ccproxy.core.plugins.middleware

ccproxy.core.plugins.middleware

Middleware management and ordering for the plugin system.

This module provides utilities for managing middleware registration and ensuring proper ordering across core and plugin middleware.

CoreMiddlewareSpec dataclass

CoreMiddlewareSpec(
    middleware_class,
    priority=APPLICATION,
    kwargs=dict(),
    source="core",
)

Bases: MiddlewareSpec

Specification for core application middleware.

Extends MiddlewareSpec with a source field to distinguish between core and plugin middleware.

MiddlewareManager

MiddlewareManager()

Manages middleware registration and ordering.

Source code in ccproxy/core/plugins/middleware.py

def __init__(self) -> None:
    """Initialize middleware manager."""
    self.middleware_specs: list[CoreMiddlewareSpec] = []

add_core_middleware

add_core_middleware(
    middleware_class, priority=APPLICATION, **kwargs
)

Add core application middleware.

Parameters:

Name	Type	Description	Default
middleware_class	type[BaseHTTPMiddleware]	Middleware class	required
priority	int	Priority for ordering	APPLICATION
**kwargs	Any	Additional middleware arguments	{}

Source code in ccproxy/core/plugins/middleware.py

def add_core_middleware(
    self,
    middleware_class: type[BaseHTTPMiddleware],
    priority: int = MiddlewareLayer.APPLICATION,
    **kwargs: Any,
) -> None:
    """Add core application middleware.

    Args:
        middleware_class: Middleware class
        priority: Priority for ordering
        **kwargs: Additional middleware arguments
    """
    spec = CoreMiddlewareSpec(
        middleware_class=middleware_class,
        priority=priority,
        kwargs=kwargs,
        source="core",
    )
    self.middleware_specs.append(spec)

add_plugin_middleware

add_plugin_middleware(plugin_name, specs)

Add middleware from a plugin.

Parameters:

Name	Type	Description	Default
plugin_name	str	Name of the plugin	required
specs	list[MiddlewareSpec]	List of middleware specifications	required

Source code in ccproxy/core/plugins/middleware.py

def add_plugin_middleware(
    self, plugin_name: str, specs: list[MiddlewareSpec]
) -> None:
    """Add middleware from a plugin.

    Args:
        plugin_name: Name of the plugin
        specs: List of middleware specifications
    """
    for spec in specs:
        core_spec = CoreMiddlewareSpec(
            middleware_class=spec.middleware_class,
            priority=spec.priority,
            kwargs=spec.kwargs,
            source=plugin_name,
        )
        self.middleware_specs.append(core_spec)
        logger.trace(
            "plugin_middleware_added",
            plugin=plugin_name,
            middleware=spec.middleware_class.__name__,
            priority=spec.priority,
            category="middleware",
        )

get_ordered_middleware

get_ordered_middleware()

Get all middleware sorted by priority.

Returns:

Type	Description
list[CoreMiddlewareSpec]	List of middleware specs sorted by priority (lower first)

Source code in ccproxy/core/plugins/middleware.py

def get_ordered_middleware(self) -> list[CoreMiddlewareSpec]:
    """Get all middleware sorted by priority.

    Returns:
        List of middleware specs sorted by priority (lower first)
    """
    # Sort by priority (lower values first)
    # Secondary sort by source (core before plugins) for same priority
    return sorted(
        self.middleware_specs,
        key=lambda x: (x.priority, x.source != "core", x.source),
    )

apply_to_app

apply_to_app(app)

Apply all middleware to the FastAPI app in correct order.

Note: Middleware in FastAPI/Starlette is applied in reverse order (last added runs first), so we add them in reverse priority order.

Parameters:

Name	Type	Description	Default
app	FastAPI	FastAPI application	required

Source code in ccproxy/core/plugins/middleware.py

def apply_to_app(self, app: FastAPI) -> None:
    """Apply all middleware to the FastAPI app in correct order.

    Note: Middleware in FastAPI/Starlette is applied in reverse order
    (last added runs first), so we add them in reverse priority order.

    Args:
        app: FastAPI application
    """
    ordered = self.get_ordered_middleware()
    applied_middleware = []
    failed_middleware = []

    # Apply in reverse order (highest priority last so it runs first)
    for spec in reversed(ordered):
        try:
            app.add_middleware(spec.middleware_class, **spec.kwargs)  # type: ignore[arg-type]
            applied_middleware.append(
                {
                    "name": spec.middleware_class.__name__,
                    "priority": spec.priority,
                    "source": spec.source,
                }
            )
        except Exception as e:
            failed_middleware.append(
                {
                    "name": spec.middleware_class.__name__,
                    "source": spec.source,
                    "error": str(e),
                }
            )
            logger.error(
                "middleware_application_failed",
                middleware=spec.middleware_class.__name__,
                source=spec.source,
                error=str(e),
                exc_info=e,
                category="middleware",
            )

    # Log aggregated success
    if applied_middleware:
        logger.info(
            "middleware_stack_configured",
            applied=len(applied_middleware),
            failed=len(failed_middleware),
            middleware=[m["name"] for m in applied_middleware],
            category="middleware",
        )

get_middleware_summary

get_middleware_summary()

Get a summary of registered middleware.

Returns:

Type	Description
dict[str, Any]	Dictionary with middleware statistics and order

Source code in ccproxy/core/plugins/middleware.py

def get_middleware_summary(self) -> dict[str, Any]:
    """Get a summary of registered middleware.

    Returns:
        Dictionary with middleware statistics and order
    """
    ordered = self.get_ordered_middleware()

    summary = {
        "total": len(ordered),
        "core": len([m for m in ordered if m.source == "core"]),
        "plugins": len([m for m in ordered if m.source != "core"]),
        "order": [
            {
                "name": spec.middleware_class.__name__,
                "priority": spec.priority,
                "layer": self._get_layer_name(spec.priority),
                "source": spec.source,
            }
            for spec in ordered
        ],
    }

    return summary

setup_default_middleware

setup_default_middleware(manager)

Setup default core middleware.

Parameters:

Name	Type	Description	Default
manager	MiddlewareManager	Middleware manager	required

Source code in ccproxy/core/plugins/middleware.py

def setup_default_middleware(manager: MiddlewareManager) -> None:
    """Setup default core middleware.

    Args:
        manager: Middleware manager
    """
    from ccproxy.api.middleware.hooks import HooksMiddleware
    from ccproxy.api.middleware.normalize_headers import NormalizeHeadersMiddleware
    from ccproxy.api.middleware.request_id import RequestIDMiddleware

    # Request ID should be first (lowest priority) to set context for all others
    manager.add_core_middleware(
        RequestIDMiddleware,
        priority=MiddlewareLayer.SECURITY - 50,  # Before security layer
    )

    # Hooks middleware should be early to capture all requests
    manager.add_core_middleware(
        HooksMiddleware,
        priority=MiddlewareLayer.SECURITY
        - 40,  # After request ID, before other middleware
    )

    # # Access logging in observability layer
    # manager.add_core_middleware(
    #     AccessLogMiddleware, priority=MiddlewareLayer.OBSERVABILITY
    # )
    #
    # Normalize headers: strip unsafe and ensure server header
    manager.add_core_middleware(
        NormalizeHeadersMiddleware,  # type: ignore[arg-type]
        priority=MiddlewareLayer.ROUTING,  # after routing layer
    )

    logger.debug("default_middleware_configured", category="middleware")