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

Language: Python
Severity: High
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-health', methods=['POST'])
def check_device_status():
    device_endpoint = request.json.get('status_url')
    auth_token = request.json.get('token', '')
    headers = {'Authorization': f'Bearer {auth_token}'}
    try:
        response = requests.get(device_endpoint, headers=headers, timeout=5)
        return jsonify({'status': 'online', 'data': response.text, 'code': response.status_code})
    except requests.RequestException as e:
        return jsonify({'status': 'offline', 'error': str(e)}), 500
```

## Explanation

The application accepts a user-controlled URL via 'status_url' parameter and directly passes it to requests.get() without validation. This allows attackers to make the server send HTTP requests to arbitrary internal or external endpoints, enabling SSRF attacks to access internal services, cloud metadata endpoints, or perform port scanning.

## Remediation

The fix validates the user-supplied URL by enforcing HTTPS-only scheme, restricting allowed ports, resolving the hostname to its IP address, and checking the IP against blocked private/internal ranges (including cloud metadata at 169.254.x.x) while allowing explicitly configured IoT device subnets. Additionally, redirects are disabled to prevent redirect-based SSRF bypasses, response size is limited, and error messages are sanitized to avoid leaking internal information.

## 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, 8080}
# Define allowed external subnets for IoT devices (example: specific subnets)
ALLOWED_DEVICE_SUBNETS = [
    ipaddress.ip_network('192.168.100.0/24'),  # Example IoT device subnet
    ipaddress.ip_network('10.0.50.0/24'),       # Example IoT device subnet
]

BLOCKED_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_allowed(ip_str):
    """Check if the resolved IP is in allowed device subnets and not in blocked ranges."""
    try:
        ip = ipaddress.ip_address(ip_str)
    except ValueError:
        return False

    # Check against blocked/private ranges
    for blocked in BLOCKED_RANGES:
        if ip in blocked:
            # Even if blocked, check if it's in an explicitly allowed device subnet
            for allowed in ALLOWED_DEVICE_SUBNETS:
                if ip in allowed:
                    return True
            return False

    return True


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

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

    # Validate scheme
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False, f"Scheme '{parsed.scheme}' not allowed. Use HTTPS."

    # Ensure hostname exists
    hostname = parsed.hostname
    if not hostname:
        return False, "No hostname in URL"

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

    # Resolve hostname and validate IP
    try:
        resolved_ips = socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror:
        return False, "Could not resolve hostname"

    if not resolved_ips:
        return False, "No addresses resolved for hostname"

    for addr_info in resolved_ips:
        ip_str = addr_info[4][0]
        if not is_ip_allowed(ip_str):
            return False, f"Target IP address is not allowed"

    return True, "Valid"


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

    device_endpoint = request.json.get('status_url')
    auth_token = request.json.get('token', '')

    # Validate the device URL before making the request
    is_valid, validation_message = validate_device_url(device_endpoint)
    if not is_valid:
        return jsonify({'status': 'error', 'error': f'Invalid device URL: {validation_message}'}), 400

    headers = {'Authorization': f'Bearer {auth_token}'}
    try:
        response = requests.get(
            device_endpoint,
            headers=headers,
            timeout=5,
            allow_redirects=False  # Prevent redirect-based SSRF bypass
        )
        # Limit response size to prevent memory issues
        response_data = response.text[:10000]
        return jsonify({'status': 'online', 'data': response_data, 'code': response.status_code})
    except requests.RequestException as e:
        return jsonify({'status': 'offline', 'error': 'Device unreachable'}), 500
```