Implement DevAddr / JoinEUI prefix filters.

This makes it possible to not only forward data of certain Gateway IDs
to a server, but also only forward data in case it matches the
configured DevAddr / JoinEUI filters.

Author: brocaar

Cargo.lock

@@ -17,6 +17,8 @@
     "derive",
   ] }
   serde = { version = "1.0", features = ["derive"] }
+  serde_json = "1.0"
+  base64 = "0.22"
   toml = { version = "0.9", default-features = false, features = [
     "std",
     "parse",

Cargo.toml

@@ -45,11 +45,36 @@ pub fn run(config: &Configuration) {
   #   # If not set, data of all gateways will be forwarded. If set, only data
   #   # from gateways with a matching Gateway ID will be forwarded.
   #   #
-  #   # Examplex:
+  #   # Example:
   #   # * "0102030405060708/32": Exact match (all 32 bits of the filter must match)
   #   # * "0102030400000000/16": All gateway IDs starting with "01020304" (filter on 16 most significant bits)
   #   gateway_id_prefixes=[]
-  {{#each multiplexer.servers}}
+  #
+  #   # Filter configuration.
+  #   [multiplexer.server.filters]
+
+  #     # DevAddr prefix filters.
+  #     #
+  #     # Example configuration:
+  #     # dev_addr_prefixes=["0000ff00/24"]
+  #     #
+  #     # The above filter means that the 24MSB of 0000ff00 will be used to
+  #     # filter DevAddrs. Uplinks with DevAddrs that do not match any of the
+  #     # configured filters will not be forwarded. Leaving this option empty
+  #     # disables filtering on DevAddr.
+  #     dev_addr_prefixes=[]
+
+  #     # JoinEUI prefix filters.
+  #     #
+  #     # Example configuration:
+  #     # join_eui_prefixes=["0000ff0000000000/24"]
+  #     #
+  #     # The above filter means that the 24MSB of 0000ff0000000000 will be used
+  #     # to filter JoinEUIs. Uplinks with JoinEUIs that do not match any of the
+  #     # configured filters will not be forwarded. Leaving this option empty
+  #     # disables filtering on JoinEUI.
+  #     join_eui_prefixes=[]
+  {{#each multiplexer.server}}
   [[multiplexer.server]]
     server="{{this.server}}"
     uplink_only={{this.uplink_only}}
@@ -59,6 +84,18 @@ pub fn run(config: &Configuration) {
       {{/each}}
     ]
 
+    [multiplexer.server.filters]
+      dev_addr_prefixes=[
+        {{#each this.filters.dev_addr_prefixes}}
+        "{{this}}",
+        {{/each}}
+      ]
+
+      join_eui_prefixes=[
+        {{#each this.filters.join_eui_prefixes}}
+        "{{this}}",
+        {{/each}}
+      ]
   {{/each}}
 
 

src/cmd/configfile.rs

@@ -66,6 +66,14 @@ pub struct Server {
     pub server: String,
     pub uplink_only: bool,
     pub gateway_id_prefixes: Vec<lrwn_filters::EuiPrefix>,
+    pub filters: Filters,
+}
+
+#[derive(Serialize, Deserialize, Default, Clone)]
+#[serde(default)]
+pub struct Filters {
+    pub dev_addr_prefixes: Vec<lrwn_filters::DevAddrPrefix>,
+    pub join_eui_prefixes: Vec<lrwn_filters::EuiPrefix>,
 }
 
 #[derive(Default, Serialize, Deserialize, Clone)]

src/config.rs

@@ -10,7 +10,7 @@ use tracing::{Instrument, debug, error, info, trace, warn};
 
 use crate::config;
 use crate::monitoring::{inc_server_udp_received_count, inc_server_udp_sent_count};
-use crate::packets::{GatewayId, PacketType, get_random_token};
+use crate::packets::{GatewayId, PacketType, PushData, get_random_token};
 use crate::traits::PrintFullError;
 
 static SERVERS: OnceCell<RwLock<Vec<Server>>> = OnceCell::const_new();
@@ -19,6 +19,7 @@ struct Server {
     server: String,
     uplink_only: bool,
     gateway_id_prefixes: Vec<lrwn_filters::EuiPrefix>,
+    filters: lrwn_filters::Filters,
     downlink_tx: UnboundedSender<(GatewayId, Vec<u8>)>,
     sockets: HashMap<GatewayId, ServerSocket>,
 }
@@ -106,6 +107,10 @@ pub async fn setup(
             server.server.clone(),
             server.uplink_only,
             server.gateway_id_prefixes.clone(),
+            lrwn_filters::Filters {
+                dev_addr_prefixes: server.filters.dev_addr_prefixes.clone(),
+                join_eui_prefixes: server.filters.join_eui_prefixes.clone(),
+            },
             downlink_tx.clone(),
         )
         .await?;
@@ -139,19 +144,31 @@ async fn handle_uplink_packet(gateway_id: GatewayId, data: &[u8]) -> Result<()>
             continue;
         }
 
+        let filters = server.filters.clone();
         let socket = server.get_server_socket(gateway_id).await?;
         socket.last_uplink = SystemTime::now();
 
         let span = tracing::info_span!("", addr = %socket.socket.peer_addr().unwrap());
         let _enter = span.enter();
 
         match packet_type {
-            PacketType::PushData => {
-                info!(packet_type = %packet_type, "Sending UDP packet");
-                socket.push_data_token = Some(random_token);
-                socket.socket.send(data).await.context("Send UDP packet")?;
-                inc_server_udp_sent_count(&server.server, packet_type).await;
-            }
+            PacketType::PushData => match PushData::from_slice(data) {
+                Ok(mut push_data) => {
+                    push_data.payload.filter_rxpk(&filters);
+
+                    if push_data.payload.is_empty() {
+                        debug!("Nothing to send, UDP packet is empty or does not match filters");
+                    } else {
+                        info!(packet_type = %packet_type, "Sending UDP packet");
+                        socket.push_data_token = Some(random_token);
+                        socket.socket.send(data).await.context("Send UDP packet")?;
+                        inc_server_udp_sent_count(&server.server, packet_type).await;
+                    }
+                }
+                Err(e) => {
+                    error!("Decode PushData payload error: {}", e);
+                }
+            },
             PacketType::PullData => {
                 info!(packet_type = %packet_type, "Sending UDP packet");
                 socket.pull_data_token = Some(random_token);
@@ -278,6 +295,7 @@ async fn add_server(
     server: String,
     uplink_only: bool,
     gateway_id_prefixes: Vec<lrwn_filters::EuiPrefix>,
+    filters: lrwn_filters::Filters,
     downlink_tx: UnboundedSender<(GatewayId, Vec<u8>)>,
 ) -> Result<()> {
     info!(
@@ -296,6 +314,7 @@ async fn add_server(
         server,
         uplink_only,
         gateway_id_prefixes,
+        filters,
         downlink_tx,
         sockets: HashMap::new(),
     });

src/forwarder.rs

@@ -1,6 +1,8 @@
+use std::collections::HashMap;
 use std::fmt;
 
 use anyhow::{Result, anyhow};
+use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Copy, Debug)]
 pub enum PacketType {
@@ -111,3 +113,95 @@ pub fn get_random_token(v: &[u8]) -> Result<u16> {
 
     Ok(u16::from_be_bytes([v[1], v[2]]))
 }
+
+pub struct PushData {
+    pub protocol_version: u8,
+    pub random_token: u16,
+    pub gateway_id: [u8; 8],
+    pub payload: PushDataPayload,
+}
+
+impl PushData {
+    pub fn from_slice(b: &[u8]) -> Result<Self> {
+        if b.len() < 14 {
+            return Err(anyhow!("At least 14 bytes are expected"));
+        }
+
+        Ok(PushData {
+            protocol_version: b[0],
+            random_token: u16::from_be_bytes([b[1], b[2]]),
+            gateway_id: {
+                let mut gateway_id: [u8; 8] = [0; 8];
+                gateway_id.copy_from_slice(&b[4..12]);
+                gateway_id
+            },
+            payload: serde_json::from_slice(&b[12..])?,
+        })
+    }
+
+    pub fn to_bytes(&self) -> Vec<u8> {
+        let mut b = vec![self.protocol_version];
+
+        b.append(&mut self.random_token.to_be_bytes().to_vec());
+        b.push(0x00);
+        b.append(&mut self.gateway_id.to_vec());
+
+        let mut j = serde_json::to_vec(&self.payload).unwrap();
+        b.append(&mut j);
+
+        b
+    }
+}
+
+#[derive(Default, Serialize, Deserialize)]
+#[serde(default)]
+pub struct PushDataPayload {
+    pub rxpk: Vec<RxPk>,
+
+    // Capture all the other fields.
+    #[serde(flatten)]
+    other: HashMap<String, serde_json::Value>,
+}
+
+impl PushDataPayload {
+    pub fn filter_rxpk(&mut self, filter: &lrwn_filters::Filters) {
+        self.rxpk = self
+            .rxpk
+            .drain(..)
+            .filter(|v| lrwn_filters::matches(&v.data, filter))
+            .collect();
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.rxpk.is_empty() && self.other.is_empty()
+    }
+}
+
+#[derive(Default, Serialize, Deserialize)]
+#[serde(default)]
+pub struct RxPk {
+    #[serde(with = "base64_codec")]
+    pub data: Vec<u8>,
+
+    // Capture all the other fields.
+    #[serde(flatten)]
+    other: HashMap<String, serde_json::Value>,
+}
+
+mod base64_codec {
+    use base64::{Engine as _, engine::general_purpose};
+    use serde::{Deserialize, Serialize};
+    use serde::{Deserializer, Serializer};
+
+    pub fn serialize<S: Serializer>(v: &Vec<u8>, s: S) -> Result<S::Ok, S::Error> {
+        let base64 = general_purpose::STANDARD.encode(v);
+        String::serialize(&base64, s)
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Vec<u8>, D::Error> {
+        let base64 = String::deserialize(d)?;
+        general_purpose::STANDARD
+            .decode(base64.as_bytes())
+            .map_err(serde::de::Error::custom)
+    }
+}

src/packets.rs

@@ -46,10 +46,12 @@ async fn test() {
     let gw_sock = UdpSocket::bind("0.0.0.0:0").await.unwrap();
     gw_sock.connect("localhost:1710").await.unwrap();
 
-    // Send PUSH_DATA.
+    // Send PUSH_DATA (unconfirmed uplink with DevAddr 01020304).
     gw_sock
         .send(&[
-            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x7d,
+            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x22,
+            0x72, 0x78, 0x70, 0x6b, 0x22, 0x3a, 0x5b, 0x7b, 0x22, 0x64, 0x61, 0x74, 0x61, 0x22,
+            0x3a, 0x22, 0x51, 0x41, 0x51, 0x44, 0x41, 0x67, 0x45, 0x3d, 0x22, 0x7d, 0x5d, 0x7d,
         ])
         .await
         .unwrap();
@@ -58,7 +60,9 @@ async fn test() {
     let size = server1_sock.recv(&mut buffer).await.unwrap();
     assert_eq!(
         &[
-            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x7d,
+            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x22,
+            0x72, 0x78, 0x70, 0x6b, 0x22, 0x3a, 0x5b, 0x7b, 0x22, 0x64, 0x61, 0x74, 0x61, 0x22,
+            0x3a, 0x22, 0x51, 0x41, 0x51, 0x44, 0x41, 0x67, 0x45, 0x3d, 0x22, 0x7d, 0x5d, 0x7d,
         ],
         &buffer[..size]
     );

tests/test_filtered_server.rs

@@ -0,0 +1,68 @@
+use std::str::FromStr;
+
+use tokio::net::UdpSocket;
+use tracing_subscriber::prelude::*;
+
+use chirpstack_packet_multiplexer::{config, forwarder, listener};
+
+#[tokio::test]
+async fn test() {
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+
+    let conf = config::Configuration {
+        multiplexer: config::Multiplexer {
+            bind: "0.0.0.0:1710".into(),
+            servers: vec![config::Server {
+                server: "localhost:1711".into(),
+                filters: config::Filters {
+                    dev_addr_prefixes: vec![
+                        lrwn_filters::DevAddrPrefix::from_str("01000000/8").unwrap(),
+                    ],
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+        },
+        ..Default::default()
+    };
+
+    let (downlink_tx, uplink_rx) = listener::setup(&conf.multiplexer.bind).await.unwrap();
+    forwarder::setup(downlink_tx, uplink_rx, conf.multiplexer.servers.clone())
+        .await
+        .unwrap();
+    let mut buffer: [u8; 65535] = [0; 65535];
+
+    // Server socket.
+    let server_sock = UdpSocket::bind("0.0.0.0:1711").await.unwrap();
+
+    // Gateway socket.
+    let gw_sock = UdpSocket::bind("0.0.0.0:0").await.unwrap();
+    gw_sock.connect("localhost:1710").await.unwrap();
+
+    // Send PUSH_DATA (unconfirmed uplink with DevAddr 01020304).
+    gw_sock
+        .send(&[
+            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x22,
+            0x72, 0x78, 0x70, 0x6b, 0x22, 0x3a, 0x5b, 0x7b, 0x22, 0x64, 0x61, 0x74, 0x61, 0x22,
+            0x3a, 0x22, 0x51, 0x41, 0x51, 0x44, 0x41, 0x67, 0x45, 0x3d, 0x22, 0x7d, 0x5d, 0x7d,
+        ])
+        .await
+        .unwrap();
+
+    // Expect PUSH_ACK.
+    let size = gw_sock.recv(&mut buffer).await.unwrap();
+    assert_eq!(&[0x02, 0x01, 0x02, 0x01], &buffer[..size]);
+
+    // Expect PUSH_DATA forwarded to server.
+    let size = server_sock.recv(&mut buffer).await.unwrap();
+    assert_eq!(
+        &[
+            0x02, 0x01, 0x02, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x7b, 0x22,
+            0x72, 0x78, 0x70, 0x6b, 0x22, 0x3a, 0x5b, 0x7b, 0x22, 0x64, 0x61, 0x74, 0x61, 0x22,
+            0x3a, 0x22, 0x51, 0x41, 0x51, 0x44, 0x41, 0x67, 0x45, 0x3d, 0x22, 0x7d, 0x5d, 0x7d,
+        ],
+        &buffer[..size]
+    );
+}