paulbsd/ipblc
1
0
Fork 0
Code Issues Pull Requests Projects Releases 15 Wiki Activity

refactor of modules

Browse Source
This commit is contained in:
Paul 2022-12-30 20:18:15 +01:00
parent bbce4547cf
commit ff99fce62b
9 changed files with 119 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
src/ipblc/full.rs → old/full.rs
View File

6
src/config/mod.rs → src/config.rs
View File

@ -1,5 +1,5 @@
use crate::ip::{BlockIpData, IpData};
use crate::utils::*;
use crate::utils::{gethostname, sleep_s};
use chrono::prelude::*;
use chrono::Duration;
@ -90,7 +90,7 @@ impl Context {
}
Err(err) => {
println!("error loading config: {err}, retrying in {CONFIG_RETRY} secs");
sleep(CONFIG_RETRY);
sleep_s(CONFIG_RETRY);
}
}
}
@ -99,7 +99,7 @@ impl Context {
pub fn argparse() -> ArgMatches {
Command::new(env!("CARGO_PKG_NAME"))
.version(format!("{}/{}", env!("CARGO_PKG_VERSION"), GIT_VERSION).as_str())
.version(format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION).as_str())
.author(env!("CARGO_PKG_AUTHORS"))
.about(env!("CARGO_PKG_DESCRIPTION"))
.arg(

3
src/firewall/mod.rs → src/fw.rs
View File

@ -36,6 +36,9 @@ pub fn block(
// add chain
batch.add(&chain, nftnl::MsgType::Add);
let rule = Rule::new(&chain);
batch.add(&rule, 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));

9
src/ip.rs
View File

@ -146,7 +146,7 @@ pub fn filter(
trustnets: &Vec<IpNet>,
regex: &Regex,
src: &String,
lastprocess: &DateTime<Local>,
last: &DateTime<Local>,
) -> isize {
let mut ips = 0;
let hostname = gethostname(true);
@ -160,15 +160,14 @@ pub fn filter(
s_ipaddr = sv4.get(0).unwrap().as_str().to_string();
}
None => {
continue;
/*match R_IPV6.captures(l.as_str()) {
match R_IPV6.captures(l.as_str()) {
Some(sv6) => {
s_ipaddr = sv6.get(0).unwrap().as_str().to_string();
}
None => {
continue;
}
};*/
};
}
};
@ -176,7 +175,7 @@ pub fn filter(
match R_DATE.captures(l.as_str()) {
Some(sdt) => {
s_date = parse_date(sdt);
if &s_date < lastprocess {
if &s_date < last {
continue;
}
}

106
src/ipblc/inc.rs → src/ipblc.rs
View File

@ -1,7 +1,13 @@
use super::*;
use crate::config::{Context, GIT_VERSION};
use crate::fw;
use crate::ip::{filter, push_ip, IpData};
use crate::utils::{gethostname, read_lines, sleep_s};
use crate::zmqcom::*;
use chrono::prelude::*;
use chrono::prelude::{DateTime, Local};
use chrono::Duration;
use nix::sys::inotify::InotifyEvent;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc::{channel, Receiver, Sender};
@ -10,17 +16,18 @@ use tokio::sync::Mutex;
const BL_CHAN_SIZE: usize = 32;
const ZMQ_CHAN_SIZE: usize = 64;
pub async fn process(ctx: &Arc<Mutex<Context>>) {
pub async fn run() {
let ctx = Arc::new(Mutex::new(Context::new().await));
println!(
"Launching {} version {}",
"Launching {}, version {}",
env!("CARGO_PKG_NAME"),
env!("CARGO_PKG_VERSION")
format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION)
);
let (ipdatatx, mut ipdatarx): (Sender<IpData>, Receiver<IpData>) = channel(ZMQ_CHAN_SIZE);
// initialize the firewall table
firewall::init(&env!("CARGO_PKG_NAME").to_string());
fw::init(&env!("CARGO_PKG_NAME").to_string());
let mut fwlen: usize = 0;
// initialize zeromq sockets
@ -85,7 +92,7 @@ pub async fn process(ctx: &Arc<Mutex<Context>>) {
}
// apply firewall blocking
firewall::block(
fw::block(
&env!("CARGO_PKG_NAME").to_string(),
&ctx.get_blocklist_toblock().await,
&mut ret,
@ -204,3 +211,90 @@ async fn compare_files_changes(
}
}
}
pub struct FileEvent {
pub inevent: InotifyEvent,
pub date: DateTime<Local>,
}
impl std::fmt::Debug for FileEvent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
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 => {}
};
}
});
}

100
src/ipblc/mod.rs
View File

@ -1,100 +0,0 @@
pub mod inc;
use crate::config::*;
use crate::firewall;
use crate::ip::*;
use crate::utils::*;
use crate::zmqcom::*;
use chrono::prelude::{DateTime, Local};
use nix::sys::inotify::InotifyEvent;
use std::sync::Arc;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;
pub struct FileEvent {
pub inevent: InotifyEvent,
pub date: DateTime<Local>,
}
impl std::fmt::Debug for FileEvent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{ie:?}", ie = self.inevent)
}
}
async fn send_to_ipbl_zmq(reqsocket: &zmq::Socket, ip: &mut IpData) {
let msg = format!("{value}", value = 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(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 => {}
};
}
});
}

9
src/main.rs
View File

@ -1,17 +1,12 @@
mod config;
mod firewall;
mod fw;
mod ip;
mod ipblc;
mod utils;
mod zmqcom;
use config::Context;
use std::sync::Arc;
use tokio::sync::Mutex;
#[tokio::main]
pub async fn main() {
// Create a new context
let ctx = Arc::new(Mutex::new(Context::new().await));
ipblc::inc::process(&ctx).await;
ipblc::run().await;
}

8
src/utils.rs
View File

@ -33,8 +33,12 @@ pub fn _dedup<T: Ord + PartialOrd>(list: &mut Vec<T>) -> usize {
list.len()
}
pub fn sleep(seconds: u64) {
std::thread::sleep(Duration::from_secs(seconds));
pub fn sleep_ms(ms: u64) {
std::thread::sleep(Duration::from_millis(ms));
}
pub fn sleep_s(s: u64) {
std::thread::sleep(Duration::from_secs(s));
}
pub fn gethostname(show_fqdn: bool) -> String {

2
src/zmqcom.rs
View File

@ -7,7 +7,7 @@ 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 zmqport = &zmqcfg.port;
let socket = zctx.socket(zmqtype).unwrap();
let connectstring = format!("{ZMQPROTO}://{zmqhost}:{zmqport}");
socket.connect(&connectstring.as_str())?;