Fix AirflowSDKConfigParser missing mask_secrets method

task-sdk/src/airflow/sdk/configuration.py

@@ -161,6 +161,35 @@ def __init__(
         if default_config is not None:
             self._update_defaults_from_string(default_config)
 
+    def mask_secrets(self) -> None:
+        """Mask sensitive config values in the secrets masker."""
+        from airflow.sdk._shared.configuration.exceptions import AirflowConfigException
+        from airflow.sdk._shared.configuration.parser import _build_kwarg_env_prefix, _collect_kwarg_env_vars
+        from airflow.sdk._shared.secrets_masker import mask_secret
+
+        for section, key in self.sensitive_config_values:
+            try:
+                with self.suppress_future_warnings():
+                    value = self.get(section, key, suppress_warnings=True)
+            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)
+
+        # Mask per-key backend kwarg env vars (AIRFLOW__SECRETS__BACKEND_KWARG__* etc.).
+        # These are not in sensitive_config_values but may contain sensitive values.
+        for _section, _kwargs_key in [
+            ("secrets", "backend_kwargs"),
+            ("workers", "secrets_backend_kwargs"),
+        ]:
+            _prefix = _build_kwarg_env_prefix(_section, _kwargs_key)
+            for _value in _collect_kwarg_env_vars(_prefix).values():
+                mask_secret(_value)
+
     def _get_custom_secret_backend(self, worker_mode: bool | None = None) -> Any | None:
         return super()._get_custom_secret_backend(
             worker_mode=worker_mode if worker_mode is not None else True