This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 490eb31ac32 Fix secrets masking in Rendered Templates for complex
objects (#61394)
490eb31ac32 is described below
commit 490eb31ac3275e901fd60c45c632a0032d716074
Author: Amogh Desai <[email protected]>
AuthorDate: Wed Feb 11 12:25:21 2026 +0530
Fix secrets masking in Rendered Templates for complex objects (#61394)
---
.../src/airflow/sdk/execution_time/task_runner.py | 13 ++++++-
.../task_sdk/execution_time/test_task_runner.py | 45 ++++++++++++++++++++++
2 files changed, 57 insertions(+), 1 deletion(-)
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 9fa13a08d36..2b3933c1071 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -957,13 +957,24 @@ def _serialize_template_field(template_field: Any, name:
str) -> str | dict | li
return tuple(sort_dict_recursively(item) for item in obj)
return obj
+ def _fallback_serialization(obj):
+ """Serialize objects with to_dict() method (eg: k8s objects) for
json.dumps() default parameter."""
+ if hasattr(obj, "to_dict"):
+ return obj.to_dict()
+ raise TypeError(f"cannot serialize {obj}")
+
max_length = conf.getint("core", "max_templated_field_length")
if not is_jsonable(template_field):
try:
serialized = template_field.serialize()
except AttributeError:
- serialized = str(template_field)
+ # check if these objects can be converted to JSON serializable
types
+ try:
+ serialized = json.dumps(template_field,
default=_fallback_serialization)
+ except (TypeError, ValueError):
+ # fall back to string representation if not
+ serialized = str(template_field)
if len(serialized) > max_length:
rendered = redact(serialized, name)
return (
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index fc832e3195d..8724877853f 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -2657,6 +2657,51 @@ class TestRuntimeTaskInstance:
in mock_supervisor_comms.send.mock_calls
)
+ @pytest.mark.enable_redact
+ def test_rendered_templates_masks_secrets_in_complex_objects(
+ self, create_runtime_ti, mock_supervisor_comms
+ ):
+ """Test that secrets in complex objects like V1EnvVar are masked
well."""
+ from kubernetes import client as k8s
+
+ from airflow.sdk._shared.secrets_masker import _secrets_masker
+
+ secret1 = "This is a longer test phrase. We are checking if this
handles regular sentences."
+ secret2 = "Lorem ipsum dolor sit amet, consectetur adipiscing elit.
Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad
minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea
commodo consequat."
+ _secrets_masker().add_mask(secret1, None)
+ _secrets_masker().add_mask(secret2, None)
+
+ class CustomOperator(BaseOperator):
+ template_fields = ("env_vars",)
+
+ def __init__(self, env_vars, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.env_vars = env_vars
+
+ def execute(self, context):
+ pass
+
+ env_vars = [
+ k8s.V1EnvVar(name="var1", value="This is a test phrase."),
+ k8s.V1EnvVar(name="var2", value=secret1),
+ k8s.V1EnvVar(name="var3", value=secret2),
+ ]
+
+ task = CustomOperator(
+ task_id="test_complex_object_masking",
+ env_vars=env_vars,
+ )
+
+ runtime_ti = create_runtime_ti(task=task,
dag_id="test_complex_object_dag")
+ run(runtime_ti, context=runtime_ti.get_template_context(),
log=mock.MagicMock())
+
+ rendered_fields =
mock_supervisor_comms.send.mock_calls[0].kwargs["msg"].rendered_fields
+ assert rendered_fields is not None
+ assert (
+ rendered_fields["env_vars"]
+ == '[{"name": "var1", "value": "This is a test phrase.",
"value_from": null}, {"name": "var2", "value": "***", "value_from": null},
{"name": "var3", "value": "***", "value_from": null}]'
+ )
+
class TestXComAfterTaskExecution:
@pytest.mark.parametrize(
