Implemented basic input filtering.

crates/bgp_packet/Cargo.toml

@@ -14,6 +14,8 @@ workspace = true
 [dependencies]
 byteorder            = "1.4.3"
 bytes.workspace      = true
+eyre.workspace       = true
 nom                  = "7.1"
 serde.workspace      = true
+serde_json.workspace = true
 tokio-util           = { version = "0.7.10", features = ["codec"] }

crates/bgp_packet/src/nlri.rs

@@ -17,11 +17,13 @@ use crate::traits::BGPParserError;
 use crate::traits::ParserContext;
 use crate::traits::ReadablePacket;
 use crate::traits::WritablePacket;
+
 use nom::bytes::complete::take;
 use nom::number::complete::be_u8;
 use nom::Err::Failure;
 use nom::IResult;
-use serde::Serialize;
+use serde::de;
+use serde::{Deserialize, Serialize};
 use std::convert::TryFrom;
 use std::convert::TryInto;
 use std::fmt;
@@ -31,7 +33,7 @@ use std::str::FromStr;
 
 // NLRI here is the Neighbor Link Reachability Information from RFC 4271.
 // Other NLRIs such as MP Reach NLRI are implemented as path attributes.
-#[derive(Debug, PartialEq, Eq, Clone, Serialize, Hash)]
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
 pub struct NLRI {
     pub afi: AddressFamilyIdentifier,
     pub prefixlen: u8,
@@ -87,6 +89,37 @@ impl ReadablePacket for NLRI {
     }
 }
 
+impl WritablePacket for NLRI {
+    fn to_wire(&self, _: &ParserContext) -> Result<Vec<u8>, &'static str> {
+        let mut buf: Vec<u8> = Vec::new();
+        buf.push(self.prefixlen);
+        buf.extend(self.prefix.as_slice());
+        Ok(buf)
+    }
+    fn wire_len(&self, _: &ParserContext) -> Result<u16, &'static str> {
+        Ok(1 + self.prefix.len() as u16)
+    }
+}
+
+impl Serialize for NLRI {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for NLRI {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        Self::try_from(String::deserialize(deserializer)?.as_str())
+            .map_err(|e| de::Error::custom(e))
+    }
+}
+
 impl TryFrom<NLRI> for Ipv6Addr {
     type Error = String;
 
@@ -165,9 +198,9 @@ impl TryInto<IpAddr> for NLRI {
     }
 }
 
