API reference¶

Everything below is importable from ryjwt.

Keys¶

ryjwt.SecretKey ¶

SecretKey(
    secret: str | bytes,
    *,
    algorithms: Sequence[HMACAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    allow_short_secret: bool = False,
)

Encodes and decodes JWTs with a shared secret, str or bytes, for the HMAC algorithms (HS256, HS384, HS512).

secret must not be empty, nor look like a public key: a PEM, an SSH key or a JWK (use PrivateKey/PublicKey).

audience and issuer are what decode checks tokens' aud and iss claims against, unless a call passes its own. Each is a str, or an iterable of them, any one of which may match. Without audience, a token that has an aud is rejected. Without issuer, iss isn't checked: set it too, above all if one key signs for several issuers.

secret must be at least as long as the hash's output (RFC 7518 §3.2): 32 bytes for HS256, 48 for HS384, 64 for HS512, the longest of them if several are allowed. A str is measured in UTF-8 bytes. Anyone holding a single token can brute-force a shorter secret offline, then forge tokens; generate one with secrets.token_bytes(32).

allow_short_secret=True skips (only) that length check, e.g. for a secret you don't control.

algorithms `property` ¶

algorithms: list[HMACAlgorithm]

The configured algorithm names.

encode ¶

encode(
    claims: Claims,
    *,
    algorithm: HMACAlgorithm | None = None,
    header: Mapping[str, Any] | None = None,
) -> str

Signs claims (a dict, a msgspec Struct or a pydantic BaseModel) and returns the token.

algorithm must be one of algorithms, and may be left out when only one is configured. header's fields are added to the token's header, which always has alg, and "typ": "JWT" unless header sets typ. Setting alg, crit or b64 in header, or more fields than decode takes (64, with alg and typ), is a ValueError; a kid that isn't a str is a TypeError. A datetime under exp, nbf or iat is written as whole seconds since the epoch; a naive one is a ValueError.

decode ¶

decode(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

decode[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

Verifies token and returns its claims as a dict, or (given into, a msgspec Struct or pydantic BaseModel class) as an instance of into.

Checks the signature, with the algorithm the token's header names: it must be one of algorithms. Then checks the claims:

exp and nbf, if the token has them, allowing leeway seconds for clock differences;
aud against audience. A token with an aud is rejected if there's no audience;
iss against issuer, if there is one.

audience and issuer default to the ones this key was created with. Passing one here replaces the key's for this call. To skip a check the key makes, use another key object.

Every rejection is an InvalidTokenError.

The payload is parsed with msgspec if it was installed when this object was created, else with jiter. They agree on all valid JSON, but may differ on exotic payloads (e.g. 1e400, or nesting over ~200 levels deep); see the documentation's "Encoding and decoding" page.

ryjwt.PrivateKey ¶

PrivateKey(
    pem: str | bytes,
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    kid: str | None = None,
)

Encodes and decodes JWTs with a private key (RS*, PS*, ES*, EdDSA).

pem is a PEM-encoded, unencrypted private key: PKCS#8 (BEGIN PRIVATE KEY), or PKCS#1 / SEC 1 (BEGIN RSA PRIVATE KEY / BEGIN EC PRIVATE KEY). A public key is rejected.

audience and issuer are what decode checks tokens' aud and iss claims against, unless a call passes its own. Each is a str, or an iterable of them, any one of which may match. Without audience, a token that has an aud is rejected. Without issuer, iss isn't checked: set it too, above all if one key signs for several issuers.

kid names the key: encode writes it into every token's header, and jwk exports it. It must be a non-empty str. decode doesn't check tokens' kid against it.

algorithms `property` ¶

algorithms: list[AsymmetricAlgorithm]

The configured algorithm names.

from_path `staticmethod` ¶

from_path(
    path: str | PathLike[str],
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    kid: str | None = None,
) -> PrivateKey

Reads the PEM from the file at path.

OS errors (FileNotFoundError, ...) propagate; unusable contents raise InvalidKeyError.

jwk ¶

jwk() -> dict[str, str]

Returns its public key as a JWK, to publish (see jwks). It never holds any private key material.

Its members, in this order: kty; kid, if the key has one; "use": "sig"; alg, only if one algorithm is configured; crv, for EC and Ed25519 keys; then the public key itself: n and e (RSA), x and y (EC), or x (Ed25519).

encode ¶

encode(
    claims: Claims,
    *,
    algorithm: AsymmetricAlgorithm | None = None,
    header: Mapping[str, Any] | None = None,
) -> str

Signs claims (a dict, a msgspec Struct or a pydantic BaseModel) and returns the token.

algorithm must be one of algorithms, and may be left out when only one is configured. header's fields are added to the token's header, which always has alg, the key's kid if it has one, and "typ": "JWT" unless header sets typ. Setting alg, crit or b64 in header, or kid when the key has one, or more fields than decode takes (64, with alg, typ and the key's kid), is a ValueError; a kid that isn't a str is a TypeError. A datetime under exp, nbf or iat is written as whole seconds since the epoch; a naive one is a ValueError.

decode ¶

decode(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

decode[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

Verifies token and returns its claims as a dict, or (given into, a msgspec Struct or pydantic BaseModel class) as an instance of into.

Checks the signature, with the algorithm the token's header names: it must be one of algorithms. Then checks the claims:

exp and nbf, if the token has them, allowing leeway seconds for clock differences;
aud against audience. A token with an aud is rejected if there's no audience;
iss against issuer, if there is one.

audience and issuer default to the ones this key was created with. Passing one here replaces the key's for this call. To skip a check the key makes, use another key object.

Every rejection is an InvalidTokenError.

The payload is parsed with msgspec if it was installed when this object was created, else with jiter. They agree on all valid JSON, but may differ on exotic payloads (e.g. 1e400, or nesting over ~200 levels deep); see the documentation's "Encoding and decoding" page.

ryjwt.PublicKey ¶

PublicKey(
    pem: str | bytes,
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    kid: str | None = None,
)

Decodes JWTs with a public key or a JWKS' keys (RS*, PS*, ES*, EdDSA). It can't encode.

pem is a PEM-encoded public key: SubjectPublicKeyInfo (BEGIN PUBLIC KEY), or PKCS#1 (BEGIN RSA PUBLIC KEY). A private key is rejected: pass its public key instead.

Built with from_jwks, it may hold several keys, and each token picks one by its kid.

audience and issuer are what decode checks tokens' aud and iss claims against, unless a call passes its own. Each is a str, or an iterable of them, any one of which may match. Without audience, a token that has an aud is rejected. Without issuer, iss isn't checked: set it too, above all if one key signs for several issuers.

kid names the key in what jwk exports. It must be a non-empty str. decode doesn't check tokens' kid against it.

algorithms `property` ¶

algorithms: list[AsymmetricAlgorithm]

The configured algorithm names.

from_path `staticmethod` ¶

from_path(
    path: str | PathLike[str],
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    kid: str | None = None,
) -> PublicKey

Reads the PEM from the file at path.

OS errors (FileNotFoundError, ...) propagate; unusable contents raise InvalidKeyError.

from_jwks `staticmethod` ¶

from_jwks(
    jwks: str | bytes | Mapping[str, Any],
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
) -> PublicKey

Takes the keys of a JWKS document (JSON, or the already-parsed Mapping).

Keys of an unsupported type, and "use": "enc" keys, are ignored; a key with its own alg is only used for that algorithm. Tokens pick their key by kid (which may only be left out if there's a single usable key): an unknown kid raises UnknownKeyError. A document holding private key material, or no usable key, raises InvalidKeyError.

jwk ¶

jwk() -> dict[str, str]

Returns the public key as a JWK, as PrivateKey.jwk does.

Built with from_jwks, it's the document's key, with the document's kid (if any). If it holds several keys, it's a ValueError: export them with jwks([key]).

decode ¶

decode(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

decode[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

Verifies token and returns its claims as a dict, or (given into, a msgspec Struct or pydantic BaseModel class) as an instance of into.

Checks the signature, with the algorithm the token's header names: it must be one of algorithms. Then checks the claims:

exp and nbf, if the token has them, allowing leeway seconds for clock differences;
aud against audience. A token with an aud is rejected if there's no audience;
iss against issuer, if there is one.

audience and issuer default to the ones this key was created with. Passing one here replaces the key's for this call. To skip a check the key makes, use another key object.

Every rejection is an InvalidTokenError.

The payload is parsed with msgspec if it was installed when this object was created, else with jiter. They agree on all valid JSON, but may differ on exotic payloads (e.g. 1e400, or nesting over ~200 levels deep); see the documentation's "Encoding and decoding" page.

ryjwt.jwks ¶

jwks(
    keys: Iterable[PrivateKey | PublicKey],
) -> dict[str, list[dict[str, str]]]

Return a JWKS of the public keys of keys, to publish (e.g. at /.well-known/jwks.json): {"keys": [key.jwk(), ...]}, in order. A PublicKey built with from_jwks adds all its keys.

Its keys must be ones PublicKey.from_jwks can pick from: at least one, and if there are several, each with its own kid. Otherwise it's a ValueError. A SecretKey is a TypeError: its secret must never be published.

JWKS URLs¶

ryjwt.JWKSClient ¶

JWKSClient(
    url: str,
    *,
    algorithms: Sequence[AsymmetricAlgorithm],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    cache_lifetime: float | timedelta = 900,
    min_cache_lifetime: float | timedelta = 60,
    max_cache_lifetime: float | timedelta = 86400,
    max_stale: float | timedelta = 86400,
    cooldown: float | timedelta = 30,
)

Decodes JWTs with the keys published at a JWKS URL, which it fetches and keeps up to date.

Unlike PyJWT's PyJWKClient, which only fetches the signing key, it decodes and verifies tokens itself.

Creating it doesn't fetch anything. Use it in one of two ways:

Automatic: decode (in sync code) or await adecode (in async code). They fetch the keys when needed: the first time, and when a token's kid is unknown. Once the keys expire, they keep using them while new ones are fetched in the background.
Manual: refresh() or await arefresh() fetch the keys when you decide (at startup, or when needs_refresh), and decode_nowait decodes with the keys held, never fetching or waiting.

If a fetch fails, the client keeps using the keys it has, for up to max_stale after they expired, and tries again after cooldown. With no usable keys, decodes raise JWKSFetchError. Keys in a fetched JWKS that can't be used are skipped.

Fetches use the standard library's urllib.request. They check TLS certificates, don't follow redirects, and give up after 2.5 seconds. One fetch runs at a time, on a thread of its own, and every caller that needs it shares it, sync or async, on any thread or event loop.

decode blocks while it waits for a fetch: in async code, use adecode.

url must be https://, or http:// to this machine (localhost, 127.0.0.1 or [::1]). URLs with credentials, whitespace, control or non-ASCII characters, or backslashes are rejected too. A URL that isn't allowed is a ValueError. algorithms are checked as PublicKey.from_jwks checks them.

audience and issuer are what decode, adecode and decode_nowait check tokens' aud and iss against, as for PublicKey. Each call may pass its own instead.

The times are seconds or timedeltas, and mustn't be negative:

cache_lifetime: how long keys stay fresh if the response doesn't say (it has no Cache-Control: max-age).
min_cache_lifetime, max_cache_lifetime: the shortest and longest time keys stay fresh, whatever the response says.
max_stale: how long expired keys stay in use while fetching new ones keeps failing.
cooldown: the least time between fetches, after a failed fetch or for an unknown kid. Keys also stay fresh for at least this long.

needs_refresh `property` ¶

needs_refresh: bool

Whether there are no keys yet, or they've expired: time to call refresh(). Still True during the cooldown after a failed fetch (refresh() then returns at once).

refresh ¶

refresh() -> None

Fetches the keys now, so the client has usable keys, or raises JWKSFetchError.

Joins the fetch in flight, if any. Within cooldown of a failed fetch, doesn't fetch. A failed fetch only raises if the client is left without usable keys: none, or expired over max_stale ago; otherwise the keys held stay in use (and needs_refresh stays True). Waits at most 2.5 seconds.

arefresh `async` ¶

arefresh() -> None

refresh, for async code.

decode ¶

decode(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

decode[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

Verifies token with the JWKS' keys, as PublicKey.decode does, fetching them first if needed (blocking: in async code, use adecode). audience and issuer, if given, replace the client's own for this call.

Raises JWKSFetchError if there are no keys (the fetch failed) or they've been out of date for over max_stale, and UnknownKeyError if the token's kid is in neither the keys held nor (cooldown permitting) refetched ones.

adecode `async` ¶

adecode(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

adecode[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

decode, for async code: waiting for a fetch lets the event loop run. With fresh keys held, it only suspends to wait for a refetch, for a token whose kid is unknown.

decode_nowait ¶

decode_nowait(
    token: str | bytes,
    *,
    into: None = None,
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> dict[str, Any]

decode_nowait[T: StructTyping | BaseModelTyping](
    token: str | bytes,
    *,
    into: type[T],
    audience: str | Iterable[str] | None = None,
    issuer: str | Iterable[str] | None = None,
    leeway: float | timedelta = 0,
) -> T

Verifies token with the keys held, as PublicKey.decode does; never fetches or waits (call refresh() for that).

Raises JWKSFetchError if there are no keys yet, or they've been expired for over max_stale (expired keys are used until then), and UnknownKeyError if the token's kid isn't in the keys held.

Inspecting a token without verifying it¶

ryjwt.unverified_header ¶

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

Return a token's header without verifying its signature; use it for routing or logging, never for trust decisions.

Nothing is checked, but the token must be well formed, as decode requires: three parts of unpadded base64url, with a JSON object in the header and in the payload. If not, it raises DecodeError.

ryjwt.unverified_claims ¶

unverified_claims(token: str | bytes) -> dict[str, Any]

Return a token's claims without verifying its signature or checking exp/nbf/aud/iss; don't trust any value in the result.

The token must still be well formed, as decode requires, or it raises DecodeError.

ryjwt.unverified_token ¶

unverified_token(
    token: str | bytes,
) -> tuple[dict[str, Any], dict[str, Any]]

Return a token's (header, claims) without verifying anything; verify with a key's decode() before trusting either.

The token must still be well formed, as decode requires, or it raises DecodeError.

Algorithms¶

ryjwt.HMACAlgorithm ¶

HMACAlgorithm = Literal['HS256', 'HS384', 'HS512']

The algorithms SecretKey takes.

ryjwt.AsymmetricAlgorithm ¶

AsymmetricAlgorithm = Literal[
    "RS256",
    "RS384",
    "RS512",
    "PS256",
    "PS384",
    "PS512",
    "ES256",
    "ES256K",
    "ES384",
    "ES512",
    "ES521",
    "EdDSA",
]

The algorithms PrivateKey and PublicKey take.

Exceptions¶

ryjwt.RYJWTError ¶

Bases: Exception

Base class for all ryjwt errors.

ryjwt.InvalidKeyError ¶

Bases: RYJWTError

A key can't be used: it doesn't parse, isn't supported, is too weak, or doesn't suit the configured algorithms.

ryjwt.JWKSFetchError ¶

Bases: RYJWTError

A JWKS client has no usable keys. Fetching them failed, and it has none cached, or they've been expired for over max_stale; or decode_nowait was called before the first fetch. It's not the token's fault: respond 503.

ryjwt.InvalidTokenError ¶

Bases: RYJWTError

The token was rejected: respond 401. Each reason has a subclass. This class itself is raised for a header with a crit field, which ryjwt must reject.

ryjwt.DecodeError ¶

Bases: InvalidTokenError

The token is malformed: e.g. not three parts, not valid base64url, or not valid JSON. Also raised when its exp or nbf claim isn't a number.

ryjwt.InvalidSignatureError ¶

Bases: DecodeError

The signature doesn't match.

ryjwt.InvalidAlgorithmError ¶

Bases: InvalidTokenError

The token's algorithm (its header's alg) is missing, or isn't one the key may be used with.

ryjwt.UnknownKeyError ¶

Bases: InvalidTokenError

No key in the JWKS has the token's kid. Also raised when the JWKS has several keys, and the token has no kid (or the keys have none) to pick one with.

ryjwt.ExpiredSignatureError ¶

Bases: InvalidTokenError

The exp claim is in the past (allowing for leeway).

ryjwt.ImmatureSignatureError ¶

Bases: InvalidTokenError

The nbf claim is in the future (allowing for leeway).

ryjwt.InvalidAudienceError ¶

Bases: InvalidTokenError

The token's audience (aud) doesn't match audience, or is missing or malformed. Also raised when the token has an aud but no audience was set, on the key or the call.

ryjwt.InvalidIssuerError ¶

Bases: InvalidTokenError

The token's issuer (iss) doesn't match issuer, or is missing or malformed.

ryjwt.ClaimsValidationError ¶

Bases: InvalidTokenError

The claims don't fit the into class given to decode: a required claim is missing, or one has the wrong type or is out of range. Its __cause__ is msgspec's or pydantic's ValidationError.

API reference¶

Keys¶

ryjwt.SecretKey ¶

algorithms property ¶

encode ¶

decode ¶

ryjwt.PrivateKey ¶

algorithms property ¶

from_path staticmethod ¶

jwk ¶

encode ¶

decode ¶

ryjwt.PublicKey ¶

algorithms property ¶

from_path staticmethod ¶

from_jwks staticmethod ¶

jwk ¶

decode ¶

ryjwt.jwks ¶

JWKS URLs¶

ryjwt.JWKSClient ¶

needs_refresh property ¶

refresh ¶

arefresh async ¶

decode ¶

adecode async ¶

decode_nowait ¶

Inspecting a token without verifying it¶

ryjwt.unverified_header ¶

ryjwt.unverified_claims ¶

ryjwt.unverified_token ¶

Algorithms¶

ryjwt.HMACAlgorithm ¶

ryjwt.AsymmetricAlgorithm ¶

Exceptions¶

ryjwt.RYJWTError ¶

ryjwt.InvalidKeyError ¶

ryjwt.JWKSFetchError ¶

ryjwt.InvalidTokenError ¶

ryjwt.DecodeError ¶

ryjwt.InvalidSignatureError ¶

ryjwt.InvalidAlgorithmError ¶

ryjwt.UnknownKeyError ¶

ryjwt.ExpiredSignatureError ¶

ryjwt.ImmatureSignatureError ¶

ryjwt.InvalidAudienceError ¶

ryjwt.InvalidIssuerError ¶

ryjwt.ClaimsValidationError ¶

algorithms `property` ¶

algorithms `property` ¶

from_path `staticmethod` ¶

algorithms `property` ¶

from_path `staticmethod` ¶

from_jwks `staticmethod` ¶

needs_refresh `property` ¶

arefresh `async` ¶

adecode `async` ¶