Redact well-known secret formats in logs and rendered fields

airflow-core/docs/security/secrets/mask-sensitive-values.rst

@@ -113,6 +113,48 @@ or
 
 The mask must be set before any log/output is produced to have any effect.
 
+Content-based masking of well-known secret formats
+""""""""""""""""""""""""""""""""""""""""""""""""""
+
+.. versionadded:: 3.2.0
+
+Registering secrets explicitly via ``mask_secret`` (or through Connections and Variables) only
+covers values Airflow was told about. Credentials that end up in Task logs or Rendered fields via
+other paths — a debug ``print`` of an environment variable, a stack trace containing a token, a
+value pulled from an XCom — are not covered by that mechanism.
+
+To catch those cases, Airflow can additionally scan every string that passes through the secrets
+masker for a small, curated set of well-known credential formats and redact any match. The set
+is intentionally narrow — each entry has a distinctive prefix so a match is overwhelmingly
+likely to be a real credential:
+
+* AWS access / session keys (``AKIA…``, ``ASIA…``)
+* GitHub tokens (``ghp_…``, ``gho_…``, ``ghu_…``, ``ghs_…``, ``ghr_…``)
+* Slack tokens (``xoxb-…``, ``xoxp-…``, ``xoxa-…``, ``xoxr-…``, ``xoxs-…``)
+* Google API keys (``AIza…``)
+* Stripe live keys (``sk_live_…``)
+* PEM-encoded private key blocks (``-----BEGIN … PRIVATE KEY-----``)
+* JSON Web Tokens with the standard ``eyJ…`` header + payload prefix
+
+This is a **defense-in-depth** measure that complements, but does not replace, explicit masking:
+formats with high false-positive rates (generic credit-card numbers, SSNs, email addresses) are
+deliberately excluded, and matches only fire on values that actually flow through the masker.
+Secrets you already know about should still be registered via ``mask_secret``.
+
+The feature is opt-in because the regex scan runs on every string that passes through the
+masker. Enable it in your Airflow config:
+
+.. code-block:: ini
+
+    [core]
+    mask_secrets_content_patterns = True
+
+or via the corresponding environment variable
+``AIRFLOW__CORE__MASK_SECRETS_CONTENT_PATTERNS=True``.
+
+When enabled, log records and redacted values containing e.g. ``AKIAIOSFODNN7EXAMPLE`` are
+rewritten so that only ``***`` appears in the output, while the surrounding text is preserved.
+
 NOT masking when using environment variables
 """"""""""""""""""""""""""""""""""""""""""""
 

airflow-core/src/airflow/config_templates/config.yml

@@ -454,6 +454,18 @@ core:
       type: string
       example: ~
       default: ""
+    mask_secrets_content_patterns:
+      description: |
+        If set to ``True``, Airflow scans string values in Task logs and Rendered fields for a small,
+        curated set of well-known credential formats (AWS access keys, GitHub tokens, Slack tokens,
+        Google API keys, Stripe live keys, PEM-encoded private key blocks, JWTs) and redacts any
+        match. This is a defense-in-depth measure that complements — but does not replace —
+        registering secrets explicitly via ``mask_secret`` or through Connections/Variables. It
+        is opt-in because the regex scan runs on every string that passes through the masker.
+      version_added: 3.2.0
+      type: boolean
+      example: ~
+      default: "False"
     default_pool_task_slot_count:
       description: |
         Task Slot counts for ``default_pool``. This setting would not have any effect in an existing

airflow-core/src/airflow/settings.py

@@ -698,12 +698,14 @@ def _configure_secrets_masker():
         sensitive_fields |= frozenset({field.strip() for field in sensitive_variable_fields.split(",")})
 
     hide_sensitive_var_conn_fields = conf.getboolean("core", "hide_sensitive_var_conn_fields")
+    mask_content_patterns = conf.getboolean("core", "mask_secrets_content_patterns", fallback=False)
 
     core_masker = secrets_masker_core()
     core_masker.min_length_to_mask = min_length_to_mask
     core_masker.sensitive_variables_fields = list(sensitive_fields)
     core_masker.secret_mask_adapter = secret_mask_adapter
     core_masker.hide_sensitive_var_conn_fields = hide_sensitive_var_conn_fields
