Identity and tenancy#

Identity is shared across HTTP, WebSocket, MCP, GraphQL and gRPC boundaries. These modules cover proof, authorization, user lifecycle and tenant lifecycle.

wreath.auth#

Authentication: identities, credentials, and authentication backends.

Authentication establishes who a request is. Authorization — what an identity is permitted to do — lives in wreath.authorization. Both are thin public facades over the private wreath._auth implementation package.

AuthenticationBackend#

What Wreath.configure_auth asks of an authentication backend.

A backend answers one question — who is this request? — with an Identity or with nothing. It never decides what that identity may then do; that is AuthorizationProvider, and they are two protocols because keeping the two questions apart is the point.

The pipeline asks at most once per request and publishes the answer on request.identity. On a route the compiled table classified as protected the backend runs before route selection; on any other route carrying a requirement it runs on first need, in the authorization stage. A route declaring none of authenticated, identify, roles, permissions, second_factor or authorize is not protected at all, so the backend is never asked there and request.identity stays None even for a caller holding a perfectly good credential.

Refuse by returning None, not by raising. None is the only refusal channel the pipeline understands: on a route that requires an identity it becomes a 401 carrying challenge, and on an identify route it is simply an anonymous caller. An exception is neither — it is not translated into a status, and on a protected route under the default bitset routing it leaves the application without passing the error boundary at all. A backend whose credential store is unreachable therefore has to decide, and None is the decision that leaves the route closed.

Backends are duck-typed: configure_auth accepts any object carrying these two members and never checks isinstance. One convention lives outside the protocol and is read with getattr: a truthy requires_session attribute tells route compilation that this backend cannot run until something global has published request.state.session, and the compile refuses rather than answering 401 to a caller holding a valid cookie.

BearerTokenBackend, SessionIdentityBackend and CompositeBackend ship as implementations.

AuthenticationBackend(*args, **kwargs)

authenticate#

async authenticate(request: Request) -> Identity | None

Identify request, or answer None when it carries no valid credential.

Awaited once per request, before any second-factor, role, permission or policy check reads request.identity. None is a value — "nobody" — rather than a failure, and it is also the only way to refuse; see the class docstring.

challenge#

challenge(request: Request) -> str | None

The WWW-Authenticate value for the 401, or None to send no header.

Consulted only on the refusal path: when authenticate yielded nothing on a route that required an identity. Synchronous, and must stay so — the pipeline does not await it. BearerTokenBackend answers "Bearer"; SessionIdentityBackend answers None, because the remediation for a browser session is the login route rather than a header.

BearerTokenBackend#

Authenticate Authorization: Bearer <token> through a verifier you supply.

The backend owns the header, and only the header: it finds the field, checks the scheme, and hands the token to verifier. Deciding whether a token is real — a signature check, a database lookup, a call to an identity provider — is the verifier's, which is why the same backend serves a JWT (JwtVerifier), an OIDC provider (OidcProvider), and an opaque token in your own table.

async def verify(token: str) -> Identity | None:
    user = await lookup(token)
    return Identity(user.id, roles=frozenset(user.roles)) if user else None

app.configure_auth(BearerTokenBackend(verify))

verifier may be a coroutine function or a plain one, and either may return an awaitable; all three are awaited correctly. Which it is, is detected once in the constructor rather than per request. The verifier is handed the token and nothing else — a decision that needs the request cannot be made here.

Every rejection is None, and there are four of them: no authorization header, more than one, a value with no space or an empty token, and a scheme that is not Bearer (matched case-insensitively, per RFC 9110). Refusing a repeated header is deliberate rather than defensive — Authorization is not a list-valued field, and a proxy and the application picking first, last or joined would authenticate different values from the same request.

The token is the remainder of the header value after the first space, verbatim, and the value is decoded as latin-1 — the encoding ASGI gives header bytes, so no byte can fail to decode.

BearerTokenBackend(verifier: Verifier) -> None

authenticate#

async authenticate(request: Request) -> Identity | None

The verifier's answer for this request's bearer token, or None.

None for any of the four rejections listed on the class, and for a token the verifier itself declined. Whatever the verifier returns is returned unchanged — this backend adds no roles, permissions or claims of its own.

challenge#

challenge(request: Request) -> str

Always "Bearer" — the WWW-Authenticate value for the 401.

The bare scheme, with no realm and no error parameter, and it does not vary with the request. An application that needs either wraps this backend or writes its own.

ClientCredentials#

Cached OAuth2 client-credentials (M2M) token acquisition.

token_path is a path on http_client's pinned origin, or a zero-arg callable returning one (so it can be resolved lazily after OIDC discovery).

ClientCredentials(*, http_client: Any, token_path: str | Callable[[], str], client_id: str, client_secret: str, scope: str | None = None) -> None

token#

async token() -> str

A valid access token, from cache or from a fresh grant.

Refreshes when there is no token yet, or when the cached one is within 30 seconds of expiry — early, so a token does not expire in flight between this call and the request that carries it. Expiry is tracked on the running loop's clock from the expires_in of the token response, defaulting to 3600 seconds when the provider omits it.

Concurrent callers refresh once: the fast path is a lock-free read, and everyone who misses it queues on one lock and re-checks inside it, so a burst of requests against a cold cache makes one token request rather than one each.

Raises:

• RuntimeError — the token endpoint answered with a status other than
• 200, or with a body carrying no string access_token. A failed
• refresh is never a cached token — the previous one is not
• returned as a fallback.

CompositeBackend#

Try each backend in order; the first Identity wins.

CompositeBackend(*backends: AuthenticationBackend) -> None

authenticate#

async authenticate(request: Request) -> Identity | None

The first identity any wrapped backend yields, or None if none does.

Each backend is awaited in constructor order and the walk stops at the first non-None answer, so the backends after it are never asked. A backend that raises stops the walk too — the exception propagates rather than being treated as that backend declining, because a bearer verifier that could not reach its store has not established that this caller is a session caller.

challenge#

challenge(request: Request) -> str | None

The first non-None challenge any wrapped backend offers.

Asked in the same constructor order, and independently of which backend (if any) attempted the authentication — so a bearer-then-session composite advertises Bearer, which is the one of the two that a WWW-Authenticate header can express. None when no backend offers one, and then the 401 carries no challenge header.

requires_session (property)#

requires_session: bool

True when any wrapped backend needs the session to be published.

A composite is usually bearer-then-session, and the session half is just as unable to run before the session exists as it would be alone -- so the requirement propagates rather than being hidden by the wrapper.

Credentials#

Credentials(scheme: 'str', value: 'str')

Credentials(scheme: str, value: str) -> None

Identity#

Who is calling, and everything that can only reduce what they may do.

limits and narrowing are the composed principal reaching the authorizer (wreath._auth.principal). Both default to absent and absent means "no restriction beyond what the providers say", so an ordinary un-delegated request pays nothing for either.

Neither field can grant. A Limits is intersected with what a provider resolved; a Narrowing is refused before the engine is consulted and has the delegating principal's own decision evaluated as a conjunct. That is checked by property test rather than asserted here -- see tests/test_principal_narrow.py.

attributes is the provisioned account data Cedar may read from the principal entity: status, account kind, or validity bounds. Machine accounts are ordinary accounts with different attributes and policies, not a privileged identity class. claims remains unrelated: it is the token's own untrusted/application-facing claims (OIDC and friends), carried verbatim, and nothing derives authority from it.

namespace distinguishes locally unique ids issued by different identity providers without changing the application-facing id. JWT identities use their verified iss; session bridges must carry it forward.

Identity(id: str, type: str = 'User', roles: frozenset[str] = frozenset(), permissions: frozenset[str] = frozenset(), claims: Mapping[str, object] = <factory>, limits: Limits | None = None, narrowing: Narrowing | None = None, attributes: Mapping[str, object] = mappingproxy({}), namespace: str = '') -> None

JwtError#

Base class for JWT construction/configuration errors.

JwtVerifier#

A callable Verifier for BearerTokenBackend.

Holds a fixed key (static-key case). The JWKS case uses the lower-level verify_jwt with a key resolver; see wreath._auth.jwks. audience must be explicit. Pass None only when another layer performs the audience check, as MCPAuth does.

JwtVerifier(*, algorithms: Iterable[str], key: JwtKey | bytes | str, issuer: str | None = None, audience: str | Sequence[str] | None | _AudienceUnset = <wreath._auth.jwt._AudienceUnset object>, leeway: int = 60, required: Iterable[str] = ('exp',), identity: IdentityMapper = <function default_identity>, revoked: RevocationCheck | None = None) -> None

OidcProvider#

One discovered OIDC identity provider.

OidcProvider(name: str, *, issuer: str, audience: str | Sequence[str] | None, http_client: Any, algorithms: Iterable[str] = ('RS256',), leeway: int = 60, required: Iterable[str] = ('exp',), identity: IdentityMapper = <function default_identity>) -> None

bearer_verifier#

bearer_verifier(*, audience: Any = <object object>)

Return an async Verifier closing over this provider's JWKS cache.

discover#

async discover() -> None

Fetch the discovery document and prefetch the JWKS. Run at startup.

RsaPublicKey#

An RSA public key (modulus/exponent). Usable only for RS/PS.

RsaPublicKey(n: int, e: int) -> None

size (property)#

size: int

Modulus length in bytes (k).

SessionIdentityBackend#

Yield an Identity from a principal previously stored in the session.

Reading the session during authentication means the session has to exist by then, which route middleware cannot promise. Wreath refuses the combination at route-compile time rather than answering 401 to a caller holding a valid cookie; requires_session is what it keys that refusal on.

SessionIdentityBackend(*, session_key: str = 'principal') -> None

authenticate#

async authenticate(request: Request) -> Identity | None

The session principal as an Identity, or None for anonymous.

Reads request.state.session[session_key] — "principal" by default — and requires it to be a mapping carrying a non-empty string sub. That subject becomes Identity.id; type defaults to "User"; string roles and permissions are whitespace-delimited, while collection forms must contain only strings; and the whole principal mapping is copied onto Identity.claims, which is what carries second_factor_at to the step-up checks.

Four things each yield None rather than an identity: no session published, no principal under the key, a principal with no usable sub, and a principal marked pending — a login still waiting on its second factor, which is authenticated but incomplete and must not be an identity anywhere. A principal carrying a numeric exp that has passed also yields None, so a signed-in session ends at whichever comes first — the provider's expiry or the session cookie's own max_age — rather than always lasting the cookie's.

challenge#

challenge(request: Request) -> str | None

Always None — a session-authenticated 401 carries no challenge.

There is no WWW-Authenticate scheme that means "go and log in through the browser", and the remediation is the login route rather than a header, so nothing is advertised. Unauthorized(challenge=None) omits the header entirely.

SymmetricKey#

An HMAC shared secret. Usable only for HS256/384/512.

SymmetricKey(secret: bytes) -> None

UnsupportedAlgorithm#

A requested algorithm is not implemented in this build.

authenticated#

authenticated() -> Callable[[Any], Any]

Require an identity, without requiring anything of it.

Marks the endpoint protected, so the compiled route table classifies its path as needing authentication and the backend runs before route selection. A request that yields no identity becomes a 401 carrying the backend's challenge(); one that yields any identity reaches the handler.

This is also what populates request.identity. An endpoint with no authentication or authorization requirement is not protected, so the backend is never asked, and request.identity is None there even for a caller holding a perfectly good token or session. A handler that reads the identity must say so -- with this decorator, or with roles, permissions or authorize, each of which implies it.

Returns:

• A decorator that records the requirement on the endpoint.

default_identity#

default_identity(claims: Mapping[str, Any]) -> Identity

Map standard/Cognito claims onto a wreath Identity.

sub -> id; roles/cognito:groups/groups -> roles; space-delimited scope -> permissions; the full claim set is retained.

identify#

identify() -> Callable[[Any], Any]

Publish request.identity without requiring one -- anonymous is a value.

Runs the authentication backend and sets request.identity to whatever it yields, including None. The route stays public: no challenge, no 401, and no access clause in the compiled route table, so an anonymous caller reaches the handler exactly as it would on a route with no requirement at all.

This is the answer to "who is this, if anyone?" -- the question every sign-in-aware page asks on load, and the one authenticated() cannot ask because it refuses the case it is asking about. Reach for it when a handler renders differently for a known caller rather than refusing an unknown one: a console deciding whether to show a sign-in form, a catalogue marking the rows you already own, a landing page greeting you by name.

@app.get("/session")
@identify()
async def whoami(request: Request) -> dict:
    identity = request.identity
    if identity is None:
        return {"signed_in": False}
    return {"signed_in": True, "subject": identity.id}

The distinction from authenticated() is the whole point and is worth stating plainly: authenticated() means refuse without an identity, and this means ask, and accept either answer. Do not reach for this to make a protected route lenient -- a route that needs an identity should say so, so that the framework can refuse before the handler runs.

Returns:

• A decorator that records the requirement on the endpoint.

jwk_thumbprint#

jwk_thumbprint(jwk: Mapping[str, Any], *, hash_name: str = 'sha-256') -> str

Return the RFC 7638 thumbprint of one JWK.

SHA-256 is the mandatory-to-implement algorithm for both RFC 7638 and RFC 9278. The JSON input is reduced to the required members for its key type, lexicographically ordered, encoded without whitespace, and hashed.

jwk_thumbprint_uri#

jwk_thumbprint_uri(jwk: Mapping[str, Any], *, hash_name: str = 'sha-256') -> str

Return the RFC 9278 URI form of a JWK thumbprint.

key_from_jwk#

key_from_jwk(jwk: Mapping[str, Any]) -> JwtKey

Build a key from a single JWK. Supports oct/RSA/EC/OKP.

key_from_pem#

key_from_pem(pem: str | bytes) -> RsaPublicKey

