# Server-Side Request Forgery (SSRF)

Language: Python
Severity: Critical
CWE: CWE-918

## Source
7

## Flow
7-8-9

## Sink
9

## Vulnerable Code
```python
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.route('/iot/device/metrics', methods=['POST'])
def fetch_device_telemetry():
    dev_payload = request.get_json()
    endpoint_url = dev_payload.get('telemetry_src')
    auth_hdr = {'X-Device-Token': dev_payload.get('tok', 'none')}
    raw_resp = requests.get(endpoint_url, headers=auth_hdr, timeout=5)
    sensor_data = raw_resp.json()
    return jsonify({'status': 'ok', 'metrics': sensor_data})
```

## Explanation

The application extracts 'telemetry_src' directly from user-controlled JSON input without any validation or sanitization, then passes it directly to requests.get(). An attacker can supply any arbitrary URL — including internal cloud metadata endpoints or internal network services — causing the server to make requests on the attacker's behalf, enabling SSRF attacks that can exfiltrate cloud credentials or probe internal infrastructure.

## Remediation

The fix implements a multi-layered SSRF defense: (1) a strict allowlist of permitted telemetry host domains ensures only pre-approved sources are contacted, (2) the scheme is restricted to HTTPS only, (3) raw IP addresses in URLs are rejected outright, (4) DNS resolution results are checked against blocked IP ranges (private, loopback, link-local, metadata IPs like 169.254.169.254) to prevent DNS rebinding attacks, and (5) HTTP redirects are disabled to prevent redirect-based SSRF bypasses. Error handling is also improved to avoid leaking internal details.

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

app = Flask(__name__)

# Allowlist of permitted telemetry source domains/hosts
ALLOWED_TELEMETRY_HOSTS = {
    "telemetry.iot-platform.example.com",
    "metrics.iot-platform.example.com",
}

ALLOWED_SCHEMES = {"https"}

# Blocked IP ranges (internal, link-local, loopback, cloud metadata, etc.)
def _is_blocked_ip(ip_str):
    try:
        addr = ipaddress.ip_address(ip_str)
        return (
            addr.is_private
            or addr.is_loopback
            or addr.is_link_local
            or addr.is_reserved
            or addr.is_multicast
            or addr.is_unspecified
        )
    except ValueError:
        return True  # If it can't be parsed, block it


def _validate_telemetry_url(url):
    """Validate that the telemetry URL is safe to fetch.
    
    Returns the URL if valid, raises ValueError otherwise.
    """
    if not url or not isinstance(url, str):
        raise ValueError("Missing or invalid telemetry_src URL.")

    parsed = urlparse(url)

    # 1. Enforce HTTPS only
    if parsed.scheme not in ALLOWED_SCHEMES:
        raise ValueError(f"Scheme '{parsed.scheme}' is not allowed. Only {ALLOWED_SCHEMES} permitted.")

    hostname = parsed.hostname
    if not hostname:
        raise ValueError("No hostname found in telemetry_src URL.")

    # 2. Block raw IP addresses in the URL to prevent direct internal IP access
    try:
        ipaddress.ip_address(hostname)
        raise ValueError("IP addresses are not allowed as telemetry source hosts. Use a domain name.")
    except ValueError as e:
        if "not allowed" in str(e):
            raise
        # hostname is not an IP literal — continue with domain validation

    # 3. Check against the allowlist of permitted domains
    if hostname not in ALLOWED_TELEMETRY_HOSTS:
        raise ValueError(f"Host '{hostname}' is not in the allowed telemetry sources list.")

    # 4. Resolve the hostname and verify it doesn't point to a blocked IP
    try:
        resolved_ips = socket.getaddrinfo(hostname, parsed.port or 443)
    except socket.gaierror:
        raise ValueError(f"Could not resolve hostname '{hostname}'.")

    for family, kind, proto, canonname, sockaddr in resolved_ips:
        ip_str = sockaddr[0]
        if _is_blocked_ip(ip_str):
            raise ValueError(
                f"Host '{hostname}' resolves to a blocked/internal IP address ({ip_str}). Request denied."
            )

    return url


@app.route('/iot/device/metrics', methods=['POST'])
def fetch_device_telemetry():
    dev_payload = request.get_json()
    if not dev_payload:
        return jsonify({'status': 'error', 'message': 'Invalid JSON payload.'}), 400

    endpoint_url = dev_payload.get('telemetry_src')

    # Validate and sanitize the telemetry source URL
    try:
        validated_url = _validate_telemetry_url(endpoint_url)
    except ValueError as e:
        return jsonify({'status': 'error', 'message': str(e)}), 400

    auth_hdr = {'X-Device-Token': dev_payload.get('tok', 'none')}

    try:
        raw_resp = requests.get(
            validated_url,
            headers=auth_hdr,
            timeout=5,
            allow_redirects=False  # Prevent open-redirect-based SSRF bypass
        )
        raw_resp.raise_for_status()
        sensor_data = raw_resp.json()
    except requests.RequestException as e:
        return jsonify({'status': 'error', 'message': f'Failed to fetch telemetry: {str(e)}'}), 502

    return jsonify({'status': 'ok', 'metrics': sensor_data})
```