+    core_masker.mask_content_patterns = mask_content_patterns
 
     from airflow.sdk._shared.secrets_masker import _secrets_masker as sdk_secrets_masker
 
@@ -712,6 +714,7 @@ def _configure_secrets_masker():
     sdk_masker.sensitive_variables_fields = list(sensitive_fields)
     sdk_masker.secret_mask_adapter = secret_mask_adapter
     sdk_masker.hide_sensitive_var_conn_fields = hide_sensitive_var_conn_fields
+    sdk_masker.mask_content_patterns = mask_content_patterns
 
 
 def configure_action_logging() -> None:

shared/secrets_masker/src/airflow_shared/secrets_masker/__init__.py

@@ -18,6 +18,7 @@
 
 from .secrets_masker import (
     DEFAULT_SENSITIVE_FIELDS,
+    KNOWN_SECRET_PATTERNS,
     Redactable,
     Redacted,
     RedactedIO,
@@ -42,6 +43,7 @@
     "should_hide_value_for_key",
     "_secrets_masker",
     "DEFAULT_SENSITIVE_FIELDS",
+    "KNOWN_SECRET_PATTERNS",
     "Redactable",
     "Redacted",
 ]

shared/secrets_masker/src/airflow_shared/secrets_masker/secrets_masker.py

@@ -79,6 +79,42 @@ def to_dict(self) -> dict[str, Any]: ...
 SECRETS_TO_SKIP_MASKING = {"airflow"}
 """Common terms that should be excluded from masking in both production and tests"""
 
