SecurityMiddleware¶

The SecurityMiddleware class is the core component of FastAPI Guard that handles all security features.

Architecture (v4.2.0+)

Since v4.2.0, SecurityMiddleware uses a modular architecture with specialized core modules. The middleware acts as an orchestration layer, delegating to specialized handlers. See Core Architecture for internal details.

Class Definition¶

class SecurityMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: Callable[[Request], Awaitable[Response]],
        config: SecurityConfig
    ):
        """
        Initialize the SecurityMiddleware.

        Args:
            app: The FastAPI/Starlette application
            config: Security configuration object

        Note:
            The middleware initializes all core components using
            dependency injection for clean separation of concerns.
        """

Architecture Overview¶

Request Processing Flow¶

The middleware processes requests through a modular pipeline:

Request → Middleware.dispatch()
    ↓
1. BypassHandler.handle_passthrough()
    ├─ No client IP? → Pass through
    └─ Excluded path? → Pass through
    ↓
2. Extract client IP and route config
    ↓
3. BypassHandler.handle_security_bypass()
    └─ Decorator bypass? → Pass through
    ↓
4. SecurityCheckPipeline.execute()
    ├─ RouteConfigCheck
    ├─ EmergencyModeCheck
    ├─ HttpsEnforcementCheck
    ├─ RequestLoggingCheck
    ├─ RequestSizeContentCheck
    ├─ RequiredHeadersCheck
    ├─ AuthenticationCheck
    ├─ ReferrerCheck
    ├─ CustomValidatorsCheck
    ├─ TimeWindowCheck
    ├─ CloudIpRefreshCheck
    ├─ IpSecurityCheck
    ├─ CloudProviderCheck
    ├─ UserAgentCheck
    ├─ RateLimitCheck
    ├─ SuspiciousActivityCheck
    └─ CustomRequestCheck
    ↓
5. BehavioralProcessor.process_usage_rules()
    ↓
6. call_next(request)  # Forward to app
    ↓
7. ErrorResponseFactory.process_response()
    ├─ Apply security headers
    ├─ Apply CORS headers
    ├─ Collect metrics
    └─ Process behavioral return rules
    ↓
Response

Core Components¶

The middleware delegates to these specialized modules:

SecurityCheckPipeline: Executes security checks in sequence
SecurityEventBus: Sends security events to monitoring agent
MetricsCollector: Collects request/response metrics
HandlerInitializer: Initializes Redis and Agent handlers
ErrorResponseFactory: Creates and processes responses
RouteConfigResolver: Resolves decorator configurations
RequestValidator: Validates request properties
BypassHandler: Handles security bypasses
BehavioralProcessor: Processes behavioral rules

See Core Architecture for detailed documentation of each module.

Public Methods¶

dispatch¶

