Skip to content

CWE-918: Server-Side Request Forgery (SSRF) - Python

Overview

Server-Side Request Forgery (SSRF) allows attackers to make the server perform HTTP requests to arbitrary destinations, reaching internal services, cloud metadata endpoints, and hosts that are only reachable from the server's own network position.

Primary Defence: Validate URLs against an allowlist of permitted domains, block private/reserved IP ranges with the ipaddress module, and restrict protocols to https:// only.

Common Vulnerable Patterns

Direct URL Usage from User Input

# VULNERABLE - No validation on user-provided URL

import requests

def fetch_image(image_url):
    # No validation - SSRF vulnerability!
    response = requests.get(image_url)
    return response.content

# Attack examples:

# http://localhost:8080/admin

# http://169.254.169.254/latest/meta-data/iam/security-credentials/

# http://internal-only.svc.cluster.local/admin

Why this is vulnerable: requests.get() will fetch whatever host it is given, and the server's network position is the whole point - localhost and 169.254.169.254 are reachable from the application and not from the attacker. Redirects are followed by default, up to 30 of them, so even a destination that passes a check can hand the client somewhere else on the next hop.

Note what is not the risk here: requests resolves a scheme to a transport adapter and ships adapters only for http and https, so file:///etc/passwd raises InvalidSchema rather than reading the file. That payload belongs with urllib and file_get_contents(), below and on the PHP page. Reaching for it against requests and watching it fail is a good way to conclude wrongly that the code is safe.

Unvalidated urllib Requests

# VULNERABLE - urllib without URL validation

from urllib.request import urlopen

def download_file(url):
    # No validation - SSRF vulnerability!
    with urlopen(url) as response:
        return response.read()

# Attack: url = "http://internal-api.local/sensitive-endpoint"

Why this is vulnerable: urlopen() accepts far more than HTTP. It handles file://, ftp:// and data: through the same call, so a URL that never touches the network can still return the contents of a local file - and unlike requests, there is no separate transport adapter to restrict. It also follows redirects by default, so a destination that passed a check can hand the client somewhere else on the next hop.

Flask/Django Without Validation

# VULNERABLE - Flask endpoint without URL validation

from flask import Flask, request
import requests

app = Flask(__name__)

@app.route('/proxy')
def proxy():
    url = request.args.get('url')
    # No validation - SSRF vulnerability!
    response = requests.get(url)
    return response.text

# Attack: /proxy?url=http://169.254.169.254/latest/meta-data/

Why this is vulnerable: The framework contributes nothing here - request.args.get('url') is an attacker-controlled string and requests.get() will fetch it. What makes a proxy endpoint worse than the earlier examples is that the response is returned to the caller, turning a blind request into a readable one: the attacker sees the body of whatever internal service answered, rather than having to infer it from timing or error behaviour.

Webhook Handler Without Validation

# VULNERABLE - Webhook without URL validation

import requests

def send_webhook(webhook_url, data):
    # No validation - SSRF vulnerability!
    response = requests.post(webhook_url, json=data)
    return response.status_code

# Attack: webhook_url = "http://billing.internal/api/v1/refunds"  # an internal HTTP API

Why this is vulnerable: User-configured callback URLs are the hardest case, because fetching an arbitrary address is the feature. The destination is stored once and used repeatedly, often by a background worker with no request context, so a URL that resolved acceptably when it was saved is re-fetched later without anyone re-checking it. POST with an attacker-chosen body also means the request can act rather than only read: an internal HTTP API that trusts its network position will carry it out. Non-HTTP services are a smaller prize than they once were - Redis, for one, closes the connection when it sees a POST or Host: line - so the demonstration to reach for is an internal HTTP endpoint, not a Redis command smuggled into a path.

Secure Patterns

URL Allowlist Validation

# SECURE - Validate URLs against allowlist

import requests
from urllib.parse import urlparse
import ipaddress
import socket

