From 336c7c08496c1938824e52361b738edbeda6b7a6 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Tue, 29 Sep 2026 10:57:28 -0700 Subject: [PATCH] test(integration): sweep proxy logs, metrics, a Datadog intake and the Logs drawer for credential canaries (#43306) * test(integration): credential canary suite harness Adds tests/integration/security with canary generation and search, sweeps over the database, GET routes, client responses, sink doubles and Redis, an owned proxy rig, a sweep sensitivity self-test and the config deployment api_key slot. Registers the security group in run.py, the manifest and the CircleCI integration matrix. * test(integration): widen canary route sweep and harden the rig Enumerate lazily registered feature routers, call parameterized routes with placeholder ids, fail on routes that return no response, skip provider pass-through routes, add an explicit admin-only route allowance, let the sink double use a configurable token, inflate gzip members anywhere in a blob, sweep Redis before the route walk, and trap outbound connections from the owned proxy. * test(integration): descend into any decoded value that can still hold an encoded canary * test(integration): bound canary decoding by depth and decoded bytes * test(integration): scope log-table and spend-log reads to the scenario window * test(integration): sweep spend-log rows in the scenario date window * test(integration): keep spend-log date window summarized * test(integration): sweep proxy logs, metrics, a gzip Datadog intake and the Logs drawer for credential canaries * test(e2e): treat an unset prompt-storage setting as unset and restore it * test(integration): name the Datadog sink slot G1d * test(e2e): search the Logs page for base64 forms of the deployment key * test(integration): resolve deployment ids, scope paginated log lists, key allowances by slot * test(integration): pass the resolved deployment id to the Datadog route sweep * test(integration): expect 404 from the caller-scoped team membership route * test(integration): use the rig's own master key and expect 404 from submission lookups * test(integration): check the overridden rig key without assuming the default key is unknown --- .../tests/integrationCritical/expected.json | 3 +- .../logsDrawerCredentialCanary.spec.ts | 224 ++++++++++++++++++ tests/integration/security/_canary.py | 1 + .../integration/security/test_datadog_sink.py | 169 +++++++++++++ .../integration/security/test_metrics_text.py | 89 +++++++ tests/integration/security/test_proxy_logs.py | 84 +++++++ 6 files changed, 569 insertions(+), 1 deletion(-) create mode 100644 tests/e2e/ui/tests/integrationCritical/logsDrawerCredentialCanary.spec.ts create mode 100644 tests/integration/security/test_datadog_sink.py create mode 100644 tests/integration/security/test_metrics_text.py create mode 100644 tests/integration/security/test_proxy_logs.py diff --git a/tests/e2e/ui/tests/integrationCritical/expected.json b/tests/e2e/ui/tests/integrationCritical/expected.json index b73c04acbe0..1614b188188 100644 --- a/tests/e2e/ui/tests/integrationCritical/expected.json +++ b/tests/e2e/ui/tests/integrationCritical/expected.json @@ -7,5 +7,6 @@ "tests/e2e/ui/tests/integrationCritical/mcpUserEnvVars.spec.ts::a server with two per-user variables reports the remaining gap until both are saved", "tests/e2e/ui/tests/integrationCritical/mcpUserEnvVars.spec.ts::a server without per-user variables shows no credential row", "tests/e2e/ui/tests/integrationCritical/mcpUserEnvVars.spec.ts::clearing credentials for a server deleted underneath the modal reports the failure without losing the page", - "tests/e2e/ui/tests/integrationCritical/costOptimizationModelGroups.spec.ts::cache leakage by model merges a deployment's resolved and requested model names into its model group" + "tests/e2e/ui/tests/integrationCritical/costOptimizationModelGroups.spec.ts::cache leakage by model merges a deployment's resolved and requested model names into its model group", + "tests/e2e/ui/tests/integrationCritical/logsDrawerCredentialCanary.spec.ts::the Logs drawer renders the stored request without the deployment api_key" ] diff --git a/tests/e2e/ui/tests/integrationCritical/logsDrawerCredentialCanary.spec.ts b/tests/e2e/ui/tests/integrationCritical/logsDrawerCredentialCanary.spec.ts new file mode 100644 index 00000000000..1d86dee107d --- /dev/null +++ b/tests/e2e/ui/tests/integrationCritical/logsDrawerCredentialCanary.spec.ts @@ -0,0 +1,224 @@ +import { test, expect, type APIRequestContext } from "@playwright/test"; +import { randomUUID } from "node:crypto"; +import { Page } from "../../fixtures/pages"; +import { dismissFeedbackPopup, navigateToPage } from "../../helpers/navigation"; + +/** + * Credential canary S8: what the Logs page renders for a request, including any client-side + * merge, never shows the deployment api_key that served it. + * + * A deployment is registered with a fresh canary api_key pointing at the owned upstream. The + * upstream must receive that canary as its bearer (positive control). The request carries a + * marker in its message content with stored prompts on, and the drawer must render that marker + * in both its pretty view and its raw request JSON view (sensitivity control: the stored request + * really reached the page) while the page's DOM holds no copy of the canary core in either view, + * raw or base64-encoded. + */ +const unhex = (): string => randomUUID().replaceAll("-", ""); + +/** + * The forms the canary core can take on the page: raw (JSON and percent encoding leave a hex + * core unchanged), and base64 in the standard and URL-safe alphabets at each of the three byte + * alignments it can start at. Each base64 form keeps only the characters that depend on core + * bytes alone, so it matches whatever bytes precede or follow the core. + */ +const canaryForms = (core: string): ReadonlyMap => { + const forms = new Map([["raw", core]]); + for (let offset = 0; offset < 3; offset++) { + const bytes = Buffer.concat([Buffer.alloc(offset), Buffer.from(core)]); + const first = offset === 0 ? 0 : 4; + const last = Math.floor(bytes.length / 3) * 4; + const text = bytes.toString("base64").slice(first, last); + forms.set(`base64@${offset}`, text); + forms.set( + `base64url@${offset}`, + text.replaceAll("+", "-").replaceAll("/", "_"), + ); + } + return forms; +}; + +/** The names of the canary forms found in ``text``; the raw form ignores case. */ +const foundForms = ( + text: string, + forms: ReadonlyMap, +): string[] => + [...forms] + .filter(([name, needle]) => + name === "raw" + ? text.toLowerCase().includes(needle) + : text.includes(needle), + ) + .map(([name]) => name); + +test("the Logs drawer renders the stored request without the deployment api_key", async ({ + page, + request, +}) => { + const master = process.env.LITELLM_MASTER_KEY ?? "sk-integration-master"; + const upstream = ( + process.env.INTEGRATION_UPSTREAM_URL ?? "http://127.0.0.1:8190" + ).replace(/\/+$/, ""); + const auth = { Authorization: `Bearer ${master}` }; + const canaryCore = unhex(); + const deploymentKey = `lkc-B1-${canaryCore}`; + const forms = canaryForms(canaryCore); + for (const prefix of ["", "k", "k:"]) { + const encoded = Buffer.from(`${prefix}${deploymentKey}`).toString("base64"); + expect( + foundForms(`Basic ${encoded}`, forms), + `the decoder misses base64 after a ${prefix.length}-byte prefix`, + ).not.toEqual([]); + } + const marker = `lkc-M0-${unhex()}`; + const model = `canary-drawer-${unhex()}`; + + const post = async (api: APIRequestContext, path: string, data: object) => { + const response = await api.post(path, { headers: auth, data }); + expect(response.status(), `POST ${path}: ${await response.text()}`).toBe( + 200, + ); + return response.json(); + }; + + const setting = await request.get( + "/config/field/info?field_name=store_prompts_in_spend_logs", + { headers: auth }, + ); + // A fresh database has no stored value, and the route answers 400 "... is not set". + const settingText = await setting.text(); + expect( + setting.status() === 200 || settingText.includes("is not set"), + settingText, + ).toBe(true); + const promptsStored: boolean | null = + setting.status() === 200 + ? JSON.parse(settingText).field_value === true + : null; + let modelId = ""; + try { + await post(request, "/config/update", { + general_settings: { store_prompts_in_spend_logs: true }, + }); + const created = await post(request, "/model/new", { + model_name: model, + litellm_params: { + model: "openai/gpt-4o-mini", + api_key: deploymentKey, + api_base: `${upstream}/v1`, + }, + }); + modelId = created.model_id; + let requestId = ""; + await expect + .poll( + async () => { + const response = await request.post("/v1/chat/completions", { + headers: auth, + data: { + model, + messages: [{ role: "user", content: `drawer ${marker}` }], + }, + }); + if (response.status() === 200) requestId = (await response.json()).id; + return response.status(); + }, + { + timeout: 30_000, + message: "the new deployment never served the request", + }, + ) + .toBe(200); + + const observed = await request.get(`${upstream}/__observations`); + const delivered = ( + (await observed.json()).requests as { + authorization: string; + body: unknown; + }[] + ).filter((entry) => JSON.stringify(entry.body).includes(marker)); + expect( + delivered.map((entry) => entry.authorization), + "Positive control: the upstream never received the deployment key", + ).toEqual([`Bearer ${deploymentKey}`]); + + await expect + .poll( + async () => { + const response = await request.get( + `/spend/logs/ui/${encodeURIComponent(requestId)}`, + { headers: auth }, + ); + return response.status() === 200 + ? JSON.stringify(await response.json()).includes(marker) + : false; + }, + { + timeout: 70_000, + message: `the stored request for ${requestId} never carried the marker`, + }, + ) + .toBe(true); + + await page.goto("/ui/login"); + await page.getByPlaceholder("Enter your username").fill("admin"); + await page.getByPlaceholder("Enter your password").fill(master); + await page.getByRole("button", { name: "Login", exact: true }).click(); + await expect(page).toHaveURL( + (url) => + url.pathname.startsWith("/ui") && !url.pathname.includes("login"), + ); + await navigateToPage(page, Page.Logs); + await dismissFeedbackPopup(page); + + const search = page + .getByTestId("datatable-search") + .filter({ visible: true }); + await expect(search).toBeVisible({ timeout: 20_000 }); + await search.fill(requestId); + const row = page + .locator("table") + .filter({ visible: true }) + .first() + .locator("tbody tr") + .filter({ hasText: requestId }); + await expect(row).toHaveCount(1, { timeout: 30_000 }); + await row.click(); + + const drawer = page.getByRole("dialog").first(); + await expect(drawer.getByText("Request & Response")).toBeVisible({ + timeout: 20_000, + }); + await expect( + drawer.getByText(marker, { exact: false }).first(), + ).toBeVisible({ timeout: 20_000 }); + expect( + foundForms(await page.content(), forms), + "the drawer's pretty view holds the deployment api_key", + ).toEqual([]); + + await drawer.getByRole("tab", { name: "JSON", exact: true }).click(); + await drawer.getByRole("tab", { name: "Request", exact: true }).click(); + const requestJson = drawer + .getByRole("tabpanel") + .filter({ hasText: marker }) + .last(); + await expect(requestJson).toBeVisible({ timeout: 20_000 }); + expect( + foundForms(await page.content(), forms), + "the drawer's request JSON holds the deployment api_key", + ).toEqual([]); + } finally { + if (modelId) await post(request, "/model/delete", { id: modelId }); + if (promptsStored === null) { + await post(request, "/config/field/delete", { + config_type: "general_settings", + field_name: "store_prompts_in_spend_logs", + }); + } else { + await post(request, "/config/update", { + general_settings: { store_prompts_in_spend_logs: promptsStored }, + }); + } + } +}); diff --git a/tests/integration/security/_canary.py b/tests/integration/security/_canary.py index 2d97c6fde0b..e58bda0936d 100644 --- a/tests/integration/security/_canary.py +++ b/tests/integration/security/_canary.py @@ -83,6 +83,7 @@ SLOTS: Final = MappingProxyType( "A1": Slot("A1", "Virtual key raw value, set as a custom key through /key/generate", prefix="sk-"), "A2": Slot("A2", "Proxy master key from the LITELLM_MASTER_KEY environment variable", prefix="sk-"), "B1": Slot("B1", "Deployment api_key declared in the proxy config.yaml model_list"), + "G1d": Slot("G1d", "Logging sink credential read from the proxy environment (DD_API_KEY)"), "C1": Slot( "C1", "Team callback langfuse_secret_key (team callback API, config team settings, callback_settings)" ), diff --git a/tests/integration/security/test_datadog_sink.py b/tests/integration/security/test_datadog_sink.py new file mode 100644 index 00000000000..be8f866e260 --- /dev/null +++ b/tests/integration/security/test_datadog_sink.py @@ -0,0 +1,169 @@ +"""Slot G1d through a Datadog intake double: the sink key reaches only its own auth header. + +The owned proxy enables the ``datadog`` callback with ``DD_API_KEY`` set to a fresh G1d canary +and ``DD_BASE_URL`` pointed at a local intake double. Datadog batches are gzip-compressed JSON +(a single event sent on the sync path is plain JSON), so the double inflates ``Content-Encoding: +gzip`` bodies, requires JSON log events, answers 202 like the real intake, and records the bytes +exactly as received for S4 (``find_canary`` inflates them). Events the route sweep itself +produces are swept again after it. + +Positive control: the intake double must receive ``DD-API-KEY: `` on the batch +carrying the scenario's marker, and the provider double ``Authorization: Bearer ``. +Sensitivity control: the marker must be found inside the gzip body (encoding ``gzip``), in the +stored spend row, on the Logs drawer route and in the generic sink. Then S1 to S5 plus the +intake double may not hold B1 or G1d anywhere, except G1d in the intake's own ``dd-api-key`` and +on the proxy admin's callback settings route (``ADMIN_ONLY_ALLOWANCES``). That route's gate for +everyone else is asserted directly: the internal user gets 401, and a ``proxy_admin_viewer`` +must read ``DD_API_KEY`` as ``REDACTED``. Routes are swept as the admin, the internal user and +that admin viewer. +""" + +from __future__ import annotations + +import gzip +import json +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from typing import Final + +import pytest +from integration._support.client import eventually, string_value +from integration._support.wire import Reply, Request, wire_server +from integration.security._canary import MARKER, Canary, canary +from integration.security._sinks import CONFIG_MODEL, GENERIC_SINK, Recorder, Rig, canary_rig, settle, team_caller +from integration.security._sweeps import ( + assert_marker_seen, + assert_no_hits, + record_route_sweep, + sweep_all, + sweep_sink, +) + +DATADOG_SINK: Final = "datadog" +DATADOG_KEY_HEADER: Final = "dd-api-key" +CALLBACK_SETTINGS_ROUTE: Final = "/get/config/callbacks" + + +def inflated(request: Request) -> bytes: + """The body as Datadog reads it: batches are gzip-compressed, single sync events are not.""" + return gzip.decompress(request.body) if request.headers.get("content-encoding") == "gzip" else request.body + + +def datadog_intake(request: Request) -> Reply: + assert request.target == "/api/v2/logs", request.target + events: Final = json.loads(inflated(request)) + assert isinstance(events, (list, dict)) and events, events + return Reply(status=202, body=b"{}") + + +def enable_datadog(config: dict[str, object], _provider_url: str) -> None: + settings: Final = config["litellm_settings"] + assert isinstance(settings, dict) + settings["callbacks"] = [*settings["callbacks"], DATADOG_SINK] + + +@pytest.fixture +def intake() -> Iterator[Recorder]: + with wire_server(datadog_intake) as wire: + yield Recorder(wire) + + +@pytest.fixture +def g1() -> Canary: + return canary("G1d") + + +@pytest.fixture +def rig(tmp_path: Path, intake: Recorder, g1: Canary) -> Iterator[Rig]: + environment: Final = {"DD_API_KEY": g1.value, "DD_SITE": "datadog.invalid", "DD_BASE_URL": intake.url} + with canary_rig(tmp_path, configure=enable_datadog, environment=environment) as value: + yield value + + +def carrying_inflated(intake: Recorder, marker: Canary) -> tuple[Request, ...]: + """Gzip batches whose inflated body holds ``marker``.""" + return tuple( + request + for request in intake.requests() + if request.headers.get("content-encoding") == "gzip" and marker.core.encode() in inflated(request) + ) + + +@pytest.mark.timeout(240) # full S1/S2 walk: every table and ~400 GET routes as three callers +def test_datadog_api_key_reaches_only_its_own_header( + rig: Rig, intake: Recorder, g1: Canary, request: pytest.FixtureRequest +) -> None: + b1: Final = rig.canaries["B1"] + marker: Final = canary(MARKER) + started: Final = datetime.now(UTC) + with rig.proxy.scenario() as scenario: + caller: Final = team_caller(scenario) + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": CONFIG_MODEL, "messages": [{"role": "user", "content": f"slot G1d {marker.value}"}]}, + key=caller.key, + ) + assert response.status_code == 200, response.text + assert [request.headers.get("authorization") for request in rig.provider.carrying(marker.value)] == [ + f"Bearer {b1.value}" + ], "Positive control: the provider double never received the B1 canary" + request_id: Final = string_value(response.json()["id"]) + settle(rig, request_id, marker) + batches: Final = eventually(lambda: carrying_inflated(intake, marker), bool, seconds=30) + assert {batch.headers.get(DATADOG_KEY_HEADER) for batch in batches} == {g1.value}, ( + "Positive control: the Datadog intake double never received the G1d canary" + ) + assert all(marker.core.encode() not in batch.body for batch in batches), "Datadog body was not compressed" + + denied: Final = rig.proxy.request("GET", CALLBACK_SETTINGS_ROUTE, key=caller.key) + assert denied.status_code == 401, f"internal_user read the callback settings: {denied.text}" + viewer: Final = scenario.key(user_id=scenario.user(user_role="proxy_admin_viewer")) + settings: Final = rig.proxy.request("GET", CALLBACK_SETTINGS_ROUTE, key=viewer) + assert settings.status_code == 200, settings.text + datadog_variables: Final = [ + entry["variables"] for entry in settings.json()["callbacks"] if entry["name"] == DATADOG_SINK + ] + assert datadog_variables and all(variables["DD_API_KEY"] == "REDACTED" for variables in datadog_variables), ( + f"The admin viewer's callback settings did not redact DD_API_KEY: {datadog_variables}" + ) + + swept: Final = intake.requests() + report: Final = sweep_all( + rig.proxy, + (marker, b1, g1), + responses=(response,), + sinks={**{name: sink.requests() for name, sink in rig.sinks.items()}, DATADOG_SINK: swept}, + ids={ + "request_id": request_id, + "team_id": caller.team_id, + "user_id": caller.user_id, + "model_id": rig.model_id, + "model": CONFIG_MODEL, + }, + callers={**caller.callers(rig), "admin_viewer": viewer}, + own_headers={**rig.own_headers, DATADOG_SINK: (DATADOG_KEY_HEADER, "G1d")}, + since=started, + ) + record_route_sweep(report.routes, request.node.nodeid) + assert_marker_seen( + report, + { + "S1": "LiteLLM_SpendLogs.proxy_server_request", + "S2": f"GET /spend/logs/ui/{request_id} as admin -> 200", + "S4": f"{GENERIC_SINK}[", + }, + ) + assert_marker_seen(report, {"S2": f"GET /spend/logs?request_id={request_id} as admin -> 200"}) + assert any( + hit.slot == MARKER and hit.location.startswith(f"{DATADOG_SINK}[") and hit.encoding == "gzip" + for hit in report.hits + ), f"Sensitivity control: S4 never inflated the marker out of the Datadog body: {report.marker_locations()}" + late: Final = sweep_sink( + f"{DATADOG_SINK} after the route sweep", + intake.requests()[len(swept) :], + (b1, g1), + own_header=(DATADOG_KEY_HEADER, "G1d"), + ) + assert_no_hits((*report.credential_hits(), *late), "slots B1 and G1d, Datadog intake") diff --git a/tests/integration/security/test_metrics_text.py b/tests/integration/security/test_metrics_text.py new file mode 100644 index 00000000000..e5a0fdc339e --- /dev/null +++ b/tests/integration/security/test_metrics_text.py @@ -0,0 +1,89 @@ +"""S7: the Prometheus ``/metrics/`` text never carries a credential canary. + +Metric label values come from request fields (caller, model, route, user agent, exception +class), so a credential copied into one of them would be served to every scraper. The owned +proxy enables the ``prometheus`` callback, sends one successful and one provider-rejected chat +completion, and searches the whole scrape. + +Positive control: the provider double must receive ``Authorization: Bearer `` for +both requests (their content carries the fresh marker, so neither is served from the response +cache). Sensitivity control: both requests send the marker as their ``User-Agent``, +which the proxy exports as the ``user_agent`` label, so the scrape must carry the marker on +the success and the failure series before the credential search counts. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from typing import Final + +import pytest +from integration._support.client import eventually +from integration.security._canary import MARKER, Canary, canary, find_canary +from integration.security._sinks import CONFIG_MODEL, PROVIDER_4XX, Rig, canary_rig, team_caller +from integration.security._sweeps import Hit, assert_no_hits + +METRICS_ROUTE: Final = "/metrics/" + + +def enable_prometheus(config: dict[str, object], _provider_url: str) -> None: + settings: Final = config["litellm_settings"] + assert isinstance(settings, dict) + settings["callbacks"] = [*settings["callbacks"], "prometheus"] + + +def sweep_metrics(text: str, canaries: tuple[Canary, ...]) -> tuple[Hit, ...]: + """Every canary in the scrape, attributed to the series line that holds it.""" + if not find_canary(text, canaries): + return () + return tuple( + Hit("S7", f"GET {METRICS_ROUTE} line {number}: {line[:160]!r}", match.slot, match.encoding) + for number, line in enumerate(text.splitlines(), start=1) + for match in find_canary(line, canaries) + ) + + +@pytest.fixture +def rig(tmp_path: Path) -> Iterator[Rig]: + with canary_rig(tmp_path, configure=enable_prometheus) as value: + yield value + + +def test_metrics_text_carries_no_credential(rig: Rig) -> None: + b1: Final = rig.canaries["B1"] + marker: Final = canary(MARKER) + agent: Final = f"canary-agent/{marker.value}" + with rig.proxy.scenario() as scenario: + caller: Final = team_caller(scenario) + responses: Final = tuple( + rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": CONFIG_MODEL, "messages": [{"role": "user", "content": text}]}, + key=caller.key, + headers={"User-Agent": agent}, + ) + for text in (f"slot B1 metrics {marker.value}", f"slot B1 metrics {marker.value} {PROVIDER_4XX}") + ) + assert [response.status_code for response in responses] == [200, 400], [r.text for r in responses] + delivered: Final = rig.provider.carrying(marker.value) + assert [request.headers.get("authorization") for request in delivered] == [f"Bearer {b1.value}"] * 2, ( + "Positive control: the provider double never received the B1 canary" + ) + + def scrape() -> str: + response: Final = rig.proxy.request("GET", METRICS_ROUTE) + assert response.status_code == 200, response.text + return response.text + + def both_outcomes_exported(text: str) -> bool: + lines: Final = text.splitlines() + return all( + any(marker.core in line and f'status_code="{status}"' in line for line in lines) + for status in ("200", "400") + ) + + hits: Final = sweep_metrics(eventually(scrape, both_outcomes_exported, seconds=30), (marker, b1)) + assert any(hit.slot == MARKER for hit in hits), "Sensitivity control: the scrape never carried the marker" + assert_no_hits(tuple(hit for hit in hits if hit.slot != MARKER), "slot B1, metrics text") diff --git a/tests/integration/security/test_proxy_logs.py b/tests/integration/security/test_proxy_logs.py new file mode 100644 index 00000000000..514bd69bb38 --- /dev/null +++ b/tests/integration/security/test_proxy_logs.py @@ -0,0 +1,84 @@ +"""S6: the owned proxy's own stdout and stderr never carry a credential canary. + +Each leg boots its own proxy (slot B1 lives in its config), sends one successful and one +provider-rejected chat completion, stops the proxy so every buffered write reaches the log +file, and then searches the whole captured log. The ``default`` leg runs with ``LITELLM_LOG`` +unset, the level an operator gets out of the box; the ``debug`` leg runs with +``LITELLM_LOG=DEBUG``, which prints request data, router decisions and provider calls. + +Positive control: the provider double must receive ``Authorization: Bearer `` for +both requests. Sensitivity control: the provider double echoes the rejected message in its +error text, and the proxy logs that error at every level, so the marker must be found in the +log; a capture that misses the log file or reads it before the writes land fails there. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import pytest +from integration._support.client import string_value +from integration._support.wire import Reply, Request +from integration.security._canary import MARKER, Canary, canary, find_canary +from integration.security._sinks import CONFIG_MODEL, PROVIDER_4XX, canary_rig, chat_upstream, settle, team_caller +from integration.security._sweeps import Hit, assert_no_hits + +LEGS: Final = MappingProxyType({"default": MappingProxyType({}), "debug": MappingProxyType({"LITELLM_LOG": "DEBUG"})}) + + +def echoing_upstream(request: Request) -> Reply: + """``chat_upstream``, except a rejection repeats the rejected message in its error text.""" + body: Final = json.loads(request.body or b"{}") + text: Final = str((body.get("messages") or [{}])[-1].get("content", "")) + if PROVIDER_4XX not in text: + return chat_upstream(request) + return Reply( + status=400, + body=json.dumps( + {"error": {"type": "invalid_request_error", "code": "canary_rejected", "message": f"rejected: {text}"}} + ).encode(), + ) + + +def sweep_log(path: Path, canaries: tuple[Canary, ...]) -> tuple[Hit, ...]: + """Every canary in the captured log, attributed to the line that holds it.""" + data: Final = path.read_bytes() + if not find_canary(data, canaries): + return () + return tuple( + Hit("S6", f"{path.name} line {number}: {line[:160]!r}", match.slot, match.encoding) + for number, line in enumerate(data.splitlines(), start=1) + for match in find_canary(line, canaries) + ) + + +@pytest.mark.parametrize("leg", tuple(LEGS)) +def test_proxy_log_carries_no_credential(leg: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("LITELLM_LOG", raising=False) + marker: Final = canary(MARKER) + with canary_rig(tmp_path, environment=LEGS[leg], upstream=echoing_upstream) as rig: + b1: Final = rig.canaries["B1"] + with rig.proxy.scenario() as scenario: + caller: Final = team_caller(scenario) + responses: Final = tuple( + rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": CONFIG_MODEL, "messages": [{"role": "user", "content": f"slot B1 {suffix}"}]}, + key=caller.key, + ) + for suffix in (marker.value, f"{marker.value} {PROVIDER_4XX}") + ) + assert [response.status_code for response in responses] == [200, 400], [r.text for r in responses] + delivered: Final = rig.provider.carrying(marker.value) + assert [request.headers.get("authorization") for request in delivered] == [f"Bearer {b1.value}"] * 2, ( + "Positive control: the provider double never received the B1 canary" + ) + settle(rig, string_value(responses[0].json()["id"]), marker) + log: Final = rig.owned.log + hits: Final = sweep_log(log, (marker, b1)) + assert any(hit.slot == MARKER for hit in hits), f"Sensitivity control: the marker never reached {log}" + assert_no_hits(tuple(hit for hit in hits if hit.slot != MARKER), f"slot B1, proxy log, {leg} level")