Core API#

The core is framework independent and can be used directly.

Framework independent OpenID Connect client.

The OIDCAuth class implements the common OpenID Connect and OAuth 2 flows used by the framework adapters.

It is intentionally framework independent. The FastAPI, Flask, Quart, Tornado, Litestar, and Django integrations expose HTTP endpoints and decorators that call into this class.

Supported flows

Authorization code flow with PKCE
Refresh token flow
Device authorization flow
Userinfo lookup
Provider initiated logout (end session)

Quick start

from py_oidc_auth.auth_base import OIDCAuth

auth = OIDCAuth(
    client_id="my client",
    client_secret="secret",
    discovery_url="https://idp.example.org/realms/demo/.well-known/openid-configuration",
    scopes="openid profile email",
    offline_access=True,
    broker_mode=True,
    broker_store_url="postgresql+asyncpg://user:pw@db/myapp",
    broker_audience="myapp-api",
    trusted_issuers=["https://other-instance.example.org"],
)

login_url = await auth.login(
    redirect_uri="https://app.example.org/auth/callback",
    prompt="login",
    offline_access=True,
)

token_payload = await auth.callback(code="...", state="...")

The value returned by OIDCAuth.login() is a URL that you redirect the browser to. The code and state values are sent back to your callback endpoint by the identity provider.

class py_oidc_auth.auth_base.OIDCAuth(client_id: str = '', discovery_url: str = '', client_secret: str | None = None, scopes: str = 'profile email', audience: str | None = None, appname: str = 'py-oidc-auth', proxy: str = '', claims: Dict[str, Any] | None = None, offline_access: bool = True, timeout_sec: int = 10, jwks_uri: str | None = None, issuer: str | None = None, broker_mode: bool = False, broker_store_url: str | None = None, broker_store_obj: BrokerStore | None = None, broker_audience: str = 'py-oidc-auth', trusted_issuers: list[str] | None = None, broker_jwks_path: str = '/auth/v2/.well-known/jwks.json')#

Async OIDC client with a minimal, stable API.

Instances of this class hold configuration and a lazy initialized TokenVerifier.

Parameters:

client_id – OIDC client identifier.
discovery_url – URL of the provider discovery document.
client_secret – Client secret for confidential clients.
scopes – Default scopes as a space separated string.
proxy – Public base URL of your application.
claims – Optional claim constraints for token validation.
audience – Optional audience constraints for token validation.
offline_access – If true, include offline_access in scope to request a refresh token.
timeout_sec – HTTP timeout for discovery and provider calls.
jwks_uri – Use this jwks uri instead of the one provided by the discovery-url
issuer – Use this issuer instead of the one provided by the disovery-url
broker_mode – Enable token broker mode. When True, the library mints its own RS256-signed JWTs instead of passing IDP tokens through. required() and optional() verify against the broker JWKS. A token endpoint must be configured in create_auth_router when broker mode is enabled.
broker_store_url – Connection URL for the broker storage backend. Defaults to a local SQLite file. Supported schemes: memory://, mongodb://, sqlite+aiosqlite:///, postgresql+asyncpg://, mysql+aiomysql://.
broker_store_obj – A pre-instantiated BrokerStore. Use this when you want to share an existing database connection rather than have the library create its own. Takes precedence over broker_store_url. For example, pass a MongoDBBrokerStore built from your application’s existing Motor/pymongo client, or a SQLAlchemyBrokerStore built from your existing async engine.
broker_audience – aud claim written into minted JWTs. Defaults to py-oidc-auth.
trusted_issuers – List of peer instance base URLs whose JWTs are accepted for cross-instance federation.
broker_jwks_path – Path appended to peer URLs when fetching JWKS.

Example

from py_oidc_auth import OIDCAuth

auth = OIDCAuth(
    client_id="my client",
    discovery_url="https://idp.example.org/.well-known/openid-configuration",
    client_secret="secret",
    scopes="myscope profile email",
    appname="my-app",
    audience="my-aud",
    broker_mode=True,
    broker_store_url="postgresql+asyncpg://user:pw@db/myapp",
    broker_audience="myapp-api",
    trusted_issuers=["https://other-instance.example.org"],
)

With an existing database connection:

from pymongo import AsyncMongoClient
from py_oidc_auth import OIDCAuth, MongoDBBrokerStore

mongo_client = AsyncMongoClient("mongodb://user:pass@host")
mongo_store = MongoDBBrokerStore(db=mongo_client["my_app"])
auth = OIDCAuth(
    client_id="my client",
    discovery_url="https://idp.example.org/.well-known/openid-configuration",
    client_secret="secret",
    scopes="myscope profile email",
    appname="my-app",
    audience="my-aud",
    broker_mode=True,
    broker_store_obj=mongo_store,
    broker_audience="myapp-api",
    trusted_issuers=["https://other-instance.example.org"],
)

async make_oidc_request(method: str, endpoint_key: str, *, data: Dict[str, str] | None = None, headers: Dict[str, str] | None = None) → Dict[str, Any]#

Call a provider endpoint from the discovery document.

The discovery document contains URLs such as token_endpoint and device_authorization_endpoint.

Parameters:

method – HTTP method, for example POST.
endpoint_key – Key in the discovery document.
data – Optional form data.
headers – Optional request headers.

Returns:

JSON response decoded into a dict.

Raises:

InvalidRequest – If the endpoint is missing or the request fails.

Example

jwks = await auth.make_oidc_request("GET", "jwks_uri")

async login(redirect_uri: str | None, prompt: Literal['none', 'login', 'consent', 'select_account'], offline_access: bool = False, scope: str | None = None) → str#

Create the authorization URL for the authorization code flow.

This method generates a URL for the provider authorization endpoint. It includes PKCE parameters and stores state information in the state parameter.

Parameters:

redirect_uri – Absolute URL of your callback endpoint.
prompt – Provider prompt parameter.
offline_access – If true, include offline_access in scope.
scope – Optional scope override.

Returns:

URL to redirect the browser to.

Raises:

InvalidRequest – If required configuration is missing.

Example

url = await auth.login(
    redirect_uri="https://app.example.org/auth/callback",
    prompt="login",
    offline_access=True,
)

Request example#

GET /auth/v2/login?redirect_uri=https%3A%2F%2Fapp.example.org%2Fauth%2Fcallback HTTP/1.1
Host: app.example.org

async callback(code: str | None = None, state: str | None = None) → Dict[str, str | int]#

Handle the callback from the authorization code flow.

Providers call your callback endpoint with query parameters code and state. This method exchanges the code for tokens by calling the provider token endpoint.

Parameters:

code – Authorization code from the provider.
state – Opaque state created by login().

Returns:

Raw JSON response from the token endpoint.

Raises:

InvalidRequest – If inputs are missing or the exchange fails.

Request example#

GET /auth/v2/callback?code=abc&state=xyz HTTP/1.1
Host: app.example.org

async device_flow() → DeviceStartResponse#

Start the OAuth 2 device authorization flow.

This is useful for devices without a browser. The provider returns a user code and a verification URI. The user visits the URI, enters the code, and authorizes the device.

Returns:: Device authorization information.
Raises:: InvalidRequest – If the provider response is malformed.

Request example#

POST /auth/v2/device HTTP/1.1
Host: app.example.org
Content-Type: application/x-www-form-urlencoded
Content-Length: 0

Exchange, refresh, or poll for an access token.

Exactly one of code, refresh_token, or device_code must be provided.

Parameters:

endpoint – Local endpoint path used to compute a default redirect URI when exchanging an authorization code.
code – Authorization code.
redirect_uri – Redirect URI to send to the token endpoint.
refresh_token – Refresh token for renewing access.
device_code – Device code for polling in the device flow.
code_verifier – PKCE verifier used during the login step.

Returns:

Parsed Token.

Raises:

InvalidRequest – If inputs are missing or the provider call fails.

Request examples#

Authorization code exchange

POST /auth/v2/token HTTP/1.1
Host: app.example.org
Content-Type: application/x-www-form-urlencoded

code=abc&redirect_uri=https%3A%2F%2Fapp.example.org%2Fauth%2Fcallback&code_verifier=xyz

Refresh token

POST /auth/v2/token HTTP/1.1
Host: app.example.org
Content-Type: application/x-www-form-urlencoded

refresh_token=ref

Device polling

