From 863d632ee4260997efb69aba76518ea8b0717590 Mon Sep 17 00:00:00 2001 From: Deepak Kumar Date: Sun, 2 Aug 2026 17:28:58 -0700 Subject: [PATCH] Redact well-known secret formats in logs and rendered fields Airflow's SecretsMasker decides what to hide by key name, not by the shape of the value. As the current docs acknowledge, keys in a Connection extra whose names do not match a sensitive keyword are not redacted, and values that flow through XCom or any other side channel reach Task logs and Rendered fields as-is even on a correctly configured deployment. Users routinely leak live credentials this way by printing environment variables during debugging, logging stack traces that include a token or PEM block, or pushing values through XCom keys whose names do not happen to match the sensitive-keyword list. Registering secrets ahead of time via mask_secret(), Connections, and Variables only covers values Airflow was told about. This adds a second line of defense so that even when Airflow was never told a specific value was sensitive, values that match a small curated set of well-known credential formats are redacted before they reach logs or rendered fields. The feature is opt-in (default False) so no existing deployment changes behavior. The initial pattern set is deliberately narrow - each entry has a distinctive fixed prefix so a match is overwhelmingly likely to be a real credential, and formats with high false-positive rates in log data are left out of the built-ins. Deployments that want additional formats can register them via add_content_patterns() rather than editing Airflow. All patterns use bounded or fixed-width bodies so the regex engine's work stays strictly linear on any input. related: #58514 --- .../secrets/mask-sensitive-values.rst | 42 ++++ .../src/airflow/config_templates/config.yml | 12 ++ airflow-core/src/airflow/settings.py | 3 + .../airflow_shared/secrets_masker/__init__.py | 2 + .../secrets_masker/secrets_masker.py | 109 ++++++++++- .../secrets_masker/test_secrets_masker.py | 182 ++++++++++++++++++ 6 files changed, 347 insertions(+), 3 deletions(-) diff --git a/airflow-core/docs/security/secrets/mask-sensitive-values.rst b/airflow-core/docs/security/secrets/mask-sensitive-values.rst index 780aceb70fbc6..30c45496f6e01 100644 --- a/airflow-core/docs/security/secrets/mask-sensitive-values.rst +++ b/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 """""""""""""""""""""""""""""""""""""""""""" diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index c065b277716c4..a0d634772ad93 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/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 diff --git a/airflow-core/src/airflow/settings.py b/airflow-core/src/airflow/settings.py index 0a0e1c6619726..bf33a5c12f3b3 100644 --- a/airflow-core/src/airflow/settings.py +++ b/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: diff --git a/shared/secrets_masker/src/airflow_shared/secrets_masker/__init__.py b/shared/secrets_masker/src/airflow_shared/secrets_masker/__init__.py index 2eb43c030db9e..9c3890cccbb90 100644 --- a/shared/secrets_masker/src/airflow_shared/secrets_masker/__init__.py +++ b/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", ] diff --git a/shared/secrets_masker/src/airflow_shared/secrets_masker/secrets_masker.py b/shared/secrets_masker/src/airflow_shared/secrets_masker/secrets_masker.py index 9157a91a7fb25..d6478c5fd85e0 100644 --- a/shared/secrets_masker/src/airflow_shared/secrets_masker/secrets_masker.py +++ b/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"(? 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): diff --git a/shared/secrets_masker/tests/secrets_masker/test_secrets_masker.py b/shared/secrets_masker/tests/secrets_masker/test_secrets_masker.py index b3b40475a2050..dda5eb800070c 100644 --- a/shared/secrets_masker/tests/secrets_masker/test_secrets_masker.py +++ b/shared/secrets_masker/tests/secrets_masker/test_secrets_masker.py @@ -31,6 +31,7 @@ from airflow_shared.secrets_masker.secrets_masker import ( DEFAULT_SENSITIVE_FIELDS, + KNOWN_SECRET_PATTERNS, RedactedIO, SecretsMasker, mask_secret, @@ -1611,3 +1612,184 @@ def test_k8s_objects_still_detected_when_imported(self): # Should be redacted since "password" is a sensitive field name assert redacted["value"] == "***" assert redacted["name"] == "password" + + +class TestContentPatternMasking: + """Value-content pattern masking for well-known credential formats.""" + + @pytest.fixture + def masker(self): + m = SecretsMasker() + configure_secrets_masker_for_test(m) + return m + + @pytest.mark.parametrize( + ("label", "sample"), + [ + ("aws_access_key", "AKIAIOSFODNN7EXAMPLE"), + ("aws_session_key", "ASIAY34FZKBOKMUTVV7A"), + ("github_pat", "ghp_" + "a" * 40), + ("github_oauth", "gho_" + "b" * 40), + ("slack_bot", "xoxb-1234567890-abcdefghij"), + ("slack_user", "xoxp-1234567890-0987654321-abcdefghij"), + ("google_api_key", "AIza" + "A" * 35), + ("stripe_live_key", "sk_live_" + "0" * 24), + ( + "jwt", + "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U", + ), + ], + ) + def test_known_pattern_is_redacted_when_enabled(self, masker, label, sample): + masker.enable_content_pattern_masking() + try: + text = f"prefix {sample} suffix" + assert masker.redact(text) == "prefix *** suffix" + finally: + masker.disable_content_pattern_masking() + + def test_pem_private_key_block_is_redacted_when_enabled(self, masker): + masker.enable_content_pattern_masking() + try: + pem = ( + "-----BEGIN RSA PRIVATE KEY-----\n" + "MIIBOgIBAAJBAKj34GkxFhD90vcNLYLInFEX6Ppy1tPf9Cnzj4p4WGeKLs1Pt8Qu\n" + "-----END RSA PRIVATE KEY-----" + ) + redacted = masker.redact(f"key: {pem} end") + assert redacted == "key: *** end" + finally: + masker.disable_content_pattern_masking() + + def test_off_by_default(self, masker): + # By default, with a fresh masker and no explicit enable, an AWS-shaped + # value must pass through unchanged. + assert not masker.is_content_pattern_masking_enabled() + aws = "AKIAIOSFODNN7EXAMPLE" + assert masker.redact(aws) == aws + + def test_disable_restores_passthrough(self, masker): + masker.enable_content_pattern_masking() + aws = "AKIAIOSFODNN7EXAMPLE" + assert masker.redact(aws) == "***" + masker.disable_content_pattern_masking() + assert masker.redact(aws) == aws + + @pytest.mark.parametrize( + "benign", + [ + "AKIA_LOOKS_LIKE_ONE", + "not-a-jwt.header.only", + "sk_live_short", + "gh_notatokentype_prefix", + "xox-not-a-slack-token", + "just some normal log line with no secrets in it at all", + ], + ) + def test_benign_strings_are_not_redacted(self, masker, benign): + masker.enable_content_pattern_masking() + try: + assert masker.redact(benign) == benign + finally: + masker.disable_content_pattern_masking() + + def test_content_masking_composes_with_key_name_masking(self, masker): + # A dict whose key name is sensitive triggers the existing recursive + # redaction; content-pattern masking on top must not regress that path. + masker.enable_content_pattern_masking() + try: + data = {"password": "some_user_password", "log_line": "token=AKIAIOSFODNN7EXAMPLE end"} + redacted = masker.redact(data) + assert redacted["password"] == "***" + assert redacted["log_line"] == "token=*** end" + finally: + masker.disable_content_pattern_masking() + + def test_content_masking_composes_with_explicit_add_mask(self, masker): + # A secret registered via add_mask() must still be redacted alongside + # a same-string pattern-detected secret in the same value. + masker.enable_content_pattern_masking() + try: + masker.add_mask("my-custom-secret-1234") + text = "my-custom-secret-1234 and gh" + "p_" + "z" * 40 + redacted = masker.redact(text) + assert "my-custom-secret-1234" not in redacted + assert "ghp_" not in redacted + assert redacted.count("***") == 2 + finally: + masker.disable_content_pattern_masking() + + def test_add_content_patterns_registers_and_masks(self, masker): + masker.enable_content_pattern_masking() + try: + masker.add_content_patterns({"acme_key": r"\bACME-[A-Z0-9]{8}\b"}) + assert masker.redact("id=ACME-ABCD1234 done") == "id=*** done" + finally: + masker.disable_content_pattern_masking() + + def test_add_content_patterns_ignores_invalid_regex(self, masker, caplog): + # Invalid regex must not raise or break the existing pattern set. + masker.enable_content_pattern_masking() + try: + masker.add_content_patterns({"broken": "([unterminated"}) + # Known patterns still work after a bad entry was rejected. + assert masker.redact("AKIAIOSFODNN7EXAMPLE") == "***" + finally: + masker.disable_content_pattern_masking() + + def test_reset_masker_restores_default_content_patterns(self, masker): + masker.enable_content_pattern_masking() + try: + masker.add_content_patterns({"custom_x": r"\bCUSTOMX-[0-9]{4}\b"}) + assert masker.redact("CUSTOMX-1234") == "***" + masker.reset_masker() + # Custom pattern gone. + assert masker.redact("CUSTOMX-1234") == "CUSTOMX-1234" + # Built-in defaults restored. + assert masker.redact("AKIAIOSFODNN7EXAMPLE") == "***" + finally: + masker.disable_content_pattern_masking() + + def test_default_pattern_set_matches_known_secret_patterns_constant(self, masker): + # Guardrail so a rename of the module-level constant does not silently + # drop the default set from newly-constructed maskers. + assert set(masker._content_pattern_sources) == set(KNOWN_SECRET_PATTERNS) + + def test_log_filter_masks_content_pattern_without_registered_secret(self, caplog): + # The log filter previously short-circuited when no explicit masks + # were registered; with content-pattern masking enabled it must run. + logging.config.dictConfig( + { + "version": 1, + "handlers": { + __name__: { + "class": "logging.StreamHandler", + "stream": "ext://sys.stdout", + } + }, + "loggers": { + __name__: { + "handlers": [__name__], + "level": logging.INFO, + "propagate": False, + } + }, + "disable_existing_loggers": False, + } + ) + formatter = logging.Formatter("%(levelname)s %(message)s") + logger = logging.getLogger(__name__) + caplog.handler.setFormatter(formatter) + logger.handlers = [caplog.handler] + + filt = SecretsMasker() + configure_secrets_masker_for_test(filt) + filt.enable_content_pattern_masking() + SecretsMasker.enable_log_masking() + logger.addFilter(filt) + try: + logger.info("token=AKIAIOSFODNN7EXAMPLE end") + assert caplog.text == "INFO token=*** end\n" + finally: + filt.disable_content_pattern_masking() + SecretsMasker.disable_log_masking()