OAuth¶

OAuth 2.0 social login. Enable it with oauth={...} on CRUDAuth. An OpenID Connect provider needs only OAuthCredentials(issuer=...), which configures GenericOIDCProvider; anything else is added by implementing AbstractOAuthProvider and registering it with OAuthProviderFactory.

crudauth.oauth.OAuthCredentials ¶

Bases: BaseModel

Client credentials for a provider, supplied via oauth={...}.

Leave client_secret empty for a public client (PKCE only). Set issuer for any OpenID Connect provider that isn't built in (Keycloak, Zitadel, Authentik, Auth0, Okta, Entra ID, ...): its endpoints are then read from {issuer}/.well-known/openid-configuration when the app starts, instead of the name being looked up in OAuthProviderFactory.

Example

oauth={
    "google": OAuthCredentials(client_id=..., client_secret=...),
    "keycloak": OAuthCredentials(
        client_id=..., client_secret=..., issuer="https://sso.example.com/realms/main"
    ),
}

crudauth.oauth.OAuthUserInfo ¶

Bases: BaseModel

Normalized profile returned by AbstractOAuthProvider.process_user_info.

crudauth.oauth.GenericOIDCProvider ¶

GenericOIDCProvider(
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    *,
    issuer: str,
    provider_name: str = "oidc",
    scopes: list[str] | None = None,
    authorize_endpoint: str = "",
    token_endpoint: str = "",
    userinfo_endpoint: str = "",
    transport: Any | None = None,
)

Bases: AbstractOAuthProvider

An OIDC provider whose endpoints come from its issuer's discovery document.

Configure it through oauth= by giving the credentials an issuer:

Example

auth = CRUDAuth(
    ...,
    oauth={
        "keycloak": OAuthCredentials(
            client_id="...",
            client_secret="...",
            issuer="https://sso.example.com/realms/main",
        )
    },
    redirect_base_url="https://app.example.com",
)

The key you use is the provider name, so it's also the column the account is linked on (keycloak -> keycloak_id) and the path segment in /oauth/keycloak/authorize. process_user_info reads the standard OIDC claims, which every conformant provider returns from /userinfo.

Note

The provider is unusable until initialize() has resolved the endpoints. CRUDAuth.initialize() does that, which is the same lifespan call Redis-backed storage already needs.

Build an unresolved provider; initialize() fills in the endpoints.

Parameters:

Name	Type	Description	Default
`issuer`	`str`	The provider's issuer identifier, e.g. `https://sso.example.com/realms/main`. A trailing slash is ignored.	required
`provider_name`	`str`	The linked-account name, so `"keycloak"` stores its account id in `keycloak_id`.	`'oidc'`
`scopes`	`list[str] \| None`	Override the default `openid profile email`.	`None`
`authorize_endpoint`	`str`	Skip discovery for this endpoint by passing it (along with the other two) explicitly.	`''`
`token_endpoint`	`str`	See `authorize_endpoint`.	`''`
`userinfo_endpoint`	`str`	See `authorize_endpoint`.	`''`
`transport`	`Any \| None`	An `httpx` transport for every outbound call, for a proxy, a client certificate, or a test double.	`None`

from_discovery `async` `classmethod` ¶

from_discovery(
    issuer: str,
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    *,
    provider_name: str = "oidc",
    scopes: list[str] | None = None,
    transport: Any | None = None,
) -> GenericOIDCProvider

Build a provider and resolve its endpoints in one await.

For code that owns its providers directly (a hand-written OAuth route, a script). Through oauth= you don't need this: CRUDAuth constructs the provider and initialize() resolves it.

initialize `async` ¶

initialize() -> None

Fetch the discovery document and resolve the endpoints (idempotent).

Raises:

Type	Description
`ValueError`	The document is missing a required endpoint, or declares an issuer other than the configured one.
`HTTPStatusError`	The discovery endpoint returned an error status.

process_user_info `async` ¶

process_user_info(
    user_info: dict[str, Any],
) -> OAuthUserInfo

Normalize the standard OIDC userinfo claims into OAuthUserInfo.

sub is the account's stable identifier, so a response without one is an error rather than a user linked to the string "None". email_verified is passed through as the provider stated it: a verified provider email is what lets the callback link the profile to an existing local account, so a provider that doesn't claim verification doesn't get that.

crudauth.oauth.AbstractOAuthProvider ¶

