From 1fbf5c532a5802e87415c2aa35d6976e7fef7e70 Mon Sep 17 00:00:00 2001 From: WaffleBits <19478926+WaffleBits@users.noreply.github.com> Date: Sat, 11 Jul 2026 09:09:10 -0400 Subject: [PATCH] Add OpenAI-compatible inference backend adapter --- README.md | 24 ++++- ROADMAP.md | 7 +- deploy/kubernetes/gateway.yaml | 1 + docker-compose.yml | 3 + docs/PORTFOLIO_REVIEW.md | 6 +- gateway/app.py | 41 +++++++- gateway/backend_adapter.py | 140 +++++++++++++++++++++++++++ tests/test_backend_adapter.py | 168 +++++++++++++++++++++++++++++++++ 8 files changed, 381 insertions(+), 9 deletions(-) create mode 100644 gateway/backend_adapter.py create mode 100644 tests/test_backend_adapter.py diff --git a/README.md b/README.md index 162f9ac..7e19313 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ This is the flagship portfolio project for the platform-security-to-AI-infrastru - Synthetic capacity and cost-to-serve planning artifact tied to configured model policies. - Prometheus, OpenTelemetry Collector, and Grafana provisioning files for local observability review. - Mock GPU inference backend with latency metadata. +- Optional OpenAI-compatible backend adapter for vLLM/SGLang-style `/v1/completions` endpoints, with bounded timeouts and validated response shapes. - Focused unit tests for policy and limiter behavior. - Threat model and architecture notes. - Kubernetes deployment example with health probes and scrape annotations. @@ -145,6 +146,24 @@ python -m gateway.resilience_drill --output artifacts/resilience-drill-evidence. The checked resilience artifact composes workload and deployment evidence with synthetic backend degradation probes. It records latency-spike, error-burst, queue-saturation, and audit-backpressure checks, detection signals, mitigation paths, rollback actions, and recovery gates without request bodies, decoded text, identities, secrets, access reasons, or production logs. +## Optional Model-Serving Adapter + +The mock backend is the default. To route an allowed request to a vLLM- or +SGLang-style OpenAI-compatible completion endpoint, set +`INFERENCE_BACKEND_COMPLETIONS_URL` to either the backend base URL, its `/v1` +URL, or the full `/v1/completions` URL. The adapter sends a bounded non-streaming +completion request, validates the returned text shape, applies a five-second +default timeout, and returns a generic `502` with the trace ID when the backend +is unavailable. Set `INFERENCE_BACKEND_API_KEY` through a secret when the +backend requires authentication; the key, prompt, output, and endpoint errors +are not written to audit or trace records. + +```powershell +$env:INFERENCE_BACKEND_COMPLETIONS_URL = "http://localhost:8001/v1" +$env:INFERENCE_BACKEND_TIMEOUT_MS = "5000" +python -m uvicorn gateway.app:app --port 8000 +``` + Local dashboard stack: ```bash @@ -193,6 +212,7 @@ This project covers: - 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. - Deployment-readiness evidence that connects capacity planning, workload guardrail coverage, distributed limiter coverage, staged rollout phases, and rollback triggers before policy or serving-path changes are treated as locally reviewable. - Resilience-drill evidence that connects workload and deployment readiness to synthetic latency spike, backend error burst, queue saturation, audit backpressure, mitigation, and rollback checks. +- An optional vLLM/SGLang-style backend adapter with schema validation, timeout handling, generic backend-error responses, and mock-by-default operation. - 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. @@ -205,8 +225,8 @@ This project covers: - Replace local HS256 review tokens with JWKS-backed OIDC key rotation. - 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. -- Replace synthetic resilience probes with measured backend error-rate, queue-depth, and recovery evidence once a real serving adapter exists. +- Replace synthetic capacity inputs with measured backend profiles after exercising a real model-serving endpoint. +- Replace synthetic resilience probes with measured backend error-rate, queue-depth, and recovery evidence after exercising a real serving endpoint. - Add policy-as-code examples, redaction controls, and negative authorization tests. - Add CI supply-chain evidence such as SBOM generation, dependency scanning, and container scanning. - Add SOC2/FedRAMP-inspired control mapping notes without claiming certification or production authorization. diff --git a/ROADMAP.md b/ROADMAP.md index b263ae6..4236ed4 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -72,13 +72,14 @@ Status: partially implemented. ## Milestone 6: Model-Serving Integration -Status: planned. +Status: in progress. - Add adapters for Triton-compatible, vLLM-compatible, or SGLang-compatible backends. +- The first adapter is an optional, mock-by-default OpenAI-compatible completion path with timeout, schema, and generic backend-error handling. - Keep mock backend tests so the repo remains reviewable without GPU hardware. -- Replace synthetic capacity profiles with measured backend profiles once a real adapter exists. +- Replace synthetic capacity profiles with measured backend profiles after exercising a real serving endpoint. - Add GPU/DCGM telemetry correlation when real backend integration exists. -- Replace the synthetic resilience drill with real queue depth, backend error rate, recovery time, and per-model latency evidence when backend integration exists. +- Replace the synthetic resilience drill with real queue depth, backend error rate, recovery time, and per-model latency evidence after exercising a real serving endpoint. ## Milestone 7: Control Mapping diff --git a/deploy/kubernetes/gateway.yaml b/deploy/kubernetes/gateway.yaml index 38e5665..28678eb 100644 --- a/deploy/kubernetes/gateway.yaml +++ b/deploy/kubernetes/gateway.yaml @@ -13,6 +13,7 @@ data: OIDC_ISSUER: https://issuer.example.com OIDC_AUDIENCE: secure-gpu-inference-gateway ALLOW_DEMO_PRINCIPALS: "false" + INFERENCE_BACKEND_TIMEOUT_MS: "5000" PYTHONDONTWRITEBYTECODE: "1" --- apiVersion: apps/v1 diff --git a/docker-compose.yml b/docker-compose.yml index f9a1943..1eae8e6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,6 +8,9 @@ services: TRACE_EXPORT_PATH: /var/log/gateway/traces.jsonl OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: http://otel-collector:4318/v1/traces ALLOW_DEMO_PRINCIPALS: "true" + # Optional: set these for a local vLLM/SGLang-compatible backend. + # INFERENCE_BACKEND_COMPLETIONS_URL: http://model-backend:8000/v1 + # INFERENCE_BACKEND_API_KEY: ${INFERENCE_BACKEND_API_KEY:-} volumes: - gateway-logs:/var/log/gateway depends_on: diff --git a/docs/PORTFOLIO_REVIEW.md b/docs/PORTFOLIO_REVIEW.md index ccbb217..8b09746 100644 --- a/docs/PORTFOLIO_REVIEW.md +++ b/docs/PORTFOLIO_REVIEW.md @@ -19,6 +19,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - `gateway/capacity_plan.py`: synthetic capacity and cost-to-serve projection tied to model policies. - `gateway/deployment_readiness.py`: synthetic deployment-readiness review that composes capacity, workload, and limiter evidence into shadow, canary, staged rollout, and rollback gates. - `gateway/resilience_drill.py`: synthetic resilience drill for latency spike, backend error burst, queue saturation, audit backpressure, mitigation, and rollback evidence. +- `gateway/backend_adapter.py`: optional OpenAI-compatible completion adapter for vLLM/SGLang-style serving paths, with bounded timeout and response validation. - `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. @@ -51,6 +52,7 @@ This project is intentionally designed as a public-safe AI security infrastructu - Exercising synthetic backend degradation paths with detection signals, mitigation paths, rollback actions, and recovery gates. - Providing Prometheus/Grafana review files that turn gateway metrics into operational panels. - Designing a mockable inference boundary that can later route to real model-serving backends. +- Routing allowed requests through an explicitly configured vLLM/SGLang-style endpoint while preserving mock-by-default reviewability and sanitized audit/trace boundaries. - Showing Kubernetes deployment thinking without requiring cloud credentials. - Keeping the next-build path focused on platform security and AI infrastructure controls instead of generic dashboard work. - Keeping public portfolio code free of secrets, credentials, private data, and production logs. @@ -68,8 +70,8 @@ This project is intentionally designed as a public-safe AI security infrastructu - Add mTLS notes for gateway-to-backend communication. - 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. -- Replace synthetic resilience probes with measured backend error-rate, queue-depth, and recovery evidence after model-serving integration exists. +- Replace synthetic capacity profiles with measured backend profiles after exercising a real model-serving endpoint. +- Replace synthetic resilience probes with measured backend error-rate, queue-depth, and recovery evidence after exercising a real serving endpoint. - Add policy-as-code and redaction examples with positive and negative test cases. - Add CI supply-chain evidence such as SBOM generation, dependency scanning, and container scanning. - Add SOC2/FedRAMP-inspired control mapping notes without implying certification or production authorization. diff --git a/gateway/app.py b/gateway/app.py index 7607776..158db96 100644 --- a/gateway/app.py +++ b/gateway/app.py @@ -6,9 +6,9 @@ from pydantic import BaseModel, Field from gateway.audit import JsonlAuditSink +from gateway.backend_adapter import BackendAdapterError, run_configured_inference from gateway.identity import AuthSettings, ResolvedPrincipal, resolve_principal from gateway.metrics import GatewayMetrics -from gateway.mock_inference import run_mock_inference from gateway.models import AuditEvent from gateway.policy import evaluate_policy from gateway.rate_limit import FixedWindowRateLimiter, FixedWindowTokenBudgetLimiter @@ -221,7 +221,44 @@ def infer( headers={"traceparent": format_traceparent(trace)}, ) - result = run_mock_inference(model_id, request.input) + try: + result = run_configured_inference(model_id, request.input) + except BackendAdapterError: + backend_error_reasons = ("inference backend unavailable",) + metrics.record_request(model_id, "backend_error", backend_error_reasons) + metrics.record_input_tokens(model_id, "backend_error", estimated_input_tokens) + audit_sink.write( + audit_event( + resolved, + trace, + model_id=model_id, + allowed=True, + reason=request.reason, + decision_reasons=backend_error_reasons, + estimated_input_tokens=estimated_input_tokens, + token_budget_limit=token_budget_limit, + ) + ) + export_trace_span( + trace, + model_id=model_id, + outcome="backend_error", + auth_method=resolved.auth_method, + decision_reasons=backend_error_reasons, + latency_ms=None, + estimated_input_tokens=estimated_input_tokens, + token_budget_limit=token_budget_limit, + started_at_unix_nano=started_at_unix_nano, + ) + raise HTTPException( + status_code=502, + detail={ + "reason": "inference backend unavailable", + "trace_id": trace.trace_id, + }, + headers={"traceparent": format_traceparent(trace)}, + ) + latency_ms = float(result["latency_ms"]) metrics.record_request(model_id, "allowed") metrics.record_input_tokens(model_id, "allowed", estimated_input_tokens) diff --git a/gateway/backend_adapter.py b/gateway/backend_adapter.py new file mode 100644 index 0000000..943089c --- /dev/null +++ b/gateway/backend_adapter.py @@ -0,0 +1,140 @@ +"""Optional OpenAI-compatible adapter for vLLM and SGLang-style backends.""" + +from __future__ import annotations + +import json +import os +import time +from collections.abc import Callable +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from gateway.mock_inference import run_mock_inference + +DEFAULT_TIMEOUT_SECONDS = 5.0 +MAX_COMPLETION_TOKENS = 256 + + +class BackendAdapterError(RuntimeError): + """Raised when an external inference backend cannot return a valid result.""" + + +def completion_url(endpoint: str) -> str: + """Normalize a base URL or completion URL without adding duplicate paths.""" + normalized = endpoint.strip().rstrip("/") + if not normalized: + raise BackendAdapterError("backend endpoint must not be empty") + if normalized.endswith("/completions"): + return normalized + if normalized.endswith("/v1"): + return f"{normalized}/completions" + return f"{normalized}/v1/completions" + + +def run_openai_compatible_inference( + model_id: str, + user_input: str, + *, + endpoint: str, + api_key: str | None = None, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + opener: Callable[..., Any] = urlopen, +) -> dict[str, object]: + """Call a vLLM/SGLang-style completion endpoint without logging payloads.""" + if timeout_seconds <= 0: + raise BackendAdapterError("backend timeout must be positive") + + payload = { + "model": model_id, + "prompt": user_input, + "max_tokens": MAX_COMPLETION_TOKENS, + "stream": False, + } + headers = { + "Accept": "application/json", + "Content-Type": "application/json", + } + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + request = Request( + completion_url(endpoint), + data=json.dumps(payload).encode("utf-8"), + headers=headers, + method="POST", + ) + started = time.perf_counter() + try: + with opener(request, timeout=timeout_seconds) as response: + raw_payload = response.read() + except (HTTPError, URLError, TimeoutError, OSError) as error: + raise BackendAdapterError("inference backend request failed") from error + + try: + response_payload = json.loads(raw_payload) + except (TypeError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise BackendAdapterError("inference backend returned invalid JSON") from error + + output, response_model = extract_completion(response_payload) + usage = response_payload.get("usage") + if usage is not None and not isinstance(usage, dict): + raise BackendAdapterError("inference backend returned invalid usage metadata") + + latency_ms = round((time.perf_counter() - started) * 1000, 3) + result: dict[str, object] = { + "model_id": response_model or model_id, + "output": output, + "latency_ms": latency_ms, + "backend": "openai-compatible", + } + if usage: + result["usage"] = { + key: int(value) + for key, value in usage.items() + if key in {"prompt_tokens", "completion_tokens", "total_tokens"} + and isinstance(value, int) + and not isinstance(value, bool) + } + return result + + +def extract_completion(response_payload: Any) -> tuple[str, str | None]: + """Extract text from completion or chat-completion response shapes.""" + if not isinstance(response_payload, dict): + raise BackendAdapterError("inference backend response must be an object") + choices = response_payload.get("choices") + if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict): + raise BackendAdapterError("inference backend response has no choices") + + choice = choices[0] + output = choice.get("text") + if output is None: + message = choice.get("message") + output = message.get("content") if isinstance(message, dict) else None + if not isinstance(output, str): + raise BackendAdapterError("inference backend response has no text content") + + model = response_payload.get("model") + return output, model if isinstance(model, str) and model else None + + +def run_configured_inference(model_id: str, user_input: str) -> dict[str, object]: + """Use the external adapter only when explicitly configured; otherwise use the mock.""" + endpoint = os.getenv("INFERENCE_BACKEND_COMPLETIONS_URL", "").strip() + if not endpoint: + return run_mock_inference(model_id, user_input) + + timeout_ms = os.getenv("INFERENCE_BACKEND_TIMEOUT_MS", "5000") + try: + timeout_seconds = float(timeout_ms) / 1000 + except ValueError as error: + raise BackendAdapterError("backend timeout must be numeric") from error + + return run_openai_compatible_inference( + model_id, + user_input, + endpoint=endpoint, + api_key=os.getenv("INFERENCE_BACKEND_API_KEY"), + timeout_seconds=timeout_seconds, + ) diff --git a/tests/test_backend_adapter.py b/tests/test_backend_adapter.py new file mode 100644 index 0000000..9a209c1 --- /dev/null +++ b/tests/test_backend_adapter.py @@ -0,0 +1,168 @@ +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch +from urllib.error import URLError + +from fastapi import HTTPException, Response + +import gateway.app as app_module +from gateway.audit import JsonlAuditSink +from gateway.metrics import GatewayMetrics +from gateway.backend_adapter import ( + BackendAdapterError, + completion_url, + extract_completion, + run_configured_inference, + run_openai_compatible_inference, +) + + +class FakeResponse: + def __init__(self, payload: dict[str, object]) -> None: + self.payload = json.dumps(payload).encode("utf-8") + + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return self.payload + + +class BackendAdapterTest(unittest.TestCase): + def test_completion_url_accepts_base_or_full_endpoint(self) -> None: + self.assertEqual( + completion_url("http://vllm:8000"), + "http://vllm:8000/v1/completions", + ) + self.assertEqual( + completion_url("http://sglang:30000/v1"), + "http://sglang:30000/v1/completions", + ) + self.assertEqual( + completion_url("http://backend/v1/completions"), + "http://backend/v1/completions", + ) + + def test_posts_prompt_and_extracts_completion_usage(self) -> None: + calls: list[tuple[object, float]] = [] + + def opener(request: object, timeout: float) -> FakeResponse: + calls.append((request, timeout)) + return FakeResponse( + { + "model": "served-model", + "choices": [{"text": "safe response"}], + "usage": {"prompt_tokens": 4, "completion_tokens": 2}, + } + ) + + result = run_openai_compatible_inference( + "mission-summarizer", + "synthetic prompt", + endpoint="http://vllm:8000/v1", + api_key="secret-token", + timeout_seconds=2.5, + opener=opener, + ) + + request = calls[0][0] + self.assertEqual(calls[0][1], 2.5) + self.assertEqual(request.full_url, "http://vllm:8000/v1/completions") + self.assertEqual(request.get_header("Authorization"), "Bearer secret-token") + self.assertEqual(json.loads(request.data), { + "model": "mission-summarizer", + "prompt": "synthetic prompt", + "max_tokens": 256, + "stream": False, + }) + self.assertEqual(result["model_id"], "served-model") + self.assertEqual(result["output"], "safe response") + self.assertEqual(result["usage"], {"prompt_tokens": 4, "completion_tokens": 2}) + + def test_extracts_chat_completion_shape(self) -> None: + self.assertEqual( + extract_completion( + {"choices": [{"message": {"content": "chat response"}}]} + ), + ("chat response", None), + ) + + def test_rejects_invalid_backend_payload(self) -> None: + with self.assertRaises(BackendAdapterError): + extract_completion({"choices": []}) + + def opener(request: object, timeout: float) -> FakeResponse: + return FakeResponse({"choices": [{"finish_reason": "stop"}]}) + + with self.assertRaises(BackendAdapterError): + run_openai_compatible_inference( + "model", + "prompt", + endpoint="http://backend/v1", + opener=opener, + ) + + def test_translates_transport_failure_without_exposing_details(self) -> None: + def opener(request: object, timeout: float) -> FakeResponse: + raise URLError("secret-hostname") + + with self.assertRaisesRegex(BackendAdapterError, "request failed"): + run_openai_compatible_inference( + "model", + "prompt", + endpoint="http://backend/v1", + opener=opener, + ) + + def test_mock_backend_remains_default(self) -> None: + with patch.dict("os.environ", {}, clear=True): + result = run_configured_inference("benchmark-echo", "prompt") + self.assertEqual(result["backend"], "mock-gpu") + + def test_app_returns_generic_502_when_backend_fails(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + old_audit_sink = app_module.audit_sink + old_metrics = app_module.metrics + try: + app_module.audit_sink = JsonlAuditSink(str(Path(tmpdir) / "audit.jsonl")) + app_module.metrics = GatewayMetrics() + with patch.object( + app_module, + "run_configured_inference", + side_effect=BackendAdapterError("secret backend hostname"), + ): + with self.assertRaises(HTTPException) as raised: + app_module.infer( + "benchmark-echo", + app_module.InferenceRequest(input="prompt"), + Response(), + x_principal_id="analyst-1", + authorization=None, + traceparent=None, + ) + + self.assertEqual(raised.exception.status_code, 502) + self.assertEqual( + raised.exception.detail["reason"], + "inference backend unavailable", + ) + self.assertNotIn("secret backend hostname", str(raised.exception.detail)) + rendered_metrics = app_module.metrics.render_prometheus( + app_module.MODEL_POLICIES + ) + self.assertIn( + 'security_gateway_requests_total{model_id="benchmark-echo",outcome="backend_error"} 1', + rendered_metrics, + ) + finally: + app_module.audit_sink = old_audit_sink + app_module.metrics = old_metrics + + +if __name__ == "__main__": + unittest.main()