Fix 400 and 500s on POST /connections/defaults in multi-team mode

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -19,7 +19,6 @@
 import posixpath
 from collections.abc import Callable, Coroutine
 from contextlib import suppress
-from json import JSONDecodeError
 from typing import TYPE_CHECKING, Annotated, Any, cast
 from urllib.parse import ParseResult, unquote, urljoin, urlparse
 
@@ -1326,14 +1325,23 @@ async def _collect_teams_to_check(
     if method != "POST":
         teams.add(get_existing_team(resource_id) if resource_id else None)
     if method in ("POST", "PUT"):
-        try:
-            body = await request.json()
-        except JSONDecodeError:
-            # Fail closed: reject unparsable bodies before any authz decision.
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Request body is not valid JSON",
-            )
+        body: Any = None
+        # ``POST /connections/defaults`` declares no body model, so FastAPI parses nothing for it
+        # and the read below is its only parse: a bodyless request there names no team and must
+        # still be served. Every other route arriving here declares a body, which FastAPI has
+        # already parsed and cached by the time this dependency runs.
+        if await request.body():
+            try:
+                body = await request.json()
+            except Exception as e:
+                # Broad because ``json.loads`` also raises UnicodeDecodeError, a bare ValueError
+                # (a number over ``int_max_str_digits``) and RecursionError, none of them
+                # JSONDecodeError. Fail closed: a body that cannot be read may carry a
+                # ``team_name``, so reject it rather than treat it as naming no team.
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Request body is not valid JSON",
+                ) from e
         raw = body.get("team_name") if isinstance(body, dict) else None
         if raw is not None and not isinstance(raw, str):
             # Fail closed: reject non-string team_name before authz / DB lookup.

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py

@@ -1798,6 +1798,14 @@ def test_should_respond_204(self, test_client, session):
         assert response.content == b""
         _check_last_log(session, dag_id=None, event="create_default_connections", logical_date=None)
 
+    @conf_vars({("core", "multi_team"): "True"})
+    def test_should_respond_204_in_multi_team_mode(self, test_client, session):
+        """The route declares no body model, so the team-scoping dependency's read is the only
+        parse of the body: a correct bodyless request must not be rejected as unparsable."""
+        response = test_client.post("/connections/defaults")
+        assert response.status_code == 204
+        assert response.content == b""
+
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.post("/connections/defaults")
         assert response.status_code == 401

airflow-core/tests/unit/api_fastapi/core_api/test_security.py

@@ -894,6 +894,7 @@ async def test_requires_access_connection(
         mock_get_name_if_exists.return_value = team_name
         fastapi_request = Mock()
         fastapi_request.path_params = {"connection_id": "conn_id"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={})
         user = Mock()
 
@@ -920,6 +921,7 @@ async def test_requires_access_connection_team_from_body(
         mock_get_name_if_exists.return_value = "team1"
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "team1"})
         user = Mock()
 
@@ -948,6 +950,7 @@ async def test_requires_access_connection_put_checks_both_teams(
         mock_get_name_if_exists.side_effect = lambda name: name
         fastapi_request = Mock()
         fastapi_request.path_params = {"connection_id": "conn_id"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "new-team"})
         user = Mock()
 
@@ -976,6 +979,7 @@ async def test_requires_access_connection_post_invalid_team_returns_400(
         mock_get_name_if_exists.return_value = None
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "nonexistent"})
         user = Mock()
 