POST /auth/v2/token HTTP/1.1
Host: app.example.org
Content-Type: application/x-www-form-urlencoded

device_code=device

async logout(post_logout_redirect_uri: str | None) → str#

Create a provider logout redirect target.

If the provider advertises an end_session_endpoint in the discovery document, the returned URL points to that endpoint. Otherwise the method returns the local post_logout_redirect_uri or /.

Parameters:: post_logout_redirect_uri – Local URI to redirect to after logout.
Returns:: Redirect target URL.

Request example

GET /auth/v2/logout?post_logout_redirect_uri=https%3A%2F%2Fapp.example.org HTTP/1.1
Host: app.example.org

async userinfo(id_token: IDToken, header: Dict[str, Dict[str, PayloadContent] | List[PayloadContent] | PayloadContent | None]) → UserInfo#

Fetch user details using the userinfo endpoint.

The method first tries to create UserInfo from the already decoded token. If required fields are missing it calls the provider userinfo endpoint.

Parameters:

id_token – Verified token obtained through _get_token().
header – Request headers. The method uses Authorization.

Returns:

Parsed user information.

Raises:

InvalidRequest – If the provider request fails.

Request example

GET /auth/v2/userinfo HTTP/1.1
Host: app.example.org
Authorization: Bearer <access token>

async broker_jwks() → Dict[str, Any]#

Return the broker public key as a JWKS document.

Framework adapters expose this via a GET /.well-known/jwks.json endpoint so external services can verify broker JWTs.

Returns:: JWKS document as a plain dict.
Raises:: RuntimeError – If broker_mode is False.

Example

# Flask
@app.get("/auth/v2/.well-known/jwks.json")
async def jwks():
    return jsonify(await auth.broker_jwks())

Unified broker token endpoint — framework-agnostic entry point.

Handles all grant types supported in broker mode:

Auth code — pass code + redirect_uri (+ optional code_verifier for PKCE).
Device code — pass device_code.
Broker refresh — pass refresh_token (a previously issued broker JWT); extracts the jti, looks up the stored IDP refresh token, rotates the session and returns a new broker JWT.
RFC 8693 token exchange — pass grant_type='urn:ietf:params:oauth:grant-type:token-exchange' and subject_token=<IDP access token>; validates the IDP token and issues a broker JWT directly.

In all cases the response Token contains the broker JWT as both access_token and refresh_token.

Parameters:

token_endpoint – Full path used to compute default redirect URIs.
code – Authorization code (auth-code flow).
redirect_uri – Redirect URI for the auth-code exchange.
refresh_token – Broker JWT to refresh.
device_code – Device code for polling.
code_verifier – PKCE verifier.
subject_token – IDP access token for RFC 8693 exchange.
grant_type – Grant type; pass the RFC 8693 URN for token exchange.

Returns:

Token with broker JWT.

Raises:

InvalidRequest – On IDP errors, invalid tokens or missing args.

Example (FastAPI adapter internal call)#

token = await auth.broker_token(
    token_endpoint="/api/auth/v2/token",
    device_code="DEV-123",
)

async mint_and_store(idp_token: Token, expiry_seconds: int = 3600) → Token#

Validate an IDP token, mint a broker JWT and persist the session.

Called after any successful IDP exchange (auth-code, device-code). Validates the IDP access token claims, resolves the username, mints a broker JWT and stores the IDP refresh token for later rotation.

Parameters:

idp_token – Raw IDP token from token().
expiry_seconds – Broker JWT lifetime in seconds.

Returns:

Token carrying the broker JWT as both access_token and refresh_token.

Raises:

InvalidRequest – If IDP token validation fails.

async broker_refresh(freva_jwt: str, token_endpoint: str) → Token#

Refresh a broker session using the stored IDP refresh token.

Accepts expired broker JWTs — only the jti claim is needed to look up the session. The old session is deleted before the new one is created (rotation).

Parameters:

freva_jwt – Current broker JWT (may be expired).
token_endpoint – Token endpoint path for the IDP refresh call.

Returns:

Fresh Token with new broker JWT.

Raises:

InvalidRequest – If the JWT is unparsable or the session is gone.

async broker_exchange(subject_token: str) → Token#

RFC 8693 token exchange: validate an IDP access token, mint broker JWT.

