Merge pull request #21 from FastNetMon/fix/bmp-replay-router-boundary

Stop BMP replay identity capture at the router boundary

Author: nuclearcat

src/ingress/register.rs

@@ -284,8 +284,10 @@ impl Register {
         Some((emit, entries.get(&emit)?.history_generation))
     }
 
-    /// Copy a bounded lineage under one lock: child -> peer -> router.
-    /// Missing ancestors or cycles cannot be silently encoded as default peers.
+    /// Copy a bounded wire-identity lineage under one lock: child -> peer ->
+    /// router. A router's parent is an opaque unit ID, not encoding metadata.
+    /// Missing required ancestors or cycles cannot be silently encoded as
+    /// default peers.
     pub(crate) fn snapshot_lineage(
         &self,
         id: IngressId,
@@ -295,6 +297,9 @@ impl Register {
         let first = inline.cloned().or_else(|| entries.get(&id).cloned())?;
         let mut snapshot = vec![(id, first)];
         while let Some(parent) = snapshot.last()?.1.parent_ingress {
+            if snapshot.last()?.1.ingress_type == Some(IngressType::Bmp) {
+                break;
+            }
             if snapshot.len() >= 8
                 || snapshot.iter().any(|(id, _)| *id == parent)
             {

src/units/bmp_tcp_out/replay.rs

@@ -183,6 +183,90 @@ mod tests {
             RouteStatus::Active,
         ))
     }
+
+    #[test]
+    fn missing_required_router_still_rejects_replay_capture() {
+        let registry = Register::default();
+        registry.update_info(7, peer(65001).with_parent_ingress(3u32));
+        assert!(SnapshotCache::default()
+            .capture(&route(7), &registry)
+            .is_none());
+    }
+
+    #[tokio::test]
+    async fn replay_stops_at_router_before_opaque_unit_parent() {
+        // BMP input reserves a unit ID without an IngressInfo entry. The
+        // router points at it for internal ownership; it is not wire identity.
+        for source in [7, 8] {
+            let registry = Register::default();
+            let unit = registry.register();
+            registry.update_info(
+                3,
+                IngressInfo::new()
+                    .with_ingress_type(IngressType::Bmp)
+                    .with_parent_ingress(unit)
+                    .with_name("example-router"),
+            );
+            registry.update_info(7, peer(65001).with_parent_ingress(3u32));
+            registry.update_info(
+                8,
+                IngressInfo::new()
+                    .with_ingress_type(IngressType::BgpPath)
+                    .with_parent_ingress(7u32)
+                    .with_path_id(33u32),
+            );
+            assert!(registry.get(unit).is_none());
+
+            let (tx, mut rx) = mpsc::channel(8);
+            let client = Arc::new(ClientState::new(
+                "127.0.0.1:1234".parse().unwrap(),
+                tx,
+                10,
+                1_000_000,
+            ));
+            assert!(matches!(
+                client
+                    .buffer_update_if_dumping(route(source), &registry)
+                    .await,
+                super::super::client_state::BufferUpdateResult::Buffered
+            ));
+            for id in [8, 7, 3] {
+                registry.remove(id);
+            }
+            let queued = client.take_buffered_updates().await;
+            assert_eq!(queued.len(), 1);
+            let identities = &queued[0].identities;
+            assert_eq!(identities.session_identity(source), Some((7, 0)));
+            assert!(identities.get(unit).is_none());
+            assert_eq!(
+                identities.get(3).unwrap().name.as_deref(),
+                Some("example-router")
+            );
+            assert!(
+                send_update_to_client(
+                    &client,
+                    &queued[0].update,
+                    identities,
+                    true,
+                    FanInPeerDistinguisher::default(),
+                    false,
+                )
+                .await
+            );
+            let up = rx.try_recv().unwrap();
+            assert_eq!(up[5], 3);
+            assert_eq!(
+                u32::from_be_bytes(up[32..36].try_into().unwrap()),
+                65001
+            );
+            assert!(up
+                .windows(b"example-router".len())
+                .any(|part| part == b"example-router"));
+            assert_eq!(rx.try_recv().unwrap()[5], 0);
+            assert!(rx.try_recv().is_err());
+        }
+    }
+
     #[tokio::test]
     async fn replay_keeps_original_headers_across_retirement_and_same_id_reconnect(
     ) {