Zum Inhalt

Overview

package wittgenstein_oidc_backend

wittgenstein-oidc-backend — shared Keycloak/OIDC backend toolkit.

from wittgenstein_oidc_backend import decode_token, has_role, get_roles

Classes

  • OidcMiddleware — ASGI middleware: decodes the bearer token (if present) with the given callable and makes the resulting payload available to get_current_payload()/has_role for the rest of the request.

  • JwksClient — Fetches and caches a Keycloak realm's JWKS document.

  • TokenMissingIdentityError — The token has neither an email (human) nor an azp (service-account client id) claim — there is nothing to match or create a user by.

  • UserStore — The minimal interface a consuming app implements against its own user/role schema for jit_provision_user() to drive.

  • TokenError — Raised for any token decode/verification failure — callers should treat this uniformly (401/403), never let the underlying jwt/httpx exception leak into a response.

Functions

  • create_oidc_bff_router — Build the /oidc router for one Keycloak realm.

  • get_groups — Raw Keycloak group paths (e.g. ['/12345678901234']) — callers that need to derive something from a group name (like a tax-id-from-group convention) do that parsing themselves.

  • get_roles — Realm roles (realm_access.roles) — the only place Keycloak puts a role NAME in the token (see wittgenstein-core's own KEYCLOAK_KNOWN_ROLE_NAMES comment); permission strings for each name are a local concern of whoever maps roles to permissions.

  • get_current_payload — The decoded token payload for the request currently being handled, or None if no token was presented/decoded.

  • set_current_payload

  • has_role — Require the caller's token to carry at least one of roles (checked against the Keycloak realm roles claim, see claims.get_roles). Raises 403 if the token is missing, invalid, or lacks all of the given roles — never a 500.

  • jit_provision_user — Find-or-create the local user for this Keycloak identity, and sync its roles from the token's realm roles.

  • decode_token — Decode a Keycloak-issued JWT and return its claims payload.

wittgenstein_oidc_backend.create_oidc_bff_router

mkapi_definition_mkapi create_oidc_bff_router(authority: str, client_id: str, *, client_secret: str | None = None, cookie_name: str = DEFAULT_COOKIE_NAME, cookie_path: str = '/oidc', cookie_secure: bool = True, cookie_samesite: str = 'lax', cookie_max_age: int | None = None, allowed_origins: list[str] | None = None, timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None) → APIRouter

Build the /oidc router for one Keycloak realm.

authority is the realm URL (e.g. https://auth.example.com/realms/acme); client_id is the SPA's realm client, and client_secret its secret when the client is confidential. transport is injectable for tests only (httpx.MockTransport) — real callers omit it.

allowed_origins (when set) fences the refresh grant: a request whose Origin header is present and not listed is refused. An absent Origin is allowed (same-origin and non-browser callers do not always send one; SameSite=Lax already fences the cookie).

wittgenstein_oidc_backend.get_groups

mkapi_definition_mkapi get_groups(payload: dict[str, Any]) → list[str]

Raw Keycloak group paths (e.g. ['/12345678901234']) — callers that need to derive something from a group name (like a tax-id-from-group convention) do that parsing themselves.

wittgenstein_oidc_backend.get_roles

mkapi_definition_mkapi get_roles(payload: dict[str, Any]) → list[str]

Realm roles (realm_access.roles) — the only place Keycloak puts a role NAME in the token (see wittgenstein-core's own KEYCLOAK_KNOWN_ROLE_NAMES comment); permission strings for each name are a local concern of whoever maps roles to permissions.

wittgenstein_oidc_backend.OidcMiddleware

mkapi_definition_mkapi class OidcMiddleware(app: Any, decode: Any)

ASGI middleware: decodes the bearer token (if present) with the given callable and makes the resulting payload available to get_current_payload()/has_role for the rest of the request.

decode is expected to be a zero-arg-besides-token async callable already bound to a mode/issuer/jwks_client — see wittgenstein_oidc_backend.tokens.decode_token partially applied, e.g. via functools.partial. Decode failures are swallowed here (payload stays None) — routes that require auth still get a 401/403 from has_role; routes that don't require it keep working unauthenticated (config-gated integrations must never hard-fail on a missing/bad token if the route itself doesn't demand one).

wittgenstein_oidc_backend.get_current_payload

mkapi_definition_mkapi get_current_payload() → dict[str, Any] | None

The decoded token payload for the request currently being handled, or None if no token was presented/decoded.

wittgenstein_oidc_backend.set_current_payload

mkapi_definition_mkapi set_current_payload(payload: dict[str, Any] | None) → None

wittgenstein_oidc_backend.has_role

mkapi_definition_mkapi has_role(roles: list[str]) → Callable[[F], F]

Require the caller's token to carry at least one of roles (checked against the Keycloak realm roles claim, see claims.get_roles). Raises 403 if the token is missing, invalid, or lacks all of the given roles — never a 500.

wittgenstein_oidc_backend.JwksClient

mkapi_definition_mkapi class JwksClient(jwks_url: str, *, cache_ttl: int = _DEFAULT_CACHE_TTL)

Fetches and caches a Keycloak realm's JWKS document.

One instance per realm/issuer is expected to be shared across requests (it is not a per-request object) — construct it once at app startup.

Methods

wittgenstein_oidc_backend.JwksClient.get_keys

mkapi_definition_mkapi async method JwksClient.get_keys(self) → list[dict[str, Any]]

wittgenstein_oidc_backend.JwksClient.get_key

mkapi_definition_mkapi async method JwksClient.get_key(self, kid: str) → dict[str, Any] | None

wittgenstein_oidc_backend.TokenMissingIdentityError

mkapi_definition_mkapi class TokenMissingIdentityError()

Bases : Exception

The token has neither an email (human) nor an azp (service-account client id) claim — there is nothing to match or create a user by.

wittgenstein_oidc_backend.UserStore

mkapi_definition_mkapi class UserStore()

Bases : Protocol

The minimal interface a consuming app implements against its own user/role schema for jit_provision_user() to drive.

Methods

  • find_by_email

  • find_by_client_id

  • link_keycloak_identity — Update an EXISTING user's Keycloak identity fields. Called on every match (not just first-time creation) so an account that existed before SSO was enabled gets linked to the Keycloak identity that first authenticates as it, instead of staying unlinked.

  • create_user

  • sync_roles — Reconcile the user's local role assignments from the RAW realm roles on the token. The store decides which of role_names are recognized (e.g. intersecting with its own Role table before assigning — wittgenstein-core's KEYCLOAK_KNOWN_ROLE_NAMES is one such filter) and reconciles to that set (add missing, remove extra). Called for both a freshly-provisioned user and an existing one on every token verification, so a role change in Keycloak takes effect without recreating the account.

wittgenstein_oidc_backend.UserStore.find_by_email

mkapi_definition_mkapi method UserStore.find_by_email(email: str) → Any | None

wittgenstein_oidc_backend.UserStore.find_by_client_id

mkapi_definition_mkapi method UserStore.find_by_client_id(client_id: str) → Any | None

mkapi_definition_mkapi method UserStore.link_keycloak_identity(user: Any, *, keycloak_sub: str, keycloak_client_id: str | None) → None

Update an EXISTING user's Keycloak identity fields. Called on every match (not just first-time creation) so an account that existed before SSO was enabled gets linked to the Keycloak identity that first authenticates as it, instead of staying unlinked.

wittgenstein_oidc_backend.UserStore.create_user

mkapi_definition_mkapi method UserStore.create_user(*, email: str | None, username: str, full_name: str | None, keycloak_sub: str, keycloak_client_id: str | None, is_service_account: bool) → Any

wittgenstein_oidc_backend.UserStore.sync_roles

mkapi_definition_mkapi method UserStore.sync_roles(user: Any, role_names: set[str]) → None

Reconcile the user's local role assignments from the RAW realm roles on the token. The store decides which of role_names are recognized (e.g. intersecting with its own Role table before assigning — wittgenstein-core's KEYCLOAK_KNOWN_ROLE_NAMES is one such filter) and reconciles to that set (add missing, remove extra). Called for both a freshly-provisioned user and an existing one on every token verification, so a role change in Keycloak takes effect without recreating the account.

wittgenstein_oidc_backend.jit_provision_user

mkapi_definition_mkapi jit_provision_user(payload: dict[str, Any], store: UserStore) → Any

Find-or-create the local user for this Keycloak identity, and sync its roles from the token's realm roles.

Human tokens are matched/created by email (so accounts that existed before SSO was enabled get linked instead of duplicated). Service-account tokens (Keycloak client-credentials grant — no email, only azp) are matched/created by keycloak_client_id; they carry no realm roles, so role sync is skipped for them (the store decides what a service account's default access looks like).

Raises

wittgenstein_oidc_backend.TokenError

mkapi_definition_mkapi class TokenError()

Bases : Exception

Raised for any token decode/verification failure — callers should treat this uniformly (401/403), never let the underlying jwt/httpx exception leak into a response.

wittgenstein_oidc_backend.decode_token

mkapi_definition_mkapi async decode_token(token: str, *, mode: OidcMode, jwks_client: JwksClient | None = None, issuer: str | None = None, audience: str | None = None) → dict[str, Any]

Decode a Keycloak-issued JWT and return its claims payload.

mode="validate" requires jwks_client and issuer. mode="trust" needs neither — it only decodes.

Raises