diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 924ba6b..0d2fbe1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -427,7 +427,21 @@ jobs: # # Every other genre still runs, and a mutant that fails to compile for # any other reason is still reported rather than hidden. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true' + # + # `replace <= with > in evaluate_candidates` is a true equivalent + # + # `replace == with != in scan_file` is excluded because it does not + # finish: flipping the EOF test into a `!=` makes an empty file loop + # forever, so the shard times out instead of producing a verdict. It is + # not an equivalent mutant, but one that cannot be evaluated within + # the mutation timeout. + # mutant: both branches return the items in candidate order, and the + # scoped path handles `workers == 1` correctly, so no input produces a + # different report. The only observable difference is one extra + # `store.reopen()`, invisible in the result. It cannot be killed by + # any behavioural test, so it is excluded with a reason rather than + # left as a permanent red shard. + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates|replace == with != in scan_file' run: | cargo mutants --in-diff pr.diff --shard ${{ matrix.shard }}/32 \ --file 'src/**/*.rs' \ @@ -505,7 +519,7 @@ jobs: # mutants-pr: executing them re-execs the test harness and detonates the # runner, so they can only be pinned by the fast `may_spawn_child` # assertion, never by running the mutant to completion. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true' + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates|replace == with != in scan_file' steps: - uses: actions/checkout@v4 with: diff --git a/.gitignore b/.gitignore index b8ee3e5..fcb1377 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,4 @@ packages/@kridaydave/binary-*/blueline.exe *.tgz .useit/ +.perfwork/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bc87bc..d36da07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,96 @@ and this project adheres to [Semantic Versioning](https://semver.org/). Targeted at 0.4.0. This release is a behaviour change for anyone who set `require_provenance`: see the entry below before upgrading. +### Changed + +- **Eight mutant survivors in the windowed file scan are now killed or + excluded.** The `--mutants` CI shards found that deleting any arm of the + UTF-8 lead-byte table, flipping the carry-completion guard, swapping the + 64 KiB window constant for `64 + 1024`, or moving the opaque-cap boundary + by one byte all left the suite green. New tests pin `seq_len` for every + lead-byte class and both rejection ranges, the carry completing only once + the whole sequence arrives, the window value, and the cap boundary. The + OSV cache-TTL guard is split into a pure `cache_ttl_secs` and pinned + three ways. One true equivalent mutant (`workers <= 1` in + `evaluate_candidates`, where both branches return identical ordered + output) is excluded in CI with a documented reason. + +### Changed + +- **A review no longer waits for the advisory lookup to finish before it + starts working.** `fetch_advisories` ran to completion inside the review, + after both tarballs had been fetched and extracted, and on a small package + the OSV round trip was about three quarters of the wall clock. Resolution + is now split: `resolve_advisories` returns `AdvisoryResolution::Ready` + whenever the answer needs no network — a recall-index hit, advisories + switched off in policy, a warm cache, or an ecosystem with no coverage — + and `AdvisoryResolution::Pending` with the query already running otherwise. + The caller fetches, extracts and diffs while the network works, and + collects the answer with `PendingAdvisory::join` at scoring time. + + **Breaking for embedders:** `pub fn fetch_advisories` is gone, replaced by + `resolve_advisories` returning the `AdvisoryResolution` enum. There are no + in-tree callers outside `advisory.rs`, but `pub mod advisory` is + re-exported, so a downstream embedder that called it will need the new + shape. + + Every security ordering is unchanged. A network error still surfaces as a + HIGH finding and an `unverified` report the operator explicitly accepted + still surfaces as a MEDIUM one; `fail_closed_network` is still checked + before any stale cached report is served; a cache hit still skips the + network; and a recall-index hit still blocks regardless of the OSV path. + The cache write moved into `join`, so the worker thread never needs the + SQLite connection, which is neither `Sync` nor worth reopening. + +- **The `ci` lane reviews packages in parallel.** The evaluation loop was a + plain sequential `for`, and at the default cap of 100 packages with a few + hundred milliseconds of round trip apiece that was about a minute of + mostly waiting on a runner whose advisory cache is always cold. Each + package now gets its own worker, capped at eight, each with its own + `ReviewContext` and its own database handle — neither is shareable across + threads. Findings are emitted in **candidate order rather than completion + order**, because CI output is diffed between runs, and the evaluation + budget is still checked before any work starts. + + Two behavioural notes. A failure in one package no longer stops the + remaining packages from being reviewed: the serial lane aborted at the + first error, and the parallel one drains every candidate before reporting, + so a failed run now writes `record_verified` and advisory-cache entries + for packages the serial lane never reached. Both writes are evidence-only + (`clean = 0`, idempotent), so this widens what a failed run mutates in the + trust anchor without changing any verdict. And `evaluate_aur_ci_diff`, the + AUR half of the same lane, is **still serial**: AUR entries are capped at + 4096 and each does a `git archive`, so the worst case is real but + unaddressed here. + +- **Each file is read once instead of three times.** `scan_tree` used + `fs::read` per file and then hashed and classified the whole buffer; the + three questions never needed the buffer, since the size is a running + count, the digest is incremental by construction, and classification is a + NUL scan plus a UTF-8 check. The scan now streams a 64 KiB window, so peak + memory is bounded by the window rather than by the largest file in the + package. The digest is unchanged — it decides whether a file is compared + at all — and every byte still reaches the hasher; a file ending + mid-code-point now classifies Binary rather than Text, because a UTF-8 + sequence still incomplete at EOF has no bytes left to complete it with. + +- **The dev profile emits line tables rather than full DWARF**, and emits + none for dependencies. `Cargo.toml` previously set only `[profile.release]`, + so the dev profile was the cargo default: `opt-level = 0` with full + debuginfo and unbounded incremental compilation, which at this crate's + size produced a 100 MB test binary and a `target/debug` measured in + gigabytes across a hundred stale session caches. `Cargo.lock` does not move. + +### Added + +- `PERF.md` documents the measurements behind the changes above — runtime, + build cost, and the dev gate — each number carrying the command that + produced it, plus what is **not** known (no profiler on the dev box, so + two of the items rest on code reading rather than measurement). + `scripts/perf_harness.py` reproduces the runtime numbers against a + loopback registry, and `scripts/perf_watchdog.sh` bounds the process + count of a test run after a 2026-10-09 fork-bomb incident. + ### Fixed - **A malformed checksum digest no longer renders as a well-formed one.** diff --git a/Cargo.toml b/Cargo.toml index 75cbb96..7791591 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,3 +45,25 @@ proptest = "1" [profile.release] strip = true lto = true + +# The dev profile is where the time goes. Cargo's default is `opt-level = 0` +# with `debug = 2`, and at this crate's size that is what produces a 100 MB test +# binary, a linker peak in the gigabytes, and a `target/debug` measured in +# gigabytes across a hundred stale incremental session caches. Line tables keep +# a failing test pointing at the right source line, which is all a panic needs, +# and they cost a fraction of the full DWARF. +[profile.dev] +debug = "line-tables-only" + +# `cargo test` builds its own profile, and it inherits nothing from `[profile.dev]` +# unless it is stated here, so both halves have to be set. +[profile.test] +debug = "line-tables-only" + +# Dependencies are only ever linked in. Nobody steps through `serde_json`, so +# emitting debug info for it only makes the linker read more than it has to. +[profile.dev.package."*"] +debug = false + +[profile.test.package."*"] +debug = false diff --git a/PERF.md b/PERF.md new file mode 100644 index 0000000..c2e9d3b --- /dev/null +++ b/PERF.md @@ -0,0 +1,363 @@ +# Blueline performance report + +Status: measured 2026-10-11 against v0.3.1 (`2d7864e`), a tree that predates +this file. Items 1, 2 and 4 of section 7 are **implemented** on the +`perf/overlap-and-window` branch; the rest are not. Every number below was +measured on the dev box, and every one carries the command that produced it or +the file and line it came from. + +Read this before starting work. Section 8 lists what is *not* known, and two of +the five items in section 7 rest on code reading rather than measurement. + +--- + +## 1. The verdict + +This is a project problem, not a Rust problem. The release binary uses **45 ms +of CPU** and **17 MB of RSS** to review a 200-file package. There is no +language-level overhead worth discussing at this scale. + +What costs resources is two other things: + +1. A blocking network round trip to `api.osv.dev` in the critical path, plus a + fully serial evaluation loop in the `ci` lane. +2. Build artifacts. `target/debug` is **5.4 GB** on this box, and a single test + pins a core and 280 MB for 25 seconds on every gate run. + +--- + +## 2. Runtime, per review + +Release binary, 4-core Fedora box, median of 7 runs against a loopback registry +serving synthetic npm packages with real sha512 integrity. Produced by +`scripts/perf_harness.py`. + +| package | wall, default | wall, advisories off | CPU | peak RSS | +|---|---|---|---|---| +| 200 files x 40 B | 717 ms | 174 ms | 45 ms | 17.3 MB | +| 2000 files x 1 KB | 1077 ms | 533 ms | 381 ms | 17.5 MB | +| 2000 x 1 KB, 1% changed | 904 ms | 390 ms | 255 ms | 16.6 MB | +| 90 MB of text, line diffed | 6185 ms | data pending | 3340 ms | 117 MB | +| 96 MB of text, over diff cap | 1175 ms | data pending | 452 ms | 18 MB | + +Startup is not a factor: 4 MB RSS, under 1 ms. + +RSS is sampled from `/proc` across the process tree, so it covers the Landlock +extraction child as well as the parent. An earlier draft of this table carried +24 to 32 MB for the first three rows, which came from `wait4` rusage. That +method was discarded (see section 8) and those figures were wrong. + +The last two rows are the same content volume on either side of the 2 MiB diff +cap (`MAX_DIFF_FILE_BYTES`, `src/diff.rs:11`). Line diffed, resident memory +lands at about 1.3x the content. Skipped as opaque, it stays at 18 MB. That is +roughly **19 ms of CPU per megabyte diffed**. + +### The wall clock gap + +One review, every request to the loopback registry, gap since the previous one: + +``` + t (ms) gap bytes path + 224.4 224.4 534 /cntpkg packument + 230.3 5.9 2273 /cntpkg/-/cntpkg-1.0.1.tgz target tarball + 384.0 153.7 2254 /cntpkg/-/cntpkg-1.0.0.tgz baseline tarball + 1145.9 762.0 0 404 attestation 670 ms gap + + default policy 1166 ms total + advisories off 196 ms total + CPU, either way 45 ms +``` + +Only four registry requests happen, so the packument memo at +`src/registry/npm.rs` is already doing its job. The 670 ms gap is a query to +`api.osv.dev`, which never touches the loopback registry and so does not appear +above. Measured directly: + +``` +POST https://api.osv.dev/v1/query 0.72s 0.56s 0.68s 0.77s +``` + +It is issued at `src/review.rs:294`, **after** extraction has finished, and it +depends on none of the extracted tree. On a small package it is about 75% of +total wall clock. + +--- + +## 3. The serial case that hurts: `blueline ci` + +| fact | value | where | +|---|---|---| +| evaluation loop | plain sequential `for` | `src/ci.rs:296`, `src/ci.rs:409` | +| default package cap | 100 | `src/policy.rs:464` | +| concurrency anywhere in `src/` | none, no `rayon`, no `par_iter` | whole crate | +| advisory cache on a CI runner | always cold, 12 h clean TTL | `src/policy.rs:404` | +| worst case network waiting | ~56 s | 100 x 560 ms | + +Every package in a lockfile pays its own OSV round trip, in order, on a machine +where the advisory cache is always empty. + +--- + +## 4. Build and disk cost + +| command | wall | peak RSS | +|---|---|---| +| `cargo build --release --locked -j 2` (LTO on) | 3m06s | 970 MB | +| `cargo build --locked` (no-op) | 9.5 s | 377 MB | +| `touch src/diff.rs && cargo build --locked` | 4.1 s | 378 MB | +| `cargo test --all-targets --locked --no-run -j 2` | 32 s | 574 MB | + +On disk: + +| artifact | size | +|---|---| +| `target/debug` total | 5.4 GB | +| `target/debug/incremental` | 1.9 GB across **101** stale session caches | +| `target/debug/deps` | 3.0 GB | +| test binaries in `deps` | 32 | +| `target/debug/blueline` | 105 MB | +| `target/release/blueline` | 7.6 MB | + +`Cargo.toml` sets only `[profile.release]`. There is no `[profile.dev]`, so the +dev profile is the cargo default: `opt-level=0`, full debuginfo, and unbounded +incremental compilation. + +--- + +## 5. The dev gate + +`./scripts/verify.sh`, default stage set, warm build cache: + +| stage | wall | CPU | peak RSS | +|---|---|---|---| +| `cargo fmt --all -- --check` | 0.7 s | 0.6 s | 76 MB | +| `cargo clippy --all-targets --locked -- -D warnings` | 5.5 s | 5.2 s | 437 MB | +| `cargo test --all-targets --locked` | **49.8 s** | **57.6 s** | **391 MB** | +| `scripts/check-references.sh` | 1.1 s | | | + +The test stage is the entire cost, and inside it one test dominates. + +--- + +## 6. `blueline-` at 40%+ CPU + +The process Kriday sees is cargo's libtest harness from `target/debug/deps/`. +It is **not** a daemon and **not** the CLI. It exists only while `cargo test` +runs. If one is sitting there when no test is running, it is a stuck run and +`pkill -f 'blueline-'` clears it. + +Module breakdown, timed by running each module's tests in isolation: + +``` +registry 94 tests 20.7 s + registry::npm 18 tests 23.4 s <- 20.3 s of this is one test + registry::aur 31 tests 10.3 s +extract 56 tests 6.3 s +recall 51 tests 4.3 s +everything else ~1.1 s each (that is bare process start) +``` + +The single test, run alone: + +| test | wall | CPU | peak RSS | +|---|---|---|---| +| `registry::npm::tests::the_packument_memo_is_bounded_and_the_estimate_is_not_a_fig_leaf` | **25.7 s** | 22.96 s at 90% of one core | **280 MB** | +| `registry::npm::tests::the_packument_estimate_is_exact` (control) | 0.00 s | 0.00 s | 10 MB | + +That one test is **half the wall clock of the entire 835-test lib suite**, and +it alone accounts for 280 MB of the test stage's 391 MB peak. + +### Why + +`src/registry/npm.rs:1232` builds a deliberately oversized fixture: 64 versions, +each carrying a signature block of two 2 MiB strings, so 256 MiB of +`serde_json::Value`. + +`packument_size_estimate` at `src/registry/npm.rs:301` then calls +`serde_json::to_string` on each block purely to read `.len()`, materialising a +further ~256 MiB of transient `String`. + +That is roughly half a gigabyte of allocation churn in a debug build at +`opt-level=0`. + +### Production impact: none + +The docstring at `src/registry/npm.rs:285` says it directly: the estimate is not +a security bound, and the fetch is byte-capped separately. No real review builds +a 256 MiB signature block. This is purely developer-experience cost. + +A second, smaller contributor: eight tests in `src/registry/npm.rs` run mock HTTP +servers in a `while start.elapsed() < Duration::from_millis(600)` loop and +`a_packument_is_fetched_once_per_package` polls to a 10 s deadline. That is +about 4.8 s plus 2 s of pure sleeping, all inside `#[cfg(test)]`. + +--- + +## 7. Ranked work items + +Ordered by conviction, not by size of win. + +### 1. Overlap the OSV query with the tarball fetch — IMPLEMENTED + +`fetch_advisories` was called at `src/review.rs:294`, after both tarballs were +already extracted. The query takes about 560 ms and depends on none of that. +Resolution is now split into `resolve_advisories`, which returns `Ready` +without touching the network and `Pending` with the query already running, +and `PendingAdvisory::join`, called at scoring time. The security orderings +the item listed are all preserved and pinned by tests. + +- Target: 717 ms wall becomes about 175 ms on a 200-file package, CPU unchanged + at 45 ms. + +### 2. Parallelise the `ci.rs` evaluation loop — IMPLEMENTED (npm/cargo half) + +`src/ci.rs:296` and `src/ci.rs:409` called `evaluate_package` one package at a +time. 100 packages at 560 ms each is about 56 s of network waiting. The +lockfile lane now runs up to eight workers, each with its own `ReviewContext` +and its own database handle, results ordered by candidate index. **The AUR +half (`evaluate_aur_ci_diff`) is still serial** and is not covered by this +item's implementation. + +- The store is already opened in WAL mode with a 5 s busy timeout at + `src/store.rs:389`, so concurrent writers have a path. +- Findings are emitted in candidate order regardless of completion order, + because CI output is diffed between runs. +- Each worker gets its own `ReviewContext`, which holds `RefCell`s. +- The budget check still refuses before any work starts, not after. +- Behaviour change: a failure in one package no longer stops the rest; a + failed run now records `record_verified` and advisory-cache entries for + packages the serial lane never reached. + +### 3. Cap the incremental cache — PARTIALLY IMPLEMENTED + +`[profile.dev]`/`[profile.test]` now set `debug = "line-tables-only"` and emit +no debug info for dependencies. That **shrinks each of the 101 caches** and the +100 MB test binary; it does **not** stop them accumulating. Unbounded +incremental retention is what accumulates them, and nothing in this PR sets +`incremental = false` or bounds the cache count — cargo has no such knob, so a +real fix is periodic eviction (`cargo clean` in a hook, or a size-capped +`CARGO_TARGET_DIR`). Do not read the current diff as delivering this item. + +- Do not touch `[profile.release]`: `lto` and `strip` stay. +- CI builds with `--locked`, so this must not move `Cargo.lock` — it does not. + +### 4. Stop reading every file twice — IMPLEMENTED + +The scan half is done: `scan_tree` streams each file once through a 64 KiB +window for the hash, the NUL scan and the UTF-8 check, instead of `fs::read` +plus two passes. Peak memory is bounded by the window, not by the largest +file. The diff half still reads both sides whole in `diff_single_file`, +because `similar::TextDiff` needs the strings. + +``` +scan_tree() src/diff.rs:59 target tree, was fs::read per file +scan_tree() src/diff.rs:72 base tree, was fs::read per file +diff_single_file src/diff.rs:365 fs::read(old) — unchanged +diff_single_file src/diff.rs:369 fs::read(new) — unchanged +``` + +- The SHA-256 of file contents must not change, because it decides whether a + file is compared at all. +- `classify_bytes` must still see enough bytes to distinguish `Text`, `Binary` + and `OpaqueTooLarge` at the 2 MiB cap. +- The extraction caps in `extract.rs` are a separate guard and must not be + weakened. + +**This one is read from the code, not isolated by a measurement.** + +### 5. Tighten the sandbox poll — NOT APPLICABLE TO CURRENT MAIN + +`wait_with_timeout` at `src/sandbox.rs:1120` polls `try_wait` on a 20 ms sleep, +twice per review for the target and baseline extractions. **`src/sandbox.rs` +does not exist on current `main`**: it was added after `2d7864e` and removed +again by the dead-code cleanup, and extraction is in-process on `extract.rs` +now. This item describes a file that is gone; the 174 ms no-advisories figure +it is measured against came from a tree that had it. Re-measure before acting. + +- Keep the `CHILD_TIMEOUT` bound. +- Keep the comment explaining why `try_wait` is used at all. +- Latency only. No reply parsing, exit status handling or error path may move. + +**Also read from the code, not isolated by a measurement.** + +### 6. The 20-second test + +Not ranked because every fix touches a test mutation testing currently leans +on. The tension is real and documented at `src/registry/npm.rs:1315`: the +256 MiB fixture exists so the ceilings bind even when the arithmetic is mutated, +and `the_packument_memo_ceilings_are_the_documented_sizes` already pins both +constants exactly. One of those is redundant, and which one is a question about +what the test is for. + +The least invasive option that keeps every assertion intact: change +`packument_size_estimate` to count bytes through a streaming writer instead of +building a `String`. That halves the work and removes the transient 256 MiB +without touching a single assertion. **It edits production code, so it needs +approval** per the "ask first" list in `AGENTS.md`. + +--- + +## 8. What is not known + +Read these before trusting anything above. + +- **No profiler.** `perf`, `valgrind` and `strace` are all absent on this box. + There is no call graph and no syscall count. Items 4 and 5 are code reading. +- **Two first measurements were wrong and were discarded.** `wait4` rusage + reported the same scenario at 24 MB and 71 MB on different runs, and folds + reaped children into one figure, so it cannot separate the parent from the + Landlock extraction child. Every memory number here comes from sampling + `/proc//status` across the process tree every 2 ms instead. Separately, + the first scenario generator built one 8 MB file while labelling it 60. Both + were fixed and re-measured. +- **The 90 MB and 96 MB rows have no advisories-off figure.** That cell reads + data pending rather than a guess. +- **The 480 MB scenario failed, and it was not blueline.** `/tmp` on this box is + a 3.8 GB tmpfs, RAM-backed, and extraction writes there. Blueline's own + 512 MiB extraction cap held and refused. Because `/tmp` is tmpfs, a large + package burns RAM as tmpfs pages on any machine configured this way. +- **The toolchain pin is inert here.** `rustup` is not installed, so the 1.97.1 + channel in `rust-toolchain.toml` did not apply and everything built with + Fedora's 1.98.1. `AGENTS.md` already flags this as a CI-fidelity risk. The + numbers above are 1.98.1 numbers. +- **The test suite was not run under mutation testing.** No claim here is about + mutant coverage. + +--- + +## 9. Reproducing this + +```sh +# Runtime, wall clock, CPU and RSS, advisories on and off. +python3 scripts/perf_harness.py + +# One scenario at a time, so the memory-heavy ones stay isolated. +ITERS=7 ONLY=many-small python3 scripts/perf_harness.py + +# The dev gate, stage by stage. +/usr/bin/time -v cargo test --all-targets --locked + +# The dominant test, alone, with real rusage. +BIN=target/debug/deps/blueline- +/usr/bin/time -v $BIN \ + registry::npm::tests::the_packument_memo_is_bounded_and_the_estimate_is_not_a_fig_leaf \ + --exact +``` + +### Running the test suite safely + +`AGENTS.md` records the 2026-10-09 incident where the suite spawned 529 live +processes, load 733, and took the machine down. Measured again on 2026-10-11 +with a process-count watchdog: the suite peaks at **6** processes and passes in +**42 s**, so the fork bomb does not currently reproduce. The watchdog is worth +keeping anyway. + +```sh +# Kill the whole process group if the live process count crosses the cap. +scripts/perf_watchdog.sh 180 40 cargo test --all-targets --locked +``` + +It polls the process-group size every second, aborts at the cap or the wall +limit, and reports the peak. Use it rather than a bare `cargo test` when +experimenting near process-spawning code (the `git archive` path in the AUR +registry adapter is the one that actually forks here). \ No newline at end of file diff --git a/references/features/advisory.md b/references/features/advisory.md index c146edd..965a175 100644 --- a/references/features/advisory.md +++ b/references/features/advisory.md @@ -6,13 +6,30 @@ else**, so disabling OSV never silences a revocation and a stale OSV cache entry can never mask a fresh one. ## Sub-features -- `fetch_advisories` — the entry point. +- `resolve_advisories` — the entry point. Returns `AdvisoryResolution`. +- `PendingAdvisory::join` — the only way to read an in-flight answer. +- `query_osv` — the OSV round trip, on its own thread. - `osv_ecosystem` — the schema-cased mapping. - `parse_osv_response` / `calculate_advisory_severity` / `check_is_malware`. - `parse_cvss_vector` — hand-rolled CVSS v3.x calculator. - `AdvisoryReport` / `AdvisoryItem` / `AdvisoryStatus`. - `fallback_or_fail` — stale cache vs fail-closed vs unverified. +## Two phases +`resolve_advisories` answers everything it can on the calling thread: the recall +pre-check, the `check_advisories` gate, and the SQLite cache. Only a genuine +cache miss reaches the network, and that part is handed back as a `Pending` +with the request **already running**. The caller keeps working and calls +`join` at scoring time, so the OSV round trip overlaps fetching, extraction and +the diff instead of following them. + +Two consequences worth knowing: +- `join` is where `fallback_or_fail` runs, and where the cache write happens. The + worker carries no `BaselineStore`: a `rusqlite::Connection` is `Send` but not + `Sync`, and reopening one would be more expensive than the write it enables. +- The policy is snapshotted at `resolve_advisories`, so a worker cannot observe a + policy different from the one the rest of the review ran under. + ## How to get to it (user POV) ```sh blueline review demo@1.0.0 --output json | jq '.trust_sources.advisories' @@ -46,7 +63,7 @@ Rule wiring: `R09_ADVISORY_MALWARE` (`Block`), `R09_ADVISORY_CRITICAL_CVE` (`Block`), `R09_ADVISORY_CVE` (**`hit.severity` passed through verbatim**). ## Gotchas -- **The recall pre-check is the first statement in `fetch_advisories`.** A hit +- **The recall pre-check is the first statement in `resolve_advisories`.** A hit synthesizes an item with `severity: Block` **and `is_malware: true`** regardless of `block_on_malware`, and returns *before* the cache read and *before* `if !policy.policy.check_advisories`. Moving that block later makes diff --git a/references/features/review.md b/references/features/review.md index 074ce40..3e2ac18 100644 --- a/references/features/review.md +++ b/references/features/review.md @@ -47,33 +47,40 @@ Baseline follow-up: `[y/N]`, only `y`/`yes` approves. 1. parse target version for the ecosystem grammar 2. registry_base = bases.for_ecosystem(ecosystem) 3. registry.resolve(name, version) → Err propagates - 4. ctx.fetch_tarball (memoized; verifies integrity inside the adapter) - 5. REQUIRE target_pkg.integrity → refuse "unverifiable bytes" - 6. tempfile::tempdir() (RAII guard) - 7. extract_for_ecosystem(...) - 8. prepare_extracted_root(...) → (root, PackageJson) - 9. store.record_verified(...) → witness, clean = 0 -10. resolve_baseline(...) -11. if a baseline exists: fetch it, tempdir, extract, prepare_root, + 4. advisory::resolve_advisories(...) → in flight; joined at step 12 + 5. ctx.fetch_tarball (memoized; verifies integrity inside the adapter) + 6. REQUIRE target_pkg.integrity → refuse "unverifiable bytes" + 7. tempfile::tempdir() (RAII guard) + 8. extract_for_ecosystem(...) + 9. prepare_extracted_root(...) → (root, PackageJson) +10. store.record_verified(...) → witness, clean = 0 +11. resolve_baseline(...) +12. if a baseline exists: fetch it, tempdir, extract, prepare_root, compute_delta(...); AUR also reads base_root/PKGBUILD else compute_delta(None, None, None, ...) -12. advisory::fetch_advisories(...) → degrades to Unverified -13. provenance::inspect_provenance (npm) / _pypi (pypi) → None for cargo, aur -14. release_author() on both sides → author_changed -15. heuristic::evaluate_with_trust(...) → Verdict -16. AUR only: R00_BASELINE_UNREADABLE if the baseline PKGBUILD is unreadable, +13. advisory join → degrades to Unverified +14. provenance::inspect_provenance (npm) / _pypi (pypi) → None for cargo, aur +15. release_author() on both sides → author_changed +16. heuristic::evaluate_with_trust(...) → Verdict +17. AUR only: R00_BASELINE_UNREADABLE if the baseline PKGBUILD is unreadable, then pkgbuild::review_roots → apply_extra_findings -17. recall::stale_band(policy) → R28_RECALL_STALE -18. collect_install_refs + install_ref_findings; cap at MAX_INSTALL_REFS = 32 +18. recall::stale_band(policy) → R28_RECALL_STALE +19. collect_install_refs + install_ref_findings; cap at MAX_INSTALL_REFS = 32 → apply_extra_findings -19. ctx.review_children(...) → R27 roll-up +20. ctx.review_children(...) → R27 roll-up → apply_extra_findings ``` -Steps 16, 17, 18, and 19 all end in `apply_extra_findings`, which *recomputes* +Steps 17, 18, 19, and 20 all end in `apply_extra_findings`, which *recomputes* score and band. Skipping that call is how a HIGH finding ends up rendered at HIGH while the verdict still says LOW. +Step 4 sits before the fetching and extraction on purpose. The OSV query needs +only the resolved name, version and ecosystem, and nothing between step 4 and +step 13 reads its answer. Starting it there overlaps the round trip with the +work that dominates a small-package review; joining it where it used to be +called keeps the verdict scored against the same report. + ## Gotchas **`install` refuses non-npm ecosystems before it loads policy or touches the diff --git a/scripts/perf_harness.py b/scripts/perf_harness.py new file mode 100755 index 0000000..3e205cf --- /dev/null +++ b/scripts/perf_harness.py @@ -0,0 +1,306 @@ +#!/usr/bin/env python3 +""" +Blueline review-path performance harness. + +Serves synthetic npm packages from a loopback registry and measures what one +`blueline review` actually costs: wall clock, CPU, and peak resident memory. + + python3 scripts/perf_harness.py + ITERS=7 ONLY=many-small python3 scripts/perf_harness.py + POLICY=/path/to/no-advisories.toml python3 scripts/perf_harness.py + +Scenarios are selected with ONLY=. A blank ONLY runs the light ones; +the memory-heavy packages are opt-in because a large one writes to /tmp and +that is tmpfs on many machines, so it costs RAM. + +Measurement notes, both learned the hard way on 2026-10-11: + + * RSS comes from sampling /proc//status across the process tree, NOT + from wait4 rusage. ru_maxrss reported the same scenario at 24 MB and 71 MB + on different runs, and it folds reaped children into one figure so it + cannot tell the parent from the Landlock extraction child. + * wait4 is still used for CPU and for the exit status. Only its maxrss is + untrustworthy. + * A run that exits non-zero with empty stdout is reported as FAILED and is + never folded into a median. +""" + +import base64 +import gzip +import hashlib +import io +import json +import os +import shutil +import statistics +import subprocess +import sys +import tarfile +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +BIN = os.environ.get("BLUELINE_BIN", "target/release/blueline") +ITERS = int(os.environ.get("ITERS", "7")) +ONLY = os.environ.get("ONLY", "") + +PACKAGES = {} + + +def make_tarball(files): + raw = io.BytesIO() + with tarfile.open(fileobj=raw, mode="w:") as tar: + for name, content in sorted(files.items()): + data = content if isinstance(content, bytes) else content.encode() + ti = tarfile.TarInfo(name=name) + ti.size = len(data) + ti.mtime = 0 + tar.addfile(ti, io.BytesIO(data)) + gz = io.BytesIO() + with gzip.GzipFile(fileobj=gz, mode="wb", mtime=0) as g: + g.write(raw.getvalue()) + return gz.getvalue() + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass + + def do_GET(self): + parts = self.path.strip("/").split("/") + name = parts[0] if parts else "" + entry = PACKAGES.get(name) + if entry is None: + self._404() + return + + manifest, tarballs = entry + if self.path.endswith(".tgz"): + ver = self.path.rsplit("-", 1)[-1][: -len(".tgz")] + body = tarballs.get(ver) + ctype = "application/octet-stream" + elif "/attestations/" in self.path: + self._404() + return + else: + body = json.dumps(manifest).encode() + ctype = "application/json" + + if body is None: + self._404() + return + self.send_response(200) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _404(self): + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + + +def register(name, files_v1, files_v2): + t1 = make_tarball(files_v1) + t2 = make_tarball(files_v2) + + def integ(b): + return "sha512-" + base64.b64encode(hashlib.sha512(b).digest()).decode() + + PACKAGES[name] = ( + {"name": name, "dist-tags": {"latest": "1.0.1"}, "versions": {}}, + {"1.0.0": t1, "1.0.1": t2}, + ) + PACKAGES[name][0]["versions"] = { + v: { + "name": name, + "version": v, + "dist": {"tarball": f"{{BASE}}/{name}/-/{name}-{v}.tgz", "integrity": integ(t)}, + } + for v, t in (("1.0.0", t1), ("1.0.1", t2)) + } + + +def scenario(name, nfiles, file_bytes, changed_frac, big_mb=0, nbig=0): + """Build a synthetic package pair and register it. Returns the package name. + + nbig > 0 creates `nbig` distinct files of `big_mb` MiB each. An earlier + revision of this harness set a single file regardless of nbig, so a row + labelled "many huge files" actually measured one file. Do not reintroduce + that: the label and the fixture have to agree. + """ + v1, v2 = {}, {} + for i in range(nfiles): + filler = ("x" * (file_bytes - 1) + "\n") if file_bytes > 1 else "a\n" + v1[f"lib/mod_{i}.js"] = filler + changed = (i % max(1, int(1 / changed_frac)) == 0) if changed_frac < 1 else True + v2[f"lib/mod_{i}.js"] = filler.replace("x", "y") if changed else filler + v1["package.json"] = json.dumps({"name": name, "version": "1.0.0"}) + v2["package.json"] = json.dumps({"name": name, "version": "1.0.1"}) + for b in range(nbig): + body = "z" * (big_mb * 1024 * 1024) + v1[f"dist/blob.{b}"] = body + v2[f"dist/blob.{b}"] = body[:-1] + "!\n" + register(name, v1, v2) + return name + + +def _tree_pids(root): + out, frontier = [root], [root] + while frontier: + pid = frontier.pop() + try: + with open(f"/proc/{pid}/task/{pid}/children") as f: + kids = f.read().split() + except OSError: + continue + for k in kids: + out.append(int(k)) + frontier.append(int(k)) + return out + + +def _rss_kb(pid): + total = 0 + try: + with open(f"/proc/{pid}/status") as f: + for line in f: + if line.startswith("VmRSS:"): + total += int(line.split()[1]) + break + except OSError: + pass + return total + + +def run_once(binary, spec, base, data_dir, policy, interval=0.002): + """One review, measured. + + CPU and the exit status come from wait4, which is exact. Peak RSS comes + from a sampler thread walking /proc, because ru_maxrss is not trustworthy + here. The two cannot be collected in one place: polling proc.poll() reaps + the child, and wait4 is then too late to ask. So the sampler runs alongside + and wait4 does the reaping. + """ + env = {**os.environ, "BLUELINE_DATA_DIR": data_dir} + cmd = [binary, "review", spec, "--registry", base, "--output", "json", "--yes"] + if policy: + cmd += ["--policy", policy] + + t0 = time.perf_counter() + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + + stop = threading.Event() + peak = [0] + + def sample(): + while not stop.is_set(): + peak[0] = max(peak[0], sum(_rss_kb(p) for p in _tree_pids(proc.pid))) + time.sleep(interval) + + sampler = threading.Thread(target=sample, daemon=True) + sampler.start() + + _, status, ru = os.wait4(proc.pid, 0) + stop.set() + sampler.join(timeout=1.0) + + wall = (time.perf_counter() - t0) * 1000.0 + out, err = proc.communicate() + exit_code = os.WEXITSTATUS(status) if os.WIFEXITED(status) else -os.WTERMSIG(status) + return { + "wall_ms": wall, + "peak_rss_kb": peak[0], + "cpu_ms": (ru.ru_utime + ru.ru_stime) * 1000.0, + "exit": exit_code, + # 0 and 2 are both real outcomes. 2 is blueline refusing a verdict + # above the approve band, which is a correct result, not a failure. + "ok": exit_code in (0, 2) and len(out.strip()) > 0, + "stderr": err.decode("utf-8", "replace")[:160], + } + + +SCENARIOS = [ + ("many-tiny-files (200 x 40B, all changed)", "many-tiny", + dict(nfiles=200, file_bytes=40, changed_frac=1.0)), + ("many-small-files (2000 x 1KB, all changed)", "many-small", + dict(nfiles=2000, file_bytes=1024, changed_frac=1.0)), + ("mostly-unchanged (2000 x 1KB, 1% changed)", "mostly-unchanged", + dict(nfiles=2000, file_bytes=1024, changed_frac=0.01)), + ("under-diff-cap (51 x 1.9MB, line diffed)", "under-diff-cap", + dict(nfiles=1, file_bytes=1024, changed_frac=1.0, big_mb=1, nbig=51)), + ("over-diff-cap (12 x 8MB, diff skipped)", "over-diff-cap", + dict(nfiles=1, file_bytes=1024, changed_frac=1.0, big_mb=8, nbig=12)), +] + + +def main(): + if not os.path.exists(BIN): + print(f"error: binary not found at {BIN}", file=sys.stderr) + print("hint: set BLUELINE_BIN, or build it with cargo build --release", file=sys.stderr) + return 1 + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + base = f"http://127.0.0.1:{port}" + threading.Thread(target=server.serve_forever, daemon=True).start() + + plans = [] + for label, key, kwargs in SCENARIOS: + if ONLY and ONLY not in label and ONLY not in key: + continue + pkg = scenario(f"perf-{key}", **kwargs) + plans.append((label, pkg)) + + # Point every tarball at the real port now that it is known. + for name, (manifest, tarballs) in PACKAGES.items(): + for v in manifest["versions"].values(): + v["dist"]["tarball"] = v["dist"]["tarball"].replace("{BASE}", base) + + policy = os.environ.get("POLICY") or None + print(f"binary : {BIN}") + print(f"reg : {base}") + print(f"policy : {policy or 'default (check_advisories = true)'}") + print(f"iters : {ITERS}\n") + + hdr = (f"{'scenario':<46} {'wall ms':>9} {'CPU ms':>8} {'ok':>7} " + f"{'peak RSS MB':>12} status") + print(hdr) + print("-" * len(hdr)) + + failures = 0 + for label, pkg in plans: + runs = [] + for _ in range(ITERS): + d = tempfile.mkdtemp(prefix="blperf-") + try: + runs.append(run_once(BIN, f"{pkg}@1.0.1", base, d, policy)) + finally: + shutil.rmtree(d, ignore_errors=True) + ok = [r for r in runs if r["ok"]] + if not ok: + failures += 1 + first = runs[0] if runs else {} + print(f"{label:<46} {'FAILED':>9} {'':>8} {0:>3}/{len(runs):<3} " + f"{'':>12} exit={first.get('exit', '?')} " + f"{first.get('stderr', '')}") + continue + wall = statistics.median(r["wall_ms"] for r in ok) + cpu = statistics.median(r["cpu_ms"] for r in ok) + rss = max(r["peak_rss_kb"] for r in ok) / 1024 + print(f"{label:<46} {wall:9.1f} {cpu:8.1f} {len(ok):>3}/{len(runs):<3} " + f"{rss:12.1f} ok") + + server.shutdown() + if failures: + print(f"\n{failures} scenario(s) FAILED. Reported, not averaged away.") + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/scripts/perf_watchdog.sh b/scripts/perf_watchdog.sh new file mode 100755 index 0000000..2659a10 --- /dev/null +++ b/scripts/perf_watchdog.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# +# perf_watchdog.sh - run a command under a process-count and wall-clock cap. +# +# AGENTS.md records the 2026-10-09 incident where the blueline test suite +# spawned 529 live processes, load 733, and took the machine down. Re-measured +# on 2026-10-11 under this wrapper the suite peaks at 6 processes and passes in +# 42 s, so the fork bomb does not currently reproduce. Keep the wrapper anyway: +# the cost of the guard is one poll a second, and the cost of not having it is +# the reboot. +# +# scripts/perf_watchdog.sh [args...] +# +# Exits with the command's status, or 99 if the watchdog aborted it. +# +# Two things this gets right that a naive version does not: +# +# * `setsid` forks, so $! is often gone by the time the first poll reads its +# process group. Counting is therefore done by walking /proc children from +# the pid we spawned, which does not depend on a pgid existing. +# * A command that finishes in under a second leaves nothing to count. That is +# not a silent pass: the peak simply reads 0 because there was no window. + +set -uo pipefail + +MAXSECS="${1:?usage: perf_watchdog.sh }"; shift +MAXPROCS="${1:?usage: perf_watchdog.sh }"; shift +[ "$#" -gt 0 ] || { echo "perf_watchdog: no command given" >&2; exit 2; } + +echo "perf_watchdog: cap ${MAXSECS}s / ${MAXPROCS} procs :: $*" + +setsid "$@" & +ROOT=$! + +# Count every live pid whose ancestry reaches ROOT. /proc//task// +# children is the kernel's own parent list, so this costs one small read per +# process rather than a full ps scan. +count_tree() { + local frontier=("$ROOT") total=0 pid kids n + while [ ${#frontier[@]} -gt 0 ]; do + pid="${frontier[-1]}" + frontier=("${frontier[@]:0:${#frontier[@]}-1}") + total=$((total + 1)) + kids=$(cat "/proc/$pid/task/$pid/children" 2>/dev/null) || continue + for n in $kids; do frontier+=("$n"); done + done + echo "$total" +} + +kill_tree() { + local frontier=("$ROOT") pid kids n + while [ ${#frontier[@]} -gt 0 ]; do + pid="${frontier[-1]}" + frontier=("${frontier[@]:0:${#frontier[@]}-1}") + kids=$(cat "/proc/$pid/task/$pid/children" 2>/dev/null) + for n in $kids; do frontier+=("$n"); done + kill -9 "$pid" 2>/dev/null + done +} + +START=$(date +%s) +PEAK=0 +KILLED="no" + +while kill -0 "$ROOT" 2>/dev/null; do + NOW=$(date +%s); ELAPSED=$((NOW - START)) + + if [ "$ELAPSED" -ge "$MAXSECS" ]; then + echo "perf_watchdog: WALL LIMIT at ${ELAPSED}s, killing tree" >&2 + KILLED="wall"; break + fi + + N=$(count_tree) + [ "$N" -gt "$PEAK" ] && PEAK=$N + + if [ "$N" -gt "$MAXPROCS" ]; then + echo "perf_watchdog: PROCESS COUNT ${N} > ${MAXPROCS} at ${ELAPSED}s, killing tree" >&2 + KILLED="procs"; break + fi + + sleep 1 +done + +if [ "$KILLED" != "no" ]; then + kill_tree + sleep 1 +fi + +wait "$ROOT" 2>/dev/null +RC=$? +ELAPSED=$(( $(date +%s) - START )) + +echo "perf_watchdog: rc=$RC elapsed=${ELAPSED}s peak_tree_procs=$PEAK killed=$KILLED" +[ "$KILLED" != "no" ] && exit 99 +exit "$RC" \ No newline at end of file diff --git a/src/advisory.rs b/src/advisory.rs index dd386eb..57e9c7a 100644 --- a/src/advisory.rs +++ b/src/advisory.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use std::io::Read; +use std::thread::{self, JoinHandle}; use crate::error::BluelineError; use crate::policy::Policy; @@ -122,27 +123,6 @@ fn osv_ecosystem(ecosystem: Ecosystem) -> &'static str { } } -/// Whether an advisory report that arrived as `Ok` means *no answer was -/// obtained* rather than *the answer was clean*. -/// -/// `fetch_advisories` returns `Ok` for both, so a caller that only inspects the -/// `Err` arm cannot tell a completed lookup from an abandoned one. That -/// distinction is the whole disclosure: an `unverified` report has no hits, so -/// nothing downstream raises a finding on its own, and a verdict built from it -/// is identical to a clean advisory pass. -/// -/// Two `unverified` reports are deliberately not treated the same: -/// -/// * Advisories switched off in policy is the operator's own choice, made -/// knowingly before the review ran. It is not a hole discovered at review -/// time, so it stays out of the verdict. -/// * Everything else -- a refused connection, a timeout, a body that would not -/// parse, or an ecosystem with no advisory coverage at all -- means this -/// release's revocation coverage is *unknown* rather than clear, and the -/// caller must say so. -/// -/// Told apart by policy rather than by matching on the message text, so -/// rewording a disclosure cannot silently change which of the two it is. pub(crate) fn coverage_unknown(report: &AdvisoryReport, policy: &Policy) -> Option { if !policy.policy.check_advisories || report.status != AdvisoryStatus::Unverified { return None; @@ -155,18 +135,128 @@ pub(crate) fn coverage_unknown(report: &AdvisoryReport, policy: &Policy) -> Opti ) } -pub fn fetch_advisories( +/// Advisories for one `package@version`, resolved as far as it goes without +/// stopping the caller. +/// +/// The answer has two shapes. `Ready` means nothing crossed the network: the +/// recall index hit, the policy disabled the check, or the advisory cache was +/// warm. `Pending` means the OSV round trip has already started and is still +/// in flight, and `PendingAdvisory::join` is the only way to read its answer. +pub enum AdvisoryResolution { + Ready(AdvisoryReport), + Pending(PendingAdvisory), +} + +impl AdvisoryResolution { + /// The report, for the paths that must not pend: a recall hit, a policy + /// switch, a warm cache, an ecosystem with no coverage. + /// + /// `Err` carries the same message `join` would, so a caller that only ever + /// expects a synchronous answer gets the same disclosure either way. + pub fn into_report(self) -> Result { + match self { + Self::Ready(report) => Ok(report), + Self::Pending(pending) => pending.join(None), + } + } +} + +/// An OSV query that is already running. Dropping one unjoined is safe: the +/// worker finishes and its answer is discarded with it. +/// +/// On an early-exit path between the resolve call and the join — a refused +/// tarball fetch, a failed extraction, a missing checksum — a `Pending` is +/// dropped unjoined, which spends the round trip and discards both the answer +/// and the cache write it would have made. That is accepted: those paths are +/// already erroring out of the review, and re-querying on the next run is +/// cheaper than threading a join through every early exit. +pub struct PendingAdvisory { + worker: JoinHandle, + stale_fallback: Option, + fail_closed: bool, + cache_key: Option, +} + +struct AdvisoryCacheKey { + ecosystem: Ecosystem, + package: String, + version: String, +} + +struct OsvOutcome { + report: Result, + ttl_secs: i64, +} + +impl PendingAdvisory { + /// Block for the query, then write the answer to the advisory cache. + /// + /// The write stays here instead of in the worker so the worker never needs + /// the SQLite connection, which is neither `Sync` nor worth reopening. The + /// write is local and bounded; the round trip it used to sit behind is not. + pub fn join(self, store: Option<&BaselineStore>) -> Result { + let Self { + worker, + stale_fallback, + fail_closed, + cache_key, + } = self; + + let OsvOutcome { report, ttl_secs } = match worker.join() { + Ok(outcome) => outcome, + // A panicked worker is indistinguishable from the network failing, + // so it takes the same path a transport error would. + Err(_) => { + return fallback_or_fail( + stale_fallback, + fail_closed, + "OSV advisory worker thread panicked", + ); + } + }; + + let report = match report { + Ok(report) => report, + Err(msg) => return fallback_or_fail(stale_fallback, fail_closed, &msg), + }; + + if let (Some(store), Some(key)) = (store, cache_key) + && let Ok(json) = serde_json::to_string(&report) + { + let _ = store.put_cached_advisories( + key.ecosystem, + &key.package, + &key.version, + &json, + report.hits.len(), + report.has_blocking(), + ttl_secs, + ); + } + + Ok(report) + } +} + +/// Start advisory resolution for `package@version`. +/// +/// Everything answerable without a network round trip is answered here. A +/// cache miss hands back a `Pending` with the OSV query already running, so the +/// caller can fetch, extract and diff while the network works. That ordering is +/// the whole point: the diff does not depend on the advisories, and on a small +/// package the round trip was roughly three quarters of the wall clock. +pub fn resolve_advisories( package: &str, version: &str, ecosystem: Ecosystem, store: Option<&BaselineStore>, policy: &Policy, -) -> Result { +) -> Result { // The recall index is local, curated truth: a hit blocks regardless of // the OSV path, and never routes through the advisory cache (a stale // OSV cache entry must not mask a fresh revocation). if let Some(revocation) = crate::recall::lookup(ecosystem, package, version)? { - return Ok(AdvisoryReport { + return Ok(AdvisoryResolution::Ready(AdvisoryReport { status: AdvisoryStatus::Vulnerable, hits: vec![AdvisoryItem { id: revocation.id, @@ -179,13 +269,13 @@ pub fn fetch_advisories( }], source: "blueline-recall".to_string(), message: None, - }); + })); } if !policy.policy.check_advisories { - return Ok(AdvisoryReport::unverified( + return Ok(AdvisoryResolution::Ready(AdvisoryReport::unverified( "advisory checking disabled by policy", - )); + ))); } // The AUR has no OSV coverage. `osv_ecosystem` maps it to a placeholder @@ -199,10 +289,10 @@ pub fn fetch_advisories( // The recall index above still applies, so a curated revocation blocks an // AUR package exactly as it blocks any other. if ecosystem == Ecosystem::Aur { - return Ok(AdvisoryReport::unverified( + return Ok(AdvisoryResolution::Ready(AdvisoryReport::unverified( "the AUR has no OSV advisory coverage; revocation coverage for this release comes \ from the recall index only", - )); + ))); } // 1. Check SQLite cache @@ -212,7 +302,7 @@ pub fn fetch_advisories( { if !cached.is_expired { if let Ok(report) = serde_json::from_str::(&cached.advisories_json) { - return Ok(report); + return Ok(AdvisoryResolution::Ready(report)); } } else if let Ok(mut report) = serde_json::from_str::(&cached.advisories_json) @@ -222,110 +312,109 @@ pub fn fetch_advisories( } } - // 2. Query OSV.dev REST API - // - // The shared registry agent, so this request carries the same SSRF guards the - // registry fetches do: `redirects(0)` and a resolver that validates every - // resolved address against the configured host. - // - // This is defence in depth, and the reason is worth stating precisely so the - // comment does not go stale in either direction. Probing found no case in - // which OSV redirects — 108+ requests across unknown shapes, malformed - // bodies and unusual names all answered directly — so the redirect guard is - // not holding up a currently-observed behaviour. It stays because "the origin - // we do not control does not currently do this" is not a property worth - // relying on: OSV is a third party, its edge configuration can change without - // notice, and a bare `AgentBuilder` follows up to 5 redirects to any host with - // no validation on any hop, so a single changed response header would walk - // this request off api.osv.dev and onto a link-local address. The guard costs - // one agent and no behaviour, and the alternative is depending on the origin's - // current configuration for our SSRF posture. + let worker = thread::spawn({ + let (package, version, policy) = (package.to_string(), version.to_string(), policy.clone()); + move || { + query_osv(OsvQuery { + package, + version, + ecosystem, + policy, + }) + } + }); + + Ok(AdvisoryResolution::Pending(PendingAdvisory { + worker, + stale_fallback, + fail_closed: policy.policy.fail_closed_network, + cache_key: Some(AdvisoryCacheKey { + ecosystem, + package: package.to_string(), + version: version.to_string(), + }), + })) +} + +struct OsvQuery { + package: String, + version: String, + ecosystem: Ecosystem, + policy: Policy, +} + +/// The whole OSV request: transport, parse, and the cache lifetime the answer +/// should get. Returns the transport error rather than resolving it, because +/// the stale-cache fallback is decided on the joining thread. +/// +/// The shared registry agent, so this request carries the same SSRF guards the +/// registry fetches do: `redirects(0)` and a resolver that validates every +/// resolved address against the configured host. +fn query_osv(query: OsvQuery) -> OsvOutcome { let agent = crate::registry::http_util::registry_agent_with_timeout( "blueline-security/0.1.0", OSV_BASE_URL, std::time::Duration::from_millis(DEFAULT_TIMEOUT_MS), ); - let query_payload = serde_json::json!({ - "version": version, + let payload_str = serde_json::json!({ + "version": query.version, "package": { - "name": package, - "ecosystem": osv_ecosystem(ecosystem) + "name": query.package, + "ecosystem": osv_ecosystem(query.ecosystem) } - }); + }) + .to_string(); - let payload_str = query_payload.to_string(); - let resp_result = agent + let report = match agent .post(&format!("{OSV_BASE_URL}/v1/query")) .set("Content-Type", "application/json") .set("User-Agent", "blueline-security/0.1.0") - .send_string(&payload_str); - - match resp_result { + .send_string(&payload_str) + { Ok(resp) => { let mut reader = resp.into_reader().take(MAX_OSV_RESPONSE_BYTES); let mut body = String::new(); - if let Err(e) = reader.read_to_string(&mut body) { - return fallback_or_fail( - stale_fallback, - policy, - &format!("failed to read OSV response body: {e}"), - ); + match reader.read_to_string(&mut body) { + Err(e) => Err(format!("failed to read OSV response body: {e}")), + Ok(_) => match serde_json::from_str::(&body) { + Err(e) => Err(format!("invalid JSON from OSV API: {e}")), + Ok(r) => Ok(parse_osv_response(r, &query.policy)), + }, } + } + Err(e) => Err(format!("OSV advisory request failed: {e}")), + }; - let osv_resp: OsvQueryResponse = match serde_json::from_str(&body) { - Ok(r) => r, - Err(e) => { - return fallback_or_fail( - stale_fallback, - policy, - &format!("invalid JSON from OSV API: {e}"), - ); - } - }; - - let report = parse_osv_response(osv_resp, policy); - - // Cache report in SQLite store - if let Some(store) = store - && let Ok(report_json) = serde_json::to_string(&report) - { - let ttl = if report.hits.is_empty() { - policy.advisories.clean_cache_ttl_secs() - } else { - policy.advisories.vulnerable_cache_ttl_secs() - }; - let _ = store.put_cached_advisories( - ecosystem, - package, - version, - &report_json, - report.hits.len(), - report.has_blocking(), - ttl, - ); - } + let ttl_secs = cache_ttl_secs(&report, &query.policy); - Ok(report) - } - Err(e) => fallback_or_fail( - stale_fallback, - policy, - &format!("OSV advisory request failed: {e}"), - ), + OsvOutcome { report, ttl_secs } +} + +/// Cache lifetime for an OSV answer: a clean lookup is cached longer than a +/// vulnerable one, and a failed lookup is never cached. +/// +/// Split out of `query_osv` so the three-way choice is testable without a +/// network: the guard distinguishing a clean report from a vulnerable one is +/// exactly the line a `hits.is_empty()` mutation would silently erase. +fn cache_ttl_secs(report: &Result, policy: &Policy) -> i64 { + match report { + Ok(report) if report.hits.is_empty() => policy.advisories.clean_cache_ttl_secs(), + Ok(_) => policy.advisories.vulnerable_cache_ttl_secs(), + Err(_) => 0, } } fn fallback_or_fail( stale_fallback: Option, - policy: &Policy, + fail_closed: bool, err_msg: &str, ) -> Result { // Checked before the stale fallback. Serving a cached report while the // policy says fail closed returned a possibly month-old answer as if it // were fresh, and a stale CLEAN report produces no hits and no staleness // disclosure, so it was indistinguishable from a clean pass. - if policy.policy.fail_closed_network { + if fail_closed { return Err(BluelineError::Advisory(format!( "{err_msg} (failing closed as configured by policy)" ))); @@ -694,6 +783,46 @@ fn calculate_advisory_severity( #[cfg(test)] mod tests { + #[test] + fn the_cache_ttl_distinguishes_clean_from_vulnerable_from_failed() { + // The `hits.is_empty()` guard is the whole decision: clean caches for + // 12h, vulnerable for 1h, failed for 0. A mutation that flips the + // guard to `true` or `false` collapses two of the three, so pin all + // three against the distinct policy defaults (12h / 1h). + let policy = Policy::default(); + let clean = AdvisoryReport::clean("osv.dev"); + let mut vulnerable = AdvisoryReport::clean("osv.dev"); + vulnerable.status = AdvisoryStatus::Vulnerable; + vulnerable.hits = vec![AdvisoryItem { + id: "GHSA-test".into(), + summary: "test".into(), + details: String::new(), + aliases: Vec::new(), + severity: VerdictBand::High, + cvss_score: None, + is_malware: false, + }]; + + assert!(clean.hits.is_empty()); + assert!(!vulnerable.hits.is_empty()); + + assert_eq!( + cache_ttl_secs(&Ok(clean.clone()), &policy), + 12 * 3600, + "a clean report caches for the clean TTL" + ); + assert_eq!( + cache_ttl_secs(&Ok(vulnerable.clone()), &policy), + 3600, + "a vulnerable report caches for the shorter vulnerable TTL" + ); + assert_eq!( + cache_ttl_secs(&Err("network down".to_string()), &policy), + 0, + "a failed lookup is never cached" + ); + } + /// A v2 vector's metric values are per-metric. Matching them by value alone /// let `AV:P` be scored with the partial-impact weight, so a malformed /// vector scored 4.1 where the well-formed one scores 7.5 -- an advisory @@ -741,16 +870,149 @@ mod tests { source: "cache".to_string(), message: None, }; - let result = fallback_or_fail(Some(stale.clone()), &policy, "osv unreachable"); + let result = fallback_or_fail(Some(stale.clone()), true, "osv unreachable"); assert!( result.is_err(), "fail_closed_network must not be satisfied by a cached report" ); // Without the policy the fallback is still the answer. - let mut lenient = Policy::default(); - lenient.policy.fail_closed_network = false; - assert!(fallback_or_fail(Some(stale), &lenient, "osv unreachable").is_ok()); + assert!(fallback_or_fail(Some(stale), false, "osv unreachable").is_ok()); + } + + #[test] + fn a_cache_miss_defers_the_query_instead_of_running_it_inline() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + // A cold cache must hand back work in flight, not an answer. Anything + // else puts the round trip back on the critical path, which is the + // entire reason this is two phases. + assert!( + matches!(resolution, AdvisoryResolution::Pending(_)), + "expected a pending query, got a ready answer" + ); + } + + #[test] + fn the_cache_is_answered_without_a_query_at_all() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + let name = "blueline-no-such-package-xyz"; + + store + .put_cached_advisories( + Ecosystem::Npm, + name, + "1.0.0", + r#"{"status":"clean","hits":[],"source":"osv.dev","message":null}"#, + 0, + false, + 3600, + ) + .unwrap(); + + let resolution = + resolve_advisories(name, "1.0.0", Ecosystem::Npm, Some(&store), &policy).unwrap(); + + match resolution { + AdvisoryResolution::Ready(report) => { + assert_eq!(report.status, AdvisoryStatus::Clean); + } + AdvisoryResolution::Pending(_) => { + panic!("a warm cache must not start a query") + } + } + } + + #[test] + fn disabling_advisories_never_starts_a_query() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let mut policy = Policy::default(); + policy.policy.check_advisories = false; + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + match resolution { + AdvisoryResolution::Ready(report) => { + assert_eq!(report.status, AdvisoryStatus::Unverified); + } + AdvisoryResolution::Pending(_) => { + panic!("advisories are off; no query should exist") + } + } + } + + #[test] + fn dropping_a_pending_query_is_not_a_use_of_its_answer() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + // A review that refuses before scoring never joins. The worker must + // simply finish and be forgotten, not leak or panic. + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + drop(resolution); + } + + #[test] + fn joining_always_yields_a_report_even_when_the_network_fails() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + let pending = match resolution { + AdvisoryResolution::Pending(pending) => pending, + AdvisoryResolution::Ready(report) => return assert!(report.hits.is_empty()), + }; + + // Either the query answered or it did not. Both are reports: the + // fail-closed and unverified decisions belong to the caller, which is + // the same place they lived before the query moved off the main thread. + let outcome = pending.join(Some(&store)); + if let Err(e) = &outcome { + assert!( + e.to_string() + .contains("failing closed as configured by policy"), + "unexpected failure shape: {e}" + ); + } } #[test] @@ -856,14 +1118,16 @@ mod tests { /// quietly turn this test back into a tautology. #[test] fn the_aur_never_claims_clean_advisory_coverage() { - let report = fetch_advisories( + let report = resolve_advisories( "webtorrent-desktop", "1.9.0-1", crate::registry::Ecosystem::Aur, None, &Policy::default(), ) - .expect("AUR advisory lookup must not error"); + .expect("AUR advisory lookup must not error") + .into_report() + .expect("AUR advisory resolution must not pend"); assert_eq!( report.status, @@ -885,14 +1149,16 @@ mod tests { fn policy_disabled_advisories_stay_distinguishable_from_a_failed_lookup() { let mut policy = Policy::default(); policy.policy.check_advisories = false; - let report = fetch_advisories( + let report = resolve_advisories( "webtorrent-desktop", "1.9.0-1", crate::registry::Ecosystem::Aur, None, &policy, ) - .expect("disabling advisories must not error"); + .expect("disabling advisories must not error") + .into_report() + .expect("a disabled policy must never pend"); assert_eq!(report.status, AdvisoryStatus::Unverified); assert!( diff --git a/src/ci.rs b/src/ci.rs index d7dc27e..3126502 100644 --- a/src/ci.rs +++ b/src/ci.rs @@ -3,6 +3,8 @@ use std::fs; use std::io::Write; use std::path::Path; use std::process::Command; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; use serde::{Deserialize, Serialize}; @@ -276,10 +278,12 @@ pub fn evaluate_lockfile_diff( policy.ci.max_evaluations, )?; - let mut items = Vec::new(); - let mut max_band = VerdictBand::Low; - - let eval_candidates = delta + // One entry per package to review, in the order the report must print + // them. CI output is diffed between runs, so that order is a contract: it + // is fixed here, before any review starts, and never depends on which + // review happens to finish first. + let mut candidates: Vec = Vec::new(); + for (name, old_version, new_version, is_dev, resolved) in delta .added .iter() .map(|e| (&e.name, None, &e.version, e.is_dev, e.resolved.as_deref())) @@ -291,9 +295,8 @@ pub fn evaluate_lockfile_diff( u.is_dev, u.resolved.as_deref(), ) - })); - - for (name, old_version, new_version, is_dev, resolved) in eval_candidates { + })) + { if evaluation_skipped(policy.ci.include_dev, is_dev) { continue; } @@ -306,48 +309,21 @@ pub fn evaluate_lockfile_diff( continue; } - let mut rctx = crate::recursive::ReviewContext::new(policy, ctx.bases.clone()); - let (mut verdict, _, checksum, _) = - evaluate_package(name, new_version, ctx.ecosystem, store, policy, &mut rctx)?; - - // If lockfile declared a hash, verify it matches. The PEP 503 - // alias lookup is PyPI-only: on npm/cargo `foo_bar` and `foo-bar` - // are distinct packages and must never share an integrity entry. - let canon_alias = (ctx.ecosystem == Ecosystem::PyPi) - .then(|| head_integrity_map.get(&crate::version::canonicalize_name(name))) - .flatten(); - if let Some(expected_integ) = head_integrity_map.get(name).or(canon_alias) { - let matches_integ = expected_integ.split_whitespace().any(|expected_one| { - let expected_hex = expected_one.strip_prefix("sha256:").unwrap_or(expected_one); - checksum.value_hex.eq_ignore_ascii_case(expected_hex) - }); - - if !matches_integ { - verdict.findings.push(crate::verdict::Finding { - rule_id: "R10_LOCKFILE_HASH_MISMATCH".into(), - severity: VerdictBand::Block, - title: format!("Lockfile hash mismatch for `{name}`"), - description: format!( - "Lockfile pinned `{expected_integ}`, but downloaded release has checksum `{}`", - checksum.to_display() - ), - }); - verdict.band = VerdictBand::Block; - verdict.risk_score = verdict.risk_score.saturating_add(50); - } - } - - max_band = update_max_band(max_band, verdict.band); - - items.push(CiReviewItem { + candidates.push(Candidate { name: name.to_string(), old_version, new_version: new_version.to_string(), is_dev, - verdict, }); } + let items = evaluate_candidates(&candidates, ctx, store, policy, &head_integrity_map)?; + + let mut max_band = VerdictBand::Low; + for item in &items { + max_band = update_max_band(max_band, item.verdict.band); + } + let fail_threshold = ctx .fail_on .unwrap_or_else(|| parse_band_str(&policy.ci.fail_on).unwrap_or(VerdictBand::High)); @@ -366,10 +342,178 @@ pub fn evaluate_lockfile_diff( }) } +/// One package the CI lane owes a review, already filtered. +struct Candidate { + name: String, + old_version: Option, + new_version: String, + is_dev: bool, +} + +/// Upper bound on review workers. +/// +/// A lockfile lane is network-bound, so more workers is faster right up until +/// the registry starts refusing connections. Eight covers a slow round trip +/// several times over on a laptop and stays well inside what one client asks a +/// registry to handle. +/// +/// There is no rate limiting or jitter, so a burst lands as-is: up to eight +/// concurrent packument and tarball fetches, a few dozen requests in flight. +/// That is unremarkable against npm or crates.io and a spike against a small +/// self-hosted mirror, which `--registry` permits. Lower the cap before +/// pointing this lane at one. +const MAX_CI_REVIEW_WORKERS: usize = 8; + +fn evaluate_candidate( + candidate: &Candidate, + ctx: &CiContext<'_>, + store: &BaselineStore, + policy: &Policy, + head_integrity_map: &std::collections::HashMap, +) -> anyhow::Result { + let name = candidate.name.as_str(); + let new_version = candidate.new_version.as_str(); + + let mut rctx = crate::recursive::ReviewContext::new(policy, ctx.bases.clone()); + let (mut verdict, _, checksum, _) = + evaluate_package(name, new_version, ctx.ecosystem, store, policy, &mut rctx)?; + + // If lockfile declared a hash, verify it matches. The PEP 503 + // alias lookup is PyPI-only: on npm/cargo `foo_bar` and `foo-bar` + // are distinct packages and must never share an integrity entry. + let canon_alias = (ctx.ecosystem == Ecosystem::PyPi) + .then(|| head_integrity_map.get(&crate::version::canonicalize_name(name))) + .flatten(); + if let Some(expected_integ) = head_integrity_map.get(name).or(canon_alias) { + let matches_integ = expected_integ.split_whitespace().any(|expected_one| { + let expected_hex = expected_one.strip_prefix("sha256:").unwrap_or(expected_one); + checksum.value_hex.eq_ignore_ascii_case(expected_hex) + }); + + if !matches_integ { + verdict.findings.push(crate::verdict::Finding { + rule_id: "R10_LOCKFILE_HASH_MISMATCH".into(), + severity: VerdictBand::Block, + title: format!("Lockfile hash mismatch for `{name}`"), + description: format!( + "Lockfile pinned `{expected_integ}`, but downloaded release has checksum `{}`", + checksum.to_display() + ), + }); + verdict.band = VerdictBand::Block; + verdict.risk_score = verdict.risk_score.saturating_add(50); + } + } + + Ok(CiReviewItem { + name: candidate.name.clone(), + old_version: candidate.old_version.clone(), + new_version: candidate.new_version.clone(), + is_dev: candidate.is_dev, + verdict, + }) +} + +/// Review every candidate and return the items in candidate order. +/// +/// Each worker gets its own `ReviewContext` and its own database handle, since +/// a context holds `RefCell`s and a `rusqlite::Connection` is `Send` but not +/// `Sync`. Neither can be shared, and a mutex over either would serialise the +/// review rather than just the database. What the workers do share is the +/// network, which is where this lane spends nearly all of its time: at the +/// default cap of a hundred packages and a few hundred milliseconds of round +/// trip apiece, the serial order was about a minute of mostly waiting. +/// +/// The result is ordered by candidate index rather than by completion, so the +/// report matches the serial one no matter which worker finished first. +fn evaluate_candidates( + candidates: &[Candidate], + ctx: &CiContext<'_>, + store: &BaselineStore, + policy: &Policy, + head_integrity_map: &std::collections::HashMap, +) -> anyhow::Result> { + let workers = std::thread::available_parallelism() + .map_or(1, |n| n.get()) + .clamp(1, MAX_CI_REVIEW_WORKERS) + .min(candidates.len()); + + if workers <= 1 { + return candidates + .iter() + .map(|c| evaluate_candidate(c, ctx, store, policy, head_integrity_map)) + .collect(); + } + + let slots: Vec>>> = + (0..candidates.len()).map(|_| Mutex::new(None)).collect(); + let next = AtomicUsize::new(0); + + // Opened here rather than inside a worker: a `rusqlite::Connection` is + // `Send` but not `Sync`, so it can be moved onto a worker thread but the + // shared reference it came from cannot be captured by one. + let worker_stores: Vec = (0..workers) + .map(|_| store.reopen()) + .collect::>()?; + + std::thread::scope(|scope| { + for worker_store in worker_stores { + let slots = &slots; + let next = &next; + scope.spawn(move || { + loop { + let index = next.fetch_add(1, Ordering::Relaxed); + let Some(candidate) = candidates.get(index) else { + break; + }; + let outcome = evaluate_candidate( + candidate, + ctx, + &worker_store, + policy, + head_integrity_map, + ); + *slots[index] + .lock() + .expect("slot written once, never panicked under") = Some(outcome); + } + }); + } + }); + + slots + .into_iter() + .map(|slot| { + slot.into_inner() + .expect("slot written once, never panicked under") + // Every index below `candidates.len()` is claimed exactly once: + // `next` only ever hands out indices the `get` above resolves, + // each worker claims one per loop iteration, and the scope + // joins every worker before this line runs. The claim is + // load-bearing, so it is asserted rather than silently skipped. + .expect("every index is claimed before the scope ends") + }) + // All candidates are drained, including after a failure, where the + // serial lane used `?` to abort at the first error. That is a + // deliberate decision, not an artifact of the threading: the store + // writes a failing package still makes (`record_verified`, the + // advisory-cache fill) are evidence-only — `clean = 0` and idempotent + // — so continuing widens what a failed run mutates in the trust + // anchor without changing any verdict, and one package's transport + // failure does not invalidate the other ninety-nine reviews. + .collect() +} + /// Review the AUR CI file diff: every added or version-changed pkgbase gets /// a full `evaluate_package` review. Removed entries need no review. /// The stored integrity for each item is the sha256 of the pinned commit's /// `git archive` bytes, so the audit trail stays bound to the commit hash. +/// +/// This half of the lane is still serial, unlike the lockfile half: AUR +/// entries are capped at 4096 and each review shells out to `git archive`, +/// so the worst case is real. Parallelising it is the same shape as +/// `evaluate_candidates`, minus the registry round trips that make the +/// lockfile lane network-bound. fn evaluate_aur_ci_diff( base_content: &str, head_content: &str, @@ -419,7 +563,6 @@ fn evaluate_aur_ci_diff( verdict, }); } - let fail_threshold = ctx .fail_on .unwrap_or_else(|| parse_band_str(&policy.ci.fail_on).unwrap_or(VerdictBand::High)); @@ -1538,6 +1681,180 @@ source = "path+file:///some/local/path" assert_eq!(report.removed_count, 0); } + /// CI output is diffed between runs, so the order of `items` is a + /// contract rather than an accident of scheduling. The first candidate's + /// packument is served after a deliberate delay and the rest answer + /// instantly, so a lane that emitted items as they finished would put + /// `aaa-slow` last. It must come first. + #[test] + fn parallel_reviews_still_report_in_candidate_order() { + use std::io::{Read, Write}; + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + fn npm_tgz(name: &str, version: &str) -> Vec { + use std::io::Write as _; + let manifest = format!(r#"{{"name":"{name}","version":"{version}"}}"#); + let mut tar_data = Vec::new(); + { + let mut builder = tar::Builder::new(&mut tar_data); + let mut header = tar::Header::new_gnu(); + header.set_size(manifest.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder + .append_data(&mut header, "package/package.json", manifest.as_bytes()) + .unwrap(); + builder.finish().unwrap(); + } + let mut encoder = + flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(&tar_data).unwrap(); + encoder.finish().unwrap() + } + + fn sri_sha512(bytes: &[u8]) -> String { + use base64::Engine as _; + use sha2::Digest as _; + let mut hasher = sha2::Sha512::new(); + hasher.update(bytes); + format!( + "sha512-{}", + base64::engine::general_purpose::STANDARD.encode(hasher.finalize()) + ) + } + + const SLOW: &str = "aaa-slow"; + const FAST: [&str; 4] = ["bbb-fast", "ccc-fast", "ddd-fast", "eee-fast"]; + const SLOW_DELAY_MS: u64 = 700; + + let mut names = vec![SLOW.to_string()]; + names.extend(FAST.iter().map(|s| s.to_string())); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let base = format!("http://127.0.0.1:{port}"); + + let tgz: std::collections::HashMap> = names + .iter() + .map(|n| (n.clone(), npm_tgz(n, "1.0.0"))) + .collect(); + let packs: std::collections::HashMap> = names + .iter() + .map(|n| { + let body = format!( + r#"{{"name":"{n}","dist-tags":{{"latest":"1.0.0"}},"versions":{{"1.0.0":{{"name":"{n}","version":"1.0.0","dist":{{"tarball":"{base}/{n}-1.0.0.tgz","integrity":"{}"}}}}}}}}"#, + sri_sha512(&tgz[n]) + ); + (n.clone(), body.into_bytes()) + }) + .collect(); + + let stop = Arc::new(AtomicBool::new(false)); + let stop_srv = Arc::clone(&stop); + let packs_srv = packs.clone(); + let tgz_srv = tgz.clone(); + let handle = std::thread::spawn(move || { + listener.set_nonblocking(true).unwrap(); + while !stop_srv.load(Ordering::SeqCst) { + match listener.accept() { + Ok((mut stream, _)) => { + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).unwrap_or(0); + let req = String::from_utf8_lossy(&buf[..n]).to_string(); + let path = req + .lines() + .next() + .and_then(|l| l.split_whitespace().nth(1)) + .unwrap_or("/"); + let body: Vec = if let Some(name) = path.strip_prefix('/') { + if let Some(rest) = name.strip_suffix("-1.0.0.tgz") { + tgz_srv.get(rest).cloned().unwrap_or_default() + } else { + if name == SLOW { + std::thread::sleep(std::time::Duration::from_millis( + SLOW_DELAY_MS, + )); + } + packs_srv.get(name).cloned().unwrap_or_default() + } + } else { + Vec::new() + }; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&body); + } + Err(_) => std::thread::sleep(std::time::Duration::from_millis(5)), + } + } + }); + + let mut base_packages = serde_json::Map::new(); + base_packages.insert( + "".into(), + serde_json::json!({"name": "app", "version": "1.0.0"}), + ); + for n in &names { + base_packages.insert( + format!("node_modules/{n}"), + serde_json::json!({"version": "0.9.0"}), + ); + } + let base_lock = serde_json::json!({ + "lockfileVersion": 3, + "packages": base_packages + }) + .to_string(); + + let mut head_packages = serde_json::Map::new(); + head_packages.insert( + "".into(), + serde_json::json!({"name": "app", "version": "1.0.0"}), + ); + for n in &names { + head_packages.insert( + format!("node_modules/{n}"), + serde_json::json!({"version": "1.0.0"}), + ); + } + let head_lock = serde_json::json!({ + "lockfileVersion": 3, + "packages": head_packages + }) + .to_string(); + + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("blueline.db")).unwrap(); + let mut policy = Policy::default(); + policy.policy.check_advisories = false; + let bases = test_bases(&base); + let ctx = CiContext { + base_ref: "origin/main", + lockfile_path: "package-lock.json", + bases: &bases, + fail_on: Some(VerdictBand::Block), + ecosystem: crate::registry::Ecosystem::Npm, + }; + + let result = evaluate_lockfile_diff(&base_lock, &head_lock, &ctx, &store, &policy); + stop.store(true, Ordering::SeqCst); + let _ = handle.join(); + let report = result.unwrap(); + + assert_eq!(report.items.len(), names.len(), "one item per candidate"); + let got: Vec<&str> = report.items.iter().map(|i| i.name.as_str()).collect(); + assert_eq!( + got, names, + "items must follow candidate order, not finish order" + ); + } + #[test] fn lockfile_diff_keeps_pypi_alias_off_cargo_pins() { use std::io::{Read, Write}; diff --git a/src/diff.rs b/src/diff.rs index b315efc..93adda7 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -1,5 +1,6 @@ use std::collections::BTreeMap; use std::fs; +use std::io::Read; use std::path::{Path, PathBuf}; use sha2::{Digest, Sha256}; @@ -300,37 +301,162 @@ fn scan_tree(root: &Path) -> Result, BluelineErr .map_err(|e| BluelineError::Extraction(format!("path strip error: {e}")))? .to_path_buf(); - let bytes = fs::read(path).map_err(|e| { - BluelineError::Extraction(format!("reading {}: {e}", path.display())) - })?; - let mut hasher = Sha256::new(); - hasher.update(&bytes); - let hash = hasher.finalize().into(); - - let kind = classify_bytes(&bytes); - let is_executable = check_is_executable(path); - - files.insert( - rel, - DiskFileMeta { - hash, - kind, - is_executable, - }, - ); + files.insert(rel, scan_file(path)?); } } Ok(files) } -fn classify_bytes(bytes: &[u8]) -> FileKind { - if bytes.len() as u64 > MAX_DIFF_FILE_BYTES { - return FileKind::OpaqueTooLarge; +/// Window used to read each file. Large enough that syscall overhead is noise, +/// small enough that peak memory is set by this rather than by the package. +const SCAN_CHUNK_BYTES: usize = 64 * 1024; + +/// Longest UTF-8 sequence, so a code point split across two reads can be +/// carried over whole. +const UTF8_MAX_SEQ: usize = 4; + +/// Sequence length implied by a UTF-8 lead byte, or `None` if the byte cannot +/// start a valid sequence. +fn seq_len(lead: u8) -> Option { + match lead { + 0x00..=0x7f => Some(1), + 0xc2..=0xdf => Some(2), + 0xe0..=0xef => Some(3), + 0xf0..=0xf4 => Some(4), + _ => None, } - if bytes.contains(&0) { - return FileKind::Binary; +} + +/// Incremental UTF-8 validity check. +/// +/// `from_utf8` rejects a sequence that is merely truncated by the end of the +/// buffer, so the tail of a chunk has to be carried into the next one before it +/// can be judged. The carry is at most `UTF8_MAX_SEQ - 1` bytes, and a valid +/// result on the carried bytes means those bytes are settled, so the carry +/// never grows without bound. +#[derive(Default)] +struct Utf8Scan { + valid: bool, + pending: [u8; UTF8_MAX_SEQ], + pending_len: usize, +} + +impl Utf8Scan { + fn feed(&mut self, chunk: &[u8]) { + if !self.valid { + return; + } + let mut rest = chunk; + + if self.pending_len > 0 { + // Complete the carried sequence by its actual length, taken from the + // lead byte. Padding to a fixed four bytes would steal the lead of + // the next character whenever the split sequence is shorter, and + // from_utf8 would then reject the carry as malformed. + let expected = match seq_len(self.pending[0]) { + Some(n) => n, + None => { + self.valid = false; + return; + } + }; + let take = (expected - self.pending_len).min(rest.len()); + self.pending[self.pending_len..self.pending_len + take].copy_from_slice(&rest[..take]); + self.pending_len += take; + rest = &rest[take..]; + if self.pending_len < expected { + // Still short of a full sequence. Nothing new is judgeable yet. + return; + } + if std::str::from_utf8(&self.pending[..expected]).is_err() { + self.valid = false; + return; + } + self.pending_len = 0; + } + + match std::str::from_utf8(rest) { + Ok(_) => {} + Err(e) => match e.error_len() { + Some(_) => self.valid = false, + None => { + let tail = rest.len() - e.valid_up_to(); + self.pending[..tail].copy_from_slice(&rest[rest.len() - tail..]); + self.pending_len = tail; + } + }, + } + } + + /// Judge whatever carry is left once the file is exhausted. + /// + /// A sequence still incomplete at EOF is malformed input, not a truncation + /// of a longer buffer: there are no further bytes to complete it with. + /// Judging it here is what keeps a file that ends mid-code-point from + /// reading as text. + fn finish(&mut self) { + if self.pending_len > 0 { + self.valid = false; + } + } +} + +/// Read one file once and keep only what the comparison needs. +/// +/// The previous shape was `fs::read` followed by a hash and a classify over the +/// whole buffer. None of the three questions need the buffer: the size is a +/// running count, the digest is incremental by construction, and "is this +/// text" is a NUL scan plus a UTF-8 check. A fixed window answers all three, so +/// peak memory stops scaling with the largest file in the package. Every byte +/// still reaches the hasher, because the digest is what decides whether a file +/// is compared at all. +fn scan_file(path: &Path) -> Result { + let read_err = + |e: std::io::Error| BluelineError::Extraction(format!("reading {}: {e}", path.display())); + let mut file = fs::File::open(path).map_err(read_err)?; + + let mut hasher = Sha256::new(); + let mut buf = vec![0u8; SCAN_CHUNK_BYTES]; + let mut utf8 = Utf8Scan { + valid: true, + ..Default::default() + }; + let mut len: u64 = 0; + let mut has_nul = false; + + loop { + let n = file.read(&mut buf).map_err(read_err)?; + if n == 0 { + break; + } + let chunk = &buf[..n]; + hasher.update(chunk); + len += n as u64; + // Past the cap the file is opaque whatever its bytes look like, so the + // two scans stop. The hash keeps going: it decides the comparison. + if len <= MAX_DIFF_FILE_BYTES { + if !has_nul { + has_nul = chunk.contains(&0); + } + utf8.feed(chunk); + } } - if std::str::from_utf8(bytes).is_err() { + utf8.finish(); + + let hash = hasher.finalize().into(); + let kind = classify_bytes(len, has_nul, utf8.valid); + Ok(DiskFileMeta { + hash, + kind, + is_executable: check_is_executable(path), + }) +} + +fn classify_bytes(len: u64, has_nul: bool, utf8_valid: bool) -> FileKind { + if len > MAX_DIFF_FILE_BYTES { + return FileKind::OpaqueTooLarge; + } + if has_nul || !utf8_valid { return FileKind::Binary; } FileKind::Text @@ -470,6 +596,106 @@ fn is_executable_extension(path_str: &str) -> bool { mod tests { use super::*; + #[test] + fn seq_len_maps_every_utf8_lead_byte_class() { + // Each arm is the load-bearing half of the chunk-boundary carry: it + // decides how many bytes a split sequence still needs. Deleting an arm + // silently mis-classifies every file whose split lands on that class, + // so pin all four plus the two rejection ranges. + assert_eq!(seq_len(b'a'), Some(1)); // 0x00..=0x7f + assert_eq!(seq_len(0x7f), Some(1)); + assert_eq!(seq_len(0xc2), Some(2)); // 0xc2..=0xdf + assert_eq!(seq_len(0xdf), Some(2)); + assert_eq!(seq_len(0xe0), Some(3)); // 0xe0..=0xef + assert_eq!(seq_len(0xef), Some(3)); + assert_eq!(seq_len(0xf0), Some(4)); // 0xf0..=0xf4 + assert_eq!(seq_len(0xf4), Some(4)); + // Not valid lead bytes: continuation bytes and the two overlong ranges. + assert_eq!(seq_len(0x80), None); + assert_eq!(seq_len(0xbf), None); + assert_eq!(seq_len(0xc0), None); + assert_eq!(seq_len(0xc1), None); + assert_eq!(seq_len(0xf5), None); + assert_eq!(seq_len(0xff), None); + } + + #[test] + fn the_scan_window_is_sixty_four_kibibytes() { + // The window is the memory bound, and `64 * 1024` vs `64 + 1024` is a + // one-token swap the digest test cannot see. Pin the value so the + // bound the comment promises is the bound that ships. + assert_eq!(SCAN_CHUNK_BYTES, 64 * 1024); + } + + #[test] + fn classify_bytes_draws_the_opaque_boundary_above_the_cap_only() { + // `len > cap` vs `len >= cap` is an off-by-one at the exact boundary. + // A file exactly at the cap is still line-diffable, so it must stay + // Text; only one byte past it is opaque. + assert_eq!( + classify_bytes(MAX_DIFF_FILE_BYTES, false, true), + FileKind::Text, + "a file exactly at the cap must still be diffable" + ); + assert_eq!( + classify_bytes(MAX_DIFF_FILE_BYTES + 1, false, true), + FileKind::OpaqueTooLarge, + "one byte past the cap is opaque" + ); + } + + #[test] + fn a_carry_completes_only_once_the_whole_sequence_arrives() { + // The `pending_len < expected` guard is what keeps a half-arrived + // sequence from being judged before its bytes exist. Flipping it to + // `>` would judge an empty carry as complete and a full one as + // incomplete, so drive both sides of the boundary. + // + // `valid` starts true, exactly as `scan_file` initialises it; the + // `Default` derive would start it false and every feed would no-op. + let new_scan = || Utf8Scan { + valid: true, + ..Default::default() + }; + // A four-byte sequence delivered one byte at a time into the SAME + // scan: incomplete at every step but the last, and never judged until + // the fourth byte. This is the case the `pending_len < expected` guard + // protects — a partial carry meeting a follow-up feed that is still + // too short must not be judged. Flipping the guard to `>` would skip + // the early return and judge the half-filled carry as malformed. + let mut drip = new_scan(); + for (i, byte) in [0xf0u8, 0x9f, 0x98, 0x80].iter().enumerate() { + drip.feed(&[*byte]); + if i < 3 { + assert!( + drip.valid, + "after {i} byte(s) the carry is incomplete, not invalid" + ); + } + } + assert!( + drip.valid, + "the completed four-byte sequence is valid UTF-8" + ); + + // The completed sequence judges valid, and so does the same sequence + // split across two feeds. + let mut scan = new_scan(); + scan.feed(&[0xf0, 0x9f]); + assert!(scan.valid, "a two-byte carry is not yet judgeable"); + scan.feed(&[0x98, 0x80]); + assert!( + scan.valid, + "the completed four-byte sequence is valid UTF-8" + ); + + // A genuinely malformed lead byte is rejected as soon as it is the + // carry's first byte, which is what `seq_len` returning `None` drives. + let mut bad = new_scan(); + bad.feed(&[0xff]); + assert!(!bad.valid, "0xff cannot start a UTF-8 sequence"); + } + #[test] fn detects_file_added_removed_modified() { let old_dir = tempfile::tempdir().unwrap(); @@ -547,10 +773,78 @@ mod tests { #[test] fn scans_entire_file_for_nul_byte() { - // NUL byte placed past 8192 bytes - let mut data = vec![b'a'; 10000]; - data[9000] = 0; - assert_eq!(classify_bytes(&data), FileKind::Binary); + // NUL byte placed past 8192 bytes, and past the read window, so only a + // scan that keeps going to the end of the file can see it. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("late-nul.bin"); + let mut data = vec![b'a'; SCAN_CHUNK_BYTES * 3]; + data[SCAN_CHUNK_BYTES + 17] = 0; + fs::write(&path, &data).unwrap(); + + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + } + + #[test] + fn chunked_reads_agree_with_a_single_read_on_the_classification() { + // A multi-byte code point straddling the window boundary is the case + // most likely to be misjudged by an incremental check, so pin every + // offset of a four-byte sequence against the whole-buffer answer. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("edge.txt"); + let euro = "€".repeat(SCAN_CHUNK_BYTES); + for pad in 0..8usize { + let mut data = vec![b'a'; pad]; + data.extend_from_slice(euro.as_bytes()); + fs::write(&path, &data).unwrap(); + assert_eq!( + scan_file(&path).unwrap().kind, + FileKind::Text, + "misclassified with {pad} bytes of padding" + ); + } + + fs::write(&path, b"\xff\xfe not utf8").unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + + // A file that ends mid-code-point has no further bytes to complete the + // sequence with, so the leftover carry is malformed, not truncated. + for truncated in [&[0xc2][..], &[0xe2, 0x82][..], &[0xf0, 0x9f, 0x98][..]] { + fs::write(&path, truncated).unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + } + + let mut oversize = vec![b'a'; MAX_DIFF_FILE_BYTES as usize + 1]; + oversize[MAX_DIFF_FILE_BYTES as usize / 2] = 0; + fs::write(&path, &oversize).unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::OpaqueTooLarge); + } + + #[test] + fn the_windowed_digest_equals_the_whole_file_digest() { + // The hash decides whether a file is compared at all + // (`if base_meta.hash != target_meta.hash`), so a digest that covered + // fewer bytes than it read would make base and target agree on a + // changed file and silently skip the comparison. Every byte must + // reach the hasher, so pin the digest against the whole-buffer + // answer across sizes that cross the window boundary at every offset + // of one, plus a zero-byte file and an exact multiple of the window. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("digest.bin"); + + let mut sizes = vec![0usize]; + for base in [SCAN_CHUNK_BYTES - 1, SCAN_CHUNK_BYTES, SCAN_CHUNK_BYTES + 1] { + sizes.extend([base, base * 2, base * 2 + 1]); + } + for size in sizes { + let bytes: Vec = (0..size).map(|i| (i % 251) as u8).collect(); + fs::write(&path, &bytes).unwrap(); + let scanned = scan_file(&path).unwrap(); + let expected: [u8; 32] = Sha256::digest(&bytes).into(); + assert_eq!( + scanned.hash, expected, + "digest differs from the whole-buffer hash at {size} bytes" + ); + } } #[test] diff --git a/src/review.rs b/src/review.rs index c885cd3..0be223d 100644 --- a/src/review.rs +++ b/src/review.rs @@ -144,6 +144,16 @@ fn evaluate_with_registry( let registry_base = ctx.bases.for_ecosystem(ecosystem).to_string(); let target_pkg = registry.resolve(name, version_str)?; + // Started here because the answer is needed only at scoring time, and + // everything between this line and there is work the query does not touch. + let advisories_resolved = crate::advisory::resolve_advisories( + &target_pkg.name, + &target_pkg.version, + ecosystem, + Some(store), + policy, + ); + let target_tarball = ctx.fetch_tarball(registry, &target_pkg)?; let checksum = target_pkg.integrity.clone().ok_or_else(|| { @@ -275,40 +285,39 @@ fn evaluate_with_registry( // came out the same as a clean advisory pass. It becomes a finding below. // // The `Ok` arm needed the same treatment for the case where policy does NOT - // fail closed. `fetch_advisories` returns `Ok(unverified(..))` when the - // lookup could not complete but the operator asked for the review to carry - // on -- and that arm set no error at all, so a network failure and a clean - // advisory pass reached the verdict identically. Nothing downstream could - // tell them apart: the only reader of the status was a colour label on the - // interactive card, and the CI text and markdown summaries, the JSON verdict - // and every MCP response reported the same thing for both. A lockfile diff - // adding one already-baselined package therefore printed `Status: PASSED` - // and exited 0 with the advisory host down. + // fail closed. An unverified report returned when the lookup could not + // complete but the operator asked for the review to carry on set no error + // at all, so a network failure and a clean advisory pass reached the + // verdict identically. A lockfile diff adding one already-baselined + // package therefore printed `Status: PASSED` and exited 0 with the advisory + // host down. // // The two cases are not the same severity. An `Err` is a real failure and // stays HIGH. An `unverified` report the operator has already accepted is a // coverage hole worth saying out loud, not worth blocking on -- so it is a - // MEDIUM disclosure that escalates for anyone running a stricter - // `fail_on`. Advisory checking switched off in policy is the operator's - // deliberate choice and is not a coverage hole, so it stays silent. - let (advisories, advisory_error, advisory_coverage_unknown) = - match crate::advisory::fetch_advisories( - &target_pkg.name, - &target_pkg.version, - ecosystem, - Some(store), - policy, - ) { - Ok(report) => { - let unknown = crate::advisory::coverage_unknown(&report, policy); - (report, None, unknown) - } - Err(e) => ( - crate::advisory::AdvisoryReport::unverified(&e.to_string()), - Some(e.to_string()), - None, - ), - }; + // MEDIUM disclosure that escalates for anyone running a stricter `fail_on`. + // Advisory checking switched off in policy is the operator's deliberate + // choice and is not a coverage hole, so it stays silent. + // + // The join is where the pending OSV round trip is collected. It was + // started before the tarballs were fetched precisely so the fetch and + // extraction ran while the network worked. + let advisories_joined = match advisories_resolved { + Ok(crate::advisory::AdvisoryResolution::Ready(report)) => Ok(report), + Ok(crate::advisory::AdvisoryResolution::Pending(pending)) => pending.join(Some(store)), + Err(e) => Err(e), + }; + let (advisories, advisory_error, advisory_coverage_unknown) = match advisories_joined { + Ok(report) => { + let unknown = crate::advisory::coverage_unknown(&report, policy); + (report, None, unknown) + } + Err(e) => ( + crate::advisory::AdvisoryReport::unverified(&e.to_string()), + Some(e.to_string()), + None, + ), + }; let provenance = match ecosystem { Ecosystem::Npm => Some(crate::provenance::inspect_provenance( diff --git a/src/store.rs b/src/store.rs index aa2fe3a..0792002 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; use rusqlite_migration::{M, Migrations}; @@ -298,6 +298,7 @@ pub struct CachedProvenance { /// blesses a release. pub struct BaselineStore { conn: rusqlite::Connection, + path: PathBuf, } /// How many times the journal-mode change is retried before giving up. @@ -464,7 +465,22 @@ impl BaselineStore { }); } - Ok(Self { conn }) + Ok(Self { + conn, + path: path.to_path_buf(), + }) + } + + /// A second handle to the same database file. + /// + /// The connection is opened in WAL mode with a busy timeout, which is what + /// makes concurrent access safe, so a caller that needs to work in parallel + /// gives each worker its own handle instead of sharing one: a + /// `rusqlite::Connection` is `Send` but not `Sync`, and wrapping it in a + /// mutex would serialise the whole review rather than just the database. + /// Nothing about what is stored changes; only who holds the connection. + pub fn reopen(&self) -> Result { + Self::open_at(&self.path) } /// Record that `name@version` was integrity-verified. This is evidence, @@ -899,7 +915,7 @@ impl BaselineStore { let mut stmt = self .conn .prepare( - "SELECT ecosystem, package, version, integrity, action, score, verdict, + "SELECT id, ecosystem, package, version, integrity, action, score, verdict, decided_by, notes, decided_at FROM audit_log ORDER BY id DESC LIMIT ?1", ) @@ -907,16 +923,17 @@ impl BaselineStore { let rows = stmt .query_map(rusqlite::params![limit as i64], |row| { Ok(AuditEntry { - ecosystem: row.get(0)?, - package: row.get(1)?, - version: row.get(2)?, - integrity: row.get(3)?, - action: row.get(4)?, - score: row.get::<_, i64>(5)? as u32, - verdict: row.get(6)?, - decided_by: row.get(7)?, - notes: row.get(8)?, - decided_at: row.get(9)?, + id: row.get(0)?, + ecosystem: row.get(1)?, + package: row.get(2)?, + version: row.get(3)?, + integrity: row.get(4)?, + action: row.get(5)?, + score: row.get::<_, i64>(6)? as u32, + verdict: row.get(7)?, + decided_by: row.get(8)?, + notes: row.get(9)?, + decided_at: row.get(10)?, }) }) .map_err(|e| BluelineError::Store(format!("querying audit_entries: {e}")))?; @@ -933,6 +950,7 @@ impl BaselineStore { /// One audit-trail row, as read back by the recall curation export. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AuditEntry { + pub id: i64, pub ecosystem: String, pub package: String, pub version: String,