OAuth¶
OAuth 2.0 social login. Enable it with oauth={...} on CRUDAuth. Add a
provider by implementing AbstractOAuthProvider and registering it with
OAuthProviderFactory.
crudauth.oauth.OAuthCredentials
¶
Bases: BaseModel
Client credentials for a provider, supplied via oauth={...}.
crudauth.oauth.OAuthUserInfo
¶
Bases: BaseModel
Normalized profile returned by AbstractOAuthProvider.process_user_info.
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,
)
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.
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.
generate_pkce_codes
staticmethod
¶
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
|
None
|
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
|
dict[str, str]
|
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The provider's raw token response ( |
Raises:
| Type | Description |
|---|---|
HTTPStatusError
|
If the token endpoint returns an error status. |
get_user_info
async
¶
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]
|
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.
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 |
crudauth.oauth.OAuthAccountService
¶
OAuthAccountService(repo: UserRepository)
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
|
|
bool
|
inserted (provider-id and email-link hits return the existing user). |
Raises:
| Type | Description |
|---|---|
BadRequestException
|
When an unverified email matches an existing account (see the linking Note), or no email is available to create an account. |
Note
Auto-linking to an existing account requires info.email_verified.
Attaching a provider to an account on an unverified,
attacker-influenceable email is an account-takeover vector, so an
unverified email matching an existing account is refused and routed
to manual linking.
Note
Creating a new account is deliberately allowed on an unverified
email - there is no existing account to hijack - but the row is
created with email_verified=False so it is never treated as
proven. Linking is the asymmetric case precisely because it touches
an account the OAuth user may not own.