diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 4466166..f6e8403 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -8,6 +8,7 @@ on: branches: [main] paths: - "socketsecurity/**/*.py" + - "socketsecurity/**/*.mjs" - "tests/unit/**/*.py" - "tests/core/**/*.py" - "pyproject.toml" diff --git a/CHANGELOG.md b/CHANGELOG.md index 00d15d1..094b546 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 2.11.0 + +### Changed: download the Coana CLI from Socket instead of npm + +- The Coana CLI is now downloaded from Socket's servers instead of npm, and is now cached in + `$XDG_CACHE_HOME/socket/coana` (`~/.cache/socket/coana` by default). +- The `SOCKET_CLI_COANA_LAUNCHER` environment variable no longer has an effect. + ## 2.10.12 ### Changed: bump pinned @coana-tech/cli to 15.12.8 diff --git a/Dockerfile b/Dockerfile index 06e8b0a..55abd0f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -75,7 +75,7 @@ RUN if [ "$TARGETARCH" = "amd64" ]; then \ fi # Install additional tools -RUN npm install @coana-tech/cli socket -g && \ +RUN npm install socket -g && \ gem install bundler && \ curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && \ . ~/.cargo/env && \ diff --git a/docs/cli-reference.md b/docs/cli-reference.md index f6efd1f..7b045a3 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -338,7 +338,7 @@ If you don't want to provide the Socket API Token every time then you can use th | Parameter | Required | Default | Description | |:---------------------------------|:---------|:--------|:---------------------------------------------------------------------------------------------------------------------------| | `--reach` | False | False | Enable reachability analysis to identify which vulnerable functions are actually called by your code. Creates a full application reachability scan (`scan_type=socket_tier1`). | -| `--reach-version` | False | 15.12.8 | Version of @coana-tech/cli to use. Defaults to the pinned version that ships with this CLI release, so the engine only changes when you upgrade the Socket CLI. Pass `latest` to always use the newest published version (opt-in auto-update), or an explicit version (e.g. `1.2.3`) to pin it. | +| `--reach-version` | False | 15.12.8 | Version of the Coana CLI to use. Defaults to the pinned version that ships with this CLI release, so the engine only changes when you upgrade the Socket CLI. Pass `latest` to always use the newest published version (opt-in auto-update), or an explicit version (e.g. `1.2.3`) to pin it. | | `--reach-analysis-timeout` | False | 10m | Timeout for each reachability analysis run, e.g. `90s`, `10m` or `1h`. Omitted by default, so coana applies its own default (`10m`). Alias: `--reach-timeout` | | `--reach-analysis-memory-limit` | False | 8GB | Memory limit for each reachability analysis run, e.g. `512MB` or `8GB`. Omitted by default, so coana applies its own default (`8GB`). Alias: `--reach-memory-limit` | | `--reach-concurrency` | False | 1 | Control parallel analysis execution (must be >= 1). Omitted by default, so coana applies its own default. | @@ -359,9 +359,7 @@ If you don't want to provide the Socket API Token every time then you can use th **Reachability Analysis Requirements:** The Python CLI verifies the following **up front** (before invoking the analysis engine) and exits with code **3** if any are unmet: -- `npm` - Required (verified up front; ships alongside `npx`) -- `npx` - Required to fetch (on first use) and run `@coana-tech/cli` (the analysis engine) -- `node` - Required to run the engine (used directly by the `npm install` fallback) +- `node` - Required to run the analysis engine (the Coana CLI). The engine needs Node.js 20 or later. - `uv` - Required by the analysis engine - An **Enterprise** Socket organization plan (any `enterprise*` plan, including Enterprise trials) @@ -419,12 +417,7 @@ Sample config files: For CI-specific examples and guidance, see [`ci-cd.md`](ci-cd.md). -The CLI runs a pinned `@coana-tech/cli` version via `npx --yes --force` (the same flags the Socket Node CLI passes for coana); it does **not** auto-update the engine or install it globally. `--yes` skips npx's interactive install prompt so non-interactive/CI runs don't hang. If the `npx` launcher is unavailable or fails before the engine starts, the CLI falls back to `npm install`-ing the pinned version into a temp directory and running it via `node`. Pass `--reach-version latest` to opt into the newest published version. Use `--reach` to enable reachability analysis during a full scan, or add `--only-facts-file` (with `--reach`) to submit only the reachability facts file (`.socket.facts.json`) when creating the full scan. - -The launcher can be tuned via the `SOCKET_CLI_COANA_LAUNCHER` environment variable: -- `auto` (default when unset) — try `npx` first; fall back to `npm install` + `node` if the launcher fails before the engine starts. -- `npm-install` — skip `npx` entirely and always use the `npm install` + `node` path (useful where `npx` is known-broken). -- `npx` — never fall back; surface the `npx` failure directly. +The CLI runs a pinned version of the Coana CLI, the analysis engine; it does **not** auto-update the engine. It downloads the engine from the Socket API on first use and keeps it in `$XDG_CACHE_HOME/socket/coana/v1` (`~/.cache/socket/coana/v1` when `XDG_CACHE_HOME` is unset). Pass `--reach-version latest` to opt into the newest published version. Use `--reach` to enable reachability analysis during a full scan, or add `--only-facts-file` (with `--reach`) to submit only the reachability facts file (`.socket.facts.json`) when creating the full scan. #### Advanced Configuration | Parameter | Required | Default | Description | diff --git a/pyproject.toml b/pyproject.toml index 0b5c1ba..be5d1f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "hatchling.build" [project] name = "socketsecurity" -version = "2.10.12" +version = "2.11.0" requires-python = ">= 3.11" license = {"file" = "LICENSE"} dependencies = [ diff --git a/socketsecurity/__init__.py b/socketsecurity/__init__.py index 025b61a..8850b2e 100644 --- a/socketsecurity/__init__.py +++ b/socketsecurity/__init__.py @@ -1,3 +1,3 @@ __author__ = 'socket.dev' -__version__ = '2.10.12' +__version__ = '2.11.0' USER_AGENT = f'SocketPythonCLI/{__version__}' diff --git a/socketsecurity/config.py b/socketsecurity/config.py index 1f107cf..c13dacd 100644 --- a/socketsecurity/config.py +++ b/socketsecurity/config.py @@ -1069,7 +1069,7 @@ def create_argument_parser() -> argparse.ArgumentParser: "--reach-version", dest="reach_version", metavar="", - help="Version of @coana-tech/cli to use. Defaults to the version pinned to this CLI " + help="Version of the Coana CLI to use. Defaults to the version pinned to this CLI " "release; pass 'latest' to always use the newest published version (opt-in " "auto-update), or an explicit version (e.g. '1.2.3') to pin it." ) diff --git a/socketsecurity/core/cli_client.py b/socketsecurity/core/cli_client.py index 405a744..afa1347 100644 --- a/socketsecurity/core/cli_client.py +++ b/socketsecurity/core/cli_client.py @@ -55,13 +55,26 @@ def request( return response except requests.exceptions.RequestException as e: - logger.error(f"API request failed: {str(e)}") + server_message = self._server_error_message(e.response) + detail = f"{e}: {server_message}" if server_message else str(e) + logger.error(f"API request failed: {detail}") # Carry the status forward. Callers that need to react to a specific # code -- the GitLab auth fallback to the other token scheme, and # APIFailure.is_transient_error -- have no other way to recover it # once the requests exception has been translated. status_code = e.response.status_code if e.response is not None else None - raise APIFailure(f"Request failed: {str(e)}", status_code=status_code) from e + raise APIFailure(f"Request failed: {detail}", status_code=status_code) from e + + @staticmethod + def _server_error_message(response: Optional[requests.Response]) -> Optional[str]: + """The explanation the Socket API gives in its ``{"error": {"message": ...}}`` body.""" + if response is None: + return None + try: + message = response.json()["error"]["message"] + except (ValueError, KeyError, TypeError): + return None + return message if isinstance(message, str) and message else None def post_telemetry_events(self, org_slug: str, events: List[Dict]) -> None: """Post telemetry events one at a time to the v0 telemetry API. Fire-and-forget — logs errors but never raises.""" diff --git a/socketsecurity/core/tools/coana-cache.mjs b/socketsecurity/core/tools/coana-cache.mjs new file mode 100644 index 0000000..52423b7 --- /dev/null +++ b/socketsecurity/core/tools/coana-cache.mjs @@ -0,0 +1,198 @@ +// Generated from coana-package-manager's packages/coana-cache. Don't edit it here. +// See "Launching Coana" in packages/cli/docs/gcs-releases.md there. + +// src/cli.ts +import { cp, readdir as readdir2, rename as rename2, rm as rm3 } from "node:fs/promises"; +import { join as join3 } from "node:path"; + +// src/cache.ts +import { mkdir as mkdir2, open, readdir, readFile, rm as rm2, stat, utimes } from "node:fs/promises"; +import { homedir } from "node:os"; +import { dirname as dirname2, isAbsolute, join as join2, relative } from "node:path"; + +// ../utils/src/atomic-fs.ts +import { randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { chmod, mkdir, mkdtemp, rename, rm, writeFile } from "node:fs/promises"; +import { basename, dirname, join } from "node:path"; +async function populateDirAtomically(targetDir, populate) { + if (existsSync(targetDir)) return; + const parentDir = dirname(targetDir); + await mkdir(parentDir, { recursive: true }); + const tmpDir = await mkdtemp(join(parentDir, `.${basename(targetDir)}-`)); + try { + await populate(tmpDir); + await chmod(tmpDir, 493); + await rename(tmpDir, targetDir).catch((e) => { + if (!existsSync(targetDir)) throw e; + }); + } finally { + await rm(tmpDir, { recursive: true, force: true }).catch(() => { + }); + } +} +async function writeFileAtomically(path, data) { + const tmpPath = `${path}.${randomUUID()}.tmp`; + try { + await writeFile(tmpPath, data); + await rename(tmpPath, path); + } finally { + await rm(tmpPath, { force: true }).catch(() => { + }); + } +} + +// src/cache.ts +var SHA256 = /^[0-9a-f]{64}$/; +var VERSION = /^[0-9A-Za-z][0-9A-Za-z.+-]*$/; +var COMPONENT = /^[A-Za-z][0-9A-Za-z-]*$/; +var TEMP_DIR = /^\.[0-9a-f]{64}-/; +var ABANDONED_TEMP_DIR_MS = 24 * 60 * 6e4; +function cacheRoot() { + const xdgCacheHome = process.env.XDG_CACHE_HOME; + const cacheHome = xdgCacheHome && isAbsolute(xdgCacheHome) ? xdgCacheHome : join2(homedir(), ".cache"); + return join2(cacheHome, "socket", "coana", "v1"); +} +var OUT_OF_SPACE_CODES = /* @__PURE__ */ new Set(["ENOSPC", "EDQUOT"]); +var NOT_WRITABLE_CODES = /* @__PURE__ */ new Set(["EACCES", "EPERM", "EROFS", ...OUT_OF_SPACE_CODES]); +var CacheNotWritable = class _CacheNotWritable extends Error { + constructor(outOfSpace, cause) { + super( + outOfSpace ? `The disk holding the Coana cache in ${cacheRoot()} is out of space (${cause.message}). Free some up, or set XDG_CACHE_HOME to a directory on another disk.` : `Could not write to the Coana cache in ${cacheRoot()} (${cause.message}). Set XDG_CACHE_HOME to a writable directory.`, + { cause } + ); + } + /** `e` as a {@link CacheNotWritable} when it is such a failure of a path in the cache. */ + static from(e) { + if (e instanceof _CacheNotWritable) return e; + if (!(e instanceof Error)) return void 0; + const { code = "", path: source, dest } = e; + if (!NOT_WRITABLE_CODES.has(code)) return void 0; + const outOfSpace = OUT_OF_SPACE_CODES.has(code); + const paths = (outOfSpace && dest !== void 0 ? [dest] : [source, dest]).filter((p) => p !== void 0); + const ofCache = paths.length === 0 ? outOfSpace : paths.some((path) => within(cacheRoot(), path) || within(path, cacheRoot())); + return ofCache ? new _CacheNotWritable(outOfSpace, e) : void 0; + function within(dir, path) { + const fromDir = relative(dir, path); + return !fromDir.startsWith("..") && !isAbsolute(fromDir); + } + } +}; +var InvalidCacheKey = class extends Error { +}; +async function writing(f) { + try { + return await f(); + } catch (e) { + throw CacheNotWritable.from(e) ?? e; + } +} +function objectDir(sha256) { + return join2(cacheRoot(), "objects", sha256); +} +function versionSegment(version) { + if (!VERSION.test(version)) throw new InvalidCacheKey(`Invalid Coana version: ${version}`); + return version; +} +function indexFile(version, component) { + if (!COMPONENT.test(component)) throw new InvalidCacheKey(`Invalid component name: ${component}`); + return join2(cacheRoot(), "versions", versionSegment(version), component); +} +async function lookup(version, component) { + const sha256 = (await readFile(indexFile(version, component), "utf8").catch(() => "")).trim(); + if (!SHA256.test(sha256)) return void 0; + const dir = objectDir(sha256); + const isDirectory = await stat(dir).then( + (s) => s.isDirectory(), + () => false + ); + return isDirectory ? dir : void 0; +} +async function install(version, component, sha256, populate) { + if (!SHA256.test(sha256)) throw new InvalidCacheKey(`Invalid SHA-256: ${sha256}`); + const index = indexFile(version, component); + const target = objectDir(sha256); + await writing(() => mkdir2(dirname2(index), { recursive: true })); + await writing( + () => populateDirAtomically(target, (dir) => { + void sweepAbandonedTempDirs(); + return populate(dir); + }) + ); + await writing(() => writeFileAtomically(index, sha256)); + return target; +} +var sweeps = /* @__PURE__ */ new Map(); +function sweepAbandonedTempDirs() { + const objects = join2(cacheRoot(), "objects"); + let sweep = sweeps.get(objects); + if (sweep === void 0) { + sweep = removeAbandoned(); + sweeps.set(objects, sweep); + } + return sweep; + async function removeAbandoned() { + const names = await readdir(objects).catch(() => []); + await Promise.all( + names.filter((name) => TEMP_DIR.test(name)).map(async (name) => { + const dir = join2(objects, name); + const age = await ageMs(dir); + if (age !== void 0 && age > ABANDONED_TEMP_DIR_MS) + await rm2(dir, { recursive: true, force: true }).catch(() => { + }); + }) + ); + } +} +async function ageMs(path) { + return stat(path).then( + (s) => Date.now() - s.ctimeMs, + () => void 0 + ); +} + +// src/cli.ts +var USAGE = `usage: coana-cache.mjs lookup + coana-cache.mjs install `; +var NOT_CACHED = 4; +async function main(args) { + const [command, ...rest] = args; + if (command === "lookup" && rest.length === 2) { + const [version, component] = rest; + const dir = await lookup(version, component); + if (dir === void 0) return NOT_CACHED; + process.stdout.write(`${dir} +`); + return 0; + } + if (command === "install" && rest.length === 4) { + const [version, component, sha256, extracted] = rest; + const dir = await install(version, component, sha256, (tmp) => moveInto(extracted, tmp)); + await rm3(extracted, { recursive: true, force: true }).catch(() => { + }); + process.stdout.write(`${dir} +`); + return 0; + } + process.stderr.write(`${USAGE} +`); + return 2; + async function moveInto(extractedDir, dir) { + for (const name of await readdir2(extractedDir)) { + const from = join3(extractedDir, name); + const to = join3(dir, name); + try { + await rename2(from, to); + } catch (e) { + if (e.code !== "EXDEV") throw e; + await cp(from, to, { recursive: true, verbatimSymlinks: true }); + } + } + } +} +process.exitCode = await main(process.argv.slice(2)).catch((e) => { + process.stderr.write(`${e instanceof Error ? e.message : String(e)} +`); + if (e instanceof InvalidCacheKey) return 2; + return e instanceof CacheNotWritable ? 3 : 1; +}); diff --git a/socketsecurity/core/tools/reachability.py b/socketsecurity/core/tools/reachability.py index 2343683..c03cea6 100644 --- a/socketsecurity/core/tools/reachability.py +++ b/socketsecurity/core/tools/reachability.py @@ -1,39 +1,82 @@ -import atexit +import hashlib import json import logging import os import pathlib import platform -import shutil +import re import subprocess import sys +import tarfile import tempfile -from typing import Any, Dict, Final, List, Optional +from dataclasses import dataclass +from typing import Any, Dict, Final, List, Mapping, Optional -from socketdev import socketdev +import requests -from socketsecurity import __version__ +from socketsecurity import USER_AGENT, __version__ +from socketsecurity.core.cli_client import CliClient log = logging.getLogger(__name__) -# Pinned @coana-tech/cli version. Bumped deliberately per Python CLI release so the +# Pinned Coana CLI version. Bumped deliberately per Python CLI release so the # reachability engine version only changes through a standard pip upgrade (advance notice). # Pass --reach-version latest to opt into the newest published version instead. DEFAULT_COANA_CLI_VERSION: Final = "15.12.8" -# Resolved @coana-tech/cli script paths from the npm-install fallback, keyed by version. -# Lives for the process lifetime so repeated fallback invocations install only once -# (mirrors the Node CLI's installedCoanaScriptPathsByVersion). -_INSTALLED_COANA_SCRIPT_PATHS: Dict[str, str] = {} +# Vendored from coana-package-manager, where it is generated; the only code that may read or +# write the cache Coana shares with the Socket CLIs. See "Launching Coana" in that repo's +# packages/cli/docs/gcs-releases.md. +_COANA_CACHE_SCRIPT: Final = pathlib.Path(__file__).with_name("coana-cache.mjs") +_COANA_COMPONENT: Final = "cli" +_NOT_CACHED_EXIT_CODE: Final = 4 -# Temp dirs created by the npm-install fallback, removed at process exit. -_COANA_INSTALL_DIRS: List[str] = [] +_EXACT_VERSION: Final = re.compile(r"\d+\.\d+\.\d+(-[0-9A-Za-z.-]+)?") +_SHA256: Final = re.compile(r"[0-9a-f]{64}") +# Seconds to connect to the download host, and seconds it may go without sending a byte. +_DOWNLOAD_TIMEOUT: Final = (30, 60) -@atexit.register -def _cleanup_coana_install_dirs() -> None: - for install_dir in _COANA_INSTALL_DIRS: - shutil.rmtree(install_dir, ignore_errors=True) +@dataclass(frozen=True) +class _CoanaComponent: + sha256: str + size: int + url: str + + +@dataclass(frozen=True) +class _CoanaArtifacts: + version: str + cli: _CoanaComponent + + @classmethod + def parse(cls, value: object) -> Optional["_CoanaArtifacts"]: + """Parse the Socket API's response to a request for the ``cli`` component.""" + if not isinstance(value, dict): + return None + version = value.get("version") + components = value.get("components") + cli = components.get(_COANA_COMPONENT) if isinstance(components, dict) else None + if not isinstance(cli, dict): + return None + sha256, size, url = cli.get("sha256"), cli.get("size"), cli.get("url") + if ( + not isinstance(version, str) + or not _EXACT_VERSION.fullmatch(version) + or not isinstance(sha256, str) + or not _SHA256.fullmatch(sha256) + # bool is an int subclass. + or type(size) is not int + or size <= 0 + or not isinstance(url, str) + or not url.startswith("https://") + ): + return None + return cls(version=version, cli=_CoanaComponent(sha256=sha256, size=size, url=url)) + + +class _DownloadRefused(Exception): + """The download host refused a signed URL with a 4xx, e.g. because it expired.""" def _build_caller_user_agent() -> str: @@ -50,30 +93,9 @@ def _build_caller_user_agent() -> str: class ReachabilityAnalyzer: - def __init__(self, sdk: socketdev, api_token: str): - self.sdk = sdk - self.api_token = api_token - - def _resolve_coana_package_spec(self, version: Optional[str] = None) -> str: - """ - Resolve the @coana-tech/cli package spec to run (e.g. '@coana-tech/cli@15.12.8'). - - Args: - version: Coana CLI version to use. - - None: the pinned ``DEFAULT_COANA_CLI_VERSION`` (no auto-update). - - 'latest': always the newest published version (opt-in to auto-update). - - '': that exact version. + def __init__(self, client: CliClient): + self.client = client - Returns: - str: The package specifier to use with npx (e.g. '@coana-tech/cli@15.12.8'). - """ - return f"@coana-tech/cli@{self._resolve_coana_version(version)}" - - def _resolve_coana_version(self, version: Optional[str] = None) -> str: - """Resolve the effective @coana-tech/cli version string (see _resolve_coana_package_spec).""" - return (version or DEFAULT_COANA_CLI_VERSION).strip() - - def run_reachability_analysis( self, org_slug: str, @@ -94,7 +116,6 @@ def run_reachability_analysis( version: Optional[str] = None, concurrency: Optional[int] = None, additional_params: Optional[List[str]] = None, - allow_unverified: bool = False, enable_debug: bool = False, use_only_pregenerated_sboms: bool = False, continue_on_analysis_errors: bool = False, @@ -125,20 +146,17 @@ def run_reachability_analysis( detailed_analysis_log_file: Print detailed analysis log file path repo_name: Repository name branch_name: Branch name - version: @coana-tech/cli version to use. None uses the pinned + version: Coana CLI version spec to use. None uses the pinned DEFAULT_COANA_CLI_VERSION (no auto-update); 'latest' opts into the newest published version; '' pins an explicit version. concurrency: Concurrency level for analysis (must be >= 1) additional_params: Additional parameters to pass to coana CLI - allow_unverified: Disable SSL certificate verification (sets NODE_TLS_REJECT_UNAUTHORIZED=0) enable_debug: Enable debug mode (passes -d flag to coana CLI) use_only_pregenerated_sboms: Use only pre-generated CDX and SPDX files for the scan Returns: Dict containing scan_id and report_path """ - # Build the coana CLI arguments (everything after the package spec). The launcher - # (npx, or the npm-install + node fallback) is chosen in _spawn_coana() below. coana_args = ["run", "."] # Add required arguments @@ -215,11 +233,13 @@ def run_reachability_analysis( coana_args.extend(additional_params) # Set up environment variables - env = os.environ.copy() - + env = self._sanitize_coana_env(os.environ) + # Required environment variables for Coana CLI env["SOCKET_ORG_SLUG"] = org_slug - env["SOCKET_CLI_API_TOKEN"] = self.api_token + env["SOCKET_CLI_API_TOKEN"] = self.client.config.api_key + # Coana expects the base URL to include the trailing slash after /v0. + env["SOCKET_CLI_API_BASE_URL"] = self.client.config.api_url.rstrip("/") + "/" # Identify the calling CLI to the coana tool / backend (parity with the Node CLI). env["SOCKET_CLI_VERSION"] = __version__ @@ -239,8 +259,7 @@ def run_reachability_analysis( if branch_name: env["SOCKET_BRANCH_NAME"] = branch_name - # Set NODE_TLS_REJECT_UNAUTHORIZED=0 if allow_unverified is True - if allow_unverified: + if self.client.config.allow_unverified_ssl: env["NODE_TLS_REJECT_UNAUTHORIZED"] = "0" # Execute coana @@ -248,9 +267,16 @@ def run_reachability_analysis( log.debug(f"Environment: SOCKET_ORG_SLUG={org_slug}, SOCKET_REPO_NAME={repo_name or 'not set'}, SOCKET_BRANCH_NAME={branch_name or 'not set'}") try: - # Prefer npx (with caching disabled); fall back to `npm install` + `node` - # if the npx launcher fails before coana starts (parity with the Node CLI). - returncode = self._spawn_coana(coana_args, version, env, target_directory) + cli_dir = self._get_coana_cli_dir((version or DEFAULT_COANA_CLI_VERSION).strip(), org_slug, env) + cmd = ["node", os.path.join(cli_dir, "cli-wrapper.mjs"), *coana_args] + log.debug(f"Reachability command: {' '.join(cmd)}") + returncode = subprocess.run( + cmd, + env=env, + cwd=target_directory, + stdout=sys.stderr, # Send stdout to stderr so the user sees it + stderr=sys.stderr, + ).returncode if returncode != 0: log.error(f"Reachability analysis failed with exit code {returncode}") @@ -274,9 +300,12 @@ def run_reachability_analysis( raise Exception(f"Failed to run reachability analysis: {str(e)}") @staticmethod - def _sanitize_coana_env(env: Dict[str, str]) -> Dict[str, str]: + def _sanitize_coana_env(env: Mapping[str, str]) -> Dict[str, str]: """Drop npm-injected ``npm_package_*`` vars before spawning coana. + TODO: probably unnecessary. The E2BIG came from npm exec re-adding these vars, and + coana no longer runs through npm, so its env is only slightly larger than our own. + npm/pnpm/yarn populate one env var per leaf of the cwd's package.json (``npm_package_dependencies_*`` etc.). In large monorepos this can be tens of KB and push argv+env past the OS ARG_MAX, making the spawn fail with E2BIG before @@ -285,187 +314,105 @@ def _sanitize_coana_env(env: Dict[str, str]) -> Dict[str, str]: """ return {k: v for k, v in env.items() if not k.startswith("npm_package_")} - @staticmethod - def _npx_launcher_failed_before_coana(returncode: int) -> bool: - """Heuristic: did npx fail *before* coana started (so retrying is worthwhile)? + def _get_coana_cli_dir(self, version_spec: str, org_slug: str, env: Dict[str, str]) -> str: + """The directory of the Coana CLI's ``cli`` component for ``version_spec``. - We stream coana's output (no capture), so we classify by exit code alone, like the - Node CLI does with inherited stdio: signal kills (negative codes) and codes >= 128 - are conventionally launcher/signal failures -> retry. Small positive codes (1..127) - are ambiguous (coana's own exit codes are small ints), so we do NOT retry. + ``version_spec`` is an exact version, a range or ``latest``. On a cache miss the + component is downloaded from the Socket API and installed into the cache. ``env`` + reaches the cache script, which reads ``XDG_CACHE_HOME`` from it. """ - return returncode < 0 or returncode >= 128 - - @staticmethod - def _resolve_coana_launcher_mode() -> str: - """Resolve the coana launcher mode: ``auto``, ``npx``, or ``npm-install``. - ``SOCKET_CLI_COANA_LAUNCHER`` wins when set to a recognized value; an unrecognized - value warns and behaves as ``auto``. Only when it is unset/empty do the legacy vars - apply: ``SOCKET_CLI_COANA_FORCE_NPM_INSTALL`` -> ``npm-install``, else - ``SOCKET_CLI_COANA_DISABLE_NPM_FALLBACK`` -> ``npx``. Mirrors the Node CLI. - """ - raw = os.environ.get("SOCKET_CLI_COANA_LAUNCHER", "") - mode = raw.strip().lower() - if mode in ("auto", "npx", "npm-install"): - return mode - if mode: - log.warning( - f'Ignoring unrecognized SOCKET_CLI_COANA_LAUNCHER value "{raw}"; ' - 'expected "auto", "npm-install", or "npx".' + def run_coana_cache(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["node", str(_COANA_CACHE_SCRIPT), *args], env=env, capture_output=True, text=True ) - return "auto" - if os.environ.get("SOCKET_CLI_COANA_FORCE_NPM_INSTALL"): - return "npm-install" - if os.environ.get("SOCKET_CLI_COANA_DISABLE_NPM_FALLBACK"): - return "npx" - return "auto" - - def _spawn_coana( - self, - coana_args: List[str], - version: Optional[str], - env: Dict[str, str], - cwd: str, - ) -> int: - """Run coana for the given args, returning the process exit code. - - We run a pinned, versioned spec via npx and intentionally do NOT ``npm install -g``: - that would silently auto-update the engine on every run and mutate the user's global - install. The pinned version rides with the Python CLI release instead (see - ``DEFAULT_COANA_CLI_VERSION``). - - Primary path: ``npx --yes --force @coana-tech/cli@ ...`` — the exact flags the - Socket Node CLI passes for coana (``--yes`` skips npx's interactive install prompt so - CI runs don't hang). - - Fallback path: if npx is missing, or its launcher dies before coana starts, install - @coana-tech/cli into a temp dir via ``npm install`` and run it directly via ``node``. - Tune with ``SOCKET_CLI_COANA_LAUNCHER``: ``auto`` (default; npx with the npm-install - fallback), ``npm-install`` (skip npx, always use the fallback path), or ``npx`` - (never fall back). - """ - effective_version = self._resolve_coana_version(version) - coana_env = self._sanitize_coana_env(env) - launcher_mode = self._resolve_coana_launcher_mode() - - if launcher_mode == "npm-install": - return self._spawn_coana_via_npm_install(coana_args, effective_version, coana_env, cwd) - package_spec = f"@coana-tech/cli@{effective_version}" - npx_cmd = ["npx", "--yes", "--force", package_spec, *coana_args] - log.debug(f"Reachability command: {' '.join(npx_cmd)}") - try: - result = subprocess.run( - npx_cmd, - env=coana_env, - cwd=cwd, - stdout=sys.stderr, # Send stdout to stderr so the user sees it - stderr=sys.stderr, - ) - except FileNotFoundError: - # npx is not on PATH: the launcher provably never started coana. - if launcher_mode == "npx": - raise - log.warning("npx not found on PATH; retrying reachability analysis via `npm install` + `node`.") - return self._spawn_coana_via_npm_install(coana_args, effective_version, coana_env, cwd) - - if result.returncode == 0: - return 0 - - if launcher_mode != "npx" and self._npx_launcher_failed_before_coana(result.returncode): - log.warning( - f"npx launcher failed (exit {result.returncode}) before coana started; " - "retrying reachability analysis via `npm install` + `node`." + def printed_dir(result: subprocess.CompletedProcess[str]) -> str: + """The directory the cache script printed, which is all it prints to stdout.""" + if result.returncode != 0: + detail = result.stderr.strip() or f"{_COANA_CACHE_SCRIPT.name} exited with {result.returncode}" + raise Exception(f"The Coana cache failed: {detail}") + return result.stdout.strip() + + def lookup(version: str) -> Optional[str]: + result = run_coana_cache("lookup", version, _COANA_COMPONENT) + return None if result.returncode == _NOT_CACHED_EXIT_CODE else printed_dir(result) + + def request_artifacts(spec: str) -> _CoanaArtifacts: + # The API refuses specs it can't serve, with an explanation that APIFailure carries. + response = self.client.request( + f"orgs/{org_slug}/coana-cli/artifacts", + method="POST", + payload=json.dumps({"version": spec, "components": [_COANA_COMPONENT]}), ) - return self._spawn_coana_via_npm_install(coana_args, effective_version, coana_env, cwd) - - return result.returncode - - def _spawn_coana_via_npm_install( - self, - coana_args: List[str], - version: str, - env: Dict[str, str], - cwd: str, - ) -> int: - """Fallback launcher: ``npm install`` @coana-tech/cli into a temp dir, run via ``node``. - - Used when npx is unavailable or its launcher fails before coana boots. Mirrors the - Node CLI's npm-install fallback. Returns coana's exit code; raises if the install - itself fails or if ``node`` is unavailable. - """ - script_path = self._install_coana_to_tmpdir(version, env) - node_cmd = self._build_coana_node_cmd(script_path, coana_args) - log.debug(f"Reachability fallback command: {' '.join(node_cmd)}") - try: - result = subprocess.run(node_cmd, env=env, cwd=cwd, stdout=sys.stderr, stderr=sys.stderr) - except FileNotFoundError as e: - # The fallback exists for broken-launcher environments, but it still needs node. - raise Exception( - "`node` was not found on PATH; it is required to run the reachability engine " - "via the npm-install fallback." - ) from e - return result.returncode - - def _install_coana_to_tmpdir(self, version: str, env: Dict[str, str]) -> str: - """``npm install`` @coana-tech/cli@ into a temp dir; return its executable JS path. - - Caches the resolved path per version for the process lifetime so repeated fallback - invocations install only once (mirrors the Node CLI's installCoanaToTmpdir). Raises if - the install fails. - """ - cached = _INSTALLED_COANA_SCRIPT_PATHS.get(version) - if cached and os.path.exists(cached): + try: + body = response.json() + except ValueError: + body = None + if (artifacts := _CoanaArtifacts.parse(body)) is None: + raise Exception(f"Unexpected response from the Socket API to a request for the Coana CLI {spec}") + return artifacts + + def download(component: _CoanaComponent, dest: pathlib.Path) -> None: + with requests.get( + component.url, + headers={"User-Agent": USER_AGENT}, + stream=True, + timeout=_DOWNLOAD_TIMEOUT, + verify=not self.client.config.allow_unverified_ssl, + ) as response: + if 400 <= response.status_code < 500: + raise _DownloadRefused( + f"The download host refused the Coana CLI URL with status {response.status_code}" + ) + response.raise_for_status() + digest = hashlib.sha256() + received = 0 + with dest.open("wb") as f: + # The checksum covers the bytes as stored, so skip any Content-Encoding. + for chunk in response.raw.stream(1 << 16, decode_content=False): + received += len(chunk) + if received > component.size: + raise Exception( + f"The Coana CLI tarball is larger than the expected {component.size} bytes" + ) + digest.update(chunk) + f.write(chunk) + sha256 = digest.hexdigest() + if received != component.size or sha256 != component.sha256: + raise Exception( + "The Coana CLI tarball does not match its checksum " + f"(expected {component.size} bytes with sha256 {component.sha256}, " + f"got {received} bytes with sha256 {sha256})" + ) + + def regular_files_only(member: tarfile.TarInfo, path: str) -> Optional[tarfile.TarInfo]: + return tarfile.data_filter(member, path) if member.isfile() or member.isdir() else None + + if _EXACT_VERSION.fullmatch(version_spec) and (cached := lookup(version_spec)): + return cached + artifacts = request_artifacts(version_spec) + if artifacts.version != version_spec and (cached := lookup(artifacts.version)): return cached - install_dir = tempfile.mkdtemp(prefix="socket-coana-") - _COANA_INSTALL_DIRS.append(install_dir) - npm_cmd = [ - "npm", "install", - "--no-save", "--no-package-lock", "--no-audit", "--no-fund", - "--prefix", install_dir, - f"@coana-tech/cli@{version}", - ] - log.info("Installing reachability analysis engine via npm fallback...") - log.debug(f"npm install fallback command: {' '.join(npm_cmd)}") - install = subprocess.run(npm_cmd, env=env, stdout=sys.stderr, stderr=sys.stderr) - if install.returncode != 0: - raise Exception( - f"npm install fallback for @coana-tech/cli@{version} failed with exit code {install.returncode}" - ) - - script_path = self._resolve_coana_bin(install_dir) - _INSTALLED_COANA_SCRIPT_PATHS[version] = script_path - return script_path - - @staticmethod - def _resolve_coana_bin(install_dir: str) -> str: - """Resolve @coana-tech/cli's executable JS from its installed package.json ``bin`` field.""" - package_json_path = os.path.join( - install_dir, "node_modules", "@coana-tech", "cli", "package.json" - ) - with open(package_json_path, "r") as f: - pkg = json.load(f) - bin_field = pkg.get("bin") - relative_bin = None - if isinstance(bin_field, str): - relative_bin = bin_field - elif isinstance(bin_field, dict): - # Prefer an entry named "coana"; otherwise take the first. - relative_bin = bin_field.get("coana") or next(iter(bin_field.values()), None) - if not relative_bin: - raise Exception( - f"@coana-tech/cli package.json at {package_json_path} is missing a usable bin entry" + log.info(f"Downloading the Coana CLI {artifacts.version}...") + with tempfile.TemporaryDirectory(prefix="socket-coana-cli-") as tmp: + tarball = pathlib.Path(tmp, "cli.tar.gz") + try: + download(artifacts.cli, tarball) + except _DownloadRefused: + # Pin the new request to the resolved version, which a range is not. + artifacts = request_artifacts(artifacts.version) + download(artifacts.cli, tarball) + extract_dir = pathlib.Path(tmp, "cli") + with tarfile.open(tarball, "r:gz") as archive: + # Without extraction filters, the checked sha256 vouches for the tarball. + archive.extractall( + extract_dir, **({"filter": regular_files_only} if hasattr(tarfile, "data_filter") else {}) + ) + return printed_dir( + run_coana_cache("install", artifacts.version, _COANA_COMPONENT, artifacts.cli.sha256, str(extract_dir)) ) - return os.path.abspath(os.path.join(os.path.dirname(package_json_path), relative_bin)) - - @staticmethod - def _build_coana_node_cmd(script_path: str, coana_args: List[str]) -> List[str]: - """Run a .js/.mjs entry via ``node``; invoke a native binary directly (Node CLI parity).""" - if script_path.endswith(".js") or script_path.endswith(".mjs"): - return ["node", script_path, *coana_args] - return [script_path, *coana_args] def _extract_scan_id(self, facts_file_path: str) -> Optional[str]: """ diff --git a/socketsecurity/socketcli.py b/socketsecurity/socketcli.py index 704f1fc..d73e6ce 100644 --- a/socketsecurity/socketcli.py +++ b/socketsecurity/socketcli.py @@ -345,7 +345,7 @@ def main_code(): # Check for required dependencies if reachability analysis is enabled if config.reach: log.info("Reachability analysis enabled, checking for required dependencies...") - required_deps = ["npm", "node", "uv", "npx"] + required_deps = ["node", "uv"] missing_deps = [] found_deps = [] @@ -511,7 +511,7 @@ def main_code(): log.info(f"Manifest upload successful, tar hash: {tar_hash}") # Initialize and run reachability analyzer - analyzer = ReachabilityAnalyzer(sdk, config.api_token) + analyzer = ReachabilityAnalyzer(client) # Determine output path output_path = config.reach_output_file or ".socket.facts.json" @@ -544,7 +544,6 @@ def main_code(): version=config.reach_version, concurrency=config.reach_concurrency, additional_params=config.reach_additional_params, - allow_unverified=config.allow_unverified, enable_debug=config.enable_debug, use_only_pregenerated_sboms=config.reach_use_only_pregenerated_sboms, continue_on_analysis_errors=config.reach_continue_on_analysis_errors, diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 5efa6fb..1965d59 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -65,6 +65,25 @@ def test_request_handles_api_error(): with pytest.raises(APIFailure): client.request("test/path") +def test_request_failure_carries_the_api_error_message(client): + response = requests.Response() + response.status_code = 404 + response._content = b'{"error": {"message": "need an older Socket CLI"}}' + + with patch('requests.request', return_value=response): + with pytest.raises(APIFailure, match="404 Client Error.*: need an older Socket CLI") as exc_info: + client.request("test/path") + assert exc_info.value.status_code == 404 + +def test_request_failure_without_an_api_error_message(client): + response = requests.Response() + response.status_code = 502 + response._content = b'Bad Gateway' + + with patch('requests.request', return_value=response): + with pytest.raises(APIFailure, match=r"502 Server Error: None for url: None$"): + client.request("test/path") + def test_request_uses_custom_headers(client): """Test that custom headers override defaults""" custom_headers = {"Authorization": "Bearer token", "Custom": "Value"} diff --git a/tests/unit/test_reachability.py b/tests/unit/test_reachability.py index 71a20dc..e2c855d 100644 --- a/tests/unit/test_reachability.py +++ b/tests/unit/test_reachability.py @@ -1,57 +1,72 @@ -"""Tests for the reachability coana-CLI command/env construction (Node alignment). +"""Tests for launching the Coana CLI from ReachabilityAnalyzer. -These cover the arg-builder, the npx launcher (caching disabled via --yes --force), the -npm-install + node fallback, and the environment wiring in -``socketsecurity.core.tools.reachability.ReachabilityAnalyzer`` without actually invoking -npx/npm/node/coana: ``subprocess.run`` (and, for the fallback, ``tempfile.mkdtemp`` / -``_resolve_coana_bin``) are mocked. +A fake cli-wrapper.mjs is installed into a temporary cache with the real coana-cache.mjs, so +nothing is mocked. The fake records the args, cwd and environment it was run with. """ -from unittest.mock import MagicMock +import json +import os +import subprocess import pytest from socketsecurity import __version__ +from socketsecurity.core.cli_client import CliClient +from socketsecurity.core.socket_config import SocketConfig from socketsecurity.core.tools import reachability from socketsecurity.core.tools.reachability import ( - DEFAULT_COANA_CLI_VERSION, ReachabilityAnalyzer, _build_caller_user_agent, ) +VERSION = "1.2.3" -@pytest.fixture -def analyzer(): - return ReachabilityAnalyzer(MagicMock(), "test-api-token") - - -@pytest.fixture(autouse=True) -def _clear_coana_install_cache(): - """The npm-install fallback caches resolved script paths in a module-level dict; isolate tests.""" - reachability._INSTALLED_COANA_SCRIPT_PATHS.clear() - try: - yield - finally: - reachability._INSTALLED_COANA_SCRIPT_PATHS.clear() - - -def _spawn_mock(analyzer, mocker, returncode=0, **kwargs): - """Run run_reachability_analysis with subprocess.run mocked to a fixed exit code. +# Records what it was run with in $FAKE_COANA_RUN, or fails when passed --fail. +FAKE_CLI_WRAPPER = """ +import { writeFileSync } from 'node:fs' +const args = process.argv.slice(2) +if (args.includes('--fail')) process.exit(1) +writeFileSync(process.env.FAKE_COANA_RUN, JSON.stringify({ args, cwd: process.cwd(), env: process.env })) +""" - Uses the real resolver / _spawn_coana; returns the run mock for inspection. - """ - mocker.patch.object(analyzer, "_extract_scan_id", return_value="scan-123") - completed = MagicMock() - completed.returncode = returncode - run_mock = mocker.patch.object(reachability.subprocess, "run", return_value=completed) - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123", **kwargs) - return run_mock +@pytest.fixture(scope="module") +def cache_home(tmp_path_factory): + tmp = tmp_path_factory.mktemp("coana") + extracted = tmp / "extracted" + extracted.mkdir() + (extracted / "cli-wrapper.mjs").write_text(FAKE_CLI_WRAPPER) + subprocess.run( + ["node", str(reachability._COANA_CACHE_SCRIPT), "install", VERSION, "cli", "a" * 64, str(extracted)], + env={**os.environ, "XDG_CACHE_HOME": str(tmp / "cache")}, + check=True, + capture_output=True, + ) + return tmp / "cache" -def _run(analyzer, mocker, **kwargs): - """Invoke run_reachability_analysis on the happy npx path; return (npx argv, env).""" - run_mock = _spawn_mock(analyzer, mocker, **kwargs) - return run_mock.call_args.args[0], run_mock.call_args.kwargs["env"] +@pytest.fixture +def fake_run_path(tmp_path, monkeypatch, cache_home): + monkeypatch.setenv("XDG_CACHE_HOME", str(cache_home)) + path = tmp_path / "run.json" + monkeypatch.setenv("FAKE_COANA_RUN", str(path)) + return path + + +def _analyzer(api_url="https://api.socket.dev/v0"): + return ReachabilityAnalyzer(CliClient(SocketConfig(api_key="test-api-token", api_url=api_url))) + + +def _run(fake_run_path, analyzer=None, **kwargs): + """Run the fake Coana; return the args, cwd and env it recorded.""" + (analyzer or _analyzer()).run_reachability_analysis( + org_slug="my-org", + target_directory=str(fake_run_path.parent), + tar_hash="tar-hash-abc123", + output_path=str(fake_run_path.parent / ".socket.facts.json"), + version=VERSION, + **kwargs, + ) + return json.loads(fake_run_path.read_text()) def test_build_caller_user_agent_shape(): @@ -62,386 +77,86 @@ def test_build_caller_user_agent_shape(): assert "/" in parts[2] # platform/arch -def test_reach_debug_appends_debug_long_flag(analyzer, mocker): +def test_runs_the_cached_cli_wrapper_with_the_args_and_cwd(fake_run_path): + run = _run(fake_run_path) + assert run["args"][:2] == ["run", "."] + assert "--manifests-tar-hash" in run["args"] + assert run["cwd"] == str(fake_run_path.parent) + + +def test_reach_debug_appends_debug_long_flag(fake_run_path): """G9: --reach-debug -> coana --debug; does not emit the global -d.""" - cmd, _ = _run(analyzer, mocker, reach_debug=True) - assert "--debug" in cmd - assert "-d" not in cmd + args = _run(fake_run_path, reach_debug=True)["args"] + assert "--debug" in args + assert "-d" not in args -def test_enable_debug_still_emits_short_d(analyzer, mocker): +def test_enable_debug_still_emits_short_d(fake_run_path): """G9: existing global --enable-debug -> -d behavior is unchanged.""" - cmd, _ = _run(analyzer, mocker, enable_debug=True) - assert "-d" in cmd - assert "--debug" not in cmd + args = _run(fake_run_path, enable_debug=True)["args"] + assert "-d" in args + assert "--debug" not in args -def test_disable_external_tool_checks(analyzer, mocker): +def test_disable_external_tool_checks(fake_run_path): """G1: --reach-disable-external-tool-checks -> coana --disable-external-tool-checks.""" - cmd, _ = _run(analyzer, mocker, disable_external_tool_checks=True) - assert "--disable-external-tool-checks" in cmd + assert "--disable-external-tool-checks" in _run(fake_run_path, disable_external_tool_checks=True)["args"] + assert "--disable-external-tool-checks" not in _run(fake_run_path)["args"] - cmd2, _ = _run(analyzer, mocker) - assert "--disable-external-tool-checks" not in cmd2 +def test_timeout_and_memory_forwarded_verbatim(fake_run_path): + """coana parses the units; config-file ints are coerced to strings.""" + args = _run(fake_run_path, timeout="10m", memory_limit=2048, concurrency=1)["args"] + assert args[args.index("--analysis-timeout") + 1] == "10m" + assert args[args.index("--memory-limit") + 1] == "2048" + assert args[args.index("--concurrency") + 1] == "1" -def test_concurrency_and_memory_args(analyzer, mocker): - """G7: explicit concurrency/memory propagate as coana args.""" - cmd, _ = _run(analyzer, mocker, concurrency=1, memory_limit="8192") - assert "--concurrency" in cmd and cmd[cmd.index("--concurrency") + 1] == "1" - assert "--memory-limit" in cmd and cmd[cmd.index("--memory-limit") + 1] == "8192" - -def test_timeout_and_memory_units_forwarded_verbatim(analyzer, mocker): - """Unit-bearing timeout/memory strings are forwarded to coana untouched (coana parses them).""" - cmd, _ = _run(analyzer, mocker, timeout="10m", memory_limit="8GB") - assert cmd[cmd.index("--analysis-timeout") + 1] == "10m" - assert cmd[cmd.index("--memory-limit") + 1] == "8GB" - - -def test_timeout_and_memory_int_values_coerced_to_str(analyzer, mocker): - """Config-file values can arrive as ints (set_defaults bypasses argparse type=); they must - still reach subprocess as strings, not raw ints.""" - cmd, _ = _run(analyzer, mocker, timeout=300, memory_limit=2048) - assert cmd[cmd.index("--analysis-timeout") + 1] == "300" - assert cmd[cmd.index("--memory-limit") + 1] == "2048" - - -def test_env_identifies_python_cli(analyzer, mocker): +def test_env_identifies_python_cli(fake_run_path): """G5: SOCKET_CLI_VERSION + SOCKET_CALLER_USER_AGENT forwarded to coana.""" - _, env = _run(analyzer, mocker) + env = _run(fake_run_path)["env"] assert env["SOCKET_CLI_VERSION"] == __version__ - assert env["SOCKET_CALLER_USER_AGENT"].startswith("socket/") + assert env["SOCKET_CALLER_USER_AGENT"] == _build_caller_user_agent() assert env["SOCKET_ORG_SLUG"] == "my-org" assert env["SOCKET_CLI_API_TOKEN"] == "test-api-token" -def test_no_proxy_env_set_by_default(analyzer, mocker, monkeypatch): - """coana inherits HTTPS_PROXY/HTTP_PROXY from the passed env; we don't set - SOCKET_CLI_API_PROXY ourselves (that's reserved for a future explicit --proxy flag).""" +@pytest.mark.parametrize("api_url", ["https://api.example.test/v0", "https://api.example.test/v0/"]) +def test_env_api_base_url_follows_the_client(fake_run_path, api_url): + env = _run(fake_run_path, analyzer=_analyzer(api_url))["env"] + assert env["SOCKET_CLI_API_BASE_URL"] == "https://api.example.test/v0/" + + +def test_no_proxy_env_set_by_default(fake_run_path, monkeypatch): + """coana reads HTTPS_PROXY/HTTP_PROXY itself; SOCKET_CLI_API_PROXY is reserved for a future + explicit --proxy flag.""" monkeypatch.delenv("SOCKET_CLI_API_PROXY", raising=False) monkeypatch.setenv("HTTPS_PROXY", "http://envproxy:3128") - _, env = _run(analyzer, mocker) - # Even with HTTPS_PROXY set, we don't copy it into SOCKET_CLI_API_PROXY (coana reads it itself). + env = _run(fake_run_path)["env"] + assert env["HTTPS_PROXY"] == "http://envproxy:3128" assert "SOCKET_CLI_API_PROXY" not in env -def test_repo_branch_env_present_when_supplied(analyzer, mocker): - _, env = _run(analyzer, mocker, repo_name="acme/widget", branch_name="main") +def test_repo_and_branch_env(fake_run_path): + """G6: caller passes None for default sentinels -> env keys omitted (cache hygiene).""" + env = _run(fake_run_path, repo_name="acme/widget", branch_name="main")["env"] assert env["SOCKET_REPO_NAME"] == "acme/widget" assert env["SOCKET_BRANCH_NAME"] == "main" - -def test_repo_branch_env_absent_when_none(analyzer, mocker): - """G6: caller passes None for default sentinels -> env keys omitted (cache hygiene).""" - _, env = _run(analyzer, mocker, repo_name=None, branch_name=None) + env = _run(fake_run_path)["env"] assert "SOCKET_REPO_NAME" not in env assert "SOCKET_BRANCH_NAME" not in env -# --- Coana package-spec resolution (pinned by default, latest is opt-in) --- - - -@pytest.mark.parametrize( - ("version", "expected"), - [ - # No --reach-version -> pinned default (no auto-update). - (None, f"@coana-tech/cli@{DEFAULT_COANA_CLI_VERSION}"), - # Explicit version pinned through. - ("1.2.3", "@coana-tech/cli@1.2.3"), - # 'latest' opts into the newest published version. - ("latest", "@coana-tech/cli@latest"), - # Surrounding whitespace is stripped; always versioned (never bare '@coana-tech/cli'). - (" 1.2.3 ", "@coana-tech/cli@1.2.3"), - ], -) -def test_resolve_coana_package_spec(analyzer, version, expected): - assert analyzer._resolve_coana_package_spec(version) == expected - - -def _spec_in(cmd): - """The @coana-tech/cli@ spec from an npx argv (it follows the npx flags).""" - return next(a for a in cmd if a.startswith("@coana-tech/cli@")) - - -def test_npx_uses_yes_and_force_flags(analyzer, mocker): - """npx is invoked with --yes --force — the exact flags the Node CLI passes for coana.""" - cmd, _ = _run(analyzer, mocker) - assert cmd[0] == "npx" - assert "--yes" in cmd - assert "--force" in cmd - - -@pytest.mark.parametrize( - ("version", "expected_spec"), - [ - (None, f"@coana-tech/cli@{DEFAULT_COANA_CLI_VERSION}"), # pinned default - ("9.9.9", "@coana-tech/cli@9.9.9"), # explicit pin - ("latest", "@coana-tech/cli@latest"), # opt-in to newest - ], -) -def test_npx_runs_resolved_version(analyzer, mocker, version, expected_spec): - cmd, _ = _run(analyzer, mocker, version=version) - assert _spec_in(cmd) == expected_spec - - -def test_default_path_never_runs_npm_install(analyzer, mocker): - """On the happy path we use npx only — no `npm install` (no global mutation).""" - run_mock = _spawn_mock(analyzer, mocker) - for call in run_mock.call_args_list: - assert call.args[0][:2] != ["npm", "install"] - - -def test_env_strips_npm_package_vars(analyzer, mocker, monkeypatch): +def test_env_strips_npm_package_vars(fake_run_path, monkeypatch): """npm_package_* dropped (E2BIG guard); npm_config_* kept. Parity with the Node CLI.""" monkeypatch.setenv("npm_package_dependencies_foo", "1.0.0") monkeypatch.setenv("npm_config_registry", "https://example.test") - _, env = _run(analyzer, mocker) + env = _run(fake_run_path)["env"] assert "npm_package_dependencies_foo" not in env - assert env.get("npm_config_registry") == "https://example.test" - - -# --- npm-install + node fallback (when the npx launcher fails before coana starts) --- + assert env["npm_config_registry"] == "https://example.test" -@pytest.mark.parametrize( - ("returncode", "is_launcher_failure"), - [ - # Signal kills / >=128 -> launcher failure -> retry. - (-9, True), # killed by signal - (137, True), # 128 + SIGKILL - (249, True), # observed npx launcher failure - # Small positive exit codes are ambiguous (coana's own codes) -> do NOT retry. - (1, False), - (2, False), - (127, False), - ], -) -def test_launcher_failure_heuristic(returncode, is_launcher_failure): - f = ReachabilityAnalyzer._npx_launcher_failed_before_coana - assert f(returncode) is is_launcher_failure - - -def _capture_spawns(analyzer, mocker, npx_behavior, **kwargs): - """Drive run_reachability_analysis capturing each spawned argv. - - ``npx_behavior`` is applied when argv[0] == 'npx': an int return code, or a - callable raising (e.g. FileNotFoundError). npm/node spawns always succeed. - """ - mocker.patch.object(analyzer, "_extract_scan_id", return_value="scan-123") - mocker.patch.object(reachability.tempfile, "mkdtemp", return_value="/tmp/socket-coana-x") - mocker.patch.object( - analyzer, - "_resolve_coana_bin", - return_value="/tmp/socket-coana-x/node_modules/@coana-tech/cli/coana.js", - ) - calls = [] - - def fake_run(argv, **_kw): - calls.append(argv) - if argv[0] == "npx" and callable(npx_behavior): - npx_behavior() - m = MagicMock() - m.returncode = npx_behavior if (argv[0] == "npx" and isinstance(npx_behavior, int)) else 0 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123", **kwargs) - return calls - - -def test_falls_back_to_npm_install_when_npx_launcher_fails(analyzer, mocker): - """npx exits >=128 (launcher died) -> npm install + node run coana directly.""" - calls = _capture_spawns(analyzer, mocker, npx_behavior=137) - assert calls[0][0] == "npx" - assert calls[1][:2] == ["npm", "install"] - assert f"@coana-tech/cli@{DEFAULT_COANA_CLI_VERSION}" in calls[1] - assert calls[2][0] == "node" - assert calls[2][1] == "/tmp/socket-coana-x/node_modules/@coana-tech/cli/coana.js" - assert calls[2][2:4] == ["run", "."] - - -def test_falls_back_when_npx_missing(analyzer, mocker): - """npx not on PATH (FileNotFoundError) -> npm install + node fallback.""" - def raise_enoent(): - raise FileNotFoundError("npx") - - calls = _capture_spawns(analyzer, mocker, npx_behavior=raise_enoent) - assert calls[0][0] == "npx" - assert calls[1][:2] == ["npm", "install"] - assert calls[2][0] == "node" - - -def test_no_fallback_on_ambiguous_exit_code(analyzer, mocker): - """A small positive npx exit (coana's own failure) does NOT trigger the npm fallback.""" - mocker.patch.object(analyzer, "_extract_scan_id", return_value=None) - calls = [] - - def fake_run(argv, **_kw): - calls.append(argv) - m = MagicMock() - m.returncode = 1 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - with pytest.raises(Exception): - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - assert calls[0][0] == "npx" - assert all(c[:2] != ["npm", "install"] for c in calls) - - -def test_force_npm_install_skips_npx(analyzer, mocker, monkeypatch): - """SOCKET_CLI_COANA_FORCE_NPM_INSTALL routes straight to npm install + node.""" - monkeypatch.setenv("SOCKET_CLI_COANA_FORCE_NPM_INSTALL", "1") - calls = _capture_spawns(analyzer, mocker, npx_behavior=0) - assert all(c[0] != "npx" for c in calls) - assert calls[0][:2] == ["npm", "install"] - assert calls[1][0] == "node" - - -def test_disable_fallback_propagates_npx_failure(analyzer, mocker, monkeypatch): - """SOCKET_CLI_COANA_DISABLE_NPM_FALLBACK: a launcher failure is NOT retried via npm.""" - monkeypatch.setenv("SOCKET_CLI_COANA_DISABLE_NPM_FALLBACK", "1") - mocker.patch.object(analyzer, "_extract_scan_id", return_value=None) - calls = [] - - def fake_run(argv, **_kw): - calls.append(argv) - m = MagicMock() - m.returncode = 137 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - with pytest.raises(Exception): - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - assert all(c[:2] != ["npm", "install"] for c in calls) - - -def test_launcher_npm_install_skips_npx(analyzer, mocker, monkeypatch): - """SOCKET_CLI_COANA_LAUNCHER=npm-install routes straight to npm install + node.""" - monkeypatch.setenv("SOCKET_CLI_COANA_LAUNCHER", "npm-install") - calls = _capture_spawns(analyzer, mocker, npx_behavior=0) - assert all(c[0] != "npx" for c in calls) - assert calls[0][:2] == ["npm", "install"] - assert calls[1][0] == "node" - - -def test_launcher_npx_propagates_npx_failure(analyzer, mocker, monkeypatch): - """SOCKET_CLI_COANA_LAUNCHER=npx: a launcher failure is NOT retried via npm.""" - monkeypatch.setenv("SOCKET_CLI_COANA_LAUNCHER", "npx") - mocker.patch.object(analyzer, "_extract_scan_id", return_value=None) - calls = [] - - def fake_run(argv, **_kw): - calls.append(argv) - m = MagicMock() - m.returncode = 137 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - with pytest.raises(Exception): - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - assert calls[0][0] == "npx" - assert all(c[:2] != ["npm", "install"] for c in calls) - - -def test_launcher_overrides_legacy_vars(analyzer, mocker, monkeypatch): - """A recognized SOCKET_CLI_COANA_LAUNCHER wins; legacy vars are ignored entirely.""" - monkeypatch.setenv("SOCKET_CLI_COANA_LAUNCHER", "auto") - monkeypatch.setenv("SOCKET_CLI_COANA_FORCE_NPM_INSTALL", "1") - monkeypatch.setenv("SOCKET_CLI_COANA_DISABLE_NPM_FALLBACK", "1") - calls = _capture_spawns(analyzer, mocker, npx_behavior=137) - assert calls[0][0] == "npx" # force-npm-install ignored: npx still attempted - assert calls[1][:2] == ["npm", "install"] # disable-fallback ignored: fallback runs - assert calls[2][0] == "node" - - -def test_launcher_unrecognized_value_behaves_as_auto(analyzer, mocker, monkeypatch): - """An unrecognized SOCKET_CLI_COANA_LAUNCHER value warns and behaves as auto.""" - monkeypatch.setenv("SOCKET_CLI_COANA_LAUNCHER", "bogus") - calls = _capture_spawns(analyzer, mocker, npx_behavior=137) - assert calls[0][0] == "npx" - assert calls[1][:2] == ["npm", "install"] - assert calls[2][0] == "node" - - -def test_fallback_installs_once_per_version(analyzer, mocker): - """A second in-process fallback for the same version reuses the install (no re-install).""" - mocker.patch.object(analyzer, "_extract_scan_id", return_value="scan-123") - mocker.patch.object(reachability.tempfile, "mkdtemp", return_value="/tmp/socket-coana-cache") - mocker.patch.object( - analyzer, - "_resolve_coana_bin", - return_value="/tmp/socket-coana-cache/node_modules/@coana-tech/cli/coana.js", - ) - # The cached script path must "exist" for the 2nd run to reuse it. - mocker.patch.object(reachability.os.path, "exists", return_value=True) - calls = [] - - def fake_run(argv, **_kw): - calls.append(argv) - m = MagicMock() - m.returncode = 137 if argv[0] == "npx" else 0 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - - npm_installs = [c for c in calls if c[:2] == ["npm", "install"]] - assert len(npm_installs) == 1 # installed once, reused on the second fallback - - -def test_fallback_node_missing_raises_clear_error(analyzer, mocker): - """If `node` is missing in the fallback, surface a clear error (not opaque FileNotFoundError).""" - mocker.patch.object(analyzer, "_extract_scan_id", return_value=None) - mocker.patch.object(reachability.tempfile, "mkdtemp", return_value="/tmp/socket-coana-n") - mocker.patch.object( - analyzer, - "_resolve_coana_bin", - return_value="/tmp/socket-coana-n/node_modules/@coana-tech/cli/coana.js", - ) - - def fake_run(argv, **_kw): - if argv[0] == "npx": - m = MagicMock() - m.returncode = 137 - return m - if argv[0] == "node": - raise FileNotFoundError("node") - m = MagicMock() # npm install succeeds - m.returncode = 0 - return m - - mocker.patch.object(reachability.subprocess, "run", side_effect=fake_run) - with pytest.raises(Exception, match="node"): - analyzer.run_reachability_analysis(org_slug="my-org", target_directory=".", tar_hash="tar-hash-abc123") - - -def test_build_coana_node_cmd_js_vs_binary(): - f = ReachabilityAnalyzer._build_coana_node_cmd - assert f("/x/coana.js", ["run", "."]) == ["node", "/x/coana.js", "run", "."] - assert f("/x/coana.mjs", ["run"]) == ["node", "/x/coana.mjs", "run"] - assert f("/x/coana", ["run", "."]) == ["/x/coana", "run", "."] - - -def test_resolve_coana_bin_parses_package_json(analyzer, tmp_path): - pkg_dir = tmp_path / "node_modules" / "@coana-tech" / "cli" - pkg_dir.mkdir(parents=True) - - # string bin - (pkg_dir / "package.json").write_text('{"bin": "dist/coana.js"}') - assert analyzer._resolve_coana_bin(str(tmp_path)) == str(pkg_dir / "dist" / "coana.js") - - # dict bin, prefer the "coana" entry - (pkg_dir / "package.json").write_text('{"bin": {"coana": "dist/c.js", "other": "x.js"}}') - assert analyzer._resolve_coana_bin(str(tmp_path)) == str(pkg_dir / "dist" / "c.js") - - # dict bin without "coana" -> first value - (pkg_dir / "package.json").write_text('{"bin": {"other": "x.js"}}') - assert analyzer._resolve_coana_bin(str(tmp_path)) == str(pkg_dir / "x.js") - - # missing bin -> raises - (pkg_dir / "package.json").write_text("{}") - with pytest.raises(Exception, match="bin"): - analyzer._resolve_coana_bin(str(tmp_path)) +def test_failing_run_raises(fake_run_path): + with pytest.raises(Exception, match="exit code 1"): + _run(fake_run_path, additional_params=["--fail"]) diff --git a/uv.lock b/uv.lock index 7aeecc9..7d88693 100644 --- a/uv.lock +++ b/uv.lock @@ -1292,7 +1292,7 @@ wheels = [ [[package]] name = "socketsecurity" -version = "2.10.12" +version = "2.11.0" source = { editable = "." } dependencies = [ { name = "beautifulsoup4" },