API keys, tokens and codes

Passwords, API keys stored as digests, revoked tokens, one-time codes, the second factor (TOTP), passkeys (WebAuthn) and step-up. In fastapi_alcazaba.auth.

In the guide: Passwords, API keys, Revoked tokens and one-time codes, Second factor, Passkeys. In an app: the example, the second factor's.

fastapi_alcazaba.auth.FastAPIPasswords

FastAPIPasswords(*, parameters: ScryptParameters | None = None, pepper: str | SecretStr | None = None, min_length: int = 15, max_attempts: int = 10, lockout: int = 900, reset_ttl: int = 3600, pwned: PwnedPasswords | None = None, cache: Cache | None = None, namespace: str = 'passwords', audit: FastAPIAudit | None = None)

Passwords, from sign-up to reset.

• Hashes: scrypt with a random salt each, in a thread (it takes a fifth of a second on purpose); with a pepper, the password is first an HMAC of it, so that a stolen database without the pepper gives nothing to guess against.
• Guesses: after max_attempts wrong passwords for an account, its logins are refused for lockout seconds, the right password too; the attempts of other accounts aren't affected. Put a rate limit per client on the login route too, for one client trying many accounts.
• Unknown accounts take as long as known ones: verify hashes against a made-up hash.
• Leaked passwords: with pwned (PwnedPasswords), check says when a password is among those leaked, which attackers try first.
• Resets: reset_token makes a random token, kept as a digest for reset_ttl seconds, that use_reset_token turns into its user once; a new one replaces the user's last.

Parameters:

Name	Type	Description	Default
parameters	ScryptParameters | None	scrypt's cost.	None
pepper	str | SecretStr | None	a secret of 32 bytes or more, kept out of the database (your settings). It can't be rotated without the passwords: keep it.	None
min_length	int	characters a new password needs (NIST SP 800-63B: 15 when the password is the only factor, 8 with another).	15
max_attempts	int	wrong passwords per account before the lockout.	10
lockout	int	seconds an account's logins are refused after them.	900
reset_ttl	int	seconds a reset token lasts.	3600
pwned	PwnedPasswords | None	checks the passwords against those leaked; None for no check.	None
cache	Cache | None	the Cache whose backend keeps the attempts and the reset tokens; FastAPICache's by default.	None
namespace	str	part of the keys, after the cache's prefix (see Cache.state_key).	'passwords'
audit	FastAPIAudit | None	the audit trail that records each password checked (auth.password) and each reset (auth.password_reset), with their outcome.	None

hash async

hash(password: str) -> str

The hash of password to store, in PHC's format. Raises ValueError for a password longer than MAX_PASSWORD: check tells it first.

needs_rehash

needs_rehash(stored: str) -> bool

Whether stored was made with other parameters: hash the password again at the next login, when it's at hand.

verify async

verify(user_id: Any, password: str, stored: str | None, *, request: HTTPConnection | None = None) -> bool

Whether password is the one of stored, the hash of user_id's. False when it isn't, when the account is locked out by wrong ones, or when the backend fails. For an account that doesn't exist, stored=None (and its user_id, or None): it takes as long as one that does.

check async

check(password: str, *, user_inputs: Sequence[str] = ()) -> list[str]

What's wrong with password as a new one, as NIST SP 800-63B says: "too_short" (min_length), "too_long" (MAX_PASSWORD), "contains_user_input" (one of user_inputs, such as the username or the e-mail address's local part, of 4 characters or more), and "pwned" (among those leaked). No rule on its characters: length is what makes it strong. An empty list when it's fine.

reset_token async

reset_token(user_id: Any) -> str

A token for user_id to set a new password, for the link of the e-mail it's sent by; it replaces their last one. Raises BackendUnavailable when the backend fails.

use_reset_token async

use_reset_token(token: str, *, request: HTTPConnection | None = None) -> str | None

The user ID (as a string) whose token it is, used up; None when it's unknown, expired, used, or replaced by a newer one. Then set the new password, and end the user's sessions (UserSessions.end_all).

fastapi_alcazaba.auth.ScryptParameters dataclass

ScryptParameters(log_n: int = 17, r: int = 8, p: int = 1)

