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

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 = 'openid profile email', proxy: str = '', claims: Dict[str, Any] | None = None, timeout_sec: int = 10)#

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.
timeout_sec – HTTP timeout for discovery and provider calls.

Example

auth = OIDCAuth(
    client_id="my client",
    discovery_url="https://idp.example.org/.well-known/openid-configuration",
    client_secret="secret",
    scopes="openid profile email",
    proxy="https://app.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>

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].

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, audience: str, 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, proxy: str = '', claims: Dict[str, Any] | None = None, 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.
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: Dict[str, Any] | 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={
        "groups": ["admins"],
        "realm_access.roles": ["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)