"""Focused tests for secure Agent enrollment.""" import re import unittest from datetime import datetime, timedelta, timezone from uuid import UUID from fastapi import FastAPI from fastapi.testclient import TestClient from app.agent_enrollment.dependencies import get_agent_enrollment_service from app.agent_auth.service import hash_agent_secret from app.agent_enrollment.models import ( AgentCredentialTransitionCreateRequest, AgentCredentialTransitionRedeemRequest, AgentEnrollmentCreateRequest, AgentEnrollRequest, ) from app.agent_enrollment.service import AgentEnrollmentService from app.agent_enrollment.router import router from app.identity import ( get_current_user, machine_agent_request, required_api_permission, ) SITE_ID = UUID("00000000-0000-4000-8000-000000001700") ZONE_ID = UUID("00000000-0000-4000-8000-000000001701") USER_ID = UUID("00000000-0000-4000-8000-000000001702") ENROLLMENT_ID = UUID("00000000-0000-4000-8000-000000001703") NOW = datetime(2026, 8, 26, tzinfo=timezone.utc) class FakeAgentEnrollmentRepository: def __init__(self): self.created = None self.redeemed = None self.transition_created = None self.transition_redeemed = None def create_enrollment(self, **kwargs): self.created = kwargs return { "id": ENROLLMENT_ID, "site_id": SITE_ID, "site_name": "Test Site", "agent_zone_id": ZONE_ID, "agent_zone_name": "Test Zone", "display_name": kwargs["display_name"], "created_at": NOW, "expires_at": NOW + timedelta(minutes=30), } def list_enrollments(self, **kwargs): return [] def create_credential_transition(self, **kwargs): self.transition_created = kwargs return { "id": ENROLLMENT_ID, "target_agent_id": kwargs["target_agent_id"], "site_id": SITE_ID, "site_name": "Test Site", "agent_zone_id": ZONE_ID, "agent_zone_name": "Test Zone", "display_name": "Existing Agent", "created_at": NOW, "expires_at": NOW + timedelta(minutes=30), } def redeem_enrollment(self, **kwargs): self.redeemed = kwargs return { "agent_id": kwargs["agent_id"], "site_id": SITE_ID, "site_name": "Test Site", "agent_zone_id": ZONE_ID, "agent_zone_name": "Test Zone", "display_name": "Test Agent", "ip_ranges": ["10.30.0.0/22"], } def redeem_credential_transition(self, **kwargs): self.transition_redeemed = kwargs return { "agent_id": "existing-agent", "site_id": SITE_ID, "site_name": "Test Site", "agent_zone_id": ZONE_ID, "agent_zone_name": "Test Zone", "display_name": "Existing Agent", "ip_ranges": ["10.30.0.0/22"], } class AgentEnrollmentServiceTests(unittest.TestCase): def test_create_returns_plaintext_token_once_but_repository_gets_only_hash(self): repository = FakeAgentEnrollmentRepository() service = AgentEnrollmentService(repository) result = service.create_enrollment( AgentEnrollmentCreateRequest( site_id=SITE_ID, agent_zone_id=ZONE_ID, display_name=" Test Agent ", ), actor_user_id=USER_ID, ) token = result["enrollment_token"] self.assertGreaterEqual(len(token), 32) self.assertNotEqual(repository.created["token_hash"], token) self.assertEqual( repository.created["token_hash"], hash_agent_secret(token), ) self.assertRegex( repository.created["token_hash"], r"^[0-9a-f]{64}$", ) self.assertEqual( repository.created["display_name"], "Test Agent", ) serialized = repr(repository.created) self.assertNotIn(token, serialized) def test_redeem_returns_plaintext_secret_but_repository_gets_only_hash(self): repository = FakeAgentEnrollmentRepository() service = AgentEnrollmentService(repository) result = service.redeem( AgentEnrollRequest( token="A" * 43, ) ) secret = result["agent_secret"] self.assertGreaterEqual(len(secret), 32) self.assertNotEqual( repository.redeemed["agent_secret_hash"], secret, ) self.assertEqual( repository.redeemed["agent_secret_hash"], hash_agent_secret(secret), ) self.assertRegex( repository.redeemed["agent_secret_hash"], r"^[0-9a-f]{64}$", ) self.assertRegex( result["agent_id"], r"^agent-[0-9a-f]{32}$", ) serialized = repr(repository.redeemed) self.assertNotIn(secret, serialized) def test_existing_transition_is_server_bound_and_accepts_only_secret_hash(self): repository = FakeAgentEnrollmentRepository() service = AgentEnrollmentService(repository) created = service.create_credential_transition( AgentCredentialTransitionCreateRequest( agent_id=" existing-agent " ), actor_user_id=USER_ID, ) token = created["transition_token"] self.assertEqual( repository.transition_created["target_agent_id"], "existing-agent", ) self.assertEqual( repository.transition_created["token_hash"], hash_agent_secret(token), ) self.assertNotIn(token, repr(repository.transition_created)) credential_hash = hash_agent_secret("locally-generated-secret") result = service.redeem_credential_transition( AgentCredentialTransitionRedeemRequest( token=token, agent_secret_hash=credential_hash, ) ) self.assertEqual(result["agent_id"], "existing-agent") self.assertEqual( repository.transition_redeemed, { "token_hash": hash_agent_secret(token), "agent_secret_hash": credential_hash, }, ) self.assertNotIn("agent_secret", result) def test_enrollment_admin_permission_mapping_is_explicit(self): self.assertEqual( required_api_permission("GET", "/api/agent-enrollments"), "agents.view", ) self.assertEqual( required_api_permission("POST", "/api/agent-enrollments"), "agents.manage", ) self.assertEqual( required_api_permission( "POST", "/api/agent-credential-transitions" ), "agents.manage", ) def test_bootstrap_endpoint_is_machine_exempt_but_admin_endpoint_is_not(self): self.assertTrue(machine_agent_request("/api/agent/enroll")) self.assertTrue( machine_agent_request("/api/agent/credential-transition") ) self.assertFalse( machine_agent_request("/api/agent-enrollments") ) def test_generated_identifiers_are_bounded(self): repository = FakeAgentEnrollmentRepository() service = AgentEnrollmentService(repository) result = service.redeem( AgentEnrollRequest(token="B" * 43) ) self.assertLessEqual(len(result["agent_id"]), 255) self.assertTrue( re.fullmatch(r"agent-[0-9a-f]{32}", result["agent_id"]) ) class AgentCredentialTransitionRouterTests(unittest.TestCase): def setUp(self): self.repository = FakeAgentEnrollmentRepository() self.service = AgentEnrollmentService(self.repository) self.app = FastAPI() self.app.include_router(router) self.app.dependency_overrides[get_agent_enrollment_service] = ( lambda: self.service ) self.app.dependency_overrides[get_current_user] = lambda: { "id": str(USER_ID), "role": "Test", "permissions": ("agents.view", "agents.manage"), } self.client = TestClient(self.app) def tearDown(self): self.client.close() def test_admin_creation_is_distinct_no_store_and_identity_only(self): response = self.client.post( "/api/agent-credential-transitions", json={"agent_id": "existing-agent"}, ) self.assertEqual(response.status_code, 201) self.assertEqual(response.headers["cache-control"], "no-store") payload = response.json()["data"] self.assertEqual(payload["target_agent_id"], "existing-agent") self.assertIn("transition_token", payload) self.assertNotIn("site_id", self.repository.transition_created) self.assertNotIn("agent_zone_id", self.repository.transition_created) def test_machine_redemption_returns_no_permanent_plaintext_secret(self): response = self.client.post( "/api/agent/credential-transition", json={ "token": "T" * 43, "agent_secret_hash": hash_agent_secret( "locally-generated-router-secret" ), }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.headers["cache-control"], "no-store") payload = response.json()["data"] self.assertEqual(payload["agent_id"], "existing-agent") self.assertNotIn("agent_secret", payload) def test_view_only_admin_cannot_create_transition(self): self.app.dependency_overrides[get_current_user] = lambda: { "id": str(USER_ID), "role": "Test", "permissions": ("agents.view",), } response = self.client.post( "/api/agent-credential-transitions", json={"agent_id": "existing-agent"}, ) self.assertEqual(response.status_code, 403) if __name__ == "__main__": unittest.main()