Redact secrets in rendered templates properly to not expose them on UI

[amoghrajesh]

Problem

Secrets registered via mask_secret() on the worker / dag processor (the ones those use task sdk secrets masker) are correctly masked in task logs but exposed in as clear text in the Rendered Templates UI.

The root cause for this is that due to the recent move to move secrets masker to a shared library and using vendoring per distribution: #54449, the task sdk uses a task sdk secrets masker is recognised differently from the core secrets masker by python even though they are logically the same. Python sees different import paths (airflow.sdk. vs airflow.and assumes that the maskers are different)

The flow leading for logs vs rendered templates:

1. Task process calls mask_secret() -> sends MaskSecret message to supervisor
2. Supervisor registers the pattern in its SecretsMasker instance
3. Logs: Flow through supervisor's SecretsMasker.filter() an are masked
4. Rendered templates: Sent via SetRenderedFields message and supervisor forwards to API server which is stored in the database.
5. When UI tries to read it to display on the UI from the database, the masker that is used to redact it is the one in airflow core, hence the patterns registered with sdk masker are not recognised and masking fails.

Solution

Since the supervisor already has all secret patterns (from MaskSecret messages) but wasn't applying them to rendered template fields before sending to the API server, the solution is to redact at the worker end itself and send it across to the API server. This is the only way to do it and should not cause any ripple effects too because rendered templates are ONLY used to display to the user, and nothing else.

Test

Test 1: Running a Dag with custom operator with templated fields

Dag:

from __future__ import annotations

from datetime import datetime

from airflow.decorators import dag
from airflow.models import BaseOperator
from airflow.sdk.log import mask_secret

mask_secret("password")


class CustomOperator(BaseOperator):
    template_fields = ("tup", "se")
    def __init__(self, tup: tuple, se: set, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.tup = tup
        self.se = se

    def execute(self, context):
        print(self.tup, self.se)



@dag(
    catchup=False,
    tags=[],
    schedule=None,
    max_active_runs=1,
    start_date=datetime(2021, 1, 1),
)
def hello_dag():
    CustomOperator(
        task_id="custom",
        tup=(1, 2, "{{ds}}"),
        se={1, 2, 3, "password"},
    )

hello_dag()

Earlier:

Logs:

Rendered fields:

After the fix:

Logs:

Rendered Fields:

Test 2: Using rendered templates with macros

Plugin:

from airflow.plugins_manager import AirflowPlugin
from airflow.models import Variable
from airflow.sdk.log import mask_secret

def get_masked_var(name: str):
    value = Variable.get(name)
    mask_secret(value)
    return value

class TestPlugin(AirflowPlugin):
    name = "test_plugin"
    macros = [get_masked_var]

Set airflow variable using:

airflow variables set my_var "secret123"

DAG:

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
from datetime import datetime

with DAG('macros_and_rtif', start_date=datetime(2024, 1, 1), schedule=None) as dag:
    BashOperator(
        task_id='test',
        bash_command='echo {{ macros.test_plugin.get_masked_var("my_var") }}'
    )

Before:

Logs:

Rendered Templates:

After:

Logs:

Rendered Templates:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

Thanks for the review, merging it.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 08e8714 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

[amoghrajesh]

Manual cherry pick here: #58772