Security
- Start here: Security guide
- Common recipes:
resolve_client_address(request.scope, trusted)returns the address a trusted proxy vouched for.ClientAddressMiddlewareresolves it once per request so every consumer reads one value.JWTVerifier(JWTConfig(...)).verify_header(header)returns the claims of the bearer token a caller presented.
Security.
Checks a service runs on an inbound request. grelmicro validates what arrives, 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.
Read more in the Security docs.
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:
|
max_hops
|
Most trusted proxies to walk past. Bounds a forged chain. None means no cap.
TYPE:
|
max_entries
|
Most header entries to parse before giving up.
TYPE:
|
max_header_bytes
|
Most header bytes to parse before giving up.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
SettingsValidationError
|
If an entry is not an IP address or a
CIDR range, or a bound is outside its usable range.
|
dataclass
ClientAddress(
ip: IPAddress,
port: int | None,
reason: ClientAddressReason,
hops: int,
)
A resolved client address, with how it was reached.
instance-attribute
ip: IPAddress
The address. Always parsed, never raw header text.
instance-attribute
port: int | None
The source port, when one was given.
instance-attribute
reason: ClientAddressReason
Why this address was chosen. RESOLVED is the trusted case.
instance-attribute
hops: int
Trusted proxies skipped before landing on this address.
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.
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.
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.
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.
class-attribute
instance-attribute
RESOLVED = 'resolved'
A trusted proxy vouched for this address.
class-attribute
instance-attribute
NO_FORWARDED_HEADER = 'no-forwarded-header'
A trusted peer sent no X-Forwarded-For.
class-attribute
instance-attribute
UNTRUSTED_PEER = 'untrusted-peer'
The peer is not a trusted proxy, so the header was ignored.
class-attribute
instance-attribute
CHAIN_EXHAUSTED = 'chain-exhausted'
Every entry in the chain was a trusted proxy.
class-attribute
instance-attribute
HOP_LIMIT = 'hop-limit'
The walk passed more trusted proxies than max_hops.
class-attribute
instance-attribute
MALFORMED_ENTRY = 'malformed-entry'
An entry was not an address, which stops the walk.
class-attribute
instance-attribute
HEADER_TOO_LARGE = 'header-too-large'
The header was longer than max_header_bytes.
class-attribute
instance-attribute
TOO_MANY_ENTRIES = 'too-many-entries'
More entries than max_entries, and every one read was trusted.
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:
|
trusted
|
The proxies whose forwarded entries may be believed.
TYPE:
|
overwrite_scope_client
|
Also rewrite 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
TYPE:
|
instance-attribute
app = app
instance-attribute
trusted = trusted
instance-attribute
overwrite_scope_client = overwrite_scope_client
JWTVerifier(
config: JWTConfig, *, bans: ClientBans | None = None
)
Verifies inbound JWTs against a fixed key set and claim policy.
Keys are parsed once at construction. Verification runs in the compiled core with the GIL released, so a thread pool verifies in parallel.
Example
verifier = JWTVerifier(
JWTConfig(
keys=[JWTKey(algorithm="RS256", key=public_pem)],
audience=["grelmicro-api"],
issuer=["https://auth.example.com/"],
)
)
claims = verifier.verify(token)
Initialize the verifier, parsing every key.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
Keys and claim policy to enforce.
TYPE:
|
bans
|
Opt in to refusing callers that keep presenting tokens that do not verify. Pass
TYPE:
|
verify(
token: str, *, client: str | None = None
) -> 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.
With bans set, a caller already banned raises ClientBannedError
before the token is looked at, and a rejection is counted against it.
| PARAMETER | DESCRIPTION |
|---|---|
token
|
The encoded JWT, with no scheme prefix.
TYPE:
|
client
|
The address to hold responsible. Required when
TYPE:
|
verify_header(
header: str | None, *, client: str | None = None
) -> JWTClaims
Return the claims of the bearer token in header.
| PARAMETER | DESCRIPTION |
|---|---|
header
|
The
TYPE:
|
client
|
The address to hold responsible, when banning.
TYPE:
|
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 key set, it never decides whether a token is valid.
| PARAMETER | DESCRIPTION |
|---|---|
token
|
The encoded JWT.
TYPE:
|
Bases: JWTPolicy
A claim policy together with the keys that verify against it.
classmethod
from_jwks(
jwks: Mapping[str, Any],
*,
algorithm: str | None = None,
**policy: Any,
) -> JWTConfig
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:
|
algorithm
|
Algorithm to pin for keys that name none.
TYPE:
|
**policy
|
Any other
TYPE:
|
Example
jwks = httpx.get(f"{issuer}/.well-known/jwks.json").json()
config = JWTConfig.from_jwks(
jwks, audience=["my-api"], issuer=[issuer]
)
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.
class-attribute
instance-attribute
audience: list[str] = Field(default_factory=list)
Accepted aud values. Leaving this empty says the service identifies with no audience, and RFC 7519 then requires refusing any token that carries an aud claim, so set it whenever your tokens have one. A token that carries no aud passes this check: add aud to required to insist on one.
class-attribute
instance-attribute
issuer: list[str] = Field(default_factory=list)
Accepted iss values. Empty leaves the issuer unchecked.
class-attribute
instance-attribute
leeway: int = 0
Seconds of clock skew allowed on exp and nbf.
class-attribute
instance-attribute
required: list[str] = Field(default_factory=lambda: ["exp"])
Further claims that must be present. exp is always required and does not need naming, and naming an audience or an issuer requires that claim too, so this only ever adds to what is enforced.
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.
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.
class-attribute
instance-attribute
cache_ttl: float = 300.0
Seconds a verified token stays cached. 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() -> 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. Leave audience empty for a
provider whose tokens carry none, such as an AWS Cognito access
token.
Bases: BaseModel
One verification key and the algorithm it verifies.
instance-attribute
algorithm: str
Signature algorithm this key verifies, such as RS256.
instance-attribute
key: bytes
PEM public key, or the shared secret for the HS* family.
class-attribute
instance-attribute
kid: str | None = None
Key id this key answers to. None serves tokens with no kid.
class-attribute
instance-attribute
format: str = 'pem'
pem for PEM or an HS* secret, jwk for a JSON Web Key.
classmethod
from_jwk(
jwk: Mapping[str, Any], *, algorithm: str | None = None
) -> JWTKey
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.
| PARAMETER | DESCRIPTION |
|---|---|
jwk
|
One key from a JWKS document.
TYPE:
|
algorithm
|
Algorithm to pin when the JWK names none.
TYPE:
|
dataclass
JWTClaims(
raw: Mapping[str, Any],
subject: str | None,
issuer: str | None,
audience: str | list[str] | None,
expires_at: int | None,
issued_at: int | None,
token_id: str | None,
)
The claims of a verified token.
raw 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.
instance-attribute
raw: Mapping[str, Any]
Every claim the token carries, read-only. 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.
instance-attribute
subject: str | None
The sub claim.
instance-attribute
issuer: str | None
The iss claim.
instance-attribute
audience: str | list[str] | None
The aud claim.
instance-attribute
expires_at: int | None
The exp claim, in seconds.
instance-attribute
issued_at: int | None
The iat claim, in seconds.
instance-attribute
token_id: str | None
The jti claim.
property
scopes: frozenset[str]
The scope claim, split on whitespace.
TokenRejectedError(reason: str)
Bases: GrelmicroError, ValueError
A token failed verification.
reason is a stable tag such as expired or signature, so a caller
branches on it rather than on message text. Neither the tag 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
|
Stable tag naming what failed.
TYPE:
|
instance-attribute
reason = reason
JWKSVerifier(
config: JWKSConfig,
*,
fetch: JWKSFetcher | None = None,
bans: ClientBans | None = None,
)
Verifies tokens against keys fetched from a JWKS endpoint.
refresh is a coroutine, verify is not. Call refresh once before
serving and then on a schedule, so no request ever waits on the provider.
Example
verifier = JWKSVerifier(
JWKSConfig(
url="https://auth.example.com/.well-known/jwks.json",
audience=["my-api"],
issuer=["https://auth.example.com/"],
)
)
await verifier.refresh()
claims = verifier.verify(token)
Initialize the verifier. Keys are loaded by the first refresh.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
Endpoint and claim policy.
TYPE:
|
fetch
|
Fetcher to use. Defaults to one built on
TYPE:
|
bans
|
Opt in to refusing callers that keep presenting tokens that do not verify. Pass
TYPE:
|
property
ready: bool
Whether a key set has been loaded.
property
stale: bool
Whether the next refresh would fetch.
True before the first load, once ttl has passed, and once a token
named a key the current set does not hold.
async
refresh(*, force: bool = False) -> bool
Fetch the key set, and return whether the keys changed.
Does nothing when the current keys are fresh, and never fetches more
often than retry_interval. A failure raises and leaves the loaded
keys in place.
| PARAMETER | DESCRIPTION |
|---|---|
force
|
Fetch even when the current keys are still fresh.
TYPE:
|
verify(
token: str, *, client: str | None = None
) -> JWTClaims
Return the claims of token, or raise TokenRejectedError.
| PARAMETER | DESCRIPTION |
|---|---|
token
|
The encoded JWT, with no scheme prefix.
TYPE:
|
client
|
The address to hold responsible, when banning.
TYPE:
|
verify_header(
header: str | None, *, client: str | None = None
) -> JWTClaims
Return the claims of the bearer token in header.
| PARAMETER | DESCRIPTION |
|---|---|
header
|
The
TYPE:
|
client
|
The address to hold responsible, when banning.
TYPE:
|
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 key set, it never decides whether a token is valid.
| PARAMETER | DESCRIPTION |
|---|---|
token
|
The encoded JWT.
TYPE:
|
Bases: JWTPolicy
Where the keys come from, 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.
instance-attribute
url: str
The JWKS endpoint. Must be https, because the keys it serves decide who is believed.
class-attribute
instance-attribute
ttl: float = 3600.0
Seconds a fetched document is treated as current.
class-attribute
instance-attribute
retry_interval: float = 60.0
Least time between two fetches. This is what stops a caller presenting invented kid values from making the service fetch on demand.
class-attribute
instance-attribute
timeout: float = 5.0
Seconds to wait for the endpoint before giving up.
class-attribute
instance-attribute
max_bytes: int = _ONE_MIB
Largest document accepted, so a hostile endpoint cannot exhaust memory.
class-attribute
instance-attribute
max_keys: int = 32
Most keys accepted from one document.
class-attribute
instance-attribute
algorithm: str | None = None
Algorithm to pin for keys that publish none, as Entra ID does.
Bases: Protocol
Fetches a JWKS document.
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.
Bases: GrelmicroError, RuntimeError
The key set could not be fetched or could not be read.
Raised by refresh, never by verify. A refresh that fails leaves the
keys already loaded in place, so a provider that goes down does not take
authentication down with it until those keys expire on its side.
ClientBans(
config: ClientBansConfig | None = None,
*,
reasons: frozenset[str] | None = None,
)
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.
Example
bans = ClientBans()
if bans.banned(client_ip):
raise HTTPException(status_code=429)
try:
claims = verifier.verify_header(authorization)
except TokenRejectedError as error:
bans.record(client_ip, error.reason)
raise
Initialize the table, which starts empty.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
When to ban, and for how long.
TYPE:
|
reasons
|
Rejection reasons that count. Defaults to
TYPE:
|
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:
|
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.
| PARAMETER | DESCRIPTION |
|---|---|
client
|
The address the failure came from.
TYPE:
|
reason
|
The
TYPE:
|
forget(client: str) -> None
Drop everything remembered about client, ban included.
| PARAMETER | DESCRIPTION |
|---|---|
client
|
The address to clear.
TYPE:
|
Bases: BaseModel
When a client is refused, and for how long.
class-attribute
instance-attribute
failures: int = 10
Failures inside window before the client is banned.
class-attribute
instance-attribute
window: float = 60.0
Seconds over which failures are counted.
class-attribute
instance-attribute
duration: float = 300.0
Seconds a ban lasts. Keep it short. Addresses are shared behind NAT, so a ban reaches more people than the one caller that earned it.
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()
Bases: GrelmicroError, 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.
Initialize the error.
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:
|
trusted
|
The proxies whose forwarded entries may be believed.
TYPE:
|
async
fetch_with_httpx(
url: str, *, timeout: float, max_bytes: int
) -> bytes
Fetch a JWKS 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 JWKS endpoint.
TYPE:
|
timeout
|
Seconds to wait for the endpoint.
TYPE:
|
max_bytes
|
Largest body accepted.
TYPE:
|