Restrict proxy forwarding to an exact method/path allowlist

routstr/core/settings.py

@@ -101,6 +101,12 @@ class Settings(BaseSettings):
 
     # Network
     cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")
+    # Comma-separated METHOD:path pairs adding to the proxy's canonical
+    # endpoint allowlist, e.g. "POST:v1/rerank,GET:batches". Only for upstreams
+    # exposing an endpoint outside the OpenAI-compatible set; each addition
+    # widens what the provider credential can be spent against, so wildcards
+    # and prefixes are not supported.
+    proxy_extra_allowed_paths: str = Field(default="", env="PROXY_EXTRA_ALLOWED_PATHS")
     tor_proxy_url: str = Field(default="socks5://127.0.0.1:9050", env="TOR_PROXY_URL")
     providers_refresh_interval_seconds: int = Field(
         default=0, env="PROVIDERS_REFRESH_INTERVAL_SECONDS"
@@ -192,6 +198,11 @@ SECRET_FIELDS = frozenset({"admin_password", "nsec"})
 # neither store nor shadow them; env is always authoritative.
 ENV_ONLY_FIELDS = frozenset(
     {
+        # Widening the proxy's reachable upstream surface is a deployment
+        # decision, not a runtime toggle: it changes what the provider
+        # credential can be spent against. Keeping it env-only also lets the
+        # proxy parse it once at import without going stale.
+        "proxy_extra_allowed_paths",
         "database_pool_size",
         "database_max_overflow",
         "database_pool_timeout",

routstr/proxy.py

@@ -26,6 +26,7 @@ from .core.db import (
 )
 from .core.exceptions import UpstreamError
 from .core.not_found import build_not_found_response
+from .core.settings import settings
 from .payment.helpers import (
     calculate_discounted_max_cost,
     check_token_balance,
@@ -221,27 +222,93 @@ async def refresh_model_maps_periodically() -> None:
             )
 
 
-_API_PATH_PREFIXES = (
+# Canonical endpoints this proxy will forward, keyed by the path with any
-    "v1/",
+# leading "v1/" and trailing slash removed, mapped to the methods allowed on
-    "responses",
+# each. The provider credential is attached during forwarding, so endpoint
-    "chat/",
+# permission has to come from this table rather than from the client-supplied
-    "completions",
+# path: an upstream's key-management, organization, or billing routes live
-    "models",
+# under the same origin and must never be reachable through the proxy.
-    "embeddings",
+_ALLOWED_ENDPOINTS: dict[str, frozenset[str]] = {
-    "audio/",
+    "chat/completions": frozenset({"POST"}),
-    "images/",
+    "completions": frozenset({"POST"}),
-    "moderations",
+    "responses": frozenset({"POST"}),
-    "providers",
+    "messages": frozenset({"POST"}),
-    "tee/",
+    "embeddings": frozenset({"POST"}),
-    "attestation",
+    "moderations": frozenset({"POST"}),
-)
+    "rerank": frozenset({"POST"}),
+    "audio/speech": frozenset({"POST"}),
+    "audio/transcriptions": frozenset({"POST"}),
+    "audio/translations": frozenset({"POST"}),
+    "images/generations": frozenset({"POST"}),
+    "images/edits": frozenset({"POST"}),
+    "images/variations": frozenset({"POST"}),
+    "models": frozenset({"GET"}),
+    "attestation": frozenset({"GET"}),
+    "tee/attestation": frozenset({"GET"}),
+}
 
-# Split the allowlist by spelling so bare tokens anchor to a path segment.
+_ALLOWED_METHODS = frozenset({"GET", "POST"})
-# 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.
+def _canonical_api_path(path: str) -> str:
-_API_SLASH_PREFIXES = tuple(p for p in _API_PATH_PREFIXES if p.endswith("/"))
+    """Reduce a request path to its allowlist key.
-_API_BARE_PREFIXES = tuple(p for p in _API_PATH_PREFIXES if not p.endswith("/"))
+
+    OpenAI-style clients reach the same endpoint with or without the ``v1/``
+    prefix and with or without a trailing slash, so both spellings collapse to
+    one key. Callers must screen the path with
+    :func:`_is_ambiguously_spelled_path` first — this function assumes the path
+    has no dot segments, empty segments, or encoded separators left to resolve.
+    """
+    core = path[:-1] if path.endswith("/") else path
+    if core.startswith("v1/"):
+        core = core[len("v1/") :]
+    return core
+
+
+def _parse_extra_allowed_endpoints(raw: str) -> dict[str, frozenset[str]]:
+    """Parse operator-configured additions to the endpoint allowlist.
+
+    Deployments whose provider exposes an endpoint outside the canonical set
+    opt in explicitly with ``PROXY_EXTRA_ALLOWED_PATHS``, a comma-separated
+    list of ``METHOD:path`` pairs (e.g. ``POST:v1/rerank,GET:batches``). Every
+    entry must name one concrete method and one unambiguous path; wildcards
+    and bare prefixes are deliberately unsupported, so widening the proxy's
+    reach is always a per-endpoint decision. Malformed entries are dropped
+    with a warning rather than silently widening or narrowing the surface.
+    """
+    extra: dict[str, frozenset[str]] = {}
+    for entry in raw.split(","):
+        entry = entry.strip()
+        if not entry:
+            continue
+        method, separator, endpoint = entry.partition(":")
+        method = method.strip().upper()
+        endpoint = endpoint.strip()
+        if not separator or method not in _ALLOWED_METHODS or not endpoint:
+            logger.warning(
+                "Ignoring malformed PROXY_EXTRA_ALLOWED_PATHS entry",
+                extra={"entry": entry},
+            )
+            continue
+        if _is_ambiguously_spelled_path(endpoint):
+            logger.warning(
+                "Ignoring ambiguously spelled PROXY_EXTRA_ALLOWED_PATHS entry",
+                extra={"entry": entry},
+            )
+            continue
+        if any(character in endpoint for character in "*?["):
+            # Refuse glob syntax outright. Kept as a literal endpoint name it
+            # would never match a real request, so the operator would think
+            # they had widened the proxy when they had not.
+            logger.warning(
+                "Ignoring wildcard PROXY_EXTRA_ALLOWED_PATHS entry; "
+                "list each endpoint explicitly",
+                extra={"entry": entry},
+            )
+            continue
+        key = _canonical_api_path(endpoint)
+        extra[key] = extra.get(key, frozenset()) | {method}
+    return extra
 
 
 def _is_ambiguously_spelled_path(path: str) -> bool:
@@ -267,20 +334,35 @@ def _is_ambiguously_spelled_path(path: str) -> bool:
     return "%2e" in lowered or "%2f" in lowered or "%5c" in lowered
 
 
-def _forwarding_allowed(path: str, is_ehbp: bool) -> bool:
+_EXTRA_ALLOWED_ENDPOINTS = _parse_extra_allowed_endpoints(
-    """Gate which paths may reach an upstream at all.
+    settings.proxy_extra_allowed_paths
+)
 
-    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
+def _allowed_methods_for(endpoint: str) -> frozenset[str]:
-    known API prefix; EHBP requests are identified by header and carry their own
+    """Return the methods allowed on a canonical endpoint, empty if unknown."""
-    encrypted contract. Endpoint permission is derived from this allowlist, not
+    methods = _ALLOWED_ENDPOINTS.get(endpoint, frozenset())
-    from the client-supplied path.
+    methods |= _EXTRA_ALLOWED_ENDPOINTS.get(endpoint, frozenset())
+    return methods
+
+
+def _forwarding_allowed(path: str, method: str) -> bool:
+    """Gate which method/path pairs may reach an upstream at all.
+
+    The provider credential is attached during forwarding, so an unknown
+    endpoint must never be forwarded on the caller's say-so. The path is
+    reduced to its canonical form and looked up in the endpoint table; there is
+    no prefix match, so a known prefix no longer carries an unknown endpoint
+    (``v1/organization/api_keys`` is rejected even though ``v1/`` is familiar).
+
+    EHBP requests are gated by the same table. Their body is opaque to the
+    proxy, which is a reason to constrain the destination more tightly, not to
+    trust the caller's path: the encrypted contract covers the body, never the
+    endpoint the credential is spent against.
     """
-    if is_ehbp:
+    if method not in _ALLOWED_METHODS:
-        return True
+        return False
-    if path.startswith(_API_SLASH_PREFIXES):
+    return method in _allowed_methods_for(_canonical_api_path(path))
-        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)
@@ -310,7 +392,7 @@ async def _proxy(
     headers = dict(request.headers)
     is_ehbp = "ehbp-encapsulated-key" in headers
 
-    if not _forwarding_allowed(path, is_ehbp):
+    if not _forwarding_allowed(path, request.method):
         return build_not_found_response(request, path)
 
     is_responses_api = path.startswith("v1/responses") or path.startswith("responses")

tests/integration/test_proxy_post_endpoints.py

@@ -318,6 +318,43 @@ async def test_authenticated_post_to_unknown_path_is_rejected(
     assert response.status_code == 404
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "bad_path",
+    [
+        # Unambiguously spelled, under a prefix the proxy serves, but not an
+        # endpoint it offers. These are real upstream routes that manage keys,
+        # org membership, and billing on the same origin as inference.
+        "v1/organization/api_keys",
+        "v1/api_keys",
+        "v1/billing/usage",
+        "v1/admin/keys",
+        # An id segment is honoured one level deep, and only where an endpoint
+        # takes one at all.
+        "v1/chat/completions/abc",
+        "v1/models/gpt-4/secret",
+    ],
+)
+async def test_known_prefix_does_not_carry_an_unknown_endpoint(
+    authenticated_client: AsyncClient, bad_path: str
+) -> None:
+    """A familiar prefix must not be a passport for the rest of the origin.
+
+    Nothing here is traversal-shaped, so the spelling screen lets it by; only
+    the endpoint allowlist stops it. Forwarding raises if the guard misses,
+    so a clean 404 also proves the credential never left."""
+    with patch(
+        "routstr.upstream.base.BaseUpstreamProvider.forward_request",
+        AsyncMock(side_effect=AssertionError("must not forward unknown endpoint")),
+    ):
+        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:
@@ -328,21 +365,37 @@ async def test_get_unknown_path_is_rejected(integration_client: AsyncClient) ->
 
 @pytest.mark.integration
 @pytest.mark.asyncio
-async def test_ambiguous_path_is_rejected_before_ehbp_exemption(
+@pytest.mark.parametrize(
-    integration_client: AsyncClient,
+    "bad_path",
+    [
+        # Percent-encoded so the traversal survives the client to the server,
+        # which decodes it to "v1/../admin" before routing.
+        "v1/%2e%2e/admin",
+        # Well-spelled, so only the endpoint allowlist can stop it.
+        "v1/organization/api_keys",
+        "anything/encrypted",
+    ],
+)
+async def test_ehbp_request_is_gated_by_the_endpoint_allowlist(
+    integration_client: AsyncClient, bad_path: str
 ) -> None:
-    """The ambiguous-spelling screen runs before the EHBP header exemption, so
+    """EHBP hides the body from the proxy, not the destination.
-    an EHBP request cannot smuggle a traversal path past it."""
+
-    # Percent-encoded so the traversal survives the client to the server, which
+    The encrypted contract covers the request body; it says nothing about which
-    # decodes it to "v1/../admin" before routing.
+    endpoint the provider credential gets spent against, so an EHBP request is
-    response = await integration_client.post(
+    screened and allowlisted exactly like any other."""
-        "/v1/%2e%2e/admin",
+    with patch(
-        content=b"encrypted",
+        "routstr.proxy.forward_ehbp_request",
-        headers={
+        AsyncMock(side_effect=AssertionError("must not forward unknown endpoint")),
-            "ehbp-encapsulated-key": "x",
+    ):
-            "x-routstr-model": "gpt-4",
+        response = await integration_client.post(
-        },
+            f"/{bad_path}",
-    )
+            content=b"encrypted",
+            headers={
+                "ehbp-encapsulated-key": "x",
+                "x-routstr-model": "gpt-4",
+            },
+        )
     assert response.status_code == 404
 
 

tests/unit/test_proxy_path_allowlist.py

@@ -2,8 +2,9 @@
 
 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
+attached. The proxy now rejects ambiguous path spellings for every method, then
-requires a known API prefix before anything is forwarded.
+requires the method/path pair to name a canonical endpoint. A familiar prefix is
+no longer enough: "v1/organization/api_keys" is refused just like "internal/admin".
 """
 
 from __future__ import annotations
@@ -18,6 +19,7 @@ import pytest  # noqa: E402
 from routstr.proxy import (  # noqa: E402
     _forwarding_allowed,
     _is_ambiguously_spelled_path,
+    _parse_extra_allowed_endpoints,
 )
 
 
@@ -63,37 +65,138 @@ def test_canonical_paths_are_allowed(path: str) -> None:
 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("internal/admin", "POST") is False
-    assert _forwarding_allowed("secret-endpoint", is_ehbp=False) is False
+    assert _forwarding_allowed("secret-endpoint", "POST") is False
 
 
 @pytest.mark.parametrize(
     "path",
     [
-        "modelsdump",  # bare token "models" must not match a longer segment
+        "modelsdump",  # "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:
+def test_endpoint_name_does_not_match_a_longer_segment(path: str) -> None:
-    assert _forwarding_allowed(path, is_ehbp=False) is False
+    assert _forwarding_allowed(path, "POST") is False
+    assert _forwarding_allowed(path, "GET") is False
 
 
-def test_known_prefixes_are_forwarded() -> None:
+@pytest.mark.parametrize(
-    assert _forwarding_allowed("v1/chat/completions", is_ehbp=False) is True
+    "path",
-    assert _forwarding_allowed("chat/completions", is_ehbp=False) is True
+    [
-    # Bare tokens match a whole segment: exactly or followed by "/".
+        # A familiar prefix must not carry an unknown endpoint. These are real
-    assert _forwarding_allowed("models", is_ehbp=False) is True
+        # upstream routes that manage keys, org membership, and billing.
-    assert _forwarding_allowed("models/gpt-4", is_ehbp=False) is True
+        "v1/organization/api_keys",
-    assert _forwarding_allowed("embeddings", is_ehbp=False) is True
+        "v1/api_keys",
-    assert _forwarding_allowed("attestation", is_ehbp=False) is True
+        "v1/admin/keys",
+        "v1/billing/usage",
+        "v1/files",
+        "v1/batches",
+        "chat/internal",
+        "audio/internal",
+        "images/internal",
+        "tee/keys",
+        # No endpoint takes a trailing id segment; a resource id never widens
+        # the reachable surface.
+        "models/gpt-4",
+        "models/gpt-4/secret",
+        "chat/completions/abc",
+    ],
+)
+def test_known_prefix_does_not_carry_an_unknown_endpoint(path: str) -> None:
+    assert _forwarding_allowed(path, "POST") is False
+    assert _forwarding_allowed(path, "GET") is False
 
 
-def test_ehbp_bypasses_prefix_gate_by_header() -> None:
+@pytest.mark.parametrize(
-    # Documents a deliberate exemption: EHBP is identified by header and carries
+    ("path", "method"),
-    # its own encrypted contract, so the prefix gate does not apply. The
+    [
-    # ambiguous-spelling screen still runs on EHBP paths (see the ordering test
+        ("v1/chat/completions", "POST"),
-    # in the integration suite).
+        ("chat/completions", "POST"),
-    assert _forwarding_allowed("anything/encrypted", is_ehbp=True) is True
+        ("v1/chat/completions/", "POST"),
+        ("completions", "POST"),
+        ("v1/responses", "POST"),
+        ("v1/messages", "POST"),
+        ("v1/embeddings", "POST"),
+        ("moderations", "POST"),
+        ("audio/transcriptions", "POST"),
+        ("images/generations", "POST"),
+        ("models", "GET"),
+        ("attestation", "GET"),
+        ("tee/attestation", "GET"),
+    ],
+)
+def test_canonical_endpoints_are_forwarded(path: str, method: str) -> None:
+    assert _forwarding_allowed(path, method) is True
+
+
+@pytest.mark.parametrize(
+    ("path", "method"),
+    [
+        ("chat/completions", "GET"),  # billed endpoints are POST-only
+        ("v1/embeddings", "GET"),
+        ("models", "POST"),  # read-only endpoints are GET-only
+        ("attestation", "POST"),
+        ("v1/chat/completions", "DELETE"),  # never routed here, refused anyway
+        ("v1/chat/completions", "PUT"),
+    ],
+)
+def test_method_must_match_the_endpoint(path: str, method: str) -> None:
+    assert _forwarding_allowed(path, method) is False
+
+
+def test_operator_additions_are_parsed_per_endpoint() -> None:
+    parsed = _parse_extra_allowed_endpoints("POST:v1/rerank, GET:batches ,post:audio/x")
+    assert parsed == {
+        "rerank": frozenset({"POST"}),  # the "v1/" prefix collapses like any path
+        "batches": frozenset({"GET"}),
+        "audio/x": frozenset({"POST"}),
+    }
+
+
+def test_operator_additions_may_grant_two_methods_on_one_endpoint() -> None:
+    assert _parse_extra_allowed_endpoints("POST:batches,GET:batches") == {
+        "batches": frozenset({"POST", "GET"})
+    }
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "",
+        "   ",
+        "v1/rerank",  # no method
+        "POST:",  # no path
+        ":v1/rerank",  # empty method
+        "DELETE:v1/rerank",  # method the proxy never routes
+        "POST:*",  # wildcards are deliberately unsupported
+        "POST:v1/*",
+        "POST:../secret",  # ambiguous spellings are screened here too
+        "POST:v1//rerank",
+        "POST:%2e%2e/secret",
+    ],
+)
+def test_malformed_operator_additions_widen_nothing(raw: str) -> None:
+    assert _parse_extra_allowed_endpoints(raw) == {}
+
+
+def test_operator_additions_are_env_only() -> None:
+    # The proxy parses this once at import, so a persisted or admin-API-writable
+    # value would be read but never take effect. Keeping it env-only also means
+    # widening the reachable upstream surface takes a deploy.
+    from routstr.core.settings import ENV_ONLY_FIELDS
+
+    assert "proxy_extra_allowed_paths" in ENV_ONLY_FIELDS
+
+
+def test_ehbp_is_gated_by_the_same_allowlist() -> None:
+    # EHBP hides the request body from the proxy, which is a reason to constrain
+    # the destination more tightly rather than to trust the caller's path: the
+    # encrypted contract covers the body, never the endpoint the provider
+    # credential is spent against.
+    assert _forwarding_allowed("anything/encrypted", "POST") is False
+    assert _forwarding_allowed("v1/organization/api_keys", "POST") is False
+    assert _forwarding_allowed("v1/chat/completions", "POST") is True

tests/unit/test_proxy_tinfoil_attestation_routing.py

@@ -103,13 +103,14 @@ async def test_attestation_trailing_slash_routes_directly_to_tinfoil(
 
 
 @pytest.mark.parametrize(
-    ("path", "expected_status"),
+    "path",
     [
-        # Valid `attestation` segment but not the exact attestation route:
+        # A valid `attestation` segment is not the exact attestation route, and
-        # fails model validation (empty body -> unknown model) before auth.
+        # `attestation` takes no id segment, so the endpoint allowlist rejects
-        ("attestation/foo", 400),
+        # it at the edge rather than letting it reach model/auth handling.
-        # Not a known path segment at all: rejected at the edge before routing.
+        "attestation/foo",
-        ("attestationjunk", 404),
+        # Not a known endpoint at all: rejected at the edge before routing.
+        "attestationjunk",
     ],
 )
 @pytest.mark.asyncio
@@ -117,7 +118,6 @@ async def test_non_attestation_prefix_does_not_bypass_authentication(
     monkeypatch: pytest.MonkeyPatch,
     proxy_app: FastAPI,
     path: str,
-    expected_status: int,
 ) -> None:
     tinfoil = MagicMock()
     tinfoil.provider_type = "tinfoil"
@@ -130,9 +130,7 @@ async def test_non_attestation_prefix_does_not_bypass_authentication(
     ) as client:
         response = await client.get(f"/{path}")
 
-    assert response.status_code == expected_status
+    assert response.status_code == 404
-    if expected_status == 400:
-        assert response.json()["error"]["type"] == "invalid_model"
     tinfoil.forward_get_request.assert_not_awaited()