DaLaw2/NetGuardia
1
0
Fork 0
mirror of https://github.com/DaLaw2/NetGuardia.git synced 2026-08-24 14:10:28 +09:00
Code Issues Packages Projects Releases Wiki Activity

wip: remove singleton pattern

Browse Source
This commit is contained in:
DaLaw2 2025-09-13 10:30:50 +00:00
parent c61485bc30
commit bdcec973ab
33 changed files with 1081 additions and 1670 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
Cargo.lock generated
View File

@ -627,6 +627,19 @@ dependencies = [
"cfg-if",
]
[[package]]
name = "crossbeam"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8"
dependencies = [
"crossbeam-channel",
"crossbeam-deque",
"crossbeam-epoch",
"crossbeam-queue",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-channel"
version = "0.5.14"
@ -636,6 +649,34 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-deque"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
dependencies = [
"crossbeam-epoch",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-epoch"
version = "0.9.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-queue"
version = "0.3.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.21"
@ -1253,6 +1294,7 @@ dependencies = [
"aya-log",
"cargo_metadata",
"common",
"crossbeam",
"dotenvy",
"futures-util",
"libc",

2
config.toml
View File

@ -4,4 +4,4 @@ egress_ifindex = "enp2s0f1" # nic name
management_ifindex = "enp4s0" # nic name
alert_path = "/tmp/alert"
http_server_bind_port = 8080 # port
refresh_interval = 1 # seconds
refresh_interval = 5 # seconds

1
net-guardia/Cargo.toml
View File

@ -12,6 +12,7 @@ actix-cors = "0.7.1"
actix-web = "4.11.0"
aya = { workspace = true }
aya-log = { workspace = true }
crossbeam = "0.8.4"
libc = { workspace = true }
mime_guess = "2.0.5"
rust-embed = "8.7.2"

60
net-guardia/src/core/app_config.rs
View File

@ -1,60 +0,0 @@
use std::fs;
use std::sync::OnceLock;
use std::sync::RwLock as SyncRwLock;
use macros::log;
use tokio::sync::RwLock as AsyncRwLock;
use crate::model::config::{Config, ConfigTable};
use crate::model::error::system::SystemError;
use crate::model::error::Error;
use crate::model::log::system::SystemLog;
static SYNC_CONFIG: OnceLock<SyncRwLock<Config>> = OnceLock::new();
static ASYNC_CONFIG: OnceLock<AsyncRwLock<Config>> = OnceLock::new();
pub struct AppConfig;
impl AppConfig {
pub async fn initialization() -> Result<(), Error> {
log!(SystemLog::Initializing);
let config = Self::load_config()?;
SYNC_CONFIG.get_or_init(|| SyncRwLock::new(config.clone()));
ASYNC_CONFIG.get_or_init(move || AsyncRwLock::new(config));
log!(SystemLog::InitializeComplete);
Ok(())
}
fn load_config() -> Result<Config, Error> {
let toml_string = fs::read_to_string("./config.toml").map_err(SystemError::ConfigNotFound)?;
let config_table = toml::from_str::<ConfigTable>(&toml_string).map_err(|_| SystemError::InvalidConfig)?;
let config = config_table.config;
if !Self::validate(&config) {
Err(SystemError::InvalidConfig)?
} else {
Ok(config)
}
}
pub fn now_blocking() -> Config {
// Initialization has been ensured
let once_lock = SYNC_CONFIG.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.read().unwrap().clone()
}
pub async fn now() -> Config {
// Initialization has been ensured
let once_lock = ASYNC_CONFIG.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.read().await.clone()
}
fn validate(config: &Config) -> bool {
Self::validate_second(config.refresh_interval)
}
fn validate_second(second: u64) -> bool {
second <= 3600
}
}

222
net-guardia/src/core/control/access_control.rs
View File

@ -1,222 +0,0 @@
use std::collections::HashMap as StdHashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::OnceLock;
use aya::maps::{HashMap as AyaHashMap, MapData};
use aya::Pod;
use common::define::setting::MAX_RULES_PORT;
use common::model::ip_address::{IPv4, IPv6, Port};
use macros::log;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::core::system::System;
use crate::model::direction::FlowDirection;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::IntoNative;
use crate::model::list_type::ListType;
use crate::model::log::system::SystemLog;
use crate::utils::ip_address::convert_ports_to_vec;
static ACCESS_CONTROL: OnceLock<RwLock<AccessControl>> = OnceLock::new();
pub struct AccessControl {
ipv4_maps: StdHashMap<(FlowDirection, ListType), AccessMap<IPv4>>,
ipv6_maps: StdHashMap<(FlowDirection, ListType), AccessMap<IPv6>>,
}
impl AccessControl {
const MAP_CONFIGS: [((FlowDirection, ListType), (&'static str, &'static str)); 4] = [
(
(FlowDirection::Source, ListType::White),
("IPV4_SRC_WHITELIST", "IPV6_SRC_WHITELIST"),
),
(
(FlowDirection::Source, ListType::Black),
("IPV4_SRC_BLACKLIST", "IPV6_SRC_BLACKLIST"),
),
(
(FlowDirection::Destination, ListType::White),
("IPV4_DST_WHITELIST", "IPV6_DST_WHITELIST"),
),
(
(FlowDirection::Destination, ListType::Black),
("IPV4_DST_BLACKLIST", "IPV6_DST_BLACKLIST"),
),
];
pub async fn initialize() -> Result<(), Error> {
log!(SystemLog::Initializing);
let mut system = System::instance_mut().await;
let ebpf = &mut system.ingress_ebpf;
let mut ipv4_maps = StdHashMap::new();
let mut ipv6_maps = StdHashMap::new();
for (key, (ipv4_name, ipv6_name)) in Self::MAP_CONFIGS {
let ipv4_map = ebpf.take_map(ipv4_name).ok_or(EbpfError::MapNotFound)?;
let ipv6_map = ebpf.take_map(ipv6_name).ok_or(EbpfError::MapNotFound)?;
ipv4_maps.insert(
key,
AccessMap {
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
AccessMap {
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
}
ACCESS_CONTROL.get_or_init(|| RwLock::new(AccessControl { ipv4_maps, ipv6_maps }));
log!(SystemLog::InitializeComplete);
Ok(())
}
#[inline(always)]
pub async fn instance() -> RwLockReadGuard<'static, AccessControl> {
let once_lock = ACCESS_CONTROL.get().unwrap();
once_lock.read().await
}
#[inline(always)]
pub async fn instance_mut() -> RwLockWriteGuard<'static, AccessControl> {
let once_lock = ACCESS_CONTROL.get().unwrap();
once_lock.write().await
}
pub async fn get_ipv4_list(direction: FlowDirection, list_type: ListType) -> StdHashMap<Ipv4Addr, Vec<Port>> {
let access_list = AccessControl::instance().await;
access_list
.ipv4_maps
.get(&(direction, list_type))
.map(|map| map.get_list())
.unwrap()
}
pub async fn get_ipv6_list(direction: FlowDirection, list_type: ListType) -> StdHashMap<Ipv6Addr, Vec<Port>> {
let access_list = AccessControl::instance().await;
access_list
.ipv6_maps
.get(&(direction, list_type))
.map(|map| map.get_list())
.unwrap()
}
pub async fn add_ipv4_list(
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
let map = access_list.ipv4_maps.get_mut(&(direction, list_type)).unwrap();
map.add(ip, port)
}
pub async fn add_ipv6_list(
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
let map = access_list.ipv6_maps.get_mut(&(direction, list_type)).unwrap();
map.add(ip, port)
}
pub async fn remove_ipv4_list(
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
let map = access_list.ipv4_maps.get_mut(&(direction, list_type)).unwrap();
map.remove(ip, port)
}
pub async fn remove_ipv6_list(
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
let map = access_list.ipv6_maps.get_mut(&(direction, list_type)).unwrap();
map.remove(ip, port)
}
}
struct AccessMap<T> {
map: AyaHashMap<MapData, T, [Port; MAX_RULES_PORT]>,
}
impl<T: IntoNative + Pod> AccessMap<T> {
fn get_list(&self) -> StdHashMap<T::Native, Vec<Port>> {
self.map
.iter()
.filter_map(Result::ok)
.map(|(key, value)| (key.into_native(), convert_ports_to_vec(value)))
.collect()
}
fn add(&mut self, ip: T, port: Port) -> Result<(), Error> {
let mut new_ports = [0_u16; MAX_RULES_PORT];
if port == 0 {
new_ports[0] = 0;
} else if let Ok(ports) = self.map.get(&ip, 0) {
if ports[0] == 0 {
return Ok(());
}
let mut index = None;
for (i, &value) in ports.iter().enumerate() {
if value == port {
return Ok(());
}
if index.is_none() && value == 0 {
index = Some(i);
}
}
if index.is_none() {
Err(EbpfError::RuleReachLimit)?;
}
new_ports.copy_from_slice(&ports);
new_ports[index.unwrap()] = port;
} else {
new_ports[0] = port;
}
self.map
.insert(ip, new_ports, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
fn remove(&mut self, ip: T, port: Port) -> Result<(), Error> {
if let Ok(mut ports) = self.map.get(&ip, 0) {
if port == 0 {
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
return Ok(());
}
if let Some(index) = ports.iter().position(|&x| x == port) {
for i in index..(MAX_RULES_PORT - 1) {
ports[i] = ports[i + 1];
}
ports[MAX_RULES_PORT - 1] = 0;
if ports[0] == 0 {
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
} else {
self.map.insert(ip, ports, 0).map_err(EbpfError::MapOperationError)?;
}
}
Ok(())
} else {
Err(EbpfError::IpDoesNotExist)?
}
}
}

15
net-guardia/src/core/control/mod.rs
View File

@ -1,15 +0,0 @@
use crate::core::control::access_control::AccessControl;
use crate::core::control::service::Service;
use crate::model::error::Error;
pub mod access_control;
pub mod service;
pub struct Control;
impl Control {
pub async fn initialize() -> Result<(), Error> {
AccessControl::initialize().await?;
Service::initialize().await
}
}

422
net-guardia/src/core/control/service.rs
View File

@ -1,422 +0,0 @@
use std::collections::HashMap as StdHashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::OnceLock;
use aya::maps::{Array as AyaArray, HashMap as AyaHashMap, MapData};
use common::model::http_method::{HttpMethod, HttpMethodBitmap};
use common::model::ip_address::{AddrPortV4, AddrPortV6, IPv4, IPv6};
use common::model::placeholder::PlaceHolder;
use macros::log;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::core::system::System;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::IntoNative;
use crate::model::log::system::SystemLog;
static SERVICE: OnceLock<RwLock<Service>> = OnceLock::new();
pub struct Service {
ipv4_http_service: AyaHashMap<MapData, AddrPortV4, HttpMethodBitmap>,
ipv6_http_service: AyaHashMap<MapData, AddrPortV6, HttpMethodBitmap>,
ssh_white_list_enable: AyaArray<MapData, PlaceHolder>,
ipv4_ssh_service: AyaHashMap<MapData, AddrPortV4, PlaceHolder>,
ipv6_ssh_service: AyaHashMap<MapData, AddrPortV6, PlaceHolder>,
ipv4_ssh_white_list: AyaHashMap<MapData, IPv4, PlaceHolder>,
ipv6_ssh_white_list: AyaHashMap<MapData, IPv6, PlaceHolder>,
ipv4_ssh_black_list: AyaHashMap<MapData, IPv4, PlaceHolder>,
ipv6_ssh_black_list: AyaHashMap<MapData, IPv6, PlaceHolder>,
}
impl Service {
pub async fn initialize() -> Result<(), Error> {
log!(SystemLog::Initializing);
let mut system = System::instance_mut().await;
let ebpf = &mut system.ingress_ebpf;
let mut service = Service {
ipv4_http_service: AyaHashMap::try_from(ebpf.take_map("IPV4_HTTP_SERVICE").ok_or(EbpfError::MapNotFound)?)
.map_err(EbpfError::MapOperationError)?,
ipv6_http_service: AyaHashMap::try_from(ebpf.take_map("IPV6_HTTP_SERVICE").ok_or(EbpfError::MapNotFound)?)
.map_err(EbpfError::MapOperationError)?,
ssh_white_list_enable: AyaArray::try_from(
ebpf.take_map("SSH_WHITE_LIST_ENABLE").ok_or(EbpfError::MapNotFound)?,
)
.map_err(EbpfError::MapOperationError)?,
ipv4_ssh_service: AyaHashMap::try_from(ebpf.take_map("IPV4_SSH_SERVICE").ok_or(EbpfError::MapNotFound)?)
.map_err(EbpfError::MapOperationError)?,
ipv6_ssh_service: AyaHashMap::try_from(ebpf.take_map("IPV6_SSH_SERVICE").ok_or(EbpfError::MapNotFound)?)
.map_err(EbpfError::MapOperationError)?,
ipv4_ssh_white_list: AyaHashMap::try_from(
ebpf.take_map("IPV4_SSH_WHITE_LIST").ok_or(EbpfError::MapNotFound)?,
)
.map_err(EbpfError::MapOperationError)?,
ipv6_ssh_white_list: AyaHashMap::try_from(
ebpf.take_map("IPV6_SSH_WHITE_LIST").ok_or(EbpfError::MapNotFound)?,
)
.map_err(EbpfError::MapOperationError)?,
ipv4_ssh_black_list: AyaHashMap::try_from(
ebpf.take_map("IPV4_SSH_BLACK_LIST").ok_or(EbpfError::MapNotFound)?,
)
.map_err(EbpfError::MapOperationError)?,
ipv6_ssh_black_list: AyaHashMap::try_from(
ebpf.take_map("IPV6_SSH_BLACK_LIST").ok_or(EbpfError::MapNotFound)?,
)
.map_err(EbpfError::MapOperationError)?,
};
service
.ssh_white_list_enable
.set(0, 0_u8, 0)
.map_err(EbpfError::MapOperationError)?;
SERVICE.get_or_init(|| RwLock::new(service));
log!(SystemLog::InitializeComplete);
Ok(())
}
#[inline(always)]
pub async fn instance() -> RwLockReadGuard<'static, Service> {
let once_lock = SERVICE.get().unwrap();
once_lock.read().await
}
#[inline(always)]
pub async fn instance_mut() -> RwLockWriteGuard<'static, Service> {
let once_lock = SERVICE.get().unwrap();
once_lock.write().await
}
pub async fn get_ipv4_http_service() -> StdHashMap<SocketAddrV4, Vec<HttpMethod>> {
let service = Service::instance().await;
service
.ipv4_http_service
.iter()
.filter_map(Result::ok)
.map(|(key, value)| {
let address = Ipv4Addr::from(key.ip);
let port = key.port;
(SocketAddrV4::new(address, port), HttpMethod::convert_from_bitmap(value))
})
.collect()
}
pub async fn get_ipv6_http_service() -> StdHashMap<SocketAddrV6, Vec<HttpMethod>> {
let service = Service::instance().await;
service
.ipv6_http_service
.iter()
.filter_map(Result::ok)
.map(|(key, value)| {
let address = Ipv6Addr::from(key.ip);
let port = key.port;
(
SocketAddrV6::new(address, port, 0, 0),
HttpMethod::convert_from_bitmap(value),
)
})
.collect()
}
pub async fn add_ipv4_http_service(address: SocketAddrV4, http_method: Vec<HttpMethod>) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV4::new(ip, port);
let ebpf_method = HttpMethod::convert_to_bitmap(http_method);
let mut service = Service::instance_mut().await;
service
.ipv4_http_service
.insert(addr_port, ebpf_method, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_http_service(address: SocketAddrV6, http_method: Vec<HttpMethod>) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV6::new(ip, port);
let ebpf_method = HttpMethod::convert_to_bitmap(http_method);
let mut service = Service::instance_mut().await;
service
.ipv6_http_service
.insert(addr_port, ebpf_method, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_http_service(
address: SocketAddrV4,
removed_http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV4::new(ip, port);
let mut service = Service::instance_mut().await;
if let Ok(current_http_method) = service.ipv4_http_service.get(&addr_port, 0) {
let mut http_method = HttpMethod::convert_from_bitmap(current_http_method);
http_method.retain(|method| !removed_http_method.contains(method));
if http_method.is_empty() {
service
.ipv4_http_service
.remove(&addr_port)
.map_err(EbpfError::MapOperationError)?;
} else {
let new_http_method = HttpMethod::convert_to_bitmap(http_method);
service
.ipv4_http_service
.insert(&addr_port, new_http_method, 0)
.map_err(EbpfError::MapOperationError)?;
}
Ok(())
} else {
Err(EbpfError::IpDoesNotExist)?
}
}
pub async fn remove_ipv6_http_service(
address: SocketAddrV6,
removed_http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV6::new(ip, port);
let mut service = Service::instance_mut().await;
if let Ok(current_http_method) = service.ipv6_http_service.get(&addr_port, 0) {
let mut http_method = HttpMethod::convert_from_bitmap(current_http_method);
http_method.retain(|method| !removed_http_method.contains(method));
if http_method.is_empty() {
service
.ipv6_http_service
.remove(&addr_port)
.map_err(EbpfError::MapOperationError)?;
} else {
let new_http_method = HttpMethod::convert_to_bitmap(http_method);
service
.ipv6_http_service
.insert(&addr_port, new_http_method, 0)
.map_err(EbpfError::MapOperationError)?;
}
Ok(())
} else {
Err(EbpfError::IpDoesNotExist)?
}
}
pub async fn is_ssh_white_list_enable() -> bool {
let service = Service::instance().await;
match service.ssh_white_list_enable.get(&0, 0) {
Ok(status) => {
if status == 0 {
false
} else {
true
}
}
Err(_) => false,
}
}
pub async fn enable_ssh_white_list() -> Result<(), Error> {
let mut service = Service::instance_mut().await;
service
.ssh_white_list_enable
.set(0, 1_u8, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
pub async fn disable_ssh_white_list() -> Result<(), Error> {
let mut service = Service::instance_mut().await;
service
.ssh_white_list_enable
.set(0, 0_u8, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
pub async fn get_ipv4_ssh_service() -> Vec<SocketAddrV4> {
let service = Service::instance().await;
service
.ipv4_ssh_service
.keys()
.filter_map(Result::ok)
.map(|key| key.into_native())
.collect()
}
pub async fn get_ipv6_ssh_service() -> Vec<SocketAddrV6> {
let service = Service::instance().await;
service
.ipv6_ssh_service
.keys()
.filter_map(Result::ok)
.map(|key| key.into_native())
.collect()
}
pub async fn add_ipv4_ssh_service(address: SocketAddrV4) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV4::new(ip, port);
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_service
.insert(&addr_port, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_service(address: SocketAddrV6) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV6::new(ip, port);
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_service
.insert(&addr_port, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_service(address: SocketAddrV4) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV4::new(ip, port);
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_service
.remove(&addr_port)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_service(address: SocketAddrV6) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV6::new(ip, port);
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_service
.remove(&addr_port)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn get_ipv4_ssh_white_list() -> Vec<Ipv4Addr> {
let service = Service::instance().await;
service
.ipv4_ssh_white_list
.keys()
.filter_map(Result::ok)
.map(|key| Ipv4Addr::from(key))
.collect()
}
pub async fn get_ipv6_ssh_white_list() -> Vec<Ipv6Addr> {
let service = Service::instance().await;
service
.ipv6_ssh_white_list
.keys()
.filter_map(Result::ok)
.map(|key| Ipv6Addr::from(key))
.collect()
}
pub async fn add_ipv4_ssh_white_list(ip: Ipv4Addr) -> Result<(), Error> {
let ip: u32 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_white_list
.insert(ip, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_white_list(ip: Ipv6Addr) -> Result<(), Error> {
let ip: u128 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_white_list
.insert(ip, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_white_list(ip: Ipv4Addr) -> Result<(), Error> {
let ip: u32 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_white_list
.remove(&ip)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_white_list(ip: Ipv6Addr) -> Result<(), Error> {
let ip: u128 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_white_list
.remove(&ip)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn get_ipv4_ssh_black_list() -> Vec<Ipv4Addr> {
let service = Service::instance().await;
service
.ipv4_ssh_black_list
.keys()
.filter_map(Result::ok)
.map(|key| Ipv4Addr::from(key))
.collect()
}
pub async fn get_ipv6_ssh_black_list() -> Vec<Ipv6Addr> {
let service = Service::instance().await;
service
.ipv6_ssh_black_list
.keys()
.filter_map(Result::ok)
.map(|key| Ipv6Addr::from(key))
.collect()
}
pub async fn add_ipv4_ssh_black_list(ip: Ipv4Addr) -> Result<(), Error> {
let ip: u32 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_black_list
.insert(ip, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_black_list(ip: Ipv6Addr) -> Result<(), Error> {
let ip: u128 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_black_list
.insert(ip, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_black_list(ip: Ipv4Addr) -> Result<(), Error> {
let ip: u32 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv4_ssh_black_list
.remove(&ip)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_black_list(ip: Ipv6Addr) -> Result<(), Error> {
let ip: u128 = ip.into();
let mut service = Service::instance_mut().await;
service
.ipv6_ssh_black_list
.remove(&ip)
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
}

206
net-guardia/src/core/ebpf/access_control.rs Normal file
View File

@ -0,0 +1,206 @@
use std::collections::HashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use aya::maps::{HashMap as AyaHashMap, MapData};
use aya::{Ebpf, Pod};
use common::define::setting::MAX_RULES_PORT;
use common::model::ip_address::{IPv4, IPv6, Port};
use tokio::sync::RwLock;
use crate::model::direction::FlowDirection;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::NativeConvert;
use crate::model::list_type::ListType;
use crate::utils::ip_address::convert_ports_to_vec;
pub struct AccessControl {
ipv4_src_whitelist: RwLock<MapWrapper<IPv4>>,
ipv4_src_blacklist: RwLock<MapWrapper<IPv4>>,
ipv4_dst_whitelist: RwLock<MapWrapper<IPv4>>,
ipv4_dst_blacklist: RwLock<MapWrapper<IPv4>>,
ipv6_src_whitelist: RwLock<MapWrapper<IPv6>>,
ipv6_src_blacklist: RwLock<MapWrapper<IPv6>>,
ipv6_dst_whitelist: RwLock<MapWrapper<IPv6>>,
ipv6_dst_blacklist: RwLock<MapWrapper<IPv6>>,
}
impl AccessControl {
pub fn new(ebpf: &mut Ebpf) -> Result<Self, Error> {
let access_control = Self {
ipv4_src_whitelist: RwLock::new(MapWrapper::new(ebpf, "IPV4_SRC_WHITELIST")?),
ipv4_src_blacklist: RwLock::new(MapWrapper::new(ebpf, "IPV4_SRC_BLACKLIST")?),
ipv4_dst_whitelist: RwLock::new(MapWrapper::new(ebpf, "IPV4_DST_WHITELIST")?),
ipv4_dst_blacklist: RwLock::new(MapWrapper::new(ebpf, "IPV4_DST_BLACKLIST")?),
ipv6_src_whitelist: RwLock::new(MapWrapper::new(ebpf, "IPV6_SRC_WHITELIST")?),
ipv6_src_blacklist: RwLock::new(MapWrapper::new(ebpf, "IPV6_SRC_BLACKLIST")?),
ipv6_dst_whitelist: RwLock::new(MapWrapper::new(ebpf, "IPV6_DST_WHITELIST")?),
ipv6_dst_blacklist: RwLock::new(MapWrapper::new(ebpf, "IPV6_DST_BLACKLIST")?),
};
Ok(access_control)
}
pub async fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>> {
let map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.read().await,
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.read().await,
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.read().await,
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.read().await,
};
map_wrapper.get_list()
}
pub async fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>> {
let map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.read().await,
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.read().await,
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.read().await,
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.read().await,
};
map_wrapper.get_list()
}
pub async fn add_ipv4_list(
&self,
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write().await,
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write().await,
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write().await,
};
map_wrapper.add(ip, port)
}
pub async fn add_ipv6_list(
&self,
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write().await,
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write().await,
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write().await,
};
map_wrapper.add(ip, port)
}
pub async fn remove_ipv4_list(
&self,
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write().await,
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write().await,
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write().await,
};
map_wrapper.remove(ip, port)
}
pub async fn remove_ipv6_list(
&self,
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write().await,
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write().await,
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write().await,
};
map_wrapper.remove(ip, port)
}
}
struct MapWrapper<T> {
map: AyaHashMap<MapData, T, [Port; MAX_RULES_PORT]>,
}
impl<T: NativeConvert + Pod> MapWrapper<T> {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn get_list(&self) -> HashMap<T::Native, Vec<Port>> {
self.map
.iter()
.filter_map(Result::ok)
.map(|(key, value)| (key.into_native(), convert_ports_to_vec(value)))
.collect()
}
fn add(&mut self, ip: T, port: Port) -> Result<(), Error> {
let mut new_ports = [0_u16; MAX_RULES_PORT];
if port == 0 {
new_ports[0] = 0;
} else if let Ok(ports) = self.map.get(&ip, 0) {
if ports[0] == 0 {
return Ok(());
}
let mut index = None;
for (i, &value) in ports.iter().enumerate() {
if value == port {
return Ok(());
}
if index.is_none() && value == 0 {
index = Some(i);
}
}
if index.is_none() {
Err(EbpfError::RuleReachLimit)?;
}
new_ports.copy_from_slice(&ports);
new_ports[index.unwrap()] = port;
} else {
new_ports[0] = port;
}
self.map
.insert(ip, new_ports, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
fn remove(&mut self, ip: T, port: Port) -> Result<(), Error> {
if let Ok(mut ports) = self.map.get(&ip, 0) {
if port == 0 {
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
return Ok(());
}
if let Some(index) = ports.iter().position(|&x| x == port) {
for i in index..(MAX_RULES_PORT - 1) {
ports[i] = ports[i + 1];
}
ports[MAX_RULES_PORT - 1] = 0;
if ports[0] == 0 {
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
} else {
self.map.insert(ip, ports, 0).map_err(EbpfError::MapOperationError)?;
}
}
Ok(())
} else {
Err(EbpfError::IpDoesNotExist)?
}
}
}

59
net-guardia/src/core/ebpf/mod.rs Normal file
View File

@ -0,0 +1,59 @@
pub mod access_control;
pub mod ring_buffer;
pub mod service;
pub mod statistics;
use std::sync::Arc;
use aya::Ebpf;
use crossbeam::queue::SegQueue;
use macros::log;
use tokio::sync::oneshot;
use crate::core::ebpf::access_control::AccessControl;
use crate::core::ebpf::ring_buffer::RingBuffer;
use crate::core::ebpf::service::Service;
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::error::system::SystemError;
use crate::model::error::Error;
pub struct EbpfServices {
#[allow(dead_code)]
pub ring_buffer: Arc<RingBuffer>,
pub access_control: Arc<AccessControl>,
pub service: Arc<Service>,
pub statistics: Arc<Statistics>,
shutdowns: SegQueue<oneshot::Sender<()>>,
}
impl EbpfServices {
pub fn new(app_config: Arc<AppConfig>, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result<Self, Error> {
let ring_buffer = RingBuffer::new()?;
let access_control = AccessControl::new(ingress_ebpf)?;
let service = Service::new(ingress_ebpf)?;
let statistics = Statistics::new(app_config, ingress_ebpf, egress_ebpf)?;
let ebpf_services = Self {
ring_buffer: Arc::new(ring_buffer),
access_control: Arc::new(access_control),
service: Arc::new(service),
statistics: Arc::new(statistics),
shutdowns: SegQueue::new(),
};
Ok(ebpf_services)
}
pub async fn run(self: Arc<Self>) {
let statistics = self.statistics.clone();
let statistics_shutdown = statistics.run().await;
self.shutdowns.push(statistics_shutdown);
}
pub fn terminate(self: Arc<Self>) {
while let Some(shutdown) = self.shutdowns.pop() {
if shutdown.send(()).is_err() {
log!(SystemError::ShutdownSignalFailed);
}
}
}
}

11
net-guardia/src/core/ebpf/ring_buffer.rs Normal file
View File

@ -0,0 +1,11 @@
use crate::model::error::Error;
#[allow(dead_code)]
pub struct RingBuffer {}
impl RingBuffer {
pub fn new() -> Result<Self, Error> {
let ring_buffer = Self {};
Ok(ring_buffer)
}
}

332
net-guardia/src/core/ebpf/service.rs Normal file
View File

@ -0,0 +1,332 @@
use std::collections::HashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use aya::maps::{Array as AyaArray, HashMap as AyaHashMap, MapData};
use aya::{Ebpf, Pod};
use common::model::http_method::{HttpMethod, HttpMethodBitmap};
use common::model::ip_address::{AddrPortV4, AddrPortV6, IPv4, IPv6};
use common::model::placeholder::PlaceHolder;
use tokio::sync::RwLock;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::NativeConvert;
pub struct Service {
ipv4_http_service: RwLock<HttpServiceWrapper<AddrPortV4>>,
ipv6_http_service: RwLock<HttpServiceWrapper<AddrPortV6>>,
ssh_white_list_enable: RwLock<WhiteListControl>,
ipv4_ssh_service: RwLock<SshServiceWrapper<AddrPortV4>>,
ipv6_ssh_service: RwLock<SshServiceWrapper<AddrPortV6>>,
ipv4_ssh_white_list: RwLock<SshListWrapper<IPv4>>,
ipv6_ssh_white_list: RwLock<SshListWrapper<IPv6>>,
ipv4_ssh_black_list: RwLock<SshListWrapper<IPv4>>,
ipv6_ssh_black_list: RwLock<SshListWrapper<IPv6>>,
}
impl Service {
pub fn new(ebpf: &mut Ebpf) -> Result<Self, Error> {
let service = Self {
ipv4_http_service: RwLock::new(HttpServiceWrapper::new(ebpf, "IPV4_HTTP_SERVICE")?),
ipv6_http_service: RwLock::new(HttpServiceWrapper::new(ebpf, "IPV6_HTTP_SERVICE")?),
ssh_white_list_enable: RwLock::new(WhiteListControl::new(ebpf, "SSH_WHITE_LIST_ENABLE")?),
ipv4_ssh_service: RwLock::new(SshServiceWrapper::new(ebpf, "IPV4_SSH_SERVICE")?),
ipv6_ssh_service: RwLock::new(SshServiceWrapper::new(ebpf, "IPV6_SSH_SERVICE")?),
ipv4_ssh_white_list: RwLock::new(SshListWrapper::new(ebpf, "IPV4_SSH_WHITE_LIST")?),
ipv6_ssh_white_list: RwLock::new(SshListWrapper::new(ebpf, "IPV6_SSH_WHITE_LIST")?),
ipv4_ssh_black_list: RwLock::new(SshListWrapper::new(ebpf, "IPV4_SSH_BLACK_LIST")?),
ipv6_ssh_black_list: RwLock::new(SshListWrapper::new(ebpf, "IPV6_SSH_BLACK_LIST")?),
};
Ok(service)
}
pub async fn get_ipv4_http_service(&self) -> HashMap<SocketAddrV4, Vec<HttpMethod>> {
self.ipv4_http_service.read().await.get_http_method()
}
pub async fn get_ipv6_http_service(&self) -> HashMap<SocketAddrV6, Vec<HttpMethod>> {
self.ipv6_http_service.read().await.get_http_method()
}
pub async fn add_ipv4_http_service(
&self,
address: SocketAddrV4,
http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
self.ipv4_http_service
.write()
.await
.add_http_service(address, http_method)
}
pub async fn add_ipv6_http_service(
&self,
address: SocketAddrV6,
http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
self.ipv6_http_service
.write()
.await
.add_http_service(address, http_method)
}
pub async fn remove_ipv4_http_service(
&self,
address: SocketAddrV4,
removed_http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
self.ipv4_http_service
.write()
.await
.remove_http_service(address, removed_http_method)
}
pub async fn remove_ipv6_http_service(
&self,
address: SocketAddrV6,
removed_http_method: Vec<HttpMethod>,
) -> Result<(), Error> {
self.ipv6_http_service
.write()
.await
.remove_http_service(address, removed_http_method)
}
pub async fn is_ssh_white_list_enable(&self) -> bool {
self.ssh_white_list_enable.read().await.is_white_list_enable()
}
pub async fn enable_ssh_white_list(&self) -> Result<(), Error> {
self.ssh_white_list_enable.write().await.enable_white_list()
}
pub async fn disable_ssh_white_list(&self) -> Result<(), Error> {
self.ssh_white_list_enable.write().await.disable_white_list()
}
pub async fn get_ipv4_ssh_service(&self) -> Vec<SocketAddrV4> {
self.ipv4_ssh_service.read().await.get_ssh_service()
}
pub async fn get_ipv6_ssh_service(&self) -> Vec<SocketAddrV6> {
self.ipv6_ssh_service.read().await.get_ssh_service()
}
pub async fn add_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
self.ipv4_ssh_service.write().await.add_ssh_service(address)
}
pub async fn add_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
self.ipv6_ssh_service.write().await.add_ssh_service(address)
}
pub async fn remove_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
self.ipv4_ssh_service.write().await.remove_ssh_service(address)
}
pub async fn remove_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
self.ipv6_ssh_service.write().await.remove_ssh_service(address)
}
pub async fn get_ipv4_ssh_white_list(&self) -> Vec<Ipv4Addr> {
self.ipv4_ssh_white_list.read().await.get_list()
}
pub async fn get_ipv6_ssh_white_list(&self) -> Vec<Ipv6Addr> {
self.ipv6_ssh_white_list.read().await.get_list()
}
pub async fn add_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
self.ipv4_ssh_white_list.write().await.add_list(ip)
}
pub async fn add_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
self.ipv6_ssh_white_list.write().await.add_list(ip)
}
pub async fn remove_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
self.ipv4_ssh_white_list.write().await.remove_list(ip)
}
pub async fn remove_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
self.ipv6_ssh_white_list.write().await.remove_list(ip)
}
pub async fn get_ipv4_ssh_black_list(&self) -> Vec<Ipv4Addr> {
self.ipv4_ssh_black_list.read().await.get_list()
}
pub async fn get_ipv6_ssh_black_list(&self) -> Vec<Ipv6Addr> {
self.ipv6_ssh_black_list.read().await.get_list()
}
pub async fn add_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
self.ipv4_ssh_black_list.write().await.add_list(ip)
}
pub async fn add_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
self.ipv6_ssh_black_list.write().await.add_list(ip)
}
pub async fn remove_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
self.ipv4_ssh_black_list.write().await.remove_list(ip)
}
pub async fn remove_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
self.ipv6_ssh_black_list.write().await.remove_list(ip)
}
}
struct WhiteListControl {
map: AyaArray<MapData, PlaceHolder>,
}
impl WhiteListControl {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaArray::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn is_white_list_enable(&self) -> bool {
match self.map.get(&0, 0) {
Ok(status) => {
if status == 0 {
false
} else {
true
}
}
Err(_) => false,
}
}
fn enable_white_list(&mut self) -> Result<(), Error> {
self.map.set(0, 1_u8, 0).map_err(EbpfError::MapOperationError)?;
Ok(())
}
fn disable_white_list(&mut self) -> Result<(), Error> {
self.map.set(0, 0_u8, 0).map_err(EbpfError::MapOperationError)?;
Ok(())
}
}
struct HttpServiceWrapper<T> {
map: AyaHashMap<MapData, T, HttpMethodBitmap>,
}
impl<T: NativeConvert + Pod> HttpServiceWrapper<T> {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn get_http_method(&self) -> HashMap<T::Native, Vec<HttpMethod>> {
self.map
.iter()
.filter_map(Result::ok)
.map(|(key, value)| {
let address = key.into_native();
(address, HttpMethod::convert_from_bitmap(value))
})
.collect()
}
fn add_http_service(&mut self, address: T::Native, http_method: Vec<HttpMethod>) -> Result<(), Error> {
let address = T::from_native(address);
let ebpf_method = HttpMethod::convert_to_bitmap(http_method);
self.map
.insert(address, ebpf_method, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
fn remove_http_service(&mut self, address: T::Native, removed_http_method: Vec<HttpMethod>) -> Result<(), Error> {
let address = T::from_native(address);
if let Ok(current_http_method) = self.map.get(&address, 0) {
let mut http_method = HttpMethod::convert_from_bitmap(current_http_method);
http_method.retain(|method| !removed_http_method.contains(method));
if http_method.is_empty() {
self.map.remove(&address).map_err(EbpfError::MapOperationError)?;
} else {
let new_http_method = HttpMethod::convert_to_bitmap(http_method);
self.map
.insert(&address, new_http_method, 0)
.map_err(EbpfError::MapOperationError)?;
}
Ok(())
} else {
Err(EbpfError::IpDoesNotExist)?
}
}
}
struct SshServiceWrapper<T> {
map: AyaHashMap<MapData, T, PlaceHolder>,
}
impl<T: NativeConvert + Pod> SshServiceWrapper<T> {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn get_ssh_service(&self) -> Vec<T::Native> {
self.map
.keys()
.filter_map(Result::ok)
.map(|key| key.into_native())
.collect()
}
fn add_ssh_service(&mut self, address: T::Native) -> Result<(), Error> {
let address = T::from_native(address);
self.map
.insert(address, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
fn remove_ssh_service(&mut self, address: T::Native) -> Result<(), Error> {
let address = T::from_native(address);
self.map.remove(&address).map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
}
struct SshListWrapper<T> {
map: AyaHashMap<MapData, T, PlaceHolder>,
}
impl<T: NativeConvert + Pod> SshListWrapper<T> {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn get_list(&self) -> Vec<T::Native> {
self.map
.keys()
.filter_map(Result::ok)
.map(|key| key.into_native())
.collect()
}
fn add_list(&mut self, address: T::Native) -> Result<(), Error> {
let address = T::from_native(address);
self.map
.insert(address, 0_u8, 0)
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
fn remove_list(&mut self, address: T::Native) -> Result<(), Error> {
let address = T::from_native(address);
self.map.remove(&address).map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
}

169
net-guardia/src/core/statistics.rs → net-guardia/src/core/ebpf/statistics.rs
View File

@ -1,28 +1,28 @@
use std::collections::HashMap as StdHashMap;
use std::collections::HashMap;
use std::net::{SocketAddrV4, SocketAddrV6};
use std::sync::OnceLock;
use std::sync::Arc;
use aya::maps::{HashMap as AyaHashMap, MapData};
use aya::Pod;
use aya::{Ebpf, Pod};
use common::model::flow_stats::FlowStats;
use common::model::ip_address::{AddrPortV4, AddrPortV6};
use macros::log;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use tokio::select;
use tokio::sync::{oneshot, RwLock};
use tokio::time::{sleep, Duration};
use crate::core::system::System;
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::IntoNative;
use crate::model::log::system::SystemLog;
use crate::model::ip_address::NativeConvert;
use crate::model::time_type::TimeType;
static STATISTICS: OnceLock<RwLock<Statistics>> = OnceLock::new();
use crate::utils::boot_time::boot_time;
pub struct Statistics {
terminate: bool,
ipv4_maps: StdHashMap<(Direction, FlowDirection, TimeType), FlowMap<AddrPortV4>>,
ipv6_maps: StdHashMap<(Direction, FlowDirection, TimeType), FlowMap<AddrPortV6>>,
app_config: Arc<AppConfig>,
boot_time: u64,
ipv4_maps: HashMap<(Direction, FlowDirection, TimeType), RwLock<FlowMap<AddrPortV4>>>,
ipv6_maps: HashMap<(Direction, FlowDirection, TimeType), RwLock<FlowMap<AddrPortV6>>>,
}
impl Statistics {
@ -80,126 +80,89 @@ impl Statistics {
),
];
pub async fn initialize() -> Result<(), Error> {
log!(SystemLog::Initializing);
let mut system = System::instance_mut().await;
let mut ipv4_maps = StdHashMap::new();
let mut ipv6_maps = StdHashMap::new();
let ingress_ebpf = &mut system.ingress_ebpf;
pub fn new(
app_config: Arc<AppConfig>,
ingress_ebpf: &mut Ebpf,
egress_ebpf: &mut Ebpf,
) -> Result<Statistics, Error> {
let boot_time = boot_time();
let mut ipv4_maps = HashMap::new();
let mut ipv6_maps = HashMap::new();
for (key, (ipv4_name, ipv6_name)) in Self::INGRESS_MAPS {
let ipv4_map = ingress_ebpf.take_map(ipv4_name).ok_or(EbpfError::MapNotFound)?;
let ipv6_map = ingress_ebpf.take_map(ipv6_name).ok_or(EbpfError::MapNotFound)?;
ipv4_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv4_maps.insert(key, RwLock::new(FlowMap::new(ingress_ebpf, ipv4_name)?));
ipv6_maps.insert(key, RwLock::new(FlowMap::new(ingress_ebpf, ipv6_name)?));
}
let egress_ebpf = &mut system.egress_ebpf;
for (key, (ipv4_name, ipv6_name)) in Self::EGRESS_MAPS {
let ipv4_map = egress_ebpf.take_map(ipv4_name).ok_or(EbpfError::MapNotFound)?;
let ipv6_map = egress_ebpf.take_map(ipv6_name).ok_or(EbpfError::MapNotFound)?;
ipv4_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv4_maps.insert(key, RwLock::new(FlowMap::new(egress_ebpf, ipv4_name)?));
ipv6_maps.insert(key, RwLock::new(FlowMap::new(egress_ebpf, ipv6_name)?));
}
let statistics = Statistics {
terminate: false,
app_config,
boot_time,
ipv4_maps,
ipv6_maps,
};
STATISTICS.get_or_init(|| RwLock::new(statistics));
log!(SystemLog::InitializeComplete);
Ok(())
Ok(statistics)
}
#[inline(always)]
pub async fn instance() -> RwLockReadGuard<'static, Statistics> {
// Initialization has been ensured
let once_lock = STATISTICS.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.read().await
}
#[inline(always)]
pub async fn instance_mut() -> RwLockWriteGuard<'static, Statistics> {
// Initialization has been ensured
let once_lock = STATISTICS.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.write().await
}
pub async fn run() {
tokio::spawn(async {
pub async fn run(self: Arc<Self>) -> oneshot::Sender<()> {
let refresh_interval = self.app_config.refresh_interval;
let (sender, receiver) = oneshot::channel();
tokio::spawn(async move {
let mut receiver = receiver;
loop {
Statistics::cleanup_expired_flows().await;
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
select! {
biased;
_ = &mut receiver => break,
_ = sleep(Duration::from_secs(refresh_interval)) => {
self.cleanup_expired_flows().await;
},
}
}
});
sender
}
pub async fn terminate() {
let mut statistics = Statistics::instance_mut().await;
statistics.terminate = true;
}
pub async fn cleanup_expired_flows() {
let mut statistics = Statistics::instance_mut().await;
let boot_time = System::boot_time().await;
pub async fn cleanup_expired_flows(self: &Arc<Self>) {
let boot_time = self.boot_time;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos() as u64;
statistics
.ipv4_maps
.iter_mut()
.for_each(|((_, _, time_type), map)| map.cleanup(boot_time, now, time_type.duration()));
statistics
.ipv6_maps
.iter_mut()
.for_each(|((_, _, time_type), map)| map.cleanup(boot_time, now, time_type.duration()));
for ((_, _, time_type), map) in self.ipv4_maps.iter() {
map.write().await.cleanup(boot_time, now, time_type.duration())
}
for ((_, _, time_type), map) in self.ipv6_maps.iter() {
map.write().await.cleanup(boot_time, now, time_type.duration())
}
}
pub async fn get_ipv4_flow_data(
&self,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> StdHashMap<SocketAddrV4, FlowStats> {
let statistics = Statistics::instance().await;
statistics
.ipv4_maps
) -> HashMap<SocketAddrV4, FlowStats> {
self.ipv4_maps
.get(&(direction, flow_direction, time_type))
.map(|map| map.get_map())
.unwrap()
.write()
.await
.get_map()
}
pub async fn get_ipv6_flow_data(
&self,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> StdHashMap<SocketAddrV6, FlowStats> {
let statistics = Statistics::instance().await;
statistics
.ipv6_maps
) -> HashMap<SocketAddrV6, FlowStats> {
self.ipv6_maps
.get(&(direction, flow_direction, time_type))
.map(|map| map.get_map())
.unwrap()
.write()
.await
.get_map()
}
}
@ -207,8 +170,14 @@ struct FlowMap<T> {
map: AyaHashMap<MapData, T, FlowStats>,
}
impl<T: IntoNative + Pod> FlowMap<T> {
fn get_map(&self) -> StdHashMap<T::Native, FlowStats> {
impl<T: NativeConvert + Pod> FlowMap<T> {
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
Ok(Self { map })
}
fn get_map(&self) -> HashMap<T::Native, FlowStats> {
self.map
.iter()
.filter_map(Result::ok)

345
net-guardia/src/core/health.rs
View File

@ -1,345 +0,0 @@
use std::sync::OnceLock;
use std::time::Duration;
use macros::log;
use sysinfo::{Components, Networks, System};
use tokio::sync::{broadcast, mpsc, RwLock, RwLockReadGuard, RwLockWriteGuard};
use tokio::time::interval;
use crate::core::app_config::AppConfig;
use crate::model::error::misc::MiscError;
use crate::model::healthy::*;
static SYSTEM_HEALTH_INSTANCE: OnceLock<RwLock<SystemHealth>> = OnceLock::new();
pub struct SystemHealth {
system: System,
networks: Networks,
components: Components,
broadcast_tx: broadcast::Sender<SystemHealthMetrics>,
shutdown_tx: mpsc::UnboundedSender<()>,
ingress_interface: String,
egress_interface: String,
management_interface: String,
}
impl SystemHealth {
pub async fn initialize(monitoring_interval: Duration) {
let (broadcast_tx, _) = broadcast::channel(100);
let (shutdown_tx, shutdown_rx) = mpsc::unbounded_channel();
let config = AppConfig::now().await;
let system_health = SystemHealth {
system: System::new_all(),
networks: Networks::new_with_refreshed_list(),
components: Components::new_with_refreshed_list(),
broadcast_tx: broadcast_tx.clone(),
shutdown_tx,
ingress_interface: config.ingress_ifindex.clone(),
egress_interface: config.egress_ifindex.clone(),
management_interface: config.management_ifindex.clone(),
};
SYSTEM_HEALTH_INSTANCE.get_or_init(|| RwLock::new(system_health));
let ingress_interface = config.ingress_ifindex;
let egress_interface = config.egress_ifindex;
let management_interface = config.management_ifindex;
tokio::spawn(async move {
Self::monitoring_loop(
broadcast_tx,
shutdown_rx,
monitoring_interval,
ingress_interface,
egress_interface,
management_interface,
)
.await;
});
}
async fn monitoring_loop(
broadcast_tx: broadcast::Sender<SystemHealthMetrics>,
mut shutdown_rx: mpsc::UnboundedReceiver<()>,
monitoring_interval: Duration,
ingress_interface: String,
egress_interface: String,
management_interface: String,
) {
let mut system = System::new_all();
let mut networks = Networks::new_with_refreshed_list();
let mut components = Components::new_with_refreshed_list();
let mut interval_timer = interval(monitoring_interval);
loop {
tokio::select! {
_ = shutdown_rx.recv() => {
break;
}
_ = interval_timer.tick() => {
system.refresh_all();
networks.refresh(true);
components.refresh(true);
let metrics = Self::collect_metrics(
&system,
&networks,
&components,
&ingress_interface,
&egress_interface,
&management_interface,
);
if broadcast_tx.receiver_count() > 0 {
if let Err(err) = broadcast_tx.send(metrics) {
log!(MiscError::SendMessageError(err))
}
}
}
}
}
}
pub async fn instance() -> RwLockReadGuard<'static, SystemHealth> {
let instance = SYSTEM_HEALTH_INSTANCE.get().unwrap();
instance.read().await
}
pub async fn instance_mut() -> RwLockWriteGuard<'static, SystemHealth> {
let instance = SYSTEM_HEALTH_INSTANCE.get().unwrap();
instance.write().await
}
fn collect_metrics(
system: &System,
networks: &Networks,
components: &Components,
ingress_interface: &str,
egress_interface: &str,
management_interface: &str,
) -> SystemHealthMetrics {
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let boot_time = System::boot_time();
let uptime_seconds = timestamp - boot_time;
let system_info = Self::collect_system_info(system);
let cpu_details = Self::collect_cpu_details(system);
let memory_usage = MemoryUsage {
total: system.total_memory(),
used: system.used_memory(),
available: system.available_memory(),
usage_percent: (system.used_memory() as f32 / system.total_memory() as f32) * 100.0,
swap_total: system.total_swap(),
swap_used: system.used_swap(),
};
let network_stats =
Self::collect_configured_network_stats(networks, ingress_interface, egress_interface, management_interface);
let load_average = System::load_average();
let load_average = if load_average.one != 0.0 || load_average.five != 0.0 || load_average.fifteen != 0.0 {
Some(LoadAverage {
one_minute: load_average.one,
five_minute: load_average.five,
fifteen_minute: load_average.fifteen,
})
} else {
None
};
let temperature = components
.iter()
.find(|component| {
let label = component.label().to_lowercase();
label.contains("cpu") || label.contains("core") || label.contains("processor")
})
.and_then(|component| component.temperature());
SystemHealthMetrics {
timestamp,
boot_time,
uptime_seconds,
system_info,
cpu_details,
memory_usage,
network_stats,
load_average,
temperature,
}
}
fn collect_system_info(system: &System) -> SystemInfo {
SystemInfo {
kernel_version: System::kernel_version(),
os_name: System::name(),
os_version: System::os_version(),
architecture: std::env::consts::ARCH.to_string(),
total_processes: system.processes().len(),
}
}
fn collect_cpu_details(system: &System) -> CpuDetails {
let cpus = system.cpus();
let cpu_usage = cpus.iter().map(|cpu| cpu.cpu_usage()).sum::<f32>() / cpus.len() as f32;
let cores: Vec<CpuCoreInfo> = cpus
.iter()
.enumerate()
.map(|(index, cpu)| CpuCoreInfo {
core_id: index,
usage_percent: cpu.cpu_usage(),
frequency: cpu.frequency(),
})
.collect();
let cpu_brand = cpus
.first()
.map(|cpu| cpu.brand().to_string())
.unwrap_or_else(|| "Unknown".to_string());
let avg_frequency = if !cores.is_empty() {
cores.iter().map(|core| core.frequency).sum::<u64>() / cores.len() as u64
} else {
0
};
CpuDetails {
cpu_brand,
core_count: cores.len(),
cpu_usage,
cpu_frequency: avg_frequency,
cores,
}
}
fn collect_configured_network_stats(
networks: &Networks,
ingress_interface: &str,
egress_interface: &str,
management_interface: &str,
) -> ConfiguredNetworkStats {
let create_network_stats = |interface_name: &str| -> Option<NetworkStats> {
networks.get(interface_name).map(|network| NetworkStats {
interface: interface_name.to_string(),
bytes_received: network.total_received(),
bytes_transmitted: network.total_transmitted(),
packets_received: network.total_packets_received(),
packets_transmitted: network.total_packets_transmitted(),
errors_received: network.total_errors_on_received(),
errors_transmitted: network.total_errors_on_transmitted(),
})
};
let ingress = create_network_stats(ingress_interface);
let egress = create_network_stats(egress_interface);
let management = create_network_stats(management_interface);
if ingress.is_none() {
log!(MiscError::NetworkInterfaceNotFound(ingress_interface));
}
if egress.is_none() {
log!(MiscError::NetworkInterfaceNotFound(egress_interface));
}
if management.is_none() {
log!(MiscError::NetworkInterfaceNotFound(management_interface));
}
ConfiguredNetworkStats {
ingress,
egress,
management,
}
}
pub async fn get_current_metrics() -> SystemHealthMetrics {
let mut instance = Self::instance_mut().await;
instance.system.refresh_all();
instance.networks.refresh(true);
instance.components.refresh(true);
Self::collect_metrics(
&instance.system,
&instance.networks,
&instance.components,
&instance.ingress_interface,
&instance.egress_interface,
&instance.management_interface,
)
}
pub async fn subscribe_to_metrics() -> broadcast::Receiver<SystemHealthMetrics> {
let instance = Self::instance().await;
instance.broadcast_tx.subscribe()
}
pub async fn shutdown() {
if let Ok(instance) = SYSTEM_HEALTH_INSTANCE.get().unwrap().try_read() {
let _ = instance.shutdown_tx.send(());
}
}
pub async fn is_system_healthy() -> SystemHealthStatus {
let metrics = Self::get_current_metrics().await;
let mut status = SystemHealthStatus {
overall_healthy: true,
issues: Vec::new(),
warnings: Vec::new(),
};
if metrics.cpu_details.cpu_usage > 90.0 {
status.overall_healthy = false;
status
.issues
.push(format!("High CPU usage: {:.1}%", metrics.cpu_details.cpu_usage));
} else if metrics.cpu_details.cpu_usage > 75.0 {
status
.warnings
.push(format!("Moderate CPU usage: {:.1}%", metrics.cpu_details.cpu_usage));
}
if metrics.memory_usage.usage_percent > 95.0 {
status.overall_healthy = false;
status.issues.push(format!(
"Critical memory usage: {:.1}%",
metrics.memory_usage.usage_percent
));
} else if metrics.memory_usage.usage_percent > 80.0 {
status
.warnings
.push(format!("High memory usage: {:.1}%", metrics.memory_usage.usage_percent));
}
if let Some(temp) = metrics.temperature {
if temp > 80.0 {
status.overall_healthy = false;
status.issues.push(format!("High CPU temperature: {:.1}°C", temp));
} else if temp > 70.0 {
status.warnings.push(format!("Elevated CPU temperature: {:.1}°C", temp));
}
}
if metrics.network_stats.ingress.is_none() {
status.overall_healthy = false;
status.issues.push("Ingress interface not available".to_string());
}
if metrics.network_stats.egress.is_none() {
status.overall_healthy = false;
status.issues.push("Egress interface not available".to_string());
}
if metrics.network_stats.management.is_none() {
status.warnings.push("Management interface not available".to_string());
}
status
}
}

39
net-guardia/src/core/infrastructure/app_config.rs Normal file
View File

@ -0,0 +1,39 @@
use std::fs;
use std::ops::Deref;
use crate::model::config::{Config, ConfigTable};
use crate::model::error::system::SystemError;
use crate::model::error::Error;
pub struct AppConfig {
pub config: Config,
}
impl AppConfig {
pub fn new() -> Result<Self, Error> {
let toml_string = fs::read_to_string("./config.toml").map_err(SystemError::ConfigNotFound)?;
let config_table = toml::from_str::<ConfigTable>(&toml_string).map_err(|_| SystemError::InvalidConfig)?;
let config = config_table.config;
if !Self::validate(&config) {
Err(SystemError::InvalidConfig)?
} else {
Ok(Self { config })
}
}
fn validate(config: &Config) -> bool {
Self::validate_second(config.refresh_interval)
}
fn validate_second(second: u64) -> bool {
second <= 3600
}
}
impl Deref for AppConfig {
type Target = Config;
fn deref(&self) -> &Self::Target {
&self.config
}
}

1
net-guardia/src/core/infrastructure/mod.rs Normal file
View File

@ -0,0 +1 @@
pub mod app_config;

6
net-guardia/src/core/mod.rs
View File

@ -1,5 +1,3 @@
pub mod app_config;
pub mod control;
pub mod health;
pub mod statistics;
pub mod ebpf;
pub mod infrastructure;
pub mod system;

227
net-guardia/src/core/system.rs
View File

@ -1,35 +1,31 @@
use std::sync::OnceLock;
use std::time::Duration;
use std::sync::Arc;
use actix_web::web::route;
use actix_web::{App, HttpServer};
use actix_web::{web, App, HttpServer};
use aya::maps::{MapData, ProgramArray};
use aya::programs::{Xdp, XdpFlags};
use aya::Ebpf;
use aya_log::EbpfLogger;
use macros::log;
use sysinfo::System as SystemInfo;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::core::app_config::AppConfig;
use crate::core::control::Control;
use crate::core::health::SystemHealth;
use crate::core::statistics::Statistics;
use crate::core::ebpf::EbpfServices;
use crate::core::infrastructure::app_config::AppConfig;
// use crate::core::infrastructure::system_health::SystemHealth;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::error::Error;
use crate::model::log::ebpf::EbpfLog;
use crate::model::log::system::SystemLog;
use crate::utils::logging::Logging;
use crate::web::api::{control, default, health, misc, statistics};
static SYSTEM: OnceLock<RwLock<System>> = OnceLock::new();
use crate::web::api::{control, default, misc};
pub struct System {
pub app_config: Arc<AppConfig>,
// pub health: Arc<SystemHealth>,
pub ebpf_services: Arc<EbpfServices>,
pub ingress_ebpf: Ebpf,
pub egress_ebpf: Ebpf,
pub boot_time: u64,
#[allow(dead_code)]
ingress_program_array: ProgramArray<MapData>,
#[allow(dead_code)]
@ -37,57 +33,99 @@ pub struct System {
}
impl System {
pub async fn initialize() -> Result<(), Error> {
Logging::initialize().await?;
log!(SystemLog::Initializing);
AppConfig::initialization().await?;
SystemHealth::initialize(Duration::from_secs(5)).await;
System::ebpf_initialize().await?;
Statistics::initialize().await?;
Control::initialize().await?;
log!(SystemLog::InitializeComplete);
Ok(())
}
async fn ebpf_initialize() -> Result<(), Error> {
let config = AppConfig::now().await;
let ingress_interface = config.ingress_ifindex;
let egress_interface = config.egress_ifindex;
let boot_time = SystemInfo::boot_time() * 1_000_000_000;
Self::set_memory_limit()?;
pub async fn new() -> Result<Self, Error> {
let (mut ingress_ebpf, ingress_program_array) = System::get_ingress_ebpf()?;
let ingress_program: &mut Xdp = ingress_ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
let (mut egress_ebpf, egress_program_array) = System::get_egress_ebpf()?;
let egress_program: &mut Xdp = egress_ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
ingress_program.load().map_err(EbpfError::LoadProgramFailed)?;
ingress_program
.attach(&ingress_interface, XdpFlags::default())
.map_err(EbpfError::AttachProgramFailed)?;
egress_program.load().map_err(EbpfError::LoadProgramFailed)?;
egress_program
.attach(&egress_interface, XdpFlags::default())
.map_err(EbpfError::AttachProgramFailed)?;
let app_config = Arc::new(AppConfig::new()?);
// let system_health = Arc::new(SystemHealth::new()?);
let ebpf_services = Arc::new(EbpfServices::new(
app_config.clone(),
&mut ingress_ebpf,
&mut egress_ebpf,
)?);
let system = System {
app_config,
ebpf_services,
ingress_ebpf,
egress_ebpf,
boot_time,
ingress_program_array,
egress_program_array,
};
SYSTEM.get_or_init(|| RwLock::new(system));
log!(EbpfLog::AttachProgramSuccess);
Ok(system)
}
pub async fn run(&mut self) -> Result<(), Error> {
let ebpf_services = self.ebpf_services.clone();
Logging::initialize().await?;
log!(SystemLog::Initializing);
self.attach_ebpf().await?;
log!(SystemLog::InitializeComplete);
ebpf_services.run().await;
self.run_http_server().await?;
Ok(())
}
pub async fn terminate(&self) -> Result<(), Error> {
let ebpf_services = self.ebpf_services.clone();
log!(SystemLog::Terminating);
ebpf_services.terminate();
log!(SystemLog::TerminateComplete);
Ok(())
}
pub async fn run_http_server(&self) -> Result<(), Error> {
let access_control = self.ebpf_services.access_control.clone();
let service = self.ebpf_services.service.clone();
let statistics = self.ebpf_services.statistics.clone();
let port = self.app_config.http_server_bind_port;
HttpServer::new(move || {
let cors = actix_cors::Cors::default()
.allow_any_origin()
.allow_any_method()
.allow_any_header()
.max_age(3600);
App::new()
.wrap(cors)
.app_data(web::Data::from(access_control.clone()))
.app_data(web::Data::from(service.clone()))
.app_data(web::Data::from(statistics.clone()))
.service(control::initialize())
.service(misc::initialize())
.default_service(route().to(default::default_route))
})
.bind(format!("0.0.0.0:{}", port))
.map_err(HttpError::BindPortError)?
.run()
.await
.map_err(HttpError::ServerPanic)?;
Ok(())
}
async fn attach_ebpf(&mut self) -> Result<(), Error> {
let config = self.app_config.config.clone();
let ingress_interface = config.ingress_ifindex;
let egress_interface = config.egress_ifindex;
Self::set_memory_limit()?;
let ingress_xdp: &mut Xdp = self
.ingress_ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
let egress_xdp: &mut Xdp = self
.egress_ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
ingress_xdp.load().map_err(EbpfError::LoadProgramFailed)?;
ingress_xdp
.attach(&ingress_interface, XdpFlags::default())
.map_err(EbpfError::AttachProgramFailed)?;
egress_xdp.load().map_err(EbpfError::LoadProgramFailed)?;
egress_xdp
.attach(&egress_interface, XdpFlags::default())
.map_err(EbpfError::AttachProgramFailed)?;
Ok(())
}
@ -119,18 +157,6 @@ impl System {
Ok((egress_ebpf, program_array))
}
fn set_memory_limit() -> Result<(), Error> {
let rlim = libc::rlimit {
rlim_cur: libc::RLIM_INFINITY,
rlim_max: libc::RLIM_INFINITY,
};
let ret = unsafe { libc::setrlimit(libc::RLIMIT_MEMLOCK, &rlim) };
if ret != 0 {
Err(MiscError::RamLimitUnlockError(ret))?
}
Ok(())
}
fn load_program(
ebpf: &mut Ebpf,
program_array: &mut ProgramArray<MapData>,
@ -148,60 +174,15 @@ impl System {
Ok(())
}
pub async fn run() -> Result<(), Error> {
log!(SystemLog::Online);
Statistics::run().await;
let config = AppConfig::now().await;
HttpServer::new(|| {
let cors = actix_cors::Cors::default()
.allow_any_origin()
.allow_any_method()
.allow_any_header()
.max_age(3600);
App::new()
.wrap(cors)
.service(statistics::initialize())
.service(control::initialize())
.service(misc::initialize())
.service(health::initialize())
.default_service(route().to(default::default_route))
})
.bind(format!("0.0.0.0:{}", config.http_server_bind_port))
.map_err(HttpError::BindPortError)?
.run()
.await
.map_err(HttpError::ServerPanic)?;
fn set_memory_limit() -> Result<(), Error> {
let rlim = libc::rlimit {
rlim_cur: libc::RLIM_INFINITY,
rlim_max: libc::RLIM_INFINITY,
};
let ret = unsafe { libc::setrlimit(libc::RLIMIT_MEMLOCK, &rlim) };
if ret != 0 {
Err(MiscError::RamLimitUnlockError(ret))?
}
Ok(())
}
pub async fn terminate() -> Result<(), Error> {
log!(SystemLog::Terminating);
Statistics::terminate().await;
SystemHealth::shutdown().await;
log!(SystemLog::TerminateComplete);
Ok(())
}
pub async fn instance() -> RwLockReadGuard<'static, System> {
// Initialization has been ensured
let once_lock = SYSTEM.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.read().await
}
pub async fn instance_mut() -> RwLockWriteGuard<'static, System> {
// Initialization has been ensured
let once_lock = SYSTEM.get().unwrap();
// There is no lock acquired multiple times, so this is safe
once_lock.write().await
}
pub async fn boot_time() -> u64 {
let system = System::instance().await;
system.boot_time
}
}

6
net-guardia/src/main.rs
View File

@ -8,8 +8,8 @@ use crate::model::error::Error;
#[actix_web::main]
async fn main() -> Result<(), Error> {
System::initialize().await?;
System::run().await?;
System::terminate().await?;
let mut system = System::new().await?;
system.run().await?;
system.terminate().await?;
Ok(())
}

81
net-guardia/src/model/healthy.rs
View File

@ -1,81 +0,0 @@
use serde::Serialize;
#[derive(Debug, Clone, Serialize)]
pub struct SystemHealthMetrics {
pub timestamp: u64,
pub boot_time: u64,
pub uptime_seconds: u64,
pub system_info: SystemInfo,
pub cpu_details: CpuDetails,
pub memory_usage: MemoryUsage,
pub network_stats: ConfiguredNetworkStats,
pub load_average: Option<LoadAverage>,
pub temperature: Option<f32>,
}
#[derive(Debug, Clone, Serialize)]
pub struct SystemInfo {
pub kernel_version: Option<String>,
pub os_name: Option<String>,
pub os_version: Option<String>,
pub architecture: String,
pub total_processes: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct CpuDetails {
pub cpu_brand: String,
pub core_count: usize,
pub cpu_usage: f32,
pub cpu_frequency: u64,
pub cores: Vec<CpuCoreInfo>,
}
#[derive(Debug, Clone, Serialize)]
pub struct CpuCoreInfo {
pub core_id: usize,
pub usage_percent: f32,
pub frequency: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct MemoryUsage {
pub total: u64,
pub used: u64,
pub available: u64,
pub usage_percent: f32,
pub swap_total: u64,
pub swap_used: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct ConfiguredNetworkStats {
pub ingress: Option<NetworkStats>,
pub egress: Option<NetworkStats>,
pub management: Option<NetworkStats>,
}
#[derive(Debug, Clone, Serialize)]
pub struct NetworkStats {
pub interface: String,
pub bytes_received: u64,
pub bytes_transmitted: u64,
pub packets_received: u64,
pub packets_transmitted: u64,
pub errors_received: u64,
pub errors_transmitted: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct LoadAverage {
pub one_minute: f64,
pub five_minute: f64,
pub fifteen_minute: f64,
}
#[derive(Debug, Clone, Serialize)]
pub struct SystemHealthStatus {
pub overall_healthy: bool,
pub issues: Vec<String>,
pub warnings: Vec<String>,
}

33
net-guardia/src/model/ip_address.rs
View File

@ -3,39 +3,62 @@ use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use common::model::ip_address::*;
pub trait IntoNative: Copy {
pub trait NativeConvert: Copy {
type Native: Eq + PartialEq + Hash;
fn into_native(self) -> Self::Native;
fn from_native(native: Self::Native) -> Self;
}
impl IntoNative for IPv4 {
impl NativeConvert for IPv4 {
type Native = Ipv4Addr;
fn into_native(self) -> Self::Native {
Ipv4Addr::from(self)
}
fn from_native(native: Self::Native) -> Self {
native.to_bits()
}
}
impl IntoNative for IPv6 {
impl NativeConvert for IPv6 {
type Native = Ipv6Addr;
fn into_native(self) -> Self::Native {
Ipv6Addr::from(self)
}
fn from_native(native: Self::Native) -> Self {
native.to_bits()
}
}
impl IntoNative for AddrPortV4 {
impl NativeConvert for AddrPortV4 {
type Native = SocketAddrV4;
fn into_native(self) -> Self::Native {
SocketAddrV4::new(Ipv4Addr::from(self.ip), self.port)
}
fn from_native(native: Self::Native) -> Self {
Self {
ip: (*native.ip()).to_bits(),
port: native.port(),
}
}
}
impl IntoNative for AddrPortV6 {
impl NativeConvert for AddrPortV6 {
type Native = SocketAddrV6;
fn into_native(self) -> Self::Native {
SocketAddrV6::new(Ipv6Addr::from(self.ip), self.port, 0, 0)
}
fn from_native(native: Self::Native) -> Self {
Self {
ip: (*native.ip()).to_bits(),
port: native.port(),
}
}
}

1
net-guardia/src/model/mod.rs
View File

@ -1,7 +1,6 @@
pub mod config;
pub mod direction;
pub mod error;
pub mod healthy;
pub mod ip_address;
pub mod list_type;
pub mod log;

5
net-guardia/src/utils/boot_time.rs Normal file
View File

@ -0,0 +1,5 @@
use sysinfo::System as SystemInfo;
pub fn boot_time() -> u64 {
SystemInfo::boot_time() * 1_000_000_000
}

1
net-guardia/src/utils/mod.rs
View File

@ -1,3 +1,4 @@
pub mod ip_address;
pub mod logging;
pub mod static_files;
pub mod boot_time;

38
net-guardia/src/web/api/control/access_control.rs
View File

@ -2,7 +2,7 @@ use std::net::{SocketAddrV4, SocketAddrV6};
use actix_web::{delete, get, put, web, HttpResponse, Responder, Scope};
use crate::core::control::access_control::AccessControl;
use crate::core::ebpf::access_control::AccessControl;
use crate::model::direction::FlowDirection;
use crate::model::list_type::ListType;
@ -17,34 +17,48 @@ pub fn initialize() -> Scope {
}
#[get("/ipv4/{direction}/{list_type}")]
async fn get_ipv4_list(path: web::Path<(FlowDirection, ListType)>) -> impl Responder {
async fn get_ipv4_list(
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let (direction, list_type) = path.into_inner();
let list = AccessControl::get_ipv4_list(direction, list_type).await;
let list = access_control.get_ipv4_list(direction, list_type).await;
HttpResponse::Ok().json(list)
}
#[get("/ipv6/{direction}/{list_type}")]
async fn get_ipv6_list(path: web::Path<(FlowDirection, ListType)>) -> impl Responder {
async fn get_ipv6_list(
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let (direction, list_type) = path.into_inner();
let list = AccessControl::get_ipv6_list(direction, list_type).await;
let list = access_control.get_ipv6_list(direction, list_type).await;
HttpResponse::Ok().json(list)
}
#[put("/ipv4/{direction}/{list_type}")]
async fn add_ipv4_list(address: web::Json<SocketAddrV4>, path: web::Path<(FlowDirection, ListType)>) -> impl Responder {
async fn add_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match AccessControl::add_ipv4_list(direction, list_type, address).await {
match access_control.add_ipv4_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[put("/ipv6/{direction}/{list_type}")]
async fn add_ipv6_list(address: web::Json<SocketAddrV6>, path: web::Path<(FlowDirection, ListType)>) -> impl Responder {
async fn add_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match AccessControl::add_ipv6_list(direction, list_type, address).await {
match access_control.add_ipv6_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
@ -54,10 +68,11 @@ async fn add_ipv6_list(address: web::Json<SocketAddrV6>, path: web::Path<(FlowDi
async fn remove_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match AccessControl::remove_ipv4_list(direction, list_type, address).await {
match access_control.remove_ipv4_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
@ -67,10 +82,11 @@ async fn remove_ipv4_list(
async fn remove_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match AccessControl::remove_ipv6_list(direction, list_type, address).await {
match access_control.remove_ipv6_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}

8
net-guardia/src/web/api/control/mod.rs
View File

@ -1,10 +1,12 @@
use actix_web::{web, Scope};
pub mod access_control;
pub mod service;
pub mod statistics;
use actix_web::{web, Scope};
pub fn initialize() -> Scope {
web::scope("/control")
web::scope("/ebpf")
.service(access_control::initialize())
.service(service::initialize())
.service(statistics::initialize())
}

122
net-guardia/src/web/api/control/service.rs
View File

@ -3,7 +3,7 @@ use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use actix_web::{delete, get, post, put, web, HttpResponse, Responder, Scope};
use common::model::http_method::HttpMethod;
use crate::core::control::service::Service;
use crate::core::ebpf::service::Service;
pub fn initialize() -> Scope {
web::scope("/service")
@ -37,202 +37,214 @@ pub fn initialize() -> Scope {
}
#[get("/ipv4/http_service")]
async fn get_ipv4_http_service() -> impl Responder {
let list = Service::get_ipv4_http_service().await;
async fn get_ipv4_http_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_http_service().await;
HttpResponse::Ok().json(web::Json(list))
}
#[get("/ipv6/http_service")]
async fn get_ipv6_http_service() -> impl Responder {
let list = Service::get_ipv6_http_service().await;
async fn get_ipv6_http_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_http_service().await;
HttpResponse::Ok().json(web::Json(list))
}
#[put("/ipv4/http_service")]
async fn add_ipv4_http_service(payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>) -> impl Responder {
async fn add_ipv4_http_service(
payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match Service::add_ipv4_http_service(addr, methods).await {
match service.add_ipv4_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[put("/ipv6/http_service")]
async fn add_ipv6_http_service(payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>) -> impl Responder {
async fn add_ipv6_http_service(
payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match Service::add_ipv6_http_service(addr, methods).await {
match service.add_ipv6_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv4/http_service")]
async fn remove_ipv4_http_service(payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>) -> impl Responder {
async fn remove_ipv4_http_service(
payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match Service::remove_ipv4_http_service(addr, methods).await {
match service.remove_ipv4_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv6/http_service")]
async fn remove_ipv6_http_service(payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>) -> impl Responder {
async fn remove_ipv6_http_service(
payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match Service::remove_ipv6_http_service(addr, methods).await {
match service.remove_ipv6_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[get("/ssh_white_list")]
async fn is_ssh_white_list_enable() -> impl Responder {
let enabled = Service::is_ssh_white_list_enable().await;
async fn is_ssh_white_list_enable(service: web::Data<Service>) -> impl Responder {
let enabled = service.is_ssh_white_list_enable().await;
HttpResponse::Ok().json(enabled)
}
#[post("/ssh_white_list/enable")]
async fn enable_ssh_white_list() -> impl Responder {
match Service::enable_ssh_white_list().await {
async fn enable_ssh_white_list(service: web::Data<Service>) -> impl Responder {
match service.enable_ssh_white_list().await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[post("/ssh_white_list/disable")]
async fn disable_ssh_white_list() -> impl Responder {
match Service::disable_ssh_white_list().await {
async fn disable_ssh_white_list(service: web::Data<Service>) -> impl Responder {
match service.disable_ssh_white_list().await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[get("/ipv4/ssh_service")]
async fn get_ipv4_ssh_service() -> impl Responder {
let list = Service::get_ipv4_ssh_service().await;
async fn get_ipv4_ssh_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_service().await;
HttpResponse::Ok().json(web::Json(list))
}
#[get("/ipv6/ssh_service")]
async fn get_ipv6_ssh_service() -> impl Responder {
let list = Service::get_ipv6_ssh_service().await;
async fn get_ipv6_ssh_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_service().await;
HttpResponse::Ok().json(web::Json(list))
}
#[put("/ipv4/ssh_service")]
async fn add_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>) -> impl Responder {
match Service::add_ipv4_ssh_service(ip_addr.into_inner()).await {
async fn add_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[put("/ipv6/ssh_service")]
async fn add_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>) -> impl Responder {
match Service::add_ipv6_ssh_service(ip_addr.into_inner()).await {
async fn add_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv4/ssh_service")]
async fn remove_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>) -> impl Responder {
match Service::remove_ipv4_ssh_service(ip_addr.into_inner()).await {
async fn remove_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv6/ssh_service")]
async fn remove_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>) -> impl Responder {
match Service::remove_ipv6_ssh_service(ip_addr.into_inner()).await {
async fn remove_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[get("/ipv4/ssh_white_list")]
async fn get_ipv4_ssh_white_list() -> impl Responder {
let list = Service::get_ipv4_ssh_white_list().await;
async fn get_ipv4_ssh_white_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_white_list().await;
HttpResponse::Ok().json(web::Json(list))
}
#[get("/ipv6/ssh_white_list")]
async fn get_ipv6_ssh_white_list() -> impl Responder {
let list = Service::get_ipv6_ssh_white_list().await;
async fn get_ipv6_ssh_white_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_white_list().await;
HttpResponse::Ok().json(web::Json(list))
}
#[put("/ipv4/ssh_white_list")]
async fn add_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>) -> impl Responder {
match Service::add_ipv4_ssh_white_list(ip_addr.into_inner()).await {
async fn add_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[put("/ipv6/ssh_white_list")]
async fn add_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>) -> impl Responder {
match Service::add_ipv6_ssh_white_list(ip_addr.into_inner()).await {
async fn add_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv4/ssh_white_list")]
async fn remove_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>) -> impl Responder {
match Service::remove_ipv4_ssh_white_list(ip_addr.into_inner()).await {
async fn remove_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv6/ssh_white_list")]
async fn remove_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>) -> impl Responder {
match Service::remove_ipv6_ssh_white_list(ip_addr.into_inner()).await {
async fn remove_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[get("/ipv4/ssh_black_list")]
async fn get_ipv4_ssh_black_list() -> impl Responder {
let list = Service::get_ipv4_ssh_black_list().await;
async fn get_ipv4_ssh_black_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_black_list().await;
HttpResponse::Ok().json(web::Json(list))
}
#[get("/ipv6/ssh_black_list")]
async fn get_ipv6_ssh_black_list() -> impl Responder {
let list = Service::get_ipv6_ssh_black_list().await;
async fn get_ipv6_ssh_black_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_black_list().await;
HttpResponse::Ok().json(web::Json(list))
}
#[put("/ipv4/ssh_black_list")]
async fn add_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>) -> impl Responder {
match Service::add_ipv4_ssh_black_list(ip_addr.into_inner()).await {
async fn add_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[put("/ipv6/ssh_black_list")]
async fn add_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>) -> impl Responder {
match Service::add_ipv6_ssh_black_list(ip_addr.into_inner()).await {
async fn add_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv4/ssh_black_list")]
async fn remove_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>) -> impl Responder {
match Service::remove_ipv4_ssh_black_list(ip_addr.into_inner()).await {
async fn remove_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}
}
#[delete("/ipv6/ssh_black_list")]
async fn remove_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>) -> impl Responder {
match Service::remove_ipv6_ssh_black_list(ip_addr.into_inner()).await {
async fn remove_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
}

31
net-guardia/src/web/api/statistics.rs → net-guardia/src/web/api/control/statistics.rs
View File

@ -1,6 +1,9 @@
use std::sync::Arc;
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use crate::core::statistics::Statistics;
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::time_type::TimeType;
use crate::web::websocket::flow_websocket;
@ -14,16 +17,26 @@ pub fn initialize() -> Scope {
}
#[get("/get/ipv4/{direction}/{flow_direction}/{time_type}")]
async fn get_ipv4_flow(path: web::Path<(Direction, FlowDirection, TimeType)>) -> impl Responder {
async fn get_ipv4_flow(
path: web::Path<(Direction, FlowDirection, TimeType)>,
statistics: web::Data<Arc<Statistics>>,
) -> impl Responder {
let (direction, flow_direction, time_type) = path.into_inner();
let flow_data = Statistics::get_ipv4_flow_data(direction, flow_direction, time_type).await;
let flow_data = statistics
.get_ipv4_flow_data(direction, flow_direction, time_type)
.await;
HttpResponse::Ok().json(web::Json(flow_data))
}
#[get("/get/ipv6/{direction}/{flow_direction}/{time_type}")]
async fn get_ipv6_flow(path: web::Path<(Direction, FlowDirection, TimeType)>) -> impl Responder {
async fn get_ipv6_flow(
path: web::Path<(Direction, FlowDirection, TimeType)>,
statistics: web::Data<Arc<Statistics>>,
) -> impl Responder {
let (direction, flow_direction, time_type) = path.into_inner();
let flow_data = Statistics::get_ipv6_flow_data(direction, flow_direction, time_type).await;
let flow_data = statistics
.get_ipv6_flow_data(direction, flow_direction, time_type)
.await;
HttpResponse::Ok().json(web::Json(flow_data))
}
@ -32,8 +45,10 @@ async fn websocket_ipv4(
req: HttpRequest,
stream: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> impl Responder {
match flow_websocket::websocket_ipv4_flow(req, stream, path).await {
match flow_websocket::websocket_ipv4_flow(req, stream, path, app_config, statistics).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}
@ -44,8 +59,10 @@ async fn websocket_ipv6(
req: HttpRequest,
stream: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> impl Responder {
match flow_websocket::websocket_ipv6_flow(req, stream, path).await {
match flow_websocket::websocket_ipv6_flow(req, stream, path, app_config, statistics).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}

32
net-guardia/src/web/api/health.rs
View File

@ -1,32 +0,0 @@
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use crate::core::health::SystemHealth;
use crate::web::websocket::health_websocket;
pub fn initialize() -> Scope {
web::scope("/health")
.service(get_current_metrics)
.service(get_health_status)
.service(websocket_metrics)
}
#[get("/metrics")]
async fn get_current_metrics() -> impl Responder {
let metrics = SystemHealth::get_current_metrics().await;
HttpResponse::Ok().json(metrics)
}
#[get("/status")]
async fn get_health_status() -> impl Responder {
let status = SystemHealth::is_system_healthy().await;
HttpResponse::Ok().json(status)
}
#[get("/websocket/system_health")]
async fn websocket_metrics(req: HttpRequest, body: web::Payload) -> impl Responder {
let broadcast_rx = SystemHealth::subscribe_to_metrics().await;
match health_websocket::websocket_system_health(req, body, broadcast_rx).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}
}

9
net-guardia/src/web/api/misc.rs
View File

@ -1,13 +1,14 @@
use actix_web::{get, web, HttpResponse, Responder, Scope};
use crate::core::system::System;
use crate::utils::boot_time::boot_time;
pub fn initialize() -> Scope {
web::scope("/misc").service(boot_time)
web::scope("/misc")
.service(get_boot_time)
}
#[get("/boot_time")]
async fn boot_time() -> impl Responder {
let boot_time = System::boot_time().await;
async fn get_boot_time() -> impl Responder {
let boot_time = boot_time();
HttpResponse::Ok().json(boot_time)
}

2
net-guardia/src/web/api/mod.rs
View File

@ -1,5 +1,3 @@
pub mod control;
pub mod default;
pub mod health;
pub mod misc;
pub mod statistics;

86
net-guardia/src/web/websocket/flow_websocket.rs
View File

@ -1,11 +1,13 @@
use std::sync::Arc;
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, MessageStream, Session};
use futures_util::StreamExt;
use macros::log;
use tokio::time::{interval, Duration};
use crate::core::app_config::AppConfig;
use crate::core::statistics::Statistics;
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
@ -15,12 +17,25 @@ pub async fn websocket_ipv4_flow(
req: HttpRequest,
body: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> Result<HttpResponse> {
let app_config = app_config.into_inner();
let statistics = statistics.into_inner();
let (direction, flow_direction, time_type) = path.into_inner();
let (response, session, msg_stream) = handle(&req, body)?;
actix_web::rt::spawn(async move {
handle_ipv4_flow_connection(session, msg_stream, direction, flow_direction, time_type).await;
handle_ipv4_flow_connection(
app_config,
statistics,
session,
msg_stream,
direction,
flow_direction,
time_type,
)
.await;
});
Ok(response)
@ -30,28 +45,42 @@ pub async fn websocket_ipv6_flow(
req: HttpRequest,
body: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> Result<HttpResponse> {
let app_config = app_config.into_inner();
let statistics = statistics.into_inner();
let (direction, flow_direction, time_type) = path.into_inner();
let (response, session, msg_stream) = handle(&req, body)?;
actix_web::rt::spawn(async move {
handle_ipv6_flow_connection(session, msg_stream, direction, flow_direction, time_type).await;
handle_ipv6_flow_connection(
app_config,
statistics,
session,
msg_stream,
direction,
flow_direction,
time_type,
)
.await;
});
Ok(response)
}
async fn handle_ipv4_flow_connection(
app_config: Arc<AppConfig>,
statistics: Arc<Statistics>,
mut session: Session,
mut msg_stream: MessageStream,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) {
let config = AppConfig::now_blocking();
let config = app_config.config.clone();
let refresh_interval = Duration::from_secs(config.refresh_interval);
let mut data_interval = interval(refresh_interval);
let mut ping_interval = interval(Duration::from_secs(30));
loop {
tokio::select! {
@ -61,15 +90,10 @@ async fn handle_ipv4_flow_connection(
}
},
_ = data_interval.tick() => {
if !send_ipv4_flow_data(&mut session, direction, flow_direction, time_type).await {
if !send_ipv4_flow_data(&statistics, &mut session, direction, flow_direction, time_type).await {
break;
}
},
_ = ping_interval.tick() => {
if session.ping(b"heartbeat").await.is_err() {
break;
}
}
}
}
@ -77,16 +101,17 @@ async fn handle_ipv4_flow_connection(
}
async fn handle_ipv6_flow_connection(
app_config: Arc<AppConfig>,
statistics: Arc<Statistics>,
mut session: Session,
mut msg_stream: MessageStream,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) {
let config = AppConfig::now_blocking();
let config = app_config.config.clone();
let refresh_interval = Duration::from_secs(config.refresh_interval);
let mut data_interval = interval(refresh_interval);
let mut ping_interval = interval(Duration::from_secs(30));
loop {
tokio::select! {
@ -96,15 +121,10 @@ async fn handle_ipv6_flow_connection(
}
},
_ = data_interval.tick() => {
if !send_ipv6_flow_data(&mut session, direction, flow_direction, time_type).await {
if !send_ipv6_flow_data(&statistics, &mut session, direction, flow_direction, time_type).await {
break;
}
},
_ = ping_interval.tick() => {
if session.ping(b"heartbeat").await.is_err() {
break;
}
}
}
}
@ -116,21 +136,7 @@ async fn handle_client_message(
msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
) -> bool {
match msg_result {
Some(Ok(Message::Text(text))) => {
let text = text.trim();
match text {
"ping" => session.text("pong").await.is_ok(),
_ => {
let error_msg = serde_json::json!({
"available_commands": ["ping"]
});
match serde_json::to_string(&error_msg) {
Ok(error_json) => session.text(error_json).await.is_ok(),
Err(_) => true,
}
}
}
}
Some(Ok(Message::Text(_))) => true,
Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
Some(Ok(Message::Close(reason))) => {
let _ = (session.clone()).close(reason).await;
@ -146,12 +152,15 @@ async fn handle_client_message(
}
async fn send_ipv4_flow_data(
statistics: &Arc<Statistics>,
session: &mut Session,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> bool {
let flow_data = Statistics::get_ipv4_flow_data(direction, flow_direction, time_type).await;
let flow_data = statistics
.get_ipv4_flow_data(direction, flow_direction, time_type)
.await;
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
@ -162,12 +171,15 @@ async fn send_ipv4_flow_data(
}
async fn send_ipv6_flow_data(
statistics: &Arc<Statistics>,
session: &mut Session,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> bool {
let flow_data = Statistics::get_ipv6_flow_data(direction, flow_direction, time_type).await;
let flow_data = statistics
.get_ipv6_flow_data(direction, flow_direction, time_type)
.await;
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {

136
net-guardia/src/web/websocket/health_websocket.rs
View File

@ -1,136 +0,0 @@
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, Session};
use futures_util::StreamExt;
use macros::log;
use tokio::sync::broadcast;
use tokio::time::{interval, Duration};
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::healthy::SystemHealthMetrics;
use crate::model::log::http::HttpLog;
pub async fn websocket_system_health(
req: HttpRequest,
body: web::Payload,
broadcast_rx: broadcast::Receiver<SystemHealthMetrics>,
) -> Result<HttpResponse> {
let (response, session, msg_stream) = handle(&req, body)?;
actix_web::rt::spawn(async move {
handle_websocket_connection(session, msg_stream, broadcast_rx).await;
});
Ok(response)
}
async fn handle_websocket_connection(
mut session: Session,
mut msg_stream: actix_ws::MessageStream,
mut broadcast_rx: broadcast::Receiver<SystemHealthMetrics>,
) {
let mut ping_interval = interval(Duration::from_secs(30));
loop {
tokio::select! {
msg_result = msg_stream.next() => {
if !handle_client_message(&mut session, msg_result).await {
break;
}
},
metrics_result = broadcast_rx.recv() => {
if !handle_broadcast_message(&mut session, metrics_result).await {
break;
}
},
_ = ping_interval.tick() => {
if session.ping(b"heartbeat").await.is_err() {
break;
}
}
}
}
let _ = session.close(None).await;
}
async fn handle_client_message(
session: &mut Session,
msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
) -> bool {
match msg_result {
Some(Ok(Message::Text(text))) => {
let text = text.trim();
match text {
"ping" => session.text("pong").await.is_ok(),
"get_current" => {
let current_metrics = crate::core::health::SystemHealth::get_current_metrics().await;
match serde_json::to_string(&current_metrics) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
true
}
}
}
_ => {
let error_msg = serde_json::json!({
"available_commands": ["ping", "get_current"]
});
match serde_json::to_string(&error_msg) {
Ok(error_json) => session.text(error_json).await.is_ok(),
Err(_) => true,
}
}
}
}
Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
Some(Ok(Message::Close(reason))) => {
let _ = (session.clone()).close(reason).await;
false
}
Some(Err(err)) => {
log!(HttpError::WebSocketError(err));
false
}
None => false,
_ => true,
}
}
async fn handle_broadcast_message(
session: &mut Session,
metrics_result: Result<SystemHealthMetrics, broadcast::error::RecvError>,
) -> bool {
match metrics_result {
Ok(metrics) => {
let message = serde_json::json!(metrics);
match serde_json::to_string(&message) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
true
}
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
log!(HttpLog::WebSocketLaged(skipped));
let lag_msg = serde_json::json!({
"message": format!("Connection lagged, skipped {} messages", skipped)
});
match serde_json::to_string(&lag_msg) {
Ok(json) => session.text(json).await.is_ok(),
Err(_) => true,
}
}
Err(broadcast::error::RecvError::Closed) => {
let close_msg = serde_json::json!({
"message": "Health monitoring stopped"
});
if let Ok(json) = serde_json::to_string(&close_msg) {
let _ = session.text(json).await;
}
false
}
}
}

1
net-guardia/src/web/websocket/mod.rs
View File

@ -1,2 +1 @@
pub mod flow_websocket;
pub mod health_websocket;