lussierj
8b0d32cd6b
Bind Agent observations, review, enrichment and telemetry to current Zone
and inventory authority. Preserve immutable MAC ownership and historical
snapshots with stale-result rejection and transactional security audits.
Retire unbound direct/range/scheduled discovery and one/bulk legacy summary
writes in both modes. Require reconciled exact inventory identity or
explicit Zone-scoped IP selection for CSV metadata imports.
Independent Blocker #1 application-security re-audit: PASS on tree
cf863aadc9. Evidence: 141 focused PostgreSQL/HTTP
tests passed; Chromium and PowerShell checks passed. Two documented baseline
source assertions remain unchanged. Production deployment is not approved.
40 lines
1.7 KiB
Python
40 lines
1.7 KiB
Python
"""Reused discovery isolation scenarios exercised through real HTTP authority."""
|
|
|
|
import unittest
|
|
from tests.test_discovery_security_postgres import DiscoveryFixtures
|
|
|
|
|
|
class AgentDiscoveryIsolationPostgresTests(DiscoveryFixtures, unittest.TestCase):
|
|
def test_cross_zone_mac_and_ip_are_unmatched_without_id_disclosure(self):
|
|
mid, mac = self.miner(1)
|
|
before = self.durable()
|
|
row = self.observe(mac=mac)
|
|
self.assertEqual(row["discovery_status"], "zone_conflict")
|
|
self.assertIsNone(row["matched_miner_id"])
|
|
self.assertNotIn(str(mid), str(row))
|
|
self.assertEqual(before["miners"], self.durable()["miners"])
|
|
|
|
def test_same_zone_mac_and_ip_matches_are_preserved(self):
|
|
mid, mac = self.miner()
|
|
row = self.observe(mac=mac)
|
|
self.assertEqual(row["matched_miner_id"], str(mid))
|
|
self.assertEqual(row["discovery_status"], "known")
|
|
|
|
def test_unassigned_miner_is_not_claimed_by_discovery_matching(self):
|
|
mid, mac = self.miner()
|
|
with __import__(
|
|
"app.agent_zones.repository", fromlist=["default_connection_factory"]
|
|
).default_connection_factory() as conn:
|
|
conn.execute("UPDATE miners SET agent_zone_id=NULL WHERE id=%s", (mid,))
|
|
row = self.observe(mac=mac)
|
|
self.assertIsNone(row["matched_miner_id"])
|
|
self.assertIsNone(
|
|
self.scalar("SELECT agent_zone_id FROM miners WHERE id=%s", (mid,))
|
|
)
|
|
|
|
def test_claimed_agent_identity_cannot_override_authentication(self):
|
|
response = self.post(
|
|
"/api/agent/discoveries", {"agent_id": self.agents[1], "discoveries": []}
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|