79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""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()
|