-impl TryFrom<String> for NLRI {
+impl TryFrom<&str> for NLRI {
     type Error = String;
-    fn try_from(value: String) -> Result<Self, Self::Error> {
+    fn try_from(value: &str) -> Result<Self, Self::Error> {
         let parts: Vec<&str> = value.split("/").collect();
         if parts.len() != 2 {
             return Err(format!("Expected ip_addr/prefixlen but got: {}", value));
@@ -211,18 +244,6 @@ impl TryFrom<String> for NLRI {
     }
 }
 
-impl WritablePacket for NLRI {
-    fn to_wire(&self, _: &ParserContext) -> Result<Vec<u8>, &'static str> {
-        let mut buf: Vec<u8> = Vec::new();
-        buf.push(self.prefixlen);
-        buf.extend(self.prefix.as_slice());
-        Ok(buf)
-    }
-    fn wire_len(&self, _: &ParserContext) -> Result<u16, &'static str> {
-        Ok(1 + self.prefix.len() as u16)
-    }
-}
-
 impl fmt::Display for NLRI {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self.afi {
@@ -322,7 +343,7 @@ mod tests {
         ];
 
         for (i, case) in cases.iter().enumerate() {
-            let parsed_nlri = NLRI::try_from(case.0.clone()).unwrap();
+            let parsed_nlri = NLRI::try_from(case.0.as_str()).unwrap();
             assert_eq!(parsed_nlri.prefix, case.1, "Check prefix match ({})", i);
             assert_eq!(
                 parsed_nlri.prefixlen, case.2,
@@ -335,6 +356,13 @@ mod tests {
                 "Check std::fmt::Display match ({})",
                 i
             );
+
+            // Check that roundtripping via JSON serialize / deserialize is correct.
+            let json_encoded = serde_json::to_string(&parsed_nlri).unwrap();
+            assert_eq!(json_encoded[1..json_encoded.len() - 1], case.3);
+
+            let reparsed: NLRI = serde_json::from_str(&json_encoded).unwrap();
+            assert_eq!(reparsed, parsed_nlri);
         }
     }
 

crates/bgp_packet/src/path_attributes.rs

@@ -24,6 +24,7 @@ use byteorder::NetworkEndian;
 use nom::number::complete::{be_u16, be_u32, be_u8};
 use nom::Err::Failure;
 use nom::IResult;
+use serde::Deserialize;
 use serde::Serialize;
 use std::convert::TryInto;
 use std::fmt;
@@ -739,7 +740,7 @@ pub struct LargeCommunitiesPathAttribute {
     pub values: Vec<LargeCommunitiesPayload>,
 }
 
-#[derive(Debug, PartialEq, Eq, Clone, Serialize)]
+#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
 pub struct LargeCommunitiesPayload {
     pub global_admin: u32,
     pub ld1: u32,

crates/server/src/config.rs

@@ -12,7 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use bgp_packet::constants::{AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier};
+use bgp_packet::{
+    constants::{AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier},
+    nlri::NLRI,
+    path_attributes::{LargeCommunitiesPathAttribute, LargeCommunitiesPayload},
+};
 use serde::{Deserialize, Serialize};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
@@ -55,6 +59,34 @@ pub struct PeerConfig {
     // Announcements is a hardcoded list of BGP updates to send
     // to the peer.
     pub announcements: Vec<PrefixAnnouncement>,
+
+    /// filter_in is applied to every announcement received by the peer before being accepted into Loc-RIB.
+    pub filter_in: Vec<(FilterMatcher, FilterAction)>,
+
+    /// filter_out is applied to every entry from Loc-RIB and if evaluation passes, will be sent to Adj-RIBs-Out
+    pub filter_out: Vec<(FilterMatcher, FilterAction)>,
+}
+
+/// All the fields in a FilterMatcher must be matched if present for the action to be executed.
+// Implementation note, it's probably more efficient to JIT compile the filter into machine code
+// so only the relevant fields need to be checked.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct FilterMatcher {
+    pub nlri: Option<NLRI>,
+    pub origin_asn: Option<u32>,
+    pub large_community: Option<LargeCommunitiesPayload>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum FilterAction {
+    Accept,
+    Reject,
+    Update(UpdateAction),
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum UpdateAction {
+    AttachLargeCommunity(LargeCommunitiesPayload),
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]

crates/server/src/filter_eval.rs

@@ -0,0 +1,205 @@
+use bgp_packet::{
+    nlri::NLRI,
+    path_attributes::{LargeCommunitiesPathAttribute, PathAttribute},
+};
+
+use crate::config::{FilterAction, FilterMatcher, UpdateAction};
+
+pub struct FilterEvaluator {
+    filter_in: Vec<(FilterMatcher, FilterAction)>,
+    filter_out: Vec<(FilterMatcher, FilterAction)>,
+}
+
+impl FilterEvaluator {
+    pub fn new(
+        filter_in: Vec<(FilterMatcher, FilterAction)>,
+        filter_out: Vec<(FilterMatcher, FilterAction)>,
+    ) -> Self {
+        Self {
+            filter_in,
+            filter_out,
+        }
+    }
+
+    fn check_rule_match(
+        matcher: &FilterMatcher,
+        path_attributes: &Vec<PathAttribute>,
+        as_path: &Vec<u32>,
+        nlri: &NLRI,
+    ) -> bool {
+        if let Some(matcher_nlri) = &matcher.nlri {
+            if nlri != matcher_nlri {
+                return false;
+            }
+        }
+        if let (Some(matcher_origin_asn), Some(origin_asn)) = (matcher.origin_asn, as_path.last()) {
+            if matcher_origin_asn != *origin_asn {
+                return false;
+            }
+        }
+        if let Some(matcher_large_community) = &matcher.large_community {
+            let mut found = false;
+            for attribute in path_attributes {
+                if let PathAttribute::LargeCommunitiesPathAttribute(lcs) = attribute {
+                    if lcs.values.iter().any(|lc| lc == matcher_large_community) {
+                        found = true;
+                        break;
+                    }
+                }
+            }
+            if !found {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    fn apply_update(update_action: &UpdateAction, path_attributes: &mut Vec<PathAttribute>) {
+        match update_action {
+            UpdateAction::AttachLargeCommunity(large_community) => {
+                let mut added_existing = false;
+                for path_attribute in &mut *path_attributes {
+                    if let PathAttribute::LargeCommunitiesPathAttribute(lc_attr) = path_attribute {
+                        lc_attr.values.push(large_community.clone());
+                        added_existing = true;
+                    }
+                }
+                if !added_existing {
+                    path_attributes.push(PathAttribute::LargeCommunitiesPathAttribute(
+                        LargeCommunitiesPathAttribute {
+                            values: vec![large_community.clone()],
+                        },
+                    ))
+                }
+            }
+        }
+    }
+
+    fn evaluate(
+        rules: &Vec<(FilterMatcher, FilterAction)>,
+        path_attributes: &mut Vec<PathAttribute>,
+        as_path: &Vec<u32>,
+        nlri: &NLRI,
+    ) -> bool {
+        for rule in rules {
+            if Self::check_rule_match(&rule.0, path_attributes, as_path, nlri) {
+                match &rule.1 {
+                    FilterAction::Accept => return true,
+                    FilterAction::Reject => return false,
+                    FilterAction::Update(update_action) => {
+                        Self::apply_update(update_action, path_attributes)
+                    }
+                }
+            }
+        }
+
+        // Default behavior is to deny.
+        return false;
+    }
+
+    /// evaluate_in checks if an announced route is eligible to be accepted into the Loc-RIB.
+    /// Note that this may change the path_attributes if a FilterAction requests to do so.
+    pub fn evaluate_in(
+        &self,
+        path_attributes: &mut Vec<PathAttribute>,
+        as_path: &Vec<u32>,
+        nlri: &NLRI,
+    ) -> bool {
+        Self::evaluate(&self.filter_in, path_attributes, as_path, nlri)
+    }
+
+    /// evaluate_out checks if a route from the Loc-RIB is to be announced to a peer.
+    /// Note that this may change the path_attributes if a FilterAction requests to do so.
+    pub fn evaluate_out(
+        &self,
+        path_attributes: &mut Vec<PathAttribute>,
+        as_path: &Vec<u32>,
+        nlri: &NLRI,
+    ) -> bool {
+        Self::evaluate(&self.filter_out, path_attributes, as_path, nlri)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use bgp_packet::nlri::NLRI;
+
+    use crate::config::{FilterAction, FilterMatcher};
+
+    use super::FilterEvaluator;
+
+    #[test]
+    fn test_simple_match_nlri() {
+        let nlri = NLRI::try_from("2001:db8::/48").unwrap();
+        let matcher = FilterEvaluator::new(
+            vec![(
+                FilterMatcher {
+                    nlri: Some(nlri.clone()),
+                    origin_asn: None,
+                    large_community: None,
+                },
+                FilterAction::Accept,
+            )],
+            vec![],
+        );
+
+        assert!(matcher.evaluate_in(&mut vec![], &vec![], &nlri));
+    }
+
+    #[test]
+    fn test_simple_match_origin_asn() {
+        let matcher = FilterEvaluator::new(
+            vec![(
+                FilterMatcher {
+                    nlri: None,
+                    origin_asn: Some(65000),
+                    large_community: None,
+                },
+                FilterAction::Accept,
+            )],
+            vec![],
+        );
+
+        assert!(matcher.evaluate_in(
+            &mut vec![],
+            &vec![65000],
+            &NLRI::try_from("2001:db8::/48").unwrap()
+        ));
+    }
+
+    #[test]
+    fn test_targeted_deny() {
+        let bad_nlri = NLRI::try_from("2001:db8:bad::/48").unwrap();
+        let matcher = FilterEvaluator::new(
+            vec![
+                // Reject a specific prefix 2001:db8:bad::/48
+                (
+                    FilterMatcher {
+                        nlri: Some(bad_nlri.clone()),
+                        origin_asn: None,
+                        large_community: None,
+                    },
+                    FilterAction::Reject,
+                ),
+                // Accept everything else.
+                (
+                    FilterMatcher {
+                        nlri: None,
+                        origin_asn: None,
+                        large_community: None,
+                    },
+                    FilterAction::Accept,
+                ),
+            ],
+            vec![],
+        );
+
+        assert!(!matcher.evaluate_in(&mut vec![], &vec![], &bad_nlri));
+        assert!(matcher.evaluate_in(
+            &mut vec![],
+            &vec![],
+            &NLRI::try_from("2001:db8:1234::/48").unwrap()
+        ));
+    }
+}

crates/server/src/lib.rs

@@ -15,6 +15,7 @@
 pub mod bgp_server;
 pub mod config;
 pub mod data_structures;
+pub mod filter_eval;
 pub mod peer;
 pub mod rib_manager;
 pub mod route_server;

crates/server/src/peer.rs

@@ -17,6 +17,7 @@ use crate::config::{PeerConfig, ServerConfig};
 use crate::data_structures::RouteAnnounce;
 use crate::data_structures::RouteWithdraw;
 use crate::data_structures::{RouteInfo, RouteUpdate};
+use crate::filter_eval::FilterEvaluator;
 use crate::rib_manager::RouteManagerCommands;
 use crate::route_server::route_server::PeerStatus;
 use bgp_packet::capabilities::{
@@ -302,6 +303,10 @@ pub struct PeerStateMachine<A: Address> {
     // restarted so that the new configuration can take effect.
     config: PeerConfig,
 
+    /// FilterEvaluator checks whether a given NLRI should be accepted or not
+    /// based on the installed filters.
+    filter_evaluator: FilterEvaluator,
+
     // Store the peer's open message so we can reference it.
     peer_open_msg: Option<OpenMessage>,
 
@@ -364,7 +369,8 @@ where
         let afi = config.afi;
         PeerStateMachine {
             server_config,
-            config,
+            config: config.clone(),
+            filter_evaluator: FilterEvaluator::new(config.filter_in, config.filter_out),
             peer_open_msg: None,
             state: BGPState::Active,
             tcp_stream: None,
@@ -766,6 +772,7 @@ where
         announcements: Vec<NLRI>,
         path_attributes: Vec<PathAttribute>,
     ) -> Result<(), String> {
+        // Extract the as_path and med from the attributes.
         let mut as_path: Vec<u32> = vec![];
         let mut med: u32 = 0;
         for attr in &path_attributes {
@@ -797,7 +804,11 @@ where
         for announcement in announcements {
             let addr: A = announcement.clone().try_into().map_err(|e| e.to_string())?;
             // Should we accept this prefix?
-            let accepted: bool = self.decide_accept_prefix(addr, announcement.prefixlen);
+            let accepted = self.filter_evaluator.evaluate_in(
+                &mut route_update.path_attributes,
+                &route_update.as_path,
+                &announcement,
+            );
             let rejection_reason: Option<String> = match accepted {
                 true => Some("Filtered by policy".to_owned()),
                 false => None,
@@ -851,11 +862,6 @@ where
         Ok(())
     }
 
-    fn decide_accept_prefix(&mut self, _: A, _: u8) -> bool {
-        // TODO: Implement filtering of prefixes.
-        true
-    }
-
     fn decide_accept_message(&mut self, _: &[PathAttribute]) -> bool {
         // TODO: Implement filtering of Update messages.
 
@@ -1185,7 +1191,7 @@ where
                     _ => return Err("Found non IPv4 nexthop in announcement".to_string()),
                 }
 
-                let nlri = NLRI::try_from(announcement.prefix.clone())?;
+                let nlri = NLRI::try_from(announcement.prefix.as_str())?;
                 bgp_update_msg.announced_nlri.push(nlri);
             }
             AddressFamilyIdentifier::Ipv6 => {
@@ -1195,7 +1201,7 @@ where
                         return Err("Found non IPv6 nexthop in announcement".to_string());
                     }
                 };
-                let nlri = NLRI::try_from(announcement.prefix.clone())?;
+                let nlri = NLRI::try_from(announcement.prefix.as_str())?;
                 let mp_reach = MPReachNLRIPathAttribute {
                     afi: AddressFamilyIdentifier::Ipv6,
                     safi: SubsequentAddressFamilyIdentifier::Unicast,
@@ -1272,7 +1278,7 @@ where
                         PathAttribute::MPReachNLRIPathAttribute(nlri) => {
                             // TODO: Determine which AFI/SAFI this update corresponds to.
                             let nexthop_res = nlri.clone().nexthop_to_v6();
-                            // TODO: How do we pick whether to use the global or LLNH?
+
                             if let Some((global, _llnh_opt)) = nexthop_res {
                                 self.process_announcements(
                                     global.octets().to_vec(),
@@ -1326,6 +1332,7 @@ where
                 );
                 Ok(())
             }
+
             BGPSubmessage::KeepaliveMessage(_) => Ok(()),
             _ => Err(format!("Got unexpected message from peer: {:?}", msg)),
         }

tests/integration_tests/tests/basic_startup.rs

@@ -125,6 +125,8 @@ async fn test_bgp_listener_known_peer() {
             name: "local-test-peer".to_string(),
             local_pref: 100,
             port: None,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
         }],
     };
 
@@ -215,6 +217,8 @@ async fn test_bgp_peer_statemachine_outbound_conn() {
             announcements: vec![],
             name: "local-test-peer".to_string(),
             local_pref: 100,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
         }],
     };
 
@@ -303,6 +307,8 @@ async fn test_bgp_peer_statemachine_outbound_reconnection() {
             announcements: vec![],
             name: "local-test-peer".to_string(),
             local_pref: 100,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
         }],
     };
 
@@ -437,6 +443,8 @@ async fn test_bgp_listener_known_peer_inbound_reconnection() {
             name: "local-test-peer".to_string(),
             local_pref: 100,
             port: None,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
         }],
     };
 
@@ -588,6 +596,8 @@ async fn test_multi_instance_announce() {
             }],
             name: "config-b-peer".to_string(),
             local_pref: 100,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
         }],
     };