306 lines
10 KiB
Python
306 lines
10 KiB
Python
"""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()
|