diff --git a/CHANGELOG.md b/CHANGELOG.md index edf668b..746c016 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## Unreleased + +- Detect RRDP session desynchronization from persisted notification delta hashes + (RFC 9697), warn and recover using the current snapshot before applying deltas. +- Select automatic TA certificates using `draft-ietf-sidrops-rpki-ta-tiebreaker-06`: + prefer later notBefore, then earlier notAfter, then the newly fetched certificate. + A healthy preference for cache preserves a successful validation result. + ## 0.1.0 — 2026-09-29 - Vendor five OpenBSD TALs and constraints with pinned provenance; enable policies diff --git a/README.md b/README.md index 6492565..1619592 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,8 @@ For continuous operation see [Usage](docs/usage.md#continuous-operation). - Multiple named anchors, fixed or automatically refreshed trust anchors, and optional per-anchor resource constraints. - RRDP snapshot, delta, replace/withdraw and protocol fallback. +- RRDP session desynchronization detection (RFC 9697) and automatic TA + certificate selection (TA tiebreaker draft -06; unreleased). - HTTPS same-origin checks for RRDP references and redirects. - Independent bounded repository and object worker pools. - CA, CRL, manifest, ROA, ASPA and BGPsec router-certificate processing. diff --git a/docs/configuration.md b/docs/configuration.md index 02e8e54..894a563 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -156,10 +156,19 @@ no separate background refresh timer, and no invented HTTPS alternative for an rsync-only TAL. Candidates must pass certificate and TAL-binding validation at the configured logical validation time. +When a fetched candidate and a revalidated cached certificate are both usable, +selection follows `draft-ietf-sidrops-rpki-ta-tiebreaker-06` §2: prefer the later +`notBefore`; if equal, prefer the earlier `notAfter`; if both are equal, use the +fetched candidate. Comparisons use the signed certificate fields. Selecting a +preferred cached certificate after a successful fetch is healthy, including +under `require-fresh`: `trustAnchors[].source` is `cache`, `refreshFailed` is false, +and `selectionReason` explains the choice. It does not increment +`taCacheFallbackCount` or make the run partial. + The cache is `.ta-cache`, keyed by the TAL public-key SHA-256. Download failure permits fallback only under allow-valid-cache with a usable, revalidated cached certificate; that makes the result at least partial. -Require-fresh prohibits fallback. Existing fixed `.cer` files do not seed this +Require-fresh prohibits fallback after a failed refresh. Existing fixed `.cer` files do not seed this cache automatically; switching to automatic mode initially needs reachable URIs. One failed TA with others completing yields partial; all TAs failing yields failed. A fixed-certificate error found during preflight exits 2 immediately. diff --git a/docs/conformance-matrix.md b/docs/conformance-matrix.md index 0db3228..a55049d 100644 --- a/docs/conformance-matrix.md +++ b/docs/conformance-matrix.md @@ -1,5 +1,22 @@ # v0.1.0 conformance matrix +## Unreleased additions + +| Behaviour | Standard reference | Implementation | Evidence | +| --- | --- | --- | --- | +| Detect changed hashes for overlapping delta serials within a session, warn and recover with a snapshot | [RFC 9697 §§3–5](https://www.rfc-editor.org/rfc/rfc9697.html) | `src/repository/sync/rrdp/`, `src/repository/storage/models_summary.rs` | `tests_parts/desynchronization.rs`; `tests/security_recovery_e2e.rs` | +| Choose between valid fetched and cached TA certificates by later notBefore, earlier notAfter, then freshly fetched | [TA tiebreaker draft -06 §2](https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-rpki-ta-tiebreaker-06) | `src/cli/anchors.rs` | `anchors_tiebreaker_tests.rs`; `tests/security_recovery_e2e.rs` | + +The RRDP hash window is the last successfully processed notification's delta +references, retained across restarts. An old database without this window learns +it on its next successful sync. Failed recovery keeps the previous baseline. +No historical delta download is required for comparison. TA selection compares +revalidated DER certificates; fixed TA inputs retain their explicit selection. +The tiebreaker is a draft implementation. Failed-refresh fallback still follows +`ta-refresh-policy`; see [configuration](configuration.md#5-anchor-fields). + +## Released v0.1.0 baseline + | Behaviour | Standard reference | Implementation | Evidence | | --- | --- | --- | --- | | TAL syntax and TA public-key binding | RFC 8630 §2 | `src/validation/from_tal.rs`, `src/model/ta.rs` | parser and synthetic integration tests | @@ -29,6 +46,6 @@ revalidation of the entire trust chain by another relying party. The worker pool and interval index are implementation mechanisms, not new wire protocols; their tests must also prove deterministic output and unchanged -validation decisions. Only the table entries above are in v0.1.0 scope. +validation decisions. Only the released baseline table is in v0.1.0 scope. Adding a new source file or protocol feature requires a new row with a document revision, section, test, and reviewer decision. diff --git a/docs/usage.md b/docs/usage.md index 57f2ab1..8de9c4e 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -36,6 +36,14 @@ TAL public-key hash. Refresh failures may use that cache under `allow-valid-cache` and produce `partial`; use `require-fresh` to reject the fallback. +Automatic TA refresh compares usable fetched and cached certificates using the +[TA tiebreaker rules](configuration.md#5-anchor-fields). A preferred cached +certificate after a successful fetch keeps the run successful. RRDP also checks +the previous notification's delta hashes (RFC 9697); a changed hash within the +same session emits `rrdp_session_desynchronized` and triggers a fresh snapshot, +including when the notification serial has not advanced. This check uses stored +hashes and does not fetch old deltas. + ## Multiple anchors and constraints Use one named `[[anchors]]` table per RIR. There is no positional TAL/TA pairing: diff --git a/src/cli/anchors.rs b/src/cli/anchors.rs index a37fd16..f117e65 100644 --- a/src/cli/anchors.rs +++ b/src/cli/anchors.rs @@ -44,6 +44,54 @@ pub(crate) fn cache_root(state: &Path) -> PathBuf { PathBuf::from(name) } +struct CachedTa { + bytes: Vec, + info: Value, + fetched_at: Value, +} + +fn load_cached_ta( + path: &Path, + tal: &Tal, + strict_name: bool, + validation_time: OffsetDateTime, +) -> Result { + let cached: Value = fs::read(path) + .map_err(|error| format!("read cached TA metadata failed: {error}")) + .and_then(|bytes| serde_json::from_slice(&bytes).map_err(|error| error.to_string()))?; + let encoded = cached["der"] + .as_str() + .ok_or("cached TA metadata has no DER value")?; + let bytes = + hex::decode(encoded).map_err(|error| format!("decode cached TA failed: {error}"))?; + let info = validate_certificate(tal, &bytes, strict_name, validation_time)?; + Ok(CachedTa { + bytes, + info, + fetched_at: cached["record"]["fetchedAt"].clone(), + }) +} + +/// draft-ietf-sidrops-rpki-ta-tiebreaker-06 §2. Both inputs must already +/// pass the current TAL binding, certificate and validity checks. Compare DER +/// fields, never the unauthenticated display dates in the JSON cache record. +fn prefer_download(cached: &[u8], downloaded: &[u8]) -> Result<(bool, &'static str), String> { + use std::cmp::Ordering; + let (_, old) = X509Certificate::from_der(cached).map_err(|error| error.to_string())?; + let (_, new) = X509Certificate::from_der(downloaded).map_err(|error| error.to_string())?; + Ok( + match new.validity().not_before.cmp(&old.validity().not_before) { + Ordering::Greater => (true, "newer_not_before"), + Ordering::Less => (false, "cached_newer_not_before"), + Ordering::Equal => match new.validity().not_after.cmp(&old.validity().not_after) { + Ordering::Less => (true, "shorter_validity"), + Ordering::Greater => (false, "cached_shorter_validity"), + Ordering::Equal => (true, "most_recently_fetched"), + }, + }, + ) +} + pub(crate) fn resolve_at( config: &Config, validation_time: OffsetDateTime, @@ -99,7 +147,7 @@ fn resolve_with_at( let mut inputs = Vec::new(); let mut records = Vec::new(); for anchor in &config.anchors { - let mut record = json!({"id":anchor.id,"mode":if anchor.ta.is_some(){"fixed"}else{"auto"},"source":Value::Null,"certificateSha256":Value::Null,"fetchedAt":Value::Null,"notBefore":Value::Null,"notAfter":Value::Null,"refreshFailed":false,"errorCode":Value::Null}); + let mut record = json!({"id":anchor.id,"mode":if anchor.ta.is_some(){"fixed"}else{"auto"},"source":Value::Null,"certificateSha256":Value::Null,"fetchedAt":Value::Null,"notBefore":Value::Null,"notAfter":Value::Null,"refreshFailed":false,"errorCode":Value::Null,"selectionReason":Value::Null}); let attempt = (|| -> Result { let tal = Tal::decode_bytes(&fs::read(&anchor.tal).map_err(|e| e.to_string())?) .map_err(|e| e.to_string())?; @@ -115,11 +163,15 @@ fn resolve_with_at( .unwrap() .extend(info.as_object().unwrap().clone()); record["source"] = json!("file"); + record["selectionReason"] = json!("fixed_input"); return Ok(path.clone()); } let key = hex::encode(Sha256::digest(&tal.subject_public_key_info_der)); let path = root.join(format!("{key}.json")); let der_path = root.join(format!("{key}.cer")); + // A corrupt/expired/mismatched cache cannot veto a valid download. + // Keep its error for the existing failed-refresh fallback path. + let cached = load_cached_ta(&path, &tal, strict_name, validation_time); let mut last_error = "no usable TA URI".to_string(); let mut error_code = "download_failed"; for uri in &tal.ta_uris { @@ -127,6 +179,29 @@ fn resolve_with_at( Ok(bytes) => { match validate_certificate(&tal, &bytes, strict_name, validation_time) { Ok(info) => { + let (download, reason) = match &cached { + Ok(old) => prefer_download(&old.bytes, &bytes)?, + Err(_) => (true, "no_valid_cache"), + }; + record["selectionReason"] = json!(reason); + if !download { + let old = cached.as_ref().map_err(|error| error.to_string())?; + record + .as_object_mut() + .unwrap() + .extend(old.info.as_object().unwrap().clone()); + record["source"] = json!("cache"); + record["fetchedAt"] = old.fetched_at.clone(); + // Keep the authoritative cache and its actual fetch time; + // repair only the runtime DER projection if necessary. + super::publication::write_atomic(&der_path, &old.bytes)?; + crate::logging::emit( + crate::logging::Level::Info, + "ta_certificate_selected", + || json!({"tal_id":anchor.id,"source":"cache","reason":reason}), + ); + return Ok(der_path); + } record .as_object_mut() .unwrap() @@ -155,40 +230,17 @@ fn resolve_with_at( record["refreshFailed"] = json!(true); record["errorCode"] = json!(error_code); if config.ta_refresh_policy == "allow-valid-cache" && path.exists() { - let cached: Value = match fs::read(&path) - .map_err(|e| e.to_string()) - .and_then(|bytes| serde_json::from_slice(&bytes).map_err(|e| e.to_string())) - { - Ok(value) => value, - Err(error) => { - record["errorCode"] = json!("cache_unusable"); - return Err(format!("read cached TA metadata failed: {error}")); - } - }; - let bytes = match cached["der"].as_str() { - Some(encoded) => match hex::decode(encoded) { - Ok(bytes) => bytes, - Err(error) => { - record["errorCode"] = json!("cache_unusable"); - return Err(format!("decode cached TA failed: {error}")); - } - }, - None => { - record["errorCode"] = json!("cache_unusable"); - return Err("cached TA metadata has no DER value".into()); - } - }; - let info = validate_certificate(&tal, &bytes, strict_name, validation_time) - .inspect_err(|_| { - record["errorCode"] = json!("cache_unusable"); - })?; + let cached = cached.inspect_err(|_| { + record["errorCode"] = json!("cache_unusable"); + })?; record .as_object_mut() .unwrap() - .extend(info.as_object().unwrap().clone()); + .extend(cached.info.as_object().unwrap().clone()); record["source"] = json!("cache"); - record["fetchedAt"] = cached["record"]["fetchedAt"].clone(); - super::publication::write_atomic(&der_path, &bytes)?; + record["fetchedAt"] = cached.fetched_at; + record["selectionReason"] = json!("refresh_failed_valid_cache"); + super::publication::write_atomic(&der_path, &cached.bytes)?; crate::logging::emit( crate::logging::Level::Warn, "ta_refresh_failed", @@ -217,6 +269,10 @@ fn resolve_with_at( Ok((inputs, records)) } +#[cfg(test)] +#[path = "anchors_tiebreaker_tests.rs"] +mod tiebreaker_tests; + #[cfg(test)] mod tests { use super::{cache_root, resolve_with}; diff --git a/src/cli/anchors_tiebreaker_tests.rs b/src/cli/anchors_tiebreaker_tests.rs new file mode 100644 index 0000000..93eaa4e --- /dev/null +++ b/src/cli/anchors_tiebreaker_tests.rs @@ -0,0 +1,207 @@ +use super::*; +use crate::cli::config::Anchor; + +fn auto_config(state: &Path) -> Config { + Config { + config_version: 1, + rrdp_state_dir: Some(state.to_path_buf()), + validation_time: Some("2026-06-01T00:00:00Z".into()), + anchors: vec![Anchor { + id: "custom".into(), + tal: crate::test_support::synthetic_repository().tal(), + ta: None, + constraints: None, + }], + ..Default::default() + } +} + +fn variant(name: &str) -> Vec { + let fixture = crate::test_support::synthetic_repository(); + fs::read(if name == "base" { + fixture.trust_anchor() + } else { + fixture.ta_variant(name) + }) + .unwrap() +} + +fn cache_path(config: &Config, extension: &str) -> PathBuf { + let tal = Tal::decode_bytes(&fs::read(&config.anchors[0].tal).unwrap()).unwrap(); + cache_root(config.rrdp_state_dir.as_ref().unwrap()).join(format!( + "{}.{}", + hex::encode(Sha256::digest(&tal.subject_public_key_info_der)), + extension + )) +} + +#[test] +fn tiebreaker_older_download_cannot_replace_newer_cached_ta() { + let temp = tempfile::tempdir().unwrap(); + let config = auto_config(&temp.path().join("state")); + let cached = variant("newer"); + resolve_with(&config, |_| Ok(cached.clone())).unwrap(); + let cache_before = fs::read(cache_path(&config, "json")).unwrap(); + let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + assert_eq!(inputs.len(), 1); + assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), cached); + assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), cache_before); + assert_eq!(records[0]["source"], "cache"); + assert_eq!(records[0]["refreshFailed"], false); +} + +#[test] +fn tiebreaker_equal_start_keeps_shorter_cached_validity() { + let temp = tempfile::tempdir().unwrap(); + let config = auto_config(&temp.path().join("state")); + let cached = variant("shorter"); + resolve_with(&config, |_| Ok(cached.clone())).unwrap(); + resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), cached); +} + +#[test] +fn tiebreaker_download_wins_each_precedence_level() { + for (old, new, reason) in [ + ("shorter", "newer", "newer_not_before"), + ("longer", "base", "shorter_validity"), + ("base", "equal-period", "most_recently_fetched"), + ("base", "base", "most_recently_fetched"), + ] { + let temp = tempfile::tempdir().unwrap(); + let config = auto_config(&temp.path().join("state")); + resolve_with(&config, |_| Ok(variant(old))).unwrap(); + let selected = variant(new); + let (inputs, records) = resolve_with(&config, |_| Ok(selected.clone())).unwrap(); + assert_eq!(inputs.len(), 1, "{old} -> {new}"); + assert_eq!(records[0]["source"], "download"); + assert_eq!(records[0]["selectionReason"], reason); + assert_eq!(records[0]["refreshFailed"], false); + assert_eq!(fs::read(cache_path(&config, "cer")).unwrap(), selected); + let saved: Value = + serde_json::from_slice(&fs::read(cache_path(&config, "json")).unwrap()).unwrap(); + assert_eq!(saved["der"], hex::encode(selected)); + } +} + +#[test] +fn tiebreaker_revalidates_cache_and_ignores_display_metadata() { + let temp = tempfile::tempdir().unwrap(); + let config = auto_config(&temp.path().join("state")); + resolve_with(&config, |_| Ok(variant("newer"))).unwrap(); + let path = cache_path(&config, "json"); + let mut saved: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + saved["record"]["notBefore"] = json!("1900-01-01T00:00:00Z"); + saved["record"]["notAfter"] = json!("1901-01-01T00:00:00Z"); + saved["record"]["certificateSha256"] = json!("not-the-certificate-hash"); + fs::write(&path, serde_json::to_vec(&saved).unwrap()).unwrap(); + fs::write(cache_path(&config, "cer"), b"broken projection").unwrap(); + let (_, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + assert_eq!(records[0]["selectionReason"], "cached_newer_not_before"); + assert_eq!( + records[0]["certificateSha256"], + hex::encode(Sha256::digest(variant("newer"))) + ); + assert_ne!(records[0]["notBefore"], saved["record"]["notBefore"]); + assert_eq!( + fs::read(cache_path(&config, "cer")).unwrap(), + variant("newer") + ); +} + +#[test] +fn tiebreaker_invalid_cached_candidates_cannot_veto_valid_download() { + let mut bad_signature = variant("newer"); + *bad_signature.last_mut().unwrap() ^= 1; + let candidates = [ + variant("expired"), + variant("future"), + variant("wrong-key"), + bad_signature, + b"corrupt DER".to_vec(), + ]; + for bytes in candidates { + let temp = tempfile::tempdir().unwrap(); + let config = auto_config(&temp.path().join("state")); + let path = cache_path(&config, "json"); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write( + &path, + serde_json::to_vec(&json!({"der":hex::encode(bytes), "record":{}})).unwrap(), + ) + .unwrap(); + let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + assert_eq!(inputs.len(), 1); + assert_eq!(records[0]["selectionReason"], "no_valid_cache"); + assert_eq!( + fs::read(cache_path(&config, "cer")).unwrap(), + variant("base") + ); + } +} + +#[test] +fn tiebreaker_invalid_downloads_obey_refresh_policy() { + for candidate in ["expired", "future", "wrong-key"] { + let temp = tempfile::tempdir().unwrap(); + let mut config = auto_config(&temp.path().join("state")); + resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + let before = fs::read(cache_path(&config, "json")).unwrap(); + let (inputs, records) = resolve_with(&config, |_| Ok(variant(candidate))).unwrap(); + assert_eq!(inputs.len(), 1); + assert_eq!(records[0]["source"], "cache"); + assert_eq!(records[0]["refreshFailed"], true); + assert_eq!(records[0]["errorCode"], "certificate_invalid"); + assert_eq!(records[0]["selectionReason"], "refresh_failed_valid_cache"); + assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), before); + config.ta_refresh_policy = "require-fresh".into(); + let (inputs, records) = resolve_with(&config, |_| Ok(variant(candidate))).unwrap(); + assert!(inputs.is_empty()); + assert_eq!(records[0]["refreshFailed"], true); + assert!(records[0]["source"].is_null()); + assert_eq!(fs::read(cache_path(&config, "json")).unwrap(), before); + } +} + +#[test] +fn tiebreaker_healthy_cached_choice_works_with_require_fresh() { + let temp = tempfile::tempdir().unwrap(); + let mut config = auto_config(&temp.path().join("state")); + config.ta_refresh_policy = "require-fresh".into(); + resolve_with(&config, |_| Ok(variant("newer"))).unwrap(); + let (inputs, records) = resolve_with(&config, |_| Ok(variant("base"))).unwrap(); + assert_eq!(inputs.len(), 1); + assert_eq!(records[0]["source"], "cache"); + assert_eq!(records[0]["refreshFailed"], false); + let (inputs, records) = resolve_with(&config, |_| Err("offline".into())).unwrap(); + assert!(inputs.is_empty()); + assert_eq!(records[0]["errorCode"], "download_failed"); +} + +#[test] +fn tiebreaker_retries_uri_and_preserves_fixed_ta_mode() { + let temp = tempfile::tempdir().unwrap(); + let mut config = auto_config(&temp.path().join("state")); + let mut attempts = 0; + let (inputs, records) = resolve_with(&config, |_| { + attempts += 1; + if attempts == 1 { + Err("unavailable first URI".into()) + } else { + Ok(variant("newer")) + } + }) + .unwrap(); + assert_eq!(attempts, 2); + assert_eq!(inputs.len(), 1); + assert_eq!(records[0]["source"], "download"); + config.anchors[0].ta = Some(crate::test_support::synthetic_repository().trust_anchor()); + let (inputs, records) = resolve_with(&config, |_| panic!("fixed TA must not fetch")).unwrap(); + assert_eq!(inputs.len(), 1); + assert_eq!(records[0]["source"], "file"); + assert_eq!(records[0]["selectionReason"], "fixed_input"); + assert_eq!( + fs::read(cache_path(&config, "cer")).unwrap(), + variant("newer") + ); +} diff --git a/src/cli/validate.rs b/src/cli/validate.rs index 4f9eb6f..34d4389 100644 --- a/src/cli/validate.rs +++ b/src/cli/validate.rs @@ -151,7 +151,12 @@ fn execute(config: &Config, run_id: &str) -> Result { serde_json::from_slice(&fs::read(root.join("summary.json")).map_err(|e| e.to_string())?) .map_err(|e| e.to_string())?; let repository_failed = summary["repositoryFailedCount"].as_u64().unwrap_or(0); - let cached = anchors.iter().filter(|a| a["source"] == "cache").count(); + // A healthy tiebreaker can prefer a cached TA after a successful download. + // Only a failed refresh using cache makes the result partial. + let cached = anchors + .iter() + .filter(|a| a["source"] == "cache" && a["refreshFailed"] == true) + .count(); let status = if failed > 0 || repository_failed > 0 || cached > 0 { "partial" } else { diff --git a/src/logging/text.rs b/src/logging/text.rs index 9c9bc2a..c3f97e7 100644 --- a/src/logging/text.rs +++ b/src/logging/text.rs @@ -215,6 +215,11 @@ fn message(event: &str, fields: &Value) -> String { repository(fields), reason(fields) ), + "rrdp_session_desynchronized" => format!( + "RRDP delta {} from {} changed its hash within the same session; recovering with the latest snapshot.", + number(fields, "serial"), + repository(fields) + ), "rrdp_cross_origin_rejected" => format!( "Rejected the RRDP {} because its origin ({}) differs from the notification origin ({}).", string(fields, "resource", "resource"), @@ -231,6 +236,11 @@ fn message(event: &str, fields: &Value) -> String { string(fields, "mode", "the selected"), repository(fields) ), + "ta_certificate_selected" => format!( + "Selected the cached trust anchor for {} because it is preferred to the downloaded certificate ({}).", + string(fields, "tal_id", "the configured anchor"), + string(fields, "reason", "TA tiebreaker") + ), "rrdp_download_completed" => format!( "Finished downloading repository data from {}.", repository(fields) @@ -448,8 +458,10 @@ mod tests { "repo_terminal_failure", "http_fetch_failed", "rrdp_delta_fallback", + "rrdp_session_desynchronized", "rrdp_cross_origin_rejected", "rrdp_sync_selected", + "ta_certificate_selected", "rrdp_download_completed", "daemon_started", "daemon_state_recovered", diff --git a/src/repository/storage/models_summary.rs b/src/repository/storage/models_summary.rs index d782e78..d05bc8e 100644 --- a/src/repository/storage/models_summary.rs +++ b/src/repository/storage/models_summary.rs @@ -21,6 +21,10 @@ pub struct RrdpSourceRecord { pub last_snapshot_uri: Option, pub last_snapshot_hash: Option, pub last_error: Option, + /// Delta references in the last successfully synchronized notification. + /// Older databases have no baseline until their next successful sync. + #[serde(default)] + pub last_delta_hashes: std::collections::BTreeMap, } impl RrdpSourceRecord { @@ -43,6 +47,16 @@ impl RrdpSourceRecord { if let Some(last_error) = &self.last_error { validate_non_empty("rrdp_source.last_error", last_error)?; } + if !self.last_delta_hashes.is_empty() + && (self.last_session_id.is_none() + || self.last_serial.is_none() + || self.last_delta_hashes.keys().any(|serial| Some(*serial) > self.last_serial)) + { + return Err(StorageError::InvalidData { + entity: "rrdp_source", + detail: "delta hash history requires a session and serial covering all references".into(), + }); + } Ok(()) } } diff --git a/src/repository/sync/rrdp.rs b/src/repository/sync/rrdp.rs index 4629023..24f23d9 100644 --- a/src/repository/sync/rrdp.rs +++ b/src/repository/sync/rrdp.rs @@ -9,10 +9,10 @@ mod snapshot_apply; use crate::repository::sync::store_projection::{ build_repository_view_present_entry, build_repository_view_withdrawn_entry, - current_rrdp_owner_is, ensure_rrdp_uri_can_be_owned_by, put_repository_view_present, - put_repository_view_withdrawn, put_rrdp_source_member_present, + build_rrdp_source_record_on_success, current_rrdp_owner_is, ensure_rrdp_uri_can_be_owned_by, + put_repository_view_present, put_repository_view_withdrawn, put_rrdp_source_member_present, put_rrdp_source_member_withdrawn, put_rrdp_uri_owner_active, put_rrdp_uri_owner_withdrawn, - update_rrdp_source_record_on_success, upsert_repo_blob_bytes, + upsert_repo_blob_bytes, }; use base64::Engine; use serde::{Deserialize, Serialize}; diff --git a/src/repository/sync/rrdp/models_and_parsing.rs b/src/repository/sync/rrdp/models_and_parsing.rs index 81a7fdc..34e695f 100644 --- a/src/repository/sync/rrdp/models_and_parsing.rs +++ b/src/repository/sync/rrdp/models_and_parsing.rs @@ -321,6 +321,8 @@ pub trait Fetcher: Send + Sync { pub struct RrdpState { pub session_id: String, pub serial: u64, + #[serde(default)] + pub delta_hashes: std::collections::BTreeMap, } impl RrdpState { @@ -342,7 +344,11 @@ pub(crate) fn load_rrdp_local_state( .map_err(|e| e.to_string())? && let (Some(session_id), Some(serial)) = (record.last_session_id, record.last_serial) { - return Ok(Some(RrdpState { session_id, serial })); + return Ok(Some(RrdpState { + session_id, + serial, + delta_hashes: record.last_delta_hashes, + })); } Ok(None) @@ -351,20 +357,28 @@ pub(crate) fn load_rrdp_local_state( pub(crate) fn persist_rrdp_local_state( store: &RocksStore, notification_uri: &str, - state: &RrdpState, - sync_state: RrdpSourceSyncState, - last_snapshot_uri: Option<&str>, - last_snapshot_hash_hex: Option<&str>, + notification: &Notification, + sync_state: Option, ) -> Result<(), String> { - update_rrdp_source_record_on_success( + let mut record = build_rrdp_source_record_on_success( store, notification_uri, - state.session_id.as_str(), - state.serial, + ¬ification.session_id.to_string(), + notification.serial, sync_state, - last_snapshot_uri, - last_snapshot_hash_hex, - ) + Some(¬ification.snapshot_uri), + Some(&hex::encode(notification.snapshot_hash_sha256)), + )?; + // Retain only the previous successful notification window (RFC 9697 §3), + // rather than accumulating an unbounded history for a long-lived session. + record.last_delta_hashes = notification + .deltas + .iter() + .map(|delta| (delta.serial, delta.hash_sha256)) + .collect(); + store + .put_rrdp_source_record(&record) + .map_err(|error| error.to_string()) } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/src/repository/sync/rrdp/notification_sync.rs b/src/repository/sync/rrdp/notification_sync.rs index b9340b5..dec37d8 100644 --- a/src/repository/sync/rrdp/notification_sync.rs +++ b/src/repository/sync/rrdp/notification_sync.rs @@ -123,7 +123,46 @@ fn sync_from_notification_inner( "delta_chain_unavailable" }; - if let Some(s) = same_session_state { + if let Some(s) = same_session_state + && s.serial > notif.serial + { + return Err(RrdpError::NotificationSerialRollback { + old: s.serial, + new: notif.serial, + } + .into()); + } + + // RFC 9697 §3–5: compare references before either the noop fast path or + // fetching a new delta. Old delta files do not need to be downloaded. + let mutation = same_session_state.and_then(|previous| { + notif.deltas.iter().find_map(|delta| { + previous + .delta_hashes + .get(&delta.serial) + .filter(|old_hash| **old_hash != delta.hash_sha256) + .map(|old_hash| (delta.serial, *old_hash, delta.hash_sha256)) + }) + }); + if let Some((serial, old_hash, new_hash)) = mutation { + snapshot_reason = "session_desynchronized"; + crate::logging::emit( + crate::logging::Level::Warn, + "rrdp_session_desynchronized", + || { + serde_json::json!({ + "notify_uri": notification_uri, "session_id": notif.session_id.to_string(), + "serial": serial, "old_hash": hex::encode(old_hash), "new_hash": hex::encode(new_hash), + "recovery": "snapshot", "rfc": "RFC 9697 §3–5", + }) + }, + ); + if let Some(t) = timing.as_ref() { + t.record_count("rrdp_session_desynchronization_total", 1); + } + } + + if let Some(s) = same_session_state.filter(|_| mutation.is_none()) { if s.serial == notif.serial { log_selection("unchanged", "same_session_and_serial"); let _hydrate_step = timing @@ -142,18 +181,13 @@ fn sync_from_notification_inner( if let Some(t) = timing.as_ref() { t.record_count("rrdp_current_index_hydrated_objects_total", hydrated as u64); } + persist_rrdp_local_state(store, notification_uri, ¬if, None) + .map_err(RrdpSyncError::Storage)?; return Ok(0); } - if s.serial > notif.serial { - return Err(RrdpError::NotificationSerialRollback { - old: s.serial, - new: notif.serial, - } - .into()); - } } - if let Some(s) = same_session_state { + if let Some(s) = same_session_state.filter(|_| mutation.is_none()) { // RFC 8182 §3.4.1: if session matches, MAY use deltas when a contiguous chain from the // last processed serial to the current serial can be processed (i.e. deltas cover the gap). let want_first = s.serial + 1; @@ -273,17 +307,11 @@ fn sync_from_notification_inner( let _write_state_total = timing .as_ref() .map(|t| t.span_phase("rrdp_write_state_total")); - let new_state = RrdpState { - session_id: notif.session_id.to_string(), - serial: notif.serial, - }; persist_rrdp_local_state( store, notification_uri, - &new_state, - RrdpSourceSyncState::DeltaReady, - Some(¬if.snapshot_uri), - Some(&hex::encode(notif.snapshot_hash_sha256)), + ¬if, + Some(RrdpSourceSyncState::DeltaReady), ) .map_err(RrdpSyncError::Storage)?; drop(_write_state_step); @@ -394,17 +422,11 @@ fn sync_from_notification_inner( let _write_state_total = timing .as_ref() .map(|t| t.span_phase("rrdp_write_state_total")); - let new_state = RrdpState { - session_id: notif.session_id.to_string(), - serial: notif.serial, - }; persist_rrdp_local_state( store, notification_uri, - &new_state, - RrdpSourceSyncState::SnapshotOnly, - Some(¬if.snapshot_uri), - Some(&hex::encode(notif.snapshot_hash_sha256)), + ¬if, + Some(RrdpSourceSyncState::SnapshotOnly), ) .map_err(RrdpSyncError::Storage)?; drop(_write_state_step); diff --git a/src/repository/sync/rrdp/snapshot_sync.rs b/src/repository/sync/rrdp/snapshot_sync.rs index 33d9844..cb7112a 100644 --- a/src/repository/sync/rrdp/snapshot_sync.rs +++ b/src/repository/sync/rrdp/snapshot_sync.rs @@ -162,17 +162,11 @@ fn sync_from_notification_snapshot_inner( let _write_state_total = timing .as_ref() .map(|t| t.span_phase("rrdp_write_state_total")); - let state = RrdpState { - session_id: notif.session_id.to_string(), - serial: notif.serial, - }; persist_rrdp_local_state( store, notification_uri, - &state, - RrdpSourceSyncState::SnapshotOnly, - Some(¬if.snapshot_uri), - Some(&hex::encode(notif.snapshot_hash_sha256)), + ¬if, + Some(RrdpSourceSyncState::SnapshotOnly), ) .map_err(RrdpSyncError::Storage)?; drop(_write_state_step); diff --git a/src/repository/sync/rrdp/tests.rs b/src/repository/sync/rrdp/tests.rs index 7db3d13..9d8f789 100644 --- a/src/repository/sync/rrdp/tests.rs +++ b/src/repository/sync/rrdp/tests.rs @@ -3,3 +3,4 @@ include!("tests_parts/parsing.rs"); include!("tests_parts/delta_apply.rs"); include!("tests_parts/sync.rs"); include!("tests_parts/edge_cases.rs"); +include!("tests_parts/desynchronization.rs"); diff --git a/src/repository/sync/rrdp/tests_parts/desynchronization.rs b/src/repository/sync/rrdp/tests_parts/desynchronization.rs new file mode 100644 index 0000000..727a4fd --- /dev/null +++ b/src/repository/sync/rrdp/tests_parts/desynchronization.rs @@ -0,0 +1,323 @@ +// RFC 9697: compare notification history without downloading old delta files. +const DESYNC_NOTIFY: &str = "https://example.net/notification.xml"; +const DESYNC_SESSION: &str = "550e8400-e29b-41d4-a716-446655440000"; +const DESYNC_OBJECT: &str = "rsync://example.net/repo/a.roa"; +const DESYNC_OBSOLETE: &str = "rsync://example.net/repo/obsolete.roa"; + +struct DesyncFetcher { + files: HashMap>, + requests: std::sync::Mutex>, +} + +impl DesyncFetcher { + fn snapshot(bytes: Vec) -> Self { + Self { + files: HashMap::from([("https://example.net/snapshot.xml".to_string(), bytes)]), + requests: Default::default(), + } + } + + fn requested(&self) -> Vec { + self.requests.lock().unwrap().clone() + } +} + +impl Fetcher for DesyncFetcher { + fn fetch(&self, uri: &str) -> Result, String> { + self.requests.lock().unwrap().push(uri.to_string()); + self.files + .get(uri) + .cloned() + .ok_or_else(|| format!("not found: {uri}")) + } +} + +fn desync_notification(session: &str, serial: u64, snapshot: &[u8], old_hash: &str) -> Vec { + let snapshot_hash = hex::encode(sha2::Sha256::digest(snapshot)); + let mut deltas = vec![("old", 100, "https://example.net/delta-100.xml", old_hash)]; + let hash_101 = "22".repeat(32); + if serial == 101 { + deltas.push(("new", 101, "https://example.net/delta-101.xml", &hash_101)); + } + notification_xml_with_deltas( + session, + serial, + "https://example.net/snapshot.xml", + &snapshot_hash, + &deltas, + ) +} + +fn seed_desync_repository(store: &RocksStore) { + let snapshot = snapshot_xml( + DESYNC_SESSION, + 100, + &[(DESYNC_OBJECT, b"old"), (DESYNC_OBSOLETE, b"obsolete")], + ); + let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"11".repeat(32)); + sync_from_notification_snapshot( + store, + DESYNC_NOTIFY, + ¬ification, + &DesyncFetcher::snapshot(snapshot), + ) + .unwrap(); +} + +#[test] +fn desynchronization_same_serial_recovers_snapshot_before_noop() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]); + let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"33".repeat(32)); + let fetcher = DesyncFetcher::snapshot(snapshot); + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &fetcher).unwrap(), + 1 + ); + assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]); + assert_current_object(&store, DESYNC_OBJECT, b"recovered"); + assert!( + store + .load_current_object_bytes_by_uri(DESYNC_OBSOLETE) + .unwrap() + .is_none() + ); +} + +#[test] +fn desynchronization_advanced_serial_survives_restart_and_skips_all_deltas() { + let temp = tempfile::tempdir().unwrap(); + { + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + } + let store = RocksStore::open(temp.path()).unwrap(); + let snapshot = snapshot_xml(DESYNC_SESSION, 101, &[(DESYNC_OBJECT, b"recovered")]); + let notification = desync_notification(DESYNC_SESSION, 101, &snapshot, &"33".repeat(32)); + let fetcher = DesyncFetcher::snapshot(snapshot); + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &fetcher).unwrap(); + assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]); + assert_current_object(&store, DESYNC_OBJECT, b"recovered"); + assert!( + store + .load_current_object_bytes_by_uri(DESYNC_OBSOLETE) + .unwrap() + .is_none() + ); +} + +#[test] +fn desynchronization_failed_recovery_keeps_baseline_for_retry() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let before = store + .get_rrdp_source_record(DESYNC_NOTIFY) + .unwrap() + .unwrap(); + let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]); + let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"33".repeat(32)); + // A failed download must not bless the new notification history. + let mut failing = DesyncFetcher::snapshot(snapshot.clone()); + failing.files.clear(); + assert!(sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &failing).is_err()); + assert_eq!( + store + .get_rrdp_source_record(DESYNC_NOTIFY) + .unwrap() + .unwrap(), + before + ); + assert_current_object(&store, DESYNC_OBJECT, b"old"); + + // Nor may a downloaded snapshot with a bad hash advance the baseline. + let corrupt = DesyncFetcher::snapshot(b"corrupt snapshot".to_vec()); + assert!(matches!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &corrupt), + Err(RrdpSyncError::Rrdp(RrdpError::SnapshotHashMismatch)) + )); + assert_eq!( + store + .get_rrdp_source_record(DESYNC_NOTIFY) + .unwrap() + .unwrap(), + before + ); + + let fetcher = DesyncFetcher::snapshot(snapshot); + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &fetcher).unwrap(); + assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]); + assert_current_object(&store, DESYNC_OBJECT, b"recovered"); + // Once recovery succeeds the changed hash becomes the new baseline. + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &PanicOnRrdpFetch).unwrap(), + 0 + ); +} + +#[test] +fn desynchronization_noop_learns_new_references_before_future_mutation() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"recovered")]); + let hash = hex::encode(sha2::Sha256::digest(&snapshot)); + let h99 = "99".repeat(32); + let h100 = "11".repeat(32); + let expanded = notification_xml_with_deltas( + DESYNC_SESSION, + 100, + "https://example.net/snapshot.xml", + &hash, + &[ + ("d99", 99, "https://example.net/d99.xml", &h99), + ("d100", 100, "https://example.net/d100.xml", &h100), + ], + ); + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, &expanded, &PanicOnRrdpFetch).unwrap(), + 0 + ); + assert_current_object(&store, DESYNC_OBJECT, b"old"); + let mutated = String::from_utf8(expanded) + .unwrap() + .replace(&h99, &"88".repeat(32)); + let fetcher = DesyncFetcher::snapshot(snapshot); + sync_from_notification(&store, DESYNC_NOTIFY, mutated.as_bytes(), &fetcher).unwrap(); + assert_eq!(fetcher.requested(), ["https://example.net/snapshot.xml"]); + assert_current_object(&store, DESYNC_OBJECT, b"recovered"); +} + +#[test] +fn desynchronization_successful_delta_updates_and_prunes_history_window() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let delta = delta_xml( + DESYNC_SESSION, + 101, + &[&format!( + r#""#, + hex::encode(sha2::Sha256::digest(b"obsolete")) + )], + ); + let hash = hex::encode(sha2::Sha256::digest(&delta)); + let snapshot = snapshot_xml(DESYNC_SESSION, 101, &[(DESYNC_OBJECT, b"recovered")]); + let snapshot_hash = hex::encode(sha2::Sha256::digest(&snapshot)); + let notification = notification_xml_with_deltas( + DESYNC_SESSION, + 101, + "https://example.net/snapshot.xml", + &snapshot_hash, + &[("new", 101, "https://example.net/delta-101.xml", &hash)], + ); + let fetcher = DesyncFetcher { + files: HashMap::from([("https://example.net/delta-101.xml".into(), delta)]), + requests: Default::default(), + }; + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &fetcher).unwrap(), + 1 + ); + assert_eq!(fetcher.requested(), ["https://example.net/delta-101.xml"]); + let state = load_rrdp_local_state(&store, DESYNC_NOTIFY) + .unwrap() + .unwrap(); + assert_eq!(state.delta_hashes.len(), 1); + assert_eq!(hex::encode(state.delta_hashes[&101]), hash); + assert_current_object(&store, DESYNC_OBJECT, b"old"); + + let mutated = String::from_utf8(notification) + .unwrap() + .replace(&hash, &"ff".repeat(32)); + let recovery = DesyncFetcher::snapshot(snapshot); + sync_from_notification(&store, DESYNC_NOTIFY, mutated.as_bytes(), &recovery).unwrap(); + assert_eq!(recovery.requested(), ["https://example.net/snapshot.xml"]); + assert_current_object(&store, DESYNC_OBJECT, b"recovered"); +} + +#[test] +fn desynchronization_new_session_replaces_hash_baseline() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let session = "550e8400-e29b-41d4-a716-446655440001"; + let snapshot = snapshot_xml(session, 100, &[(DESYNC_OBJECT, b"new session")]); + let notification = desync_notification(session, 100, &snapshot, &"44".repeat(32)); + sync_from_notification( + &store, + DESYNC_NOTIFY, + ¬ification, + &DesyncFetcher::snapshot(snapshot), + ) + .unwrap(); + let state = load_rrdp_local_state(&store, DESYNC_NOTIFY) + .unwrap() + .unwrap(); + assert_eq!(state.session_id, session); + assert_eq!(state.delta_hashes[&100], [0x44; 32]); + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &PanicOnRrdpFetch).unwrap(), + 0 + ); + assert_current_object(&store, DESYNC_OBJECT, b"new session"); +} + +#[test] +fn desynchronization_legacy_source_record_defaults_to_empty_history() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let record = store + .get_rrdp_source_record(DESYNC_NOTIFY) + .unwrap() + .unwrap(); + let mut legacy = serde_cbor::value::to_value(&record).unwrap(); + if let serde_cbor::Value::Map(map) = &mut legacy { + map.remove(&serde_cbor::Value::Text("last_delta_hashes".into())); + } else { + panic!("source record must encode as a CBOR map"); + } + let decoded: crate::repository::storage::RrdpSourceRecord = + serde_cbor::from_slice(&serde_cbor::to_vec(&legacy).unwrap()).unwrap(); + assert!(decoded.last_delta_hashes.is_empty()); + store.put_rrdp_source_record(&decoded).unwrap(); + let snapshot = snapshot_xml(DESYNC_SESSION, 100, &[(DESYNC_OBJECT, b"unused")]); + let notification = desync_notification(DESYNC_SESSION, 100, &snapshot, &"55".repeat(32)); + assert_eq!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &PanicOnRrdpFetch).unwrap(), + 0 + ); + assert_eq!( + load_rrdp_local_state(&store, DESYNC_NOTIFY) + .unwrap() + .unwrap() + .delta_hashes[&100], + [0x55; 32] + ); + assert_current_object(&store, DESYNC_OBJECT, b"old"); +} + +#[test] +fn desynchronization_does_not_override_serial_rollback_guard() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + seed_desync_repository(&store); + let before = store.get_rrdp_source_record(DESYNC_NOTIFY).unwrap(); + let notification = notification_xml( + DESYNC_SESSION, + 99, + "https://example.net/snapshot.xml", + &"00".repeat(32), + ); + assert!(matches!( + sync_from_notification(&store, DESYNC_NOTIFY, ¬ification, &PanicOnRrdpFetch), + Err(RrdpSyncError::Rrdp(RrdpError::NotificationSerialRollback { + old: 100, + new: 99 + })) + )); + assert_eq!(store.get_rrdp_source_record(DESYNC_NOTIFY).unwrap(), before); +} diff --git a/src/repository/sync/rrdp/tests_parts/edge_cases.rs b/src/repository/sync/rrdp/tests_parts/edge_cases.rs index 2108222..98f9d70 100644 --- a/src/repository/sync/rrdp/tests_parts/edge_cases.rs +++ b/src/repository/sync/rrdp/tests_parts/edge_cases.rs @@ -10,16 +10,17 @@ fn load_rrdp_local_state_uses_source_record_only() { None ); - update_rrdp_source_record_on_success( + let record = build_rrdp_source_record_on_success( &store, notif_uri, "source-session", 9, - crate::repository::storage::RrdpSourceSyncState::DeltaReady, + Some(crate::repository::storage::RrdpSourceSyncState::DeltaReady), Some("https://example.net/snapshot.xml"), Some(&hex::encode([0x11; 32])), ) - .expect("write source record"); + .expect("build source record"); + store.put_rrdp_source_record(&record).expect("write source record"); let got = load_rrdp_local_state(&store, notif_uri) .expect("load source preferred") @@ -29,6 +30,7 @@ fn load_rrdp_local_state_uses_source_record_only() { RrdpState { session_id: "source-session".to_string(), serial: 9, + delta_hashes: Default::default(), } ); } diff --git a/src/repository/sync/store_projection.rs b/src/repository/sync/store_projection.rs index 312c949..ec7fd88 100644 --- a/src/repository/sync/store_projection.rs +++ b/src/repository/sync/store_projection.rs @@ -408,15 +408,15 @@ pub fn put_rrdp_uri_owner_withdrawn( .map_err(|e| e.to_string()) } -pub fn update_rrdp_source_record_on_success( +pub fn build_rrdp_source_record_on_success( store: &RocksStore, notification_uri: &str, session_id: &str, serial: u64, - sync_state: RrdpSourceSyncState, + sync_state: Option, last_snapshot_uri: Option<&str>, last_snapshot_hash_hex: Option<&str>, -) -> Result<(), String> { +) -> Result { let now = now_pack_time(); let mut record = match store .get_rrdp_source_record(notification_uri) @@ -430,10 +430,11 @@ pub fn update_rrdp_source_record_on_success( first_seen_at: now.clone(), last_seen_at: now.clone(), last_sync_at: None, - sync_state, + sync_state: sync_state.unwrap_or(RrdpSourceSyncState::Empty), last_snapshot_uri: None, last_snapshot_hash: None, last_error: None, + last_delta_hashes: Default::default(), }, }; @@ -441,14 +442,14 @@ pub fn update_rrdp_source_record_on_success( record.last_serial = Some(serial); record.last_seen_at = now.clone(); record.last_sync_at = Some(now); - record.sync_state = sync_state; + record.sync_state = sync_state.unwrap_or(record.sync_state); record.last_snapshot_uri = last_snapshot_uri.map(str::to_string); record.last_snapshot_hash = last_snapshot_hash_hex.map(str::to_string); record.last_error = None; - store - .put_rrdp_source_record(&record) - .map_err(|e| e.to_string()) + // The caller adds the notification's delta history before the single write, + // so session/serial and the history cannot be committed independently. + Ok(record) } pub fn compute_sha256_hex(bytes: &[u8]) -> String { diff --git a/src/test_support.rs b/src/test_support.rs index 91fe83d..5bda55e 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -26,6 +26,10 @@ impl SyntheticRepositoryFixture { pub(crate) fn trust_anchor(&self) -> PathBuf { self.root.join("ta/custom-ta.cer") } + + pub(crate) fn ta_variant(&self, name: &str) -> PathBuf { + self.root.join("ta-variants").join(format!("{name}.cer")) + } } pub(crate) fn synthetic_repository() -> &'static SyntheticRepositoryFixture { diff --git a/tests/security_recovery_e2e.rs b/tests/security_recovery_e2e.rs new file mode 100644 index 0000000..c99e360 --- /dev/null +++ b/tests/security_recovery_e2e.rs @@ -0,0 +1,9 @@ +#[test] +fn persisted_rrdp_desynchronization_and_ta_tiebreaker_are_end_to_end() { + let status = std::process::Command::new("python3") + .arg("tests/support/test_security_recovery.py") + .arg(env!("CARGO_BIN_EXE_panda-rpki")) + .status() + .expect("start RRDP recovery and TA selection integration driver"); + assert!(status.success(), "security recovery E2E failed: {status}"); +} diff --git a/tests/support/README.md b/tests/support/README.md index 2f60319..c04ad6d 100644 --- a/tests/support/README.md +++ b/tests/support/README.md @@ -11,3 +11,17 @@ network request is included in this directory. The generator requires Python `cryptography`, OpenSSL's `cms` command, and Python 3; CI installs those test tools explicitly. Test output is created under a temporary directory and is not part of a release artifact. +# Security recovery and TA selection + +`cargo test --offline --test security_recovery_e2e` runs the real CLI against a +local HTTPS repository with persisted state. Across three processes it checks +initial snapshot, recovery after a delta-hash mutation at unchanged session/serial, +and a subsequent noop. Request assertions prove that no historical delta is +downloaded. A newer cached TA must survive a valid older download under +`require-fresh`, with a successful result and no cache-fallback count. + +To preserve the per-step request and summary evidence: + +```sh +python3 tests/support/test_security_recovery.py target/debug/panda-rpki --proof /tmp/security-recovery.json +``` diff --git a/tests/support/generate_repository.py b/tests/support/generate_repository.py index e2d4d0b..7b7ade0 100644 --- a/tests/support/generate_repository.py +++ b/tests/support/generate_repository.py @@ -545,6 +545,29 @@ def make_https_certificates(output: Path) -> None: write_key(output / "certs/rrdp-server.key", server_key) +def make_ta_variants(output: Path, name: x509.Name, key, foreign_key) -> None: + """Real self-signed versions of one TAL-bound key, for tiebreaker tests.""" + variants = { + "older": (datetime(2025, 1, 1, tzinfo=UTC), VALID_TO), + "newer": (datetime(2026, 5, 1, tzinfo=UTC), VALID_TO), + "shorter": (VALID_FROM, datetime(2026, 12, 31, tzinfo=UTC)), + "longer": (VALID_FROM, datetime(2030, 1, 1, tzinfo=UTC)), + "equal-period": (VALID_FROM, VALID_TO), + "expired": (datetime(2025, 1, 1, tzinfo=UTC), datetime(2025, 12, 31, tzinfo=UTC)), + "future": (datetime(2027, 1, 1, tzinfo=UTC), datetime(2028, 1, 1, tzinfo=UTC)), + "wrong-key": (VALID_FROM, VALID_TO), + } + for serial, (variant, (start, end)) in enumerate(variants.items(), 0x14700100): + selected_key = foreign_key if variant == "wrong-key" else key + certificate = sign_cert( + name, name, selected_key.public_key(), selected_key, serial, True, selected_key, + ip_ext=ip_resources([CHILD_PREFIX]), as_ext=as_resources([TEST_ASN]), + ca_repository=ROOT_REPO_URI, manifest_uri=ROOT_MFT_URI, notify_uri=RRDP_URI, + not_before=start, not_after=end, + ) + write(output / f"ta-variants/{variant}.cer", certificate.public_bytes(Encoding.DER)) + + def main() -> None: global RRDP_HOST, FOREIGN_RRDP_HOST, RRDP_PORT, RSYNC_HOST, RSYNC_PORT global TAL_URI, TAL_HTTPS_URI, RRDP_URI, STRICT_NAMES @@ -638,6 +661,7 @@ def main() -> None: ).encode() write(output / "tal/custom.tal", tal) write(output / "ta/custom-ta.cer", root_cert.public_bytes(Encoding.DER)) + make_ta_variants(output, root_name, root_key, child_key) with tempfile.TemporaryDirectory(prefix="custom-rpki-cms-") as temp: work = Path(temp) common = output / ".common" diff --git a/tests/support/test_security_recovery.py b/tests/support/test_security_recovery.py new file mode 100644 index 0000000..13d70f5 --- /dev/null +++ b/tests/support/test_security_recovery.py @@ -0,0 +1,151 @@ +"""Real CLI, HTTPS and persisted state for RFC 9697 and TA tiebreaker -06.""" + +from __future__ import annotations + +import argparse +import functools +import http.server +import json +from pathlib import Path +import shutil +import ssl +import sys +import tempfile +import threading + +from test_public_config import ( + assert_successful_output, + free_port, + make_rsync_wrapper, + read_json, + run_checked, + write_config, +) + + +def run(binary: str) -> dict: + support = Path(__file__).resolve().parent + with tempfile.TemporaryDirectory(prefix="panda-rpki-security-") as tmp: + base = Path(tmp) + fixture = base / "fixture" + port = free_port() + run_checked([ + sys.executable, str(support / "generate_repository.py"), + "--rrdp-host", "127.0.0.1", "--rrdp-port", str(port), + "--rsync-host", "127.0.0.1", "--strict-names", "--output", str(fixture), + ]) + active = base / "active" + shutil.copytree(fixture / "cases/baseline-v1/http", active) + # Prefer real HTTPS for automatic TA retrieval. Retain the rsync URI: + # the validation core uses it as the key for its offline TA projection. + tal = fixture / "tal/custom.tal" + lines = tal.read_text().splitlines() + tal.write_text("\n".join( + [line for line in lines if line.startswith("https://")] + + [line for line in lines if line.startswith("rsync://")] + + [line for line in lines if not line.startswith(("https://", "rsync://"))] + ) + "\n", encoding="utf-8") + served_ta = active / "ta/custom-ta.cer" + served_ta.write_bytes((fixture / "ta-variants/newer.cer").read_bytes()) + notification = active / "rrdp/notification.xml" + original = notification.read_text() + + def advertise_hash(value: str) -> None: + reference = ( + f'' + ) + notification.write_text(original.replace("", reference + "")) + + advertise_hash("11") + requests: list[str] = [] + + class Handler(http.server.SimpleHTTPRequestHandler): + def do_GET(self) -> None: + requests.append(self.path) + super().do_GET() + + def log_message(self, format: str, *args) -> None: + pass + + server = http.server.ThreadingHTTPServer( + ("127.0.0.1", port), functools.partial(Handler, directory=str(active)) + ) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(fixture / "certs/rrdp-server.pem", fixture / "certs/rrdp-server.key") + server.socket = context.wrap_socket(server.socket, server_side=True) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + output, state = base / "output", base / "state" + wrapper, wrapper_log = base / "rsync.py", base / "rsync.log" + make_rsync_wrapper(wrapper, fixture, wrapper_log) + config = base / "validate.toml" + write_config( + config, fixture=fixture, output=output, state=state, + ccr=output / "result.ccr", wrapper=wrapper, mirror=base / "mirror", + sync_preference="rrdp-then-rsync", rrdp_mode="auto", + validation_time="2026-06-01T00:00:00Z", + resource_mode="rfc6487", scope="publication-point", + ) + text = "\n".join(line for line in config.read_text().splitlines() if not line.startswith("ta = ")) + text = text.replace("[strict]", 'ta-refresh-policy = "require-fresh"\n\n[strict]') + config.write_text(text + "\n", encoding="utf-8") + proof = {"steps": []} + + def validate(name: str) -> tuple[dict, str, list[str]]: + start = len(requests) + result = run_checked([binary, "validate", "--config", str(config)]) + assert_successful_output(output, "2026-06-01T00:00:00Z") + summary = read_json(output / "summary.json") + seen = requests[start:] + assert summary["taCacheFallbackCount"] == 0, summary + assert not any("delta-" in uri for uri in seen), seen + assert not wrapper_log.exists(), "unexpected rsync fallback" + proof["steps"].append({ + "name": name, "requests": seen, "vrps": summary["vrps"], + "resultStatus": summary["resultStatus"], + "taCacheFallbackCount": summary["taCacheFallbackCount"], + "trustAnchors": summary["trustAnchors"], + }) + return summary, result.stderr, seen + + summary, _, seen = validate("initial_snapshot_newer_ta") + assert "/rrdp/snapshot.xml" in seen, seen + assert summary["trustAnchors"][0]["source"] == "download", summary + cache = next(base.joinpath("state.ta-cache").glob("*.json")) + cache_before = cache.read_bytes() + + # A new process sees a mutated historical delta hash at the same + # session/serial, and a valid but older TA with the identical key. + advertise_hash("33") + served_ta.write_bytes((fixture / "ta/custom-ta.cer").read_bytes()) + summary, stderr, seen = validate("restarted_hash_mutation_and_older_ta") + assert "/rrdp/snapshot.xml" in seen, seen + assert "rrdp_session_desynchronized" in stderr, stderr + anchor = summary["trustAnchors"][0] + assert anchor["source"] == "cache" and not anchor["refreshFailed"], anchor + assert anchor["selectionReason"] == "cached_newer_not_before", anchor + assert cache.read_bytes() == cache_before, "older TA replaced the preferred cache" + + _, stderr, seen = validate("restarted_noop_after_recovery") + assert "/rrdp/snapshot.xml" not in seen, seen + assert "rrdp_session_desynchronized" not in stderr, stderr + assert cache.read_bytes() == cache_before + return proof + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("binary") + parser.add_argument("--proof", type=Path) + args = parser.parse_args() + proof = run(args.binary) + if args.proof: + args.proof.parent.mkdir(parents=True, exist_ok=True) + args.proof.write_text(json.dumps(proof, indent=2) + "\n", encoding="utf-8") + print(json.dumps(proof, indent=2))