Validate route origins against the RPKI and refuse invalid routes

With rpki { caches { ... } } in the configuration the server keeps the
validated ROA payloads of a validator and validates every route from a
peer outside its own AS (RFC 6811). An invalid route is refused, unless
its peer has accept_rpki_invalid. Valid and not-found routes are taken;
validity does not otherwise enter route selection.

Validity is not settled when a route arrives: the data changes every few
minutes. The routes follow, without the peer sending anything again.

- A route refused as invalid is held by its session, out of the table.
  When the data changes the session looks at what it holds, and what is
  no longer invalid goes into the table. A held route is the peer's to
  withdraw or replace like any other.
- A route in the table that becomes invalid is found by a task that
  looks through the table when the data changes, at the prefixes the
  changed VRPs cover, and is taken out and held by its session.
- If the data expires because no cache confirms it, nothing is invalid
  any more and everything held is taken.
- A session holds at most 200,000 routes; past that it asks the peer for
  a route refresh when the data changes.

So that the table can be looked through without parsing attributes, an
attribute set's key now has the route's origin AS. Memory per route is
unchanged at 52 bytes on the real table.

Shown: /api/rpki and a section of the status page with the cache, its
state and the age of the data; each route's validity; each peer's count
of held routes. Refusals are recorded under "RPKI invalid". Metrics:
bgp_rpki_vrps and bgp_rpki_updates_total.

Also: Negotiated says whether the peer supports route refresh, and
docs/RFCs.md is brought up to date.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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

crates/integration-tests/src/lib.rs

@@ -1,2 +1,3 @@
+pub mod test_cache;
 pub mod test_peer;
 pub mod test_server;

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<Mutex<(Vec<Vrp>, u32)>>,
+    /// The serial, for telling connected routers there is something new.
+    serial: watch::Sender<u32>,
+    task: JoinHandle<()>,
+}
+
+impl TestCache {
+    pub async fn start(vrps: Vec<Vrp>) -> 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<Vrp>) {
+        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<Mutex<(Vec<Vrp>, u32)>>,
+    mut serial: watch::Receiver<u32>,
+) {
+    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;
+        }
+    }
+}

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<Rpki>,
     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 }))

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<Duration>,
+    import: ImportPolicy,
+) -> Result<Setup> {
+    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<TestPeer> {
+    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<String> {
+    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<String> {
+        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::<Value, eyre::Report>(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(())
+}