[airflow] branch main updated: Fix secrets rendered in UI when task is not executed. (#22754)

[ephraimanierobi]

This is an automated email from the ASF dual-hosted git repository.

ephraimanierobi 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 ce8ea66918 Fix secrets rendered in UI when task is not executed. 
(#22754)
ce8ea66918 is described below

commit ce8ea6691820140a0e2d9a5dad5254bc05a5a270
Author: Karthikeyan Singaravelan <tir.kar...@gmail.com>
AuthorDate: Fri May 20 19:37:47 2022 +0530

    Fix secrets rendered in UI when task is not executed. (#22754)
---
 airflow/models/taskinstance.py         | 14 ++++++++++-
 tests/www/views/test_views_rendered.py | 43 +++++++++++++++++++++++++++++++---
 2 files changed, 53 insertions(+), 4 deletions(-)

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index f08b3ba812..4b2b2f9348 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -2094,8 +2094,18 @@ class TaskInstance(Base, LoggingMixin):
             for field_name, rendered_value in 
rendered_task_instance_fields.items():
                 setattr(self.task, field_name, rendered_value)
             return
+
         try:
-            self.render_templates()
+            # Task was never executed. Initialize RenderedTaskInstanceFields
+            # to render template and mask secrets. Set MASK_SECRETS_IN_LOGS
+            # to True to enable masking similar to task run.
+            original_value = settings.MASK_SECRETS_IN_LOGS
+            settings.MASK_SECRETS_IN_LOGS = True
+            rendered_task_instance = RenderedTaskInstanceFields(self)
+            rendered_fields = rendered_task_instance.rendered_fields
+            if rendered_fields:
+                for field_name, rendered_value in rendered_fields.items():
+                    setattr(self.task, field_name, rendered_value)
         except (TemplateAssertionError, UndefinedError) as e:
             raise AirflowException(
                 "Webserver does not have access to User-defined Macros or 
Filters "
@@ -2103,6 +2113,8 @@ class TaskInstance(Base, LoggingMixin):
                 "started running, please use 'airflow tasks render' for 
debugging the "
                 "rendering of template_fields."
             ) from e
+        finally:
+            settings.MASK_SECRETS_IN_LOGS = original_value
 
     @provide_session
     def get_rendered_k8s_spec(self, session=NEW_SESSION):
diff --git a/tests/www/views/test_views_rendered.py 
b/tests/www/views/test_views_rendered.py
index 129baf7c0e..31935221c0 100644
--- a/tests/www/views/test_views_rendered.py
+++ b/tests/www/views/test_views_rendered.py
@@ -20,7 +20,7 @@ from urllib.parse import quote_plus
 
 import pytest
 
-from airflow.models import DAG, RenderedTaskInstanceFields
+from airflow.models import DAG, RenderedTaskInstanceFields, Variable
 from airflow.operators.bash import BashOperator
 from airflow.serialization.serialized_objects import SerializedDAG
 from airflow.utils import timezone
@@ -61,6 +61,15 @@ def task2(dag):
     )
 
 
+@pytest.fixture()
+def task_secret(dag):
+    return BashOperator(
+        task_id='task_secret',
+        bash_command='echo {{ var.value.my_secret }} && echo {{ var.value.spam 
}}',
+        dag=dag,
+    )
+
+
 @pytest.fixture(scope="module", autouse=True)
 def init_blank_db():
     """Make sure there are no runs before we test anything.
@@ -73,7 +82,7 @@ def init_blank_db():
 
 
 @pytest.fixture(autouse=True)
-def reset_db(dag, task1, task2):
+def reset_db(dag, task1, task2, task_secret):
     yield
     clear_db_dags()
     clear_db_runs()
@@ -81,7 +90,7 @@ def reset_db(dag, task1, task2):
 
 
 @pytest.fixture()
-def create_dag_run(dag, task1, task2):
+def create_dag_run(dag, task1, task2, task_secret):
     def _create_dag_run(*, execution_date, session):
         dag_run = dag.create_dagrun(
             state=DagRunState.RUNNING,
@@ -94,6 +103,8 @@ def create_dag_run(dag, task1, task2):
         ti1.state = TaskInstanceState.SUCCESS
         ti2 = dag_run.get_task_instance(task2.task_id, session=session)
         ti2.state = TaskInstanceState.SCHEDULED
+        ti3 = dag_run.get_task_instance(task_secret.task_id, session=session)
+        ti3.state = TaskInstanceState.QUEUED
         session.flush()
         return dag_run
 
@@ -168,3 +179,29 @@ def 
test_user_defined_filter_and_macros_raise_error(admin_client, create_dag_run
     # MarkupSafe changed the exception detail from 'no filter named' to
     # 'No filter named' in 2.0 (I think), so we normalize for comparison.
     assert "originalerror: no filter named &#39;hello&#39;" in 
resp_html.lower()
+
+
+@pytest.mark.usefixtures("patch_app")
+def test_rendered_template_secret(admin_client, create_dag_run, task_secret):
+    """Test that the Rendered View masks values retrieved from secret 
variables."""
+    Variable.set("my_secret", "foo")
+    Variable.set("spam", "egg")
+
+    assert task_secret.bash_command == 'echo {{ var.value.my_secret }} && echo 
{{ var.value.spam }}'
+
+    with create_session() as session:
+        dag_run = create_dag_run(execution_date=DEFAULT_DATE, session=session)
+        ti = dag_run.get_task_instance(task_secret.task_id, session=session)
+        assert ti is not None, "task instance not found"
+        ti.refresh_from_task(task_secret)
+        assert ti.state == TaskInstanceState.QUEUED
+
+    date = quote_plus(str(DEFAULT_DATE))
+    url = 
f'rendered-templates?task_id=task_secret&dag_id=testdag&execution_date={date}'
+
+    resp = admin_client.get(url, follow_redirects=True)
+    check_content_in_response(
+        'echo</span> *** <span class="o">&amp;&amp;</span> <span 
class="nb">echo</span> egg', resp
+    )
+    ti.refresh_from_task(task_secret)
+    assert ti.state == TaskInstanceState.QUEUED