Merge branch 'develop-websocket'

.drone.yml

@@ -12,7 +12,7 @@ steps:
     image: rust:1
     commands:
       - apt-get update -y
-      - apt-get install -y libzmq3-dev libnftnl-dev libmnl-dev
+      - apt-get install -y libnftnl-dev libmnl-dev
       - cargo build --verbose --all
       - cargo test --verbose --all
     volumes:
@@ -28,7 +28,7 @@ steps:
     image: rust:1
     commands:
       - apt-get update -y
-      - apt-get install -y libzmq3-dev libnftnl-dev libmnl-dev
+      - apt-get install -y libnftnl-dev libmnl-dev
       - cargo build --release --verbose --all
       - cd target/release
       - tar -czvf ipblc-${DRONE_TAG}-${DRONE_STAGE_OS}-${DRONE_STAGE_ARCH}.tar.gz ipblc
@@ -77,7 +77,7 @@ steps:
     image: rust:1
     commands:
       - apt-get update -y
-      - apt-get install -y libzmq3-dev libnftnl-dev libmnl-dev
+      - apt-get install -y libnftnl-dev libmnl-dev
       - cargo build --verbose --all
       - cargo test --verbose --all
     volumes:
@@ -93,7 +93,7 @@ steps:
     image: rust:1
     commands:
       - apt-get update -y
-      - apt-get install -y libzmq3-dev libnftnl-dev libmnl-dev
+      - apt-get install -y libnftnl-dev libmnl-dev
       - cargo build --release --verbose --all
       - cd target/release
       - tar -czvf ipblc-${DRONE_TAG}-${DRONE_STAGE_OS}-${DRONE_STAGE_ARCH}.tar.gz ipblc

Cargo.toml

@@ -10,32 +10,22 @@ repository = "https://git.paulbsd.com/paulbsd/ipblc"
 
 [dependencies]
 chrono = { version = "0.4", features = ["serde"] }
-clap = { version = "4.0", features = ["string"] }
-git-version = "0.3.5"
+clap = { version = "4.2", features = ["string"] }
+git-version = "0.3"
 ipnet = "2.7"
 lazy_static = "1.4"
 mnl = "0.2"
 nftnl = "0.6"
 nix = "0.26"
