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:

  1. Detect potential SSRF attacks in real-time
  2. Collect data for forensic analysis
  3. Improve your security measures based on attack patterns
  4. 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:

from 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:

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

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:

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:

from datetime import datetime
from typing import List, Optional

class SSRFIncidentHandler:
    """Handler for SSRF security incidents"""
    
    def __init__(self):
        # Record of incidents
        self.incidents: List[dict] = []
    
    def log_incident(self, ip: str, url: str):
        """Log a new incident"""
        incident = {
            'timestamp': datetime.now(),
            'ip': ip,
            'url': url,
            'resolved': False
        }
        
        self.incidents.append(incident)
        print(f"[{incident['timestamp'].isoformat()}] SSRF Attempt: {ip} -> {url}")
        
        # Alert appropriate personnel
        self.alert_team(incident)
        
        # Implement mitigation measures
        self.mitigate(ip)
        
        return incident
    
    def alert_team(self, incident: dict):
        """Alert security team"""
        # In a real app, this would send emails, SMS, or integrate with a security platform
        print(f"SECURITY ALERT: SSRF attempt from {incident['ip']} targeting {incident['url']}")
    
    def mitigate(self, ip: str):
        """Implement mitigation measures"""
        # In a real app, this might temporarily block the IP, adjust WAF rules, etc.
        print(f"Mitigation: Temporarily blocking requests from {ip}")
    
    def get_incidents(self, resolved: Optional[bool] = None) -> List[dict]:
        """Get all recorded incidents"""
        if resolved is None:
            return self.incidents
        return [inc for inc in self.incidents if inc['resolved'] == resolved]
    
    def resolve_incident(self, index: int) -> bool:
        """Mark an incident as resolved"""
        if 0 <= index < len(self.incidents):
            self.incidents[index]['resolved'] = True
            return True
        return False

# Example usage in a FastAPI app
from fastapi import FastAPI, Request

app = FastAPI()
incident_handler = SSRFIncidentHandler()

def is_suspicious_url(url: str) -> bool:
    """Implement your SSRF detection logic here"""
    return 'internal' in url or 'localhost' in url

@app.middleware("http")
async def check_ssrf(request: Request, call_next):
    url = request.query_params.get('url', '')
    
    if url and is_suspicious_url(url):
        incident_handler.log_incident(request.client.host, url)
        return JSONResponse(
            status_code=403,
            content={"error": "Access denied"}
        )
    
    response = await call_next(request)
    return response

@app.get("/admin/incidents")
async def get_incidents(resolved: Optional[bool] = None):
    """Admin endpoint to view incidents"""
    # In a real app, this would require authentication
    return incident_handler.get_incidents(resolved)

This incident handler logs SSRF attempts, alerts the security team, implements mitigation measures, and provides an interface for managing incidents.

Conclusion

In this lesson, we explored the importance of monitoring and responding to SSRF incidents. We learned how to set up request logging, implement advanced SSRF detection, create an alerting system, and develop an incident response plan. By combining these techniques with the prevention measures from the previous lesson, you can create a robust defense against SSRF vulnerabilities in your FastAPI applications.

In the next lesson, we'll dive deeper into security logging and monitoring, exploring more advanced techniques to enhance your application's security posture. Stay tuned! 🚀

Previous
Next
Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal