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()