CRUDAuth¶

crudauth.CRUDAuth ¶

CRUDAuth(
    *,
    session: Callable[..., Any],
    user_model: type[Any],
    SECRET_KEY: str,
    transports: Sequence[Transport] | None = None,
    column_map: dict[str, str] | None = None,
    identity: IdentityConfig | None = None,
    oauth: dict[str, Any] | None = None,
    oauth_paths: dict[str, str] | None = None,
    oauth_response_mode: Literal[
        "redirect", "json"
    ] = "redirect",
    email: Any = None,
    channels: list[DeliveryChannel] | None = None,
    hooks: AuthHooks | None = None,
    redirect_base_url: str | None = None,
    algorithm: str = DEFAULT_ALGORITHM,
    cookies: CookieConfig | None = None,
    register_schema: type[BaseModel] | None = None,
    register_extra_fields: set[str] | None = None,
    new_user_fields: NewUserFields | None = None,
    new_user_defaults: dict[str, Any] | None = None,
    rate_limiter: RateLimiterBackend | None = None,
    redis_url: str | None = None,
    redis_client: Any = None,
    rate_limits: dict[str, RateLimit] | None = None,
    lockout: LockoutConfig | None = None,
    trusted_proxy_hops: int = 0,
    sudo: SudoConfig | None = None,
    mfa: MfaConfig | None = None,
    warn_on_memory_backend: bool = True,
    password_policy: PasswordPolicy | None = None,
)

Composition root: configure transports, mount routers, gate routes.

Construct one per auth surface. It owns the user repository, the shared AuthRuntime, the rate-limiter backend, and the assembled routers. Session auth is the default; add bearer/oauth/email by passing transports=, oauth=, email=.

Example

auth = CRUDAuth(session=get_session, user_model=User, SECRET_KEY="change-me")
app.include_router(auth.router)

@app.get("/me")
async def me(user: Principal = Depends(auth.current_user())):
    return {"id": user.user_id}

Configure the auth surface.

Parameters:

Name	Type	Description	Default
`session`	`Callable[..., Any]`	FastAPI dependency that yields an `AsyncSession` (your `get_session`); every route and the `current_user` dependency acquire the DB through it.	required
`user_model`	`type[Any]`	Your SQLAlchemy user model (typically inheriting AuthUserMixin).	required
`SECRET_KEY`	`str`	Secret used to sign session/JWT and email tokens.	required
`transports`	`Sequence[Transport] \| None`	Ordered auth channels to enable; defaults to a single SessionTransport. Order is the first-wins precedence.	`None`
`column_map`	`dict[str, str] \| None`	Maps crudauth logical field names to your model's actual column names when they differ (e.g. `{"hashed_password": "pw_hash"}`).	`None`
`oauth`	`dict[str, Any] \| None`	`{provider_name: OAuthCredentials}` to enable OAuth login; requires `redirect_base_url` and a session transport.	`None`
`oauth_paths`	`dict[str, str] \| None`	Optional OAuth router paths: `prefix`, `authorize_path`, and `callback_path`. Defaults to `{"prefix": "/oauth", "authorize_path": "/{provider}/authorize", "callback_path": "/{provider}/callback"}`. Both paths must contain `{provider}`. The provider redirect URI is `redirect_base_url` plus `prefix` plus the callback path, so `redirect_base_url` must include any prefix the app adds when mounting the router.	`None`
`oauth_response_mode`	`Literal['redirect', 'json']`	`"redirect"` (default) or `"json"`. In JSON mode `authorize` returns `{"url": ...}` and `callback` returns `{"user": ..., "csrf_token": ..., "redirect_to": ...}` with the session cookies set, or a `400` on failure.	`'redirect'`
`email`	`Any`	An EmailConfig to enable verify/reset/change flows over email (the built-in delivery channel); `None` disables email delivery. Either `email` or `channels` enables the recovery endpoints.	`None`
`channels`	`list[DeliveryChannel] \| None`	Additional DeliveryChannels to route recovery tokens over (SMS, WhatsApp, push, ...). Fired alongside the email channel if `email` is also set, every channel best-effort. With `channels` and no `email`, the recovery endpoints still mount (token lifetimes fall back to the defaults). Both `email` and `channels` need a recovery factor (`identity.recovery`).	`None`
`hooks`	`AuthHooks \| None`	Lifecycle callbacks (AuthHooks).	`None`
`redirect_base_url`	`str \| None`	Public base URL used to build OAuth redirect URIs and the post-login redirect default.	`None`
`algorithm`	`str`	JWT signing algorithm (default `"HS256"`).	`DEFAULT_ALGORITHM`
`cookies`	`CookieConfig \| None`	App-wide CookieConfig (`secure` / `samesite` / `path`); transports may override per-instance.	`None`
`register_schema`	`type[BaseModel] \| None`	Custom Pydantic body for `/register`. By default only `email`/`username` are persisted; any other field is dropped unless its name is listed in `register_extra_fields`.	`None`
`register_extra_fields`	`set[str] \| None`	App-defined model columns that `/register` is allowed to set (e.g. `{"full_name", "locale"}`). Registration is an allowlist: without opting a column in here it is dropped, so adding a column to your model never silently becomes settable at signup. crudauth's privileged fields (`is_superuser`, `email_verified`, ...) can never be opted in.	`None`
`new_user_defaults`	`dict[str, Any] \| None`	Constant app columns to set on every new user, on BOTH `/register` and OAuth signup (e.g. `{"tier_id": FREE}`). The declarative shortcut for fixed values; gated like `new_user_fields` (a crudauth-owned key is dropped + warned at construction).	`None`
`new_user_fields`	`NewUserFields \| None`	Callback (sync or async) returning extra columns to set when crudauth creates a user, for values that must be derived (and may read the DB) rather than constant - e.g. `lambda ctx: {"name": ctx.suggested_name}`. Receives a trusted NewUserContext (never the request body) and returns app columns only, as a `dict` or a `BaseModel`; any crudauth logical field it returns is dropped (crudauth stays authoritative). Merged into the single insert after `new_user_defaults`, so a derived value can override a constant default. Client-typed fields belong in `register_schema` /`register_extra_fields`, not here.	`None`
`rate_limiter`	`RateLimiterBackend \| None`	Backend for lockout/throttles; defaults to an in-process MemoryRateLimiterBackend. Use `redis_rate_limiter(...)` in production.	`None`
`redis_url`	`str \| None`	Redis URL for every server-side store: sessions and CSRF tokens, the one-time-token and OAuth-state stores, and the default rate limiter. A part configured directly (a `SessionTransport` with its own `backend`/`redis_url`/`redis_client`, or `rate_limiter=`) keeps its own setting. Mutually exclusive with `redis_client`.	`None`
`redis_client`	`Any`	An existing async Redis client to use the same way as `redis_url`, with either `decode_responses` setting. The caller owns it, so `shutdown()` doesn't close it. Mutually exclusive with `redis_url`.	`None`
`rate_limits`	`dict[str, RateLimit] \| None`	Per-action overrides merged over :data:`~crudauth.ratelimit.DEFAULT_RATE_LIMITS`. Keys must be built-in actions; a custom action takes its limit in `auth.rate_limit(action, RateLimit(...))`.	`None`
`lockout`	`LockoutConfig \| None`	LockoutConfig for the escalating login lockout shared by `/login` and `/token`, whichever transports are configured. Defaults to `LockoutConfig()`.	`None`
`trusted_proxy_hops`	`int`	Number of trusted reverse proxies in front of the app. `0` (default) ignores `X-Forwarded-For` and keys per-IP rate limits / lockout on the socket peer; set to the count of proxies you control (e.g. `1` behind a single nginx/Caddy) so the real client IP is read without trusting attacker-supplied header values. See get_client_ip.	`0`
`sudo`	`SudoConfig \| None`	Enable sudo mode (short-lived re-authentication for sensitive actions) with this SudoConfig. Requires a session transport - elevation is stamped on the server-side session. Exposes `auth.sudo` and `auth.require_sudo()`.	`None`
`mfa`	`MfaConfig \| None`	Enable TOTP two-factor authentication with this MfaConfig. The model needs the MFA columns (`make_auth_identity(mfa=True)`) and `cryptography` (`crudauth[mfa]`). Exposes `auth.mfa` and the `/mfa` routes.	`None`
`warn_on_memory_backend`	`bool`	Log a startup warning when an in-memory backend is active (the zero-config default). In-memory state is per-process, so under multiple workers it silently breaks; set `False` to silence once you've accepted that (e.g. single-worker dev).	`True`
`password_policy`	`PasswordPolicy \| None`	The PasswordPolicy every new password must meet on registration, set, change and reset. The default requires at least 8 characters.	`None`

Raises:

Type	Description
`ValueError`	If `SECRET_KEY` is empty; if `oauth` or `sudo` is set without a session transport (and `oauth` also needs `redirect_base_url`); if a configured OAuth provider has no `{provider}_id` column on the user model; if `email` or `channels` is set with `identity.recovery=None`; if `rate_limits` names an unknown action; or if `lockout` is set alongside a `SessionTransport`'s `login_*` arguments.

