update tcp api server

README.md

@@ -45,6 +45,8 @@ Options:
 - ✅ Handle zeromq data transfer
 - ✅ Code optimizations (WIP)
 - ✅ Error handing when fetching config
+- ❌ Local bound tcp api socket
+- ❌ ZMQ -> MQTT/Websocket ?
 
 
 ### Notes

src/api.rs

@@ -1,13 +1,13 @@
 use crate::config::Context;
 
+use serde_json;
 use std::io;
 use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpSocket;
-use tokio::sync::mpsc::Sender;
 use tokio::sync::Mutex;
 
-pub async fn apiserver(ctxarc: &Arc<Mutex<Context>>, apitx: Sender<String>) -> io::Result<()> {
+pub async fn apiserver(ctxarc: &Arc<Mutex<Context>>) -> io::Result<()> {
     let ctxclone = Arc::clone(ctxarc);
     let ctx = ctxclone.lock().await;
     let addr = ctx.cfg.api.parse().unwrap();
@@ -15,24 +15,41 @@ pub async fn apiserver(ctxarc: &Arc<Mutex<Context>>, apitx: Sender<String>) -> i
 
     let socket = TcpSocket::new_v4().unwrap();
     socket.bind(addr).unwrap();
+    socket.set_reuseaddr(true).unwrap();
     let listener = socket.listen(1024).unwrap();
 
     tokio::spawn(async move {
         loop {
+            //apitx.send(String::from("")).await.unwrap();
             match listener.accept().await {
-                Ok((mut socket, addr)) => {
+                Ok((stream, _addr)) => {
-                    let ctx = ctxclone.lock().await;
+                    //let mut buf = [0; 1024];
-                    let (reader, mut writer) = socket.split();
+                    let data;
-                    apitx.send(String::from("")).await.unwrap();
+                    {
-                    let msg = format!("{:?}", ctx.blocklist.len());
+                        let ctx = ctxclone.lock().await;
-                    writer
+                        data = serde_json::to_string(&ctx.blocklist);
-                        .write_all(format!("{msg}\r\n").as_bytes())
+                    }
-                        .await
+
-                        .unwrap();
+                    match data {
-                    writer.shutdown().await.unwrap();
+                        Ok(dt) => {
-                    socket.shutdown().await.unwrap();
+                            let (_reader, mut writer) = stream.into_split();
+                            match writer.write_all(format!("{dt}").as_bytes()).await {
+                                Ok(a) => {
+                                    println!("{a:?}");
+                                }
+                                Err(err) => {
+                                    println!("{err}");
+                                }
+                            }
+                        }
+                        Err(err) => {
+                            println!("unable to serialize data: {err}");
+                        }
+                    }
+                }
+                Err(err) => {
+                    println!("couldn't get client: {}", err)
                 }
-                Err(e) => println!("couldn't get client: {:?}", e),
             }
         }
     });

src/config.rs

@@ -89,7 +89,7 @@ impl Context {
                 }
                 Err(err) => {
                     println!("error loading config: {err}, retrying in {CONFIG_RETRY} secs");
-                    sleep_s(CONFIG_RETRY);
+                    sleep_s(CONFIG_RETRY).await;
                 }
             }
         }
@@ -207,6 +207,13 @@ impl Context {
         // nightly, future use
         //let drained: HashMap<String,IpData> = ctx.blocklist.drain_filter(|k,v| v.parse_date() < mindate)
         for (ip, blocked) in self.blocklist.clone().iter() {
+            match self.cfg.sets.get(&blocked.ipdata.src) {
+                Some(set) => {
+                    let mut block = self.blocklist.get_mut(ip).unwrap();
+                    block.blocktime = set.blocktime.clone();
+                }
+                None => {}
+            }
             let mindate = now - Duration::minutes(blocked.blocktime);
             if blocked.starttime < mindate {
                 self.blocklist.remove(&ip.clone()).unwrap();

src/fw.rs

@@ -1,15 +1,13 @@
 use crate::ip::IpData;
+use crate::ipblc::PKG_NAME;
 
 use nftnl::{nft_expr, Batch, Chain, FinalizedBatch, ProtoFamily, Rule, Table};
 use std::{ffi::CString, io::Error, net::Ipv4Addr};
 
-pub fn init(tablename: &String) -> (Batch, Table) {
+pub fn fwinit() -> (Batch, Table) {
     let mut batch = Batch::new();
 
-    let table = Table::new(
+    let table = Table::new(&CString::new(PKG_NAME).unwrap(), ProtoFamily::Ipv4);
-        &CString::new(tablename.as_str()).unwrap(),
-        ProtoFamily::Ipv4,
-    );
 
     batch.add(&table, nftnl::MsgType::Add);
     batch.add(&table, nftnl::MsgType::Del);
@@ -18,18 +16,17 @@ pub fn init(tablename: &String) -> (Batch, Table) {
     (batch, table)
 }
 
-pub fn block(
+pub fn fwblock(
-    tablename: &String,
     ips_add: &Vec<IpData>,
     ret: &mut Vec<String>,
     fwlen: &mut usize,
 ) -> std::result::Result<(), Error> {
     // convert chain
     let ips_add = convert(ips_add);
-    let (mut batch, table) = init(tablename);
+    let (mut batch, table) = fwinit();
 
     // build chain
-    let mut chain = Chain::new(&CString::new(tablename.as_str()).unwrap(), &table);
+    let mut chain = Chain::new(&CString::new(PKG_NAME).unwrap(), &table);
     chain.set_hook(nftnl::Hook::In, 1);
     chain.set_policy(nftnl::Policy::Accept);
 

src/ipblc.rs

@@ -1,6 +1,6 @@
 use crate::api::apiserver;
 use crate::config::{Context, GIT_VERSION};
-use crate::fw::{block, init};
+use crate::fw::{fwblock, fwinit};
 use crate::ip::{filter, IpData};
 use crate::utils::read_lines;
 use crate::ws::send_to_ipbl_ws;
@@ -15,36 +15,38 @@ use std::sync::Arc;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 use tokio::sync::Mutex;
 
+pub const PKG_NAME: &str = env!("CARGO_PKG_NAME");
 const BL_CHAN_SIZE: usize = 32;
 const ZMQ_CHAN_SIZE: usize = 64;
 const API_CHAN_SIZE: usize = 64;
 
 pub async fn run() {
-    let ctx = Arc::new(Mutex::new(Context::new().await));
+    let ctxarc = Arc::new(Mutex::new(Context::new().await));
-    println!(
-        "Launching {}, version {}",
-        env!("CARGO_PKG_NAME"),
-        format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION)
-    );
 
-    let (apitx, mut apirx): (Sender<String>, Receiver<String>) = channel(API_CHAN_SIZE);
+    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);
+
+    let (_apitx, mut apirx): (Sender<String>, Receiver<String>) = channel(API_CHAN_SIZE);
     //let tcpsocket = apiinit(&ctx, &apitx).await;
 
-    apiserver(&ctx, apitx).await.unwrap();
+    let ctxclone = Arc::clone(&ctxarc);
+    apiserver(&ctxclone).await.unwrap();
 
     // initialize the firewall table
-    init(&env!("CARGO_PKG_NAME").to_string());
+    fwinit();
     let mut fwlen: usize = 0;
 
     // initialize zeromq sockets
     let (ipdatatx, mut ipdatarx): (Sender<IpData>, Receiver<IpData>) = channel(ZMQ_CHAN_SIZE);
-    let zmqreqsocket = zmqinit(&ctx, &ipdatatx).await;
+    let zmqreqsocket = zmqinit(&ctxclone, &ipdatatx).await;
 
-    let mut blrx = watchfiles(&ctx).await;
+    let mut blrx = watchfiles(&ctxclone).await;
 
-    let ctxarc = Arc::clone(&ctx);
+    let ctxclone = Arc::clone(&ctxarc);
     tokio::spawn(async move {
-        compare_files_changes(&ctxarc, &mut blrx, &ipdatatx).await;
+        compare_files_changes(&ctxclone, &mut blrx, &ipdatatx).await;
     });
 
     let mut ip_init = IpData {
@@ -58,22 +60,19 @@ pub async fn run() {
 
     loop {
         let mut ret: Vec<String> = Vec::new();
-        let begin: DateTime<Local> = Local::now().trunc_subsecs(0);
 
         // wait for logs parse and zmq channel receive
         //let mut received_ip = ipdatarx.recv();
-        let r = ipdatarx.recv();
+        let ipdata_wait = ipdatarx.recv();
+        let apimsg_wait = apirx.recv();
 
-        let apimsg = apirx.recv();
+        let ctxclone = Arc::clone(&ctxarc);
-
-        let sl = tokio::time::sleep(tokio::time::Duration::from_millis(100));
 
         tokio::select! {
-            val = r => {
+            val = ipdata_wait => {
                 let mut received_ip = val.unwrap();
-                // lock the context mutex
+
-                let ctxarc = Arc::clone(&ctx);
+                let mut ctx = ctxclone.lock().await;
-                let mut ctx = ctxarc.lock().await;
 
                 if received_ip.ip == "".to_string() && received_ip.mode == "init".to_string() {
                     for ip_to_send in &mut ctx.get_blocklist_toblock().await {
@@ -85,62 +84,64 @@ pub async fn run() {
 
                 // refresh context blocklist
                 let filtered_ip = ctx.update_blocklist(&mut received_ip).await;
-                ctx.gc_blocklist().await;
 
                 // send ip list to ws and zmq sockets
                 if let Some(mut ip) = filtered_ip {
                     send_to_ipbl_ws(&ctx, &mut ip, &mut ret).await;
                     send_to_ipbl_zmq(&zmqreqsocket, &mut ip).await;
                 }
-
-                // apply firewall blocking
-                block(
-                    &env!("CARGO_PKG_NAME").to_string(),
-                    &ctx.get_blocklist_toblock().await,
-                    &mut ret,
-                    &mut fwlen,
-                )
-                .unwrap();
-
-                // log lines
-                if ret.len() > 0 {
-                    println!("{ret}", ret = ret.join(", "));
-                }
-
-                let end: DateTime<Local> = Local::now().trunc_subsecs(0);
-                if (end - begin) > Duration::seconds(5) {
-                    // reload configuration from the server
-                    match ctx.load().await {
-                        Ok(_) => {}
-                        Err(err) => {
-                            println!("error loading config: {err}");
-                        }
-                    }
-                };
             }
-            _val = apimsg => {
+            _val = apimsg_wait => {
-                continue;
-            }
-            _val = sl => {
-                continue;
             }
         };
+
+        let toblock;
+        {
+            let mut ctx = ctxarc.lock().await;
+            ctx.gc_blocklist().await;
+            toblock = ctx.get_blocklist_toblock().await;
+        }
+        // apply firewall blocking
+        fwblock(&toblock, &mut ret, &mut fwlen).unwrap();
+
+        // log lines
+        if ret.len() > 0 {
+            println!("{ret}", ret = ret.join(", "));
+        }
+
+        {
+            let now_cfg_reload = Local::now().trunc_subsecs(0);
+            if (now_cfg_reload - last_cfg_reload) > Duration::seconds(5) {
+                // reload configuration from the server
+                let mut ctx = ctxclone.lock().await;
+                match ctx.load().await {
+                    Ok(_) => {
+                        last_cfg_reload = Local::now().trunc_subsecs(0);
+                        drop(ctx);
+                    }
+                    Err(err) => {
+                        println!("error loading config: {err}");
+                    }
+                }
+            };
+        }
     }
 }
 
-async fn watchfiles(ctx: &Arc<Mutex<Context>>) -> Receiver<FileEvent> {
+async fn watchfiles(ctxarc: &Arc<Mutex<Context>>) -> Receiver<FileEvent> {
     let (bltx, blrx): (Sender<FileEvent>, Receiver<FileEvent>) = channel(BL_CHAN_SIZE);
-    let ctx = Arc::clone(ctx);
+    let ctxclone = Arc::clone(ctxarc);
     tokio::spawn(async move {
         loop {
-            let events: Vec<InotifyEvent>;
+            let events;
+            let instance;
             {
-                let c = ctx.lock().await;
+                let ctx = ctxclone.lock().await;
-                let instance = c.instance.clone();
+                instance = ctx.instance.clone();
-                drop(c);
-                events = instance.read_events().unwrap();
             }
 
+            events = instance.read_events().unwrap();
+
             for inevent in events {
                 let date: DateTime<Local> = Local::now().trunc_subsecs(0);
                 bltx.send(FileEvent { inevent, date }).await.unwrap();
@@ -163,7 +164,7 @@ async fn get_last_file_size(w: &mut HashMap<String, u64>, path: &str) -> (u64, b
 }
 
 async fn compare_files_changes(
-    ctx: &Arc<Mutex<Context>>,
+    ctxarc: &Arc<Mutex<Context>>,
     inrx: &mut Receiver<FileEvent>,
     ipdatatx: &Sender<IpData>,
 ) {
@@ -172,14 +173,20 @@ async fn compare_files_changes(
         let modfiles = inrx.recv().await.unwrap();
         let mut iplist: Vec<IpData> = vec![];
 
-        let mut ctx = ctx.lock().await;
+        let sask;
-        tnets = ctx.cfg.build_trustnets();
+        let sas;
+        {
+            let ctx = ctxarc.lock().await;
+            sas = ctx.clone().sas;
+            sask = sas.keys();
+            tnets = ctx.cfg.build_trustnets();
+        }
 
         match modfiles.inevent.name {
             Some(name) => {
                 let filename = name.to_str().unwrap();
-                for sak in &mut ctx.clone().sas.keys() {
+                for sak in sask {
-                    let sa = &mut ctx.sas.get_mut(sak).unwrap();
+                    let sa = sas.get(sak).unwrap();
                     if modfiles.inevent.wd == sa.wd {
                         let handle: String;
                         if sa.filename.as_str() == "" {
@@ -190,8 +197,13 @@ async fn compare_files_changes(
                             continue;
                         }
 
-                        let (filesize, sizechanged) =
+                        let (filesize, sizechanged);
-                            get_last_file_size(&mut sa.watchedfiles, &handle).await;
+                        {
+                            let mut ctx = ctxarc.lock().await;
+                            let sa = ctx.sas.get_mut(sak).unwrap();
+                            (filesize, sizechanged) =
+                                get_last_file_size(&mut sa.watchedfiles, &handle).await;
+                        }
 
                         if !sizechanged {
                             continue;

src/utils.rs

@@ -5,7 +5,7 @@ use std::boxed::Box;
 use std::fs::File;
 use std::io::*;
 use std::path::Path;
-use std::time::Duration;
+use tokio::time::{sleep, Duration};
 
 lazy_static! {
     static ref R_FILE_GZIP: Regex = Regex::new(r".*\.gz.*").unwrap();
@@ -33,12 +33,12 @@ pub fn _dedup<T: Ord + PartialOrd>(list: &mut Vec<T>) -> usize {
     list.len()
 }
 
-pub fn _sleep_ms(ms: u64) {
+pub async fn sleep_ms(ms: u64) {
-    std::thread::sleep(Duration::from_millis(ms));
+    sleep(Duration::from_millis(ms)).await;
 }
 
-pub fn sleep_s(s: u64) {
+pub async fn sleep_s(s: u64) {
-    std::thread::sleep(Duration::from_secs(s));
+    sleep(Duration::from_secs(s)).await;
 }
 
 pub fn gethostname(show_fqdn: bool) -> String {

src/ws.rs

@@ -11,7 +11,7 @@ pub async fn send_to_ipbl_ws(ctx: &Context, ip: &mut IpData, ret: &mut Vec<Strin
             }
             Err(err) => {
                 println!("{err}");
-                sleep_s(1);
+                sleep_s(1).await;
             }
         };
     }

src/zmqcom.rs

@@ -41,15 +41,17 @@ pub async fn zmqinit(ctx: &Arc<Mutex<Context>>, ipdatatx: &Sender<IpData>) -> zm
 }
 
 async fn listenpubsub(ctx: &Arc<Mutex<Context>>, txpubsub: Sender<IpData>, socket: zmq::Socket) {
-    let ctx = ctx.lock().await;
+    let prefix;
-    let prefix = format!(
+    {
-        "{sub} ",
+        let ctx = ctx.lock().await;
-        sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription
+        prefix = format!(
-    );
+            "{sub} ",
-    socket
+            sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription
-        .set_subscribe(ctx.cfg.zmq.get("pubsub").unwrap().subscription.as_bytes())
+        );
-        .expect("failed setting subscription");
+        socket
-    drop(ctx);
+            .set_subscribe(ctx.cfg.zmq.get("pubsub").unwrap().subscription.as_bytes())
+            .expect("failed setting subscription");
+    }
 
     tokio::spawn(async move {
         loop {