# SSRF via Unvalidated requests.get() URL Parameter

Language: Python
Severity: Critical
CWE: CWE-918

## Source
8

## Flow
8-12

## Sink
12

## Vulnerable Code
```python
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/iot/device-status', methods=['POST'])
def fetch_device_telemetry():
    device_endpoint = request.json.get('telemetry_url')
    auth_token = request.json.get('device_token', '')
    headers = {'Authorization': f'Bearer {auth_token}', 'User-Agent': 'IoT-Gateway/2.1'}
    try:
        telemetry_response = requests.get(device_endpoint, headers=headers, timeout=10)
        device_data = telemetry_response.json()
        return jsonify({
            'status': 'success',
            'temperature': device_data.get('temp'),
            'humidity': device_data.get('humidity'),
            'uptime': device_data.get('uptime')
        }), 200
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500
```

## Explanation

The application accepts a user-controlled URL via 'telemetry_url' parameter without validation and directly passes it to requests.get(). This allows attackers to perform Server-Side Request Forgery (SSRF) by specifying internal network endpoints, cloud metadata services (e.g., http://169.254.169.254/latest/meta-data/), or other sensitive internal resources.

## Remediation

The fix validates the user-supplied URL by checking the scheme (HTTPS only), port (restricted allowlist), and resolving the hostname to verify the IP address does not point to internal/metadata services or blocked network ranges. It also disables HTTP redirects to prevent redirect-based SSRF bypasses and uses specific exception handling to avoid leaking internal error details to the client.

## 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 device network ranges (configure as needed)
ALLOWED_SCHEMES = {'https'}
ALLOWED_PORTS = {443, 8443}
# Define allowed external subnets for IoT devices (example)
ALLOWED_DEVICE_SUBNETS = [
    ipaddress.ip_network('192.168.100.0/24'),  # Example IoT device subnet
    ipaddress.ip_network('10.0.50.0/24'),       # Example edge device subnet
]

# Blocked internal/metadata ranges
BLOCKED_RANGES = [
    ipaddress.ip_network('127.0.0.0/8'),
    ipaddress.ip_network('169.254.0.0/16'),
    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('0.0.0.0/8'),
    ipaddress.ip_network('::1/128'),
    ipaddress.ip_network('fc00::/7'),
    ipaddress.ip_network('fe80::/10'),
]


def is_ip_in_allowed_subnets(ip_str):
    """Check if the resolved IP is within explicitly allowed device subnets."""
    try:
        ip = ipaddress.ip_address(ip_str)
        for subnet in ALLOWED_DEVICE_SUBNETS:
            if ip in subnet:
                return True
        return False
    except ValueError:
        return False


def is_ip_blocked(ip_str):
    """Check if the resolved IP falls within blocked internal ranges."""
    try:
        ip = ipaddress.ip_address(ip_str)
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
            if not is_ip_in_allowed_subnets(ip_str):
                return True
        for blocked in BLOCKED_RANGES:
            if ip in blocked and not is_ip_in_allowed_subnets(ip_str):
                return True
        return False
    except ValueError:
        return True


def validate_device_url(url):
    """Validate the telemetry URL against SSRF protections."""
    if not url or not isinstance(url, str):
        return False, "Invalid URL provided"

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

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

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

    port = parsed.port or (443 if parsed.scheme == 'https' else 80)
    if port not in ALLOWED_PORTS:
        return False, f"Port {port} not allowed. Only {ALLOWED_PORTS} permitted."

    try:
        resolved_ips = socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP)
        if not resolved_ips:
            return False, "Could not resolve hostname"

        for result in resolved_ips:
            ip_str = result[4][0]
            if is_ip_blocked(ip_str):
                return False, f"Access to resolved IP address is forbidden"
    except socket.gaierror:
        return False, "DNS resolution failed for hostname"

    return True, "Valid"


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

    device_endpoint = request.json.get('telemetry_url')
    auth_token = request.json.get('device_token', '')

    is_valid, validation_message = validate_device_url(device_endpoint)
    if not is_valid:
        return jsonify({'status': 'error', 'message': f'Invalid device endpoint: {validation_message}'}), 400

    headers = {'Authorization': f'Bearer {auth_token}', 'User-Agent': 'IoT-Gateway/2.1'}
    try:
        telemetry_response = requests.get(
            device_endpoint,
            headers=headers,
            timeout=10,
            allow_redirects=False
        )
        device_data = telemetry_response.json()
        return jsonify({
            'status': 'success',
            'temperature': device_data.get('temp'),
            'humidity': device_data.get('humidity'),
            'uptime': device_data.get('uptime')
        }), 200
    except requests.exceptions.RequestException as e:
        return jsonify({'status': 'error', 'message': 'Failed to fetch device telemetry'}), 502
    except (ValueError, KeyError):
        return jsonify({'status': 'error', 'message': 'Invalid response from device'}), 502
```