diff --git a/Cargo.lock b/Cargo.lock index f6bdaf6..77fc4e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -192,6 +192,7 @@ dependencies = [ "bgp-packet", "bgp-primitives", "bgp-rib", + "bgp-rpki", "bgp-server", "bgp-web", "bytes", @@ -267,6 +268,7 @@ dependencies = [ "bgp-packet", "bgp-primitives", "bgp-rib", + "bgp-rpki", "bytes", "clap", "futures", @@ -289,6 +291,7 @@ dependencies = [ "bgp-packet", "bgp-primitives", "bgp-rib", + "bgp-rpki", "bgp-server", "futures", "prometheus", diff --git a/bin/bgp/README.md b/bin/bgp/README.md index 077435d..3eed7e6 100644 --- a/bin/bgp/README.md +++ b/bin/bgp/README.md @@ -23,6 +23,7 @@ When `http_endpoint` is set the binary also serves: | `GET /api/routes/local` | Originated routes | | `GET /api/routes/received` | Routes learned from peers; see "Finding a route" | | `GET /api/routes/rejected` | Routes peers announced that were refused, by peer | +| `GET /api/rpki` | Route origin validation: the cache, the data and its age | | `GET /api/events` | Server-sent events for session and route changes | | `GET /metricz` | Prometheus metrics | @@ -64,6 +65,34 @@ A refused route is dropped, along with the route it would have replaced. `bgp_routes_rejected_total` counts them by reason, and the status page shows how many routes each peer has and how many were refused. +## Route origin validation (RPKI) + +With `rpki { caches { address: "..." port: 3323 } }` in the configuration, +the daemon keeps the validated ROA payloads of a validator (Routinator, +rpki-client with StayRTR, FORT) over the RPKI-to-Router protocol, and +validates every route from peers outside its own AS (RFC 6811): + +- **invalid** (the holder of the addresses authorised another origin, or a + shorter prefix): refused, like any route the import policy refuses, unless + the peer has `accept_rpki_invalid: true`; +- **valid** and **not found**: taken. Validity does not otherwise change + which route is chosen. + +The data changes all the time, and the routes follow without the peer having +to send them again: a route in the table that becomes invalid is taken out, +and a route refused as invalid is kept aside ("held") and taken when it no +longer is. If no cache confirms the data for the expire interval (two hours +unless the cache or the configuration says otherwise) it is dropped, and +nothing is invalid any more. + +The status page shows the cache, how fresh the data is, each route's +validity, and how many of each peer's routes are held. `/api/rpki` has the +same as JSON; `bgp_rpki_vrps{family}` and `bgp_rpki_updates_total` are on +`/metricz`. Refused routes are counted under the reason `RPKI invalid`. + +The connection to the cache is plain TCP, without authentication: run the +validator on the same host or on a network you trust. + ## Finding a route The status page has a look-up box over the received routes, and diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 4678e23..2f327be 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -51,6 +51,7 @@ async fn main() -> Result<()> { local_rib: server.local_rib.clone(), rib: Arc::clone(&server.rib), refusals: server.refusals.clone(), + rpki: server.rpki.clone(), }; tokio::spawn(async move { if let Err(e) = bgp_web::serve_on(state, listener).await { diff --git a/crates/integration-tests/Cargo.toml b/crates/integration-tests/Cargo.toml index 7d0e14e..687ef08 100644 --- a/crates/integration-tests/Cargo.toml +++ b/crates/integration-tests/Cargo.toml @@ -12,6 +12,7 @@ publish = false bgp-packet.workspace = true bgp-primitives.workspace = true bgp-rib.workspace = true +bgp-rpki.workspace = true bgp-server.workspace = true bgp-web.workspace = true bytes.workspace = true diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs index cedbe12..694548d 100644 --- a/crates/integration-tests/src/lib.rs +++ b/crates/integration-tests/src/lib.rs @@ -1,2 +1,3 @@ +pub mod test_cache; pub mod test_peer; pub mod test_server; diff --git a/crates/integration-tests/src/test_cache.rs b/crates/integration-tests/src/test_cache.rs new file mode 100644 index 0000000..0e99be0 --- /dev/null +++ b/crates/integration-tests/src/test_cache.rs @@ -0,0 +1,124 @@ +//! An RPKI cache for tests: an RTR server that hands out whatever VRPs the +//! test says there are. + +use std::net::{Ipv4Addr, SocketAddr}; +use std::sync::{Arc, Mutex}; + +use bgp_rpki::Vrp; +use bgp_rpki::rtr::{Pdu, Timers}; +use bytes::BytesMut; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::watch; +use tokio::task::JoinHandle; + +const SESSION: u16 = 0x1234; + +/// The cache. Stops, closing its connections, when dropped. +pub struct TestCache { + address: SocketAddr, + /// The VRPs there are, and their serial. + data: Arc, u32)>>, + /// The serial, for telling connected routers there is something new. + serial: watch::Sender, + task: JoinHandle<()>, +} + +impl TestCache { + pub async fn start(vrps: Vec) -> Self { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let address = listener.local_addr().unwrap(); + let data = Arc::new(Mutex::new((vrps, 1))); + let (serial, _) = watch::channel(1); + let task = tokio::spawn({ + let (data, serial) = (Arc::clone(&data), serial.clone()); + async move { + // Connections end with this task: they are part of it. + let mut connections = tokio::task::JoinSet::new(); + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + connections.spawn(serve(stream, Arc::clone(&data), serial.subscribe())); + } + } + }); + Self { + address, + data, + serial, + task, + } + } + + pub fn address(&self) -> SocketAddr { + self.address + } + + /// Replaces the VRPs and tells connected routers. + pub fn set(&self, vrps: Vec) { + let serial = { + let mut data = self.data.lock().unwrap(); + data.0 = vrps; + data.1 += 1; + data.1 + }; + self.serial.send_replace(serial); + } +} + +impl Drop for TestCache { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// One router's connection. Asked for changes, this cache always says to +/// ask for everything: the router works out what changed. +async fn serve( + mut stream: TcpStream, + data: Arc, u32)>>, + mut serial: watch::Receiver, +) { + let mut buf = BytesMut::new(); + loop { + let mut out = BytesMut::new(); + tokio::select! { + read = stream.read_buf(&mut buf) => { + if !matches!(read, Ok(read) if read > 0) { + return; + } + while let Ok(Some((version, pdu, used))) = Pdu::decode(&buf) { + let _ = buf.split_to(used); + match pdu { + Pdu::SerialQuery { .. } => Pdu::CacheReset.encode(version, &mut out), + Pdu::ResetQuery => { + let (vrps, serial) = data.lock().unwrap().clone(); + Pdu::CacheResponse { session_id: SESSION }.encode(version, &mut out); + for vrp in vrps { + Pdu::Prefix { announce: true, vrp }.encode(version, &mut out); + } + Pdu::EndOfData { + session_id: SESSION, + serial, + timers: Some(Timers::default()), + } + .encode(version, &mut out); + } + _ => return, + } + } + } + changed = serial.changed() => { + if changed.is_err() { + return; + } + let serial = *serial.borrow_and_update(); + Pdu::SerialNotify { session_id: SESSION, serial }.encode(1, &mut out); + } + } + if stream.write_all(&out).await.is_err() { + return; + } + } +} diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index e1b1f32..0673cd0 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -5,6 +5,7 @@ use bgp_packet::message::{NlriNextHop, PathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::LocalRib; use bgp_rib::table::{Event, PeerIndex, Reader, RibTable}; +use bgp_rpki::Rpki; use bgp_server::config::Config; use bgp_server::error::ServerError; use bgp_server::refusals::RefusalLog; @@ -29,6 +30,8 @@ pub struct TestServer { pub registry: SessionRegistry, /// What peers announced that was not taken. pub refusals: RefusalLog, + /// The data for route origin validation, if the server validates. + pub rpki: Option, pub local_asn: u32, } @@ -59,6 +62,7 @@ impl TestServer { accept_unconfigured_peers: true, originate: vec![], http_endpoint: None, + rpki: None, }; customize(&mut config); @@ -72,6 +76,7 @@ impl TestServer { let local_rib = bound.local_rib.clone(); let registry = bound.registry.clone(); let refusals = bound.refusals.clone(); + let rpki = bound.rpki.clone(); let handle = tokio::spawn(bound.run()); Ok(Self { @@ -81,6 +86,7 @@ impl TestServer { local_rib, registry, refusals, + rpki, local_asn, }) } @@ -199,6 +205,7 @@ impl TestServer { local_rib: self.local_rib.clone(), rib: Arc::clone(&self.rib), refusals: self.refusals.clone(), + rpki: self.rpki.clone(), }; let handle = tokio::spawn(bgp_web::serve_on(state, listener)); Ok((base_url, HttpGuard { handle })) diff --git a/crates/integration-tests/tests/rpki.rs b/crates/integration-tests/tests/rpki.rs new file mode 100644 index 0000000..80f1172 --- /dev/null +++ b/crates/integration-tests/tests/rpki.rs @@ -0,0 +1,340 @@ +//! Route origin validation, end to end: a server that validates against a +//! cache, a peer that announces routes, and the validation data changing +//! under both. + +use std::collections::BTreeSet; +use std::net::Ipv4Addr; +use std::time::Duration; + +use bgp_integration_tests::test_cache::TestCache; +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_packet::message::{ + AsPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute, PathAttribute, + UpdateMessage, +}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rpki::{CacheState, RpkiConfig, Vrp}; +use bgp_server::config::PeerConfig; +use bgp_server::family::FamilySet; +use bgp_server::policy::ImportPolicy; +use eyre::Result; +use serde_json::Value; +use tokio::net::TcpListener; + +const LOCAL_ASN: u32 = 1299; +const PEER_ASN: u32 = 3333; +const TIMEOUT: Duration = Duration::from_secs(5); + +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, + } +} + +/// What a test holds on to: the server, and the listener its dialling +/// goes to. +struct Setup { + server: TestServer, + _listener: TcpListener, +} + +/// A server that validates against `cache`, with one configured peer at +/// the address the test connects from. Returns once the server has the +/// cache's data. +async fn server_with( + cache: &TestCache, + expire: Option, + import: ImportPolicy, +) -> Result { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let endpoint = listener.local_addr()?; + let cache = cache.address(); + let server = TestServer::start_with(LOCAL_ASN, |c| { + c.accept_unconfigured_peers = false; + c.rpki = Some(RpkiConfig { + caches: vec![cache], + retry: Some(Duration::from_millis(100)), + expire, + ..RpkiConfig::default() + }); + c.peers = vec![PeerConfig { + name: "r1".into(), + endpoint, + remote_asn: PEER_ASN, + hold_time_secs: 90, + local_address: None, + families: FamilySet::supported(), + add_path: None, + import, + }]; + }) + .await?; + let rpki = server.rpki.clone().expect("validation is configured"); + tokio::time::timeout(TIMEOUT, async { + while rpki.status().state != CacheState::Synchronised { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await?; + Ok(Setup { + server, + _listener: listener, + }) +} + +async fn established(server: &TestServer) -> Result { + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; + tokio::time::timeout(TIMEOUT, async { + while server.registry.list().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await?; + Ok(peer) +} + +/// Announces `prefixes` as originated by `origin`, reached through the +/// peer. +async fn announce(peer: &mut TestPeer, prefixes: &[&str], origin: u32) -> Result<()> { + let path = if origin == PEER_ASN { + vec![PEER_ASN] + } else { + vec![PEER_ASN, origin] + }; + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(path), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + ], + announced_routes: prefixes.iter().map(|p| prefix(p).into()).collect(), + })) + .await +} + +fn set(prefixes: &[&str]) -> BTreeSet { + prefixes.iter().map(|p| p.to_string()).collect() +} + +/// Waits until the table holds exactly `expected` and the peer has `held` +/// routes held aside. +async fn table_becomes(server: &TestServer, expected: &[&str], held: u64) -> Result<()> { + let expected = set(expected); + let table = || -> BTreeSet { + server + .received_routes() + .iter() + .map(|route| route.prefix.to_string()) + .collect() + }; + let held_now = || { + server + .registry + .list() + .first() + .map_or(0, |session| session.counters.held()) + }; + tokio::time::timeout(TIMEOUT, async { + while table() != expected || held_now() != held { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .map_err(|_| { + eyre::eyre!( + "the table has {:?} with {} held; expected {expected:?} with {held} held", + table(), + held_now() + ) + }) +} + +/// A route whose origin is not the one the holder of the addresses +/// authorised is refused. One that is authorised is taken, and so is one +/// that nothing is known about. +#[tokio::test] +async fn test_an_invalid_route_is_refused_and_the_rest_are_taken() -> Result<()> { + let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await; + let Setup { server, .. } = server_with(&cache, None, ImportPolicy::default()).await?; + let (base, _http) = server.serve_http().await?; + let mut peer = established(&server).await?; + + announce(&mut peer, &["193.0.0.0/21", "194.0.0.0/24"], PEER_ASN).await?; + // Someone else's AS as the origin of a covered prefix; and the right + // AS, more specific than it may announce. + announce(&mut peer, &["193.0.4.0/24"], 15169).await?; + announce(&mut peer, &["193.0.6.0/25"], PEER_ASN).await?; + table_becomes(&server, &["193.0.0.0/21", "194.0.0.0/24"], 1).await?; + + let client = reqwest::Client::new(); + let get = |path: &str| { + let request = client.get(format!("{base}{path}")); + async move { Ok::(request.send().await?.error_for_status()?.json().await?) } + }; + // Each route says how it stands. + let routes = get("/api/routes/received").await?; + let validity: Vec<(&str, &str)> = routes + .as_array() + .unwrap() + .iter() + .map(|r| (r["prefix"].as_str().unwrap(), r["rpki"].as_str().unwrap())) + .collect(); + assert_eq!( + validity, + [("193.0.0.0/21", "valid"), ("194.0.0.0/24", "not found")] + ); + // The refused one is on record with its reason. The too-specific one + // was refused before validation came to it. + let refused = get("/api/routes/rejected").await?; + assert_eq!( + refused[0]["by_reason"], + serde_json::json!({"RPKI invalid": 1, "prefix too specific": 1}) + ); + let peers = get("/api/peers").await?; + assert_eq!(peers[0]["routes"], 2); + assert_eq!(peers[0]["held_routes"], 1); + + let rpki = get("/api/rpki").await?; + assert_eq!(rpki["enabled"], true); + assert_eq!(rpki["state"], "synchronised"); + assert_eq!(rpki["vrps_ipv4"], 1); + assert_eq!(rpki["cache"], cache.address().to_string()); + Ok(()) +} + +/// Validity changes with the validation data, and the routes follow: one +/// that was taken goes when it becomes invalid, and one that was refused +/// as invalid is taken when it no longer is. The peer sends nothing more +/// meanwhile; it has said all it has to say. +#[tokio::test] +async fn test_routes_follow_the_validation_data() -> Result<()> { + let ripe = vrp("193.0.0.0/21", 24, PEER_ASN); + let cache = TestCache::start(vec![ripe]).await; + let Setup { server, .. } = server_with(&cache, None, ImportPolicy::default()).await?; + let mut peer = established(&server).await?; + + // Nothing is known about 8.8.8.0/24, so it is taken. 193.0.4.0/24 is + // covered, and not for this origin. + announce(&mut peer, &["8.8.8.0/24", "193.0.4.0/24"], 15169).await?; + table_becomes(&server, &["8.8.8.0/24"], 1).await?; + + // The holder of 8.8.8.0/24 authorises another AS: the route in the + // table is now invalid, and goes. + cache.set(vec![ripe, vrp("8.8.8.0/24", 24, 13335)]); + table_becomes(&server, &[], 2).await?; + let refused = &server.refusals.peers()[0].1; + assert_eq!(refused.total(), 2); + assert_eq!(refused.recent()[0].prefix, prefix("8.8.8.0/24")); + assert_eq!(&refused.recent()[0].as_path[..], [PEER_ASN, 15169]); + + // Both are authorised for the origin they have: both come back, with + // the attributes they were announced with. + cache.set(vec![ + ripe, + vrp("8.8.8.0/24", 24, 15169), + vrp("193.0.4.0/24", 24, 15169), + ]); + table_becomes(&server, &["8.8.8.0/24", "193.0.4.0/24"], 0).await?; + for route in server.received_routes() { + assert!( + route + .attributes + .contains(&AsPathAttribute::from_asns(vec![PEER_ASN, 15169])), + "{route:?}" + ); + } + assert_eq!(server.registry.list()[0].counters.routes(), 2); + Ok(()) +} + +/// A held route is the peer's to withdraw or replace like any other. One +/// it has withdrawn does not appear when it would have become valid. +#[tokio::test] +async fn test_a_held_route_the_peer_withdrew_does_not_come_back() -> Result<()> { + let ripe = vrp("193.0.0.0/21", 24, PEER_ASN); + let cache = TestCache::start(vec![ripe]).await; + let Setup { server, .. } = server_with(&cache, None, ImportPolicy::default()).await?; + let mut peer = established(&server).await?; + + announce(&mut peer, &["193.0.4.0/24", "193.0.5.0/24"], 15169).await?; + table_becomes(&server, &[], 2).await?; + + // One is withdrawn, the other announced again with the right origin. + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![prefix("193.0.4.0/24").into()], + path_attributes: vec![], + announced_routes: vec![], + })) + .await?; + announce(&mut peer, &["193.0.5.0/24"], PEER_ASN).await?; + table_becomes(&server, &["193.0.5.0/24"], 0).await?; + + // Now the old origin would be valid for both. Neither old route is + // there to be brought back. + cache.set(vec![ripe, vrp("193.0.4.0/23", 24, 15169)]); + tokio::time::sleep(Duration::from_millis(500)).await; + table_becomes(&server, &["193.0.5.0/24"], 0).await?; + assert!( + server.received_routes()[0] + .attributes + .contains(&AsPathAttribute::from_asns(vec![PEER_ASN])), + "the route the peer has now" + ); + Ok(()) +} + +/// A peer can be exempt: its invalid routes are taken, and shown for what +/// they are. +#[tokio::test] +async fn test_a_peer_can_be_exempt_from_validation() -> Result<()> { + let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await; + let import = ImportPolicy { + accept_rpki_invalid: true, + ..ImportPolicy::default() + }; + let Setup { server, .. } = server_with(&cache, None, import).await?; + let (base, _http) = server.serve_http().await?; + let mut peer = established(&server).await?; + + announce(&mut peer, &["193.0.4.0/24"], 15169).await?; + table_becomes(&server, &["193.0.4.0/24"], 0).await?; + let routes: Value = reqwest::get(format!("{base}/api/routes/received")) + .await? + .json() + .await?; + assert_eq!(routes[0]["rpki"], "invalid"); + assert!(server.refusals.peers().is_empty()); + Ok(()) +} + +/// Validation data that no cache has confirmed for too long is dropped, +/// and with it every verdict: routes held as invalid are taken. Refusing +/// routes on the word of data that may be long out of date would be worse. +#[tokio::test] +async fn test_held_routes_are_taken_when_the_validation_data_expires() -> Result<()> { + let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await; + let Setup { server, .. } = server_with( + &cache, + Some(Duration::from_millis(1500)), + ImportPolicy::default(), + ) + .await?; + let mut peer = established(&server).await?; + announce(&mut peer, &["193.0.4.0/24"], 15169).await?; + table_becomes(&server, &[], 1).await?; + + drop(cache); + table_becomes(&server, &["193.0.4.0/24"], 0).await?; + assert!(server.rpki.as_ref().unwrap().current().table.is_empty()); + Ok(()) +} diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs index b11f89f..7009047 100644 --- a/crates/rib/src/table/attrs.rs +++ b/crates/rib/src/table/attrs.rs @@ -37,6 +37,12 @@ pub struct SelectionKey { pub as_path_len: u16, /// ORIGIN: 0 for IGP, 1 for EGP, 2 for INCOMPLETE. pub origin: u8, + /// The AS the route originates from: the last AS of the path. Zero, + /// which is no AS's number, if the path is empty or ends in a set and + /// so has no one origin. Not part of selection; kept here because it + /// is what a route's validity is asked about (RFC 6811), for every + /// route, when the validation data changes. + pub origin_as: u32, } /// Reads the selection key straight off attributes in the form an @@ -51,7 +57,12 @@ fn scan(mut wire: &[u8]) -> Option { // A route without ORIGIN does not get this far; if one did, it // would be the least preferred. origin: 2, + origin_as: 0, }; + // The origin as AS4_PATH has it, if there is one: on a session with + // two-octet AS numbers the path's own last AS may be a stand-in + // (RFC 6793). + let mut wide_origin = None; let number = |value: &[u8]| Some(u32::from_be_bytes(<[u8; 4]>::try_from(value).ok()?)); while !wire.is_empty() { let (attribute, rest) = split_attribute(wire).ok()?; @@ -59,38 +70,64 @@ fn scan(mut wire: &[u8]) -> Option { let value = attribute.value; match attribute.type_code { attr_type::ORIGIN => key.origin = *value.first()?, - attr_type::AS_PATH => (key.as_path_len, key.neighbor_as) = scan_as_path(value)?, + attr_type::AS_PATH => { + let path = scan_as_path(value)?; + key.as_path_len = path.hops; + key.neighbor_as = path.first; + key.origin_as = path.last; + } + attr_type::AS4_PATH => wide_origin = Some(scan_as_path(value)?.last), attr_type::MULTI_EXIT_DISC => key.med = Some(number(value)?), attr_type::LOCAL_PREF => key.local_pref = Some(number(value)?), _ => {} } } + if let Some(origin) = wide_origin { + key.origin_as = origin; + } Some(key) } -/// The length of an AS_PATH as selection counts it, and the neighbouring -/// AS, from the attribute's value with four-octet AS numbers. -fn scan_as_path(mut segments: &[u8]) -> Option<(u16, u32)> { +/// What the selection key takes from an AS path. +struct PathScan { + /// The length as selection counts it. + hops: u16, + /// The first AS, if the path begins with a sequence; else zero. + first: u32, + /// The last AS, if the path ends in a sequence; else zero. + last: u32, +} + +/// Reads an AS_PATH or AS4_PATH value with four-octet AS numbers. +fn scan_as_path(mut segments: &[u8]) -> Option { const AS_SEQUENCE: u8 = 2; let mut hops = 0usize; - let mut neighbor_as = 0; - let mut first = true; + let (mut first, mut last) = (0, 0); + let mut at_start = true; while let [kind, count, rest @ ..] = segments { let (asns, rest) = rest.split_at_checked(usize::from(*count) * 4)?; let ordered = *kind == AS_SEQUENCE; hops += if ordered { usize::from(*count) } else { 1 }; - if first && ordered { - neighbor_as = asns - .first_chunk::<4>() - .map_or(0, |asn| u32::from_be_bytes(*asn)); + let asn = |bytes: Option<&[u8; 4]>| bytes.map_or(0, |asn| u32::from_be_bytes(*asn)); + if at_start && ordered { + first = asn(asns.first_chunk::<4>()); } - first = false; + last = if ordered { + asn(asns.last_chunk::<4>()) + } else { + 0 + }; + at_start = false; segments = rest; } if !segments.is_empty() { return None; } - Some((u16::try_from(hops).unwrap_or(u16::MAX), neighbor_as)) + Some(PathScan { + hops: u16::try_from(hops).unwrap_or(u16::MAX), + first, + last, + }) } /// One distinct list of path attributes, with the next hop that goes with @@ -396,6 +433,8 @@ mod tests { neighbor_as: 65001, as_path_len: 3, origin: 2, + // The path ends in a set: it has no one origin. + origin_as: 0, } ); @@ -422,6 +461,7 @@ mod tests { assert_eq!(fresh.key(), parsed.key()); assert_eq!(fresh.key().as_path_len, 3); assert_eq!(fresh.key().neighbor_as, 65001); + assert_eq!(fresh.key().origin_as, 65003); assert_eq!(fresh.attributes(), Some((attrs, Some(next_hop)))); // Bytes that are not attributes are refused. diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 6b8b793..c280e7c 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -11,6 +11,7 @@ version.workspace = true bgp-packet.workspace = true bgp-primitives.workspace = true bgp-rib.workspace = true +bgp-rpki.workspace = true bytes.workspace = true clap.workspace = true futures.workspace = true diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index d927bbf..5593449 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -44,6 +44,31 @@ message BgpConfig { // any AS it cares to name, and have its routes taken. That is for // testing on a closed network and nothing else. bool accept_unconfigured_peers = 9; + + // Route origin validation with the RPKI (RFC 6811). Left out, routes + // are not validated. + Rpki rpki = 10; +} + +// Where to get the validated ROA payloads that say which AS may originate +// which prefixes. With this set, a route whose origin is not authorised +// for its prefix ("invalid") is refused, from every peer that is not in +// our own AS and does not have accept_rpki_invalid set. Routes that +// nothing is known about are taken as before. +message Rpki { + // The RPKI-to-Router (RFC 8210) servers of one or more validators, such + // as Routinator, rpki-client with StayRTR, or FORT, in order of + // preference. The port is required. One is used at a time, over plain + // TCP: keep it on a network you trust. + repeated TcpEndpoint caches = 1; + + // How often to ask the cache for changes unprompted, how long to wait + // before trying again after a failure, and how long the data may be used + // without a cache confirming it, after which nothing is invalid any + // more. Each defaults to what the cache says, or to 3600, 600 and 7200. + uint32 refresh_seconds = 2; + uint32 retry_seconds = 3; + uint32 expire_seconds = 4; } // A route originated by this router. It is announced with an AS_PATH of the @@ -163,6 +188,10 @@ message ImportPolicy { // The AS numbers the peer's routes may originate from: the last AS of // the path. Empty allows any. repeated uint32 origin_asns = 10; + + // By default, where route origin validation is configured (rpki), a + // route that is invalid is refused. True takes it all the same. + bool accept_rpki_invalid = 11; } // A prefix, and optionally the more specific prefixes inside it. diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 8ccf790..45a0c7e 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -671,6 +671,7 @@ mod tests { peer_identifier: Ipv4Addr::new(10, 0, 0, n), hold_time_secs: 90, four_octet_asn: true, + route_refresh: false, families: FamilySet::supported(), add_path_receive: FamilySet::default(), add_path_send: FamilySet::default(), diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 1399a5c..5bfb993 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -8,6 +8,7 @@ use bgp_packet::open::AddPathSendRecv; use bgp_primitives::constants::{AfiSafi, IpVersion}; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::{LocalRoute, LocalRouteAttrs}; +use bgp_rpki::RpkiConfig; use crate::error::{ConfigError, ImportProblem, OriginateProblem, PeerProblem}; use crate::family::FamilySet; @@ -42,6 +43,9 @@ pub struct Config { pub originate: Vec, /// Address for the HTTP API; `None` disables it. pub http_endpoint: Option, + /// Where route origin validation gets its data; `None` for no + /// validation. + pub rpki: Option, } #[derive(Debug, Clone)] @@ -139,6 +143,11 @@ impl Config { None => None, }; + let rpki = match proto.rpki.as_ref() { + Some(rpki) => Some(parse_rpki(rpki)?), + None => None, + }; + let connect_retry = if proto.connect_retry_seconds == 0 { DEFAULT_CONNECT_RETRY } else { @@ -156,6 +165,7 @@ impl Config { accept_unconfigured_peers: proto.accept_unconfigured_peers, originate, http_endpoint, + rpki, }) } } @@ -232,6 +242,28 @@ fn parse_peer(peer: Peer) -> Result { }) } +fn parse_rpki(rpki: &proto::Rpki) -> Result { + if rpki.caches.is_empty() { + return Err(ConfigError::RpkiNoCaches); + } + let caches = rpki + .caches + .iter() + .map(|cache| match cache.port { + // There is no port a cache can be assumed to listen on. + 0 => Err(ConfigError::RpkiCachePort(cache.address.clone())), + port => parse_endpoint(&cache.address, port), + }) + .collect::>()?; + let seconds = |seconds: u32| (seconds > 0).then(|| Duration::from_secs(seconds.into())); + Ok(RpkiConfig { + caches, + refresh: seconds(rpki.refresh_seconds), + retry: seconds(rpki.retry_seconds), + expire: seconds(rpki.expire_seconds), + }) +} + fn parse_import(import: &proto::ImportPolicy) -> Result { let defaults = ImportPolicy::default(); // A length of zero is the field left out. @@ -290,6 +322,7 @@ fn parse_import(import: &proto::ImportPolicy) -> Result defaults.max_as_path_length, length => u16::try_from(length).map_err(|_| ImportProblem::PathLength(length))?, @@ -666,4 +699,49 @@ mod tests { its own length and 32" ); } + + #[test] + fn route_origin_validation_is_off_unless_caches_are_given() { + assert!(parse(BASE).unwrap().rpki.is_none()); + + let config = parse(&format!( + "{BASE}rpki {{ + caches {{ address: \"192.0.2.10\" port: 3323 }} + caches {{ address: \"2001:db8::10\" port: 8282 }} + expire_seconds: 3600 + }}" + )) + .unwrap(); + let rpki = config.rpki.unwrap(); + assert_eq!( + rpki.caches, + [ + "192.0.2.10:3323".parse().unwrap(), + "[2001:db8::10]:8282".parse().unwrap() + ] + ); + // What is not set is left to the cache. + assert_eq!(rpki.refresh, None); + assert_eq!(rpki.expire, Some(Duration::from_secs(3600))); + + assert_eq!( + rejected(&format!("{BASE}rpki {{}}")), + ConfigError::RpkiNoCaches + ); + // No port is assumed for a cache. + assert_eq!( + rejected(&format!( + "{BASE}rpki {{ caches {{ address: \"192.0.2.10\" }} }}" + )), + ConfigError::RpkiCachePort("192.0.2.10".into()) + ); + + // A peer can be exempt. + assert!(!import_of("").unwrap().accept_rpki_invalid); + assert!( + import_of("import { accept_rpki_invalid: true }") + .unwrap() + .accept_rpki_invalid + ); + } } diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 73c6141..c383326 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -14,13 +14,16 @@ use std::sync::Arc; use std::time::Duration; use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode}; +use bgp_packet::message::RouteRefreshMessage; use bgp_packet::message::{ BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Decoded, Message, PathAttribute, UpdateMessage, }; use bgp_packet::open::AddPathSendRecv; use bgp_packet::update_decode::UpdateError; +use bgp_primitives::ip_prefix::Nlri; use bgp_primitives::parser::ParserContext; use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable}; +use bgp_rpki::Rpki; use bytes::BytesMut; use prometheus::IntGauge; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; @@ -42,7 +45,8 @@ use crate::metrics::{ use crate::panic::contained; use crate::policy::{self, HoldDowns, ImportPolicy}; use crate::refusals::{PeerRefusals, RefusalLog}; -use crate::routes; +use crate::routes::{self, Held}; +use crate::rpki::Validator; use crate::session::{PeerId, SessionCounters, SessionGuard, SessionHandle, SessionRegistry}; /// Number of outbound messages that may be queued for a session before @@ -74,6 +78,8 @@ pub struct ServerContext { pub hold_downs: HoldDowns, /// What peers announced that was not taken. pub refusals: RefusalLog, + /// The data for route origin validation, where that is configured. + pub rpki: Option, } /// What is configured for the peer at the other end of a connection. @@ -211,10 +217,40 @@ struct Established { counters: Arc, /// The same number of routes, where metrics are read from. routes_metric: IntGauge, + /// Route origin validation for the peer's routes, if they are subject + /// to it: the data to validate against, and where the session is told + /// that it has changed. + validation: Option, + /// The peer's routes that are out of the table as invalid. + held: Held, + /// The address families the peer can be asked to send again, if it can + /// be asked at all. + refreshable: Option, /// Keeps the session listed in the registry. _registration: SessionGuard, } +impl Established { + /// Keeps count after the peer's routes have changed, and says whether + /// the peer now has more than it may. + fn account( + &self, + outcome: &Outcome, + rejected: usize, + import: &ImportPolicy, + ) -> Result<(), SessionError> { + let routes = self.counters.note(outcome.added, outcome.removed, rejected); + self.counters.set_held(self.held.len()); + self.routes_metric.set(routes as i64); + match import.max_prefixes { + Some(limit) if routes > u64::from(limit) => { + Err(SessionError::PrefixLimitExceeded { limit }) + } + _ => Ok(()), + } + } +} + /// The routes a peer has announced on a session, as far as their lifetime /// goes: they must not outlive it (RFC 4271 ยง8.2.2). /// @@ -283,6 +319,22 @@ impl Drop for PeerRoutes { } } +struct Validation { + validator: Validator, + /// Routes of the peer's that have become invalid, each time the + /// validation data changes. + changes: mpsc::UnboundedReceiver>, +} + +/// What a session's task wakes up for. +enum Wake { + /// Something for the state machine. + Event(Event), + /// The validation data has changed; these routes of the peer's are + /// now invalid. + Revalidate(Vec), +} + struct Session { fsm: Fsm, frames: FrameReader, @@ -406,31 +458,34 @@ impl Session { if let Some(result) = self.execute(actions).await { return result; } - let event = self.next_event().await?; - actions = self.fsm.handle(event); + actions = match self.next_event().await? { + Wake::Event(event) => self.fsm.handle(event), + Wake::Revalidate(suspects) => self.revalidate(&suspects), + }; } } /// Waits for whatever happens next. Every branch is cancel-safe: the /// branches that lose leave nothing half done. - async fn next_event(&mut self) -> Result { - tokio::select! { + async fn next_event(&mut self) -> Result { + let event = tokio::select! { frame = self.frames.next(&self.ctx) => match frame { Ok(decoded) => { self.report(&decoded.update_errors); - Ok(Event::Message(Received { + Event::Message(Received { message: decoded.message, attributes_wire: decoded.attributes_wire, - })) + }) } - Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)), - Err(e) => Err(e), + Err(SessionError::Decode(e)) => Event::DecodeFailed(e), + Err(e) => return Err(e), }, - () = expired(&mut self.hold_timer) => Ok(Event::HoldTimerExpired), - () = ticked(&mut self.keepalive_timer) => Ok(Event::KeepaliveTimerExpired), - msg = queued(&mut self.established) => Ok(Event::Outbound(msg)), - () = killed(&mut self.collision) => Ok(Event::CollisionLost), - } + () = expired(&mut self.hold_timer) => Event::HoldTimerExpired, + () = ticked(&mut self.keepalive_timer) => Event::KeepaliveTimerExpired, + wake = from_server(&mut self.established) => return Ok(wake), + () = killed(&mut self.collision) => Event::CollisionLost, + }; + Ok(Wake::Event(event)) } /// Records errors in an UPDATE that the session survived. They are the @@ -603,7 +658,7 @@ impl Session { collision.set_established(); } let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); - let handle = SessionHandle::new( + let mut handle = SessionHandle::new( peer_id, self.peer_name.clone(), self.direction, @@ -612,7 +667,24 @@ impl Session { rib_peer, outbound_tx, ); + // Routes from other networks are validated, where there is data + // to validate against and the peer's policy does not say to take + // invalid ones anyway. + let validation = self + .server + .rpki + .clone() + .filter(|_| !negotiated.ibgp && !self.import.accept_rpki_invalid) + .map(|rpki| Validation { + validator: Validator::new(rpki), + changes: handle.validated(), + }); self.established = Some(Established { + validation, + held: Held::default(), + refreshable: negotiated + .route_refresh + .then(|| negotiated.families.clone()), peer_id, rib_peer, outbound, @@ -632,21 +704,30 @@ impl Session { /// Puts the routes of a received UPDATE into the routing table. fn store( - &self, + &mut self, update: &UpdateMessage, attributes_wire: Option<&[u8]>, ) -> Result<(), SessionError> { - let Some(established) = &self.established else { + let Some(established) = &mut self.established else { return Ok(()); }; let import = routes::Import { peer: established.rib_peer, from: established.from, policy: &self.import, + vrps: established + .validation + .as_mut() + .map(|validation| validation.validator.table()), refusals: &self.refusals, }; - let routes::Applied { outcome, rejected } = - routes::apply(&self.server.rib, &import, update, attributes_wire)?; + let routes::Applied { outcome, rejected } = routes::apply( + &self.server.rib, + &import, + &mut established.held, + update, + attributes_wire, + )?; // One line for every UPDATE is too much to log by default: a full // table is a million of them. debug!( @@ -658,16 +739,54 @@ impl Session { best_changed = outcome.best_changed, "Received UPDATE" ); - let held = established - .counters - .note(outcome.added, outcome.removed, rejected); - established.routes_metric.set(held as i64); - match self.import.max_prefixes { - Some(limit) if held > u64::from(limit) => { - Err(SessionError::PrefixLimitExceeded { limit }) - } - _ => Ok(()), + established.account(&outcome, rejected, &self.import) + } + + /// Looks at the peer's routes again after the validation data has + /// changed: `suspects` are routes in the table that are now invalid. + /// Returns what there is to do as a result. + fn revalidate(&mut self, suspects: &[Nlri]) -> Vec { + let Some(established) = &mut self.established else { + return vec![]; + }; + let import = routes::Import { + peer: established.rib_peer, + from: established.from, + policy: &self.import, + vrps: established + .validation + .as_mut() + .map(|validation| validation.validator.table()), + refusals: &self.refusals, + }; + let routes::Applied { outcome, rejected } = + routes::revalidate(&self.server.rib, &import, &mut established.held, suspects); + if outcome != Outcome::default() { + info!( + peer = %established.peer_id, + now_invalid = rejected, + no_longer_invalid = outcome.added, + held = established.held.len(), + "Revalidated the peer's routes" + ); } + if let Err(e) = established.account(&outcome, rejected, &self.import) { + return self.fsm.handle(Event::LocalFailure(e)); + } + // Invalid routes that there was no room to hold may be acceptable + // now, and only the peer has them. + if established.held.take_overflowed() + && let Some(families) = &established.refreshable + { + info!(peer = %established.peer_id, "Asking the peer to send its routes again"); + return families + .iter() + .map(|family| { + Action::Send(Message::RouteRefresh(RouteRefreshMessage::request(family))) + }) + .collect(); + } + vec![] } async fn send(&mut self, msg: &Message) -> Result<(), SessionError> { @@ -713,16 +832,30 @@ async fn ticked(timer: &mut Option) { } } -async fn queued(established: &mut Option) -> Message { - match established { - // The registry holds the sending half for as long as the session is - // registered, so the queue does not close under us; if it somehow - // did, there is simply nothing more to send. - Some(established) => match established.outbound.recv().await { - Some(msg) => msg, +/// What the rest of the server has for an established session: a message +/// to send to the peer, or word that the validation data has changed. +async fn from_server(established: &mut Option) -> Wake { + let Some(Established { + outbound, + validation, + .. + }) = established + else { + return pending().await; + }; + let invalidated = async { + match validation { + Some(validation) => validation.changes.recv().await, None => pending().await, - }, - None => pending().await, + } + }; + tokio::select! { + Some(msg) = outbound.recv() => Wake::Event(Event::Outbound(msg)), + Some(suspects) = invalidated => Wake::Revalidate(suspects), + // The registry holds the sending halves for as long as the session + // is registered, so neither closes under us; if they somehow did, + // there is simply nothing more to come. + else => pending().await, } } @@ -931,6 +1064,7 @@ mod tests { collisions: CollisionTable::new(false), hold_downs: HoldDowns::default(), refusals: RefusalLog::default(), + rpki: None, }); let (mut session, _) = Session::open( stream, @@ -978,6 +1112,7 @@ mod tests { collisions: CollisionTable::new(false), hold_downs: HoldDowns::default(), refusals: RefusalLog::default(), + rpki: None, }); let (mut session, _) = Session::open( stream, @@ -1041,9 +1176,10 @@ mod tests { four_octet_asn: true, }, policy: &ImportPolicy::permissive(), + vrps: None, refusals: &RefusalLog::default().of_peer(Ipv4Addr::new(192, 0, 2, 1).into(), "test"), }; - routes::apply(&table, &import, &update, None).unwrap(); + routes::apply(&table, &import, &mut Held::default(), &update, None).unwrap(); assert_eq!(table.len(), 1); assert_eq!(table.attrs().len(), 1); diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 63fdf1e..b17bf84 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -143,6 +143,10 @@ pub enum ConfigError { Port(u32), #[error("http_endpoint.port is required")] HttpPort, + #[error("rpki: no caches are listed")] + RpkiNoCaches, + #[error("rpki: the port of cache {0:?} is required")] + RpkiCachePort(String), #[error("peer {peer:?}: {problem}")] Peer { peer: String, problem: PeerProblem }, #[error("originate {prefix:?}: {problem}")] diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index a616da8..5b06ebd 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -78,6 +78,8 @@ pub struct Negotiated { pub hold_time_secs: u16, /// Whether AS numbers are four octets wide on this session (RFC 6793). pub four_octet_asn: bool, + /// Whether the peer can be asked to send its routes again (RFC 2918). + pub route_refresh: bool, /// The address families both sides offered: the ones routes may be /// exchanged for. Never empty. pub families: FamilySet, @@ -340,6 +342,9 @@ impl Fsm { peer_identifier: open.identifier, hold_time_secs: self.config.hold_time_secs.min(open.hold_time), four_octet_asn: capability_asn.is_some(), + route_refresh: open + .capabilities() + .any(|capability| matches!(capability, Capability::RouteRefresh)), families, add_path_receive, add_path_send, @@ -572,6 +577,7 @@ mod tests { peer_identifier: PEER_ID, hold_time_secs: 30, four_octet_asn: false, + route_refresh: false, // The peer listed no families, so it speaks IPv4 unicast. families: only(AfiSafi::IPV4_UNICAST), add_path_receive: FamilySet::default(), diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index e117bec..20e10c6 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -11,6 +11,7 @@ mod panic; pub mod policy; pub mod refusals; mod routes; +mod rpki; pub mod server; pub mod session; mod sync; diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index b17ff6e..72e59c7 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -87,6 +87,23 @@ pub static PREFIX_LIMIT_EXCEEDED: LazyLock = LazyLock::new(|| { .expect("metric registration") }); +pub static RPKI_VRPS: LazyLock = LazyLock::new(|| { + register_int_gauge_vec!( + "bgp_rpki_vrps", + "Validated ROA payloads in use for route origin validation, by address family", + &["family"] + ) + .expect("metric registration") +}); + +pub static RPKI_UPDATES: LazyLock = LazyLock::new(|| { + register_int_counter!( + "bgp_rpki_updates_total", + "Times the validated ROA payloads changed" + ) + .expect("metric registration") +}); + pub static CONNECTIONS_REFUSED: LazyLock = LazyLock::new(|| { register_int_counter_vec!( "bgp_connections_refused_total", diff --git a/crates/server/src/policy.rs b/crates/server/src/policy.rs index 39c2bec..10b5709 100644 --- a/crates/server/src/policy.rs +++ b/crates/server/src/policy.rs @@ -76,6 +76,8 @@ pub struct ImportPolicy { /// The AS numbers the peer's routes may originate from; `None` for /// any. pub origin_asns: Option>, + /// Take routes that route origin validation finds invalid (RFC 6811). + pub accept_rpki_invalid: bool, } impl Default for ImportPolicy { @@ -93,6 +95,7 @@ impl Default for ImportPolicy { max_as_path_length: DEFAULT_MAX_AS_PATH_LENGTH, prefixes: None, origin_asns: None, + accept_rpki_invalid: false, } } } @@ -109,6 +112,7 @@ impl ImportPolicy { max_ipv6_length: IpVersion::Ipv6.bits(), accept_any_first_as: true, max_as_path_length: u16::MAX, + accept_rpki_invalid: true, ..Self::default() } } @@ -136,10 +140,13 @@ pub enum Rejection { /// The route does not originate from an AS the peer may announce /// routes of. OriginNotAllowed, + /// The holder of the addresses has said who may originate routes for + /// them, and it is not this route's origin (RFC 6811). + RpkiInvalid, } impl Rejection { - pub const ALL: [Self; 9] = [ + pub const ALL: [Self; 10] = [ Self::BogonPrefix, Self::TooShort, Self::TooSpecific, @@ -149,6 +156,7 @@ impl Rejection { Self::FirstAs, Self::PathTooLong, Self::OriginNotAllowed, + Self::RpkiInvalid, ]; /// Where the reason is in [`Rejection::ALL`]. @@ -167,6 +175,7 @@ impl Rejection { Self::FirstAs => "path does not begin with the peer's AS", Self::PathTooLong => "path too long", Self::OriginNotAllowed => "origin AS not allowed", + Self::RpkiInvalid => "RPKI invalid", } } } @@ -186,10 +195,13 @@ pub struct From { impl ImportPolicy { /// Judges what the routes of one UPDATE have in common: their path. - pub fn check_path(&self, attributes: &[PathAttribute], from: &From) -> Result<(), Rejection> { - if from.ibgp { - return Ok(()); - } + /// If it passes, returns the AS the routes originate from: the last AS + /// of the path, or `None` if the path is empty or ends in a set. + pub fn check_path( + &self, + attributes: &[PathAttribute], + from: &From, + ) -> Result, Rejection> { let path = attributes.iter().find_map(|attribute| match attribute { PathAttribute::AsPath(path) => Some(path), _ => None, @@ -199,6 +211,16 @@ impl ImportPolicy { _ => None, }); let facts = PathFacts::of(path, as4_path, from); + // The origin is the last AS of the path. If the path ends in a + // set, there is no telling which of its members that is. + let ends_in_sequence = as4_path + .or(path) + .and_then(|path| path.segments.last()) + .is_some_and(|segment| segment.ordered); + let origin = facts.last.filter(|_| ends_in_sequence); + if from.ibgp { + return Ok(origin); + } if facts.has_local_asn { return Err(Rejection::OwnAs); @@ -220,19 +242,12 @@ impl ImportPolicy { if !self.accept_bogons && facts.has_bogon_asn { return Err(Rejection::BogonAs); } - if let Some(allowed) = &self.origin_asns { - // The origin is the last AS of the path. If the path ends in a - // set, there is no telling which of its members that is. - let ends_in_sequence = as4_path - .or(path) - .and_then(|path| path.segments.last()) - .is_some_and(|segment| segment.ordered); - let origin = facts.last.filter(|_| ends_in_sequence); - if !origin.is_some_and(|origin| allowed.contains(&origin)) { - return Err(Rejection::OriginNotAllowed); - } + if let Some(allowed) = &self.origin_asns + && !origin.is_some_and(|origin| allowed.contains(&origin)) + { + return Err(Rejection::OriginNotAllowed); } - Ok(()) + Ok(origin) } /// Judges one prefix. @@ -618,7 +633,9 @@ mod tests { ..EXTERNAL }; assert_eq!( - ImportPolicy::default().check_path(&path(&[]), &internal), + ImportPolicy::default() + .check_path(&path(&[]), &internal) + .map(drop), Ok(()) ); } @@ -626,7 +643,10 @@ mod tests { #[test] fn the_path_must_begin_with_the_peers_as_unless_it_is_a_route_server() { let policy = ImportPolicy::default(); - assert_eq!(policy.check_path(&path(&[3333, 2914]), &EXTERNAL), Ok(())); + assert_eq!( + policy.check_path(&path(&[3333, 2914]), &EXTERNAL).map(drop), + Ok(()) + ); assert_eq!( policy.check_path(&path(&[2914, 3333]), &EXTERNAL), Err(Rejection::FirstAs) @@ -649,7 +669,9 @@ mod tests { ..ImportPolicy::default() }; assert_eq!( - route_server.check_path(&path(&[2914, 174]), &EXTERNAL), + route_server + .check_path(&path(&[2914, 174]), &EXTERNAL) + .map(drop), Ok(()) ); } @@ -677,7 +699,7 @@ mod tests { } for real in [174, 64495, 131072, 210036, 4_199_999_999] { assert_eq!( - policy.check_path(&path(&[3333, real]), &EXTERNAL), + policy.check_path(&path(&[3333, real]), &EXTERNAL).map(drop), Ok(()), "AS{real}" ); @@ -703,7 +725,7 @@ mod tests { path: vec![210036], }], })); - assert_eq!(policy.check_path(&attributes, &two_octet), Ok(())); + assert_eq!(policy.check_path(&attributes, &two_octet).map(drop), Ok(())); // What AS4_PATH reveals is judged too. let mut attributes = path(&[3333, 23456]); @@ -722,7 +744,10 @@ mod tests { #[test] fn an_absurdly_long_path_is_refused() { let policy = ImportPolicy::default(); - assert_eq!(policy.check_path(&path(&[3333; 64]), &EXTERNAL), Ok(())); + assert_eq!( + policy.check_path(&path(&[3333; 64]), &EXTERNAL), + Ok(Some(3333)) + ); assert_eq!( policy.check_path(&path(&[3333; 65]), &EXTERNAL), Err(Rejection::PathTooLong) @@ -735,8 +760,14 @@ mod tests { origin_asns: Some([3333, 2121].into()), ..ImportPolicy::default() }; - assert_eq!(policy.check_path(&path(&[3333]), &EXTERNAL), Ok(())); - assert_eq!(policy.check_path(&path(&[3333, 2121]), &EXTERNAL), Ok(())); + assert_eq!( + policy.check_path(&path(&[3333]), &EXTERNAL).map(drop), + Ok(()) + ); + assert_eq!( + policy.check_path(&path(&[3333, 2121]), &EXTERNAL).map(drop), + Ok(()) + ); assert_eq!( policy.check_path(&path(&[3333, 15169]), &EXTERNAL), Err(Rejection::OriginNotAllowed) diff --git a/crates/server/src/routes.rs b/crates/server/src/routes.rs index 24f3a00..091dd0c 100644 --- a/crates/server/src/routes.rs +++ b/crates/server/src/routes.rs @@ -3,22 +3,29 @@ use std::borrow::Cow; use std::cell::OnceCell; +use std::collections::HashMap; use std::sync::Arc; use bgp_packet::message::{NlriNextHop, PathAttribute, UpdateMessage}; use bgp_primitives::ip_prefix::Nlri; use bgp_primitives::parser::EncodeError; use bgp_rib::table::{AttrSet, Outcome, PeerIndex, RibTable}; +use bgp_rpki::{Validity, VrpTable}; use crate::policy::{self, From, ImportPolicy, Rejection}; use crate::refusals::PeerRefusals; +/// The most routes of one peer that are held aside as invalid. Far more +/// than a peer sending the whole Internet has; a bound all the same. +const MAX_HELD: usize = 200_000; + /// What applying an UPDATE did. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub(crate) struct Applied { /// What changed in the table. pub(crate) outcome: Outcome, - /// How many routes the UPDATE announced that the policy refused. + /// How many routes were announced, or found on a second look, that + /// the policy refused. pub(crate) rejected: usize, } @@ -27,10 +34,60 @@ pub(crate) struct Import<'a> { pub(crate) peer: PeerIndex, pub(crate) from: From, pub(crate) policy: &'a ImportPolicy, + /// What the RPKI says about who may originate what, if this peer's + /// routes are validated against it. + pub(crate) vrps: Option<&'a VrpTable>, /// Where what is refused is noted. pub(crate) refusals: &'a PeerRefusals, } +/// A peer's routes that are not in the routing table because route origin +/// validation found them invalid. +/// +/// Every other reason for refusing a route is a matter of the route and of +/// our configuration, and stays true for as long as the session lasts. +/// Validity is not: the data it rests on changes all the time, and a route +/// that is invalid now may be valid in ten minutes. So these routes are +/// kept, out of use, to be taken when that happens. The peer will not send +/// them again by itself. +#[derive(Default)] +pub(crate) struct Held { + routes: HashMap>, + /// Whether a route was not kept for want of room: then looking through + /// `routes` again is not enough, and the peer has to be asked. + overflowed: bool, +} + +impl Held { + pub(crate) fn len(&self) -> usize { + self.routes.len() + } + + /// Whether some invalid route was dropped instead of held; resets it. + pub(crate) fn take_overflowed(&mut self) -> bool { + std::mem::take(&mut self.overflowed) + } + + fn hold(&mut self, route: Nlri, attrs: &Arc) { + if self.routes.len() >= MAX_HELD && !self.routes.contains_key(&route) { + self.overflowed = true; + return; + } + self.routes.insert(route, Arc::clone(attrs)); + } + + /// The peer has replaced or withdrawn these routes: what was held of + /// them is no longer what the peer has. + fn forget(&mut self, routes: &[Nlri]) { + if self.routes.is_empty() { + return; + } + for route in routes { + self.routes.remove(route); + } + } +} + /// Applies `update` to `table`: its withdrawals first and then its /// announcements, so that a prefix named in both ends up announced /// (RFC 4271 ยง4.3). @@ -42,7 +99,8 @@ pub(crate) struct Import<'a> { /// prefix and path identifier together, and is a route of its own. /// /// A route the policy refuses is withdrawn instead: whatever the peer had -/// announced for it before is replaced by this, and this is not taken. +/// announced for it before is replaced by this, and this is not taken. One +/// refused as invalid by route origin validation is kept in `held`. /// /// `attributes_wire` is the attributes as the decoder kept them, if it did /// (see `DecodedUpdate::attributes_wire`). With it the table finds the @@ -50,6 +108,7 @@ pub(crate) struct Import<'a> { pub(crate) fn apply( table: &RibTable, import: &Import<'_>, + held: &mut Held, update: &UpdateMessage, attributes_wire: Option<&[u8]>, ) -> Result { @@ -58,10 +117,12 @@ pub(crate) fn apply( if !update.withdrawn_routes.is_empty() { applied.outcome += table.withdraw(peer, &update.withdrawn_routes); + held.forget(&update.withdrawn_routes); } for attribute in &update.path_attributes { if let PathAttribute::MpUnreachNlri(mp) = attribute { applied.outcome += table.withdraw(peer, &mp.prefixes); + held.forget(&mp.prefixes); } } @@ -85,6 +146,9 @@ pub(crate) fn apply( if nlri.is_empty() { return Ok(()); } + // Whatever becomes of them, these replace what was held of the + // same routes. + held.forget(nlri); let mut refuse = |refused: &[Nlri], rejection: Rejection| { import .refusals @@ -92,13 +156,37 @@ pub(crate) fn apply( applied.rejected += refused.len(); applied.outcome += table.withdraw(peer, refused); }; - if let Err(rejection) = path_verdict { - refuse(nlri, rejection); + let origin = match path_verdict { + Ok(origin) => origin, + Err(rejection) => { + refuse(nlri, rejection); + return Ok(()); + } + }; + let mut invalid = Vec::new(); + let accepted = sift( + nlri, + |route| { + import.policy.check_prefix(&route.prefix)?; + match import.vrps.map(|vrps| vrps.validate(&route.prefix, origin)) { + Some(Validity::Invalid) => { + invalid.push(*route); + Err(Rejection::RpkiInvalid) + } + _ => Ok(()), + } + }, + &mut refuse, + ); + if accepted.is_empty() && invalid.is_empty() { return Ok(()); } - let accepted = sift(import.policy, nlri, &mut refuse); + let attrs = attributes(next_hop)?; + for route in invalid { + held.hold(route, &attrs); + } if !accepted.is_empty() { - applied.outcome += table.announce(peer, &attributes(next_hop)?, &accepted); + applied.outcome += table.announce(peer, &attrs, &accepted); } Ok(()) }; @@ -126,20 +214,26 @@ fn path_of(update: &UpdateMessage) -> Arc<[u32]> { policy::asns(path).collect() } -/// The routes of `nlri` whose prefixes the policy takes; the others are -/// given to `refuse`. Nearly always that is all of them, and nothing is -/// copied. +/// The routes of `nlri` that `verdict` passes; the others are given to +/// `refuse`. Nearly always that is all of them, and nothing is copied. fn sift<'a>( - policy: &ImportPolicy, nlri: &'a [Nlri], + mut verdict: impl FnMut(&Nlri) -> Result<(), Rejection>, refuse: &mut impl FnMut(&[Nlri], Rejection), ) -> Cow<'a, [Nlri]> { - let verdict = |route: &Nlri| policy.check_prefix(&route.prefix); - let Some(first_refused) = nlri.iter().position(|route| verdict(route).is_err()) else { + let mut first_refused = None; + for (at, route) in nlri.iter().enumerate() { + if let Err(rejection) = verdict(route) { + first_refused = Some((at, rejection)); + break; + } + } + let Some((at, rejection)) = first_refused else { return Cow::Borrowed(nlri); }; - let mut accepted = nlri[..first_refused].to_vec(); - for route in &nlri[first_refused..] { + refuse(std::slice::from_ref(&nlri[at]), rejection); + let mut accepted = nlri[..at].to_vec(); + for route in &nlri[at + 1..] { match verdict(route) { Ok(()) => accepted.push(*route), Err(rejection) => refuse(std::slice::from_ref(route), rejection), @@ -147,3 +241,70 @@ fn sift<'a>( } Cow::Owned(accepted) } + +/// Looks again at a peer's routes after the validation data has changed. +/// +/// `suspects` are routes of the peer's in the table that may now be +/// invalid. Those that are (they may have changed since they were listed) +/// are taken out and held. Every route that was held and is no longer +/// invalid goes into the table. +pub(crate) fn revalidate( + table: &RibTable, + import: &Import<'_>, + held: &mut Held, + suspects: &[Nlri], +) -> Applied { + let mut applied = Applied::default(); + let Some(vrps) = import.vrps else { + return applied; + }; + let validity = |route: &Nlri, attrs: &AttrSet| { + let origin = Some(attrs.key().origin_as).filter(|origin| *origin != 0); + vrps.validate(&route.prefix, origin) + }; + + for suspect in suspects { + // As it is in the table now. + let Some(route) = table + .routes(&suspect.prefix) + .into_iter() + .find(|route| route.peer == import.peer && route.path_id == suspect.path_id) + else { + continue; + }; + if validity(suspect, &route.attrs) != Validity::Invalid { + continue; + } + applied.outcome += table.withdraw(import.peer, std::slice::from_ref(suspect)); + applied.rejected += 1; + let as_path: Arc<[u32]> = match route.attrs.attributes() { + Some((attributes, _)) => { + let path = attributes.iter().find_map(|attribute| match attribute { + PathAttribute::AsPath(path) => Some(path), + _ => None, + }); + policy::asns(path).collect() + } + None => Arc::default(), + }; + import.refusals.note( + std::slice::from_ref(suspect), + Rejection::RpkiInvalid, + &as_path, + ); + held.hold(*suspect, &route.attrs); + } + + let released: Vec = held + .routes + .iter() + .filter(|(route, attrs)| validity(route, attrs) != Validity::Invalid) + .map(|(route, _)| *route) + .collect(); + for route in released { + if let Some(attrs) = held.routes.remove(&route) { + applied.outcome += table.announce(import.peer, &attrs, std::slice::from_ref(&route)); + } + } + applied +} diff --git a/crates/server/src/rpki.rs b/crates/server/src/rpki.rs new file mode 100644 index 0000000..bdd10e8 --- /dev/null +++ b/crates/server/src/rpki.rs @@ -0,0 +1,155 @@ +//! Route origin validation in the server: where a session gets the data it +//! validates against, and what happens to routes already taken when that +//! data changes. +//! +//! A route is validated when it arrives (see [`crate::routes`]). But +//! whether a route is valid is not settled once: the validated ROA +//! payloads change every few minutes, and each change can make routes that +//! are in the table invalid, and routes that were refused as invalid +//! acceptable. The [revalidator](spawn_revalidator) deals with the first +//! by looking through the table; each session deals with the second, from +//! the routes it has [held](crate::routes::Held). + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use bgp_primitives::ip_prefix::Nlri; +use bgp_rib::table::{PeerIndex, RibTable}; +use bgp_rpki::{Rpki, Update, Validity, VrpTable}; +use tokio::task::JoinHandle; +use tracing::{info, warn}; + +use crate::metrics::{RPKI_UPDATES, RPKI_VRPS}; +use crate::session::{SessionHandle, SessionRegistry}; + +/// How long after new data is published the table is looked through. A +/// session that was in the middle of an UPDATE when the data changed +/// validated it against the old data; by now it has finished, and the look +/// through the table sees what it put there. +const SETTLE: Duration = Duration::from_millis(200); + +/// A session's view of the validation data: the table in use, replaced +/// when a newer one is published. Asking whether there is a newer one +/// costs the load of one number, which is done for every UPDATE. +pub(crate) struct Validator { + rpki: Rpki, + table: Arc, + version: u64, +} + +impl Validator { + pub(crate) fn new(rpki: Rpki) -> Self { + let Update { version, table, .. } = rpki.current(); + Self { + rpki, + table, + version, + } + } + + /// The table to validate against now. + pub(crate) fn table(&mut self) -> &VrpTable { + if self.rpki.version() != self.version { + let Update { version, table, .. } = self.rpki.current(); + self.table = table; + self.version = version; + } + &self.table + } +} + +/// Starts the task that, whenever the validation data changes, finds the +/// routes in the table that are now invalid and has their sessions take +/// them out, and has every session look again at the routes it holds. +pub(crate) fn spawn_revalidator( + rpki: Rpki, + rib: Arc, + registry: SessionRegistry, +) -> JoinHandle<()> { + tokio::spawn(async move { + let mut updates = rpki.subscribe(); + let mut seen = updates.borrow().version; + // Ends only if the client that publishes the data has gone, which + // the server notices for itself. + while updates.changed().await.is_ok() { + let update = updates.borrow_and_update().clone(); + // What changed is known only from one version to the next. + let changed = match update.changed { + Some(changed) if update.version == seen + 1 => Some(changed), + _ => None, + }; + seen = update.version; + + let status = rpki.status(); + RPKI_VRPS + .with_label_values(&["ipv4"]) + .set(status.vrps_ipv4 as i64); + RPKI_VRPS + .with_label_values(&["ipv6"]) + .set(status.vrps_ipv6 as i64); + RPKI_UPDATES.inc(); + + tokio::time::sleep(SETTLE).await; + let sessions: Vec = registry + .list() + .into_iter() + .filter(SessionHandle::is_validated) + .collect(); + let peers: Vec = sessions.iter().map(|session| session.rib_peer).collect(); + let table = Arc::clone(&update.table); + let rib = Arc::clone(&rib); + // Looks at every route, or at every prefix at least: not for a + // thread that sessions run on. + let found = tokio::task::spawn_blocking(move || { + newly_invalid(&rib, &table, changed.as_deref(), &peers) + }) + .await; + let mut suspects = match found { + Ok(suspects) => suspects, + Err(e) => { + warn!(err = %e, "Looking for routes that are now invalid failed"); + HashMap::new() + } + }; + info!( + vrps = update.table.len(), + now_invalid = suspects.values().map(Vec::len).sum::(), + sessions = sessions.len(), + "RPKI data changed; revalidating routes" + ); + // Every session, also those with nothing found: each has its + // held routes to look at again. + for session in sessions { + session.revalidate(suspects.remove(&session.rib_peer).unwrap_or_default()); + } + } + }) +} + +/// The routes of `peers` in the table that `vrps` finds invalid, by peer. +/// With `changed`, only routes that the VRPs in it cover are looked at: no +/// others can have become invalid. +fn newly_invalid( + rib: &RibTable, + vrps: &VrpTable, + changed: Option<&VrpTable>, + peers: &[PeerIndex], +) -> HashMap> { + let routes = rib.routes_where(usize::MAX, |route| { + changed.is_none_or(|changed| changed.covers(&route.prefix)) + && peers.contains(&route.peer) + && { + let origin = Some(route.attrs.key().origin_as).filter(|origin| *origin != 0); + vrps.validate(&route.prefix, origin) == Validity::Invalid + } + }); + let mut by_peer: HashMap> = HashMap::new(); + for route in routes { + by_peer.entry(route.peer).or_default().push(Nlri { + prefix: route.prefix, + path_id: route.path_id, + }); + } + by_peer +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index a66788d..d3d5ba7 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -15,6 +15,7 @@ use tracing::{error, info, warn}; use bgp_rib::table::RibTable; use bgp_rib::{LocalRib, RibStopped}; +use bgp_rpki::Rpki; use crate::advertise::spawn_advertiser; use crate::collision::{CollisionTable, Direction}; @@ -26,6 +27,7 @@ use crate::metrics::{CONNECTIONS_REFUSED, PEER_MAX_PREFIXES}; use crate::panic; use crate::policy::HoldDowns; use crate::refusals::RefusalLog; +use crate::rpki::spawn_revalidator; use crate::session::{ConfiguredPeer, SessionRegistry}; use crate::sync::IgnorePoison; @@ -71,6 +73,10 @@ pub struct BgpServer { pub registry: SessionRegistry, /// What peers announced that was not taken. pub refusals: RefusalLog, + /// The data for route origin validation, where that is configured. + pub rpki: Option, + /// The task that keeps that data up to date. + rpki_client: Option>, } impl BgpServer { @@ -100,7 +106,12 @@ impl BgpServer { remote_asn: p.remote_asn, }) .collect(); + // The client starts now: with luck the data is there before the + // first routes are. + let (rpki, rpki_client) = config.rpki.clone().map(Rpki::spawn).unzip(); Ok(BgpServer { + rpki, + rpki_client, listeners, rib: Arc::new(RibTable::new(table_shards())), local_rib, @@ -154,6 +165,7 @@ impl BgpServer { collisions: CollisionTable::new(self.config.collision_detect_established_state), hold_downs: HoldDowns::default(), refusals: self.refusals.clone(), + rpki: self.rpki.clone(), }); // The tasks the server cannot do without. None of them ends while @@ -169,6 +181,13 @@ impl BgpServer { ), )); tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib)))); + if let (Some(rpki), Some(client)) = (self.rpki.clone(), self.rpki_client) { + tasks.push(named("RPKI client", client)); + tasks.push(named( + "revalidator", + spawn_revalidator(rpki, Arc::clone(&self.rib), self.registry.clone()), + )); + } for dialer in spawn_dialers(&self.config.peers, &context, self.config.connect_retry) { tasks.push(named("dialer", dialer)); } diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index 0900574..3939c39 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -15,6 +15,7 @@ use bgp_packet::errors::{BgpError, CeaseSubcode}; use bgp_packet::message::NotificationMessage; use bgp_packet::message::Message; +use bgp_primitives::ip_prefix::Nlri; use bgp_rib::table::PeerIndex; use tokio::sync::{broadcast, mpsc}; @@ -69,6 +70,9 @@ pub struct SessionCounters { /// Routes the peer announced that its import policy refused, since the /// session came up. rejected: AtomicU64, + /// Routes of the peer's that are held out of the table as invalid, to + /// be taken if route origin validation comes to say otherwise. + held: AtomicU64, } impl SessionCounters { @@ -80,6 +84,14 @@ impl SessionCounters { self.rejected.load(Ordering::Relaxed) } + pub fn held(&self) -> u64 { + self.held.load(Ordering::Relaxed) + } + + pub(crate) fn set_held(&self, held: usize) { + self.held.store(held as u64, Ordering::Relaxed); + } + /// Notes what one UPDATE did, and returns how many routes the peer has /// in the table after it. pub(crate) fn note(&self, added: usize, removed: usize, rejected: usize) -> u64 { @@ -122,6 +134,9 @@ pub struct SessionHandle { /// How many routes the peer has, and how many were refused. pub counters: Arc, outbound: mpsc::Sender, + /// Where the session is told to look at its routes again, if they are + /// subject to route origin validation. + revalidate: Option>>, } #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] @@ -152,6 +167,31 @@ impl SessionHandle { rib_peer, counters: Arc::default(), outbound, + revalidate: None, + } + } + + /// Marks the session's routes as subject to route origin validation. + /// What is sent with [`SessionHandle::revalidate`] arrives at the + /// receiver returned. + pub(crate) fn validated(&mut self) -> mpsc::UnboundedReceiver> { + let (tx, rx) = mpsc::unbounded_channel(); + self.revalidate = Some(tx); + rx + } + + /// Whether the session's routes are subject to route origin + /// validation. + pub fn is_validated(&self) -> bool { + self.revalidate.is_some() + } + + /// Tells the session that the validation data has changed. `suspects` + /// are routes of its peer's in the table that are now invalid. + pub(crate) fn revalidate(&self, suspects: Vec) { + if let Some(revalidate) = &self.revalidate { + // Fails only if the session has just ended. + let _ = revalidate.send(suspects); } } diff --git a/crates/web/Cargo.toml b/crates/web/Cargo.toml index 051b34b..b2061b5 100644 --- a/crates/web/Cargo.toml +++ b/crates/web/Cargo.toml @@ -13,6 +13,7 @@ axum.workspace = true bgp-packet.workspace = true bgp-primitives.workspace = true bgp-rib.workspace = true +bgp-rpki.workspace = true bgp-server.workspace = true futures.workspace = true prometheus.workspace = true diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index 04dd2ed..b8cdd2c 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -3,6 +3,7 @@ use std::collections::{BTreeMap, HashMap}; use std::net::IpAddr; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use axum::Json; @@ -16,6 +17,7 @@ use bgp_primitives::constants::IpVersion; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::table::{PeerIndex, RibTable, Route}; use bgp_rib::{LocalRoute, RibStopped}; +use bgp_rpki::{CacheState, VrpTable}; use bgp_server::session::{ConfiguredPeer, DialState, PeerId, SessionHandle, SessionRegistry}; use serde::{Deserialize, Serialize}; @@ -73,6 +75,9 @@ pub struct PeerView { /// How many routes the peer announced on this session that its import /// policy refused. pub rejected_routes: Option, + /// How many of the peer's routes are held out of the table as invalid + /// by route origin validation, to be taken if that changes. + pub held_routes: Option, /// Why the last connection attempt or session ended, while in Backoff. pub last_error: Option, pub retry_at_unix: Option, @@ -95,6 +100,7 @@ impl From<&SessionHandle> for PeerView { add_path_send: Some(s.add_path_send.iter().map(|f| f.to_string()).collect()), routes: Some(s.counters.routes()), rejected_routes: Some(s.counters.rejected()), + held_routes: Some(s.counters.held()), last_error: None, retry_at_unix: None, } @@ -125,6 +131,7 @@ impl PeerView { add_path_send: None, routes: None, rejected_routes: None, + held_routes: None, last_error, retry_at_unix, } @@ -262,6 +269,11 @@ pub struct ReceivedRouteView { pub path_id: u32, /// Whether this is the route chosen for its prefix. pub best: bool, + /// What route origin validation says of the route: `valid` or + /// `not found`. (An invalid route is not in the table, unless its + /// peer's are taken regardless; then `invalid`.) Absent where + /// validation is not configured. + pub rpki: Option<&'static str>, pub summary: AttrSummary, /// The decoded attribute list. MP_REACH_NLRI is not in it: its next hop /// is in the summary and its prefixes are routes of their own. @@ -270,23 +282,29 @@ pub struct ReceivedRouteView { /// Who the routes in the table are from: the table knows peers by number, /// the sessions know which number is whose. -pub(crate) struct PeerNames(HashMap)>); +pub(crate) struct PeerNames { + peers: HashMap)>, + /// What routes are validated against, where they are. + vrps: Option>, +} impl PeerNames { - pub(crate) fn of(registry: &SessionRegistry) -> Self { - Self( - registry + pub(crate) fn of(state: &AppState) -> Self { + Self { + peers: state + .registry .list() .into_iter() .map(|session| (session.rib_peer, (session.peer, session.name))) .collect(), - ) + vrps: state.rpki.as_ref().map(|rpki| rpki.current().table), + } } /// The peers that `wanted` names: by configured name, or by address. fn matching(&self, wanted: &str) -> Vec { let address: Option = wanted.parse().ok(); - self.0 + self.peers .iter() .filter(|(_, (peer, name))| { name.as_deref() == Some(wanted) || address == Some(peer.addr.ip()) @@ -298,7 +316,8 @@ impl PeerNames { /// A view of one route. `None` if the route's session has just ended: /// the route is on its way out of the table. pub(crate) fn view(&self, route: &Route) -> Option { - let (peer, name) = self.0.get(&route.peer)?; + let (peer, name) = self.peers.get(&route.peer)?; + let origin = Some(route.attrs.key().origin_as).filter(|origin| *origin != 0); // What a session stored was accepted by the decoder, and parses. let (attributes, next_hop) = route.attrs.attributes().unwrap_or_default(); Some(ReceivedRouteView { @@ -308,6 +327,10 @@ impl PeerNames { prefix: route.prefix.to_string(), path_id: route.path_id, best: route.best, + rpki: self + .vrps + .as_ref() + .map(|vrps| vrps.validate(&route.prefix, origin).as_str()), summary: summarize(&attributes, next_hop.as_ref()), attributes, }) @@ -534,7 +557,7 @@ pub async fn received_routes( .unwrap_or(DEFAULT_ROUTE_LIMIT) .min(MAX_ROUTE_LIMIT); let scope = Scope::of(&query)?; - let names = PeerNames::of(&state.registry); + let names = PeerNames::of(&state); // Looking through the table, copying, parsing and sorting many routes // takes long enough to keep off the threads that sessions run on. let routes = @@ -545,6 +568,61 @@ pub async fn received_routes( Ok(Json(routes)) } +/// Route origin validation: where the data comes from and how fresh it is. +#[derive(Debug, Clone, Serialize)] +pub struct RpkiView { + /// Whether validation is configured at all. + pub enabled: bool, + /// The cache in use, or last tried. + pub cache: Option, + /// `connecting`, `synchronising` or `synchronised`. + pub state: Option<&'static str>, + pub session_id: Option, + pub serial: Option, + pub protocol_version: Option, + pub vrps_ipv4: usize, + pub vrps_ipv6: usize, + pub updated_at_unix: Option, + /// When the data is dropped unless a cache confirms it first. + pub expires_at_unix: Option, + pub last_error: Option, +} + +pub async fn rpki(State(state): State) -> Json { + let Some(status) = state.rpki.as_ref().map(|rpki| rpki.status()) else { + return Json(RpkiView { + enabled: false, + cache: None, + state: None, + session_id: None, + serial: None, + protocol_version: None, + vrps_ipv4: 0, + vrps_ipv6: 0, + updated_at_unix: None, + expires_at_unix: None, + last_error: None, + }); + }; + Json(RpkiView { + enabled: true, + cache: status.cache.map(|cache| cache.to_string()), + state: Some(match status.state { + CacheState::Connecting => "connecting", + CacheState::Synchronising => "synchronising", + CacheState::Synchronised => "synchronised", + }), + session_id: status.session_id, + serial: status.serial, + protocol_version: Some(status.protocol_version), + vrps_ipv4: status.vrps_ipv4, + vrps_ipv6: status.vrps_ipv6, + updated_at_unix: status.updated_at.map(unix_secs), + expires_at_unix: status.expires_at.map(unix_secs), + last_error: status.last_error, + }) +} + /// What one peer announced that was refused. #[derive(Debug, Clone, Serialize)] pub struct RefusalsView { diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index c1ffe3c..3835950 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -97,7 +97,7 @@ impl WireEvent { /// own, so one that reads slowly holds nobody else up. fn best_routes(state: &AppState) -> impl Stream + use<> { let (events, stream) = mpsc::channel(BEST_ROUTE_BUFFER); - let registry = state.registry.clone(); + let state_for_views = state.clone(); // Taken before this returns, so that nothing that happens after the // client is answered is missed. let mut reader = state.rib.reader_from_now(); @@ -140,7 +140,7 @@ fn best_routes(state: &AppState) -> impl Stream + use<> { } _ => {} }); - let names = PeerNames::of(®istry); + let names = PeerNames::of(&state_for_views); for change in changes { let event = match change { Change::Best(route) => match names.view(&route) { diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 0c82043..2ad1d67 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -19,6 +19,7 @@ use axum::response::{Html, IntoResponse, Redirect}; use axum::routing::get; use bgp_rib::LocalRib; use bgp_rib::table::RibTable; +use bgp_rpki::Rpki; use bgp_server::refusals::RefusalLog; use bgp_server::session::SessionRegistry; use tokio::net::TcpListener; @@ -26,7 +27,7 @@ use tracing::info; pub use api::{ AttrSummary, ConfigView, LocalRouteView, PeerView, ReceivedRouteView, RefusalsView, - RefusedRouteView, StatusView, + RefusedRouteView, RpkiView, StatusView, }; /// Static facts about the router shown in status output. @@ -87,6 +88,8 @@ pub struct AppState { pub rib: Arc, /// What peers announced that was not taken. pub refusals: RefusalLog, + /// The data for route origin validation, where that is configured. + pub rpki: Option, } /// The status page, embedded so the binary is self-contained. @@ -110,6 +113,7 @@ pub fn router(state: AppState) -> Router { .route("/api/routes/local", get(api::local_routes)) .route("/api/routes/received", get(api::received_routes)) .route("/api/routes/rejected", get(api::rejected_routes)) + .route("/api/rpki", get(api::rpki)) .route("/api/events", get(events::stream)) .route("/metricz", get(metrics::metricz)) .with_state(state) diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index 8fc2cf7..a048d6b 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -47,7 +47,7 @@ Router ID ..., local AS ...
@@ -63,6 +63,9 @@ View: variables, status json, < +

Route origin validation (RPKI)

+
...
+

Originated routes (...)

@@ -83,7 +86,7 @@ from
prefixoriginAS pathMEDcommunitieslarge communitiessince
- +
prefixfrom peerAS pathnext hoporiginMEDlocal prefcommunitieslarge communitiesbest
prefixfrom peerAS pathnext hoporiginMEDlocal prefcommunitieslarge communitiesRPKIbest
@@ -111,7 +114,7 @@ from const $ = id => document.getElementById(id); const esc = s => String(s).replace(/[&<>"]/g, c => ({'&':'&','<':'<','>':'>','"':'"'}[c])); - const state = { status: null, peers: [], local: [], received: [], refused: [] }; + const state = { status: null, peers: [], local: [], received: [], refused: [], rpki: null }; const DASH = '-'; // ---- helpers ----------------------------------------------------------- @@ -178,7 +181,7 @@ from ${up ? age(p.established_at_unix) : DASH}${up ? (p.four_octet_asn ? 'yes' : 'no') : DASH} ${up ? esc((p.address_families || []).join(', ')) + addPath(p) : DASH} ${up ? p.routes : DASH} - ${up ? (p.rejected_routes ? `${p.rejected_routes}` : 0) : DASH}${detail}`; + ${up ? (p.rejected_routes ? `${p.rejected_routes}` : 0) + (p.held_routes ? ` (${p.held_routes} held)` : '') : DASH}${detail}`; }).join('')); renderPeerChoices(); } @@ -247,12 +250,28 @@ from ? `showing ${state.received.length} of ${st.received_routes}, for ${st.received_prefixes} prefixes` : `${state.received.length}${st ? `, for ${st.received_prefixes} prefixes` : ''}`; if (!state.received.length) { - setHTML(b, `${looking ? 'No route matches.' : 'No routes received from peers.'}`); return; + setHTML(b, `${looking ? 'No route matches.' : 'No routes received from peers.'}`); return; } setHTML(b, state.received.map(r => { const s = r.summary; return ` ${esc(r.prefix)}${r.peer_name ? esc(r.peer_name) + ' ' : ''}${esc(r.peer_addr)} AS${r.peer_asn}${r.path_id ? ` path ${r.path_id}` : ''}${asPath(s.as_path)} ${s.next_hop ? esc(s.next_hop) : DASH}${s.origin ? esc(s.origin) : DASH}${s.med ?? DASH}${s.local_pref ?? DASH} - ${commList(s.communities, true)}${commList(s.large_communities)}${r.best ? 'best' : ''}`; }).join('')); + ${commList(s.communities, true)}${commList(s.large_communities)} + ${r.rpki ? `${esc(r.rpki)}` : DASH}${r.best ? 'best' : ''}`; }).join('')); + } + + const RPKI_CLASS = { valid: 'ok', invalid: 'bad', 'not found': 'dim' }; + const RPKI_STATE_CLASS = { synchronised: 'ok', synchronising: 'warn', connecting: 'bad' }; + function renderRpki() { + const r = state.rpki, b = $('rpki-body'); + if (!r) return; + if (!r.enabled) { setHTML(b, 'Not configured: routes are not validated.'); return; } + const data = r.updated_at_unix + ? `${r.vrps_ipv4} IPv4 and ${r.vrps_ipv6} IPv6 VRPs at serial ${r.serial ?? '?'}, confirmed ${age(r.updated_at_unix)} ago` + + (r.expires_at_unix ? `, dropped in ${until(r.expires_at_unix)} unless confirmed again` : '') + : 'no data: every route is "not found" and none is refused as invalid'; + setHTML(b, `Cache ${esc(r.cache || '?')}: ${esc(r.state)}` + + ` (RTR version ${r.protocol_version}). ${data}.` + + (r.last_error ? `
Last error: ${esc(r.last_error)}` : '')); } // For each peer a line that says how many and why, then its latest few. @@ -290,7 +309,8 @@ from renderReceived(); } async function loadRefused() { state.refused = await getJSON('/api/routes/rejected'); renderRefused(); } - async function loadAll() { await Promise.all([loadStatus(), loadPeers(), loadLocal(), loadReceived(), loadRefused()]); } + async function loadRpki() { state.rpki = await getJSON('/api/rpki'); renderRpki(); } + async function loadAll() { await Promise.all([loadStatus(), loadPeers(), loadLocal(), loadReceived(), loadRefused(), loadRpki()]); } // ---- events ------------------------------------------------------------ function logEvent(html) { diff --git a/docs/RFCs.md b/docs/RFCs.md index 832c0e0..459f33a 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -158,7 +158,6 @@ functional behaviour implemented beyond that. | RFC | Capability | Status | |-----|-----------|--------| | RFC 5291 | Outbound Route Filtering (ORF) | ๐ŸŸก | -| RFC 7911 | ADD-PATH | ๐ŸŸก | | RFC 8205 | BGPsec | ๐ŸŸก | --- @@ -171,4 +170,50 @@ https://datatracker.ietf.org/doc/html/rfc9687 | Feature | Status | |---------|--------| | `SendHoldTimer` error code (8) defined | โœ… | -| Send hold timer enforcement | โŒ | +| Send hold timer enforcement (a write that makes no progress ends the session) | โœ… | + +--- + +## Route Security + +### RFC 6811 โ€” BGP Prefix Origin Validation +https://datatracker.ietf.org/doc/html/rfc6811 + +| Feature | Status | +|---------|--------| +| Validation states (valid, invalid, not found) for routes from external peers | โœ… | +| Invalid routes refused, per peer (`accept_rpki_invalid` to take them) | โœ… | +| Routes revalidated when the VRPs change, in both directions | โœ… | +| Validation of routes from internal peers and of originated routes | โŒ | +| Validity as a step of route selection (RFC 8097 communities) | โŒ | + +### RFC 8210 โ€” The RPKI to Router Protocol, Version 1 (and RFC 6810, version 0) +https://datatracker.ietf.org/doc/html/rfc8210 + +| Feature | Status | +|---------|--------| +| Reset and serial queries, notifies, cache reset, incremental updates | โœ… | +| Refresh, retry and expire intervals; data dropped when it expires | โœ… | +| Fallback to version 0; several caches with failover | โœ… | +| Transport over plain TCP | โœ… | +| Transport over SSH, TLS, TCP-MD5 or TCP-AO | โŒ | +| Router keys (BGPsec) | ๐ŸŸก read and ignored | + +### RFC 7911 โ€” Advertisement of Multiple Paths in BGP (ADD-PATH) +https://datatracker.ietf.org/doc/html/rfc7911 + +| Feature | Status | +|---------|--------| +| Capability negotiation per family and direction | โœ… | +| Receiving several routes for a prefix | โœ… | +| Sending several routes for a prefix | โŒ (path identifiers are sent; one route each) | + +### RFC 4271 ยง9.1.2 / RFC 4486 / operational practice โ€” import policy + +| Feature | Status | +|---------|--------| +| Own AS in path refused (loop detection) | โœ… | +| Bogon prefixes and AS numbers, prefix length bounds, first-AS check | โœ… | +| Per-peer prefix and origin AS allow-lists | โœ… | +| Maximum prefixes with Cease / Maximum Number of Prefixes Reached and hold-down | โœ… | + diff --git a/testdata/server.textpb b/testdata/server.textpb index 8b61c18..c7bbbb4 100644 --- a/testdata/server.textpb +++ b/testdata/server.textpb @@ -53,6 +53,10 @@ originate { large_communities: "65000:1:2" } +# Route origin validation: refuse routes whose origin AS is not the one the +# holder of the addresses authorised. Point it at a validator's RTR server. +# rpki { caches { address: "127.0.0.1" port: 3323 } } + # Read-only status page, JSON API and metrics. It has no authentication: # keep it on loopback or behind a reverse proxy. http_endpoint { address: "127.0.0.1" port: 8080 }