Implemented basic input filtering.

Cargo.toml

@@ -1,9 +1,70 @@
 [workspace]
-
 members = [
-    "bgpd",
-    "netlink",
+    "bin",
+    "crates/bgp_packet",
+    "crates/route_client",
+    "crates/server",
 
     # Tests
-    "tests/integration_tests"
+    "tests/integration_tests",
 ]
+resolver = "2"
+
+[workspace.package]
+authors      = ["Rayhaan Jaufeerally <rayhaan@rayhaan.ch>"]
+edition      = "2021"
+homepage     = "https://github.com/net-control-plane/bgp"
+license      = "Apache 2"
+repository   = "https://github.com/net-control-plane/bgp"
+rust-version = "1.76"
+version      = "0.1.1"
+
+# [[bin]]
+# name = "bgp_server"
+# path = "src/main.rs"
+
+# [[bin]]
+# name = "route_client"
+# path = "src/route_client/main.rs"
+
+# [[bin]]
+# name = "streamer_cli"
+# path = "src/streamer_cli/main.rs"
+
+[workspace.lints]
+rust.unused_must_use = "deny"
+
+[workspace.dependencies]
+anyhow                           = "1.0.71"
+async-trait                      = "0.1.80"
+byteorder                        = "1.4.3"
+bytes                            = "1.*"
+clap                             = { version = "3.2.8", features = ["cargo", "derive"] }
+eyre                             = "0.6.12"
+futures                          = "0.3"
+ip_network_table-deps-treebitmap = "0.5.0"
+ipnet                            = "2.3.0"
+libc                             = "0.2.126"
+log                              = "0.4"
+netlink                          = "0.1.1"
+netlink-packet-route             = "0.19.0"
+netlink-packet-utils             = "0.5.2"
+nom                              = "7.1"
+prost                            = "0.8"
+rtnetlink                        = "0.14.1"
+serde                            = { version = "1.0", features = ["derive"] }
+serde_json                       = "1.0.64"
+stderrlog                        = "0.5.1"
+tokio                            = { version = "1.13.0", features = ["full"] }
+tokio-stream                     = { version = "0.1.7", features = ["net"] }
+tokio-util                       = { version = "0.6.7", features = ["codec"] }
+tonic                            = { version = "0.5", features = ["compression"] }
+tracing                          = "0.1"
+tracing-subscriber               = { version = "0.3.17", features = ["env-filter"] }
+warp                             = "0.3.5"
+
+# -- Local crates --
+bgp_packet   = { path = "crates/bgp_packet" }
+bgp_server   = { path = "crates/server" }
+bin          = { path = "bin" }
+route_client = { path = "crates/route_client" }

bgpd/Cargo.toml

@@ -1,50 +0,0 @@
-[package]
-name = "bgpd"
-version = "0.1.0"
-authors = ["Rayhaan Jaufeerally <rayhaan@rayhaan.ch>"]
-edition = "2021"
-
-[[bin]]
-name = "bgp_server"
-path = "src/main.rs"
-
-[[bin]]
-name = "route_client"
-path = "src/route_client/main.rs"
-
-[[bin]]
-name = "streamer_cli"
-path = "src/streamer_cli/main.rs"
-
-[dependencies]
-anyhow = "1.0.56"
-byteorder = "1.4.3"
-bytes = "1.*"
-clap = {version = "3.2.8", features = ["cargo", "derive"]}
-futures = "0.3"
-ipnet = "2.3.0"
-libc = "0.2.126"
-log = "0.4"
-nom = "6.1.2"
-prost = "0.8"
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0.64"
-signal-hook = { version = "0.3.10", features = ["extended-siginfo"] }
-signal-hook-tokio = "0.3.0"
-stderrlog = "0.5.1"
-tokio = { version = "1.13.0", features = ["full"] }
-tokio-stream = { version = "0.1.7", features = ["net"] }
-tokio-util = { version = "0.6.7", features = ["codec"] }
-tonic = { version = "0.5", features = ["compression"] }
-tracing = "0.1"
-tracing-subscriber = "0.2"
-ip_network_table-deps-treebitmap = "0.5.0"
-warp = "0.3"
-rtnetlink = "0.9.1"
-netlink-packet-route = "0.11.0"
-netlink = { path = "../netlink" }
-neli = "0.6.2"
-async-trait = "0.1.57"
-
-[build-dependencies]
-tonic-build = { version = "0.5.1", features = ["prost", "compression"] }

bgpd/rustfmt.toml

@@ -1 +0,0 @@
-edition = "2018"

bgpd/src/lib.rs

@@ -1,3 +0,0 @@
-pub mod bgp_packet;
-pub mod route_client;
-pub mod server;

bin/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+edition.workspace      = true
+homepage.workspace     = true
+license.workspace      = true
+name                   = "bgpd_bin"
+repository.workspace   = true
+rust-version.workspace = true
+version.workspace      = true
+
+[dependencies]
+bgp_server.workspace         = true
+clap.workspace               = true
+eyre.workspace               = true
+libc.workspace               = true
+log.workspace                = true
+route_client.workspace       = true
+serde_json.workspace         = true
+signal-hook                  = { version = "0.3.17", features = ["extended-siginfo"] }
+signal-hook-tokio            = "0.3.0"
+tokio.workspace              = true
+tracing-subscriber.workspace = true
+tracing.workspace            = true
+
+[[bin]]
+name = "bgp_server"
+path = "src/bgp_server/main.rs"
+
+[[bin]]
+name = "client"
+path = "src/client/main.rs"

bin/src/bgp_server/main.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use bgpd::server::config::ServerConfig;
-use bgpd::server::bgp_server::Server;
+use bgp_server::bgp_server::Server;
+use bgp_server::config::ServerConfig;
 use clap::{App, Arg};
 use core::sync::atomic::AtomicBool;
 use libc::SIGUSR1;

bin/src/client/main.rs

@@ -0,0 +1,73 @@
+use clap::Parser;
+use eyre::Result;
+use tracing::{info, warn};
+
+use route_client::netlink::NetlinkConnector;
+use route_client::{run_connector_v4, run_connector_v6};
+
+#[derive(Parser)]
+#[clap(
+    author = "Rayhaan Jaufeerally <rayhaan@rayhaan.ch>",
+    version = "0.1",
+    about = "Installs routes from a BGP speaker via streaming RPC to the forwarding plane"
+)]
+struct Cli {
+    #[clap(long = "route_server")]
+    route_server: String,
+    #[clap(long = "rt_table")]
+    rt_table: Option<u32>,
+    dry_run: bool,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let args = Cli::parse();
+
+    tracing_subscriber::fmt().pretty().init();
+
+    info!("Starting route client");
+
+    let rt_table = match args.rt_table {
+        Some(table) => table,
+        None => 201,
+    };
+
+    let v4_joinhandle = {
+        let server_addr = args.route_server.clone();
+        tokio::task::spawn(async move {
+            run_connector_v4::<NetlinkConnector>(
+                server_addr.clone(),
+                rt_table,
+                args.dry_run,
+                NetlinkConnector::new(Some(rt_table)).await.unwrap(),
+            )
+            .await
+            .unwrap();
+        })
+    };
+
+    let v6_joinhandle = {
+        let server_addr = args.route_server.clone();
+        tokio::task::spawn(async move {
+            run_connector_v6::<NetlinkConnector>(
+                server_addr,
+                rt_table,
+                args.dry_run,
+                NetlinkConnector::new(Some(rt_table)).await.unwrap(),
+            )
+            .await
+            .unwrap();
+        })
+    };
+
+    tokio::select! {
+        _ = v4_joinhandle => {
+            warn!("Unexpected exit of IPv4 connector");
+        },
+        _ = v6_joinhandle => {
+            warn!("Unexpected exit of IPv6 connector");
+        }
+    }
+
+    Ok(())
+}

bin/src/streamer_cli/main.rs

@@ -31,7 +31,7 @@ extern crate clap;
 #[clap(
     author = "Rayhaan Jaufeerally <rayhaan@rayhaan.ch>",
     version = "0.1",
-    about = "A program to install routes from BGP into the Linux control plane"
+    about = "A utility for dumping routes from the bgp_server."
 )]
 struct Cli {
     server_address: String,

crates/bgp_packet/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "bgp_packet"
+
+edition.workspace      = true
+homepage.workspace     = true
+license.workspace      = true
+repository.workspace   = true
+rust-version.workspace = true
+version.workspace      = true
+
+[lints]
+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/capabilities.rs

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::bgp_packet::constants::SubsequentAddressFamilyIdentifier;
-use crate::bgp_packet::traits::BGPParserError;
-use crate::bgp_packet::traits::ParserContext;
-use crate::bgp_packet::traits::ReadablePacket;
-use crate::bgp_packet::traits::WritablePacket;
+use crate::constants::AddressFamilyIdentifier;
+use crate::constants::SubsequentAddressFamilyIdentifier;
+use crate::traits::BGPParserError;
+use crate::traits::ParserContext;
+use crate::traits::ReadablePacket;
+use crate::traits::WritablePacket;
 use byteorder::{ByteOrder, NetworkEndian};
 use nom::number::complete::{be_u16, be_u8};
 use nom::Err::Failure;
@@ -48,9 +48,6 @@ pub mod BGPOpenOptionTypeValues {
     pub const CAPABILITIES: BGPOpenOptionType = BGPOpenOptionType(2);
 }
 
-/// OpenOptionValue represents something which can be in the payload of OpenOption.
-trait OpenOptionValue: ReadablePacket + WritablePacket + fmt::Debug {}
-
 #[derive(Debug, PartialEq)]
 pub struct OpenOption {
     pub option_type: BGPOpenOptionType,
@@ -694,9 +691,9 @@ mod tests {
     use super::ExtendedNextHopEncodingCapability;
     use super::FourByteASNCapability;
     use super::OpenOption;
-    use crate::bgp_packet::constants::AddressFamilyIdentifier::Ipv6;
-    use crate::bgp_packet::traits::ParserContext;
-    use crate::bgp_packet::traits::ReadablePacket;
+    use crate::constants::AddressFamilyIdentifier::Ipv6;
+    use crate::traits::ParserContext;
+    use crate::traits::ReadablePacket;
 
     #[test]
     fn test_four_byte_asn_capability() {

crates/bgp_packet/src/messages.rs

@@ -12,15 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::capabilities::OpenOption;
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::bgp_packet::constants::SubsequentAddressFamilyIdentifier;
-use crate::bgp_packet::nlri::NLRI;
-use crate::bgp_packet::path_attributes::PathAttribute;
-use crate::bgp_packet::traits::BGPParserError;
-use crate::bgp_packet::traits::ParserContext;
-use crate::bgp_packet::traits::ReadablePacket;
-use crate::bgp_packet::traits::WritablePacket;
+use crate::capabilities::OpenOption;
+use crate::constants::AddressFamilyIdentifier;
+use crate::constants::SubsequentAddressFamilyIdentifier;
+use crate::nlri::NLRI;
+use crate::path_attributes::PathAttribute;
+use crate::traits::BGPParserError;
+use crate::traits::ParserContext;
+use crate::traits::ReadablePacket;
+use crate::traits::WritablePacket;
+
 use byteorder::{ByteOrder, NetworkEndian};
 use bytes::Buf;
 use bytes::BufMut;
@@ -545,11 +546,11 @@ impl Display for UpdateMessage {
 mod tests {
     use super::BGPMessage;
     use super::Codec;
-    use crate::bgp_packet::constants::AddressFamilyIdentifier::Ipv6;
-    use crate::bgp_packet::messages::AddressFamilyIdentifier::Ipv4;
-    use crate::bgp_packet::traits::ParserContext;
-    use crate::bgp_packet::traits::ReadablePacket;
-    use crate::bgp_packet::traits::WritablePacket;
+    use crate::constants::AddressFamilyIdentifier::Ipv6;
+    use crate::messages::AddressFamilyIdentifier::Ipv4;
+    use crate::traits::ParserContext;
+    use crate::traits::ReadablePacket;
+    use crate::traits::WritablePacket;
 
     use bytes::BufMut;
     use tokio_util::codec::{Decoder, Encoder};

crates/bgp_packet/src/nlri.rs

@@ -12,16 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::bgp_packet::traits::BGPParserError;
-use crate::bgp_packet::traits::ParserContext;
-use crate::bgp_packet::traits::ReadablePacket;
-use crate::bgp_packet::traits::WritablePacket;
+use crate::constants::AddressFamilyIdentifier;
+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 {
@@ -254,10 +275,10 @@ mod tests {
     use std::convert::TryFrom;
 
     use super::NLRI;
-    use crate::bgp_packet::constants::AddressFamilyIdentifier::{Ipv4, Ipv6};
-    use crate::bgp_packet::traits::ParserContext;
-    use crate::bgp_packet::traits::ReadablePacket;
-    use crate::bgp_packet::traits::WritablePacket;
+    use crate::constants::AddressFamilyIdentifier::{Ipv4, Ipv6};
+    use crate::traits::ParserContext;
+    use crate::traits::ReadablePacket;
+    use crate::traits::WritablePacket;
 
     #[test]
     fn test_basic_nlri_v6() {
@@ -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

@@ -12,18 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::bgp_packet::constants::SubsequentAddressFamilyIdentifier;
-use crate::bgp_packet::nlri::NLRI;
-use crate::bgp_packet::traits::BGPParserError;
-use crate::bgp_packet::traits::ParserContext;
-use crate::bgp_packet::traits::ReadablePacket;
-use crate::bgp_packet::traits::WritablePacket;
+use crate::constants::AddressFamilyIdentifier;
+use crate::constants::SubsequentAddressFamilyIdentifier;
+use crate::nlri::NLRI;
+use crate::traits::BGPParserError;
+use crate::traits::ParserContext;
+use crate::traits::ReadablePacket;
+use crate::traits::WritablePacket;
 use byteorder::ByteOrder;
 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,
@@ -976,11 +977,11 @@ impl fmt::Display for MPUnreachNLRIPathAttribute {
 mod tests {
     use std::net::Ipv4Addr;
 
-    use crate::bgp_packet::constants::AddressFamilyIdentifier::Ipv6;
-    use crate::bgp_packet::constants::SubsequentAddressFamilyIdentifier::Unicast;
-    use crate::bgp_packet::traits::ParserContext;
-    use crate::bgp_packet::traits::ReadablePacket;
-    use crate::bgp_packet::traits::WritablePacket;
+    use crate::constants::AddressFamilyIdentifier::Ipv6;
+    use crate::constants::SubsequentAddressFamilyIdentifier::Unicast;
+    use crate::traits::ParserContext;
+    use crate::traits::ReadablePacket;
+    use crate::traits::WritablePacket;
 
     use super::ASPathAttribute;
     use super::CommunitiesPathAttribute;

crates/bgp_packet/src/traits.rs

@@ -14,7 +14,7 @@
 
 //! Implements high level abstractions for use in the BGP parser.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
+use crate::constants::AddressFamilyIdentifier;
 use nom::error::ErrorKind;
 use nom::error::ParseError;
 use nom::IResult;

crates/route_client/Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "route_client"
+
+edition.workspace      = true
+homepage.workspace     = true
+license.workspace      = true
+repository.workspace   = true
+rust-version.workspace = true
+version.workspace      = true
+
+[lints]
+workspace = true
+
+[dependencies]
+async-trait.workspace                      = true
+bgp_packet.workspace                       = true
+byteorder                                  = "1.4.3"
+bytes.workspace                            = true
+eyre.workspace                             = true
+futures.workspace                          = true
+ip_network_table-deps-treebitmap.workspace = true
+log.workspace                              = true
+netlink-packet-route.workspace             = true
+netlink-packet-utils.workspace             = true
+nom                                        = "7.1"
+prost.workspace                            = true
+rtnetlink.workspace                        = true
+serde.workspace                            = true
+tokio-stream                               = "0.1.14"
+tokio-util                                 = { version = "0.7.10", features = ["codec"] }
+tokio.workspace                            = true
+tonic.workspace                            = true
+tracing.workspace                          = true
+warp.workspace                             = true
+
+[build-dependencies]
+tonic-build = { version = "0.5.1", features = ["compression", "prost"] }

crates/route_client/proto/route_service.proto

@@ -0,0 +1,84 @@
+// Copyright 2021 Rayhaan Jaufeerally.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package bgpd.grpc;
+
+enum AddressFamily {
+  UNKNOWN = 0;
+  IPv4 = 1;
+  IPv6 = 2;
+}
+
+message Prefix {
+  bytes ip_prefix = 1;
+  int32 prefix_len = 2;
+  AddressFamily address_family = 3;
+}
+
+// Path represents the metadata associated with the route to a particular
+// prefix.
+message Path {
+  bytes nexthop = 1;
+  string peer_name = 2;
+  uint32 local_pref = 3;
+  uint32 med = 4;
+  repeated uint32 as_path = 5;
+  // TODO: Path attributes. Not yet supported because we need to generate proto
+  // definitions for all of them.
+}
+
+message PathSet {
+  uint64 epoch = 1;
+  Prefix prefix = 2;
+  repeated Path paths = 3;
+}
+
+message StreamPathsRequest { AddressFamily address_family = 1; }
+
+message DumpPathsRequest { AddressFamily address_family = 1; };
+
+message DumpPathsResponse {
+  uint64 epoch = 1;
+  repeated PathSet path_sets = 2;
+};
+
+service RouteService {
+  // DumpPaths returns all the paths currently in the RIB.
+  rpc DumpPaths(DumpPathsRequest) returns (DumpPathsResponse);
+  // StreamPaths dumps the existing routes and starts streaming updates to the
+  // RIB.
+  rpc StreamPaths(StreamPathsRequest) returns (stream PathSet);
+}
+
+message PeerStatusRequest {}
+
+message PeerStatus {
+  string peer_name = 1;
+  string state = 2;
+  // The following fields are only populated when the session is established.
+  optional uint64 session_established_time = 3;
+  optional uint64 last_messaage_time = 4;
+  optional uint64 route_updates_in = 5;
+  optional uint64 route_updates_out = 6;
+}
+
+message PeerStatusResponse { repeated PeerStatus peer_status = 1; }
+
+// BGPServerAdminService implements an administrative interface to
+// view the status and control the operation of this BGP server.
+service BGPServerAdminService {
+  rpc PeerStatus(PeerStatusRequest) returns (PeerStatusResponse);
+}

crates/route_client/src/fib_state.rs

@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::bgp_packet::nlri::NLRI;
-use crate::route_client::southbound_interface::SouthboundInterface;
 use futures::lock::Mutex;
 use ip_network_table_deps_treebitmap::address::Address;
 use ip_network_table_deps_treebitmap::IpLookupTable;
@@ -25,6 +22,11 @@ use std::net::{IpAddr, Ipv4Addr};
 use std::sync::Arc;
 use tracing::{trace, warn};
 
+use bgp_packet::constants::AddressFamilyIdentifier;
+use bgp_packet::nlri::NLRI;
+
+use crate::southbound_interface::SouthboundInterface;
+
 /// fib_state implements the logic to maintain forwarding routes in the FIB.
 /// This for now means the Linux Kernel via Netlink, but in the future can
 /// be extended to include other targets such as OpenFlow or even program

crates/route_client/src/lib.rs

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use bgpd::route_client::netlink::NetlinkConnector;
-use bgpd::route_client::southbound_interface::SouthboundInterface;
-use clap::Parser;
+pub mod fib_state;
+pub mod netlink;
+pub mod southbound_interface;
+
 use log::trace;
 use std::convert::TryInto;
 use std::net::IpAddr;
@@ -23,27 +24,27 @@ use std::net::Ipv4Addr;
 use std::net::Ipv6Addr;
 use std::str::FromStr;
 use std::time::Duration;
-use tonic::transport::Uri;
 
-use bgpd::bgp_packet::constants::AddressFamilyIdentifier;
-use bgpd::bgp_packet::nlri::NLRI;
-use bgpd::route_client::fib_state::FibState;
+use bgp_packet::constants::AddressFamilyIdentifier;
+use bgp_packet::nlri::NLRI;
 
+use eyre::{anyhow, Result};
 use ip_network_table_deps_treebitmap::IpLookupTable;
 use tonic::transport::Endpoint;
-use tracing::{info, warn};
-
-use anyhow::{anyhow, Result};
+use tonic::transport::Uri;
+use tracing::info;
 
+use crate::fib_state::FibState;
 use crate::proto::route_service_client::RouteServiceClient;
+use crate::southbound_interface::SouthboundInterface;
 
 pub mod proto {
     tonic::include_proto!("bgpd.grpc");
 }
 
-fn vec_to_array<T, const N: usize>(v: Vec<T>) -> Result<[T; N], anyhow::Error> {
+fn vec_to_array<T, const N: usize>(v: Vec<T>) -> Result<[T; N]> {
     v.try_into()
-        .map_err(|_| anyhow::Error::msg("Wrong size of Vec".to_string()))
+        .map_err(|_| eyre::Error::msg("Wrong size of Vec".to_string()))
 }
 
 /// Temporary hack to select the route to install to the FIB.
@@ -62,12 +63,12 @@ fn select_best_route(ps: &proto::PathSet) -> Option<proto::Path> {
     selected
 }
 
-async fn run_connector_v4<S: SouthboundInterface>(
+pub async fn run_connector_v4<S: SouthboundInterface>(
     route_server: String,
     rt_table: u32,
     dry_run: bool,
     southbound: S,
-) -> Result<(), anyhow::Error> {
+) -> Result<()> {
     // Create netlink socket.
     let mut fib_state = FibState::<Ipv4Addr, S> {
         fib: IpLookupTable::new(),
@@ -132,7 +133,7 @@ async fn run_connector_v4<S: SouthboundInterface>(
     unreachable!()
 }
 
-async fn run_connector_v6<S: SouthboundInterface>(
+pub async fn run_connector_v6<S: SouthboundInterface>(
     route_server: String,
     rt_table: u32,
     dry_run: bool,
@@ -201,72 +202,3 @@ async fn run_connector_v6<S: SouthboundInterface>(
 
     unreachable!()
 }
-
-#[derive(Parser)]
-#[clap(
-    author = "Rayhaan Jaufeerally <rayhaan@rayhaan.ch>",
-    version = "0.1",
-    about = "Installs routes from a BGP speaker via streaming RPC to the forwarding plane"
-)]
-struct Cli {
-    #[clap(long = "route_server")]
-    route_server: String,
-    #[clap(long = "rt_table")]
-    rt_table: Option<u32>,
-    dry_run: bool,
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    let args = Cli::parse();
-
-    let _init_log = stderrlog::new()
-        .verbosity(2) // Shows info level.
-        .show_module_names(true)
-        .init();
-    info!("Starting route client");
-
-    let rt_table = match args.rt_table {
-        Some(table) => table,
-        None => 201,
-    };
-
-    let v4_joinhandle = {
-        let server_addr = args.route_server.clone();
-        tokio::task::spawn(async move {
-            run_connector_v4::<NetlinkConnector>(
-                server_addr.clone(),
-                rt_table,
-                args.dry_run,
-                NetlinkConnector::new(Some(rt_table)).await.unwrap(),
-            )
-            .await
-            .unwrap();
-        })
-    };
-
-    let v6_joinhandle = {
-        let server_addr = args.route_server.clone();
-        tokio::task::spawn(async move {
-            run_connector_v6::<NetlinkConnector>(
-                server_addr,
-                rt_table,
-                args.dry_run,
-                NetlinkConnector::new(Some(rt_table)).await.unwrap(),
-            )
-            .await
-            .unwrap();
-        })
-    };
-
-    tokio::select! {
-        _ = v4_joinhandle => {
-            warn!("Unexpected exit of IPv4 connector");
-        },
-        _ = v6_joinhandle => {
-            warn!("Unexpected exit of IPv6 connector");
-        }
-    }
-
-    Ok(())
-}

crates/route_client/src/netlink.rs

@@ -1,10 +1,14 @@
-use crate::bgp_packet::{constants::AddressFamilyIdentifier, nlri::NLRI};
-use anyhow::Result;
 use async_trait::async_trait;
+use bgp_packet::{constants::AddressFamilyIdentifier, nlri::NLRI};
+use eyre::{eyre, Result};
 use futures::TryStreamExt;
-use netlink::constants::RTN_UNICAST;
-use netlink_packet_route::{rtnl::route::nlas::Nla, RouteHeader};
-use netlink_packet_route::{RouteMessage, RTPROT_STATIC};
+use netlink_packet_route::route::RouteAddress;
+use netlink_packet_route::route::RouteAttribute;
+use netlink_packet_route::route::RouteHeader;
+use netlink_packet_route::route::RouteMessage;
+use netlink_packet_route::route::RouteProtocol;
+use netlink_packet_route::route::RouteType;
+use netlink_packet_route::AddressFamily as NetlinkAddressFamily;
 use rtnetlink::IpVersion;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::{convert::TryInto, io::ErrorKind};
@@ -29,10 +33,11 @@ impl SouthboundInterface for NetlinkConnector {
         let route = self.handle.route();
         match address_family {
             AddressFamilyIdentifier::Ipv6 => {
-                let addr: Ipv6Addr = match prefix.clone().try_into()? {
+                let prefix_len = prefix.prefixlen;
+                let addr: Ipv6Addr = match prefix.try_into()? {
                     IpAddr::V6(addr) => addr,
                     _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
+                        return Err(eyre::Error::from(std::io::Error::new(
                             ErrorKind::InvalidInput,
                             "Got non-IPv6 address from NLRI",
                         )))
@@ -41,7 +46,7 @@ impl SouthboundInterface for NetlinkConnector {
                 let gw_addr: Ipv6Addr = match nexthop.clone().try_into()? {
                     IpAddr::V6(addr) => addr,
                     _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
+                        return Err(eyre::Error::from(std::io::Error::new(
                             ErrorKind::InvalidInput,
                             "Got non-IPv6 gateway for IPv6 NLRI",
                         )))
@@ -50,18 +55,19 @@ impl SouthboundInterface for NetlinkConnector {
                 let mut mutation = route
                     .add()
                     .v6()
-                    .destination_prefix(addr, prefix.prefixlen)
+                    .destination_prefix(addr, prefix_len)
                     .gateway(gw_addr);
                 if let Some(table_id) = self.table {
-                    mutation = mutation.table(table_id.try_into().unwrap());
+                    mutation = mutation.table_id(table_id.try_into().unwrap());
                 }
-                mutation.execute().await.map_err(|e| anyhow::Error::from(e))
+                mutation.execute().await.map_err(|e| eyre::Error::from(e))
             }
             AddressFamilyIdentifier::Ipv4 => {
+                let prefix_len = prefix.prefixlen;
                 let addr: Ipv4Addr = match prefix.clone().try_into()? {
                     IpAddr::V4(addr) => addr,
                     _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
+                        return Err(eyre::Error::from(std::io::Error::new(
                             ErrorKind::InvalidInput,
                             "Got non-IPv4 address from NLRI",
                         )))
@@ -70,7 +76,7 @@ impl SouthboundInterface for NetlinkConnector {
                 let gw_addr = match nexthop.clone().try_into()? {
                     IpAddr::V4(addr) => addr,
                     _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
+                        return Err(eyre::Error::from(std::io::Error::new(
                             ErrorKind::InvalidInput,
                             "Got non-IPv4 gateway for IPv4 NLRI",
                         )))
@@ -79,71 +85,52 @@ impl SouthboundInterface for NetlinkConnector {
                 let mut mutation = route
                     .add()
                     .v4()
-                    .destination_prefix(addr, prefix.prefixlen)
+                    .destination_prefix(addr, prefix_len)
                     .gateway(gw_addr);
                 if let Some(table_id) = self.table {
-                    mutation = mutation.table(table_id.try_into().unwrap());
+                    mutation = mutation.table_id(table_id.try_into().unwrap());
                 }
-                mutation.execute().await.map_err(|e| anyhow::Error::from(e))
+                mutation.execute().await.map_err(|e| eyre::Error::from(e))
             }
         }
     }
 
     async fn route_del(&mut self, prefix: NLRI, nexthop: IpAddr) -> Result<()> {
-        let nh_octets = match nexthop {
-            IpAddr::V6(addr) => addr.octets().to_vec(),
-            IpAddr::V4(addr) => addr.octets().to_vec(),
-        };
         let rt_handle = self.handle.route();
-        let address_family = match prefix.afi {
-            AddressFamilyIdentifier::Ipv4 => netlink_packet_route::rtnl::constants::AF_INET as u8,
-            AddressFamilyIdentifier::Ipv6 => netlink_packet_route::rtnl::constants::AF_INET6 as u8,
-        };
-        let header = RouteHeader {
-            address_family,
-            destination_prefix_length: prefix.prefixlen,
-            table: self.table.unwrap_or(0) as u8,
-            protocol: RTPROT_STATIC,
-            kind: RTN_UNICAST,
-            ..Default::default()
-        };
-        let mut rt_msg = RouteMessage {
-            header,
-            ..Default::default()
-        };
-        let prefix_octets = match prefix.afi {
+        let destination = match prefix.afi {
             AddressFamilyIdentifier::Ipv4 => {
-                let addr: Ipv4Addr = match prefix.clone().try_into()? {
-                    IpAddr::V4(addr) => addr,
-                    _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
-                            ErrorKind::InvalidInput,
-                            "Got non-IPv4 address from NLRI",
-                        )))
-                    }
-                };
-                addr.octets().to_vec()
+                RouteAddress::Inet(prefix.clone().try_into().map_err(|e: String| eyre!(e))?)
             }
             AddressFamilyIdentifier::Ipv6 => {
-                let addr: Ipv6Addr = match prefix.clone().try_into()? {
-                    IpAddr::V6(addr) => addr,
-                    _ => {
-                        return Err(anyhow::Error::from(std::io::Error::new(
-                            ErrorKind::InvalidInput,
-                            "Got non-IPv6 address from NLRI",
-                        )))
-                    }
-                };
-                addr.octets().to_vec()
+                RouteAddress::Inet6(prefix.clone().try_into().map_err(|e: String| eyre!(e))?)
             }
         };
-        rt_msg.nlas.push(Nla::Destination(prefix_octets));
-        rt_msg.nlas.push(Nla::Gateway(nh_octets));
+        let nexthop = match nexthop {
+            IpAddr::V4(ipv4) => RouteAddress::Inet(ipv4),
+            IpAddr::V6(ipv6) => RouteAddress::Inet6(ipv6),
+        };
+        let header = RouteHeader {
+            address_family: match prefix.afi {
+                AddressFamilyIdentifier::Ipv4 => NetlinkAddressFamily::Inet,
+                AddressFamilyIdentifier::Ipv6 => NetlinkAddressFamily::Inet6,
+            },
+            destination_prefix_length: prefix.prefixlen,
+            table: self.table.unwrap_or(0) as u8,
+            protocol: RouteProtocol::Bgp,
+            kind: RouteType::Unicast,
+            ..Default::default()
+        };
+        let mut rt_msg: RouteMessage = Default::default();
+        rt_msg.header = header;
+        rt_msg.attributes = vec![
+            RouteAttribute::Destination(destination),
+            RouteAttribute::Gateway(nexthop),
+        ];
         rt_handle
             .del(rt_msg)
             .execute()
             .await
-            .map_err(|e| anyhow::Error::from(e))
+            .map_err(|e| eyre::Error::from(e))
     }
 }
 
@@ -165,8 +152,8 @@ impl NetlinkConnector {
         });
         if let Some(table_id) = table {
             req.message_mut()
-                .nlas
-                .push(Nla::Table(table_id.try_into().unwrap()));
+                .attributes
+                .push(RouteAttribute::Table(table_id));
         }
         req.execute().try_collect().await
     }

crates/route_client/src/southbound_interface.rs

@@ -14,23 +14,25 @@
 
 use std::{collections::HashMap, net::IpAddr};
 
-use crate::bgp_packet::{constants::AddressFamilyIdentifier, nlri::NLRI};
-use anyhow::{anyhow, Result};
 use async_trait::async_trait;
+use eyre::{eyre, Result};
 use log::info;
 
+use bgp_packet::{constants::AddressFamilyIdentifier, nlri::NLRI};
+
 /// SouthboundInterface provides a uniform API to network forwarding elements
 /// These are devices or targets that perform packet routing and are the end
 /// consumers of packet routing data.
-
 #[async_trait]
 pub trait SouthboundInterface {
+    /// route_add adds a route towards a particular address_family/NLRI via the given nexthop.
     async fn route_add(
         &mut self,
         address_family: AddressFamilyIdentifier,
         prefix: NLRI,
         nexthop: IpAddr,
     ) -> Result<()>;
+    /// route_del removes the route towards a particular prefix via a given nexthop.
     async fn route_del(&mut self, prefix: NLRI, nexthop: IpAddr) -> Result<()>;
 }
 
@@ -59,7 +61,7 @@ impl SouthboundInterface for DummyVerifier {
         // Check that the route is not already present.
         match self.route_state.get(&prefix) {
             Some(value) => {
-                return Err(anyhow!(
+                return Err(eyre!(
                     "Prefix {} with nexthop {} already contained in route_state! when trying to add {} -> {}",
                     prefix, value, prefix, nexthop,
                 ));
@@ -79,7 +81,7 @@ impl SouthboundInterface for DummyVerifier {
         match self.route_state.remove(&prefix) {
             Some(entry) => {
                 if entry != nexthop {
-                    return Err(anyhow!(
+                    return Err(eyre!(
                         "Removed entry's nexthop did not match: {} vs requested {}",
                         entry,
                         nexthop
@@ -87,7 +89,7 @@ impl SouthboundInterface for DummyVerifier {
                 }
             }
             None => {
-                return Err(anyhow!(
+                return Err(eyre!(
                     "Requested removal of route {} that was not in route_state",
                     prefix
                 ));

crates/server/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "bgp_server"
+
+edition.workspace      = true
+homepage.workspace     = true
+license.workspace      = true
+repository.workspace   = true
+rust-version.workspace = true
+version.workspace      = true
+
+[lints]
+workspace = true
+
+[dependencies]
+bgp_packet.workspace                       = true
+byteorder                                  = "1.4.3"
+bytes.workspace                            = true
+chrono                                     = "0.4.38"
+eyre.workspace                             = true
+ip_network_table-deps-treebitmap.workspace = true
+log.workspace                              = true
+nom                                        = "7.1"
+prost.workspace                            = true
+serde.workspace                            = true
+tokio-stream                               = "0.1.14"
+tokio-util                                 = { version = "0.7.10", features = ["codec"] }
+tokio.workspace                            = true
+tonic.workspace                            = true
+tracing.workspace                          = true
+warp.workspace                             = true
+
+[build-dependencies]
+tonic-build = { version = "0.5.1", features = ["compression", "prost"] }

crates/server/build.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod fib_state;
-pub mod netlink;
-pub mod southbound_interface;
+fn main() {
+    tonic_build::configure()
+        .compile(&["proto/route_service.proto"], &["proto"])
+        .unwrap();
+}

crates/server/proto/route_service.proto

@@ -68,10 +68,10 @@ message PeerStatusRequest {}
 message PeerStatus {
   string peer_name = 1;
   string state = 2;
-  uint64 session_established_time = 3;
-  uint64 last_messaage_time = 4;
-  uint64 route_updates_in = 5;
-  uint64 route_updates_out = 6;
+  optional uint64 session_established_time = 3;
+  optional uint64 last_messaage_time = 4;
+  optional uint64 route_updates_in = 5;
+  optional uint64 route_updates_out = 6;
 }
 
 message PeerStatusResponse { repeated PeerStatus peer_status = 1; }

crates/server/src/bgp_server.rs

@@ -12,26 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::server::config::PeerConfig;
-use crate::server::config::ServerConfig;
-use crate::server::peer::PeerCommands;
-use crate::server::peer::PeerStateMachine;
-use crate::server::rib_manager::RibManager;
-use crate::server::rib_manager::RibSnapshot;
-use crate::server::rib_manager::RouteManagerCommands;
-use crate::server::route_server;
-use crate::server::route_server::route_server::route_service_server::RouteServiceServer;
+use crate::config::PeerConfig;
+use crate::config::ServerConfig;
+use crate::peer::PeerCommands;
+use crate::peer::PeerStateMachine;
+use crate::rib_manager::RibManager;
+use crate::rib_manager::RibSnapshot;
+use crate::rib_manager::RouteManagerCommands;
+use crate::route_server;
+use crate::route_server::route_server::bgp_server_admin_service_server::BgpServerAdminServiceServer;
+use crate::route_server::route_server::route_service_server::RouteServiceServer;
+use bgp_packet::constants::AddressFamilyIdentifier;
 use std::collections::HashMap;
 use std::net::Ipv4Addr;
 use std::net::Ipv6Addr;
 use std::net::SocketAddr;
 use tokio::net::TcpListener;
 use tokio::net::TcpStream;
-use tokio::sync::broadcast;
 use tokio::sync::mpsc::unbounded_channel;
 use tokio::sync::mpsc::UnboundedSender;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 use warp::Filter;
 use warp::Reply;
@@ -44,7 +45,7 @@ async fn socket_listener(
     c: UnboundedSender<(TcpStream, SocketAddr)>,
     listen_addr: String,
     notifier: oneshot::Sender<Result<(), String>>,
-    mut shutdown: broadcast::Receiver<()>,
+    shutdown: CancellationToken,
 ) {
     info!("Starting to listen on addr: {}", listen_addr);
     let listener_result = TcpListener::bind(&listen_addr).await;
@@ -52,7 +53,7 @@ async fn socket_listener(
         warn!("Listener for {} failed: {}", listen_addr, e.to_string());
         match notifier.send(Err(e.to_string())) {
             Ok(_) => {}
-            Err(e) => warn!("Failed to send notification of channel error: {:?}", e),
+            Err(e) => warn!(?e, "Failed to send notification of channel error"),
         }
         return;
     }
@@ -62,16 +63,19 @@ async fn socket_listener(
         Ok(_) => {}
         Err(e) => warn!("Failed to send notification of channel ready: {:?}", e),
     }
-    info!("Sucessfully spawned listner for: {}", listen_addr);
+
+    info!(listen_addr, "Spawned listner");
+
     loop {
         let conn = tokio::select! {
             res = listener.accept() => res,
-            _ = shutdown.recv() => {
+            _ = shutdown.cancelled() => {
                 info!("Shutting down listener");
                 return;
             }
         };
-        info!("Got something: {:?}", conn);
+
+        info!(?conn, "New inbound connection");
         match conn {
             Ok((stream, addr)) => {
                 info!("Accepted socket connection from {}", addr);
@@ -98,7 +102,7 @@ async fn start_http_server(
     manager6: UnboundedSender<RouteManagerCommands<Ipv6Addr>>,
     peers: HashMap<String, UnboundedSender<PeerCommands>>,
     listen_addr: SocketAddr,
-    mut shutdown: broadcast::Receiver<()>,
+    shutdown: CancellationToken,
 ) -> Result<tokio::task::JoinHandle<()>, String> {
     async fn manager_get_routes_handler<T: serde::ser::Serialize>(
         channel: UnboundedSender<RouteManagerCommands<T>>,
@@ -118,48 +122,48 @@ async fn start_http_server(
         }
     }
 
-    async fn rm_large_community(
-        chan: UnboundedSender<PeerCommands>,
-        ld1: u32,
-        ld2: u32,
-    ) -> Result<impl warp::Reply, warp::Rejection> {
-        let (tx, rx) = tokio::sync::oneshot::channel::<String>();
-        if let Err(e) = chan.send(PeerCommands::RemoveLargeCommunity((ld1, ld2), tx)) {
-            warn!("Failed to send RemoveLargeCommunity request: {}", e);
-            return Err(warp::reject());
-        }
+    // async fn rm_large_community(
+    //     chan: UnboundedSender<PeerCommands>,
+    //     ld1: u32,
+    //     ld2: u32,
+    // ) -> Result<impl warp::Reply, warp::Rejection> {
+    //     let (tx, rx) = tokio::sync::oneshot::channel::<String>();
+    //     if let Err(e) = chan.send(PeerCommands::RemoveLargeCommunity((ld1, ld2), tx)) {
+    //         warn!("Failed to send RemoveLargeCommunity request: {}", e);
+    //         return Err(warp::reject());
+    //     }
 
-        match rx.await {
-            Ok(result) => Ok(warp::reply::json(&result)),
-            Err(e) => {
-                warn!(
-                    "RemoveLargeCommunity response from peer state machine: {}",
-                    e
-                );
-                Err(warp::reject())
-            }
-        }
-    }
+    //     match rx.await {
+    //         Ok(result) => Ok(warp::reply::json(&result)),
+    //         Err(e) => {
+    //             warn!(
+    //                 "RemoveLargeCommunity response from peer state machine: {}",
+    //                 e
+    //             );
+    //             Err(warp::reject())
+    //         }
+    //     }
+    // }
 
-    async fn add_large_community(
-        chan: UnboundedSender<PeerCommands>,
-        ld1: u32,
-        ld2: u32,
-    ) -> Result<impl warp::Reply, warp::Rejection> {
-        let (tx, rx) = tokio::sync::oneshot::channel::<String>();
-        if let Err(e) = chan.send(PeerCommands::AddLargeCommunity((ld1, ld2), tx)) {
-            warn!("Failed to send AddLargeCommunity request: {}", e);
-            return Err(warp::reject());
-        }
+    // async fn add_large_community(
+    //     chan: UnboundedSender<PeerCommands>,
+    //     ld1: u32,
+    //     ld2: u32,
+    // ) -> Result<impl warp::Reply, warp::Rejection> {
+    //     let (tx, rx) = tokio::sync::oneshot::channel::<String>();
+    //     if let Err(e) = chan.send(PeerCommands::AddLargeCommunity((ld1, ld2), tx)) {
+    //         warn!("Failed to send AddLargeCommunity request: {}", e);
+    //         return Err(warp::reject());
+    //     }
 
-        match rx.await {
-            Ok(result) => Ok(warp::reply::json(&result)),
-            Err(e) => {
-                warn!("AddLargeCommunity response from peer state machine: {}", e);
-                Err(warp::reject())
-            }
-        }
-    }
+    //     match rx.await {
+    //         Ok(result) => Ok(warp::reply::json(&result)),
+    //         Err(e) => {
+    //             warn!("AddLargeCommunity response from peer state machine: {}", e);
+    //             Err(warp::reject())
+    //         }
+    //     }
+    // }
 
     // reset_peer_connection causes the PSM to close the connection, flush state, and reconnect to the peer.
     async fn reset_peer_connection(
@@ -209,7 +213,6 @@ async fn start_http_server(
             match rx.await {
                 Ok(resp) => {
                     result += &format!("Peer state: <b>{:?}</b><br/>", resp.state);
-                    result += &format!("<code>{:?}</code>", resp.config);
                 }
                 Err(e) => {
                     warn!("error on rx from peer channel: {}", e);
@@ -292,7 +295,7 @@ async fn start_http_server(
         .or(peers_restart_route);
     let (_, server) = warp::serve(routes)
         .try_bind_with_graceful_shutdown(listen_addr, async move {
-            shutdown.recv().await.ok();
+            shutdown.cancelled().await;
         })
         .map_err(|e| e.to_string())?;
     Ok(tokio::task::spawn(server))
@@ -303,7 +306,7 @@ pub struct Server {
     config: ServerConfig,
 
     // shutdown is a channel that a
-    shutdown: broadcast::Sender<()>,
+    shutdown: CancellationToken,
 
     // worker_handles contains the JoinHandle of tasks spawned by the server so that
     // we can wait on them for shutdown.
@@ -315,7 +318,7 @@ pub struct Server {
 
 impl Server {
     pub fn new(config: ServerConfig) -> Server {
-        let (shutdown, _) = broadcast::channel(1);
+        let shutdown = CancellationToken::new();
         Server {
             config,
             shutdown,
@@ -345,9 +348,12 @@ impl Server {
             info!("Starting listener for {}", listen_addr.to_string());
             let sender = tcp_in_tx.clone();
             let (ready_tx, ready_rx) = oneshot::channel();
-            let shutdown_channel = self.shutdown.subscribe();
-            let listen_handle = tokio::spawn(async move {
-                socket_listener(sender, listen_addr.to_string(), ready_tx, shutdown_channel).await;
+
+            let listen_handle = tokio::spawn({
+                let shutdown = self.shutdown.clone();
+                async move {
+                    socket_listener(sender, listen_addr.to_string(), ready_tx, shutdown).await;
+                }
             });
             self.worker_handles.push(listen_handle);
             if wait_startup {
@@ -363,7 +369,7 @@ impl Server {
         let (rp6_tx, rp6_rx) = unbounded_channel::<RouteManagerCommands<Ipv6Addr>>();
         self.mgr_v6 = Some(rp6_tx.clone());
         let mut rib_manager6: RibManager<Ipv6Addr> =
-            RibManager::<Ipv6Addr>::new(rp6_rx, self.shutdown.subscribe()).unwrap();
+            RibManager::<Ipv6Addr>::new(rp6_rx, self.shutdown.clone()).unwrap();
         tokio::spawn(async move {
             match rib_manager6.run().await {
                 Ok(_) => {}
@@ -376,7 +382,7 @@ impl Server {
         let (rp4_tx, rp4_rx) = unbounded_channel::<RouteManagerCommands<Ipv4Addr>>();
         self.mgr_v4 = Some(rp4_tx.clone());
         let mut rib_manager4: RibManager<Ipv4Addr> =
-            RibManager::<Ipv4Addr>::new(rp4_rx, self.shutdown.subscribe()).unwrap();
+            RibManager::<Ipv4Addr>::new(rp4_rx, self.shutdown.clone()).unwrap();
         tokio::spawn(async move {
             match rib_manager4.run().await {
                 Ok(_) => {}
@@ -388,7 +394,6 @@ impl Server {
 
         // Start a PeerStateMachine for every peer that is configured and store its channel so that
         // we can communicate with it.
-
         let mut peer_statemachines: HashMap<String, (PeerConfig, UnboundedSender<PeerCommands>)> =
             HashMap::new();
 
@@ -402,11 +407,10 @@ impl Server {
                         psm_rx,
                         psm_tx.clone(),
                         rp6_tx.clone(),
-                        self.shutdown.subscribe(),
+                        self.shutdown.clone(),
                     );
                     self.worker_handles.push(tokio::spawn(async move {
                         psm.run().await;
-                        warn!("Should not reach here");
                     }));
                 }
                 AddressFamilyIdentifier::Ipv4 => {
@@ -416,11 +420,10 @@ impl Server {
                         psm_rx,
                         psm_tx.clone(),
                         rp4_tx.clone(),
-                        self.shutdown.subscribe(),
+                        self.shutdown.clone(),
                     );
                     self.worker_handles.push(tokio::spawn(async move {
                         psm.run().await;
-                        warn!("Should not reach here");
                     }));
                 }
                 _ => panic!("Unsupported address family: {}", peer_config.afi),
@@ -442,7 +445,7 @@ impl Server {
                 rp6_tx.clone(),
                 peer_chan_map.clone(),
                 addr,
-                self.shutdown.subscribe(),
+                self.shutdown.clone(),
             )
             .await
             .unwrap();
@@ -458,10 +461,12 @@ impl Server {
                 peer_state_machines: peer_chan_map,
             };
 
-            let svc = RouteServiceServer::new(rs);
+            let rs_svc = RouteServiceServer::new(rs.clone());
+            let adm_svc = BgpServerAdminServiceServer::new(rs);
             tokio::spawn(async move {
                 if let Err(e) = tonic::transport::Server::builder()
-                    .add_service(svc)
+                    .add_service(rs_svc)
+                    .add_service(adm_svc)
                     .serve(addr)
                     .await
                 {
@@ -471,12 +476,12 @@ impl Server {
         }
 
         // Event loop for processing inbound connections.
-        let mut shutdown_recv = self.shutdown.subscribe();
+        let shutdown = self.shutdown.clone();
         self.worker_handles.push(tokio::spawn(async move {
             loop {
                 let next = tokio::select! {
                     cmd = tcp_in_rx.recv() => cmd,
-                    _ = shutdown_recv.recv() => {
+                    _ = shutdown.cancelled() => {
                         warn!("Peer connection dispatcher shutting down due to shutdown signal.");
                         return;
                     }
@@ -508,13 +513,7 @@ impl Server {
     }
 
     pub async fn shutdown(&mut self) {
-        match self.shutdown.send(()) {
-            Ok(_) => {}
-            Err(e) => {
-                warn!("Failed to send shutdown signal: {}", e);
-                return;
-            }
-        }
+        self.shutdown.cancel();
         for handle in &mut self.worker_handles {
             match handle.await {
                 Ok(_) => {}

crates/server/src/config.rs

@@ -12,7 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::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)]
@@ -65,9 +97,26 @@ pub struct PrefixAnnouncement {
     /// Linklocal nexthop to be used for IPv6 announcements.
     pub llnh: Option<Ipv6Addr>,
 
-    // Path attributes
+    /// Path attributes
     pub local_pref: Option<u32>,
+    /// Multi exit discriminator
     pub med: Option<u32>,
+    /// Legacy communities [RFC 1997]
     pub communities: Option<Vec<String>>,
+    /// Large communities [RFC 8092]
     pub large_communities: Option<Vec<String>>,
 }
+
+impl Default for PrefixAnnouncement {
+    fn default() -> Self {
+        Self {
+            prefix: "::/0".to_owned(),
+            nexthop: IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1)),
+            llnh: Default::default(),
+            local_pref: Default::default(),
+            med: Default::default(),
+            communities: Default::default(),
+            large_communities: Default::default(),
+        }
+    }
+}

crates/server/src/data_structures.rs

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::nlri::NLRI;
-use crate::bgp_packet::path_attributes::PathAttribute;
-use std::time::SystemTime;
+use bgp_packet::nlri::NLRI;
+use bgp_packet::path_attributes::PathAttribute;
+
+use chrono::{DateTime, Utc};
 
 /// RouteInfo encapsulates information received about a particular BGP route.
 #[derive(Clone, Debug)]
@@ -30,9 +31,9 @@ pub struct RouteInfo<A> {
     pub rejection_reason: Option<String>,
 
     /// Time at which this path was learned from the peer.
-    pub learned: SystemTime,
+    pub learned: DateTime<Utc>,
     /// Time at which this path was last updated by the peer.
-    pub updated: SystemTime,
+    pub updated: DateTime<Utc>,
 
     /// The current path attributes from the UPDATE message where this path
     /// was learned.

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

@@ -12,40 +12,42 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::capabilities::{
+use crate::config::PrefixAnnouncement;
+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::{
     BGPCapability, BGPCapabilityTypeValues, BGPCapabilityValue, BGPOpenOptionTypeValues,
     FourByteASNCapability, MultiprotocolCapability, OpenOption, OpenOptionCapabilities,
     OpenOptions,
 };
-use crate::bgp_packet::constants::{
-    AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier, AS_TRANS,
-};
-use crate::bgp_packet::messages::BGPMessage;
-use crate::bgp_packet::messages::BGPMessageTypeValues;
-use crate::bgp_packet::messages::BGPMessageTypeValues::OPEN_MESSAGE;
-use crate::bgp_packet::messages::BGPMessageTypeValues::UPDATE_MESSAGE;
-use crate::bgp_packet::messages::BGPSubmessage;
-use crate::bgp_packet::messages::Codec;
-use crate::bgp_packet::messages::KeepaliveMessage;
-use crate::bgp_packet::messages::NotificationMessage;
-use crate::bgp_packet::messages::OpenMessage;
-use crate::bgp_packet::messages::UpdateMessage;
-use crate::bgp_packet::nlri::NLRI;
-use crate::bgp_packet::path_attributes::ASPathAttribute;
-use crate::bgp_packet::path_attributes::NextHopPathAttribute;
-use crate::bgp_packet::path_attributes::OriginPathAttribute;
-use crate::bgp_packet::path_attributes::PathAttribute;
-use crate::bgp_packet::path_attributes::{
+use bgp_packet::constants::{AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier, AS_TRANS};
+use bgp_packet::messages::BGPMessage;
+use bgp_packet::messages::BGPMessageTypeValues;
+use bgp_packet::messages::BGPMessageTypeValues::OPEN_MESSAGE;
+use bgp_packet::messages::BGPMessageTypeValues::UPDATE_MESSAGE;
+use bgp_packet::messages::BGPSubmessage;
+use bgp_packet::messages::Codec;
+use bgp_packet::messages::KeepaliveMessage;
+use bgp_packet::messages::NotificationMessage;
+use bgp_packet::messages::OpenMessage;
+use bgp_packet::messages::UpdateMessage;
+use bgp_packet::nlri::NLRI;
+use bgp_packet::path_attributes::ASPathAttribute;
+use bgp_packet::path_attributes::NextHopPathAttribute;
+use bgp_packet::path_attributes::OriginPathAttribute;
+use bgp_packet::path_attributes::PathAttribute;
+use bgp_packet::path_attributes::{
     LargeCommunitiesPathAttribute, LargeCommunitiesPayload, MPReachNLRIPathAttribute,
 };
-use crate::bgp_packet::traits::ParserContext;
-use crate::server::config::PrefixAnnouncement;
-use crate::server::config::{PeerConfig, ServerConfig};
-use crate::server::data_structures::RouteAnnounce;
-use crate::server::data_structures::RouteWithdraw;
-use crate::server::data_structures::{RouteInfo, RouteUpdate};
-use crate::server::rib_manager::RouteManagerCommands;
+use bgp_packet::traits::ParserContext;
 use bytes::BytesMut;
+use chrono::{DateTime, Utc};
+use eyre::{bail, eyre};
 use ip_network_table_deps_treebitmap::address::Address;
 use ip_network_table_deps_treebitmap::IpLookupTable;
 use std::convert::TryFrom;
@@ -59,7 +61,6 @@ use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 use tokio::net::tcp;
 use tokio::net::TcpStream;
-use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::sync::Mutex;
@@ -76,15 +77,6 @@ type PeerInterface = mpsc::UnboundedSender<PeerCommands>;
 // not be expensive, and other tasks such as picking the best route
 // will be done in a different threading model.
 
-/// PeerStatus contians the current state of the PSM for monitoring
-/// and debugging.
-#[derive(Clone, Debug)]
-pub struct PeerStatus {
-    pub name: String,
-    pub config: PeerConfig,
-    pub state: BGPState,
-}
-
 /// BGPState represents which state of the BGP state machine the peer
 /// is currently in.
 #[derive(Copy, Clone, Debug, PartialEq)]
@@ -178,7 +170,7 @@ async fn run_timer(
 async fn check_hold_timer(
     cancel_token: CancellationToken,
     iface: PeerInterface,
-    last_msg_time: Arc<RwLock<std::time::SystemTime>>,
+    last_msg_time: Arc<RwLock<DateTime<Utc>>>,
     hold_time: std::time::Duration,
 ) {
     loop {
@@ -189,23 +181,16 @@ async fn check_hold_timer(
             }
             _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
                 let last =  last_msg_time.read().unwrap();
-                let elapsed_time = std::time::SystemTime::now().duration_since(*last);
-                match elapsed_time {
-                    Ok(duration) => {
-                        if duration > hold_time {
-                            match iface.send(PeerCommands::TimerEvent(PeerTimerEvent::HoldTimerExpire())) {
-                                Ok(()) => {},
-                                Err(e) => {
-                                    warn!("Failed to send HoldTimerExpire message: {}", e);
-                                }
-                            }
-                            // Exit the hold timer task since it's expired already and is not needed anymore.
-                            return;
+                let elapsed_time = Utc::now() - *last;
+                if elapsed_time.num_seconds() as u64 > hold_time.as_secs() {
+                    match iface.send(PeerCommands::TimerEvent(PeerTimerEvent::HoldTimerExpire())) {
+                        Ok(()) => {},
+                        Err(e) => {
+                            warn!("Failed to send HoldTimerExpire message: {}", e);
                         }
                     }
-                    Err(e) => {
-                        warn!("Failed to check duration since last message: {}", e);
-                    }
+                    // Exit the hold timer task since it's expired already and is not needed anymore.
+                    return;
                 }
             }
 
@@ -312,10 +297,16 @@ pub struct PeerStateMachine<A: Address> {
     // server_config is the server wide config that we use here for
     // reading global options.
     server_config: ServerConfig,
+
     /// The current configuration for this peer.
     // To apply a new configuration the peer must be shutdown and
     // 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>,
 
@@ -345,9 +336,12 @@ pub struct PeerStateMachine<A: Address> {
     /// updates from the peer go to rib_in.
     route_manager: mpsc::UnboundedSender<RouteManagerCommands<A>>,
 
+    // The time at which the session was established.
+    established_time: Option<DateTime<Utc>>,
+
     // Keep track of the time of the last message to efficiently implement
     // the hold timer.
-    last_msg_time: Arc<RwLock<std::time::SystemTime>>,
+    last_msg_time: Arc<RwLock<DateTime<Utc>>>,
 
     // Timers and cancellation token to spawned tasks
     connect_timer: Option<(JoinHandle<()>, CancellationToken)>,
@@ -355,7 +349,7 @@ pub struct PeerStateMachine<A: Address> {
     keepalive_timer: Option<(JoinHandle<()>, CancellationToken)>,
     read_cancel_token: Option<CancellationToken>,
 
-    shutdown: broadcast::Receiver<()>,
+    shutdown: CancellationToken,
 }
 
 impl<A: Address> PeerStateMachine<A>
@@ -370,12 +364,13 @@ where
         iface_rx: mpsc::UnboundedReceiver<PeerCommands>,
         iface_tx: mpsc::UnboundedSender<PeerCommands>,
         route_manager: mpsc::UnboundedSender<RouteManagerCommands<A>>,
-        shutdown: broadcast::Receiver<()>,
+        shutdown: CancellationToken,
     ) -> PeerStateMachine<A> {
         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,
@@ -389,7 +384,8 @@ where
             iface_rx,
             iface_tx,
             route_manager,
-            last_msg_time: Arc::new(RwLock::new(std::time::SystemTime::UNIX_EPOCH)),
+            established_time: None,
+            last_msg_time: Arc::new(RwLock::new(DateTime::from_timestamp(0, 0).unwrap())),
             connect_timer: None,
             hold_timer: None,
             keepalive_timer: None,
@@ -423,7 +419,7 @@ where
         loop {
             let next = tokio::select! {
                 cmd = self.iface_rx.recv() => cmd,
-                _ = self.shutdown.recv() => {
+                _ = self.shutdown.cancelled() => {
                     warn!("PSM shutting down due to shutdown signal.");
                     return;
                 },
@@ -446,7 +442,7 @@ where
         }
     }
 
-    async fn handle_chan_msg(&mut self, c: PeerCommands) -> Result<(), std::io::Error> {
+    async fn handle_chan_msg(&mut self, c: PeerCommands) -> eyre::Result<()> {
         match c {
             PeerCommands::NewConnection(mut conn) => {
                 let peer_addr = conn.peer_addr()?;
@@ -528,7 +524,7 @@ where
             }
 
             PeerCommands::AddLargeCommunity(c, sender) => {
-                for mut a in self.config.announcements.iter_mut() {
+                for a in self.config.announcements.iter_mut() {
                     if let Some(lcs) = a.large_communities.as_mut() {
                         lcs.push(format!("{}:{}:{}", self.config.asn, c.0, c.1));
                     } else {
@@ -570,16 +566,14 @@ where
 
             PeerCommands::MessageFromPeer(msg) => match self.handle_msg(msg).await {
                 Ok(_) => {
-                    // Update the last time counter
-                    // We call unwrap here because it indicates that some other thread which
-                    // was accessing the lock had a panic.
-                    // TODO: This should be handled more gracefully, maybe by shutting down the
-                    // peer and starting it up again.
-                    let mut last_time_lock = (*self.last_msg_time).write().unwrap();
-                    *last_time_lock = std::time::SystemTime::now();
+                    let mut last_time = self
+                        .last_msg_time
+                        .write()
+                        .map_err(|e| eyre!(e.to_string()))?;
+                    *last_time = Utc::now();
                 }
                 Err(e) => {
-                    return Err(std::io::Error::new(std::io::ErrorKind::Other, e));
+                    bail!(e);
                 }
             },
             PeerCommands::TimerEvent(timer_event) => match timer_event {
@@ -626,9 +620,12 @@ where
             },
             PeerCommands::GetStatus(sender) => {
                 let state = PeerStatus {
-                    name: self.config.name.clone(),
-                    config: self.config.clone(),
-                    state: self.state,
+                    peer_name: self.config.name.clone(),
+                    state: format!("{:?}", self.state),
+                    session_established_time: self.established_time.map(|t| t.timestamp() as u64),
+                    last_messaage_time: Some(self.last_msg_time.read().unwrap().timestamp() as u64),
+                    route_updates_in: Some(0),  /* todo */
+                    route_updates_out: Some(0), /* todo */
                 };
                 match sender.send(state) {
                     Ok(()) => {}
@@ -717,6 +714,7 @@ where
 
         // Set the state machine back to the expected.
         self.state = BGPState::Active;
+        self.established_time = None;
 
         // Restart the connect timer to try and connect periodically.
         {
@@ -774,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 {
@@ -805,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,
@@ -822,7 +825,7 @@ where
                     // Update the route_info, we need to clone it then reassign.
                     let mut new_route_info: RouteInfo<A> = route_info.clone();
                     new_route_info.path_attributes = route_update.path_attributes.clone();
-                    new_route_info.updated = std::time::SystemTime::now();
+                    new_route_info.updated = Utc::now();
                     self.prefixes_in
                         .insert(addr, announcement.prefixlen.into(), new_route_info);
                 }
@@ -834,8 +837,8 @@ where
                         nlri: announcement.clone(),
                         accepted,
                         rejection_reason,
-                        learned: std::time::SystemTime::now(),
-                        updated: std::time::SystemTime::now(),
+                        learned: Utc::now(),
+                        updated: Utc::now(),
                         path_attributes: route_update.path_attributes.clone(),
                     };
                     self.prefixes_in
@@ -859,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.
 
@@ -955,6 +953,7 @@ where
                         self.config.name, o.asn
                     );
                     self.state = BGPState::Active;
+                    self.established_time = None;
                     if let Some(stream) = self.tcp_stream.as_mut() {
                         stream.shutdown().await.map_err(|e| e.to_string())?;
                     }
@@ -1089,6 +1088,7 @@ where
             BGPSubmessage::KeepaliveMessage(_) => {
                 // Switch the state from OpenConfirm to ESTABLISHED.
                 self.state = BGPState::Established;
+                self.established_time = Some(Utc::now());
 
                 if hold_time > 0 {
                     // Set keepalive timer.
@@ -1191,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 => {
@@ -1201,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,
@@ -1278,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(),
@@ -1332,6 +1332,7 @@ where
                 );
                 Ok(())
             }
+
             BGPSubmessage::KeepaliveMessage(_) => Ok(()),
             _ => Err(format!("Got unexpected message from peer: {:?}", msg)),
         }

crates/server/src/rib_manager.rs

@@ -12,27 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::nlri::NLRI;
-use crate::server::data_structures::RouteAnnounce;
-use std::collections::BTreeMap;
-
-use crate::bgp_packet::path_attributes::PathAttribute;
-use crate::server::config::PeerConfig;
-use crate::server::data_structures::RouteUpdate;
-use crate::server::peer::PeerCommands;
-
-use tracing::{info, trace, warn};
+use crate::config::PeerConfig;
+use crate::data_structures::RouteAnnounce;
+use crate::data_structures::RouteUpdate;
+use crate::peer::PeerCommands;
 
 use std::cmp::Eq;
+use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::convert::TryInto;
+use std::sync::Mutex;
 
+use bgp_packet::nlri::NLRI;
+use bgp_packet::path_attributes::PathAttribute;
 use ip_network_table_deps_treebitmap::address::Address;
 use serde::Serialize;
-use std::sync::Mutex;
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
+use tracing::{info, trace, warn};
 
 use super::data_structures::RouteWithdraw;
 
@@ -107,16 +106,17 @@ pub enum RouteManagerCommands<A> {
 
 pub struct RibManager<A: Address> {
     mgr_rx: mpsc::UnboundedReceiver<RouteManagerCommands<A>>,
+
+    /// Peers configured on this server instance.
     peers: HashMap<String, (PeerConfig, PeerInterface)>,
 
-    // We need to use a mutex for PathSet because IpLookupTable does not return a mut ptr.
     rib: ip_network_table_deps_treebitmap::IpLookupTable<A, Mutex<PathSet<A>>>,
     epoch: u64,
 
     // Handle for streaming updates to PathSets in the RIB.
     pathset_streaming_handle: broadcast::Sender<(u64, PathSet<A>)>,
 
-    shutdown: broadcast::Receiver<()>,
+    shutdown: CancellationToken,
 }
 
 impl<A: Address> RibManager<A>
@@ -127,7 +127,7 @@ where
 {
     pub fn new(
         chan: mpsc::UnboundedReceiver<RouteManagerCommands<A>>,
-        shutdown: broadcast::Receiver<()>,
+        shutdown: CancellationToken,
     ) -> Result<Self, std::io::Error> {
         // TODO: Make this a flag that can be configured.
         let (pathset_tx, _) = broadcast::channel(10_000_000);
@@ -145,8 +145,8 @@ where
         loop {
             let next = tokio::select! {
                 cmd = self.mgr_rx.recv() => cmd,
-                _ = self.shutdown.recv() => {
-                    warn!("RIB manager shutting down due to shutdown signal.");
+                _ = self.shutdown.cancelled() => {
+                    warn!("RIB manager shutting down.");
                     return Ok(());
                 }
             };
@@ -227,7 +227,7 @@ where
                 // reannouncement or fresh announcement.
                 match path_set.paths.get_mut(&update.peer) {
                     // Peer already announced this route before.
-                    Some(mut existing) => {
+                    Some(existing) => {
                         trace!(
                             "Updating existing path attributes for NLRI: {}/{}",
                             addr,
@@ -329,24 +329,24 @@ where
 
 #[cfg(test)]
 mod tests {
-    use crate::bgp_packet::constants::AddressFamilyIdentifier;
-    use crate::bgp_packet::nlri::NLRI;
-    use crate::server::rib_manager::RibManager;
-    use crate::server::rib_manager::RouteAnnounce;
-    use crate::server::rib_manager::RouteManagerCommands;
-    use crate::server::rib_manager::RouteUpdate;
+    use crate::rib_manager::RibManager;
+    use crate::rib_manager::RouteAnnounce;
+    use crate::rib_manager::RouteManagerCommands;
+    use crate::rib_manager::RouteUpdate;
+
+    use bgp_packet::constants::AddressFamilyIdentifier;
+    use bgp_packet::nlri::NLRI;
 
     use std::net::Ipv6Addr;
     use std::str::FromStr;
     use tokio::sync::mpsc;
+    use tokio_util::sync::CancellationToken;
 
     #[test]
     fn test_manager_process_single() {
         let (_, rp_rx) = mpsc::unbounded_channel::<RouteManagerCommands<Ipv6Addr>>();
-        // Nothing spaawned here so no need to send the shutdown signal.
-        let (_shutdown_tx, shutdown_rx) = tokio::sync::broadcast::channel(1);
         let mut rib_manager: RibManager<Ipv6Addr> =
-            RibManager::<Ipv6Addr>::new(rp_rx, shutdown_rx).unwrap();
+            RibManager::<Ipv6Addr>::new(rp_rx, CancellationToken::new()).unwrap();
 
         let nexthop = Ipv6Addr::new(0x20, 0x01, 0xd, 0xb8, 0, 0, 0, 0x1);
 

crates/server/src/route_server.rs

@@ -12,20 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bgp_packet::constants::AddressFamilyIdentifier;
-use crate::server::peer::PeerCommands;
-use crate::server::rib_manager;
-use crate::server::rib_manager::RibSnapshot;
-use crate::server::rib_manager::RouteManagerCommands;
-use crate::server::route_server::route_server::route_service_server::RouteService;
-use crate::server::route_server::route_server::AddressFamily;
-use crate::server::route_server::route_server::DumpPathsRequest;
-use crate::server::route_server::route_server::DumpPathsResponse;
-use crate::server::route_server::route_server::Path;
-use crate::server::route_server::route_server::PathSet;
-use crate::server::route_server::route_server::Prefix;
-use crate::server::route_server::route_server::StreamPathsRequest;
-use log::warn;
+use crate::peer::PeerCommands;
+use crate::rib_manager;
+use crate::rib_manager::RibSnapshot;
+use crate::rib_manager::RouteManagerCommands;
+use crate::route_server::route_server::bgp_server_admin_service_server::BgpServerAdminService;
+use crate::route_server::route_server::route_service_server::RouteService;
+use crate::route_server::route_server::AddressFamily;
+use crate::route_server::route_server::DumpPathsRequest;
+use crate::route_server::route_server::DumpPathsResponse;
+use crate::route_server::route_server::Path;
+use crate::route_server::route_server::PathSet;
+use crate::route_server::route_server::Prefix;
+use crate::route_server::route_server::StreamPathsRequest;
+use bgp_packet::constants::AddressFamilyIdentifier;
+use route_server::PeerStatusRequest;
+use route_server::PeerStatusResponse;
 use std::collections::HashMap;
 use std::net::Ipv4Addr;
 use std::net::Ipv6Addr;
@@ -36,11 +38,13 @@ use tokio::sync::oneshot;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::Response;
 use tonic::Status;
+use tracing::warn;
 
 pub mod route_server {
     tonic::include_proto!("bgpd.grpc");
 }
 
+#[derive(Clone)]
 pub struct RouteServer {
     pub ip4_manager: UnboundedSender<RouteManagerCommands<Ipv4Addr>>,
     pub ip6_manager: UnboundedSender<RouteManagerCommands<Ipv6Addr>>,
@@ -98,6 +102,31 @@ impl RouteServer {
     }
 }
 
+#[tonic::async_trait]
+impl BgpServerAdminService for RouteServer {
+    async fn peer_status(
+        &self,
+        request: tonic::Request<PeerStatusRequest>,
+    ) -> Result<Response<PeerStatusResponse>, Status> {
+        let mut result = PeerStatusResponse::default();
+
+        for peer in &self.peer_state_machines {
+            let (tx, rx) = oneshot::channel();
+            if let Err(e) = peer.1.send(PeerCommands::GetStatus(tx)) {
+                warn!(
+                    peer = peer.0,
+                    "Peer channel dead when trying to send state request"
+                );
+                continue;
+            }
+            let resp = rx.await.map_err(|e| Status::internal(format!("{}", e)))?;
+            result.peer_status.push(resp);
+        }
+
+        Ok(Response::new(result))
+    }
+}
+
 #[tonic::async_trait]
 impl RouteService for RouteServer {
     async fn dump_paths(

tests/integration_tests/Cargo.toml

@@ -1,21 +1,20 @@
 [package]
-name = "integration_tests"
-version = "0.1.0"
 authors = ["Rayhaan Jaufeerally <rayhaan@rayhaan.ch>"]
 edition = "2018"
 license = "Apache-2.0"
+name    = "integration_tests"
+version = "0.1.0"
 
 [dependencies]
-bgpd = { path = "../../bgpd" }
-bytes = "1.*"
-tokio = { version = "1.6.1", features = ["full"] }
-tokio-util = { version = "0.6.7", features = ["codec"] }
-tracing = "0.1"
-tracing-subscriber = "0.2"
-libc = "0.2.126"
+bgp_packet.workspace   = true
+bgp_server.workspace   = true
+bytes                  = "1.*"
+libc                   = "0.2.126"
+route_client.workspace = true
+tokio                  = { version = "1.6.1", features = ["full"] }
+tokio-util             = { version = "0.7.10", features = ["codec"] }
+tracing                = "0.1"
+tracing-subscriber     = "0.2"
 
 [dev-dependencies]
 serial_test = "0.5.1"
-
-[unstable]
-thread_id_value = true

tests/integration_tests/tests/basic_startup.rs

@@ -12,12 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use bgpd::bgp_packet;
-use bgpd::bgp_packet::constants::{AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier};
-use bgpd::bgp_packet::messages::BGPSubmessage;
-use bgpd::bgp_packet::traits::ParserContext;
-use bgpd::server::bgp_server::Server;
-use bgpd::server::config::{PeerConfig, ServerConfig};
 use std::io::{Read, Write};
 use std::mem::size_of;
 use std::net::Ipv4Addr;
@@ -30,12 +24,20 @@ use std::time::Duration;
 use tokio_util::codec::Decoder;
 use tracing::info;
 
+use bgp_packet::constants::{AddressFamilyIdentifier, SubsequentAddressFamilyIdentifier};
+use bgp_packet::messages::BGPSubmessage;
+use bgp_packet::traits::ParserContext;
+use bgp_server::bgp_server::Server;
+use bgp_server::config::{PeerConfig, PrefixAnnouncement, ServerConfig};
+use bgp_server::route_server::route_server::bgp_server_admin_service_client::BgpServerAdminServiceClient;
+use bgp_server::route_server::route_server::PeerStatusRequest;
+
 #[macro_use]
 extern crate serial_test;
 
 fn init() {
     match tracing_subscriber::fmt()
-        .with_env_filter("bgpd=trace,tokio=trace,basic_startup=trace")
+        // .with_env_filter("server=trace,tokio=trace,basic_startup=trace")
         .try_init()
     {
         Ok(()) => {}
@@ -123,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(),
         }],
     };
 
@@ -213,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(),
         }],
     };
 
@@ -301,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(),
         }],
     };
 
@@ -435,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(),
         }],
     };
 
@@ -513,7 +523,6 @@ async fn test_bgp_listener_known_peer_inbound_reconnection() {
         }
     }
 
-    // conn.shutdown(std::net::Shutdown::Both).unwrap();
     drop(conn);
 
     // Try to connect to localhost:9179 and it should connect and send the OPEN message.
@@ -556,3 +565,76 @@ async fn test_bgp_listener_known_peer_inbound_reconnection() {
 
     bgp_server.shutdown().await;
 }
+
+// Spawn two instances of the BGP server and make them peer with each other and exchange a route.
+// We then check that the route is accepted by the receiving peer.
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn test_multi_instance_announce() {
+    init();
+
+    let v6_addr: Ipv6Addr = "::1".parse().unwrap();
+    let config_a = ServerConfig {
+        asn: 65535,
+        hold_time: 3,
+        identifier: Ipv4Addr::new(127, 0, 0, 1),
+        grpc_addr: Some("[::]:9181".to_owned()),
+        http_addr: None,
+        listen_addrs: vec!["[::]:9179".to_owned()],
+        peers: vec![PeerConfig {
+            afi: AddressFamilyIdentifier::Ipv6,
+            safi: SubsequentAddressFamilyIdentifier::Unicast,
+            asn: 65536,
+            ip: IpAddr::V6(v6_addr),
+            port: Some(9180),
+            announcements: vec![PrefixAnnouncement {
+                prefix: "2001:db8:babe::/48".to_owned(),
+                nexthop: "2001:db8::1".parse().unwrap(),
+                local_pref: Some(100),
+                med: Some(100),
+                ..Default::default()
+            }],
+            name: "config-b-peer".to_string(),
+            local_pref: 100,
+            filter_in: Vec::default(),
+            filter_out: Vec::default(),
+        }],
+    };
+
+    let mut config_b = config_a.clone();
+    config_b.asn = 65536;
+    config_b.listen_addrs = vec!["[::]:9180".to_owned()];
+    config_b.grpc_addr = Some("[::]:9182".to_owned());
+    config_b.peers[0].asn = 65535;
+    config_b.peers[0].port = Some(9179);
+    config_b.peers[0].name = "config-a-peer".to_owned();
+
+    let mut server_a = Server::new(config_a);
+    server_a.start(true).await.unwrap();
+
+    let mut server_b = Server::new(config_b);
+    server_b.start(true).await.unwrap();
+
+    // Connect to the grpc endpoint of server_b and check until the peer is healthy.
+    let mut stub = BgpServerAdminServiceClient::connect("http://[::1]:9182")
+        .await
+        .unwrap();
+
+    let mut established = false;
+
+    for _ in 1..10 {
+        let response = stub
+            .peer_status(PeerStatusRequest::default())
+            .await
+            .unwrap()
+            .into_inner();
+        info!(?response);
+
+        if response.peer_status[0].state == "Established" {
+            established = true;
+        }
+        tokio::time::sleep(Duration::from_secs(1)).await;
+    }
+
+    assert!(established);
+}