# is_global is the right starting point and not the whole answer. Measured on
# CPython 3.13.12: 224.0.0.1 and ff02::1 are is_global True, and so are
# 64:ff9b::7f00:1 (the NAT64 well-known-prefix spelling of 127.0.0.1, which a
# compliant translator drops per RFC 6052 section 3.1), ::ffff:0:7f00:1 (the
# IPv4-translated form), fec0::1 (site-local space deprecated in 2004 and still
# in use) and 192.0.0.9 and 192.0.0.10, the two anycast hosts it exempts from
# 192.0.0.0/24. It does handle the mapped form: ::ffff:127.0.0.1 is False and
# ::ffff:8.8.8.8 is True. 3.13 also already treats 64:ff9b:1::/48 (RFC 8215),
# 2002::/16 and 2001::/32 as non-global; they stay in the list because older
# interpreters do not, and an entry that is redundant on one version costs
# less than one that is missing on another.
_BLOCKED = tuple(ipaddress.ip_network(cidr) for cidr in (
    '192.0.0.0/24',      # IETF protocol assignments - the whole block, as the other language pages do
    '64:ff9b::/96',      # NAT64 (RFC 6052)
    '64:ff9b:1::/48',    # NAT64 local use (RFC 8215)
    '2002::/16',         # 6to4 (RFC 3056)
    '2001::/32',         # Teredo (RFC 4380)
    '::ffff:0:0:0/96',   # IPv4-translated (RFC 6145) - one zero group on from the mapped form
    'fec0::/10',         # site-local, deprecated but still a private range
))


def blocked_address(ip) -> bool:
    """One address policy, shared by every example on this page."""
    ip = _unwrap_ipv4(ip)
    if not ip.is_global or ip.is_multicast:
        return True
    return any(ip in network for network in _BLOCKED if network.version == ip.version)


def _unwrap_ipv4(ip):
    """Return the IPv4 address an IPv6 one carries, where it carries one.

    `is_global` already handles the mapped form, but not the compatible one:
    measured on CPython 3.13, `::7f00:1` is `is_global` True, and it is
    127.0.0.1 - as `::a9fe:a9fe` is the metadata address. Anything whose
    integer value fits in 32 bits is that form; `::` and `::1` are the
    unspecified and loopback addresses rather than embedded IPv4, and
    unwrapping `::1` to 0.0.0.1 would lose what made it worth blocking.
    """
    if ip.version == 6 and ip.ipv4_mapped:
        return ip.ipv4_mapped
    if ip.version == 6 and 1 < int(ip) < 2 ** 32:
        return ipaddress.IPv4Address(int(ip))
    return ip


# The fetcher is its own module. It imports the session from "Validating the
# address actually connected to" below, which imports the policy above as
# ssrf_policy - so the dependency runs one way and there is no cycle
from ssrf_transport import secure_session, read_capped


class SafeImageFetcher:
    ALLOWED_HOSTS = {
        'api.example.com',
        'cdn.example.com',
        'images.example.com'
    }

    ALLOWED_SCHEMES = {'https'}

    def fetch_image(self, image_url: str) -> bytes:
        validated_url = self._validate_url(image_url)

        # A bare requests.get() would resolve the name a second time and take
        # http_proxy/https_proxy from the environment; secure_session() dials
        # only addresses it has checked and takes nothing from the environment
        with secure_session() as session:
            response = session.get(validated_url, timeout=10, allow_redirects=False, stream=True)
            response.raise_for_status()
            return read_capped(response)

    def _validate_url(self, url: str) -> str:
        try:
            parsed = urlparse(url)
        except Exception:
            raise SecurityError("Invalid URL")

        # Validate scheme
        if parsed.scheme not in self.ALLOWED_SCHEMES:
            raise SecurityError(f"Invalid URL scheme: {parsed.scheme}")

        # Validate host
        host = parsed.hostname
        if not host or host.lower() not in self.ALLOWED_HOSTS:
            raise SecurityError(f"Host not allowed: {host}")

        # Block private IP ranges
        if self._is_private_ip(host):
            raise SecurityError("Private IP addresses not allowed")

        return url

    def _is_private_ip(self, host: str) -> bool:
        try:
            # Resolve hostname to IP
            ip_addresses = socket.getaddrinfo(host, None)

            for ip_info in ip_addresses:
                ip_str = ip_info[4][0]
                ip = ipaddress.ip_address(ip_str)

                if blocked_address(ip):
                    return True

                # Check for AWS metadata IP
                if str(ip) == '169.254.169.254':
                    return True

            return False
        except Exception:
            # If DNS fails, block it
            return True

# The framework examples below import this exception as url_validation.
class SecurityError(Exception):
    pass

Why this works:

  • Host allowlist: Pre-approved domains (api.example.com, cdn.example.com, images.example.com) prevent arbitrary target selection
  • Scheme validation: urlparse identifies the scheme so the allowlist can reject file://, ftp://, and other non-HTTP protocols before the client runs
  • DNS resolution: socket.getaddrinfo resolves all IPs (IPv4/IPv6), then blocked_address classifies each one. is_global does most of that work - private, loopback, link-local, reserved, unspecified and the CGN range are all rejected by it, and so is ::ffff:127.0.0.1 - but it is not the whole answer, and the gaps are why the helper exists. Measured on CPython 3.13, 224.0.0.1 and ff02::1 are is_global True, so multicast needs its own test; so are 64:ff9b::7f00:1 and 64:ff9b:1::7f00:1, the NAT64 spellings of 127.0.0.1 (only the second can pass a compliant translator: RFC 6052 section 3.1 drops the first, RFC 8215 permits the second), ::ffff:0:7f00:1, the IPv4-translated spelling of the same address, and fec0::1, deprecated site-local space
  • Explicit AWS metadata protection: 169.254.169.254 is named in the code, documenting the cloud-metadata threat even though is_global already rejects link-local addresses
  • Fail-closed behavior: DNS errors block the request instead of falling back to a best-effort fetch
  • Redirect blocking: allow_redirects=False matters even with an allowlisted host. requests follows up to 30 redirects by default and re-checks none of them, so an allowlisted host - or one an attacker has a route to influence - can answer with 302 Location: http://169.254.169.254/, and the fetch lands there
  • Defense-in-depth: Multi-layer validation (allowlist -> scheme -> DNS -> IP) blocks common SSRF targets, and secure_session() makes the addresses checked at dial time the only ones connected to. A bare requests.get(validated_url) would resolve the name again - the race Validating the address actually connected to describes - and would honour https_proxy from the environment
  • Bounded read: timeout=10 caps each connect and each wait for data, not the whole exchange, so read_capped is what stops a slow trickle from an allowlisted host turning into an unbounded download
  • All-IP validation: Not just the first record, so a getaddrinfo answer that mixes public and private addresses is still rejected

Requests with Validation and Timeout

# SECURE - requests library with comprehensive validation

import requests
from urllib.parse import urlparse
import re
import ipaddress
import socket

# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address
# The SecurityError raised by the validation helpers above, as its own module
from url_validation import SecurityError
# The pinned session from "Validating the address actually connected to"
from ssrf_transport import secure_session

class SecureWebhookHandler:
    # Only allow specific domain pattern
    ALLOWED_URL_PATTERN = re.compile(r'^https://([a-z0-9-]+\.)*example\.com/.*$')

    def __init__(self):
        # trust_env=False and ValidatingAdapter mounted: the session takes no
        # proxy from the environment - through a proxy the hostname is resolved
        # at the proxy, not by the lookup below - and dials only addresses that
        # pass blocked_address, so the lookup below is not the last check
        self.session = secure_session()

    def send_webhook(self, webhook_url: str, data: dict) -> int:
        validated_url = self._validate_webhook_url(webhook_url)

        try:
            response = self.session.post(
                validated_url,
                json=data,
                timeout=10,
                allow_redirects=False
            )
            return response.status_code
        except requests.exceptions.RequestException as e:
            raise SecurityError(f"Request failed: {str(e)}")

    def _validate_webhook_url(self, url: str) -> str:
        if not url:
            raise SecurityError("URL cannot be empty")

        # fullmatch(), not match(): `$` also matches before a trailing newline,
        # so match() accepts an otherwise-valid URL with a newline appended -
        # measured on CPython 3.13 - and fullmatch() does not
        if not self.ALLOWED_URL_PATTERN.fullmatch(url):
            raise SecurityError(f"URL not allowed: {url}")

        parsed = urlparse(url)

        # Only HTTPS
        if parsed.scheme != 'https':
            raise SecurityError("Only HTTPS allowed")

        # Block private IPs
        if self._is_private_address(parsed.hostname):
            raise SecurityError("Private IP addresses not allowed")

        return url

    def _is_private_address(self, host: str) -> bool:
        try:
            # Resolve DNS
            ip_addresses = socket.getaddrinfo(host, None)

            for ip_info in ip_addresses:
                ip_str = ip_info[4][0]
                ip = ipaddress.ip_address(ip_str)

                if blocked_address(ip):
                    return True

            return False
        except Exception:
            return True

Why this works:

  • Redirect blocking: allow_redirects=False prevents redirect-based SSRF (e.g., example.com/redirect?to=localhost:6379)
  • Strict domain validation: Regex ^https://([a-z0-9-]+\.)*example\.com/.*$ prevents typosquatted lookalikes (examp1e.com) and null byte injection (example.com%00.attacker.com)
  • Protocol restriction: HTTPS-only blocks file://, ftp:// and gopher://
  • DNS rebinding defense: socket.getaddrinfo + IP checks reject a bad host before any socket opens, and secure_session() repeats the check on every address it dials, so a second DNS answer is inspected on the same terms as the first
  • Defense-in-depth: the session carries trust_env = False, so an https_proxy variable in the environment cannot route the request through a proxy that resolves the host itself; the 10s timeout caps each connect and each wait for data rather than the whole exchange; allow_redirects=False is set on the call, since requests has no session-level switch for it

URL Validator Class

# SECURE - Reusable URL validator. The framework examples below import
# this class as url_validator.

from urllib.parse import urlparse
import ipaddress
import socket
from typing import Set

# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address

class UrlValidator:
    def __init__(
        self,
        allowed_schemes: Set[str],
        allowed_hosts: Set[str],
        block_private_ips: bool = True
    ):
        self.allowed_schemes = {s.lower() for s in allowed_schemes}
        self.allowed_hosts = {h.lower() for h in allowed_hosts}
        self.block_private_ips = block_private_ips

    def validate(self, url: str) -> str:
        try:
            parsed = urlparse(url)
        except Exception:
            raise ValueError("Invalid URL")

        # Validate scheme
        if parsed.scheme.lower() not in self.allowed_schemes:
            raise ValueError(f"Scheme not allowed: {parsed.scheme}")

        # Validate host
        host = parsed.hostname
        if not host:
            raise ValueError("No host in URL")

        host_lower = host.lower()

        if not self._is_host_allowed(host_lower):
            raise ValueError(f"Host not allowed: {host}")

        # Block private IPs
        if self.block_private_ips and self._is_private_ip(host):
            raise ValueError("Private IP addresses not allowed")

        # Block localhost variants
        if self._is_localhost(host_lower):
            raise ValueError("Localhost not allowed")

        return url

    def _is_host_allowed(self, host: str) -> bool:
        # Exact match
        if host in self.allowed_hosts:
            return True

        # Wildcard subdomain match (*.example.com)
        for allowed_host in self.allowed_hosts:
            if allowed_host.startswith('*.'):
                if host.endswith(allowed_host[1:]):
                    return True

        return False

    def _is_private_ip(self, host: str) -> bool:
        try:
            # Resolve hostname
            ip_addresses = socket.getaddrinfo(host, None)

            for ip_info in ip_addresses:
                ip_str = ip_info[4][0]
                ip = ipaddress.ip_address(ip_str)

                if blocked_address(ip):
                    return True

                # AWS metadata endpoint
                if self._is_aws_metadata(ip):
                    return True

                # Docker internal network
                if self._is_docker_internal(ip):
                    return True

            return False
        except Exception:
            return True

    def _is_aws_metadata(self, ip: ipaddress.IPv4Address) -> bool:
        return str(ip) == '169.254.169.254'

    def _is_docker_internal(self, ip: ipaddress.IPv4Address) -> bool:
        # Docker default bridge: 172.17.0.0/16
        return str(ip).startswith('172.17.')

    def _is_localhost(self, host: str) -> bool:
        return host in ('localhost', '127.0.0.1', '::1', '0.0.0.0')

# Usage

validator = UrlValidator(
    allowed_schemes={'https'},
    allowed_hosts={'api.example.com', '*.cdn.example.com'},
    block_private_ips=True
)

safe_url = validator.validate(user_input)

Why this works:

  • Centralized reusable class: Constructor-based configuration (allowed schemes, hosts, private IP blocking) enables consistent security across requests, urllib, httpx, async clients
  • Flexible domain matching: Wildcard subdomains (*.cdn.example.com) with case-normalization (host.lower()) prevents bypass via mixed-case
  • IP detection: Catches localhost variants (127.0.0.1, ::1, 0.0.0.0), AWS metadata (169.254.169.254), Docker networks (172.17.x), reserved, multicast, and shared-address ranges
  • DNS rebinding defense: Pre-resolves hostnames with socket.getaddrinfo() before validation; pair this with connection pinning or dial-time validation to avoid TOCTOU races
  • OOP benefits: Enables unit testing (mock DNS), per-environment config, framework integration (Django/Flask/FastAPI) without duplicating logic

