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

rtr client 开发

Browse Source
This commit is contained in:
xiuting.xu
2026-04-15 15:43:59 +08:00
parent c1d3112a45
commit 17c1a02c90
27 changed files with 1991 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
src/bin/rpki_rs_test_client/README.md Normal file
View File

@@ -0,0 +1,112 @@
# rpki_rs_test_client
`rpki_rs_test_client` 是一个基于 `rpki-rs` RTR 客户端接口的测试工具，参数风格对齐 `rtr_debug_client`。
实现上直接调用外部 crate 的 client API（`Client` / `PayloadTarget`），不重复实现 RTR 客户端状态机。
## 构建
```bash
cargo build --bin rpki_rs_test_client
```
## 基本用法
```bash
cargo run --bin rpki_rs_test_client -- <addr> <version> [reset|serial|serial <session_id> <serial>] [options]
```
默认值：
- `addr`: `127.0.0.1:323`
- `version`: `2`
- `mode`: `reset`
## 常用参数
- `--steps <n>`: 执行 `client.step()` 次数（默认 `1`）
- `--follow`: bootstrap 结束后持续执行 `client.step()`（常驻模式）
- `--print-records`: 打印当前收敛后的 payload 记录
- `--assert-min-records <n>`: 断言收敛记录数下限
- `--assert-substr <text>`: 在 payload 的 `Debug` 输出中做字符串断言（可重复）
TLS 参数：
- `--tls`
- `--ca-cert <path>`
- `--server-name <name>`
- `--client-cert <path>`
- `--client-key <path>`
## 限制说明
- 当前 `rpki-rs v0.18` client API 不支持显式覆盖初始版本，因此这里只接受 `version=2`。
- 支持 `serial`（无参数）模式：会基于 client 内部 state 自动走 serial 更新。
- 当前不支持 `serial <session_id> <serial>` 显式注入状态（传入会直接报错）。
## 示例
TCP 连通 + 最小记录数断言：
```bash
cargo run --bin rpki_rs_test_client -- \
127.0.0.1:323 \
2 reset \
--steps 1 \
--assert-min-records 1
```
自动 serial（无需传 sid/serial）：
```bash
cargo run --bin rpki_rs_test_client -- \
127.0.0.1:323 \
2 serial --steps 2 --follow
```
结合 `mini_data` 的内容做字符串断言：
```bash
cargo run --bin rpki_rs_test_client -- \
127.0.0.1:323 \
2 reset \
--assert-substr "10.0.1.0" \
--assert-substr "65003"
```
TLS 场景：
```bash
cargo run --bin rpki_rs_test_client -- \
127.0.0.1:324 \
2 reset \
--tls \
--ca-cert tests/fixtures/tls/client-ca.crt \
--server-name localhost
```
## Docker 启动（deploy）
构建并启动（Linux 服务器，`host` 网络）：
```bash
docker compose -f deploy/rpki-rs-client/docker-compose.yml up --build
```
按需覆盖运行参数（覆盖 compose 默认 `command`）：
```bash
docker compose -f deploy/rpki-rs-client/docker-compose.yml run --rm \
rpki-rs-test-client 127.0.0.1:323 2 reset --steps 1 --assert-min-records 1
```
常驻跟进模式（先 bootstrap，再持续 serial/notify）：
```bash
docker compose -f deploy/rpki-rs-client/docker-compose.yml run --rm \
rpki-rs-test-client 127.0.0.1:323 2 serial --steps 2 --follow
```
停止：
```bash
docker compose -f deploy/rpki-rs-client/docker-compose.yml down
```

563
src/bin/rpki_rs_test_client/main.rs Normal file
View File

