Core Architecture (Internal)¶

Internal Implementation Details

As of v5.0.0, all core modules live in the guard-core package (guard_core/core/). They should NOT be imported directly in user code. Always use the public API via from tornadoapi_guard import ... or from tornadoapi_guard import SecurityMiddleware.

This documentation is provided for contributors and advanced users who want to understand the internal architecture.

Overview¶

TornadoAPI Guard uses a modular core architecture introduced in v4.2.0 that separates security concerns into specialized, independently testable modules. This transformation improved maintainability from MI 0.00 (Rank C) to MI 54.51 (Rank A) while maintaining 100% backward compatibility.

Architecture Principles¶

Single Responsibility: Each module has one clear purpose
Dependency Injection: Clean dependencies via context objects
Chain of Responsibility: Security checks execute in a pipeline
Testability: Each component independently testable
Performance: Better caching and optimization opportunities

Module Overview¶

guard_core/core/
├── checks/              # Security check pipeline (Chain of Responsibility)
├── 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

Module: guard_core/core/checks/¶

Purpose: Security check implementations using the Chain of Responsibility pattern

Pattern: Pipeline-based execution where each check can block or pass the request

Key Components¶

SecurityCheck (Base Class)¶

Abstract base class for all security checks.

Location: guard_core/core/checks/base.py

Key Methods:

class SecurityCheck:
    check_name: str

    async def check(self, request: GuardRequest) -> GuardResponse | None: ...

    async def create_error_response(
        self, status_code: int, default_message: str
    ) -> GuardResponse: ...

Usage: All security checks inherit from this base class.

SecurityCheckPipeline¶

Orchestrates the execution of multiple security checks in sequence.

Location: guard_core/core/checks/pipeline.py

Key Methods:

class SecurityCheckPipeline:
    def __init__(self, checks: list[SecurityCheck]): ...

    async def execute(self, request: GuardRequest) -> GuardResponse | None: ...

    def get_check_names(self) -> list[str]: ...

Features:

Sequential execution with early termination
Error handling with fail-secure option
Logging of blocking checks
Dynamic check addition/removal

Check Implementations¶

Location: guard_core/core/checks/implementations/

Execution Order¶

Checks execute in this order (defined in middleware.py._build_security_pipeline()):

RouteConfigCheck - Extract route config and client IP
EmergencyModeCheck - Emergency mode (highest priority)
HttpsEnforcementCheck - HTTPS enforcement (can redirect)
RequestLoggingCheck - Log request
RequestSizeContentCheck - Validate size/content
RequiredHeadersCheck - Check required headers
AuthenticationCheck - Verify authentication
ReferrerCheck - Check referrer
CustomValidatorsCheck - Custom validators
TimeWindowCheck - Time-based access
CloudIpRefreshCheck - Periodic maintenance
IpSecurityCheck - IP whitelist/blacklist
CloudProviderCheck - Cloud provider blocking
UserAgentCheck - User agent filtering
RateLimitCheck - Rate limiting
SuspiciousActivityCheck - Threat detection
CustomRequestCheck - Custom checks

Implementation Details¶

Each check is self-contained and follows this pattern:

from guard_core.core.checks.base import SecurityCheck
from guard_core.protocols import GuardRequest, GuardResponse


class ExampleCheck(SecurityCheck):
    check_name = "example_check"

    def __init__(self, middleware: "SecurityMiddleware"):
        self.middleware = middleware
        self.config = middleware.config
        self.logger = middleware.logger

    async def check(self, request: GuardRequest) -> GuardResponse | None:
        client_ip = request.state.client_ip
        route_config = request.state.route_config

        if should_block:
            return await self.create_error_response(403, "Check failed")

        return None

Helper Functions¶

Location: guard_core/core/checks/helpers.py

Common utilities shared across check implementations:

check_route_ip_access() - IP whitelist/blacklist validation
check_user_agent_allowed() - User agent validation
get_client_country() - Country lookup
Other shared validation logic

Module: guard_core/core/events/¶

Purpose: Event system for middleware actions and metrics collection

SecurityEventBus¶

Centralized event dispatching for security-related events.

Location: guard_core/core/events/middleware_events.py

Key Methods:

class SecurityEventBus:
    async def send_middleware_event(
        self,
        event_type: str,
        request: GuardRequest,
        action_taken: str,
        reason: str,
        **kwargs,
    ) -> None: ...

    async def send_https_violation_event(
        self, request: GuardRequest, route_config: RouteConfig | None
    ) -> None: ...

    async def send_cloud_detection_events(
        self,
        request: GuardRequest,
        client_ip: str,
        cloud_providers: list[str],
        route_config: RouteConfig | None,
        cloud_handler: Any,
        passive_mode: bool,
    ) -> None: ...

Usage: Sends events to optional monitoring agent for observability.

MetricsCollector¶

Collects and reports request metrics.

Location: guard_core/core/events/metrics.py

Key Methods:

class MetricsCollector:
    async def send_security_metric(
        self, metric_name: str, value: float, tags: dict[str, str] | None = None
    ) -> None: ...

    async def collect_request_metrics(
        self,
        request: GuardRequest,
        response: GuardResponse,
        response_time: float,
        client_ip: str,
    ) -> None: ...

Module: guard_core/core/initialization/¶

Purpose: Centralized handler initialization logic

HandlerInitializer¶

Manages initialization of Redis, Agent, and other handlers.

Location: guard_core/core/initialization/handler_initializer.py

Key Methods:

class HandlerInitializer:
    async def initialize_redis_handlers(self) -> None:
        """Initialize Redis for all handlers."""
        ...

    async def initialize_agent_for_handlers(self) -> None:
        """Initialize agent in all handlers."""
        ...

    async def initialize_agent_integrations(self) -> None:
        """Initialize agent and its integrations."""
        ...

    async def initialize_dynamic_rule_manager(self) -> None:
        """Initialize dynamic rule manager if configured."""
        ...

Usage: Called from SecurityMiddleware.initialize() to set up all handlers.

Module: guard_core/core/responses/¶

Purpose: Response creation and processing

ErrorResponseFactory¶

Centralized response creation, error handling, and header application.

Location: guard_core/core/responses/factory.py

Key Methods:

class ErrorResponseFactory:
    async def create_error_response(
        self, status_code: int, default_message: str
    ) -> GuardResponse: ...

    async def create_https_redirect(self, request: GuardRequest) -> GuardResponse: ...

    async def process_response(
        self,
        request: GuardRequest,
        response: GuardResponse,
        response_time: float,
        route_config: RouteConfig | None,
        process_behavioral_rules: Callable | None = None,
    ) -> GuardResponse: ...

Features:

Custom error message support
Security header application
CORS header handling
Metrics collection
Behavioral rule processing

ResponseContext¶

Dependency injection container for response factory.

Location: guard_core/core/responses/context.py

Module: guard_core/core/routing/¶

Purpose: Route configuration and decorator resolution

RouteConfigResolver¶

Resolves route-specific security configuration from decorators.

Location: guard_core/core/routing/resolver.py

Key Methods:

class RouteConfigResolver:
    def get_route_config(self, request: GuardRequest) -> RouteConfig | None: ...

    def should_bypass_check(
        self, check_name: str, route_config: RouteConfig | None
    ) -> bool: ...

    def get_cloud_providers_to_check(
        self, route_config: RouteConfig | None
    ) -> list[str] | None: ...

Usage: Resolves decorator configurations for route-level security overrides.

RoutingContext¶

Dependency injection container for route resolver.

Location: guard_core/core/routing/context.py

Module: guard_core/core/validation/¶

Purpose: Request validation utilities

RequestValidator¶

Provides request validation logic for HTTPS, proxies, time windows, and path exclusions.

Location: guard_core/core/validation/validator.py

Key Methods:

class RequestValidator:
    def is_request_https(self, request: GuardRequest) -> bool: ...

    def is_trusted_proxy(self, client_ip: str) -> bool: ...

    async def check_time_window(
        self, time_restrictions: dict[str, str]
    ) -> bool: ...

    def is_path_excluded(self, path: str) -> bool: ...

ValidationContext¶

Dependency injection container for validator.

Location: guard_core/core/validation/context.py

Module: guard_core/core/bypass/¶

Purpose: Handle security bypass cases

BypassHandler¶

Handles passthrough cases and decorator-based security bypasses.

Location: guard_core/core/bypass/handler.py

Key Methods:

class BypassHandler:
    async def handle_passthrough(
        self,
        request: GuardRequest,
        call_next: Callable[[GuardRequest], Awaitable[GuardResponse]],
    ) -> GuardResponse | None: ...

    async def handle_security_bypass(
        self,
        request: GuardRequest,
        call_next: Callable[[GuardRequest], Awaitable[GuardResponse]],
        route_config: RouteConfig | None,
    ) -> GuardResponse | None: ...

Features:

No client IP detection (passthrough)
Excluded path handling
Decorator bypass support
Early exit optimization

BypassContext¶

Dependency injection container for bypass handler.

Location: guard_core/core/bypass/context.py

Module: guard_core/core/behavioral/¶

Purpose: Behavioral rule processing

BehavioralProcessor¶

Processes decorator-based behavioral rules for usage and return monitoring.

Location: guard_core/core/behavioral/processor.py

Key Methods:

class BehavioralProcessor:
    async def process_usage_rules(
        self, request: GuardRequest, client_ip: str, route_config: RouteConfig
    ) -> None: ...

    async def process_return_rules(
        self,
        request: GuardRequest,
        response: GuardResponse,
        client_ip: str,
        route_config: RouteConfig,
    ) -> None: ...

    def get_endpoint_id(self, request: GuardRequest) -> str: ...

BehavioralContext¶

Dependency injection container for behavioral processor.

Location: guard_core/core/behavioral/context.py

Dependency Injection Pattern¶

All core modules use Context objects for clean dependency injection:

# Example: ResponseContext
@dataclass
class ResponseContext:
    config: SecurityConfig
    logger: logging.Logger
    metrics_collector: MetricsCollector
    agent_handler: Any
    guard_decorator: BaseSecurityDecorator | None

Benefits:

Explicit dependencies
Easy testing with mocks
Prevents tight coupling
Type-safe dependency passing

Adding New Security Checks¶

To add a custom security check:

Create Implementation

File: guard_core/core/checks/implementations/my_custom_check.py

from guard_core.core.checks.base import SecurityCheck
from guard_core.protocols import GuardRequest, GuardResponse


class MyCustomCheck(SecurityCheck):
    check_name = "my_custom_check"

    def __init__(self, middleware: "SecurityMiddleware"):
        self.middleware = middleware
        self.config = middleware.config
        self.logger = middleware.logger

    async def check(self, request: GuardRequest) -> GuardResponse | None:
        client_ip = request.state.client_ip
        route_config = request.state.route_config

        if condition_fails:
            self.logger.warning(f"Check failed for {client_ip}")
            return await self.create_error_response(
                403, "Custom check failed"
            )

        return None

Register in Pipeline

File: guard/middleware.py in _build_security_pipeline() method

from guard_core.core.checks.implementations.my_custom_check import MyCustomCheck

def _build_security_pipeline(self) -> None:
    checks = [
        # ... existing checks
        MyCustomCheck(self),  # Add your check
    ]
    self.security_pipeline = SecurityCheckPipeline(checks)

Export (Optional)

File: guard_core/core/checks/__init__.py

from guard_core.core.checks.implementations.my_custom_check import MyCustomCheck

__all__ = [
    # ... existing exports
    "MyCustomCheck",
]

Test

import pytest
from guard_core.core.checks.implementations.my_custom_check import MyCustomCheck

@pytest.mark.asyncio
async def test_my_custom_check(test_middleware, test_request):
    check = MyCustomCheck(test_middleware)
    response = await check.check(test_request)
    assert response is None  # or assert response.status_code == 403

Testing Core Modules¶

Each module is independently testable:

# Test a specific check
from guard_core.core.checks.implementations import IpSecurityCheck

async def test_ip_security():
    middleware = create_test_middleware()
    check = IpSecurityCheck(middleware)

    request = create_test_request(client_ip="10.0.0.1")
    response = await check.check(request)

    assert response is None  # IP allowed

# Test the pipeline
from guard_core.core.checks import SecurityCheckPipeline

async def test_pipeline():
    checks = [Check1(middleware), Check2(middleware)]
    pipeline = SecurityCheckPipeline(checks)

    response = await pipeline.execute(request)
    assert response is None  # All checks passed

Performance Considerations¶

Pipeline Optimization¶

Early Termination: Pipeline stops at first blocking check
Conditional Checks: Some checks skip execution based on config
Cached Results: Some checks cache results per request

Modular Benefits¶

Independent Caching: Each module can cache independently
Targeted Optimization: Identify and optimize specific bottlenecks
Profiling: Easy to profile individual checks

Migration from Monolithic Architecture¶

The modular architecture provides several improvements over the original monolithic middleware.py:

Aspect	Old (Monolithic)	New (Modular)
File Size	2,087 LOC	471 LOC + 9 modules
Maintainability	MI 0.00 (C)	MI 54.51 (A)
Complexity	~15 avg	2.35 avg
Testability	Hard (coupled)	Easy (isolated)
Extensibility	Difficult	Simple

For Contributors: All new features should follow the modular pattern.

References¶

Full Architecture Documentation: See ARCHITECTURE_CHANGES.md in project root
Public API Documentation: See other API reference pages
Task Documentation: .claude/tasks/COMPLETED/002-* and 003-*
Code Quality Metrics: All core modules achieve MI 56-82 (Rank A)

Core Architecture (Internal)¶

Overview¶

Architecture Principles¶

Module Overview¶

Module: guard_core/core/checks/¶

Key Components¶

SecurityCheck (Base Class)¶

SecurityCheckPipeline¶

Check Implementations¶

Execution Order¶

Implementation Details¶

Helper Functions¶

Module: guard_core/core/events/¶

SecurityEventBus¶

MetricsCollector¶

Module: guard_core/core/initialization/¶

HandlerInitializer¶

Module: guard_core/core/responses/¶

ErrorResponseFactory¶

ResponseContext¶

Module: guard_core/core/routing/¶

RouteConfigResolver¶

RoutingContext¶

Module: guard_core/core/validation/¶

RequestValidator¶

ValidationContext¶

Module: guard_core/core/bypass/¶

BypassHandler¶

BypassContext¶

Module: guard_core/core/behavioral/¶

BehavioralProcessor¶

BehavioralContext¶

Dependency Injection Pattern¶

Adding New Security Checks¶

Testing Core Modules¶

Performance Considerations¶

Pipeline Optimization¶

Modular Benefits¶

Migration from Monolithic Architecture¶

References¶

See Also¶