Judge every route a peer announces: import policy, prefix limit, hold-down

A configured peer could announce any prefix with any path, in any
number, and all of it was stored and could be chosen as best. Each
peer now has an import policy, and every route goes through it before
it reaches the routing table.

Checks that hold for any peer unless turned off for it:
- no prefixes that are not routed on the Internet (private, loopback,
  link-local, documentation, multicast, reserved; for IPv6, anything
  outside global unicast);
- nothing longer than /24 or /48, no default route, nothing shorter
  than /8 or /16;
- no reserved, private or documentation AS number in the path;
- a path that begins with the peer's own AS (turned off for a route
  server), and of at most 64 AS numbers.

Always: a route with our own AS number in its path is refused. RFC 4271
requires this loop check and it was missing.

Per peer, if configured: the prefixes it may announce, each with an
optional longest more-specific, and the AS numbers its routes may
originate from. This is the list generated from the routing registries.

max_prefixes limits how many routes a peer may have. One more ends the
session with Cease / Maximum Number of Prefixes Reached, and the peer
is kept down for hold_down_seconds (ten minutes unless set): its
connections are not taken and it is not dialled. Refused routes do not
count towards the limit.

A refused route is withdrawn rather than ignored, so that it replaces
whatever the peer had announced for it before.

On a session with two-octet AS numbers the path is judged with its
AS4_PATH merged in, so that the real AS numbers are what is checked.

Peers that are not configured (accept_unconfigured_peers, for test
networks) get a policy that takes everything but loops.

Refusals are counted in bgp_routes_rejected_total by reason. The status
page and /api/peers show, for each session, how many routes the peer
has and how many were refused. The server warns at startup about peers
with no limit.

The prefix check costs about 4 ns a route with the default policy; the
end-to-end benchmark is unchanged.

Also: the benchmark's server AS is one that is in none of its generated
paths, and a table test that depended on the hasher's seed is fixed.

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

crates/integration-tests/tests/add_path.rs

@@ -16,6 +16,7 @@ use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
 use bgp_rib::{LocalRoute, LocalRouteAttrs};
 use bgp_server::config::PeerConfig;
 use bgp_server::family::{FamilySet, SUPPORTED};
+use bgp_server::policy::ImportPolicy;
 use eyre::Result;
 use serde_json::Value;
 
@@ -46,6 +47,7 @@ async fn server_offering(
             local_address: None,
             families: FamilySet::supported(),
             add_path,
+            import: ImportPolicy::permissive(),
         }];
         c.originate = originate;
         // Keep the dialer out of the way.

crates/integration-tests/tests/admission.rs

@@ -8,6 +8,7 @@ use bgp_integration_tests::test_peer::TestPeer;
 use bgp_integration_tests::test_server::TestServer;
 use bgp_server::config::PeerConfig;
 use bgp_server::family::FamilySet;
+use bgp_server::policy::ImportPolicy;
 use eyre::Result;
 use tokio::net::TcpListener;
 
@@ -24,15 +25,13 @@ fn peer_at(endpoint: SocketAddr) -> PeerConfig {
         local_address: None,
         families: FamilySet::supported(),
         add_path: None,
+        import: ImportPolicy::permissive(),
     }
 }
 
 /// Whether the server closed the connection without sending anything.
 async fn closed_unanswered(peer: &mut TestPeer) -> bool {
-    matches!(
-        tokio::time::timeout(TIMEOUT, peer.recv()).await,
-        Ok(Err(_))
-    )
+    matches!(tokio::time::timeout(TIMEOUT, peer.recv()).await, Ok(Err(_)))
 }
 
 /// A host that is not a configured peer gets nothing: no OPEN to answer,

crates/integration-tests/tests/collision.rs

@@ -18,6 +18,7 @@ use bgp_rib::{LocalRoute, LocalRouteAttrs};
 use bgp_server::collision::Direction;
 use bgp_server::config::PeerConfig;
 use bgp_server::family::FamilySet;
+use bgp_server::policy::ImportPolicy;
 use eyre::Result;
 use tokio::net::TcpListener;
 
