Implementing GuardResponse and GuardResponseFactory¶

Protocol Definitions¶

Both protocols live at guard_core/protocols/response_protocol.py:

from collections.abc import MutableMapping
from typing import Protocol, runtime_checkable


@runtime_checkable
class GuardResponse(Protocol):
    @property
    def status_code(self) -> int: ...
    @property
    def headers(self) -> MutableMapping[str, str]: ...
    @property
    def body(self) -> bytes | None: ...


@runtime_checkable
class GuardResponseFactory(Protocol):
    def create_response(self, content: str, status_code: int) -> GuardResponse: ...
    def create_redirect_response(self, url: str, status_code: int) -> GuardResponse: ...

GuardResponse¶

The response object returned by your factory and ultimately sent back to the client. The headers property must be mutable -- guard-core's ErrorResponseFactory writes security headers (HSTS, CSP, X-Frame-Options, etc.) directly onto the response via response.headers[header_name] = header_value.

GuardResponseFactory¶

A factory that produces GuardResponse instances. Guard-core's ErrorResponseFactory calls your factory's methods to build framework-native responses. You pass an instance of your factory into the ResponseContext, and the entire pipeline uses it from that point forward.

How ErrorResponseFactory Uses Your Factory¶

The ErrorResponseFactory in guard_core/core/responses/factory.py delegates response creation to your GuardResponseFactory:

class ErrorResponseFactory:
    def __init__(self, context: ResponseContext):
        self.context = context

    async def create_error_response(
        self, status_code: int, default_message: str
    ) -> GuardResponse:
        custom_message = self.context.config.custom_error_responses.get(
            status_code, default_message
        )
        response = self.context.response_factory.create_response(
            custom_message, status_code
        )
        response = await self.apply_security_headers(response)
        response = await self.apply_modifier(response)
        return response

    async def create_https_redirect(self, request: GuardRequest) -> GuardResponse:
        https_url = request.url_replace_scheme("https")
        redirect_response = self.context.response_factory.create_redirect_response(
            str(https_url), 301
        )
        return await self.apply_modifier(redirect_response)

The flow is:

A security check returns await self.create_error_response(403, "Forbidden").
This calls middleware.create_error_response(), which delegates to ErrorResponseFactory.
ErrorResponseFactory checks SecurityConfig.custom_error_responses for a user-defined message override.
Your GuardResponseFactory.create_response() builds the actual framework response.
Security headers are applied via response.headers[...] = ....
If SecurityConfig.custom_response_modifier is set, it receives the response for final transformation.

Implementation: FastAPI / Starlette¶

from collections.abc import MutableMapping

from starlette.responses import PlainTextResponse, RedirectResponse, Response


class StarletteGuardResponse:
    def __init__(self, response: Response) -> None:
        self._response = response

    @property
    def status_code(self) -> int:
        return self._response.status_code

    @property
    def headers(self) -> MutableMapping[str, str]:
        return self._response.headers

    @property
    def body(self) -> bytes | None:
        if hasattr(self._response, "body"):
            return self._response.body
        return None


class StarletteResponseFactory:
    def create_response(
        self, content: str, status_code: int
    ) -> StarletteGuardResponse:
        response = PlainTextResponse(content, status_code=status_code)
        return StarletteGuardResponse(response)

    def create_redirect_response(
        self, url: str, status_code: int
    ) -> StarletteGuardResponse:
        response = RedirectResponse(url=url, status_code=status_code)
        return StarletteGuardResponse(response)

If your framework's Response already has status_code, mutable headers, and body as properties, you may not need a wrapper at all. Starlette's Response satisfies all three natively, so a simpler approach is to use it directly:

class StarletteResponseFactory:
    def create_response(self, content: str, status_code: int) -> Response:
        return PlainTextResponse(content, status_code=status_code)

    def create_redirect_response(self, url: str, status_code: int) -> Response:
        return RedirectResponse(url=url, status_code=status_code)

This works because Starlette's Response structurally satisfies GuardResponse.

Implementation: Flask¶

Flask's Response has mutable headers but uses data instead of body:

