Skip to content

Commit 9fafec1

Browse files
flowstateflowstate
committed
added new endpoints
1 parent e0ca3cd commit 9fafec1

File tree

5 files changed

+179
-8
lines changed

5 files changed

+179
-8
lines changed

‎socketdev/__init__.py‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
from socketdev.dependencies import Dependencies
55
from socketdev.export import Export
66
from socketdev.fullscans import FullScans
7+
from socketdev.historical import Historical
78
from socketdev.npm import NPM
89
from socketdev.openapi import OpenAPI
910
from socketdev.org import Orgs
@@ -14,6 +15,7 @@
1415
from socketdev.repositories import Repositories
1516
from socketdev.sbom import Sbom
1617
from socketdev.settings import Settings
18+
from socketdev.triage import Triage
1719
from socketdev.utils import Utils, IntegrationType, INTEGRATION_TYPES
1820
from socketdev.version import __version__
1921

@@ -42,18 +44,20 @@ def __init__(self, token: str, timeout: int = 1200):
4244
self.api.set_timeout(timeout)
4345

4446
self.dependencies = Dependencies(self.api)
47+
self.export = Export(self.api)
48+
self.fullscans = FullScans(self.api)
49+
self.historical = Historical(self.api)
4550
self.npm = NPM(self.api)
4651
self.openapi = OpenAPI(self.api)
4752
self.org = Orgs(self.api)
53+
self.purl = Purl(self.api)
4854
self.quota = Quota(self.api)
4955
self.report = Report(self.api)
50-
self.sbom = Sbom(self.api)
51-
self.purl = Purl(self.api)
52-
self.fullscans = FullScans(self.api)
53-
self.export = Export(self.api)
54-
self.repositories = Repositories(self.api)
5556
self.repos = Repos(self.api)
57+
self.repositories = Repositories(self.api)
58+
self.sbom = Sbom(self.api)
5659
self.settings = Settings(self.api)
60+
self.triage = Triage(self.api)
5761
self.utils = Utils()
5862

5963
@staticmethod

‎socketdev/dependencies/__init__.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ def post(self, files: list, params: dict) -> dict:
2222
else:
2323
result = {}
2424
log.error(f"Error posting {files} to the Dependency API")
25-
print(response.text)
25+
log.error(response.text)
2626
return result
2727

