20261016 修复 Manifest failed-fetch 缓存与子树恢复
Some checks failed
ci / rust (push) Failing after 4s
ci / docker (push) Failing after 1m27s
ci / audit (push) Failing after 3s

This commit is contained in:
2026-10-16 15:53:40 +08:00
parent 5f46837b61
commit 2d7501db08
28 changed files with 1565 additions and 143 deletions

View File

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

View File

@@ -4,6 +4,7 @@
| Behaviour | Standard reference | Implementation | Evidence | | 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` | | 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` | | 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` | | 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` |

43
docs/failed-fetch.md Normal file
View File

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

View File

@@ -158,7 +158,12 @@ fn execute(config: &Config, run_id: &str) -> Result<i32, String> {
.iter() .iter()
.filter(|a| a["source"] == "cache" && a["refreshFailed"] == true) .filter(|a| a["source"] == "cache" && a["refreshFailed"] == true)
.count(); .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" "partial"
} else { } else {
"success" "success"

View File

@@ -30,6 +30,7 @@ fn registered_event_level(kind: &str) -> Option<crate::logging::Level> {
| "rrdp_sync_failed" | "rrdp_sync_failed"
| "rrdp_cross_origin_rejected" | "rrdp_cross_origin_rejected"
| "repo_terminal_failure" | "repo_terminal_failure"
| "failed_fetch_cache_reused"
| "rrdp_failed_fallback_rsync" | "rrdp_failed_fallback_rsync"
) { ) {
crate::logging::Level::Warn crate::logging::Level::Warn

View File

@@ -246,7 +246,7 @@ pub struct RoaIpAddress {
pub max_length: Option<u16>, pub max_length: Option<u16>,
} }
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
pub struct IpPrefix { pub struct IpPrefix {
pub afi: RoaAfi, pub afi: RoaAfi,
/// Prefix length in bits. /// Prefix length in bits.

View File

@@ -1,9 +1,9 @@
use serde::Serialize; use serde::{Deserialize, Serialize};
use sha2::Digest; use sha2::Digest;
use crate::validation::policy::Policy; use crate::validation::policy::Policy;
#[derive(Clone, Debug, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum AuditObjectKind { pub enum AuditObjectKind {
Manifest, Manifest,
@@ -15,7 +15,7 @@ pub enum AuditObjectKind {
Other, Other,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum AuditObjectResult { pub enum AuditObjectResult {
Ok, Ok,
@@ -23,7 +23,7 @@ pub enum AuditObjectResult {
Error, Error,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ObjectAuditEntry { pub struct ObjectAuditEntry {
pub rsync_uri: String, pub rsync_uri: String,
pub sha256_hex: String, pub sha256_hex: String,
@@ -33,7 +33,7 @@ pub struct ObjectAuditEntry {
pub detail: Option<String>, pub detail: Option<String>,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AuditWarning { pub struct AuditWarning {
pub message: String, pub message: String,
pub category: 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 { pub struct PublicationPointAudit {
/// Monotonic node ID assigned by the traversal engine. /// Monotonic node ID assigned by the traversal engine.
/// ///
@@ -148,7 +148,7 @@ pub struct PublicationPointAudit {
pub objects: Vec<ObjectAuditEntry>, pub objects: Vec<ObjectAuditEntry>,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscoveredFrom { pub struct DiscoveredFrom {
pub parent_manifest_rsync_uri: String, pub parent_manifest_rsync_uri: String,
pub child_ca_certificate_rsync_uri: String, pub child_ca_certificate_rsync_uri: String,

View File

@@ -1,5 +1,6 @@
use rocksdb::{ColumnFamilyDescriptor, DBCompressionType, Options}; 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_REPOSITORY_VIEW: &str = "repository_view";
pub const CF_RAW_BY_HASH: &str = "raw_by_hash"; pub const CF_RAW_BY_HASH: &str = "raw_by_hash";
pub const CF_RAW_BLOB: &str = "raw_blob"; 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] = &[ pub const ALL_COLUMN_FAMILY_NAMES: &[&str] = &[
CF_REPOSITORY_VIEW, CF_REPOSITORY_VIEW,
CF_VALIDATED_PUBLICATION_POINT,
CF_RAW_BY_HASH, CF_RAW_BY_HASH,
CF_RAW_BLOB, CF_RAW_BLOB,
CF_MANIFEST_ANTI_ROLLBACK, CF_MANIFEST_ANTI_ROLLBACK,

View File

@@ -19,10 +19,20 @@ impl RocksStore {
/// Re-read and validate under the same lock as the write. RocksDB itself /// Re-read and validate under the same lock as the write. RocksDB itself
/// excludes a second writable process opening this database. /// excludes a second writable process opening this database.
#[cfg(test)]
pub(crate) fn update_ca_manifest_history<E: From<StorageError>>( pub(crate) fn update_ca_manifest_history<E: From<StorageError>>(
&self, &self,
identity: &CaIdentity, identity: &CaIdentity,
update: impl FnOnce(Option<&CaManifestHistory>) -> Result<CaManifestHistory, E>, update: impl FnOnce(Option<&CaManifestHistory>) -> Result<CaManifestHistory, E>,
) -> Result<CaManifestHistory, E> {
self.update_ca_manifest_history_and_cache(identity, None, update)
}
pub(crate) fn update_ca_manifest_history_and_cache<E: From<StorageError>>(
&self,
identity: &CaIdentity,
mut cache: Option<crate::validation::failed_fetch::CachedPublicationPoint>,
update: impl FnOnce(Option<&CaManifestHistory>) -> Result<CaManifestHistory, E>,
) -> Result<CaManifestHistory, E> { ) -> Result<CaManifestHistory, E> {
let _guard = let _guard =
self.manifest_history_commit self.manifest_history_commit
@@ -36,9 +46,57 @@ impl RocksStore {
history.validate_internal()?; history.validate_internal()?;
let cf = self.cf(CF_MANIFEST_ANTI_ROLLBACK)?; let cf = self.cf(CF_MANIFEST_ANTI_ROLLBACK)?;
let bytes = encode_cbor(&history, "ca_manifest")?; 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 self.db
.put_cf(cf, ca_manifest_key(identity).as_bytes(), bytes) .write(batch)
.map_err(|e| StorageError::RocksDb(e.to_string()))?; .map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(history) Ok(history)
} }
} }
impl RocksStore {
pub(crate) fn get_validated_publication_point(
&self,
identity: &CaIdentity,
) -> StorageResult<Option<crate::validation::failed_fetch::CachedPublicationPoint>> {
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))
}
}

View File

@@ -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); 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::<StorageError>(&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::<StorageError>(&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());
}

View File

@@ -421,6 +421,7 @@ struct ValidationSummary {
rrdp_delta_updates: usize, rrdp_delta_updates: usize,
vrps: usize, vrps: usize,
publication_points: usize, publication_points: usize,
failed_fetch_cached: usize,
validation_ms: u64, validation_ms: u64,
} }
@@ -466,6 +467,11 @@ fn write_summary(
rrdp_noop_repositories: 0, rrdp_noop_repositories: 0,
rrdp_delta_updates: 0, rrdp_delta_updates: 0,
vrps: shared.vrps.len(), 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(), publication_points: shared.publication_points.len(),
validation_ms, validation_ms,
}; };

View File

@@ -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::write(rollback.join("rrdp/notification.xml"), notification).unwrap();
std::fs::remove_file(&active).unwrap(); std::fs::remove_file(&active).unwrap();
std::os::unix::fs::symlink(&rollback, &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(); let store = RocksStore::open(&db).unwrap();
assert_eq!( assert_eq!(
store.get_ca_manifest_history(&identity).unwrap(), store.get_ca_manifest_history(&identity).unwrap(),

View File

@@ -232,9 +232,14 @@ pub struct TaConstraints {
allow_asn: IntervalIndex<AsInterval>, allow_asn: IntervalIndex<AsInterval>,
deny_asn: IntervalIndex<AsInterval>, deny_asn: IntervalIndex<AsInterval>,
warnings: Vec<String>, warnings: Vec<String>,
cache_fingerprint: String,
} }
impl TaConstraints { impl TaConstraints {
pub(crate) fn cache_fingerprint(&self) -> &str {
&self.cache_fingerprint
}
pub fn from_file(path: &Path) -> Result<Self, String> { pub fn from_file(path: &Path) -> Result<Self, String> {
let contents = std::fs::read_to_string(path) let contents = std::fs::read_to_string(path)
.map_err(|error| format!("read {}: {error}", path.display()))?; .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("allow AS", &mut allow_asn, &mut warnings);
normalize_as_intervals("deny AS", &mut deny_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 { Ok(Self {
source, source,
cache_fingerprint,
allow_v4: IntervalIndex::new(allow_v4), allow_v4: IntervalIndex::new(allow_v4),
deny_v4: IntervalIndex::new(deny_v4), deny_v4: IntervalIndex::new(deny_v4),
allow_v6: IntervalIndex::new(allow_v6), allow_v6: IntervalIndex::new(allow_v6),

View File

@@ -311,3 +311,17 @@ fn generated_very_large_ipv4_constraints_use_interval_tree() {
assert!(!constraints.allow_v4.uses_tree()); assert!(!constraints.allow_v4.uses_tree());
assert!(constraints.warnings.is_empty()); 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());
}

View File

@@ -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<Vrp>,
aspas: Vec<AspaAttestation>,
router_keys: Vec<RouterKeyPayload>,
stats: ObjectsStats,
pub audit: PublicationPointAudit,
pub children: Vec<DiscoveredChildCaInstance>,
pub ccr: CcrManifestProjection,
}
pub(crate) fn identity(store: &RocksStore, ca: &CaInstanceHandle) -> Result<CaIdentity, String> {
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<String, String> {
// 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<OffsetDateTime, String> {
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<OffsetDateTime, String> {
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<Option<Self>, 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<Self, String> {
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;

View File

@@ -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<Vec<u8>, String> {
panic!("unexpected HTTP request")
}
}
impl RsyncFetcher for Fixture {
fn fetch_objects(&self, base: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>> {
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<u8>)>,
) {
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::<CaManifestHistory, crate::repository::storage::StorageError>(
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");
}
}
}

View File

@@ -88,23 +88,33 @@ fn checked_manifest_history(
Ok(next) Ok(next)
} }
#[cfg(test)]
pub(crate) fn commit_manifest_history( pub(crate) fn commit_manifest_history(
store: &RocksStore, store: &RocksStore,
candidate: &ManifestHistoryCandidate, candidate: &ManifestHistoryCandidate,
) -> Result<(CaManifestHistory, Option<Warning>), ManifestFreshError> {
commit_manifest_history_with_cache(store, candidate, None)
}
pub(crate) fn commit_manifest_history_with_cache(
store: &RocksStore,
candidate: &ManifestHistoryCandidate,
cache: Option<crate::validation::failed_fetch::CachedPublicationPoint>,
) -> Result<(CaManifestHistory, Option<Warning>), ManifestFreshError> { ) -> Result<(CaManifestHistory, Option<Warning>), ManifestFreshError> {
let mut warning = None; let mut warning = None;
let history = store.update_ca_manifest_history(&candidate.identity, |old| { let history =
let next = checked_manifest_history(old, candidate)?; store.update_ca_manifest_history_and_cache(&candidate.identity, cache, |old| {
warning = old.and_then(|old| { let next = checked_manifest_history(old, candidate)?;
rename_warning_for_history( warning = old.and_then(|old| {
old, rename_warning_for_history(
&candidate.identity, old,
&candidate.manifest_sia_uris, &candidate.identity,
&candidate.history.manifest_rsync_uri, &candidate.manifest_sia_uris,
) &candidate.history.manifest_rsync_uri,
}); )
Ok::<_, ManifestFreshError>(next) });
})?; Ok::<_, ManifestFreshError>(next)
})?;
Ok((history, warning)) Ok((history, warning))
} }

View File

@@ -4,6 +4,7 @@
pub enum PublicationPointSource { pub enum PublicationPointSource {
Fresh, Fresh,
FailedFetchNoCache, FailedFetchNoCache,
FailedFetchCached,
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]

View File

@@ -1,6 +1,7 @@
pub mod ca_instance; pub mod ca_instance;
pub mod ca_path; pub mod ca_path;
pub mod cert_path; pub mod cert_path;
pub(crate) mod failed_fetch;
pub mod from_tal; pub mod from_tal;
pub mod manifest; pub mod manifest;
pub mod objects; pub mod objects;

View File

@@ -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 struct Vrp {
pub asn: u32, pub asn: u32,
pub prefix: IpPrefix, pub prefix: IpPrefix,
pub max_length: u16, pub max_length: u16,
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AspaAttestation { pub struct AspaAttestation {
pub customer_as_id: u32, pub customer_as_id: u32,
pub provider_as_ids: Vec<u32>, pub provider_as_ids: Vec<u32>,
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RouterKeyPayload { pub struct RouterKeyPayload {
pub as_id: u32, pub as_id: u32,
pub ski: Vec<u8>, pub ski: Vec<u8>,
@@ -89,7 +89,7 @@ pub struct ObjectsOutput {
pub stats: ObjectsStats, pub stats: ObjectsStats,
pub audit: Vec<ObjectAuditEntry>, pub audit: Vec<ObjectAuditEntry>,
} }
#[derive(Clone, Debug, Default, PartialEq, Eq)] #[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ObjectsStats { pub struct ObjectsStats {
pub roa_total: usize, pub roa_total: usize,
pub roa_ok: usize, pub roa_ok: usize,

View File

@@ -4,6 +4,7 @@ use crate::output::report::Warning;
use crate::validation::manifest::PublicationPointSource; use crate::validation::manifest::PublicationPointSource;
use crate::validation::objects::{AspaAttestation, ObjectsOutput, RouterKeyPayload, Vrp}; use crate::validation::objects::{AspaAttestation, ObjectsOutput, RouterKeyPayload, Vrp};
use crate::validation::publication_point::PublicationPointSnapshot; use crate::validation::publication_point::PublicationPointSnapshot;
use serde::{Deserialize, Serialize};
use std::borrow::Cow; use std::borrow::Cow;
use std::sync::{Arc, OnceLock}; 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) ca_depth_is_allowed(config, child_depth).then_some(child_depth)
} }
#[derive(Clone, Debug)] #[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CaCertificateRef { pub enum CaCertificateRef {
InlineDer(Vec<u8>), InlineDer(Vec<u8>),
/// A previously validated child reached through a failed-fetch cache edge.
/// Its entire branch must use cached results without repository requests.
CachedDer(Vec<u8>),
RepoBytes { RepoBytes {
sha256_hex: String, sha256_hex: String,
#[serde(skip)]
cached_der: Arc<OnceLock<Arc<[u8]>>>, cached_der: Arc<OnceLock<Arc<[u8]>>>,
}, },
} }
@@ -53,7 +58,8 @@ pub enum CaCertificateRef {
impl PartialEq for CaCertificateRef { impl PartialEq for CaCertificateRef {
fn eq(&self, other: &Self) -> bool { fn eq(&self, other: &Self) -> bool {
match (self, other) { 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 { Self::RepoBytes {
sha256_hex: left, .. sha256_hex: left, ..
@@ -83,7 +89,7 @@ impl CaCertificateRef {
pub fn sha256_hex(&self) -> Option<&str> { pub fn sha256_hex(&self) -> Option<&str> {
match self { match self {
Self::InlineDer(_) => None, Self::InlineDer(_) | Self::CachedDer(_) => None,
Self::RepoBytes { sha256_hex, .. } => Some(sha256_hex.as_str()), Self::RepoBytes { sha256_hex, .. } => Some(sha256_hex.as_str()),
} }
} }
@@ -93,7 +99,7 @@ impl CaCertificateRef {
store: &crate::repository::storage::RocksStore, store: &crate::repository::storage::RocksStore,
) -> Result<Cow<'a, [u8]>, String> { ) -> Result<Cow<'a, [u8]>, String> {
match self { 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 { Self::RepoBytes {
sha256_hex, sha256_hex,
cached_der, cached_der,
@@ -116,7 +122,7 @@ impl CaCertificateRef {
} }
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CaInstanceHandle { pub struct CaInstanceHandle {
pub anchor_namespace: [u8; 32], pub anchor_namespace: [u8; 32],
pub depth: usize, pub depth: usize,
@@ -142,6 +148,10 @@ pub struct CaInstanceHandle {
} }
impl 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 { pub fn with_depth(mut self, depth: usize) -> Self {
self.depth = depth; self.depth = depth;
self self
@@ -160,7 +170,7 @@ impl CaInstanceHandle {
pub fn ca_certificate_sha256_32(&self) -> Option<[u8; 32]> { pub fn ca_certificate_sha256_32(&self) -> Option<[u8; 32]> {
match &self.ca_certificate { match &self.ca_certificate {
CaCertificateRef::InlineDer(bytes) => { CaCertificateRef::InlineDer(bytes) | CaCertificateRef::CachedDer(bytes) => {
use sha2::Digest as _; use sha2::Digest as _;
let digest = sha2::Sha256::digest(bytes); let digest = sha2::Sha256::digest(bytes);
Some(digest.into()) Some(digest.into())
@@ -184,17 +194,18 @@ pub struct PublicationPointRunResult {
/// Candidate child CA instances to enqueue after this publication point completes. /// Candidate child CA instances to enqueue after this publication point completes.
/// ///
/// - For `Fresh`, these are discovered from the current validated publication point. /// - For `Fresh`, these are discovered from the current validated publication point.
/// - For failed synchronization, the result contains no child instances; /// - For `FailedFetchCached`, these are cache-only edges: no descendant
/// callers may retry the repository on a later run. /// acquisition or fresh validation is permitted during this run.
/// - Without usable cached results, this list is empty.
pub discovered_children: Vec<DiscoveredChildCaInstance>, pub discovered_children: Vec<DiscoveredChildCaInstance>,
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscoveredChildEntryProjection { pub struct DiscoveredChildEntryProjection {
pub child_ski: String, pub child_ski: String,
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscoveredChildCaInstance { pub struct DiscoveredChildCaInstance {
pub handle: CaInstanceHandle, pub handle: CaInstanceHandle,
pub discovered_from: DiscoveredFrom, pub discovered_from: DiscoveredFrom,

View File

@@ -93,6 +93,21 @@ fn start_queued_ca_instances(
continue; continue;
} }
*instances_started += 1; *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) { match repo_runtime.request_publication_point_repo(&node.handle, 0) {
Ok(RepoSyncRequestStatus::Ready { mut outcome, .. }) => { Ok(RepoSyncRequestStatus::Ready { mut outcome, .. }) => {
// Ready here means this CA is reusing repo work that has already completed // 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, ready_queue_len_after_pop,
..ReadyStageMetrics::default() ..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 mut warnings = ready.repo_outcome.warnings.clone();
let repo_outcome = ready.repo_outcome.clone(); let repo_outcome = ready.repo_outcome.clone();
let stage_fresh_started = Instant::now(); let stage_fresh_started = Instant::now();

View File

@@ -276,78 +276,29 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
snapshot_pack_ms, snapshot_pack_ms,
); );
let mut ccr_projection = None; let ccr_projection_build_started = std::time::Instant::now();
let mut ccr_projection_build_ms = 0; let subordinate_skis = discovered_children
let mut ccr_append_ms = 0; .iter()
if self.ccr_accumulator.is_some() { .map(|child| {
let ccr_projection_build_started = std::time::Instant::now(); if let Some(projection) = &child.child_entry_projection {
let subordinate_skis = discovered_children return hex::decode(&projection.child_ski).map_err(|error| error.to_string());
.iter() }
.map(|child| { let der = child.handle.ca_certificate_der(self.store)?;
if let Some(projection) = &child.child_entry_projection { let certificate = ResourceCertificate::decode_der(der.as_ref())
return hex::decode(&projection.child_ski) .map_err(|error| error.to_string())?;
.map_err(|error| error.to_string()); certificate
} .tbs
let der = child.handle.ca_certificate_der(self.store)?; .extensions
let certificate = ResourceCertificate::decode_der(der.as_ref()) .subject_key_identifier
.map_err(|error| error.to_string())?; .clone()
certificate .ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string())
.tbs })
.extensions .collect::<Result<Vec<_>, String>>()?;
.subject_key_identifier let ccr_manifest_projection =
.clone() crate::ccr::projection::from_snapshot(ca, &pack, subordinate_skis)?;
.ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string()) let ccr_projection_build_ms = ccr_projection_build_started.elapsed().as_millis() as u64;
}) let ccr_projection = ccr_manifest_projection;
.collect::<Result<Vec<_>, String>>()?; let ccr_append_ms;
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 audit_build_started = std::time::Instant::now(); let audit_build_started = std::time::Instant::now();
let audit = build_publication_point_audit_from_snapshot( let audit = build_publication_point_audit_from_snapshot(
ca, ca,
@@ -374,6 +325,60 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
audit_build_ms, 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 { Ok(FreshPublicationPointFinalizeOutput {
result: PublicationPointRunResult { result: PublicationPointRunResult {
source: PublicationPointSource::Fresh, source: PublicationPointSource::Fresh,

View File

@@ -24,6 +24,7 @@ pub(super) fn kind_from_rsync_uri(uri: &str) -> AuditObjectKind {
pub(super) fn source_label(source: PublicationPointSource) -> String { pub(super) fn source_label(source: PublicationPointSource) -> String {
match source { match source {
PublicationPointSource::Fresh => "fresh".to_string(), PublicationPointSource::Fresh => "fresh".to_string(),
PublicationPointSource::FailedFetchCached => "failed_fetch_cached".to_string(),
PublicationPointSource::FailedFetchNoCache => "failed_fetch_no_cache".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 { pub(super) fn terminal_state_label(source: PublicationPointSource) -> &'static str {
match source { match source {
PublicationPointSource::Fresh => "fresh", PublicationPointSource::Fresh => "fresh",
PublicationPointSource::FailedFetchCached => "failed_cached",
PublicationPointSource::FailedFetchNoCache => "failed_no_cache", PublicationPointSource::FailedFetchNoCache => "failed_no_cache",
} }
} }

View File

@@ -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 = let attempted_rrdp =
self.policy.sync_preference == crate::validation::policy::SyncPreference::RrdpThenRsync; self.policy.sync_preference == crate::validation::policy::SyncPreference::RrdpThenRsync;
let repo_started = std::time::Instant::now(); let repo_started = std::time::Instant::now();
@@ -97,45 +111,15 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
let stage = match stage { let stage = match stage {
Ok(stage) => stage, Ok(stage) => stage,
Err(stage_error) => { Err(stage_error) => {
let mut warnings = stage_error.warnings; return self.failed_fetch_result(
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(
ca, ca,
stage_error.warnings,
stage_error.error,
repo_sync_source.as_deref(), repo_sync_source.as_deref(),
repo_sync_phase.as_deref(), repo_sync_phase.as_deref(),
Some(repo_sync_duration_ms), repo_sync_duration_ms,
repo_sync_error.as_deref(), 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) Ok(result)
} }
} }
impl Rpkiv1PublicationPointRunner<'_> {
fn failed_fetch_result(
&self,
ca: &CaInstanceHandle,
mut warnings: Vec<Warning>,
error: ManifestFreshError,
repo_source: Option<&str>,
repo_phase: Option<&str>,
duration: u64,
repo_error: Option<&str>,
) -> Result<PublicationPointRunResult, String> {
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)
}
}

22
tests/failed_fetch_e2e.rs Normal file
View File

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

View File

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

View File

@@ -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'<snapshot xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{session}" serial="1">'
+ ''.join(g.publish('rsync://127.0.0.1:873/custom/'+str(p.relative_to(repository)), p) for p in objects)
+ '</snapshot>').encode()
(active/'rrdp/snapshot.xml').write_bytes(snapshot)
(active/'rrdp/notification.xml').write_text(f'<notification xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{session}" serial="1"><snapshot uri="https://127.0.0.1:{port}/rrdp/snapshot.xml" hash="{hashlib.sha256(snapshot).hexdigest()}"/></notification>')
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)))