Redact secrets in rendered templates properly when truncating it

[amoghrajesh]

Problem & Solution

When rendered template fields in a dag exceed the max_templated_field_length (default is 4096 characters), data could be exposed in cleartext in the Rendered Templates UI. This occurred because serialization of those fields while sending to API server used a secret masker instance without patterns registered via mask_secret() calls (ie, it used core secrets masker but patterns were registered against sdk secrets masker). This was a ripple effect of moving secrets masker onto the shared library: #54449

The fix is to create a local _serialize_template_field() function in task_runner that uses the SDK secrets masker, which contains patterns registered via mask_secret() calls in user dag code. This ensures secrets are properly masked before truncation occurs.

This fix is anyways in line with our 3.2 goal of client server separation by getting rid of the serialize_template_field import from airflow core in task sdk.

Testing

Defined a macro via a plugin like this:

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


def mask_str(var: str):
    """
    Takes a string and masks it.
    Returns the same string for templating.
    """
    mask_secret(var)
    return var


class TruncMacrosPlugin(AirflowPlugin):
    name = "trunc_macro"
    macros = [mask_str]

DAG:

import json
from datetime import datetime

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

with DAG('trunc_dag', start_date=datetime(2024, 1, 1), schedule=None) as dag:
    env = {}
    secret = "postgresql+psycopg2://username:testpass123@host/db"
    for i in range(70):
        env[f"URL_{i}"] = f"{{{{ macros.trunc_macro.mask_str('{secret}') }}}}"

    print("len of env is", len(json.dumps(env, indent=4)))

    task = BashOperator(
        task_id="example",
        bash_command="echo done",
        env=env,
    )

Before:

After:

---

^ 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]

Ah forgot to add the backport label!

[amoghrajesh]

CP here: #59704