Doug/add new endpoints (#36)

* Adding new endpoints

* Added new endpoints

* Added new endpoints to docs

Author: dacoburn

README.rst

@@ -575,3 +575,205 @@ Retrieve the Issues associated with a package and version.
 
 - **package (str)** - The name of the NPM package.
 - **version (str)** - The version of the NPM Package.
+
+labels.list(org_slug)
+"""""""""""""""""""""""
+List all repository labels for the given organization.
+
+**Usage:**
+
+.. code-block:: python
+
+    from socketdev import socketdev
+
+    socket = socketdev(token="REPLACE_ME")
+    print(socket.labels.list("org_slug"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+
+labels.post(org_slug, label_name)
+"""""""""""""""""""""""""""""""""""
+Create a new label in the organization.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.post("org_slug", "my-label"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_name (str)** – Name of the label to create
+
+labels.get(org_slug, label_id)
+"""""""""""""""""""""""""""""""""
+Retrieve a single label by its ID.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.get("org_slug", "label_id"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (str)** – The label ID
+
+labels.delete(org_slug, label_id)
+"""""""""""""""""""""""""""""""""""
+Delete a label by ID.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.delete("org_slug", "label_id"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (str)** – The label ID
+
+labels.associate(org_slug, label_id, repo_id)
+"""""""""""""""""""""""""""""""""""""""""""""""
+Associate a label with a repository.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.associate("org_slug", 1234, "repo_id"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (int)** – The label ID
+- **repo_id (str)** – The repository ID
+
+labels.disassociate(org_slug, label_id, repo_id)
+"""""""""""""""""""""""""""""""""""""""""""""""""
+Disassociate a label from a repository.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.disassociate("org_slug", 1234, "repo_id"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (int)** – The label ID
+- **repo_id (str)** – The repository ID
+
+labels.setting.get(org_slug, label_id, setting_key)
+"""""""""""""""""""""""""""""""""""""""""""""""""""""
+Get a setting for a specific label.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.setting.get("org_slug", 1234, "severity"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (int)** – The label ID
+- **setting_key (str)** – The key of the setting
+
+labels.setting.put(org_slug, label_id, settings)
+"""""""""""""""""""""""""""""""""""""""""""""""""""
+Update settings for a specific label.
+
+**Usage:**
+
+.. code-block:: python
+
+    settings = {"severity": {"value": {"level": "high"}}}
+    print(socket.labels.setting.put("org_slug", 1234, settings))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (int)** – The label ID
+- **settings (dict)** – A dictionary of label settings
+
+labels.setting.delete(org_slug, label_id, setting_key)
+"""""""""""""""""""""""""""""""""""""""""""""""""""""""
+Delete a setting from a label.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.labels.setting.delete("org_slug", 1234, "severity"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **label_id (int)** – The label ID
+- **setting_key (str)** – The setting key to delete
+
+historical.list(org_slug, query_params=None)
+"""""""""""""""""""""""""""""""""""""""""""""""
+List historical alerts for an organization.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.historical.list("org_slug", {"repo": "example-repo"}))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **query_params (dict, optional)** – Optional query parameters
+
+historical.trend(org_slug, query_params=None)
+"""""""""""""""""""""""""""""""""""""""""""""""
+Retrieve alert trend data across time.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.historical.trend("org_slug", {"range": "30d"}))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **query_params (dict, optional)** – Optional query parameters
+
+historical.snapshots.create(org_slug)
+""""""""""""""""""""""""""""""""""""""""
+Create a new snapshot of historical data.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.historical.snapshots.create("org_slug"))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+
+historical.snapshots.list(org_slug, query_params=None)
+"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
+List all historical snapshots for an organization.
+
+**Usage:**
+
+.. code-block:: python
+
+    print(socket.historical.snapshots.list("org_slug", {"repo": "example-repo"}))
+
+**PARAMETERS:**
+
+- **org_slug (str)** – The organization name
+- **query_params (dict, optional)** – Optional query parameters

example-socket-export.py

@@ -0,0 +1,54 @@
+import json
+import os
+import logging
+from socketdev import socketdev
+logging.basicConfig(level=logging.INFO)
+
+
+sdk = socketdev(token=os.getenv("SOCKET_SECURITY_API_KEY"))
+
+orgs = sdk.org.get()
+if len(orgs) > 0:
+    org_id = None
+    org_slug = None
+    for org_key in orgs['organizations']:
+        org = orgs['organizations'][org_key]
+        org_id = org_key
+        org_slug = org['slug']
+else:
+    print("Something went wrong with getting org info")
+    exit(1)
+per_page = 100
+response = sdk.repos.get(org_slug, per_page=per_page)
+next_page = response.get("nextPage")
+repos = response.get("results")
+while next_page is not None and next_page != 0:
+    response = sdk.repos.get(org_slug, per_page=per_page, page=next_page)
+    next_page = response.get("nextPage")
+    repos.extend(response.get("results"))
+    if len(response.get("results", [])) == 0:
+        break
+
+# repos = repos[:20]
+head_full_scans_ids = []
+for repo in repos:
+    head_full_scans_id = repo.get("head_full_scan_id")
+    if head_full_scans_id:
+        head_full_scans_ids.append(head_full_scans_id)
+
+socket_results = {}
+for head_full_scan_id in head_full_scans_ids:
+    full_scan_metadata = sdk.fullscans.metadata(org_slug=org_slug, full_scan_id=head_full_scan_id)
+    full_scan_result = {
+        "repo": full_scan_metadata.get("repo"),
+        "branch": full_scan_metadata.get("branch"),
+        "commit_hash": full_scan_metadata.get("commit_hash"),
+        "commit_message": full_scan_metadata.get("commit_message"),
+        "pull_request_url": full_scan_metadata.get("pull_request_url"),
+        "committers": full_scan_metadata.get("committers"),
+        "created_at": full_scan_metadata.get("created_at"),
+        "results": sdk.fullscans.stream(org_slug=org_slug, full_scan_id=head_full_scan_id) or None
+    }
+    socket_results[head_full_scan_id] = full_scan_result
+
+print(json.dumps(socket_results, indent=4))

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/apitokens/__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 {}