NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260624 修复future notBefore PP缓存并固化231发布

Browse Source
This commit is contained in:
yuyr
2026-06-24 13:36:39 +08:00
parent adb68fd469
commit af25316d68
6 changed files with 1064 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
src/storage.rs
View File

@@ -80,6 +80,13 @@ enum PublicationPointCacheProjectionIndexState {
},
}
#[derive(Clone, Copy)]
pub(crate) enum PublicationPointCacheProjectionWriteAction<'a> {
Keep,
Write(&'a PublicationPointCacheProjection),
Delete { manifest_rsync_uri: &'a str },
}
fn process_vm_rss_kb() -> Option<u64> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
status.lines().find_map(|line| {
@@ -2533,20 +2540,27 @@ fn write_roa_cache_projection_to_batch(
fn write_publication_point_cache_projection_to_batch(
projection_cf: &ColumnFamily,
batch: &mut WriteBatch,
projection: Option<&PublicationPointCacheProjection>,
action: PublicationPointCacheProjectionWriteAction<'_>,
timing: Option<&mut VcirReplaceTimingBreakdown>,
) -> StorageResult<()> {
let Some(projection) = projection else {
return Ok(());
};
projection.validate_internal()?;
let key = publication_point_cache_projection_key(&projection.manifest_rsync_uri);
let value = encode_cbor(projection, "publication_point_cache_projection")?;
if let Some(timing) = timing {
timing.publication_point_cache_projection_value_bytes = value.len() as u64;
match action {
PublicationPointCacheProjectionWriteAction::Keep => Ok(()),
PublicationPointCacheProjectionWriteAction::Write(projection) => {
projection.validate_internal()?;
let key = publication_point_cache_projection_key(&projection.manifest_rsync_uri);
let value = encode_cbor(projection, "publication_point_cache_projection")?;
if let Some(timing) = timing {
timing.publication_point_cache_projection_value_bytes = value.len() as u64;
}
batch.put_cf(projection_cf, key.as_bytes(), value);
Ok(())
}
PublicationPointCacheProjectionWriteAction::Delete { manifest_rsync_uri } => {
let key = publication_point_cache_projection_key(manifest_rsync_uri);
batch.delete_cf(projection_cf, key.as_bytes());
Ok(())
}
}
batch.put_cf(projection_cf, key.as_bytes(), value);
Ok(())
}
impl RocksStore {
@@ -2975,6 +2989,22 @@ impl RocksStore {
vcir: &ValidatedCaInstanceResult,
roa_cache_context: Option<&RoaCacheProjectionContext>,
publication_point_projection: Option<&PublicationPointCacheProjection>,
) -> StorageResult<()> {
let publication_point_projection_action = publication_point_projection
.map(PublicationPointCacheProjectionWriteAction::Write)
.unwrap_or(PublicationPointCacheProjectionWriteAction::Keep);
self.put_vcir_with_projection_action(
vcir,
roa_cache_context,
publication_point_projection_action,
)
}
fn put_vcir_with_projection_action(
&self,
vcir: &ValidatedCaInstanceResult,
roa_cache_context: Option<&RoaCacheProjectionContext>,
publication_point_projection_action: PublicationPointCacheProjectionWriteAction<'_>,
) -> StorageResult<()> {
vcir.validate_internal()?;
let vcir_cf = self.cf(CF_VCIR)?;
@@ -3000,11 +3030,13 @@ impl RocksStore {
write_publication_point_cache_projection_to_batch(
pp_projection_cf,
&mut batch,
publication_point_projection,
publication_point_projection_action,
None,
)?;
self.write_batch(batch)?;
self.update_publication_point_cache_projection_index(publication_point_projection)?;
self.apply_publication_point_cache_projection_index_action(
publication_point_projection_action,
)?;
Ok(())
}
@@ -3032,6 +3064,22 @@ impl RocksStore {
vcir: &ValidatedCaInstanceResult,
roa_cache_context: Option<&RoaCacheProjectionContext>,
publication_point_projection: Option<&PublicationPointCacheProjection>,
) -> StorageResult<VcirReplaceTimingBreakdown> {
let publication_point_projection_action = publication_point_projection
.map(PublicationPointCacheProjectionWriteAction::Write)
.unwrap_or(PublicationPointCacheProjectionWriteAction::Keep);
self.replace_vcir_manifest_replay_meta_and_projection_action(
vcir,
roa_cache_context,
publication_point_projection_action,
)
}
pub(crate) fn replace_vcir_manifest_replay_meta_and_projection_action(
&self,
vcir: &ValidatedCaInstanceResult,
roa_cache_context: Option<&RoaCacheProjectionContext>,
publication_point_projection_action: PublicationPointCacheProjectionWriteAction<'_>,
) -> StorageResult<VcirReplaceTimingBreakdown> {
let mut timing = VcirReplaceTimingBreakdown {
rss_before_kb: process_vm_rss_kb(),
@@ -3085,7 +3133,7 @@ impl RocksStore {
write_publication_point_cache_projection_to_batch(
pp_projection_cf,
&mut batch,
publication_point_projection,
publication_point_projection_action,
Some(&mut timing),
)?;
timing.publication_point_cache_projection_encode_ms =
@@ -3100,7 +3148,9 @@ impl RocksStore {
let write_batch_started = std::time::Instant::now();
self.write_batch(batch)?;
self.update_publication_point_cache_projection_index(publication_point_projection)?;
self.apply_publication_point_cache_projection_index_action(
publication_point_projection_action,
)?;
timing.write_batch_ms = write_batch_started.elapsed().as_millis() as u64;
timing.rss_after_write_batch_kb = process_vm_rss_kb();
Ok(timing)
@@ -3357,13 +3407,25 @@ impl RocksStore {
Ok(entries)
}
fn apply_publication_point_cache_projection_index_action(
&self,
action: PublicationPointCacheProjectionWriteAction<'_>,
) -> StorageResult<()> {
match action {
PublicationPointCacheProjectionWriteAction::Keep => Ok(()),
PublicationPointCacheProjectionWriteAction::Write(projection) => {
self.update_publication_point_cache_projection_index(projection)
}
PublicationPointCacheProjectionWriteAction::Delete { manifest_rsync_uri } => {
self.delete_publication_point_cache_projection_index_entry(manifest_rsync_uri)
}
}
}
fn update_publication_point_cache_projection_index(
&self,
projection: Option<&PublicationPointCacheProjection>,
projection: &PublicationPointCacheProjection,
) -> StorageResult<()> {
let Some(projection) = projection else {
return Ok(());
};
let mut guard = self
.publication_point_cache_projection_index
.lock()
@@ -3376,17 +3438,22 @@ impl RocksStore {
index,
bytes: total_bytes,
} => {
*total_bytes = total_bytes.saturating_add(bytes.len());
index.insert(
if let Some(previous) = index.insert(
projection.manifest_rsync_uri.clone(),
Arc::<[u8]>::from(bytes),
);
Arc::<[u8]>::from(bytes.clone()),
) {
*total_bytes = total_bytes.saturating_sub(previous.len());
}
*total_bytes = total_bytes.saturating_add(bytes.len());
}
PublicationPointCacheProjectionIndexState::BuildingFromEmpty {
index,
bytes: total_bytes,
limit,
} => {
if let Some(previous) = index.remove(&projection.manifest_rsync_uri) {
*total_bytes = total_bytes.saturating_sub(previous.len());
}
if total_bytes.saturating_add(bytes.len()) <= *limit {
*total_bytes += bytes.len();
index.insert(
@@ -3400,11 +3467,13 @@ impl RocksStore {
PublicationPointCacheProjectionIndexState::LoadedMmap {
dirty, dirty_bytes, ..
} => {
*dirty_bytes = dirty_bytes.saturating_add(bytes.len());
dirty.insert(
if let Some(previous) = dirty.insert(
projection.manifest_rsync_uri.clone(),
Arc::<[u8]>::from(bytes),
);
Arc::<[u8]>::from(bytes.clone()),
) {
*dirty_bytes = dirty_bytes.saturating_sub(previous.len());
}
*dirty_bytes = dirty_bytes.saturating_add(bytes.len());
}
PublicationPointCacheProjectionIndexState::Uninitialized
| PublicationPointCacheProjectionIndexState::Disabled => {}
@@ -3412,6 +3481,84 @@ impl RocksStore {
Ok(())
}
fn delete_publication_point_cache_projection_index_entry(
&self,
manifest_rsync_uri: &str,
) -> StorageResult<()> {
let mut invalidate_mmap_files = false;
{
let mut guard = self
.publication_point_cache_projection_index
.lock()
.map_err(|e| {
StorageError::RocksDb(format!(
"publication point cache index lock poisoned: {e}"
))
})?;
match &mut *guard {
PublicationPointCacheProjectionIndexState::Loaded {
index,
bytes: total_bytes,
} => {
if let Some(previous) = index.remove(manifest_rsync_uri) {
*total_bytes = total_bytes.saturating_sub(previous.len());
}
}
PublicationPointCacheProjectionIndexState::BuildingFromEmpty {
index,
bytes: total_bytes,
..
} => {
if let Some(previous) = index.remove(manifest_rsync_uri) {
*total_bytes = total_bytes.saturating_sub(previous.len());
}
}
PublicationPointCacheProjectionIndexState::LoadedMmap { .. } => {
*guard = PublicationPointCacheProjectionIndexState::Disabled;
invalidate_mmap_files = true;
}
PublicationPointCacheProjectionIndexState::Uninitialized
| PublicationPointCacheProjectionIndexState::Disabled => {}
}
}
if invalidate_mmap_files {
self.remove_publication_point_cache_mmap_index_files()?;
let mut guard = self
.publication_point_cache_projection_index
.lock()
.map_err(|e| {
StorageError::RocksDb(format!(
"publication point cache index lock poisoned: {e}"
))
})?;
if matches!(*guard, PublicationPointCacheProjectionIndexState::Disabled) {
*guard = PublicationPointCacheProjectionIndexState::Uninitialized;
}
}
Ok(())
}
fn remove_publication_point_cache_mmap_index_files(&self) -> StorageResult<()> {
let dir = &self.publication_point_cache_index_dir;
if !dir.exists() {
return Ok(());
}
for entry in std::fs::read_dir(dir).map_err(|e| StorageError::RocksDb(e.to_string()))? {
let entry = entry.map_err(|e| StorageError::RocksDb(e.to_string()))?;
let path = entry.path();
let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
continue;
};
if file_name == "current.idx"
|| (file_name.starts_with("segment-") && file_name.ends_with(".idx"))
{
std::fs::remove_file(&path).map_err(|e| StorageError::RocksDb(e.to_string()))?;
}
}
Ok(())
}
pub fn refresh_publication_point_cache_mmap_index(
&self,
) -> StorageResult<Option<PpCacheIndexRefreshStats>> {

155
src/validation/tree_runner.rs
View File

@@ -26,11 +26,11 @@ use crate::replay::delta_archive::ReplayDeltaArchiveIndex;
use crate::report::{RfcRef, Warning};
use crate::storage::{
PackFile, PackTime, PublicationPointCacheChild, PublicationPointCacheOutput,
PublicationPointCacheProjection, RawByHashEntry, RoaCacheProjectionContext, RocksStore,
ValidatedCaInstanceResult, VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus,
VcirAuditSummary, VcirCcrManifestProjection, VcirChildEntry, VcirInstanceGate, VcirLocalOutput,
VcirLocalOutputPayload, VcirOutputType, VcirRelatedArtifact, VcirReplaceTimingBreakdown,
VcirSourceObjectType, VcirSummary,
PublicationPointCacheProjection, PublicationPointCacheProjectionWriteAction, RawByHashEntry,
RoaCacheProjectionContext, RocksStore, ValidatedCaInstanceResult, VcirArtifactKind,
VcirArtifactRole, VcirArtifactValidationStatus, VcirAuditSummary, VcirCcrManifestProjection,
VcirChildEntry, VcirInstanceGate, VcirLocalOutput, VcirLocalOutputPayload, VcirOutputType,
VcirRelatedArtifact, VcirReplaceTimingBreakdown, VcirSourceObjectType, VcirSummary,
};
use crate::sync::repo::{
sync_publication_point, sync_publication_point_replay, sync_publication_point_replay_delta,
@@ -90,6 +90,7 @@ pub(crate) struct PersistVcirTimingBreakdown {
pub(crate) embedded_store_ms: u64,
pub(crate) build_vcir_ms: u64,
pub(crate) replace_vcir_ms: u64,
pub(crate) publication_point_cache_future_notbefore_guarded: bool,
pub(crate) build_vcir: BuildVcirTimingBreakdown,
pub(crate) replace_vcir: VcirReplaceTimingBreakdown,
}
@@ -1021,6 +1022,9 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
.related_artifacts_ms
.saturating_mul(1_000_000),
);
if persist_vcir_timing.publication_point_cache_future_notbefore_guarded {
timing.record_count("publication_point_cache_future_notbefore_guarded", 1);
}
}
self.record_publication_point_step_ms(
&ca.manifest_rsync_uri,
@@ -1585,6 +1589,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
"persist_related_artifacts_ms": persist_vcir_timing.build_vcir.related_artifacts_ms,
"persist_vcir_struct_ms": persist_vcir_timing.build_vcir.struct_build_ms,
"persist_replace_breakdown": &persist_vcir_timing.replace_vcir,
"publication_point_cache_future_notbefore_guarded": persist_vcir_timing.publication_point_cache_future_notbefore_guarded,
"ccr_projection_build_ms": ccr_projection_build_ms,
"ccr_append_ms": ccr_append_ms,
"audit_build_ms": audit_build_ms,
@@ -1630,6 +1635,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
"persist_related_artifacts_ms": persist_vcir_timing.build_vcir.related_artifacts_ms,
"persist_vcir_struct_ms": persist_vcir_timing.build_vcir.struct_build_ms,
"persist_replace_breakdown": &persist_vcir_timing.replace_vcir,
"publication_point_cache_future_notbefore_guarded": persist_vcir_timing.publication_point_cache_future_notbefore_guarded,
"ccr_projection_build_ms": ccr_projection_build_ms,
"ccr_append_ms": ccr_append_ms,
"audit_build_ms": audit_build_ms,
@@ -4181,22 +4187,45 @@ fn persist_vcir_for_fresh_result_with_timing(
timing.build_vcir = build_vcir_timing;
let replace_vcir_started = std::time::Instant::now();
let publication_point_cache_projection = if write_publication_point_cache_projection {
Some(build_publication_point_cache_projection_from_fresh(
policy, ca, pack, &vcir,
)?)
let future_not_before_cache_guard = write_publication_point_cache_projection
&& publication_point_cache_has_future_not_before_risk(
pack,
objects,
child_audits,
validation_time,
policy,
);
timing.publication_point_cache_future_notbefore_guarded = future_not_before_cache_guard;
let publication_point_cache_projection =
if write_publication_point_cache_projection && !future_not_before_cache_guard {
Some(build_publication_point_cache_projection_from_fresh(
policy, ca, pack, &vcir,
)?)
} else {
None
};
let publication_point_cache_projection_action = if !write_publication_point_cache_projection {
PublicationPointCacheProjectionWriteAction::Keep
} else if future_not_before_cache_guard {
PublicationPointCacheProjectionWriteAction::Delete {
manifest_rsync_uri: &vcir.manifest_rsync_uri,
}
} else {
None
PublicationPointCacheProjectionWriteAction::Write(
publication_point_cache_projection
.as_ref()
.expect("publication point projection must exist when guard is not active"),
)
};
let replace_timing = store
.replace_vcir_manifest_replay_meta_and_projections(
.replace_vcir_manifest_replay_meta_and_projection_action(
&vcir,
Some(&RoaCacheProjectionContext {
parent_context_digest: parent_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(policy),
object_meta: objects.roa_cache_object_meta.clone(),
}),
publication_point_cache_projection.as_ref(),
publication_point_cache_projection_action,
)
.map_err(|e| format!("store VCIR and manifest replay meta failed: {e}"))?;
timing.replace_vcir_ms = replace_vcir_started.elapsed().as_millis() as u64;
@@ -4226,6 +4255,108 @@ fn build_publication_point_cache_projection_from_fresh(
.map_err(|e| e.to_string())
}
fn publication_point_cache_has_future_not_before_risk(
pack: &PublicationPointSnapshot,
objects: &crate::validation::objects::ObjectsOutput,
child_audits: &[ObjectAuditEntry],
validation_time: time::OffsetDateTime,
policy: &Policy,
) -> bool {
let mut files_by_uri: HashMap<&str, &PackFile> = HashMap::new();
for file in &pack.files {
files_by_uri.insert(file.rsync_uri.as_str(), file);
}
objects
.audit
.iter()
.chain(child_audits.iter())
.any(|entry| {
audit_entry_has_certificate_time_error(entry)
&& files_by_uri
.get(entry.rsync_uri.as_str())
.map(|file| {
audit_entry_has_future_not_before(entry, file, validation_time, policy)
})
.unwrap_or(true)
})
}
fn audit_entry_has_certificate_time_error(entry: &ObjectAuditEntry) -> bool {
entry.result == AuditObjectResult::Error
&& entry
.detail
.as_deref()
.is_some_and(|detail| detail.contains("certificate not valid at validation_time"))
}
fn audit_entry_has_future_not_before(
entry: &ObjectAuditEntry,
file: &PackFile,
validation_time: time::OffsetDateTime,
policy: &Policy,
) -> bool {
match entry.kind {
AuditObjectKind::Roa => signed_object_ee_not_before(file, policy, SignedObjectKind::Roa)
.map(|not_before| validation_time < not_before)
.unwrap_or(true),
AuditObjectKind::Aspa => signed_object_ee_not_before(file, policy, SignedObjectKind::Aspa)
.map(|not_before| validation_time < not_before)
.unwrap_or(true),
AuditObjectKind::Certificate | AuditObjectKind::RouterCertificate => file
.bytes()
.ok()
.and_then(|bytes| ResourceCertificate::decode_der(bytes).ok())
.map(|cert| validation_time < cert.tbs.validity_not_before)
.unwrap_or(true),
_ => true,
}
}
#[derive(Clone, Copy)]
enum SignedObjectKind {
Roa,
Aspa,
}
fn signed_object_ee_not_before(
file: &PackFile,
policy: &Policy,
kind: SignedObjectKind,
) -> Option<time::OffsetDateTime> {
let bytes = file.bytes().ok()?;
match kind {
SignedObjectKind::Roa => {
let object = RoaObject::decode_der_with_strict_options(
bytes,
policy.strict.cms_der,
policy.strict.name,
)
.ok()?;
Some(
object.signed_object.signed_data.certificates[0]
.resource_cert
.tbs
.validity_not_before,
)
}
SignedObjectKind::Aspa => {
let object = AspaObject::decode_der_with_strict_options(
bytes,
policy.strict.cms_der,
policy.strict.name,
)
.ok()?;
Some(
object.signed_object.signed_data.certificates[0]
.resource_cert
.tbs
.validity_not_before,
)
}
}
}
fn build_vcir_from_fresh_result_with_timing(
ca: &CaInstanceHandle,
pack: &PublicationPointSnapshot,

86
src/validation/tree_runner/tests.rs
View File

@@ -4,7 +4,8 @@ use crate::data_model::roa::RoaAfi;
use crate::fetch::rsync::LocalDirRsyncFetcher;
use crate::fetch::rsync::{RsyncFetchError, RsyncFetcher};
use crate::storage::{
PackFile, PackTime, PublicationPointCacheProjection, RawByHashEntry, RepositoryViewEntry,
PackFile, PackTime, PublicationPointCacheProjection,
PublicationPointCacheProjectionWriteAction, RawByHashEntry, RepositoryViewEntry,
RepositoryViewState, RocksStore, ValidatedCaInstanceResult, ValidatedManifestMeta,
VcirArtifactKind, VcirArtifactRole, VcirArtifactValidationStatus, VcirAuditSummary,
VcirChildEntry, VcirInstanceGate, VcirLocalOutput, VcirLocalOutputPayload, VcirOutputType,
@@ -1914,6 +1915,89 @@ fn seed_publication_point_cache_projection(
vcir
}
#[test]
fn publication_point_cache_future_notbefore_guard_detects_future_roa_notbefore_only() {
let uri = "rsync://example.test/repo/issuer/future.roa";
let bytes = std::fs::read(
std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/AS4538.roa"),
)
.expect("read ROA fixture");
let roa = RoaObject::decode_der(&bytes).expect("decode ROA fixture");
let ee = &roa.signed_object.signed_data.certificates[0].resource_cert;
let file = PackFile::from_bytes_compute_sha256(uri, bytes);
let pack = dummy_pack_with_files(vec![file]);
let objects = crate::validation::objects::ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: Default::default(),
audit: vec![ObjectAuditEntry {
rsync_uri: uri.to_string(),
sha256_hex: sha256_hex_from_32(&pack.files[0].sha256),
kind: AuditObjectKind::Roa,
result: AuditObjectResult::Error,
detail: Some(
"EE certificate path validation failed: certificate not valid at validation_time"
.to_string(),
),
}],
roa_cache_stats: Default::default(),
roa_cache_object_meta: Vec::new(),
};
assert!(publication_point_cache_has_future_not_before_risk(
&pack,
&objects,
&[],
ee.tbs.validity_not_before - time::Duration::seconds(1),
&Policy::default(),
));
assert!(!publication_point_cache_has_future_not_before_risk(
&pack,
&objects,
&[],
ee.tbs.validity_not_after + time::Duration::seconds(1),
&Policy::default(),
));
}
#[test]
fn publication_point_cache_delete_action_removes_existing_projection() {
let store_dir = tempfile::tempdir().expect("store dir");
let store = RocksStore::open(store_dir.path()).expect("open rocksdb");
let policy = Policy::default();
let validation_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::minutes(1);
let ca = publication_point_cache_fixture_ca();
let vcir = seed_publication_point_cache_projection(&store, &policy, &ca, validation_time);
assert!(
store
.get_publication_point_cache_projection_cached(&ca.manifest_rsync_uri)
.expect("load projection")
.is_some()
);
store
.replace_vcir_manifest_replay_meta_and_projection_action(
&vcir,
None,
PublicationPointCacheProjectionWriteAction::Delete {
manifest_rsync_uri: &vcir.manifest_rsync_uri,
},
)
.expect("delete projection");
assert!(
store
.get_publication_point_cache_projection_cached(&ca.manifest_rsync_uri)
.expect("load projection after delete")
.is_none()
);
}
fn publication_point_cache_fixture_ca() -> CaInstanceHandle {
CaInstanceHandle {
depth: 1,