refactor: logging improvements

Author: Brooke-white

redshift_connector/plugin/adfs_credentials_provider.py

@@ -61,7 +61,7 @@ def form_based_authentication(self: "AdfsCredentialsProvider") -> str:
             _logger.error("A unknown error occurred when requesting SAML assertion to refresh credentials")
             raise InterfaceError(e)
 
-        _logger.debug(response.text)
+        _logger.debug("ADFS form based authentication response length: {}".format(len(response.text)))
 
         try:
             soup = bs4.BeautifulSoup(response.text, features="lxml")

redshift_connector/plugin/azure_credentials_provider.py

@@ -103,7 +103,7 @@ def azure_oauth_based_authentication(self: "AzureCredentialsProvider") -> str:
             _logger.error("A unknown error occurred when requesting authentication from Azure.")
             raise InterfaceError(e)
 
-        _logger.debug(response.text)
+        _logger.debug("Azure Oauth authentication response length: {}".format(len(response.text)))
 
         # parse the JSON response to grab access_token field which contains Base64 encoded SAML
         # Assertion and decode it

redshift_connector/plugin/browser_azure_credentials_provider.py

@@ -148,7 +148,7 @@ def fetch_saml_response(self: "BrowserAzureCredentialsProvider", token):
             _logger.error("A unknown error occurred when requesting authentication from Azure")
             raise InterfaceError(e)
 
-        _logger.debug(response.text)
+        _logger.debug("Azure authentication response length: {}".format(len(response.text)))
 
         try:
             saml_assertion: str = response.json()["access_token"]

redshift_connector/plugin/ping_credentials_provider.py

@@ -65,7 +65,7 @@ def get_saml_assertion(self: "PingCredentialsProvider") -> str:
                 _logger.error("A unknown error occurred when requesting SAML assertion to refresh credentials")
                 raise InterfaceError(e)
 
-            _logger.debug(response.content)
+            _logger.debug("response length: {}".format(len(response.content)))
 
             try:
                 soup = bs4.BeautifulSoup(response.text)

redshift_connector/utils/logging_utils.py

@@ -12,27 +12,78 @@ def make_divider_block() -> str:
 def mask_secure_info_in_props(info: "RedshiftProperty") -> "RedshiftProperty":
     from redshift_connector import RedshiftProperty
 
+    logging_allow_list: typing.Tuple[str, ...] = (
+        # "access_key_id",
+        "allow_db_user_override",
+        "app_id",
+        "app_name",
+        "application_name",
+        "auth_profile",
+        "auto_create",
+        # "client_id",
+        "client_protocol_version",
+        # "client_secret",
+        "cluster_identifier",
+        "credentials_provider",
+        "database_metadata_current_db_only",
+        "db_groups",
+        "db_name",
+        "db_user",
+        "duration",
+        "endpoint_url",
+        "force_lowercase",
+        "group_federation",
+        "host",
+        "iam",
+        "iam_disable_cache",
+        "idp_host",
+        "idpPort",
+        "idp_response_timeout",
+        "idp_tenant",
+        "is_serverless",
+        "listen_port",
+        "login_url",
+        "max_prepared_statements",
+        "numeric_to_float",
+        "partner_sp_id",
+        # "password",
+        "port",
+        "preferred_role",
+        "principal",
+        "profile",
+        "provider_name",
+        "region",
+        "replication",
+        "role_arn",
+        "role_session_name",
+        "scope",
+        # "secret_access_key",
+        "serverless_acct_id",
+        "serverless_work_group",
+        # "session_token",
+        "source_address",
+        "ssl",
+        "ssl_insecure",
+        "sslmode",
+        "tcp_keepalive",
+        "timeout",
+        "unix_sock",
+        "user_name",
+        # "web_identity_token",
+    )
+
     if info is None:
         return info
-    secure_info_found: bool = False
-    placeholder_value: str = "***"
 
-    temp: RedshiftProperty = copy.deepcopy(info)
+    temp: RedshiftProperty = RedshiftProperty()
 
     def is_populated(field: typing.Optional[str]):
         return field is not None and field != ""
 
-    if is_populated(temp.password):
-        secure_info_found = True
-        temp.password = placeholder_value
-    if is_populated(temp.access_key_id):
-        secure_info_found = True
-        temp.access_key_id = placeholder_value
-    if is_populated(temp.secret_access_key):
-        secure_info_found = True
-        temp.secret_access_key = placeholder_value
-    if is_populated(temp.session_token):
-        secure_info_found = True
-        temp.session_token = placeholder_value
-
-    return temp if secure_info_found else info
+    for parameter, value in info.__dict__.items():
+        if parameter in logging_allow_list:
+            temp.put(parameter, value)
+        elif is_populated(value):
+            temp.put(parameter, "***")
+
+    return temp

setup.cfg

@@ -23,4 +23,4 @@ addopts =
     --cov-report term-missing
     --cov-report html:build/coverage
     --cov-report xml:build/coverage/coverage.xml
-testpaths=test
+    test

test/unit/test_logging_utils.py

@@ -3,7 +3,15 @@
 from redshift_connector import RedshiftProperty
 from redshift_connector.utils.logging_utils import mask_secure_info_in_props
 
-secret_rp_values = ("password", "access_key_id", "session_token", "secret_access_key")
+secret_rp_values = (
+    "password",
+    "access_key_id",
+    "session_token",
+    "secret_access_key",
+    "client_id",
+    "client_secret",
+    "web_identity_token",
+)
 
 
 @pytest.mark.parametrize("rp_arg", secret_rp_values)