Security

• Start here: Security guide
• Common recipes: resolve_client_address(request.scope, trusted) returns the address a trusted proxy vouched for. ClientAddressMiddleware resolves it once per request so every consumer reads one value. JWTVerifier.keys(JWTKey.pem(...), audience="my-api").verify_header(header) returns the claims of the bearer token a caller presented. ClientCredentials("payments-api", audience="payments-api").auth() sends a token for the service to another API, fetched from the registered OAuthClient.

grelmicro.security

Security.

Checks a service runs on an inbound request, and the tokens it sends on an outbound one. grelmicro validates what arrives and authenticates what it sends. It never issues credentials.

TrustedProxies names your own proxies and resolve_client_address returns the address one of them vouched for, so a spoofed X-Forwarded-For never becomes a rate limiter key or an audit record. ClientAddressMiddleware resolves it once per request.

OAuthClient registers the service with its authorization server, and ClientCredentials and TokenExchange get the tokens it calls other APIs with.

Read more in the Security docs.

Each name loads its submodule on first access.

TrustedProxies

TrustedProxies(
    networks: Iterable[str | IPAddress | IPNetwork],
    /,
    *,
    max_hops: int | None = None,
    max_entries: int = 64,
    max_header_bytes: int = 8192,
)

The proxies whose X-Forwarded-For entries may be believed.

Build one at startup and reuse it. Every entry is parsed strictly, so a typo fails here rather than becoming a rule that silently matches nothing.

There is no wildcard. To trust every peer, pass ["0.0.0.0/0", "::/0"], which is visible in a configuration review in a way a * is not.

Compile the trusted set, rejecting anything unparsable.

PARAMETER	DESCRIPTION
networks	Addresses and CIDR ranges of your own proxies. Required, with no default. An empty iterable is legal and means trust nothing, so resolution always returns the transport peer. TYPE: Iterable[str | IPAddress | IPNetwork]
max_hops	Most trusted proxies to walk past. Bounds a forged chain. None means no cap. TYPE: int | None DEFAULT: None
max_entries	Most header entries to parse before giving up. TYPE: int DEFAULT: 64
max_header_bytes	Most header bytes to parse before giving up. TYPE: int DEFAULT: 8192

RAISES	DESCRIPTION
SettingsValidationError	If an entry is not an IP address or a CIDR range, or a bound is outside its usable range. max_entries and max_header_bytes cap work, so zero or less caps nothing. max_hops counts proxies to walk past, so it may be zero but never negative.

ClientAddress dataclass

ClientAddress(
    ip: IPAddress,
    port: int | None,
    reason: ClientAddressReason,
    hops: int,
)

A resolved client address, with how it was reached.

ip instance-attribute

ip: IPAddress

The address. Always parsed, never raw header text.

port instance-attribute

port: int | None

The source port, when one was given.

reason instance-attribute

reason: ClientAddressReason

Why this address was chosen. RESOLVED is the trusted case.

hops instance-attribute

hops: int

Trusted proxies skipped before landing on this address.

forwarded property

forwarded: bool

Whether a trusted proxy vouched for this address.

The test to use behind a proxy for anything that treats the address as the caller's identity, such as an allowlist or a private network check. It is True for RESOLVED alone.

degraded property

degraded: bool

Whether ip is one of your own proxies rather than a caller.

True when the peer was a trusted proxy but its chain could not be walked to a client, so ip is that proxy's own address and is never the caller's. Any check that treats it as the caller must refuse.

False is not the same as vouched for. It also covers the two cases where the peer is the caller only if nothing unlisted sits in front of the app: UNTRUSTED_PEER and NO_FORWARDED_HEADER. Use forwarded when the address has to carry an identity behind a proxy.

key property

key: str

Canonical form, safe to use as a rate limiter key.

An IPv4-mapped address folds onto its IPv4 form, so one caller never occupies two buckets.

It is a bucket, not an identity. When degraded is True every caller behind the proxy shares one key, so anything that has to keep callers apart, such as an idempotency key or an allowlist, checks forwarded first.

ClientAddressReason

Bases: StrEnum

Why ClientAddress.ip holds the address it holds.

RESOLVED is the one value the library can vouch for on its own: a trusted proxy wrote the entry. Every other value means ip is the verified transport peer, and they split in two.

With NO_FORWARDED_HEADER and UNTRUSTED_PEER the peer is the caller, as long as nothing sits in front of the app that is missing from the trusted set. That is a claim about your topology, not one this module can check.

Every remaining value means the peer is itself a trusted proxy whose chain could not be walked to a client, so ip is one of your own proxies and never the caller. ClientAddress.degraded marks exactly those.

RESOLVED class-attribute instance-attribute

RESOLVED = 'resolved'

A trusted proxy vouched for this address.

NO_FORWARDED_HEADER class-attribute instance-attribute

NO_FORWARDED_HEADER = 'no-forwarded-header'

A trusted peer sent no X-Forwarded-For.

UNTRUSTED_PEER class-attribute instance-attribute

UNTRUSTED_PEER = 'untrusted-peer'

The peer is not a trusted proxy, so the header was ignored.

CHAIN_EXHAUSTED class-attribute instance-attribute

CHAIN_EXHAUSTED = 'chain-exhausted'

Every entry in the chain was a trusted proxy.

HOP_LIMIT class-attribute instance-attribute

HOP_LIMIT = 'hop-limit'

The walk passed more trusted proxies than max_hops.

MALFORMED_ENTRY class-attribute instance-attribute

MALFORMED_ENTRY = 'malformed-entry'

An entry was not an address, which stops the walk.

HEADER_TOO_LARGE class-attribute instance-attribute

HEADER_TOO_LARGE = 'header-too-large'

The header was longer than max_header_bytes.

TOO_MANY_ENTRIES class-attribute instance-attribute

TOO_MANY_ENTRIES = 'too-many-entries'

More entries than max_entries, and every one read was trusted.

ClientAddressMiddleware

ClientAddressMiddleware(
    app: ASGIApp,
    *,
    trusted: TrustedProxies,
    overwrite_scope_client: bool = False,
)

Resolve the client address once and cache it on the request.

Stores a ClientAddress under scope["state"]["client_address"], so every handler and every other middleware reads one value instead of each resolving its own and drifting apart.

from grelmicro.security import ClientAddressMiddleware, TrustedProxies

app.add_middleware(
    ClientAddressMiddleware, trusted=TrustedProxies(["10.0.0.0/8"])
)

Pure ASGI, so it works on any ASGI server. It acts on http and websocket scopes and passes every other scope through untouched.

Initialize the middleware with the trusted set.

PARAMETER	DESCRIPTION
app	The next ASGI application in the chain. TYPE: ASGIApp
trusted	The proxies whose forwarded entries may be believed. TYPE: TrustedProxies
overwrite_scope_client	Also rewrite scope["client"] with the resolved address. Off by default, because it discards the verified transport peer that an audit record wants, and because a server that already rewrites it would then be running the walk twice. Turn it on for code that reads request.client.host and cannot be changed. TYPE: bool DEFAULT: False

app instance-attribute

app = app

trusted instance-attribute

trusted = trusted

overwrite_scope_client instance-attribute

overwrite_scope_client = overwrite_scope_client

JWTVerifier

JWTVerifier(*args: object, **kwargs: object)

Bases: Reconfigurable[JWTKeysConfig | JWKSConfig | DiscoveryConfig]

Verifies inbound JWTs against a key set and a claim policy.

Build one with a factory that names where the keys come from: JWTVerifier.discover for an issuer that publishes its metadata, JWTVerifier.jwks for a JWKS endpoint, JWTVerifier.keys for keys you hold, or JWTVerifier.from_config for a config assembled elsewhere. There is no bare constructor, because no key source is a safe default.

Keys are parsed once, when they are loaded. Verification runs in the compiled core with the GIL released, so a thread pool verifies in parallel. Keys a provider publishes are fetched by refresh, a coroutine, and never on the request path.

keys and jwks also read the environment unless GREL_ENV_LOAD=0, under GREL_JWTVERIFIER_, or GREL_JWTVERIFIER_{NAME}_ for a named verifier, and a keyword always wins. The environment may say whose tokens to trust, such as the issuer or the audience. It never chooses the key source, the algorithm or the key material, and a mounted file changes only what a verification costs while the service runs.

Example

verifier = JWTVerifier.keys(
    JWTKey.pem(public_pem, algorithm="RS256"),
    audience="grelmicro-api",
    issuer="https://auth.example.com/",
)
claims = verifier.verify(token)

Refuse construction: a verifier has no default key source.

name property

name: str

The instance name.

ready property

ready: bool

Whether a key set is loaded. Always true for keys held in code.

metadata_url property

metadata_url: str | None

The issuer metadata document discovery last read keys through.

None for a verifier that does not discover, and until its first discovery succeeds.

stale property

stale: bool

Whether the next refresh would fetch.

Never true for keys held in code. For keys a provider publishes, true before the first load, once ttl has passed, and once a token named a key the current set does not hold.

keys classmethod

keys(
    *keys: JWTKey,
    audience: str | Sequence[str] | Unset | None = UNSET,
    issuer: str | Sequence[str] | None = None,
    leeway: int | None = None,
    token_type: Literal["at+jwt"] | None = None,
    scope_claims: str | Sequence[str] | None = None,
    required: Sequence[str] | None = None,
    cache_size: int | None = None,
    cache_key: Literal["sha256", "token"] | None = None,
    cache_ttl: int | timedelta | None = None,
    name: str = "default",
    env_load: bool | None = None,
) -> Self

Build a verifier from keys you hold, such as a PEM or a secret.

A setting left out is read from the environment, then takes the default JWTKeysConfig gives it. Key material only ever comes from code.

PARAMETER	DESCRIPTION
*keys	Keys this verifier accepts, selected by the token's kid. TYPE: JWTKey DEFAULT: ()
audience	Accepted aud values. None answers to no audience, which refuses any token that names one, and only code can say it. Left out, it is read from the environment, where it is required. TYPE: str | Sequence[str] | Unset | None DEFAULT: UNSET
issuer	Accepted iss values. Left out, nothing checks the issuer. TYPE: str | Sequence[str] | None DEFAULT: None
leeway	Seconds of clock skew allowed on exp, nbf and iat. TYPE: int | None DEFAULT: None
token_type	Type every token must declare in its typ header. TYPE: Literal['at+jwt'] | None DEFAULT: None
scope_claims	Claims the granted scopes are read from, in order. TYPE: str | Sequence[str] | None DEFAULT: None
required	Further claims that must be present. exp always is. TYPE: Sequence[str] | None DEFAULT: None
cache_size	Verified tokens held in memory. Zero turns the cache off. TYPE: int | None DEFAULT: None
cache_key	What the cache keys on: a digest, or the encoded token. TYPE: Literal['sha256', 'token'] | None DEFAULT: None
cache_ttl	How long a verified token stays cached, in whole seconds or as a timedelta. A float is refused. Zero turns the cache off. TYPE: int | timedelta | None DEFAULT: None
name	Instance name, which is the environment namespace: GREL_JWTVERIFIER_{NAME}_. The default instance reads GREL_JWTVERIFIER_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a key or a setting is refused.

jwks classmethod

jwks(
    url: str | None = None,
    *,
    audience: str | Sequence[str] | Unset | None = UNSET,
    issuer: str | Sequence[str] | None = None,
    algorithm: _AsymmetricAlgorithm | None = None,
    ttl: int | timedelta | None = None,
    retry_interval: float | None = None,
    timeout: float | None = None,
    max_bytes: int | None = None,
    max_keys: int | None = None,
    leeway: int | None = None,
    token_type: Literal["at+jwt"] | None = None,
    scope_claims: str | Sequence[str] | None = None,
    required: Sequence[str] | None = None,
    cache_size: int | None = None,
    cache_key: Literal["sha256", "token"] | None = None,
    cache_ttl: int | timedelta | None = None,
    fetch: Fetcher | None = None,
    name: str = "default",
    env_load: bool | None = None,
) -> Self

Build a verifier from keys a provider publishes at a JWKS endpoint.

Nothing is fetched here. Open the verifier with async with, or await refresh, before the first request. A setting left out is read from the environment, then takes the default JWKSConfig gives it.

PARAMETER	DESCRIPTION
url	The JWKS endpoint. Must be https. Left out, it is read from the environment. TYPE: str | None DEFAULT: None
audience	Accepted aud values. None answers to no audience, which refuses any token that names one, and only code can say it. Left out, it is read from the environment, where it is required. TYPE: str | Sequence[str] | Unset | None DEFAULT: UNSET
issuer	Accepted iss values. Left out, nothing checks the issuer. TYPE: str | Sequence[str] | None DEFAULT: None
algorithm	Algorithm to pin for keys that publish none, as Entra ID does. Only code chooses it. TYPE: _AsymmetricAlgorithm | None DEFAULT: None
ttl	How long a fetched document is treated as current, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
retry_interval	Least time between two fetches. TYPE: float | None DEFAULT: None
timeout	Seconds to wait for the endpoint before giving up. TYPE: float | None DEFAULT: None
max_bytes	Largest document accepted. TYPE: int | None DEFAULT: None
max_keys	Most keys accepted from one document. TYPE: int | None DEFAULT: None
leeway	Seconds of clock skew allowed on exp, nbf and iat. TYPE: int | None DEFAULT: None
token_type	Type every token must declare in its typ header. TYPE: Literal['at+jwt'] | None DEFAULT: None
scope_claims	Claims the granted scopes are read from, in order. TYPE: str | Sequence[str] | None DEFAULT: None
required	Further claims that must be present. exp always is. TYPE: Sequence[str] | None DEFAULT: None
cache_size	Verified tokens held in memory. Zero turns the cache off. TYPE: int | None DEFAULT: None
cache_key	What the cache keys on: a digest, or the encoded token. TYPE: Literal['sha256', 'token'] | None DEFAULT: None
cache_ttl	How long a verified token stays cached, in whole seconds or as a timedelta. A float is refused. Zero turns the cache off. TYPE: int | timedelta | None DEFAULT: None
fetch	Fetcher to use. Defaults to one built on httpx. TYPE: Fetcher | None DEFAULT: None
name	Instance name, which is the environment namespace: GREL_JWTVERIFIER_{NAME}_. The default instance reads GREL_JWTVERIFIER_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused, or a variable names the algorithm.

discover classmethod

discover(
    issuer: str | None = None,
    *,
    audience: str | Sequence[str] | Unset | None = UNSET,
    algorithm: _AsymmetricAlgorithm | None = None,
    ttl: int | timedelta | None = None,
    retry_interval: float | None = None,
    timeout: float | None = None,
    max_bytes: int | None = None,
    max_keys: int | None = None,
    leeway: int | None = None,
    token_type: Literal["at+jwt"] | None = None,
    scope_claims: str | Sequence[str] | None = None,
    required: Sequence[str] | None = None,
    cache_size: int | None = None,
    cache_key: Literal["sha256", "token"] | None = None,
    cache_ttl: int | timedelta | None = None,
    fetch: Fetcher | None = None,
    name: str = "default",
    env_load: bool | None = None,
) -> Self

