From bc5b3b761d033409497df52ad115fffe34e82ed3 Mon Sep 17 00:00:00 2001 From: lelia <2418071+lelia@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:26:56 -0400 Subject: [PATCH 1/4] feat(enums): make every API-sourced enum forward-compatible Five of the six enums populated from API responses coerced strictly, so a value the backend added would raise ValueError inside from_dict and empty an entire response rather than degrade one field. That is the mechanism behind issue #78 and the unknown `generic` purl type; each was fixed on the single enum that fired, leaving the rest holding the same landmine. All five now fall back to a documented member and log the unrecognized value. The fallbacks are chosen rather than convenient: SocketIssueSeverity and DiffType gain an explicit UNKNOWN, since guessing an existing level would either hide a real finding or invent one, and SecurityAction defers. A generalized test discovers every enum in the package, including ones added later, and fails if any raises. A drift check compares the enums against the public OpenAPI spec; it found 10 purl types the SDK was missing, which are added here. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/api-drift-check.yml | 37 +++++++ CHANGELOG.md | 35 +++++++ scripts/check_api_enum_drift.py | 139 +++++++++++++++++++++++++ socketdev/core/enums.py | 35 +++++++ socketdev/fullscans/__init__.py | 59 +++++++---- socketdev/settings/__init__.py | 10 ++ socketdev/version.py | 2 +- tests/unit/test_enum_forward_compat.py | 110 +++++++++++++++++++ 8 files changed, 407 insertions(+), 20 deletions(-) create mode 100644 .github/workflows/api-drift-check.yml create mode 100755 scripts/check_api_enum_drift.py create mode 100644 socketdev/core/enums.py create mode 100644 tests/unit/test_enum_forward_compat.py diff --git a/.github/workflows/api-drift-check.yml b/.github/workflows/api-drift-check.yml new file mode 100644 index 0000000..e25c55a --- /dev/null +++ b/.github/workflows/api-drift-check.yml @@ -0,0 +1,37 @@ +name: API Drift Check + +# Compares this SDK's hand-maintained enums against the live Socket OpenAPI +# spec. The spec is public and unauthenticated, so this job needs no secrets, +# no org and no fixture data. +# +# Deliberately NOT a pull_request trigger: it tests the API, not the diff, and a +# backend change must never block an unrelated SDK pull request. +# +# TODO: once this has run green manually a few times, uncomment the schedule +# below to turn it into an early-warning signal instead of a manual check. +on: + workflow_dispatch: + # schedule: + # - cron: '0 14 * * 1' # Mondays, 14:00 UTC + +permissions: + contents: read + +jobs: + enum-drift: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 1 + persist-credentials: false + + - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 + with: + python-version: "3.12" + + - name: Install the SDK + run: python -m pip install . + + - name: Compare SDK enums against the live OpenAPI spec + run: python scripts/check_api_enum_drift.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 14a7ce2..b002357 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,40 @@ # Changelog +## 3.6.0 + +### Changed: every API-sourced enum now tolerates unknown values + +- `SocketIssueSeverity`, `SocketCategory`, `DiffType`, `ScanType` and + `SecurityAction` now fall back to a documented member instead of raising + `ValueError` when the API sends a value this release does not know about. + `SocketPURL_Type` already behaved this way; the other five did not, so each + was one backend addition away from emptying a response the same way issue #78 + and the unknown `generic` purl type did. +- Fallbacks are deliberate rather than convenient. `SocketIssueSeverity` and + `DiffType` gained an explicit `UNKNOWN` member because guessing an existing + level would either hide a real finding or invent one, and `SecurityAction` + falls back to `DEFER` because that already means "use the configured + default". Every fallback logs a warning naming the unrecognized value. +- Added the 10 purl types the API defines that this SDK was missing: `alpm`, + `chrome`, `clawhub`, `edge-extension`, `firefox-extension`, `qpkg`, `socket`, + `swid`, `vscode` and `vscode-extension`. Artifacts with those types were + being flattened to `unknown`. + +### Added: enum forward-compatibility is now an enforced invariant + +- `tests/unit/test_enum_forward_compat.py` discovers every enum in the package, + including ones added later, and fails if any raises on an unrecognized value. + The two prior incidents were each fixed with a bespoke test on the single + enum that happened to fire; this replaces that pattern. + +### Added: scheduled check for enum drift against the live API + +- `scripts/check_api_enum_drift.py` compares the SDK's enums against + `https://api.socket.dev/v0/openapi`. The spec is public, so the check needs no + token, org or fixture data. Run by `.github/workflows/api-drift-check.yml`, + which is manual-dispatch only for now and is not a pull request check --- it + tests the API rather than the diff. + ## 3.5.0 ### Changed: bound runtime dependency ranges and pin build backend diff --git a/scripts/check_api_enum_drift.py b/scripts/check_api_enum_drift.py new file mode 100755 index 0000000..0c1f2da --- /dev/null +++ b/scripts/check_api_enum_drift.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""Compare this SDK's enums against the live Socket OpenAPI specification. + +The SDK's enums are hand-maintained copies of value sets the API owns, and +nothing has ever told us when the API adds to one. Both prior incidents were +both found by a customer, after the fact, via an empty report. + +The spec at ``https://api.socket.dev/v0/openapi`` is public and unauthenticated, +so this needs no token, no org and no fixture data -- it is a plain GET plus a +set comparison. It catches schema drift only; it deliberately says nothing about +runtime behaviour, response shapes or auth, which need the authenticated +integration checks. + +Exit codes: + 0 no missing values (extras and unmapped enums are reported, not fatal) + 1 the API defines values this SDK does not know about + 2 the spec could not be fetched or parsed +""" + +import argparse +import json +import sys +import urllib.error +import urllib.request + +from socketdev.fullscans import ( + DiffType, + ScanType, + SocketCategory, + SocketIssueSeverity, + SocketPURL_Type, +) +from socketdev.settings import SecurityAction + +DEFAULT_SPEC_URL = "https://api.socket.dev/v0/openapi" + +# SDK enum -> the schema in components/schemas that defines the same value set. +# None means the API does not expose the value set as a named schema, so this +# check cannot cover it; those are reported so the gap stays visible rather than +# looking like a pass. +ENUM_TO_SCHEMA = { + SocketPURL_Type: "SocketPURL_Type", + SocketIssueSeverity: "SocketIssueSeverity", + SocketCategory: "SocketCategory", + DiffType: "SocketDiffArtifactType", + ScanType: None, + SecurityAction: None, +} + +# Members this SDK adds deliberately, which the API will never send. They are +# the documented _missing_ fallbacks (see socketdev/core/enums.py), so their +# absence from the spec is expected rather than drift. +SDK_ONLY_VALUES = { + "SocketPURL_Type": {"unknown"}, + "SocketIssueSeverity": {"unknown"}, + "SocketCategory": {"miscellaneous"}, + "DiffType": {"unknown"}, +} + + +def fetch_spec(url): + with urllib.request.urlopen(url, timeout=60) as response: + if response.status != 200: + raise RuntimeError(f"{url} returned HTTP {response.status}") + return json.load(response) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--spec-url", default=DEFAULT_SPEC_URL) + args = parser.parse_args() + + try: + spec = fetch_spec(args.spec_url) + schemas = spec["components"]["schemas"] + except (urllib.error.URLError, KeyError, ValueError, RuntimeError) as exc: + print(f"::error::could not read the OpenAPI spec: {exc}") + return 2 + + drifted = False + unmapped = [] + + for enum_cls, schema_name in ENUM_TO_SCHEMA.items(): + if schema_name is None: + unmapped.append(enum_cls.__name__) + continue + schema = schemas.get(schema_name) + if not schema or "enum" not in schema: + print( + f"::warning::schema {schema_name!r} for {enum_cls.__name__} is gone " + f"or no longer an enum -- the API may have restructured it" + ) + continue + + api_values = set(schema["enum"]) + sdk_values = {member.value for member in enum_cls} + sdk_only = SDK_ONLY_VALUES.get(enum_cls.__name__, set()) + + missing = sorted(api_values - sdk_values) + extra = sorted(sdk_values - api_values - sdk_only) + + if missing: + drifted = True + print( + f"::error::{enum_cls.__name__} is missing {len(missing)} value(s) " + f"the API defines: {', '.join(missing)}" + ) + if extra: + # Not fatal: the spec omitting a value the SDK accepts is usually a + # spec gap, and dropping a member would be a breaking change. + print( + f"::warning::{enum_cls.__name__} defines {len(extra)} value(s) " + f"absent from the spec: {', '.join(extra)}" + ) + if not missing and not extra: + print(f"ok: {enum_cls.__name__} matches {schema_name} " + f"({len(api_values)} values)") + + if unmapped: + print( + f"::warning::not covered by this check, because the API exposes no " + f"named schema for them: {', '.join(sorted(unmapped))}" + ) + + if drifted: + print( + "\nAdd the missing members to the SDK enum. Existing values are " + "still parsed correctly in the meantime -- _missing_ maps unknown " + "values to a fallback -- so this is a loss of fidelity, not an " + "outage." + ) + return 1 + + print("\nno missing enum values") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/socketdev/core/enums.py b/socketdev/core/enums.py new file mode 100644 index 0000000..a2b7d85 --- /dev/null +++ b/socketdev/core/enums.py @@ -0,0 +1,35 @@ +"""Forward-compatibility helpers for enums sourced from the Socket API. + +The Socket API adds values to its enums (purl types, alert categories, policy +actions) without a corresponding SDK release. When a strict ``Enum`` coercion +meets one of those values it raises ``ValueError``, and because coercion happens +inside ``from_dict`` that error takes down the whole response parse rather than +the single field it applies to. That is how issue #78 (an unknown +``SocketCategory``) and the unknown ``generic`` purl type each produced +empty reports from otherwise-successful scans. + +Every enum in this package that is populated from an API response therefore +defines ``_missing_`` and falls back to a documented sentinel instead of +raising. ``tests/unit/test_enum_forward_compat.py`` enforces that as an +invariant across the package, including for enums added later. +""" + +import logging + +log = logging.getLogger("socketdev") + + +def unknown_enum_value(enum_name: str, value: object, fallback): + """Log an unrecognized API enum value and return the enum's fallback member. + + Callers are ``_missing_`` implementations, so returning ``fallback`` is what + turns the would-be ``ValueError`` into a usable member. + """ + log.warning( + "Unknown %s %r; falling back to %s. " + "Upgrade socketdev to pick up newer values.", + enum_name, + value, + fallback.name, + ) + return fallback diff --git a/socketdev/fullscans/__init__.py b/socketdev/fullscans/__init__.py index 2fe6f09..770440d 100644 --- a/socketdev/fullscans/__init__.py +++ b/socketdev/fullscans/__init__.py @@ -5,6 +5,7 @@ from dataclasses import dataclass, asdict, field import urllib.parse from ..core.dedupe import Dedupe +from ..core.enums import unknown_enum_value from ..utils import IntegrationType, Utils log = logging.getLogger("socketdev") @@ -12,9 +13,12 @@ class SocketPURL_Type(str, Enum): UNKNOWN = "unknown" + ALPM = "alpm" APK = "apk" BITBUCKET = "bitbucket" CARGO = "cargo" + CHROME = "chrome" + CLAWHUB = "clawhub" COCOAPODS = "cocoapods" COMPOSER = "composer" CONAN = "conan" @@ -22,6 +26,8 @@ class SocketPURL_Type(str, Enum): CRAN = "cran" DEB = "deb" DOCKER = "docker" + EDGE_EXTENSION = "edge-extension" + FIREFOX_EXTENSION = "firefox-extension" GEM = "gem" GENERIC = "generic" GITHUB = "github" @@ -37,21 +43,17 @@ class SocketPURL_Type(str, Enum): OCI = "oci" PUB = "pub" PYPI = "pypi" + QPKG = "qpkg" RPM = "rpm" + SOCKET = "socket" + SWID = "swid" SWIFT = "swift" + VSCODE = "vscode" + VSCODE_EXTENSION = "vscode-extension" @classmethod def _missing_(cls, value): - # The API can emit purl types this SDK does not know about yet. Fall - # back to UNKNOWN instead of raising so one artifact cannot fail an - # entire response parse (same forward-compat approach as - # SocketCategory, https://github.com/SocketDev/socket-sdk-python/issues/78). - log.warning( - "Unknown SocketPURL_Type %r; falling back to UNKNOWN. " - "Upgrade socketdev to pick up newer purl types.", - value, - ) - return cls.UNKNOWN + return unknown_enum_value(cls.__name__, value, cls.UNKNOWN) class SocketIssueSeverity(str, Enum): @@ -59,6 +61,15 @@ class SocketIssueSeverity(str, Enum): MIDDLE = "middle" HIGH = "high" CRITICAL = "critical" + # Sentinel for severities the API adds later. Deliberately not an existing + # level: falling back to LOW would hide a serious finding and CRITICAL + # would manufacture one, so consumers get an explicit "not understood" + # value to branch on instead of a guess. + UNKNOWN = "unknown" + + @classmethod + def _missing_(cls, value): + return unknown_enum_value(cls.__name__, value, cls.UNKNOWN) class SocketCategory(str, Enum): @@ -70,6 +81,10 @@ class SocketCategory(str, Enum): MISCELLANEOUS = "miscellaneous" OTHER = "other" # Added to match backend API responses + @classmethod + def _missing_(cls, value): + return unknown_enum_value(cls.__name__, value, cls.MISCELLANEOUS) + class DiffType(str, Enum): ADDED = "added" @@ -77,12 +92,24 @@ class DiffType(str, Enum): UNCHANGED = "unchanged" REPLACED = "replaced" UPDATED = "updated" + # Not UNCHANGED: treating an unrecognized change as "nothing happened" + # would drop a real diff entry out of a comparison silently. + UNKNOWN = "unknown" + + @classmethod + def _missing_(cls, value): + return unknown_enum_value(cls.__name__, value, cls.UNKNOWN) class ScanType(str, Enum): SOCKET = "socket" SOCKET_TIER1 = "socket_tier1" SOCKET_BASICS = "socket_basics" + UNKNOWN = "unknown" + + @classmethod + def _missing_(cls, value): + return unknown_enum_value(cls.__name__, value, cls.UNKNOWN) @dataclass(kw_only=True) @@ -481,15 +508,9 @@ def to_dict(self): @classmethod def from_dict(cls, data: dict) -> "SocketAlert": - try: - category = SocketCategory(data["category"]) - except ValueError: - log.warning( - "Unknown SocketCategory %r; falling back to MISCELLANEOUS. " - "Upgrade socketdev to pick up newer categories.", - data["category"], - ) - category = SocketCategory.MISCELLANEOUS + # SocketCategory._missing_ handles unrecognized values; see + # socketdev/core/enums.py. + category = SocketCategory(data["category"]) return cls( key=data["key"], type=data["type"], diff --git a/socketdev/settings/__init__.py b/socketdev/settings/__init__.py index a9cbe8e..3812726 100644 --- a/socketdev/settings/__init__.py +++ b/socketdev/settings/__init__.py @@ -3,6 +3,8 @@ from typing import Dict, Optional, Union from dataclasses import dataclass, asdict +from ..core.enums import unknown_enum_value + log = logging.getLogger("socketdev") @@ -13,6 +15,14 @@ class SecurityAction(str, Enum): MONITOR = "monitor" IGNORE = "ignore" + @classmethod + def _missing_(cls, value): + # DEFER, not a new UNKNOWN sentinel: an unrecognized action is a policy + # decision this SDK cannot make, and "defer" already means "use the + # configured default". IGNORE would silently disable a rule and ERROR + # would fail builds on a value the API considers routine. + return unknown_enum_value(cls.__name__, value, cls.DEFER) + @dataclass class SecurityPolicyRule: diff --git a/socketdev/version.py b/socketdev/version.py index dcbfb52..85197cb 100644 --- a/socketdev/version.py +++ b/socketdev/version.py @@ -1 +1 @@ -__version__ = "3.5.0" +__version__ = "3.6.0" diff --git a/tests/unit/test_enum_forward_compat.py b/tests/unit/test_enum_forward_compat.py new file mode 100644 index 0000000..3511b32 --- /dev/null +++ b/tests/unit/test_enum_forward_compat.py @@ -0,0 +1,110 @@ +"""Package-wide invariant: API-sourced enums must tolerate unknown values. + +This generalizes two point fixes. Issue #78 (an unrecognized ``SocketCategory``) +and the unrecognized ``generic`` purl type were the same bug: the Socket +API added an enum value, the SDK coerced it strictly inside ``from_dict``, and +the resulting ``ValueError`` emptied an entire response instead of degrading one +field. Each was fixed on the one enum that happened to fire, leaving the others +holding the same landmine. + +Rather than add a third bespoke regression test the next time it happens, this +discovers every ``Enum`` in the package -- including ones added after this file +was written -- and asserts the invariant directly. A new enum has to opt out +explicitly and say why. +""" + +import enum +import importlib +import logging +import pkgutil +import unittest + +import socketdev + +# Enums that are only ever used to *build* requests, never to parse a response. +# Strictness is correct there: a bad value is the caller's typo and should raise +# rather than be silently coerced. Add an entry only with a comment justifying +# that the enum never sees API-supplied values. +REQUEST_ONLY_ENUMS = frozenset() + +# A value the API will never legitimately send. +SENTINEL = "__value_the_api_would_never_send__" + + +def _all_enums(): + """Every Enum subclass defined under the socketdev package.""" + found = {} + for module_info in pkgutil.walk_packages( + socketdev.__path__, prefix="socketdev." + ): + try: + module = importlib.import_module(module_info.name) + except Exception: # pragma: no cover - an unimportable module is its own bug + continue + for name in dir(module): + obj = getattr(module, name) + if ( + isinstance(obj, type) + and issubclass(obj, enum.Enum) + and obj.__module__.startswith("socketdev") + and len(obj) > 0 + ): + found[f"{obj.__module__}.{obj.__name__}"] = obj + return found + + +class TestEnumForwardCompatibility(unittest.TestCase): + """Every response-parsed enum degrades instead of raising.""" + + def test_enums_are_discovered(self): + # Guards against the discovery walk silently finding nothing, which + # would make every other test in this file vacuously pass. + self.assertGreaterEqual( + len(_all_enums()), 6, "enum discovery found suspiciously few enums" + ) + + def test_unknown_value_does_not_raise(self): + for qualname, enum_cls in sorted(_all_enums().items()): + if enum_cls.__name__ in REQUEST_ONLY_ENUMS: + continue + with self.subTest(enum=qualname): + try: + result = enum_cls(SENTINEL) + except ValueError: + self.fail( + f"{qualname} raised ValueError on an unrecognized value. " + f"Add a _missing_ that returns a documented fallback " + f"(see socketdev/core/enums.py), or add it to " + f"REQUEST_ONLY_ENUMS with a justification." + ) + self.assertIsInstance( + result, + enum_cls, + f"{qualname}._missing_ must return a member of its own enum", + ) + + def test_unknown_value_warns(self): + # The fallback is a silent downgrade in accuracy, so it has to leave a + # trace that something drifted. + for qualname, enum_cls in sorted(_all_enums().items()): + if enum_cls.__name__ in REQUEST_ONLY_ENUMS: + continue + with self.subTest(enum=qualname): + with self.assertLogs("socketdev", level=logging.WARNING) as captured: + enum_cls(SENTINEL) + self.assertTrue( + any(enum_cls.__name__ in line for line in captured.output), + f"{qualname} fell back without naming itself in the warning; " + f"got: {captured.output}", + ) + + def test_known_values_still_round_trip(self): + # Forward-compat must not swallow legitimate values. + for qualname, enum_cls in sorted(_all_enums().items()): + for member in enum_cls: + with self.subTest(enum=qualname, member=member.name): + self.assertIs(enum_cls(member.value), member) + + +if __name__ == "__main__": + unittest.main() From d99483b237610a8c3c72a7f074c68e04293d733b Mon Sep 17 00:00:00 2001 From: lelia <2418071+lelia@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:26:56 -0400 Subject: [PATCH 2/4] docs: drop Linear IDs from test docstrings and comments Ticket identifiers belong in the pull request description, not in code that outlives the ticket. The GitHub issue reference in the purl test docstring stays, since that resolves for anyone reading the repository. Co-Authored-By: Claude Opus 5 (1M context) --- tests/unit/test_socket_alert_category.py | 2 +- tests/unit/test_socket_purl_type.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_socket_alert_category.py b/tests/unit/test_socket_alert_category.py index c55113b..bf937db 100644 --- a/tests/unit/test_socket_alert_category.py +++ b/tests/unit/test_socket_alert_category.py @@ -34,7 +34,7 @@ def test_known_category_is_preserved(self): self.assertEqual(alert.severity, SocketIssueSeverity.LOW) def test_other_category_is_recognized(self): - # "other" is a known backend category as of CE-225; it should resolve to + # "other" is a known backend category; it should resolve to # SocketCategory.OTHER rather than falling back to MISCELLANEOUS. alert = SocketAlert.from_dict(self._base_payload("other")) self.assertEqual(alert.category, SocketCategory.OTHER) diff --git a/tests/unit/test_socket_purl_type.py b/tests/unit/test_socket_purl_type.py index 6454fbc..691384a 100644 --- a/tests/unit/test_socket_purl_type.py +++ b/tests/unit/test_socket_purl_type.py @@ -1,5 +1,5 @@ """ -Unit tests for lenient SocketPURL_Type parsing (CE-362). +Unit tests for lenient SocketPURL_Type parsing. The Socket API can emit purl types the SDK does not yet know about (e.g. ``"generic"``, which was missing from the enum entirely). Strict enum parsing From 203b5e82616de372d6199ee08333e74f5d366306 Mon Sep 17 00:00:00 2001 From: lelia <2418071+lelia@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:26:56 -0400 Subject: [PATCH 3/4] chore: add .gitattributes and normalize line endings to LF Eleven Python files were committed with CRLF, socketdev/__init__.py among them. Any tooling that reads and rewrites one of those files converts it to LF on the way out, so a two-line edit arrives as a whole-file diff with the real change buried in it. That happened while writing the enum change in this same branch. This normalizes all of them once and pins the setting so it cannot recur. Reviewable with `git diff -w`, which shows only .gitattributes: no file content changed, and the unit suite is unaffected. Co-Authored-By: Claude Opus 5 (1M context) --- .gitattributes | 29 + README.rst | 2970 ++++++++++++++-------------- socketdev/__init__.py | 230 +-- socketdev/core/classes.py | 296 +-- socketdev/dependencies/__init__.py | 106 +- socketdev/npm/__init__.py | 56 +- socketdev/openapi/__init__.py | 38 +- socketdev/org/__init__.py | 68 +- socketdev/quota/__init__.py | 38 +- socketdev/report/__init__.py | 174 +- socketdev/repositories/__init__.py | 62 +- socketdev/settings/__init__.py | 364 ++-- socketdev/tools/__init__.py | 130 +- 13 files changed, 2295 insertions(+), 2266 deletions(-) create mode 100644 .gitattributes diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..6162e5d --- /dev/null +++ b/.gitattributes @@ -0,0 +1,29 @@ +# Normalize line endings to LF in the repository. +# +# Eleven Python files were committed with CRLF, socketdev/__init__.py among +# them. Any tooling that reads and rewrites one of those files converts it to +# LF on the way out, which turns a two-line change into a whole-file diff and +# buries the actual edit. Normalizing once and pinning the setting here stops +# that from recurring. +* text=auto + +*.py text eol=lf +*.pyi text eol=lf +*.md text eol=lf +*.rst text eol=lf +*.toml text eol=lf +*.cfg text eol=lf +*.yml text eol=lf +*.yaml text eol=lf +*.json text eol=lf +*.sh text eol=lf +*.lock text eol=lf + +# Byte-exact, never line-ending-normalized. +*.png binary +*.jpg binary +*.jpeg binary +*.gif binary +*.ico binary +*.whl binary +*.gz binary diff --git a/README.rst b/README.rst index 61b64ce..9ee3762 100644 --- a/README.rst +++ b/README.rst @@ -1,1485 +1,1485 @@ - -socketdev -######### - -Purpose -------- - -The Socket.dev Python SDK provides a wrapper around the Socket.dev REST API to simplify making calls to the API from Python. - -Socket API v0 - https://docs.socket.dev/reference/introduction-to-socket-api - -Initializing the module ------------------------ - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME", timeout=30) - -**PARAMETERS:** - -- **token (str)** - The Socket API Key for your Organization -- **timeout (int)** - The number of seconds to wait before failing the connection -- **allow_unverified (bool)** - Whether to skip SSL certificate verification (default: False). Set to True for testing with self-signed certificates. -- **user_agent (str, optional)** - Custom User-Agent string to use in API requests. If not provided, defaults to "SocketSDKPython/{version}" - -Supported Functions -------------------- - - -purl.post(license, components, org_slug=None) -""""""""""""""""""""""""""""""""""""""""""""" -Retrieve package information for one or more PURLs. Pass ``org_slug`` to use the -current org-scoped endpoint. Omitting ``org_slug`` keeps the legacy deprecated -endpoint for backwards compatibility. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - org_slug = "your-org-slug" - license = "true" - components = [ - { - "purl": "pkg:pypi/pyonepassword@5.0.0" - }, - { - "purl": "pkg:pypi/socketsecurity" - } - ] - print(socket.purl.post(license, components, org_slug=org_slug)) - -**PARAMETERS:** - -- **license (str)** - The license parameter if enabled will show alerts and license information. If disabled will only show the basic package metadata and scores. Default is true -- **components (array{dict})** - The components list of packages urls -- **org_slug (str, optional)** - Organization slug for the supported org-scoped PURL endpoint. If omitted, the SDK uses the deprecated legacy endpoint for backwards compatibility. - -export.cdx_bom(org_slug, id, query_params) -"""""""""""""""""""""""""""""""""""""""""" -Export a Socket SBOM as a CycloneDX SBOM - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - from socketdev.export import ExportQueryParams - - socket = socketdev(token="REPLACE_ME") - query_params = ExportQueryParams( - author="john_doe", - project_name="my-project" - ) - print(socket.export.cdx_bom("org_slug", "sbom_id", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **id (str)** - The ID of either a full scan or an SBOM report -- **query_params (ExportQueryParams)** - Optional query parameters for filtering: - - **author (str)** - Filter by author - - **project_group (str)** - Filter by project group - - **project_name (str)** - Filter by project name - - **project_version (str)** - Filter by project version - - **project_id (str)** - Filter by project ID - -export.spdx_bom(org_slug, id, query_params) -""""""""""""""""""""""""""""""""""""""""""" -Export a Socket SBOM as an SPDX SBOM - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - from socketdev.export import ExportQueryParams - - socket = socketdev(token="REPLACE_ME") - query_params = ExportQueryParams( - project_name="my-project", - project_version="1.0.0" - ) - print(socket.export.spdx_bom("org_slug", "sbom_id", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **id (str)** - The ID of either a full scan or an SBOM report -- **query_params (ExportQueryParams)** - Optional query parameters for filtering: - - **author (str)** - Filter by author - - **project_group (str)** - Filter by project group - - **project_name (str)** - Filter by project name - - **project_version (str)** - Filter by project version - - **project_id (str)** - Filter by project ID - -export.openvex_bom(org_slug, id, query_params) -"""""""""""""""""""""""""""""""""""""""""""""" -Export a Socket SBOM as an OpenVEX SBOM - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - from socketdev.export import ExportQueryParams - - socket = socketdev(token="REPLACE_ME") - query_params = ExportQueryParams( - project_name="my-project", - project_version="1.0.0" - ) - print(socket.export.openvex_bom("org_slug", "sbom_id", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **id (str)** - The ID of either a full scan or an SBOM report -- **query_params (ExportQueryParams)** - Optional query parameters for filtering: - - **author (str)** - Filter by author - - **project_group (str)** - Filter by project group - - **project_name (str)** - Filter by project name - - **project_version (str)** - Filter by project version - - **project_id (str)** - Filter by project ID - -fullscans.get(org_slug, params) -""""""""""""""""""""""""""""""" -Retrieve the Fullscans information for an Organization with query parameters - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - - # Query parameters for filtering full scans - params = { - "repo": "my-repo", - "branch": "main", - "limit": 10, - "offset": 0 - } - print(socket.fullscans.get("org_slug", params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **params (dict)** - Query parameters for filtering results (required) - -fullscans.post(files, params) -""""""""""""""""""""""""""""" -Create a full scan from a set of package manifest files. Returns a full scan including all SBOM artifacts. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - from socketdev.fullscans import FullScanParams - - socket = socketdev(token="REPLACE_ME") - files = [ - "/path/to/manifest/package.json" - ] - params = FullScanParams( - org_slug="org_name", - repo="TestRepo", - workspace="my-workspace", - branch="main", - commit_message="Test Commit Message", - commit_hash="abc123def456", - pull_request=123, - committers=["committer1", "committer2"], - make_default_branch=False, - set_as_pending_head=False - ) - - print(socket.fullscans.post(files, params)) - -**PARAMETERS:** - -- **files (list)** - List of file paths of manifest files -- **params (FullScanParams)** - FullScanParams object containing scan configuration - -+------------------------+------------+-------------------------------------------------------------------------------+ -| Parameter | Required | Description | -+========================+============+===============================================================================+ -| org_slug | True | The string name in a git approved name for organization. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| repo | True | The string name in a git approved name for repositories. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| branch | False | The string name in a git approved name for branches. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| committers | False | List of committer names (List[str]). | -+------------------------+------------+-------------------------------------------------------------------------------+ -| pull_request | False | The integer for the PR or MR number. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| commit_message | False | The string for a commit message if there is one. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| make_default_branch | False | Boolean to signal that this is the default branch. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| commit_hash | False | Optional git commit hash | -+------------------------+------------+-------------------------------------------------------------------------------+ -| set_as_pending_head | False | Boolean to set as pending head | -+------------------------+------------+-------------------------------------------------------------------------------+ -| tmp | False | Boolean temporary flag | -+------------------------+------------+-------------------------------------------------------------------------------+ -| workspace | False | The workspace of the repository to associate the full-scan with. | -+------------------------+------------+-------------------------------------------------------------------------------+ -| integration_type | False | IntegrationType enum value (e.g., "api", "github") | -+------------------------+------------+-------------------------------------------------------------------------------+ -| integration_org_slug | False | Organization slug for integration | -+------------------------+------------+-------------------------------------------------------------------------------+ -| scan_type | False | ScanType enum value: "socket", "socket_tier1", or "socket_basics" | -+------------------------+------------+-------------------------------------------------------------------------------+ - -fullscans.delete(org_slug, full_scan_id) -"""""""""""""""""""""""""""""""""""""""" -Delete an existing full scan. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fullscans.delete("org_slug", "full_scan_id")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **full_scan_id (str)** - The ID of the full scan - -fullscans.stream_diff(org_slug, before, after, use_types=True, include_license_details="true", \*\*kwargs) -"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Stream a diff between two full scans. Returns a scan diff. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fullscans.stream_diff("org_slug", "before_scan_id", "after_scan_id")) - - # With additional parameters - print(socket.fullscans.stream_diff( - "org_slug", - "before_scan_id", - "after_scan_id", - use_types=False, - include_license_details="false" - )) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **before (str)** - The base full scan ID -- **after (str)** - The comparison full scan ID -- **use_types (bool)** - Whether to return typed response objects (default: True) -- **include_license_details (str)** - Include license details ("true"/"false"). Can greatly increase response size. Defaults to "true". -- **kwargs** - Additional query parameters - -fullscans.stream(org_slug, full_scan_id, use_types=False) -""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Stream all SBOM artifacts for a full scan. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fullscans.stream("org_slug", "full_scan_id")) - - # With typed response - print(socket.fullscans.stream("org_slug", "full_scan_id", use_types=True)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **full_scan_id (str)** - The ID of the full scan -- **use_types (bool)** - Whether to return typed response objects (default: False) - -fullscans.metadata(org_slug, full_scan_id, use_types=False) -""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Get metadata for a single full scan - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fullscans.metadata("org_slug", "full_scan_id")) - - # With typed response - print(socket.fullscans.metadata("org_slug", "full_scan_id", use_types=True)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **full_scan_id (str)** - The ID of the full scan -- **use_types (bool)** - Whether to return typed response objects (default: False) - -fullscans.gfm(org_slug, before, after) -"""""""""""""""""""""""""""""""""""""" -Get GitHub Flavored Markdown diff between two full scans. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fullscans.gfm("org_slug", "before_scan_id", "after_scan_id")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **before (str)** - The base full scan ID -- **after (str)** - The comparison full scan ID - -fullscans.finalize_tier1(full_scan_id, tier1_reachability_scan_id) -"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Finalize a tier 1 reachability scan by associating it with a full scan. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - success = socket.fullscans.finalize_tier1("full_scan_id", "tier1_reachability_scan_id") - print(f"Finalization successful: {success}") - -**PARAMETERS:** - -- **full_scan_id (str)** - The ID of the full scan to associate with the tier 1 scan -- **tier1_reachability_scan_id (str)** - The tier 1 reachability scan ID from the facts file - -basics.get_config(org_slug, use_types) -"""""""""""""""""""""""""""""""""""""" -Get Socket Basics configuration for an organization. Socket Basics is a CI/CD security scanning suite that includes SAST scanning, secret detection, container security, and dependency analysis. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - - # Basic usage - returns dictionary - config = socket.basics.get_config("org_slug") - print(f"Python SAST enabled: {config['pythonSastEnabled']}") - print(f"Secret scanning enabled: {config['secretScanningEnabled']}") - - # Using typed response objects - from socketdev.basics import SocketBasicsConfig, SocketBasicsResponse - response = socket.basics.get_config("org_slug", use_types=True) - if response.success and response.config: - print(f"JavaScript SAST: {response.config.javascriptSastEnabled}") - print(f"Trivy scanning: {response.config.trivyImageEnabled}") - -**PARAMETERS:** - -- **org_slug (str)** - The organization name -- **use_types (bool)** - Whether to return typed response objects (default: False) - -**Socket Basics Features:** - -- **Python SAST** - Static analysis for Python code -- **Go SAST** - Static analysis for Go code -- **JavaScript SAST** - Static analysis for JavaScript/TypeScript code -- **Secret Scanning** - Detection of hardcoded secrets and credentials -- **Trivy Image Scanning** - Vulnerability scanning for Docker images -- **Trivy Dockerfile Scanning** - Vulnerability scanning for Dockerfiles -- **Socket SCA** - Supply chain analysis for dependencies -- **Socket Scanning** - General dependency security scanning -- **Additional Parameters** - Custom configuration options - -dependencies.get(limit, offset) -""""""""""""""""""""""""""""""" -Retrieve the dependencies for the organization associated with the API Key - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.dependencies.get(10, 0)) - -**PARAMETERS:** - -- **limit (int)** - The maximum number of dependencies to return -- **offset (int)** - The index to start from for pulling the dependencies - -dependencies.post(files, params) -"""""""""""""""""""""""""""""""" -Retrieve the dependencies for the organization associated with the API Key - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - file_names = [ - "path/to/package.json" - ] - params = { - "repository": "username/repo-name", - "branch": "dependency-branch" - } - print(socket.dependencies.post(file_names, params)) - -**PARAMETERS:** - -- **files (list)** - The file paths of the manifest files to import into the Dependency API. -- **params (dict)** - A dictionary of the `repository` and `branch` options for the API - -repos.get() -""""""""""" -Get a list of information about the tracked repositories - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.repos.get(sort="name", direction="asc", per_page=100, page=1)) - -**PARAMETERS:** - -- **sort** - The key to sort on from the repo properties. Defaults to `created_at` -- **direction** - Can be `desc` or `asc`. Defaults to `desc` -- **per_page** - Integer between 1 to 100. Defaults to `10` -- **page** - Integer page number defaults to `1`. If there are no more results it will be `0` - -repos.post() -"""""""""""" -Create a new Socket Repository - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print( - socket.repos.post( - name="example", - description="Info about Repo", - homepage="http://homepage", - visibility='public', - archived=False, - default_branch='not-main' - ) - ) - -**PARAMETERS:** - -- **name(required)** - The name of the Socket Repository -- **description(optional)** - String description of the repository -- **homepage(optional)** - URL of the homepage of the -- **visibility(optional)** - Can be `public` or `private` and defaults to `private` -- **archived(optional)** - Boolean on if the repository is archived. Defaults to `False` -- **default_branch(optional)** - String name of the default branch for the repository. Defaults to `main` - -repos.repo() -"""""""""""" -Get a list of information about the tracked repositories - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.repos.repo(org_slug="example", repo_name="example-repo")) - -repos.update() -"""""""""""""" -Update an existing Socket Repository - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print( - socket.repos.update( - org_slug="example-org", - repo_name="example", - name="new-name-example", - description="Info about Repo", - homepage="http://homepage", - visibility='public', - archived=False, - default_branch='not-main' - ) - ) - -- **name(optional)** - The name of the Socket Repository -- **description(optional)** - String description of the repository -- **homepage(optional)** - URL of the homepage of the -- **visibility(optional)** - Can be `public` or `private` and defaults to `private` -- **archived(optional)** - Boolean on if the repository is archived. Defaults to `False` -- **default_branch(optional)** - String name of the default branch for the repository. Defaults to `main` - -repos.delete() -"""""""""""""" -Delete a Socket Repository - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.repos.delete(org_slug="example", repo_name="example-repo")) - -**PARAMETERS:** - -- **org_slug** - Name of the Socket Org -- **repo_name** - The name of the Socket Repository to delete - -org.get() -""""""""" -Retrieve the Socket.dev org information - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.org.get()) - -quota.get() -""""""""""" -Retrieve the the current quota available for your API Key - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.quota.get()) - -settings.get() -"""""""""""""" -Retrieve the Socket Organization Settings - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.settings.get()) - -report.supported() -"""""""""""""""""" -Retrieve the supported types of manifest files for creating a report - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.report.supported()) - -Deprecated: report.list() -""""""""""""""""""""""""" -Retrieve the list of all reports for the organization - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.report.list(from_time=1726183485)) - -**PARAMETERS:** - -- **from_time (int)** - The Unix Timestamp in Seconds to limit the reports pulled - -Deprecated: report.delete(report_id) -"""""""""""""""""""""""""""""""""""" -Delete the specified report - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.report.delete("report-id")) - -**PARAMETERS:** - -- **report_id (str)** - The report ID of the report to delete - -Deprecated: report.view(report_id) -"""""""""""""""""""""""""""""""""" -Retrieve the information for a Project Health Report - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.report.view("report_id")) - -**PARAMETERS:** - -- **report_id (str)** - The report ID of the report to view - -Deprecated: report.create(files) -"""""""""""""""""""""""""""""""" -Create a new project health report with the provided files - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - files = [ - "/path/to/manifest/package.json" - ] - print(socket.report.create(files)) - -**PARAMETERS:** - -- **files (list)** - List of file paths of manifest files - -Deprecated: repositories.get() -"""""""""""""""""""""""""""""" -Get a list of information about the tracked repositories - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.repositories.get()) - -Deprecated: sbom.view(report_id) -"""""""""""""""""""""""""""""""" -Retrieve the information for a SBOM Report - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.sbom.view("report_id")) - -Deprecated: npm.issues(package, version) -"""""""""""""""""""""""""""""""""""""""" -Retrieve the Issues associated with a package and version. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.npm.issues("hardhat-gas-report", "1.1.25")) - -**PARAMETERS:** - -- **package (str)** - The name of the NPM package. -- **version (str)** - The version of the NPM Package. - -Deprecated: npm.score(package, version) -""""""""""""""""""""""""""""""""""""""" -Retrieve the Issues associated with a package and version. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.npm.score("hardhat-gas-report", "1.1.25")) - -**PARAMETERS:** - -- **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 - -diffscans.list(org_slug, params=None) -""""""""""""""""""""""""""""""""""""" -List all diff scans for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.diffscans.list("org_slug", {"limit": 10, "offset": 0})) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **params (dict, optional)** – Optional query parameters for filtering - -diffscans.get(org_slug, diff_scan_id) -""""""""""""""""""""""""""""""""""""" -Fetch a specific diff scan by ID. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.diffscans.get("org_slug", "diff_scan_id")) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **diff_scan_id (str)** – The ID of the diff scan to retrieve - -diffscans.create_from_ids(org_slug, params) -""""""""""""""""""""""""""""""""""""""""""" -Create a diff scan from two full scan IDs. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - params = { - "before": "full_scan_id_1", - "after": "full_scan_id_2", - "description": "Compare two scans" - } - print(socket.diffscans.create_from_ids("org_slug", params)) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **params (dict)** – Parameters including before and after scan IDs - -diffscans.create_from_repo(org_slug, repo_slug, files, params=None) -""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Create a diff scan from repository files. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - files = ["/path/to/package.json"] - params = {"branch": "main", "commit": "abc123"} - print(socket.diffscans.create_from_repo("org_slug", "repo_slug", files, params)) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **repo_slug (str)** – The repository name -- **files (list)** – List of file paths to scan -- **params (dict, optional)** – Optional parameters for the scan - -diffscans.gfm(org_slug, diff_scan_id) -""""""""""""""""""""""""""""""""""""" -Get GitHub Flavored Markdown comments for a diff scan. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.diffscans.gfm("org_slug", "diff_scan_id")) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **diff_scan_id (str)** – The ID of the diff scan - -diffscans.delete(org_slug, diff_scan_id) -"""""""""""""""""""""""""""""""""""""""" -Delete a specific diff scan. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.diffscans.delete("org_slug", "diff_scan_id")) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **diff_scan_id (str)** – The ID of the diff scan to delete - -threatfeed.get(org_slug=None, \*\*kwargs) -""""""""""""""""""""""""""""""""""""""""""" -Get threat feed items for an organization or globally. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - - # Get org-specific threat feed - print(socket.threatfeed.get("org_slug", per_page=50, sort="created_at")) - - # Get global threat feed (deprecated) - print(socket.threatfeed.get()) - -**PARAMETERS:** - -- **org_slug (str, optional)** – The organization name (recommended for new implementations) -- **kwargs** – Query parameters like per_page, page_cursor, sort, etc. - -apitokens.create(org_slug, \*\*kwargs) -"""""""""""""""""""""""""""""""""""""" -Create a new API token for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - token_config = { - "name": "My API Token", - "permissions": ["read", "write"], - "expires_at": "2024-12-31T23:59:59Z" - } - print(socket.apitokens.create("org_slug", **token_config)) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **kwargs** – Token configuration parameters - -apitokens.update(org_slug, \*\*kwargs) -"""""""""""""""""""""""""""""""""""""" -Update an existing API token. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - update_params = { - "token_id": "token_123", - "name": "Updated Token Name", - "permissions": ["read"] - } - print(socket.apitokens.update("org_slug", **update_params)) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **kwargs** – Token update parameters - -auditlog.get(org_slug, \*\*kwargs) -"""""""""""""""""""""""""""""""""""" -Get audit log entries for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.auditlog.get("org_slug", limit=100, cursor="abc123")) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **kwargs** – Query parameters like limit, cursor, etc. - -analytics.get_org(filter, \*\*kwargs) -""""""""""""""""""""""""""""""""""""""" -Get organization analytics (deprecated - use Historical module instead). - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - # DEPRECATED: Use socket.historical.list() or socket.historical.trend() instead - print(socket.analytics.get_org("alerts", start_date="2024-01-01")) - -**PARAMETERS:** - -- **filter (str)** – Analytics filter type -- **kwargs** – Additional query parameters - -analytics.get_repo(name, filter, \*\*kwargs) -"""""""""""""""""""""""""""""""""""""""""""""" -Get repository analytics (deprecated - use Historical module instead). - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - # DEPRECATED: Use socket.historical.list() or socket.historical.trend() instead - print(socket.analytics.get_repo("repo_name", "alerts", start_date="2024-01-01")) - -**PARAMETERS:** - -- **name (str)** – Repository name -- **filter (str)** – Analytics filter type -- **kwargs** – Additional query parameters - -alerttypes.get(alert_types=None, language="en-US", \*\*kwargs) -""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Get alert types metadata. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - - # Get metadata for specific alert types - alert_list = ["supply_chain_risk", "license_risk"] - print(socket.alerttypes.get(alert_list, language="en-US")) - - # Get all alert types metadata - print(socket.alerttypes.get()) - -**PARAMETERS:** - -- **alert_types (list, optional)** – List of alert type strings to get metadata for -- **language (str)** – Language for alert metadata (default: en-US) -- **kwargs** – Additional query parameters - -triage.list_alert_triage(org_slug, query_params=None) -""""""""""""""""""""""""""""""""""""""""""""""""""""" -Get list of triaged alerts for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - query_params = {"status": "triaged", "limit": 50} - print(socket.triage.list_alert_triage("org_slug", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** – The organization name -- **query_params (dict, optional)** – Optional query parameters for filtering - -openapi.get() -""""""""""""" -Retrieve the OpenAPI specification for the Socket API. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.openapi.get()) - -**PARAMETERS:** - -None required. - -webhooks.list(org_slug, \*\*query_params) -""""""""""""""""""""""""""""""""""""""""""" -List all webhooks for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.webhooks.list("org_slug")) - - # With query parameters - print(socket.webhooks.list("org_slug", limit=10, offset=0)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params** - Optional query parameters for filtering - -webhooks.create(org_slug, \*\*kwargs) -"""""""""""""""""""""""""""""""""""""" -Create a new webhook for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - webhook_config = { - "url": "https://example.com/webhook", - "events": ["alert.created", "scan.completed"] - } - print(socket.webhooks.create("org_slug", **webhook_config)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **kwargs** - Webhook configuration parameters - -webhooks.get(org_slug, webhook_id) -"""""""""""""""""""""""""""""""""" -Get details for a specific webhook. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.webhooks.get("org_slug", "webhook_id")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **webhook_id (str)** - The webhook ID - -webhooks.update(org_slug, webhook_id, \*\*kwargs) -""""""""""""""""""""""""""""""""""""""""""""""""""" -Update an existing webhook. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - updates = { - "url": "https://example.com/new-webhook", - "events": ["alert.created"] - } - print(socket.webhooks.update("org_slug", "webhook_id", **updates)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **webhook_id (str)** - The webhook ID -- **kwargs** - Webhook configuration parameters to update - -webhooks.delete(org_slug, webhook_id) -""""""""""""""""""""""""""""""""""""" -Delete a webhook. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.webhooks.delete("org_slug", "webhook_id")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **webhook_id (str)** - The webhook ID - -telemetry.get_config(org_slug) -"""""""""""""""""""""""""""""" -Get telemetry configuration for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.telemetry.get_config("org_slug")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug - -telemetry.update_config(org_slug, \*\*kwargs) -"""""""""""""""""""""""""""""""""""""""""""""" -Update telemetry configuration for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - config = { - "enabled": True, - "sampling_rate": 0.5 - } - print(socket.telemetry.update_config("org_slug", **config)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **kwargs** - Configuration parameters to update - -alerts.get(org_slug, \*\*query_params) -""""""""""""""""""""""""""""""""""""""" -Get alerts for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.alerts.get("org_slug")) - - # With query parameters - print(socket.alerts.get("org_slug", severity="high", limit=50)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params** - Optional query parameters for filtering - -fixes.get(org_slug, \*\*query_params) -"""""""""""""""""""""""""""""""""""""" -Get available fixes for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.fixes.get("org_slug")) - - # With query parameters - print(socket.fixes.get("org_slug", ecosystem="npm")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params** - Optional query parameters for filtering - -supportedfiles.get(org_slug) -"""""""""""""""""""""""""""" -Get list of supported manifest file types for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.supportedfiles.get("org_slug")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug - -alertfullscansearch.search(org_slug, \*\*query_params) -""""""""""""""""""""""""""""""""""""""""""""""""""""""" -Search alerts across full scans. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - search_params = { - "query": "CVE-2024-1234", - "limit": 20 - } - print(socket.alertfullscansearch.search("org_slug", **search_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params** - Optional query parameters for filtering - -historical.dependencies_trend(org_slug, query_params) -"""""""""""""""""""""""""""""""""""""""""""""""""""""" -Get historical dependency trends data for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - query_params = { - "from": "2024-01-01", - "to": "2024-12-31" - } - print(socket.historical.dependencies_trend("org_slug", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params (dict, optional)** - Optional query parameters for date filtering - -historical.snapshots.create(org_slug) -""""""""""""""""""""""""""""""""""""" -Create a new snapshot for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.historical.snapshots.create("org_slug")) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug - -historical.snapshots.list(org_slug, query_params) -""""""""""""""""""""""""""""""""""""""""""""""""" -List historical snapshots for an organization. - -**Usage:** - -.. code-block:: python - - from socketdev import socketdev - socket = socketdev(token="REPLACE_ME") - print(socket.historical.snapshots.list("org_slug")) - - # With query parameters - query_params = {"limit": 10, "offset": 0} - print(socket.historical.snapshots.list("org_slug", query_params)) - -**PARAMETERS:** - -- **org_slug (str)** - The organization slug -- **query_params (dict, optional)** - Optional query parameters for filtering + +socketdev +######### + +Purpose +------- + +The Socket.dev Python SDK provides a wrapper around the Socket.dev REST API to simplify making calls to the API from Python. + +Socket API v0 - https://docs.socket.dev/reference/introduction-to-socket-api + +Initializing the module +----------------------- + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME", timeout=30) + +**PARAMETERS:** + +- **token (str)** - The Socket API Key for your Organization +- **timeout (int)** - The number of seconds to wait before failing the connection +- **allow_unverified (bool)** - Whether to skip SSL certificate verification (default: False). Set to True for testing with self-signed certificates. +- **user_agent (str, optional)** - Custom User-Agent string to use in API requests. If not provided, defaults to "SocketSDKPython/{version}" + +Supported Functions +------------------- + + +purl.post(license, components, org_slug=None) +""""""""""""""""""""""""""""""""""""""""""""" +Retrieve package information for one or more PURLs. Pass ``org_slug`` to use the +current org-scoped endpoint. Omitting ``org_slug`` keeps the legacy deprecated +endpoint for backwards compatibility. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + org_slug = "your-org-slug" + license = "true" + components = [ + { + "purl": "pkg:pypi/pyonepassword@5.0.0" + }, + { + "purl": "pkg:pypi/socketsecurity" + } + ] + print(socket.purl.post(license, components, org_slug=org_slug)) + +**PARAMETERS:** + +- **license (str)** - The license parameter if enabled will show alerts and license information. If disabled will only show the basic package metadata and scores. Default is true +- **components (array{dict})** - The components list of packages urls +- **org_slug (str, optional)** - Organization slug for the supported org-scoped PURL endpoint. If omitted, the SDK uses the deprecated legacy endpoint for backwards compatibility. + +export.cdx_bom(org_slug, id, query_params) +"""""""""""""""""""""""""""""""""""""""""" +Export a Socket SBOM as a CycloneDX SBOM + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + from socketdev.export import ExportQueryParams + + socket = socketdev(token="REPLACE_ME") + query_params = ExportQueryParams( + author="john_doe", + project_name="my-project" + ) + print(socket.export.cdx_bom("org_slug", "sbom_id", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **id (str)** - The ID of either a full scan or an SBOM report +- **query_params (ExportQueryParams)** - Optional query parameters for filtering: + - **author (str)** - Filter by author + - **project_group (str)** - Filter by project group + - **project_name (str)** - Filter by project name + - **project_version (str)** - Filter by project version + - **project_id (str)** - Filter by project ID + +export.spdx_bom(org_slug, id, query_params) +""""""""""""""""""""""""""""""""""""""""""" +Export a Socket SBOM as an SPDX SBOM + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + from socketdev.export import ExportQueryParams + + socket = socketdev(token="REPLACE_ME") + query_params = ExportQueryParams( + project_name="my-project", + project_version="1.0.0" + ) + print(socket.export.spdx_bom("org_slug", "sbom_id", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **id (str)** - The ID of either a full scan or an SBOM report +- **query_params (ExportQueryParams)** - Optional query parameters for filtering: + - **author (str)** - Filter by author + - **project_group (str)** - Filter by project group + - **project_name (str)** - Filter by project name + - **project_version (str)** - Filter by project version + - **project_id (str)** - Filter by project ID + +export.openvex_bom(org_slug, id, query_params) +"""""""""""""""""""""""""""""""""""""""""""""" +Export a Socket SBOM as an OpenVEX SBOM + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + from socketdev.export import ExportQueryParams + + socket = socketdev(token="REPLACE_ME") + query_params = ExportQueryParams( + project_name="my-project", + project_version="1.0.0" + ) + print(socket.export.openvex_bom("org_slug", "sbom_id", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **id (str)** - The ID of either a full scan or an SBOM report +- **query_params (ExportQueryParams)** - Optional query parameters for filtering: + - **author (str)** - Filter by author + - **project_group (str)** - Filter by project group + - **project_name (str)** - Filter by project name + - **project_version (str)** - Filter by project version + - **project_id (str)** - Filter by project ID + +fullscans.get(org_slug, params) +""""""""""""""""""""""""""""""" +Retrieve the Fullscans information for an Organization with query parameters + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + + # Query parameters for filtering full scans + params = { + "repo": "my-repo", + "branch": "main", + "limit": 10, + "offset": 0 + } + print(socket.fullscans.get("org_slug", params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **params (dict)** - Query parameters for filtering results (required) + +fullscans.post(files, params) +""""""""""""""""""""""""""""" +Create a full scan from a set of package manifest files. Returns a full scan including all SBOM artifacts. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + from socketdev.fullscans import FullScanParams + + socket = socketdev(token="REPLACE_ME") + files = [ + "/path/to/manifest/package.json" + ] + params = FullScanParams( + org_slug="org_name", + repo="TestRepo", + workspace="my-workspace", + branch="main", + commit_message="Test Commit Message", + commit_hash="abc123def456", + pull_request=123, + committers=["committer1", "committer2"], + make_default_branch=False, + set_as_pending_head=False + ) + + print(socket.fullscans.post(files, params)) + +**PARAMETERS:** + +- **files (list)** - List of file paths of manifest files +- **params (FullScanParams)** - FullScanParams object containing scan configuration + ++------------------------+------------+-------------------------------------------------------------------------------+ +| Parameter | Required | Description | ++========================+============+===============================================================================+ +| org_slug | True | The string name in a git approved name for organization. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| repo | True | The string name in a git approved name for repositories. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| branch | False | The string name in a git approved name for branches. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| committers | False | List of committer names (List[str]). | ++------------------------+------------+-------------------------------------------------------------------------------+ +| pull_request | False | The integer for the PR or MR number. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| commit_message | False | The string for a commit message if there is one. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| make_default_branch | False | Boolean to signal that this is the default branch. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| commit_hash | False | Optional git commit hash | ++------------------------+------------+-------------------------------------------------------------------------------+ +| set_as_pending_head | False | Boolean to set as pending head | ++------------------------+------------+-------------------------------------------------------------------------------+ +| tmp | False | Boolean temporary flag | ++------------------------+------------+-------------------------------------------------------------------------------+ +| workspace | False | The workspace of the repository to associate the full-scan with. | ++------------------------+------------+-------------------------------------------------------------------------------+ +| integration_type | False | IntegrationType enum value (e.g., "api", "github") | ++------------------------+------------+-------------------------------------------------------------------------------+ +| integration_org_slug | False | Organization slug for integration | ++------------------------+------------+-------------------------------------------------------------------------------+ +| scan_type | False | ScanType enum value: "socket", "socket_tier1", or "socket_basics" | ++------------------------+------------+-------------------------------------------------------------------------------+ + +fullscans.delete(org_slug, full_scan_id) +"""""""""""""""""""""""""""""""""""""""" +Delete an existing full scan. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fullscans.delete("org_slug", "full_scan_id")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **full_scan_id (str)** - The ID of the full scan + +fullscans.stream_diff(org_slug, before, after, use_types=True, include_license_details="true", \*\*kwargs) +"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Stream a diff between two full scans. Returns a scan diff. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fullscans.stream_diff("org_slug", "before_scan_id", "after_scan_id")) + + # With additional parameters + print(socket.fullscans.stream_diff( + "org_slug", + "before_scan_id", + "after_scan_id", + use_types=False, + include_license_details="false" + )) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **before (str)** - The base full scan ID +- **after (str)** - The comparison full scan ID +- **use_types (bool)** - Whether to return typed response objects (default: True) +- **include_license_details (str)** - Include license details ("true"/"false"). Can greatly increase response size. Defaults to "true". +- **kwargs** - Additional query parameters + +fullscans.stream(org_slug, full_scan_id, use_types=False) +""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Stream all SBOM artifacts for a full scan. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fullscans.stream("org_slug", "full_scan_id")) + + # With typed response + print(socket.fullscans.stream("org_slug", "full_scan_id", use_types=True)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **full_scan_id (str)** - The ID of the full scan +- **use_types (bool)** - Whether to return typed response objects (default: False) + +fullscans.metadata(org_slug, full_scan_id, use_types=False) +""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Get metadata for a single full scan + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fullscans.metadata("org_slug", "full_scan_id")) + + # With typed response + print(socket.fullscans.metadata("org_slug", "full_scan_id", use_types=True)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **full_scan_id (str)** - The ID of the full scan +- **use_types (bool)** - Whether to return typed response objects (default: False) + +fullscans.gfm(org_slug, before, after) +"""""""""""""""""""""""""""""""""""""" +Get GitHub Flavored Markdown diff between two full scans. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fullscans.gfm("org_slug", "before_scan_id", "after_scan_id")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **before (str)** - The base full scan ID +- **after (str)** - The comparison full scan ID + +fullscans.finalize_tier1(full_scan_id, tier1_reachability_scan_id) +"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Finalize a tier 1 reachability scan by associating it with a full scan. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + success = socket.fullscans.finalize_tier1("full_scan_id", "tier1_reachability_scan_id") + print(f"Finalization successful: {success}") + +**PARAMETERS:** + +- **full_scan_id (str)** - The ID of the full scan to associate with the tier 1 scan +- **tier1_reachability_scan_id (str)** - The tier 1 reachability scan ID from the facts file + +basics.get_config(org_slug, use_types) +"""""""""""""""""""""""""""""""""""""" +Get Socket Basics configuration for an organization. Socket Basics is a CI/CD security scanning suite that includes SAST scanning, secret detection, container security, and dependency analysis. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + + # Basic usage - returns dictionary + config = socket.basics.get_config("org_slug") + print(f"Python SAST enabled: {config['pythonSastEnabled']}") + print(f"Secret scanning enabled: {config['secretScanningEnabled']}") + + # Using typed response objects + from socketdev.basics import SocketBasicsConfig, SocketBasicsResponse + response = socket.basics.get_config("org_slug", use_types=True) + if response.success and response.config: + print(f"JavaScript SAST: {response.config.javascriptSastEnabled}") + print(f"Trivy scanning: {response.config.trivyImageEnabled}") + +**PARAMETERS:** + +- **org_slug (str)** - The organization name +- **use_types (bool)** - Whether to return typed response objects (default: False) + +**Socket Basics Features:** + +- **Python SAST** - Static analysis for Python code +- **Go SAST** - Static analysis for Go code +- **JavaScript SAST** - Static analysis for JavaScript/TypeScript code +- **Secret Scanning** - Detection of hardcoded secrets and credentials +- **Trivy Image Scanning** - Vulnerability scanning for Docker images +- **Trivy Dockerfile Scanning** - Vulnerability scanning for Dockerfiles +- **Socket SCA** - Supply chain analysis for dependencies +- **Socket Scanning** - General dependency security scanning +- **Additional Parameters** - Custom configuration options + +dependencies.get(limit, offset) +""""""""""""""""""""""""""""""" +Retrieve the dependencies for the organization associated with the API Key + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.dependencies.get(10, 0)) + +**PARAMETERS:** + +- **limit (int)** - The maximum number of dependencies to return +- **offset (int)** - The index to start from for pulling the dependencies + +dependencies.post(files, params) +"""""""""""""""""""""""""""""""" +Retrieve the dependencies for the organization associated with the API Key + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + file_names = [ + "path/to/package.json" + ] + params = { + "repository": "username/repo-name", + "branch": "dependency-branch" + } + print(socket.dependencies.post(file_names, params)) + +**PARAMETERS:** + +- **files (list)** - The file paths of the manifest files to import into the Dependency API. +- **params (dict)** - A dictionary of the `repository` and `branch` options for the API + +repos.get() +""""""""""" +Get a list of information about the tracked repositories + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.repos.get(sort="name", direction="asc", per_page=100, page=1)) + +**PARAMETERS:** + +- **sort** - The key to sort on from the repo properties. Defaults to `created_at` +- **direction** - Can be `desc` or `asc`. Defaults to `desc` +- **per_page** - Integer between 1 to 100. Defaults to `10` +- **page** - Integer page number defaults to `1`. If there are no more results it will be `0` + +repos.post() +"""""""""""" +Create a new Socket Repository + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print( + socket.repos.post( + name="example", + description="Info about Repo", + homepage="http://homepage", + visibility='public', + archived=False, + default_branch='not-main' + ) + ) + +**PARAMETERS:** + +- **name(required)** - The name of the Socket Repository +- **description(optional)** - String description of the repository +- **homepage(optional)** - URL of the homepage of the +- **visibility(optional)** - Can be `public` or `private` and defaults to `private` +- **archived(optional)** - Boolean on if the repository is archived. Defaults to `False` +- **default_branch(optional)** - String name of the default branch for the repository. Defaults to `main` + +repos.repo() +"""""""""""" +Get a list of information about the tracked repositories + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.repos.repo(org_slug="example", repo_name="example-repo")) + +repos.update() +"""""""""""""" +Update an existing Socket Repository + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print( + socket.repos.update( + org_slug="example-org", + repo_name="example", + name="new-name-example", + description="Info about Repo", + homepage="http://homepage", + visibility='public', + archived=False, + default_branch='not-main' + ) + ) + +- **name(optional)** - The name of the Socket Repository +- **description(optional)** - String description of the repository +- **homepage(optional)** - URL of the homepage of the +- **visibility(optional)** - Can be `public` or `private` and defaults to `private` +- **archived(optional)** - Boolean on if the repository is archived. Defaults to `False` +- **default_branch(optional)** - String name of the default branch for the repository. Defaults to `main` + +repos.delete() +"""""""""""""" +Delete a Socket Repository + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.repos.delete(org_slug="example", repo_name="example-repo")) + +**PARAMETERS:** + +- **org_slug** - Name of the Socket Org +- **repo_name** - The name of the Socket Repository to delete + +org.get() +""""""""" +Retrieve the Socket.dev org information + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.org.get()) + +quota.get() +""""""""""" +Retrieve the the current quota available for your API Key + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.quota.get()) + +settings.get() +"""""""""""""" +Retrieve the Socket Organization Settings + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.settings.get()) + +report.supported() +"""""""""""""""""" +Retrieve the supported types of manifest files for creating a report + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.report.supported()) + +Deprecated: report.list() +""""""""""""""""""""""""" +Retrieve the list of all reports for the organization + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.report.list(from_time=1726183485)) + +**PARAMETERS:** + +- **from_time (int)** - The Unix Timestamp in Seconds to limit the reports pulled + +Deprecated: report.delete(report_id) +"""""""""""""""""""""""""""""""""""" +Delete the specified report + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.report.delete("report-id")) + +**PARAMETERS:** + +- **report_id (str)** - The report ID of the report to delete + +Deprecated: report.view(report_id) +"""""""""""""""""""""""""""""""""" +Retrieve the information for a Project Health Report + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.report.view("report_id")) + +**PARAMETERS:** + +- **report_id (str)** - The report ID of the report to view + +Deprecated: report.create(files) +"""""""""""""""""""""""""""""""" +Create a new project health report with the provided files + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + files = [ + "/path/to/manifest/package.json" + ] + print(socket.report.create(files)) + +**PARAMETERS:** + +- **files (list)** - List of file paths of manifest files + +Deprecated: repositories.get() +"""""""""""""""""""""""""""""" +Get a list of information about the tracked repositories + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.repositories.get()) + +Deprecated: sbom.view(report_id) +"""""""""""""""""""""""""""""""" +Retrieve the information for a SBOM Report + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.sbom.view("report_id")) + +Deprecated: npm.issues(package, version) +"""""""""""""""""""""""""""""""""""""""" +Retrieve the Issues associated with a package and version. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.npm.issues("hardhat-gas-report", "1.1.25")) + +**PARAMETERS:** + +- **package (str)** - The name of the NPM package. +- **version (str)** - The version of the NPM Package. + +Deprecated: npm.score(package, version) +""""""""""""""""""""""""""""""""""""""" +Retrieve the Issues associated with a package and version. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.npm.score("hardhat-gas-report", "1.1.25")) + +**PARAMETERS:** + +- **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 + +diffscans.list(org_slug, params=None) +""""""""""""""""""""""""""""""""""""" +List all diff scans for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.diffscans.list("org_slug", {"limit": 10, "offset": 0})) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **params (dict, optional)** – Optional query parameters for filtering + +diffscans.get(org_slug, diff_scan_id) +""""""""""""""""""""""""""""""""""""" +Fetch a specific diff scan by ID. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.diffscans.get("org_slug", "diff_scan_id")) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **diff_scan_id (str)** – The ID of the diff scan to retrieve + +diffscans.create_from_ids(org_slug, params) +""""""""""""""""""""""""""""""""""""""""""" +Create a diff scan from two full scan IDs. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + params = { + "before": "full_scan_id_1", + "after": "full_scan_id_2", + "description": "Compare two scans" + } + print(socket.diffscans.create_from_ids("org_slug", params)) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **params (dict)** – Parameters including before and after scan IDs + +diffscans.create_from_repo(org_slug, repo_slug, files, params=None) +""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Create a diff scan from repository files. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + files = ["/path/to/package.json"] + params = {"branch": "main", "commit": "abc123"} + print(socket.diffscans.create_from_repo("org_slug", "repo_slug", files, params)) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **repo_slug (str)** – The repository name +- **files (list)** – List of file paths to scan +- **params (dict, optional)** – Optional parameters for the scan + +diffscans.gfm(org_slug, diff_scan_id) +""""""""""""""""""""""""""""""""""""" +Get GitHub Flavored Markdown comments for a diff scan. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.diffscans.gfm("org_slug", "diff_scan_id")) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **diff_scan_id (str)** – The ID of the diff scan + +diffscans.delete(org_slug, diff_scan_id) +"""""""""""""""""""""""""""""""""""""""" +Delete a specific diff scan. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.diffscans.delete("org_slug", "diff_scan_id")) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **diff_scan_id (str)** – The ID of the diff scan to delete + +threatfeed.get(org_slug=None, \*\*kwargs) +""""""""""""""""""""""""""""""""""""""""""" +Get threat feed items for an organization or globally. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + + # Get org-specific threat feed + print(socket.threatfeed.get("org_slug", per_page=50, sort="created_at")) + + # Get global threat feed (deprecated) + print(socket.threatfeed.get()) + +**PARAMETERS:** + +- **org_slug (str, optional)** – The organization name (recommended for new implementations) +- **kwargs** – Query parameters like per_page, page_cursor, sort, etc. + +apitokens.create(org_slug, \*\*kwargs) +"""""""""""""""""""""""""""""""""""""" +Create a new API token for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + token_config = { + "name": "My API Token", + "permissions": ["read", "write"], + "expires_at": "2024-12-31T23:59:59Z" + } + print(socket.apitokens.create("org_slug", **token_config)) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **kwargs** – Token configuration parameters + +apitokens.update(org_slug, \*\*kwargs) +"""""""""""""""""""""""""""""""""""""" +Update an existing API token. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + update_params = { + "token_id": "token_123", + "name": "Updated Token Name", + "permissions": ["read"] + } + print(socket.apitokens.update("org_slug", **update_params)) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **kwargs** – Token update parameters + +auditlog.get(org_slug, \*\*kwargs) +"""""""""""""""""""""""""""""""""""" +Get audit log entries for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.auditlog.get("org_slug", limit=100, cursor="abc123")) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **kwargs** – Query parameters like limit, cursor, etc. + +analytics.get_org(filter, \*\*kwargs) +""""""""""""""""""""""""""""""""""""""" +Get organization analytics (deprecated - use Historical module instead). + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + # DEPRECATED: Use socket.historical.list() or socket.historical.trend() instead + print(socket.analytics.get_org("alerts", start_date="2024-01-01")) + +**PARAMETERS:** + +- **filter (str)** – Analytics filter type +- **kwargs** – Additional query parameters + +analytics.get_repo(name, filter, \*\*kwargs) +"""""""""""""""""""""""""""""""""""""""""""""" +Get repository analytics (deprecated - use Historical module instead). + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + # DEPRECATED: Use socket.historical.list() or socket.historical.trend() instead + print(socket.analytics.get_repo("repo_name", "alerts", start_date="2024-01-01")) + +**PARAMETERS:** + +- **name (str)** – Repository name +- **filter (str)** – Analytics filter type +- **kwargs** – Additional query parameters + +alerttypes.get(alert_types=None, language="en-US", \*\*kwargs) +""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Get alert types metadata. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + + # Get metadata for specific alert types + alert_list = ["supply_chain_risk", "license_risk"] + print(socket.alerttypes.get(alert_list, language="en-US")) + + # Get all alert types metadata + print(socket.alerttypes.get()) + +**PARAMETERS:** + +- **alert_types (list, optional)** – List of alert type strings to get metadata for +- **language (str)** – Language for alert metadata (default: en-US) +- **kwargs** – Additional query parameters + +triage.list_alert_triage(org_slug, query_params=None) +""""""""""""""""""""""""""""""""""""""""""""""""""""" +Get list of triaged alerts for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + query_params = {"status": "triaged", "limit": 50} + print(socket.triage.list_alert_triage("org_slug", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** – The organization name +- **query_params (dict, optional)** – Optional query parameters for filtering + +openapi.get() +""""""""""""" +Retrieve the OpenAPI specification for the Socket API. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.openapi.get()) + +**PARAMETERS:** + +None required. + +webhooks.list(org_slug, \*\*query_params) +""""""""""""""""""""""""""""""""""""""""""" +List all webhooks for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.webhooks.list("org_slug")) + + # With query parameters + print(socket.webhooks.list("org_slug", limit=10, offset=0)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params** - Optional query parameters for filtering + +webhooks.create(org_slug, \*\*kwargs) +"""""""""""""""""""""""""""""""""""""" +Create a new webhook for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + webhook_config = { + "url": "https://example.com/webhook", + "events": ["alert.created", "scan.completed"] + } + print(socket.webhooks.create("org_slug", **webhook_config)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **kwargs** - Webhook configuration parameters + +webhooks.get(org_slug, webhook_id) +"""""""""""""""""""""""""""""""""" +Get details for a specific webhook. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.webhooks.get("org_slug", "webhook_id")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **webhook_id (str)** - The webhook ID + +webhooks.update(org_slug, webhook_id, \*\*kwargs) +""""""""""""""""""""""""""""""""""""""""""""""""""" +Update an existing webhook. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + updates = { + "url": "https://example.com/new-webhook", + "events": ["alert.created"] + } + print(socket.webhooks.update("org_slug", "webhook_id", **updates)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **webhook_id (str)** - The webhook ID +- **kwargs** - Webhook configuration parameters to update + +webhooks.delete(org_slug, webhook_id) +""""""""""""""""""""""""""""""""""""" +Delete a webhook. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.webhooks.delete("org_slug", "webhook_id")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **webhook_id (str)** - The webhook ID + +telemetry.get_config(org_slug) +"""""""""""""""""""""""""""""" +Get telemetry configuration for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.telemetry.get_config("org_slug")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug + +telemetry.update_config(org_slug, \*\*kwargs) +"""""""""""""""""""""""""""""""""""""""""""""" +Update telemetry configuration for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + config = { + "enabled": True, + "sampling_rate": 0.5 + } + print(socket.telemetry.update_config("org_slug", **config)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **kwargs** - Configuration parameters to update + +alerts.get(org_slug, \*\*query_params) +""""""""""""""""""""""""""""""""""""""" +Get alerts for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.alerts.get("org_slug")) + + # With query parameters + print(socket.alerts.get("org_slug", severity="high", limit=50)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params** - Optional query parameters for filtering + +fixes.get(org_slug, \*\*query_params) +"""""""""""""""""""""""""""""""""""""" +Get available fixes for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.fixes.get("org_slug")) + + # With query parameters + print(socket.fixes.get("org_slug", ecosystem="npm")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params** - Optional query parameters for filtering + +supportedfiles.get(org_slug) +"""""""""""""""""""""""""""" +Get list of supported manifest file types for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.supportedfiles.get("org_slug")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug + +alertfullscansearch.search(org_slug, \*\*query_params) +""""""""""""""""""""""""""""""""""""""""""""""""""""""" +Search alerts across full scans. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + search_params = { + "query": "CVE-2024-1234", + "limit": 20 + } + print(socket.alertfullscansearch.search("org_slug", **search_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params** - Optional query parameters for filtering + +historical.dependencies_trend(org_slug, query_params) +"""""""""""""""""""""""""""""""""""""""""""""""""""""" +Get historical dependency trends data for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + query_params = { + "from": "2024-01-01", + "to": "2024-12-31" + } + print(socket.historical.dependencies_trend("org_slug", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params (dict, optional)** - Optional query parameters for date filtering + +historical.snapshots.create(org_slug) +""""""""""""""""""""""""""""""""""""" +Create a new snapshot for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.historical.snapshots.create("org_slug")) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug + +historical.snapshots.list(org_slug, query_params) +""""""""""""""""""""""""""""""""""""""""""""""""" +List historical snapshots for an organization. + +**Usage:** + +.. code-block:: python + + from socketdev import socketdev + socket = socketdev(token="REPLACE_ME") + print(socket.historical.snapshots.list("org_slug")) + + # With query parameters + query_params = {"limit": 10, "offset": 0} + print(socket.historical.snapshots.list("org_slug", query_params)) + +**PARAMETERS:** + +- **org_slug (str)** - The organization slug +- **query_params (dict, optional)** - Optional query parameters for filtering diff --git a/socketdev/__init__.py b/socketdev/__init__.py index 2a9dde2..c257d07 100644 --- a/socketdev/__init__.py +++ b/socketdev/__init__.py @@ -1,115 +1,115 @@ -import os -from socketdev.core.api import API -from socketdev.dependencies import Dependencies -from socketdev.diffscans import DiffScans -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 -from socketdev.purl import Purl -from socketdev.quota import Quota -from socketdev.report import Report -from socketdev.repos import Repos -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__ -from socketdev.labels import Labels -from socketdev.licensemetadata import LicenseMetadata -from socketdev.threatfeed import ThreatFeed -from socketdev.apitokens import ApiTokens -from socketdev.auditlog import AuditLog -from socketdev.analytics import Analytics -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__ -__all__ = ["socketdev", "Utils", "IntegrationType", "INTEGRATION_TYPES"] - - -global encoded_key -encoded_key: str - -api_url = "https://api.socket.dev/v0" -request_timeout = 1200 - - -# TODO: Add debug flag to constructor to enable verbose error logging for API response parsing. - - -class socketdev: - def __init__(self, token: Optional[str] = None, timeout: int = 1200, allow_unverified: bool = False, user_agent: Optional[str] = None): - # 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) - self.api.set_timeout(timeout) - self.api.set_allow_unverified(allow_unverified) - if user_agent is not None: - self.api.set_user_agent(user_agent) - - 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.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() - self.labels = Labels(self.api) - self.licensemetadata = LicenseMetadata(self.api) - self.diffscans = DiffScans(self.api) - self.threatfeed = ThreatFeed(self.api) - self.apitokens = ApiTokens(self.api) - self.auditlog = AuditLog(self.api) - self.analytics = Analytics(self.api) - 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): - # Kept for backwards compatibility - pass +import os +from socketdev.core.api import API +from socketdev.dependencies import Dependencies +from socketdev.diffscans import DiffScans +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 +from socketdev.purl import Purl +from socketdev.quota import Quota +from socketdev.report import Report +from socketdev.repos import Repos +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__ +from socketdev.labels import Labels +from socketdev.licensemetadata import LicenseMetadata +from socketdev.threatfeed import ThreatFeed +from socketdev.apitokens import ApiTokens +from socketdev.auditlog import AuditLog +from socketdev.analytics import Analytics +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__ +__all__ = ["socketdev", "Utils", "IntegrationType", "INTEGRATION_TYPES"] + + +global encoded_key +encoded_key: str + +api_url = "https://api.socket.dev/v0" +request_timeout = 1200 + + +# TODO: Add debug flag to constructor to enable verbose error logging for API response parsing. + + +class socketdev: + def __init__(self, token: Optional[str] = None, timeout: int = 1200, allow_unverified: bool = False, user_agent: Optional[str] = None): + # 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) + self.api.set_timeout(timeout) + self.api.set_allow_unverified(allow_unverified) + if user_agent is not None: + self.api.set_user_agent(user_agent) + + 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.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() + self.labels = Labels(self.api) + self.licensemetadata = LicenseMetadata(self.api) + self.diffscans = DiffScans(self.api) + self.threatfeed = ThreatFeed(self.api) + self.apitokens = ApiTokens(self.api) + self.auditlog = AuditLog(self.api) + self.analytics = Analytics(self.api) + 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): + # Kept for backwards compatibility + pass diff --git a/socketdev/core/classes.py b/socketdev/core/classes.py index 3a21401..0d3a37e 100644 --- a/socketdev/core/classes.py +++ b/socketdev/core/classes.py @@ -1,148 +1,148 @@ -import json - - -class Score: - supplyChain: float - quality: float - maintenance: float - license: float - overall: float - vulnerability: float - - def __init__(self, **kwargs): - if kwargs: - for key, value in kwargs.items(): - setattr(self, key, value) - for score_name in self.__dict__: - score = getattr(self, score_name) - if score <= 1: - score = score * 100 - setattr(self, score_name, score) - - def __str__(self): - return json.dumps(self.__dict__) - - -class Package: - type: str - name: str - version: str - release: str - id: str - direct: bool - manifestFiles: list - author: list - size: int - score: dict - scores: Score - alerts: list - error_alerts: list - alert_counts: dict - topLevelAncestors: list - url: str - transitives: int - license: str - license_text: str - purl: str - - def __init__(self, **kwargs): - if kwargs: - for key, value in kwargs.items(): - setattr(self, key, value) - if not hasattr(self, "direct"): - self.direct = False - else: - if str(self.direct).lower() == "true": - self.direct = True - self.url = f"https://socket.dev/{self.type}/package/{self.name}/overview/{self.version}" - if hasattr(self, 'score'): - self.scores = Score(**self.score) - if not hasattr(self, "alerts"): - self.alerts = [] - if not hasattr(self, "topLevelAncestors"): - self.topLevelAncestors = [] - if not hasattr(self, "manifestFiles"): - self.manifestFiles = [] - if not hasattr(self, "transitives"): - self.transitives = 0 - if not hasattr(self, "author"): - self.author = [] - if not hasattr(self, "size"): - self.size = 0 - self.alert_counts = { - "critical": 0, - "high": 0, - "middle": 0, - "low": 0 - } - self.error_alerts = [] - if not hasattr(self, "license"): - self.license = "NoLicenseFound" - if not hasattr(self, "license_text"): - self.license_text = "" - self.url = f"https://socket.dev/{self.type}/package/{self.name}/overview/{self.version}" - self.purl = f"{self.type}/{self.name}@{self.version}" - - -class Dependency: - branch: str - id: int - name: str - type: str - version: str - namespace: str - repository: str - - def __init__(self, **kwargs): - if kwargs: - for key, value in kwargs.items(): - setattr(self, key, value) - - def __str__(self): - return json.dumps(self.__dict__) - - -class Org: - id: int - image: str - name: str - plan: str - - def __init__(self, **kwargs): - if kwargs: - for key, value in kwargs.items(): - setattr(self, key, value) - - def __str__(self): - return json.dumps(self.__dict__) - - -class Response: - text: str - error: bool - status_code: int - - def __init__(self, text: str, error: bool, status_code: int): - self.text = text - self.error = error - self.status_code = status_code - - def __str__(self): - return json.dumps(self.__dict__) - - def json(self): - return self.__dict__ - - -class DependGetData: - url: str - headers: dict - payload: str - - def __init__(self, **kwargs): - if kwargs: - for key, value in kwargs.items(): - setattr(self, key, value) - - def __str__(self): - return json.dumps(self.__dict__) +import json + + +class Score: + supplyChain: float + quality: float + maintenance: float + license: float + overall: float + vulnerability: float + + def __init__(self, **kwargs): + if kwargs: + for key, value in kwargs.items(): + setattr(self, key, value) + for score_name in self.__dict__: + score = getattr(self, score_name) + if score <= 1: + score = score * 100 + setattr(self, score_name, score) + + def __str__(self): + return json.dumps(self.__dict__) + + +class Package: + type: str + name: str + version: str + release: str + id: str + direct: bool + manifestFiles: list + author: list + size: int + score: dict + scores: Score + alerts: list + error_alerts: list + alert_counts: dict + topLevelAncestors: list + url: str + transitives: int + license: str + license_text: str + purl: str + + def __init__(self, **kwargs): + if kwargs: + for key, value in kwargs.items(): + setattr(self, key, value) + if not hasattr(self, "direct"): + self.direct = False + else: + if str(self.direct).lower() == "true": + self.direct = True + self.url = f"https://socket.dev/{self.type}/package/{self.name}/overview/{self.version}" + if hasattr(self, 'score'): + self.scores = Score(**self.score) + if not hasattr(self, "alerts"): + self.alerts = [] + if not hasattr(self, "topLevelAncestors"): + self.topLevelAncestors = [] + if not hasattr(self, "manifestFiles"): + self.manifestFiles = [] + if not hasattr(self, "transitives"): + self.transitives = 0 + if not hasattr(self, "author"): + self.author = [] + if not hasattr(self, "size"): + self.size = 0 + self.alert_counts = { + "critical": 0, + "high": 0, + "middle": 0, + "low": 0 + } + self.error_alerts = [] + if not hasattr(self, "license"): + self.license = "NoLicenseFound" + if not hasattr(self, "license_text"): + self.license_text = "" + self.url = f"https://socket.dev/{self.type}/package/{self.name}/overview/{self.version}" + self.purl = f"{self.type}/{self.name}@{self.version}" + + +class Dependency: + branch: str + id: int + name: str + type: str + version: str + namespace: str + repository: str + + def __init__(self, **kwargs): + if kwargs: + for key, value in kwargs.items(): + setattr(self, key, value) + + def __str__(self): + return json.dumps(self.__dict__) + + +class Org: + id: int + image: str + name: str + plan: str + + def __init__(self, **kwargs): + if kwargs: + for key, value in kwargs.items(): + setattr(self, key, value) + + def __str__(self): + return json.dumps(self.__dict__) + + +class Response: + text: str + error: bool + status_code: int + + def __init__(self, text: str, error: bool, status_code: int): + self.text = text + self.error = error + self.status_code = status_code + + def __str__(self): + return json.dumps(self.__dict__) + + def json(self): + return self.__dict__ + + +class DependGetData: + url: str + headers: dict + payload: str + + def __init__(self, **kwargs): + if kwargs: + for key, value in kwargs.items(): + setattr(self, key, value) + + def __str__(self): + return json.dumps(self.__dict__) diff --git a/socketdev/dependencies/__init__.py b/socketdev/dependencies/__init__.py index 06c65b0..dbe4fd7 100644 --- a/socketdev/dependencies/__init__.py +++ b/socketdev/dependencies/__init__.py @@ -1,53 +1,53 @@ -import json -from urllib.parse import urlencode -import logging -from socketdev.tools import load_files -from ..utils import Utils - -log = logging.getLogger("socketdev") - -# TODO: Add types for responses. Not currently used in the CLI. - - -class Dependencies: - def __init__(self, api): - self.api = api - - def post(self, files: list, params: dict, use_lazy_loading: bool = True, workspace: str = None, base_path: str = None) -> dict: - if use_lazy_loading: - loaded_files = Utils.load_files_for_sending_lazy(files, workspace, base_path=base_path) - else: - loaded_files = [] - loaded_files = load_files(files, loaded_files) - - path = "dependencies/upload?" + urlencode(params) - response = self.api.do_request(path=path, files=loaded_files, method="POST") - if response.status_code == 200: - result = response.json() - else: - result = {} - log.error(f"Error posting {files} to the Dependency API") - log.error(response.text) - return result - - def get(self, org_slug: str = None, ecosystem: str = None, package: str = None, version: str = None, **kwargs) -> dict: - # If all specific parameters are provided, use the specific dependency endpoint - if org_slug and ecosystem and package and version: - path = f"orgs/{org_slug}/dependencies/{ecosystem}/{package}/{version}" - response = self.api.do_request(path=path, method="GET") - else: - # Otherwise use the search endpoint - limit = kwargs.get('limit', 50) - offset = kwargs.get('offset', 0) - path = "dependencies/search" - payload = {"limit": limit, "offset": offset} - payload_str = json.dumps(payload) - response = self.api.do_request(path=path, method="POST", payload=payload_str) - - if response.status_code == 200: - result = response.json() - else: - result = {} - log.error("Unable to retrieve Dependencies") - log.error(response.text) - return result +import json +from urllib.parse import urlencode +import logging +from socketdev.tools import load_files +from ..utils import Utils + +log = logging.getLogger("socketdev") + +# TODO: Add types for responses. Not currently used in the CLI. + + +class Dependencies: + def __init__(self, api): + self.api = api + + def post(self, files: list, params: dict, use_lazy_loading: bool = True, workspace: str = None, base_path: str = None) -> dict: + if use_lazy_loading: + loaded_files = Utils.load_files_for_sending_lazy(files, workspace, base_path=base_path) + else: + loaded_files = [] + loaded_files = load_files(files, loaded_files) + + path = "dependencies/upload?" + urlencode(params) + response = self.api.do_request(path=path, files=loaded_files, method="POST") + if response.status_code == 200: + result = response.json() + else: + result = {} + log.error(f"Error posting {files} to the Dependency API") + log.error(response.text) + return result + + def get(self, org_slug: str = None, ecosystem: str = None, package: str = None, version: str = None, **kwargs) -> dict: + # If all specific parameters are provided, use the specific dependency endpoint + if org_slug and ecosystem and package and version: + path = f"orgs/{org_slug}/dependencies/{ecosystem}/{package}/{version}" + response = self.api.do_request(path=path, method="GET") + else: + # Otherwise use the search endpoint + limit = kwargs.get('limit', 50) + offset = kwargs.get('offset', 0) + path = "dependencies/search" + payload = {"limit": limit, "offset": offset} + payload_str = json.dumps(payload) + response = self.api.do_request(path=path, method="POST", payload=payload_str) + + if response.status_code == 200: + result = response.json() + else: + result = {} + log.error("Unable to retrieve Dependencies") + log.error(response.text) + return result diff --git a/socketdev/npm/__init__.py b/socketdev/npm/__init__.py index d26dc86..bdb213c 100644 --- a/socketdev/npm/__init__.py +++ b/socketdev/npm/__init__.py @@ -1,28 +1,28 @@ -import logging - -log = logging.getLogger("socketdev") - -# TODO: Add response type classes for NPM endpoints - - -class NPM: - def __init__(self, api): - self.api = api - - def issues(self, package: str, version: str) -> list: - path = f"npm/{package}/{version}/issues" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error getting npm issues: {response.status_code}") - log.error(response.text) - return [] - - def score(self, package: str, version: str) -> list: - path = f"npm/{package}/{version}/score" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error getting npm score: {response.status_code}") - log.error(response.text) - return [] +import logging + +log = logging.getLogger("socketdev") + +# TODO: Add response type classes for NPM endpoints + + +class NPM: + def __init__(self, api): + self.api = api + + def issues(self, package: str, version: str) -> list: + path = f"npm/{package}/{version}/issues" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error getting npm issues: {response.status_code}") + log.error(response.text) + return [] + + def score(self, package: str, version: str) -> list: + path = f"npm/{package}/{version}/score" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error getting npm score: {response.status_code}") + log.error(response.text) + return [] diff --git a/socketdev/openapi/__init__.py b/socketdev/openapi/__init__.py index 70854ba..a9f2950 100644 --- a/socketdev/openapi/__init__.py +++ b/socketdev/openapi/__init__.py @@ -1,19 +1,19 @@ -import logging - -log = logging.getLogger("socketdev") - -# TODO: Add response type classes for OpenAPI endpoints - - -class OpenAPI: - def __init__(self, api): - self.api = api - - def get(self) -> dict: - path = "openapi" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error getting OpenAPI spec: {response.status_code}") - log.error(response.text) - return {} +import logging + +log = logging.getLogger("socketdev") + +# TODO: Add response type classes for OpenAPI endpoints + + +class OpenAPI: + def __init__(self, api): + self.api = api + + def get(self) -> dict: + path = "openapi" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error getting OpenAPI spec: {response.status_code}") + log.error(response.text) + return {} diff --git a/socketdev/org/__init__.py b/socketdev/org/__init__.py index 20a02dc..5496479 100644 --- a/socketdev/org/__init__.py +++ b/socketdev/org/__init__.py @@ -1,34 +1,34 @@ -from typing import TypedDict, Dict -import logging - -log = logging.getLogger("socketdev") - - -class Organization(TypedDict): - id: str - name: str - image: str - plan: str - slug: str - - -class OrganizationsResponse(TypedDict): - organizations: Dict[str, Organization] - # Add other fields from the response if needed - - -class Orgs: - def __init__(self, api): - self.api = api - - def get(self, use_types: bool = False) -> OrganizationsResponse: - path = "organizations" - response = self.api.do_request(path=path) - if response.status_code == 200: - result = response.json() - if use_types: - return OrganizationsResponse(result) - return result - log.error(f"Error getting organizations: {response.status_code}") - log.error(response.text) - return {"organizations": {}} +from typing import TypedDict, Dict +import logging + +log = logging.getLogger("socketdev") + + +class Organization(TypedDict): + id: str + name: str + image: str + plan: str + slug: str + + +class OrganizationsResponse(TypedDict): + organizations: Dict[str, Organization] + # Add other fields from the response if needed + + +class Orgs: + def __init__(self, api): + self.api = api + + def get(self, use_types: bool = False) -> OrganizationsResponse: + path = "organizations" + response = self.api.do_request(path=path) + if response.status_code == 200: + result = response.json() + if use_types: + return OrganizationsResponse(result) + return result + log.error(f"Error getting organizations: {response.status_code}") + log.error(response.text) + return {"organizations": {}} diff --git a/socketdev/quota/__init__.py b/socketdev/quota/__init__.py index 2fc797a..d7a7a2c 100644 --- a/socketdev/quota/__init__.py +++ b/socketdev/quota/__init__.py @@ -1,19 +1,19 @@ -import logging - -log = logging.getLogger("socketdev") - -# TODO: Add response type classes for Quota endpoints - - -class Quota: - def __init__(self, api): - self.api = api - - def get(self) -> dict: - path = "quota" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error getting quota: {response.status_code}") - log.error(response.text) - return {} +import logging + +log = logging.getLogger("socketdev") + +# TODO: Add response type classes for Quota endpoints + + +class Quota: + def __init__(self, api): + self.api = api + + def get(self) -> dict: + path = "quota" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error getting quota: {response.status_code}") + log.error(response.text) + return {} diff --git a/socketdev/report/__init__.py b/socketdev/report/__init__.py index a5c255b..f0331ea 100644 --- a/socketdev/report/__init__.py +++ b/socketdev/report/__init__.py @@ -1,87 +1,87 @@ -import logging -from datetime import datetime, timedelta, timezone - -log = logging.getLogger("socketdev") - -# TODO: Add response type classes for Report endpoints - - -class Report: - def __init__(self, api): - self.api = api - - def list(self, from_time: int = None) -> dict: - """ - This function will return all reports from time specified. - :param from_time: Unix epoch time in seconds. Will default self, to 30 days - """ - if from_time is None: - from_time = int((datetime.now(timezone.utc) - timedelta(days=30)).timestamp()) - else: - from_time = int((datetime.now(timezone.utc) - timedelta(seconds=from_time)).timestamp()) - - path = "report/list" - if from_time is not None: - path += f"?from={from_time}" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error listing reports: {response.status_code}") - log.error(response.text) - return {} - - def delete(self, report_id: str) -> bool: - path = f"report/delete/{report_id}" - response = self.api.do_request(path=path, method="DELETE") - if response.status_code == 200: - return True - log.error(f"Error deleting report: {response.status_code}") - log.error(response.text) - return False - - def view(self, report_id) -> dict: - path = f"report/view/{report_id}" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error viewing report: {response.status_code}") - log.error(response.text) - return {} - - def supported(self) -> dict: - path = "report/supported" - response = self.api.do_request(path=path) - if response.status_code == 200: - return response.json() - log.error(f"Error getting supported reports: {response.status_code}") - log.error(response.text) - return {} - - def create(self, files: list) -> dict: - # Handle both file path strings and file tuples - open_files = [] - for file_entry in files: - if isinstance(file_entry, tuple) and len(file_entry) == 2: - name, file_data = file_entry - if isinstance(file_data, tuple) and len(file_data) == 2: - # Format: [("field_name", ("filename", file_obj))] - filename, file_obj = file_data - file_info = (name, (filename, file_obj, "text/plain")) - open_files.append(file_info) - else: - # Format: [("field_name", "file_path")] - file_info = (name, (name, open(file_data, "rb"), "text/plain")) - open_files.append(file_info) - else: - # Handle other formats if needed - log.error(f"Unexpected file format: {file_entry}") - return {} - - path = "report/upload" - payload = {} - response = self.api.do_request(path=path, method="PUT", files=open_files, payload=payload) - if response.status_code in (200, 201): - return response.json() - log.error(f"Error creating report: {response.status_code}") - log.error(response.text) - return {} +import logging +from datetime import datetime, timedelta, timezone + +log = logging.getLogger("socketdev") + +# TODO: Add response type classes for Report endpoints + + +class Report: + def __init__(self, api): + self.api = api + + def list(self, from_time: int = None) -> dict: + """ + This function will return all reports from time specified. + :param from_time: Unix epoch time in seconds. Will default self, to 30 days + """ + if from_time is None: + from_time = int((datetime.now(timezone.utc) - timedelta(days=30)).timestamp()) + else: + from_time = int((datetime.now(timezone.utc) - timedelta(seconds=from_time)).timestamp()) + + path = "report/list" + if from_time is not None: + path += f"?from={from_time}" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error listing reports: {response.status_code}") + log.error(response.text) + return {} + + def delete(self, report_id: str) -> bool: + path = f"report/delete/{report_id}" + response = self.api.do_request(path=path, method="DELETE") + if response.status_code == 200: + return True + log.error(f"Error deleting report: {response.status_code}") + log.error(response.text) + return False + + def view(self, report_id) -> dict: + path = f"report/view/{report_id}" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error viewing report: {response.status_code}") + log.error(response.text) + return {} + + def supported(self) -> dict: + path = "report/supported" + response = self.api.do_request(path=path) + if response.status_code == 200: + return response.json() + log.error(f"Error getting supported reports: {response.status_code}") + log.error(response.text) + return {} + + def create(self, files: list) -> dict: + # Handle both file path strings and file tuples + open_files = [] + for file_entry in files: + if isinstance(file_entry, tuple) and len(file_entry) == 2: + name, file_data = file_entry + if isinstance(file_data, tuple) and len(file_data) == 2: + # Format: [("field_name", ("filename", file_obj))] + filename, file_obj = file_data + file_info = (name, (filename, file_obj, "text/plain")) + open_files.append(file_info) + else: + # Format: [("field_name", "file_path")] + file_info = (name, (name, open(file_data, "rb"), "text/plain")) + open_files.append(file_info) + else: + # Handle other formats if needed + log.error(f"Unexpected file format: {file_entry}") + return {} + + path = "report/upload" + payload = {} + response = self.api.do_request(path=path, method="PUT", files=open_files, payload=payload) + if response.status_code in (200, 201): + return response.json() + log.error(f"Error creating report: {response.status_code}") + log.error(response.text) + return {} diff --git a/socketdev/repositories/__init__.py b/socketdev/repositories/__init__.py index 19aaa5f..e63faf9 100644 --- a/socketdev/repositories/__init__.py +++ b/socketdev/repositories/__init__.py @@ -1,31 +1,31 @@ -from typing import TypedDict, Union -import logging - -log = logging.getLogger("socketdev") - - -class Repo(TypedDict): - name: str - description: str - homepage: str - visibility: str - archived: bool - default_branch: str - - -class Repositories: - def __init__(self, api): - self.api = api - - def list(self, use_types: bool = False) -> Union[dict, list[Repo]]: - path = "repos" - response = self.api.do_request(path=path) - if response.status_code == 200: - result = response.json() - if use_types: - return [Repo(repo) for repo in result] - return result - - log.error(f"Error listing repositories: {response.status_code}") - log.error(response.text) - return [] +from typing import TypedDict, Union +import logging + +log = logging.getLogger("socketdev") + + +class Repo(TypedDict): + name: str + description: str + homepage: str + visibility: str + archived: bool + default_branch: str + + +class Repositories: + def __init__(self, api): + self.api = api + + def list(self, use_types: bool = False) -> Union[dict, list[Repo]]: + path = "repos" + response = self.api.do_request(path=path) + if response.status_code == 200: + result = response.json() + if use_types: + return [Repo(repo) for repo in result] + return result + + log.error(f"Error listing repositories: {response.status_code}") + log.error(response.text) + return [] diff --git a/socketdev/settings/__init__.py b/socketdev/settings/__init__.py index 3812726..1ba8724 100644 --- a/socketdev/settings/__init__.py +++ b/socketdev/settings/__init__.py @@ -1,182 +1,182 @@ -import logging -from enum import Enum -from typing import Dict, Optional, Union -from dataclasses import dataclass, asdict - -from ..core.enums import unknown_enum_value - -log = logging.getLogger("socketdev") - - -class SecurityAction(str, Enum): - DEFER = "defer" - ERROR = "error" - WARN = "warn" - MONITOR = "monitor" - IGNORE = "ignore" - - @classmethod - def _missing_(cls, value): - # DEFER, not a new UNKNOWN sentinel: an unrecognized action is a policy - # decision this SDK cannot make, and "defer" already means "use the - # configured default". IGNORE would silently disable a rule and ERROR - # would fail builds on a value the API considers routine. - return unknown_enum_value(cls.__name__, value, cls.DEFER) - - -@dataclass -class SecurityPolicyRule: - action: SecurityAction - - def __getitem__(self, key): - return getattr(self, key) - - def to_dict(self): - return asdict(self) - - @classmethod - def from_dict(cls, data: dict) -> "SecurityPolicyRule": - return cls(action=SecurityAction(data["action"])) - - -@dataclass -class OrgSecurityPolicyResponse: - success: bool - status: int - securityPolicyRules: Optional[Dict[str, SecurityPolicyRule]] = None - message: Optional[str] = None - - def __getitem__(self, key): - return getattr(self, key) - - def to_dict(self): - return asdict(self) - - @classmethod - def from_dict(cls, data: dict) -> "OrgSecurityPolicyResponse": - return cls( - securityPolicyRules={k: SecurityPolicyRule.from_dict(v) for k, v in data["securityPolicyRules"].items()} - if data.get("securityPolicyRules") - else None, - success=data["success"], - status=data["status"], - message=data.get("message"), - ) - - -class Settings: - def __init__(self, api): - self.api = api - - def create_params_string(self, params: dict) -> str: - param_str = "" - - for name, value in params.items(): - if value: - if name == "committers" and isinstance(value, list): - # Handle committers specially - add multiple params - for committer in value: - param_str += f"&{name}={committer}" - else: - param_str += f"&{name}={value}" - - param_str = "?" + param_str.lstrip("&") - return param_str - - def get( - self, org_slug: str, custom_rules_only: bool = False, use_types: bool = False - ) -> Union[dict, OrgSecurityPolicyResponse]: - path = f"orgs/{org_slug}/settings/security-policy" - params = {"custom_rules_only": custom_rules_only} - params_args = self.create_params_string(params) if custom_rules_only else "" - path += params_args - response = self.api.do_request(path=path, method="GET") - - if response.status_code == 200: - rules = response.json() - if use_types: - return OrgSecurityPolicyResponse.from_dict( - {"securityPolicyRules": rules.get("securityPolicyRules", {}), "success": True, "status": 200} - ) - return rules - - error_message = response.json().get("error", {}).get("message", "Unknown error") - log.error(f"Failed to get security policy: {response.status_code}, message: {error_message}") - if use_types: - return OrgSecurityPolicyResponse.from_dict( - {"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 {} +import logging +from enum import Enum +from typing import Dict, Optional, Union +from dataclasses import dataclass, asdict + +from ..core.enums import unknown_enum_value + +log = logging.getLogger("socketdev") + + +class SecurityAction(str, Enum): + DEFER = "defer" + ERROR = "error" + WARN = "warn" + MONITOR = "monitor" + IGNORE = "ignore" + + @classmethod + def _missing_(cls, value): + # DEFER, not a new UNKNOWN sentinel: an unrecognized action is a policy + # decision this SDK cannot make, and "defer" already means "use the + # configured default". IGNORE would silently disable a rule and ERROR + # would fail builds on a value the API considers routine. + return unknown_enum_value(cls.__name__, value, cls.DEFER) + + +@dataclass +class SecurityPolicyRule: + action: SecurityAction + + def __getitem__(self, key): + return getattr(self, key) + + def to_dict(self): + return asdict(self) + + @classmethod + def from_dict(cls, data: dict) -> "SecurityPolicyRule": + return cls(action=SecurityAction(data["action"])) + + +@dataclass +class OrgSecurityPolicyResponse: + success: bool + status: int + securityPolicyRules: Optional[Dict[str, SecurityPolicyRule]] = None + message: Optional[str] = None + + def __getitem__(self, key): + return getattr(self, key) + + def to_dict(self): + return asdict(self) + + @classmethod + def from_dict(cls, data: dict) -> "OrgSecurityPolicyResponse": + return cls( + securityPolicyRules={k: SecurityPolicyRule.from_dict(v) for k, v in data["securityPolicyRules"].items()} + if data.get("securityPolicyRules") + else None, + success=data["success"], + status=data["status"], + message=data.get("message"), + ) + + +class Settings: + def __init__(self, api): + self.api = api + + def create_params_string(self, params: dict) -> str: + param_str = "" + + for name, value in params.items(): + if value: + if name == "committers" and isinstance(value, list): + # Handle committers specially - add multiple params + for committer in value: + param_str += f"&{name}={committer}" + else: + param_str += f"&{name}={value}" + + param_str = "?" + param_str.lstrip("&") + return param_str + + def get( + self, org_slug: str, custom_rules_only: bool = False, use_types: bool = False + ) -> Union[dict, OrgSecurityPolicyResponse]: + path = f"orgs/{org_slug}/settings/security-policy" + params = {"custom_rules_only": custom_rules_only} + params_args = self.create_params_string(params) if custom_rules_only else "" + path += params_args + response = self.api.do_request(path=path, method="GET") + + if response.status_code == 200: + rules = response.json() + if use_types: + return OrgSecurityPolicyResponse.from_dict( + {"securityPolicyRules": rules.get("securityPolicyRules", {}), "success": True, "status": 200} + ) + return rules + + error_message = response.json().get("error", {}).get("message", "Unknown error") + log.error(f"Failed to get security policy: {response.status_code}, message: {error_message}") + if use_types: + return OrgSecurityPolicyResponse.from_dict( + {"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 {} diff --git a/socketdev/tools/__init__.py b/socketdev/tools/__init__.py index 3333fa2..5cdf2e4 100644 --- a/socketdev/tools/__init__.py +++ b/socketdev/tools/__init__.py @@ -1,65 +1,65 @@ -import glob -import sys -import platform - - -def find_package_files(folder: str, file_types: list) -> list: - files = [] - for file_type in file_types: - search_pattern = f"{folder}/**/{file_type}" - result = glob.glob(search_pattern, recursive=True) - if sys.platform.lower() == "win32": - result = fix_file_path(result) - files.extend(result) - return files - - -def fix_file_path(files) -> list: - fixed_files = [] - for file in files: - file = file.replace("\\", "/") - fixed_files.append(file) - return fixed_files - - - - -def load_files(files: list, loaded_files: list, workspace: str = None) -> list: - for file in files: - if platform.system() == "Windows": - file = file.replace("\\", "/") - if "/" in file: - path, name = file.rsplit("/", 1) - else: - path = "." - name = file - full_path = f"{path}/{name}" - - # Calculate key based on workspace if provided - if workspace and full_path.startswith(workspace): - key = full_path[len(workspace):] - key = key.lstrip("/") - key = key.lstrip("./") - else: - key = full_path - - payload = (key, (name, open(full_path, "rb"))) - loaded_files.append(payload) - return loaded_files - - -def prepare_for_csv(dependencies: list, packages: dict) -> list: - output = [] - for dependency in dependencies: - if dependency.name in packages: - for package in packages[dependency.name]: - output_object = [ - dependency.repository, - dependency.branch, - package.name, - package.version, - package.license, - package.repository, - ] - output.append(output_object) - return output +import glob +import sys +import platform + + +def find_package_files(folder: str, file_types: list) -> list: + files = [] + for file_type in file_types: + search_pattern = f"{folder}/**/{file_type}" + result = glob.glob(search_pattern, recursive=True) + if sys.platform.lower() == "win32": + result = fix_file_path(result) + files.extend(result) + return files + + +def fix_file_path(files) -> list: + fixed_files = [] + for file in files: + file = file.replace("\\", "/") + fixed_files.append(file) + return fixed_files + + + + +def load_files(files: list, loaded_files: list, workspace: str = None) -> list: + for file in files: + if platform.system() == "Windows": + file = file.replace("\\", "/") + if "/" in file: + path, name = file.rsplit("/", 1) + else: + path = "." + name = file + full_path = f"{path}/{name}" + + # Calculate key based on workspace if provided + if workspace and full_path.startswith(workspace): + key = full_path[len(workspace):] + key = key.lstrip("/") + key = key.lstrip("./") + else: + key = full_path + + payload = (key, (name, open(full_path, "rb"))) + loaded_files.append(payload) + return loaded_files + + +def prepare_for_csv(dependencies: list, packages: dict) -> list: + output = [] + for dependency in dependencies: + if dependency.name in packages: + for package in packages[dependency.name]: + output_object = [ + dependency.repository, + dependency.branch, + package.name, + package.version, + package.license, + package.repository, + ] + output.append(output_object) + return output From d7bc034d335e99d5b7a91062d0985c6c26170730 Mon Sep 17 00:00:00 2001 From: lelia <2418071+lelia@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:40:02 -0400 Subject: [PATCH 4/4] fix(enums): keep ScanType strict, it only ever builds requests ScanType never parses an API response. FullScanParams.to_dict() is urlencoded onto the create-scan query string, so giving it a _missing_ fallback meant a caller typo silently shipped scan_type=unknown to the API instead of failing at construction. The same from_dict already passes integration_type through uncoerced for that reason. It is now recorded in REQUEST_ONLY_ENUMS, the opt-out the invariant test always had and this branch had left empty, and a new test asserts request-only enums keep raising so the exemption cannot quietly become a skip. Also bumps actions/setup-python in the new workflow to v7.0.0, matching the pin already used by .github/actions/setup-sfw. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/api-drift-check.yml | 2 +- CHANGELOG.md | 12 ++++++----- scripts/check_api_enum_drift.py | 11 +++++++++- socketdev/fullscans/__init__.py | 10 ++++----- tests/unit/test_enum_forward_compat.py | 28 +++++++++++++++++++++++++- 5 files changed, 50 insertions(+), 13 deletions(-) diff --git a/.github/workflows/api-drift-check.yml b/.github/workflows/api-drift-check.yml index e25c55a..d308abf 100644 --- a/.github/workflows/api-drift-check.yml +++ b/.github/workflows/api-drift-check.yml @@ -26,7 +26,7 @@ jobs: fetch-depth: 1 persist-credentials: false - - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: python-version: "3.12" diff --git a/CHANGELOG.md b/CHANGELOG.md index b002357..71f6099 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,14 @@ ### Changed: every API-sourced enum now tolerates unknown values -- `SocketIssueSeverity`, `SocketCategory`, `DiffType`, `ScanType` and - `SecurityAction` now fall back to a documented member instead of raising +- `SocketIssueSeverity`, `SocketCategory`, `DiffType` and `SecurityAction` + now fall back to a documented member instead of raising `ValueError` when the API sends a value this release does not know about. - `SocketPURL_Type` already behaved this way; the other five did not, so each - was one backend addition away from emptying a response the same way issue #78 - and the unknown `generic` purl type did. + `SocketPURL_Type` already behaved this way; the others did not, so each was + one backend addition away from emptying a response the same way issue #78 and + the unknown `generic` purl type did. `ScanType` is deliberately left strict: + it is only ever urlencoded onto the create-scan request, so an unrecognized + value is a caller typo rather than API drift. - Fallbacks are deliberate rather than convenient. `SocketIssueSeverity` and `DiffType` gained an explicit `UNKNOWN` member because guessing an existing level would either hide a real finding or invent one, and `SecurityAction` diff --git a/scripts/check_api_enum_drift.py b/scripts/check_api_enum_drift.py index 0c1f2da..af9b9c3 100755 --- a/scripts/check_api_enum_drift.py +++ b/scripts/check_api_enum_drift.py @@ -43,10 +43,14 @@ SocketIssueSeverity: "SocketIssueSeverity", SocketCategory: "SocketCategory", DiffType: "SocketDiffArtifactType", - ScanType: None, SecurityAction: None, } +# Enums the SDK only ever sends, never parses. Drift in the API's copy cannot +# break parsing here, and these are intentionally strict so a caller typo fails +# at construction, so they are out of scope for this check rather than a gap. +REQUEST_ONLY = (ScanType,) + # Members this SDK adds deliberately, which the API will never send. They are # the documented _missing_ fallbacks (see socketdev/core/enums.py), so their # absence from the spec is expected rather than drift. @@ -122,6 +126,11 @@ def main(): f"named schema for them: {', '.join(sorted(unmapped))}" ) + print( + f"not applicable (request-only, intentionally strict): " + f"{', '.join(sorted(e.__name__ for e in REQUEST_ONLY))}" + ) + if drifted: print( "\nAdd the missing members to the SDK enum. Existing values are " diff --git a/socketdev/fullscans/__init__.py b/socketdev/fullscans/__init__.py index 770440d..45c64f1 100644 --- a/socketdev/fullscans/__init__.py +++ b/socketdev/fullscans/__init__.py @@ -102,14 +102,14 @@ def _missing_(cls, value): class ScanType(str, Enum): + # Deliberately strict, unlike the response-parsed enums in this module. + # ScanType only ever travels outbound: FullScanParams.to_dict() is + # urlencoded onto the create-scan query string, so an unrecognized value is + # the caller's typo, not API drift. Coercing it to a fallback would send + # scan_type=unknown to the API instead of failing at construction. SOCKET = "socket" SOCKET_TIER1 = "socket_tier1" SOCKET_BASICS = "socket_basics" - UNKNOWN = "unknown" - - @classmethod - def _missing_(cls, value): - return unknown_enum_value(cls.__name__, value, cls.UNKNOWN) @dataclass(kw_only=True) diff --git a/tests/unit/test_enum_forward_compat.py b/tests/unit/test_enum_forward_compat.py index 3511b32..948b64e 100644 --- a/tests/unit/test_enum_forward_compat.py +++ b/tests/unit/test_enum_forward_compat.py @@ -25,7 +25,15 @@ # Strictness is correct there: a bad value is the caller's typo and should raise # rather than be silently coerced. Add an entry only with a comment justifying # that the enum never sees API-supplied values. -REQUEST_ONLY_ENUMS = frozenset() +REQUEST_ONLY_ENUMS = frozenset( + { + # Only ever travels outbound: FullScanParams.to_dict() is urlencoded + # onto the create-scan query string. It never parses an API response, so + # an unrecognized value is a caller typo that should surface at + # construction rather than reach the API as scan_type=unknown. + "ScanType", + } +) # A value the API will never legitimately send. SENTINEL = "__value_the_api_would_never_send__" @@ -98,6 +106,24 @@ def test_unknown_value_warns(self): f"got: {captured.output}", ) + def test_request_only_enums_stay_strict(self): + # The opt-out is not a "skip this one" marker: these enums must actively + # keep raising. Coercing a caller's typo to a fallback would send the + # fallback to the API instead of failing at construction, which is how + # the forward-compat change first got ScanType wrong. + by_name = {cls.__name__: cls for cls in _all_enums().values()} + for name in sorted(REQUEST_ONLY_ENUMS): + with self.subTest(enum=name): + enum_cls = by_name.get(name) + self.assertIsNotNone( + enum_cls, f"{name} is exempted but no longer exists" + ) + with self.assertRaises( + ValueError, + msg=f"{name} is request-only and must reject unknown values", + ): + enum_cls(SENTINEL) + def test_known_values_still_round_trip(self): # Forward-compat must not swallow legitimate values. for qualname, enum_cls in sorted(_all_enums().items()):