feat: bump 1.7.0

Reviewed-on: #14

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ipblc"
-version = "1.6.3"
+version = "1.7.0"
 edition = "2021"
 authors = ["PaulBSD <paul@paulbsd.com>"]
 description = "ipblc is a tool that search and send attacking ip addresses to ipbl"
@@ -21,8 +21,9 @@ regex = "1.10"
 reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "1.34", features = ["full", "sync"] }
+sd-notify = { version = "0.4" }
-tungstenite = { version = "0.20", features = ["handshake", "rustls-tls-native-roots"] }
+tokio = { version = "1.35", features = ["full", "sync"] }
+tungstenite = { version = "0.21", features = ["handshake", "rustls-tls-native-roots"] }
 
 ## to optimize binary size (slow compile time)
 #[profile.release]

src/config.rs

@@ -17,7 +17,7 @@ use std::path::Path;
 pub const GIT_VERSION: &str = git_version!(args = ["--always", "--dirty="]);
 const MASTERSERVER: &str = "ipbl.paulbsd.com";
 const WSSUBSCRIPTION: &str = "ipbl";
-const CONFIG_RETRY: u64 = 2;
+const CONFIG_RETRY_INTERVAL: u64 = 2;
 const WEB_CLIENT_TIMEOUT: i64 = 5;
 
 #[derive(Debug)]
@@ -103,12 +103,12 @@ impl Context {
             .send()
             .await;
         let req = match resp {
-            Ok(re) => re,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
         let data: Discovery = match req.json().await {
-            Ok(res) => res,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
         Ok(data)
     }
@@ -127,10 +127,10 @@ impl Context {
                     }
                     break;
                 }
-                Err(err) => {
+                Err(e) => {
-                    println!("error loading config: {err}, retrying in {CONFIG_RETRY}s");
+                    println!("error loading config: {e}, retrying in {CONFIG_RETRY_INTERVAL}s");
                     last_in_err = true;
-                    sleep_s(CONFIG_RETRY).await;
+                    sleep_s(CONFIG_RETRY_INTERVAL).await;
                 }
             };
         }
