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

增加cache restart和cache shutdown

Browse Source
This commit is contained in:
xiuting.xu
2026-07-24 11:10:32 +08:00
parent 5fb79a7c8d
commit 859a56c434
20 changed files with 949 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
tests/test_ccr.rs
View File

@@ -92,7 +92,7 @@ fn snapshot_to_payloads_with_options_skips_invalid_aspa_when_not_strict() {
}],
vaps: vec![ParsedAspa {
customer_asn: 174,
provider_asns: vec![0],
provider_asns: vec![0, 64496],
}],
};
@@ -102,29 +102,5 @@ fn snapshot_to_payloads_with_options_skips_invalid_aspa_when_not_strict() {
assert_eq!(conversion.payloads.len(), 1);
assert!(conversion.invalid_vrps.is_empty());
assert_eq!(conversion.invalid_vaps.len(), 1);
assert!(conversion.invalid_vaps[0].contains("provider list must not contain AS0"));
}
#[test]
fn generated_mini_ccr_files_are_parseable() {
let cases = [
("20260403T000001Z-mini-a.ccr", 2usize, 1usize),
("20260403T000101Z-mini-b.ccr", 3usize, 1usize),
("20260403T000201Z-mini-c.ccr", 2usize, 2usize),
];
for (name, expect_vrps, expect_vaps) in cases {
let snapshot = load_ccr_snapshot_from_file(fixture_path(name))
.unwrap_or_else(|e| panic!("failed to parse {}: {:?}", name, e));
assert_eq!(
snapshot.vrps.len(),
expect_vrps,
"vrp count mismatch for {name}"
);
assert_eq!(
snapshot.vaps.len(),
expect_vaps,
"vap count mismatch for {name}"
);
}
assert!(conversion.invalid_vaps[0].contains("provider list containing AS0"));
}

9
tests/test_pdu.rs
View File

@@ -217,9 +217,12 @@ fn aspa_announcement_rejects_as0() {
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains("AS0"));
let err = PayloadAspa::new(Asn::from(64496u32), vec![Asn::from(0u32)])
.validate_announcement()
.unwrap_err();
let err = PayloadAspa::new(
Asn::from(64496u32),
vec![Asn::from(0u32), Asn::from(64497u32)],
)
.validate_announcement()
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains("AS0"));
}

4
tests/test_rtr_debug_client_ssh_cli.rs
View File

@@ -32,7 +32,9 @@ fn ssh_requires_key() {
]);
assert!(!output.status.success());
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(stderr.contains("SSH mode requires --ssh-key <path>"));
assert!(stderr.contains("SSH mode requires authentication"));
assert!(stderr.contains("--ssh-key <path>"));
assert!(stderr.contains("--ssh-password <value>"));
}
#[test]

146
tests/test_rtr_error_codes.rs Normal file
View File

