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(concurrency): migrate ten lock sites to lock-free primitives

Browse Source 
Move hot and snapshot-style state off Mutex/RwLock onto atomic
primitives, ArcSwap, DashSet, moka cache, or capacity-1 mpsc:

- DropCounters: Mutex<struct of u64> -> 9x AtomicU64 with snapshot()
- Telegram rate limiter: Mutex<(u32, Instant)> -> packed AtomicU64 + CAS
- DNS blacklist: RwLock<HashSet<DnsName>> -> DashSet<DnsName>
- geo_block.blocked_countries: RwLock<HashSet> -> ArcSwap with rcu()
- SOAR playbooks + admin_whitelist caches: RwLock -> ArcSwap
- EbpfHealth: Arc<parking_lot::RwLock<EbpfHealth>> -> Arc<ArcSwap>
- SuricataHealth: Arc<RwLock> -> Arc<ArcSwap>
- GeoIP cache: tokio::sync::RwLock<LruCache> -> moka::sync::Cache
- ShutdownHandle: Mutex<Option<oneshot::Sender>> -> mpsc(1)+try_send

Setting rename driven by removing a hardcoded 60s window in the
Telegram limiter (no backward compat — pre-release):
- telegram_max_messages_per_minute -> telegram_rate_limit_max_messages
- new key telegram_rate_limit_window_secs (default 60)

SystemLog::TelegramLocalRateLimitDropped fields renamed accordingly.

Add moka 0.12 (sync feature) dependency.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
DaLaw2 2026-04-18 21:37:52 +08:00
parent 36cdf1a3d9
commit 37ba734048
20 changed files with 260 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
Cargo.lock generated
View File

