Authentication at the Gateway — Deep Dive into Auth Strategies
In this tutorial, you'll learn about Authentication Deep. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Authentication at the gateway centralizes identity verification so backend services can trust that every request reaching them has already been authenticated.
What You'll Learn
By the end of this lesson, you will implement gateway-level JWT validation, token introspection, API key authentication with rotation, and multi-factor authentication enforcement.
Why It Matters
Centralizing authentication at the gateway eliminates duplicate auth logic in every microservice, reduces the attack surface, and simplifies security audits.
Real-World Use
Durga Antivirus Pro authenticates all API requests at its gateway using JWT tokens for human users and API keys for machine clients, with automatic key rotation every 90 days.
Gateway Authentication Architecture
flowchart LR
Client-->Gateway
subgraph Gateway[API Gateway Auth]
JWT[JWT Validator]
APIKey[API Key Check]
OAuth[OAuth Introspection]
MFA[MFA Enforcer]
end
Gateway-->AuthService[Auth Service]
AuthService-->UserDB[(User DB)]
Gateway-->Backend[Protected Service]
JWT Validation at Gateway
The gateway validates JWT tokens before forwarding requests, rejecting invalid or expired tokens early.
import jwt
from datetime import datetime, timedelta
from typing import Dict, Optional, Tuple
import requests
class GatewayJWTAuth:
def __init__(self, jwks_url: Optional[str] = None,
secret: Optional[str] = None):
self.jwks_url = jwks_url
self.secret = secret
self.public_keys: Dict[str, str] = {}
def fetch_jwks(self):
if self.jwks_url:
response = requests.get(self.jwks_url)
jwks = response.json()
for key in jwks.get("keys", []):
self.public_keys[key["kid"]] = key
def validate_token(self, token: str
) -> Tuple[bool, Optional[Dict], Optional[str]]:
try:
if self.secret:
payload = jwt.decode(
token, self.secret,
algorithms=["HS256"]
)
else:
unverified = jwt.decode(
token, options={"verify_signature": False}
)
kid = unverified.get("kid")
if kid not in self.public_keys:
return False, None, "Unknown key ID"
public_key = self.public_keys[kid]
payload = jwt.decode(
token, public_key,
algorithms=["RS256"]
)
return True, payload, None
except jwt.ExpiredSignatureError:
return False, None, "Token expired"
except jwt.InvalidTokenError as e:
return False, None, str(e)
def extract_user(self, token: str) -> Optional[Dict]:
valid, payload, error = self.validate_token(token)
if not valid:
return None
return {
"user_id": payload.get("sub"),
"roles": payload.get("roles", []),
"scopes": payload.get("scope", "").split(),
}
auth = GatewayJWTAuth(secret="my-secret")
token = jwt.encode(
{"sub": "user-1", "roles": ["admin"], "exp": datetime.utcnow() + timedelta(hours=1)},
"my-secret", algorithm="HS256"
)
user = auth.extract_user(token)
print(f"Authenticated user: {user}")
API Key Authentication with Rotation
API keys for machine-to-machine communication require rotation and revocation support.
import hashlib
import hmac
import secrets
from datetime import datetime, timedelta
from typing import Dict, Optional, Tuple
class APIKeyManager:
def __init__(self):
self.keys: Dict[str, Dict] = {}
self.hash_keys: Dict[str, str] = {}
def create_key(self, client_id: str,
expires_in_days: int = 90) -> Tuple[str, str]:
raw_key = f"dag_{secrets.token_hex(24)}"
key_id = hashlib.sha256(
raw_key.encode()
).hexdigest()[:12]
hashed = hashlib.sha256(raw_key.encode()).hexdigest()
expires = datetime.utcnow() + timedelta(days=expires_in_days)
self.keys[key_id] = {
"client_id": client_id,
"hashed": hashed,
"created": datetime.utcnow(),
"expires": expires,
"active": True
}
self.hash_keys[hashed] = key_id
return key_id, raw_key
def validate_key(self, raw_key: str
) -> Tuple[bool, Optional[str]]:
hashed = hashlib.sha256(raw_key.encode()).hexdigest()
key_id = self.hash_keys.get(hashed)
if not key_id:
return False, "Unknown key"
key_data = self.keys.get(key_id)
if not key_data or not key_data["active"]:
return False, "Key revoked"
if datetime.utcnow() > key_data["expires"]:
return False, "Key expired"
return True, None
def revoke_key(self, key_id: str):
if key_id in self.keys:
self.keys[key_id]["active"] = False
def rotate_key(self, key_id: str,
expires_in_days: int = 90
) -> Optional[Tuple[str, str]]:
key_data = self.keys.get(key_id)
if not key_data:
return None
self.revoke_key(key_id)
return self.create_key(
key_data["client_id"], expires_in_days
)
manager = APIKeyManager()
key_id, raw = manager.create_key("scan-service")
print(f"Created key: {key_id}")
valid, error = manager.validate_key(raw)
print(f"Valid: {valid}")
manager.revoke_key(key_id)
valid, error = manager.validate_key(raw)
print(f"After revoke: {valid}, error: {error}")
Token Introspection for OAuth2
For OAuth2, the gateway introspects tokens with the authorization server on every request.
import requests
from typing import Dict, Optional, Tuple
from datetime import datetime
class TokenIntrospector:
def __init__(self, introspection_url: str,
client_id: str, client_secret: str):
self.url = introspection_url
self.auth = (client_id, client_secret)
self.cache: Dict[str, Tuple[bool, float]] = {}
def introspect(self, token: str,
use_cache: bool = True
) -> Tuple[bool, Optional[Dict]]:
if use_cache and token in self.cache:
valid, expiry = self.cache[token]
if datetime.utcnow().timestamp() < expiry:
return valid, None
response = requests.post(
self.url,
auth=self.auth,
data={"token": token},
timeout=2
)
result = response.json()
active = result.get("active", False)
if use_cache and active:
exp = result.get("exp", 0)
cache_ttl = min(exp - datetime.utcnow().timestamp(), 60)
self.cache[token] = (active, cache_ttl)
return active, result if active else None
def gateway_check(self, token: str,
required_scope: Optional[str] = None
) -> Tuple[bool, Optional[str]]:
active, data = self.introspect(token)
if not active:
return False, "Token inactive or invalid"
if required_scope:
scopes = data.get("scope", "").split()
if required_scope not in scopes:
return False, f"Missing scope: {required_scope}"
return True, None
# Usage:
# introspector = TokenIntrospector(
# "https://auth.example.com/introspect",
# "gateway-client", "client-secret"
# )
# valid, error = introspector.gateway_check("some-token", "scan:read")
Common Mistakes
Mistake 1: Validating JWT Without Signature Check
Decoding a JWT without verifying the signature lets attackers forge tokens. Always verify.
Mistake 2: Not Handling Token Expiration Gracefully
Returning a generic 401 without indicating expiration forces clients to guess the reason.
Mistake 3: Caching Introspection Results Too Long
OAuth2 tokens can be revoked at any time. Cache introspection for seconds, not hours.
Mistake 4: Mixing Authentication and Authorization
The gateway authenticates. Authorization checks belong in the service that knows the business rules.
Mistake 5: Logging Tokens
Never log tokens or API keys in plain text. Hash them before logging.
Practice Questions
- Why should authentication be centralized at the gateway rather than per-service?
- What is the difference between JWT validation and token introspection?
- How does API key rotation work without breaking existing clients?
- What headers should the gateway set for downstream services after authentication?
- How do you handle MFA enforcement at the gateway level?
Challenge
Build a gateway authentication plugin that supports both JWT and API key authentication, sets X-User-Id and X-User-Roles headers for downstream services, and returns appropriate WWW-Authenticate headers on failure.
FAQ
Mini Project
Build a gateway authentication module that supports JWT (RS256 with JWKS), API keys (with rotation and revocation), and OAuth2 token introspection, with a configurable priority chain and request header injection for downstream services.
What's Next
Learn about OAuth2 Gateway for delegated authorization flows, or explore JWT Authentication best practices and common attacks.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro