SSRF: validate the resolved IP, not the URL string

Vulnerable

import requests

def fetch_preview(url: str) -> str:
    if "localhost" in url or "127.0.0.1" in url:
        raise ValueError("blocked")
    return requests.get(url, timeout=5).text

# http://169.254.169.254/latest/meta-data/iam/security-credentials/
# http://2130706433/  (decimal for 127.0.0.1)
# http://attacker.tld/  -> 302 -> http://169.254.169.254/

Secure

import ipaddress
import socket
from urllib.parse import urlsplit

import requests

ALLOWED_SCHEMES = {"http", "https"}
ALLOWED_PORTS = {80, 443}


def _resolve_public(host: str) -> str:
    """Every A/AAAA record must be public; return one verified address."""
    infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
    addresses = {info[4][0] for info in infos}
    if not addresses:
        raise ValueError("unresolvable host")
    for address in addresses:
        ip = ipaddress.ip_address(address)
        if ip.version == 6 and ip.ipv4_mapped:
            ip = ip.ipv4_mapped
        if (ip.is_private or ip.is_loopback or ip.is_link_local
                or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
            raise ValueError(f"blocked address {ip}")
    return sorted(addresses)[0]


def fetch_preview(url: str) -> str:
    parts = urlsplit(url)
    if parts.scheme not in ALLOWED_SCHEMES or not parts.hostname:
        raise ValueError("blocked scheme")
    if (parts.port or (443 if parts.scheme == "https" else 80)) not in ALLOWED_PORTS:
        raise ValueError("blocked port")

    verified_ip = _resolve_public(parts.hostname)

    # Connect to the address we checked, keeping the Host header and TLS SNI
    # intact. Re-resolving here is what DNS rebinding attacks rely on.
    target = parts._replace(netloc=verified_ip if parts.port is None
                            else f"{verified_ip}:{parts.port}").geturl()
    response = requests.get(
        target,
        headers={"Host": parts.netloc},
        allow_redirects=False,          # re-run the whole check per hop instead
        timeout=5,
        stream=True,
    )
    if response.status_code in {301, 302, 303, 307, 308}:
        raise ValueError("redirects are not followed")
    return response.raw.read(1_000_000, decode_content=True).decode("utf-8", "replace")

Vulnerable

const res = await fetch(req.body.webhookUrl);   // follows redirects by default

Secure

const dns = require("node:dns/promises");
const net = require("node:net");
const ipaddr = require("ipaddr.js");

const BLOCKED = ["unspecified", "private", "loopback", "linkLocal",
                 "uniqueLocal", "reserved", "carrierGradeNat", "broadcast"];

async function assertPublic(hostname) {
  const records = await dns.lookup(hostname, { all: true, verbatim: true });
  if (records.length === 0) throw new Error("unresolvable");
  for (const { address } of records) {
    let ip = ipaddr.parse(address);
    if (ip.kind() === "ipv6" && ip.isIPv4MappedAddress()) ip = ip.toIPv4Address();
    if (BLOCKED.includes(ip.range())) throw new Error(`blocked ${address}`);
  }
  return records[0].address;
}

const url = new URL(req.body.webhookUrl);
if (!["http:", "https:"].includes(url.protocol)) throw new Error("blocked scheme");
if (net.isIP(url.hostname) === 0) await assertPublic(url.hostname);
else await assertPublic(url.hostname);

const res = await fetch(url, {
  redirect: "manual",                     // fetch follows redirects unless told not to
  signal: AbortSignal.timeout(5000),
});