From 0217002ea19d30bbb24e861daebd872f043ba84e Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 23 Aug 2026 16:46:00 +0200 Subject: [PATCH] Gate proxy forwarding behind a segment-anchored API path allowlist --- routstr/proxy.py | 60 +++++++++-- .../integration/test_proxy_post_endpoints.py | 54 ++++++++++ tests/unit/test_proxy_path_allowlist.py | 99 +++++++++++++++++++ .../test_proxy_tinfoil_attestation_routing.py | 21 +++- 4 files changed, 224 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_proxy_path_allowlist.py diff --git a/routstr/proxy.py b/routstr/proxy.py index 4542d008..c367d9a2 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -236,6 +236,52 @@ _API_PATH_PREFIXES = ( "attestation", ) +# Split the allowlist by spelling so bare tokens anchor to a path segment. +# A slash-terminated prefix ("v1/") is already segment-anchored under +# startswith. A bare token ("models") must match a whole segment — exactly or +# followed by "/" — so "modelsdump" / "attestationadmin" cannot slip through. +_API_SLASH_PREFIXES = tuple(p for p in _API_PATH_PREFIXES if p.endswith("/")) +_API_BARE_PREFIXES = tuple(p for p in _API_PATH_PREFIXES if not p.endswith("/")) + + +def _is_ambiguously_spelled_path(path: str) -> bool: + """Reject paths whose spelling could resolve somewhere the allowlist did not. + + ``{path:path}`` arrives percent-decoded, so a client that sent ``%2e%2e`` or + ``%2f`` shows up here as ``..`` / ``/``. Dot segments, backslashes, duplicate + or leading separators, NUL bytes, and any residual encoded separator are + treated as unsafe: they let a caller walk off the canonical API surface (and + onto a sensitive upstream endpoint) even though the literal prefix check + would pass. Reject rather than trying to rewrite the path. + """ + if not path or path != path.strip() or path.startswith("/"): + return True + if "\x00" in path or "\\" in path: + return True + # A single trailing slash is canonical (e.g. "attestation/"); ignore it, + # then no remaining segment may be empty (covers "//") or a dot segment. + core = path[:-1] if path.endswith("/") else path + if any(segment in ("", ".", "..") for segment in core.split("/")): + return True + lowered = path.lower() + return "%2e" in lowered or "%2f" in lowered or "%5c" in lowered + + +def _forwarding_allowed(path: str, is_ehbp: bool) -> bool: + """Gate which paths may reach an upstream at all. + + The provider credential is attached during forwarding, so an unknown path + must never be forwarded on the caller's say-so. A path must resolve to a + known API prefix; EHBP requests are identified by header and carry their own + encrypted contract. Endpoint permission is derived from this allowlist, not + from the client-supplied path. + """ + if is_ehbp: + return True + if path.startswith(_API_SLASH_PREFIXES): + return True + return any(path == p or path.startswith(p + "/") for p in _API_BARE_PREFIXES) + @proxy_router.api_route("/{path:path}", methods=["GET", "POST"], response_model=None) async def proxy( @@ -255,14 +301,17 @@ async def proxy( async def _proxy( request: Request, path: str, session: AsyncSession ) -> Response | StreamingResponse: - # GET requests must hit a known API prefix; otherwise return a 404 (HTML - # for browsers, JSON for API clients). POST requests are always forwarded - # so that OpenAI-style endpoints work with or without the `v1/` prefix - # (e.g. `/chat/completions` as well as `/v1/chat/completions`). - if request.method == "GET" and not path.startswith(_API_PATH_PREFIXES): + # Screen the path before any routing decision: reject ambiguous spellings, + # then require a known API prefix so nothing unknown is forwarded with the + # provider credential attached. + if _is_ambiguously_spelled_path(path): return build_not_found_response(request, path) headers = dict(request.headers) + is_ehbp = "ehbp-encapsulated-key" in headers + + if not _forwarding_allowed(path, is_ehbp): + return build_not_found_response(request, path) is_responses_api = path.startswith("v1/responses") or path.startswith("responses") request_body = await request.body() @@ -272,7 +321,6 @@ async def _proxy( # extract the model id, so the SDK sends it in X-Routstr-Model. Forward the # raw encrypted body to the upstream's /private/ endpoint and stream the # encrypted response back untouched — the SDK's SecureClient decrypts it. - is_ehbp = "ehbp-encapsulated-key" in headers if is_ehbp: request_body_dict = {} model_id = headers.get("x-routstr-model", "") diff --git a/tests/integration/test_proxy_post_endpoints.py b/tests/integration/test_proxy_post_endpoints.py index 8d5fab36..9765ffd4 100644 --- a/tests/integration/test_proxy_post_endpoints.py +++ b/tests/integration/test_proxy_post_endpoints.py @@ -289,8 +289,62 @@ async def test_proxy_post_unauthorized_access(integration_client: AsyncClient) - assert response.status_code in [400, 401] +@pytest.mark.integration +@pytest.mark.asyncio +@pytest.mark.parametrize( + "bad_path", + [ + "internal/admin", # unknown endpoint, no API prefix + "v1/../admin", # traversal onto a sibling path + "v1//models", # duplicate separator + "%2e%2e/secret", # encoded dot segment + ], +) +async def test_authenticated_post_to_unknown_path_is_rejected( + authenticated_client: AsyncClient, bad_path: str +) -> None: + """An authenticated POST to an unknown/traversal path must be rejected at + the edge (404) and never forwarded — the provider credential must not reach + an endpoint the caller merely spelled into the URL. If the guard let it + through, forwarding would raise and this would not be a clean 404.""" + with patch( + "routstr.upstream.base.BaseUpstreamProvider.forward_request", + AsyncMock(side_effect=AssertionError("must not forward unknown path")), + ): + response = await authenticated_client.post( + f"/{bad_path}", + json={"model": "gpt-3.5-turbo", "messages": []}, + ) + assert response.status_code == 404 +@pytest.mark.integration +@pytest.mark.asyncio +async def test_get_unknown_path_is_rejected(integration_client: AsyncClient) -> None: + """A GET to an unknown (no API prefix) path is rejected at the edge.""" + response = await integration_client.get("/internal/admin") + assert response.status_code == 404 + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_ambiguous_path_is_rejected_before_ehbp_exemption( + integration_client: AsyncClient, +) -> None: + """The ambiguous-spelling screen runs before the EHBP header exemption, so + an EHBP request cannot smuggle a traversal path past it.""" + # Percent-encoded so the traversal survives the client to the server, which + # decodes it to "v1/../admin" before routing. + response = await integration_client.post( + "/v1/%2e%2e/admin", + content=b"encrypted", + headers={ + "ehbp-encapsulated-key": "x", + "x-routstr-model": "gpt-4", + }, + ) + assert response.status_code == 404 + @pytest.mark.integration @pytest.mark.asyncio diff --git a/tests/unit/test_proxy_path_allowlist.py b/tests/unit/test_proxy_path_allowlist.py new file mode 100644 index 00000000..a98679d5 --- /dev/null +++ b/tests/unit/test_proxy_path_allowlist.py @@ -0,0 +1,99 @@ +"""Unit tests for the proxy edge path allowlist (arbitrary-upstream-path-proxy). + +An authenticated POST used to be forwarded for ANY path, so a caller could reach +arbitrary or traversal-shaped upstream endpoints with the provider credential +attached. The proxy now rejects ambiguous path spellings for every method and +requires a known API prefix before anything is forwarded. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("UPSTREAM_BASE_URL", "http://test") +os.environ.setdefault("UPSTREAM_API_KEY", "test") + +import pytest # noqa: E402 + +from routstr.proxy import ( # noqa: E402 + _forwarding_allowed, + _is_ambiguously_spelled_path, +) + + +@pytest.mark.parametrize( + "path", + [ + "../secret", + "v1/../admin", + "v1/./models", + "..", + "v1//models", # duplicate separator + "/v1/models", # leading slash / absolute override + "v1/models/..", + "%2e%2e/secret", # residual encoded dot segment + "v1/%2fadmin", # residual encoded slash + "v1\\models", # backslash + "v1/models\x00", # NUL byte + " v1/models", # leading whitespace + "", + ], +) +def test_ambiguous_paths_are_rejected(path: str) -> None: + assert _is_ambiguously_spelled_path(path) is True + + +@pytest.mark.parametrize( + "path", + [ + "v1/chat/completions", + "chat/completions", + "v1/responses", + "v1/embeddings", + "models", + "v1/models/gpt-4", + "attestation/", # a single trailing slash is canonical + "tee/attestation/", + ], +) +def test_canonical_paths_are_allowed(path: str) -> None: + assert _is_ambiguously_spelled_path(path) is False + + +def test_unknown_paths_are_not_forwarded() -> None: + # The credential is attached during forwarding, so an unknown endpoint must + # never be forwarded on the caller's say-so. + assert _forwarding_allowed("internal/admin", is_ehbp=False) is False + assert _forwarding_allowed("secret-endpoint", is_ehbp=False) is False + + +@pytest.mark.parametrize( + "path", + [ + "modelsdump", # bare token "models" must not match a longer segment + "attestationadmin", + "providers-secret", + "embeddingsx", + "completions-internal", + ], +) +def test_bare_prefix_does_not_match_a_longer_segment(path: str) -> None: + assert _forwarding_allowed(path, is_ehbp=False) is False + + +def test_known_prefixes_are_forwarded() -> None: + assert _forwarding_allowed("v1/chat/completions", is_ehbp=False) is True + assert _forwarding_allowed("chat/completions", is_ehbp=False) is True + # Bare tokens match a whole segment: exactly or followed by "/". + assert _forwarding_allowed("models", is_ehbp=False) is True + assert _forwarding_allowed("models/gpt-4", is_ehbp=False) is True + assert _forwarding_allowed("embeddings", is_ehbp=False) is True + assert _forwarding_allowed("attestation", is_ehbp=False) is True + + +def test_ehbp_bypasses_prefix_gate_by_header() -> None: + # Documents a deliberate exemption: EHBP is identified by header and carries + # its own encrypted contract, so the prefix gate does not apply. The + # ambiguous-spelling screen still runs on EHBP paths (see the ordering test + # in the integration suite). + assert _forwarding_allowed("anything/encrypted", is_ehbp=True) is True diff --git a/tests/unit/test_proxy_tinfoil_attestation_routing.py b/tests/unit/test_proxy_tinfoil_attestation_routing.py index b6ba2f88..cd335f3b 100644 --- a/tests/unit/test_proxy_tinfoil_attestation_routing.py +++ b/tests/unit/test_proxy_tinfoil_attestation_routing.py @@ -102,10 +102,22 @@ async def test_attestation_trailing_slash_routes_directly_to_tinfoil( tinfoil.forward_get_request.assert_awaited_once() -@pytest.mark.parametrize("path", ["attestation/foo", "attestationjunk"]) +@pytest.mark.parametrize( + ("path", "expected_status"), + [ + # Valid `attestation` segment but not the exact attestation route: + # fails model validation (empty body -> unknown model) before auth. + ("attestation/foo", 400), + # Not a known path segment at all: rejected at the edge before routing. + ("attestationjunk", 404), + ], +) @pytest.mark.asyncio async def test_non_attestation_prefix_does_not_bypass_authentication( - monkeypatch: pytest.MonkeyPatch, proxy_app: FastAPI, path: str + monkeypatch: pytest.MonkeyPatch, + proxy_app: FastAPI, + path: str, + expected_status: int, ) -> None: tinfoil = MagicMock() tinfoil.provider_type = "tinfoil" @@ -118,8 +130,9 @@ async def test_non_attestation_prefix_does_not_bypass_authentication( ) as client: response = await client.get(f"/{path}") - assert response.status_code == 400 - assert response.json()["error"]["type"] == "invalid_model" + assert response.status_code == expected_status + if expected_status == 400: + assert response.json()["error"]["type"] == "invalid_model" tinfoil.forward_get_request.assert_not_awaited()