diff --git a/socketdev/core/api.py b/socketdev/core/api.py index 8c35231..1187f45 100644 --- a/socketdev/core/api.py +++ b/socketdev/core/api.py @@ -1,4 +1,5 @@ import base64 +from typing import Optional from socketdev.log import log import requests @@ -47,6 +48,8 @@ def do_request( payload: [dict, str] = None, files: list = None, method: str = "GET", + *, + request_timeout: Optional[float] = None, ) -> Response: if self.encoded_key is None or self.encoded_key == "": raise APIKeyMissing @@ -68,7 +71,8 @@ def format_headers(headers_dict): response = requests.request( method.upper(), url, headers=headers, data=payload, files=files, - timeout=self.request_timeout, verify=not self.allow_unverified + timeout=self.request_timeout if request_timeout is None else request_timeout, + verify=not self.allow_unverified ) request_duration = time.time() - start_time diff --git a/socketdev/fullscans/__init__.py b/socketdev/fullscans/__init__.py index 45c64f1..187060a 100644 --- a/socketdev/fullscans/__init__.py +++ b/socketdev/fullscans/__init__.py @@ -4,9 +4,12 @@ from typing import Any, Dict, List, Optional, Union from dataclasses import dataclass, asdict, field import urllib.parse +import math +import time from ..core.dedupe import Dedupe from ..core.enums import unknown_enum_value from ..utils import IntegrationType, Utils +from ..exceptions import APIFailure, APITimeout log = logging.getLogger("socketdev") @@ -960,15 +963,71 @@ def stream_diff( ) return {} - def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> Union[dict, FullScanStreamResponse]: - path = "orgs/" + org_slug + "/full-scans/" + full_scan_id - response = self.api.do_request(path=path, method="GET") + def stream( + self, + org_slug: str, + full_scan_id: str, + use_types: bool = False, + *, + poll_timeout_seconds: Optional[float] = None, + poll_initial_interval_seconds: float = 5.0, + poll_max_interval_seconds: float = 10.0, + ) -> Union[dict, FullScanStreamResponse]: + request_timeout = self.api.request_timeout + timeout = request_timeout if poll_timeout_seconds is None else poll_timeout_seconds + values = (timeout, poll_initial_interval_seconds, poll_max_interval_seconds, request_timeout) + if any( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value <= 0 + for value in values + ): + raise ValueError("Full scan polling timeout and intervals must be positive finite numbers") + if poll_initial_interval_seconds > poll_max_interval_seconds: + raise ValueError("Initial polling interval must not exceed maximum polling interval") + + path = ( + "orgs/" + urllib.parse.quote(org_slug, safe="") + "/full-scans/" + + urllib.parse.quote(full_scan_id, safe="") + "?cached=true" + ) + deadline = time.monotonic() + timeout + interval = poll_initial_interval_seconds + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise APITimeout("Timed out waiting for full scan results") + try: + response = self.api.do_request( + path=path, + method="GET", + request_timeout=min(30.0, float(request_timeout), remaining), + ) + except APIFailure as error: + if not error.is_transient_error(): + raise + remaining = deadline - time.monotonic() + if remaining <= 0: + raise APITimeout("Timed out waiting for full scan results") from error + time.sleep(min(interval, remaining)) + interval = min(poll_max_interval_seconds, interval * 1.5) + continue + if response.status_code == 202: + response.close() + remaining = deadline - time.monotonic() + if remaining <= 0: + raise APITimeout("Timed out waiting for full scan results") + time.sleep(min(interval, remaining)) + interval = min(poll_max_interval_seconds, interval * 1.5) + continue + break if response.status_code == 200: try: stream_str = [] artifacts = {} result = response.text + response.close() result = result.strip('"').strip() for line in result.split("\n"): if line != '"' and line != "" and line is not None: @@ -993,6 +1052,7 @@ def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> U return artifacts except Exception as e: + response.close() error_message = f"Error parsing stream response: {str(e)}" log.error(error_message) if use_types: @@ -1001,7 +1061,10 @@ def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> U ) return {} - error_message = response.json().get("error", {}).get("message", "Unknown error") + try: + error_message = response.json().get("error", {}).get("message", "Unknown error") + finally: + response.close() log.error(f"Error streaming full scan: {response.status_code}, message: {error_message}") if use_types: return FullScanStreamResponse.from_dict( diff --git a/socketdev/version.py b/socketdev/version.py index 85197cb..46f67e7 100644 --- a/socketdev/version.py +++ b/socketdev/version.py @@ -1 +1 @@ -__version__ = "3.6.0" +__version__ = "3.7.0" diff --git a/tests/support/socket_api_fixture.py b/tests/support/socket_api_fixture.py new file mode 100644 index 0000000..d884842 --- /dev/null +++ b/tests/support/socket_api_fixture.py @@ -0,0 +1,121 @@ +"""Reusable localhost HTTP fixture for real Socket SDK transport tests.""" + +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Callable, Dict, List, Optional, Tuple +from urllib.parse import parse_qs, urlsplit + + +@dataclass +class SocketRequest: + method: str + path: str + query: Dict[str, List[str]] + headers: Dict[str, str] + body: bytes + + +@dataclass +class SocketResponse: + status: int + payload: object + raw_body: Optional[bytes] = None + + +Route = Callable[[SocketRequest], SocketResponse] + + +class SocketApiFixture: + def __init__(self) -> None: + self.requests: List[SocketRequest] = [] + self._routes: Dict[Tuple[str, str], Route] = {} + self._server: Optional[ThreadingHTTPServer] = None + self._thread: Optional[threading.Thread] = None + + @property + def base_url(self) -> str: + if self._server is None: + raise RuntimeError("Socket API fixture is not running") + host, port = self._server.server_address + return f"http://{host}:{port}/v0" + + def route( + self, + method: str, + path: str, + response: SocketResponse | Route, + ) -> None: + if callable(response): + handler = response + else: + + def handler(_request: SocketRequest) -> SocketResponse: + return response + + self._routes[(method.upper(), path)] = handler + + def start(self) -> None: + fixture = self + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self._dispatch() + + def do_POST(self) -> None: + self._dispatch() + + def _dispatch(self) -> None: + parsed = urlsplit(self.path) + length = int(self.headers.get("Content-Length", "0")) + request = SocketRequest( + method=self.command, + path=parsed.path.removeprefix("/v0/"), + query=parse_qs(parsed.query, keep_blank_values=True), + headers=dict(self.headers.items()), + body=self.rfile.read(length), + ) + fixture.requests.append(request) + route = fixture._routes.get((request.method, request.path)) + if route is None and request.path == "organizations": + response = SocketResponse( + 200, + {"organizations": {"org-fixture": {"slug": "fixture-org"}}}, + ) + elif route is None: + response = SocketResponse(404, {"error": {"message": "Not found"}}) + else: + response = route(request) + body = response.raw_body + if body is None: + body = json.dumps(response.payload).encode("utf-8") + self.send_response(response.status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, _format: str, *args: object) -> None: + return + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + + def close(self) -> None: + if self._server is not None: + self._server.shutdown() + self._server.server_close() + self._server = None + if self._thread is not None: + self._thread.join(timeout=2) + self._thread = None + + +def socket_api_fixture() -> SocketApiFixture: + fixture = SocketApiFixture() + fixture.start() + return fixture diff --git a/tests/unit/test_full_scan_cached_polling_http.py b/tests/unit/test_full_scan_cached_polling_http.py new file mode 100644 index 0000000..e3e9f90 --- /dev/null +++ b/tests/unit/test_full_scan_cached_polling_http.py @@ -0,0 +1,155 @@ +import json + +import pytest + +from socketdev import socketdev +from socketdev.fullscans import FullScanStreamResponse +from socketdev.exceptions import APIAccessDenied, APIResourceNotFound, APITimeout +from tests.support.socket_api_fixture import SocketApiFixture, SocketResponse + + +@pytest.fixture +def api_fixture(): + fixture = SocketApiFixture() + fixture.start() + yield fixture + fixture.close() + + +def sdk_for(fixture, timeout=1): + sdk = socketdev(token="synthetic-token", timeout=timeout) + sdk.api.api_url = fixture.base_url + return sdk + + +def ndjson(*records): + return SocketResponse(200, None, raw_body="\n".join(json.dumps(record) for record in records).encode()) + + +def test_cached_polling_returns_dictionary_and_typed_results(api_fixture): + path = "orgs/org/full-scans/scan" + artifact = { + "id": "artifact-1", + "alerts": [{"key": "test-key", "type": "test", "severity": "high", "category": "quality"}], + } + api_fixture.route("GET", path, ndjson(artifact)) + sdk = sdk_for(api_fixture) + + result = sdk.fullscans.stream("org", "scan") + typed = sdk.fullscans.stream("org", "scan", use_types=True) + + assert result["artifact-1"]["alerts"][0]["type"] == "test" + assert isinstance(typed, FullScanStreamResponse) + assert typed.artifacts["artifact-1"].id == "artifact-1" + assert typed.artifacts["artifact-1"].alerts[0].type == "test" + assert all(request.query == {"cached": ["true"]} for request in api_fixture.requests) + + +def test_processing_then_ready_and_empty_stream(api_fixture): + path = "orgs/org/full-scans/scan" + responses = iter([SocketResponse(202, {"status": "processing"}), SocketResponse(202, {}), ndjson()]) + api_fixture.route("GET", path, lambda _request: next(responses)) + sdk = sdk_for(api_fixture) + + assert sdk.fullscans.stream("org", "scan", poll_timeout_seconds=0.5, poll_initial_interval_seconds=0.01) == {} + assert len(api_fixture.requests) == 3 + + +def test_transient_error_retries_and_invalid_ndjson_preserves_unsuccessful_response(api_fixture): + path = "orgs/org/full-scans/scan" + responses = iter([SocketResponse(503, {"error": {"message": "try again"}}), ndjson()]) + api_fixture.route("GET", path, lambda _request: next(responses)) + sdk = sdk_for(api_fixture) + assert sdk.fullscans.stream( + "org", "scan", poll_timeout_seconds=0.5, poll_initial_interval_seconds=0.01 + ) == {} + assert len(api_fixture.requests) == 2 + + api_fixture.requests.clear() + api_fixture.route("GET", path, SocketResponse(200, None, raw_body=b"{invalid json")) + response = sdk.fullscans.stream("org", "scan", use_types=True) + assert isinstance(response, FullScanStreamResponse) + assert not response.success + + +def test_perpetual_processing_times_out(api_fixture): + api_fixture.route("GET", "orgs/org/full-scans/scan", SocketResponse(202, {})) + with pytest.raises(APITimeout, match="full scan results"): + sdk_for(api_fixture).fullscans.stream( + "org", "scan", poll_timeout_seconds=0.06, + poll_initial_interval_seconds=0.02, poll_max_interval_seconds=0.02, + ) + + +def test_persistent_transient_error_expires_with_poll_budget(api_fixture): + api_fixture.route("GET", "orgs/org/full-scans/scan", SocketResponse(503, {})) + with pytest.raises(APITimeout, match="full scan results"): + sdk_for(api_fixture).fullscans.stream( + "org", "scan", poll_timeout_seconds=0.06, + poll_initial_interval_seconds=0.02, poll_max_interval_seconds=0.02, + ) + assert len(api_fixture.requests) >= 1 + + +def test_duplicate_artifacts_merge_alerts_and_preserve_score_event(api_fixture): + path = "orgs/org/full-scans/scan" + common = { + "id": "artifact-1", "type": "pypi", "name": "example", "version": "1.0", + "inputPurl": "pkg:pypi/example@1.0", "score": { + "supplyChain": 1, "quality": 2, "maintenance": 3, + "vulnerability": 4, "license": 5, "overall": 6, + }, + } + first = dict(common, release="wheel", alerts=[{ + "key": "alert-1", "type": "test", "severity": "high", "category": "quality", + }]) + second = dict(common, release="source", alerts=[{ + "key": "alert-1", "type": "test", "severity": "high", "category": "quality", + }]) + api_fixture.route("GET", path, ndjson(first, second)) + + result = sdk_for(api_fixture).fullscans.stream("org", "scan") + + artifact = result["artifact-1"] + assert artifact["releases"] == ["source", "wheel"] + assert artifact["score"]["overall"] == 6 + assert len(artifact["alerts"]) == 1 + assert artifact["alerts"][0]["releases"] == ["wheel", "source"] + + +@pytest.mark.parametrize("status", [403, 404]) +def test_deterministic_errors_stop_immediately(api_fixture, status): + api_fixture.route("GET", "orgs/org/full-scans/scan", SocketResponse(status, {"error": {"message": "denied"}})) + error_type = APIAccessDenied if status == 403 else APIResourceNotFound + with pytest.raises(error_type): + sdk_for(api_fixture).fullscans.stream("org", "scan", poll_timeout_seconds=0.1) + assert len(api_fixture.requests) == 1 + + +def test_invalid_settings_and_encoded_components(api_fixture): + sdk = sdk_for(api_fixture) + with pytest.raises(ValueError): + sdk.fullscans.stream("org", "scan", poll_initial_interval_seconds=2, poll_max_interval_seconds=1) + with pytest.raises(ValueError): + sdk.fullscans.stream("org", "scan", poll_timeout_seconds=0.1, poll_initial_interval_seconds=True) + assert not api_fixture.requests + api_fixture.route("GET", "orgs/org%3Fx%3D1/full-scans/scan%3Fx%3D1", ndjson()) + sdk.fullscans.stream("org?x=1", "scan?x=1") + assert api_fixture.requests[0].query == {"cached": ["true"]} + + +@pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True]) +@pytest.mark.parametrize("parameter", ["poll_timeout_seconds", "poll_initial_interval_seconds", "poll_max_interval_seconds"]) +def test_invalid_polling_numbers_fail_before_request(api_fixture, parameter, value): + with pytest.raises(ValueError): + sdk_for(api_fixture).fullscans.stream("org", "scan", **{parameter: value}) + assert not api_fixture.requests + + +@pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True]) +def test_invalid_configured_request_timeout_fails_before_request(api_fixture, value): + sdk = sdk_for(api_fixture) + sdk.api.request_timeout = value + with pytest.raises(ValueError): + sdk.fullscans.stream("org", "scan", poll_timeout_seconds=0.1) + assert not api_fixture.requests diff --git a/tests/unit/test_socket_purl_type.py b/tests/unit/test_socket_purl_type.py index 691384a..1b53c2c 100644 --- a/tests/unit/test_socket_purl_type.py +++ b/tests/unit/test_socket_purl_type.py @@ -21,13 +21,14 @@ import logging import unittest +from socketdev import socketdev from socketdev.fullscans import ( - FullScans, FullScanStreamResponse, SocketArtifact, SocketPURL, SocketPURL_Type, ) +from tests.support.socket_api_fixture import SocketApiFixture, SocketResponse def _artifact_payload(artifact_id: str, purl_type: str) -> dict: @@ -109,32 +110,39 @@ def test_malformed_artifact_is_skipped_not_fatal(self): ) def test_full_scans_stream_skips_artifact_without_id(self): - class Response: - status_code = 200 - text = "\n".join( + fixture = SocketApiFixture() + fixture.start() + try: + body = "\n".join( json.dumps(artifact) for artifact in ( _artifact_payload("good", "npm"), {"type": "npm", "name": "bad", "alerts": []}, ) + ).encode("utf-8") + fixture.route( + "GET", + "orgs/org/full-scans/scan", + SocketResponse(200, None, raw_body=body), ) - - class API: - def do_request(self, **kwargs): - return Response() - - with self.assertLogs("socketdev", level=logging.WARNING) as captured: - response = FullScans(API()).stream("org", "scan", use_types=True) - - self.assertTrue(response.success) - self.assertEqual(list(response.artifacts), ["good"]) - self.assertTrue( - any( - "Skipping artifact without a usable id" in message - for message in captured.output - ), - f"expected a warning about the skipped artifact, got: {captured.output}", - ) + sdk = socketdev(token="synthetic-token", timeout=1) + sdk.api.api_url = fixture.base_url + + with self.assertLogs("socketdev", level=logging.WARNING) as captured: + response = sdk.fullscans.stream("org", "scan", use_types=True) + + self.assertTrue(response.success) + self.assertEqual(list(response.artifacts), ["good"]) + self.assertTrue( + any( + "Skipping artifact without a usable id" in message + for message in captured.output + ), + f"expected a warning about the skipped artifact, got: {captured.output}", + ) + self.assertEqual(fixture.requests[0].query, {"cached": ["true"]}) + finally: + fixture.close() if __name__ == "__main__":