diff --git a/devplacepy/services/openai_gateway/CLAUDE.md b/devplacepy/services/openai_gateway/CLAUDE.md index 31da160..bafd18a 100644 --- a/devplacepy/services/openai_gateway/CLAUDE.md +++ b/devplacepy/services/openai_gateway/CLAUDE.md @@ -14,6 +14,15 @@ Every NON-STREAMING gateway response (chat, embeddings, images, passthrough; suc **A STREAMING chat response (`stream: true`) is the one exception and carries only `X-Gateway-Model`/`X-Gateway-Backend`/`X-App-Reference`** - HTTP headers are sent before the body, and cost/tokens for a streamed call are only known once the stream ends, so they cannot be headers on that same response. The call is still fully metered: the ledger row (including `ttft_ms`/`inter_token_ms`) is written server-side once the stream completes (see "Real upstream streaming" below), and a client that requests `stream_options.include_usage` still gets the upstream's real `usage` object on the final SSE chunk, exactly as the OpenAI streaming API itself works - it is simply not summarized into response headers. +### Upstream header collision policy (`usage.upstream_reported_model`, `gateway._apply_served_model`) + +An upstream this gateway forwards to may itself be a DevPlace-style gateway (or any provider that happens to emit the same `X-Gateway-*` header names on its own response), so its raw response headers can collide with the ones we are about to build. **Exactly one upstream header is ever trusted and relayed: `X-Gateway-Model`.** Every other `X-Gateway-*`/`X-App-Reference` header is always our own measurement, computed as if the upstream had sent nothing at all - an upstream cannot be an authority on our own cost table, our own tokenizer counts, our own socket timings, or our own caller attribution. + +- **Why only the model name.** It is the one field an upstream can legitimately know better than we do: it may have resolved an alias, redirected to a fallback, or served a different pinned version than what we requested. Cost/tokens/latency/context/app-reference are, by construction, things WE measured about OUR call (our pricing table, our tokenizer expectations, our socket, our caller) - an upstream's own figures for those describe its own accounting, not ours, and blending them in would corrupt `gateway_usage_ledger`'s per-model cost/token rollups and the quota math that reads these exact header names back out (`usage.parse_usage_headers`, consumed by `services/correction.py`/`services/ai_modifier.py`). +- **`usage.upstream_reported_model(headers)`** extracts that one value defensively: it treats the upstream response as untrusted input and never raises. It looks up `X-Gateway-Model` case-insensitively (a direct `.get`, then a manual case-insensitive scan for a plain non-case-insensitive mapping), and rejects anything that is not a non-empty string, exceeds 200 characters, or contains a control character (`\r`/`\n`/`\x00`, a header-injection guard) - any of those makes it return `None`, i.e. "upstream said nothing usable." +- **`gateway._apply_served_model(headers, resp)`** is the one place this is applied: called at the tail of every response-header build (chat non-streaming and streaming, embeddings, images, passthrough), it overwrites `headers["X-Gateway-Model"]` with the upstream's claim ONLY when one was found, and leaves every other header untouched. It never touches the ledger row itself (`GatewayUsageLedger.record`'s `model` field, and therefore `context_utilization`/pricing lookups, stay keyed on the model WE resolved and billed against) - the override is display-only, informing the client which model actually answered without altering what we measured or charged for. +- **Tests:** `tests/unit/services/openai_gateway/usage.py` (`upstream_reported_model` - case-insensitivity, missing/oversized/control-character rejection, non-dict/`None` inputs) and `tests/unit/services/openai_gateway/gateway.py` (`test_upstream_x_gateway_model_header_is_forwarded_*` - chat/streaming/embeddings/images/passthrough all forward an upstream-reported model while every cost/token header stays ours; `test_upstream_x_gateway_model_header_absent_keeps_our_own_model` - no upstream header is a no-op). + ## App-reference header (`X-App-Reference`) Callers may send an optional `X-App-Reference` header to tag gateway calls by application. The value is validated and stored in the `app_reference` column of `gateway_usage_ledger`, surfaced in analytics and admin reporting. diff --git a/devplacepy/services/openai_gateway/gateway.py b/devplacepy/services/openai_gateway/gateway.py index b96008b..9d55933 100644 --- a/devplacepy/services/openai_gateway/gateway.py +++ b/devplacepy/services/openai_gateway/gateway.py @@ -31,6 +31,7 @@ from devplacepy.services.openai_gateway.usage import ( extract_params, parse_context_map, pricing_from_cfg, + upstream_reported_model, usage_response_headers, ) from devplacepy.services.openai_gateway.vision import VisionAugmenter, VisionCache @@ -44,6 +45,19 @@ def _call_failed(resp, exc, timing: dict) -> bool: return resp is not None and resp.status_code >= 400 +def _apply_served_model(headers: dict, resp) -> dict: + # Everything else in `headers` (cost, tokens, latency, context) is our + # own measurement and must never be overwritten by an upstream that + # happens to echo the same X-Gateway-* header names - only the served + # model name is worth trusting from upstream. See usage.upstream_reported_model. + if not headers or resp is None: + return headers + served = upstream_reported_model(getattr(resp, "headers", None)) + if served: + headers["X-Gateway-Model"] = served + return headers + + def _fallback_headers(cfg: dict, key_field: str) -> dict: headers = {"Content-Type": "application/json"} if cfg.get(key_field): @@ -407,7 +421,7 @@ class GatewayRuntime: chat_cost = float((row or {}).get("cost_usd") or 0.0) headers["X-Gateway-Cost-USD"] = f"{chat_cost + vision_cost:.8f}" headers["X-Gateway-Vision-Cost-USD"] = f"{vision_cost:.8f}" - return headers + return _apply_served_model(headers, resp) if timing["circuit_open"]: resp_headers = finalize(503, False, "circuit_open") @@ -465,11 +479,14 @@ class GatewayRuntime: log, ), media_type="text/event-stream", - headers={ - "X-Gateway-Model": model, - "X-Gateway-Backend": "chat", - "X-App-Reference": app_reference or "default", - }, + headers=_apply_served_model( + { + "X-Gateway-Model": model, + "X-Gateway-Backend": "chat", + "X-App-Reference": app_reference or "default", + }, + resp, + ), ) try: data = resp.json() @@ -734,7 +751,7 @@ class GatewayRuntime: chat_cost = float((row or {}).get("cost_usd") or 0.0) headers["X-Gateway-Cost-USD"] = f"{chat_cost + vision_cost:.8f}" headers["X-Gateway-Vision-Cost-USD"] = f"{vision_cost:.8f}" - return headers + return _apply_served_model(headers, resp) if timing["circuit_open"]: resp_headers = finalize(503, False, "circuit_open") @@ -939,7 +956,7 @@ class GatewayRuntime: base["error_category"] = category base["usage"] = usage row = self._ledger.record(base, pricing, context_map) - return usage_response_headers(row) + return _apply_served_model(usage_response_headers(row), resp) if timing["circuit_open"]: resp_headers = finalize(503, False, "circuit_open") @@ -1064,7 +1081,7 @@ class GatewayRuntime: chat_cost = float((row or {}).get("cost_usd") or 0.0) headers["X-Gateway-Cost-USD"] = f"{chat_cost + vision_cost:.8f}" headers["X-Gateway-Vision-Cost-USD"] = f"{vision_cost:.8f}" - return headers + return _apply_served_model(headers, resp) if timing["circuit_open"]: resp_headers = finalize(503, False, "circuit_open") diff --git a/devplacepy/services/openai_gateway/usage.py b/devplacepy/services/openai_gateway/usage.py index abe84c3..f8d260b 100644 --- a/devplacepy/services/openai_gateway/usage.py +++ b/devplacepy/services/openai_gateway/usage.py @@ -408,6 +408,49 @@ def usage_response_headers(row: Optional[dict]) -> dict: return headers +UPSTREAM_MODEL_HEADER_MAX_LEN = 200 + + +def upstream_reported_model(headers: Any) -> Optional[str]: + """Extract a served-model name an upstream (which may itself be another + DevPlace-style gateway) reported on its own X-Gateway-Model response + header, if present and well-formed. + + This is the ONE upstream response header this gateway ever trusts and + relays: model name is the single field an upstream can know better than + we do (it may have resolved an alias/fallback on its own side). Every + other X-Gateway-* figure (cost, tokens, latency, context) stays our own + measurement regardless of what an upstream claims - see usage_response_headers. + + An upstream response is untrusted input, so this never raises: a missing, + malformed, oversized, or control-character-laced value is treated as + absent rather than propagated or allowed to break our own response. + """ + if not headers: + return None + value = None + try: + value = headers.get("X-Gateway-Model") + except Exception: + value = None + if value is None: + try: + for key, candidate in headers.items(): + if isinstance(key, str) and key.lower() == "x-gateway-model": + value = candidate + break + except Exception: + value = None + if not isinstance(value, str): + return None + value = value.strip() + if not value or len(value) > UPSTREAM_MODEL_HEADER_MAX_LEN: + return None + if any(ch in value for ch in ("\r", "\n", "\x00")): + return None + return value + + def parse_usage_headers(headers) -> Optional[dict]: if not headers or "X-Gateway-Cost-USD" not in headers: return None diff --git a/pyproject.toml b/pyproject.toml index 108494f..e3dc038 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "devplacepy" -version = "1.0.4" +version = "1.0.5" description = "DevPlace - The Developer Social Network" requires-python = ">=3.12" dependencies = [ diff --git a/tests/unit/services/openai_gateway/gateway.py b/tests/unit/services/openai_gateway/gateway.py index 58c5941..ad7581b 100644 --- a/tests/unit/services/openai_gateway/gateway.py +++ b/tests/unit/services/openai_gateway/gateway.py @@ -8,11 +8,18 @@ from devplacepy.utils import generate_uid import devplacepy.services.openai_gateway.gateway as gwmod from devplacepy.services.openai_gateway import GatewayService class FakeResp_openai_gateway: - def __init__(self, status=200, payload=None, ctype="application/json", content=b""): + def __init__( + self, + status=200, + payload=None, + ctype="application/json", + content=b"", + extra_headers=None, + ): self.status_code = status self._payload = payload self.text = json.dumps(payload) if payload is not None else "" - self.headers = {"content-type": ctype} + self.headers = {"content-type": ctype, **(extra_headers or {})} self.content = content def json(self): @@ -1309,3 +1316,258 @@ def test_models_endpoint_publishes_molodetz(local_db): finally: routing.model_store.remove("molodetz") routing.model_store.remove("molodetz-pro") + + +# Upstream header collision policy: only X-Gateway-Model is ever trusted and +# forwarded from an upstream that happens to speak our own X-Gateway-* header +# convention (e.g. another DevPlace-style gateway). Every other figure +# (cost/tokens/latency/context/app-reference) must remain ours regardless of +# what an upstream claims. See usage.upstream_reported_model / CLAUDE.md. + + +class FakeClientUpstreamModelHeader_openai_gateway(FakeClient_openai_gateway): + async def send(self, request, stream=False): + self.calls.append((request.url, request.json_body)) + body = request.json_body or {} + return FakeResp_openai_gateway( + payload={ + "id": "x", + "model": body.get("model"), + "choices": [{"message": {"content": "hi there"}}], + }, + extra_headers={"X-Gateway-Model": "upstream/served-model-x"}, + ) + + +class FakeClientHostileModelHeader_openai_gateway(FakeClient_openai_gateway): + async def send(self, request, stream=False): + self.calls.append((request.url, request.json_body)) + body = request.json_body or {} + return FakeResp_openai_gateway( + payload={ + "id": "x", + "model": body.get("model"), + "choices": [{"message": {"content": "hi there"}}], + }, + extra_headers={"X-Gateway-Model": "evil\r\nX-Injected: true"}, + ) + + +class FakeEmbedClientUpstreamModelHeader_openai_gateway(FakeEmbedClient_openai_gateway): + async def send(self, request, stream=False): + self.calls.append((request.url, request.json_body)) + body = request.json_body or {} + return FakeResp_openai_gateway( + payload={ + "object": "list", + "model": body.get("model"), + "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}], + "usage": {"prompt_tokens": 5, "total_tokens": 5}, + }, + extra_headers={"X-Gateway-Model": "upstream/embed-served"}, + ) + + +class FakeImageClientUpstreamModelHeader_openai_gateway(FakeImageClient_openai_gateway): + async def send(self, request, stream=False): + self.calls.append((request.url, request.json_body)) + body = request.json_body or {} + return FakeResp_openai_gateway( + payload={ + "created": 1, + "model": body.get("model"), + "data": [{"b64_json": "aGVsbG8="}], + "usage": {"cost": 0.05}, + }, + extra_headers={"X-Gateway-Model": "upstream/image-served"}, + ) + + +class FakePassthroughClientUpstreamModelHeader_openai_gateway: + def __init__(self, *a, **k): + self.calls = [] + + def build_request(self, method, url, headers=None, json=None, content=None): + return FakeRequest(method, url, json) + + async def send(self, request, stream=False): + self.calls.append((request.url, request.json_body)) + return FakeResp_openai_gateway( + payload={"ok": True}, + extra_headers={"X-Gateway-Model": "upstream/passthrough-served"}, + ) + + async def aclose(self): + pass + + +def test_upstream_x_gateway_model_header_is_forwarded_for_chat(local_db, monkeypatch): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakeClientUpstreamModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_force_model"] = True + cfg["gateway_model"] = "deepseek-chat" + rt = svc.runtime() + resp = run_async( + rt.handle_chat( + {"messages": [{"role": "user", "content": "hi"}]}, + cfg, + ("guest", "upstream_model_chat"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "upstream/served-model-x" + row = get_table("gateway_usage_ledger").find_one(owner_id="upstream_model_chat") + assert row is not None + assert row["model"] == "deepseek-chat" + + +def test_upstream_x_gateway_model_header_is_forwarded_for_streaming_chat( + local_db, monkeypatch +): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakeClientUpstreamModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + rt = svc.runtime() + resp = run_async( + rt.handle_chat( + {"messages": [{"role": "user", "content": "hi"}], "stream": True}, + cfg, + ("guest", "upstream_model_stream"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "upstream/served-model-x" + assert resp.headers["X-Gateway-Backend"] == "chat" + + async def drain(): + async for _ in resp.body_iterator: + pass + + run_async(drain()) + + +def test_upstream_x_gateway_model_header_is_forwarded_for_embeddings( + local_db, monkeypatch +): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakeEmbedClientUpstreamModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_embed_enabled"] = True + cfg["gateway_embed_model"] = "qwen/qwen3-embedding-8b" + rt = svc.runtime() + resp = run_async( + rt.handle_embeddings( + {"input": "hello"}, + cfg, + ("guest", "upstream_model_embed"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "upstream/embed-served" + row = get_table("gateway_usage_ledger").find_one(owner_id="upstream_model_embed") + assert row is not None + assert row["model"] == "qwen/qwen3-embedding-8b" + + +def test_upstream_x_gateway_model_header_is_forwarded_for_images(local_db, monkeypatch): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakeImageClientUpstreamModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_image_enabled"] = True + cfg["gateway_image_model"] = "black-forest-labs/flux.2-pro" + rt = svc.runtime() + resp = run_async( + rt.handle_images( + {"prompt": "badge"}, + cfg, + ("guest", "upstream_model_img"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "upstream/image-served" + row = get_table("gateway_usage_ledger").find_one(owner_id="upstream_model_img") + assert row is not None + assert row["model"] == "black-forest-labs/flux.2-pro" + + +def test_upstream_x_gateway_model_header_is_forwarded_for_passthrough( + local_db, monkeypatch +): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakePassthroughClientUpstreamModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_model"] = "deepseek-chat" + rt = svc.runtime() + resp = run_async( + rt.handle_passthrough( + "POST", + "responses", + "application/json", + b"{}", + cfg, + ("guest", "upstream_model_passthrough"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "upstream/passthrough-served" + row = get_table("gateway_usage_ledger").find_one( + owner_id="upstream_model_passthrough" + ) + assert row is not None + assert row["model"] == "deepseek-chat" + + +def test_no_upstream_model_header_keeps_our_own_model(local_db, monkeypatch): + monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_force_model"] = True + cfg["gateway_model"] = "deepseek-chat" + rt = svc.runtime() + resp = run_async( + rt.handle_chat( + {"messages": [{"role": "user", "content": "hi"}]}, + cfg, + ("guest", "no_upstream_header"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "deepseek-chat" + + +def test_hostile_upstream_model_header_is_ignored_end_to_end(local_db, monkeypatch): + monkeypatch.setattr( + gwmod.httpx, "AsyncClient", FakeClientHostileModelHeader_openai_gateway + ) + svc = GatewayService() + cfg = svc.effective_config() + cfg["gateway_force_model"] = True + cfg["gateway_model"] = "deepseek-chat" + rt = svc.runtime() + resp = run_async( + rt.handle_chat( + {"messages": [{"role": "user", "content": "hi"}]}, + cfg, + ("guest", "hostile_header"), + "test", + "default", + ) + ) + assert resp.headers["X-Gateway-Model"] == "deepseek-chat" diff --git a/tests/unit/services/openai_gateway/usage.py b/tests/unit/services/openai_gateway/usage.py index 22fe9ae..a177c01 100644 --- a/tests/unit/services/openai_gateway/usage.py +++ b/tests/unit/services/openai_gateway/usage.py @@ -11,6 +11,7 @@ from devplacepy.services.openai_gateway.usage import ( new_usage_totals, normalize_usage, parse_usage_headers, + upstream_reported_model, usage_metric_cards, ) @@ -248,3 +249,79 @@ def test_validate_app_reference_strips_whitespace(): assert _validate_app_reference(" my-app ") == "my-app" assert _validate_app_reference("\tdevplace\t") == "devplace" + + +class DictOnlyHeaders(dict): + """A plain mapping with no case-insensitive .get, like a hand-rolled + fake response's headers dict in a test double - exercises the + manual case-insensitive fallback scan.""" + + +def test_upstream_reported_model_reads_exact_case_via_get(): + assert upstream_reported_model({"X-Gateway-Model": "vendor/served-model"}) == ( + "vendor/served-model" + ) + + +def test_upstream_reported_model_is_case_insensitive_via_real_headers(): + import httpx + + headers = httpx.Headers({"x-gateway-model": "vendor/lowercase-model"}) + assert upstream_reported_model(headers) == "vendor/lowercase-model" + + +def test_upstream_reported_model_falls_back_to_manual_scan_for_plain_dicts(): + headers = DictOnlyHeaders({"x-GATEWAY-model": "vendor/mixed-case"}) + assert upstream_reported_model(headers) == "vendor/mixed-case" + + +def test_upstream_reported_model_strips_surrounding_whitespace(): + assert upstream_reported_model({"X-Gateway-Model": " vendor/spaced "}) == ( + "vendor/spaced" + ) + + +def test_upstream_reported_model_missing_header_returns_none(): + assert upstream_reported_model({}) is None + assert upstream_reported_model({"Content-Type": "application/json"}) is None + + +def test_upstream_reported_model_handles_falsy_and_malformed_containers(): + assert upstream_reported_model(None) is None + assert upstream_reported_model({}) is None + assert upstream_reported_model(object()) is None + assert upstream_reported_model(["not", "a", "mapping"]) is None + + +def test_upstream_reported_model_rejects_empty_or_whitespace_only(): + assert upstream_reported_model({"X-Gateway-Model": ""}) is None + assert upstream_reported_model({"X-Gateway-Model": " "}) is None + + +def test_upstream_reported_model_rejects_non_string_value(): + assert upstream_reported_model({"X-Gateway-Model": 12345}) is None + assert upstream_reported_model({"X-Gateway-Model": None}) is None + + +def test_upstream_reported_model_rejects_oversized_value(): + huge = "x" * 201 + assert upstream_reported_model({"X-Gateway-Model": huge}) is None + exactly_max = "x" * 200 + assert upstream_reported_model({"X-Gateway-Model": exactly_max}) == exactly_max + + +def test_upstream_reported_model_rejects_control_characters(): + assert upstream_reported_model({"X-Gateway-Model": "vendor\r\nInjected: header"}) is None + assert upstream_reported_model({"X-Gateway-Model": "vendor\nmodel"}) is None + assert upstream_reported_model({"X-Gateway-Model": "vendor\x00model"}) is None + + +def test_upstream_reported_model_get_raising_never_propagates(): + class HostileHeaders: + def get(self, key, default=None): + raise RuntimeError("boom") + + def items(self): + raise RuntimeError("boom too") + + assert upstream_reported_model(HostileHeaders()) is None