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.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

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"}

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"}'

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

Python

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)

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)

Python

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")

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")

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...

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"}'

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

Python

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)

Python

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")

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...

Introduction

Understanding SSRF in FastAPI

The Vulnerable Code

Join the 1M+ learners on CodeSignal

Exploiting the Vulnerability

Input Validation

URL Whitelisting

Using urlparse for Validation

Implementing IP Address Restrictions

Conclusion