SSRF Remediation

import urllib.parse
from ipaddress import ip_address, ip_network

ALLOWED_DOMAINS = {'api.example.com', 'cdn.example.com', 'images.example.com'}
ALLOWED_SCHEMES = {'https'}

def is_url_allowed(url: str) -> bool:
    """Validate URL against allowlist"""
    try:
        parsed = urllib.parse.urlparse(url)
        
        # Check scheme
        if parsed.scheme not in ALLOWED_SCHEMES:
            return False
        
        # Check domain against allowlist
        if parsed.hostname not in ALLOWED_DOMAINS:
            return False
        
        # Prevent port manipulation
        if parsed.port and parsed.port not in [443, 80]:
            return False
            
        return True
    except Exception:
        return False

# Usage
url = request.form.get('url')
if not is_url_allowed(url):
    abort(400, 'Invalid URL')
    
response = requests.get(url, timeout=5)

import socket
from ipaddress import ip_address, ip_network

# Define private/reserved ranges
BLOCKED_NETWORKS = [
    ip_network('127.0.0.0/8'),      # Loopback
    ip_network('10.0.0.0/8'),       # Private
    ip_network('172.16.0.0/12'),    # Private
    ip_network('192.168.0.0/16'),   # Private
    ip_network('169.254.0.0/16'),   # Link-local (cloud metadata!)
    ip_network('0.0.0.0/8'),        # Current network
    ip_network('100.64.0.0/10'),    # Shared address space
    ip_network('192.0.0.0/24'),     # IETF protocol assignments
    ip_network('192.0.2.0/24'),     # TEST-NET-1
    ip_network('198.51.100.0/24'),  # TEST-NET-2
    ip_network('203.0.113.0/24'),   # TEST-NET-3
    ip_network('fc00::/7'),         # IPv6 unique local
    ip_network('fe80::/10'),        # IPv6 link-local
    ip_network('::1/128'),          # IPv6 loopback
]

def is_ip_allowed(ip_str: str) -> bool:
    """Check if IP is not in blocked ranges"""
    try:
        ip = ip_address(ip_str)
        for network in BLOCKED_NETWORKS:
            if ip in network:
                return False
        return True
    except ValueError:
        return False

def resolve_and_validate(hostname: str) -> str:
    """Resolve hostname and validate IP"""
    try:
        ip = socket.gethostbyname(hostname)
        if not is_ip_allowed(ip):
            raise ValueError(f"Blocked IP: {ip}")
        return ip
    except socket.gaierror:
        raise ValueError("DNS resolution failed")

import requests
from requests.adapters import HTTPAdapter

class SSRFSafeAdapter(HTTPAdapter):
    """HTTP adapter that validates IPs before connection"""
    
    def send(self, request, **kwargs):
        # Parse hostname from request
        from urllib.parse import urlparse
        parsed = urlparse(request.url)
        hostname = parsed.hostname
        
        # Resolve and validate before each request
        ip = socket.gethostbyname(hostname)
        if not is_ip_allowed(ip):
            raise ValueError(f"Blocked IP: {ip}")
        
        return super().send(request, **kwargs)

# Use custom session
session = requests.Session()
session.mount('http://', SSRFSafeAdapter())
session.mount('https://', SSRFSafeAdapter())

# Make request - IP validated at connection time
response = session.get(url, timeout=5)

# AWS - Require IMDSv2 (instance metadata service v2)
# This requires a PUT request with hop limit, blocking SSRF via GET

# Via AWS CLI
aws ec2 modify-instance-metadata-options \
    --instance-id i-1234567890abcdef0 \
    --http-tokens required \
    --http-put-response-hop-limit 1

# Terraform
resource "aws_instance" "example" {
  # ... other config
  
  metadata_options {
    http_tokens                 = "required"  # Enforce IMDSv2
    http_put_response_hop_limit = 1
    http_endpoint               = "enabled"
  }
}

# iptables - Block metadata endpoint from application
iptables -A OUTPUT -d 169.254.169.254 -j DROP

# AWS Security Group - Block outbound to metadata (not always possible)
# Better: Use IMDSv2 and validate at application level

# Docker - Block from containers
docker network create --driver bridge \
    --opt com.docker.network.bridge.enable_ip_masquerade=true \
    --opt com.docker.network.bridge.host_binding_ipv4=0.0.0.0 \
    no-metadata-network

# Kubernetes NetworkPolicy
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: block-metadata
spec:
  podSelector: {}
  policyTypes:
  - Egress
  egress:
  - to:
    - ipBlock:
        cidr: 0.0.0.0/0
        except:
        - 169.254.169.254/32

def fetch_url_safely(url: str, max_size: int = 1024 * 1024) -> bytes:
    """Fetch URL with safety controls"""
    
    # Validate URL first
    if not is_url_allowed(url):
        raise ValueError("URL not allowed")
    
    # Resolve and validate IP
    from urllib.parse import urlparse
    parsed = urlparse(url)
    resolve_and_validate(parsed.hostname)
    
    # Make request with safety controls
    response = requests.get(
        url,
        timeout=5,                    # Connection timeout
        allow_redirects=False,        # Don't follow redirects (or validate each)
        stream=True                   # Don't load entire response to memory
    )
    
    # Check content length
    content_length = response.headers.get('Content-Length')
    if content_length and int(content_length) > max_size:
        raise ValueError("Response too large")
    
    # Read with size limit
    content = b''
    for chunk in response.iter_content(chunk_size=8192):
        content += chunk
        if len(content) > max_size:
            raise ValueError("Response too large")
    
    return content