@@ -42,6 +43,7 @@ async fn server_dialing(listener: &TcpListener) -> Result<TestServer> {
             local_address: None,
             families: FamilySet::supported(),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
     })
     .await

crates/integration-tests/tests/dialer.rs

@@ -12,6 +12,7 @@ use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::{LocalRoute, LocalRouteAttrs};
 use bgp_server::config::PeerConfig;
 use bgp_server::family::FamilySet;
+use bgp_server::policy::ImportPolicy;
 use bgp_server::session::DialState;
 use eyre::Result;
 use tokio::net::TcpListener;
@@ -33,6 +34,7 @@ async fn server_dialing(listener: &TcpListener, remote_asn: u32) -> Result<TestS
             local_address: None,
             families: FamilySet::supported(),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
     })
     .await
@@ -187,6 +189,7 @@ async fn test_inbound_from_configured_address_uses_config() -> Result<()> {
             local_address: None,
             families: FamilySet::supported(),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
     })
     .await?;
@@ -235,6 +238,7 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> {
             local_address: Some(Ipv4Addr::LOCALHOST.into()),
             families: FamilySet::supported(),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
     })
     .await?;
@@ -252,6 +256,7 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> {
             local_address: Some(Ipv4Addr::new(192, 0, 2, 1).into()),
             families: FamilySet::supported(),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
     })
     .await?;

crates/integration-tests/tests/import_policy.rs

