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.
Describe this scheme as an OpenAPI Security Scheme Object.
Return the object published under components.securitySchemes, or
None when the scheme cannot be described. A route guarded by an
undescribed scheme is published with no security requirement - which
asserts the endpoint is open - so the schema build warns rather than
doing that silently.
Only the scheme knows what it reads and how a client should send it, so it answers for itself: a subclass adding a new authentication style implements this and is published like a built-in.
Usage::
class CertHeaderAuth(SecurityScheme):
__slots__ = ()
def openapi_scheme(self):
return {"type": "apiKey", "in": "header", "name": "X-Cert"}
Bases: _APIKeyBase
API Key authentication via HTTP header.
Describe an API key, in the location this subclass reads it from.
Build 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.
Describe an API key, in the location this subclass reads it from.
Build 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.
Describe an API key, in the location this subclass reads it from.
Build 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 authentication, published with the scheme it advertises.
dataclass
¶
HTTP Basic auth credentials.
eq=False keeps the identity equality - and so the hashability - these
credentials had before they were a dataclass; a generated __eq__ sets
__hash__ to None, which stops them being usable as a dict key or set
member.
Bases: _BearerScheme
HTTP Bearer token authentication.
HTTP authentication, published with the scheme it advertises.
The scheme is scheme_name, which is also what __call__ matches the
Authorization header against. Publishing a fixed "bearer" meant a
custom scheme changed what the server accepts and not what the document
told a client to send. Lower-cased because OpenAPI 3.1 names the IANA
registry entry, whose entries are lower-case.
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.
HTTP authentication, published with the scheme it advertises.
dataclass
¶
Parsed Digest auth challenge response - RFC 7616 Sec. 3.4.
eq=False for the same reason as HTTPBasicCredentials: it keeps the
identity equality, and the hashability, these had before the dataclass.
Bases: _OAuth2BearerScheme
OAuth2 Password Bearer flow - extracts token from Authorization header.
Describe the password flow, with its token endpoint and scopes.
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,
)
Describe the authorization-code flow; refreshUrl is omitted when unset.
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.
OpenID Connect discovery, published as its single URL.
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.
The OpenAPI document describes this as an apiKey credential read from the
session cookie. Pass cookie_name= when SessionMiddleware is configured
with a name other than the default, so the document names the cookie a
client actually has to send.
subject_key= / scopes_key= name the session slots this reads. They are
the same slots login_session writes, so an application that overrides
either must pass the matching key to login_session too - otherwise the
scheme reads a slot the login never wrote and every request is anonymous.
Describe the session credential, as a cookie-borne API key.
OpenAPI has no session-specific scheme type; a cookie credential is an
apiKey read from cookie, which is how APIKeyCookie describes the
same transport.
login_session(request: Request, subject: str, *, scopes: Iterable[str] = (), subject_key: str = SESSION_SUBJECT_KEY, scopes_key: str = SESSION_SCOPES_KEY, **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.
subject_key / scopes_key must name the same slots as the SessionAuth
that reads the session back. A scheme built with a non-default key reads a
slot this helper never wrote, and every request resolves anonymous.
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 bare misuse raises - so the
documented except BadResetToken works without breaking a caller who
catches TypeError instead.
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.
Audit¶
The structured form of Veloce.security_audit(). veloce.audit.run(app)
returns Finding objects; startup refuses to serve on an error.
dataclass
¶
One thing an audit found.
message states what is wrong; fix states what to do about it, kept
separate so a tool can present or suppress the remedy independently. id
is a stable handle for the finding, which is what SILENCED_AUDIT_IDS
matches on, so an accepted finding can be turned off without turning the
audit off.
Usage::
Finding(
"TENANT_SIGNING_KEY is not set - tenant headers are unverified.",
severity="error",
fix="set TENANT_SIGNING_KEY",
id="tenant-signing-key-missing",
)
dataclass
¶
What an audited middleware is given.
routes_final is False when the application was imported but never
started, which is how veloce check runs it. A middleware that reads the
route table sets audit_needs_routes and is not called at all in that
case, so this flag is for the rarer check that can narrow its scope rather
than skip.
Bases: VeloceError, ValueError
An error-severity finding refused the application's startup.
Also a ValueError, which is what a middleware raised for the same
condition before findings carried a severity, so existing handling of a
misconfigured middleware still catches it. The findings that caused the
failure are on .findings.
Middleware reports on itself through Middleware.audit without registering
anywhere. Something that hardens or exposes the app without being middleware
says so with app.register_auditable(...), which puts it on the same terms:
give it an audit(ctx) yielding Findings.
Register component to report to veloce check and security_audit.
Middleware reports on itself through Middleware.audit without being
registered anywhere - it is already in the stack. Something that hardens
or exposes the app without being middleware has nowhere to say so, and
a mounted MCP endpoint is the case in tree: it registers routes, so
without this the audit had nothing to ask about a tool-execution
endpoint with no authentication.
component needs an audit(ctx) yielding Findings, exactly as a
middleware's does; it may also carry the Auditable class markers
(sets_hardening_headers, audit_needs_routes). Returns component,
so it can be used as a decorator on a class.
Usage::
class TelemetryEndpoint:
def audit(self, ctx):
if not self.authenticated:
yield Finding(
"telemetry endpoint accepts unauthenticated writes",
severity="error",
fix="pass auth=...",
id="telemetry-unauthenticated",
)
app.register_auditable(TelemetryEndpoint())