Skip to content

MFA

TOTP two-factor authentication. Configure with mfa=MfaConfig(...) on CRUDAuth; the service is auth.mfa. See the guide.

crudauth.mfa.MfaConfig dataclass 

MfaConfig(
    issuer: str,
    encryption_key: str | bytes | Sequence[str | bytes],
    required: MfaRequirement = False,
    oauth: bool = False,
    challenge_ttl_seconds: int = DEFAULT_CHALLENGE_TTL_SECONDS,
    max_code_attempts: int = DEFAULT_MAX_CODE_ATTEMPTS,
    recovery_code_count: int = RECOVERY_CODE_COUNT,
)

Opt-in TOTP (authenticator app) second factor, passed as CRUDAuth(mfa=...).

Parameters:

Name Type Description Default
issuer str

Name the authenticator app shows next to the account.

 required
encryption_key str | bytes | Sequence[str | bytes]

Fernet key that encrypts TOTP secrets at rest, or a list of keys for rotation (the first encrypts, any decrypts). Must differ from SECRET_KEY.

 required
required MfaRequirement

Whether an account must use MFA: False (default), True, or a sync/async predicate over the user row. A required account that isn't enrolled enrolls during login.

 False
oauth bool

Also challenge OAuth logins. Off by default, since the identity provider owns the second factor there.

 False
challenge_ttl_seconds int

How long a login challenge can be answered.

 DEFAULT_CHALLENGE_TTL_SECONDS
max_code_attempts int

Wrong codes a challenge survives.

 DEFAULT_MAX_CODE_ATTEMPTS
recovery_code_count int

Recovery codes issued at enrollment and on regeneration.

 RECOVERY_CODE_COUNT

Raises:

Type Description
ValueError

If issuer or encryption_key is empty, or a number isn't positive.
Example 
auth = CRUDAuth(
    ...,
    mfa=MfaConfig(
        issuer="Acme",
        encryption_key=os.environ["MFA_ENCRYPTION_KEY"],
        required=lambda user: user.is_superuser,
    ),
)

encryption_keys property 

encryption_keys: list[str]

encryption_key as a list of strings, the encrypting key first.

crudauth.mfa.MfaRequirement module-attribute 

MfaRequirement = (
    bool | Callable[[Any], bool | Awaitable[bool]]
)

True, False, or a sync/async predicate over the user row.

crudauth.mfa.MfaService 

MfaService(
    *,
    runtime: AuthRuntime,
    config: MfaConfig,
    challenge_store: AbstractSessionStorage[MfaChallenge],
)

TOTP enrollment and verification, reachable as auth.mfa.

/login and /token call challenge_login after a correct password; /mfa/verify calls complete_challenge. A hand-written login does the same to keep the second factor.

Example 
user = await auth.authenticate_password(
    db, form.username, form.password, request=request, record_success=False
)
challenge = await auth.mfa.challenge_login(
    db, user, request=request, transport="session",
    lockout_identifier=form.username, options={},
)
if challenge is not None:
    return challenge

is_enrolled 

is_enrolled(user: Any) -> bool

Whether the account has a confirmed authenticator.

is_required async 

is_required(user: Any) -> bool

Whether MfaConfig.required applies to this account.

status async 

status(user: Any) -> dict[str, Any]

{"enabled", "required", "recovery_codes_remaining"} for GET /mfa.

verify_totp async 

verify_totp(db: AsyncSession, user: Any, code: str) -> bool

Check a code from the enrolled authenticator, claiming its time step.

A step is accepted once, so the same code (or an older one) fails afterwards.

verify_code async 

verify_code(
    db: AsyncSession,
    user: Any,
    code: str,
    *,
    context: HookContext | None = None,
) -> bool

Check an authenticator code, or else consume a recovery code.

Fires on_after_recovery_code_used when a recovery code is spent.

begin_setup async 

begin_setup(db: AsyncSession, user: Any) -> dict[str, str]

Start (or restart) enrollment with a new secret.

Returns:

Type Description
dict[str, str]

{"secret", "otpauth_uri"} for the authenticator app.

Raises:

Type Description
BadRequestException

If an authenticator is already enabled.

confirm_setup async 

confirm_setup(
    db: AsyncSession,
    user: Any,
    code: str,
    *,
    context: HookContext | None = None,
) -> list[str]

Enable the pending authenticator once a code from it checks out.

Returns:

Type Description
list[str]

The recovery codes, in plain text, for the user to store. They aren't
list[str]

retrievable later.

Raises:

Type Description
BadRequestException

If MFA is already enabled or no setup is pending.
UnauthorizedException

If the code is wrong.

disable async 

disable(
    db: AsyncSession,
    user: Any,
    *,
    context: HookContext | None = None,
) -> None

Remove the authenticator and recovery codes (an admin reset calls this directly).

regenerate_recovery_codes async 

regenerate_recovery_codes(
    db: AsyncSession, user: Any
) -> list[str]

Replace every recovery code with a new set.

Raises:

Type Description
BadRequestException

If MFA isn't enabled.

challenge_login async 

challenge_login(
    db: AsyncSession,
    user: Any,
    *,
    request: Request,
    transport: str,
    lockout_identifier: str,
    options: dict[str, Any],
) -> dict[str, Any] | None

Start the second step of a login whose password checked out.

Parameters:

Name Type Description Default
db AsyncSession

Active async session.

 required
user Any

The authenticated user row.

 required
request Request

The login request.

 required
transport str

Name of the transport that finishes the login.

 required
lockout_identifier str

The login identifier the lockout is keyed on; a correct code clears it, a wrong one counts against it.

 required
options dict[str, Any]

What the transport's complete_login needs afterwards.

 required

Returns:

Type Description
dict[str, Any] | None

None when the account neither uses nor requires MFA (the lockout is
dict[str, Any] | None

cleared, and the caller completes the login). Otherwise
dict[str, Any] | None

{"mfa_required": True, "challenge", "expires_in"}, plus
dict[str, Any] | None

setup: {"secret", "otpauth_uri"} when a required account enrolls now.

describe_challenge async 

describe_challenge(
    db: AsyncSession, token: str
) -> dict[str, Any]

{"setup": {"secret", "otpauth_uri"} | None} for a live challenge.

For a frontend that received only the challenge token (an OAuth redirect) and needs the enrollment details. Doesn't count as an attempt.

Raises:

Type Description
BadRequestException

If the challenge doesn't exist.

complete_challenge async 

complete_challenge(
    db: AsyncSession,
    token: str,
    code: str,
    *,
    request: Request,
    response: Response,
) -> dict[str, Any]

Check the code for a challenge and issue the credential its login started.

An authenticator code or a recovery code answers a login challenge; a setup challenge needs a code from the authenticator being enrolled, and the response then carries its recovery_codes. Each call counts toward the challenge's max_code_attempts and the login lockout; a correct code consumes the challenge and clears the lockout.

Returns:

Type Description
dict[str, Any]

The transport's login response, plus recovery_codes after
dict[str, Any]

enrollment and redirect_to for an OAuth login.

Raises:

Type Description
BadRequestException

Unknown, expired, used-up or exhausted challenge, or the password was reset or changed since the challenge began.
ForbiddenException

A cross-site request for a cookie-setting login.
RateLimitException

The login lockout is engaged.
UnauthorizedException

Wrong code.