Introduction

Welcome to the final lesson in our Server-Side Request Forgery (SSRF) Prevention in FastAPI course! In this lesson, we'll explore security logging and monitoring in depth. Effective logging and monitoring are crucial components of a comprehensive security strategy, as they help you detect, investigate, and respond to security incidents promptly. Let's dive in and discover how to implement these practices in your FastAPI applications! 📊

The Role of Security Logging

Security logging is the practice of recording events related to security concerns within your application. Properly implemented logs serve multiple purposes:

  1. Detecting Security Incidents: Logs can reveal suspicious activities that may indicate ongoing attacks.
  2. Investigating Breaches: After a security incident, logs provide valuable data for forensic analysis.
  3. Compliance Requirements: Many regulatory frameworks require specific logging practices.
  4. System Auditing: Logs help track user activities and system changes over time.

Let's implement a comprehensive logging system using Python's built-in logging module:

from fastapi import FastAPI, Request
import logging
import time

app = FastAPI()

# Configure logging format and handlers
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('logs/error.log', mode='a'),
        logging.FileHandler('logs/combined.log', mode='a')
    ]
)

logger = logging.getLogger('web-service')

# Set error log to only log errors
error_handler = logging.FileHandler('logs/error.log')
error_handler.setLevel(logging.ERROR)
logger.addHandler(error_handler)

# Logging middleware
@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

# Add custom security logging middleware
@app.middleware("http")
async def security_logging_middleware(request: Request, call_next):
    # Log potentially suspicious activities
    if request.method == "POST":
        try:
            body = await request.json()
            url = body.get('url')
            if url:
                logger.warning('URL parameter detected', extra={
                    'ip': request.client.host,
                    'url': url,
                    'method': request.method,
                    'path': request.url.path,
                    'user_agent': request.headers.get('user-agent', '')
                })
        except:
            pass
    
    response = await call_next(request)
    return response

# Example route
@app.get("/")
async def root():
    return {"message": "Hello, world!"}

if __name__ == "__main__":
    import uvicorn
    logger.info('Server running on port 3000')
    uvicorn.run(app, host="localhost", port=3000)

This code sets up a robust logging system using Python's logging module. It logs all HTTP requests and errors to both the console and files, making it easier to monitor and analyze application activities.

Structured Logging for Security Events

To make security logs more useful, it's important to use structured logging with consistent fields:

from enum import Enum
import logging
import time

# Define security event types for consistency
class SecurityEventType(str, Enum):
    AUTHENTICATION_FAILURE = 'authentication_failure'
    AUTHORIZATION_FAILURE = 'authorization_failure'
    INPUT_VALIDATION_FAILURE = 'input_validation_failure'
    POTENTIAL_SSRF_ATTEMPT = 'potential_ssrf_attempt'
    SENSITIVE_DATA_ACCESS = 'sensitive_data_access'

logger = logging.getLogger('web-service')

# Function to log security events with consistent structure
def log_security_event(
    event_type: SecurityEventType,
    message: str,
    metadata: dict
):
    logger.warning(
        message,
        extra={
            'security_event_type': event_type.value,
            'timestamp': time.time(),
            **metadata
        }
    )

# Example usage in a route
@app.post('/fetch-url')
async def fetch_url(request: Request):
    try:
        body = await request.json()
        url = body.get('url')
    except:
        url = None
    
    if not url or not isinstance(url, str):
        log_security_event(
            SecurityEventType.INPUT_VALIDATION_FAILURE,
            'Invalid URL input',
            {
                'ip': request.client.host,
                'input': url,
                'user_id': getattr(request.state, 'user_id', None),
                'path': request.url.path
            }
        )
        
        return JSONResponse(
            status_code=400,
            content={'error': 'Invalid URL'}
        )
    
    # Check for potential SSRF attempt
    if is_suspicious_url(url):
        log_security_event(
            SecurityEventType.POTENTIAL_SSRF_ATTEMPT,
            'Potential SSRF attempt detected',
            {
                'ip': request.client.host,
                'url': url,
                'user_id': getattr(request.state, 'user_id', None),
                'user_agent': request.headers.get('user-agent', '')
            }
        )
        
        return JSONResponse(
            status_code=403,
            content={'error': 'Access denied'}
        )
    
    # Process the URL (securely)
    return {'success': True}

def is_suspicious_url(url: str) -> bool:
    suspicious_patterns = ['localhost', '127.0.0.1', 'internal', '192.168.']
    return any(pattern in url for pattern in suspicious_patterns)

This approach ensures that security events are logged with consistent fields, making it easier to analyze and correlate events across your application.

Real-time Log Monitoring

Logging is only effective if someone is monitoring the logs. Let's implement a simple real-time monitoring system using WebSockets:

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import logging
import asyncio
from typing import List

app = FastAPI()

# Store active WebSocket connections
active_connections: List[WebSocket] = []

# Custom logging handler for WebSocket broadcasting
class WebSocketHandler(logging.Handler):
    def emit(self, record):
        log_entry = self.format(record)
        # Broadcast to all connected clients
        asyncio.create_task(broadcast_log(log_entry))

async def broadcast_log(message: str):
    for connection in active_connections:
        try:
            await connection.send_text(message)
        except:
            pass

# Configure logger with WebSocket handler
logger = logging.getLogger('web-service')
ws_handler = WebSocketHandler()
ws_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
logger.addHandler(ws_handler)

@app.websocket("/ws/logs")
async def websocket_logs(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except:
        active_connections.remove(websocket)

# Example route with security logging
@app.get('/api/data')
async def get_data(request: Request):
    logger.info(f'Data accessed from {request.client.host}')
    return {'data': 'sensitive information'}

# Example route that triggers a security warning
@app.post('/api/fetch')
async def fetch_api(request: Request):
    try:
        body = await request.json()
        url = body.get('url')
    except:
        url = None
    
    if url and is_suspicious_url(url):
        logger.warning(f'Potential SSRF attempt from {request.client.host}: {url}')
        return JSONResponse(status_code=403, content={'error': 'Access denied'})
    
    return {'success': True}

def is_suspicious_url(url: str) -> bool:
    return 'internal' in url or 'localhost' in url

# Simple admin dashboard to view logs in real-time
@app.get('/admin/logs', response_class=HTMLResponse)
async def admin_logs():
    return """
    <html>
      <head>
        <title>Security Log Monitor</title>
        <script>
          const ws = new WebSocket('ws://localhost:3000/ws/logs');
          
          ws.onmessage = (event) => {
            const logElement = document.createElement('div');
            logElement.textContent = event.data;
            if (event.data.includes('WARNING') || event.data.includes('ERROR')) {
              logElement.style.color = 'red';
              logElement.style.fontWeight = 'bold';
            } else {
              logElement.style.color = 'blue';
            }
            document.getElementById('logs').prepend(logElement);
          };
        </script>
        <style>
          #logs { max-height: 500px; overflow-y: auto; }
        </style>
      </head>
      <body>
        <h1>Security Log Monitor</h1>
        <div id="logs"></div>
      </body>
    </html>
    """

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

This implementation uses WebSockets to broadcast log events to a simple admin dashboard, allowing for real-time monitoring of security events.

Setting Up Alerts Based on Log Patterns

To proactively respond to security events, we can set up alerts based on specific log patterns:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import logging
import smtplib
from email.message import EmailMessage
from collections import defaultdict
import time
from typing import Dict, List

app = FastAPI()
logger = logging.getLogger('web-service')

# Class to handle alerts based on log patterns
class LogAlerter:
    def __init__(self):
        self.alert_thresholds = {
            'potential_ssrf_attempt': {
                'count': 3,
                'time_window': 10 * 60,  # 10 minutes in seconds
                'last_alert': 0
            },
            'authentication_failure': {
                'count': 5,
                'time_window': 15 * 60,  # 15 minutes
                'last_alert': 0
            }
        }
        self.event_counts: Dict[str, List[Dict]] = defaultdict(list)
    
    def log_event(self, event_type: str, ip: str):
        now = time.time()
        
        # Add event to the count
        self.event_counts[event_type].append({
            'timestamp': now,
            'ip': ip
        })
        
        # Check if we need to send an alert
        self.check_alert_threshold(event_type)
    
    def check_alert_threshold(self, event_type: str):
        if event_type not in self.alert_thresholds:
            return
        
        threshold = self.alert_thresholds[event_type]
        now = time.time()
        
        # Filter events within the time window
        events = self.event_counts[event_type]
        recent_events = [e for e in events if e['timestamp'] > now - threshold['time_window']]
        
        # Update event list to only include recent events
        self.event_counts[event_type] = recent_events
        
        # Check if we've hit the threshold
        if (len(recent_events) >= threshold['count'] and 
            now - threshold['last_alert'] > threshold['time_window']):
            
            # Group by IP address
            ip_counts = defaultdict(int)
            for event in recent_events:
                ip_counts[event['ip']] += 1
            
            # Find the IP with the most events
            max_ip = max(ip_counts.items(), key=lambda x: x[1])
            
            # Send the alert
            self.send_alert(event_type, len(recent_events), max_ip[0], max_ip[1])
            
            # Update last alert timestamp
            threshold['last_alert'] = now
    
    def send_alert(self, event_type: str, total_count: int, main_ip: str, ip_count: int):
        subject = f"SECURITY ALERT: {self.format_event_type(event_type)}"
        message = f"""
        We detected {total_count} instances of {self.format_event_type(event_type)} in the last few minutes.
        
        The primary source appears to be IP: {main_ip} ({ip_count} events).
        
        Please investigate immediately.
        """
        
        # Log the alert
        logger.error(f'{subject} - Total: {total_count}, Main IP: {main_ip} ({ip_count})')
        
        # Send email alert
        try:
            msg = EmailMessage()
            msg.set_content(message)
            msg['Subject'] = subject
            msg['From'] = 'security@example.com'
            msg['To'] = 'admin@example.com'
            
            with smtplib.SMTP('localhost') as smtp:
                smtp.send_message(msg)
            
            logger.info('Alert email sent successfully')
        except Exception as e:
            logger.error(f'Failed to send alert email: {str(e)}')
    
    def format_event_type(self, event_type: str) -> str:
        return event_type.replace('_', ' ').title()

# Initialize the log alerter
alerter = LogAlerter()

# Example route with security logging
@app.post('/api/fetch')
async def fetch_api(request: Request):
    try:
        body = await request.json()
        url = body.get('url')
    except:
        url = None
    
    if url and is_suspicious_url(url):
        logger.warning(f'Potential SSRF attempt: {request.client.host} -> {url}')
        alerter.log_event('potential_ssrf_attempt', request.client.host)
        return JSONResponse(status_code=403, content={'error': 'Access denied'})
    
    return {'success': True}

def is_suspicious_url(url: str) -> bool:
    return 'internal' in url or 'localhost' in url

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

This implementation sets up a sophisticated alerting system that monitors log patterns and sends alerts when certain thresholds are exceeded, helping you respond quickly to potential security incidents.

Conclusion

In this lesson, we explored the importance of security logging and monitoring in protecting FastAPI applications from SSRF and other attacks. We learned how to implement structured logging, real-time monitoring, and automated alerting based on log patterns. By integrating these practices into your security strategy, you can significantly enhance your ability to detect and respond to security incidents.

Throughout this course, we've covered the fundamentals of SSRF, prevention techniques in FastAPI, incident response, and comprehensive monitoring. These skills will help you build more secure applications and protect your users' data from potential threats.

Remember that security is an ongoing process, not a one-time implementation. Continue to stay informed about emerging threats and best practices to ensure your applications remain secure in an ever-evolving landscape. Thank you for joining us on this journey to better security! 🚀

Previous
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