# Server-Side Request Forgery (SSRF)

Language: Python
Severity: Critical
CWE: CWE-918

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.route('/iot/device/telemetry', methods=['POST'])
def fetch_device_telemetry():
    dev_cfg = request.get_json()
    endpoint_url = dev_cfg.get('metrics_endpoint')
    auth_hdr = dev_cfg.get('auth_token', '')
    hdrs = {'Authorization': f'Bearer {auth_hdr}', 'Accept': 'application/json'}
    resp = requests.get(endpoint_url, headers=hdrs, timeout=5)
    telemetry_data = resp.json()
    return jsonify({'status': 'ok', 'payload': telemetry_data})
```

## Explanation

The 'metrics_endpoint' value is taken directly from user-supplied JSON (dev_cfg) on line 7 without any validation or allowlist check, and is passed as-is into requests.get() on line 10. This allows an attacker to supply any arbitrary URL, including internal metadata services or private network endpoints, causing the server to make requests on the attacker's behalf. The forwarding of the auth token in the headers further amplifies the impact.

## Remediation

The fix introduces a comprehensive URL validation function (is_safe_url) that enforces HTTPS-only schemes, checks the hostname against an explicit allowlist of trusted IoT device hosts, resolves the hostname to its IP addresses and rejects any that fall within private, loopback, link-local (including the 169.254.0.0/16 cloud metadata range), or other internal IP ranges. Additionally, allow_redirects=False is set on the request to prevent open-redirect-based SSRF bypasses, and proper input validation and error handling are added.

## 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 domains/hosts for device telemetry endpoints
ALLOWED_HOSTS = {
    # Add your trusted IoT device hosts/domains here
    # e.g., 'devices.example.com', 'telemetry.iot.example.com'
}

ALLOWED_SCHEMES = {'https'}

BLOCKED_IP_RANGES = [
    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('169.254.0.0/16'),    # Link-local / cloud metadata
    ipaddress.ip_network('100.64.0.0/10'),     # Shared address space
    ipaddress.ip_network('0.0.0.0/8'),         # "This" network
    ipaddress.ip_network('::1/128'),           # IPv6 loopback
    ipaddress.ip_network('fc00::/7'),          # IPv6 private
    ipaddress.ip_network('fe80::/10'),         # IPv6 link-local
]


def is_safe_url(url):
    """Validate that the URL is safe to request: scheme, host allowlist, and IP block check."""
    try:
        parsed = urlparse(url)
    except Exception:
        return False

    # Enforce allowed schemes (HTTPS only)
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False

    hostname = parsed.hostname
    if not hostname:
        return False

    # Reject URLs with credentials embedded (e.g., http://user:pass@host)
    if parsed.username or parsed.password:
        return False

    # Check against the host allowlist
    if ALLOWED_HOSTS and hostname not in ALLOWED_HOSTS:
        return False

    # Resolve the hostname and check all resolved IPs against blocked ranges
    try:
        addr_infos = socket.getaddrinfo(hostname, parsed.port or 443)
    except socket.gaierror:
        return False

    for family, _, _, _, sockaddr in addr_infos:
        ip = ipaddress.ip_address(sockaddr[0])
        for blocked_range in BLOCKED_IP_RANGES:
            if ip in blocked_range:
                return False

    return True


@app.route('/iot/device/telemetry', methods=['POST'])
def fetch_device_telemetry():
    dev_cfg = request.get_json()
    if not dev_cfg or not isinstance(dev_cfg, dict):
        return jsonify({'status': 'error', 'message': 'Invalid request body'}), 400

    endpoint_url = dev_cfg.get('metrics_endpoint')
    if not endpoint_url or not isinstance(endpoint_url, str):
        return jsonify({'status': 'error', 'message': 'Missing or invalid metrics_endpoint'}), 400

    if not is_safe_url(endpoint_url):
        return jsonify({'status': 'error', 'message': 'The provided metrics_endpoint is not allowed'}), 403

    auth_hdr = dev_cfg.get('auth_token', '')
    hdrs = {'Authorization': f'Bearer {auth_hdr}', 'Accept': 'application/json'}

    try:
        resp = requests.get(endpoint_url, headers=hdrs, timeout=5, allow_redirects=False)
        resp.raise_for_status()
        telemetry_data = resp.json()
    except requests.RequestException as e:
        return jsonify({'status': 'error', 'message': f'Failed to fetch telemetry: {str(e)}'}), 502

    return jsonify({'status': 'ok', 'payload': telemetry_data})
```