@@ -0,0 +1,146 @@
use rpki::rtr::error_type::{
ERROR_CODE_VERSION_SPECS, ErrorCode, ErrorCodeDisposition, ErrorCodeVersionSpec,
};
#[test]
fn error_code_version_specs_match_rfc_and_draft_table() {
use ErrorCode::*;
use ErrorCodeDisposition::*;
let expected = [
ErrorCodeVersionSpec {
code: CorruptData,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: InternalError,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: NoDataAvailable,
min_version: 0,
max_version: 2,
disposition: NonFatal,
},
ErrorCodeVersionSpec {
code: InvalidRequest,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: UnsupportedProtocolVersion,
min_version: 0,
max_version: 1,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: UnsupportedProtocolVersion,
min_version: 2,
max_version: 2,
disposition: NonFatal,
},
ErrorCodeVersionSpec {
code: UnsupportedPduType,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: WithdrawalOfUnknownRecord,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: DuplicateAnnouncement,
min_version: 0,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: UnexpectedProtocolVersion,
min_version: 1,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: AspaProviderListError,
min_version: 2,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: TransportFailed,
min_version: 2,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: OrderingError,
min_version: 2,
max_version: 2,
disposition: Fatal,
},
ErrorCodeVersionSpec {
code: CacheRestart,
min_version: 2,
max_version: 2,
disposition: NonFatal,
},
ErrorCodeVersionSpec {
code: CacheShutdown,
min_version: 2,
max_version: 2,
disposition: Fatal,
},
];
assert_eq!(ERROR_CODE_VERSION_SPECS, &expected);
}
#[test]
fn error_code_definition_scope_is_version_sensitive() {
assert!(ErrorCode::DuplicateAnnouncement.is_defined_in_version(0));
assert!(!ErrorCode::UnexpectedProtocolVersion.is_defined_in_version(0));
assert!(ErrorCode::UnexpectedProtocolVersion.is_defined_in_version(1));
assert!(!ErrorCode::AspaProviderListError.is_defined_in_version(1));
assert!(ErrorCode::CacheShutdown.is_defined_in_version(2));
}
#[test]
fn error_code_disposition_is_version_sensitive() {
assert_eq!(
ErrorCode::NoDataAvailable.disposition_in_version(0),
Some(ErrorCodeDisposition::NonFatal)
);
assert_eq!(
ErrorCode::UnsupportedProtocolVersion.disposition_in_version(1),
Some(ErrorCodeDisposition::Fatal)
);
assert_eq!(
ErrorCode::UnsupportedProtocolVersion.disposition_in_version(2),
Some(ErrorCodeDisposition::NonFatal)
);
assert_eq!(
ErrorCode::CacheRestart.disposition_in_version(2),
Some(ErrorCodeDisposition::NonFatal)
);
assert_eq!(ErrorCode::CacheRestart.disposition_in_version(1), None);
assert_eq!(
ErrorCode::CacheShutdown.disposition_in_version(2),
Some(ErrorCodeDisposition::Fatal)
);
}
#[test]
fn raw_error_code_registry_mapping_accepts_defined_codes_only() {
assert_eq!(ErrorCode::try_from(12), Ok(ErrorCode::CacheRestart));
assert_eq!(ErrorCode::try_from(13), Ok(ErrorCode::CacheShutdown));
assert!(ErrorCode::try_from(14).is_err());
assert!(ErrorCode::try_from(255).is_err());
}

12
tests/test_server_transports.rs
View File

@@ -16,8 +16,8 @@ use tokio_rustls::TlsConnector;
use rpki::rtr::cache::{RtrCacheBuilder, SessionIds, SharedRtrCache};
use rpki::rtr::payload::Timing;
use rpki::rtr::pdu::{CacheResponse, EndOfDataV1, ResetQuery};
use rpki::rtr::server::RtrService;
use rpki::rtr::server::ssh::SshAuthMode;
use rpki::rtr::server::{RtrService, RtrShutdownReason};
use russh::client;
use russh::keys;
use russh::keys::ssh_key::LineEnding;
@@ -135,7 +135,7 @@ async fn unified_server_tcp_handles_reset_query() {
assert_eq!(eod.session_id(), 42);
assert_eq!(eod.serial_number(), 100);
running.shutdown();
running.shutdown(RtrShutdownReason::CloseOnly);
running.wait().await;
}
@@ -171,7 +171,7 @@ async fn unified_server_tls_handles_reset_query() {
assert_eq!(eod.session_id(), 42);
assert_eq!(eod.serial_number(), 100);
running.shutdown();
running.shutdown(RtrShutdownReason::CloseOnly);
running.wait().await;
}
@@ -220,7 +220,7 @@ async fn unified_server_ssh_opens_listener_and_emits_banner() {
banner
);
running.shutdown();
running.shutdown(RtrShutdownReason::CloseOnly);
running.wait().await;
}
@@ -295,7 +295,7 @@ async fn unified_server_ssh_accepts_password_when_configured() {
assert_eq!(eod.session_id(), 42);
assert_eq!(eod.serial_number(), 100);
running.shutdown();
running.shutdown(RtrShutdownReason::CloseOnly);
running.wait().await;
}
@@ -344,6 +344,6 @@ async fn unified_server_ssh_rejects_password_when_not_configured() {
.expect("password auth result");
assert!(!auth_result.success(), "password auth should be rejected");
running.shutdown();
running.shutdown(RtrShutdownReason::CloseOnly);
running.wait().await;
}

