Merge pull request #60 from SocketDev/doug/add-integration-settings-endpoints

feat: Add comprehensive SDK enhancements and new endpoint modules

Author: dc-larsen

socketdev/__init__.py

@@ -1,3 +1,4 @@
+import os
 from socketdev.core.api import API
 from socketdev.dependencies import Dependencies
 from socketdev.diffscans import DiffScans
@@ -26,7 +27,14 @@
 from socketdev.alerttypes import AlertTypes
 from socketdev.basics import Basics
 from socketdev.uploadmanifests import UploadManifests
+from socketdev.alertfullscansearch import AlertFullScanSearch
+from socketdev.alerts import Alerts
+from socketdev.fixes import Fixes
+from socketdev.supportedfiles import SupportedFiles
+from socketdev.webhooks import Webhooks
+from socketdev.telemetry import Telemetry
 from socketdev.log import log
+from typing import Optional
 
 __author__ = "socket.dev"
 __version__ = __version__
@@ -44,7 +52,22 @@
 
 
 class socketdev:
-    def __init__(self, token: str, timeout: int = 1200, allow_unverified: bool = False):
+    def __init__(self, token: Optional[str] = None, timeout: int = 1200, allow_unverified: bool = False):
+        # Try to get token from environment variables if not provided
+        if token is None:
+            token = (
+                os.getenv("SOCKET_SECURITY_API_TOKEN") or
+                os.getenv("SOCKET_SECURITY_API_KEY") or
+                os.getenv("SOCKET_API_KEY") or
+                os.getenv("SOCKET_API_TOKEN")
+            )
+        
+        if token is None:
+            raise ValueError(
+                "API token is required. Provide it as a parameter or set one of these environment variables: "
+                "SOCKET_SECURITY_API_TOKEN, SOCKET_SECURITY_API_KEY, SOCKET_API_KEY, SOCKET_API_TOKEN"
+            )
+        
         self.api = API()
         self.token = token + ":"
         self.api.encode_key(self.token)
@@ -77,6 +100,12 @@ def __init__(self, token: str, timeout: int = 1200, allow_unverified: bool = Fal
         self.alerttypes = AlertTypes(self.api)
         self.basics = Basics(self.api)
         self.uploadmanifests = UploadManifests(self.api)
+        self.alertfullscansearch = AlertFullScanSearch(self.api)
+        self.alerts = Alerts(self.api)
+        self.fixes = Fixes(self.api)
+        self.supportedfiles = SupportedFiles(self.api)
+        self.webhooks = Webhooks(self.api)
+        self.telemetry = Telemetry(self.api)
 
     @staticmethod
     def set_timeout(timeout: int):

socketdev/alertfullscansearch/__init__.py

@@ -0,0 +1,33 @@
+import logging
+from urllib.parse import urlencode
+
+log = logging.getLogger("socketdev")
+
+
+class AlertFullScanSearch:
+    def __init__(self, api):
+        self.api = api
+
+    def search(self, org_slug: str, **query_params) -> dict:
+        """
+        Search alerts across full scans.
+
+        Args:
+            org_slug: Organization slug
+            **query_params: Optional query parameters for filtering
+
+        Returns:
+            dict containing search results
+        """
+        path = f"orgs/{org_slug}/alert-full-scan-search"
+        if query_params:
+            path += "?" + urlencode(query_params)
+        
+        response = self.api.do_request(path=path)
+        
+        if response.status_code == 200:
+            return response.json()
+        
+        log.error(f"Error searching alerts: {response.status_code}")
+        log.error(response.text)
+        return {}

socketdev/alerts/__init__.py

@@ -0,0 +1,33 @@
+import logging
+from urllib.parse import urlencode
+
+log = logging.getLogger("socketdev")
+
+
+class Alerts:
+    def __init__(self, api):
+        self.api = api
+
+    def get(self, org_slug: str, **query_params) -> dict:
+        """
+        Get alerts for an organization.
+
+        Args:
+            org_slug: Organization slug
+            **query_params: Optional query parameters for filtering
+
+        Returns:
+            dict containing alerts data
+        """
+        path = f"orgs/{org_slug}/alerts"
+        if query_params:
+            path += "?" + urlencode(query_params)
+        
+        response = self.api.do_request(path=path)
+        
+        if response.status_code == 200:
+            return response.json()
+        
+        log.error(f"Error getting alerts: {response.status_code}")
+        log.error(response.text)
+        return {}

socketdev/export/__init__.py

@@ -73,3 +73,28 @@ def spdx_bom(
         log.error(f"Error exporting SPDX BOM: {response.status_code}")
         log.error(response.text)
         return {}
+
+    def openvex_bom(
+        self, org_slug: str, id: str, query_params: Optional[ExportQueryParams] = None, use_types: bool = False
+    ) -> dict:
+        """
+        Export a Socket SBOM as an OpenVEX 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
+        :param use_types: Optional[bool] - Whether to return typed responses
+        :return: dict
+        """
+        path = f"orgs/{org_slug}/export/openvex/{id}"
+        if query_params:
+            path += query_params.to_query_params()
+        response = self.api.do_request(path=path)
+
+        if response.status_code == 200:
+            return response.json()
+            # TODO: Add typed response when types are defined
+
+        log.error(f"Error exporting OpenVEX BOM: {response.status_code}")
+        log.error(response.text)
+        return {}
+

socketdev/fixes/__init__.py

@@ -0,0 +1,33 @@
+import logging
+from urllib.parse import urlencode
+
+log = logging.getLogger("socketdev")
+
+
+class Fixes:
+    def __init__(self, api):
+        self.api = api
+
+    def get(self, org_slug: str, **query_params) -> dict:
+        """
+        Get available fixes for an organization.
+
+        Args:
+            org_slug: Organization slug
+            **query_params: Optional query parameters for filtering
+
+        Returns:
+            dict containing available fixes
+        """
+        path = f"orgs/{org_slug}/fixes"
+        if query_params:
+            path += "?" + urlencode(query_params)
+        
+        response = self.api.do_request(path=path)
+        
+        if response.status_code == 200:
+            return response.json()
+        
+        log.error(f"Error getting fixes: {response.status_code}")
+        log.error(response.text)
+        return {}