Build a verifier from an issuer, finding its keys in its metadata.

Nothing is fetched here. Open the verifier with async with, or await refresh, before the first request. The issuer's RFC 8414 authorization server metadata is read first, then its OpenID Connect discovery document. The one that answers must name this issuer exactly and an https JWKS endpoint, and every token must carry the issuer in iss.

PARAMETER	DESCRIPTION
issuer	The issuer, exactly as its tokens carry it in iss. Must be https. Left out, it is read from the environment. TYPE: str | None DEFAULT: None
audience	Accepted aud values. None answers to no audience, which refuses any token that names one, and only code can say it. Left out, it is read from the environment, where it is required. TYPE: str | Sequence[str] | Unset | None DEFAULT: UNSET
algorithm	Algorithm to pin for keys that publish none, as Entra ID does. Only code chooses it. TYPE: _AsymmetricAlgorithm | None DEFAULT: None
ttl	How long the metadata and the key set are treated as current, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
retry_interval	Least time between two fetches. TYPE: float | None DEFAULT: None
timeout	Seconds to wait for the provider before giving up. TYPE: float | None DEFAULT: None
max_bytes	Largest document accepted. TYPE: int | None DEFAULT: None
max_keys	Most keys accepted from one document. TYPE: int | None DEFAULT: None
leeway	Seconds of clock skew allowed on exp, nbf and iat. TYPE: int | None DEFAULT: None
token_type	Type every token must declare in its typ header. TYPE: Literal['at+jwt'] | None DEFAULT: None
scope_claims	Claims the granted scopes are read from, in order. TYPE: str | Sequence[str] | None DEFAULT: None
required	Further claims that must be present. exp always is. TYPE: Sequence[str] | None DEFAULT: None
cache_size	Verified tokens held in memory. Zero turns the cache off. TYPE: int | None DEFAULT: None
cache_key	What the cache keys on: a digest, or the encoded token. TYPE: Literal['sha256', 'token'] | None DEFAULT: None
cache_ttl	How long a verified token stays cached, in whole seconds or as a timedelta. A float is refused. Zero turns the cache off. TYPE: int | timedelta | None DEFAULT: None
fetch	Fetcher to use. Defaults to one built on httpx. TYPE: Fetcher | None DEFAULT: None
name	Instance name, which is the environment namespace: GREL_JWTVERIFIER_{NAME}_. The default instance reads GREL_JWTVERIFIER_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

Example

verifier = JWTVerifier.discover(
    "https://auth.example.com/", audience="orders-api"
)

async with verifier:
    claims = verifier.verify_header(authorization)

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused, or a variable names the algorithm.

from_config classmethod

from_config(
    config: JWTKeysConfig | JWKSConfig | DiscoveryConfig,
    *,
    fetch: Fetcher | None = None,
    name: str = "default",
) -> Self

Build a verifier from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the verifier is not registered for live reload.

PARAMETER	DESCRIPTION
config	The keys, or where they are published, and the claim policy to enforce, taken as they are. TYPE: JWTKeysConfig | JWKSConfig | DiscoveryConfig
fetch	Fetcher for a JWKSConfig or a DiscoveryConfig. Defaults to one built on httpx, and is refused for keys held in code. TYPE: Fetcher | None DEFAULT: None
name	Instance name. TYPE: str DEFAULT: 'default'

RAISES	DESCRIPTION
SettingsValidationError	If the core cannot read a key.
TypeError	If fetch is given for keys held in code.

refresh async

refresh(*, force: bool = False) -> bool

Fetch the published key set, and return whether the keys changed.

Does nothing for keys held in code. For keys a provider publishes, it does nothing while they are fresh and never fetches more often than retry_interval. A failure raises and leaves the loaded keys in place.

One fetch runs at a time. A caller arriving while one runs waits for it and gets its answer, so a burst of tokens naming a new key costs the provider one request. That makes it safe to await from a request that was refused with unknown-key, and to verify again when it returns True:

try:
    claims = verifier.verify(token)
except TokenRejectedError as error:
    if error.reason is not TokenRejectedReason.UNKNOWN_KEY or not await verifier.refresh():
        raise
    claims = verifier.verify(token)

PARAMETER	DESCRIPTION
force	Fetch even when the current keys are still fresh. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
SigningKeysUnavailableError	If the document cannot be fetched, or holds no usable key.

verify

verify(token: str) -> JWTClaims

Return the claims of token, or raise TokenRejectedError.

A token verified earlier in this process is answered from the cache until cache_ttl or its own exp passes, whichever comes first, so a cache hit is never staler than a full verification.

PARAMETER	DESCRIPTION
token	The encoded JWT, with no scheme prefix. TYPE: str

RAISES	DESCRIPTION
TokenRejectedError	If the token does not verify.
SigningKeysUnavailableError	If no key set has loaded yet.

verify_header

verify_header(header: str | None) -> JWTClaims

Return the claims of the bearer token in header.

The scheme is matched without regard to case, and the token read behind one or more spaces, as RFC 7235 asks.

PARAMETER	DESCRIPTION
header	The Authorization header value, or None. TYPE: str | None

RAISES	DESCRIPTION
TokenRejectedError	With scheme when header carries no bearer token, or with the reason the token itself failed.

JWTKeysConfig

Bases: JWTPolicy

A claim policy together with the keys that verify against it.

keys instance-attribute

keys: list[JWTKey]

Keys this verifier accepts, selected by the token's kid.

from_jwks classmethod

from_jwks(
    jwks: Mapping[str, Any],
    *,
    algorithm: _AsymmetricAlgorithm | None = None,
    **policy: Any,
) -> JWTKeysConfig

Build a config from a JWKS document.

Keys marked for encryption are skipped, and so are symmetric keys: a signature is never verified with the first, and the second would be a shared secret published to anyone who can read the document. Everything else in policy is passed through.

PARAMETER	DESCRIPTION
jwks	A JWKS document, as an OIDC provider serves it. TYPE: Mapping[str, Any]
algorithm	Algorithm to pin for keys that name none. TYPE: _AsymmetricAlgorithm | None DEFAULT: None
**policy	Any other JWTKeysConfig setting. TYPE: Any DEFAULT: {}

Example

jwks = httpx.get(f"{issuer}/.well-known/jwks.json").json()
config = JWTKeysConfig.from_jwks(
    jwks, audience="my-api", issuer=issuer
)

JWTPolicy

Bases: BaseModel

What a verifier checks, and what it remembers.

Everything here is independent of where the keys came from, so a verifier built from a PEM and one built from a JWKS endpoint enforce it the same way.

audience instance-attribute

audience: list[str] | None

Accepted aud values, one or several. Required, because a resource server has to check that a token was issued for it. None says the service identifies with no audience, and RFC 7519 then requires refusing any token that carries an aud claim, which fits a provider whose access tokens carry none, such as AWS Cognito.

issuer class-attribute instance-attribute

issuer: list[str] = Field(default_factory=list)

Accepted iss values. Empty leaves the issuer unchecked.

leeway class-attribute instance-attribute

leeway: int = 0

Seconds of clock skew allowed on exp, nbf and iat.

token_type class-attribute instance-attribute

token_type: Literal['at+jwt'] | None = None

Type every token must declare in its typ header. None accepts a token that declares none, JWT, JOSE or at+jwt, and refuses any other type, such as a DPoP proof or a logout token. at+jwt requires the access token type of RFC 9068, which Microsoft Entra ID does not send and Keycloak sends only when a client asks for it.

scope_claims class-attribute instance-attribute

scope_claims: list[str] = Field(
    default_factory=lambda: ["scope", "scp"]
)