AbstractOAuthProvider(
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    *,
    scopes: list[str],
    authorize_endpoint: str,
    token_endpoint: str,
    userinfo_endpoint: str,
    provider_name: str,
    transport: Any | None = None,
)

Bases: ABC

Port for an OAuth provider - implements the Authorization-Code-with-PKCE flow.

Subclass it, pass the three endpoints + scopes + provider_name to super().__init__, implement process_user_info, and register it with OAuthProviderFactory. Set email_verified honestly - auto-linking to an existing account requires a verified provider email.

Set requires_client_secret = True on a provider that never accepts a public client, so a missing secret fails at startup instead of at the first login.

Note

A custom provider named "gitlab" requires a gitlab_id column on your user model (that's where its account id is stored and matched). Add it to your model (or map it via column_map=); CRUDAuth raises at startup if a configured provider has no {provider}_id column. Only google_id/github_id ship on AuthUserMixin.

Example

class GitLabOAuthProvider(AbstractOAuthProvider):
    def __init__(self, client_id, client_secret, redirect_uri, scopes=None):
        super().__init__(
            client_id, client_secret, redirect_uri,
            scopes=scopes or ["read_user"],
            authorize_endpoint="https://gitlab.com/oauth/authorize",
            token_endpoint="https://gitlab.com/oauth/token",
            userinfo_endpoint="https://gitlab.com/api/v4/user",
            provider_name="gitlab",
        )

    async def process_user_info(self, info):
        return OAuthUserInfo(
            provider="gitlab", provider_user_id=str(info["id"]),
            email=info.get("email"), email_verified=True, raw_data=info,
        )

OAuthProviderFactory.register_provider("gitlab", GitLabOAuthProvider)
# ...and add `gitlab_id` to your user model.

initialize `async` ¶

initialize() -> None

Resolve whatever the provider needs before it can serve a login.

A no-op for a provider with static endpoints. CRUDAuth.initialize awaits this for every configured provider, so a provider that discovers its endpoints can do it there.

generate_state `staticmethod` ¶

generate_state() -> str

Return a fresh, URL-safe CSRF state value.

generate_pkce_codes `staticmethod` ¶

generate_pkce_codes() -> dict[str, str]

Return a PKCE pair: {"code_verifier": ..., "code_challenge": ...} (S256).

get_authorization_url ¶

get_authorization_url(
    state: str | None = None,
    pkce: bool = True,
    extra_params: dict[str, str] | None = None,
) -> dict[str, str]

Build the provider authorization URL and the values to stash server-side.

Parameters:

Name	Type	Description	Default
`state`	`str \| None`	CSRF state to embed; generated if omitted.	`None`
`pkce`	`bool`	Include a PKCE challenge (recommended).	`True`
`extra_params`	`dict[str, str] \| None`	Provider-specific query params to merge in (e.g. Google's `access_type`/`prompt`).	`None`

Returns:

Type	Description
`dict[str, str]`	`{"url": <redirect target>, "state": ..., "code_verifier": ...}` -
`dict[str, str]`	`code_verifier` is present only when `pkce` is true and must be
`dict[str, str]`	persisted to verify the callback.

exchange_code `async` ¶

exchange_code(
    code: str,
    code_verifier: str | None = None,
    headers: dict[str, str] | None = None,
) -> dict[str, Any]

Exchange an authorization code for tokens at the token endpoint.

Parameters:

Name	Type	Description	Default
`code`	`str`	The authorization code from the callback.	required
`code_verifier`	`str \| None`	The stored PKCE verifier (required if PKCE was used).	`None`
`headers`	`dict[str, str] \| None`	Extra request headers (some providers need `Accept`).	`None`

Returns:

Type	Description
`dict[str, Any]`	The provider's raw token response (`access_token`, ...).

Raises:

Type	Description
`HTTPStatusError`	If the token endpoint returns an error status.

Note

client_secret is sent only when set, so a public client (PKCE only) sends no client authentication. It rides in the form body (client_secret_post) unless token_auth_method says the provider wants HTTP Basic (client_secret_basic).

get_user_info `async` ¶

get_user_info(access_token: str) -> dict[str, Any]

Fetch the raw user profile from the userinfo endpoint.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	A valid provider access token.	required

Returns:

Type	Description
`dict[str, Any]`	The provider's raw profile JSON (normalize it in
`dict[str, Any]`	process_user_info).

Raises:

Type	Description
`HTTPStatusError`	If the userinfo endpoint returns an error status.

process_user_info `abstractmethod` `async` ¶

process_user_info(
    user_info: dict[str, Any],
) -> OAuthUserInfo

Normalize the provider's raw user payload into OAuthUserInfo.

crudauth.oauth.OAuthProviderFactory ¶

Process-wide registry of provider name -> provider class.

Built-in providers ("google", "github") register themselves when [crudauth.oauth][] is imported; register your own with register_provider.

Example

OAuthProviderFactory.register_provider("gitlab", GitLabOAuthProvider)

register_provider `classmethod` ¶

register_provider(
    provider_name: str,
    provider_class: type[AbstractOAuthProvider],
) -> None

Register provider_class under provider_name (overwrites any existing).

get_provider_class `classmethod` ¶

get_provider_class(
    provider_name: str,
) -> type[AbstractOAuthProvider] | None

Return the registered class for provider_name, or None.

create_provider `classmethod` ¶

create_provider(
    provider_name: str,
    client_id: str,
    client_secret: str,
    redirect_uri: str,
    scopes: list[str] | None = None,
) -> AbstractOAuthProvider

Instantiate a registered provider.

Parameters:

Name	Type	Description	Default
`provider_name`	`str`	A registered provider name.	required
`client_id`	`str`	OAuth client id.	required
`client_secret`	`str`	OAuth client secret.	required
`redirect_uri`	`str`	The callback URI registered with the provider.	required
`scopes`	`list[str] \| None`	Override the provider's default scopes.	`None`

Returns:

Type	Description
`AbstractOAuthProvider`	A configured AbstractOAuthProvider.

Raises:

Type	Description
`ValueError`	If `provider_name` isn't registered. This is a config-time error (raised while building the app), distinct from the request-time `BadRequestException` the OAuth router raises for an unknown provider in a URL - different layers, deliberately different error types.

crudauth.oauth.OAuthAccountService ¶

OAuthAccountService(
    repo: UserRepository,
    new_user_fields: NewUserFields | None = None,
    new_user_defaults: dict[str, Any] | None = None,
    *,
    session_manager: "SessionManager | None" = None,
)

Resolve an OAuth identity to a user, creating or linking as needed.

The linking rules live here (lookup order: provider id → verified email → create), so a hand-written callback reuses them. Reachable as auth.oauth (None when OAuth isn't configured). session_manager is used to sign out the sessions of an unverified account that a provider claims.

Example

if auth.oauth is not None:
    user, created = await auth.oauth.get_or_create_user(info, db)

get_or_create_user `async` ¶

get_or_create_user(
    info: OAuthUserInfo, db: AsyncSession
) -> tuple[Any, bool]

Resolve an OAuth identity to a user; lookup order: provider id → email → create.

Returns:

Type	Description
`Any`	`(user, created)` - `created` is `True` only when a new row was
`bool`	inserted (provider-id and email-link hits return the existing user).

Raises:

Type	Description
`OAuthAccountException`	When the provider gives no email, an unverified email, an email longer than the `email` column, or the matching account is already linked to a different account of the same provider.

Note

Only a verified provider email links or creates an account. Linking to an account whose own email was never verified claims it: its password becomes unusable, any two-factor enrollment is removed, its token_version is bumped, its sessions are signed out, and its email is marked verified.

OAuth¶

crudauth.oauth.OAuthCredentials ¶

crudauth.oauth.OAuthUserInfo ¶

crudauth.oauth.GenericOIDCProvider ¶

from_discovery async classmethod ¶

initialize async ¶

process_user_info async ¶

crudauth.oauth.AbstractOAuthProvider ¶

initialize async ¶

generate_state staticmethod ¶

generate_pkce_codes staticmethod ¶

get_authorization_url ¶

exchange_code async ¶

get_user_info async ¶

process_user_info abstractmethod async ¶

crudauth.oauth.OAuthProviderFactory ¶

register_provider classmethod ¶

get_provider_class classmethod ¶

create_provider classmethod ¶

crudauth.oauth.OAuthAccountService ¶

get_or_create_user async ¶

from_discovery `async` `classmethod` ¶

initialize `async` ¶

process_user_info `async` ¶

initialize `async` ¶

generate_state `staticmethod` ¶

generate_pkce_codes `staticmethod` ¶

exchange_code `async` ¶

get_user_info `async` ¶

process_user_info `abstractmethod` `async` ¶

register_provider `classmethod` ¶

get_provider_class `classmethod` ¶

create_provider `classmethod` ¶

get_or_create_user `async` ¶