137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""Focused tests for per-Agent authentication and legacy compatibility."""
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from app.agent_auth.exceptions import (
|
|
AgentAuthenticationFailed,
|
|
AgentAuthenticationUnavailable,
|
|
AgentCredentialRevoked,
|
|
AgentIdentityUnauthorized,
|
|
)
|
|
from app.agent_auth.service import AgentAuthService, hash_agent_secret
|
|
from app.config import settings
|
|
|
|
|
|
NOW = datetime(2026, 8, 26, tzinfo=timezone.utc)
|
|
|
|
|
|
class FakeAgentAuthRepository:
|
|
def __init__(self, rows=None):
|
|
self.rows = rows or {}
|
|
|
|
def get_credential(self, agent_id):
|
|
return self.rows.get(agent_id)
|
|
|
|
|
|
class AgentAuthServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.old_id = settings.minerops_agent_id
|
|
self.old_secret = settings.minerops_agent_shared_secret
|
|
settings.minerops_agent_id = "legacy-agent"
|
|
settings.minerops_agent_shared_secret = "legacy-secret"
|
|
|
|
def tearDown(self):
|
|
settings.minerops_agent_id = self.old_id
|
|
settings.minerops_agent_shared_secret = self.old_secret
|
|
|
|
def service(self, rows=None):
|
|
return AgentAuthService(FakeAgentAuthRepository(rows))
|
|
|
|
def test_high_entropy_secret_hash_is_sha256_hex(self):
|
|
digest = hash_agent_secret("example-secret")
|
|
self.assertEqual(len(digest), 64)
|
|
self.assertRegex(digest, r"^[0-9a-f]{64}$")
|
|
|
|
def test_legacy_agent_without_registry_row_still_authenticates(self):
|
|
self.assertEqual(
|
|
self.service().authenticate("legacy-agent", "legacy-secret"),
|
|
"legacy-agent",
|
|
)
|
|
|
|
def test_legacy_agent_with_null_credential_still_authenticates(self):
|
|
rows = {
|
|
"legacy-agent": {
|
|
"agent_id": "legacy-agent",
|
|
"agent_secret_hash": None,
|
|
"credential_created_at": None,
|
|
"credential_revoked_at": None,
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
self.service(rows).authenticate("legacy-agent", "legacy-secret"),
|
|
"legacy-agent",
|
|
)
|
|
|
|
def test_per_agent_credential_authenticates_independently(self):
|
|
rows = {
|
|
"agent-new": {
|
|
"agent_id": "agent-new",
|
|
"agent_secret_hash": hash_agent_secret("unique-secret"),
|
|
"credential_created_at": NOW,
|
|
"credential_revoked_at": None,
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
self.service(rows).authenticate("agent-new", "unique-secret"),
|
|
"agent-new",
|
|
)
|
|
|
|
def test_legacy_secret_never_authenticates_per_agent_identity(self):
|
|
settings.minerops_agent_id = "agent-new"
|
|
rows = {
|
|
"agent-new": {
|
|
"agent_id": "agent-new",
|
|
"agent_secret_hash": hash_agent_secret("unique-secret"),
|
|
"credential_created_at": NOW,
|
|
"credential_revoked_at": None,
|
|
}
|
|
}
|
|
with self.assertRaises(AgentAuthenticationFailed):
|
|
self.service(rows).authenticate("agent-new", "legacy-secret")
|
|
|
|
def test_wrong_per_agent_secret_is_rejected(self):
|
|
rows = {
|
|
"agent-new": {
|
|
"agent_id": "agent-new",
|
|
"agent_secret_hash": hash_agent_secret("unique-secret"),
|
|
"credential_created_at": NOW,
|
|
"credential_revoked_at": None,
|
|
}
|
|
}
|
|
with self.assertRaises(AgentAuthenticationFailed):
|
|
self.service(rows).authenticate("agent-new", "wrong-secret")
|
|
|
|
def test_revoked_per_agent_credential_never_falls_back(self):
|
|
settings.minerops_agent_id = "agent-new"
|
|
rows = {
|
|
"agent-new": {
|
|
"agent_id": "agent-new",
|
|
"agent_secret_hash": hash_agent_secret("unique-secret"),
|
|
"credential_created_at": NOW,
|
|
"credential_revoked_at": NOW,
|
|
}
|
|
}
|
|
for secret in ("unique-secret", "legacy-secret"):
|
|
with self.subTest(secret=secret), self.assertRaises(AgentCredentialRevoked):
|
|
self.service(rows).authenticate("agent-new", secret)
|
|
|
|
def test_unknown_identity_does_not_use_shared_secret(self):
|
|
with self.assertRaises(AgentIdentityUnauthorized):
|
|
self.service().authenticate("unknown-agent", "legacy-secret")
|
|
|
|
def test_missing_identity_or_secret_is_rejected(self):
|
|
with self.assertRaises(AgentAuthenticationFailed):
|
|
self.service().authenticate(None, "legacy-secret")
|
|
with self.assertRaises(AgentAuthenticationFailed):
|
|
self.service().authenticate("legacy-agent", None)
|
|
|
|
def test_legacy_path_reports_unavailable_if_not_configured(self):
|
|
settings.minerops_agent_shared_secret = ""
|
|
with self.assertRaises(AgentAuthenticationUnavailable):
|
|
self.service().authenticate("legacy-agent", "anything")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|