added websocket feature

Cargo.toml

@@ -22,4 +22,4 @@ reqwest = { version = "0.11", default-features = false, features = ["json","rust
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tokio = { version = "1.23", features = ["full", "sync"] }
-zmq = "0.10"
+tungstenite = { version = "0.18", features = ["handshake","rustls-tls-native-roots"] }

src/config.rs

@@ -17,6 +17,7 @@ use std::path::Path;
 pub const GIT_VERSION: &str = git_version!();
 const MASTERSERVER: &str = "ipbl.paulbsd.com";
 const ZMQSUBSCRIPTION: &str = "ipbl";
+const WSSUBSCRIPTION: &str = "ipbl";
 const CONFIG_RETRY: u64 = 10;
 
 #[derive(Debug, Clone)]
@@ -286,7 +287,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 zmq: HashMap<String, ZMQCfg>,
     pub api: String,
 }
 
@@ -337,12 +339,21 @@ 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(),
+                endpoint: format!("wss://{}/wsps", MASTERSERVER.to_string()),
+                subscription: WSSUBSCRIPTION.to_string(),
+            }),("reqrep".to_string(), WebSocketCfg {
+                t: "reqrep".to_string(),
+                endpoint: format!("wss://{}/wsrr", MASTERSERVER.to_string()),
+                subscription: WSSUBSCRIPTION.to_string(),
+            })]),
+            zmq: HashMap::from([("pubsub".to_string(),ZMQCfg{
                 t: "pubsub".to_string(),
                 hostname: MASTERSERVER.to_string(),
                 port: 9999,
                 subscription: ZMQSUBSCRIPTION.to_string(),
-            }),("reqrep".to_string(),ZMQ {
+            }),("reqrep".to_string(),ZMQCfg {
                 t: "reqrep".to_string(),
                 hostname: MASTERSERVER.to_string(),
                 port: 9998,
@@ -357,6 +368,7 @@ impl Config {
         self.get_trustnets(&ctx).await?;
         self.get_sets(&ctx).await?;
         self.get_zmq_config(&ctx).await?;
+        self.get_ws_config(&ctx).await?;
         Ok(())
     }
 
@@ -443,9 +455,9 @@ impl Config {
             Ok(re) => re,
             Err(err) => return Err(err),
         };
-        let data: HashMap<String, ZMQ> = match req.json::<Vec<ZMQ>>().await {
+        let data: HashMap<String, ZMQCfg> = match req.json::<Vec<ZMQCfg>>().await {
             Ok(res) => {
-                let mut out: HashMap<String, ZMQ> = HashMap::new();
+                let mut out: HashMap<String, ZMQCfg> = HashMap::new();
                 res.into_iter().map(|x| x).for_each(|x| {
                     out.insert(x.t.to_string(), x);
                 });
@@ -457,6 +469,30 @@ impl Config {
         Ok(())
     }
 
+    async fn get_ws_config(&mut self, ctx: &Context) -> Result<(), ReqError> {
+        let resp: Result<Response, ReqError> = ctx
+            .client
+            .get(format!("{server}/config/ws", server = ctx.flags.server))
+            .send()
+            .await;
+        let req = match resp {
+            Ok(re) => re,
+            Err(err) => return Err(err),
+        };
+        let data: HashMap<String, WebSocketCfg> = match req.json::<Vec<WebSocketCfg>>().await {
+            Ok(res) => {
+                let mut out: HashMap<String, WebSocketCfg> = HashMap::new();
+                res.into_iter().map(|x| x).for_each(|x| {
+                    out.insert(x.t.to_string(), x);
+                });
+                out
+            }
+            Err(err) => return Err(err),
+        };
+        self.ws = data;
+        Ok(())
+    }
+
     pub fn build_trustnets(&self) -> Vec<IpNet> {
         let mut trustnets: Vec<IpNet> = vec![];
         for trustnet in &self.trustnets {
@@ -469,6 +505,20 @@ impl Config {
         }
         trustnets
     }
+
+    pub fn bootstrap_event(&self) -> IpEvent {
+        IpEvent {
+            msgtype: String::from("bootstrap"),
+            mode: String::from("socket"),
+            hostname: gethostname(true),
+            ipdata: IpData {
+                ip: "".to_string(),
+                src: "".to_string(),
+                date: "".to_string(),
+                hostname: "".to_string(),
+            },
+        }
+    }
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
@@ -488,7 +538,7 @@ pub struct Set {
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
-pub struct ZMQ {
+pub struct ZMQCfg {
     #[serde(rename = "type")]
     pub t: String,
     pub hostname: String,
@@ -496,6 +546,14 @@ pub struct ZMQ {
     pub subscription: String,
 }
 
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct WebSocketCfg {
+    #[serde(rename = "type")]
+    pub t: String,
+    pub endpoint: String,
+    pub subscription: String,
+}
+
 #[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct Discovery {
     pub version: String,

src/ipblc.rs

@@ -3,8 +3,8 @@ use crate::config::{Context, GIT_VERSION};
 use crate::fw::{fwblock, fwinit};
 use crate::ip::{filter, IpData, IpEvent};
 use crate::utils::{gethostname, read_lines, sleep_ms};
-use crate::ws::send_to_ipbl_ws;
+use crate::webservice::send_to_ipbl_api;
-use crate::zmqcom::{send_to_ipbl_zmq, zmqinit};
+use crate::websocket::{send_to_ipbl_websocket, websocketinit};
 
 use chrono::prelude::*;
 use chrono::prelude::{DateTime, Local};
@@ -25,25 +25,22 @@ pub async fn run() {
     let globalctx = Context::new().await;
     let ctxarc = Arc::new(RwLock::new(globalctx));
     let mut ret: Vec<String> = Vec::new();
+    let mut fwlen: usize = 0;
 
     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);
+    fwinit();
 
     let (_apitx, mut apirx): (Sender<String>, Receiver<String>) = channel(API_CHAN_SIZE);
-    //let tcpsocket = apiinit(&ctx, &apitx).await;
 
     let ctxclone = Arc::clone(&ctxarc);
     apiserver(&ctxclone).await.unwrap();
 
-    // initialize the firewall table
+    // initialize sockets
-    fwinit();
-    let mut fwlen: usize = 0;
-
-    // initialize zeromq sockets
     let (ipeventtx, mut ipeventrx): (Sender<IpEvent>, Receiver<IpEvent>) = channel(ZMQ_CHAN_SIZE);
-    let zmqreqsocket = zmqinit(&ctxclone, &ipeventtx).await;
+    let (wspubsubsocket, mut wsreqsocket) = websocketinit(&ctxclone, &ipeventtx).await;
 
     let mut blrx = watchfiles(&ctxclone).await;
 
@@ -52,18 +49,9 @@ pub async fn run() {
         compare_files_changes(&ctxclone, &mut blrx, &ipeventtx).await;
     });
 
-    let ipevent_bootstrap = IpEvent {
+    let ctxclone = Arc::clone(&ctxarc);
-        msgtype: String::from("bootstrap"),
+    let bootstrap_event = ctxclone.read().await.cfg.bootstrap_event().clone();
-        mode: String::from("zmq"),
+    send_to_ipbl_websocket(&mut wsreqsocket, &bootstrap_event, &mut ret).await;
-        hostname: fqdn.clone(),
-        ipdata: IpData {
-            ip: "".to_string(),
-            src: "".to_string(),
-            date: "".to_string(),
-            hostname: "".to_string(),
-        },
-    };
-    send_to_ipbl_zmq(&zmqreqsocket, &ipevent_bootstrap, &mut ret).await;
 
     loop {
         ret = Vec::new();
@@ -90,7 +78,7 @@ pub async fn run() {
                             hostname: fqdn.clone(),
                             ipdata: ip_to_send,
                         };
-                        send_to_ipbl_zmq(&zmqreqsocket, &ipe, &mut ret).await;
+                        send_to_ipbl_websocket(&mut wsreqsocket, &ipe, &mut ret).await;
                     }
                     continue
                 }
@@ -98,18 +86,18 @@ pub async fn run() {
                 // refresh context blocklist
                 let filtered_ipevent = ctx.update_blocklist(&received_ip).await;
 
-                // send ip list to ws and zmq sockets
+                // send ip list to api and ws sockets
                 if let Some(ipevent) = filtered_ipevent {
                     if received_ip.msgtype != "init" {
-                        println!("sending {} to ws and zmq", ipevent.ipdata.ip);
+                        println!("sending {} to api and ws", ipevent.ipdata.ip);
                         let event = IpEvent{
                             msgtype: String::from("add"),
-                            mode: String::from("zmq"),
+                            mode: String::from("socket"),
                             hostname: fqdn.clone(),
                             ipdata: ipevent.ipdata,
                         };
-                        send_to_ipbl_ws(&ctx, &event, &mut ret).await;
+                        send_to_ipbl_api(&ctx, &event, &mut ret).await;
-                        send_to_ipbl_zmq(&zmqreqsocket, &event, &mut ret).await;
+                        send_to_ipbl_websocket(&mut wsreqsocket, &event, &mut ret).await;
                     }
                 }
             }

src/main.rs

@@ -4,8 +4,8 @@ mod fw;
 mod ip;
 mod ipblc;
 mod utils;
-mod ws;
+mod webservice;
-mod zmqcom;
+mod websocket;
 
 #[tokio::main]
 pub async fn main() {

src/webservice.rs

@@ -4,7 +4,7 @@ use crate::utils::sleep_s;
 
 use reqwest::Error as ReqError;
 
-pub async fn send_to_ipbl_ws(ctx: &Context, ip: &IpEvent, ret: &mut Vec<String>) {
+pub async fn send_to_ipbl_api(ctx: &Context, ip: &IpEvent, ret: &mut Vec<String>) {
     ret.push(format!("host: {hostname}", hostname = ctx.hostname));
     loop {
         match push_ip(&ctx, &ip.ipdata, ret).await {

src/websocket.rs

@@ -0,0 +1,105 @@
+use crate::config::{Context, WebSocketCfg};
+use crate::ip::IpEvent;
+use crate::utils::gethostname;
+use std::net::TcpStream;
+use tungstenite::stream::*;
+use tungstenite::*;
+
+use std::sync::Arc;
+use tokio::sync::mpsc::Sender;
+use tokio::sync::RwLock;
+
+pub async fn websocketconnect<'a>(
+    wscfg: &WebSocketCfg,
+) -> Result<WebSocket<MaybeTlsStream<TcpStream>>, Error> {
+    let (socket, response) = connect(&wscfg.endpoint).expect("Can't connect");
+    Ok(socket)
+}
+
+pub async fn websocketinit(
+    ctx: &Arc<RwLock<Context>>,
+    ipeventtx: &Sender<IpEvent>,
+) -> (
+    Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
+    WebSocket<MaybeTlsStream<TcpStream>>,
+) {
+    let ctxarc = Arc::clone(&ctx);
+
+    let wsreqsocket;
+    let wssubsocket;
+    let wssubsocketcb;
+    {
+        let ctx = ctxarc.read().await;
+        wsreqsocket = websocketconnect(&ctx.cfg.ws.get("reqrep").unwrap())
+            .await
+            .unwrap();
+        wssubsocket = Arc::new(RwLock::new(
+            websocketconnect(&ctx.cfg.ws.get("pubsub").unwrap())
+                .await
+                .unwrap(),
+        ));
+
+        wssubsocketcb = wssubsocket.clone();
+    }
+
+    wslistenpubsub(wssubsocket, ipeventtx.clone()).await;
+    return (wssubsocketcb, wsreqsocket);
+}
+
+async fn wslistenpubsub(
+    socket: Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
+    txpubsub: Sender<IpEvent>,
+) {
+    tokio::spawn(async move {
+        loop {
+            let msgs: Option<String>;
+            {
+                let mut socket = socket.write().await;
+                msgs = match socket.read_message() {
+                    Ok(s) => {
+                        println!("msg: {}", s);
+                        None
+                    }
+                    Err(e) => {
+                        println!("error: {e:?}");
+                        socket.close(None).unwrap();
+                        return;
+                    }
+                };
+            }
+            match msgs {
+                Some(ss) => {
+                    let tosend: IpEvent = serde_json::from_str(ss.as_str()).unwrap();
+                    if tosend.ipdata.hostname != gethostname(true)
+                        || tosend.msgtype == "init".to_string()
+                    {
+                        txpubsub.send(tosend).await.unwrap();
+                    }
+                }
+                None => {}
+            };
+        }
+    });
+}
+
+pub async fn send_to_ipbl_websocket(
+    reqsocket: &mut WebSocket<MaybeTlsStream<TcpStream>>,
+    ip: &IpEvent,
+    _ret: &mut Vec<String>,
+) {
+    let msg = format!("{val}", val = serde_json::to_string(&ip).unwrap());
+    match reqsocket.write_message(Message::Text(msg)) {
+        Ok(_) => {}
+        Err(e) => {
+            println!("{e:?}")
+        }
+    };
+    match reqsocket.read_message() {
+        Ok(o) => {
+            println!("{o}")
+        }
+        Err(e) => {
+            println!("{e:?}")
+        }
+    };
+}

src/zmqcom.rs

@@ -1,4 +1,4 @@
-use crate::config::{Context, ZMQ};
+use crate::config::{Context, ZMQCfg};
 use crate::ip::IpEvent;
 use crate::utils::gethostname;
 
@@ -9,7 +9,7 @@ use tokio::sync::RwLock;
 const ZMQPROTO: &str = "tcp";
 
 pub async fn zconnect<'a>(
-    zmqcfg: &ZMQ,
+    zmqcfg: &ZMQCfg,
     zmqtype: zmq::SocketType,
 ) -> Result<zmq::Socket, zmq::Error> {
     let zctx = zmq::Context::new();