loom.rest.auth¶

Authentication for the Loom REST layer.

The layer is mechanism-agnostic: AuthenticationMiddleware drives any Authenticator and publishes the resulting Identity for the duration of the request. JwtAuthenticator and its ready-made JwtAuthMiddleware are the batteries-included implementation, bound from the app.rest.auth.jwt config section.

Install the optional JWT dependency with:

pip install "loom-kernel[jwt]"

class loom.rest.auth.AuthenticationMiddleware(app, *, authenticator, exclude_paths=())[source]¶

Bases: object

Authenticates every HTTP request through a pluggable mechanism.

On each HTTP request whose path is not excluded:

Builds RequestCredentials from the ASGI scope.
Asks the Authenticator for an identity.
On refusal, answers 401 with the framework’s standard error body and a WWW-Authenticate challenge. The message is deliberately generic for every failure mode (no oracle), and the refusal is logged at INFO — the response and the log have different audiences, and only the response has an attacker in it. Never the credential: a log holding a bearer token turns log access into API access.
On success, installs the identity for the duration of the request and restores the previous one in a finally — without it, a reused worker task would inherit the previous caller.

Non-HTTP scopes (WebSocket, lifespan) are passed through unchanged.

Parameters:

app (_ASGIApp) – The ASGI application to wrap.
authenticator (Authenticator) – Mechanism that verifies callers.
exclude_paths (Sequence[str]) – Exact request paths served without authentication.

Example:

app.add_middleware(
    AuthenticationMiddleware,
    authenticator=MyApiKeyAuthenticator(store),
    exclude_paths=("/health",),
)

class loom.rest.auth.Authenticator(*args, **kwargs)[source]¶

Bases: Protocol

Turns the credentials of one request into a verified identity.

Implementations must be stateless with respect to the request and safe to share across concurrent calls: one instance serves the whole application.

Example:

class ApiKeyAuthenticator:
    name = "api-key"
    provides_roles = True

    async def authenticate(self, credentials):
        key = credentials.header("x-api-key")
        owner = await self._keys.owner_of(key) if key else None
        if owner is None:
            return None
        return Identity(subject=owner.id, roles=owner.roles, mechanism=self.name)

property name: str¶: Short label of the mechanism, recorded on every identity it issues.

property provides_roles: bool¶

Whether the mechanism binds roles to the identity.

Startup gates rely on this: an endpoint whose authorization is role-based refuses to mount behind a mechanism that issues none, rather than letting every authenticated caller pick their own privileges.

async authenticate(credentials)[source]¶

Verify credentials and return the caller they identify.

Parameters:: credentials (RequestCredentials) – Headers, path and peer address of the request.
Returns:: The verified identity, or None to refuse the request. The refusal carries no reason on purpose: the response must not become an oracle about which part of the credentials failed.
Return type:: Identity | None

class loom.rest.auth.JwtAuthConfig(*, secret_path=None, public_keys=<factory>, algorithms=(), audience=None, issuer=None, leeway_seconds=0, exclude_paths=('/docs', '/redoc', '/openapi.json', '/metrics', '/health'), roles_claim=None)[source]¶

Bases: LoomFrozenStruct

Validated settings for JwtAuthMiddleware.

Binds from the app.rest.auth.jwt config section. Validation runs on construction (fail-fast): exactly one key source must be provided and the algorithm allowlist must be non-empty and coherent with that key source.

secret_path¶

Filesystem path of the shared secret for symmetric algorithms (HS*). A path and not the value: this is a msgspec.Struct, so any serializer emits its fields verbatim and a config dump would publish the key that verifies and signs. Mutually exclusive with public_keys.

Type:: str | None

public_keys¶

Static PEM-encoded public keys for asymmetric algorithms (RS*/ES*/EdDSA), keyed by the kid that selects them. Mutually exclusive with secret.

Type:: dict[str, str]

algorithms¶

Explicit allowlist of accepted JWT algorithms. The none algorithm is always forbidden.

Type:: tuple[str, …]

audience¶

Expected aud claim. Validated only when set.