Claims the granted scopes are read from, in order. The first one the token carries decides, as a space-separated string or an array of strings. Microsoft Entra ID writes scp as a string and Okta writes it as an array. A claim of any other shape grants nothing.

required class-attribute instance-attribute

required: list[str] = Field(default_factory=lambda: ["exp"])

Further claims that must be present. exp and sub are always required and do not need naming, and naming an audience or an issuer requires that claim too, so this only ever adds to what is enforced.

cache_size class-attribute instance-attribute

cache_size: int = 1024

Verified tokens held in memory. A client resends one token until it expires, so a hit answers without repeating the signature check. Zero turns the cache off.

cache_key class-attribute instance-attribute

cache_key: str = 'sha256'

What the cache holds as its key. sha256 keeps a digest, so a live bearer token is not held in memory for the lifetime of the entry. It is computed in the core and costs around 80 ns on a hit, under 1% of a verification. token keeps the encoded token instead, which is what an in-process cache normally does and is the faster of the two.

cache_ttl class-attribute instance-attribute

cache_ttl: Retention = timedelta(minutes=5)

How long a verified token stays cached, in whole seconds or as a timedelta. A float is refused. Zero turns the cache off. An entry also never outlives the token's own exp, so this is the bound that matters for a long-lived token: it caps how long a token withdrawn upstream keeps being accepted from memory.

enforced_claims

enforced_claims() -> list[str]

Return the claims a token must carry under this policy.

Naming an audience or an issuer requires the matching claim. A token that simply omits aud satisfies an audience check otherwise, which turns a configured check into one that silently does not apply to the tokens most worth checking. Set audience=None for a provider whose tokens carry none, such as an AWS Cognito access token.

JWTKey

Bases: BaseModel

One verification key and the algorithm it verifies.

Build one with pem, secret or jwk. The constructor takes the same fields, for a key that arrives as data, such as from a settings file.

The key material is held as a secret, so it never appears in a repr, a log line or a dumped configuration.

Example

JWTKey.pem(public_pem, algorithm="RS256", kid="2026-09")

algorithm instance-attribute

algorithm: _Algorithm

Signature algorithm this key verifies, such as RS256.

key instance-attribute

key: SecretBytes

The key material: a PEM public key, an HS* secret, or one JWK written as JSON.

kid class-attribute instance-attribute

kid: str | None = None

Key id this key answers to. None serves tokens with no kid.

format class-attribute instance-attribute

format: Literal['jwk', 'pem', 'secret'] | None = None

How key is written. None reads a secret for an HS* algorithm and a PEM for every other one.

pem classmethod

pem(
    key: bytes | str,
    *,
    algorithm: _AsymmetricAlgorithm,
    kid: str | None = None,
) -> Self

Build a key from a PEM public key.

PARAMETER	DESCRIPTION
key	The PEM public key. TYPE: bytes | str
algorithm	Signature algorithm this key verifies, such as RS256. TYPE: _AsymmetricAlgorithm
kid	Key id this key answers to. None serves tokens with none. TYPE: str | None DEFAULT: None

secret classmethod

secret(
    key: bytes | str,
    *,
    algorithm: _SecretAlgorithm,
    kid: str | None = None,
) -> Self

Build a key from a shared secret, for tokens your own service signs.

The secret must be at least as long as the algorithm's hash: 32 bytes for HS256, 48 for HS384 and 64 for HS512, as RFC 7518 requires. Load it from a secret store or a mounted file, never from a document that is published, which is why a JWKS never supplies one.

PARAMETER	DESCRIPTION
key	The shared secret. TYPE: bytes | str
algorithm	HS256, HS384 or HS512. TYPE: _SecretAlgorithm
kid	Key id this key answers to. None serves tokens with none. TYPE: str | None DEFAULT: None

jwk classmethod

jwk(
    jwk: Mapping[str, Any],
    *,
    algorithm: _AsymmetricAlgorithm | None = None,
) -> Self

Build a key from one JWK, as published at a JWKS endpoint.

The kid and alg are read from the JWK. A provider that publishes no alg, as Entra ID does, needs one here or gets the algorithm its key type implies. A symmetric JWK is refused, because a published key cannot be a shared secret.

PARAMETER	DESCRIPTION
jwk	One key from a JWKS document. TYPE: Mapping[str, Any]
algorithm	Algorithm to pin when the JWK names none. TYPE: _AsymmetricAlgorithm | None DEFAULT: None

JWTClaims dataclass

JWTClaims(
    claims: Mapping[str, Any],
    subject: str | None,
    issuer: str | None,
    audience: str | tuple[str, ...] | None,
    expires_at: int | None,
    issued_at: int | None,
    token_id: str | None,
    scopes: frozenset[str] = frozenset(),
)

The claims of a verified token.

claims holds every claim as it arrived. The named fields are the registered claims lifted out of it, so the common path needs no dict lookups. It satisfies Principal, and answers what Starlette reads off request.user, so it can stand for the caller in a request scope.

claims instance-attribute

claims: Mapping[str, Any]

Every claim the token carries, read-only all the way down: an object is a read-only mapping and an array a tuple. A verified claim set is shared by every request presenting that token while it is cached, so writing into it would change what a later request is authorized as.

subject instance-attribute

subject: str | None

The sub claim.

issuer instance-attribute

issuer: str | None

The iss claim.

audience instance-attribute

audience: str | tuple[str, ...] | None

The aud claim.

expires_at instance-attribute

expires_at: int | None

The exp claim, in whole seconds, a fraction rounded down.

issued_at instance-attribute

issued_at: int | None

The iat claim, in whole seconds, a fraction rounded down.

token_id instance-attribute

token_id: str | None

The jti claim.

scopes class-attribute instance-attribute

scopes: frozenset[str] = frozenset()

What the token grants, read from the policy's scope_claims.

is_authenticated property

is_authenticated: bool

Always true: a verified token authenticates its caller.

identity property

identity: str

The subject, or an empty string for a token that names none.

display_name property

display_name: str

The subject, which is what a token names its caller by.

Principal

Bases: Protocol

The party a request was authenticated as, whatever proved it.

A handler reads this rather than the credential behind it, so the same handler serves a bearer token today and another scheme later. JWTClaims satisfies it.

Key a caller by issuer and subject together. A subject is unique only within the issuer that assigned it, and an email or a username can be reassigned to somebody else.

subject property

subject: str | None

Who the caller is, as the issuer names them.

issuer property

issuer: str | None

Who vouched for the caller.

scopes property

scopes: frozenset[str]

What the caller was granted.

claims property

claims: Mapping[str, Any]

Everything the credential says about the caller, read-only.

is_authenticated property

is_authenticated: bool

Whether the caller was authenticated at all.

TokenRejectedError

TokenRejectedError(
    reason: TokenRejectedReason | str,
    *,
    subject: str | None = None,
)

Bases: GrelmicroError, ValueError

A token failed verification.

reason names what failed, so a caller branches on it rather than on message text. Neither the reason nor the message quotes the token: it is a live credential and the message reaches logs and error responses.

Initialize the error.

PARAMETER	DESCRIPTION
reason	What failed. A tag this module does not know reads as invalid. TYPE: TokenRejectedReason | str
subject	The sub of the token, only once its signature verified. TYPE: str | None DEFAULT: None

resource_metadata class-attribute instance-attribute

resource_metadata: str | None = None

The URL of the protected resource metadata its challenge points at.

Set by AuthenticatedRequestsMiddleware on a refusal a route raised, when the service publishes the metadata, so the challenge names it wherever the refusal is rendered.

subject class-attribute instance-attribute

subject: str | None = subject

The sub of a token whose signature verified before a claim failed.

