Security Logging and Monitoring in FastAPI Applications

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: Detecting Security Incidents: Logs can reveal suspicious activities that may indicate ongoing attacks. Investigating Breaches: After a security incident, logs provide valuable data for forensic analysis. Compliance Requirements: Many regulatory frameworks require specific logging practices. 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: Pythonfrom 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)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.

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

Python

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)

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)

Python

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)

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)

Python

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)

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)

Python

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)

Python

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)

Python

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)

Introduction

The Role of Security Logging

Join the 1M+ learners on CodeSignal

Structured Logging for Security Events

Real-time Log Monitoring

Setting Up Alerts Based on Log Patterns

Conclusion