# SSRF via requests.get() on Untrusted URL

Language: Python
Severity: High
CWE: CWE-918

## Source
8

## Flow
8-14

## Sink
14

## 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_callback_url = request.json.get('callback_endpoint')
    device_id = request.json.get('device_id', 'unknown')
    
    if not device_callback_url:
        return jsonify({'error': 'callback_endpoint required'}), 400
    
    try:
        health_response = requests.get(device_callback_url, timeout=5)
        device_status = health_response.json()
        
        return jsonify({
            'device_id': device_id,
            'status': 'reachable',
            'health_data': device_status,
            'http_code': health_response.status_code
        }), 200
    except Exception as e:
        return jsonify({
            'device_id': device_id,
            'status': 'unreachable',
            'error': str(e)
        }), 500
```

## Explanation

The application accepts a user-controlled callback_endpoint URL without validation and directly passes it to requests.get(), enabling Server-Side Request Forgery (SSRF). An attacker can force the server to make requests to internal services, cloud metadata endpoints, or other restricted resources.

## Remediation

The fix adds a URL validation function that checks the scheme, port, and resolves the hostname to ensure it does not point to private, loopback, link-local, or reserved IP addresses (blocking access to cloud metadata endpoints like 169.254.169.254). Additionally, redirects are disabled to prevent redirect-based SSRF bypasses, and error messages are sanitized to avoid leaking internal network 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 schemes and optional domain allowlist
ALLOWED_SCHEMES = {'http', 'https'}
ALLOWED_PORTS = {80, 443, 8080, 8443}
# Optional: restrict to known device network ranges or domains
# ALLOWED_DOMAINS = {'devices.example.com', 'iot.example.com'}

def is_private_ip(hostname):
    """Check if a hostname resolves to a private/reserved IP address."""
    try:
        resolved_ips = socket.getaddrinfo(hostname, None)
        for entry in resolved_ips:
            ip = ipaddress.ip_address(entry[4][0])
            if (ip.is_private or ip.is_loopback or ip.is_reserved or
                ip.is_multicast or ip.is_link_local or
                ip.is_unspecified):
                return True
    except (socket.gaierror, ValueError):
        return True  # If we can't resolve, deny by default
    return False


def validate_callback_url(url):
    """Validate that the callback URL is safe to request."""
    try:
        parsed = urlparse(url)
    except Exception:
        return False, "Invalid URL format"

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

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

    # Check port if specified
    port = parsed.port
    if port and port not in ALLOWED_PORTS:
        return False, f"Port {port} not allowed"

    # Block IP-based URLs that resolve to private/internal addresses
    if is_private_ip(hostname):
        return False, "URL resolves to a restricted internal address"

    # Optional: enforce domain allowlist
    # if hostname not in ALLOWED_DOMAINS:
    #     return False, f"Domain '{hostname}' not in allowed device domains"

    return True, None


@app.route('/api/iot/device/health', methods=['POST'])
def check_device_health():
    device_callback_url = request.json.get('callback_endpoint')
    device_id = request.json.get('device_id', 'unknown')

    if not device_callback_url:
        return jsonify({'error': 'callback_endpoint required'}), 400

    # Validate the callback URL before making the request
    is_valid, validation_error = validate_callback_url(device_callback_url)
    if not is_valid:
        return jsonify({
            'device_id': device_id,
            'status': 'rejected',
            'error': f'Invalid callback URL: {validation_error}'
        }), 400

    try:
        health_response = requests.get(
            device_callback_url,
            timeout=5,
            allow_redirects=False  # Prevent redirect-based SSRF bypass
        )
        device_status = health_response.json()

        return jsonify({
            'device_id': device_id,
            'status': 'reachable',
            'health_data': device_status,
            'http_code': health_response.status_code
        }), 200
    except Exception as e:
        return jsonify({
            'device_id': device_id,
            'status': 'unreachable',
            'error': 'Device health check failed'
        }), 500
```