bgp unnumbered: consolidate NDP-backed unnumbered state

.github/buildomat/jobs/test-ndp.sh

@@ -15,4 +15,4 @@ set -e
 
 source .github/buildomat/test-common.sh
 
-cargo nextest run -p ndp
+cargo nextest run -p ndp -p unnumbered

.github/buildomat/jobs/test-proptest.sh

@@ -6,7 +6,7 @@
 #: rust_toolchain = "stable"
 #: output_rules = [
 #:   "/work/*.log",
-#:   "/work/proptest-regressions/*",
+#:   "/work/proptest-regressions/**/*",
 #:   "/tmp/*.db",
 #: ]
 #: access_repos = [
@@ -30,21 +30,25 @@ cargo install cargo-nextest --version "${NEXTEST_VERSION}"
 source .github/buildomat/test-common.sh
 
 # Run property-based tests with high intensity (default is 256)
+export PROPTEST_CASES=1000000
 
-# RDB proptest suite
-pushd rdb
-PROPTEST_CASES=1000000 cargo nextest run --lib proptest
-cp ./*.log /work/ 2>/dev/null || true
-if [[ -d proptest-regressions ]]; then
-    cp -r proptest-regressions /work/rdb-proptest-regressions
-fi
-popd
-
-# BGP proptest suite
-pushd bgp
-PROPTEST_CASES=1000000 cargo nextest run --lib proptest
-cp ./*.log /work/ 2>/dev/null || true
-if [[ -d proptest-regressions ]]; then
-    cp -r proptest-regressions /work/bgp-proptest-regressions
-fi
-popd
+run_crate_proptests() {
+    local crate="$1"
+
+    pushd "${crate}"
+    cargo nextest run --lib proptest
+    cp ./*.log /work/ 2>/dev/null || true
+
+    if [[ -d proptest-regressions ]]; then
+        mkdir -p "/work/proptest-regressions/${crate}"
+        cp -R proptest-regressions/. "/work/proptest-regressions/${crate}/"
+    fi
+
+    popd
+}
+
+run_crate_proptests rdb
+
+run_crate_proptests bgp
+
+run_crate_proptests unnumbered

Cargo.toml

@@ -29,6 +29,7 @@ default-members = [
     "ndp",
     "xtask",
     "falcon-lab",
+    "unnumbered",
 ]
 
 members = [
@@ -59,6 +60,7 @@ members = [
     "ndp",
     "xtask",
     "falcon-lab",
+    "unnumbered",
 ]
 
 [workspace.dependencies]
@@ -78,6 +80,7 @@ ddm-api = { path = "ddm-api" }
 ddm-api-types = { path = "ddm-api-types" }
 ddm-api-types-versions = { path = "ddm-api-types/versions" }
 ddm-protocol = { path = "ddm-protocol" }
+unnumbered = { path = "unnumbered" }
 
 # External
 slog = { version = "2.8.2", features = ["max_level_trace", "release_max_level_debug"] }

bgp/Cargo.toml

@@ -26,6 +26,7 @@ uuid.workspace = true
 rand.workspace = true
 iddqd.workspace = true
 clap = { workspace = true, optional = true }
+unnumbered.workspace = true
 
 [target.'cfg(target_os = "illumos")'.dependencies]
 libnet = { workspace = true, optional = true }

bgp/src/connection.rs

@@ -7,16 +7,16 @@ use crate::{
     error::Error,
     messages::Message,
     router::SessionMap,
-    session::{FsmEvent, SessionInfo},
-    unnumbered::UnnumberedManager,
+    session::{FsmEvent, PeerId, SessionInfo, SessionRunner},
 };
-use slog::Logger;
+use slog::{Logger, debug, error};
 use std::{
     net::{SocketAddr, ToSocketAddrs},
     sync::{Arc, Mutex, mpsc::Sender},
     thread::JoinHandle,
     time::Duration,
 };
+use unnumbered::BgpUnnumbered;
 
 pub(crate) use mg_api_types::bgp::session::{
     ConnectionDirection, ConnectionId,
@@ -31,6 +31,60 @@ pub const MAX_MD5SIG_KEYLEN: usize = 80;
 #[cfg(target_os = "macos")]
 pub const MAX_MD5SIG_KEYLEN: usize = 80;
 
+/// Accepted inbound connection and its resolved session metadata.
+pub struct AcceptedConnection<Cnx: BgpConnection + 'static> {
+    pub connection: Cnx,
+    pub session_key: PeerId,
+    pub runner: Arc<SessionRunner<Cnx>>,
+    pub min_ttl: Option<u8>,
+    pub md5_key: Option<String>,
+}
+
+/// Resolve an incoming peer address to the session key used by BGP.
+pub fn resolve_session_key(
+    peer_addr: SocketAddr,
+    unnumbered_manager: Option<&Arc<dyn BgpUnnumbered>>,
+    log: &Logger,
+) -> PeerId {
+    // Try interface-based key for IPv6 link-local addresses
+    if let Some(mgr) = unnumbered_manager
+        && let SocketAddr::V6(v6_addr) = peer_addr
+        && v6_addr.ip().is_unicast_link_local()
+    {
+        match mgr.get_active_interface_by_scope(v6_addr.scope_id()) {
+            Ok(Some(interface)) => {
+                if interface.discovered_neighbor == Some(*v6_addr.ip()) {
+                    return PeerId::Interface(interface.interface);
+                } else if let Some(neighbor) = interface.discovered_neighbor {
+                    debug!(log,
+                        "incoming unnumbered peer does not match discovered NDP neighbor";
+                        "interface" => interface.interface.as_str(),
+                        "peer" => peer_addr,
+                        "discovered" => neighbor.to_string(),
+                    );
+                } else {
+                    debug!(log,
+                        "incoming unnumbered peer has no discovered NDP neighbor";
+                        "interface" => interface.interface.as_str(),
+                        "peer" => peer_addr,
+                    );
+                }
+            }
+            // Fallback to IP-based key on failure/missing interface
+            Ok(None) => {}
+            Err(e) => {
+                error!(log,
+                    "active unnumbered interface query error: {e}";
+                    "error" => format!("{e}")
+                );
+            }
+        }
+    }
+
+    // Default to IP-based key
+    PeerId::Ip(peer_addr.ip())
+}
+
 /// Implementors of this trait listen to and accept inbound BGP connections.
 pub trait BgpListener<Cnx: BgpConnection> {
     /// Bind to an address and listen for connections.
@@ -41,7 +95,7 @@ pub trait BgpListener<Cnx: BgpConnection> {
     fn bind<A: ToSocketAddrs>(
         addr: A,
         log: Logger,
-        unnumbered_manager: Option<Arc<dyn UnnumberedManager>>,
+        unnumbered_manager: Option<Arc<dyn BgpUnnumbered>>,
     ) -> Result<Self, Error>
     where
         Self: Sized;
@@ -55,7 +109,7 @@ pub trait BgpListener<Cnx: BgpConnection> {
         log: Logger,
         sessions: Arc<Mutex<SessionMap<Cnx>>>,
         timeout: Duration,
-    ) -> Result<Cnx, Error>;
+    ) -> Result<AcceptedConnection<Cnx>, Error>;
 
     /// Apply policy to an established connection. This is called by the
     /// Dispatcher after accept() returns and session lookup is completed.

bgp/src/connection_channel.rs

@@ -11,15 +11,14 @@ use crate::{
     IO_TIMEOUT,
     clock::ConnectionClock,
     connection::{
-        BgpConnection, BgpConnector, BgpListener, ConnectionDirection,
-        ConnectionId, ThreadState,
+        AcceptedConnection, BgpConnection, BgpConnector, BgpListener,
+        ConnectionDirection, ConnectionId, ThreadState, resolve_session_key,
     },
     error::Error,
     log::{connection_log, connection_log_lite},
     messages::Message,
     router::SessionMap,
-    session::{ConnectionEvent, FsmEvent, PeerId, SessionInfo},
-    unnumbered::UnnumberedManager,
+    session::{ConnectionEvent, FsmEvent, SessionInfo},
 };
 use mg_common::lock;
 use slog::{Logger, info};
@@ -34,6 +33,7 @@ use std::{
     thread::{JoinHandle, spawn},
     time::{Duration, Instant},
 };
+use unnumbered::BgpUnnumbered;
 
 const UNIT_CONNECTION: &str = "connection_channel";
 
@@ -193,33 +193,14 @@ pub fn count_failed_connections_to(peer: SocketAddr) -> usize {
 pub struct BgpListenerChannel {
     listener: Listener,
     bind_addr: SocketAddr,
-    unnumbered_manager: Option<Arc<dyn UnnumberedManager>>,
-}
-
-impl BgpListenerChannel {
-    /// Resolve incoming peer address to appropriate PeerId.
-    fn resolve_session_key(&self, peer_addr: SocketAddr) -> PeerId {
-        // Try interface-based routing for IPv6 link-local addresses
-        if let Some(ref mgr) = self.unnumbered_manager
-            && let SocketAddr::V6(v6_addr) = peer_addr
-            && v6_addr.ip().is_unicast_link_local()
-        {
-            let scope_id = v6_addr.scope_id();
-            if let Some(interface) = mgr.get_interface_by_scope(scope_id) {
-                return PeerId::Interface(interface);
-            }
-        }
-
-        // Default to IP-based routing
-        PeerId::Ip(peer_addr.ip())
-    }
+    unnumbered_manager: Option<Arc<dyn BgpUnnumbered>>,
 }
 
 impl BgpListener<BgpConnectionChannel> for BgpListenerChannel {
     fn bind<A: ToSocketAddrs>(
         addr: A,
         log: Logger,
-        unnumbered_manager: Option<Arc<dyn UnnumberedManager>>,
+        unnumbered_manager: Option<Arc<dyn BgpUnnumbered>>,
     ) -> Result<Self, Error>
     where
         Self: Sized,
@@ -245,30 +226,41 @@ impl BgpListener<BgpConnectionChannel> for BgpListenerChannel {
         log: Logger,
         sessions: Arc<Mutex<SessionMap<BgpConnectionChannel>>>,
         timeout: Duration,
-    ) -> Result<BgpConnectionChannel, Error> {
+    ) -> Result<AcceptedConnection<BgpConnectionChannel>, Error> {
         let (peer, endpoint) = self.listener.accept(timeout)?;
 
         let local = self.bind_addr;
 
-        // Resolve peer address to appropriate PeerId (IP or Interface)
-        let key = self.resolve_session_key(peer);
+        let key =
+            resolve_session_key(peer, self.unnumbered_manager.as_ref(), &log);
 
         let runner = lock!(sessions)
             .get(&key)
             .cloned()
             .ok_or_else(|| Error::UnknownPeer(key.clone()))?;
 
-        let config = lock!(runner.session);
-        Ok(BgpConnectionChannel::with_conn(
-            local,
-            peer,
-            endpoint,
-            runner.event_tx.clone(),
-            IO_TIMEOUT,
-            log,
-            ConnectionDirection::Inbound,
-            &config,
-        ))
+        let (connection, min_ttl, md5_key) = {
+            let config = lock!(runner.session);
+            let connection = BgpConnectionChannel::with_conn(
+                local,
+                peer,
+                endpoint,
+                runner.event_tx.clone(),
+                IO_TIMEOUT,
+                log,
+                ConnectionDirection::Inbound,
+                &config,
+            );
+            (connection, config.min_ttl, config.md5_auth_key.clone())
+        };
+
+        Ok(AcceptedConnection {
+            connection,
+            session_key: key,
+            runner,
+            min_ttl,
+            md5_key,
+        })
     }
 
     fn apply_policy(