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

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

Python

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)

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)

Python

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)

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)

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)

Python

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)

Introduction

The Importance of Monitoring

Setting Up Request Logging

Advanced SSRF Detection

Join the 1M+ learners on CodeSignal

Setting Up Alerts

Incident Response Plan

Conclusion