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: Redesigning error handling

Browse Source
This commit is contained in:
DaLaw2 2025-09-13 03:57:14 +00:00
parent b74e49284d
commit 4514e257ce
36 changed files with 690 additions and 600 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Cargo.lock generated
View File

@ -296,12 +296,6 @@ version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
[[package]]
name = "anyhow"
version = "1.0.99"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0674a1ddeecb70197781e945de4b3b8ffb61fa939a5597bcf48503737663100"
[[package]]
name = "assert_matches"
version = "1.5.0"
@ -1255,7 +1249,6 @@ dependencies = [
"actix-cors",
"actix-web",
"actix-ws",
"anyhow",
"aya",
"aya-log",
"cargo_metadata",
@ -1263,6 +1256,7 @@ dependencies = [
"dotenvy",
"futures-util",
"libc",
"macros",
"mime_guess",
"rust-embed",
"serde",

220
macros/src/error_enum.rs Normal file
View File

@ -0,0 +1,220 @@
use proc_macro::TokenStream;
use quote::quote;
use syn::parse::{Parse, ParseStream};
use syn::spanned::Spanned;
use syn::{Attribute, Error, Expr, Ident, LitStr, Result, Token, Type};
pub struct ErrorVariant {
pub attributes: Vec<Attribute>,
pub error_msg: LitStr,
pub name: Ident,
pub fields: Vec<(Ident, Type)>,
pub level: Expr,
}
impl ErrorVariant {
pub fn has_no_source(&self) -> bool {
self.attributes.iter().any(|attr| attr.path().is_ident("no_source"))
}
pub fn should_generate_constructor(&self, force_no_source: bool) -> bool {
if force_no_source || self.has_no_source() {
!self.fields.is_empty()
} else {
true
}
}
}
pub struct ErrorEnumInput {
pub enum_name: Ident,
pub variants: Vec<ErrorVariant>,
}
impl Parse for ErrorEnumInput {
fn parse(input: ParseStream) -> Result<Self> {
let enum_name = input.parse::<Ident>()?;
let content;
syn::braced!(content in input);
let mut variants = Vec::new();
while !content.is_empty() {
let mut attributes = Vec::new();
while content.peek(Token![#]) {
attributes.push(content.call(Attribute::parse_outer)?);
}
let attributes: Vec<_> = attributes.into_iter().flatten().collect();
let error_attr = attributes
.iter()
.find(|attr| attr.path().is_ident("error"))
.ok_or_else(|| Error::new(content.span(), "Missing #[error] attribute"))?;
let error_msg = match &error_attr.meta {
syn::Meta::List(list) => syn::parse2::<LitStr>(list.tokens.clone())?,
_ => {
return Err(Error::new(error_attr.span(), "Invalid error attribute format"));
}
};
let name = content.parse::<Ident>()?;
let mut fields = Vec::new();
if content.peek(syn::token::Brace) {
let fields_content;
syn::braced!(fields_content in content);
while !fields_content.is_empty() {
let field_name = fields_content.parse::<Ident>()?;
fields_content.parse::<Token![:]>()?;
let field_type = fields_content.parse::<Type>()?;
fields.push((field_name, field_type));
if !fields_content.is_empty() {
fields_content.parse::<Token![,]>()?;
}
}
}
content.parse::<Token![=>]>()?;
let level = content.parse::<Expr>()?;
if !content.is_empty() {
content.parse::<Token![,]>()?;
}
variants.push(ErrorVariant {
attributes,
error_msg,
name,
fields,
level,
});
}
Ok(ErrorEnumInput { enum_name, variants })
}
}
pub fn generate_error_enum(input: TokenStream, force_no_source: bool) -> TokenStream {
let input = syn::parse_macro_input!(input as ErrorEnumInput);
let enum_name = &input.enum_name;
let variants = &input.variants;
let enum_variants = variants.iter().map(|variant| {
let name = &variant.name;
let error_msg = &variant.error_msg;
let fields = &variant.fields;
let field_definitions = fields.iter().map(|(name, ty)| {
quote! { #name: #ty }
});
if force_no_source || variant.has_no_source() {
if variant.fields.is_empty() {
quote! {
#[error(#error_msg)]
#name
}
} else {
quote! {
#[error(#error_msg)]
#name { #(#field_definitions,)* }
}
}
} else {
quote! {
#[error(#error_msg)]
#name {
#(#field_definitions,)*
err: String
}
}
}
});
let level_match_arms = variants.iter().map(|variant| {
let name = &variant.name;
let level = &variant.level;
if force_no_source || variant.has_no_source() {
if variant.fields.is_empty() {
quote! {
Self::#name => #level
}
} else {
quote! {
Self::#name { .. } => #level
}
}
} else {
quote! {
Self::#name { err: _, .. } => #level
}
}
});
let constructors = variants.iter().filter_map(|variant| {
if !variant.should_generate_constructor(force_no_source) {
return None;
}
let name = &variant.name;
let fields = &variant.fields;
let params = fields.iter().map(|(field_name, field_type)| {
quote! { #field_name: impl Into<#field_type> }
});
let field_assignments = fields.iter().map(|(field_name, _)| {
quote! { #field_name: #field_name.into() }
});
if force_no_source || variant.has_no_source() {
Some(quote! {
#[allow(non_snake_case)]
pub fn #name(#(#params),*) -> Self {
Self::#name {
#(#field_assignments,)*
}
}
})
} else {
Some(quote! {
#[allow(non_snake_case)]
pub fn #name(#(#params,)* source: impl std::fmt::Display) -> Self {
Self::#name {
#(#field_assignments,)*
err: source.to_string()
}
}
})
}
});
let expanded = quote! {
#[allow(dead_code)]
#[derive(Debug, Clone, thiserror::Error, serde::Serialize, serde::Deserialize)]
pub enum #enum_name {
#(#enum_variants,)*
}
impl #enum_name {
#[allow(dead_code)]
pub fn level(&self) -> tracing::Level {
match self {
#(#level_match_arms,)*
}
}
#(#constructors)*
}
};
TokenStream::from(expanded)
}

1
macros/src/lib.rs
View File

@ -1,3 +1,4 @@
mod error_enum;
mod log;
mod loggable;
mod traceable;

140
macros/src/loggable.rs
View File

@ -1,143 +1,7 @@
use proc_macro::TokenStream;
use quote::quote;
use syn::parse::{Parse, ParseStream};
use syn::{braced, parse_macro_input, Attribute, Fields, Ident, LitStr, Token, Type, Visibility};
struct LoggableVariant {
error_message: LitStr,
name: Ident,
fields: Fields,
level: syn::Expr,
}
struct LoggableInput {
enum_name: Ident,
variants: Vec<LoggableVariant>,
}
impl Parse for LoggableInput {
fn parse(input: ParseStream) -> syn::Result<Self> {
let enum_name = input.parse::<Ident>()?;
let content;
braced!(content in input);
let mut variants = Vec::new();
while !content.is_empty() {
let attrs = content.call(Attribute::parse_outer)?;
let error_attr = attrs
.iter()
.find(|attr| attr.path().is_ident("error"))
.ok_or_else(|| content.error("Expected #[error(...)] attribute"))?;
let error_message = error_attr.parse_args::<LitStr>()?;
let name = content.parse::<Ident>()?;
let fields = if content.peek(syn::token::Brace) {
let field_content;
braced!(field_content in content);
let mut named_fields = syn::punctuated::Punctuated::new();
while !field_content.is_empty() {
let field_name = field_content.parse::<Ident>()?;
field_content.parse::<Token![:]>()?;
let field_type = field_content.parse::<Type>()?;
named_fields.push(syn::Field {
attrs: vec![],
vis: Visibility::Inherited,
mutability: syn::FieldMutability::None,
ident: Some(field_name),
colon_token: Some(Default::default()),
ty: field_type,
});
if field_content.peek(Token![,]) {
field_content.parse::<Token![,]>()?;
}
}
Fields::Named(syn::FieldsNamed {
brace_token: Default::default(),
named: named_fields,
})
} else {
Fields::Unit
};
content.parse::<Token![=>]>()?;
let level = content.parse::<syn::Expr>()?;
if content.peek(Token![,]) {
content.parse::<Token![,]>()?;
}
variants.push(LoggableVariant {
error_message,
name,
fields,
level,
});
}
Ok(LoggableInput {
enum_name,
variants,
})
}
}
use crate::error_enum;
pub fn loggable_impl(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as LoggableInput);
let enum_name = &input.enum_name;
let variants = &input.variants;
let enum_variants = variants.iter().map(|variant| {
let name = &variant.name;
let error_attr = &variant.error_message;
let fields = &variant.fields;
quote! {
#[error(#error_attr)]
#name #fields
}
});
let level_match_arms = variants.iter().map(|variant| {
let name = &variant.name;
let level = &variant.level;
let field_pattern = match &variant.fields {
Fields::Unit => quote! {},
Fields::Named(fields) => {
let field_names = fields.named.iter().map(|f| &f.ident);
quote! { { #(#field_names: _),* } }
}
Fields::Unnamed(_) => quote! { (..) },
};
quote! {
Self::#name #field_pattern => #level
}
});
quote! {
#[allow(dead_code)]
#[derive(Debug, Clone, thiserror::Error, serde::Serialize, serde::Deserialize)]
pub enum #enum_name {
#(#enum_variants,)*
}
impl #enum_name {
#[allow(dead_code)]
pub fn level(&self) -> tracing::Level {
match self {
#(#level_match_arms,)*
}
}
}
}
.into()
error_enum::generate_error_enum(input, true)
}

228
macros/src/traceable.rs
View File

