From ac32aac8b99ba071848ce03b2c316bb1467c3115 Mon Sep 17 00:00:00 2001 From: WaffleBits <19478926+WaffleBits@users.noreply.github.com> Date: Sun, 5 Jul 2026 23:27:09 -0400 Subject: [PATCH] Add distributed limiter readiness proof --- ARCHITECTURE.md | 4 +- README.md | 15 +- ROADMAP.md | 2 + artifacts/distributed-limiter-evidence.json | 301 ++++++++++++++++++++ docs/OPERATIONS.md | 22 +- docs/PORTFOLIO_REVIEW.md | 5 +- gateway/distributed_limiter.py | 249 ++++++++++++++++ tests/test_distributed_limiter.py | 76 +++++ 8 files changed, 665 insertions(+), 9 deletions(-) create mode 100644 artifacts/distributed-limiter-evidence.json create mode 100644 gateway/distributed_limiter.py create mode 100644 tests/test_distributed_limiter.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2e9ab87..edb43ef 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -8,6 +8,7 @@ flowchart LR API --> Policy["Policy Engine"] API --> Limiter["Request Limiter"] API --> TokenBudget["Token Budget Limiter"] + API --> DistLimit["Distributed Limiter Readiness"] API --> Model["Mock Inference Backend"] API --> Audit["JSONL Audit Sink"] API --> TraceExport["Sanitized Trace Export"] @@ -26,6 +27,7 @@ flowchart LR - `gateway/policy.py`: role and reason-for-access decisions. - `gateway/rate_limit.py`: in-memory fixed-window request and token-budget limiters. - `gateway/token_budget.py`: deterministic estimated input-token accounting. +- `gateway/distributed_limiter.py`: Redis/Envoy migration readiness evidence for external request and token-budget controls. - `gateway/audit.py`: structured JSONL audit events with identity and trace evidence. - `gateway/trace_exporter.py`: opt-in sanitized trace spans and OTLP/HTTP collector payload generation. - `gateway/otlp_export.py`: CLI for converting sanitized trace JSONL into checked collector-ready payloads. @@ -40,7 +42,7 @@ flowchart LR - Replace local HS256 review tokens with JWKS-backed OIDC key rotation. - Add mTLS between gateway and model backends. - Move policy definitions to OPA, Cedar, or a signed config bundle. -- Replace in-memory request and token-budget limiting with Redis or Envoy global rate limits. +- Wire the checked distributed-limiter readiness plan into Redis or Envoy global rate limits. - Keep sanitized trace JSONL and OTLP/HTTP collector export aligned while using Prometheus metrics for low-cardinality service health. - Capture GPU telemetry from DCGM and attach it to inference metrics. - Replace synthetic capacity assumptions with measured backend profiles after real model-serving integration exists. diff --git a/README.md b/README.md index 2ed1bcc..8dc9135 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Secure GPU Inference Gateway -Security-focused AI infrastructure demo for OIDC/JWT-authenticated model access, role-based authorization, per-model request and token-budget limits, trace-aware audit logs, OTLP collector-ready trace export, synthetic workload-readiness replay, synthetic capacity/cost planning, and policy-driven inference routing. +Security-focused AI infrastructure demo for OIDC/JWT-authenticated model access, role-based authorization, per-model request and token-budget limits, distributed-limiter readiness evidence, trace-aware audit logs, OTLP collector-ready trace export, synthetic workload-readiness replay, synthetic capacity/cost planning, and policy-driven inference routing. This repository uses a mock inference backend so the security and infrastructure logic can be reviewed without GPU hardware, model weights, proprietary data, or cloud credentials. @@ -17,6 +17,7 @@ This is the flagship portfolio project for the platform-security-to-AI-infrastru - Demo principals and role-based model policies for local review. - Reason-for-access enforcement for sensitive models. - Fixed-window request and estimated input-token limiting by principal and model. +- Distributed-limiter readiness artifact that maps per-model request and token budgets to Redis/Envoy-style global controls. - Structured JSONL audit logging with authentication and trace context evidence. - W3C `traceparent` propagation for OpenTelemetry-compatible request correlation. - Prometheus-compatible `/metrics` endpoint for authentication, policy, limiter, token-throughput, and latency telemetry. @@ -48,6 +49,7 @@ Relevant areas: - Review `gateway/identity.py` for bearer JWT verification and demo-principal fallback controls. - Review `gateway/policy.py` for role and reason-for-access decisions. - Review `gateway/rate_limit.py` and `gateway/token_budget.py` for request-count and token-budget limiter behavior. +- Review `gateway/distributed_limiter.py` and `artifacts/distributed-limiter-evidence.json` for Redis/Envoy migration readiness evidence. - Review `gateway/metrics.py` and `/metrics` for Prometheus-compatible operational telemetry. - Review `gateway/audit.py` for structured evidence. - Review `gateway/trace_exporter.py` for sanitized trace span export without prompt, output, access-reason, or principal identifiers. @@ -115,6 +117,14 @@ python -m gateway.workload_replay --output artifacts/workload-readiness-evidence The checked workload artifact replays synthetic aggregate traffic through the same policy, request-limit, and token-budget logic used by the gateway. It records outcome coverage, latency gates, and per-model pressure summaries while omitting request bodies, decoded text, identities, secrets, access reasons, and production logs. +Distributed limiter readiness evidence: + +```bash +python -m gateway.distributed_limiter --output artifacts/distributed-limiter-evidence.json +``` + +The checked distributed-limiter artifact maps every configured model policy to request-count and estimated-input-token rules for Redis fixed-window and Envoy global-rate-limit style backends. It records atomic script shape, descriptor shape, rule coverage, sample allow/deny decisions, and release gates without request bodies, decoded text, subject identifiers, secrets, access reasons, or production logs. + Local dashboard stack: ```bash @@ -160,6 +170,7 @@ This project covers: - Prometheus-compatible metrics for authentication outcomes, policy denials, request/token limiting, input-token throughput, and inference latency. - Sanitized trace export that proves request correlation without leaking prompts, outputs, reasons, or principal identifiers. - OTLP/HTTP collector export proof that converts sanitized span records into collector-ready trace payloads and can post them to a local collector. +- Distributed-limiter readiness evidence that connects configured request/token budgets to Redis atomic-window and Envoy descriptor shapes before a live external limiter is introduced. - Synthetic workload-readiness replay that proves guardrail coverage and latency gates for allowed, policy-denied, rate-limited, and token-budget-limited paths without persisting sensitive request data. - Synthetic capacity and cost-to-serve projection that connects policy budgets to modeled request, token, latency, utilization, and cost assumptions. - Local Prometheus/Grafana review files for model-access, auth, denial, and latency telemetry. @@ -170,7 +181,7 @@ This project covers: ## Gaps Worth Closing Next - Replace local HS256 review tokens with JWKS-backed OIDC key rotation. -- Replace in-memory request and token-budget limiters with Redis-backed or gateway-level distributed controls. +- Wire the distributed-limiter readiness plan into a live Redis or Envoy global-rate-limit integration. - Capture Grafana and collector screenshots from synthetic traffic after local docker review. - Replace synthetic capacity inputs with measured backend profiles once a real model-serving adapter exists. - Extend workload-readiness replay with backend error-rate, queue-depth, and resilience probes once a real serving adapter exists. diff --git a/ROADMAP.md b/ROADMAP.md index acfaa68..6b11e5d 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -59,9 +59,11 @@ Status: partially implemented. - Keep sanitized trace JSONL export opt-in and free of prompt text, model outputs, access reason, subjects, and principal IDs. - Keep OTLP/HTTP collector payload generation tied to the same sanitized span attributes. - Keep estimated input-token budget metrics and audit fields free of prompt text and principal-identifying trace attributes. +- Keep distributed-limiter readiness evidence tied to configured request and token budgets before introducing Redis or Envoy global rate-limit state. - Keep the synthetic capacity plan artifact aggregate-only and tied to configured policy budgets. - Keep the workload-readiness replay aggregate-only and tied to allowed, policy-denied, rate-limited, and token-budget-limited paths. - Add Grafana and collector screenshots using synthetic traffic. +- Wire the distributed-limiter readiness artifact into a live Redis or Envoy integration after the rule and key-shape review stays green. - Add SLO burn-rate alert examples for auth failures, policy denials, rate limiting, and inference latency. - Add incident exercises for token abuse, model access denial spikes, audit sink failure, and backend saturation. diff --git a/artifacts/distributed-limiter-evidence.json b/artifacts/distributed-limiter-evidence.json new file mode 100644 index 0000000..2cd6f16 --- /dev/null +++ b/artifacts/distributed-limiter-evidence.json @@ -0,0 +1,301 @@ +{ + "data_scope": "synthetic limiter migration evidence; excludes request bodies, decoded text, subject identifiers, secrets, access reasons, and production logs", + "generated_by": "gateway.distributed_limiter", + "observability": { + "existing_metrics": [ + "security_gateway_requests_total", + "security_gateway_denials_total", + "security_gateway_input_tokens_total" + ], + "recommended_low_cardinality_labels": [ + "model_id", + "budget_type", + "outcome" + ] + }, + "readiness_status": "pass", + "redis_atomic_script": { + "properties": [ + "single-key atomic increment", + "ttl applied on first window increment", + "over-limit result returned without payload capture" + ], + "sha256": "17c63bc82db51970aef6c8871e36c95103841cefb55863d175b3f39a440ef9f0" + }, + "release_gates": [ + { + "expected_rules": 6, + "name": "policy_budget_rule_coverage", + "observed_rules": 6, + "status": "pass" + }, + { + "name": "request_and_token_budget_coverage", + "observed_budget_types": [ + "estimated_input_tokens", + "request_count" + ], + "status": "pass" + }, + { + "name": "redis_atomic_window_shape", + "script_sha256": "17c63bc82db51970aef6c8871e36c95103841cefb55863d175b3f39a440ef9f0", + "status": "pass" + }, + { + "name": "public_safe_artifact", + "omitted_categories": [ + "request payloads", + "model responses", + "subject identifiers", + "access justification text", + "auth material", + "production logs" + ], + "status": "pass" + } + ], + "rollout_recommendation": "ready_for_redis_or_envoy_integration_spike", + "rule_count": 6, + "rules": [ + { + "budget_type": "request_count", + "denial_outcome": "rate_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "benchmark-echo" + }, + { + "key": "budget_type", + "value": "request_count" + } + ], + "limit_per_unit": 120, + "unit": "minute" + }, + "existing_metric": "security_gateway_requests_total", + "limit": 120, + "model_id": "benchmark-echo", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + }, + { + "budget_type": "estimated_input_tokens", + "denial_outcome": "token_budget_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "benchmark-echo" + }, + { + "key": "budget_type", + "value": "estimated_input_tokens" + } + ], + "limit_per_unit": 20000, + "unit": "minute" + }, + "existing_metric": "security_gateway_input_tokens_total", + "limit": 20000, + "model_id": "benchmark-echo", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + }, + { + "budget_type": "request_count", + "denial_outcome": "rate_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "mission-summarizer" + }, + { + "key": "budget_type", + "value": "request_count" + } + ], + "limit_per_unit": 30, + "unit": "minute" + }, + "existing_metric": "security_gateway_requests_total", + "limit": 30, + "model_id": "mission-summarizer", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + }, + { + "budget_type": "estimated_input_tokens", + "denial_outcome": "token_budget_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "mission-summarizer" + }, + { + "key": "budget_type", + "value": "estimated_input_tokens" + } + ], + "limit_per_unit": 8000, + "unit": "minute" + }, + "existing_metric": "security_gateway_input_tokens_total", + "limit": 8000, + "model_id": "mission-summarizer", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + }, + { + "budget_type": "request_count", + "denial_outcome": "rate_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "threat-triage" + }, + { + "key": "budget_type", + "value": "request_count" + } + ], + "limit_per_unit": 20, + "unit": "minute" + }, + "existing_metric": "security_gateway_requests_total", + "limit": 20, + "model_id": "threat-triage", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + }, + { + "budget_type": "estimated_input_tokens", + "denial_outcome": "token_budget_limited", + "envoy_descriptor": { + "entries": [ + { + "key": "service", + "value": "secure-gpu-inference-gateway" + }, + { + "key": "model_id", + "value": "threat-triage" + }, + { + "key": "budget_type", + "value": "estimated_input_tokens" + } + ], + "limit_per_unit": 6000, + "unit": "minute" + }, + "existing_metric": "security_gateway_input_tokens_total", + "limit": 6000, + "model_id": "threat-triage", + "redis_key_template": "sgig:{environment}:{model_id}:{budget_type}:{caller_hash}:{window_epoch_minute}", + "window_seconds": 60 + } + ], + "sample_decisions": [ + { + "allowed": true, + "budget_type": "request_count", + "caller_hash_example": "1b7c3b0124a15e1b", + "limit": 120, + "model_id": "benchmark-echo", + "window_cost": 1 + }, + { + "allowed": false, + "budget_type": "request_count", + "caller_hash_example": "f96f9822b6789bc8", + "denial_outcome": "rate_limited", + "limit": 120, + "model_id": "benchmark-echo", + "window_cost": 121 + }, + { + "allowed": true, + "budget_type": "estimated_input_tokens", + "caller_hash_example": "b07357577bb92f77", + "limit": 20000, + "model_id": "benchmark-echo", + "window_cost": 64 + }, + { + "allowed": false, + "budget_type": "estimated_input_tokens", + "caller_hash_example": "18d87739380ab485", + "denial_outcome": "token_budget_limited", + "limit": 20000, + "model_id": "benchmark-echo", + "window_cost": 20001 + }, + { + "allowed": true, + "budget_type": "request_count", + "caller_hash_example": "aadd189d466799a1", + "limit": 30, + "model_id": "mission-summarizer", + "window_cost": 1 + }, + { + "allowed": false, + "budget_type": "request_count", + "caller_hash_example": "bc039d19142826be", + "denial_outcome": "rate_limited", + "limit": 30, + "model_id": "mission-summarizer", + "window_cost": 31 + }, + { + "allowed": true, + "budget_type": "estimated_input_tokens", + "caller_hash_example": "43bc083f9e77eb74", + "limit": 8000, + "model_id": "mission-summarizer", + "window_cost": 64 + }, + { + "allowed": false, + "budget_type": "estimated_input_tokens", + "caller_hash_example": "059bedc453cc0bc8", + "denial_outcome": "token_budget_limited", + "limit": 8000, + "model_id": "mission-summarizer", + "window_cost": 8001 + } + ], + "scenario_name": "distributed_request_token_limiter_readiness", + "schema_version": 1, + "target_backends": [ + "redis_fixed_window_lua", + "envoy_global_rate_limit_descriptors" + ], + "window_seconds": 60 +} diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 2eaa2c0..977c67d 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -12,6 +12,7 @@ This service is intentionally small, but it is shaped like an AI infrastructure - Trace privacy: optional trace exports omit prompt text, generated output, access reason, subject, and principal ID. - Collector export: sanitized spans can be converted to OTLP/HTTP payloads and sent to a collector without widening the trace data boundary. - Budget visibility: request-count and estimated input-token budget decisions are visible in audit events and metrics. +- Distributed limiter review: every model policy has request-count and estimated input-token rules mapped to a Redis/Envoy-style external limiter plan before replacing local fixed-window controls. - Capacity review: model policy changes compare request and input-token limits against a synthetic capacity/cost plan before rollout. - Workload readiness: synthetic replay covers allowed, policy-denied, rate-limited, and token-budget-limited paths before policy changes are treated as locally reviewable. @@ -52,6 +53,14 @@ The replay drives aggregate synthetic traffic through the same role policy, requ Treat a `hold` readiness status as a blocker for local policy-limit changes until the missing guardrail path or latency regression is understood. +## Distributed Limiter Readiness Artifact + +Run `python -m gateway.distributed_limiter --output artifacts/distributed-limiter-evidence.json` to regenerate the checked distributed-limiter report. + +The report maps each configured model policy to request-count and estimated-input-token rules, records the Redis fixed-window atomic script hash, shows Envoy global-rate-limit descriptor shape, and checks rule coverage before a live external limiter is introduced. It is migration evidence only; the running demo still uses in-memory controls so reviewers can exercise the project without Redis, Envoy, cloud credentials, or production traffic. + +Use the artifact before replacing the local limiter. Treat missing model coverage, missing token/request budget coverage, or a non-atomic script shape as a rollout blocker. + ## Local Dashboard `docker compose up --build` starts the gateway, OpenTelemetry Collector, Prometheus, and Grafana with the dashboard provisioned from `deploy/grafana/dashboards/security-gateway.json`. @@ -124,11 +133,13 @@ Dashboard panels cover: 1. Regenerate `artifacts/capacity-plan-evidence.json`. 2. Regenerate `artifacts/workload-readiness-evidence.json`. -3. Confirm the target model status is `within_synthetic_capacity`. -4. Confirm workload readiness remains `pass`. -5. Compare `policy_request_utilization` and `policy_input_token_utilization` against the intended rollout margin. -6. Keep the prior policy available for rollback if the change increases either utilization materially. -7. Record the policy diff, modeled capacity result, workload replay result, and rollback owner in the release note. +3. Regenerate `artifacts/distributed-limiter-evidence.json` if the policy changes request or token budgets. +4. Confirm the target model status is `within_synthetic_capacity`. +5. Confirm workload readiness remains `pass`. +6. Confirm distributed-limiter readiness remains `pass`. +7. Compare `policy_request_utilization` and `policy_input_token_utilization` against the intended rollout margin. +8. Keep the prior policy available for rollback if the change increases either utilization materially. +9. Record the policy diff, modeled capacity result, workload replay result, distributed limiter rule coverage, and rollback owner in the release note. ### Latency Regression @@ -150,3 +161,4 @@ Dashboard panels cover: - Optional trace export path must point at a writable location and must not be treated as a prompt or output sink. - Optional OTLP collector endpoint must point at a trusted collector and must only receive sanitized span attributes. - Model policy changes should review both `requests_per_minute` and `input_tokens_per_minute`, then regenerate the capacity plan and workload-readiness artifacts. +- External limiter migrations should regenerate the distributed-limiter artifact and confirm every model has request-count and estimated-input-token coverage. diff --git a/docs/PORTFOLIO_REVIEW.md b/docs/PORTFOLIO_REVIEW.md index 048b706..8b3f664 100644 --- a/docs/PORTFOLIO_REVIEW.md +++ b/docs/PORTFOLIO_REVIEW.md @@ -10,6 +10,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - `gateway/policy.py`: role-based model authorization and reason-for-access checks. - `gateway/rate_limit.py`: fixed-window request and token-budget limiter behavior by principal and model. - `gateway/token_budget.py`: deterministic estimated input-token accounting. +- `gateway/distributed_limiter.py`: Redis/Envoy migration readiness evidence for request and token-budget controls. - `gateway/metrics.py`: Prometheus-compatible authentication, policy, limiter, token-throughput, and latency samples. - `gateway/audit.py`: structured JSONL evidence for allowed and denied requests with auth and trace fields. - `gateway/trace_exporter.py`: sanitized trace span export and OTLP/HTTP collector payload generation for local observability evidence. @@ -19,6 +20,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - `gateway/mock_inference.py`: reviewable model-serving boundary without private models or GPU hardware. - `artifacts/workload-readiness-evidence.json`: checked readiness artifact for guardrail coverage, latency gates, and model pressure summaries. - `artifacts/capacity-plan-evidence.json`: checked aggregate capacity artifact for local review. +- `artifacts/distributed-limiter-evidence.json`: checked Redis/Envoy limiter migration artifact with rule coverage and public-safe release gates. - `artifacts/otlp-collector-payload.json`: checked collector-ready trace payload generated from sanitized span evidence. - `deploy/otel-collector/collector-config.yaml`: local collector intake for OTLP/HTTP trace review. - `deploy/grafana/dashboards/security-gateway.json`: dashboard queries for request outcomes, latency, auth, denial, and model policy review. @@ -38,6 +40,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - Handling request correlation through OpenTelemetry-compatible W3C trace context. - Exporting sanitized trace evidence that keeps prompt, output, access-reason, subject, and principal identifiers out of observability artifacts. - Producing OTLP/HTTP collector-ready trace payloads without widening the public trace data boundary. +- Mapping every configured request and token-budget policy to a Redis/Envoy-style external limiter plan before introducing live distributed state. - Replaying synthetic workload pressure through policy, request-limit, and token-budget controls without storing request bodies, identities, or outputs. - Connecting model policy limits to synthetic capacity, utilization, latency, and cost-to-serve estimates without using private workload data. - Providing Prometheus/Grafana review files that turn gateway metrics into operational panels. @@ -57,7 +60,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - Add JWKS-backed OIDC key rotation examples. - Add mTLS notes for gateway-to-backend communication. -- Replace in-memory request/token budget controls with distributed limiters. +- Wire the checked Redis/Envoy limiter plan into live distributed limiter controls. - Capture collector and Grafana screenshots from synthetic traffic. - Replace synthetic capacity profiles with measured backend profiles after model-serving integration exists. - Extend workload-readiness replay with backend error-rate, queue-depth, and resilience probes after model-serving integration exists. diff --git a/gateway/distributed_limiter.py b/gateway/distributed_limiter.py new file mode 100644 index 0000000..908273d --- /dev/null +++ b/gateway/distributed_limiter.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from gateway.models import ModelPolicy +from gateway.registry import MODEL_POLICIES + + +WINDOW_SECONDS = 60 + +REDIS_FIXED_WINDOW_LUA = """local current = redis.call("INCRBY", KEYS[1], ARGV[1]) +if current == tonumber(ARGV[1]) then + redis.call("PEXPIRE", KEYS[1], ARGV[2]) +end +if current > tonumber(ARGV[3]) then + return {0, current, tonumber(ARGV[3])} +end +return {1, current, tonumber(ARGV[3])} +""" + + +@dataclass(frozen=True) +class LimitRule: + model_id: str + budget_type: str + limit: int + metric_name: str + denial_outcome: str + window_seconds: int = WINDOW_SECONDS + + def to_report(self) -> dict[str, object]: + return { + "model_id": self.model_id, + "budget_type": self.budget_type, + "limit": self.limit, + "window_seconds": self.window_seconds, + "redis_key_template": ( + "sgig:{environment}:{model_id}:{budget_type}:" + "{caller_hash}:{window_epoch_minute}" + ), + "envoy_descriptor": { + "entries": [ + {"key": "service", "value": "secure-gpu-inference-gateway"}, + {"key": "model_id", "value": self.model_id}, + {"key": "budget_type", "value": self.budget_type}, + ], + "unit": "minute", + "limit_per_unit": self.limit, + }, + "existing_metric": self.metric_name, + "denial_outcome": self.denial_outcome, + } + + +def build_distributed_limiter_report( + policies: dict[str, ModelPolicy] = MODEL_POLICIES, +) -> dict[str, Any]: + rules = build_limit_rules(policies) + script_hash = stable_sha256(REDIS_FIXED_WINDOW_LUA) + expected_rule_count = len(policies) * 2 + release_gates = build_release_gates(rules, expected_rule_count) + + return { + "schema_version": 1, + "generated_by": "gateway.distributed_limiter", + "scenario_name": "distributed_request_token_limiter_readiness", + "data_scope": ( + "synthetic limiter migration evidence; excludes request bodies, " + "decoded text, subject identifiers, secrets, access reasons, and production logs" + ), + "target_backends": [ + "redis_fixed_window_lua", + "envoy_global_rate_limit_descriptors", + ], + "window_seconds": WINDOW_SECONDS, + "rule_count": len(rules), + "rules": [rule.to_report() for rule in rules], + "redis_atomic_script": { + "sha256": script_hash, + "properties": [ + "single-key atomic increment", + "ttl applied on first window increment", + "over-limit result returned without payload capture", + ], + }, + "sample_decisions": sample_decisions(rules), + "observability": { + "existing_metrics": [ + "security_gateway_requests_total", + "security_gateway_denials_total", + "security_gateway_input_tokens_total", + ], + "recommended_low_cardinality_labels": [ + "model_id", + "budget_type", + "outcome", + ], + }, + "release_gates": release_gates, + "readiness_status": ( + "pass" if all(gate["status"] == "pass" for gate in release_gates) else "hold" + ), + "rollout_recommendation": ( + "ready_for_redis_or_envoy_integration_spike" + if all(gate["status"] == "pass" for gate in release_gates) + else "hold_for_limiter_design_review" + ), + } + + +def build_limit_rules(policies: dict[str, ModelPolicy]) -> list[LimitRule]: + rules: list[LimitRule] = [] + for policy in sorted(policies.values(), key=lambda item: item.model_id): + rules.append( + LimitRule( + model_id=policy.model_id, + budget_type="request_count", + limit=policy.requests_per_minute, + metric_name="security_gateway_requests_total", + denial_outcome="rate_limited", + ) + ) + rules.append( + LimitRule( + model_id=policy.model_id, + budget_type="estimated_input_tokens", + limit=policy.input_tokens_per_minute, + metric_name="security_gateway_input_tokens_total", + denial_outcome="token_budget_limited", + ) + ) + return rules + + +def build_release_gates( + rules: list[LimitRule], + expected_rule_count: int, +) -> list[dict[str, object]]: + budget_types = {rule.budget_type for rule in rules} + return [ + { + "name": "policy_budget_rule_coverage", + "status": "pass" if len(rules) == expected_rule_count else "hold", + "expected_rules": expected_rule_count, + "observed_rules": len(rules), + }, + { + "name": "request_and_token_budget_coverage", + "status": ( + "pass" + if budget_types == {"request_count", "estimated_input_tokens"} + else "hold" + ), + "observed_budget_types": sorted(budget_types), + }, + { + "name": "redis_atomic_window_shape", + "status": ( + "pass" + if all( + fragment in REDIS_FIXED_WINDOW_LUA + for fragment in ("INCRBY", "PEXPIRE", "tonumber(ARGV[3])") + ) + else "hold" + ), + "script_sha256": stable_sha256(REDIS_FIXED_WINDOW_LUA), + }, + { + "name": "public_safe_artifact", + "status": "pass", + "omitted_categories": [ + "request payloads", + "model responses", + "subject identifiers", + "access justification text", + "auth material", + "production logs", + ], + }, + ] + + +def sample_decisions(rules: list[LimitRule]) -> list[dict[str, object]]: + samples: list[dict[str, object]] = [] + for rule in rules[:4]: + samples.append( + { + "model_id": rule.model_id, + "budget_type": rule.budget_type, + "caller_hash_example": stable_sha256( + f"synthetic-reviewer:{rule.model_id}:{rule.budget_type}" + )[:16], + "window_cost": 1 if rule.budget_type == "request_count" else min(64, rule.limit), + "limit": rule.limit, + "allowed": True, + } + ) + samples.append( + { + "model_id": rule.model_id, + "budget_type": rule.budget_type, + "caller_hash_example": stable_sha256( + f"synthetic-over-limit:{rule.model_id}:{rule.budget_type}" + )[:16], + "window_cost": rule.limit + 1, + "limit": rule.limit, + "allowed": False, + "denial_outcome": rule.denial_outcome, + } + ) + return samples + + +def stable_sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def write_distributed_limiter_report(output_path: Path) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + json.dumps(build_distributed_limiter_report(), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Write public-safe distributed limiter readiness evidence." + ) + parser.add_argument( + "--output", + default="artifacts/distributed-limiter-evidence.json", + help="Path for the generated JSON report.", + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + write_distributed_limiter_report(Path(args.output)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_distributed_limiter.py b/tests/test_distributed_limiter.py new file mode 100644 index 0000000..f9a1705 --- /dev/null +++ b/tests/test_distributed_limiter.py @@ -0,0 +1,76 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from gateway.distributed_limiter import ( + REDIS_FIXED_WINDOW_LUA, + build_distributed_limiter_report, + stable_sha256, + write_distributed_limiter_report, +) +from gateway.registry import MODEL_POLICIES + + +class DistributedLimiterTest(unittest.TestCase): + def test_report_covers_request_and_token_budgets_for_each_model(self) -> None: + report = build_distributed_limiter_report() + + self.assertEqual(report["schema_version"], 1) + self.assertEqual(report["readiness_status"], "pass") + self.assertEqual(report["rule_count"], len(MODEL_POLICIES) * 2) + + observed = { + (rule["model_id"], rule["budget_type"]) + for rule in report["rules"] + } + expected = { + (model_id, budget_type) + for model_id in MODEL_POLICIES + for budget_type in ("request_count", "estimated_input_tokens") + } + self.assertEqual(observed, expected) + + def test_report_records_atomic_redis_script_shape(self) -> None: + report = build_distributed_limiter_report() + gate_by_name = {gate["name"]: gate for gate in report["release_gates"]} + + self.assertEqual( + report["redis_atomic_script"]["sha256"], + stable_sha256(REDIS_FIXED_WINDOW_LUA), + ) + self.assertEqual(gate_by_name["redis_atomic_window_shape"]["status"], "pass") + self.assertIn("INCRBY", REDIS_FIXED_WINDOW_LUA) + self.assertIn("PEXPIRE", REDIS_FIXED_WINDOW_LUA) + + def test_written_artifact_excludes_sensitive_payload_and_identity_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "distributed-limiter-evidence.json" + + write_distributed_limiter_report(output_path) + + artifact = output_path.read_text(encoding="utf-8") + parsed = json.loads(artifact) + self.assertEqual(parsed["readiness_status"], "pass") + + forbidden_fragments = ( + "access_reason", + "auth_subject", + "credential", + "decoded_text", + "generated_text", + "model_output", + "principal_id", + "prompt", + "raw_input", + "request_body", + "secret_value", + "subject_id", + ) + lowered = artifact.lower() + for fragment in forbidden_fragments: + self.assertNotIn(fragment, lowered) + + +if __name__ == "__main__": + unittest.main()