@@ -0,0 +1,280 @@
+//! What is taken from a peer: the checks every route goes through, and the
+//! limit on how many a peer may have.
+
+use std::collections::BTreeSet;
+use std::net::Ipv4Addr;
+use std::time::Duration;
+
+use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_server::TestServer;
+use bgp_packet::errors::{BgpError, CeaseSubcode};
+use bgp_packet::message::{
+    AsPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute, PathAttribute,
+    UpdateMessage,
+};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_server::config::PeerConfig;
+use bgp_server::family::FamilySet;
+use bgp_server::policy::{ImportPolicy, PrefixSet};
+use bgp_server::session::DialState;
+use eyre::Result;
+use tokio::net::TcpListener;
+
+/// AS numbers as they are on the Internet: the checks refuse the private
+/// ones the other tests use.
+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()
+}
+
+/// A server with one configured peer, at the address the test connects
+/// from, whose routes are judged by `import`. Returns the listener that
+/// keeps the server's own dialling occupied.
+async fn server_with(import: ImportPolicy) -> Result<(TestServer, TcpListener)> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+    let endpoint = listener.local_addr()?;
+    let server = TestServer::start_with(LOCAL_ASN, |c| {
+        c.accept_unconfigured_peers = false;
+        c.connect_retry = Duration::from_millis(50);
+        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?;
+    Ok((server, 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 IPv4 `prefixes` with the AS path `path`.
+async fn announce(peer: &mut TestPeer, prefixes: &[&str], path: &[u32]) -> Result<()> {
+    peer.send(&Message::Update(UpdateMessage {
+        withdrawn_routes: vec![],
+        path_attributes: vec![
+            PathAttribute::Origin(OriginPathAttribute::IGP),
+            AsPathAttribute::from_asns(path.to_vec()),
+            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
+        ],
+        announced_routes: prefixes.iter().map(|p| prefix(p).into()).collect(),
+    }))
+    .await
+}
+
+/// Waits until the server has dealt with `count` announced routes of the
+/// session's, taken or refused, and returns the prefixes in its table.
+async fn table_after(server: &TestServer, taken: u64, refused: u64) -> Result<BTreeSet<String>> {
+    let counters = server.registry.list()[0].counters.clone();
+    tokio::time::timeout(TIMEOUT, async {
+        while (counters.routes(), counters.rejected()) != (taken, refused) {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .map_err(|_| {
+        eyre::eyre!(
+            "the server took {} routes and refused {}; expected {taken} and {refused}",
+            counters.routes(),
+            counters.rejected()
+        )
+    })?;
+    Ok(server
+        .received_routes()
+        .iter()
+        .map(|route| route.prefix.to_string())
+        .collect())
+}
+
+fn set(prefixes: &[&str]) -> BTreeSet<String> {
+    prefixes.iter().map(|p| p.to_string()).collect()
+}
+
+/// With nothing configured for a peer but its address and AS, its routes
+/// still go through the checks that hold for anyone: only space that is
+/// routed on the Internet, nothing too specific, a path that begins with
+/// the peer and has no reserved AS in it, and never one that has been
+/// through us. Everything a peer sent used to be taken.
+#[tokio::test]
+async fn test_routes_that_no_peer_should_announce_are_refused() -> Result<()> {
+    let (server, _listener) = server_with(ImportPolicy::default()).await?;
+    let mut peer = established(&server).await?;
+
+    // One UPDATE with good and bad prefixes: the good ones are taken.
+    announce(
+        &mut peer,
+        &[
+            "193.0.0.0/21",
+            "10.0.0.0/8",    // private
+            "193.0.10.0/25", // more specific than is routed
+            "0.0.0.0/0",     // default
+            "193.0.16.0/24",
+        ],
+        &[PEER_ASN],
+    )
+    .await?;
+    assert_eq!(
+        table_after(&server, 2, 3).await?,
+        set(&["193.0.0.0/21", "193.0.16.0/24"])
+    );
+
+    // Paths that are refused whatever the prefix: not the peer's own,
+    // with a private AS in it, and one that has been through us.
+    announce(&mut peer, &["194.0.1.0/24"], &[174, PEER_ASN]).await?;
+    announce(&mut peer, &["194.0.2.0/24"], &[PEER_ASN, 64512]).await?;
+    announce(&mut peer, &["194.0.3.0/24"], &[PEER_ASN, LOCAL_ASN, 174]).await?;
+    assert_eq!(
+        table_after(&server, 2, 6).await?,
+        set(&["193.0.0.0/21", "193.0.16.0/24"])
+    );
+    Ok(())
+}
+
+/// A route that is refused takes the place of the one the peer had
+/// announced before, which goes: the peer no longer has the old route,
+/// and the new one is not acceptable.
+#[tokio::test]
+async fn test_a_refused_route_replaces_the_one_that_was_taken() -> Result<()> {
+    let (server, _listener) = server_with(ImportPolicy::default()).await?;
+    let mut peer = established(&server).await?;
+
+    announce(&mut peer, &["193.0.0.0/21", "193.0.16.0/24"], &[PEER_ASN]).await?;
+    assert_eq!(table_after(&server, 2, 0).await?.len(), 2);
+
+    announce(&mut peer, &["193.0.0.0/21"], &[PEER_ASN, 64512]).await?;
+    assert_eq!(table_after(&server, 1, 1).await?, set(&["193.0.16.0/24"]));
+    Ok(())
+}
+
+/// A peer with a list of what it may announce is held to it: its own
+/// prefixes, originated by the AS numbers given.
+#[tokio::test]
+async fn test_a_peer_is_held_to_its_prefix_list_and_origins() -> Result<()> {
+    let mut prefixes = PrefixSet::default();
+    prefixes.allow(prefix("193.0.0.0/21"), 24);
+    let (server, _listener) = server_with(ImportPolicy {
+        prefixes: Some(prefixes),
+        origin_asns: Some([PEER_ASN, 2121].into()),
+        ..ImportPolicy::default()
+    })
+    .await?;
+    let mut peer = established(&server).await?;
+
+    // Its own, and a more specific that the list allows.
+    announce(&mut peer, &["193.0.0.0/21", "193.0.4.0/24"], &[PEER_ASN]).await?;
+    // A customer of its that is listed as an origin.
+    announce(&mut peer, &["193.0.6.0/24"], &[PEER_ASN, 2121]).await?;
+    // Someone else's prefix, as a hijack or a leak would be.
+    announce(&mut peer, &["8.8.8.0/24"], &[PEER_ASN]).await?;
+    // Its own prefix, but passed on from an origin that is not listed.
+    announce(&mut peer, &["193.0.7.0/24"], &[PEER_ASN, 15169]).await?;
+
+    assert_eq!(
+        table_after(&server, 3, 2).await?,
+        set(&["193.0.0.0/21", "193.0.4.0/24", "193.0.6.0/24"])
+    );
+    Ok(())
+}
+
+/// A peer that sends more routes than it is allowed is cut off and told
+/// why, and kept down for a while: neither its connections are taken nor
+/// is it dialled. A session that came straight back would only fill up
+/// again.
+#[tokio::test]
+async fn test_a_peer_over_its_limit_is_cut_off_and_kept_down() -> Result<()> {
+    let (server, listener) = server_with(ImportPolicy {
+        max_prefixes: Some(3),
+        hold_down: Duration::from_millis(1500),
+        ..ImportPolicy::default()
+    })
+    .await?;
+    // The server dials the peer as well; that connection is left unanswered
+    // until the end.
+    let dialled = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
+
+    let mut peer = established(&server).await?;
+    // Refused routes do not count towards the limit, and replacing a
+    // route does not either.
+    announce(
+        &mut peer,
+        &["193.0.0.0/24", "193.0.1.0/24", "10.0.0.0/8"],
+        &[PEER_ASN],
+    )
+    .await?;
+    announce(&mut peer, &["193.0.1.0/24", "193.0.2.0/24"], &[PEER_ASN]).await?;
+    assert_eq!(table_after(&server, 3, 1).await?.len(), 3);
+
+    announce(&mut peer, &["193.0.3.0/24"], &[PEER_ASN]).await?;
+    let notification = tokio::time::timeout(TIMEOUT, async {
+        loop {
+            if let Message::Notification(notification) = peer.recv().await? {
+                return Ok::<_, eyre::Report>(notification);
+            }
+        }
+    })
+    .await??;
+    assert_eq!(
+        notification.error,
+        BgpError::Cease(CeaseSubcode::MaxPrefixes)
+    );
+    drop(peer);
+    tokio::time::timeout(TIMEOUT, async {
+        while !server.received_routes().is_empty() {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+
+    // Held down: a new connection gets nothing.
+    let mut early = TestPeer::connect(server.addr).await?;
+    assert!(
+        matches!(
+            tokio::time::timeout(TIMEOUT, early.recv()).await,
+            Ok(Err(_))
+        ),
+        "a peer that is held down is not answered"
+    );
+    // Nor is it dialled: once the connection the server had open to it is
+    // gone, the dialer waits out the hold-down and says so.
+    drop(dialled);
+    let reason = tokio::time::timeout(TIMEOUT, async {
+        loop {
+            if let Some((_, DialState::Backoff { last_error, .. })) =
+                server.registry.configured_peers().into_iter().next()
+                && last_error.contains("held down")
+            {
+                return last_error;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    assert!(reason.contains("limit of routes"), "{reason}");
+
+    // When the time is up the peer is dialled again, and taken back.
+    tokio::time::timeout(TIMEOUT, listener.accept()).await??;
+    let mut later = TestPeer::connect(server.addr).await?;
+    later
+        .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90)
+        .await?;
+    Ok(())
+}

crates/integration-tests/tests/negotiation.rs

@@ -17,6 +17,7 @@ use bgp_primitives::ip_prefix::IpPrefix;
 use bgp_rib::{LocalRoute, LocalRouteAttrs};
 use bgp_server::config::PeerConfig;
 use bgp_server::family::{FamilySet, SUPPORTED};
+use bgp_server::policy::ImportPolicy;
 use eyre::{Result, bail};
 
 const LOCAL_ASN: u32 = 65000;
@@ -189,6 +190,7 @@ async fn test_no_family_in_common_refuses_the_session() -> Result<()> {
             local_address: None,
             families: only(AfiSafi::IPV6_UNICAST),
             add_path: None,
+            import: ImportPolicy::permissive(),
         }];
         // Keep the dialer out of the way.
         c.connect_retry = Duration::from_secs(3600);