From c24bd50d7ef4858699eb76c9487e20ce79df05ef Mon Sep 17 00:00:00 2001 From: DaLaw2 Date: Sat, 21 Mar 2026 23:52:01 +0800 Subject: [PATCH] refactor: hexagonal architecture with ports, adapters, and CQRS event bus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- Cargo.lock | 17 +- net-guardia/Cargo.toml | 4 + .../src/{web/api => adapter/http}/acl.rs | 16 +- .../src/{web/api => adapter/http}/auth.rs | 10 +- .../src/{web/api => adapter/http}/default.rs | 2 +- .../src/{web/api => adapter/http}/filter.rs | 8 +- .../src/{web/api => adapter/http}/health.rs | 0 .../src/{web/api => adapter/http}/ml.rs | 0 .../src/{web/api => adapter/http}/mod.rs | 1 - .../{web/api => adapter/http}/rate_limit.rs | 6 +- .../src/{web/api => adapter/http}/stats.rs | 0 .../src/{web/api => adapter/http}/system.rs | 10 +- net-guardia/src/adapter/mod.rs | 3 + .../database => adapter/persistence}/mod.rs | 0 .../persistence}/repository.rs | 47 ++ .../websocket/alert_websocket.rs | 0 .../websocket/drop_websocket.rs | 0 .../websocket/flow_websocket.rs | 0 .../websocket/health_websocket.rs | 0 .../src/{web => adapter}/websocket/mod.rs | 1 + .../api/ws.rs => adapter/websocket/routes.rs} | 2 +- net-guardia/src/core/auth/jwt.rs | 2 +- net-guardia/src/core/email/report.rs | 2 +- net-guardia/src/core/email/scheduler.rs | 2 +- .../src/core/infrastructure/statistics.rs | 22 + net-guardia/src/core/mod.rs | 1 - net-guardia/src/core/system.rs | 439 ++---------------- .../infrastructure/communication_manager.rs | 360 ++++++++++++++ net-guardia/src/infrastructure/http_server.rs | 97 ++++ net-guardia/src/infrastructure/mod.rs | 3 + .../src/infrastructure/service_factory.rs | 389 ++++++++++++++++ .../src/interface/communication/command.rs | 16 + .../interface/communication/command_types.rs | 116 +++++ .../src/interface/communication/event.rs | 9 + .../interface/communication/event_types.rs | 83 ++++ .../src/interface/communication/message.rs | 3 + .../src/interface/communication/mod.rs | 7 + .../src/interface/communication/query.rs | 16 + .../interface/communication/query_types.rs | 88 ++++ net-guardia/src/interface/mod.rs | 2 + net-guardia/src/interface/port/auth.rs | 19 + net-guardia/src/interface/port/health.rs | 11 + net-guardia/src/interface/port/mod.rs | 4 + .../src/interface/port/notification.rs | 9 + net-guardia/src/interface/port/repository.rs | 39 ++ net-guardia/src/main.rs | 4 +- net-guardia/src/model/error/misc.rs | 12 + net-guardia/src/model/flow_stats.rs | 21 +- net-guardia/src/web/mod.rs | 2 - 49 files changed, 1449 insertions(+), 456 deletions(-) rename net-guardia/src/{web/api => adapter/http}/acl.rs (96%) rename net-guardia/src/{web/api => adapter/http}/auth.rs (97%) rename net-guardia/src/{web/api => adapter/http}/default.rs (99%) rename net-guardia/src/{web/api => adapter/http}/filter.rs (98%) rename net-guardia/src/{web/api => adapter/http}/health.rs (100%) rename net-guardia/src/{web/api => adapter/http}/ml.rs (100%) rename net-guardia/src/{web/api => adapter/http}/mod.rs (92%) rename net-guardia/src/{web/api => adapter/http}/rate_limit.rs (96%) rename net-guardia/src/{web/api => adapter/http}/stats.rs (100%) rename net-guardia/src/{web/api => adapter/http}/system.rs (90%) create mode 100644 net-guardia/src/adapter/mod.rs rename net-guardia/src/{core/database => adapter/persistence}/mod.rs (100%) rename net-guardia/src/{core/database => adapter/persistence}/repository.rs (81%) rename net-guardia/src/{web => adapter}/websocket/alert_websocket.rs (100%) rename net-guardia/src/{web => adapter}/websocket/drop_websocket.rs (100%) rename net-guardia/src/{web => adapter}/websocket/flow_websocket.rs (100%) rename net-guardia/src/{web => adapter}/websocket/health_websocket.rs (100%) rename net-guardia/src/{web => adapter}/websocket/mod.rs (86%) rename net-guardia/src/{web/api/ws.rs => adapter/websocket/routes.rs} (97%) create mode 100644 net-guardia/src/infrastructure/communication_manager.rs create mode 100644 net-guardia/src/infrastructure/http_server.rs create mode 100644 net-guardia/src/infrastructure/mod.rs create mode 100644 net-guardia/src/infrastructure/service_factory.rs create mode 100644 net-guardia/src/interface/communication/command.rs create mode 100644 net-guardia/src/interface/communication/command_types.rs create mode 100644 net-guardia/src/interface/communication/event.rs create mode 100644 net-guardia/src/interface/communication/event_types.rs create mode 100644 net-guardia/src/interface/communication/message.rs create mode 100644 net-guardia/src/interface/communication/mod.rs create mode 100644 net-guardia/src/interface/communication/query.rs create mode 100644 net-guardia/src/interface/communication/query_types.rs create mode 100644 net-guardia/src/interface/mod.rs create mode 100644 net-guardia/src/interface/port/auth.rs create mode 100644 net-guardia/src/interface/port/health.rs create mode 100644 net-guardia/src/interface/port/mod.rs create mode 100644 net-guardia/src/interface/port/notification.rs create mode 100644 net-guardia/src/interface/port/repository.rs delete mode 100644 net-guardia/src/web/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 2265cb0..a0b17c7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", ] diff --git a/net-guardia/Cargo.toml b/net-guardia/Cargo.toml index 8727e17..aa47a4f 100644 --- a/net-guardia/Cargo.toml +++ b/net-guardia/Cargo.toml @@ -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 } diff --git a/net-guardia/src/web/api/acl.rs b/net-guardia/src/adapter/http/acl.rs similarity index 96% rename from net-guardia/src/web/api/acl.rs rename to net-guardia/src/adapter/http/acl.rs index 5d27325..1c11161 100644 --- a/net-guardia/src/web/api/acl.rs +++ b/net-guardia/src/adapter/http/acl.rs @@ -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, path: web::Path<(FlowDirection, ListType)>, access_control: web::Data, - db: web::Data, + db: web::Data, ) -> 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, path: web::Path<(FlowDirection, ListType)>, access_control: web::Data, - db: web::Data, + db: web::Data, ) -> 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, path: web::Path<(FlowDirection, ListType)>, access_control: web::Data, - db: web::Data, + db: web::Data, ) -> 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, path: web::Path<(FlowDirection, ListType)>, access_control: web::Data, - db: web::Data, + db: web::Data, ) -> 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, geo_block: web::Data, - db: web::Data, + db: web::Data, ) -> 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, geo_block: web::Data, - db: web::Data, + db: web::Data, ) -> impl Responder { let codes = body.into_inner().country_codes; for code in &codes { diff --git a/net-guardia/src/web/api/auth.rs b/net-guardia/src/adapter/http/auth.rs similarity index 97% rename from net-guardia/src/web/api/auth.rs rename to net-guardia/src/adapter/http/auth.rs index b250b23..7c0ac31 100644 --- a/net-guardia/src/web/api/auth.rs +++ b/net-guardia/src/adapter/http/auth.rs @@ -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, - db: web::Data, + db: web::Data, jwt: web::Data, ) -> impl Responder { let req = body.into_inner(); @@ -106,7 +108,7 @@ async fn login( async fn register( req: HttpRequest, body: web::Json, - db: web::Data, + db: web::Data, ) -> impl Responder { // Check caller is admin let claims = req.extensions().get::().cloned(); @@ -166,7 +168,7 @@ async fn me(req: HttpRequest) -> impl Responder { async fn change_password( req: HttpRequest, body: web::Json, - db: web::Data, + db: web::Data, ) -> impl Responder { let claims = match req.extensions().get::().cloned() { Some(c) => c, diff --git a/net-guardia/src/web/api/default.rs b/net-guardia/src/adapter/http/default.rs similarity index 99% rename from net-guardia/src/web/api/default.rs rename to net-guardia/src/adapter/http/default.rs index b46f256..215f180 100644 --- a/net-guardia/src/web/api/default.rs +++ b/net-guardia/src/adapter/http/default.rs @@ -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"), } -} \ No newline at end of file +} diff --git a/net-guardia/src/web/api/filter.rs b/net-guardia/src/adapter/http/filter.rs similarity index 98% rename from net-guardia/src/web/api/filter.rs rename to net-guardia/src/adapter/http/filter.rs index 7eb2dc7..48be8c3 100644 --- a/net-guardia/src/web/api/filter.rs +++ b/net-guardia/src/adapter/http/filter.rs @@ -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) -> impl Responder { async fn add_dns_blacklist( payload: web::Json, service: web::Data, - db: web::Data, + db: web::Data, ) -> 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, service: web::Data, - db: web::Data, + db: web::Data, ) -> impl Responder { let domains = payload.into_inner().domains; for domain in &domains { diff --git a/net-guardia/src/web/api/health.rs b/net-guardia/src/adapter/http/health.rs similarity index 100% rename from net-guardia/src/web/api/health.rs rename to net-guardia/src/adapter/http/health.rs diff --git a/net-guardia/src/web/api/ml.rs b/net-guardia/src/adapter/http/ml.rs similarity index 100% rename from net-guardia/src/web/api/ml.rs rename to net-guardia/src/adapter/http/ml.rs diff --git a/net-guardia/src/web/api/mod.rs b/net-guardia/src/adapter/http/mod.rs similarity index 92% rename from net-guardia/src/web/api/mod.rs rename to net-guardia/src/adapter/http/mod.rs index ac66ed7..bd119f0 100644 --- a/net-guardia/src/web/api/mod.rs +++ b/net-guardia/src/adapter/http/mod.rs @@ -7,4 +7,3 @@ pub mod ml; pub mod rate_limit; pub mod stats; pub mod system; -pub mod ws; diff --git a/net-guardia/src/web/api/rate_limit.rs b/net-guardia/src/adapter/http/rate_limit.rs similarity index 96% rename from net-guardia/src/web/api/rate_limit.rs rename to net-guardia/src/adapter/http/rate_limit.rs index f09fc87..1b5f4a7 100644 --- a/net-guardia/src/web/api/rate_limit.rs +++ b/net-guardia/src/adapter/http/rate_limit.rs @@ -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, config: web::Data, - db: web::Data, + db: web::Data, ) -> impl Responder { let s = settings.into_inner(); if let Some(v) = s.packet_rate { diff --git a/net-guardia/src/web/api/stats.rs b/net-guardia/src/adapter/http/stats.rs similarity index 100% rename from net-guardia/src/web/api/stats.rs rename to net-guardia/src/adapter/http/stats.rs diff --git a/net-guardia/src/web/api/system.rs b/net-guardia/src/adapter/http/system.rs similarity index 90% rename from net-guardia/src/web/api/system.rs rename to net-guardia/src/adapter/http/system.rs index 7eb7b7e..4ea1859 100644 --- a/net-guardia/src/web/api/system.rs +++ b/net-guardia/src/adapter/http/system.rs @@ -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) -> impl Responder { +async fn get_enforce_mode(db: web::Data) -> 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) -> impl Responder { async fn set_enforce_mode( body: web::Json, - db: web::Data, + db: web::Data, ) -> 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) -> impl Responder { +async fn get_xdp_mode(db: web::Data) -> 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") diff --git a/net-guardia/src/adapter/mod.rs b/net-guardia/src/adapter/mod.rs new file mode 100644 index 0000000..91f4394 --- /dev/null +++ b/net-guardia/src/adapter/mod.rs @@ -0,0 +1,3 @@ +pub mod http; +pub mod persistence; +pub mod websocket; diff --git a/net-guardia/src/core/database/mod.rs b/net-guardia/src/adapter/persistence/mod.rs similarity index 100% rename from net-guardia/src/core/database/mod.rs rename to net-guardia/src/adapter/persistence/mod.rs diff --git a/net-guardia/src/core/database/repository.rs b/net-guardia/src/adapter/persistence/repository.rs similarity index 81% rename from net-guardia/src/core/database/repository.rs rename to net-guardia/src/adapter/persistence/repository.rs index 0ecb3bf..c67d393 100644 --- a/net-guardia/src/core/database/repository.rs +++ b/net-guardia/src/adapter/persistence/repository.rs @@ -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, 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, 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, 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, Error> { self.load_geo_countries() } + fn get_setting(&self, key: &str) -> Result, 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, 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 { self.user_count() } + fn record_login_failure(&self, username: &str) -> Result<(u32, Option), Error> { self.record_login_failure(username) } + fn check_login_locked(&self, username: &str) -> Result, 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); + } } diff --git a/net-guardia/src/web/websocket/alert_websocket.rs b/net-guardia/src/adapter/websocket/alert_websocket.rs similarity index 100% rename from net-guardia/src/web/websocket/alert_websocket.rs rename to net-guardia/src/adapter/websocket/alert_websocket.rs diff --git a/net-guardia/src/web/websocket/drop_websocket.rs b/net-guardia/src/adapter/websocket/drop_websocket.rs similarity index 100% rename from net-guardia/src/web/websocket/drop_websocket.rs rename to net-guardia/src/adapter/websocket/drop_websocket.rs diff --git a/net-guardia/src/web/websocket/flow_websocket.rs b/net-guardia/src/adapter/websocket/flow_websocket.rs similarity index 100% rename from net-guardia/src/web/websocket/flow_websocket.rs rename to net-guardia/src/adapter/websocket/flow_websocket.rs diff --git a/net-guardia/src/web/websocket/health_websocket.rs b/net-guardia/src/adapter/websocket/health_websocket.rs similarity index 100% rename from net-guardia/src/web/websocket/health_websocket.rs rename to net-guardia/src/adapter/websocket/health_websocket.rs diff --git a/net-guardia/src/web/websocket/mod.rs b/net-guardia/src/adapter/websocket/mod.rs similarity index 86% rename from net-guardia/src/web/websocket/mod.rs rename to net-guardia/src/adapter/websocket/mod.rs index d4e5d08..a488f84 100644 --- a/net-guardia/src/web/websocket/mod.rs +++ b/net-guardia/src/adapter/websocket/mod.rs @@ -2,3 +2,4 @@ pub mod alert_websocket; pub mod drop_websocket; pub mod flow_websocket; pub mod health_websocket; +pub mod routes; diff --git a/net-guardia/src/web/api/ws.rs b/net-guardia/src/adapter/websocket/routes.rs similarity index 97% rename from net-guardia/src/web/api/ws.rs rename to net-guardia/src/adapter/websocket/routes.rs index 8e38114..2117e18 100644 --- a/net-guardia/src/web/api/ws.rs +++ b/net-guardia/src/adapter/websocket/routes.rs @@ -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 { diff --git a/net-guardia/src/core/auth/jwt.rs b/net-guardia/src/core/auth/jwt.rs index 0cb516c..c8aeb6d 100644 --- a/net-guardia/src/core/auth/jwt.rs +++ b/net-guardia/src/core/auth/jwt.rs @@ -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; diff --git a/net-guardia/src/core/email/report.rs b/net-guardia/src/core/email/report.rs index 56b4aa9..03c7015 100644 --- a/net-guardia/src/core/email/report.rs +++ b/net-guardia/src/core/email/report.rs @@ -1,4 +1,4 @@ -use crate::core::database::Database; +use crate::adapter::persistence::Database; use crate::model::error::Error; use std::sync::Arc; diff --git a/net-guardia/src/core/email/scheduler.rs b/net-guardia/src/core/email/scheduler.rs index 4443243..6c71d7d 100644 --- a/net-guardia/src/core/email/scheduler.rs +++ b/net-guardia/src/core/email/scheduler.rs @@ -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; diff --git a/net-guardia/src/core/infrastructure/statistics.rs b/net-guardia/src/core/infrastructure/statistics.rs index 15014b7..11209dc 100644 --- a/net-guardia/src/core/infrastructure/statistics.rs +++ b/net-guardia/src/core/infrastructure/statistics.rs @@ -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, } diff --git a/net-guardia/src/core/mod.rs b/net-guardia/src/core/mod.rs index 428fd52..856e8d1 100644 --- a/net-guardia/src/core/mod.rs +++ b/net-guardia/src/core/mod.rs @@ -1,5 +1,4 @@ pub mod auth; -pub mod database; pub mod email; pub mod ebpf; pub mod infrastructure; diff --git a/net-guardia/src/core/system.rs b/net-guardia/src/core/system.rs index 4469533..4cb2e2f 100644 --- a/net-guardia/src/core/system.rs +++ b/net-guardia/src/core/system.rs @@ -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, pub inference_config: Arc, @@ -59,163 +40,19 @@ pub struct System { impl System { pub async fn new() -> Result { - 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::() { - 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::() { - 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 { - 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 { - 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, 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::::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, - 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 } } diff --git a/net-guardia/src/infrastructure/communication_manager.rs b/net-guardia/src/infrastructure/communication_manager.rs new file mode 100644 index 0000000..02511db --- /dev/null +++ b/net-guardia/src/infrastructure/communication_manager.rs @@ -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 { + pub sender: broadcast::Sender, +} + +impl EventBroadcaster for TypedEventBroadcaster { + fn subscribe_typed(&self) -> Box { + Box::new(self.sender.subscribe()) + } + + fn broadcast_event(&self, event: Box) -> Result<(), Error> { + let typed_event = *event.downcast::().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, + query_handlers: DashMap, + event_broadcasters: DashMap>, + 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( + self: Arc, + service: Arc, + ) -> ServiceRegistrar { + ServiceRegistrar::new(service, self) + } + + pub fn register_command_handler( + &self, + handler: Arc + Send + Sync>, + ) { + let type_id = TypeId::of::(); + let boxed_handler: CommandHandlerFn = Box::new(move |command: Box| { + let handler = handler.clone(); + Box::pin(async move { + let command = *command + .downcast::() + .map_err(|_| MiscError::TypeMismatch)?; + handler.handle_command(command).await + }) as CommandFuture + }); + + self.command_handlers.insert(type_id, boxed_handler); + } + + pub async fn send_command(&self, command: C) -> Result<(), Error> { + let type_id = TypeId::of::(); + if let Some(handler) = self.command_handlers.get(&type_id) { + handler(Box::new(command)).await + } else { + Err(MiscError::HandlerNotFound)? + } + } + + pub fn register_query_handler( + &self, + handler: Arc + Send + Sync>, + ) { + let type_id = TypeId::of::(); + let boxed_handler: QueryHandlerFn = Box::new(move |query: Box| { + let handler = handler.clone(); + Box::pin(async move { + let query = *query.downcast::().map_err(|_| MiscError::TypeMismatch)?; + let response = handler.handle_query(query).await?; + Ok(Box::new(response) as Box) + }) as QueryFuture + }); + + self.query_handlers.insert(type_id, boxed_handler); + } + + pub async fn send_query(&self, query: Q) -> Result { + let type_id = TypeId::of::(); + if let Some(handler) = self.query_handlers.get(&type_id) { + let response = handler(Box::new(query)).await?; + Ok(*response + .downcast::() + .map_err(|_| MiscError::TypeMismatch)?) + } else { + Err(MiscError::HandlerNotFound)? + } + } + + pub fn register_event_type(&self) { + let type_id = TypeId::of::(); + let (tx, _) = broadcast::channel::(self.channel_capacity); + let broadcaster = TypedEventBroadcaster { sender: tx }; + self.event_broadcasters + .insert(type_id, Box::new(broadcaster)); + } + + pub fn subscribe_event(&self) -> Result, Error> { + let type_id = TypeId::of::(); + let broadcaster = self + .event_broadcasters + .get(&type_id) + .ok_or(MiscError::TypeNotRegistered)?; + let receiver_box = broadcaster.subscribe_typed(); + let receiver = *receiver_box + .downcast::>() + .map_err(|_| MiscError::TypeMismatch)?; + Ok(receiver) + } + + pub async fn publish_event(&self, event: E) -> Result<(), Error> { + let type_id = TypeId::of::(); + 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 { + service: Arc, + comm: Arc, +} + +impl ServiceRegistrar { + fn new(service: Arc, comm: Arc) -> Self { + Self { service, comm } + } + + pub fn command(self) -> Self + where + S: CommandHandler, + { + let handler: Arc + Send + Sync> = self.service.clone(); + self.comm.register_command_handler::(handler); + self + } + + pub fn query(self) -> Self + where + S: QueryHandler, + { + let handler: Arc + Send + Sync> = self.service.clone(); + self.comm.register_query_handler::(handler); + self + } + + pub fn event(self) -> Self { + self.comm.register_event_type::(); + self + } + + pub fn build(self) -> Arc { + 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>>, + } + + #[async_trait] + impl CommandHandler 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 for TestQueryHandler { + async fn handle_query(&self, query: TestQuery) -> Result { + 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::(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::(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::(); + + let mut receiver = comm.subscribe_event::().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::(); + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_event_multiple_subscribers() { + let comm = CommunicationManager::new(); + comm.register_event_type::(); + + let mut rx1 = comm.subscribe_event::().unwrap(); + let mut rx2 = comm.subscribe_event::().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::() + .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::(); + assert!(comm.subscribe_event::().is_ok()); + + comm.clear_handlers(); + assert!(comm.subscribe_event::().is_err()); + } +} diff --git a/net-guardia/src/infrastructure/http_server.rs b/net-guardia/src/infrastructure/http_server.rs new file mode 100644 index 0000000..ddfceb1 --- /dev/null +++ b/net-guardia/src/infrastructure/http_server.rs @@ -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, + pub inference_config: Arc, + pub ebpf_services: Arc, + pub app_services: Arc, + pub db: Arc, + pub jwt_service: Arc, + #[cfg(feature = "license")] + pub license_info: Arc, +} + +/// 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)) + .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(()) +} diff --git a/net-guardia/src/infrastructure/mod.rs b/net-guardia/src/infrastructure/mod.rs new file mode 100644 index 0000000..e724318 --- /dev/null +++ b/net-guardia/src/infrastructure/mod.rs @@ -0,0 +1,3 @@ +pub mod communication_manager; +pub mod http_server; +pub mod service_factory; diff --git a/net-guardia/src/infrastructure/service_factory.rs b/net-guardia/src/infrastructure/service_factory.rs new file mode 100644 index 0000000..dac3b59 --- /dev/null +++ b/net-guardia/src/infrastructure/service_factory.rs @@ -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, + pub inference_config: Arc, + pub ebpf_services: Arc, + pub app_services: Arc, + pub db: Arc, + pub jwt_service: Arc, + #[cfg(feature = "license")] + pub license_info: Arc, + pub ingress_ebpf: Ebpf, + pub egress_ebpf: Ebpf, + #[allow(dead_code)] + pub ingress_program_array: ProgramArray, +} + +/// 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 { + 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 { + 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, 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::::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, + 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 { + 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::() { + 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::() { + 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); + } + } + } +} diff --git a/net-guardia/src/interface/communication/command.rs b/net-guardia/src/interface/communication/command.rs new file mode 100644 index 0000000..31ada71 --- /dev/null +++ b/net-guardia/src/interface/communication/command.rs @@ -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> + Send + 'static>>; +pub type CommandHandlerFn = Box) -> CommandFuture + Send + Sync>; + +pub trait Command: Message {} + +#[async_trait] +pub trait CommandHandler { + async fn handle_command(&self, command: C) -> Result<(), Error>; +} diff --git a/net-guardia/src/interface/communication/command_types.rs b/net-guardia/src/interface/communication/command_types.rs new file mode 100644 index 0000000..708ccbb --- /dev/null +++ b/net-guardia/src/interface/communication/command_types.rs @@ -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, +} + +impl Message for BlockGeoCountriesCommand { + type Response = (); +} +impl Command for BlockGeoCountriesCommand {} + +pub struct UnblockGeoCountriesCommand { + pub country_codes: Vec, +} + +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 {} diff --git a/net-guardia/src/interface/communication/event.rs b/net-guardia/src/interface/communication/event.rs new file mode 100644 index 0000000..faeaa07 --- /dev/null +++ b/net-guardia/src/interface/communication/event.rs @@ -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; + fn broadcast_event(&self, event: Box) -> Result<(), Error>; +} diff --git a/net-guardia/src/interface/communication/event_types.rs b/net-guardia/src/interface/communication/event_types.rs new file mode 100644 index 0000000..bfcf470 --- /dev/null +++ b/net-guardia/src/interface/communication/event_types.rs @@ -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 {} diff --git a/net-guardia/src/interface/communication/message.rs b/net-guardia/src/interface/communication/message.rs new file mode 100644 index 0000000..aaf1df9 --- /dev/null +++ b/net-guardia/src/interface/communication/message.rs @@ -0,0 +1,3 @@ +pub trait Message: Send + 'static { + type Response: Send + 'static; +} diff --git a/net-guardia/src/interface/communication/mod.rs b/net-guardia/src/interface/communication/mod.rs new file mode 100644 index 0000000..b49b1a5 --- /dev/null +++ b/net-guardia/src/interface/communication/mod.rs @@ -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; diff --git a/net-guardia/src/interface/communication/query.rs b/net-guardia/src/interface/communication/query.rs new file mode 100644 index 0000000..5a5c6de --- /dev/null +++ b/net-guardia/src/interface/communication/query.rs @@ -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, Error>> + Send + 'static>>; +pub type QueryHandlerFn = Box) -> QueryFuture + Send + Sync>; + +pub trait Query: Message {} + +#[async_trait] +pub trait QueryHandler { + async fn handle_query(&self, query: Q) -> Result; +} diff --git a/net-guardia/src/interface/communication/query_types.rs b/net-guardia/src/interface/communication/query_types.rs new file mode 100644 index 0000000..56278ec --- /dev/null +++ b/net-guardia/src/interface/communication/query_types.rs @@ -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; +} +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; +} +impl Query for GetDnsDomainsQuery {} + +// ── Geo Queries ────────────────────────────────────────────────────── + +pub struct GetGeoBlockedCountriesQuery; + +impl Message for GetGeoBlockedCountriesQuery { + type Response = Vec; +} +impl Query for GetGeoBlockedCountriesQuery {} diff --git a/net-guardia/src/interface/mod.rs b/net-guardia/src/interface/mod.rs new file mode 100644 index 0000000..27daa2a --- /dev/null +++ b/net-guardia/src/interface/mod.rs @@ -0,0 +1,2 @@ +pub mod communication; +pub mod port; diff --git a/net-guardia/src/interface/port/auth.rs b/net-guardia/src/interface/port/auth.rs new file mode 100644 index 0000000..341362b --- /dev/null +++ b/net-guardia/src/interface/port/auth.rs @@ -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; + fn validate_token(&self, token: &str) -> Result; + fn hash_password(&self, password: &str) -> Result; + fn verify_password(&self, password: &str, hash: &str) -> Result; +} diff --git a/net-guardia/src/interface/port/health.rs b/net-guardia/src/interface/port/health.rs new file mode 100644 index 0000000..e44a8af --- /dev/null +++ b/net-guardia/src/interface/port/health.rs @@ -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; +} diff --git a/net-guardia/src/interface/port/mod.rs b/net-guardia/src/interface/port/mod.rs new file mode 100644 index 0000000..12766f8 --- /dev/null +++ b/net-guardia/src/interface/port/mod.rs @@ -0,0 +1,4 @@ +pub mod repository; +pub mod auth; +pub mod notification; +pub mod health; diff --git a/net-guardia/src/interface/port/notification.rs b/net-guardia/src/interface/port/notification.rs new file mode 100644 index 0000000..fd34d35 --- /dev/null +++ b/net-guardia/src/interface/port/notification.rs @@ -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>; +} diff --git a/net-guardia/src/interface/port/repository.rs b/net-guardia/src/interface/port/repository.rs new file mode 100644 index 0000000..8aca0da --- /dev/null +++ b/net-guardia/src/interface/port/repository.rs @@ -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, Error>; + + // --- Rate Limit --- + fn set_rate_limit(&self, key: &str, value: u64) -> Result<(), Error>; + fn load_rate_limit_config(&self) -> Result, 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, 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, Error>; + + // --- Settings --- + fn get_setting(&self, key: &str) -> Result, Error>; + fn set_setting(&self, key: &str, value: &str) -> Result<(), Error>; + + // --- Users --- + fn find_user(&self, username: &str) -> Result, 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; + + // --- Login Rate Limiting --- + fn record_login_failure(&self, username: &str) -> Result<(u32, Option), Error>; + fn check_login_locked(&self, username: &str) -> Result, Error>; + fn clear_login_failures(&self, username: &str) -> Result<(), Error>; +} diff --git a/net-guardia/src/main.rs b/net-guardia/src/main.rs index 3f9d173..82440f0 100644 --- a/net-guardia/src/main.rs +++ b/net-guardia/src/main.rs @@ -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; diff --git a/net-guardia/src/model/error/misc.rs b/net-guardia/src/model/error/misc.rs index 6947201..b7ea19d 100644 --- a/net-guardia/src/model/error/misc.rs +++ b/net-guardia/src/model/error/misc.rs @@ -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, } } diff --git a/net-guardia/src/model/flow_stats.rs b/net-guardia/src/model/flow_stats.rs index a9f75be..a82956f 100644 --- a/net-guardia/src/model/flow_stats.rs +++ b/net-guardia/src/model/flow_stats.rs @@ -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 { diff --git a/net-guardia/src/web/mod.rs b/net-guardia/src/web/mod.rs deleted file mode 100644 index a3766ff..0000000 --- a/net-guardia/src/web/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod api; -pub mod websocket;