Secure API Gateway

What is an API Gateway?

A single entry point that sits between clients and backend microservices, enforcing security policies.

In a microservices architecture, an API Gateway acts as a reverse proxy that routes requests to the appropriate backend service. Beyond routing, it's the first line of defense for security, handling:

• Authentication: Who are you? (Verify identity)
• Authorization: What can you do? (Check permissions)
• Rate Limiting: Prevent API abuse and DDoS attacks
• Request Validation: Block malicious payloads (SQL injection, XSS)
• TLS Termination: Encrypt traffic between client and gateway
• Observability: Centralized logging, metrics, and tracing

Authentication Mechanisms

Authentication verifies the identity of the user or service making the request. The gateway validates credentials and rejects unauthorized requests.

1. API Keys

How it works: Clients include a secret key in request headers (X-API-Key: abc123). The gateway validates the key against a database or cache.

Pros: Simple, fast, stateless.

Cons: No user identity (all clients with same key look identical), keys can leak and be abused.

2. JWT (JSON Web Tokens)

How it works: Client authenticates with an auth service and receives a signed JWT. The gateway validates the JWT signature and extracts claims (user ID, roles, expiration).

Pros: Stateless, contains user identity and claims, can be verified without calling auth service.

Cons: Tokens can't be revoked (until expiration), larger payload size.

JWT Validation Example (Python)

import jwt
from flask import Flask, request, jsonify

