implement compatibility with MP-BGP updates

nxthdr · Nov 3, 2024 · dc79489 · dc79489
1 parent 4825cae
commit dc79489
Show file tree

Hide file tree

Showing 5 changed files with 59 additions and 9 deletions.
diff --git a/src/bmp.rs b/src/bmp.rs
@@ -4,7 +4,7 @@ use bgpkit_parser::parser::bmp::messages::{BmpMessage, BmpMessageBody};
 use bytes::Bytes;
 use chrono::Utc;
 use config::Config;
-use log::debug;
+use log::{debug, info};
 use std::io;
 use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
@@ -77,12 +77,12 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
 
     match message.message_body {
         BmpMessageBody::PeerUpNotification(body) => {
-            debug!("PEER_UP_NOTIFICATION - {:?}", body);
+            debug!("{:?}", body);
             // Simply add the peer if we did not see it before
             router.add_peer(&peer);
         }
         BmpMessageBody::RouteMonitoring(body) => {
-            debug!("ROUTE_MONITORING - {:?}", body);
+            debug!("{:?}", body);
             let potential_updates = decode_updates(body).unwrap_or(Vec::new());
 
             let mut legitimate_updates = Vec::new();
@@ -96,12 +96,12 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
             // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in legitimate_updates {
                 let update = format_update(&router, &peer, &update);
-                debug!("UPDATE - {:?}", update);
+                info!("{:?}", update);
                 send_to_kafka(&kafka_host, &kafka_topic, update.as_bytes());
             }
         }
         BmpMessageBody::PeerDownNotification(body) => {
-            debug!("PEER_DOWN_NOTIFICATION - {:?}", body);
+            debug!("{:?}", body);
             // Remove the peer and the associated prefixes
             // To do so, we start by emiting synthetic withdraw updates
             let mut synthetic_updates = Vec::new();
@@ -120,7 +120,7 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
             // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in synthetic_updates {
                 let update = format_update(&router, &peer, &update);
-                debug!("UPDATE - {:?}", update);
+                info!("{:?}", update);
                 send_to_kafka(&kafka_host, &kafka_topic, update.as_bytes());
             }
         }

diff --git a/src/update.rs b/src/update.rs
@@ -21,6 +21,7 @@ pub fn decode_updates(message: RouteMonitoring) -> Option<Vec<Update>> {
 
     match message.bgp_message {
         bgpkit_parser::models::BgpMessage::Update(bgp_update) => {
+            // https://datatracker.ietf.org/doc/html/rfc4271
             let mut prefixes_to_update = Vec::new();
             for prefix in bgp_update.announced_prefixes {
                 prefixes_to_update.push((prefix, true));
@@ -29,13 +30,33 @@ pub fn decode_updates(message: RouteMonitoring) -> Option<Vec<Update>> {
                 prefixes_to_update.push((prefix, false));
             }
 
+            // https://datatracker.ietf.org/doc/html/rfc4760
             let attributes = bgp_update.attributes;
+            match attributes.get_reachable_nlri() {
+                Some(nlri) => {
+                    for prefix in &nlri.prefixes {
+                        prefixes_to_update.push((*prefix, true));
+                    }
+                }
+                None => (),
+            }
+            match attributes.get_unreachable_nlri() {
+                Some(nlri) => {
+                    for prefix in &nlri.prefixes {
+                        prefixes_to_update.push((*prefix, false));
+                    }
+                }
+                None => (),
+            }
+
+            // Get the other attributes
             let origin = attributes.origin();
             let path = match attributes.as_path() {
                 Some(path) => Some(path.clone()),
                 None => None,
             };
             let communities: Vec<MetaCommunity> = attributes.iter_communities().collect();
+
             for (prefix, announced) in prefixes_to_update {
                 updates.push(Update {
                     prefix: prefix,

diff --git a/testbed/compose.yml b/testbed/compose.yml
@@ -25,7 +25,7 @@ services:
         ipv4_address: 10.0.0.20
   risotto:
     build: ..
-    command: --debug --config /config/risotto
+    command: --config /config/risotto
     volumes:
       - ./config/risotto/risotto.yml:/config/risotto.yml
     networks:

diff --git a/testbed/config/bird/bird_10.conf b/testbed/config/bird/bird_10.conf
@@ -4,18 +4,24 @@ log syslog all;
 
 protocol device { scan time 5; }
 protocol direct { ipv4; }
+protocol direct { ipv6; }
 
 protocol bmp {
     station address ip 10.0.0.100 port 4000;
     monitoring rib in pre_policy;
     monitoring rib in post_policy;
 }
 
-protocol static Local {
+protocol static LocalV4 {
     ipv4;
     route 172.16.10.0/24 reject;
 }
 
+protocol static LocalV6 {
+    ipv6;
+    route fd49:6166:cafe::/48 reject;
+}
+
 protocol bgp Uplink {
     local 10.0.0.10 as 65010;
     neighbor 10.0.0.20 as 65020;
@@ -28,4 +34,13 @@ protocol bgp Uplink {
             } else reject;
         };
     };
+    ipv6 {
+        import all;
+        export filter {
+            if (net ~ [ fd49:6166:cafe::/48+ ]) then {
+                bgp_community.add((65010,41));
+                accept;
+            } else reject;
+        };
+    };
 }
diff --git a/testbed/config/bird/bird_20.conf b/testbed/config/bird/bird_20.conf
@@ -4,18 +4,24 @@ log syslog all;
 
 protocol device { scan time 5; }
 protocol direct { ipv4; }
+protocol direct { ipv6; }
 
 protocol bmp {
     station address ip 10.0.0.100 port 4000;
     monitoring rib in pre_policy;
     monitoring rib in post_policy;
 }
 
-protocol static Local {
+protocol static LocalV4 {
     ipv4;
     route 172.16.20.0/24 reject;
 }
 
+protocol static LocalV6 {
+    ipv6;
+    route fd49:6166:beef::/48 reject;
+}
+
 protocol bgp Uplink {
     local 10.0.0.20 as 65020;
     neighbor 10.0.0.10 as 65010;
@@ -26,4 +32,12 @@ protocol bgp Uplink {
             else reject;
         };
     };
+    ipv6 {
+        import all;
+        export filter {
+            if (net ~ [ fd49:6166:beef::/48+ ]) then {
+                accept;
+            } else reject;
+        };
+    };
 }