Implement DevAddr and JoinEUI prefix filters.

Closes #25.

Author: brocaar

Cargo.lock

@@ -19,6 +19,7 @@ clap = { version = "4.1", default-features = false, features = [
 chirpstack_api = { version = "4.3.1", default-features = false, features = [
     "json",
 ] }
+lrwn_filters = { version = "4.3.3-test.1", features = ["serde"] }
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 log = "0.4"

Cargo.toml

@@ -56,13 +56,18 @@ impl Backend {
         info!("Received gateway id, gateway_id: {}", gateway_id);
 
         tokio::spawn({
-            let forward_crc_ok = conf.backend.forward_crc_ok;
-            let forward_crc_invalid = conf.backend.forward_crc_invalid;
-            let forward_crc_missing = conf.backend.forward_crc_missing;
+            let forward_crc_ok = conf.backend.filters.forward_crc_ok;
+            let forward_crc_invalid = conf.backend.filters.forward_crc_invalid;
+            let forward_crc_missing = conf.backend.filters.forward_crc_missing;
+            let filters = lrwn_filters::Filters {
+                dev_addr_prefixes: conf.backend.filters.dev_addr_prefixes.clone(),
+                join_eui_prefixes: conf.backend.filters.join_eui_prefixes.clone(),
+            };
 
             async move {
                 event_loop(
                     event_sock,
+                    filters,
                     forward_crc_ok,
                     forward_crc_invalid,
                     forward_crc_missing,
@@ -148,6 +153,7 @@ impl BackendTrait for Backend {
 
 async fn event_loop(
     event_sock: zmq::Socket,
+    filters: lrwn_filters::Filters,
     forward_crc_ok: bool,
     forward_crc_invalid: bool,
     forward_crc_missing: bool,
@@ -187,6 +193,7 @@ async fn event_loop(
                 if let Err(err) = handle_event_msg(
                     &msg[0],
                     &msg[1],
+                    &filters,
                     forward_crc_ok,
                     forward_crc_invalid,
                     forward_crc_missing,
@@ -212,6 +219,7 @@ async fn event_loop(
 async fn handle_event_msg(
     event: &[u8],
     pl: &[u8],
+    filters: &lrwn_filters::Filters,
     forward_crc_ok: bool,
     forward_crc_invalid: bool,
     forward_crc_missing: bool,
@@ -235,11 +243,18 @@ async fn handle_event_msg(
                 }
             }
 
-            info!(
-                "Received uplink frame, uplink_id: {}",
-                pl.rx_info.as_ref().map(|v| v.uplink_id).unwrap_or_default(),
-            );
-            send_uplink_frame(&pl).await?;
+            if lrwn_filters::matches(&pl.phy_payload, filters) {
+                info!(
+                    "Received uplink frame, uplink_id: {}",
+                    pl.rx_info.as_ref().map(|v| v.uplink_id).unwrap_or_default(),
+                );
+                send_uplink_frame(&pl).await?;
+            } else {
+                debug!(
+                    "Ignoring uplink frame because of dev_addr and join_eui filters, uplink_id: {}",
+                    pl.rx_info.as_ref().map(|v| v.uplink_id).unwrap_or_default()
+                );
+            }
         }
         "stats" => {
             let pl = gw::GatewayStats::decode(pl)?;

src/backend/concentratord.rs

@@ -7,7 +7,7 @@ use std::time::{Duration, SystemTime};
 use anyhow::Result;
 use async_trait::async_trait;
 use chirpstack_api::gw;
-use log::{info, trace, warn};
+use log::{debug, info, trace, warn};
 use prost::Message;
 use tokio::net::UdpSocket;
 use tokio::sync::{Mutex, RwLock};
@@ -26,6 +26,10 @@ struct State {
     pull_addr: RwLock<Option<SocketAddr>>,
     stats: Mutex<Stats>,
     time_fallback_enabled: bool,
+    forward_crc_ok: bool,
+    forward_crc_invalid: bool,
+    forward_crc_missing: bool,
+    filters: lrwn_filters::Filters,
 }
 
 #[derive(Clone)]
@@ -238,6 +242,13 @@ impl Backend {
             pull_addr: RwLock::new(None),
             stats: Mutex::new(Stats::default()),
             time_fallback_enabled: conf.backend.semtech_udp.time_fallback_enabled,
+            forward_crc_invalid: conf.backend.filters.forward_crc_invalid,
+            forward_crc_missing: conf.backend.filters.forward_crc_missing,
+            forward_crc_ok: conf.backend.filters.forward_crc_ok,
+            filters: lrwn_filters::Filters {
+                dev_addr_prefixes: conf.backend.filters.dev_addr_prefixes.clone(),
+                join_eui_prefixes: conf.backend.filters.join_eui_prefixes.clone(),
+            },
         };
         let state = Arc::new(state);
 
@@ -348,8 +359,29 @@ async fn handle_push_data(state: &Arc<State>, data: &[u8], remote: &SocketAddr)
     let gateway_stats = pl.to_proto_gateway_stats()?;
 
     for uf in &uplink_frames {
-        state.count_uplink(uf).await?;
-        send_uplink_frame(uf).await?;
+        if let Some(rx_info) = &uf.rx_info {
+            if !((rx_info.crc_status() == gw::CrcStatus::CrcOk && state.forward_crc_ok)
+                || (rx_info.crc_status() == gw::CrcStatus::BadCrc && state.forward_crc_invalid)
+                || (rx_info.crc_status() == gw::CrcStatus::NoCrc && state.forward_crc_missing))
+            {
+                debug!(
+                    "Ignoring uplink frame because of forward_crc_ flags, uplink_id: {}",
+                    uf.rx_info.as_ref().map(|v| v.uplink_id).unwrap_or_default(),
+                );
+
+                continue;
+            }
+        }
+
+        if lrwn_filters::matches(&uf.phy_payload, &state.filters) {
+            state.count_uplink(uf).await?;
+            send_uplink_frame(uf).await?;
+        } else {
+            debug!(
+                "Ignoring uplink frame because of dev_addr and join_eui filters, uplink_id: {}",
+                uf.rx_info.as_ref().map(|v| v.uplink_id).unwrap_or_default()
+            );
+        }
     }
 
     if let Some(mut stats) = gateway_stats {

src/backend/semtech_udp/mod.rs

@@ -106,14 +106,47 @@ pub fn run(config: &Configuration) {
   #   * semtech_udp
   enabled="{{ backend.enabled }}"
 
-  # Forward CRC ok.
-  forward_crc_ok={{ backend.forward_crc_ok }}
+  # Filters.
+  [backend.filters]
 
-  # Forward CRC invalid.
-  forward_crc_invalid={{ backend.forward_crc_invalid }}
+    # Forward CRC ok.
+    forward_crc_ok={{ backend.filters.forward_crc_ok }}
 
-  # Forward CRC missing.
-  forward_crc_missing={{ backend.forward_crc_missing }}
+    # Forward CRC invalid.
+    forward_crc_invalid={{ backend.filters.forward_crc_invalid }}
+
+    # Forward CRC missing.
+    forward_crc_missing={{ backend.filters.forward_crc_missing }}
+
+    # 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=[
+      {{#each backend.filters.dev_addr_prefixes}}
+      "{{this}}",
+      {{/each}}
+    ]
+
+    # 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 backend.filters.join_eui_prefixes}}
+      "{{this}}",
+      {{/each}}
+    ]
 
 
   # ChirpStack Concentratord backend configuration.

src/cmd/configfile.rs

@@ -79,9 +79,7 @@ impl Default for Mqtt {
 #[serde(default)]
 pub struct Backend {
     pub enabled: String,
-    pub forward_crc_ok: bool,
-    pub forward_crc_invalid: bool,
-    pub forward_crc_missing: bool,
+    pub filters: Filters,
     pub gateway_id: String,
     pub semtech_udp: SemtechUdp,
     pub concentratord: Concentratord,
@@ -91,16 +89,36 @@ impl Default for Backend {
     fn default() -> Self {
         Backend {
             enabled: "semtech_udp".to_string(),
-            forward_crc_ok: true,
-            forward_crc_invalid: false,
-            forward_crc_missing: false,
+            filters: Filters::default(),
             gateway_id: "".into(),
             semtech_udp: SemtechUdp::default(),
             concentratord: Concentratord::default(),
         }
     }
 }
 
+#[derive(Serialize, Deserialize)]
+#[serde(default)]
+pub struct Filters {
+    pub forward_crc_ok: bool,
+    pub forward_crc_invalid: bool,
+    pub forward_crc_missing: bool,
+    pub dev_addr_prefixes: Vec<lrwn_filters::DevAddrPrefix>,
+    pub join_eui_prefixes: Vec<lrwn_filters::EuiPrefix>,
+}
+
+impl Default for Filters {
+    fn default() -> Self {
+        Filters {
+            forward_crc_ok: true,
+            forward_crc_invalid: false,
+            forward_crc_missing: false,
+            dev_addr_prefixes: vec![],
+            join_eui_prefixes: vec![],
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 #[serde(default)]
 pub struct Concentratord {

src/config.rs

@@ -9,7 +9,7 @@ pub fn setup(name: &str, level: log::Level, syslog: bool) -> Result<()> {
             facility: Facility::LOG_USER,
             hostname: None,
             process: name.to_string(),
-            pid: process::id() as u32,
+            pid: process::id(),
         };
         let logger = syslog::unix(formatter).map_err(|e| anyhow!("{}", e))?;
         log::set_boxed_logger(Box::new(BasicLogger::new(logger)))