diff --git a/airflow-core/newsfragments/70936.bugfix.rst b/airflow-core/newsfragments/70936.bugfix.rst new file mode 100644 index 0000000000000..a919995570aca --- /dev/null +++ b/airflow-core/newsfragments/70936.bugfix.rst @@ -0,0 +1 @@ +Fix API clients discarding JSON error bodies when a reverse proxy rewrites the ``content-type`` header. The media type was compared to the literal ``application/json``, so ``application/json; charset=utf-8`` or ``application/problem+json`` skipped error parsing entirely: ``airflowctl`` printed a raw ``httpx.HTTPStatusError`` traceback in place of the server's message, and in the Task SDK the exception was no longer a ``ServerResponseError``, so the status-specific handling in the supervisor and the task runner stopped matching. Two adjacent ``airflowctl`` crashes are fixed with it — a non-object JSON error body made the warning log raise, and a body labelled JSON that does not decode raised ``ValueError`` instead of falling through to ``raise_for_status()``. diff --git a/airflow-ctl/src/airflowctl/api/client.py b/airflow-ctl/src/airflowctl/api/client.py index e160cf295b7d2..e23d027711851 100644 --- a/airflow-ctl/src/airflowctl/api/client.py +++ b/airflow-ctl/src/airflowctl/api/client.py @@ -113,7 +113,10 @@ def get_json_error(response: httpx.Response): # This part is used in integration tests to verify the error message # If you are updating here don't forget to update the airflow-ctl-tests if not response.request.extensions.get("airflowctl_suppress_error_log"): - log.warning("Server error ", extra=dict(err.response.json())) + detail = err.response.json() + # A JSON error body is not necessarily an object; anything else would make + # dict() raise and turn a handled error into a traceback. + log.warning("Server error ", extra=detail if isinstance(detail, dict) else {"detail": detail}) raise err diff --git a/airflow-ctl/src/airflowctl/api/operations.py b/airflow-ctl/src/airflowctl/api/operations.py index c5f0dc392f02c..8989a8c2cb14f 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -106,6 +106,18 @@ def _build_query_params(**values: Any) -> dict[str, Any]: return {name: _serialize_query_param(value) for name, value in values.items() if value is not None} +def _has_json_body(response: httpx.Response) -> bool: + """Check whether a response declares a JSON body.""" + # Kept in sync with airflow.sdk.api.client._has_json_body. + # Media types are case-insensitive and may carry parameters (RFC 9110), so a proxy in + # front of the API server that rewrites "application/json" into + # "application/json; charset=utf-8" must not defeat the check — otherwise the friendly + # error handling built on ServerResponseError is silently skipped and callers get a + # bare httpx.HTTPStatusError instead. + media_type = response.headers.get("content-type", "").partition(";")[0].strip().lower() + return media_type == "application/json" or media_type.endswith("+json") + + # Generic Server Response Error class ServerResponseError(httpx.HTTPStatusError): """Server response error (Generic).""" @@ -115,16 +127,23 @@ def from_response(cls, response: httpx.Response) -> ServerResponseError | None: if response.status_code < 400: return None - if response.headers.get("content-type") != "application/json": + if not _has_json_body(response): return None # httpx runs response event hooks before it reads the body, so the body has to be # pulled in explicitly here or ``.json()`` raises ``httpx.ResponseNotRead``. response.read() + try: + detail = response.json() + except ValueError: + # Undecodable body despite a JSON content-type (truncated response, or a proxy + # mislabelling its own error page) — leave it to raise_for_status(). + return None + error_kind = "Client" if response.status_code < 500 else "Server" return cls( - message=f"{error_kind} error message: {response.json()}", + message=f"{error_kind} error message: {detail}", request=response.request, response=response, ) diff --git a/airflow-ctl/tests/airflow_ctl/api/test_client.py b/airflow-ctl/tests/airflow_ctl/api/test_client.py index 22ad51d796e04..6e6bbd31daba5 100644 --- a/airflow-ctl/tests/airflow_ctl/api/test_client.py +++ b/airflow-ctl/tests/airflow_ctl/api/test_client.py @@ -23,6 +23,7 @@ import subprocess import sys import tempfile +from typing import Any from unittest.mock import MagicMock, patch import httpx @@ -55,7 +56,9 @@ def handle_request(request: httpx.Request) -> httpx.Response: return Client(base_url="", token="", mounts={"'http://": httpx.MockTransport(handle_request)}) -def make_unread_json_response(status_code: int, payload: dict, **kwargs) -> httpx.Response: +def make_unread_json_response( + status_code: int, payload: Any, content_type: str = "application/json", **kwargs +) -> httpx.Response: """ Build a JSON response whose body has not been read yet. @@ -65,7 +68,7 @@ def make_unread_json_response(status_code: int, payload: dict, **kwargs) -> http """ return httpx.Response( status_code, - headers={"content-type": "application/json"}, + headers={"content-type": content_type}, content=iter([json.dumps(payload).encode()]), **kwargs, ) @@ -142,6 +145,65 @@ def test_error_parsing_with_unread_body(self, status_code, expected_message): assert err.value.args == (expected_message,) + @pytest.mark.parametrize( + "content_type", + [ + pytest.param("application/json", id="bare"), + pytest.param("application/json; charset=utf-8", id="charset"), + pytest.param("application/json;charset=UTF-8", id="charset-no-space"), + pytest.param("Application/JSON", id="mixed-case"), + pytest.param("application/problem+json", id="problem-json"), + ], + ) + @pytest.mark.parametrize("status_code", [404, 500]) + def test_error_parsing_json_media_type_variants(self, content_type, status_code): + response = make_unread_json_response( + status_code, + {"detail": "boom"}, + content_type=content_type, + request=httpx.Request("GET", "http://error"), + ) + + with pytest.raises(ServerResponseError): + get_json_error(response) + + def test_error_parsing_media_type_merely_prefixed_with_json(self): + response = make_unread_json_response( + 404, + {"detail": "boom"}, + content_type="application/jsonp", + request=httpx.Request("GET", "http://error"), + ) + + assert get_json_error(response) is None + + def test_error_parsing_undecodable_json_body(self): + response = httpx.Response( + 404, + headers={"content-type": "application/json"}, + content=iter([b"{truncated"]), + request=httpx.Request("GET", "http://error"), + ) + + assert get_json_error(response) is None + + @pytest.mark.parametrize( + ("payload", "expected_extra"), + [ + pytest.param({"detail": "Not found"}, {"detail": "Not found"}, id="object"), + pytest.param(["a", "b"], {"detail": ["a", "b"]}, id="array"), + pytest.param("plain message", {"detail": "plain message"}, id="string"), + ], + ) + @patch("airflowctl.api.client.log.warning") + def test_error_log_accepts_non_object_json_body(self, mock_warning, payload, expected_extra): + response = make_unread_json_response(404, payload, request=httpx.Request("GET", "http://error")) + + with pytest.raises(ServerResponseError): + get_json_error(response) + + assert mock_warning.call_args.kwargs["extra"] == expected_extra + @pytest.mark.parametrize( ("suppress_error_log", "expected_warning_count"), [ diff --git a/task-sdk/src/airflow/sdk/api/client.py b/task-sdk/src/airflow/sdk/api/client.py index ea220c68fde71..0affc04d57394 100644 --- a/task-sdk/src/airflow/sdk/api/client.py +++ b/task-sdk/src/airflow/sdk/api/client.py @@ -1328,6 +1328,15 @@ def dags(self) -> DagsOperations: return DagsOperations(self) +def _has_json_body(response: httpx.Response) -> bool: + """Check whether a response declares a JSON body.""" + # Kept in sync with airflowctl.api.operations._has_json_body. + # Media types are case-insensitive and may carry parameters (RFC 9110), so a proxy that + # rewrites "application/json" into "application/json; charset=utf-8" must not defeat the check. + media_type = response.headers.get("content-type", "").partition(";")[0].strip().lower() + return media_type == "application/json" or media_type.endswith("+json") + + # This is only used for parsing. ServerResponseError is raised instead class _ErrorBody(BaseModel): detail: list[RemoteValidationError] | dict[str, Any] | str @@ -1354,7 +1363,7 @@ def from_response(cls, response: httpx.Response) -> ServerResponseError | None: if not (400 <= response.status_code < 600): return None - if response.headers.get("content-type") != "application/json": + if not _has_json_body(response): return None detail: list[RemoteValidationError] | dict[str, Any] | None = None diff --git a/task-sdk/tests/task_sdk/api/test_client.py b/task-sdk/tests/task_sdk/api/test_client.py index 684d4e7dcbb17..0642330fe04b0 100644 --- a/task-sdk/tests/task_sdk/api/test_client.py +++ b/task-sdk/tests/task_sdk/api/test_client.py @@ -175,6 +175,35 @@ def test_error_parsing_other_json(self): assert err.value.args == ("Not found",) assert err.value.detail is None + @pytest.mark.parametrize( + "content_type", + [ + pytest.param("application/json; charset=utf-8", id="charset"), + pytest.param("application/json;charset=UTF-8", id="charset-no-space"), + pytest.param("Application/JSON", id="mixed-case"), + pytest.param("application/problem+json", id="problem-json"), + ], + ) + def test_error_parsing_json_media_type_variants(self, content_type): + responses = [ + httpx.Response(404, json={"detail": "Not found"}, headers={"content-type": content_type}) + ] + client = make_client_w_responses(responses) + + with pytest.raises(ServerResponseError) as err: + client.get("http://error") + assert err.value.args == ("Not found",) + + def test_error_parsing_media_type_merely_prefixed_with_json(self): + responses = [ + httpx.Response(404, json={"detail": "Not found"}, headers={"content-type": "application/jsonp"}) + ] + client = make_client_w_responses(responses) + + with pytest.raises(httpx.HTTPStatusError) as err: + client.get("http://error") + assert not isinstance(err.value, ServerResponseError) + def test_server_response_error_pickling(self): responses = [httpx.Response(404, json={"detail": {"message": "Invalid input"}})] client = make_client_w_responses(responses)