2828
def get(
@@ -38,6 +38,6 @@ def get(
3838
result = response.json()
3939
else:
4040
result = {}
41-
print("Unable to retrieve Dependencies")
42-
print(response.text)
41+
log.error("Unable to retrieve Dependencies")
42+
log.error(response.text)
4343
return result

‎socketdev/historical/__init__.py‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
import logging
2+
from urllib.parse import urlencode
3+
4+
log = logging.getLogger("socketdev")
5+
6+
7+
class Historical:
8+
def __init__(self, api):
9+
self.api = api
10+
11+
def list(self, org_slug: str, query_params: dict = None) -> dict:
12+
"""Get historical alerts list for an organization.
13+
14+
Args:
15+
org_slug: Organization slug
16+
query_params: Optional dictionary of query parameters
17+
"""
18+
path = f"orgs/{org_slug}/alerts/historical"
19+
if query_params:
20+
path += "?" + urlencode(query_params)
21+
22+
response = self.api.do_request(path=path)
23+
if response.status_code == 200:
24+
return response.json()
25+
26+
log.error(f"Error getting historical alerts: {response.status_code}")
27+
log.error(response.text)
28+
return {}
29+
30+
def trend(self, org_slug: str, query_params: dict = None) -> dict:
31+
"""Get historical alerts trend data for an organization.
32+
33+
Args:
34+
org_slug: Organization slug
35+
query_params: Optional dictionary of query parameters
36+
"""
37+
path = f"orgs/{org_slug}/alerts/historical/trend"
38+
if query_params:
39+
path += "?" + urlencode(query_params)
40+
41+
response = self.api.do_request(path=path)
42+
if response.status_code == 200:
43+
return response.json()
44+
45+
log.error(f"Error getting historical trend: {response.status_code}")
46+
log.error(response.text)
47+
return {}

‎socketdev/settings/__init__.py‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,3 +97,76 @@ def get(
9797
{"securityPolicyRules": {}, "success": False, "status": response.status_code, "message": error_message}
9898
)
9999
return {}
100+
101+
def integration_events(self, org_slug: str, integration_id: str) -> dict:
102+
"""Get integration events for a specific integration.
103+
104+
Args:
105+
org_slug: Organization slug
106+
integration_id: Integration ID
107+
"""
108+
path = f"orgs/{org_slug}/settings/integrations/{integration_id}"
109+
response = self.api.do_request(path=path)
110+
111+
if response.status_code == 200:
112+
return response.json()
113+
114+
error_message = response.json().get("error", {}).get("message", "Unknown error")
115+
log.error(f"Error getting integration events: {response.status_code}, message: {error_message}")
116+
return {}
117+
118+
def get_license_policy(self, org_slug: str) -> dict:
119+
"""Get license policy settings for an organization.
120+
121+
Args:
122+
org_slug: Organization slug
123+
"""
124+
path = f"orgs/{org_slug}/settings/license-policy"
125+
response = self.api.do_request(path=path)
126+
127+
if response.status_code == 200:
128+
return response.json()
129+
130+
error_message = response.json().get("error", {}).get("message", "Unknown error")
131+
log.error(f"Error getting license policy: {response.status_code}, message: {error_message}")
132+
return {}
133+
134+
def update_security_policy(self, org_slug: str, body: dict, custom_rules_only: bool = False) -> dict:
135+
"""Update security policy settings for an organization.
136+
137+
Args:
138+
org_slug: Organization slug
139+
body: Security policy configuration to update
140+
custom_rules_only: Optional flag to update only custom rules
141+
"""
142+
path = f"orgs/{org_slug}/settings/security-policy"
143+
if custom_rules_only:
144+
path += "?custom_rules_only=true"
145+
146+
response = self.api.do_request(path=path, method="POST", payload=body)
147+
148+
if response.status_code == 200:
149+
return response.json()
150+
151+
error_message = response.json().get("error", {}).get("message", "Unknown error")
152+
log.error(f"Error updating security policy: {response.status_code}, message: {error_message}")
153+
return {}
154+
155+
def update_license_policy(self, org_slug: str, body: dict, merge_update: bool = False) -> dict:
156+
"""Update license policy settings for an organization.
157+
158+
Args:
159+
org_slug: Organization slug
160+
body: License policy configuration to update
161+
merge_update: Optional flag to merge updates instead of replacing (defaults to False)
162+
"""
163+
path = f"orgs/{org_slug}/settings/license-policy?merge_update={str(merge_update).lower()}"
164+
165+
response = self.api.do_request(path=path, method="POST", payload=body)
166+
167+
if response.status_code == 200:
168+
return response.json()
169+
170+
error_message = response.json().get("error", {}).get("message", "Unknown error")
171+
log.error(f"Error updating license policy: {response.status_code}, message: {error_message}")
172+
return {}

‎socketdev/triage/__init__.py‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
import logging
2+
from urllib.parse import urlencode
3+
4+
log = logging.getLogger("socketdev")
5+
6+
7+
class Triage:
8+
def __init__(self, api):
9+
self.api = api
10+
11+
def list_alert_triage(self, org_slug: str, query_params: dict = None) -> dict:
12+
"""Get list of triaged alerts for an organization.
13+
14+
Args:
15+
org_slug: Organization slug
16+
query_params: Optional dictionary of query parameters
17+
"""
18+
path = f"orgs/{org_slug}/triage/alerts"
19+
if query_params:
20+
path += "?" + urlencode(query_params)
21+
22+
response = self.api.do_request(path=path)
23+
24+
if response.status_code == 200:
25+
return response.json()
26+
27+
error_message = response.json().get("error", {}).get("message", "Unknown error")
28+
log.error(f"Error getting alert triage list: {response.status_code}, message: {error_message}")
29+
return {}
30+
31+
def update_alert_triage(self, org_slug: str, body: dict) -> dict:
32+
"""Update triaged alerts for an organization.
33+
34+
Args:
35+
org_slug: Organization slug
36+
body: Alert triage configuration to update
37+
"""
38+
path = f"orgs/{org_slug}/triage/alerts"
39+
40+
response = self.api.do_request(path=path, method="POST", payload=body)
41+
42+
if 200 <= response.status_code < 300:
43+
return response.json()
44+
45+
error_message = response.json().get("error", {}).get("message", "Unknown error")
46+
log.error(f"Error updating alert triage: {response.status_code}, message: {error_message}")
47+
return {}

0 commit comments

Comments
 (0)