@ -2448,6 +2448,23 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "moka"
version = "0.12.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046"
dependencies = [
"crossbeam-channel",
"crossbeam-epoch",
"crossbeam-utils",
"equivalent",
"parking_lot",
"portable-atomic",
"smallvec",
"tagptr",
"uuid",
]
[[package]]
name = "ndarray"
version = "0.16.1"
@ -2498,6 +2515,7 @@ dependencies = [
"macros",
"maxminddb",
"mime_guess",
"moka",
"network-types",
"notify",
"parking_lot",
@ -3803,6 +3821,12 @@ dependencies = [
"windows",
]
[[package]]
name = "tagptr"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
[[package]]
name = "tar"
version = "0.4.44"

1
net-guardia/Cargo.toml
View File

@ -58,6 +58,7 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
async-trait = "0.1"
dashmap = "6"
arc-swap = "1"
moka = { version = "0.12", features = ["sync"] }
notify = { version = "7", default-features = false, features = ["macos_kqueue"] }
# Utilities

23
net-guardia/src/adapter/ebpf/dns_filter.rs
View File

@ -1,8 +1,7 @@
use core::str;
use std::collections::HashSet;
use common::model::dns_name::DnsName;
use parking_lot::RwLock;
use dashmap::DashSet;
use crate::interface::port::dns_filter_api::DnsFilterPort;
use crate::interface::port::dns_query_filter::DnsQueryFilter;
@ -10,30 +9,33 @@ use crate::model::error::Error;
use crate::model::error::misc::MiscError;
pub struct DnsFilter {
blacklist: RwLock<HashSet<DnsName>>,
blacklist: DashSet<DnsName>,
}
impl DnsFilter {
pub fn new() -> Self {
Self {
blacklist: RwLock::new(HashSet::new()),
blacklist: DashSet::new(),
}
}
pub fn add_domain(&self, domain: &str) -> Result<(), Error> {
let name = domain_to_wire_format(domain)?;
self.blacklist.write().insert(name);
self.blacklist.insert(name);
Ok(())
}
pub fn remove_domain(&self, domain: &str) -> Result<(), Error> {
let name = domain_to_wire_format(domain)?;
self.blacklist.write().remove(&name);
self.blacklist.remove(&name);
Ok(())
}
pub fn list_domains(&self) -> Vec<String> {
self.blacklist.read().iter().filter_map(wire_format_to_domain).collect()
self.blacklist
.iter()
.filter_map(|entry| wire_format_to_domain(&entry))
.collect()
}
/// Fast-path helper combining `parse_query_name` + `is_blacklisted` — used
@ -47,12 +49,11 @@ impl DnsFilter {
/// Check if a DNS query name (in wire format) or any of its parent domains is blacklisted.
pub fn is_blacklisted(&self, name: &DnsName, name_len: usize) -> bool {
let bl = self.blacklist.read();
if bl.is_empty() {
if self.blacklist.is_empty() {
return false;
}
// Check exact match
if bl.contains(name) {
if self.blacklist.contains(name) {
return true;
}
// Check parent domains
@ -76,7 +77,7 @@ impl DnsFilter {
let mut parent = DnsName::zeroed();
let remaining = name_len - offset;
parent.data[..remaining.min(128)].copy_from_slice(&name.data[offset..offset + remaining.min(128)]);
if bl.contains(&parent) {
if self.blacklist.contains(&parent) {
return true;
}
}

39
net-guardia/src/adapter/ebpf/drop_monitor.rs
View File

@ -1,6 +1,7 @@
use std::mem;
use std::net::Ipv6Addr;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use aya::maps::{MapData, RingBuf};
@ -9,14 +10,13 @@ use tokio::time::interval;
use common::define::drop_reason::*;
use common::model::drop_event::DropEvent as RawDropEvent;
use parking_lot::Mutex;
use crate::model::config::constants::DROP_CHANNEL_CAPACITY;
use crate::model::monitoring::drop_event::{DropCounters, DropEventMessage};
use crate::model::monitoring::drop_event::{DropCounters, DropCountersAtomic, DropEventMessage};
pub struct DropMonitor {
broadcast_tx: broadcast::Sender<DropEventMessage>,
counters: Mutex<DropCounters>,
counters: DropCountersAtomic,
}
impl DropMonitor {
@ -24,7 +24,7 @@ impl DropMonitor {
let (tx, _) = broadcast::channel(DROP_CHANNEL_CAPACITY);
Self {
broadcast_tx: tx,
counters: Mutex::new(DropCounters::default()),
counters: DropCountersAtomic::default(),
}
}
@ -33,25 +33,24 @@ impl DropMonitor {
}
pub fn get_counters(&self) -> DropCounters {
self.counters.lock().clone()
self.counters.snapshot()
}
fn process_event(&self, raw: &RawDropEvent) {
// Update counters
{
let mut c = self.counters.lock();
c.total += 1;
match raw.reason {
DROP_REASON_ACL_BLACKLIST => c.acl_blacklist += 1,
DROP_REASON_RATE_LIMIT_PKT => c.rate_limit_pkt += 1,
DROP_REASON_RATE_LIMIT_SYN => c.rate_limit_syn += 1,
DROP_REASON_RATE_LIMIT_UDP => c.rate_limit_udp += 1,
DROP_REASON_RATE_LIMIT_DNS => c.rate_limit_dns += 1,
DROP_REASON_PROTOCOL_FILTER => c.protocol_filter += 1,
DROP_REASON_DNS_BLACKLIST => c.dns_blacklist += 1,
DROP_REASON_GEO_BLOCK => c.geo_block += 1,
_ => {}
}
self.counters.total.fetch_add(1, Ordering::Relaxed);
let bucket = match raw.reason {
DROP_REASON_ACL_BLACKLIST => Some(&self.counters.acl_blacklist),
DROP_REASON_RATE_LIMIT_PKT => Some(&self.counters.rate_limit_pkt),
DROP_REASON_RATE_LIMIT_SYN => Some(&self.counters.rate_limit_syn),
DROP_REASON_RATE_LIMIT_UDP => Some(&self.counters.rate_limit_udp),
DROP_REASON_RATE_LIMIT_DNS => Some(&self.counters.rate_limit_dns),
DROP_REASON_PROTOCOL_FILTER => Some(&self.counters.protocol_filter),
DROP_REASON_DNS_BLACKLIST => Some(&self.counters.dns_blacklist),
DROP_REASON_GEO_BLOCK => Some(&self.counters.geo_block),
_ => None,
};
if let Some(counter) = bucket {
counter.fetch_add(1, Ordering::Relaxed);
}
let reason_str = reason_to_str(raw.reason);

31
net-guardia/src/adapter/ebpf/geo_block.rs
View File

@ -1,6 +1,7 @@
use std::collections::{HashMap as StdHashMap, HashSet};
use std::sync::Arc;
use arc_swap::ArcSwap;
use aya::Ebpf;
use aya::maps::MapData;
use aya::maps::lpm_trie::{Key, LpmTrie};
@ -23,7 +24,7 @@ struct GeoIndex {
pub struct GeoBlock {
geo_block_v4: RwLock<Option<LpmTrie<MapData, u32, u8>>>,
geo_block_v6: RwLock<Option<LpmTrie<MapData, u128, u8>>>,
blocked_countries: RwLock<HashSet<String>>,
blocked_countries: ArcSwap<HashSet<String>>,
index: Arc<GeoIndex>,
}
@ -43,7 +44,7 @@ impl GeoBlock {
Ok(Self {
geo_block_v4: RwLock::new(Some(v4_trie)),
geo_block_v6: RwLock::new(Some(v6_trie)),
blocked_countries: RwLock::new(HashSet::new()),
blocked_countries: ArcSwap::from_pointee(HashSet::new()),
index: Arc::new(index),
})
}
@ -63,7 +64,7 @@ impl GeoBlock {
Self {
geo_block_v4: RwLock::new(None),
geo_block_v6: RwLock::new(None),
blocked_countries: RwLock::new(HashSet::new()),
blocked_countries: ArcSwap::from_pointee(HashSet::new()),
index: Arc::new(index),
}
}
@ -116,42 +117,44 @@ impl GeoBlock {
/// Block multiple countries at once, rebuilding tries only once.
pub fn block_countries(&self, country_codes: &[String]) -> Result<u64, Error> {
{
let mut countries = self.blocked_countries.write();
self.blocked_countries.rcu(|cur| {
let mut next: HashSet<String> = (**cur).clone();
for code in country_codes {
let upper = code.trim().to_uppercase();
if upper.len() == 2 && upper.chars().all(|c| c.is_ascii_alphabetic()) {
countries.insert(upper);
next.insert(upper);
}
}
}
next
});
self.rebuild_tries()
}
/// Unblock multiple countries at once, rebuilding tries only once.
pub fn unblock_countries(&self, country_codes: &[String]) -> Result<u64, Error> {
{
let mut countries = self.blocked_countries.write();
self.blocked_countries.rcu(|cur| {
let mut next: HashSet<String> = (**cur).clone();
for code in country_codes {
countries.remove(&code.trim().to_uppercase());
next.remove(&code.trim().to_uppercase());
}
}
next
});
self.rebuild_tries()
}
pub fn get_blocked_countries(&self) -> Vec<String> {
self.blocked_countries.read().iter().cloned().collect()
self.blocked_countries.load().iter().cloned().collect()
}
/// Rebuild LPM tries from pre-indexed data. Fast — no DB scan.
fn rebuild_tries(&self) -> Result<u64, Error> {
let countries = self.blocked_countries.read().clone();
let countries = self.blocked_countries.load_full();
// Collect entries from index (no DB scan)
let mut v4_entries: Vec<(Key<u32>, u8)> = Vec::new();
let mut v6_entries: Vec<(Key<u128>, u8)> = Vec::new();
for code in &countries {
for code in countries.iter() {
if let Some(prefixes) = self.index.v4.get(code) {
for &(ip_be, prefix_len) in prefixes {
v4_entries.push((Key::new(prefix_len, ip_be), 1u8));

4
net-guardia/src/adapter/http/health.rs
View File

@ -22,11 +22,11 @@ async fn get_health_status(health: web::Data<SystemHealth>) -> impl Responder {
}
async fn get_ebpf_health(health: web::Data<SystemHealth>) -> impl Responder {
let ebpf = health.ebpf_health().read().clone();
let ebpf = (**health.ebpf_health().load()).clone();
HttpResponse::Ok().json(ebpf)
}
async fn get_suricata_health(manager: web::Data<SuricataManager>) -> impl Responder {
let state = manager.health().read().clone();
let state = (**manager.health().load()).clone();
HttpResponse::Ok().json(state)
}

88
net-guardia/src/adapter/telegram/mod.rs
View File

@ -1,9 +1,9 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use async_trait::async_trait;
use macros::log;
use parking_lot::Mutex;
use reqwest::Client;
use tokio::time::sleep;
@ -22,8 +22,10 @@ pub struct TelegramAdapter {
notif: Arc<dyn SettingRepo>,
repo: Arc<dyn AppRepo>,
secrets: Option<Arc<dyn SecretStorePort>>,
/// Rate limiter: (count, window_start)
rate_state: Mutex<(u32, Instant)>,
/// Packed rate-limit state: high 32 bits = window-start unix seconds,
/// low 32 bits = count consumed in this window. Updated via CAS so the
/// hot path stays lock-free.
rate_state: AtomicU64,
}
impl TelegramAdapter {
@ -42,7 +44,7 @@ impl TelegramAdapter {
notif,
repo,
secrets,
rate_state: Mutex::new((0, Instant::now())),
rate_state: AtomicU64::new(0),
})
}
@ -73,30 +75,60 @@ impl TelegramAdapter {
}
}
/// Check rate limit. Returns true if send is allowed.
fn check_rate_limit(&self) -> bool {
let mut state = self.rate_state.lock();
let (count, window_start) = &mut *state;
// Reset window if >60s has passed
if window_start.elapsed() > Duration::from_secs(60) {
*count = 0;
*window_start = Instant::now();
}
let max_per_min: u32 = self
.repo
.get_setting("telegram_max_messages_per_minute")
/// Read the configured per-window message cap from settings.
fn rate_limit_max_messages(&self) -> u32 {
self.repo
.get_setting("telegram_rate_limit_max_messages")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(20);
if *count >= max_per_min {
.unwrap_or(20)
}
/// Read the configured window length (seconds) from settings.
fn rate_limit_window_secs(&self) -> u32 {
self.repo
.get_setting("telegram_rate_limit_window_secs")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(60)
}
/// Check rate limit. Returns true if send is allowed.
fn check_rate_limit(&self) -> bool {
let max_messages = self.rate_limit_max_messages();
if max_messages == 0 {
return false;
}
let window_secs = self.rate_limit_window_secs().max(1);
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as u32)
.unwrap_or(0);
*count += 1;
true
loop {
let cur = self.rate_state.load(Ordering::Acquire);
let count = cur as u32;
let window = (cur >> 32) as u32;
let (next_count, next_window) = if now.saturating_sub(window) >= window_secs {
(1u32, now)
} else if count >= max_messages {
return false;
} else {
(count + 1, window)
};
let new = ((next_window as u64) << 32) | next_count as u64;
if self
.rate_state
.compare_exchange_weak(cur, new, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
return true;
}
}
}
/// Send a message via Telegram Bot API with retry on 429.
@ -200,15 +232,9 @@ impl AlertNotifier for TelegramAdapter {
};
if !self.check_rate_limit() {
let max_per_min: u32 = self
.repo
.get_setting("telegram_max_messages_per_minute")
.ok()
.flatten()
.and_then(|v| v.parse().ok())
.unwrap_or(20);
log!(SystemLog::TelegramLocalRateLimitDropped(
max_per_min,
self.rate_limit_max_messages(),
self.rate_limit_window_secs(),
payload.source_ip.clone(),
));
return Ok(());

8
net-guardia/src/core/config_service.rs
View File

@ -56,7 +56,10 @@ const SETTINGS_MAP: &[(&str, &[&str])] = &[
("misc", &["geoip_db_name"]),
("soar", &["soar_max_auto_block_cap", "soar_max_ttl_secs"]),
("ml", &["ml_drift_window_secs"]),
("telegram", &["telegram_max_messages_per_minute"]),
(
"telegram",
&["telegram_rate_limit_max_messages", "telegram_rate_limit_window_secs"],
),
("dns", &["dns_max_domains_per_request"]),
("smtp", &["smtp_host", "smtp_port", "smtp_username", "smtp_recipient"]),
];
@ -129,7 +132,8 @@ impl ConfigService {
"ml_drift_window_secs": get("ml_drift_window_secs"),
},
"telegram": {
"telegram_max_messages_per_minute": get("telegram_max_messages_per_minute"),
"telegram_rate_limit_max_messages": get("telegram_rate_limit_max_messages"),
"telegram_rate_limit_window_secs": get("telegram_rate_limit_window_secs"),
},
"dns": {
"dns_max_domains_per_request": get("dns_max_domains_per_request"),

4
net-guardia/src/core/soar/actions.rs
View File

@ -41,7 +41,7 @@ impl SoarEngine {
}
// Check admin whitelist
if self.admin_whitelist.read().contains(&event.source_ip) {
if self.admin_whitelist.load().contains(&event.source_ip) {
log!(SoarLog::WhitelistSkipped(
event.source_ip.clone(),
playbook.name.clone()
@ -465,7 +465,7 @@ impl SoarEngine {
/// Only fires when source_ip is present.
pub(super) async fn execute_fallback(&self, event: &ThreatDetectedEvent) -> Result<(), Error> {
// Check admin whitelist — never block admin IPs even in fallback
if self.admin_whitelist.read().contains(&event.source_ip) {
if self.admin_whitelist.load().contains(&event.source_ip) {
log!(SoarLog::WhitelistSkipped(
event.source_ip.clone(),
"fallback".to_string()

29
net-guardia/src/core/soar/engine.rs
View File

@ -3,10 +3,10 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU32, Ordering};
use std::time::Instant;
use arc_swap::ArcSwap;
use chrono::{NaiveDateTime, Utc};
use dashmap::DashMap;
use macros::log;
use parking_lot::RwLock;
use serde_json::Value;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{Mutex as TokioMutex, broadcast};
@ -45,9 +45,9 @@ pub struct SoarEngine {
pub(super) db: Arc<dyn AppRepo>,
pub(super) access_control: Arc<dyn AccessControlPort>,
/// In-memory cache of playbooks (loaded at startup, refreshed on change).
pub(super) playbooks: RwLock<Vec<Playbook>>,
pub(super) playbooks: ArcSwap<Vec<Playbook>>,
/// In-memory cache of admin whitelist IPs.
pub(super) admin_whitelist: RwLock<HashSet<String>>,
pub(super) admin_whitelist: ArcSwap<HashSet<String>>,
/// Cooldown tracker: maps (playbook_id, source_ip) → last execution time.
pub(super) cooldowns: DashMap<CooldownKey, Instant>,
/// Frequency tracker for frequency-based conditions.
@ -83,8 +83,8 @@ impl SoarEngine {
let engine = Self {
db,
access_control,
playbooks: RwLock::new(Vec::new()),
admin_whitelist: RwLock::new(HashSet::new()),
playbooks: ArcSwap::from_pointee(Vec::new()),
admin_whitelist: ArcSwap::from_pointee(HashSet::new()),
cooldowns: DashMap::new(),
frequency_tracker: FrequencyTracker::new(),
active_block_count: AtomicU32::new(0),
@ -201,21 +201,20 @@ impl SoarEngine {
}
}
*self.playbooks.write() = playbooks;
let playbook_count = playbooks.len();
self.playbooks.store(Arc::new(playbooks));
// Load admin whitelist
let whitelist = self.db.load_admin_whitelist()?;
*self.admin_whitelist.write() = whitelist.into_iter().collect();
let whitelist: HashSet<String> = whitelist.into_iter().collect();
let whitelist_count = whitelist.len();
self.admin_whitelist.store(Arc::new(whitelist));
// Initialize block counter from DB
let count = self.db.count_active_soar_blocks()?;
self.active_block_count.store(count, Ordering::SeqCst);
log!(SoarLog::CacheLoaded(
self.playbooks.read().len(),
self.admin_whitelist.read().len(),
count,
));
log!(SoarLog::CacheLoaded(playbook_count, whitelist_count, count));
Ok(())
}
@ -759,18 +758,18 @@ mod tests {
let engine = SoarEngine::new(db.clone(), mock, None, None, None, cache, None).expect("Failed to create engine");
// Should have loaded default playbooks
let count = engine.playbooks.read().len();
let count = engine.playbooks.load().len();
assert!(count > 0, "Should have loaded default playbooks");
// Add a new playbook directly to DB
db.insert_playbook("test_pb", "port_scan", None, None, None, 60).ok();
// Cache should not have it yet
assert_eq!(engine.playbooks.read().len(), count);
assert_eq!(engine.playbooks.load().len(), count);
// After reload, should have one more
engine.reload_cache().expect("reload should succeed");
assert_eq!(engine.playbooks.read().len(), count + 1);
assert_eq!(engine.playbooks.load().len(), count + 1);
}
fn test_event(confidence: f32, country: Option<&str>, ip: &str, repeat: bool) -> ThreatDetectedEvent {

26
net-guardia/src/core/soar/matcher.rs
View File

@ -33,8 +33,8 @@ const DEFAULT_COOLDOWN_EXPIRY_SECS: u64 = 3600;
impl SoarEngine {
/// Find playbooks matching the event via trigger_event + multi-condition AND logic.
pub(super) fn find_matching_playbooks(&self, event: &ThreatDetectedEvent) -> Vec<Playbook> {
let playbooks = self.playbooks.read();
playbooks
self.playbooks
.load()
.iter()
.filter(|pb| pb.enabled && pb.trigger_event == event.attack_type)
.filter(|pb| self.evaluate_conditions(pb, event))
@ -257,14 +257,13 @@ impl SoarEngine {
/// Remove expired cooldown entries to prevent unbounded growth.
/// Called by TTL scheduler every 60 seconds.
pub fn cleanup_expired_cooldowns(&self) {
let max_cooldown_secs = {
let playbooks = self.playbooks.read();
playbooks
.iter()
.map(|p| p.cooldown_secs as u64)
.max()
.unwrap_or(DEFAULT_COOLDOWN_EXPIRY_SECS)
};
let max_cooldown_secs = self
.playbooks
.load()
.iter()
.map(|p| p.cooldown_secs as u64)
.max()
.unwrap_or(DEFAULT_COOLDOWN_EXPIRY_SECS);
let expiry = Duration::from_secs(max_cooldown_secs.saturating_mul(2).max(DEFAULT_COOLDOWN_EXPIRY_SECS));
let before = self.cooldowns.len();
self.cooldowns.retain(|_, instant| instant.elapsed() < expiry);
@ -310,8 +309,11 @@ impl SoarEngine {
/// admin so they don't assume a `would_fire=true` playbook will
/// definitely fire on the next matching real event.
pub fn dry_run(&self, event: &ThreatDetectedEvent) -> Vec<DryRunMatch> {
let playbooks = self.playbooks.read();
playbooks.iter().map(|pb| simulate_playbook(pb, event)).collect()
self.playbooks
.load()
.iter()
.map(|pb| simulate_playbook(pb, event))
.collect()
}
/// Check if an IP address is private/loopback/link-local (SSRF protection).

3
net-guardia/src/infrastructure/app_config.rs
View File

@ -69,7 +69,8 @@ impl AppConfig {
// ML
("ml_drift_window_secs", "3600".into()),
// Telegram
("telegram_max_messages_per_minute", "20".into()),
("telegram_rate_limit_max_messages", "20".into()),
("telegram_rate_limit_window_secs", "60".into()),
// Directories
("report_dir", "/var/lib/netguardia/reports".into()),
("log_dir", "logs".into()),

3
net-guardia/src/infrastructure/app_services.rs
View File

@ -2,6 +2,7 @@ use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use arc_swap::ArcSwap;
use crossbeam::queue::SegQueue;
use macros::log;
use tokio::sync::oneshot;
@ -50,7 +51,7 @@ impl AppServices {
inference_config: Arc<MLInferenceConfig>,
ml_manifest: Option<ModelManifest>,
drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
ebpf_health: Arc<parking_lot::RwLock<EbpfHealth>>,
ebpf_health: Arc<ArcSwap<EbpfHealth>>,
comm: Arc<CommunicationManager>,
) -> Result<Self, Error> {
let health = SystemHealth::new(app_config.clone(), ebpf_health)?;

23
net-guardia/src/infrastructure/geoip.rs
View File

@ -2,10 +2,8 @@ use std::net::IpAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use lru::LruCache;
use maxminddb::{MaxMindDbError, Reader, geoip2};
use std::num::NonZeroUsize;
use tokio::sync::RwLock;
use moka::sync::Cache;
use tokio::task;
use crate::model::monitoring::geolocation::GeoLocation;
@ -13,7 +11,7 @@ use crate::utils::ip_address;
pub struct GeoIpService {
reader: Arc<Reader<Vec<u8>>>,
cache: Arc<RwLock<LruCache<IpAddr, Option<GeoLocation>>>>,
cache: Cache<IpAddr, Option<GeoLocation>>,
}
impl GeoIpService {
@ -24,12 +22,11 @@ impl GeoIpService {
pub fn with_cache_size<P: AsRef<Path>>(db_path: P, cache_size: usize) -> Result<Self, MaxMindDbError> {
let reader = Reader::open_readfile(db_path)?;
// SAFETY: 10000 is a non-zero literal, NonZeroUsize::new is infallible.
let cache_capacity = NonZeroUsize::new(cache_size).unwrap_or_else(|| NonZeroUsize::new(10000).unwrap());
let capacity = if cache_size == 0 { 10_000 } else { cache_size } as u64;
Ok(Self {
reader: Arc::new(reader),
cache: Arc::new(RwLock::new(LruCache::new(cache_capacity))),
cache: Cache::new(capacity),
})
}
@ -45,11 +42,8 @@ impl GeoIpService {
}));
}
{
let cache = self.cache.read().await;
if let Some(cached) = cache.peek(&ip) {
return Ok(cached.clone());
}
if let Some(cached) = self.cache.get(&ip) {
return Ok(cached);
}
let reader = self.reader.clone();
@ -60,10 +54,7 @@ impl GeoIpService {
offset: None,
})??;
{
let mut cache = self.cache.write().await;
cache.put(ip, result.clone());
}
self.cache.insert(ip, result.clone());
Ok(result)
}

11
net-guardia/src/infrastructure/health.rs
View File

@ -2,6 +2,7 @@ use std::env::consts;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use arc_swap::ArcSwap;
use macros::log;
use sysinfo::{Components, Networks, System};
use tokio::sync::{RwLock, broadcast, oneshot};
@ -22,11 +23,11 @@ pub struct SystemHealth {
broadcast_tx: broadcast::Sender<SystemHealthMetrics>,
ingress_interface: String,
egress_interface: String,
ebpf_health: Arc<parking_lot::RwLock<EbpfHealth>>,
ebpf_health: Arc<ArcSwap<EbpfHealth>>,
}
impl SystemHealth {
pub fn new(config: Arc<AppConfig>, ebpf_health: Arc<parking_lot::RwLock<EbpfHealth>>) -> Result<Self, Error> {
pub fn new(config: Arc<AppConfig>, ebpf_health: Arc<ArcSwap<EbpfHealth>>) -> Result<Self, Error> {
let (broadcast_tx, _) = broadcast::channel(100);
let health = SystemHealth {
@ -74,7 +75,7 @@ impl SystemHealth {
let networks = self.networks.read().await;
let components = self.components.read().await;
let ebpf = self.ebpf_health.read().clone();
let ebpf = (**self.ebpf_health.load()).clone();
let metrics = Self::collect_metrics(
&system,
&networks,
@ -243,7 +244,7 @@ impl SystemHealth {
let networks = self.networks.read().await;
let components = self.components.read().await;
let ebpf = self.ebpf_health.read().clone();
let ebpf = (**self.ebpf_health.load()).clone();
Self::collect_metrics(
&system,
&networks,
@ -257,7 +258,7 @@ impl SystemHealth {
/// Returns a handle to the shared eBPF health state. Consumers (HTTP
/// handlers, setup wizard, frontend) can read the current eBPF state
/// without going through the full metrics broadcast.
pub fn ebpf_health(&self) -> &Arc<parking_lot::RwLock<EbpfHealth>> {
pub fn ebpf_health(&self) -> &Arc<ArcSwap<EbpfHealth>> {
&self.ebpf_health
}

9
net-guardia/src/infrastructure/service_factory.rs
View File

@ -5,7 +5,8 @@ use std::sync::Arc;
use std::sync::atomic::AtomicU8;
use std::time::Duration;
use parking_lot::{Mutex, RwLock};
use arc_swap::ArcSwap;
use parking_lot::Mutex;
use aya::Ebpf;
use aya::maps::{Array, MapData, ProgramArray};
@ -91,7 +92,7 @@ pub struct AppState {
/// or to `Unavailable { stage, category, reason }` when any stage fails.
/// Read by `SystemHealth` for the metrics broadcast and by HTTP handlers
/// that render runtime status to the frontend.
pub ebpf_health: Arc<RwLock<EbpfHealth>>,
pub ebpf_health: Arc<ArcSwap<EbpfHealth>>,
pub suricata_manager: Arc<SuricataManager>,
}
@ -138,7 +139,7 @@ impl ServiceFactory {
// Shared eBPF health handle. Initialized Healthy; downgraded to
// Unavailable with a classified reason if any stage below fails.
let ebpf_health = Arc::new(RwLock::new(EbpfHealth::Healthy));
let ebpf_health = Arc::new(ArcSwap::from_pointee(EbpfHealth::Healthy));
// Attempt full eBPF bring-up. On any failure we classify the error,
// write it into `ebpf_health`, and fall back to an `EbpfServices`
@ -152,7 +153,7 @@ impl ServiceFactory {
Err((stage, err)) => {
let health = ebpf_preflight::classify(stage, &err, None);
log!(SystemLog::EbpfBringupFailed(format!("{:?}", health)));
*ebpf_health.write() = health;
ebpf_health.store(Arc::new(health));
(
None,
None,

24
net-guardia/src/infrastructure/suricata_manager.rs
View File

@ -15,8 +15,8 @@ use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use arc_swap::ArcSwap;
use macros::log;
use parking_lot::RwLock;
use tokio::process::{Child, Command};
use tokio::sync::oneshot;
use tokio::time::{sleep, timeout};
@ -29,7 +29,7 @@ use crate::model::system::suricata::SuricataHealth;
pub struct SuricataManager {
config: Arc<AppConfig>,
health: Arc<RwLock<SuricataHealth>>,
health: Arc<ArcSwap<SuricataHealth>>,
}
impl SuricataManager {
@ -43,12 +43,12 @@ impl SuricataManager {
};
Arc::new(Self {
config,
health: Arc::new(RwLock::new(initial)),
health: Arc::new(ArcSwap::from_pointee(initial)),
})
}
/// Shared handle for HTTP handlers and the health broadcast.
pub fn health(&self) -> Arc<RwLock<SuricataHealth>> {
pub fn health(&self) -> Arc<ArcSwap<SuricataHealth>> {
self.health.clone()
}
@ -73,20 +73,22 @@ impl SuricataManager {
loop {
// Pre-flight: validate binary + config exist before spawning.
if let Err(e) = Self::preflight(&self.config) {
*self.health.write() = SuricataHealth::Stopped { reason: e.to_string() };
self.health
.store(Arc::new(SuricataHealth::Stopped { reason: e.to_string() }));
return;
}
let mut child = match self.spawn_child() {
Ok(c) => c,
Err(e) => {
*self.health.write() = SuricataHealth::Stopped { reason: e.to_string() };
self.health
.store(Arc::new(SuricataHealth::Stopped { reason: e.to_string() }));
return;
}
};
let pid = child.id().unwrap_or(0);
*self.health.write() = SuricataHealth::Running { pid };
self.health.store(Arc::new(SuricataHealth::Running { pid }));
log!(SuricataLog::Started(pid));
tokio::select! {
@ -98,21 +100,21 @@ impl SuricataManager {
if self.config.suricata.auto_restart_on_crash {
let backoff = self.config.suricata.restart_backoff_secs;
log!(SuricataLog::CrashedRestartPending(reason.clone(), backoff));
*self.health.write() = SuricataHealth::Stopped { reason };
self.health.store(Arc::new(SuricataHealth::Stopped { reason }));
sleep(Duration::from_secs(backoff)).await;
continue;
} else {
log!(SuricataLog::Stopped(reason.clone()));
*self.health.write() = SuricataHealth::Stopped { reason };
self.health.store(Arc::new(SuricataHealth::Stopped { reason }));
return;
}
}
_ = &mut shutdown_rx => {
log!(SuricataLog::ShutdownRequested);
Self::graceful_stop(&mut child).await;
*self.health.write() = SuricataHealth::Stopped {
self.health.store(Arc::new(SuricataHealth::Stopped {
reason: "shutdown".to_string(),
};
}));
return;
}
}

39
net-guardia/src/infrastructure/system.rs
View File

@ -3,10 +3,10 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use arc_swap::ArcSwap;
use aya::Ebpf;
use aya::maps::{MapData, ProgramArray};
use macros::log;
use parking_lot::{Mutex, RwLock};
use sd_notify::NotifyState;
use tokio::signal::ctrl_c;
use tokio::sync::broadcast::{Receiver, error::RecvError};
@ -65,26 +65,23 @@ pub enum ShutdownMode {
Restart,
}
/// Handle for triggering shutdown from HTTP endpoints.
/// Uses a parking_lot::Mutex<Option<oneshot::Sender>> so it can be shared as app_data.
/// Handle for triggering shutdown from HTTP endpoints. Backed by a
/// capacity-1 mpsc so the first trigger atomically wins via `try_send`,
/// and subsequent calls receive `TrySendError::Full` — no lock, no
/// `Option::take`, no `Mutex`.
pub struct ShutdownHandle {
tx: Mutex<Option<oneshot::Sender<ShutdownMode>>>,
tx: mpsc::Sender<ShutdownMode>,
}
impl ShutdownHandle {
fn new(tx: oneshot::Sender<ShutdownMode>) -> Self {
Self {
tx: Mutex::new(Some(tx)),
}
fn new(tx: mpsc::Sender<ShutdownMode>) -> Self {
Self { tx }
}
/// Trigger shutdown. Returns false if already triggered.
/// Trigger shutdown. Returns false if already triggered or the
/// receiver has been dropped.
pub fn trigger(&self, mode: ShutdownMode) -> bool {
if let Some(tx) = self.tx.lock().take() {
tx.send(mode).is_ok()
} else {
false
}
self.tx.try_send(mode).is_ok()
}
}
@ -115,7 +112,7 @@ pub struct System {
pub drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
pub shutdown_handle: Option<Arc<ShutdownHandle>>,
_ingress_program_array: Option<ProgramArray<MapData>>,
pub ebpf_health: Arc<RwLock<EbpfHealth>>,
pub ebpf_health: Arc<ArcSwap<EbpfHealth>>,
pub suricata_manager: Arc<SuricataManager>,
suricata_shutdown: Option<oneshot::Sender<()>>,
}
@ -222,7 +219,7 @@ impl System {
let iface = self.app_config.network.ingress_ifname.as_str();
let health = ebpf_preflight::classify(EbpfFailStage::AfXdpBind, &e, Some(iface));
log!(SystemLog::EbpfBringupFailed(format!("{:?}", health)));
*self.ebpf_health.write() = health;
self.ebpf_health.store(Arc::new(health));
}
app_services.run().await?;
@ -314,8 +311,10 @@ impl System {
// SOAR engine started above (self.soar_engine.start succeeded)
readiness_state.soar_engine_running.store(true, Ordering::SeqCst);
// Create shutdown channel for API-triggered shutdown/restart
let (shutdown_tx, shutdown_rx) = oneshot::channel::<ShutdownMode>();
// Create shutdown channel for API-triggered shutdown/restart.
// Capacity 1 means only the first `try_send` lands a value; later
// ones report Full, which `ShutdownHandle::trigger` surfaces as `false`.
let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<ShutdownMode>(1);
let shutdown_handle = Arc::new(ShutdownHandle::new(shutdown_tx));
self.shutdown_handle = Some(shutdown_handle.clone());
@ -393,7 +392,7 @@ impl System {
_ = ctrl_c() => {
Ok(ShutdownMode::Shutdown)
}
mode = shutdown_rx => {
mode = shutdown_rx.recv() => {
Ok(mode.unwrap_or(ShutdownMode::Shutdown))
}
}
@ -535,7 +534,7 @@ impl System {
};
let health = ebpf_preflight::classify(EbpfFailStage::XdpAttach, err, Some(iface));
log!(SystemLog::EbpfBringupFailed(format!("{:?}", health)));
*self.ebpf_health.write() = health;
self.ebpf_health.store(Arc::new(health));
}
}

4
net-guardia/src/model/log/system.rs
View File

@ -111,8 +111,8 @@ loggable! {
#[error("Telegram not configured, skipping alert")]
TelegramNotConfiguredSkipped => tracing::Level::DEBUG,
#[error("Telegram rate limit reached ({max_per_min}/min), dropping alert for IP {source_ip}")]
TelegramLocalRateLimitDropped { max_per_min: u32, source_ip: String } => tracing::Level::WARN,
#[error("Telegram rate limit reached ({max_messages} per {window_secs}s), dropping alert for IP {source_ip}")]
TelegramLocalRateLimitDropped { max_messages: u32, window_secs: u32, source_ip: String } => tracing::Level::WARN,
#[error("Stats aggregator started (1h interval)")]
StatsAggregatorStarted => tracing::Level::INFO,

36
net-guardia/src/model/monitoring/drop_event.rs
View File

@ -1,3 +1,5 @@
use std::sync::atomic::{AtomicU64, Ordering};
use serde::Serialize;
#[derive(Debug, Clone, Serialize)]
@ -12,6 +14,8 @@ pub struct DropEventMessage {
pub ip_version: u8,
}
/// Wire snapshot of drop counts. Returned by `DropMonitor::snapshot` and
/// serialized to JSON for the HTTP stats endpoint.
#[derive(Default, Clone, Serialize)]
pub struct DropCounters {
pub acl_blacklist: u64,
@ -24,3 +28,35 @@ pub struct DropCounters {
pub geo_block: u64,
pub total: u64,
}
/// Lock-free atomic counters incremented on the drop ring-buffer consumer
/// path. `Relaxed` is sufficient — counters are independent and the
/// snapshot does not require a global consistent ordering across them.
#[derive(Default)]
pub struct DropCountersAtomic {
pub acl_blacklist: AtomicU64,
pub rate_limit_pkt: AtomicU64,
pub rate_limit_syn: AtomicU64,
pub rate_limit_udp: AtomicU64,
pub rate_limit_dns: AtomicU64,
pub protocol_filter: AtomicU64,
pub dns_blacklist: AtomicU64,
pub geo_block: AtomicU64,
pub total: AtomicU64,
}
impl DropCountersAtomic {
pub fn snapshot(&self) -> DropCounters {
DropCounters {
acl_blacklist: self.acl_blacklist.load(Ordering::Relaxed),
rate_limit_pkt: self.rate_limit_pkt.load(Ordering::Relaxed),
rate_limit_syn: self.rate_limit_syn.load(Ordering::Relaxed),
rate_limit_udp: self.rate_limit_udp.load(Ordering::Relaxed),
rate_limit_dns: self.rate_limit_dns.load(Ordering::Relaxed),
protocol_filter: self.protocol_filter.load(Ordering::Relaxed),
dns_blacklist: self.dns_blacklist.load(Ordering::Relaxed),
geo_block: self.geo_block.load(Ordering::Relaxed),
total: self.total.load(Ordering::Relaxed),
}
}
}