20260330 完成live bundle录制，远程录制，以及与routinator/rpki-client replay对比

2026-03-31 17:34:32 +08:00
parent cd0ba15286
commit 6edc420ce2
19 changed files with 4059 additions and 22 deletions
--- a/src/bin/replay_bundle_capture_delta.rs
+++ b/src/bin/replay_bundle_capture_delta.rs
@@ -0,0 +1,531 @@
+use rpki::bundle::{
+    BundleManifest, BundleManifestEntry, RecordingHttpFetcher,
+    RecordingRsyncFetcher, build_vap_compare_rows, build_vrp_compare_rows,
+    write_live_delta_replay_bundle_inputs, write_vap_csv, write_vrp_csv,
+};
+use rpki::ccr::{build_ccr_from_run, decode_content_info, verify_content_info, write_ccr_file};
+use rpki::fetch::http::{BlockingHttpFetcher, HttpFetcherConfig};
+use rpki::fetch::rsync_system::{SystemRsyncConfig, SystemRsyncFetcher};
+use rpki::policy::Policy;
+use rpki::storage::RocksStore;
+use rpki::sync::rrdp::Fetcher;
+use rpki::validation::run_tree_from_tal::{
+    run_tree_from_tal_and_ta_der_payload_delta_replay_serial_audit,
+    run_tree_from_tal_and_ta_der_payload_replay_serial_audit,
+    run_tree_from_tal_and_ta_der_serial_audit,
+};
+use rpki::validation::tree::TreeRunConfig;
+use sha2::Digest;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::time::Instant;
+use time::format_description::well_known::Rfc3339;
+
+#[derive(Debug, Default, PartialEq, Eq)]
+struct Args {
+    rir: Option<String>,
+    base_bundle_dir: Option<PathBuf>,
+    out_dir: Option<PathBuf>,
+    validation_time: Option<time::OffsetDateTime>,
+    http_timeout_secs: u64,
+    rsync_timeout_secs: u64,
+    rsync_mirror_root: Option<PathBuf>,
+    max_depth: Option<usize>,
+    max_instances: Option<usize>,
+    trust_anchor: Option<String>,
+}
+
+fn usage() -> &'static str {
+    "Usage: replay_bundle_capture_delta --rir <name> --base-bundle-dir <path> --out-dir <path> [--validation-time <rfc3339>] [--http-timeout-secs <n>] [--rsync-timeout-secs <n>] [--rsync-mirror-root <path>] [--max-depth <n>] [--max-instances <n>] [--trust-anchor <name>]"
+}
+
+fn parse_args(argv: &[String]) -> Result<Args, String> {
+    let mut args = Args {
+        http_timeout_secs: 20,
+        rsync_timeout_secs: 60,
+        ..Args::default()
+    };
+    let mut i = 1usize;
+    while i < argv.len() {
+        match argv[i].as_str() {
+            "--help" | "-h" => return Err(usage().to_string()),
+            "--rir" => {
+                i += 1;
+                args.rir = Some(argv.get(i).ok_or("--rir requires a value")?.clone());
+            }
+            "--base-bundle-dir" => {
+                i += 1;
+                args.base_bundle_dir =
+                    Some(PathBuf::from(argv.get(i).ok_or("--base-bundle-dir requires a value")?));
+            }
+            "--out-dir" => {
+                i += 1;
+                args.out_dir = Some(PathBuf::from(argv.get(i).ok_or("--out-dir requires a value")?));
+            }
+            "--validation-time" => {
+                i += 1;
+                let value = argv.get(i).ok_or("--validation-time requires a value")?;
+                args.validation_time = Some(
+                    time::OffsetDateTime::parse(value, &Rfc3339)
+                        .map_err(|e| format!("invalid --validation-time: {e}"))?,
+                );
+            }
+            "--http-timeout-secs" => {
+                i += 1;
+                args.http_timeout_secs = argv
+                    .get(i)
+                    .ok_or("--http-timeout-secs requires a value")?
+                    .parse()
+                    .map_err(|e| format!("invalid --http-timeout-secs: {e}"))?;
+            }
+            "--rsync-timeout-secs" => {
+                i += 1;
+                args.rsync_timeout_secs = argv
+                    .get(i)
+                    .ok_or("--rsync-timeout-secs requires a value")?
+                    .parse()
+                    .map_err(|e| format!("invalid --rsync-timeout-secs: {e}"))?;
+            }
+            "--rsync-mirror-root" => {
+                i += 1;
+                args.rsync_mirror_root =
+                    Some(PathBuf::from(argv.get(i).ok_or("--rsync-mirror-root requires a value")?));
+            }
+            "--max-depth" => {
+                i += 1;
+                args.max_depth = Some(
+                    argv.get(i)
+                        .ok_or("--max-depth requires a value")?
+                        .parse()
+                        .map_err(|e| format!("invalid --max-depth: {e}"))?,
+                );
+            }
+            "--max-instances" => {
+                i += 1;
+                args.max_instances = Some(
+                    argv.get(i)
+                        .ok_or("--max-instances requires a value")?
+                        .parse()
+                        .map_err(|e| format!("invalid --max-instances: {e}"))?,
+                );
+            }
+            "--trust-anchor" => {
+                i += 1;
+                args.trust_anchor = Some(argv.get(i).ok_or("--trust-anchor requires a value")?.clone());
+            }
+            other => return Err(format!("unknown argument: {other}\n{}", usage())),
+        }
+        i += 1;
+    }
+
+    if args.rir.is_none() {
+        return Err(format!("--rir is required\n{}", usage()));
+    }
+    if args.base_bundle_dir.is_none() {
+        return Err(format!("--base-bundle-dir is required\n{}", usage()));
+    }
+    if args.out_dir.is_none() {
+        return Err(format!("--out-dir is required\n{}", usage()));
+    }
+    Ok(args)
+}
+
+fn sha256_hex(bytes: &[u8]) -> String {
+    hex::encode(sha2::Sha256::digest(bytes))
+}
+
+fn write_json(path: &Path, value: &impl serde::Serialize) -> Result<(), String> {
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent)
+            .map_err(|e| format!("create parent failed: {}: {e}", parent.display()))?;
+    }
+    let bytes = serde_json::to_vec_pretty(value).map_err(|e| e.to_string())?;
+    fs::write(path, bytes).map_err(|e| format!("write json failed: {}: {e}", path.display()))
+}
+
+fn copy_dir_all(src: &Path, dst: &Path) -> Result<(), String> {
+    fs::create_dir_all(dst)
+        .map_err(|e| format!("create directory failed: {}: {e}", dst.display()))?;
+    for entry in fs::read_dir(src).map_err(|e| format!("read_dir failed: {}: {e}", src.display()))? {
+        let entry = entry.map_err(|e| format!("read_dir entry failed: {}: {e}", src.display()))?;
+        let ty = entry
+            .file_type()
+            .map_err(|e| format!("file_type failed: {}: {e}", entry.path().display()))?;
+        let to = dst.join(entry.file_name());
+        if ty.is_dir() {
+            copy_dir_all(&entry.path(), &to)?;
+        } else if ty.is_file() {
+            if let Some(parent) = to.parent() {
+                fs::create_dir_all(parent)
+                    .map_err(|e| format!("create parent failed: {}: {e}", parent.display()))?;
+            }
+            fs::copy(entry.path(), &to)
+                .map_err(|e| format!("copy failed: {} -> {}: {e}", entry.path().display(), to.display()))?;
+        }
+    }
+    Ok(())
+}
+
+fn load_validation_time(path: &Path) -> Result<time::OffsetDateTime, String> {
+    let json: serde_json::Value = serde_json::from_slice(
+        &fs::read(path).map_err(|e| format!("read json failed: {}: {e}", path.display()))?,
+    )
+    .map_err(|e| format!("parse json failed: {}: {e}", path.display()))?;
+    let value = json
+        .get("validationTime")
+        .or_else(|| json.get("validation_time"))
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| format!("validationTime missing in {}", path.display()))?;
+    time::OffsetDateTime::parse(value, &Rfc3339)
+        .map_err(|e| format!("invalid validationTime in {}: {e}", path.display()))
+}
+
+fn ensure_recorded_target_snapshots(
+    store: &RocksStore,
+    base_bundle_dir: &Path,
+    http: &RecordingHttpFetcher<BlockingHttpFetcher>,
+) -> Result<(), String> {
+    let base_locks: serde_json::Value = serde_json::from_slice(
+        &fs::read(base_bundle_dir.join("base-locks.json"))
+            .map_err(|e| format!("read base locks failed: {e}"))?,
+    )
+    .map_err(|e| format!("parse base locks failed: {e}"))?;
+    let base_rrdp = base_locks
+        .get("rrdp")
+        .and_then(|v| v.as_object())
+        .cloned()
+        .unwrap_or_default();
+
+    for (notify_uri, base_lock) in base_rrdp {
+        let Some(base_transport) = base_lock.get("transport").and_then(|v| v.as_str()) else {
+            continue;
+        };
+        if base_transport != "rrdp" {
+            continue;
+        }
+        let Some(base_session) = base_lock.get("session").and_then(|v| v.as_str()) else {
+            continue;
+        };
+        let Some(base_serial) = base_lock.get("serial").and_then(|v| v.as_u64()) else {
+            continue;
+        };
+        let Some(record) = store
+            .get_rrdp_source_record(&notify_uri)
+            .map_err(|e| format!("read rrdp source record failed for {notify_uri}: {e}"))?
+        else {
+            continue;
+        };
+        let Some(target_session) = record.last_session_id.as_deref() else {
+            continue;
+        };
+        let Some(target_serial) = record.last_serial else {
+            continue;
+        };
+        if target_session != base_session || target_serial <= base_serial {
+            continue;
+        }
+        let Some(snapshot_uri) = record.last_snapshot_uri.as_deref() else {
+            continue;
+        };
+        if http.snapshot_responses().contains_key(snapshot_uri) {
+            continue;
+        }
+        let _ = http
+            .fetch(snapshot_uri)
+            .map_err(|e| format!("fetch target snapshot for {notify_uri} failed: {e}"))?;
+    }
+    Ok(())
+}
+
+fn run(args: Args) -> Result<PathBuf, String> {
+    let rir = args.rir.as_ref().unwrap();
+    let rir_normalized = rir.to_ascii_lowercase();
+    let out_root = args.out_dir.as_ref().unwrap();
+    let base_root = args.base_bundle_dir.as_ref().unwrap();
+    let base_rir_dir = base_root.join(&rir_normalized);
+    if !base_rir_dir.is_dir() {
+        return Err(format!("base bundle rir dir not found: {}", base_rir_dir.display()));
+    }
+    if out_root.exists() {
+        fs::remove_dir_all(out_root)
+            .map_err(|e| format!("remove old out dir failed: {}: {e}", out_root.display()))?;
+    }
+    copy_dir_all(base_root, out_root)?;
+    let rir_dir = out_root.join(&rir_normalized);
+
+    let trust_anchor = args
+        .trust_anchor
+        .clone()
+        .unwrap_or_else(|| rir_normalized.clone());
+    let tal_bytes =
+        fs::read(rir_dir.join("tal.tal")).map_err(|e| format!("read tal from base bundle failed: {e}"))?;
+    let ta_bytes =
+        fs::read(rir_dir.join("ta.cer")).map_err(|e| format!("read ta from base bundle failed: {e}"))?;
+    let base_validation_time = load_validation_time(&rir_dir.join("base-locks.json"))?;
+    let target_validation_time = args.validation_time.unwrap_or_else(time::OffsetDateTime::now_utc);
+
+    let target_store_dir = out_root.join(".tmp").join(format!("{rir}-live-target-db"));
+    let self_replay_dir = out_root.join(".tmp").join(format!("{rir}-self-delta-db"));
+    let _ = fs::remove_dir_all(&target_store_dir);
+    let _ = fs::remove_dir_all(&self_replay_dir);
+    if let Some(parent) = target_store_dir.parent() {
+        fs::create_dir_all(parent)
+            .map_err(|e| format!("create tmp dir failed: {}: {e}", parent.display()))?;
+    }
+    let target_store =
+        RocksStore::open(&target_store_dir).map_err(|e| format!("open target rocksdb failed: {e}"))?;
+
+    let _base = run_tree_from_tal_and_ta_der_payload_replay_serial_audit(
+        &target_store,
+        &Policy::default(),
+        &tal_bytes,
+        &ta_bytes,
+        None,
+        &rir_dir.join("base-payload-archive"),
+        &rir_dir.join("base-locks.json"),
+        base_validation_time,
+        &TreeRunConfig {
+            max_depth: args.max_depth,
+            max_instances: args.max_instances,
+        },
+    )
+    .map_err(|e| format!("base bootstrap replay failed: {e}"))?;
+
+    let http = RecordingHttpFetcher::new(
+        BlockingHttpFetcher::new(HttpFetcherConfig {
+            timeout: std::time::Duration::from_secs(args.http_timeout_secs),
+            ..HttpFetcherConfig::default()
+        })
+        .map_err(|e| format!("create http fetcher failed: {e}"))?,
+    );
+    let rsync = RecordingRsyncFetcher::new(SystemRsyncFetcher::new(SystemRsyncConfig {
+        timeout: std::time::Duration::from_secs(args.rsync_timeout_secs),
+        mirror_root: args.rsync_mirror_root.clone(),
+        ..SystemRsyncConfig::default()
+    }));
+
+    let started = Instant::now();
+    let target_out = run_tree_from_tal_and_ta_der_serial_audit(
+        &target_store,
+        &Policy::default(),
+        &tal_bytes,
+        &ta_bytes,
+        None,
+        &http,
+        &rsync,
+        target_validation_time,
+        &TreeRunConfig {
+            max_depth: args.max_depth,
+            max_instances: args.max_instances,
+        },
+    )
+    .map_err(|e| format!("live target run failed: {e}"))?;
+    let duration = started.elapsed();
+    ensure_recorded_target_snapshots(&target_store, &rir_dir, &http)?;
+
+    let delta_ccr = build_ccr_from_run(
+        &target_store,
+        &[target_out.discovery.trust_anchor.clone()],
+        &target_out.tree.vrps,
+        &target_out.tree.aspas,
+        &target_out.tree.router_keys,
+        target_validation_time,
+    )
+    .map_err(|e| format!("build delta ccr failed: {e}"))?;
+    let delta_ccr_path = rir_dir.join("delta.ccr");
+    write_ccr_file(&delta_ccr_path, &delta_ccr).map_err(|e| format!("write delta ccr failed: {e}"))?;
+    let delta_ccr_bytes =
+        fs::read(&delta_ccr_path).map_err(|e| format!("read delta ccr failed: {}: {e}", delta_ccr_path.display()))?;
+    let delta_decoded =
+        decode_content_info(&delta_ccr_bytes).map_err(|e| format!("decode delta ccr failed: {e}"))?;
+    let delta_verify =
+        verify_content_info(&delta_decoded).map_err(|e| format!("verify delta ccr failed: {e}"))?;
+
+    let delta_vrp_rows = build_vrp_compare_rows(&target_out.tree.vrps, &trust_anchor);
+    let delta_vap_rows = build_vap_compare_rows(&target_out.tree.aspas, &trust_anchor);
+    let (ccr_vrps, ccr_vaps) = rpki::bundle::decode_ccr_compare_views(&delta_decoded, &trust_anchor)?;
+    if delta_vrp_rows != ccr_vrps {
+        return Err("record-delta.csv compare view does not match delta.ccr".to_string());
+    }
+    if delta_vap_rows != ccr_vaps {
+        return Err("record-delta-vaps.csv compare view does not match delta.ccr".to_string());
+    }
+    write_vrp_csv(&rir_dir.join("record-delta.csv"), &delta_vrp_rows)?;
+    write_vap_csv(&rir_dir.join("record-delta-vaps.csv"), &delta_vap_rows)?;
+
+    let capture = write_live_delta_replay_bundle_inputs(
+        &rir_dir,
+        &rir_normalized,
+        target_validation_time,
+        &target_out.publication_points,
+        &target_store,
+        &http.snapshot_responses(),
+        &rsync.snapshot_fetches(),
+    )?;
+
+    let self_store =
+        RocksStore::open(&self_replay_dir).map_err(|e| format!("open self replay db failed: {e}"))?;
+    let replay_out = run_tree_from_tal_and_ta_der_payload_delta_replay_serial_audit(
+        &self_store,
+        &Policy::default(),
+        &tal_bytes,
+        &ta_bytes,
+        None,
+        &rir_dir.join("base-payload-archive"),
+        &rir_dir.join("base-locks.json"),
+        &rir_dir.join("payload-delta-archive"),
+        &rir_dir.join("locks-delta.json"),
+        base_validation_time,
+        target_validation_time,
+        &TreeRunConfig {
+            max_depth: args.max_depth,
+            max_instances: args.max_instances,
+        },
+    )
+    .map_err(|e| format!("self delta replay failed: {e}"))?;
+    let replay_vrps = build_vrp_compare_rows(&replay_out.tree.vrps, &trust_anchor);
+    let replay_vaps = build_vap_compare_rows(&replay_out.tree.aspas, &trust_anchor);
+    if replay_vrps != delta_vrp_rows {
+        return Err("self delta replay VRP compare view mismatch".to_string());
+    }
+    if replay_vaps != delta_vap_rows {
+        return Err("self delta replay VAP compare view mismatch".to_string());
+    }
+
+    fs::create_dir_all(rir_dir.join("timings"))
+        .map_err(|e| format!("create timings dir failed: {e}"))?;
+    write_json(
+        &rir_dir.join("timings").join("delta-produce.json"),
+        &serde_json::json!({
+            "mode": "delta",
+            "validationTime": target_validation_time
+                .format(&Rfc3339)
+                .map_err(|e| format!("format validation time failed: {e}"))?,
+            "durationSeconds": duration.as_secs_f64(),
+        }),
+    )?;
+
+    let mut bundle_json: serde_json::Value = serde_json::from_slice(
+        &fs::read(rir_dir.join("bundle.json")).map_err(|e| format!("read base bundle.json failed: {e}"))?,
+    )
+    .map_err(|e| format!("parse base bundle.json failed: {e}"))?;
+    bundle_json["deltaValidationTime"] = serde_json::Value::String(
+        target_validation_time
+            .format(&Rfc3339)
+            .map_err(|e| format!("format delta validation time failed: {e}"))?,
+    );
+    bundle_json["deltaCcrSha256"] = serde_json::Value::String(sha256_hex(&delta_ccr_bytes));
+    bundle_json["deltaVrpCount"] = serde_json::Value::from(delta_vrp_rows.len() as u64);
+    bundle_json["deltaVapCount"] = serde_json::Value::from(delta_vap_rows.len() as u64);
+    bundle_json["hasAspa"] = serde_json::Value::Bool(
+        bundle_json
+            .get("hasAspa")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false)
+            || !delta_vap_rows.is_empty(),
+    );
+    bundle_json["hasRouterKey"] = serde_json::Value::Bool(
+        bundle_json
+            .get("hasRouterKey")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false)
+            || delta_verify.router_key_count > 0,
+    );
+    write_json(&rir_dir.join("bundle.json"), &bundle_json)?;
+
+    let mut verification_json: serde_json::Value = serde_json::from_slice(
+        &fs::read(rir_dir.join("verification.json"))
+            .map_err(|e| format!("read base verification.json failed: {e}"))?,
+    )
+    .map_err(|e| format!("parse base verification.json failed: {e}"))?;
+    verification_json["delta"] = serde_json::json!({
+        "validationTime": target_validation_time
+            .format(&Rfc3339)
+            .map_err(|e| format!("format delta validation time failed: {e}"))?,
+        "ccr": {
+            "path": "delta.ccr",
+            "sha256": sha256_hex(&delta_ccr_bytes),
+            "stateHashesOk": delta_verify.state_hashes_ok,
+            "manifestInstances": delta_verify.manifest_instances,
+            "roaVrpCount": delta_verify.roa_vrp_count,
+            "aspaPayloadSets": delta_verify.aspa_payload_sets,
+            "routerKeyCount": delta_verify.router_key_count,
+        },
+        "compareViews": {
+            "vrpsSelfMatch": true,
+            "vapsSelfMatch": true,
+            "deltaVrpCount": delta_vrp_rows.len(),
+            "deltaVapCount": delta_vap_rows.len(),
+        },
+        "capture": {
+            "captureId": capture.capture_id,
+            "rrdpRepoCount": capture.rrdp_repo_count,
+            "rsyncModuleCount": capture.rsync_module_count,
+            "selfReplayOk": true,
+        }
+    });
+    write_json(&rir_dir.join("verification.json"), &verification_json)?;
+
+    let bundle_manifest = BundleManifest {
+        schema_version: "20260330-v1".to_string(),
+        bundle_producer: "ours".to_string(),
+        recorded_at_rfc3339_utc: time::OffsetDateTime::now_utc()
+            .format(&Rfc3339)
+            .map_err(|e| format!("format recorded_at failed: {e}"))?,
+        rirs: vec![rir_normalized.clone()],
+        per_rir_bundles: vec![BundleManifestEntry {
+            rir: rir_normalized.clone(),
+            relative_path: rir_normalized,
+            base_validation_time: base_validation_time
+                .format(&Rfc3339)
+                .map_err(|e| format!("format base validation time failed: {e}"))?,
+            delta_validation_time: Some(
+                target_validation_time
+                    .format(&Rfc3339)
+                    .map_err(|e| format!("format delta validation time failed: {e}"))?,
+            ),
+            has_aspa: bundle_json["hasAspa"].as_bool().unwrap_or(false),
+        }],
+    };
+    write_json(&out_root.join("bundle-manifest.json"), &bundle_manifest)?;
+
+    let _ = fs::remove_dir_all(&target_store_dir);
+    let _ = fs::remove_dir_all(&self_replay_dir);
+
+    Ok(out_root.clone())
+}
+
+fn main() -> Result<(), String> {
+    let args = parse_args(&std::env::args().collect::<Vec<_>>())?;
+    let out = run(args)?;
+    println!("{}", out.display());
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_args_requires_required_flags() {
+        let argv = vec![
+            "replay_bundle_capture_delta".to_string(),
+            "--rir".to_string(),
+            "apnic".to_string(),
+            "--base-bundle-dir".to_string(),
+            "base".to_string(),
+            "--out-dir".to_string(),
+            "out".to_string(),
+        ];
+        let args = parse_args(&argv).expect("parse");
+        assert_eq!(args.rir.as_deref(), Some("apnic"));
+        assert_eq!(args.base_bundle_dir.as_deref(), Some(Path::new("base")));
+        assert_eq!(args.out_dir.as_deref(), Some(Path::new("out")));
+    }
+
+    #[test]
+    fn parse_args_rejects_missing_requireds() {
+        let err = parse_args(&["replay_bundle_capture_delta".to_string()]).unwrap_err();
+        assert!(err.contains("--rir is required"), "{err}");
+    }
+}