Masking configuration values irrelevant to DAG author

airflow/configuration.py

@@ -772,6 +772,21 @@ def _create_future_warning(name: str, section: str, current_value: Any, new_valu
             stacklevel=3,
         )
 
+    def mask_secrets(self):
+        from airflow.utils.log.secrets_masker import mask_secret
+
+        for section, key in conf.sensitive_config_values:
+            try:
+                value = conf.get(section, key)
+            except AirflowConfigException:
+                log.debug(
+                    "Could not retrieve value from section %s, for key %s. Skipping redaction of this conf.",
+                    section,
+                    key,
+                )
+                continue
+            mask_secret(value)
+
     def _env_var_name(self, section: str, key: str) -> str:
         return f"{ENV_VAR_PREFIX}{section.replace('.', '_').upper()}__{key.upper()}"
 

airflow/settings.py

@@ -741,6 +741,9 @@ def initialize():
     configure_orm()
     configure_action_logging()
 
+    # mask the sensitive_config_values
+    conf.mask_secrets()
+
     # Run any custom runtime checks that needs to be executed for providers
     run_providers_custom_runtime_checks()
 

tests/core/test_configuration.py

@@ -1763,3 +1763,22 @@ def test_config_paths_is_directory(self):
 
         with pytest.raises(IsADirectoryError, match="configuration file, but got a directory"):
             write_default_airflow_configuration_if_needed()
+
+    @patch.object(
+        conf,
+        "sensitive_config_values",
+        new_callable=lambda: [("mysection1", "mykey1"), ("mysection2", "mykey2")],
+    )
+    @patch("airflow.utils.log.secrets_masker.mask_secret")
+    @patch("airflow.configuration.conf.get")
+    def test_mask_conf_values(self, mock_get, mock_mask_secret, mock_sensitive_config_values):
+        mock_get.side_effect = ["supersecret1", "supersecret2"]
+        conf.mask_secrets()
+
+        mock_get.assert_any_call("mysection1", "mykey1")
+        mock_get.assert_any_call("mysection2", "mykey2")
+        mock_mask_secret.assert_any_call("supersecret1")
+        mock_mask_secret.assert_any_call("supersecret2")
+
+        assert mock_get.call_count == 2
+        assert mock_mask_secret.call_count == 2

tests/core/test_settings.py

@@ -31,7 +31,11 @@
 from airflow.api_internal.internal_api_call import InternalApiConfig
 from airflow.configuration import conf
 from airflow.exceptions import AirflowClusterPolicyViolation, AirflowConfigException
-from airflow.settings import _ENABLE_AIP_44, TracebackSession, is_usage_data_collection_enabled
+from airflow.settings import (
+    _ENABLE_AIP_44,
+    TracebackSession,
+    is_usage_data_collection_enabled,
+)
 from airflow.utils.session import create_session
 
 from tests_common.test_utils.config import conf_vars