Header Manipulation at the Gateway — Request and Response Header Transformation
In this tutorial, you'll learn about Header Manipulation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Header manipulation at the gateway allows you to add, remove, or transform HTTP headers as requests pass through, enabling consistent header policies across all backend services.
What You'll Learn
By the end of this lesson, you will implement header add, remove, and transform rules, forward client IP and original protocol, strip sensitive headers, and inject correlation IDs and authentication context.
Why It Matters
Centralized header management at the gateway eliminates duplicate header logic in every service, ensures consistent header policies, and prevents sensitive header leakage.
Real-World Use
Durga Antivirus Pro uses gateway header manipulation to inject X-Correlation-Id, X-User-Id, and X-Forwarded-For headers while stripping internal headers like X-Debug and X-Internal-Token.
Header Transformation Flow
flowchart LR
Client-->|Request Headers|Gateway
Gateway-->Add[Add Headers]
Gateway-->Remove[Remove Headers]
Gateway-->Transform[Transform Headers]
Add-->Backend[Backend Service]
Remove-->Backend
Transform-->Backend
Backend-->|Response Headers|Gateway
Gateway-->Strip[Strip Internal Headers]
Strip-->Client
Header Manipulation Engine
A configurable engine that applies header rules to requests and responses.
from typing import Dict, List, Optional, Callable, Tuple
import re
class HeaderRule:
def __init__(self, action: str, header: str,
value: Optional[str] = None,
pattern: Optional[str] = None):
self.action = action
self.header = header
self.value = value
self.pattern = re.compile(pattern) if pattern else None
class HeaderManipulator:
def __init__(self):
self.request_rules: List[HeaderRule] = []
self.response_rules: List[HeaderRule] = []
def add_request_rule(self, rule: HeaderRule):
self.request_rules.append(rule)
def add_response_rule(self, rule: HeaderRule):
self.response_rules.append(rule)
def apply_request_rules(self, headers: Dict
) -> Dict:
result = dict(headers)
for rule in self.request_rules:
if rule.action == "add" and rule.header not in result:
result[rule.header] = rule.value
elif rule.action == "set":
result[rule.header] = rule.value
elif rule.action == "remove":
result.pop(rule.header, None)
elif rule.action == "transform" and rule.pattern:
if rule.header in result:
result[rule.header] = rule.pattern.sub(
rule.value or "", result[rule.header]
)
return result
def apply_response_rules(self, headers: Dict
) -> Dict:
result = dict(headers)
for rule in self.response_rules:
if rule.action == "remove":
result.pop(rule.header, None)
elif rule.action == "set":
result[rule.header] = rule.value
return result
manipulator = HeaderManipulator()
manipulator.add_request_rule(HeaderRule("add", "X-Correlation-Id", "auto"))
manipulator.add_request_rule(HeaderRule("remove", "X-Debug"))
manipulator.add_request_rule(HeaderRule("remove", "Authorization"))
manipulator.add_request_rule(HeaderRule("set", "X-Forwarded-Proto", "https"))
manipulator.add_response_rule(HeaderRule("remove", "X-Internal-Token"))
manipulator.add_response_rule(HeaderRule("remove", "Server"))
headers = {
"Authorization": "Bearer token123",
"X-Debug": "true",
"Content-Type": "application/json"
}
result = manipulator.apply_request_rules(headers)
print(f"Modified request headers: {result}")
Client IP Forwarding
Forward the real client IP to backend services through standard headers.
from typing import Dict, Optional, Tuple
class IPForwarder:
def __init__(self, trusted_proxies: Optional[list] = None):
self.trusted_proxies = set(trusted_proxies or [])
def get_client_ip(self, remote_addr: str,
headers: Dict) -> str:
if remote_addr in self.trusted_proxies:
forwarded = headers.get("X-Forwarded-For", "")
if forwarded:
return forwarded.split(",")[0].strip()
return remote_addr
def set_forwarded_headers(self, remote_addr: str,
headers: Dict) -> Dict:
result = dict(headers)
client_ip = self.get_client_ip(remote_addr, headers)
existing = headers.get("X-Forwarded-For", "")
if existing:
result["X-Forwarded-For"] = f"{existing}, {client_ip}"
else:
result["X-Forwarded-For"] = client_ip
result["X-Real-IP"] = client_ip
return result
def is_internal_request(self, remote_addr: str) -> bool:
return remote_addr.startswith("10.") or \
remote_addr.startswith("172.16.") or \
remote_addr.startswith("192.168.")
forwarder = IPForwarder(trusted_proxies=["10.0.0.1", "10.0.0.2"])
result = forwarder.set_forwarded_headers(
"10.0.0.1",
{"X-Forwarded-For": "203.0.113.42"}
)
print(f"Forwarded headers: {result}")
client_ip = forwarder.get_client_ip("10.0.0.1", result)
print(f"Real client IP: {client_ip}")
Response Header Security
Strip or set security-related headers on responses passing through the gateway.
from typing import Dict
class ResponseSecurityHeaders:
def __init__(self):
self.security_headers = {
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"X-XSS-Protection": "1; mode=block",
"Strict-Transport-Security": "max-age=31536000; includeSubDomains",
"Referrer-Policy": "strict-origin-when-cross-origin",
}
self.strip_headers = {
"Server", "X-Powered-By", "X-AspNet-Version"
}
def apply(self, response_headers: Dict
) -> Dict:
result = dict(response_headers)
for h in self.strip_headers:
result.pop(h, None)
for h, v in self.security_headers.items():
if h not in result:
result[h] = v
return result
def add_custom_header(self, name: str, value: str):
self.security_headers[name] = value
security = ResponseSecurityHeaders()
security.add_custom_header("Permissions-Policy",
"camera=(), microphone=()")
response_headers = {
"Server": "Kestrel",
"Content-Type": "application/json"
}
result = security.apply(response_headers)
print(f"Secure response headers: {result}")
Common Mistakes
Mistake 1: Forwarding the Authorization Header to Backends
Once authenticated, strip the Authorization header. Backends should use gateway-injected headers for identity.
Mistake 2: Not Validating X-Forwarded-For
Any client can set X-Forwarded-For. Only trust it from known proxies.
Mistake 3: Overwriting Existing Headers
When adding headers, check if they already exist. Overwriting client-set headers may break legitimate use cases.
Mistake 4: Exposing Internal Headers
Headers like X-Debug, X-Internal, or stack trace headers must be stripped from responses.
Mistake 5: Header Injection via Malicious Input
If header values come from request data, sanitize them to prevent HTTP header injection attacks.
Practice Questions
- Why should the gateway strip the Authorization header before forwarding to backends?
- What is the purpose of the X-Forwarded-For header?
- How do you prevent header injection attacks?
- What security headers should every API response include?
- How do you handle header manipulation for Websocket upgrades?
Challenge
Build a header manipulation module for the gateway that adds X-Correlation-Id if missing, strips Authorization and X-Debug headers from requests, sets security headers on responses (CSP, HSTS, X-Frame-Options), and forwards the real client IP.
FAQ
Mini Project
Build a header manipulation middleware for the gateway that supports add, set, remove, and transform rules for both requests and responses, with separate rulesets for internal and external routes, security header injection, and client IP forwarding with trusted proxy validation.
What's Next
Learn about CORS Gateway for cross-origin request handling, or explore Compression for response size optimization.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro