DFP-4138 Filtered sensitive request body from logging incase of exc

spark_pipeline_framework/utilities/api_helper/http_request.py

@@ -3,6 +3,7 @@
 """
 
 import json
+from copy import deepcopy
 from enum import Enum
 from os import environ
 from typing import Optional, Dict, Any, List, Callable, Union, NamedTuple, Tuple
@@ -152,15 +153,15 @@ def get_response(
             self.cert,
             self.verify,
         )
-        arguments = {"headers": self.headers}
+        arguments: Dict[str, Any] = {"headers": self.headers}
         request_function = None
         if self.request_type == RequestType.GET:
             arguments["params"] = self.payload
             request_function = session.get
         elif self.request_type == RequestType.POST:
             # https://requests.readthedocs.io/en/master/user/quickstart/#more-complicated-post-requests
             arguments["data"] = (
-                json.dumps(self.payload)  # type: ignore
+                json.dumps(self.payload)
                 if self.post_as_json_formatted_string
                 else self.payload
             )
@@ -171,13 +172,18 @@ def get_response(
         # remove None arguments
         arguments = {k: v for k, v in arguments.items() if v is not None}
 
+        # Hide sensitive tokens to protect against unauthorized access
+        filtered_arguments = deepcopy(arguments)
+        headers = filtered_arguments.get("headers")
+        if isinstance(headers, dict) and headers.get("Authorization"):
+            filtered_arguments["headers"]["Authorization"] = "[FILTERED]"
         response = self._send_request(request_function, arguments=arguments)
         if self.raise_error:
             try:
                 response.raise_for_status()
             except HTTPError as e:
                 if self.logger:
-                    error_text = f"Request to {self.url} with arguments {json.dumps(arguments)} failed"
+                    error_text = f"Request to {self.url} with arguments {json.dumps(filtered_arguments)} failed"
                     if e.response:
                         error_text += (
                             f" with {e.response.status_code}: {e.response.content!r}."

spark_pipeline_framework/utilities/api_helper/v2/http_request.py

@@ -1,6 +1,7 @@
 import json
 from enum import Enum
 from os import environ
+from copy import deepcopy
 from typing import (
     Optional,
     Dict,
@@ -148,14 +149,14 @@ async def get_response_async(
         """
         session: ClientSession
         async with self._get_session() as session:
-            arguments = {"headers": self.headers}
+            arguments: Dict[str, Any] = {"headers": self.headers}
             request_function = None
             if self.request_type == RequestType.GET:
                 arguments["params"] = self.payload
                 request_function = session.get
             elif self.request_type == RequestType.POST:
                 arguments["data"] = (
-                    json.dumps(self.payload)  # type: ignore
+                    json.dumps(self.payload)
                     if self.post_as_json_formatted_string
                     else self.payload
                 )
@@ -165,10 +166,15 @@ async def get_response_async(
 
             arguments = {k: v for k, v in arguments.items() if v is not None}
 
+            # Hide sensitive tokens to protect against unauthorized access
+            filtered_arguments = deepcopy(arguments)
+            headers = filtered_arguments.get("headers")
+            if isinstance(headers, dict) and headers.get("Authorization"):
+                filtered_arguments["headers"]["Authorization"] = "[FILTERED]"
             response = await self._send_request_async(request_function, arguments)  # type: ignore[arg-type]
             if self.raise_error:
                 if response.status >= 400:
-                    error_text = f"Request to {self.url} with arguments {json.dumps(arguments)} failed with {response.status}: {await response.text()}."
+                    error_text = f"Request to {self.url} with arguments {json.dumps(filtered_arguments)} failed with {response.status}: {await response.text()}."
                     if self.logger:
                         self.logger.error(error_text)
                     raise ClientError(error_text)