@@ -1001,6 +1005,7 @@ async def test_requires_access_connection_body_parse_failure_returns_400(
         mock_get_team_name.return_value = None
         fastapi_request = Mock()
         fastapi_request.path_params = {"connection_id": "conn_id"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(side_effect=JSONDecodeError("expecting value", "", 0))
         user = Mock()
 
@@ -1011,6 +1016,67 @@ async def test_requires_access_connection_body_parse_failure_returns_400(
         assert exc_info.value.status_code == 400
         auth_manager.is_authorized_connection.assert_not_called()
 
+    @pytest.mark.db_test
+    @pytest.mark.parametrize("method", ["POST", "PUT"])
+    @pytest.mark.parametrize(
+        "parse_error",
+        [
+            UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte"),
+            ValueError("Exceeds the limit (4300 digits) for integer string conversion"),
+            RecursionError("maximum recursion depth exceeded while decoding a JSON array"),
+        ],
+        ids=["invalid-utf8", "oversized-int", "deep-nesting"],
+    )
+    @patch.object(Team, "get_name_if_exists")
+    @patch.object(Connection, "get_team_name")
+    @patch("airflow.api_fastapi.core_api.security.get_auth_manager")
+    async def test_requires_access_connection_body_parse_failure_other_than_jsondecodeerror(
+        self, mock_get_auth_manager, mock_get_team_name, mock_get_name_if_exists, parse_error, method
+    ):
+        """``json.loads`` raises these three besides JSONDecodeError; each must fail closed."""
+        auth_manager = Mock()
+        auth_manager.is_authorized_connection.return_value = False
+        mock_get_auth_manager.return_value = auth_manager
+        mock_get_team_name.return_value = None
+        fastapi_request = Mock()
+        fastapi_request.path_params = {"connection_id": "conn_id"}
+        fastapi_request.body = AsyncMock(return_value=b"<unparsable>")
+        fastapi_request.json = AsyncMock(side_effect=parse_error)
+        user = Mock()
+
+        with conf_vars({("core", "multi_team"): "True"}):
+            with pytest.raises(HTTPException) as exc_info:
+                await requires_access_connection(method)(fastapi_request, user)
+
+        assert exc_info.value.status_code == 400
+        auth_manager.is_authorized_connection.assert_not_called()
+
+    @pytest.mark.db_test
+    @patch.object(Team, "get_name_if_exists")
+    @patch.object(Connection, "get_team_name")
+    @patch("airflow.api_fastapi.core_api.security.get_auth_manager")
+    async def test_requires_access_connection_empty_body_names_no_team(
+        self, mock_get_auth_manager, mock_get_team_name, mock_get_name_if_exists
+    ):
+        """``POST /connections/defaults`` declares no body model, so a bodyless request reaches
+        here unparsed: it names no team and must be authorized, not rejected as unparsable."""
+        auth_manager = Mock()
+        auth_manager.is_authorized_connection.return_value = True
+        mock_get_auth_manager.return_value = auth_manager
+        mock_get_team_name.return_value = None
+        fastapi_request = Mock()
+        fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"")
+        fastapi_request.json = AsyncMock(return_value={})
+        user = Mock()
+
+        with conf_vars({("core", "multi_team"): "True"}):
+            await requires_access_connection("POST")(fastapi_request, user)
+
+        fastapi_request.json.assert_not_called()
+        auth_manager.is_authorized_connection.assert_called_once()
+        assert auth_manager.is_authorized_connection.call_args.kwargs["details"].team_name is None
+
     @pytest.mark.db_test
     @pytest.mark.parametrize("method", ["POST", "PUT"])
     @pytest.mark.parametrize("bad_team_name", [123, ["x"], {"name": "x"}, True])
@@ -1032,6 +1098,7 @@ async def test_requires_access_connection_non_string_team_name_returns_400(
         mock_get_team_name.return_value = None
         fastapi_request = Mock()
         fastapi_request.path_params = {"connection_id": "conn_id"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": bad_team_name})
         user = Mock()
 
@@ -1154,6 +1221,7 @@ async def test_requires_access_variable(
         mock_get_name_if_exists.return_value = team_name
         fastapi_request = Mock()
         fastapi_request.path_params = {"variable_key": "var_key"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={})
         user = Mock()
 
@@ -1180,6 +1248,7 @@ async def test_requires_access_variable_team_from_body(
         mock_get_name_if_exists.return_value = "team1"
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "team1"})
         user = Mock()
 
@@ -1208,6 +1277,7 @@ async def test_requires_access_variable_put_checks_both_teams(
         mock_get_name_if_exists.side_effect = lambda name: name
         fastapi_request = Mock()
         fastapi_request.path_params = {"variable_key": "var_key"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "new-team"})
         user = Mock()
 
@@ -1236,6 +1306,7 @@ async def test_requires_access_variable_post_invalid_team_returns_400(
         mock_get_name_if_exists.return_value = None
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "nonexistent"})
         user = Mock()
 
@@ -1330,6 +1401,7 @@ async def test_requires_access_pool(
         mock_get_name_if_exists.return_value = team_name
         fastapi_request = Mock()
         fastapi_request.path_params = {"pool_name": "pool"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={})
         user = Mock()
 
@@ -1354,6 +1426,7 @@ async def test_requires_access_pool_team_from_body(self, mock_get_auth_manager,
         mock_get_name_if_exists.return_value = "team1"
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "team1"})
         user = Mock()
 
@@ -1382,6 +1455,7 @@ async def test_requires_access_pool_put_checks_both_teams(
         mock_get_name_if_exists.side_effect = lambda name: name
         fastapi_request = Mock()
         fastapi_request.path_params = {"pool_name": "pool"}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "new-team"})
         user = Mock()
 
@@ -1410,6 +1484,7 @@ async def test_requires_access_pool_post_invalid_team_returns_400(
         mock_get_name_if_exists.return_value = None
         fastapi_request = Mock()
         fastapi_request.path_params = {}
+        fastapi_request.body = AsyncMock(return_value=b"{}")
         fastapi_request.json = AsyncMock(return_value={"team_name": "nonexistent"})
         user = Mock()