DNS Rebinding Attack Prevention

DNS rebinding is a bypass technique where attackers:

  1. Create a domain that initially resolves to a legitimate IP
  2. Application validates the IP (passes allowlist)
  3. Attacker changes DNS to point to internal IP (127.0.0.1, 192.168.x.x)
  4. Application makes request using cached DNS or re-resolves
  5. Request goes to internal service

Protection against DNS rebinding:

# Validate hostname, validate resolved addresses, then pin or revalidate at connection time

from urllib.parse import urlparse
import socket
import ipaddress

# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address

def validate_before_and_after_dns(url: str, allowed_domains: set) -> str:
    """Validate URL before a request; pair with connection pinning for rebinding defense."""
    parsed = urlparse(url)
    hostname = parsed.hostname

    # Step 1: Validate hostname against allowlist
    if hostname not in allowed_domains:
        raise SecurityException("Domain not allowed")

    # Step 2: Resolve DNS and validate ALL resolved IPs
    try:
        ip_addresses = socket.getaddrinfo(hostname, None)
        for ip_info in ip_addresses:
            ip_str = ip_info[4][0]
            ip = ipaddress.ip_address(ip_str)

            if blocked_address(ip):
                raise SecurityException(f"Domain resolves to private IP: {ip_str}")

            # Check for AWS metadata endpoint
            if str(ip) == '169.254.169.254':
                raise SecurityException("Access to AWS metadata endpoint blocked")
    except socket.gaierror:
        raise SecurityException("Cannot resolve hostname")

    return url

class SecurityException(Exception):
    pass

This is necessary and not sufficient, and the gap is the whole attack. The function returns the URL, and requests.get(url) then performs its own DNS lookup. Nothing connects the address that was validated to the address that is connected to, so a second answer with a short TTL wins the race. Use it to reject obviously bad destinations early, then close the race below.

Validating the address actually connected to

Resolve and validate every candidate before opening a socket, then connect to a validated numeric address. Checking only getpeername() is too late to prevent TCP port probing: even a connection closed before HTTP is sent reaches the target.

# SECURE - validate all candidates before connecting to a checked numeric address

import ipaddress
import socket

import requests
from requests.adapters import HTTPAdapter
from urllib3.connection import HTTPConnection, HTTPSConnection
from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
from urllib3.poolmanager import PoolManager
from urllib3.util.connection import create_connection
from urllib3.exceptions import ConnectTimeoutError, NameResolutionError, NewConnectionError

# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address


class SecurityException(Exception):
    pass


def _new_public_connection(connection):
    try:
        candidates = socket.getaddrinfo(
            connection._dns_host, connection.port, type=socket.SOCK_STREAM
        )
    except socket.gaierror as error:
        raise NameResolutionError(connection.host, connection, error) from error

    if not candidates:
        raise SecurityException("Host has no usable addresses")
    for candidate in candidates:
        if blocked_address(ipaddress.ip_address(candidate[4][0])):
            raise SecurityException("Destination blocked")

    # Keep the connection's hostname for HTTP Host, TLS SNI and certificate
    # verification. Only the socket destination changes to a numeric address.
    last_error = None
    for candidate in candidates:
        try:
            sock = create_connection(
                (candidate[4][0], connection.port), connection.timeout,
                source_address=connection.source_address,
                socket_options=connection.socket_options,
            )
        except OSError as error:
            last_error = error
            continue
        _reject_non_global(sock)
        return sock

    if isinstance(last_error, socket.timeout):
        raise ConnectTimeoutError(connection, "Connection timed out") from last_error
    raise NewConnectionError(connection, "Connection failed") from last_error


def _reject_non_global(sock):
    # getpeername() is the address this socket is connected to - not a name
    # that could resolve differently a moment later
    ip = ipaddress.ip_address(sock.getpeername()[0])
    if blocked_address(ip):
        sock.close()
        raise SecurityException(f"Blocked connection to non-global address {ip}")


class _ValidatingHTTPConnection(HTTPConnection):
    def _new_conn(self):
        return _new_public_connection(self)


class _ValidatingHTTPSConnection(HTTPSConnection):
    def _new_conn(self):
        return _new_public_connection(self)


class _ValidatingHTTPPool(HTTPConnectionPool):
    ConnectionCls = _ValidatingHTTPConnection