Type:: str | None

issuer¶

Expected iss claim. Validated only when set.

Type:: str | None

leeway_seconds¶

Clock-skew tolerance applied to time-based claims.

Type:: int

exclude_paths¶

Exact request paths that bypass authentication.

Type:: tuple[str, …]

roles_claim¶

Name of the verified claim carrying the roles the caller is authorized to use. Consumed by the SQL endpoint: the effective roles are the claim values intersected with the connection allowlist, and the request body can only narrow that set.

Type:: str | None

Raises:

ConfigError – If key sources, algorithms, leeway, or the roles claim name are invalid.

Parameters:

secret_path (str | None)
public_keys (dict[str, str])
algorithms (tuple[str, ...])
audience (str | None)
issuer (str | None)
leeway_seconds (int)
exclude_paths (tuple[str, ...])
roles_claim (str | None)

Example YAML:

app:
  rest:
    auth:
      jwt:
        secret_path: ${oc.env:LOOM_JWT_SECRET_PATH}
        algorithms: [HS256]
        roles_claim: loom_sql_roles

classmethod from_signing_key(private_key_path=None, *, private_key_ref=None, key_ref_region=None, kid, algorithms, audience=None, issuer=None, leeway_seconds=0, exclude_paths=DEFAULT_EXCLUDE_PATHS, roles_claim=None, additional_public_keys=None)[source]¶

Build a verifier whose public key is derived from the signing key.

For the service that both issues and verifies. Two configured values that must match are two values that can disagree, and a stale public key does not fail loudly — it accepts nothing, or accepts what a rotated key signed. Deriving removes the second value.

Parameters:

private_key_path (str | None) – Filesystem path of the PEM signing key. Only the derived public key is kept; the private material is dropped. Mutually exclusive with private_key_ref.
private_key_ref (str | None) – Managed-store reference of the PEM signing key, as accepted by JwtIssuerConfig.private_key_ref. Mutually exclusive with private_key_path.
key_ref_region (str | None) – AWS region for the ref resolver. Requires private_key_ref.
kid (str) – Key identifier the derived public key is published under.
algorithms (tuple[str, ...]) – Explicit allowlist of accepted JWT algorithms.
audience (str | None) – Expected aud claim. Validated only when set.
issuer (str | None) – Expected iss claim. Validated only when set.
leeway_seconds (int) – Clock-skew tolerance for time-based claims.
exclude_paths (tuple[str, ...]) – Exact request paths that bypass authentication.
roles_claim (str | None) – Verified claim carrying the caller roles.
additional_public_keys (dict[str, str] | None) – Extra public keys by kid, so the previous key stays published for one rotation window. Public material, so unlike the signing key it is safe to carry as a value. Reusing the derived kid here is refused: this method exists so the published key cannot drift from the signing key, and an override would reintroduce the drift.

Returns:

A validated JwtAuthConfig publishing the derived key.

Raises:

ConfigError – If the sources are not exactly one, or the signing key cannot be read or parsed. The cause is chained: parser errors name the reason and never carry the material.

Return type:

JwtAuthConfig

verification_key(key_id)[source]¶

Return the key that verifies a token, selected by its kid.

Selection is explicit and never exhaustive: trying every configured key in turn would decouple each algorithm from its key family, which is what makes algorithm confusion structurally impossible here.

Parameters:: key_id (str | None) – kid header of the token, or None when it carries none.
Returns:: The key material, or None when no configured key applies — an unknown kid, or a missing one while several keys are configured.
Return type:: str | None

class loom.rest.auth.JwtAuthMiddleware(app, *, config)[source]¶

Bases: object

Stateless JWT bearer authentication, as a ready-made middleware.

Thin composition over AuthenticationMiddleware and JwtAuthenticator: it exists so applications that only need JWT wire one class instead of two.

Parameters:

app (_ASGIApp) – The ASGI application to wrap.
config (JwtAuthConfig) – Validated JwtAuthConfig.

Raises:

ImportError – If the optional pyjwt dependency is not installed.

Example — FastAPI:

from loom.rest.auth import JwtAuthConfig, JwtAuthMiddleware

config = JwtAuthConfig(secret_path="/run/secrets/jwt", algorithms=("HS256",))
app.add_middleware(JwtAuthMiddleware, config=config)

class loom.rest.auth.JwtAuthenticator(config)[source]¶

Bases: object

Authenticates callers from a stateless JWT bearer token.

Verification is fully stateless: no server-side session storage and no remote JWKS fetch. Signature, exp and sub are always required (a token without a subject carries no identity to bind an authorization decision to, nor to audit afterwards); aud/iss are validated only when configured.

Parameters:: config (JwtAuthConfig) – Validated JWT settings.
Raises:: ImportError – If the optional pyjwt dependency is not installed.

Example:

authenticator = JwtAuthenticator(
    JwtAuthConfig(
        secret_path="/run/secrets/jwt",
        algorithms=("HS256",),
        roles_claim="loom_sql_roles",
    )
)

property name: str¶: Return the mechanism label recorded on issued identities.

property provides_roles: bool¶: Whether a verified claim binds roles to the caller identity.

async authenticate(credentials)[source]¶

Verify the bearer token and project its claims onto an identity.

Parameters:: credentials (RequestCredentials) – Headers and path of the request.
Returns:: The verified identity, or None when the header is absent, uses another scheme, or the token fails verification.
Return type:: Identity | None

class loom.rest.auth.JwtIssuer(config)[source]¶

Bases: object

Mints JWT bearer tokens for a verified Identity.

Everything the token says comes from the identity: there is no way for a caller to add a claim, so nobody can widen their own roles or speak for another subject through this door.

The signing key is read once here and never kept on the config, which is a msgspec.Struct whose fields any serializer would publish.

Every issuing is logged at INFO with the subject and the roles granted. That is a deliberate audit trail, not diagnostics: without it an access is not attributable to who asked for the token nor to the privileges it carried. The subject is usually personal data, so route these logs accordingly. The token and the key are never logged.

Parameters:

config (JwtIssuerConfig) – Validated issuer settings.

Raises:

ImportError – If the optional pyjwt dependency is not installed.
ConfigError – If the signing key cannot be read, or cannot sign with the configured algorithm.

Example:

issuer = JwtIssuer(JwtIssuerConfig(
    private_key_path="/run/secrets/jwt.pem", algorithm="EdDSA",
    audience="my-api", issuer="my-gateway", roles_claim="loom_sql_roles",
))
issued = issuer.issue(identity)

issue(identity, *, ttl=None)[source]¶

Mint a token for identity.

Parameters:

identity (Identity) – Verified caller the token speaks for.
ttl (timedelta | None) – Lifetime override, bounded by the configured one.

Returns:

The token, its expiry and its jti.

Raises:

ValueError – If the identity is anonymous, carries no role while a roles claim is configured, holds an attribute that would be unreadable once encoded, or ttl is out of bounds.
RuntimeError – If signing fails. The cause is not chained, so the caller learns nothing about the key or the algorithm.

Return type:

IssuedToken

class loom.rest.auth.JwtIssuerConfig(*, private_key_path=None, private_key_ref=None, key_ref_region=None, secret_path=None, algorithm='', audience='', issuer='', roles_claim='', ttl_seconds=900, kid=None, allow_symmetric_signing=False)[source]¶

Bases: LoomFrozenStruct

Validated settings for JwtIssuer.

Separate from JwtAuthConfig on purpose, and not a few extra fields on it: a service that only verifies must have no configuration path through which signing material can land in its process. The fields diverge anyway — one algorithm instead of an allowlist, audience and issuer mandatory rather than optional, a lifetime the verifier has no use for.

The private key is read from private_key_path and never held as a field. __repr__ redaction would not be enough: this is a msgspec.Struct, so msgspec.json.encode and to_builtins emit every field verbatim without going through it, and a config dump would publish the signing key.

private_key_path¶

Filesystem path of the PEM signing key, for asymmetric algorithms. Mutually exclusive with the other sources.

