NASP/rpki
75
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260716 增加verification-only缓存正确性复核

Browse Source
This commit is contained in:
yuyr 2026-07-16 19:08:16 +08:00
parent 7aac4855bd
commit 59623b449b
14 changed files with 2269 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
scripts/soak/build_portable_soak_package.sh
View File

@ -14,7 +14,7 @@ Usage:
scripts/soak/build_portable_soak_package.sh [--out-dir <path>] [--profile <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"

69
scripts/soak/run_soak.sh
View File

@ -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

50
scripts/verification/README.md Normal file
View File

@ -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，按不匹配状态集合继续排查。

102
scripts/verification/run_verification_only.sh Executable file
View File

@ -0,0 +1,102 @@
#!/usr/bin/env bash
set -euo pipefail
usage() {
cat <<'USAGE'
Usage:
run_verification_only.sh \
--source-run-dir <run_dir> \
--source-state-root <state_dir> \
--out-dir <verification_dir> \
[--rpki-bin <path>] [--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"

171
src/bin/verification_ccr_compare.rs Normal file
View File

@ -0,0 +1,171 @@
use std::path::PathBuf;
fn usage() -> &'static str {
"Usage: verification_ccr_compare --source-ccr <path> --verification-ccr <path> --out-json <path> [--out-md <path>]"
}
#[derive(Debug, PartialEq, Eq)]
struct Args {
source_ccr: PathBuf,
verification_ccr: PathBuf,
out_json: PathBuf,
out_md: Option<PathBuf>,
}
fn parse_args(argv: &[String]) -> Result<Args, String> {
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::<Vec<_>>();
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<String> {
std::iter::once("verification_ccr_compare")
.chain(values.iter().copied())
.map(str::to_string)
.collect()
}
fn sample_ccr(asn: u32) -> Vec<u8> {
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")
);
}
}

45
src/blob_store.rs
View File

@ -77,6 +77,7 @@ pub struct ExternalRawStoreDb {
pub struct ExternalRepoBytesDb {
path: PathBuf,
db: Arc<DB>,
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<u8>)],
) -> StorageResult<()> {
if blobs.is_empty() {
return Ok(());
}
let hashes = blobs
.iter()
.map(|(sha256_hex, _)| sha256_hex.clone())
.collect::<Vec<_>>();
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<Option<Vec<u8>>> {
validate_blob_sha256_hex(sha256_hex)?;
let key = repo_bytes_key(sha256_hex);

303
src/cli.rs
View File

@ -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<PathBuf>,
pub verification_source_state_root: Option<PathBuf>,
pub verification_out_dir: Option<PathBuf>,
pub keep_verification_scratch: bool,
pub validation_contract_out_path: Option<PathBuf>,
pub tal_urls: Vec<String>,
pub tal_paths: Vec<PathBuf>,
pub ta_paths: Vec<PathBuf>,
@ -184,8 +190,19 @@ fn usage() -> String {
Usage:
{bin} --db <path> --tal-url <url> [--tal-url <url> ...] [options]
{bin} --db <path> --tal-path <path> --ta-path <path> [--tal-path <path> --ta-path <path> ...] [options]
{bin} --verification-only --verification-source-run <path> --verification-source-state <path> --verification-out <path> [options]
Options:
--verification-only Revalidate the latest successful run without validation caches or network access
--verification-source-run <path>
Source normal run directory containing input.cir/result.ccr/validation-contract.json
--verification-source-state <path>
Source state root containing db/work-db and db/repo-bytes.db
--verification-out <path> 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 <path>
Write the effective normal-run validation contract (not valid in verification-only mode)
--db <path> RocksDB directory path (required)
--raw-store-db <path> External raw-by-hash store DB path (optional)
--repo-bytes-db <path> External repo object bytes DB path (optional)
@ -273,6 +290,12 @@ Options:
}
pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let mut verification_only = false;
let mut verification_source_run_dir: Option<PathBuf> = None;
let mut verification_source_state_root: Option<PathBuf> = None;
let mut verification_out_dir: Option<PathBuf> = None;
let mut keep_verification_scratch = false;
let mut validation_contract_out_path: Option<PathBuf> = None;
let mut tal_urls: Vec<String> = Vec::new();
let mut tal_paths: Vec<PathBuf> = Vec::new();
let mut ta_paths: Vec<PathBuf> = Vec::new();
@ -333,6 +356,39 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
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<CliArgs, String> {
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<CliArgs, String> {
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<CliArgs, String> {
}
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,9 +2153,54 @@ 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())?;
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;
}
@ -2022,9 +2210,31 @@ pub fn run(argv: &[String]) -> Result<(), String> {
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::<Result<Vec<_>, _>>()?;
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(())
}

57
src/cli/tests.rs
View File

@ -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![

178
src/fetch/current_repository.rs Normal file
View File

@ -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<RocksStore>,
}
impl CurrentRepositoryViewRsyncFetcher {
pub fn new(store: Arc<RocksStore>) -> Self {
Self { store }
}
}
impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher {
fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>> {
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<Vec<u8>> {
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<_>>(),
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"));
}
}

1
src/fetch/mod.rs
View File

@ -1,3 +1,4 @@
pub mod current_repository;
pub mod http;
pub mod rsync;
pub mod rsync_system;

2
src/lib.rs
View File

@ -42,3 +42,5 @@ pub mod sync;
pub mod tools;
#[cfg(feature = "full")]
pub mod validation;
#[cfg(feature = "full")]
pub mod verification_only;

130
src/storage.rs
View File

@ -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<T> = Result<T, StorageError>;
#[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<Option<ChildCertificateCacheProjection>>,
@ -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<Self> {
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<Self> {
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<RepositoryBlobVerificationSummary> {
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::<RepositoryViewEntry>(&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::<Vec<_>>();
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;

118
src/storage/tests.rs
View File

@ -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}");
}

1040
src/verification_only.rs Normal file
View File

File diff suppressed because it is too large Load Diff