AIP-72: Move Secrets Masker to task SDK

.pre-commit-config.yaml

@@ -1223,12 +1223,12 @@ repos:
           ^airflow/serialization/serde.py$ |
           ^airflow/utils/file.py$ |
           ^airflow/utils/helpers.py$ |
-          ^airflow/utils/log/secrets_masker.py$ |
           ^providers/ |
           ^tests/ |
           ^providers/tests/ |
           ^providers/.*/tests/ |
           ^task_sdk/src/airflow/sdk/definitions/dag.py$ |
+          ^task_sdk/src/airflow/sdk/execution_time/secrets_masker.py$ |
           ^task_sdk/src/airflow/sdk/definitions/_internal/node.py$ |
           ^dev/.*\.py$ |
           ^scripts/.*\.py$ |

airflow/api_connexion/schemas/connection_schema.py

@@ -53,7 +53,7 @@ class ConnectionSchema(ConnectionCollectionItemSchema):
     def serialize_extra(obj: Connection):
         if obj.extra is None:
             return
-        from airflow.utils.log.secrets_masker import redact
+        from airflow.sdk.execution_time.secrets_masker import redact
 
         try:
             extra = json.loads(obj.extra)

airflow/api_fastapi/core_api/datamodels/assets.py

@@ -22,7 +22,7 @@
 from pydantic import Field, field_validator
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
-from airflow.utils.log.secrets_masker import redact
+from airflow.sdk.execution_time.secrets_masker import redact
 
 
 class DagScheduleAssetReference(StrictBaseModel):

airflow/api_fastapi/core_api/datamodels/connections.py

@@ -23,7 +23,7 @@
 from pydantic_core.core_schema import ValidationInfo
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
-from airflow.utils.log.secrets_masker import redact
+from airflow.sdk.execution_time.secrets_masker import redact
 
 
 # Response Models

airflow/api_fastapi/core_api/datamodels/variables.py

@@ -23,8 +23,8 @@
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
 from airflow.models.base import ID_LEN
+from airflow.sdk.execution_time.secrets_masker import redact
 from airflow.typing_compat import Self
-from airflow.utils.log.secrets_masker import redact
 
 
 class VariableResponse(BaseModel):

airflow/api_fastapi/logging/decorators.py

@@ -29,7 +29,7 @@
 from airflow.api_fastapi.core_api.security import get_user_with_exception_handling
 from airflow.auth.managers.models.base_user import BaseUser
 from airflow.models import Log
-from airflow.utils.log import secrets_masker
+from airflow.sdk.execution_time import secrets_masker
 
 logger = logging.getLogger(__name__)
 

airflow/cli/commands/remote_commands/task_command.py

@@ -48,6 +48,7 @@
 from airflow.models.dagrun import DagRun
 from airflow.models.taskinstance import TaskReturnCode
 from airflow.sdk.definitions.param import ParamsDict
+from airflow.sdk.execution_time.secrets_masker import RedactedIO
 from airflow.settings import IS_EXECUTOR_CONTAINER, IS_K8S_EXECUTOR_POD
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS
@@ -61,7 +62,6 @@
 )
 from airflow.utils.log.file_task_handler import _set_task_deferred_context_var
 from airflow.utils.log.logging_mixin import StreamLogWriter
-from airflow.utils.log.secrets_masker import RedactedIO
 from airflow.utils.net import get_hostname
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 from airflow.utils.session import NEW_SESSION, create_session, provide_session

airflow/config_templates/airflow_local_settings.py

