Monitoring and Responding to SSRF Incidents

Introduction

Welcome to the third lesson of our Server-Side Request Forgery (SSRF) Prevention in FastAPI course! We've covered what SSRF is and how to prevent it in FastAPI applications. Now, let's focus on an equally important aspect: monitoring and responding to SSRF incidents . Even with robust prevention measures, it's essential to detect and respond to potential attacks quickly. Let's dive in! 🔍

The Importance of Monitoring

Monitoring is a critical component of a comprehensive security strategy. It allows you to: Detect potential SSRF attacks in real-time Collect data for forensic analysis Improve your security measures based on attack patterns Respond quickly to minimize damage Let's explore how to set up effective monitoring for SSRF vulnerabilities in FastAPI applications.

Setting Up Request Logging

The first step in monitoring is to set up comprehensive request logging. This allows you to track and analyze all incoming requests, making it easier to detect suspicious activity: Pythonfrom fastapi import FastAPI, Request import logging import time app = FastAPI() # Configure logging logging.basicConfig( filename='access.log', level=logging.INFO, format='%(asctime)s - %(message)s' ) logger = logging.getLogger(__name__) @app.middleware("http") async def log_requests(request: Request, call_next): start_time = time.time() # Log the request logger.info(f"{request.client.host} - {request.method} {request.url.path}") response = await call_next(request) # Log response details process_time = time.time() - start_time logger.info(f"Status: {response.status_code} - Time: {process_time:.3f}s") return response @app.get("/") async def root(): return {"message": "Hello, world!"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="localhost", port=3000) In this example, we use FastAPI middleware to log all HTTP requests to a file. The logging includes information such as the IP address, request method, URL, status code, and processing time.

Advanced SSRF Detection

To detect potential SSRF attacks, we need to implement more sophisticated monitoring. Let's create middleware that specifically looks for suspicious URL patterns: Pythonfrom fastapi import FastAPI, Request from urllib.parse import urlparse import logging app = FastAPI() logger = logging.getLogger(__name__) @app.middleware("http") async def ssrf_detection_middleware(request: Request, call_next): # Extract URL from request body or query url = None if request.method == "POST": try: body = await request.json() url = body.get('url', '') except: pass else: url = request.query_params.get('url', '') if url and isinstance(url, str): try: parsed_url = urlparse(url) # Check for suspicious patterns suspicious_patterns = [ '127.0.0.1', 'localhost', 'internal', '169.254.169.254', # Cloud metadata service 'file:', 'dict:', 'gopher:', '10.', '172.16.', '192.168.' ] is_suspicious = any( pattern in (parsed_url.hostname or '') or pattern in url for pattern in suspicious_patterns ) if is_suspicious: # Log the suspicious request logger.warning(f"POTENTIAL SSRF ATTACK: {request.client.host} tried to access {url}") # Note: In production, you could integrate with an alerting service # For blocking operations, use a message queue or background task system except: # URL parsing failed, but we'll continue processing the request pass response = await call_next(request) return response @app.post("/fetch-url") async def fetch_url(request: Request): # Your secure URL fetching logic here return {"success": True} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="localhost", port=3000) This middleware checks for suspicious URL patterns that might indicate an SSRF attack attempt. When detected, it logs the incident for further analysis. For production environments, you would typically integrate this with a message queue or background task system to trigger alerts without blocking the request processing.

Join the 1M+ learners on CodeSignal

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

Setting Up Alerts

Alerting is crucial for timely response to potential security incidents. Let's implement a system to alert administrators when suspicious activity is detected:

Python

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
import smtplib
from email.message import EmailMessage
import os
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

def send_email_alert(subject: str, message: str):
    """Send email alert to administrator (blocking function)"""
    email_user = os.getenv('EMAIL_USER', 'your-email@gmail.com')
    email_password = os.getenv('EMAIL_PASSWORD', 'your-email-password')
    admin_email = os.getenv('ADMIN_EMAIL', 'admin@example.com')
    
    try:
        msg = EmailMessage()
        msg.set_content(message)
        msg['Subject'] = subject
        msg['From'] = email_user
        msg['To'] = admin_email
        
        # For Gmail, use SMTP_SSL on port 465
        with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
            smtp.login(email_user, email_password)
            smtp.send_message(msg)
        
        logger.info(f"Email sent: {subject}")
    except Exception as e:
        logger.error(f"Error sending email: {e}")

@app.post("/report-incident")
async def report_incident(request: Request, background_tasks: BackgroundTasks):
    """Endpoint to report security incidents"""
    try:
        body = await request.json()
        incident = body.get('incident')
        
        if not incident:
            raise HTTPException(status_code=400, detail="Incident details required")
        
        # Log the incident
        logger.warning(f"Security incident reported: {incident}")
        
        # Alert administrators asynchronously using background tasks
        background_tasks.add_task(
            send_email_alert, 
            'SSRF Incident Reported', 
            f'Details: {incident}'
        )
        
        return {"success": True}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=3000)

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
import smtplib
from email.message import EmailMessage
import os
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

def send_email_alert(subject: str, message: str):
    """Send email alert to administrator (blocking function)"""
    email_user = os.getenv('EMAIL_USER', 'your-email@gmail.com')
    email_password = os.getenv('EMAIL_PASSWORD', 'your-email-password')
    admin_email = os.getenv('ADMIN_EMAIL', 'admin@example.com')
    
    try:
        msg = EmailMessage()
        msg.set_content(message)
        msg['Subject'] = subject
        msg['From'] = email_user
        msg['To'] = admin_email
        
        # For Gmail, use SMTP_SSL on port 465
        with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
            smtp.login(email_user, email_password)
            smtp.send_message(msg)
        
        logger.info(f"Email sent: {subject}")
    except Exception as e:
        logger.error(f"Error sending email: {e}")

@app.post("/report-incident")
async def report_incident(request: Request, background_tasks: BackgroundTasks):
    """Endpoint to report security incidents"""
    try:
        body = await request.json()
        incident = body.get('incident')
        
        if not incident:
            raise HTTPException(status_code=400, detail="Incident details required")
        
        # Log the incident
        logger.warning(f"Security incident reported: {incident}")
        
        # Alert administrators asynchronously using background tasks
        background_tasks.add_task(
            send_email_alert, 
            'SSRF Incident Reported', 
            f'Details: {incident}'
        )
        
        return {"success": True}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=3000)

This code sets up an email alert system using Python's built-in email libraries. When a security incident is reported, an email is sent to the administrator with details of the incident.

Incident Response Plan

Having a solid incident response plan is essential for handling security breaches effectively. Let's create a simple incident response handler for SSRF attacks: