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

Language: Python
Severity: Critical
CWE: CWE-918

## Source
8-9

## Flow
8-9-10-11

## Sink
11

## 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_health():
    device_endpoint = request.json.get('device_url')
    health_path = request.json.get('health_check_path', '/status')
    full_url = f"{device_endpoint}{health_path}"
    response = requests.get(full_url, timeout=5)
    return jsonify({
        'device_status': response.status_code,
        'health_data': response.text,
        'response_time': response.elapsed.total_seconds()
    })
```

## Explanation

The application accepts user-controlled 'device_url' and 'health_check_path' parameters and directly concatenates them to construct a URL for requests.get() without any validation or sanitization. An attacker can manipulate these inputs to make the server send HTTP requests to arbitrary internal or external resources, enabling SSRF attacks to access internal services, cloud metadata endpoints, or perform port scanning.

## Remediation

The fix implements a multi-layered SSRF defense: it validates the URL scheme and port against allowlists, resolves the hostname to an IP address and checks it against both a blocklist of internal/reserved ranges (including cloud metadata at 169.254.x.x) and an allowlist of permitted IoT device network ranges, sanitizes the health check path to prevent injection, disables HTTP redirects to prevent redirect-based SSRF bypasses, and truncates response data to limit information exfiltration.

## Secure Code
```python
import requests
import ipaddress
from urllib.parse import urlparse
from flask import Flask, request, jsonify

app = Flask(__name__)

# Allowlist of permitted device network ranges (configure as needed)
ALLOWED_DEVICE_NETWORKS = [
    ipaddress.ip_network('192.168.1.0/24'),
    ipaddress.ip_network('10.0.1.0/24'),
]

# Blocked IP ranges (internal services, cloud metadata, loopback, link-local)
BLOCKED_NETWORKS = [
    ipaddress.ip_network('127.0.0.0/8'),
    ipaddress.ip_network('169.254.0.0/16'),
    ipaddress.ip_network('10.255.255.0/24'),
    ipaddress.ip_network('0.0.0.0/8'),
    ipaddress.ip_network('::1/128'),
    ipaddress.ip_network('fc00::/7'),
]

ALLOWED_SCHEMES = ('http', 'https')
ALLOWED_PORTS = (80, 443, 8080, 8443)


def is_safe_url(url):
    """Validate that a URL is safe to fetch (not targeting internal resources)."""
    try:
        parsed = urlparse(url)
    except Exception:
        return False, "Invalid URL format"

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

    # Validate hostname exists
    hostname = parsed.hostname
    if not hostname:
        return False, "No hostname provided 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} is not allowed"

    # Resolve hostname to IP and validate
    import socket
    try:
        resolved_ips = socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror:
        return False, f"Cannot resolve hostname: {hostname}"

    for family, type_, proto, canonname, sockaddr in resolved_ips:
        ip = ipaddress.ip_address(sockaddr[0])

        # Check against blocked networks
        for blocked in BLOCKED_NETWORKS:
            if ip in blocked:
                return False, f"Access to {ip} is blocked (internal/reserved range)"

        # Check against allowed device networks
        in_allowed = any(ip in network for network in ALLOWED_DEVICE_NETWORKS)
        if not in_allowed:
            return False, f"IP {ip} is not in the allowed device network ranges"

    return True, "URL is safe"


def sanitize_health_path(health_path):
    """Sanitize the health check path to prevent path traversal or injection."""
    if not health_path:
        return '/status'

    # Ensure path starts with /
    if not health_path.startswith('/'):
        health_path = '/' + health_path

    # Block path traversal
    if '..' in health_path:
        return None

    # Only allow alphanumeric, slashes, hyphens, underscores in path
    import re
    if not re.match(r'^[/a-zA-Z0-9_\-\.]+$', health_path):
        return None

    return health_path


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

    device_endpoint = request.json.get('device_url')
    health_path = request.json.get('health_check_path', '/status')

    if not device_endpoint:
        return jsonify({'error': 'device_url is required'}), 400

    # Sanitize health check path
    sanitized_path = sanitize_health_path(health_path)
    if sanitized_path is None:
        return jsonify({'error': 'Invalid health_check_path'}), 400

    full_url = f"{device_endpoint.rstrip('/')}{sanitized_path}"

    # Validate the full constructed URL against SSRF
    is_safe, reason = is_safe_url(full_url)
    if not is_safe:
        return jsonify({'error': f'URL validation failed: {reason}'}), 403

    try:
        response = requests.get(full_url, timeout=5, allow_redirects=False)
        return jsonify({
            'device_status': response.status_code,
            'health_data': response.text[:1024],
            'response_time': response.elapsed.total_seconds()
        })
    except requests.exceptions.RequestException as e:
        return jsonify({'error': f'Failed to reach device: {str(e)}'}), 502
```