Protocols¶

Protocols are the most important API surface for adapter developers. Guard-core uses Python Protocol classes (PEP 544) to define the contracts that adapters must satisfy. All protocols are @runtime_checkable.

GuardRequest¶

`guard_core.protocols.request_protocol.GuardRequest` ¶

Bases: Protocol

`client_host` `property` ¶

`headers` `property` ¶

`method` `property` ¶

`query_params` `property` ¶

`scope` `property` ¶

`state` `property` ¶

`url_full` `property` ¶

`url_path` `property` ¶

`url_scheme` `property` ¶

`body()` `async` ¶

Source code in guard_core/protocols/request_protocol.py

async def body(self) -> bytes: ...

`url_replace_scheme(scheme)` ¶

Source code in guard_core/protocols/request_protocol.py

def url_replace_scheme(self, scheme: str) -> str: ...

The request protocol defines how guard-core reads incoming request data. Adapters must wrap their framework's request object to satisfy this interface.

@runtime_checkable
class GuardRequest(Protocol):
    @property
    def url_path(self) -> str: ...
    @property
    def url_scheme(self) -> str: ...
    @property
    def url_full(self) -> str: ...
    def url_replace_scheme(self, scheme: str) -> str: ...
    @property
    def method(self) -> str: ...
    @property
    def client_host(self) -> str | None: ...
    @property
    def headers(self) -> Mapping[str, str]: ...
    @property
    def query_params(self) -> Mapping[str, str]: ...
    async def body(self) -> bytes: ...
    @property
    def state(self) -> Any: ...
    @property
    def scope(self) -> dict[str, Any]: ...

Member Details¶

Member	Return Type	Description
`url_path`	`str`	The path component of the URL (e.g., `"/api/users"`).
`url_scheme`	`str`	The URL scheme (`"http"` or `"https"`).
`url_full`	`str`	The full URL string including scheme, host, path, and query.
`url_replace_scheme`	`str`	Returns the full URL with the scheme replaced. Used for HTTPS redirects.
`method`	`str`	The HTTP method (`"GET"`, `"POST"`, etc.).
`client_host`	`str \\| None`	The connecting client's IP address. `None` if unavailable.
`headers`	`Mapping[str, str]`	Request headers as a read-only mapping. Case handling depends on adapter.
`query_params`	`Mapping[str, str]`	URL query parameters as a read-only mapping.
`body()`	`bytes` (async)	The raw request body. May be called multiple times by detection checks.
`state`	`Any`	A mutable state object for passing data between checks. Must support attribute assignment.
`scope`	`dict[str, Any]`	ASGI-style scope dict. Used for route resolution (`scope["route"]`).

Implementation Notes¶

state: Guard-core sets these attributes on state during pipeline execution:

state.route_config -- RouteConfig | None
state.client_ip -- str
state.is_whitelisted -- bool

The adapter's state object must support dynamic attribute assignment (e.g., a simple namespace or the framework's built-in state).

scope: Must include a "route" key with an object that has an endpoint attribute for decorator resolution. If the framework does not have route objects in scope, decorator-based features will not activate.

body(): The detection engine calls body() to scan request bodies for threats. Adapters should ensure the body is buffered and can be read multiple times (not consumed on first read).

headers: Must be iterable via .items() for header scanning. Guard-core reads headers case-insensitively in many places (e.g., headers.get("User-Agent")), but the mapping itself does not need to be case-insensitive.

GuardResponse¶

`guard_core.protocols.response_protocol.GuardResponse` ¶

Bases: Protocol

`body` `property` ¶

`headers` `property` ¶

`status_code` `property` ¶

The response protocol defines how guard-core reads and modifies outgoing responses.

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

Member Details¶

Member	Return Type	Description
`status_code`	`int`	HTTP status code.
`headers`	`MutableMapping[str, str]`	Response headers. Must be mutable for security header injection.
`body`	`bytes \\| None`	Response body bytes. Used by behavioral return pattern matching.

GuardResponseFactory¶

`guard_core.protocols.response_protocol.GuardResponseFactory` ¶

Bases: Protocol

`create_redirect_response(url, status_code)` ¶

Source code in guard_core/protocols/response_protocol.py

def create_redirect_response(self, url: str, status_code: int) -> GuardResponse: ...

`create_response(content, status_code)` ¶

Source code in guard_core/protocols/response_protocol.py

def create_response(self, content: str, status_code: int) -> GuardResponse: ...

Adapters must provide a factory that creates framework-native response objects.

@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: ...

Method	Purpose
`create_response`	Creates a plain text/JSON error response.
`create_redirect_response`	Creates an HTTP redirect (used for HTTPS enforcement).

GuardMiddlewareProtocol¶

`guard_core.protocols.middleware_protocol.GuardMiddlewareProtocol` ¶

Bases: Protocol

`agent_handler` `property` ¶

`config` `instance-attribute` ¶

`event_bus` `property` ¶

`geo_ip_handler` `property` ¶

`guard_response_factory` `property` ¶

`last_cloud_ip_refresh` `instance-attribute` ¶

`logger` `instance-attribute` ¶

`rate_limit_handler` `property` ¶

`response_factory` `property` ¶

`route_resolver` `property` ¶