124
tests/test_session.rs
View File

@@ -34,6 +34,7 @@ use rpki::rtr::pdu::{
};
use rpki::rtr::server::connection::handle_tls_connection;
use rpki::rtr::server::tls::load_rustls_server_config_with_options;
use rpki::rtr::server::{RtrServiceControl, RtrShutdownReason};
use rpki::rtr::session::RtrSession;
use rpki::rtr::store::RtrStore;
@@ -46,14 +47,14 @@ async fn start_session_server(
) -> (
SocketAddr,
broadcast::Sender<()>,
watch::Sender<bool>,
watch::Sender<RtrServiceControl>,
JoinHandle<()>,
) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (notify_tx, notify_rx) = broadcast::channel(16);
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running);
let handle = tokio::spawn(async move {
let Ok((stream, _)) = listener.accept().await else {
@@ -73,14 +74,14 @@ async fn start_session_server_with_transport_timeout(
) -> (
SocketAddr,
broadcast::Sender<()>,
watch::Sender<bool>,
watch::Sender<RtrServiceControl>,
JoinHandle<()>,
) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (notify_tx, notify_rx) = broadcast::channel(16);
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running);
let handle = tokio::spawn(async move {
let Ok((stream, _)) = listener.accept().await else {
@@ -99,14 +100,14 @@ async fn start_session_server_returning_result(
cache: SharedRtrCache,
) -> (
SocketAddr,
watch::Sender<bool>,
watch::Sender<RtrServiceControl>,
JoinHandle<anyhow::Result<()>>,
) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (_notify_tx, notify_rx) = broadcast::channel(16);
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running);
let handle = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
@@ -119,7 +120,7 @@ async fn start_session_server_returning_result(
async fn start_tls_session_server(
cache: SharedRtrCache,
) -> (SocketAddr, watch::Sender<bool>, JoinHandle<()>) {
) -> (SocketAddr, watch::Sender<RtrServiceControl>, JoinHandle<()>) {
start_tls_session_server_with_cert(cache, "server.crt", "server.key").await
}
@@ -127,12 +128,12 @@ async fn start_tls_session_server_with_cert(
cache: SharedRtrCache,
cert_name: &str,
key_name: &str,
) -> (SocketAddr, watch::Sender<bool>, JoinHandle<()>) {
) -> (SocketAddr, watch::Sender<RtrServiceControl>, JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (_notify_tx, notify_rx) = broadcast::channel(16);
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running);
let tls_config = Arc::new(
load_rustls_server_config_with_options(
fixture_path(cert_name),
@@ -167,19 +168,22 @@ async fn start_tls_session_server_with_cert(
async fn shutdown_server(
mut client: TcpStream,
shutdown_tx: watch::Sender<bool>,
shutdown_tx: watch::Sender<RtrServiceControl>,
server_handle: JoinHandle<()>,
) {
shutdown_io(&mut client, shutdown_tx, server_handle).await;
}
async fn shutdown_io<S>(io: &mut S, shutdown_tx: watch::Sender<bool>, server_handle: JoinHandle<()>)
where
async fn shutdown_io<S>(
io: &mut S,
shutdown_tx: watch::Sender<RtrServiceControl>,
server_handle: JoinHandle<()>,
) where
S: AsyncWrite + Unpin,
{
let _ = io.shutdown().await;
let _ = shutdown_tx.send(true);
let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly));
match timeout(Duration::from_secs(1), server_handle).await {
Ok(join_res) => {
@@ -252,7 +256,7 @@ async fn connect_tls_client_with_server_name(
connector.connect(server_name, tcp).await.unwrap()
}
/// 用于 dump Serial Notify，保持输出风格一致。
/// Dump Serial Notify in the same style as the other test helpers.
fn dump_serial_notify(notify: &SerialNotify) -> serde_json::Value {
json!({
"version": notify.version(),
@@ -310,7 +314,7 @@ async fn wait_for_store_serials(store: &RtrStore, expected: [u32; 3]) {
);
}
/// 测试：Reset Query 会返回完整 snapshot，并以 End of Data 结束响应。
/// Reset Query returns a full snapshot and finishes with End of Data.
#[tokio::test]
async fn reset_query_returns_snapshot_and_end_of_data() {
let prefix = IPAddressPrefix {
@@ -397,6 +401,70 @@ async fn reset_query_uses_version_specific_session_id() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
#[tokio::test]
async fn version_two_cache_restart_shutdown_sends_error_code_12() {
let cache = RtrCacheBuilder::new()
.session_ids(SessionIds::from_array([40, 41, 42]))
.serials(serials_all(100))
.timing(Timing::new(600, 600, 7200))
.build();
let server_cache = shared_cache(cache);
let (addr, _notify_tx, shutdown_tx, server_handle) = start_session_server(server_cache).await;
let mut client = TcpStream::connect(addr).await.unwrap();
ResetQuery::new(2).write(&mut client).await.unwrap();
let _response = CacheResponse::read(&mut client).await.unwrap();
let _eod = EndOfDataV1::read(&mut client).await.unwrap();
shutdown_tx
.send(RtrServiceControl::Shutdown(RtrShutdownReason::CacheRestart))
.unwrap();
let report = ErrorReport::read(&mut client).await.unwrap();
assert_error_report_matches(&report, 2, ErrorCode::CacheRestart, &[]);
assert!(String::from_utf8_lossy(report.text()).contains("restarting"));
timeout(Duration::from_secs(1), server_handle)
.await
.expect("server task did not exit within timeout")
.unwrap();
}
#[tokio::test]
async fn version_one_cache_restart_shutdown_does_not_send_v2_error_code() {
let cache = RtrCacheBuilder::new()
.session_ids(SessionIds::from_array([40, 41, 42]))
.serials(serials_all(100))
.timing(Timing::new(600, 600, 7200))
.build();
let server_cache = shared_cache(cache);
let (addr, _notify_tx, shutdown_tx, server_handle) = start_session_server(server_cache).await;
let mut client = TcpStream::connect(addr).await.unwrap();
ResetQuery::new(1).write(&mut client).await.unwrap();
let _response = CacheResponse::read(&mut client).await.unwrap();
let _eod = EndOfDataV1::read(&mut client).await.unwrap();
shutdown_tx
.send(RtrServiceControl::Shutdown(RtrShutdownReason::CacheRestart))
.unwrap();
let read_res = timeout(Duration::from_secs(1), Header::read(&mut client))
.await
.expect("timed out waiting for session close");
assert!(
read_res.is_err(),
"v1 session must not receive v2 error code 12"
);
timeout(Duration::from_secs(1), server_handle)
.await
.expect("server task did not exit within timeout")
.unwrap();
}
#[tokio::test]
async fn restart_restores_versioned_state_and_serves_queries() {
let dir = tempfile::tempdir().unwrap();
@@ -488,7 +556,7 @@ async fn restart_restores_versioned_state_and_serves_queries() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：当 Serial Query 的 session_id 和 serial 都与当前 cache 一致时，仅返回 End of Data。
/// Serial Query returns only End of Data when the client is up to date.
#[tokio::test]
async fn serial_query_returns_end_of_data_when_up_to_date() {
let cache = RtrCacheBuilder::new()
@@ -534,7 +602,7 @@ async fn serial_query_returns_end_of_data_when_up_to_date() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：当已建立 session 后收到错误的 session_id 时，返回 CorruptData 并关闭连接。
/// Established sessions reject mismatched Session IDs with Corrupt Data.
#[tokio::test]
async fn serial_query_returns_corrupt_data_when_session_id_mismatch() {
let cache = RtrCacheBuilder::new()
@@ -581,7 +649,7 @@ async fn serial_query_returns_corrupt_data_when_session_id_mismatch() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：当增量更新可用时，Serial Query 返回 Cache Response + delta payload + End of Data。
/// Serial Query returns Cache Response, delta payloads, and End of Data when incremental updates exist.
#[tokio::test]
async fn serial_query_returns_deltas_when_incremental_update_available() {
let prefix = IPAddressPrefix {
@@ -751,7 +819,7 @@ async fn serial_query_returns_cache_reset_for_future_serial_across_wraparound()
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：Reset Query 返回的 payload 顺序符合当前实现的 RTR 排序规则。
/// Reset Query returns payloads in RTR ordering.
#[tokio::test]
async fn reset_query_returns_payloads_in_rtr_order() {
let v4_low_prefix = IPAddressPrefix {
@@ -850,7 +918,7 @@ async fn reset_query_returns_payloads_in_rtr_order() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：Serial Query 返回的增量中，announcement 在前，withdrawal 在后，且各自内部顺序符合当前实现。
/// Serial Query returns announcements before withdrawals, with each side sorted.
#[tokio::test]
async fn serial_query_returns_announcements_before_withdrawals() {
let announced_low_prefix = IPAddressPrefix {
@@ -984,7 +1052,7 @@ async fn serial_query_returns_announcements_before_withdrawals() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：session 建立后，收到 notify 广播时会发送 Serial Notify。
/// Established sessions send Serial Notify after receiving cache update notifications.
#[tokio::test]
async fn established_session_sends_serial_notify() {
let cache = RtrCacheBuilder::new()
@@ -1026,7 +1094,7 @@ async fn established_session_sends_serial_notify() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：首个 PDU 版本过高时，返回 UnsupportedProtocolVersion 错误并关闭连接。
/// A too-high first-PDU protocol version returns Unsupported Protocol Version.
#[tokio::test]
async fn first_pdu_with_too_high_version_returns_unsupported_version_error() {
let cache = RtrCacheBuilder::new()
@@ -1066,7 +1134,7 @@ async fn first_pdu_with_too_high_version_returns_unsupported_version_error() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：版本协商完成后，如果后续请求更换了协议版本，返回 UnexpectedProtocolVersion 并关闭连接。
/// Established sessions reject later protocol version changes.
#[tokio::test]
async fn session_rejects_version_change_after_negotiation() {
let cache = RtrCacheBuilder::new()
@@ -1119,7 +1187,7 @@ async fn session_rejects_version_change_after_negotiation() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：在版本协商完成前，即使收到 notify 广播，也不能发送 Serial Notify。
/// Notifications before version negotiation do not send Serial Notify.
#[tokio::test]
async fn notify_is_not_sent_before_version_negotiation() {
let cache = RtrCacheBuilder::new()
@@ -1156,7 +1224,7 @@ async fn notify_is_not_sent_before_version_negotiation() {
shutdown_server(client, shutdown_tx, server_handle).await;
}
/// 测试：同一 session 在一分钟窗口内连续收到 notify 广播时，只会发送一个 Serial Notify。
/// Serial Notify is rate-limited to once per minute per session.
#[tokio::test]
async fn serial_notify_is_rate_limited_to_once_per_minute() {
let cache = RtrCacheBuilder::new()
@@ -1908,7 +1976,7 @@ async fn invalid_timing_prevents_end_of_data_response() {
"server should close instead of sending invalid EndOfData"
);
let _ = shutdown_tx.send(true);
let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly));
let join = timeout(Duration::from_secs(1), server_handle)
.await
.expect("server task did not exit within timeout")
@@ -1947,7 +2015,7 @@ async fn invalid_aspa_prevents_snapshot_response() {
"server should close instead of sending invalid ASPA"
);
let _ = shutdown_tx.send(true);
let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly));
let join = timeout(Duration::from_secs(1), server_handle)
.await
.expect("server task did not exit within timeout")
@@ -1989,7 +2057,7 @@ async fn invalid_router_key_prevents_snapshot_response() {
"server should close instead of sending invalid RouterKey"
);
let _ = shutdown_tx.send(true);
let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly));
let join = timeout(Duration::from_secs(1), server_handle)
.await
.expect("server task did not exit within timeout")