diff --git a/Cargo.toml b/Cargo.toml index 68669ba..8aba6a0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,11 @@ panic = "abort" [profile.release] panic = "abort" +opt-level = 3 +lto = true +strip = true +debug = false +overflow-checks = false [profile.release.package.net-guardia-ebpf] debug = 2 diff --git a/net-guardia/src/core/statistics.rs b/net-guardia/src/core/statistics.rs index 140f29f..c4848a4 100644 --- a/net-guardia/src/core/statistics.rs +++ b/net-guardia/src/core/statistics.rs @@ -1,6 +1,8 @@ use crate::core::system::System; +use crate::model::direction::Direction; use crate::model::flow_stats::FlowStats; -use crate::model::flow_type::FlowType; +use crate::model::ip_address::SocketAddressType; +use crate::model::time_type::TimeType; use crate::utils::log_entry::system::SystemEntry; use aya::maps::{HashMap as AyaHashMap, MapData}; use aya::Pod; @@ -16,39 +18,62 @@ static STATISTICS: OnceLock> = OnceLock::new(); pub struct Statistics { terminate: bool, - ipv4_src_1min: AyaHashMap, - ipv4_src_10min: AyaHashMap, - ipv4_src_1hour: AyaHashMap, - ipv6_src_1min: AyaHashMap, - ipv6_src_10min: AyaHashMap, - ipv6_src_1hour: AyaHashMap, - ipv4_dst_1min: AyaHashMap, - ipv4_dst_10min: AyaHashMap, - ipv4_dst_1hour: AyaHashMap, - ipv6_dst_1min: AyaHashMap, - ipv6_dst_10min: AyaHashMap, - ipv6_dst_1hour: AyaHashMap, + ipv4_maps: StdHashMap<(Direction, TimeType), FlowMap>, + ipv6_maps: StdHashMap<(Direction, TimeType), FlowMap>, } impl Statistics { + const MAP_CONFIGS: [((Direction, TimeType), (&'static str, &'static str)); 6] = [ + ( + (Direction::Source, TimeType::_1Min), + ("IPV4_SRC_1MIN", "IPV6_SRC_1MIN"), + ), + ( + (Direction::Source, TimeType::_10Min), + ("IPV4_SRC_10MIN", "IPV6_SRC_10MIN"), + ), + ( + (Direction::Source, TimeType::_1Hour), + ("IPV4_SRC_1HOUR", "IPV6_SRC_1HOUR"), + ), + ( + (Direction::Destination, TimeType::_1Min), + ("IPV4_DST_1MIN", "IPV6_DST_1MIN"), + ), + ( + (Direction::Destination, TimeType::_10Min), + ("IPV4_DST_10MIN", "IPV6_DST_10MIN"), + ), + ( + (Direction::Destination, TimeType::_1Hour), + ("IPV4_DST_1HOUR", "IPV6_DST_1HOUR"), + ), + ]; + pub async fn initialize() -> anyhow::Result<()> { info!("{}", SystemEntry::Initializing); let mut system = System::instance_mut().await; let ebpf = &mut system.ebpf; + let mut ipv4_maps = StdHashMap::new(); + let mut ipv6_maps = StdHashMap::new(); + for (key, (ipv4_name, ipv6_name)) in Self::MAP_CONFIGS { + ipv4_maps.insert( + key, + FlowMap { + map: AyaHashMap::try_from(ebpf.take_map(ipv4_name).unwrap())?, + }, + ); + ipv6_maps.insert( + key, + FlowMap { + map: AyaHashMap::try_from(ebpf.take_map(ipv6_name).unwrap())?, + }, + ); + } let monitor = Statistics { terminate: false, - ipv4_src_1min: AyaHashMap::try_from(ebpf.take_map("IPV4_SRC_1MIN").unwrap())?, - ipv4_src_10min: AyaHashMap::try_from(ebpf.take_map("IPV4_SRC_10MIN").unwrap())?, - ipv4_src_1hour: AyaHashMap::try_from(ebpf.take_map("IPV4_SRC_1HOUR").unwrap())?, - ipv6_src_1min: AyaHashMap::try_from(ebpf.take_map("IPV6_SRC_1MIN").unwrap())?, - ipv6_src_10min: AyaHashMap::try_from(ebpf.take_map("IPV6_SRC_10MIN").unwrap())?, - ipv6_src_1hour: AyaHashMap::try_from(ebpf.take_map("IPV6_SRC_1HOUR").unwrap())?, - ipv4_dst_1min: AyaHashMap::try_from(ebpf.take_map("IPV4_DST_1MIN").unwrap())?, - ipv4_dst_10min: AyaHashMap::try_from(ebpf.take_map("IPV4_DST_10MIN").unwrap())?, - ipv4_dst_1hour: AyaHashMap::try_from(ebpf.take_map("IPV4_DST_1HOUR").unwrap())?, - ipv6_dst_1min: AyaHashMap::try_from(ebpf.take_map("IPV6_DST_1MIN").unwrap())?, - ipv6_dst_10min: AyaHashMap::try_from(ebpf.take_map("IPV6_DST_10MIN").unwrap())?, - ipv6_dst_1hour: AyaHashMap::try_from(ebpf.take_map("IPV6_DST_1HOUR").unwrap())?, + ipv4_maps, + ipv6_maps, }; STATISTICS.get_or_init(|| RwLock::new(monitor)); info!("{}", SystemEntry::InitializeComplete); @@ -86,89 +111,72 @@ impl Statistics { } pub async fn cleanup_expired_flows() { - const ONE_MIN: u64 = 60 * 1_000_000_000; - const TEN_MIN: u64 = 10 * ONE_MIN; - const ONE_HOUR: u64 = 60 * ONE_MIN; - let mut monitor = Statistics::instance_mut().await; + let boot_time = System::boot_time().await; let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_nanos() as u64; - - Statistics::cleanup_map(&mut monitor.ipv4_src_1min, now, ONE_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv4_dst_1min, now, ONE_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv4_src_10min, now, TEN_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv4_dst_10min, now, TEN_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv4_src_1hour, now, ONE_HOUR).await; - Statistics::cleanup_map(&mut monitor.ipv4_dst_1hour, now, ONE_HOUR).await; - Statistics::cleanup_map(&mut monitor.ipv6_src_1min, now, ONE_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv6_dst_1min, now, ONE_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv6_src_10min, now, TEN_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv6_dst_10min, now, TEN_MIN).await; - Statistics::cleanup_map(&mut monitor.ipv6_src_1hour, now, ONE_HOUR).await; - Statistics::cleanup_map(&mut monitor.ipv6_dst_1hour, now, ONE_HOUR).await; + monitor + .ipv4_maps + .iter_mut() + .for_each(|((_, time_type), map)| map.cleanup(boot_time, now, time_type.duration())); + monitor + .ipv6_maps + .iter_mut() + .for_each(|((_, time_type), map)| map.cleanup(boot_time, now, time_type.duration())); } - async fn cleanup_map(map: &mut AyaHashMap, now: u64, window: u64) - where - K: Pod, - { - let boot_time = System::boot_time().await; - let expired_keys: Vec = map - .iter() - .filter_map(|result| { - if let Ok((key, stats)) = result { - if now - stats[2] - boot_time > window { - Some(key) - } else { - None - } - } else { - None - } - }) - .collect(); - for key in expired_keys { - let _ = map.remove(&key); - } - } - - pub async fn get_ipv4_flow_data(flow_type: FlowType) -> StdHashMap { + pub async fn get_ipv4_flow_data( + direction: Direction, + time_type: TimeType, + ) -> StdHashMap { let monitor = Statistics::instance().await; - let iter = match flow_type { - FlowType::Src1Min => monitor.ipv4_src_1min.iter(), - FlowType::Src10Min => monitor.ipv4_src_10min.iter(), - FlowType::Src1Hour => monitor.ipv4_src_1hour.iter(), - FlowType::Dst1Min => monitor.ipv4_dst_1min.iter(), - FlowType::Dst10Min => monitor.ipv4_dst_10min.iter(), - FlowType::Dst1Hour => monitor.ipv4_dst_1hour.iter(), - }; - iter.filter_map(Result::ok) - .map(|(key, value)| { - let ip = Ipv4Addr::from(key[0]); - let port = key[1] as u16; - (SocketAddrV4::new(ip, port), FlowStats::from(value)) - }) - .collect() + monitor + .ipv4_maps + .get(&(direction, time_type)) + .map(|map| map.get_map()) + .unwrap() } - pub async fn get_ipv6_flow_data(flow_type: FlowType) -> StdHashMap { + pub async fn get_ipv6_flow_data( + direction: Direction, + time_type: TimeType, + ) -> StdHashMap { let monitor = Statistics::instance().await; - let iter = match flow_type { - FlowType::Src1Min => monitor.ipv6_src_1min.iter(), - FlowType::Src10Min => monitor.ipv6_src_10min.iter(), - FlowType::Src1Hour => monitor.ipv6_src_1hour.iter(), - FlowType::Dst1Min => monitor.ipv6_dst_1min.iter(), - FlowType::Dst10Min => monitor.ipv6_dst_10min.iter(), - FlowType::Dst1Hour => monitor.ipv6_dst_1hour.iter(), - }; - iter.filter_map(Result::ok) - .map(|(key, value)| { - let ip = Ipv6Addr::from(key[0]); - let port = key[1] as u16; - (SocketAddrV6::new(ip, port, 0, 0), FlowStats::from(value)) - }) - .collect() + monitor + .ipv6_maps + .get(&(direction, time_type)) + .map(|map| map.get_map()) + .unwrap() + } +} + +struct FlowMap { + map: AyaHashMap, +} + +impl FlowMap { + fn get_map(&self) -> StdHashMap { + self.map + .iter() + .filter_map(Result::ok) + .map(|(key, value)| (key.into_native(), FlowStats::from(value))) + .collect() + } + + fn cleanup(&mut self, boot_time: u64, now: u64, window: u64) { + let expired_keys: Vec = self + .map + .iter() + .filter_map(|result| { + result + .ok() + .and_then(|(key, stats)| (now - stats[2] - boot_time > window).then_some(key)) + }) + .collect(); + expired_keys.iter().for_each(|key| { + let _ = self.map.remove(key); + }); } } diff --git a/net-guardia/src/model/flow_type.rs b/net-guardia/src/model/flow_type.rs deleted file mode 100644 index 48de016..0000000 --- a/net-guardia/src/model/flow_type.rs +++ /dev/null @@ -1,18 +0,0 @@ -use serde::{Deserialize, Serialize}; - -#[derive(Serialize, Deserialize, Debug, Copy, Clone)] -#[serde(rename_all = "lowercase")] -pub enum FlowType { - #[serde(rename = "src_1min")] - Src1Min, - #[serde(rename = "src_10min")] - Src10Min, - #[serde(rename = "src_1hour")] - Src1Hour, - #[serde(rename = "dst_1min")] - Dst1Min, - #[serde(rename = "dst_10min")] - Dst10Min, - #[serde(rename = "dst_1hour")] - Dst1Hour, -} diff --git a/net-guardia/src/model/ip_address.rs b/net-guardia/src/model/ip_address.rs index aa309b9..25e6fa9 100644 --- a/net-guardia/src/model/ip_address.rs +++ b/net-guardia/src/model/ip_address.rs @@ -1,6 +1,6 @@ -use net_guardia_common::model::ip_address::{IPv4, IPv6}; +use net_guardia_common::model::ip_address::{IPv4, IPv6, EbpfAddrPortV4, EbpfAddrPortV6}; use std::hash::Hash; -use std::net::{Ipv4Addr, Ipv6Addr}; +use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; pub trait IpAddressType: Copy { type Native: Eq + PartialEq + Hash; @@ -31,3 +31,33 @@ impl IpAddressType for IPv6 { native.into() } } + +pub trait SocketAddressType: Copy { + type Native: Eq + PartialEq + Hash; + fn into_native(self) -> Self::Native; + fn from_native(native: Self::Native) -> Self; +} + +impl SocketAddressType for EbpfAddrPortV4 { + type Native = SocketAddrV4; + + fn into_native(self) -> Self::Native { + SocketAddrV4::new(Ipv4Addr::from(self[0]), self[1] as u16) + } + + fn from_native(native: Self::Native) -> Self { + [(*native.ip()).into(), native.port() as u32] + } +} + +impl SocketAddressType for EbpfAddrPortV6 { + type Native = SocketAddrV6; + + fn into_native(self) -> Self::Native { + SocketAddrV6::new(Ipv6Addr::from(self[0]), self[1] as u16, 0, 0) + } + + fn from_native(native: Self::Native) -> Self { + [(*native.ip()).into(), native.port() as u128] + } +} diff --git a/net-guardia/src/model/mod.rs b/net-guardia/src/model/mod.rs index 76472d6..6d3d5aa 100644 --- a/net-guardia/src/model/mod.rs +++ b/net-guardia/src/model/mod.rs @@ -1,7 +1,7 @@ pub mod config; pub mod direction; pub mod flow_stats; -pub mod flow_type; pub mod http_method; pub mod ip_address; pub mod list_type; +pub mod time_type; diff --git a/net-guardia/src/model/time_type.rs b/net-guardia/src/model/time_type.rs new file mode 100644 index 0000000..83dce19 --- /dev/null +++ b/net-guardia/src/model/time_type.rs @@ -0,0 +1,19 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Hash)] +#[serde(rename_all = "lowercase")] +pub enum TimeType { + #[serde(rename = "1min")] + _1Min = 60 * 1_000_000_000, + #[serde(rename = "10min")] + _10Min = 600 * 1_000_000_000, + #[serde(rename = "1hour")] + _1Hour = 3600 * 1_000_000_000, +} + +impl TimeType { + #[inline] + pub fn duration(&self) -> u64 { + *self as u64 + } +} diff --git a/net-guardia/src/web/api/statistics.rs b/net-guardia/src/web/api/statistics.rs index 67c2ca7..10a3c77 100644 --- a/net-guardia/src/web/api/statistics.rs +++ b/net-guardia/src/web/api/statistics.rs @@ -1,122 +1,58 @@ use crate::core::statistics::Statistics; -use crate::model::flow_type::FlowType; use crate::web::utils::flow_websocket::{IPv4FlowWebSocket, IPv6FlowWebSocket}; use actix_web::{get, web, Error, HttpRequest, HttpResponse, Responder, Scope}; use actix_web_actors::ws::start; +use crate::model::direction::Direction; +use crate::model::time_type::TimeType; pub fn initialize() -> Scope { web::scope("/statistics") - .service(get_ipv4_src_1min) - .service(get_ipv4_src_10min) - .service(get_ipv4_src_1hour) - .service(get_ipv6_src_1min) - .service(get_ipv6_src_10min) - .service(get_ipv6_src_1hour) - .service(get_ipv4_dst_1min) - .service(get_ipv4_dst_10min) - .service(get_ipv4_dst_1hour) - .service(get_ipv6_dst_1min) - .service(get_ipv6_dst_10min) - .service(get_ipv6_dst_1hour) + .service(get_ipv4_flow) + .service(get_ipv6_flow) .service(websocket_ipv4) .service(websocket_ipv6) } -#[get("/get/ipv4/src/1min")] -async fn get_ipv4_src_1min() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Src1Min).await; +#[get("/get/ipv4/{direction}/{time_type}")] +async fn get_ipv4_flow(path: web::Path<(Direction, TimeType)>) -> impl Responder { + let (direction, time_type) = path.into_inner(); + let flow_data = Statistics::get_ipv4_flow_data(direction, time_type).await; HttpResponse::Ok().json(web::Json(flow_data)) } -#[get("/get/ipv4/src/10min")] -async fn get_ipv4_src_10min() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Src10Min).await; + +#[get("/get/ipv6/{direction}/{time_type}")] +async fn get_ipv6_flow(path: web::Path<(Direction, TimeType)>) -> impl Responder { + let (direction, time_type) = path.into_inner(); + let flow_data = Statistics::get_ipv6_flow_data(direction, time_type).await; HttpResponse::Ok().json(web::Json(flow_data)) } -#[get("/get/ipv4/src/1hour")] -async fn get_ipv4_src_1hour() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Src1Hour).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/src/1min")] -async fn get_ipv6_src_1min() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src1Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/src/10min")] -async fn get_ipv6_src_10min() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src10Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/src/1hour")] -async fn get_ipv6_src_1hour() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src1Hour).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv4/dst/1min")] -async fn get_ipv4_dst_1min() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Dst1Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv4/dst/10min")] -async fn get_ipv4_dst_10min() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Dst10Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv4/dst/1hour")] -async fn get_ipv4_dst_1hour() -> impl Responder { - let flow_data = Statistics::get_ipv4_flow_data(FlowType::Dst1Hour).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/dst/1min")] -async fn get_ipv6_dst_1min() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src1Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/dst/10min")] -async fn get_ipv6_dst_10min() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src10Min).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/get/ipv6/dst/1hour")] -async fn get_ipv6_dst_1hour() -> impl Responder { - let flow_data = Statistics::get_ipv6_flow_data(FlowType::Src1Hour).await; - HttpResponse::Ok().json(web::Json(flow_data)) -} - -#[get("/websocket/ipv4/{flow_type}")] +#[get("/websocket/ipv4/{direction}/{time_type}")] async fn websocket_ipv4( req: HttpRequest, stream: web::Payload, - path: web::Path, + path: web::Path<(Direction, TimeType)>, ) -> Result { - let flow_type = path.into_inner(); + let (direction, time_type) = path.into_inner(); let websocket = IPv4FlowWebSocket { - flow_type, + direction, + time_type, interval: None, }; start(websocket, &req, stream) } -#[get("/websocket/ipv6/{flow_type}")] +#[get("/websocket/ipv6/{direction}/{time_type}")] async fn websocket_ipv6( req: HttpRequest, stream: web::Payload, - path: web::Path, + path: web::Path<(Direction, TimeType)>, ) -> Result { - let flow_type = path.into_inner(); + let (direction, time_type) = path.into_inner(); let websocket = IPv6FlowWebSocket { - flow_type, + direction, + time_type, interval: None, }; start(websocket, &req, stream) diff --git a/net-guardia/src/web/utils/flow_websocket.rs b/net-guardia/src/web/utils/flow_websocket.rs index b4f6cb2..71d7422 100644 --- a/net-guardia/src/web/utils/flow_websocket.rs +++ b/net-guardia/src/web/utils/flow_websocket.rs @@ -1,12 +1,14 @@ use crate::core::config_manager::ConfigManager; use crate::core::statistics::Statistics; -use crate::model::flow_type::FlowType; use actix::prelude::*; use actix_web_actors::ws; use std::time::Duration; +use crate::model::direction::Direction; +use crate::model::time_type::TimeType; pub struct IPv4FlowWebSocket { - pub flow_type: FlowType, + pub direction: Direction, + pub time_type: TimeType, pub interval: Option, } @@ -17,8 +19,9 @@ impl Actor for IPv4FlowWebSocket { let config = ConfigManager::now_blocking(); let refresh_interval = Duration::from_secs(config.refresh_interval); let interval = ctx.run_interval(refresh_interval, |actor, ctx| { - let flow_type = actor.flow_type.clone(); - let future = async move { Statistics::get_ipv4_flow_data(flow_type).await }; + let direction = actor.direction.clone(); + let time_type = actor.time_type.clone(); + let future = async move { Statistics::get_ipv4_flow_data(direction, time_type).await }; ctx.wait(future.into_actor(actor).map(|flow_data, _, ctx| { if let Ok(json) = serde_json::to_string(&flow_data) { ctx.text(json); @@ -50,7 +53,8 @@ impl StreamHandler> for IPv4FlowWebSocket } pub struct IPv6FlowWebSocket { - pub flow_type: FlowType, + pub direction: Direction, + pub time_type: TimeType, pub interval: Option, } @@ -61,8 +65,9 @@ impl Actor for IPv6FlowWebSocket { let config = ConfigManager::now_blocking(); let refresh_interval = Duration::from_secs(config.refresh_interval); let interval = ctx.run_interval(refresh_interval, |actor, ctx| { - let flow_type = actor.flow_type.clone(); - let future = async move { Statistics::get_ipv6_flow_data(flow_type).await }; + let direction = actor.direction.clone(); + let time_type = actor.time_type.clone(); + let future = async move { Statistics::get_ipv6_flow_data(direction, time_type).await }; ctx.wait(future.into_actor(actor).map(|flow_data, _, ctx| { if let Ok(json) = serde_json::to_string(&flow_data) { ctx.text(json);