class _ValidatingHTTPSPool(HTTPSConnectionPool):
    ConnectionCls = _ValidatingHTTPSConnection


class _ValidatingPoolManager(PoolManager):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.pool_classes_by_scheme = {
            "http": _ValidatingHTTPPool,
            "https": _ValidatingHTTPSPool,
        }


class ValidatingAdapter(HTTPAdapter):
    """Rejects any connection whose peer address is not globally routable."""

    def init_poolmanager(self, connections, maxsize, block=False, **kwargs):
        self.poolmanager = _ValidatingPoolManager(
            num_pools=connections, maxsize=maxsize, block=block, **kwargs
        )


def secure_session() -> requests.Session:
    """A Session that dials only checked addresses.

    The other examples on this page import it as ssrf_transport. Keep the
    allowlist check in front of it: this establishes that an address is
    public, not that it is one you meant to talk to.
    """
    session = requests.Session()
    # trust_env=False: a Session reads http_proxy/https_proxy from the
    # environment by default, and through a proxy the socket _new_conn opens
    # is to the proxy - getpeername() reports the proxy's address, and the
    # target is resolved there
    session.trust_env = False
    session.mount("http://", ValidatingAdapter())
    session.mount("https://", ValidatingAdapter())
    return session


def read_capped(response, limit: int = 10 * 1024 * 1024) -> bytes:
    """Read at most `limit` bytes of a response opened with stream=True.

    requests' timeout is a connect wait and a per-read wait, not an overall
    deadline: a peer that sends one byte every few seconds keeps an unbounded
    download alive for as long as it likes. The cap is what bounds the work.
    """
    body = response.raw.read(limit + 1, decode_content=True)
    if len(body) > limit:
        raise SecurityException("Response too large")
    return body


# Usage
with secure_session() as session:
    response = session.get(validated_url, timeout=10, allow_redirects=False, stream=True)
    body = read_capped(response)

Why this works:

  • The socket receives a checked numeric address. Every DNS candidate is validated before any connection attempt; the numeric destination cannot acquire a different DNS answer. The peer check is an additional assertion.
  • It covers redirects and retries too. Every connection the session opens goes through _new_conn, so a redirect that the client follows is checked on the same terms as the original request - a validated URL that redirects to an internal address is stopped at connect, not at parse.
  • It fails closed before connecting. A blocked DNS candidate rejects the whole answer set, preventing both HTTP requests and TCP probing of that target.
  • trust_env = False keeps the check on the target. requests honours http_proxy, https_proxy and all_proxy from the environment unless told not to, and a proxied connection is to the proxy: getpeername() returns a public, acceptable address, the request line carries the target hostname, and the proxy resolves it where none of this ran. A variable set in a container image is enough. Where a proxy is required for egress, the destination control has to live on the proxy.
  • read_capped bounds the body. timeout=10 is a connect wait and a per-read wait; a host that keeps sending keeps the download open. The cap, not the timeout, is what limits the work an allowlisted host can cause.

Assert both directions, because a guard that blocks everything looks identical to one that works:

  internal http (must block)     -> Destination blocked; listener accepts no connection
  legitimate https (must work)   -> 200

Keep the allowlist and scheme checks in front of this. Dial-time validation tells you the address is publicly routable; it does not tell you the host is one you intended to talk to, and only the allowlist does that.

Framework-Specific Guidance

Django

# SECURE - Django view with URL validation

import logging

from django.http import JsonResponse, HttpResponseBadRequest
from django.views.decorators.http import require_http_methods

# The validator class from the section above, as its own module
from url_validator import UrlValidator
# The pinned session from "Validating the address actually connected to"
from ssrf_transport import secure_session, read_capped

logger = logging.getLogger(__name__)

# Create validator instance

url_validator = UrlValidator(
    allowed_schemes={'https'},
    allowed_hosts={'api.example.com'},
    block_private_ips=True
)

@require_http_methods(["GET"])
def proxy_view(request):
    url = request.GET.get('url')

    if not url:
        return HttpResponseBadRequest("URL parameter required")

    try:
        # Validate URL
        validated_url = url_validator.validate(url)

        # The session dials only the addresses it has checked and takes no
        # proxy from the environment; the body is capped because timeout=10
        # bounds each wait, not the whole download
        with secure_session() as session:
            response = session.get(validated_url, timeout=10, allow_redirects=False, stream=True)
            body = read_capped(response)

        return JsonResponse({
            'status': response.status_code,
            'content': body.decode(response.encoding or 'utf-8', errors='replace')
        })

    except ValueError:
        return HttpResponseBadRequest("Invalid URL")
    except Exception as e:
        # The exception names the address that was refused or the host that
        # did not answer - a map of the internal network, one request at a
        # time. It goes to the log; the caller gets a fixed string
        logger.warning("outbound request rejected: %s", e)
        return HttpResponseBadRequest("Request failed")

