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

完善routerkey和aspa功能

Browse Source
This commit is contained in:
xiuting.xu
2026-03-26 10:04:09 +08:00
parent 03c0ab0ec7
commit 7432fbcac4
20 changed files with 1156 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/bin/rtr_debug_client/README.md
View File

@@ -59,7 +59,7 @@ cargo run --bin rtr_debug_client -- <addr> <version> [reset|serial <session_id>
```
默认值：
- `addr`: `127.0.0.1:3323`
- `addr`: `127.0.0.1:323`
- `version`: `1`
- `mode`: `reset`
- `timeout`: `30`
@@ -70,19 +70,19 @@ cargo run --bin rtr_debug_client -- <addr> <version> [reset|serial <session_id>
发送 `Reset Query`：
```sh
cargo run --bin rtr_debug_client -- 127.0.0.1:3323 1 reset
cargo run --bin rtr_debug_client -- 127.0.0.1:323 1 reset
```
发送 `Serial Query`：
```sh
cargo run --bin rtr_debug_client -- 127.0.0.1:3323 1 serial 42 100
cargo run --bin rtr_debug_client -- 127.0.0.1:323 1 serial 42 100
```
持续观察错误路径：
```sh
cargo run --bin rtr_debug_client -- 127.0.0.1:3323 1 reset --keep-after-error
cargo run --bin rtr_debug_client -- 127.0.0.1:323 1 reset --keep-after-error
```
## TLS 示例
@@ -91,7 +91,7 @@ cargo run --bin rtr_debug_client -- 127.0.0.1:3323 1 reset --keep-after-error
```sh
cargo run --bin rtr_debug_client -- \
127.0.0.1:3324 1 reset \
127.0.0.1:324 1 reset \
--tls \
--ca-cert tests/fixtures/tls/client-ca.crt \
--server-name localhost
@@ -101,7 +101,7 @@ cargo run --bin rtr_debug_client -- \
```sh
cargo run --bin rtr_debug_client -- \
127.0.0.1:3324 1 reset \
127.0.0.1:324 1 reset \
--tls \
--ca-cert tests/fixtures/tls/client-ca.crt \
--server-name localhost \
@@ -113,7 +113,7 @@ cargo run --bin rtr_debug_client -- \
```sh
cargo run --bin rtr_debug_client -- \
127.0.0.1:3324 1 reset \
127.0.0.1:324 1 reset \
--tls \
--ca-cert tests/fixtures/tls/client-ca.crt \
--server-name localhost \

225
src/bin/rtr_debug_client/main.rs
View File

@@ -16,7 +16,7 @@ mod protocol;
use crate::wire::{read_pdu, send_reset_query, send_serial_query};
use crate::pretty::{
parse_end_of_data_info, parse_serial_notify_serial, print_pdu,
parse_end_of_data_info, parse_serial_notify_serial, print_pdu, print_raw_pdu,
};
use crate::protocol::{PduHeader, PduType, QueryMode};
@@ -53,10 +53,6 @@ async fn main() -> io::Result<()> {
println!();
print_help();
let stream = connect_stream(&config).await?;
println!("connected to {}", config.addr);
let (mut reader, mut writer) = tokio_io::split(stream);
let mut state = ClientState::new(
config.version,
config.read_timeout_secs,
@@ -64,85 +60,168 @@ async fn main() -> io::Result<()> {
config.keep_after_error,
);
match config.mode {
QueryMode::Reset => {
send_reset_query(&mut writer, config.version).await?;
println!("sent Reset Query");
}
QueryMode::Serial { session_id, serial } => {
state.session_id = Some(session_id);
state.serial = Some(serial);
send_serial_query(&mut writer, config.version, session_id, serial).await?;
println!("sent Serial Query");
}
}
state.schedule_next_poll();
println!();
let stdin = tokio::io::stdin();
let mut stdin_lines = BufReader::new(stdin).lines();
loop {
let poll_sleep = tokio::time::sleep_until(state.poll_deadline());
tokio::pin!(poll_sleep);
let stream = loop {
match connect_stream(&config).await {
Ok(stream) => {
println!("connected to {}", config.addr);
break stream;
}
Err(err) => {
let delay = state.reconnect_delay_secs();
eprintln!(
"connect failed: {}. retry after {}s",
err,
delay
);
tokio::time::sleep(Duration::from_secs(delay)).await;
}
}
};
tokio::select! {
line = stdin_lines.next_line() => {
match line {
Ok(Some(line)) => {
let should_quit = handle_console_command(
&line,
&mut writer,
&mut state,
).await?;
let (mut reader, mut writer) = tokio_io::split(stream);
send_resume_query(&mut writer, &mut state, &config.mode).await?;
state.schedule_next_poll();
println!();
if should_quit {
println!("quit requested, closing client.");
break;
let reconnect = loop {
let poll_sleep = tokio::time::sleep_until(state.poll_deadline());
tokio::pin!(poll_sleep);
tokio::select! {
line = stdin_lines.next_line() => {
match line {
Ok(Some(line)) => {
match handle_console_command(
&line,
&mut writer,
&mut state,
).await {
Ok(should_quit) => {
if should_quit {
println!("quit requested, closing client.");
return Ok(());
}
}
Err(err) if should_reconnect(&err) => {
eprintln!("command failed due to disconnected transport: {}", err);
break true;
}
Err(err) => return Err(err),
}
}
Ok(None) => {
println!("stdin closed, continue network loop.");
}
Err(err) => {
eprintln!("read stdin failed: {}", err);
}
}
Ok(None) => {
println!("stdin closed, continue network loop.");
}
_ = &mut poll_sleep => {
match handle_poll_tick(&mut writer, &mut state).await {
Ok(()) => state.schedule_next_poll(),
Err(err) if should_reconnect(&err) => {
eprintln!("auto poll failed due to disconnected transport: {}", err);
break true;
}
Err(err) => return Err(err),
}
Err(err) => {
eprintln!("read stdin failed: {}", err);
}
read_result = timeout(
Duration::from_secs(state.read_timeout_secs),
read_pdu(&mut reader)
) => {
match read_result {
Ok(Ok(pdu)) => {
print_raw_pdu(&pdu.header, &pdu.body);
print_pdu(&pdu.header, &pdu.body);
match handle_incoming_pdu(&mut writer, &mut state, &pdu.header, &pdu.body).await {
Ok(()) => {}
Err(err) if should_reconnect(&err) => {
eprintln!("connection dropped while handling incoming PDU: {}", err);
break true;
}
Err(err) => return Err(err),
}
}
Ok(Err(err)) => {
eprintln!("read PDU failed: {}", err);
if should_reconnect(&err) {
break true;
}
return Err(err);
}
Err(_) => {
println!(
"[timeout] no PDU received in {}s, connection kept open.",
state.read_timeout_secs
);
}
}
}
}
};
_ = &mut poll_sleep => {
handle_poll_tick(&mut writer, &mut state).await?;
state.schedule_next_poll();
}
read_result = timeout(
Duration::from_secs(state.read_timeout_secs),
read_pdu(&mut reader)
) => {
match read_result {
Ok(Ok(pdu)) => {
print_pdu(&pdu.header, &pdu.body);
handle_incoming_pdu(&mut writer, &mut state, &pdu.header, &pdu.body).await?;
}
Ok(Err(err)) => {
eprintln!("read PDU failed: {}", err);
return Err(err);
}
Err(_) => {
println!(
"[timeout] no PDU received in {}s, connection kept open.",
state.read_timeout_secs
);
}
}
}
if reconnect {
let delay = state.reconnect_delay_secs();
state.current_session_id = None;
println!(
"[reconnect] transport disconnected, retry after {}s",
delay
);
tokio::time::sleep(Duration::from_secs(delay)).await;
}
}
}
async fn send_resume_query(
writer: &mut ClientWriter,
state: &mut ClientState,
mode: &QueryMode,
) -> io::Result<()> {
match (state.session_id, state.serial) {
(Some(session_id), Some(serial)) => {
println!(
"reconnected, send Serial Query with session_id={}, serial={}",
session_id,
serial
);
send_serial_query(writer, state.version, session_id, serial).await?;
}
_ => match mode {
QueryMode::Reset => {
send_reset_query(writer, state.version).await?;
println!("sent Reset Query");
}
QueryMode::Serial { session_id, serial } => {
state.session_id = Some(*session_id);
state.serial = Some(*serial);
send_serial_query(writer, state.version, *session_id, *serial).await?;
println!("sent Serial Query");
}
},
}
Ok(())
}
fn should_reconnect(err: &io::Error) -> bool {
matches!(
err.kind(),
io::ErrorKind::UnexpectedEof
| io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionReset
| io::ErrorKind::BrokenPipe
| io::ErrorKind::NotConnected
)
}
async fn handle_incoming_pdu(
writer: &mut ClientWriter,
state: &mut ClientState,
@@ -590,6 +669,16 @@ impl ClientState {
"default"
}
}
fn reconnect_delay_secs(&self) -> u64 {
if self.should_prefer_retry_poll() {
self.retry
.map(|v| v as u64)
.unwrap_or(self.default_poll_secs)
} else {
self.default_poll_secs
}
}
}
#[derive(Debug)]
@@ -615,7 +704,9 @@ impl Config {
while let Some(arg) = args.next() {
match arg.as_str() {
"--tls" => {
transport = TransportConfig::Tls(TlsConfig::default());
if matches!(transport, TransportConfig::Tcp) {
transport = TransportConfig::Tls(TlsConfig::default());
}
}
"--ca-cert" => {
let path = args.next().ok_or_else(|| {
@@ -670,7 +761,7 @@ impl Config {
let addr = positional
.next()
.unwrap_or_else(|| "127.0.0.1:3323".to_string());
.unwrap_or_else(|| "127.0.0.1:323".to_string());
let version = positional
.next()

84
src/bin/rtr_debug_client/pretty.rs
View File

@@ -220,20 +220,18 @@ fn print_serial_query(header: &PduHeader, body: &[u8]) {
println!("serial : {}", serial);
}
fn print_router_key(header: &PduHeader, body: &[u8]) {
println!("session_id : {}", header.session_id());
fn print_router_key(_header: &PduHeader, body: &[u8]) {
if body.len() < ROUTER_KEY_FIXED_BODY_LEN {
println!("invalid Router Key body length: {}", body.len());
println!("raw body : {}", hex_bytes(body));
return;
}
let flags = body[0];
let zero = body[1];
let ski = &body[2..22];
let asn = u32::from_be_bytes([body[22], body[23], body[24], body[25]]);
let spki = &body[26..];
let flags = _header.flags();
let zero = _header.zero();
let ski = &body[0..20];
let asn = u32::from_be_bytes([body[20], body[21], body[22], body[23]]);
let spki = &body[24..];
println!("flags : 0x{:02x} ({})", flags, flag_meaning(flags));
println!("zero : {}", zero);
@@ -241,6 +239,32 @@ fn print_router_key(header: &PduHeader, body: &[u8]) {
println!("asn : {}", asn);
println!("spki_len : {}", spki.len());
println!("spki : {}", hex_bytes(spki));
if flags & !0x01 != 0 {
println!("warning : Router Key flags use reserved bits");
}
if zero != 0 {
println!("warning : Router Key reserved zero octet is non-zero");
}
if asn == 0 {
println!("warning : Router Key ASN is AS0");
}
if spki.is_empty() {
println!("warning : Router Key SPKI is empty");
}
}
pub fn print_raw_pdu(header: &PduHeader, body: &[u8]) {
let mut raw = Vec::with_capacity(8 + body.len());
raw.push(header.version);
raw.push(header.pdu_type_raw);
raw.extend_from_slice(&header.field1.to_be_bytes());
raw.extend_from_slice(&header.length.to_be_bytes());
raw.extend_from_slice(body);
println!("--------------------------------------------------");
println!("[raw] pdu_type : {}", header.pdu_type());
println!("[raw] octets : {}", hex_bytes(&raw));
}
fn error_code_name(code: u16) -> &'static str {
@@ -271,26 +295,22 @@ fn parse_encapsulated_header(encapsulated: &[u8]) -> Option<PduHeader> {
Some(PduHeader::from_bytes(header))
}
fn print_aspa(header: &PduHeader, body: &[u8]) {
println!("session_id : {}", header.session_id());
fn print_aspa(_header: &PduHeader, body: &[u8]) {
if body.len() < ASPA_FIXED_BODY_LEN {
println!("invalid ASPA body length: {}", body.len());
println!("raw body : {}", hex_bytes(body));
return;
}
let flags = body[0];
let zero1 = body[1];
let zero2 = body[2];
let zero3 = body[3];
let customer_asn = u32::from_be_bytes([body[4], body[5], body[6], body[7]]);
let flags = _header.flags();
let zero = _header.zero();
let customer_asn = u32::from_be_bytes([body[0], body[1], body[2], body[3]]);
println!("flags : 0x{:02x} ({})", flags, flag_meaning(flags));
println!("reserved : [{}, {}, {}]", zero1, zero2, zero3);
println!("reserved : {}", zero);
println!("customer_as : {}", customer_asn);
let providers_raw = &body[8..];
let providers_raw = &body[4..];
if providers_raw.len() % 4 != 0 {
println!("invalid ASPA providers length: {}", providers_raw.len());
println!("providers : {}", hex_bytes(providers_raw));
@@ -303,6 +323,34 @@ fn print_aspa(header: &PduHeader, body: &[u8]) {
}
println!("providers : {:?}", providers);
if flags & !0x01 != 0 {
println!("warning : ASPA flags use reserved bits");
}
if zero != 0 {
println!("warning : ASPA reserved zero octet is non-zero");
}
if customer_asn == 0 {
println!("warning : ASPA customer ASN is AS0");
}
let is_announcement = flags & 0x01 == 0x01;
if is_announcement {
if providers.is_empty() {
println!("warning : ASPA announcement has empty provider list");
}
} else if !providers.is_empty() {
println!("warning : ASPA withdrawal must not contain providers");
}
if providers.iter().any(|asn| *asn == 0) {
println!("warning : ASPA provider list contains AS0");
}
let strictly_increasing = providers.windows(2).all(|pair| pair[0] < pair[1]);
if !strictly_increasing && providers.len() > 1 {
println!("warning : ASPA providers are not in strictly increasing order");
}
}
pub fn parse_serial_notify_serial(body: &[u8]) -> Option<u32> {

14
src/bin/rtr_debug_client/protocol.rs
View File

@@ -2,14 +2,14 @@ use std::fmt;
pub const HEADER_LEN: usize = 8;
pub const SERIAL_QUERY_LEN: usize = 12;
pub const MAX_PDU_LEN: u32 = 1024 * 1024; // 1 MiB
pub const MAX_PDU_LEN: u32 = 65_535;
pub const IPV4_PREFIX_BODY_LEN: usize = 12;
pub const IPV6_PREFIX_BODY_LEN: usize = 24;
pub const END_OF_DATA_V0_BODY_LEN: usize = 4;
pub const END_OF_DATA_V1_BODY_LEN: usize = 16;
pub const ROUTER_KEY_FIXED_BODY_LEN: usize = 26;
pub const ASPA_FIXED_BODY_LEN: usize = 8;
pub const ROUTER_KEY_FIXED_BODY_LEN: usize = 24;
pub const ASPA_FIXED_BODY_LEN: usize = 4;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueryMode {
@@ -126,6 +126,14 @@ impl PduHeader {
pub fn error_code(&self) -> u16 {
self.field1
}
pub fn flags(&self) -> u8 {
self.field1.to_be_bytes()[0]
}
pub fn zero(&self) -> u8 {
self.field1.to_be_bytes()[1]
}
}
#[derive(Debug, Clone)]

37
src/main.rs
View File

@@ -8,7 +8,7 @@ use tokio::task::JoinHandle;
use tracing::{info, warn};
use rpki::rtr::cache::{RtrCache, SharedRtrCache};
use rpki::rtr::loader::load_vrps_from_file;
use rpki::rtr::loader::{load_aspas_from_file, load_router_keys_from_file, load_vrps_from_file};
use rpki::rtr::payload::Timing;
use rpki::rtr::server::{RtrNotifier, RtrService, RtrServiceConfig, RunningRtrService};
use rpki::rtr::store::RtrStore;
@@ -21,6 +21,8 @@ struct AppConfig {
db_path: String,
vrp_file: String,
aspa_file: String,
router_key_file: String,
tls_cert_path: String,
tls_key_path: String,
tls_client_ca_path: String,
@@ -35,17 +37,19 @@ impl Default for AppConfig {
fn default() -> Self {
Self {
enable_tls: false,
tcp_addr: "0.0.0.0:3323".parse().expect("invalid default tcp_addr"),
tls_addr: "0.0.0.0:3324".parse().expect("invalid default tls_addr"),
tcp_addr: "0.0.0.0:323".parse().expect("invalid default tcp_addr"),
tls_addr: "0.0.0.0:324".parse().expect("invalid default tls_addr"),
db_path: "./rtr-db".to_string(),
vrp_file: r"C:\Users\xuxiu\git_code\rpki\data\vrps.txt".to_string(),
aspa_file: "./data/aspas.txt".to_string(),
router_key_file: "./data/router-keys.txt".to_string(),
tls_cert_path: "./certs/server.crt".to_string(),
tls_key_path: "./certs/server.key".to_string(),
tls_client_ca_path: "./certs/client-ca.crt".to_string(),
max_delta: 100,
refresh_interval: Duration::from_secs(10),
refresh_interval: Duration::from_secs(300),
service_config: RtrServiceConfig {
max_connections: 512,
@@ -81,6 +85,12 @@ impl AppConfig {
if let Some(value) = env_var("RPKI_RTR_VRP_FILE")? {
config.vrp_file = value;
}
if let Some(value) = env_var("RPKI_RTR_ASPA_FILE")? {
config.aspa_file = value;
}
if let Some(value) = env_var("RPKI_RTR_ROUTER_KEY_FILE")? {
config.router_key_file = value;
}
if let Some(value) = env_var("RPKI_RTR_TLS_CERT_PATH")? {
config.tls_cert_path = value;
}
@@ -176,7 +186,7 @@ fn init_shared_cache(config: &AppConfig, store: &RtrStore) -> Result<SharedRtrCa
store,
config.max_delta,
Timing::default(),
|| load_vrps_from_file(&config.vrp_file),
|| load_payloads_from_files(&config.vrp_file, &config.aspa_file, &config.router_key_file),
)?;
let shared_cache: SharedRtrCache = Arc::new(RwLock::new(initial_cache));
@@ -220,6 +230,8 @@ fn spawn_refresh_task(
) -> JoinHandle<()> {
let refresh_interval = config.refresh_interval;
let vrp_file = config.vrp_file.clone();
let aspa_file = config.aspa_file.clone();
let router_key_file = config.router_key_file.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(refresh_interval);
@@ -227,7 +239,7 @@ fn spawn_refresh_task(
loop {
interval.tick().await;
match load_vrps_from_file(&vrp_file) {
match load_payloads_from_files(&vrp_file, &aspa_file, &router_key_file) {
Ok(payloads) => {
let payload_count = payloads.len();
let updated = {
@@ -303,6 +315,8 @@ fn log_startup_config(config: &AppConfig) {
}
info!("vrp_file={}", config.vrp_file);
info!("aspa_file={}", config.aspa_file);
info!("router_key_file={}", config.router_key_file);
info!("max_delta={}", config.max_delta);
info!(
"refresh_interval_secs={}",
@@ -357,3 +371,14 @@ fn parse_bool(value: &str, name: &str) -> Result<bool> {
_ => Err(anyhow!("invalid {} '{}': expected boolean", name, value)),
}
}
fn load_payloads_from_files(
vrp_file: &str,
aspa_file: &str,
router_key_file: &str,
) -> Result<Vec<rpki::rtr::payload::Payload>> {
let mut payloads = load_vrps_from_file(vrp_file)?;
payloads.extend(load_aspas_from_file(aspa_file)?);
payloads.extend(load_router_keys_from_file(router_key_file)?);
Ok(payloads)
}

221
src/rtr/loader.rs
View File

@@ -7,7 +7,7 @@ use anyhow::{anyhow, Context, Result};
use crate::data_model::resources::as_resources::Asn;
use crate::data_model::resources::ip_resources::{IPAddress, IPAddressPrefix};
use crate::rtr::payload::{Payload, RouteOrigin};
use crate::rtr::payload::{Aspa, Payload, RouteOrigin, RouterKey, Ski};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParsedVrp {
@@ -17,6 +17,19 @@ pub struct ParsedVrp {
pub asn: u32,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParsedAspa {
pub customer_asn: u32,
pub provider_asns: Vec<u32>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParsedRouterKey {
pub ski: [u8; 20],
pub asn: u32,
pub spki: Vec<u8>,
}
/// 从文本文件中加载 VRP，并转换成 RTR Payload::RouteOrigin。
///
/// 文件格式：
@@ -52,6 +65,56 @@ pub fn load_vrps_from_file(path: impl AsRef<Path>) -> Result<Vec<Payload>> {
Ok(payloads)
}
pub fn load_aspas_from_file(path: impl AsRef<Path>) -> Result<Vec<Payload>> {
let path = path.as_ref();
let content = fs::read_to_string(path)
.with_context(|| format!("failed to read ASPA file: {}", path.display()))?;
let mut payloads = Vec::new();
for (idx, raw_line) in content.lines().enumerate() {
let line_no = idx + 1;
let line = raw_line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
let aspa = parse_aspa_line(line)
.with_context(|| format!("invalid ASPA line {}: {}", line_no, raw_line))?;
payloads.push(Payload::Aspa(build_aspa(aspa)?));
}
Ok(payloads)
}
pub fn load_router_keys_from_file(path: impl AsRef<Path>) -> Result<Vec<Payload>> {
let path = path.as_ref();
let content = fs::read_to_string(path)
.with_context(|| format!("failed to read Router Key file: {}", path.display()))?;
let mut payloads = Vec::new();
for (idx, raw_line) in content.lines().enumerate() {
let line_no = idx + 1;
let line = raw_line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
let router_key = parse_router_key_line(line)
.with_context(|| format!("invalid Router Key line {}: {}", line_no, raw_line))?;
payloads.push(Payload::RouterKey(build_router_key(router_key)?));
}
Ok(payloads)
}
/// 解析单行 VRP。
///
/// 格式：
@@ -93,6 +156,59 @@ pub fn parse_vrp_line(line: &str) -> Result<ParsedVrp> {
})
}
pub fn parse_aspa_line(line: &str) -> Result<ParsedAspa> {
let parts: Vec<_> = line.split(',').map(|s| s.trim()).collect();
if parts.len() != 2 {
return Err(anyhow!(
"expected format: <customer_asn>,<provider_asn> [provider_asn ...]"
));
}
let customer_asn = u32::from_str(parts[0])
.with_context(|| format!("invalid customer_asn: {}", parts[0]))?;
let provider_asns = parts[1]
.split_whitespace()
.map(|provider| {
u32::from_str(provider)
.with_context(|| format!("invalid provider_asn: {}", provider))
})
.collect::<Result<Vec<_>>>()?;
validate_aspa(customer_asn, &provider_asns)?;
Ok(ParsedAspa {
customer_asn,
provider_asns,
})
}
pub fn parse_router_key_line(line: &str) -> Result<ParsedRouterKey> {
let parts: Vec<_> = line.split(',').map(|s| s.trim()).collect();
if parts.len() != 3 {
return Err(anyhow!(
"expected format: <ski_hex>,<asn>,<spki_hex>"
));
}
let ski_vec = decode_hex(parts[0])
.with_context(|| format!("invalid SKI hex: {}", parts[0]))?;
if ski_vec.len() != 20 {
return Err(anyhow!("SKI must be exactly 20 bytes"));
}
let mut ski = [0u8; 20];
ski.copy_from_slice(&ski_vec);
let asn = u32::from_str(parts[1])
.with_context(|| format!("invalid asn: {}", parts[1]))?;
let spki = decode_hex(parts[2])
.with_context(|| format!("invalid SPKI hex: {}", parts[2]))?;
validate_router_key(asn, &spki)?;
Ok(ParsedRouterKey { ski, asn, spki })
}
fn validate_vrp(prefix_addr: IpAddr, prefix_len: u8, max_len: u8) -> Result<()> {
match prefix_addr {
IpAddr::V4(_) => {
@@ -121,6 +237,33 @@ fn validate_vrp(prefix_addr: IpAddr, prefix_len: u8, max_len: u8) -> Result<()>
Ok(())
}
fn validate_aspa(customer_asn: u32, provider_asns: &[u32]) -> Result<()> {
if customer_asn == 0 {
return Err(anyhow!("customer_asn must not be AS0"));
}
if provider_asns.is_empty() {
return Err(anyhow!("provider list must not be empty"));
}
if provider_asns.iter().any(|asn| *asn == 0) {
return Err(anyhow!("provider list must not contain AS0"));
}
Ok(())
}
fn validate_router_key(asn: u32, spki: &[u8]) -> Result<()> {
crate::rtr::payload::RouterKey::new(
Ski::default(),
Asn::from(asn),
spki.to_vec(),
)
.validate()
.map_err(|err| anyhow!(err.to_string()))?;
Ok(())
}
pub fn build_route_origin(vrp: ParsedVrp) -> Result<RouteOrigin> {
let address = match vrp.prefix_addr {
IpAddr::V4(addr) => IPAddress::from_ipv4(addr),
@@ -133,53 +276,37 @@ pub fn build_route_origin(vrp: ParsedVrp) -> Result<RouteOrigin> {
Ok(RouteOrigin::new(prefix, vrp.max_len, asn))
}
#[cfg(test)]
mod tests {
use super::*;
pub fn build_aspa(aspa: ParsedAspa) -> Result<Aspa> {
let customer_asn = Asn::from(aspa.customer_asn);
let provider_asns = aspa
.provider_asns
.into_iter()
.map(Asn::from)
.collect::<Vec<_>>();
let aspa = Aspa::new(customer_asn, provider_asns);
aspa.validate_announcement()?;
Ok(aspa)
}
#[test]
fn parse_ipv4_vrp_line() {
let got = parse_vrp_line("10.0.0.0/24,24,65001").unwrap();
assert_eq!(
got,
ParsedVrp {
prefix_addr: IpAddr::from_str("10.0.0.0").unwrap(),
prefix_len: 24,
max_len: 24,
asn: 65001,
}
);
pub fn build_router_key(router_key: ParsedRouterKey) -> Result<RouterKey> {
let asn = Asn::from(router_key.asn);
let ski = Ski::from_bytes(router_key.ski);
let router_key = RouterKey::new(ski, asn, router_key.spki);
Ok(router_key)
}
fn decode_hex(input: &str) -> Result<Vec<u8>> {
let trimmed = input.trim();
if trimmed.len() % 2 != 0 {
return Err(anyhow!("hex string must have even length"));
}
#[test]
fn parse_ipv6_vrp_line() {
let got = parse_vrp_line("2001:db8::/32,48,65003").unwrap();
assert_eq!(
got,
ParsedVrp {
prefix_addr: IpAddr::from_str("2001:db8::").unwrap(),
prefix_len: 32,
max_len: 48,
asn: 65003,
}
);
}
(0..trimmed.len())
.step_by(2)
.map(|idx| {
u8::from_str_radix(&trimmed[idx..idx + 2], 16)
.map_err(|err| anyhow!("invalid hex at byte {}: {}", idx / 2, err))
})
.collect()
}
#[test]
fn parse_rejects_invalid_max_len() {
let err = parse_vrp_line("10.0.0.0/24,16,65001").unwrap_err();
assert!(err.to_string().contains("max_len"));
}
#[test]
fn parse_rejects_invalid_ip() {
let err = parse_vrp_line("10.0.0.999/24,24,65001").unwrap_err();
assert!(err.to_string().contains("invalid IP"));
}
#[test]
fn parse_rejects_invalid_format() {
let err = parse_vrp_line("10.0.0.0/24,24").unwrap_err();
assert!(err.to_string().contains("expected format"));
}
}

44
src/rtr/payload.rs
View File

@@ -4,6 +4,8 @@ use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::data_model::resources::as_resources::Asn;
use crate::data_model::resources::ip_resources::IPAddressPrefix;
use x509_parser::prelude::FromDer;
use x509_parser::x509::SubjectPublicKeyInfo;
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
@@ -23,6 +25,12 @@ impl AsRef<[u8]> for Ski {
}
}
impl Ski {
pub fn from_bytes(bytes: [u8; 20]) -> Self {
Self(bytes)
}
}
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
pub struct RouteOrigin {
prefix: IPAddressPrefix,
@@ -80,6 +88,42 @@ impl RouterKey {
pub fn spki(&self) -> &[u8] {
&self.subject_public_key_info
}
pub fn validate(&self) -> Result<(), io::Error> {
if self.asn.into_u32() == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey ASN must not be AS0",
));
}
if self.subject_public_key_info.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey SPKI must not be empty",
));
}
let (rem, _) = SubjectPublicKeyInfo::from_der(&self.subject_public_key_info)
.map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("RouterKey SPKI is not valid DER: {err}"),
)
})?;
if !rem.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"RouterKey SPKI DER has trailing bytes: {}",
rem.len()
),
));
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]

218
src/rtr/pdu.rs
View File

@@ -333,6 +333,8 @@ impl HeaderWithFlags {
pub fn flags(self) -> Flags{Flags(self.flags)}
pub fn zero(self) -> u8 { self.zero }
pub fn length(self) -> u32{u32::from_be(self.length)}
}
@@ -968,6 +970,58 @@ impl RouterKey {
pub const PDU: u8 = 9;
const BASE_LEN: usize = HEADER_LEN + 20 + 4;
pub async fn read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
) -> Result<Self, io::Error> {
let header = HeaderWithFlags::read(sock)
.await
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))?;
if header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"PDU type mismatch when expecting RouterKey",
));
}
Self::read_payload(header, sock).await
}
pub async fn read_payload<Sock: AsyncRead + Unpin>(
header: HeaderWithFlags,
sock: &mut Sock,
) -> Result<Self, io::Error> {
let total_len = usize::try_from(header.length()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey PDU too large for this system to handle",
)
})?;
if total_len < Self::BASE_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid length for RouterKey PDU",
));
}
let body_len = total_len - HEADER_LEN;
let mut body = vec![0u8; body_len];
sock.read_exact(&mut body).await?;
let mut ski = [0u8; 20];
ski.copy_from_slice(&body[..20]);
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());
let res = Self {
header,
flags: header.flags(),
ski: Ski::from_bytes(ski),
asn,
subject_public_key_info,
};
res.validate()?;
Ok(res)
}
pub async fn write<A: AsyncWrite + Unpin>(
&self,
w: &mut A,
@@ -1024,6 +1078,46 @@ impl RouterKey {
pub fn spki(&self) -> &[u8] {
&self.subject_public_key_info
}
fn validate(&self) -> Result<(), io::Error> {
if self.header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"unexpected PDU type for RouterKey",
));
}
if usize::try_from(self.header.length()).unwrap_or(0) < Self::BASE_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey PDU shorter than fixed wire size",
));
}
if self.header.zero() != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey reserved zero octet must be zero",
));
}
if self.header.flags().into_u8() & !0x01 != 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey flags use reserved bits",
));
}
if self.asn.into_u32() == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey ASN must not be AS0",
));
}
if self.subject_public_key_info.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"RouterKey SPKI must not be empty",
));
}
Ok(())
}
}
@@ -1041,6 +1135,63 @@ impl Aspa {
pub const PDU: u8 = 11;
const BASE_LEN: usize = HEADER_LEN + 4;
pub async fn read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
) -> Result<Self, io::Error> {
let header = HeaderWithFlags::read(sock)
.await
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))?;
if header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"PDU type mismatch when expecting ASPA",
));
}
Self::read_payload(header, sock).await
}
pub async fn read_payload<Sock: AsyncRead + Unpin>(
header: HeaderWithFlags,
sock: &mut Sock,
) -> Result<Self, io::Error> {
let total_len = usize::try_from(header.length()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
"ASPA PDU too large for this system to handle",
)
})?;
if total_len < Self::BASE_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid length for ASPA PDU",
));
}
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",
));
}
let body_len = total_len - HEADER_LEN;
let mut body = vec![0u8; body_len];
sock.read_exact(&mut body).await?;
let customer_asn = u32::from_be_bytes(body[..4].try_into().unwrap());
let mut provider_asns = Vec::with_capacity((body.len() - 4) / 4);
for chunk in body[4..].chunks_exact(4) {
provider_asns.push(u32::from_be_bytes(chunk.try_into().unwrap()));
}
let res = Self {
header,
customer_asn,
provider_asns,
};
res.validate()?;
Ok(res)
}
pub async fn write<A: AsyncWrite + Unpin>(
&self,
w: &mut A,
@@ -1086,6 +1237,73 @@ impl Aspa {
}
}
fn validate(&self) -> Result<(), io::Error> {
if self.header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"unexpected PDU type for ASPA",
));
}
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,
"ASPA PDU shorter than fixed wire size",
));
}
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,
"ASPA flags use reserved bits",
));
}
if self.customer_asn == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA customer ASN must not be AS0",
));
}
let is_announcement = self.header.flags().is_announce();
if is_announcement && self.provider_asns.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA announcement must contain at least one provider ASN",
));
}
if !is_announcement && !self.provider_asns.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA withdrawal must not contain provider ASNs",
));
}
if self.provider_asns.iter().any(|asn| *asn == 0) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA provider list must not contain AS0",
));
}
if self.provider_asns.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ASPA provider ASNs must be strictly increasing",
));
}
Ok(())
}
}

