"""
JWT Authentication Manager for Kailash Middleware
Provides enterprise-grade JWT authentication with support for both HS256 and RSA algorithms.
This consolidates the functionality of both JWTAuthManager and KailashJWTAuthManager.
"""
import logging
import os
import secrets
import time
import uuid
from datetime import datetime, timedelta, timezone
from functools import lru_cache
from typing import Any, Dict, List, Optional, Union
# JWT and cryptography imports
try:
import jwt
from cryptography.hazmat.primitives.asymmetric import rsa
except ImportError:
jwt = None
rsa = None
from ...utils.secure_logging import sanitize_log_value
from .exceptions import (
InvalidTokenError,
RefreshTokenError,
TokenBlacklistedError,
TokenExpiredError,
)
# Import models and utilities (no circular dependencies)
from .models import JWTConfig, RefreshTokenData, TokenPair, TokenPayload
from .revocation import InMemoryTokenRevocationStore, TokenRevocationStore
from .utils import generate_secret_key, is_token_expired
logger = logging.getLogger(__name__)
#: Environment variable holding the HS256 signing secret. Read when neither
#: ``secret_key=`` nor ``JWTConfig.secret_key`` is supplied, so an operator has
#: a wiring path that needs no code change — the failure mode #2041 documented
#: was an error message naming a variable nothing ever read.
JWT_SECRET_KEY_ENV = "KAILASH_JWT_SECRET_KEY"
#: Shortest secret accepted from :data:`JWT_SECRET_KEY_ENV`. RFC 7518 §3.2
#: requires an HMAC-SHA256 key of at least 256 bits, which is where the 32 comes
#: from; it is the same NUMBER as
#: ``kailash.trust.auth.jwt.JWTConfig.MIN_SECRET_LENGTH``
#: (``src/kailash/trust/auth/jwt.py:151``) and as the
#: ``KAILASH_API_GATEWAY_SECRET`` check
#: (``src/kailash/middleware/communication/api_gateway.py:261-268``).
#:
#: It is NOT the same MEASUREMENT, and calling it a match would be wrong. That
#: gateway check counts UTF-8 BYTES and does not trim; this one counts
#: non-whitespace CHARACTERS. The two disagree on any non-ASCII secret:
#: ``"a" * 20 + "é" * 6`` is 26 characters but 32 bytes, so it PASSES there and
#: FAILS here. Both are deliberate at their own site — the gateway secret feeds
#: an HMAC that consumes bytes, while this one is a human-set passphrase where
#: trailing newlines from a config file are the common failure — and neither is
#: an entropy floor: ``"a" * 32`` clears both.
#:
#: Scoped to the ENVIRONMENT path deliberately. This variable is introduced by
#: the #2083 fix, so validating it breaks nothing that exists — whereas an
#: unvalidated new key surface would hand operators a four-character signing
#: secret as the ergonomic path, which is the failure this fix exists to
#: prevent wearing different clothes. The pre-existing ``secret_key=`` argument
#: is deliberately NOT gated here: that is a separate defect class (weak key,
#: not absent key) with an unbounded set of existing callers, and it is
#: reported rather than silently changed under this issue.
MIN_ENV_SECRET_LENGTH = 32
@lru_cache(maxsize=1)
def _warn_ephemeral_signing_key() -> None:
"""Announce, once per process, that signing keys are ephemeral.
ERROR rather than INFO, and once per process rather than once per manager.
The line it replaces was ``logger.info("Generated new HS256 secret key")``,
which named neither the consequence nor the wiring that avoids it — and an
INFO line from a library during start-up is exactly what an operator learns
to scroll past. Matches the one-time-ERROR shape ``SecretManager`` adopted
for the same failure class in #2041 / PR #2063.
The variable name is spelled out literally rather than interpolated from
:data:`JWT_SECRET_KEY_ENV`. Nothing sensitive is involved either way — this
is the NAME of an environment variable, never a value — but CodeQL's
``py/clear-text-logging-sensitive-data`` matches on identifier names and
reads a constant called ``..._SECRET_KEY_ENV`` as key material. A literal
has no dataflow into the sink at all, which answers the alert rather than
suppressing it. Drift between the two is what
``test_the_loud_signal_names_the_current_constants`` pins.
The generated secret itself is NEVER logged: a fix that fails loudly must
not turn a durability bug into a disclosure bug (``security.md``).
"""
logger.error(
"JWT signing keys were GENERATED, not configured: auto_generate_keys=True "
"and no signing key was supplied. The keys exist only inside this "
"process, so every token issued now becomes invalid at the next restart, "
"and any other replica signs with a different key and will reject these "
"tokens. This is for local development only. Set "
"KAILASH_JWT_SECRET_KEY, or pass secret_key= (HS256) or "
"private_key=/public_key= (RSA), and leave auto_generate_keys at its "
"default of False."
)
[docs]
class JWTAuthManager:
"""
Enterprise JWT Authentication Manager.
Provides comprehensive JWT token management with security best practices:
- Support for both HS256 (default) and RSA algorithms
- RSA key pair generation and rotation (when using RSA)
- Refresh token management
- Token revocation via a pluggable revocation store
- Comprehensive audit logging
- Rate limiting protection
This consolidates both JWTAuthManager and KailashJWTAuthManager functionality.
Token revocation in multi-worker deployments
---------------------------------------------
Token revocation is backed by a :class:`TokenRevocationStore`. By default
(``revocation_store`` omitted, ``config.enable_blacklist=True``) an in-memory
:class:`InMemoryTokenRevocationStore` is used, which is **process-local**: a
token revoked through one worker is NOT rejected by other workers. For any
multi-worker / multi-pod deployment supply a SHARED store (Redis, database,
distributed cache) implementing :class:`TokenRevocationStore` via the
``revocation_store`` constructor argument so revocation propagates to every
worker that shares it (issue #1356).
Known process-local state (NOT covered by ``revocation_store``): refresh-token
tracking (``refresh_access_token`` / ``revoke_refresh_token`` /
``revoke_all_user_tokens``) and rate-limit accounting are also per-instance
in-memory and behave per-worker. Shared-backend coverage for those is tracked
separately; only access-token revocation propagates through the store.
"""
[docs]
def __init__(
self,
config: Optional[JWTConfig] = None,
secret_key: Optional[str] = None,
algorithm: Optional[str] = None,
use_rsa: Optional[bool] = None,
revocation_store: Optional[TokenRevocationStore] = None,
**kwargs,
):
"""
Initialize JWT Auth Manager.
Args:
config: JWTConfig object with full configuration
secret_key: Secret key for HS256 (overrides config)
algorithm: Algorithm to use (overrides config)
use_rsa: Whether to use RSA (overrides config)
revocation_store: Backend that records and checks token revocation.
When ``config.enable_blacklist`` is True and this is omitted, a
process-local :class:`InMemoryTokenRevocationStore` is used —
which means a token revoked on one worker is NOT rejected by
other workers. Supply a SHARED store (Redis, database,
distributed cache) implementing :class:`TokenRevocationStore` so
revocation propagates across every worker that shares it
(issue #1356). Ignored when ``config.enable_blacklist`` is False.
**kwargs: Additional config parameters
"""
self.config = config or JWTConfig()
# Override config with direct parameters for backward compatibility
if secret_key is not None:
self.config.secret_key = secret_key
if algorithm is not None:
self.config.algorithm = algorithm
if use_rsa is not None:
self.config.use_rsa = use_rsa
if use_rsa:
self.config.algorithm = "RS256"
# Apply any additional kwargs to config
for key, value in kwargs.items():
if hasattr(self.config, key):
setattr(self.config, key, value)
# Key management
self._private_key: Optional[Any] = (
None # rsa.RSAPrivateKey when rsa is available
)
self._public_key: Optional[Any] = None # rsa.RSAPublicKey when rsa is available
self._secret_key: Optional[str] = self.config.secret_key
self._key_id = str(uuid.uuid4())
self._key_generated_at = datetime.now(timezone.utc)
# Token revocation backend. When blacklisting is enabled, use the
# injected shared store if provided, else a process-local default.
# When disabled, no store is held and revocation is a no-op.
self._revocation_store: Optional[TokenRevocationStore] = None
if self.config.enable_blacklist:
self._revocation_store = revocation_store or InMemoryTokenRevocationStore()
self._refresh_tokens: Dict[str, Dict[str, Any]] = {}
self._failed_attempts: Dict[str, List[datetime]] = {}
# Initialize keys based on algorithm
self._initialize_keys()
def _initialize_keys(self):
"""Initialize keys based on configured algorithm.
A signing key is REQUIRED and there is no default (issue #2083). With
nothing configured this raises rather than minting a throwaway key:
releases up to kailash 2.64 generated one here and continued at INFO
level, which invalidated every outstanding token at each restart and
made a multi-replica deployment authenticate non-deterministically.
Generation is still available for local development, but only as an
explicit ``auto_generate_keys=True`` opt-in, and it announces itself
once per process at ERROR level.
"""
if self.config.use_rsa or self.config.algorithm.startswith("RS"):
# RSA mode
if self.config.private_key and self.config.public_key:
# Load provided keys
self._load_rsa_keys()
elif self.config.auto_generate_keys:
# Explicit opt-in: generate, but say so loudly. The signal fires
# here rather than inside _generate_key_pair so that deliberate
# key ROTATION on a configured manager stays quiet.
_warn_ephemeral_signing_key()
self._generate_key_pair()
else:
raise ValueError(
"RSA mode requires a configured key pair: pass "
"private_key= and public_key= (PEM), or a JWTConfig "
"carrying them. Set auto_generate_keys=True ONLY for local "
"development — generated keys live in this process, so all "
"tokens die at restart and other replicas reject them "
"(issue #2083)."
)
else:
# HS256 mode
if not self._secret_key:
# Environment wiring, so an operator can configure this without
# a code seam — the error below names this variable, and an
# error naming a variable nothing reads is worse than silence.
env_secret = os.environ.get(JWT_SECRET_KEY_ENV)
if env_secret and env_secret.strip():
# Validated because this variable is NEW here: shipping a
# fresh key-input surface that accepts a four-character
# signing secret would re-open the same hole one door down.
# Measured stripped, used verbatim — trimming would change
# the signing key for anyone whose secret ends in
# whitespace, and every token signed before the trim would
# stop verifying.
if len(env_secret.strip()) < MIN_ENV_SECRET_LENGTH:
# Names the length, never the secret: this message
# reaches logs and crash reports (`security.md`).
raise ValueError(
f"The signing secret in KAILASH_JWT_SECRET_KEY "
f"carries {len(env_secret.strip())} non-whitespace "
f"characters; at least {MIN_ENV_SECRET_LENGTH} are "
f"required. RFC 7518 section 3.2 requires an "
f"HMAC-SHA256 key of at least 256 bits, and a "
f"shorter secret is brute-forceable offline from a "
f"single captured token — which would let an "
f"attacker mint tokens this manager accepts. Note "
f"that leading and trailing whitespace IS "
f"significant in the secret itself: it is measured "
f"trimmed but used verbatim."
)
self._secret_key = env_secret
self.config.secret_key = env_secret
if not self._secret_key:
if self.config.auto_generate_keys:
# Explicit opt-in: generate, but say so loudly.
_warn_ephemeral_signing_key()
self._secret_key = secrets.token_urlsafe(32)
self.config.secret_key = self._secret_key
else:
raise ValueError(
"HS256 mode requires a configured signing secret: set "
"the KAILASH_JWT_SECRET_KEY environment variable, or "
"pass secret_key=. Set auto_generate_keys=True ONLY for "
"local development — a generated secret lives in this "
"process, so all tokens die at restart and other "
"replicas reject them (issue #2083)."
)
def _load_rsa_keys(self):
"""Load RSA keys from PEM strings."""
try:
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
if self.config.private_key is None or self.config.public_key is None:
raise ValueError("RSA mode requires both private_key and public_key")
self._private_key = serialization.load_pem_private_key(
self.config.private_key.encode(),
password=None,
backend=default_backend(),
)
self._public_key = serialization.load_pem_public_key(
self.config.public_key.encode(), backend=default_backend()
)
logger.info("Loaded RSA keys from configuration")
except Exception as e:
logger.error(f"Failed to load RSA keys: {e}")
raise
def _generate_key_pair(self):
"""Generate new RSA key pair for token signing."""
if rsa is None:
raise ImportError("cryptography package is required for RSA key generation")
self._private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048
)
self._public_key = self._private_key.public_key()
self._key_id = str(uuid.uuid4())
self._key_generated_at = datetime.now(timezone.utc)
logger.info(f"Generated new JWT key pair with ID: {self._key_id}")
def _should_rotate_keys(self) -> bool:
"""Check if keys should be rotated."""
if not self._key_generated_at:
return True
rotation_threshold = timedelta(days=self.config.key_rotation_days)
return datetime.now(timezone.utc) - self._key_generated_at > rotation_threshold
def _create_token_payload(
self,
user_id: str,
token_type: str = "access",
tenant_id: Optional[str] = None,
session_id: Optional[str] = None,
permissions: Optional[List[str]] = None,
roles: Optional[List[str]] = None,
**kwargs,
) -> TokenPayload:
"""Create token payload with all claims."""
now = datetime.now(timezone.utc)
# Determine expiration
if token_type == "access":
expire_delta = timedelta(minutes=self.config.access_token_expire_minutes)
else: # refresh
expire_delta = timedelta(days=self.config.refresh_token_expire_days)
return TokenPayload(
sub=user_id,
iss=self.config.issuer,
aud=self.config.audience,
exp=int((now + expire_delta).timestamp()),
iat=int(now.timestamp()),
jti=str(uuid.uuid4()),
tenant_id=tenant_id,
session_id=session_id,
token_type=token_type,
permissions=permissions or [],
roles=roles or [],
**kwargs,
)
[docs]
def create_access_token(
self,
user_id: str,
tenant_id: Optional[str] = None,
session_id: Optional[str] = None,
permissions: Optional[List[str]] = None,
roles: Optional[List[str]] = None,
**kwargs,
) -> str:
"""Create JWT access token."""
# Only rotate keys in RSA mode
if self.config.use_rsa and self._should_rotate_keys():
self._generate_key_pair()
payload = self._create_token_payload(
user_id=user_id,
token_type="access",
tenant_id=tenant_id,
session_id=session_id,
permissions=permissions,
roles=roles,
**kwargs,
)
# Convert payload to dict for encoding
payload_dict = {
"sub": payload.sub,
"iss": payload.iss,
"aud": payload.aud,
"exp": payload.exp,
"iat": payload.iat,
"jti": payload.jti,
"tenant_id": payload.tenant_id,
"session_id": payload.session_id,
"token_type": payload.token_type,
"permissions": payload.permissions,
"roles": payload.roles,
}
payload_dict.update(kwargs)
if jwt is None:
raise ImportError("PyJWT package is required for JWT operations")
# Sign token based on algorithm
if self.config.use_rsa or self.config.algorithm.startswith("RS"):
# RSA signing
headers = {"kid": self._key_id}
token = jwt.encode(
payload_dict,
self._private_key, # type: ignore[reportArgumentType]
algorithm=self.config.algorithm,
headers=headers,
)
else:
# HS256 signing
token = jwt.encode(
payload_dict,
self._secret_key, # type: ignore[reportArgumentType]
algorithm=self.config.algorithm,
)
logger.debug(
"Created access token for user %s", sanitize_log_value(user_id, 128)
)
return token
[docs]
def create_refresh_token(
self,
user_id: str,
tenant_id: Optional[str] = None,
session_id: Optional[str] = None,
**kwargs,
) -> str:
"""Create JWT refresh token."""
payload = self._create_token_payload(
user_id=user_id,
token_type="refresh",
tenant_id=tenant_id,
session_id=session_id,
**kwargs,
)
# Convert payload to dict
payload_dict = {
"sub": payload.sub,
"iss": payload.iss,
"aud": payload.aud,
"exp": payload.exp,
"iat": payload.iat,
"jti": payload.jti,
"tenant_id": payload.tenant_id,
"session_id": payload.session_id,
"token_type": payload.token_type,
"refresh_count": payload.refresh_count,
}
payload_dict.update(kwargs)
if jwt is None:
raise ImportError("PyJWT package is required for JWT operations")
# Sign token based on algorithm
if self.config.use_rsa or self.config.algorithm.startswith("RS"):
headers = {"kid": self._key_id}
token = jwt.encode(
payload_dict,
self._private_key, # type: ignore[reportArgumentType]
algorithm=self.config.algorithm,
headers=headers,
)
else:
token = jwt.encode(
payload_dict,
self._secret_key, # type: ignore[reportArgumentType]
algorithm=self.config.algorithm,
)
# Store refresh token metadata
self._refresh_tokens[payload.jti] = {
"user_id": user_id,
"tenant_id": tenant_id,
"session_id": session_id,
"created_at": datetime.now(timezone.utc),
"refresh_count": 0,
"last_used": None,
}
logger.debug(
"Created refresh token for user %s", sanitize_log_value(user_id, 128)
)
return token
[docs]
def create_token_pair(
self,
user_id: str,
tenant_id: Optional[str] = None,
session_id: Optional[str] = None,
permissions: Optional[List[str]] = None,
roles: Optional[List[str]] = None,
**kwargs,
) -> TokenPair:
"""Create access and refresh token pair."""
access_token = self.create_access_token(
user_id, tenant_id, session_id, permissions, roles, **kwargs
)
refresh_token = self.create_refresh_token(
user_id, tenant_id, session_id, **kwargs
)
expires_at = datetime.now(timezone.utc) + timedelta(
minutes=self.config.access_token_expire_minutes
)
return TokenPair(
access_token=access_token,
refresh_token=refresh_token,
expires_in=self.config.access_token_expire_minutes * 60,
expires_at=expires_at,
)
[docs]
def verify_token(self, token: str) -> Dict[str, Any]:
"""
Verify and decode JWT token.
Returns:
Decoded token payload or raises exception
"""
if jwt is None:
raise ImportError("PyJWT package is required for JWT operations")
try:
# Get verification key based on algorithm
if self.config.use_rsa or self.config.algorithm.startswith("RS"):
# RSA verification
# Decode without verification first to get header
unverified_header = jwt.get_unverified_header(token)
key_id = unverified_header.get("kid")
# Verify key ID matches current key (optional check)
if key_id and key_id != self._key_id:
# `key_id` comes from `get_unverified_header` -- it is read
# BEFORE any signature check, because reading it is how the
# verification key gets selected, so it is wholly
# unauthenticated attacker-chosen input. Interpolated, an
# embedded newline lets an unauthenticated caller forge
# additional well-formed WARNING records on the very path
# a key-ID prober drives repeatedly (issue #2104).
logger.warning(
"Token signed with unknown key ID: %s",
sanitize_log_value(key_id, 128),
)
# In production, you might want to support multiple keys
# for graceful key rotation
# Verify and decode token
payload = jwt.decode(
token,
self._public_key, # type: ignore[reportArgumentType]
algorithms=[self.config.algorithm],
issuer=self.config.issuer,
audience=self.config.audience,
)
else:
# HS256 verification
payload = jwt.decode(
token,
self._secret_key, # type: ignore[reportArgumentType]
algorithms=[self.config.algorithm],
issuer=self.config.issuer,
audience=self.config.audience,
)
# Reject revoked tokens. Checked AFTER decode so the revocation
# identity (jti) is available — this is what a shared store keys on
# so revocation propagates across workers (issue #1356).
# INVARIANT: pass BOTH jti AND token — revoke() keys on `jti or token`,
# so a token revoked before its jti was known (decode-failure path) is
# keyed by raw token; dropping `token=` here would silently stop
# enforcing those revocations.
if (
self._revocation_store is not None
and self._revocation_store.is_revoked(
jti=payload.get("jti"), token=token
)
):
raise jwt.InvalidTokenError("Token has been revoked")
return payload
except jwt.ExpiredSignatureError:
logger.debug("Token has expired")
raise
except jwt.InvalidTokenError as e:
# PyJWT builds several of these messages FROM the presented token --
# `DecodeError(f"Invalid header string: {e}")` is the clearest case
# -- so the exception text is attacker-influenced on exactly the
# path an unauthenticated caller drives (issue #2104).
logger.warning("Invalid token: %s", sanitize_log_value(e))
raise
except Exception as e:
logger.error("Token verification error: %s", sanitize_log_value(e))
raise jwt.InvalidTokenError(f"Token verification failed: {e}")
[docs]
def refresh_access_token(self, refresh_token: str) -> TokenPair:
"""
Create new access token using refresh token.
Args:
refresh_token: Valid refresh token
Returns:
New token pair with refreshed access token
"""
if jwt is None:
raise ImportError("PyJWT package is required for JWT operations")
try:
# Verify refresh token
payload = self.verify_token(refresh_token)
if payload.get("token_type") != "refresh":
raise jwt.InvalidTokenError("Token is not a refresh token")
jti = payload.get("jti")
if jti is None or jti not in self._refresh_tokens:
raise jwt.InvalidTokenError("Refresh token not found")
refresh_data = self._refresh_tokens[jti]
# Check refresh count limit
if refresh_data["refresh_count"] >= self.config.max_refresh_count:
self.revoke_refresh_token(jti)
raise jwt.InvalidTokenError("Refresh token has exceeded usage limit")
# Update refresh count
refresh_data["refresh_count"] += 1
refresh_data["last_used"] = datetime.now(timezone.utc)
# Create new token pair
tenant_id: Optional[str] = payload.get("tenant_id")
session_id: Optional[str] = payload.get("session_id")
return self.create_token_pair(
user_id=payload["sub"],
tenant_id=tenant_id,
session_id=session_id,
permissions=payload.get("permissions", []),
roles=payload.get("roles", []),
)
except Exception as e:
logger.error("Token refresh failed: %s", sanitize_log_value(e))
raise
[docs]
def revoke_token(self, token: str):
"""Revoke a token via the revocation store.
With a shared store this propagates to every worker that shares it; with
the default in-memory store it is process-local (issue #1356).
"""
if self._revocation_store is None:
return
try:
payload = self.verify_token(token)
jti = payload.get("jti")
exp = payload.get("exp")
expires_at = datetime.fromtimestamp(exp, tz=timezone.utc) if exp else None
self._revocation_store.revoke(jti=jti, token=token, expires_at=expires_at)
# `jti` is decoded from a token the CALLER presented. It verified,
# so it is not unauthenticated -- but a holder of any valid token
# still chooses this value if it minted the token elsewhere, and a
# verified caller is not a trusted log author (issue #2104).
if jti:
logger.info("Revoked token %s", sanitize_log_value(jti, 128))
else:
logger.info("Revoked token")
except Exception:
# Even if verification fails, revoke by raw token string so a
# malformed/expired token presented for revocation is still recorded.
# Bound the entry's TTL so an attacker spamming revoke with unique
# invalid strings cannot grow the store without limit. The longest a
# legitimately-issued token can remain presentable is the refresh
# window; cap the entry there so a FORGED far-future `exp` in an
# unverified token cannot extend the entry's lifetime beyond it
# (no presentable token outlives the cap, so the entry self-purges
# without ever evicting a still-valid token).
ttl_cap = datetime.now(timezone.utc) + timedelta(
days=self.config.refresh_token_expire_days
)
expires_at = ttl_cap
try:
unverified = jwt.decode(
token, options={"verify_signature": False, "verify_exp": False}
)
exp = unverified.get("exp")
if exp:
expires_at = min(
datetime.fromtimestamp(exp, tz=timezone.utc), ttl_cap
)
except Exception:
expires_at = ttl_cap
self._revocation_store.revoke(jti=None, token=token, expires_at=expires_at)
[docs]
def revoke_refresh_token(self, jti: str):
"""Revoke specific refresh token."""
if jti in self._refresh_tokens:
del self._refresh_tokens[jti]
logger.info("Revoked refresh token %s", sanitize_log_value(jti, 128))
[docs]
def revoke_all_user_tokens(self, user_id: str):
"""Revoke all tokens for a specific user."""
# Remove all refresh tokens for user
to_remove = []
for jti, data in self._refresh_tokens.items():
if data["user_id"] == user_id:
to_remove.append(jti)
for jti in to_remove:
del self._refresh_tokens[jti]
logger.info("Revoked all tokens for user %s", sanitize_log_value(user_id, 128))
[docs]
def cleanup_expired_tokens(self):
"""Remove expired tokens from tracking."""
now = datetime.now(timezone.utc)
# Clean up expired refresh tokens
expired_refresh = []
for jti, data in self._refresh_tokens.items():
# Check if token is older than refresh token lifetime
token_age = now - data["created_at"]
if token_age > timedelta(days=self.config.refresh_token_expire_days):
expired_refresh.append(jti)
for jti in expired_refresh:
del self._refresh_tokens[jti]
# Clean up old failed attempts (keep only last hour)
cutoff = now - timedelta(hours=1)
for ip, attempts in list(self._failed_attempts.items()):
recent_attempts = [t for t in attempts if t > cutoff]
if recent_attempts:
self._failed_attempts[ip] = recent_attempts
else:
del self._failed_attempts[ip]
if expired_refresh or self._failed_attempts:
logger.debug(f"Cleaned up {len(expired_refresh)} expired refresh tokens")
[docs]
def get_public_key_jwks(self) -> Dict[str, Any]:
"""Get public key in JWKS format for external verification."""
if not self._public_key:
return {}
# Convert public key to JWKS format
public_numbers = self._public_key.public_numbers()
return {
"keys": [
{
"kty": "RSA",
"kid": self._key_id,
"use": "sig",
"alg": self.config.algorithm,
"n": self._encode_number(public_numbers.n),
"e": self._encode_number(public_numbers.e),
}
]
}
def _encode_number(self, number: int) -> str:
"""Encode number for JWKS format."""
import base64
byte_length = (number.bit_length() + 7) // 8
number_bytes = number.to_bytes(byte_length, "big")
return base64.urlsafe_b64encode(number_bytes).decode("ascii").rstrip("=")
[docs]
def get_stats(self) -> Dict[str, Any]:
"""Get authentication manager statistics.
Note: ``blacklisted_tokens`` retains the legacy field name (the config
flag ``enable_blacklist`` likewise) for backward compatibility; the
underlying mechanism is the revocation store. The value is the store's
``count()`` — an int for the in-memory default, but it MAY be ``None``
for a shared backend that cannot cheaply count (per the
``TokenRevocationStore.count`` contract), meaning "unknown".
"""
return {
"active_refresh_tokens": len(self._refresh_tokens),
"blacklisted_tokens": (
self._revocation_store.count()
if self._revocation_store is not None
else 0
),
"key_id": self._key_id,
"key_age_days": (
(datetime.now(timezone.utc) - self._key_generated_at).days
if self._key_generated_at
else 0
),
"failed_attempts_tracked": len(self._failed_attempts),
"config": {
"algorithm": self.config.algorithm,
"access_token_expire_minutes": self.config.access_token_expire_minutes,
"refresh_token_expire_days": self.config.refresh_token_expire_days,
},
}