JWT Authentication at the API Gateway — Validation and Claims Extraction
In this tutorial, you'll learn about JWT Gateway. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
JSON Web Token validation at the API Gateway ensures every request carries a valid, unexpired, properly signed token before reaching backend services.
What You'll Learn
By the end of this lesson, you will implement JWT validation with RS256 and HS256 algorithms, JWKS endpoint integration, claims extraction and header injection, and automated key rotation handling.
Why It Matters
Validating JWT at the gateway removes duplicate verification logic from every service and provides a single point to enforce token policies and detect invalid tokens.
Real-World Use
Durga Antivirus Pro validates JWT tokens at the gateway for all user-facing API requests, extracting user ID and role claims and injecting them as headers for downstream services.
JWT Validation Flow
flowchart LR
Client-->|Bearer Token|Gateway
Gateway-->Decode[Decode JWT]
Decode-->Verify{Verify Signature}
Verify-->|JWKS|Fetch[Fetch Public Key]
Verify-->|Secret|Local[Local Secret]
Fetch-->Check{Claims Valid?}
Local-->Check
Check-->|Yes|Inject[Inject Headers]
Check-->|No|Reject[401 Unauthorized]
Inject-->Backend[Backend Service]
JWT Validator with JWKS Support
A robust JWT validator that fetches public keys from a JWKS endpoint and caches them.
import jwt
import requests
from datetime import datetime, timedelta
from typing import Dict, Optional, Tuple, List, Any
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import base64
import json
import time
class JWKSValidator:
def __init__(self, jwks_url: str,
cache_ttl: int = 3600,
allowed_algorithms: Optional[List[str]] = None):
self.jwks_url = jwks_url
self.cache_ttl = cache_ttl
self.allowed_algs = allowed_algorithms or [
"RS256", "RS384", "RS512"
]
self.keys: Dict[str, Any] = {}
self.last_fetch: float = 0
def _fetch_keys(self):
if time.time() - self.last_fetch < self.cache_ttl:
return
try:
response = requests.get(
self.jwks_url, timeout=5
)
response.raise_for_status()
jwks = response.json()
for key_data in jwks.get("keys", []):
kid = key_data.get("kid")
if kid:
public_key = self._build_public_key(key_data)
if public_key:
self.keys[kid] = public_key
self.last_fetch = time.time()
except requests.RequestException:
pass
def _build_public_key(self, key_data: Dict) -> Optional[Any]:
try:
n_bytes = self._base64url_decode(
key_data["n"]
)
e_bytes = self._base64url_decode(
key_data["e"]
)
n = int.from_bytes(n_bytes, "big")
e = int.from_bytes(e_bytes, "big")
public_key = rsa.RSAPublicNumbers(e, n).public_key()
return public_key
except (KeyError, ValueError):
return None
def _base64url_decode(self, data: str) -> bytes:
padding = 4 - len(data) % 4
if padding != 4:
data += "=" * padding
return base64.urlsafe_b64decode(data)
def validate(self, token: str
) -> Tuple[bool, Optional[Dict], Optional[str]]:
self._fetch_keys()
try:
unverified = jwt.decode(
token, options={"verify_signature": False}
)
kid = unverified.get("kid")
alg = unverified.get("alg", "")
if kid and kid in self.keys:
public_key = self.keys[kid]
payload = jwt.decode(
token, public_key,
algorithms=self.allowed_algs
)
return True, payload, None
elif not kid:
return False, None, "Missing kid header"
else:
return False, None, "Unknown key ID"
except jwt.ExpiredSignatureError:
return False, None, "Token expired"
except jwt.InvalidTokenError as e:
return False, None, str(e)
validator = JWKSValidator(
"https://auth.example.com/.well-known/jwks.json"
)
# valid, payload, error = validator.validate("eyJhbGci...")
Claims Extraction and Header Injection
After validation, extract claims and inject them as request headers for downstream services.
from typing import Dict, Optional, Tuple
import re
class ClaimsExtractor:
def __init__(self):
self.claim_to_header = {
"sub": "X-User-Id",
"email": "X-User-Email",
"roles": "X-User-Roles",
"tenant_id": "X-Tenant-Id",
"name": "X-User-Name",
}
def extract_and_inject(self,
payload: Dict,
request_headers: Dict
) -> Dict:
headers = request_headers.copy()
for claim, header_name in self.claim_to_header.items():
value = payload.get(claim)
if value is not None:
if isinstance(value, list):
value = ",".join(value)
headers[header_name] = str(value)
headers.pop("Authorization", None)
return headers
def validate_required_claims(self, payload: Dict,
required: list[str]
) -> Tuple[bool, Optional[str]]:
for claim in required:
if claim not in payload:
return False, f"Missing required claim: {claim}"
return True, None
extractor = ClaimsExtractor()
payload = {
"sub": "user-42",
"email": "user@example.com",
"roles": ["admin", "scanner"],
"tenant_id": "tenant-1"
}
headers = extractor.extract_and_inject(
payload,
{"Authorization": "Bearer token", "Content-Type": "application/json"}
)
print(f"Downstream headers: {headers}")
Handling Token Refresh
The gateway can detect expired tokens and guide clients to refresh.
from datetime import datetime
from typing import Dict, Optional, Tuple
class TokenExpiryHandler:
def __init__(self, refresh_grace_period: int = 300):
self.grace_period = refresh_grace_period
def check_expiry(self, payload: Dict
) -> Tuple[str, Optional[Dict]]:
exp = payload.get("exp")
if not exp:
return "valid", None
now = datetime.utcnow().timestamp()
remaining = exp - now
if remaining < 0:
return "expired", {
"error": "token_expired",
"message": "Token has expired. Please refresh."
}
elif remaining < self.grace_period:
return "expiring", {
"warning": "token_expiring",
"expires_in": int(remaining)
}
return "valid", None
def build_refresh_response(self,
refresh_token: Optional[str]
) -> Dict:
if not refresh_token:
return {
"status": 401,
"body": {
"error": "token_expired",
"message": "Refresh token required"
}
}
return {
"status": 200,
"body": {
"message": "Use refresh token to obtain new access token"
}
}
handler = TokenExpiryHandler()
payload = {"sub": "user-1", "exp": datetime.utcnow().timestamp() - 10}
status, info = handler.check_expiry(payload)
print(f"Token status: {status}, info: {info}")
Common Mistakes
Mistake 1: Not Verifying the Signature
Decoding without verification accepts forged tokens. Always verify using the correct key and algorithm.
Mistake 2: Using HS256 with Shared Secrets in Microservices
HS256 requires all services to share the secret. Use RS256 so only the gateway needs the secret.
Mistake 3: Ignoring the kid Header
Without kid, the gateway cannot rotate keys. The token must include kid to identify which key signed it.
Mistake 4: Forwarding the Raw Token to Backends
Once validated, remove the Authorization header. Backends should trust the gateway-injected headers.
Mistake 5: Not Handling Clock Skew
Services may have slightly different clocks. Allow a few seconds of leeway in expiration validation.
Practice Questions
- What is the difference between HS256 and RS256 for gateway JWT validation?
- Why does the gateway need to fetch keys from a JWKS endpoint?
- What claims should a gateway extract and forward to downstream services?
- How do you handle JWT expiration in a gateway?
- What is the purpose of the kid header in a JWT?
Challenge
Build a JWT validation middleware for the gateway that fetches JWKS keys, validates RS256 signatures, extracts sub and roles claims, sets X-User-Id and X-User-Roles headers, and returns a structured 401 response with error details.
FAQ
Mini Project
Build a JWT gateway middleware that validates RS256 JWTs using a JWKS endpoint, caches public keys for 1 hour, extracts sub, email, and roles claims into headers, and returns proper 401 responses with www-authenticate headers for invalid or expired tokens.
What's Next
Learn about OAuth2 Gateway for OAuth2 integration, or explore API Key Authentication for machine-to-machine authentication.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro