Add flow-filter stage to assert packets belong to a valid peering connection

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "dpdk-sys",
     "dpdk-sysroot-helper",
     "errno",
+    "flow-filter",
     "flow-info",
     "gwname",
     "hardware",
@@ -55,6 +56,7 @@ dpdk-sys = { path = "./dpdk-sys", package = "dataplane-dpdk-sys", features = []
 dpdk-sysroot-helper = { path = "./dpdk-sysroot-helper", package = "dataplane-dpdk-sysroot-helper", features = [] }
 dplane-rpc = { git = "https://github.com/githedgehog/dplane-rpc.git", rev = "e8fc33db10e1d00785f2a2b90cbadcad7900f200", features = [] }
 errno = { path = "./errno", package = "dataplane-errno", features = [] }
+flow-filter = { path = "./flow-filter", package = "dataplane-flow-filter", features = [] }
 flow-info = { path = "./flow-info", package = "dataplane-flow-info", features = [] }
 gateway_config = { git = "https://github.com/githedgehog/gateway-proto", tag = "v0.20.0", features = [] }
 gwname = { path = "./gwname", package = "dataplane-gwname", features = [] }

dataplane/Cargo.toml

@@ -15,6 +15,7 @@ concurrency = { workspace = true }
 ctrlc = { workspace = true, features = ["termination"] }
 dpdk = { workspace = true }
 dyn-iter = { workspace = true }
+flow-filter = { workspace = true }
 futures = { workspace = true }
 gwname = { workspace = true }
 hyper = { workspace = true }

dataplane/src/main.rs

@@ -165,6 +165,7 @@ fn main() {
             nattablesw: setup.nattablesw,
             natallocatorw: setup.natallocatorw,
             vpcdtablesw: setup.vpcdtablesw,
+            flowfilterw: setup.flowfiltertablesw,
             vpc_stats_store: setup.vpc_stats_store,
         },
     })

dataplane/src/packet_processor/mod.rs

@@ -12,6 +12,7 @@ use super::packet_processor::ipforward::IpForwarder;
 
 use concurrency::sync::Arc;
 
+use flow_filter::{FlowFilter, FlowFilterTableWriter};
 use pkt_meta::dst_vpcd_lookup::{DstVpcdLookup, VpcDiscTablesWriter};
 use pkt_meta::flow_table::{ExpirationsNF, FlowTable, LookupNF};
 
@@ -39,6 +40,7 @@ where
     pub nattablesw: NatTablesWriter,
     pub natallocatorw: NatAllocatorWriter,
     pub vpcdtablesw: VpcDiscTablesWriter,
+    pub flowfiltertablesw: FlowFilterTableWriter,
     pub stats: StatsCollector,
     pub vpc_stats_store: Arc<VpcStatsStore>,
 }
@@ -50,6 +52,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let nattablesw = NatTablesWriter::new();
     let natallocatorw = NatAllocatorWriter::new();
     let vpcdtablesw = VpcDiscTablesWriter::new();
+    let flowfiltertablesw = FlowFilterTableWriter::new();
     let router = Router::new(params)?;
     let vpcmapw = VpcMapWriter::<VpcMapName>::new();
 
@@ -66,6 +69,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let iftr_factory = router.get_iftabler_factory();
     let fibtr_factory = router.get_fibtr_factory();
     let vpcdtablesr_factory = vpcdtablesw.get_reader_factory();
+    let flowfiltertablesr_factory = flowfiltertablesw.get_reader_factory();
     let atabler_factory = router.get_atabler_factory();
     let nattabler_factory = nattablesw.get_reader_factory();
     let natallocator_factory = natallocatorw.get_reader_factory();
@@ -87,6 +91,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         let stats_stage = Stats::new("stats", writer.clone());
         let flow_lookup_nf = LookupNF::new("flow-lookup", flow_table.clone());
         let flow_expirations_nf = ExpirationsNF::new(flow_table.clone());
+        let flow_filter = FlowFilter::new("flow-filter", flowfiltertablesr_factory.handle());
 
         // Build the pipeline for a router. The composition of the pipeline (in stages) is currently
         // hard-coded. In any pipeline, the Stats and ExpirationsNF stages should go last
@@ -95,6 +100,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
             .add_stage(iprouter1)
             .add_stage(dst_vpcd_lookup)
             .add_stage(flow_lookup_nf)
+            .add_stage(flow_filter)
             .add_stage(stateless_nat)
             .add_stage(stateful_nat)
             .add_stage(iprouter2)
@@ -111,6 +117,7 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         nattablesw,
         natallocatorw,
         vpcdtablesw,
+        flowfiltertablesw,
         stats,
         vpc_stats_store,
     })

flow-filter/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "dataplane-flow-filter"
+edition.workspace = true
+license.workspace = true
+publish.workspace = true
+version.workspace = true
+
+[dependencies]
+config = { workspace = true }
+left-right = { workspace = true }
+linkme = { workspace = true }
+lpm = { workspace = true }
+net = { workspace = true }
+pipeline = { workspace = true }
+tracectl = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+lpm = { workspace = true, features = ["testing"] }
+tracing-test = { workspace = true, features = [] }

flow-filter/src/filter_rw.rs

@@ -0,0 +1,80 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Open Network Fabric Authors
+
+//! Left-right integration for [`FlowFilterTable`]
+
+use crate::tables::FlowFilterTable;
+use left_right::{Absorb, ReadGuard, ReadHandle, ReadHandleFactory, WriteHandle, new_from_empty};
+use tracing::debug;
+
+#[derive(Debug)]
+pub(crate) enum FlowFilterTableChange {
+    UpdateFlowFilterTable(FlowFilterTable),
+}
+
+impl Absorb<FlowFilterTableChange> for FlowFilterTable {
+    fn absorb_first(&mut self, change: &mut FlowFilterTableChange, _: &Self) {
+        match change {
+            FlowFilterTableChange::UpdateFlowFilterTable(table) => {
+                *self = table.clone();
+            }
+        }
+    }
+    fn drop_first(self: Box<Self>) {}
+    fn sync_with(&mut self, first: &Self) {
+        *self = first.clone();
+    }
+}
+
+#[derive(Debug)]
+pub struct FlowFilterTableReader(ReadHandle<FlowFilterTable>);
+
+impl FlowFilterTableReader {
+    pub(crate) fn enter(&self) -> Option<ReadGuard<'_, FlowFilterTable>> {
+        self.0.enter()
+    }
+
+    #[must_use]
+    pub fn factory(&self) -> FlowFilterTableReaderFactory {
+        FlowFilterTableReaderFactory(self.0.factory())
+    }
+}
+
+#[derive(Debug)]
+pub struct FlowFilterTableReaderFactory(ReadHandleFactory<FlowFilterTable>);
+
+impl FlowFilterTableReaderFactory {
+    #[must_use]
+    pub fn handle(&self) -> FlowFilterTableReader {
+        FlowFilterTableReader(self.0.handle())
+    }
+}
+
+#[derive(Debug)]
+pub struct FlowFilterTableWriter(WriteHandle<FlowFilterTable, FlowFilterTableChange>);
+
+impl FlowFilterTableWriter {
+    #[must_use]
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> FlowFilterTableWriter {
+        let (w, _r) =
+            new_from_empty::<FlowFilterTable, FlowFilterTableChange>(FlowFilterTable::new());
+        FlowFilterTableWriter(w)
+    }
+
+    #[must_use]
+    pub fn get_reader(&self) -> FlowFilterTableReader {
+        FlowFilterTableReader(self.0.clone())
+    }
+
+    pub fn get_reader_factory(&self) -> FlowFilterTableReaderFactory {
+        self.get_reader().factory()
+    }
+
+    pub fn update_flow_filter_table(&mut self, table: FlowFilterTable) {
+        self.0
+            .append(FlowFilterTableChange::UpdateFlowFilterTable(table));
+        self.0.publish();
+        debug!("Updated flow filter table");
+    }
+}