scrypt's cost: N = 2^log_n, r, p. OWASP's: 2^17, 8, 1 (about 128 MiB and a fifth of a second each).

memory property

memory: int

Bytes scrypt needs, with room: what maxmem must allow.

fastapi_alcazaba.auth.PwnedPasswords

PwnedPasswords(*, timeout: float = 2.0, fail_open: bool = True, client: Any = None, url: str = 'https://api.pwnedpasswords.com/range/')

Whether a password is among those leaked, by Have I Been Pwned's Pwned Passwords API, without sending it: only the first five characters of its SHA-1 go out (k-anonymity), and the answer is padded, so that its size doesn't tell them either. Needs httpx2.

Parameters:

Name	Type	Description	Default
timeout	float	seconds for the call.	2.0
fail_open	bool	when the API doesn't answer, take the password as not leaked (the default: the check is advice, and a sign-up mustn't wait for it), or raise.	True
client	Any	the httpx2 client to call it with; one of its own by default.	None
url	str	the API's range endpoint.	'https://api.pwnedpasswords.com/range/'

count async

count(password: str) -> int

How many times password is in the leaks; 0 when it isn't (or the API didn't answer, with fail_open).

fastapi_alcazaba.auth.FastAPIAPIKeys

FastAPIAPIKeys(lookup: Lookup[R], *, header: str = 'X-API-Key', prefix: str = 'ak', bearer: bool = False, pepper: Secrets | None = None, cache_ttl: float = 0.0, cache_size: int = 1024, scheme_name: str = 'APIKey', metrics: Metrics | None = None)

Bases: Generic[R]

FastAPI dependency that authenticates a request by its API key and returns its record.

keys = FastAPIAPIKeys(load_key, pepper=settings.api_key_pepper)

@app.get("/me")
async def me(key: Annotated[StoredKey, Depends(keys)]) -> ...

@app.delete("/orders/{id}", dependencies=[Depends(keys.require("orders:write"))])

The key comes in the header (X-API-Key by default) or, with bearer=True, as Authorization: Bearer <key>; both appear in the OpenAPI schema, for the docs' Authorize button. A missing, malformed, unknown, wrong or expired key gets the same 401 (with WWW-Authenticate); a valid key without a scope that require asks for gets a 403. Keys and secrets are never logged; put a FastAPIRateLimiter in front, since each request with a well-formed key costs a lookup.

Revoking a key: delete its record, or set its expires_at to the past. With cache_ttl, records are kept in each process for that many seconds, so a revocation takes up to that long to reach every process.

Parameters:

Name	Type	Description	Default
lookup	Lookup[R]	lookup(key_id): the stored record of that id, or None. Async, or a plain function, which runs in a thread like FastAPI's sync dependencies.	required
header	str	the request header holding the key.	'X-API-Key'
prefix	str	the start of every key (ak_...): letters, digits and underscores, starting with a letter. Secret scanners (GitHub's, GitLab's) can be taught to spot it.	'ak'
bearer	bool	also accept the key as Authorization: Bearer <key>.	False
pepper	Secrets | None	a server secret (at least 32 bytes) mixed into every digest, kept out of the database (an environment variable, a secret manager). A list rotates it: the first one is used for new keys, all of them are accepted; old peppers must stay as long as keys made with them do. Changing it without keeping the old one invalidates every key.	None
cache_ttl	float	seconds each process keeps a record it looked up; 0 looks up every request.	0.0
cache_size	int	records kept at most per process (the least recently used are dropped).	1024
scheme_name	str	name of the scheme in the OpenAPI schema.	'APIKey'
metrics	Metrics | None	where to count the outcomes (fastapi_alcazaba.api_keys); FastAPICache's by default.	None

require

require(*scopes: str) -> Callable[..., Awaitable[R]]

A dependency like this one that also needs every scope in scopes (a 403 without them): Depends(keys.require("orders:write")).

generate

generate(*, scopes: Iterable[str] = (), expires_at: datetime | None = None) -> NewAPIKey

A new key, to show its owner once, and its record, to store. scopes are free-form names ("orders:read"); expires_at a datetime with its time zone.

authenticate async

authenticate(key: str | None) -> R

