user-engine/src/user_engine/runtime.py

"""Production runtime assembly for the WSGI portal."""

from __future__ import annotations

import os
from wsgiref.simple_server import make_server

from user_engine.adapters import (
    LocalAuthorizationCheckPort,
    PostgresUserEngineStore,
    VerifiedIdentityClaimsAdapter,
    HTTPIdentityProvisioningAdapter,
    HTTPTenantManagementAdapter,
)
from user_engine.service import UserEngineService
from user_engine.oidc import OIDCClient
from user_engine.web import PortalApplication


def create_application() -> PortalApplication:
    """Assemble the runtime from secret-backed environment references.

    The local authorization adapter is an explicit pre-production bridge. A
    flex-auth HTTP adapter must replace it before the production gate.
    """

    try:
        import psycopg
    except ImportError as exc:  # pragma: no cover - deployment guard
        raise RuntimeError("install user-engine[postgres] for the runtime") from exc

    database_url = _required("USER_ENGINE_DATABASE_URL")
    store = PostgresUserEngineStore(psycopg.connect(database_url))
    store.migrate()
    service = UserEngineService(
        store=store,
        identity_adapter=VerifiedIdentityClaimsAdapter(
            expected_issuer=_required("USER_ENGINE_OIDC_ISSUER"),
            expected_audience=_required("USER_ENGINE_OIDC_AUDIENCE"),
        ),
        authorization=LocalAuthorizationCheckPort(),
    )
    tenant_management = None
    if os.environ.get("USER_ENGINE_TENANT_MANAGEMENT_URL"):
        tenant_management = HTTPTenantManagementAdapter(
            base_url=_required("USER_ENGINE_TENANT_MANAGEMENT_URL"),
            bearer_token=_required("USER_ENGINE_TENANT_MANAGEMENT_TOKEN"),
        )
    return PortalApplication(
        service,
        trusted_proxy_secret=_required("USER_ENGINE_PROXY_SECRET"),
        login_url=_required("USER_ENGINE_LOGIN_URL"),
        public_registration=os.environ.get("USER_ENGINE_PUBLIC_REGISTRATION", "false").lower()
        == "true",
        oidc_client=OIDCClient(
            issuer=_required("USER_ENGINE_OIDC_ISSUER"),
            client_id=_required("USER_ENGINE_OIDC_CLIENT_ID"),
            redirect_uri=_required("USER_ENGINE_OIDC_REDIRECT_URI"),
            audience=_required("USER_ENGINE_OIDC_AUDIENCE"),
            backend_url=os.environ.get("USER_ENGINE_OIDC_BACKEND_URL"),
        ),
        provisioning=HTTPIdentityProvisioningAdapter(
            base_url=_required("USER_ENGINE_PROVISIONING_URL"),
            bearer_token=_required("USER_ENGINE_PROVISIONING_TOKEN"),
        ),
        tenant_management=tenant_management,
    )


def main() -> None:
    host = os.environ.get("USER_ENGINE_HOST", "0.0.0.0")
    # ``USER_ENGINE_PORT`` is reserved by Kubernetes service-link injection
    # (for example ``tcp://10.43.0.1:8080``), so use an unambiguous setting.
    port = int(os.environ.get("USER_ENGINE_HTTP_PORT", "8080"))
    with make_server(host, port, create_application()) as server:
        server.serve_forever()


def _required(name: str) -> str:
    # Values sourced from files or `kubectl create secret --from-file` commonly
    # retain one trailing newline.  Such a value is unusable in HTTP headers
    # (notably USER_ENGINE_PROXY_SECRET and provisioning bearer tokens), and
    # comparing it byte-for-byte makes the trusted boundary impossible to
    # exercise.  Normalize transport whitespace at the runtime boundary; the
    # domain and adapters still receive an opaque non-empty value.
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


if __name__ == "__main__":
    main()