Preventing SSRF in FastAPI

Introduction

Welcome back! In the previous lesson, we explored the concept of Server-Side Request Forgery (SSRF) vulnerabilities and how to detect them. Now, we'll focus on preventing SSRF in FastAPI applications. By the end of this lesson, you'll understand how to secure your FastAPI applications against SSRF attacks, ensuring a safer web environment. Let's dive in! 🌟

Understanding SSRF in FastAPI

FastAPI is a modern, fast web framework for building APIs with Python, known for its simplicity and automatic API documentation. However, this flexibility can sometimes lead to vulnerabilities if not handled properly. SSRF vulnerabilities in FastAPI often arise when user input is not validated, allowing attackers to manipulate server-side requests. When a FastAPI application receives a request, it processes the input and may make further requests to external resources. If this input is not properly validated, an attacker can craft a request that tricks the server into making unintended requests, potentially accessing sensitive data or services.

The Vulnerable Code

Let's examine a piece of code that demonstrates how SSRF vulnerabilities can occur in a FastAPI application: Pythonfrom fastapi import FastAPI from pydantic import BaseModel import httpx app = FastAPI() class URLRequest(BaseModel): url: str @app.post("/fetch-url") async def fetch_url(request: URLRequest): try: # Proceed with the request to the validated URL with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(request.url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except Exception as e: return {"error": "Failed to fetch URL"}from fastapi import FastAPI from pydantic import BaseModel import httpx app = FastAPI() class URLRequest(BaseModel): url: str @app.post("/fetch-url") async def fetch_url(request: URLRequest): try: # Proceed with the request to the validated URL with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(request.url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except Exception as e: return {"error": "Failed to fetch URL"} In this code, the application accepts a URL from the user and fetches data from it using httpx. However, there's no validation to ensure the URL is safe or trusted. This lack of validation can lead to SSRF vulnerabilities, as attackers can provide malicious URLs to exploit the server.

Exploiting the Vulnerability

An attacker can exploit this vulnerability by providing a malicious URL to the /fetch-url endpoint. Here's an example of how an attack might be performed: Shell # Attacker crafts a request to the vulnerable endpoint curl -X POST http://localhost:3000/fetch-url -H "Content-Type: application/json" -d '{"url": "http://internal-service.local"}' # Attacker crafts a request to the vulnerable endpoint curl -X POST http://localhost:3000/fetch-url -H "Content-Type: application/json" -d '{"url": "http://internal-service.local"}' In this example, the attacker sends a POST request to the /fetch-url endpoint with a URL pointing to an internal service. If the application does not validate the URL, it may inadvertently access internal resources, leading to potential data breaches or unauthorized actions. This is especially dangerous in environments like cloud platforms (AWS, Azure, GCP), where internal endpoints such as metadata services are exposed only to internal IPs. By exploiting SSRF in such cases, an attacker can retrieve sensitive data like instance credentials or configuration tokens, which may then be used to escalate privileges or access cloud APIs directly.

Input Validation

To prevent SSRF attacks, it's crucial to implement robust input validation. Let's start by validating the user input to ensure it meets our security requirements: Python @app.post("/fetch-url") async def fetch_url(request: URLRequest): url = request.url # Validate URL if not url or not isinstance(url, str): raise HTTPException(status_code=400, detail="Invalid URL") # ... rest of the code @app.post("/fetch-url") async def fetch_url(request: URLRequest): url = request.url # Validate URL if not url or not isinstance(url, str): raise HTTPException(status_code=400, detail="Invalid URL") # ... rest of the code In this step, we check if the URL is present and is a string. This basic validation helps ensure that the input is in the expected format before proceeding.

URL Whitelisting

Next, we'll implement URL whitelisting to restrict requests to trusted domains only: Pythonfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel import httpx from urllib.parse import urlparse app = FastAPI() # List of allowed domains ALLOWED_DOMAINS = ['trusted-domain.com', 'api.trusted-domain.com'] class URLRequest(BaseModel): url: str # Example of SSRF prevention @app.post("/fetch-url") async def fetch_url(request: URLRequest): url = request.url # Basic validation if not url or not isinstance(url, str): raise HTTPException(status_code=400, detail="Invalid URL format") try: # Parse the URL to validate its components parsed_url = urlparse(url) # Ensure only HTTP/HTTPS schemes are used if parsed_url.scheme not in ['http', 'https']: raise HTTPException(status_code=400, detail="Only HTTP/HTTPS protocols are allowed") # Check if domain is in our whitelist is_domain_allowed = any( parsed_url.hostname == domain or parsed_url.hostname.endswith(f'.{domain}') for domain in ALLOWED_DOMAINS ) if not is_domain_allowed: raise HTTPException(status_code=400, detail="Domain not allowed") # DNS resolution check to prevent DNS rebinding attacks import socket try: ip_addresses = socket.getaddrinfo(parsed_url.hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) for addr_info in ip_addresses: ip_str = addr_info[4][0] # Get the IP address if is_private_ip(ip_str): raise HTTPException(status_code=400, detail=f"Hostname resolves to private IP: {ip_str}") except socket.gaierror: raise HTTPException(status_code=400, detail="Failed to resolve hostname") # Proceed with the request to the validated URL with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except ValueError: raise HTTPException(status_code=400, detail="Invalid URL") except httpx.TimeoutException: raise HTTPException(status_code=408, detail="Request timeout") except httpx.RequestError as e: raise HTTPException(status_code=502, detail=f"Request failed: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail="Failed to fetch URL") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="localhost", port=3000)from fastapi import FastAPI, HTTPException from pydantic import BaseModel import httpx from urllib.parse import urlparse app = FastAPI() # List of allowed domains ALLOWED_DOMAINS = ['trusted-domain.com', 'api.trusted-domain.com'] class URLRequest(BaseModel): url: str # Example of SSRF prevention @app.post("/fetch-url") async def fetch_url(request: URLRequest): url = request.url # Basic validation if not url or not isinstance(url, str): raise HTTPException(status_code=400, detail="Invalid URL format") try: # Parse the URL to validate its components parsed_url = urlparse(url) # Ensure only HTTP/HTTPS schemes are used if parsed_url.scheme not in ['http', 'https']: raise HTTPException(status_code=400, detail="Only HTTP/HTTPS protocols are allowed") # Check if domain is in our whitelist is_domain_allowed = any( parsed_url.hostname == domain or parsed_url.hostname.endswith(f'.{domain}') for domain in ALLOWED_DOMAINS ) if not is_domain_allowed: raise HTTPException(status_code=400, detail="Domain not allowed") # DNS resolution check to prevent DNS rebinding attacks import socket try: ip_addresses = socket.getaddrinfo(parsed_url.hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) for addr_info in ip_addresses: ip_str = addr_info[4][0] # Get the IP address if is_private_ip(ip_str): raise HTTPException(status_code=400, detail=f"Hostname resolves to private IP: {ip_str}") except socket.gaierror: raise HTTPException(status_code=400, detail="Failed to resolve hostname") # Proceed with the request to the validated URL with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except ValueError: raise HTTPException(status_code=400, detail="Invalid URL") except httpx.TimeoutException: raise HTTPException(status_code=408, detail="Request timeout") except httpx.RequestError as e: raise HTTPException(status_code=502, detail=f"Request failed: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail="Failed to fetch URL") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="localhost", port=3000) By checking if the domain is in our whitelist, we limit the scope of where requests can be made. This prevents attackers from directing the server to malicious or internal URLs.