oauth_providers `property` ¶

oauth_providers: Mapping[str, AbstractOAuthProvider]

The configured OAuth providers, by the name they were configured under.

Read-only; the provider objects carry what was resolved for them, such as an OIDC provider's discovery document.

sessions `property` ¶

sessions

The SessionManager of the configured session transport.

emails `property` ¶

emails: EmailFlowService | None

The EmailFlowService, or None when no recovery is configured.

Drives the recovery flows (request_recovery_verification, reset_password, request_email_change, ...) so a hand-written route can trigger them with the same token mint/verify the built-in endpoints use.

Example

if auth.emails is not None:
    await auth.emails.request_password_reset(db, email)

oauth `property` ¶

oauth: OAuthAccountService | None

The OAuthAccountService, or None when OAuth isn't configured.

Exposes get_or_create_user (provider-id → verified-email link → create) so a hand-written OAuth callback can reuse the linking/creation rules.

Example

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

oauth_router `property` ¶

oauth_router: APIRouter

The configured OAuth routes, for apps keeping their own auth routes.

mfa `property` ¶

mfa: MfaService | None

The MfaService, or None when MFA isn't configured.

rate_limits `property` ¶

rate_limits: dict[str, RateLimit]

The per-action limits in effect: the defaults with rate_limits= applied.

rate_limiter `property` ¶

rate_limiter: RateLimiterBackend | None

The configured rate-limit backend.

router `property` ¶

router: APIRouter

The full router to mount: shared (/register, /me) plus every transport's routes, plus OAuth and email routes when configured.

Returns:

Type	Description
`APIRouter`	An `APIRouter` to pass to `app.include_router`.

Example

app.include_router(auth.router)

session_router `property` ¶

session_router: APIRouter

Only the session transport's routes (/login, /logout).

Raises:

Type	Description
`RuntimeError`	If no SessionTransport is configured.

bearer_router `property` ¶

bearer_router: APIRouter

Only the bearer transport's routes (/token, /refresh).

Raises:

Type	Description
`RuntimeError`	If no BearerTransport is configured.

validate_password `async` ¶

validate_password(
    password: str,
    *,
    user: Any = None,
    source: PasswordSource = "set",
    field: str = "password",
) -> None

Check password against password_policy before your own code hashes it.

Raises PasswordPolicyException (422) at field when it fails. Pass the user it's for so validators taking a PasswordContext see its username and email.

resolve_principal `async` ¶

resolve_principal(
    request: Request, update_activity: bool = False
) -> Principal | None

Resolve the request principal outside FastAPI dependency injection.

This is intended for middleware and other request-level code. It tries transports in configured order, returns None for anonymous or invalid credentials, does not enforce CSRF, and does not slide sessions unless update_activity=True. A later current_user() in the same request reuses the result, reloading the user through its own session.

It opens its own DB session by calling the session dependency directly, so FastAPI's dependency_overrides don't apply to it.

authenticate_password `async` ¶

authenticate_password(
    db: Any,
    identifier: str,
    password: str,
    *,
    request: Request,
    record_success: bool = True,
) -> Any

Verify a username/email + password with the full login hardening.

The hardened credential check behind /login and /token, exposed so a hand-written login route gets the same protections (shared escalating lockout, timing-equalized verification, disabled-account check) instead of reassembling them. Returns the user row; raises RateLimitException on lockout and UnauthorizedException on bad credentials. Delegates to AuthRuntime.authenticate_password.

Example

@app.post("/my-login")
async def my_login(request: Request, form: MyForm, db=Depends(get_db)):
    user = await auth.authenticate_password(
        db, form.username, form.password, request=request
    )
    sid, csrf = await auth.sessions.create_session(request, auth.repo.user_id(user))
    ...

issue_tokens ¶

issue_tokens(
    user: Any, *, scopes: list[str] | None = None
) -> dict[str, Any]

Mint a bearer access (+refresh) token pair for a user.

The hardened issuance behind /token, exposed for a hand-written token endpoint: scopes are clamped to the transport's grantable_scopes (no self-grant) and both tokens carry the token_version epoch (so a password reset revokes them). The refresh token is returned under refresh_token (there's no Response to set a cookie on). Delegates to BearerTransport.issue_tokens.

Raises:

Type	Description
`RuntimeError`	If no BearerTransport is configured.

Example

