Gateway Security — Protecting Your API Gateway
In this tutorial, you'll learn about Gateway Security. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
An API gateway sits at the perimeter of your system, making it the first line of defense against attacks and the most critical component to secure properly.
What You'll Learn
By the end of this lesson, you will implement gateway-level security controls including request validation, authentication enforcement, rate limiting, and IP filtering.
Why It Matters
A compromised gateway exposes every downstream service. Securing the gateway prevents attackers from reaching internal services and limits Blast Radius.
Real-World Use
Durga Antivirus Pro uses its API gateway to validate all incoming scan requests, reject malformed payloads before they reach the scanning service, and enforce per-client rate limits.
Gateway Security Layers
flowchart LR
Client-->Gateway
subgraph Gateway[API Gateway Security Layers]
WAF[Web App Firewall]
Auth[Authentication]
Rate[Rate Limiting]
Validate[Input Validation]
end
Gateway-->Backend[Internal Services]
Request Validation Middleware
The gateway must validate every incoming request before forwarding. Invalid requests get rejected early, saving backend resources.
import json
from typing import Dict, Any, Optional
class GatewaySecurityFilter:
BLOCKED_HEADERS = ["x-internal", "x-debug", "x-admin"]
MAX_BODY_SIZE = 1024 * 100 # 100KB
ALLOWED_METHODS = {"GET", "POST", "PUT", "DELETE", "PATCH"}
def validate_request(self, method: str, path: str,
headers: Dict[str, str],
body: Optional[bytes]) -> Optional[str]:
if method not in self.ALLOWED_METHODS:
return f"Method {method} not allowed"
for h in self.BLOCKED_HEADERS:
if h in headers:
return f"Blocked header: {h}"
if body and len(body) > self.MAX_BODY_SIZE:
return "Request body too large"
path_segments = path.split("/")
for seg in path_segments:
if ".." in seg or "//" in seg:
return "Path traversal detected"
return None
filter = GatewaySecurityFilter()
error = filter.validate_request("GET", "/api/scan", {}, None)
print(error) # None - valid
error = filter.validate_request("DELETE", "/api/scan",
{"x-internal": "true"}, None)
print(error) # Blocked header: x-internal
Authentication Enforcement
The gateway should reject unauthenticated requests before they reach any service.
import jwt
from datetime import datetime, timedelta
from typing import Optional
class GatewayAuthenticator:
def __init__(self, secret: str):
self.secret = secret
self.public_paths = {"/health", "/docs", "/favicon.ico"}
def is_public(self, path: str) -> bool:
return path in self.public_paths
def authenticate(self, token: str) -> Optional[Dict]:
try:
payload = jwt.decode(token, self.secret,
algorithms=["HS256"])
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
def gateway_handler(self, path: str,
auth_header: Optional[str]) -> Optional[str]:
if self.is_public(path):
return None
if not auth_header or not auth_header.startswith("Bearer "):
return "Missing or invalid authorization header"
token = auth_header.split(" ")[1]
payload = self.authenticate(token)
if payload is None:
return "Invalid or expired token"
return None
auth = GatewayAuthenticator("my-secret-key")
result = auth.gateway_handler("/api/scan", "Bearer invalid-token")
print(result) # Invalid or expired token
Rate Limiting by Client
Rate limiting at the gateway protects backends from abuse and fair-share allocation.
import time
from collections import defaultdict
from typing import Dict, Tuple
class SlidingWindowRateLimiter:
def __init__(self, max_requests: int = 100,
window_seconds: int = 60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: Dict[str, list] = defaultdict(list)
def check(self, client_id: str) -> Tuple[bool, int]:
now = time.time()
cutoff = now - self.window_seconds
self.requests[client_id] = [
t for t in self.requests[client_id]
if t > cutoff
]
self.requests[client_id].append(now)
remaining = self.max_requests - len(self.requests[client_id])
allowed = remaining >= 0
return allowed, max(0, remaining)
limiter = SlidingWindowRateLimiter(5, 10)
for i in range(7):
allowed, remaining = limiter.check("client-1")
print(f"Request {i+1}: allowed={allowed}, remaining={remaining}")
Common Mistakes
Mistake 1: Trusting the X-Forwarded-For Header
Attackers can spoof this header. Always validate against the known proxy IP list.
# BAD - trusts any X-Forwarded-For
client_ip = request.headers.get("X-Forwarded-For", "")
# GOOD - validates upstream
trusted_proxies = {"10.0.0.1", "10.0.0.2"}
upstream = request.remote_addr
if upstream in trusted_proxies:
client_ip = request.headers.get("X-Forwarded-For", upstream)
Mistake 2: Exposing Internal Routes
Internal API paths should never be accessible through the gateway.
# BAD - exposes internal path
route.add("/internal/admin/health")
# GOOD - only public routes
route.add("/v1/health")
Mistake 3: Ignoring Websocket Origin Checks
WebSocket connections must validate the Origin header during the upgrade.
# BAD - no origin check
upgrader.upgrade(request)
# GOOD - validates origin
allowed_origins = {"https://app.example.com"}
origin = request.headers.get("Origin", "")
if origin not in allowed_origins:
return 403
Mistake 4: Not Validating Content-Type
Attackers send XML to an endpoint expecting JSON to trigger XXE or other parser exploits.
# BAD - accepts any content type
body = request.get_data()
# GOOD - validates expected type
expected = "application/json"
if request.content_type != expected:
return 415
Mistake 5: Leaking Stack Traces
Gateway error responses should never expose internal details.
# BAD - exposes internals
return {"error": str(traceback.format_exc())}
# GOOD - generic error
return {"error": "Internal server error", "code": "GATEWAY_500"}
Practice Questions
What is the first security check a gateway should perform on every request? Answer: Validate the HTTP method, path, and headers to reject obviously malicious requests early.
Why should the gateway validate content length? Answer: To prevent buffer overflow attacks and denial of service through oversized payloads.
What header spoofing attack targets gateways that trust X-Forwarded-For? Answer: IP spoofing where an attacker sets a fake client IP to bypass IP-based access controls.
How does rate limiting at the gateway differ from rate limiting at the application layer? Answer: Gateway rate limiting protects all downstream services globally, while application rate limiting is service-specific.
What is the purpose of the Origin header check for WebSocket upgrades? Answer: Prevents cross-site WebSocket hijacking by ensuring only authorized web origins can establish WebSocket connections.
Challenge
Build a gateway security filter that blocks requests containing SQL Injection patterns in query parameters and returns a 403 with a sanitized error message.
FAQ
Mini Project
Build a gateway security proxy that intercepts all requests, validates JWT tokens from the Authorization header, enforces a rate limit of 50 requests per minute per client, and logs all rejected requests with the reason for rejection.
What's Next
Explore OAuth2 Gateway for delegated authorization, or learn about GraphQL Gateway for API composition patterns.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro