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

refactor: hexagonal architecture with ports, adapters, and CQRS event bus

Browse Source 
- Architecture: reorganize into hexagonal layers (interface/, adapter/,
  infrastructure/) with strict unidirectional dependency rules
- interface/communication: CQRS message bus (Command, Query, Event traits)
  adapted from MirrorSphere's CommunicationManager pattern
- interface/port: define 5 port traits (RepositoryPort, AuthPort, HealthPort,
  NotificationPort, PacketProcessorPort) for dependency inversion
- infrastructure: extract ServiceFactory and HttpServer from God Object
  (system.rs reduced from 503 to ~120 lines), add CommunicationManager
- adapter/http: move web/api/ handlers, use dyn RepositoryPort trait objects
  instead of concrete Database type
- adapter/websocket: move web/websocket/ handlers + route definitions
- adapter/persistence: move core/database/, implement RepositoryPort trait
- Fix layer violations: model/ no longer imports core/, adapters don't
  cross-import each other
- Define 10 command types, 8 query types, 6 event types for subsystem
  communication
- Add 10 new tests (29 total): CommunicationManager dispatch (9 tests),
  RepositoryPort trait object verification (1 test)

Dependency rules enforced:
  model/ → (no imports from other layers)
  interface/ → model/ only
  adapter/ → interface/ + model/ (no cross-adapter imports)
  infrastructure/ → all layers (composition root)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
DaLaw2 2026-03-21 23:52:01 +08:00
parent 8138c5d751
commit c24bd50d7e
49 changed files with 1449 additions and 456 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
Cargo.lock generated
View File