Set for a token that is expired, not yet valid, issued by or for someone else, missing a claim, or bound to a key. None when the token failed before or at its signature, or carries no non-empty string sub, so a claim from a token nobody vouched for never reaches it.

reason instance-attribute

reason: TokenRejectedReason = known

TokenRejectedReason

Bases: StrEnum

Why a token was rejected.

A stable tag a caller branches on rather than message text. Each member compares equal to the string it names, so error.reason == "expired" reads the same as error.reason is TokenRejectedReason.EXPIRED.

ALGORITHM class-attribute instance-attribute

ALGORITHM = 'algorithm'

The token asks for an algorithm its key does not verify.

AUDIENCE class-attribute instance-attribute

AUDIENCE = 'audience'

The token was issued for another audience.

BINDING class-attribute instance-attribute

BINDING = 'binding'

The token is bound to a key, and no proof of possession is checked.

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

The token is past its exp.

INVALID class-attribute instance-attribute

INVALID = 'invalid'

The token failed for a reason with no tag of its own.

ISSUER class-attribute instance-attribute

ISSUER = 'issuer'

The token was issued by another issuer.

MALFORMED class-attribute instance-attribute

MALFORMED = 'malformed'

The token is not a well-formed JWS.

MISSING_CLAIM class-attribute instance-attribute

MISSING_CLAIM = 'missing-claim'

The token lacks a claim the policy requires.

NOT_YET_VALID class-attribute instance-attribute

NOT_YET_VALID = 'not-yet-valid'

The token is before its nbf, or claims an iat still to come.

REVOKED class-attribute instance-attribute

REVOKED = 'revoked'

The token verified, and the service no longer accepts it.

SCHEME class-attribute instance-attribute

SCHEME = 'scheme'

The Authorization header carries no bearer token.

SIGNATURE class-attribute instance-attribute

SIGNATURE = 'signature'

The signature does not match the key.

TYPE class-attribute instance-attribute

TYPE = 'type'

The token declares a type other than an access token.

UNKNOWN_KEY class-attribute instance-attribute

UNKNOWN_KEY = 'unknown-key'

No loaded key matches the token's kid.

DiscoveryConfig

Bases: JWTPolicy, _KeyPublishing

An issuer whose metadata says where its keys are published.

The issuer is the one URL to configure. Its RFC 8414 authorization server metadata is read first, then its OpenID Connect discovery document, and the first one that answers names the JWKS endpoint. That document must name the issuer exactly, and every token must carry it in iss.

JWKSConfig

Bases: JWTPolicy, _KeyPublishing

Where the keys are published, and the claim policy they enforce.

Carries every JWTPolicy setting, so a verifier fed from a JWKS endpoint checks exactly what one built from a PEM checks.

url instance-attribute

url: str

The JWKS endpoint. Must be https, because the keys it serves decide who is believed.

SigningKeysUnavailableError

Bases: GrelmicroError, RuntimeError

The signing keys could not be loaded.

Raised by refresh when a document cannot be fetched or read, and by verify before any key set has loaded. A refresh that fails keeps the keys already loaded, so a provider that goes down does not take authentication down with it until those keys expire on its side.

Fetcher

Bases: Protocol

Fetches a document a verifier loads its keys from.

Supply your own to reuse a client that already carries your proxy settings, certificate authority, or mutual TLS identity, and to keep the request inside whatever tracing and retry policy that client has.

ClientBans

ClientBans(
    *,
    failures: int | None = None,
    window: int | timedelta | None = None,
    duration: int | timedelta | None = None,
    max_clients: int | None = None,
    reasons: frozenset[str] | None = None,
    name: str = "default",
    env_load: bool | None = None,
)

Bases: Reconfigurable[ClientBansConfig]

Tracks failing clients and refuses the ones that keep failing.

banned is the only call on the request path and does one dictionary lookup. record runs when a token was already refused, so it is never on the path of a request that succeeds.

Built with keywords, it also reads GREL_CLIENTBANS_, or GREL_CLIENTBANS_{NAME}_ for a named table, unless GREL_ENV_LOAD=0. Every setting can change while the service runs: a ban only ever costs capacity, never trust.

Example

bans = ClientBans()

if bans.banned(client_ip):
    raise ClientBannedError(retry_after=bans.banned_for(client_ip))

try:
    claims = verifier.verify_header(authorization)
except TokenRejectedError as error:
    bans.record(client_ip, error.reason)
    raise

Initialize the table, which starts empty.

A setting left out is read from the environment, then takes the default ClientBansConfig gives it.

PARAMETER	DESCRIPTION
failures	Failures inside window before the client is banned. TYPE: int | None DEFAULT: None
window	How long failures are counted, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
duration	How long a ban lasts, in whole seconds or as a timedelta. A float is refused. Keep it short: addresses are shared. TYPE: int | timedelta | None DEFAULT: None
max_clients	Addresses tracked at once. TYPE: int | None DEFAULT: None
reasons	Rejection reasons that count. Defaults to ABUSIVE_REASONS. TYPE: frozenset[str] | None DEFAULT: None
name	Instance name, which is the environment namespace: GREL_CLIENTBANS_{NAME}_. The default instance reads GREL_CLIENTBANS_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused.

name property

name: str

Return the instance name.

from_config classmethod

from_config(
    config: ClientBansConfig,
    *,
    reasons: frozenset[str] | None = None,
    name: str = "default",
) -> Self

Build the table from a configuration that is already whole.

The one declarative door. What you pass is what runs: no environment variable is read, and the table is not registered for live reload.

PARAMETER	DESCRIPTION
config	When to ban, and for how long. TYPE: ClientBansConfig
reasons	Rejection reasons that count. Defaults to ABUSIVE_REASONS. TYPE: frozenset[str] | None DEFAULT: None
name	Instance name, carried by the ban records and metrics. TYPE: str DEFAULT: 'default'

active

active() -> int

Return how many clients are banned right now.

banned

banned(client: str) -> bool

Whether client is currently refused.

One dictionary lookup, and nothing is written, so an honest request pays this and no more.

PARAMETER	DESCRIPTION
client	The address a trusted proxy vouched for, never a raw header. TYPE: str

banned_for

banned_for(client: str) -> float

Return the seconds client stays refused, 0.0 when it is not.

Read once banned has said yes, to tell the caller when to come back, so an honest request never pays for it.

PARAMETER	DESCRIPTION
client	The address a trusted proxy vouched for, never a raw header. TYPE: str

record

record(client: str, reason: str) -> bool

Count a rejection, and return whether it banned the client.

A reason outside the configured set is not counted at all, so a key rotation or a batch of expired tokens never bans anyone.

A ban already running is never shortened. The counting window is shorter than a ban, so a client that keeps failing rolls its window over while still banned, and taking the new count at face value would let it clear its own ban by carrying on. forget is what lifts a ban.

A ban that starts writes one record on grelmicro.security.events and counts grelmicro.client_bans.started. A ban extended while it runs writes nothing.

PARAMETER	DESCRIPTION
client	The address the failure came from. TYPE: str
reason	The TokenRejectedError reason. TYPE: str

forget

forget(client: str) -> None

Drop everything remembered about client, ban included.

PARAMETER	DESCRIPTION
client	The address to clear. TYPE: str

ClientBansConfig

Bases: BaseModel

When a client is refused, and for how long.

failures class-attribute instance-attribute

failures: int = 10

Failures inside window before the client is banned.

window class-attribute instance-attribute

window: Duration = timedelta(minutes=1)

How long failures are counted, in whole seconds or as a timedelta. A float is refused. From text, such as an environment variable, it reads whole seconds ("60") or an ISO 8601 duration ("PT0.5S").

duration class-attribute instance-attribute

duration: Duration = timedelta(minutes=5)

