Protocol Reference¶
guard-core defines its boundaries through typing.Protocol classes. Adapter developers must provide concrete implementations of these protocols to bridge their framework's native types into the guard-core engine.
All protocols are decorated with @runtime_checkable, meaning you can verify conformance at runtime with isinstance().
GuardRequest¶
Location: guard_core/protocols/request_protocol.py
The abstraction over any HTTP request object. Every security check receives a GuardRequest.
from collections.abc import Mapping
from typing import Any, Protocol, runtime_checkable
@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]: ...
Property Reference¶
| Property | 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 complete URL including scheme, host, path, and query string |
url_replace_scheme(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 when unavailable (e.g. Unix sockets) |
headers |
Mapping[str, str] |
HTTP headers as a read-only mapping. Header names are case-insensitive in HTTP but the mapping implementation determines lookup behavior |
query_params |
Mapping[str, str] |
URL query parameters as a read-only mapping |
body() |
async -> bytes |
The raw request body. This is an async method because some frameworks read the body lazily |
state |
Any |
A mutable state object for attaching per-request data. guard-core stores client_ip, route_config, and is_whitelisted here |
scope |
dict[str, Any] |
ASGI-style scope dict. Must contain "app" key for route resolution to work. Also checked for "route" key by behavioral processing |
Framework Mapping¶
This table shows what each property maps to in common frameworks:
| GuardRequest | FastAPI/Starlette | Flask | Django |
|---|---|---|---|
url_path |
request.url.path |
request.path |
request.path |
url_scheme |
request.url.scheme |
request.scheme |
request.scheme |
url_full |
str(request.url) |
request.url |
request.build_absolute_uri() |
url_replace_scheme(s) |
str(request.url.replace(scheme=s)) |
manual construction | manual construction |
method |
request.method |
request.method |
request.method |
client_host |
request.client.host |
request.remote_addr |
request.META['REMOTE_ADDR'] |
headers |
request.headers |
request.headers |
request.headers (Django 2.2+) |
query_params |
request.query_params |
request.args |
request.GET |
body() |
await request.body() |
request.get_data() |
request.body |
state |
request.state |
custom attribute bag | custom attribute bag |
scope |
request.scope |
synthetic dict | synthetic dict |
The state object
guard-core writes to request.state during pipeline execution. The following attributes are set by built-in checks:
state.client_ip-- set byRouteConfigCheck(check #1)state.route_config-- set byRouteConfigCheck(check #1)state.is_whitelisted-- set byIpSecurityCheck(check #12)
Your adapter's GuardRequest.state implementation must support arbitrary attribute assignment.
The scope dict
Route resolution depends on request.scope["app"] having a .routes attribute and request.scope["app"].state.guard_decorator for decorator lookup. For non-ASGI frameworks, you will need to synthesize this scope dict with at least the "app" key pointing to an object that exposes .routes and .state.
GuardResponse¶
Location: guard_core/protocols/response_protocol.py
The abstraction over any HTTP response object.
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: ...
Property Reference¶
| Property | Type | Description |
|---|---|---|
status_code |
int |
HTTP status code (e.g. 403, 429, 503) |
headers |
MutableMapping[str, str] |
Response headers. Must be mutable -- guard-core writes security headers, CORS headers, and HSTS headers into this mapping |
body |
bytes \| None |
The response body, or None if no body |
Framework Mapping¶
| GuardResponse | FastAPI/Starlette | Flask | Django |
|---|---|---|---|
status_code |
response.status_code |
response.status_code |
response.status_code |
headers |
response.headers |
response.headers |
response.headers (Django HttpResponse) |
body |
response.body |
response.get_data() |
response.content |
MutableMapping requirement
The headers property must return a MutableMapping[str, str]. guard-core's ErrorResponseFactory writes security headers directly into response.headers[header_name] = header_value. If your framework's response headers are immutable, you must wrap them.
GuardResponseFactory¶
Location: guard_core/protocols/response_protocol.py
A factory protocol for creating response objects. The adapter provides this so guard-core can create error responses and redirects without knowing the framework's response class.
@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 Reference¶
| Method | Parameters | Returns | Purpose |
|---|---|---|---|
create_response |
content: str, status_code: int |
GuardResponse |
Creates a plain-text or JSON error response |
create_redirect_response |
url: str, status_code: int |
GuardResponse |
Creates an HTTP redirect response (used for HTTPS enforcement) |
Example Implementation (FastAPI/Starlette)¶
from starlette.responses import JSONResponse, RedirectResponse
from guard_core.protocols.response_protocol import GuardResponse, GuardResponseFactory
class StarletteResponseFactory:
def create_response(self, content: str, status_code: int) -> GuardResponse:
return JSONResponse(
content={"detail": content},
status_code=status_code,
)
def create_redirect_response(self, url: str, status_code: int) -> GuardResponse:
return RedirectResponse(url=url, status_code=status_code)
The factory instance is passed to ResponseContext.response_factory and used by ErrorResponseFactory internally.
GuardMiddlewareProtocol¶
Location: guard_core/protocols/middleware_protocol.py
Defines what the middleware instance must expose so that security checks can access shared state and services. Your adapter's middleware class must satisfy this protocol.
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from guard_core.models import SecurityConfig
@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: ...
Attribute Reference¶
| Attribute | Type | Description |
|---|---|---|
config |
SecurityConfig |
The security configuration object |
logger |
logging.Logger |
Logger instance used by all checks |
last_cloud_ip_refresh |
int |
Unix timestamp of the last cloud IP range refresh |
suspicious_request_counts |
dict[str, int] |
Per-IP suspicious request counters (for auto-ban threshold) |
Property Reference¶
| Property | Typical Type | Description |
|---|---|---|
event_bus |
SecurityEventBus |
Dispatches security events to the agent |
route_resolver |
RouteConfigResolver |
Resolves route-level decorator configurations |
response_factory |
ErrorResponseFactory |
Creates error responses and applies security headers |
rate_limit_handler |
RateLimitManager |
Checks and enforces rate limits |
agent_handler |
AgentHandlerProtocol \| None |
Guard agent telemetry client (or None if disabled) |
geo_ip_handler |
GeoIPHandler \| None |
GeoIP lookup handler (or None if not configured) |
guard_response_factory |
GuardResponseFactory |
The adapter-provided response factory |
Method Reference¶
| Method | Description |
|---|---|
create_error_response(status_code, default_message) |
Delegates to ErrorResponseFactory.create_error_response(). Applies custom error messages, security headers, and response modifiers |
refresh_cloud_ip_ranges() |
Triggers a refresh of cloud provider IP ranges from upstream sources |
Security checks access the middleware
Every SecurityCheck subclass receives a GuardMiddlewareProtocol reference in its constructor. Checks use self.middleware.event_bus, self.middleware.route_resolver, self.middleware.rate_limit_handler, etc. to access shared services.
Additional Protocols¶
GeoIPHandler¶
Location: guard_core/protocols/geo_ip_protocol.py
@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: ...
guard-core ships with IPInfoManager as a default implementation. Adapters can provide custom implementations for other GeoIP providers.
RedisHandlerProtocol¶
Location: guard_core/protocols/redis_protocol.py
@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]: ...
guard-core ships with RedisManager as the default implementation using the redis Python package.
AgentHandlerProtocol¶
Location: guard_core/protocols/agent_protocol.py
@runtime_checkable
class AgentHandlerProtocol(Protocol):
async def initialize_redis(self, redis_handler: RedisHandlerProtocol) -> None: ...
async def send_event(self, event: Any) -> None: ...
async def send_metric(self, metric: Any) -> None: ...
async def start(self) -> None: ...
async def stop(self) -> None: ...
async def flush_buffer(self) -> None: ...
async def get_dynamic_rules(self) -> Any | None: ...
async def health_check(self) -> bool: ...
Implemented by the guard-agent package. Only active when SecurityConfig.enable_agent = True.