The record of key, if it's valid; raises an HTTPException (401) otherwise. For keys that come some other way (a WebSocket message, a test).

forget

forget(key_id: str | None = None) -> None

Drop the record of key_id (every record when None) from this process's cache: call it after revoking a key, for this process to see it at once.

fastapi_alcazaba.auth.APIKeyRecord dataclass

APIKeyRecord(id: str, digest: str, scopes: frozenset[str] = frozenset(), expires_at: datetime | None = None)

What to store for a key: everything but the key itself.

fastapi_alcazaba.auth.TokenBlocklist

TokenBlocklist(cache: Cache | None = None, *, namespace: str = 'revoked', fail_open: bool = False, leeway: int = 60)

Revoked tokens (by their jti) and subjects logged out everywhere (revoke_subject).

blocklist = TokenBlocklist()

async def current_user(claims: Annotated[dict[str, Any], Depends(decode_jwt)]) -> str:
    sub, iat = claims["sub"], claims["iat"]
    if await blocklist.is_revoked(claims["jti"], subject=sub, issued_at=iat):
        raise HTTPException(401)
    return sub

Each mark lasts while the token could still be accepted, leeway included, and then the backend drops it. When the cache has a signing_key or encryption_key, the marks are sealed with it and planted ones are ignored; but whoever can delete keys in the backend can still un-revoke tokens, so keep it out of reach.

Parameters:

Name	Type	Description	Default
cache	Cache | None	the Cache whose backend keeps the marks; FastAPICache's by default, resolved at each call (so the blocklist can be created before FastAPICache.init).	None
namespace	str	part of the keys, after the cache's prefix (see Cache.state_key).	'revoked'
fail_open	bool	when the backend fails, is_revoked answers False (tokens are accepted) instead of True (refused, the default).	False
leeway	int	seconds your JWT library still accepts a token after its exp (PyJWT's leeway): marks are kept that much longer.	60

revoke async

revoke(token_id: str, *, expires_at: Moment | None = None, ttl: int | None = None) -> None

Refuse the token token_id (its jti) until it would have expired anyway: give its exp as expires_at, or the seconds it has left as ttl. Raises BackendUnavailable when the backend fails: the token isn't revoked then.

revoke_subject async

revoke_subject(subject: str, *, ttl: int) -> None

Refuse every token of subject issued before the current second: log out everywhere, e.g. after a password change. ttl is the longest lifetime of your tokens. iat has whole seconds, so the tokens issued during this second stay valid, among them the one you issue right after for the current session. Raises BackendUnavailable when the backend fails.

is_revoked async

is_revoked(token_id: str, *, subject: str | None = None, issued_at: Moment | None = None) -> bool

Whether the token is refused: its jti was revoked, or its subject was logged out everywhere after it was issued (give its sub and iat). When the backend fails, not fail_open.

fastapi_alcazaba.auth.OneTimeCodes

OneTimeCodes(secret: Secrets, cache: Cache | None = None, *, digits: int = 6, ttl: int = 300, max_attempts: int = 5, namespace: str = 'otp')

Codes of digits digits, sent by e-mail or SMS, that prove who someone is: each lasts ttl seconds, works once and is lost after max_attempts wrong guesses.

codes = OneTimeCodes(secret=settings.otp_secret)

code = await codes.issue(user.email, purpose="login")  # send it to the user
...
if not await codes.verify(form.email, form.code, purpose="login"):
    raise HTTPException(400, "Invalid or expired code")

The backend holds an HMAC of each code with secret, never the code: a plain digest of a 6-digit code is brute-forced in milliseconds by whoever reads the backend. Issuing a code replaces the previous one and gives max_attempts again, so put a FastAPIRateLimiter on the route that issues them.

Parameters:

Name	Type	Description	Default
secret	Secrets	at least 32 bytes (secrets.token_urlsafe(32)), from the environment or a secret manager. A list rotates them: the first one is used, all of them are accepted.	required
cache	Cache | None	the Cache whose backend keeps the codes; FastAPICache's by default, resolved at each call.	None
digits	int	length of the codes, 4 to 10.	6
ttl	int	seconds each code lasts.	300
max_attempts	int	guesses allowed for each code, right or wrong.	5
namespace	str	part of the keys, after the cache's prefix (see Cache.state_key).	'otp'

issue async

issue(subject: str, purpose: str = '') -> str

A new code for subject (an e-mail address, a user id...) and purpose ("login", "reset-password"...), replacing the previous one. Raises BackendUnavailable when the backend fails.

verify async

verify(subject: str, code: str, purpose: str = '') -> bool

Whether code is the current code of subject for purpose; it's used up then. Of several requests with the right code at the same time, only one gets True. False when the backend fails.

fastapi_alcazaba.auth.FastAPITOTP

FastAPITOTP(*, issuer: str, secret: Secrets, cache: Cache | None = None, digits: int = 6, period: int = 30, window: int = 1, algorithm: Algorithm = 'SHA1', max_attempts: int = 5, lockout: int = 900, namespace: str = 'totp', audit: FastAPIAudit | None = None)

The codes of authenticator apps as a second factor (RFC 6238), recovery codes, and step-up for the routes that need a recent one.

• The secrets are sealed with AES-256-GCM, each for its user: store sealed, never the secret. A sealed secret copied to another user doesn't open, nor without secret.
• A code is accepted within window steps of period seconds before or after now, for clocks out of step; each one once (RFC 6238, 5.2): another request with the same code, from someone who looked over the user's shoulder, is refused.
• Guesses: after max_attempts wrong codes (recovery codes included), the user's codes are refused for lockout seconds, whatever they are: a 6-digit code isn't guessed.
• When the backend fails, nothing is accepted: no code is checked for replay, nor guesses counted.

Parameters:

Name	Type	Description	Default
issuer	str	the name the authenticator app shows for the account ("Example").	required
secret	Secrets	at least 32 bytes (secrets.token_urlsafe(32)), from your settings: it seals the TOTP secrets and makes the recovery codes' digests. A list rotates it: the first one seals, all of them open.	required
cache	Cache | None	the Cache whose backend keeps the codes used and the guesses; FastAPICache's by default, resolved at each call.	None
digits	int	of each code, 6 or 8.	6
period	int	seconds each code lasts.	30
window	int	steps accepted before and after the current one.	1
algorithm	Algorithm	SHA1, which every app supports, SHA256 or SHA512.	'SHA1'
max_attempts	int	wrong codes before the lockout.	5
lockout	int	seconds the codes are refused after max_attempts wrong ones, counted from the first of them.	900
namespace	str	part of the keys, after the cache's prefix (see Cache.state_key).	'totp'
audit	FastAPIAudit | None	the audit trail that records each second factor checked (auth.mfa) and each recovery code tried (auth.recovery_code), with their outcome.	None

enroll

enroll(user_id: str | int | UUID, label: str) -> TOTPEnrollment

A new secret for user_id's app; label is what the app shows under the issuer (their e-mail address, their username). Store sealed once a code of it has been verified.

code

code(user_id: str | int | UUID, sealed: str, at: float | None = None) -> str

The code the app shows at at (now by default): for tests, not for checking codes.

verify async

verify(user_id: str | int | UUID, sealed: str, code: str, *, request: HTTPConnection | None = None) -> bool

Whether code is one the app of user_id, whose secret is sealed, shows now (or window steps away), and hasn't been used. False when the user is locked out by wrong codes, or the backend fails. Raises ValueError when sealed isn't user_id's.

recovery_codes

recovery_codes(user_id: str | int | UUID, count: int = 10) -> RecoveryCodes

count new recovery codes for user_id, which replace the old ones: show codes once, store digests (instead of the old ones).

use_recovery_code async

use_recovery_code(user_id: str | int | UUID, code: str, digests: Collection[str], *, request: HTTPConnection | None = None) -> str | None

The digest of digests that code is, used up: delete it from those stored. None when it's none of them, was used already, the user is locked out, or the backend fails. Counts as a guess like a wrong TOTP code.

mark

mark(request: HTTPConnection, user_id: str | int | UUID) -> None

Note in the session that user_id just gave their code (stepup.mark), and give the session a new id: it holds more privileges now.

verified_at

verified_at(request: HTTPConnection, user_id: str | int | UUID) -> float | None

When user_id last gave a second factor in this session; None when they didn't, or someone else did (stepup.verified_at).

require_recent

require_recent(user_id: Callable[..., Any], max_age: int = 300) -> Callable[..., Any]

A dependency that needs a second factor (this one, or a passkey) of the last max_age seconds in this session: stepup.require_recent.

fastapi_alcazaba.auth.TOTPEnrollment dataclass

TOTPEnrollment(uri: str, secret: SecretStr, sealed: str)

A new secret for a user's authenticator app.

Parameters:

Name	Type	Description	Default
uri	str	the otpauth:// URI, to show as a QR code for the app to scan.	required
secret	SecretStr	the secret in base32, for those who type it in the app.	required
sealed	str	the secret encrypted for its user, to store with them once they've typed a code that verify accepts: it's worth nothing without FastAPITOTP's secret.	required

fastapi_alcazaba.auth.RecoveryCodes dataclass

RecoveryCodes(codes: list[str], digests: list[str])

Codes that stand in for the app once each, for a lost phone.

Parameters:

Name	Type	Description	Default
codes	list[str]	to show the user once, to print or keep in a password manager.	required
digests	list[str]	to store with the user; use_recovery_code tells which one was used.	required

fastapi_alcazaba.auth.FastAPIWebAuthn

FastAPIWebAuthn(*, rp_id: str, rp_name: str, origins: Collection[str], user_verification: UserVerification = 'required', algorithms: Sequence[int] = (ES256, EDDSA, RS256), timeout: int = 300, cache: Cache | None = None, namespace: str = 'webauthn', audit: FastAPIAudit | None = None)

Passkeys and security keys: registering them (registration_options, verify_registration) and logging in with them (authentication_options, verify_authentication), as WebAuthn Level 2 says.

• The challenges are random (32 bytes), kept in the cache's backend for timeout seconds, bound to the ceremony and to the user they were made for, and used once: of two answers to the same challenge, one is checked.
• The origin of the answer must be one of origins, never in a cross-origin iframe, and the authenticator's data must be for rp_id: a phishing site's origin is refused, and so is its RP ID, which the browser wouldn't sign for another site.
• The user must be present (a touch), and verified (a PIN, a fingerprint, the phone's lock) with user_verification="required", the default: a passkey is then both factors.
• The counter: a credential whose counter goes back (or doesn't move, for those that count) was cloned: refused. Synced passkeys count 0, and aren't checked.
• When the backend fails, no answer is checked: the challenges can't be used once.

Parameters:

Name	Type	Description	Default
rp_id	str	the relying party's ID: the site's domain, or one it's under (example.com for https://login.example.com). The credentials only work there.	required
rp_name	str	the name the authenticator shows ("Example").	required
origins	Collection[str]	the origins the answers come from, https://example.com (or http://localhost:8000, in development); each on rp_id or under it.	required
user_verification	UserVerification	required, preferred or discouraged.	'required'
algorithms	Sequence[int]	the COSE algorithms accepted, in order of preference.	(ES256, EDDSA, RS256)
timeout	int	seconds a challenge lasts, and the browser waits for the user.	300
cache	Cache | None	the Cache whose backend keeps the challenges; FastAPICache's by default.	None
namespace	str	part of the keys, after the cache's prefix (see Cache.state_key).	'webauthn'
audit	FastAPIAudit | None	the audit trail that records each passkey registered (auth.passkey.register) and each login with one (auth.passkey), with their outcome.	None

registration_options async

registration_options(user_id: UserId, user_name: str, *, display_name: str | None = None, user_handle: str | None = None, exclude: Sequence[Passkey] = (), resident_key: Literal['required', 'preferred', 'discouraged'] = 'required') -> dict[str, Any]

The options of navigator.credentials.create(), as JSON, for user_id to register a passkey. user_name and display_name are what the authenticator shows; user_handle is the user's handle (base64url) when they have passkeys already, else a new random one. exclude are their passkeys, which the authenticator won't register twice. resident_key="required" makes a passkey, which logs in without a username.

verify_registration async

verify_registration(user_id: UserId, credential: Mapping[str, Any], *, request: HTTPConnection | None = None) -> Passkey

The Passkey of credential (the browser's credential.toJSON() of create()), to store with user_id; WebAuthnRefused when it isn't a valid answer to a challenge made for them. Check that no other user has its id before storing it.

authentication_options async

authentication_options(user_id: UserId | None = None, *, allow: Sequence[Passkey] = ()) -> dict[str, Any]

The options of navigator.credentials.get(), as JSON. Without a user, any passkey of this site may answer (a login without a username, or the browser's autofill); with user_id and allow (their passkeys), only theirs, for a second factor.

verify_authentication async

verify_authentication(credential: Mapping[str, Any], passkey: Passkey, *, user_id: UserId | None = None, request: HTTPConnection | None = None) -> Passkey

The Passkey with its new counter and last_used, to store again, when credential (the browser's credential.toJSON() of get()) is a valid answer by passkey, the one whose id is credential_id(credential); WebAuthnRefused otherwise. user_id is the user passkey is stored with: the challenge, when made for a user, must be theirs.

mark

mark(request: HTTPConnection, user_id: UserId) -> None

Note in the session that user_id just logged in with a passkey (stepup.mark), for the routes of require_recent, and give the session a new id.

require_recent

require_recent(user_id: Callable[..., Any], max_age: int = 300) -> Callable[..., Any]

A dependency that needs a second factor (a passkey, or a TOTP code) of the last max_age seconds in this session: stepup.require_recent.

fastapi_alcazaba.auth.Passkey dataclass

Passkey(id: str, public_key: str, algorithm: int, user_handle: str, sign_count: int = 0, transports: tuple[str, ...] = (), backup_eligible: bool = False, backed_up: bool = False, created: float = time.time(), last_used: float | None = None)

A registered credential, to store with its user (as to_dict(), JSON) and to give back to verify_authentication. It holds a public key: nothing in it opens a door by itself.

Parameters:

Name	Type	Description	Default
id	str	the credential's ID (base64url), by which the browser names it.	required
public_key	str	its COSE public key (base64url).	required
algorithm	int	its COSE algorithm: -7 (ES256), -8 (EdDSA) or -257 (RS256).	required
user_handle	str	the user's handle (base64url) the authenticator keeps with it.	required
sign_count	int	the signature counter it last showed (0 for those that don't count).	0
transports	tuple[str, ...]	how the browser reaches it ("internal", "hybrid", "usb"...).	()
backup_eligible	bool	it can be synced (a passkey in iCloud Keychain, Google Password Manager).	False
backed_up	bool	it is synced now.	False
created	float	when it was registered (Unix time).	time()
last_used	float | None	when it last logged someone in (Unix time); None until it does.	None

fastapi_alcazaba.auth.WebAuthnRefused

WebAuthnRefused(reason: str, detail: str = '')

Bases: Exception

A registration or a login that isn't valid: the user's answer is refused, a 400 or a 401 for your route. reason says why, for the logs (not for the user).

Parameters:

Name	Type	Description	Default
reason	str	short and stable: "challenge", "origin", "rp_id", "user_presence", "user_verification", "signature", "counter", "algorithm", "malformed"...	required

fastapi_alcazaba.auth.credential_id

credential_id(credential: Mapping[str, Any]) -> str

The ID of the credential the browser answered with (base64url): what to look its Passkey up by, before verify_authentication.

fastapi_alcazaba.auth.stepup.mark

mark(request: HTTPConnection, user_id: UserId, method: str) -> None

Note in the session that user_id just gave a second factor (method: "totp", "passkey"...), and give the session a new id (with ServerSessionMiddleware): it holds more privileges now.

fastapi_alcazaba.auth.stepup.verified_at

verified_at(request: HTTPConnection, user_id: UserId) -> float | None

When user_id last gave a second factor in this session (Unix time); None when they didn't, or someone else did.

fastapi_alcazaba.auth.stepup.require_recent

require_recent(user_id: Callable[..., Any], max_age: int = 300) -> Callable[..., Any]

A dependency that lets through the users who gave a second factor in the last max_age seconds in this session, and returns their ID; the others get a 403 ("Second factor required"), to ask for one and mark the session, and requests without a user a 401. user_id is a dependency that returns the request's user ID, or None.