Merge pull request #35773 from HuanQian571/litellm_fix_management_v1_get_flat_params

fix(proxy): restore management_v1 query-param validation under fastapi>=0.140.7

litellm/proxy/management_endpoints/management_v1/common.py

@@ -4,7 +4,8 @@ from typing import Final
 from urllib.parse import urlencode
 
 from fastapi import Request
-from fastapi.dependencies.utils import get_flat_dependant
+from fastapi.dependencies.utils import get_flat_params
+from fastapi.params import ParamTypes
 from fastapi.responses import JSONResponse
 
 from litellm.types.proxy.management_endpoints.management_v1 import (
@@ -42,7 +43,13 @@ def _declared_query_params(request: Request) -> frozenset[str]:
     dependant: Final = getattr(route, "dependant", None)
     if dependant is None:
         return frozenset()
-    return frozenset(field.alias for field in get_flat_dependant(dependant, skip_repeats=True).query_params)
+    # fastapi>=0.140.7 removed get_flat_dependant(); get_flat_params() returns the
+    # flattened (deduped) param list. Filter to query params to match the old behavior.
+    return frozenset(
+        field.alias
+        for field in get_flat_params(dependant)
+        if getattr(field.field_info, "in_", None) == ParamTypes.query
+    )
 
 
 def escape_like(value: str) -> str:

tests/test_litellm/proxy/management_endpoints/management_v1/test_common.py

@@ -0,0 +1,95 @@
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, Header, Query, Request
+from fastapi.testclient import TestClient
+
+from litellm.proxy.management_endpoints.management_v1.common import (
+    ManagementProblem,
+    PROBLEM_CONTENT_TYPE,
+    _declared_query_params,
+    problem_response,
+    reject_unknown_query_params,
+)
+
+
+def _client() -> TestClient:
+    app = FastAPI()
+
+    @app.exception_handler(ManagementProblem)
+    async def _handle(_request: Request, exc: ManagementProblem):
+        return problem_response(exc.problem)
+
+    @app.get("/things/{thing_id}", dependencies=[Depends(reject_unknown_query_params)])
+    def _handler(
+        thing_id: str,
+        request: Request,
+        status: Annotated[str | None, Query(alias="filter[status]")] = None,
+        page: Annotated[int, Query(ge=1)] = 1,
+        x_trace: Annotated[str | None, Header()] = None,
+    ) -> dict[str, bool]:
+        return {"ok": True}
+
+    return TestClient(app, raise_server_exceptions=False)
+
+
+def test_a_declared_query_param_is_accepted_by_its_alias():
+    response = _client().get("/things/abc", params={"filter[status]": "active", "page": "2"})
+    assert response.status_code == 200, response.text
+
+
+def test_an_unknown_query_param_is_rejected_as_a_problem():
+    response = _client().get("/things/abc", params={"bogus": "x"})
+    assert response.status_code == 400
+    assert response.headers["content-type"].startswith(PROBLEM_CONTENT_TYPE)
+    assert "bogus" in response.json()["detail"]
+
+
+def test_a_path_param_name_is_not_a_declared_query_param():
+    """The flatten step returns path+query+header together; only query names count as declared.
+
+    If the ParamTypes.query filter were dropped, `thing_id` (a path param) would leak
+    into the declared set and this request would be wrongly accepted.
+    """
+    response = _client().get("/things/abc", params={"thing_id": "x"})
+    assert response.status_code == 400
+    assert "thing_id" in response.json()["detail"]
+
+
+def test_a_header_param_name_is_not_a_declared_query_param():
+    response = _client().get("/things/abc", params={"x-trace": "x"})
+    assert response.status_code == 400
+    assert "x-trace" in response.json()["detail"]
+
+
+def test_declared_query_params_isolates_query_aliases_from_other_param_types():
+    captured: dict[str, frozenset[str]] = {}
+    app = FastAPI()
+
+    @app.get("/things/{thing_id}")
+    def _handler(
+        thing_id: str,
+        request: Request,
+        status: Annotated[str | None, Query(alias="filter[status]")] = None,
+        page: Annotated[int, Query(ge=1)] = 1,
+        x_trace: Annotated[str | None, Header()] = None,
+    ) -> dict[str, bool]:
+        captured["declared"] = _declared_query_params(request)
+        return {"ok": True}
+
+    TestClient(app).get("/things/abc")
+    assert captured["declared"] == frozenset({"filter[status]", "page"})
+
+
+def test_declared_query_params_is_empty_when_the_route_has_no_dependant():
+    request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "scheme": "http",
+            "root_path": "",
+            "path": "/things/abc",
+            "query_string": b"",
+            "headers": [(b"host", b"testserver")],
+        }
+    )
+    assert _declared_query_params(request) == frozenset()