@@ -0,0 +1,563 @@
use std::env;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use rustls::{ClientConfig as RustlsClientConfig, RootCertStore};
use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::TcpStream;
use tokio_rustls::TlsConnector;
use rpki_rs::rtr::client::{Client, PayloadError, PayloadTarget};
use rpki_rs::rtr::payload::{Action, Payload, Timing};
const DEFAULT_TIMEOUT_SECS: u64 = 10;
const DEFAULT_STEPS: usize = 1;
trait AsyncStream: AsyncRead + AsyncWrite + Unpin + Send {}
impl<T> AsyncStream for T where T: AsyncRead + AsyncWrite + Unpin + Send {}
type DynStream = Box<dyn AsyncStream>;
#[derive(Debug, Clone)]
struct Config {
addr: String,
version: u8,
mode: QueryMode,
steps: usize,
follow: bool,
transport: TransportConfig,
assert_substr: Vec<String>,
assert_min_records: Option<usize>,
print_records: bool,
}
#[derive(Debug, Clone, Copy)]
enum QueryMode {
Reset,
SerialAuto,
Serial { session_id: u16, serial: u32 },
}
impl Config {
fn from_args() -> io::Result<Self> {
let mut args = env::args().skip(1);
let mut positional = Vec::new();
let mut steps = DEFAULT_STEPS;
let mut follow = false;
let mut transport = TransportConfig::Tcp;
let mut assert_substr = Vec::new();
let mut assert_min_records = None;
let mut print_records = false;
while let Some(arg) = args.next() {
match arg.as_str() {
"--version" => {
let _ = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--version requires value")
})?;
// rpki-rs v0.18 client only exposes Client::new without
// initial-version override. Keep this option as reserved.
}
"--steps" => {
let v = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--steps requires value")
})?;
steps = parse_usize_arg(&v, "--steps")?;
}
"--follow" => {
follow = true;
}
"--tls" => {
if matches!(transport, TransportConfig::Tcp) {
transport = TransportConfig::Tls(TlsConfig::default());
}
}
"--ca-cert" => {
let v = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--ca-cert requires path")
})?;
ensure_tls(&mut transport)?.ca_cert = Some(PathBuf::from(v));
}
"--server-name" => {
let v = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--server-name requires value")
})?;
ensure_tls(&mut transport)?.server_name = Some(v);
}
"--client-cert" => {
let v = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--client-cert requires path")
})?;
ensure_tls(&mut transport)?.client_cert = Some(PathBuf::from(v));
}
"--client-key" => {
let v = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--client-key requires path")
})?;
ensure_tls(&mut transport)?.client_key = Some(PathBuf::from(v));
}
"--assert-substr" => {
let v = args.next().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"--assert-substr requires value",
)
})?;
assert_substr.push(v);
}
"--assert-min-records" => {
let v = args.next().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"--assert-min-records requires value",
)
})?;
assert_min_records = Some(parse_usize_arg(&v, "--assert-min-records")?);
}
"--print-records" => {
print_records = true;
}
"--timeout" => {
let _ = args.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "--timeout requires value")
})?;
// This binary relies on rpki-rs client's built-in IO timeout.
}
_ if arg.starts_with("--") => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unknown option '{}'", arg),
));
}
_ => positional.push(arg),
}
}
let mut positional = positional.into_iter();
let addr = positional
.next()
.unwrap_or_else(|| "127.0.0.1:323".to_string());
let version = positional
.next()
.map(|v| parse_u8_arg(&v, "version"))
.transpose()?
.unwrap_or(2);
let mode = match positional.next().as_deref() {
None | Some("reset") => QueryMode::Reset,
Some("serial") if positional.clone().next().is_none() => QueryMode::SerialAuto,
Some("serial") => {
let session_id = parse_u16_arg(
&positional.next().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"serial mode requires session_id and serial",
)
})?,
"session_id",
)?;
let serial = parse_u32_arg(
&positional.next().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "serial mode requires serial")
})?,
"serial",
)?;
QueryMode::Serial { session_id, serial }
}
Some(other) => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid mode '{}', expected 'reset' or 'serial'", other),
));
}
};
let transport = finalize_transport(transport, &addr)?;
Ok(Self {
addr,
version,
mode,
steps,
follow,
transport,
assert_substr,
assert_min_records,
print_records,
})
}
}
#[derive(Debug, Clone)]
enum TransportConfig {
Tcp,
Tls(TlsConfig),
}
#[derive(Debug, Clone, Default)]
struct TlsConfig {
server_name: Option<String>,
ca_cert: Option<PathBuf>,
client_cert: Option<PathBuf>,
client_key: Option<PathBuf>,
}
fn ensure_tls(transport: &mut TransportConfig) -> io::Result<&mut TlsConfig> {
if matches!(transport, TransportConfig::Tcp) {
*transport = TransportConfig::Tls(TlsConfig::default());
}
match transport {
TransportConfig::Tls(cfg) => Ok(cfg),
TransportConfig::Tcp => Err(io::Error::other("tls configuration unavailable")),
}
}
fn finalize_transport(transport: TransportConfig, addr: &str) -> io::Result<TransportConfig> {
match transport {
TransportConfig::Tcp => Ok(TransportConfig::Tcp),
TransportConfig::Tls(mut cfg) => {
let ca_cert = cfg.ca_cert.take().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"TLS mode requires --ca-cert <path>",
)
})?;
match (&cfg.client_cert, &cfg.client_key) {
(Some(_), Some(_)) | (None, None) => {}
_ => {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"TLS client auth requires both --client-cert and --client-key",
));
}
}
let server_name = cfg
.server_name
.take()
.or_else(|| default_server_name_for_addr(addr))
.ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"TLS mode requires --server-name or parseable host",
)
})?;
Ok(TransportConfig::Tls(TlsConfig {
server_name: Some(server_name),
ca_cert: Some(ca_cert),
client_cert: cfg.client_cert,
client_key: cfg.client_key,
}))
}
}
}
#[derive(Debug, Default)]
struct InMemoryTarget {
records: Vec<Payload>,
timing: Option<Timing>,
announced: u64,
withdrawn: u64,
}
impl InMemoryTarget {
fn dump_text(&self) -> String {
self.records
.iter()
.map(|p| format!("{:?}", p))
.collect::<Vec<_>>()
.join("\n")
}
}
impl PayloadTarget for InMemoryTarget {
type Update = Vec<(Action, Payload)>;
fn start(&mut self, reset: bool) -> Self::Update {
if reset {
self.records.clear();
}
Vec::new()
}
fn apply(&mut self, update: Self::Update, timing: Timing) -> Result<(), PayloadError> {
for (action, payload) in update {
match action {
Action::Announce => {
self.announced += 1;
if self.records.iter().any(|p| p == &payload) {
return Err(PayloadError::DuplicateAnnounce);
}
self.records.push(payload);
}
Action::Withdraw => {
self.withdrawn += 1;
if let Some(pos) = self.records.iter().position(|p| p == &payload) {
self.records.swap_remove(pos);
} else {
return Err(PayloadError::UnknownWithdraw);
}
}
}
}
self.timing = Some(timing);
Ok(())
}
}
#[tokio::main]
async fn main() -> io::Result<()> {
let config = Config::from_args()?;
println!("== rpki_rs_test_client ==");
println!("target : {}", config.addr);
println!("version : {}", config.version);
println!(
"mode : {}",
match config.mode {
QueryMode::Reset => "reset".to_string(),
QueryMode::SerialAuto => "serial(auto)".to_string(),
QueryMode::Serial { session_id, serial } => {
format!("serial sid={} serial={}", session_id, serial)
}
}
);
println!("steps : {}", config.steps);
println!("follow : {}", config.follow);
println!(
"timeout : {}s (from rpki-rs client IO timeout)",
DEFAULT_TIMEOUT_SECS
);
if config.version != 2 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"rpki-rs v0.18 client API does not expose initial-version override; please use version 2",
));
}
if matches!(config.mode, QueryMode::Serial { .. }) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"rpki-rs v0.18 Client::new cannot bootstrap explicit serial state; use reset mode",
));
}
let stream = connect_stream(&config).await?;
let target = InMemoryTarget::default();
let mut client = Client::new(stream, target, None);
let bootstrap_steps = match config.mode {
QueryMode::SerialAuto if config.steps < 2 => 2,
_ => config.steps,
};
if bootstrap_steps != config.steps {
println!(
"steps adjusted : {} -> {} (serial(auto) needs at least 2 steps)",
config.steps, bootstrap_steps
);
}
for idx in 0..bootstrap_steps {
client.step().await.map_err(|err| {
io::Error::new(err.kind(), format!("step {} failed: {}", idx + 1, err))
})?;
println!("[step] bootstrap {} ok", idx + 1);
}
if config.follow {
println!("[follow] enabled, entering continuous step loop");
let mut step_index = bootstrap_steps;
loop {
step_index += 1;
client.step().await.map_err(|err| {
io::Error::new(err.kind(), format!("step {} failed: {}", step_index, err))
})?;
println!("[step] follow {} ok", step_index);
}
}
let negotiated_state = client.state();
let target = client.into_target();
println!("state : {:?}", negotiated_state);
if let Some(timing) = target.timing {
println!(
"timing : refresh={} retry={} expire={}",
timing.refresh, timing.retry, timing.expire
);
}
println!("records : {}", target.records.len());
println!(
"updates : announce={} withdraw={}",
target.announced, target.withdrawn
);
if config.print_records {
println!("-- records --");
for rec in &target.records {
println!("{:?}", rec);
}
}
run_assertions(&config, &target)?;
println!("[assert] passed");
Ok(())
}
fn run_assertions(config: &Config, target: &InMemoryTarget) -> io::Result<()> {
if let Some(min) = config.assert_min_records
&& target.records.len() < min
{
return Err(io::Error::other(format!(
"assertion failed: records {} < {}",
target.records.len(),
min
)));
}
if !config.assert_substr.is_empty() {
let dump = target.dump_text();
for needle in &config.assert_substr {
if !dump.contains(needle) {
return Err(io::Error::other(format!(
"assertion failed: missing substring '{}'",
needle
)));
}
}
}
Ok(())
}
async fn connect_stream(config: &Config) -> io::Result<DynStream> {
match &config.transport {
TransportConfig::Tcp => Ok(Box::new(TcpStream::connect(&config.addr).await?)),
TransportConfig::Tls(tls) => connect_tls_stream(&config.addr, tls).await,
}
}
async fn connect_tls_stream(addr: &str, tls: &TlsConfig) -> io::Result<DynStream> {
let stream = TcpStream::connect(addr).await?;
let connector = build_tls_connector(tls)?;
let server_name_str = tls
.server_name
.as_ref()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing TLS server name"))?;
let server_name = ServerName::try_from(server_name_str.clone()).map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid TLS server name '{}': {}", server_name_str, err),
)
})?;
let tls_stream = connector.connect(server_name, stream).await.map_err(|err| {
io::Error::new(
io::ErrorKind::ConnectionAborted,
format!("TLS handshake failed: {}", err),
)
})?;
Ok(Box::new(tls_stream))
}
fn build_tls_connector(tls: &TlsConfig) -> io::Result<TlsConnector> {
let ca_cert_path = tls
.ca_cert
.as_ref()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing TLS CA cert"))?;
let ca_certs = load_certs(ca_cert_path)?;
let mut roots = RootCertStore::empty();
let (added, _) = roots.add_parsable_certificates(ca_certs);
if added == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("no valid CA certs in {}", ca_cert_path.display()),
));
}
let builder = RustlsClientConfig::builder().with_root_certificates(roots);
let cfg = match (&tls.client_cert, &tls.client_key) {
(Some(cert_path), Some(key_path)) => {
let certs = load_certs(cert_path)?;
let key = load_private_key(key_path)?;
builder.with_client_auth_cert(certs, key).map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid client cert/key: {}", err),
)
})?
}
(None, None) => builder.with_no_client_auth(),
_ => unreachable!(),
};
Ok(TlsConnector::from(Arc::new(cfg)))
}
fn load_certs(path: &Path) -> io::Result<Vec<CertificateDer<'static>>> {
let mut reader = std::io::BufReader::new(std::fs::File::open(path)?);
let certs = rustls_pemfile::certs(&mut reader)
.collect::<Result<Vec<_>, _>>()
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
if certs.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("no certs found in {}", path.display()),
));
}
Ok(certs)
}
fn load_private_key(path: &Path) -> io::Result<PrivateKeyDer<'static>> {
let mut reader = std::io::BufReader::new(std::fs::File::open(path)?);
rustls_pemfile::private_key(&mut reader)
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?
.ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("no private key in {}", path.display()),
)
})
}
fn default_server_name_for_addr(addr: &str) -> Option<String> {
if let Some(rest) = addr.strip_prefix('[') {
return rest.split(']').next().map(str::to_string);
}
addr.rsplit_once(':').map(|(host, _)| host.to_string())
}
fn parse_u8_arg(value: &str, name: &str) -> io::Result<u8> {
value.parse::<u8>().map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid {} '{}': {}", name, value, err),
)
})
}
fn parse_u16_arg(value: &str, name: &str) -> io::Result<u16> {
value.parse::<u16>().map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid {} '{}': {}", name, value, err),
)
})
}
fn parse_u32_arg(value: &str, name: &str) -> io::Result<u32> {
value.parse::<u32>().map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid {} '{}': {}", name, value, err),
)
})
}
fn parse_usize_arg(value: &str, name: &str) -> io::Result<usize> {
value.parse::<usize>().map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("invalid {} '{}': {}", name, value, err),
)
})
}

