Return headers on request exception

socketdev/__init__.py

@@ -4,6 +4,7 @@
 from socketdev.dependencies import Dependencies
 from socketdev.export import Export
 from socketdev.fullscans import FullScans
+from socketdev.historical import Historical
 from socketdev.npm import NPM
 from socketdev.openapi import OpenAPI
 from socketdev.org import Orgs
@@ -14,6 +15,7 @@
 from socketdev.repositories import Repositories
 from socketdev.sbom import Sbom
 from socketdev.settings import Settings
+from socketdev.triage import Triage
 from socketdev.utils import Utils, IntegrationType, INTEGRATION_TYPES
 from socketdev.version import __version__
 
@@ -23,7 +25,6 @@
 __all__ = ["socketdev", "Utils", "IntegrationType", "INTEGRATION_TYPES"]
 
 
-
 global encoded_key
 encoded_key: str
 
@@ -32,6 +33,8 @@
 log = logging.getLogger("socketdev")
 log.addHandler(logging.NullHandler())
 
+# TODO: Add debug flag to constructor to enable verbose error logging for API response parsing.
+
 
 class socketdev:
     def __init__(self, token: str, timeout: int = 1200):
@@ -41,18 +44,20 @@ def __init__(self, token: str, timeout: int = 1200):
         self.api.set_timeout(timeout)
 
         self.dependencies = Dependencies(self.api)
+        self.export = Export(self.api)
+        self.fullscans = FullScans(self.api)
+        self.historical = Historical(self.api)
         self.npm = NPM(self.api)
         self.openapi = OpenAPI(self.api)
         self.org = Orgs(self.api)
+        self.purl = Purl(self.api)
         self.quota = Quota(self.api)
         self.report = Report(self.api)
-        self.sbom = Sbom(self.api)
-        self.purl = Purl(self.api)
-        self.fullscans = FullScans(self.api)
-        self.export = Export(self.api)
-        self.repositories = Repositories(self.api)
         self.repos = Repos(self.api)
+        self.repositories = Repositories(self.api)
+        self.sbom = Sbom(self.api)
         self.settings = Settings(self.api)
+        self.triage = Triage(self.api)
         self.utils = Utils()
 
     @staticmethod

socketdev/core/api.py

@@ -2,9 +2,16 @@
 import requests
 from socketdev.core.classes import Response
 from socketdev.exceptions import (
-    APIKeyMissing, APIFailure, APIAccessDenied, APIInsufficientQuota, 
-    APIResourceNotFound, APITimeout, APIConnectionError, APIBadGateway,
-    APIInsufficientPermissions, APIOrganizationNotAllowed
+    APIKeyMissing,
+    APIFailure,
+    APIAccessDenied,
+    APIInsufficientQuota,
+    APIResourceNotFound,
+    APITimeout,
+    APIConnectionError,
+    APIBadGateway,
+    APIInsufficientPermissions,
+    APIOrganizationNotAllowed,
 )
 from socketdev.version import __version__
 from requests.exceptions import Timeout, ConnectionError
@@ -24,7 +31,12 @@ def set_timeout(self, timeout: int):
         self.request_timeout = timeout
 
     def do_request(
-        self, path: str, headers: dict | None = None, payload: [dict, str] = None, files: list = None, method: str = "GET"
+        self,
+        path: str,
+        headers: dict | None = None,
+        payload: [dict, str] = None,
+        files: list = None,
+        method: str = "GET",
     ) -> Response:
         if self.encoded_key is None or self.encoded_key == "":
             raise APIKeyMissing
@@ -36,33 +48,39 @@ def do_request(
                 "accept": "application/json",
             }
         url = f"{self.api_url}/{path}"
+
+        def format_headers(headers_dict):
+            return "\n".join(f"{k}: {v}" for k, v in headers_dict.items())
+
         try:
             start_time = time.time()
             response = requests.request(
                 method.upper(), url, headers=headers, data=payload, files=files, timeout=self.request_timeout
             )
             request_duration = time.time() - start_time
-
+
+            headers_str = f"\n\nHeaders:\n{format_headers(response.headers)}" if response.headers else ""
+            path_str = f"\nPath: {url}"
+
             if response.status_code == 401:
-                raise APIAccessDenied("Unauthorized")
+                raise APIAccessDenied(f"Unauthorized{path_str}{headers_str}")
             if response.status_code == 403:
                 try:
-                    error_message = response.json().get('error', {}).get('message', '')
+                    error_message = response.json().get("error", {}).get("message", "")
                     if "Insufficient permissions for API method" in error_message:
-                        raise APIInsufficientPermissions(error_message)
+                        raise APIInsufficientPermissions(f"{error_message}{path_str}{headers_str}")
                     elif "Organization not allowed" in error_message:
-                        raise APIOrganizationNotAllowed(error_message)
+                        raise APIOrganizationNotAllowed(f"{error_message}{path_str}{headers_str}")
                     elif "Insufficient max quota" in error_message:
-                        raise APIInsufficientQuota(error_message)
+                        raise APIInsufficientQuota(f"{error_message}{path_str}{headers_str}")
                     else:
-                        raise APIAccessDenied(error_message or "Access denied")
+                        raise APIAccessDenied(f"{error_message or 'Access denied'}{path_str}{headers_str}")
                 except ValueError:
-                    # If JSON parsing fails
-                    raise APIAccessDenied("Access denied")
+                    raise APIAccessDenied(f"Access denied{path_str}{headers_str}")
             if response.status_code == 404:
-                raise APIResourceNotFound(f"Path not found {path}")
+                raise APIResourceNotFound(f"Path not found {path}{path_str}{headers_str}")
             if response.status_code == 429:
-                retry_after = response.headers.get('retry-after')
+                retry_after = response.headers.get("retry-after")
                 if retry_after:
                     try:
                         seconds = int(retry_after)
@@ -73,23 +91,34 @@ def do_request(
                         time_msg = f" Retry after: {retry_after}"
                 else:
                     time_msg = ""
-                raise APIInsufficientQuota(f"Insufficient quota for API route.{time_msg}")
+                raise APIInsufficientQuota(f"Insufficient quota for API route.{time_msg}{path_str}{headers_str}")
             if response.status_code == 502:
-                raise APIBadGateway("Upstream server error")
+                raise APIBadGateway(f"Upstream server error{path_str}{headers_str}")
             if response.status_code >= 400:
-                raise APIFailure(f"Bad Request: HTTP {response.status_code}")
-
+                raise APIFailure(
+                    f"Bad Request: HTTP original_status_code:{response.status_code}{path_str}{headers_str}",
+                    status_code=500,
+                )
+
             return response
-            
+
         except Timeout:
             request_duration = time.time() - start_time
             raise APITimeout(f"Request timed out after {request_duration:.2f} seconds")
         except ConnectionError as error:
             request_duration = time.time() - start_time
             raise APIConnectionError(f"Connection error after {request_duration:.2f} seconds: {error}")
-        except (APIAccessDenied, APIInsufficientQuota, APIResourceNotFound, APIFailure,
-                APITimeout, APIConnectionError, APIBadGateway, APIInsufficientPermissions,
-                APIOrganizationNotAllowed):
+        except (
+            APIAccessDenied,
+            APIInsufficientQuota,
+            APIResourceNotFound,
+            APIFailure,
+            APITimeout,
+            APIConnectionError,
+            APIBadGateway,
+            APIInsufficientPermissions,
+            APIOrganizationNotAllowed,
+        ):
             # Let all our custom exceptions propagate up unchanged
             raise
         except Exception as error:

socketdev/dependencies/__init__.py

@@ -1,8 +1,12 @@
 import json
 from urllib.parse import urlencode
-
+import logging
 from socketdev.tools import load_files
 
+log = logging.getLogger("socketdev")
+
+# TODO: Add types for responses. Not currently used in the CLI.
+
 
 class Dependencies:
     def __init__(self, api):
@@ -17,8 +21,8 @@ def post(self, files: list, params: dict) -> dict:
             result = response.json()
         else:
             result = {}
-            print(f"Error posting {files} to the Dependency API")
-            print(response.text)
+            log.error(f"Error posting {files} to the Dependency API")
+            log.error(response.text)
         return result
 
     def get(
@@ -34,6 +38,6 @@ def get(
             result = response.json()
         else:
             result = {}
-            print("Unable to retrieve Dependencies")
-            print(response.text)
+            log.error("Unable to retrieve Dependencies")
+            log.error(response.text)
         return result

socketdev/export/__init__.py

@@ -1,6 +1,9 @@
 from urllib.parse import urlencode
 from dataclasses import dataclass, asdict
 from typing import Optional
+import logging
+
+log = logging.getLogger("socketdev")
 
 
 @dataclass
@@ -23,40 +26,50 @@ class Export:
     def __init__(self, api):
         self.api = api
 
-    def cdx_bom(self, org_slug: str, id: str, query_params: Optional[ExportQueryParams] = None) -> dict:
+    def cdx_bom(
+        self, org_slug: str, id: str, query_params: Optional[ExportQueryParams] = None, use_types: bool = False
+    ) -> dict:
         """
         Export a Socket SBOM as a CycloneDX SBOM
         :param org_slug: String - The slug of the organization
         :param id: String - The id of either a full scan or an sbom report
         :param query_params: Optional[ExportQueryParams] - Query parameters for filtering
-        :return:
+        :param use_types: Optional[bool] - Whether to return typed responses
+        :return: dict
         """
         path = f"orgs/{org_slug}/export/cdx/{id}"
         if query_params:
             path += query_params.to_query_params()
         response = self.api.do_request(path=path)
-        try:
-            sbom = response.json()
-            sbom["success"] = True
-        except Exception as error:
-            sbom = {"success": False, "message": str(error)}
-        return sbom
-
-    def spdx_bom(self, org_slug: str, id: str, query_params: Optional[ExportQueryParams] = None) -> dict:
+
+        if response.status_code == 200:
+            return response.json()
+            # TODO: Add typed response when types are defined
+
+        log.error(f"Error exporting CDX BOM: {response.status_code}")
+        print(response.text)
+        return {}
+
+    def spdx_bom(
+        self, org_slug: str, id: str, query_params: Optional[ExportQueryParams] = None, use_types: bool = False
+    ) -> dict:
         """
         Export a Socket SBOM as an SPDX SBOM
         :param org_slug: String - The slug of the organization
         :param id: String - The id of either a full scan or an sbom report
         :param query_params: Optional[ExportQueryParams] - Query parameters for filtering
-        :return:
+        :param use_types: Optional[bool] - Whether to return typed responses
+        :return: dict
         """
         path = f"orgs/{org_slug}/export/spdx/{id}"
         if query_params:
             path += query_params.to_query_params()
         response = self.api.do_request(path=path)
-        try:
-            sbom = response.json()
-            sbom["success"] = True
-        except Exception as error:
-            sbom = {"success": False, "message": str(error)}
-        return sbom
+
+        if response.status_code == 200:
+            return response.json()
+            # TODO: Add typed response when types are defined
+
+        log.error(f"Error exporting SPDX BOM: {response.status_code}")
+        print(response.text)
+        return {}