Merge pull request #192 from torresashjiancyber/VC-53770-logos-fix-c

security: CWE-532: Redact sensitive data from debug logs

Author: stkomitov

vcert/common.py

@@ -755,7 +755,7 @@ def process_server_response(r):
             log.debug(r.content.decode())
             return r.status_code, r.content.decode()
         elif content_type.startswith(MIME_OCTET_STREAM):
-            log.debug(r.content)
+            log.debug(f"Received {len(r.content)} bytes (octet-stream body not logged)")
             return r.status_code, r.content
         else:
             log.error(f"Unexpected content type: {content_type} for request {r.request.url}")

vcert/connection_tpp.py

@@ -90,7 +90,7 @@ def put(self, args):
     def _get(self, url="", params=None):
         if not self._token or self._token[1] < time.time() + 1:
             self.auth()
-            log.debug(f"Token is {self._token[0]}, timeout is {self._token[1]}")
+            log.debug(f"Token is [REDACTED], timeout is {self._token[1]}")
 
         r = requests.get(f"{self._base_url}{url}",
                          headers={TOKEN_HEADER_NAME: self._token[0],
@@ -103,7 +103,7 @@ def _get(self, url="", params=None):
     def _post(self, url, data=None):
         if not self._token or self._token[1] < time.time() + 1:
             self.auth()
-            log.debug(f"Token is {self._token[0]}, timeout is {self._token[1]}")
+            log.debug(f"Token is [REDACTED], timeout is {self._token[1]}")
 
         if isinstance(data, dict):
             r = requests.post(f"{self._base_url}{url}",
@@ -120,7 +120,7 @@ def _post(self, url, data=None):
     def _put(self, url, data=None):
         if not self._token or self._token[1] < time.time() + 1:
             self.auth()
-            log.debug(f"Token is {self._token[0]}, timeout is {self._token[1]}")
+            log.debug(f"Token is [REDACTED], timeout is {self._token[1]}")
 
         if isinstance(data, dict):
             r = requests.put(f"{self._base_url}{url}",

vcert/connection_tpp_token.py

@@ -126,7 +126,9 @@ def _post(self, url=None, data=None, check_token=True, include_token_header=True
             headers[HEADER_AUTHORIZATION] = token
 
         if isinstance(data, dict):
-            log.debug(f"POST Request\n\tURL: {self._base_url+url}\n\tHeaders:{headers}\n\tBody:{data}\n")
+            safe_headers = {k: ('***' if k == HEADER_AUTHORIZATION else v) for k, v in headers.items()}
+            safe_data = {k: ('***' if k in ('password', 'Password', 'refresh_token', 'client_secret', 'PrivateKeyPassphrase') else v) for k, v in data.items()}
+            log.debug(f"POST Request\n\tURL: {self._base_url+url}\n\tHeaders:{safe_headers}\n\tBody:{safe_data}\n")
             r = requests.post(self._base_url + url, headers=headers, json=data,  **self._http_request_kwargs)  # nosec B113
         else:
             log.error(f"Unexpected client data type: {type(data)} for {url}")
@@ -146,7 +148,9 @@ def _put(self, url, data=None, check_token=True, include_token_header=True):
             headers[HEADER_AUTHORIZATION] = token
 
         if isinstance(data, dict):
-            log.debug(f"POST Request\n\tURL: {self._base_url + url}\n\tHeaders:{headers}\n\tBody:{data}\n")
+            safe_headers = {k: ('***' if k == HEADER_AUTHORIZATION else v) for k, v in headers.items()}
+            safe_data = {k: ('***' if k in ('password', 'Password', 'refresh_token', 'client_secret', 'PrivateKeyPassphrase') else v) for k, v in data.items()}
+            log.debug(f"POST Request\n\tURL: {self._base_url + url}\n\tHeaders:{safe_headers}\n\tBody:{safe_data}\n")
             r = requests.put(self._base_url + url, headers=headers, json=data,
                               **self._http_request_kwargs)  # nosec B113
         else:
@@ -157,13 +161,13 @@ def _put(self, url, data=None, check_token=True, include_token_header=True):
     def _check_token(self):
         if not self._auth.access_token:
             self.get_access_token()
-            log.debug(f"Token is {self._auth.access_token}, expire date is {self._auth.token_expires}")
+            log.debug(f"Token is [REDACTED], expire date is {self._auth.token_expires}")
 
         # Token expired, get new token
         elif self._auth.token_expires and self._auth.token_expires < time.time():
             if self._auth.refresh_token:
                 self.refresh_access_token()
-                log.debug(f"Token is {self._auth.access_token}, expire date is {self._auth.token_expires}")
+                log.debug(f"Token is [REDACTED], expire date is {self._auth.token_expires}")
             else:
                 raise AuthenticationError("Access Token expired. No refresh token provided.")