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

新增slurm功能

Browse Source
This commit is contained in:
xiuting.xu
2026-04-01 16:24:01 +08:00
parent 8d6173f605
commit cd0330e8ae
49 changed files with 2738 additions and 1052 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
src/rtr/cache/core.rs vendored
View File

@@ -1,14 +1,14 @@
use std::collections::{BTreeMap, VecDeque};
use std::cmp::Ordering;
use std::sync::Arc;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::collections::{BTreeMap, VecDeque};
use std::sync::Arc;
use tracing::{debug, info, warn};
use crate::rtr::payload::{Payload, Timing};
use super::model::{Delta, DualTime, Snapshot};
use super::ordering::{change_key, ChangeKey};
use super::ordering::{ChangeKey, change_key};
const SERIAL_HALF_RANGE: u32 = 1 << 31;
@@ -166,9 +166,7 @@ impl RtrCacheBuilder {
let serial = self.serial.unwrap_or(0);
let created_at = self.created_at.unwrap_or_else(|| now.clone());
let availability = self.availability.unwrap_or(CacheAvailability::Ready);
let session_ids = self
.session_ids
.unwrap_or_else(SessionIds::random_distinct);
let session_ids = self.session_ids.unwrap_or_else(SessionIds::random_distinct);
RtrCache {
availability,
@@ -235,8 +233,7 @@ impl RtrCache {
self.serial = self.serial.wrapping_add(1);
debug!(
"RTR cache advanced serial: old_serial={}, new_serial={}",
old,
self.serial
old, self.serial
);
self.serial
}
@@ -251,9 +248,7 @@ impl RtrCache {
let snapshot_wire_size = estimate_snapshot_payload_wire_size(&self.snapshot);
let mut cumulative_delta_wire_size =
estimate_delta_window_payload_wire_size(&self.deltas);
while !self.deltas.is_empty()
&& cumulative_delta_wire_size >= snapshot_wire_size
{
while !self.deltas.is_empty() && cumulative_delta_wire_size >= snapshot_wire_size {
if let Some(oldest) = self.deltas.pop_front() {
dropped_serials.push(oldest.serial());
cumulative_delta_wire_size =
@@ -262,9 +257,7 @@ impl RtrCache {
}
debug!(
"RTR cache delta-size pruning evaluated: snapshot_wire_size={}, cumulative_delta_wire_size={}, dropped_serials={:?}",
snapshot_wire_size,
cumulative_delta_wire_size,
dropped_serials
snapshot_wire_size, cumulative_delta_wire_size, dropped_serials
);
}
debug!(
@@ -292,7 +285,10 @@ impl RtrCache {
}
}
pub(super) fn apply_update(&mut self, new_payloads: Vec<Payload>) -> Result<Option<AppliedUpdate>> {
pub(super) fn apply_update(
&mut self,
new_payloads: Vec<Payload>,
) -> Result<Option<AppliedUpdate>> {
self.last_update_begin = DualTime::now();
info!(
"RTR cache applying update: availability={:?}, current_serial={}, incoming_payloads={}",
@@ -319,14 +315,15 @@ impl RtrCache {
self.last_update_end = DualTime::now();
if !changed {
debug!("RTR cache update produced empty snapshot but cache was already unavailable; no state change");
debug!(
"RTR cache update produced empty snapshot but cache was already unavailable; no state change"
);
return Ok(None);
}
info!(
"RTR cache update cleared usable data and marked cache unavailable: serial={}, session_ids={:?}",
self.serial,
self.session_ids
self.serial, self.session_ids
);
return Ok(Some(AppliedUpdate {
@@ -349,8 +346,7 @@ impl RtrCache {
self.last_update_end = DualTime::now();
debug!(
"RTR cache update detected identical snapshot content: serial={}, session_ids={:?}",
self.serial,
self.session_ids
self.serial, self.session_ids
);
return Ok(None);
}
@@ -455,8 +451,7 @@ impl RtrCache {
if client_serial == self.serial {
debug!(
"RTR cache delta query is already up to date: client_serial={}, cache_serial={}",
client_serial,
self.serial
client_serial, self.serial
);
return SerialResult::UpToDate;
}
@@ -467,8 +462,7 @@ impl RtrCache {
) {
warn!(
"RTR cache delta query requires reset due to invalid/newer client serial: client_serial={}, cache_serial={}",
client_serial,
self.serial
client_serial, self.serial
);
return SerialResult::ResetRequired;
}
@@ -489,8 +483,7 @@ impl RtrCache {
if deltas.is_empty() {
debug!(
"RTR cache delta query resolved to no deltas: client_serial={}, cache_serial={}",
client_serial,
self.serial
client_serial, self.serial
);
return SerialResult::UpToDate;
}
@@ -633,7 +626,11 @@ fn estimate_payload_wire_size(payload: &Payload, announce: bool) -> usize {
},
Payload::RouterKey(key) => 8 + 20 + 4 + key.spki().len(),
Payload::Aspa(aspa) => {
let providers = if announce { aspa.provider_asns().len() } else { 0 };
let providers = if announce {
aspa.provider_asns().len()
} else {
0
};
8 + 4 + providers * 4
}
}

16
src/rtr/cache/model.rs vendored
View File

@@ -195,7 +195,12 @@ impl Snapshot {
}
if !self.same_aspas(new_snapshot) {
diff_aspas(&self.aspas, &new_snapshot.aspas, &mut announced, &mut withdrawn);
diff_aspas(
&self.aspas,
&new_snapshot.aspas,
&mut announced,
&mut withdrawn,
);
}
(announced, withdrawn)
@@ -206,9 +211,8 @@ impl Snapshot {
}
pub fn payloads(&self) -> Vec<Payload> {
let mut v = Vec::with_capacity(
self.origins.len() + self.router_keys.len() + self.aspas.len(),
);
let mut v =
Vec::with_capacity(self.origins.len() + self.router_keys.len() + self.aspas.len());
v.extend(self.origins.iter().cloned().map(Payload::RouteOrigin));
v.extend(self.router_keys.iter().cloned().map(Payload::RouterKey));
@@ -268,9 +272,7 @@ impl Snapshot {
}
pub fn is_empty(&self) -> bool {
self.origins.is_empty()
&& self.router_keys.is_empty()
&& self.aspas.is_empty()
self.origins.is_empty() && self.router_keys.is_empty() && self.aspas.is_empty()
}
}

20
src/rtr/cache/ordering.rs vendored
View File

@@ -64,8 +64,18 @@ enum PayloadPduType {
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub(crate) enum RouteOriginKey {
V4 { addr: u32, plen: u8, mlen: u8, asn: u32 },
V6 { addr: u128, plen: u8, mlen: u8, asn: u32 },
V4 {
addr: u32,
plen: u8,
mlen: u8,
asn: u32,
},
V6 {
addr: u128,
plen: u8,
mlen: u8,
asn: u32,
},
}
pub(crate) fn change_key(payload: &Payload) -> ChangeKey {
@@ -287,7 +297,11 @@ fn payload_brief(payload: &Payload) -> String {
match payload {
Payload::RouteOrigin(origin) => format!(
"{} prefix {:?}/{} max={} asn={}",
if route_origin_is_ipv4(origin) { "IPv4" } else { "IPv6" },
if route_origin_is_ipv4(origin) {
"IPv4"
} else {
"IPv6"
},
origin.prefix().address,
origin.prefix().prefix_length,
origin.max_length(),

9
src/rtr/cache/store.rs vendored
View File

@@ -18,12 +18,9 @@ impl RtrCache {
timing: Timing,
file_loader: impl Fn() -> Result<Vec<Payload>>,
) -> Result<Self> {
if let Some(cache) = try_restore_from_store(
store,
max_delta,
prune_delta_by_snapshot_size,
timing,
)? {
if let Some(cache) =
try_restore_from_store(store, max_delta, prune_delta_by_snapshot_size, timing)?
{
tracing::info!(
"RTR cache restored from store: availability={:?}, session_ids={:?}, serial={}, snapshot(route_origins={}, router_keys={}, aspas={})",
cache.availability(),

361
src/rtr/ccr.rs
View File

@@ -1,361 +0,0 @@
use std::fs;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::path::{Path, PathBuf};
use anyhow::{anyhow, Context, Result};
use der_parser::ber::{BerObject, BerObjectContent};
use der_parser::der::parse_der;
use crate::rtr::loader::{ParsedAspa, ParsedVrp, build_aspa, build_route_origin};
use crate::rtr::payload::Payload;
const VRPS_INDEX: usize = 3;
const VAPS_INDEX: usize = 4;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ParsedCcrSnapshot {
pub content_type_oid: String,
pub produced_at: Option<String>,
pub vrps: Vec<ParsedVrp>,
pub vaps: Vec<ParsedAspa>,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct CcrPayloadConversion {
pub payloads: Vec<Payload>,
pub invalid_vrps: Vec<String>,
pub invalid_vaps: Vec<String>,
}
pub fn load_ccr_snapshot_from_file(path: impl AsRef<Path>) -> Result<ParsedCcrSnapshot> {
let path = path.as_ref();
let bytes = fs::read(path)
.with_context(|| format!("failed to read CCR file: {}", path.display()))?;
parse_ccr_bytes(&bytes).with_context(|| format!("failed to parse CCR file: {}", path.display()))
}
pub fn load_ccr_payloads_from_file(path: impl AsRef<Path>) -> Result<Vec<Payload>> {
let snapshot = load_ccr_snapshot_from_file(path)?;
snapshot_to_payloads(&snapshot)
}
pub fn load_ccr_payloads_from_file_with_options(
path: impl AsRef<Path>,
strict: bool,
) -> Result<CcrPayloadConversion> {
let snapshot = load_ccr_snapshot_from_file(path)?;
snapshot_to_payloads_with_options(&snapshot, strict)
}
pub fn find_latest_ccr_file(dir: impl AsRef<Path>) -> Result<PathBuf> {
let dir = dir.as_ref();
let mut latest: Option<PathBuf> = None;
for entry in fs::read_dir(dir)
.with_context(|| format!("failed to read CCR directory: {}", dir.display()))?
{
let entry =
entry.with_context(|| format!("failed to iterate CCR directory: {}", dir.display()))?;
let path = entry.path();
if !path.is_file() {
continue;
}
if path.extension().and_then(|ext| ext.to_str()) != Some("ccr") {
continue;
}
if latest.as_ref().is_none_or(|current| {
file_name_key(&path) > file_name_key(current)
}) {
latest = Some(path);
}
}
latest.ok_or_else(|| anyhow!("no .ccr files found in {}", dir.display()))
}
pub fn snapshot_to_payloads(snapshot: &ParsedCcrSnapshot) -> Result<Vec<Payload>> {
Ok(snapshot_to_payloads_with_options(snapshot, true)?.payloads)
}
pub fn snapshot_to_payloads_with_options(
snapshot: &ParsedCcrSnapshot,
strict: bool,
) -> Result<CcrPayloadConversion> {
let mut payloads = Vec::with_capacity(snapshot.vrps.len() + snapshot.vaps.len());
let mut invalid_vrps = Vec::new();
let mut invalid_vaps = Vec::new();
for vrp in &snapshot.vrps {
match build_route_origin(vrp.clone()) {
Ok(origin) => payloads.push(Payload::RouteOrigin(origin)),
Err(err) => {
let msg = format!("invalid CCR VRP: {:?}: {}", vrp, err);
if strict {
return Err(anyhow!(msg));
}
invalid_vrps.push(msg);
}
}
}
for vap in &snapshot.vaps {
match build_aspa(vap.clone()) {
Ok(aspa) => payloads.push(Payload::Aspa(aspa)),
Err(err) => {
let msg = format!("invalid CCR VAP/ASPA: {:?}: {}", vap, err);
if strict {
return Err(anyhow!(msg));
}
invalid_vaps.push(msg);
}
}
}
Ok(CcrPayloadConversion {
payloads,
invalid_vrps,
invalid_vaps,
})
}
pub fn parse_ccr_bytes(bytes: &[u8]) -> Result<ParsedCcrSnapshot> {
let (rem, root) = parse_der(bytes).map_err(|err| anyhow!("failed to parse CCR DER: {err}"))?;
if !rem.is_empty() {
return Err(anyhow!("CCR DER has {} trailing bytes", rem.len()));
}
let root_items = sequence_items(&root)?;
if root_items.len() != 2 {
return Err(anyhow!("CCR root must contain exactly 2 items"));
}
let content_type_oid = match &root_items[0].content {
BerObjectContent::OID(oid) => oid.to_string(),
other => {
return Err(anyhow!(
"CCR root first element must be content type OID, got {other:?}"
));
}
};
let payload = decode_context_wrapped_sequence(&root_items[1])?;
let payload_items = sequence_items(&payload)?;
let produced_at = payload_items.get(1).and_then(|obj| match &obj.content {
BerObjectContent::GeneralizedTime(t) => Some(t.to_string()),
_ => None,
});
let vrps = if let Some(vrps_field) = payload_items.get(VRPS_INDEX) {
parse_vrps(vrps_field)?
} else {
Vec::new()
};
let vaps = if let Some(vaps_field) = payload_items.get(VAPS_INDEX) {
parse_vaps(vaps_field)?
} else {
Vec::new()
};
Ok(ParsedCcrSnapshot {
content_type_oid,
produced_at,
vrps,
vaps,
})
}
fn parse_vrps(field: &BerObject<'_>) -> Result<Vec<ParsedVrp>> {
let vrp_state = decode_context_wrapped_sequence(field)?;
let vrp_state_items = sequence_items(&vrp_state)?;
let roa_payload_sets = vrp_state_items
.first()
.ok_or_else(|| anyhow!("ROA payload state missing payload set list"))?;
let roa_payload_sets = sequence_items(roa_payload_sets)?;
let mut vrps = Vec::new();
for payload_set in roa_payload_sets {
let payload_set_items = sequence_items(payload_set)?;
if payload_set_items.len() != 2 {
return Err(anyhow!(
"ROAPayloadSet must contain 2 items, got {}",
payload_set_items.len()
));
}
let asn = as_u32(&payload_set_items[0], "ROAPayloadSet.asID")?;
let families = sequence_items(&payload_set_items[1])?;
for family in families {
let family_items = sequence_items(family)?;
if family_items.len() != 2 {
return Err(anyhow!(
"ROAIPAddressFamily must contain 2 items, got {}",
family_items.len()
));
}
let address_family = as_octets(&family_items[0], "ROAIPAddressFamily.addressFamily")?;
let addresses = sequence_items(&family_items[1])?;
for address in addresses {
let address_items = sequence_items(address)?;
let (prefix_addr, prefix_len, max_len) =
parse_roa_address(address_family, address_items)?;
vrps.push(ParsedVrp {
prefix_addr,
prefix_len,
max_len,
asn,
});
}
}
}
Ok(vrps)
}
fn parse_vaps(field: &BerObject<'_>) -> Result<Vec<ParsedAspa>> {
let vap_state = decode_context_wrapped_sequence(field)?;
let vap_state_items = sequence_items(&vap_state)?;
let aspa_payload_sets = vap_state_items
.first()
.ok_or_else(|| anyhow!("ASPA payload state missing payload set list"))?;
let aspa_payload_sets = sequence_items(aspa_payload_sets)?;
let mut vaps = Vec::new();
for payload_set in aspa_payload_sets {
let payload_set_items = sequence_items(payload_set)?;
if payload_set_items.len() != 2 {
return Err(anyhow!(
"ASPAPayloadSet must contain 2 items, got {}",
payload_set_items.len()
));
}
let customer_asn = as_u32(&payload_set_items[0], "ASPAPayloadSet.customerASID")?;
let provider_set = sequence_items(&payload_set_items[1])?;
let mut provider_asns = Vec::with_capacity(provider_set.len());
for provider in provider_set {
provider_asns.push(as_u32(provider, "ASPAPayloadSet.providerASID")?);
}
vaps.push(ParsedAspa {
customer_asn,
provider_asns,
});
}
Ok(vaps)
}
fn parse_roa_address(
address_family: &[u8],
items: &[BerObject<'_>],
) -> Result<(IpAddr, u8, u8)> {
let address = items
.first()
.ok_or_else(|| anyhow!("ROAIPAddress missing address field"))?;
let (unused_bits, bit_string) = match &address.content {
BerObjectContent::BitString(unused_bits, bit_string) => (*unused_bits, bit_string),
other => {
return Err(anyhow!(
"ROAIPAddress.address must be BIT STRING, got {other:?}"
));
}
};
let prefix_len = (bit_string.data.len() * 8)
.checked_sub(usize::from(unused_bits))
.ok_or_else(|| anyhow!("invalid ROAIPAddress BIT STRING length"))?;
let prefix_len = u8::try_from(prefix_len)
.map_err(|_| anyhow!("prefix length {prefix_len} does not fit in u8"))?;
let max_len = match items.get(1) {
Some(value) => {
let max_len = as_u32(value, "ROAIPAddress.maxLength")?;
u8::try_from(max_len)
.map_err(|_| anyhow!("maxLength {max_len} does not fit in u8"))?
}
None => prefix_len,
};
let prefix_addr = match address_family {
[0, 1] | [0, 1, ..] => {
let mut octets = [0u8; 4];
if bit_string.data.len() > octets.len() {
return Err(anyhow!(
"IPv4 ROAIPAddress too long: {} bytes",
bit_string.data.len()
));
}
octets[..bit_string.data.len()].copy_from_slice(bit_string.data);
IpAddr::V4(Ipv4Addr::from(octets))
}
[0, 2] | [0, 2, ..] => {
let mut octets = [0u8; 16];
if bit_string.data.len() > octets.len() {
return Err(anyhow!(
"IPv6 ROAIPAddress too long: {} bytes",
bit_string.data.len()
));
}
octets[..bit_string.data.len()].copy_from_slice(bit_string.data);
IpAddr::V6(Ipv6Addr::from(octets))
}
_ => {
return Err(anyhow!(
"unsupported ROA address family octets: {:?}",
address_family
));
}
};
Ok((prefix_addr, prefix_len, max_len))
}
fn sequence_items<'a>(obj: &'a BerObject<'a>) -> Result<&'a [BerObject<'a>]> {
match &obj.content {
BerObjectContent::Sequence(items) => Ok(items),
other => Err(anyhow!("expected SEQUENCE, got {other:?}")),
}
}
fn decode_context_wrapped_sequence<'a>(obj: &'a BerObject<'a>) -> Result<BerObject<'a>> {
match &obj.content {
BerObjectContent::Unknown(any) => {
let (rem, inner) = parse_der(any.data)
.map_err(|err| anyhow!("failed to parse encapsulated DER: {err}"))?;
if !rem.is_empty() {
return Err(anyhow!(
"encapsulated DER has {} trailing bytes",
rem.len()
));
}
Ok(inner)
}
other => Err(anyhow!(
"expected context-specific wrapped field, got {other:?}"
)),
}
}
fn as_u32(obj: &BerObject<'_>, field_name: &str) -> Result<u32> {
obj.as_u32()
.map_err(|err| anyhow!("{field_name} must be INTEGER fitting in u32: {err}"))
}
fn as_octets<'a>(obj: &'a BerObject<'a>, field_name: &str) -> Result<&'a [u8]> {
match &obj.content {
BerObjectContent::OctetString(bytes) => Ok(bytes),
other => Err(anyhow!("{field_name} must be OCTET STRING, got {other:?}")),
}
}
fn file_name_key(path: &Path) -> String {
path.file_name()
.and_then(|name| name.to_str())
.map(|name| name.to_ascii_lowercase())
.unwrap_or_default()
}

42
src/rtr/error_type.rs
View File

@@ -19,7 +19,6 @@ pub enum ErrorCode {
}
impl ErrorCode {
#[inline]
pub fn as_u16(self) -> u16 {
self as u16
@@ -27,41 +26,29 @@ impl ErrorCode {
pub fn description(self) -> &'static str {
match self {
ErrorCode::CorruptData =>
"Corrupt Data",
ErrorCode::CorruptData => "Corrupt Data",
ErrorCode::InternalError =>
"Internal Error",
ErrorCode::InternalError => "Internal Error",
ErrorCode::NoDataAvailable =>
"No Data Available",
ErrorCode::NoDataAvailable => "No Data Available",
ErrorCode::InvalidRequest =>
"Invalid Request",
ErrorCode::InvalidRequest => "Invalid Request",
ErrorCode::UnsupportedProtocolVersion =>
"Unsupported Protocol Version",
ErrorCode::UnsupportedProtocolVersion => "Unsupported Protocol Version",
ErrorCode::UnsupportedPduType =>
"Unsupported PDU Type",
ErrorCode::UnsupportedPduType => "Unsupported PDU Type",
ErrorCode::WithdrawalOfUnknownRecord =>
"Withdrawal of Unknown Record",
ErrorCode::WithdrawalOfUnknownRecord => "Withdrawal of Unknown Record",
ErrorCode::DuplicateAnnouncement =>
"Duplicate Announcement Received",
ErrorCode::DuplicateAnnouncement => "Duplicate Announcement Received",
ErrorCode::UnexpectedProtocolVersion =>
"Unexpected Protocol Version",
ErrorCode::UnexpectedProtocolVersion => "Unexpected Protocol Version",
ErrorCode::AspaProviderListError =>
"ASPA Provider List Error",
ErrorCode::AspaProviderListError => "ASPA Provider List Error",
ErrorCode::TransportFailed =>
"Transport Failed",
ErrorCode::TransportFailed => "Transport Failed",
ErrorCode::OrderingError =>
"Ordering Error",
ErrorCode::OrderingError => "Ordering Error",
}
}
}
@@ -90,9 +77,6 @@ impl TryFrom<u16> for ErrorCode {
impl fmt::Display for ErrorCode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{} ({})",
self.description(),
*self as u16
)
write!(f, "{} ({})", self.description(), *self as u16)
}
}

40
src/rtr/loader.rs
View File

@@ -3,7 +3,7 @@ use std::net::IpAddr;
use std::path::Path;
use std::str::FromStr;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use crate::data_model::resources::as_resources::Asn;
use crate::data_model::resources::ip_resources::{IPAddress, IPAddressPrefix};
@@ -131,10 +131,9 @@ pub fn parse_vrp_line(line: &str) -> Result<ParsedVrp> {
}
let prefix_part = parts[0];
let max_len = u8::from_str(parts[1])
.with_context(|| format!("invalid max_len: {}", parts[1]))?;
let asn = u32::from_str(parts[2])
.with_context(|| format!("invalid asn: {}", parts[2]))?;
let max_len =
u8::from_str(parts[1]).with_context(|| format!("invalid max_len: {}", parts[1]))?;
let asn = u32::from_str(parts[2]).with_context(|| format!("invalid asn: {}", parts[2]))?;
let (addr_str, prefix_len_str) = prefix_part
.split_once('/')
@@ -164,14 +163,13 @@ pub fn parse_aspa_line(line: &str) -> Result<ParsedAspa> {
));
}
let customer_asn = u32::from_str(parts[0])
.with_context(|| format!("invalid customer_asn: {}", parts[0]))?;
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))
u32::from_str(provider).with_context(|| format!("invalid provider_asn: {}", provider))
})
.collect::<Result<Vec<_>>>()?;
@@ -186,23 +184,18 @@ pub fn parse_aspa_line(line: &str) -> Result<ParsedAspa> {
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>"
));
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]))?;
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]))?;
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)?;
@@ -254,13 +247,9 @@ fn validate_aspa(customer_asn: u32, provider_asns: &[u32]) -> Result<()> {
}
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()))?;
crate::rtr::payload::RouterKey::new(Ski::default(), Asn::from(asn), spki.to_vec())
.validate()
.map_err(|err| anyhow!(err.to_string()))?;
Ok(())
}
@@ -309,4 +298,3 @@ fn decode_hex(input: &str) -> Result<Vec<u8>> {
})
.collect()
}

13
src/rtr/mod.rs
View File

@@ -1,10 +1,9 @@
pub mod pdu;
pub mod cache;
pub mod payload;
pub mod store;
pub mod session;
pub mod error_type;
pub mod state;
pub mod server;
pub mod loader;
pub mod ccr;
pub mod payload;
pub mod pdu;
pub mod server;
pub mod session;
pub mod state;
pub mod store;

44
src/rtr/payload.rs
View File

@@ -1,13 +1,12 @@
use crate::data_model::resources::as_resources::Asn;
use crate::data_model::resources::ip_resources::IPAddressPrefix;
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use std::io;
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)]
enum PayloadPduType {
Ipv4Prefix = 4,
@@ -16,7 +15,9 @@ enum PayloadPduType {
Aspa = 11,
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[derive(
Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Ord, PartialOrd, Serialize, Deserialize,
)]
pub struct Ski([u8; 20]);
impl AsRef<[u8]> for Ski {
@@ -60,7 +61,6 @@ impl RouteOrigin {
}
}
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
pub struct RouterKey {
subject_key_identifier: Ski,
@@ -104,8 +104,8 @@ impl RouterKey {
));
}
let (rem, _) = SubjectPublicKeyInfo::from_der(&self.subject_public_key_info)
.map_err(|err| {
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}"),
@@ -115,10 +115,7 @@ impl RouterKey {
if !rem.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"RouterKey SPKI DER has trailing bytes: {}",
rem.len()
),
format!("RouterKey SPKI DER has trailing bytes: {}", rem.len()),
));
}
@@ -177,7 +174,6 @@ impl Aspa {
}
}
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
pub enum Payload {
@@ -191,7 +187,6 @@ pub enum Payload {
Aspa(Aspa),
}
// Timing
#[derive(Clone, Copy, Debug)]
pub struct Timing {
@@ -202,7 +197,7 @@ pub struct Timing {
pub retry: u32,
/// The number of secionds before data expires if not refreshed.
pub expire: u32
pub expire: u32,
}
impl Timing {
@@ -214,7 +209,11 @@ impl Timing {
pub const MAX_EXPIRE: u32 = 172_800;
pub const fn new(refresh: u32, retry: u32, expire: u32) -> Self {
Self { refresh, retry, expire }
Self {
refresh,
retry,
expire,
}
}
pub fn validate(self) -> Result<(), io::Error> {
@@ -223,7 +222,9 @@ impl Timing {
io::ErrorKind::InvalidData,
format!(
"refresh interval {} out of range {}..={}",
self.refresh, Self::MIN_REFRESH, Self::MAX_REFRESH
self.refresh,
Self::MIN_REFRESH,
Self::MAX_REFRESH
),
));
}
@@ -233,7 +234,9 @@ impl Timing {
io::ErrorKind::InvalidData,
format!(
"retry interval {} out of range {}..={}",
self.retry, Self::MIN_RETRY, Self::MAX_RETRY
self.retry,
Self::MIN_RETRY,
Self::MAX_RETRY
),
));
}
@@ -243,7 +246,9 @@ impl Timing {
io::ErrorKind::InvalidData,
format!(
"expire interval {} out of range {}..={}",
self.expire, Self::MIN_EXPIRE, Self::MAX_EXPIRE
self.expire,
Self::MIN_EXPIRE,
Self::MAX_EXPIRE
),
));
}
@@ -282,7 +287,6 @@ impl Timing {
pub fn expire(self) -> Duration {
Duration::from_secs(u64::from(self.expire))
}
}
impl Default for Timing {

340
src/rtr/pdu.rs
View File

@@ -1,16 +1,16 @@
use std::{cmp, mem};
use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use crate::data_model::resources::as_resources::Asn;
use crate::rtr::error_type::ErrorCode;
use crate::rtr::payload::{Ski, Timing};
use std::io;
use tokio::io::{AsyncWrite};
use anyhow::Result;
use std::io;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::{cmp, mem};
use tokio::io::AsyncWrite;
use std::slice;
use anyhow::bail;
use serde::Serialize;
use std::slice;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
pub const HEADER_LEN: usize = 8;
@@ -27,10 +27,7 @@ macro_rules! common {
#[allow(dead_code)]
impl $type {
/// Writes a value to a writer.
pub async fn write<A: AsyncWrite + Unpin>(
&self,
a: &mut A
) -> Result<(), io::Error> {
pub async fn write<A: AsyncWrite + Unpin>(&self, a: &mut A) -> Result<(), io::Error> {
a.write_all(self.as_ref()).await
}
}
@@ -38,10 +35,7 @@ macro_rules! common {
impl AsRef<[u8]> for $type {
fn as_ref(&self) -> &[u8] {
unsafe {
slice::from_raw_parts(
self as *const Self as *const u8,
mem::size_of::<Self>()
)
slice::from_raw_parts(self as *const Self as *const u8, mem::size_of::<Self>())
}
}
}
@@ -49,14 +43,11 @@ macro_rules! common {
impl AsMut<[u8]> for $type {
fn as_mut(&mut self) -> &mut [u8] {
unsafe {
slice::from_raw_parts_mut(
self as *mut Self as *mut u8,
mem::size_of::<Self>()
)
slice::from_raw_parts_mut(self as *mut Self as *mut u8, mem::size_of::<Self>())
}
}
}
}
};
}
macro_rules! concrete {
@@ -94,28 +85,20 @@ macro_rules! concrete {
///
/// If a value with a different PDU type is received, returns an
/// error.
pub async fn read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
) -> Result<Self, io::Error> {
pub async fn read<Sock: AsyncRead + Unpin>(sock: &mut Sock) -> Result<Self, io::Error> {
let mut res = Self::default();
sock.read_exact(res.header.as_mut()).await?;
if res.header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
concat!(
"PDU type mismatch when expecting ",
stringify!($type)
)
))
concat!("PDU type mismatch when expecting ", stringify!($type)),
));
}
if res.header.length() as usize != res.as_ref().len() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
concat!(
"invalid length for ",
stringify!($type)
)
))
concat!("invalid length for ", stringify!($type)),
));
}
sock.read_exact(&mut res.as_mut()[Header::LEN..]).await?;
Ok(res)
@@ -126,32 +109,26 @@ macro_rules! concrete {
/// If a different PDU type is received, returns the header as
/// the error case of the ok case.
pub async fn try_read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
sock: &mut Sock,
) -> Result<Result<Self, Header>, io::Error> {
let mut res = Self::default();
sock.read_exact(res.header.as_mut()).await?;
if res.header.pdu() == ErrorReport::PDU {
// Since we should drop the session after an error, we
// can safely ignore all the rest of the error for now.
return Ok(Err(res.header))
return Ok(Err(res.header));
}
if res.header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
concat!(
"PDU type mismatch when expecting ",
stringify!($type)
)
))
concat!("PDU type mismatch when expecting ", stringify!($type)),
));
}
if res.header.length() as usize != res.as_ref().len() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
concat!(
"invalid length for ",
stringify!($type)
)
))
concat!("invalid length for ", stringify!($type)),
));
}
sock.read_exact(&mut res.as_mut()[Header::LEN..]).await?;
Ok(Ok(res))
@@ -163,17 +140,14 @@ macro_rules! concrete {
/// `header`, the function reads the rest of the PUD from the
/// reader and returns the complete value.
pub async fn read_payload<Sock: AsyncRead + Unpin>(
header: Header, sock: &mut Sock
header: Header,
sock: &mut Sock,
) -> Result<Self, io::Error> {
if header.length() as usize != mem::size_of::<Self>() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
concat!(
"invalid length for ",
stringify!($type),
" PDU"
)
))
concat!("invalid length for ", stringify!($type), " PDU"),
));
}
let mut res = Self::default();
sock.read_exact(&mut res.as_mut()[Header::LEN..]).await?;
@@ -181,10 +155,9 @@ macro_rules! concrete {
Ok(res)
}
}
}
};
}
// 所有PDU公共头部信息
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
@@ -196,7 +169,6 @@ pub struct Header {
}
impl Header {
const LEN: usize = mem::size_of::<Self>();
pub fn new(version: u8, pdu: u8, session: u16, length: u32) -> Self {
Header {
@@ -208,7 +180,7 @@ impl Header {
}
pub async fn read_raw<S: AsyncRead + Unpin>(
sock: &mut S
sock: &mut S,
) -> Result<[u8; HEADER_LEN], io::Error> {
let mut buf = [0u8; HEADER_LEN];
sock.read_exact(&mut buf).await?;
@@ -229,10 +201,7 @@ impl Header {
}
if length > MAX_PDU_LEN {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"PDU too large",
));
return Err(io::Error::new(io::ErrorKind::InvalidData, "PDU too large"));
}
Ok(Self {
@@ -247,13 +216,21 @@ impl Header {
Self::from_raw(Self::read_raw(sock).await?)
}
pub fn version(self) -> u8{self.version}
pub fn version(self) -> u8 {
self.version
}
pub fn pdu(self) -> u8{self.pdu}
pub fn pdu(self) -> u8 {
self.pdu
}
pub fn session_id(self) -> u16{u16::from_be(self.session_id)}
pub fn session_id(self) -> u16 {
u16::from_be(self.session_id)
}
pub fn length(self) -> u32{u32::from_be(self.length)}
pub fn length(self) -> u32 {
u32::from_be(self.length)
}
pub fn pdu_len(self) -> Result<usize, io::Error> {
usize::try_from(self.length()).map_err(|_| {
@@ -268,7 +245,6 @@ impl Header {
debug_assert_eq!(self.pdu(), ErrorReport::PDU);
self.session_id()
}
}
common!(Header);
@@ -304,12 +280,10 @@ impl HeaderWithFlags {
let pdu = buf[1];
let flags = buf[2];
let zero = buf[3];
let length = u32::from_be_bytes([
buf[4], buf[5], buf[6], buf[7],
]);
let length = u32::from_be_bytes([buf[4], buf[5], buf[6], buf[7]]);
// 3. 基础合法性校验
if length < HEADER_LEN as u32{
if length < HEADER_LEN as u32 {
bail!("Invalid PDU length");
}
@@ -327,18 +301,27 @@ impl HeaderWithFlags {
})
}
pub fn version(self) -> u8{self.version}
pub fn version(self) -> u8 {
self.version
}
pub fn pdu(self) -> u8{self.pdu}
pub fn pdu(self) -> u8 {
self.pdu
}
pub fn flags(self) -> Flags{Flags(self.flags)}
pub fn flags(self) -> Flags {
Flags(self.flags)
}
pub fn zero(self) -> u8 { self.zero }
pub fn zero(self) -> u8 {
self.zero
}
pub fn length(self) -> u32{u32::from_be(self.length)}
pub fn length(self) -> u32 {
u32::from_be(self.length)
}
}
// Serial Notify
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
@@ -351,20 +334,18 @@ impl SerialNotify {
pub const PDU: u8 = 0;
pub fn new(version: u8, session_id: u16, serial_number: u32) -> Self {
SerialNotify{
SerialNotify {
header: Header::new(version, Self::PDU, session_id, Self::size()),
serial_number: serial_number.to_be()
serial_number: serial_number.to_be(),
}
}
pub fn serial_number(self) -> u32 {
u32::from_be(self.serial_number)
}
}
concrete!(SerialNotify);
// Serial Query
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
@@ -377,9 +358,9 @@ impl SerialQuery {
pub const PDU: u8 = 1;
pub fn new(version: u8, session_id: u16, serial_number: u32) -> Self {
SerialQuery{
SerialQuery {
header: Header::new(version, Self::PDU, session_id, Self::size()),
serial_number: serial_number.to_be()
serial_number: serial_number.to_be(),
}
}
@@ -390,12 +371,11 @@ impl SerialQuery {
concrete!(SerialQuery);
// Reset Query
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
pub struct ResetQuery {
header: Header
header: Header,
}
impl ResetQuery {
@@ -410,7 +390,6 @@ impl ResetQuery {
concrete!(ResetQuery);
// Cache Response
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
@@ -430,7 +409,6 @@ impl CacheResponse {
concrete!(CacheResponse);
// Flags
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
pub struct Flags(u8);
@@ -464,7 +442,7 @@ pub struct IPv4Prefix {
max_len: u8,
zero: u8,
prefix: u32,
asn: u32
asn: u32,
}
impl IPv4Prefix {
@@ -475,7 +453,7 @@ impl IPv4Prefix {
prefix_len: u8,
max_len: u8,
prefix: Ipv4Addr,
asn: Asn
asn: Asn,
) -> Self {
IPv4Prefix {
header: Header::new(version, Self::PDU, ZERO_16, IPV4_PREFIX_LEN),
@@ -488,12 +466,22 @@ impl IPv4Prefix {
}
}
pub fn flag(self) -> Flags{self.flags}
pub fn flag(self) -> Flags {
self.flags
}
pub fn prefix_len(self) -> u8{self.prefix_len}
pub fn max_len(self) -> u8{self.max_len}
pub fn prefix(self) -> Ipv4Addr{u32::from_be(self.prefix).into()}
pub fn asn(self) -> Asn{u32::from_be(self.asn).into()}
pub fn prefix_len(self) -> u8 {
self.prefix_len
}
pub fn max_len(self) -> u8 {
self.max_len
}
pub fn prefix(self) -> Ipv4Addr {
u32::from_be(self.prefix).into()
}
pub fn asn(self) -> Asn {
u32::from_be(self.asn).into()
}
}
concrete!(IPv4Prefix);
@@ -509,7 +497,7 @@ pub struct IPv6Prefix {
max_len: u8,
zero: u8,
prefix: u128,
asn: u32
asn: u32,
}
impl IPv6Prefix {
@@ -520,7 +508,7 @@ impl IPv6Prefix {
prefix_len: u8,
max_len: u8,
prefix: Ipv6Addr,
asn: Asn
asn: Asn,
) -> Self {
IPv6Prefix {
header: Header::new(version, Self::PDU, ZERO_16, IPV6_PREFIX_LEN),
@@ -533,17 +521,26 @@ impl IPv6Prefix {
}
}
pub fn flag(self) -> Flags{self.flags}
pub fn flag(self) -> Flags {
self.flags
}
pub fn prefix_len(self) -> u8{self.prefix_len}
pub fn max_len(self) -> u8{self.max_len}
pub fn prefix(self) -> Ipv6Addr{u128::from_be(self.prefix).into()}
pub fn asn(self) -> Asn{u32::from_be(self.asn).into()}
pub fn prefix_len(self) -> u8 {
self.prefix_len
}
pub fn max_len(self) -> u8 {
self.max_len
}
pub fn prefix(self) -> Ipv6Addr {
u128::from_be(self.prefix).into()
}
pub fn asn(self) -> Asn {
u32::from_be(self.asn).into()
}
}
concrete!(IPv6Prefix);
// End of Data
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize)]
pub enum EndOfData {
@@ -559,14 +556,20 @@ impl EndOfData {
timing: Timing,
) -> Result<Self, io::Error> {
if version == 0 {
Ok(EndOfData::V0(EndOfDataV0::new(version, session_id, serial_number)))
}
else {
Ok(EndOfData::V1(EndOfDataV1::new(version, session_id, serial_number, timing)?))
Ok(EndOfData::V0(EndOfDataV0::new(
version,
session_id,
serial_number,
)))
} else {
Ok(EndOfData::V1(EndOfDataV1::new(
version,
session_id,
serial_number,
timing,
)?))
}
}
}
#[repr(C, packed)]
@@ -587,11 +590,12 @@ impl EndOfDataV0 {
}
}
pub fn serial_number(self) -> u32{u32::from_be(self.serial_number)}
pub fn serial_number(self) -> u32 {
u32::from_be(self.serial_number)
}
}
concrete!(EndOfDataV0);
#[repr(C, packed)]
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq, Serialize)]
pub struct EndOfDataV1 {
@@ -601,7 +605,6 @@ pub struct EndOfDataV1 {
refresh_interval: u32,
retry_interval: u32,
expire_interval: u32,
}
impl EndOfDataV1 {
@@ -640,9 +643,11 @@ impl EndOfDataV1 {
})
}
pub fn serial_number(self) -> u32{u32::from_be(self.serial_number)}
pub fn serial_number(self) -> u32 {
u32::from_be(self.serial_number)
}
pub fn timing(self) -> Timing{
pub fn timing(self) -> Timing {
Timing {
refresh: u32::from_be(self.refresh_interval),
retry: u32::from_be(self.retry_interval),
@@ -654,22 +659,20 @@ impl EndOfDataV1 {
self.timing().validate()
}
pub async fn read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
) -> Result<Self, io::Error> {
pub async fn read<Sock: AsyncRead + Unpin>(sock: &mut Sock) -> Result<Self, io::Error> {
let mut res = Self::default();
sock.read_exact(res.header.as_mut()).await?;
if res.header.pdu() != Self::PDU {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"PDU type mismatch when expecting EndOfDataV1",
))
));
}
if res.header.length() as usize != mem::size_of::<Self>() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid length for EndOfDataV1",
))
));
}
sock.read_exact(&mut res.as_mut()[Header::LEN..]).await?;
res.validate()?;
@@ -677,13 +680,14 @@ impl EndOfDataV1 {
}
pub async fn read_payload<Sock: AsyncRead + Unpin>(
header: Header, sock: &mut Sock
header: Header,
sock: &mut Sock,
) -> Result<Self, io::Error> {
if header.length() as usize != mem::size_of::<Self>() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid length for EndOfDataV1 PDU",
))
));
}
let mut res = Self::default();
sock.read_exact(&mut res.as_mut()[Header::LEN..]).await?;
@@ -704,23 +708,21 @@ pub struct CacheReset {
impl CacheReset {
pub const PDU: u8 = 8;
pub fn new(version: u8) -> Self{
pub fn new(version: u8) -> Self {
CacheReset {
header: Header::new(version, Self::PDU, ZERO_16, HEADER_LEN as u32)
header: Header::new(version, Self::PDU, ZERO_16, HEADER_LEN as u32),
}
}
}
concrete!(CacheReset);
// Error Report
#[derive(Clone, Debug, Default, Eq, Hash, PartialEq, Serialize)]
pub struct ErrorReport {
octets: Vec<u8>,
}
impl ErrorReport {
/// The PDU type of an error PDU.
pub const PDU: u8 = 10;
@@ -741,27 +743,19 @@ impl ErrorReport {
let text_len = cmp::min(text.len(), text_room);
let size = Self::FIXED_PART_LEN + pdu_len + text_len;
let header = Header::new(
version, 10, error_code, u32::try_from(size).unwrap()
);
let header = Header::new(version, 10, error_code, u32::try_from(size).unwrap());
let mut octets = Vec::with_capacity(size);
octets.extend_from_slice(header.as_ref());
octets.extend_from_slice(
u32::try_from(pdu_len).unwrap().to_be_bytes().as_ref()
);
octets.extend_from_slice(u32::try_from(pdu_len).unwrap().to_be_bytes().as_ref());
octets.extend_from_slice(&pdu[..pdu_len]);
octets.extend_from_slice(
u32::try_from(text_len).unwrap().to_be_bytes().as_ref()
);
octets.extend_from_slice(u32::try_from(text_len).unwrap().to_be_bytes().as_ref());
octets.extend_from_slice(&text[..text_len]);
ErrorReport { octets }
}
pub async fn read<Sock: AsyncRead + Unpin>(
sock: &mut Sock
) -> Result<Self, io::Error> {
pub async fn read<Sock: AsyncRead + Unpin>(sock: &mut Sock) -> Result<Self, io::Error> {
let header = Header::read(sock).await?;
if header.pdu() != Self::PDU {
return Err(io::Error::new(
@@ -787,7 +781,8 @@ impl ErrorReport {
let mut octets = Vec::with_capacity(total_len);
octets.extend_from_slice(header.as_ref());
octets.resize(total_len, 0);
sock.read_exact(&mut octets[mem::size_of::<Header>()..]).await?;
sock.read_exact(&mut octets[mem::size_of::<Header>()..])
.await?;
let res = ErrorReport { octets };
res.validate()?;
@@ -813,7 +808,8 @@ impl ErrorReport {
/// Skips over the payload of the error PDU.
pub async fn skip_payload<Sock: AsyncRead + Unpin>(
header: Header, sock: &mut Sock
header: Header,
sock: &mut Sock,
) -> Result<(), io::Error> {
let Some(mut remaining) = header.pdu_len()?.checked_sub(mem::size_of::<Header>()) else {
return Err(io::Error::new(
@@ -840,9 +836,7 @@ impl ErrorReport {
}
/// Writes the PUD to a writer.
pub async fn write<A: AsyncWrite + Unpin>(
&self, a: &mut A
) -> Result<(), io::Error> {
pub async fn write<A: AsyncWrite + Unpin>(&self, a: &mut A) -> Result<(), io::Error> {
a.write_all(self.as_ref()).await
}
@@ -860,7 +854,7 @@ impl ErrorReport {
u32::from_be_bytes(
self.octets[Header::LEN..Header::LEN + 4]
.try_into()
.unwrap()
.unwrap(),
) as usize
}
@@ -876,11 +870,7 @@ impl ErrorReport {
fn text_len(&self) -> usize {
let offset = self.text_len_offset();
u32::from_be_bytes(
self.octets[offset..offset + 4]
.try_into()
.unwrap()
) as usize
u32::from_be_bytes(self.octets[offset..offset + 4].try_into().unwrap()) as usize
}
fn text_range(&self) -> std::ops::Range<usize> {
@@ -916,7 +906,10 @@ impl ErrorReport {
let pdu_len = self.erroneous_pdu_len();
let text_len_offset = Header::LEN + 4 + pdu_len;
let Some(text_len_end) = text_len_offset.checked_add(4) else {
return Err(io::Error::new(io::ErrorKind::InvalidData, "ErrorReport length overflow"));
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ErrorReport length overflow",
));
};
if text_len_end > self.octets.len() {
return Err(io::Error::new(
@@ -928,10 +921,13 @@ impl ErrorReport {
let text_len = u32::from_be_bytes(
self.octets[text_len_offset..text_len_end]
.try_into()
.unwrap()
.unwrap(),
) as usize;
let Some(text_end) = text_len_end.checked_add(text_len) else {
return Err(io::Error::new(io::ErrorKind::InvalidData, "ErrorReport text overflow"));
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"ErrorReport text overflow",
));
};
if text_end != self.octets.len() {
return Err(io::Error::new(
@@ -951,7 +947,6 @@ impl ErrorReport {
}
}
// TODO: 补全
/// Router Key
#[derive(Clone, Debug, Default, Eq, Hash, PartialEq, Serialize)]
@@ -966,13 +961,10 @@ pub struct RouterKey {
}
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> {
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()))?;
@@ -1022,25 +1014,19 @@ impl RouterKey {
Ok(res)
}
pub async fn write<A: AsyncWrite + Unpin>(
&self,
w: &mut A,
) -> Result<(), io::Error> {
pub async fn write<A: AsyncWrite + Unpin>(&self, w: &mut A) -> Result<(), io::Error> {
let length = Self::BASE_LEN + self.subject_public_key_info.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 as u32);
w.write_all(&[
header.version(),
header.pdu(),
header.flags().into_u8(),
ZERO_8,
]).await?;
])
.await?;
w.write_all(&(length as u32).to_be_bytes()).await?;
w.write_all(self.ski.as_ref()).await?;
@@ -1120,24 +1106,20 @@ impl RouterKey {
}
}
// ASPA
#[derive(Clone, Debug, Default, Eq, Hash, PartialEq, Serialize)]
pub struct Aspa{
pub struct Aspa {
header: HeaderWithFlags,
customer_asn: u32,
provider_asns: Vec<u32>
provider_asns: Vec<u32>,
}
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> {
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()))?;
@@ -1192,11 +1174,7 @@ impl Aspa {
Ok(res)
}
pub async fn write<A: AsyncWrite + Unpin>(
&self,
w: &mut A,
) -> Result<(), io::Error> {
pub async fn write<A: AsyncWrite + Unpin>(&self, w: &mut A) -> Result<(), io::Error> {
let length = Self::BASE_LEN + (self.provider_asns.len() * 4);
let header = HeaderWithFlags::new(
@@ -1211,7 +1189,8 @@ impl Aspa {
header.pdu(),
header.flags().into_u8(),
ZERO_8,
]).await?;
])
.await?;
w.write_all(&(length as u32).to_be_bytes()).await?;
w.write_all(&self.customer_asn.to_be_bytes()).await?;
@@ -1222,12 +1201,7 @@ impl Aspa {
Ok(())
}
pub fn new(
version: u8,
flags: Flags,
customer_asn: u32,
provider_asns: Vec<u32>,
) -> Self {
pub fn new(version: u8, flags: Flags, customer_asn: u32, provider_asns: Vec<u32>) -> Self {
let length = Self::BASE_LEN + (provider_asns.len() * 4);
Self {
@@ -1306,7 +1280,6 @@ impl Aspa {
}
}
//--- AsRef and AsMut
impl AsRef<[u8]> for ErrorReport {
fn as_ref(&self) -> &[u8] {
@@ -1319,4 +1292,3 @@ impl AsMut<[u8]> for ErrorReport {
self.octets.as_mut()
}
}

35
src/rtr/server/connection.rs
View File

@@ -6,7 +6,7 @@ use std::sync::{
use anyhow::{Context, Result, anyhow};
use tokio::net::TcpStream;
use tokio::sync::{broadcast, watch, OwnedSemaphorePermit};
use tokio::sync::{OwnedSemaphorePermit, broadcast, watch};
use tracing::{info, warn};
use x509_parser::extensions::GeneralName;
use x509_parser::prelude::{FromDer, X509Certificate};
@@ -22,10 +22,7 @@ pub struct ConnectionGuard {
}
impl ConnectionGuard {
pub fn new(
active_connections: Arc<AtomicUsize>,
permit: OwnedSemaphorePermit,
) -> Self {
pub fn new(active_connections: Arc<AtomicUsize>, permit: OwnedSemaphorePermit) -> Self {
active_connections.fetch_add(1, Ordering::Relaxed);
Self {
active_connections,
@@ -72,8 +69,12 @@ pub async fn handle_tls_connection(
.await
.with_context(|| format!("TLS handshake failed for {}", peer_addr))?;
info!("RTR TLS handshake completed for {}", peer_addr);
verify_peer_certificate_ip(&tls_stream, peer_addr.ip())
.with_context(|| format!("TLS client certificate SAN IP validation failed for {}", peer_addr))?;
verify_peer_certificate_ip(&tls_stream, peer_addr.ip()).with_context(|| {
format!(
"TLS client certificate SAN IP validation failed for {}",
peer_addr
)
})?;
info!("RTR TLS client certificate validated for {}", peer_addr);
let session = RtrSession::new(cache, tls_stream, notify_rx, shutdown_rx);
@@ -122,16 +123,16 @@ fn verify_peer_certificate_ip(
GeneralName::IPAddress(bytes) => {
let bytes = *bytes;
match (peer_ip, bytes.len()) {
(IpAddr::V4(ip), 4) => <[u8; 4]>::try_from(bytes)
.map(IpAddr::from)
.map(|cert_ip| cert_ip == IpAddr::V4(ip))
.unwrap_or(false),
(IpAddr::V6(ip), 16) => <[u8; 16]>::try_from(bytes)
.map(IpAddr::from)
.map(|cert_ip| cert_ip == IpAddr::V6(ip))
.unwrap_or(false),
_ => false,
}
(IpAddr::V4(ip), 4) => <[u8; 4]>::try_from(bytes)
.map(IpAddr::from)
.map(|cert_ip| cert_ip == IpAddr::V4(ip))
.unwrap_or(false),
(IpAddr::V6(ip), 16) => <[u8; 16]>::try_from(bytes)
.map(IpAddr::from)
.map(|cert_ip| cert_ip == IpAddr::V6(ip))
.unwrap_or(false),
_ => false,
}
}
_ => false,
});

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

@@ -1,28 +1,22 @@
use std::net::SocketAddr;
use std::path::Path;
use std::sync::{
Arc,
atomic::AtomicUsize,
};
use std::sync::{Arc, atomic::AtomicUsize};
use std::time::Duration;
use anyhow::{Context, Result};
use socket2::{SockRef, TcpKeepalive};
use tokio::net::TcpListener;
use tokio::sync::{broadcast, watch, Semaphore};
use tokio::sync::{Semaphore, broadcast, watch};
use tracing::{info, warn};
use rustls::ServerConfig;
use tokio_rustls::TlsAcceptor;
use crate::rtr::cache::SharedRtrCache;
use crate::rtr::server::connection::{
ConnectionGuard,
handle_tcp_connection,
handle_tls_connection,
is_expected_disconnect,
};
use crate::rtr::server::config::RtrServiceConfig;
use crate::rtr::server::connection::{
ConnectionGuard, handle_tcp_connection, handle_tls_connection, is_expected_disconnect,
};
use crate::rtr::server::tls::load_rustls_server_config_with_options;
pub struct RtrServer {
@@ -65,7 +59,8 @@ impl RtrServer {
}
pub fn active_connections(&self) -> usize {
self.active_connections.load(std::sync::atomic::Ordering::Relaxed)
self.active_connections
.load(std::sync::atomic::Ordering::Relaxed)
}
pub async fn run_tcp(self) -> Result<()> {
@@ -293,10 +288,7 @@ impl RtrServer {
}
}
fn apply_keepalive(
stream: &tokio::net::TcpStream,
keepalive: Option<Duration>,
) -> Result<()> {
fn apply_keepalive(stream: &tokio::net::TcpStream, keepalive: Option<Duration>) -> Result<()> {
let Some(keepalive) = keepalive else {
return Ok(());
};

2
src/rtr/server/mod.rs
View File

@@ -9,4 +9,4 @@ pub use config::RtrServiceConfig;
pub use listener::RtrServer;
pub use notifier::RtrNotifier;
pub use service::{RtrService, RunningRtrService};
pub use tls::load_rustls_server_config;
pub use tls::load_rustls_server_config;

2
src/rtr/server/notifier.rs
View File

@@ -13,4 +13,4 @@ impl RtrNotifier {
pub fn notify_cache_updated(&self) {
let _ = self.tx.send(());
}
}
}

10
src/rtr/server/service.rs
View File

@@ -5,7 +5,7 @@ use std::sync::{
atomic::{AtomicUsize, Ordering},
};
use tokio::sync::{broadcast, watch, Semaphore};
use tokio::sync::{Semaphore, broadcast, watch};
use tokio::task::JoinHandle;
use tracing::{error, warn};
@@ -114,7 +114,10 @@ impl RtrService {
let server = self.tls_server(bind_addr);
tokio::spawn(async move {
if let Err(err) = server.run_tls_from_pem(cert_path, key_path, client_ca_path).await {
if let Err(err) = server
.run_tls_from_pem(cert_path, key_path, client_ca_path)
.await
{
error!("RTR TLS server {} exited with error: {:?}", bind_addr, err);
}
})
@@ -129,7 +132,8 @@ impl RtrService {
client_ca_path: impl AsRef<Path>,
) -> RunningRtrService {
let tcp_handle = self.spawn_tcp(tcp_bind_addr);
let tls_handle = self.spawn_tls_from_pem(tls_bind_addr, cert_path, key_path, client_ca_path);
let tls_handle =
self.spawn_tls_from_pem(tls_bind_addr, cert_path, key_path, client_ca_path);
RunningRtrService {
shutdown_tx: self.shutdown_tx.clone(),

13
src/rtr/server/tls.rs
View File

@@ -3,7 +3,7 @@ use std::io::BufReader;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use rustls::server::WebPkiClientVerifier;
use rustls::{RootCertStore, ServerConfig};
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
@@ -36,8 +36,12 @@ pub fn load_rustls_server_config_with_options(
let key = load_private_key(&key_path)
.with_context(|| format!("failed to load private key from {}", key_path.display()))?;
let client_ca_certs = load_certs(&client_ca_path)
.with_context(|| format!("failed to load client CA certs from {}", client_ca_path.display()))?;
let client_ca_certs = load_certs(&client_ca_path).with_context(|| {
format!(
"failed to load client CA certs from {}",
client_ca_path.display()
)
})?;
let mut client_roots = RootCertStore::empty();
let (added, _) = client_roots.add_parsable_certificates(client_ca_certs);
if added == 0 {
@@ -100,8 +104,7 @@ fn load_certs(path: &Path) -> Result<Vec<CertificateDer<'static>>> {
let file = File::open(path)?;
let mut reader = BufReader::new(file);
let certs = rustls_pemfile::certs(&mut reader)
.collect::<std::result::Result<Vec<_>, _>>()?;
let certs = rustls_pemfile::certs(&mut reader).collect::<std::result::Result<Vec<_>, _>>()?;
if certs.is_empty() {
return Err(anyhow!("no certificates found in {}", path.display()));

124
src/rtr/session.rs
View File

@@ -1,7 +1,7 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{anyhow, bail, Result};
use anyhow::{Result, anyhow, bail};
use tokio::io;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{broadcast, watch};
@@ -14,13 +14,11 @@ use crate::rtr::cache::{
validate_payloads_for_rtr,
};
use crate::rtr::error_type::ErrorCode;
use crate::rtr::pdu::{
Aspa as AspaPdu,
CacheReset, CacheResponse, EndOfData, ErrorReport, Flags, Header, IPv4Prefix, IPv6Prefix,
ResetQuery, RouterKey as RouterKeyPdu, SerialNotify, SerialQuery,
HEADER_LEN,
};
use crate::rtr::payload::{Aspa, Payload, RouteOrigin, RouterKey};
use crate::rtr::pdu::{
Aspa as AspaPdu, CacheReset, CacheResponse, EndOfData, ErrorReport, Flags, HEADER_LEN, Header,
IPv4Prefix, IPv6Prefix, ResetQuery, RouterKey as RouterKeyPdu, SerialNotify, SerialQuery,
};
const SUPPORTED_MAX_VERSION: u8 = 2;
const SUPPORTED_MIN_VERSION: u8 = 0;
@@ -83,10 +81,7 @@ where
}
async fn run_inner(&mut self) -> Result<()> {
info!(
"RTR session started: {}",
self.session_summary()
);
info!("RTR session started: {}", self.session_summary());
loop {
let transport_timeout = self.transport_timeout();
tokio::select! {
@@ -265,7 +260,10 @@ where
self.session_summary()
);
} else {
debug!("RTR session transport shutdown completed: {}", self.session_summary());
debug!(
"RTR session transport shutdown completed: {}",
self.session_summary()
);
}
}
@@ -351,7 +349,7 @@ where
offending_pdu,
msg.as_bytes(),
)
.await
.await
}
async fn send_unexpected_version(
@@ -362,8 +360,7 @@ where
) -> io::Result<()> {
let msg = format!(
"unexpected protocol version {}, established version is {}",
received_version,
established_version
received_version, established_version
);
self.send_error(
@@ -372,7 +369,7 @@ where
offending_pdu,
msg.as_bytes(),
)
.await
.await
}
async fn send_corrupt_session_id(
@@ -393,7 +390,7 @@ where
offending_pdu,
msg.as_bytes(),
)
.await
.await
}
async fn send_corrupt_data(
@@ -402,13 +399,8 @@ where
offending_pdu: &[u8],
detail: &[u8],
) -> io::Result<()> {
self.send_error(
version,
ErrorCode::CorruptData,
offending_pdu,
detail,
)
.await
self.send_error(version, ErrorCode::CorruptData, offending_pdu, detail)
.await
}
async fn send_no_data_available(
@@ -458,7 +450,8 @@ where
self.state = SessionState::Closed;
bail!(
"router version {} higher than cache max {}",
version, SUPPORTED_MAX_VERSION
version,
SUPPORTED_MAX_VERSION
);
}
self.negotiate_version(version).await?;
@@ -495,14 +488,16 @@ where
self.state = SessionState::Closed;
bail!(
"router version {} higher than cache max {}",
version, SUPPORTED_MAX_VERSION
version,
SUPPORTED_MAX_VERSION
);
}
self.negotiate_version(version).await?;
let session_id = query.session_id();
let serial = query.serial_number();
self.handle_serial(version, session_id, serial, query.as_ref()).await?;
self.handle_serial(version, session_id, serial, query.as_ref())
.await?;
self.state = SessionState::Established;
info!(
"RTR session established after Serial Query: negotiated_version={}, client_session_id={}, client_serial={}, {}",
@@ -613,7 +608,10 @@ where
.cache
.read()
.map_err(|_| anyhow!("cache read lock poisoned"))?;
(cache.is_data_available(), cache.session_id_for_version(version))
(
cache.is_data_available(),
cache.session_id_for_version(version),
)
};
if !data_available {
@@ -723,7 +721,10 @@ where
let now = Instant::now();
if let Some(last) = self.last_notify_at {
if now.duration_since(last) < NOTIFY_MIN_INTERVAL {
debug!("RTR session notify skipped due to rate limit: {}", self.session_summary());
debug!(
"RTR session notify skipped due to rate limit: {}",
self.session_summary()
);
return Ok(());
}
}
@@ -824,8 +825,7 @@ where
let version = self.version()?;
debug!(
"RTR session writing Cache Response: version={}, session_id={}",
version,
session_id
version, session_id
);
CacheResponse::new(version, session_id)
.write(&mut self.stream)
@@ -835,10 +835,7 @@ where
async fn write_cache_reset(&mut self) -> Result<()> {
let version = self.version()?;
info!(
"RTR session writing Cache Reset: version={}",
version
);
info!("RTR session writing Cache Reset: version={}", version);
CacheReset::new(version).write(&mut self.stream).await?;
Ok(())
}
@@ -880,8 +877,7 @@ where
// References:
// https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-11.4
// https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-12
validate_payloads_for_rtr(payloads, announce)
.map_err(|err| anyhow!(err.to_string()))?;
validate_payloads_for_rtr(payloads, announce).map_err(|err| anyhow!(err.to_string()))?;
let (route_origins, router_keys, aspas) = count_payloads(payloads);
debug!(
"RTR session sending snapshot payloads: announce={}, total={}, route_origins={}, router_keys={}, aspas={}",
@@ -906,8 +902,7 @@ where
// References:
// https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-11.4
// https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-12
validate_payload_updates_for_rtr(&updates)
.map_err(|err| anyhow!(err.to_string()))?;
validate_payload_updates_for_rtr(&updates).map_err(|err| anyhow!(err.to_string()))?;
let (announced, withdrawn, route_origins, router_keys, aspas) =
count_payload_updates(&updates);
debug!(
@@ -1010,8 +1005,7 @@ where
});
let providers = if announce {
aspa
.provider_asns()
aspa.provider_asns()
.iter()
.map(|asn| asn.into_u32())
.collect::<Vec<_>>()
@@ -1019,18 +1013,12 @@ where
Vec::new()
};
let pdu = AspaPdu::new(
version,
flags,
aspa.customer_asn().into_u32(),
providers,
);
let pdu = AspaPdu::new(version, flags, aspa.customer_asn().into_u32(), providers);
pdu.write(&mut self.stream).await?;
Ok(())
}
async fn send_error(
&mut self,
version: u8,
@@ -1052,11 +1040,7 @@ where
.await
}
async fn handle_pdu_read_error(
&mut self,
header: Header,
err: io::Error,
) -> Result<()> {
async fn handle_pdu_read_error(&mut self, header: Header, err: io::Error) -> Result<()> {
warn!(
"RTR session failed to read established-session PDU payload: pdu={}, version={}, err={}",
header.pdu(),
@@ -1076,11 +1060,7 @@ where
Ok(())
}
async fn handle_first_pdu_read_error(
&mut self,
header: Header,
err: io::Error,
) -> Result<()> {
async fn handle_first_pdu_read_error(&mut self, header: Header, err: io::Error) -> Result<()> {
warn!(
"RTR session failed to read first PDU payload: pdu={}, version={}, err={}",
header.pdu(),
@@ -1089,13 +1069,12 @@ where
);
if err.kind() == io::ErrorKind::InvalidData {
let offending = self.read_full_pdu_bytes(header).await?;
let err_version = if (SUPPORTED_MIN_VERSION..=SUPPORTED_MAX_VERSION)
.contains(&header.version())
{
header.version()
} else {
SUPPORTED_MAX_VERSION
};
let err_version =
if (SUPPORTED_MIN_VERSION..=SUPPORTED_MAX_VERSION).contains(&header.version()) {
header.version()
} else {
SUPPORTED_MAX_VERSION
};
let detail = err.to_string();
let _ = self
@@ -1114,13 +1093,14 @@ where
) -> Result<()> {
warn!(
"RTR session handling invalid header bytes: raw_header={:02X?}, err={}",
raw_header,
err
raw_header, err
);
if err.kind() == io::ErrorKind::InvalidData {
let version = match self.version {
Some(version) => version,
None if (SUPPORTED_MIN_VERSION..=SUPPORTED_MAX_VERSION).contains(&raw_header[0]) => {
None if (SUPPORTED_MIN_VERSION..=SUPPORTED_MAX_VERSION)
.contains(&raw_header[0]) =>
{
raw_header[0]
}
None => SUPPORTED_MAX_VERSION,
@@ -1139,10 +1119,7 @@ where
async fn handle_transport_timeout(&mut self, offending_pdu: &[u8]) -> Result<()> {
let version = self.version.unwrap_or(SUPPORTED_MAX_VERSION);
let timeout = self.transport_timeout();
let detail = format!(
"transport stalled for longer than {:?}",
timeout
);
let detail = format!("transport stalled for longer than {:?}", timeout);
warn!(
"RTR session transport timeout: version={}, offending_pdu_len={}, timeout={:?}",
version,
@@ -1177,7 +1154,8 @@ where
bytes.resize(total_len, 0);
timeout(
self.transport_timeout(),
self.stream.read_exact(&mut bytes[HEADER_LEN..HEADER_LEN + payload_len]),
self.stream
.read_exact(&mut bytes[HEADER_LEN..HEADER_LEN + payload_len]),
)
.await
.map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "transport read timed out"))??;

2
src/rtr/state.rs
View File

@@ -12,7 +12,7 @@ impl State {
pub fn session_ids(self) -> SessionIds {
self.session_ids
}
pub fn serial(self) -> u32 {
self.serial
}

142
src/rtr/store.rs
View File

@@ -1,6 +1,6 @@
use anyhow::{Result, anyhow};
use rocksdb::{ColumnFamilyDescriptor, DB, Direction, IteratorMode, Options, WriteBatch};
use anyhow::{anyhow, Result};
use serde::{de::DeserializeOwned, Serialize};
use serde::{Serialize, de::DeserializeOwned};
use std::path::Path;
use std::sync::Arc;
use tokio::task;
@@ -66,7 +66,10 @@ impl RtrStore {
/// Common serialize/put.
fn put_cf<T: Serialize>(&self, cf: &str, key: &[u8], value: &T) -> Result<()> {
let cf_handle = self.db.cf_handle(cf).ok_or_else(|| anyhow!("CF not found"))?;
let cf_handle = self
.db
.cf_handle(cf)
.ok_or_else(|| anyhow!("CF not found"))?;
let data = serde_json::to_vec(value)?;
self.db.put_cf(cf_handle, key, data)?;
Ok(())
@@ -74,7 +77,10 @@ impl RtrStore {
/// Common get/deserialize.
fn get_cf<T: DeserializeOwned>(&self, cf: &str, key: &[u8]) -> Result<Option<T>> {
let cf_handle = self.db.cf_handle(cf).ok_or_else(|| anyhow!("CF not found"))?;
let cf_handle = self
.db
.cf_handle(cf)
.ok_or_else(|| anyhow!("CF not found"))?;
if let Some(value) = self.db.get_cf(cf_handle, key)? {
let obj = serde_json::from_slice(&value)?;
Ok(Some(obj))
@@ -85,7 +91,10 @@ impl RtrStore {
/// Common delete.
fn delete_cf(&self, cf: &str, key: &[u8]) -> Result<()> {
let cf_handle = self.db.cf_handle(cf).ok_or_else(|| anyhow!("CF not found"))?;
let cf_handle = self
.db
.cf_handle(cf)
.ok_or_else(|| anyhow!("CF not found"))?;
self.db.delete_cf(cf_handle, key)?;
Ok(())
}
@@ -137,10 +146,12 @@ impl RtrStore {
pub fn set_delta_window(&self, min_serial: u32, max_serial: u32) -> Result<()> {
debug!(
"RTR store persisting delta window metadata: min_serial={}, max_serial={}",
min_serial,
max_serial
min_serial, max_serial
);
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(meta_cf, META_DELTA_MIN, serde_json::to_vec(&min_serial)?);
batch.put_cf(meta_cf, META_DELTA_MAX, serde_json::to_vec(&max_serial)?);
@@ -150,7 +161,10 @@ impl RtrStore {
pub fn clear_delta_window(&self) -> Result<()> {
debug!("RTR store clearing delta window metadata");
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.delete_cf(meta_cf, META_DELTA_MIN);
batch.delete_cf(meta_cf, META_DELTA_MAX);
@@ -166,8 +180,7 @@ impl RtrStore {
(Some(min), Some(max)) => {
debug!(
"RTR store loaded delta window metadata: min_serial={}, max_serial={}",
min,
max
min, max
);
Ok(Some((min, max)))
}
@@ -189,7 +202,10 @@ impl RtrStore {
// ===============================
pub fn save_snapshot(&self, snapshot: &Snapshot) -> Result<()> {
let cf_handle = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let cf_handle = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let mut batch = WriteBatch::default();
let data = serde_json::to_vec(snapshot)?;
batch.put_cf(cf_handle, b"current", data);
@@ -206,8 +222,14 @@ impl RtrStore {
}
pub fn save_snapshot_and_state(&self, snapshot: &Snapshot, state: &State) -> Result<()> {
let snapshot_cf = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
@@ -234,8 +256,14 @@ impl RtrStore {
serial: u32,
) -> Result<()> {
let mut batch = WriteBatch::default();
let snapshot_cf = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SESSION_IDS, serde_json::to_vec(session_ids)?);
@@ -266,15 +294,28 @@ impl RtrStore {
snapshot.router_keys().len(),
snapshot.aspas().len()
);
let snapshot_cf = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let delta_cf = self.db.cf_handle(CF_DELTA).ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
let delta_cf = self
.db
.cf_handle(CF_DELTA)
.ok_or_else(|| anyhow!("CF_DELTA not found"))?;
let mut batch = WriteBatch::default();
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SESSION_IDS, serde_json::to_vec(session_ids)?);
batch.put_cf(meta_cf, META_SERIAL, serde_json::to_vec(&serial)?);
batch.put_cf(meta_cf, META_AVAILABILITY, serde_json::to_vec(&availability)?);
batch.put_cf(
meta_cf,
META_AVAILABILITY,
serde_json::to_vec(&availability)?,
);
if let Some(delta) = delta {
debug!(
@@ -283,7 +324,11 @@ impl RtrStore {
delta.announced().len(),
delta.withdrawn().len()
);
batch.put_cf(delta_cf, delta_key(delta.serial()), serde_json::to_vec(delta)?);
batch.put_cf(
delta_cf,
delta_key(delta.serial()),
serde_json::to_vec(delta)?,
);
}
if clear_delta_window {
@@ -318,8 +363,7 @@ impl RtrStore {
} else {
debug!(
"RTR store found no stale delta records outside window [{}, {}]",
min_serial,
max_serial
min_serial, max_serial
);
}
for key in stale_keys {
@@ -334,8 +378,14 @@ impl RtrStore {
pub fn save_snapshot_and_serial(&self, snapshot: &Snapshot, serial: u32) -> Result<()> {
let mut batch = WriteBatch::default();
let snapshot_cf = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", serde_json::to_vec(snapshot)?);
batch.put_cf(meta_cf, META_SERIAL, serde_json::to_vec(&serial)?);
self.db.write(batch)?;
@@ -352,8 +402,14 @@ impl RtrStore {
task::spawn_blocking(move || {
let mut batch = WriteBatch::default();
let snapshot_cf = self.db.cf_handle(CF_SNAPSHOT).ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self.db.cf_handle(CF_META).ok_or_else(|| anyhow!("CF_META not found"))?;
let snapshot_cf = self
.db
.cf_handle(CF_SNAPSHOT)
.ok_or_else(|| anyhow!("CF_SNAPSHOT not found"))?;
let meta_cf = self
.db
.cf_handle(CF_META)
.ok_or_else(|| anyhow!("CF_META not found"))?;
batch.put_cf(snapshot_cf, b"current", snapshot_bytes);
batch.put_cf(meta_cf, META_SERIAL, serial_bytes);
self.db.write(batch)?;
@@ -370,7 +426,9 @@ impl RtrStore {
match (snapshot, state) {
(Some(snap), Some(state)) => Ok(Some((snap, state))),
(None, None) => Ok(None),
_ => Err(anyhow!("Inconsistent DB state: snapshot and state mismatch")),
_ => Err(anyhow!(
"Inconsistent DB state: snapshot and state mismatch"
)),
}
}
@@ -380,7 +438,9 @@ impl RtrStore {
match (snapshot, serial) {
(Some(snap), Some(serial)) => Ok(Some((snap, serial))),
(None, None) => Ok(None),
_ => Err(anyhow!("Inconsistent DB state: snapshot and serial mismatch")),
_ => Err(anyhow!(
"Inconsistent DB state: snapshot and serial mismatch"
)),
}
}
@@ -413,8 +473,8 @@ impl RtrStore {
for item in iter {
let (key, value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let parsed = delta_key_serial(key.as_ref())
.ok_or_else(|| anyhow!("Invalid delta key"))?;
let parsed =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
if parsed <= serial {
continue;
@@ -430,8 +490,7 @@ impl RtrStore {
pub fn load_delta_window(&self, min_serial: u32, max_serial: u32) -> Result<Vec<Delta>> {
info!(
"RTR store loading persisted delta window: min_serial={}, max_serial={}",
min_serial,
max_serial
min_serial, max_serial
);
let cf_handle = self
.db
@@ -442,8 +501,8 @@ impl RtrStore {
for item in iter {
let (key, value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let parsed = delta_key_serial(key.as_ref())
.ok_or_else(|| anyhow!("Invalid delta key"))?;
let parsed =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
// Restore by the persisted window bounds instead of load_deltas_since().
// The latter follows lexicographic key order and is not safe across serial
@@ -493,7 +552,11 @@ impl RtrStore {
Ok(keys)
}
fn list_delta_keys_outside_window(&self, min_serial: u32, max_serial: u32) -> Result<Vec<Vec<u8>>> {
fn list_delta_keys_outside_window(
&self,
min_serial: u32,
max_serial: u32,
) -> Result<Vec<Vec<u8>>> {
let cf_handle = self
.db
.cf_handle(CF_DELTA)
@@ -503,8 +566,8 @@ impl RtrStore {
for item in iter {
let (key, _value) = item.map_err(|e| anyhow!("rocksdb iterator error: {}", e))?;
let serial = delta_key_serial(key.as_ref())
.ok_or_else(|| anyhow!("Invalid delta key"))?;
let serial =
delta_key_serial(key.as_ref()).ok_or_else(|| anyhow!("Invalid delta key"))?;
if !serial_in_window(serial, min_serial, max_serial) {
keys.push(key.to_vec());
}
@@ -522,8 +585,7 @@ fn validate_delta_window(deltas: &[Delta], min_serial: u32, max_serial: u32) ->
if deltas.is_empty() {
warn!(
"RTR store delta window validation failed: no persisted deltas for window [{}, {}]",
min_serial,
max_serial
min_serial, max_serial
);
return Err(anyhow!(
"delta window [{}, {}] has no persisted deltas",