Flask Server-Side Request Forgery (SSRF) via requests

# Secure: Safe URL handling with SSRF protection in Flask

import requests
from flask import Flask, request, jsonify
from urllib.parse import urlparse
import socket
import ipaddress
import re
from marshmallow import Schema, fields, ValidationError as MarshmallowValidationError

app = Flask(__name__)

# Safe: Configuration with allowed domains
app.config['ALLOWED_DOMAINS'] = [
    'api.example.com',
    'feeds.example.com',
    'images.example.com',
    'webhook.example.com'
]

app.config['MAX_REQUEST_SIZE'] = 10 * 1024 * 1024  # 10MB
app.config['REQUEST_TIMEOUT'] = 10  # seconds

# Input validation schemas
class URLFetchSchema(Schema):
    url = fields.Url(required=True)

class WebhookSchema(Schema):
    webhook_id = fields.Str(required=True)
    payload = fields.Dict(required=True)

# Safe: URL validation and SSRF protection
def validate_url_for_ssrf(url):
    """Validate URL against SSRF attacks"""
    try:
        parsed = urlparse(url)
    except Exception:
        raise ValueError('Invalid URL format')
    
    # Validate scheme
    if parsed.scheme not in ['http', 'https']:
        raise ValueError('Only HTTP and HTTPS protocols are allowed')
    
    # Validate hostname
    if not parsed.hostname:
        raise ValueError('URL must have a valid hostname')
    
    # Check against allowlist
    if not is_allowed_domain(parsed.hostname):
        raise ValueError('Domain not allowed')
    
    # Prevent SSRF to internal networks
    check_internal_network_access(parsed.hostname)
    
    return url

def is_allowed_domain(hostname):
    """Check if domain is in allowlist"""
    allowed_domains = app.config['ALLOWED_DOMAINS']
    
    # Check exact match or subdomain
    for domain in allowed_domains:
        if hostname == domain or hostname.endswith('.' + domain):
            return True
    
    return False

def check_internal_network_access(hostname):
    """Prevent access to internal networks"""
    try:
        # Resolve hostname to IP
        ip = socket.gethostbyname(hostname)
        ip_obj = ipaddress.ip_address(ip)
        
        # Block private/internal networks
        if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local:
            raise ValueError('Requests to internal networks are not allowed')
        
        # Block specific dangerous IP ranges
        dangerous_ranges = [
            ipaddress.ip_network('169.254.0.0/16'),  # AWS metadata
            ipaddress.ip_network('10.0.0.0/8'),     # Private class A
            ipaddress.ip_network('172.16.0.0/12'),  # Private class B
            ipaddress.ip_network('192.168.0.0/16'), # Private class C
        ]
        
        for network in dangerous_ranges:
            if ip_obj in network:
                raise ValueError('IP address not allowed')
                
    except socket.gaierror:
        raise ValueError('Unable to resolve hostname')
    except ValueError as e:
        if 'not allowed' in str(e) or 'internal networks' in str(e):
            raise e
        raise ValueError('Invalid IP address')

def make_safe_request(url, method='GET', **kwargs):
    """Make a safe HTTP request with proper configurations"""
    # Set safe defaults
    safe_kwargs = {
        'timeout': app.config['REQUEST_TIMEOUT'],
        'allow_redirects': False,  # Prevent redirect SSRF
        'stream': True,  # Stream response for size checking
        'headers': kwargs.get('headers', {})
    }
    
    # Add User-Agent
    safe_kwargs['headers']['User-Agent'] = 'SafeFlaskApp/1.0'
    
    # Make request
    if method.upper() == 'GET':
        response = requests.get(url, **safe_kwargs)
    elif method.upper() == 'POST':
        safe_kwargs['json'] = kwargs.get('json')
        safe_kwargs['data'] = kwargs.get('data')
        response = requests.post(url, **safe_kwargs)
    elif method.upper() == 'HEAD':
        response = requests.head(url, **safe_kwargs)
    else:
        raise ValueError('HTTP method not allowed')
    
    # Check response size
    content_length = response.headers.get('content-length')
    if content_length and int(content_length) > app.config['MAX_REQUEST_SIZE']:
        raise ValueError('Response too large')
    
    # Read content with size limit
    content = b''
    for chunk in response.iter_content(chunk_size=8192):
        content += chunk
        if len(content) > app.config['MAX_REQUEST_SIZE']:
            raise ValueError('Response too large')
    
    # Replace response content with size-limited content
    response._content = content
    
    return response

# Safe: URL fetching with validation
@app.route('/fetch', methods=['POST'])
def safe_fetch_url():
    try:
        # Validate input
        schema = URLFetchSchema()
        data = schema.load(request.json or {})
        
        # Validate URL for SSRF
        validated_url = validate_url_for_ssrf(data['url'])
        
        # Make safe request
        response = make_safe_request(validated_url)
        
        return jsonify({
            'status_code': response.status_code,
            'content_preview': response.text[:500],  # Limit preview
            'content_length': len(response.content),
            'content_type': response.headers.get('content-type', 'unknown')
        })
    except (ValueError, MarshmallowValidationError, requests.RequestException) as e:
        return jsonify({'error': str(e)}), 400

# Safe: Proxy with validation
@app.route('/proxy')
def safe_proxy_request():
    try:
        target_url = request.args.get('url', '')
        method = request.args.get('method', 'GET')
        
        # Validate method
        if method.upper() not in ['GET', 'HEAD']:
            raise ValueError('HTTP method not allowed')
        
        # Validate URL
        validated_url = validate_url_for_ssrf(target_url)
        
        # Make safe request
        response = make_safe_request(validated_url, method)
        
        return jsonify({
            'status': response.status_code,
            'headers': dict(response.headers),
            'content_length': len(response.content)
        })
    except (ValueError, requests.RequestException) as e:
        return jsonify({'error': str(e)}), 400

