diff --git a/docs/command-line-reference.md b/docs/command-line-reference.md index 7721955..ff2b158 100644 --- a/docs/command-line-reference.md +++ b/docs/command-line-reference.md @@ -135,3 +135,9 @@ TA certificate on each cycle; providing `ta` selects fixed mode. TOML is read once at process start. Put `log-format = "text"` and `ta-refresh-policy` at the top level, before any `[[anchors]]` tables. See [Docker reload steps](docker.md#graceful-stop-edit-and-reload) and [complete container TOML examples](../docker/examples/README.md). + +Manifest failed-fetch can retain still-valid, previously validated results from +persistent state. `summary.json.failed_fetch_cached` counts publication points +using those results. Such a run remains partial; it does not advance +`publish-dir/latest`. See [failed-fetch recovery](failed-fetch.md) for validity +bounds, policy changes and cached-subtree behaviour. diff --git a/docs/conformance-matrix.md b/docs/conformance-matrix.md index da920d5..0e32f27 100644 --- a/docs/conformance-matrix.md +++ b/docs/conformance-matrix.md @@ -4,6 +4,7 @@ | Behaviour | Standard reference | Implementation | Evidence | | --- | --- | --- | --- | +| Retain still-valid per-CA results on failed-fetch, persist them across restarts, and traverse descendants using cached results only | [RFC 9286 §6.6](https://www.rfc-editor.org/rfc/rfc9286.html#section-6.6) | `src/validation/failed_fetch.rs`, `tree_runner/`, `tree_parallel/`, atomic history/result storage | `tests/failed_fetch_e2e.rs`, cache/expiry/context/storage tests; [behaviour and limits](failed-fetch.md) | | CA-key manifest history across filename changes, monotonic thisUpdate, exact EE SIA location matching, rename warnings and coordinated commits | [RFC 9981 §§2–4](https://www.rfc-editor.org/rfc/rfc9981.html) | `src/validation/manifest/`, `src/repository/storage/store_manifest.rs`, `src/validation/tree_runner/` | `history_tests.rs`, storage history fault tests, `tests/manifest_history_e2e.rs` | | 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` | diff --git a/docs/failed-fetch.md b/docs/failed-fetch.md new file mode 100644 index 0000000..5116194 --- /dev/null +++ b/docs/failed-fetch.md @@ -0,0 +1,43 @@ +# Manifest failed-fetch recovery + +After a publication point passes validation, Panda keeps its direct VRPs, ASPAs, +router keys, child CA references, audit information and CCR contribution in the +state database. If a later fetch fails, it warns and uses that previously +validated result while it remains usable, following +[RFC 9286 section 6.6](https://www.rfc-editor.org/rfc/rfc9286.html#section-6.6). +This is automatic and independent of whether CCR output is enabled. + +Results belong to a CA public key within its TAL namespace. Reuse additionally +requires the same CA certificate, publication locations, effective resources, +validation policy and TA resource constraints. A different certificate or policy +requires a fresh successful validation before a result can be reused. + +The cache ends at the earliest relevant deadline: manifest nextUpdate, CRL +nextUpdate, CA/manifest EE validity, or validity of an accepted payload or child +certificate. Panda uses a conservative deadline for the whole instance. At the +boundary it stops reusing the result; repeated failures never extend this time. +A validation time preceding the cached validation time also prevents reuse. + +When an instance fails, its cached descendants use cached results only: they do +not request another repository sync or validate newly downloaded objects. A +missing, expired or mismatched descendant cache stops that branch. Normal tree +limits still apply. Successful fresh parents can delete or revoke child CAs; +those children are not restored from an earlier cache. A successful empty result +supersedes the previous payloads. + +The manifest anti-rollback history and result cache commit atomically. Reuse +never advances that history. Cache corruption, an inconsistent history, or a +storage/issuer failure cannot authorize reuse. Transport state is separate from +validated results, so a mutable repository view cannot silently replace cached +payloads or cached child certificates. + +A cache hit preserves the fetch failure. `summary.json` includes +`failed_fetch_cached`, and the result remains `partial` with exit 3 when +`require-complete` is enabled. Publication to `publish-dir/latest` still requires +a complete success. Structured logs use `failed_fetch_cache_reused` with the +original verification time and effective deadline; publication-point audits use +`failed_fetch_cached` and terminal state `failed_cached`. + +Keep the state database between invocations or daemon cycles to retain these +results. This does not require serializing RRDP objects to a filesystem mirror, +and does not change the current RRDP/rsync storage architecture. diff --git a/src/cli/validate.rs b/src/cli/validate.rs index 57b453e..ecd4d04 100644 --- a/src/cli/validate.rs +++ b/src/cli/validate.rs @@ -158,7 +158,12 @@ fn execute(config: &Config, run_id: &str) -> Result { .iter() .filter(|a| a["source"] == "cache" && a["refreshFailed"] == true) .count(); - let status = if failed > 0 || repository_failed > 0 || manifest_failed > 0 || cached > 0 { + let status = if failed > 0 + || repository_failed > 0 + || manifest_failed > 0 + || cached > 0 + || summary["failed_fetch_cached"].as_u64().unwrap_or(0) > 0 + { "partial" } else { "success" diff --git a/src/logging/progress.rs b/src/logging/progress.rs index 5ec496d..3ac1eb7 100644 --- a/src/logging/progress.rs +++ b/src/logging/progress.rs @@ -30,6 +30,7 @@ fn registered_event_level(kind: &str) -> Option { | "rrdp_sync_failed" | "rrdp_cross_origin_rejected" | "repo_terminal_failure" + | "failed_fetch_cache_reused" | "rrdp_failed_fallback_rsync" ) { crate::logging::Level::Warn diff --git a/src/model/roa.rs b/src/model/roa.rs index 2dcdd51..1189502 100644 --- a/src/model/roa.rs +++ b/src/model/roa.rs @@ -246,7 +246,7 @@ pub struct RoaIpAddress { pub max_length: Option, } -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)] pub struct IpPrefix { pub afi: RoaAfi, /// Prefix length in bits. diff --git a/src/output/audit.rs b/src/output/audit.rs index 6c62e64..eb4f460 100644 --- a/src/output/audit.rs +++ b/src/output/audit.rs @@ -1,9 +1,9 @@ -use serde::Serialize; +use serde::{Deserialize, Serialize}; use sha2::Digest; use crate::validation::policy::Policy; -#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum AuditObjectKind { Manifest, @@ -15,7 +15,7 @@ pub enum AuditObjectKind { Other, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum AuditObjectResult { Ok, @@ -23,7 +23,7 @@ pub enum AuditObjectResult { Error, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ObjectAuditEntry { pub rsync_uri: String, pub sha256_hex: String, @@ -33,7 +33,7 @@ pub struct ObjectAuditEntry { pub detail: Option, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct AuditWarning { pub message: String, pub category: String, @@ -110,7 +110,7 @@ impl From<&crate::output::report::Warning> for AuditWarning { } } -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct PublicationPointAudit { /// Monotonic node ID assigned by the traversal engine. /// @@ -148,7 +148,7 @@ pub struct PublicationPointAudit { pub objects: Vec, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct DiscoveredFrom { pub parent_manifest_rsync_uri: String, pub child_ca_certificate_rsync_uri: String, diff --git a/src/repository/storage/config.rs b/src/repository/storage/config.rs index 7b2e705..f7c597b 100644 --- a/src/repository/storage/config.rs +++ b/src/repository/storage/config.rs @@ -1,5 +1,6 @@ use rocksdb::{ColumnFamilyDescriptor, DBCompressionType, Options}; +pub const CF_VALIDATED_PUBLICATION_POINT: &str = "validated_publication_point"; pub const CF_REPOSITORY_VIEW: &str = "repository_view"; pub const CF_RAW_BY_HASH: &str = "raw_by_hash"; pub const CF_RAW_BLOB: &str = "raw_blob"; @@ -12,6 +13,7 @@ pub const CF_RRDP_URI_OWNER: &str = "rrdp_uri_owner"; pub const ALL_COLUMN_FAMILY_NAMES: &[&str] = &[ CF_REPOSITORY_VIEW, + CF_VALIDATED_PUBLICATION_POINT, CF_RAW_BY_HASH, CF_RAW_BLOB, CF_MANIFEST_ANTI_ROLLBACK, diff --git a/src/repository/storage/store_manifest.rs b/src/repository/storage/store_manifest.rs index 59402e6..41f2ce3 100644 --- a/src/repository/storage/store_manifest.rs +++ b/src/repository/storage/store_manifest.rs @@ -19,10 +19,20 @@ impl RocksStore { /// Re-read and validate under the same lock as the write. RocksDB itself /// excludes a second writable process opening this database. + #[cfg(test)] pub(crate) fn update_ca_manifest_history>( &self, identity: &CaIdentity, update: impl FnOnce(Option<&CaManifestHistory>) -> Result, + ) -> Result { + self.update_ca_manifest_history_and_cache(identity, None, update) + } + + pub(crate) fn update_ca_manifest_history_and_cache>( + &self, + identity: &CaIdentity, + mut cache: Option, + update: impl FnOnce(Option<&CaManifestHistory>) -> Result, ) -> Result { let _guard = self.manifest_history_commit @@ -36,9 +46,57 @@ impl RocksStore { history.validate_internal()?; let cf = self.cf(CF_MANIFEST_ANTI_ROLLBACK)?; let bytes = encode_cbor(&history, "ca_manifest")?; + let cache_cf = self.cf(CF_VALIDATED_PUBLICATION_POINT)?; + let key = ca_manifest_key(identity); + let mut batch = WriteBatch::default(); + batch.put_cf(cf, key.as_bytes(), bytes); + if let Some(cache) = cache.as_mut() { + cache.history = history.clone(); + batch.put_cf( + cache_cf, + key.as_bytes(), + encode_cbor(cache, "validated_publication_point")?, + ); + } else { + // A fresh success supersedes the old result even when this result + // is ineligible for reuse (for example, drop-publication-point). + batch.delete_cf(cache_cf, key.as_bytes()); + } self.db - .put_cf(cf, ca_manifest_key(identity).as_bytes(), bytes) + .write(batch) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(history) } } + +impl RocksStore { + pub(crate) fn get_validated_publication_point( + &self, + identity: &CaIdentity, + ) -> StorageResult> { + let _guard = + self.manifest_history_commit + .lock() + .map_err(|_| StorageError::InvalidData { + entity: "validated_publication_point", + detail: "cache read lock poisoned".into(), + })?; + let cf = self.cf(CF_VALIDATED_PUBLICATION_POINT)?; + let Some(bytes) = self + .db + .get_cf(cf, ca_manifest_key(identity).as_bytes()) + .map_err(|e| StorageError::RocksDb(e.to_string()))? + else { + return Ok(None); + }; + let cache: crate::validation::failed_fetch::CachedPublicationPoint = + decode_cbor(&bytes, "validated_publication_point")?; + if self.get_ca_manifest_history(identity)?.as_ref() != Some(&cache.history) { + return Err(StorageError::InvalidData { + entity: "validated_publication_point", + detail: "cache does not match committed manifest history".into(), + }); + } + Ok(Some(cache)) + } +} diff --git a/src/repository/storage/tests_parts/manifest_history.rs b/src/repository/storage/tests_parts/manifest_history.rs index 0d5cab3..aa999f2 100644 --- a/src/repository/storage/tests_parts/manifest_history.rs +++ b/src/repository/storage/tests_parts/manifest_history.rs @@ -82,3 +82,17 @@ fn manifest_history_write_failure_is_returned_and_does_not_create_success() { )); assert_eq!(store.get_ca_manifest_history(&id).unwrap(), None); } + +#[test] +fn result_cache_storage_failure_cannot_commit_history_alone() { + let td = tempfile::tempdir().unwrap(); + let mut store = RocksStore::open(td.path()).unwrap(); + let id = CaIdentity::from_spki([5; 32], b"key"); + let history = sample_ca_history(); + store.update_ca_manifest_history::(&id, |_| Ok(history.clone())).unwrap(); + store.db.drop_cf(CF_VALIDATED_PUBLICATION_POINT).unwrap(); + let mut newer = history.clone(); newer.manifest_number_be = vec![2]; + assert!(store.update_ca_manifest_history::(&id, |_| Ok(newer)).is_err()); + assert_eq!(store.get_ca_manifest_history(&id).unwrap(), Some(history)); + assert!(store.get_validated_publication_point(&id).is_err()); +} diff --git a/src/runtime/run.rs b/src/runtime/run.rs index e83e98d..dc9b0fb 100644 --- a/src/runtime/run.rs +++ b/src/runtime/run.rs @@ -421,6 +421,7 @@ struct ValidationSummary { rrdp_delta_updates: usize, vrps: usize, publication_points: usize, + failed_fetch_cached: usize, validation_ms: u64, } @@ -466,6 +467,11 @@ fn write_summary( rrdp_noop_repositories: 0, rrdp_delta_updates: 0, vrps: shared.vrps.len(), + failed_fetch_cached: shared + .publication_points + .iter() + .filter(|p| p.source == "failed_fetch_cached") + .count(), publication_points: shared.publication_points.len(), validation_ms, }; diff --git a/src/runtime/tests.rs b/src/runtime/tests.rs index dc1dd51..2c84818 100644 --- a/src/runtime/tests.rs +++ b/src/runtime/tests.rs @@ -300,7 +300,67 @@ fn synthetic_snapshot_delta_restart_rollback_and_log_equivalence() { std::fs::write(rollback.join("rrdp/notification.xml"), notification).unwrap(); std::fs::remove_file(&active).unwrap(); std::os::unix::fs::symlink(&rollback, &active).unwrap(); - run("rollback-rejected", 0, 8, "info"); + let (rollback_out, rollback_summary) = run("rollback-rejected", 2, 8, "info"); + assert_eq!(rollback_summary["failed_fetch_cached"], 1); + assert_eq!( + std::fs::read(rollback_out.join("vrps.csv")).unwrap(), + std::fs::read(delta.join("vrps.csv")).unwrap() + ); + let report: serde_json::Value = + serde_json::from_slice(&std::fs::read(rollback_out.join("report.json")).unwrap()).unwrap(); + assert!( + report["publication_points"] + .as_array() + .unwrap() + .iter() + .any(|point| point["source"] == "failed_fetch_cached") + ); + // The failed ancestor snapshot also replaces the mutable child objects + // with the older repository. Parallel traversal must use only the durable + // child result (two VRPs), without staging fresh child validation. + use base64::Engine as _; + let original_snapshot = std::fs::read_to_string(rollback.join("rrdp/snapshot.xml")).unwrap(); + let root_manifest = + std::fs::read(fixture.join("cases/baseline-v1/repository/root/root.mft")).unwrap(); + let failed_snapshot = original_snapshot + .replace( + &base64::engine::general_purpose::STANDARD.encode(root_manifest), + &base64::engine::general_purpose::STANDARD.encode(b"broken ancestor manifest"), + ) + .replace( + "22222222-2222-4333-8444-555555555555", + "33333333-2222-4333-8444-555555555555", + ); + assert_ne!(failed_snapshot, original_snapshot); + let failed_notification = std::fs::read_to_string(rollback.join("rrdp/notification.xml")) + .unwrap() + .replace( + &hex::encode(sha2::Sha256::digest(original_snapshot.as_bytes())), + &hex::encode(sha2::Sha256::digest(failed_snapshot.as_bytes())), + ) + .replace( + "22222222-2222-4333-8444-555555555555", + "33333333-2222-4333-8444-555555555555", + ); + std::fs::write(rollback.join("rrdp/snapshot.xml"), failed_snapshot).unwrap(); + std::fs::write(rollback.join("rrdp/notification.xml"), failed_notification).unwrap(); + let (ancestor_out, ancestor_summary) = run("ancestor-failed-cache-only", 2, 8, "info"); + assert_eq!(ancestor_summary["failed_fetch_cached"], 2); + assert_eq!( + std::fs::read(ancestor_out.join("vrps.csv")).unwrap(), + std::fs::read(delta.join("vrps.csv")).unwrap() + ); + let report: serde_json::Value = + serde_json::from_slice(&std::fs::read(ancestor_out.join("report.json")).unwrap()).unwrap(); + let points = report["publication_points"].as_array().unwrap(); + assert!(points.iter().all(|p| p["source"] == "failed_fetch_cached")); + assert_eq!( + points + .iter() + .filter(|p| p["repo_sync_phase"] == "cache_only") + .count(), + 1 + ); let store = RocksStore::open(&db).unwrap(); assert_eq!( store.get_ca_manifest_history(&identity).unwrap(), diff --git a/src/ta_constraints/implementation.rs b/src/ta_constraints/implementation.rs index 49b03a9..1ea3927 100644 --- a/src/ta_constraints/implementation.rs +++ b/src/ta_constraints/implementation.rs @@ -232,9 +232,14 @@ pub struct TaConstraints { allow_asn: IntervalIndex, deny_asn: IntervalIndex, warnings: Vec, + cache_fingerprint: String, } impl TaConstraints { + pub(crate) fn cache_fingerprint(&self) -> &str { + &self.cache_fingerprint + } + pub fn from_file(path: &Path) -> Result { let contents = std::fs::read_to_string(path) .map_err(|error| format!("read {}: {error}", path.display()))?; @@ -324,8 +329,19 @@ impl TaConstraints { normalize_as_intervals("allow AS", &mut allow_asn, &mut warnings); normalize_as_intervals("deny AS", &mut deny_asn, &mut warnings); + use sha2::Digest; + let cache_fingerprint = hex::encode(sha2::Sha256::digest( + format!( + "{:?}", + ( + &allow_v4, &deny_v4, &allow_v6, &deny_v6, &allow_asn, &deny_asn, + ) + ) + .as_bytes(), + )); Ok(Self { source, + cache_fingerprint, allow_v4: IntervalIndex::new(allow_v4), deny_v4: IntervalIndex::new(deny_v4), allow_v6: IntervalIndex::new(allow_v6), diff --git a/src/ta_constraints/tests.rs b/src/ta_constraints/tests.rs index 78219b2..dac2af9 100644 --- a/src/ta_constraints/tests.rs +++ b/src/ta_constraints/tests.rs @@ -311,3 +311,17 @@ fn generated_very_large_ipv4_constraints_use_interval_tree() { assert!(!constraints.allow_v4.uses_tree()); assert!(constraints.warnings.is_empty()); } + +#[test] +fn cache_fingerprint_tracks_normalized_policy_not_source_path_or_formatting() { + let first = TaConstraints::parse("first".into(), "allow 192.0.2.0/24\nallow 64496\n").unwrap(); + let same = TaConstraints::parse( + "second".into(), + "# same policy\nallow 64496\nallow 192.0.2.0/24\n", + ) + .unwrap(); + let changed = + TaConstraints::parse("first".into(), "allow 192.0.2.0/24\nallow 64497\n").unwrap(); + assert_eq!(first.cache_fingerprint(), same.cache_fingerprint()); + assert_ne!(first.cache_fingerprint(), changed.cache_fingerprint()); +} diff --git a/src/validation/failed_fetch.rs b/src/validation/failed_fetch.rs new file mode 100644 index 0000000..f968e34 --- /dev/null +++ b/src/validation/failed_fetch.rs @@ -0,0 +1,214 @@ +//! RFC 9286 §6.6: immutable, previously validated results, independent of the +//! mutable transport view. A cache edge never authorizes fresh child work. +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use time::OffsetDateTime; +use x509_parser::prelude::{FromDer, X509Certificate}; + +use crate::ccr::projection::CcrManifestProjection; +use crate::model::{rc::ResourceCertificate, signed_object::RpkiSignedObject}; +use crate::output::audit::{AuditObjectKind, AuditObjectResult, PublicationPointAudit}; +use crate::repository::storage::{CaIdentity, CaManifestHistory, PackTime, RocksStore}; +use crate::validation::objects::{ + AspaAttestation, ObjectsOutput, ObjectsStats, RouterKeyPayload, Vrp, +}; +use crate::validation::policy::Policy; +use crate::validation::publication_point::PublicationPointSnapshot; +use crate::validation::tree::{CaCertificateRef, CaInstanceHandle, DiscoveredChildCaInstance}; + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct CachedPublicationPoint { + pub history: CaManifestHistory, + context: String, + verified_at: PackTime, + pub(crate) effective_until: PackTime, + vrps: Vec, + aspas: Vec, + router_keys: Vec, + stats: ObjectsStats, + pub audit: PublicationPointAudit, + pub children: Vec, + pub ccr: CcrManifestProjection, +} + +pub(crate) fn identity(store: &RocksStore, ca: &CaInstanceHandle) -> Result { + let der = ca.ca_certificate_der(store)?; + let cert = ResourceCertificate::decode_der(&der).map_err(|e| e.to_string())?; + Ok(CaIdentity::from_spki( + ca.anchor_namespace, + &cert.tbs.subject_public_key_info, + )) +} + +fn context(store: &RocksStore, ca: &CaInstanceHandle, policy: &Policy) -> Result { + // The cache-only marker and traversal depth are control state, not trust context. + // Hash the validated DER, not a lazy storage reference or a mutable URI lookup. + let der = ca.ca_certificate_der(store)?; + let encoded = serde_json::to_vec(&( + ca.anchor_namespace, + &ca.tal_id, + Sha256::digest(&der).as_slice(), + &ca.ca_certificate_rsync_uri, + &ca.parent_manifest_rsync_uri, + &ca.manifest_rsync_uri, + &ca.manifest_sia_uris, + &ca.publication_point_rsync_uri, + &ca.rsync_base_uri, + &ca.rrdp_notification_uri, + &ca.effective_ip_resources, + &ca.effective_as_resources, + policy, + policy + .ta_constraints + .for_tal(&ca.tal_id) + .map(|c| c.cache_fingerprint()), + )) + .map_err(|e| e.to_string())?; + Ok(hex::encode(Sha256::digest(encoded))) +} + +fn certificate_deadline(bytes: &[u8]) -> Result { + let (rest, certificate) = X509Certificate::from_der(bytes).map_err(|e| e.to_string())?; + if !rest.is_empty() { + return Err("cached certificate has trailing bytes".into()); + } + Ok(certificate.validity().not_after.to_datetime()) +} + +fn signed_deadline(bytes: &[u8]) -> Result { + let parsed = RpkiSignedObject::parse_der(bytes).map_err(|e| e.to_string())?; + let certificates = parsed + .signed_data + .certificates + .ok_or("cached signed object has no EE")?; + let ee = certificates + .first() + .ok_or("cached signed object has no EE")?; + certificate_deadline(ee) +} + +impl CachedPublicationPoint { + pub(crate) fn prepare( + store: &RocksStore, + ca: &CaInstanceHandle, + policy: &Policy, + history: &CaManifestHistory, + pack: &PublicationPointSnapshot, + objects: &ObjectsOutput, + audit: &PublicationPointAudit, + children: &[DiscoveredChildCaInstance], + ccr: CcrManifestProjection, + ) -> Result, String> { + if objects.stats.publication_point_dropped { + return Ok(None); + } + let verified_at = pack.verified_at.parse()?; + let mut deadline = pack + .next_update + .parse()? + .min(certificate_deadline(&ca.ca_certificate_der(store)?)?) + .min(signed_deadline(&pack.manifest_bytes)?); + // Only accepted payloads/children constrain their result cache. All + // parseable CRLs are conservative bounds (including the manifest CRL). + let audit_by_uri: std::collections::HashMap<_, _> = audit + .objects + .iter() + .map(|e| (e.rsync_uri.as_str(), e)) + .collect(); + for file in &pack.files { + let entry = audit_by_uri.get(file.rsync_uri.as_str()).copied(); + let Some(entry) = entry.filter(|e| e.result == AuditObjectResult::Ok) else { + continue; + }; + let until = match entry.kind { + AuditObjectKind::Roa | AuditObjectKind::Aspa => signed_deadline(file.bytes()?)?, + AuditObjectKind::Certificate | AuditObjectKind::RouterCertificate => { + certificate_deadline(file.bytes()?)? + } + AuditObjectKind::Crl => { + crate::model::crl::RpkixCrl::decode_der(file.bytes()?) + .map_err(|e| e.to_string())? + .next_update + .utc + } + _ => continue, + }; + deadline = deadline.min(until); + } + for router in &objects.router_keys { + deadline = deadline.min(router.item_effective_until.parse()?); + } + if deadline <= verified_at { + return Ok(None); + } + let mut children = children.to_vec(); + for child in &mut children { + // Self-contained old certificate; no dependency on current repo bytes. + let der = child.handle.ca_certificate_der(store)?.into_owned(); + child.handle.ca_certificate = CaCertificateRef::CachedDer(der); + } + Ok(Some(Self { + history: history.clone(), + context: context(store, ca, policy)?, + verified_at: pack.verified_at.clone(), + effective_until: PackTime::from_utc_offset_datetime(deadline), + vrps: objects.vrps.clone(), + aspas: objects.aspas.clone(), + router_keys: objects.router_keys.clone(), + stats: objects.stats.clone(), + audit: audit.clone(), + children, + ccr, + })) + } + + pub(crate) fn check(&self, expected_context: &str, now: OffsetDateTime) -> Result<(), String> { + if self.context != expected_context { + return Err("failed-fetch cache context changed".into()); + } + let verified = self.verified_at.parse()?; + let deadline = self.effective_until.parse()?; + if now < verified { + return Err("failed-fetch cache is from the future (clock rollback)".into()); + } + if now >= deadline || deadline <= verified { + return Err("failed-fetch cache is stale".into()); + } + if self.stats.publication_point_dropped { + return Err("failed-fetch cache contains dropped results".into()); + } + if self.children.iter().any(|c| !c.handle.is_cache_only()) { + return Err("failed-fetch cache contains a fresh child edge".into()); + } + Ok(()) + } + + pub(crate) fn load( + store: &RocksStore, + ca: &CaInstanceHandle, + policy: &Policy, + now: OffsetDateTime, + ) -> Result { + let identity = identity(store, ca)?; + let cached = store + .get_validated_publication_point(&identity) + .map_err(|e| e.to_string())? + .ok_or("no previously validated publication point")?; + cached.check(&context(store, ca, policy)?, now)?; + Ok(cached) + } + + pub(crate) fn objects(&self) -> ObjectsOutput { + ObjectsOutput { + vrps: self.vrps.clone(), + aspas: self.aspas.clone(), + router_keys: self.router_keys.clone(), + stats: self.stats.clone(), + warnings: Vec::new(), + audit: self.audit.objects.clone(), + } + } +} + +#[cfg(test)] +mod tests; diff --git a/src/validation/failed_fetch/tests.rs b/src/validation/failed_fetch/tests.rs new file mode 100644 index 0000000..5b155fc --- /dev/null +++ b/src/validation/failed_fetch/tests.rs @@ -0,0 +1,532 @@ +use super::*; +use crate::repository::fetch::rsync::{RsyncFetchError, RsyncFetchResult, RsyncFetcher}; +use crate::validation::tree::{PublicationPointRunner, TreeRunConfig, run_tree_serial_audit}; +use crate::validation::tree_runner::Rpkiv1PublicationPointRunner; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +struct Fixture { + failed: AtomicBool, + requests: AtomicUsize, +} +impl crate::repository::sync::rrdp::Fetcher for Fixture { + fn fetch(&self, _: &str) -> Result, String> { + panic!("unexpected HTTP request") + } +} +impl RsyncFetcher for Fixture { + fn fetch_objects(&self, base: &str) -> RsyncFetchResult)>> { + self.requests.fetch_add(1, Ordering::Relaxed); + if self.failed.load(Ordering::Relaxed) { + return Err(RsyncFetchError::Fetch("injected repository outage".into())); + } + let root = crate::test_support::synthetic_repository().case_repository("baseline-v1"); + let relative = url::Url::parse(base) + .unwrap() + .path() + .trim_start_matches("/custom/") + .to_string(); + fn read( + root: &std::path::Path, + path: &std::path::Path, + output: &mut Vec<(String, Vec)>, + ) { + for entry in std::fs::read_dir(path).unwrap() { + let path = entry.unwrap().path(); + if path.is_dir() { + read(root, &path, output); + } else { + output.push(( + format!( + "rsync://rpki-local-rsync:873/custom/{}", + path.strip_prefix(root).unwrap().display() + ), + std::fs::read(path).unwrap(), + )); + } + } + } + let mut output = Vec::new(); + read(&root, &root.join(relative), &mut output); + Ok(output) + } +} +fn fixture() -> Fixture { + Fixture { + failed: AtomicBool::new(false), + requests: AtomicUsize::new(0), + } +} +fn now() -> OffsetDateTime { + OffsetDateTime::from_unix_timestamp(1788825600).unwrap() +} +fn root() -> CaInstanceHandle { + let fixture = crate::test_support::synthetic_repository(); + let discovery = crate::validation::from_tal::discover_root_ca_instance_from_tal_and_ta_der( + &std::fs::read(fixture.tal()).unwrap(), + &std::fs::read(fixture.trust_anchor()).unwrap(), + None, + ) + .unwrap(); + crate::validation::run_tree_from_tal::root_handle_from_trust_anchor( + &discovery.trust_anchor, + "custom".into(), + None, + &discovery.ca_instance, + ) +} +fn runner<'a>( + store: &'a RocksStore, + policy: &'a Policy, + fetcher: &'a Fixture, +) -> Rpkiv1PublicationPointRunner<'a> { + Rpkiv1PublicationPointRunner { + store, + policy, + http_fetcher: fetcher, + rsync_fetcher: fetcher, + validation_time: now(), + timing: None, + download_log: None, + current_repo_index: None, + repo_sync_runtime: None, + parallel_phase2_config: None, + parallel_roa_worker_pool: None, + ccr_accumulator: None, + } +} +fn populate(store: &RocksStore, policy: &Policy, fetcher: &Fixture) -> CachedPublicationPoint { + let tree = run_tree_serial_audit( + root(), + &runner(store, policy, fetcher), + &TreeRunConfig::default(), + ) + .unwrap(); + assert_eq!(tree.tree.vrps.len(), 1); + assert_eq!(tree.tree.aspas.len(), 1); + CachedPublicationPoint::load(store, &root(), policy, now()).unwrap() +} + +#[test] +fn failed_fetch_serial_restart_cache_only_and_depth_limits() { + let temp = tempfile::tempdir().unwrap(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let fetcher = fixture(); + let store = RocksStore::open(temp.path()).unwrap(); + let initial = populate(&store, &policy, &fetcher); + assert_eq!(initial.children.len(), 1); + assert!(initial.children[0].handle.is_cache_only()); + let child = + CachedPublicationPoint::load(&store, &initial.children[0].handle, &policy, now()).unwrap(); + assert_eq!(child.vrps.len(), 1); + assert_eq!(child.aspas.len(), 1); + let history = store + .get_ca_manifest_history(&identity(&store, &root()).unwrap()) + .unwrap(); + drop(store); + let store = RocksStore::open(temp.path()).unwrap(); + fetcher.failed.store(true, Ordering::Relaxed); + for limit in [Some(0), Some(1), None] { + let before = fetcher.requests.load(Ordering::Relaxed); + let tree = run_tree_serial_audit( + root(), + &runner(&store, &policy, &fetcher), + &TreeRunConfig { + max_depth: limit, + build_ccr_accumulator: false, + ..Default::default() + }, + ) + .unwrap(); + assert_eq!( + fetcher.requests.load(Ordering::Relaxed), + before + 1, + "only failed parent may acquire data" + ); + assert_eq!(tree.tree.vrps.len(), usize::from(limit != Some(0))); + assert!( + tree.publication_points + .iter() + .all(|p| p.repo_terminal_state == "failed_cached") + ); + } + assert_eq!( + store + .get_ca_manifest_history(&identity(&store, &root()).unwrap()) + .unwrap(), + history + ); + assert_eq!( + CachedPublicationPoint::load(&store, &root(), &policy, now()) + .unwrap() + .verified_at, + initial.verified_at + ); +} + +#[test] +fn failed_fetch_cache_context_and_temporal_boundaries() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let fetcher = fixture(); + let cache = populate(&store, &policy, &fetcher); + let context = context(&store, &root(), &policy).unwrap(); + assert!(cache.check(&context, now()).is_ok()); + assert!( + cache + .check(&context, now() - time::Duration::seconds(1)) + .unwrap_err() + .contains("future") + ); + let end = cache.effective_until.parse().unwrap(); + assert!( + cache + .check(&context, end - time::Duration::seconds(1)) + .is_ok() + ); + assert!(cache.check(&context, end).unwrap_err().contains("stale")); + assert!( + cache + .check("different", now()) + .unwrap_err() + .contains("context") + ); + let mut changed = policy.clone(); + changed.strict.name = !changed.strict.name; + assert!(CachedPublicationPoint::load(&store, &root(), &changed, now()).is_err()); + let mut changed = root(); + changed.effective_as_resources = Some(crate::model::rc::AsResourceSet { + asnum: None, + rdi: None, + }); + assert!(CachedPublicationPoint::load(&store, &changed, &policy, now()).is_err()); + let mut changed = root(); + changed.manifest_rsync_uri.push_str(".new"); + assert!(CachedPublicationPoint::load(&store, &changed, &policy, now()).is_err()); + let mut changed = root(); + changed.anchor_namespace = [9; 32]; + assert!(CachedPublicationPoint::load(&store, &changed, &policy, now()).is_err()); + let mut reissued = root(); + let der = + std::fs::read(crate::test_support::synthetic_repository().ta_variant("newer")).unwrap(); + reissued.ca_certificate = CaCertificateRef::InlineDer(der); + assert_eq!( + identity(&store, &root()).unwrap(), + identity(&store, &reissued).unwrap() + ); + assert!(CachedPublicationPoint::load(&store, &reissued, &policy, now()).is_err()); + let mut different_ca = root(); + different_ca.ca_certificate = cache.children[0].handle.ca_certificate.clone(); + assert!(CachedPublicationPoint::load(&store, &different_ca, &policy, now()).is_err()); + let mut changed_tal = root(); + changed_tal.tal_id = "another".into(); + assert!(CachedPublicationPoint::load(&store, &changed_tal, &policy, now()).is_err()); + let tal_path = temp.path().join("custom.tal"); + let constraints_path = temp.path().join("changed.constraints"); + std::fs::write(&constraints_path, "allow 192.0.2.0/24\nallow 64496\n").unwrap(); + let mut constrained = policy.clone(); + constrained.ta_constraints = crate::ta_constraints::TaConstraintsByTal::load_for_tals( + &[crate::scheduler::types::TalInputSpec::from_file_path( + tal_path, + )], + &[format!("custom={}", constraints_path.display())], + ) + .unwrap(); + assert!(CachedPublicationPoint::load(&store, &root(), &constrained, now()).is_err()); + let child = + CachedPublicationPoint::load(&store, &cache.children[0].handle, &policy, now()).unwrap(); + let roa = std::fs::read( + crate::test_support::synthetic_repository() + .case_repository("baseline-v1") + .join("child/valid.roa"), + ) + .unwrap(); + assert_eq!( + child.effective_until.parse().unwrap(), + signed_deadline(&roa).unwrap(), + "accepted EE bounds reuse before manifest expires" + ); + let mut invalid = cache.clone(); + invalid.children[0].handle.ca_certificate = CaCertificateRef::InlineDer(vec![]); + assert!( + invalid + .check(&context, now()) + .unwrap_err() + .contains("fresh child") + ); + let mut invalid = cache.clone(); + invalid.stats.publication_point_dropped = true; + assert!( + invalid + .check(&context, now()) + .unwrap_err() + .contains("dropped") + ); + let mut invalid = cache; + invalid.effective_until.rfc3339_utc = "bad".into(); + assert!(invalid.check(&context, now()).is_err()); +} + +#[test] +fn failed_fetch_cache_atomic_replace_clear_and_rejected_commit() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let fetcher = fixture(); + let cache = populate(&store, &policy, &fetcher); + let identity = identity(&store, &root()).unwrap(); + let failure = + store.update_ca_manifest_history_and_cache(&identity, Some(cache.clone()), |_| { + Err::( + crate::repository::storage::StorageError::InvalidData { + entity: "test", + detail: "reject stale candidate".into(), + }, + ) + }); + assert!(failure.is_err()); + assert_eq!( + store + .get_validated_publication_point(&identity) + .unwrap() + .unwrap() + .history, + cache.history + ); + store + .update_ca_manifest_history_and_cache(&identity, None, |old| { + Ok::<_, crate::repository::storage::StorageError>(old.unwrap().clone()) + }) + .unwrap(); + assert!( + store + .get_validated_publication_point(&identity) + .unwrap() + .is_none() + ); + assert_eq!( + store.get_ca_manifest_history(&identity).unwrap(), + Some(cache.history) + ); + fetcher.failed.store(true, Ordering::Relaxed); + let result = runner(&store, &policy, &fetcher) + .run_publication_point(&root()) + .unwrap(); + assert_eq!( + result.source, + crate::validation::manifest::PublicationPointSource::FailedFetchNoCache + ); + assert!(result.discovered_children.is_empty()); +} + +#[test] +fn failed_fetch_cache_corruption_and_history_mismatch_fail_closed() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let fetcher = fixture(); + let cache = populate(&store, &policy, &fetcher); + let id = identity(&store, &root()).unwrap(); + let key = format!( + "ca_manifest:{}:{}", + hex::encode(id.anchor_namespace), + hex::encode(id.spki_sha256) + ); + drop(store); + for corrupt in [true, false] { + { + let db = rocksdb::DB::open_cf( + &rocksdb::Options::default(), + temp.path(), + crate::repository::storage::ALL_COLUMN_FAMILY_NAMES, + ) + .unwrap(); + let cf = db.cf_handle("validated_publication_point").unwrap(); + let mut mismatched = cache.clone(); + mismatched.history.generation += 1; + let value = if corrupt { + b"not CBOR".to_vec() + } else { + serde_cbor::to_vec(&mismatched).unwrap() + }; + db.put_cf(cf, key.as_bytes(), value).unwrap(); + } + let store = RocksStore::open(temp.path()).unwrap(); + assert!(CachedPublicationPoint::load(&store, &root(), &policy, now()).is_err()); + fetcher.failed.store(true, Ordering::Relaxed); + let result = runner(&store, &policy, &fetcher) + .run_publication_point(&root()) + .unwrap(); + assert_eq!( + result.source, + crate::validation::manifest::PublicationPointSource::FailedFetchNoCache + ); + assert!(result.objects.vrps.is_empty()); + assert!( + result + .warnings + .iter() + .any(|w| w.message.contains("cache unavailable")) + ); + } +} + +#[test] +fn failed_fetch_preparation_deadline_and_ineligible_fresh_result() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let fetcher = fixture(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let runner = runner(&store, &policy, &fetcher); + let result = runner.run_publication_point(&root()).unwrap(); + let mut pack = result.snapshot.unwrap(); + let history = store + .get_ca_manifest_history(&identity(&store, &root()).unwrap()) + .unwrap() + .unwrap(); + let cached = CachedPublicationPoint::load(&store, &root(), &policy, now()).unwrap(); + let prepare = |pack: &PublicationPointSnapshot, objects: &ObjectsOutput| { + CachedPublicationPoint::prepare( + &store, + &root(), + &policy, + &history, + pack, + objects, + &result.audit, + &result.discovered_children, + cached.ccr.clone(), + ) + }; + pack.next_update = PackTime::from_utc_offset_datetime(now()); + assert!(prepare(&pack, &result.objects).unwrap().is_none()); + pack.next_update = PackTime::from_utc_offset_datetime(now() + time::Duration::seconds(20)); + assert_eq!( + prepare(&pack, &result.objects) + .unwrap() + .unwrap() + .effective_until, + pack.next_update + ); + let mut dropped = result.objects.clone(); + dropped.stats.publication_point_dropped = true; + assert!(prepare(&pack, &dropped).unwrap().is_none()); + assert!(certificate_deadline(b"invalid").is_err()); + let mut der = root().ca_certificate_der(&store).unwrap().into_owned(); + der.push(0); + assert!(certificate_deadline(&der).unwrap_err().contains("trailing")); + assert!(signed_deadline(b"invalid").is_err()); +} + +#[test] +fn failed_fetch_cache_concurrent_history_and_payload_commit() { + let temp = tempfile::tempdir().unwrap(); + let store = std::sync::Arc::new(RocksStore::open(temp.path()).unwrap()); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let fetcher = fixture(); + let cache = populate(&store, &policy, &fetcher); + let identity = identity(&store, &root()).unwrap(); + let candidate = |number, delta| { + let mut history = cache.history.clone(); + history.manifest_number_be = vec![number]; + history.manifest_this_update = PackTime::from_utc_offset_datetime( + history.manifest_this_update.parse().unwrap() + time::Duration::seconds(delta), + ); + history.manifest_sha256 = vec![number; 32]; + crate::validation::manifest::ManifestHistoryCandidate { + identity: identity.clone(), + manifest_sia_uris: vec![history.manifest_rsync_uri.clone()], + history, + } + }; + let newer = candidate(3, 2); + let older = candidate(2, 1); + let barrier = std::sync::Barrier::new(2); + std::thread::scope(|scope| { + scope.spawn(|| { + let mut newer_cache = cache.clone(); + newer_cache.ccr.manifest_number_be = vec![3]; + crate::validation::manifest::commit_manifest_history_with_cache( + &store, + &newer, + Some(newer_cache), + ) + .unwrap(); + barrier.wait(); + }); + scope.spawn(|| { + barrier.wait(); + assert!( + crate::validation::manifest::commit_manifest_history_with_cache( + &store, + &older, + Some(cache.clone()) + ) + .is_err() + ); + for _ in 0..20 { + let loaded = store + .get_validated_publication_point(&identity) + .unwrap() + .unwrap(); + assert_eq!(loaded.history.manifest_number_be, vec![3]); + assert_eq!(loaded.ccr.manifest_number_be, vec![3]); + } + }); + }); +} + +#[test] +fn failed_fetch_missing_child_cache_stops_only_that_branch_without_fetch() { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let fetcher = fixture(); + let policy = Policy { + sync_preference: crate::validation::policy::SyncPreference::RsyncOnly, + ..Default::default() + }; + let cache = populate(&store, &policy, &fetcher); + let id = identity(&store, &cache.children[0].handle).unwrap(); + store + .update_ca_manifest_history_and_cache(&id, None, |old| { + Ok::<_, crate::repository::storage::StorageError>(old.unwrap().clone()) + }) + .unwrap(); + fetcher.failed.store(true, Ordering::Relaxed); + for instances in [Some(1), Some(2), None] { + let before = fetcher.requests.load(Ordering::Relaxed); + let tree = run_tree_serial_audit( + root(), + &runner(&store, &policy, &fetcher), + &TreeRunConfig { + max_instances: instances, + ..Default::default() + }, + ) + .unwrap(); + assert!(tree.tree.vrps.is_empty()); + assert_eq!(fetcher.requests.load(Ordering::Relaxed), before + 1); + assert_eq!(tree.publication_points[0].source, "failed_fetch_cached"); + if instances != Some(1) { + assert_eq!(tree.publication_points[1].source, "failed_fetch_no_cache"); + } + } +} diff --git a/src/validation/manifest/history.rs b/src/validation/manifest/history.rs index f9b8e55..bc5138b 100644 --- a/src/validation/manifest/history.rs +++ b/src/validation/manifest/history.rs @@ -88,23 +88,33 @@ fn checked_manifest_history( Ok(next) } +#[cfg(test)] pub(crate) fn commit_manifest_history( store: &RocksStore, candidate: &ManifestHistoryCandidate, +) -> Result<(CaManifestHistory, Option), ManifestFreshError> { + commit_manifest_history_with_cache(store, candidate, None) +} + +pub(crate) fn commit_manifest_history_with_cache( + store: &RocksStore, + candidate: &ManifestHistoryCandidate, + cache: Option, ) -> Result<(CaManifestHistory, Option), ManifestFreshError> { let mut warning = None; - let history = store.update_ca_manifest_history(&candidate.identity, |old| { - let next = checked_manifest_history(old, candidate)?; - warning = old.and_then(|old| { - rename_warning_for_history( - old, - &candidate.identity, - &candidate.manifest_sia_uris, - &candidate.history.manifest_rsync_uri, - ) - }); - Ok::<_, ManifestFreshError>(next) - })?; + let history = + store.update_ca_manifest_history_and_cache(&candidate.identity, cache, |old| { + let next = checked_manifest_history(old, candidate)?; + warning = old.and_then(|old| { + rename_warning_for_history( + old, + &candidate.identity, + &candidate.manifest_sia_uris, + &candidate.history.manifest_rsync_uri, + ) + }); + Ok::<_, ManifestFreshError>(next) + })?; Ok((history, warning)) } diff --git a/src/validation/manifest/models_and_process.rs b/src/validation/manifest/models_and_process.rs index 0a8c684..4e0c7bc 100644 --- a/src/validation/manifest/models_and_process.rs +++ b/src/validation/manifest/models_and_process.rs @@ -4,6 +4,7 @@ pub enum PublicationPointSource { Fresh, FailedFetchNoCache, + FailedFetchCached, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/src/validation/mod.rs b/src/validation/mod.rs index cab35a5..e79f5d3 100644 --- a/src/validation/mod.rs +++ b/src/validation/mod.rs @@ -1,6 +1,7 @@ pub mod ca_instance; pub mod ca_path; pub mod cert_path; +pub(crate) mod failed_fetch; pub mod from_tal; pub mod manifest; pub mod objects; diff --git a/src/validation/objects/outputs.rs b/src/validation/objects/outputs.rs index b250371..5a02891 100644 --- a/src/validation/objects/outputs.rs +++ b/src/validation/objects/outputs.rs @@ -58,18 +58,18 @@ fn extra_rfc_refs_for_crl_selection(error: &ObjectValidateError) -> &'static [Rf } } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct Vrp { pub asn: u32, pub prefix: IpPrefix, pub max_length: u16, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct AspaAttestation { pub customer_as_id: u32, pub provider_as_ids: Vec, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct RouterKeyPayload { pub as_id: u32, pub ski: Vec, @@ -89,7 +89,7 @@ pub struct ObjectsOutput { pub stats: ObjectsStats, pub audit: Vec, } -#[derive(Clone, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ObjectsStats { pub roa_total: usize, pub roa_ok: usize, diff --git a/src/validation/tree.rs b/src/validation/tree.rs index 7130cd0..e553629 100644 --- a/src/validation/tree.rs +++ b/src/validation/tree.rs @@ -4,6 +4,7 @@ use crate::output::report::Warning; use crate::validation::manifest::PublicationPointSource; use crate::validation::objects::{AspaAttestation, ObjectsOutput, RouterKeyPayload, Vrp}; use crate::validation::publication_point::PublicationPointSnapshot; +use serde::{Deserialize, Serialize}; use std::borrow::Cow; use std::sync::{Arc, OnceLock}; @@ -41,11 +42,15 @@ pub(crate) fn next_allowed_ca_depth(config: &TreeRunConfig, parent_depth: usize) ca_depth_is_allowed(config, child_depth).then_some(child_depth) } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub enum CaCertificateRef { InlineDer(Vec), + /// A previously validated child reached through a failed-fetch cache edge. + /// Its entire branch must use cached results without repository requests. + CachedDer(Vec), RepoBytes { sha256_hex: String, + #[serde(skip)] cached_der: Arc>>, }, } @@ -53,7 +58,8 @@ pub enum CaCertificateRef { impl PartialEq for CaCertificateRef { fn eq(&self, other: &Self) -> bool { match (self, other) { - (Self::InlineDer(left), Self::InlineDer(right)) => left == right, + (Self::InlineDer(left), Self::InlineDer(right)) + | (Self::CachedDer(left), Self::CachedDer(right)) => left == right, ( Self::RepoBytes { sha256_hex: left, .. @@ -83,7 +89,7 @@ impl CaCertificateRef { pub fn sha256_hex(&self) -> Option<&str> { match self { - Self::InlineDer(_) => None, + Self::InlineDer(_) | Self::CachedDer(_) => None, Self::RepoBytes { sha256_hex, .. } => Some(sha256_hex.as_str()), } } @@ -93,7 +99,7 @@ impl CaCertificateRef { store: &crate::repository::storage::RocksStore, ) -> Result, String> { match self { - Self::InlineDer(bytes) => Ok(Cow::Borrowed(bytes.as_slice())), + Self::InlineDer(bytes) | Self::CachedDer(bytes) => Ok(Cow::Borrowed(bytes.as_slice())), Self::RepoBytes { sha256_hex, cached_der, @@ -116,7 +122,7 @@ impl CaCertificateRef { } } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct CaInstanceHandle { pub anchor_namespace: [u8; 32], pub depth: usize, @@ -142,6 +148,10 @@ pub struct CaInstanceHandle { } impl CaInstanceHandle { + pub fn is_cache_only(&self) -> bool { + matches!(self.ca_certificate, CaCertificateRef::CachedDer(_)) + } + pub fn with_depth(mut self, depth: usize) -> Self { self.depth = depth; self @@ -160,7 +170,7 @@ impl CaInstanceHandle { pub fn ca_certificate_sha256_32(&self) -> Option<[u8; 32]> { match &self.ca_certificate { - CaCertificateRef::InlineDer(bytes) => { + CaCertificateRef::InlineDer(bytes) | CaCertificateRef::CachedDer(bytes) => { use sha2::Digest as _; let digest = sha2::Sha256::digest(bytes); Some(digest.into()) @@ -184,17 +194,18 @@ pub struct PublicationPointRunResult { /// Candidate child CA instances to enqueue after this publication point completes. /// /// - For `Fresh`, these are discovered from the current validated publication point. - /// - For failed synchronization, the result contains no child instances; - /// callers may retry the repository on a later run. + /// - For `FailedFetchCached`, these are cache-only edges: no descendant + /// acquisition or fresh validation is permitted during this run. + /// - Without usable cached results, this list is empty. pub discovered_children: Vec, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct DiscoveredChildEntryProjection { pub child_ski: String, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct DiscoveredChildCaInstance { pub handle: CaInstanceHandle, pub discovered_from: DiscoveredFrom, diff --git a/src/validation/tree_parallel/ready_stage.rs b/src/validation/tree_parallel/ready_stage.rs index 128493b..f699e02 100644 --- a/src/validation/tree_parallel/ready_stage.rs +++ b/src/validation/tree_parallel/ready_stage.rs @@ -93,6 +93,21 @@ fn start_queued_ca_instances( continue; } *instances_started += 1; + if node.handle.is_cache_only() { + ready_queue.push_back(ReadyCaInstance { + node, + ready_enqueued_at: Instant::now(), + repo_outcome: RepoSyncRuntimeOutcome { + repo_sync_ok: false, + repo_sync_err: None, + repo_sync_source: Some("cache".into()), + repo_sync_phase: Some("cache_only".into()), + repo_sync_duration_ms: 0, + warnings: Vec::new(), + }, + }); + continue; + } match repo_runtime.request_publication_point_repo(&node.handle, 0) { Ok(RepoSyncRequestStatus::Ready { mut outcome, .. }) => { // Ready here means this CA is reusing repo work that has already completed @@ -138,6 +153,15 @@ fn compute_ready_publication_point_stage( ready_queue_len_after_pop, ..ReadyStageMetrics::default() }; + if ready.node.handle.is_cache_only() { + return ( + StageOutcome::FreshError(Box::new(FreshErrorOutcome { + ready, + publication_point_started, + })), + metrics, + ); + } let mut warnings = ready.repo_outcome.warnings.clone(); let repo_outcome = ready.repo_outcome.clone(); let stage_fresh_started = Instant::now(); diff --git a/src/validation/tree_runner/fresh_pipeline.rs b/src/validation/tree_runner/fresh_pipeline.rs index 94a7451..7900d93 100644 --- a/src/validation/tree_runner/fresh_pipeline.rs +++ b/src/validation/tree_runner/fresh_pipeline.rs @@ -276,78 +276,29 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { snapshot_pack_ms, ); - let mut ccr_projection = None; - let mut ccr_projection_build_ms = 0; - let mut ccr_append_ms = 0; - if self.ccr_accumulator.is_some() { - let ccr_projection_build_started = std::time::Instant::now(); - let subordinate_skis = discovered_children - .iter() - .map(|child| { - if let Some(projection) = &child.child_entry_projection { - return hex::decode(&projection.child_ski) - .map_err(|error| error.to_string()); - } - let der = child.handle.ca_certificate_der(self.store)?; - let certificate = ResourceCertificate::decode_der(der.as_ref()) - .map_err(|error| error.to_string())?; - certificate - .tbs - .extensions - .subject_key_identifier - .clone() - .ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string()) - }) - .collect::, String>>()?; - let ccr_manifest_projection = - crate::ccr::projection::from_snapshot(ca, &pack, subordinate_skis)?; - ccr_projection_build_ms = ccr_projection_build_started.elapsed().as_millis() as u64; - ccr_projection = Some(ccr_manifest_projection); - } - let (_, commit_warning) = crate::validation::manifest::commit_manifest_history( - self.store, - &fresh_point.history_candidate, - ) - .map_err(|error| format!("persist manifest freshness failed: {error}"))?; - if let Some(warning) = commit_warning { - warnings.retain(|item| !item.message.starts_with("manifest filename changed:")); - warnings.push(warning); - } - for warning in &mut warnings { - if warning.message.starts_with("manifest filename changed:") { - warning.message.push_str(&format!( - " new_number={} new_this_update={} candidate_accepted=true", - hex::encode_upper(&pack.manifest_number_be), - pack.this_update.rfc3339_utc - )); - } - } - if let Some(projection) = ccr_projection { - let started = std::time::Instant::now(); - self.append_ccr_manifest_projection(&projection)?; - ccr_append_ms = started.elapsed().as_millis() as u64; - } - if let Some(timing) = self.timing.as_ref() { - timing.record_phase_nanos( - "fresh_ccr_projection_build_total", - ccr_projection_build_ms.saturating_mul(1_000_000), - ); - timing.record_phase_nanos( - "fresh_ccr_append_total", - ccr_append_ms.saturating_mul(1_000_000), - ); - } - self.record_publication_point_step_ms( - &ca.manifest_rsync_uri, - "fresh_ccr_projection_build", - ccr_projection_build_ms, - ); - self.record_publication_point_step_ms( - &ca.manifest_rsync_uri, - "fresh_ccr_append", - ccr_append_ms, - ); - + let ccr_projection_build_started = std::time::Instant::now(); + let subordinate_skis = discovered_children + .iter() + .map(|child| { + if let Some(projection) = &child.child_entry_projection { + return hex::decode(&projection.child_ski).map_err(|error| error.to_string()); + } + let der = child.handle.ca_certificate_der(self.store)?; + let certificate = ResourceCertificate::decode_der(der.as_ref()) + .map_err(|error| error.to_string())?; + certificate + .tbs + .extensions + .subject_key_identifier + .clone() + .ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string()) + }) + .collect::, String>>()?; + let ccr_manifest_projection = + crate::ccr::projection::from_snapshot(ca, &pack, subordinate_skis)?; + let ccr_projection_build_ms = ccr_projection_build_started.elapsed().as_millis() as u64; + let ccr_projection = ccr_manifest_projection; + let ccr_append_ms; let audit_build_started = std::time::Instant::now(); let audit = build_publication_point_audit_from_snapshot( ca, @@ -374,6 +325,60 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { audit_build_ms, ); + let cache = crate::validation::failed_fetch::CachedPublicationPoint::prepare( + self.store, + ca, + self.policy, + &fresh_point.history_candidate.history, + &pack, + &objects, + &audit, + &discovered_children, + ccr_projection.clone(), + )?; + let (_, commit_warning) = crate::validation::manifest::commit_manifest_history_with_cache( + self.store, + &fresh_point.history_candidate, + cache, + ) + .map_err(|error| format!("persist manifest freshness failed: {error}"))?; + if let Some(warning) = commit_warning { + warnings.retain(|item| !item.message.starts_with("manifest filename changed:")); + warnings.push(warning); + } + for warning in &mut warnings { + if warning.message.starts_with("manifest filename changed:") { + warning.message.push_str(&format!( + " new_number={} new_this_update={} candidate_accepted=true", + hex::encode_upper(&pack.manifest_number_be), + pack.this_update.rfc3339_utc + )); + } + } + let ccr_append_started = std::time::Instant::now(); + self.append_ccr_manifest_projection(&ccr_projection)?; + ccr_append_ms = ccr_append_started.elapsed().as_millis() as u64; + if let Some(timing) = self.timing.as_ref() { + timing.record_phase_nanos( + "fresh_ccr_projection_build_total", + ccr_projection_build_ms.saturating_mul(1_000_000), + ); + timing.record_phase_nanos( + "fresh_ccr_append_total", + ccr_append_ms.saturating_mul(1_000_000), + ); + } + self.record_publication_point_step_ms( + &ca.manifest_rsync_uri, + "fresh_ccr_projection_build", + ccr_projection_build_ms, + ); + self.record_publication_point_step_ms( + &ca.manifest_rsync_uri, + "fresh_ccr_append", + ccr_append_ms, + ); + Ok(FreshPublicationPointFinalizeOutput { result: PublicationPointRunResult { source: PublicationPointSource::Fresh, diff --git a/src/validation/tree_runner/labels.rs b/src/validation/tree_runner/labels.rs index 98f3449..0c57899 100644 --- a/src/validation/tree_runner/labels.rs +++ b/src/validation/tree_runner/labels.rs @@ -24,6 +24,7 @@ pub(super) fn kind_from_rsync_uri(uri: &str) -> AuditObjectKind { pub(super) fn source_label(source: PublicationPointSource) -> String { match source { PublicationPointSource::Fresh => "fresh".to_string(), + PublicationPointSource::FailedFetchCached => "failed_fetch_cached".to_string(), PublicationPointSource::FailedFetchNoCache => "failed_fetch_no_cache".to_string(), } } @@ -59,6 +60,7 @@ pub(super) fn repo_sync_failure_phase_label( pub(super) fn terminal_state_label(source: PublicationPointSource) -> &'static str { match source { PublicationPointSource::Fresh => "fresh", + PublicationPointSource::FailedFetchCached => "failed_cached", PublicationPointSource::FailedFetchNoCache => "failed_no_cache", } } diff --git a/src/validation/tree_runner/publication_point_runner.rs b/src/validation/tree_runner/publication_point_runner.rs index 0435255..93db709 100644 --- a/src/validation/tree_runner/publication_point_runner.rs +++ b/src/validation/tree_runner/publication_point_runner.rs @@ -21,6 +21,20 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { }), ); + if ca.is_cache_only() { + return self.failed_fetch_result( + ca, + Vec::new(), + ManifestFreshError::RepoSyncFailed { + detail: "ancestor failed-fetch: cached subtree only".into(), + }, + Some("cache"), + Some("cache_only"), + 0, + None, + ); + } + let attempted_rrdp = self.policy.sync_preference == crate::validation::policy::SyncPreference::RrdpThenRsync; let repo_started = std::time::Instant::now(); @@ -97,45 +111,15 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { let stage = match stage { Ok(stage) => stage, Err(stage_error) => { - let mut warnings = stage_error.warnings; - warnings.push( - Warning::new(format!( - "publication point processing failed: {}", - stage_error.error - )) - .with_category(stage_error.error.warning_category()) - .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) - .with_context(&ca.manifest_rsync_uri), - ); - let audit = build_publication_point_audit_from_failed_fetch( + return self.failed_fetch_result( ca, + stage_error.warnings, + stage_error.error, repo_sync_source.as_deref(), repo_sync_phase.as_deref(), - Some(repo_sync_duration_ms), + repo_sync_duration_ms, repo_sync_error.as_deref(), - &warnings, - &stage_error.error, ); - crate::logging::progress::emit!( - "repo_terminal_failure", - serde_json::json!({ - "manifest_rsync_uri": ca.manifest_rsync_uri, - "repo_sync_source": repo_sync_source, - "repo_sync_phase": repo_sync_phase, - "repo_sync_error": repo_sync_error, - "repo_sync_duration_ms": repo_sync_duration_ms, - "terminal_state": "failed_no_cache", - "error": stage_error.error.to_string(), - }), - ); - return Ok(PublicationPointRunResult { - source: PublicationPointSource::FailedFetchNoCache, - snapshot: None, - warnings, - objects: empty_objects_output(), - audit, - discovered_children: Vec::new(), - }); } }; @@ -231,3 +215,107 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { Ok(result) } } + +impl Rpkiv1PublicationPointRunner<'_> { + fn failed_fetch_result( + &self, + ca: &CaInstanceHandle, + mut warnings: Vec, + error: ManifestFreshError, + repo_source: Option<&str>, + repo_phase: Option<&str>, + duration: u64, + repo_error: Option<&str>, + ) -> Result { + warnings.push( + Warning::new(format!("publication point processing failed: {error}")) + .with_category(error.warning_category()) + .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) + .with_context(&ca.manifest_rsync_uri), + ); + // Storage failure must not be disguised by reuse of another record. + let cached = if matches!( + error, + ManifestFreshError::Storage(_) | ManifestFreshError::IssuerCaLoadFailed { .. } + ) { + Err("failed-fetch cache disabled after storage/issuer failure".to_string()) + } else { + crate::validation::failed_fetch::CachedPublicationPoint::load( + self.store, + ca, + self.policy, + self.validation_time, + ) + }; + let result = match cached { + Ok(cache) => { + crate::logging::progress::emit!( + "failed_fetch_cache_reused", + serde_json::json!({ + "manifest_rsync_uri": ca.manifest_rsync_uri, + "verified_at": cache.audit.verified_at_rfc3339_utc, + "effective_until": cache.effective_until.rfc3339_utc, + "repo_sync_phase": repo_phase, + "children": cache.children.len(), + }) + ); + self.append_ccr_manifest_projection(&cache.ccr)?; + warnings.push(Warning::new("using previously validated publication point; no fresh subordinate work (RFC 9286 §6.6)") + .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]).with_context(&ca.manifest_rsync_uri)); + let objects = cache.objects(); + let mut audit = cache.audit; + audit.source = source_label(PublicationPointSource::FailedFetchCached); + audit.repo_terminal_state = "failed_cached".into(); + audit.repo_sync_source = repo_source.map(str::to_owned); + audit.repo_sync_phase = repo_phase.map(str::to_owned); + audit.repo_sync_duration_ms = Some(duration); + audit.repo_sync_error = repo_error.map(str::to_owned); + audit + .warnings + .extend(warnings.iter().map(AuditWarning::from)); + PublicationPointRunResult { + source: PublicationPointSource::FailedFetchCached, + snapshot: None, + warnings, + objects, + audit, + discovered_children: cache.children, + } + } + Err(reason) => { + warnings.push( + Warning::new(format!("failed-fetch cache unavailable: {reason}")) + .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) + .with_context(&ca.manifest_rsync_uri), + ); + let audit = build_publication_point_audit_from_failed_fetch( + ca, + repo_source, + repo_phase, + Some(duration), + repo_error, + &warnings, + &error, + ); + PublicationPointRunResult { + source: PublicationPointSource::FailedFetchNoCache, + snapshot: None, + warnings, + objects: empty_objects_output(), + audit, + discovered_children: Vec::new(), + } + } + }; + crate::logging::progress::emit!( + "repo_terminal_failure", + serde_json::json!({ + "manifest_rsync_uri": ca.manifest_rsync_uri, "repo_sync_source": repo_source, + "repo_sync_phase": repo_phase, "repo_sync_error": repo_error, + "repo_sync_duration_ms": duration, "terminal_state": terminal_state_label(result.source), + "error": error.to_string(), + }) + ); + Ok(result) + } +} diff --git a/tests/failed_fetch_e2e.rs b/tests/failed_fetch_e2e.rs new file mode 100644 index 0000000..ac87126 --- /dev/null +++ b/tests/failed_fetch_e2e.rs @@ -0,0 +1,22 @@ +#[test] +fn valid_failed_fetch_results_and_subtree_survive_cli_restart() { + let status = std::process::Command::new("python3") + .arg("tests/support/test_failed_fetch.py") + .arg(env!("CARGO_BIN_EXE_panda-rpki")) + .status() + .expect("start failed-fetch CLI driver"); + assert!(status.success(), "failed-fetch E2E failed: {status}"); +} + +#[test] +fn signed_failed_fetch_mutations_over_both_transports() { + let status = std::process::Command::new("python3") + .arg("tests/support/test_failed_fetch_extended.py") + .arg(env!("CARGO_BIN_EXE_panda-rpki")) + .status() + .expect("start extended failed-fetch driver"); + assert!( + status.success(), + "extended failed-fetch E2E failed: {status}" + ); +} diff --git a/tests/support/test_failed_fetch.py b/tests/support/test_failed_fetch.py new file mode 100644 index 0000000..c70e68e --- /dev/null +++ b/tests/support/test_failed_fetch.py @@ -0,0 +1,72 @@ +"""Real CLI restarts: failed-fetch must retain valid per-instance results.""" +from pathlib import Path +import argparse +import json +import shutil +import subprocess +import sys +import tempfile + +from test_public_config import make_rsync_wrapper, read_json, run_checked, write_config + + +def run(binary: str, evidence: Path | None = None, expect_no_cache: bool = False) -> dict: + proof = {'runs': []} + with tempfile.TemporaryDirectory(prefix='panda-failed-fetch-') as temp: + base = Path(temp) + fixture = base / 'fixture' + run_checked([sys.executable, str(Path(__file__).with_name('generate_repository.py')), + '--output', str(fixture), '--rsync-host', '127.0.0.1', '--strict-names']) + repository = base / 'repository' + shutil.copytree(fixture / 'cases/baseline-v1/repository', repository) + wrapper = base / 'rsync.py' + log = base / 'rsync.log' + make_rsync_wrapper(wrapper, fixture, log) + wrapper.write_text(wrapper.read_text().replace(str(fixture / 'cases/baseline-v1/repository'), str(repository))) + wrapper.chmod(0o755) + baseline = None + for name in ('initial', 'missing-object', 'missing-parent', 'recovered'): + if name == 'missing-object': + (repository / 'child/valid.roa').write_bytes(b'corrupted downloaded ROA') + if name == 'missing-parent': + (repository / 'root/root.mft').write_bytes(b'corrupted downloaded manifest') + if name == 'recovered': + shutil.rmtree(repository) + shutil.copytree(fixture / 'cases/baseline-v2/repository', repository) + out = base / name + config = base / 'validate.toml' + write_config(config, fixture=fixture, output=out, state=base / 'state', ccr=out / 'result.ccr', + wrapper=wrapper, mirror=base / 'mirror', sync_preference='rsync-only', rrdp_mode='auto', + validation_time='2026-06-01T00:00:00Z', resource_mode='rfc6487', scope='publication-point') + start = len(log.read_text().splitlines()) if log.exists() else 0 + result = subprocess.run([binary, 'validate', '--config', str(config)], capture_output=True, text=True, timeout=120) + summary = read_json(out / 'summary.json') + expected = 2 if name == 'recovered' else (0 if expect_no_cache and name.startswith('missing') else 1) + assert summary['vrps'] == expected, (name, summary, result.stderr[-4000:]) + if name.startswith('missing'): + assert summary['resultStatus'] == 'partial', summary + assert result.returncode == 3, result + else: + assert summary['resultStatus'] == 'success' and result.returncode == 0, (summary, result.stderr[-3000:]) + csv = (out / 'vrps.csv').read_bytes() + if name == 'initial': + baseline = csv + elif name.startswith('missing') and not expect_no_cache: + assert baseline == csv, 'failed-fetch changed still-valid payloads' + requests = log.read_text().splitlines()[start:] + if name == 'missing-parent': + assert len(requests) == 1, ('cached subtree must not fetch children', requests) + proof['runs'].append({'name': name, 'status': summary['resultStatus'], 'vrps': summary['vrps'], + 'exit': result.returncode, 'rsync_requests': requests}) + if evidence: + evidence.write_text(json.dumps(proof, indent=2) + '\n') + return proof + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('binary') + parser.add_argument('--evidence', type=Path) + parser.add_argument('--expect-no-cache', action='store_true') + args = parser.parse_args() + print(json.dumps(run(args.binary, args.evidence, args.expect_no_cache))) diff --git a/tests/support/test_failed_fetch_extended.py b/tests/support/test_failed_fetch_extended.py new file mode 100644 index 0000000..a157629 --- /dev/null +++ b/tests/support/test_failed_fetch_extended.py @@ -0,0 +1,214 @@ +"""Signed, multi-level repository mutations and real CLI restarts over both transports.""" +import argparse, functools, hashlib, http.server, json, shutil, ssl, subprocess, sys, tempfile, threading +from datetime import timedelta +from pathlib import Path +from cryptography import x509 +from cryptography.hazmat.primitives import serialization, hashes +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import Encoding +import generate_repository as g +from test_public_config import free_port, make_rsync_wrapper, read_json, run_checked, write_config + + +def canonical_ccr(blob): + # producedAt describes artifact generation, not the validated state. Compare + # every other DER byte while normalizing only that top-level CCR timestamp. + def tlv(offset): + tag = blob[offset]; length = blob[offset+1]; start = offset+2 + if length & 128: + count = length & 127; length = int.from_bytes(blob[start:start+count], 'big'); start += count + return tag, start, start+length + tag, outer, end = tlv(0); assert tag == 0x30 and end == len(blob) + _, _, oid_end = tlv(outer) + tag, explicit, _ = tlv(oid_end); assert tag == 0xa0 + tag, start, end = tlv(explicit); assert tag == 0x30 + while start < end: + tag, value, next_field = tlv(start) + if tag == 0x18: + return blob[:value] + b'0'*(next_field-value) + blob[next_field:] + start = next_field + raise AssertionError('CCR producedAt missing') + + +def run(binary, evidence=None): + proof = {'transports': [], 'binary_sha256': hashlib.sha256(Path(binary).read_bytes()).hexdigest(), 'ccr_comparison': 'all DER bytes except producedAt; CSV byte identical'} + with tempfile.TemporaryDirectory(prefix='panda-failed-fetch-extended-') as temp: + base = Path(temp); fixture = base/'fixture'; port = free_port() + run_checked([sys.executable, str(Path(__file__).with_name('generate_repository.py')), '--output', str(fixture), + '--rrdp-host', '127.0.0.1', '--rrdp-port', str(port), '--rsync-host', '127.0.0.1', '--strict-names']) + g.STRICT_NAMES = True + root_uri = 'rsync://127.0.0.1:873/custom/root/' + child_uri = 'rsync://127.0.0.1:873/custom/child/' + grand_uri = 'rsync://127.0.0.1:873/custom/grand/' + sibling_uri = 'rsync://127.0.0.1:873/custom/sibling/' + notify = f'https://127.0.0.1:{port}/rrdp/notification.xml' + active = base/'active'; active.mkdir(); (active/'rrdp').mkdir() + repository = base/'repository' + requests = [] + class Handler(http.server.SimpleHTTPRequestHandler): + def do_GET(self): requests.append(self.path); super().do_GET() + def log_message(self, *_): pass + server = http.server.ThreadingHTTPServer(('127.0.0.1', port), functools.partial(Handler, directory=str(active))) + tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + tls.load_cert_chain(fixture/'certs/rrdp-server.pem', fixture/'certs/rrdp-server.key') + server.socket = tls.wrap_socket(server.socket, server_side=True) + thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start() + original = fixture/'cases/baseline-v1/repository' + root = x509.load_der_x509_certificate((original/'ta.cer').read_bytes()) + child = x509.load_der_x509_certificate((original/'root/child.cer').read_bytes()) + root_key = serialization.load_pem_private_key((fixture/'keys/root.key').read_bytes(), None) + child_key = serialization.load_pem_private_key((fixture/'keys/child.key').read_bytes(), None) + work = base/'signing'; work.mkdir() + def manifest(folder, number, names, issuer, key, ca_uri): + uri = {'root': root_uri, 'child': child_uri, 'grand': grand_uri, 'sibling': sibling_uri}[folder] + ee_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + ee = g.sign_cert(x509.Name([g.common_name('Failed Fetch Manifest')]), issuer, ee_key.public_key(), key, + x509.random_serial_number(), False, ee_key, key, ip_ext=g.ip_inherit(), as_ext=g.as_resources(None), + signed_object_uri=uri+folder+'.mft', crl_uri=uri+folder+'.crl', ca_issuer_uri=ca_uri) + content = g.manifest_content(number, [(n, (repository/folder/n).read_bytes()) for n in names]) + g.write_cms(content, g.OID_MANIFEST_ECONTENT, ee, ee_key, repository/folder/(folder+'.mft'), work) + def publish(index): + session = f'{index:08d}-1111-4111-8111-222222222222' + objects = sorted(p for p in repository.rglob('*') if p.is_file() and p.suffix in {'.mft', '.crl', '.cer', '.roa', '.asa'}) + snapshot = (f'' + + ''.join(g.publish('rsync://127.0.0.1:873/custom/'+str(p.relative_to(repository)), p) for p in objects) + + '').encode() + (active/'rrdp/snapshot.xml').write_bytes(snapshot) + (active/'rrdp/notification.xml').write_text(f'') + def reset(case): + if repository.exists(): shutil.rmtree(repository) + shutil.copytree(fixture/'cases'/case/'repository', repository) + try: + for transport in ('rsync-only', 'rrdp-then-rsync'): + target = base/transport; target.mkdir(); log = target/'rsync.log'; wrapper = target/'rsync.py' + make_rsync_wrapper(wrapper, fixture, log) + wrapper.write_text(wrapper.read_text().replace(str(original), str(repository)).replace( + 'destination.mkdir(parents=True, exist_ok=True)', + 'if destination.exists(): shutil.rmtree(destination)\ndestination.mkdir(parents=True, exist_ok=True)')) + wrapper.chmod(0o755) + reset('baseline-v1'); runs = []; initial_vrps = initial_aspas = initial_ccr = None + def execute(name, expected, partial=False, *, stamp='2026-06-01T00:00:00Z', config_change=None, no_child_fetch=False): + nonlocal initial_vrps, initial_aspas, initial_ccr + publish(len(runs)+1) + out = target/name; config = target/'validate.toml' + write_config(config, fixture=fixture, output=out, state=target/'state', ccr=out/'result.ccr', wrapper=wrapper, + mirror=target/'mirror', sync_preference=transport, rrdp_mode='auto', validation_time=stamp, + resource_mode='rfc6487', scope='publication-point') + if config_change: config.write_text(config_change(config.read_text())) + start = len(requests); rsync_start = len(log.read_text().splitlines()) if log.exists() else 0 + result = subprocess.run([binary, 'validate', '--config', str(config)], capture_output=True, text=True, timeout=120) + summary = read_json(out/'summary.json') + assert summary['vrps'] == expected, (transport, name, summary, result.stderr[-5000:]) + assert summary['resultStatus'] == ('partial' if partial else 'success'), (name, summary, result.stderr[-5000:]) + assert result.returncode == (3 if partial else 0), (name, result.returncode, result.stderr[-5000:]) + events = [] + for line in result.stderr.splitlines(): + try: event = json.loads(line) + except ValueError: continue + if event.get('event') == 'failed_fetch_cache_reused': events.append(event['fields']) + cached = summary['failed_fetch_cached'] + # Structured cache events also expose the unchanged validity bounds. + assert cached == len(events), (name, cached, events) + rsync = log.read_text().splitlines()[rsync_start:] if log.exists() else [] + if no_child_fetch: + if transport == 'rsync-only': assert len(rsync) == 1, (name, rsync) + assert sum(e['repo_sync_phase'] == 'cache_only' for e in events) == max(0, cached-1), events + if name == 'initial': + initial_vrps = (out/'vrps.csv').read_bytes(); initial_aspas = (out/'vaps.csv').read_bytes(); initial_ccr = canonical_ccr((out/'result.ccr').read_bytes()) + if name in {'corrupt-object', 'repeat-failure', 'missing-object', 'bad-crl', 'corrupt-parent'}: + assert (out/'vrps.csv').read_bytes() == initial_vrps + assert (out/'vaps.csv').read_bytes() == initial_aspas + assert canonical_ccr((out/'result.ccr').read_bytes()) == initial_ccr + if transport == 'rrdp-then-rsync': assert not rsync, ('unexpected rsync fallback', name, rsync) + runs.append({'name': name, 'vrps': expected, 'status': summary['resultStatus'], 'cached': cached, + 'publication_points': summary['publication_points'], 'http_requests': requests[start:], 'rsync_requests': len(rsync), + 'verified_at': [p['verified_at'] for p in sorted(events, key=lambda e:e['manifest_rsync_uri'])]}) + execute('initial', 1) + good_roa = (repository/'child/valid.roa').read_bytes() + (repository/'child/valid.roa').write_bytes(b'corrupt downloaded object') + execute('corrupt-object', 1, True) + execute('repeat-failure', 1, True, config_change=lambda s:s.replace('parallel-phase2-object-workers = 2', 'parallel-phase2-object-workers = 1')) + assert runs[-1]['verified_at'] == runs[-2]['verified_at'], 'failure refreshed validation time' + execute('policy-changed', 0, True, config_change=lambda s:s.replace('name = true', 'name = false')) + execute('clock-rollback', 0, True, stamp='2026-05-31T00:00:00Z') + execute('expired', 0, True, stamp='2027-12-01T01:00:00Z') + (repository/'child/valid.roa').unlink(); execute('missing-object', 1, True) + (repository/'child/valid.roa').write_bytes(good_roa) + good_crl = (repository/'child/child.crl').read_bytes(); (repository/'child/child.crl').write_bytes(b'bad crl') + execute('bad-crl', 1, True); (repository/'child/child.crl').write_bytes(good_crl) + (repository/'root/root.mft').write_bytes(b'bad parent'); execute('corrupt-parent', 1, True, no_child_fetch=True) + reset('baseline-v2'); execute('recovered', 2) + manifest('child', 3, ['child.crl'], child.subject, child_key, child_uri+'child.cer') + execute('fresh-empty-child', 0) + (repository/'child/child.mft').write_bytes(b'bad empty child'); execute('cached-empty-child', 0, True) + manifest('root', 4, ['root.crl'], root.subject, root_key, root_uri+'ta.cer') + execute('fresh-deleted-child', 0) + (repository/'root/root.mft').write_bytes(b'bad deleted parent'); execute('cached-deleted-child', 0, True, no_child_fetch=True) + # Restore a valid child, then revoke its certificate in a signed root CRL. + shutil.copy2(fixture/'cases/baseline-v2/repository/child/child.mft', repository/'child/child.mft') + manifest('child', 4, ['child.crl', 'valid.roa', 'valid2.roa', 'valid.asa'], child.subject, child_key, child_uri+'child.cer') + manifest('root', 5, ['root.crl', 'child.cer'], root.subject, root_key, root_uri+'ta.cer') + execute('restored-child', 2) + revoked = x509.RevokedCertificateBuilder().serial_number(child.serial_number).revocation_date(g.VALID_FROM+timedelta(days=32)).build() + crl = (x509.CertificateRevocationListBuilder().issuer_name(root.subject).last_update(g.VALID_FROM+timedelta(days=33)) + .next_update(g.VALID_TO-timedelta(days=30)).add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(root_key.public_key()), False) + .add_extension(x509.CRLNumber(2), False).add_revoked_certificate(revoked).sign(root_key, hashes.SHA256())) + (repository/'root/root.crl').write_bytes(crl.public_bytes(Encoding.DER)) + manifest('root', 6, ['root.crl', 'child.cer'], root.subject, root_key, root_uri+'ta.cer'); execute('fresh-revoked-child', 0) + (repository/'root/root.mft').write_bytes(b'bad revoked parent'); execute('cached-revoked-child', 0, True, no_child_fetch=True) + # A genuine root -> child -> grandchild chain, with payload only at depth 2. + (repository/'root/root.crl').write_bytes((original/'root/root.crl').read_bytes()) + grand_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + grand_name = x509.Name([g.common_name('Grandchild CA')]) + grand = g.sign_cert(grand_name, child.subject, grand_key.public_key(), child_key, 909, True, grand_key, child_key, + ip_ext=g.ip_resources([g.CHILD_PREFIX]), as_ext=g.as_resources([g.TEST_ASN]), + ca_repository=grand_uri, manifest_uri=grand_uri+'grand.mft', notify_uri=notify, + crl_uri=child_uri+'child.crl', ca_issuer_uri=root_uri+'child.cer') + (repository/'child/grand.cer').write_bytes(grand.public_bytes(Encoding.DER)); (repository/'grand').mkdir() + (repository/'grand/grand.crl').write_bytes(g.make_crl(grand_name, grand_key, 1).public_bytes(Encoding.DER)) + roa_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + roa_ee = g.sign_cert(x509.Name([g.common_name('Grandchild ROA')]), grand_name, roa_key.public_key(), grand_key, + x509.random_serial_number(), False, roa_key, grand_key, + ip_ext=g.ip_resources([g.CHILD_PREFIX]), signed_object_uri=grand_uri+'valid.roa', + crl_uri=grand_uri+'grand.crl', ca_issuer_uri=child_uri+'grand.cer', + not_before=g.VALID_FROM+timedelta(days=31), not_after=g.VALID_TO-timedelta(days=31)) + g.write_cms(g.roa_content(g.CHILD_PREFIX, 24), g.OID_ROA_ECONTENT, roa_ee, roa_key, repository/'grand/valid.roa', work) + manifest('grand', 2, ['grand.crl', 'valid.roa'], grand_name, grand_key, child_uri+'grand.cer') + manifest('child', 5, ['child.crl', 'grand.cer'], child.subject, child_key, child_uri+'child.cer') + manifest('root', 7, ['root.crl', 'child.cer'], root.subject, root_key, root_uri+'ta.cer') + execute('three-level-fresh', 1) + (repository/'root/root.mft').write_bytes(b'bad three-level parent') + execute('three-level-cache', 1, True, no_child_fetch=True) + execute('three-level-cache-ccr-off', 1, True, config_change=lambda s:'\n'.join(l for l in s.splitlines() if not l.startswith('ccr-out ='))+'\n', no_child_fetch=True) + execute('three-level-depth-limit', 0, True, config_change=lambda s:s.replace('require-complete = true', 'require-complete = true\nmax-ca-depth = 1'), no_child_fetch=True) + # A later-expiring sibling stays usable when the first leaf expires. + sibling_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + sibling_name = x509.Name([g.common_name('Sibling CA')]) + sibling = g.sign_cert(sibling_name, child.subject, sibling_key.public_key(), child_key, 910, True, sibling_key, child_key, + ip_ext=g.ip_resources([g.SECOND_PREFIX]), as_ext=g.as_resources([g.TEST_ASN]), + ca_repository=sibling_uri, manifest_uri=sibling_uri+'sibling.mft', notify_uri=notify, + crl_uri=child_uri+'child.crl', ca_issuer_uri=root_uri+'child.cer') + (repository/'child/sibling.cer').write_bytes(sibling.public_bytes(Encoding.DER)); (repository/'sibling').mkdir() + (repository/'sibling/sibling.crl').write_bytes(g.make_crl(sibling_name, sibling_key, 1).public_bytes(Encoding.DER)) + ee_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + ee = g.sign_cert(x509.Name([g.common_name('Sibling ROA')]), sibling_name, ee_key.public_key(), sibling_key, + x509.random_serial_number(), False, ee_key, sibling_key, ip_ext=g.ip_resources([g.SECOND_PREFIX]), + signed_object_uri=sibling_uri+'valid.roa', crl_uri=sibling_uri+'sibling.crl', ca_issuer_uri=child_uri+'sibling.cer', + not_before=g.VALID_FROM+timedelta(days=31), not_after=g.VALID_TO-timedelta(days=10)) + g.write_cms(g.roa_content(g.SECOND_PREFIX, 25), g.OID_ROA_ECONTENT, ee, ee_key, repository/'sibling/valid.roa', work) + manifest('sibling', 1, ['sibling.crl', 'valid.roa'], sibling_name, sibling_key, child_uri+'sibling.cer') + manifest('child', 6, ['child.crl', 'grand.cer', 'sibling.cer'], child.subject, child_key, child_uri+'child.cer') + manifest('root', 8, ['root.crl', 'child.cer'], root.subject, root_key, root_uri+'ta.cer') + execute('siblings-fresh', 2) + (repository/'root/root.mft').write_bytes(b'bad sibling parent') + execute('siblings-cache', 2, True, no_child_fetch=True) + execute('one-cached-leaf-expired', 1, True, stamp='2027-12-01T00:00:00Z', no_child_fetch=True) + proof['transports'].append({'transport': transport, 'runs': runs}) + finally: + server.shutdown(); server.server_close(); thread.join(timeout=5) + if evidence: evidence.write_text(json.dumps(proof, indent=2)+'\n') + return proof + +if __name__ == '__main__': + parser = argparse.ArgumentParser(); parser.add_argument('binary'); parser.add_argument('--evidence', type=Path); args = parser.parse_args() + print(json.dumps(run(args.binary, args.evidence)))