Fix 500 from POST /auth/token on a request body it cannot parse

airflow-core/src/airflow/api_fastapi/auth/managers/simple/utils.py

@@ -16,9 +16,12 @@
 # under the License.
 from __future__ import annotations
 
+from json import JSONDecodeError
+
 from fastapi import HTTPException, Request, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
+from starlette.requests import ClientDisconnect
 
 from airflow.api_fastapi.auth.managers.simple.datamodels.login import LoginBody
 from airflow.api_fastapi.common.headers import HeaderContentTypeJsonOrForm
@@ -29,20 +32,61 @@ async def parse_login_body(
     request: Request,
     content_type: HeaderContentTypeJsonOrForm,
 ) -> LoginBody:
-    try:
-        if content_type == Mimetype.JSON:
+    # ``/token`` takes its body through this dependency rather than declaring one, so FastAPI
+    # parses no body for it and the reads below are the only parse. The route is unauthenticated,
+    # so anything left unhandled here is a 500 that any caller can reach without credentials.
+    if content_type == Mimetype.JSON:
+        try:
             body = await request.json()
-        elif content_type == Mimetype.FORM:
-            form = await request.form()
-            body = {
-                "username": form.get("username"),
-                "password": form.get("password"),
-            }
-        else:
+        except JSONDecodeError as e:
+            # This arm and the next are the chain FastAPI runs around its own ``request.json()``,
+            # so an unreadable body gets the same status here as on ``/token/cli``, where the
+            # body is declared and FastAPI validates it natively.
+            raise RequestValidationError(
+                [{"type": "json_invalid", "loc": ["body", e.pos], "msg": "JSON decode error"}]
+            ) from e
+        except Exception as e:
+            # ``json.loads`` also raises UnicodeDecodeError, a bare ValueError (a number over
+            # ``int_max_str_digits``) and RecursionError. A server-side failure such as
+            # MemoryError is reported as a client error too, which is the trade FastAPI makes
+            # at the same point.
             raise HTTPException(
-                status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
-                detail="Unsupported Media Type",
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="There was an error parsing the body",
+            ) from e
+        if not isinstance(body, dict):
+            # Valid JSON, but a list, scalar or null cannot be splatted into the model below.
+            raise RequestValidationError(
+                [
+                    {
+                        "type": "model_attributes_type",
+                        "loc": ["body"],
+                        "msg": "Input should be a valid dictionary or object to extract fields from",
+                    }
+                ]
             )
+    elif content_type == Mimetype.FORM:
+        try:
+            form = await request.form()
+        except ClientDisconnect as e:
+            # Only the transport failure is translated. Starlette reports its own form-parser
+            # limits as a 400 carrying a specific detail ("Too many fields..."), so those are
+            # left to propagate rather than be flattened into the generic message above.
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="There was an error parsing the body",
+            ) from e
+        body = {
+            "username": form.get("username"),
+            "password": form.get("password"),
+        }
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+            detail="Unsupported Media Type",
+        )
+
+    try:
         return LoginBody(**body)
     except ValidationError as e:
         raise RequestValidationError(repr(e))

airflow-core/tests/unit/api_fastapi/auth/managers/simple/routes/test_login.py

@@ -20,6 +20,8 @@
 
 from unittest.mock import patch
 
+import pytest
+
 from tests_common.test_utils.config import conf_vars
 
 
@@ -131,3 +133,59 @@ def test_create_token_invalid_user_password_cli(self, test_client):
         )
         assert response.status_code == 401
         assert response.json()["detail"] == "Invalid credentials"
+
+
+# Bodies that reach ``parse_login_body`` without being valid credentials: the first five make
+# ``json.loads`` itself raise, the last three parse to something that is not a JSON object.
+UNPARSABLE_JSON_BODIES = [
+    pytest.param(b"{", id="malformed"),
+    pytest.param(b"", id="empty"),
+    pytest.param(b'{"username": "\xff\xfe"}', id="invalid_utf8"),
+    pytest.param(b'{"username": ' + b"1" * 4301 + b"}", id="oversized_int"),
+    pytest.param(b"[" * 20_000, id="deeply_nested"),
+    pytest.param(b'["test1", "DUMMY_PASS"]', id="json_array"),
+    pytest.param(b"5", id="json_scalar"),
+    pytest.param(b"null", id="json_null"),
+]
+
+
+class TestCreateTokenUnparsableBody:
+    """
+    ``/auth/token`` must answer a body it cannot read rather than raise out of the dependency.
+
+    Unlike ``/auth/token/cli``, which declares ``body: LoginBody`` and so gets FastAPI's own
+    validation, this route takes its body through ``Depends(parse_login_body)``. FastAPI parses
+    no body for it, leaving that dependency the only parse. The route is unauthenticated, so
+    anything it raises is reachable without credentials.
+    """
+
+    @pytest.mark.parametrize("payload", UNPARSABLE_JSON_BODIES)
+    def test_answers_like_the_natively_parsed_route(self, payload, test_client):
+        """
+        Both routes take the same body and must answer it the same way.
+
+        Asserting the pair agree rather than naming status codes keeps this pinned to whatever
+        FastAPI does with a body it cannot read, instead of to numbers copied out of it.
+        """
+        headers = {"Content-Type": "application/json"}
+        declared = test_client.post("/auth/token/cli", content=payload, headers=headers)
+        via_dependency = test_client.post("/auth/token", content=payload, headers=headers)
+
+        assert declared.status_code < 500
+        assert via_dependency.status_code == declared.status_code
+
+    def test_form_parser_error_keeps_its_own_detail(self, test_client):
+        """
+        A form body the parser rejects must keep starlette's message, not the generic one.
+
+        The catch-all that turns an unreadable body into a 400 also sees the 400 starlette
+        raises from its own form parser, so it has to let that one through untouched.
+        """
+        response = test_client.post(
+            "/auth/token",
+            content=b"&".join(b"a=1" for _ in range(2000)),
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        )
+
+        assert response.status_code == 400
+        assert "Too many fields" in response.json()["detail"]