65 lines
2 KiB
Python
65 lines
2 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,
|
||
|
|
)
|
||
|
|
from user_engine.service import UserEngineService
|
||
|
|
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",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
host = os.environ.get("USER_ENGINE_HOST", "0.0.0.0")
|
||
|
|
port = int(os.environ.get("USER_ENGINE_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()
|