Gate raw Dag deserialization error detail behind api.expose_stacktrace and log it server-side

airflow-core/src/airflow/api_fastapi/common/exceptions.py

@@ -116,9 +116,21 @@ def __init__(self):
 
     def exception_handler(self, request: Request, exc: DeserializationError):
         """Handle Dag deserialization exceptions."""
+        if conf.get("api", "expose_stacktrace") == "True":
+            log.error("Error while trying to deserialize Dag: %s", exc, exc_info=exc)
+            detail = f"An error occurred while trying to deserialize Dag: {exc}"
+        else:
+            # Only mint a correlation id when the detail is redacted, so the
+            # generic client message can be tied back to the server-side log.
+            exception_id = get_random_string()
+            log.error("Error with id %s while trying to deserialize Dag: %s", exception_id, exc, exc_info=exc)
+            detail = (
+                "An error occurred while trying to deserialize the Dag. Check the api server "
+                f"logs for more details - look for ID {exception_id}."
+            )
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"An error occurred while trying to deserialize Dag: {exc}",
+            detail=detail,
         )
 
 

airflow-core/tests/unit/api_fastapi/common/test_exceptions.py

@@ -406,6 +406,7 @@ class TestDagErrorHandler:
             "ValueError",
         ],
     )
+    @conf_vars({("api", "expose_stacktrace"): "True"})
     def test_handle_deserialization_error(self, cause: Exception) -> None:
         deserialization_error = DeserializationError("test_dag_id")
         deserialization_error.__cause__ = cause
@@ -418,6 +419,26 @@ def test_handle_deserialization_error(self, cause: Exception) -> None:
         with pytest.raises(HTTPException, match=re.escape(expected_exception.detail)):
             DagErrorHandler().exception_handler(Mock(), deserialization_error)
 
+    @conf_vars({("api", "expose_stacktrace"): "False"})
+    def test_handle_deserialization_error_without_stacktrace(self) -> None:
+        deserialization_error = DeserializationError("secret_dag_id")
+        deserialization_error.__cause__ = RuntimeError("internal credential leak xyz")
+
+        with patch("airflow.api_fastapi.common.exceptions.log") as mock_log:
+            with pytest.raises(HTTPException) as exc_info:
+                DagErrorHandler().exception_handler(Mock(), deserialization_error)
+
+        detail = exc_info.value.detail
+        assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+        # Raw exception text is withheld when expose_stacktrace is off ...
+        assert str(deserialization_error) not in detail
+        assert "internal credential leak xyz" not in detail
+        # ... and the caller gets an id to correlate with the server-side log.
+        assert "look for ID" in detail
+        # The error is still logged server-side, so detail is not silently lost.
+        mock_log.error.assert_called_once()
+
+    @conf_vars({("api", "expose_stacktrace"): "True"})
     @pytest.mark.usefixtures("testing_dag_bundle")
     @pytest.mark.need_serialized_dag
     def test_handle_real_dag_deserialization_error(self, session: Session, dag_maker: DagMaker) -> None: