# Ssrf Prevention > Implements defense-in-depth SSRF prevention including URL validation, IP blocking, DNS rebinding protection, and cloud metadata endpoint hardening across Python, Node.js, and Go applications. - Skill: `paulpas/ssrf-prevention` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/ssrf-prevention` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/ssrf-prevention/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/ssrf-prevention --- # 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-validation` instead - 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-kubernetes` for network policies) - Cryptographic URL signing or tamper-proof redirects — use authentication/authorization patterns --- ## Core Workflow 1. **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. 2. **Enforce Scheme Whitelist** — Parse the URL and reject anything not on the allowed schemes (typically `http` and `https`). Explicitly block `file`, `gopher`, `dict`, `ssh`, `ftp`, `data`, `javascript`, and any custom protocol. **Checkpoint:** Reject URLs with empty scheme or malformed scheme before any DNS resolution occurs. 3. **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. 4. **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. 5. **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. 6. **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. ```python """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. ```typescript /** 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 { if (METADATA_IPS.has(hostname)) { throw new SsrfValidationError(originalUrl, `Blocked cloud metadata IP: ${hostname}`); } const dnsPromise = new Promise((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 { 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. ```go // 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. ```python """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. ```python """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,", "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)