137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
"""V2R1 runtime composition root for authenticated channel ingress."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Mapping
|
|
|
|
from .channel_adapter import ChannelAdapter, IngressResult
|
|
from .identity import (
|
|
AuthenticationError,
|
|
IngressAuthenticator,
|
|
RuntimeIdentity,
|
|
RuntimeIdentityResolver,
|
|
)
|
|
from .message_ledger import MessageLedger
|
|
from .router import TenantRouter
|
|
from .session_pool import SessionPool
|
|
from .tenant_context import ConflictError, ScopeError, TenantContext, ValidationError, utc_now
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeIngressResponse:
|
|
"""Sanitised boundary response; no exception text crosses external ingress."""
|
|
|
|
status: str
|
|
result: IngressResult | None = None
|
|
|
|
|
|
class RuntimeWiring:
|
|
"""Own the only external-to-runtime identity binding and message path.
|
|
|
|
The request accepted by :meth:`ingest` is an HMAC-authenticated envelope;
|
|
tenant, cell, channel, account, session, principal, and message claims are
|
|
signed together. The envelope is then resolved against ``TenantRouter``.
|
|
Only that resolved identity can mint the short-lived internal context passed
|
|
to ``ChannelAdapter``.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
router: TenantRouter,
|
|
ledger: MessageLedger,
|
|
session_pool: SessionPool,
|
|
channel_adapter: ChannelAdapter,
|
|
identity_resolver: RuntimeIdentityResolver,
|
|
ingress_authenticator: IngressAuthenticator,
|
|
internal_signing_key: bytes,
|
|
clock: Callable[[], datetime],
|
|
) -> None:
|
|
self.router = router
|
|
self.ledger = ledger
|
|
self.session_pool = session_pool
|
|
self.channel_adapter = channel_adapter
|
|
self.identity_resolver = identity_resolver
|
|
self._ingress_authenticator = ingress_authenticator
|
|
self._internal_signing_key = internal_signing_key
|
|
self._clock = clock
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
root: str | Path,
|
|
*,
|
|
signing_key: bytes,
|
|
ingress_hmac_key: bytes | None = None,
|
|
clock: Callable[[], datetime] = utc_now,
|
|
) -> "RuntimeWiring":
|
|
if not isinstance(signing_key, bytes) or len(signing_key) < 32:
|
|
raise ValidationError("internal signing key must be at least 32 bytes")
|
|
# Kept optional for one controlled migration release; the external
|
|
# request path still cannot supply a TenantContext token.
|
|
hmac_key = ingress_hmac_key if ingress_hmac_key is not None else signing_key
|
|
root_path = Path(root)
|
|
root_path.mkdir(parents=True, exist_ok=True)
|
|
router = TenantRouter(root_path / "router.sqlite3", clock=clock)
|
|
ledger = MessageLedger(root_path / "message_ledger.sqlite3")
|
|
session_pool = SessionPool(router)
|
|
adapter = ChannelAdapter(
|
|
router=router,
|
|
ledger=ledger,
|
|
session_pool=session_pool,
|
|
signing_key=signing_key,
|
|
clock=clock,
|
|
)
|
|
return cls(
|
|
router=router,
|
|
ledger=ledger,
|
|
session_pool=session_pool,
|
|
channel_adapter=adapter,
|
|
identity_resolver=RuntimeIdentityResolver(router),
|
|
ingress_authenticator=IngressAuthenticator(hmac_key),
|
|
internal_signing_key=signing_key,
|
|
clock=clock,
|
|
)
|
|
|
|
def _issue_internal_context(self, identity: RuntimeIdentity) -> TenantContext:
|
|
"""Mint internal scope only after all claims matched the router route."""
|
|
return TenantContext.issue(
|
|
lifetime_seconds=300,
|
|
now=self._clock(),
|
|
tenant_id=identity.tenant_id,
|
|
cell_id=identity.cell_id,
|
|
principal_id=identity.principal_id,
|
|
principal_type="authenticated_user",
|
|
session_id=identity.session_id,
|
|
task_id=None,
|
|
channel=identity.channel,
|
|
channel_account_id=identity.channel_account_id,
|
|
)
|
|
|
|
def ingest(self, *, request: Mapping[str, Any]) -> RuntimeIngressResponse:
|
|
"""Authenticate, resolve, issue internal context, then append and lease.
|
|
|
|
The only successful route is ``IngressAuthenticator`` ->
|
|
``RuntimeIdentityResolver`` -> ``ChannelAdapter`` ->
|
|
``MessageLedger.append`` + ``SessionPool``. The legacy caller-supplied
|
|
TenantContext token interface is intentionally not accepted.
|
|
"""
|
|
try:
|
|
ingress = self._ingress_authenticator.authenticate(request)
|
|
identity = self.identity_resolver.resolve(ingress.principal, ingress.claims)
|
|
context = self._issue_internal_context(identity)
|
|
result = self.channel_adapter.ingest(
|
|
token=context.sign(self._internal_signing_key), message=ingress.message
|
|
)
|
|
return RuntimeIngressResponse(status="ACCEPTED", result=result)
|
|
except (AuthenticationError, ScopeError):
|
|
return RuntimeIngressResponse(status="AUTH_FAILED")
|
|
except ValidationError:
|
|
return RuntimeIngressResponse(status="VALIDATION_FAILED")
|
|
except ConflictError:
|
|
return RuntimeIngressResponse(status="CONFLICT")
|
|
except Exception:
|
|
return RuntimeIngressResponse(status="INTERNAL_ERROR")
|