Security¶
Authentication schemes, token handling, password hashing, and the signing primitives underneath them.
Base contract for callable authentication schemes.
Owns the auto_error field and documents the resolver's expected shape:
a __call__(self, request) that returns the extracted credential, or
None when authentication is absent and auto_error is False. Not an
abc.ABC: the hook raises NotImplementedError so subclasses that forget
to override fail loudly without pulling in the ABC machinery.
Bases: _APIKeyBase
API Key authentication via HTTP header.
The WWW-Authenticate challenge sent on a 401.
Returns {WWW-Authenticate: APIKey realm="..."} when a realm is
configured, else the bare APIKey token, which still satisfies
RFC 9110 Sec. 11.6.1. Subclasses may override to emit a custom
challenge.
Bases: _APIKeyBase
API Key authentication via query parameter.
The WWW-Authenticate challenge sent on a 401.
Returns {WWW-Authenticate: APIKey realm="..."} when a realm is
configured, else the bare APIKey token, which still satisfies
RFC 9110 Sec. 11.6.1. Subclasses may override to emit a custom
challenge.
Bases: _APIKeyBase
API Key authentication via cookie.
The WWW-Authenticate challenge sent on a 401.
Returns {WWW-Authenticate: APIKey realm="..."} when a realm is
configured, else the bare APIKey token, which still satisfies
RFC 9110 Sec. 11.6.1. Subclasses may override to emit a custom
challenge.
Bases: SecurityScheme
HTTP Basic authentication - extracts username:password from Authorization header.
HTTP Basic auth credentials.
Bases: _BearerScheme
HTTP Bearer token authentication.
Bases: SecurityScheme
HTTP Digest authentication - RFC 7616.
Parses the Authorization: Digest ... header into the named fields
and returns them as HTTPDigestCredentials. This class does NOT
validate the response hash - the application owns the secret
(HA1) and must compute the expected digest itself; Digest's whole
point is that the secret never crosses the wire. Veloce's job is to
parse the challenge response and to emit a 401 + WWW-Authenticate:
Digest ... header when auth is missing or malformed.
The scheme's responsibility is the parse + challenge dance; verifying the response is application logic.
Parsed Digest auth challenge response - RFC 7616 Sec. 3.4.
Bases: _OAuth2BearerScheme
OAuth2 Password Bearer flow - extracts token from Authorization header.
OAuth2 password request form data.
async
classmethod
¶
from_request(request: Request) -> OAuth2PasswordRequestForm
Parse an OAuth2 password grant from the request form data.
Bases: OAuth2PasswordRequestForm
OAuth2PasswordRequestForm with a mandatory grant_type.
The non-strict form leaves grant_type optional;
the strict form requires it and constrains the value to the
literal password (RFC 6749 Sec. 4.3.2). Missing or mismatched values
fail validation with 422.
async
classmethod
¶
from_request(request: Request) -> OAuth2PasswordRequestFormStrict
Parse and validate that grant_type is present and equals 'password'.
Bases: _OAuth2BearerScheme
OAuth2 Authorization-Code (with PKCE) Bearer flow.
Extracts a Bearer token from the Authorization: header exactly
like OAuth2PasswordBearer; the difference is the OpenAPI security
scheme it advertises (authorizationUrl + tokenUrl + scopes),
which is what an interactive OAuth2 client (Swagger UI's "Authorize"
button, an SPA's auth library) uses to start the redirect dance.
The construction shape is chosen so an OpenAPI snippet generated from a standard OpenAPI document can be replayed against veloce without rewrites:
oauth2 = OAuth2AuthorizationCodeBearer(
authorizationUrl="https://auth.example.com/authorize",
tokenUrl="https://auth.example.com/token",
refreshUrl=None,
scopes={"read:items": "Read items", "write:items": "Write items"},
auto_error=True,
)
Bases: _OAuth2BearerScheme
OpenID Connect Bearer authentication.
Same Bearer extraction logic as the OAuth2 schemes; the OpenAPI
scheme advertises a single openIdConnectUrl pointing at the
provider's .well-known/openid-configuration document. Clients
auto-discover everything else from there.
Bases: SecurityScheme
Resolve the current Principal from the request's session.
Usage::
from veloce import Depends, Veloce
from veloce.security.session import SessionAuth, login_session
app = Veloce(secret_key="...")
app.add_middleware(SessionMiddleware, secret_key="...")
session_auth = SessionAuth()
@app.post("/login")
async def login(request: Request):
login_session(request, "user-42", scopes={"items:read"})
return {"ok": True}
@app.get("/me")
async def me(principal=Depends(session_auth)):
return {"user": principal.subject}
Returns the Principal and publishes it via set_principal, so
current_principal() resolves for anything further down the request -
including a dependency shared with an MCP-exposed handler.
With auto_error=False an anonymous request resolves to None instead of
raising, for routes that render differently when signed in. A missing
SessionMiddleware is a configuration error rather than an anonymous
request, and still raises under either setting.
Pass loader= to build a richer principal from the stored subject (a
database lookup, say); it receives (request, subject) and returns a
Principal, or None to reject the session.
login_session(request: Request, subject: str, *, scopes: Iterable[str] = (), **claims: Any) -> None
Sign subject into the request's session and publish the principal.
Rotates the session id first, so a session id planted before login cannot be replayed against the now-authenticated session.
logout_session(request: Request) -> None
Clear the session's identity and the request's principal.
Clears the whole session rather than only the identity keys: leftover per-user state on a session that has changed hands is a data-leak shape, not a convenience.
Sign claims into a compact JWS token using the given HMAC algorithm.
decode_jwt(token: str, secret: str | bytes, *, algorithms: Sequence[str], audience: str | Sequence[str] | None = None, issuer: str | None = None, require: Sequence[str] = (), leeway: float = 0, now: float | None = None) -> Claims
Verify a compact JWS token and return its claims as a read-only mapping.
Bases: Mapping[str, Any]
Read-only mapping over a decoded JWT payload.
Usage::
claims = decode_jwt(token, secret, algorithms=["HS256"])
user_id = claims["sub"]
Bases: VeloceError
Base class for all JWT decode/encode failures.
Bases: JWTError
Malformed structure: not three segments, bad base64, or bad JSON.
Derive a salted verifier for password.
Returns a self-describing string of the form
method$params$salt$hash where each segment is URL-safe base64
(no padding). Pass this string verbatim to verify_password later.
method:
- "scrypt" (default): RFC 7914, memory-hard.
- "pbkdf2:sha256": NIST SP 800-132, CPU-only.
salt_length is the number of random bytes used for the salt;
16 is the OWASP minimum.
async
¶
hash_password_async(password: str | bytes, method: str = 'scrypt', salt_length: int = _SALT_BYTES) -> str
Async-safe wrapper for hash_password - runs the KDF on a thread.
hash_password calls hashlib.scrypt / pbkdf2_hmac synchronously;
those are deliberately slow (~100 ms) and would block the event loop
if called directly from an async handler. This wrapper offloads the
work to the default executor so the loop stays free for other
requests. Use this from async def handlers; keep the sync
hash_password for sync handlers / scripts / CLI tools.
Compare candidate against a stored verifier string.
Returns False (never raises) for any malformed stored, unknown
method, or mismatch. Uses hmac.compare_digest for the final byte
comparison so timing attacks can't leak partial matches.
async
¶
Async-safe wrapper for verify_password - runs the KDF on a thread.
Same rationale as hash_password_async: the scrypt / PBKDF2 verify
is ~100 ms of CPU; calling it synchronously from an async handler
blocks the event loop. Offload it.
Verify candidate and report whether stored should be upgraded.
Returns (ok, needs_update):
- ok is the same boolean verify_password returns.
- needs_update is True only when ok is True AND the stored
verifier is weaker than the current defaults (see needs_rehash).
It is always False on a failed verify - there is nothing to upgrade
for a credential that did not match.
Usage::
ok, upgrade = verify_and_needs_update(user.pw_hash, form_password)
if not ok:
raise Unauthorized()
if upgrade:
user.pw_hash = hash_password(form_password)
db.save(user)
async
¶
Async-safe wrapper for verify_and_needs_update.
Offloads the KDF verify to a thread for the same reason as
verify_password_async; needs_rehash is a cheap string parse and
runs inline on the worker thread alongside the verify.
Whether stored should be re-derived with the current defaults.
Returns True when the stored verifier was produced with a weaker
configuration than hash_password would produce today - either a
non-default method, or cost parameters below the current module
defaults. An app can call this after a successful verify_password
and transparently re-hash the password (the plaintext is in hand at
that moment) so credentials drift up to the current work factor on
each login without a forced reset.
A malformed or unparseable stored returns False - it is not a
rehash candidate (it would not verify in the first place), so the
caller's normal verify-failure path handles it.
Cheap policy check - not exhaustive.
Returns True only when the password meets a minimum baseline: at
least min_length characters and contains at least one digit AND
one alphabetic character. Callers that want NIST SP 800-63B-style
policy (block known-leaked passwords, drop max-length caps, etc.)
should layer on top.
Bind a caller-supplied state fingerprint into a signed reset token.
check_reset_token(token: str, state: bytes, *, secret: str | bytes, max_age: int, fallback_secrets: Sequence[str | bytes] = (), salt: str | bytes = RESET_TOKEN_SALT) -> bool
Return True iff the token is authentic, unexpired, and still bound to state.
Bases: VeloceError, TypeError
Raised on programmer misuse; invalid tokens return False instead.
Also a TypeError, which is what the misuse used to raise - so the
documented except BadResetToken works without breaking a caller already
catching TypeError.
dataclass
¶
The authenticated identity and granted scopes for the current request.
Usage::
from veloce import Principal, set_principal
set_principal(Principal(subject="user-42", scopes={"mcp:tools"}))
current_principal() -> Principal | None
Return the authenticated Principal for the current request, or None.
set_principal(principal: Principal | None) -> None
Set the authenticated Principal for the current request.
Call this from whatever authenticates a request - an HTTP auth middleware or
dependency, or the MCP transport's token verifier - so downstream code reads
one identity through current_principal, regardless of which door the request
arrived on.
Hold a str/bytes secret while resisting accidental disclosure.
Usage::
token = Secret(os.environ["API_TOKEN"])
send(token.reveal())
HMAC-SHA256 signer for arbitrary JSON-serialisable values.
Usage
s = Signer(secret="server-secret", salt="reset-token") token = s.dumps({"user_id": 42}) ... data = s.loads(token, max_age=3600) # raises if older than 1h
Add an additional secret accepted for verification (not signing).
Used during secret rotation: configure the new secret as primary, keep the old one as a fallback for the rotation window. Tokens signed with the fallback still verify; new tokens use the primary.
Verify token and return the original data.
Raises BadSignature on tamper / unknown secret, BadTimeSignature
when max_age is set and the token's timestamp is older than that.
Bases: VeloceError
The token's signature did not verify against the configured secret.
Bases: BadSignature
The signature verified but the token is older than max_age.
Bases: BadSignature
The token's payload could not be decoded (malformed base64 / JSON).
Compare two secrets without leaking their contents through timing.
Wraps hmac.compare_digest; str inputs are UTF-8 encoded first. Use
this when the operands may be str (or mixed str/bytes). Callers
that already hold two equal-typed bytes values - the signing, JWT,
reset-token, password, and Secret verify paths - call hmac.compare_digest
directly: routing them through here would add an isinstance ladder and a
redundant encode/copy on a security-hot verify path for no behavioural gain
(the False-on-type-mismatch branch is unreachable when both operands are
statically bytes).
Join paths onto directory, returning None on any escape.
Returns the absolute joined path if it equals directory or is a
descendant. Returns None if:
- any component in paths is an absolute path,
- any component contains a NUL byte,
- on Windows, any segment names a reserved device (COM1, NUL, ...),
- the resolved path is outside directory.
The check is performed via os.path.abspath, which collapses ..
segments before comparison. Symlinks are not resolved - callers
that distrust symlinks must use os.path.realpath themselves.
Return a safe basename for name.
- Strips directory separators (
/,\) and any non-ASCII characters. - Replaces unsafe characters with underscores; collapses repeats.
- Strips leading/trailing dots/spaces/underscores (blocks
.and..). - Prefixes Windows reserved names (
CON,PRN, ...) with_. - Returns
""when nothing survives sanitisation.
Empty or whitespace-only input returns "". The caller is responsible
for treating that as a rejection - secure_filename will not raise.