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/.github/workflows/api-drift-check.yml b/.github/workflows/api-drift-check.yml new file mode 100644 index 0000000..d308abf --- /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@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.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..71f6099 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,42 @@ # Changelog +## 3.6.0 + +### Changed: every API-sourced enum now tolerates unknown values + +- `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 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` + 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/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/scripts/check_api_enum_drift.py b/scripts/check_api_enum_drift.py new file mode 100755 index 0000000..af9b9c3 --- /dev/null +++ b/scripts/check_api_enum_drift.py @@ -0,0 +1,148 @@ +#!/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", + 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. +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))}" + ) + + 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 " + "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/__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/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/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/fullscans/__init__.py b/socketdev/fullscans/__init__.py index 2fe6f09..45c64f1 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,9 +92,21 @@ 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): + # 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" @@ -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/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 a9cbe8e..1ba8724 100644 --- a/socketdev/settings/__init__.py +++ b/socketdev/settings/__init__.py @@ -1,172 +1,182 @@ -import logging -from enum import Enum -from typing import Dict, Optional, Union -from dataclasses import dataclass, asdict - -log = logging.getLogger("socketdev") - - -class SecurityAction(str, Enum): - DEFER = "defer" - ERROR = "error" - WARN = "warn" - MONITOR = "monitor" - IGNORE = "ignore" - - -@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 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..948b64e --- /dev/null +++ b/tests/unit/test_enum_forward_compat.py @@ -0,0 +1,136 @@ +"""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( + { + # 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__" + + +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_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()): + 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() 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