"""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")