mirror of
https://github.com/DaLaw2/NetGuardia.git
synced 2026-08-24 14:10:28 +09:00
refactor(concurrency): finish lock-free migration across 8 subsystems
Migrate the remaining lock sites flagged by the eng review. Pattern
choice was driven by the actual contention shape and what the call
site can tolerate, not by uniformity:
- parking_lot::RwLock for the eBPF admin maps (access_control,
protocol_filter). Critical sections are tiny aya syscalls that
never cross await — replacing tokio::sync::RwLock removes async
overhead without changing semantics.
- moka::sync::Cache<FlowKey, Arc<Mutex<FlowData>>> for the per-queue
FlowTracker. The previous LruCache+Mutex stalled XSK rx for the
duration of the inference snapshot pass; per-shard moka locks plus
per-flow mutexes let the inference loop iterate in parallel with
packet ingest, and W-TinyLFU resists burst-noise eviction of high-
rate attack flows.
- DashMap for AttackAggregator (single-caller, but the &self API
removes a Mutex<AttackAggregator> wrapper from Engine).
- tokio owner-task + mpsc handles for state that needs serialized
read-modify-write across systems:
* DriftDetector (hot-path update is fire-and-forget try_send;
check_drift round-trips via oneshot)
* SOAR rate_limit (DB settings + eBPF RATE_LIMIT_CONFIG batch —
a SQLite tx couldn't cover the eBPF half)
- ArcSwap for snapshot publishing where readers want lock-free reads
(SystemHealth: refresh task owns sysinfo handles; readers just
load the published SystemHealthMetrics).
- AtomicBool CAS gate for non-queueing single-flight (PromoteLock:
concurrent model uploads now get 409 Conflict instead of waiting
silently behind the first promote).
Surface impact: AccessControlPort, AccessControlAdminPort, and
ProtocolFilter drop async_trait; AclService and SOAR call sites
become sync; Engine drops Mutex<AttackAggregator> and
Arc<Mutex<DriftDetector>> in favor of typed handles; SystemHealth
HTTP handlers no longer .await.
All 279 tests pass; clippy -D warnings clean.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
8c0a313a68
commit
10505d88d1
@ -1,8 +1,6 @@
|
||||
use std::net::{IpAddr, SocketAddrV4, SocketAddrV6};
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::adapter::ebpf::access_control::AccessControl;
|
||||
use crate::interface::port::access_control::AccessControlPort;
|
||||
use crate::model::access_control::list_type::ListType;
|
||||
@ -21,9 +19,8 @@ impl EbpfAccessControlAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AccessControlPort for EbpfAccessControlAdapter {
|
||||
async fn block_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
fn block_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
let addr: IpAddr = ip
|
||||
.parse()
|
||||
.map_err(|_| Error::from(EbpfError::InvalidIpAddress(ip.to_string())))?;
|
||||
@ -32,18 +29,16 @@ impl AccessControlPort for EbpfAccessControlAdapter {
|
||||
let socket = SocketAddrV4::new(v4, 0);
|
||||
self.access_control
|
||||
.add_ipv4_list(FlowDirection::Source, ListType::Black, socket)
|
||||
.await
|
||||
}
|
||||
IpAddr::V6(v6) => {
|
||||
let socket = SocketAddrV6::new(v6, 0, 0, 0);
|
||||
self.access_control
|
||||
.add_ipv6_list(FlowDirection::Source, ListType::Black, socket)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn unblock_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
fn unblock_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
let addr: IpAddr = ip
|
||||
.parse()
|
||||
.map_err(|_| Error::from(EbpfError::InvalidIpAddress(ip.to_string())))?;
|
||||
@ -52,13 +47,11 @@ impl AccessControlPort for EbpfAccessControlAdapter {
|
||||
let socket = SocketAddrV4::new(v4, 0);
|
||||
self.access_control
|
||||
.remove_ipv4_list(FlowDirection::Source, ListType::Black, socket)
|
||||
.await
|
||||
}
|
||||
IpAddr::V6(v6) => {
|
||||
let socket = SocketAddrV6::new(v6, 0, 0, 0);
|
||||
self.access_control
|
||||
.remove_ipv6_list(FlowDirection::Source, ListType::Black, socket)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,12 +1,11 @@
|
||||
use std::collections::HashMap;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use aya::maps::{HashMap as AyaHashMap, MapData};
|
||||
use aya::{Ebpf, Pod};
|
||||
use common::model::ip_address::{IPv4, IPv6, Port};
|
||||
use common::model::port_rule::PortRule;
|
||||
use tokio::sync::RwLock;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use crate::interface::port::access_control_admin::AccessControlAdminPort;
|
||||
use crate::model::access_control::ip_address::NativeConvert;
|
||||
@ -57,27 +56,27 @@ impl AccessControl {
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>> {
|
||||
pub fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>> {
|
||||
let map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.read().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.read().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.read().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.read().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.read(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.read(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.read(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.read(),
|
||||
};
|
||||
map_wrapper.get_list()
|
||||
}
|
||||
|
||||
pub async fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>> {
|
||||
pub fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>> {
|
||||
let map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.read().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.read().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.read().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.read().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.read(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.read(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.read(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.read(),
|
||||
};
|
||||
map_wrapper.get_list()
|
||||
}
|
||||
|
||||
pub async fn add_ipv4_list(
|
||||
pub fn add_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
@ -86,15 +85,15 @@ impl AccessControl {
|
||||
let ip: u32 = (*address.ip()).to_bits().to_be();
|
||||
let port = address.port();
|
||||
let mut map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write(),
|
||||
};
|
||||
map_wrapper.add(ip, port)
|
||||
}
|
||||
|
||||
pub async fn add_ipv6_list(
|
||||
pub fn add_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
@ -103,15 +102,15 @@ impl AccessControl {
|
||||
let ip: u128 = (*address.ip()).to_bits().to_be();
|
||||
let port = address.port();
|
||||
let mut map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write(),
|
||||
};
|
||||
map_wrapper.add(ip, port)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4_list(
|
||||
pub fn remove_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
@ -120,15 +119,15 @@ impl AccessControl {
|
||||
let ip: u32 = (*address.ip()).to_bits().to_be();
|
||||
let port = address.port();
|
||||
let mut map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv4_src_blacklist.write(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv4_dst_whitelist.write(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv4_dst_blacklist.write(),
|
||||
};
|
||||
map_wrapper.remove(ip, port)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6_list(
|
||||
pub fn remove_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
@ -137,54 +136,43 @@ impl AccessControl {
|
||||
let ip: u128 = (*address.ip()).to_bits().to_be();
|
||||
let port = address.port();
|
||||
let mut map_wrapper = match (direction, list_type) {
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write().await,
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write().await,
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write().await,
|
||||
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write(),
|
||||
(FlowDirection::Source, ListType::Black) => self.ipv6_src_blacklist.write(),
|
||||
(FlowDirection::Destination, ListType::White) => self.ipv6_dst_whitelist.write(),
|
||||
(FlowDirection::Destination, ListType::Black) => self.ipv6_dst_blacklist.write(),
|
||||
};
|
||||
map_wrapper.remove(ip, port)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AccessControlAdminPort for AccessControl {
|
||||
async fn add_ipv4_list(
|
||||
fn add_ipv4_list(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.add_ipv4_list(direction, list_type, address)
|
||||
}
|
||||
fn add_ipv6_list(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.add_ipv6_list(direction, list_type, address)
|
||||
}
|
||||
fn remove_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error> {
|
||||
self.add_ipv4_list(direction, list_type, address).await
|
||||
self.remove_ipv4_list(direction, list_type, address)
|
||||
}
|
||||
async fn add_ipv6_list(
|
||||
fn remove_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error> {
|
||||
self.add_ipv6_list(direction, list_type, address).await
|
||||
self.remove_ipv6_list(direction, list_type, address)
|
||||
}
|
||||
async fn remove_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error> {
|
||||
self.remove_ipv4_list(direction, list_type, address).await
|
||||
fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>> {
|
||||
self.get_ipv4_list(direction, list_type)
|
||||
}
|
||||
async fn remove_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error> {
|
||||
self.remove_ipv6_list(direction, list_type, address).await
|
||||
}
|
||||
async fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>> {
|
||||
self.get_ipv4_list(direction, list_type).await
|
||||
}
|
||||
async fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>> {
|
||||
self.get_ipv6_list(direction, list_type).await
|
||||
fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>> {
|
||||
self.get_ipv6_list(direction, list_type)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -6,7 +6,7 @@ use aya::{Ebpf, Pod};
|
||||
use common::model::http_method::{HttpMethod, HttpMethodBitmap};
|
||||
use common::model::ip_address::{AddrPortV4, AddrPortV6, IPv4, IPv6};
|
||||
use common::model::placeholder::PlaceHolder;
|
||||
use tokio::sync::RwLock;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use crate::model::access_control::ip_address::NativeConvert;
|
||||
use crate::model::error::Error;
|
||||
@ -55,140 +55,124 @@ impl ProtocolFilter {
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_ipv4_http_service(&self) -> HashMap<SocketAddrV4, Vec<HttpMethod>> {
|
||||
self.ipv4_http_service.read().await.get_http_method()
|
||||
pub fn get_ipv4_http_service(&self) -> HashMap<SocketAddrV4, Vec<HttpMethod>> {
|
||||
self.ipv4_http_service.read().get_http_method()
|
||||
}
|
||||
|
||||
pub async fn get_ipv6_http_service(&self) -> HashMap<SocketAddrV6, Vec<HttpMethod>> {
|
||||
self.ipv6_http_service.read().await.get_http_method()
|
||||
pub fn get_ipv6_http_service(&self) -> HashMap<SocketAddrV6, Vec<HttpMethod>> {
|
||||
self.ipv6_http_service.read().get_http_method()
|
||||
}
|
||||
|
||||
pub async fn add_ipv4_http_service(
|
||||
&self,
|
||||
address: SocketAddrV4,
|
||||
http_method: Vec<HttpMethod>,
|
||||
) -> Result<(), Error> {
|
||||
self.ipv4_http_service
|
||||
.write()
|
||||
.await
|
||||
.add_http_service(address, http_method)
|
||||
pub fn add_ipv4_http_service(&self, address: SocketAddrV4, http_method: Vec<HttpMethod>) -> Result<(), Error> {
|
||||
self.ipv4_http_service.write().add_http_service(address, http_method)
|
||||
}
|
||||
|
||||
pub async fn add_ipv6_http_service(
|
||||
&self,
|
||||
address: SocketAddrV6,
|
||||
http_method: Vec<HttpMethod>,
|
||||
) -> Result<(), Error> {
|
||||
self.ipv6_http_service
|
||||
.write()
|
||||
.await
|
||||
.add_http_service(address, http_method)
|
||||
pub fn add_ipv6_http_service(&self, address: SocketAddrV6, http_method: Vec<HttpMethod>) -> Result<(), Error> {
|
||||
self.ipv6_http_service.write().add_http_service(address, http_method)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4_http_service(
|
||||
pub fn remove_ipv4_http_service(
|
||||
&self,
|
||||
address: SocketAddrV4,
|
||||
removed_http_method: Vec<HttpMethod>,
|
||||
) -> Result<(), Error> {
|
||||
self.ipv4_http_service
|
||||
.write()
|
||||
.await
|
||||
.remove_http_service(address, removed_http_method)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6_http_service(
|
||||
pub fn remove_ipv6_http_service(
|
||||
&self,
|
||||
address: SocketAddrV6,
|
||||
removed_http_method: Vec<HttpMethod>,
|
||||
) -> Result<(), Error> {
|
||||
self.ipv6_http_service
|
||||
.write()
|
||||
.await
|
||||
.remove_http_service(address, removed_http_method)
|
||||
}
|
||||
|
||||
pub async fn is_ssh_white_list_enable(&self) -> bool {
|
||||
self.ssh_white_list_enable.read().await.is_white_list_enable()
|
||||
pub fn is_ssh_white_list_enable(&self) -> bool {
|
||||
self.ssh_white_list_enable.read().is_white_list_enable()
|
||||
}
|
||||
|
||||
pub async fn enable_ssh_white_list(&self) -> Result<(), Error> {
|
||||
self.ssh_white_list_enable.write().await.enable_white_list()
|
||||
pub fn enable_ssh_white_list(&self) -> Result<(), Error> {
|
||||
self.ssh_white_list_enable.write().enable_white_list()
|
||||
}
|
||||
|
||||
pub async fn disable_ssh_white_list(&self) -> Result<(), Error> {
|
||||
self.ssh_white_list_enable.write().await.disable_white_list()
|
||||
pub fn disable_ssh_white_list(&self) -> Result<(), Error> {
|
||||
self.ssh_white_list_enable.write().disable_white_list()
|
||||
}
|
||||
|
||||
pub async fn get_ipv4_ssh_service(&self) -> Vec<SocketAddrV4> {
|
||||
self.ipv4_ssh_service.read().await.get_all()
|
||||
pub fn get_ipv4_ssh_service(&self) -> Vec<SocketAddrV4> {
|
||||
self.ipv4_ssh_service.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn get_ipv6_ssh_service(&self) -> Vec<SocketAddrV6> {
|
||||
self.ipv6_ssh_service.read().await.get_all()
|
||||
pub fn get_ipv6_ssh_service(&self) -> Vec<SocketAddrV6> {
|
||||
self.ipv6_ssh_service.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn add_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.ipv4_ssh_service.write().await.add(address)
|
||||
pub fn add_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.ipv4_ssh_service.write().add(address)
|
||||
}
|
||||
|
||||
pub async fn add_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.ipv6_ssh_service.write().await.add(address)
|
||||
pub fn add_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.ipv6_ssh_service.write().add(address)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.ipv4_ssh_service.write().await.remove(address)
|
||||
pub fn remove_ipv4_ssh_service(&self, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.ipv4_ssh_service.write().remove(address)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.ipv6_ssh_service.write().await.remove(address)
|
||||
pub fn remove_ipv6_ssh_service(&self, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.ipv6_ssh_service.write().remove(address)
|
||||
}
|
||||
|
||||
pub async fn get_ipv4_ssh_white_list(&self) -> Vec<Ipv4Addr> {
|
||||
self.ipv4_ssh_white_list.read().await.get_all()
|
||||
pub fn get_ipv4_ssh_white_list(&self) -> Vec<Ipv4Addr> {
|
||||
self.ipv4_ssh_white_list.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn get_ipv6_ssh_white_list(&self) -> Vec<Ipv6Addr> {
|
||||
self.ipv6_ssh_white_list.read().await.get_all()
|
||||
pub fn get_ipv6_ssh_white_list(&self) -> Vec<Ipv6Addr> {
|
||||
self.ipv6_ssh_white_list.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn add_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_white_list.write().await.add(ip)
|
||||
pub fn add_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_white_list.write().add(ip)
|
||||
}
|
||||
|
||||
pub async fn add_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_white_list.write().await.add(ip)
|
||||
pub fn add_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_white_list.write().add(ip)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_white_list.write().await.remove(ip)
|
||||
pub fn remove_ipv4_ssh_white_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_white_list.write().remove(ip)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_white_list.write().await.remove(ip)
|
||||
pub fn remove_ipv6_ssh_white_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_white_list.write().remove(ip)
|
||||
}
|
||||
|
||||
pub async fn get_ipv4_ssh_black_list(&self) -> Vec<Ipv4Addr> {
|
||||
self.ipv4_ssh_black_list.read().await.get_all()
|
||||
pub fn get_ipv4_ssh_black_list(&self) -> Vec<Ipv4Addr> {
|
||||
self.ipv4_ssh_black_list.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn get_ipv6_ssh_black_list(&self) -> Vec<Ipv6Addr> {
|
||||
self.ipv6_ssh_black_list.read().await.get_all()
|
||||
pub fn get_ipv6_ssh_black_list(&self) -> Vec<Ipv6Addr> {
|
||||
self.ipv6_ssh_black_list.read().get_all()
|
||||
}
|
||||
|
||||
pub async fn add_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_black_list.write().await.add(ip)
|
||||
pub fn add_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_black_list.write().add(ip)
|
||||
}
|
||||
|
||||
pub async fn add_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_black_list.write().await.add(ip)
|
||||
pub fn add_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_black_list.write().add(ip)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_black_list.write().await.remove(ip)
|
||||
pub fn remove_ipv4_ssh_black_list(&self, ip: Ipv4Addr) -> Result<(), Error> {
|
||||
self.ipv4_ssh_black_list.write().remove(ip)
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_black_list.write().await.remove(ip)
|
||||
pub fn remove_ipv6_ssh_black_list(&self, ip: Ipv6Addr) -> Result<(), Error> {
|
||||
self.ipv6_ssh_black_list.write().remove(ip)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -27,13 +27,13 @@ pub fn initialize() -> Scope {
|
||||
|
||||
async fn get_ipv4_list(path: web::Path<(FlowDirection, ListType)>, acl: web::Data<AclService>) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
let list = acl.access_control().get_ipv4_list(direction, list_type).await;
|
||||
let list = acl.access_control().get_ipv4_list(direction, list_type);
|
||||
HttpResponse::Ok().json(list)
|
||||
}
|
||||
|
||||
async fn get_ipv6_list(path: web::Path<(FlowDirection, ListType)>, acl: web::Data<AclService>) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
let list = acl.access_control().get_ipv6_list(direction, list_type).await;
|
||||
let list = acl.access_control().get_ipv6_list(direction, list_type);
|
||||
HttpResponse::Ok().json(list)
|
||||
}
|
||||
|
||||
@ -43,7 +43,7 @@ async fn add_ipv4_list(
|
||||
acl: web::Data<AclService>,
|
||||
) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
match acl.add_ipv4(direction, list_type, address.into_inner()).await {
|
||||
match acl.add_ipv4(direction, list_type, address.into_inner()) {
|
||||
Ok(_) => HttpResponse::Ok().finish(),
|
||||
Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})),
|
||||
}
|
||||
@ -55,7 +55,7 @@ async fn add_ipv6_list(
|
||||
acl: web::Data<AclService>,
|
||||
) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
match acl.add_ipv6(direction, list_type, address.into_inner()).await {
|
||||
match acl.add_ipv6(direction, list_type, address.into_inner()) {
|
||||
Ok(_) => HttpResponse::Ok().finish(),
|
||||
Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})),
|
||||
}
|
||||
@ -67,7 +67,7 @@ async fn remove_ipv4_list(
|
||||
acl: web::Data<AclService>,
|
||||
) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
match acl.remove_ipv4(direction, list_type, address.into_inner()).await {
|
||||
match acl.remove_ipv4(direction, list_type, address.into_inner()) {
|
||||
Ok(_) => HttpResponse::Ok().finish(),
|
||||
Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})),
|
||||
}
|
||||
@ -79,7 +79,7 @@ async fn remove_ipv6_list(
|
||||
acl: web::Data<AclService>,
|
||||
) -> impl Responder {
|
||||
let (direction, list_type) = path.into_inner();
|
||||
match acl.remove_ipv6(direction, list_type, address.into_inner()).await {
|
||||
match acl.remove_ipv6(direction, list_type, address.into_inner()) {
|
||||
Ok(_) => HttpResponse::Ok().finish(),
|
||||
Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({"error": e.to_string()})),
|
||||
}
|
||||
|
||||
@ -111,11 +111,11 @@ fn ssh_blacklist_scope() -> Scope {
|
||||
// --- HTTP service handlers ---
|
||||
|
||||
async fn get_ipv4_http_service(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv4_http_service().await)
|
||||
HttpResponse::Ok().json(service.get_ipv4_http_service())
|
||||
}
|
||||
|
||||
async fn get_ipv6_http_service(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv6_http_service().await)
|
||||
HttpResponse::Ok().json(service.get_ipv6_http_service())
|
||||
}
|
||||
|
||||
async fn add_ipv4_http_service(
|
||||
@ -123,7 +123,7 @@ async fn add_ipv4_http_service(
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
let (addr, methods) = payload.into_inner();
|
||||
ok_or_error(service.add_ipv4_http_service(addr, methods).await)
|
||||
ok_or_error(service.add_ipv4_http_service(addr, methods))
|
||||
}
|
||||
|
||||
async fn add_ipv6_http_service(
|
||||
@ -131,7 +131,7 @@ async fn add_ipv6_http_service(
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
let (addr, methods) = payload.into_inner();
|
||||
ok_or_error(service.add_ipv6_http_service(addr, methods).await)
|
||||
ok_or_error(service.add_ipv6_http_service(addr, methods))
|
||||
}
|
||||
|
||||
async fn remove_ipv4_http_service(
|
||||
@ -139,7 +139,7 @@ async fn remove_ipv4_http_service(
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
let (addr, methods) = payload.into_inner();
|
||||
ok_or_error(service.remove_ipv4_http_service(addr, methods).await)
|
||||
ok_or_error(service.remove_ipv4_http_service(addr, methods))
|
||||
}
|
||||
|
||||
async fn remove_ipv6_http_service(
|
||||
@ -147,113 +147,113 @@ async fn remove_ipv6_http_service(
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
let (addr, methods) = payload.into_inner();
|
||||
ok_or_error(service.remove_ipv6_http_service(addr, methods).await)
|
||||
ok_or_error(service.remove_ipv6_http_service(addr, methods))
|
||||
}
|
||||
|
||||
// --- SSH service handlers ---
|
||||
|
||||
async fn get_ipv4_ssh_service(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_service().await)
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_service())
|
||||
}
|
||||
|
||||
async fn get_ipv6_ssh_service(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_service().await)
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_service())
|
||||
}
|
||||
|
||||
async fn add_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv4_ssh_service(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv4_ssh_service(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn add_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv6_ssh_service(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv6_ssh_service(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv4_ssh_service(
|
||||
ip_addr: web::Json<SocketAddrV4>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv4_ssh_service(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv4_ssh_service(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv6_ssh_service(
|
||||
ip_addr: web::Json<SocketAddrV6>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv6_ssh_service(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv6_ssh_service(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
// --- SSH whitelist handlers ---
|
||||
|
||||
async fn is_ssh_white_list_enable(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.is_ssh_white_list_enable().await)
|
||||
HttpResponse::Ok().json(service.is_ssh_white_list_enable())
|
||||
}
|
||||
|
||||
async fn enable_ssh_white_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.enable_ssh_white_list().await)
|
||||
ok_or_error(service.enable_ssh_white_list())
|
||||
}
|
||||
|
||||
async fn disable_ssh_white_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.disable_ssh_white_list().await)
|
||||
ok_or_error(service.disable_ssh_white_list())
|
||||
}
|
||||
|
||||
async fn get_ipv4_ssh_white_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_white_list().await)
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_white_list())
|
||||
}
|
||||
|
||||
async fn get_ipv6_ssh_white_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_white_list().await)
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_white_list())
|
||||
}
|
||||
|
||||
async fn add_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv4_ssh_white_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv4_ssh_white_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn add_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv6_ssh_white_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv6_ssh_white_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv4_ssh_white_list(
|
||||
ip_addr: web::Json<Ipv4Addr>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv4_ssh_white_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv4_ssh_white_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv6_ssh_white_list(
|
||||
ip_addr: web::Json<Ipv6Addr>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv6_ssh_white_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv6_ssh_white_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
// --- SSH blacklist handlers ---
|
||||
|
||||
async fn get_ipv4_ssh_black_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_black_list().await)
|
||||
HttpResponse::Ok().json(service.get_ipv4_ssh_black_list())
|
||||
}
|
||||
|
||||
async fn get_ipv6_ssh_black_list(service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_black_list().await)
|
||||
HttpResponse::Ok().json(service.get_ipv6_ssh_black_list())
|
||||
}
|
||||
|
||||
async fn add_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv4_ssh_black_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv4_ssh_black_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn add_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<ProtocolFilter>) -> impl Responder {
|
||||
ok_or_error(service.add_ipv6_ssh_black_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.add_ipv6_ssh_black_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv4_ssh_black_list(
|
||||
ip_addr: web::Json<Ipv4Addr>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv4_ssh_black_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv4_ssh_black_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
async fn remove_ipv6_ssh_black_list(
|
||||
ip_addr: web::Json<Ipv6Addr>,
|
||||
service: web::Data<ProtocolFilter>,
|
||||
) -> impl Responder {
|
||||
ok_or_error(service.remove_ipv6_ssh_black_list(ip_addr.into_inner()).await)
|
||||
ok_or_error(service.remove_ipv6_ssh_black_list(ip_addr.into_inner()))
|
||||
}
|
||||
|
||||
@ -12,12 +12,12 @@ pub fn initialize() -> Scope {
|
||||
}
|
||||
|
||||
async fn get_current_metrics(health: web::Data<SystemHealth>) -> impl Responder {
|
||||
let metrics = health.get_current_metrics().await;
|
||||
let metrics = health.get_current_metrics();
|
||||
HttpResponse::Ok().json(metrics)
|
||||
}
|
||||
|
||||
async fn get_health_status(health: web::Data<SystemHealth>) -> impl Responder {
|
||||
let status = health.is_system_healthy().await;
|
||||
let status = health.is_system_healthy();
|
||||
HttpResponse::Ok().json(status)
|
||||
}
|
||||
|
||||
|
||||
@ -16,7 +16,7 @@ pub fn initialize() -> Scope {
|
||||
async fn get_status(engine: web::Data<Engine>) -> impl Responder {
|
||||
let trackers = engine.trackers();
|
||||
let num_trackers = trackers.len();
|
||||
let total_flows: usize = trackers.iter().map(|t| t.lock().flow_count()).sum();
|
||||
let total_flows: usize = trackers.iter().map(|t| t.flow_count()).sum();
|
||||
let has_traffic_logger = engine.has_traffic_logger();
|
||||
|
||||
HttpResponse::Ok().json(serde_json::json!({
|
||||
|
||||
@ -20,6 +20,7 @@ use std::io;
|
||||
use std::io::Read;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use actix_multipart::Multipart;
|
||||
@ -29,7 +30,6 @@ use serde_json::Value as JsonValue;
|
||||
use sha2::{Digest, Sha256};
|
||||
use tokio::fs;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tokio::sync::Mutex as AsyncMutex;
|
||||
use tokio::task;
|
||||
use uuid::Uuid;
|
||||
|
||||
@ -75,11 +75,50 @@ const AUDIT_ACTOR_SECURITY_ADMIN_PREFIX: &str = "SecurityAdmin";
|
||||
/// chain too.
|
||||
const AUDIT_ACTION_MODEL_SWAP: &str = "model_swap";
|
||||
|
||||
/// Process-wide lock serializing the rename step of every promote.
|
||||
/// The critical section is tiny (three `tokio::fs::rename` syscalls)
|
||||
/// but must never interleave: a concurrent promote mid-rename could
|
||||
/// Process-wide gate that ensures only one promote ever runs the rename
|
||||
/// section at a time. The critical section is tiny (three `tokio::fs::rename`
|
||||
/// syscalls) but must never interleave: a concurrent promote mid-rename could
|
||||
/// leave `models/` pointing at a manifest whose ONNX hasn't landed yet.
|
||||
pub type PromoteLock = AsyncMutex<()>;
|
||||
///
|
||||
/// Unlike a mutex, the gate does not queue. A second concurrent promote sees
|
||||
/// the gate held and gets `PromoteError::ConcurrentPromote` immediately —
|
||||
/// administrators wanting to swap models should know another swap is in flight
|
||||
/// rather than silently waiting behind it.
|
||||
#[derive(Default)]
|
||||
pub struct PromoteGate {
|
||||
in_progress: AtomicBool,
|
||||
}
|
||||
|
||||
impl PromoteGate {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Try to claim the gate. Returns `Some(guard)` on success; `None` when
|
||||
/// another promote is already inside the rename section. The guard
|
||||
/// releases the gate when dropped, including on panic.
|
||||
fn try_acquire(&self) -> Option<PromoteGuard<'_>> {
|
||||
if self
|
||||
.in_progress
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
{
|
||||
Some(PromoteGuard { gate: self })
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct PromoteGuard<'a> {
|
||||
gate: &'a PromoteGate,
|
||||
}
|
||||
|
||||
impl Drop for PromoteGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.gate.in_progress.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn initialize() -> Scope {
|
||||
web::scope("/models").route("/upload", web::post().to(upload))
|
||||
@ -96,7 +135,7 @@ async fn upload(
|
||||
app_config: web::Data<Arc<AppConfig>>,
|
||||
inference: web::Data<Inference>,
|
||||
comm: web::Data<CommunicationManager>,
|
||||
promote_lock: web::Data<PromoteLock>,
|
||||
promote_lock: web::Data<PromoteGate>,
|
||||
claims: AuthClaims,
|
||||
payload: Multipart,
|
||||
) -> impl Responder {
|
||||
@ -440,7 +479,7 @@ async fn validate_and_promote(
|
||||
summary: &UploadSummary,
|
||||
inference: &Inference,
|
||||
comm: &CommunicationManager,
|
||||
promote_lock: &PromoteLock,
|
||||
promote_lock: &PromoteGate,
|
||||
actor_username: &str,
|
||||
batch_size: usize,
|
||||
) -> Result<PromoteReport, PromoteError> {
|
||||
@ -487,7 +526,7 @@ async fn validate_and_promote(
|
||||
|
||||
let before_status = inference.current_status();
|
||||
|
||||
let _guard = promote_lock.lock().await;
|
||||
let _guard = promote_lock.try_acquire().ok_or(PromoteError::ConcurrentPromote)?;
|
||||
let models_dir = PathBuf::from(MODELS_DIR);
|
||||
let target_onnx = models_dir.join(&declared_onnx);
|
||||
fs::rename(&staged_onnx, &target_onnx)
|
||||
@ -552,6 +591,7 @@ enum PromoteError {
|
||||
UnsupportedAdapter,
|
||||
StagingIo(String),
|
||||
PromoteIo(String),
|
||||
ConcurrentPromote,
|
||||
}
|
||||
|
||||
impl PromoteError {
|
||||
@ -567,10 +607,12 @@ impl PromoteError {
|
||||
),
|
||||
Self::StagingIo(err) => (500, format!("staging io error: {err}")),
|
||||
Self::PromoteIo(err) => (500, format!("promote io error: {err}")),
|
||||
Self::ConcurrentPromote => (409, "another model promote is already in progress".to_string()),
|
||||
};
|
||||
let body = serde_json::json!({ "error": message });
|
||||
match status {
|
||||
422 => HttpResponse::UnprocessableEntity().json(body),
|
||||
409 => HttpResponse::Conflict().json(body),
|
||||
_ => HttpResponse::InternalServerError().json(body),
|
||||
}
|
||||
}
|
||||
|
||||
@ -32,13 +32,8 @@ impl AclService {
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn add_ipv4(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error> {
|
||||
self.access_control.add_ipv4_list(direction, list_type, address).await?;
|
||||
pub fn add_ipv4(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV4) -> Result<(), Error> {
|
||||
self.access_control.add_ipv4_list(direction, list_type, address)?;
|
||||
if let Err(e) = self.db.insert_acl_rule(
|
||||
4,
|
||||
direction_str(direction),
|
||||
@ -46,11 +41,7 @@ impl AclService {
|
||||
&address.ip().to_string(),
|
||||
address.port(),
|
||||
) {
|
||||
if let Err(rollback_err) = self
|
||||
.access_control
|
||||
.remove_ipv4_list(direction, list_type, address)
|
||||
.await
|
||||
{
|
||||
if let Err(rollback_err) = self.access_control.remove_ipv4_list(direction, list_type, address) {
|
||||
log!(EbpfError::RollbackFailed(rollback_err));
|
||||
}
|
||||
return Err(e);
|
||||
@ -58,13 +49,8 @@ impl AclService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn add_ipv6(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error> {
|
||||
self.access_control.add_ipv6_list(direction, list_type, address).await?;
|
||||
pub fn add_ipv6(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV6) -> Result<(), Error> {
|
||||
self.access_control.add_ipv6_list(direction, list_type, address)?;
|
||||
if let Err(e) = self.db.insert_acl_rule(
|
||||
6,
|
||||
direction_str(direction),
|
||||
@ -72,11 +58,7 @@ impl AclService {
|
||||
&address.ip().to_string(),
|
||||
address.port(),
|
||||
) {
|
||||
if let Err(rollback_err) = self
|
||||
.access_control
|
||||
.remove_ipv6_list(direction, list_type, address)
|
||||
.await
|
||||
{
|
||||
if let Err(rollback_err) = self.access_control.remove_ipv6_list(direction, list_type, address) {
|
||||
log!(EbpfError::RollbackFailed(rollback_err));
|
||||
}
|
||||
return Err(e);
|
||||
@ -84,15 +66,13 @@ impl AclService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn remove_ipv4(
|
||||
pub fn remove_ipv4(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error> {
|
||||
self.access_control
|
||||
.remove_ipv4_list(direction, list_type, address)
|
||||
.await?;
|
||||
self.access_control.remove_ipv4_list(direction, list_type, address)?;
|
||||
if let Err(e) = self.db.delete_acl_rule(
|
||||
4,
|
||||
direction_str(direction),
|
||||
@ -100,7 +80,7 @@ impl AclService {
|
||||
&address.ip().to_string(),
|
||||
address.port(),
|
||||
) {
|
||||
if let Err(rollback_err) = self.access_control.add_ipv4_list(direction, list_type, address).await {
|
||||
if let Err(rollback_err) = self.access_control.add_ipv4_list(direction, list_type, address) {
|
||||
log!(EbpfError::RollbackFailed(rollback_err));
|
||||
}
|
||||
return Err(e);
|
||||
@ -108,15 +88,13 @@ impl AclService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn remove_ipv6(
|
||||
pub fn remove_ipv6(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error> {
|
||||
self.access_control
|
||||
.remove_ipv6_list(direction, list_type, address)
|
||||
.await?;
|
||||
self.access_control.remove_ipv6_list(direction, list_type, address)?;
|
||||
if let Err(e) = self.db.delete_acl_rule(
|
||||
6,
|
||||
direction_str(direction),
|
||||
@ -124,7 +102,7 @@ impl AclService {
|
||||
&address.ip().to_string(),
|
||||
address.port(),
|
||||
) {
|
||||
if let Err(rollback_err) = self.access_control.add_ipv6_list(direction, list_type, address).await {
|
||||
if let Err(rollback_err) = self.access_control.add_ipv6_list(direction, list_type, address) {
|
||||
log!(EbpfError::RollbackFailed(rollback_err));
|
||||
}
|
||||
return Err(e);
|
||||
|
||||
@ -4,20 +4,21 @@
|
||||
//! fire based on how many hits a given attack class needs (manifest-driven)
|
||||
//! and how far the rolling-average score beats the confidence threshold.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use dashmap::DashMap;
|
||||
|
||||
use crate::model::detection::ml_detection::FlowKey;
|
||||
|
||||
pub struct AttackAggregator {
|
||||
detections: HashMap<FlowKey, Vec<(Instant, f32)>>,
|
||||
detections: DashMap<FlowKey, Vec<(Instant, f32)>>,
|
||||
window_duration: Duration,
|
||||
}
|
||||
|
||||
impl AttackAggregator {
|
||||
pub fn new(window_secs: u64) -> Self {
|
||||
Self {
|
||||
detections: HashMap::new(),
|
||||
detections: DashMap::new(),
|
||||
window_duration: Duration::from_secs(window_secs),
|
||||
}
|
||||
}
|
||||
@ -30,7 +31,7 @@ impl AttackAggregator {
|
||||
/// - `alert_multiplier`: scales `threshold` before the average-score
|
||||
/// comparison, driven by the manifest's `thresholds.alert_multiplier`.
|
||||
pub fn should_alert(
|
||||
&mut self,
|
||||
&self,
|
||||
flow_key: &FlowKey,
|
||||
score: f32,
|
||||
threshold: f32,
|
||||
@ -39,7 +40,7 @@ impl AttackAggregator {
|
||||
) -> bool {
|
||||
let now = Instant::now();
|
||||
|
||||
let detections = self.detections.entry(flow_key.clone()).or_default();
|
||||
let mut detections = self.detections.entry(flow_key.clone()).or_default();
|
||||
detections.retain(|(time, _)| now.duration_since(*time) < self.window_duration);
|
||||
detections.push((now, score));
|
||||
|
||||
@ -51,7 +52,7 @@ impl AttackAggregator {
|
||||
false
|
||||
}
|
||||
|
||||
pub fn cleanup(&mut self) {
|
||||
pub fn cleanup(&self) {
|
||||
let now = Instant::now();
|
||||
self.detections.retain(|_, detections| {
|
||||
detections.retain(|(time, _)| now.duration_since(*time) < self.window_duration);
|
||||
@ -79,7 +80,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn required_three_fires_on_third_hit() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
assert!(!agg.should_alert(&key, 5.0, 1.0, 3, TEST_MULTIPLIER));
|
||||
assert!(!agg.should_alert(&key, 5.0, 1.0, 3, TEST_MULTIPLIER));
|
||||
@ -88,14 +89,14 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn required_one_fires_immediately() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
assert!(agg.should_alert(&key, 5.0, 1.0, 1, TEST_MULTIPLIER));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn required_six_needs_six_hits() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
for _ in 0..5 {
|
||||
assert!(!agg.should_alert(&key, 5.0, 1.0, 6, TEST_MULTIPLIER));
|
||||
@ -105,7 +106,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn average_score_at_or_below_scaled_threshold_does_not_fire() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
// score 1.0, threshold 1.0, multiplier 1.2 → gate is 1.2; 1.0 misses.
|
||||
assert!(!agg.should_alert(&key, 1.0, 1.0, 1, TEST_MULTIPLIER));
|
||||
@ -113,7 +114,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn larger_multiplier_raises_the_bar() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
// multiplier 2.5, threshold 1.0 → gate is 2.5; score 2.0 misses.
|
||||
assert!(!agg.should_alert(&key, 2.0, 1.0, 1, 2.5));
|
||||
@ -121,7 +122,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn cleanup_preserves_fresh_entries() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key = test_key();
|
||||
agg.should_alert(&key, 5.0, 1.0, 10, TEST_MULTIPLIER);
|
||||
assert!(agg.detections.contains_key(&key));
|
||||
@ -131,7 +132,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn independent_flows_track_separately() {
|
||||
let mut agg = AttackAggregator::new(60);
|
||||
let agg = AttackAggregator::new(60);
|
||||
let key_a = test_key();
|
||||
let mut key_b = test_key();
|
||||
key_b.dst_port = 81;
|
||||
|
||||
@ -1,11 +1,18 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
use crate::model::detection::drift::{DriftReport, FeatureBaselines};
|
||||
|
||||
/// Maximum number of snapshots to retain, preventing unbounded memory growth.
|
||||
const MAX_SNAPSHOTS: usize = 10_000;
|
||||
|
||||
/// Channel depth for the owner-task command queue. With a typical inference
|
||||
/// batch of 100 flows per second, 1024 gives ~10s of cushion before the
|
||||
/// hot path begins shedding samples.
|
||||
const DRIFT_CMD_CHANNEL_CAPACITY: usize = 1024;
|
||||
|
||||
/// Tracks rolling mean/stddev of normalized input features over a configurable window.
|
||||
/// Compares against training-time baselines to detect data drift.
|
||||
pub struct DriftDetector {
|
||||
@ -106,6 +113,64 @@ impl DriftDetector {
|
||||
}
|
||||
}
|
||||
|
||||
/// Command queue between drift-detector callers and the owner task.
|
||||
enum DriftCmd {
|
||||
Update(Vec<f64>),
|
||||
CheckDrift {
|
||||
reply: oneshot::Sender<Option<DriftReport>>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Lock-free handle to a `DriftDetector` running on its own tokio task.
|
||||
///
|
||||
/// The hot path is `update`, called from the ML engine's inference tick on
|
||||
/// the spawn-blocking pool — it must not await, so we use `try_send` and
|
||||
/// silently drop the sample when the channel is full. Drift is a statistical
|
||||
/// signal computed over thousands of snapshots in a window; losing a few
|
||||
/// samples under back-pressure does not change the verdict.
|
||||
///
|
||||
/// `check_drift` is called from the periodic drift monitor (tokio task), so
|
||||
/// it can `await` the round-trip naturally.
|
||||
#[derive(Clone)]
|
||||
pub struct DriftDetectorHandle {
|
||||
tx: mpsc::Sender<DriftCmd>,
|
||||
}
|
||||
|
||||
impl DriftDetectorHandle {
|
||||
/// Spawn the owner task on the current tokio runtime and return a handle.
|
||||
pub fn spawn(baselines: Option<FeatureBaselines>, drift_window: Duration) -> Self {
|
||||
let (tx, mut rx) = mpsc::channel::<DriftCmd>(DRIFT_CMD_CHANNEL_CAPACITY);
|
||||
tokio::spawn(async move {
|
||||
let mut detector = DriftDetector::new(baselines, drift_window);
|
||||
while let Some(cmd) = rx.recv().await {
|
||||
match cmd {
|
||||
DriftCmd::Update(features) => detector.update(&features),
|
||||
DriftCmd::CheckDrift { reply } => {
|
||||
let _ = reply.send(detector.check_drift());
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
Self { tx }
|
||||
}
|
||||
|
||||
/// Fire-and-forget update. Drops the sample silently when the channel is
|
||||
/// full or the owner task has shut down (statistical tolerance — see the
|
||||
/// type-level doc).
|
||||
pub fn update(&self, features: Vec<f64>) {
|
||||
let _ = self.tx.try_send(DriftCmd::Update(features));
|
||||
}
|
||||
|
||||
/// Round-trip drift query. Returns `None` if the channel is closed.
|
||||
pub async fn check_drift(&self) -> Option<DriftReport> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
if self.tx.send(DriftCmd::CheckDrift { reply: reply_tx }).await.is_err() {
|
||||
return None;
|
||||
}
|
||||
reply_rx.await.unwrap_or(None)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@ -8,14 +8,13 @@ use std::sync::Arc;
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use macros::log;
|
||||
use parking_lot::Mutex;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio::task::spawn_blocking;
|
||||
use tokio::time::interval;
|
||||
|
||||
use super::aggregator::AttackAggregator;
|
||||
use super::alert::MLAlert;
|
||||
use super::drift_detector::DriftDetector;
|
||||
use super::drift_detector::DriftDetectorHandle;
|
||||
use super::flow_tracker::{FlowData, FlowTracker};
|
||||
use super::inference::Inference;
|
||||
use super::traffic_logger::TrafficLogger;
|
||||
@ -41,13 +40,15 @@ const ML_MIN_PACKETS_FLOOR: usize = 5;
|
||||
|
||||
/// Per-queue tracker. With symmetric hash in eBPF, both directions of a flow
|
||||
/// land on the same queue, so per-queue trackers correctly see bidirectional flows.
|
||||
pub type ThreadTracker = Arc<Mutex<FlowTracker>>;
|
||||
/// `FlowTracker` itself is internally synchronized (DashMap), so the
|
||||
/// per-queue handle is a plain `Arc`.
|
||||
pub type ThreadTracker = Arc<FlowTracker>;
|
||||
|
||||
pub struct Engine {
|
||||
trackers: Vec<ThreadTracker>,
|
||||
inference_pipeline: Arc<Inference>,
|
||||
aggregator: Mutex<AttackAggregator>,
|
||||
drift_detector: Arc<Mutex<DriftDetector>>,
|
||||
aggregator: AttackAggregator,
|
||||
drift_detector: DriftDetectorHandle,
|
||||
ml_alert: Arc<MLAlert>,
|
||||
min_packets: usize,
|
||||
/// Confirmations count used when the active manifest's label has no
|
||||
@ -64,7 +65,7 @@ impl Engine {
|
||||
pub fn new(
|
||||
inference_pipeline: Arc<Inference>,
|
||||
ml_alert: Arc<MLAlert>,
|
||||
drift_detector: Arc<Mutex<DriftDetector>>,
|
||||
drift_detector: DriftDetectorHandle,
|
||||
engine_config: EngineConfig,
|
||||
traffic_logger: Option<Arc<TrafficLogger>>,
|
||||
num_threads: u32,
|
||||
@ -72,11 +73,11 @@ impl Engine {
|
||||
let interval_secs = engine_config.inference_interval_secs.max(1);
|
||||
let ticks_per_window = engine_config.aggregator_window_secs / interval_secs;
|
||||
let default_confirmations = (ticks_per_window / DEFAULT_CONFIRMATION_WINDOW_FRACTION).max(1) as usize;
|
||||
let aggregator = Mutex::new(AttackAggregator::new(engine_config.aggregator_window_secs));
|
||||
let aggregator = AttackAggregator::new(engine_config.aggregator_window_secs);
|
||||
|
||||
let max_flows_per_thread = engine_config.max_flows / (num_threads as usize).max(1);
|
||||
let trackers: Vec<ThreadTracker> = (0..num_threads)
|
||||
.map(|_| Arc::new(Mutex::new(FlowTracker::new(max_flows_per_thread))))
|
||||
.map(|_| Arc::new(FlowTracker::new(max_flows_per_thread)))
|
||||
.collect();
|
||||
|
||||
Self {
|
||||
@ -216,14 +217,12 @@ impl Engine {
|
||||
.unwrap_or(0);
|
||||
|
||||
for tracker in &self.trackers {
|
||||
let mut t = tracker.lock();
|
||||
t.cleanup_stale_flows(now_us);
|
||||
tracker.cleanup_stale_flows(now_us);
|
||||
}
|
||||
|
||||
for tracker in &self.trackers {
|
||||
let mut t = tracker.lock();
|
||||
total_count += t.flow_count();
|
||||
all_flows.extend(t.get_uninferred_flows().into_iter().filter(|flow| {
|
||||
total_count += tracker.flow_count();
|
||||
all_flows.extend(tracker.get_uninferred_flows().into_iter().filter(|flow| {
|
||||
let total_packets = flow.packet_count();
|
||||
total_packets >= Self::effective_min_packets(&flow.flow_key, self.min_packets)
|
||||
&& !Self::is_strong_benign(
|
||||
@ -272,7 +271,6 @@ impl Engine {
|
||||
fn update_drift(&self, batch: &[FlowData]) {
|
||||
let batch = &batch[..batch.len().min(self.batch_size)];
|
||||
let config = &self.inference_pipeline.config;
|
||||
let mut dd = self.drift_detector.lock();
|
||||
for flow in batch {
|
||||
let features = FlowFeatures::extract(flow, &config.ae_feature_names);
|
||||
let normalized: Vec<f64> = features
|
||||
@ -281,7 +279,7 @@ impl Engine {
|
||||
.zip(config.ae_scaler_mean.iter().zip(config.ae_scaler_std.iter()))
|
||||
.map(|(&val, (&mean, &std))| if std.abs() > 1e-12 { (val - mean) / std } else { 0.0 })
|
||||
.collect();
|
||||
dd.update(&normalized);
|
||||
self.drift_detector.update(normalized);
|
||||
}
|
||||
}
|
||||
|
||||
@ -309,7 +307,6 @@ impl Engine {
|
||||
stats.flows_per_second
|
||||
));
|
||||
|
||||
let mut aggregator = self.aggregator.lock();
|
||||
let config = &self.inference_pipeline.config;
|
||||
for result in &results {
|
||||
if result.is_attack {
|
||||
@ -318,7 +315,7 @@ impl Engine {
|
||||
.as_deref()
|
||||
.and_then(|at| self.inference_pipeline.confirmations_for_attack_type(at))
|
||||
.unwrap_or(self.default_confirmations);
|
||||
let should_alert = aggregator.should_alert(
|
||||
let should_alert = self.aggregator.should_alert(
|
||||
&result.flow_key_raw,
|
||||
result.confidence,
|
||||
config.class_min_confidence,
|
||||
@ -340,7 +337,7 @@ impl Engine {
|
||||
}
|
||||
}
|
||||
|
||||
aggregator.cleanup();
|
||||
self.aggregator.cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
@ -351,7 +348,7 @@ struct QueueTrackerSink {
|
||||
|
||||
impl PacketSink for QueueTrackerSink {
|
||||
fn process_packet(&self, packet: UserPacket, is_ingress: bool) {
|
||||
self.tracker.lock().process_packet(packet, is_ingress);
|
||||
self.tracker.process_packet(packet, is_ingress);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,7 +1,8 @@
|
||||
use std::num::NonZero;
|
||||
use std::sync::Arc;
|
||||
|
||||
use common::define::tcp_flags::*;
|
||||
use lru::LruCache;
|
||||
use moka::sync::Cache;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use crate::model::config::constants::{
|
||||
FLOW_BULK_MIN_BYTES, FLOW_BULK_MIN_PACKETS, FLOW_IDLE_THRESHOLD_US, FLOW_IDLE_TIMEOUT_US,
|
||||
@ -205,31 +206,41 @@ impl FlowData {
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-thread flow tracker. No locks — each XSK thread owns one.
|
||||
/// RSS guarantees the same flow always goes to the same thread.
|
||||
/// Uses LruCache for O(1) eviction instead of O(n) min_by_key scan.
|
||||
/// Per-flow handle: an `Arc` so map operations stay copy-cheap, with an inner
|
||||
/// `Mutex` because `add_packet` is a read-modify-write that needs exclusive
|
||||
/// access. Same-flow packets land on the same XSK queue (symmetric eBPF
|
||||
/// hash), so this mutex is effectively single-writer; the inference tick
|
||||
/// briefly contends only when it clones the entry for a snapshot.
|
||||
type FlowEntry = Arc<Mutex<FlowData>>;
|
||||
|
||||
/// Per-queue flow tracker backed by a sharded W-TinyLFU cache (`moka`).
|
||||
///
|
||||
/// The hot path (`process_packet`) acquires only the per-shard moka lock
|
||||
/// and the per-flow entry mutex — never a global tracker lock — so the
|
||||
/// inference loop's snapshot pass (`get_uninferred_flows`,
|
||||
/// `cleanup_stale_flows`) can run in parallel without stalling AF_XDP rx.
|
||||
/// W-TinyLFU's frequency sketch keeps high-rate attack flows resident
|
||||
/// even when burst noise floods the cache, which a strict-LRU eviction
|
||||
/// policy would mishandle.
|
||||
pub struct FlowTracker {
|
||||
active: LruCache<FlowKey, FlowData>,
|
||||
active: Cache<FlowKey, FlowEntry>,
|
||||
}
|
||||
|
||||
impl FlowTracker {
|
||||
pub fn new(max_flows: usize) -> Self {
|
||||
// SAFETY: max(1, max_flows) ensures NonZero is never zero.
|
||||
let cap = NonZero::new(max_flows.max(1)).unwrap_or_else(|| unreachable!());
|
||||
let cap = max_flows.max(1) as u64;
|
||||
Self {
|
||||
active: LruCache::new(cap),
|
||||
active: Cache::builder().max_capacity(cap).build(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn process_packet(&mut self, mut packet: UserPacket, is_ingress: bool) {
|
||||
pub fn process_packet(&self, mut packet: UserPacket, is_ingress: bool) {
|
||||
let packet_key = FlowKey::from_packet(&packet);
|
||||
let reversed_key = packet_key.reverse();
|
||||
|
||||
// Try to match an existing flow first (canonical key already established).
|
||||
// Use peek() to avoid promoting — we'll promote via get_mut() below.
|
||||
let (actual_key, is_forward) = if self.active.peek(&packet_key).is_some() {
|
||||
let (actual_key, is_forward) = if self.active.contains_key(&packet_key) {
|
||||
(packet_key, true)
|
||||
} else if self.active.peek(&reversed_key).is_some() {
|
||||
} else if self.active.contains_key(&reversed_key) {
|
||||
(reversed_key, false)
|
||||
} else {
|
||||
// New flow: determine initiator using TCP flags, fall back to is_ingress.
|
||||
@ -261,7 +272,6 @@ impl FlowTracker {
|
||||
|
||||
packet.is_forward = is_forward;
|
||||
|
||||
// Record which interface the initiator is on for this flow.
|
||||
let initiator_direction = if is_forward {
|
||||
if is_ingress {
|
||||
Direction::Ingress
|
||||
@ -274,62 +284,59 @@ impl FlowTracker {
|
||||
Direction::Ingress
|
||||
};
|
||||
|
||||
// LruCache::push handles eviction automatically when capacity is exceeded (O(1)).
|
||||
// If the flow already exists, get_mut promotes it to MRU; otherwise push creates it.
|
||||
if let Some(flow) = self.active.get_mut(&actual_key) {
|
||||
flow.add_packet(&packet);
|
||||
} else {
|
||||
let mut flow = FlowData::new(actual_key.clone(), &packet, initiator_direction);
|
||||
flow.add_packet(&packet);
|
||||
self.active.push(actual_key, flow);
|
||||
}
|
||||
let key_for_init = actual_key.clone();
|
||||
let entry = self.active.get_with(actual_key, || {
|
||||
Arc::new(Mutex::new(FlowData::new(key_for_init, &packet, initiator_direction)))
|
||||
});
|
||||
entry.lock().add_packet(&packet);
|
||||
}
|
||||
|
||||
/// Get all active flows (clone, no drain). Used by WebSocket.
|
||||
pub fn get_flows(&self) -> Vec<FlowData> {
|
||||
self.active.iter().map(|(_, flow)| flow.clone()).collect()
|
||||
self.active.iter().map(|(_, entry)| entry.lock().clone()).collect()
|
||||
}
|
||||
|
||||
/// Get flows that received new packets since their last inference,
|
||||
/// and mark them as inferred. Used by ML engine.
|
||||
pub fn get_uninferred_flows(&mut self) -> Vec<FlowData> {
|
||||
/// Get flows that received new packets since their last inference, and
|
||||
/// mark them as inferred. Used by ML engine.
|
||||
pub fn get_uninferred_flows(&self) -> Vec<FlowData> {
|
||||
let mut result = Vec::new();
|
||||
// iter_mut does NOT promote entries (preserves LRU order)
|
||||
for (_, flow) in self.active.iter_mut() {
|
||||
for (_, entry) in self.active.iter() {
|
||||
let mut flow = entry.lock();
|
||||
if flow.last_time_us > flow.last_inferred_us {
|
||||
result.push(flow.clone());
|
||||
let snapshot = flow.clone();
|
||||
flow.last_inferred_us = flow.last_time_us;
|
||||
result.push(snapshot);
|
||||
}
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub fn flow_count(&self) -> usize {
|
||||
self.active.len()
|
||||
self.active.entry_count() as usize
|
||||
}
|
||||
|
||||
/// Remove flows that have been idle too long or are terminated (FIN/RST seen).
|
||||
/// `now_us`: current timestamp in microseconds (same scale as packet timestamps).
|
||||
/// Returns the number of flows removed.
|
||||
pub fn cleanup_stale_flows(&mut self, now_us: u64) -> usize {
|
||||
// LruCache doesn't have retain(), so collect keys to remove then pop them.
|
||||
let keys_to_remove: Vec<FlowKey> = self
|
||||
.active
|
||||
.iter()
|
||||
.filter(|(_, flow)| {
|
||||
let idle = now_us.saturating_sub(flow.last_time_us);
|
||||
let is_terminated = flow.fin_count > 0 || flow.rst_count > 0;
|
||||
if is_terminated {
|
||||
idle >= FLOW_TERMINATED_TIMEOUT_US
|
||||
} else {
|
||||
idle >= FLOW_IDLE_TIMEOUT_US
|
||||
}
|
||||
})
|
||||
.map(|(k, _)| k.clone())
|
||||
.collect();
|
||||
let removed = keys_to_remove.len();
|
||||
pub fn cleanup_stale_flows(&self, now_us: u64) -> usize {
|
||||
let mut keys_to_remove = Vec::new();
|
||||
for (key, entry) in self.active.iter() {
|
||||
let flow = entry.lock();
|
||||
let idle = now_us.saturating_sub(flow.last_time_us);
|
||||
let is_terminated = flow.fin_count > 0 || flow.rst_count > 0;
|
||||
let stale = if is_terminated {
|
||||
idle >= FLOW_TERMINATED_TIMEOUT_US
|
||||
} else {
|
||||
idle >= FLOW_IDLE_TIMEOUT_US
|
||||
};
|
||||
if stale {
|
||||
keys_to_remove.push((*key).clone());
|
||||
}
|
||||
}
|
||||
let mut removed = 0;
|
||||
for key in keys_to_remove {
|
||||
self.active.pop(&key);
|
||||
self.active.invalidate(&key);
|
||||
removed += 1;
|
||||
}
|
||||
removed
|
||||
}
|
||||
@ -357,74 +364,71 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn sync_count(tracker: &FlowTracker) -> usize {
|
||||
tracker.active.run_pending_tasks();
|
||||
tracker.flow_count()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cleanup_removes_idle_flows() {
|
||||
let mut tracker = FlowTracker::new(10000);
|
||||
let tracker = FlowTracker::new(10000);
|
||||
let base_ts = 1_000_000_000u64; // 1000 seconds
|
||||
|
||||
// Insert a flow with old timestamp
|
||||
let pkt = make_packet(base_ts, 0x02); // SYN
|
||||
tracker.process_packet(pkt, false);
|
||||
assert_eq!(tracker.flow_count(), 1);
|
||||
assert_eq!(sync_count(&tracker), 1);
|
||||
|
||||
// 130 seconds later — should be cleaned up (idle > 120s)
|
||||
let now = base_ts + 130_000_000;
|
||||
let removed = tracker.cleanup_stale_flows(now);
|
||||
assert_eq!(removed, 1);
|
||||
assert_eq!(tracker.flow_count(), 0);
|
||||
assert_eq!(sync_count(&tracker), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cleanup_keeps_active_flows() {
|
||||
let mut tracker = FlowTracker::new(10000);
|
||||
let tracker = FlowTracker::new(10000);
|
||||
let base_ts = 1_000_000_000u64;
|
||||
|
||||
let pkt = make_packet(base_ts, 0x02);
|
||||
tracker.process_packet(pkt, false);
|
||||
|
||||
// Only 10 seconds later — should NOT be cleaned up
|
||||
let now = base_ts + 10_000_000;
|
||||
let removed = tracker.cleanup_stale_flows(now);
|
||||
assert_eq!(removed, 0);
|
||||
assert_eq!(tracker.flow_count(), 1);
|
||||
assert_eq!(sync_count(&tracker), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cleanup_removes_terminated_flows_after_short_idle() {
|
||||
let mut tracker = FlowTracker::new(10000);
|
||||
let tracker = FlowTracker::new(10000);
|
||||
let base_ts = 1_000_000_000u64;
|
||||
|
||||
// SYN packet
|
||||
let pkt1 = make_packet(base_ts, 0x02);
|
||||
tracker.process_packet(pkt1, false);
|
||||
|
||||
// FIN packet 1 second later
|
||||
let pkt2 = make_packet(base_ts + 1_000_000, 0x01); // FIN
|
||||
tracker.process_packet(pkt2, false);
|
||||
|
||||
// 6 seconds after FIN — terminated flow should be removed (idle > 5s)
|
||||
let now = base_ts + 7_000_000;
|
||||
let removed = tracker.cleanup_stale_flows(now);
|
||||
assert_eq!(removed, 1);
|
||||
assert_eq!(tracker.flow_count(), 0);
|
||||
assert_eq!(sync_count(&tracker), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cleanup_keeps_recently_terminated_flows() {
|
||||
let mut tracker = FlowTracker::new(10000);
|
||||
let tracker = FlowTracker::new(10000);
|
||||
let base_ts = 1_000_000_000u64;
|
||||
|
||||
let pkt1 = make_packet(base_ts, 0x02);
|
||||
tracker.process_packet(pkt1, false);
|
||||
|
||||
// FIN packet
|
||||
let pkt2 = make_packet(base_ts + 1_000_000, 0x01);
|
||||
tracker.process_packet(pkt2, false);
|
||||
|
||||
// Only 2 seconds after FIN — should still be around
|
||||
let now = base_ts + 3_000_000;
|
||||
let removed = tracker.cleanup_stale_flows(now);
|
||||
assert_eq!(removed, 0);
|
||||
assert_eq!(tracker.flow_count(), 1);
|
||||
assert_eq!(sync_count(&tracker), 1);
|
||||
}
|
||||
}
|
||||
|
||||
@ -231,7 +231,7 @@ impl PlaybookService {
|
||||
let source_ip = &block.1;
|
||||
|
||||
// Remove from eBPF ACL
|
||||
self.access_control.unblock_ip(source_ip).await?;
|
||||
self.access_control.unblock_ip(source_ip)?;
|
||||
|
||||
// Atomically drop acl_rules entry AND mark soar_block_rules unblocked
|
||||
// in one transaction (R2 mitigation, tx-3 per M2_CARVE_PLAN §4b).
|
||||
|
||||
@ -166,7 +166,7 @@ impl SoarEngine {
|
||||
}
|
||||
|
||||
// Block IP via AccessControlPort (handles IPv4/IPv6 dispatch internally)
|
||||
if let Err(e) = self.access_control.block_ip(&event.source_ip).await {
|
||||
if let Err(e) = self.access_control.block_ip(&event.source_ip) {
|
||||
self.decrement_block_count();
|
||||
return Err(e);
|
||||
}
|
||||
@ -183,7 +183,7 @@ impl SoarEngine {
|
||||
.commit_soar_block_to_db(&event.source_ip, ip_version, playbook_id, &expires_str)
|
||||
{
|
||||
// DB tx rolled back both rows; now roll back the eBPF block.
|
||||
if let Err(unblock_err) = self.access_control.unblock_ip(&event.source_ip).await {
|
||||
if let Err(unblock_err) = self.access_control.unblock_ip(&event.source_ip) {
|
||||
log!(SoarLog::EventHandlingFailed(format!(
|
||||
"CRITICAL: Failed to unblock IP {} after DB error — queueing for retry: {}",
|
||||
event.source_ip, unblock_err
|
||||
@ -211,7 +211,7 @@ impl SoarEngine {
|
||||
action: &PlaybookAction,
|
||||
event: &ThreatDetectedEvent,
|
||||
) -> Result<String, Error> {
|
||||
let rate_limit = self.rate_limit.as_ref().ok_or(SoarError::RateLimitUnavailable)?;
|
||||
let owner = self.rate_limit.as_ref().ok_or(SoarError::RateLimitUnavailable)?;
|
||||
|
||||
let factor = action.params.get("factor").and_then(|v| v.as_f64()).unwrap_or(0.5);
|
||||
let ttl_secs = action.params.get("ttl_secs").and_then(|v| v.as_u64()).unwrap_or(600);
|
||||
@ -231,67 +231,9 @@ impl SoarEngine {
|
||||
Err(SoarError::InvalidTtl(ttl_secs, max_ttl))?;
|
||||
}
|
||||
|
||||
// Acquire lock to serialize rate limit read-save-write (Item 6: atomicity)
|
||||
let _guard = self.rate_limit_lock.lock().await;
|
||||
|
||||
// Read current rates, save originals, apply reduced rates
|
||||
let current_packet = rate_limit.get_packet_rate().unwrap_or(10000);
|
||||
let current_syn = rate_limit.get_syn_rate().unwrap_or(1000);
|
||||
let current_udp = rate_limit.get_udp_rate().unwrap_or(5000);
|
||||
let current_dns = rate_limit.get_dns_rate().unwrap_or(2000);
|
||||
|
||||
// Store original rates for restoration (only if not already adjusted)
|
||||
let key = "soar_rate_limit_original";
|
||||
if self.db.get_setting(key)?.filter(|s| !s.is_empty()).is_none() {
|
||||
let original = serde_json::json!({
|
||||
"packet_rate": current_packet,
|
||||
"syn_rate": current_syn,
|
||||
"udp_rate": current_udp,
|
||||
"dns_rate": current_dns,
|
||||
});
|
||||
self.db.set_setting(key, &original.to_string())?;
|
||||
}
|
||||
|
||||
// Store TTL for restoration
|
||||
let expires_at = Utc::now() + ChronoDuration::seconds(ttl_secs as i64);
|
||||
self.db.set_setting(
|
||||
"soar_rate_limit_expires",
|
||||
&expires_at.format("%Y-%m-%d %H:%M:%S").to_string(),
|
||||
)?;
|
||||
|
||||
// Apply reduced rates
|
||||
let new_packet = (current_packet as f64 * factor) as u64;
|
||||
let new_syn = (current_syn as f64 * factor) as u64;
|
||||
let new_udp = (current_udp as f64 * factor) as u64;
|
||||
let new_dns = (current_dns as f64 * factor) as u64;
|
||||
|
||||
rate_limit.set_packet_rate(new_packet.max(1))?;
|
||||
rate_limit.set_syn_rate(new_syn.max(1))?;
|
||||
rate_limit.set_udp_rate(new_udp.max(1))?;
|
||||
rate_limit.set_dns_rate(new_dns.max(1))?;
|
||||
|
||||
log!(SoarLog::RateLimitAdjusted(
|
||||
format!("{}", factor),
|
||||
ttl_secs,
|
||||
event.source_ip.clone(),
|
||||
event.attack_type.clone(),
|
||||
format!(
|
||||
"packet {}→{}, syn {}→{}, udp {}→{}, dns {}→{}",
|
||||
current_packet,
|
||||
new_packet.max(1),
|
||||
current_syn,
|
||||
new_syn.max(1),
|
||||
current_udp,
|
||||
new_udp.max(1),
|
||||
current_dns,
|
||||
new_dns.max(1),
|
||||
),
|
||||
));
|
||||
|
||||
Ok(format!(
|
||||
"Rate limits reduced by factor {} for {}s (triggered by {})",
|
||||
factor, ttl_secs, event.source_ip
|
||||
))
|
||||
owner
|
||||
.adjust(factor, ttl_secs, event.source_ip.clone(), event.attack_type.clone())
|
||||
.await
|
||||
}
|
||||
|
||||
/// Send Telegram notification.
|
||||
|
||||
@ -4,14 +4,14 @@ 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 serde_json::Value;
|
||||
use tokio::sync::broadcast;
|
||||
use tokio::sync::broadcast::error::RecvError;
|
||||
use tokio::sync::{Mutex as TokioMutex, broadcast};
|
||||
|
||||
use crate::core::soar::frequency::FrequencyTracker;
|
||||
use crate::core::soar::rate_limit_owner::RateLimitOwnerHandle;
|
||||
use crate::infrastructure::communication_manager::CommunicationManager;
|
||||
use crate::infrastructure::geoip::GeoIpService;
|
||||
use crate::interface::port::access_control::AccessControlPort;
|
||||
@ -58,12 +58,11 @@ pub struct SoarEngine {
|
||||
pub(super) alert_notifier: Option<Arc<dyn AlertNotifier>>,
|
||||
/// Optional GeoIP service for country lookups.
|
||||
pub(super) geoip: Option<Arc<GeoIpService>>,
|
||||
/// Optional rate limit config for adjust_rate_limit action.
|
||||
pub(super) rate_limit: Option<Arc<dyn RateLimitPort>>,
|
||||
/// Serializes rate-limit read-save-write sequences so concurrent SOAR
|
||||
/// actions can't race the `soar_rate_limit_*` DB settings into an
|
||||
/// inconsistent pair.
|
||||
pub(super) rate_limit_lock: TokioMutex<()>,
|
||||
/// Owner-task handle that serializes the rate-limit DB+eBPF
|
||||
/// read-modify-write batch. `None` when no `RateLimitPort` was wired
|
||||
/// up (eBPF unavailable); SOAR actions that need rate-limit then
|
||||
/// fail with `SoarError::RateLimitUnavailable`.
|
||||
pub(super) rate_limit: Option<RateLimitOwnerHandle>,
|
||||
/// Cached enforce level: Monitor=0, MlOnly=1, Enforce=2.
|
||||
pub(super) enforce_level_cache: Arc<AtomicU8>,
|
||||
/// Secret store for decrypting SMTP passwords etc.
|
||||
@ -80,6 +79,7 @@ impl SoarEngine {
|
||||
enforce_level_cache: Arc<AtomicU8>,
|
||||
secrets: Option<Arc<dyn SecretStorePort>>,
|
||||
) -> Result<Self, Error> {
|
||||
let rate_limit_owner = rate_limit.map(|rl| RateLimitOwnerHandle::spawn(db.clone(), rl));
|
||||
let engine = Self {
|
||||
db,
|
||||
access_control,
|
||||
@ -90,8 +90,7 @@ impl SoarEngine {
|
||||
active_block_count: AtomicU32::new(0),
|
||||
alert_notifier,
|
||||
geoip,
|
||||
rate_limit,
|
||||
rate_limit_lock: TokioMutex::new(()),
|
||||
rate_limit: rate_limit_owner,
|
||||
enforce_level_cache,
|
||||
secrets,
|
||||
};
|
||||
@ -290,7 +289,7 @@ impl SoarEngine {
|
||||
|
||||
for (_id, source_ip, _playbook_id, _expires_at) in &active_blocks {
|
||||
// Preserve original error-swallowing behavior during recovery
|
||||
if let Err(e) = self.access_control.block_ip(source_ip).await {
|
||||
if let Err(e) = self.access_control.block_ip(source_ip) {
|
||||
log!(SoarLog::RecoveryFailed(source_ip.clone(), e.to_string()));
|
||||
}
|
||||
}
|
||||
@ -326,7 +325,7 @@ impl SoarEngine {
|
||||
continue;
|
||||
}
|
||||
|
||||
match self.access_control.unblock_ip(&source_ip).await {
|
||||
match self.access_control.unblock_ip(&source_ip) {
|
||||
Ok(()) => {
|
||||
let _ = self.db.delete_pending_unblock(id);
|
||||
log!(SoarLog::EventHandlingFailed(format!(
|
||||
@ -351,76 +350,10 @@ impl SoarEngine {
|
||||
/// Restore original rate limits if the TTL has expired.
|
||||
/// Called by TTL scheduler on each sweep.
|
||||
pub async fn check_rate_limit_restoration(&self) -> Result<(), Error> {
|
||||
let _guard = self.rate_limit_lock.lock().await;
|
||||
|
||||
let expires_str = match self
|
||||
.db
|
||||
.get_setting("soar_rate_limit_expires")?
|
||||
.filter(|s| !s.is_empty())
|
||||
{
|
||||
Some(s) => s,
|
||||
None => return Ok(()), // No active adjustment
|
||||
};
|
||||
|
||||
let expires = NaiveDateTime::parse_from_str(&expires_str, "%Y-%m-%d %H:%M:%S")
|
||||
.map(|dt| dt.and_utc())
|
||||
.unwrap_or_else(|_| Utc::now());
|
||||
|
||||
if Utc::now() < expires {
|
||||
return Ok(()); // Not yet expired
|
||||
match &self.rate_limit {
|
||||
Some(owner) => owner.restore_if_expired().await,
|
||||
None => Ok(()),
|
||||
}
|
||||
|
||||
// Restore original rates
|
||||
let original_str = match self
|
||||
.db
|
||||
.get_setting("soar_rate_limit_original")?
|
||||
.filter(|s| !s.is_empty())
|
||||
{
|
||||
Some(s) => s,
|
||||
None => {
|
||||
// No originals saved, just clean up
|
||||
self.db.set_setting("soar_rate_limit_expires", "")?;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
if let (Some(rate_limit), Ok(original)) = (
|
||||
&self.rate_limit,
|
||||
serde_json::from_str::<serde_json::Value>(&original_str),
|
||||
) {
|
||||
let mut restore_errors = Vec::new();
|
||||
if let Some(v) = original.get("packet_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_packet_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("packet_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("syn_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_syn_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("syn_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("udp_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_udp_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("udp_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("dns_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_dns_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("dns_rate: {}", e));
|
||||
}
|
||||
if restore_errors.is_empty() {
|
||||
log!(SoarLog::RateLimitRestored);
|
||||
} else {
|
||||
log!(SoarLog::RateLimitRestoreFailed(restore_errors.join(", ")));
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up settings
|
||||
self.db.set_setting("soar_rate_limit_original", "")?;
|
||||
self.db.set_setting("soar_rate_limit_expires", "")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@ -429,7 +362,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::model::error::ebpf::EbpfError;
|
||||
use crate::model::event::DetectionSource;
|
||||
use chrono::Duration as ChronoDuration;
|
||||
use chrono::{Duration as ChronoDuration, Utc};
|
||||
use parking_lot::Mutex;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
|
||||
@ -450,9 +383,8 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl AccessControlPort for MockAccessControl {
|
||||
async fn block_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
fn block_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
if self.should_fail.load(Ordering::SeqCst) {
|
||||
Err(EbpfError::UnknownError)?;
|
||||
}
|
||||
@ -460,7 +392,7 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn unblock_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
fn unblock_ip(&self, ip: &str) -> Result<(), Error> {
|
||||
if self.should_fail.load(Ordering::SeqCst) {
|
||||
Err(EbpfError::UnknownError)?;
|
||||
}
|
||||
|
||||
@ -2,4 +2,5 @@ pub mod actions;
|
||||
pub mod engine;
|
||||
pub mod frequency;
|
||||
pub mod matcher;
|
||||
pub mod rate_limit_owner;
|
||||
pub mod scheduler;
|
||||
|
||||
226
net-guardia/src/core/soar/rate_limit_owner.rs
Normal file
226
net-guardia/src/core/soar/rate_limit_owner.rs
Normal file
@ -0,0 +1,226 @@
|
||||
//! Owner task that serializes SOAR rate-limit adjustments.
|
||||
//!
|
||||
//! The "adjust" and "restore" sequences each touch two systems back-to-back
|
||||
//! (the `soar_rate_limit_*` settings rows and the eBPF `RATE_LIMIT_CONFIG`
|
||||
//! map). The atomicity that the original `TokioMutex<()>` was protecting is
|
||||
//! exactly "no other adjust/restore interleaves between the read and the
|
||||
//! write" — a SQLite transaction can't cover the eBPF half, so we move the
|
||||
//! read-modify-write inside a single tokio task that owns both ports. All
|
||||
//! callers dispatch over an mpsc channel and wait on a one-shot reply.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::{Duration as ChronoDuration, NaiveDateTime, Utc};
|
||||
use macros::log;
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
use crate::interface::port::app_repo::AppRepo;
|
||||
use crate::interface::port::rate_limit_api::RateLimitPort;
|
||||
use crate::model::error::Error;
|
||||
use crate::model::error::soar::SoarError;
|
||||
use crate::model::log::soar::SoarLog;
|
||||
|
||||
/// Channel depth for the owner-task command queue. SOAR rate-limit operations
|
||||
/// are bursty but rare (operator action / playbook trigger), so 64 is plenty.
|
||||
const RATE_LIMIT_CMD_CHANNEL_CAPACITY: usize = 64;
|
||||
|
||||
/// Settings keys persisted across restarts so the next process can resume the
|
||||
/// same TTL window. Stable wire format with the DB.
|
||||
const KEY_ORIGINAL: &str = "soar_rate_limit_original";
|
||||
const KEY_EXPIRES: &str = "soar_rate_limit_expires";
|
||||
|
||||
enum RateLimitCmd {
|
||||
Adjust {
|
||||
factor: f64,
|
||||
ttl_secs: u64,
|
||||
source_ip: String,
|
||||
attack_type: String,
|
||||
reply: oneshot::Sender<Result<String, Error>>,
|
||||
},
|
||||
RestoreIfExpired {
|
||||
reply: oneshot::Sender<Result<(), Error>>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Lock-free handle to the rate-limit owner task. Cloning is cheap (just an
|
||||
/// `mpsc::Sender`).
|
||||
#[derive(Clone)]
|
||||
pub struct RateLimitOwnerHandle {
|
||||
tx: mpsc::Sender<RateLimitCmd>,
|
||||
}
|
||||
|
||||
impl RateLimitOwnerHandle {
|
||||
/// Spawn the owner task on the current tokio runtime. The owner holds
|
||||
/// the only mutating references to the DB rate-limit settings and the
|
||||
/// eBPF rate-limit map for the duration of an Adjust / Restore batch.
|
||||
pub fn spawn(db: Arc<dyn AppRepo>, rate_limit: Arc<dyn RateLimitPort>) -> Self {
|
||||
let (tx, mut rx) = mpsc::channel::<RateLimitCmd>(RATE_LIMIT_CMD_CHANNEL_CAPACITY);
|
||||
tokio::spawn(async move {
|
||||
while let Some(cmd) = rx.recv().await {
|
||||
match cmd {
|
||||
RateLimitCmd::Adjust {
|
||||
factor,
|
||||
ttl_secs,
|
||||
source_ip,
|
||||
attack_type,
|
||||
reply,
|
||||
} => {
|
||||
let result = adjust(&db, rate_limit.as_ref(), factor, ttl_secs, &source_ip, &attack_type);
|
||||
let _ = reply.send(result);
|
||||
}
|
||||
RateLimitCmd::RestoreIfExpired { reply } => {
|
||||
let result = restore_if_expired(&db, rate_limit.as_ref());
|
||||
let _ = reply.send(result);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
Self { tx }
|
||||
}
|
||||
|
||||
pub async fn adjust(
|
||||
&self,
|
||||
factor: f64,
|
||||
ttl_secs: u64,
|
||||
source_ip: String,
|
||||
attack_type: String,
|
||||
) -> Result<String, Error> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
self.tx
|
||||
.send(RateLimitCmd::Adjust {
|
||||
factor,
|
||||
ttl_secs,
|
||||
source_ip,
|
||||
attack_type,
|
||||
reply: reply_tx,
|
||||
})
|
||||
.await
|
||||
.map_err(|_| SoarError::RateLimitOwnerUnavailable)?;
|
||||
reply_rx.await.map_err(|_| SoarError::RateLimitOwnerUnavailable)?
|
||||
}
|
||||
|
||||
pub async fn restore_if_expired(&self) -> Result<(), Error> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
self.tx
|
||||
.send(RateLimitCmd::RestoreIfExpired { reply: reply_tx })
|
||||
.await
|
||||
.map_err(|_| SoarError::RateLimitOwnerUnavailable)?;
|
||||
reply_rx.await.map_err(|_| SoarError::RateLimitOwnerUnavailable)?
|
||||
}
|
||||
}
|
||||
|
||||
fn adjust(
|
||||
db: &Arc<dyn AppRepo>,
|
||||
rate_limit: &dyn RateLimitPort,
|
||||
factor: f64,
|
||||
ttl_secs: u64,
|
||||
source_ip: &str,
|
||||
attack_type: &str,
|
||||
) -> Result<String, Error> {
|
||||
let current_packet = rate_limit.get_packet_rate().unwrap_or(10000);
|
||||
let current_syn = rate_limit.get_syn_rate().unwrap_or(1000);
|
||||
let current_udp = rate_limit.get_udp_rate().unwrap_or(5000);
|
||||
let current_dns = rate_limit.get_dns_rate().unwrap_or(2000);
|
||||
|
||||
if db.get_setting(KEY_ORIGINAL)?.filter(|s| !s.is_empty()).is_none() {
|
||||
let original = serde_json::json!({
|
||||
"packet_rate": current_packet,
|
||||
"syn_rate": current_syn,
|
||||
"udp_rate": current_udp,
|
||||
"dns_rate": current_dns,
|
||||
});
|
||||
db.set_setting(KEY_ORIGINAL, &original.to_string())?;
|
||||
}
|
||||
|
||||
let expires_at = Utc::now() + ChronoDuration::seconds(ttl_secs as i64);
|
||||
db.set_setting(KEY_EXPIRES, &expires_at.format("%Y-%m-%d %H:%M:%S").to_string())?;
|
||||
|
||||
let new_packet = (current_packet as f64 * factor) as u64;
|
||||
let new_syn = (current_syn as f64 * factor) as u64;
|
||||
let new_udp = (current_udp as f64 * factor) as u64;
|
||||
let new_dns = (current_dns as f64 * factor) as u64;
|
||||
|
||||
rate_limit.set_packet_rate(new_packet.max(1))?;
|
||||
rate_limit.set_syn_rate(new_syn.max(1))?;
|
||||
rate_limit.set_udp_rate(new_udp.max(1))?;
|
||||
rate_limit.set_dns_rate(new_dns.max(1))?;
|
||||
|
||||
log!(SoarLog::RateLimitAdjusted(
|
||||
format!("{}", factor),
|
||||
ttl_secs,
|
||||
source_ip.to_string(),
|
||||
attack_type.to_string(),
|
||||
format!(
|
||||
"packet {}→{}, syn {}→{}, udp {}→{}, dns {}→{}",
|
||||
current_packet,
|
||||
new_packet.max(1),
|
||||
current_syn,
|
||||
new_syn.max(1),
|
||||
current_udp,
|
||||
new_udp.max(1),
|
||||
current_dns,
|
||||
new_dns.max(1),
|
||||
),
|
||||
));
|
||||
|
||||
Ok(format!(
|
||||
"Rate limits reduced by factor {} for {}s (triggered by {})",
|
||||
factor, ttl_secs, source_ip
|
||||
))
|
||||
}
|
||||
|
||||
fn restore_if_expired(db: &Arc<dyn AppRepo>, rate_limit: &dyn RateLimitPort) -> Result<(), Error> {
|
||||
let expires_str = match db.get_setting(KEY_EXPIRES)?.filter(|s| !s.is_empty()) {
|
||||
Some(s) => s,
|
||||
None => return Ok(()),
|
||||
};
|
||||
|
||||
let expires = NaiveDateTime::parse_from_str(&expires_str, "%Y-%m-%d %H:%M:%S")
|
||||
.map(|dt| dt.and_utc())
|
||||
.unwrap_or_else(|_| Utc::now());
|
||||
|
||||
if Utc::now() < expires {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let original_str = match db.get_setting(KEY_ORIGINAL)?.filter(|s| !s.is_empty()) {
|
||||
Some(s) => s,
|
||||
None => {
|
||||
db.set_setting(KEY_EXPIRES, "")?;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
if let Ok(original) = serde_json::from_str::<serde_json::Value>(&original_str) {
|
||||
let mut restore_errors = Vec::new();
|
||||
if let Some(v) = original.get("packet_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_packet_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("packet_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("syn_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_syn_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("syn_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("udp_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_udp_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("udp_rate: {}", e));
|
||||
}
|
||||
if let Some(v) = original.get("dns_rate").and_then(|v| v.as_u64())
|
||||
&& let Err(e) = rate_limit.set_dns_rate(v)
|
||||
{
|
||||
restore_errors.push(format!("dns_rate: {}", e));
|
||||
}
|
||||
if restore_errors.is_empty() {
|
||||
log!(SoarLog::RateLimitRestored);
|
||||
} else {
|
||||
log!(SoarLog::RateLimitRestoreFailed(restore_errors.join(", ")));
|
||||
}
|
||||
}
|
||||
|
||||
db.set_setting(KEY_ORIGINAL, "")?;
|
||||
db.set_setting(KEY_EXPIRES, "")?;
|
||||
Ok(())
|
||||
}
|
||||
@ -82,7 +82,7 @@ impl TtlScheduler {
|
||||
}
|
||||
|
||||
// Remove from eBPF ACL via AccessControlPort
|
||||
if let Err(e) = self.access_control.unblock_ip(source_ip).await {
|
||||
if let Err(e) = self.access_control.unblock_ip(source_ip) {
|
||||
log!(SoarLog::RecoveryFailed(
|
||||
source_ip.clone(),
|
||||
format!("unblock failed: {}", e)
|
||||
|
||||
@ -10,7 +10,7 @@ use tokio::sync::oneshot;
|
||||
use crate::core::detection::metrics::FusionMetrics;
|
||||
use crate::core::ml::adapter::ModelSourceState;
|
||||
use crate::core::ml::alert::MLAlert;
|
||||
use crate::core::ml::drift_detector::DriftDetector;
|
||||
use crate::core::ml::drift_detector::DriftDetectorHandle;
|
||||
use crate::core::ml::engine::Engine;
|
||||
use crate::core::ml::inference::Inference;
|
||||
use crate::core::ml::manifest::ModelManifest;
|
||||
@ -50,7 +50,7 @@ impl AppServices {
|
||||
app_config: Arc<AppConfig>,
|
||||
inference_config: Arc<MLInferenceConfig>,
|
||||
ml_manifest: Option<ModelManifest>,
|
||||
drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
|
||||
drift_detector: DriftDetectorHandle,
|
||||
ebpf_health: Arc<ArcSwap<EbpfHealth>>,
|
||||
comm: Arc<CommunicationManager>,
|
||||
) -> Result<Self, Error> {
|
||||
|
||||
@ -5,7 +5,7 @@ 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};
|
||||
use tokio::sync::{broadcast, oneshot};
|
||||
use tokio::time::interval;
|
||||
|
||||
use crate::infrastructure::app_config::AppConfig;
|
||||
@ -16,10 +16,16 @@ use crate::model::system::health::{
|
||||
SystemHealthMetrics, SystemHealthStatus, SystemInfo,
|
||||
};
|
||||
|
||||
/// Lock-free system health.
|
||||
///
|
||||
/// A single owner task on the tokio runtime owns the sysinfo handles
|
||||
/// (`System`, `Networks`, `Components`). It refreshes them on a tick,
|
||||
/// computes a fresh `SystemHealthMetrics`, publishes the snapshot via
|
||||
/// `ArcSwap`, and broadcasts to streaming subscribers. Readers
|
||||
/// (`get_current_metrics`, `is_system_healthy`, HTTP handlers) just
|
||||
/// `.load()` the `ArcSwap` — no locks crossed, no `await` needed.
|
||||
pub struct SystemHealth {
|
||||
system: RwLock<System>,
|
||||
networks: RwLock<Networks>,
|
||||
components: RwLock<Components>,
|
||||
metrics: Arc<ArcSwap<SystemHealthMetrics>>,
|
||||
broadcast_tx: broadcast::Sender<SystemHealthMetrics>,
|
||||
ingress_interface: String,
|
||||
egress_interface: String,
|
||||
@ -29,35 +35,74 @@ pub struct SystemHealth {
|
||||
impl SystemHealth {
|
||||
pub fn new(config: Arc<AppConfig>, ebpf_health: Arc<ArcSwap<EbpfHealth>>) -> Result<Self, Error> {
|
||||
let (broadcast_tx, _) = broadcast::channel(100);
|
||||
let ingress_interface = config.network.ingress_ifname.clone();
|
||||
let egress_interface = config.network.egress_ifname.clone();
|
||||
|
||||
let health = SystemHealth {
|
||||
system: RwLock::new(System::new_all()),
|
||||
networks: RwLock::new(Networks::new_with_refreshed_list()),
|
||||
components: RwLock::new(Components::new_with_refreshed_list()),
|
||||
// Bootstrap snapshot so readers don't have to handle a "no metrics yet"
|
||||
// case before the refresh task fires for the first time. The
|
||||
// `*_with_refreshed_list` constructors already do an initial refresh.
|
||||
let system = System::new_all();
|
||||
let networks = Networks::new_with_refreshed_list();
|
||||
let components = Components::new_with_refreshed_list();
|
||||
let initial = Self::collect_metrics(
|
||||
&system,
|
||||
&networks,
|
||||
&components,
|
||||
&ingress_interface,
|
||||
&egress_interface,
|
||||
(**ebpf_health.load()).clone(),
|
||||
);
|
||||
|
||||
Ok(SystemHealth {
|
||||
metrics: Arc::new(ArcSwap::from_pointee(initial)),
|
||||
broadcast_tx,
|
||||
ingress_interface: config.network.ingress_ifname.clone(),
|
||||
egress_interface: config.network.egress_ifname.clone(),
|
||||
ingress_interface,
|
||||
egress_interface,
|
||||
ebpf_health,
|
||||
};
|
||||
|
||||
Ok(health)
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn run(self: Arc<Self>, monitoring_interval: Duration) -> oneshot::Sender<()> {
|
||||
let (sender, mut receiver) = oneshot::channel();
|
||||
let health = self.clone();
|
||||
let metrics = self.metrics.clone();
|
||||
let broadcast_tx = self.broadcast_tx.clone();
|
||||
let ingress_interface = self.ingress_interface.clone();
|
||||
let egress_interface = self.egress_interface.clone();
|
||||
let ebpf_health = self.ebpf_health.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
// Owner task exclusively holds these sysinfo handles, so no locks
|
||||
// are needed on the data plane.
|
||||
let mut system = System::new_all();
|
||||
let mut networks = Networks::new_with_refreshed_list();
|
||||
let mut components = Components::new_with_refreshed_list();
|
||||
let mut interval_timer = interval(monitoring_interval);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = &mut receiver => {
|
||||
break;
|
||||
}
|
||||
_ = &mut receiver => break,
|
||||
_ = interval_timer.tick() => {
|
||||
health.refresh_and_broadcast().await;
|
||||
system.refresh_all();
|
||||
networks.refresh(true);
|
||||
components.refresh(true);
|
||||
|
||||
let snapshot = Self::collect_metrics(
|
||||
&system,
|
||||
&networks,
|
||||
&components,
|
||||
&ingress_interface,
|
||||
&egress_interface,
|
||||
(**ebpf_health.load()).clone(),
|
||||
);
|
||||
|
||||
metrics.store(Arc::new(snapshot.clone()));
|
||||
|
||||
if broadcast_tx.receiver_count() > 0
|
||||
&& let Err(e) = broadcast_tx.send(snapshot)
|
||||
{
|
||||
log!(Health::BroadcastFailed(e.to_string()));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -66,36 +111,6 @@ impl SystemHealth {
|
||||
sender
|
||||
}
|
||||
|
||||
async fn refresh_and_broadcast(&self) {
|
||||
self.system.write().await.refresh_all();
|
||||
self.networks.write().await.refresh(true);
|
||||
self.components.write().await.refresh(true);
|
||||
|
||||
let system = self.system.read().await;
|
||||
let networks = self.networks.read().await;
|
||||
let components = self.components.read().await;
|
||||
|
||||
let ebpf = (**self.ebpf_health.load()).clone();
|
||||
let metrics = Self::collect_metrics(
|
||||
&system,
|
||||
&networks,
|
||||
&components,
|
||||
&self.ingress_interface,
|
||||
&self.egress_interface,
|
||||
ebpf,
|
||||
);
|
||||
|
||||
drop(system);
|
||||
drop(networks);
|
||||
drop(components);
|
||||
|
||||
if self.broadcast_tx.receiver_count() > 0
|
||||
&& let Err(e) = self.broadcast_tx.send(metrics)
|
||||
{
|
||||
log!(Health::BroadcastFailed(e.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
fn collect_metrics(
|
||||
system: &System,
|
||||
networks: &Networks,
|
||||
@ -239,20 +254,8 @@ impl SystemHealth {
|
||||
ConfiguredNetworkStats { ingress, egress }
|
||||
}
|
||||
|
||||
pub async fn get_current_metrics(&self) -> SystemHealthMetrics {
|
||||
let system = self.system.read().await;
|
||||
let networks = self.networks.read().await;
|
||||
let components = self.components.read().await;
|
||||
|
||||
let ebpf = (**self.ebpf_health.load()).clone();
|
||||
Self::collect_metrics(
|
||||
&system,
|
||||
&networks,
|
||||
&components,
|
||||
&self.ingress_interface,
|
||||
&self.egress_interface,
|
||||
ebpf,
|
||||
)
|
||||
pub fn get_current_metrics(&self) -> SystemHealthMetrics {
|
||||
(**self.metrics.load()).clone()
|
||||
}
|
||||
|
||||
/// Returns a handle to the shared eBPF health state. Consumers (HTTP
|
||||
@ -266,8 +269,8 @@ impl SystemHealth {
|
||||
self.broadcast_tx.subscribe()
|
||||
}
|
||||
|
||||
pub async fn is_system_healthy(&self) -> SystemHealthStatus {
|
||||
let metrics = self.get_current_metrics().await;
|
||||
pub fn is_system_healthy(&self) -> SystemHealthStatus {
|
||||
let metrics = self.get_current_metrics();
|
||||
|
||||
let mut status = SystemHealthStatus {
|
||||
overall_healthy: true,
|
||||
|
||||
@ -9,7 +9,7 @@ use actix_web::{App, HttpResponse, HttpServer, web};
|
||||
use macros::log;
|
||||
|
||||
use crate::adapter::ebpf::EbpfServices;
|
||||
use crate::adapter::http::model_upload::PromoteLock;
|
||||
use crate::adapter::http::model_upload::PromoteGate;
|
||||
use crate::adapter::http::{
|
||||
acl, api_keys, audit as audit_api, auth, byo, default, filter, flow_trace, fusion, health as health_api,
|
||||
logs as logs_api, ml, model_upload, notification as notification_api, rate_limit as rate_limit_api,
|
||||
@ -243,7 +243,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
|
||||
// serialize their rename-into-`models/` critical section. Built
|
||||
// here rather than threaded through HttpServerParams because
|
||||
// nothing outside the HTTP boundary needs to observe it.
|
||||
let promote_lock: Arc<PromoteLock> = Arc::new(PromoteLock::new(()));
|
||||
let promote_lock: Arc<PromoteGate> = Arc::new(PromoteGate::new());
|
||||
|
||||
HttpServer::new(move || {
|
||||
let app = App::new()
|
||||
|
||||
@ -6,7 +6,6 @@ use std::sync::atomic::AtomicU8;
|
||||
use std::time::Duration;
|
||||
|
||||
use arc_swap::ArcSwap;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use aya::Ebpf;
|
||||
use aya::maps::{Array, MapData, ProgramArray};
|
||||
@ -24,7 +23,7 @@ use crate::core::acl_service::AclService;
|
||||
use crate::core::config_service::ConfigService;
|
||||
use crate::core::dns_filter_service::DnsFilterService;
|
||||
use crate::core::email::scheduler::ReportScheduler;
|
||||
use crate::core::ml::drift_detector::DriftDetector;
|
||||
use crate::core::ml::drift_detector::DriftDetectorHandle;
|
||||
use crate::core::ml::manifest::ModelManifest;
|
||||
use crate::core::notification_service::NotificationService;
|
||||
use crate::core::playbook_service::PlaybookService;
|
||||
@ -82,7 +81,7 @@ pub struct AppState {
|
||||
pub playbook_service: Arc<PlaybookService>,
|
||||
pub rate_limit_service: Arc<RateLimitService>,
|
||||
pub geoip: Option<Arc<GeoIpService>>,
|
||||
pub drift_detector: Arc<Mutex<DriftDetector>>,
|
||||
pub drift_detector: DriftDetectorHandle,
|
||||
pub ingress_ebpf: Option<Ebpf>,
|
||||
pub egress_ebpf: Option<Ebpf>,
|
||||
/// Held to keep the eBPF program array map FD alive. `None` when eBPF
|
||||
@ -181,10 +180,7 @@ impl ServiceFactory {
|
||||
.flatten()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(3600);
|
||||
let drift_detector = Arc::new(Mutex::new(DriftDetector::new(
|
||||
baselines,
|
||||
Duration::from_secs(drift_window_secs),
|
||||
)));
|
||||
let drift_detector = DriftDetectorHandle::spawn(baselines, Duration::from_secs(drift_window_secs));
|
||||
|
||||
// Create AtomicU8 enforce-level cache (Monitor=0, MlOnly=1, Enforce=2)
|
||||
let enforce_level_cache = Arc::new(AtomicU8::new({
|
||||
@ -575,12 +571,9 @@ impl ServiceFactory {
|
||||
};
|
||||
let result = match ip_version {
|
||||
4 => match ip_address.parse::<Ipv4Addr>() {
|
||||
Ok(addr) => {
|
||||
ebpf_services
|
||||
.access_control
|
||||
.add_ipv4_list(dir, lt, SocketAddrV4::new(addr, *port))
|
||||
.await
|
||||
}
|
||||
Ok(addr) => ebpf_services
|
||||
.access_control
|
||||
.add_ipv4_list(dir, lt, SocketAddrV4::new(addr, *port)),
|
||||
Err(e) => {
|
||||
log!(SystemLog::AclIpv4ParseFailed(ip_address.clone(), e.to_string()));
|
||||
continue;
|
||||
@ -591,7 +584,6 @@ impl ServiceFactory {
|
||||
ebpf_services
|
||||
.access_control
|
||||
.add_ipv6_list(dir, lt, SocketAddrV6::new(addr, *port, 0, 0))
|
||||
.await
|
||||
}
|
||||
Err(e) => {
|
||||
log!(SystemLog::AclIpv6ParseFailed(ip_address.clone(), e.to_string()));
|
||||
|
||||
@ -42,8 +42,7 @@ impl FlowStatistics {
|
||||
pub fn get_all_flows(&self) -> Vec<FlowStatsEntry> {
|
||||
let mut entries = Vec::new();
|
||||
for tracker in self.engine.trackers() {
|
||||
let t = tracker.lock();
|
||||
entries.extend(t.get_flows().iter().map(FlowStatsEntry::from));
|
||||
entries.extend(tracker.get_flows().iter().map(FlowStatsEntry::from));
|
||||
}
|
||||
entries
|
||||
}
|
||||
|
||||
@ -25,7 +25,7 @@ use crate::core::detection::beaconing::BeaconingDetector;
|
||||
use crate::core::detection::orchestrator::DetectionOrchestrator;
|
||||
use crate::core::dns_filter_service::DnsFilterService;
|
||||
use crate::core::email::scheduler::ReportScheduler;
|
||||
use crate::core::ml::drift_detector::DriftDetector;
|
||||
use crate::core::ml::drift_detector::DriftDetectorHandle;
|
||||
use crate::core::ml::model_watcher::ModelWatcher;
|
||||
use crate::core::notification_service::NotificationService;
|
||||
use crate::core::playbook_service::PlaybookService;
|
||||
@ -109,7 +109,7 @@ pub struct System {
|
||||
pub playbook_service: Arc<PlaybookService>,
|
||||
pub rate_limit_service: Arc<RateLimitService>,
|
||||
pub geoip: Option<Arc<GeoIpService>>,
|
||||
pub drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
|
||||
pub drift_detector: DriftDetectorHandle,
|
||||
pub shutdown_handle: Option<Arc<ShutdownHandle>>,
|
||||
_ingress_program_array: Option<ProgramArray<MapData>>,
|
||||
pub ebpf_health: Arc<ArcSwap<EbpfHealth>>,
|
||||
@ -430,14 +430,11 @@ impl System {
|
||||
}
|
||||
|
||||
/// Periodically check the drift detector and publish DriftDetectedEvent when drift is found.
|
||||
async fn run_drift_monitor(
|
||||
drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
|
||||
comm: Arc<CommunicationManager>,
|
||||
) {
|
||||
async fn run_drift_monitor(drift_detector: DriftDetectorHandle, comm: Arc<CommunicationManager>) {
|
||||
let mut interval = interval(Duration::from_secs(60));
|
||||
loop {
|
||||
interval.tick().await;
|
||||
let report = drift_detector.lock().check_drift();
|
||||
let report = drift_detector.check_drift().await;
|
||||
if let Some(report) = report {
|
||||
log!(SystemLog::DriftDetected(
|
||||
report.drifted_features.len(),
|
||||
|
||||
@ -1,16 +1,13 @@
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::model::error::Error;
|
||||
|
||||
/// Port for blocking/unblocking IP addresses in the network data plane.
|
||||
/// Adapters: EbpfAccessControlAdapter (wraps eBPF AccessControl)
|
||||
#[async_trait]
|
||||
pub trait AccessControlPort: Send + Sync {
|
||||
/// Block an IP address (adds to source blacklist in the data plane).
|
||||
/// Accepts both IPv4 ("1.2.3.4") and IPv6 ("::1") strings.
|
||||
async fn block_ip(&self, ip: &str) -> Result<(), Error>;
|
||||
fn block_ip(&self, ip: &str) -> Result<(), Error>;
|
||||
|
||||
/// Unblock an IP address (removes from source blacklist in the data plane).
|
||||
/// Accepts both IPv4 and IPv6 strings. No-op if IP was not blocked.
|
||||
async fn unblock_ip(&self, ip: &str) -> Result<(), Error>;
|
||||
fn unblock_ip(&self, ip: &str) -> Result<(), Error>;
|
||||
}
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
use std::collections::HashMap;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use common::model::ip_address::Port;
|
||||
|
||||
use crate::model::access_control::list_type::ListType;
|
||||
@ -13,38 +12,27 @@ use crate::model::monitoring::direction::FlowDirection;
|
||||
/// Distinct from `AccessControlPort` (which only exposes `block_ip` /
|
||||
/// `unblock_ip` for SOAR). `AclService` uses this richer API to serve the
|
||||
/// `/api/acl` HTTP routes.
|
||||
#[async_trait]
|
||||
#[allow(dead_code)]
|
||||
pub trait AccessControlAdminPort: Send + Sync {
|
||||
async fn add_ipv4_list(
|
||||
fn add_ipv4_list(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV4) -> Result<(), Error>;
|
||||
|
||||
fn add_ipv6_list(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV6) -> Result<(), Error>;
|
||||
|
||||
fn remove_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error>;
|
||||
|
||||
async fn add_ipv6_list(
|
||||
fn remove_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error>;
|
||||
|
||||
async fn remove_ipv4_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV4,
|
||||
) -> Result<(), Error>;
|
||||
fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>>;
|
||||
|
||||
async fn remove_ipv6_list(
|
||||
&self,
|
||||
direction: FlowDirection,
|
||||
list_type: ListType,
|
||||
address: SocketAddrV6,
|
||||
) -> Result<(), Error>;
|
||||
|
||||
async fn get_ipv4_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv4Addr, Vec<Port>>;
|
||||
|
||||
async fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>>;
|
||||
fn get_ipv6_list(&self, direction: FlowDirection, list_type: ListType) -> HashMap<Ipv6Addr, Vec<Port>>;
|
||||
}
|
||||
|
||||
@ -55,5 +55,9 @@ traceable! {
|
||||
#[no_source]
|
||||
#[error("Unknown SOAR condition type: {condition_type}")]
|
||||
UnknownConditionType { condition_type: String } => tracing::Level::WARN,
|
||||
|
||||
#[no_source]
|
||||
#[error("Rate-limit owner task is unavailable (channel closed)")]
|
||||
RateLimitOwnerUnavailable => tracing::Level::ERROR,
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user