Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airflow-core/newsfragments/70936.bugfix.rst
Original file line number Diff line number Diff line change
@@ -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()``.
5 changes: 4 additions & 1 deletion airflow-ctl/src/airflowctl/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
23 changes: 21 additions & 2 deletions airflow-ctl/src/airflowctl/api/operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)."""
Expand All @@ -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,
)
Expand Down
66 changes: 64 additions & 2 deletions airflow-ctl/tests/airflow_ctl/api/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import subprocess
import sys
import tempfile
from typing import Any
from unittest.mock import MagicMock, patch

import httpx
Expand Down Expand Up @@ -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.

Expand All @@ -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,
)
Expand Down Expand Up @@ -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"),
[
Expand Down
11 changes: 10 additions & 1 deletion task-sdk/src/airflow/sdk/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
rjgoyln marked this conversation as resolved.
# 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
Expand All @@ -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
Expand Down
29 changes: 29 additions & 0 deletions task-sdk/tests/task_sdk/api/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down