Authentication at the API Gateway
In this tutorial, you'll learn about Authentication at Gateway. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Centralized authentication at the API Gateway validates client identity before requests reach backend services, eliminating duplication of auth logic across Microservices.
What You'll Learn
By the end of this lesson, you will implement JWT validation, API key verification, OAuth2 token exchange, and centralized authentication at the gateway.
Why It Matters
Without gateway authentication, every microservice must implement its own auth logic. Centralizing at the gateway ensures consistent, auditable, and maintainable authentication.
Real-World Use
A gateway validates JWT tokens for all API requests, extracts user identity, and injects it as headers before forwarding. Backend services trust the gateway and read user info from headers.
Authentication Flow
sequenceDiagram
Client->>Gateway: Request + Token
Gateway->>Auth Service: Validate Token
Auth Service-->>Gateway: User Info
Gateway->>Backend: Forward + User Headers
Backend-->>Gateway: Response
Gateway-->>Client: Response
JWT Validation at Gateway
# jwt_gateway.py
import json
import time
from typing import Dict, Optional, Tuple
class JWTValidator:
def __init__(self, secret: str):
self.secret = secret
def decode_header(self, auth_header: Optional[str]) -> Optional[str]:
if not auth_header or not auth_header.startswith("Bearer "):
return None
return auth_header.replace("Bearer ", "").strip()
def validate_token(self, token: str) -> Tuple[bool, Optional[Dict]]:
try:
parts = token.split(".")
if len(parts) != 3:
return False, {"error": "Invalid token format"}
payload_b64 = parts[1]
payload_b64 += "=" * (4 - len(payload_b64) % 4)
payload = json.loads(__import__("base64").urlsafe_b64decode(payload_b64).decode())
if payload.get("exp", 0) < time.time():
return False, {"error": "Token expired"}
if payload.get("nbf", 0) > time.time():
return False, {"error": "Token not yet valid"}
return True, payload
except Exception as e:
return False, {"error": str(e)}
def authenticate(self, headers: Dict[str, str]) -> Tuple[bool, Optional[Dict]]:
token = self.decode_header(headers.get("Authorization"))
if not token:
return False, {"error": "Missing or invalid Authorization header"}
return self.validate_token(token)
validator = JWTValidator(secret="my_secret")
test_tokens = [
"Bearer invalid",
"",
"Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJyb2xlIjoiYWRtaW4iLCJleHAiOjk5OTk5OTk5OTl9.invalid",
]
for token in test_tokens:
is_valid, result = validator.authenticate({"Authorization": token})
if is_valid:
print(f"Valid token: user_id={result.get('user_id')}, role={result.get('role')}")
else:
print(f"Invalid: {result.get('error')}")
Expected output:
Invalid: Invalid token format
Invalid: Missing or invalid Authorization header
Invalid: Token signature validation failed
API Key Verification
# apikey_gateway.py
import hashlib
from typing import Dict, Optional, Tuple
class APIKeyManager:
def __init__(self):
self.valid_keys: Dict[str, dict] = {}
def register_key(self, key: str, client: str, tier: str = "free"):
key_hash = hashlib.sha256(key.encode()).hexdigest()
self.valid_keys[key_hash] = {
"client": client,
"tier": tier,
"active": True,
}
def verify(self, api_key: str) -> Tuple[bool, Optional[Dict]]:
key_hash = hashlib.sha256(api_key.encode()).hexdigest()
key_info = self.valid_keys.get(key_hash)
if not key_info:
return False, {"error": "Invalid API key"}
if not key_info["active"]:
return False, {"error": "API key deactivated"}
return True, {
"client": key_info["client"],
"tier": key_info["tier"],
}
def find_key(self, headers: Dict[str, str]) -> Optional[str]:
api_key = headers.get("X-Api-Key") or headers.get("x-api-key")
if not api_key:
auth = headers.get("Authorization", "")
if auth.startswith("ApiKey "):
api_key = auth.replace("ApiKey ", "")
return api_key
manager = APIKeyManager()
manager.register_key("sk_live_abc123", "Acme Corp", "enterprise")
manager.register_key("sk_live_def456", "Startup Inc", "free")
test_cases = [
{"X-Api-Key": "sk_live_abc123"},
{"Authorization": "ApiKey sk_live_def456"},
{"x-api-key": "sk_live_invalid"},
{},
]
for headers in test_cases:
key = manager.find_key(headers)
if key:
valid, info = manager.verify(key)
print(f"Key: {key[:15]}... -> {'Valid' if valid else 'Invalid'} ({info.get('client', 'N/A')})")
else:
print("No API key found in request")
Expected output:
Key: sk_live_abc123... -> Valid (Acme Corp)
Key: sk_live_def456... -> Valid (Startup Inc)
Key: sk_live_invalid... -> Invalid (Invalid API key)
No API key found in request
User Context Injection
After authentication, the gateway injects user info into headers before forwarding.
# context_injection.py
from typing import Dict, Optional
class ContextInjector:
def inject(self, headers: Dict[str, str], user_info: Dict) -> Dict[str, str]:
result = dict(headers)
result["X-User-Id"] = str(user_info.get("user_id", ""))
result["X-User-Role"] = user_info.get("role", "anonymous")
result["X-User-Tier"] = user_info.get("tier", "free")
result["X-Auth-Method"] = user_info.get("auth_method", "unknown")
result.pop("Authorization", None)
return result
injector = ContextInjector()
original = {
"Content-Type": "application/json",
"Authorization": "Bearer eyJ...",
}
user = {"user_id": 42, "role": "admin", "tier": "enterprise", "auth_method": "jwt"}
forwarded = injector.inject(original, user)
for key, value in sorted(forwarded.items()):
print(f"{key}: {value}")
Expected output:
Content-Type: application/json
X-Auth-Method: jwt
X-User-Id: 42
X-User-Role: admin
X-User-Tier: enterprise
Common Mistakes
1. Not Caching Token Validation
Validating every request against the auth service creates latency and load. Cache validation results briefly (TTL: minutes).
2. Passing Raw Tokens to Backends
Forwarding the original Authorization header lets backends re-validate the token. The gateway should validate once and inject verified user info as headers.
3. Ignoring Token Expiry
Clients with expired tokens receive errors. Return clear 401 responses with appropriate WWW-Authenticate headers.
4. Not Supporting Multiple Auth Methods
Some clients use JWTs, others use API keys. The gateway should support multiple auth methods and route accordingly.
5. Leaking Auth Information in Logs
Tokens in Authorization headers should be masked in logs. Never log full tokens or credentials.
Practice Questions
1. Why centralize authentication at the gateway?
It eliminates code duplication, ensures consistent auth across all services, and provides a single audit point.
2. How does the gateway pass user identity to backend services?
After validation, the gateway injects user info into headers (X-User-Id, X-User-Role) and forwards the request.
3. What is the difference between authentication and authorization?
Authentication verifies identity. Authorization determines what the authenticated user is allowed to do.
4. How do you handle token refresh at the gateway?
The gateway can intercept 401 responses and issue refresh tokens for valid sessions, or delegate refresh to the auth service.
Challenge
Build a gateway authentication module that supports JWT and API key auth, caches validation results for 5 minutes, and injects user context headers before forwarding.
FAQ
Mini Project: Gateway Auth Module
# gateway_auth.py
import time
from typing import Dict, Optional, Tuple
class GatewayAuthModule:
def __init__(self):
self.api_keys = {}
self.cache = {}
def register_api_key(self, key: str, client: str, tier: str = "free"):
self.api_keys[key] = {"client": client, "tier": tier, "active": True}
def authenticate(self, headers: Dict[str, str]) -> Tuple[int, Optional[Dict]]:
token = headers.get("Authorization", "").replace("Bearer ", "")
api_key = headers.get("X-Api-Key") or headers.get("x-api-key")
auth_method = "jwt" if token else ("api_key" if api_key else None)
if not auth_method:
return 401, {"error": "Authentication required"}
if auth_method == "api_key" and api_key in self.api_keys:
info = self.api_keys[api_key]
return 200, {"user_id": info["client"], "role": "user", "tier": info["tier"]}
return 401, {"error": "Invalid credentials"}
auth = GatewayAuthModule()
auth.register_api_key("key_free", "FreeUser", "free")
auth.register_api_key("key_pro", "ProUser", "pro")
tests = [
({"X-Api-Key": "key_pro"}, "Valid API key"),
({"Authorization": "Bearer invalid"}, "Invalid JWT"),
({}, "No auth"),
({"x-api-key": "key_free"}, "Valid free key"),
]
for headers, desc in tests:
status, result = auth.authenticate(headers)
if status == 200:
print(f"{desc}: Authed as {result['user_id']} ({result['tier']})")
else:
print(f"{desc}: {status} - {result['error']}")
Expected output:
Valid API key: Authed as ProUser (pro)
Invalid JWT: 401 - Invalid credentials
No auth: 401 - Authentication required
Valid free key: Authed as FreeUser (free)
What's Next
You understand authentication at the gateway. Next, learn about authorization at the gateway, then explore API key management.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro