Added new endpoints

Author: dacoburn

socketdev/__init__.py

@@ -16,6 +16,7 @@
 from socketdev.triage import Triage
 from socketdev.utils import Utils, IntegrationType, INTEGRATION_TYPES
 from socketdev.version import __version__
+from socketdev.labels import Labels
 from socketdev.log import log
 
 
@@ -57,6 +58,7 @@ def __init__(self, token: str, timeout: int = 1200):
         self.settings = Settings(self.api)
         self.triage = Triage(self.api)
         self.utils = Utils()
+        self.labels = Labels(self.api)
 
     @staticmethod
     def set_timeout(timeout: int):

socketdev/historical/__init__.py

@@ -7,6 +7,7 @@
 class Historical:
     def __init__(self, api):
         self.api = api
+        self.snapshots = self.Snapshots(api)
 
     def list(self, org_slug: str, query_params: dict = None) -> dict:
         """Get historical alerts list for an organization.
@@ -15,7 +16,7 @@ def list(self, org_slug: str, query_params: dict = None) -> dict:
             org_slug: Organization slug
             query_params: Optional dictionary of query parameters
         """
-        path = f"orgs/{org_slug}/alerts/historical"
+        path = f"orgs/{org_slug}/historical/alerts"
         if query_params:
             path += "?" + urlencode(query_params)
 
@@ -28,13 +29,13 @@ def list(self, org_slug: str, query_params: dict = None) -> dict:
         return {}
 
     def trend(self, org_slug: str, query_params: dict = None) -> dict:
-        """Get historical alerts trend data for an organization.
+        """Get historical alert trends data for an org.
 
         Args:
             org_slug: Organization slug
             query_params: Optional dictionary of query parameters
         """
-        path = f"orgs/{org_slug}/alerts/historical/trend"
+        path = f"orgs/{org_slug}/historical/alerts/trend"
         if query_params:
             path += "?" + urlencode(query_params)
 
@@ -45,3 +46,39 @@ def trend(self, org_slug: str, query_params: dict = None) -> dict:
         log.error(f"Error getting historical trend: {response.status_code}")
         log.error(response.text)
         return {}
+
+    class Snapshots:
+        """Submodule for managing historical snapshots."""
+
+        def __init__(self, api):
+            self.api = api
+
+        def create(self, org_slug: str) -> dict:
+            """Create a new snapshot for an organization.
+
+            Args:
+                org_slug: Organization slug
+                data: Dictionary containing snapshot data
+            """
+            path = f"orgs/{org_slug}/historical/snapshots"
+            response = self.api.do_request(path=path, method="POST")
+            if response.status_code == 200:
+                return response.json()
+
+            log.error(f"Error creating snapshot: {response.status_code}")
+            log.error(response.text)
+            return {}
+
+        def list(self, org_slug: str, query_params: dict = None) -> dict:
+            """List historical snapshots for an organization."""
+            path = f"orgs/{org_slug}/historical/snapshots"
+            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 listing snapshots: {response.status_code}")
+            log.error(response.text)
+            return {}

socketdev/labels/__init__.py

@@ -0,0 +1,121 @@
+import json
+import logging
+from typing import Any
+from urllib.parse import urlencode
+
+
+log = logging.getLogger("socketdev")
+
+class Setting:
+    def __init__(self, api):
+        self.api = api
+
+    def create_url(self, org_slug: str, label_id: int):
+        return "orgs/" + org_slug + f"/repos/labels/{label_id}/label-setting"
+
+    def get(self, org_slug: str, label_id: int, setting_key: str):
+        url = self.create_url(org_slug, label_id)
+        path = f"{url}?setting_key={setting_key}"
+        response = self.api.do_request(path=path)
+        if response.status_code == 201:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error getting label setting {setting_key} for {label_id}: {response.status_code}, message: {error_message}")
+        return {}
+
+    def put(self, org_slug: str, label_id: int, settings: dict[str, dict[str, dict[str, str]]]):
+        path = self.create_url(org_slug, label_id)
+        response = self.api.do_request(method="PUT", path=path, payload=json.dumps(settings))
+
+        if response.status_code == 201:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error updating label settings for {label_id}: {response.status_code}, message: {error_message}")
+        return {}
+
+    def delete(self, org_slug: str, label_id: int, settings_key: str):
+        path = self.create_url(org_slug, label_id)
+        path += "?setting_key=" + settings_key
+        response = self.api.do_request(path=path, method="DELETE")
+
+        if response.status_code == 201:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error updating label settings for {label_id}: {response.status_code}, message: {error_message}")
+        return {}
+
+
+class Labels:
+    def __init__(self, api):
+        self.api = api
+        self.setting = Setting(api)
+
+    def list(self, org_slug: str):
+        path = f"orgs/" + org_slug + f"/repos/labels"
+        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 labels: {response.status_code}, message: {error_message}")
+        return {}
+
+    def post(self, org_slug: str, label_name: str) -> dict:
+        path = f"orgs/{org_slug}/repos/labels"
+        payload = json.dumps({"name": label_name})
+        response = self.api.do_request(path=path, method="POST", payload=payload)
+
+        if response.status_code == 201:
+            result = response.json()
+            return result
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        print(f"Failed to create repository label: {response.status_code}, message: {error_message}")
+        return {}
+
+    def get(self, org_slug: str, label_id: str) -> dict:
+        path = f"orgs/{org_slug}/repos/labels/{label_id}"
+        response = self.api.do_request(path=path)
+        if response.status_code == 200:
+            result = response.json()
+            return result
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        print(f"Failed to get repository label: {response.status_code}, message: {error_message}")
+        return {}
+
+    def delete(self, org_slug: str, label_id: str) -> dict:
+        path = f"orgs/{org_slug}/repos/labels/{label_id}"
+        response = self.api.do_request(path=path, method="DELETE")
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error deleting repository label: {response.status_code}, message: {error_message}")
+        return {}
+
+
+    def associate(self, org_slug: str, label_id: int, repo_id: str) -> dict:
+        path = f"orgs/{org_slug}/repos/labels/{label_id}/associate"
+        payload = json.dumps({"repository_id": repo_id})
+        response = self.api.do_request(path=path, method="POST", payload=payload)
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error associating repository label: {response.status_code}, message: {error_message}")
+        return {}
+
+    def disassociate(self, org_slug: str, label_id: int, repo_id: str) -> dict:
+        path = f"orgs/{org_slug}/repos/labels/{label_id}/disassociate"
+        payload = json.dumps({"repository_id": repo_id})
+        response = self.api.do_request(path=path, method="POST", payload=payload)
+        if response.status_code == 200:
+            return response.json()
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Error associating repository label: {response.status_code}, message: {error_message}")
+        return {}

socketdev/version.py

@@ -1 +1 @@
-__version__ = "2.0.21"
+__version__ = "2.1.0"