Decrypt SecureString value obtained by SsmHook

airflow/providers/amazon/aws/hooks/ssm.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.utils.log.secrets_masker import mask_secret
 from airflow.utils.types import NOTSET, ArgNotSet
 
 
@@ -40,6 +41,7 @@ def __init__(self, *args, **kwargs) -> None:
     def get_parameter_value(self, parameter: str, default: str | ArgNotSet = NOTSET) -> str:
         """
         Returns the value of the provided Parameter or an optional default.
+        If value exists, and it is encrypted, then decrypt and mask them for loggers.
 
         .. seealso::
             - :external+boto3:py:meth:`SSM.Client.get_parameter`
@@ -48,7 +50,11 @@ def get_parameter_value(self, parameter: str, default: str | ArgNotSet = NOTSET)
         :param default: Optional default value to return if none is found.
         """
         try:
-            return self.conn.get_parameter(Name=parameter)["Parameter"]["Value"]
+            param = self.conn.get_parameter(Name=parameter, WithDecryption=True)["Parameter"]
+            value = param["Value"]
+            if param["Type"] == "SecureString":
+                mask_secret(value)
+            return value
         except self.conn.exceptions.ParameterNotFound:
             if isinstance(default, ArgNotSet):
                 raise

tests/providers/amazon/aws/hooks/test_ssm.py

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+from unittest import mock
+
 import botocore.exceptions
 import pytest
 from moto import mock_ssm
@@ -33,13 +35,20 @@
 DEFAULT_VALUE = "default"
 
 
-class TestSsmHooks:
-    @pytest.fixture(autouse=True)
-    def setup_tests(self):
+class TestSsmHook:
+    @pytest.fixture(
+        autouse=True,
+        params=[
+            pytest.param("String", id="unencrypted-string"),
+            pytest.param("SecureString", id="encrypted-string"),
+        ],
+    )
+    def setup_tests(self, request):
         with mock_ssm():
             self.hook = SsmHook(region_name=REGION)
+            self.param_type = request.param
             self.hook.conn.put_parameter(
-                Type="String", Name=EXISTING_PARAM_NAME, Value=PARAM_VALUE, Overwrite=True
+                Type=self.param_type, Name=EXISTING_PARAM_NAME, Value=PARAM_VALUE, Overwrite=True
             )
             yield
 
@@ -62,6 +71,14 @@ def test_get_parameter_value_happy_cases(self, param_name, default_value, expect
         else:
             assert self.hook.get_parameter_value(param_name) == expected_result
 
+    @mock.patch("airflow.providers.amazon.aws.hooks.ssm.mask_secret")
+    def test_get_parameter_masking(self, mock_masker: mock.MagicMock):
+        self.hook.get_parameter_value(EXISTING_PARAM_NAME)
+        if self.param_type == "SecureString":
+            mock_masker.assert_called_once_with(PARAM_VALUE)
+        else:
+            mock_masker.assert_not_called()
+
     def test_get_parameter_value_param_does_not_exist_no_default_provided(self) -> None:
         with pytest.raises(botocore.exceptions.ClientError) as raised_exception:
             self.hook.get_parameter_value(BAD_PARAM_NAME)