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


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:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"{name} is required")
    return value


if __name__ == "__main__":
    main()