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 self.sensitive_config_values:
+            try:
+                value = self.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,18 @@ def test_config_paths_is_directory(self):
 
         with pytest.raises(IsADirectoryError, match="configuration file, but got a directory"):
             write_default_airflow_configuration_if_needed()
+
+    @conf_vars({("mysection1", "mykey1"): "supersecret1", ("mysection2", "mykey2"): "supersecret2"})
+    @patch.object(
+        conf,
+        "sensitive_config_values",
+        new_callable=lambda: [("mysection1", "mykey1"), ("mysection2", "mykey2")],
+    )
+    @patch("airflow.utils.log.secrets_masker.mask_secret")
+    def test_mask_conf_values(self, mock_mask_secret, mock_sensitive_config_values):
+        conf.mask_secrets()
+
+        mock_mask_secret.assert_any_call("supersecret1")
+        mock_mask_secret.assert_any_call("supersecret2")
+
+        assert mock_mask_secret.call_count == 2