@ -1,231 +1,7 @@
use proc_macro::TokenStream;
use quote::quote;
use syn::{
parse::{Parse, ParseStream}, parse_macro_input, spanned::Spanned, Attribute, Error, Expr, Ident, LitStr,
Result,
Token,
Type,
};
struct ErrorVariant {
attributes: Vec<Attribute>,
error_msg: LitStr,
name: Ident,
fields: Vec<(Ident, Type)>,
level: Expr,
}
impl ErrorVariant {
fn has_no_source(&self) -> bool {
self.attributes
.iter()
.any(|attr| attr.path().is_ident("no_source"))
}
fn should_generate_constructor(&self) -> bool {
if self.has_no_source() {
!self.fields.is_empty()
} else {
true
}
}
}
struct TraceableInput {
enum_name: Ident,
variants: Vec<ErrorVariant>,
}
impl Parse for TraceableInput {
fn parse(input: ParseStream) -> Result<Self> {
let enum_name = input.parse::<Ident>()?;
let content;
syn::braced!(content in input);
let mut variants = Vec::new();
while !content.is_empty() {
let mut attributes = Vec::new();
while content.peek(Token![#]) {
attributes.push(content.call(Attribute::parse_outer)?);
}
let attributes: Vec<_> = attributes.into_iter().flatten().collect();
let error_attr = attributes
.iter()
.find(|attr| attr.path().is_ident("error"))
.ok_or_else(|| Error::new(content.span(), "Missing #[error] attribute"))?;
let error_msg = match &error_attr.meta {
syn::Meta::List(list) => syn::parse2::<LitStr>(list.tokens.clone())?,
_ => {
return Err(Error::new(
error_attr.span(),
"Invalid error attribute format",
));
}
};
let name = content.parse::<Ident>()?;
let mut fields = Vec::new();
if content.peek(syn::token::Brace) {
let fields_content;
syn::braced!(fields_content in content);
while !fields_content.is_empty() {
let field_name = fields_content.parse::<Ident>()?;
fields_content.parse::<Token![:]>()?;
let field_type = fields_content.parse::<Type>()?;
fields.push((field_name, field_type));
if !fields_content.is_empty() {
fields_content.parse::<Token![,]>()?;
}
}
}
content.parse::<Token![=>]>()?;
let level = content.parse::<Expr>()?;
if !content.is_empty() {
content.parse::<Token![,]>()?;
}
variants.push(ErrorVariant {
attributes,
error_msg,
name,
fields,
level,
});
}
Ok(TraceableInput {
enum_name,
variants,
})
}
}
use crate::error_enum;
pub fn traceable_impl(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as TraceableInput);
let enum_name = &input.enum_name;
let variants = &input.variants;
let enum_variants = variants.iter().map(|variant| {
let name = &variant.name;
let error_msg = &variant.error_msg;
let fields = &variant.fields;
let field_definitions = fields.iter().map(|(name, ty)| {
quote! { #name: #ty }
});
if variant.has_no_source() {
if variant.fields.is_empty() {
quote! {
#[error(#error_msg)]
#name
}
} else {
quote! {
#[error(#error_msg)]
#name { #(#field_definitions,)* }
}
}
} else {
quote! {
#[error(#error_msg)]
#name {
#(#field_definitions,)*
err: String
}
}
}
});
let level_match_arms = variants.iter().map(|variant| {
let name = &variant.name;
let level = &variant.level;
if variant.has_no_source() {
if variant.fields.is_empty() {
quote! {
Self::#name => #level
}
} else {
quote! {
Self::#name { .. } => #level
}
}
} else {
quote! {
Self::#name { err: _, .. } => #level
}
}
});
let constructors = variants.iter().filter_map(|variant| {
if !variant.should_generate_constructor() {
return None;
}
let name = &variant.name;
let fields = &variant.fields;
let params = fields.iter().map(|(field_name, field_type)| {
quote! { #field_name: impl Into<#field_type> }
});
let field_assignments = fields.iter().map(|(field_name, _)| {
quote! { #field_name: #field_name.into() }
});
if variant.has_no_source() {
Some(quote! {
#[allow(non_snake_case)]
pub fn #name(#(#params),*) -> Self {
Self::#name {
#(#field_assignments,)*
}
}
})
} else {
Some(quote! {
#[allow(non_snake_case)]
pub fn #name(#(#params,)* source: impl std::fmt::Display) -> Self {
Self::#name {
#(#field_assignments,)*
err: source.to_string()
}
}
})
}
});
let expanded = quote! {
#[allow(dead_code)]
#[derive(Debug, Clone, thiserror::Error, serde::Serialize, serde::Deserialize)]
pub enum #enum_name {
#(#enum_variants,)*
}
impl #enum_name {
#[allow(dead_code)]
pub fn level(&self) -> tracing::Level {
match self {
#(#level_match_arms,)*
}
}
#(#constructors)*
}
};
TokenStream::from(expanded)
error_enum::generate_error_enum(input, false)
}

2
net-guardia/Cargo.toml
View File

@ -5,11 +5,11 @@ edition = "2024"
[dependencies]
common = { path = "../common", features = ["user"] }
macros = { path = "../macros" }
actix = "0.13.5"
actix-cors = "0.7.1"
actix-web = "4.11.0"
anyhow = { version = "1.0.99", default-features = true }
aya = { workspace = true }
aya-log = { workspace = true }
libc = { workspace = true }

3
net-guardia/build.rs
View File

@ -1,4 +1,3 @@
use cargo_metadata::{Artifact, CompilerMessage, Message, Metadata, MetadataCommand, Package, Target, TargetKind};
use std::env;
use std::fs;
use std::io::{BufRead as _, BufReader};
@ -6,6 +5,8 @@ use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::time::SystemTime;
use cargo_metadata::{Artifact, CompilerMessage, Message, Metadata, MetadataCommand, Package, Target, TargetKind};
fn main() {
build_ingress_ebpf();
build_egress_ebpf();

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

@ -2,12 +2,13 @@ use std::fs;
use std::sync::OnceLock;
use std::sync::RwLock as SyncRwLock;
use anyhow::anyhow;
use macros::log;
use tokio::sync::RwLock as AsyncRwLock;
use tracing::{error, info};
use crate::model::config::{Config, ConfigTable};
use crate::utils::log::system::SystemEntry;
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();
@ -15,33 +16,23 @@ static ASYNC_CONFIG: OnceLock<AsyncRwLock<Config>> = OnceLock::new();
pub struct AppConfig;
impl AppConfig {
pub async fn initialization() -> anyhow::Result<()> {
info!("{}", SystemEntry::Initializing);
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));
info!("{}", SystemEntry::InitializeComplete);
log!(SystemLog::InitializeComplete);
Ok(())
}
fn load_config() -> anyhow::Result<Config> {
let parse_result = (|| {
let toml_string = fs::read_to_string("./config.toml").map_err(|_| anyhow!(SystemEntry::ConfigNotFound))?;
let config_table =
toml::from_str::<ConfigTable>(&toml_string).map_err(|_| anyhow!(SystemEntry::InvalidConfig))?;
let config = config_table.config;
if !Self::validate(&config) {
Err(anyhow!(SystemEntry::InvalidConfig))
} else {
Ok(config)
}
})();
match parse_result {
Ok(config) => Ok(config),
Err(err) => {
error!("{}", SystemEntry::InvalidConfig);
Err(err)
}
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)
}
}

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

@ -6,16 +6,17 @@ 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 tracing::info;
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;
use crate::utils::log::ebpf::EbpfEntry;
use crate::utils::log::system::SystemEntry;
static ACCESS_CONTROL: OnceLock<RwLock<AccessControl>> = OnceLock::new();
@ -44,28 +45,30 @@ impl AccessControl {
),
];
pub async fn initialize() -> anyhow::Result<()> {
info!("{}", SystemEntry::Initializing);
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(ebpf.take_map(ipv4_name).unwrap())?,
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
AccessMap {
map: AyaHashMap::try_from(ebpf.take_map(ipv6_name).unwrap())?,
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
}
ACCESS_CONTROL.get_or_init(|| RwLock::new(AccessControl { ipv4_maps, ipv6_maps }));
info!("{}", SystemEntry::InitializeComplete);
log!(SystemLog::InitializeComplete);
Ok(())
}
@ -103,7 +106,7 @@ impl AccessControl {
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
@ -115,7 +118,7 @@ impl AccessControl {
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
@ -127,7 +130,7 @@ impl AccessControl {
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV4,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
@ -139,7 +142,7 @@ impl AccessControl {
direction: FlowDirection,
list_type: ListType,
address: SocketAddrV6,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let mut access_list = AccessControl::instance_mut().await;
@ -161,7 +164,7 @@ impl<T: IntoNative + Pod> AccessMap<T> {
.collect()
}
fn add(&mut self, ip: T, port: Port) -> anyhow::Result<()> {
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;
@ -179,7 +182,7 @@ impl<T: IntoNative + Pod> AccessMap<T> {
}
}
if index.is_none() {
return Err(EbpfEntry::RuleReachLimit.into());
Err(EbpfError::RuleReachLimit)?;
}
new_ports.copy_from_slice(&ports);
new_ports[index.unwrap()] = port;
@ -188,14 +191,14 @@ impl<T: IntoNative + Pod> AccessMap<T> {
}
self.map
.insert(ip, new_ports, 0)
.map_err(|_| EbpfEntry::MapOperationError)?;
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
fn remove(&mut self, ip: T, port: Port) -> anyhow::Result<()> {
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(|_| EbpfEntry::MapOperationError)?;
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
return Ok(());
}
@ -206,16 +209,14 @@ impl<T: IntoNative + Pod> AccessMap<T> {
ports[MAX_RULES_PORT - 1] = 0;
if ports[0] == 0 {
self.map.remove(&ip).map_err(|_| EbpfEntry::MapOperationError)?;
self.map.remove(&ip).map_err(EbpfError::MapOperationError)?;
} else {
self.map
.insert(ip, ports, 0)
.map_err(|_| EbpfEntry::MapOperationError)?;
self.map.insert(ip, ports, 0).map_err(EbpfError::MapOperationError)?;
}
}
Ok(())
} else {
Err(EbpfEntry::IpDoesNotExist)?
Err(EbpfError::IpDoesNotExist)?
}
}
}

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

@ -1,5 +1,6 @@
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;
@ -7,7 +8,7 @@ pub mod service;
pub struct Control;
impl Control {
pub async fn initialize() -> anyhow::Result<()> {
pub async fn initialize() -> Result<(), Error> {
AccessControl::initialize().await?;
Service::initialize().await
}

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

@ -6,13 +6,14 @@ 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 tracing::{error, info};
use crate::core::system::System;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::ip_address::IntoNative;
use crate::utils::log::ebpf::EbpfEntry;
use crate::utils::log::system::SystemEntry;
use crate::model::log::system::SystemLog;
static SERVICE: OnceLock<RwLock<Service>> = OnceLock::new();
@ -29,26 +30,46 @@ pub struct Service {
}
impl Service {
pub async fn initialize() -> anyhow::Result<()> {
info!("{}", SystemEntry::Initializing);
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").unwrap())?,
ipv6_http_service: AyaHashMap::try_from(ebpf.take_map("IPV6_HTTP_SERVICE").unwrap())?,
ssh_white_list_enable: AyaArray::try_from(ebpf.take_map("SSH_WHITE_LIST_ENABLE").unwrap())?,
ipv4_ssh_service: AyaHashMap::try_from(ebpf.take_map("IPV4_SSH_SERVICE").unwrap())?,
ipv6_ssh_service: AyaHashMap::try_from(ebpf.take_map("IPV6_SSH_SERVICE").unwrap())?,
ipv4_ssh_white_list: AyaHashMap::try_from(ebpf.take_map("IPV4_SSH_WHITE_LIST").unwrap())?,
ipv6_ssh_white_list: AyaHashMap::try_from(ebpf.take_map("IPV6_SSH_WHITE_LIST").unwrap())?,
ipv4_ssh_black_list: AyaHashMap::try_from(ebpf.take_map("IPV4_SSH_BLACK_LIST").unwrap())?,
ipv6_ssh_black_list: AyaHashMap::try_from(ebpf.take_map("IPV6_SSH_BLACK_LIST").unwrap())?,
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)?,
};
if service.ssh_white_list_enable.set(0, 0_u8, 0).is_err() {
error!(" ");
}
service
.ssh_white_list_enable
.set(0, 0_u8, 0)
.map_err(EbpfError::MapOperationError)?;
SERVICE.get_or_init(|| RwLock::new(service));
info!("{}", SystemEntry::InitializeComplete);
log!(SystemLog::InitializeComplete);
Ok(())
}
@ -95,7 +116,7 @@ impl Service {
.collect()
}
pub async fn add_ipv4_http_service(address: SocketAddrV4, http_method: Vec<HttpMethod>) -> anyhow::Result<()> {
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);
@ -104,11 +125,11 @@ impl Service {
service
.ipv4_http_service
.insert(addr_port, ebpf_method, 0)
.map_err(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_http_service(address: SocketAddrV6, http_method: Vec<HttpMethod>) -> anyhow::Result<()> {
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);
@ -117,14 +138,14 @@ impl Service {
service
.ipv6_http_service
.insert(addr_port, ebpf_method, 0)
.map_err(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_http_service(
address: SocketAddrV4,
removed_http_method: Vec<HttpMethod>,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV4::new(ip, port);
@ -136,24 +157,24 @@ impl Service {
service
.ipv4_http_service
.remove(&addr_port)
.map_err(|_| EbpfEntry::MapOperationError)?;
.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(|_| EbpfEntry::MapOperationError)?;
.map_err(EbpfError::MapOperationError)?;
}
Ok(())
} else {
Err(EbpfEntry::IpDoesNotExist)?
Err(EbpfError::IpDoesNotExist)?
}
}
pub async fn remove_ipv6_http_service(
address: SocketAddrV6,
removed_http_method: Vec<HttpMethod>,
) -> anyhow::Result<()> {
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let port = address.port();
let addr_port = AddrPortV6::new(ip, port);
@ -165,17 +186,17 @@ impl Service {
service
.ipv6_http_service
.remove(&addr_port)
.map_err(|_| EbpfEntry::MapOperationError)?;
.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(|_| EbpfEntry::MapOperationError)?;
.map_err(EbpfError::MapOperationError)?;
}
Ok(())
} else {
Err(EbpfEntry::IpDoesNotExist)?
Err(EbpfError::IpDoesNotExist)?
}
}
@ -193,21 +214,21 @@ impl Service {
}
}
pub async fn enable_ssh_white_list() -> anyhow::Result<()> {
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(|_| EbpfEntry::MapOperationError)?;
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
pub async fn disable_ssh_white_list() -> anyhow::Result<()> {
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(|_| EbpfEntry::MapOperationError)?;
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
@ -231,7 +252,7 @@ impl Service {
.collect()
}
pub async fn add_ipv4_ssh_service(address: SocketAddrV4) -> anyhow::Result<()> {
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);
@ -239,11 +260,11 @@ impl Service {
service
.ipv4_ssh_service
.insert(&addr_port, 0_u8, 0)
.map_err(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_service(address: SocketAddrV6) -> anyhow::Result<()> {
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);
@ -251,11 +272,11 @@ impl Service {
service
.ipv6_ssh_service
.insert(&addr_port, 0_u8, 0)
.map_err(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_service(address: SocketAddrV4) -> anyhow::Result<()> {
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);
@ -263,11 +284,11 @@ impl Service {
service
.ipv4_ssh_service
.remove(&addr_port)
.map_err(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_service(address: SocketAddrV6) -> anyhow::Result<()> {
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);
@ -275,7 +296,7 @@ impl Service {
service
.ipv6_ssh_service
.remove(&addr_port)
.map_err(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
@ -299,43 +320,43 @@ impl Service {
.collect()
}
pub async fn add_ipv4_ssh_white_list(ip: Ipv4Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_white_list(ip: Ipv6Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_white_list(ip: Ipv4Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_white_list(ip: Ipv6Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
@ -359,43 +380,43 @@ impl Service {
.collect()
}
pub async fn add_ipv4_ssh_black_list(ip: Ipv4Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn add_ipv6_ssh_black_list(ip: Ipv6Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::RuleReachLimit)?;
.map_err(|_| EbpfError::RuleReachLimit)?;
Ok(())
}
pub async fn remove_ipv4_ssh_black_list(ip: Ipv4Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
pub async fn remove_ipv6_ssh_black_list(ip: Ipv6Addr) -> anyhow::Result<()> {
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(|_| EbpfEntry::IpDoesNotExist)?;
.map_err(|_| EbpfError::IpDoesNotExist)?;
Ok(())
}
}

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

@ -1,12 +1,13 @@
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 tracing::{error, info, warn};
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();
@ -57,8 +58,6 @@ impl SystemHealth {
)
.await;
});
info!("System health monitoring initialized with configured interfaces");
}
async fn monitoring_loop(
@ -77,7 +76,6 @@ impl SystemHealth {
loop {
tokio::select! {
_ = shutdown_rx.recv() => {
info!("System health monitoring shutdown requested");
break;
}
_ = interval_timer.tick() => {
@ -95,15 +93,13 @@ impl SystemHealth {
);
if broadcast_tx.receiver_count() > 0 {
if let Err(e) = broadcast_tx.send(metrics) {
error!("Failed to broadcast system health metrics: {}", e);
if let Err(err) = broadcast_tx.send(metrics) {
log!(MiscError::SendMessageError(err))
}
}
}
}
}
info!("System health monitoring stopped");
}
pub async fn instance() -> RwLockReadGuard<'static, SystemHealth> {
@ -248,13 +244,13 @@ impl SystemHealth {
let management = create_network_stats(management_interface);
if ingress.is_none() {
warn!("Ingress interface '{}' not found", ingress_interface);
log!(MiscError::NetworkInterfaceNotFound(ingress_interface));
}
if egress.is_none() {
warn!("Egress interface '{}' not found", egress_interface);
log!(MiscError::NetworkInterfaceNotFound(egress_interface));
}
if management.is_none() {
warn!("Management interface '{}' not found", management_interface);
log!(MiscError::NetworkInterfaceNotFound(management_interface));
}
ConfiguredNetworkStats {

24
net-guardia/src/core/statistics.rs
View File

@ -6,14 +6,16 @@ use aya::maps::{HashMap as AyaHashMap, MapData};
use aya::Pod;
use common::model::flow_stats::FlowStats;
use common::model::ip_address::{AddrPortV4, AddrPortV6};
use macros::log;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use tracing::info;
use crate::core::system::System;
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::time_type::TimeType;
use crate::utils::log::system::SystemEntry;
static STATISTICS: OnceLock<RwLock<Statistics>> = OnceLock::new();
@ -78,38 +80,42 @@ impl Statistics {
),
];
pub async fn initialize() -> anyhow::Result<()> {
info!("{}", SystemEntry::Initializing);
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;
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(ingress_ebpf.take_map(ipv4_name).unwrap())?,
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(ingress_ebpf.take_map(ipv6_name).unwrap())?,
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
}
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(egress_ebpf.take_map(ipv4_name).unwrap())?,
map: AyaHashMap::try_from(ipv4_map).map_err(EbpfError::MapOperationError)?,
},
);
ipv6_maps.insert(
key,
FlowMap {
map: AyaHashMap::try_from(egress_ebpf.take_map(ipv6_name).unwrap())?,
map: AyaHashMap::try_from(ipv6_map).map_err(EbpfError::MapOperationError)?,
},
);
}
@ -119,7 +125,7 @@ impl Statistics {
ipv6_maps,
};
STATISTICS.get_or_init(|| RwLock::new(statistics));
info!("{}", SystemEntry::InitializeComplete);
log!(SystemLog::InitializeComplete);
Ok(())
}

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

@ -3,20 +3,24 @@ use std::time::Duration;
use actix_web::web::route;
use actix_web::{App, HttpServer};
use anyhow::Context;
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 tracing::{error, info, warn};
use crate::core::app_config::AppConfig;
use crate::core::control::Control;
use crate::core::health::SystemHealth;
use crate::core::statistics::Statistics;
use crate::utils::log::ebpf::EbpfEntry;
use crate::utils::log::system::SystemEntry;
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};
@ -33,9 +37,9 @@ pub struct System {
}
impl System {
pub async fn initialize() -> anyhow::Result<()> {
pub async fn initialize() -> Result<(), Error> {
Logging::initialize().await?;
info!("{}", SystemEntry::Initializing);
log!(SystemLog::Initializing);
AppConfig::initialization().await?;
@ -45,28 +49,36 @@ impl System {
Statistics::initialize().await?;
Control::initialize().await?;
info!("{}", SystemEntry::InitializeComplete);
log!(SystemLog::InitializeComplete);
Ok(())
}
async fn ebpf_initialize() -> anyhow::Result<()> {
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()?;
let (mut ingress_ebpf, ingress_program_array) = System::get_ingress_ebpf()?;
let ingress_program: &mut Xdp = ingress_ebpf.program_mut("net_guardia").unwrap().try_into()?;
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").unwrap().try_into()?;
ingress_program.load()?;
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())
.context(EbpfEntry::AttachProgramFailed)?;
egress_program.load()?;
.map_err(EbpfError::AttachProgramFailed)?;
egress_program.load().map_err(EbpfError::LoadProgramFailed)?;
egress_program
.attach(&egress_interface, XdpFlags::default())
.context(EbpfEntry::AttachProgramFailed)?;
.map_err(EbpfError::AttachProgramFailed)?;
let system = System {
ingress_ebpf,
egress_ebpf,
@ -75,48 +87,46 @@ impl System {
egress_program_array,
};
SYSTEM.get_or_init(|| RwLock::new(system));
info!("{}", EbpfEntry::AttachProgramSuccess);
log!(EbpfLog::AttachProgramSuccess);
Ok(())
}
fn get_ingress_ebpf() -> anyhow::Result<(Ebpf, ProgramArray<MapData>)> {
fn get_ingress_ebpf() -> Result<(Ebpf, ProgramArray<MapData>), Error> {
let mut ingress_ebpf = Ebpf::load(aya::include_bytes_aligned!(concat!(
env!("OUT_DIR"),
"/net-guardia-ingress"
)))?;
if let Err(e) = aya_log::EbpfLogger::init(&mut ingress_ebpf) {
error!("{}", e);
warn!("{}", EbpfEntry::LoggerInitializeFailed);
}
let mut ingress_program_array = ProgramArray::try_from(ingress_ebpf.take_map("PROGRAM_ARRAY").unwrap())?;
Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "access_control", 0)?;
Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "service", 1)?;
Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "statistics", 2)?;
Ok((ingress_ebpf, ingress_program_array))
)))
.map_err(EbpfError::EbpfNotFound)?;
EbpfLogger::init(&mut ingress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
let program_array = ingress_ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(program_array).map_err(EbpfError::MapOperationError)?;
Self::load_program(&mut ingress_ebpf, &mut program_array, "access_control", 0)?;
Self::load_program(&mut ingress_ebpf, &mut program_array, "service", 1)?;
Self::load_program(&mut ingress_ebpf, &mut program_array, "statistics", 2)?;
Ok((ingress_ebpf, program_array))
}
fn get_egress_ebpf() -> anyhow::Result<(Ebpf, ProgramArray<MapData>)> {
fn get_egress_ebpf() -> Result<(Ebpf, ProgramArray<MapData>), Error> {
let mut egress_ebpf = Ebpf::load(aya::include_bytes_aligned!(concat!(
env!("OUT_DIR"),
"/net-guardia-egress"
)))?;
if let Err(e) = aya_log::EbpfLogger::init(&mut egress_ebpf) {
error!("{}", e);
warn!("{}", EbpfEntry::LoggerInitializeFailed);
}
let mut egress_program_array = ProgramArray::try_from(egress_ebpf.take_map("PROGRAM_ARRAY").unwrap())?;
Self::load_program(&mut egress_ebpf, &mut egress_program_array, "statistics", 0)?;
Ok((egress_ebpf, egress_program_array))
)))
.map_err(EbpfError::EbpfNotFound)?;
EbpfLogger::init(&mut egress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
let program_array = egress_ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(program_array).map_err(EbpfError::MapOperationError)?;
Self::load_program(&mut egress_ebpf, &mut program_array, "statistics", 0)?;
Ok((egress_ebpf, program_array))
}
fn set_memory_limit() -> anyhow::Result<()> {
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 {
info!("Failed to remove limit on locked memory, ret is: {}", ret);
Err(MiscError::RamLimitUnlockError(ret))?
}
Ok(())
}
@ -126,16 +136,20 @@ impl System {
program_array: &mut ProgramArray<MapData>,
function_name: &str,
index: u32,
) -> anyhow::Result<()> {
let program: &mut Xdp = ebpf.program_mut(function_name).unwrap().try_into()?;
program.load()?;
let fd = program.fd()?;
program_array.set(index, fd, 0)?;
) -> Result<(), Error> {
let program: &mut Xdp = ebpf
.program_mut(function_name)
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::MapOperationError)?;
program.load().map_err(EbpfError::AttachProgramFailed)?;
let fd = program.fd().map_err(|_| EbpfError::UnknownError)?;
program_array.set(index, fd, 0).map_err(EbpfError::MapOperationError)?;
Ok(())
}
pub async fn run() -> anyhow::Result<()> {
info!("{}", SystemEntry::Online);
pub async fn run() -> Result<(), Error> {
log!(SystemLog::Online);
Statistics::run().await;
@ -154,19 +168,21 @@ impl System {
.service(health::initialize())
.default_service(route().to(default::default_route))
})
.bind(format!("0.0.0.0:{}", config.http_server_bind_port))?
.bind(format!("0.0.0.0:{}", config.http_server_bind_port))
.map_err(HttpError::BindPortError)?
.run()
.await?;
.await
.map_err(HttpError::ServerPanic)?;
Ok(())
}
pub async fn terminate() -> anyhow::Result<()> {
info!("{}", SystemEntry::Terminating);
pub async fn terminate() -> Result<(), Error> {
log!(SystemLog::Terminating);
Statistics::terminate().await;
SystemHealth::shutdown().await;
info!("{}", SystemEntry::TerminateComplete);
log!(SystemLog::TerminateComplete);
Ok(())
}

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

@ -4,9 +4,10 @@ mod utils;
mod web;
use crate::core::system::System;
use crate::model::error::Error;
#[actix_web::main]
async fn main() -> anyhow::Result<()> {
async fn main() -> Result<(), Error> {
System::initialize().await?;
System::run().await?;
System::terminate().await?;

44
net-guardia/src/model/error/ebpf.rs Normal file
View File

@ -0,0 +1,44 @@
use macros::traceable;
use tracing;
traceable! {
EbpfError {
#[error("Failed to initialize eBPF logger")]
LoggerInitFailed => tracing::Level::ERROR,
#[error("Ebpf program not found")]
EbpfNotFound => tracing::Level::ERROR,
#[no_source]
#[error("Failed to load XDP program")]
ProgramNotFound => tracing::Level::ERROR,
#[error("Failed to load XDP program")]
GetProgramFailed => tracing::Level::ERROR,
#[error("Failed to load XDP program")]
LoadProgramFailed => tracing::Level::ERROR,
#[error("Failed to attach the XDP program")]
AttachProgramFailed => tracing::Level::ERROR,
#[no_source]
#[error("Map not found")]
MapNotFound => tracing::Level::ERROR,
#[error("An error occurred during map operation")]
MapOperationError => tracing::Level::ERROR,
#[no_source]
#[error("The ip required for operation does not exist")]
IpDoesNotExist => tracing::Level::ERROR,
#[no_source]
#[error("Amount of rules has reached the upper limit")]
RuleReachLimit => tracing::Level::ERROR,
#[no_source]
#[error("Unknown error")]
UnknownError => tracing::Level::ERROR,
}
}

14
net-guardia/src/model/error/http.rs Normal file
View File

@ -0,0 +1,14 @@
use macros::traceable;
traceable! {
HttpError {
#[error("Bind port error")]
BindPortError => tracing::Level::ERROR,
#[error("Http Server panic")]
ServerPanic => tracing::Level::ERROR,
#[error("WebSocket error")]
WebSocketError => tracing::Level::ERROR,
}
}

10
net-guardia/src/model/error/io.rs Normal file
View File

@ -0,0 +1,10 @@
use std::path::PathBuf;
use macros::traceable;
traceable! {
IOError {
#[error("Failed to create directory: {path}")]
CreateDirectoryFailed { path: PathBuf } => tracing::Level::ERROR,
}
}

22
net-guardia/src/model/error/misc.rs Normal file
View File

@ -0,0 +1,22 @@
use macros::traceable;
traceable! {
MiscError {
#[no_source]
#[error("Failed to remove limit on locked memory, ret is: {ret}")]
RamLimitUnlockError { ret: i32 } => tracing::Level::ERROR,
#[error("Failed to send message to receiver")]
SendMessageError => tracing::Level::ERROR,
#[error("Failed to serialize data")]
SerializeError => tracing::Level::ERROR,
#[error("Failed to deserialize data")]
DeserializeError => tracing::Level::ERROR,
#[no_source]
#[error("Network interface '{interface}' not found")]
NetworkInterfaceNotFound { interface: String } => tracing::Level::ERROR,
}
}

57
net-guardia/src/model/error/mod.rs Normal file
View File

@ -0,0 +1,57 @@
pub mod ebpf;
pub mod http;
pub mod io;
pub mod misc;
pub mod system;
use serde::{Deserialize, Serialize};
use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::io::IOError;
use crate::model::error::misc::MiscError;
use crate::model::error::system::SystemError;
#[derive(Clone, Debug, thiserror::Error, Serialize, Deserialize)]
pub enum Error {
#[error("{0}")]
Ebpf(EbpfError),
#[error("{0}")]
Http(HttpError),
#[error("{0}")]
IO(IOError),
#[error("{0}")]
Misc(MiscError),
#[error("{0}")]
System(SystemError),
}
impl From<EbpfError> for Error {
fn from(error: EbpfError) -> Self {
Self::Ebpf(error)
}
}
impl From<HttpError> for Error {
fn from(error: HttpError) -> Self {
Self::Http(error)
}
}
impl From<IOError> for Error {
fn from(error: IOError) -> Self {
Self::IO(error)
}
}
impl From<MiscError> for Error {
fn from(error: MiscError) -> Self {
Self::Misc(error)
}
}
impl From<SystemError> for Error {
fn from(error: SystemError) -> Self {
Self::System(error)
}
}

33
net-guardia/src/model/error/system.rs Normal file
View File

@ -0,0 +1,33 @@
use macros::traceable;
traceable! {
SystemError {
#[no_source]
#[error("Unable to run as administrator")]
RunAsAdminFailed => tracing::Level::ERROR,
#[no_source]
#[error("Invalid configuration")]
InvalidConfig => tracing::Level::ERROR,
#[error("Configuration not found")]
ConfigNotFound => tracing::Level::ERROR,
#[error("Failed to terminate instance")]
TerminateError => tracing::Level::ERROR,
#[no_source]
#[error("Failed to send shutdown signal")]
ShutdownSignalFailed => tracing::Level::ERROR,
#[error("Unexcepted thread panic")]
ThreadPanic => tracing::Level::ERROR,
#[error("Unexcepted error")]
UnexpectError => tracing::Level::ERROR,
#[no_source]
#[error("Unknown error")]
UnknownError => tracing::Level::ERROR,
}
}

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

@ -78,4 +78,4 @@ pub struct SystemHealthStatus {
pub overall_healthy: bool,
pub issues: Vec<String>,
pub warnings: Vec<String>,
}
}

2
net-guardia/src/model/list_type.rs
View File

@ -1,4 +1,4 @@
use serde::{Serialize, Deserialize};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Hash)]
#[serde(rename_all = "lowercase")]

9
net-guardia/src/model/log/ebpf.rs Normal file
View File

@ -0,0 +1,9 @@
use macros::loggable;
use tracing;
loggable! {
EbpfLog {
#[error("Attach XDP program success")]
AttachProgramSuccess => tracing::Level::INFO,
}
}

9
net-guardia/src/model/log/http.rs Normal file
View File

@ -0,0 +1,9 @@
use macros::loggable;
use tracing;
loggable! {
HttpLog {
#[error("Health WebSocket lagged, skipped {skipped} messages")]
WebSocketLaged { skipped: u64 } => tracing::Level::WARN,
}
}

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

@ -1,2 +1,3 @@
pub mod ebpf;
pub mod http;
pub mod system;

27
net-guardia/src/model/log/system.rs Normal file
View File

@ -0,0 +1,27 @@
use macros::loggable;
use tracing;
loggable! {
SystemLog {
#[error("Online now")]
Online => tracing::Level::INFO,
#[error("Initializing")]
Initializing => tracing::Level::INFO,
#[error("Initialization completed")]
InitializeComplete => tracing::Level::INFO,
#[error("Termination in process")]
Terminating => tracing::Level::INFO,
#[error("Termination completed")]
TerminateComplete => tracing::Level::INFO,
#[error("Invalid configuration")]
InvalidConfig => tracing::Level::INFO,
#[error("Configuration not found")]
ConfigNotFound => tracing::Level::INFO,
}
}

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

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

17
net-guardia/src/utils/log/ebpf.rs
View File

@ -1,17 +0,0 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum EbpfEntry {
#[error("Failed to initialize eBPF logger")]
LoggerInitializeFailed,
#[error("Attach XDP program success")]
AttachProgramSuccess,
#[error("Failed to attach the XDP program")]
AttachProgramFailed,
#[error("An error occurred during map operation")]
MapOperationError,
#[error("The ip required for operation does not exist")]
IpDoesNotExist,
#[error("Amount of rules has reached the upper limit")]
RuleReachLimit,
}

19
net-guardia/src/utils/log/system.rs
View File

@ -1,19 +0,0 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum SystemEntry {
#[error("Online now")]
Online,
#[error("Initializing")]
Initializing,
#[error("Initialization completed")]
InitializeComplete,
#[error("Termination in process")]
Terminating,
#[error("Termination completed")]
TerminateComplete,
#[error("Invalid configuration")]
InvalidConfig,
#[error("Configuration not found")]
ConfigNotFound,
}

9
net-guardia/src/utils/logging.rs
View File

@ -5,12 +5,17 @@ use tracing_subscriber::filter::EnvFilter;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use crate::model::error::io::IOError;
use crate::model::error::Error;
pub struct Logging;
impl Logging {
pub async fn initialize() -> anyhow::Result<()> {
pub async fn initialize() -> Result<(), Error> {
let log_directory = "logs";
fs::create_dir_all(log_directory).await?;
fs::create_dir_all(log_directory)
.await
.map_err(|err| IOError::CreateDirectoryFailed(log_directory, err))?;
let file_appender = RollingFileAppender::new(Rotation::DAILY, log_directory, "NetGuardia");

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

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

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

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

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

@ -1,12 +1,14 @@
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 tracing::error;
use crate::core::app_config::AppConfig;
use crate::core::statistics::Statistics;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::time_type::TimeType;
pub async fn websocket_ipv4_flow(
@ -135,7 +137,7 @@ async fn handle_client_message(
false
}
Some(Err(err)) => {
error!("WebSocket error: {}", err);
log!(HttpError::WebSocketError(err));
false
}
None => false,
@ -153,7 +155,7 @@ async fn send_ipv4_flow_data(
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
error!("Failed to serialize IPv4 flow data: {}", err);
log!(MiscError::SerializeError(err));
true
}
}
@ -169,7 +171,7 @@ async fn send_ipv6_flow_data(
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
error!("Failed to serialize IPv6 flow data: {}", err);
log!(MiscError::SerializeError(err));
true
}
}

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

@ -1,11 +1,14 @@
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 tracing::{error, warn};
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,
@ -65,7 +68,7 @@ async fn handle_client_message(
match serde_json::to_string(&current_metrics) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
error!("Failed to serialize current metrics: {}", err);
log!(MiscError::SerializeError(err));
true
}
}
@ -87,7 +90,7 @@ async fn handle_client_message(
false
}
Some(Err(err)) => {
error!("WebSocket error: {}", err);
log!(HttpError::WebSocketError(err));
false
}
None => false,
@ -105,13 +108,13 @@ async fn handle_broadcast_message(
match serde_json::to_string(&message) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
error!("Failed to serialize health metrics: {}", err);
log!(MiscError::SerializeError(err));
true
}
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
warn!("Health WebSocket lagged, skipped {} messages", skipped);
log!(HttpLog::WebSocketLaged(skipped));
let lag_msg = serde_json::json!({
"message": format!("Connection lagged, skipped {} messages", skipped)
});

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

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