@@ -169,18 +169,19 @@ impl Context {
     }
 
     pub async fn update_blocklist(&mut self, ipevent: &IpEvent) -> Option<IpEvent> {
-        match self.cfg.sets.get(&ipevent.ipdata.src) {
+        match &ipevent.ipdata {
+            Some(ipdata) => match self.cfg.sets.get(&ipdata.src) {
                 Some(set) => {
-                let starttime = DateTime::parse_from_rfc3339(ipevent.ipdata.date.as_str())
+                    let starttime = DateTime::parse_from_rfc3339(ipdata.date.as_str())
                         .unwrap()
                         .with_timezone(&chrono::Local);
                     let blocktime = set.blocktime;
                     if ipevent.mode == "file".to_string() && gethostname(true) == ipevent.hostname {
-                    let block = self
+                        let block =
-                        .blocklist
+                            self.blocklist
-                        .entry(ipevent.ipdata.ip.to_string())
+                                .entry(ipdata.ip.to_string())
                                 .or_insert(BlockIpData {
-                            ipdata: ipevent.ipdata.clone(),
+                                    ipdata: ipdata.clone(),
                                     tryfail: 0,
                                     starttime,
                                     blocktime,
@@ -192,9 +193,9 @@ impl Context {
                         }
                     } else {
                         self.blocklist
-                        .entry(ipevent.ipdata.ip.to_string())
+                            .entry(ipdata.ip.to_string())
                             .or_insert(BlockIpData {
-                            ipdata: ipevent.ipdata.clone(),
+                                ipdata: ipdata.clone(),
                                 tryfail: set.tryfail,
                                 starttime,
                                 blocktime,
@@ -202,6 +203,8 @@ impl Context {
                     }
                 }
                 None => {}
+            },
+            None => {}
         }
         None
     }
@@ -362,12 +365,12 @@ impl Config {
             .send()
             .await;
         let req = match resp {
-            Ok(re) => re,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
         let data: GlobalConfigV2 = match req.json::<GlobalConfigV2>().await {
-            Ok(res) => res,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
 
         for d in data.sets {
@@ -397,13 +400,13 @@ impl Config {
             .await;
 
         let req = match resp {
-            Ok(re) => re,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
 
         let data: Vec<IpData> = match req.json::<Vec<IpData>>().await {
-            Ok(res) => res,
+            Ok(o) => o,
-            Err(err) => return Err(err),
+            Err(e) => return Err(e),
         };
 
         Ok(data)
@@ -414,8 +417,8 @@ impl Config {
         for trustnet in &self.trustnets {
             match trustnet.parse() {
                 Ok(net) => trustnets.push(net),
-                Err(err) => {
+                Err(e) => {
-                    println!("error parsing {trustnet}, error: {err}");
+                    println!("error parsing {trustnet}, error: {e}");
                 }
             };
         }
@@ -427,13 +430,7 @@ impl Config {
             msgtype: String::from("bootstrap"),
             mode: String::from("ws"),
             hostname: gethostname(true),
-            ipdata: IpData {
+            ipdata: None,
-                t: 4,
-                ip: "".to_string(),
-                src: "".to_string(),
-                date: "".to_string(),
-                hostname: "".to_string(),
-            },
         }
     }
 }
@@ -521,13 +518,13 @@ mod test {
                 msgtype: String::from("add"),
                 mode: String::from("ws"),
                 hostname: String::from("localhost"),
-                ipdata: IpData {
+                ipdata: Some(IpData {
                     t: 4,
                     ip: "1.1.1.1".to_string(),
                     hostname: "test1".to_string(),
                     date: now.to_rfc3339().to_string(),
                     src: "ssh".to_string(),
-                },
+                }),
             })
             .await;
         }
@@ -537,13 +534,13 @@ mod test {
                 msgtype: String::from("add"),
                 mode: String::from("ws"),
                 hostname: String::from("localhost"),
-                ipdata: IpData {
+                ipdata: Some(IpData {
                     t: 4,
                     ip: "1.1.1.2".to_string(),
                     hostname: "test2".to_string(),
                     date: now.to_rfc3339().to_string(),
                     src: "http".to_string(),
-                },
+                }),
             })
             .await;
         }
@@ -552,13 +549,13 @@ mod test {
             msgtype: String::from("add"),
             mode: String::from("ws"),
             hostname: String::from("localhost"),
-            ipdata: IpData {
+            ipdata: Some(IpData {
                 t: 4,
                 ip: "1.1.1.3".to_string(),
                 hostname: "testgood".to_string(),
                 date: now.to_rfc3339().to_string(),
                 src: "http".to_string(),
-            },
+            }),
         })
         .await;
 
@@ -566,13 +563,13 @@ mod test {
             msgtype: String::from("add"),
             mode: String::from("ws"),
             hostname: String::from("localhost"),
-            ipdata: IpData {
+            ipdata: Some(IpData {
                 t: 4,
                 ip: "1.1.1.4".to_string(),
                 hostname: "testgood".to_string(),
                 date: now.to_rfc3339().to_string(),
                 src: "http".to_string(),
-            },
+            }),
         })
         .await;
 
@@ -580,26 +577,26 @@ mod test {
             msgtype: String::from("add"),
             mode: String::from("ws"),
             hostname: String::from("localhost"),
-            ipdata: IpData {
+            ipdata: Some(IpData {
                 t: 4,
                 ip: "1.1.1.4".to_string(),
                 hostname: "testgood".to_string(),
                 date: now.to_rfc3339().to_string(),
                 src: "http".to_string(),
-            },
+            }),
         })
         .await;
         ctx.update_blocklist(&mut IpEvent {
             msgtype: String::from("add"),
             mode: String::from("ws"),
             hostname: String::from("localhost"),
-            ipdata: IpData {
+            ipdata: Some(IpData {
                 t: 6,
                 ip: "2a00:1450:4007:805::2003".to_string(),
                 hostname: "testgood".to_string(),
                 date: now.to_rfc3339().to_string(),
                 src: "http".to_string(),
-            },
+            }),
         })
         .await;
 

src/fw.rs

@@ -21,7 +21,6 @@ pub fn fwglobalinit<'a>() -> ((Batch, Table), (Batch, Table)) {
 
 macro_rules! initrules {
     ($batch:expr, $table:expr, $chain:ident) => {
-        let mut $chain = Chain::new(&CString::new(PKG_NAME).unwrap(), &$table);
         $chain.set_hook(nftnl::Hook::In, 1);
         $chain.set_policy(nftnl::Policy::Accept);
 
@@ -29,13 +28,11 @@ macro_rules! initrules {
         $batch.add(&Rule::new(&$chain), nftnl::MsgType::Del);
 
         let mut rule = Rule::new(&$chain);
-
         rule.add_expr(&nft_expr!(ct state));
         rule.add_expr(&nft_expr!(bitwise mask 4u32, xor 0u32));
         rule.add_expr(&nft_expr!(cmp != 0u32));
         rule.add_expr(&nft_expr!(counter));
         rule.add_expr(&nft_expr!(verdict accept));
-
         $batch.add(&rule, nftnl::MsgType::Add);
     };
 }
@@ -86,17 +83,33 @@ fn fwinit(t: FwTableType) -> (Batch, Table) {
 }
 
 pub fn fwblock(
-    ips_add: &Vec<IpData>,
+    ips_add_all: &Vec<IpData>,
     ret: &mut Vec<String>,
     fwlen: &mut usize,
 ) -> std::result::Result<(), Error> {
     let ((mut batch4, table4), (mut batch6, table6)) = fwglobalinit();
 
+    let mut chain4 = Chain::new(&CString::new(PKG_NAME).unwrap(), &table4);
+    let mut chain6 = Chain::new(&CString::new(PKG_NAME).unwrap(), &table6);
+
     initrules!(batch4, table4, chain4);
     initrules!(batch6, table6, chain6);
 
+    let mut factor = 1;
+    if ips_add_all.len() > 100 {
+        factor = (ips_add_all.len() / 10) as usize
+    }
+
+    let ips_add_tmp: Vec<IpData> = ips_add_all.clone().iter().map(|x| x.clone()).collect();
+    let mut ips_add_iter = ips_add_tmp.chunks(factor);
+    let mut ips_add: Vec<&[IpData]> = vec![];
+    while let Some(x) = ips_add_iter.next() {
+        ips_add.push(x);
+    }
+
     // build and add rules
-    for ipdata in ips_add.clone() {
+    for ipdata_group in ips_add.clone() {
+        for ipdata in ipdata_group {
             match ipdata.t {
                 4 => {
                     createrules!(ipdata, chain4, batch4, Ipv4Addr, ipv4);
@@ -107,16 +120,25 @@ pub fn fwblock(
                 _ => {}
             }
         }
+    }
 
     // validate and send batch
     for b in [batch4, batch6] {
         let bf = b.finalize();
-        send_and_process(&bf).unwrap();
+        match send_and_process(&bf) {
+            Ok(_) => {}
+            Err(e) => {
+                println!("error sending batch: {e}");
             }
-    if fwlen != &mut ips_add.len() {
+        };
-        ret.push(format!("{length} ip in firewall", length = ips_add.len()));
     }
-    *fwlen = ips_add.len();
+    if fwlen != &mut ips_add_all.len() {
+        ret.push(format!(
+            "{length} ip in firewall",
+            length = ips_add_all.len()
+        ));
+    }
+    *fwlen = ips_add_all.len();
     Ok(())
 }
 

src/ip.rs

@@ -22,7 +22,7 @@ pub struct IpEvent {
     pub msgtype: String,
     pub mode: String,
     pub hostname: String,
-    pub ipdata: IpData,
+    pub ipdata: Option<IpData>,
 }
 
 #[macro_export]
@@ -35,6 +35,14 @@ macro_rules! ipevent {
             ipdata: $ipdata,
         }
     };
+    ($msgtype:expr,$mode:expr,$hostname:expr) => {
+        IpEvent {
+            msgtype: String::from($msgtype),
+            mode: String::from($mode),
+            hostname: $hostname,
+            ipdata: None,
+        }
+    };
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]

src/ipblc.rs

@@ -11,6 +11,7 @@ use chrono::prelude::*;
 use chrono::prelude::{DateTime, Local};
 use chrono::Duration;
 use nix::sys::inotify::{InitFlags, Inotify, InotifyEvent};
+use sd_notify::*;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
@@ -21,6 +22,13 @@ const BL_CHAN_SIZE: usize = 32;
 const WS_CHAN_SIZE: usize = 64;
 const LOOP_MAX_WAIT: u64 = 5;
 
+macro_rules! log_with_systemd {
+    ($msg:expr) => {
+        println!("{}", $msg);
+        notify(false, &[NotifyState::Status(format!("{}", $msg).as_str())]).unwrap();
+    };
+}
+
 pub async fn run() {
     let inotify = Inotify::init(InitFlags::empty()).unwrap();
     let globalctx = Context::new(&inotify).await;
@@ -31,7 +39,7 @@ pub async fn run() {
     let pkgversion = format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION);
 
     let mut last_cfg_reload: DateTime<Local> = Local::now().trunc_subsecs(0);
-    println!("Launching {}, version {}", PKG_NAME, pkgversion);
+    log_with_systemd!(format!("Launching {}, version {}", PKG_NAME, pkgversion));
     fwglobalinit();
 
     let ctxapi = Arc::clone(&ctxarc);
@@ -60,6 +68,8 @@ pub async fn run() {
         compare_files_changes(&ctxclone, &mut blrx, &ipeventclone).await;
     });
 
+    notify(false, &[NotifyState::Ready]).unwrap();
+
     loop {
         let mut ret: Vec<String> = Vec::new();
 
@@ -76,8 +86,9 @@ pub async fn run() {
 
                 if received_ip.msgtype == "bootstrap".to_string() {
                     for ip_to_send in toblock {
-                        let ipe = ipevent!("init","ws",gethostname(true),ip_to_send);
+                        let ipe = ipevent!("init","ws",gethostname(true),Some(ip_to_send));
                         if !send_to_ipbl_websocket(&mut wssocketrr, &ipe).await {
+                            wssocketrr.close(None).unwrap();
                             wssocketrr = websocketreqrep(&ctxwsrr).await;
                             break;
                         }
@@ -93,18 +104,24 @@ pub async fn run() {
                 // send ip list to api and ws sockets
                 if let Some(ipevent) = filtered_ipevent {
                     if received_ip.msgtype != "init" {
-                        println!("sending {} to api and ws", ipevent.ipdata.ip);
+                        log_with_systemd!(format!("sending {} to api and ws", ipevent.ipdata.clone().unwrap().ip));
                         let ipe = ipevent!("add","ws",gethostname(true),ipevent.ipdata);
                         send_to_ipbl_api(&server.clone(), &ipe).await;
-                        let status = send_to_ipbl_websocket(&mut wssocketrr, &ipe).await;
+                        if !send_to_ipbl_websocket(&mut wssocketrr, &ipe).await {
-                        if !status {
+                            wssocketrr.close(None).unwrap();
                             wssocketrr = websocketreqrep(&ctxwsrr).await;
                             continue;
                         }
                     }
                 }
             }
-            _val = sleep_s(LOOP_MAX_WAIT) => {}
+            _val = sleep_s(LOOP_MAX_WAIT) => {
+                let ipe = ipevent!("ping", "ws", gethostname(true));
+                if !send_to_ipbl_websocket(&mut wssocketrr, &ipe).await {
+                    wssocketrr.close(None).unwrap();
+                    wssocketrr = websocketreqrep(&ctxwsrr).await;
+                }
+            }
         };
 
         let ctxclone = Arc::clone(&ctxarc);
@@ -112,7 +129,8 @@ pub async fn run() {
 
         // log lines
         if ret.len() > 0 {
-            println!("{ret}", ret = ret.join(", "));
+            let result = ret.join(", ");
+            log_with_systemd!(format!("{result}"));
         }
 
         let ctxclone = Arc::clone(&ctxarc);
@@ -128,8 +146,26 @@ async fn handle_cfg_reload(
 ) {
     let now_cfg_reload = Local::now().trunc_subsecs(0);
     if (now_cfg_reload - *last_cfg_reload) > Duration::seconds(LOOP_MAX_WAIT as i64) {
-        let inotify = inoarc.read().await;
+        let inotify;
-        match ctxclone.write().await.load(&inotify).await {
+        loop {
+            inotify = match inoarc.try_read() {
+                Ok(o) => o,
+                Err(e) => {
+                    println!("{e}");
+                    sleep_s(1).await;
+                    continue;
+                }
+            };
+            break;
+        }
+        let mut ctxtest = match ctxclone.try_write() {
+            Ok(o) => o,
+            Err(e) => {
+                println!("{e}");
+                return;
+            }
+        };
+        match ctxtest.load(&inotify).await {
             Ok(_) => {
                 *last_cfg_reload = Local::now().trunc_subsecs(0);
             }
@@ -153,8 +189,8 @@ async fn handle_fwblock(ctxclone: Arc<RwLock<Context>>, ret: &mut Vec<String>, f
     // apply firewall blocking
     match fwblock(&toblock, ret, fwlen) {
         Ok(_) => {}
-        Err(err) => {
+        Err(e) => {
-            println!("Err: {err}, unable to push firewall rules, use super user")
+            println!("err: {e}, unable to push firewall rules, use super user")
         }
     };
 }
@@ -205,8 +241,7 @@ async fn compare_files_changes(
         match modfiles.inevent.name {
             Some(name) => {
                 let filename = name.to_str().unwrap();
-                for sak in sas.clone().keys() {
+                for (sak, sa) in sas.clone().iter_mut() {
-                    let sa = sas.get(sak).unwrap();
                     if modfiles.inevent.wd == sa.wd {
                         let handle: String;
                         if sa.filename.as_str() == "" {
@@ -244,7 +279,7 @@ async fn compare_files_changes(
                     }
                 }
                 for ip in iplist {
-                    let ipe = ipevent!("add", "file", gethostname(true), ip);
+                    let ipe = ipevent!("add", "file", gethostname(true), Some(ip));
                     let ipetx = ipeventtx.read().await;
                     ipetx.send(ipe).await.unwrap();
                 }

src/monitoring.rs

@@ -10,7 +10,13 @@ use tokio::sync::RwLock;
 pub async fn apiserver(ctxarc: &Arc<RwLock<Context>>) -> io::Result<()> {
     let ctxarc = ctxarc.clone();
     let addr: String = { ctxarc.read().await.cfg.api.parse().unwrap() };
-    let listener = TcpListener::bind(addr).await.unwrap();
+    let listener = match TcpListener::bind(addr).await {
+        Ok(o) => o,
+        Err(e) => {
+            println!("error: {e}");
+            std::process::exit(1);
+        }
+    };
 
     tokio::spawn(async move {
         loop {
@@ -23,13 +29,13 @@ pub async fn apiserver(ctxarc: &Arc<RwLock<Context>>) -> io::Result<()> {
                             match socket.try_read(&mut buf) {
                                 Ok(_) => {}
                                 Err(e) => {
-                                    println!("{e}");
+                                    println!("error: {e}");
                                     continue;
                                 }
                             };
                         }
                         Err(e) => {
-                            println!("{e}");
+                            println!("error: {e}");
                             continue;
                         }
                     }
@@ -43,8 +49,8 @@ pub async fn apiserver(ctxarc: &Arc<RwLock<Context>>) -> io::Result<()> {
 
                     match socket.write_all(res.as_bytes()).await {
                         Ok(_) => {}
-                        Err(err) => {
+                        Err(e) => {
-                            println!("ee {err}");
+                            println!("error: {e}");
                         }
                     }
                 }

src/utils.rs

@@ -6,9 +6,9 @@ use tokio::time::{sleep, Duration};
 
 pub fn read_lines(filename: &String, offset: u64) -> Option<Box<dyn Read>> {
     let mut file = match File::open(filename) {
-        Ok(f) => f,
+        Ok(o) => o,
-        Err(err) => {
+        Err(e) => {
-            println!("{err}");
+            println!("error: {e}");
             return None;
         }
     };
@@ -21,6 +21,11 @@ pub async fn sleep_s(s: u64) {
     sleep(Duration::from_secs(s)).await;
 }
 
+#[allow(dead_code)]
+pub async fn sleep_ms(m: u64) {
+    sleep(Duration::from_millis(m)).await;
+}
+
 pub fn gethostname(show_fqdn: bool) -> String {
     let hostname_cstr = unistd::gethostname().expect("Failed getting hostname");
     let fqdn = hostname_cstr

src/webservice.rs

@@ -11,12 +11,12 @@ pub async fn send_to_ipbl_api(server: &str, ip: &IpEvent) {
     let mut try_req = 0;
     let client = httpclient();
     loop {
-        match push_ip(&client, &server, &ip.ipdata).await {
+        match push_ip(&client, &server, &ip.ipdata.clone().unwrap()).await {
             Ok(_) => {
                 break;
             }
-            Err(err) => {
+            Err(e) => {
-                println!("{err}");
+                println!("error: {e}");
                 sleep_s(1).await;
                 if try_req == MAX_FAILED_API_RATE {
                     break;

src/websocket.rs

@@ -14,13 +14,13 @@ use tungstenite::*;
 pub async fn websocketreqrep(
     ctxarc: &Arc<RwLock<Context>>,
 ) -> WebSocket<MaybeTlsStream<TcpStream>> {
-    let (mut wssocketrr, bootstrap_event, cfg);
+    let (mut wssocketrr, bootstrap_event, wscfg);
     {
         let ctx = ctxarc.read().await;
         bootstrap_event = ctx.cfg.bootstrap_event().clone();
-        cfg = ctx.cfg.ws.get("reqrep").unwrap().clone();
+        wscfg = ctx.cfg.ws.get("reqrep").unwrap().clone();
     }
-    wssocketrr = websocketconnect(&cfg, &gethostname(true)).await.unwrap();
+    wssocketrr = websocketconnect(&wscfg, &gethostname(true)).await.unwrap();
     send_to_ipbl_websocket(&mut wssocketrr, &bootstrap_event).await;
 
     return wssocketrr;
@@ -45,17 +45,26 @@ pub async fn websocketpubsub(
                 Ok(msg) => {
                     let tosend: IpEvent = match serde_json::from_str(msg.to_string().as_str()) {
                         Ok(o) => o,
-                        Err(_e) => {
+                        Err(e) => {
+                            println!("error in pubsub: {e:?}");
                             continue;
                         }
                     };
-                    if tosend.ipdata.hostname != gethostname(true)
+                    match tosend.ipdata.clone() {
+                        Some(o) => {
+                            if o.hostname != gethostname(true)
                                 || tosend.msgtype == "init".to_string()
                             {
                                 let txps = txpubsub.read().await;
                                 txps.send(tosend).await.unwrap();
                             }
                         }
+                        None => {
+                            let txps = txpubsub.read().await;
+                            txps.send(tosend.clone()).await.unwrap();
+                        }
+                    }
+                }
                 Err(e) => {
                     println!("error in pubsub: {e:?}");
                     ws.close(None).unwrap();
@@ -105,11 +114,12 @@ pub async fn send_to_ipbl_websocket(
             Ok(_) => {}
             Err(e) => {
                 println!("err send read: {e:?}");
-                return handle_websocket_error(ws);
+                return false;
             }
         };
     } else {
-        return handle_websocket_error(ws);
+        println!("can't write to socket");
+        return false;
     };
 
     if ws.can_read() {
@@ -117,16 +127,13 @@ pub async fn send_to_ipbl_websocket(
             Ok(_) => {}
             Err(e) => {
                 println!("err send read: {e:?}");
-                return handle_websocket_error(ws);
+                return false;
             }
         };
     } else {
-        return handle_websocket_error(ws);
+        println!("can't read from socket");
+        return false;
     };
+
     true
 }
-
-fn handle_websocket_error(ws: &mut WebSocket<MaybeTlsStream<TcpStream>>) -> bool {
-    ws.close(None).unwrap();
-    return false;
-}