async def dispatch(
    self,
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    """
    Main request handler that orchestrates security checks.

    This method is pure orchestration - it delegates all logic
    to specialized handlers and maintains no business logic itself.

    Args:
        request: The incoming request
        call_next: The next middleware/handler in the chain

    Returns:
        Response: Either from security checks (blocking) or from the app

    Flow:
        1. Handle passthrough cases (no client, excluded paths)
        2. Get route config and client IP
        3. Handle security bypasses
        4. Execute security pipeline
        5. Process behavioral usage rules
        6. Call next handler
        7. Process response (headers, metrics, behavioral return rules)
    """

create_error_response¶

async def create_error_response(
    self,
    status_code: int,
    default_message: str
) -> Response:
    """
    Create standardized error responses.

    Delegates to ErrorResponseFactory for response creation.

    Args:
        status_code: HTTP status code
        default_message: Default error message

    Returns:
        Response: Error response with optional custom message

    Note:
        Custom error messages can be configured in SecurityConfig
        via the custom_error_responses dict.
    """

initialize¶

async def initialize(self) -> None:
    """
    Initialize all components asynchronously.

    This method should be called after adding the middleware to the app,
    typically in a startup event handler.

    Tasks performed:
        - Build security check pipeline
        - Initialize Redis handlers (if enabled)
        - Initialize agent integrations (if enabled)
        - Initialize dynamic rule manager (if configured)

    Example:
        ```python
        @app.on_event("startup")
        async def startup():
            # Get middleware instance
            for middleware in app.user_middleware:
                if isinstance(middleware.cls, SecurityMiddleware):
                    await middleware.cls.initialize()
        ```
    """

set_decorator_handler¶

def set_decorator_handler(
    self,
    decorator_handler: BaseSecurityDecorator | None
) -> None:
    """
    Set the SecurityDecorator instance for decorator support.

    This enables route-level security configuration via decorators.

    Args:
        decorator_handler: SecurityDecorator instance or None

    Example:
        ```python
        guard_deco = SecurityDecorator(config)
        middleware.set_decorator_handler(guard_deco)
        # Or set on app state:
        app.state.guard_decorator = guard_deco
        ```
    """

configure_cors¶

@staticmethod
def configure_cors(app: FastAPI, config: SecurityConfig) -> bool:
    """
    Configure FastAPI's CORS middleware based on SecurityConfig.

    This is a convenience method for setting up CORS.

    Args:
        app: FastAPI application instance
        config: Security configuration with CORS settings

    Returns:
        bool: True if CORS was configured, False otherwise

    Example:
        ```python
        SecurityMiddleware.configure_cors(app, config)
        app.add_middleware(SecurityMiddleware, config=config)
        ```
    """

Handler Integration¶

The middleware works with singleton handler instances:

All handler classes (IPBanManager, CloudManager, etc.) use the singleton pattern
The middleware initializes these existing instances conditionally based on configuration
IPInfoManager is only initialized when country filtering is enabled
CloudManager is only loaded when cloud provider blocking is configured
This selective loading improves performance when not all features are used

Initialization Process¶

The middleware uses HandlerInitializer to set up all handlers:

# In __init__
self.handler_initializer = HandlerInitializer(
    config=self.config,
    redis_handler=self.redis_handler,
    agent_handler=self.agent_handler,
    geo_ip_handler=self.geo_ip_handler,
    rate_limit_handler=self.rate_limit_handler,
    guard_decorator=self.guard_decorator,
)

# In initialize()
await self.handler_initializer.initialize_redis_handlers()
await self.handler_initializer.initialize_agent_integrations()

Redis Configuration¶

Enable Redis in SecurityConfig:

config = SecurityConfig(
    enable_redis=True,
    redis_url="redis://prod:6379/0",
    redis_prefix="prod_security:"
)

The middleware automatically initializes: - CloudManager cloud provider IP ranges - IPBanManager distributed banning - IPInfoManager IP geolocation - RateLimitManager rate limiting - RedisManager Redis caching - SusPatternsManager suspicious patterns

Proxy Security Configuration¶

The middleware supports secure handling of proxy headers:

config = SecurityConfig(
    trusted_proxies=["10.0.0.1", "192.168.1.0/24"],  # List of trusted proxy IPs/ranges
    trusted_proxy_depth=1,  # Number of proxies in the chain
    trust_x_forwarded_proto=True,  # Trust X-Forwarded-Proto header from trusted proxies
)

This prevents IP spoofing attacks through header manipulation.

Usage Examples¶

Basic Setup¶

from fastapi import FastAPI
from guard.middleware import SecurityMiddleware
from guard.models import SecurityConfig

app = FastAPI()

config = SecurityConfig(
    rate_limit=100,
    enable_https=True,
    enable_cors=True
)

app.add_middleware(SecurityMiddleware, config=config)

With Decorators¶

from fastapi import FastAPI
from guard.middleware import SecurityMiddleware
from guard.models import SecurityConfig
from guard.decorators import SecurityDecorator

app = FastAPI()

config = SecurityConfig(rate_limit=100)
guard_deco = SecurityDecorator(config)

# Apply decorators to routes
@app.get("/api/limited")
@guard_deco.rate_limit(requests=10, window=300)
def limited_endpoint():
    return {"data": "limited"}

# Add middleware and set decorator
app.add_middleware(SecurityMiddleware, config=config)
app.state.guard_decorator = guard_deco  # Required for decorator integration

With Async Initialization¶

from fastapi import FastAPI
from guard.middleware import SecurityMiddleware
from guard.models import SecurityConfig

app = FastAPI()

config = SecurityConfig(
    enable_redis=True,
    redis_url="redis://localhost:6379"
)

# Add middleware
middleware_instance = None
for mw in app.user_middleware:
    if isinstance(mw.cls, type) and issubclass(mw.cls, SecurityMiddleware):
        middleware_instance = mw.cls
        break

app.add_middleware(SecurityMiddleware, config=config)

@app.on_event("startup")
async def startup():
    # Initialize async components
    if middleware_instance:
        await middleware_instance.initialize()

Internal Architecture¶

For Contributors

The internal architecture is documented in Core Architecture. This section provides a high-level overview.

Modular Design (v4.2.0+)¶

The middleware delegates to specialized modules in guard/core/:

checks/: Security check implementations (Chain of Responsibility pattern)
events/: Event bus and metrics collection
initialization/: Handler initialization logic
responses/: Response creation and processing
routing/: Route configuration resolution
validation/: Request validation utilities
bypass/: Security bypass handling
behavioral/: Behavioral rule processing

Benefits of Modular Architecture¶

Maintainability: Each module < 200 LOC, single responsibility
Testability: Each component independently testable
Performance: Better caching and optimization opportunities
Extensibility: Easy to add new checks or modify behavior
Development Speed: 2-3x faster feature additions (projected)