# SSRF via requests.get() on User-Controlled URL

Language: Python
Severity: High
CWE: CWE-918

## Source
14

## Flow
14-15-7-8

## Sink
8

## Vulnerable Code
```python
import requests
import json

def fetch_iot_device_telemetry(device_endpoint, auth_token):
    headers = {'Authorization': f'Bearer {auth_token}', 'Content-Type': 'application/json'}
    try:
        telemetry_url = f"{device_endpoint}/api/v2/telemetry"
        response = requests.get(telemetry_url, headers=headers, timeout=10)
        if response.status_code == 200:
            return {'status': 'success', 'data': response.json(), 'device': device_endpoint}
        return {'status': 'error', 'message': 'Device unreachable', 'code': response.status_code}
    except requests.exceptions.RequestException as e:
        return {'status': 'error', 'message': str(e)}

device_url = request.form.get('device_url')
token = session.get('iot_token')
telemetry_data = fetch_iot_device_telemetry(device_url, token)
```

## Explanation

The application accepts a user-controlled device URL without validation and directly uses it in requests.get(), enabling Server-Side Request Forgery (SSRF). An attacker can manipulate the device_url parameter to target internal services, cloud metadata endpoints, or perform port scanning on internal infrastructure.

## Remediation

The fix adds comprehensive URL validation before making the HTTP request, including scheme enforcement (HTTPS only), DNS resolution checks against blocked internal/private IP ranges, an allowlist for registered device hosts, credential stripping prevention, and disabling of redirects to prevent redirect-based SSRF bypasses. Error messages are also sanitized to avoid leaking internal infrastructure details.

## Secure Code
```python
import requests
import json
from urllib.parse import urlparse
import ipaddress
import socket

# Allowlist of permitted device hostname patterns or registered devices
ALLOWED_DEVICE_HOSTS = set()  # Populate from database of registered devices
ALLOWED_SCHEMES = {'https'}
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 an IP address falls within blocked ranges (internal/private networks)."""
    try:
        ip = ipaddress.ip_address(ip_str)
        for network in BLOCKED_IP_RANGES:
            if ip in network:
                return True
        return False
    except ValueError:
        return True

def validate_device_url(device_endpoint):
    """Validate that the device URL is safe and points to an allowed external host."""
    if not device_endpoint:
        return False, "Device URL is required"

    try:
        parsed = urlparse(device_endpoint)
    except Exception:
        return False, "Invalid URL format"

    # Enforce HTTPS only
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False, f"Only {ALLOWED_SCHEMES} schemes are allowed"

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

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

    # Check against allowlist of registered device hosts
    if ALLOWED_DEVICE_HOSTS and hostname not in ALLOWED_DEVICE_HOSTS:
        return False, "Device host is not registered in the platform"

    # Resolve hostname and check against blocked IP ranges
    try:
        resolved_ips = socket.getaddrinfo(hostname, parsed.port or 443)
        for entry in resolved_ips:
            ip_str = entry[4][0]
            if is_ip_blocked(ip_str):
                return False, "Device URL resolves to a blocked internal address"
    except socket.gaierror:
        return False, "Unable to resolve device hostname"

    return True, None

def fetch_iot_device_telemetry(device_endpoint, auth_token):
    """Fetch telemetry from a validated IoT device endpoint."""
    # Validate the URL before making the request
    is_valid, error_message = validate_device_url(device_endpoint)
    if not is_valid:
        return {'status': 'error', 'message': f'Invalid device URL: {error_message}'}

    headers = {'Authorization': f'Bearer {auth_token}', 'Content-Type': 'application/json'}
    try:
        telemetry_url = f"{device_endpoint}/api/v2/telemetry"

        # Re-validate the constructed URL to prevent path-based bypasses
        is_valid, error_message = validate_device_url(telemetry_url.rsplit('/api/v2/telemetry', 1)[0])
        if not is_valid:
            return {'status': 'error', 'message': f'Invalid constructed URL: {error_message}'}

        response = requests.get(telemetry_url, headers=headers, timeout=10, allow_redirects=False)
        if response.status_code == 200:
            return {'status': 'success', 'data': response.json(), 'device': device_endpoint}
        return {'status': 'error', 'message': 'Device unreachable', 'code': response.status_code}
    except requests.exceptions.RequestException:
        return {'status': 'error', 'message': 'Failed to connect to device'}

device_url = request.form.get('device_url')
token = session.get('iot_token')
telemetry_data = fetch_iot_device_telemetry(device_url, token)
```