from collections.abc import MutableMapping
from typing import Any

from flask import Response, redirect
from werkzeug.datastructures import Headers


class FlaskGuardResponse:
    def __init__(
        self, content: str = "", status_code: int = 200, headers: dict | None = None
    ) -> None:
        self._response = Response(content, status=status_code, headers=headers)

    @property
    def status_code(self) -> int:
        return self._response.status_code

    @property
    def headers(self) -> MutableMapping[str, str]:
        return self._response.headers

    @property
    def body(self) -> bytes | None:
        return self._response.data


class FlaskResponseFactory:
    def create_response(
        self, content: str, status_code: int
    ) -> FlaskGuardResponse:
        return FlaskGuardResponse(content, status_code)

    def create_redirect_response(
        self, url: str, status_code: int
    ) -> FlaskGuardResponse:
        return FlaskGuardResponse(
            f"Redirecting to {url}",
            status_code,
            {"Location": url},
        )

Implementation: Django¶

Django's HttpResponse has a dict-like headers property (Django 3.2+):

from collections.abc import MutableMapping

from django.http import HttpResponse, HttpResponseRedirect


class DjangoGuardResponse:
    def __init__(self, response: HttpResponse) -> None:
        self._response = response

    @property
    def status_code(self) -> int:
        return self._response.status_code

    @property
    def headers(self) -> MutableMapping[str, str]:
        return self._response.headers

    @property
    def body(self) -> bytes | None:
        return self._response.content


class DjangoResponseFactory:
    def create_response(
        self, content: str, status_code: int
    ) -> DjangoGuardResponse:
        response = HttpResponse(content, status=status_code)
        return DjangoGuardResponse(response)

    def create_redirect_response(
        self, url: str, status_code: int
    ) -> DjangoGuardResponse:
        response = HttpResponseRedirect(url)
        response.status_code = status_code
        return DjangoGuardResponse(response)

ResponseContext¶

Your GuardResponseFactory instance is passed into the ResponseContext dataclass, which is then used to construct the ErrorResponseFactory:

from guard_core.core.responses.context import ResponseContext
from guard_core.core.responses.factory import ErrorResponseFactory
from guard_core.core.events import MetricsCollector


response_context = ResponseContext(
    config=security_config,
    logger=logger,
    metrics_collector=metrics_collector,
    agent_handler=agent_handler,
    guard_decorator=guard_decorator,
    response_factory=your_response_factory,  # Your GuardResponseFactory instance
)

error_factory = ErrorResponseFactory(response_context)

The ResponseContext dataclass is defined in guard_core/core/responses/context.py:

@dataclass
class ResponseContext:
    config: SecurityConfig
    logger: Logger
    metrics_collector: MetricsCollector

    agent_handler: Any | None = None
    guard_decorator: BaseSecurityDecorator | None = None
    response_factory: Any = field(default=None)

Security Headers¶

ErrorResponseFactory.apply_security_headers() iterates over the headers produced by security_headers_manager.get_headers() and writes them onto your response:

async def apply_security_headers(
    self, response: GuardResponse, request_path: str | None = None
) -> GuardResponse:
    headers_config = self.context.config.security_headers
    if headers_config and headers_config.get("enabled", True):
        security_headers = await security_headers_manager.get_headers(request_path)
        for header_name, header_value in security_headers.items():
            response.headers[header_name] = header_value
    return response

This is why GuardResponse.headers must return a MutableMapping. If your framework's response uses an immutable header collection, you must wrap it with a mutable proxy.

Custom Response Modifier¶

If SecurityConfig.custom_response_modifier is set, ErrorResponseFactory.apply_modifier() passes every response through it:

async def apply_modifier(self, response: GuardResponse) -> GuardResponse:
    if self.context.config.custom_response_modifier:
        return await self.context.config.custom_response_modifier(response)
    return response

The modifier receives and returns a GuardResponse, so your wrapper must be what gets passed in. If you use framework-native responses directly (no wrapper), the modifier will receive the framework response -- which is fine as long as it satisfies GuardResponse.