diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 145573d..e28c428 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,7 +44,7 @@ jobs: env: RUSTFLAGS: "" run: | - for target in message message_body prefix_text config; do + for target in message message_body prefix_text config rtr; do mkdir -p "corpus/$target" cargo fuzz run "$target" "corpus/$target" "seeds/$target" -- -max_total_time=60 done diff --git a/Cargo.lock b/Cargo.lock index c3f83b5..f6bdaf6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -248,6 +248,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "bgp-rpki" +version = "0.0.2" +dependencies = [ + "bgp-primitives", + "bytes", + "proptest", + "thiserror 2.0.12", + "tokio", + "tracing", +] + [[package]] name = "bgp-server" version = "0.0.2" diff --git a/Cargo.toml b/Cargo.toml index 4c7d5fd..1caff74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,7 @@ unimplemented = "warn" bgp-packet = { path = "crates/packet" } bgp-primitives = { path = "crates/primitives" } bgp-rib = { path = "crates/rib" } +bgp-rpki = { path = "crates/rpki" } bgp-server = { path = "crates/server" } bgp-web = { path = "crates/web" } bgp-integration-tests = { path = "crates/integration-tests" } diff --git a/crates/primitives/src/constants.rs b/crates/primitives/src/constants.rs index 337f8d5..c13f0a0 100644 --- a/crates/primitives/src/constants.rs +++ b/crates/primitives/src/constants.rs @@ -11,7 +11,7 @@ pub struct UnknownAfi(pub u16); /// The IP version of a prefix. For the address family field of a capability /// or an MP_REACH_NLRI, which can name families that are not IP, see [`Afi`]. -#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize_repr)] +#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize_repr)] #[repr(u16)] pub enum IpVersion { Ipv4 = 1, diff --git a/crates/primitives/src/ip_prefix.rs b/crates/primitives/src/ip_prefix.rs index e332446..93b28ef 100644 --- a/crates/primitives/src/ip_prefix.rs +++ b/crates/primitives/src/ip_prefix.rs @@ -31,7 +31,7 @@ pub enum PrefixError { /// value. A prefix has one representation: the bits past its length are /// always zero, so two prefixes are equal exactly when they are the same /// prefix. -#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash, Serialize)] +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Hash, Serialize)] pub struct IpPrefix { address_family: IpVersion, length: u8, diff --git a/crates/rpki/Cargo.toml b/crates/rpki/Cargo.toml new file mode 100644 index 0000000..af1cf22 --- /dev/null +++ b/crates/rpki/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "bgp-rpki" +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +version.workspace = true + +[dependencies] +bgp-primitives.workspace = true +bytes.workspace = true +thiserror.workspace = true +tokio = { workspace = true, features = ["full"] } +tracing.workspace = true + +[dev-dependencies] +proptest.workspace = true + +[lints] +workspace = true diff --git a/crates/rpki/src/client.rs b/crates/rpki/src/client.rs new file mode 100644 index 0000000..8e51076 --- /dev/null +++ b/crates/rpki/src/client.rs @@ -0,0 +1,647 @@ +//! Keeps the list of VRPs in step with a validator, over the RPKI-to-Router +//! protocol (RFC 8210). +//! +//! The client connects to a cache (the validator's RTR server), asks for +//! everything once, and from then on for what has changed: when the cache +//! says there is something new, and otherwise every so often. Each time the +//! data changes a new [`VrpTable`] is published for whoever validates +//! routes. +//! +//! A cache that cannot be reached is tried again, and the other caches in +//! turn. Meanwhile the data last received is used, but not for ever: a list +//! that has not been confirmed for the expire interval is dropped, and +//! routes are then neither valid nor invalid. Using stale data would be +//! worse: it would go on refusing routes whose holders have since +//! authorised them. + +use std::collections::HashSet; +use std::io; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::{Duration, SystemTime}; + +use bgp_primitives::constants::IpVersion; +use bytes::BytesMut; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::sync::watch; +use tokio::task::JoinHandle; +use tokio::time::{Instant, sleep_until, timeout}; +use tracing::{info, warn}; + +use crate::rtr::{Pdu, RtrError, Timers, VERSION_0, VERSION_1, error_code}; +use crate::vrp::{Vrp, VrpTable}; + +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// How long a cache may take to send the next part of an answer. +const RESPONSE_TIMEOUT: Duration = Duration::from_secs(60); + +/// How long to wait before trying the next cache after one has failed. +const NEXT_CACHE_DELAY: Duration = Duration::from_secs(1); + +/// Where the VRPs come from, and how patiently. +#[derive(Debug, Clone, Default)] +pub struct RpkiConfig { + /// The caches, in order of preference. One is used at a time; the next + /// is tried when it fails. + pub caches: Vec, + /// If set, used in place of what the cache says (or of the defaults of + /// RFC 8210 §6, with a cache that says nothing). + pub refresh: Option, + pub retry: Option, + pub expire: Option, +} + +/// The VRPs as they are after a change. +#[derive(Debug, Clone)] +pub struct Update { + /// Goes up by one with every update. + pub version: u64, + pub table: Arc, + /// The VRPs that were added or removed to get here from the version + /// before: only routes that these cover can have changed validity. + /// `None` if any route may have. + pub changed: Option>, +} + +/// How things stand with the caches. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CacheState { + /// No cache has been reached yet, or the one in use was lost. + Connecting, + /// Connected and waiting for the data. + Synchronising, + /// The data is complete as of the cache's serial. + Synchronised, +} + +/// What there is to say about the client, for showing. +#[derive(Debug, Clone)] +pub struct Status { + /// The cache in use, or last tried. + pub cache: Option, + pub state: CacheState, + /// The cache's session and the serial the data is at. + pub session_id: Option, + pub serial: Option, + pub protocol_version: u8, + pub vrps_ipv4: usize, + pub vrps_ipv6: usize, + /// When the data was last confirmed by a cache. + pub updated_at: Option, + /// When it will be dropped unless confirmed again. + pub expires_at: Option, + /// Why the last attempt failed, if it did. + pub last_error: Option, +} + +struct Shared { + updates: watch::Sender, + /// The version of the latest update, for a look that costs nothing. + version: AtomicU64, + status: Mutex, +} + +/// Handle to the VRPs and to the client that keeps them. Cheap to clone. +#[derive(Clone)] +pub struct Rpki { + shared: Arc, +} + +impl Rpki { + /// Starts a client for `config` and returns the handle to its data, + /// with the task that runs it. Until the first cache has answered the + /// table is empty, and every route is [`crate::Validity::NotFound`]. + pub fn spawn(config: RpkiConfig) -> (Self, JoinHandle<()>) { + let (updates, _) = watch::channel(Update { + version: 0, + table: Arc::default(), + changed: None, + }); + let rpki = Self { + shared: Arc::new(Shared { + updates, + version: AtomicU64::new(0), + status: Mutex::new(Status { + cache: None, + state: CacheState::Connecting, + session_id: None, + serial: None, + protocol_version: VERSION_1, + vrps_ipv4: 0, + vrps_ipv6: 0, + updated_at: None, + expires_at: None, + last_error: None, + }), + }), + }; + let client = Client { + shared: Arc::clone(&rpki.shared), + config, + vrps: HashSet::new(), + place: None, + protocol_version: VERSION_1, + timers: Timers::default(), + expires: None, + }; + (rpki, tokio::spawn(client.run())) + } + + /// The VRPs as they are now. + pub fn current(&self) -> Update { + self.shared.updates.borrow().clone() + } + + /// The version of the latest [`Update`]. Costs a load of one number, + /// for those who ask before every use whether theirs is still current. + pub fn version(&self) -> u64 { + self.shared.version.load(Ordering::Acquire) + } + + /// Follows the updates. A follower that is busy sees the latest only; + /// [`Update::version`] tells it whether it skipped any. + pub fn subscribe(&self) -> watch::Receiver { + self.shared.updates.subscribe() + } + + pub fn status(&self) -> Status { + self.shared + .status + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } +} + +/// Why a session with a cache ended. +#[derive(Debug, thiserror::Error)] +enum SessionError { + #[error("{0}")] + Io(#[from] io::Error), + #[error("the cache sent something that is not RTR: {0}")] + Malformed(#[from] RtrError), + #[error("the cache closed the connection")] + Closed, + #[error("the cache did not answer within {RESPONSE_TIMEOUT:?}")] + Timeout, + #[error("the cache reported an error: {text} (code {code})")] + Reported { code: u16, text: String }, + #[error("the cache sent {0} where it has no place")] + Unexpected(&'static str), + #[error("the cache {0}")] + Inconsistent(&'static str), + /// Not a failure of the cache: it speaks the older version, and is to + /// be asked again in that. + #[error("the cache speaks protocol version 0")] + Downgrade, +} + +/// Where the data is at: which of the cache's sessions, and which serial. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Place { + cache: SocketAddr, + session_id: u16, + serial: u32, +} + +struct Client { + shared: Arc, + config: RpkiConfig, + /// The VRPs as last received. + vrps: HashSet, + /// Where `vrps` is at, if it came from a cache that can be asked for + /// what has changed since. + place: Option, + protocol_version: u8, + /// As the cache last gave them, or the defaults. + timers: Timers, + /// When `vrps` is to be dropped unless a cache confirms it first. + expires: Option, +} + +/// A connection to a cache, read a PDU at a time. +struct Connection { + stream: TcpStream, + buf: BytesMut, +} + +impl Connection { + async fn send(&mut self, version: u8, pdu: &Pdu) -> io::Result<()> { + let mut out = BytesMut::new(); + pdu.encode(version, &mut out); + self.stream.write_all(&out).await + } + + /// The next PDU, and the protocol version it was sent with. + /// Cancel-safe: what has arrived of a PDU stays in the buffer. + async fn next(&mut self) -> Result<(u8, Pdu), SessionError> { + loop { + if let Some((version, pdu, used)) = Pdu::decode(&self.buf)? { + let _ = self.buf.split_to(used); + return Ok((version, pdu)); + } + if self.stream.read_buf(&mut self.buf).await? == 0 { + return Err(SessionError::Closed); + } + } + } +} + +impl Client { + fn status(&self, change: impl FnOnce(&mut Status)) { + change( + &mut self + .shared + .status + .lock() + .unwrap_or_else(PoisonError::into_inner), + ); + } + + fn refresh(&self) -> Duration { + self.config.refresh.unwrap_or(Duration::from_secs( + self.timers.refresh.clamp(1, 86400).into(), + )) + } + + fn retry(&self) -> Duration { + self.config + .retry + .unwrap_or(Duration::from_secs(self.timers.retry.clamp(1, 7200).into())) + } + + fn expire(&self) -> Duration { + self.config.expire.unwrap_or(Duration::from_secs( + self.timers.expire.clamp(600, 172_800).into(), + )) + } + + async fn run(mut self) { + if self.config.caches.is_empty() { + return std::future::pending().await; + } + let mut attempt = 0usize; + loop { + let cache = self.config.caches[attempt % self.config.caches.len()]; + self.status(|status| { + status.cache = Some(cache); + status.state = CacheState::Connecting; + }); + let ended = self.session(cache).await; + if matches!(ended, SessionError::Downgrade) { + info!(%cache, "RPKI cache speaks RTR version 0; asking again in that"); + self.protocol_version = VERSION_0; + continue; + } + warn!(%cache, err = %ended, "RPKI cache session ended"); + self.status(|status| { + status.state = CacheState::Connecting; + status.last_error = Some(ended.to_string()); + }); + + // The next cache at once; when all have been tried, a rest. + attempt += 1; + // What is new may be at a newer version. + self.protocol_version = VERSION_1; + let wait = if attempt.is_multiple_of(self.config.caches.len()) { + self.retry() + } else { + NEXT_CACHE_DELAY + }; + self.wait(wait).await; + } + } + + /// Waits for `duration`, dropping the data on the way if its time runs + /// out. + async fn wait(&mut self, duration: Duration) { + let until = Instant::now() + duration; + if let Some(expires) = self.expires + && expires <= until + { + sleep_until(expires).await; + self.expired(); + } + sleep_until(until).await; + } + + /// The data has gone unconfirmed for too long to be used. + fn expired(&mut self) { + warn!( + vrps = self.vrps.len(), + "RPKI data has expired without a cache to confirm it; dropping it" + ); + self.vrps.clear(); + self.place = None; + self.expires = None; + self.publish(Arc::default(), None); + self.status(|status| { + status.session_id = None; + status.serial = None; + status.expires_at = None; + }); + } + + /// Runs a session with `cache` until it ends, which it does only by + /// failing. + async fn session(&mut self, cache: SocketAddr) -> SessionError { + let stream = match timeout(CONNECT_TIMEOUT, TcpStream::connect(cache)).await { + Ok(Ok(stream)) => stream, + Ok(Err(e)) => return e.into(), + Err(_) => return SessionError::Timeout, + }; + let mut connection = Connection { + stream, + buf: BytesMut::with_capacity(64 * 1024), + }; + info!(%cache, "Connected to RPKI cache"); + self.status(|status| { + status.state = CacheState::Synchronising; + status.protocol_version = self.protocol_version; + }); + + loop { + if let Err(e) = self.synchronise(cache, &mut connection).await { + return e; + } + // Until the cache says there is something new, or it is time + // to ask anyway. If the data runs out first, it is the cache + // that has stopped confirming it. + let ask = Instant::now() + self.refresh(); + let expires = self.expires.unwrap_or(ask); + tokio::select! { + pdu = connection.next() => match pdu { + Ok((_, Pdu::SerialNotify { .. })) => {} + Ok((_, Pdu::ErrorReport { code, text })) => { + return SessionError::Reported { code, text }; + } + Ok(_) => return SessionError::Unexpected("an answer nobody asked for"), + Err(e) => return e, + }, + () = sleep_until(ask) => {} + () = sleep_until(expires), if expires < ask => { + self.expired(); + return SessionError::Timeout; + } + } + } + } + + /// Asks the cache for what has changed, or for everything, and takes + /// the answer. + async fn synchronise( + &mut self, + cache: SocketAddr, + connection: &mut Connection, + ) -> Result<(), SessionError> { + // Changes can be asked for only of the cache the data came from. + let mut since = self.place.filter(|place| place.cache == cache); + loop { + let query = match since { + Some(place) => Pdu::SerialQuery { + session_id: place.session_id, + serial: place.serial, + }, + None => Pdu::ResetQuery, + }; + connection.send(self.protocol_version, &query).await?; + match self.answer(cache, connection, since).await? { + Answer::Complete => return Ok(()), + // The cache cannot say what changed: ask for everything. + Answer::Reset => since = None, + } + } + } + + /// Reads the answer to a query. `since` is the place a Serial Query + /// asked about; `None` after a Reset Query. + async fn answer( + &mut self, + cache: SocketAddr, + connection: &mut Connection, + since: Option, + ) -> Result { + let mut next = async || match timeout(RESPONSE_TIMEOUT, connection.next()).await { + Ok(pdu) => pdu, + Err(_) => Err(SessionError::Timeout), + }; + + // The answer begins with a Cache Response, or is a Cache Reset. + let session_id = loop { + let (version, pdu) = next().await?; + if version < self.protocol_version { + // An older cache answers in its own version; the query has + // to be put again in that (RFC 8210 §7). + return Err(SessionError::Downgrade); + } + match pdu { + Pdu::CacheResponse { session_id } => break session_id, + Pdu::CacheReset if since.is_some() => return Ok(Answer::Reset), + // News of what is being asked for anyway. + Pdu::SerialNotify { .. } => {} + Pdu::ErrorReport { code, .. } + if code == error_code::UNSUPPORTED_VERSION + && self.protocol_version != VERSION_0 => + { + return Err(SessionError::Downgrade); + } + Pdu::ErrorReport { code, text } => { + return Err(SessionError::Reported { code, text }); + } + _ => return Err(SessionError::Unexpected("data before a Cache Response")), + } + }; + if since.is_some_and(|place| place.session_id != session_id) { + // The cache has started over; its serials mean something else + // now. + return Ok(Answer::Reset); + } + + // Then the VRPs, and End of Data. Nothing takes effect until the + // answer is complete (RFC 8210 §8.1), so what it says is kept + // aside in the order it came. + let mut records: Vec<(Vrp, bool)> = Vec::new(); + let (serial, timers) = loop { + match next().await?.1 { + Pdu::Prefix { announce, vrp } => records.push((vrp, announce)), + Pdu::RouterKey => {} + Pdu::EndOfData { + session_id: ended, + serial, + timers, + } => { + if ended != session_id { + return Err(SessionError::Inconsistent( + "ended an answer in another session than it began it", + )); + } + break (serial, timers); + } + Pdu::ErrorReport { code, text } => { + return Err(SessionError::Reported { code, text }); + } + _ => { + return Err(SessionError::Unexpected( + "a PDU that is not part of an answer", + )); + } + } + }; + + // The first data there is can change any route's validity; after + // that, only where a VRP came or went. + let first = self.place.is_none() && self.expires.is_none(); + let before = match since { + // Changes, to what there is. + Some(_) => std::mem::take(&mut self.vrps), + // Everything anew. + None => HashSet::new(), + }; + let replaced = match since { + Some(_) => None, + None => Some(std::mem::take(&mut self.vrps)), + }; + // Building the tables is work for a thread that may take its time. + let built = tokio::task::spawn_blocking(move || apply(before, replaced, &records, first)) + .await + .map_err(|_| SessionError::Inconsistent("could not be followed (a bug here)"))?; + let built = match built { + Ok(built) => built, + Err((vrps, problem)) => { + // What there was stays as it was. + self.vrps = vrps; + return Err(SessionError::Inconsistent(problem)); + } + }; + + self.vrps = built.vrps; + self.place = Some(Place { + cache, + session_id, + serial, + }); + self.timers = timers.unwrap_or_default(); + self.expires = Some(Instant::now() + self.expire()); + if first || built.changes > 0 { + self.publish(built.table, built.changed); + } + info!( + %cache, + serial, + vrps = self.vrps.len(), + changed = built.changes, + "RPKI data is up to date" + ); + let expires_at = SystemTime::now() + self.expire(); + self.status(|status| { + status.state = CacheState::Synchronised; + status.session_id = Some(session_id); + status.serial = Some(serial); + status.updated_at = Some(SystemTime::now()); + status.expires_at = Some(expires_at); + status.last_error = None; + }); + Ok(Answer::Complete) + } + + /// Makes `table`, which holds the VRPs as they now are, the one in + /// use. + fn publish(&self, table: Arc, changed: Option>) { + let count = |version| { + self.vrps + .iter() + .filter(|vrp| vrp.prefix.address_family() == version) + .count() + }; + let (ipv4, ipv6) = (count(IpVersion::Ipv4), count(IpVersion::Ipv6)); + self.status(|status| { + status.vrps_ipv4 = ipv4; + status.vrps_ipv6 = ipv6; + }); + self.shared.updates.send_modify(|update| { + update.version += 1; + update.table = table; + update.changed = changed; + self.shared.version.store(update.version, Ordering::Release); + }); + } +} + +/// The VRPs after an answer, ready to be used. +struct Built { + vrps: HashSet, + table: Arc, + /// See [`Update::changed`]. + changed: Option>, + /// How many VRPs came or went. + changes: usize, +} + +/// Applies the `records` of a complete answer, in order, to `vrps`: each +/// a VRP and whether it is announced or withdrawn. `replaced` is the set +/// that an answer to a Reset Query takes the place of. +/// +/// If the answer contradicts itself or what there was, what there was is +/// given back with the reason. +fn apply( + mut vrps: HashSet, + replaced: Option>, + records: &[(Vrp, bool)], + first: bool, +) -> Result, &'static str)> { + let mut problem = None; + let mut applied = 0; + for (vrp, announce) in records { + let ok = match announce { + true => vrps.insert(*vrp), + false => vrps.remove(vrp), + }; + if !ok { + problem = Some(match announce { + true => "announced a VRP twice", + false => "withdrew a VRP it had not announced", + }); + break; + } + applied += 1; + } + if let Some(problem) = problem { + return Err(match replaced { + Some(replaced) => (replaced, problem), + None => { + // Undo, last first. + for (vrp, announce) in records[..applied].iter().rev() { + match announce { + true => vrps.remove(vrp), + false => vrps.insert(*vrp), + }; + } + (vrps, problem) + } + }); + } + + // What came or went: against the set replaced, or else whatever the + // answer touched. + let changed: Vec = match &replaced { + Some(replaced) => vrps.symmetric_difference(replaced).copied().collect(), + None => records.iter().map(|(vrp, _)| *vrp).collect(), + }; + Ok(Built { + table: Arc::new(VrpTable::new(vrps.iter().copied())), + changed: (!first).then(|| Arc::new(VrpTable::new(changed.iter().copied()))), + changes: changed.len(), + vrps, + }) +} + +/// How a query was answered. +enum Answer { + /// With data, which is now in use. + Complete, + /// With a request to ask for everything instead. + Reset, +} diff --git a/crates/rpki/src/lib.rs b/crates/rpki/src/lib.rs new file mode 100644 index 0000000..a39c943 --- /dev/null +++ b/crates/rpki/src/lib.rs @@ -0,0 +1,19 @@ +//! Route origin validation with the RPKI (RFC 6811). +//! +//! The holder of an address block can sign a statement of which AS may +//! originate routes for it. Validators collect and check those statements +//! and hand routers the result, a list of validated ROA payloads (VRPs), +//! over the RPKI-to-Router protocol (RFC 8210). With that list a router can +//! tell a route whose origin is authorised from one whose origin is not: +//! the commonest kind of hijack and of mistake. +//! +//! - [`vrp`] holds the list and answers whether a route is valid. +//! - [`rtr`] is the protocol's messages. +//! - [`client`] keeps the list in step with a validator. + +pub mod client; +pub mod rtr; +pub mod vrp; + +pub use client::{CacheState, Rpki, RpkiConfig, Status, Update}; +pub use vrp::{Validity, Vrp, VrpTable}; diff --git a/crates/rpki/src/rtr.rs b/crates/rpki/src/rtr.rs new file mode 100644 index 0000000..b94ad94 --- /dev/null +++ b/crates/rpki/src/rtr.rs @@ -0,0 +1,553 @@ +//! The messages of the RPKI-to-Router protocol (RFC 8210, and RFC 6810 for +//! version 0), as bytes and back. No I/O here. +//! +//! Every message (a PDU) begins with the same eight octets: the protocol +//! version, the type, a field whose meaning depends on the type, and the +//! length of the whole PDU. + +use bgp_primitives::constants::IpVersion; +use bgp_primitives::ip_prefix::IpPrefix; +use bytes::{BufMut, BytesMut}; + +use crate::vrp::Vrp; + +/// The protocol version this client asks for. Version 1 added the timers in +/// End of Data and router keys. +pub const VERSION_1: u8 = 1; +/// What an older validator speaks. +pub const VERSION_0: u8 = 0; + +const HEADER_LEN: usize = 8; + +/// The longest PDU this client reads. Every PDU but Error Report and Router +/// Key has a fixed, small length, and those two have no need to be large. +pub const MAX_PDU_LEN: usize = 16 * 1024; + +mod pdu_type { + pub const SERIAL_NOTIFY: u8 = 0; + pub const SERIAL_QUERY: u8 = 1; + pub const RESET_QUERY: u8 = 2; + pub const CACHE_RESPONSE: u8 = 3; + pub const IPV4_PREFIX: u8 = 4; + pub const IPV6_PREFIX: u8 = 6; + pub const END_OF_DATA: u8 = 7; + pub const CACHE_RESET: u8 = 8; + pub const ROUTER_KEY: u8 = 9; + pub const ERROR_REPORT: u8 = 10; +} + +/// Error codes of an Error Report (RFC 8210 §12). +pub mod error_code { + pub const CORRUPT_DATA: u16 = 0; + pub const INTERNAL_ERROR: u16 = 1; + pub const NO_DATA_AVAILABLE: u16 = 2; + pub const INVALID_REQUEST: u16 = 3; + pub const UNSUPPORTED_VERSION: u16 = 4; + pub const UNSUPPORTED_PDU_TYPE: u16 = 5; + pub const WITHDRAWAL_OF_UNKNOWN_RECORD: u16 = 6; + pub const DUPLICATE_ANNOUNCEMENT: u16 = 7; + pub const UNEXPECTED_VERSION: u16 = 8; +} + +/// The timers a validator tells its routers to use (RFC 8210 §6), in +/// seconds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Timers { + /// How long to wait before asking for changes unprompted. + pub refresh: u32, + /// How long to wait before trying again after a failure. + pub retry: u32, + /// How long the data may be used without a successful update. + pub expire: u32, +} + +impl Default for Timers { + /// What RFC 8210 §6 recommends, and what a version 0 session uses. + fn default() -> Self { + Self { + refresh: 3600, + retry: 600, + expire: 7200, + } + } +} + +/// One PDU. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Pdu { + /// The validator has new data, up to `serial`. + SerialNotify { session_id: u16, serial: u32 }, + /// Send what has changed since `serial`. + SerialQuery { session_id: u16, serial: u32 }, + /// Send everything. + ResetQuery, + /// The answer to a query begins: prefixes follow, then End of Data. + CacheResponse { session_id: u16 }, + /// A VRP, announced or withdrawn. + Prefix { announce: bool, vrp: Vrp }, + /// The answer is complete, and the data is now at `serial`. `timers` + /// are there from version 1 on. + EndOfData { + session_id: u16, + serial: u32, + timers: Option, + }, + /// The validator cannot say what changed since the serial asked about; + /// ask for everything. + CacheReset, + /// A BGPsec router key. Not used here; read so as to be skipped. + RouterKey, + /// Something went wrong at the sender's end, or with what we sent. + ErrorReport { code: u16, text: String }, +} + +/// Bytes that are not a PDU this client can read. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum RtrError { + #[error("PDU of type {pdu_type} has length {length}, which is not valid for it")] + Length { pdu_type: u8, length: u32 }, + #[error("unknown PDU type {0}")] + UnknownType(u8), + #[error("protocol version {0} is not supported")] + Version(u8), + #[error("a prefix PDU with prefix length {length} and maximum length {max_length}")] + Prefix { length: u8, max_length: u8 }, +} + +impl RtrError { + /// The error code that reports this to the sender (RFC 8210 §12). + pub fn code(&self) -> u16 { + match self { + Self::Length { .. } | Self::Prefix { .. } => error_code::CORRUPT_DATA, + Self::UnknownType(_) => error_code::UNSUPPORTED_PDU_TYPE, + Self::Version(_) => error_code::UNSUPPORTED_VERSION, + } + } +} + +fn be_u16(bytes: &[u8]) -> u16 { + u16::from_be_bytes([bytes[0], bytes[1]]) +} + +fn be_u32(bytes: &[u8]) -> u32 { + u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) +} + +impl Pdu { + /// Reads the PDU at the front of `buf`, if all of it has arrived: + /// the protocol version it was sent with, the PDU, and how many octets + /// it took. `Ok(None)` means more octets are needed. + pub fn decode(buf: &[u8]) -> Result, RtrError> { + let Some(header) = buf.first_chunk::() else { + return Ok(None); + }; + let (version, pdu_type) = (header[0], header[1]); + let field = be_u16(&header[2..]); + let length = be_u32(&header[4..]); + let bad_length = || RtrError::Length { pdu_type, length }; + + if version > VERSION_1 { + return Err(RtrError::Version(version)); + } + let total = usize::try_from(length).map_err(|_| bad_length())?; + if !(HEADER_LEN..=MAX_PDU_LEN).contains(&total) { + return Err(bad_length()); + } + // The fixed lengths are checked before waiting for the rest, so + // that a wrong one is not waited for. + let fixed = match pdu_type { + pdu_type::SERIAL_NOTIFY | pdu_type::SERIAL_QUERY => Some(12), + pdu_type::RESET_QUERY | pdu_type::CACHE_RESPONSE | pdu_type::CACHE_RESET => Some(8), + pdu_type::IPV4_PREFIX => Some(20), + pdu_type::IPV6_PREFIX => Some(32), + pdu_type::END_OF_DATA => Some(if version == VERSION_0 { 12 } else { 24 }), + pdu_type::ROUTER_KEY | pdu_type::ERROR_REPORT => None, + other => return Err(RtrError::UnknownType(other)), + }; + if fixed.is_some_and(|fixed| fixed != total) { + return Err(bad_length()); + } + let Some(body) = buf.get(HEADER_LEN..total) else { + return Ok(None); + }; + + let pdu = match pdu_type { + pdu_type::SERIAL_NOTIFY => Self::SerialNotify { + session_id: field, + serial: be_u32(body), + }, + pdu_type::SERIAL_QUERY => Self::SerialQuery { + session_id: field, + serial: be_u32(body), + }, + pdu_type::RESET_QUERY => Self::ResetQuery, + pdu_type::CACHE_RESPONSE => Self::CacheResponse { session_id: field }, + pdu_type::IPV4_PREFIX | pdu_type::IPV6_PREFIX => { + let version = if pdu_type == pdu_type::IPV4_PREFIX { + IpVersion::Ipv4 + } else { + IpVersion::Ipv6 + }; + let (flags, length, max_length) = (body[0], body[1], body[2]); + let address = &body[4..body.len() - 4]; + let bad_prefix = || RtrError::Prefix { length, max_length }; + if max_length < length || max_length > version.bits() { + return Err(bad_prefix()); + } + Self::Prefix { + announce: flags & 1 == 1, + vrp: Vrp { + prefix: IpPrefix::new(version, address, length) + .map_err(|_| bad_prefix())?, + max_length, + asn: be_u32(&body[body.len() - 4..]), + }, + } + } + pdu_type::END_OF_DATA => Self::EndOfData { + session_id: field, + serial: be_u32(body), + timers: (version != VERSION_0).then(|| Timers { + refresh: be_u32(&body[4..]), + retry: be_u32(&body[8..]), + expire: be_u32(&body[12..]), + }), + }, + pdu_type::CACHE_RESET => Self::CacheReset, + pdu_type::ROUTER_KEY => Self::RouterKey, + pdu_type::ERROR_REPORT => { + // The length of the PDU that was at fault, that PDU, the + // length of the text, the text. + let text = (|| { + let pdu_len = usize::try_from(be_u32(body.get(..4)?)).ok()?; + let rest = body.get(4usize.checked_add(pdu_len)?..)?; + let text_len = usize::try_from(be_u32(rest.get(..4)?)).ok()?; + let text = rest.get(4..)?; + (text.len() == text_len).then(|| String::from_utf8_lossy(text).into_owned()) + })() + .ok_or_else(bad_length)?; + Self::ErrorReport { code: field, text } + } + _ => unreachable!("unknown types were refused above"), + }; + Ok(Some((version, pdu, total))) + } + + /// Writes the PDU, for a session that speaks protocol version + /// `version`, to the end of `out`. + pub fn encode(&self, version: u8, out: &mut BytesMut) { + let start = out.len(); + let header = |out: &mut BytesMut, pdu_type: u8, field: u16| { + out.put_u8(version); + out.put_u8(pdu_type); + out.put_u16(field); + // The length, filled in below. + out.put_u32(0); + }; + match self { + Self::SerialNotify { session_id, serial } => { + header(out, pdu_type::SERIAL_NOTIFY, *session_id); + out.put_u32(*serial); + } + Self::SerialQuery { session_id, serial } => { + header(out, pdu_type::SERIAL_QUERY, *session_id); + out.put_u32(*serial); + } + Self::ResetQuery => header(out, pdu_type::RESET_QUERY, 0), + Self::CacheResponse { session_id } => { + header(out, pdu_type::CACHE_RESPONSE, *session_id); + } + Self::Prefix { announce, vrp } => { + let (pdu_type, address_len) = match vrp.prefix.address_family() { + IpVersion::Ipv4 => (pdu_type::IPV4_PREFIX, 4), + IpVersion::Ipv6 => (pdu_type::IPV6_PREFIX, 16), + }; + header(out, pdu_type, 0); + out.put_u8(u8::from(*announce)); + out.put_u8(vrp.prefix.length()); + out.put_u8(vrp.max_length); + out.put_u8(0); + let mut address = [0u8; 16]; + address[..vrp.prefix.bytes().len()].copy_from_slice(vrp.prefix.bytes()); + out.put_slice(&address[..address_len]); + out.put_u32(vrp.asn); + } + Self::EndOfData { + session_id, + serial, + timers, + } => { + header(out, pdu_type::END_OF_DATA, *session_id); + out.put_u32(*serial); + if version != VERSION_0 { + let timers = timers.unwrap_or_default(); + out.put_u32(timers.refresh); + out.put_u32(timers.retry); + out.put_u32(timers.expire); + } + } + Self::CacheReset => header(out, pdu_type::CACHE_RESET, 0), + Self::RouterKey => header(out, pdu_type::ROUTER_KEY, 0), + Self::ErrorReport { code, text } => { + header(out, pdu_type::ERROR_REPORT, *code); + // No copy of the PDU at fault. + out.put_u32(0); + out.put_u32(text.len() as u32); + out.put_slice(text.as_bytes()); + } + } + let length = (out.len() - start) as u32; + out[start + 4..start + 8].copy_from_slice(&length.to_be_bytes()); + } +} + +#[cfg(test)] +mod tests { + use proptest::prelude::*; + + use super::*; + + fn decode_one(bytes: &[u8]) -> (u8, Pdu) { + let (version, pdu, used) = Pdu::decode(bytes).unwrap().expect("a whole PDU"); + assert_eq!(used, bytes.len()); + (version, pdu) + } + + /// The PDUs as RFC 8210 §5 draws them. + #[test] + fn pdus_are_read_as_the_rfc_lays_them_out() { + #[rustfmt::skip] + let ipv4 = [ + 1, 4, 0, 0, 0, 0, 0, 20, + 1, 21, 24, 0, + 193, 0, 0, 0, + 0, 0, 0x0d, 0x05, + ]; + assert_eq!( + decode_one(&ipv4), + ( + 1, + Pdu::Prefix { + announce: true, + vrp: Vrp { + prefix: IpPrefix::try_from("193.0.0.0/21").unwrap(), + max_length: 24, + asn: 3333, + }, + } + ) + ); + #[rustfmt::skip] + let ipv6 = [ + 1, 6, 0, 0, 0, 0, 0, 32, + 0, 48, 48, 0, + 0x20, 0x01, 0x06, 0x7c, 0x02, 0xe8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0x0d, 0x05, + ]; + assert_eq!( + decode_one(&ipv6), + ( + 1, + Pdu::Prefix { + announce: false, + vrp: Vrp { + prefix: IpPrefix::try_from("2001:67c:2e8::/48").unwrap(), + max_length: 48, + asn: 3333, + }, + } + ) + ); + #[rustfmt::skip] + let end_of_data = [ + 1, 7, 0x12, 0x34, 0, 0, 0, 24, + 0, 0, 0, 99, + 0, 0, 0x0e, 0x10, 0, 0, 0x02, 0x58, 0, 0, 0x1c, 0x20, + ]; + assert_eq!( + decode_one(&end_of_data), + ( + 1, + Pdu::EndOfData { + session_id: 0x1234, + serial: 99, + timers: Some(Timers::default()), + } + ) + ); + // Version 0 has no timers. + assert_eq!( + decode_one(&[0, 7, 0x12, 0x34, 0, 0, 0, 12, 0, 0, 0, 99]), + ( + 0, + Pdu::EndOfData { + session_id: 0x1234, + serial: 99, + timers: None, + } + ) + ); + #[rustfmt::skip] + let error = [ + 1, 10, 0, 2, 0, 0, 0, 30, + 0, 0, 0, 8, 1, 1, 0, 0, 0, 0, 0, 12, + 0, 0, 0, 6, b'n', b'o', b' ', b'd', b'a', b't', + ]; + assert_eq!( + decode_one(&error), + ( + 1, + Pdu::ErrorReport { + code: error_code::NO_DATA_AVAILABLE, + text: "no dat".into(), + } + ) + ); + } + + #[test] + fn a_pdu_that_has_not_all_arrived_is_waited_for() { + let mut wire = BytesMut::new(); + Pdu::SerialNotify { + session_id: 7, + serial: 42, + } + .encode(VERSION_1, &mut wire); + for cut in 0..wire.len() { + assert_eq!(Pdu::decode(&wire[..cut]), Ok(None), "{cut} octets"); + } + assert!(Pdu::decode(&wire).unwrap().is_some()); + } + + #[test] + fn what_is_not_a_pdu_is_refused_from_its_header() { + // A version from the future. + assert_eq!( + Pdu::decode(&[2, 2, 0, 0, 0, 0, 0, 8]), + Err(RtrError::Version(2)) + ); + assert_eq!( + Pdu::decode(&[1, 99, 0, 0, 0, 0, 0, 8]), + Err(RtrError::UnknownType(99)) + ); + // A Reset Query that claims to be long, and a length no PDU has: + // refused without waiting for the octets. + for length in [9u32, 7, 0, 1 << 30] { + let mut header = vec![1, 2, 0, 0]; + header.extend_from_slice(&length.to_be_bytes()); + assert_eq!( + Pdu::decode(&header), + Err(RtrError::Length { + pdu_type: 2, + length + }), + ); + } + // A maximum length shorter than the prefix, and longer than an + // address. + for (length, max_length) in [(24, 21), (24, 33)] { + #[rustfmt::skip] + let prefix = [ + 1, 4, 0, 0, 0, 0, 0, 20, + 1, length, max_length, 0, 193, 0, 0, 0, 0, 0, 0, 1, + ]; + assert_eq!( + Pdu::decode(&prefix), + Err(RtrError::Prefix { length, max_length }) + ); + } + // An Error Report whose inner lengths do not add up. + #[rustfmt::skip] + let error = [1, 10, 0, 2, 0, 0, 0, 16, 0, 0, 0, 9, 0, 0, 0, 0]; + assert!(matches!(Pdu::decode(&error), Err(RtrError::Length { .. }))); + } + + fn any_prefix() -> impl Strategy { + ( + any::(), + any::<[u8; 16]>(), + 0u8..=128, + 0u8..=128, + any::(), + ) + .prop_map(|(v6, bytes, length, extra, asn)| { + let version = if v6 { IpVersion::Ipv6 } else { IpVersion::Ipv4 }; + let length = length.min(version.bits()); + Vrp { + prefix: IpPrefix::new(version, bytes, length).unwrap(), + max_length: length.saturating_add(extra).min(version.bits()), + asn, + } + }) + } + + fn any_pdu() -> impl Strategy { + prop_oneof![ + (any::(), any::()) + .prop_map(|(session_id, serial)| Pdu::SerialNotify { session_id, serial }), + (any::(), any::()) + .prop_map(|(session_id, serial)| Pdu::SerialQuery { session_id, serial }), + Just(Pdu::ResetQuery), + any::().prop_map(|session_id| Pdu::CacheResponse { session_id }), + (any::(), any_prefix()).prop_map(|(announce, vrp)| Pdu::Prefix { announce, vrp }), + (any::(), any::(), any::<[u32; 3]>()).prop_map( + |(session_id, serial, [refresh, retry, expire])| Pdu::EndOfData { + session_id, + serial, + timers: Some(Timers { + refresh, + retry, + expire + }), + } + ), + Just(Pdu::CacheReset), + (any::(), "[ -~]{0,60}").prop_map(|(code, text)| Pdu::ErrorReport { code, text }), + ] + } + + proptest! { + #[test] + fn a_pdu_that_is_written_reads_back_equal(pdu in any_pdu(), version in 0u8..=1) { + let mut wire = BytesMut::new(); + pdu.encode(version, &mut wire); + // Version 0 does not carry the timers. + let expected = match (&pdu, version) { + (Pdu::EndOfData { session_id, serial, .. }, VERSION_0) => Pdu::EndOfData { + session_id: *session_id, + serial: *serial, + timers: None, + }, + _ => pdu.clone(), + }; + prop_assert_eq!(Pdu::decode(&wire), Ok(Some((version, expected, wire.len())))); + } + + /// Whatever arrives, decoding returns: a PDU, a wait for more, or + /// an error. What it reads is within what arrived. + #[test] + fn noise_is_never_more_than_an_error(bytes in proptest::collection::vec(any::(), 0..80)) { + if let Ok(Some((_, _, used))) = Pdu::decode(&bytes) { + prop_assert!(used <= bytes.len()); + } + } + + /// Noise behind a header that could be a PDU's, to get past it. + #[test] + fn noise_behind_a_plausible_header_is_never_more_than_an_error( + version in 0u8..=1, + pdu_type in 0u8..12, + field: u16, + length in 8u32..64, + body in proptest::collection::vec(any::(), 0..64), + ) { + let mut bytes = vec![version, pdu_type]; + bytes.extend_from_slice(&field.to_be_bytes()); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.extend_from_slice(&body); + if let Ok(Some((_, _, used))) = Pdu::decode(&bytes) { + prop_assert!(used <= bytes.len()); + } + } + } +} diff --git a/crates/rpki/src/vrp.rs b/crates/rpki/src/vrp.rs new file mode 100644 index 0000000..834b07b --- /dev/null +++ b/crates/rpki/src/vrp.rs @@ -0,0 +1,429 @@ +//! The validated ROA payloads, and what they say about a route. + +use bgp_primitives::constants::IpVersion; +use bgp_primitives::ip_prefix::IpPrefix; + +/// A validated ROA payload: `asn` may originate `prefix`, and any prefix +/// inside it of at most `max_length` bits. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct Vrp { + pub prefix: IpPrefix, + pub max_length: u8, + pub asn: u32, +} + +/// What the VRPs say about a route (RFC 6811 §2). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum Validity { + /// No VRP covers the route's prefix: nothing is known about it. Most + /// of the Internet's routes are in this state, and are used. + NotFound, + /// A VRP covers the prefix and allows this origin at this length. + Valid, + /// VRPs cover the prefix and none of them allows this origin at this + /// length: the holder of the addresses has said who may announce + /// them, and this is not it. + Invalid, +} + +impl Validity { + pub fn as_str(self) -> &'static str { + match self { + Self::NotFound => "not found", + Self::Valid => "valid", + Self::Invalid => "invalid", + } + } +} + +/// A set of VRPs, arranged for asking about routes: this is asked for +/// every route received. +/// +/// It is built once and not changed. When the VRPs change, another is built +/// and takes its place. +#[derive(Debug, Default)] +pub struct VrpTable { + v4: Trie, + v6: Trie, + len: usize, +} + +/// A binary trie over prefixes: a node for each prefix that is a VRP's or +/// lies on the way to one. +#[derive(Debug)] +struct Trie { + nodes: Vec, + /// What the VRPs of each node allow, in runs by node: (maximum length, + /// origin AS). + allowed: Vec<(u8, u32)>, + /// A short cut past the first sixteen levels, which every lookup of a + /// longer prefix would otherwise walk: for each value of an address's + /// first sixteen bits, the node those bits lead to, or zero if there + /// is none. [`WALK`] is set where a VRP lies on the way there, which + /// is rare; such lookups walk from the root. + jump: Box<[u32]>, +} + +/// How many leading bits [`Trie::jump`] is indexed by. +const JUMP_BITS: u8 = 16; +/// Set in a [`Trie::jump`] entry: walk from the root. +const WALK: u32 = 1 << 31; + +#[derive(Debug, Clone, Copy, Default)] +struct Node { + /// The nodes for this prefix extended by a zero bit and by a one bit; + /// zero where there is none (the root is nobody's child). + children: [u32; 2], + /// Where in `allowed` this prefix's VRPs are, and how many. + first: u32, + count: u32, +} + +impl Default for Trie { + fn default() -> Self { + Self { + nodes: vec![Node::default()], + allowed: Vec::new(), + jump: vec![0; 1 << JUMP_BITS].into(), + } + } +} + +/// The bit of `bytes` at `index`, counting from the most significant bit +/// of the first byte. +fn bit(bytes: &[u8], index: u8) -> usize { + usize::from(bytes[usize::from(index / 8)] >> (7 - index % 8) & 1) +} + +impl Trie { + /// `vrps` must be sorted, so that those of one prefix are together. + fn build(vrps: &[Vrp]) -> Self { + let mut trie = Self::default(); + for group in vrps.chunk_by(|a, b| a.prefix == b.prefix) { + let prefix = &group[0].prefix; + let mut at = 0usize; + for index in 0..prefix.length() { + let side = bit(prefix.bytes(), index); + at = match trie.nodes[at].children[side] { + 0 => { + trie.nodes.push(Node::default()); + let new = trie.nodes.len() - 1; + trie.nodes[at].children[side] = new as u32; + new + } + child => child as usize, + }; + } + trie.nodes[at].first = trie.allowed.len() as u32; + trie.nodes[at].count = group.len() as u32; + trie.allowed + .extend(group.iter().map(|vrp| (vrp.max_length, vrp.asn))); + } + assert!( + trie.nodes.len() < WALK as usize, + "more VRPs than a table can hold" + ); + for slot in 0..=u16::MAX { + trie.jump[usize::from(slot)] = trie.jump_for(slot); + } + trie + } + + /// The [`Trie::jump`] entry for addresses that begin with `bits`. + fn jump_for(&self, bits: u16) -> u32 { + let bytes = bits.to_be_bytes(); + let mut at = 0usize; + for index in 0..JUMP_BITS { + if self.nodes[at].count > 0 { + return WALK; + } + match self.nodes[at].children[bit(&bytes, index)] { + 0 => return 0, + child => at = child as usize, + } + } + at as u32 + } + + /// Calls `visit` with what the VRPs of each prefix that covers `prefix` + /// allow, from the shortest; stops when `visit` returns true, and + /// returns whether it did. + fn covering(&self, prefix: &IpPrefix, mut visit: impl FnMut(&[(u8, u32)]) -> bool) -> bool { + let bytes = prefix.bytes(); + let mut at = 0usize; + let mut index = 0; + if let [first, second, ..] = bytes + && prefix.length() >= JUMP_BITS + { + match self.jump[usize::from(u16::from_be_bytes([*first, *second]))] { + WALK => {} + // Nothing covers an address that begins like this. + 0 => return false, + node => (at, index) = (node as usize, JUMP_BITS), + } + } + loop { + let node = &self.nodes[at]; + if node.count > 0 { + let first = node.first as usize; + if visit(&self.allowed[first..first + node.count as usize]) { + return true; + } + } + if index == prefix.length() { + return false; + } + match node.children[bit(bytes, index)] { + 0 => return false, + child => at = child as usize, + } + index += 1; + } + } +} + +impl VrpTable { + /// A table of `vrps`. Listing one more than once is the same as listing + /// it once. + pub fn new(vrps: impl IntoIterator) -> Self { + let mut vrps: Vec = vrps.into_iter().collect(); + vrps.sort_unstable(); + vrps.dedup(); + let v6_from = vrps.partition_point(|vrp| vrp.prefix.address_family() == IpVersion::Ipv4); + debug_assert!( + vrps[..v6_from] + .iter() + .all(|vrp| vrp.prefix.address_family() == IpVersion::Ipv4) + ); + Self { + v4: Trie::build(&vrps[..v6_from]), + v6: Trie::build(&vrps[v6_from..]), + len: vrps.len(), + } + } + + /// How many VRPs the table holds. + pub fn len(&self) -> usize { + self.len + } + + pub fn is_empty(&self) -> bool { + self.len == 0 + } + + fn trie(&self, prefix: &IpPrefix) -> &Trie { + match prefix.address_family() { + IpVersion::Ipv4 => &self.v4, + IpVersion::Ipv6 => &self.v6, + } + } + + /// The validity of a route for `prefix` that originates from `origin` + /// (RFC 6811 §2). `None` is a route with no one origin, as when its AS + /// path ends in a set: no VRP can allow it. + pub fn validate(&self, prefix: &IpPrefix, origin: Option) -> Validity { + let mut covered = false; + let valid = self.trie(prefix).covering(prefix, |allowed| { + covered = true; + // AS 0 in a VRP says that nobody may originate the prefix + // (RFC 6483 §4); it is the origin of no route. + origin.is_some_and(|origin| { + origin != 0 + && allowed + .iter() + .any(|(max_length, asn)| *asn == origin && prefix.length() <= *max_length) + }) + }); + match (valid, covered) { + (true, _) => Validity::Valid, + (false, true) => Validity::Invalid, + (false, false) => Validity::NotFound, + } + } + + /// Whether some VRP covers `prefix`: whether the VRPs say anything + /// about routes for it. + pub fn covers(&self, prefix: &IpPrefix) -> bool { + self.trie(prefix).covering(prefix, |_| true) + } +} + +#[cfg(test)] +mod tests { + use proptest::prelude::*; + + use super::*; + + fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() + } + + fn vrp(p: &str, max_length: u8, asn: u32) -> Vrp { + Vrp { + prefix: prefix(p), + max_length, + asn, + } + } + + /// The cases of RFC 6811 §2. + #[test] + fn a_route_is_valid_invalid_or_not_found() { + let table = VrpTable::new([ + vrp("193.0.0.0/21", 24, 3333), + vrp("193.0.0.0/21", 21, 2121), + vrp("2001:67c:2e8::/48", 48, 3333), + ]); + let validate = |p: &str, origin| table.validate(&prefix(p), Some(origin)); + + // Covered, by the right AS, no longer than allowed. + assert_eq!(validate("193.0.0.0/21", 3333), Validity::Valid); + assert_eq!(validate("193.0.4.0/24", 3333), Validity::Valid); + assert_eq!(validate("193.0.0.0/21", 2121), Validity::Valid); + assert_eq!(validate("2001:67c:2e8::/48", 3333), Validity::Valid); + // The right AS, but more specific than it may announce. + assert_eq!(validate("193.0.4.0/24", 2121), Validity::Invalid); + assert_eq!(validate("193.0.4.0/25", 3333), Validity::Invalid); + // Someone else's AS: a hijack, or a mistake. + assert_eq!(validate("193.0.0.0/21", 15169), Validity::Invalid); + assert_eq!(validate("2001:67c:2e8::/48", 15169), Validity::Invalid); + // Nothing covers these: a shorter prefix, another block, another + // family. + assert_eq!(validate("193.0.0.0/20", 3333), Validity::NotFound); + assert_eq!(validate("194.0.0.0/21", 3333), Validity::NotFound); + assert_eq!(validate("2001:67c:2e9::/48", 3333), Validity::NotFound); + + // A route with no one origin matches no VRP. + assert_eq!( + table.validate(&prefix("193.0.0.0/21"), None), + Validity::Invalid + ); + assert_eq!( + table.validate(&prefix("194.0.0.0/21"), None), + Validity::NotFound + ); + assert_eq!(table.len(), 3); + assert!(table.covers(&prefix("193.0.7.0/24"))); + assert!(!table.covers(&prefix("193.0.8.0/24"))); + } + + /// A VRP for AS 0 says the prefix is not to be routed at all. + #[test] + fn as_zero_allows_nobody() { + let table = VrpTable::new([vrp("203.0.113.0/24", 24, 0)]); + assert_eq!( + table.validate(&prefix("203.0.113.0/24"), Some(0)), + Validity::Invalid + ); + assert_eq!( + table.validate(&prefix("203.0.113.0/24"), Some(64500)), + Validity::Invalid + ); + } + + /// A VRP covers a route however many other VRPs cover it too: one that + /// allows it is enough. + #[test] + fn any_covering_vrp_can_make_a_route_valid() { + let table = VrpTable::new([ + vrp("10.0.0.0/8", 8, 1), + vrp("10.1.0.0/16", 24, 2), + vrp("0.0.0.0/0", 0, 3), + ]); + assert_eq!( + table.validate(&prefix("10.1.2.0/24"), Some(2)), + Validity::Valid + ); + assert_eq!( + table.validate(&prefix("10.1.2.0/24"), Some(1)), + Validity::Invalid + ); + assert_eq!( + table.validate(&prefix("10.0.0.0/8"), Some(1)), + Validity::Valid + ); + assert_eq!( + table.validate(&prefix("0.0.0.0/0"), Some(3)), + Validity::Valid + ); + // The VRP for the default route covers everything. + assert_eq!( + table.validate(&prefix("192.0.2.0/24"), Some(9)), + Validity::Invalid + ); + } + + #[test] + fn an_empty_table_knows_nothing() { + let table = VrpTable::default(); + assert!(table.is_empty()); + assert_eq!( + table.validate(&prefix("193.0.0.0/21"), Some(3333)), + Validity::NotFound + ); + assert_eq!(table.validate(&prefix("::/0"), None), Validity::NotFound); + } + + fn any_vrp() -> impl Strategy { + // Few distinct addresses and AS numbers, so that routes and VRPs + // meet. + (any::(), 0u8..4, 0u8..=24, 0u8..=8, 0u32..4).prop_map( + |(v6, address, length, extra, asn)| { + let version = if v6 { IpVersion::Ipv6 } else { IpVersion::Ipv4 }; + let bytes = [ + address << 6 | address, + address, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + ]; + Vrp { + prefix: IpPrefix::new(version, bytes, length).unwrap(), + max_length: length + extra, + asn, + } + }, + ) + } + + proptest! { + /// The table agrees with reading RFC 6811 out over the list. + #[test] + fn the_table_agrees_with_looking_through_the_list( + vrps in proptest::collection::vec(any_vrp(), 0..30), + routes in proptest::collection::vec((any_vrp(), any::()), 1..40), + ) { + let table = VrpTable::new(vrps.iter().copied()); + for (route, has_origin) in routes { + let origin = has_origin.then_some(route.asn); + let covering: Vec<&Vrp> = vrps + .iter() + .filter(|vrp| vrp.prefix.contains(&route.prefix)) + .collect(); + let expected = if covering.is_empty() { + Validity::NotFound + } else if covering.iter().any(|vrp| { + origin.is_some_and(|origin| origin != 0 && origin == vrp.asn) + && route.prefix.length() <= vrp.max_length + }) { + Validity::Valid + } else { + Validity::Invalid + }; + prop_assert_eq!(table.validate(&route.prefix, origin), expected, "{:?}", route); + prop_assert_eq!(table.covers(&route.prefix), !covering.is_empty()); + } + } + } +} diff --git a/crates/rpki/tests/client.rs b/crates/rpki/tests/client.rs new file mode 100644 index 0000000..4358f7c --- /dev/null +++ b/crates/rpki/tests/client.rs @@ -0,0 +1,507 @@ +//! The RTR client against a cache played by the test: what it asks for, +//! what it makes of the answers, and what it does when the cache misbehaves +//! or goes away. + +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rpki::rtr::{Pdu, Timers, error_code}; +use bgp_rpki::{CacheState, Rpki, RpkiConfig, Update, Validity, Vrp}; +use bytes::BytesMut; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::watch; + +const TIMEOUT: Duration = Duration::from_secs(5); +const SESSION: u16 = 0x1234; + +fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() +} + +fn vrp(p: &str, max_length: u8, asn: u32) -> Vrp { + Vrp { + prefix: prefix(p), + max_length, + asn, + } +} + +fn announce(vrp: Vrp) -> Pdu { + Pdu::Prefix { + announce: true, + vrp, + } +} + +fn withdraw(vrp: Vrp) -> Pdu { + Pdu::Prefix { + announce: false, + vrp, + } +} + +fn end_of_data(serial: u32) -> Pdu { + Pdu::EndOfData { + session_id: SESSION, + serial, + timers: Some(Timers::default()), + } +} + +/// A cache played by the test. +struct Cache { + listener: TcpListener, +} + +/// One connection from the client to the cache. +struct Connection { + stream: TcpStream, + buf: BytesMut, +} + +impl Cache { + async fn start() -> Self { + Self { + listener: TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(), + } + } + + fn address(&self) -> SocketAddr { + self.listener.local_addr().unwrap() + } + + async fn accept(&self) -> Connection { + let (stream, _) = tokio::time::timeout(TIMEOUT, self.listener.accept()) + .await + .expect("the client connects") + .unwrap(); + Connection { + stream, + buf: BytesMut::new(), + } + } +} + +impl Connection { + /// The next PDU from the client and the version it was sent in. + async fn received(&mut self) -> (u8, Pdu) { + let reading = async { + loop { + if let Some((version, pdu, used)) = Pdu::decode(&self.buf).unwrap() { + let _ = self.buf.split_to(used); + return (version, pdu); + } + assert_ne!( + self.stream.read_buf(&mut self.buf).await.unwrap(), + 0, + "the client closed the connection" + ); + } + }; + tokio::time::timeout(TIMEOUT, reading) + .await + .expect("the client sends something") + } + + async fn send(&mut self, version: u8, pdus: &[Pdu]) { + let mut out = BytesMut::new(); + for pdu in pdus { + pdu.encode(version, &mut out); + } + self.stream.write_all(&out).await.unwrap(); + } + + /// Answers a query with everything: `vrps`, at `serial`. + async fn send_all(&mut self, vrps: &[Vrp], serial: u32) { + let mut pdus = vec![Pdu::CacheResponse { + session_id: SESSION, + }]; + pdus.extend(vrps.iter().copied().map(announce)); + pdus.push(end_of_data(serial)); + self.send(1, &pdus).await; + } +} + +fn client_of(caches: &[&Cache]) -> (Rpki, watch::Receiver) { + let (rpki, _task) = Rpki::spawn(RpkiConfig { + caches: caches.iter().map(|cache| cache.address()).collect(), + retry: Some(Duration::from_millis(50)), + ..RpkiConfig::default() + }); + let updates = rpki.subscribe(); + (rpki, updates) +} + +async fn next_update(updates: &mut watch::Receiver) -> Update { + tokio::time::timeout(TIMEOUT, updates.changed()) + .await + .expect("the VRPs change") + .unwrap(); + updates.borrow_and_update().clone() +} + +/// The ordinary life of a session: everything once, then what changes. +#[tokio::test] +async fn the_client_takes_everything_once_and_then_the_changes() { + let cache = Cache::start().await; + let (rpki, mut updates) = client_of(&[&cache]); + // Nothing is known until a cache has answered. + assert_eq!(rpki.version(), 0); + assert!(rpki.current().table.is_empty()); + + let mut connection = cache.accept().await; + assert_eq!(connection.received().await, (1, Pdu::ResetQuery)); + let ripe = vrp("193.0.0.0/21", 24, 3333); + let other = vrp("2001:67c:2e8::/48", 48, 3333); + connection.send_all(&[ripe, other], 7).await; + + let update = next_update(&mut updates).await; + assert_eq!(update.version, 1); + assert_eq!(rpki.version(), 1); + assert!( + update.changed.is_none(), + "the first data can change anything" + ); + assert_eq!(update.table.len(), 2); + assert_eq!( + update.table.validate(&prefix("193.0.4.0/24"), Some(3333)), + Validity::Valid + ); + assert_eq!( + update.table.validate(&prefix("193.0.4.0/24"), Some(15169)), + Validity::Invalid + ); + let status = rpki.status(); + assert_eq!(status.state, CacheState::Synchronised); + assert_eq!((status.session_id, status.serial), (Some(SESSION), Some(7))); + assert_eq!((status.vrps_ipv4, status.vrps_ipv6), (1, 1)); + assert_eq!(status.cache, Some(cache.address())); + + // The cache has news: the client asks for what changed since serial 7. + connection + .send( + 1, + &[Pdu::SerialNotify { + session_id: SESSION, + serial: 8, + }], + ) + .await; + assert_eq!( + connection.received().await, + ( + 1, + Pdu::SerialQuery { + session_id: SESSION, + serial: 7 + } + ) + ); + let new = vrp("8.8.8.0/24", 24, 15169); + connection + .send( + 1, + &[ + Pdu::CacheResponse { + session_id: SESSION, + }, + announce(new), + withdraw(other), + end_of_data(8), + ], + ) + .await; + + let update = next_update(&mut updates).await; + assert_eq!(update.version, 2); + assert_eq!(update.table.len(), 2); + assert_eq!( + update.table.validate(&prefix("8.8.8.0/24"), Some(15169)), + Validity::Valid + ); + assert_eq!( + update + .table + .validate(&prefix("2001:67c:2e8::/48"), Some(3333)), + Validity::NotFound, + "withdrawn" + ); + // Only routes that what came and went covers can have changed. + let changed = update.changed.expect("a change to what there was"); + assert!(changed.covers(&prefix("8.8.8.0/24"))); + assert!(changed.covers(&prefix("2001:67c:2e8::/48"))); + assert!(!changed.covers(&prefix("193.0.0.0/21"))); + assert_eq!(rpki.status().serial, Some(8)); +} + +/// A cache that cannot say what changed tells the client to start over. +/// What the client then publishes is the difference to what it had. +#[tokio::test] +async fn a_cache_reset_makes_the_client_ask_for_everything() { + let cache = Cache::start().await; + let (rpki, mut updates) = client_of(&[&cache]); + let mut connection = cache.accept().await; + connection.received().await; + let kept = vrp("193.0.0.0/21", 24, 3333); + let dropped = vrp("8.8.8.0/24", 24, 15169); + connection.send_all(&[kept, dropped], 7).await; + next_update(&mut updates).await; + + connection + .send( + 1, + &[Pdu::SerialNotify { + session_id: SESSION, + serial: 20, + }], + ) + .await; + assert!(matches!( + connection.received().await.1, + Pdu::SerialQuery { serial: 7, .. } + )); + connection.send(1, &[Pdu::CacheReset]).await; + assert_eq!(connection.received().await.1, Pdu::ResetQuery); + let added = vrp("1.1.1.0/24", 24, 13335); + connection.send_all(&[kept, added], 20).await; + + let update = next_update(&mut updates).await; + assert_eq!(update.table.len(), 2); + let changed = update.changed.expect("known what changed"); + assert!(changed.covers(&prefix("8.8.8.0/24"))); + assert!(changed.covers(&prefix("1.1.1.0/24"))); + assert!(!changed.covers(&prefix("193.0.0.0/21")), "it stayed"); + assert_eq!(rpki.status().serial, Some(20)); +} + +/// An answer takes effect whole or not at all: one that contradicts itself +/// changes nothing, and the session is given up. +#[tokio::test] +async fn an_answer_that_contradicts_itself_changes_nothing() { + let cache = Cache::start().await; + let (rpki, mut updates) = client_of(&[&cache]); + let mut connection = cache.accept().await; + connection.received().await; + let held = vrp("193.0.0.0/21", 24, 3333); + connection.send_all(&[held], 7).await; + next_update(&mut updates).await; + + connection + .send( + 1, + &[Pdu::SerialNotify { + session_id: SESSION, + serial: 8, + }], + ) + .await; + connection.received().await; + // A good change, then the withdrawal of something never announced. + connection + .send( + 1, + &[ + Pdu::CacheResponse { + session_id: SESSION, + }, + announce(vrp("1.1.1.0/24", 24, 13335)), + withdraw(vrp("9.9.9.0/24", 24, 19281)), + end_of_data(8), + ], + ) + .await; + + // The client gives the session up and comes back; what it has is + // what it had, and it asks from where it was. + let mut again = cache.accept().await; + assert_eq!( + again.received().await.1, + Pdu::SerialQuery { + session_id: SESSION, + serial: 7 + } + ); + let current = rpki.current(); + assert_eq!(current.version, 1); + assert_eq!(current.table.len(), 1); + assert!( + rpki.status() + .last_error + .is_some_and(|e| e.contains("withdrew a VRP it had not announced")) + ); +} + +/// A validator that only speaks version 0 says so, by an error or by +/// answering in version 0. The client asks again in version 0. +#[tokio::test] +async fn a_cache_that_speaks_version_zero_is_spoken_to_in_it() { + for refuses_with_an_error in [true, false] { + let cache = Cache::start().await; + let (rpki, mut updates) = client_of(&[&cache]); + let mut connection = cache.accept().await; + assert_eq!(connection.received().await, (1, Pdu::ResetQuery)); + if refuses_with_an_error { + connection + .send( + 0, + &[Pdu::ErrorReport { + code: error_code::UNSUPPORTED_VERSION, + text: "version 0 only".into(), + }], + ) + .await; + } else { + connection + .send( + 0, + &[Pdu::CacheResponse { + session_id: SESSION, + }], + ) + .await; + } + + let mut connection = cache.accept().await; + assert_eq!(connection.received().await, (0, Pdu::ResetQuery)); + connection + .send( + 0, + &[ + Pdu::CacheResponse { + session_id: SESSION, + }, + announce(vrp("193.0.0.0/21", 24, 3333)), + Pdu::EndOfData { + session_id: SESSION, + serial: 3, + timers: None, + }, + ], + ) + .await; + assert_eq!(next_update(&mut updates).await.table.len(), 1); + assert_eq!(rpki.status().protocol_version, 0); + } +} + +/// When the cache goes away the data is kept and used while the client +/// tries again, and the next cache is tried. A cache the data did not come +/// from is asked for everything. +#[tokio::test] +async fn a_lost_cache_is_replaced_by_the_next_and_the_data_is_kept() { + let first = Cache::start().await; + let second = Cache::start().await; + let (rpki, mut updates) = client_of(&[&first, &second]); + + let mut connection = first.accept().await; + connection.received().await; + connection + .send_all(&[vrp("193.0.0.0/21", 24, 3333)], 7) + .await; + next_update(&mut updates).await; + + drop(connection); + drop(first); + let mut connection = second.accept().await; + assert_eq!(connection.received().await.1, Pdu::ResetQuery); + // Still in use meanwhile. + assert_eq!(rpki.current().table.len(), 1); + + connection + .send_all( + &[vrp("193.0.0.0/21", 24, 3333), vrp("1.1.1.0/24", 24, 13335)], + 100, + ) + .await; + let update = next_update(&mut updates).await; + assert_eq!(update.table.len(), 2); + let changed = update.changed.expect("the difference to what there was"); + assert!(changed.covers(&prefix("1.1.1.0/24"))); + assert!(!changed.covers(&prefix("193.0.0.0/21"))); + assert_eq!(rpki.status().cache, Some(second.address())); +} + +/// Data that no cache has confirmed for the expire interval is dropped: +/// routes are then neither valid nor invalid. +#[tokio::test] +async fn data_that_is_not_confirmed_in_time_is_dropped() { + let cache = Cache::start().await; + let (rpki, _task) = Rpki::spawn(RpkiConfig { + caches: vec![cache.address()], + retry: Some(Duration::from_millis(50)), + expire: Some(Duration::from_millis(300)), + ..RpkiConfig::default() + }); + let mut updates = rpki.subscribe(); + let mut connection = cache.accept().await; + connection.received().await; + connection + .send_all(&[vrp("193.0.0.0/21", 24, 3333)], 7) + .await; + assert_eq!(next_update(&mut updates).await.table.len(), 1); + + // The cache goes and does not come back. + drop(connection); + drop(cache); + let update = next_update(&mut updates).await; + assert!(update.table.is_empty()); + assert!(update.changed.is_none(), "anything may have changed"); + assert_eq!( + update.table.validate(&prefix("193.0.0.0/21"), Some(15169)), + Validity::NotFound + ); + let status = rpki.status(); + assert_eq!(status.state, CacheState::Connecting); + assert_eq!(status.serial, None); +} + +/// A cache that reports an error, or sends what is not RTR, is given up +/// on and tried again; nothing of what it sent is used. +#[tokio::test] +async fn a_cache_that_misbehaves_is_given_up_on() { + let cache = Cache::start().await; + let (rpki, _updates) = client_of(&[&cache]); + + let mut connection = cache.accept().await; + connection.received().await; + connection + .send( + 1, + &[Pdu::ErrorReport { + code: error_code::NO_DATA_AVAILABLE, + text: "still validating".into(), + }], + ) + .await; + let mut connection = cache.accept().await; + assert_eq!(connection.received().await.1, Pdu::ResetQuery); + assert!( + rpki.status() + .last_error + .is_some_and(|e| e.contains("still validating")) + ); + + // Half an answer, then noise. + connection + .send( + 1, + &[ + Pdu::CacheResponse { + session_id: SESSION, + }, + announce(vrp("193.0.0.0/21", 24, 3333)), + ], + ) + .await; + connection + .stream + .write_all(&[1, 99, 0, 0, 0, 0, 0, 8]) + .await + .unwrap(); + let mut connection = cache.accept().await; + assert_eq!(connection.received().await.1, Pdu::ResetQuery); + assert!(rpki.current().table.is_empty()); + assert_eq!(rpki.version(), 0); +} diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index 498713c..e4ff614 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -17,6 +17,7 @@ cargo-fuzz = true [dependencies] bgp-packet = { path = "../crates/packet" } bgp-primitives = { path = "../crates/primitives" } +bgp-rpki = { path = "../crates/rpki" } bgp-server = { path = "../crates/server" } bytes = "1.10.1" libfuzzer-sys = "0.4" @@ -52,3 +53,10 @@ path = "fuzz_targets/config.rs" test = false doc = false bench = false + +[[bin]] +name = "rtr" +path = "fuzz_targets/rtr.rs" +test = false +doc = false +bench = false diff --git a/fuzz/README.md b/fuzz/README.md index a62241b..7e0c481 100644 --- a/fuzz/README.md +++ b/fuzz/README.md @@ -12,6 +12,7 @@ you leave them running. | `message_body` | A message body behind a correct header (reaches the body parsers) | | `prefix_text` | A prefix as text | | `config` | The configuration file | +| `rtr` | A stream of RTR PDUs, as an RPKI cache sends them | For the two message targets a finding is either a panic, or a message the decoder accepted that cannot be written and read back as itself. diff --git a/fuzz/fuzz_targets/rtr.rs b/fuzz/fuzz_targets/rtr.rs new file mode 100644 index 0000000..a8f19df --- /dev/null +++ b/fuzz/fuzz_targets/rtr.rs @@ -0,0 +1,27 @@ +//! Bytes from an RPKI cache: a stream of RTR PDUs. A cache is trusted for +//! its data, not to be free of bugs or to be the cache we meant to reach. +#![no_main] + +use bgp_rpki::rtr::Pdu; +use bytes::BytesMut; +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + let mut rest = data; + // As the client reads: one PDU after another until one is incomplete + // or wrong. + while let Ok(Some((version, pdu, used))) = Pdu::decode(rest) { + assert!(used >= 8 && used <= rest.len()); + // What was read can be shown and written. + let _ = format!("{pdu:?}"); + let mut out = BytesMut::new(); + pdu.encode(version, &mut out); + // And what is written reads back as itself. A router key is read + // only to be skipped, and an error report is written without the + // PDU it complains about. + if !matches!(pdu, Pdu::RouterKey | Pdu::ErrorReport { .. }) { + assert_eq!(Pdu::decode(&out), Ok(Some((version, pdu, out.len())))); + } + rest = &rest[used..]; + } +}); diff --git a/fuzz/seeds/rtr/error b/fuzz/seeds/rtr/error new file mode 100644 index 0000000..604427a Binary files /dev/null and b/fuzz/seeds/rtr/error differ diff --git a/fuzz/seeds/rtr/full_answer b/fuzz/seeds/rtr/full_answer new file mode 100644 index 0000000..4a1eba6 Binary files /dev/null and b/fuzz/seeds/rtr/full_answer differ diff --git a/fuzz/seeds/rtr/notify b/fuzz/seeds/rtr/notify new file mode 100644 index 0000000..cb83cdf Binary files /dev/null and b/fuzz/seeds/rtr/notify differ