app = Flask(__name__)
PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-----END PUBLIC KEY-----"""

def verify_jwt(token):
    """Validate JWT signature and extract claims"""
    try:
        payload = jwt.decode(
            token,
            PUBLIC_KEY,
            algorithms=["RS256"],
            audience="my-api",
            issuer="https://auth.example.com"
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise Exception("Token has expired")
    except jwt.InvalidTokenError:
        raise Exception("Invalid token")

@app.before_request
def authenticate():
    """Middleware to validate JWT on every request"""
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        return jsonify({"error": "Missing or invalid Authorization header"}), 401
    
    token = auth_header.split(' ')[1]
    try:
        user = verify_jwt(token)
        request.user = user  # Attach user to request context
    except Exception as e:
        return jsonify({"error": str(e)}), 401

@app.route('/api/protected')
def protected_route():
    return jsonify({"message": f"Hello, {request.user['sub']}!"})

3. OAuth 2.0

How it works: Delegated authorization protocol. Users authorize third-party apps to access their resources without sharing passwords. Gateway validates OAuth access tokens.

Flows:

• Authorization Code: For web apps (most secure)
• Client Credentials: For machine-to-machine communication
• Implicit/PKCE: For single-page apps (SPAs)

Authorization Patterns

After authentication, authorization determines what the authenticated user is allowed to do.

1. Role-Based Access Control (RBAC)

Users are assigned roles (e.g., Admin, Editor, Viewer), and roles have permissions. The gateway checks if the user's role allows the requested action.

RBAC Example

from functools import wraps
from flask import request, jsonify

# Define role permissions
ROLE_PERMISSIONS = {
    "admin": ["read", "write", "delete"],
    "editor": ["read", "write"],
    "viewer": ["read"]
}

def require_permission(required_permission):
    """Decorator to enforce permission checks"""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            user_role = request.user.get('role')  # From JWT claims
            allowed_permissions = ROLE_PERMISSIONS.get(user_role, [])
            
            if required_permission not in allowed_permissions:
                return jsonify({"error": "Forbidden: insufficient permissions"}), 403
            
            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route('/api/users', methods=['GET'])
@require_permission('read')
def list_users():
    return jsonify({"users": ["alice", "bob"]})

@app.route('/api/users', methods=['DELETE'])
@require_permission('delete')
def delete_user():
    return jsonify({"message": "User deleted"})

2. Attribute-Based Access Control (ABAC)

More granular than RBAC. Decisions based on attributes: user attributes (department, job title), resource attributes (owner, sensitivity), and context (time, location).

Example: "Allow if user.department == resource.owner_department AND time.hour between 9 and 17"

Rate Limiting & Throttling

Prevent API abuse by limiting the number of requests a client can make within a time window.

Common Algorithms

Rate Limiting Strategies

• Per-IP: Limit requests by client IP address
• Per-User: Limit requests by authenticated user ID
• Per-API Key: Different limits for different tiers (free vs paid)
• Global: Protect backend from total overload

Token Bucket Implementation (Redis + Python)

import redis
import time

redis_client = redis.StrictRedis(host='localhost', decode_responses=True)

def token_bucket_allow(key, capacity=100, refill_rate=10):
    """
    Token bucket rate limiter using Redis.
    
    Args:
        key: Unique identifier (IP, user ID, API key)
        capacity: Max tokens in bucket
        refill_rate: Tokens added per second
    
    Returns:
        True if request allowed, False if rate limited
    """
    now = time.time()
    bucket_key = f"rate_limit:{key}"
    
    # Get last update time and current tokens
    last_update = float(redis_client.hget(bucket_key, 'last_update') or now)
    tokens = float(redis_client.hget(bucket_key, 'tokens') or capacity)
    
    # Calculate tokens to add based on time elapsed
    elapsed = now - last_update
    tokens = min(capacity, tokens + elapsed * refill_rate)
    
    # Try to consume one token
    if tokens >= 1:
        tokens -= 1
        redis_client.hset(bucket_key, 'tokens', tokens)
        redis_client.hset(bucket_key, 'last_update', now)
        redis_client.expire(bucket_key, 3600)  # TTL: 1 hour
        return True
    else:
        return False

# Usage in Flask middleware
@app.before_request
def rate_limit():
    client_id = request.headers.get('X-API-Key') or request.remote_addr
    if not token_bucket_allow(client_id, capacity=100, refill_rate=10):
        return jsonify({"error": "Rate limit exceeded. Try again later."}), 429

Threat Protection

API Gateways must defend against common attack vectors.

TLS & Mutual TLS (mTLS)

Encrypt traffic between clients and the gateway, and between gateway and backend services.

TLS Termination at Gateway

The gateway handles TLS encryption/decryption, communicating with backend services over plain HTTP (within a secure internal network) or re-encrypting with TLS.

Benefits:

• Offload CPU-intensive encryption from backend services
• Centralize certificate management
• Inspect and modify requests (logging, transformation)

Mutual TLS (mTLS) for Service-to-Service

What is mTLS? Both client and server verify each other's certificates. Ensures only authorized services can communicate.

Use Case: Gateway-to-backend communication, microservices mesh.

Generating mTLS Certificates (Dev/Test)

# 1. Generate Certificate Authority (CA)
openssl genrsa -out ca.key 2048
openssl req -x509 -new -nodes -key ca.key -subj "/CN=MyCA" -days 3650 -out ca.crt

# 2. Generate Server Certificate
openssl genrsa -out server.key 2048
openssl req -new -key server.key -subj "/CN=api-gateway.local" -out server.csr
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365

# 3. Generate Client Certificate
openssl genrsa -out client.key 2048
openssl req -new -key client.key -subj "/CN=backend-service" -out client.csr
openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt -days 365

# Now configure gateway to require client.crt for mTLS verification

Observability & Monitoring

Comprehensive logging and metrics are essential for detecting security incidents and performance issues.

What to Log

• Access Logs: Request ID, timestamp, client IP, user ID, endpoint, HTTP status, latency
• Auth Failures: Failed login attempts, invalid tokens, permission denials
• Rate Limit Events: Which clients are being throttled
• Anomalies: Unusual traffic patterns, suspicious IPs

Key Metrics

Structured Logging Example

import json
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def log_request(request, response, user=None, duration=None):
    """Log request with structured JSON format"""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "request_id": request.headers.get('X-Request-ID'),
        "client_ip": request.remote_addr,
        "user_id": user.get('sub') if user else None,
        "method": request.method,
        "path": request.path,
        "query": request.query_string.decode(),
        "status_code": response.status_code,
        "duration_ms": duration * 1000 if duration else None,
        "user_agent": request.headers.get('User-Agent')
    }
    logger.info(json.dumps(log_entry))

# Usage in Flask
@app.after_request
def after_request(response):
    duration = time.time() - g.start_time  # Set in before_request
    log_request(request, response, request.user, duration)
    return response