Utilities¶

The utils module provides various helper functions for security operations.

Logging Functions¶

setup_custom_logging¶

async def setup_custom_logging(
    log_file: str
) -> logging.Logger:
    """
    Setup custom logging for the application.
    """

log_activity¶

async def log_activity(
    request: Request,
    logger: logging.Logger,
    log_type: str = "request",
    reason: str = "",
    passive_mode: bool = False,
    trigger_info: str = "",
    level: Literal["INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"] | None = "WARNING"
):
    """
    Universal logging function for all types of requests and activities.
    """

Parameters:

request: The FastAPI request object
logger: The logger instance
log_type: Type of log entry (default: "request", can also be "suspicious")
reason: Reason for flagging an activity
passive_mode: Whether to enable passive mode logging format
trigger_info: Details about what triggered detection
level: The logging level to use. If None, logging is disabled. Defaults to "WARNING".

This is a unified logging function that handles regular requests, suspicious activities, and passive mode logging.

Security Check Functions¶

is_user_agent_allowed¶

async def is_user_agent_allowed(
    user_agent: str,
    config: SecurityConfig
) -> bool:
    """
    Check if user agent is allowed.
    """

check_ip_country¶

async def check_ip_country(
    request: str | Request,
    config: SecurityConfig,
    ipinfo_db: IPInfoManager
) -> bool:
    """
    Check if IP is from a blocked country.
    """

is_ip_allowed¶

async def is_ip_allowed(
    ip: str,
    config: SecurityConfig,
    ipinfo_db: IPInfoManager | None = None
) -> bool:
    """
    Check if IP address is allowed.
    """

The ipinfo_db parameter is now properly optional - it's only needed when country filtering is configured. If it's not provided when country filtering is configured, the function will work correctly but won't apply country filtering rules rules.

This function intelligently handles:

Whitelist/blacklist checking
Country filtering (only when IPInfoManager is provided)
Cloud provider detection (only when cloud blocking is configured)

This selective processing aligns with FastAPI Guard's smart resource loading to optimize performance.

detect_penetration_attempt¶

async def detect_penetration_attempt(
    request: Request
) -> tuple[bool, str]

Detect potential penetration attempts in the request.

This function checks various parts of the request (query params, body, path, headers) against a list of suspicious patterns to identify potential security threats.

Returns a tuple where:

First element is a boolean: True if a potential attack is detected, False otherwise
Second element is a string with details about what triggered the detection, or empty string if no attack detected

Example usage:

from fastapi import Request
from guard.utils import detect_penetration_attempt

@app.post("/api/submit")
async def submit_data(request: Request):
    is_suspicious, trigger_info = await detect_penetration_attempt(request)
    if is_suspicious:
        # Log the detection with details
        logger.warning(f"Attack detected: {trigger_info}")
        return {"error": "Suspicious activity detected"}
    return {"success": True}

extract_client_ip¶

def extract_client_ip(request: Request, config: SecurityConfig) -> str:
    """
    Securely extract the client IP address from the request, considering trusted proxies.

    This function implements a secure approach to IP extraction that protects against
    X-Forwarded-For header injection attacks.
    """

This function provides a secure way to extract client IPs by:

Only trusting X-Forwarded-For headers from configured trusted proxies
Using the connecting IP when not from a trusted proxy
Properly handling proxy chains based on configured depth

Usage Examples¶

from guard.utils import (
    setup_custom_logging,
    log_activity,
    detect_penetration_attempt
)

# Setup logging
logger = await setup_custom_logging("security.log")

# Log regular request
await log_activity(request, logger)

# Log suspicious activity
await log_activity(
    request,
    logger,
    log_type="suspicious",
    reason="Suspicious pattern detected"
)

# Check for penetration attempts
is_suspicious, trigger_info = await detect_penetration_attempt(request)