Secret masker ignores passwords with special chars (#36692)

* Secret masker ignores passwords with special chars #36688

Connection uri's get connection uses quote to change the
password and certain other fields to escape special chars
due to this, when the connection object is passed through
the masker this changed string is skipped.

* Added a test for the logging change

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:[email protected]: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(