Using urlparse for Validation

For more robust URL validation, we can use Python's urlparse module: Pythonfrom urllib.parse import urlparse # Validate URL using urlparse try: parsed_url = urlparse(url) # Ensure only HTTP/HTTPS schemes are used if parsed_url.scheme not in ['http', 'https']: raise HTTPException(status_code=400, detail="Only HTTP/HTTPS protocols are allowed") # Check hostname against whitelist allowed_domains = ['trusted-domain.com', 'api.trusted-domain.com'] if not any(parsed_url.hostname == domain or parsed_url.hostname.endswith(f'.{domain}') for domain in allowed_domains): raise HTTPException(status_code=400, detail="Domain not allowed") # Now we can safely make the request with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except ValueError: raise HTTPException(status_code=400, detail="Invalid URL") except Exception as e: raise HTTPException(status_code=500, detail="Failed to fetch URL")from urllib.parse import urlparse # Validate URL using urlparse try: parsed_url = urlparse(url) # Ensure only HTTP/HTTPS schemes are used if parsed_url.scheme not in ['http', 'https']: raise HTTPException(status_code=400, detail="Only HTTP/HTTPS protocols are allowed") # Check hostname against whitelist allowed_domains = ['trusted-domain.com', 'api.trusted-domain.com'] if not any(parsed_url.hostname == domain or parsed_url.hostname.endswith(f'.{domain}') for domain in allowed_domains): raise HTTPException(status_code=400, detail="Domain not allowed") # Now we can safely make the request with safety controls timeout = httpx.Timeout(10.0, connect=5.0) # 10s total, 5s connect timeout limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) async with httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False # Prevent redirect-based allowlist bypass ) as client: response = await client.get(url) # Check response size to prevent memory exhaustion if len(response.content) > 1024 * 1024: # 1MB limit raise HTTPException(status_code=413, detail="Response too large") return {"data": response.text} except ValueError: raise HTTPException(status_code=400, detail="Invalid URL") except Exception as e: raise HTTPException(status_code=500, detail="Failed to fetch URL") Using urlparse allows us to easily validate different parts of the URL, such as the scheme and hostname.

Implementing IP Address Restrictions

To further enhance security, we can block requests to internal IP addresses: Python import ipaddress # Function to check if an IP address is private def is_private_ip(hostname: str) -> bool: try: # Try to parse the hostname as an IP address ip = ipaddress.ip_address(hostname) return (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified) except ValueError: # If parsing fails, it's not a valid IP address return False # In your route handler try: parsed_url = urlparse(url) # Check if hostname is an IP address and if it's private if is_private_ip(parsed_url.hostname): raise HTTPException(status_code=400, detail="Private IP addresses not allowed") # Rest of your validation logic... except Exception as e: # Handle errors... import ipaddress # Function to check if an IP address is private def is_private_ip(hostname: str) -> bool: try: # Try to parse the hostname as an IP address ip = ipaddress.ip_address(hostname) return (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified) except ValueError: # If parsing fails, it's not a valid IP address return False # In your route handler try: parsed_url = urlparse(url) # Check if hostname is an IP address and if it's private if is_private_ip(parsed_url.hostname): raise HTTPException(status_code=400, detail="Private IP addresses not allowed") # Rest of your validation logic... except Exception as e: # Handle errors... This function uses Python's ipaddress library to check if the hostname is a private IP address, preventing requests to internal resources.

Conclusion

In this lesson, we explored how to prevent SSRF vulnerabilities in FastAPI applications. We learned about the importance of input validation, URL whitelisting, and secure HTTP requests. By implementing these security measures, you can significantly reduce the risk of SSRF attacks in your applications. In the next lesson, we'll dive deeper into monitoring and responding to SSRF incidents, helping you build a comprehensive security strategy for your FastAPI applications.