Files
panda-rpki-oss/src/validation/tree_runner/fresh_pipeline.rs
yuyr 5f46837b61
Some checks failed
ci / rust (push) Failing after 17s
ci / docker (push) Failing after 1m12s
ci / audit (push) Failing after 8s
20261015_4 修复Manifest编号与改名抗重放校验
2026-10-16 11:16:20 +08:00

393 lines
15 KiB
Rust

impl<'a> Rpkiv1PublicationPointRunner<'a> {
pub(crate) fn ccr_accumulator_snapshot(&self) -> Option<CcrAccumulator> {
self.ccr_accumulator
.as_ref()
.and_then(|accumulator| accumulator.lock().ok().map(|guard| guard.clone()))
}
pub(crate) fn append_ccr_manifest_projection(
&self,
projection: &CcrManifestProjection,
) -> Result<(), String> {
if let Some(accumulator) = self.ccr_accumulator.as_ref() {
accumulator
.lock()
.map_err(|_| "lock CCR accumulator failed".to_string())?
.append_manifest_projection(projection)?;
}
Ok(())
}
pub(crate) fn stage_fresh_publication_point_after_repo_ready(
&self,
ca: &CaInstanceHandle,
repo_sync_ok: bool,
repo_sync_err: Option<&str>,
) -> Result<FreshPublicationPointStage, FreshPublicationPointStageError> {
let snapshot_prepare_started = std::time::Instant::now();
let issuer_ca_der = ca_certificate_der_for_validation(ca, self.store, self.timing.as_ref())
.map_err(|detail| FreshPublicationPointStageError {
warnings: Vec::new(),
error: ManifestFreshError::IssuerCaLoadFailed { detail },
snapshot_prepare_ms: snapshot_prepare_started.elapsed().as_millis() as u64,
})?;
let issuer_ca_der: Arc<[u8]> = Arc::from(issuer_ca_der.as_ref());
let identity = ResourceCertificate::decode_der(&issuer_ca_der)
.map(|cert| {
crate::repository::storage::CaIdentity::from_spki(
ca.anchor_namespace,
&cert.tbs.subject_public_key_info,
)
})
.map_err(|e| FreshPublicationPointStageError {
warnings: Vec::new(),
error: ManifestFreshError::IssuerCaLoadFailed {
detail: e.to_string(),
},
snapshot_prepare_ms: snapshot_prepare_started.elapsed().as_millis() as u64,
})?;
let rename_warnings: Vec<Warning> = crate::validation::manifest::manifest_rename_warning(
self.store,
&identity,
&ca.manifest_sia_uris,
&ca.manifest_rsync_uri,
)
.map_err(|error| FreshPublicationPointStageError {
warnings: Vec::new(),
error,
snapshot_prepare_ms: snapshot_prepare_started.elapsed().as_millis() as u64,
})?
.into_iter()
.collect();
for warning in &rename_warnings {
crate::logging::emit(
crate::logging::Level::Warn,
"manifest_filename_changed",
|| serde_json::json!({"manifest_rsync_uri": ca.manifest_rsync_uri, "message": warning.message, "candidate_accepted": null}),
);
}
let fresh_publication_point = {
let _manifest_total = self
.timing
.as_ref()
.map(|t| t.span_phase("manifest_processing_total"));
process_manifest_publication_point_fresh_after_repo_sync_with_timing(
self.store,
&ca.anchor_namespace,
&ca.manifest_rsync_uri,
&ca.publication_point_rsync_uri,
self.current_repo_index.as_ref(),
issuer_ca_der.as_ref(),
ca.ca_certificate_rsync_uri.as_deref(),
self.validation_time,
repo_sync_ok,
repo_sync_err,
)
};
let snapshot_prepare_ms = snapshot_prepare_started.elapsed().as_millis() as u64;
let (fresh_point, snapshot_prepare_timing) =
fresh_publication_point.map_err(|error| FreshPublicationPointStageError {
warnings: rename_warnings
.iter()
.cloned()
.map(|mut warning| {
warning
.message
.push_str(&format!(" candidate_accepted=false reason={error}"));
warning
})
.collect(),
error,
snapshot_prepare_ms,
})?;
if let Some(timing) = self.timing.as_ref() {
timing.record_phase_nanos(
"fresh_snapshot_prepare_total",
snapshot_prepare_ms.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_manifest_load_total",
snapshot_prepare_timing
.manifest_load_ms
.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_manifest_decode_total",
snapshot_prepare_timing
.manifest_decode_ms
.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_anti_rollback_total",
snapshot_prepare_timing
.anti_rollback_guard_ms
.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_manifest_entries_total",
snapshot_prepare_timing
.manifest_entries_ms
.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_pack_files_total",
snapshot_prepare_timing
.pack_files_ms
.saturating_mul(1_000_000),
);
timing.record_phase_nanos(
"fresh_snapshot_ee_path_validate_total",
snapshot_prepare_timing
.ee_path_validate_ms
.saturating_mul(1_000_000),
);
timing.record_count("fresh_publication_points", 1);
timing.record_count(
"fresh_manifest_files_total",
snapshot_prepare_timing.manifest_file_count as u64,
);
}
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_prepare",
snapshot_prepare_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_manifest_load",
snapshot_prepare_timing.manifest_load_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_manifest_decode",
snapshot_prepare_timing.manifest_decode_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_anti_rollback",
snapshot_prepare_timing.anti_rollback_guard_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_manifest_entries",
snapshot_prepare_timing.manifest_entries_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_pack_files",
snapshot_prepare_timing.pack_files_ms,
);
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_ee_path_validate",
snapshot_prepare_timing.ee_path_validate_ms,
);
let child_discovery_started = std::time::Instant::now();
let out = {
let _child_disc_total = self
.timing
.as_ref()
.map(|t| t.span_phase("child_discovery_total"));
discover_children_from_fresh_snapshot_with_audit_with_issuer_der(
ca,
issuer_ca_der.as_ref(),
&fresh_point,
self.validation_time,
self.timing.as_ref(),
self.policy,
)
};
let (discovered_children, child_audits, discovered_router_keys, mut warnings) = match out {
Ok(out) => (out.children, out.audits, out.router_keys, Vec::new()),
Err(e) => (
Vec::new(),
Vec::new(),
Vec::new(),
vec![
Warning::new(format!("child CA discovery failed: {e}"))
.with_category(crate::output::report::WarningCategory::Certificate)
.with_rfc_refs(&[RfcRef("RFC 6487 §7.2")])
.with_context(&ca.manifest_rsync_uri),
],
),
};
let child_discovery_ms = child_discovery_started.elapsed().as_millis() as u64;
if let Some(timing) = self.timing.as_ref() {
timing.record_phase_nanos(
"fresh_child_discovery_total",
child_discovery_ms.saturating_mul(1_000_000),
);
timing.record_count(
"fresh_children_discovered",
discovered_children.len() as u64,
);
timing.record_count("fresh_child_audits", child_audits.len() as u64);
timing.record_count(
"fresh_router_keys_discovered",
discovered_router_keys.len() as u64,
);
}
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_child_discovery",
child_discovery_ms,
);
warnings.extend(rename_warnings);
Ok(FreshPublicationPointStage {
fresh_point,
issuer_ca_der,
snapshot_prepare_timing,
snapshot_prepare_ms,
discovered_children,
child_audits,
discovered_router_keys,
child_discovery_ms,
warnings,
})
}
pub(crate) fn finalize_fresh_publication_point_from_reducer(
&self,
ca: &CaInstanceHandle,
fresh_point: &FreshValidatedPublicationPoint,
mut warnings: Vec<Warning>,
objects: crate::validation::objects::ObjectsOutput,
child_audits: Vec<ObjectAuditEntry>,
discovered_children: Vec<DiscoveredChildCaInstance>,
repo_sync_source: Option<&str>,
repo_sync_phase: Option<&str>,
repo_sync_duration_ms: u64,
repo_sync_err: Option<&str>,
) -> Result<FreshPublicationPointFinalizeOutput, String> {
let snapshot_pack_started = std::time::Instant::now();
let pack = fresh_point.to_publication_point_snapshot();
let snapshot_pack_ms = snapshot_pack_started.elapsed().as_millis() as u64;
if let Some(timing) = self.timing.as_ref() {
timing.record_phase_nanos(
"fresh_snapshot_pack_total",
snapshot_pack_ms.saturating_mul(1_000_000),
);
}
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_snapshot_pack",
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::<Result<Vec<_>, 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 audit_build_started = std::time::Instant::now();
let audit = build_publication_point_audit_from_snapshot(
ca,
PublicationPointSource::Fresh,
repo_sync_source,
repo_sync_phase,
Some(repo_sync_duration_ms),
repo_sync_err,
&pack,
&warnings,
&objects,
&child_audits,
);
let audit_build_ms = audit_build_started.elapsed().as_millis() as u64;
if let Some(timing) = self.timing.as_ref() {
timing.record_phase_nanos(
"fresh_audit_build_total",
audit_build_ms.saturating_mul(1_000_000),
);
}
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
"fresh_audit_build",
audit_build_ms,
);
Ok(FreshPublicationPointFinalizeOutput {
result: PublicationPointRunResult {
source: PublicationPointSource::Fresh,
snapshot: Some(pack),
warnings,
objects,
audit,
discovered_children,
},
snapshot_pack_ms,
ccr_projection_build_ms,
ccr_append_ms,
audit_build_ms,
})
}
}