159 lines
5.7 KiB
Python
159 lines
5.7 KiB
Python
"""Fail-closed identity binding for authenticated external ingress.
|
|
|
|
External payload claims are not authority. An HMAC-authenticated envelope is
|
|
the sole factory for :class:`AuthenticatedPrincipal`; the resolver then checks
|
|
every claim against the authoritative ``TenantRouter`` before the runtime signs
|
|
an internal ``TenantContext``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import hmac
|
|
import re
|
|
from typing import Any, Mapping
|
|
|
|
from .router import Route, TenantRouter
|
|
from .tenant_context import ScopeError, ValidationError, canonical_json, require_ident
|
|
|
|
|
|
_PROVENANCE = object()
|
|
_SIGNATURE_RE = re.compile(r"^[0-9a-f]{64}$")
|
|
_SCOPE_FIELDS = (
|
|
"principal_id", "tenant_id", "cell_id", "channel",
|
|
"channel_account_id", "session_id", "message",
|
|
)
|
|
|
|
|
|
class AuthenticationError(ValidationError):
|
|
"""Authentication or identity resolution failed without disclosing scope."""
|
|
|
|
|
|
@dataclass(frozen=True, init=False)
|
|
class AuthenticatedPrincipal:
|
|
"""A principal which only the verified envelope factory can create."""
|
|
|
|
principal_id: str
|
|
|
|
def __init__(self, principal_id: str, *, _provenance: object) -> None:
|
|
if _provenance is not _PROVENANCE:
|
|
raise TypeError("AuthenticatedPrincipal is created by authenticated ingress only")
|
|
object.__setattr__(self, "principal_id", require_ident("principal_id", principal_id))
|
|
|
|
@classmethod
|
|
def _verified(cls, principal_id: str) -> "AuthenticatedPrincipal":
|
|
return cls(principal_id, _provenance=_PROVENANCE)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IdentityClaims:
|
|
tenant_id: str
|
|
cell_id: str
|
|
channel: str
|
|
channel_account_id: str
|
|
session_id: str
|
|
|
|
def __post_init__(self) -> None:
|
|
require_ident("tenant_id", self.tenant_id, tenant=True)
|
|
for field in ("cell_id", "channel", "channel_account_id", "session_id"):
|
|
require_ident(field, getattr(self, field))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthenticatedIngress:
|
|
principal: AuthenticatedPrincipal
|
|
claims: IdentityClaims
|
|
message: Mapping[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeIdentity:
|
|
tenant_id: str
|
|
cell_id: str
|
|
principal_id: str
|
|
channel: str
|
|
channel_account_id: str
|
|
session_id: str
|
|
|
|
|
|
class IngressAuthenticator:
|
|
"""Verify a canonical HMAC envelope before any identity lookup occurs."""
|
|
|
|
def __init__(self, hmac_key: bytes) -> None:
|
|
if not isinstance(hmac_key, bytes) or len(hmac_key) < 32:
|
|
raise ValidationError("ingress HMAC key must be at least 32 bytes")
|
|
self._hmac_key = hmac_key
|
|
|
|
@staticmethod
|
|
def _signed_payload(request: Mapping[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, Mapping):
|
|
raise ValidationError("ingress request must be an object")
|
|
if set(request) != {*_SCOPE_FIELDS, "signature"}:
|
|
raise ValidationError("ingress request fields are invalid")
|
|
values = {field: request[field] for field in _SCOPE_FIELDS}
|
|
if not isinstance(values["message"], Mapping):
|
|
raise ValidationError("ingress message must be an object")
|
|
return values
|
|
|
|
def authenticate(self, request: Mapping[str, Any]) -> AuthenticatedIngress:
|
|
values = self._signed_payload(request)
|
|
signature = request["signature"]
|
|
if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
|
|
raise AuthenticationError("invalid ingress signature")
|
|
expected = hmac.new(
|
|
self._hmac_key, canonical_json(values), hashlib.sha256
|
|
).hexdigest()
|
|
if not hmac.compare_digest(expected, signature):
|
|
raise AuthenticationError("invalid ingress signature")
|
|
return AuthenticatedIngress(
|
|
principal=AuthenticatedPrincipal._verified(values["principal_id"]),
|
|
claims=IdentityClaims(
|
|
tenant_id=values["tenant_id"],
|
|
cell_id=values["cell_id"],
|
|
channel=values["channel"],
|
|
channel_account_id=values["channel_account_id"],
|
|
session_id=values["session_id"],
|
|
),
|
|
message=dict(values["message"]),
|
|
)
|
|
|
|
|
|
class RuntimeIdentityResolver:
|
|
"""Resolve authenticated ingress identity through ``TenantRouter`` only."""
|
|
|
|
def __init__(self, router: TenantRouter) -> None:
|
|
self._router = router
|
|
|
|
def resolve(
|
|
self, principal: AuthenticatedPrincipal, claims: IdentityClaims
|
|
) -> RuntimeIdentity:
|
|
if not isinstance(principal, AuthenticatedPrincipal):
|
|
raise AuthenticationError("untrusted principal type")
|
|
try:
|
|
route: Route = self._router.resolve(
|
|
channel=claims.channel,
|
|
channel_account_id=claims.channel_account_id,
|
|
principal_id=principal.principal_id,
|
|
session_id=claims.session_id,
|
|
)
|
|
except (ScopeError, ValidationError) as exc:
|
|
raise AuthenticationError("identity is not registered") from None
|
|
expected = (
|
|
claims.tenant_id, claims.cell_id, claims.channel,
|
|
claims.channel_account_id, principal.principal_id, claims.session_id,
|
|
)
|
|
actual = (
|
|
route.tenant_id, route.cell_id, route.channel,
|
|
route.channel_account_id, route.principal_id, route.session_id,
|
|
)
|
|
if actual != expected:
|
|
raise AuthenticationError("identity claims do not match authoritative route")
|
|
return RuntimeIdentity(
|
|
tenant_id=route.tenant_id,
|
|
cell_id=route.cell_id,
|
|
principal_id=route.principal_id,
|
|
channel=route.channel,
|
|
channel_account_id=route.channel_account_id,
|
|
session_id=route.session_id or "",
|
|
)
|