diff --git a/scripts/soak/build_portable_soak_package.sh b/scripts/soak/build_portable_soak_package.sh index d2ff988..0eac491 100755 --- a/scripts/soak/build_portable_soak_package.sh +++ b/scripts/soak/build_portable_soak_package.sh @@ -14,7 +14,7 @@ Usage: scripts/soak/build_portable_soak_package.sh [--out-dir ] [--profile ] Requires release binaries to already exist. Build them first, for example: - cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer + cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer --bin verification_ccr_compare USAGE } @@ -53,7 +53,7 @@ else TARGET_BIN_DIR="$REPO_ROOT/target/$PROFILE" fi -REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer) +REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer verification_ccr_compare) OPTIONAL_BINS=( ccr_dump ccr_state_compare @@ -82,7 +82,7 @@ STAGE_DIR="$BUILD_ROOT/$PACKAGE_DIR_NAME" ARCHIVE_PATH="$OUT_DIR/$PACKAGE_NAME.tar.gz" rm -rf "$STAGE_DIR" "$ARCHIVE_PATH" -mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" \ +mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" "$STAGE_DIR/scripts/verification" \ "$STAGE_DIR/runs" "$STAGE_DIR/state" "$STAGE_DIR/logs" "$STAGE_DIR/tmp" install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh" @@ -92,6 +92,8 @@ install -m 0755 "$SCRIPT_DIR/fixed_phase_loop.sh" "$STAGE_DIR/scripts/soak/fixed install -m 0755 "$SCRIPT_DIR/hourly_soak_report.py" "$STAGE_DIR/scripts/soak/hourly_soak_report.py" install -m 0755 "$SCRIPT_DIR/publish_remote231.sh" "$STAGE_DIR/scripts/soak/publish_remote231.sh" install -m 0755 "$SCRIPT_DIR/publish_remote231_full.sh" "$STAGE_DIR/scripts/soak/publish_remote231_full.sh" +install -m 0755 "$REPO_ROOT/scripts/verification/run_verification_only.sh" "$STAGE_DIR/scripts/verification/run_verification_only.sh" +install -m 0644 "$REPO_ROOT/scripts/verification/README.md" "$STAGE_DIR/scripts/verification/README.md" install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/.env" install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/portable-soak.env.example" diff --git a/scripts/soak/run_soak.sh b/scripts/soak/run_soak.sh index 15d1b13..350be4f 100755 --- a/scripts/soak/run_soak.sh +++ b/scripts/soak/run_soak.sh @@ -48,6 +48,7 @@ TMP_DIR="${TMP_DIR:-$RUN_ROOT/tmp}" RSYNC_MIRROR_ROOT="${RSYNC_MIRROR_ROOT:-$STATE_ROOT/rsync-mirror}" INVALID_ROOT="$STATE_ROOT/invalid" RESET_STAGING_ROOT="$STATE_ROOT/reset-staging" +STATE_EXECUTION_LOCK_PATH="$STATE_ROOT/locks/state-execution.lock" LIVE_TA_REFRESH_DIR="${LIVE_TA_REFRESH_DIR:-$META_DIR/live-ta-refresh}" LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS="${LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS:-15}" LIVE_TA_REFRESH_MAX_TIME_SECS="${LIVE_TA_REFRESH_MAX_TIME_SECS:-120}" @@ -1080,6 +1081,7 @@ build_child_args() { CHILD_ARGS+=( --report-json "{run_out}/report.json" + --validation-contract-out "{run_out}/validation-contract.json" ) if is_true "$OUTPUT_COMPACT_REPORT"; then CHILD_ARGS+=(--report-json-compact) @@ -1109,6 +1111,59 @@ build_child_args() { fi } +write_current_state_binding() { + local run_dir="$1" + local run_id="$2" + local output_path="$META_DIR/current-state-binding.json" + python3 - "$run_dir" "$run_id" "$output_path" <<'PY' +import hashlib +import json +import os +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +run_dir = Path(sys.argv[1]) +run_id = sys.argv[2] +output_path = Path(sys.argv[3]) +summary = json.loads((run_dir / "run-summary.json").read_text(encoding="utf-8")) +if summary.get("status") != "success": + raise SystemExit(f"cannot bind unsuccessful source run: {run_id}") + +def digest(name): + path = run_dir / name + if not path.is_file(): + raise SystemExit(f"current state binding input is missing: {path}") + hasher = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + hasher.update(chunk) + return hasher.hexdigest() + +binding = { + "schemaVersion": 1, + "currentRunId": run_id, + "sourceCirSha256": digest("input.cir"), + "sourceCcrSha256": digest("result.ccr"), + "validationContractSha256": digest("validation-contract.json"), + "updatedAtRfc3339Utc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), +} +output_path.parent.mkdir(parents=True, exist_ok=True) +fd, tmp_name = tempfile.mkstemp(prefix=output_path.name + ".", dir=output_path.parent) +try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(binding, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, output_path) +finally: + if os.path.exists(tmp_name): + os.unlink(tmp_name) +PY +} + copy_inner_run_outputs() { local daemon_state_root="$1" local run_dir="$2" @@ -1304,6 +1359,12 @@ run_one_round() { final_status="failed" fi fi + if [[ "$final_status" == "success" ]]; then + if ! write_current_state_binding "$run_dir" "$run_id"; then + warn "failed to bind successful run to current state: $run_id" + final_status="failed" + fi + fi write_run_meta "$run_dir/run-meta.json" "$final_status" "$run_index" "$run_id" "$sync_mode" \ "$snapshot_reason" "$previous_run_id" "$previous_success_value" "$started_at" "$completed_at" \ "$INVALID_DB_PATH" "$INVALID_STATE_PATH" "$INVALID_TMP_PATH" "$daemon_exit_code" "$PACKAGE_ROOT" "$ENV_FILE" \ @@ -1327,6 +1388,7 @@ main() { require_command python3 require_command date require_command find + require_command flock if [[ "$TAL_INPUT_MODE" == "file-live-ta" ]]; then require_command curl validate_positive_int "LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" "$LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" @@ -1360,7 +1422,7 @@ main() { fi done - mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR" + mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR" "$(dirname "$STATE_EXECUTION_LOCK_PATH")" if is_true "$ALLOW_RSYNC_MIRROR_REUSE"; then mkdir -p "$RSYNC_MIRROR_ROOT" fi @@ -1386,6 +1448,9 @@ main() { local any_failed=0 while (( run_forever == 1 || next_index <= stop_index )); do + local state_lock_fd + exec {state_lock_fd}>"$STATE_EXECUTION_LOCK_PATH" + flock "$state_lock_fd" INVALID_DB_PATH="" INVALID_STATE_PATH="" INVALID_TMP_PATH="" @@ -1472,6 +1537,8 @@ main() { echo "completed run $(printf 'run_%04d' "$next_index") status=failed" >&2 any_failed=1 fi + flock -u "$state_lock_fd" + exec {state_lock_fd}>&- if (( STOP_AFTER_SECS > 0 )); then elapsed_secs=$(( $(date +%s) - started_epoch )) if (( elapsed_secs >= STOP_AFTER_SECS )); then diff --git a/scripts/verification/README.md b/scripts/verification/README.md new file mode 100644 index 0000000..af0347a --- /dev/null +++ b/scripts/verification/README.md @@ -0,0 +1,50 @@ +# verification-only 使用说明 + +`verification-only` 用于复核启用验证缓存的最新成功 normal run。它冻结当前仓库视图，关闭 PP、ROA、child certificate cache、prefetch、VCIR 持久化和网络访问，在源 CIR 的同一验证时间重新执行完整验证，并比较 CCR 的 MFT、VRP、VAP、TA、router key 五类 state digest。 + +## 前提 + +- 源 run 必须是当前 state DB 对应的最新成功 normal run。 +- 源 run 必须包含 `input.cir`、`result.ccr`、`run-summary.json` 和 `validation-contract.json`。 +- state root 必须包含 `db/work-db`、`db/repo-bytes.db` 和 `meta/current-state-binding.json`。 +- 源 normal run 必须启用至少一种验证缓存；仅启用 transport prefetch 不算验证缓存。 +- verification-only 与 normal soak 共用 `state/locks/state-execution.lock`，不能绕过 wrapper 并发执行。 + +## 执行 + +在 portable package 根目录执行： + +```bash +./scripts/verification/run_verification_only.sh \ + --source-run-dir /data/ours-rp/runs/run_0123 \ + --source-state-root /data/ours-rp/state \ + --out-dir /data/ours-rp/verification-runs/run_0123 +``` + +验证时间自动从源 `input.cir` 读取，不接受命令行覆盖。成功后 scratch work-db 默认删除；排障时可加 `--keep-scratch`。 + +## 输出 + +- `input.cir`、`result.ccr`、`report.json` +- `vrps.csv`、`vaps.csv` +- `stage-timing.json`、`validation-events.jsonl` +- `validation-contract.json`、`verification-meta.json` +- `compare/ccr-state-compare.json`、`compare/ccr-state-compare.md` +- `stdout.log`、`stderr.log`、`process-time.txt`、`wrapper-timing.json` + +通过条件是 compare JSON 中 `stateDigestMatch=true`。`producedAt` 不参与比较。 + +## 隔离保证 + +- 不分配 normal run id，不更新 lifecycle、periodic snapshot 计数和 retention。 +- 源 work-db 只用于创建 checkpoint，后续写入只发生在独立 scratch work-db。 +- 源 repo-bytes.db 以只读模式打开。 +- 输出目录必须不存在，防止覆盖旧验证证据。 + +## 常见失败 + +- `current state binding`：所选 run 已不是当前 state 对应的最新成功 run，需选择 binding 指向的 run。 +- `differs from current state binding`：源 CIR、CCR 或 contract 已变化，不能继续复核。 +- `blob is missing`：repository view 引用的原始 bytes 不完整，验证不会启动。 +- `did not enable a validation cache`：源 run 不是缓存正确性对比的有效基准。 +- `CCR state digest mismatch`：查看 `compare/` 下 JSON/Markdown，按不匹配状态集合继续排查。 diff --git a/scripts/verification/run_verification_only.sh b/scripts/verification/run_verification_only.sh new file mode 100755 index 0000000..b954023 --- /dev/null +++ b/scripts/verification/run_verification_only.sh @@ -0,0 +1,102 @@ +#!/usr/bin/env bash +set -euo pipefail + +usage() { + cat <<'USAGE' +Usage: + run_verification_only.sh \ + --source-run-dir \ + --source-state-root \ + --out-dir \ + [--rpki-bin ] [--keep-scratch] + +validation time is always read from the source CIR. The wrapper serializes access +with normal soak runs and never writes into the normal run sequence. +USAGE +} + +SOURCE_RUN_DIR="" +SOURCE_STATE_ROOT="" +OUT_DIR="" +RPKI_BIN="" +KEEP_SCRATCH=0 +while [[ $# -gt 0 ]]; do + case "$1" in + --source-run-dir) SOURCE_RUN_DIR="${2:?missing value}"; shift 2 ;; + --source-state-root) SOURCE_STATE_ROOT="${2:?missing value}"; shift 2 ;; + --out-dir) OUT_DIR="${2:?missing value}"; shift 2 ;; + --rpki-bin) RPKI_BIN="${2:?missing value}"; shift 2 ;; + --keep-scratch) KEEP_SCRATCH=1; shift ;; + -h|--help) usage; exit 0 ;; + *) echo "unknown argument: $1" >&2; usage >&2; exit 2 ;; + esac +done + +[[ -n "$SOURCE_RUN_DIR" ]] || { echo "--source-run-dir is required" >&2; exit 2; } +[[ -n "$SOURCE_STATE_ROOT" ]] || { echo "--source-state-root is required" >&2; exit 2; } +[[ -n "$OUT_DIR" ]] || { echo "--out-dir is required" >&2; exit 2; } +if [[ -z "$RPKI_BIN" ]]; then + RPKI_BIN="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)/target/release/rpki" +fi +[[ -x "$RPKI_BIN" ]] || { echo "rpki binary is not executable: $RPKI_BIN" >&2; exit 2; } +command -v flock >/dev/null || { echo "flock is required" >&2; exit 2; } +command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; } + +now_ms() { + python3 -c 'import time; print(time.time_ns() // 1_000_000)' +} + +mkdir -p "$SOURCE_STATE_ROOT/locks" "$(dirname "$OUT_DIR")" +[[ ! -e "$OUT_DIR" ]] || { echo "verification output already exists: $OUT_DIR" >&2; exit 2; } +LOCK_PATH="$SOURCE_STATE_ROOT/locks/state-execution.lock" +STDOUT_TMP="${OUT_DIR}.stdout.tmp" +STDERR_TMP="${OUT_DIR}.stderr.tmp" +PROCESS_TIME_TMP="${OUT_DIR}.process-time.tmp" +WAIT_STARTED_MS="$(now_ms)" +exec {LOCK_FD}>"$LOCK_PATH" +flock "$LOCK_FD" +LOCK_ACQUIRED_MS="$(now_ms)" + +ARGS=( + --verification-only + --verification-source-run "$SOURCE_RUN_DIR" + --verification-source-state "$SOURCE_STATE_ROOT" + --verification-out "$OUT_DIR" +) +if [[ "$KEEP_SCRATCH" -eq 1 ]]; then + ARGS+=(--keep-verification-scratch) +fi + +set +e +if [[ -x /usr/bin/time ]]; then + RPKI_STATE_EXECUTION_LOCK_HELD=1 /usr/bin/time -v -o "$PROCESS_TIME_TMP" -- \ + "$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP" +else + RPKI_STATE_EXECUTION_LOCK_HELD=1 "$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP" +fi +EXIT_CODE=$? +set -e +LOCK_RELEASED_MS="$(now_ms)" +flock -u "$LOCK_FD" +exec {LOCK_FD}>&- + +mkdir -p "$OUT_DIR" +mv "$STDOUT_TMP" "$OUT_DIR/stdout.log" +mv "$STDERR_TMP" "$OUT_DIR/stderr.log" +if [[ -f "$PROCESS_TIME_TMP" ]]; then + mv "$PROCESS_TIME_TMP" "$OUT_DIR/process-time.txt" +fi +python3 - "$OUT_DIR/wrapper-timing.json" "$WAIT_STARTED_MS" "$LOCK_ACQUIRED_MS" "$LOCK_RELEASED_MS" "$EXIT_CODE" <<'PY' +import json +import sys +from pathlib import Path + +path = Path(sys.argv[1]) +wait_started_ms, lock_acquired_ms, lock_released_ms, exit_code = map(int, sys.argv[2:]) +path.write_text(json.dumps({ + "lockWaitMs": lock_acquired_ms - wait_started_ms, + "lockHeldMs": lock_released_ms - lock_acquired_ms, + "exitCode": exit_code, +}, indent=2, sort_keys=True) + "\n", encoding="utf-8") +PY +exit "$EXIT_CODE" diff --git a/src/bin/verification_ccr_compare.rs b/src/bin/verification_ccr_compare.rs new file mode 100644 index 0000000..e810380 --- /dev/null +++ b/src/bin/verification_ccr_compare.rs @@ -0,0 +1,171 @@ +use std::path::PathBuf; + +fn usage() -> &'static str { + "Usage: verification_ccr_compare --source-ccr --verification-ccr --out-json [--out-md ]" +} + +#[derive(Debug, PartialEq, Eq)] +struct Args { + source_ccr: PathBuf, + verification_ccr: PathBuf, + out_json: PathBuf, + out_md: Option, +} + +fn parse_args(argv: &[String]) -> Result { + let mut source_ccr = None; + let mut verification_ccr = None; + let mut out_json = None; + let mut out_md = None; + let mut index = 1usize; + while index < argv.len() { + let target = match argv[index].as_str() { + "--source-ccr" => &mut source_ccr, + "--verification-ccr" => &mut verification_ccr, + "--out-json" => &mut out_json, + "--out-md" => &mut out_md, + "-h" | "--help" => return Err(usage().to_string()), + other => return Err(format!("unknown argument: {other}\n{}", usage())), + }; + index += 1; + *target = + Some(PathBuf::from(argv.get(index).ok_or_else(|| { + format!("{} requires a value", argv[index - 1]) + })?)); + index += 1; + } + let source_ccr = source_ccr.ok_or_else(|| format!("--source-ccr is required\n{}", usage()))?; + let verification_ccr = + verification_ccr.ok_or_else(|| format!("--verification-ccr is required\n{}", usage()))?; + let out_json = out_json.ok_or_else(|| format!("--out-json is required\n{}", usage()))?; + Ok(Args { + source_ccr, + verification_ccr, + out_json, + out_md, + }) +} + +fn run(args: &Args) -> Result<(), String> { + let comparison = + rpki::verification_only::compare_ccr_files(&args.source_ccr, &args.verification_ccr)?; + rpki::verification_only::write_ccr_comparison( + &args.out_json, + args.out_md.as_deref(), + &comparison, + )?; + if !comparison.state_digest_match { + return Err(format!( + "CCR state digest mismatch: {}", + comparison.mismatched_states.join(",") + )); + } + println!("CCR state digests match: {}", args.out_json.display()); + Ok(()) +} + +fn main() -> Result<(), String> { + let argv = std::env::args().collect::>(); + run(&parse_args(&argv)?) +} + +#[cfg(test)] +mod tests { + use super::*; + use rpki::ccr::{ + CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation, + build_roa_payload_state, encode_content_info, + }; + use rpki::data_model::roa::{IpPrefix, RoaAfi}; + use rpki::validation::objects::Vrp; + use std::fs; + + fn argv(values: &[&str]) -> Vec { + std::iter::once("verification_ccr_compare") + .chain(values.iter().copied()) + .map(str::to_string) + .collect() + } + + fn sample_ccr(asn: u32) -> Vec { + let vrps = build_roa_payload_state(&[Vrp { + asn, + prefix: IpPrefix { + afi: RoaAfi::Ipv4, + prefix_len: 24, + addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + }, + max_length: 24, + }]) + .expect("build VRP state"); + encode_content_info(&CcrContentInfo::new(RpkiCanonicalCacheRepresentation { + version: 0, + hash_alg: CcrDigestAlgorithm::Sha256, + produced_at: time::OffsetDateTime::UNIX_EPOCH, + mfts: None, + vrps: Some(vrps), + vaps: None, + tas: None, + rks: None, + })) + .expect("encode CCR") + } + + #[test] + fn parse_args_accepts_complete_input_and_rejects_bad_forms() { + let parsed = parse_args(&argv(&[ + "--source-ccr", + "source.ccr", + "--verification-ccr", + "verification.ccr", + "--out-json", + "result.json", + "--out-md", + "result.md", + ])) + .expect("parse"); + assert_eq!(parsed.source_ccr, PathBuf::from("source.ccr")); + assert_eq!(parsed.out_md, Some(PathBuf::from("result.md"))); + assert!(parse_args(&argv(&[])).unwrap_err().contains("source-ccr")); + assert!( + parse_args(&argv(&["--source-ccr"])) + .unwrap_err() + .contains("requires a value") + ); + assert!( + parse_args(&argv(&["--unknown"])) + .unwrap_err() + .contains("unknown argument") + ); + assert!(parse_args(&argv(&["--help"])).is_err()); + } + + #[test] + fn run_writes_matching_outputs_and_rejects_mismatch() { + let temp = tempfile::tempdir().expect("tempdir"); + let source = temp.path().join("source.ccr"); + let verification = temp.path().join("verification.ccr"); + let out_json = temp.path().join("out/result.json"); + let out_md = temp.path().join("out/result.md"); + fs::write(&source, sample_ccr(64496)).expect("source"); + fs::write(&verification, sample_ccr(64496)).expect("verification"); + let args = Args { + source_ccr: source.clone(), + verification_ccr: verification.clone(), + out_json: out_json.clone(), + out_md: Some(out_md.clone()), + }; + run(&args).expect("matching run"); + assert!(out_json.is_file()); + assert!(out_md.is_file()); + + fs::write(&verification, sample_ccr(64497)).expect("mismatch"); + let error = run(&args).unwrap_err(); + assert!(error.contains("vrps")); + assert!( + fs::read_to_string(out_json) + .expect("comparison JSON") + .contains("mismatchedStates") + ); + } +} diff --git a/src/blob_store.rs b/src/blob_store.rs index d5c622b..cd65327 100644 --- a/src/blob_store.rs +++ b/src/blob_store.rs @@ -77,6 +77,7 @@ pub struct ExternalRawStoreDb { pub struct ExternalRepoBytesDb { path: PathBuf, db: Arc, + read_only: bool, } impl ExternalRawStoreDb { @@ -189,6 +190,7 @@ impl ExternalRepoBytesDb { Ok(Self { path, db: Arc::new(db), + read_only: false, }) } @@ -201,6 +203,7 @@ impl ExternalRepoBytesDb { Ok(Self { path, db: Arc::new(db), + read_only: true, }) } @@ -208,6 +211,12 @@ impl ExternalRepoBytesDb { if blobs.is_empty() { return Ok(()); } + if self.read_only { + return Err(StorageError::RocksDb(format!( + "repo-bytes DB is read-only: {}", + self.path.display() + ))); + } let mut batch = WriteBatch::default(); for (sha256_hex, bytes) in blobs { validate_blob_sha256_hex(sha256_hex)?; @@ -221,6 +230,42 @@ impl ExternalRepoBytesDb { Ok(()) } + pub(crate) fn is_read_only(&self) -> bool { + self.read_only + } + + pub(crate) fn require_existing_blob_bytes_batch( + &self, + blobs: &[(String, Vec)], + ) -> StorageResult<()> { + if blobs.is_empty() { + return Ok(()); + } + let hashes = blobs + .iter() + .map(|(sha256_hex, _)| sha256_hex.clone()) + .collect::>(); + let existing = self.get_blob_bytes_batch(&hashes)?; + for ((sha256_hex, expected_bytes), actual_bytes) in blobs.iter().zip(existing) { + match actual_bytes { + Some(actual_bytes) if actual_bytes == *expected_bytes => {} + Some(_) => { + return Err(StorageError::InvalidData { + entity: "read_only_repo_bytes", + detail: format!("existing bytes differ for SHA-256 {sha256_hex}"), + }); + } + None => { + return Err(StorageError::InvalidData { + entity: "read_only_repo_bytes", + detail: format!("blob is missing for SHA-256 {sha256_hex}"), + }); + } + } + } + Ok(()) + } + pub fn get_blob_bytes(&self, sha256_hex: &str) -> StorageResult>> { validate_blob_sha256_hex(sha256_hex)?; let key = repo_bytes_key(sha256_hex); diff --git a/src/cli.rs b/src/cli.rs index 6e7e75b..b41341d 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -115,6 +115,12 @@ fn vcir_storage_summary_enabled() -> bool { #[derive(Clone, Debug, PartialEq, Eq)] pub struct CliArgs { + pub verification_only: bool, + pub verification_source_run_dir: Option, + pub verification_source_state_root: Option, + pub verification_out_dir: Option, + pub keep_verification_scratch: bool, + pub validation_contract_out_path: Option, pub tal_urls: Vec, pub tal_paths: Vec, pub ta_paths: Vec, @@ -184,8 +190,19 @@ fn usage() -> String { Usage: {bin} --db --tal-url [--tal-url ...] [options] {bin} --db --tal-path --ta-path [--tal-path --ta-path ...] [options] + {bin} --verification-only --verification-source-run --verification-source-state --verification-out [options] Options: + --verification-only Revalidate the latest successful run without validation caches or network access + --verification-source-run + Source normal run directory containing input.cir/result.ccr/validation-contract.json + --verification-source-state + Source state root containing db/work-db and db/repo-bytes.db + --verification-out Independent output directory for the one-shot verification run + --keep-verification-scratch + Keep the private work-db checkpoint after a successful verification + --validation-contract-out + Write the effective normal-run validation contract (not valid in verification-only mode) --db RocksDB directory path (required) --raw-store-db External raw-by-hash store DB path (optional) --repo-bytes-db External repo object bytes DB path (optional) @@ -273,6 +290,12 @@ Options: } pub fn parse_args(argv: &[String]) -> Result { + let mut verification_only = false; + let mut verification_source_run_dir: Option = None; + let mut verification_source_state_root: Option = None; + let mut verification_out_dir: Option = None; + let mut keep_verification_scratch = false; + let mut validation_contract_out_path: Option = None; let mut tal_urls: Vec = Vec::new(); let mut tal_paths: Vec = Vec::new(); let mut ta_paths: Vec = Vec::new(); @@ -333,6 +356,39 @@ pub fn parse_args(argv: &[String]) -> Result { let arg = argv[i].as_str(); match arg { "--help" | "-h" => return Err(usage()), + "--verification-only" => { + verification_only = true; + } + "--verification-source-run" => { + i += 1; + verification_source_run_dir = Some(PathBuf::from( + argv.get(i) + .ok_or("--verification-source-run requires a value")?, + )); + } + "--verification-source-state" => { + i += 1; + verification_source_state_root = Some(PathBuf::from( + argv.get(i) + .ok_or("--verification-source-state requires a value")?, + )); + } + "--verification-out" => { + i += 1; + verification_out_dir = Some(PathBuf::from( + argv.get(i).ok_or("--verification-out requires a value")?, + )); + } + "--keep-verification-scratch" => { + keep_verification_scratch = true; + } + "--validation-contract-out" => { + i += 1; + validation_contract_out_path = Some(PathBuf::from( + argv.get(i) + .ok_or("--validation-contract-out requires a value")?, + )); + } "--tal-url" => { i += 1; let v = argv.get(i).ok_or("--tal-url requires a value")?; @@ -717,10 +773,91 @@ pub fn parse_args(argv: &[String]) -> Result { i += 1; } + if verification_only { + if verification_source_run_dir.is_none() { + return Err(format!( + "--verification-only requires --verification-source-run\n\n{}", + usage() + )); + } + if verification_source_state_root.is_none() { + return Err(format!( + "--verification-only requires --verification-source-state\n\n{}", + usage() + )); + } + if verification_out_dir.is_none() { + return Err(format!( + "--verification-only requires --verification-out\n\n{}", + usage() + )); + } + if db_path.is_some() + || raw_store_db.is_some() + || repo_bytes_db.is_some() + || !tal_urls.is_empty() + || !tal_paths.is_empty() + || !ta_paths.is_empty() + || policy_path.is_some() + || strict_policy.is_some() + || resource_validation_mode.is_some() + || validation_time.is_some() + || max_ca_depth_option.is_some() + || max_instances.is_some() + || enable_roa_validation_cache + || enable_child_certificate_validation_cache + || publication_point_cache_observe_only + || enable_publication_point_validation_cache + || enable_transport_request_prefetch + || rsync_local_dir.is_some() + || rsync_command.is_some() + || !http_root_cert_paths.is_empty() + || payload_replay_archive.is_some() + || payload_replay_locks.is_some() + || payload_base_archive.is_some() + || payload_base_locks.is_some() + || payload_delta_archive.is_some() + || payload_delta_locks.is_some() + || validation_contract_out_path.is_some() + { + return Err(format!( + "verification-only derives DB, TAL, validation time, policy and cache settings from the source run; conflicting normal-run options are not allowed\n\n{}", + usage() + )); + } + if report_json_path.is_some() + || ccr_out_path.is_some() + || vrps_csv_out_path.is_some() + || vaps_csv_out_path.is_some() + || cir_enabled + || cir_out_path.is_some() + { + return Err(format!( + "verification-only output paths are derived from --verification-out\n\n{}", + usage() + )); + } + db_path = Some( + verification_out_dir + .as_ref() + .expect("checked verification out") + .join("scratch/work-db"), + ); + } else if verification_source_run_dir.is_some() + || verification_source_state_root.is_some() + || verification_out_dir.is_some() + || keep_verification_scratch + { + return Err(format!( + "verification source/output options require --verification-only\n\n{}", + usage() + )); + } + let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?; let tal_mode_count = (!tal_urls.is_empty()) as u8 + (!tal_paths.is_empty()) as u8; - if tal_mode_count != 1 { + if !verification_only && tal_mode_count != 1 { return Err(format!( "must specify either one-or-more --tal-url or one-or-more --tal-path/--ta-path pairs\n\n{}", usage() @@ -774,13 +911,13 @@ pub fn parse_args(argv: &[String]) -> Result { usage() )); } - if !tal_urls.is_empty() && !ta_paths.is_empty() { + if !verification_only && !tal_urls.is_empty() && !ta_paths.is_empty() { return Err(format!( "--ta-path cannot be used with --tal-url mode\n\n{}", usage() )); } - if !tal_paths.is_empty() { + if !verification_only && !tal_paths.is_empty() { if !ta_paths.is_empty() { if ta_paths.len() != tal_paths.len() { return Err(format!( @@ -955,6 +1092,12 @@ pub fn parse_args(argv: &[String]) -> Result { } Ok(CliArgs { + verification_only, + verification_source_run_dir, + verification_source_state_root, + verification_out_dir, + keep_verification_scratch, + validation_contract_out_path, tal_urls, tal_paths, ta_paths, @@ -1869,7 +2012,7 @@ where H: crate::sync::rrdp::Fetcher + Clone + 'static, R: crate::fetch::rsync::RsyncFetcher + Clone + 'static, { - if args.tal_inputs.len() > 1 { + if args.verification_only || args.tal_inputs.len() > 1 { return if let Some(t) = timing { run_tree_from_multiple_tals_parallel_phase2_audit_with_timing( store, @@ -2010,21 +2153,88 @@ where } pub fn run(argv: &[String]) -> Result<(), String> { - let args = parse_args(argv)?; + let mut args = parse_args(argv)?; + let mut prepared_verification = if args.verification_only { + let prepared = crate::verification_only::prepare_verification( + args.verification_source_run_dir + .as_deref() + .expect("validated by parse_args"), + args.verification_source_state_root + .as_deref() + .expect("validated by parse_args"), + args.verification_out_dir + .as_deref() + .expect("validated by parse_args"), + )?; + args.db_path = prepared.artifacts.scratch_work_db.clone(); + args.repo_bytes_db = Some(prepared.artifacts.source_repo_bytes_db.clone()); + args.report_json_path = Some(prepared.artifacts.report_json.clone()); + args.report_json_compact = true; + args.skip_report_build = false; + args.skip_vcir_persist = true; + args.enable_roa_validation_cache = false; + args.enable_child_certificate_validation_cache = false; + args.publication_point_cache_observe_only = false; + args.enable_publication_point_validation_cache = false; + args.enable_transport_request_prefetch = false; + args.ccr_out_path = Some(prepared.artifacts.result_ccr.clone()); + args.vrps_csv_out_path = Some(prepared.artifacts.vrps_csv.clone()); + args.vaps_csv_out_path = Some(prepared.artifacts.vaps_csv.clone()); + args.compare_view_trust_anchor = Some("verification-only".to_string()); + args.cir_enabled = true; + args.cir_out_path = Some(prepared.artifacts.result_cir.clone()); + args.cir_tal_uris = crate::verification_only::cir_tal_uris(&prepared.cir); + args.tal_inputs = crate::verification_only::tal_inputs_from_cir(&prepared.cir); + args.validation_time = Some(prepared.cir.validation_time); + args.max_ca_depth = prepared.contract.max_ca_depth; + args.max_instances = prepared.contract.max_instances; + args.disable_rrdp = true; + args.rsync_local_dir = None; + Some(prepared) + } else { + None + }; - let mut policy = read_policy(args.policy_path.as_deref())?; - if let Some(strict_policy) = args.strict_policy { - policy.strict = strict_policy; - } - if let Some(resource_validation_mode) = args.resource_validation_mode { - policy.resource_validation_mode = resource_validation_mode; - } - if args.disable_rrdp { + let mut policy = if let Some(prepared) = prepared_verification.as_ref() { + prepared.contract.policy.clone() + } else { + read_policy(args.policy_path.as_deref())? + }; + if !args.verification_only { + if let Some(strict_policy) = args.strict_policy { + policy.strict = strict_policy; + } + if let Some(resource_validation_mode) = args.resource_validation_mode { + policy.resource_validation_mode = resource_validation_mode; + } + if args.disable_rrdp { + policy.sync_preference = crate::policy::SyncPreference::RsyncOnly; + } + } else { policy.sync_preference = crate::policy::SyncPreference::RsyncOnly; + policy.ca_failed_fetch_policy = crate::policy::CaFailedFetchPolicy::StopAllOutput; } let validation_time = args .validation_time .unwrap_or_else(time::OffsetDateTime::now_utc); + let validation_time = + time::OffsetDateTime::from_unix_timestamp(validation_time.unix_timestamp()) + .map_err(|error| format!("normalize validation time failed: {error}"))?; + if let Some(path) = args.validation_contract_out_path.as_deref() { + let contract = crate::verification_only::ValidationContract::for_current_binary( + validation_time, + policy.clone(), + args.max_ca_depth, + args.max_instances, + crate::verification_only::ValidationCacheContract { + publication_point: args.enable_publication_point_validation_cache, + roa: args.enable_roa_validation_cache, + child_certificate: args.enable_child_certificate_validation_cache, + transport_prefetch: args.enable_transport_request_prefetch, + }, + )?; + crate::verification_only::write_validation_contract(path, &contract)?; + } let http_root_certificates_pem = args .http_root_cert_paths .iter() @@ -2034,7 +2244,12 @@ pub fn run(argv: &[String]) -> Result<(), String> { }) .collect::, _>>()?; - let store = if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() { + let store = if let Some(prepared) = prepared_verification.as_mut() { + prepared + .store + .take() + .expect("prepared verification store must be available") + } else if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() { Arc::new( RocksStore::open_with_external_stores( &args.db_path, @@ -2137,7 +2352,23 @@ pub fn run(argv: &[String]) -> Result<(), String> { ); let validation_started = std::time::Instant::now(); let collect_current_repo_objects = false; - let out = if delta_replay_mode { + let out = if args.verification_only { + let http = crate::verification_only::NetworkDisabledHttpFetcher; + let rsync = crate::fetch::current_repository::CurrentRepositoryViewRsyncFetcher::new( + Arc::clone(&store), + ); + run_online_validation_with_fetchers( + Arc::clone(&store), + &policy, + &args, + &http, + &rsync, + validation_time, + &config, + collect_current_repo_objects, + timing.as_ref().map(|(_, t)| t), + )? + } else if delta_replay_mode { let tal_path = args .tal_path .as_ref() @@ -2666,6 +2897,62 @@ pub fn run(argv: &[String]) -> Result<(), String> { } print_summary_from_shared(validation_time, &shared); + if let Some(prepared) = prepared_verification.as_ref() { + std::fs::create_dir_all(&prepared.artifacts.compare_dir).map_err(|error| { + format!( + "create verification compare directory failed: {}: {error}", + prepared.artifacts.compare_dir.display() + ) + })?; + let comparison = crate::verification_only::compare_ccr_files( + &prepared.artifacts.source_ccr, + &prepared.artifacts.result_ccr, + )?; + crate::verification_only::write_ccr_comparison( + &prepared.artifacts.compare_dir.join("ccr-state-digest.json"), + Some(&prepared.artifacts.compare_dir.join("ccr-state-digest.md")), + &comparison, + )?; + let scratch_retained = !comparison.state_digest_match || args.keep_verification_scratch; + let source_run_id = prepared + .artifacts + .source_run_dir + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("unknown") + .to_string(); + crate::verification_only::write_verification_meta( + &prepared.artifacts.verification_meta, + &crate::verification_only::VerificationRunMeta { + schema_version: 1, + status: if comparison.state_digest_match { + "success".to_string() + } else { + "mismatch".to_string() + }, + source_run_id, + validation_time: crate::verification_only::format_validation_time(validation_time)?, + wall_ms: total_started.elapsed().as_millis() as u64, + repository_objects_checked: prepared.blob_summary.current_objects, + repository_bytes_checked: prepared.blob_summary.bytes_verified, + state_digest_match: comparison.state_digest_match, + scratch_retained, + }, + )?; + if !comparison.state_digest_match { + return Err(format!( + "verification-only CCR state digest mismatch: {}", + comparison.mismatched_states.join(",") + )); + } + if !args.keep_verification_scratch { + drop(shared); + drop(store); + std::fs::remove_dir_all(prepared.artifacts.output_dir.join("scratch")).map_err( + |error| format!("remove verification scratch directory failed: {error}"), + )?; + } + } Ok(()) } diff --git a/src/cli/tests.rs b/src/cli/tests.rs index 4d47d1f..471d06c 100644 --- a/src/cli/tests.rs +++ b/src/cli/tests.rs @@ -49,6 +49,63 @@ fn parse_rejects_unknown_argument() { assert!(err.contains("unknown argument"), "{err}"); } +#[test] +fn parse_accepts_verification_only_without_manual_validation_time() { + let argv = vec![ + "rpki".to_string(), + "--verification-only".to_string(), + "--verification-source-run".to_string(), + "/run/run_0002".to_string(), + "--verification-source-state".to_string(), + "/run/state".to_string(), + "--verification-out".to_string(), + "/verification/check_0002".to_string(), + ]; + let args = parse_args(&argv).expect("parse verification-only args"); + assert!(args.verification_only); + assert!(args.validation_time.is_none()); + assert_eq!( + args.db_path, + PathBuf::from("/verification/check_0002/scratch/work-db") + ); +} + +#[test] +fn parse_rejects_manual_validation_time_in_verification_only() { + let argv = vec![ + "rpki".to_string(), + "--verification-only".to_string(), + "--verification-source-run".to_string(), + "/run/run_0002".to_string(), + "--verification-source-state".to_string(), + "/run/state".to_string(), + "--verification-out".to_string(), + "/verification/check_0002".to_string(), + "--validation-time".to_string(), + "2026-07-16T00:00:00Z".to_string(), + ]; + let error = parse_args(&argv).expect_err("manual time must be rejected"); + assert!(error.contains("validation time"), "{error}"); +} + +#[test] +fn parse_accepts_normal_validation_contract_output() { + let argv = vec![ + "rpki".to_string(), + "--db".to_string(), + "db".to_string(), + "--tal-url".to_string(), + "https://example.test/root.tal".to_string(), + "--validation-contract-out".to_string(), + "out/validation-contract.json".to_string(), + ]; + let args = parse_args(&argv).expect("parse normal contract output"); + assert_eq!( + args.validation_contract_out_path, + Some(PathBuf::from("out/validation-contract.json")) + ); +} + #[test] fn parse_rejects_both_tal_url_and_tal_path() { let argv = vec![ diff --git a/src/fetch/current_repository.rs b/src/fetch/current_repository.rs new file mode 100644 index 0000000..5e494b2 --- /dev/null +++ b/src/fetch/current_repository.rs @@ -0,0 +1,178 @@ +use std::sync::Arc; + +use crate::fetch::rsync::{ + RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri, +}; +use crate::storage::{RepositoryViewState, RocksStore}; + +#[derive(Clone)] +pub struct CurrentRepositoryViewRsyncFetcher { + store: Arc, +} + +impl CurrentRepositoryViewRsyncFetcher { + pub fn new(store: Arc) -> Self { + Self { store } + } +} + +impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher { + fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult)>> { + let base = normalize_rsync_base_uri(rsync_base_uri); + let entries = self + .store + .list_repository_view_entries_with_prefix(&base) + .map_err(|error| { + RsyncFetchError::Fetch(format!( + "list frozen repository view failed for {base}: {error}" + )) + })?; + let mut objects = Vec::with_capacity(entries.len()); + for entry in entries { + if !matches!( + entry.state, + RepositoryViewState::Present | RepositoryViewState::Replaced + ) { + continue; + } + let bytes = self + .store + .load_current_object_bytes_by_uri(&entry.rsync_uri) + .map_err(|error| { + RsyncFetchError::Fetch(format!( + "load frozen repository object failed for {}: {error}", + entry.rsync_uri + )) + })? + .ok_or_else(|| { + RsyncFetchError::Fetch(format!( + "frozen repository object missing for {}", + entry.rsync_uri + )) + })?; + objects.push((entry.rsync_uri, bytes)); + } + objects.sort_by(|left, right| left.0.cmp(&right.0)); + if objects.is_empty() { + return Err(RsyncFetchError::Fetch(format!( + "frozen repository view contains no current objects under {base}" + ))); + } + Ok(objects) + } + + fn fetch_object(&self, rsync_uri: &str) -> RsyncFetchResult> { + self.store + .load_current_object_bytes_by_uri(rsync_uri) + .map_err(|error| { + RsyncFetchError::Fetch(format!( + "load frozen repository object failed for {rsync_uri}: {error}" + )) + })? + .ok_or_else(|| { + RsyncFetchError::Fetch(format!("frozen repository object not found: {rsync_uri}")) + }) + } + + fn dedup_key(&self, rsync_base_uri: &str) -> String { + normalize_rsync_base_uri(rsync_base_uri) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::{RepositoryViewEntry, RocksStore}; + + #[test] + fn current_repository_fetcher_returns_sorted_present_objects() { + let dir = tempfile::tempdir().expect("tempdir"); + let store = Arc::new(RocksStore::open(dir.path()).expect("open store")); + let entries = [ + ("rsync://example.test/repo/b.roa", b"b".as_slice()), + ("rsync://example.test/repo/a.mft", b"a".as_slice()), + ]; + for (uri, bytes) in entries { + let hash = hex::encode(crate::cir::sha256(bytes)); + store + .put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())]) + .expect("put blob"); + store + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: uri.to_string(), + repository_source: Some("fixture".to_string()), + object_type: Some("roa".to_string()), + state: RepositoryViewState::Present, + current_hash: Some(hash), + }) + .expect("put view"); + } + let fetcher = CurrentRepositoryViewRsyncFetcher::new(store); + let objects = fetcher + .fetch_objects("rsync://example.test/repo/") + .expect("fetch objects"); + assert_eq!( + objects + .iter() + .map(|(uri, _)| uri.as_str()) + .collect::>(), + vec![ + "rsync://example.test/repo/a.mft", + "rsync://example.test/repo/b.roa" + ] + ); + assert_eq!( + fetcher + .fetch_object("rsync://example.test/repo/a.mft") + .expect("fetch one"), + b"a" + ); + assert!( + fetcher + .fetch_object("rsync://example.test/repo/missing.roa") + .is_err() + ); + assert_eq!( + fetcher.dedup_key("rsync://example.test/repo"), + "rsync://example.test/repo/" + ); + assert!( + fetcher + .fetch_objects("rsync://empty.example/repo/") + .unwrap_err() + .to_string() + .contains("no current objects") + ); + } + + #[test] + fn current_repository_fetcher_ignores_withdrawn_and_reports_missing_blob() { + let dir = tempfile::tempdir().expect("tempdir"); + let store = Arc::new(RocksStore::open(dir.path()).expect("open store")); + store + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: "rsync://example.test/repo/withdrawn.roa".to_string(), + repository_source: Some("fixture".to_string()), + object_type: Some("roa".to_string()), + state: RepositoryViewState::Withdrawn, + current_hash: None, + }) + .expect("put withdrawn view"); + let missing_hash = "ab".repeat(32); + store + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: "rsync://example.test/repo/missing.roa".to_string(), + repository_source: Some("fixture".to_string()), + object_type: Some("roa".to_string()), + state: RepositoryViewState::Present, + current_hash: Some(missing_hash), + }) + .expect("put missing view"); + let fetcher = CurrentRepositoryViewRsyncFetcher::new(store); + let error = fetcher + .fetch_objects("rsync://example.test/repo/") + .unwrap_err() + .to_string(); + assert!(error.contains("blob bytes missing")); + } +} diff --git a/src/fetch/mod.rs b/src/fetch/mod.rs index 02e16b0..5f92b39 100644 --- a/src/fetch/mod.rs +++ b/src/fetch/mod.rs @@ -1,3 +1,4 @@ +pub mod current_repository; pub mod http; pub mod rsync; pub mod rsync_system; diff --git a/src/lib.rs b/src/lib.rs index f9e88d1..e85aa1c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,3 +42,5 @@ pub mod sync; pub mod tools; #[cfg(feature = "full")] pub mod validation; +#[cfg(feature = "full")] +pub mod verification_only; diff --git a/src/storage.rs b/src/storage.rs index 4ce04e2..9ad4e75 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -8,6 +8,7 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use base64::Engine; +use rocksdb::checkpoint::Checkpoint; use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch}; use serde::{Deserialize, Serialize}; @@ -53,6 +54,13 @@ pub enum StorageError { pub type StorageResult = Result; +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct RepositoryBlobVerificationSummary { + pub current_objects: u64, + pub bytes_verified: u64, + pub batches: u64, +} + #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct ChildCertificateCacheMmapLookup { pub projections: Vec>, @@ -2893,6 +2901,38 @@ fn write_publication_point_cache_projection_to_batch( } impl RocksStore { + pub fn create_read_only_checkpoint(source: &Path, destination: &Path) -> StorageResult<()> { + if destination.exists() { + return Err(StorageError::InvalidData { + entity: "work_db_checkpoint", + detail: format!( + "checkpoint destination already exists: {}", + destination.display() + ), + }); + } + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent) + .map_err(|error| StorageError::RocksDb(error.to_string()))?; + } + let mut options = Options::default(); + let blob_mode = work_db_blob_mode_from_env(); + let memory_profile = work_db_memory_profile_from_env(); + configure_work_db_options(&mut options, blob_mode, memory_profile); + let db = DB::open_cf_descriptors_read_only( + &options, + source, + column_family_descriptors_for_blob_mode(blob_mode), + false, + ) + .map_err(|error| StorageError::RocksDb(error.to_string()))?; + let checkpoint = + Checkpoint::new(&db).map_err(|error| StorageError::RocksDb(error.to_string()))?; + checkpoint + .create_checkpoint(destination) + .map_err(|error| StorageError::RocksDb(error.to_string())) + } + pub fn open(path: &Path) -> StorageResult { let mut base_opts = Options::default(); base_opts.create_if_missing(true); @@ -2931,6 +2971,15 @@ impl RocksStore { Self::open_with_external_stores(path, None, Some(repo_bytes_path)) } + pub fn open_with_external_repo_bytes_read_only( + path: &Path, + repo_bytes_path: &Path, + ) -> StorageResult { + let mut store = Self::open(path)?; + store.external_repo_bytes = Some(ExternalRepoBytesDb::open_read_only(repo_bytes_path)?); + Ok(store) + } + pub fn open_with_external_stores( path: &Path, raw_store_path: Option<&Path>, @@ -3147,6 +3196,53 @@ impl RocksStore { .collect() } + pub fn verify_current_repository_blobs( + &self, + batch_size: usize, + ) -> StorageResult { + if batch_size == 0 { + return Err(StorageError::InvalidData { + entity: "repository_blob_verification", + detail: "batch_size must be greater than zero".to_string(), + }); + } + let cf = self.cf(CF_REPOSITORY_VIEW)?; + let prefix = repository_view_prefix(""); + let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward); + let mut summary = RepositoryBlobVerificationSummary::default(); + let mut batch = Vec::with_capacity(batch_size); + for item in self.db.iterator_cf(cf, mode) { + let (key, value) = item.map_err(|error| StorageError::RocksDb(error.to_string()))?; + if !key.starts_with(prefix.as_bytes()) { + break; + } + let entry = decode_cbor::(&value, "repository_view")?; + entry.validate_internal()?; + if !matches!( + entry.state, + RepositoryViewState::Present | RepositoryViewState::Replaced + ) { + continue; + } + let hash = entry + .current_hash + .clone() + .ok_or(StorageError::InvalidData { + entity: "repository_blob_verification", + detail: format!("current hash missing for {}", entry.rsync_uri), + })?; + batch.push((entry.rsync_uri, hash)); + if batch.len() >= batch_size { + verify_repository_blob_batch(self, &batch, &mut summary)?; + batch.clear(); + } + } + if !batch.is_empty() { + verify_repository_blob_batch(self, &batch, &mut summary)?; + } + Ok(summary) + } + pub fn put_raw_by_hash_entry(&self, entry: &RawByHashEntry) -> StorageResult<()> { entry.validate_internal()?; if let Some(raw_store) = self.external_raw_store.as_ref() { @@ -3206,6 +3302,9 @@ impl RocksStore { return Ok(()); } if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { + if repo_bytes.is_read_only() { + return repo_bytes.require_existing_blob_bytes_batch(blobs); + } return repo_bytes.put_blob_bytes_batch(blobs); } if let Some(raw_store) = self.external_raw_store.as_ref() { @@ -4682,6 +4781,37 @@ impl RocksStore { } } +fn verify_repository_blob_batch( + store: &RocksStore, + batch: &[(String, String)], + summary: &mut RepositoryBlobVerificationSummary, +) -> StorageResult<()> { + let hashes = batch + .iter() + .map(|(_, hash)| hash.clone()) + .collect::>(); + let blobs = store.get_blob_bytes_batch(&hashes)?; + for ((uri, expected_hash), blob) in batch.iter().zip(blobs.into_iter()) { + let bytes = blob.as_ref().ok_or(StorageError::InvalidData { + entity: "repository_blob_verification", + detail: format!("blob missing for URI {uri} (hash={expected_hash})"), + })?; + let actual_hash = hex::encode(compute_sha256_32(bytes)); + if !actual_hash.eq_ignore_ascii_case(expected_hash) { + return Err(StorageError::InvalidData { + entity: "repository_blob_verification", + detail: format!( + "blob hash mismatch for URI {uri}: expected={expected_hash}, actual={actual_hash}" + ), + }); + } + summary.current_objects += 1; + summary.bytes_verified += bytes.len() as u64; + } + summary.batches += 1; + Ok(()) +} + #[cfg(test)] #[path = "storage/tests.rs"] mod tests; diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 0848e8d..bd18bd0 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -2276,3 +2276,121 @@ fn pack_file_can_lazy_load_bytes_from_external_repo_bytes_store() { assert_eq!(file.bytes_cloned().expect("cloned repo bytes"), bytes); assert_eq!(file.compute_sha256().expect("compute sha256"), file.sha256); } + +#[test] +fn read_only_checkpoint_isolated_from_source_work_db() { + let td = tempfile::tempdir().expect("tempdir"); + let source_path = td.path().join("source-work-db"); + let checkpoint_path = td.path().join("checkpoint-work-db"); + let uri = "rsync://example.test/repo/a.roa"; + let bytes = b"checkpoint-object"; + let hash = sha256_hex(bytes); + { + let source = RocksStore::open(&source_path).expect("open source"); + source + .put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())]) + .expect("put blob"); + source + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: uri.to_string(), + current_hash: Some(hash), + repository_source: Some("fixture".to_string()), + object_type: Some("roa".to_string()), + state: RepositoryViewState::Present, + }) + .expect("put view"); + } + + RocksStore::create_read_only_checkpoint(&source_path, &checkpoint_path) + .expect("create checkpoint"); + let checkpoint = RocksStore::open(&checkpoint_path).expect("open checkpoint"); + checkpoint + .delete_repository_view_entry(uri) + .expect("delete checkpoint entry"); + drop(checkpoint); + + let source = RocksStore::open(&source_path).expect("reopen source"); + assert!( + source + .get_repository_view_entry(uri) + .expect("get source entry") + .is_some() + ); +} + +#[test] +fn read_only_external_repo_bytes_rejects_writes() { + let td = tempfile::tempdir().expect("tempdir"); + let path = td.path().join("repo-bytes.db"); + let hash = sha256_hex(b"repo-object"); + { + let writable = ExternalRepoBytesDb::open(&path).expect("open writable repo bytes"); + writable + .put_blob_bytes_batch(&[(hash.clone(), b"repo-object".to_vec())]) + .expect("seed repo bytes"); + } + let read_only = ExternalRepoBytesDb::open_read_only(&path).expect("open read-only repo bytes"); + assert_eq!( + read_only + .get_blob_bytes(&hash) + .expect("read repo bytes") + .expect("blob"), + b"repo-object" + ); + assert!( + read_only + .put_blob_bytes_batch(&[(hash, b"repo-object".to_vec())]) + .is_err() + ); +} + +#[test] +fn read_only_store_accepts_only_idempotent_existing_blob_apply() { + let td = tempfile::tempdir().expect("tempdir"); + let work_db_path = td.path().join("work-db"); + let repo_bytes_path = td.path().join("repo-bytes.db"); + let bytes = b"repo-object".to_vec(); + let hash = sha256_hex(&bytes); + { + let repo_bytes = ExternalRepoBytesDb::open(&repo_bytes_path).expect("open repo bytes"); + repo_bytes + .put_blob_bytes_batch(&[(hash.clone(), bytes.clone())]) + .expect("seed repo bytes"); + } + let store = + RocksStore::open_with_external_repo_bytes_read_only(&work_db_path, &repo_bytes_path) + .expect("open read-only store"); + store + .put_blob_bytes_batch(&[(hash.clone(), bytes)]) + .expect("idempotent apply"); + assert!( + store + .put_blob_bytes_batch(&[(hash, b"different".to_vec())]) + .is_err() + ); +} + +#[test] +fn repository_blob_verification_detects_missing_external_blob() { + let td = tempfile::tempdir().expect("tempdir"); + let repo_bytes_path = td.path().join("repo-bytes.db"); + ExternalRepoBytesDb::open(&repo_bytes_path).expect("create repo bytes"); + let store = RocksStore::open_with_external_repo_bytes_read_only( + &td.path().join("work-db"), + &repo_bytes_path, + ) + .expect("open work db"); + store + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: "rsync://example.test/repo/missing.roa".to_string(), + current_hash: Some(sha256_hex(b"missing")), + repository_source: Some("fixture".to_string()), + object_type: Some("roa".to_string()), + state: RepositoryViewState::Present, + }) + .expect("put view"); + let error = store + .verify_current_repository_blobs(16) + .expect_err("missing blob must fail"); + assert!(error.to_string().contains("missing.roa"), "{error}"); +} diff --git a/src/verification_only.rs b/src/verification_only.rs new file mode 100644 index 0000000..9cc609e --- /dev/null +++ b/src/verification_only.rs @@ -0,0 +1,1040 @@ +use std::fs; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use time::format_description::well_known::Rfc3339; + +use crate::parallel::types::TalInputSpec; +use crate::policy::Policy; +use crate::storage::{RepositoryBlobVerificationSummary, RocksStore}; + +pub const VALIDATION_CONTRACT_SCHEMA_VERSION: u32 = 1; +pub const CURRENT_STATE_BINDING_SCHEMA_VERSION: u32 = 1; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ValidationCacheContract { + pub publication_point: bool, + pub roa: bool, + pub child_certificate: bool, + pub transport_prefetch: bool, +} + +impl ValidationCacheContract { + pub fn any_validation_cache_enabled(&self) -> bool { + self.publication_point || self.roa || self.child_certificate + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ValidationContract { + pub schema_version: u32, + pub validation_time: String, + pub binary_sha256: String, + pub policy: Policy, + pub max_ca_depth: usize, + pub max_instances: Option, + pub cache: ValidationCacheContract, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CurrentStateBinding { + pub schema_version: u32, + pub current_run_id: String, + pub source_cir_sha256: String, + pub source_ccr_sha256: String, + pub validation_contract_sha256: String, + pub updated_at_rfc3339_utc: String, +} + +impl CurrentStateBinding { + pub fn validate(&self) -> Result<(), String> { + if self.schema_version != CURRENT_STATE_BINDING_SCHEMA_VERSION { + return Err(format!( + "unsupported current state binding schemaVersion: {}", + self.schema_version + )); + } + if self.current_run_id.trim().is_empty() { + return Err("current state binding currentRunId must not be empty".to_string()); + } + validate_sha256_hex( + "current state binding sourceCirSha256", + &self.source_cir_sha256, + )?; + validate_sha256_hex( + "current state binding sourceCcrSha256", + &self.source_ccr_sha256, + )?; + validate_sha256_hex( + "current state binding validationContractSha256", + &self.validation_contract_sha256, + )?; + time::OffsetDateTime::parse(&self.updated_at_rfc3339_utc, &Rfc3339).map_err(|error| { + format!("invalid current state binding updatedAtRfc3339Utc: {error}") + })?; + Ok(()) + } +} + +impl ValidationContract { + pub fn for_current_binary( + validation_time: time::OffsetDateTime, + policy: Policy, + max_ca_depth: usize, + max_instances: Option, + cache: ValidationCacheContract, + ) -> Result { + Ok(Self { + schema_version: VALIDATION_CONTRACT_SCHEMA_VERSION, + validation_time: format_validation_time(validation_time)?, + binary_sha256: current_binary_sha256()?, + policy, + max_ca_depth, + max_instances, + cache, + }) + } + + pub fn validation_time(&self) -> Result { + time::OffsetDateTime::parse(&self.validation_time, &Rfc3339) + .map(|value| value.to_offset(time::UtcOffset::UTC)) + .map_err(|error| format!("invalid validation contract validationTime: {error}")) + } + + pub fn validate(&self) -> Result<(), String> { + if self.schema_version != VALIDATION_CONTRACT_SCHEMA_VERSION { + return Err(format!( + "unsupported validation contract schemaVersion: {}", + self.schema_version + )); + } + self.validation_time()?; + validate_sha256_hex("validation contract binarySha256", &self.binary_sha256)?; + if self.max_ca_depth == 0 { + return Err("validation contract maxCaDepth must be greater than zero".to_string()); + } + Ok(()) + } + + pub fn require_current_binary(&self) -> Result<(), String> { + let current = current_binary_sha256()?; + if current != self.binary_sha256 { + return Err(format!( + "verification-only binary differs from source run: source={}, current={current}", + self.binary_sha256 + )); + } + Ok(()) + } +} + +#[derive(Clone, Debug)] +pub struct VerificationArtifacts { + pub source_run_dir: PathBuf, + pub source_state_root: PathBuf, + pub output_dir: PathBuf, + pub source_cir: PathBuf, + pub source_ccr: PathBuf, + pub source_contract: PathBuf, + pub source_work_db: PathBuf, + pub source_repo_bytes_db: PathBuf, + pub scratch_work_db: PathBuf, + pub report_json: PathBuf, + pub result_cir: PathBuf, + pub result_ccr: PathBuf, + pub vrps_csv: PathBuf, + pub vaps_csv: PathBuf, + pub compare_dir: PathBuf, + pub source_binding: PathBuf, + pub source_run_summary: PathBuf, + pub result_contract: PathBuf, + pub verification_meta: PathBuf, +} + +impl VerificationArtifacts { + pub fn new( + source_run_dir: impl Into, + source_state_root: impl Into, + output_dir: impl Into, + ) -> Self { + let source_run_dir = source_run_dir.into(); + let source_state_root = source_state_root.into(); + let output_dir = output_dir.into(); + Self { + source_cir: source_run_dir.join("input.cir"), + source_ccr: source_run_dir.join("result.ccr"), + source_contract: source_run_dir.join("validation-contract.json"), + source_work_db: source_state_root.join("db/work-db"), + source_repo_bytes_db: source_state_root.join("db/repo-bytes.db"), + scratch_work_db: output_dir.join("scratch/work-db"), + report_json: output_dir.join("report.json"), + result_cir: output_dir.join("input.cir"), + result_ccr: output_dir.join("result.ccr"), + vrps_csv: output_dir.join("vrps.csv"), + vaps_csv: output_dir.join("vaps.csv"), + compare_dir: output_dir.join("compare"), + source_binding: source_state_root.join("meta/current-state-binding.json"), + source_run_summary: source_run_dir.join("run-summary.json"), + result_contract: output_dir.join("validation-contract.json"), + verification_meta: output_dir.join("verification-meta.json"), + source_run_dir, + source_state_root, + output_dir, + } + } +} + +pub struct PreparedVerification { + pub artifacts: VerificationArtifacts, + pub cir: crate::cir::CanonicalInputRepresentation, + pub contract: ValidationContract, + pub blob_summary: RepositoryBlobVerificationSummary, + pub store: Option>, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct VerificationCcrStateComparison { + pub name: String, + pub source_present: bool, + pub verification_present: bool, + pub source_hash_hex: Option, + pub verification_hash_hex: Option, + pub matches: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct VerificationCcrComparison { + pub state_digest_match: bool, + pub source_version: u32, + pub verification_version: u32, + pub source_hash_algorithm_oid: String, + pub verification_hash_algorithm_oid: String, + pub mismatched_states: Vec, + pub states: Vec, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct VerificationRunMeta { + pub schema_version: u32, + pub status: String, + pub source_run_id: String, + pub validation_time: String, + pub wall_ms: u64, + pub repository_objects_checked: u64, + pub repository_bytes_checked: u64, + pub state_digest_match: bool, + pub scratch_retained: bool, +} + +pub fn read_validation_contract(path: &Path) -> Result { + let bytes = fs::read(path).map_err(|error| { + format!( + "read validation contract failed: {}: {error}", + path.display() + ) + })?; + let contract: ValidationContract = serde_json::from_slice(&bytes).map_err(|error| { + format!( + "decode validation contract failed: {}: {error}", + path.display() + ) + })?; + contract.validate()?; + Ok(contract) +} + +pub fn read_current_state_binding(path: &Path) -> Result { + let bytes = fs::read(path).map_err(|error| { + format!( + "read current state binding failed: {}: {error}", + path.display() + ) + })?; + let binding: CurrentStateBinding = serde_json::from_slice(&bytes).map_err(|error| { + format!( + "decode current state binding failed: {}: {error}", + path.display() + ) + })?; + binding.validate()?; + Ok(binding) +} + +pub fn write_validation_contract(path: &Path, contract: &ValidationContract) -> Result<(), String> { + contract.validate()?; + let bytes = serde_json::to_vec_pretty(contract) + .map_err(|error| format!("encode validation contract failed: {error}"))?; + atomic_write(path, &bytes) +} + +pub fn read_cir(path: &Path) -> Result { + let bytes = fs::read(path) + .map_err(|error| format!("read source CIR failed: {}: {error}", path.display()))?; + crate::cir::decode_cir(&bytes) + .map_err(|error| format!("decode source CIR failed: {}: {error}", path.display())) +} + +pub fn tal_inputs_from_cir(cir: &crate::cir::CanonicalInputRepresentation) -> Vec { + cir.trust_anchors + .iter() + .map(|trust_anchor| { + TalInputSpec::from_ta_der( + trust_anchor.tal_uri.clone(), + trust_anchor.tal_bytes.clone(), + trust_anchor.ta_certificate_der.clone(), + ) + }) + .collect() +} + +pub fn cir_tal_uris(cir: &crate::cir::CanonicalInputRepresentation) -> Vec { + cir.trust_anchors + .iter() + .map(|trust_anchor| trust_anchor.tal_uri.clone()) + .collect() +} + +pub fn prepare_verification( + source_run_dir: &Path, + source_state_root: &Path, + output_dir: &Path, +) -> Result { + if std::env::var("RPKI_STATE_EXECUTION_LOCK_HELD").as_deref() != Ok("1") { + return Err( + "verification-only requires the state execution lock; use scripts/verification/run_verification_only.sh" + .to_string(), + ); + } + let artifacts = VerificationArtifacts::new(source_run_dir, source_state_root, output_dir); + validate_output_isolated(&artifacts)?; + validate_source_run_success(&artifacts.source_run_summary)?; + + let binding = read_current_state_binding(&artifacts.source_binding)?; + let source_run_id = artifacts + .source_run_dir + .file_name() + .and_then(|value| value.to_str()) + .ok_or_else(|| "source run directory must have a UTF-8 run id basename".to_string())?; + if binding.current_run_id != source_run_id { + return Err(format!( + "source run is not bound to the current state: requested={source_run_id}, current={}", + binding.current_run_id + )); + } + verify_bound_file( + "source CIR", + &artifacts.source_cir, + &binding.source_cir_sha256, + )?; + verify_bound_file( + "source CCR", + &artifacts.source_ccr, + &binding.source_ccr_sha256, + )?; + verify_bound_file( + "validation contract", + &artifacts.source_contract, + &binding.validation_contract_sha256, + )?; + + let cir = read_cir(&artifacts.source_cir)?; + let contract = read_validation_contract(&artifacts.source_contract)?; + if cir.validation_time.to_offset(time::UtcOffset::UTC) != contract.validation_time()? { + return Err(format!( + "source CIR validation time differs from validation contract: cir={}, contract={}", + format_validation_time(cir.validation_time)?, + contract.validation_time + )); + } + contract.require_current_binary()?; + if !contract.cache.any_validation_cache_enabled() { + return Err( + "source run did not enable PP, ROA, or child-certificate validation cache".to_string(), + ); + } + if !artifacts.source_work_db.is_dir() { + return Err(format!( + "source work-db is missing: {}", + artifacts.source_work_db.display() + )); + } + if !artifacts.source_repo_bytes_db.is_dir() { + return Err(format!( + "source repo-bytes.db is missing: {}", + artifacts.source_repo_bytes_db.display() + )); + } + + fs::create_dir_all(&artifacts.output_dir).map_err(|error| { + format!( + "create verification output directory failed: {}: {error}", + artifacts.output_dir.display() + ) + })?; + RocksStore::create_read_only_checkpoint(&artifacts.source_work_db, &artifacts.scratch_work_db) + .map_err(|error| format!("create frozen work-db checkpoint failed: {error}"))?; + let store = Arc::new( + RocksStore::open_with_external_repo_bytes_read_only( + &artifacts.scratch_work_db, + &artifacts.source_repo_bytes_db, + ) + .map_err(|error| format!("open frozen verification store failed: {error}"))?, + ); + let blob_summary = store + .verify_current_repository_blobs(1024) + .map_err(|error| format!("verify frozen repository blobs failed: {error}"))?; + write_validation_contract(&artifacts.result_contract, &contract)?; + + Ok(PreparedVerification { + artifacts, + cir, + contract, + blob_summary, + store: Some(store), + }) +} + +pub fn compare_ccr_files( + source_path: &Path, + verification_path: &Path, +) -> Result { + let source = fs::read(source_path) + .map_err(|error| format!("read source CCR failed: {}: {error}", source_path.display()))?; + let verification = fs::read(verification_path).map_err(|error| { + format!( + "read verification CCR failed: {}: {error}", + verification_path.display() + ) + })?; + let comparison = crate::ccr::compare_state_digests(&source, &verification) + .map_err(|error| format!("compare CCR state digests failed: {error}"))?; + let states = comparison + .states + .iter() + .map(|state| VerificationCcrStateComparison { + name: state.name.to_string(), + source_present: state.ours_present, + verification_present: state.peer_present, + source_hash_hex: state.ours_hash_hex.clone(), + verification_hash_hex: state.peer_hash_hex.clone(), + matches: state.matches, + }) + .collect::>(); + Ok(VerificationCcrComparison { + state_digest_match: comparison.matches(), + source_version: comparison.ours.version, + verification_version: comparison.peer.version, + source_hash_algorithm_oid: comparison.ours.hash_alg_oid.clone(), + verification_hash_algorithm_oid: comparison.peer.hash_alg_oid.clone(), + mismatched_states: comparison + .mismatched_state_names() + .into_iter() + .map(str::to_string) + .collect(), + states, + }) +} + +pub fn write_ccr_comparison( + json_path: &Path, + markdown_path: Option<&Path>, + comparison: &VerificationCcrComparison, +) -> Result<(), String> { + let json = serde_json::to_vec_pretty(comparison) + .map_err(|error| format!("encode CCR comparison failed: {error}"))?; + atomic_write(json_path, &json)?; + if let Some(markdown_path) = markdown_path { + let mut markdown = String::from( + "# Verification-only CCR Comparison\n\n| State | Source hash | Verification hash | Match |\n|---|---|---|---|\n", + ); + for state in &comparison.states { + markdown.push_str(&format!( + "| {} | {} | {} | {} |\n", + state.name, + state.source_hash_hex.as_deref().unwrap_or("absent"), + state.verification_hash_hex.as_deref().unwrap_or("absent"), + if state.matches { "yes" } else { "no" } + )); + } + markdown.push_str(&format!( + "\nOverall state digest match: **{}**\n", + if comparison.state_digest_match { + "yes" + } else { + "no" + } + )); + atomic_write(markdown_path, markdown.as_bytes())?; + } + Ok(()) +} + +pub fn write_verification_meta(path: &Path, meta: &VerificationRunMeta) -> Result<(), String> { + let bytes = serde_json::to_vec_pretty(meta) + .map_err(|error| format!("encode verification metadata failed: {error}"))?; + atomic_write(path, &bytes) +} + +#[derive(Clone)] +pub struct NetworkDisabledHttpFetcher; + +impl crate::sync::rrdp::Fetcher for NetworkDisabledHttpFetcher { + fn fetch(&self, uri: &str) -> Result, String> { + Err(format!( + "network access is disabled in verification-only mode: {uri}" + )) + } +} + +pub fn sha256_file(path: &Path) -> Result { + let bytes = fs::read(path) + .map_err(|error| format!("read file for SHA-256 failed: {}: {error}", path.display()))?; + Ok(hex::encode(Sha256::digest(bytes))) +} + +pub fn current_binary_sha256() -> Result { + let path = std::env::current_exe() + .map_err(|error| format!("resolve current executable failed: {error}"))?; + sha256_file(&path) +} + +pub fn format_validation_time(value: time::OffsetDateTime) -> Result { + value + .to_offset(time::UtcOffset::UTC) + .format(&Rfc3339) + .map_err(|error| format!("format validation time failed: {error}")) +} + +fn validate_sha256_hex(label: &str, value: &str) -> Result<(), String> { + if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(format!( + "{label} must be a 64-character hexadecimal SHA-256" + )); + } + Ok(()) +} + +fn validate_source_run_success(path: &Path) -> Result<(), String> { + let bytes = fs::read(path).map_err(|error| { + format!( + "read source run summary failed: {}: {error}", + path.display() + ) + })?; + let value: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { + format!( + "decode source run summary failed: {}: {error}", + path.display() + ) + })?; + if value.get("status").and_then(serde_json::Value::as_str) != Some("success") { + return Err(format!("source run is not successful: {}", path.display())); + } + Ok(()) +} + +fn validate_output_isolated(artifacts: &VerificationArtifacts) -> Result<(), String> { + if artifacts.output_dir.exists() { + let mut entries = fs::read_dir(&artifacts.output_dir).map_err(|error| { + format!( + "read verification output directory failed: {}: {error}", + artifacts.output_dir.display() + ) + })?; + if entries.next().is_some() { + return Err(format!( + "verification output directory must be absent or empty: {}", + artifacts.output_dir.display() + )); + } + } + if artifacts.output_dir.starts_with(&artifacts.source_run_dir) + || artifacts.source_run_dir.starts_with(&artifacts.output_dir) + { + return Err("verification output must be independent from the source run directory".into()); + } + if let Some(run_root) = artifacts.source_state_root.parent() { + if artifacts.output_dir.starts_with(run_root.join("runs")) { + return Err("verification output must not use the normal runs directory".into()); + } + } + Ok(()) +} + +fn verify_bound_file(label: &str, path: &Path, expected_sha256: &str) -> Result<(), String> { + let actual = sha256_file(path)?; + if actual != expected_sha256 { + return Err(format!( + "{label} differs from current state binding: path={}, expected={}, actual={actual}", + path.display(), + expected_sha256 + )); + } + Ok(()) +} + +fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), String> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .map_err(|error| format!("create directory failed: {}: {error}", parent.display()))?; + } + let tmp = path.with_extension(format!( + "{}.tmp", + path.extension() + .and_then(|value| value.to_str()) + .unwrap_or("json") + )); + let mut file = fs::File::create(&tmp) + .map_err(|error| format!("create temporary file failed: {}: {error}", tmp.display()))?; + file.write_all(bytes) + .map_err(|error| format!("write temporary file failed: {}: {error}", tmp.display()))?; + file.write_all(b"\n") + .map_err(|error| format!("write temporary file failed: {}: {error}", tmp.display()))?; + file.sync_all() + .map_err(|error| format!("sync temporary file failed: {}: {error}", tmp.display()))?; + fs::rename(&tmp, path).map_err(|error| { + format!( + "rename temporary file failed: {} -> {}: {error}", + tmp.display(), + path.display() + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ccr::{ + CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation, + build_roa_payload_state, encode_content_info, + }; + use crate::cir::{CanonicalInputRepresentation, CirTrustAnchor, encode_cir}; + use crate::data_model::roa::{IpPrefix, RoaAfi}; + use crate::sync::rrdp::Fetcher; + use crate::validation::objects::Vrp; + use std::sync::Mutex; + + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + struct SourceFixture { + _temp: tempfile::TempDir, + source_run: PathBuf, + state_root: PathBuf, + output: PathBuf, + } + + fn sample_contract() -> ValidationContract { + ValidationContract { + schema_version: VALIDATION_CONTRACT_SCHEMA_VERSION, + validation_time: "2026-07-16T00:00:00Z".to_string(), + binary_sha256: "ab".repeat(32), + policy: Policy::default(), + max_ca_depth: 32, + max_instances: None, + cache: ValidationCacheContract { + publication_point: true, + roa: true, + child_certificate: true, + transport_prefetch: false, + }, + } + } + + fn sample_cir() -> CanonicalInputRepresentation { + let ta_der = b"verification-only-test-ta".to_vec(); + CanonicalInputRepresentation::new_v4( + time::OffsetDateTime::parse( + "2026-07-16T00:00:00Z", + &time::format_description::well_known::Rfc3339, + ) + .expect("validation time"), + Vec::new(), + Vec::new(), + vec![CirTrustAnchor { + ta_rsync_uri: "rsync://example.test/ta.cer".to_string(), + tal_uri: "https://example.test/ta.tal".to_string(), + tal_bytes: b"rsync://example.test/ta.cer\n\nAQID\n".to_vec(), + ta_certificate_sha256: crate::cir::sha256(&ta_der), + ta_certificate_der: ta_der, + }], + Vec::new(), + Vec::new(), + ) + } + + fn sample_ccr(asn: u32) -> Vec { + let vrps = build_roa_payload_state(&[Vrp { + asn, + prefix: IpPrefix { + afi: RoaAfi::Ipv4, + prefix_len: 24, + addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + }, + max_length: 24, + }]) + .expect("build VRP state"); + encode_content_info(&CcrContentInfo::new(RpkiCanonicalCacheRepresentation { + version: 0, + hash_alg: CcrDigestAlgorithm::Sha256, + produced_at: time::OffsetDateTime::UNIX_EPOCH, + mfts: None, + vrps: Some(vrps), + vaps: None, + tas: None, + rks: None, + })) + .expect("encode CCR") + } + + fn refresh_binding(fixture: &SourceFixture) { + let binding = CurrentStateBinding { + schema_version: CURRENT_STATE_BINDING_SCHEMA_VERSION, + current_run_id: "run_0001".to_string(), + source_cir_sha256: sha256_file(&fixture.source_run.join("input.cir")) + .expect("hash CIR"), + source_ccr_sha256: sha256_file(&fixture.source_run.join("result.ccr")) + .expect("hash CCR"), + validation_contract_sha256: sha256_file( + &fixture.source_run.join("validation-contract.json"), + ) + .expect("hash contract"), + updated_at_rfc3339_utc: "2026-07-16T00:01:00Z".to_string(), + }; + fs::create_dir_all(fixture.state_root.join("meta")).expect("create meta"); + fs::write( + fixture.state_root.join("meta/current-state-binding.json"), + serde_json::to_vec_pretty(&binding).expect("encode binding"), + ) + .expect("write binding"); + } + + fn source_fixture() -> SourceFixture { + let temp = tempfile::tempdir().expect("tempdir"); + let source_run = temp.path().join("runs/run_0001"); + let state_root = temp.path().join("state"); + let output = temp.path().join("verification/run_0001"); + fs::create_dir_all(&source_run).expect("create source run"); + fs::create_dir_all(state_root.join("db")).expect("create state db root"); + let store = RocksStore::open_with_external_repo_bytes( + &state_root.join("db/work-db"), + &state_root.join("db/repo-bytes.db"), + ) + .expect("create source databases"); + drop(store); + fs::write( + source_run.join("input.cir"), + encode_cir(&sample_cir()).expect("encode CIR"), + ) + .expect("write CIR"); + fs::write(source_run.join("result.ccr"), sample_ccr(64496)).expect("write CCR"); + fs::write( + source_run.join("run-summary.json"), + br#"{"status":"success"}"#, + ) + .expect("write summary"); + let contract = ValidationContract::for_current_binary( + sample_cir().validation_time, + Policy::default(), + 32, + None, + ValidationCacheContract { + publication_point: true, + roa: false, + child_certificate: false, + transport_prefetch: true, + }, + ) + .expect("contract"); + write_validation_contract(&source_run.join("validation-contract.json"), &contract) + .expect("write contract"); + let fixture = SourceFixture { + _temp: temp, + source_run, + state_root, + output, + }; + refresh_binding(&fixture); + fixture + } + + fn prepare_error(fixture: &SourceFixture) -> String { + match prepare_verification(&fixture.source_run, &fixture.state_root, &fixture.output) { + Ok(_) => panic!("verification preparation unexpectedly succeeded"), + Err(error) => error, + } + } + + #[test] + fn validation_contract_roundtrips() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("validation-contract.json"); + let expected = sample_contract(); + write_validation_contract(&path, &expected).expect("write contract"); + assert_eq!( + read_validation_contract(&path).expect("read contract"), + expected + ); + } + + #[test] + fn validation_contract_rejects_invalid_time_and_hash() { + let mut contract = sample_contract(); + contract.validation_time = "not-a-time".to_string(); + assert!(contract.validate().is_err()); + contract.validation_time = "2026-07-16T00:00:00Z".to_string(); + contract.binary_sha256 = "bad".to_string(); + assert!(contract.validate().is_err()); + } + + #[test] + fn cache_contract_distinguishes_prefetch_from_validation_caches() { + let cache = ValidationCacheContract { + publication_point: false, + roa: false, + child_certificate: false, + transport_prefetch: true, + }; + assert!(!cache.any_validation_cache_enabled()); + } + + #[test] + fn validation_contract_and_binding_validation_cover_rejection_paths() { + let time = time::OffsetDateTime::parse( + "2026-07-16T00:00:00Z", + &time::format_description::well_known::Rfc3339, + ) + .expect("time"); + let contract = ValidationContract::for_current_binary( + time, + Policy::default(), + 32, + Some(10), + sample_contract().cache, + ) + .expect("current contract"); + assert_eq!(contract.validation_time().expect("contract time"), time); + contract.validate().expect("valid contract"); + contract.require_current_binary().expect("same binary"); + + let mut invalid = contract.clone(); + invalid.schema_version += 1; + assert!(invalid.validate().unwrap_err().contains("schemaVersion")); + invalid = contract.clone(); + invalid.max_ca_depth = 0; + assert!(invalid.validate().unwrap_err().contains("maxCaDepth")); + invalid = contract.clone(); + invalid.binary_sha256 = "00".repeat(32); + assert!(invalid.require_current_binary().is_err()); + + let valid_binding = CurrentStateBinding { + schema_version: CURRENT_STATE_BINDING_SCHEMA_VERSION, + current_run_id: "run_0001".to_string(), + source_cir_sha256: "11".repeat(32), + source_ccr_sha256: "22".repeat(32), + validation_contract_sha256: "33".repeat(32), + updated_at_rfc3339_utc: "2026-07-16T00:00:00Z".to_string(), + }; + valid_binding.validate().expect("valid binding"); + for mutate in 0..6 { + let mut binding = valid_binding.clone(); + match mutate { + 0 => binding.schema_version += 1, + 1 => binding.current_run_id.clear(), + 2 => binding.source_cir_sha256 = "bad".into(), + 3 => binding.source_ccr_sha256 = "bad".into(), + 4 => binding.validation_contract_sha256 = "bad".into(), + _ => binding.updated_at_rfc3339_utc = "bad".into(), + } + assert!(binding.validate().is_err()); + } + } + + #[test] + fn artifact_helpers_read_write_compare_and_disable_network() { + let temp = tempfile::tempdir().expect("tempdir"); + let source = temp.path().join("runs/run_0001"); + let state = temp.path().join("state"); + let output = temp.path().join("verification/run_0001"); + let artifacts = VerificationArtifacts::new(&source, &state, &output); + assert_eq!(artifacts.source_cir, source.join("input.cir")); + assert_eq!( + artifacts.source_binding, + state.join("meta/current-state-binding.json") + ); + assert_eq!(artifacts.result_ccr, output.join("result.ccr")); + + fs::create_dir_all(&source).expect("create source"); + let cir = sample_cir(); + fs::write(&artifacts.source_cir, encode_cir(&cir).expect("encode CIR")).expect("write CIR"); + let decoded = read_cir(&artifacts.source_cir).expect("read CIR"); + assert_eq!(cir_tal_uris(&decoded), vec!["https://example.test/ta.tal"]); + let tal_inputs = tal_inputs_from_cir(&decoded); + assert_eq!(tal_inputs.len(), 1); + + let first = temp.path().join("first.ccr"); + let second = temp.path().join("second.ccr"); + fs::write(&first, sample_ccr(64496)).expect("first CCR"); + fs::write(&second, sample_ccr(64496)).expect("second CCR"); + let equal = compare_ccr_files(&first, &second).expect("equal compare"); + assert!(equal.state_digest_match); + let json = temp.path().join("compare/result.json"); + let markdown = temp.path().join("compare/result.md"); + write_ccr_comparison(&json, Some(&markdown), &equal).expect("write compare"); + assert!( + fs::read_to_string(json) + .expect("json") + .contains("stateDigestMatch") + ); + assert!( + fs::read_to_string(markdown) + .expect("md") + .contains("Overall state") + ); + + fs::write(&second, sample_ccr(64497)).expect("different CCR"); + let unequal = compare_ccr_files(&first, &second).expect("unequal compare"); + assert!(!unequal.state_digest_match); + assert_eq!(unequal.mismatched_states, vec!["vrps"]); + + let meta = VerificationRunMeta { + schema_version: 1, + status: "success".to_string(), + source_run_id: "run_0001".to_string(), + validation_time: "2026-07-16T00:00:00Z".to_string(), + wall_ms: 12, + repository_objects_checked: 3, + repository_bytes_checked: 4, + state_digest_match: true, + scratch_retained: false, + }; + let meta_path = temp.path().join("meta/result.json"); + write_verification_meta(&meta_path, &meta).expect("write meta"); + assert!( + fs::read_to_string(meta_path) + .expect("meta") + .contains("run_0001") + ); + assert!( + NetworkDisabledHttpFetcher + .fetch("https://example.test") + .is_err() + ); + assert_eq!(sha256_file(&first).expect("hash").len(), 64); + assert!(read_cir(&first).is_err()); + } + + #[test] + fn verification_preflight_rejects_stale_inputs_then_prepares_frozen_store() { + let _guard = ENV_LOCK.lock().expect("env lock"); + let fixture = source_fixture(); + unsafe { std::env::remove_var("RPKI_STATE_EXECUTION_LOCK_HELD") }; + assert!(prepare_error(&fixture).contains("state execution lock")); + unsafe { std::env::set_var("RPKI_STATE_EXECUTION_LOCK_HELD", "1") }; + + fs::create_dir_all(&fixture.output).expect("output"); + fs::write(fixture.output.join("occupied"), b"x").expect("occupied"); + assert!(prepare_error(&fixture).contains("absent or empty")); + fs::remove_dir_all(&fixture.output).expect("remove output"); + + fs::write( + fixture.source_run.join("run-summary.json"), + br#"{"status":"failed"}"#, + ) + .expect("failed summary"); + assert!(prepare_error(&fixture).contains("not successful")); + fs::write( + fixture.source_run.join("run-summary.json"), + br#"{"status":"success"}"#, + ) + .expect("success summary"); + + let binding_path = fixture.state_root.join("meta/current-state-binding.json"); + let mut binding = read_current_state_binding(&binding_path).expect("binding"); + binding.current_run_id = "run_0002".to_string(); + fs::write( + &binding_path, + serde_json::to_vec_pretty(&binding).expect("binding JSON"), + ) + .expect("write binding"); + assert!(prepare_error(&fixture).contains("not bound")); + refresh_binding(&fixture); + + fs::write(fixture.source_run.join("input.cir"), b"tampered").expect("tamper CIR"); + assert!(prepare_error(&fixture).contains("differs from current state binding")); + fs::write( + fixture.source_run.join("input.cir"), + encode_cir(&sample_cir()).expect("encode CIR"), + ) + .expect("restore CIR"); + refresh_binding(&fixture); + + let mut contract = + read_validation_contract(&fixture.source_run.join("validation-contract.json")) + .expect("contract"); + contract.cache.publication_point = false; + contract.cache.transport_prefetch = true; + write_validation_contract( + &fixture.source_run.join("validation-contract.json"), + &contract, + ) + .expect("write no-cache contract"); + refresh_binding(&fixture); + assert!(prepare_error(&fixture).contains("did not enable")); + + contract.cache.roa = true; + write_validation_contract( + &fixture.source_run.join("validation-contract.json"), + &contract, + ) + .expect("restore cache contract"); + refresh_binding(&fixture); + let mut prepared = + prepare_verification(&fixture.source_run, &fixture.state_root, &fixture.output) + .expect("prepare verification"); + assert_eq!(prepared.blob_summary.current_objects, 0); + assert_eq!(prepared.blob_summary.bytes_verified, 0); + assert_eq!(prepared.contract.validation_time, "2026-07-16T00:00:00Z"); + assert_eq!(cir_tal_uris(&prepared.cir).len(), 1); + assert!(prepared.artifacts.scratch_work_db.is_dir()); + assert!(prepared.artifacts.result_contract.is_file()); + drop(prepared.store.take()); + unsafe { std::env::remove_var("RPKI_STATE_EXECUTION_LOCK_HELD") }; + } + + #[test] + fn private_validation_helpers_report_bad_files_and_paths() { + let temp = tempfile::tempdir().expect("tempdir"); + let summary = temp.path().join("summary.json"); + fs::write(&summary, b"not-json").expect("summary"); + assert!(validate_source_run_success(&summary).is_err()); + assert!(validate_source_run_success(&temp.path().join("missing")).is_err()); + + let file = temp.path().join("value"); + fs::write(&file, b"value").expect("file"); + assert!(verify_bound_file("value", &file, &"00".repeat(32)).is_err()); + assert!(verify_bound_file("missing", &temp.path().join("missing"), "").is_err()); + + let binding = temp.path().join("binding.json"); + fs::write(&binding, b"{}").expect("binding"); + assert!(read_current_state_binding(&binding).is_err()); + assert!(read_validation_contract(&binding).is_err()); + + let source = temp.path().join("normal/runs/run_0001"); + let state = temp.path().join("normal/state"); + let nested_output = source.join("verification"); + let nested = VerificationArtifacts::new(&source, &state, &nested_output); + assert!(validate_output_isolated(&nested).is_err()); + let normal_output = temp.path().join("normal/runs/run_verify"); + let normal = VerificationArtifacts::new(&source, &state, &normal_output); + assert!(validate_output_isolated(&normal).is_err()); + } +}