user = await auth.authenticate_password(db, ident, pw, request=request)
tokens = auth.issue_tokens(user, scopes=["read"])

current_user ¶

current_user(
    *,
    optional: bool = False,
    superuser: bool = False,
    verified: bool = False,
    scopes: list[str] | None = None,
    transport: str | list[str] | None = None,
    check: Callable[[Principal], Any] | None = None,
) -> Callable[..., Any]

Build a FastAPI dependency that authenticates and authorizes a request.

Every gate is a keyword: optional, superuser, verified, scopes, transport (narrow to one/some transports), and check.

Note

check is a predicate (sync or async) run last on the resolved principal. Returning False denies the request with 403. To deny with a custom status/message, raise your own exception from inside check. Returning None (or anything that isn't False) allows - so both styles work: a boolean predicate (check=lambda p: p.is_superuser) and a raise-to-deny callback that simply returns nothing on success.

Note

Transports are tried in order, first credential wins. A transport returns None when its credential is absent (move to the next), but RAISES for a present-but-invalid one (e.g. a session cookie that fails the CSRF header check on a mutation). That hard-fail propagates even under optional=True - a tampered credential is an attack signal, not "treat me as anonymous".

Returns:

Type	Description
`Callable[..., Any]`	An async dependency yielding the Principal (or `None` when
`Callable[..., Any]`	`optional` and no credential is present).

Example

@app.get("/admin")
async def admin(_: Principal = Depends(auth.current_user(superuser=True))):
    ...

require_sudo ¶

require_sudo() -> Callable[..., Any]

Build a dependency that requires a current sudo elevation.

Authenticates like current_user (reusing the per-request principal cache) and then demands an unexpired sudo stamp, raising 403 otherwise. Compose it with current_user gates on the same route to also enforce identity/role:

Example

@app.post("/account/close")
async def close(
    user: Principal = Depends(auth.current_user(superuser=True)),
    _: Principal = Depends(auth.require_sudo()),
):
    ...

Raises:

Type	Description
`RuntimeError`	If sudo isn't configured (pass `sudo=SudoConfig()`).

rate_limit ¶

rate_limit(
    action: str,
    limit: RateLimit | RateLimitResolver | None = None,
    *,
    key: KeyBy | Callable[..., str] = IP,
    transport: str | list[str] | None = None,
) -> Callable[..., Any]

Build a FastAPI dependency that throttles an endpoint.

limit is a RateLimit or a sync/async function of (request, principal) returning one (None for no limit). Without it, action must be a built-in action, and its rate_limits= override or :data:~crudauth.ratelimit.DEFAULT_RATE_LIMITS entry applies. key picks who shares a budget: a KeyBy member, key(request), or key(request, principal). KeyBy.IP keys an IPv6 client by its /64 (see client_ip_key). transport narrows which credentials identify the caller, as on current_user. Writes X-RateLimit-* headers and raises RateLimitException (429) when the caller exceeds the window.

Example

@app.post("/contact", dependencies=[Depends(auth.rate_limit("contact", RateLimit(5, 60)))])
async def contact(...): ...

initialize `async` ¶

initialize() -> None

Open storage/limiter connections; call from your app's lifespan startup.

Idempotent per component. Required for server-side backends (redis); a no-op for the in-memory defaults.

Example

@asynccontextmanager
async def lifespan(app):
    await auth.initialize()
    yield
    await auth.shutdown()

shutdown `async` ¶

shutdown() -> None

Close connections. Call in lifespan teardown.

Every component is closed even when an earlier one fails; the first failure is re-raised once all of them were attempted.

CRUDAuth¶

crudauth.CRUDAuth ¶

oauth_providers property ¶

sessions property ¶

emails property ¶

oauth property ¶

oauth_router property ¶

mfa property ¶

rate_limits property ¶

rate_limiter property ¶

router property ¶

session_router property ¶

bearer_router property ¶

validate_password async ¶

resolve_principal async ¶

authenticate_password async ¶

issue_tokens ¶

current_user ¶

require_sudo ¶

rate_limit ¶

initialize async ¶

shutdown async ¶

oauth_providers `property` ¶

sessions `property` ¶

emails `property` ¶

oauth `property` ¶

oauth_router `property` ¶

mfa `property` ¶

rate_limits `property` ¶

rate_limiter `property` ¶

router `property` ¶

session_router `property` ¶

bearer_router `property` ¶

validate_password `async` ¶

resolve_principal `async` ¶

authenticate_password `async` ¶

initialize `async` ¶

shutdown `async` ¶