103 lines
4.2 KiB
Python
103 lines
4.2 KiB
Python
"""Per-Agent authentication with bounded legacy compatibility."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
from app.agent_auth.exceptions import (
|
|
AgentAuthenticationFailed,
|
|
AgentAuthenticationUnavailable,
|
|
AgentCredentialRevoked,
|
|
AgentIdentityUnauthorized,
|
|
)
|
|
from app.config import settings
|
|
|
|
|
|
def hash_agent_secret(secret: str) -> str:
|
|
"""Hash a high-entropy machine-generated Agent secret for persistence."""
|
|
return hashlib.sha256(secret.encode("utf-8")).hexdigest()
|
|
|
|
|
|
class AgentAuthService:
|
|
def __init__(self, repository):
|
|
self.repository = repository
|
|
|
|
def authenticate(self, agent_id: str | None, provided_secret: str | None) -> str:
|
|
identity = (agent_id or "").strip()
|
|
secret = (provided_secret or "").strip()
|
|
|
|
if not secret:
|
|
raise AgentAuthenticationFailed("Invalid or missing Agent secret")
|
|
|
|
# The production Alpha Agent predates the identity header. Missing-ID
|
|
# compatibility is therefore bound to exactly the configured legacy
|
|
# identity; no request value is allowed to select another Agent.
|
|
legacy_id = settings.minerops_agent_id.strip()
|
|
legacy_secret = settings.minerops_agent_shared_secret.strip()
|
|
missing_identity = not identity
|
|
if missing_identity:
|
|
if not legacy_id:
|
|
raise AgentAuthenticationUnavailable(
|
|
"Legacy Agent authentication is not configured"
|
|
)
|
|
identity = legacy_id
|
|
|
|
credential = self.repository.get_credential(identity)
|
|
|
|
# Revocation is authoritative regardless of which credential path
|
|
# would otherwise apply. Never allow malformed or historical
|
|
# credential state to fall through to the legacy shared secret.
|
|
if credential and credential.get("credential_revoked_at") is not None:
|
|
raise AgentCredentialRevoked("Agent credential has been revoked")
|
|
|
|
# This value comes from the identity-level retirement ledger, not the
|
|
# deletable current Agent row. It is therefore authoritative even when
|
|
# the Agent was deleted and has not yet been recreated.
|
|
legacy_auth_retired = bool(
|
|
credential
|
|
and credential.get("legacy_auth_disabled_at") is not None
|
|
)
|
|
|
|
# Any Agent with a per-Agent credential is permanently on the new
|
|
# authentication path. The legacy shared secret must never authenticate
|
|
# that identity, even if the Agent later becomes revoked.
|
|
if credential and credential.get("agent_secret_hash"):
|
|
# Once the configured legacy identity has ever received a
|
|
# per-Agent credential, secret-only fallback is permanently
|
|
# disabled. The legacy global secret cannot authenticate it and a
|
|
# per-Agent secret cannot be used without an explicit identity.
|
|
if missing_identity:
|
|
raise AgentAuthenticationFailed(
|
|
"Explicit Agent identity is required for enrolled Agents"
|
|
)
|
|
|
|
supplied_hash = hash_agent_secret(secret)
|
|
if not hmac.compare_digest(
|
|
supplied_hash,
|
|
str(credential["agent_secret_hash"]),
|
|
):
|
|
raise AgentAuthenticationFailed("Invalid or missing Agent secret")
|
|
|
|
return identity
|
|
|
|
# The durable identity retirement survives Agent-row deletion,
|
|
# credential-field corruption, administrative mistakes, and future
|
|
# credential rotation workflows.
|
|
if legacy_auth_retired:
|
|
raise AgentAuthenticationFailed(
|
|
"Legacy Agent authentication has been permanently disabled"
|
|
)
|
|
|
|
# Temporary compatibility for the single pre-enrollment Agent.
|
|
if not legacy_id or not legacy_secret:
|
|
raise AgentAuthenticationUnavailable(
|
|
"Legacy Agent authentication is not configured"
|
|
)
|
|
|
|
if not hmac.compare_digest(identity, legacy_id):
|
|
raise AgentIdentityUnauthorized("Agent identity is not authorized")
|
|
|
|
if not hmac.compare_digest(secret, legacy_secret):
|
|
raise AgentAuthenticationFailed("Invalid or missing Agent secret")
|
|
|
|
return identity
|