diff --git a/CHANGELOG.md b/CHANGELOG.md index a5ee2e5..b598422 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.3.0 - 2026-09-13 + +- Snapshot authenticated SQLite bytes into a private file before queries and sign + sealed receipt bytes, removing mutable-path time-of-check/time-of-use windows. +- Require consumer-supplied reviewer trust roots during signed verification and + activation; support retired reviewer keys with OpenSSH validity epochs. +- Hash the exact source descriptor stream consumed by parsers and add configurable + expanded-byte, record, database-growth, query-output and diff-output limits. +- Bind identity decisions to entity metadata, redact descriptions from typed diffs, + include diff attribution and keep CrUX job IDs stable across output-only limits. +- Migrate writer stores to schema/rules version 3 while allowing activation over + pinned version 1 and version 2 generations for revocation checks. + ## 0.2.0 - 2026-09-13 - Open generations from an authenticated immutable SQLite file descriptor and diff --git a/Cargo.lock b/Cargo.lock index 5ce0190..5533a49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -78,14 +78,14 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "argand-atomic" -version = "0.2.0" +version = "0.3.0" dependencies = [ "tempfile", ] [[package]] name = "argand-site-registry" -version = "0.2.0" +version = "0.3.0" dependencies = [ "anyhow", "argand-atomic", diff --git a/Cargo.toml b/Cargo.toml index c411b22..67416bb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ resolver = "3" members = ["crates/argand-atomic", "crates/argand-site-registry"] [workspace.package] -version = "0.2.0" +version = "0.3.0" authors = ["Nic Weyand"] edition = "2024" license = "AGPL-3.0-or-later" diff --git a/README.md b/README.md index 153c094..0b9039c 100644 --- a/README.md +++ b/README.md @@ -121,8 +121,8 @@ have no dataset, review, signing or activation authority. The code remains **AGPL-3.0-or-later**; the complete license is in [LICENSE](LICENSE). Original attribution is retained. [UPSTREAM.json](UPSTREAM.json) records the signed -Argand extraction revision and original file hashes. Version 0.2 advances the -standalone database to schema version 2; the migration and consumer rules are +Argand extraction revision and original file hashes. Version 0.3 advances the +standalone database to schema version 3; the migration and consumer rules are documented in [CONSUMERS.md](docs/CONSUMERS.md). Argand still uses its embedded copy; its switch to a pinned upstream release is a coordinated downstream change. This repository does not operate a public approved-link dataset. diff --git a/crates/argand-site-registry/README.md b/crates/argand-site-registry/README.md index 25d1492..db345e9 100644 --- a/crates/argand-site-registry/README.md +++ b/crates/argand-site-registry/README.md @@ -271,7 +271,9 @@ argand-site-registry sign --generation "$ARGAND_SITE_DATA/generation-2" \ --allowed-reviewers /secure/reviewer-allowed-signers argand-site-registry activate --generation "$ARGAND_SITE_DATA/generation-2" \ --current "$ARGAND_SITE_DATA/current.json" \ - --allowed-signers /secure/registry-allowed-signers --identity registry-publisher + --allowed-signers /secure/registry-allowed-signers \ + --allowed-reviewers /secure/reviewer-allowed-signers \ + --identity registry-publisher ``` Before signing, the CLI replays every stored reviewer signature against the supplied @@ -280,7 +282,8 @@ operator-controlled SSH release key. The external release allowed-signers file follows OpenSSH syntax: `registry-publisher ssh-ed25519 PUBLIC_KEY`. Neither keys nor the trust file should come from the downloaded dataset. Consumers can open `Registry::open(path, trusted_receipt_sha256)` once and reuse its indexed -queries, or verify a publisher with `release::verify_signed` first. A hash proves +queries, or verify both a publisher and the retained reviewer proofs with +`release::verify_signed` first. A hash proves integrity only relative to a trusted pin. Signature verification authenticates the publisher, not the truth of a source assertion. @@ -313,7 +316,7 @@ downloads. No scheduled job signs, approves, renews approvals or activates links ## Storage and operating limits -Migrations `migrations/001.sql` and `002.sql` own schema version 2. `sources`, `records` and +Migrations `migrations/001.sql`, `002.sql` and `003.sql` own schema version 3. `sources`, `records` and `facts` preserve snapshot/native IDs, licenses, retrieval times and confidence; `reviews` and their cryptographic `review_auth` proofs are append-only. Complete source selection is latest retrieval time per provider/scope, with digest as the deterministic tie break. Use the **same scope** @@ -331,21 +334,28 @@ join. The canonical label rule prefers labels, then English, then language/text order. Original labels/aliases are kept. Popularity has its own source, target, observation period and audience scope and never creates an ownership edge. All derivations bind input fact IDs, PSL identity and the -`argand.site-rules/v2` contract through their generation receipt. A v1 writer store +`argand.site-rules/v3` contract through their generation receipt. A v1 or v2 writer store migrates in place while retaining review history. Any legacy unauthenticated -decision makes release signing fail closed; start a reviewed v2 store from the +decision makes release signing fail closed; start a reviewed v3 store from the pinned source inputs rather than deleting historical decisions. Imports use transactions of 256 relevant records with durable replay checkpoints. -Restart replays the compressed stream and skips committed records. Large source -records are capped at 16 MiB; SQLite has an 8 MiB page cache and disk-backed sorts. +Restart replays the compressed stream and skips committed records. The exact open +descriptor stream is hashed, and any integrity or resource failure removes all +partially committed rows for that source. Large source records are capped at 16 MiB; +imports also have finite expanded-byte, record and database-growth ceilings. Override +them with `import --maximum-expanded-bytes`, `--maximum-records` and +`--maximum-database-growth-bytes` when a reviewed source requires different bounds. +SQLite has an 8 MiB page cache and disk-backed sorts. Builds stream a canonical sorted copy and never load the full registry into RAM. Names and entity metadata exposed by lookup are capped at 256 facts each, with uncapped totals; the complete assertions remain available in the database/export. -Lookup returns at most 100 edges and reports all pre-limit ambiguity counts. +Lookup returns at most 100 edges, reports all pre-limit ambiguity counts and caps +aggregate serialized candidate data at 64 MiB. Typed diff events are capped at +16 MiB and the full stream at 1 GiB; descriptions are redacted and the header +contains source attribution. The full store/history and each generation consume disk; there is no automatic -pruning. The original compressed source is hashed before/after import, so expect -extra sequential disk reads. These bounds are not a full-dump throughput claim. +pruning. These bounds are not a full-dump throughput claim. The `observation` module validates and deterministically normalizes future crawler evidence for redirects, canonical links, hreflang, JSON-LD sameAs, sitemaps and diff --git a/crates/argand-site-registry/migrations/003.sql b/crates/argand-site-registry/migrations/003.sql new file mode 100644 index 0000000..62d4f8a --- /dev/null +++ b/crates/argand-site-registry/migrations/003.sql @@ -0,0 +1,3 @@ +-- By Nic Weyand! Bind identity decisions to all displayed evidence fields. +UPDATE registry_metadata SET rules='argand.site-rules/v3' WHERE singleton=1; +PRAGMA user_version=3; diff --git a/crates/argand-site-registry/src/adapters/curlie.rs b/crates/argand-site-registry/src/adapters/curlie.rs index d2ac167..70b9494 100644 --- a/crates/argand-site-registry/src/adapters/curlie.rs +++ b/crates/argand-site-registry/src/adapters/curlie.rs @@ -14,7 +14,9 @@ pub(super) struct Curlie; impl SourceAdapter for Curlie { fn ingest(&self, input: &mut dyn BufRead, sink: &mut dyn RecordSink) -> anyhow::Result<()> { - let gzip = flate2::read::MultiGzDecoder::new(input); + const MAXIMUM_EXPANDED_ARCHIVE_BYTES: u64 = 1024 * 1024 * 1024 * 1024; + let gzip = + flate2::read::MultiGzDecoder::new(input).take(MAXIMUM_EXPANDED_ARCHIVE_BYTES + 1); let mut archive = tar::Archive::new(gzip); let mut entries = 0_u64; for entry in archive.entries()? { @@ -97,6 +99,7 @@ impl SourceAdapter for Curlie { "unexpected archive trailing data" ); } + ensure!(gzip.limit() > 0, "Curlie archive expansion cap exceeded"); ensure!(entries > 0, "Curlie archive has no entries"); Ok(()) } diff --git a/crates/argand-site-registry/src/catalog.rs b/crates/argand-site-registry/src/catalog.rs index b0fbab3..184b986 100644 --- a/crates/argand-site-registry/src/catalog.rs +++ b/crates/argand-site-registry/src/catalog.rs @@ -315,15 +315,18 @@ impl Registry { }, )?; let mut observations = Vec::new(); + let mut output_bytes = 0; for row in rows { let (fact, source, target, value, derived) = row?; - observations.push(PopularityObservation { + let observation = PopularityObservation { source, target, value: serde_json::from_str(&value)?, domain: serde_json::from_str(&derived)?, provenance: evidence::fact(&self.db, &fact)?, - }); + }; + crate::query::account_output(&mut output_bytes, &observation)?; + observations.push(observation); } Ok(PopularityLookup { input: input.into(), @@ -349,12 +352,14 @@ impl Registry { ); let mut statement = self.db.prepare("SELECT f.id FROM facts f JOIN selected_sources s ON s.id=f.source_id WHERE f.predicate='category' AND json_extract(f.value,'$.category_id')=?1 ORDER BY f.id")?; let mut metadata = Vec::new(); + let mut output_bytes = 0; for id in statement.query_map([category_id], |row| row.get::<_, String>(0))? { let mut fact = evidence::fact(&self.db, &id?)?; if let Some(value) = fact["value"].as_object_mut() { value.remove("description"); value.insert("description_redacted".into(), json!(true)); } + crate::query::account_output(&mut output_bytes, &fact)?; metadata.push(fact); } let total_members = self.db.query_row("SELECT count(*) FROM edges e WHERE e.entity IN(SELECT f.subject FROM facts f JOIN selected_sources s ON s.id=f.source_id WHERE f.predicate='category_membership' AND json_extract(f.value,'$.category_id')=?1)",[category_id],|row|crate::store::unsigned(row,0))?; @@ -384,10 +389,14 @@ impl Registry { let fingerprints = statement .query_map(params, |row| row.get::<_, String>(0))? .collect::, _>>()?; - fingerprints - .iter() - .map(|fingerprint| self.candidate(fingerprint)) - .collect() + let mut candidates = Vec::new(); + let mut output_bytes = 0; + for fingerprint in fingerprints { + let candidate = self.candidate(&fingerprint)?; + crate::query::account_output(&mut output_bytes, &candidate)?; + candidates.push(candidate); + } + Ok(candidates) } fn property_edges( diff --git a/crates/argand-site-registry/src/cli.rs b/crates/argand-site-registry/src/cli.rs index a585637..cea8e90 100644 --- a/crates/argand-site-registry/src/cli.rs +++ b/crates/argand-site-registry/src/cli.rs @@ -98,6 +98,12 @@ enum Command { input: PathBuf, #[arg(long)] manifest: PathBuf, + #[arg(long)] + maximum_expanded_bytes: Option, + #[arg(long)] + maximum_records: Option, + #[arg(long)] + maximum_database_growth_bytes: Option, }, /// Build a new immutable generation; output must not exist. Build { @@ -260,6 +266,8 @@ enum Command { #[arg(long)] allowed_signers: PathBuf, #[arg(long)] + allowed_reviewers: PathBuf, + #[arg(long)] identity: String, }, /// Run configured acquisition/import/build; never approves or activates. @@ -269,6 +277,7 @@ enum Command { }, } +#[allow(clippy::too_many_lines)] // One exhaustive command dispatch table is easier to audit. pub(super) async fn run() -> anyhow::Result<()> { let value = match Args::parse().command { command @ (Command::Download { .. } @@ -287,9 +296,17 @@ pub(super) async fn run() -> anyhow::Result<()> { database, input, manifest, - } => { - serde_json::json!({"source_id":registry::store::import(&mut registry::store::open(&database)?,®istry::read_json(&manifest)?,&input)?}) - } + maximum_expanded_bytes, + maximum_records, + maximum_database_growth_bytes, + } => import_source( + &database, + &input, + &manifest, + maximum_expanded_bytes, + maximum_records, + maximum_database_growth_bytes, + )?, Command::Build { database, output } => { ensure!(database.is_file(), "import database does not exist"); let receipt = registry::build::build(®istry::store::open(&database)?, &output)?; @@ -303,15 +320,15 @@ pub(super) async fn run() -> anyhow::Result<()> { signature, allowed_reviewers, identity, - } => { - let (review, authentication) = registry::review::authenticate( - &decision, - &signature, - &allowed_reviewers, - &identity, - )?; - serde_json::json!({"review_sequence":registry::review::record_authenticated(®istry::store::open(&database)?,&Registry::open(&generation,&pin)?,&review,&authentication)?,"authenticated_reviewer":identity,"rebuild_required":true}) - } + } => record_review( + &database, + &generation, + &pin, + &decision, + &signature, + &allowed_reviewers, + &identity, + )?, Command::Export { generation, pin, @@ -354,11 +371,15 @@ pub(super) async fn run() -> anyhow::Result<()> { generation, current, allowed_signers, + allowed_reviewers, identity, - } => { - registry::release::activate(&generation, ¤t, &allowed_signers, &identity)?; - serde_json::json!({"current":current}) - } + } => activate_release( + &generation, + ¤t, + &allowed_signers, + &allowed_reviewers, + &identity, + )?, Command::Update { config } => { let config = read_config(&config)?; serde_json::json!({"candidate":registry::update::run(&config).await?}) @@ -367,6 +388,74 @@ pub(super) async fn run() -> anyhow::Result<()> { write_json(&value) } +fn record_review( + database: &std::path::Path, + generation: &std::path::Path, + pin: &str, + decision: &std::path::Path, + signature: &std::path::Path, + allowed_reviewers: &std::path::Path, + identity: &str, +) -> anyhow::Result { + let (review, authentication) = + registry::review::authenticate(decision, signature, allowed_reviewers, identity)?; + let sequence = registry::review::record_authenticated( + ®istry::store::open(database)?, + &Registry::open(generation, pin)?, + &review, + &authentication, + )?; + Ok( + serde_json::json!({"review_sequence":sequence,"authenticated_reviewer":identity,"rebuild_required":true}), + ) +} + +fn activate_release( + generation: &std::path::Path, + current: &std::path::Path, + allowed_signers: &std::path::Path, + allowed_reviewers: &std::path::Path, + identity: &str, +) -> anyhow::Result { + registry::release::activate( + generation, + current, + allowed_signers, + identity, + allowed_reviewers, + )?; + Ok(serde_json::json!({"current":current})) +} + +fn import_source( + database: &std::path::Path, + input: &std::path::Path, + manifest_path: &std::path::Path, + maximum_expanded_bytes: Option, + maximum_records: Option, + maximum_database_growth_bytes: Option, +) -> anyhow::Result { + let manifest = registry::read_json(manifest_path)?; + let mut limits = registry::store::ImportLimits::for_manifest(&manifest); + if let Some(value) = maximum_expanded_bytes { + limits.maximum_expanded_bytes = value; + } + if let Some(value) = maximum_records { + limits.maximum_records = value; + } + if let Some(value) = maximum_database_growth_bytes { + limits.maximum_database_growth_bytes = value; + } + Ok( + serde_json::json!({"source_id":registry::store::import_with_limits( + &mut registry::store::open(database)?, + &manifest, + input, + limits, + )?}), + ) +} + fn write_json(value: &serde_json::Value) -> anyhow::Result<()> { writeln!( std::io::stdout().lock(), diff --git a/crates/argand-site-registry/src/crux.rs b/crates/argand-site-registry/src/crux.rs index 00248ed..5990633 100644 --- a/crates/argand-site-registry/src/crux.rs +++ b/crates/argand-site-registry/src/crux.rs @@ -41,7 +41,7 @@ pub struct CruxDownload { /// Returns authentication, billing, schema, size, expiry, or transport failures. pub async fn download(cache: &Path, request: &CruxDownload) -> anyhow::Result { let query = query(request)?; - let key = crate::digest(&serde_json::to_vec(request)?); + let key = job_key(request)?; let root = cache.join("crux").join(&key); fs::create_dir_all(&root)?; let lock = OpenOptions::new() // atomic-writes: allow advisory lock inode must remain stable @@ -175,13 +175,14 @@ async fn download_pages( job_id: &str, part: &Path, ) -> anyhow::Result<()> { + const HEADER: &[u8] = b"origin,rank,yyyymm,country_code\n"; let result_url = format!( "https://bigquery.googleapis.com/bigquery/v2/projects/{}/queries/{job_id}", request.project ); let mut output = File::create(part)?; // atomic-writes: allow unpublished replay file, synced and renamed before manifest publication - output.write_all(b"origin,rank,yyyymm,country_code\n")?; - let mut written = 29_u64; + output.write_all(HEADER)?; + let mut written = u64::try_from(HEADER.len())?; let mut count = 0_u64; let mut page_token = String::new(); let mut pending = 0_u32; @@ -280,6 +281,15 @@ fn query(request: &CruxDownload) -> anyhow::Result { )) } +fn job_key(request: &CruxDownload) -> anyhow::Result { + Ok(crate::digest(&serde_json::to_vec(&( + &request.project, + &request.month, + &request.country, + request.maximum_bytes_billed, + ))?)) +} + async fn bounded_json(mut response: reqwest::Response) -> anyhow::Result { ensure!( response.status().is_success(), @@ -338,6 +348,11 @@ mod tests { maximum_output_bytes: 1_000_000, }; let sql = query(&request)?; + let original_job = job_key(&request)?; + request.maximum_output_bytes += 1; + assert_eq!(job_key(&request)?, original_job); + request.maximum_bytes_billed += 1; + assert_ne!(job_key(&request)?, original_job); assert!(sql.contains("`chrome-ux-report.country_gb.202608`")); assert!(sql.contains("experimental.popularity.rank")); request.maximum_bytes_billed = 0; diff --git a/crates/argand-site-registry/src/diff.rs b/crates/argand-site-registry/src/diff.rs index 3b68ba1..0a8a1dc 100644 --- a/crates/argand-site-registry/src/diff.rs +++ b/crates/argand-site-registry/src/diff.rs @@ -6,6 +6,35 @@ use rusqlite::{Connection, OptionalExtension}; use serde_json::{Value, json}; use std::io::Write; +const MAXIMUM_EVENT_BYTES: usize = 16 * 1024 * 1024; +const MAXIMUM_OUTPUT_BYTES: u64 = 1024 * 1024 * 1024; + +struct Output<'a> { + inner: &'a mut dyn Write, + written: u64, +} + +impl Output<'_> { + fn line(&mut self, value: &Value) -> anyhow::Result<()> { + let bytes = serde_json::to_vec(value)?; + anyhow::ensure!( + bytes.len() <= MAXIMUM_EVENT_BYTES, + "diff event exceeds 16 MiB" + ); + self.written = self + .written + .checked_add(u64::try_from(bytes.len())? + 1) + .ok_or_else(|| anyhow::anyhow!("diff output size overflow"))?; + anyhow::ensure!( + self.written <= MAXIMUM_OUTPUT_BYTES, + "diff output exceeds 1 GiB" + ); + self.inner.write_all(&bytes)?; + self.inner.write_all(b"\n")?; + Ok(()) + } +} + struct Table { subject: &'static str, scan: &'static str, @@ -70,21 +99,17 @@ const TABLES: &[Table] = &[ /// # Errors /// Returns database, JSON, or output failures. pub fn write(old: &Registry, new: &Registry, output: &mut dyn Write) -> anyhow::Result<()> { - writeln!( - output, - "{}", - json!({"schema":"argand.site-diff/v2","type":"header","old":old.identity,"new":new.identity}) - )?; + let mut output = Output { + inner: output, + written: 0, + }; + output.line(&json!({"schema":"argand.site-diff/v3","type":"header","old":old.identity,"new":new.identity,"attribution":crate::release::attribution(),"descriptions_included":false}))?; let mut changes = 0_u64; for table in TABLES { - changes += removed_or_changed(table, &old.db, &new.db, output)?; - changes += added(table, &new.db, &old.db, output)?; + changes += removed_or_changed(table, &old.db, &new.db, &mut output)?; + changes += added(table, &new.db, &old.db, &mut output)?; } - writeln!( - output, - "{}", - json!({"schema":"argand.site-diff/v2","type":"summary","changes":changes}) - )?; + output.line(&json!({"schema":"argand.site-diff/v3","type":"summary","changes":changes}))?; Ok(()) } @@ -92,7 +117,7 @@ fn removed_or_changed( table: &Table, from: &Connection, to: &Connection, - output: &mut dyn Write, + output: &mut Output<'_>, ) -> anyhow::Result { let mut count = 0; let mut scan = from.prepare(table.scan)?; @@ -129,7 +154,7 @@ fn added( table: &Table, from: &Connection, to: &Connection, - output: &mut dyn Write, + output: &mut Output<'_>, ) -> anyhow::Result { let mut count = 0; let mut scan = from.prepare(table.scan)?; @@ -152,7 +177,7 @@ fn added( } fn event( - output: &mut dyn Write, + output: &mut Output<'_>, subject: &str, change: &str, key: &str, @@ -161,14 +186,54 @@ fn event( ) -> anyhow::Result<()> { let parse = |value: Option<&str>| -> anyhow::Result> { value - .map(serde_json::from_str) + .map(|raw| { + let mut value: Value = serde_json::from_str(raw)?; + redact_descriptions(&mut value); + Ok(value) + }) .transpose() - .map_err(Into::into) }; - writeln!( - output, - "{}", - json!({"schema":"argand.site-diff/v2","type":"change","subject":subject,"change":change,"key":key,"before":parse(before)?,"after":parse(after)?}) - )?; + output.line(&json!({"schema":"argand.site-diff/v3","type":"change","subject":subject,"change":change,"key":key,"before":parse(before)?,"after":parse(after)?}))?; Ok(()) } + +fn redact_descriptions(value: &mut Value) { + match value { + Value::Object(object) => { + if object.get("predicate").and_then(Value::as_str) == Some("description") { + object.insert("value".into(), Value::Null); + object.insert("description_redacted".into(), Value::Bool(true)); + } + if object.remove("description").is_some() { + object.insert("description_redacted".into(), Value::Bool(true)); + } + for child in object.values_mut() { + redact_descriptions(child); + } + } + Value::Array(values) => { + for child in values { + redact_descriptions(child); + } + } + _ => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn redacts_description_facts_and_nested_fields() { + let mut value = json!({ + "predicate": "description", + "value": {"text": "secret"}, + "nested": {"description": "also secret"} + }); + redact_descriptions(&mut value); + assert!(value["value"].is_null()); + assert_eq!(value["description_redacted"], true); + assert!(value["nested"].get("description").is_none()); + } +} diff --git a/crates/argand-site-registry/src/generation.rs b/crates/argand-site-registry/src/generation.rs index 58c3b14..ee6682f 100644 --- a/crates/argand-site-registry/src/generation.rs +++ b/crates/argand-site-registry/src/generation.rs @@ -13,7 +13,7 @@ use std::{ }; #[cfg(target_os = "linux")] -use std::os::{fd::AsRawFd, unix::fs::OpenOptionsExt}; +use std::os::unix::fs::OpenOptionsExt; impl Registry { /// Opens only a complete, externally pinned generation. @@ -21,6 +21,18 @@ impl Registry { /// # Errors /// Rejects altered receipts/databases and unsupported contracts. pub fn open(path: &Path, expected_pin: &str) -> anyhow::Result { + Self::open_with_rules(path, expected_pin, false) + } + + pub(crate) fn open_previous(path: &Path, expected_pin: &str) -> anyhow::Result { + Self::open_with_rules(path, expected_pin, true) + } + + fn open_with_rules( + path: &Path, + expected_pin: &str, + allow_legacy_rules: bool, + ) -> anyhow::Result { ensure!( crate::model::valid_digest(expected_pin), "provide a full trusted receipt SHA-256" @@ -39,9 +51,10 @@ impl Registry { && crate::digest(&attribution) == receipt.attribution_sha256, "registry license or attribution digest mismatch" ); + let rules_supported = crate::store::supported_rule_version(&receipt.rules) + || (allow_legacy_rules && crate::store::legacy_rule_version(&receipt.rules)); ensure!( - receipt.schema == "argand.site-registry/v1" - && crate::store::supported_rule_version(&receipt.rules), + receipt.schema == "argand.site-registry/v1" && rules_supported, "unsupported registry contract" ); let database = path.join("registry.sqlite"); @@ -127,27 +140,38 @@ pub(crate) fn open_no_follow(_path: &Path) -> anyhow::Result { #[cfg(target_os = "linux")] fn open_authenticated_database(path: &Path, expected: &str) -> anyhow::Result<(Connection, File)> { - let mut file = open_no_follow(path)?; + let mut source = open_no_follow(path)?; ensure!( - file.metadata()?.is_file(), + source.metadata()?.is_file(), "database must be a regular file" ); + // SQLite's immutable mode trusts that its backing inode cannot change. Copy + // the authenticated bytes into a mode-0700 private directory, open them, + // then unlink the directory so no writer can alter the verified snapshot. + let directory = tempfile::tempdir()?; + let snapshot = directory.path().join("registry.sqlite"); + let mut file = OpenOptions::new() // atomic-writes: allow private create-new verified snapshot + .read(true) + .write(true) + .create_new(true) + .mode(0o600) + .open(&snapshot)?; let mut hash = Sha256::new(); let mut buffer = [0; 8192]; loop { - let count = file.read(&mut buffer)?; + let count = source.read(&mut buffer)?; if count == 0 { break; } hash.update(&buffer[..count]); + std::io::Write::write_all(&mut file, &buffer[..count])?; } ensure!( format!("{:x}", hash.finalize()) == expected, "registry database digest mismatch" ); file.rewind()?; - let descriptor = format!("/proc/self/fd/{}", file.as_raw_fd()); - let mut uri = url::Url::from_file_path(descriptor) + let mut uri = url::Url::from_file_path(&snapshot) .map_err(|()| anyhow::anyhow!("cannot construct immutable SQLite URI"))?; uri.query_pairs_mut() .append_pair("mode", "ro") @@ -156,6 +180,7 @@ fn open_authenticated_database(path: &Path, expected: &str) -> anyhow::Result<(C | rusqlite::OpenFlags::SQLITE_OPEN_URI | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX; let db = Connection::open_with_flags(uri.as_str(), flags)?; + drop(directory); Ok((db, file)) } diff --git a/crates/argand-site-registry/src/identity.rs b/crates/argand-site-registry/src/identity.rs index 31de350..542442d 100644 --- a/crates/argand-site-registry/src/identity.rs +++ b/crates/argand-site-registry/src/identity.rs @@ -51,6 +51,7 @@ pub fn propose(registry: &Registry, left: &str, right: &str) -> anyhow::Result(used: &mut usize, value: &T) -> anyhow::Result<()> { + *used = used + .checked_add(serde_json::to_vec(value)?.len()) + .ok_or_else(|| anyhow::anyhow!("query output size overflow"))?; + ensure!( + *used <= MAXIMUM_QUERY_OUTPUT_BYTES, + "query output exceeds 64 MiB" + ); + Ok(()) +} + pub use crate::resolution::{Resolution, ResolutionCounts, ResolutionStatus}; /// Open verified generation. Hashes are checked once, outside the query path. pub struct Registry { pub(crate) db: Connection, - // Keep the authenticated inode alive for the SQLite /proc/self/fd reader. + // Keep the private authenticated snapshot alive with the SQLite reader. pub(crate) _database: File, /// External receipt pin supplied by the caller. pub identity: String, @@ -162,10 +175,13 @@ impl Registry { let ids = stmt .query_map(params![key, limit], |r| r.get::<_, String>(0))? .collect::, _>>()?; - let candidates = ids - .iter() - .map(|id| self.candidate(id)) - .collect::>>()?; + let mut candidates = Vec::new(); + let mut output_bytes = 0; + for id in ids { + let candidate = self.candidate(&id)?; + account_output(&mut output_bytes, &candidate)?; + candidates.push(candidate); + } Ok(Lookup { query: key, total_entities, @@ -192,7 +208,7 @@ impl Registry { let signer:Option=r.get(10)?; Ok(json!({"sequence":sequence,"decision":r.get::<_,String>(1)?,"reviewer":r.get::<_,String>(2)?,"reason":r.get::<_,String>(3)?,"evidence":r.get::<_,String>(4)?,"retrieved_at":r.get::<_,String>(5)?,"expires_at":r.get::<_,String>(6)?,"role":r.get::<_,String>(7)?,"locale":r.get::<_,String>(8)?,"country":r.get::<_,String>(9)?,"authentication":signer.map(|identity|json!({"identity":identity,"signature_sha256":r.get::<_,String>(11).unwrap_or_default(),"namespace":r.get::<_,String>(12).unwrap_or_default()})),"source":"argand_operator_review","source_identifier":format!("{}:{sequence}",fingerprint),"license":"CC0-1.0","license_url":"https://creativecommons.org/publicdomain/zero/1.0/","confidence":9000})) }).optional()?; - Ok(Candidate { + let candidate = Candidate { identity_provenance: Vec::new(), entity: crate::evidence::entity(&self.db, &entity, &name)?, property_scopes: crate::evidence::scopes( @@ -217,6 +233,9 @@ impl Registry { provenance, review, eligible, - }) + }; + let mut output_bytes = 0; + account_output(&mut output_bytes, &candidate)?; + Ok(candidate) } } diff --git a/crates/argand-site-registry/src/release.rs b/crates/argand-site-registry/src/release.rs index 62400b4..73120ef 100644 --- a/crates/argand-site-registry/src/release.rs +++ b/crates/argand-site-registry/src/release.rs @@ -8,7 +8,6 @@ use std::{ fs::{self, File, OpenOptions}, io::{BufWriter, Write}, path::Path, - process::Command, }; /// Source terms shipped and authenticated with every generation. @@ -86,19 +85,16 @@ pub fn sign( pin: &str, allowed_reviewers: &Path, ) -> anyhow::Result<()> { + let receipt = crate::ssh::sealed_input(&generation.join("COMPLETE.json"), 1024 * 1024)?; + ensure!(crate::digest(&receipt.bytes) == pin, "receipt pin mismatch"); let registry = Registry::open(generation, pin)?; crate::review::verify_all(®istry.db, allowed_reviewers)?; - ensure!( - !generation.join("COMPLETE.json.sig").exists(), - "signature already exists" - ); - let status = Command::new("ssh-keygen") - .args(["-Y", "sign", "-n", "argand-site-registry", "-f"]) - .arg(key) - .arg(generation.join("COMPLETE.json")) - .status()?; - ensure!(status.success(), "SSH signing failed"); - File::open(generation.join("COMPLETE.json.sig"))?.sync_all()?; + crate::ssh::sign( + "argand-site-registry", + &receipt.bytes, + key, + &generation.join("COMPLETE.json.sig"), + )?; File::open(generation)?.sync_all()?; Ok(()) } @@ -111,6 +107,7 @@ pub fn verify_signed( generation: &Path, signers: &Path, identity: &str, + allowed_reviewers: &Path, ) -> anyhow::Result { let receipt = crate::ssh::sealed_input(&generation.join("COMPLETE.json"), 1024 * 1024)?; let signature = crate::ssh::sealed_input(&generation.join("COMPLETE.json.sig"), 64 * 1024)?; @@ -124,7 +121,7 @@ pub fn verify_signed( identity, )?; let registry = Registry::open(generation, &pin)?; - crate::review::ensure_all_authenticated(®istry.db)?; + crate::review::verify_all(®istry.db, allowed_reviewers)?; Ok(registry) } @@ -138,8 +135,9 @@ pub fn activate( current: &Path, signers: &Path, identity: &str, + allowed_reviewers: &Path, ) -> anyhow::Result<()> { - let registry = verify_signed(generation, signers, identity)?; + let registry = verify_signed(generation, signers, identity, allowed_reviewers)?; let parent = current .parent() .context("current pointer needs a parent directory")?; @@ -165,7 +163,7 @@ pub fn activate( .context("invalid current pointer")?, "rollback would discard revocations; rebuild using the current review log" ); - let old = Registry::open( + let old = Registry::open_previous( Path::new( previous["generation"] .as_str() diff --git a/crates/argand-site-registry/src/review.rs b/crates/argand-site-registry/src/review.rs index 8c7823a..6c512b0 100644 --- a/crates/argand-site-registry/src/review.rs +++ b/crates/argand-site-registry/src/review.rs @@ -259,10 +259,6 @@ pub fn verify_all(db: &Connection, allowed_reviewers: &Path) -> anyhow::Result<( verify_rows(db, Some(&allowed.bytes)) } -pub(crate) fn ensure_all_authenticated(db: &Connection) -> anyhow::Result<()> { - verify_rows(db, None) -} - fn verify_rows(db: &Connection, allowed_reviewers: Option<&[u8]>) -> anyhow::Result<()> { let missing: u64 = db.query_row( "SELECT count(*) FROM reviews r LEFT JOIN review_auth a USING(sequence) WHERE a.sequence IS NULL", @@ -300,12 +296,13 @@ fn verify_rows(db: &Connection, allowed_reviewers: Option<&[u8]>) -> anyhow::Res }; validate_authentication(&review, &authentication)?; if let Some(allowed) = allowed_reviewers { - crate::ssh::verify( + crate::ssh::verify_at( SIGNATURE_NAMESPACE, &authentication.decision_json, &authentication.signature, allowed, &authentication.signer, + Some(review.reviewed_at), )?; } } diff --git a/crates/argand-site-registry/src/ssh.rs b/crates/argand-site-registry/src/ssh.rs index dff7bf3..df5fc93 100644 --- a/crates/argand-site-registry/src/ssh.rs +++ b/crates/argand-site-registry/src/ssh.rs @@ -35,6 +35,24 @@ pub(crate) fn verify( signature: &[u8], allowed_signers: &[u8], identity: &str, +) -> anyhow::Result<()> { + verify_at( + namespace, + message, + signature, + allowed_signers, + identity, + None, + ) +} + +pub(crate) fn verify_at( + namespace: &str, + message: &[u8], + signature: &[u8], + allowed_signers: &[u8], + identity: &str, + verify_time: Option>, ) -> anyhow::Result<()> { // OpenSSH requires paths for its signature and trust file. Private, // create-new temporary copies bind the command to the exact bytes already @@ -49,11 +67,15 @@ pub(crate) fn verify( allowed_file.write_all(allowed_signers)?; allowed_file.flush()?; - let status = Command::new("ssh-keygen") - .args(["-Y", "verify", "-n", namespace, "-f"]) - .arg(allowed_file.path()) - .arg("-I") - .arg(identity) + let mut command = Command::new("ssh-keygen"); + command.args(["-Y", "verify", "-n", namespace, "-f"]); + command.arg(allowed_file.path()).arg("-I").arg(identity); + if let Some(time) = verify_time { + command + .arg("-O") + .arg(format!("verify-time={}", time.format("%Y%m%d%H%M%SZ"))); + } + let status = command .arg("-s") .arg(signature_file.path()) .stdin(Stdio::from(message_file.reopen()?)) @@ -62,3 +84,29 @@ pub(crate) fn verify( ensure!(status.success(), "untrusted SSH signature"); Ok(()) } + +pub(crate) fn sign( + namespace: &str, + message: &[u8], + key: &Path, + output: &Path, +) -> anyhow::Result<()> { + ensure!(!output.exists(), "signature already exists"); + let directory = tempfile::tempdir()?; + let message_path = directory.path().join("message"); + let mut message_file = std::fs::OpenOptions::new() // atomic-writes: allow private create-new signing input + .write(true) + .create_new(true) + .open(&message_path)?; + message_file.write_all(message)?; + message_file.sync_all()?; + let status = Command::new("ssh-keygen") + .args(["-Y", "sign", "-n", namespace, "-f"]) + .arg(key) + .arg(&message_path) + .status()?; + ensure!(status.success(), "SSH signing failed"); + let signature = sealed_input(&message_path.with_extension("sig"), 64 * 1024)?; + argand_atomic::create_durable(output, &signature.bytes)?; + Ok(()) +} diff --git a/crates/argand-site-registry/src/store.rs b/crates/argand-site-registry/src/store.rs index 0002204..d3e340f 100644 --- a/crates/argand-site-registry/src/store.rs +++ b/crates/argand-site-registry/src/store.rs @@ -7,15 +7,17 @@ use crate::{ }; use anyhow::{Context, ensure}; use rusqlite::{Connection, OptionalExtension, params}; +use sha2::{Digest, Sha256}; use std::{ - fs::File, + cell::RefCell, io::{BufReader, Read}, path::Path, + rc::Rc, time::Duration, }; /// Adapter/normalization contract recorded in all generation identities. -pub const RULE_VERSION: &str = "argand.site-rules/v2"; +pub const RULE_VERSION: &str = "argand.site-rules/v3"; /// Whether a signed immutable generation uses a reader-compatible rule contract. #[must_use] @@ -23,6 +25,10 @@ pub fn supported_rule_version(version: &str) -> bool { version == RULE_VERSION } +pub(crate) fn legacy_rule_version(version: &str) -> bool { + matches!(version, "argand.site-rules/v1" | "argand.site-rules/v2") +} + /// Opens or migrates the local assertion store with bounded page cache. /// /// # Errors @@ -39,14 +45,21 @@ pub fn open(path: &Path) -> anyhow::Result { db.execute_batch("BEGIN IMMEDIATE")?; db.execute_batch(include_str!("../migrations/001.sql"))?; db.execute_batch(include_str!("../migrations/002.sql"))?; + db.execute_batch(include_str!("../migrations/003.sql"))?; db.execute_batch("COMMIT")?; } 1 => { db.execute_batch("BEGIN IMMEDIATE")?; db.execute_batch(include_str!("../migrations/002.sql"))?; + db.execute_batch(include_str!("../migrations/003.sql"))?; db.execute_batch("COMMIT")?; } - 2 => {} + 2 => { + db.execute_batch("BEGIN IMMEDIATE")?; + db.execute_batch(include_str!("../migrations/003.sql"))?; + db.execute_batch("COMMIT")?; + } + 3 => {} _ => anyhow::bail!("unsupported registry schema {version}"), } let rules: String = db.query_row( @@ -67,6 +80,35 @@ pub(crate) fn configure(db: &Connection) -> anyhow::Result<()> { Ok(()) } +/// Hard resource ceilings applied to one source import. +#[derive(Clone, Copy, Debug)] +pub struct ImportLimits { + /// Maximum bytes produced by the manifest-declared compression layer. + pub maximum_expanded_bytes: u64, + /// Maximum source records accepted in this snapshot. + pub maximum_records: u64, + /// Maximum logical SQLite growth caused by this snapshot. + pub maximum_database_growth_bytes: u64, +} + +impl ImportLimits { + /// Computes finite defaults scaled to the pinned compressed source size. + #[must_use] + pub fn for_manifest(manifest: &SourceManifest) -> Self { + Self { + maximum_expanded_bytes: manifest + .bytes + .saturating_mul(100) + .clamp(64 * 1024 * 1024, 1024 * 1024 * 1024 * 1024), + maximum_records: 500_000_000, + maximum_database_growth_bytes: manifest + .bytes + .saturating_mul(8) + .clamp(1024 * 1024 * 1024, 1024 * 1024 * 1024 * 1024), + } + } +} + /// Verifies a source object before and after parsing, publishing completion last. /// Incomplete transactions are rolled back; committed batches replay idempotently. /// @@ -76,9 +118,33 @@ pub fn import( db: &mut Connection, manifest: &SourceManifest, path: &Path, +) -> anyhow::Result { + import_with_limits(db, manifest, path, ImportLimits::for_manifest(manifest)) +} + +/// Imports with explicit resource ceilings, primarily for constrained operators. +/// +/// # Errors +/// Returns source integrity, resource, parser, filesystem, or database errors. +pub fn import_with_limits( + db: &mut Connection, + manifest: &SourceManifest, + path: &Path, + limits: ImportLimits, ) -> anyhow::Result { manifest.validate()?; - verify_input(manifest, path)?; + ensure!( + limits.maximum_expanded_bytes > 0 + && limits.maximum_records > 0 + && limits.maximum_database_growth_bytes > 0, + "import limits must be positive" + ); + let mut file = crate::generation::open_no_follow(path)?; + let metadata = file.metadata()?; + ensure!( + metadata.is_file() && metadata.len() == manifest.bytes, + "source length/type mismatch" + ); let id = manifest.id()?; let declaration = serde_json::to_string(manifest)?; db.execute("INSERT OR IGNORE INTO sources(id,source,scope,retrieved_at,manifest) VALUES(?1,?2,?3,?4,?5)", params![id, manifest.source.key(),manifest.scope,manifest.retrieved_at.to_rfc3339(),declaration])?; @@ -88,14 +154,39 @@ pub fn import( |r| Ok((unsigned(r, 0)?, r.get(1)?)), )?; if complete { + let mut hash = Sha256::new(); + let bytes = std::io::copy(&mut file, &mut HashWriter(&mut hash))?; + ensure!( + bytes == manifest.bytes && format!("{:x}", hash.finalize()) == manifest.sha256, + "source digest mismatch" + ); return Ok(id); } - let file = File::open(path)?; - let reader: Box = match manifest.compression { - Compression::None => Box::new(file), - Compression::Gzip => Box::new(flate2::read::MultiGzDecoder::new(file)), - Compression::Bzip2 => Box::new(bzip2::read::MultiBzDecoder::new(file)), + let observation = Rc::new(RefCell::new(StreamObservation::default())); + let observed = ObservedReader { + inner: file, + observation: Rc::clone(&observation), }; + let reader: Box = match manifest.compression { + Compression::None => Box::new(observed), + Compression::Gzip => Box::new(flate2::read::MultiGzDecoder::new(observed)), + Compression::Bzip2 => Box::new(bzip2::read::MultiBzDecoder::new(observed)), + }; + let mut reader = BufReader::new(ExpandedReader { + inner: reader, + read: 0, + maximum: limits.maximum_expanded_bytes, + }); + let database_before = database_bytes(db)?; + let page_size: u64 = db.query_row("PRAGMA page_size", [], |row| unsigned(row, 0))?; + let original_max_pages: u64 = + db.query_row("PRAGMA max_page_count", [], |row| unsigned(row, 0))?; + let import_max_pages = database_before + .saturating_add(limits.maximum_database_growth_bytes) + .checked_div(page_size) + .context("invalid SQLite page size")? + .min(original_max_pages); + db.pragma_update(None, "max_page_count", i64::try_from(import_max_pages)?)?; db.execute_batch("BEGIN IMMEDIATE")?; let result = { let mut sink = SqlSink { @@ -103,15 +194,23 @@ pub fn import( source: &id, ordinal: 0, checkpoint, + limits, + database_before, }; - let parsed = - adapters::adapter(manifest.format).ingest(&mut BufReader::new(reader), &mut sink); + let parsed = adapters::adapter(manifest.format).ingest(&mut reader, &mut sink); parsed.and_then(|()| { + std::io::copy(&mut reader, &mut std::io::sink())?; ensure!( sink.ordinal > 0 && sink.ordinal >= checkpoint, "source empty or shorter than checkpoint" ); - verify_input(manifest, path)?; + let observed = observation.borrow(); + ensure!( + observed.bytes == manifest.bytes + && format!("{:x}", observed.hash.clone().finalize()) == manifest.sha256, + "source digest mismatch" + ); + sink.ensure_database_growth()?; sink.db.execute( "UPDATE sources SET checkpoint=?2,complete=1 WHERE id=?1", params![id, i64::try_from(sink.ordinal)?], @@ -123,35 +222,57 @@ pub fn import( Ok(()) => db.execute_batch("COMMIT")?, Err(error) => { db.execute_batch("ROLLBACK")?; + discard_incomplete_source(db, &id)?; + db.pragma_update(None, "max_page_count", i64::try_from(original_max_pages)?)?; return Err(error); } } + db.pragma_update(None, "max_page_count", i64::try_from(original_max_pages)?)?; Ok(id) } -fn verify_input(manifest: &SourceManifest, path: &Path) -> anyhow::Result<()> { - let metadata = std::fs::symlink_metadata(path)?; - ensure!( - metadata.is_file() && metadata.len() == manifest.bytes, - "source length/type mismatch" - ); - ensure!( - crate::file_digest(path)? == manifest.sha256, - "source digest mismatch" - ); +fn discard_incomplete_source(db: &Connection, id: &str) -> anyhow::Result<()> { + db.execute_batch("BEGIN IMMEDIATE")?; + db.execute("DELETE FROM facts WHERE source_id=?1", [id])?; + db.execute("DELETE FROM records WHERE source_id=?1", [id])?; + db.execute("DELETE FROM sources WHERE id=?1 AND complete=0", [id])?; + db.execute_batch("COMMIT")?; Ok(()) } +fn database_bytes(db: &Connection) -> anyhow::Result { + let page_count: u64 = db.query_row("PRAGMA page_count", [], |row| unsigned(row, 0))?; + let page_size: u64 = db.query_row("PRAGMA page_size", [], |row| unsigned(row, 0))?; + Ok(page_count.saturating_mul(page_size)) +} + struct SqlSink<'a> { db: &'a Connection, source: &'a str, ordinal: u64, checkpoint: u64, + limits: ImportLimits, + database_before: u64, +} + +impl SqlSink<'_> { + fn ensure_database_growth(&self) -> anyhow::Result<()> { + ensure!( + database_bytes(self.db)?.saturating_sub(self.database_before) + <= self.limits.maximum_database_growth_bytes, + "source database growth cap exceeded" + ); + Ok(()) + } } impl RecordSink for SqlSink<'_> { fn emit(&mut self, record: Record) -> anyhow::Result<()> { self.ordinal += 1; + ensure!( + self.ordinal <= self.limits.maximum_records, + "source record cap exceeded" + ); if self.ordinal <= self.checkpoint { return Ok(()); } @@ -184,6 +305,7 @@ impl RecordSink for SqlSink<'_> { ])?; } if self.ordinal.is_multiple_of(256) { + self.ensure_database_growth()?; self.db.execute( "UPDATE sources SET checkpoint=?2 WHERE id=?1", params![self.source, i64::try_from(self.ordinal)?], @@ -194,6 +316,63 @@ impl RecordSink for SqlSink<'_> { } } +#[derive(Default)] +struct StreamObservation { + hash: Sha256, + bytes: u64, +} + +struct ObservedReader { + inner: R, + observation: Rc>, +} + +impl Read for ObservedReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let count = self.inner.read(buffer)?; + let mut observation = self.observation.borrow_mut(); + observation.hash.update(&buffer[..count]); + observation.bytes = observation.bytes.saturating_add(count as u64); + Ok(count) + } +} + +struct ExpandedReader { + inner: Box, + read: u64, + maximum: u64, +} + +struct HashWriter<'a>(&'a mut Sha256); + +impl std::io::Write for HashWriter<'_> { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + self.0.update(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl Read for ExpandedReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let remaining = self.maximum.saturating_sub(self.read); + if remaining == 0 { + let mut probe = [0_u8; 1]; + if self.inner.read(&mut probe)? != 0 { + return Err(std::io::Error::other("source expansion cap exceeded")); + } + return Ok(0); + } + let bound = usize::try_from(remaining.min(buffer.len() as u64)).unwrap_or(buffer.len()); + let count = self.inner.read(&mut buffer[..bound])?; + self.read = self.read.saturating_add(count as u64); + Ok(count) + } +} + /// Returns the manifest of a complete source snapshot. /// /// # Errors diff --git a/crates/argand-site-registry/tests/cli.rs b/crates/argand-site-registry/tests/cli.rs index cd5c4cc..2c64cca 100644 --- a/crates/argand-site-registry/tests/cli.rs +++ b/crates/argand-site-registry/tests/cli.rs @@ -337,6 +337,8 @@ fn release_lifecycle( text(&root.join("current.json"))?, "--allowed-signers", text(&root.join("allowed_signers"))?, + "--allowed-reviewers", + text(&root.join("allowed_signers"))?, "--identity", "fixture" ]) @@ -391,6 +393,24 @@ fn sign_and_activate(root: &Path, approved: &Path, approved_pin: &str) -> anyhow text(¤t)?, "--allowed-signers", text(&allowed)?, + "--allowed-reviewers", + text(&wrong_reviewers)?, + "--identity", + "fixture" + ]) + .is_err() + ); + assert!( + run(&[ + "activate", + "--generation", + text(approved)?, + "--current", + text(¤t)?, + "--allowed-signers", + text(&allowed)?, + "--allowed-reviewers", + text(&allowed)?, "--identity", "untrusted" ]) @@ -404,6 +424,8 @@ fn sign_and_activate(root: &Path, approved: &Path, approved_pin: &str) -> anyhow text(¤t)?, "--allowed-signers", text(&allowed)?, + "--allowed-reviewers", + text(&allowed)?, "--identity", "fixture", ])?; diff --git a/crates/argand-site-registry/tests/failures.rs b/crates/argand-site-registry/tests/failures.rs index cbe4972..3a86da1 100644 --- a/crates/argand-site-registry/tests/failures.rs +++ b/crates/argand-site-registry/tests/failures.rs @@ -6,7 +6,7 @@ use argand_site_registry::{ download::CachedSource, model::{Compression, Format, Source}, query::Registry, - store, + store::{self, ImportLimits}, }; use serde_json::json; use std::{fs, io::Write, process::Command}; @@ -23,7 +23,7 @@ fn version_one_store_migrates_without_losing_review_history() -> anyhow::Result< let migrated = store::open(&path)?; assert_eq!( migrated.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))?, - 2 + 3 ); assert_eq!( migrated.query_row("SELECT rules FROM registry_metadata", [], |row| row @@ -83,7 +83,8 @@ fn externally_signed_unauthenticated_reviews_are_rejected() -> anyhow::Result<() argand_site_registry::release::verify_signed( &root.path().join("generation"), &root.path().join("allowed"), - "fixture" + "fixture", + &root.path().join("allowed") ) .is_err() ); @@ -123,6 +124,168 @@ fn unsigned_sqlite_sidecars_and_generation_symlinks_are_rejected() -> anyhow::Re Ok(()) } +#[test] +fn opened_registry_uses_private_authenticated_database_bytes() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let db = common::fixture(root.path())?; + let registry = common::build(&db, root.path(), "private-snapshot")?; + let mutable = rusqlite::Connection::open(root.path().join("private-snapshot/registry.sqlite"))?; + mutable.execute( + "UPDATE properties SET url='https://changed.invalid/' WHERE url='https://facebook.com/'", + [], + )?; + assert_eq!( + registry.lookup("FB", 1)?.candidates[0].url, + "https://facebook.com/" + ); + Ok(()) +} + +#[test] +fn import_limits_rollback_all_partial_source_rows() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let mut db = store::open(&root.path().join("limited.sqlite"))?; + let compressed = common::gzip(common::CRUX.as_bytes())?; + let input = root.path().join("limited.csv.gz"); + fs::write(&input, &compressed)?; + let mut manifest = common::manifest(Source::Crux, Format::CruxCsv, &compressed)?; + manifest.compression = Compression::Gzip; + assert!( + store::import_with_limits( + &mut db, + &manifest, + &input, + ImportLimits { + maximum_expanded_bytes: 16, + maximum_records: 100, + maximum_database_growth_bytes: 1024 * 1024, + }, + ) + .is_err() + ); + assert_eq!( + db.query_row("SELECT count(*) FROM sources", [], |row| row + .get::<_, i64>(0))?, + 0 + ); + Ok(()) +} + +#[test] +fn retired_reviewer_keys_verify_history_only_with_a_validity_epoch() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let db = common::fixture(root.path())?; + let generation = common::build(&db, root.path(), "reviewer-epoch")?; + let fingerprint = generation.lookup("FB", 1)?.candidates[0] + .fingerprint + .clone(); + let key = root.path().join("reviewer"); + assert!( + Command::new("ssh-keygen") + .args(["-q", "-t", "ed25519", "-N", "", "-f"]) + .arg(&key) + .status()? + .success() + ); + let decision = root.path().join("decision.json"); + fs::write( + &decision, + serde_json::to_vec(&json!({ + "fingerprint":fingerprint,"decision":"revoke","reviewer":"retired", + "reason":"synthetic key epoch test","evidence":"synthetic:key-epoch", + "reviewed_at":"2026-09-12T12:00:00Z","expires_at":"2026-09-12T12:00:00Z", + "role":"unspecified","locale":"","country":"" + }))?, + )?; + assert!( + Command::new("ssh-keygen") + .args(["-Y", "sign", "-n", "argand-site-registry-review", "-f"]) + .arg(&key) + .arg(&decision) + .status()? + .success() + ); + let public = fs::read_to_string(key.with_extension("pub"))?; + let allowed = root.path().join("allowed"); + fs::write(&allowed, format!("retired {public}"))?; + let (review, authentication) = argand_site_registry::review::authenticate( + &decision, + &decision.with_extension("json.sig"), + &allowed, + "retired", + )?; + argand_site_registry::review::record_authenticated(&db, &generation, &review, &authentication)?; + fs::write( + &allowed, + format!("retired valid-before=\"20260913000000Z\" {public}"), + )?; + assert!(argand_site_registry::review::verify_all(&db, &allowed).is_ok()); + fs::write( + &allowed, + format!("retired valid-before=\"20260912000000Z\" {public}"), + )?; + assert!(argand_site_registry::review::verify_all(&db, &allowed).is_err()); + Ok(()) +} + +#[test] +fn activation_accepts_a_pinned_v1_previous_generation() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let db = common::fixture(root.path())?; + let old = common::build(&db, root.path(), "legacy-current")?; + drop(old); + let receipt_path = root.path().join("legacy-current/COMPLETE.json"); + let mut receipt: serde_json::Value = argand_site_registry::read_json(&receipt_path)?; + receipt["rules"] = json!("argand.site-rules/v1"); + fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt)?)?; + let old_pin = argand_site_registry::file_digest(&receipt_path)?; + let current = root.path().join("current.json"); + fs::write( + ¤t, + serde_json::to_vec_pretty(&json!({ + "schema":"argand.site-current/v1", + "generation":root.path().join("legacy-current").canonicalize()?, + "receipt_sha256":old_pin, + "signer":"legacy-publisher", + "revocation_sequence":0 + }))?, + )?; + + let new = common::build(&db, root.path(), "current-rules")?; + let key = root.path().join("publisher"); + assert!( + Command::new("ssh-keygen") + .args(["-q", "-t", "ed25519", "-N", "", "-f"]) + .arg(&key) + .status()? + .success() + ); + let allowed = root.path().join("allowed-publisher"); + fs::write( + &allowed, + format!( + "publisher {}", + fs::read_to_string(key.with_extension("pub"))? + ), + )?; + argand_site_registry::release::sign( + &root.path().join("current-rules"), + &key, + &new.identity, + &allowed, + )?; + argand_site_registry::release::activate( + &root.path().join("current-rules"), + ¤t, + &allowed, + "publisher", + &allowed, + )?; + let activated: serde_json::Value = argand_site_registry::read_json(¤t)?; + assert_eq!(activated["receipt_sha256"], new.identity); + Ok(()) +} + #[test] fn removed_entity_websites_retire_the_previous_selection() -> anyhow::Result<()> { let root = tempfile::tempdir()?; @@ -263,6 +426,8 @@ fn typed_diff_reports_review_only_changes() -> anyhow::Result<()> { assert!(rows.iter().any(|row| { row["type"] == "change" && row["subject"] == "review" && row["change"] == "added" })); + assert_eq!(rows[0]["descriptions_included"], false); + assert!(rows[0]["attribution"]["curlie"].is_object()); assert_eq!(rows.last().and_then(|row| row["changes"].as_u64()), Some(1)); Ok(()) } diff --git a/crates/argand-site-registry/tests/identity.rs b/crates/argand-site-registry/tests/identity.rs index e65f7c7..a0d8225 100644 --- a/crates/argand-site-registry/tests/identity.rs +++ b/crates/argand-site-registry/tests/identity.rs @@ -171,3 +171,28 @@ fn name_collision_remains_ambiguous_until_review_and_changes_invalidate_link() - assert!(review::record(&db, &baseline, &tampered).is_err()); Ok(()) } + +#[test] +fn metadata_changes_invalidate_identity_fingerprint() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let mut db = common::fixture(root.path())?; + let baseline = common::build(&db, root.path(), "metadata-baseline")?; + let wiki = baseline.lookup("FB", 1)?.candidates[0].entity_id.clone(); + let curlie = baseline.lookup("Facebook directory listing", 1)?.candidates[0] + .entity_id + .clone(); + let before = identity::propose(&baseline, &wiki, &curlie)?; + + let mut entity = common::entity("Q355", "Facebook", &["FB"], &["https://facebook.com/"]); + entity["claims"]["P17"] = json!([{"id":"Q355$country","rank":"normal","mainsnak":{"property":"P17","snaktype":"value","datavalue":{"type":"wikibase-entityid","value":{"id":"Q30"}}}}]); + let bytes = serde_json::to_vec(&json!({"entities":{"Q355":entity}}))?; + let mut source = common::manifest(Source::Wikidata, Format::WikidataEntities, &bytes)?; + source.retrieved_at += chrono::Duration::hours(1); + let input = root.path().join("metadata-change.json"); + std::fs::write(&input, &bytes)?; + store::import(&mut db, &source, &input)?; + let changed = common::build(&db, root.path(), "metadata-changed")?; + let after = identity::propose(&changed, &wiki, &curlie)?; + assert_ne!(before.fingerprint, after.fingerprint); + Ok(()) +} diff --git a/crates/argand-site-registry/tests/registry.rs b/crates/argand-site-registry/tests/registry.rs index e6d6757..194e7b3 100644 --- a/crates/argand-site-registry/tests/registry.rs +++ b/crates/argand-site-registry/tests/registry.rs @@ -324,11 +324,12 @@ fn compressed_dumps_resume_and_corruption_fail_closed() -> anyhow::Result<()> { m.compression = Compression::Gzip; let path = dir.path().join("dump.gz"); std::fs::write(&path, &bytes)?; - // Interrupt the sink after its first committed batch; rerun the exact input. + // A handled sink error removes committed rows whose complete stream digest + // was not established. A process crash still leaves the durable checkpoint. db.execute_batch("CREATE TRIGGER simulated_crash BEFORE INSERT ON records WHEN NEW.ordinal=270 BEGIN SELECT RAISE(ABORT,'simulated interruption'); END;")?; assert!(store::import(&mut db, &m, &path).is_err()); - let checkpoint: i64 = db.query_row("SELECT checkpoint FROM sources", [], |r| r.get(0))?; - assert_eq!(checkpoint, 256); + let partial: i64 = db.query_row("SELECT count(*) FROM sources", [], |r| r.get(0))?; + assert_eq!(partial, 0); db.execute_batch("DROP TRIGGER simulated_crash")?; store::import(&mut db, &m, &path)?; let records: i64 = db.query_row("SELECT count(*) FROM records", [], |r| r.get(0))?; @@ -340,12 +341,12 @@ fn compressed_dumps_resume_and_corruption_fail_closed() -> anyhow::Result<()> { let mut truncated = common::manifest(Source::Wikidata, Format::WikidataDump, &damaged)?; truncated.compression = Compression::Gzip; assert!(store::import(&mut db, &truncated, &path).is_err()); - let complete: i64 = db.query_row( - "SELECT complete FROM sources WHERE id=?1", + let retained: i64 = db.query_row( + "SELECT count(*) FROM sources WHERE id=?1", [truncated.id()?], |r| r.get(0), )?; - assert_eq!(complete, 0); + assert_eq!(retained, 0); Ok(()) } diff --git a/docs/CONSUMERS.md b/docs/CONSUMERS.md index 54ba604..0480212 100644 --- a/docs/CONSUMERS.md +++ b/docs/CONSUMERS.md @@ -2,9 +2,11 @@ ## Rust library -Use `argand_site_registry::query::Registry::open(generation, trusted_pin)` once per -immutable generation and reuse the reader. `lookup(query, limit)` returns evidence -and complete ambiguity counts; `resolve_explained` returns a reviewed candidate or +Use `release::verify_signed(generation, publisher_signers, publisher_identity, +reviewer_signers)` once per production generation and reuse the returned reader. +`Registry::open(generation, trusted_pin)` is the lower-level path when the pin +distributor is also trusted for the complete review decision. `lookup(query, +limit)` returns evidence and complete ambiguity counts; `resolve_explained` returns a reviewed candidate or a typed abstention reason with counts. Exact reverse views cover entity IDs, URLs/domains, popularity and Curlie categories. Check the compiled example and API docs for exact types. `selection_context` binds the full alternative set for @@ -39,9 +41,10 @@ omits descriptions and includes fact provenance plus an attribution envelope. It is an assertion export, not a self-contained signed list of admitted routes. Raw SQL inspection is useful for audit; it does not implement resolution policy. -Code version 0.2.0 uses schema version 2 and `argand.site-rules/v2`. -It adds authenticated reviewer proofs, exact immutable SQLite opening, typed diffs -and audit/evaluation APIs. Schema/rule contracts remain versioned independently in +Code version 0.3.0 uses schema version 3 and `argand.site-rules/v3`. +It adds reviewer-trust enforcement for consumers, private authenticated SQLite +snapshots, exact-stream import checks, bounded outputs and metadata-bound identity +decisions. Schema/rule contracts remain versioned independently in receipts. Unsupported contracts fail closed. Pin source releases, compile consumers and replay fixed fixtures before upgrades. Preserve import and review history; never mutate complete generations to migrate them. diff --git a/docs/PUBLISHING.md b/docs/PUBLISHING.md index 1d5ecae..7fbf23f 100644 --- a/docs/PUBLISHING.md +++ b/docs/PUBLISHING.md @@ -14,6 +14,12 @@ one accepted principal and public key per line: operator@example.org ssh-ed25519 REVIEWER_PUBLIC_KEY ``` +For rotation, retain an old public key with an OpenSSH `valid-before` option +covering its signed decision times. New decisions are checked at append time, so +an expired key cannot submit backdated reviews; historical release verification +uses each authenticated `reviewed_at`. Remove a retired key only after no retained +generation or review log depends on it. + Distribute the release publisher public key to consumers through an independent authenticated channel. Do not put private keys, production trust files or source datasets in Git or CI. The isolated source CI runner has none of these files. @@ -62,6 +68,7 @@ argand-site-registry sign --generation /data/registry/reviewed \ argand-site-registry activate --generation /data/registry/reviewed \ --current /data/registry/current.json \ --allowed-signers /secure/release-allowed-signers \ + --allowed-reviewers /secure/reviewer-allowed-signers \ --identity registry-publisher ``` diff --git a/docs/TRUST.md b/docs/TRUST.md index e73a2df..99a4f61 100644 --- a/docs/TRUST.md +++ b/docs/TRUST.md @@ -18,9 +18,11 @@ writer retains their exact decision/signature bytes in the append-only review lo Generations bind the database, license document and attribution to a completion receipt. Consumers provide a trusted hash or verify an external publisher key. Release signing re-verifies every stored decision against an external reviewer -trust file. Activation checks publisher signatures, rejects structurally incomplete -review proofs and refuses rollback that loses distributed revocations. Updates -build candidates and cannot approve, sign or activate them. +trust file. Activation checks publisher signatures, re-verifies every review +against a separately supplied reviewer trust file and refuses rollback that loses +distributed revocations. Reviewer validity epochs are evaluated at decision time +for retained history while new decisions must pass the trust policy at append +time. Updates build candidates and cannot approve, sign or activate them. ## What a publisher must establish @@ -47,8 +49,11 @@ enforced quorum. ## What consumers must preserve -Authenticate a release before opening it. Keep the full receipt pin and required -attribution with caches and exports. Use `resolve` for reviewed destinations, keep +Authenticate a release before opening it. Consumers that rely on reviewer +separation must use `release::verify_signed` or `activate` with independently +distributed publisher and reviewer trust files; a receipt pin alone delegates the +whole release decision to whoever distributed that pin. Keep the full receipt pin +and required attribution with caches and exports. Use `resolve` for reviewed destinations, keep null as abstention, and enforce application-specific malware/content/navigation policy separately. A verified signature authenticates the publisher, not the truth of every assertion. An official website may later be compromised.