How long a ban lasts, in whole seconds or as a timedelta. A float is refused. Keep it short. Addresses are shared behind NAT, so a ban reaches more people than the one caller that earned it.

max_clients class-attribute instance-attribute

max_clients: int = 10000

Addresses tracked at once. Reached, the least recently recorded are dropped, so failing from many addresses costs an attacker the bans they had rather than the service its memory.

ClientBannedError

ClientBannedError(*, retry_after: float = 0.0)

Bases: AdmissionError, RuntimeError

The caller is banned, so its token was not looked at.

Distinct from TokenRejectedError because it says nothing about the token. Answer it with 429, not 401: the caller is being refused for what it did before this request, and a fresh token would not change it. retry_after says how long the ban has left.

Initialize the error.

PARAMETER	DESCRIPTION
retry_after	Seconds the ban has left, 0.0 when not known. TYPE: float DEFAULT: 0.0

retry_after instance-attribute

retry_after = retry_after

OAuthClient

OAuthClient(*args: object, **kwargs: object)

Bases: Reconfigurable[OAuthClientConfig]

A service registered with its authorization server, for outbound tokens.

Build one with a factory: OAuthClient.discover for an issuer that publishes its metadata, OAuthClient.endpoint for a token endpoint, or OAuthClient.from_config. Register it in Grelmicro(uses=[...]), and ClientCredentials and TokenExchange find it through the app, the way Lock finds its backend.

discover and endpoint also read the environment unless GREL_ENV_LOAD=0, under GREL_OAUTHCLIENT_, or GREL_OAUTHCLIENT_{NAME}_ for a named client, and a keyword always wins. The environment supplies the issuer, the client ID, a secret or an assertion file path. It never chooses how the service authenticates.

Example

micro = Grelmicro(
    uses=[
        OAuthClient.discover(
            "https://auth.example.com/",
            client_id="orders-api",
            client_auth=ClientAuth.secret(),
        ),
    ]
)

Refuse construction: a client has no default authorization server.

kind class-attribute

kind: str = 'oauthclient'

The component kind patterns resolve the client under.

name property

name: str

The name the client is registered under.

config property

config: OAuthClientConfig

The configuration in effect.

token_endpoint property

token_endpoint: str | None

The token endpoint, once known. None until discovery succeeds.

discover classmethod

discover(
    issuer: str | None = None,
    *,
    client_auth: ClientAuth,
    client_id: str | None = None,
    refresh_before: int | timedelta | None = None,
    default_lifetime: int | timedelta | None = None,
    timeout: float | None = None,
    retry_interval: float | None = None,
    name: str = "default",
    env_load: bool | None = None,
) -> Self

Build a client from an issuer, reading its token endpoint from its metadata.

Nothing is fetched here. The metadata is read when the client opens. The issuer's RFC 8414 metadata is read first, then its OpenID Connect discovery document, and the one that answers must name the issuer exactly and an https token endpoint.

PARAMETER	DESCRIPTION
issuer	The issuer, exactly as its metadata names it. Must be https. Left out, it is read from the environment. TYPE: str | None DEFAULT: None
client_auth	How the service proves who it is. TYPE: ClientAuth
client_id	The client ID. Left out, it is read from the environment, where it is required. TYPE: str | None DEFAULT: None
refresh_before	A token is refreshed at least this long before it expires, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
default_lifetime	How long a token lives when its response omits expires_in, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
timeout	Seconds to wait for the authorization server. TYPE: float | None DEFAULT: None
retry_interval	Seconds a failed token request is remembered for. TYPE: float | None DEFAULT: None
name	Registration name, and the environment namespace: GREL_OAUTHCLIENT_{NAME}_. The default client reads GREL_OAUTHCLIENT_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused, no issuer is named, or the environment names what client_auth does not use.

endpoint classmethod

endpoint(
    token_endpoint: str | None = None,
    *,
    client_auth: ClientAuth,
    client_id: str | None = None,
    issuer: str | None = None,
    refresh_before: int | timedelta | None = None,
    default_lifetime: int | timedelta | None = None,
    timeout: float | None = None,
    retry_interval: float | None = None,
    name: str = "default",
    env_load: bool | None = None,
) -> Self

Build a client from a token endpoint, for a server that publishes no metadata.

With no metadata, ClientAuth.secret() sends the secret in Authorization: Basic unless method= says otherwise.

PARAMETER	DESCRIPTION
token_endpoint	The token endpoint. Must be https. Left out, it is read from the environment. TYPE: str | None DEFAULT: None
client_auth	How the service proves who it is. TYPE: ClientAuth
client_id	The client ID. Left out, it is read from the environment, where it is required. TYPE: str | None DEFAULT: None
issuer	The issuer, which a signed assertion names as its audience. Only code passes it here. TYPE: str | None DEFAULT: None
refresh_before	A token is refreshed at least this long before it expires, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
default_lifetime	How long a token lives when its response omits expires_in, in whole seconds or as a timedelta. A float is refused. TYPE: int | timedelta | None DEFAULT: None
timeout	Seconds to wait for the authorization server. TYPE: float | None DEFAULT: None
retry_interval	Seconds a failed token request is remembered for. TYPE: float | None DEFAULT: None
name	Registration name, and the environment namespace: GREL_OAUTHCLIENT_{NAME}_. TYPE: str DEFAULT: 'default'
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused, no endpoint is named, a variable names the issuer, or the environment names what client_auth does not use.

from_config classmethod

from_config(
    config: OAuthClientConfig,
    *,
    client_auth: ClientAuth,
    name: str = "default",
) -> Self

Build a client from a configuration that is already whole.

What you pass is what runs: no environment variable is read, and the client is not registered for live reload.

A secret or a path client_auth carries is used, and one set both there and in config is refused rather than one silently winning.

PARAMETER	DESCRIPTION
config	The client, taken as it is. It discovers when it names an issuer and no token endpoint. TYPE: OAuthClientConfig
client_auth	How the service proves who it is. TYPE: ClientAuth
name	Registration name. TYPE: str DEFAULT: 'default'

RAISES	DESCRIPTION
SettingsValidationError	If config holds what client_auth does not use, lacks what it needs, or holds a secret or path client_auth carries too.
TypeError	If client_auth is not a ClientAuth.

OAuthClientConfig

Bases: BaseModel

Where the authorization server is, who the service is, and how tokens are kept.

client_id instance-attribute

client_id: str

The client ID the authorization server knows the service by.

issuer class-attribute instance-attribute

issuer: str | None = None

The issuer, exactly as its metadata names it. Must be https. Discovery reads the token endpoint from its metadata, and a signed assertion names it as its audience.

token_endpoint class-attribute instance-attribute

token_endpoint: str | None = None

The token endpoint, for a server that publishes no metadata. Must be https.

client_secret class-attribute instance-attribute

client_secret: SecretStr | None = None

The client secret, for ClientAuth.secret.

assertion_file class-attribute instance-attribute

assertion_file: str | None = None

The assertion file, for ClientAuth.assertion_file.

refresh_before class-attribute instance-attribute

refresh_before: Duration = timedelta(minutes=1)

A token is refreshed at least this long before it expires, capped at half its lifetime, in whole seconds or as a timedelta. A float is refused.

default_lifetime class-attribute instance-attribute

default_lifetime: Duration = timedelta(minutes=5)

How long a token lives when its response omits expires_in, in whole seconds or as a timedelta. A float is refused.

timeout class-attribute instance-attribute

timeout: float = 5.0

Seconds to wait for the authorization server.

retry_interval class-attribute instance-attribute

retry_interval: float = 5.0

Seconds a failed token request is remembered for.

max_bytes class-attribute instance-attribute

