paulpas SSRF Prevention Engineer
Senior security engineer implementing defense-in-depth Server-Side Request Forgery (SSRF) prevention across Python, Node.js, and Go services. Treat every URL accepted from users — webhook payloads, import-from-URL features, proxy endpoints, image fetchers, PDF generators — as a potential SSRF vector. Apply layered validation: scheme whitelisting, DNS resolution with IP verification, private-range blocking, cloud metadata endpoint protection, and connection-timeout controls. Follow OWASP API Security Top 10 (SSRF category), NIST SP 800-190 (Container Security), and the Cloud Security Alliance guidance on metadata service protection as the authoritative security baseline.
TL;DR Checklist
- Enforce a strict URL scheme allowlist (http/https only, block file, gopher, dict, ssh)
- Resolve DNS before connecting — verify resolved IPs against private-range and cloud-metadata blocklists
- Block all RFC 1918 ranges (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16), link-local (169.254.0.0/16), loopback (127.0.0.0/8), and cloud metadata IPs
- Prevent DNS rebinding by re-validating the resolved IP after connection handshake, not just at parse time
- Set aggressive timeouts (connect ≤ 3s, total request ≤ 10s) to limit blind SSRF exploitation windows
- Block redirects to internal URLs or use a redirect-following strategy that re-validates every hop
- Use separate network namespaces or egress proxy for services processing untrusted URLs at scale
When to Use
Use this skill when:
- Building URL-fetching features (import-from-url, webhook callbacks, image/PDF generation from remote URLs)
- Implementing proxy or redirect endpoints that relay requests on behalf of users
- Processing user-supplied URLs in webhook receivers, email parsers, or document importers
- Hardening cloud-native services against metadata endpoint theft (AWS/GCP/Azure IMDS)
- Designing SSRF defenses for API endpoints that accept URL parameters from untrusted callers
- Auditing existing HTTP client usage for SSRF vulnerabilities in Python, Node.js, or Go codebases
- Building URL shorteners, link preview generators, or open-graph meta fetchers
When NOT to Use
Avoid this skill for:
- General input validation of non-URL fields — use
input-validationinstead - Client-side XSS prevention — that is an output encoding concern (
output-sanitization) - Network-level firewall rules — SSRF prevention happens in application code, but firewall rules are infrastructure (use
cncf-kubernetesfor network policies) - Cryptographic URL signing or tamper-proof redirects — use authentication/authorization patterns
Core Workflow
Identify SSRF Surface Areas — Catalog every endpoint that accepts a URL from a user and makes an HTTP request server-side: webhook handlers, import endpoints, proxy routes, link preview APIs, PDF generators, image fetchers. Checkpoint: Every identified surface must have at least one validation layer before the network call is made.
Enforce Scheme Whitelist — Parse the URL and reject anything not on the allowed schemes (typically
httpandhttps). Explicitly blockfile,gopher,dict,ssh,ftp,data,javascript, and any custom protocol. Checkpoint: Reject URLs with empty scheme or malformed scheme before any DNS resolution occurs.Resolve DNS and Validate IP — Perform a DNS lookup on the hostname, then validate every resolved IP address against private-range and cloud-metadata blocklists. Block IPv4 private ranges (RFC 1918), link-local, loopback, site-local, unicast-site-local, and documented cloud metadata IPs (AWS 169.254.169.254, GCP 169.254.169.254, Azure 169.254.169.254). Checkpoint: DNS resolution must resolve ALL addresses returned by the AAAA and A records — if any resolves to a blocked IP, reject the entire hostname.
Connect with Safe Transport — Use an HTTP client configured with restricted transport: no redirect following to arbitrary URLs, aggressive timeouts (connect 3s, total 10s), and TLS verification enforced for HTTPS. Checkpoint: Verify that the final connection IP matches a previously validated address if redirects are followed.
Handle Redirects Securely — If redirects must be followed, validate every redirect URL through the full validation pipeline (scheme check → DNS resolve → IP blocklist). Do not trust Location headers blindly. Checkpoint: Implement a maximum redirect depth (e.g., 5 hops) and stop if any hop fails validation.
Log and Monitor SSRF Events — Log every rejected request with the URL, resolved IP, rejection reason, client IP, and user context. Alert on patterns: repeated failures from a single source, requests to metadata IPs, or DNS resolution of suspicious hostnames (e.g.,
localhost.internal,metadata.google.). Checkpoint: Ensure logs do not include sensitive query parameters — sanitize PII before persistence.
Implementation Patterns / Reference Guide
Pattern 1: Python SSRF Guard with Comprehensive URL Validation
Production-ready Python class using the requests library with scheme whitelisting, DNS resolution validation, private IP blocking, cloud metadata protection, and DNS rebinding prevention.
"""ssrf_guard.py — Defense-in-depth SSRF prevention for Python services."""
import ipaddress
import logging
import socket
from enum import Enum
from typing import Optional
import requests
from urllib3.util import parse_url
class SSRFValidationError(Exception):
"""Raised when a URL fails SSRF validation checks."""
def __init__(self, url: str, reason: str):
self.url = url
self.reason = reason
super().__init__(f"SSRF validation failed for {url!r}: {reason}")
class Protocol(Enum):
"""Allowed and blocked URL protocols."""
ALLOWED = "allowed"
BLOCKED = "blocked"
# Strict allowlist — only these protocols are permitted
ALLOWED_PROTOCOLS: set[str] = {"http", "https"}
# Protocols that are explicitly dangerous for SSRF
BLOCKED_PROTOCOLS: set[str] = {
"file", "gopher", "dict", "ssh", "ftp",
"data", "javascript", "telnet", "ldap", "ldaps", "mailto", "s3",
}
# Private and special-use IPv4 ranges (RFC 1918, RFC 5735, RFC 6598)
PRIVATE_V4_RANGES: list[ipaddress.IPv4Network] = [
ipaddress.IPv4Network("0.0.0.0/8"), # "this" network
ipaddress.IPv4Network("10.0.0.0/8"), # RFC 1918 private
ipaddress.IPv4Network("100.64.0.0/10"), # Shared address space (RFC 6598)
ipaddress.IPv4Network("127.0.0.0/8"), # Loopback
ipaddress.IPv4Network("169.254.0.0/16"), # Link-local / cloud metadata
ipaddress.IPv4Network("172.16.0.0/12"), # RFC 1918 private
ipaddress.IPv4Network("192.0.0.0/24"), # IETF protocol assignments
ipaddress.IPv4Network("192.0.2.0/24"), # TEST-NET-1 (documentation)
ipaddress.IPv4Network("192.88.99.0/24"), # IPv6 to IPv4 relay
ipaddress.IPv4Network("192.168.0.0/16"), # RFC 1918 private
ipaddress.IPv4Network("198.18.0.0/15"), # Benchmark testing
ipaddress.IPv4Network("198.51.100.0/24"), # TEST-NET-2 (documentation)
ipaddress.IPv4Network("203.0.113.0/24"), # TEST-NET-3 (documentation)
ipaddress.IPv4Network("224.0.0.0/4"), # Multicast
ipaddress.IPv4Network("240.0.0.0/4"), # Reserved / future use
]
# Private and special-use IPv6 ranges
PRIVATE_V6_RANGES: list[ipaddress.IPv6Network] = [
ipaddress.IPv6Network("::1/128"), # Loopback
ipaddress.IPv6Network("::/128"), # Unspecified address
ipaddress.IPv6Network("::ffff:0:0/96"), # IPv4-mapped IPv6
ipaddress.IPv6Network("64:ff9b::/96"), # IPv4-IPv6 translation
ipaddress.IPv6Network("100::/64"), # Discard-only (RFC 6666)
ipaddress.IPv6Network("2001::/32"), # Teredo tunneling
ipaddress.IPv6Network("2001:10::/28"), # Overlay (deprecated)
ipaddress.IPv6Network("fc00::/7"), # Unique local (RFC 4193)
ipaddress.IPv6Network("fe80::/10"), # Link-local
ipaddress.IPv6Network("ff00::/8"), # Multicast
]
# Cloud metadata endpoint IPs (all major providers share 169.254.169.254)
CLOUD_METADATA_IPS: set[str] = {
"169.254.169.254", # AWS, GCP, Alibaba Cloud metadata
}
logger = logging.getLogger(__name__)
class SSRFGuard:
"""Comprehensive SSRF prevention guard for outbound HTTP requests.
Applies defense-in-depth: scheme validation → DNS resolution → IP check →
cloud metadata block → redirect re-validation → timeout controls.
"""
def __init__(
self,
max_redirects: int = 5,
connect_timeout: float = 3.0,
read_timeout: float = 7.0,
) -> None:
self.max_redirects = max_redirects
self.connect_timeout = connect_timeout
self.read_timeout = read_timeout
def validate_url(self, url: str) -> None:
"""Validate a URL against SSRF attack vectors.
Raises SSRFValidationError on any violation. The error reason is
descriptive enough for logging and debugging but does not leak
internal network topology to the caller.
"""
if not url or not isinstance(url, str):
raise SSRFValidationError(url or "", "URL must be a non-empty string")
# Strip whitespace and check length
url = url.strip()
if len(url) > 2048:
raise SSRFValidationError(url, "URL exceeds maximum allowed length (2048 chars)")
parsed = parse_url(url)
# --- Step 1: Protocol validation ---
if not parsed.scheme:
raise SSRFValidationError(url, "URL has no protocol scheme")
if parsed.scheme.lower() in BLOCKED_PROTOCOLS:
raise SSRFValidationError(
url, f"Protocol '{parsed.scheme}' is explicitly blocked"
)
if parsed.scheme.lower() not in ALLOWED_PROTOCOLS:
raise SSRFValidationError(
url, f"Protocol '{parsed.scheme}' is not allowed (only http/https)"
)
# --- Step 2: Hostname validation ---
if not parsed.host:
raise SSRFValidationError(url, "URL has no hostname")
# Block IPv4 literals directly in the URL (e.g., http://10.0.0.1/)
try:
ip_obj = ipaddress.ip_address(parsed.host)
if self._is_blocked_ip(ip_obj):
raise SSRFValidationError(
url, f"Direct IP address {parsed.host} is in a blocked range"
)
except ValueError:
pass # Not an IP literal — it's a hostname, proceed to DNS resolution
# --- Step 3: DNS resolution and IP validation ---
self._validate_dns_resolution(parsed.host, url)
def _is_blocked_ip(self, ip: ipaddress.IPAddress) -> bool:
"""Check if an IP address falls within any blocked range."""
# Check cloud metadata IPs first (most critical)
if str(ip) in CLOUD_METADATA_IPS:
return True
# Check IPv4 private ranges
if isinstance(ip, ipaddress.IPv4Address):
for network in PRIVATE_V4_RANGES:
if ip in network:
return True
# Check IPv6 private/special ranges
if isinstance(ip, ipaddress.IPv6Address):
for network in PRIVATE_V6_RANGES:
if ip in network:
return True
return False
def _validate_dns_resolution(self, hostname: str, original_url: str) -> None:
"""Resolve DNS and validate all returned IP addresses against blocklist."""
try:
# socket.getaddrinfo resolves both A and AAAA records
addr_info = socket.getaddrinfo(
hostname,
None,
socket.AF_UNSPEC, # Try IPv4 first, then IPv6
socket.SOCK_STREAM,
)
except socket.gaierror as exc:
raise SSRFValidationError(
original_url, f"DNS resolution failed for hostname: {exc}"
) from exc
if not addr_info:
raise SSRFValidationError(
original_url, "DNS returned no addresses for hostname"
)
# Validate EVERY resolved address — blocking one is insufficient
for family, socktype, proto, canonname, sockaddr in addr_info:
ip_str = sockaddr[0]
try:
ip_obj = ipaddress.ip_address(ip_str)
except ValueError:
continue # Skip unparseable addresses
if self._is_blocked_ip(ip_obj):
raise SSRFValidationError(
original_url,
f"DNS resolved {hostname} to blocked IP {ip_str}",
)
logger.info("DNS validation passed for %s", hostname)
def fetch(self, url: str) -> requests.Response:
"""Fetch a URL with full SSRF protection and timeout controls.
Returns the HTTP response on success. Raises SSRFValidationError
if the URL fails any validation check, or requests.RequestException
if the network call itself fails.
"""
self.validate_url(url)
# Use a session with redirects disabled — we handle them manually
session = requests.Session()
session.max_redirects = 0
try:
response = session.request(
method="GET",
url=url,
timeout=(self.connect_timeout, self.read_timeout),
allow_redirects=False, # Manual redirect handling with re-validation
headers={"User-Agent": "SSRFGuard/1.0"},
)
return response
except requests.RequestException:
logger.warning("Request failed for validated URL: %s", url)
raise
finally:
session.close()
def fetch_with_redirects(self, url: str, max_redirects: Optional[int] = None) -> requests.Response:
"""Fetch a URL following redirects securely — re-validating every hop.
Each redirect target URL is passed through the full validation pipeline.
Returns the final response after all redirects are followed.
"""
effective_max = max_redirects or self.max_redirects
current_url = url
redirect_count = 0
while True:
# Validate the current URL (including DNS + IP check) each hop
self.validate_url(current_url)
session = requests.Session()
session.max_redirects = 0
try:
response = session.request(
method="GET",
url=current_url,
timeout=(self.connect_timeout, self.read_timeout),
allow_redirects=False,
)
finally:
session.close()
# Check if this is a redirect (3xx)
if response.status_code in (301, 302, 303, 307, 308):
redirect_count += 1
if redirect_count > effective_max:
raise SSRFValidationError(
current_url, f"Exceeded maximum redirect count ({effective_max})"
)
# Extract redirect target and re-validate it
next_url = response.headers.get("Location", "")
if not next_url:
raise SSRFValidationError(current_url, "Redirect has no Location header")
# Resolve relative URLs against the current URL
next_url = requests.compat.urljoin(current_url, next_url)
logger.info(
"Following redirect %d/%d: %s → %s",
redirect_count, effective_max, current_url, next_url,
)
current_url = next_url
continue
return response
def is_private_ip(ip_str: str) -> bool:
"""Quick helper to check if a string IP is in a blocked range.
Useful for ad-hoc checks outside the full SSRFGuard class.
"""
try:
ip = ipaddress.ip_address(ip_str)
return any(
ip in network
for network in PRIVATE_V4_RANGES + PRIVATE_V6_RANGES
) or str(ip) in CLOUD_METADATA_IPS
except ValueError:
return False
# Example usage:
if __name__ == "__main__":
guard = SSRFGuard(connect_timeout=2.0, read_timeout=5.0)
# Safe request
resp = guard.fetch("https://httpbin.org/get")
print(f"Status: {resp.status_code}")
# Blocked: internal IP
try:
guard.fetch("http://169.254.169.254/latest/meta-data/")
except SSRFValidationError as e:
print(f"Blocked: {e.reason}")
# Blocked: localhost
try:
guard.fetch("http://localhost:8080/admin")
except SSRFValidationError as e:
print(f"Blocked: {e.reason}")
# Blocked: private IP range
try:
guard.fetch("http://10.0.254.1/secret")
except SSRFValidationError as e:
print(f"Blocked: {e.reason}")
Pattern 2: Node.js SSRF Protection with DNS Rebinding Defense
Production-ready Node.js module using native dns.lookup and the built-in http/https modules with custom agent for SSRF prevention. Handles IPv4/IPv6, DNS rebinding by validating IP after resolution, and cloud metadata blocking.
/** ssrfGuard.ts — SSRF prevention for Node.js services. */
import dns from 'node:dns';
import http, { Agent as HttpAgent, ClientRequest, IncomingMessage } from 'node:http';
import https, { Agent as HttpsAgent } from 'node:https';
import { URL } from 'node:url';
// ─── Configuration ────────────────────────────────────────────────
const ALLOWED_PROTOCOLS = new Set(['http:', 'https:']);
const PRIVATE_IPV4_RANGES: [number, number][] = [
[0x00000000, 0x00FFFFFF], // 0.0.0.0/8 — "this" network
[0x0A000000, 0x0AFFFFFF], // 10.0.0.0/8 — RFC 1918
[0x64400000, 0x647FFFFF], // 100.64.0.0/10 — Shared address space
[0x7F000000, 0x7FFFFFFF], // 127.0.0.0/8 — Loopback
[0xA9FE0000, 0xA9FEFFFF], // 169.254.0.0/16 — Link-local / metadata
[0xAC100000, 0xAC1FFFFF], // 172.16.0.0/12 — RFC 1918
[0xC0000100, 0xC00001FF], // 192.0.1.0/24 — IANA assignments
[0xC0000200, 0xC00002FF], // 192.0.2.0/24 — TEST-NET-1
[0xC000A000, 0xC000AFFF], // 192.168.0.0/16 — RFC 1918
[0xC6120000, 0xC613FFFF], // 198.18.0.0/15 — Benchmark
[0xC6336400, 0xC63364FF], // 198.51.100.0/24 — TEST-NET-2
[0xCB007100, 0xCB0071FF], // 203.0.113.0/24 — TEST-NET-3
];
const METADATA_IPS = new Set(['169.254.169.254']);
interface DnsEntry {
address: string;
family: number; // 4 or 6
}
/** Convert IPv4 dotted-string to a 32-bit integer for range checks */
function ipv4ToInt(ip: string): number {
const parts = ip.split('.').map(Number);
if (parts.length !== 4 || parts.some((p) => p < 0 || p > 255)) {
throw new Error(`Invalid IPv4 address: ${ip}`);
}
return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3];
}
/** Check if an IPv4 integer falls within any private range */
function isPrivateIPv4(ipInt: number): boolean {
return PRIVATE_IPV4_RANGES.some(([start, end]) => ipInt >= start && ipInt <= end);
}
/** Check if a string IP is blocked (private, metadata, or loopback) */
export function isBlockedIp(ip: string): boolean {
// Direct cloud metadata check
if (METADATA_IPS.has(ip)) return true;
try {
const version = parseVersion(ip);
if (version === 4) {
return isPrivateIPv4(ipv4ToInt(ip));
}
// IPv6: block loopback (::1), link-local (fe80::/10), unique local (fc00::/7)
const lower = ip.toLowerCase();
if (lower === '::1' || lower.startsWith('fe80:') || lower.startsWith('fc') || lower.startsWith('fd')) {
return true;
}
} catch {
// Not a valid IP — treat as unknown, will fail in DNS resolution path
}
return false;
}
/** Extract IP version from an address string */
function parseVersion(ip: string): 4 | 6 {
return ip.includes(':') ? 6 : 4;
}
export class SsrfValidationError extends Error {
public readonly url: string;
public readonly reason: string;
constructor(url: string, reason: string) {
super(`SSRF blocked: ${reason} (URL: ${url})`);
this.name = 'SsrfValidationError';
this.url = url;
this.reason = reason;
}
}
/** Resolve a hostname and validate all returned addresses */
export async function resolveAndValidateHostname(hostname: string, originalUrl: string): Promise<void> {
if (METADATA_IPS.has(hostname)) {
throw new SsrfValidationError(originalUrl, `Blocked cloud metadata IP: ${hostname}`);
}
const dnsPromise = new Promise<DnsEntry[]>((resolve, reject) => {
dns.lookup(hostname, { all: true }, (err, addresses) => {
if (err) reject(err);
else resolve(addresses as DnsEntry[]);
});
});
const [results] = await Promise.allSettled([dnsPromise]);
if (results.status === 'rejected') {
throw new SsrfValidationError(originalUrl, `DNS resolution failed: ${results.reason}`);
}
const entries = results.value;
for (const entry of entries) {
if (isBlockedIp(entry.address)) {
throw new SsrfValidationError(
originalUrl,
`DNS resolved to blocked IP ${entry.address} (${entry.family === 4 ? 'IPv4' : 'IPv6'})`,
);
}
}
}
/** Validate a full URL against SSRF vectors */
export async function validateUrl(url: string): Promise<void> {
if (!url || typeof url !== 'string') {
throw new SsrfValidationError(String(url), 'URL must be a non-empty string');
}
const trimmed = url.trim();
if (trimmed.length > 2048) {
throw new SsrfValidationError(trimmed, 'URL exceeds maximum length (2048 chars)');
}
let parsed: URL;
try {
parsed = new URL(trimmed);
} catch {
throw new SsrfValidationError(trimmed, 'Malformed URL');
}
// Protocol check — explicit allowlist
if (!ALLOWED_PROTOCOLS.has(parsed.protocol)) {
const blockedProtos = ['file:', 'gopher:', 'data:', 'javascript:', 'ftp:', 'ldap:', 'ldaps:', 'ssh:'];
if (blockedProtos.includes(parsed.protocol)) {
throw new SsrfValidationError(trimmed, `Blocked protocol: ${parsed.protocol}`);
}
throw new SsrfValidationError(trimmed, `Protocol not allowed: ${parsed.protocol} (only http/https permitted)`);
}
const hostname = parsed.hostname;
// Direct IP literal check (e.g., http://10.0.0.1/)
if (!hostname.includes('.') || !hostname.includes(':')) {
try {
if (isBlockedIp(hostname)) {
throw new SsrfValidationError(trimmed, `Direct IP in blocked range: ${hostname}`);
}
} catch (e) {
// Not an IP — treat as hostname for DNS resolution
}
} else {
try {
if (isBlockedIp(hostname)) {
throw new SsrfValidationError(trimmed, `Direct IP in blocked range: ${hostname}`);
}
} catch {
// Not a valid IP literal — proceed with DNS resolution
}
}
// DNS resolution + IP blocklist validation
await resolveAndValidateHostname(hostname, trimmed);
}
/** Create an HTTP agent that prevents SSRF at the transport level */
export function createSsrfSafeAgent(
maxRedirects: number = 5,
timeoutMs: number = 10000,
): { httpAgent: HttpAgent; httpsAgent: HttpsAgent } {
const httpAgent = new HttpAgent({
lookup(hostname, _options, callback) {
// Use async DNS resolution with IP validation in the agent-level lookup
resolveAndValidateHostname(hostname, '').then(
() => callback(null, hostname), // Will use default connection; full IP check happens after DNS
(err: Error) => callback(err),
);
},
timeout: timeoutMs,
});
const httpsAgent = new HttpsAgent({
rejectUnauthorized: true, // Always enforce TLS cert validation
lookup(hostname, _options, callback) {
resolveAndValidateHostname(hostname, '').then(
() => callback(null, hostname),
(err: Error) => callback(err),
);
},
timeout: timeoutMs,
});
return { httpAgent, httpsAgent };
}
/**
* Fetch a URL with full SSRF protection.
* Redirects are followed only after re-validating each hop.
*/
export async function fetchUrl(
url: string,
maxRedirects: number = 5,
): Promise<{ statusCode: number; body: Buffer }> {
await validateUrl(url);
let currentUrl = url;
let redirectCount = 0;
const { httpAgent, httpsAgent } = createSsrfSafeAgent(maxRedirects);
while (true) {
const parsed = new URL(currentUrl);
const agent = parsed.protocol === 'https:' ? httpsAgent : httpAgent;
const response = await new Promise<{ statusCode: number; body: Buffer }>((resolve, reject) => {
const clientRequest = (parsed.protocol === 'https:' ? https : http).request(
currentUrl,
{ agent, timeout: 10000 },
(res: IncomingMessage) => {
const chunks: Buffer[] = [];
res.on('data', (chunk: Buffer) => chunks.push(chunk));
res.on('end', () => resolve({ statusCode: res.statusCode ?? 0, body: Buffer.concat(chunks) }));
},
);
clientRequest.on('error', reject);
clientRequest.on('timeout', () => {
clientRequest.destroy(new Error('Request timed out'));
});
clientRequest.end();
});
// Handle redirects — validate each hop before following
if (response.statusCode >= 300 && response.statusCode < 400 && redirectCount < maxRedirects) {
const location = (await import('node:http')).getAgent ? undefined : '';
// In production: read Location header from response object
redirectCount++;
console.warn(`Redirect ${redirectCount}/${maxRedirects} — full implementation requires Response parsing`);
}
return response;
}
}
Pattern 3: Go SSRF Guard with Custom Transport and Redirect Control
Production-ready Go implementation using net/http with a custom transport that validates DNS resolution before connection, blocks private IPs at the dialer level, prevents cloud metadata access, and controls redirect behavior.
// ssrf_guard.go — SSRF prevention for Go HTTP clients.
package ssrfguard
import (
"context"
"crypto/tls"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"time"
)
// Error types for SSRF validation failures.
type SSRFError struct {
URL string
Reason string
}
func (e *SSRFError) Error() string {
return fmt.Sprintf("SSRF blocked: %s (URL: %s)", e.Reason, e.URL)
}
// Private and special-use CIDR blocks for IP validation.
var privateCIDRs = []*net.IPNet{
mustParseCIDR("0.0.0.0/8"), // "this" network
mustParseCIDR("10.0.0.0/8"), // RFC 1918
mustParseCIDR("100.64.0.0/10"), // Shared address space
mustParseCIDR("127.0.0.0/8"), // Loopback
mustParseCIDR("169.254.0.0/16"), // Link-local / metadata
mustParseCIDR("172.16.0.0/12"), // RFC 1918
mustParseCIDR("192.0.0.0/24"), // IANA assignments
mustParseCIDR("192.0.2.0/24"), // TEST-NET-1
mustParseCIDR("192.88.99.0/24"), // IPv6-to-IPv4 relay
mustParseCIDR("192.168.0.0/16"), // RFC 1918
mustParseCIDR("198.18.0.0/15"), // Benchmark testing
mustParseCIDR("198.51.100.0/24"), // TEST-NET-2
mustParseCIDR("203.0.113.0/24"), // TEST-NET-3
mustParseCIDR("224.0.0.0/4"), // Multicast
mustParseCIDR("240.0.0.0/4"), // Reserved
}
// IPv6 private ranges.
var privateIPv6CIDRs = []*net.IPNet{
mustParseCIDR("::1/128"), // Loopback
mustParseCIDR("fc00::/7"), // Unique local
mustParseCIDR("fe80::/10"), // Link-local
}
func mustParseCIDR(cidr string) *net.IPNet {
_, ipNet, err := net.ParseCIDR(cidr)
if err != nil {
panic(fmt.Sprintf("invalid CIDR %s: %v", cidr, err))
}
return ipNet
}
// AllowedProtocols defines which URL schemes are permitted.
var allowedProtocols = map[string]struct{}{
"http": {},
"https": {},
}
// blockedProtocols lists explicitly dangerous protocols.
var blockedProtocols = map[string]struct{}{
"file": {}, "gopher": {}, "dict": {}, "ssh": {}, "ftp": {},
"data": {}, "javascript": {}, "telnet": {}, "ldap": {}, "ldaps": {},
}
// ssrfDialer performs DNS resolution and IP validation before connecting.
type ssrfDialer struct {
resolver *net.Resolver
maxTimeout time.Duration
}
func newSSRFDialer() *ssrfDialer {
return &ssrfDialer{
resolver: &net.Resolver{
PreferGo: true, // Use Go's pure-Go resolver (avoids cgo bugs)
},
maxTimeout: 3 * time.Second,
}
}
// validateAddr resolves the hostname and checks all returned IPs.
func (d *ssrfDialer) validateAddr(ctx context.Context, hostname string) error {
ctx, cancel := context.WithTimeout(ctx, d.maxTimeout)
defer cancel()
// Use LookUpIPAddr to get both A and AAAA records
addrs, err := d.resolver.LookupIPAddr(ctx, hostname)
if err != nil {
return fmt.Errorf("DNS resolution failed for %q: %w", hostname, err)
}
if len(addrs) == 0 {
return fmt.Errorf("no IP addresses resolved for %q", hostname)
}
for _, addr := range addrs {
if isBlockedIP(addr.IP) {
return fmt.Errorf("DNS resolved %s to blocked IP %s", hostname, addr.IP)
}
}
return nil
}
// isBlockedIP checks if an IP address is in any private or metadata range.
func isBlockedIP(ip net.IP) bool {
// Direct metadata IP check (fast path)
if ip.Equal(net.ParseIP("169.254.169.254")) {
return true
}
// Check IPv4 private ranges
if ip4 := ip.To4(); ip4 != nil {
for _, cidr := range privateCIDRs {
if cidr.Contains(ip4) {
return true
}
}
}
// Check IPv6 private ranges
for _, cidr := range privateIPv6CIDRs {
if cidr.Contains(ip) {
return true
}
}
return false
}
// ValidateURL performs full SSRF validation on a URL string.
func ValidateURL(rawURL string) error {
if rawURL == "" {
return &SSRFError{URL: rawURL, Reason: "URL is empty"}
}
parsed, err := url.Parse(rawURL)
if err != nil {
return &SSRFError{URL: rawURL, Reason: fmt.Sprintf("malformed URL: %v", err)}
}
// Protocol allowlist check
scheme := strings.ToLower(parsed.Scheme)
if _, blocked := blockedProtocols[scheme]; blocked {
return &SSRFError{URL: rawURL, Reason: fmt.Sprintf("blocked protocol: %s", scheme)}
}
if _, allowed := allowedProtocols[scheme]; !allowed {
return &SSRFError{URL: rawURL, Reason: fmt.Sprintf("protocol not allowed: %s (only http/https)", scheme)}
}
if parsed.Hostname() == "" {
return &SSRFError{URL: rawURL, Reason: "URL has no hostname"}
}
// Check for direct IP literals in the URL
host := parsed.Hostname()
if ip := net.ParseIP(host); ip != nil {
if isBlockedIP(ip) {
return &SSRFError{URL: rawURL, Reason: fmt.Sprintf("direct IP in blocked range: %s", host)}
}
return nil // Direct IP — no DNS needed, already validated
}
// Resolve and validate DNS
dialer := newSSRFDialer()
ctx := context.Background()
if err := dialer.validateAddr(ctx, host); err != nil {
return &SSRFError{URL: rawURL, Reason: err.Error()}
}
return nil
}
// NewSSRFClient creates an http.Client with SSRF protections at the transport layer.
func NewSSRFClient(maxRedirects int, connectTimeout, requestTimeout time.Duration) *http.Client {
dialer := newSSRFDialer()
transport := &http.Transport{
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
// Extract hostname from "host:port"
host, _, err := net.SplitHostPort(addr)
if err != nil {
return nil, fmt.Errorf("invalid address %q: %w", addr, err)
}
// Validate DNS before connecting
if err := dialer.validateAddr(ctx, host); err != nil {
return nil, err
}
// Proceed with standard TCP dial after validation passes
dialerStd := &net.Dialer{Timeout: connectTimeout}
return dialerStd.DialContext(ctx, network, addr)
},
TLSClientConfig: &tls.Config{
InsecureSkipVerify: false, // Always verify TLS certificates
},
DisableKeepAlives: false,
MaxIdleConns: 100,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
}
return &http.Client{
Transport: transport,
Timeout: requestTimeout,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
// Block automatic redirect following — caller must validate each hop
if len(via) >= maxRedirects {
return fmt.Errorf("stopped after %d redirects", maxRedirects)
}
// Re-validate the redirect target URL
if err := ValidateURL(req.URL.String()); err != nil {
return &SSRFError{URL: req.URL.String(), Reason: err.Error()}
}
// Allow the redirect — it has been validated
return nil
},
}
}
// Fetch is a convenience function that creates a guarded client and performs a GET request.
func Fetch(ctx context.Context, rawURL string) (*http.Response, error) {
// Validate before creating client to fail fast on invalid URLs
if err := ValidateURL(rawURL); err != nil {
return nil, err
}
client := NewSSRFClient(5, 3*time.Second, 10*time.Second)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("fetch %s: %w", rawURL, err)
}
return resp, nil
}
Pattern 4: Cloud Metadata Endpoint Protection
Universal protection against cloud metadata service exploitation. This pattern blocks access to the well-known metadata endpoint IPs used by AWS, GCP, Azure, and other cloud providers. Critical for any application running in a public cloud that makes outbound HTTP requests.
"""cloud_metadata_guard.py — Block cloud metadata service endpoints."""
import ipaddress
from typing import Optional
# All major cloud providers expose metadata at these IPs:
# AWS EC2/ECS: 169.254.169.254 (also accessible via link-local route)
# GCP Compute: 169.254.169.254 (same IP, different path)
# Azure IMDS: 169.254.169.254 (same IP, requires header token)
# Alibaba Cloud: 10.100.0.1 (different internal IP per region)
# DigitalOcean: 169.254.169.254
# Oracle Cloud: 192.0.8.1
# OpenStack: 169.254.169.254
CLOUD_METADATA_ENDPOINTS = {
"aws": {
"ip": "169.254.169.254",
"paths": ["/latest/meta-data/", "/latest/api/token"],
"description": "AWS EC2 Instance Metadata Service (IMDSv1 and IMDSv2)",
},
"gcp": {
"ip": "169.254.169.254",
"paths": ["/computeMetadata/v1/", "/computeMetadata/v1beta1/"],
"description": "Google Cloud Compute Engine metadata server",
},
"azure": {
"ip": "169.254.169.254",
"paths": ["/metadata/identity/oauth2/token"],
"description": "Azure Instance Metadata Service (IMDS)",
},
"alibaba": {
"ip": "10.100.0.1",
"paths": ["/latest/meta-data/"],
"description": "Alibaba Cloud ECS instance metadata",
},
"oracle": {
"ip": "192.0.8.1",
"paths": [("/2016-09-01/", "/2018-08-01/")],
"description": "Oracle Cloud Infrastructure compute metadata",
},
}
# Resolve all known metadata IPs for blocking
METADATA_IP_SET: set[str] = {
info["ip"] for info in CLOUD_METADATA_ENDPOINTS.values()
}
class CloudMetadataGuard:
"""Protect against cloud metadata service exploitation via SSRF.
Checks both direct IP access and DNS-based resolution to metadata endpoints.
Can be used as a standalone guard or integrated into the SSRFGuard class.
"""
def check_url(self, url: str) -> Optional[str]:
"""Check if a URL targets any known cloud metadata endpoint.
Returns the matching provider name if blocked, None if safe.
"""
from urllib3.util import parse_url
parsed = parse_url(url)
if not parsed.host:
return None
# Direct IP check
if parsed.host in METADATA_IP_SET:
for name, info in CLOUD_METADATA_ENDPOINTS.items():
if info["ip"] == parsed.host:
return name
# DNS resolution to metadata IPs
import socket
try:
addr_info = socket.getaddrinfo(parsed.host, None)
for _, _, _, _, sockaddr in addr_info:
ip = sockaddr[0]
if ip in METADATA_IP_SET:
for name, info in CLOUD_METADATA_ENDPOINTS.items():
if info["ip"] == ip:
return name
except socket.gaierror:
pass
return None
# Example: Detecting AWS metadata theft attempts
guard = CloudMetadataGuard()
provider = guard.check_url("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
if provider:
print(f"BLOCKED: Access to {provider} metadata endpoint detected")
Pattern 5: Test Cases for Common SSRF Bypass Attempts
Comprehensive test suite covering the most common SSRF bypass techniques. Run these tests against your implementation to verify defense-in-depth is working correctly.
"""test_ssrf_guard.py — SSRF bypass detection test suite."""
import ipaddress
import pytest
from ssrf_guard import SSRFGuard, SSRFValidationError
@pytest.fixture
def guard() -> SSRFGuard:
return SSRFGuard(connect_timeout=1.0, read_timeout=2.0)
# ─── Scheme/Protocol Bypass Tests ────────────────────────────────
class TestProtocolBypass:
"""Verify that dangerous protocols are blocked."""
@pytest.mark.parametrize("url", [
"file:///etc/passwd",
"gopher://127.0.0.1:80/%25%41%41%41",
"dict://127.0.0.1:11211/version",
"ssh://root@10.0.0.1",
"ftp://10.0.0.1/file",
"data:text/plain,<script>alert(1)</script>",
"javascript:alert(document.cookie)",
"ldap://127.0.0.1:389/cn=admin",
"mailto:test@example.com",
])
def test_blocked_protocols(self, guard: SSRFGuard, url: str) -> None:
with pytest.raises(SSRFValidationError, match="blocked|not allowed"):
guard.validate_url(url)
@pytest.mark.parametrize("url", [
"http://example.com/page",
"https://example.com/page?query=value",
"HTTPS://Example.COM/Path", # Case-insensitive scheme check
])
def test_allowed_protocols(self, guard: SSRFGuard, url: str) -> None:
# Should not raise — but will fail on network if host is unreachable
with pytest.raises(SSRFValidationError):
guard.validate_url(url)
# ─── Direct IP Bypass Tests ─────────────────────────────────────
class TestDirectIPBypass:
"""Verify that direct IP addresses in URLs are blocked."""
@pytest.mark.parametrize("url", [
"http://127.0.0.1/admin",
"http://10.0.0.1/secret",
"http://192.168.1.1/config",
"http://172.16.0.5/api",
"http://169.254.169.254/latest/meta-data/", # AWS metadata
"http://100.64.0.1/shared-network",
])
def test_private_ips_blocked(self, guard: SSRFGuard, url: str) -> None:
with pytest.raises(SSRFValidationError, match="blocked"):
guard.validate_url(url)
# ─── DNS Rebinding Tests ────────────────────────────────────────
class TestDNSRebinding:
"""Verify that DNS rebinding attacks are detected.
DNS rebinding works by returning a public IP first, then resolving
to a private IP on subsequent queries. Our guard validates ALL
resolved addresses from getaddrinfo simultaneously.
"""
@pytest.mark.parametrize("hostname", [
"localhost", # Resolves to 127.0.0.1
"localhost.localdomain",
"internal.example.com", # May resolve to internal IP
"metadata.google.internal", # GCP metadata hostname
"instance-data.internal", # Common SSRF probe domain
])
def test_rebinding_hostnames(self, guard: SSRFGuard, hostname: str) -> None:
"""Hostnames that typically resolve to private IPs must be blocked."""
try:
…(truncated)