Files
rpki/src/validation/tree.rs

695 lines
25 KiB
Rust

use crate::audit::{DiscoveredFrom, ObjectAuditEntry, PublicationPointAudit};
use crate::cir::{CirInputAccumulator, CirInputSnapshot};
use crate::data_model::rc::{AsResourceSet, IpResourceSet};
use crate::report::Warning;
use crate::validation::manifest::PublicationPointSource;
use crate::validation::objects::{
AspaAttestation, ObjectsOutput, RoaValidationCacheStats, RouterKeyPayload, Vrp,
};
use crate::validation::publication_point::PublicationPointSnapshot;
use std::borrow::Cow;
use std::sync::{Arc, OnceLock};
pub const DEFAULT_MAX_CA_DEPTH: usize = 32;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TreeRunConfig {
/// Max CA instance depth to process (0 = root only).
pub max_depth: Option<usize>,
/// Max number of CA instances to process.
pub max_instances: Option<usize>,
/// Drop per-object audit payload when the caller only needs validation outputs.
pub compact_audit: bool,
/// Persist per-CA VCIR state for future reuse/delta runs.
pub persist_vcir: bool,
/// Build online CCR manifest projections during phase2 validation.
pub build_ccr_accumulator: bool,
/// Reuse accepted ROA validation outputs from previous VCIR when explicitly enabled.
pub enable_roa_validation_cache: bool,
/// Reuse validated child certificate discovery results when explicitly enabled.
pub enable_child_certificate_validation_cache: bool,
/// Evaluate publication-point cache eligibility without changing validation results.
pub publication_point_cache_observe_only: bool,
/// Reuse complete publication-point validation projections when explicitly enabled.
pub enable_publication_point_validation_cache: bool,
/// Prefetch transport requests from the previous run before normal tree traversal.
pub enable_transport_request_prefetch: bool,
}
impl Default for TreeRunConfig {
fn default() -> Self {
Self {
max_depth: Some(DEFAULT_MAX_CA_DEPTH),
max_instances: None,
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
enable_child_certificate_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
}
}
}
pub(crate) fn ca_depth_is_allowed(config: &TreeRunConfig, depth: usize) -> bool {
config.max_depth.is_none_or(|max_depth| depth <= max_depth)
}
pub(crate) fn next_allowed_ca_depth(config: &TreeRunConfig, parent_depth: usize) -> Option<usize> {
let child_depth = parent_depth.checked_add(1)?;
ca_depth_is_allowed(config, child_depth).then_some(child_depth)
}
#[derive(Clone, Debug)]
pub enum CaCertificateRef {
InlineDer(Vec<u8>),
RepoBytes {
sha256_hex: String,
cached_der: Arc<OnceLock<Arc<[u8]>>>,
},
}
impl PartialEq for CaCertificateRef {
fn eq(&self, other: &Self) -> bool {
match (self, other) {
(Self::InlineDer(left), Self::InlineDer(right)) => left == right,
(
Self::RepoBytes {
sha256_hex: left, ..
},
Self::RepoBytes {
sha256_hex: right, ..
},
) => left == right,
_ => false,
}
}
}
impl Eq for CaCertificateRef {}
impl CaCertificateRef {
pub fn inline_der(bytes: Vec<u8>) -> Self {
Self::InlineDer(bytes)
}
pub fn repo_bytes(sha256_hex: String) -> Self {
Self::RepoBytes {
sha256_hex,
cached_der: Arc::new(OnceLock::new()),
}
}
pub fn sha256_hex(&self) -> Option<&str> {
match self {
Self::InlineDer(_) => None,
Self::RepoBytes { sha256_hex, .. } => Some(sha256_hex.as_str()),
}
}
pub fn der<'a>(&'a self, store: &crate::storage::RocksStore) -> Result<Cow<'a, [u8]>, String> {
match self {
Self::InlineDer(bytes) => Ok(Cow::Borrowed(bytes.as_slice())),
Self::RepoBytes {
sha256_hex,
cached_der,
} => {
if cached_der.get().is_none() {
let bytes = store
.get_blob_bytes(sha256_hex)
.map_err(|e| format!("load CA certificate bytes failed: {e}"))?
.ok_or_else(|| {
format!("missing CA certificate repo bytes for sha256={sha256_hex}")
})?;
let _ = cached_der.set(Arc::from(bytes));
}
let bytes = cached_der.get().ok_or_else(|| {
format!("missing cached CA certificate bytes for sha256={sha256_hex}")
})?;
Ok(Cow::Borrowed(bytes.as_ref()))
}
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CaInstanceHandle {
pub depth: usize,
pub tal_id: String,
pub parent_manifest_rsync_uri: Option<String>,
/// CA certificate bytes or a lazy repo-bytes reference for this CA instance.
pub ca_certificate: CaCertificateRef,
/// rsync URI of this CA certificate object (where it is published).
///
/// This is used for strict AIA binding checks (RFC 6487 §4.8.7) when validating
/// subordinate CA and EE certificates. Trust anchor certificates downloaded from
/// TAL URIs may not have an rsync publication URI, so this can be `None`.
pub ca_certificate_rsync_uri: Option<String>,
/// Effective (fully resolved) resources for this CA instance, used for resource-path
/// validation of subordinate CA certificates.
pub effective_ip_resources: Option<IpResourceSet>,
pub effective_as_resources: Option<AsResourceSet>,
pub rsync_base_uri: String,
pub manifest_rsync_uri: String,
pub publication_point_rsync_uri: String,
pub rrdp_notification_uri: Option<String>,
}
impl CaInstanceHandle {
pub fn with_depth(mut self, depth: usize) -> Self {
self.depth = depth;
self
}
pub fn ca_certificate_der<'a>(
&'a self,
store: &crate::storage::RocksStore,
) -> Result<Cow<'a, [u8]>, String> {
self.ca_certificate.der(store)
}
pub fn ca_certificate_sha256_hex(&self) -> Option<&str> {
self.ca_certificate.sha256_hex()
}
pub fn ca_certificate_sha256_32(&self) -> Option<[u8; 32]> {
match &self.ca_certificate {
CaCertificateRef::InlineDer(bytes) => {
use sha2::Digest as _;
let digest = sha2::Sha256::digest(bytes);
Some(digest.into())
}
CaCertificateRef::RepoBytes { sha256_hex, .. } => {
let mut out = [0u8; 32];
hex::decode_to_slice(sha256_hex, &mut out).ok()?;
Some(out)
}
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PublicationPointRunResult {
pub source: PublicationPointSource,
pub snapshot: Option<PublicationPointSnapshot>,
pub warnings: Vec<Warning>,
pub objects: ObjectsOutput,
pub audit: PublicationPointAudit,
pub cir_fresh_objects: Vec<ObjectAuditEntry>,
pub cir_cached_objects: Vec<ObjectAuditEntry>,
/// Candidate child CA instances to enqueue after this publication point completes.
///
/// - For `Fresh`, these are discovered from the current validated publication point.
/// - For `VcirCurrentInstance`, these are restored from the current instance VCIR and then
/// processed fresh-first when their own turn arrives.
pub discovered_children: Vec<DiscoveredChildCaInstance>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DiscoveredChildEntryProjection {
pub child_ski: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DiscoveredChildCaInstance {
pub handle: CaInstanceHandle,
pub discovered_from: DiscoveredFrom,
pub child_entry_projection: Option<DiscoveredChildEntryProjection>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TreeRunOutput {
pub instances_processed: usize,
pub instances_failed: usize,
pub warnings: Vec<Warning>,
pub vrps: Vec<Vrp>,
pub aspas: Vec<AspaAttestation>,
pub router_keys: Vec<RouterKeyPayload>,
}
#[derive(Debug, thiserror::Error)]
pub enum TreeRunError {
#[error("publication point runner failed: {0}")]
Runner(String),
}
pub trait PublicationPointRunner {
fn run_publication_point(
&self,
ca: &CaInstanceHandle,
) -> Result<PublicationPointRunResult, String>;
fn prefetch_discovered_children(
&self,
_children: &[DiscoveredChildCaInstance],
) -> Result<(), String> {
Ok(())
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TreeRunAuditOutput {
pub tree: TreeRunOutput,
pub publication_points: Vec<PublicationPointAudit>,
pub roa_cache_stats: RoaValidationCacheStats,
pub cir_input: CirInputSnapshot,
}
pub(crate) fn submit_publication_point_cir_input(
cir_input: &mut CirInputAccumulator,
source: PublicationPointSource,
audit: &PublicationPointAudit,
cir_fresh_objects: &[ObjectAuditEntry],
cir_cached_objects: &[ObjectAuditEntry],
) -> Result<(), TreeRunError> {
match source {
PublicationPointSource::Fresh => {
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, &audit.objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
PublicationPointSource::FailedFetchNoCache => {
let entries = if cir_fresh_objects.is_empty() {
&audit.objects
} else {
cir_fresh_objects
};
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, entries)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
PublicationPointSource::VcirCurrentInstance
| PublicationPointSource::PublicationPointCache => {
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Fresh, cir_fresh_objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
cir_input
.submit_audit_entries(crate::cir::CirInputSection::Cached, cir_cached_objects)
.map_err(|e| TreeRunError::Runner(e.to_string()))?;
}
}
Ok(())
}
pub fn run_tree_serial(
root: CaInstanceHandle,
runner: &dyn PublicationPointRunner,
config: &TreeRunConfig,
) -> Result<TreeRunOutput, TreeRunError> {
Ok(run_tree_serial_audit(root, runner, config)?.tree)
}
pub fn run_tree_serial_audit(
root: CaInstanceHandle,
runner: &dyn PublicationPointRunner,
config: &TreeRunConfig,
) -> Result<TreeRunAuditOutput, TreeRunError> {
run_tree_serial_audit_multi_root(vec![root], runner, config)
}
pub fn run_tree_serial_audit_multi_root(
roots: Vec<CaInstanceHandle>,
runner: &dyn PublicationPointRunner,
config: &TreeRunConfig,
) -> Result<TreeRunAuditOutput, TreeRunError> {
#[derive(Clone, Debug)]
struct QueuedCaInstance {
id: u64,
handle: CaInstanceHandle,
parent_id: Option<u64>,
discovered_from: Option<DiscoveredFrom>,
}
let mut next_id: u64 = 0;
let mut queue: std::collections::VecDeque<QueuedCaInstance> = std::collections::VecDeque::new();
for root in roots {
queue.push_back(QueuedCaInstance {
id: next_id,
handle: root,
parent_id: None,
discovered_from: None,
});
next_id += 1;
}
let mut visited_manifest_uris: std::collections::HashSet<String> =
std::collections::HashSet::new();
let mut instances_processed = 0usize;
let mut instances_failed = 0usize;
let mut warnings: Vec<Warning> = Vec::new();
let mut vrps: Vec<Vrp> = Vec::new();
let mut aspas: Vec<AspaAttestation> = Vec::new();
let mut router_keys: Vec<RouterKeyPayload> = Vec::new();
let mut publication_points: Vec<PublicationPointAudit> = Vec::new();
let mut roa_cache_stats = RoaValidationCacheStats::default();
let mut cir_input = CirInputAccumulator::default();
while let Some(node) = queue.pop_front() {
let ca = &node.handle;
if !visited_manifest_uris.insert(ca.manifest_rsync_uri.clone()) {
continue;
}
if let Some(max_instances) = config.max_instances {
if instances_processed >= max_instances {
break;
}
}
if !ca_depth_is_allowed(config, ca.depth) {
continue;
}
let res = match runner.run_publication_point(ca) {
Ok(v) => v,
Err(e) => {
instances_failed += 1;
warnings.push(
Warning::new(format!("publication point failed: {e}"))
.with_context(&ca.manifest_rsync_uri),
);
continue;
}
};
instances_processed += 1;
warnings.extend(res.warnings);
warnings.extend(res.objects.warnings);
roa_cache_stats.add_assign(&res.objects.roa_cache_stats);
vrps.extend(res.objects.vrps);
aspas.extend(res.objects.aspas);
router_keys.extend(res.objects.router_keys);
let source = res.source;
let cir_fresh_objects = res.cir_fresh_objects;
let cir_cached_objects = res.cir_cached_objects;
let mut audit = res.audit;
audit.node_id = Some(node.id);
audit.parent_node_id = node.parent_id;
audit.discovered_from = node.discovered_from.clone();
submit_publication_point_cir_input(
&mut cir_input,
source,
&audit,
&cir_fresh_objects,
&cir_cached_objects,
)?;
if config.compact_audit {
audit.objects.clear();
audit.warnings.clear();
}
publication_points.push(audit);
let Some(child_depth) = next_allowed_ca_depth(config, ca.depth) else {
continue;
};
let mut children = res.discovered_children;
children.sort_by(|a, b| {
a.handle
.manifest_rsync_uri
.cmp(&b.handle.manifest_rsync_uri)
.then_with(|| {
a.discovered_from
.child_ca_certificate_rsync_uri
.cmp(&b.discovered_from.child_ca_certificate_rsync_uri)
})
});
if let Err(e) = runner.prefetch_discovered_children(&children) {
warnings.push(
Warning::new(format!("prefetch discovered children failed: {e}"))
.with_context(&ca.manifest_rsync_uri),
);
}
for child in children {
queue.push_back(QueuedCaInstance {
id: next_id,
handle: child.handle.with_depth(child_depth),
parent_id: Some(node.id),
discovered_from: Some(child.discovered_from),
});
next_id += 1;
}
}
Ok(TreeRunAuditOutput {
tree: TreeRunOutput {
instances_processed,
instances_failed,
warnings,
vrps,
aspas,
router_keys,
},
publication_points,
roa_cache_stats,
cir_input: cir_input.finalize(),
})
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use crate::audit::{
AuditObjectKind, AuditObjectResult, DiscoveredFrom, ObjectAuditEntry, PublicationPointAudit,
};
use crate::validation::objects::{ObjectsOutput, ObjectsStats};
use super::{
CaCertificateRef, CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult,
PublicationPointRunner, TreeRunConfig, run_tree_serial_audit,
run_tree_serial_audit_multi_root,
};
fn sample_handle(manifest: &str) -> CaInstanceHandle {
CaInstanceHandle {
depth: 0,
tal_id: "arin".to_string(),
parent_manifest_rsync_uri: None,
ca_certificate: CaCertificateRef::inline_der(vec![1, 2, 3]),
ca_certificate_rsync_uri: None,
effective_ip_resources: None,
effective_as_resources: None,
rsync_base_uri: "rsync://example.test/repo/".to_string(),
manifest_rsync_uri: manifest.to_string(),
publication_point_rsync_uri: "rsync://example.test/repo/".to_string(),
rrdp_notification_uri: None,
}
}
struct PrefetchRecordingRunner {
seen_prefetch_children: Arc<Mutex<Vec<String>>>,
}
impl PublicationPointRunner for PrefetchRecordingRunner {
fn prefetch_discovered_children(
&self,
children: &[DiscoveredChildCaInstance],
) -> Result<(), String> {
let mut seen = self
.seen_prefetch_children
.lock()
.expect("prefetch lock poisoned");
seen.extend(
children
.iter()
.map(|child| child.handle.manifest_rsync_uri.clone()),
);
Ok(())
}
fn run_publication_point(
&self,
ca: &CaInstanceHandle,
) -> Result<PublicationPointRunResult, String> {
let children = if ca.manifest_rsync_uri.ends_with("root.mft") {
vec![DiscoveredChildCaInstance {
handle: sample_handle("rsync://example.test/repo/child.mft"),
discovered_from: DiscoveredFrom {
parent_manifest_rsync_uri: ca.manifest_rsync_uri.clone(),
child_ca_certificate_rsync_uri: "rsync://example.test/repo/child.cer"
.to_string(),
child_ca_certificate_sha256_hex: "00".repeat(32),
},
child_entry_projection: None,
}]
} else {
Vec::new()
};
Ok(PublicationPointRunResult {
source: crate::validation::manifest::PublicationPointSource::Fresh,
snapshot: None,
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats: crate::validation::objects::RoaValidationCacheStats::default(),
roa_cache_object_meta: Vec::new(),
},
audit: PublicationPointAudit::default(),
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: children,
})
}
}
#[test]
fn run_tree_serial_audit_invokes_prefetch_hook_for_discovered_children() {
let seen = Arc::new(Mutex::new(Vec::new()));
let runner = PrefetchRecordingRunner {
seen_prefetch_children: Arc::clone(&seen),
};
let out = run_tree_serial_audit(
sample_handle("rsync://example.test/repo/root.mft"),
&runner,
&TreeRunConfig::default(),
)
.expect("tree run");
assert_eq!(out.tree.instances_processed, 2);
assert_eq!(
seen.lock().expect("lock").as_slice(),
["rsync://example.test/repo/child.mft"]
);
}
#[test]
fn run_tree_serial_audit_multi_root_processes_all_roots() {
let seen = Arc::new(Mutex::new(Vec::new()));
let runner = PrefetchRecordingRunner {
seen_prefetch_children: Arc::clone(&seen),
};
let out = run_tree_serial_audit_multi_root(
vec![
sample_handle("rsync://example.test/repo/root-a.mft"),
sample_handle("rsync://example.test/repo/root-b.mft"),
],
&runner,
&TreeRunConfig::default(),
)
.expect("tree run");
assert_eq!(out.tree.instances_processed, 2);
assert_eq!(out.publication_points.len(), 2);
}
#[test]
fn run_tree_serial_audit_accumulates_fresh_and_cached_cir_input() {
struct CirRunner;
impl PublicationPointRunner for CirRunner {
fn run_publication_point(
&self,
ca: &CaInstanceHandle,
) -> Result<PublicationPointRunResult, String> {
if ca.manifest_rsync_uri.ends_with("cached.mft") {
let fresh_reject = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/cached/current.mft".to_string(),
sha256_hex: "11".repeat(32),
kind: AuditObjectKind::Manifest,
result: AuditObjectResult::Error,
detail: Some("fresh manifest rejected".to_string()),
};
let cached = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/cached/old.roa".to_string(),
sha256_hex: "22".repeat(32),
kind: AuditObjectKind::Roa,
result: AuditObjectResult::Ok,
detail: None,
};
return Ok(PublicationPointRunResult {
source:
crate::validation::manifest::PublicationPointSource::VcirCurrentInstance,
snapshot: None,
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats:
crate::validation::objects::RoaValidationCacheStats::default(),
roa_cache_object_meta: Vec::new(),
},
audit: PublicationPointAudit {
objects: vec![fresh_reject.clone(), cached.clone()],
..PublicationPointAudit::default()
},
cir_fresh_objects: vec![fresh_reject],
cir_cached_objects: vec![cached],
discovered_children: Vec::new(),
});
}
let fresh = ObjectAuditEntry {
rsync_uri: "rsync://example.test/repo/fresh/a.roa".to_string(),
sha256_hex: "33".repeat(32),
kind: AuditObjectKind::Roa,
result: AuditObjectResult::Ok,
detail: None,
};
Ok(PublicationPointRunResult {
source: crate::validation::manifest::PublicationPointSource::Fresh,
snapshot: None,
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats:
crate::validation::objects::RoaValidationCacheStats::default(),
roa_cache_object_meta: Vec::new(),
},
audit: PublicationPointAudit {
objects: vec![fresh],
..PublicationPointAudit::default()
},
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: Vec::new(),
})
}
}
let out = run_tree_serial_audit_multi_root(
vec![
sample_handle("rsync://example.test/repo/fresh/root.mft"),
sample_handle("rsync://example.test/repo/cached/cached.mft"),
],
&CirRunner,
&TreeRunConfig::default(),
)
.expect("tree run");
assert_eq!(out.cir_input.fresh_validated_objects.len(), 2);
assert_eq!(out.cir_input.cached_validated_objects.len(), 1);
assert_eq!(out.cir_input.fresh_rejected_objects.len(), 1);
assert!(
out.cir_input
.fresh_validated_objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/fresh/a.roa")
);
assert!(
out.cir_input
.cached_validated_objects
.iter()
.any(|item| item.rsync_uri == "rsync://example.test/repo/cached/old.roa")
);
}
}