`suspicious_request_counts` `instance-attribute` ¶

`create_error_response(status_code, default_message)` `async` ¶

Source code in guard_core/protocols/middleware_protocol.py

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

`refresh_cloud_ip_ranges()` `async` ¶

Source code in guard_core/protocols/middleware_protocol.py

async def refresh_cloud_ip_ranges(self) -> None: ...

Defines the interface that the adapter's middleware class must expose to the security check pipeline.

@runtime_checkable
class GuardMiddlewareProtocol(Protocol):
    config: SecurityConfig
    logger: logging.Logger
    last_cloud_ip_refresh: int
    suspicious_request_counts: dict[str, int]

    @property
    def event_bus(self) -> Any: ...
    @property
    def route_resolver(self) -> Any: ...
    @property
    def response_factory(self) -> Any: ...
    @property
    def rate_limit_handler(self) -> Any: ...
    @property
    def agent_handler(self) -> Any: ...
    @property
    def geo_ip_handler(self) -> Any: ...
    @property
    def guard_response_factory(self) -> Any: ...

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

    async def refresh_cloud_ip_ranges(self) -> None: ...

Attributes¶

Attribute	Type	Description
`config`	`SecurityConfig`	The security configuration.
`logger`	`logging.Logger`	Logger instance for the middleware.
`last_cloud_ip_refresh`	`int`	Timestamp of the last cloud IP refresh.
`suspicious_request_counts`	`dict[str, int]`	Per-IP suspicious request counters for auto-ban.

Properties¶

Property	Purpose
`event_bus`	`SecurityEventBus` for emitting security events.
`route_resolver`	`RouteConfigResolver` for decorator resolution.
`response_factory`	`ErrorResponseFactory` for creating error responses.
`rate_limit_handler`	`RateLimitManager` instance.
`agent_handler`	Agent handler or `None`.
`geo_ip_handler`	`GeoIPHandler` or `None`.
`guard_response_factory`	`GuardResponseFactory` from the adapter.

GeoIPHandler¶

`guard_core.protocols.geo_ip_protocol.GeoIPHandler` ¶

Bases: Protocol

`is_initialized` `property` ¶

`get_country(ip)` ¶

Source code in guard_core/protocols/geo_ip_protocol.py

def get_country(self, ip: str) -> str | None: ...

`initialize()` `async` ¶

Source code in guard_core/protocols/geo_ip_protocol.py

async def initialize(self) -> None: ...

`initialize_agent(agent_handler)` `async` ¶

Source code in guard_core/protocols/geo_ip_protocol.py

async def initialize_agent(self, agent_handler: AgentHandlerProtocol) -> None: ...

`initialize_redis(redis_handler)` `async` ¶

Source code in guard_core/protocols/geo_ip_protocol.py

async def initialize_redis(self, redis_handler: RedisHandlerProtocol) -> None: ...

Protocol for geolocation services. Adapters can provide any implementation (MaxMind, IPInfo, custom).

@runtime_checkable
class GeoIPHandler(Protocol):
    @property
    def is_initialized(self) -> bool: ...
    async def initialize(self) -> None: ...
    async def initialize_redis(self, redis_handler: RedisHandlerProtocol) -> None: ...
    async def initialize_agent(self, agent_handler: AgentHandlerProtocol) -> None: ...
    def get_country(self, ip: str) -> str | None: ...

Method	Description
`is_initialized`	Whether the handler has been initialized.
`initialize()`	Perform async initialization (e.g., download DB).
`get_country(ip)`	Return ISO country code for the IP, or `None`.

RedisHandlerProtocol¶

`guard_core.protocols.redis_protocol.RedisHandlerProtocol` ¶

Bases: Protocol

`delete(namespace, key)` `async` ¶

Source code in guard_core/protocols/redis_protocol.py

async def delete(self, namespace: str, key: str) -> int | None: ...

`get_connection()` ¶

Source code in guard_core/protocols/redis_protocol.py

def get_connection(self) -> AsyncContextManager[Any]: ...

`get_key(namespace, key)` `async` ¶

Source code in guard_core/protocols/redis_protocol.py

async def get_key(self, namespace: str, key: str) -> Any: ...

`initialize()` `async` ¶

Source code in guard_core/protocols/redis_protocol.py

async def initialize(self) -> None: ...

`keys(pattern)` `async` ¶

Source code in guard_core/protocols/redis_protocol.py

async def keys(self, pattern: str) -> list[str] | None: ...

`set_key(namespace, key, value, ttl=None)` `async` ¶

Source code in guard_core/protocols/redis_protocol.py

async def set_key(
    self, namespace: str, key: str, value: Any, ttl: int | None = None
) -> bool | None: ...

Protocol for Redis operations. Matches the RedisManager interface.

@runtime_checkable
class RedisHandlerProtocol(Protocol):
    async def get_key(self, namespace: str, key: str) -> Any: ...
    async def set_key(self, namespace: str, key: str, value: Any, ttl: int | None = None) -> bool | None: ...
    async def delete(self, namespace: str, key: str) -> int | None: ...
    async def keys(self, pattern: str) -> list[str] | None: ...
    async def initialize(self) -> None: ...
    def get_connection(self) -> AsyncContextManager[Any]: ...