@@ -71,7 +71,7 @@
     },
     "filters": {
         "mask_secrets": {
-            "()": "airflow.utils.log.secrets_masker.SecretsMasker",
+            "()": "airflow.sdk.execution_time.secrets_masker.SecretsMasker",
         },
     },
     "handlers": {

airflow/config_templates/config.yml

@@ -886,10 +886,10 @@ logging:
     secret_mask_adapter:
       description: |
         An import path to a function to add adaptations of each secret added with
-        ``airflow.utils.log.secrets_masker.mask_secret`` to be masked in log messages. The given function
-        is expected to require a single parameter: the secret to be adapted. It may return a
-        single adaptation of the secret or an iterable of adaptations to each be masked as secrets.
-        The original secret will be masked as well as any adaptations returned.
+        ``airflow.sdk.execution_time.secrets_masker.mask_secret`` to be masked in log messages.
+        The given function is expected to require a single parameter: the secret to be adapted.
+        It may return a single adaptation of the secret or an iterable of adaptations to each be
+        masked as secrets. The original secret will be masked as well as any adaptations returned.
       version_added: 2.6.0
       type: string
       default: ""

airflow/configuration.py

@@ -782,7 +782,7 @@ def _create_future_warning(name: str, section: str, current_value: Any, new_valu
         )
 
     def mask_secrets(self):
-        from airflow.utils.log.secrets_masker import mask_secret
+        from airflow.sdk.execution_time.secrets_masker import mask_secret
 
         for section, key in self.sensitive_config_values:
             try:

airflow/models/connection.py

@@ -32,10 +32,10 @@
 from airflow.exceptions import AirflowException, AirflowNotFoundException
 from airflow.models.base import ID_LEN, Base
 from airflow.models.crypto import get_fernet
+from airflow.sdk.execution_time.secrets_masker import mask_secret
 from airflow.secrets.cache import SecretCache
 from airflow.utils.helpers import prune_dict
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.log.secrets_masker import mask_secret
 from airflow.utils.module_loading import import_string
 
 log = logging.getLogger(__name__)

airflow/models/renderedtifields.py

@@ -145,7 +145,7 @@ def __repr__(self):
         return prefix + ">"
 
     def _redact(self):
-        from airflow.utils.log.secrets_masker import redact
+        from airflow.sdk.execution_time.secrets_masker import redact
 
         if self.k8s_pod_yaml:
             self.k8s_pod_yaml = redact(self.k8s_pod_yaml)

airflow/models/taskinstance.py

@@ -3267,7 +3267,7 @@ def get_rendered_template_fields(self, session: Session = NEW_SESSION) -> None:
 
         try:
             # If we get here, either the task hasn't run or the RTIF record was purged.
-            from airflow.utils.log.secrets_masker import redact
+            from airflow.sdk.execution_time.secrets_masker import redact
 
             self.render_templates()
             for field_name in self.task.template_fields:

airflow/models/variable.py

@@ -28,10 +28,10 @@
 from airflow.configuration import ensure_secrets_loaded
 from airflow.models.base import ID_LEN, Base
 from airflow.models.crypto import get_fernet
+from airflow.sdk.execution_time.secrets_masker import mask_secret
 from airflow.secrets.cache import SecretCache
 from airflow.secrets.metastore import MetastoreBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.log.secrets_masker import mask_secret
 from airflow.utils.session import provide_session
 
 if TYPE_CHECKING:

airflow/serialization/helpers.py

@@ -21,8 +21,8 @@
 from typing import Any
 
 from airflow.configuration import conf
+from airflow.sdk.execution_time.secrets_masker import redact
 from airflow.settings import json
-from airflow.utils.log.secrets_masker import redact
 
 
 def serialize_template_field(template_field: Any, name: str) -> str | dict | list | int | float:

airflow/settings.py

@@ -328,7 +328,7 @@ def _is_sqlite_db_path_relative(sqla_conn_str: str) -> bool:
 
 def configure_orm(disable_connection_pool=False, pool_class=None):
     """Configure ORM using SQLAlchemy."""
-    from airflow.utils.log.secrets_masker import mask_secret
+    from airflow.sdk.execution_time.secrets_masker import mask_secret
 
     if _is_sqlite_db_path_relative(SQL_ALCHEMY_CONN):
         from airflow.exceptions import AirflowConfigException

airflow/utils/cli.py

@@ -35,9 +35,9 @@
 
 from airflow import settings
 from airflow.exceptions import AirflowException
+from airflow.sdk.execution_time.secrets_masker import should_hide_value_for_key
 from airflow.utils import cli_action_loggers, timezone
 from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler
-from airflow.utils.log.secrets_masker import should_hide_value_for_key
 from airflow.utils.platform import getuser, is_terminal_support_colors
 
 T = TypeVar("T", bound=Callable)

airflow/www/decorators.py

@@ -31,7 +31,7 @@
 
 from airflow.api_fastapi.app import get_auth_manager
 from airflow.models import Log
-from airflow.utils.log import secrets_masker
+from airflow.sdk.execution_time import secrets_masker
 from airflow.utils.session import create_session
 
 T = TypeVar("T", bound=Callable)

airflow/www/views.py

@@ -114,6 +114,7 @@
 from airflow.plugins_manager import PLUGINS_ATTRIBUTES_TO_DUMP
 from airflow.providers_manager import ProvidersManager
 from airflow.sdk.definitions.asset import Asset, AssetAlias
+from airflow.sdk.execution_time import secrets_masker
 from airflow.security import permissions
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS
@@ -127,7 +128,6 @@
 from airflow.utils.db import get_query_count
 from airflow.utils.docs import get_doc_url_for_provider, get_docs_url
 from airflow.utils.helpers import exactly_one
-from airflow.utils.log import secrets_masker
 from airflow.utils.log.log_reader import TaskLogReader
 from airflow.utils.net import get_hostname
 from airflow.utils.session import NEW_SESSION, create_session, provide_session

docs/apache-airflow/security/secrets/mask-sensitive-values.rst

@@ -58,7 +58,7 @@ your DAG file or operator's ``execute`` function using the ``mask_secret`` funct
 
     @task
     def my_func():
-        from airflow.utils.log.secrets_masker import mask_secret
+        from airflow.sdk.execution_time.secrets_masker import mask_secret
 
         mask_secret("custom_value")
 
@@ -71,7 +71,7 @@ or
 
     class MyOperator(BaseOperator):
         def execute(self, context):
-            from airflow.utils.log.secrets_masker import mask_secret
+            from airflow.sdk.execution_time.secrets_masker import mask_secret
 
             mask_secret("custom_value")
 

newsfragments/46375.significant.rst

@@ -0,0 +1,22 @@
+``SecretsMasker`` has now been moved into the task SDK to be consumed by DAG authors and users
+
+Any occurrences of the ``secrets_masker`` module will have to be updated from ``airflow.utils.log.secrets_masker`` to the new path: ``airflow.sdk.execution_time.secrets_masker``
+
+* Types of change
+
+  * [ ] Dag changes
+  * [ ] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [ ] Code interface changes
+
+* Migration rules needed
+
+  * ruff
+
+    * AIR302
+
+      * [ ] ``airflow.utils.log.secrets_masker`` -> ``airflow.sdk.execution_time.secrets_masker``

providers/openlineage/src/airflow/providers/openlineage/plugins/adapter.py

@@ -50,8 +50,13 @@
     from datetime import datetime
 
     from airflow.providers.openlineage.extractors import OperatorLineage
-    from airflow.utils.log.secrets_masker import SecretsMasker
+    from airflow.sdk.execution_time.secrets_masker import SecretsMasker, _secrets_masker
     from airflow.utils.state import DagRunState
+else:
+    try:
+        from airflow.sdk.execution_time.secrets_masker import SecretsMasker, _secrets_masker
+    except ImportError:
+        from airflow.utils.log.secrets_masker import SecretsMasker, _secrets_masker
 
 _PRODUCER = f"https://github.com/apache/airflow/tree/providers-openlineage/{OPENLINEAGE_PROVIDER_VERSION}"
 
@@ -71,8 +76,6 @@ def __init__(self, client: OpenLineageClient | None = None, secrets_masker: Secr
         super().__init__()
         self._client = client
         if not secrets_masker:
-            from airflow.utils.log.secrets_masker import _secrets_masker
-
             secrets_masker = _secrets_masker()
         self._redacter = OpenLineageRedactor.from_masker(secrets_masker)
 

providers/openlineage/src/airflow/providers/openlineage/utils/utils.py

@@ -54,12 +54,6 @@
 from airflow.sensors.base import BaseSensorOperator
 from airflow.serialization.serialized_objects import SerializedBaseOperator
 from airflow.utils.context import AirflowContextDeprecationWarning
-from airflow.utils.log.secrets_masker import (
-    Redactable,
-    Redacted,
-    SecretsMasker,
-    should_hide_value_for_key,
-)
 from airflow.utils.module_loading import import_string
 from airflow.utils.session import NEW_SESSION, provide_session
 from openlineage.client.utils import RedactMixin
@@ -68,6 +62,12 @@
     from airflow.models import TaskInstance
     from airflow.providers.common.compat.assets import Asset
     from airflow.sdk import DAG, BaseOperator, MappedOperator
+    from airflow.sdk.execution_time.secrets_masker import (
+        Redactable,
+        Redacted,
+        SecretsMasker,
+        should_hide_value_for_key,
+    )
     from airflow.utils.state import DagRunState, TaskInstanceState
     from openlineage.client.event_v2 import Dataset as OpenLineageDataset
     from openlineage.client.facet_v2 import RunFacet, processing_engine_run
@@ -86,6 +86,21 @@
             # dataset is renamed to asset since Airflow 3.0
             from airflow.datasets import Dataset as Asset
 
+    try:
+        from airflow.sdk.execution_time.secrets_masker import (
+            Redactable,
+            Redacted,
+            SecretsMasker,
+            should_hide_value_for_key,
+        )
+    except ImportError:
+        from airflow.utils.log.secrets_masker import (
+            Redactable,
+            Redacted,
+            SecretsMasker,
+            should_hide_value_for_key,
+        )
+
 log = logging.getLogger(__name__)
 _NOMINAL_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
 

providers/openlineage/tests/provider_tests/openlineage/plugins/test_utils.py

@@ -20,7 +20,7 @@
 import json
 import uuid
 from json import JSONEncoder
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -45,7 +45,6 @@
 )
 from airflow.serialization.enums import DagAttributeTypes
 from airflow.utils import timezone
-from airflow.utils.log.secrets_masker import _secrets_masker
 from airflow.utils.state import State
 from airflow.utils.types import DagRunType
 
@@ -57,6 +56,14 @@
 if AIRFLOW_V_3_0_PLUS:
     from airflow.utils.types import DagRunTriggeredByType
 
+if TYPE_CHECKING:
+    from airflow.sdk.execution_time.secrets_masker import _secrets_masker
+else:
+    try:
+        from airflow.sdk.execution_time.secrets_masker import _secrets_masker
+    except ImportError:
+        from airflow.utils.log.secrets_masker import _secrets_masker
+
 
 class SafeStrDict(dict):
     def __str__(self):
@@ -236,7 +243,7 @@ def __init__(self):
 
 @pytest.mark.enable_redact
 def test_redact_with_exclusions(monkeypatch):
-    redactor = OpenLineageRedactor.from_masker(_secrets_masker())
+    redactor = OpenLineageRedactor.from_masker(_secrets_masker())  # type: ignore[assignment]
 
     class NotMixin:
         def __init__(self):

providers/src/airflow/providers/amazon/aws/hooks/base_aws.py

@@ -58,7 +58,6 @@
 from airflow.providers_manager import ProvidersManager
 from airflow.utils.helpers import exactly_one
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.log.secrets_masker import mask_secret
 
 BaseAwsConnection = TypeVar("BaseAwsConnection", bound=Union[boto3.client, boto3.resource])
 
@@ -68,6 +67,12 @@
     from botocore.credentials import ReadOnlyCredentials
 
     from airflow.models.connection import Connection
+    from airflow.sdk.execution_time.secrets_masker import mask_secret
+else:
+    try:
+        from airflow.sdk.execution_time.secrets_masker import mask_secret
+    except ImportError:
+        from airflow.utils.log.secrets_masker import mask_secret
 
 _loader = botocore.loaders.Loader()
 """

providers/src/airflow/providers/amazon/aws/hooks/ecr.py

@@ -23,11 +23,17 @@
 from typing import TYPE_CHECKING
 
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
-from airflow.utils.log.secrets_masker import mask_secret
 
 if TYPE_CHECKING:
     from datetime import datetime
 
+    from airflow.sdk.execution_time.secrets_masker import mask_secret
+else:
+    try:
+        from airflow.sdk.execution_time.secrets_masker import mask_secret
+    except ImportError:
+        from airflow.utils.log.secrets_masker import mask_secret
+
 logger = logging.getLogger(__name__)