FlawOpen

flawopen.com/ssrf/Python

● CWE-918 · High
FlawOpen Security Research

Python SSRF: secure requests and urllib3

Python SSRF lets user URLs reach internal services or cloud metadata. Validate and pin resolved IPs, block redirects, and restrict outbound traffic.

💡 Plain English Explainer (ELI5)

Imagine a parcel desk that checks a delivery address in the public directory, then hands the job to a courier who looks the name up again and obeys forwarding slips. The second lookup or a redirect can send the parcel into a staff-only room. A Python service that checks a URL and then calls requests.get can make the same mistake: the client resolves the host again and follows redirects. Parse with urlsplit, require every resolved ip.is_global address to pass, connect to a validated IP while preserving TLS server_hostname, and set redirect=False. urllib3 lets the code make those steps explicit.

Core Concepts & Subsystem Terms

Cloud metadata endpoint (169.254.169.254)
Cloud VMs expose credentials over plain HTTP at 169.254.169.254. In the 2019 Capital One breach an SSRF reached that address and returned IAM role keys. IMDSv2 requires a PUT-issued token and a hop limit, which blocks most SSRF-based reads.
Check-then-fetch and DNS rebinding
socket.gethostbyname(host) returns one address at check time, then requests resolves the name again when it connects. A DNS server with a short TTL can answer with a public address the first time and 127.0.0.1 the second (DNS rebinding).
Redirects to internal addresses
requests follows 301 and 302 responses by default, including for POST. A public server that answers Location: http://169.254.169.254/ sends your client straight to the internal address after every check has passed.
ipaddress is_global and IPv4-mapped IPv6
is_private misses shared and reserved ranges such as 100.64.0.0/10, while is_global is True only for publicly routable addresses. Unwrap ::ffff:127.0.0.1 with ipv4_mapped before checking, and keep Python patched for CVE-2024-4032, which corrected these properties.
Pinning the validated IP (server_hostname)
Connect to the address you validated instead of the hostname: HTTPSConnectionPool(ip, server_hostname=host, assert_hostname=host) sends the right SNI and checks the certificate for the hostname, so TLS still works and no second DNS lookup happens.

Framework mappings

Step-by-Step Attack Flow

Step 1

A user supplies a URL

A link-preview, import or webhook feature accepts url in a JSON body and passes it to requests.get or requests.post.

Step 2

The server connects to an internal address

A DNS precheck does not pin the socket: the HTTP client resolves the host when it connects, so rebinding can change a public answer to a private destination. Pin the validated IP while preserving TLS hostname checks.

Step 3

The internal service answers the trusted caller

requests follows redirects by default. A public preview or webhook URL can therefore send the backend to 169.254.169.254 after the original URL was checked.

Step 4

The attacker reads secrets or pivots

The response path creates the disclosure: a Flask route that returns response.text. Return only the data the feature needs, and keep internal credentials and response bodies out of caller-controlled fetch results.

Source Code: Flaw vs. Secure Implementation

UNPATCHED FLAW
# fetch.py: both endpoints let the caller choose the destination
import ipaddress
import socket
from urllib.parse import urlparse

import requests
from flask import Flask, abort, request

app = Flask(__name__)


# any URL is fetched, including http://169.254.169.254/
@app.post("/preview")
def preview():
    return requests.get(request.json["url"], timeout=10).text[:5000]


# checks one address, then requests resolves again and follows redirects
@app.post("/webhooks/test")
def test_webhook():
    url = request.json["url"]
    host = urlparse(url).hostname or ""
    if ipaddress.ip_address(socket.gethostbyname(host)).is_private:
        abort(400)
    return {"status": requests.post(url, json={"ping": True}, timeout=10).status_code}
HARDENED SECURE PATCH
# fetch.py: validate every resolved address and connect to that IP
import ipaddress
import socket
from urllib.parse import urlsplit

import urllib3
from flask import Flask, abort, request

app = Flask(__name__)
MAX_BYTES = 1_000_000


# is_global rejects private, loopback, link-local and reserved ranges
def resolve_public(host: str, port: int) -> str:
    ips = []
    for info in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM):
        ip = ipaddress.ip_address(info[4][0])
        if ip.version == 6 and ip.ipv4_mapped:
            ip = ip.ipv4_mapped
        ips.append(ip)
    if not ips or not all(ip.is_global for ip in ips):
        abort(400, "destination is not a public address")
    return str(ips[0])


# connect to the checked IP; SNI and certificate still use the hostname
def fetch(url: str, method: str = "GET", body: bytes | None = None) -> urllib3.BaseHTTPResponse:
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https") or not parts.hostname:
        abort(400, "only http and https URLs are allowed")
    port = parts.port or (443 if parts.scheme == "https" else 80)
    ip = resolve_public(parts.hostname, port)
    timeout = urllib3.Timeout(connect=3, read=5)
    if parts.scheme == "https":
        pool = urllib3.HTTPSConnectionPool(ip, port, timeout=timeout, retries=False,
                                           server_hostname=parts.hostname, assert_hostname=parts.hostname)
    else:
        pool = urllib3.HTTPConnectionPool(ip, port, timeout=timeout, retries=False)
    target = (parts.path or "/") + (f"?{parts.query}" if parts.query else "")
    host_header = parts.hostname if parts.port is None else f"{parts.hostname}:{parts.port}"
    return pool.urlopen(method, target, body=body, headers={"Host": host_header, "Content-Type": "application/json"},
                        redirect=False, preload_content=False)


# no redirects, short timeouts, capped response size
@app.post("/preview")
def preview():
    return fetch(request.json["url"]).read(MAX_BYTES).decode("utf-8", "replace")[:5000]


@app.post("/webhooks/test")
def test_webhook():
    return {"status": fetch(request.json["url"], "POST", b'{"ping": true}').status}

Engineering & System Hardening Checklist

Sources