# settings.py - Configure allowed hosts

SSRF_ALLOWED_HOSTS = [
    'api.example.com',
    '*.cdn.example.com'
]

Flask

# SECURE - Flask with URL validation

import logging

from flask import Flask, request, jsonify
# The validator class from the section above, as its own module
from url_validator import UrlValidator
# The pinned session from "Validating the address actually connected to"
from ssrf_transport import secure_session, read_capped

app = Flask(__name__)
logger = logging.getLogger(__name__)

# Initialize validator

url_validator = UrlValidator(
    allowed_schemes={'https'},
    allowed_hosts={'api.example.com', 'public-api.example.org'},
    block_private_ips=True
)

@app.route('/proxy')
def proxy():
    url = request.args.get('url')

    if not url:
        return jsonify({'error': 'URL parameter required'}), 400

    try:
        # Validate URL
        validated_url = url_validator.validate(url)

        # Pinned session, no environment proxy, capped body
        with secure_session() as session:
            response = session.get(validated_url, timeout=10, allow_redirects=False, stream=True)
            body = read_capped(response)

        return jsonify({
            'status': response.status_code,
            'content': body.decode(response.encoding or 'utf-8', errors='replace')
        })

    except ValueError:
        return jsonify({'error': 'Invalid URL'}), 400
    except Exception as e:
        logger.warning("outbound request rejected: %s", e)   # detail stays server-side
        return jsonify({'error': 'Request failed'}), 502

FastAPI

# SECURE - FastAPI with URL validation

import logging

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
# The validator class from the section above, as its own module
from url_validator import UrlValidator
# The pinned session from "Validating the address actually connected to"
from ssrf_transport import secure_session, read_capped

app = FastAPI()
logger = logging.getLogger(__name__)

# Initialize validator

url_validator = UrlValidator(
    allowed_schemes={'https'},
    allowed_hosts={'api.example.com'},
    block_private_ips=True
)

class ProxyResponse(BaseModel):
    status: int
    content: str

# def, not async def: getaddrinfo and requests both block, and a blocking call
# inside an async route stalls the event loop for every other request. FastAPI
# runs a plain def in a worker thread. An async route needs an async client
# whose connector enforces the same policy - see the aiohttp example below
@app.get("/proxy", response_model=ProxyResponse)
def proxy(url: str = Query(..., description="URL to fetch")):
    try:
        # Validate URL
        validated_url = url_validator.validate(url)

        # Pinned session, no environment proxy, capped body
        with secure_session() as session:
            response = session.get(validated_url, timeout=10, allow_redirects=False, stream=True)
            body = read_capped(response)

        return ProxyResponse(
            status=response.status_code,
            content=body.decode(response.encoding or 'utf-8', errors='replace')
        )

    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid URL")
    except Exception as e:
        logger.warning("outbound request rejected: %s", e)   # detail stays server-side
        raise HTTPException(status_code=502, detail="Request failed")

aiohttp (Async)

# SECURE - aiohttp with URL validation and a resolver that checks every address

import ipaddress
import logging
import socket

import aiohttp
from aiohttp import web
from aiohttp.abc import AbstractResolver
from aiohttp.resolver import DefaultResolver
# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address
# The validator class from the section above, as its own module
from url_validator import UrlValidator

logger = logging.getLogger(__name__)
MAX_BODY = 10 * 1024 * 1024

url_validator = UrlValidator(
    allowed_schemes={'https'},
    allowed_hosts={'api.example.com'},
    block_private_ips=True
)


class ValidatingResolver(AbstractResolver):
    """aiohttp connects to whatever its resolver returns, so checking here is
    the pin: there is no later lookup for a second DNS answer to win. Measured
    on aiohttp 3.14, a URL whose host is an IP literal never reaches the
    resolver, which is why UrlValidator - with its own literal check - stays in
    front of it.
    """

    def __init__(self):
        self._inner = DefaultResolver()

    async def resolve(self, host, port=0, family=socket.AF_UNSPEC):
        results = await self._inner.resolve(host, port, family)
        for result in results:
            if blocked_address(ipaddress.ip_address(result['host'])):
                raise OSError(f"Destination blocked for {host}")
        return results

    async def close(self):
        await self._inner.close()


