Use single source of truth for sensitive config items

airflow/config_templates/config.yml

@@ -434,6 +434,7 @@ core:
       description: Kwargs to supply to dataset manager.
       version_added: 2.4.0
       type: string
+      sensitive: true
       default: ~
       example: '{"some_param": "some_value"}'
     database_access_isolation:
@@ -662,6 +663,7 @@ logging:
         log files will be deleted after they are uploaded to remote location.
       version_added: 2.6.0
       type: string
+      sensitive: true
       example: '{"delete_local_copy": true}'
       default: ""
     encrypt_s3_logs:
@@ -1052,6 +1054,7 @@ secrets:
         ``{{"connections_prefix": "/airflow/connections", "profile_name": "default"}}``
       version_added: 1.10.10
       type: string
+      sensitive: true
       example: ~
       default: ""
 cli:
@@ -1929,6 +1932,7 @@ sentry:
       description: ~
       version_added: 1.10.6
       type: string
+      sensitive: true
       example: ~
       default: ""
     before_send:
@@ -2197,6 +2201,7 @@ celery_broker_transport_options:
         https://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/redis.html#configuration
       version_added: 2.7.0
       type: string
+      sensitive: true
       example: '{"password": "password_for_redis_server"}'
       default: ~
 dask:

airflow/configuration.py

@@ -37,7 +37,7 @@
 from contextlib import contextmanager, suppress
 from json.decoder import JSONDecodeError
 from re import Pattern
-from typing import IO, Any, Dict, Iterable, Tuple, Union
+from typing import IO, Any, Dict, Iterable, Set, Tuple, Union
 from urllib.parse import urlsplit
 
 from typing_extensions import overload
@@ -146,21 +146,6 @@ def default_config_yaml() -> dict[str, Any]:
         return yaml.safe_load(config_file)
 
 
-SENSITIVE_CONFIG_VALUES = {
-    ("database", "sql_alchemy_conn"),
-    ("core", "fernet_key"),
-    ("celery", "broker_url"),
-    ("celery", "flower_basic_auth"),
-    ("celery", "result_backend"),
-    ("atlas", "password"),
-    ("smtp", "smtp_password"),
-    ("webserver", "secret_key"),
-    ("secrets", "backend_kwargs"),
-    # The following options are deprecated
-    ("core", "sql_alchemy_conn"),
-}
-
-
 class AirflowConfigParser(ConfigParser):
     """Custom Airflow Configparser supporting defaults and deprecated options."""
 
@@ -170,7 +155,19 @@ class AirflowConfigParser(ConfigParser):
     # These configs can also be fetched from Secrets backend
     # following the "{section}__{name}__secret" pattern
 
-    sensitive_config_values: set[tuple[str, str]] = SENSITIVE_CONFIG_VALUES
+    @functools.cached_property
+    def sensitive_config_values(self) -> Set[tuple[str, str]]:  # noqa: UP006
+        default_config = default_config_yaml()
+        flattened = {
+            (s, k): item for s, s_c in default_config.items() for k, item in s_c.get("options").items()
+        }
+        sensitive = {(section, key) for (section, key), v in flattened.items() if v.get("sensitive") is True}
+        depr_option = {self.deprecated_options[x][:-1] for x in sensitive if x in self.deprecated_options}
+        depr_section = {
+            (self.deprecated_sections[s][0], k) for s, k in sensitive if s in self.deprecated_sections
+        }
+        sensitive.update(depr_section, depr_option)
+        return sensitive
 
     # A mapping of (new section, new option) -> (old section, old option, since_version).
     # When reading new option, the old option will be checked to see if it exists. If it does a

airflow/www/views.py

@@ -4159,11 +4159,9 @@ def conf(self):
         # TODO remove "if raw" usage in Airflow 3.0. Configuration can be fetched via the REST API.
         if raw:
             if expose_config == "non-sensitive-only":
