diff --git a/webservice/components/request_adapter.py b/webservice/components/request_adapter.py index 4f50a168..4b23bf6d 100644 --- a/webservice/components/request_adapter.py +++ b/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) diff --git a/webservice/tests/test_webservice.py b/webservice/tests/test_webservice.py index a05b9702..d220c624 100644 --- a/webservice/tests/test_webservice.py +++ b/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" + )