async def proxy_handler(request):
    url = request.query.get('url')

    if not url:
        return web.Response(text='URL parameter required', status=400)

    try:
        # Validate URL
        validated_url = url_validator.validate(url)

        # ClientSession takes no proxy from the environment unless trust_env=True
        # is passed, so there is nothing to switch off here
        timeout = aiohttp.ClientTimeout(total=10)
        connector = aiohttp.TCPConnector(resolver=ValidatingResolver())
        async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
            async with session.get(validated_url, allow_redirects=False) as response:
                body = await response.content.read(MAX_BODY + 1)
                if len(body) > MAX_BODY:
                    return web.Response(text='Request failed', status=502)

                return web.json_response({
                    'status': response.status,
                    # charset from Content-Type, or a default: get_encoding() would
                    # try to sniff a body it has not been given
                    'content': body.decode(response.charset or 'utf-8', errors='replace')
                })

    except ValueError:
        return web.Response(text='Invalid URL', status=400)
    except aiohttp.ClientError as e:
        logger.warning("outbound request rejected: %s", e)   # detail stays server-side
        return web.Response(text='Request failed', status=502)

app = web.Application()
app.router.add_get('/proxy', proxy_handler)

Protecting Cloud Metadata Endpoints

# SECURE - Block AWS/Azure/GCP metadata endpoints

import ipaddress
import socket
from urllib.parse import urlparse

# The address policy from the top of this page, kept in one module
from ssrf_policy import blocked_address

class MetadataProtection:
    BLOCKED_HOSTS = {
        '169.254.169.254',           # AWS/Azure metadata
        'metadata.google.internal',  # GCP metadata
        'metadata'
    }

    BLOCKED_PATHS = {
        '/latest/meta-data',
        '/latest/user-data',
        '/latest/dynamic',
        '/computeMetadata/v1',
        '/metadata/instance'
    }

    def validate_not_metadata(self, url: str):
        parsed = urlparse(url)

        host = parsed.hostname.lower() if parsed.hostname else ''
        path = parsed.path

        # Block metadata service hostnames
        if host in self.BLOCKED_HOSTS:
            raise ValueError("Access to metadata service blocked")

        # Block metadata paths
        for blocked_path in self.BLOCKED_PATHS:
            if path.startswith(blocked_path):
                raise ValueError("Access to metadata endpoint blocked")

        # Every answer goes through the one policy. A local is_link_local test
        # here would skip the IPv6 spellings of the same address: measured on
        # CPython 3.13, 64:ff9b:1::a9fe:a9fe and ::a9fe:a9fe are both
        # is_link_local False, and both spell 169.254.169.254
        try:
            ip_addresses = socket.getaddrinfo(host, None)
        except socket.gaierror:
            raise ValueError("DNS resolution failed")

        for ip_info in ip_addresses:
            ip = ipaddress.ip_address(ip_info[4][0])
            if blocked_address(ip):
                raise ValueError("Non-public address blocked")

Common Pitfalls

  • Resolving and validating a hostname with socket.gethostbyname(), then calling requests.get(url) - requests/urllib3 performs its own DNS resolution when opening the connection, so the validated address and the connected address aren't guaranteed to match. Validate every candidate before connecting to a checked numeric address, as ValidatingAdapter in Validating the address actually connected to does. A peer check alone runs after the TCP connection and cannot prevent port probing.
  • Leaving requests's default allow_redirects=True after validating only the initial URL - each redirect in the chain is followed automatically without being re-checked against the allowlist, so allow_redirects=False plus a manual per-hop validation loop is needed, not a one-time check before the request.
  • Mounting ValidatingAdapter and leaving Session.trust_env at its default - requests reads http_proxy/https_proxy/all_proxy from the environment and sends the request through that proxy, so the peer _new_conn inspects is the proxy and the target is resolved at the other end. session.trust_env = False disables it, along with .netrc lookup.
  • Wrapping the check in try: ipaddress.ip_address(host) and only validating when that succeeds - ipaddress.ip_address() raises ValueError for anything that isn't a literal IP, so a hostname (the common case) skips the private/link-local check entirely unless the code explicitly resolves it first and validates the resolved address instead of the original host string.

Additional Resources