-regex = "1.7"
+regex = "1.8"
 reqwest = { version = "0.11", default-features = false, features = ["json","rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "1.23", features = ["full"] }
-zmq = "0.10"
+tokio = { version = "1.28", features = ["full", "sync"] }
+tungstenite = { version = "0.19", features = ["handshake","rustls-tls-native-roots"] }
 
-# [target.aarch64-unknown-linux-gnu.dependencies]
-# flate2 = { version = "1.0", features = ["cloudflare_zlib"] }
-
-# [target.aarch64-linux-android.dependencies]
-# flate2 = { version = "1.0", features = ["zlib"] }
-
-# [target.armv7-unknown-linux-gnueabihf.dependencies]
-# flate2 = { version = "1.0", features = ["zlib"] }
-
-# [target.x86_64-unknown-linux-gnu.dependencies]
-# flate2 = { version = "1.0", features = ["cloudflare_zlib"] }
-
-[profile.release]
-debug = false
-opt-level = 3
+## to optimize binary size (slow compile time)
+#[profile.release]
+#strip = true
+#lto = true
+#opt-level = "z"

NOTES.md

@@ -0,0 +1,10 @@
+# Notes
+
+### Date formats
+
+```
+nginx:    2006-01-02T15:04:05+01:00
+ssh:      2006-01-02T15:04:05.000000+01:00
+openvpn:  2006-01-02 15:04:05
+mail:     2006-01-02T15:04:05.000000+01:00
+```

README.md

@@ -26,14 +26,15 @@ cargo build --release
 ### Usage
 
 ```
-USAGE:
-    ipblc [OPTIONS]
+ipblc is a tool that search and send attacking ip addresses to ipbl
 
-OPTIONS:
-    -d                           Enable debugging
-    -h, --help                   Print help information
-    -s, --server <server>        Sets a ipbl server [default: https://ipbl.paulbsd.com]
-    -V, --version                Print version informatio
+Usage: ipblc [OPTIONS]
+
+Options:
+  -s, --server <server>  Sets a http server [default: https://ipbl.paulbsd.com]
+  -d                     Enable debugging
+  -h, --help             Print help information
+  -V, --version          Print version information
 ```
 
 ### TODO
@@ -44,20 +45,19 @@ OPTIONS:
 - ✅ Handle zeromq data transfer
 - ✅ Code optimizations (WIP)
 - ✅ Error handing when fetching config
+- ✅ Local bound tcp api socket
+- ✅ ZMQ -> Websocket
+- ✅ Bug in RwLocks (agent often give up)
 
-### Date formats
 
-```
-nginx:    2006-01-02T15:04:05+01:00
-ssh:      2006-01-02T15:04:05.000000+01:00
-openvpn:  2006-01-02 15:04:05
-mail:     2006-01-02T15:04:05.000000+01:00
-```
+### Notes
+
+See [here](NOTES.md)
 
 ## License
 
 ```text
-Copyright (c) 2021, 2022 PaulBSD
+Copyright (c) 2021, 2022, 2023 PaulBSD
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without

old/zmqcom.rs

@@ -0,0 +1,106 @@
+use crate::config::{Context, ZMQCfg};
+use crate::ip::IpEvent;
+use crate::utils::gethostname;
+
+use std::sync::Arc;
+use tokio::sync::mpsc::Sender;
+use tokio::sync::RwLock;
+
+const ZMQPROTO: &str = "tcp";
+
+pub async fn zconnect<'a>(
+    zmqcfg: &ZMQCfg,
+    zmqtype: zmq::SocketType,
+) -> Result<zmq::Socket, zmq::Error> {
+    let zctx = zmq::Context::new();
+    let zmqhost = &zmqcfg.hostname;
+    let zmqport = &zmqcfg.port;
+    let socket = zctx.socket(zmqtype).unwrap();
+    let connectstring = format!("{ZMQPROTO}://{zmqhost}:{zmqport}");
+    socket.connect(&connectstring.as_str())?;
+    Ok(socket)
+}
+
+pub async fn zmqinit(ctx: &Arc<RwLock<Context>>, ipeventtx: &Sender<IpEvent>) -> zmq::Socket {
+    let ctxarc = Arc::clone(&ctx);
+
+    let zmqreqsocket;
+    let zmqsubsocket;
+    {
+        let zmqctx = ctxarc.read().await;
+        zmqreqsocket = zconnect(&zmqctx.cfg.zmq.get("reqrep").unwrap(), zmq::REQ)
+            .await
+            .unwrap();
+        zmqsubsocket = zconnect(&zmqctx.cfg.zmq.get("pubsub").unwrap(), zmq::SUB)
+            .await
+            .unwrap();
+    }
+
+    listenpubsub(&ctx, ipeventtx.clone(), zmqsubsocket).await;
+    return zmqreqsocket;
+}
+
+async fn listenpubsub(ctx: &Arc<RwLock<Context>>, txpubsub: Sender<IpEvent>, socket: zmq::Socket) {
+    let prefix;
+    {
+        let ctx = ctx.read().await;
+        prefix = format!(
+            "{sub} ",
+            sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription
+        );
+        socket
+            .set_subscribe(ctx.cfg.zmq.get("pubsub").unwrap().subscription.as_bytes())
+            .expect("failed setting subscription");
+    }
+
+    tokio::spawn(async move {
+        loop {
+            let msgs: Option<String> = match socket.recv_string(0) {
+                Ok(s) => match s {
+                    Ok(ss) => Some(ss),
+                    Err(e) => {
+                        println!("{e:?}");
+                        None
+                    }
+                },
+                Err(e) => {
+                    println!("{e:?}");
+                    None
+                }
+            };
+            match msgs {
+                Some(ss) => {
+                    let msg = ss.strip_prefix(prefix.as_str()).unwrap();
+                    let tosend: IpEvent = serde_json::from_str(msg).unwrap();
+                    if tosend.ipdata.hostname != gethostname(true)
+                        || tosend.msgtype == "init".to_string()
+                    {
+                        txpubsub.send(tosend).await.unwrap();
+                    }
+                }
+                None => {}
+            };
+        }
+    });
+}
+
+pub async fn send_to_ipbl_zmq(reqsocket: &zmq::Socket, ip: &IpEvent, _ret: &mut Vec<String>) {
+    let msg = format!("{val}", val = serde_json::to_string(&ip).unwrap());
+    match reqsocket.send(&msg, 0) {
+        Ok(_) => {}
+        Err(e) => {
+            println!("{e:?}")
+        }
+    };
+    match reqsocket.recv_string(0) {
+        Ok(o) => match o {
+            Ok(_) => {}
+            Err(ee) => {
+                println!("{ee:?}")
+            }
+        },
+        Err(e) => {
+            println!("{e:?}")
+        }
+    };
+}

scripts/upgrade.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+git pull
+cargo b --release
+sudo systemctl stop ipblc
+sudo cp target/release/ipblc /usr/local/apps/ipblc/ipblc
+sudo systemctl start ipblc

src/config.rs

@@ -1,4 +1,4 @@
-use crate::ip::{BlockIpData, IpData};
+use crate::ip::{BlockIpData, IpData, IpEvent};
 use crate::utils::{gethostname, sleep_s};
 
 use chrono::prelude::*;
@@ -16,17 +16,16 @@ use std::path::Path;
 
 pub const GIT_VERSION: &str = git_version!();
 const MASTERSERVER: &str = "ipbl.paulbsd.com";
-const ZMQSUBSCRIPTION: &str = "ipbl";
-const CONFIG_RETRY: u64 = 10;
+const WSSUBSCRIPTION: &str = "ipbl";
+const CONFIG_RETRY: u64 = 1;
+const WEB_CLIENT_TIMEOUT: i64 = 5;
 
 #[derive(Debug, Clone)]
 pub struct Context {
     pub blocklist: HashMap<String, BlockIpData>,
     pub cfg: Config,
-    pub client: Client,
     pub discovery: Discovery,
     pub flags: Flags,
-    pub hostname: String,
     pub instance: Box<Inotify>,
     pub sas: HashMap<String, SetMap>,
     pub hashwd: HashMap<String, WatchDescriptor>,
@@ -49,10 +48,9 @@ pub struct Flags {
 }
 
 impl Context {
-    pub async fn new<'a>() -> Self {
+    pub async fn new() -> Self {
         // Get flags
         let argp: ArgMatches = Context::argparse();
-        //let debug: bool = argp.contains_id("debug");
         let debug: bool = argp.get_one::<bool>("debug").unwrap().to_owned();
         let server: String = argp.get_one::<String>("server").unwrap().to_string();
 
@@ -60,43 +58,23 @@ impl Context {
         let mut ctx = Context {
             cfg: Config::new(),
             flags: Flags { debug, server },
-            hostname: gethostname(true),
             discovery: Discovery {
                 version: "1.0".to_string(),
                 urls: HashMap::new(),
             },
-            client: Client::builder()
-                .user_agent(format!(
-                    "{}/{}@{}/{}",
-                    env!("CARGO_PKG_NAME"),
-                    env!("CARGO_PKG_VERSION"),
-                    GIT_VERSION,
-                    gethostname(false)
-                ))
-                .build()
-                .unwrap(),
             sas: HashMap::new(),
             instance: Box::new(Inotify::init(InitFlags::empty()).unwrap()),
             blocklist: HashMap::new(),
             hashwd: HashMap::new(),
         };
 
-        loop {
-            print!("Loading config ... ");
-            match ctx.load().await {
-                Ok(_) => {
-                    break;
-                }
-                Err(err) => {
-                    println!("error loading config: {err}, retrying in {CONFIG_RETRY} secs");
-                    sleep_s(CONFIG_RETRY);
-                }
-            }
-        }
+        print!("Loading config ... ");
+        ctx.load().await.unwrap();
+
         ctx
     }
 
-    pub fn argparse<'a>() -> ArgMatches {
+    pub fn argparse() -> ArgMatches {
         Command::new(env!("CARGO_PKG_NAME"))
             .version(format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION))
             .author(env!("CARGO_PKG_AUTHORS"))
@@ -118,9 +96,9 @@ impl Context {
             .get_matches()
     }
 
+    #[allow(dead_code)]
     pub async fn discovery(&self) -> Result<Discovery, ReqError> {
-        let resp: Result<Response, ReqError> = self
-            .client
+        let resp: Result<Response, ReqError> = httpclient()
             .get(format!("{server}/discovery", server = self.flags.server))
             .send()
             .await;
@@ -139,9 +117,30 @@ impl Context {
         if cfg!(test) {
             return Ok(());
         }
-        self.discovery = self.discovery().await?;
-        self.cfg.load(self.to_owned()).await?;
+        let mut last_in_err = false;
+        loop {
+            let res = self.cfg.load(&self.flags.server).await;
+            match res {
+                Ok(()) => {
+                    if last_in_err {
+                        println!("loaded config");
+                    }
+                    break;
+                }
+                Err(err) => {
+                    println!("error loading config: {err}, retrying in {CONFIG_RETRY} secs");
+                    last_in_err = true;
+                    sleep_s(CONFIG_RETRY).await;
+                }
+            };
+        }
+        if last_in_err {
+            println!("creating sas");
+        }
         self.create_sas().await?;
+        if last_in_err {
+            println!("created sas");
+        }
         Ok(())
     }
 
@@ -156,43 +155,49 @@ impl Context {
 
     pub async fn get_blocklist_toblock(&mut self) -> Vec<IpData> {
         let mut res: Vec<IpData> = vec![];
-        let now: DateTime<Local> = Local::now().trunc_subsecs(0);
         for (_, block) in self.blocklist.iter_mut() {
-            let set = self.cfg.sets.get(&block.ipdata.src.to_string()).unwrap();
-            if block.tryfail >= set.tryfail {
-                res.push(block.ipdata.clone());
-                if block.tryfail == set.tryfail {
-                    block.starttime = DateTime::from(now);
+            match self.cfg.sets.get(&block.ipdata.src) {
+                Some(set) => {
+                    if block.tryfail >= set.tryfail {
+                        res.push(block.ipdata.clone());
+                    }
                 }
+                None => {}
             }
         }
         res
     }
 
-    pub async fn update_blocklist(&mut self, ipdata: &mut IpData) -> Option<IpData> {
-        match self.cfg.sets.get(&ipdata.src) {
+    pub async fn update_blocklist(&mut self, ipevent: &IpEvent) -> Option<IpEvent> {
+        match self.cfg.sets.get(&ipevent.ipdata.src) {
             Some(set) => {
-                if self.blocklist.contains_key(&ipdata.ip)
-                    && self.hostname == ipdata.hostname
-                    && ipdata.mode == "file".to_string()
-                {
-                    let mut block = self.blocklist.get_mut(&ipdata.ip).unwrap();
+                let starttime = DateTime::parse_from_rfc3339(ipevent.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
+                        .blocklist
+                        .entry(ipevent.ipdata.ip.to_string())
+                        .or_insert(BlockIpData {
+                            ipdata: ipevent.ipdata.clone(),
+                            tryfail: 0,
+                            starttime,
+                            blocktime,
+                        });
                     block.tryfail += 1;
-                    block.blocktime = set.blocktime;
+                    block.blocktime = blocktime;
                     if block.tryfail >= set.tryfail {
-                        return Some(block.ipdata.clone());
+                        return Some(ipevent.clone());
                     }
                 } else {
-                    let starttime = DateTime::parse_from_rfc3339(ipdata.date.as_str())
-                        .unwrap()
-                        .with_timezone(&chrono::Local);
                     self.blocklist
-                        .entry(ipdata.ip.to_string())
+                        .entry(ipevent.ipdata.ip.to_string())
                         .or_insert(BlockIpData {
-                            ipdata: ipdata.clone(),
-                            tryfail: 100,
+                            ipdata: ipevent.ipdata.clone(),
+                            tryfail: set.tryfail,
                             starttime,
-                            blocktime: set.blocktime,
+                            blocktime,
                         });
                 }
             }
@@ -205,8 +210,15 @@ impl Context {
         let mut removed: Vec<IpData> = vec![];
         let now: DateTime<Local> = Local::now().trunc_subsecs(0);
         // nightly, future use
-        //let drained: HashMap<String,IpData> = ctx.blocklist.drain_filter(|k,v| v.parse_date() < mindate)
+        // 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();
@@ -273,7 +285,8 @@ pub struct Config {
     pub sets: HashMap<String, Set>,
     #[serde(skip_serializing)]
     pub trustnets: Vec<String>,
-    pub zmq: HashMap<String, ZMQ>,
+    pub ws: HashMap<String, WebSocketCfg>,
+    pub api: String,
 }
 
 impl Config {
@@ -323,34 +336,59 @@ impl Config {
                 "172.16.0.0/12".to_string(),
                 "192.168.0.0/16".to_string(),
             ],
-            zmq: HashMap::from([("pubsub".to_string(),ZMQ{
+            ws: HashMap::from([("pubsub".to_string(),WebSocketCfg{
                 t: "pubsub".to_string(),
-                hostname: MASTERSERVER.to_string(),
-                port: 9999,
-                subscription: ZMQSUBSCRIPTION.to_string(),
-            }),("reqrep".to_string(),ZMQ {
+                endpoint: format!("wss://{}/wsps", MASTERSERVER.to_string()),
+                subscription: WSSUBSCRIPTION.to_string(),
+            }),("reqrep".to_string(), WebSocketCfg {
                 t: "reqrep".to_string(),
-                hostname: MASTERSERVER.to_string(),
-                port: 9998,
-                subscription: String::new(),
-            })])
+                endpoint: format!("wss://{}/wsrr", MASTERSERVER.to_string()),
+                subscription: WSSUBSCRIPTION.to_string(),
+            })]),
+            api: String::from("127.0.0.1:8060")
         }
     }
 
-    pub async fn load(&mut self, ctx: Context) -> Result<(), ReqError> {
-        self.get_trustnets(&ctx).await?;
-        self.get_sets(&ctx).await?;
-        self.get_zmq_config(&ctx).await?;
+    pub async fn load(&mut self, server: &String) -> Result<(), ReqError> {
+        self.get_global_config(server).await?;
+        self.get_trustnets(server).await?;
+        self.get_sets(server).await?;
+        self.get_ws_config(server).await?;
         Ok(())
     }
 
-    async fn get_trustnets(&mut self, ctx: &Context) -> Result<(), ReqError> {
-        let resp: Result<Response, ReqError> = ctx
-            .client
-            .get(format!(
-                "{server}/config/trustlist",
-                server = ctx.flags.server
-            ))
+    async fn get_global_config(&mut self, server: &String) -> Result<(), ReqError> {
+        let resp: Result<Response, ReqError> =
+            httpclient().get(format!("{server}/config")).send().await;
+        let req = match resp {
+            Ok(re) => re,
+            Err(err) => return Err(err),
+        };
+        let data: HashMap<String, GlobalConfig> = match req.json::<Vec<GlobalConfig>>().await {
+            Ok(res) => {
+                let mut out: HashMap<String, GlobalConfig> = HashMap::new();
+                res.into_iter().map(|x| x).for_each(|x| {
+                    out.insert(x.key.to_string(), x);
+                });
+                out
+            }
+            Err(err) => return Err(err),
+        };
+        let key = "".to_string();
+        self.api = data
+            .get(&key.to_string())
+            .unwrap_or(&GlobalConfig {
+                key: "api".to_string(),
+                value: "127.0.0.1:8060".to_string(),
+            })
+            .value
+            .clone();
+        Ok(())
+    }
+
+    async fn get_trustnets(&mut self, server: &String) -> Result<(), ReqError> {
+        let resp: Result<Response, ReqError> = httpclient()
+            .get(format!("{server}/config/trustlist"))
             .send()
             .await;
         let req = match resp {
@@ -365,10 +403,9 @@ impl Config {
         Ok(())
     }
 
-    async fn get_sets(&mut self, ctx: &Context) -> Result<(), ReqError> {
-        let resp: Result<Response, ReqError> = ctx
-            .client
-            .get(format!("{server}/config/sets", server = ctx.flags.server))
+    async fn get_sets(&mut self, server: &String) -> Result<(), ReqError> {
+        let resp: Result<Response, ReqError> = httpclient()
+            .get(format!("{server}/config/sets"))
             .send()
             .await;
         let req = match resp {
@@ -385,19 +422,16 @@ impl Config {
         Ok(())
     }
 
-    async fn get_zmq_config(&mut self, ctx: &Context) -> Result<(), ReqError> {
-        let resp: Result<Response, ReqError> = ctx
-            .client
-            .get(format!("{server}/config/zmq", server = ctx.flags.server))
-            .send()
-            .await;
+    async fn get_ws_config(&mut self, server: &String) -> Result<(), ReqError> {
+        let resp: Result<Response, ReqError> =
+            httpclient().get(format!("{server}/config/ws")).send().await;
         let req = match resp {
             Ok(re) => re,
             Err(err) => return Err(err),
         };
-        let data: HashMap<String, ZMQ> = match req.json::<Vec<ZMQ>>().await {
+        let data: HashMap<String, WebSocketCfg> = match req.json::<Vec<WebSocketCfg>>().await {
             Ok(res) => {
-                let mut out: HashMap<String, ZMQ> = HashMap::new();
+                let mut out: HashMap<String, WebSocketCfg> = HashMap::new();
                 res.into_iter().map(|x| x).for_each(|x| {
                     out.insert(x.t.to_string(), x);
                 });
@@ -405,7 +439,7 @@ impl Config {
             }
             Err(err) => return Err(err),
         };
-        self.zmq = data;
+        self.ws = data;
         Ok(())
     }
 
@@ -421,6 +455,41 @@ impl Config {
         }
         trustnets
     }
+
+    pub fn bootstrap_event(&self) -> IpEvent {
+        IpEvent {
+            msgtype: String::from("bootstrap"),
+            mode: String::from("ws"),
+            hostname: gethostname(true),
+            ipdata: IpData {
+                ip: "".to_string(),
+                src: "".to_string(),
+                date: "".to_string(),
+                hostname: "".to_string(),
+            },
+        }
+    }
+}
+
+pub fn httpclient() -> Client {
+    let client = Client::builder()
+        .user_agent(format!(
+            "{}/{}@{}/{}",
+            env!("CARGO_PKG_NAME"),
+            env!("CARGO_PKG_VERSION"),
+            GIT_VERSION,
+            gethostname(false)
+        ))
+        .timeout(Duration::seconds(WEB_CLIENT_TIMEOUT).to_std().unwrap())
+        .build()
+        .unwrap();
+    client
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct GlobalConfig {
+    pub key: String,
+    pub value: String,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
@@ -434,11 +503,10 @@ pub struct Set {
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
-pub struct ZMQ {
+pub struct WebSocketCfg {
     #[serde(rename = "type")]
     pub t: String,
-    pub hostname: String,
-    pub port: i64,
+    pub endpoint: String,
     pub subscription: String,
 }
 
@@ -478,51 +546,71 @@ mod test {
         ctx.blocklist = HashMap::new();
 
         for _i in 0..10 {
-            ctx.update_blocklist(&mut IpData {
-                ip: "1.1.1.1".to_string(),
-                hostname: "test1".to_string(),
-                date: now.to_rfc3339().to_string(),
-                src: "ssh".to_string(),
-                mode: "file".to_string(),
+            ctx.update_blocklist(&mut IpEvent {
+                msgtype: String::from("add"),
+                mode: String::from("ws"),
+                hostname: String::from("localhost"),
+                ipdata: IpData {
+                    ip: "1.1.1.1".to_string(),
+                    hostname: "test1".to_string(),
+                    date: now.to_rfc3339().to_string(),
+                    src: "ssh".to_string(),
+                },
             })
             .await;
         }
 
         for _ in 0..10 {
-            ctx.update_blocklist(&mut IpData {
-                ip: "1.1.1.2".to_string(),
-                hostname: "test2".to_string(),
-                date: now.to_rfc3339().to_string(),
-                src: "http".to_string(),
-                mode: "file".to_string(),
+            ctx.update_blocklist(&mut IpEvent {
+                msgtype: String::from("add"),
+                mode: String::from("ws"),
+                hostname: String::from("localhost"),
+                ipdata: IpData {
+                    ip: "1.1.1.2".to_string(),
+                    hostname: "test2".to_string(),
+                    date: now.to_rfc3339().to_string(),
+                    src: "http".to_string(),
+                },
             })
             .await;
         }
 
-        ctx.update_blocklist(&mut IpData {
-            ip: "1.1.1.3".to_string(),
-            hostname: "testgood".to_string(),
-            date: now.to_rfc3339().to_string(),
-            src: "http".to_string(),
-            mode: "file".to_string(),
+        ctx.update_blocklist(&mut IpEvent {
+            msgtype: String::from("add"),
+            mode: String::from("ws"),
+            hostname: String::from("localhost"),
+            ipdata: IpData {
+                ip: "1.1.1.3".to_string(),
+                hostname: "testgood".to_string(),
+                date: now.to_rfc3339().to_string(),
+                src: "http".to_string(),
+            },
         })
         .await;
 
-        ctx.update_blocklist(&mut IpData {
-            ip: "1.1.1.4".to_string(),
-            hostname: "testgood".to_string(),
-            date: now.to_rfc3339().to_string(),
-            src: "http".to_string(),
-            mode: "file".to_string(),
+        ctx.update_blocklist(&mut IpEvent {
+            msgtype: String::from("add"),
+            mode: String::from("ws"),
+            hostname: String::from("localhost"),
+            ipdata: IpData {
+                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 IpData {
-            ip: "1.1.1.4".to_string(),
-            hostname: "testgood".to_string(),
-            date: now.to_rfc3339().to_string(),
-            src: "http".to_string(),
-            mode: "file".to_string(),
+        ctx.update_blocklist(&mut IpEvent {
+            msgtype: String::from("add"),
+            mode: String::from("ws"),
+            hostname: String::from("localhost"),
+            ipdata: IpData {
+                ip: "1.1.1.4".to_string(),
+                hostname: "testgood".to_string(),
+                date: now.to_rfc3339().to_string(),
+                src: "http".to_string(),
+            },
         })
         .await;
 

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(
-        &CString::new(tablename.as_str()).unwrap(),
-        ProtoFamily::Ipv4,
-    );
+    let table = Table::new(&CString::new(PKG_NAME).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(
-    tablename: &String,
+pub fn fwblock(
     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/ip.rs

@@ -1,4 +1,4 @@
-use crate::config::Context;
+use crate::config::httpclient;
 use crate::utils::gethostname;
 
 use chrono::prelude::*;
@@ -18,13 +18,20 @@ lazy_static! {
     static ref R_DATE: Regex = Regex::new(include_str!("regexps/date.txt")).unwrap();
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IpEvent {
+    pub msgtype: String,
+    pub mode: String,
+    pub hostname: String,
+    pub ipdata: IpData,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, Eq)]
 pub struct IpData {
     pub ip: String,
     pub src: String,
     pub date: String,
     pub hostname: String,
-    pub mode: String,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -60,89 +67,18 @@ impl Display for IpData {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(
             f,
-            "ip: {ip}, src: {src}, date: {date}, hostname: {hostname}, mode: {mode}",
+            "ip: {ip}, src: {src}, date: {date}, hostname: {hostname}",
             ip = self.ip,
             src = self.src,
             date = self.date,
             hostname = self.hostname,
-            mode = self.mode,
         )
     }
 }
 
-pub async fn push_ip(ctx: &Context, ip: &IpData, ret: &mut Vec<String>) -> Result<(), ReqError> {
-    let result: String;
-    let mut data: Vec<IpData> = vec![];
-
-    data.push(IpData {
-        ip: ip.ip.to_string(),
-        src: ip.src.to_string(),
-        date: ip.date.to_string(),
-        hostname: ip.hostname.to_string(),
-        mode: "file".to_string(),
-    });
-
-    let resp = ctx
-        .client
-        .post(format!("{server}/ips", server = ctx.flags.server))
-        .json(&data)
-        .send()
-        .await?;
-    ret.push(format!("status: {status}", status = resp.status()));
-
-    let res = resp.text().await.unwrap();
-
-    if res.trim().len() > 0 {
-        result = res.trim().to_string();
-    } else {
-        result = "".to_string();
-    }
-
-    ret.push(format!("response: {result}"));
-    Ok(())
-}
-
-pub async fn _push_ip_bulk(
-    ctx: &Context,
-    ips: &Vec<IpData>,
-    ret: &mut Vec<String>,
-) -> Result<(), ReqError> {
-    let result: String;
-    let mut data: Vec<IpData> = vec![];
-
-    for ip in ips {
-        data.push(IpData {
-            ip: ip.ip.to_string(),
-            src: ip.src.to_string(),
-            date: ip.date.to_string(),
-            hostname: ip.hostname.to_string(),
-            mode: "file".to_string(),
-        })
-    }
-
-    let resp = ctx
-        .client
-        .post(format!("{server}/ips", server = ctx.flags.server))
-        .json(&data)
-        .send()
-        .await?;
-    ret.push(format!("status: {status}", status = resp.status()));
-
-    let res = resp.text().await.unwrap();
-
-    if res.trim().len() > 0 {
-        result = res.trim().to_string();
-    } else {
-        result = "".to_string();
-    }
-
-    ret.push(format!("response: {result}"));
-    Ok(())
-}
-
 pub fn filter(
     lines: Box<dyn Read>,
-    list: &mut Vec<IpData>,
+    iplist: &mut Vec<IpData>,
     trustnets: &Vec<IpNet>,
     regex: &Regex,
     src: &String,
@@ -193,12 +129,11 @@ pub fn filter(
                 };
 
                 if !is_trusted(&ipaddr, &trustnets) {
-                    list.push(IpData {
+                    iplist.push(IpData {
                         ip: s_ipaddr,
                         src: src.to_owned(),
                         date: s_date.to_rfc3339().to_owned(),
                         hostname: hostname.to_owned(),
-                        mode: "file".to_owned(),
                     });
                     ips += 1;
                 };
@@ -222,8 +157,15 @@ fn parse_date(input: regex::Captures) -> DateTime<Local> {
     }
 
     let date = Local
-        .ymd(ymd[0] as i32, ymd[1] as u32, ymd[2] as u32)
-        .and_hms(hms[0] as u32, hms[1] as u32, hms[2] as u32);
+        .with_ymd_and_hms(
+            ymd[0] as i32,
+            ymd[1] as u32,
+            ymd[2] as u32,
+            hms[0] as u32,
+            hms[1] as u32,
+            hms[2] as u32,
+        )
+        .unwrap();
     date
 }
 
@@ -236,10 +178,9 @@ fn is_trusted(ip: &IpAddr, trustnets: &Vec<IpNet>) -> bool {
     false
 }
 
-pub async fn _get_last(ctx: &Context) -> Result<Vec<IpData>, ReqError> {
-    let resp = ctx
-        .client
-        .get(format!("{server}/ips/last", server = ctx.flags.server))
+pub async fn _get_last(server: &String) -> Result<Vec<IpData>, ReqError> {
+    let resp = httpclient()
+        .get(format!("{server}/ips/last"))
         .query(&[("interval", "3 hours")])
         .send()
         .await;

src/ipblc.rs

@@ -1,8 +1,10 @@
 use crate::config::{Context, GIT_VERSION};
-use crate::fw::{block, init};
-use crate::ip::{filter, push_ip, IpData};
+use crate::fw::{fwblock, fwinit};
+use crate::ip::{filter, IpData, IpEvent};
+use crate::monitoring::apiserver;
 use crate::utils::{gethostname, read_lines, sleep_s};
-use crate::zmqcom::zconnect;
+use crate::webservice::send_to_ipbl_api;
+use crate::websocket::{send_to_ipbl_websocket, websocketpubsub, websocketreqrep};
 
 use chrono::prelude::*;
 use chrono::prelude::{DateTime, Local};
@@ -11,126 +13,167 @@ use nix::sys::inotify::InotifyEvent;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
+pub const PKG_NAME: &str = env!("CARGO_PKG_NAME");
 const BL_CHAN_SIZE: usize = 32;
-const ZMQ_CHAN_SIZE: usize = 64;
+const WS_CHAN_SIZE: usize = 64;
+const LOOP_MAX_WAIT: u64 = 5;
 
 pub async fn run() {
-    let ctx = Arc::new(Mutex::new(Context::new().await));
-    println!(
-        "Launching {}, version {}",
-        env!("CARGO_PKG_NAME"),
-        format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION)
-    );
-
-    let (ipdatatx, mut ipdatarx): (Sender<IpData>, Receiver<IpData>) = channel(ZMQ_CHAN_SIZE);
-
-    // initialize the firewall table
-    init(&env!("CARGO_PKG_NAME").to_string());
+    let globalctx = Context::new().await;
+    let ctxarc = Arc::new(RwLock::new(globalctx));
     let mut fwlen: usize = 0;
 
-    // initialize zeromq sockets
-    let reqsocket;
-    let subsocket;
-    {
-        let ctxarc = Arc::clone(&ctx);
-        let zmqctx = ctxarc.lock().await;
-        reqsocket = zconnect(&zmqctx.cfg.zmq.get("reqrep").unwrap(), zmq::REQ)
-            .await
-            .unwrap();
-        subsocket = zconnect(&zmqctx.cfg.zmq.get("pubsub").unwrap(), zmq::SUB)
-            .await
-            .unwrap();
-    }
+    let pkgversion = format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION);
 
-    listenpubsub(&ctx, ipdatatx.clone(), subsocket).await;
+    let mut last_cfg_reload: DateTime<Local> = Local::now().trunc_subsecs(0);
+    println!("Launching {}, version {}", PKG_NAME, pkgversion);
+    fwinit();
 
-    let mut blrx = watchfiles(&ctx).await;
+    let ctxapi = Arc::clone(&ctxarc);
+    apiserver(&ctxapi).await.unwrap();
 
-    let ctxarc = Arc::clone(&ctx);
+    // initialize sockets
+    let (ipeventtx, mut ipeventrx): (Sender<IpEvent>, Receiver<IpEvent>) = channel(WS_CHAN_SIZE);
+    let ipeventtxarc = Arc::new(RwLock::new(ipeventtx));
+
+    // init pubsub
+    let ctxwsps = Arc::clone(&ctxarc);
+    let ipeventws = Arc::clone(&ipeventtxarc);
+    websocketpubsub(&ctxwsps, ipeventws).await;
+
+    let ctxwsrr = Arc::clone(&ctxarc);
+    let mut wssocketrr = websocketreqrep(&ctxwsrr).await;
+
+    // init file watcher
+    let mut blrx = watchfiles(&ctxarc).await;
+
+    let ctxclone = Arc::clone(&ctxarc);
+    let ipeventclone = Arc::clone(&ipeventtxarc);
     tokio::spawn(async move {
-        compare_files_changes(&ctxarc, &mut blrx, &ipdatatx).await;
+        compare_files_changes(&ctxclone, &mut blrx, &ipeventclone).await;
     });
 
-    let mut ip_init = IpData {
-        ip: "".to_string(),
-        src: "".to_string(),
-        date: "".to_string(),
-        hostname: "".to_string(),
-        mode: "init".to_string(),
-    };
-    send_to_ipbl_zmq(&reqsocket, &mut ip_init).await;
-
     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().await.unwrap();
+        let ctxclone = Arc::clone(&ctxarc);
 
-        // lock the context mutex
-        let ctxarc = Arc::clone(&ctx);
-        let mut ctx = ctxarc.lock().await;
+        tokio::select! {
+            ipevent = ipeventrx.recv() => {
+                let received_ip = ipevent.unwrap();
 
-        if received_ip.ip == "".to_string() && received_ip.mode == "init".to_string() {
-            for ip_to_send in &mut ctx.get_blocklist_toblock().await {
-                ip_to_send.mode = "init".to_string();
-                send_to_ipbl_zmq(&reqsocket, ip_to_send).await;
+                let (toblock,server);
+                {
+                    let mut ctx = ctxclone.write().await;
+                    toblock = ctx.get_blocklist_toblock().await;
+                    server = ctx.flags.server.clone();
+                }
+
+                if received_ip.msgtype == "bootstrap".to_string() {
+                    for ip_to_send in toblock {
+                        let ipe = IpEvent{
+                            msgtype: String::from("init"),
+                            mode: String::from("ws"),
+                            hostname: gethostname(true),
+                            ipdata: ip_to_send,
+                        };
+                        if !send_to_ipbl_websocket(&mut wssocketrr, &ipe).await {
+                            wssocketrr = websocketreqrep(&ctxwsrr).await;
+                            break;
+                        }
+                    }
+                    continue
+                }
+
+                // refresh context blocklist
+                let filtered_ipevent;
+                {
+                    let mut ctx = ctxarc.write().await;
+                    filtered_ipevent = ctx.update_blocklist(&received_ip).await;
+                }
+
+                // 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);
+                        let ipe = IpEvent{
+                            msgtype: String::from("add"),
+                            mode: String::from("ws"),
+                            hostname: gethostname(true),
+                            ipdata: ipevent.ipdata,
+                        };
+                        send_to_ipbl_api(&server.clone(), &ipe).await;
+                        let status = send_to_ipbl_websocket(&mut wssocketrr, &ipe).await;
+                        if !status {
+                            wssocketrr = websocketreqrep(&ctxwsrr).await;
+                            continue;
+                        }
+                    }
+                }
             }
-            continue;
-        }
+            _val = sleep_s(LOOP_MAX_WAIT) => {}
+        };
 
-        // 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(&reqsocket, &mut ip).await;
-        }
-
-        // apply firewall blocking
-        block(
-            &env!("CARGO_PKG_NAME").to_string(),
-            &ctx.get_blocklist_toblock().await,
-            &mut ret,
-            &mut fwlen,
-        )
-        .unwrap();
+        let ctxclone = Arc::clone(&ctxarc);
+        handle_fwblock(ctxclone, &mut ret, &mut fwlen).await;
 
         // 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}");
-                }
-            }
-        }
+        let ctxclone = Arc::clone(&ctxarc);
+        handle_cfg_reload(&ctxclone, &mut last_cfg_reload).await;
     }
 }
 
-async fn watchfiles(ctx: &Arc<Mutex<Context>>) -> Receiver<FileEvent> {
+async fn handle_cfg_reload(ctxclone: &Arc<RwLock<Context>>, last_cfg_reload: &mut DateTime<Local>) {
+    let now_cfg_reload = Local::now().trunc_subsecs(0);
+    if (now_cfg_reload - *last_cfg_reload) > Duration::seconds(LOOP_MAX_WAIT as i64) {
+        let mut ctx = ctxclone.write().await;
+        match ctx.load().await {
+            Ok(_) => {
+                *last_cfg_reload = Local::now().trunc_subsecs(0);
+            }
+            Err(_) => {
+                println!("error reloading config");
+            }
+        }
+    };
+}
+
+async fn handle_fwblock(ctxclone: Arc<RwLock<Context>>, ret: &mut Vec<String>, fwlen: &mut usize) {
+    let toblock = {
+        let mut ctx = ctxclone.write().await;
+        ctx.gc_blocklist().await;
+        ctx.get_blocklist_toblock().await
+    };
+
+    // apply firewall blocking
+    match fwblock(&toblock, ret, fwlen) {
+        Ok(_) => {}
+        Err(err) => {
+            println!("Err: {err}, unable to push firewall rules, use super user")
+        }
+    };
+}
+
+async fn watchfiles(ctxarc: &Arc<RwLock<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 instance = c.instance.clone();
-                drop(c);
-                events = instance.read_events().unwrap();
+                let ctx = ctxclone.read().await;
+                instance = ctx.instance.clone();
             }
 
+            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();
@@ -153,23 +196,29 @@ 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<RwLock<Context>>,
     inrx: &mut Receiver<FileEvent>,
-    ipdatatx: &Sender<IpData>,
+    ipeventtx: &Arc<RwLock<Sender<IpEvent>>>,
 ) {
     let mut tnets;
     loop {
         let modfiles = inrx.recv().await.unwrap();
         let mut iplist: Vec<IpData> = vec![];
 
-        let mut ctx = ctx.lock().await;
-        tnets = ctx.cfg.build_trustnets();
+        let sask;
+        let sas;
+        {
+            let ctx = ctxarc.read().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() {
-                    let sa = &mut ctx.sas.get_mut(sak).unwrap();
+                for sak in sask {
+                    let sa = sas.get(sak).unwrap();
                     if modfiles.inevent.wd == sa.wd {
                         let handle: String;
                         if sa.filename.as_str() == "" {
@@ -180,8 +229,13 @@ async fn compare_files_changes(
                             continue;
                         }
 
-                        let (filesize, sizechanged) =
-                            get_last_file_size(&mut sa.watchedfiles, &handle).await;
+                        let (filesize, sizechanged);
+                        {
+                            let mut ctx = ctxarc.write().await;
+                            let sa = ctx.sas.get_mut(sak).unwrap();
+                            (filesize, sizechanged) =
+                                get_last_file_size(&mut sa.watchedfiles, &handle).await;
+                        }
 
                         if !sizechanged {
                             continue;
@@ -204,7 +258,14 @@ async fn compare_files_changes(
                     }
                 }
                 for ip in iplist {
-                    ipdatatx.send(ip).await.unwrap();
+                    let ipevent = IpEvent {
+                        msgtype: String::from("add"),
+                        hostname: gethostname(true),
+                        mode: String::from("file"),
+                        ipdata: ip,
+                    };
+                    let ipetx = ipeventtx.write().await;
+                    ipetx.send(ipevent).await.unwrap();
                 }
             }
             None => {}
@@ -222,79 +283,3 @@ impl std::fmt::Debug for FileEvent {
         write!(f, "{ie:?}", ie = self.inevent)
     }
 }
-
-async fn send_to_ipbl_zmq(reqsocket: &zmq::Socket, ip: &mut IpData) {
-    let msg = format!("{val}", val = serde_json::to_string(&ip).unwrap());
-    match reqsocket.send(&msg, 0) {
-        Ok(_) => {}
-        Err(e) => {
-            println!("{e:?}")
-        }
-    };
-    match reqsocket.recv_string(0) {
-        Ok(o) => match o {
-            Ok(_) => {}
-            Err(ee) => {
-                println!("{ee:?}")
-            }
-        },
-        Err(e) => {
-            println!("{e:?}")
-        }
-    };
-}
-
-async fn send_to_ipbl_ws(ctx: &Context, ip: &mut IpData, ret: &mut Vec<String>) {
-    ret.push(format!("host: {hostname}", hostname = ctx.hostname));
-    loop {
-        match push_ip(&ctx, &ip, ret).await {
-            Ok(_) => {
-                break;
-            }
-            Err(err) => {
-                println!("{err}");
-                sleep_s(1);
-            }
-        };
-    }
-}
-
-async fn listenpubsub(ctx: &Arc<Mutex<Context>>, txpubsub: Sender<IpData>, socket: zmq::Socket) {
-    let ctx = ctx.lock().await;
-    let prefix = format!(
-        "{sub} ",
-        sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription
-    );
-    socket
-        .set_subscribe(ctx.cfg.zmq.get("pubsub").unwrap().subscription.as_bytes())
-        .expect("failed setting subscription");
-    drop(ctx);
-
-    tokio::spawn(async move {
-        loop {
-            let msgs: Option<String> = match socket.recv_string(0) {
-                Ok(s) => match s {
-                    Ok(ss) => Some(ss),
-                    Err(e) => {
-                        println!("{e:?}");
-                        None
-                    }
-                },
-                Err(e) => {
-                    println!("{e:?}");
-                    None
-                }
-            };
-            match msgs {
-                Some(ss) => {
-                    let msg = ss.strip_prefix(prefix.as_str()).unwrap();
-                    let tosend: IpData = serde_json::from_str(msg).unwrap();
-                    if tosend.hostname != gethostname(true) || tosend.mode == "init".to_string() {
-                        txpubsub.send(tosend).await.unwrap();
-                    }
-                }
-                None => {}
-            };
-        }
-    });
-}

src/main.rs

@@ -2,11 +2,12 @@ mod config;
 mod fw;
 mod ip;
 mod ipblc;
+mod monitoring;
 mod utils;
-mod zmqcom;
+mod webservice;
+mod websocket;
 
 #[tokio::main]
 pub async fn main() {
-    // Create a new context
     ipblc::run().await;
 }

src/monitoring.rs

@@ -0,0 +1,57 @@
+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::RwLock;
+
+pub async fn apiserver(ctxarc: &Arc<RwLock<Context>>) -> io::Result<()> {
+    let ctxarc = ctxarc.clone();
+    let addr;
+    {
+        let ctx = ctxarc.read().await;
+        addr = ctx.cfg.api.parse().unwrap();
+    }
+
+    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((stream, _addr)) => {
+                    //let mut buf = [0; 1024];
+                    let data;
+                    {
+                        let ctx = ctxarc.read().await;
+                        data = serde_json::to_string(&ctx.blocklist);
+                    }
+
+                    match data {
+                        Ok(dt) => {
+                            let (_reader, mut writer) = stream.into_split();
+                            match writer.write_all(format!("{dt}").as_bytes()).await {
+                                Ok(_) => {}
+                                Err(err) => {
+                                    println!("{err}");
+                                }
+                            }
+                        }
+                        Err(err) => {
+                            println!("unable to serialize data: {err}");
+                        }
+                    }
+                }
+                Err(err) => {
+                    println!("couldn't get client: {}", err)
+                }
+            }
+        }
+    });
+    Ok(())
+}

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) {
-    std::thread::sleep(Duration::from_millis(ms));
+pub async fn _sleep_ms(ms: u64) {
+    sleep(Duration::from_millis(ms)).await;
 }
 
-pub fn sleep_s(s: u64) {
-    std::thread::sleep(Duration::from_secs(s));
+pub async fn sleep_s(s: u64) {
+    sleep(Duration::from_secs(s)).await;
 }
 
 pub fn gethostname(show_fqdn: bool) -> String {

src/webservice.rs

@@ -0,0 +1,83 @@
+use crate::config::{httpclient, Context};
+use crate::ip::{IpData, IpEvent};
+use crate::utils::sleep_s;
+
+use reqwest::Client;
+use reqwest::Error as ReqError;
+
+const MAX_FAILED_API_RATE: u64 = 10;
+
+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 {
+            Ok(_) => {
+                break;
+            }
+            Err(err) => {
+                println!("{err}");
+                sleep_s(1).await;
+                if try_req == MAX_FAILED_API_RATE {
+                    break;
+                }
+                try_req += 1;
+            }
+        };
+    }
+}
+
+async fn push_ip(client: &Client, server: &str, ip: &IpData) -> Result<(), ReqError> {
+    let mut data: Vec<IpData> = vec![];
+
+    data.push(IpData {
+        ip: ip.ip.to_string(),
+        src: ip.src.to_string(),
+        date: ip.date.to_string(),
+        hostname: ip.hostname.to_string(),
+    });
+
+    client
+        .post(format!("{server}/ips"))
+        .json(&data)
+        .send()
+        .await?;
+
+    Ok(())
+}
+
+async fn _push_ip_bulk(
+    ctx: &Context,
+    ips: &Vec<IpData>,
+    ret: &mut Vec<String>,
+) -> Result<(), ReqError> {
+    let result: String;
+    let mut data: Vec<IpData> = vec![];
+
+    for ip in ips {
+        data.push(IpData {
+            ip: ip.ip.to_string(),
+            src: ip.src.to_string(),
+            date: ip.date.to_string(),
+            hostname: ip.hostname.to_string(),
+        })
+    }
+
+    let resp = httpclient()
+        .post(format!("{server}/ips", server = ctx.flags.server))
+        .json(&data)
+        .send()
+        .await?;
+    ret.push(format!("status: {status}", status = resp.status()));
+
+    let res = resp.text().await.unwrap();
+
+    if res.trim().len() > 0 {
+        result = res.trim().to_string();
+    } else {
+        result = "".to_string();
+    }
+
+    ret.push(format!("response: {result}"));
+    Ok(())
+}

src/websocket.rs

@@ -0,0 +1,134 @@
+use crate::config::{Context, WebSocketCfg};
+use crate::ip::IpEvent;
+use crate::utils::{gethostname, sleep_s};
+
+use serde_json::json;
+use std::io::{self, Write};
+use std::net::TcpStream;
+use std::sync::Arc;
+use tokio::sync::mpsc::Sender;
+use tokio::sync::RwLock;
+use tungstenite::stream::*;
+use tungstenite::*;
+
+pub async fn websocketreqrep(
+    ctxarc: &Arc<RwLock<Context>>,
+) -> WebSocket<MaybeTlsStream<TcpStream>> {
+    let (mut wssocketrr, bootstrap_event, cfg);
+    {
+        let ctx = ctxarc.read().await;
+        bootstrap_event = ctx.cfg.bootstrap_event().clone();
+        cfg = ctx.cfg.ws.get("reqrep").unwrap().clone();
+    }
+    wssocketrr = websocketconnect(&cfg, &gethostname(true)).await.unwrap();
+    send_to_ipbl_websocket(&mut wssocketrr, &bootstrap_event).await;
+
+    return wssocketrr;
+}
+
+pub async fn websocketpubsub(
+    ctxarc: &Arc<RwLock<Context>>,
+    txpubsub: Arc<RwLock<Sender<IpEvent>>>,
+) {
+    let cfg;
+    {
+        let ctx = ctxarc.read().await;
+        cfg = ctx.cfg.ws.get("pubsub").unwrap().clone();
+    }
+    let mut websocket = Arc::new(RwLock::new(
+        websocketconnect(&cfg, &gethostname(true)).await.unwrap(),
+    ));
+    tokio::spawn(async move {
+        loop {
+            let mut ws = websocket.write().await;
+            match ws.read_message() {
+                Ok(msg) => {
+                    let tosend: IpEvent = match serde_json::from_str(msg.to_string().as_str()) {
+                        Ok(o) => o,
+                        Err(_e) => {
+                            continue;
+                        }
+                    };
+                    if tosend.ipdata.hostname != gethostname(true)
+                        || tosend.msgtype == "init".to_string()
+                    {
+                        let txps = txpubsub.write().await;
+                        txps.send(tosend).await.unwrap();
+                    }
+                }
+                Err(e) => {
+                    println!("error in pubsub: {e:?}");
+                    ws.close(None).unwrap();
+                    drop(ws);
+                    websocket = Arc::new(RwLock::new(
+                        websocketconnect(&cfg, &gethostname(true)).await.unwrap(),
+                    ));
+                }
+            };
+        }
+    });
+}
+
+pub async fn websocketconnect<'a>(
+    wscfg: &WebSocketCfg,
+    hostname: &String,
+) -> Result<WebSocket<MaybeTlsStream<TcpStream>>, Error> {
+    let endpoint = &wscfg.endpoint;
+    print!("connecting to {} ... ", endpoint);
+    io::stdout().flush().unwrap();
+    let mut socket;
+    loop {
+        (socket, _) = match connect(endpoint) {
+            Ok((o, e)) => (o, e),
+            _ => {
+                println!("error connecting to {endpoint}, retrying");
+                sleep_s(1).await;
+                continue;
+            }
+        };
+        break;
+    }
+    println!("connected to {endpoint}");
+    let msg = json!({ "hostname": hostname });
+    socket
+        .write_message(Message::Text(msg.to_string()))
+        .unwrap();
+    Ok(socket)
+}
+
+pub async fn send_to_ipbl_websocket(
+    ws: &mut WebSocket<MaybeTlsStream<TcpStream>>,
+    ip: &IpEvent,
+) -> bool {
+    let msg = format!("{val}", val = serde_json::to_string(&ip).unwrap());
+
+    if ws.can_write() {
+        match ws.write_message(Message::Text(msg)) {
+            Ok(_) => {}
+            Err(e) => {
+                println!("err send read: {e:?}");
+                return handle_websocket_error(ws);
+            }
+        };
+    } else {
+        return handle_websocket_error(ws);
+    };
+
+    if ws.can_read() {
+        match ws.read_message() {
+            Ok(_) => {}
+            Err(e) => {
+                println!("err send read: {e:?}");
+                return handle_websocket_error(ws);
+            }
+        };
+    } else {
+        return handle_websocket_error(ws);
+    };
+    true
+}
+
+fn handle_websocket_error(ws: &mut WebSocket<MaybeTlsStream<TcpStream>>) -> bool {
+    ws.close(None).unwrap();
+    return false;
+}

src/zmqcom.rs

@@ -1,13 +0,0 @@
-use crate::config::ZMQ;
-
-const ZMQPROTO: &str = "tcp";
-
-pub async fn zconnect(zmqcfg: &ZMQ, zmqtype: zmq::SocketType) -> Result<zmq::Socket, zmq::Error> {
-    let zctx = zmq::Context::new();
-    let zmqhost = &zmqcfg.hostname;
-    let zmqport = &zmqcfg.port;
-    let socket = zctx.socket(zmqtype).unwrap();
-    let connectstring = format!("{ZMQPROTO}://{zmqhost}:{zmqport}");
-    socket.connect(&connectstring.as_str())?;
-    Ok(socket)
-}