From 7e3320a5c31e4e093e68fb21e0839c4c35860688 Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 31 Oct 2026 03:51:17 +0000 Subject: [PATCH] test(responses): cover comparison ID ownership and wire decoding Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../test_responses_id_ownership.py | 201 +++++++++++++++++- .../providers/test_responses_core_wire.py | 169 ++++++++++++++- tests/unit/test_responses_id_security.py | 56 +++++ 3 files changed, 422 insertions(+), 4 deletions(-) diff --git a/tests/integration/authorization/test_responses_id_ownership.py b/tests/integration/authorization/test_responses_id_ownership.py index 2ab26e75f50..e56651df9fa 100644 --- a/tests/integration/authorization/test_responses_id_ownership.py +++ b/tests/integration/authorization/test_responses_id_ownership.py @@ -3,7 +3,7 @@ from __future__ import annotations import json import os import uuid -from collections.abc import Callable +from collections.abc import Callable, Mapping from pathlib import Path from typing import Final, Literal from urllib.parse import urlsplit @@ -16,7 +16,7 @@ from integration._support.process import owned_proxy from integration._support.responses_vendor import same_response from integration._support.wire import Reply, Request, Wire, wire_server from integration._support.worker_readiness import wait_for_worker_readiness -from openai.types.responses import ResponseInputParam +from openai.types.responses import ResponseCompletedEvent, ResponseInputParam from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter pytestmark: Final = pytest.mark.shares_worker_state @@ -138,6 +138,203 @@ def _every_id_route(client: openai.OpenAI, model: str, response_id: str) -> tupl ) +def _comparison_body(response_id: str, nested: bool) -> Mapping[str, object]: + options: Final = {"comparison_response_id": response_id} + if nested: + return { + "prompt_cache_options": {"mode": "explicit", "ttl": "12h"}, + "extra_body": {"prompt_cache_options": options}, + } + return {"prompt_cache_options": options} + + +@pytest.mark.parametrize( + ("prefix", "nested", "stream"), + (("/v1", False, False), ("", True, False), ("/openai/v1", False, True), ("/v1", True, True)), +) +def test_regression_comparison_response_id_owner_repeated_requests_reach_raw_provider_id( + gateway: Gateway, prefix: str, nested: bool, stream: bool +) -> None: + raw_id: Final = f"resp_comparison_{uuid.uuid4().hex}" + + def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return _model_discovery_reply() + assert (request.method, request.target) == ("POST", "/v1/responses") + body: Final = _JSON_OBJECT.validate_json(request.body) + if body["input"] == "create comparison response": + assert body == {"model": _MODEL, "input": "create comparison response"}, request.body + return Reply(body=_response(raw_id)) + assert body == { + "model": _MODEL, + "input": "compare owned response", + "stream": stream, + "prompt_cache_options": {"comparison_response_id": raw_id}, + }, request.body + if stream: + return Reply( + content_type="text/event-stream", + chunks=( + b"data: " + + json.dumps({"type": "response.completed", "response": json.loads(_response(raw_id))}).encode() + + b"\n\n", + b"data: [DONE]\n\n", + ), + ) + return Reply(body=_response(raw_id)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, wire) + team: Final = scenario.team(models=[model]) + user: Final = scenario.member(team) + with _client(gateway, _key(scenario, team, user, model), prefix) as client: + created: Final = client.responses.create(model=model, input="create comparison response") + assert created.id != raw_id, created.model_dump_json() + assert len(_contract_requests(wire)) == 1 + + def compare_again() -> None: + result: Final = client.responses.create( + model=model, + input="compare owned response", + stream=stream, + extra_body=_comparison_body(created.id, nested), + ) + if isinstance(result, openai.Stream): + events: Final = tuple(result) + assert isinstance(events[-1], ResponseCompletedEvent), events + assert events[-1].response.status == "completed", events + else: + assert result.status == "completed", result.model_dump_json() + + for _ in range(2): + compare_again() + assert len(_contract_requests(wire)) == 2 + + +@pytest.mark.parametrize("nested", (False, True)) +@pytest.mark.parametrize("stranger", ("user", "team")) +def test_regression_comparison_response_id_other_identity_is_denied_before_dispatch( + gateway: Gateway, nested: bool, stranger: str +) -> None: + raw_id: Final = f"resp_comparison_{uuid.uuid4().hex}" + + def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return _model_discovery_reply() + return Reply(body=_response(raw_id)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, wire) + team_a: Final = scenario.team(models=[model]) + team_b: Final = scenario.team(models=[model]) + user_a: Final = scenario.member(team_a) + user_b: Final = scenario.member(team_a) + gateway.post("/team/member_add", {"team_id": team_b, "member": {"role": "user", "user_id": user_a}}) + with ( + _client(gateway, _key(scenario, team_a, user_a, model), "/v1") as owner, + _client( + gateway, + _key( + scenario, team_a if stranger == "user" else team_b, user_b if stranger == "user" else user_a, model + ), + "/v1", + ) as other, + ): + created: Final = owner.responses.create(model=model, input="create private comparison") + assert created.id != raw_id + assert len(_contract_requests(wire)) == 1 + _refused( + lambda: other.responses.create( + model=model, input="compare another identity", extra_body=_comparison_body(created.id, nested) + ), + _OTHER_USER_DETAIL if stranger == "user" else _OTHER_TEAM_DETAIL, + ) + assert _contract_requests(wire) == (), "A cross-identity comparison reached the provider" + + +@pytest.mark.parametrize("nested", (False, True)) +@pytest.mark.parametrize("comparison_id", ("resp_unmanaged_comparison", "resp_!!!not-base64", "cmF3X3ZlbmRvcl9pZA==")) +def test_guard_unmanaged_comparison_response_ids_pass_through_without_opt_in( + gateway: Gateway, nested: bool, comparison_id: str +) -> None: + def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return _model_discovery_reply() + assert (request.method, request.target) == ("POST", "/v1/responses") + assert _JSON_OBJECT.validate_json(request.body) == { + "model": _MODEL, + "input": "unmanaged comparison", + "prompt_cache_options": {"comparison_response_id": comparison_id}, + }, request.body + return Reply(body=_response(f"resp_result_{uuid.uuid4().hex}")) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, wire) + team: Final = scenario.team(models=[model]) + user: Final = scenario.member(team) + with _client(gateway, _key(scenario, team, user, model), "/v1") as client: + compared: Final = client.responses.create( + model=model, input="unmanaged comparison", extra_body=_comparison_body(comparison_id, nested) + ) + assert compared.status == "completed", compared.model_dump_json() + assert len(_contract_requests(wire)) == 1 + + +@pytest.mark.parametrize("effective_owner", (False, True), ids=("guard-raw", "regression-owned")) +def test_comparison_extra_body_ignores_foreign_top_level(gateway: Gateway, effective_owner: bool) -> None: + foreign_raw_id: Final = f"resp_ignored_{uuid.uuid4().hex}" + owner_raw_id: Final = f"resp_effective_{uuid.uuid4().hex}" + unmanaged_id: Final = "resp_raw_effective_comparison" + + def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return _model_discovery_reply() + assert (request.method, request.target) == ("POST", "/v1/responses") + body: Final = _JSON_OBJECT.validate_json(request.body) + if body["input"] in ("seed ignored comparison", "seed effective comparison"): + assert body == {"model": _MODEL, "input": body["input"]}, request.body + return Reply(body=_response(foreign_raw_id if body["input"] == "seed ignored comparison" else owner_raw_id)) + assert body == { + "model": _MODEL, + "input": "compare effective options", + "prompt_cache_options": {"comparison_response_id": owner_raw_id if effective_owner else unmanaged_id}, + }, request.body + return Reply(body=_response(owner_raw_id)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, wire) + team: Final = scenario.team(models=[model]) + owner_user: Final = scenario.member(team) + foreign_user: Final = scenario.member(team) + with ( + _client(gateway, _key(scenario, team, owner_user, model), "/v1") as owner, + _client(gateway, _key(scenario, team, foreign_user, model), "/v1") as foreign, + ): + foreign_response: Final = foreign.responses.create(model=model, input="seed ignored comparison") + assert foreign_response.id != foreign_raw_id + owner_response: Final = owner.responses.create(model=model, input="seed effective comparison") + assert owner_response.id != owner_raw_id + assert len(_contract_requests(wire)) == 2 + compared: Final = owner.responses.create( + model=model, + input="compare effective options", + extra_body={ + "prompt_cache_options": { + "comparison_response_id": foreign_response.id, + "mode": "explicit", + "ttl": "12h", + }, + "extra_body": { + "prompt_cache_options": { + "comparison_response_id": owner_response.id if effective_owner else unmanaged_id + } + }, + }, + ) + assert compared.status == "completed", compared.model_dump_json() + assert len(_contract_requests(wire)) == 1 + + @pytest.mark.parametrize("prefix", _ALIASES) def test_response_ids_are_scoped_to_the_issuing_user_and_team(gateway: Gateway, prefix: str) -> None: raw_id: Final = f"resp_owned_{uuid.uuid4().hex}" diff --git a/tests/integration/providers/test_responses_core_wire.py b/tests/integration/providers/test_responses_core_wire.py index 9a24903a20d..9166c21893c 100644 --- a/tests/integration/providers/test_responses_core_wire.py +++ b/tests/integration/providers/test_responses_core_wire.py @@ -3,8 +3,9 @@ from __future__ import annotations import json import os import uuid -from collections.abc import Callable -from typing import Final +from collections.abc import AsyncIterator, Callable, Iterator +from copy import deepcopy +from typing import Final, Protocol, runtime_checkable import openai import pytest @@ -14,6 +15,9 @@ from integration._support.worker_readiness import wait_for_worker_readiness from openai.types.responses import ResponseCompletedEvent from pydantic import JsonValue, TypeAdapter +import litellm +from litellm.types.llms.openai import ResponsesAPIResponse + _MODEL: Final = "gpt-5" _API_KEY: Final = "synthetic-responses-key" _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) @@ -21,6 +25,16 @@ _PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", " _ALIASES: Final = ("/v1", "", "/openai/v1") +@runtime_checkable +class _AsyncResponseEvents(Protocol): + def __aiter__(self) -> AsyncIterator[object]: ... + + +@runtime_checkable +class _ResponseEvents(Protocol): + def __iter__(self) -> Iterator[object]: ... + + def _deployments_in_group(gateway: Gateway, group: str) -> int: entries: Final = gateway.get("/model/info")["data"] assert isinstance(entries, list), entries @@ -69,6 +83,157 @@ def _stream(identity: str) -> tuple[bytes, ...]: ) + (b"data: [DONE]\n\n",) +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async")) +@pytest.mark.parametrize("stream", (False, True), ids=("nonstream", "stream")) +@pytest.mark.parametrize("extra", (False, True), ids=("top-level", "extra-body-precedence")) +async def test_regression_sdk_comparison_response_id_is_raw_on_the_wire( + asynchronous: bool, stream: bool, extra: bool +) -> None: + raw_id: Final = f"resp_sdk_comparison_{uuid.uuid4().hex}" + + def respond(request: Request) -> Reply: + assert (request.method, request.target) == ("POST", "/v1/responses") + body: Final = _JSON_OBJECT.validate_json(request.body) + if body["input"] == "seed comparison": + assert body == {"model": _MODEL, "input": "seed comparison"}, request.body + return Reply(body=_response(raw_id)) + assert body == { + "model": _MODEL, + "input": "compare response", + "stream": stream, + "previous_response_id": raw_id, + "prompt_cache_options": {"comparison_response_id": raw_id}, + }, request.body + return ( + Reply(content_type="text/event-stream", chunks=_stream(raw_id)) if stream else Reply(body=_response(raw_id)) + ) + + with wire_server(respond) as wire: + seeded: Final = litellm.responses( + model=f"openai/{_MODEL}", input="seed comparison", api_base=f"{wire.url}/v1", api_key=_API_KEY + ) + assert isinstance(seeded, ResponsesAPIResponse), seeded + assert seeded.id != raw_id, seeded + params: Final = { + "model": f"openai/{_MODEL}", + "input": "compare response", + "stream": stream, + "previous_response_id": seeded.id, + "api_base": f"{wire.url}/v1", + "api_key": _API_KEY, + "num_retries": 0, + **( + { + "prompt_cache_options": {"mode": "explicit", "ttl": "12h"}, + "extra_body": {"prompt_cache_options": {"comparison_response_id": seeded.id}}, + } + if extra + else {"prompt_cache_options": {"comparison_response_id": seeded.id}} + ), + } + original: Final = deepcopy(params) + result: Final = await litellm.aresponses(**params) if asynchronous else litellm.responses(**params) + if stream: + if asynchronous: + assert isinstance(result, _AsyncResponseEvents), result + async_events: Final = tuple([event async for event in result]) + async_completed: Final = ResponseCompletedEvent.model_validate(async_events[-1]) + assert async_completed.response.status == "completed", async_events + else: + assert isinstance(result, _ResponseEvents), result + sync_events: Final = tuple(result) + sync_completed: Final = ResponseCompletedEvent.model_validate(sync_events[-1]) + assert sync_completed.response.status == "completed", sync_events + else: + assert isinstance(result, ResponsesAPIResponse), result + assert result.status == "completed", result + assert params == original, "The caller's comparison options were mutated" + assert len(wire.drain()) == 2 + + +@pytest.mark.parametrize("extra", (False, True)) +@pytest.mark.parametrize("comparison_id", ("resp_custom_vendor_id", "resp_!!!not-base64", "cmF3X3ZlbmRvcl9pZA==")) +def test_guard_sdk_custom_host_preserves_unmanaged_comparison_ids(extra: bool, comparison_id: str) -> None: + def respond(request: Request) -> Reply: + assert (request.method, request.target) == ("POST", "/v1/responses") + assert _JSON_OBJECT.validate_json(request.body) == { + "model": _MODEL, + "input": "custom comparison", + "prompt_cache_options": {"comparison_response_id": comparison_id, "mode": "implicit", "ttl": "30m"}, + }, request.body + return Reply(body=_response("resp_custom_result")) + + with wire_server(respond) as wire: + options: Final = { + "comparison_response_id": comparison_id, + "mode": "implicit", + "ttl": "30m", + } + params: Final = ( + {"extra_body": {"prompt_cache_options": options}} if extra else {"prompt_cache_options": options} + ) + original: Final = deepcopy(params) + result: Final = litellm.responses( + model=f"openai/{_MODEL}", input="custom comparison", api_base=f"{wire.url}/v1", api_key=_API_KEY, **params + ) + assert isinstance(result, ResponsesAPIResponse), result + assert result.status == "completed", result + assert params == original + assert len(wire.drain()) == 1 + + +@pytest.mark.parametrize("provider", ("azure", "xai")) +@pytest.mark.parametrize("extra", (False, True)) +def test_guard_inherited_responses_providers_preserve_comparison_wrappers(provider: str, extra: bool) -> None: + raw_id: Final = f"resp_inherited_{uuid.uuid4().hex}" + + def seed(request: Request) -> Reply: + assert _JSON_OBJECT.validate_json(request.body) == {"model": _MODEL, "input": "seed inherited comparison"} + return Reply(body=_response(raw_id)) + + with wire_server(seed) as seed_wire: + seeded: Final = litellm.responses( + model=f"openai/{_MODEL}", + input="seed inherited comparison", + api_base=f"{seed_wire.url}/v1", + api_key=_API_KEY, + ) + assert isinstance(seeded, ResponsesAPIResponse), seeded + assert seeded.id != raw_id + assert len(seed_wire.drain()) == 1 + + def respond(request: Request) -> Reply: + assert request.method == "POST", request + expected_target: Final = ( + "/openai/responses?api-version=2025-03-01-preview" if provider == "azure" else "/v1/responses" + ) + assert request.target == expected_target, request.target + assert _JSON_OBJECT.validate_json(request.body) == { + "model": _MODEL, + "input": "inherited comparison", + "prompt_cache_options": {"comparison_response_id": seeded.id}, + }, request.body + return Reply(body=_response(raw_id)) + + with wire_server(respond) as wire: + options: Final = {"comparison_response_id": seeded.id} + params: Final = ( + {"extra_body": {"prompt_cache_options": options}} if extra else {"prompt_cache_options": options} + ) + result: Final = litellm.responses( + model=f"{provider}/{_MODEL}", + input="inherited comparison", + api_base=wire.url if provider == "azure" else f"{wire.url}/v1", + api_key=_API_KEY, + api_version="2025-03-01-preview" if provider == "azure" else None, + **params, + ) + assert isinstance(result, ResponsesAPIResponse), result + assert result.status == "completed", result + assert len(wire.drain()) == 1 + + def _register_in_group(scenario: Scenario, wire: Wire, group: str, order: int | None = None) -> str: existing: Final = _deployments_in_group(scenario.gateway, group) created: Final = scenario.gateway.post( diff --git a/tests/unit/test_responses_id_security.py b/tests/unit/test_responses_id_security.py index 7e74bfd0f17..b6b4665f286 100644 --- a/tests/unit/test_responses_id_security.py +++ b/tests/unit/test_responses_id_security.py @@ -4,11 +4,16 @@ Tests for ResponsesIDSecurity hook. Tests the security hook that prevents user B from seeing response from user A. """ +from collections.abc import Mapping +from typing import Final from unittest.mock import MagicMock, patch import pytest from fastapi import HTTPException +from pydantic import TypeAdapter +from litellm.caching.caching import DualCache +from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.hooks.responses_id_security import ( ResponsesIDSecurity, _streams_response_ids, @@ -854,6 +859,57 @@ def _issue_managed_id(hook, owner, provider_response_id=_FABRICATED_PROVIDER_RES return issued.id +def _comparison_options(data: object, nested: bool) -> Mapping[str, str]: + request: Final = TypeAdapter(dict[str, object]).validate_python(data) + container: Final = TypeAdapter(dict[str, object]).validate_python(request["extra_body"]) if nested else request + return TypeAdapter(dict[str, str]).validate_python(container["prompt_cache_options"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("nested", (False, True)) +@pytest.mark.parametrize("stranger", ("user", "team")) +async def test_regression_comparison_id_reentry_keeps_ownership_after_owner_retry( + salt_key_env: str, nested: bool, stranger: str +) -> None: + hook: Final = ResponsesIDSecurity(general_settings_reader=lambda: {}, signing_key_reader=lambda: salt_key_env) + owner: Final = UserAPIKeyAuth(user_id="owner-user", team_id="owner-team") + caller: Final = UserAPIKeyAuth( + user_id="stranger-user" if stranger == "user" else owner.user_id, + team_id="stranger-team" if stranger == "team" else owner.team_id, + ) + cache: Final = DualCache() + issued: Final = ResponsesAPIResponse.model_validate( + await hook.async_post_call_success_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary + data={}, + user_api_key_dict=owner, + response=ResponsesAPIResponse( + id=_FABRICATED_PROVIDER_RESPONSE_ID, created_at=1234567890, output=[], status="completed" + ), + ) + ) + assert issued.id != _FABRICATED_PROVIDER_RESPONSE_ID + options: Final = {"comparison_response_id": issued.id} + request: Final = {"extra_body": {"prompt_cache_options": options}} if nested else {"prompt_cache_options": options} + first: Final = TypeAdapter(dict[str, object]).validate_python( + await hook.async_pre_call_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary + user_api_key_dict=owner, cache=cache, data=request, call_type="aresponses" + ) + ) + retried: Final = TypeAdapter(dict[str, object]).validate_python( + await hook.async_pre_call_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary + user_api_key_dict=owner, cache=cache, data=request, call_type="aresponses" + ) + ) + with pytest.raises(HTTPException) as refusal: + await hook.async_pre_call_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary + user_api_key_dict=caller, cache=cache, data=request, call_type="aresponses" + ) + assert refusal.value.status_code == 403 + assert f"not associated with the {stranger}" in refusal.value.detail + assert _comparison_options(first, nested) == {"comparison_response_id": _FABRICATED_PROVIDER_RESPONSE_ID} + assert _comparison_options(retried, nested) == {"comparison_response_id": _FABRICATED_PROVIDER_RESPONSE_ID} + + class TestUnrecognizedResponseIdIsRejected: """An id this proxy never issued carries no owner, so it must not reach the provider."""