Gate proxy forwarding behind a segment-anchored API path allowlist

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
+    # Screen the path before any routing decision: reject ambiguous spellings,
-    # for browsers, JSON for API clients). POST requests are always forwarded
+    # then require a known API prefix so nothing unknown is forwarded with the
-    # so that OpenAI-style endpoints work with or without the `v1/` prefix
+    # provider credential attached.
-    # (e.g. `/chat/completions` as well as `/v1/chat/completions`).
+    if _is_ambiguously_spelled_path(path):
-    if request.method == "GET" and not path.startswith(_API_PATH_PREFIXES):
         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", "")

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

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

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,7 +130,8 @@ 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.status_code == expected_status
+    if expected_status == 400:
         assert response.json()["error"]["type"] == "invalid_model"
     tinfoil.forward_get_request.assert_not_awaited()