added new endpoints

Author: flowstate

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__
 
@@ -42,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/dependencies/__init__.py

@@ -22,7 +22,7 @@ def post(self, files: list, params: dict) -> dict:
         else:
             result = {}
             log.error(f"Error posting {files} to the Dependency API")
-            print(response.text)
+            log.error(response.text)
         return result
 
     def get(
@@ -38,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/historical/__init__.py

@@ -0,0 +1,47 @@
+import logging
+from urllib.parse import urlencode
+
+log = logging.getLogger("socketdev")
+
+
+class Historical:
+    def __init__(self, api):
+        self.api = api
+
+    def list(self, org_slug: str, query_params: dict = None) -> dict:
+        """Get historical alerts list for an organization.
+
+        Args:
+            org_slug: Organization slug
+            query_params: Optional dictionary of query parameters
+        """
+        path = f"orgs/{org_slug}/alerts/historical"
+        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 historical alerts: {response.status_code}")
+        log.error(response.text)
+        return {}
+
+    def trend(self, org_slug: str, query_params: dict = None) -> dict:
+        """Get historical alerts trend data for an organization.
+
+        Args:
+            org_slug: Organization slug
+            query_params: Optional dictionary of query parameters
+        """
+        path = f"orgs/{org_slug}/alerts/historical/trend"
+        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 historical trend: {response.status_code}")
+        log.error(response.text)
+        return {}

socketdev/settings/__init__.py

@@ -97,3 +97,76 @@ def get(
                 {"securityPolicyRules": {}, "success": False, "status": response.status_code, "message": error_message}
             )
         return {}
+
+    def integration_events(self, org_slug: str, integration_id: str) -> dict:
+        """Get integration events for a specific integration.
+
+        Args:
+            org_slug: Organization slug
+            integration_id: Integration ID
+        """
+        path = f"orgs/{org_slug}/settings/integrations/{integration_id}"
+        response = self.api.do_request(path=path)
+
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error getting integration events: {response.status_code}, message: {error_message}")
+        return {}
+
+    def get_license_policy(self, org_slug: str) -> dict:
+        """Get license policy settings for an organization.
+
+        Args:
+            org_slug: Organization slug
+        """
+        path = f"orgs/{org_slug}/settings/license-policy"
+        response = self.api.do_request(path=path)
+
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error getting license policy: {response.status_code}, message: {error_message}")
+        return {}
+
+    def update_security_policy(self, org_slug: str, body: dict, custom_rules_only: bool = False) -> dict:
+        """Update security policy settings for an organization.
+
+        Args:
+            org_slug: Organization slug
+            body: Security policy configuration to update
+            custom_rules_only: Optional flag to update only custom rules
+        """
+        path = f"orgs/{org_slug}/settings/security-policy"
+        if custom_rules_only:
+            path += "?custom_rules_only=true"
+
+        response = self.api.do_request(path=path, method="POST", payload=body)
+
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error updating security policy: {response.status_code}, message: {error_message}")
+        return {}
+
+    def update_license_policy(self, org_slug: str, body: dict, merge_update: bool = False) -> dict:
+        """Update license policy settings for an organization.
+
+        Args:
+            org_slug: Organization slug
+            body: License policy configuration to update
+            merge_update: Optional flag to merge updates instead of replacing (defaults to False)
+        """
+        path = f"orgs/{org_slug}/settings/license-policy?merge_update={str(merge_update).lower()}"
+
+        response = self.api.do_request(path=path, method="POST", payload=body)
+
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error updating license policy: {response.status_code}, message: {error_message}")
+        return {}

socketdev/triage/__init__.py

@@ -0,0 +1,47 @@
+import logging
+from urllib.parse import urlencode
+
+log = logging.getLogger("socketdev")
+
+
+class Triage:
+    def __init__(self, api):
+        self.api = api
+
+    def list_alert_triage(self, org_slug: str, query_params: dict = None) -> dict:
+        """Get list of triaged alerts for an organization.
+
+        Args:
+            org_slug: Organization slug
+            query_params: Optional dictionary of query parameters
+        """
+        path = f"orgs/{org_slug}/triage/alerts"
+        if query_params:
+            path += "?" + urlencode(query_params)
+
+        response = self.api.do_request(path=path)
+
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error getting alert triage list: {response.status_code}, message: {error_message}")
+        return {}
+
+    def update_alert_triage(self, org_slug: str, body: dict) -> dict:
+        """Update triaged alerts for an organization.
+
+        Args:
+            org_slug: Organization slug
+            body: Alert triage configuration to update
+        """
+        path = f"orgs/{org_slug}/triage/alerts"
+
+        response = self.api.do_request(path=path, method="POST", payload=body)
+
+        if 200 <= response.status_code < 300:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error updating alert triage: {response.status_code}, message: {error_message}")
+        return {}