+KNOWN_SECRET_PATTERNS: dict[str, str] = {
+    # Word-boundary lookarounds keep AWS keys from matching inside longer
+    # uppercase runs (e.g. an unrelated 20-char identifier that happens to
+    # start with "AKIA").
+    "aws_access_key": r"(?<![A-Z0-9])(?:AKIA|ASIA)[0-9A-Z]{16}(?![A-Z0-9])",
+    "github_token": r"\bgh[pousr]_[A-Za-z0-9]{36,255}\b",
+    "slack_token": r"\bxox[baprs]-[A-Za-z0-9-]{10,255}\b",
+    "google_api_key": r"\bAIza[0-9A-Za-z_\-]{35}\b",
+    "stripe_live_key": r"\bsk_live_[0-9A-Za-z]{24,64}\b",
+    # Match the full PEM block, including header/footer, so the entire
+    # key material is redacted rather than only the label line. Body is
+    # bounded to keep the regex engine's work strictly linear on any input.
+    "pem_private_key_block": (
+        r"-----BEGIN (?:RSA |EC |OPENSSH |DSA |PGP )?PRIVATE KEY-----"
+        r"[\s\S]{1,16384}?-----END (?:RSA |EC |OPENSSH |DSA |PGP )?PRIVATE KEY-----"
+    ),
+    # Require both header and payload segments to start with "eyJ" (base64
+    # of `{"`) so we don't match arbitrary dotted base64-looking blobs. Each
+    # segment is bounded so an adversarial input cannot force unbounded work.
+    "jwt": (
+        r"\beyJ[A-Za-z0-9_\-]{10,4096}"
+        r"\.eyJ[A-Za-z0-9_\-]{10,4096}"
+        r"\.[A-Za-z0-9_\-]{10,4096}\b"
+    ),
+}
+"""Regexes for well-known secret formats detected by value.
+
+The set is intentionally narrow: each entry has a distinctive prefix
+(``AKIA``/``ASIA``, ``gh[pousr]_``, ``xox[baprs]-``, ``AIza``, ``sk_live_``,
+``-----BEGIN … PRIVATE KEY-----``, ``eyJ…eyJ…``) so a match is overwhelmingly
+likely to be a real credential. Formats with high false-positive rates
+(generic credit-card, SSN, email) are deliberately excluded from this
+list — deployments that want them can register their own patterns via
+:meth:`SecretsMasker.add_content_patterns`.
+"""
+
 
 def should_hide_value_for_key(name):
     """
@@ -196,6 +232,7 @@ class SecretsMasker(logging.Filter):
     MAX_RECURSION_DEPTH = 5
     _has_warned_short_secret = False
     mask_secrets_in_logs = False
+    mask_content_patterns = False
 
     min_length_to_mask = 5
     secret_mask_adapter = None
@@ -205,6 +242,8 @@ def __init__(self):
         self.patterns = set()
         self.sensitive_variables_fields = []
         self.hide_sensitive_var_conn_fields = True
+        self._content_pattern_sources: dict[str, str] = dict(KNOWN_SECRET_PATTERNS)
+        self._content_pattern_replacer: Pattern | None = None
 
     @classmethod
     def __init_subclass__(cls, **kwargs):
@@ -244,6 +283,58 @@ def is_log_masking_enabled(cls) -> bool:
         """Check if secret masking in logs is enabled."""
         return cls.mask_secrets_in_logs
 
+    @classmethod
+    def enable_content_pattern_masking(cls) -> None:
+        """Enable value-content pattern masking (well-known secret formats)."""
+        cls.mask_content_patterns = True
+
+    @classmethod
+    def disable_content_pattern_masking(cls) -> None:
+        """Disable value-content pattern masking."""
+        cls.mask_content_patterns = False
+
+    @classmethod
+    def is_content_pattern_masking_enabled(cls) -> bool:
+        """Check if value-content pattern masking is enabled."""
+        return cls.mask_content_patterns
+
+    def add_content_patterns(self, patterns: dict[str, str]) -> None:
+        """
+        Register additional named regex patterns for value-content masking.
+
+        Keys are pattern names (used for diagnostics), values are regex
+        source strings. Existing entries with the same name are replaced.
+        Invalid regexes are skipped with a warning rather than raising, so
+        a misconfigured deployment does not disable the whole masker.
+        """
+        changed = False
+        for name, source in patterns.items():
+            try:
+                re.compile(source)
+            except re.error as exc:
+                log.warning(
+                    "Skipping invalid content-mask pattern %r: %s",
+                    name,
+                    exc,
+                    extra={self.ALREADY_FILTERED_FLAG: True},
+                )
+                continue
+            self._content_pattern_sources[name] = source
+            changed = True
+        if changed:
+            self._content_pattern_replacer = None
+
+    def _get_content_pattern_replacer(self) -> Pattern | None:
+        """Return the compiled union of registered content patterns, or ``None`` if empty."""
+        if self._content_pattern_replacer is not None:
+            return self._content_pattern_replacer
+        sources = list(self._content_pattern_sources.values())
+        if not sources:
+            return None
+        combined = "|".join(f"(?:{src})" for src in sources)
+        self._content_pattern_replacer = re.compile(combined)
+        return self._content_pattern_replacer
+
     @cached_property
     def _record_attrs_to_ignore(self) -> Iterable[str]:
         # Doing log.info(..., extra={'foo': 2}) sets extra properties on
@@ -307,7 +398,9 @@ def filter(self, record) -> bool:
             # "private" flag that stops us needing to process it more than once
             return True
 
-        if self.replacer:
+        # Redact when either explicit masks are registered or value-content
+        # pattern masking is enabled — otherwise there is nothing to look for.
+        if self.replacer or self.mask_content_patterns:
             for k, v in record.__dict__.items():
                 if k not in self._record_attrs_to_ignore:
                     record.__dict__[k] = self.redact(v)
@@ -408,12 +501,20 @@ def _redact(
                     )
                 return tmp
             if isinstance(item, str):
+                content_replacer = (
+                    self._get_content_pattern_replacer() if self.mask_content_patterns else None
+                )
+                if not self.replacer and content_replacer is None:
+                    return item
+                text = str(item)
                 if self.replacer:
                     # We can't replace specific values, but the key-based redacting
                     # can still happen, so we can't short-circuit, we need to walk
                     # the structure.
-                    return self.replacer.sub(replacement, str(item))
-                return item
+                    text = self.replacer.sub(replacement, text)
+                if content_replacer is not None:
+                    text = content_replacer.sub(replacement, text)
+                return text
             return item
         # I think this should never happen, but it does not hurt to leave it just in case
         # Well. It happened (see https://github.com/apache/airflow/issues/19816#issuecomment-983311373)
@@ -634,6 +735,8 @@ def reset_masker(self):
         """Reset the patterns and the replacer in the masker instance."""
         self.patterns = set()
         self.replacer = None
+        self._content_pattern_sources = dict(KNOWN_SECRET_PATTERNS)
+        self._content_pattern_replacer = None
 
 
 class RedactedIO(TextIO):