# supply-chain-risk-auditor Audits a project's dependencies for supply-chain risk: version-matched advisories for direct dependencies and the full lockfile tree, abandoned or archived upstreams, npm publisher concentration, and install-time script execution. Use when asked to audit dependencies, assess supply-chain or third-party package risk, or review a dependency tree before an engagement. - **Kind:** skill - **Source:** https://github.com/trailofbits/skills - **Page:** https://forefy.com/skills/7c2f5b50-4e0b-4228-ab1c-1b599957dc5c - **API (JSON + files):** https://forefy.com/api/asr/7c2f5b50-4e0b-4228-ab1c-1b599957dc5c --- ## SKILL.md --- name: supply-chain-risk-auditor description: "Audits a project's dependencies for supply-chain risk: version-matched advisories for direct dependencies and the full lockfile tree, abandoned or archived upstreams, npm publisher concentration, and install-time script execution. Use when asked to audit dependencies, assess supply-chain or third-party package risk, or review a dependency tree before an engagement." allowed-tools: Read Write Bash Glob Grep --- # Supply Chain Risk Auditor Generates a supply-chain risk report for a project's direct dependencies (npm, PyPI, Go), plus an advisory sweep of everything its lockfile resolves. Two deterministic scripts do the measuring; your job is the judgment they refuse to automate. ## Why the scripts do the measuring, not you Every figure in this report is a claim about somebody else's project, and hand-collected figures were measured wrong before this skill was rebuilt around scripts: GitHub contributor counts said five-plus people maintain `lodash` where npm's ACL says one, and `gh` saw zero downloads for a package that moves 164 million a week. Do not estimate maintainer counts, downloads, staleness, or CVE history from `gh`, web search, or memory — run the collector, and quote what it measured. The scripts enforce two rules worth knowing before you read their output: - **Unavailable data is never evidence of risk.** Every criterion resolves to assessed-clean, assessed-flagged, or unassessable-with-a-reason. - **An absent measurement is never a clean verdict.** A run that measured nothing exits non-zero instead of printing a report that finds nothing. ## Workflow 1. Confirm the target directory has manifests: `package.json`, `pyproject.toml`, `requirements*.txt`, or `go.mod`. If none exist, say so and stop — do not audit an ecosystem this collector does not parse by hand. Lockfiles read for exact versions and the transitive sweep: `package-lock.json`/`npm-shrinkwrap.json`, `uv.lock`, and a go 1.17+ `go.mod`. `yarn.lock`, `pnpm-lock.yaml`, and `poetry.lock` are not read — the report says so when they are present, and versions fall back to pins or the latest release. 2. Check `gh auth status`. Unauthenticated GitHub allows 60 requests/hour against 5,000, and the collector makes several per dependency; expect repository criteria to come back unassessable without it. Say so rather than fixing it silently. 3. Collect, then render. Put outputs somewhere outside the audited repository unless asked otherwise: ```sh uv run {baseDir}/scripts/collect.py --json /findings.json uv run {baseDir}/scripts/render.py /findings.json --out /report.md ``` Expect a few minutes for ~50 dependencies — several HTTP requests per dependency, more with many Go modules, and slower without authenticated `gh`. If `collect.py` exits non-zero, it is refusing to report — relay its message verbatim instead of retrying or working around it. 4. Read `report.md` and `findings.json`. The report is the deliverable; the JSON carries the datum behind every verdict when you need to cite one. 5. Add what the collector cannot, clearly separated from what it measured: - A short narrative for this reader: what to act on first, and why. - Upgrade paths for advisory findings — check whether the fix is a patch or a major version away. - Replacement candidates for abandoned or archived dependencies. Verify a candidate exists in the registry before naming it, and label these as judgment, not measurement. - For flagged install scripts: whether `npm ci --ignore-scripts` is viable for this project's build. ## Style for what you add Write added prose the way a security report reads, and apply the same register to the report addendum and the final reply alike — replies get pasted into tickets and reports verbatim. State the finding, the datum behind it, and the action. - Impersonal and declarative: no first or second person ("I ran the collector", "you should upgrade"), no contractions, no exclamation points. - Active voice, with the subject matter as the actor: "upgrading to 1.19.0 clears all 25 advisories", not "it is recommended that axios be upgraded". - Objective: no intensifiers or subjective framing ("very", "significant", "fortunately"), and no guesses about why the project chose what it chose. - Tense: past for what the audit did, present for the state of the dependencies, future for the consequences of acting or not. - Constructive: a recommendation names the action and its cost, never a culprit. If the `report-writing:writing-style` skill is available in the session, follow it — it is the full version of this register. The rendered report carries facts only. The interpretive rules below are instructions to you, not content for the reader — do not copy them into the deliverable as caveats or framing. ## Reading the report - **Unassessable is not risk.** PyPI publishes no maintainer ACL and Go has no registry; those rows say what could not be known, not what is wrong. - **The coverage table bounds every claim.** "No advisories" means "none among what was assessed" — check the assessed count before repeating a clean verdict. - **Quote figures verbatim.** Do not re-derive, round, or embellish the report's numbers; every one is reproducible from the artifact. - **Absence from the findings is not endorsement.** A dependency with no findings was measured against these criteria only. ## Rationalizations to reject - "`gh` can give me maintainer counts faster than the collector." Measured wrong — repo contributors and registry publish rights are different populations. - "No findings, so the dependencies are safe." Read the coverage table; on PyPI and Go, half the criteria are structurally unassessable. - "The unassessable rows would just confuse the reader; I'll drop them." They are the boundary of every claim in the report. Dropping them turns partial coverage into a clean bill of health, which is the failure this skill was rebuilt to prevent. - "The version is probably close enough." A range checked at latest-release and a lockfile-resolved version are different claims; the report labels which one it makes. Keep the label. ## When not to use - License compliance auditing. - Scanning the target's own source for vulnerabilities or secrets — this skill never reads dependency source, only registry, advisory, and repository metadata. - Judging whether the project installs or builds. The audit is designed to work from nothing more than the dependency list — manifests and lockfiles — and never installs, builds, or executes anything. Broken installs and import-time breakage are out of scope, and worth saying so if the user seems to expect them. - Ecosystems other than npm, PyPI, and Go; say the ecosystem is unsupported rather than improvising an audit for it. ## agents ``` ``` ## agents/openai.yaml ```yaml interface: icon_small: "assets/trail-of-bits-mark.svg" icon_large: "assets/trail-of-bits-mark.svg" brand_color: "#D83A34" ``` ## assets ``` ``` ## assets/trail-of-bits-mark.svg ``` ``` ## scripts ``` ``` ## scripts/collect.py ```python #!/usr/bin/env python3 # /// script # requires-python = ">=3.11" # /// """Collect supply-chain risk signals for a project's direct dependencies. Emits a JSON artifact; `render.py` turns it into a report. Direct dependencies only, by design — the scope is what keeps an audit sizeable. Dependency source is never fetched or read; every signal comes from a registry, an advisory database, or repository metadata. Two rules the code is arranged to enforce rather than remember: - **Unavailable data is not evidence of risk.** Every criterion resolves to assessed-clean, assessed-flagged, or unassessable-with-reason. - **An absent measurement is not a clean verdict.** An empty answer from a vulnerability database means "no advisories" only for a package known to exist. Usage: uv run collect.py [--json out.json] [--cache DIR] [--offline] """ from __future__ import annotations import argparse import json import re import sys import tempfile import tomllib from datetime import UTC, datetime from pathlib import Path import sources from model import ( CRITERIA, TIER_SCORECARD, Dependency, ReconciliationError, Signal, to_json, ) # Two years without a push is "stale". A one-year threshold flagged jinja2 and # itsdangerous at 14 months — maintained-but-finished libraries whose flags teach a # reader to skim past the criterion — while the packages that motivated it (stream-throttle # at 11 years, dev-null at 9, escape-html at 4) clear two years easily. Reported with the # actual date so a reader who draws the line elsewhere can. STALE_DAYS = 730 # An absolute floor, not a rank. The original skill defined low popularity relative to the # target's other dependencies, which is uncomputable across ecosystems: npm gives weekly # downloads, PyPI returns -1, Go has no concept at all. LOW_DOWNLOADS_PER_WEEK = 1000 REPO_CRITERIA = ("archived", "staleness", "security_policy") # Shared by every criterion of every dependency that resolves outside its public # registry, so the report can group them into one bullet per criterion. The specific # source is the signal's value, and each dependency also gets its own Method note. NON_REGISTRY_REASON = ( "the dependency resolves from outside its public registry, so registry-keyed data " "does not apply to it (the source is named in Method and caveats)" ) # ------------------------------------------------------------------ npm manifests def _npm_locked_versions(project: Path) -> dict[str, str]: """Resolved direct-dependency versions from an npm lockfile, if one exists.""" for name in ("package-lock.json", "npm-shrinkwrap.json"): path = project / name if not path.exists(): continue data = _read_json(path) out: dict[str, str] = {} for key, entry in (data.get("packages") or {}).items(): if key.startswith("node_modules/") and entry.get("version"): out[key.removeprefix("node_modules/")] = entry["version"] for pkg, entry in (data.get("dependencies") or {}).items(): if isinstance(entry, dict) and entry.get("version") and pkg not in out: out[pkg] = entry["version"] if out: return out return {} # One lockfile-resolved package: (ecosystem, name, version, dev-only or None). LockedPackage = tuple[str, str, str, bool | None] # A lockfile entry whose registry existence cannot be attested, with the reason. Unverifiable = dict def _npm_all_locked( project: Path, ) -> tuple[list[LockedPackage], list[Unverifiable], str | None, str | None]: """Every package the npm lockfile resolves, split by registry attestation. Returns (attested packages, unverifiable entries, lockfile name or None, note or None). Attested means the entry carries an integrity hash for a registry tarball — the datum that lets an empty advisory answer read as clean. Git, file, and private-registry entries go in the unverifiable list instead: OSV data is keyed by public-registry name, so querying it for a package that resolves elsewhere either proves nothing (absence) or attributes another package's advisories to it. Only v2+ lockfiles carry the flat `packages` table; a v1-only lockfile yields a note rather than a silent zero that would read as "no transitive packages". """ for lockname in ("package-lock.json", "npm-shrinkwrap.json"): path = project / lockname if not path.exists(): continue packages = _read_json(path).get("packages") if not isinstance(packages, dict): return ( [], [], None, f"{lockname} predates npm 7's flat package table, so the transitive tree " f"was not read from it.", ) out: list[LockedPackage] = [] unverifiable: list[Unverifiable] = [] for key, entry in packages.items(): if "node_modules/" not in key or entry.get("link") or not entry.get("version"): continue name = entry.get("name") or key.rsplit("node_modules/", 1)[1] resolved = str(entry.get("resolved") or "") if entry.get("integrity") and resolved.startswith("https://registry.npmjs.org/"): out.append(("npm", name, entry["version"], bool(entry.get("dev")))) else: unverifiable.append( { "ecosystem": "npm", "name": name, "version": entry["version"], "reason": ( f"resolves from {resolved or 'an undeclared source'}, not the " f"npm registry, so registry-keyed advisory data does not " f"apply to it" ) if not resolved.startswith("https://registry.npmjs.org/") else "lockfile entry carries no integrity hash", } ) return out, unverifiable, lockname, None return [], [], None, None _EXACT_SEMVER = re.compile(r"^\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.\-+]+)?$") _ALIAS = re.compile(r"^npm:(?P@?[^@]+(?:/[^@]+)?)@") # Specs that are not a registry version range. The dependency key is then a local or # remote artifact rather than a package name that can be looked up. _NON_REGISTRY = ("file:", "link:", "workspace:", "portal:", "git+", "git:", "http://", "https://") def _npm_spec_kind(spec: str) -> str: """Classify an npm dependency spec: `registry`, `alias`, or a non-registry source.""" spec = (spec or "").strip() if spec.startswith("npm:"): return "alias" for prefix in _NON_REGISTRY: if spec.startswith(prefix): return "non-registry" # `owner/repo` with no version operator is GitHub shorthand, not a registry range. if "/" in spec and not any(ch in spec for ch in "^~<>= .*") and spec.count("/") == 1: return "non-registry" return "registry" def _exact_npm_pin(spec: str) -> str | None: """Return an exact version only for a genuine pin. `1.x`, `1.2.x` and `2` are ranges. Accepting them as pins fed the range string to OSV as if it were a version and suppressed resolution of the real release. """ spec = (spec or "").strip() return spec if _EXACT_SEMVER.match(spec) else None def parse_npm(project: Path) -> tuple[list[Dependency], list[str]]: """Direct npm dependencies from package.json, with the resolved version where known. Returns: The dependencies, and notes for anything that could not be treated as a registry package. `optionalDependencies` are installed by default and so are included; `peerDependencies` are the consumer's responsibility and are noted, not audited. """ manifest = project / "package.json" if not manifest.exists(): return [], [] data = _read_json(manifest) locked = _npm_locked_versions(project) deps: list[Dependency] = [] notes: list[str] = [] fields = (("dependencies", False), ("devDependencies", True), ("optionalDependencies", False)) for field_name, is_dev in fields: table = data.get(field_name) or {} if not isinstance(table, dict): raise SystemExit(f"error: {manifest}: '{field_name}' is not a table of name -> spec") for name, spec in table.items(): dep, note = _npm_dependency(name, str(spec), is_dev, locked) if dep: deps.append(dep) if note: notes.append(note) peers = data.get("peerDependencies") or {} if peers: count = "1 peerDependency was" if len(peers) == 1 else f"{len(peers)} peerDependencies were" notes.append( f"{count} not audited: they are supplied by the consuming project rather than " f"installed by this one." ) return deps, notes def _npm_dependency( name: str, spec: str, is_dev: bool, locked: dict[str, str] ) -> tuple[Dependency | None, str | None]: kind = _npm_spec_kind(spec) declared = name if kind == "alias": match = _ALIAS.match(spec) target = match.group("name") if match else None if not target: return None, f"`{name}` uses an npm alias this parser could not read ({spec})." note = f"`{name}` is an alias for `{target}`; the audit follows the target." name = target kind = "registry" else: note = None if kind == "non-registry": reason = f"resolves from {spec}, not the npm registry" return ( Dependency(ecosystem="npm", name=name, dev=is_dev, non_registry_reason=reason), f"`{name}` {reason}, so no registry or advisory data applies to it.", ) # The lockfile keys the declared name (node_modules/), so an alias must # be looked up under it, not under the rewritten target. version, source = locked.get(declared), "lockfile" if not version: version, source = _exact_npm_pin(spec), "manifest-pin" if not version: source = "unresolved" return Dependency( ecosystem="npm", name=name, version=version, version_source=source, dev=is_dev ), note # ----------------------------------------------------------------- PyPI manifests _REQ_SPLIT = re.compile(r"[<>=!~\[;@]") # A permissive PEP 440 shape: enough to reject line-continuation and inline-option # debris ("2.19.0 \\", "2.19.0 --hash") that would otherwise be sent to OSV as a # version and printed in the report as a version-matched claim. # # The optional `v` is load-bearing: PEP 440 permits it, pip accepts `django==v3.2.0`, # and requiring a leading digit turned that legal pin into an unresolved version, so # advisories were matched against the latest release and 62 real ones read as clean. The # prefix is stripped rather than kept, because the canonical form is what OSV, the # report, and any hand-check should agree on. _VALID_VERSION = re.compile(r"^v?[0-9][0-9A-Za-z.+!_-]*$") _PEP503 = re.compile(r"[-_.]+") _VALID_NAME = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._-]*[A-Za-z0-9])?$") def normalize_pypi_name(name: str) -> str: """PEP 503 normalisation: lowercase, runs of -_. collapsed to one hyphen.""" return _PEP503.sub("-", name).lower() def _strip_marker(line: str) -> str: """Drop a PEP 508 environment marker. Markers contain `==`, so leaving one in place made `psutil; sys_platform == 'win32'` resolve to the version `'win32'` — fabricated, then sent to OSV and reported as fact. """ return line.split(";", 1)[0].strip() # `name [extras] @ url` — PEP 508's direct-reference form. `@` is also a name # terminator in _REQ_SPLIT, so without this check the URL is silently discarded and a # git or file fork is looked up on PyPI under the public package's name. _DIRECT_REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\s*(?:\[[^\]]*\]\s*)?@\s*(\S+)") def _direct_reference_url(line: str) -> str | None: """The URL of a PEP 508 direct reference, if this requirement uses one.""" spec = _strip_marker(line.split("#", 1)[0].strip()) match = _DIRECT_REF.match(spec) return match.group(1) if match else None def _requirement_name(line: str) -> str | None: """Extract a distribution name, or None when the line declares no requirement.""" line = line.split("#", 1)[0].strip() if not line or line.startswith("-"): return None candidate = _REQ_SPLIT.split(_strip_marker(line), 1)[0].strip() return candidate or None # Extra and requirements-file names that conventionally hold development dependencies. # `coverage` and `pytest` arriving from a `test` extra are not production dependencies, and # saying they ship in the built artifact is a claim the manifest does not support. # Matched as whole tokens, not substrings: substring matching classified the runtime # extras `docker` ("doc") and `tracing` ("ci") as build-time only, which understates # blast radius — the unsafe direction for this report. _DEV_GROUP_TOKENS = frozenset( "test tests testing dev development doc docs lint linting type types typing " "typecheck check checks bench benchmark benchmarks ci".split() ) _TOKEN_SPLIT = re.compile(r"[^a-z0-9]+") def _is_dev_group(name: str) -> bool: return any(token in _DEV_GROUP_TOKENS for token in _TOKEN_SPLIT.split(name.lower())) def _pypi_specs(project: Path) -> tuple[dict[str, tuple[str, bool]], list[str]]: """Gather name -> (raw spec, is_dev) for every direct Python requirement.""" specs: dict[str, tuple[str, bool]] = {} notes: list[str] = [] pyproject = project / "pyproject.toml" if pyproject.exists(): data = _read_toml(pyproject) groups = [((data.get("project") or {}).get("dependencies") or [], False)] for extra, items in ( (data.get("project") or {}).get("optional-dependencies") or {} ).items(): groups.append((items, _is_dev_group(extra))) for group in (data.get("dependency-groups") or {}).values(): # PEP 735 groups exist for development dependencies by definition. groups.append(([g for g in group if isinstance(g, str)], True)) poetry = ((data.get("tool") or {}).get("poetry") or {}).get("dependencies") or {} if poetry: notes.append( f"{len(poetry)} Poetry dependencies in the tool.poetry.dependencies " f"table were not parsed; only PEP 621 and PEP 735 tables are read." ) for group, is_dev in groups: _absorb_specs(group, specs, notes, is_dev) for candidate in sorted(project.glob("requirements*.txt")): lines = _read_text(candidate).splitlines() includes = [ln for ln in lines if ln.strip().startswith(("-r", "--requirement"))] if includes: notes.append( f"{candidate.name} includes {len(includes)} other requirements file(s) that " f"were not followed, so its dependency list may be incomplete." ) _absorb_specs(lines, specs, notes, _is_dev_group(candidate.stem)) return specs, notes def _absorb_specs( raw_lines: list[str], specs: dict[str, tuple[str, bool]], notes: list[str], is_dev: bool ) -> None: for raw in raw_lines: if not isinstance(raw, str): continue name = _requirement_name(raw) if not name: continue if not _VALID_NAME.match(name): notes.append(f"skipped an unparseable requirement: {raw.strip()!r}") continue canonical = normalize_pypi_name(name) existing = specs.get(canonical) if existing is None or (existing[1] and not is_dev): specs[canonical] = (raw, is_dev) # What terminates the value of a `==` pin. Deliberately not `_REQ_SPLIT`, which also # splits on `!` for the `!=` operator and would truncate a PEP 440 epoch — `1!2.0` # became the version `1`, a wrong pin reported as version-matched. _VERSION_END = re.compile(r"[\s,;]") def _pypi_version(raw: str, locked: dict[str, str], canonical: str) -> tuple[str | None, str]: if canonical in locked: return locked[canonical], "lockfile" spec = _strip_marker(raw) if "==" in spec: candidate = _VERSION_END.split(spec.split("==", 1)[1].strip().strip(","), 1)[0].strip() # `1.0.*` is a range, not a pin, and pip-compile hash lines leave debris # (`2.19.0 \\`) that must not be reported as a version. if candidate and "*" not in candidate and _VALID_VERSION.match(candidate): return candidate.removeprefix("v"), "manifest-pin" return None, "unresolved" def _uv_lock_versions(project: Path) -> dict[str, str]: lock = project / "uv.lock" if not lock.exists(): return {} data = _read_toml(lock) return { normalize_pypi_name(pkg["name"]): pkg["version"] for pkg in data.get("package") or [] if pkg.get("name") and pkg.get("version") } def _uv_non_registry_sources(project: Path) -> dict[str, str]: """Canonical name -> source kind for uv.lock entries that do not resolve from PyPI. A direct dependency in this map must not be looked up on PyPI by name: the lock says the project installs it from git, a directory, or a local path, so a same-named public package's advisories and metadata do not apply to it. """ lock = project / "uv.lock" if not lock.exists(): return {} out: dict[str, str] = {} for pkg in _read_toml(lock).get("package") or []: source = pkg.get("source") or {} if not pkg.get("name") or source.get("registry"): continue if source.get("editable") or source.get("virtual"): continue out[normalize_pypi_name(pkg["name"])] = next(iter(source), "unknown") return out def _uv_all_locked( project: Path, ) -> tuple[list[LockedPackage], list[Unverifiable], str | None, str | None]: """Every package uv.lock resolves, split by registry attestation. The lock includes the audited project itself as an editable or virtual source; that entry is the subject, not a dependency. Git, directory, and path sources are unverifiable against PyPI — a vendored fork at `../vendor/flask` must not inherit PyPI flask's advisories, since vendoring a patched fork is a common way to fix exactly those. uv.lock does not mark which packages are development-only per entry, so the dev marker is None throughout. """ lock = project / "uv.lock" if not lock.exists(): return [], [], None, None out: list[LockedPackage] = [] unverifiable: list[Unverifiable] = [] for pkg in _read_toml(lock).get("package") or []: source = pkg.get("source") or {} if not pkg.get("name") or not pkg.get("version"): continue if source.get("editable") or source.get("virtual"): continue name = normalize_pypi_name(pkg["name"]) if source.get("registry"): out.append(("PyPI", name, pkg["version"], None)) else: kind = next(iter(source), "unknown") unverifiable.append( { "ecosystem": "PyPI", "name": name, "version": pkg["version"], "reason": ( f"resolves from a {kind} source, not PyPI, so PyPI-keyed " f"advisory data does not apply to it" ), } ) return out, unverifiable, "uv.lock", None def parse_pypi(project: Path) -> tuple[list[Dependency], list[str]]: """Direct Python dependencies from pyproject.toml and/or requirements files.""" specs, notes = _pypi_specs(project) if not specs: return [], notes locked = _uv_lock_versions(project) non_registry = _uv_non_registry_sources(project) deps = [] for canonical, (raw, is_dev) in specs.items(): version, source = _pypi_version(raw, locked, canonical) reason = None url = _direct_reference_url(raw) if url: reason = f"resolves from {url}, not PyPI" elif canonical in non_registry: reason = f"resolves from a {non_registry[canonical]} source, not PyPI" if reason: notes.append(f"`{canonical}` {reason}, so no registry or advisory data applies to it.") deps.append( Dependency( ecosystem="PyPI", name=canonical, version=version, version_source=source, dev=is_dev, non_registry_reason=reason, ) ) return deps, notes # ------------------------------------------------------------------- Go manifests _GOMOD_REQUIRE = re.compile(r"^\s*(?:require\s+)?([\w.\-]+(?:\.[\w.\-]+)*/[^\s]+)\s+(v[^\s]+)") _GOMOD_BLOCK_OPEN = re.compile(r"^(require|replace|exclude|retract)\s*\($") _GOMOD_OTHER = ("replace ", "exclude ", "retract ") def _gomod_line_kind(line: str, block: str | None) -> tuple[str | None, str]: """Classify one go.mod line, tracking which block it sits in. Returns: The block in effect after this line, and one of `require`, `indirect`, `directive`, or `ignore`. Block context is what separates a requirement from a `replace` entry, since both read as `module version`. """ stripped = line.strip() opener = _GOMOD_BLOCK_OPEN.match(stripped) if opener: return opener.group(1), "ignore" if stripped == ")": return None, "ignore" if block in {"replace", "exclude", "retract"} or stripped.startswith(_GOMOD_OTHER): return block, "directive" if block not in {"require", None}: return block, "ignore" if not _GOMOD_REQUIRE.match(line): return block, "ignore" if block is None and not stripped.startswith("require "): return block, "ignore" if "// indirect" in line: return block, "indirect" return block, "require" def _go_dependency(line: str) -> Dependency: module, version = _GOMOD_REQUIRE.match(line).groups() return Dependency( ecosystem="Go", name=module, # OSV wants Go versions without the `v` prefix. version=version.removeprefix("v"), # go.mod records a *minimum*; module-version selection can pick higher. version_source="go-mod-minimum", # go.mod has no dev section; asserting either way would invent a distinction. dev=None, ) def parse_go(project: Path) -> tuple[list[Dependency], list[str]]: """Direct Go modules from go.mod. Block context matters: every block-form directive line looks like `module version`, so a parser without it reads `replace (...)` entries as requirements. That reports the module the project replaced *away from*, at the abandoned version, while never assessing the replacement actually built — and reports `exclude` entries as used. """ gomod = project / "go.mod" if not gomod.exists(): return [], [] deps: list[Dependency] = [] notes: list[str] = [] block: str | None = None replaced = 0 for line in _read_text(gomod).splitlines(): block, kind = _gomod_line_kind(line, block) if kind == "directive": replaced += 1 elif kind == "require": deps.append(_go_dependency(line)) if replaced: notes.append( f"{replaced} replace/exclude/retract directives in go.mod were not treated as " f"dependencies; where a module is replaced, the replacement was not audited." ) return deps, notes _GO_DIRECTIVE = re.compile(r"^go\s+(\d+)\.(\d+)", re.MULTILINE) def _go_indirect( project: Path, ) -> tuple[list[LockedPackage], list[Unverifiable], str | None, str | None]: """Indirect modules from go.mod, which is complete only for go 1.17 and later. Modules declaring an older go directive list an arbitrary subset of their indirect requirements, so reading them would report "the transitive tree was checked" about a tree that was mostly absent. go.mod carries no dev marker; None throughout. go.mod attests nothing about existence — the split into attested and unverifiable happens in `sweep_transitive`, which asks the Go module proxy about each entry. """ gomod = project / "go.mod" if not gomod.exists(): return [], [], None, None text = _read_text(gomod) match = _GO_DIRECTIVE.search(text) if not match or (int(match.group(1)), int(match.group(2))) < (1, 17): return ( [], [], None, "go.mod declares a go directive older than 1.17 (or none), so its indirect " "module list is incomplete and the transitive tree was not read from it.", ) out: list[LockedPackage] = [] block: str | None = None for line in text.splitlines(): block, kind = _gomod_line_kind(line, block) if kind == "indirect": module, version = _GOMOD_REQUIRE.match(line).groups() out.append(("Go", module, version.removeprefix("v"), None)) return out, [], "go.mod", None PARSERS = (parse_npm, parse_pypi, parse_go) def _read_text(path: Path) -> str: """Read a plain-text manifest as UTF-8, refusing anything that will not decode. The JSON and TOML readers below already turn an undecodable file into a refusal. Without this, `requirements*.txt` and `go.mod` did not: `main()` catches only `ReconciliationError`, so a `UnicodeDecodeError` escaped as a traceback. That is not hypothetical on the platform this exists to support — PowerShell 5.1 redirection writes UTF-16LE with a BOM, so any requirements file generated with `>` on a stock Windows box begins with two bytes that are invalid UTF-8. Args: path: The file to read. Returns: The decoded text. Raises: SystemExit: The file is not UTF-8, or could not be read. """ try: return path.read_text(encoding="utf-8") except UnicodeDecodeError as exc: raise SystemExit( f"error: {path} is not UTF-8, so its contents cannot be read: {exc}. " f"A file produced by PowerShell redirection is UTF-16 — re-save it as " f"UTF-8, or regenerate it through `Out-File -Encoding utf8`." ) from exc except OSError as exc: raise SystemExit(f"error: cannot read {path}: {exc}") from exc def _read_json(path: Path) -> dict: try: data = json.loads(path.read_text(encoding="utf-8"), strict=False) if not isinstance(data, dict): raise SystemExit( f"error: {path} holds a JSON {type(data).__name__}, not the object this " f"manifest format requires" ) return data except json.JSONDecodeError as exc: raise SystemExit(f"error: {path} is not valid JSON: {exc}") from exc # UnicodeDecodeError is a ValueError, not a JSONDecodeError, so the two clauses # around this one do not catch it and it used to escape as a traceback. JSON is # UTF-8 by RFC 8259, so a manifest that is not decodable is malformed input and # earns the same refusal as invalid syntax. except UnicodeDecodeError as exc: raise SystemExit(f"error: {path} is not UTF-8, which JSON requires: {exc}") from exc except OSError as exc: raise SystemExit(f"error: cannot read {path}: {exc}") from exc def _read_toml(path: Path) -> dict: try: return tomllib.loads(path.read_text(encoding="utf-8")) except tomllib.TOMLDecodeError as exc: raise SystemExit(f"error: {path} is not valid TOML: {exc}") from exc except UnicodeDecodeError as exc: raise SystemExit(f"error: {path} is not UTF-8, which TOML requires: {exc}") from exc except OSError as exc: raise SystemExit(f"error: cannot read {path}: {exc}") from exc def discover(project: Path) -> tuple[list[Dependency], list[str]]: """Parse every manifest, dropping the weaker of any duplicate.""" found: list[Dependency] = [] notes: list[str] = [] for parser in PARSERS: deps, parser_notes = parser(project) found.extend(deps) notes.extend(parser_notes) seen: dict[str, Dependency] = {} for dep in found: existing = seen.get(dep.key) if existing is None or (existing.dev is True and dep.dev is not True): seen[dep.key] = dep return list(seen.values()), notes # ---------------------------------------------------------------------- signalling def _reason(exc: Exception, source: str) -> str: """A human-facing reason. The raw URL stays in the artifact, not in the report.""" text = str(exc) if isinstance(exc, sources.NotFound): return f"not published in {source}" if "rate limited" in text: return f"{source} rate-limited this audit" if "offline" in text: return f"{source} was not in the local cache and this run was offline" return f"{source} did not answer" def _advisory_signal(dep: Dependency, found: dict[str, list[str]]) -> Signal: """Advisories, refusing to read an empty answer as safety for an unknown package.""" if dep.key not in found: return Signal.unassessable("OSV did not answer for this package") ids = found[dep.key] if not ids: if dep.exists is not True: return Signal.unassessable( "OSV recorded no advisories, but this package was not confirmed to exist " "in its registry, and an empty answer cannot distinguish 'no advisories' " "from 'unknown package'" ) return Signal.clean("no advisories recorded", []) count = f"{len(ids)} advisory" if len(ids) == 1 else f"{len(ids)} advisories" verb = "affects" if len(ids) == 1 else "affect" described = { "lockfile": f"{count} {verb} the installed {dep.version}", "manifest-pin": f"{count} {verb} the pinned {dep.version}", "go-mod-minimum": ( f"{count} {verb} {dep.version}, go.mod's minimum — module-version selection " f"may build a higher version" ), "latest-release": ( f"{count} {verb} {dep.version}, the current latest release; the manifest gives " f"a range, so the version this project installs was not resolved" ), } return Signal.flagged( described.get( dep.version_source, f"{count} recorded for the package; no version resolved, so this is historical " f"rather than a statement about what is installed", ), ids, ) def human_days(days: int) -> str: """A duration at the precision a reader can use. "no push in 3,884 days" is four significant figures on a decade, and reads as scanner output rather than as a judgement. The exact date travels alongside for anyone who wants it. """ if days < 60: return f"{days} days" if days < 730: return f"{round(days / 30.4)} months" return f"{days / 365.25:.0f} years" def _staleness_signal(pushed: str | None) -> Signal: if not pushed: return Signal.unassessable("repository reports no last-push date") try: when = datetime.fromisoformat(pushed.replace("Z", "+00:00")) except ValueError: return Signal.unassessable(f"could not read the last-push date ({pushed!r})") if when.tzinfo is None: return Signal.unassessable(f"last-push date has no timezone ({pushed!r})") days = (datetime.now(UTC) - when).days if days < 0: return Signal.clean("pushed today (repository clock is ahead)", pushed) if days > STALE_DAYS: return Signal.flagged(f"no push in {human_days(days)} (last {pushed[:10]})", pushed) return Signal.clean(f"pushed {human_days(days)} ago", pushed) def _concentration_signal(meta: dict) -> Signal: """Who can actually ship code into this package. Branches on provenance because CI publishing moves the trust boundary off the registry ACL and onto repository merge rights, which GitHub does not expose to third parties (`/collaborators` returns 403). "1 maintainer, low risk" for a CI-published package would be confidently wrong. """ if meta["provenance"]: return Signal.unassessable( "publishes from CI with provenance, so the effective publisher set is whoever " "can merge to the release branch — not externally observable", meta["maintainers"], ) if not meta["maintainers"]: return Signal.unassessable("registry lists no maintainers", []) humans, bots = meta["human_maintainers"], meta["automated_maintainers"] if not humans: return Signal.unassessable( "every listed maintainer looks automated; the human population is unknown", meta["maintainers"], ) if len(humans) == 1: # Name the filtering: the bot heuristic is a guess, and it is the step that turns # two listed maintainers into a flag. aside = f"; {', '.join(bots)} looked automated and was excluded" if bots else "" return Signal.flagged( f"single human publisher ({humans[0]}) of {len(meta['maintainers'])} listed{aside}", meta["maintainers"], ) return Signal.clean(f"{len(humans)} human publishers hold publish rights", meta["maintainers"]) def _fill(dep: Dependency, criteria: tuple[str, ...], signal: Signal) -> None: for criterion in criteria: dep.signals[criterion] = signal # ------------------------------------------------------------------- enrichment def enrich_repo(http: sources.Http, dep: Dependency, token: str | None) -> None: """Repository-derived signals, shared by every ecosystem.""" if dep.repo is None: _fill( dep, REPO_CRITERIA, Signal.unassessable("no source repository could be resolved for this package"), ) _fill(dep, TIER_SCORECARD, Signal.unassessable("no source repository to score")) return try: repo = sources.github_repo(http, dep.repo, token) except sources.RepoIdentifierError as exc: _fill(dep, REPO_CRITERIA, Signal.unassessable(str(exc), dep.repo)) _fill(dep, TIER_SCORECARD, Signal.unassessable(str(exc), dep.repo)) return except sources.Unavailable as exc: _fill(dep, REPO_CRITERIA, Signal.unassessable(_reason(exc, "the GitHub API"), str(exc))) _fill(dep, TIER_SCORECARD, Signal.unassessable(_reason(exc, "the GitHub API"), str(exc))) return dep.signals["archived"] = ( Signal.flagged("repository is archived", True) if repo["archived"] else Signal.clean("repository is active", False) ) dep.signals["staleness"] = _staleness_signal(repo["pushed_at"]) policy = repo["security_policy"] dep.signals["security_policy"] = ( Signal.unassessable("could not determine whether a security policy is published") if policy is None else Signal.clean( "publishes a security policy" if policy else "no security policy found", policy ) ) enrich_scorecard(http, dep) def enrich_scorecard(http: sources.Http, dep: Dependency) -> None: """OpenSSF Scorecard individual checks. Never the aggregate score.""" try: scores = sources.scorecard_checks(http, dep.repo or "") except sources.Unavailable as exc: reason = ( "OpenSSF Scorecard has no report for this repository" if isinstance(exc, sources.NotFound) else _reason(exc, "the Scorecard API") ) _fill(dep, TIER_SCORECARD, Signal.unassessable(reason)) return for check, (criterion, threshold) in sources.SCORECARD_CHECKS.items(): score = scores.get(check) if score is None: dep.signals[criterion] = Signal.unassessable( f"Scorecard did not report {check} for this repository" ) elif score < 0: # Scorecard's own "could not evaluate", commonly for want of admin access. dep.signals[criterion] = Signal.unassessable( f"Scorecard could not evaluate {check} (score -1)" ) elif threshold is not None and score < threshold: dep.signals[criterion] = Signal.flagged( f"{check} scores {score}/10 (below {threshold})", score ) else: dep.signals[criterion] = Signal.clean(f"{check} scores {score}/10", score) def enrich_npm(http: sources.Http, dep: Dependency, token: str | None) -> None: """npm publishes the richest metadata of the three ecosystems.""" try: meta = sources.npm_metadata(http, dep.name) except sources.Unavailable as exc: reason = _reason(exc, "the npm registry") _fill( dep, ("deprecated", "publisher_concentration", "install_script"), Signal.unassessable(reason), ) dep.signals["provenance"] = Signal.unassessable(reason) else: dep.signals["deprecated"] = ( Signal.flagged( f"deprecated by its maintainers: {meta['deprecated']}", meta["deprecated"] ) if meta["deprecated"] else Signal.clean("not deprecated", False) ) dep.signals["publisher_concentration"] = _concentration_signal(meta) dep.signals["install_script"] = ( Signal.flagged( "runs an install script; `npm ci --ignore-scripts` prevents execution", True ) if meta["has_install_script"] else Signal.clean("no install script", False) ) dep.signals["provenance"] = Signal.clean( "publishes with provenance" if meta["provenance"] else "no publish provenance", meta["provenance"], ) try: dep.signals["downloads"] = _download_signal(sources.npm_downloads(http, dep.name)) except sources.Unavailable as exc: dep.signals["downloads"] = Signal.unassessable( _reason(exc, "the npm download API"), str(exc) ) enrich_repo(http, dep, token) def _download_signal(count: int) -> Signal: """Download volume, measured and never flagged. A floor of 1,000/week flagged nothing across 141 dependencies in three real projects, so as a detector it did no work. As context it does a great deal: a single-publisher package at 450M downloads/week is a different proposition from one at 126K, and that belongs beside the finding rather than as a finding of its own. """ scale = "low" if count < LOW_DOWNLOADS_PER_WEEK else "normal" return Signal.clean(f"{count:,} downloads/week ({scale} volume)", count) def enrich_pypi(http: sources.Http, dep: Dependency, token: str | None) -> None: """PyPI publishes no upload ACL and disables its download counters.""" dep.signals["publisher_concentration"] = Signal.unassessable( "PyPI publishes no upload ACL, so who can publish this package is not observable" ) dep.signals["downloads"] = Signal.unassessable( "PyPI's download counters are disabled and return -1" ) dep.signals["provenance"] = Signal.unassessable( "PyPI publish attestations are not read by this collector" ) dep.signals["install_script"] = Signal.unassessable( "install-time execution depends on whether a wheel or an sdist is installed, which " "this collector does not determine" ) try: meta = sources.pypi_metadata(http, dep.name, dep.version) except sources.Unavailable as exc: dep.signals["deprecated"] = Signal.unassessable(_reason(exc, "PyPI"), str(exc)) else: dep.signals["deprecated"] = _yank_signal(meta, dep) enrich_repo(http, dep, token) def _yank_signal(meta: dict, dep: Dependency) -> Signal: """Yank status is per release, so it must be read for the version in play.""" if meta["yanked"] is None: return Signal.unassessable( f"PyPI does not list release {dep.version} for this package, so its yank " f"status is unknown" ) if meta["yanked"]: return Signal.flagged( f"release {meta['yank_version']} is yanked: " f"{meta['yanked_reason'] or 'no reason given'}", True, ) return Signal.clean(f"release {meta['yank_version']} is not yanked", False) def enrich_go(http: sources.Http, dep: Dependency, token: str | None) -> None: """Go has no registry: a module path is a VCS path, so there is no ACL to read.""" dep.signals["publisher_concentration"] = Signal.unassessable( "Go has no registry ACL; publishing is repository write access, which GitHub does " "not expose to third parties" ) dep.signals["downloads"] = Signal.unassessable("Go has no download-count concept") dep.signals["provenance"] = Signal.unassessable( "Go has no publish-provenance concept; module integrity comes from the checksum " "database instead" ) dep.signals["deprecated"] = Signal.unassessable( "module deprecation is declared in the module's own go.mod, which is not read" ) # Not a gap: the Go toolchain has no install-time script hook, so the risk is absent. dep.signals["install_script"] = Signal.clean( "Go modules have no install-time script execution", False ) enrich_repo(http, dep, token) ENRICHERS = {"npm": enrich_npm, "PyPI": enrich_pypi, "Go": enrich_go} # ----------------------------------------------------------------------- resolve def resolve_from_registry(http: sources.Http, deps: list[Dependency]) -> None: """Establish existence, a version, and a repository before querying OSV. Existence matters most: an empty OSV answer only means "no advisories" for a package that is really published. Version resolution matters because "every advisory ever recorded against tornado" is close to useless next to "advisories affecting the release you would install". Repository resolution needs a concrete version, so without this step every unpinned dependency lost its repository signals. """ for dep in deps: meta = _registry_metadata(http, dep) if meta is None: continue dep.exists = True if not dep.version and meta.get("latest"): dep.version = meta["latest"] dep.version_source = "latest-release" if not dep.repo and meta.get("repository"): dep.repo = meta["repository"] def _registry_metadata(http: sources.Http, dep: Dependency) -> dict | None: try: if dep.ecosystem == "npm": return sources.npm_metadata(http, dep.name) if dep.ecosystem == "PyPI": return sources.pypi_metadata(http, dep.name, dep.version) if dep.ecosystem == "Go": # The proxy answering is the existence proof — so the proxy must be asked. # An earlier version asserted this in a comment while never making the # request, and nonexistent modules read as assessed-clean. sources.go_module_latest(http, dep.name) dep.exists = True except sources.NotFound: dep.exists = False return None except sources.Unavailable: return None return None def resolve_go_repos(http: sources.Http, deps: list[Dependency]) -> None: """Map Go module paths to repositories, preferring deps.dev over the path itself. Vanity paths are not VCS paths: truncating go.opentelemetry.io/otel to three segments yields a host that does not exist, and the fallback must not then be stated as fact. """ for dep in deps: if dep.ecosystem != "Go" or dep.repo: continue try: dep.repo = sources.depsdev_repo(http, dep.ecosystem, dep.name, dep.version) except sources.Unavailable: parts = dep.name.split("/") guess = "/".join(parts[:3]) if len(parts) >= 3 else dep.name dep.repo = guess if guess.startswith(("github.com/", "gitlab.com/")) else None def resolve_repos(http: sources.Http, deps: list[Dependency]) -> None: for dep in deps: if dep.repo or not dep.version: continue try: dep.repo = sources.depsdev_repo(http, dep.ecosystem, dep.name, dep.version) except sources.Unavailable: continue # --------------------------------------------------------------------------- run def _advisory_map( http: sources.Http, deps: list[Dependency], notes: list[str] ) -> dict[str, list[str]]: queries = [(d.ecosystem, d.name, d.version) for d in deps] try: ids_per_query = sources.osv_advisories(http, queries) except sources.Unavailable as exc: notes.append(f"OSV was unreachable ({exc}); advisory coverage is zero.") return {} return {dep.key: ids for dep, ids in zip(deps, ids_per_query, strict=True)} def _locked_beyond_direct( project: Path, deps: list[Dependency] ) -> tuple[dict[tuple[str, str, str], bool | None], list[Unverifiable], dict, list[str], list[str]]: """Lockfile-resolved (ecosystem, name, version) triples that are not direct deps. Returns the attested triples with their dev-only markers, the unverifiable entries with reasons, the ledger (distinct lockfile triples, and how many were excluded as direct-covered), the lockfiles read, and notes. Exclusion is keyed on the full triple, never on the name: the direct sweep checks a direct dependency only at its own resolved version, so a nested copy of the same package pinned at another version by some other dependency is still this sweep's responsibility. A name-keyed exclusion silently dropped exactly that copy — a genuinely installed, possibly vulnerable version checked by neither sweep while the counts still balanced. """ notes: list[str] = [] gathered: list[LockedPackage] = [] unverifiable: list[Unverifiable] = [] lock_sources: list[str] = [] for packages, unattested, source, note in ( _npm_all_locked(project), _uv_all_locked(project), _go_indirect(project), ): gathered.extend(packages) unverifiable.extend(unattested) if source: lock_sources.append(source) if note: notes.append(note) direct_triples = {(d.ecosystem, d.name, d.version) for d in deps if d.version} # Counted from the lock readers' output, before any exclusion or bucketing, so a # triple dropped without landing in a named bucket breaks validation instead of # vanishing while the remaining counts reconcile among themselves. all_triples = {(e, n, v) for e, n, v, _ in gathered} all_triples.update((e["ecosystem"], e["name"], e["version"]) for e in unverifiable) merged: dict[tuple[str, str, str], bool | None] = {} for eco, name, version, dev in gathered: key = (eco, name, version) if key in direct_triples: continue # npm hoists one package into several paths, dev-only in one and runtime in # another; runtime wins, matching discover()'s rule for duplicate declarations. if key not in merged or dev is False: merged[key] = dev deduped = _dedup_unverifiable(unverifiable, direct_triples, set(merged)) ledger = { "lockfile_entries": len(all_triples), "excluded_direct": len(all_triples & direct_triples), } return merged, deduped, ledger, lock_sources, notes def _dedup_unverifiable( unverifiable: list[Unverifiable], direct_triples: set[tuple[str, str, str]], seen: set[tuple[str, str, str]], ) -> list[Unverifiable]: """Drop unverifiable entries that duplicate a direct triple or an attested triple.""" out = [] for entry in unverifiable: key = (entry["ecosystem"], entry["name"], entry["version"]) if key in direct_triples or key in seen: continue seen.add(key) out.append(entry) return out def _attest_go_modules( http: sources.Http, merged: dict[tuple[str, str, str], bool | None], unverifiable: list[Unverifiable], ) -> None: """Split Go entries by whether the module proxy resolves them. go.mod carries no integrity data for its indirect list, so existence must be measured per module; without this, a typo'd or private module path reads as assessed-clean the moment OSV has nothing recorded against it. """ for eco, name, version in sorted(merged): if eco != "Go": continue try: sources.go_module_latest(http, name) except sources.NotFound: reason = "the Go module proxy has no such module" except sources.Unavailable: reason = "the Go module proxy did not answer, so existence is unestablished" else: continue del merged[(eco, name, version)] unverifiable.append({"ecosystem": eco, "name": name, "version": version, "reason": reason}) def sweep_transitive( http: sources.Http, project: Path, deps: list[Dependency] ) -> tuple[dict, list[str]]: """Check every attested lockfile package beyond the direct set for advisories. Advisories only: no other criterion is assessed at this depth, and the report says so. The existence rule still holds — an empty advisory answer may only read as clean for a package known to exist — so only attested entries are queried: npm entries with a registry integrity hash, uv.lock entries from a registry source, and Go modules the module proxy resolves. Everything else is reported as unverifiable with its reason, never as clean. Returns: The transitive accounting for the artifact, and notes for the report. """ merged, unverifiable, ledger, lock_sources, notes = _locked_beyond_direct(project, deps) empty = { "examined": False, "reason": None, "sources": [], "total": 0, "checked": 0, "lockfile_entries": 0, "excluded_direct": 0, "unverifiable": [], } if not lock_sources: return { **empty, "reason": "no lockfile resolves the transitive tree (package-lock.json, " "uv.lock, or a go 1.17+ go.mod)", "flagged": [], }, notes _attest_go_modules(http, merged, unverifiable) triples = sorted(merged) accounted = { **empty, "examined": True, "sources": lock_sources, "total": len(triples) + len(unverifiable), **ledger, "unverifiable": sorted(unverifiable, key=lambda e: (e["ecosystem"], e["name"])), } if not triples: return {**accounted, "flagged": []}, notes try: ids_per_query = sources.osv_advisories(http, [(e, n, v) for e, n, v in triples]) except sources.Unavailable as exc: notes.append( f"OSV was unreachable for the transitive sweep ({exc}); transitive advisory " f"coverage is zero." ) return {**accounted, "reason": "OSV was unreachable", "flagged": []}, notes flagged = [ {"ecosystem": e, "name": n, "version": v, "dev": merged[(e, n, v)], "advisories": ids} for (e, n, v), ids in zip(triples, ids_per_query, strict=True) if ids ] return {**accounted, "checked": len(triples), "flagged": flagged}, notes def _cross_check_pip_audit(deps: list[Dependency], found: dict[str, list[str]]) -> str | None: """Compare OSV's PyPI verdicts against pip-audit, when pip-audit is installed. The versions checked are whatever the collector resolved — a lockfile's, a pin's, or the latest release. Calling them "pinned" claimed the project chose them when it may not have, so the note says "resolved" and the version-source caveats stay in force. """ resolved = [d for d in deps if d.ecosystem == "PyPI" and d.version] if not resolved: return None with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as handle: for dep in resolved: handle.write(f"{dep.name}=={dep.version}\n") path = Path(handle.name) try: flagged_by_tool = {normalize_pypi_name(n) for n in sources.pip_audit_vulnerable(path)} except sources.Unavailable as exc: return f"pip-audit is installed but did not produce a cross-check ({exc})." finally: path.unlink(missing_ok=True) osv_flagged = {d.name for d in resolved if found.get(d.key)} only_tool = sorted(flagged_by_tool - osv_flagged) only_osv = sorted(osv_flagged - flagged_by_tool) if not only_tool and not only_osv: return ( f"pip-audit agreed with OSV on all {len(resolved)} Python dependencies at " f"their resolved versions." ) parts = [] if only_tool: parts.append(f"pip-audit alone flagged {', '.join(only_tool)}") if only_osv: parts.append(f"OSV alone flagged {', '.join(only_osv)}") return "Advisory databases disagree, which is itself worth knowing: " + "; ".join(parts) + "." def _tooling_notes(deps: list[Dependency], found: dict[str, list[str]]) -> list[str]: tools = sources.detect_tools() present = sorted(name for name, ok in tools.items() if ok) absent = sorted(name for name, ok in tools.items() if not ok) notes = [ f"Optional tooling detected: {', '.join(present) or 'none'}. " f"Not installed, so not used: {', '.join(absent) or 'none'}." ] if tools.get("pip-audit"): cross = _cross_check_pip_audit(deps, found) if cross: notes.append(cross) return notes # Recognised but unread; each produces a note so the fallback to pins or # latest-release is disclosed where the reader will see it. UNREAD_LOCKFILES = ("yarn.lock", "pnpm-lock.yaml", "poetry.lock") def _unread_lockfile_notes(project: Path) -> list[str]: return [ f"{name} is present but not read: direct-dependency versions fall back to " f"manifest pins or the latest release, and its transitive tree was not examined." for name in UNREAD_LOCKFILES if (project / name).exists() ] MANIFEST_NAMES = ( "package.json", "package-lock.json", "npm-shrinkwrap.json", "pyproject.toml", "uv.lock", "go.mod", ) def _git_commit(project: Path) -> str | None: """The scanned project's own HEAD commit, for reproducibility. Reads the target's git metadata, never a dependency's source. Returns None when the target is not a git checkout, which is the normal case for an extracted tarball. """ git_dir = project / ".git" head = git_dir / "HEAD" if not head.exists(): return None # The target's .git contents are untrusted input: a crafted HEAD can point outside # .git (`ref: ../../etc/passwd`) or hold bytes that are not UTF-8. Both degrade to # None rather than leaking file content into the report or aborting the run — # UnicodeDecodeError is a ValueError, which the original OSError guard missed. try: content = head.read_text(encoding="utf-8").strip() if content.startswith("ref: "): ref = (git_dir / content.removeprefix("ref: ")).resolve() if not ref.is_relative_to(git_dir.resolve()): return None content = ref.read_text(encoding="utf-8").strip() if ref.exists() else "" return content[:12] or None except (OSError, ValueError): return None def scan_metadata(project: Path) -> dict: """What was examined, so a reader knows whose dependency tree this describes. A report that says "43 direct dependencies" without naming the subject is ambiguous between the reader's project and someone else's, and a reader who assumes wrongly acts on recommendations they have no authority over. """ manifests = [name for name in MANIFEST_NAMES if (project / name).exists()] manifests += [p.name for p in sorted(project.glob("requirements*.txt"))] return { "subject": project.resolve().name, "path": str(project), "commit": _git_commit(project), "manifests": manifests, "scanned_at": datetime.now(UTC).isoformat(timespec="seconds"), } def _repo_sharing_note(deps: list[Dependency]) -> str | None: """Warn when repository-level findings will appear more times than they occur. Archived, maintenance activity, security policy and every Scorecard check describe the source repository, not the package. Monorepos publish many packages from one repository, so a single fact becomes several findings: six of axios's `@rollup/*` dependencies share github.com/rollup/plugins, and all six carried the same checked-in-binaries score. """ repos = [d.repo for d in deps if d.repo] if not repos: return None distinct = len(set(repos)) if distinct == len(repos): return None shared = sorted({r for r in repos if repos.count(r) > 1}) return ( f"Repository-level criteria (archived, maintenance activity, security policy, and " f"the Scorecard checks) describe the source repository rather than the package. " f"{len(repos)} dependencies resolve to {distinct} distinct repositories, so one " f"repository can appear as several findings. Shared: {', '.join(shared[:5])}" + (" ..." if len(shared) > 5 else "") ) def _version_notes(deps: list[Dependency]) -> list[str]: notes = [] for source, wording in ( ( "latest-release", "are specified as a version range with no lockfile, so their advisories were " "matched against the current latest release rather than against what this " "project installs — commit a lockfile for an exact answer", ), ( "unresolved", "had no resolvable version, so their advisory results are historical for the " "package rather than version-matched", ), ( "go-mod-minimum", "come from go.mod, which records a minimum version; module-version selection " "may build something higher", ), ): named = sorted(d.name for d in deps if d.version_source == source) if named: notes.append( f"{len(named)} dependencies {wording}: {', '.join(named[:8])}" + (" ..." if len(named) > 8 else "") ) return notes def collect(project: Path, cache: Path, offline: bool) -> dict: """Collect every signal for every direct dependency and assemble the artifact.""" deps, notes = discover(project) if not deps: hints = (" " + " ".join(notes)) if notes else "" raise SystemExit( f"error: no direct dependencies found under {project}. A run that assesses " f"nothing must not report that nothing is wrong.{hints}" ) token = sources.gh_token() http = sources.Http(cache, offline=offline, auth_marker="gh" if token else "anon") if http.cache_owner_caveat: notes.append(http.cache_owner_caveat) if token is None: notes.append( "gh is not authenticated. GitHub allows 60 requests/hour unauthenticated " "against 5000 authenticated, so repository signals may be unassessable." ) notes.extend(_unread_lockfile_notes(project)) # The one choke point for registry identity: a dependency that resolves from # somewhere other than its public registry is never looked up by name — a # same-named public package's advisories, publishers, and deprecation belong to # code this project does not install. Every criterion is unassessable, with the # source as the reason, and the dependency stays in the report and its coverage. registry_deps = [d for d in deps if not d.non_registry_reason] for dep in deps: if dep.non_registry_reason: # One shared reason across all 13 criteria, with the per-dependency source # carried in the signal's value and in the Method note. Embedding the source # in the reason gave every dependency a unique string, which defeated the # report's grouping: 13 criteria x 7 workspace packages produced 91 # near-identical bullets, scaling linearly with monorepo size. _fill( dep, CRITERIA, Signal.unassessable(NON_REGISTRY_REASON, dep.non_registry_reason), ) resolve_from_registry(http, registry_deps) found = _advisory_map(http, registry_deps, notes) transitive, transitive_notes = sweep_transitive(http, project, deps) notes.extend(transitive_notes) resolve_repos(http, registry_deps) resolve_go_repos(http, registry_deps) for dep in registry_deps: dep.signals["advisories"] = _advisory_signal(dep, found) ENRICHERS[dep.ecosystem](http, dep, token) sources.polite_pause() notes.extend(_version_notes(registry_deps)) sharing = _repo_sharing_note(deps) if sharing: notes.append(sharing) notes.extend(_tooling_notes(registry_deps, found)) notes.append(_scope_note(transitive)) notes.append(_cache_note(http)) return to_json(deps, scan_metadata(project), notes, transitive) def _scope_note(transitive: dict) -> str: """State what depth each claim in the report reaches. The umbrella-package caveat matters most when the transitive tree went unexamined: advisories attach to the package that ships the affected code, so a clean direct tree says little about what it pulls in. """ unverifiable = len(transitive.get("unverifiable") or []) if transitive["examined"] and transitive["checked"] + unverifiable == transitive["total"]: if transitive["total"] == 0: return ( f"The lockfile ({', '.join(transitive['sources'])}) resolves no packages " f"beyond the direct dependencies." ) caveat = ( f" {unverifiable} lockfile entries could not be verified against a public " f"registry and were not checked." if unverifiable else "" ) return ( f"Every criterion except advisories applies to direct dependencies only. The " f"{transitive['checked']} registry-verified packages resolved by " f"{', '.join(transitive['sources'])} were checked for known advisories at " f"their locked versions, and for nothing else." + caveat ) reason = transitive["reason"] or "the transitive sweep did not complete" return ( "Direct dependencies only. Advisories attach to the package that ships the " "affected code, so an umbrella package can look clean while its components are " "not — rails 5.0.0 reports 0 advisories where actionpack 5.0.0 reports 10. " f"Transitive dependencies were not examined: {reason}." ) def _cache_note(http: sources.Http) -> str: stats = http.stats oldest = http.oldest_hit_seconds / 3600 note = ( f"HTTP sources: {stats['fetched']} fetched, {stats['hits']} served from cache " f"(oldest {oldest:.1f}h old), {stats['stale']} refetched as stale, " f"{stats['offline_misses']} unavailable offline, {stats['errors']} errors." ) if http.offline and http.oldest_hit_seconds > sources.CACHE_MAX_AGE_SECONDS: note += ( " This offline run served entries past the freshness bound; " "repository-derived signals such as maintenance activity describe the state " "at fetch time, not today." ) return note def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("project", type=Path, help="path to the project to audit") parser.add_argument("--json", type=Path, help="write the artifact here (default: stdout)") parser.add_argument( "--cache", type=Path, # Outside the working directory: the target of an audit is somebody else's # repository and this should not leave a directory in it. default=Path(tempfile.gettempdir()) / "supply-chain-risk-auditor-cache", help="HTTP cache directory (default: a stable path under the system temp dir)", ) parser.add_argument("--offline", action="store_true", help="use only cached responses") args = parser.parse_args() if not args.project.is_dir(): raise SystemExit(f"error: {args.project} is not a directory") try: artifact = collect(args.project, args.cache, args.offline) except ReconciliationError as exc: # A refusal, not a crash: say what is wrong rather than printing a traceback. raise SystemExit(f"error: this run cannot be reported: {exc}") from exc text = json.dumps(artifact, indent=2, sort_keys=True) if args.json: args.json.write_text(text + "\n", encoding="utf-8") print(f"wrote {args.json} ({artifact['coverage']['total_dependencies']} dependencies)") else: print(text) return 0 if __name__ == "__main__": sys.exit(main()) ``` ## scripts/model.py ```python """Tri-state signal model and the checks that keep a report honest. Unavailable data is never evidence of risk, and an absent measurement is never a clean verdict. Both directions matter: an early version of this collector held three 404s proving a package was unpublished and still reported "no advisories recorded" as a clean result, because an empty answer from a vulnerability database was mistaken for evidence. So `Signal.clean` demands a datum, exactly as `Signal.flagged` and `Signal.unassessable` demand a reason. A clean state with nothing behind it is the shape of that bug. `validate_artifact` is the guard that can actually fail. An earlier `coverage()` compared per-state counts against a total that was computed from the same loop, so it was true by construction and could never fire. """ from __future__ import annotations from dataclasses import asdict, dataclass, field from enum import Enum class State(str, Enum): """What we learned about one criterion for one dependency.""" CLEAN = "assessed_clean" FLAGGED = "assessed_flagged" UNASSESSABLE = "unassessable" @dataclass class Signal: """One criterion's verdict for one dependency, with the datum behind it.""" state: State detail: str value: object = None @classmethod def clean(cls, detail: str, value: object) -> Signal: """A negative finding, which is still a claim and still needs evidence. Raises: ValueError: If no detail is given, or `value` is None. A clean verdict with no supporting datum cannot be distinguished from an unasked question. """ if not detail: raise ValueError("a clean signal must say what was measured") if value is None: raise ValueError( f"a clean signal needs the datum behind it ({detail!r} carries none); " f"use Signal.unassessable when nothing was measured" ) return cls(State.CLEAN, detail, value) @classmethod def flagged(cls, detail: str, value: object) -> Signal: """A positive finding. Raises: ValueError: If no detail is given. """ if not detail: raise ValueError("a flagged signal must say what was flagged") return cls(State.FLAGGED, detail, value) @classmethod def unassessable(cls, reason: str, value: object = None) -> Signal: """Nothing was measured, and this is not a finding. Raises: ValueError: If no reason is given. """ if not reason: raise ValueError("an unassessable signal must say why it could not be assessed") return cls(State.UNASSESSABLE, reason, value) # Comparable across every ecosystem, so only Tier A reaches the headline verdict. TIER_A = ("advisories", "deprecated", "archived", "staleness") # Registry-dependent: npm publishes these, PyPI publishes none of them, Go has no # registry at all. TIER_B = ("publisher_concentration", "install_script") # From OpenSSF Scorecard's individual checks. Never its aggregate score, which measures # adherence to OSS security hygiene and correlates with project size rather than with # takeover risk: p-limit scores 3.8 and chalk 4.6 against lodash's 7.2, while lodash is # the one with a single publisher and 165M weekly downloads. TIER_SCORECARD = ("dangerous_workflow", "token_permissions", "binary_artifacts", "code_review") # Measured and counted, never flagged. # # `provenance` and `security_policy` describe hygiene rather than likelihood of compromise, # and are absent for most packages — flagging them put 5 of 8 dependencies in the findings # table on a trial run and buried the real findings. # # `downloads` earned its place here by measurement: a floor of 1,000/week flagged 0 of 141 # dependencies across three real projects, so it detected nothing. It is retained because # it is the number that distinguishes a single-publisher package at 450M downloads/week # from one at 126K, which is prioritisation context rather than a finding. TIER_INFO = ("provenance", "security_policy", "downloads") CRITERIA = TIER_A + TIER_B + TIER_SCORECARD + TIER_INFO # OpenSSF Scorecard check -> (criterion, flag threshold). The single source of truth for # which Scorecard criteria can flag: the collector scores against the thresholds and the # renderer derives its never-flags set from `threshold is None`. Keeping this knowledge # in one place is what stops the renderer's prose from drifting against the collector's # behaviour — an earlier draft maintained the demoted set by hand in the renderer and # described the two checks that DO flag as "not flagged — poor precision". # # Only the two that name a concrete mechanism flag. Measured against axios's 43 # dependencies, `Token-Permissions` below 6 flagged 23 of 35 and `Code-Review` below 4 # flagged 15 of 39 — two thirds and a third of the tree. Both describe CI configuration # maturity, which tracks project size rather than the likelihood of someone shipping # malicious code, and `Code-Review` largely restates publisher concentration. SCORECARD_CHECKS = { # Scorecard found an actual script-injection or untrusted-checkout pattern. "Dangerous-Workflow": ("dangerous_workflow", 10), # Binaries committed to the repository, which nobody can review. "Binary-Artifacts": ("binary_artifacts", 10), "Token-Permissions": ("token_permissions", None), "Code-Review": ("code_review", None), } @dataclass class Dependency: """A direct dependency of the target project.""" ecosystem: str name: str version: str | None = None # How the version was established, which decides how strong any advisory claim is: # lockfile exactly what the project installs # manifest-pin an exact pin in the manifest # go-mod-minimum go.mod's minimum; module-version selection may pick higher # latest-release the registry's current release, not the project's choice # unresolved nothing; advisory results are historical for the package version_source: str = "unresolved" # True = build-time only, False = ships in the artifact, None = the manifest declares # no distinction. Go is the None case: go.mod has no dev section, so claiming a module # ships to production would assert something the ecosystem never states. dev: bool | None = None repo: str | None = None # True when the package is known to exist in its registry. An empty answer from a # vulnerability database only means "no advisories" if the package is real. exists: bool | None = None # Set when the dependency resolves from somewhere other than its public registry # (file:, workspace:, git, a vendored directory). Registry-keyed data must never be # queried for such a dependency: a same-named public package's advisories, # publishers, and deprecation would be attributed to code the project never # installs. Every criterion becomes unassessable with this reason. non_registry_reason: str | None = None signals: dict[str, Signal] = field(default_factory=dict) @property def key(self) -> str: return f"{self.ecosystem}:{self.name}" def flagged(self) -> list[str]: return [c for c, s in self.signals.items() if s.state is State.FLAGGED] class ReconciliationError(RuntimeError): """The artifact does not account for every dependency and criterion.""" def coverage(deps: list[Dependency], criteria: tuple[str, ...] = CRITERIA) -> dict: """Count each criterion's states, refusing anything that hides a dependency. Args: deps: The dependencies that were collected. criteria: Criterion names every dependency must carry. Returns: A mapping of criterion -> per-state counts, plus the dependency total. Raises: ReconciliationError: If two dependencies share a key, or any dependency's signal keys are not exactly `criteria`. Both are ways a dependency disappears from the report while the totals still look plausible. """ keys = [dep.key for dep in deps] duplicates = sorted({key for key in keys if keys.count(key) > 1}) if duplicates: raise ReconciliationError( f"duplicate dependencies would be counted twice: {', '.join(duplicates)}" ) expected = set(criteria) for dep in deps: present = set(dep.signals) if present != expected: missing = sorted(expected - present) extra = sorted(present - expected) raise ReconciliationError( f"{dep.key} carries the wrong criteria — missing {missing}, unexpected " f"{extra}. A missing criterion reads as a clean one; an unexpected one is " f"counted nowhere and rendered nowhere." ) out = {} for criterion in criteria: counts = {state.value: 0 for state in State} for dep in deps: counts[dep.signals[criterion].state.value] += 1 out[criterion] = counts return {"total_dependencies": len(deps), "criteria": out} def by_ecosystem(deps: list[Dependency]) -> dict[str, int]: counts: dict[str, int] = {} for dep in deps: counts[dep.ecosystem] = counts.get(dep.ecosystem, 0) + 1 return dict(sorted(counts.items())) def assessed(counts: dict[str, int]) -> int: return counts[State.CLEAN.value] + counts[State.FLAGGED.value] def tier_a_was_assessed(artifact: dict) -> bool: """Did the run establish anything at all on the universally-available criteria?""" criteria = artifact["coverage"]["criteria"] return any(assessed(criteria[name]) > 0 for name in TIER_A if name in criteria) def validate_artifact(artifact: dict) -> None: """Refuse an artifact that cannot support a report. This is deliberately a checker that fails when it inspects zero items, which is what the repository's contributing guide asks of anything that counts or filters. Args: artifact: A collector artifact. Raises: ReconciliationError: If the artifact is empty, describes no dependencies, has a coverage row that does not account for every dependency, or measured nothing on Tier A. """ total = artifact.get("coverage", {}).get("total_dependencies") criteria = artifact.get("coverage", {}).get("criteria") or {} if not total: raise ReconciliationError( "artifact describes zero dependencies; a run that assessed nothing must not " "be rendered as a report finding nothing" ) if not criteria: raise ReconciliationError("artifact has no coverage rows") for criterion, counts in criteria.items(): if sum(counts.values()) != total: raise ReconciliationError( f"{criterion}: {counts} does not account for all {total} dependencies" ) if len(artifact.get("dependencies") or []) != total: raise ReconciliationError( f"coverage claims {total} dependencies but " f"{len(artifact.get('dependencies') or [])} are listed" ) if not tier_a_was_assessed(artifact): raise ReconciliationError( "no Tier A criterion was assessable for any dependency, so this run measured " "nothing; every source was unavailable" ) _validate_transitive(artifact) def _validate_transitive(artifact: dict) -> None: """The transitive accounting must exist and must reconcile. Absence is the failure mode: a report with no transitive statement reads as though the tree was covered, which is the old design's absence-means-clean defect at one remove. """ transitive = artifact.get("transitive") if not isinstance(transitive, dict) or "examined" not in transitive: raise ReconciliationError( "artifact carries no transitive accounting, so the report cannot state " "whether the transitive tree was examined" ) if not transitive["examined"]: if not transitive.get("reason"): raise ReconciliationError("an unexamined transitive tree must say why") return _reconcile_transitive_counts(transitive) def _reconcile_transitive_counts(transitive: dict) -> None: total, checked = transitive.get("total", 0), transitive.get("checked", 0) flagged = transitive.get("flagged") unverifiable = transitive.get("unverifiable") if flagged is None: raise ReconciliationError("transitive accounting lists no flagged collection") if unverifiable is None: raise ReconciliationError( "transitive accounting lists no unverifiable collection; entries that " "cannot be verified must be named, not folded into the checked count" ) # A sweep that checked nothing is only accountable with a stated reason (OSV # unreachable); without one, zero must reconcile like any other number — the # earlier `if checked and ...` guard could not fire on the value that matters most. if checked + len(unverifiable) != total and not (checked == 0 and transitive.get("reason")): raise ReconciliationError( f"transitive sweep checked {checked} and lists {len(unverifiable)} " f"unverifiable of {total} resolved; the difference is unaccounted for" ) _reconcile_transitive_ledger(transitive, total) if checked > total: raise ReconciliationError( f"transitive sweep claims {checked} packages checked of {total} resolved" ) if len(flagged) > checked: raise ReconciliationError( f"transitive sweep flags {len(flagged)} packages but checked only {checked}" ) _reconcile_transitive_entries(flagged, unverifiable) def _reconcile_transitive_ledger(transitive: dict, total: int) -> None: """Every lockfile triple must land in a named bucket. `total` is derived from the buckets themselves, so it reconciles by construction even when a triple is silently dropped before bucketing — which is how a nested copy of a direct dependency once vanished from the sweep with the counts still balancing. `lockfile_entries` is counted from the lock readers' output, before any exclusion or bucketing, so a triple dropped after they return breaks this equation instead of disappearing. A drop inside a reader is outside this guard's reach: the readers are the origin of the count, so nothing independent can contradict them. """ lockfile_entries = transitive.get("lockfile_entries") excluded = transitive.get("excluded_direct") if lockfile_entries is None or excluded is None: raise ReconciliationError( "transitive accounting carries no lockfile ledger (lockfile_entries and " "excluded_direct); without it a dropped entry cannot be detected" ) if total + excluded != lockfile_entries: raise ReconciliationError( f"the lockfile resolves {lockfile_entries} distinct packages but only " f"{total} are accounted for after excluding {excluded} direct-covered " f"entries; the difference vanished from the sweep" ) def _reconcile_transitive_entries(flagged: list, unverifiable: list) -> None: for entry in unverifiable: if not entry.get("reason"): raise ReconciliationError(f"unverifiable entry {entry.get('name')!r} carries no reason") for entry in flagged: if not entry.get("advisories"): raise ReconciliationError( f"transitive flag for {entry.get('name')!r} carries no advisory ids" ) def count_flags(artifact: dict) -> dict[str, int]: """Flags per criterion, taken from the dependency list rather than from coverage. The renderer compares this against the coverage table it prints. If a criterion is counted in one place and rendered in the other, the report contradicts itself, and that is how a finding reaches the coverage table and no reader. """ out: dict[str, int] = {} for dep in artifact["dependencies"]: for name, signal in dep["signals"].items(): if signal["state"] == State.FLAGGED.value: out[name] = out.get(name, 0) + 1 return out def to_json(deps: list[Dependency], scan: dict, notes: list[str], transitive: dict) -> dict: """Assemble the machine-readable artifact, refusing one that cannot reconcile. Args: deps: Every direct dependency, with all criteria populated. scan: What was examined — subject, manifests, commit, timestamp. notes: Scope and caveat lines for the report. transitive: Accounting for the lockfile-resolved packages beyond the direct set. """ artifact = { "scan": scan, "target": scan.get("path", ""), "ecosystems": by_ecosystem(deps), "coverage": coverage(deps), "transitive": transitive, "notes": notes, "dependencies": [ { **{k: v for k, v in asdict(dep).items() if k != "signals"}, "flagged": dep.flagged(), "signals": { name: {"state": sig.state.value, "detail": sig.detail, "value": sig.value} for name, sig in dep.signals.items() }, } for dep in sorted(deps, key=lambda d: (d.ecosystem, d.name)) ], } validate_artifact(artifact) return artifact ``` ## scripts/pyproject.toml ```toml [project] name = "supply-chain-risk-auditor-scripts" version = "0.1.0" description = "Direct-dependency supply-chain signal collector and report renderer" requires-python = ">=3.11" # Stdlib only, like every other plugin's scripts/ in this repo. The HTTP client is # urllib; `make python-tests` runs suites with `uv run --no-project`, so anything # third-party would be unimportable there anyway. dependencies = [] [tool.ruff] line-length = 100 [tool.ruff.lint] # C901 is not in the repo's default select list, so the ≤8 complexity limit the # contributing guide states was unenforced here until it was added. extend-select = ["C901"] [tool.ruff.lint.mccabe] max-complexity = 8 ``` ## scripts/render.py ```python #!/usr/bin/env python3 # /// script # requires-python = ">=3.11" # /// """Turn a collector artifact into a Markdown supply-chain report. The failure this format exists to prevent is a reader inferring safety from silence: a criterion that could not be assessed for 84 of 177 dependencies must not read as an all-clear. The weakest coverage figure is therefore in the summary, while the full table sits behind the findings — an earlier version opened with the table and spent a reader's first two screens on the tool's limits before anything actionable. Order after that follows blast radius: what reaches production, then what reaches only the build, then what belongs to somebody else's repository. Nothing here skips quietly. A criterion the artifact lacks, or a flag counted in coverage but rendered in no table, raises — a renderer whose purpose is to surface gaps must not have gaps of its own. The report carries facts only: what was measured, what was flagged, what could not be assessed and why. Interpretive rules — unassessable is not risk, low coverage is not an all-clear — are instructions to the agent adding judgment, and they live in SKILL.md. An earlier draft printed them at the reader, where they read as framing about the workflow rather than information about the dependencies. Usage: uv run render.py findings.json [--out report.md] """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path from model import ( SCORECARD_CHECKS, TIER_A, TIER_B, TIER_INFO, TIER_SCORECARD, ReconciliationError, State, assessed, count_flags, validate_artifact, ) LABELS = { "advisories": "Known advisories", "deprecated": "Deprecated or yanked", "archived": "Repository archived", "staleness": "Maintenance activity", "publisher_concentration": "Publisher concentration", "install_script": "Install-time script execution", "downloads": "Download volume", "dangerous_workflow": "Dangerous CI workflow", "token_permissions": "CI token permissions", "binary_artifacts": "Checked-in binaries", "code_review": "Changes reviewed by a second person", "provenance": "Publish provenance", "security_policy": "Security policy published", } # Coverage is reported per tier because availability differs per tier. Findings are not: # a reader wants one row per dependency listing everything wrong with it, rather than the # same package repeated in a table per criterion group. The split that does matter to a # reader is whether anything can be done about it. FINDING_GROUPS = ( (TIER_A + TIER_B, "Findings"), (TIER_SCORECARD, "Upstream repository and CI hygiene — OpenSSF Scorecard"), ) COVERAGE_TIERS = (("A", TIER_A), ("B", TIER_B), ("scorecard", TIER_SCORECARD)) # Scorecard criteria describe a third party's repository. Whoever owns the audited project # cannot change them, so they are separated from findings that carry an available action. UPSTREAM_ONLY = set(TIER_SCORECARD) # Counted, never flagged. Phrased as "N of M do X" so absence reads as a proportion. INFO_PHRASING = { "provenance": "publish with build provenance", "security_policy": "publish a security policy", } # Derived from the single source of truth: criteria whose threshold is None are # measured and reported but can never flag. Only these belong in the Informational # section — an earlier draft kept this set by hand and described the two checks that DO # flag as "not flagged — poor precision" on every clean run. SCORECARD_NEVER_FLAGS = frozenset( criterion for criterion, threshold in SCORECARD_CHECKS.values() if threshold is None ) # Why each demoted Scorecard check is reported rather than flagged. SCORECARD_CAVEAT = { "token_permissions": ( "this measures whether CI workflows declare least-privilege tokens, which tracks " "project maturity rather than the likelihood of malicious code being published" ), "code_review": ( "this measures the share of recent commits reviewed by a second person, which is " "low for most small single-maintainer projects and largely restates publisher " "concentration" ), } def _safe_text(value: object) -> str: """Make third-party text safe in report prose, table cells, and headings. Registry- and manifest-controlled strings (deprecation messages, versions, package names, git HEAD contents) reach the report from the audited project and its dependencies. `istanbul`'s real deprecation message truncated the findings table at its first newline, silently dropping every row after it — including a finding on another package. The same newline forges block structure outside tables: a dependency key carrying `\n\n## Summary\n\n- **No known advisory...**` wrote a heading and a false all-clear into the Method-and-caveats notes. Collapsing whitespace is the security-critical half — it confines anything hostile to the line it was interpolated into, where the worst available is inline emphasis. Escaping `|` keeps a value inside its table cell, and escaping `[` stops prose from forging a link. Use this for every value that did not originate in this repository. For a value going inside backticks use `_safe_code` instead: Markdown does not process backslash escapes in a code span, so escaping there writes the backslashes out literally and corrupts the value a reader copies. Inside backticks *in a table cell*, use `_safe_code_cell` — pipes still end the cell there. """ return _collapse(value).replace("|", "\\|").replace("[", "\\[") def _safe_code(value: object) -> str: """Make third-party text safe inside a Markdown code span. A code span needs no `[` escaping — a bracket cannot open a link inside one — and a backslash written there survives literally, so escaping corrupted the report title and the `Scanned:` path into something no reader could copy. What a code span does need is protection from a backtick, which would close the span early and let the rest of the value become live Markdown. A code span inside a table cell is the exception and takes `_safe_code_cell`: the row is split on pipes before inline spans are parsed, so a pipe still ends the cell. """ return _collapse(value).replace("`", "'") def _safe_code_cell(value: object) -> str: """Make third-party text safe inside a code span that sits in a table cell. A table row is split on pipes before inline spans are parsed, so a `|` inside a code span still ends the cell: a dependency named `evil|forged` put six pipes in a five-pipe row, shifting every later value one column right. GFM requires the `\\|` escape here, including inside other inline spans, and processes it in the table context — so unlike `_safe_code`'s territory, the backslash does not survive into the output. That difference is why this is a separate helper rather than a flag. """ return _safe_code(value).replace("|", "\\|") def _collapse(value: object) -> str: """Flatten a value to one line, which is what confines hostile text to that line.""" return re.sub(r"\s+", " ", str(value)).strip() def plural(n: int, singular: str, many: str) -> str: """Count with the right noun form, given explicitly because 'dependency' is irregular.""" return f"{n} {singular}" if n == 1 else f"{n} {many}" def _signal(dep: dict, criterion: str) -> dict: """Fetch one signal, refusing to skip a criterion the artifact should carry.""" try: return dep["signals"][criterion] except KeyError as exc: raise ReconciliationError( f"{dep['name']} carries no {criterion!r} signal; the renderer would drop it " f"silently and the report would read as though it had been assessed" ) from exc def _flagged(dep: dict, criteria: tuple[str, ...]) -> list[tuple[str, dict]]: out = [] for criterion in criteria: signal = _signal(dep, criterion) if signal["state"] == State.FLAGGED.value: out.append((criterion, signal)) return out def coverage_table(artifact: dict) -> list[str]: """Coverage grouped by tier, which is the distinction that matters. The artifact serialises with sorted keys, so iterating it directly loses the grouping. """ total = artifact["coverage"]["total_dependencies"] criteria = artifact["coverage"]["criteria"] rows = ["| Criterion | Tier | Assessed | Flagged | Not assessable |", "|---|---|---|---|---|"] ordered = [(tier, name) for tier, names in COVERAGE_TIERS for name in names] ordered += [("info", name) for name in TIER_INFO] seen = {name for _, name in ordered} missing = sorted(set(criteria) - seen) if missing: raise ReconciliationError( f"the artifact carries criteria this renderer knows nothing about: {missing}. " f"They would be counted in coverage and rendered in no table." ) for tier, name in ordered: counts = criteria.get(name) if counts is None: raise ReconciliationError(f"coverage has no row for {name!r}") rows.append( f"| {LABELS.get(name, name)} | {tier} | {assessed(counts)}/{total} | " f"{counts[State.FLAGGED.value]} | {counts[State.UNASSESSABLE.value]} |" ) return rows def _version_label(dep: dict) -> str: """Mark how the version was established; the finding means different things by source.""" version = dep["version"] if not version: return "unresolved" suffix = { "latest-release": " (latest, not the project's pin)", "go-mod-minimum": " (go.mod minimum)", "manifest-pin": "", "lockfile": "", }.get(dep.get("version_source", ""), "") return f"{version}{suffix}" def _volume(dep: dict) -> str: """Download volume as prioritisation context beside a finding.""" value = _signal(dep, "downloads").get("value") return f"{value:,}/wk" if isinstance(value, int) else "—" def _finding_rows(hits: list[tuple[dict, list]]) -> list[str]: rows = ["| Dependency | Version | Weekly downloads | Findings |", "|---|---|---|---|"] for dep, flags in sorted(hits, key=lambda x: (-len(x[1]), x[0]["name"])): detail = _safe_text("; ".join(s["detail"] for _, s in flags)) rows.append( f"| `{_safe_code_cell(dep['name'])}` | {_safe_text(_version_label(dep))} | {_volume(dep)} | {detail} |" ) return rows def findings_section(artifact: dict, criteria: tuple[str, ...], heading: str) -> list[str]: """Findings for one tier, split by whether the dependency reaches production. The split is the first thing a reader needs. A compromised build-time dependency reaches the build host; a compromised production dependency reaches users. Presenting both at equal weight — as an earlier version did, with a nine-year-abandoned build tool listed beside a shipped package — makes the report harder to act on than a bare list. """ out = [f"## {heading}", ""] if set(criteria) & UPSTREAM_ONLY: out += [ "These criteria describe each dependency's own repository, not the audited", "project. Remediation, where any exists, is upstream.", "", ] hits = [(d, f) for d in artifact["dependencies"] if (f := _flagged(d, criteria))] if not hits: rows = artifact["coverage"]["criteria"] if all(assessed(rows[c]) == 0 for c in criteria if c in rows): out += ["No criterion in this tier was assessable for any dependency.", ""] else: out += ["No dependency was flagged on these criteria.", ""] return out groups = ( ( [(d, f) for d, f in hits if d.get("dev") is False], "Reaches production", "", ), ( [(d, f) for d, f in hits if d.get("dev") is True], "Build-time only", "Declared as development dependencies. A compromise here reaches build and CI " "hosts rather than users of the shipped artifact.", ), ( [(d, f) for d, f in hits if d.get("dev") is None], "Production or build-time not declared", "These manifests draw no distinction between runtime and development " "dependencies — go.mod has no dev section — so whether these reach a shipped " "artifact cannot be read from the manifest.", ), ) for rows, heading, preamble in groups: if not rows: continue out += [f"### {heading}", ""] if preamble: out += [preamble, ""] out += _finding_rows(rows) out.append("") return out def _transitive_summary(transitive: dict) -> str: """One summary line stating what the transitive sweep did or could not do.""" if not transitive["examined"]: return f"- Transitive dependencies were **not examined**: {transitive['reason']}." if transitive["total"] == 0: return "- The lockfile resolves no packages beyond the direct dependencies." unverifiable = transitive.get("unverifiable") or [] checkable = transitive["total"] - len(unverifiable) caveat = ( f" {plural(len(unverifiable), 'entry was', 'entries were')} unverifiable against " f"a public registry and not checked (see Transitive advisories)." if unverifiable else "" ) if checkable and transitive["checked"] < checkable: return ( f"- {checkable} transitive packages were resolved but **could not " f"be checked**: {transitive['reason'] or 'the sweep did not complete'}." ) if transitive["flagged"]: return ( f"- **{plural(len(transitive['flagged']), 'transitive package carries', 'transitive packages carry')} " f"known advisories** at the locked versions — see Transitive advisories." + caveat ) if not checkable: return ( f"- All {transitive['total']} packages beyond the direct set were " f"unverifiable against a public registry; none was checked for advisories." ) return ( f"- No known advisory affects any of the {checkable} registry-verified " f"transitive packages at their locked versions." + caveat ) def transitive_section(artifact: dict) -> list[str]: """Advisory results for the lockfile-resolved packages beyond the direct set. Advisories only, and the section says so: leaving the depth to be inferred is how a reader mistakes "the direct tree is clean" for "the tree is clean". Only entries attested against a public registry are checked; the rest are named with reasons, because an empty advisory answer about an unverifiable entry proves nothing. """ transitive = artifact["transitive"] out = ["## Transitive advisories", ""] if not transitive["examined"]: out += [ f"Not examined: {transitive['reason']}. Commit a lockfile to close this gap.", "", ] return out src = ", ".join(f"`{s}`" for s in transitive["sources"]) if transitive["total"] == 0: out += [f"{src} resolves no packages beyond the direct dependencies.", ""] return out unverifiable = transitive.get("unverifiable") or [] checkable = transitive["total"] - len(unverifiable) if checkable and transitive["checked"] < checkable: out += [ f"{checkable} packages beyond the direct set are resolved by {src}, " f"but none was checked: " f"{transitive['reason'] or 'the sweep did not complete'}.", "", ] return out if not transitive["flagged"]: if checkable: out += [ f"No known advisory affects any of the {checkable} registry-verified " f"packages beyond the direct set (resolved by {src}), at the locked " f"versions. Only advisories were checked at this depth; every other " f"criterion in this report describes direct dependencies only.", "", ] out += _unverifiable_lines(transitive) return out out += [ f"{plural(len(transitive['flagged']), 'package', 'packages')} of the " f"{checkable} registry-verified packages beyond the direct set (resolved by " f"{src}) {'carries' if len(transitive['flagged']) == 1 else 'carry'} known " f"advisories at the locked versions. Only advisories were checked at this depth.", "", "| Package | Version | Reaches | Advisories |", "|---|---|---|---|", ] for entry in sorted(transitive["flagged"], key=lambda e: (e["ecosystem"], e["name"])): reaches = {True: "build-time only", False: "production"}.get( entry.get("dev"), "not declared" ) ids = ", ".join(entry["advisories"][:6]) if len(entry["advisories"]) > 6: ids += f" and {len(entry['advisories']) - 6} more" out.append( f"| `{_safe_code_cell(entry['name'])}` ({_safe_text(entry['ecosystem'])}) | " f"{_safe_text(entry['version'])} | {reaches} | {_safe_text(ids)} |" ) out.append("") out += _unverifiable_lines(transitive) return out def _unverifiable_lines(transitive: dict) -> list[str]: """Name every lockfile entry that could not be verified against a registry. These are neither clean nor flagged: registry-keyed advisory data does not apply to a git checkout, a vendored directory, or a module the proxy cannot resolve, and an empty answer about them proves nothing. """ entries = transitive.get("unverifiable") or [] if not entries: return [] out = [ f"{plural(len(entries), 'entry', 'entries')} in the lockfile could not be " f"verified against a public registry and " f"{'was' if len(entries) == 1 else 'were'} not checked for advisories:", "", ] for entry in entries[:10]: out.append( f"- `{_safe_code(entry['name'])}` {_safe_text(entry['version'])} " f"({_safe_text(entry['ecosystem'])}) — {_safe_text(entry['reason'])}" ) if len(entries) > 10: out.append(f"- and {len(entries) - 10} more, listed in the artifact") out.append("") return out def production_section(artifact: dict) -> list[str]: """Every production dependency with a verdict, including the clean ones. A findings-only report cannot state this. Three of axios's four runtime dependencies appeared nowhere in an earlier draft because nothing was wrong with them — and for a reader whose question is "what reaches my users", a named package with a version and a clean advisory result is the most valuable line in the document. """ production = [d for d in artifact["dependencies"] if d.get("dev") is False] undeclared = [d for d in artifact["dependencies"] if d.get("dev") is None] out = ["## Production dependencies", ""] if not production: if undeclared: out += [ f"No manifest here declares which dependencies are development-only, so " f"none of the {len(undeclared)} can be identified as production or not. " f"Advisory status for all of them is in the findings and coverage below.", "", ] else: out += [ "Every direct dependency is declared as a development dependency, so none", "of them ships in the built artifact.", "", ] return out out += [ f"{plural(len(production), 'dependency', 'dependencies')} " f"{'is' if len(production) == 1 else 'are'} declared as runtime dependencies and " f"ship in the built artifact. Advisory status is given for every one, clean or not.", "", "| Dependency | Version | Advisories | Other findings |", "|---|---|---|---|", ] for dep in sorted(production, key=lambda d: d["name"]): advisories = _signal(dep, "advisories") verdict = { State.CLEAN.value: "none known", State.FLAGGED.value: f"**{advisories['detail']}**", }.get(advisories["state"], f"not established — {advisories['detail']}") others = [ LABELS.get(c, c) for c in dep["flagged"] if c != "advisories" and c not in UPSTREAM_ONLY ] out.append( f"| `{_safe_code_cell(dep['name'])}` | {_safe_text(_version_label(dep))} | {_safe_text(verdict)} " f"| {_safe_text(', '.join(others) or '—')} |" ) out.append("") return out def unassessable_section(artifact: dict) -> list[str]: """Every gap, grouped by criterion and reason, with the affected packages named.""" out = ["## Not assessable", ""] grouped: dict[str, dict[str, list[str]]] = {} for dep in artifact["dependencies"]: for criterion, signal in dep["signals"].items(): if signal["state"] != State.UNASSESSABLE.value: continue grouped.setdefault(criterion, {}).setdefault(signal["detail"], []).append(dep["name"]) if not grouped: out += ["Every criterion was assessable for every dependency.", ""] return out for criterion, reasons in sorted(grouped.items()): out += [f"**{LABELS.get(criterion, criterion)}**", ""] for reason, names in sorted(reasons.items(), key=lambda kv: -len(kv[1])): sample = ", ".join(f"`{_safe_code(n)}`" for n in sorted(names)[:6]) more = f" and {len(names) - 6} more" if len(names) > 6 else "" out.append( f"- {plural(len(names), 'dependency', 'dependencies')} — " f"{_safe_text(reason)}: {sample}{more}" ) out.append("") return out def scorecard_distribution(artifact: dict) -> list[str]: """Score spreads for the Scorecard checks that are measured but never flagged. Reported rather than dropped: a check demoted for poor precision still tells a reader something, and silently omitting it would leave a criterion counted in the coverage table and shown nowhere. """ out = [] for criterion in TIER_SCORECARD: if criterion not in SCORECARD_NEVER_FLAGS: # The flagging checks report through the findings tables and coverage; # describing them here as "not flagged" contradicted both. continue scores = [ s["value"] for dep in artifact["dependencies"] if isinstance((s := _signal(dep, criterion)).get("value"), int) ] if not scores: continue low = sum(1 for s in scores if s < 6) out.append( f"- **{LABELS.get(criterion, criterion)}**: median " f"{sorted(scores)[len(scores) // 2]}/10 across {len(scores)} scored " f"dependencies; {low} score below 6. Not flagged — " f"{SCORECARD_CAVEAT[criterion]}." ) return out def _download_line(artifact: dict) -> str: """Download volume is an integer, not a yes/no, so it gets its own summary line. Treating it as a boolean proportion printed "not determinable for any dependency" beside a coverage table that said 41 of 44 were assessed — the report contradicting itself about its own coverage. """ total = artifact["coverage"]["total_dependencies"] counted = sorted( (value, dep["name"]) for dep in artifact["dependencies"] if isinstance((value := _signal(dep, "downloads").get("value")), int) ) label = LABELS["downloads"] if not counted: return f"- **{label}**: not determinable for any dependency in this project." median = counted[len(counted) // 2][0] lowest = ", ".join(f"`{_safe_code(name)}` ({value:,}/wk)" for value, name in counted[:3]) return ( f"- **{label}**: established for {len(counted)} of {total}; median " f"{median:,}/week. Lowest: {lowest}." ) def informational_section(artifact: dict) -> list[str]: """Measured but never flagged, reported as proportions with the absentees named.""" out = ["## Informational", "", "Measured, not flagged.", ""] out += scorecard_distribution(artifact) for criterion in TIER_INFO: if criterion == "downloads": out.append(_download_line(artifact)) continue yes, no = [], [] for dep in artifact["dependencies"]: value = _signal(dep, criterion).get("value") if value is True: yes.append(dep["name"]) elif value is False: no.append(dep["name"]) determined = len(yes) + len(no) if not determined: out.append( f"- **{LABELS.get(criterion, criterion)}**: not determinable for any " f"dependency in this project." ) continue out.append( f"- **{LABELS.get(criterion, criterion)}**: {len(yes)} of {determined} " f"{INFO_PHRASING.get(criterion, 'satisfy this')}." ) if no: sample = ", ".join(f"`{_safe_code(n)}`" for n in sorted(no)[:10]) more = f" and {len(no) - 10} more" if len(no) > 10 else "" out.append(f" Without: {sample}{more}") out.append("") return out def check_no_forged_lines(lines: list[str]) -> None: """Refuse a document whose structure did not come from this renderer. Escaping is applied per interpolation site, so its coverage is only as good as the author's memory — this defect reached separate call sites across three successive commits, each fix correct and each leaving siblings unguarded. These two invariants hold regardless of which site leaks, including sites not yet written. A forged heading, bullet, or row needs a newline inside a line the renderer meant as one line, and every legitimate line is appended to `lines` as its own element, so an embedded newline is precisely the signature of interpolated hostile text. A pipe surviving into a table row is the same story for cells: GFM splits the row before it parses inline spans, so an unescaped pipe silently adds a column and shifts every later value right. Raises: ReconciliationError: If any assembled line carries a newline, or a table row's unescaped-pipe count differs from its header's. """ for line in lines: if "\n" in line: raise ReconciliationError( f"a line assembled by the renderer contains a newline, so third-party " f"text reached it unescaped: {line!r}" ) expected = None for line in lines: if not line.startswith("|"): expected = None continue # Only the pipes GFM splits on: an escaped `\|` is cell content. count = len(re.findall(r"(? None: """Every flag counted in coverage must appear in a findings table. This is the assertion that fails when it inspects zero items. A criterion added to the collector but not to this renderer would otherwise be counted in the coverage table and shown to no reader. """ counted = count_flags(artifact) coverage = artifact["coverage"]["criteria"] from_coverage = { name: counts[State.FLAGGED.value] for name, counts in coverage.items() if counts[State.FLAGGED.value] } if counted != from_coverage: raise ReconciliationError( f"flags in the dependency list {counted} disagree with the coverage table " f"{from_coverage}" ) renderable = {name for names, _ in FINDING_GROUPS for name in names} unrendered = sorted(set(from_coverage) - renderable) if unrendered: raise ReconciliationError( f"these criteria have flags that no findings table renders: {unrendered}" ) if from_coverage and "| Dependency |" not in rendered: raise ReconciliationError( f"coverage reports {sum(from_coverage.values())} flags but the report contains " f"no findings table" ) if artifact["transitive"].get("flagged") and "## Transitive advisories" not in rendered: raise ReconciliationError( f"{len(artifact['transitive']['flagged'])} transitive packages carry " f"advisories but no section renders them" ) def header(artifact: dict) -> list[str]: """Name the subject. Without it, a reader cannot tell whose dependency tree this is.""" scan = artifact.get("scan") or {} total = artifact["coverage"]["total_dependencies"] ecosystems = ", ".join(f"{eco} {n}" for eco, n in artifact["ecosystems"].items()) subject = scan.get("subject") or Path(artifact["target"]).name or "unnamed project" lines = [f"# Supply Chain Risk Report — `{_safe_code(subject)}`", ""] lines.append(f"**Scanned:** `{_safe_code(scan.get('path', artifact['target']))}` ") if scan.get("commit"): lines.append(f"**Commit:** `{_safe_code(scan['commit'])}` ") if scan.get("manifests"): joined = ", ".join(f"`{_safe_code(m)}`" for m in scan["manifests"]) lines.append(f"**Manifests read:** {joined} ") if scan.get("scanned_at"): lines.append(f"**Scanned at:** {scan['scanned_at']} ") lines += [f"**Direct dependencies:** {total} ({ecosystems})", ""] return lines def summary(artifact: dict) -> list[str]: """A factual headline plus the one coverage sentence a reader must not skip.""" deps = artifact["dependencies"] criteria = artifact["coverage"]["criteria"] total = artifact["coverage"]["total_dependencies"] adv = criteria["advisories"] flagged = [d for d in deps if d["flagged"]] prod_flagged = [d for d in flagged if d.get("dev") is False] weakest = min( ((assessed(criteria[name]), name) for name in TIER_A + TIER_B if name in criteria), default=(0, ""), ) lines = ["## Summary", ""] # "Checked at the versions this project resolves" is only true when every version # came from the project itself. Saying it about latest-release fallbacks turned a # missing lockfile into a stronger claim than a present one supports. # Whitelist the strong sources so any new version source defaults to the weaker # claim: go-mod-minimum escaped an earlier blacklist and a Go summary asserted # "the versions this project resolves" over floor versions. unresolved = sum(1 for d in deps if d.get("version_source") not in ("lockfile", "manifest-pin")) if adv[State.FLAGGED.value]: lines.append( f"- **{adv[State.FLAGGED.value]} of {total} dependencies have known " f"advisories.** See the findings below." ) elif assessed(adv) == total: if unresolved: lines.append( f"- **No known advisory affects any of the {total} direct dependencies** " f"— but {plural(unresolved, 'was', 'of them were')} not checked at a " f"project-resolved version (latest-release fallback or go.mod minimum; " f"see Method and caveats)." ) else: lines.append( f"- **No known advisory affects any of the {total} direct dependencies**, " f"checked at the versions this project resolves." ) else: lines.append( f"- Advisory status was established for {assessed(adv)} of {total} " f"dependencies; the rest could not be checked." ) lines.append(_transitive_summary(artifact["transitive"])) undeclared = [d for d in deps if d.get("dev") is None] if undeclared and not any(d.get("dev") is False for d in deps): lines.append( f"- {len(flagged)} of {total} dependencies carry at least one finding. These " f"manifests declare no runtime/development split, so which of them reach a " f"shipped artifact cannot be determined here." ) else: lines.append( f"- {len(flagged)} of {total} dependencies carry at least one finding, " f"{len(prod_flagged)} of which " f"{'reaches' if len(prod_flagged) == 1 else 'reach'} production." ) lines.append( f"- Weakest coverage: **{LABELS.get(weakest[1], weakest[1])}**, established for " f"{weakest[0]} of {total}; the Coverage section lists every criterion." ) lines.append("") return lines def render(artifact: dict) -> str: """Render the whole report, refusing an artifact that cannot support one. Order is deliberate. An earlier version opened with the full coverage table, and a reader with twenty minutes read two screens about the tool's limits before reaching anything they could act on. The one coverage sentence that matters is in the summary; the detail sits behind the findings. """ validate_artifact(artifact) lines = header(artifact) lines += summary(artifact) lines += production_section(artifact) for criteria, heading in FINDING_GROUPS: lines += findings_section(artifact, criteria, heading) lines += transitive_section(artifact) lines += informational_section(artifact) lines += ["## Coverage", ""] lines += [ "What was and was not measured, per criterion.", "", *coverage_table(artifact), "", ] lines += unassessable_section(artifact) lines += ["## Method and caveats", ""] lines += [f"- {_safe_text(note)}" for note in artifact["notes"]] lines.append("") check_no_forged_lines(lines) text = "\n".join(lines) check_flags_reconcile(artifact, text) return text def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("artifact", type=Path, help="collector JSON") parser.add_argument("--out", type=Path, help="write Markdown here (default: stdout)") args = parser.parse_args() try: text = render(json.loads(args.artifact.read_text(encoding="utf-8"), strict=False)) except ReconciliationError as exc: raise SystemExit(f"error: {args.artifact} cannot be rendered: {exc}") from exc if args.out: args.out.write_text(text, encoding="utf-8") print(f"wrote {args.out}") else: print(text) return 0 if __name__ == "__main__": sys.exit(main()) ``` ## scripts/sources.py ```python """Data-source adapters. Each returns measurements; none decides risk. Every function either returns a datum or raises `Unavailable`. Turning an `Unavailable` into a verdict belongs to `collect.py`, which only ever turns it into `Signal.unassessable`. Notes earned by getting them wrong first: - npm's full packument for popular packages is **invalid JSON** — lodash's 247KB response carries unescaped control characters and `jq` rejects it. Parse with `json.loads(..., strict=False)`. - The packument-level `.maintainers` is the *current* publish ACL. The version-level `.maintainers` is a publish-time snapshot and disagrees: lodash reads 1 current against 3 at v4.18.1. - OSV advisories must be keyed by ecosystem+package, never by repository. Repo-keyed gives 3 for lodash where ecosystem-keyed gives 5. - deps.dev keys Go versions *with* a `v` prefix; OSV wants it stripped. - deps.dev's `relatedProjects` is unordered and typed. Across 104 cached responses the first entry was `ISSUE_TRACKER` 51 times and `SOURCE_REPO` 53 times, so taking the first entry was right only because the two ids happened to coincide. - GitHub reports primary rate-limit exhaustion as **403** with `x-ratelimit-remaining: 0`, not 429, so matching only 429 misses the throttle that actually happens. """ from __future__ import annotations import hashlib import json import os import re import subprocess import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from email.message import Message from pathlib import Path from shutil import which from model import SCORECARD_CHECKS UA = {"User-Agent": "trailofbits-supply-chain-risk-auditor/0.1"} TIMEOUT = 45 OSV_BATCH_MAX = 500 # Entries older than this are refetched when online. Without a bound, a cached # `pushed_at` ages alongside the repository and manufactures staleness flags: a repo last # pushed 300 days before its document was cached, read from a 100-day-old entry, reports # "no push in 400 days" about a project that may have shipped yesterday. CACHE_MAX_AGE_SECONDS = 6 * 60 * 60 class Unavailable(Exception): """A source could not supply this datum. Never a risk verdict.""" class NotFound(Unavailable): """The resource provably does not exist (HTTP 404). Distinct from its parent because absence is evidence and every other failure is not: a 404 on SECURITY.md means there is no security policy, while a rate limit on the same request means nothing at all. """ @dataclass class CacheEntry: body: dict age_seconds: float @dataclass class _Response: """The slice of an HTTP response the client needs, however urllib delivered it.""" status: int text: str headers: Message = field(default_factory=Message) class Http: """Caching HTTP client with a freshness bound and atomic writes. Built on urllib so the plugin carries no third-party dependency. The default opener follows redirects, and that is load-bearing: `jrburke/r.js` 301s to `requirejs/r.js`, and a client that stopped at the 301 would call the repository missing. """ def __init__(self, cache_dir: Path, offline: bool = False, auth_marker: str = "anon"): self.cache_dir = cache_dir self.offline = offline # Cache keys include an auth marker: a private or SSO-gated repository 404s # anonymously, and without this the negative result would be served to every # later authenticated run — the run the user fixed their credentials for. self.auth_marker = auth_marker self.cache_dir.mkdir(parents=True, exist_ok=True, mode=0o700) # None once ownership is confirmed; a caveat string when the platform cannot be # asked. Callers surface it — see collect.collect, which appends it to the report's # notes so an unenforced control appears in "Method and caveats" rather than # nowhere. self.cache_owner_caveat = self._check_cache_owner() self.opener = urllib.request.build_opener() self.stats = {"hits": 0, "fetched": 0, "offline_misses": 0, "errors": 0, "stale": 0} self.oldest_hit_seconds = 0.0 def _check_cache_owner(self) -> str | None: """Refuse a cache owned by another user, or report why that was not checked. The cache holds registry and advisory responses, and the collector trusts them. A cache another user can write turns a compromised package into a clean "no advisories" verdict, so on POSIX a foreign owner aborts the run. `os.getuid` is POSIX-only and `st_uid` is always 0 on Windows, so the check cannot be evaluated there. Rather than drop it silently, this returns a caveat for the caller to surface: an unenforced control the report admits to is a smaller problem than one nobody mentions. Returns: None when ownership was verified, or the caveat text when the platform offers no way to verify it. Raises: SystemExit: The cache directory belongs to another user. """ if not hasattr(os, "getuid"): # Windows return ( f"cache directory ownership was not verified: os.getuid is POSIX-only, so " f"a cache at {self.cache_dir} owned by another user would not be detected " f"on this platform. Keep the cache under your own user profile, or pass " f"--cache with a private path." ) if self.cache_dir.stat().st_uid != os.getuid(): raise SystemExit( f"error: cache directory {self.cache_dir} is owned by another user; " f"refusing to trust or write it — pass --cache with a private path" ) return None # ------------------------------------------------------------------ cache def _path(self, method: str, url: str, body: str = "") -> Path: digest = hashlib.sha256(f"{method} {url} {body} {self.auth_marker}".encode()).hexdigest()[ :32 ] return self.cache_dir / f"{digest}.json" def _read_cache(self, path: Path) -> CacheEntry | None: if not path.exists(): return None try: stored = json.loads(path.read_text(encoding="utf-8"), strict=False) except (json.JSONDecodeError, UnicodeDecodeError, OSError): # A run interrupted mid-write leaves a truncated file that would otherwise # crash every later run. Drop it and refetch. self.stats["errors"] += 1 path.unlink(missing_ok=True) return None meta = stored.get("__meta") if not isinstance(meta, dict): path.unlink(missing_ok=True) return None age = time.time() - float(meta.get("fetched_at", 0)) if age < 0: # A timestamp from the future (clock skew, or a planted entry) would read # as age zero forever; treat it as infinitely stale instead. age = float("inf") return CacheEntry(body=stored, age_seconds=age) def _write_cache(self, path: Path, body: dict, status: int) -> None: payload = {"__meta": {"fetched_at": time.time(), "status": status}, "body": body} # Atomic: a partial file must never become a cache hit. tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(payload), encoding="utf-8") os.replace(tmp, path) def _serve(self, entry: CacheEntry, url: str) -> dict: self.stats["hits"] += 1 self.oldest_hit_seconds = max(self.oldest_hit_seconds, entry.age_seconds) if entry.body["__meta"].get("status") == 404: raise NotFound(f"404 from {url}") return entry.body["body"] def _usable(self, entry: CacheEntry | None) -> bool: if entry is None: return False return self.offline or entry.age_seconds <= CACHE_MAX_AGE_SECONDS # ------------------------------------------------------------------ fetch def _rate_limited(self, resp: _Response) -> bool: if resp.status == 429: return True # GitHub's primary rate limit is a 403 with the remaining count at zero. return resp.status == 403 and resp.headers.get("x-ratelimit-remaining") == "0" def _retry_delay(self, resp: _Response) -> float: after = resp.headers.get("Retry-After") if after and after.isdigit(): return min(float(after), 30.0) return 3.0 def _open(self, method: str, url: str, headers: dict, payload: dict | None) -> _Response: """One HTTP exchange. Error statuses are data here; only transport failures raise.""" data = json.dumps(payload).encode() if payload is not None else None request = urllib.request.Request(url, data=data, headers={**UA, **headers}, method=method) if payload is not None: request.add_header("Content-Type", "application/json") try: with self.opener.open(request, timeout=TIMEOUT) as resp: return _Response(resp.status, resp.read().decode("utf-8", "replace"), resp.headers) except urllib.error.HTTPError as exc: with exc: return _Response(exc.code, exc.read().decode("utf-8", "replace"), exc.headers) except (urllib.error.URLError, TimeoutError, OSError) as exc: self.stats["errors"] += 1 raise Unavailable(f"request failed: {exc}") from exc def _send(self, method: str, url: str, headers: dict, payload: dict | None) -> dict: resp = self._open(method, url, headers, payload) if self._rate_limited(resp): time.sleep(self._retry_delay(resp)) resp = self._open(method, url, headers, payload) if self._rate_limited(resp): self.stats["errors"] += 1 raise Unavailable(f"rate limited by {url}") return self._decode(resp, url) def _decode(self, resp: _Response, url: str) -> dict: # 410 Gone is proxy.golang.org's "module does not exist"; treat both as provable # absence rather than as an outage. if resp.status in (404, 410): raise NotFound(f"{resp.status} from {url}") if resp.status >= 400: self.stats["errors"] += 1 raise Unavailable(f"HTTP {resp.status} from {url}") self.stats["fetched"] += 1 return json.loads(resp.text, strict=False) def _request(self, method: str, url: str, headers: dict, payload: dict | None) -> dict: body = json.dumps(payload, sort_keys=True) if payload else "" path = self._path(method, url, body) entry = self._read_cache(path) if self._usable(entry): return self._serve(entry, url) if entry is not None: self.stats["stale"] += 1 if self.offline: self.stats["offline_misses"] += 1 raise Unavailable(f"not in the local cache and this run is offline: {url}") try: data = self._request_and_cache(method, url, headers, payload, path) except NotFound: self._write_cache(path, {}, 404) raise return data def _request_and_cache( self, method: str, url: str, headers: dict, payload: dict | None, path: Path ) -> dict: data = self._send(method, url, headers, payload) self._write_cache(path, data, 200) return data def get_json(self, url: str, headers: dict | None = None) -> dict: return self._request("GET", url, headers or {}, None) def post_json(self, url: str, payload: dict) -> dict: return self._request("POST", url, {}, payload) # --------------------------------------------------------------------------- OSV def osv_advisories(http: Http, queries: list[tuple[str, str, str | None]]) -> list[list[str]]: """Batch-query OSV. One HTTP call per 500 packages. Args: http: Caching client. queries: (ecosystem, name, version) triples. A None version asks for every advisory recorded against the package, a historical claim rather than a statement about what the project installs. Returns: Advisory-ID lists, positional with `queries`. Positional rather than keyed by package, because a transitive tree routinely holds the same package at two versions and a name-keyed mapping would silently keep only one. Raises: Unavailable: If OSV returns a different number of results than queries sent. Results are positional, so a short response would shift every advisory onto the wrong package — a flagged verdict whose evidence belongs to something else. The mismatch is detectable, so it must not be tolerated. """ out: list[list[str]] = [] for start in range(0, len(queries), OSV_BATCH_MAX): chunk = queries[start : start + OSV_BATCH_MAX] payload: dict = {"queries": []} for eco, name, version in chunk: query: dict = {"package": {"name": name, "ecosystem": eco}} if version: query["version"] = version payload["queries"].append(query) data = http.post_json("https://api.osv.dev/v1/querybatch", payload) results = data.get("results", []) if len(results) != len(chunk): raise Unavailable( f"OSV returned {len(results)} results for {len(chunk)} queries; positional " f"pairing would attribute advisories to the wrong packages" ) out.extend([v["id"] for v in (result or {}).get("vulns", [])] for result in results) return out # --------------------------------------------------------------------------- npm # Substring hints. Conservative on purpose: a false negative overstates the human # publisher count, which is the safer direction, since a count of 1 is what creates a # flag. `dependabot` and `renovate` are named explicitly because neither contains a # hyphen next to "bot". _AUTOMATION_HINTS = ("-bot", "bot-", "-ci", "npm-cli", "github-actions", "dependabot", "renovate") def looks_automated(account: str) -> bool: """Heuristic: does this registry account look like a bot rather than a person?""" lowered = account.lower() return lowered in {"bot", "ci"} or any(hint in lowered for hint in _AUTOMATION_HINTS) def _maintainer_names(raw: object) -> list[str]: """Registry maintainer lists hold dicts now and bare strings in older packuments.""" names = [] for entry in raw or []: if isinstance(entry, dict): name = entry.get("name") else: name = str(entry) if name: names.append(name) return names def npm_metadata(http: Http, name: str) -> dict: """Current publish ACL, provenance, install-script flag, deprecation, repository. Raises: NotFound: If the package is not published. Unavailable: If `dist-tags.latest` names a version the packument does not carry, in which case install-script and provenance would otherwise read as determined negatives derived from an absent document. """ quoted = urllib.parse.quote(name, safe="@").replace("/", "%2F") doc = http.get_json(f"https://registry.npmjs.org/{quoted}") latest = (doc.get("dist-tags") or {}).get("latest") versions = doc.get("versions") or {} if not latest or latest not in versions: raise Unavailable("npm's latest tag points at a version absent from the packument") version_doc = versions[latest] maintainers = _maintainer_names(doc.get("maintainers")) repository = doc.get("repository") repo_url = repository.get("url") if isinstance(repository, dict) else repository return { "latest": latest, "maintainers": maintainers, "human_maintainers": [m for m in maintainers if not looks_automated(m)], "automated_maintainers": [m for m in maintainers if looks_automated(m)], "has_install_script": bool(version_doc.get("hasInstallScript")), "provenance": bool((version_doc.get("dist") or {}).get("attestations")), "deprecated": version_doc.get("deprecated") or doc.get("deprecated"), "repository": normalize_repo(repo_url) if isinstance(repo_url, str) else None, } def npm_downloads(http: Http, name: str) -> int: quoted = urllib.parse.quote(name, safe="@").replace("/", "%2F") data = http.get_json(f"https://api.npmjs.org/downloads/point/last-week/{quoted}") count = data.get("downloads") if count is None: raise Unavailable("npm returned no download count") return int(count) # ------------------------------------------------------------------------- PyPI def pypi_metadata(http: Http, name: str, version: str | None = None) -> dict: """PyPI exposes no upload ACL, and its download fields return -1 by design. Args: http: Caching client. name: Distribution name, PEP 503 normalised. version: The version to report yank status for. Yank is per-release, so checking the latest release's status and displaying it against a pinned version states something false about that version. """ doc = http.get_json(f"https://pypi.org/pypi/{name}/json") info = doc.get("info") or {} releases = doc.get("releases") or {} latest = info.get("version") target = version if version in releases else None files = releases.get(target) if target else None urls = info.get("project_urls") or {} repo_url = None for label in ("Source", "Source Code", "Repository", "Code", "GitHub", "Homepage"): candidate = urls.get(label) if isinstance(candidate, str) and ("github.com" in candidate or "gitlab" in candidate): repo_url = candidate break return { "latest": latest, "yank_version": target, "yanked": any(f.get("yanked") for f in files) if files else None, "yanked_reason": ( next((f.get("yanked_reason") for f in files if f.get("yanked")), None) if files else None ), "repository": normalize_repo(repo_url) if repo_url else None, } # -------------------------------------------------------------------- Go module proxy def go_module_latest(http: Http, module: str) -> dict: """Resolve a module against proxy.golang.org, which is the Go existence check. Go has no registry, so a module path that OSV has no advisories for is indistinguishable from a typo unless the proxy actually resolves it — and an early version of this collector asserted the proxy check in a comment while never making the request, which let nonexistent modules read as assessed-clean. Raises: NotFound: If the proxy has no such module (404, or its 410 Gone). Unavailable: If the proxy could not be reached. """ # The proxy escapes uppercase letters as "!" (github.com/Azure -> # github.com/!azure). escaped = re.sub(r"[A-Z]", lambda m: "!" + m.group(0).lower(), module) return http.get_json(f"https://proxy.golang.org/{escaped}/@latest") # ---------------------------------------------------------------------- deps.dev _DEPSDEV_SYSTEM = {"npm": "npm", "PyPI": "pypi", "Go": "go", "crates.io": "cargo"} def depsdev_repo(http: Http, ecosystem: str, name: str, version: str | None) -> str: """Resolve a package to its source repository. Prefers the `SOURCE_REPO` relation. `relatedProjects` is unordered and also carries `ISSUE_TRACKER`, which can point at a different repository entirely. """ system = _DEPSDEV_SYSTEM.get(ecosystem) if not system or not version: raise Unavailable(f"no deps.dev lookup for {ecosystem} without a resolved version") if system == "go" and not version.startswith("v"): version = f"v{version}" quoted = name.replace("/", "%2F") doc = http.get_json( f"https://api.deps.dev/v3alpha/systems/{system}/packages/{quoted}/versions/{version}" ) for project in doc.get("relatedProjects") or []: if project.get("relationType") != "SOURCE_REPO": continue ident = (project.get("projectKey") or {}).get("id") if ident: return normalize_repo(ident) for link in doc.get("links") or []: if link.get("label") == "SOURCE_REPO" and link.get("url"): return normalize_repo(link["url"]) raise Unavailable("deps.dev lists no source repository") # ------------------------------------------------------------- repo identifiers _HOST_SHORTHAND = {"github": "github.com", "gitlab": "gitlab.com", "bitbucket": "bitbucket.org"} _KNOWN_FORGES = ("github.com", "gitlab.com", "bitbucket.org") def normalize_repo(raw: str) -> str: """Canonicalise a repository identifier to `//`. deps.dev and the registries return whatever a manifest declared: npm's bare `owner/repo` shorthand (which means GitHub), `git+https://….git`, scp-style `git@host:owner/repo`, mixed-case hosts, or a deep link into a file on a branch. Getting any of these wrong produces a false "hosted outside GitHub" claim, which is this report's worst failure mode — a confident statement about a third party derived from a string-formatting detail. """ text = (raw or "").strip().removeprefix("git+") for prefix, host in _HOST_SHORTHAND.items(): if text.startswith(f"{prefix}:") and not text.startswith(f"{prefix}://"): text = f"{host}/{text[len(prefix) + 1 :]}" break for scheme in ("https://", "http://", "ssh://", "git://"): text = text.removeprefix(scheme) if "@" in text.split("/", 1)[0]: text = text.split("@", 1)[1].replace(":", "/", 1) text = text.removesuffix(".git").strip("/").removeprefix("www.") parts = text.split("/") if parts and "." not in parts[0] and len(parts) == 2: # npm's bare `owner/repo` shorthand implies GitHub. parts = ["github.com", *parts] if parts: parts[0] = parts[0].lower() # Truncate deep links: a Homepage of github.com/psf/requests/blob/main/README.md is # a repository identifier with noise on the end. if parts and parts[0] in _KNOWN_FORGES and len(parts) > 3: parts = parts[:3] return "/".join(parts) # ------------------------------------------------------------------------ GitHub def gh_token() -> str | None: """Read the gh CLI's token. Availability is not authentication. Unauthenticated GitHub allows 60 requests/hour against 5000 authenticated, and this collector makes several per dependency. """ try: out = subprocess.run( ["gh", "auth", "token"], capture_output=True, text=True, encoding="utf-8", timeout=15, check=False, ) except (OSError, subprocess.SubprocessError): return None return out.stdout.strip() or None class RepoIdentifierError(Unavailable): """The identifier could not be parsed or is not on a host we can query.""" def github_repo(http: Http, repo_id: str, token: str | None) -> dict: """Staleness, archived state, and security-policy presence. Raises: RepoIdentifierError: If the identifier is unparseable or not on GitHub. Distinct from a request failure so the report does not blame the GitHub API for a request it never made. Unavailable: If GitHub could not be reached. """ slug = normalize_repo(repo_id) if not slug.startswith("github.com/"): raise RepoIdentifierError(f"not a GitHub repository: {slug}") owner_repo = slug.removeprefix("github.com/") if owner_repo.count("/") != 1: raise RepoIdentifierError(f"cannot parse owner/repo from {repo_id}") headers = {"Accept": "application/vnd.github+json"} if token: headers["Authorization"] = f"Bearer {token}" doc = http.get_json(f"https://api.github.com/repos/{owner_repo}", headers=headers) return { "owner_repo": owner_repo, "pushed_at": doc.get("pushed_at"), "archived": bool(doc.get("archived")), "security_policy": _security_policy(http, owner_repo, headers), } def _security_policy(http: Http, owner_repo: str, headers: dict) -> bool | None: """Is a security policy published? Returns True, False, or None when the question could not be answered. Only a 404 proves absence — a rate limit or a network error proves nothing, and collapsing the two reported "no security policy" as a measured fact about projects that have one. GitHub also serves the owner's `.github` repository as a default, so an in-repo-only check is wrong: expressjs/express 404s in-repo while expressjs/.github/SECURITY.md is a 200. """ owner = owner_repo.split("/")[0] candidates = ( f"{owner_repo}/contents/SECURITY.md", f"{owner_repo}/contents/.github/SECURITY.md", f"{owner_repo}/contents/docs/SECURITY.md", f"{owner}/.github/contents/SECURITY.md", f"{owner}/.github/contents/.github/SECURITY.md", ) inconclusive = False for candidate in candidates: try: http.get_json(f"https://api.github.com/repos/{candidate}", headers=headers) return True except NotFound: continue except Unavailable: inconclusive = True return None if inconclusive else False # -------------------------------------------------------------------- Scorecard # Check names and flag thresholds live in model.SCORECARD_CHECKS, the single source of # truth shared with the renderer. A Scorecard score of -1 means the check could not run # — commonly for want of repository-admin access — and is a gap, not a finding. def scorecard_checks(http: Http, repo_id: str) -> dict[str, int]: """Individual OpenSSF Scorecard check scores, keyed by Scorecard's own check names. Raises: NotFound: If Scorecard has no report for this repository. Coverage is incomplete — paulmillr/noble-curves, audited cryptography, is a 404. """ slug = normalize_repo(repo_id) doc = http.get_json(f"https://api.scorecard.dev/projects/{slug}") scores = {c["name"]: c.get("score", -1) for c in doc.get("checks") or []} return {name: scores[name] for name in SCORECARD_CHECKS if name in scores} # ------------------------------------------------------- optional local tooling OPTIONAL_TOOLS = ("pip-audit", "osv-scanner", "npm", "cargo-audit", "bundler-audit") def detect_tools() -> dict[str, bool]: """Which optional ecosystem tools are on PATH. Never required, only used if present.""" return {tool: which(tool) is not None for tool in OPTIONAL_TOOLS} def pip_audit_vulnerable(requirements: Path) -> set[str]: """Cross-check PyPI advisories with pip-audit, when it is installed. Returns the set of PEP 503 normalised names pip-audit reports as vulnerable. A disagreement with OSV is worth reporting in its own right — two databases differing tells a reader more than either alone. Raises: Unavailable: If pip-audit is missing, times out, or returns unparseable output. """ try: out = subprocess.run( [ "pip-audit", "--requirement", str(requirements), # Both flags are required. Without them pip-audit resolves the full # tree through pip, downloading distributions and running setup.py for # anything without a wheel — executing the untrusted code this tool # promises never to run. Measured: `--no-deps` alone still audited the # resolved transitive set, so pip was still involved; `--disable-pip` # (valid only with `--no-deps`) removes pip entirely and audits exactly # the listed pins from registry metadata. "--no-deps", "--disable-pip", "--format", "json", "--progress-spinner", "off", ], capture_output=True, text=True, encoding="utf-8", timeout=300, check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise Unavailable(f"pip-audit could not run: {exc}") from exc if not out.stdout.strip(): raise Unavailable(f"pip-audit produced no output (exit {out.returncode})") try: data = json.loads(out.stdout) except json.JSONDecodeError as exc: raise Unavailable(f"pip-audit output was not JSON: {exc}") from exc entries = data.get("dependencies") if isinstance(data, dict) else data vulnerable = set() for entry in entries or []: if entry.get("vulns"): vulnerable.add(str(entry.get("name", "")).lower()) return vulnerable def polite_pause(seconds: float = 0.05) -> None: time.sleep(seconds) ``` ## scripts/test_collect.py ```python """Collector tests: manifest parsing, signal judgement, and the offline pipeline. The end-to-end tests seed the HTTP cache through `Http`'s own writer and run with `offline=True`, so they exercise the real cache and code paths with no network. The negatives are the point: a package that 404s everywhere must produce no flag and no clean verdict, and a run in which every source is down must refuse to report. """ from __future__ import annotations import ast import json import os import subprocess import sys import tempfile import textwrap from datetime import UTC, datetime, timedelta from pathlib import Path import pytest import sources from collect import ( STALE_DAYS, _read_json, _advisory_signal, _concentration_signal, _exact_npm_pin, _go_indirect, _npm_all_locked, _npm_spec_kind, _pypi_version, _staleness_signal, _uv_all_locked, collect, discover, parse_go, parse_npm, parse_pypi, sweep_transitive, ) from model import Dependency, ReconciliationError, State, _validate_transitive # ---------------------------------------------------------------------- npm parsing def test_ranges_are_not_pins(): assert _exact_npm_pin("1.2.3") == "1.2.3" for spec in ("^1.2.3", "~1.2", "1.x", "1.2.x", "2", "*", ">=1.0.0"): assert _exact_npm_pin(spec) is None, spec def test_non_registry_specs_classified(): for spec in ("file:../local", "workspace:*", "git+https://x/y.git", "owner/repo"): assert _npm_spec_kind(spec) == "non-registry", spec assert _npm_spec_kind("npm:real@^2") == "alias" assert _npm_spec_kind("^1.0.0") == "registry" def test_npm_alias_audits_the_target(tmp_path: Path): (tmp_path / "package.json").write_text( json.dumps({"dependencies": {"my-alias": "npm:real-pkg@^2.0.0"}}) ) deps, notes = parse_npm(tmp_path) assert [d.name for d in deps] == ["real-pkg"] assert any("alias" in n for n in notes) def test_peer_dependencies_noted_not_audited(tmp_path: Path): (tmp_path / "package.json").write_text( json.dumps({"dependencies": {"a": "^1"}, "peerDependencies": {"react": "^18"}}) ) deps, notes = parse_npm(tmp_path) assert [d.name for d in deps] == ["a"] assert any("peerDependenc" in n for n in notes) def test_duplicate_declaration_keeps_the_runtime_one(tmp_path: Path): (tmp_path / "package.json").write_text( json.dumps({"dependencies": {"a": "^1"}, "devDependencies": {"a": "^1"}}) ) deps, _ = discover(tmp_path) assert len(deps) == 1 and deps[0].dev is False REG = "https://registry.npmjs.org" def registry_entry(name: str, version: str, **extra: object) -> dict: return { "version": version, "resolved": f"{REG}/{name}/-/{name}-{version}.tgz", "integrity": "sha512-fake", **extra, } def test_npm_lockfile_closure_reads_nested_paths_and_dev(tmp_path: Path): (tmp_path / "package-lock.json").write_text( json.dumps( { "packages": { "": {"name": "demo"}, "node_modules/direct": registry_entry("direct", "1.0.0"), "node_modules/leftover": registry_entry("leftover", "2.0.0", dev=True), "node_modules/leftover/node_modules/inner": registry_entry( "inner", "3.0.0", dev=True ), "node_modules/linked": {"version": "9.9.9", "link": True}, } } ) ) packages, unverifiable, source, note = _npm_all_locked(tmp_path) assert source == "package-lock.json" and note is None assert ("npm", "inner", "3.0.0", True) in packages assert all(name != "linked" for _, name, _, _ in packages) assert unverifiable == [] def test_npm_lockfile_git_and_hashless_entries_are_unverifiable(tmp_path: Path): (tmp_path / "package-lock.json").write_text( json.dumps( { "packages": { "node_modules/good": registry_entry("good", "1.0.0"), "node_modules/evil-git-dep": { "version": "2.0.0", "resolved": "git+ssh://git@github.com/acme/evil-git-dep.git#abc", }, "node_modules/@acme/private": { "version": "3.0.0", "resolved": "https://npm.acme.internal/@acme/private/-/private-3.0.0.tgz", "integrity": "sha512-x", }, } } ) ) packages, unverifiable, _, _ = _npm_all_locked(tmp_path) assert [n for _, n, _, _ in packages] == ["good"] names = {e["name"] for e in unverifiable} assert names == {"evil-git-dep", "@acme/private"} assert all("not the npm registry" in e["reason"] for e in unverifiable) def test_v1_lockfile_yields_a_note_not_a_silent_zero(tmp_path: Path): (tmp_path / "package-lock.json").write_text( json.dumps({"dependencies": {"a": {"version": "1.0.0"}}}) ) packages, unverifiable, source, note = _npm_all_locked(tmp_path) assert packages == [] and unverifiable == [] and source is None assert note and "transitive tree was not read" in note # --------------------------------------------------------------------- PyPI parsing def test_pep508_marker_does_not_fabricate_a_version(): version, source = _pypi_version("psutil; sys_platform == 'win32'", {}, "psutil") assert version is None and source == "unresolved" def test_pinned_version_extraction(): assert _pypi_version("requests==2.19.0", {}, "requests") == ("2.19.0", "manifest-pin") assert _pypi_version("foo==1.0.*", {}, "foo") == (None, "unresolved") # pip-compile continuation and inline options end the version without corrupting it assert _pypi_version("requests==2.19.0 \\", {}, "requests") == ("2.19.0", "manifest-pin") assert _pypi_version("requests==2.19.0 --hash=sha256:abc", {}, "requests") == ( "2.19.0", "manifest-pin", ) # PEP 440 permits a `v` prefix and pip accepts it; requiring a leading digit made # this legal pin unresolved, so advisories matched the latest release and 62 real # ones for django v3.2.0 read as clean assert _pypi_version("django==v3.2.0", {}, "django") == ("3.2.0", "manifest-pin") # an epoch must survive: splitting on `!` (there for `!=`) truncated 1!2.0 to 1 assert _pypi_version("x==1!2.0+local", {}, "x") == ("1!2.0+local", "manifest-pin") # anything the gate cannot recognise stays unresolved rather than reaching OSV assert _pypi_version("x==1.2.3[extra]", {}, "x") == (None, "unresolved") def test_requirements_filename_hints_dev(tmp_path: Path): (tmp_path / "requirements.txt").write_text("flask==2.0.0\n") # sorts before requirements.txt; a package in both must stay runtime at the # runtime pin, not collapse to the first-seen dev entry (tmp_path / "requirements-dev.txt").write_text("pytest==8.0.0\nflask==1.0.2\n") # PEP 508 direct reference: the fork URL is the identity, not the public name (tmp_path / "requirements.txt").write_text( "flask==2.0.0\ninternal-lib @ git+ssh://git@github.com/acme/internal-lib.git\n" ) deps, _ = parse_pypi(tmp_path) by_name = {d.name: d for d in deps} assert by_name["flask"].dev is False and by_name["flask"].version == "2.0.0" assert by_name["pytest"].dev is True assert by_name["internal-lib"].non_registry_reason assert "git+ssh://" in by_name["internal-lib"].non_registry_reason def test_uv_lock_closure_skips_the_project_itself(tmp_path: Path): (tmp_path / "uv.lock").write_text( "\n".join( [ "[[package]]", 'name = "demo"', 'version = "0.1.0"', 'source = { editable = "." }', "", "[[package]]", 'name = "Left_Pad"', 'version = "1.0.0"', 'source = { registry = "https://pypi.org/simple" }', ] ) ) packages, unverifiable, source, note = _uv_all_locked(tmp_path) assert source == "uv.lock" and note is None assert packages == [("PyPI", "left-pad", "1.0.0", None)] assert unverifiable == [] def test_uv_lock_non_registry_sources_are_unverifiable(tmp_path: Path): (tmp_path / "uv.lock").write_text( "\n".join( [ "[[package]]", 'name = "internal-lib"', 'version = "3.0.0"', 'source = { git = "ssh://git@github.com/acme/internal-lib" }', "", "[[package]]", 'name = "vendored-flask"', 'version = "1.0.2"', 'source = { directory = "../vendor/flask" }', "", "[[package]]", 'name = "real"', 'version = "2.0.0"', 'source = { registry = "https://pypi.org/simple" }', ] ) ) packages, unverifiable, _, _ = _uv_all_locked(tmp_path) assert [n for _, n, _, _ in packages] == ["real"] reasons = {e["name"]: e["reason"] for e in unverifiable} assert "git source" in reasons["internal-lib"] assert "directory source" in reasons["vendored-flask"] # a DIRECT dependency with a non-registry lock source carries the marker, so the # pipeline never looks it up on PyPI by name (tmp_path / "pyproject.toml").write_text( '[project]\nname = "demo"\nversion = "0"\ndependencies = ["internal-lib", "real"]\n' ) deps, _ = parse_pypi(tmp_path) by_name = {d.name: d for d in deps} assert by_name["internal-lib"].non_registry_reason assert "git source" in by_name["internal-lib"].non_registry_reason assert by_name["real"].non_registry_reason is None # ----------------------------------------------------------------------- Go parsing GOMOD = """\ module example.com/demo go 1.21 require ( \tgithub.com/direct/one v1.2.3 \tgolang.org/x/text v0.9.0 // indirect ) replace ( \tgithub.com/direct/one => github.com/fork/one v9.9.9 ) exclude github.com/bad/mod v0.0.1 """ def test_gomod_blocks_and_dev_tristate(tmp_path: Path): (tmp_path / "go.mod").write_text(GOMOD) deps, notes = parse_go(tmp_path) assert [(d.name, d.version) for d in deps] == [("github.com/direct/one", "1.2.3")] # go.mod declares no runtime/dev split; asserting one would invent a distinction. assert deps[0].dev is None assert any("replace/exclude" in n for n in notes) def test_go_indirect_modules_are_the_transitive_set(tmp_path: Path): (tmp_path / "go.mod").write_text(GOMOD) packages, unverifiable, source, note = _go_indirect(tmp_path) assert source == "go.mod" and note is None assert packages == [("Go", "golang.org/x/text", "0.9.0", None)] assert unverifiable == [] def test_pre_117_gomod_is_not_a_closure(tmp_path: Path): (tmp_path / "go.mod").write_text(GOMOD.replace("go 1.21", "go 1.16")) packages, unverifiable, source, note = _go_indirect(tmp_path) assert packages == [] and unverifiable == [] and source is None assert note and "1.17" in note # ------------------------------------------------------------------ signal judgement def test_empty_advisory_answer_needs_proof_of_existence(): dep = Dependency(ecosystem="npm", name="ghost", version="1.0.0", exists=None) signal = _advisory_signal(dep, {dep.key: []}) assert signal.state is State.UNASSESSABLE dep.exists = True assert _advisory_signal(dep, {dep.key: []}).state is State.CLEAN def test_advisory_wording_matches_version_source(): dep = Dependency( ecosystem="Go", name="example.com/m", version="1.0.0", version_source="go-mod-minimum", exists=True, ) signal = _advisory_signal(dep, {dep.key: ["GO-2026-1"]}) assert signal.state is State.FLAGGED and "minimum" in signal.detail def test_staleness_threshold(): old = (datetime.now(UTC) - timedelta(days=STALE_DAYS + 30)).isoformat() fresh = (datetime.now(UTC) - timedelta(days=400)).isoformat() assert _staleness_signal(old).state is State.FLAGGED # 400 days is a finished library, not an abandoned one; flagging it buried the # decade-abandoned packages this criterion exists for. assert _staleness_signal(fresh).state is State.CLEAN assert _staleness_signal(None).state is State.UNASSESSABLE assert _staleness_signal("2020-01-01T00:00:00").state is State.UNASSESSABLE def test_concentration_judgement(): base = { "provenance": False, "maintainers": [], "human_maintainers": [], "automated_maintainers": [], } assert _concentration_signal({**base, "provenance": True}).state is State.UNASSESSABLE assert _concentration_signal(base).state is State.UNASSESSABLE lone = { **base, "maintainers": ["alice", "release-bot"], "human_maintainers": ["alice"], "automated_maintainers": ["release-bot"], } signal = _concentration_signal(lone) assert signal.state is State.FLAGGED and "release-bot" in signal.detail two = {**base, "maintainers": ["alice", "bob"], "human_maintainers": ["alice", "bob"]} assert _concentration_signal(two).state is State.CLEAN # ------------------------------------------------------------------- offline seeding def seeded_http(cache_dir: Path, responses: dict[tuple[str, str, str], object]) -> sources.Http: """Write canned responses through Http's own cache writer, then serve offline.""" http = sources.Http(cache_dir, offline=True) for (method, url, body), payload in responses.items(): path = http._path(method, url, body) if payload is None: http._write_cache(path, {}, 404) else: http._write_cache(path, payload, 200) return http def osv_body(triples: list[tuple[str, str, str | None]]) -> str: queries = [] for eco, name, version in triples: query: dict = {"package": {"name": name, "ecosystem": eco}} if version: query["version"] = version queries.append(query) return json.dumps({"queries": queries}, sort_keys=True) OSV = "https://api.osv.dev/v1/querybatch" # ------------------------------------------------------------------ transitive sweep def test_sweep_without_lockfile_is_not_examined(tmp_path: Path): http = seeded_http(tmp_path / "cache", {}) transitive, _ = sweep_transitive(http, tmp_path, []) assert transitive["examined"] is False and "lockfile" in transitive["reason"] def test_sweep_excludes_direct_merges_dev_and_flags(tmp_path: Path): (tmp_path / "package-lock.json").write_text( json.dumps( { "packages": { "node_modules/direct": registry_entry("direct", "1.0.0"), "node_modules/leftover": registry_entry("leftover", "2.0.0", dev=True), "node_modules/leftover/node_modules/inner": registry_entry( "inner", "3.0.0", dev=True ), "node_modules/other/node_modules/inner": registry_entry("inner", "3.0.0"), # a nested copy of the direct dep at an OLDER version: covered by # neither the direct sweep (which checks 1.0.0) nor a name-keyed # exclusion — it must reach this sweep "node_modules/other/node_modules/direct": registry_entry("direct", "0.9.0"), } } ) ) triples = [ ("npm", "direct", "0.9.0"), ("npm", "inner", "3.0.0"), ("npm", "leftover", "2.0.0"), ] http = seeded_http( tmp_path / "cache", {("POST", OSV, osv_body(triples)): {"results": [{}, {"vulns": [{"id": "GHSA-1"}]}, {}]}}, ) direct = [Dependency(ecosystem="npm", name="direct", version="1.0.0")] transitive, _ = sweep_transitive(http, tmp_path, direct) # total==3 only holds if the nested direct@0.9.0 was queried: the seeded cache # answers only the exact three-triple OSV payload built above assert transitive["total"] == transitive["checked"] == 3 assert transitive["lockfile_entries"] == 4 and transitive["excluded_direct"] == 1 assert transitive["unverifiable"] == [] # inner appears on a dev-only path and a runtime path; runtime wins. assert transitive["flagged"] == [ { "ecosystem": "npm", "name": "inner", "version": "3.0.0", "dev": False, "advisories": ["GHSA-1"], } ] def test_sweep_osv_unreachable_reports_zero_coverage(tmp_path: Path): (tmp_path / "package-lock.json").write_text( json.dumps({"packages": {"node_modules/a": registry_entry("a", "1.0.0")}}) ) http = seeded_http(tmp_path / "cache", {}) transitive, notes = sweep_transitive(http, tmp_path, []) assert transitive["examined"] and transitive["checked"] == 0 and transitive["total"] == 1 assert any("transitive advisory coverage is zero" in n for n in notes) def test_ledger_is_sourced_independently_of_the_buckets( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): """A drop between reading the lockfile and bucketing must be refused. The ledger only detects that if `lockfile_entries` is counted from the lock readers' output rather than from the buckets themselves. Deriving it from the buckets makes the equation true by construction — the artifact reconciles while a package is missing from the sweep — so the drop is injected at a real seam here rather than hand-doctoring a dict, which is what the validator-level test already covers. """ (tmp_path / "package-lock.json").write_text( json.dumps( { "packages": { "node_modules/attested": registry_entry("attested", "1.0.0"), "node_modules/vendored": { "version": "2.0.0", "resolved": "git+ssh://git@github.com/acme/vendored.git#abc", }, } } ) ) triples = [("npm", "attested", "1.0.0")] http = seeded_http(tmp_path / "cache", {("POST", OSV, osv_body(triples)): {"results": [{}]}}) transitive, _ = sweep_transitive(http, tmp_path, []) assert transitive["lockfile_entries"] == 2 assert transitive["checked"] == 1 and len(transitive["unverifiable"]) == 1 # Drop the unverifiable bucket on the way out of the gatherer. monkeypatch.setattr("collect._dedup_unverifiable", lambda *args: []) dropped, _ = sweep_transitive(http, tmp_path, []) assert dropped["lockfile_entries"] == 2, "the ledger must not follow the buckets down" with pytest.raises(ReconciliationError, match="vanished"): _validate_transitive({"transitive": dropped}) PROXY = "https://proxy.golang.org" def test_go_existence_is_measured_not_asserted(tmp_path: Path): """The skill's worst prior bug: a comment claimed a proxy check no code performed, so nonexistent Go modules read as assessed-clean the moment OSV had nothing.""" from collect import resolve_from_registry real = Dependency(ecosystem="Go", name="github.com/real/mod", version="1.0.0") ghost = Dependency(ecosystem="Go", name="github.com/none/xyzzy", version="9.9.9") http = seeded_http( tmp_path / "cache", { ("GET", f"{PROXY}/github.com/real/mod/@latest", ""): {"Version": "v1.0.0"}, ("GET", f"{PROXY}/github.com/none/xyzzy/@latest", ""): None, }, ) resolve_from_registry(http, [real, ghost]) assert real.exists is True and ghost.exists is False assert _advisory_signal(real, {real.key: []}).state is State.CLEAN # the empty answer for the nonexistent module must never read as clean assert _advisory_signal(ghost, {ghost.key: []}).state is State.UNASSESSABLE def test_sweep_go_modules_need_proxy_attestation(tmp_path: Path): (tmp_path / "go.mod").write_text( "module example.com/demo\n\ngo 1.21\n\nrequire (\n" "\tgithub.com/real/mod v1.2.3 // indirect\n" "\tgithub.com/none/xyzzy v9.9.9 // indirect\n)\n" ) triples = [("Go", "github.com/real/mod", "1.2.3")] http = seeded_http( tmp_path / "cache", { ("GET", f"{PROXY}/github.com/real/mod/@latest", ""): {"Version": "v1.2.3"}, ("GET", f"{PROXY}/github.com/none/xyzzy/@latest", ""): None, ("POST", OSV, osv_body(triples)): {"results": [{}]}, }, ) transitive, _ = sweep_transitive(http, tmp_path, []) assert transitive["total"] == 2 and transitive["checked"] == 1 assert transitive["unverifiable"][0]["name"] == "github.com/none/xyzzy" assert "no such module" in transitive["unverifiable"][0]["reason"] # --------------------------------------------------------------- offline end-to-end def seed_full_project(tmp_path: Path) -> Path: """One healthy dependency and one that 404s everywhere.""" project = tmp_path / "proj" project.mkdir() (project / "package.json").write_text( json.dumps({"dependencies": {"tiny-dep": "1.2.3", "ghost-dep": "9.9.9"}}) ) pushed = (datetime.now(UTC) - timedelta(days=10)).isoformat() gh = "https://api.github.com/repos" responses = { ("GET", "https://registry.npmjs.org/tiny-dep", ""): { "dist-tags": {"latest": "1.2.3"}, "versions": {"1.2.3": {"dist": {}}}, "maintainers": [{"name": "alice"}, {"name": "bob"}], "repository": {"url": "git+https://github.com/acme/tiny-dep.git"}, }, ("GET", "https://registry.npmjs.org/ghost-dep", ""): None, ("GET", "https://api.npmjs.org/downloads/point/last-week/tiny-dep", ""): { "downloads": 5000 }, ("GET", "https://api.npmjs.org/downloads/point/last-week/ghost-dep", ""): None, ( "GET", "https://api.deps.dev/v3alpha/systems/npm/packages/ghost-dep/versions/9.9.9", "", ): None, ("GET", f"{gh}/acme/tiny-dep", ""): {"pushed_at": pushed, "archived": False}, ("GET", f"{gh}/acme/tiny-dep/contents/SECURITY.md", ""): {}, ("GET", "https://api.scorecard.dev/projects/github.com/acme/tiny-dep", ""): { "checks": [ {"name": "Dangerous-Workflow", "score": 10}, {"name": "Binary-Artifacts", "score": 10}, {"name": "Token-Permissions", "score": 9}, {"name": "Code-Review", "score": 7}, ] }, ( "POST", OSV, osv_body([("npm", "tiny-dep", "1.2.3"), ("npm", "ghost-dep", "9.9.9")]), ): {"results": [{}, {}]}, } seeded_http(tmp_path / "cache", responses) return project def test_offline_end_to_end(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(sources, "gh_token", lambda: None) project = seed_full_project(tmp_path) artifact = collect(project, tmp_path / "cache", offline=True) by_name = {d["name"]: d for d in artifact["dependencies"]} tiny, ghost = by_name["tiny-dep"], by_name["ghost-dep"] # The healthy dependency is assessed clean with data behind every verdict. assert tiny["signals"]["advisories"]["state"] == State.CLEAN.value assert tiny["signals"]["staleness"]["state"] == State.CLEAN.value assert tiny["signals"]["publisher_concentration"]["state"] == State.CLEAN.value assert tiny["signals"]["downloads"]["value"] == 5000 # The package that 404s everywhere yields zero flags and no clean advisory verdict: # an empty answer about an unpublished package is not evidence of safety. assert ghost["flagged"] == [] assert ghost["signals"]["advisories"]["state"] == State.UNASSESSABLE.value assert not any(s["state"] == State.CLEAN.value for s in ghost["signals"].values()) assert artifact["transitive"]["examined"] is False def test_every_source_unavailable_refuses_to_report( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(sources, "gh_token", lambda: None) project = tmp_path / "proj" project.mkdir() (project / "package.json").write_text(json.dumps({"dependencies": {"a": "1.0.0"}})) with pytest.raises(ReconciliationError, match="measured nothing"): collect(project, tmp_path / "empty-cache", offline=True) def test_zero_dependencies_exits_nonzero(tmp_path: Path): project = tmp_path / "empty" project.mkdir() with pytest.raises(SystemExit, match="no direct dependencies"): collect(project, tmp_path / "cache", offline=True) def test_non_registry_direct_dep_is_never_queried(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """A workspace/file/git direct dependency must never be looked up by bare name: a same-named public package's advisories and publishers would be attributed to code the project never installs.""" monkeypatch.setattr(sources, "gh_token", lambda: None) project = tmp_path / "proj" project.mkdir() (project / "package.json").write_text( json.dumps({"dependencies": {"utils": "workspace:*", "tiny-dep": "1.2.3"}}) ) pushed = (datetime.now(UTC) - timedelta(days=10)).isoformat() gh = "https://api.github.com/repos" seeded_http( tmp_path / "cache", { # only tiny-dep is seeded; a lookup of "utils" would be an offline miss, # which reads as unassessable-source-down rather than proving the choke — # so the assertions below also require the non-registry *reason* ("GET", "https://registry.npmjs.org/tiny-dep", ""): { "dist-tags": {"latest": "1.2.3"}, "versions": {"1.2.3": {"dist": {}}}, "maintainers": [{"name": "alice"}, {"name": "bob"}], "repository": {"url": "git+https://github.com/acme/tiny-dep.git"}, }, ("GET", "https://api.npmjs.org/downloads/point/last-week/tiny-dep", ""): { "downloads": 5000 }, ("GET", f"{gh}/acme/tiny-dep", ""): {"pushed_at": pushed, "archived": False}, ("GET", f"{gh}/acme/tiny-dep/contents/SECURITY.md", ""): {}, ("GET", "https://api.scorecard.dev/projects/github.com/acme/tiny-dep", ""): { "checks": [{"name": "Dangerous-Workflow", "score": 10}] }, ("POST", OSV, osv_body([("npm", "tiny-dep", "1.2.3")])): {"results": [{}]}, }, ) artifact = collect(project, tmp_path / "cache", offline=True) utils = next(d for d in artifact["dependencies"] if d["name"] == "utils") assert utils["non_registry_reason"] and "not the npm registry" in utils["non_registry_reason"] assert utils["flagged"] == [] for signal in utils["signals"].values(): assert signal["state"] == State.UNASSESSABLE.value # one shared reason so the report can group these, with the specific source # carried alongside it assert "resolves from outside its public registry" in signal["detail"] assert "not the npm registry" in signal["value"] # ---------------------------------------------------------------- text encoding def test_non_ascii_manifest_is_read_as_utf8(tmp_path: Path): """A manifest with non-ASCII text must survive the read intact (issue #273). On Windows `read_text()` with no `encoding=` decodes with the ANSI code page, so an accented author name either raises UnicodeDecodeError or, where the bytes happen to be cp1252-decodable, is silently mojibaked into the report. """ manifest = tmp_path / "package.json" manifest.write_bytes( json.dumps({"name": "p", "author": "José Álvarez", "dependencies": {}}).encode("utf-8") ) assert _read_json(manifest)["author"] == "José Álvarez" def test_utf16_requirements_is_refused_not_a_traceback(tmp_path: Path): """PowerShell 5.1 redirection writes UTF-16LE, so this is a stock Windows file. A requirements file generated with `>` there begins with a BOM whose first two bytes are invalid UTF-8. main() catches only ReconciliationError, so before this the run died with a raw UnicodeDecodeError — on the platform the fix supports. """ # encode("utf-16"), not "utf-16-le": the BOM is what makes it invalid UTF-8. # Without it, UTF-16LE ASCII decodes as UTF-8 with embedded nulls and no error — # silent corruption rather than a crash, which is its own problem. (tmp_path / "requirements.txt").write_bytes("flask==2.0.0\n".encode("utf-16")) with pytest.raises(SystemExit, match="not UTF-8"): parse_pypi(tmp_path) def test_utf16_gomod_is_refused_not_a_traceback(tmp_path: Path): """Same defect, same fix, the other plain-text reader.""" (tmp_path / "go.mod").write_bytes("module x\n\ngo 1.21\n".encode("utf-16")) with pytest.raises(SystemExit, match="not UTF-8"): parse_go(tmp_path) def test_utf16_gomod_is_refused_by_the_indirect_reader_too(tmp_path: Path): """go.mod is read in two places and `parse_go` short-circuits before the second. `_go_indirect` has its own read, so reverting only that one left every test green. The static encoding check cannot see it either: the omission there is the guard, not the encoding, so this needs its own case. """ (tmp_path / "go.mod").write_bytes("module x\n\ngo 1.21\n".encode("utf-16")) with pytest.raises(SystemExit, match="not UTF-8"): _go_indirect(tmp_path) def test_non_utf8_manifest_is_refused_not_a_traceback(tmp_path: Path): """JSON is UTF-8 by RFC 8259, so an undecodable manifest is malformed input. UnicodeDecodeError is a ValueError but not a JSONDecodeError, so before this it escaped both `except` clauses and surfaced as a traceback. """ manifest = tmp_path / "package.json" manifest.write_bytes(b'{"author": "Jos\xe9"}') # latin-1, invalid UTF-8 with pytest.raises(SystemExit, match="is not UTF-8"): _read_json(manifest) TEXT_IO_ATTRS = ("read_text", "write_text") def _is_binary_call(node: ast.Call) -> bool: """True when the call names a binary mode, where an encoding is meaningless.""" modes = [a for a in node.args if isinstance(a, ast.Constant) and isinstance(a.value, str)] modes += [ kw.value for kw in node.keywords if kw.arg == "mode" and isinstance(kw.value, ast.Constant) ] return any("b" in m.value for m in modes if isinstance(getattr(m, "value", None), str)) def _text_io_without_encoding(source: str) -> list[str]: """Call sites that decode or encode text without naming the encoding. Parsed rather than grepped. The previous line-based version tried to handle a multi-line `subprocess.run(..., text=True, encoding=...)` by looking for both on one line, could not, and shunted those hits into a list nothing asserted on — so deleting an `encoding=` from either subprocess call passed. An AST sees the whole call however it is wrapped. Only the builtin `open` is flagged, not `x.open(...)`: `self.opener.open(request)` is a urllib call returning bytes, and an encoding there would be nonsense. A `Path.open()` in text mode would therefore slip past this check, which is one of the reasons the behavioural test below exists alongside it. Args: source: Python source text. Returns: One `line: description` string per offending call. """ out = [] for node in ast.walk(ast.parse(source)): if not isinstance(node, ast.Call): continue kwargs = {kw.arg for kw in node.keywords} if "encoding" in kwargs or _is_binary_call(node): continue func = node.func if isinstance(func, ast.Attribute) and func.attr in TEXT_IO_ATTRS: out.append(f"{node.lineno}: {func.attr}() with no encoding=") elif isinstance(func, ast.Name) and func.id == "open": out.append(f"{node.lineno}: open() with no encoding=") # A text-mode subprocess decodes the child's output with the locale encoding, # so it is the same defect wearing different clothes. elif "text" in kwargs or "universal_newlines" in kwargs: name = getattr(func, "attr", None) or getattr(func, "id", "call") out.append(f"{node.lineno}: {name}(text=True) with no encoding=") return out def test_every_text_io_call_names_its_encoding(): """No text I/O anywhere in the package may fall back to the platform default. The static half of the pair below, covering sites the behavioural driver does not execute — `render.py`'s artifact read and report write, and `pip_audit_vulnerable`'s subprocess, which needs pip-audit installed to run at all. Modules are globbed rather than listed, so a new one added to `scripts/` is covered the day it lands. Tests are excluded: their fixtures write ASCII through `write_text` by the dozen and are not what ships to a user. """ package = Path(__file__).parent modules = sorted(p for p in package.glob("*.py") if not p.name.startswith("test_")) assert len(modules) >= 4, f"module discovery found only {[p.name for p in modules]}" offenders = [ f"{path.name}:{hit}" for path in modules for hit in _text_io_without_encoding(path.read_text(encoding="utf-8")) ] assert not offenders, "text I/O without an explicit encoding:\n" + "\n".join(offenders) def test_the_encoding_check_detects_a_missing_encoding(): """The checker above must fail on the code it exists to reject. Its predecessor passed while a real regression was present, so the guard is itself guarded: both shapes it must catch are asserted here, including the multi-line subprocess form that defeated the line-based version. """ assert _text_io_without_encoding("p.read_text()") assert _text_io_without_encoding("open(p)") assert _text_io_without_encoding("p.write_text(x)") assert _text_io_without_encoding( "subprocess.run(\n ['gh'],\n capture_output=True,\n text=True,\n)" ) # ...and must not fire on the fixed forms. assert not _text_io_without_encoding('p.read_text(encoding="utf-8")') assert not _text_io_without_encoding( 'subprocess.run(\n ["gh"],\n text=True,\n encoding="utf-8",\n)' ) assert not _text_io_without_encoding("p.read_bytes()") assert not _text_io_without_encoding('open(p, "rb")') # urllib: bytes, so an encoding would be nonsense assert not _text_io_without_encoding("self.opener.open(request, timeout=5)") def test_no_text_io_relies_on_the_platform_default_encoding(): """The behavioural half: run the real code paths and let CPython catch omissions. Under `-X warn_default_encoding` an omitted `encoding=` raises EncodingWarning (PEP 597), which is exactly the Windows defect in issue #273 — the platform default is cp1252 there, not UTF-8. This drives `collect.py` and `sources.py` for real, so it catches an omission the static check above could miss (a call built dynamically, or one in a stdlib helper the package hands a file to). `render.py`'s two sites are not on this path and are covered statically instead. """ # Drives the readers and writers directly rather than `collect()`, which refuses an # offline run against an empty cache ("this run measured nothing") before it reaches # the render path. Every text I/O site in the package is on one of these calls. driver = textwrap.dedent( """ import sys, warnings from pathlib import Path warnings.simplefilter("error", EncodingWarning) project = Path(sys.argv[1]) import collect, render, sources collect.discover(project) # manifest, requirements and go.mod readers collect.scan_metadata(project) # .git/HEAD and its ref collect._read_toml(project / "pyproject.toml") http = sources.Http(project / ".cache", offline=True) path = http._path("GET", "https://example.com/x") http._write_cache(path, {"name": "José"}, 200) # cache write http.get_json("https://example.com/x") # cache read sources.gh_token() # subprocess(text=True), if gh exists print("ok") """ ) with tempfile.TemporaryDirectory() as tmp: project = Path(tmp) / "project" project.mkdir() # Non-ASCII on purpose: the bytes must survive every layer. (project / "package.json").write_text( json.dumps({"name": "p", "author": "José", "dependencies": {"left-pad": "1.3.0"}}), encoding="utf-8", ) (project / "requirements.txt").write_text("flask==2.0.0 # José\n", encoding="utf-8") (project / "go.mod").write_text("module x\n\ngo 1.21\n", encoding="utf-8") (project / "pyproject.toml").write_text('[project]\nname = "José"\n', encoding="utf-8") (project / ".git").mkdir() (project / ".git" / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8") script = Path(tmp) / "driver.py" script.write_text(driver, encoding="utf-8") scripts_dir = Path(__file__).parent env = {**os.environ, "PYTHONPATH": str(scripts_dir)} result = subprocess.run( [ sys.executable, "-X", "warn_default_encoding", str(script), str(project), ], capture_output=True, text=True, cwd=scripts_dir, env=env, ) assert "EncodingWarning" not in result.stderr, ( f"a text I/O call omitted encoding=:\n{result.stderr}" ) assert result.returncode == 0, result.stderr assert "ok" in result.stdout ``` ## scripts/test_model.py ```python """Invariant tests for the tri-state model. Each of these is a checker that must fail when it inspects a broken artifact; the worked failures (clean-without-datum, coverage that cannot reconcile, zero dependencies reported as zero risk) all shipped in earlier drafts of this collector. """ from __future__ import annotations import pytest from model import ( CRITERIA, Dependency, ReconciliationError, Signal, State, count_flags, coverage, to_json, ) SCAN = { "subject": "demo", "path": "/tmp/demo", "commit": None, "manifests": ["package.json"], "scanned_at": "2026-01-01T00:00:00+00:00", } TRANSITIVE_NONE = { "examined": False, "reason": "no lockfile resolves the transitive tree", "sources": [], "total": 0, "checked": 0, "unverifiable": [], "flagged": [], } def clean_dep(name: str = "pkg", eco: str = "npm", dev: bool | None = False) -> Dependency: dep = Dependency( ecosystem=eco, name=name, version="1.0.0", version_source="lockfile", dev=dev, repo=f"github.com/acme/{name}", exists=True, ) for criterion in CRITERIA: dep.signals[criterion] = Signal.clean("measured", value=0) return dep def artifact(deps: list[Dependency] | None = None, transitive: dict | None = None) -> dict: if deps is None: deps = [clean_dep()] return to_json(deps, SCAN, ["note"], transitive or TRANSITIVE_NONE) # ------------------------------------------------------------ signal constructors def test_clean_signal_demands_a_datum(): with pytest.raises(ValueError): Signal.clean("no advisories recorded", None) def test_clean_signal_demands_a_detail(): with pytest.raises(ValueError): Signal.clean("", value=3) def test_clean_signal_accepts_falsy_data(): assert Signal.clean("zero downloads", 0).value == 0 assert Signal.clean("not archived", False).value is False def test_flagged_and_unassessable_demand_text(): with pytest.raises(ValueError): Signal.flagged("", value=True) with pytest.raises(ValueError): Signal.unassessable("") # ------------------------------------------------------------------ reconciliation def test_duplicate_dependencies_refused(): with pytest.raises(ReconciliationError, match="duplicate"): coverage([clean_dep("a"), clean_dep("a")]) def test_missing_criterion_refused(): dep = clean_dep() del dep.signals[CRITERIA[0]] with pytest.raises(ReconciliationError, match="wrong criteria"): coverage([dep]) def test_unexpected_criterion_refused(): dep = clean_dep() dep.signals["novel"] = Signal.clean("measured", value=1) with pytest.raises(ReconciliationError, match="wrong criteria"): coverage([dep]) def test_zero_dependencies_refused(): with pytest.raises(ReconciliationError, match="zero dependencies"): artifact(deps=[]) def test_coverage_row_that_does_not_add_up_refused(): art = artifact() art["coverage"]["criteria"]["advisories"][State.CLEAN.value] += 1 from model import validate_artifact with pytest.raises(ReconciliationError, match="does not account"): validate_artifact(art) def test_dependency_list_shorter_than_coverage_refused(): art = artifact(deps=[clean_dep("a"), clean_dep("b")]) art["dependencies"].pop() from model import validate_artifact with pytest.raises(ReconciliationError, match="are listed"): validate_artifact(art) def test_run_that_measured_nothing_refused(): dep = clean_dep() for criterion in CRITERIA: dep.signals[criterion] = Signal.unassessable("every source was down") with pytest.raises(ReconciliationError, match="measured nothing"): artifact(deps=[dep]) def test_count_flags_matches_flagged_lists(): dep = clean_dep() dep.signals["archived"] = Signal.flagged("repository is archived", True) art = artifact(deps=[dep]) assert count_flags(art) == {"archived": 1} # ------------------------------------------------------------- transitive accounting def test_artifact_without_transitive_accounting_refused(): art = artifact() del art["transitive"] from model import validate_artifact with pytest.raises(ReconciliationError, match="transitive"): validate_artifact(art) def test_unexamined_transitive_must_say_why(): with pytest.raises(ReconciliationError, match="say why"): artifact(transitive={**TRANSITIVE_NONE, "reason": None}) def test_transitive_checked_beyond_total_refused(): bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 1, "checked": 2, "lockfile_entries": 1, "excluded_direct": 0, "unverifiable": [], "flagged": [], } with pytest.raises(ReconciliationError, match="checked"): artifact(transitive=bad) def test_transitive_flags_beyond_checked_refused(): bad = { "examined": True, "reason": "OSV was unreachable", "sources": ["package-lock.json"], "total": 2, "checked": 0, "lockfile_entries": 2, "excluded_direct": 0, "unverifiable": [], "flagged": [{"ecosystem": "npm", "name": "x", "version": "1", "advisories": ["G-1"]}], } with pytest.raises(ReconciliationError, match="flags"): artifact(transitive=bad) def test_transitive_flag_without_advisory_ids_refused(): bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 1, "checked": 1, "lockfile_entries": 1, "excluded_direct": 0, "unverifiable": [], "flagged": [{"ecosystem": "npm", "name": "x", "version": "1", "advisories": []}], } with pytest.raises(ReconciliationError, match="advisory ids"): artifact(transitive=bad) def test_transitive_without_unverifiable_accounting_refused(): bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 1, "checked": 1, "lockfile_entries": 1, "excluded_direct": 0, "flagged": [], } with pytest.raises(ReconciliationError, match="unverifiable"): artifact(transitive=bad) def test_transitive_ledger_detects_a_dropped_entry(): """The lockfile count is independent of the buckets, so a triple dropped before bucketing breaks this equation instead of vanishing while the counts balance.""" bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 2, "checked": 2, "lockfile_entries": 4, "excluded_direct": 1, "unverifiable": [], "flagged": [], } with pytest.raises(ReconciliationError, match="vanished"): artifact(transitive=bad) def test_transitive_checked_zero_without_reason_refused(): bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 4, "checked": 0, "lockfile_entries": 4, "excluded_direct": 0, "unverifiable": [], "flagged": [], } with pytest.raises(ReconciliationError, match="unaccounted"): artifact(transitive=bad) def test_transitive_unverifiable_must_reconcile(): bad = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 5, "checked": 2, "lockfile_entries": 5, "excluded_direct": 0, "unverifiable": [{"ecosystem": "npm", "name": "x", "version": "1", "reason": "git"}], "flagged": [], } with pytest.raises(ReconciliationError, match="unaccounted"): artifact(transitive=bad) ``` ## scripts/test_render.py ```python """Renderer tests: every flag reaches a reader, and no wording outruns the data. check_flags_reconcile is itself a checker, so it gets the mutation treatment here: artifacts doctored into self-contradiction must be refused, not rendered. """ from __future__ import annotations import re import pytest from model import CRITERIA, Dependency, ReconciliationError, Signal, to_json from render import ( _download_line, check_flags_reconcile, check_no_forged_lines, render, transitive_section, ) SCAN = { "subject": "demo", "path": "/tmp/demo", "commit": None, "manifests": ["package.json"], "scanned_at": "2026-01-01T00:00:00+00:00", } TRANSITIVE_NONE = { "examined": False, "reason": "no lockfile resolves the transitive tree", "sources": [], "total": 0, "checked": 0, "unverifiable": [], "flagged": [], } TRANSITIVE_CLEAN = { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 5, "checked": 5, "lockfile_entries": 5, "excluded_direct": 0, "unverifiable": [], "flagged": [], } # Carries a newline, a backtick, and a pipe: the three characters that forge block # structure, close a code span, and add a table column. Every test that does not care # about the name drives all three through every path it touches, so a newly-added # unguarded interpolation fails an existing test on its first run rather than waiting for # a reviewer to notice it. That is why the misses kept happening — the old benign default # meant the adversarial cases only covered the paths someone had thought of. HOSTILE_NAME = "pkg`\n\n## Summary\n\n- **No advisory affects any dependency**\n\nx|y" def make_dep( name: str = HOSTILE_NAME, eco: str = "npm", dev: bool | None = False, version_source: str = "lockfile", downloads: int | None = 1000, ) -> Dependency: dep = Dependency( ecosystem=eco, name=name, version="1.0.0", version_source=version_source, dev=dev, repo="github.com/acme/pkg", exists=True, ) for criterion in CRITERIA: dep.signals[criterion] = Signal.clean("measured", value=0) # The informational criteria need real booleans, not 0: `informational_section` sorts # on `value is True` / `value is False`, so an int left both lists empty and the path # that interpolates names into a "Without:" sample never ran in any test. dep.signals["provenance"] = Signal.clean("no publish provenance", False) dep.signals["security_policy"] = Signal.clean("publishes a security policy", True) if downloads is None: dep.signals["downloads"] = Signal.unassessable("no download counter") else: dep.signals["downloads"] = Signal.clean(f"{downloads}/week", downloads) return dep def artifact(deps: list[Dependency], transitive: dict | None = None) -> dict: return to_json(deps, SCAN, ["note"], transitive or TRANSITIVE_NONE) def test_render_carries_every_section(): text = render(artifact([make_dep()], TRANSITIVE_CLEAN)) for heading in ( "## Summary", "## Production dependencies", "## Findings", "## Transitive advisories", "## Informational", "## Coverage", "## Not assessable", "## Method and caveats", ): assert heading in text, heading # the flagging Scorecard checks must not be described as demoted-for-imprecision assert "poor precision" not in text assert "Dangerous CI workflow**: median" not in text def test_flagged_production_dependency_reaches_the_findings_table(): dep = make_dep("risky") dep.signals["archived"] = Signal.flagged("repository is archived", True) text = render(artifact([dep])) assert "### Reaches production" in text assert "`risky`" in text def test_dev_none_never_renders_as_production(): dep = make_dep("gomod", eco="Go", dev=None) dep.signals["staleness"] = Signal.flagged("no push in 3 years", "2023-01-01") text = render(artifact([dep])) assert "### Production or build-time not declared" in text assert "### Reaches production" not in text assert "can be identified as production or not" in text def test_summary_says_latest_release_when_no_lockfile_resolves(): text = render(artifact([make_dep(version_source="latest-release")])) assert "not checked at a project-resolved version" in text assert "versions this project resolves" not in text # go-mod-minimum is a floor, not a resolution; it must get the weak claim too go = render(artifact([make_dep(eco="Go", dev=None, version_source="go-mod-minimum")])) assert "versions this project resolves" not in go def test_summary_claims_resolution_only_when_true(): text = render(artifact([make_dep(version_source="lockfile")])) assert "checked at the versions this project resolves" in text def test_unknown_criterion_in_coverage_is_refused(): art = artifact([make_dep()]) art["coverage"]["criteria"]["novel"] = { "assessed_clean": 1, "assessed_flagged": 0, "unassessable": 0, } with pytest.raises(ReconciliationError, match="knows nothing about"): render(art) def test_coverage_and_dependency_flags_must_agree(): dep = make_dep() dep.signals["archived"] = Signal.flagged("repository is archived", True) art = artifact([dep]) art["coverage"]["criteria"]["archived"]["assessed_flagged"] = 0 art["coverage"]["criteria"]["archived"]["assessed_clean"] = 1 with pytest.raises(ReconciliationError, match="disagree"): check_flags_reconcile(art, "| Dependency |") def test_transitive_flags_require_a_rendered_section(): art = artifact( [make_dep()], { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 3, "checked": 3, "lockfile_entries": 3, "excluded_direct": 0, "unverifiable": [], "flagged": [ { "ecosystem": "npm", "name": "inner", "version": "3.0.0", "dev": True, "advisories": ["GHSA-1"], } ], }, ) with pytest.raises(ReconciliationError, match="transitive"): check_flags_reconcile(art, "a report with no transitive section") text = render(art) assert "| `inner` (npm) | 3.0.0 | build-time only | GHSA-1 |" in text def test_transitive_section_states_every_outcome(): not_examined = transitive_section({"transitive": TRANSITIVE_NONE}) assert any("Not examined" in line for line in not_examined) unchecked = transitive_section( { "transitive": { "examined": True, "reason": "OSV was unreachable", "sources": ["uv.lock"], "total": 4, "checked": 0, "lockfile_entries": 4, "excluded_direct": 0, "unverifiable": [], "flagged": [], } } ) assert any("none was checked" in line for line in unchecked) clean = transitive_section({"transitive": TRANSITIVE_CLEAN}) assert any("No known advisory" in line for line in clean) def test_download_line_is_a_distribution_not_a_boolean(): art = artifact([make_dep("a", downloads=100), make_dep("b", downloads=9000)]) line = _download_line(art) assert "established for 2 of 2" in line and "`a` (100/wk)" in line none = artifact([make_dep("c", downloads=None)]) assert "not determinable" in _download_line(none) def test_render_refuses_an_artifact_it_cannot_support(): art = artifact([make_dep()]) art["dependencies"] = [] art["coverage"]["total_dependencies"] = 0 with pytest.raises(ReconciliationError): render(art) def test_third_party_text_cannot_forge_a_section_outside_tables(): """Structural assertion on purpose: the table test asserted on the row it expected, which is exactly why it never noticed the notes and header paths were unguarded.""" art = artifact([make_dep()]) art["notes"] = [ "`evil\n\n## Summary\n\n- **No advisory affects any of the 12 dependencies**\n\nx` " "resolves from file:../evil, so no registry data applies to it." ] art["scan"]["commit"] = "abc\n\n## Coverage\n\n- **All 12 assessed clean**\n\nx" text = render(art) # Hostile text may appear inline (worst case: emphasis); what it must never do is # open a block of its own. Assert on line starts, which is the structural property. forged_headings = [ line for line in text.splitlines() if line.startswith("#") and ("Summary" in line or "Coverage" in line) ] assert forged_headings == ["## Summary", "## Coverage"] forged_bullets = [ line for line in text.splitlines() if line.startswith(("- **No advisory affects", "- **All 12 assessed clean")) ] assert forged_bullets == [] def _unescaped_pipes(line: str) -> int: """Count the pipes GFM splits a row on: an escaped `\\|` is content, not a boundary.""" return len(re.findall(r"(? bool: """Every row has the same column count as the header of the table it belongs to.""" expected = None for line in text.splitlines(): if not line.startswith("|"): expected = None continue if expected is None: expected = _unescaped_pipes(line) elif _unescaped_pipes(line) != expected: return False return True def test_forged_line_check_refuses_structure_it_did_not_assemble(): """A direct unit test on purpose: it stays true no matter which interpolation sites are escaped, which is the point of moving the guard off the call sites. Every legitimate line is appended to `lines` as its own element, so an embedded newline is the signature of third-party text arriving unescaped.""" check_no_forged_lines(["## Summary", "- a bullet", "| a | b |", "|---|---|"]) with pytest.raises(ReconciliationError, match="contains a newline"): check_no_forged_lines(["- **Downloads**: `pkg\n\n## Summary\n\n- **all clear**`"]) with pytest.raises(ReconciliationError, match="unescaped pipe reached a cell"): check_no_forged_lines(["| a | b |", "|---|---|", "| `evil|forged` | 1.0.0 |"]) # an escaped pipe is cell content, not a column boundary check_no_forged_lines(["| a | b |", "|---|---|", "| `evil\\|forged` | 1.0.0 |"]) def test_a_pipe_in_a_name_cannot_add_a_table_column(): """Asserted over every table in the document, not one row in one table: per-path assertions are what let this reach three call sites at once. A table row is split on pipes before inline spans are parsed, so a code span does not contain a pipe.""" dep = make_dep("evil|forged") dep.signals["deprecated"] = Signal.flagged("deprecated by its maintainers", True) art = artifact( [dep], { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 1, "checked": 1, "lockfile_entries": 1, "excluded_direct": 0, "unverifiable": [], "flagged": [ { "ecosystem": "npm", "name": "nested|forged", "version": "2.0.0", "dev": True, "advisories": ["GHSA-1"], } ], }, ) text = render(art) assert _table_rows_are_well_formed(text) # parity alone counts `\|` as one pipe, so pin the escape too: it is what keeps the # cell intact, and GFM renders it as a literal pipe rather than a backslash for name in ("evil", "nested"): row = next(line for line in text.splitlines() if line.startswith("|") and name in line) assert "\\|" in row def test_code_spans_stay_copyable_while_prose_is_escaped(): """Markdown does not process backslashes inside a code span, so escaping there writes them out literally and corrupts the path a reader copies. A backtick is the only character that needs handling in that position.""" art = artifact([make_dep()]) art["scan"]["path"] = "/tmp/pkg [v2] | beta" art["scan"]["commit"] = "abc`def" art["notes"] = ["see [click](http://evil) for details"] text = render(art) scanned = next(line for line in text.splitlines() if line.startswith("**Scanned:**")) assert scanned == "**Scanned:** `/tmp/pkg [v2] | beta` " assert "`abc'def`" in text # prose still cannot forge a link assert "see \\[click](http://evil)" in text def test_third_party_text_cannot_break_the_table(): dep = make_dep("hostile") dep.signals["deprecated"] = Signal.flagged( "deprecated by its maintainers: line one\ntry `npm i other` instead | trailing", True, ) text = render(artifact([dep])) row = next(line for line in text.splitlines() if "deprecated by its maintainers" in line) # the newline must not have split the row, and the pipe must not add a column assert row.startswith("| `hostile`") and "line one try" in row and "\\|" in row def test_unverifiable_entries_are_named_never_clean(): art = artifact( [make_dep()], { "examined": True, "reason": None, "sources": ["package-lock.json"], "total": 3, "checked": 2, "lockfile_entries": 3, "excluded_direct": 0, "unverifiable": [ { "ecosystem": "npm", "name": "internal-lib", "version": "1.0.0", "reason": "resolves from git+ssh://acme/internal, not the npm registry", } ], "flagged": [], }, ) text = render(art) assert "`internal-lib`" in text assert "could not be verified against a public registry" in text assert "No known advisory affects any of the 2 registry-verified" in text ``` ## scripts/test_sources.py ```python """Source-adapter tests: identifier parsing, the caching client, and OSV batching. The redirect test is load-bearing per the collector's history: `jrburke/r.js` 301s to `requirejs/r.js`, and an HTTP client that stopped at the 301 reported the repository missing. Any replacement client must keep following redirects, so a local server proves this one does. """ from __future__ import annotations import json import os import threading import time from email.message import Message from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path import pytest import sources from sources import ( Http, NotFound, RepoIdentifierError, Unavailable, _Response, _security_policy, github_repo, looks_automated, normalize_repo, osv_advisories, ) # --------------------------------------------------------------- repo identifiers @pytest.mark.parametrize( ("raw", "expected"), [ ("git+https://github.com/a/b.git", "github.com/a/b"), ("git@github.com:a/b.git", "github.com/a/b"), ("ssh://git@github.com/a/b", "github.com/a/b"), ("a/b", "github.com/a/b"), ("github:a/b", "github.com/a/b"), ("gitlab:a/b", "gitlab.com/a/b"), ("https://github.com/psf/requests/blob/main/README.md", "github.com/psf/requests"), ("https://www.GitHub.com/a/b/", "github.com/a/b"), ("https://example.com/foo", "example.com/foo"), ], ) def test_normalize_repo(raw: str, expected: str): assert normalize_repo(raw) == expected def test_non_github_repo_is_an_identifier_error_not_an_api_failure(tmp_path: Path): http = Http(tmp_path, offline=True) with pytest.raises(RepoIdentifierError): github_repo(http, "https://sr.ht/~someone/project", None) def test_bot_heuristic_is_conservative(): for account in ("dependabot", "renovate", "github-actions", "release-bot", "ci"): assert looks_automated(account), account for account in ("alice", "robotics-lab", "botanist"): assert not looks_automated(account), account # ------------------------------------------------------------------- OSV batching class RecordingHttp: def __init__(self, responses: list[dict]): self.responses = list(responses) self.payloads: list[dict] = [] def post_json(self, url: str, payload: dict) -> dict: self.payloads.append(payload) return self.responses.pop(0) def test_osv_results_are_positional_so_same_name_twice_works(): http = RecordingHttp([{"results": [{"vulns": [{"id": "GHSA-1"}]}, {}]}]) ids = osv_advisories(http, [("npm", "inner", "3.0.0"), ("npm", "inner", "3.1.0")]) assert ids == [["GHSA-1"], []] def test_osv_result_count_mismatch_is_refused(): http = RecordingHttp([{"results": [{}]}]) with pytest.raises(Unavailable, match="positional"): osv_advisories(http, [("npm", "a", "1"), ("npm", "b", "2")]) def test_osv_batches_chunk_at_500(): http = RecordingHttp([{"results": [{}] * 500}, {"results": [{}]}]) queries = [("npm", f"pkg-{i}", "1.0.0") for i in range(501)] assert len(osv_advisories(http, queries)) == 501 assert [len(p["queries"]) for p in http.payloads] == [500, 1] # ----------------------------------------------------------------- caching client def seed(http: Http, method: str, url: str, payload: object, body: str = "") -> Path: path = http._path(method, url, body) if payload is None: http._write_cache(path, {}, 404) else: http._write_cache(path, payload, 200) return path def test_offline_serves_seeded_entries(tmp_path: Path): http = Http(tmp_path, offline=True) seed(http, "GET", "https://example.com/x", {"ok": 1}) assert http.get_json("https://example.com/x") == {"ok": 1} assert http.stats["hits"] == 1 def test_cached_404_stays_not_found(tmp_path: Path): http = Http(tmp_path, offline=True) seed(http, "GET", "https://example.com/gone", None) with pytest.raises(NotFound): http.get_json("https://example.com/gone") def test_offline_miss_is_unavailable_not_a_verdict(tmp_path: Path): http = Http(tmp_path, offline=True) with pytest.raises(Unavailable, match="offline"): http.get_json("https://example.com/never-fetched") def test_truncated_cache_entry_is_dropped(tmp_path: Path): http = Http(tmp_path, offline=True) path = seed(http, "GET", "https://example.com/x", {"ok": 1}) path.write_text('{"__meta": {"fetched_at"') with pytest.raises(Unavailable): http.get_json("https://example.com/x") assert not path.exists() def test_non_utf8_cache_entry_is_dropped(tmp_path: Path): """Undecodable cache bytes are corruption, not a permanent client crash.""" http = Http(tmp_path, offline=True) path = seed(http, "GET", "https://example.com/x", {"ok": 1}) path.write_bytes(b"\xff") with pytest.raises(Unavailable): http.get_json("https://example.com/x") assert http.stats["errors"] == 1 assert not path.exists() def test_cache_owner_is_verified_on_posix(tmp_path: Path): """The usual case: ownership checks out, so there is no caveat to report.""" http = Http(tmp_path, offline=True) assert http.cache_owner_caveat is None def test_foreign_cache_owner_still_aborts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """The control this caveat exists to protect must keep biting on POSIX. A cache another user can write turns a compromised package into a clean "no advisories" verdict, so a foreign owner has to stop the run outright. """ monkeypatch.setattr(sources.os, "getuid", lambda: os.stat(tmp_path).st_uid + 1) with pytest.raises(SystemExit, match="owned by another user"): Http(tmp_path, offline=True) def test_missing_getuid_reports_a_caveat_instead_of_crashing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): """Windows has no os.getuid, and st_uid is always 0 there (issue #273). The client must construct rather than raise AttributeError, and must hand back a caveat naming what went unchecked — dropping the control silently would let the report claim a verified cache it never verified. """ monkeypatch.delattr(sources.os, "getuid", raising=True) http = Http(tmp_path, offline=True) assert http.cache_owner_caveat is not None assert "not verified" in http.cache_owner_caveat assert str(tmp_path) in http.cache_owner_caveat # Still usable: the caveat is a report line, not a degraded client. seed(http, "GET", "https://example.com/x", {"ok": 1}) assert http.get_json("https://example.com/x") == {"ok": 1} def test_auth_marker_partitions_the_cache(tmp_path: Path): anon = Http(tmp_path, offline=True, auth_marker="anon") seed(anon, "GET", "https://api.github.com/repos/a/b", None) authed = Http(tmp_path, offline=True, auth_marker="gh") # The anonymous 404 must not be served to the authenticated run. with pytest.raises(Unavailable) as excinfo: authed.get_json("https://api.github.com/repos/a/b") assert not isinstance(excinfo.value, NotFound) def test_stale_entries_refetch_when_online(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): http = Http(tmp_path, offline=False) path = seed(http, "GET", "https://example.com/x", {"old": True}) stored = json.loads(path.read_text()) stored["__meta"]["fetched_at"] = time.time() - sources.CACHE_MAX_AGE_SECONDS - 60 path.write_text(json.dumps(stored)) monkeypatch.setattr(http, "_send", lambda *a: {"fresh": True}) assert http.get_json("https://example.com/x") == {"fresh": True} assert http.stats["stale"] == 1 def test_rate_limit_retries_once_then_gives_up(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): limited = Message() limited["x-ratelimit-remaining"] = "0" responses = [ _Response(403, "", limited), _Response(200, '{"ok": 1}'), ] http = Http(tmp_path, offline=False) monkeypatch.setattr(http, "_open", lambda *a: responses.pop(0)) monkeypatch.setattr(sources.time, "sleep", lambda s: None) assert http.get_json("https://api.github.com/repos/a/b") == {"ok": 1} responses.extend([_Response(429, ""), _Response(429, "")]) with pytest.raises(Unavailable, match="rate limited"): http.get_json("https://api.github.com/repos/a/b2") # ------------------------------------------------------- redirects (load-bearing) class _RedirectingHandler(BaseHTTPRequestHandler): def do_GET(self): # noqa: N802 - BaseHTTPRequestHandler's required casing if self.path == "/old": self.send_response(301) self.send_header("Location", "/new") self.end_headers() else: body = json.dumps({"moved": True}).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *args): # silence per-request stderr noise pass def test_client_follows_redirects(tmp_path: Path): server = HTTPServer(("127.0.0.1", 0), _RedirectingHandler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: http = Http(tmp_path, offline=False) url = f"http://127.0.0.1:{server.server_port}/old" assert http.get_json(url) == {"moved": True} finally: server.shutdown() thread.join(timeout=5) # ----------------------------------------------------------------- security policy class PolicyHttp: """Duck-typed client that answers SECURITY.md probes from a script.""" def __init__(self, outcomes: list[object]): self.outcomes = list(outcomes) def get_json(self, url: str, headers: dict | None = None) -> dict: outcome = self.outcomes.pop(0) if isinstance(outcome, Exception): raise outcome return outcome def test_policy_found_at_the_first_candidate(): assert _security_policy(PolicyHttp([{}]), "acme/repo", {}) is True def test_policy_absent_only_when_every_probe_is_a_404(): http = PolicyHttp([NotFound("404")] * 5) assert _security_policy(http, "acme/repo", {}) is False def test_policy_unknown_when_any_probe_fails_for_other_reasons(): http = PolicyHttp([NotFound("404"), Unavailable("rate limited")] + [NotFound("404")] * 3) assert _security_policy(http, "acme/repo", {}) is None ``` ## scripts/uv.lock ``` version = 1 revision = 3 requires-python = ">=3.11" [options] exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P1W" [[package]] name = "supply-chain-risk-auditor-scripts" version = "0.1.0" source = { virtual = "." } ```