Type:: str | None

private_key_ref¶

Managed-store reference of the PEM signing key, as "<resolver>:<key>" with resolver one of secrets or ssm (e.g. "secrets:/myapp/prod/jwt-signing-key"). Resolved once, when the issuer loads the key, so the material never touches disk or config — and rotating the stored value requires a restart. The Secrets Manager resolver navigates dots as JSON paths, so the key name must not contain .. Mutually exclusive with the other sources.

Type:: str | None

key_ref_region¶

AWS region the ref resolver is built with. Defaults to boto3’s own resolution chain. Requires private_key_ref.

Type:: str | None

secret_path¶

Filesystem path of the shared secret for HS* algorithms. Requires allow_symmetric_signing. Mutually exclusive with the other sources.

Type:: str | None

algorithm¶

The single algorithm tokens are signed with. Issuing chooses one; only verification negotiates an allowlist.

Type:: str

audience¶

aud stamped on every token. Mandatory: a token without an audience is valid at any service that shares the key.

Type:: str

issuer¶

iss stamped on every token. Mandatory.

Type:: str

roles_claim¶

Claim carrying the caller roles. Mandatory, and never a registered claim: an issuer that does not own that name lets an attribute impersonate it. An identity without roles is refused.

Type:: str

ttl_seconds¶

Lifetime of a minted token, in 1..MAX_ISSUER_TTL_SECONDS, and the ceiling for any per-call override.

Type:: int

kid¶

Key identifier stamped on the token header, so a verifier can select the right key during a rotation overlap.

Type:: str | None

allow_symmetric_signing¶

Opt-in required for HS*. With a shared secret every verifier can also mint, so the choice must be deliberate.

Type:: bool

Raises:

ConfigError – If key sources, algorithm, audience, issuer, roles claim or lifetime are invalid.

Parameters:

private_key_path (str | None)
private_key_ref (str | None)
key_ref_region (str | None)
secret_path (str | None)
algorithm (str)
audience (str)
issuer (str)
roles_claim (str)
ttl_seconds (int)
kid (str | None)
allow_symmetric_signing (bool)

Note

Built programmatically, not bound from a config section: issuing happens in an application use case that injects the issuer, not in the REST layer that owns app.rest.auth. Resolve the values however the service resolves its own settings.

Example:

config = JwtIssuerConfig(
    private_key_path=os.environ["JWT_SIGNING_KEY_PATH"],
    algorithm="EdDSA",
    audience="my-api",
    issuer="my-gateway",
    roles_claim="loom_sql_roles",
    ttl_seconds=900,
    kid="2026-08",
)

load_signing_key()[source]¶

Load the signing key, so it lives in the issuer and not in the config.

Returns:: The key material, read from disk or resolved from the managed store when private_key_ref is set.
Raises:: ConfigError – If the file cannot be read or is not UTF-8 text (the cause is never chained: an OS error carries the path, and a UnicodeDecodeError carries the file contents on exc.object), or if the managed store cannot answer (that cause is chained: resolver errors never carry the material).
Return type:: str

class loom.rest.auth.RequestCredentials(headers, path, client_host=None)[source]¶

Bases: object

What an authenticator is allowed to look at, free of any ASGI type.

Deliberately narrow: everything an authentication mechanism may legitimately read, and nothing that would let it reach into request handling. The body is absent on purpose — authenticating on it would require buffering the request before deciding whether the caller exists.

Parameters:

headers (Mapping[str, str])
path (str)
client_host (str | None)

headers¶

Request headers keyed by lowercase name.

Type:: collections.abc.Mapping[str, str]

path¶

Request path, so a mechanism can scope itself per route.

Type:: str

client_host¶

Peer address when the server exposes one, else None.

Type:: str | None

Example:

credentials = RequestCredentials(
    headers={"authorization": "Bearer ..."},
    path="/sql/analytics",
)

header(name)[source]¶

Return a header value by case-insensitive name.

Parameters:: name (str) – Header name in any casing.
Returns:: The header value, or None when the header is absent.
Return type:: str | None