# Server-Side Request Forgery (SSRF) via Unvalidated Internal Metadata URL Fetch

Language: Python
Severity: Critical
CWE: CWE-918

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
import requests
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/cloud/inspect', methods=['POST'])
def fetch_instance_diagnostics():
    payload = request.get_json()
    svc_endpoint = payload.get('diagnostics_url')
    auth_hdr = payload.get('bearer_token', '')
    hdrs = {'Authorization': f'Bearer {auth_hdr}', 'Metadata': 'true'}
    resp = requests.get(svc_endpoint, headers=hdrs, timeout=5)
    diag_data = resp.json()
    return jsonify({'status': 'ok', 'diagnostics': diag_data})
```

## Explanation

The application accepts a user-controlled URL ('diagnostics_url') from the POST request body without any validation or allowlist checking. This URL is directly passed to 'requests.get()', enabling Server-Side Request Forgery (SSRF). An attacker can supply internal cloud metadata URLs (e.g., http://169.254.169.254/latest/meta-data/iam/security-credentials/) along with a crafted bearer token, causing the server to fetch and return sensitive IAM credentials or other internal resources.

## Remediation

The fix implements a multi-layered SSRF defense: (1) a strict URL scheme allowlist limiting requests to HTTPS only, (2) a hostname allowlist that only permits pre-approved diagnostic service domains, (3) DNS resolution of the target hostname with validation against blocked IP ranges (link-local 169.254.x.x, loopback, RFC1918 private ranges) to prevent DNS rebinding attacks, (4) removal of the dangerous 'Metadata: true' header that could facilitate cloud metadata exfiltration, and (5) disabling of redirects via allow_redirects=False to prevent redirect-based SSRF bypasses.

## Secure Code
```python
import requests
from flask import Flask, request, jsonify
from urllib.parse import urlparse
import ipaddress
import socket

app = Flask(__name__)

# Allowlist of permitted diagnostic endpoint hostnames/domains
ALLOWED_DIAGNOSTIC_HOSTS = {
    "diagnostics.example.com",
    "monitoring.internal.example.com",
}

# Blocked IP ranges (link-local, loopback, private, etc.)
BLOCKED_IP_NETWORKS = [
    ipaddress.ip_network("169.254.0.0/16"),   # Link-local / cloud metadata
    ipaddress.ip_network("127.0.0.0/8"),       # Loopback
    ipaddress.ip_network("10.0.0.0/8"),        # Private
    ipaddress.ip_network("172.16.0.0/12"),     # Private
    ipaddress.ip_network("192.168.0.0/16"),    # Private
    ipaddress.ip_network("::1/128"),           # IPv6 loopback
    ipaddress.ip_network("fc00::/7"),          # IPv6 unique local
    ipaddress.ip_network("fe80::/10"),         # IPv6 link-local
]

ALLOWED_SCHEMES = {"https"}


def is_url_safe(url):
    """Validate that the URL is safe to fetch: scheme allowlist, hostname
    allowlist, and resolved-IP blocklist to prevent SSRF."""
    try:
        parsed = urlparse(url)
    except Exception:
        return False, "Malformed URL"

    # 1. Scheme check
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False, f"Scheme '{parsed.scheme}' is not allowed. Use HTTPS."

    hostname = parsed.hostname
    if not hostname:
        return False, "No hostname found in URL"

    # 2. Hostname allowlist check
    if hostname not in ALLOWED_DIAGNOSTIC_HOSTS:
        return False, f"Host '{hostname}' is not in the allowed diagnostics hosts"

    # 3. Resolve hostname and check against blocked IP ranges
    try:
        resolved_ips = socket.getaddrinfo(hostname, parsed.port or 443)
    except socket.gaierror:
        return False, f"Unable to resolve hostname '{hostname}'"

    for family, socktype, proto, canonname, sockaddr in resolved_ips:
        ip = ipaddress.ip_address(sockaddr[0])
        for blocked_net in BLOCKED_IP_NETWORKS:
            if ip in blocked_net:
                return False, f"Resolved IP {ip} falls within a blocked network range"

    return True, None


@app.route('/cloud/inspect', methods=['POST'])
def fetch_instance_diagnostics():
    payload = request.get_json()
    if not payload:
        return jsonify({'status': 'error', 'message': 'Invalid JSON body'}), 400

    svc_endpoint = payload.get('diagnostics_url')
    if not svc_endpoint or not isinstance(svc_endpoint, str):
        return jsonify({'status': 'error', 'message': 'Missing or invalid diagnostics_url'}), 400

    # Validate URL against allowlist and blocked IP ranges
    is_safe, rejection_reason = is_url_safe(svc_endpoint)
    if not is_safe:
        return jsonify({'status': 'error', 'message': f'URL rejected: {rejection_reason}'}), 403

    auth_hdr = payload.get('bearer_token', '')
    hdrs = {'Authorization': f'Bearer {auth_hdr}'}
    # Note: 'Metadata' header intentionally removed to avoid forwarding
    # cloud-metadata headers to external services

    try:
        resp = requests.get(svc_endpoint, headers=hdrs, timeout=5, allow_redirects=False)
        resp.raise_for_status()
    except requests.RequestException as e:
        return jsonify({'status': 'error', 'message': f'Request failed: {str(e)}'}), 502

    try:
        diag_data = resp.json()
    except ValueError:
        return jsonify({'status': 'error', 'message': 'Response is not valid JSON'}), 502

    return jsonify({'status': 'ok', 'diagnostics': diag_data})
```