# Safe: Webhook with validation
@app.route('/webhook', methods=['POST'])
def safe_send_webhook():
    try:
        # Validate input
        schema = WebhookSchema()
        data = schema.load(request.json or {})
        
        # Get webhook configuration from database
        webhook_config = get_webhook_config(data['webhook_id'])
        
        # Validate webhook URL (should be done when webhook is created)
        validated_url = validate_url_for_ssrf(webhook_config['url'])
        
        # Prepare payload
        payload = data['payload']
        
        # Add signature if secret is provided
        headers = {'Content-Type': 'application/json'}
        if webhook_config.get('secret'):
            signature = generate_webhook_signature(payload, webhook_config['secret'])
            headers['X-Webhook-Signature'] = signature
        
        # Make safe request
        response = make_safe_request(
            validated_url,
            method='POST',
            json=payload,
            headers=headers
        )
        
        return jsonify({
            'webhook_status': response.status_code,
            'delivery_id': generate_delivery_id()
        })
    except (ValueError, MarshmallowValidationError, requests.RequestException) as e:
        return jsonify({'error': str(e)}), 400

def get_webhook_config(webhook_id):
    """Get webhook configuration from database"""
    # Mock function - would query database
    webhooks = {
        'webhook_1': {
            'url': 'https://webhook.example.com/endpoint',
            'secret': 'webhook_secret_123'
        }
    }
    
    if webhook_id not in webhooks:
        raise ValueError('Webhook not found')
    
    return webhooks[webhook_id]

def generate_webhook_signature(payload, secret):
    """Generate webhook signature"""
    import hmac
    import hashlib
    import json
    
    payload_str = json.dumps(payload, sort_keys=True)
    signature = hmac.new(
        secret.encode(),
        payload_str.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return f'sha256={signature}'

def generate_delivery_id():
    """Generate unique delivery ID"""
    import uuid
    return str(uuid.uuid4())

# Safe: API integration with validation
@app.route('/api-call')
def safe_api_call():
    try:
        api_name = request.args.get('api', '')
        endpoint = request.args.get('endpoint', '')
        
        # Validate API name
        api_config = get_api_config(api_name)
        
        # Validate endpoint
        validated_endpoint = validate_api_endpoint(endpoint, api_config)
        
        # Construct URL
        full_url = f"{api_config['base_url']}/{validated_endpoint}"
        
        # Make safe request
        headers = {'Authorization': f"Bearer {api_config['api_key']}"}
        response = make_safe_request(full_url, headers=headers)
        
        return jsonify({
            'api_response': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text[:1000],
            'status_code': response.status_code
        })
    except (ValueError, requests.RequestException) as e:
        return jsonify({'error': str(e)}), 400

def get_api_config(api_name):
    """Get API configuration"""
    apis = {
        'weather': {
            'base_url': 'https://api.example.com',
            'api_key': 'weather_api_key'
        }
    }
    
    if api_name not in apis:
        raise ValueError('API not configured')
    
    return apis[api_name]

def validate_api_endpoint(endpoint, api_config):
    """Validate API endpoint"""
    if not endpoint:
        raise ValueError('Endpoint is required')
    
    # Only allow safe characters
    if not re.match(r'^[a-zA-Z0-9/_.-]+$', endpoint):
        raise ValueError('Endpoint contains invalid characters')
    
    # Prevent traversal
    if '..' in endpoint:
        raise ValueError('Path traversal not allowed')
    
    return endpoint

# Safe: Image processing with validation
@app.route('/process-image', methods=['POST'])
def safe_process_image_from_url():
    try:
        image_url = request.json.get('image_url', '') if request.json else ''
        
        # Validate URL
        validated_url = validate_url_for_ssrf(image_url)
        
        # Make safe request
        response = make_safe_request(validated_url)
        
        # Validate content type
        content_type = response.headers.get('content-type', '')
        if not content_type.startswith('image/'):
            raise ValueError('URL does not point to an image')
        
        # Process image safely
        image_info = process_image_safely(response.content, content_type)
        
        return jsonify(image_info)
    except (ValueError, requests.RequestException) as e:
        return jsonify({'error': str(e)}), 400

def process_image_safely(image_data, content_type):
    """Process image data safely"""
    from PIL import Image
    import io
    
    try:
        # Validate image
        image = Image.open(io.BytesIO(image_data))
        image.verify()  # Verify it's a valid image
        
        return {
            'status': 'Image processed successfully',
            'format': image.format,
            'size': image.size,
            'mode': image.mode,
            'content_type': content_type
        }
    except Exception as e:
        raise ValueError(f'Image processing failed: {str(e)}')

# Safe: URL checker with validation
@app.route('/check-url')
def safe_check_url():
    try:
        url_to_check = request.args.get('url', '')
        
        # Validate URL
        validated_url = validate_url_for_ssrf(url_to_check)
        
        # Make safe HEAD request
        response = make_safe_request(validated_url, method='HEAD')
        
        return jsonify({
            'url': validated_url,
            'status_code': response.status_code,
            'accessible': True,
            'response_time': response.elapsed.total_seconds(),
            'content_type': response.headers.get('content-type')
        })
    except (ValueError, requests.RequestException) as e:
        return jsonify({
            'url': request.args.get('url', ''),
            'accessible': False,
            'error': str(e)
        }), 400

if __name__ == '__main__':
    app.run(debug=False)