feat: integrate idpyoidc server into app lifespan

This commit is contained in:
Johan Lundberg 2026-02-16 13:29:39 +01:00
parent 2426e0675c
commit 95d184ce0f
No known key found for this signature in database
GPG key ID: A6C152738D03C7D1
2 changed files with 33 additions and 0 deletions

View file

@ -16,6 +16,7 @@ from fastapi_oidc_op.authn.webauthn import WebAuthnService
from fastapi_oidc_op.config import Settings, StorageBackend from fastapi_oidc_op.config import Settings, StorageBackend
from fastapi_oidc_op.invite.service import MagicLinkService from fastapi_oidc_op.invite.service import MagicLinkService
from fastapi_oidc_op.manage.routes import router as manage_router from fastapi_oidc_op.manage.routes import router as manage_router
from fastapi_oidc_op.oidc.provider import create_oidc_server
from fastapi_oidc_op.store.sqlite.migrations import run_migrations from fastapi_oidc_op.store.sqlite.migrations import run_migrations
from fastapi_oidc_op.store.sqlite.repositories import ( from fastapi_oidc_op.store.sqlite.repositories import (
SQLiteCredentialRepository, SQLiteCredentialRepository,
@ -57,6 +58,24 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
ttl=settings.invite_ttl, ttl=settings.invite_ttl,
) )
# OIDC Server
oidc_server = create_oidc_server(settings)
app.state.oidc_server = oidc_server
# Register management client
manage_secret = settings.session_secret or secrets.token_hex(32)
oidc_server.context.cdb[settings.manage_client_id] = {
"client_id": settings.manage_client_id,
"client_secret": manage_secret,
"redirect_uris": [(f"{settings.issuer}/manage/callback", {})],
"response_types_supported": ["code"],
"token_endpoint_auth_method": "client_secret_basic",
"scope": ["openid", "profile", "email"],
"allowed_scopes": ["openid", "profile", "email"],
"client_salt": secrets.token_hex(8),
}
oidc_server.keyjar.add_symmetric(settings.manage_client_id, manage_secret)
yield yield
await db.close() await db.close()
else: else:

View file

@ -0,0 +1,14 @@
from httpx import AsyncClient
async def test_oidc_server_on_app_state(client: AsyncClient) -> None:
app = client._transport.app # type: ignore[union-attr]
assert hasattr(app.state, "oidc_server")
assert app.state.oidc_server is not None
async def test_manage_client_registered(client: AsyncClient) -> None:
app = client._transport.app # type: ignore[union-attr]
oidc_server = app.state.oidc_server
settings = app.state.settings
assert settings.manage_client_id in oidc_server.context.cdb