Parse an RSA public key from a PEM PUBLIC KEY (SPKI) or RSA PUBLIC KEY (PKCS#1) block, without a third-party dependency.

oauth_step_up#

oauth_step_up(*, max_age: int | None = None, acr_values: Iterable[str] = ()) -> Callable[[Any], Any]

Require OAuth authentication recency or class and emit an RFC 9470 challenge.

The identity must expose the access token's auth_time and acr claims in Identity.claims. A valid token that misses either declared requirement is answered with a 401 Bearer insufficient_user_authentication challenge, including the max_age and acr_values a client can carry into a new authorization request.

This is deliberately separate from second_factor(). That decorator owns Wreath's browser-session verification flow and answers 403; this one owns an OAuth resource-server challenge and answers 401. Combining them is refused when the route is declared because their remediation paths conflict.

Args:

• max_age — Non-negative whole seconds since the token's active
• authentication event.
• acr_values — Acceptable authentication context class references, in
• preference order. The token's acr must equal one of them.

Returns:

• A decorator that records the OAuth step-up requirement on the endpoint.

Raises:

• TypeError — max_age is not an integer.
• ValueError — neither requirement is present, a value is invalid, or the
• same authentication class appears twice.

public#

public() -> Callable[[Any], Any]

Declare that an endpoint intentionally admits anonymous callers.

This changes no dispatch behavior. It is the positive declaration consumed by Wreath(require_access_declarations=True) and the route manifest, so a reviewed public endpoint is distinguishable from a forgotten guard.

second_factor#

second_factor(*, max_age: float = 300.0) -> Callable[[Any], Any]

Require a second factor proved within the last max_age seconds -- step-up.

Implies authenticated(). A caller with no identity is a 401; an identity that has not proved a factor recently enough is a 403 whose detail is second_factor_required, because they were identified and found wanting. The remediation is POST /auth/2fa/verify, which stamps the session and rotates its id, after which the same request succeeds.

@app.delete("/accounts/{account_id}")
@second_factor(max_age=300)
async def close_account(request: Request) -> dict:
    ...

This is what makes a second factor worth having in a framework that already owns authorization, rather than a login-time formality. The freshness lives on the route, so a handler never threads a "did they step up?" flag through its arguments, and a route that forgets to ask is not a route that quietly accepts a session from eight hours ago.

An identity that carries no second-factor stamp never satisfies this. A bearer token or an OIDC login has no such stamp, so a route guarded this way refuses it rather than treating an absent record as a fresh one. Guard only the actions that warrant re-prompting; a whole API behind this is a five-minute session with extra steps.

Args:

• max_age — seconds. Five minutes is the default -- long enough to survive
• a confirmation dialogue, short enough that a walked-away-from
• browser is not still authorized to delete things.

Returns:

• A decorator that records the requirement on the endpoint.

Raises:

• ValueError — max_age is not positive and finite. A window of zero or
• less can never be satisfied, while a non-finite window cannot
• express recency.

wreath.authorization#

Authorization: roles, permissions, policies, and authorization decisions.

Authorization determines what an authenticated identity may do. Establishing identity lives in wreath.auth. Both are thin public facades over the private wreath._auth implementation package.

LiveDocument is re-exported here because permission_document returns one: it is the per-principal document primitive -- an ETag and a change stream -- and permissions are its first caller.

ANY_SCOPE (value)#

ANY_SCOPE: _AnyScope = ANY_SCOPE

AuthRequirement#

AuthRequirement(authenticated: 'bool' = False, identify: 'bool' = False, role_checks: 'tuple[SetRequirement, ...]' = (), permission_checks: 'tuple[SetRequirement, ...]' = (), policies: 'tuple[PolicyRequirement, ...]' = (), second_factor: 'float | None' = None, oauth_step_up: 'OAuthStepUpRequirement | None' = None, public: 'bool' = False)

AuthRequirement(authenticated: bool = False, identify: bool = False, role_checks: tuple[SetRequirement, ...] = (), permission_checks: tuple[SetRequirement, ...] = (), policies: tuple[PolicyRequirement, ...] = (), second_factor: float | None = None, oauth_step_up: OAuthStepUpRequirement | None = None, public: bool = False) -> None

access_level (property)#

access_level: int

How much this endpoint asks of the caller: 0 nothing, 1 a principal, 2 an administrator.

The single definition of "is anything required here", asked by the HTTP and WebSocket dispatchers, by Wreath._authorize_request, and by MCP's _authorize -- which is the point. Four spellings of this question existed, differing in which fields they remembered to name, and second_factor was in only one of them: MCP admitted a tool declaring AuthRequirement(second_factor=300) without ever asking for the factor, because it read a level that did not count it. Every field that can refuse a caller is named here exactly once.

identify is deliberately absent: it asks the backend to answer, not the caller to satisfy anything, and an anonymous caller is one of its two acceptable answers. needs_backend adds it back for the one question it does change.

declares_access (property)#

declares_access: bool

Whether the endpoint explicitly says who may call it.

needs_backend (property)#

needs_backend: bool

Whether this endpoint needs identity resolution or an access check.

AuthorizationDecision#

AuthorizationDecision(allowed: 'bool', reason: 'str | None' = None, diagnostics: 'tuple[str, ...]' = ())

AuthorizationDecision(allowed: bool, reason: str | None = None, diagnostics: tuple[str, ...] = ()) -> None

AuthorizationProvider#

What Wreath.configure_auth asks of a policy authorizer.

The other half of the decision. Authentication established who; this decides whether that principal may perform one declared action on one resource. It is consulted only for endpoints carrying authorize policies — roles, permissions and second_factor are checked by the framework itself and never reach a provider.

Every authorize decorator on an endpoint contributes one PolicyRequirement, and they are asked in declaration order, and every one of them must allow. The first decision whose allowed is false ends the request with a 403 whose detail is that decision's reason (or "Forbidden" when it carries none), and the policies after it are not asked. There is no mode in which one policy admits a caller another refused.

A provider runs only after authentication, the second-factor window, the role checks and the permission checks have all passed — authorize implies authenticated, so request.identity is never None by then. CedarAuthorizer denies an anonymous request anyway rather than relying on that.

An absent provider refuses, and so does a broken one. An endpoint declaring a policy on an application configured without an authorizer raises RuntimeError when it is requested, and MCP refuses the call with an explanation naming the action; neither admits the caller because nothing was installed. A provider that raises is not a denial either — it becomes a 500 through the ordinary error boundary — so a provider that cannot reach what it needs should return a denying AuthorizationDecision rather than let the exception out.

AuthorizationProvider(*args, **kwargs)

authorize#

async authorize(request: Request, requirement: PolicyRequirement) -> AuthorizationDecision

Decide requirement.action against requirement.resource for this request.

requirement.resource is whatever the authorize decorator was handed, unresolved: when the decorator was given a callable it is still that callable, and calling it with the request — and awaiting the result if it is awaitable — is the provider's job. CedarAuthorizer does both.

Only allowed gates the request. reason becomes the 403's detail, so it reaches the client and should be written for one; diagnostics is carried on the decision for logging and is not sent.

AuthorizationVocabulary#

The complete, typed set of policy actions an application may declare.

Build it from one or more StrEnum classes and use those enum members in @authorize. Wreath stores the wire value, so the route manifest and Cedar still see ordinary strings while a misspelled action is caught by the type checker before startup.

AuthorizationVocabulary(*enums: type[StrEnum]) -> None

unknown#

unknown(actions: Iterable[str]) -> tuple[str, ...]

Declared actions absent from this vocabulary, sorted for diagnostics.

unused#

unused(actions: Iterable[str]) -> tuple[str, ...]

Vocabulary actions no route or mounted surface currently declares.

CedarAuthorizer#

The AuthorizationProvider that turns a request into a Cedar query.

This is the adapter, not the engine. It owns the five mappings from a Wreath request to Cedar's (principal, action, resource, context, entities) and hands them to whatever CedarEngine it was constructed with; the policy language, the evaluation order and the decision belong to the engine. Pair it with the built-in one and there is nothing else to wire:

engine = CedarPolicies(policy_text)
app.configure_auth(BearerTokenBackend(verify), CedarAuthorizer(engine=engine))

Each mapper is a constructor argument and can be replaced individually; the defaults model the common case, and are described in this module's own documentation. In short: the principal is EntityUid(identity.type, identity.id); the action is Action::"<name>" with the decorator's action string used verbatim as the id, never parsed or split; the resource passes through untouched; the entity set is the caller as one entity whose parents are Role::"..." for each of its roles, sorted; and the context is the request method and path, plus second_factor_age in seconds only when the identity has proved a factor.

A mapper may be synchronous or asynchronous — anything awaitable one returns is awaited before it is passed on.

There is deliberately no public accessor for the engine. A caller holding it could call is_authorized directly and bypass every mapper above, which is the work this class exists to do; fingerprint, source and policies hand out the engine's identifying values without handing out the engine.

CedarAuthorizer(*, engine: CedarEngine, principal: Callable[[Identity], object] = <function _default_principal>, action: Callable[[str, Request], object] = <function _default_action>, resource: Callable[[object, Request], object] = <function _default_resource>, resource_entities: Callable[[object, Request], object] | None = None, entities: Callable[[Request], object] = <function _default_entities>, context: Callable[[Request], Mapping[str, object]] = <function _default_context>, flags: Any = None, regions: Any = None, location: Callable[[Request], object] = <function _default_location>, organizations: Any = None, entitlements: Any = None, quota: Any = None, clock: Callable[[], object] = <built-in function time>) -> None

authorize#

async authorize(request: Request, requirement: PolicyRequirement) -> AuthorizationDecision

Map this request into a Cedar query, evaluate it, and normalize the answer.

An anonymous request is denied without consulting the engine, with the reason "anonymous". The route pipeline already refuses one before reaching here, since authorize implies authenticated; this is the second line, for a caller invoking the authorizer directly.

Otherwise: a callable requirement.resource is called with the request and awaited if need be, the five mappers run, and the engine's result goes through the shared normalizer — which denies any shape it does not recognize. See CedarEngine.is_authorized for what the shapes are.

facts_for#

facts_for(request: Request) -> dict[str, frozenset[str]]

Every declared set fact for this request, keyed by context attribute.

What the permission manifest tags with. Adding a fact must change the ETag, or a manifest outlives the membership or entitlement that produced it -- and enumerating them here rather than naming them one by one is what stops the next fact from being forgotten.

fingerprint (property)#

fingerprint: object

The engine's fingerprint, if it has one.

First of the three tags the permission manifest probes for. Delegated verbatim — this class computes nothing.

Raises:

• AttributeError — the engine offers no fingerprint. Absence stays
• absence — the probe reads these with getattr(..., None), so a
• raise is reported as "not offered" and it falls through to the
• next tag. A property that always resolved and returned None
• would instead promise that there is a fingerprint and that it
• is nothing.

flags_for#

flags_for(request: Request) -> frozenset[str]

This request's enabled flags, from the same resolution authorize uses.

The permission manifest tags its answer with everything that can change it, and once a policy can read context.flags that includes the flags. Reading them back through the authorizer keeps the provider and the vocabulary private to this class, and shares the per-request cache, so tagging a manifest costs no extra resolution.

policies (property)#

policies: object

The engine's policies, if it has one.

Last of the three tags the permission manifest probes for, and used only when it answers with bytes or str.

Raises:

• AttributeError — the engine offers no policies, on the same terms
• as fingerprint.

regions_for#

regions_for(request: Request) -> frozenset[str]

This request's containing regions, from the resolution authorize uses.

The manifest's counterpart to flags_for: once a policy can read context.regions, a caller's answer changes when they move, and a tag that ignored position would let a stale manifest outlive the geofence that produced it.

schema_owners (property)#

schema_owners: tuple[Any, ...]

Stores delegated through the organisation fact provider.

source (property)#

source: object

The engine's source, if it has one.

For CedarPolicies this is the policy text the set was parsed from, which is why the shipped configuration gets a content-derived manifest tag — the same on every worker and across a restart — rather than a per-instance one.

Raises:

• AttributeError — the engine offers no source, on the same terms as
• fingerprint.

CedarEngine#

The policy evaluator CedarAuthorizer delegates to.

One method, and it is a protocol rather than a base class so a different Cedar evaluator can be substituted without touching the adapter, the decorators, or anything that reads a decision. CedarPolicies — the built-in engine, parsed in Python and evaluated in C — satisfies it, and is what CedarAuthorizer(engine=...) is normally handed (Wreath implements Cedar itself).

Three attributes are optional and are read off the authorizer, which delegates each straight through to its engine: fingerprint, source, then policies, in that order, the first that answers with bytes or str winning. They exist to tag a cached permission manifest by the content of the policy set, so a client can revalidate an ETag across workers and across a restart. An engine offering none of them is entirely valid; the manifest then falls back to a per-instance token, which still changes when the policy set is replaced but agrees with no other worker.

CedarEngine(*args, **kwargs)

is_authorized#

is_authorized(*, principal: object, action: object, resource: object, context: Mapping[str, object], entities: object) -> object

Decide one request. Every argument is keyword-only.

CedarAuthorizer produces all five from the request through its mappers, awaiting any that come back awaitable, and awaits the result of this call too — so an engine may be synchronous or asynchronous.

The return value is normalized rather than trusted: an AuthorizationDecision is used as-is; a plain bool becomes a decision with the reason "cedar"; anything else is read for allowed and diagnostics attributes, with allowed defaulting to false when absent. An unrecognized shape therefore denies, and a reason on a duck-typed result is dropped — only an AuthorizationDecision carries its reason through to the 403.

CedarEntity#

One entity: its uid, its attributes, and its parents in the hierarchy.

CedarEntity(uid: EntityUid, attrs: Mapping[str, object] = <factory>, parents: tuple[EntityUid, ...] = ()) -> None

CedarParseError#

A policy set (or entity reference) that is not valid Cedar.

CedarParseError(message: str, line: int | None = None, column: int | None = None) -> None

CedarPolicies#

A parsed Cedar policy set that acts as the authorization engine.

Parsing happens once, at construction — a syntax error is an application bug and surfaces at startup, never during a request. entities given here are the static hierarchy (roles, groups, resource ownership); the per-request entities handed to is_authorized are merged over them.

CedarPolicies(source: str, *, entities: Iterable[CedarEntity] = (), schema: CedarSchema | str | None = None) -> None

context_attributes_for_action#

context_attributes_for_action(action: object) -> frozenset[str]

Context keys reachable by policies that can match this action.

Only a different exact-equality action scope proves irrelevance here. Hierarchical and set scopes remain in the candidate set because request entities can make them match even when their literal target differs.

is_authorized#

is_authorized(*, principal: object, action: object, resource: object, context: Mapping[str, object] | None = None, entities: object = None) -> AuthorizationDecision

Evaluate this policy set for one request. Keyword arguments only.

principal, action and resource are each an EntityUid or a Type::"id" string; anything else is a TypeError. context is a mapping converted into the compiled value model, and entities is a CedarEntity or an iterable of them, layered over the static hierarchy given to the constructor — a request entity sharing a static uid replaces it.

The decision follows Cedar: forbid overrides permit, the default is deny, and a policy that errors while evaluating is skipped and named in diagnostics rather than counting as satisfied. reason says which of the three it was — "explicit forbid", "cedar permit", or "no permit policy matched" — and diagnostics names every policy that matched or was skipped, and why. reason names no policy, which matters because it is what reaches the client as the 403's detail; the policy ids stay in diagnostics, which is not sent.

Evaluation runs in C. This does no parsing — that happened once, in __init__.

principal_entity_for_action#

principal_entity_for_action(action: object) -> bool

Whether this action can read request principal attributes or ancestry.

reads_context#

reads_context(attribute: str) -> bool

Whether any policy reads context.<attribute>, in any shape.

The scalar counterpart to the referenced_* walks. CedarAuthorizer asks it about delegated and actor to decide whether a delegated request needs a second evaluation, which is a question about presence rather than about members.

referenced_entitlements#

referenced_entitlements() -> frozenset[str] | None

Every entitlement this policy set tests, or None for "all".

referenced_flags#

referenced_flags() -> frozenset[str] | None

Every feature-flag name this policy set tests, or None for "all of them".

Two jobs. It is the vocabulary a CedarAuthorizer validates its flag provider against at startup, so context.flags.contains("new_iu") fails where it is written rather than denying quietly forever; and it is the set the authorizer resolves per request, which measured at a fifth of the cost of resolving every configured flag.

None means the policy set reads context.flags in a shape whose names are not statically knowable -- isEmpty(), or a computed argument -- so the caller must resolve them all. An empty set means no policy reads flags at all, which is the overwhelmingly common case and costs nothing.

Optional capability, probed with getattr the way source is — an outside CedarEngine that cannot answer simply gets the safe path.

referenced_org_roles#

referenced_org_roles() -> frozenset[str] | None

Every role-within-an-organisation this policy set tests, or None.

Unlike organisation ids these are a declared vocabulary, so a policy naming "acme:admni" is refused at startup. The qualified form carries an organisation id that cannot be enumerated, so validation checks the role half against the declared roles and lets the organisation half through -- see Memberships.names.

referenced_organizations#

referenced_organizations() -> frozenset[str] | None

Every organisation id this policy set tests, or None for "all".

Read for one purpose only, and it is not validation: an organisation id is a row, not a declared vocabulary, so refusing to boot because a policy names an organisation nobody has created yet would be wrong. What it decides is whether membership is resolved at all -- an empty answer means no policy reads context.organizations, so the caller skips the lookup entirely and the fact costs nothing.

referenced_quota#

referenced_quota() -> frozenset[str] | None

Every quota state this policy set tests, or None for "all".

referenced_regions#

referenced_regions() -> frozenset[str] | None

Every geofence region name this policy set tests, or None for "all".

The geospatial counterpart of referenced_flags, and the same two jobs: the vocabulary a CedarAuthorizer validates its region set against at startup, and the names it resolves per request.

The difference is what the work costs. A flag is a hash and a comparison; a region is a great-circle distance against the caller's position, so a policy naming two of a hundred regions does two of them. context.regions.contains(resource.reserve) — the shape a geofence is most naturally written in — has a computed argument, so this answers None and every region is resolved. That is the honest answer rather than a short one, and it is why a region set is worth keeping small.

source (property)#

source: str

The Cedar text this policy set was parsed from.

Public because it identifies the policy set by content, and callers that cache a decision against it need a tag that is the same on every worker and across a restart. id() is not that: CPython reuses addresses, so an address-derived tag can survive a reload that replaced the policies. Read-only — the parse happens once, in __init__, and a settable source would let the text drift from _policies.

CedarSchema#

A parsed Cedar schema that rejects bad policies before startup.

The supported input is Cedar's human-readable schema syntax. Action hierarchy declarations are also compiled into ordinary Action entity parents, so action in Action::"group" runs in the native evaluator without request-time schema work.

CedarSchema(source: str) -> None

action_parents#

action_parents(name: str) -> tuple[str, ...]

actions (property)#

actions: tuple[str, ...]

Action ids declared by this schema, in declaration order.

contexts#

contexts(actions: set[str]) -> tuple[Any, ...]

descendants#

descendants(name: str) -> frozenset[str]

require_action#

require_action(name: str) -> None

EntityUid#

A Cedar entity reference: a type path and an id, User::"alice".

EntityUid(type: str, id: str) -> None

parse (classmethod)#

parse(text: str) -> EntityUid

Parse Type::"id" (Cedar syntax) or the bare Type::id form.

Limits#

What a composed principal is restricted to, never what it is granted.

Every field is None for "not restricted" or a set for "at most these". A provider decides what a principal actually has; these narrow that answer and can never extend it. An application that composes human(user) | member_of("acme") for a caller the organisation store does not place in acme gets an empty membership set, not a fabricated one.

plan is the same rule in scalar form: claiming a plan the entitlement provider disagrees with yields no entitlements rather than the claimed plan's.

Limits(organizations: frozenset[str] | None = None, org_roles: frozenset[str] | None = None, entitlements: frozenset[str] | None = None, active_organization: str | None = None, plan: str | None = None) -> None

bound#

bound(attribute: str, resolved: frozenset[str]) -> frozenset[str]

Apply the restriction for attribute to what a provider resolved.

The single place a limit meets a provider's answer, so there is one place to read when asking whether composition can grant. It cannot: the result is a subset of resolved for every possible limit.

merge#

merge(other: Limits) -> Limits

Combine two limits by intersection, never by union.

active_organization and plan are scalars, so "combining" them is a last-writer-wins on the right-hand side -- composing member_of("a") | member_of("b") makes b active while leaving both in the membership restriction, which is the reading that cannot grant: the active organisation is a focus, and the restriction is the bound.

LiveDocument#

The change signal for one kind of per-principal document.

bus is optional and its absence is a supported configuration: a single-worker deployment or a test wants the local half without a database behind it. With a bus, a notification raised on the worker that took the write reaches the worker holding the stream in one hop -- and never travels a second one, because BusBridge has no path from receive to publish.

watch names the models whose writes make a document stale, as wreath._orm_events announces them. Those announcements are model-grained, not row-grained, so a write to the roles model wakes every subscriber rather than the one user who was promoted. That is the right trade here: the cost is one conditional refetch each, and a conditional refetch that finds nothing changed is a 304. Row-grained would mean recording a read set per request to save it.

fingerprint returns the part of every subscriber's tag that is not per-principal -- for permissions, the policy set and the route vocabulary. An open stream re-reads it on each keep-alive tick, so a policy set replaced in-process is noticed without anyone having to remember to call notify_all from a reload hook.

LiveDocument(*, channel: str, bus: Any = None, fingerprint: Callable[[], str] | None = None, watch: Iterable[Any] = (), watch_reason: str = 'write', max_subscribers: int = 1024, max_per_principal: int = 4, keepalive: float = 15.0) -> None

attached (property)#

attached: bool

Whether there is a bus to reach the other workers through.

channel (property)#

channel: str

The bus channel this document's change notifications travel on.

Fixed at construction, and read back from the bridge rather than stored twice. It names the channel whether or not a bus was supplied — an unattached document is one nobody publishes on, not one with no name.

close_all#

close_all() -> None

End every open stream. For shutdown, and for tests.

fingerprint#

fingerprint() -> str

The non-per-principal part of the tag, cached for a moment.

"" when the caller declared none, which turns the drift check in change_events off rather than making it lie.

keepalive (property)#

keepalive: float

Seconds a stream waits before it ticks, 15.0 by default.

The idle timeout every stream over this document uses unless it was opened with a keepalive of its own. A tick emits an SSE comment, which keeps a proxy from closing an idle stream and is also how a vanished client is discovered; it is also when the stream re-reads fingerprint, so this is the upper bound on how long a policy set replaced in-process can go unnoticed by an already-open subscriber.

notify#

notify(principal: str, reason: str, *, etag: str | None = None) -> None

One principal's document is stale, here and on the other workers.

notify_all#

notify_all(reason: str, *, etag: str | None = None) -> None

Every document is stale -- a policy reload, a deploy, a config change.

subscribe#

subscribe(principal: str) -> Subscription | None

A slot for one stream, or None when the registry is full.

Refused rather than queued or evicted. A caller without a stream is not broken -- it still revalidates the document with If-None-Match, which is this feature minus the push -- whereas evicting somebody else's tab invites a reconnect that evicts the next one.

subscribers (property)#

subscribers: int

How many streams are open right now, across every principal.

The number max_subscribers is checked against, so it is also how close this document is to refusing the next subscribe. It counts slots, not people: one principal holding four tabs is four.

watching (property)#

watching: bool

Whether this document is currently listening for ORM writes.

Narrowing#

A delegation: who is acting, what they may do, and until when.

Produced by Principal.narrow and carried on Identity.narrowing. The authorizer refuses an expired or out-of-scope narrowing before consulting the engine, and evaluates the delegating principal's own decision as a conjunct, so this can only ever subtract.

Args:

• actor — the agent doing the acting. Distinct from the subject, which
• stays the human -- an audit record that cannot name both has not
• recorded a delegation.
• scope — the action names this delegation permits, or None for the
• explicit ANY. Matched against the action= string a route declared,
• verbatim, because that string is already the id of the Action — :
• entity and inventing a second spelling here would let the two drift.
• expires_at — epoch seconds after which every decision denies, or None
• when the delegation carries no expiry of its own.
• on_behalf_of — the subject at the moment of narrowing, kept so a chain of
• sub-delegations still names the human at the bottom of it.
• depth — how many times this has been narrowed. A sub-agent's token is
• derived from its parent's, never a copy, and the depth is what makes
• "how many hops from a person" answerable in a policy or an audit
• record.

Narrowing(actor: str, scope: frozenset[str] | None = None, expires_at: float | None = None, on_behalf_of: str = '', depth: int = 1) -> None

expired#

expired(now: float) -> bool

Whether this delegation has run out, at now (epoch seconds).

permits#

permits(action: str) -> bool

Whether action is inside this delegation's scope.

An empty scope permits nothing. That is the fail-closed reading and the one worth stating: scope=set() is a plausible spelling of "no restrictions" and it means the opposite here. ANY_SCOPE is how the unrestricted case is written.

then#

then(other: Narrowing) -> Narrowing

Compose a further narrowing onto this one.

Scope intersects, expiry takes the earlier of the two, the actor becomes the innermost one, and the human at the bottom is preserved. A sub-agent therefore receives authority derived from its parent's rather than a copy of it, which is the mitigation for privilege escalation through an agent hierarchy.

PrecisionLadder#

An ordered set of actions, finest first, answering how precisely.

The generalisation of a withheld field from a boolean to a value. A withheld coordinate is either present or absent, which is the whole vocabulary a serializer normally has; real deployments want a third and a fourth answer, and they want the policy engine to choose between them:

ladder = PrecisionLadder(
    ("read_location_exact", None),      # exact
    ("read_location_fine", 1_000),      # 1 km
    ("read_location_coarse", 10_000),   # 10 km
)

Each rung is an ordinary Cedar action and an ordinary Cedar decision, asked in order until one permits. A caller permitted none sees no coordinate at all — absent, not null, which is the argument the camera-trap example already makes for a boolean and this extends to a scale.

Ordering is checked at declaration: a ladder whose rungs coarsen out of order, or repeat a resolution, is a mistake that would otherwise show up as a caller silently receiving a finer answer than the rung above allowed.

Args:

• rungs — (action, metres) finest first; metres of None means exact
• and is only valid on the first rung.

Raises:

• ValueError — no rungs, a duplicate action, an exact rung that is not
• first, or resolutions that do not strictly coarsen.

PrecisionLadder(*rungs: tuple[str, float | None]) -> None

actions#

actions() -> tuple[str, ...]

Every action this ladder asks, finest first.

The vocabulary declared_actions needs, so a ladder's rungs are part of the permission manifest rather than a second list that drifts from it.

apply#

apply(point: Coordinate | None, metres: float | None) -> Coordinate | None

point at the resolution metres names.

The one place a resolution becomes a value: None metres is exact, a number coarsens, and withheld (expressed by the caller passing no rung at all) never reaches here.

rungs (property)#

rungs: Sequence[tuple[str, float | None]]

The rungs, finest first, as declared.

Principal#

An identity, what it is limited to, and any delegation it arrived under.

Composed with | and narrowed with narrow; both return a new value and neither can widen. identity is what the auth stack carries, so a composed principal is installed by handing principal.identity to whatever sets request.identity.

principal = human(user) | member_of("acme", role="admin") | on_plan("pro")
delegated = principal.narrow(actor=agent_id, scope={"photos:read"}, ttl=300)

Principal(identity: Any, limits: Limits = Limits(organizations=None, org_roles=None, entitlements=None, active_organization=None, plan=None), narrowing: Narrowing | None = None) -> None

bind#

bind() -> Any

The Identity to install on a request, carrying limits and narrowing.

narrow#

narrow(*, actor: str, scope: ScopeArg, ttl: float | None = None, now: float | None = None) -> Principal

Delegate to actor, restricted to scope and expiring after ttl.

Args:

• actor — the agent that will act. Required and non-empty: a delegation
• nobody can name is one an audit record cannot record.
• scope — the action names permitted, or ANY_SCOPE. Required, with no
• default, because a defaulted security parameter gets defaulted.
• ttl — seconds until this delegation expires. None means it carries
• no expiry of its own -- which is not the same as unbounded, since
• a narrowing composed onto an already-expiring one keeps the
• earlier expiry.
• now — epoch seconds the ttl is measured from; defaults to the wall
• clock. Injectable so a test can pin it without patching.

Returns:

• A new Principal whose decisions are a subset of this one's.

Regions#

Named areas, and which of them contain a point.

The vocabulary a geofencing policy names. Each region is either a circle (a centre and a radius in metres) or a BoundingBox; a point is inside a circle when the great-circle distance is within the radius, and inside a box when the box contains it.

regions = Regions(
    depot=(Coordinate(lat=-25.1, lon=133.4), 5_000),
    reserve=BoundingBox(-26.0, -24.0, 132.0, 135.0),
)

Names are the strings a policy writes, so they are validated once here rather than failing as a silent non-match later: a name that is not a non-empty string is refused where the region is declared.

Args:

• regions — name to (centre, radius_metres) or BoundingBox.

Raises:

• ValueError — a name is empty or not a string, a radius is not positive,
• or a region is neither a circle pair nor a BoundingBox.

Regions(regions: Mapping[str, Any] | None = None, **named: Any) -> None

containing#

containing(point: Coordinate, names: Iterable[str] | None = None) -> frozenset[str]

The declared regions containing point.

Args:

• names — resolve only these, which is what the authorizer passes when
• the policy set's references are statically knowable. None
• resolves every declared region — the honest answer when a policy
• computes the name it tests.

names#

names() -> frozenset[str]

Every declared region name.

The enumeration CedarAuthorizer validates a policy set against at startup, so a misspelled region fails where it is written rather than never matching.

authorize#

authorize(*, action: str | StrEnum, resource: object | Callable[[Any], object]) -> Callable[[Any], Any]

Require a policy decision from the configured authorizer.

Implies authenticated(): the decision needs a principal, so the endpoint is protected and a caller with no identity is refused before the policy is ever consulted. A policy that denies is a 403, not a 401 -- the caller was identified and found wanting, which is a different remediation.

action and the resource type are also the vocabulary permissions_router and wreath typegen read off the routes, so the strings here are the ones a UI asks about. There is no second list to keep in step.

Args:

• action — The action name the policy is asked about, e.g. "read".
• resource — The resource the action applies to. A callable receives the
• request and returns the resource, for a target that is only known
• per request; anything else is used as-is.

Returns:

• A decorator that records the requirement on the endpoint.

Raises:

• ValueError — action is empty.

coarsen#

coarsen(point: Coordinate, metres: float) -> Coordinate

point rounded onto a fixed grid of roughly metres on a side.

The grid is absolute, not relative to the point, and there is no randomness anywhere in it. That is the whole security property: the alternative — adding jitter per request — averages away, so an attacker who can ask repeatedly recovers the true position to arbitrary precision. A grid cell reveals the cell and nothing further however many times it is asked, which is why coarsen is a pure function of (point, metres) and takes no seed, no salt and no request.

The cell is at least metres in both dimensions. Latitude rounds on a constant step; longitude rounds on that step divided by the cosine of the rounded latitude, so a cell near the poles stays as wide on the ground as one at the equator and the band a point falls in cannot depend on the part of its own longitude being removed. Within roughly 30 km of a pole a cell would span the globe, and the longitude collapses to 0 rather than pretending to a precision the grid cannot express.

Args:

• metres — cell size. Must be positive and finite.

Raises:

• ValueError — metres is not a positive finite number.

declared_actions#

declared_actions(app: Any) -> dict[str, tuple[str, ...]]

Resource type -> the actions this application enforces on it.

Read off the declarations themselves, so it cannot disagree with what is enforced. All four protocols wreath serves land here, and the differences between them are only in where the declaration sits:

• REST declares @authorize(action=...) on a route handler.
• gRPC declares action= on a method, which is a route -- so it arrives through the same requirement, with no second path.
• MCP declares action= on a tool, a resource or a prompt, and its endpoint advertises them through SURFACE_ACTIONS.
• GraphQL declares a policy per field, and its endpoint advertises the schema's types the same way.

A surface with nothing enforcing it contributes nothing, exactly as a route with no policy does.

human#

human(identity: Any) -> Principal

Start a composition from an authenticated identity.

member_of#

member_of(organization: str, *, role: str | None = None) -> Limits

Restrict to one organisation, optionally to one role within it.

The role is namespaced by the organisation ("acme:admin"), because a role name alone cannot say where it applies and an admin of one tenant is not an admin of another. That spelling is what context.org_roles carries.

on_plan#

on_plan(name: str) -> Limits

Restrict to the entitlements of one plan.

Claiming a plan the entitlement provider disagrees with yields no entitlements rather than the claimed plan's, which is the composition law in its most tempting-to-get-wrong form.

permission_document#

permission_document(app: Any, *, bus: Any = None, roles_model: Any = None, channel: str = 'wreath_permissions', keepalive: float = 15.0, max_subscribers: int = 1024, max_per_principal: int = 4) -> LiveDocument

The change signal behind {prefix}/stream.

Build one yourself when you need a handle on it -- to call document.notify_all("policies") from a reload hook, or document.close_all() at shutdown -- and pass it to permissions_router. Otherwise that function builds one.

bus (app.messaging(...)) is what makes a role change on the worker that took the write reach the worker holding the stream. Without it the signal is local, which is right for one worker and for tests.

roles_model is your role-membership model, or its name; a committed write to it is what makes one user's manifest stale. It has to be named because Wreath cannot know which of your tables grants a role -- and without it the stream still works, still notices a policy-set change, and still delivers an explicit notify.

permissions#

permissions(*values: str, mode: Mode = 'all') -> Callable[[Any], Any]

Require the identity to carry named permissions.

Implies authenticated(). The same 401/403 split as roles: no identity is a 401, an identity without the permissions is a 403.

An SSO identity carries permissions as well as roles, so a route guarded this way admits the same person whether they arrived with a bearer token or a browser session.

Args:

• values — Permission names. Empty strings are discarded.
• mode — "all" (the default) demands every permission; "any" demands one.

Returns:

• A decorator that records the requirement on the endpoint.

Raises:

• ValueError — mode is not "all" or "any", or no non-empty permission
• was given.

permissions_router#

permissions_router(app: Any, *, prefix: str = '/permissions', manifest_path: str = '/manifest', stream_path: str = '/stream', bus: Any = None, roles_model: Any = None, document: LiveDocument | None = None, max_ids: int = 200, max_concurrency: int = 8) -> Router

Routes that answer what the caller may do, from the app's own policies.

GET {prefix} the vocabulary, so a client can discover it GET {prefix}/manifest what this caller may ever do (ETag) GET {prefix}/stream SSE: that manifest moved, ask again POST {prefix} what this caller may do to these rows

Mount it with app.include_router(permissions_router(app)). The application is passed in because every answer is read off its routes and evaluated by its authorizer -- the endpoint cannot drift from enforcement even in principle, and it is only ever read at request time, so the routes it describes may be declared after this call.

All four require an authenticated caller, the vocabulary included. It is the complete map of the authorization surface -- every resource type and every action the application enforces -- and nothing runtime needs it anonymously: the generated client bakes the vocabulary in at build time, from the app object, never over HTTP.

max_ids bounds one batch request. The endpoint runs len(ids) x len(actions) policy evaluations, so an unbounded list is a denial of service one authenticated caller can post. Over the ceiling it refuses, naming the limit, rather than truncating: a truncated answer draws a UI that is confidently wrong, and an absent answer is only absent.

max_concurrency bounds how many of those evaluations are in flight at once. It matters only for a remote authorizer, which the CedarEngine protocol invites: the built-in engine is in-process and the loop never yields, so concurrency there is the same work in the same order. Evaluating a full batch one ask at a time is max_ids x len(actions) round trips end to end; this makes it that over max_concurrency. It is deliberately not unbounded -- gathering the whole product would turn one caller's batch into a burst of that size against a service that is not yours to overload.

An authorizer must therefore tolerate concurrent calls carrying one request. The built-in one does; it reads the request and never writes to it. If yours cannot, pass max_concurrency=1 and evaluation returns to strictly sequential.

bus and roles_model configure the stream; see permission_document, which builds the document when you do not pass one.

roles#

roles(*values: str, mode: Mode = 'all') -> Callable[[Any], Any]

Require the identity to carry named roles.

Implies authenticated(). A caller with no identity is a 401; a caller whose identity lacks the roles is a 403.

Args:

• values — Role names. Empty strings are discarded.
• mode — "all" (the default) demands every role; "any" demands one.

Returns:

• A decorator that records the requirement on the endpoint.

Raises:

• ValueError — mode is not "all" or "any", or no non-empty role was
• given -- a requirement satisfied by everyone is a mistake, not a
• permissive default.

with_entitlements#

with_entitlements(*names: str) -> Limits

Restrict to at most these entitlements, whatever the plan grants.

wreath.users#

User-management lifecycle flows — the fastapi-users equivalent.

Builds registration / login / logout / email-verification / password-reset / me on top of wreath's existing auth: login writes the same signed-session principal ({"sub", "type", "roles"}) that wreath.auth.SessionIdentityBackend already reads, so this integrates with the auth stack rather than reinventing it. It writes no exp and no permissions, so a session minted here lasts the session cookie's own lifetime and carries roles only.

The security-sensitive core (scrypt hashing, HMAC action tokens, flow logic) lives in the stdlib-only wreath._userkit; this module is the thin wreath glue.

store = InMemoryUserStore() # or an ORM-backed UserStore sessions = PostgresSessionStore(database) app.include_router( user_router(store, sessions=sessions, secret=SECRET, base_url="https://app") )

SessionPolicy is required for login: without it POST /login answers 500 rather than signing anyone in. POST /logout and GET /me degrade instead -- logout reports success having cleared nothing, and /me answers 401.

Email delivery is a pluggable EmailSender. The default LogEmailSender logs a redacted delivery notice; an explicit LogEmailSender(expose_tokens=True) is a local-development escape hatch. SmtpEmailSender (stdlib smtplib, STARTTLS or implicit TLS on port 465) is the shipped production transport.

AttemptLimiter#

Atomic keyed admission for authentication attempts.

RateLimitPolicy implements this contract. Back it with PostgresRateLimitStore when a limit must span workers; admit_key then makes the refill, consume, and admission decision in one database statement. It returns None when admitted and a refusal response when the key is over budget.

AttemptLimiter(*args, **kwargs)

admit_key#

async admit_key(key: str) -> Any | None

BulkSecondFactorRemovalStore#

A second-factor store that can remove one credential set together.

remove_many#

async remove_many(user_id: str, credential_ids: Sequence[str]) -> None

Delete the named credentials only when they belong to user_id.

add (inherited from SecondFactorStore)#

async add(credential: SecondFactor) -> SecondFactor

Store credential and return what was stored.

The id is the caller's, not the store's, and must be treated as taken: overwriting an existing one would drop a live credential and, for TOTP, the replay counter with it. InMemorySecondFactorStore raises ValueError on a duplicate id and an implementation should do the same rather than replace.

credentials (inherited from SecondFactorStore)#

async credentials(user_id: str) -> list[SecondFactor]

Every credential registered to user_id, or an empty list.

A user with no second factor is an empty list, never None — "has none" is an answer, and the flows read it as one.

remove (inherited from SecondFactorStore)#

async remove(user_id: str, credential_id: str) -> None

Delete credential_id, only if it belongs to user_id.

The ownership check is the implementation's to make and is not optional: the credential id is the only thing a caller supplies, so a store that deletes by id alone lets any user delete any user's factor.

Removing something that is not there is not an error. The outcome asked for is that the credential is gone, and it is.

touch (inherited from SecondFactorStore)#

async touch(credential_id: str, *, counter: int, at: datetime) -> bool

Advance the replay counter, atomically, and report whether this won.

Store counter and at against credential_id only when counter is strictly greater than the counter already stored, and return True exactly when the store did. Anything else returns False: an equal or lower counter, and a credential that is not there.

The return value is the whole point — see the class docstring for why a read-then-write cannot be substituted for a conditional one.

BulkSecondFactorStore#

A second-factor store that can persist one credential set together.

add_many#

async add_many(credentials: Sequence[SecondFactor]) -> tuple[SecondFactor, ...]

Store every credential and return them in input order.

add (inherited from SecondFactorStore)#

async add(credential: SecondFactor) -> SecondFactor

Store credential and return what was stored.

The id is the caller's, not the store's, and must be treated as taken: overwriting an existing one would drop a live credential and, for TOTP, the replay counter with it. InMemorySecondFactorStore raises ValueError on a duplicate id and an implementation should do the same rather than replace.

credentials (inherited from SecondFactorStore)#

async credentials(user_id: str) -> list[SecondFactor]

Every credential registered to user_id, or an empty list.

A user with no second factor is an empty list, never None — "has none" is an answer, and the flows read it as one.

remove (inherited from SecondFactorStore)#

async remove(user_id: str, credential_id: str) -> None

Delete credential_id, only if it belongs to user_id.

The ownership check is the implementation's to make and is not optional: the credential id is the only thing a caller supplies, so a store that deletes by id alone lets any user delete any user's factor.

Removing something that is not there is not an error. The outcome asked for is that the credential is gone, and it is.

touch (inherited from SecondFactorStore)#

async touch(credential_id: str, *, counter: int, at: datetime) -> bool

Advance the replay counter, atomically, and report whether this won.

Store counter and at against credential_id only when counter is strictly greater than the counter already stored, and return True exactly when the store did. Anything else returns False: an equal or lower counter, and a credential that is not there.

The return value is the whole point — see the class docstring for why a read-then-write cannot be substituted for a conditional one.

CapturingEmailSender#

Test sender — records (email, link) pairs instead of sending.

CapturingEmailSender(verifications: list[tuple[str, str]] = <factory>, resets: list[tuple[str, str]] = <factory>, messages: list[Message] = <factory>, suppression: SuppressionList | None = None) -> None

send#

async send(message: Message) -> None

Append message to messages, applying suppression if one is set.

The suppression check is duplicated here rather than inherited so that a test using this sender exercises the same refusal a real one would -- a capturing double that accepts what production refuses is how a suppression bug reaches production green.

send_password_reset#

async send_password_reset(email: str, link: str) -> None

Append (email, link) to resets. Nothing is delivered.

send_verification#

async send_verification(email: str, link: str) -> None

Append (email, link) to verifications. Nothing is delivered.

DiscoverableSecondFactorStore#

The additional indexed lookup required by first-factor passkeys.

Kept separate from SecondFactorStore so existing TOTP and second-factor WebAuthn stores remain valid. second_factor_router(passkey_login=True) refuses at construction unless this extension is implemented.

credential#

async credential(credential_id: str) -> SecondFactor | None

Look up one credential by its framework id, or return None.

Discoverable login derives this opaque id from the public WebAuthn credential id with the same one-way function enrolment uses. The lookup must be indexed; enumerating every user's credentials makes login proportional to the deployment's account count.

add (inherited from SecondFactorStore)#

async add(credential: SecondFactor) -> SecondFactor

Store credential and return what was stored.

The id is the caller's, not the store's, and must be treated as taken: overwriting an existing one would drop a live credential and, for TOTP, the replay counter with it. InMemorySecondFactorStore raises ValueError on a duplicate id and an implementation should do the same rather than replace.

credentials (inherited from SecondFactorStore)#

async credentials(user_id: str) -> list[SecondFactor]

Every credential registered to user_id, or an empty list.

A user with no second factor is an empty list, never None — "has none" is an answer, and the flows read it as one.

remove (inherited from SecondFactorStore)#

async remove(user_id: str, credential_id: str) -> None

Delete credential_id, only if it belongs to user_id.

The ownership check is the implementation's to make and is not optional: the credential id is the only thing a caller supplies, so a store that deletes by id alone lets any user delete any user's factor.

Removing something that is not there is not an error. The outcome asked for is that the credential is gone, and it is.

touch (inherited from SecondFactorStore)#

async touch(credential_id: str, *, counter: int, at: datetime) -> bool

Advance the replay counter, atomically, and report whether this won.

Store counter and at against credential_id only when counter is strictly greater than the counter already stored, and return True exactly when the store did. Anything else returns False: an equal or lower counter, and a credential that is not there.

The return value is the whole point — see the class docstring for why a read-then-write cannot be substituted for a conditional one.

EmailSender#

Delivery seam. SmtpEmailSender below is the shipped transport.

EmailSender(*args, **kwargs)

send_password_reset#

async send_password_reset(email: str, link: str) -> None

Deliver the password-reset link to a user who asked for one.

The same contract as send_verification, and the same reason to be quick: "forgot password" answers identically whether or not the address is registered, and only the sender can make the timing match too.

send_verification#

async send_verification(email: str, link: str) -> None

Deliver the "confirm your address" link to a newly registered user.

link is built by the application's link_builder and already carries the signed token; the sender's job is delivery and nothing else.

Registration awaits this, so a slow or failing transport is a slow or failing registration. It is also the last remaining timing signal that distinguishes a new address from one already registered — the flow is uniform in every other respect — so a transport that queues rather than delivers inline closes that gap as well as the latency one.

InMemorySecondFactorStore#

A dict-backed SecondFactorStore for development and tests.

Per-process and unshared, exactly like InMemoryUserStore: two workers do not see each other's credentials, and the replay counter is per process too, so this is not a store to run a real second factor on. Use wreath.users.OrmSecondFactorStore, or your own, in production.

InMemorySecondFactorStore(_rows: dict[str, SecondFactor] = <factory>) -> None

add#

async add(credential: SecondFactor) -> SecondFactor

Store credential.

Raises:

• ValueError — the id is already taken. Silently overwriting would drop
• a live credential -- and, for TOTP, its replay counter with it.

add_many#

async add_many(credentials: Sequence[SecondFactor]) -> tuple[SecondFactor, ...]

Store one credential set without suspending between rows.

credential#

async credential(credential_id: str) -> SecondFactor | None

Return one credential by id, or None.

credentials#

async credentials(user_id: str) -> list[SecondFactor]

Every credential belonging to user_id, in insertion order.

remove#

async remove(user_id: str, credential_id: str) -> None

Delete one credential, but only if it belongs to user_id.

A miss is not an error; the outcome asked for is that the credential is gone. The ownership check is not decoration -- the id is the only thing a caller passes, so without it any user could delete any factor.

remove_many#

async remove_many(user_id: str, credential_ids: Sequence[str]) -> None

Delete one owned credential set.

touch#

async touch(credential_id: str, *, counter: int, at: datetime) -> bool

Record that credential_id was just used at step counter.

Returns:

• True when the counter advanced. False when it did not -- a counter
• at or below the stored one, or a credential that is no longer there.
• A caller verifying a code must read False as a replay — either the
• step was already spent, or another request spent it while this one
• was checking the code. It is not an error, and it is deliberately
• not an exception — a losing racer is an ordinary refusal, and two
• requests arriving with the same code is exactly what this is for.

InMemoryUserStore#

A dict-backed store for dev and tests. Emails are normalized (lower/trim).

InMemoryUserStore(_by_id: dict[str, UserRecord] = <factory>, _by_email: dict[str, str] = <factory>, _seq: int = 0) -> None

compare_and_set_password#

async compare_and_set_password(user_id: str, expected: str, replacement: str) -> bool

create#

async create(email: str, hashed_password: str) -> UserRecord

Store a new user under the next id in a per-instance counter.

Ids are "1", "2", and so on, which is stable within one process and means nothing outside it. The email is normalized before it is stored, so the record carries the trimmed, lower-cased form. created_at and updated_at are both set to now.

get_by_email#

async get_by_email(email: str) -> UserRecord | None

The user registered under this address, trimmed and lower-cased first.

get_by_id#

async get_by_id(user_id: str) -> UserRecord | None

The user with this id, or None.

get_many_by_id#

async get_many_by_id(user_ids: Iterable[str]) -> list[UserRecord | None]

One ordered dictionary lookup per id, retaining repeats and misses.

update#

async update(user: UserRecord) -> UserRecord

Store user against its own id, stamping updated_at to now.

An upsert: a record whose id is not present is simply inserted. The email index gains an entry for the record's current address, and a previous address the record no longer carries is not removed from it, so a changed email stays resolvable under the old one. That has no effect on the shipped flows, which never change an address — but it is a reason not to reach for this store beyond development and tests.

LogEmailSender#

Development sender that logs delivery without exposing tokens by default.

Pass expose_tokens=True only for a local development process when a link must be copied into a browser. The safe default records that delivery would have happened without putting a bearer token into logs.

Args:

• printer — Where a line goes. print by default; pass a logger's method
• to route it somewhere durable.
• expose_tokens — Include bearer-token links in output. Disabled by default.

LogEmailSender(printer: Callable[[str], None] = <built-in function print>, *, expose_tokens: bool = False) -> None

send#

async send(message: Message) -> None

Print one line for message, including its class. Nothing is sent.

send_password_reset#

async send_password_reset(email: str, link: str) -> None

Log reset delivery, redacting the link unless opted in.

send_verification#

async send_verification(email: str, link: str) -> None

Log verification delivery, redacting the link unless opted in.

OrmSecondFactorStore#

Reference SecondFactorStore over a wreath ORM session and a model.

The model must carry the columns default_second_factor_model() declares. Reads use Model.select().where(...) with session.fetch; writes use the unit-of-work API -- session.add(instance) or session.delete(instance) then await session.flush(). Nothing here commits: the transaction belongs to whoever opened the session.

Two refusals in here are security properties rather than tidiness, and both are raises so that python -O cannot delete them:

• remove deletes only when the row's user_id matches the one passed. The credential id is the only thing an HTTP caller supplies, so without the check any user could delete any other user's second factor by guessing or observing an id.
• touch advances the counter in one conditional statement, and says whether it won. That counter is the replay defence, and a read-then-write cannot enforce it: verification reads the credential, awaits the code check, and only then writes, so two requests carrying the same observed code both read the same counter and a plain UPDATE ... SET counter = $1 lets both of them through. WHERE counter < $1 RETURNING 1 is the whole fix -- PostgreSQL decides, once, and the loser is told it lost.

Args:

• session — an open ORM session; every method awaits it and none commits.
• model — the ORM model class carrying the columns above.

OrmSecondFactorStore(session: Any, model: type[Any]) -> None

add#

async add(credential: SecondFactor) -> SecondFactor

Insert one credential and flush; returns the argument.

The row is added and flushed, not committed, so it is visible to later statements in the same transaction and lands when that transaction does. credential.id is written through, so the id the caller minted is the id the database holds -- which is what lets confirm_totp_enrolment return a credential that can be looked up again.

add_many#

async add_many(credentials: Sequence[SecondFactor]) -> tuple[SecondFactor, ...]

Insert one credential set with a single unit-of-work flush.

credential#

async credential(credential_id: str) -> SecondFactor | None

Return one credential by indexed primary key, or None.

credentials#

async credentials(user_id: str) -> list[SecondFactor]

Every credential belonging to user_id. An empty list is an answer.

remove#

async remove(user_id: str, credential_id: str) -> None

Delete one credential, but only when it belongs to user_id.

A row that does not exist is not an error -- the outcome asked for is that it is gone. A row belonging to somebody else is silently not deleted, which is the same observable answer, so this cannot be used to probe which ids exist.

remove_many#

async remove_many(user_id: str, credential_ids: Sequence[str]) -> None

Delete one user's named credentials with one read and one flush.

store_id (property)#

store_id: object

The model, database, and tenant. OrmUserStore.store_id says why.

touch#

async touch(credential_id: str, *, counter: int, at: Any) -> bool

Record a use -- the newest accepted step and when -- if it is newer.

One statement, no read first: the counter is compared and set inside PostgreSQL, so of two requests replaying the same code exactly one updates a row. Written as a read, a check and a write, both would.

This is deliberately not the unit-of-work path the other writers here use. A loaded row mutated and flushed carries the value this session read before it awaited anything, which is the stale value the race is made of.

Returns:

• True when this call advanced the counter. False when it did not --
• the stored counter is already at or past counter, or the
• credential is gone. wreath._secondfactor.verify_second_factor
• reads False as a replay and refuses.

OrmUserStore#

Reference UserStore over a wreath ORM session and a user model.

The model must carry id, email, hashed_password, is_active and is_verified; default_user_model() builds one, and any model naming those columns works. Reads use Model.select().where(...) with session.fetch_one and session.get; writes use wreath's unit-of-work API, session.add(instance) then await session.flush() (there is no session.update — a loaded row is mutated and flushed). Nothing here commits: the transaction belongs to whoever opened the session.

It hashes nothing. create stores the string it is handed, and the flows in wreath._userkit hash before calling it. Handing it a plaintext password stores a plaintext password.

Emails are normalized with strip().lower() on create and on get_by_email, so a lookup matches rows this store wrote whatever case the caller used. The comparison is a plain SQL = rather than lower(email) or a citext column, so a row some other writer inserted with upper-case characters is not found — the case-insensitivity is normalization on both sides, not a case-insensitive column.

created_at/updated_at are a stated gap: no method writes them and _to_record does not read them, so every UserRecord this store returns carries 0.0 for both.

Args:

• session — an open ORM session; every method awaits it and none commits.
• model — the ORM model class carrying the columns above.

OrmUserStore(session: Any, model: type[Any]) -> None

compare_and_set_password#

async compare_and_set_password(user_id: str, expected: str, replacement: str) -> bool

create#

async create(email: str, hashed_password: str) -> UserRecord

Insert a user from an already-hashed password and return the record.

hashed_password is stored verbatim. The email is normalized with strip().lower(); is_active and is_verified take the model's own defaults. Uniqueness belongs to the database — the reference model marks email unique, so a duplicate surfaces as an error from the flush rather than as an existing record.

The row is added and flushed, not committed, so it is visible to later statements in the same transaction and lands when that transaction does.

get_by_email#

async get_by_email(email: str) -> UserRecord | None

Return the user with this email, or None when no row matches.

The argument is normalized with strip().lower() before the comparison, which is what makes the lookup case-insensitive for rows create wrote. A miss is an answer, not an error.

get_by_id#

async get_by_id(user_id: str) -> UserRecord | None

Return the user with this primary key, or None when no row matches.

user_id reaches session.get unconverted, so the model's primary-key column has to accept a str. The Uuid key default_user_model declares does not: the ORM coerces a comparison value against the column type and raises TypeError: expected UUID, got str. Pair a uuid model with a store that converts, or give the model a text primary key.

get_many_by_id#

async get_many_by_id(user_ids: Iterable[str]) -> list[UserRecord | None]

Return ordered users and misses after one query for all supplied ids.

Repeated ids repeat their result without repeating the database query. An empty input returns without crossing into the session. Primary-key values follow the same typing rule as get_by_id.

store_id (property)#

store_id: object

The model, database, and tenant that decide the rows served.

Two of these over one model are one store wearing two objects, and a deployment that builds one inline for each router has exactly that. See UserStore for the convention and _same_store for what reads it; the session itself is deliberately not part of it, since two sessions over one database and tenant still see the same table.

update#

async update(user: UserRecord) -> UserRecord

Write user back over its row and flush. Returns the argument.

Not a partial update. The row is loaded by primary key and then email, hashed_password, is_active and is_verified are all assigned from user, so a record that was read, changed in one field, and passed back rewrites the other three with whatever it was still holding. Loading it by primary key carries the same typing rule as get_by_id.

created_at/updated_at are left untouched, the email is written as given rather than normalized, and the returned record is the argument rather than a re-read of the row. A user.id with no row raises AttributeError on the first assignment — this never inserts.

SecondFactor#

One enrolled second factor: a TOTP secret, or one recovery code's hash.

Frozen, because the two mutable parts -- the counter and the last-used stamp -- move through SecondFactorStore.touch so that a store can enforce that the counter only ever goes forwards. Read a changed copy back from the store rather than assigning to one of these.

material is deliberately opaque and deliberately not in the repr: for a TOTP credential it is the shared secret, and a dataclass repr reaches logs, tracebacks and error reporters without anyone choosing to put it there. Nothing in wreath.users ever renders it.

Args:

• kind — totp or recovery today; webauthn is accepted and reserved.
• label — shown to the user when listing factors ("iPhone", "1Password").
• counter — for TOTP, the newest step accepted; 0 means none yet.

Raises:

• ValueError — an empty id or user id, or a kind outside the three above.

SecondFactor(id: str, user_id: str, kind: Kind, label: str, created_at: datetime, last_used_at: datetime | None, material: bytes, counter: int = 0) -> None

SecondFactorStore#

Persistence seam for second factors, sibling to UserStore.

A separate protocol rather than four more methods on UserStore: a user has zero or many credentials, and get/create/update is the wrong shape for a collection. InMemorySecondFactorStore and wreath.users.OrmSecondFactorStore both implement it.

An implementation owes two guarantees the flows rely on and cannot check:

• remove must be scoped to user_id, so one user cannot delete another's factor by id.
• touch must be an atomic conditional advance, and must report whether it won. It stores counter and at only when counter is strictly greater than the counter already stored, and returns True exactly when it did. A credential that is not there loses, like any other non-advance.

A plain UPDATE ... SET counter = $1 is not enough, and that is the whole reason this method returns something. Verification reads the credential, awaits, checks the code, awaits, and only then writes -- so two requests carrying the same observed code both read the same stored counter and both write it, and the second one succeeds. Racing the legitimate user with a code read over their shoulder is the attack the counter exists to stop, so the comparison has to happen where the write happens: WHERE counter < $1 in the same statement, test_and_set in the same non-suspending function.

SecondFactorStore(*args, **kwargs)

add#

async add(credential: SecondFactor) -> SecondFactor

Store credential and return what was stored.

The id is the caller's, not the store's, and must be treated as taken: overwriting an existing one would drop a live credential and, for TOTP, the replay counter with it. InMemorySecondFactorStore raises ValueError on a duplicate id and an implementation should do the same rather than replace.

credentials#

async credentials(user_id: str) -> list[SecondFactor]

Every credential registered to user_id, or an empty list.

A user with no second factor is an empty list, never None — "has none" is an answer, and the flows read it as one.

remove#

async remove(user_id: str, credential_id: str) -> None

Delete credential_id, only if it belongs to user_id.

The ownership check is the implementation's to make and is not optional: the credential id is the only thing a caller supplies, so a store that deletes by id alone lets any user delete any user's factor.

Removing something that is not there is not an error. The outcome asked for is that the credential is gone, and it is.

touch#

async touch(credential_id: str, *, counter: int, at: datetime) -> bool

Advance the replay counter, atomically, and report whether this won.

Store counter and at against credential_id only when counter is strictly greater than the counter already stored, and return True exactly when the store did. Anything else returns False: an equal or lower counter, and a credential that is not there.

The return value is the whole point — see the class docstring for why a read-then-write cannot be substituted for a conditional one.

SmtpEmailSender#

Real SMTP delivery via stdlib smtplib/email (zero-dep).

STARTTLS by default (or implicit TLS on port=465). The blocking smtplib work runs in a worker thread so the event loop is never blocked. Build from env with from_env (WREATH_SMTP_HOST/_FROM/_PORT/_USER/_PASSWORD/_TLS).

Three things beyond the transport, each of which a 2026 sender needs and none of which an SMTP library supplies:

• dkim signs the exact bytes handed to the MTA. Pair it with SPF and a DMARC policy; wreath.doctor.check_email_deliverability reports whether the DNS actually backs the configuration up, which is the failure this class would otherwise be silent about.
• suppression is consulted before every send. Marketing mail to a suppressed address is refused; transactional mail is delivered, because a newsletter complaint must not cost someone their password reset.
• Message.mail_class decides both of those, and has no default.

Counters (sent, suppressed, transport_failures) are readable at any time. They exist because the failure mode here is silence: mail that "sent", nothing raised, and nobody received it.

SmtpEmailSender(host: str, from_addr: str, port: int = 587, username: str | None = None, password: str | None = None, use_tls: bool = True, timeout: float = 30.0, verify_subject: str = 'Verify your email', reset_subject: str = 'Reset your password', dkim: DkimSigner | None = None, suppression: SuppressionList | None = None, sent: int = 0, suppressed: int = 0, transport_failures: int = 0) -> None

build#

build(message: Message) -> bytes

Serialise message to the exact bytes that go on the wire.

Public because it is what the tests sign and verify, and because a caller integrating a different transport needs the same bytes: the signature covers these bytes and no others. Re-serialising an EmailMessage after signing it is the most common way a valid DKIM signature becomes an invalid one, which is why this returns bytes and the sender hands those bytes straight to sendmail.

from_env (classmethod)#

from_env() -> SmtpEmailSender

Build a sender from the WREATH_SMTP_* environment variables.

WREATH_SMTP_HOST and WREATH_SMTP_FROM are required. _PORT defaults to 587, _USER and _PASSWORD to unset (no login), and _TLS to on — it is off only for the literal values 0, false or False, so a typo leaves TLS enabled rather than silently disabling it.

The subjects and the timeout are not read from the environment; set them on the instance if you want them changed.

Raises:

• KeyError — WREATH_SMTP_HOST or WREATH_SMTP_FROM is not set.
• ValueError — WREATH_SMTP_PORT is not an integer.

send#

async send(message: Message) -> None

Deliver one Message, honouring its class.

Raises:

• SuppressedError — message is marketing and its recipient is
• suppressed. Raised rather than returned quietly, because a
• caller that keeps sending to a suppressed list is the thing that
• drives a domain past the 0.10% complaint threshold, and it
• should find out.
• OSError — the transport failed. Counted in transport_failures
• first, then re-raised — see the note in _deliver.

send_password_reset#

async send_password_reset(email: str, link: str) -> None

Send a plain-text password-reset mail carrying link.

Subject is reset_subject; otherwise identical to send_verification, transactional class and all.

send_verification#

async send_verification(email: str, link: str) -> None

Send a plain-text verification mail carrying link.

Subject is verify_subject. Transactional, so it reaches a suppressed address: someone who unsubscribed from a newsletter still has to be able to confirm their own account. The connection is opened, used and closed for this one message, on a worker thread, so the event loop is never blocked by smtplib. Anything smtplib raises propagates — delivery failure is not swallowed here.

TotpEnrolment#

A minted TOTP secret that is not enrolled yet.

begin_totp_enrolment returns one of these and writes nothing anywhere. Show uri as a QR code and secret_base32 for hand entry, keep the secret somewhere it survives one round trip, and pass it to confirm_totp_enrolment along with a code the user read off their phone. Only that call creates a credential. Enrolling before a code has verified is the most common bug in hand-rolled MFA and it locks people out of their own accounts.

Neither secret nor uri appears in the repr; the URI embeds the secret.

TotpEnrolment(secret: bytes, uri: str, label: str, digits: int = 6, period: int = 30) -> None

secret_base32 (property)#

secret_base32: str

The secret as unpadded base32, for an app that cannot scan a code.

UserRecord#

UserRecord(id: 'str', email: 'str', hashed_password: 'str', is_active: 'bool' = True, is_verified: 'bool' = False, created_at: 'float' = 0.0, updated_at: 'float' = 0.0)

UserRecord(id: str, email: str, hashed_password: str, is_active: bool = True, is_verified: bool = False, created_at: float = 0.0, updated_at: float = 0.0) -> None

UserStore#

Persistence seam for users — supply your own, or use InMemory/an ORM adapter.

A store that does not hold its own data should declare a store_id. wreath.users has to decide whether a user_router and a second_factor_router serve the same users, and two objects over one database are two objects: OrmUserStore(session, model) keeps nothing, so building one inline for each router is a mistake that reads as correct at the call site. store_id is any hashable value equal across the stores that share rows — the ORM adapters return their model class. Declaring none means "this object is the identity", which is right for InMemoryUserStore and for anything else that owns what it serves.

UserStore(*args, **kwargs)

compare_and_set_password#

async compare_and_set_password(user_id: str, expected: str, replacement: str) -> bool

Replace one password only while its previously-read hash still matches.

create#

async create(email: str, hashed_password: str) -> UserRecord

Create a user and return the stored record, with its id assigned.

hashed_password is already hashed — wreath.users hashes off the event loop before calling, and a store must never be handed or asked to handle a plaintext one. New users start is_active=True and is_verified=False; verification is a later update.

Called only after get_by_email returned None, so a store need not make the duplicate check its own — but it is registering an address as unique, and a real one should enforce that.

get_by_email#

async get_by_email(email: str) -> UserRecord | None

The user with this email address, or None if there is none.

The lookup every sign-in and every "forgot password" starts with. None is an ordinary answer rather than an error, and the flows are careful to respond identically whichever it is, so that a caller cannot learn which addresses are registered.

Matching should be case- and whitespace-insensitive, as InMemoryUserStore is: an address entered with a capital or a trailing space is the same account.

get_by_id#

async get_by_id(user_id: str) -> UserRecord | None

The user with this id, or None if there is none.

The id is UserRecord.id, which the store minted in create — it is also what a verification or reset token carries as its subject, so this is what turns a redeemed token back into a user.

get_many_by_id#

async get_many_by_id(user_ids: Iterable[str]) -> list[UserRecord | None]

Return one result per supplied id, in the same order.

Missing ids produce None, and repeated ids repeat their result. This exact shape lets a caller retain its own ordering while a database store serves the whole request with one query.

update#

async update(user: UserRecord) -> UserRecord

Persist a modified record and return what was stored.

The whole record, not a patch. The shipped flows use it for exactly two things — marking an account verified, and replacing a password hash — and both pass a copy of the record they read, so a store may treat the record's id as the key.

WebAuthnAssertion#

A verified assertion: which credential answered, and how.

user_verified is the outcome of this ceremony's user verification -- a PIN or a fingerprint rather than a bare touch. It is reported rather than enforced, because whether a bare touch is enough is a policy question; verify_webauthn_assertion(require_user_verification=True) is how a caller that has decided turns it into a refusal.

counter is the signature counter this assertion carried, already written back to the store.

WebAuthnAssertion(credential: SecondFactor, user_verified: bool, counter: int) -> None

WebAuthnCeremony#

A begun WebAuthn ceremony: what to remember, and what to send.

options is the PublicKeyCredentialCreationOptions or PublicKeyCredentialRequestOptions dictionary, with every byte string already base64url-encoded, so it is JSON as it stands. challenge is the raw bytes to keep server-side and hand back to the confirming call.

Neither member is in the repr. A challenge is not a secret -- it is sent to the browser -- but it is the thing that makes a ceremony single-use, and the plan asks for challenges to stay out of logs rather than to be argued about.

WebAuthnCeremony(ceremony: "Literal['register', 'authenticate']", challenge: bytes, options: dict[str, Any]) -> None

WebAuthnError#

A ceremony that did not verify, or a wire format that did not parse.

A ValueError, so a caller that already funnels malformed input through one except ValueError keeps working. The message names what failed, which is for the server's own operator: wreath.users.second_factor_router answers every one of these with the same opaque body, so the detail never reaches the caller who triggered it.

default_second_factor_model#

default_second_factor_model(table: str = 'user_second_factors') -> type[Any]

Build a reference ORM model for second-factor credentials.

A companion table rather than columns on the user model, which is what makes this additive for a deployment that already has users: adding second factors is a CREATE TABLE, not an ALTER on the hottest table in the schema. A user has zero or many credentials, so a table is also simply the right shape.

Declares id, user_id (indexed, since every read is "this user's factors"), kind, label, secret_material, counter, created_at and last_used_at.

secret_material is bytea and holds the TOTP shared secret or a recovery code's scrypt hash. It is named for what it is rather than for the SecondFactor.material field it maps to, and that is load-bearing: the column name is what wreath.crud.SENSITIVE_FIELD and the GraphQL schema builder match on to hide a column, and material on its own matches nothing. A model of your own that calls it something innocuous will have that something rendered into a generated API response.

counter is the newest accepted TOTP step, and it is what makes a code single-use.

The primary key is text, not uuid, unlike default_user_model. That is deliberate: SecondFactor.id is a string everywhere else, and the ORM coerces a comparison value against its column's type, so a uuid key would make store.remove(user_id, "…") raise TypeError: expected UUID, got str on the id the router just handed out. The default is wreath._secondfactor.new_credential_id, 128 bits of hex.

Import-lazy, like default_user_model, and likewise a fresh class per call -- build it once and pass it around.

default_user_model#

default_user_model(table: str = 'users') -> type[Any]

Build a reference ORM user model. Import-lazy so importing users needs no DB.

Declares exactly the columns the flows use -- id, email, hashed_password, is_active, is_verified -- plus nullable created_at and updated_at that nothing in this module writes. email is unique and indexed, which is what makes a duplicate registration a database refusal rather than a second row.

The uuid primary key auto-generates via default=uuid.uuid4 (wreath's client-side default convention, applied when the instance is constructed); the app may supply its own model instead. Calling this twice builds two distinct classes, so build the model once and pass it around.

generate_recovery_codes#

generate_recovery_codes(count: int = 10) -> list[str]

Mint count single-use recovery codes, formatted for a human to copy.

Each is 16 characters from a 30-symbol alphabet -- about 78 bits, which is not guessable -- grouped in fours with hyphens, and drawn from an alphabet with no 0, 1, i, l, o or u in it, because these get read off paper by somebody who has just lost their phone. Hyphens and case are ignored when one is redeemed.

These are the plaintext. Show them once, store only what confirm_totp_enrolment stores, which is hash_recovery_code of each.

Raises:

• ValueError — a count below 1 or above 50.

hash_password#

hash_password(password: str, *, n: int = 16384, r: int = 8, p: int = 1) -> str

Hash password with a fresh salt; returns scrypt$n$r$p$salt$hash.

second_factor_router#

second_factor_router(users: UserStore, factors: SecondFactorStore, *, issuer: str = '', prefix: str = '/auth/2fa', session_key: str = 'principal', pending_key: str = 'pending_second_factor', enrolment_key: str = 'pending_totp_enrolment', label: str = 'Authenticator app', digits: int = 6, period: int = 30, skew: int = 1, recovery_codes: int = 10, enrolment_ttl: float = 600.0, pending_ttl: float = 300.0, step_up_ttl: float = 300.0, max_verify_attempts: int = 5, verify_window: float = 300.0, enrolments: Any = None, challenges: Any = None, rp_id: str = '', rp_name: str = '', origins: Sequence[str] = (), webauthn_key: str = 'pending_webauthn', webauthn_label: str = 'Security key', webauthn_ttl: float = 300.0, user_verification: str = 'preferred', passkey_login: bool = False, clock: Callable[[], float] = <built-in function time>) -> Router

Build a mountable Router for TOTP enrolment, verification, and step-up.

Mounts, under prefix, POST /totp/begin, POST /totp/confirm, GET / (the prefix itself, listing what is enrolled), POST /verify and DELETE /{factor_id}. It is opt-in and separate from user_router: an application that does not want a second factor mounts neither this router nor its table.

Separate, but not independent. Building this router records it against the users store it was given, and a user_router over that same store with no second_factors= of its own refuses the login of anybody who has a factor here rather than issuing a full session. Pass the same SecondFactorStore to both and none of that machinery is reached; the coupling exists for the deployment that did not.

Pass rp_id and four routes appear: POST /webauthn/begin and POST /webauthn/confirm register a passkey or security key, and POST /webauthn/verify/begin and POST /webauthn/verify are the assertion half, serving a pending login and a step-up with one pair exactly as POST /verify does for codes. Without rp_id the router is TOTP-only, so an application gets passkey endpoints only when it asks for them.

Set passkey_login=True to add POST /webauthn/login/begin and POST /webauthn/login. These are a discoverable, usernameless first factor: begin sends no allowCredentials, the authenticator chooses a resident passkey, and completion resolves its owner with one indexed store lookup. Enrolments made by this router then require a resident credential and user verification; existing non-discoverable second-factor keys continue to work on the verify routes but cannot identify an account at login.

A WebAuthn ceremony's challenge is minted by begin, held wherever enrolments says, bound to the user who began it, and deleted on every exit from confirm -- success, failure, expiry, and a session that changed hands alike. Single-use is the property the whole ceremony rests on: a challenge that survives a failed attempt is a challenge a recorded assertion can be replayed against.

Enrolment is deliberately two-phase. begin mints a secret, returns the otpauth:// URI, and enrols nothing; confirm takes one code from the user's authenticator and only then writes the credential and issues the recovery codes. A secret that was displayed but never confirmed simply expires out of the session, so a scan that went wrong cannot lock anybody out. Both require an already-authenticated session, since enrolling a factor is something a signed-in user does to their own account.

A begun enrolment, and a begun WebAuthn ceremony, also end when the session does: user_router's login and logout clear enrolment_key and webauthn_key and delete the rows behind them, wherever enrolments puts them. Renaming either key here is safe -- the names travel with the registration described above -- but an application that signs users in without user_router (an OAuth2 callback writing the principal itself) clears nothing, and there the ceremony's own record of who began it is what refuses a session that has changed hands.

Adding a factor to an account that already has one is itself a step-up, and it is refused with 403 second_factor_required unless a factor was proved within step_up_ttl -- the same predicate DELETE /{factor_id} uses, read in one place so the two cannot drift. /totp/begin, /webauthn/begin and both confirm routes carry it; the confirms re-check at the write, so a ceremony begun on a fresh stamp cannot land on a stale one. The first factor is exempt, because there is nothing to step up from, and recovery codes do not count as one -- they are minted alongside TOTP and never alone.

Enrolling a factor stamps the session only when it is the first one. A stamp says the caller proved a factor the account already had; a factor the caller has just chosen proves possession of their own authenticator instead. Enrolment is refused for an already signed-in session that has not completed step-up, including when the caller intends to prove the new factor next.

POST /verify is the other end, and it serves two moments with one route. Given a pending login it finishes it, rotating the session id before promoting it, so an id an attacker planted before the password step is not the id that ends up authenticated. Given an already signed-in session it is step-up: the same code check, the same rotation, and the principal is re-stamped with the moment it happened. It accepts a TOTP code or a recovery code in the same field; a recovery code is deleted when it is redeemed.

That stamp -- second_factor_at, Unix seconds on the session principal and so on Identity.claims -- is what wreath.auth.second_factor(max_age=...) reads, and what the Cedar context exposes as second_factor_age. A route guarded by either asks when, not whether, so the application never threads a flag through a handler to find out.

DELETE /{factor_id} un-enrols one factor, and requires a factor proved within step_up_ttl seconds -- turning off a second factor is exactly the act somebody who has stolen a session wants to perform, so it is guarded by the thing they do not have. It is owner-scoped twice over: the id is looked up among the caller's own credentials, and SecondFactorStore.remove is scoped to the user as well. Removing the last real factor removes the recovery codes with it, so "off" means off rather than a login that still demands a code the user no longer has; see wreath._secondfactor.remove_second_factor for why a recovery credential cannot be deleted by id at all.

Failed verifications are counted per user, not per client address, which is the granularity that matters: a six-digit code is a million guesses and an attacker with a botnet has as many addresses as they like. max_verify_attempts (5) within verify_window (300s) applies to confirm and verify alike, answering 429 with a Retry-After. The counter lives in memory in this process, exactly as LoginLimiter documents.

Args:

• users — the same UserStore user_router was built with.
• factors — where credentials live; InMemorySecondFactorStore for
• development, OrmSecondFactorStore or your own for production.
• issuer — your application's name, as the authenticator app displays it.
• pending_key — must match the user_router this pairs with.
• enrolment_ttl — seconds a begun-but-unconfirmed enrolment stays offerable.
• pending_ttl — seconds a pending login may wait for its second factor.
• step_up_ttl — seconds a proved factor counts as recent for DELETE.
• enrolments — a wreath.session_store.SessionStore -- pass the same one
• SessionPolicy was given. With it, a begun enrolment lives
• server-side under an opaque id and the cookie carries only that id;
• without it the unconfirmed secret rides in the session itself, which
• is a cookie unless the session middleware also has a store. The
• secret is not a credential either way -- nothing refers to it until
• a code verifies -- but a cookie persists where a response body does
• not, and can reach a disk, a log, or a shared machine's profile.
• **A WebAuthn challenge is only genuinely single-use with a store** —
• without one it lives in the cookie, and a caller who kept an older
• copy of that cookie has kept the challenge with it. Building the
• router without one raises no error -- an in-memory development app
• is a legitimate configuration -- but it does emit a UserWarning
• naming what is degraded, because the alternative is a deployment
• discovering it by being wrong.
• rp_id — the relying party id the passkey routes are scoped to -- the
• site's registrable domain, example.com rather than
• login.example.com, because a credential is bound to it for as long
• as it exists. Passing it is what mounts the WebAuthn routes; the
• router is TOTP-only without it, so an application that does not want
• passkeys gets no passkey endpoints.
• rp_name — the name an authenticator shows for this site; defaults to
• rp_id.
• origins — the origins a ceremony may be collected at, matched exactly.
• Defaults to https — //{rp_id} -- **and http://{rp_id} as well when
• `rp_id is loopback** (localhost, ` — :1, 127.0.0.0/8), where a
• browser grants a secure context over plain HTTP and WebAuthn really
• works. A loopback origin may also carry any port, so
• `http — //localhost:8000` matches without being named. That exception
• is loopback and nothing else — no other host is ever admitted over
• `http — //`, and no host at all is admitted by a wildcard.
• user_verification — what the ceremonies ask of the authenticator.
• `preferred for a second factor` — verify the user when the device
• can, and let a security key with no PIN still work. The outcome is
• recorded on the session as second_factor_uv either way, so a
• policy that cares can read it rather than guess.
• passkey_login — mount discoverable first-factor passkey login. Requires
• rp_id, a DiscoverableSecondFactorStore.credential
• implementation, resident credentials at enrolment, and user
• verification on every login.
• clock — Unix-seconds source. It is what TOTP steps are counted from as
• well as what the two timeouts are measured with, so a server whose
• clock has drifted more than skew steps from its users' phones
• rejects every correct code. Injectable for deterministic tests.

Raises:

• ValueError — a skew, digit count or period the code generator refuses.

totp_uri#

totp_uri(secret: bytes, *, account: str, issuer: str = '', digits: int = 6, period: int = 30, algorithm: str = 'sha1') -> str

Build the otpauth://totp/... URI an authenticator app scans as a QR code.

The de-facto format (Google's, which every app implements): a label of issuer:account, then the secret in base32 and the parameters as a query. The issuer appears twice -- in the label and as a parameter -- because older apps read one and newer ones read the other.

The URI contains the secret in clear. It is meant for one pair of eyes, over TLS, and belongs in a response body or a QR image, never in a log line or a URL that a proxy will record.

Args:

• account — how the entry is named in the app, usually an email address.
• issuer — the application's name; omitted from the label when empty.

Raises:

• ValueError — an empty account, or an account or issuer containing a colon
• (which would move the boundary between the two halves of the label).

user_router#

user_router(store: UserStore, *, secret: str, email_sender: EmailSender | None = None, base_url: str = '', prefix: str = '/users', session_key: str = 'principal', verify_ttl: int = 86400, reset_ttl: int = 3600, link_builder: Callable[[str, str], str] | None = None, sessions: Any = None, max_login_attempts: int = 10, login_window: float = 300.0, login_limiter: AttemptLimiter | None = None, max_registration_attempts: int = 20, registration_window: float = 60.0, max_reset_requests: int = 3, reset_window: float = 900.0, reset_limiter: AttemptLimiter | None = None, second_factors: SecondFactorStore | None = None, pending_key: str = 'pending_second_factor', clock: Callable[[], float] = <built-in function time>) -> Router

Build a mountable Router with the full user lifecycle.

Mounts, under prefix, POST /register, /login, /logout, /verify, /forgot-password and /reset-password, plus GET /verify/{token} and GET /me. Login writes the session principal the auth stack already reads, so it requires SessionPolicy; without it /login answers 500 with {"error": "session_middleware_required"} and signs nobody in.

secret signs the action tokens (use a stable app secret). link_builder maps (purpose, token) -> URL for the emailed links; the default points at this router's own verify route and a reset-password?token= query.

sessions is the required server-side session store. Its delete_for method lets a password reset end the sessions somebody else is already holding; construction refuses without that revocation seam. A non-default session_key is handed to sessions.delete_for so it enumerates by the key this router writes -- give the store the same key (PostgresSessionStore(..., session_key=...)) so a caller of delete_for outside this router agrees too. A store whose delete_for takes only a subject works unchanged on the default key and raises on a renamed one. It is also where login and logout delete a half-finished second-factor ceremony's row, though a second_factor_router mounted with its own enrolments= store is cleared there too, whichever store that is.

max_login_attempts/login_window throttle failed sign-ins per identifier with the local, per-router LoginLimiter. max_reset_requests/ reset_window independently bound reset-email issuance per normalized identifier while preserving the uniform response. Those defaults are for a single-process deployment. To enforce a budget across workers, pass separately configured RateLimitPolicy instances as login_limiter and reset_limiter, backed by PostgresRateLimitStore. Their admit_key calls are the one atomic shared-store boundary before each protected attempt. Shared policies count every admitted attempt; the legacy local login limiter counts failures and clears on success. _userkit.authenticate stays unguarded for direct callers.

No response here reveals whether an account exists. Register answers 202 and forgot-password answers 200 either way, a failed login answers 401 {"error": "invalid_credentials"} whether the email is unknown or the password is wrong, and a throttled login reuses that same body with 429 plus a Retry-After of login_window seconds rather than saying the account is locked. The timing is kept comparable too — an unknown email still pays a dummy scrypt verify, and registration hashes before it checks for an existing account — but the emails these flows send are not queued, so a mail send remains an observable difference.

Pass second_factors -- any SecondFactorStore -- and login grows a third outcome. A user with an enrolled factor is authenticated but incomplete: the response is 200 {"status": "second_factor_required"} carrying no profile, and the session holds a pending marker under pending_key rather than a principal. SessionIdentityBackend reads the principal, so that session is not an identity anywhere in the framework and no protected route will admit it. wreath.users.second_factor_router's POST /verify is what turns it into one, and both routers must agree on pending_key.

Forgetting second_factors fails the login closed. Mounting second_factor_router without passing its store here would otherwise let a user enrol a factor, see it listed, and then sign in with a password alone -- protected everywhere except where it counts. When this router has no store of its own and the account being signed in has a non-recovery factor enrolled in a second_factor_router built over this same UserStore, the login answers 500 {"error": "second_factor_not_wired", "detail": ...} and writes no session. The detail names what to pass. Yes, that locks out a misconfigured deployment; refusing rather than half-wiring is why that is the right way round, and a user with nothing enrolled is unaffected either way.

Signing in and signing out both end a half-finished ceremony. A begun TOTP enrolment and a live WebAuthn challenge are cleared from the session, and the rows behind them deleted, at both ends -- a session is a sitting, and an unconfirmed secret that outlives one on a shared browser is state outliving the ceremony it belonged to.

These handlers answer with plain JSON {"status": ...} / {"error": ...} bodies. Failures raised by the framework around them keep the standard shape -- a malformed or invalid body is an RFC 9457 application/problem+json, 400 for unparseable JSON and 422 for a body that fails validation, the latter carrying an errors list of one entry per field.

verify_password#

verify_password(password: str, encoded: str) -> bool

Constant-time verify password against a stored encoded hash.

wreath.tokens#

Purpose-scoped application action tokens with rotation and optional replay refusal.

ActionTokens#

Issue and verify bounded, versioned, purpose-scoped action tokens.

ActionTokens(keys: Mapping[str, bytes], *, current: str, purposes: Iterable[TokenPurpose], ledger: TokenLedger | None = None, clock: Callable[[], float] = <built-in function time>, max_token_bytes: int = 4096) -> None

issue#

issue(purpose: str, subject: str, *, bound: str = '', now: float | None = None) -> str

Issue a token, registering its nonce first when the purpose is single-use.

verify#

verify(purpose: str, token: str, *, bound: str = '', now: float | None = None) -> TokenClaims | None

Return verified claims, or None for every invalid token shape.

MemoryTokenLedger#

Bounded in-process ledger for single-use action tokens.

Use an application-owned shared ledger when consumption must span workers. This class is explicit about its ownership instead of presenting local memory as distributed replay protection.

MemoryTokenLedger(*, max_entries: int = 10000) -> None

consume#

consume(token_id: str, *, now: float) -> bool

register#

register(token_id: str, *, ttl: int, now: float) -> bool

TokenClaims#

TokenClaims(purpose: 'str', subject: 'str', issued_at: 'int', expires_at: 'int', bound: 'str', key_id: 'str')

TokenClaims(purpose: str, subject: str, issued_at: int, expires_at: int, bound: str, key_id: str) -> None

TokenLedger#

Application-owned storage contract for single-use token nonces.

TokenLedger(*args, **kwargs)

consume#

consume(token_id: str, *, now: float) -> bool

register#

register(token_id: str, *, ttl: int, now: float) -> bool

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

TokenPurpose#

Policy for one action such as verify_email or invite_accept.

TokenPurpose(name: str, ttl: int, single_use: bool = False) -> None

token_fingerprint#

token_fingerprint(token: str) -> str

A log-safe stable identifier; never returns token material.

wreath.session_store#

Server-side session storage.

The default SessionPolicy keeps the whole session in a signed cookie: nothing on the server, but nothing revocable either, and a 4 KiB ceiling. Hand the policy a store and the cookie carries only a signed session id, while the contents live in PostgreSQL:

store = PostgresSessionStore(app.postgres("main"))
app.configure_http_policy(HttpPolicy(session=SessionPolicy(secret="…", store=store)))

Then a session can be revoked (delete the row), inspected, and grown past what a cookie holds. No Redis: the database the application already has is the store, the same choice PostgresRateLimitStore makes.

The table is not created for you. Apply PostgresSessionStore.schema_sql as a migration, so schema changes stay in the migration history where the rest of the schema lives.

Expired rows are removed by PostgresSessionStore.purge_pass -- a chunked, resumable, paced walk you hand to the job runner rather than a background thread, so it retries, does not duplicate across workers, and cannot hold one long transaction open over a table that grows with every login.

PostgresSessionStore#

Sessions in a PostgreSQL table, shared by every worker.

jsonb rather than text: the contents stay queryable, so "log out every session for this user" is one statement rather than a scan.

PostgreSQL owns the clock (clock_timestamp()), so workers on disagreeing wall clocks cannot disagree about expiry -- the same reasoning, and now the same wreath.store primitive, as the PostgreSQL rate-limit store.

Unlike the other two stores over that primitive, the lifetime is not the store's to decide: the cookie's max_age is what a session should outlive, so it is bound per write rather than fixed at construction.

Args:

• database — the Database this store's statements run on.
• table — the backing table, an unqualified plain identifier.
• session_key — which session key holds the principal, matching
• user_router(session_key=...) and
• SessionIdentityBackend(session_key=...). Only delete_for reads
• inside the payload, so this affects nothing else. It must match the
• key used by the router and identity backend for session revocation
• to find the principal.

PostgresSessionStore(database: Any, *, table: str = 'wreath_session', session_key: str = 'principal') -> None

component#

component() -> Any

This store's claim on the wreath schema.

The session table is unqualified -- wreath_session, resolved through search_path -- and it stays that way. Moving it into the wreath schema is not additive: a worker still on the previous version looks for the unqualified name and would not find it, which is precisely what the additive rule exists to prevent. So the component is registered where the rows actually are, and a move is a later, staged concern.

delete#

async delete(sid: str) -> None

Revoke one session. Not an error when it is already gone.

Takes effect for every worker at once, which is the thing a cookie-only session cannot do.

delete_for#

async delete_for(subject: str, session_key: str | None = None, *, namespace: str = '') -> int

Drop every session whose principal is subject.

One statement over the payload, because the alternative -- read every row, decode each one, delete the matches -- is the whole table across the network to end a handful of sessions.

The predicate has to read the key the session backend actually writes, and that key is configurable. It was hardcoded to principal here, so a deployment on user_router(session_key=...) / SessionIdentityBackend(session_key=...) got a password reset that matched no row, ended no sessions, and reported success -- a security control silently doing nothing. session_key defaults to whatever this store was constructed with, and wreath.users passes its own so the two cannot drift apart.

Args:

• subject — the principal's sub, as login wrote it.
• session_key — the session key holding the principal. Defaults to the
• store's own session_key=.
• namespace — the principal's issuer namespace. The empty default
• matches local sessions that have no iss claim.

Returns:

• How many rows were deleted, or 0 when the driver reported no count.

load#

async load(sid: str) -> dict[str, Any] | None

The session under sid, or None when it is absent or expired.

Expiry is decided by the database's clock in the same statement as the lookup, so an expired row is never returned even though nothing has purged it yet. Anything that is not a JSON object -- a row whose payload was written by something else -- also reads as None rather than reaching a handler as a value it cannot use.

purge#

async purge() -> str

Delete expired rows in one unbounded statement.

Safe to run concurrently, and fine for a small table -- but on a large one it is the long transaction purge_pass exists to prevent.

purge_pass#

purge_pass(*, chunk: int = 1000, **options: Any) -> Any

A recurring pass that deletes expired sessions, chunk by chunk.

The supported way to keep the table small:

jobs.drive(store.purge_pass(), cron="*/10 * * * *")

A session table is one row per login rather than one per request, so it is exactly the size where a single unbounded DELETE starts holding a snapshot open long enough for the application to notice afterwards. The pass walks it in expiry order, one transaction per chunk, resumably, and paced. See wreath.passes.

save#

async save(sid: str, data: dict[str, Any], max_age: int) -> None

Store data under sid, expiring max_age seconds from now.

One INSERT ... ON CONFLICT DO UPDATE, so creating a session and rewriting one are the same statement and neither can lose a race to the other. Every write moves the deadline, which is what a session wants and is the opposite of the rule the other stores over this primitive follow: a session should survive for max_age past the last time it was used, while an idempotency key must not be extendable by the request holding it.

Args:

• sid — the session id, as the cookie carries it.
• data — the whole session, as jsonb. Replaces; this is not a merge.
• max_age — seconds, applied against the database's clock, not this worker's.

save_if_present#

async save_if_present(sid: str, data: dict[str, Any], max_age: int) -> bool

Replace only a still-live row, atomically preserving revocation.

schema_database (property)#

schema_database: Any

The database component()'s tables belong to.

The application never saw this store constructed -- a caller builds it and hands it to SessionPolicy -- so it cannot know which app.postgres() the tables go to unless the store says. This is that contract, and it is one name rather than a list of plausible ones: Wreath._schema_database reads exactly schema_database.

schema_sql#

schema_sql() -> str

DDL for the backing table, semicolon-joined. A derivation of component().

SessionStore#

Where server-side session contents live.

SessionStore(*args, **kwargs)

delete#

async delete(sid: str) -> None

Drop sid. Must not fail when it is already gone.

load#

async load(sid: str) -> dict[str, Any] | None

The session for sid, or None when absent or expired.

save#

async save(sid: str, data: dict[str, Any], max_age: int) -> None

Store data under sid, expiring max_age seconds from now.

save_if_present#

async save_if_present(sid: str, data: dict[str, Any], max_age: int) -> bool

Replace a live session, returning false if it was revoked meanwhile.

wreath.organizations#

Organisations, memberships, roles within them, and invitations.

Tenancy at the identity layer. wreath.orm.TenantContext is the other one — it binds a schema and a role transaction-locally, which is tenancy at the data layer — and the two are deliberately separate: a user may belong to two organisations with different roles while every one of their requests reads the same schema, and a single-tenant deployment may still want organisations.

Everything enterprise hangs off this. SCIM provisions into it, SAML asserts into it, an audit record names the organisation a write happened in, and an admin screen redacts a field per role within it. Building any of those first would have produced a second membership model, so this lands first and they adapt onto it.

This is not an ORM model you must adopt. wreath.users already made the right call with a pluggable UserStore, and the same reasoning applies harder here: organisations are the table an existing application is most likely to already have, under a different name, with columns nobody wants to migrate. OrganizationStore is the seam. InMemoryOrganizationStore is the bounded development/reference implementation; PostgresOrganizationStore is the durable implementation shared by API workers and process restarts.

store = PostgresOrganizationStore( app.postgres("main"), roles={"admin", "member", "billing"} ) await store.add_member("acme", "alice", roles={"admin"})

app.configure_auth(backend, CedarAuthorizer( engine=CedarPolicies(POLICY), organizations=Memberships(store), ))

A policy then reads them as sets, the same shape context.flags and context.regions already use:

permit (principal, action == Action::"read", resource) when { context.organizations.contains(resource.owner) };

permit (principal, action == Action::"invite", resource) when { context.org_roles.contains("acme:admin") };

Roles are namespaced by organisation ("acme:admin") because a bare role name cannot say where it applies, and an admin of one tenant is not an admin of another. That is the whole cross-tenant leak in one sentence, so the spelling that makes it impossible is the one that ships.

Resolution is synchronous, on purpose#

Memberships — the authorizer's seam — is not async, and that is a design decision rather than a limitation. Authorization runs on the request path, which the request-boundary baseline expects to stay cheap; a database round trip per decision is the single easiest way to make Cedar the slowest thing in a request. The membership set a decision needs is already known when the session was established, so the shape that works is: load once where identity is established, resolve synchronously where policy is evaluated.

Memberships therefore reads from a synchronous source — the in-memory store directly, or a snapshot an authentication backend loaded. load_into is the helper for the second, and wreath.organizations never opens a connection on the authorization path.

SCIM provisions into this, and adds nothing to it#

scim_router is the RFC 7643/7644 adapter, and it is an adapter in the strict sense: it has no user table, no group table and no membership table of its own.

app.include_router(scim_router(app, users=users, organizations=store, organization="acme"))

A SCIM User is a wreath.users record, a SCIM Group is a role in this store's declared vocabulary, and a SCIM group membership is this module's Membership. That is what makes a directory de-provisioning take effect: the row a directory removes is the row context.org_roles reads. It is also why a group cannot be created over SCIM — the role vocabulary is configuration a Cedar policy names by string, not data — and why every SCIM route is authorized by the application's own Cedar policies rather than by anything SCIM decides.

Listing an organisation's members is the one question the authorization path never asks, so it is a separate seam: MemberDirectory, which both shipped stores implement and scim_router requires.

Invitation#

An offer of membership that survives the invitee having no account yet.

That is the whole reason this is a record rather than a call: the common case is inviting an email address belonging to nobody, and an invitation keyed on a user id cannot express it. The invitee is an email address until acceptance, and acceptance is what binds it to a user.

Args:

• token — the secret in the invitation link. The in-memory store compares
• it in constant time; the PostgreSQL store indexes its fixed-width
• SHA-256 digest. Single-use in both.
• organization — where the invitation is to
• email — who was invited, before they exist
• roles — what they get on acceptance
• expires_at — epoch seconds; an expired invitation is refused
• accepted_by — the user id that accepted, or None

Invitation(token: str, organization: str, email: str, roles: frozenset[str] = frozenset(), expires_at: float | None = None, accepted_by: str | None = None) -> None

expired#

expired(now: float) -> bool

InMemoryOrganizationStore#

The reference OrganizationStore, and what the tests run against.

Args:

• roles — the declared role vocabulary. Required and non-empty: an
• application with no declared roles cannot have a policy naming one
• validated, and silently skipping that check is the failure this
• whole design is trying to avoid.
• organizations — optional initial organisations

InMemoryOrganizationStore(*, roles: Iterable[str], organizations: Iterable[Organization] = ()) -> None

accept#

async accept(token: str, user_id: str, *, now: float | None = None) -> Membership

Bind an invitation to a user, once.

Raises:

• ValueError — the token is unknown, already used, or expired. One
• message per branch, distinct, so a test asserting the refusal
• cannot pass on whichever branch happened to fire -- every
• refusal here would otherwise mention the token.

add_member#

async add_member(org_id: str, user_id: str, *, roles: Iterable[str] = ()) -> Membership

Place user_id in org_id with roles, replacing any prior roles.

create#

async create(organization: Organization) -> Organization

Add an organisation; refuses to replace an existing id.

invitations#

async invitations(org_id: str) -> tuple[Invitation, ...]

Every invitation issued for org_id, accepted or not.

invite#

async invite(org_id: str, email: str, *, roles: Iterable[str] = (), ttl: float | None = None, now: float | None = None) -> Invitation

Offer membership to an email address that may belong to nobody yet.

members#

async members(org_id: str) -> tuple[Membership, ...]

Every membership in org_id, in user id order -- the MemberDirectory half.

Scans the membership table rather than keeping a reverse index. The table is user -> org -> roles because that is the shape the authorization path reads, and a second index maintained beside it is one more thing that can disagree with the first; a store this size can afford the scan, and a database-backed one answers with a WHERE clause instead of either.

memberships#

async memberships(user_id: str) -> tuple[Membership, ...]

memberships_for#

memberships_for(user_id: str) -> tuple[Membership, ...]

The synchronous half, which Memberships reads on the request path.

Present here because this store is in memory, so the query is a dict lookup and there is nothing to await. A database-backed store offers no such method and participates through load_into instead -- which is why Memberships probes for this rather than requiring it.

organization#

async organization(org_id: str) -> Organization | None

remove_member#

async remove_member(org_id: str, user_id: str) -> bool

Remove a membership. Returns whether there was one.

roles#

roles() -> frozenset[str]

The declared role vocabulary.

Membership#

A user's place in one organisation.

Args:

• organization — the organisation id
• user_id — the member
• roles — role names within that organisation, unqualified. The qualified
• `"<org> — <role>"` spelling is produced on the way into policy context
• and is never stored, so there is one place the namespacing rule
• lives.

Membership(organization: str, user_id: str, roles: frozenset[str] = frozenset()) -> None

qualified_roles#

qualified_roles() -> frozenset[str]

This membership's roles, namespaced by organisation.

MemberDirectory#

The other direction: who is in an organisation.

Deliberately separate from OrganizationStore, because the two questions have opposite cost profiles and opposite callers. Authorization asks "what does this user hold", once per session, and never needs a list. A directory -- SCIM's GET /Users, an admin screen's member table -- asks "who is in this tenant", and cannot be answered by the first question at all.

A store that only ever backs authorization does not have to implement this, and scim_router refuses at build time rather than discovering it missing on the first request.

MemberDirectory(*args, **kwargs)

members#

async members(org_id: str) -> tuple[Membership, ...]

Every membership in org_id.

Memberships#

The authorizer's seam: this caller's organisations and roles, synchronously.

Two sources, checked in order:

1. a snapshot placed on the request by load_into, which is how a database-backed store participates without a query on the authorization path;
2. a synchronous memberships_for(user_id) on the source, which InMemoryOrganizationStore gains through this class.

A caller with neither resolves to nothing, and nothing denies. There is deliberately no third source and no fallback to an async query: a fallback that sometimes queried would make the authorization path's cost depend on whether a snapshot happened to be loaded, which is the kind of variance that shows up as a p99 nobody can explain.

Args:

• store — the organisation store. Only roles() is used on the request
• path, for the startup vocabulary; membership data comes from the
• snapshot.

Memberships(store: Any) -> None

for_request#

for_request(request: Any) -> tuple[Membership, ...]

This caller's memberships, from the snapshot or a synchronous source.

names#

names() -> frozenset[str]

The declared role vocabulary, qualified and unqualified.

The enumeration half of the provider surface, the same shape FeatureFlags.names() offers, and read by CedarAuthorizer at startup to refuse a policy naming a role nobody declared.

Unqualified names are included because a policy may test context.org_roles.contains("admin") against the active organisation's unqualified roles, which this provider also supplies. Organisation ids are data, not vocabulary, so the qualified form cannot be enumerated here -- and context.organizations is validated for that reason.

schema_owners (property)#

schema_owners: tuple[Any, ...]

Durable stores whose schema the configured authorizer must own.

Organization#

One tenant.

Args:

• id — stable identifier, used verbatim in context.organizations and in
• the "<org> — <role>" spelling, so it must not contain a colon --
• checked, because a colon here silently forges a role in another
• organisation.
• name — display name
• metadata — anything the application wants to carry along

Organization(id: str, name: str = '', metadata: Mapping[str, object] = <factory>) -> None

OrganizationFederation#

OrganizationFederation(organizations: OrganizationStore, installations: Any) -> None

resolve#

async resolve(external: Any, binding: Any) -> Any

OrganizationFederationError#

OrganizationStore#

Where organisations, memberships and invitations live.

Async, because a real one is a database. The authorization path does not use this protocol -- see the module docstring -- so nothing here is on the request's hot path.

OrganizationStore(*args, **kwargs)

memberships#

async memberships(user_id: str) -> tuple[Membership, ...]

Every membership user_id holds, in any organisation.

organization#

async organization(org_id: str) -> Organization | None

The organisation, or None.

roles#

roles() -> frozenset[str]

The declared role vocabulary, unqualified.

Synchronous and configuration rather than data: roles are a fixed vocabulary an application declares, which is what lets a policy naming "acme:admni" be refused at startup instead of denying forever.

PostgresOrganizationStore#

Durable organisations, memberships, and invitations in PostgreSQL.

Every instance is stateless apart from lazily prepared statements. Building a new instance after a process restart therefore reads the same rows, and two workers share invitation consumption rather than each accepting its own in-memory copy.

Invitation acceptance is one PostgreSQL statement: it conditionally marks the invitation accepted and upserts the membership from the row it marked. The returned membership is proof that this caller consumed the token. Concurrent acceptance can return at most one membership.

Tokens are looked up by a fixed-width SHA-256 digest. The raw token remains in the row so an operator can list or resend invitations, but it is never used in an index comparison and a presented token is never interpolated into SQL.

Args:

• database — the application database shared by every API worker.
• roles — the non-empty, declared role vocabulary.
• table — base table name. _membership and _invitation are
• appended for the related tables; all three are plain identifiers.
• prefix — prepared-statement prefix. Defaults to table.

PostgresOrganizationStore(database: Any, *, roles: Iterable[str], table: str = 'wreath_organization', prefix: str | None = None) -> None

accept#

async accept(token: str, user_id: str, *, now: float | None = None) -> Membership

add_member#

async add_member(org_id: str, user_id: str, *, roles: Iterable[str] = ()) -> Membership

component#

component() -> Any

The durable store's additive, application-owned schema claim.

create#

async create(organization: Organization) -> Organization

invitations#

async invitations(org_id: str) -> tuple[Invitation, ...]

invite#

async invite(org_id: str, email: str, *, roles: Iterable[str] = (), ttl: float | None = None, now: float | None = None) -> Invitation

members#

async members(org_id: str) -> tuple[Membership, ...]

memberships#

async memberships(user_id: str) -> tuple[Membership, ...]

organization#

async organization(org_id: str) -> Organization | None

remove_member#

async remove_member(org_id: str, user_id: str) -> bool

roles#

roles() -> frozenset[str]

The declared role vocabulary.

schema_database (property)#

schema_database: Any

The database that owns component()'s three tables.

schema_sql#

schema_sql() -> str

DDL for all three durable tables, derived from component().

active_organization#

active_organization(request: Any) -> str | None

Which organisation this request is acting in, or None.

Read from the session under ACTIVE_ORGANIZATION_KEY, then from the identity's limits. A user in two organisations is acting in one of them at a time, and every answer that depends on which -- the unqualified roles in context.org_roles, an admin screen's scope, an audit record's tenant -- reads it here rather than deriving it separately.

load_into#

load_into(request: Any, memberships: Iterable[Membership]) -> None

Place a membership snapshot on the request, for the authorizer to read.

Called from wherever identity is established -- a session backend that has just read the user, a token backend that has just verified one. That is the query that was going to happen anyway, so authorization adds none.

scim_router#

scim_router(app: Any, *, users: Any, organizations: Any, organization: str | Callable[[Any], str], prefix: str = '/scim/v2', read_action: str = 'scim_read', write_action: str = 'scim_write', page_size: int = 100, max_page_size: int = 200, max_filter_scan: int = 1000, cursor_secret: bytes | str | None = None, cursor_timeout: int = 3600, revoke_sessions: Callable[[str], Any] | None = None) -> Router

SCIM 2.0 provisioning endpoints for one organisation, over stores you already have.

app.configure_auth(backend, CedarAuthorizer(engine=CedarPolicies(POLICY),
                                            organizations=Memberships(orgs)))
app.include_router(scim_router(app, users=users, organizations=orgs,
                               organization="acme"))

SCIM keeps no store of its own. A User is a record in users, a Group is a role in organizations.roles(), and membership is a Membership in the organisation. Provisioning writes what any other part of the application would write, which is what makes a de-provisioned user actually lose access: context.organizations and context.org_roles read the same rows.

There is no authorization decision in this module. Every route carries @authorize(action=..., resource='Organization::"<org>"') and the application's own CedarAuthorizer answers it, so a SCIM client is bounded by the same policy set as everything else and appears in permissions_router's vocabulary. A policy has to permit it:

permit (principal in Group::"directory", action == Action::"scim_write", resource == Organization::"acme");

Building the router refuses when the application has no authorizer configured, rather than serving a provisioning API to any authenticated caller -- the refuse-rather-than-half-wire rule in AGENTS.md is why that is the right way round.

The two de-provisioning verbs, which are not the same#

• DELETE /Users/{id} removes the membership. The account survives, and so does everything it owns -- de-provisioning a directory user must not delete their content, and a cascade here would make an accidental one unrecoverable. The resource then answers 404, which is what a directory expects of a user it removed.
• PATCH or PUT setting active: false disables the account, which is UserRecord.is_active and is not scoped to an organisation. When the user is a member of another organisation as well, that write is refused with scimType: mutability naming the count: one tenant's directory does not get to sign a user out of another tenant. DELETE is the scoped verb and the refusal says so.

POST /Users with a userName that already has an account adopts it into the organisation and answers 201; the same call for someone already provisioned here answers 409 uniqueness. A directory that retries a call it already made therefore converges rather than duplicating.

What it does not implement, and why the refusals are loud#

Bulk operations and ETag are absent, and ServiceProviderConfig says so. Lists support bounded sortBy/sortOrder. POST and DELETE on /Groups answer 501: a group here is a role in the declared vocabulary, which is configuration a policy names by string, not a row a client may mint. A filter naming an attribute this provider does not hold -- externalId, name.givenName, anything else section 4.1 makes optional -- is a 400 rather than an empty page, because an empty page is what makes a directory decide the user is missing and create them twice.

Args:

• app — the application these routes will be mounted on. Read for its
• authorizer and its authentication backend; the routes are its own.
• users — a wreath.users UserStore -- get_by_id, get_many_by_id,
• get_by_email, create, update.
• organizations — an OrganizationStore that also offers
• members(org_id), which the authorization path never needs and a
• directory listing always does.
• organization — which organisation this router provisions into, or a
• (request) -> str reading it from a path parameter or a subdomain.
• The same value becomes the Cedar resource, so the data scope and the
• policy decision cannot disagree about which tenant is meant.
• prefix — where the routes mount. {name} placeholders are filled from
• the request's path parameters when building meta.location.
• read_action — the Cedar action name the read routes ask about.
• write_action — the Cedar action name the write routes ask about.
• page_size — count when a client sends none.
• max_page_size — the largest count honoured.
• max_filter_scan — how many members, roles, or role assignments a request
• that materializes them may examine before refusing with tooMany.
• User filters read candidate accounts in one batch; group resources
• expand role membership, so this bounds both forms of work.
• cursor_secret — key authenticating opaque cursors. A single-process
• deployment may omit it; every worker in a fleet must share one.
• cursor_timeout — seconds a cursor remains valid between page requests.

Returns:

• A Router to pass to app.include_router.

Raises:

• ValueError — a store missing a method, an application with no authorizer,
• an empty action name, an inconsistent page size, or an organisation
• id that cannot be spelled inside a Cedar entity reference.

wreath.tenancy#

Tenant resolution and PostgreSQL-enforced isolation.

A TenantDirectory is the application-owned authority that maps a trusted tenant id to its schema, role and lifecycle state. TenantHostLabel and TenantHeader extract a candidate id; TenancyMiddleware resolves it through the directory, refuses unknown, suspended or unready tenants, and binds the result for the request. Request host, path, header and token values never become schema or role names directly.

SchemaMode.isolated(isolation="role") combines transaction-local search_path selection with SET LOCAL ROLE. The tenant role owns privileges on its schema and read access to the central schema. The application login role is NOINHERIT, so it has no ambient tenant-table privilege outside a bound transaction.

The boundary has explicit limits:

• a deliberate SET ROLE to another role of which the login is a member can cross tenants; source hardening and isolation_report expose that residual;
• a superuser or schema owner bypasses grants and is refused by verify_isolation;
• PostgreSQL catalog rows reveal object names even when grants prevent access to their data; confidential tenant names require a database-per-tenant design.

connection_budget prices database-per-tenant alternatives from server memory and max_connections. Central models remain read-only to tenant roles and stay behind the tenant schema in the search path, so tenant-local and shared data can be joined in one statement.

ConnectionBudget#

Whether a fleet fits, and what it would cost if it did.

Reported rather than silently accepted, because the failure it prevents is the worst kind: the deployment works in staging with three tenants and refuses connections in production at two hundred.

ConnectionBudget(tenants: int, workers: int, per_tenant_connections: int, max_connections: int) -> None

explain#

explain() -> str

fits (property)#

fits: bool

required (property)#

required: int

server_memory_bytes (property)#

server_memory_bytes: int

FromTenant#

Bind this session to the tenant the request resolved to.

Written inside FromORM, never on its own:

session: Annotated[Session, FromORM("main", tenant=FromTenant())]

That spelling is the whole point of this class. Before it, the only route to a tenant-bound session was building Session(registry, workload, tenant=TenantContext(...)) by hand in the handler body -- so the framework's own convenience produced an unbound session and the safe path was the one you had to remember. Now the declarative path is the safe one and the unsafe one is refused at route-compile time.

FromTenant() -> None

resolve#

resolve(request: Any) -> TenantContext

The bound tenant's context, or refuse.

Reads the ambient scope rather than the request, because the scope is what a job, a pass and a workflow also run under -- one binding to resolve, wherever the work is happening.

InMemoryTenantDirectory#

A directory held in the process, for tests and single-tenant fixtures.

Not for a fleet: it cannot see a tenant another worker provisioned. The lookup is a dict, so resolution is one hash of a short string on the request path rather than a scan.

InMemoryTenantDirectory(tenants: Iterable[Tenant] = ()) -> None

add#

add(tenant: Tenant) -> None

all#

all() -> tuple[Tenant, ...]

resolve#

resolve(key: str) -> Tenant

IsolationReport#

What a tenant role can actually reach, read out of the catalog.

From information_schema rather than from the code that intended it: isolation you cannot audit is isolation you are hoping for.

IsolationReport(tenant: str, role: str, readable_schemas: tuple[str, ...], writable_schemas: tuple[str, ...], ungranted_tables: tuple[str, ...], inherits_by_membership: tuple[str, ...]) -> None

crosses_into#

crosses_into(other: str) -> bool

Tenancy#

Resolution: a source that names a tenant, and a directory that finds it.

The two halves are separate because collapsing them is the whole vulnerability. A header used directly as a schema is a header that can name any schema in the database; a header used as a key into a directory can only ever name a tenant somebody provisioned.

Tenancy(*, directory: TenantDirectory, source: TenantSource | None = None) -> None

directory (property)#

directory: TenantDirectory

resolve_name#

resolve_name(name: str) -> Tenant

The directory's answer for one name, with its status checked.

resolve_request#

resolve_request(request: Any) -> Tenant

The tenant this request belongs to.

Three steps and no fourth: read a name, look it up, check it may serve. Everything expensive about tenancy happens at provisioning time.

source (property)#

source: TenantSource

TenancyError#

A refusal made before any tenant SQL is sent.

TenancyMiddleware#

Resolve the request's tenant and bind it for the rest of the request.

Global middleware, not route middleware, and for the same reason SessionPolicy is: the binding has to exist before a route's own tape runs, because an authorization hook that reads tenant-scoped data would otherwise run unbound. wreath.app already refuses the wrong spelling for sessions and the same argument applies here.

The ContextVar is set in before and reset in after, which is the lifetime the tape gives a hook. A request that never reaches after -- because something above it raised -- leaves the token unreset on a task that is ending anyway; a ContextVar is per-task, so nothing else can observe it.

TenancyMiddleware(tenancy: Tenancy, *, optional: bool = False) -> None

after#

async after(request: Any, response: Any) -> Any

before#

async before(request: Any) -> None

Tenant#

One tenant: who it is, where its data lives, and whether it may serve.

key is the name the outside world uses -- a subdomain label, a header value, a claim. schema and role are physical placement, and they are deliberately separate fields rather than derived from the key: a key is customer-visible and may have to change, while a schema holding a terabyte may not, and deriving one from the other welds them together.

Tenant(key: str, schema: str, role: str | None = None, status: TenantStatus = <TenantStatus.ACTIVE: active>, metadata: Mapping[str, Any] = <factory>) -> None

context#

context() -> TenantContext

The transaction-local binding for this tenant.

require_bindable#

require_bindable() -> None

Raise unless this tenant may serve a request right now.

Checked at the bind rather than at the route. On a route it is one forgotten decorator away from not happening; at the bind there is no way to reach the data without passing through it.

TenantDirectory#

The application's own record of who its tenants are.

A protocol rather than a base class: the directory is usually a table the application already has, and migrations.TenantState already says the runner never invents tenant identity.

TenantDirectory(*args, **kwargs)

all#

all() -> tuple[Tenant, ...]

resolve#

resolve(key: str) -> Tenant

TenantHeader#

The tenant name arrives in a header.

For a gateway that has already authenticated the caller and rewritten the header. On an internet-facing listener a client-supplied header is a client choosing its own tenant name -- which is safe only because the name is looked up rather than used, and because the identity check still has to pass.

TenantHeader(header: str, trusted: bool = False) -> None

describe#

describe() -> str

name_for#

name_for(request: Any) -> str | None

TenantHostLabel#

The tenant name is the first label of the host, under a fixed suffix.

acme.example.com under suffix="example.com" is acme. The suffix is required and matched exactly: without it the apex example.com resolves example as a tenant, and www becomes a customer.

TenantHostLabel(suffix: str) -> None

describe#

describe() -> str

name_for#

name_for(request: Any) -> str | None

TenantNotBound#

A tenant-scoped registry was reached with no tenant resolved.

TenantNotReady#

The tenant exists and its schema is not yet at the target migration.

TenantSessionClaim#

The tenant name is a claim on the established session.

The strongest of the three, and the one that needs an authenticated request: the name comes from state the server wrote, so a caller cannot name a tenant at all.

TenantSessionClaim(claim: str = 'tenant') -> None

describe#

describe() -> str

name_for#

name_for(request: Any) -> str | None

TenantSource#

How a request names a tenant. Never how it chooses one.

TenantSource(*args, **kwargs)

describe#

describe() -> str

name_for#

name_for(request: Any) -> str | None

TenantStatus#

Where a tenant is in its life.

Four values rather than a boolean, because "exists" and "may serve a request" are different questions and collapsing them is how a half-migrated tenant answers a request with a missing-relation error deep inside a handler.

TenantSuspended#

The tenant exists and may not be bound.

UnknownTenant#

The directory holds no tenant under that name.

A miss is never a fallback. Serving a request against "no tenant" means serving it against whatever namespace the pooled connection last held, which belongs to somebody else.

cedar_context#

cedar_context() -> dict[str, str]

The active tenant as Cedar sees it.

A separate key from context.organizations, deliberately. An organisation is tenancy at the identity layer -- who you are acting as -- and a tenant is where the rows live. A deployment can have one without the other, and a policy that conflated them would be right until the first customer with two organisations in one database.

connection_budget#

connection_budget(*, tenants: int, workers: int = 1, per_tenant_connections: int = 1, max_connections: int = 100) -> ConnectionBudget

Price connection-per-tenant isolation before choosing it.

max_connections defaults to PostgreSQL's own default rather than to something generous: a deployment that has not raised it is the one this is most likely to surprise.

check_enqueue_tenant#

check_enqueue_tenant(explicit: str | None) -> str

Reconcile an explicit tenant= with the bound scope.

JobRunner.enqueue(tenant="") defaults to empty, so work enqueued inside a tenant request runs later against no tenant at all -- on a worker, hours away, with no request left to attribute it to. Inside a scope the scope wins, and an explicit mismatch is refused rather than silently resolved in either direction: one of the two spellings is a bug and there is no way to tell which.

current_tenant#

current_tenant() -> Tenant

The tenant bound to this task, or raise TenantNotBound.

current_tenant_or_none#

current_tenant_or_none() -> Tenant | None

The tenant bound to this task, or None. For code that may run outside one.

deprovision_tenant#

async deprovision_tenant(connection: Any, tenant: Tenant, *, force: bool = False) -> None

Drop a tenant's schema and role. Irreversible, so it asks first.

Refuses while the schema holds any table unless force, following privacy.erase: an irreversible act recomputes what it is about to destroy and refuses when that does not match what the caller expected.

find_schema_literals#

find_schema_literals(source: str) -> tuple[str, ...]

Every tenant-schema literal in one piece of application source.

One pass of a compiled pattern over the text rather than a parse: the input is a file, the answer is a list of names, and a SQL parser here would be a second dialect to keep correct for no extra finding.

isolation_report#

async isolation_report(connection: Any, tenant: Tenant) -> IsolationReport

What this tenant's role can reach, according to the catalog.

Takes no central=: the point is to report every schema the role can reach, so naming the one it is expected to reach would let the report agree with the expectation instead of with the database.

provision_tenant#

async provision_tenant(connection: Any, *, key: str, schema: str | None = None, role: str | None = None, central: str = 'central', login_role: str, status: TenantStatus = <TenantStatus.PROVISIONING: provisioning>) -> Tenant

Create one tenant's schema, role and grants, idempotently.

Idempotent throughout, so a run stopped by a lock or a deploy is finished by running it again rather than by reasoning about which of eleven statements completed -- the property migrations.apply_fleet already holds for the fleet, for the same reason.

Returns the tenant in PROVISIONING: the schema exists and the artifact has not been applied, and those are different facts. Move it to ACTIVE once migrations.apply has run against it.

require_connection_budget#

require_connection_budget(budget: ConnectionBudget) -> None

Refuse a fleet that cannot fit, at startup, with the arithmetic shown.

Naming the numbers rather than the conclusion: an operator who disagrees can raise max_connections, cut workers, or go back to role isolation, and none of those choices is wreath's to make.

tenant_scope#

tenant_scope(tenant: Tenant | str, *, directory: TenantDirectory | None = None) -> Iterator[Tenant]

Bind one tenant for the duration of the block.

Takes a Tenant or, with a directory, a key. Restores the previous binding on exit rather than clearing it, so a nested scope in a background task cannot unbind its caller's.

telemetry_attributes#

telemetry_attributes() -> dict[str, str]

The active tenant as a metric and span dimension.

"Which tenant is slow" is the first question of every incident, and wreath.metrics already carries instance for the same reason.

verify_isolation#

async verify_isolation(connection: Any, *, schemas: Iterable[str] = ()) -> None

Refuse a connection whose role makes the boundary decorative.

Both checks are about privileges that cannot be revoked from their holder, so no grant set can compensate for either. At startup, naming the role -- the one moment somebody can act on it. A deployment that discovers this in production discovers it as a breach.

wreath.saml#

Verify a SAML 2.0 assertion, and turn what it proves into facts.

An enterprise identity provider signs an XML assertion saying "this person authenticated, here is who they are, and here is what I know about them". This module checks that signature against keys you configured, checks the assertion is addressed to you and is live right now, spends its identifier so it cannot be presented twice, and hands back the values it carries.

It decides nothing. A verified assertion is a VerifiedAssertion, and VerifiedAssertion.facts() is a mapping for Cedar context. Whether the person it names may do the thing they asked for is a policy question, and the policy set answers it — exactly as wreath.signatures establishes for a signed request. "Verified" means the identity provider said this; it does not mean "trusted".

What this is, and what it is not#

This is the service provider half, and only the receiving end of it. It reads a Response or a bare Assertion that arrived by some route you own — the POST body of your assertion consumer endpoint, most often. It does not act as an identity provider, publish service-provider metadata, decrypt an EncryptedAssertion, or mount the redirect/POST binding endpoints.

The three properties it is built on#

Transforms are an allow-list of exactly two. A Reference may declare enveloped-signature and exclusive canonicalization, in that order, and nothing else. An unknown transform is a refusal naming the algorithm, never a skip: a permissive transform list is a signature-wrapping vector in its own right, because a transform the verifier ignores is a transform that changed what the digest covers.

Verification runs over source bytes. wreath.xml records the byte range every element was parsed from and canonicalizes by re-reading those bytes, so the subtree the digest covers and the subtree the values are read from are the same handle. The reference is resolved once, by Document.find_id, which refuses a repeated identifier rather than picking one; and the element it resolves to must be the very object this module then reads. There is no second lookup for an attacker to divert.

An assertion identifier is spent once. Replay protection is one atomic insert-or-reclaim rather than a read followed by a write. MemoryReplayLedger owns that claim in one worker; PostgresStore owns it across several.

Keys are configured, never fetched#

IdentityProvider takes the certificates your deployment obtained from the identity provider out of band, and parses them once, at construction. Nothing here reads KeyInfo to decide which key to use and nothing here opens a socket. A signature under a key you did not configure is simply unverified. That rule is wreath.signatures': resolving a key by fetching lets an unauthenticated caller aim an outbound request from your network, and the assertion has not been verified yet at the moment the fetch would happen.

from wreath.saml import IdentityProvider, MemoryReplayLedger, ServiceProvider, verify_response

IDP = IdentityProvider(entity_id="https://idp.example/metadata", certificates=[CERT_PEM])
SP = ServiceProvider(entity_id="https://app.example/saml", acs_url="https://app.example/acs")
SEEN = MemoryReplayLedger(ttl=600.0)

assertion = await verify_response(body, idp=IDP, sp=SP, ledger=SEEN)
context = assertion.facts()   # for a Cedar policy to read; not a decision

LIMITS (value)#

LIMITS: Limits = Limits(max_bytes=524288, max_depth=40, max_elements=65536, max_attributes=256, max_attribute_bytes=65536)

IdentityProvider#

The directory whose assertions this service provider accepts.

certificates are the signing certificates obtained out of band — from the identity provider's metadata document, downloaded and reviewed by a person, not by this process. Give it more than one during a rotation: every configured key is tried, so the old and new certificates can both be live while the identity provider switches over.

Each is a PEM CERTIFICATE block, a PEM PUBLIC KEY block, or the bare base64 DER that a <ds:X509Certificate> element holds. They are parsed here, at construction, so a malformed certificate fails when the application is described rather than on the first login.

IdentityProvider(entity_id: str, certificates: tuple[str, ...]) -> None

MemoryReplayLedger#

A bounded, fail-closed assertion ledger for one worker.

MemoryReplayLedger(*, ttl: float, max_entries: int = 4096, clock: Callable[[], float] = <built-in function monotonic>) -> None

claim#

claim(key: str) -> bool

Spend key, refusing duplicates and raising when full.

ReplayLedger#

Where an assertion identifier is spent, exactly once.

MemoryReplayLedger.claim is synchronous and PostgresStore.claim is a coroutine, so a result that is awaitable is awaited.

Give the store a TTL at least as long as MAX_LIFETIME plus your clock skew. The ledger is what makes single use true, and an entry that ages out before the assertion expires reopens exactly the window it closed.

ReplayLedger(*args, **kwargs)

claim#

claim(key: str) -> Any

True when this caller took key, False when it was already held.

ReplayLedgerCapacityError#

The in-process replay ledger cannot safely admit another identifier.

SamlRefusal#

A SAML payload was refused, with a stable reason code.

Every refusal in this module carries one. The message says which check failed and what it saw; the code is what a log filter or a metric groups on, and it does not change when the prose does.

SamlRefusal(reason: str, message: str) -> None

ServiceProvider#

This application, as the assertion has to address it.

entity_id is what an AudienceRestriction must name and acs_url is what a bearer SubjectConfirmationData/@Recipient must equal. Both are required: an assertion minted for a different service provider verifying here is the whole reason AudienceRestriction exists.

ServiceProvider(entity_id: str, acs_url: str, clock_skew: float = 60.0) -> None

VerifiedAssertion#

What a verified assertion says. Facts, not a decision.

Every field here is the identity provider's claim, checked to have been signed by a key you configured and addressed to you, within its validity window, and not seen before. None of that makes it an authorization: pass facts() into Cedar context and let a policy decide what the values mean.

VerifiedAssertion(assertion_id: str, issuer: str, name_id: str, name_id_format: str, session_index: str, authn_context: str, authn_instant: datetime, not_before: datetime, not_on_or_after: datetime, attributes: Mapping[str, tuple[str, ...]]) -> None

facts#

facts() -> dict[str, Any]

This assertion as a mapping for Cedar context.

Flat, JSON-shaped, and named the way the rest of Wreath names facts. It is deliberately not a Principal and deliberately not a session: turning an assertion into either is the application's decision, and one this module must not make on its behalf.

ledger_declaration#

ledger_declaration(*, ttl: float, table: str = 'wreath_saml_seen', prefix: str = 'wreath_saml') -> Any

The Keyed declaration behind a shared replay ledger.

claim=True: the generated statement is one insert-or-reclaim, which is the whole point — a read followed by a write lets two workers both conclude they were first, and two workers both accepting one assertion is the replay this exists to prevent. There are no payload columns, because what is stored is the fact that the identifier was seen.

ttl is required rather than defaulted: how long an identifier must be remembered is a function of the identity provider's assertion lifetime, and a guess here silently reopens the window this closes. MAX_LIFETIME plus your clock skew is the smallest safe answer.

verify_response#

async verify_response(raw: bytes, *, idp: IdentityProvider, sp: ServiceProvider, ledger: ReplayLedger, in_response_to: str | None = None, now: datetime | None = None, limits: Limits | None = None) -> VerifiedAssertion

Verify a SAML 2.0 Response (or bare Assertion) and read its facts.

raw is the XML as it arrived — the decoded SAMLResponse form field, not the base64 wrapper. in_response_to is the ID of the authentication request this service provider issued, when it issued one; leave it None for an identity-provider-initiated login, and an assertion claiming to answer a request is then refused rather than accepted as unsolicited.

Raises SamlRefusal for everything: a malformed document, a transform outside the allow-list, a signature no configured key verifies, an assertion outside its window or addressed elsewhere, and an identifier already spent. Every refusal carries a reason code and says which check failed.

Returns facts. It does not sign anybody in, does not create a session, and does not decide anything — VerifiedAssertion.facts() goes into Cedar context and a policy reads it there.

wreath.sso#

Single sign-on: the flows, over the verification wreath already owned.

wreath.saml.verify_response does the hard half — exclusive canonicalization, signature wrapping made unexpressible by re-reading the parsed byte range, a replay ledger, every refusal named. It was also the module's only public symbol, so there was no AuthnRequest, no assertion-consumer endpoint, no service-provider metadata, and no login. wreath.auth verifies an OIDC id token and drives no relying-party flow. This module is the flows.

Nothing here re-implements verification. SamlServiceProvider.consume builds the IdentityProvider, ServiceProvider and in_response_to that verify_response already takes, and hands the answer to provisioning.

The reason this is more than glue: one identity provider per organisation#

A single-tenant application configures one IdP and is done. A B2B application has one per customer -- Okta here, Entra there, Google Workspace for the small ones -- and that changes the threat model rather than the amount of configuration.

Every tenant's IdP is a trusted signer of that tenant, and of nothing else. Verify an assertion against the union of every configured certificate and the smallest customer's Okta can mint an identity in the largest customer's account, with every signature check passing while it happens. There is no signature to notice: the assertion really is signed by a key this application really does trust.

So the signer set is scoped to the organisation, and the organisation comes from the login that began, never from the assertion. Reading it out of the assertion would let the assertion choose its own trust anchor, which is the same defect one indirection further along. PendingLogin is what carries it: the request id is minted here, stored with its organisation, and spent by the ACS.

Two rules inherited from elsewhere in the tree#

Nothing is fetched on the request path. OIDC discovery and JWKS refresh at lifespan startup and on an explicit refresh(), exactly as wreath.signatures refreshes its directories -- a key fetch driven by a request lets an unauthenticated caller aim an outbound request at a host they chose, and puts an identity provider's outage in the path of every login rather than of the refresh. An unknown kid is simply unverified.

Every refusal is distinct. A test that asserts only "it was refused" passes on whichever branch fired, including the one nobody was testing.

AttributeMapping#

Which assertion attributes become which fields. Declared, never inferred.

A heuristic that reads email and misses mail is confident and wrong, and the failure is one duplicate account per user per identity provider. So an attribute this mapping does not name is refused rather than guessed at or dropped: a directory sending something unexpected is a configuration question, and answering it silently is how it stays unanswered.

AttributeMapping(email: str = 'email', display_name: str | None = None, external_id: str | None = None) -> None

apply#

apply(attributes: Mapping[str, Any]) -> dict[str, Any]

Read the declared attributes out, refusing anything undeclared.

declared#

declared() -> tuple[str, ...]

IdentityProviderConfig#

One organisation's identity provider.

certificates are that organisation's signing certificates and only that organisation's. More than one during a rotation, which is why this is a tuple rather than a single value: the old and new can both be live while the directory switches over.

IdentityProviderConfig(organization: str, entity_id: str, sso_url: str, certificates: tuple[str, ...], mapping: AttributeMapping | None = None, roles: tuple[str, ...] = (), require_second_factor: bool = False) -> None

IdentityProviderDirectory#

Which identity provider signs for which organisation.

A miss is a refusal, never a fallback to "any configured provider" -- that fallback is the cross-tenant forgery this module exists to prevent.

IdentityProviderDirectory(providers: Iterable[IdentityProviderConfig] = ()) -> None

add#

add(provider: IdentityProviderConfig) -> None

for_organization#

for_organization(organization: str) -> IdentityProviderConfig

organizations#

organizations() -> tuple[str, ...]

JitProvisioning#

Turn a verified assertion into an account and a membership.

Both, together: an account with no membership sees nothing and reads as a bug in login, and a membership with no account cannot be signed in.

An existing account is adopted, never duplicated -- the choice scim_router's POST /Users already makes, for the same reason: somebody who signed up with a password before their company bought SSO keeps their data.

JitProvisioning(*, roles: Iterable[str] = (), vocabulary: Iterable[str] = ('member', 'admin'), require_second_factor: bool = False, revoke_sessions: Callable[[str], Awaitable[Any]] | None = None) -> None

provision#

provision(*, organization: str, email: str) -> ProvisionedLogin

revoke#

async revoke(*, organization: str, email: str) -> int

De-provision. A revoked user holding a live cookie is why SSO was bought.

OidcRelyingParty#

The authorization-code flow, with PKCE, against one organisation's issuer.

Three single-use values and each defends a different thing: state binds the callback to the browser that began (without it, an attacker's code redeemed in a victim's browser logs them into the attacker's account), nonce binds the id token to this authorization request, and the PKCE verifier binds the code redemption to this client.

OidcRelyingParty(*, issuer: str, client_id: str, pkce: "Literal['S256', 'plain']" = 'S256', ttl: float = 600.0, max_pending: int = 10000) -> None

begin_login#

begin_login(*, organization: str, session_id: str, now: float | None = None) -> _OidcFlow

check_nonce#

check_nonce(claims: Mapping[str, Any], *, expected_nonce: str) -> None

The claim that binds an id token to one authorization request.

consume_state#

consume_state(state: str, *, session_id: str, now: float | None = None) -> _OidcFlow

Spend the state, refusing a replay or another browser's callback.

fetches_on_request_path (property)#

fetches_on_request_path: bool

Always False, and asserted rather than promised.

key_for#

key_for(kid: str) -> Any

The signing key for a kid, or refuse. Never fetches.

refresh#

async refresh(fetch: Any) -> int

Reload discovery and JWKS. Called at lifespan startup, never per request.

PendingLogin#

An AuthnRequest this application issued and has not yet seen answered.

The request_id is what makes a response solicited. Without one, any assertion the identity provider ever signed is accepted whenever it arrives, which is a login as anybody from a captured POST body.

It also carries the organisation, and that is the load-bearing part: the ACS reads the trust anchor from here rather than from the assertion.

PendingLogin(request_id: str, relay_state: str, organization: str, issued_at: float, session_id: str = '', redirect_url: str = '') -> None

ProvisionedLogin#

What a completed SSO login produced.

ProvisionedLogin(user_id: str, organization: str, membership: tuple[str, ...], session_state: "Literal['authenticated', 'pending']" = 'authenticated', created: bool = False) -> None

SamlServiceProvider#

This application, as an identity provider has to address it.

entity_id is what an AudienceRestriction must name, acs_url is what a bearer confirmation must equal, and directory says whose signature counts for which organisation.

SamlServiceProvider(entity_id: str, acs_url: str, directory: IdentityProviderDirectory, certificates: tuple[str, ...] = (), pending: PendingLoginStore = <factory>) -> None

authn_request_xml#

authn_request_xml(pending: PendingLogin) -> str

The AuthnRequest for a pending login, as XML.

begin_login#

begin_login(*, organization: str, session_id: str, now: float | None = None) -> PendingLogin

Mint an AuthnRequest for one organisation and remember it.

Resolving the provider here is what makes an unconfigured organisation a refusal at the start of a login rather than a signature failure at the end of one.

consume#

async consume(raw: bytes, *, in_response_to: str, relay_state: str, session_id: str, ledger: Any, now: Any = None) -> Any

Verify an assertion against that organisation's signers only.

The organisation comes from the pending login, so the trust anchor is decided by the request this application issued rather than by the document answering it. That ordering is the whole defence against one customer's identity provider minting an identity in another's.

Returns wreath.saml.VerifiedAssertion: facts, not a session. Signing anybody in is JitProvisioning's job and a policy's after that.

metadata_xml#

metadata_xml() -> str

The document an administrator pastes into their identity provider.

Generated rather than written by hand: an entityID or an ACS URL that disagrees with what this application actually verifies is a login that fails with a signature error, which is the least informative way to discover a typo.

organization_for_request#

organization_for_request(request_id: str) -> str

Which organisation this request id was issued for.

SsoRefusal#

A login this application will not complete, and why.

Carries a reason code as well as prose so a caller can branch without matching on a message, the way wreath.saml.SamlRefusal already does.

SsoRefusal(reason: str, message: str) -> None

UnknownIdentityProvider#

No identity provider is configured for that organisation.

UnknownIdentityProvider(organization: str) -> None

wreath.oauth#

Issuing OAuth 2.1 tokens, for a deployment that decided to.

Wreath verifies bearer tokens with wreath.auth.JwtVerifier and issues them for deployments that act as their own identity provider.

The first obligation is that what this mints is what wreath already verifies. An issuer whose output its own verifier rejects is two features rather than one, so tests/oauth/ drives a minted token straight through JwtVerifier and BearerTokenBackend.

Two signers, and how to choose#

Es256Signer is the one most deployments want. It signs with ECDSA on P-256, publishes a real key set from jwks(), and adds no dependency -- wreath._webpush already signs ES256 VAPID tokens with hedged nonces over the standard library, and wreath.auth's JwtVerifier already verifies ES256 against an EcPublicKey. Both halves of the loop were already in the tree; this module was written claiming they were not, which was wrong.

HS256 is the default, and that is a choice about ceremony rather than about strength. A shared secret needs no key to generate, store, rotate or publish, which is right when the issuer and the resource server are the same deployment reading the same configuration. Its cost is that there is nothing safe to publish, so jwks() is honestly empty -- an oct entry carrying k would hand every reader the ability to mint tokens.

So: the moment anything outside this deployment verifies your tokens, pass signer=Es256Signer.generate(), keep the private scalar in configuration, and jwks() becomes a real key set.

The asymmetric signer does substantially more arithmetic than HMAC. It is the right trade at a login and the wrong operation to repeat on every application request. A deployment that needs both public verifiability and volume should issue long-lived tokens and refresh them rarely, which is what the shape is for anyway.

The three replay defences, and why each is separate#

• An authorization code is single use, and redeeming it twice revokes the token the first redemption issued. Refusing the second alone leaves the attacker's token live if they got there first, so the only safe answer to "two parties hold this code" is that neither keeps anything.
• A refresh token rotates, and presenting a rotated one revokes the whole chain. Rotation without reuse detection is rotation that tells you nothing.
• PKCE is required. plain is refused: a challenge that is the verifier protects nothing.

AuthorizationServer#

An OAuth 2.1 authorization server over wreath's own JWT vocabulary.

AuthorizationServer(*, issuer: str, secret: bytes | str = b'', clients: Iterable[ClientRegistration] = (), lifetime: float = 3600.0, code_ttl: float = 60.0, refresh_ttl: float = 2592000.0, signer: Any = None, authorization_detail_types: Mapping[str, type] | None = None, max_pending_grants: int = 10000) -> None

authorization_response#

authorization_response(*, code: str | None = None, error: str | None = None, state: str | None = None) -> dict[str, str]

Build RFC 9207 authorization response parameters.

The issuer is present on both success and error responses. Keeping this in the authorization server prevents endpoint glue from remembering the mix-up defence on one branch and forgetting it on another.

authorize#

authorize(*, client_id: str, redirect_uri: str, challenge_method: str = 'S256') -> ClientRegistration

Check an authorization request before anything is minted for it.

client_credentials#

client_credentials(*, client_id: str, client_secret: bytes | str | None = None, subject: str | None = None, scope: Iterable[str] = (), dpop_jkt: str = '', authorization_details: Any = ()) -> IssuedToken

A machine token. It carries no sub, and asking for one is refused.

A machine token with a subject is a machine that can act as a person, and every audit trail downstream then attributes its writes to them.

counters#

counters() -> Any

What signing has cost, so the ceiling is watched rather than warned about.

wreath.metrics.collect gathers by asking anything that offers counters(), so these reach a dashboard with no second registration.

ES256 signing is synchronous CPU work, so the request issuing a token waits for it even when unrelated requests do not. Divide signing_nanoseconds by elapsed wall-clock nanoseconds for the fraction of a core this issuer is spending; a threshold baked in here would be one guess applied to every deployment's latency budget.

introspection_jwt#

introspection_jwt(token: str, *, client_id: str, client_secret: bytes | str | None, now: float | None = None) -> str

is_revoked#