max_bytes: int = 1 << 20

Largest response accepted from the authorization server.

ClientAuth

ClientAuth(*args: object, **kwargs: object)

How a service proves who it is to its authorization server.

Build one with a factory: ClientAuth.secret, ClientAuth.private_key or ClientAuth.assertion_file. There is no bare constructor, because no method is a safe default. Code chooses the method, and the environment only ever supplies a secret or a path.

Example

OAuthClient.discover(
    "https://auth.example.com/",
    client_id="orders-api",
    client_auth=ClientAuth.secret(),
)

Refuse construction: a client has no default way to authenticate.

secret classmethod

secret(
    secret: str | SecretStr | None = None,
    *,
    method: Literal["basic", "post"] | None = None,
) -> Self

Authenticate with a client secret.

PARAMETER	DESCRIPTION
secret	The client secret. Left out, it is read from GREL_OAUTHCLIENT_CLIENT_SECRET, or GREL_OAUTHCLIENT_{NAME}_CLIENT_SECRET for a named client. TYPE: str | SecretStr | None DEFAULT: None
method	Pins how the secret is sent: basic in Authorization: Basic, post in the request body. Left out, the server's metadata decides, and Basic is used when it lists both or says nothing. TYPE: Literal['basic', 'post'] | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If method is neither basic nor post.

private_key classmethod

private_key(
    key: bytes | str,
    *,
    algorithm: SigningAlgorithm,
    kid: str | None = None,
    certificate: bytes | str | None = None,
    audience: Literal[
        "issuer", "token_endpoint"
    ] = "issuer",
) -> Self

Authenticate with an assertion signed by a private key, RFC 7523.

The key is read once, here, so a key that cannot sign under algorithm is refused at startup rather than on the first fetch. An encrypted key is refused. Every fetch signs a fresh assertion, valid for one minute, with a new jti.

PARAMETER	DESCRIPTION
key	The private key, as PEM. RSA keys are read from PKCS #8 or PKCS #1, EC keys from PKCS #8 or SEC 1, Ed25519 keys from PKCS #8. Only code passes it. TYPE: bytes | str
algorithm	The algorithm the assertion is signed under. TYPE: SigningAlgorithm
kid	The key ID the server knows the key by, sent as kid. TYPE: str | None DEFAULT: None
certificate	The certificate of the key, as PEM. Its SHA-256 thumbprint is sent as x5t#S256, which is how Microsoft Entra ID finds the key. TYPE: bytes | str | None DEFAULT: None
audience	What the assertion names as its audience. issuer, the default, is what RFC 7523bis requires. token_endpoint is what Okta and Microsoft Entra ID still require. TYPE: Literal['issuer', 'token_endpoint'] DEFAULT: 'issuer'

RAISES	DESCRIPTION
SettingsValidationError	If the key or the certificate cannot be read, or the key cannot sign under algorithm.
DependencyNotFoundError	If grelmicro[jwt] is not installed.

assertion_file classmethod

assertion_file(
    path: str | PathLike[str] | None = None,
) -> Self

Authenticate with a signed assertion read from a file on every fetch.

The file is read again for every token request, so an assertion the platform rotates in place is always current.

PARAMETER	DESCRIPTION
path	The file holding the signed assertion, such as a projected service account token. Left out, it is read from GREL_OAUTHCLIENT_ASSERTION_FILE. TYPE: str | PathLike[str] | None DEFAULT: None

ClientCredentials

ClientCredentials(
    name: str,
    *,
    audience: str | None = None,
    resource: str | None = None,
    scopes: str | Sequence[str] | None = None,
    client: OAuthClient | str | None = None,
    env_load: bool | None = None,
)

Bases: _TokenPattern[ClientCredentialsConfig]

A token for the service itself, to call another API.

The name is the API it calls, and its environment namespace. It finds the registered OAuthClient through the app, or takes client=. The token is fetched on first use, cached, and refreshed shortly before it expires, however many requests use it.

Example

payments_token = ClientCredentials("payments-api", audience="payments-api")
payments = httpx.AsyncClient(
    base_url="https://payments.internal", auth=payments_token.auth()
)

Name the API, and how the token for it is requested.

PARAMETER	DESCRIPTION
name	The API the token is for, and the environment namespace: GREL_CLIENTCREDENTIALS_{NAME}_. TYPE: str
audience	Sent as audience. TYPE: str | None DEFAULT: None
resource	Sent as resource, RFC 8707. TYPE: str | None DEFAULT: None
scopes	Sent as scope. TYPE: str | Sequence[str] | None DEFAULT: None
client	The client to use: an OAuthClient, or the name of a registered one. Left out, the registered default. TYPE: OAuthClient | str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused.

config property

config: ClientCredentialsConfig

The target in effect.

from_config classmethod

from_config(
    name: str,
    config: ClientCredentialsConfig,
    *,
    client: OAuthClient | str | None = None,
) -> Self

Build from a configuration that is already whole, reading no variable.

PARAMETER	DESCRIPTION
name	The API the token is for. TYPE: str
config	The target, taken as it is. TYPE: ClientCredentialsConfig
client	The client to use. TYPE: OAuthClient | str | None DEFAULT: None

token async

token() -> AccessToken

Return a valid token for the API, fetching one when needed.

RAISES	DESCRIPTION
OutOfContextError	If no client is open to fetch it with.
TokenUnavailableError	If no valid token is cached and none can be fetched.

auth

auth() -> Any

Return an httpx.Auth that sends the token with every request.

It works with httpx.AsyncClient, from httpx or httpx2. On a 401 refusing the token, it drops the token, and sends a request that is safe to repeat again once with a new one.

RAISES	DESCRIPTION
DependencyNotFoundError	If neither httpx nor httpx2 is installed.

ClientCredentialsConfig

Bases: _TargetConfig

The API a ClientCredentials requests its token for.

TokenExchange

TokenExchange(
    name: str,
    *,
    audience: str | None = None,
    resource: str | None = None,
    scopes: str | Sequence[str] | None = None,
    cache_size: int | None = None,
    client: OAuthClient | str | None = None,
    env_load: bool | None = None,
)

Bases: _TokenPattern[TokenExchangeConfig]

A token for another API, exchanged for the token a request presented.

The issued token still names the user as its subject. It follows RFC 8693 token exchange, and TokenExchange.on_behalf_of uses the on-behalf-of grant instead, which Microsoft Entra ID uses.

Only a VerifiedToken, which CurrentToken gives, can be exchanged. An exchanged token is cached per caller and never outlives the caller's own token. The cache keys on a SHA-256 digest of it, never on the token.

Example

payments_for_user = TokenExchange("payments-api", audience="payments-api")


@app.post("/orders")
async def create_order(token: CurrentToken) -> dict[str, str]:
    response = await payments.post(
        "/charges", auth=payments_for_user.auth(token)
    )
    return response.json()

Name the API, and exchange for it with RFC 8693 token exchange.

PARAMETER	DESCRIPTION
name	The API the token is for, and the environment namespace: GREL_TOKENEXCHANGE_{NAME}_. TYPE: str
audience	Sent as audience. TYPE: str | None DEFAULT: None
resource	Sent as resource, RFC 8707. TYPE: str | None DEFAULT: None
scopes	Sent as scope. TYPE: str | Sequence[str] | None DEFAULT: None
cache_size	Exchanged tokens held in memory. TYPE: int | None DEFAULT: None
client	The client to use: an OAuthClient, or the name of a registered one. Left out, the registered default. TYPE: OAuthClient | str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused.

config property

config: TokenExchangeConfig

The target in effect.

on_behalf_of classmethod

