fixup! webservice: log calls

webservice/components/request_adapter.py

@@ -4,6 +4,7 @@
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
 import logging
+from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
 import requests
 
@@ -12,6 +13,21 @@
 _logger = logging.getLogger(__name__)
 
 
+def sanitize_url_for_log(url, blacklisted_keys=None):
+    """Sanitize url to avoid loggin sensitive data"""
+    blacklisted_keys = blacklisted_keys or ("apikey", "password", "pwd")
+    parsed = urlparse(url)
+    query = parse_qs(parsed.query, keep_blank_values=False)
+    clean_query = {}
+    for k, v in query.items():
+        for bl_key in blacklisted_keys:
+            if bl_key.lower() in k.lower():
+                continue
+        clean_query[k] = v
+    parsed = parsed._replace(query=urlencode(query, True))
+    return urlunparse(parsed)
+
+
 class BaseRestRequestsAdapter(Component):
     """Generic adapter for HTTP requests."""
 
@@ -23,7 +39,8 @@ class BaseRestRequestsAdapter(Component):
     def _request(self, method, url=None, url_params=None, **kwargs):
         url = self._get_url(url=url, url_params=url_params)
         # TODO: turn on/off debug from webservice setting?
-        _logger.info("%s CALL to %s", method, url)
+        url_to_log = self._sanitize_url_for_log(url)
+        _logger.info("%s call to %s", method, url_to_log)
         new_kwargs = kwargs.copy()
         new_kwargs.update(
             {"auth": self._get_auth(**kwargs), "headers": self._get_headers(**kwargs)}
@@ -32,6 +49,9 @@ def _request(self, method, url=None, url_params=None, **kwargs):
         request.raise_for_status()
         return request.content
 
+    def _sanitize_url_for_log(self, url):
+        return sanitize_url_for_log(url)
+
     def get(self, **kwargs):
         return self._request("get", **kwargs)
 

webservice/tests/test_webservice.py

@@ -190,3 +190,9 @@ def test_web_service_call_args(self):
             responses.calls[0].request.headers["Content-Type"], "application/xml"
         )
         self.assertEqual(responses.calls[0].request.headers["demo_header"], "HEADER")
+
+    def test_sanitize_url(self):
+        url = "https://custom.url/?a=1&apikey=secret&password=moresecret"
+        self.assertEqual(
+            self.webservice._sanitize_url_for_log(url), "https://custom.url/?a=1"
+        )