diff --git a/docs/architecture.md b/docs/architecture.md index 30766ce..9f5eaaf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -21,8 +21,11 @@ constitute a general billing cap or a global distributed rate limiter. The backend sees only what Cerberus sends it: a request Cerberus has already authorized against the caller's identity and alias policy, under the credential the alias names. A caller cannot select the panel, the analyst, or the tool -surface of a fusion request, and cannot reach the backend except through a -fusion alias it is allowed to use. Fusion-mode deliberation is not streamed. +surface of a fusion request: a body carrying `tools`, `tool_choice`, or any of +the model-selection keys every alias refuses (see Dispatch control plane) is +answered 400 before the backend is called. A caller cannot reach the backend +except through a fusion alias it is allowed to use. Fusion-mode deliberation is +not streamed. ## Dispatch control plane @@ -43,6 +46,23 @@ per-request failures are excluded at their own scope by the cooldown store. Normal label lookup reads the in-memory snapshot and performs no SQL. Routing and usage records are written asynchronously after a decision. +The request body cannot add a path either. Cerberus overwrites `model` with the +target it selected, and refuses with a 400, before any upstream call, a body +carrying a key an upstream reads as model selection: `models` and `route` +(OpenRouter's fallback list), `plugins`, `preset`, or a server tool, meaning an +entry in `tools` or a `tool_choice` whose `type` starts with `openrouter:` +(`openrouter:fusion` names a panel and analyst of its own). Any of them would let +a caller run, on the operator's credential, a model the pinned revision never +named and the free/paid cost gate never weighed. The rule is +`engine.caller_model_selection`. It applies to every alias mode and every +provider, OpenRouter or not, because the failover loop may reach any candidate. +Refusing rather than stripping keeps the substitution from ever being silent, as +with `reasoning_effort_override`. The refusal is recorded as a routing event, +outcome `upstream_error` with status 400 and no attempts, the same shape fusion +records. Function tools, a `tool_choice` naming one, and keys that leave the +model alone pass through unchanged. That includes `provider` preferences, which +only choose among hosts for the model Cerberus already pinned. + `/health` is the only ungated endpoint and answers liveness only — `{"status": "ok", "service": "cerberus"}` — which is what a container or orchestrator probe needs. Operational diagnostics (config version and checksum, diff --git a/src/cerberus/fusion/dispatch.py b/src/cerberus/fusion/dispatch.py index e7c7411..2bacb90 100644 --- a/src/cerberus/fusion/dispatch.py +++ b/src/cerberus/fusion/dispatch.py @@ -22,12 +22,15 @@ from cerberus.fusion.backend import FusionBackend, FusionError, FusionRequest from cerberus.registry.loader import ConfigDocument from cerberus.registry.schema import Alias, CerberusConfig +from cerberus.router.engine import CALLER_ROUTING_KEYS from cerberus.telemetry import RoutingAttempt, RoutingEvent, TelemetryEmitter # Caller-supplied keys a fusion alias cannot honor: the backend owns the tool # surface during deliberation, so accepting them would silently drop or reroute -# the caller's intent. Rejected up front rather than forwarded. -_REJECTED_CALLER_KEYS = ("tools", "tool_choice", "plugins") +# the caller's intent. Rejected up front rather than forwarded. The keys every +# alias refuses because they choose models upstream (models, route, ...) would +# otherwise reach OpenRouter beside the deliberation, so they are refused too. +_REJECTED_CALLER_KEYS = ("tools", "tool_choice", *CALLER_ROUTING_KEYS) def fusion_aliases(config: CerberusConfig) -> list[str]: diff --git a/src/cerberus/router/dispatch.py b/src/cerberus/router/dispatch.py index e00f7e9..1232c25 100644 --- a/src/cerberus/router/dispatch.py +++ b/src/cerberus/router/dispatch.py @@ -26,7 +26,7 @@ ) from cerberus.registry.loader import ConfigDocument from cerberus.router.availability import runtime_exclusion -from cerberus.router.engine import Target, cost_eligible, ordered_targets +from cerberus.router.engine import Target, caller_model_selection, cost_eligible, ordered_targets from cerberus.state.cooldowns import InMemoryCooldownStore from cerberus.telemetry import RoutingAttempt, RoutingEvent, RoutingOutcome, TelemetryEmitter @@ -199,6 +199,41 @@ async def dispatch( targets = ordered_targets(config, alias_name) ordered = [f"{t.provider_id}/{t.credential_id}/{t.model}" for t in targets] + + # The pinned revision alone chooses the model. A body that would choose or + # add one upstream is refused before any provider sees it, whichever + # candidate the loop would have reached, so the refusal is never silent. + rejected = caller_model_selection(body) + if rejected: + telemetry.emit( + _event( + request_id=request_id, + alias=alias_name, + mode=alias.mode, + target=None, + attempts=[], + http_status=400, + outcome="upstream_error", + started_at=started_at, + usage=None, + streaming=streaming, + fallback=False, + identity=identity_name, + config_version=document.version, + config_checksum=document.checksum, + candidates=ordered, + ) + ) + return JSONResponse( + status_code=400, + content={ + "error": { + "message": f"alias does not accept caller-supplied model selection: {', '.join(rejected)}", + "request_id": request_id, + } + }, + ) + eligible, exclusions = cost_eligible(alias, targets) attempts: list[RoutingAttempt] = [] attempted = 0 diff --git a/src/cerberus/router/engine.py b/src/cerberus/router/engine.py index aaef640..a29970e 100644 --- a/src/cerberus/router/engine.py +++ b/src/cerberus/router/engine.py @@ -1,8 +1,9 @@ -"""Deterministic target resolution and cost-tier gating. +"""Deterministic target resolution, cost-tier gating and the caller routing gate. Policy dispatch is deterministic: same config version + same alias -> same ordered target list. Runtime failover over that order is state-dependent and -belongs to the dispatch loop, not here. +belongs to the dispatch loop, not here. Neither may a request body widen that +list: caller_model_selection names the keys that would choose models upstream. """ from __future__ import annotations @@ -13,6 +14,17 @@ from cerberus.registry.schema import Alias, CerberusConfig, CostTier +# Caller-supplied body keys an upstream reads as model selection. OpenRouter +# honours every one: models/route are its fallback list, and plugins and preset +# can carry routing of their own. Forwarding any of them would let a caller run, +# on the operator's credential, a model the pinned revision never named and +# cost_eligible never gated. +CALLER_ROUTING_KEYS = ("models", "route", "plugins", "preset") +# OpenRouter server tools execute upstream under the operator's credential, and +# some run models of their own (openrouter:fusion names a whole panel and +# analyst). Refused as a class by type prefix; function tools are untouched. +SERVER_TOOL_PREFIX = "openrouter:" + @dataclass(frozen=True, slots=True) class Target: @@ -96,3 +108,25 @@ def cost_eligible(alias: Alias, targets: list[Target]) -> tuple[list[Target], li continue eligible.append(target) return eligible, exclusions + + +def caller_model_selection(body: Mapping[str, Any]) -> list[str]: + """The request-body keys that would choose or add models upstream. + + Empty means the body leaves model selection to the pinned revision. A + non-empty answer is refused before any upstream call rather than stripped, + so the caller learns its routing was not honoured instead of being served + by a model it did not ask for. + """ + found = [key for key in CALLER_ROUTING_KEYS if key in body] + tools = body.get("tools") + if any(_is_server_tool(tool) for tool in (tools if isinstance(tools, list) else [tools])): + found.append("tools") + if _is_server_tool(body.get("tool_choice")): + found.append("tool_choice") + return found + + +def _is_server_tool(value: Any) -> bool: + kind = value.get("type") if isinstance(value, Mapping) else value + return isinstance(kind, str) and kind.strip().lower().startswith(SERVER_TOOL_PREFIX) diff --git a/tests/test_caller_model_selection.py b/tests/test_caller_model_selection.py new file mode 100644 index 0000000..775f9fc --- /dev/null +++ b/tests/test_caller_model_selection.py @@ -0,0 +1,132 @@ +"""A request body cannot choose models upstream on a dispatch-path alias. + +``build_upstream_request`` overwrites only ``model``, so any other key reaches +the provider as sent. OpenRouter reads several of them as model selection +(``models``/``route`` fallback lists, ``plugins``, ``preset``, ``openrouter:*`` +server tools), which would let a caller run models absent from the pinned +revision on the operator's credential. The upstream is mocked at the HTTP +boundary; these prove such a body is refused before it, while tool calling and +keys that leave the model alone are forwarded untouched. +""" + +import httpx +import pytest + +from cerberus.app import create_app +from cerberus.registry import load_config_document +from cerberus.router.engine import caller_model_selection +from tests.test_control import ok_upstream, write_config +from tests.test_fusion_dispatch import AUTH, Capture, client_for, fusion_raw + +FUSION_TOOL = { + "type": "openrouter:fusion", + "parameters": {"analysis_models": ["openai/gpt-5", "anthropic/claude-opus-4"], "model": "openai/gpt-5"}, +} +FUNCTION_TOOL = { + "type": "function", + "function": {"name": "lookup", "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}}, +} +ALIASES = ("cerberus/dispatch-dev", "cerberus/free-dev") + + +def raw_config() -> dict: + """The fusion-dev slice (an OpenRouter dispatch candidate) plus a free alias.""" + raw = fusion_raw() + raw["aliases"]["cerberus/free-dev"] = { + "mode": "free", + "candidates": [{"provider": "openrouter", "credential": "main", "model": "free-b"}], + } + raw["identities"]["dev"]["allowed_modes"].append("free") + raw["identities"]["dev"]["allowed_aliases"].append("cerberus/free-dev") + return raw + + +def make_app(monkeypatch, tmp_path): + monkeypatch.setenv("OPENROUTER_API_KEY", "or-key") + monkeypatch.setenv("CB_KEY_DEV", "cb-dev") + monkeypatch.setenv("CB_KEY_DISPATCH", "cb-dispatch") + doc = load_config_document(write_config(tmp_path, "v.yaml", raw_config()), validate_credentials=False) + upstream = Capture(ok_upstream) + return create_app(doc, http_transport=httpx.MockTransport(upstream)), upstream + + +async def call(app, body): + async with app.router.lifespan_context(app): + async with client_for(app) as client: + resp = await client.post("/v1/chat/completions", json=body, headers=AUTH) + events = (await client.get("/admin/events")).json()["events"] + return resp, events + + +def request(alias: str, **extra) -> dict: + return {"model": alias, "messages": [{"role": "user", "content": "hi"}], **extra} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("alias", ALIASES) +@pytest.mark.parametrize( + ("key", "extra"), + [ + ("models", {"models": ["openai/gpt-5", "anthropic/claude-opus-4"]}), + ("route", {"route": "fallback"}), + ("plugins", {"plugins": [{"id": "jev-router"}]}), + ("preset", {"preset": "operator-preset"}), + ("tools", {"tools": [FUNCTION_TOOL, FUSION_TOOL]}), + ("tool_choice", {"tool_choice": {"type": "openrouter:fusion"}}), + ("models", {"models": ["openai/gpt-5"], "stream": True}), + ], + ids=["models", "route", "plugins", "preset", "openrouter-fusion-tool", "openrouter-tool-choice", "streaming"], +) +async def test_model_selection_keys_are_rejected_before_any_upstream_call(monkeypatch, tmp_path, alias, key, extra): + app, upstream = make_app(monkeypatch, tmp_path) + resp, events = await call(app, request(alias, **extra)) + + assert upstream.requests == [], "a body that selects models must never reach the provider" + assert resp.status_code == 400 + error = resp.json()["error"] + assert key in error["message"] and error["request_id"] + assert len(events) == 1 + event = events[0] + assert event["outcome"] == "upstream_error" and event["http_status"] == 400 + assert event["alias"] == alias and event["identity"] == "dev" + assert event["request_id"] == error["request_id"] + assert event["attempt_count"] == 0 and event["provider"] is None and event["model"] is None + assert event["streaming"] is bool(extra.get("stream")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("alias", ALIASES) +async def test_function_tools_and_non_routing_keys_are_forwarded_unchanged(monkeypatch, tmp_path, alias): + app, upstream = make_app(monkeypatch, tmp_path) + tool_choice = {"type": "function", "function": {"name": "lookup"}} + # provider preferences narrow which host serves the pinned model; they never pick another one + provider = {"data_collection": "deny", "allow_fallbacks": False} + resp, events = await call( + app, request(alias, tools=[FUNCTION_TOOL], tool_choice=tool_choice, provider=provider, temperature=0.2) + ) + + assert resp.status_code == 200 + assert len(upstream.requests) == 1 + sent = upstream.body + assert sent["model"] == ("free-a" if alias == "cerberus/dispatch-dev" else "free-b") + assert sent["tools"] == [FUNCTION_TOOL] + assert sent["tool_choice"] == tool_choice + assert sent["provider"] == provider and sent["temperature"] == 0.2 + assert events[0]["outcome"] == "success" + + +@pytest.mark.parametrize( + ("body", "expected"), + [ + ({"messages": []}, []), + ({"tools": [FUNCTION_TOOL], "tool_choice": "auto"}, []), + ({"tool_choice": "required", "provider": {"order": ["x"]}}, []), + ({"tools": None, "tool_choice": None}, []), + ({"tools": [{"type": "OpenRouter:Web_Search"}]}, ["tools"]), + ({"tools": FUSION_TOOL}, ["tools"]), + ({"tool_choice": "openrouter:fusion"}, ["tool_choice"]), + ({"models": [], "route": None, "plugins": [], "preset": ""}, ["models", "route", "plugins", "preset"]), + ], +) +def test_caller_model_selection_classifies_bodies(body, expected): + assert caller_model_selection(body) == expected diff --git a/tests/test_fusion_dispatch.py b/tests/test_fusion_dispatch.py index 83167cc..2d1857d 100644 --- a/tests/test_fusion_dispatch.py +++ b/tests/test_fusion_dispatch.py @@ -216,9 +216,11 @@ async def test_caller_cannot_override_panel_or_tool_surface(monkeypatch, tmp_pat resp, _ = await call(app, {**FUSION_REQ, "stream": True}) assert resp.status_code == 200 assert capture.body["model"] == "free-a" and "stream" not in capture.body - # ... but tools/tool_choice/plugins cannot be honored through a fusion alias: reject, never forward + # ... but tools/tool_choice/plugins cannot be honored through a fusion alias, and models/route/preset + # would choose models beside the deliberation: reject, never forward for key, value in (("plugins", [{"id": "fusion", "analysis_models": ["free-c"]}]), ("tool_choice", "none"), - ("tools", [{"type": "function", "function": {"name": "x"}}])): + ("tools", [{"type": "function", "function": {"name": "x"}}]), + ("models", ["paid-x"]), ("route", "fallback"), ("preset", "operator-preset")): app, capture = make_app(monkeypatch, tmp_path) resp, events = await call(app, {**FUSION_REQ, key: value}) assert resp.status_code == 400, key