on_behalf_of(
    name: str,
    *,
    scopes: str | Sequence[str] | None = None,
    cache_size: int | None = None,
    client: OAuthClient | str | None = None,
    env_load: bool | None = None,
) -> Self

Exchange with the on-behalf-of grant, which names the API by scopes.

PARAMETER	DESCRIPTION
name	The API the token is for, and the environment namespace: GREL_TOKENEXCHANGE_{NAME}_. TYPE: str
scopes	Sent as scope, such as api://payments/.default. TYPE: str | Sequence[str] | None DEFAULT: None
cache_size	Exchanged tokens held in memory. TYPE: int | None DEFAULT: None
client	The client to use. TYPE: OAuthClient | str | None DEFAULT: None
env_load	Whether to read environment variables. None follows the process-wide GREL_ENV_LOAD flag. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If a setting is refused, or a variable names an audience or a resource, which this grant does not send.

from_config classmethod

from_config(
    name: str,
    config: TokenExchangeConfig,
    *,
    grant: Literal[
        "token_exchange", "on_behalf_of"
    ] = "token_exchange",
    client: OAuthClient | str | None = None,
) -> Self

Build from a configuration that is already whole, reading no variable.

PARAMETER	DESCRIPTION
name	The API the token is for. TYPE: str
config	The target, taken as it is. TYPE: TokenExchangeConfig
grant	The grant to exchange with. TYPE: Literal['token_exchange', 'on_behalf_of'] DEFAULT: 'token_exchange'
client	The client to use. TYPE: OAuthClient | str | None DEFAULT: None

RAISES	DESCRIPTION
SettingsValidationError	If grant is not one this knows, or the on-behalf-of grant is given an audience or a resource.

token async

token(token: VerifiedToken) -> AccessToken

Return a token for the API, exchanged for the caller's token.

PARAMETER	DESCRIPTION
token	The caller's verified token, as CurrentToken gives it. TYPE: VerifiedToken

RAISES	DESCRIPTION
TypeError	If token is not a VerifiedToken.
OutOfContextError	If no client is open to exchange it with.
TokenUnavailableError	If no valid token is cached and the exchange fails or is refused.

auth

auth(token: VerifiedToken) -> Any

Return an httpx.Auth that sends a token exchanged for the caller's.

Pass it per request, client.post(..., auth=exchange.auth(token)), since each request carries its own caller.

PARAMETER	DESCRIPTION
token	The caller's verified token, as CurrentToken gives it. TYPE: VerifiedToken

RAISES	DESCRIPTION
TypeError	If token is not a VerifiedToken.
DependencyNotFoundError	If neither httpx nor httpx2 is installed.

TokenExchangeConfig

Bases: _TargetConfig

The API a TokenExchange exchanges for, and how many tokens it keeps.

cache_size class-attribute instance-attribute

cache_size: int = 1024

Exchanged tokens held in memory. Zero turns the cache off.

AccessToken dataclass

AccessToken(
    value: str,
    token_type: str,
    expires_at: int,
    scopes: frozenset[str] = frozenset(),
)

A token issued for an outbound call.

auth() is how it should reach a request. Read one with token() to send it through another client. Its repr never shows the token.

value class-attribute instance-attribute

value: str = field(repr=False)

The token.

token_type instance-attribute

token_type: str

The scheme the token is sent under, such as Bearer.

expires_at instance-attribute

expires_at: int

When the token expires, in seconds since the epoch.

scopes class-attribute instance-attribute

scopes: frozenset[str] = frozenset()

What the server granted. Empty when the response names no scope.

VerifiedToken

VerifiedToken(*args: object, **kwargs: object)

The bearer token a request presented, once it verified.

AuthenticatedRequests builds one for every request it authenticates, and a handler reads it through CurrentToken to act for the caller, such as exchanging it for a token issued to another API. It cannot be built from a string, so a token that never verified cannot pass for one.

Its repr never shows the token.

Refuse construction: only a token that verified becomes one.

value instance-attribute

value: str

The encoded token, exactly as the request presented it.

expires_at instance-attribute

expires_at: int | None

When the token expires, in seconds since the epoch, or None when its verifier does not say.

TokenUnavailableError

TokenUnavailableError(
    message: str,
    *,
    error: str | None = None,
    description: str | None = None,
)

Bases: GrelmicroError, RuntimeError

No valid token could be had for an outbound call.

Raised by token(), and by a request sent through auth(), once no cached token is left and fetching one failed or was refused. A failure is remembered for a while, so requests arriving in that time raise at once.

error holds the code the authorization server sent, such as invalid_grant, or None when it sent none. description holds its description. It is kept here and never written to a log or to the message, because a server can echo input into it.

Hold the message, and what the authorization server sent.

error instance-attribute

error = error

The error code the authorization server sent, or None.

description instance-attribute

description = description

The server's description of the error, or None.

ClientRejectedError

ClientRejectedError(
    message: str,
    *,
    error: str | None = None,
    description: str | None = None,
)

Bases: TokenUnavailableError

The authorization server refused the service itself.

A wrong secret, a key it does not know, or a client not allowed this grant. Waiting does not fix it. It is also written on the grelmicro.security.events logger, because a client the server stops accepting is often a rotated or leaked credential.

resolve_client_address

resolve_client_address(
    scope: Scope, trusted: TrustedProxies
) -> ClientAddress | None

Resolve the client address a trusted proxy vouched for.

Returns None when the transport peer is absent or unparsable, which a caller must handle rather than being handed a fabricated address.

The header is read only when the peer itself is trusted, and the walk runs right to left, because the rightmost entries are the ones your own proxies wrote. The first entry that is not a trusted proxy is the client. A malformed entry stops the walk rather than being skipped, since skipping lets padding shift which entry is chosen.

When every entry is trusted, the peer is returned with CHAIN_EXHAUSTED rather than the leftmost entry. No trusted proxy vouched for that entry against an untrusted peer, so it is exactly the value that must not become a rate limiter key. The same walk ending inside a header the max_entries cap truncated returns TOO_MANY_ENTRIES, since the entries beyond the cap were never read.

An untrusted peer that sends a non-empty X-Forwarded-For is logged on the grelmicro.security.clientip logger, once per peer and for at most eight peers. A caller probing the header therefore cannot flood the log, nor take the report a misconfigured proxy of yours needs. Nothing is logged when the trusted set is empty, which is the deployment that means trust nothing.

PARAMETER	DESCRIPTION
scope	The ASGI scope of the request. TYPE: Scope
trusted	The proxies whose forwarded entries may be believed. TYPE: TrustedProxies

unverified_header

unverified_header(token: str) -> dict[str, Any]

Return the alg and kid of token without checking its signature.

Nothing it returns is trustworthy. It routes a token to the right verifier, it never decides whether a token is valid, and it reads no key.

PARAMETER	DESCRIPTION
token	The encoded JWT. TYPE: str

RAISES	DESCRIPTION
TokenRejectedError	If token is not a well-formed JWS.

fetch_with_httpx async

fetch_with_httpx(
    url: str, *, timeout: float, max_bytes: int
) -> bytes

Fetch a document with httpx, the default fetcher.

Either httpx or httpx2 will do, whichever the application already has, because the ecosystem is split across the two lines.

The body is read in chunks and abandoned the moment it passes max_bytes, rather than trusting the length the server declares. Redirects are not followed: a key set that answers from somewhere else is a key set from somewhere else.

PARAMETER	DESCRIPTION
url	The document to fetch. TYPE: str
timeout	Seconds to wait for the endpoint. TYPE: float
max_bytes	Largest body accepted. TYPE: int

RAISES	DESCRIPTION
SigningKeysUnavailableError	If the endpoint cannot be reached, times out, answers anything but 200, or serves too large a body.