6
src/main.rs
View File

@@ -134,6 +134,12 @@ impl AppConfig {
err
)
})?;
if secs == 0 {
return Err(anyhow!(
"invalid RPKI_RTR_REFRESH_INTERVAL_SECS '{}': must be >= 1",
value
));
}
config.refresh_interval = Duration::from_secs(secs);
}
if let Some(value) = env_var("RPKI_RTR_MAX_CONNECTIONS")? {

2
src/rtr/payload.rs
View File

@@ -292,7 +292,7 @@ impl Timing {
impl Default for Timing {
fn default() -> Self {
Self {
refresh: 3600,
refresh: 60,
retry: 600,
expire: 7200,
}

86
src/rtr/pdu.rs
View File

@@ -11,7 +11,9 @@ use tokio::io::AsyncWrite;
use anyhow::bail;
use serde::Serialize;
use std::slice;
use std::sync::Once;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tracing::debug;
pub const HEADER_LEN: usize = 8;
pub const MAX_PDU_LEN: u32 = 65535;
@@ -22,6 +24,9 @@ pub const END_OF_DATA_V1_LEN: u32 = 24;
pub const ZERO_16: u16 = 0;
pub const ZERO_8: u8 = 0;
static ROUTER_KEY_RESERVED_ZERO_NONZERO_LOG_ONCE: Once = Once::new();
static ASPA_RESERVED_ZERO_NONZERO_LOG_ONCE: Once = Once::new();
macro_rules! common {
( $type:ident ) => {
#[allow(dead_code)]
@@ -1003,6 +1008,15 @@ impl RouterKey {
let asn = Asn::from(u32::from_be_bytes(body[20..24].try_into().unwrap()));
let subject_public_key_info = Arc::<[u8]>::from(body[24..].to_vec());
if header.zero() != 0 {
ROUTER_KEY_RESERVED_ZERO_NONZERO_LOG_ONCE.call_once(|| {
debug!(
"received RouterKey PDU with non-zero reserved field (zero={}); ignoring per protocol compatibility",
header.zero()
);
});
}
let res = Self {
header,
flags: header.flags(),
@@ -1015,10 +1029,25 @@ impl RouterKey {
}
pub async fn write<A: AsyncWrite + Unpin>(&self, w: &mut A) -> Result<(), io::Error> {
self.validate()?;
let length = Self::BASE_LEN + self.subject_public_key_info.len();
let length_u32 = u32::try_from(length).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey PDU length exceeds u32",
)
})?;
if length_u32 > MAX_PDU_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"RouterKey PDU too large: {} octets exceeds max {}",
length_u32, MAX_PDU_LEN
),
));
}
let header =
HeaderWithFlags::new(self.header.version(), Self::PDU, self.flags, length as u32);
let header = HeaderWithFlags::new(self.header.version(), Self::PDU, self.flags, length_u32);
w.write_all(&[
header.version(),
@@ -1028,7 +1057,7 @@ impl RouterKey {
])
.await?;
w.write_all(&(length as u32).to_be_bytes()).await?;
w.write_all(&length_u32.to_be_bytes()).await?;
w.write_all(self.ski.as_ref()).await?;
w.write_all(&self.asn.into_u32().to_be_bytes()).await?;
w.write_all(&self.subject_public_key_info).await?;
@@ -1043,9 +1072,10 @@ impl RouterKey {
subject_public_key_info: Arc<[u8]>,
) -> Self {
let length = Self::BASE_LEN + subject_public_key_info.len();
let wire_length = u32::try_from(length).unwrap_or(u32::MAX);
Self {
header: HeaderWithFlags::new(version, Self::PDU, flags, length as u32),
header: HeaderWithFlags::new(version, Self::PDU, flags, wire_length),
flags,
ski,
asn,
@@ -1072,16 +1102,17 @@ impl RouterKey {
"unexpected PDU type for RouterKey",
));
}
if usize::try_from(self.header.length()).unwrap_or(0) < Self::BASE_LEN {
let total_len = usize::try_from(self.header.length()).unwrap_or(0);
if total_len < Self::BASE_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey PDU shorter than fixed wire size",
));
}
if self.header.zero() != 0 {
if self.header.length() > MAX_PDU_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey reserved zero octet must be zero",
"RouterKey PDU length exceeds MAX_PDU_LEN",
));
}
if self.header.flags().into_u8() & !0x01 != 0 {
@@ -1165,6 +1196,15 @@ impl Aspa {
provider_asns.push(u32::from_be_bytes(chunk.try_into().unwrap()));
}
if header.zero() != 0 {
ASPA_RESERVED_ZERO_NONZERO_LOG_ONCE.call_once(|| {
debug!(
"received ASPA PDU with non-zero reserved field (zero={}); ignoring per protocol compatibility",
header.zero()
);
});
}
let res = Self {
header,
customer_asn,
@@ -1175,13 +1215,26 @@ impl Aspa {
}
pub async fn write<A: AsyncWrite + Unpin>(&self, w: &mut A) -> Result<(), io::Error> {
self.validate()?;
let length = Self::BASE_LEN + (self.provider_asns.len() * 4);
let length_u32 = u32::try_from(length).map_err(|_| {
io::Error::new(io::ErrorKind::InvalidData, "ASPA PDU length exceeds u32")
})?;
if length_u32 > MAX_PDU_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"ASPA PDU too large: {} octets exceeds max {}",
length_u32, MAX_PDU_LEN
),
));
}
let header = HeaderWithFlags::new(
self.header.version(),
Self::PDU,
self.header.flags(),
length as u32,
length_u32,
);
w.write_all(&[
@@ -1192,7 +1245,7 @@ impl Aspa {
])
.await?;
w.write_all(&(length as u32).to_be_bytes()).await?;
w.write_all(&length_u32.to_be_bytes()).await?;
w.write_all(&self.customer_asn.to_be_bytes()).await?;
for asn in &self.provider_asns {
@@ -1203,9 +1256,10 @@ impl Aspa {
}
pub fn new(version: u8, flags: Flags, customer_asn: u32, provider_asns: Vec<u32>) -> Self {
let length = Self::BASE_LEN + (provider_asns.len() * 4);
let wire_length = u32::try_from(length).unwrap_or(u32::MAX);
Self {
header: HeaderWithFlags::new(version, Self::PDU, flags, length as u32),
header: HeaderWithFlags::new(version, Self::PDU, flags, wire_length),
customer_asn,
provider_asns,
}
@@ -1225,18 +1279,18 @@ impl Aspa {
"ASPA PDU shorter than fixed wire size",
));
}
if self.header.length() > MAX_PDU_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA PDU length exceeds MAX_PDU_LEN",
));
}
if (total_len - Self::BASE_LEN) % 4 != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA provider list length must be a multiple of four octets",
));
}
if self.header.zero() != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA reserved zero octet must be zero",
));
}
if self.header.flags().into_u8() & !0x01 != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,

16
src/source/pipeline.rs
View File

@@ -98,13 +98,17 @@ fn apply_slurm_to_payloads_from_dir(
}
fn read_slurm_files(slurm_dir: &str) -> Result<Vec<(String, SlurmFile)>> {
let mut paths = std::fs::read_dir(slurm_dir)
let mut paths = Vec::<PathBuf>::new();
for entry in std::fs::read_dir(slurm_dir)
.map_err(|err| anyhow!("failed to read SLURM directory '{}': {}", slurm_dir, err))?
.filter_map(|entry| entry.ok())
.map(|entry| entry.path())
.filter(|path| path.is_file())
.filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("slurm"))
.collect::<Vec<PathBuf>>();
{
let entry = entry
.map_err(|err| anyhow!("failed to enumerate SLURM directory '{}': {}", slurm_dir, err))?;
let path = entry.path();
if path.is_file() && path.extension().and_then(|ext| ext.to_str()) == Some("slurm") {
paths.push(path);
}
}
paths.sort_by_key(|path| {
path.file_name()