# SSRF via Unvalidated requests.get 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('/api/iot/sensor/proxy', methods=['POST'])
def fetch_sensor_telemetry():
    device_endpoint = request.json.get('sensor_url')
    auth_token = request.json.get('token', 'default-key')
    headers = {'Authorization': f'Bearer {auth_token}'}
    sensor_data = requests.get(device_endpoint, headers=headers, timeout=5)
    return jsonify({'status': 'success', 'telemetry': sensor_data.json(), 'source': device_endpoint})
```

## Explanation

The application accepts a user-controlled 'sensor_url' parameter and directly passes it to requests.get() without any validation or allowlisting. An attacker can provide internal network URLs (e.g., http://localhost:6379, http://169.254.169.254/latest/meta-data/) to scan internal services, access cloud metadata endpoints, or exfiltrate data from restricted network segments.

## Remediation

The fix implements multiple layers of SSRF protection: a URL validation function that enforces an allowlist of permitted sensor hostnames, restricts the scheme to HTTPS only, resolves the hostname to check that the IP doesn't fall within private/internal/link-local ranges, and blocks URLs with embedded credentials. Additionally, the response no longer returns the source URL (preventing internal network topology disclosure), redirects are disabled to prevent redirect-based SSRF bypasses, and proper error handling is added.

## 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 sensor network domains/IPs
ALLOWED_SENSOR_HOSTS = [
    'sensors.example.com',
    'iot-gateway.example.com',
]

# Allowed schemes
ALLOWED_SCHEMES = ['https']

# Blocked IP ranges (private, loopback, link-local, metadata endpoints)
BLOCKED_IP_RANGES = [
    ipaddress.ip_network('127.0.0.0/8'),
    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'),
    ipaddress.ip_network('0.0.0.0/8'),
    ipaddress.ip_network('::1/128'),
    ipaddress.ip_network('fc00::/7'),
    ipaddress.ip_network('fe80::/10'),
]


def is_ip_blocked(ip_str):
    """Check if a resolved IP address falls within blocked ranges."""
    try:
        ip = ipaddress.ip_address(ip_str)
        for blocked_range in BLOCKED_IP_RANGES:
            if ip in blocked_range:
                return True
        return False
    except ValueError:
        return True


def validate_sensor_url(url):
    """Validate that the sensor URL is safe to fetch."""
    if not url or not isinstance(url, str):
        return False, "Invalid URL provided"

    try:
        parsed = urlparse(url)
    except Exception:
        return False, "Malformed URL"

    # Check scheme
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False, f"Scheme '{parsed.scheme}' not allowed. Only {ALLOWED_SCHEMES} permitted."

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

    # Check against allowlist
    if hostname not in ALLOWED_SENSOR_HOSTS:
        return False, f"Host '{hostname}' is not in the allowed sensor hosts list"

    # Resolve hostname and check IP against blocked ranges
    try:
        resolved_ips = socket.getaddrinfo(hostname, parsed.port or 443)
        for result in resolved_ips:
            ip_str = result[4][0]
            if is_ip_blocked(ip_str):
                return False, f"Resolved IP address is in a blocked range"
    except socket.gaierror:
        return False, "Unable to resolve hostname"

    # Block URLs with credentials
    if parsed.username or parsed.password:
        return False, "URLs with embedded credentials are not allowed"

    return True, None


@app.route('/api/iot/sensor/proxy', methods=['POST'])
def fetch_sensor_telemetry():
    if not request.json:
        return jsonify({'status': 'error', 'message': 'Request body must be JSON'}), 400

    device_endpoint = request.json.get('sensor_url')
    auth_token = request.json.get('token', 'default-key')

    # Validate the target URL before making the request
    is_valid, error_message = validate_sensor_url(device_endpoint)
    if not is_valid:
        return jsonify({'status': 'error', 'message': error_message}), 400

    headers = {'Authorization': f'Bearer {auth_token}'}

    try:
        sensor_data = requests.get(device_endpoint, headers=headers, timeout=5, allow_redirects=False)
        sensor_data.raise_for_status()
        return jsonify({'status': 'success', 'telemetry': sensor_data.json()})
    except requests.exceptions.RequestException as e:
        return jsonify({'status': 'error', 'message': 'Failed to fetch sensor data'}), 502
```