CM-25061 - Hide sensitive API responses from debug logs

cycode/cli/main.py

@@ -149,7 +149,7 @@ def code_scan(
         if output == 'json':
             context.obj['no_progress_meter'] = True
 
-    context.obj['client'] = get_cycode_client(client_id, secret)
+    context.obj['client'] = get_cycode_client(client_id, secret, not context.obj['show_secret'])
     context.obj['severity_threshold'] = severity_threshold
     context.obj['monitor'] = monitor
     context.obj['report'] = report
@@ -234,15 +234,15 @@ def main_cli(
         CycodeClientBase.enrich_user_agent(user_agent_option.user_agent_suffix)
 
 
-def get_cycode_client(client_id: str, client_secret: str) -> 'ScanClient':
+def get_cycode_client(client_id: str, client_secret: str, hide_response_log: bool) -> 'ScanClient':
     if not client_id or not client_secret:
         client_id, client_secret = _get_configured_credentials()
         if not client_id:
             raise click.ClickException('Cycode client id needed.')
         if not client_secret:
             raise click.ClickException('Cycode client secret is needed.')
 
-    return create_scan_client(client_id, client_secret)
+    return create_scan_client(client_id, client_secret, hide_response_log)
 
 
 def _get_configured_credentials() -> Tuple[str, str]:

cycode/cyclient/auth_client.py

@@ -24,7 +24,7 @@ def get_api_token(self, session_id: str, code_verifier: str) -> Optional[models.
         path = f'{self.AUTH_CONTROLLER_PATH}/token'
         body = {'session_id': session_id, 'code_verifier': code_verifier}
         try:
-            response = self.cycode_client.post(url_path=path, body=body)
+            response = self.cycode_client.post(url_path=path, body=body, hide_response_content_log=True)
             return self.parse_api_token_polling_response(response)
         except (NetworkError, HttpUnauthorizedError) as e:
             return self.parse_api_token_polling_response(e.response)

cycode/cyclient/cycode_client_base.py

@@ -53,7 +53,13 @@ def get(self, url_path: str, headers: Optional[dict] = None, **kwargs) -> Respon
         return self._execute(method='get', endpoint=url_path, headers=headers, **kwargs)
 
     def _execute(
-        self, method: str, endpoint: str, headers: Optional[dict] = None, without_auth: bool = False, **kwargs
+        self,
+        method: str,
+        endpoint: str,
+        headers: Optional[dict] = None,
+        without_auth: bool = False,
+        hide_response_content_log: bool = False,
+        **kwargs,
     ) -> Response:
         url = self.build_full_url(self.api_url, endpoint)
         logger.debug(f'Executing {method.upper()} request to {url}')
@@ -62,7 +68,8 @@ def _execute(
             headers = self.get_request_headers(headers, without_auth=without_auth)
             response = request(method=method, url=url, timeout=self.timeout, headers=headers, **kwargs)
 
-            logger.debug(f'Response {response.status_code} from {url}. Content: {response.text}')
+            content = 'HIDDEN' if hide_response_content_log else response.text
+            logger.debug(f'Response {response.status_code} from {url}. Content: {content}')
 
             response.raise_for_status()
             return response

cycode/cyclient/cycode_token_based_client.py

@@ -35,6 +35,7 @@ def refresh_api_token(self) -> None:
             url_path='api/v1/auth/api-token',
             body={'clientId': self.client_id, 'secret': self.client_secret},
             without_auth=True,
+            hide_response_content_log=True,
         )
         auth_response_data = auth_response.json()
 

cycode/cyclient/scan_client.py

@@ -11,22 +11,23 @@
 
 
 class ScanClient:
-    def __init__(self, scan_cycode_client: CycodeClientBase, scan_config: ScanConfigBase) -> None:
+    def __init__(
+        self, scan_cycode_client: CycodeClientBase, scan_config: ScanConfigBase, hide_response_log: bool = True
+    ) -> None:
         self.scan_cycode_client = scan_cycode_client
         self.scan_config = scan_config
+
         self.SCAN_CONTROLLER_PATH = 'api/v1/scan'
         self.DETECTIONS_SERVICE_CONTROLLER_PATH = 'api/v1/detections'
 
+        self._hide_response_log = hide_response_log
+
     def content_scan(self, scan_type: str, file_name: str, content: str, is_git_diff: bool = True) -> models.ScanResult:
         path = f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/content'
         body = {'name': file_name, 'content': content, 'is_git_diff': is_git_diff}
-        response = self.scan_cycode_client.post(url_path=path, body=body)
-        return self.parse_scan_response(response)
-
-    def file_scan(self, scan_type: str, path: str) -> models.ScanResult:
-        url_path = f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}'
-        files = {'file': open(path, 'rb')}  # noqa: SIM115 requests lib should care about closing
-        response = self.scan_cycode_client.post(url_path=url_path, files=files)
+        response = self.scan_cycode_client.post(
+            url_path=path, body=body, hide_response_content_log=self._hide_response_log
+        )
         return self.parse_scan_response(response)
 
     def zipped_file_scan(
@@ -39,6 +40,7 @@ def zipped_file_scan(
             url_path=url_path,
             data={'scan_id': scan_id, 'is_git_diff': is_git_diff, 'scan_parameters': json.dumps(scan_parameters)},
             files=files,
+            hide_response_content_log=self._hide_response_log,
         )
 
         return self.parse_zipped_file_scan_response(response)
@@ -96,7 +98,9 @@ def get_scan_detections(self, scan_id: str) -> List[dict]:
             params['page_size'] = page_size
             params['page_number'] = page_number
 
-            response = self.scan_cycode_client.get(url_path=url_path, params=params).json()
+            response = self.scan_cycode_client.get(
+                url_path=url_path, params=params, hide_response_content_log=self._hide_response_log
+            ).json()
             detections.extend(response)
 
             page_number += 1
@@ -116,7 +120,9 @@ def commit_range_zipped_file_scan(
             f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/commit-range-zipped-file'
         )
         files = {'file': ('multiple_files_scan.zip', zip_file.read())}
-        response = self.scan_cycode_client.post(url_path=url_path, data={'scan_id': scan_id}, files=files)
+        response = self.scan_cycode_client.post(
+            url_path=url_path, data={'scan_id': scan_id}, files=files, hide_response_content_log=self._hide_response_log
+        )
         return self.parse_zipped_file_scan_response(response)
 
     def report_scan_status(self, scan_type: str, scan_id: str, scan_status: dict) -> None:

cycode/cyclient/scan_config/scan_config_creator.py

@@ -8,13 +8,13 @@
 from cycode.cyclient.scan_config.scan_config_base import DefaultScanConfig, DevScanConfig
 
 
-def create_scan_client(client_id: str, client_secret: str) -> ScanClient:
+def create_scan_client(client_id: str, client_secret: str, hide_response_log: bool) -> ScanClient:
     if dev_mode:
         scan_cycode_client, scan_config = create_scan_for_dev_env()
     else:
         scan_cycode_client, scan_config = create_scan(client_id, client_secret)
 
-    return ScanClient(scan_cycode_client=scan_cycode_client, scan_config=scan_config)
+    return ScanClient(scan_cycode_client, scan_config, hide_response_log)
 
 
 def create_scan(client_id: str, client_secret: str) -> Tuple[CycodeTokenBasedClient, DefaultScanConfig]:

tests/conftest.py

@@ -19,7 +19,7 @@
 
 @pytest.fixture(scope='session')
 def scan_client() -> ScanClient:
-    return create_scan_client(_CLIENT_ID, _CLIENT_SECRET)
+    return create_scan_client(_CLIENT_ID, _CLIENT_SECRET, hide_response_log=False)
 
 
 @pytest.fixture(scope='session')