Secret masker ignores passwords with special chars

airflow/models/connection.py

@@ -173,6 +173,7 @@ def __init__(
 
         if self.password:
             mask_secret(self.password)
+            mask_secret(quote(self.password))
 
     @staticmethod
     def _validate_extra(extra, conn_id) -> None:
@@ -206,6 +207,7 @@ def _validate_extra(extra, conn_id) -> None:
     def on_db_load(self):
         if self.password:
             mask_secret(self.password)
+            mask_secret(quote(self.password))
 
     def parse_from_uri(self, **uri):
         """Use uri parameter in constructor, this method is deprecated."""

tests/always/test_connection.py

@@ -22,6 +22,7 @@
 import re
 from collections import namedtuple
 from unittest import mock
+from urllib.parse import quote
 
 import pytest
 import sqlalchemy
@@ -397,6 +398,7 @@ def test_connection_from_uri(self, test_config: UriTestCaseConfig):
         expected_calls = []
         if test_config.test_conn_attributes.get("password"):
             expected_calls.append(mock.call(test_config.test_conn_attributes["password"]))
+            expected_calls.append(mock.call(quote(test_config.test_conn_attributes["password"])))
 
         if test_config.test_conn_attributes.get("extra_dejson"):
             expected_calls.append(mock.call(test_config.test_conn_attributes["extra_dejson"]))
@@ -627,18 +629,18 @@ def test_from_json_special_characters(self, val, expected):
     @mock.patch.dict(
         "os.environ",
         {
-            "AIRFLOW_CONN_TEST_URI": "postgresql://username:password@ec2.compute.com:5432/the_database",
+            "AIRFLOW_CONN_TEST_URI": "postgresql://username:password%21@ec2.compute.com:5432/the_database",
         },
     )
     def test_using_env_var(self):
         conn = SqliteHook.get_connection(conn_id="test_uri")
         assert "ec2.compute.com" == conn.host
         assert "the_database" == conn.schema
         assert "username" == conn.login
-        assert "password" == conn.password
+        assert "password!" == conn.password
         assert 5432 == conn.port
 
-        self.mask_secret.assert_called_once_with("password")
+        self.mask_secret.assert_has_calls([mock.call("password!"), mock.call(quote("password!"))])
 
     @mock.patch.dict(
         "os.environ",
@@ -752,7 +754,7 @@ def test_masking_from_db(self):
             conn = Connection(
                 conn_id=f"test-{os.getpid()}",
                 conn_type="http",
-                password="s3cr3t",
+                password="s3cr3t!",
                 extra='{"apikey":"masked too"}',
             )
             session.add(conn)
@@ -768,7 +770,8 @@ def test_masking_from_db(self):
 
             assert self.mask_secret.mock_calls == [
                 # We should have called it _again_ when loading from the DB
-                mock.call("s3cr3t"),
+                mock.call("s3cr3t!"),
+                mock.call(quote("s3cr3t!")),
                 mock.call({"apikey": "masked too"}),
             ]
         finally:

tests/utils/log/test_secrets_masker.py

@@ -30,6 +30,7 @@
 import pytest
 
 from airflow import settings
+from airflow.models import Connection
 from airflow.utils.log.secrets_masker import (
     RedactedIO,
     SecretsMasker,
@@ -80,7 +81,6 @@ def logger(caplog):
     logger.addFilter(filt)
 
     filt.add_mask("password")
-
     return logger
 
 
@@ -340,6 +340,22 @@ def test_redact_state_enum(self, logger, caplog, state, expected):
         assert caplog.text == f"INFO State: {expected}\n"
         assert "TypeError" not in caplog.text
 
+    def test_masking_quoted_strings_in_connection(self, logger, caplog):
+        secrets_masker = [fltr for fltr in logger.filters if isinstance(fltr, SecretsMasker)][0]
+        with patch("airflow.utils.log.secrets_masker._secrets_masker", return_value=secrets_masker):
+            test_conn_attributes = dict(
+                conn_type="scheme",
+                host="host/location",
+                schema="schema",
+                login="user",
+                password="should_be_hidden!",
+                port=1234,
+                extra=None,
+            )
+            conn = Connection(**test_conn_attributes)
+            logger.info(conn.get_uri())
+            assert "should_be_hidden" not in caplog.text
+
 
 class TestShouldHideValueForKey:
     @pytest.mark.parametrize(