@ -883,6 +883,20 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "dashmap"
version = "6.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf"
dependencies = [
"cfg-if",
"crossbeam-utils",
"hashbrown 0.14.5",
"lock_api",
"once_cell",
"parking_lot_core",
]
[[package]]
name = "data-encoding"
version = "2.10.0"
@ -1957,6 +1971,7 @@ dependencies = [
"actix-web",
"actix-ws",
"argon2",
"async-trait",
"aya",
"aya-log",
"base64",
@ -1964,6 +1979,7 @@ dependencies = [
"chrono",
"common",
"crossbeam",
"dashmap",
"dotenvy",
"ed25519-dalek",
"futures-util",
@ -2953,7 +2969,6 @@ dependencies = [
"cfg-if",
"libc",
"psm",
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]

4
net-guardia/Cargo.toml
View File

@ -47,6 +47,10 @@ tract-onnx = { workspace = true }
lettre = { version = "0.11", default-features = false, features = ["builder", "hostname", "smtp-transport", "tokio1-rustls-tls"] }
chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
# Architecture
async-trait = "0.1"
dashmap = "6"
# Utilities
parking_lot = { workspace = true }
thiserror = { workspace = true }

16
net-guardia/src/web/api/acl.rs → net-guardia/src/adapter/http/acl.rs
View File

@ -3,7 +3,9 @@ use std::net::{SocketAddrV4, SocketAddrV6};
use actix_web::{web, HttpResponse, Responder, Scope};
use serde::Deserialize;
use crate::core::database::Database;
use crate::interface::port::repository::RepositoryPort;
type Repo = dyn RepositoryPort;
use crate::core::ebpf::access_control::AccessControl;
use crate::core::ebpf::geo_block::GeoBlock;
use crate::model::direction::FlowDirection;
@ -57,7 +59,7 @@ async fn add_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
@ -74,7 +76,7 @@ async fn add_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
@ -91,7 +93,7 @@ async fn remove_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
@ -108,7 +110,7 @@ async fn remove_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
@ -131,7 +133,7 @@ async fn get_geo_blocked(
async fn block_geo_countries(
body: web::Json<CountryCodesRequest>,
geo_block: web::Data<GeoBlock>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let codes = body.into_inner().country_codes;
for code in &codes {
@ -153,7 +155,7 @@ async fn block_geo_countries(
async fn unblock_geo_countries(
body: web::Json<CountryCodesRequest>,
geo_block: web::Data<GeoBlock>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let codes = body.into_inner().country_codes;
for code in &codes {

10
net-guardia/src/web/api/auth.rs → net-guardia/src/adapter/http/auth.rs
View File

@ -3,7 +3,9 @@ use serde::Deserialize;
use crate::core::auth::jwt::{Claims, JwtService};
use crate::core::auth::password;
use crate::core::database::Database;
use crate::interface::port::repository::RepositoryPort;
type Repo = dyn RepositoryPort;
#[derive(Deserialize)]
struct LoginRequest {
@ -51,7 +53,7 @@ fn validate_password(password: &str) -> Result<(), &'static str> {
async fn login(
body: web::Json<LoginRequest>,
db: web::Data<Database>,
db: web::Data<Repo>,
jwt: web::Data<JwtService>,
) -> impl Responder {
let req = body.into_inner();
@ -106,7 +108,7 @@ async fn login(
async fn register(
req: HttpRequest,
body: web::Json<RegisterRequest>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
// Check caller is admin
let claims = req.extensions().get::<Claims>().cloned();
@ -166,7 +168,7 @@ async fn me(req: HttpRequest) -> impl Responder {
async fn change_password(
req: HttpRequest,
body: web::Json<ChangePasswordRequest>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let claims = match req.extensions().get::<Claims>().cloned() {
Some(c) => c,

2
net-guardia/src/web/api/default.rs → net-guardia/src/adapter/http/default.rs
View File

@ -39,4 +39,4 @@ pub async fn default_route(req: HttpRequest) -> impl Responder {
.body(page.data.into_owned()),
None => HttpResponse::NotFound().body("404 Not Found"),
}
}
}

8
net-guardia/src/web/api/filter.rs → net-guardia/src/adapter/http/filter.rs
View File

@ -5,7 +5,9 @@ use actix_web::{web, HttpResponse, Responder, Scope};
use common::model::http_method::HttpMethod;
use serde::Deserialize;
use crate::core::database::Database;
use crate::interface::port::repository::RepositoryPort;
type Repo = dyn RepositoryPort;
use crate::core::ebpf::dns_filter::DnsFilter;
use crate::core::ebpf::protocol_filter::ProtocolFilter;
@ -49,7 +51,7 @@ async fn get_dns_blacklist(service: web::Data<DnsFilter>) -> impl Responder {
async fn add_dns_blacklist(
payload: web::Json<DnsDomainsPayload>,
service: web::Data<DnsFilter>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let domains = payload.into_inner().domains;
if domains.len() > MAX_DNS_DOMAINS_PER_REQUEST {
@ -74,7 +76,7 @@ async fn add_dns_blacklist(
async fn remove_dns_blacklist(
payload: web::Json<DnsDomainsPayload>,
service: web::Data<DnsFilter>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let domains = payload.into_inner().domains;
for domain in &domains {

0
net-guardia/src/web/api/health.rs → net-guardia/src/adapter/http/health.rs
View File

0
net-guardia/src/web/api/ml.rs → net-guardia/src/adapter/http/ml.rs
View File

1
net-guardia/src/web/api/mod.rs → net-guardia/src/adapter/http/mod.rs
View File

@ -7,4 +7,3 @@ pub mod ml;
pub mod rate_limit;
pub mod stats;
pub mod system;
pub mod ws;

6
net-guardia/src/web/api/rate_limit.rs → net-guardia/src/adapter/http/rate_limit.rs
View File

@ -2,7 +2,9 @@ use actix_web::{web, HttpResponse, Responder, Scope};
use serde::{Deserialize, Serialize};
use common::define::setting::*;
use crate::core::database::Database;
use crate::interface::port::repository::RepositoryPort;
type Repo = dyn RepositoryPort;
use crate::core::ebpf::rate_limit::RateLimitConfig;
#[derive(Serialize, Deserialize)]
@ -35,7 +37,7 @@ async fn get_config(
async fn set_config(
settings: web::Json<RateLimitSettings>,
config: web::Data<RateLimitConfig>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let s = settings.into_inner();
if let Some(v) = s.packet_rate {

0
net-guardia/src/web/api/stats.rs → net-guardia/src/adapter/http/stats.rs
View File

10
net-guardia/src/web/api/system.rs → net-guardia/src/adapter/http/system.rs
View File

@ -1,7 +1,9 @@
use actix_web::{web, HttpResponse, Responder, Scope};
use serde::Deserialize;
use crate::core::database::Database;
use crate::interface::port::repository::RepositoryPort;
type Repo = dyn RepositoryPort;
#[derive(Deserialize)]
struct EnforceModeRequest {
@ -25,7 +27,7 @@ async fn get_boot_time() -> impl Responder {
HttpResponse::Ok().json(crate::utils::boot_time::boot_time())
}
async fn get_enforce_mode(db: web::Data<Database>) -> impl Responder {
async fn get_enforce_mode(db: web::Data<Repo>) -> impl Responder {
match db.get_setting("enforce_mode") {
Ok(Some(mode)) => HttpResponse::Ok().json(serde_json::json!({"mode": mode})),
Ok(None) => HttpResponse::Ok().json(serde_json::json!({"mode": "monitor"})),
@ -36,7 +38,7 @@ async fn get_enforce_mode(db: web::Data<Database>) -> impl Responder {
async fn set_enforce_mode(
body: web::Json<EnforceModeRequest>,
db: web::Data<Database>,
db: web::Data<Repo>,
) -> impl Responder {
let mode = &body.mode;
if mode != "monitor" && mode != "enforce" {
@ -54,7 +56,7 @@ async fn set_enforce_mode(
}
}
async fn get_xdp_mode(db: web::Data<Database>) -> impl Responder {
async fn get_xdp_mode(db: web::Data<Repo>) -> impl Responder {
let ingress = db.get_setting("xdp_ingress_mode")
.ok().flatten().unwrap_or_else(|| "unknown".to_string());
let egress = db.get_setting("xdp_egress_mode")

3
net-guardia/src/adapter/mod.rs Normal file
View File

@ -0,0 +1,3 @@
pub mod http;
pub mod persistence;
pub mod websocket;

0
net-guardia/src/core/database/mod.rs → net-guardia/src/adapter/persistence/mod.rs
View File

47
net-guardia/src/core/database/repository.rs → net-guardia/src/adapter/persistence/repository.rs
View File

@ -294,6 +294,31 @@ impl Database {
}
}
/// Implement the RepositoryPort trait, proving Database satisfies the port contract.
/// This enables adapter-level testing with mock implementations.
impl crate::interface::port::repository::RepositoryPort for Database {
fn insert_acl_rule(&self, ip_version: u8, direction: &str, list_type: &str, ip_address: &str, port: u16) -> Result<(), Error> { self.insert_acl_rule(ip_version, direction, list_type, ip_address, port) }
fn delete_acl_rule(&self, ip_version: u8, direction: &str, list_type: &str, ip_address: &str, port: u16) -> Result<(), Error> { self.delete_acl_rule(ip_version, direction, list_type, ip_address, port) }
fn load_acl_rules(&self) -> Result<Vec<(u8, String, String, String, u16)>, Error> { self.load_acl_rules() }
fn set_rate_limit(&self, key: &str, value: u64) -> Result<(), Error> { self.set_rate_limit(key, value) }
fn load_rate_limit_config(&self) -> Result<Vec<(String, u64)>, Error> { self.load_rate_limit_config() }
fn insert_dns_domain(&self, domain: &str) -> Result<(), Error> { self.insert_dns_domain(domain) }
fn delete_dns_domain(&self, domain: &str) -> Result<(), Error> { self.delete_dns_domain(domain) }
fn load_dns_domains(&self) -> Result<Vec<String>, Error> { self.load_dns_domains() }
fn insert_geo_country(&self, code: &str) -> Result<(), Error> { self.insert_geo_country(code) }
fn delete_geo_country(&self, code: &str) -> Result<(), Error> { self.delete_geo_country(code) }
fn load_geo_countries(&self) -> Result<Vec<String>, Error> { self.load_geo_countries() }
fn get_setting(&self, key: &str) -> Result<Option<String>, Error> { self.get_setting(key) }
fn set_setting(&self, key: &str, value: &str) -> Result<(), Error> { self.set_setting(key, value) }
fn find_user(&self, username: &str) -> Result<Option<(i64, String, String, String, bool)>, Error> { self.find_user(username) }
fn insert_user(&self, username: &str, password_hash: &str, role: &str, force_password_change: bool) -> Result<(), Error> { self.insert_user(username, password_hash, role, force_password_change) }
fn update_user_password(&self, user_id: i64, password_hash: &str) -> Result<(), Error> { self.update_user_password(user_id, password_hash) }
fn user_count(&self) -> Result<i64, Error> { self.user_count() }
fn record_login_failure(&self, username: &str) -> Result<(u32, Option<u64>), Error> { self.record_login_failure(username) }
fn check_login_locked(&self, username: &str) -> Result<Option<u64>, Error> { self.check_login_locked(username) }
fn clear_login_failures(&self, username: &str) -> Result<(), Error> { self.clear_login_failures(username) }
}
#[cfg(test)]
mod tests {
use super::*;
@ -434,4 +459,26 @@ mod tests {
let db = test_db();
assert!(db.find_user("nobody").unwrap().is_none());
}
/// Verify that Database satisfies the RepositoryPort trait contract.
/// This test ensures the trait impl compiles and can be used via trait object.
#[test]
fn test_repository_port_trait_object() {
use crate::interface::port::repository::RepositoryPort;
let db = test_db();
let repo: &dyn RepositoryPort = &db;
// Use via trait object — proves the abstraction works
repo.set_setting("test_key", "test_value").unwrap();
assert_eq!(repo.get_setting("test_key").unwrap(), Some("test_value".to_string()));
repo.insert_acl_rule(4, "source", "blacklist", "10.0.0.1", 443).unwrap();
let rules = repo.load_acl_rules().unwrap();
assert_eq!(rules.len(), 1);
assert_eq!(repo.user_count().unwrap(), 0);
repo.insert_user("test", "hash", "viewer", false).unwrap();
assert_eq!(repo.user_count().unwrap(), 1);
}
}

0
net-guardia/src/web/websocket/alert_websocket.rs → net-guardia/src/adapter/websocket/alert_websocket.rs
View File

0
net-guardia/src/web/websocket/drop_websocket.rs → net-guardia/src/adapter/websocket/drop_websocket.rs
View File

0
net-guardia/src/web/websocket/flow_websocket.rs → net-guardia/src/adapter/websocket/flow_websocket.rs
View File

0
net-guardia/src/web/websocket/health_websocket.rs → net-guardia/src/adapter/websocket/health_websocket.rs
View File

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

@ -2,3 +2,4 @@ pub mod alert_websocket;
pub mod drop_websocket;
pub mod flow_websocket;
pub mod health_websocket;
pub mod routes;

2
net-guardia/src/web/api/ws.rs → net-guardia/src/adapter/websocket/routes.rs
View File

@ -6,7 +6,7 @@ use crate::core::ebpf::drop_monitor::DropMonitor;
use crate::core::infrastructure::health::SystemHealth;
use crate::core::infrastructure::statistics::FlowStatistics;
use crate::core::ml::alert::MLAlert;
use crate::web::websocket::{alert_websocket, drop_websocket, flow_websocket, health_websocket};
use super::{alert_websocket, drop_websocket, flow_websocket, health_websocket};
#[derive(Deserialize)]
struct WsQuery {

2
net-guardia/src/core/auth/jwt.rs
View File

@ -1,7 +1,7 @@
use jsonwebtoken::{decode, encode, errors::ErrorKind, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use crate::core::database::Database;
use crate::adapter::persistence::Database;
use crate::model::error::auth::AuthError;
use crate::model::error::Error;

2
net-guardia/src/core/email/report.rs
View File

@ -1,4 +1,4 @@
use crate::core::database::Database;
use crate::adapter::persistence::Database;
use crate::model::error::Error;
use std::sync::Arc;

2
net-guardia/src/core/email/scheduler.rs
View File

@ -1,4 +1,4 @@
use crate::core::database::Database;
use crate::adapter::persistence::Database;
use crate::model::error::database::DatabaseError;
use crate::model::error::Error;
use lettre::message::header::ContentType;

22
net-guardia/src/core/infrastructure/statistics.rs
View File

@ -2,8 +2,30 @@ use std::sync::Arc;
use std::time;
use crate::core::ml::engine::Engine;
use crate::core::ml::flow_tracker::FlowData;
use crate::model::flow_stats::{FlowStatsEntry, FlowSubscription, StatsSummary};
/// Conversion from core::ml::FlowData to model::FlowStatsEntry.
/// Placed here (core layer) to maintain dependency rule: model/ must not import core/.
impl From<&FlowData> for FlowStatsEntry {
fn from(flow: &FlowData) -> Self {
Self {
direction: flow.direction,
src_ip: flow.flow_key.src_ip_string(),
dst_ip: flow.flow_key.dst_ip_string(),
src_port: flow.flow_key.src_port,
dst_port: flow.flow_key.dst_port,
protocol: flow.flow_key.protocol,
fwd_packets: flow.fwd_packets.len(),
bwd_packets: flow.bwd_packets.len(),
fwd_bytes: flow.fwd_total_bytes,
bwd_bytes: flow.bwd_total_bytes,
duration_us: flow.duration_us(),
last_seen_us: flow.last_time_us,
}
}
}
pub struct FlowStatistics {
engine: Arc<Engine>,
}

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

@ -1,5 +1,4 @@
pub mod auth;
pub mod database;
pub mod email;
pub mod ebpf;
pub mod infrastructure;

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

@ -1,47 +1,28 @@
use std::collections::HashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc;
use actix_web::web::route;
use actix_web::{web, App, HttpServer};
use aya::maps::{Array, MapData, ProgramArray};
use aya::programs::{Xdp, XdpFlags};
use aya::maps::{MapData, ProgramArray};
use aya::Ebpf;
use aya_log::EbpfLogger;
use common::define::pipeline::*;
use macros::log;
use crate::core::auth::jwt::JwtService;
use crate::core::auth::password;
use crate::core::database::Database;
use crate::adapter::persistence::Database;
use crate::core::ebpf::EbpfServices;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::MLService;
use crate::core::ml::config_loader::InferenceConfig;
#[cfg(feature = "license")]
use crate::core::license::LicenseInfo;
#[cfg(feature = "license")]
use crate::core::license::validator::validate_license;
use crate::core::ml::config_loader::InferenceConfig;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::direction::FlowDirection;
use crate::infrastructure::http_server::HttpServerParams;
use crate::infrastructure::service_factory::ServiceFactory;
use crate::model::error::Error;
use crate::model::list_type::ListType;
use crate::model::log::ml::MLLog;
use crate::model::log::system::SystemLog;
use crate::utils::logging::Logging;
use crate::web::api::{acl, auth, filter, rate_limit as rate_limit_api, stats, health as health_api, ml, system as system_api, default, ws};
/// Maps stage name (from config.toml) to (function_name, stage_id)
fn stage_registry() -> HashMap<&'static str, (&'static str, u32)> {
HashMap::from([
("access_control", ("access_control", STAGE_ACCESS_CONTROL)),
("rate_limit", ("rate_limit", STAGE_RATE_LIMIT)),
("service", ("protocol_filter", STAGE_SERVICE)),
])
}
/// Thin wrapper around infrastructure services.
/// Delegates construction to `ServiceFactory::build()` and HTTP to
/// `infrastructure::http_server::run()`.
/// Will be removed in a later refactoring phase.
pub struct System {
pub app_config: Arc<AppConfig>,
pub inference_config: Arc<InferenceConfig>,
@ -59,163 +40,19 @@ pub struct System {
impl System {
pub async fn new() -> Result<Self, Error> {
let mut ingress_ebpf = Self::load_ebpf("ingress")?;
let mut egress_ebpf = Self::load_ebpf("egress")?;
let app_config = Arc::new(AppConfig::new()?);
#[cfg(feature = "license")]
let license_info = Arc::new(validate_license(
&app_config.misc.license_file,
&app_config.network.ingress_ifname,
&app_config.network.egress_ifname,
)?);
let ingress_program_array = Self::configure_ingress_pipeline(
&mut ingress_ebpf,
&app_config.pipeline.ingress,
)?;
let inference_config = Arc::new(InferenceConfig::load_file(&app_config.inference.models_config_name)?);
// Write queue count to eBPF maps for symmetric hash redirect
let num_queues = app_config.network.combined_queue_count;
Self::write_num_queues(&mut ingress_ebpf, num_queues)?;
Self::write_num_queues(&mut egress_ebpf, num_queues)?;
let db = Arc::new(Database::new(&app_config.misc.database_path)?);
// Create default admin user if no users exist
if db.user_count().unwrap_or(0) == 0 {
let hash = password::hash_password("admin")?;
db.insert_user("admin", &hash, "admin", true)?;
tracing::warn!("Default admin user created with password 'admin' — you must change it on first login");
}
// Ensure enforce_mode setting exists (default: monitor)
if db.get_setting("enforce_mode")?.is_none() {
db.set_setting("enforce_mode", "monitor")?;
}
let jwt_service = Arc::new(JwtService::new(&db, app_config.http.jwt_expiry_hours)?);
let ebpf_services = Arc::new(EbpfServices::new(
app_config.clone(),
&mut ingress_ebpf,
&mut egress_ebpf,
)?);
let app_services = Arc::new(MLService::new(app_config.clone(), inference_config.clone())?);
// Load persisted DNS blacklist
if let Ok(domains) = db.load_dns_domains() {
for domain in &domains {
if let Err(e) = ebpf_services.dns_filter.add_domain(domain) {
tracing::warn!("Failed to restore DNS domain '{}': {}", domain, e);
}
}
if !domains.is_empty() {
tracing::info!("Restored {} DNS blacklist domains from database", domains.len());
}
}
// Load persisted geo-blocked countries
if let Ok(countries) = db.load_geo_countries() {
if !countries.is_empty() {
if let Err(e) = ebpf_services.geo_block.block_countries(&countries) {
tracing::warn!("Failed to restore geo-blocked countries: {}", e);
} else {
tracing::info!("Restored {} geo-blocked countries from database", countries.len());
}
}
}
// Load persisted rate limit config
if let Ok(configs) = db.load_rate_limit_config() {
for (key, value) in &configs {
let result = match key.as_str() {
"packet_rate" => ebpf_services.rate_limit.set_packet_rate(*value),
"syn_rate" => ebpf_services.rate_limit.set_syn_rate(*value),
"udp_rate" => ebpf_services.rate_limit.set_udp_rate(*value),
"dns_rate" => ebpf_services.rate_limit.set_dns_rate(*value),
"window_ns" => ebpf_services.rate_limit.set_window_ns(*value),
_ => Ok(()),
};
if let Err(e) = result {
tracing::warn!("Failed to restore rate limit '{}': {}", key, e);
}
}
if !configs.is_empty() {
tracing::info!("Restored {} rate limit settings from database", configs.len());
}
}
// Load persisted ACL rules
if let Ok(rules) = db.load_acl_rules() {
let mut restored = 0u32;
for (ip_version, direction, list_type, ip_address, port) in &rules {
let dir = match direction.as_str() {
"source" => FlowDirection::Source,
"destination" => FlowDirection::Destination,
other => {
tracing::warn!("Unknown ACL direction '{}', skipping", other);
continue;
}
};
let lt = match list_type.as_str() {
"whitelist" => ListType::White,
"blacklist" => ListType::Black,
other => {
tracing::warn!("Unknown ACL list type '{}', skipping", other);
continue;
}
};
let result = match ip_version {
4 => {
match ip_address.parse::<Ipv4Addr>() {
Ok(addr) => ebpf_services.access_control.add_ipv4_list(dir, lt, SocketAddrV4::new(addr, *port)).await,
Err(e) => {
tracing::warn!("Failed to parse IPv4 address '{}': {}", ip_address, e);
continue;
}
}
}
6 => {
match ip_address.parse::<Ipv6Addr>() {
Ok(addr) => ebpf_services.access_control.add_ipv6_list(dir, lt, SocketAddrV6::new(addr, *port, 0, 0)).await,
Err(e) => {
tracing::warn!("Failed to parse IPv6 address '{}': {}", ip_address, e);
continue;
}
}
}
other => {
tracing::warn!("Unknown IP version {}, skipping", other);
continue;
}
};
if let Err(e) = result {
tracing::warn!("Failed to restore ACL rule ({} {} {}:{}): {}", direction, list_type, ip_address, port, e);
} else {
restored += 1;
}
}
if restored > 0 {
tracing::info!("Restored {} ACL rules from database", restored);
}
}
let state = ServiceFactory::build().await?;
Ok(System {
app_config,
inference_config,
ebpf_services,
app_services,
db,
jwt_service,
app_config: state.app_config,
inference_config: state.inference_config,
ebpf_services: state.ebpf_services,
app_services: state.app_services,
db: state.db,
jwt_service: state.jwt_service,
#[cfg(feature = "license")]
license_info,
ingress_ebpf,
egress_ebpf,
ingress_program_array,
license_info: state.license_info,
ingress_ebpf: state.ingress_ebpf,
egress_ebpf: state.egress_ebpf,
ingress_program_array: state.ingress_program_array,
})
}
@ -237,7 +74,7 @@ impl System {
attacks: self.inference_config.num_attack_types()
});
self.aya_log_init()?;
ServiceFactory::aya_log_init(&mut self.ingress_ebpf, &mut self.egress_ebpf)?;
log!(SystemLog::InitializeComplete);
self.attach_ebpf()?;
@ -258,19 +95,13 @@ impl System {
Ok(())
}
fn aya_log_init(&mut self) -> Result<(), Error> {
EbpfLogger::init(&mut self.ingress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
EbpfLogger::init(&mut self.egress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
Ok(())
}
fn attach_ebpf(&mut self) -> Result<(), Error> {
let ingress_ifname = self.app_config.network.ingress_ifname.clone();
let egress_ifname = self.app_config.network.egress_ifname.clone();
Self::set_memory_limit()?;
ServiceFactory::set_memory_limit()?;
let ingress_mode = Self::attach_xdp(&mut self.ingress_ebpf, &ingress_ifname, true)?;
let egress_mode = Self::attach_xdp(&mut self.egress_ebpf, &egress_ifname, false)?;
let ingress_mode = ServiceFactory::attach_xdp(&mut self.ingress_ebpf, &ingress_ifname, true)?;
let egress_mode = ServiceFactory::attach_xdp(&mut self.egress_ebpf, &egress_ifname, false)?;
// Store XDP mode in settings for health API reporting
if let Err(e) = self.db.set_setting("xdp_ingress_mode", &ingress_mode) {
@ -283,221 +114,17 @@ impl System {
Ok(())
}
fn attach_xdp(ebpf: &mut Ebpf, ifname: &str, already_loaded: bool) -> Result<String, Error> {
let xdp: &mut Xdp = ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
if !already_loaded {
xdp.load().map_err(EbpfError::LoadProgramFailed)?;
}
// Try DRV_MODE first (native XDP, best performance)
match xdp.attach(ifname, XdpFlags::DRV_MODE) {
Ok(_) => {
tracing::info!("XDP attached to {} in native DRV_MODE", ifname);
return Ok("drv".to_string());
}
Err(drv_err) => {
tracing::warn!(
"XDP DRV_MODE failed on {}: {}. Falling back to SKB_MODE.",
ifname, drv_err
);
}
}
// Fallback to SKB_MODE (generic XDP, reduced performance)
match xdp.attach(ifname, XdpFlags::SKB_MODE) {
Ok(_) => {
tracing::warn!(
"XDP attached to {} in generic SKB_MODE (reduced performance). \
For best performance, use a NIC with native XDP support (e.g., virtio-net, Intel i40e/ice).",
ifname
);
Ok("skb".to_string())
}
Err(skb_err) => {
tracing::error!(
"XDP attach failed on {} with both DRV_MODE and SKB_MODE. \
Ensure the interface exists and supports XDP. \
Supported NICs: virtio-net, Intel i40e/ice/i350, Mellanox mlx5. \
SKB error: {}",
ifname, skb_err
);
Err(EbpfError::AttachProgramFailed(skb_err).into())
}
}
}
async fn run_http_server(&self) -> Result<(), Error> {
let app_config = self.app_config.clone();
let inference_config = self.inference_config.clone();
let access_control = self.ebpf_services.access_control.clone();
let protocol_filter = self.ebpf_services.protocol_filter.clone();
let dns_filter = self.ebpf_services.dns_filter.clone();
let geo_block = self.ebpf_services.geo_block.clone();
let rate_limit = self.ebpf_services.rate_limit.clone();
let health = self.app_services.health.clone();
let ml_alert = self.app_services.ml_alert.clone();
let ml_engine = self.app_services.ml_engine.clone();
let flow_statistics = self.app_services.flow_statistics.clone();
let drop_monitor = self.ebpf_services.drop_monitor.clone();
let db = self.db.clone();
let jwt_service = self.jwt_service.clone();
#[cfg(feature = "license")]
let license_info = self.license_info.clone();
let port = self.app_config.http.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);
let app = App::new()
.wrap(cors)
.app_data(web::Data::from(app_config.clone()))
.app_data(web::Data::from(inference_config.clone()))
.app_data(web::Data::from(access_control.clone()))
.app_data(web::Data::from(protocol_filter.clone()))
.app_data(web::Data::from(dns_filter.clone()))
.app_data(web::Data::from(geo_block.clone()))
.app_data(web::Data::from(rate_limit.clone()))
.app_data(web::Data::from(health.clone()))
.app_data(web::Data::from(ml_alert.clone()))
.app_data(web::Data::from(ml_engine.clone()))
.app_data(web::Data::from(flow_statistics.clone()))
.app_data(web::Data::from(drop_monitor.clone()))
.app_data(web::Data::from(db.clone()))
.app_data(web::Data::from(jwt_service.clone()));
let params = HttpServerParams {
app_config: self.app_config.clone(),
inference_config: self.inference_config.clone(),
ebpf_services: self.ebpf_services.clone(),
app_services: self.app_services.clone(),
db: self.db.clone(),
jwt_service: self.jwt_service.clone(),
#[cfg(feature = "license")]
let app = app.app_data(web::Data::from(license_info.clone()));
app.service(
web::scope("/api")
.wrap(crate::core::auth::middleware::AuthMiddleware)
.service(auth::initialize())
.service(acl::initialize())
.service(filter::initialize())
.service(rate_limit_api::initialize())
.service(stats::initialize())
.service(health_api::initialize())
.service(ml::initialize())
.service(system_api::initialize())
)
.service(ws::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(())
}
fn load_ebpf(name: &str) -> Result<Ebpf, Error> {
let bytes = match name {
"ingress" => aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/net-guardia-ingress")),
"egress" => aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/net-guardia-egress")),
_ => return Err(EbpfError::ProgramNotFound.into()),
license_info: self.license_info.clone(),
};
Ok(Ebpf::load(bytes).map_err(EbpfError::EbpfNotFound)?)
}
/// Configure the ingress pipeline based on config.toml [Pipeline] section.
/// Loads each stage program into ProgramArray and wires NEXT_STAGE map.
fn configure_ingress_pipeline(
ebpf: &mut Ebpf,
stages: &[String],
) -> Result<ProgramArray<MapData>, Error> {
let registry = stage_registry();
let entry: &mut Xdp = ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
entry.load().map_err(EbpfError::LoadProgramFailed)?;
let pa_map = ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(pa_map).map_err(EbpfError::MapOperationError)?;
let ns_map = ebpf.take_map("NEXT_STAGE").ok_or(EbpfError::MapNotFound)?;
let mut next_stage = Array::<MapData, u32>::try_from(ns_map).map_err(EbpfError::MapOperationError)?;
Self::load_program(ebpf, &mut program_array, "transmission", STAGE_TRANSMISSION)?;
if stages.is_empty() {
next_stage
.set(STAGE_ENTRY as u32, STAGE_TRANSMISSION, 0)
.map_err(EbpfError::MapOperationError)?;
return Ok(program_array);
}
let mut slots: Vec<(u32, u32)> = Vec::new();
for (i, stage_name) in stages.iter().enumerate() {
let (func_name, stage_id) = registry
.get(stage_name.as_str())
.ok_or(EbpfError::ProgramNotFound)?;
let slot = (i + 1) as u32;
Self::load_program(ebpf, &mut program_array, func_name, slot)?;
slots.push((*stage_id, slot));
}
next_stage
.set(STAGE_ENTRY as u32, slots[0].1, 0)
.map_err(EbpfError::MapOperationError)?;
for i in 0..slots.len() {
let (stage_id, _) = slots[i];
let next_slot = if i + 1 < slots.len() {
slots[i + 1].1
} else {
STAGE_TRANSMISSION
};
next_stage
.set(stage_id as u32, next_slot, 0)
.map_err(EbpfError::MapOperationError)?;
}
Ok(program_array)
}
fn load_program(
ebpf: &mut Ebpf,
program_array: &mut ProgramArray<MapData>,
function_name: &str,
slot: u32,
) -> 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(slot, fd, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
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 write_num_queues(ebpf: &mut Ebpf, num_queues: u32) -> Result<(), Error> {
let map = ebpf.map_mut("NUM_QUEUES").ok_or(EbpfError::MapNotFound)?;
let mut arr = Array::<_, u32>::try_from(map).map_err(EbpfError::MapOperationError)?;
arr.set(0, num_queues, 0).map_err(EbpfError::MapOperationError)?;
Ok(())
crate::infrastructure::http_server::run(params).await
}
}

360
net-guardia/src/infrastructure/communication_manager.rs Normal file
View File

@ -0,0 +1,360 @@
use crate::interface::communication::command::*;
use crate::interface::communication::event::Event;
use crate::interface::communication::event::EventBroadcaster;
use crate::interface::communication::query::*;
use crate::model::error::misc::MiscError;
use crate::model::error::Error;
use dashmap::DashMap;
use std::any::{Any, TypeId};
use std::sync::Arc;
use tokio::sync::broadcast;
/// Default broadcast channel capacity for event types.
const DEFAULT_CHANNEL_CAPACITY: usize = 256;
/// Inline TypedEventBroadcaster (adapted from MirrorSphere's model).
pub struct TypedEventBroadcaster<E: Event> {
pub sender: broadcast::Sender<E>,
}
impl<E: Event + 'static> EventBroadcaster for TypedEventBroadcaster<E> {
fn subscribe_typed(&self) -> Box<dyn Any + Send> {
Box::new(self.sender.subscribe())
}
fn broadcast_event(&self, event: Box<dyn Any + Send>) -> Result<(), Error> {
let typed_event = *event.downcast::<E>().map_err(|_| MiscError::TypeMismatch)?;
let _ = self.sender.send(typed_event);
Ok(())
}
}
/// Central communication hub using the command/query/event pattern.
/// Adapted from MirrorSphere's CommunicationManager for NetGuardia.
pub struct CommunicationManager {
command_handlers: DashMap<TypeId, CommandHandlerFn>,
query_handlers: DashMap<TypeId, QueryHandlerFn>,
event_broadcasters: DashMap<TypeId, Box<dyn EventBroadcaster>>,
channel_capacity: usize,
}
impl CommunicationManager {
pub fn new() -> Self {
Self {
command_handlers: DashMap::new(),
query_handlers: DashMap::new(),
event_broadcasters: DashMap::new(),
channel_capacity: DEFAULT_CHANNEL_CAPACITY,
}
}
pub fn with_capacity(channel_capacity: usize) -> Self {
Self {
command_handlers: DashMap::new(),
query_handlers: DashMap::new(),
event_broadcasters: DashMap::new(),
channel_capacity,
}
}
pub fn with_service<S: Send + Sync + 'static>(
self: Arc<Self>,
service: Arc<S>,
) -> ServiceRegistrar<S> {
ServiceRegistrar::new(service, self)
}
pub fn register_command_handler<C: Command + 'static>(
&self,
handler: Arc<dyn CommandHandler<C> + Send + Sync>,
) {
let type_id = TypeId::of::<C>();
let boxed_handler: CommandHandlerFn = Box::new(move |command: Box<dyn Any + Send>| {
let handler = handler.clone();
Box::pin(async move {
let command = *command
.downcast::<C>()
.map_err(|_| MiscError::TypeMismatch)?;
handler.handle_command(command).await
}) as CommandFuture
});
self.command_handlers.insert(type_id, boxed_handler);
}
pub async fn send_command<C: Command + 'static>(&self, command: C) -> Result<(), Error> {
let type_id = TypeId::of::<C>();
if let Some(handler) = self.command_handlers.get(&type_id) {
handler(Box::new(command)).await
} else {
Err(MiscError::HandlerNotFound)?
}
}
pub fn register_query_handler<Q: Query + 'static>(
&self,
handler: Arc<dyn QueryHandler<Q> + Send + Sync>,
) {
let type_id = TypeId::of::<Q>();
let boxed_handler: QueryHandlerFn = Box::new(move |query: Box<dyn Any + Send>| {
let handler = handler.clone();
Box::pin(async move {
let query = *query.downcast::<Q>().map_err(|_| MiscError::TypeMismatch)?;
let response = handler.handle_query(query).await?;
Ok(Box::new(response) as Box<dyn Any + Send>)
}) as QueryFuture
});
self.query_handlers.insert(type_id, boxed_handler);
}
pub async fn send_query<Q: Query + 'static>(&self, query: Q) -> Result<Q::Response, Error> {
let type_id = TypeId::of::<Q>();
if let Some(handler) = self.query_handlers.get(&type_id) {
let response = handler(Box::new(query)).await?;
Ok(*response
.downcast::<Q::Response>()
.map_err(|_| MiscError::TypeMismatch)?)
} else {
Err(MiscError::HandlerNotFound)?
}
}
pub fn register_event_type<E: Event + 'static>(&self) {
let type_id = TypeId::of::<E>();
let (tx, _) = broadcast::channel::<E>(self.channel_capacity);
let broadcaster = TypedEventBroadcaster { sender: tx };
self.event_broadcasters
.insert(type_id, Box::new(broadcaster));
}
pub fn subscribe_event<E: Event + 'static>(&self) -> Result<broadcast::Receiver<E>, Error> {
let type_id = TypeId::of::<E>();
let broadcaster = self
.event_broadcasters
.get(&type_id)
.ok_or(MiscError::TypeNotRegistered)?;
let receiver_box = broadcaster.subscribe_typed();
let receiver = *receiver_box
.downcast::<broadcast::Receiver<E>>()
.map_err(|_| MiscError::TypeMismatch)?;
Ok(receiver)
}
pub async fn publish_event<E: Event + 'static>(&self, event: E) -> Result<(), Error> {
let type_id = TypeId::of::<E>();
let broadcaster = self
.event_broadcasters
.get(&type_id)
.ok_or(MiscError::TypeNotRegistered)?;
broadcaster.broadcast_event(Box::new(event))
}
pub fn clear_handlers(&self) {
self.command_handlers.clear();
self.query_handlers.clear();
self.event_broadcasters.clear();
}
}
/// Fluent builder for registering a service's command/query/event handlers.
pub struct ServiceRegistrar<S> {
service: Arc<S>,
comm: Arc<CommunicationManager>,
}
impl<S: Send + Sync + 'static> ServiceRegistrar<S> {
fn new(service: Arc<S>, comm: Arc<CommunicationManager>) -> Self {
Self { service, comm }
}
pub fn command<C: Command + 'static>(self) -> Self
where
S: CommandHandler<C>,
{
let handler: Arc<dyn CommandHandler<C> + Send + Sync> = self.service.clone();
self.comm.register_command_handler::<C>(handler);
self
}
pub fn query<Q: Query + 'static>(self) -> Self
where
S: QueryHandler<Q>,
{
let handler: Arc<dyn QueryHandler<Q> + Send + Sync> = self.service.clone();
self.comm.register_query_handler::<Q>(handler);
self
}
pub fn event<E: Event + 'static>(self) -> Self {
self.comm.register_event_type::<E>();
self
}
pub fn build(self) -> Arc<CommunicationManager> {
self.comm
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::interface::communication::message::Message;
use crate::interface::communication::command::Command;
use crate::interface::communication::query::Query;
use crate::interface::communication::event::Event;
use async_trait::async_trait;
// ── Test Command ─────────────────────────────────────────────────
struct TestCommand {
value: String,
}
impl Message for TestCommand {
type Response = ();
}
impl Command for TestCommand {}
struct TestCommandHandler {
received: Arc<std::sync::Mutex<Vec<String>>>,
}
#[async_trait]
impl CommandHandler<TestCommand> for TestCommandHandler {
async fn handle_command(&self, command: TestCommand) -> Result<(), Error> {
self.received.lock().unwrap().push(command.value);
Ok(())
}
}
// ── Test Query ───────────────────────────────────────────────────
struct TestQuery {
input: i32,
}
impl Message for TestQuery {
type Response = i32;
}
impl Query for TestQuery {}
struct TestQueryHandler;
#[async_trait]
impl QueryHandler<TestQuery> for TestQueryHandler {
async fn handle_query(&self, query: TestQuery) -> Result<i32, Error> {
Ok(query.input * 2)
}
}
// ── Test Event ───────────────────────────────────────────────────
#[derive(Debug, Clone)]
struct TestEvent {
message: String,
}
impl Event for TestEvent {}
// ── Tests ────────────────────────────────────────────────────────
#[tokio::test]
async fn test_command_dispatch() {
let received = Arc::new(std::sync::Mutex::new(Vec::new()));
let handler = Arc::new(TestCommandHandler { received: received.clone() });
let comm = Arc::new(CommunicationManager::new());
comm.register_command_handler::<TestCommand>(handler);
comm.send_command(TestCommand { value: "hello".into() }).await.unwrap();
let msgs = received.lock().unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0], "hello");
}
#[tokio::test]
async fn test_command_not_found() {
let comm = CommunicationManager::new();
let result = comm.send_command(TestCommand { value: "nope".into() }).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_query_dispatch() {
let handler = Arc::new(TestQueryHandler);
let comm = Arc::new(CommunicationManager::new());
comm.register_query_handler::<TestQuery>(handler);
let result = comm.send_query(TestQuery { input: 21 }).await.unwrap();
assert_eq!(result, 42);
}
#[tokio::test]
async fn test_query_not_found() {
let comm = CommunicationManager::new();
let result = comm.send_query(TestQuery { input: 1 }).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_event_pub_sub() {
let comm = CommunicationManager::new();
comm.register_event_type::<TestEvent>();
let mut receiver = comm.subscribe_event::<TestEvent>().unwrap();
comm.publish_event(TestEvent { message: "ping".into() }).await.unwrap();
let event = receiver.recv().await.unwrap();
assert_eq!(event.message, "ping");
}
#[tokio::test]
async fn test_event_not_registered() {
let comm = CommunicationManager::new();
let result = comm.subscribe_event::<TestEvent>();
assert!(result.is_err());
}
#[tokio::test]
async fn test_event_multiple_subscribers() {
let comm = CommunicationManager::new();
comm.register_event_type::<TestEvent>();
let mut rx1 = comm.subscribe_event::<TestEvent>().unwrap();
let mut rx2 = comm.subscribe_event::<TestEvent>().unwrap();
comm.publish_event(TestEvent { message: "broadcast".into() }).await.unwrap();
assert_eq!(rx1.recv().await.unwrap().message, "broadcast");
assert_eq!(rx2.recv().await.unwrap().message, "broadcast");
}
#[tokio::test]
async fn test_service_registrar() {
let received = Arc::new(std::sync::Mutex::new(Vec::new()));
let handler = Arc::new(TestCommandHandler { received: received.clone() });
let comm = Arc::new(CommunicationManager::new());
let _comm = comm.clone()
.with_service(handler)
.command::<TestCommand>()
.build();
comm.send_command(TestCommand { value: "via_registrar".into() }).await.unwrap();
let msgs = received.lock().unwrap();
assert_eq!(msgs[0], "via_registrar");
}
#[test]
fn test_clear_handlers() {
let comm = CommunicationManager::new();
comm.register_event_type::<TestEvent>();
assert!(comm.subscribe_event::<TestEvent>().is_ok());
comm.clear_handlers();
assert!(comm.subscribe_event::<TestEvent>().is_err());
}
}

97
net-guardia/src/infrastructure/http_server.rs Normal file
View File

@ -0,0 +1,97 @@
use std::sync::Arc;
use actix_web::web::route;
use actix_web::{web, App, HttpServer};
use crate::core::auth::jwt::JwtService;
use crate::adapter::persistence::Database;
use crate::core::ebpf::EbpfServices;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::MLService;
use crate::core::ml::config_loader::InferenceConfig;
#[cfg(feature = "license")]
use crate::core::license::LicenseInfo;
use crate::model::error::http::HttpError;
use crate::model::error::Error;
use crate::adapter::http::{acl, auth, default, filter, health as health_api, ml, rate_limit as rate_limit_api, stats, system as system_api};
use crate::adapter::websocket::routes as ws;
use crate::interface::port::repository::RepositoryPort;
/// Parameters for starting the HTTP server, avoiding `#[cfg]` on function params.
pub struct HttpServerParams {
pub app_config: Arc<AppConfig>,
pub inference_config: Arc<InferenceConfig>,
pub ebpf_services: Arc<EbpfServices>,
pub app_services: Arc<MLService>,
pub db: Arc<Database>,
pub jwt_service: Arc<JwtService>,
#[cfg(feature = "license")]
pub license_info: Arc<LicenseInfo>,
}
/// Run the HTTP server with the given parameters.
pub async fn run(params: HttpServerParams) -> Result<(), Error> {
let access_control = params.ebpf_services.access_control.clone();
let protocol_filter = params.ebpf_services.protocol_filter.clone();
let dns_filter = params.ebpf_services.dns_filter.clone();
let geo_block = params.ebpf_services.geo_block.clone();
let rate_limit = params.ebpf_services.rate_limit.clone();
let health = params.app_services.health.clone();
let ml_alert = params.app_services.ml_alert.clone();
let ml_engine = params.app_services.ml_engine.clone();
let flow_statistics = params.app_services.flow_statistics.clone();
let drop_monitor = params.ebpf_services.drop_monitor.clone();
let app_config = params.app_config;
let inference_config = params.inference_config;
let db = params.db;
let jwt_service = params.jwt_service;
#[cfg(feature = "license")]
let license_info = params.license_info;
let port = app_config.http.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);
let app = App::new()
.wrap(cors)
.app_data(web::Data::from(app_config.clone()))
.app_data(web::Data::from(inference_config.clone()))
.app_data(web::Data::from(access_control.clone()))
.app_data(web::Data::from(protocol_filter.clone()))
.app_data(web::Data::from(dns_filter.clone()))
.app_data(web::Data::from(geo_block.clone()))
.app_data(web::Data::from(rate_limit.clone()))
.app_data(web::Data::from(health.clone()))
.app_data(web::Data::from(ml_alert.clone()))
.app_data(web::Data::from(ml_engine.clone()))
.app_data(web::Data::from(flow_statistics.clone()))
.app_data(web::Data::from(drop_monitor.clone()))
.app_data(web::Data::from(db.clone() as Arc<dyn RepositoryPort>))
.app_data(web::Data::from(jwt_service.clone()));
#[cfg(feature = "license")]
let app = app.app_data(web::Data::from(license_info.clone()));
app.service(
web::scope("/api")
.wrap(crate::core::auth::middleware::AuthMiddleware)
.service(auth::initialize())
.service(acl::initialize())
.service(filter::initialize())
.service(rate_limit_api::initialize())
.service(stats::initialize())
.service(health_api::initialize())
.service(ml::initialize())
.service(system_api::initialize())
)
.service(ws::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(())
}

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

@ -0,0 +1,3 @@
pub mod communication_manager;
pub mod http_server;
pub mod service_factory;

389
net-guardia/src/infrastructure/service_factory.rs Normal file
View File

@ -0,0 +1,389 @@
use std::collections::HashMap;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc;
use aya::maps::{Array, MapData, ProgramArray};
use aya::programs::{Xdp, XdpFlags};
use aya::Ebpf;
use aya_log::EbpfLogger;
use common::define::pipeline::*;
use crate::core::auth::jwt::JwtService;
use crate::core::auth::password;
use crate::adapter::persistence::Database;
use crate::core::ebpf::EbpfServices;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::MLService;
#[cfg(feature = "license")]
use crate::core::license::LicenseInfo;
#[cfg(feature = "license")]
use crate::core::license::validator::validate_license;
use crate::core::ml::config_loader::InferenceConfig;
use crate::model::direction::FlowDirection;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::misc::MiscError;
use crate::model::error::Error;
use crate::model::list_type::ListType;
/// Holds all Arc-wrapped services that make up the running application.
pub struct AppState {
pub app_config: Arc<AppConfig>,
pub inference_config: Arc<InferenceConfig>,
pub ebpf_services: Arc<EbpfServices>,
pub app_services: Arc<MLService>,
pub db: Arc<Database>,
pub jwt_service: Arc<JwtService>,
#[cfg(feature = "license")]
pub license_info: Arc<LicenseInfo>,
pub ingress_ebpf: Ebpf,
pub egress_ebpf: Ebpf,
#[allow(dead_code)]
pub ingress_program_array: ProgramArray<MapData>,
}
/// Maps stage name (from config.toml) to (function_name, stage_id).
fn stage_registry() -> HashMap<&'static str, (&'static str, u32)> {
HashMap::from([
("access_control", ("access_control", STAGE_ACCESS_CONTROL)),
("rate_limit", ("rate_limit", STAGE_RATE_LIMIT)),
("service", ("protocol_filter", STAGE_SERVICE)),
])
}
/// Factory responsible for creating and wiring all application services.
pub struct ServiceFactory;
impl ServiceFactory {
/// Build all services and return the complete application state.
pub async fn build() -> Result<AppState, Error> {
let mut ingress_ebpf = Self::load_ebpf("ingress")?;
let mut egress_ebpf = Self::load_ebpf("egress")?;
let app_config = Arc::new(AppConfig::new()?);
#[cfg(feature = "license")]
let license_info = Arc::new(validate_license(
&app_config.misc.license_file,
&app_config.network.ingress_ifname,
&app_config.network.egress_ifname,
)?);
let ingress_program_array = Self::configure_ingress_pipeline(
&mut ingress_ebpf,
&app_config.pipeline.ingress,
)?;
let inference_config = Arc::new(InferenceConfig::load_file(&app_config.inference.models_config_name)?);
// Write queue count to eBPF maps for symmetric hash redirect
let num_queues = app_config.network.combined_queue_count;
Self::write_num_queues(&mut ingress_ebpf, num_queues)?;
Self::write_num_queues(&mut egress_ebpf, num_queues)?;
let db = Arc::new(Database::new(&app_config.misc.database_path)?);
// Create default admin user if no users exist
if db.user_count().unwrap_or(0) == 0 {
let hash = password::hash_password("admin")?;
db.insert_user("admin", &hash, "admin", true)?;
tracing::warn!("Default admin user created with password 'admin' — you must change it on first login");
}
// Ensure enforce_mode setting exists (default: monitor)
if db.get_setting("enforce_mode")?.is_none() {
db.set_setting("enforce_mode", "monitor")?;
}
let jwt_service = Arc::new(JwtService::new(&db, app_config.http.jwt_expiry_hours)?);
let ebpf_services = Arc::new(EbpfServices::new(
app_config.clone(),
&mut ingress_ebpf,
&mut egress_ebpf,
)?);
let app_services = Arc::new(MLService::new(app_config.clone(), inference_config.clone())?);
// Restore persisted state from database
Self::restore_dns_blacklist(&db, &ebpf_services);
Self::restore_geo_countries(&db, &ebpf_services);
Self::restore_rate_limits(&db, &ebpf_services);
Self::restore_acl_rules(&db, &ebpf_services).await;
Ok(AppState {
app_config,
inference_config,
ebpf_services,
app_services,
db,
jwt_service,
#[cfg(feature = "license")]
license_info,
ingress_ebpf,
egress_ebpf,
ingress_program_array,
})
}
// --- eBPF loading helpers ---
fn load_ebpf(name: &str) -> Result<Ebpf, Error> {
let bytes = match name {
"ingress" => aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/net-guardia-ingress")),
"egress" => aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/net-guardia-egress")),
_ => return Err(EbpfError::ProgramNotFound.into()),
};
Ok(Ebpf::load(bytes).map_err(EbpfError::EbpfNotFound)?)
}
fn configure_ingress_pipeline(
ebpf: &mut Ebpf,
stages: &[String],
) -> Result<ProgramArray<MapData>, Error> {
let registry = stage_registry();
let entry: &mut Xdp = ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
entry.load().map_err(EbpfError::LoadProgramFailed)?;
let pa_map = ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(pa_map).map_err(EbpfError::MapOperationError)?;
let ns_map = ebpf.take_map("NEXT_STAGE").ok_or(EbpfError::MapNotFound)?;
let mut next_stage = Array::<MapData, u32>::try_from(ns_map).map_err(EbpfError::MapOperationError)?;
Self::load_program(ebpf, &mut program_array, "transmission", STAGE_TRANSMISSION)?;
if stages.is_empty() {
next_stage
.set(STAGE_ENTRY as u32, STAGE_TRANSMISSION, 0)
.map_err(EbpfError::MapOperationError)?;
return Ok(program_array);
}
let mut slots: Vec<(u32, u32)> = Vec::new();
for (i, stage_name) in stages.iter().enumerate() {
let (func_name, stage_id) = registry
.get(stage_name.as_str())
.ok_or(EbpfError::ProgramNotFound)?;
let slot = (i + 1) as u32;
Self::load_program(ebpf, &mut program_array, func_name, slot)?;
slots.push((*stage_id, slot));
}
next_stage
.set(STAGE_ENTRY as u32, slots[0].1, 0)
.map_err(EbpfError::MapOperationError)?;
for i in 0..slots.len() {
let (stage_id, _) = slots[i];
let next_slot = if i + 1 < slots.len() {
slots[i + 1].1
} else {
STAGE_TRANSMISSION
};
next_stage
.set(stage_id as u32, next_slot, 0)
.map_err(EbpfError::MapOperationError)?;
}
Ok(program_array)
}
fn load_program(
ebpf: &mut Ebpf,
program_array: &mut ProgramArray<MapData>,
function_name: &str,
slot: u32,
) -> 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(slot, fd, 0)
.map_err(EbpfError::MapOperationError)?;
Ok(())
}
fn write_num_queues(ebpf: &mut Ebpf, num_queues: u32) -> Result<(), Error> {
let map = ebpf.map_mut("NUM_QUEUES").ok_or(EbpfError::MapNotFound)?;
let mut arr = Array::<_, u32>::try_from(map).map_err(EbpfError::MapOperationError)?;
arr.set(0, num_queues, 0).map_err(EbpfError::MapOperationError)?;
Ok(())
}
pub 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 fn attach_xdp(ebpf: &mut Ebpf, ifname: &str, already_loaded: bool) -> Result<String, Error> {
let xdp: &mut Xdp = ebpf
.program_mut("net_guardia")
.ok_or(EbpfError::ProgramNotFound)?
.try_into()
.map_err(EbpfError::GetProgramFailed)?;
if !already_loaded {
xdp.load().map_err(EbpfError::LoadProgramFailed)?;
}
// Try DRV_MODE first (native XDP, best performance)
match xdp.attach(ifname, XdpFlags::DRV_MODE) {
Ok(_) => {
tracing::info!("XDP attached to {} in native DRV_MODE", ifname);
return Ok("drv".to_string());
}
Err(drv_err) => {
tracing::warn!(
"XDP DRV_MODE failed on {}: {}. Falling back to SKB_MODE.",
ifname, drv_err
);
}
}
// Fallback to SKB_MODE (generic XDP, reduced performance)
match xdp.attach(ifname, XdpFlags::SKB_MODE) {
Ok(_) => {
tracing::warn!(
"XDP attached to {} in generic SKB_MODE (reduced performance). \
For best performance, use a NIC with native XDP support (e.g., virtio-net, Intel i40e/ice).",
ifname
);
Ok("skb".to_string())
}
Err(skb_err) => {
tracing::error!(
"XDP attach failed on {} with both DRV_MODE and SKB_MODE. \
Ensure the interface exists and supports XDP. \
Supported NICs: virtio-net, Intel i40e/ice/i350, Mellanox mlx5. \
SKB error: {}",
ifname, skb_err
);
Err(EbpfError::AttachProgramFailed(skb_err).into())
}
}
}
pub fn aya_log_init(ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result<(), Error> {
EbpfLogger::init(ingress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
EbpfLogger::init(egress_ebpf).map_err(EbpfError::LoggerInitFailed)?;
Ok(())
}
// --- State restoration helpers ---
fn restore_dns_blacklist(db: &Database, ebpf_services: &EbpfServices) {
if let Ok(domains) = db.load_dns_domains() {
for domain in &domains {
if let Err(e) = ebpf_services.dns_filter.add_domain(domain) {
tracing::warn!("Failed to restore DNS domain '{}': {}", domain, e);
}
}
if !domains.is_empty() {
tracing::info!("Restored {} DNS blacklist domains from database", domains.len());
}
}
}
fn restore_geo_countries(db: &Database, ebpf_services: &EbpfServices) {
if let Ok(countries) = db.load_geo_countries() {
if !countries.is_empty() {
if let Err(e) = ebpf_services.geo_block.block_countries(&countries) {
tracing::warn!("Failed to restore geo-blocked countries: {}", e);
} else {
tracing::info!("Restored {} geo-blocked countries from database", countries.len());
}
}
}
}
fn restore_rate_limits(db: &Database, ebpf_services: &EbpfServices) {
if let Ok(configs) = db.load_rate_limit_config() {
for (key, value) in &configs {
let result = match key.as_str() {
"packet_rate" => ebpf_services.rate_limit.set_packet_rate(*value),
"syn_rate" => ebpf_services.rate_limit.set_syn_rate(*value),
"udp_rate" => ebpf_services.rate_limit.set_udp_rate(*value),
"dns_rate" => ebpf_services.rate_limit.set_dns_rate(*value),
"window_ns" => ebpf_services.rate_limit.set_window_ns(*value),
_ => Ok(()),
};
if let Err(e) = result {
tracing::warn!("Failed to restore rate limit '{}': {}", key, e);
}
}
if !configs.is_empty() {
tracing::info!("Restored {} rate limit settings from database", configs.len());
}
}
}
async fn restore_acl_rules(db: &Database, ebpf_services: &EbpfServices) {
if let Ok(rules) = db.load_acl_rules() {
let mut restored = 0u32;
for (ip_version, direction, list_type, ip_address, port) in &rules {
let dir = match direction.as_str() {
"source" => FlowDirection::Source,
"destination" => FlowDirection::Destination,
other => {
tracing::warn!("Unknown ACL direction '{}', skipping", other);
continue;
}
};
let lt = match list_type.as_str() {
"whitelist" => ListType::White,
"blacklist" => ListType::Black,
other => {
tracing::warn!("Unknown ACL list type '{}', skipping", other);
continue;
}
};
let result = match ip_version {
4 => {
match ip_address.parse::<Ipv4Addr>() {
Ok(addr) => ebpf_services.access_control.add_ipv4_list(dir, lt, SocketAddrV4::new(addr, *port)).await,
Err(e) => {
tracing::warn!("Failed to parse IPv4 address '{}': {}", ip_address, e);
continue;
}
}
}
6 => {
match ip_address.parse::<Ipv6Addr>() {
Ok(addr) => ebpf_services.access_control.add_ipv6_list(dir, lt, SocketAddrV6::new(addr, *port, 0, 0)).await,
Err(e) => {
tracing::warn!("Failed to parse IPv6 address '{}': {}", ip_address, e);
continue;
}
}
}
other => {
tracing::warn!("Unknown IP version {}, skipping", other);
continue;
}
};
if let Err(e) = result {
tracing::warn!("Failed to restore ACL rule ({} {} {}:{}): {}", direction, list_type, ip_address, port, e);
} else {
restored += 1;
}
}
if restored > 0 {
tracing::info!("Restored {} ACL rules from database", restored);
}
}
}
}

16
net-guardia/src/interface/communication/command.rs Normal file
View File

@ -0,0 +1,16 @@
use crate::interface::communication::message::Message;
use crate::model::error::Error;
use async_trait::async_trait;
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
pub type CommandFuture = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
pub type CommandHandlerFn = Box<dyn Fn(Box<dyn Any + Send>) -> CommandFuture + Send + Sync>;
pub trait Command: Message<Response = ()> {}
#[async_trait]
pub trait CommandHandler<C: Command> {
async fn handle_command(&self, command: C) -> Result<(), Error>;
}

116
net-guardia/src/interface/communication/command_types.rs Normal file
View File

@ -0,0 +1,116 @@
use crate::interface::communication::command::Command;
use crate::interface::communication::message::Message;
// ── ACL Commands ─────────────────────────────────────────────────────
pub struct AddAclRuleCommand {
pub ip_version: u8,
pub direction: String,
pub list_type: String,
pub ip_address: String,
pub port: u16,
}
impl Message for AddAclRuleCommand {
type Response = ();
}
impl Command for AddAclRuleCommand {}
pub struct RemoveAclRuleCommand {
pub ip_version: u8,
pub direction: String,
pub list_type: String,
pub ip_address: String,
pub port: u16,
}
impl Message for RemoveAclRuleCommand {
type Response = ();
}
impl Command for RemoveAclRuleCommand {}
// ── Geo Commands ─────────────────────────────────────────────────────
pub struct BlockGeoCountriesCommand {
pub country_codes: Vec<String>,
}
impl Message for BlockGeoCountriesCommand {
type Response = ();
}
impl Command for BlockGeoCountriesCommand {}
pub struct UnblockGeoCountriesCommand {
pub country_codes: Vec<String>,
}
impl Message for UnblockGeoCountriesCommand {
type Response = ();
}
impl Command for UnblockGeoCountriesCommand {}
// ── DNS Commands ─────────────────────────────────────────────────────
pub struct AddDnsDomainCommand {
pub domain: String,
}
impl Message for AddDnsDomainCommand {
type Response = ();
}
impl Command for AddDnsDomainCommand {}
pub struct RemoveDnsDomainCommand {
pub domain: String,
}
impl Message for RemoveDnsDomainCommand {
type Response = ();
}
impl Command for RemoveDnsDomainCommand {}
// ── Rate Limit Commands ──────────────────────────────────────────────
pub struct SetRateLimitCommand {
pub key: String,
pub value: u64,
}
impl Message for SetRateLimitCommand {
type Response = ();
}
impl Command for SetRateLimitCommand {}
// ── System Commands ──────────────────────────────────────────────────
pub struct ChangeEnforceModeCommand {
pub mode: String,
}
impl Message for ChangeEnforceModeCommand {
type Response = ();
}
impl Command for ChangeEnforceModeCommand {}
// ── Auth Commands ────────────────────────────────────────────────────
pub struct ChangePasswordCommand {
pub user_id: i64,
pub new_password_hash: String,
}
impl Message for ChangePasswordCommand {
type Response = ();
}
impl Command for ChangePasswordCommand {}
pub struct RegisterUserCommand {
pub username: String,
pub password_hash: String,
pub role: String,
}
impl Message for RegisterUserCommand {
type Response = ();
}
impl Command for RegisterUserCommand {}

9
net-guardia/src/interface/communication/event.rs Normal file
View File

@ -0,0 +1,9 @@
use crate::model::error::Error;
use std::any::Any;
pub trait Event: Send + Clone + 'static {}
pub trait EventBroadcaster: Send + Sync {
fn subscribe_typed(&self) -> Box<dyn Any + Send>;
fn broadcast_event(&self, event: Box<dyn Any + Send>) -> Result<(), Error>;
}

83
net-guardia/src/interface/communication/event_types.rs Normal file
View File

@ -0,0 +1,83 @@
use crate::interface::communication::event::Event;
use crate::model::direction::Direction;
// ── ML Events ────────────────────────────────────────────────────────
/// Fired when the ML engine detects a potential threat.
#[derive(Debug, Clone)]
pub struct ThreatDetectedEvent {
pub flow_key: String,
pub direction: Direction,
pub attack_type: String,
pub confidence: f32,
pub ae_score: f32,
}
impl Event for ThreatDetectedEvent {}
/// Fired after each ML inference tick with summary stats.
#[derive(Debug, Clone)]
pub struct InferenceCompletedEvent {
pub total_flows: usize,
pub malicious_flows: usize,
pub benign_flows: usize,
pub elapsed_ms: u32,
}
impl Event for InferenceCompletedEvent {}
// ── System Events ────────────────────────────────────────────────────
/// Fired when enforce mode changes (monitor ↔ enforce).
#[derive(Debug, Clone)]
pub struct EnforceModeChangedEvent {
pub old_mode: String,
pub new_mode: String,
}
impl Event for EnforceModeChangedEvent {}
/// Fired when XDP attachment completes (or falls back).
#[derive(Debug, Clone)]
pub struct XdpAttachedEvent {
pub interface: String,
pub mode: String, // "drv" or "skb"
}
impl Event for XdpAttachedEvent {}
// ── ACL Events ───────────────────────────────────────────────────────
/// Fired when an ACL rule is added or removed.
#[derive(Debug, Clone)]
pub struct AclRuleChangedEvent {
pub action: String, // "added" or "removed"
pub ip_version: u8,
pub direction: String,
pub list_type: String,
pub ip_address: String,
pub port: u16,
}
impl Event for AclRuleChangedEvent {}
// ── Auth Events ──────────────────────────────────────────────────────
/// Fired when a login attempt fails (for auditing).
#[derive(Debug, Clone)]
pub struct LoginFailedEvent {
pub username: String,
pub failure_count: u32,
pub locked: bool,
}
impl Event for LoginFailedEvent {}
/// Fired when a user changes their password.
#[derive(Debug, Clone)]
pub struct PasswordChangedEvent {
pub user_id: i64,
pub username: String,
}
impl Event for PasswordChangedEvent {}

3
net-guardia/src/interface/communication/message.rs Normal file
View File

@ -0,0 +1,3 @@
pub trait Message: Send + 'static {
type Response: Send + 'static;
}

7
net-guardia/src/interface/communication/mod.rs Normal file
View File

@ -0,0 +1,7 @@
pub mod message;
pub mod command;
pub mod query;
pub mod event;
pub mod command_types;
pub mod query_types;
pub mod event_types;

16
net-guardia/src/interface/communication/query.rs Normal file
View File

@ -0,0 +1,16 @@
use crate::interface::communication::message::Message;
use crate::model::error::Error;
use async_trait::async_trait;
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
pub type QueryFuture = Pin<Box<dyn Future<Output = Result<Box<dyn Any + Send>, Error>> + Send + 'static>>;
pub type QueryHandlerFn = Box<dyn Fn(Box<dyn Any + Send>) -> QueryFuture + Send + Sync>;
pub trait Query: Message {}
#[async_trait]
pub trait QueryHandler<Q: Query> {
async fn handle_query(&self, query: Q) -> Result<Q::Response, Error>;
}

88
net-guardia/src/interface/communication/query_types.rs Normal file
View File

@ -0,0 +1,88 @@
use crate::interface::communication::message::Message;
use crate::interface::communication::query::Query;
use crate::model::health::{SystemHealthMetrics, SystemHealthStatus};
// ── System Queries ───────────────────────────────────────────────────
pub struct GetEnforceModeQuery;
impl Message for GetEnforceModeQuery {
type Response = String;
}
impl Query for GetEnforceModeQuery {}
pub struct GetXdpModeQuery;
impl Message for GetXdpModeQuery {
type Response = XdpModeResponse;
}
impl Query for GetXdpModeQuery {}
#[derive(Debug, Clone)]
pub struct XdpModeResponse {
pub ingress_mode: String,
pub egress_mode: String,
}
// ── Health Queries ───────────────────────────────────────────────────
pub struct GetHealthMetricsQuery;
impl Message for GetHealthMetricsQuery {
type Response = SystemHealthMetrics;
}
impl Query for GetHealthMetricsQuery {}
pub struct GetHealthStatusQuery;
impl Message for GetHealthStatusQuery {
type Response = SystemHealthStatus;
}
impl Query for GetHealthStatusQuery {}
// ── ACL Queries ──────────────────────────────────────────────────────
pub struct GetAclRulesQuery;
impl Message for GetAclRulesQuery {
type Response = Vec<(u8, String, String, String, u16)>;
}
impl Query for GetAclRulesQuery {}
// ── Settings Queries ─────────────────────────────────────────────────
pub struct GetSettingQuery {
pub key: String,
}
impl Message for GetSettingQuery {
type Response = Option<String>;
}
impl Query for GetSettingQuery {}
// ── Rate Limit Queries ───────────────────────────────────────────────
pub struct GetRateLimitConfigQuery;
impl Message for GetRateLimitConfigQuery {
type Response = Vec<(String, u64)>;
}
impl Query for GetRateLimitConfigQuery {}
// ── DNS Queries ──────────────────────────────────────────────────────
pub struct GetDnsDomainsQuery;
impl Message for GetDnsDomainsQuery {
type Response = Vec<String>;
}
impl Query for GetDnsDomainsQuery {}
// ── Geo Queries ──────────────────────────────────────────────────────
pub struct GetGeoBlockedCountriesQuery;
impl Message for GetGeoBlockedCountriesQuery {
type Response = Vec<String>;
}
impl Query for GetGeoBlockedCountriesQuery {}

2
net-guardia/src/interface/mod.rs Normal file
View File

@ -0,0 +1,2 @@
pub mod communication;
pub mod port;

19
net-guardia/src/interface/port/auth.rs Normal file
View File

@ -0,0 +1,19 @@
use crate::model::error::Error;
/// Claims extracted from a validated JWT token.
#[derive(Debug, Clone)]
pub struct TokenClaims {
pub sub: i64,
pub username: String,
pub role: String,
pub exp: usize,
}
/// Port for authentication operations.
/// Adapters: JWT (current), could be OAuth, etc.
pub trait AuthPort: Send + Sync {
fn create_token(&self, user_id: i64, username: &str, role: &str) -> Result<String, Error>;
fn validate_token(&self, token: &str) -> Result<TokenClaims, Error>;
fn hash_password(&self, password: &str) -> Result<String, Error>;
fn verify_password(&self, password: &str, hash: &str) -> Result<bool, Error>;
}

11
net-guardia/src/interface/port/health.rs Normal file
View File

@ -0,0 +1,11 @@
use crate::model::error::Error;
use crate::model::health::{SystemHealthMetrics, SystemHealthStatus};
use async_trait::async_trait;
/// Port for system health monitoring.
/// Adapters: sysinfo-based (current)
#[async_trait]
pub trait HealthPort: Send + Sync {
async fn get_metrics(&self) -> SystemHealthMetrics;
async fn is_healthy(&self) -> SystemHealthStatus;
}

4
net-guardia/src/interface/port/mod.rs Normal file
View File

@ -0,0 +1,4 @@
pub mod repository;
pub mod auth;
pub mod notification;
pub mod health;

9
net-guardia/src/interface/port/notification.rs Normal file
View File

@ -0,0 +1,9 @@
use crate::model::error::Error;
use async_trait::async_trait;
/// Port for outbound notifications (alerts, reports).
/// Adapters: WebSocket (alerts), SMTP (weekly report)
#[async_trait]
pub trait NotificationPort: Send + Sync {
async fn send_weekly_report(&self) -> Result<(), Error>;
}

39
net-guardia/src/interface/port/repository.rs Normal file
View File

@ -0,0 +1,39 @@
use crate::model::error::Error;
/// Port for persistent storage operations.
/// Adapters: SQLite (current), could be Postgres, etc.
pub trait RepositoryPort: Send + Sync {
// --- ACL ---
fn insert_acl_rule(&self, ip_version: u8, direction: &str, list_type: &str, ip_address: &str, port: u16) -> Result<(), Error>;
fn delete_acl_rule(&self, ip_version: u8, direction: &str, list_type: &str, ip_address: &str, port: u16) -> Result<(), Error>;
fn load_acl_rules(&self) -> Result<Vec<(u8, String, String, String, u16)>, Error>;
// --- Rate Limit ---
fn set_rate_limit(&self, key: &str, value: u64) -> Result<(), Error>;
fn load_rate_limit_config(&self) -> Result<Vec<(String, u64)>, Error>;
// --- DNS ---
fn insert_dns_domain(&self, domain: &str) -> Result<(), Error>;
fn delete_dns_domain(&self, domain: &str) -> Result<(), Error>;
fn load_dns_domains(&self) -> Result<Vec<String>, Error>;
// --- Geo ---
fn insert_geo_country(&self, code: &str) -> Result<(), Error>;
fn delete_geo_country(&self, code: &str) -> Result<(), Error>;
fn load_geo_countries(&self) -> Result<Vec<String>, Error>;
// --- Settings ---
fn get_setting(&self, key: &str) -> Result<Option<String>, Error>;
fn set_setting(&self, key: &str, value: &str) -> Result<(), Error>;
// --- Users ---
fn find_user(&self, username: &str) -> Result<Option<(i64, String, String, String, bool)>, Error>;
fn insert_user(&self, username: &str, password_hash: &str, role: &str, force_password_change: bool) -> Result<(), Error>;
fn update_user_password(&self, user_id: i64, password_hash: &str) -> Result<(), Error>;
fn user_count(&self) -> Result<i64, Error>;
// --- Login Rate Limiting ---
fn record_login_failure(&self, username: &str) -> Result<(u32, Option<u64>), Error>;
fn check_login_locked(&self, username: &str) -> Result<Option<u64>, Error>;
fn clear_login_failures(&self, username: &str) -> Result<(), Error>;
}

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

@ -1,7 +1,9 @@
mod adapter;
mod core;
mod infrastructure;
mod interface;
mod model;
mod utils;
mod web;
use crate::core::system::System;
use crate::model::error::Error;

12
net-guardia/src/model/error/misc.rs
View File

@ -30,5 +30,17 @@ traceable! {
#[no_source]
#[error("Invalid DNS domain name: {reason}")]
InvalidDnsName { reason: String } => tracing::Level::WARN,
#[no_source]
#[error("Type mismatch during message dispatch")]
TypeMismatch => tracing::Level::ERROR,
#[no_source]
#[error("No handler registered for this message type")]
HandlerNotFound => tracing::Level::ERROR,
#[no_source]
#[error("Event type not registered with communication manager")]
TypeNotRegistered => tracing::Level::ERROR,
}
}

21
net-guardia/src/model/flow_stats.rs
View File

@ -1,6 +1,5 @@
use serde::{Deserialize, Serialize};
use crate::core::ml::flow_tracker::FlowData;
use crate::model::direction::Direction;
#[derive(Debug, Clone, Serialize)]
@ -19,24 +18,8 @@ pub struct FlowStatsEntry {
pub last_seen_us: u64,
}
impl From<&FlowData> for FlowStatsEntry {
fn from(flow: &FlowData) -> Self {
Self {
direction: flow.direction,
src_ip: flow.flow_key.src_ip_string(),
dst_ip: flow.flow_key.dst_ip_string(),
src_port: flow.flow_key.src_port,
dst_port: flow.flow_key.dst_port,
protocol: flow.flow_key.protocol,
fwd_packets: flow.fwd_packets.len(),
bwd_packets: flow.bwd_packets.len(),
fwd_bytes: flow.fwd_total_bytes,
bwd_bytes: flow.bwd_total_bytes,
duration_us: flow.duration_us(),
last_seen_us: flow.last_time_us,
}
}
}
// NOTE: From<&FlowData> impl moved to core/infrastructure/statistics.rs
// to maintain the dependency rule: model/ must not import core/
#[derive(Debug, Clone, Serialize)]
pub struct StatsSummary {

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

@ -1,2 +0,0 @@
pub mod api;
pub mod websocket;