-                from airflow.configuration import SENSITIVE_CONFIG_VALUES
-
                 updater = configupdater.ConfigUpdater()
                 updater.read(AIRFLOW_CONFIG)
-                for sect, key in SENSITIVE_CONFIG_VALUES:
+                for sect, key in conf.sensitive_config_values:
                     if updater.has_option(sect, key):
                         updater[sect][key].value = "< hidden >"
                 config = str(updater)

tests/core/test_configuration.py

@@ -36,6 +36,7 @@
     AirflowConfigException,
     AirflowConfigParser,
     conf,
+    default_config_yaml,
     expand_env_var,
     get_airflow_config,
     get_airflow_home,
@@ -1447,3 +1448,35 @@ def test_future_warning_only_for_code_ref(self, key):
             w = captured.pop()
             assert "your `conf.get*` call to use the new name" in str(w.message)
             assert w.category == FutureWarning
+
+
+def test_sensitive_values():
+    from airflow.settings import conf
+
+    # this list was hardcoded prior to 2.6.2
+    # included here to avoid regression in refactor
+    # inclusion of keys ending in "password" or "kwargs" is automated from 2.6.2
+    # items not matching this pattern must be added here manually
+    sensitive_values = {
+        ("database", "sql_alchemy_conn"),
+        ("core", "fernet_key"),
+        ("celery", "broker_url"),
+        ("celery", "flower_basic_auth"),
+        ("celery", "result_backend"),
+        ("atlas", "password"),
+        ("smtp", "smtp_password"),
+        ("webserver", "secret_key"),
+        ("secrets", "backend_kwargs"),
+        ("sentry", "sentry_dsn"),
+        ("database", "sql_alchemy_engine_args"),
+        ("core", "sql_alchemy_conn"),
+    }
+    default_config = default_config_yaml()
+    all_keys = {(s, k) for s, v in default_config.items() for k in v.get("options")}
+    suspected_sensitive = {(s, k) for (s, k) in all_keys if k.endswith(("password", "kwargs"))}
+    exclude_list = {
+        ("kubernetes_executor", "delete_option_kwargs"),
+    }
+    suspected_sensitive -= exclude_list
+    sensitive_values.update(suspected_sensitive)
+    assert sensitive_values == conf.sensitive_config_values

tests/www/views/test_views_configuration.py

@@ -18,7 +18,7 @@
 
 import html
 
-from airflow.configuration import SENSITIVE_CONFIG_VALUES, conf
+from airflow.configuration import conf
 from tests.test_utils.config import conf_vars
 from tests.test_utils.www import check_content_in_response, check_content_not_in_response
 
@@ -36,7 +36,7 @@ def test_user_cant_view_configuration(admin_client):
 @conf_vars({("webserver", "expose_config"): "True"})
 def test_user_can_view_configuration(admin_client):
     resp = admin_client.get("configuration", follow_redirects=True)
-    for section, key in SENSITIVE_CONFIG_VALUES:
+    for section, key in conf.sensitive_config_values:
         value = conf.get(section, key, fallback="")
         if not value:
             continue
@@ -46,7 +46,7 @@ def test_user_can_view_configuration(admin_client):
 @conf_vars({("webserver", "expose_config"): "non-sensitive-only"})
 def test_configuration_redacted(admin_client):
     resp = admin_client.get("configuration", follow_redirects=True)
-    for section, key in SENSITIVE_CONFIG_VALUES:
+    for section, key in conf.sensitive_config_values:
         value = conf.get(section, key, fallback="")
         if not value or value == "airflow":
             continue
@@ -58,7 +58,7 @@ def test_configuration_redacted(admin_client):
 @conf_vars({("webserver", "expose_config"): "non-sensitive-only"})
 def test_configuration_redacted_in_running_configuration(admin_client):
     resp = admin_client.get("configuration", follow_redirects=True)
-    for section, key in SENSITIVE_CONFIG_VALUES:
+    for section, key in conf.sensitive_config_values:
         value = conf.get(section, key, fallback="")
         if not value or value == "airflow":
             continue