Security Gateway¶

security_gateway — mvp.security_gateway

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview¶

Centralised security enforcement pipeline: authentication (API key), RBAC role checking, CBRN (chemical/biological/radiological/nuclear) content classification, per-tier rate limiting, CSF safety gating, and tamper-evident audit logging. Returns a structured SecurityVerdict with decision, stage evidence, warning card, and canonical completion_state (verified, qualified-draft, or blocked-escalated).

When to use:

• Enforcing a single security boundary across all G6 server surfaces
• Classifying requests for harmful content before routing to downstream agents
• Generating a compliance-ready audit log of all security decisions

Example:

from mvp.security_gateway import APIKeyStore, Role, SecurityGatewayBlock, SecurityRequest, Tier

keys = APIKeyStore()
api_key = keys.create_key("user_123", Role.USER, Tier.BASIC)

block = SecurityGatewayBlock(name="security", key_store=keys)
result = block.infer(SecurityRequest(
    api_key=api_key,
    target_component="grounding",
    operation="infer",
    params_hash="...",
    content_summary="Verify this workflow output.",
    source_ip="127.0.0.1",
    transport="rest",
))
verdict = result.value
# verdict.decision, verdict.reason, verdict.risk_score
# verdict.completion_state, verdict.warning_card, verdict.evidence

Diagnostics: gateway_health, gateway_capabilities, and MCP gateway_dry_run are read-only/dry-run surfaces. They do not expose API-key creation, RBAC mutation, audit deletion, or policy writes.

CBRN sessions: the default CBRN session accumulator is process-local memory. Set SECURITY_CBRN_SESSION_BACKEND=redis with Redis configuration for shared multi-worker accumulation.

Works well with: middleware, csf_gate, csf_audit

Public API¶

SecurityAuditLogger¶

Append-only security event log backed by SQLite.

Constructor:

Methods:

log_event(event_type: str, detail: dict, caller_id: str | None = None, source_ip: str | None = None, transport: str | None = None) -> None¶

Append a structured event to the audit log (hash-chained).

verify_hash_chain() -> tuple[bool, int | None]¶

Walk the chain and detect tampering.

query(event_type: str | None = None, limit: int = 100) -> list[dict]¶

Return recent events, optionally filtered by event_type.

prune_old_events(retention_days: int = 365) -> int¶

Delete events older than retention_days. Returns count deleted.

Authenticator¶

Verify caller identity from API keys.

Constructor:

Methods:

authenticate(request: SecurityRequest) -> Result[CallerIdentity]¶

Return a CallerIdentity or an error Result.

Authorizer¶

RBAC authoriser backed by SQLite.

Constructor:

Methods:

reload() -> None¶

Load rules from SQLite into memory.

authorize(caller: CallerIdentity, component: str, operation: str) -> Result[bool]¶

Check if caller may invoke component/operation.

authorize_pipeline(caller: CallerIdentity, steps: list[tuple[str, str]]) -> Result[bool]¶

Check RBAC for every step and enforce tier pipeline-step limits.

CBRNSessionBackend(Protocol)¶

Rolling CBRN session accumulation backend.

Methods:

accumulate(caller_id: str, score: float, window_seconds: float) -> float¶

Record score and return cumulative score within the rolling window.

MemoryCBRNSessionBackend¶

Process-local CBRN session backend.

Methods:

accumulate(caller_id: str, score: float, window_seconds: float) -> float¶

RedisCBRNSessionBackend¶

Redis sorted-set backend for shared multi-worker CBRN accumulation.

Constructor:

Methods:

accumulate(caller_id: str, score: float, window_seconds: float) -> float¶

CBRNRuleEngine¶

Regex-based additive scoring against a configurable rule set.

Constructor:

Methods:

score(content: str, component: str, operation: str) -> CBRNResult¶

Score content against all rules, apply context multiplier.

score_pipeline(content: str, steps: list[tuple[str, str]]) -> CBRNResult¶

Score content using the highest-risk component in steps, then add pipeline bonus.

LLMSafetyJudge¶

Optional Tier 2 judge backed by an Anthropic Claude model.

Constructor:

Methods:

evaluate(content: str, component: str, operation: str) -> CBRNResult¶

Call the LLM and parse a structured safety verdict. Fail-closed on error.

CBRNClassifier¶

Two-tier classifier with optional session-level accumulation.

Constructor:

Methods:

session_backend_kind() -> str¶

session_scope() -> str¶

classify(content: str, component: str, operation: str, caller_id: str = '') -> CBRNResult¶

Classify content. Escalate to LLM if rule engine is ambiguous.

classify_pipeline(content: str, steps: list[tuple[str, str]], caller_id: str = '') -> CBRNResult¶

Composition-aware classification across pipeline steps.

CSFChecker¶

Check operations against the G6 Computational Safety Framework.

Methods:

check(component: str, operation: str) -> Result[bool]¶

Check a single operation against the CSF safety budget.

check_pipeline(steps: list[tuple[str, str]]) -> Result[bool]¶

Check a pipeline of operations against the cumulative CSF budget.

SecurityGatewayBlock(AIBlock[SecurityRequest, SecurityVerdict, None])¶

Centralised 6-stage security pipeline for all G6 transports.

Methods:

infer(data: SecurityRequest) -> Result[SecurityVerdict]¶

Run the 6-stage security pipeline. Always returns Result.ok(SecurityVerdict).

reload_policy() -> None¶

Reload RBAC policy from database.

dry_run(data: SecurityRequest) -> SecurityVerdict¶

Run diagnostics without mutating this gateway's audit, limiter, or CBRN session state.

gateway_health() -> dict¶

Return read-only runtime diagnostics without executing gateway stages.

gateway_capabilities() -> dict¶

Return diagnostic capabilities; mutation operations are intentionally absent.

APIKeyStore¶

Manages API keys in a SQLite database. Keys are stored as SHA-256 hashes.

Constructor:

Methods:

create_key(caller_id: str, role: Role, tier: Tier) -> str¶

Generate a new API key. Returns the plaintext key (store it safely).

verify_key(key: str) -> CallerIdentity | None¶

Return CallerIdentity if key is valid and not revoked, else None.

revoke_key(key: str) -> None¶

Soft-revoke a key by its plaintext value.

revoke_by_caller_id(caller_id: str) -> None¶

Revoke all keys belonging to caller_id.

list_keys() -> list[dict]¶

Return metadata for all active (non-revoked) keys. No plaintext or hash.

RateLimiterBackend(Protocol)¶

Backend contract for sliding-window rate limiters.

Methods:

check(key: str, limit: int, window_seconds: float) -> tuple[bool, int]¶

Attempt to record one hit for key.

MemoryBackend¶

In-process sliding-window backend.

Methods:

check(key: str, limit: int, window_seconds: float) -> tuple[bool, int]¶

RedisBackend¶

Redis-backed sliding-window backend.

Constructor:

Methods:

check(key: str, limit: int, window_seconds: float) -> tuple[bool, int]¶

SlidingWindowRateLimiter¶

Sliding window rate limiter facade over a pluggable backend.

Constructor:

Methods:

check(caller_id: str, tier: Tier) -> Result[bool]¶

Return Result.ok(True) if allowed, Result.fail(...) if over limit.

Role(Enum)¶

Caller privilege level.

Tier(Enum)¶

Subscription tier with associated rate limits.

Methods:

rate_limit() -> int¶

Requests per minute.

max_pipeline_steps() -> int | None¶

Max pipeline steps (None = unlimited).

CallerIdentity¶

Authenticated caller metadata.

SecurityRequest¶

Inbound request to be checked by the security pipeline.

SecurityVerdict¶

Output of a security pipeline stage.

Functions¶

make_session_backend(backend: str | CBRNSessionBackend = 'memory', redis_url: str = '', key_prefix: str = 'g6:cbrn:session:') -> CBRNSessionBackend¶

Factory for CBRN rolling-session accumulation backends.

make_backend(backend: str = 'memory', redis_url: str = '', key_prefix: str = 'g6:ratelimit:', use_lua: bool = True) -> RateLimiterBackend¶

Factory. backend in {"memory", "redis"}.