6
src/rtr/server/listener.rs
View File

@@ -98,9 +98,6 @@ impl RtrServer {
}
};
if let Err(err) = stream.set_nodelay(true) {
warn!("failed to enable TCP_NODELAY for {}: {}", peer_addr, err);
}
if let Err(err) = apply_keepalive(&stream, self.config.tcp_keepalive) {
warn!("failed to configure TCP keepalive for {}: {}", peer_addr, err);
}
@@ -209,9 +206,6 @@ impl RtrServer {
}
};
if let Err(err) = stream.set_nodelay(true) {
warn!("failed to enable TCP_NODELAY for {}: {}", peer_addr, err);
}
if let Err(err) = apply_keepalive(&stream, self.config.tcp_keepalive) {
warn!("failed to configure TCP keepalive for {}: {}", peer_addr, err);
}

21
src/rtr/session.rs
View File

@@ -3,7 +3,7 @@ use std::time::{Duration, Instant};
use anyhow::{anyhow, bail, Result};
use tokio::io;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{broadcast, watch};
use tokio::time::timeout;
use tracing::{debug, error, info, warn};
@@ -77,6 +77,12 @@ where
}
pub async fn run(mut self) -> Result<()> {
let result = self.run_inner().await;
self.close_transport().await;
result
}
async fn run_inner(&mut self) -> Result<()> {
info!(
"RTR session started: {}",
self.session_summary()
@@ -251,6 +257,18 @@ where
}
}
async fn close_transport(&mut self) {
if let Err(err) = self.stream.shutdown().await {
debug!(
"RTR session transport shutdown returned error: err={}, {}",
err,
self.session_summary()
);
} else {
debug!("RTR session transport shutdown completed: {}", self.session_summary());
}
}
async fn negotiate_version(&mut self, router_version: u8) -> io::Result<u8> {
if let Some(current) = self.version {
if current == router_version {
@@ -958,6 +976,7 @@ where
async fn send_router_key(&mut self, key: &RouterKey, announce: bool) -> Result<()> {
let version = self.version()?;
key.validate()?;
let flags = Flags::new(if announce {
ANNOUNCE_FLAG