The subject_token must be a valid IDP access token. It is verified against the IDP JWKS. On success a broker JWT is issued.

Because a plain token exchange does not yield an IDP refresh token, the resulting broker session cannot be silently refreshed via broker_refresh(). Clients that need long-lived sessions should use the device-code or auth-code flow instead.

Parameters:: subject_token – IDP access token to exchange.
Returns:: Token with broker JWT.
Raises:: InvalidRequest – If the IDP token is invalid.

Pydantic models and types used by py oidc auth.

The framework adapters return these models from the built in authentication endpoints. They are also used as types for dependency injection and decorators.

All models are compatible with OpenAPI generation in supported frameworks.

Decoded OpenID Connect token.

Standard claims are optional to allow interoperability across providers. Additional provider specific claims are preserved.

Example

token = IDToken(**payload)
print(token.sub)
print(token.get("groups"))

model_config = {'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property flattened_roles: List[str]#: Flat list of roles from any standard location.

classmethod from_token(token: str, algorithms: List[str] | None = None) → IDToken#

Create an IDToken from an encoded JWT.

Decodes without signature verification when no key is provided.

Parameters:

token – Encoded JWT string.
key – Public key or secret for signature verification.
algorithms – Accepted algorithms, e.g. ["RS256"].
audience – Expected aud claim.
issuer – Expected iss claim.

Returns:

Populated IDToken instance.

Example

# No verification
token = IDToken.from_token(encoded)

# With full verification
token = IDToken.from_token(
    encoded,
    key=public_key,
    algorithms=["RS256"],
    audience="freva-api",
)

class py_oidc_auth.schema.Token(*, access_token: str, token_type: str, expires: int, refresh_token: str, refresh_expires: int, scope: str)#

Token response returned by the token endpoint.

This model normalises common fields across providers.

Example

token = Token(
    access_token="...",
    token_type="Bearer",
    expires=1710000000,
    refresh_token="...",
    refresh_expires=1710003600,
    scope="openid profile",
)

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class py_oidc_auth.schema.DeviceStartResponse(*, device_code: str, user_code: str, verification_uri: str, verification_uri_complete: str | None = None, expires_in: int, interval: int = 5)#

Response returned when starting the device authorization flow.

The user should open verification_uri and enter user_code.

Example

start = await auth.device_flow()
print(start.verification_uri)
print(start.user_code)

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class py_oidc_auth.schema.TokenisedUser(*, pw_name: str)#

A minimal user identifier.

This model is useful in places where only a single stable identifier is required.

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class py_oidc_auth.schema.UserInfo(*, username: Annotated[str, MinLen(min_length=1)], last_name: str, first_name: str, pw_name: str, email: str | None = None)#

Normalised user profile.

Providers use different claim names for user information. The library attempts to map common claim names into this structure.

Example

info = await auth.userinfo(token, headers)
print(info.email)

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

JWT validation for OpenID Connect.

This module implements signature verification and claim validation for JWTs issued by an OpenID Connect provider.

It fetches the provider JSON Web Key Set and caches it to support key rotation.

Most users interact with this functionality indirectly through OIDCAuth.

class py_oidc_auth.token_validation.JWKSCache(jwks_uri: str, ttl: int = 3600, timeout: Timeout | None = None)#

Fetch and cache a JSON Web Key Set.

Keys are refreshed when the cache expires or when a token refers to a key id that is not currently cached.

Parameters:

jwks_uri – Provider JWKS URI.
ttl – Cache time to live in seconds.
timeout – HTTP timeout for fetching keys.

async get_key(kid: str) → Payload#

Return the JWK for a given key id.

Parameters:: kid – Key id from the JWT header.
Returns:: JWK mapping.
Raises:: KeyError – If the provider does not expose a matching key.

class py_oidc_auth.token_validation.TokenVerifier(jwks_uri: str, issuer: str | None, audience: str | None, algorithms: Sequence[str] = ('RS256',), jwks_ttl: int = 3600, timeout: Timeout | None = None)#

Validate JWTs issued by an OpenID Connect provider.

Parameters:

jwks_uri – Provider JWKS URI.
issuer – Expected issuer claim.
audience – Expected audience claim, typically your client id.
algorithms – Accepted signing algorithms.
jwks_ttl – Cache time to live for the JWKS.
timeout – HTTP timeout for fetching keys.

Example

verifier = TokenVerifier(
    jwks_uri=cfg.oidc_overview["jwks_uri"],
    issuer=cfg.oidc_overview["issuer"],
    audience=cfg.client_id,
)
token = await verifier.verify(raw_jwt)

async verify(token: str) → IDToken#

Decode and validate a JWT.

Parameters:: token – Encoded JWT without the Bearer prefix.
Returns:: Decoded token as IDToken.
Raises:: pyjwt.InvalidTokenError – On any validation failure.

Exceptions raised by py oidc auth.

This package raises a single public exception type, InvalidRequest. Integrations for web frameworks typically translate it into framework specific HTTP errors.

The exception is designed to be simple to map to an HTTP response. It carries an HTTP status code and a human readable detail message.

Example

from py_oidc_auth.exceptions import InvalidRequest

raise InvalidRequest(status_code=401, detail="Not authenticated")

exception py_oidc_auth.exceptions.InvalidRequest(status_code: int, detail: str = '')#

An error that can be represented as an HTTP response.

Parameters:

status_code – HTTP status code to return to the client.
detail – Human readable error message.

The exception is used throughout the core implementation and the framework adapters.

Example

try:
    ...
except InvalidRequest as exc:
    return {"status": exc.status_code, "detail": exc.detail}

Utility helpers and configuration.

This module contains:

OIDCConfig, a dataclass that holds all OpenID Connect settings
HTTP helper functions used by OIDCAuth
Claim and header helpers used by the framework adapters

The functions in this module are part of the public surface of the package. They are small, stable building blocks that you can use if you build your own adapter.

Look up a header or payload value with flexible key casing.

The function checks several common casing variants of key and returns the first match.

Parameters:

payload – Mapping to search in.
key – Key to look up.

Returns:

The matching value or None.

Example

authorization = process_payload(dict(request.headers), "authorization")

class py_oidc_auth.utils.OIDCConfig(client_id: str, discovery_url: str = '', client_secret: str | None = None, scopes: List[str] | None = None, audience: str | None = None, proxy: str = '', claims: Dict[str, Any] | None = None, offline_access: bool = True, timeout: Timeout | None = None)#

Configuration required to talk to an OpenID Connect provider.

Parameters:

client_id – OIDC client identifier.
discovery_url – Full URL to the discovery document.
client_secret – Client secret for confidential clients.
scopes – Default scopes.
proxy – Public base URL of your application.
claims – Optional claim constraints.
offline_access – If true, include offline_access in scope.
timeout – HTTP timeout for outbound requests.

The discovery document is fetched lazily when oidc_overview is accessed.

Example

cfg = OIDCConfig(
    client_id="my client",
    discovery_url="https://idp.example.org/.well-known/openid-configuration",
    scopes=["openid", "profile"],
)
token_endpoint = cfg.oidc_overview["token_endpoint"]

property oidc_overview: Dict[str, Dict[str, PayloadContent] | List[PayloadContent] | PayloadContent | None]#

The provider discovery document.

If the discovery document has not been loaded yet, it is fetched from discovery_url.

Returns:: Discovery document as a dict. Returns an empty dict on failure.

Notes

The HTTP client uses verify=False and follows redirects. If you require strict TLS verification, wrap this class or adjust the implementation.

class py_oidc_auth.utils.SystemUserInfo#: User information extracted from token or userinfo response.

class py_oidc_auth.utils.CacheTokenPayload#: Payload format used by cache tokens.

py_oidc_auth.utils.string_to_dict(string: str) → Dict[str, List[str]]#

Parse a simple key:value list into a dict.

The input is a comma separated list. Duplicate keys are collected into a list and duplicates are removed.

Parameters:: string – For example "key1:value1,key2:value2,key1:value2".
Returns:: A mapping like {"key1": ["value1", "value2"], "key2": ["value2"]}.

Example

assert string_to_dict("a:1,a:2,b:2") == {"a": ["1", "2"], "b": ["2"]}

py_oidc_auth.utils.token_field_matches(token: str, claims: str | Iterable[str] | Dict[str, Iterable[str]] | None = None) → bool#

Check claim constraints against an encoded JWT.

The function decodes the JWT without verifying the signature and checks that a set of claim constraints matches.

The claims argument maps a claim path to a list of acceptable values. Nested claims can be expressed using dot notation.

Parameters:

token – Encoded JWT.
claims – Mapping from claim path to acceptable values.

Returns:

True if all constraints match.

Example

ok = token_field_matches(
    token,
    claims=["admins", "offline_access"]
)

py_oidc_auth.utils.get_userinfo(user_info: Dict[str, str]) → SystemUserInfo#

Map provider specific user fields into a normalised structure.

Providers use different claim names for the same concept. This helper applies a best effort mapping.

Parameters:: user_info – Mapping created from token claims or a userinfo response.
Returns:: A normalised mapping.

Example

mapped = get_userinfo({"preferred_username": "janedoe", "mail": "a@b"})

async py_oidc_auth.utils.oidc_request(url: str, method: str, *, data: Dict[str, str] | None = None, headers: Dict[str, str] | None = None, timeout: Timeout | None = None) → Dict[str, Any]#

Make an HTTP request to an OpenID Connect provider.

Parameters:

url – Target URL.
method – HTTP method.
data – Optional form data.
headers – Optional request headers.
timeout – Optional httpx timeout.

Returns:

Response JSON.

Raises:

InvalidRequest – If the provider responds with an error or the request fails.

Example

data = {"grant_type": "refresh_token", "refresh_token": "..."}
result = await oidc_request(token_endpoint, "POST", data=data)

async py_oidc_auth.utils.query_user(token_data: Dict[str, str], authorization: str, cfg: OIDCConfig) → UserInfo#

Create UserInfo from token claims or the userinfo endpoint.

The function first attempts to build UserInfo from token claims. If required fields are missing, it calls the provider userinfo endpoint.

Parameters:

token_data – Token claims in a lower cased mapping.
authorization – Value of the Authorization header.
cfg – OIDC configuration.

Returns:

Parsed user information.

Raises:

InvalidRequest – If user information cannot be obtained.

async py_oidc_auth.utils.get_username(current_user: IDToken | None, header: Dict[str, Any], cfg: OIDCConfig) → str | None#

Return a usable username.

The function prefers explicit username fields in the token. If they are missing it tries the userinfo endpoint. As a last resort it returns the sub claim.

Parameters:

current_user – Verified token.
header – Request headers.
cfg – OIDC configuration.

Returns:

Username or None.

Example

username = await get_username(token, dict(request.headers), auth.config)

Token minting and federation#

JWT token broker — issuance, verification and federation.

TokenBroker is the orchestrator used by OIDCAuth when broker_mode=True. It:

Lazily loads or generates an RSA-2048 signing key via the configured BrokerStore.
Mints and verifies RS256 signed JWTs with a configurable aud claim.
Manages a peer JWKS cache for cross-instance token acceptance (federation). Unknown kid values trigger a rate-limited lazy refresh backed by sync httpx so that verification stays synchronous.

class py_oidc_auth.broker.issuer.TokenBroker(store: BrokerStore, issuer: str, audience: str, trusted_issuers: list[str] | None = None, jwks_path: str = '/auth/v2/.well-known/jwks.json')#

Issues and verifies broker-scoped RS256 JWTs.

Parameters:

store – Storage backend for keys, sessions and peer JWKS.
issuer – iss claim written into minted JWTs and used for issuer validation of own tokens.
audience – aud claim written into minted JWTs.
trusted_issuers – Peer instance URLs whose tokens are accepted. Peer JWKS are fetched at startup and cached.
jwks_path – Path segment appended to each peer URL when fetching JWKS. Defaults to /auth/v2/.well-known/jwks.json.

Usage:

broker = TokenBroker(
    store=create_broker_store("mongodb://localhost/myapp"),
    issuer="https://api.example.org",
    audience="my-api",
)
await broker.setup()          # idempotent, call in lifespan
token, jti = broker.mint(
    sub="janedoe",
    email="jane@example.org",
    roles=["hpcuser"],
)
claims = broker.verify(token)

async setup() → None#

Initialise the store and load all keys.

Idempotent — safe to call multiple times. Recommended inside a FastAPI lifespan handler so startup errors surface early.

property private_key: RSAPrivateKey#: The RSA private key — raises RuntimeError before setup().

mint(sub: str, email: str | None, roles: list[str], preferred_username: str | None = None, expiry_seconds: int = 3600) → tuple[str, str]#

Mint a signed broker JWT.

Parameters:

sub – Subject (human-readable username).
email – Email address or None.
roles – Flat list of role strings.
preferred_username – Display name; defaults to sub.
expiry_seconds – Token lifetime in seconds.

Returns:

(encoded_jwt, jti) tuple.

verify(token: str) → IDToken#

Verify a broker JWT and return decoded claims.

Accepts both own tokens and tokens from trusted peer instances. An unknown kid triggers a rate-limited lazy JWKS refresh.

Parameters:

token – Encoded JWT string.

Returns:

Decoded IDToken.

Raises:

pyjwt.PyJWTError – For invalid, expired or wrong-audience tokens.
pyjwt.InvalidIssuerError – For untrusted peer issuers.

jwks() → JWKSDict#: Return the public key as a JWKS document for the /.well-known/jwks.json endpoint.

async save_session(jti: str, sub: str | None, refresh_token: str, expires_at: int) → None#: Persist an IDP refresh-token session keyed by jti.

async get_session(jti: str) → tuple[str, str] | None#: Return (sub, refresh_token) or None.

async delete_session(jti: str) → None#: Remove a session entry.

async load_peer_keys() → None#: Re-fetch all peer JWKS (e.g. from a scheduled background task).

Pluggable token broker storage backends.

The BrokerStore abstract base class defines the storage interface required by TokenBroker. Choose a backend that matches your deployment:

URL prefix	Backend
`memory://`	`InMemoryBrokerStore` — testing only
`mongodb://`	`MongoDBBrokerStore` — requires `pymongo`
`sqlite+...`	`SQLAlchemyBrokerStore` — requires `sqlalchemy`
`postgresql+...`	`SQLAlchemyBrokerStore` — requires `sqlalchemy`
`mysql+...`	`SQLAlchemyBrokerStore` — requires `sqlalchemy`

Use create_broker_store() to instantiate the right backend from a URL:

store = create_broker_store("mongodb://localhost/py_oidc_auth")
store = create_broker_store("postgresql+asyncpg://user:pw@host/db")
store = create_broker_store("sqlite+aiosqlite:///~/.local/share/py-oidc-auth/broker.sqlite")
store = create_broker_store("memory://")   # testing

Multi-worker key generation:

All backends handle the signing-key creation race safely:

MongoDB uses $setOnInsert with upsert=True so only one document is ever written even if two workers race.
SQL backends catch IntegrityError from a UNIQUE constraint violation and re-read the winning worker’s key.
SQLite additionally sets PRAGMA journal_mode=WAL so concurrent readers do not block the single writer.

Session expiry:

MongoDB uses a native TTL index on the expires_at datetime field. No application-level cleanup is needed.
SQL backends perform a best-effort DELETE of expired rows on BrokerStore.setup() and verify expiry on every get_session() call. For long-running processes you may additionally schedule periodic calls to SQLAlchemyBrokerStore.purge_expired().

class py_oidc_auth.broker.store.InMemoryBrokerStore#

Ephemeral in-memory store.

Suitable for unit tests. Not safe for multiple processes — each process gets its own isolated store.

async setup() → None#: Perform in-memory setup.

async load_or_create_signing_key() → str#: Return the PEM-encoded RSA private key, creating it if absent.

async save_session(jti: str, sub: str, refresh_token: str, expires_at: int) → None#

Persist or replace an IDP refresh-token session.

Parameters:

jti – JWT ID claim from the minted freva JWT.
sub – Subject identifier.
refresh_token – IDP refresh token to store.
expires_at – Expiry as a Unix timestamp.

async get_session(jti: str) → tuple[str, str] | None#

Return (sub, refresh_token) or None if absent or expired.

Parameters:: jti – JWT ID to look up.

async delete_session(jti: str) → None#

Remove a session (token rotation or logout).

Parameters:: jti – JWT ID to remove.

async save_peer_jwks(issuer_url: str, jwks: JWKSDict) → None#

Persist peer public keys for cross-instance federation.

Parameters:

issuer_url – Canonical URL of the peer instance.
jwks – JWKS document from the peer.

async load_all_peer_jwks() → list[tuple[str, JWKSDict]]#

Return all stored peer JWKS documents.

Returns:: List of (issuer_url, jwks) tuples.

class py_oidc_auth.broker.store.MongoDBBrokerStore(url: str | None = None, db: 'AsyncDatabase[dict[str, object]]' | None = None)#

MongoDB-backed store using async pymongo.

Parameters:

url – MongoDB connection URL including database name, e.g. mongodb://user:pass@host/mydb. The database is resolved via get_default_database() so the name must be present in the URL.
db – A pre-existing AsyncDatabase instance. Use this to share an existing client. Takes precedence over url.

Requires pymongo:

pip install pymongo

async setup() → None#: Create mongoDB indexes.

async load_or_create_signing_key() → str#: Return the PEM-encoded RSA private key, creating it if absent.

async save_session(jti: str, sub: str, refresh_token: str, expires_at: int) → None#

Persist or replace an IDP refresh-token session.

Parameters:

jti – JWT ID claim from the minted freva JWT.
sub – Subject identifier.
refresh_token – IDP refresh token to store.
expires_at – Expiry as a Unix timestamp.

async get_session(jti: str) → tuple[str, str] | None#

Return (sub, refresh_token) or None if absent or expired.

Parameters:: jti – JWT ID to look up.

async delete_session(jti: str) → None#

Remove a session (token rotation or logout).

Parameters:: jti – JWT ID to remove.

async save_peer_jwks(issuer_url: str, jwks: JWKSDict) → None#

Persist peer public keys for cross-instance federation.

Parameters:

issuer_url – Canonical URL of the peer instance.
jwks – JWKS document from the peer.

async load_all_peer_jwks() → list[tuple[str, JWKSDict]]#

Return all stored peer JWKS documents.

Returns:: List of (issuer_url, jwks) tuples.

class py_oidc_auth.broker.store.SQLAlchemyBrokerStore(url: str | None = None, db: 'AsyncEngine' | None = None)#

Async SQLAlchemy store supporting PostgreSQL, MySQL and SQLite.

Parameters:

url – SQLAlchemy async connection URL.
db – A pre-existing sqlalchemy.engine instance. Use this to share an existing client. Takes precedence over url.

Requires an sqlalchemy async driver:

pip install asyncpg         # PostgreSQL
pip install aiomysql        # MySQL

SQLite specifics:

PRAGMA journal_mode=WAL is set on every connection for safe concurrent access within a single process.
SQLite does not support multiple writer processes safely even in WAL mode. For multi-process deployments use PostgreSQL or MySQL.

Session expiry

Expired rows are deleted on setup() and on every get_session() call. Call purge_expired() from a background task for long-running processes.

async setup() → None#: Create tables if absent and delete expired sessions.

async purge_expired() → int#

Delete expired session rows.

Returns:: Number of rows removed.

Call this from a background task for long-running processes.

async load_or_create_signing_key() → str#

Return the signing key PEM, creating it if absent.

Race-safe: concurrent workers that both attempt insertion will get an IntegrityError from the UNIQUE primary key constraint; the loser re-reads the winner’s key.

async save_session(jti: str, sub: str, refresh_token: str, expires_at: int) → None#

Persist or replace an IDP refresh-token session.

Parameters:

jti – JWT ID claim from the minted freva JWT.
sub – Subject identifier.
refresh_token – IDP refresh token to store.
expires_at – Expiry as a Unix timestamp.

async get_session(jti: str) → tuple[str, str] | None#

Return (sub, refresh_token) or None if absent or expired.

Parameters:: jti – JWT ID to look up.

async delete_session(jti: str) → None#

Remove a session (token rotation or logout).

Parameters:: jti – JWT ID to remove.

async save_peer_jwks(issuer_url: str, jwks: JWKSDict) → None#

Persist peer public keys for cross-instance federation.

Parameters:

issuer_url – Canonical URL of the peer instance.
jwks – JWKS document from the peer.

async load_all_peer_jwks() → list[tuple[str, JWKSDict]]#

Return all stored peer JWKS documents.

Returns:: List of (issuer_url, jwks) tuples.