mirror of
https://github.com/DaLaw2/NetGuardia.git
synced 2026-08-24 14:10:28 +09:00
fix: address PR #19 review feedback across 22 issues
Correctness: - ml/config_loader: gate output_names check on adapter kind so Binary/Multiclass/Autoencoder manifests load cleanly - ml/model_watcher: try_send in notify callback (no more inotify-thread stall on burst) + spawn_blocking around try_reload (no more tokio executor stall on ONNX load) - http/fusion: truncated flag now reflects real overflow, not matches.len() == cap - http/soar: reject empty sources[] in dry-run instead of synthesising an inconsistent active_source_count=1 event - detection/attack_type: port_scan now maps to CanonicalAttackType::PortScan across ML / Correlation / Suricata so cross-source fusion fires - detection/orchestrator: full-scan cleanup_expired, since LRU order reflects access time not insertion time - soar/actions: block_ip/unblock_ip via spawn_blocking so parking_lot + aya syscalls never run on a tokio worker - ml/flow_tracker: fwd/bwd total bytes use packet_length (full IP), matching user intuition and on-wire byte counts - ebpf/xsk_manager: userspace DNS-blacklist drops now update DropMonitor; TX drop accounting uses nb_submitted so ring backpressure is counted - config_service: scrub plaintext smtp_password on SecretStore write - utils/logging: preserve RUST_LOG per-target directives across set_level - Cargo.toml: let notify auto-select FS backend — previous macos_kqueue-only feature meant ModelWatcher silently failed on Linux Cleanup: - Remove #[allow(dead_code)] from 9 port traits; delete the genuinely dead trait methods + associated Database impl stubs this revealed - detection/orchestrator: MAX_DEDUP_ENTRIES becomes const NonZero; tuple-style log! constructors per CODE_STYLE - model/system/health: drop unused is_healthy + task-ref comment - observability/log_buffer: document why parking_lot::Mutex is required - http/model_upload: module doc now correctly describes PromoteGate as AtomicBool reject-not-queue (not a mutex) All 289 tests pass; cargo clippy --tests -- -D warnings clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
d6547aaec3
commit
48effea826
10
Cargo.lock
generated
10
Cargo.lock
generated
@ -1377,6 +1377,15 @@ dependencies = [
|
||||
"percent-encoding",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fsevent-sys"
|
||||
version = "4.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.32"
|
||||
@ -2601,6 +2610,7 @@ checksum = "c533b4c39709f9ba5005d8002048266593c1cfaf3c5f0739d5b8ab0c6c504009"
|
||||
dependencies = [
|
||||
"bitflags 2.11.0",
|
||||
"filetime",
|
||||
"fsevent-sys",
|
||||
"inotify",
|
||||
"kqueue",
|
||||
"libc",
|
||||
|
||||
@ -59,7 +59,7 @@ async-trait = "0.1"
|
||||
dashmap = "6"
|
||||
arc-swap = "1"
|
||||
moka = { version = "0.12", features = ["sync"] }
|
||||
notify = { version = "7", default-features = false, features = ["macos_kqueue"] }
|
||||
notify = "7"
|
||||
|
||||
# Utilities
|
||||
parking_lot = { workspace = true }
|
||||
|
||||
@ -36,6 +36,29 @@ impl DropMonitor {
|
||||
self.counters.snapshot()
|
||||
}
|
||||
|
||||
/// Record a userspace drop decision (XSK worker's DNS filter) by the
|
||||
/// per-reason counter. Callers at this layer haven't parsed src/dst yet,
|
||||
/// so no broadcast event is emitted — `/api/stats/drops` stays correct,
|
||||
/// `/ws/drops` simply does not surface the individual packet. Parse the
|
||||
/// packet upstream if you need a structured event.
|
||||
pub fn record_userspace_drop_count_only(&self, reason: u8) {
|
||||
self.counters.total.fetch_add(1, Ordering::Relaxed);
|
||||
let bucket = match reason {
|
||||
DROP_REASON_ACL_BLACKLIST => Some(&self.counters.acl_blacklist),
|
||||
DROP_REASON_RATE_LIMIT_PKT => Some(&self.counters.rate_limit_pkt),
|
||||
DROP_REASON_RATE_LIMIT_SYN => Some(&self.counters.rate_limit_syn),
|
||||
DROP_REASON_RATE_LIMIT_UDP => Some(&self.counters.rate_limit_udp),
|
||||
DROP_REASON_RATE_LIMIT_DNS => Some(&self.counters.rate_limit_dns),
|
||||
DROP_REASON_PROTOCOL_FILTER => Some(&self.counters.protocol_filter),
|
||||
DROP_REASON_DNS_BLACKLIST => Some(&self.counters.dns_blacklist),
|
||||
DROP_REASON_GEO_BLOCK => Some(&self.counters.geo_block),
|
||||
_ => None,
|
||||
};
|
||||
if let Some(counter) = bucket {
|
||||
counter.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
fn process_event(&self, raw: &RawDropEvent) {
|
||||
self.counters.total.fetch_add(1, Ordering::Relaxed);
|
||||
let bucket = match raw.reason {
|
||||
|
||||
@ -87,7 +87,12 @@ impl EbpfServices {
|
||||
pub async fn run(self: Arc<Self>, sink_factory: Arc<dyn PacketSinkFactory>) -> Result<(), Error> {
|
||||
let xsk_manager = self.xsk_manager.clone();
|
||||
let dns: Arc<dyn DnsQueryFilter> = self.dns_filter.clone();
|
||||
xsk_manager.run(Some(sink_factory), Some(dns), &self.shutdowns)?;
|
||||
xsk_manager.run(
|
||||
Some(sink_factory),
|
||||
Some(dns),
|
||||
Some(self.drop_monitor.clone()),
|
||||
&self.shutdowns,
|
||||
)?;
|
||||
|
||||
let ring_buf = self.drop_ring_buf.lock().take();
|
||||
if let Some(ring_buf) = ring_buf {
|
||||
|
||||
@ -16,6 +16,9 @@ use tokio::sync::oneshot::{self, error::TryRecvError};
|
||||
use xsk_rs::config::{BindFlags, FrameSize, Interface, LibxdpFlags, QueueSize, SocketConfig, UmemConfig};
|
||||
use xsk_rs::{CompQueue, FillQueue, FrameDesc, RxQueue, Socket, TxQueue, Umem};
|
||||
|
||||
use common::define::drop_reason::DROP_REASON_DNS_BLACKLIST;
|
||||
|
||||
use crate::adapter::ebpf::drop_monitor::DropMonitor;
|
||||
use crate::infrastructure::app_config::AppConfig;
|
||||
use crate::interface::port::dns_query_filter::DnsQueryFilter;
|
||||
use crate::interface::port::packet_sink::{PacketSink, PacketSinkFactory};
|
||||
@ -93,6 +96,7 @@ impl XskManager {
|
||||
&self,
|
||||
sinks: Option<Arc<dyn PacketSinkFactory>>,
|
||||
dns_filter: Option<Arc<dyn DnsQueryFilter>>,
|
||||
drop_monitor: Option<Arc<DropMonitor>>,
|
||||
shutdowns: &SegQueue<oneshot::Sender<()>>,
|
||||
) -> Result<(), Error> {
|
||||
// If eBPF failed to load, there are no XSK maps to bind and no queues
|
||||
@ -119,6 +123,7 @@ impl XskManager {
|
||||
Direction::Ingress,
|
||||
sink.clone(),
|
||||
dns_filter.clone(),
|
||||
drop_monitor.clone(),
|
||||
)?;
|
||||
|
||||
let egress_xsk = XskPair::new(
|
||||
@ -129,6 +134,7 @@ impl XskManager {
|
||||
Direction::Egress,
|
||||
sink,
|
||||
None,
|
||||
drop_monitor.clone(),
|
||||
)?;
|
||||
|
||||
let mut xsk_guard = self.xsk_map.lock();
|
||||
@ -172,11 +178,13 @@ pub struct XskPair {
|
||||
frame_pool: Vec<FrameDesc>,
|
||||
sink: Option<Arc<dyn PacketSink>>,
|
||||
dns_filter: Option<Arc<dyn DnsQueryFilter>>,
|
||||
drop_monitor: Option<Arc<DropMonitor>>,
|
||||
packet_buffer_size: usize,
|
||||
buffer_pool_capacity: usize,
|
||||
}
|
||||
|
||||
impl XskPair {
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub fn new(
|
||||
config: NetworkConfig,
|
||||
queue_id: u32,
|
||||
@ -185,6 +193,7 @@ impl XskPair {
|
||||
direction: Direction,
|
||||
sink: Option<Arc<dyn PacketSink>>,
|
||||
dns_filter: Option<Arc<dyn DnsQueryFilter>>,
|
||||
drop_monitor: Option<Arc<DropMonitor>>,
|
||||
) -> Result<Self, Error> {
|
||||
let rx_ifname_c = CString::new(rx_ifname).map_err(|_| SystemError::InvalidConfig)?;
|
||||
|
||||
@ -241,6 +250,7 @@ impl XskPair {
|
||||
frame_pool: pool_frames,
|
||||
sink,
|
||||
dns_filter,
|
||||
drop_monitor,
|
||||
packet_buffer_size: config.packet_buffer_size,
|
||||
buffer_pool_capacity: config.buffer_pool_capacity,
|
||||
};
|
||||
@ -353,10 +363,16 @@ impl XskPair {
|
||||
|
||||
let raw = &contents[..packet_len];
|
||||
|
||||
// DNS blacklist check — drop blacklisted DNS queries before forwarding
|
||||
// DNS blacklist check — drop blacklisted DNS queries before forwarding.
|
||||
// Report to DropMonitor so `/api/stats/drops` and `/ws/drops`
|
||||
// reflect userspace-decided drops (the kernel eBPF never saw
|
||||
// this packet's DNS payload, so it emits no DROP_EVENTS entry).
|
||||
if let Some(ref dns) = self.dns_filter
|
||||
&& dns.is_query_blacklisted(raw)
|
||||
{
|
||||
if let Some(ref monitor) = self.drop_monitor {
|
||||
monitor.record_userspace_drop_count_only(DROP_REASON_DNS_BLACKLIST);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
@ -439,7 +455,6 @@ impl XskPair {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let sent_count = frames.len();
|
||||
for (frame, packet) in frames.iter_mut().zip(packets_to_send.iter()) {
|
||||
unsafe {
|
||||
self.umem
|
||||
@ -465,8 +480,12 @@ impl XskPair {
|
||||
log!(EbpfLog::TXWakeupFailed(e.to_string()));
|
||||
}
|
||||
|
||||
// Log dropped packets when frames < packets
|
||||
let dropped = total_packets - sent_count;
|
||||
// Drop accounting: a packet is dropped whenever we couldn't put it
|
||||
// on the TX ring. That includes both the frame-pool-exhausted path
|
||||
// (frames.len() < total_packets) and the TX-ring backpressure path
|
||||
// (nb_submitted < frames.len()). Using `nb_submitted` as the sent
|
||||
// count covers both.
|
||||
let dropped = total_packets - nb_submitted;
|
||||
if dropped > 0 {
|
||||
log!(EbpfLog::FramePoolExhausted(dropped));
|
||||
}
|
||||
|
||||
@ -63,11 +63,11 @@ async fn explain_ip(req: HttpRequest, audit: web::Data<dyn AuditRepo>) -> impl R
|
||||
}
|
||||
};
|
||||
|
||||
let matches = filter_fusion_evidence_for_ip(&entries, &src_ip, FUSION_EXPLAIN_RESPONSE_CAP);
|
||||
let (matches, truncated) = filter_fusion_evidence_for_ip(&entries, &src_ip, FUSION_EXPLAIN_RESPONSE_CAP);
|
||||
HttpResponse::Ok().json(serde_json::json!({
|
||||
"src_ip": src_ip,
|
||||
"match_count": matches.len(),
|
||||
"truncated": matches.len() >= FUSION_EXPLAIN_RESPONSE_CAP,
|
||||
"truncated": truncated,
|
||||
"entries": matches,
|
||||
}))
|
||||
}
|
||||
@ -78,17 +78,29 @@ async fn explain_ip(req: HttpRequest, audit: web::Data<dyn AuditRepo>) -> impl R
|
||||
/// append-only, so a malformed row is an integrity concern for the
|
||||
/// audit-verify endpoint to surface, not this handler.
|
||||
///
|
||||
/// Returns `(entries_up_to_cap, truncated)`. `truncated` is `true` when
|
||||
/// at least one matching entry was dropped — `matches.len() == cap` does
|
||||
/// NOT imply truncation, so we look at `cap + 1` candidates and set the
|
||||
/// flag only when the overflow entry exists.
|
||||
///
|
||||
/// Extracted as a free function so tests can cover the filter /
|
||||
/// ordering / cap behaviour without an in-memory DB.
|
||||
pub fn filter_fusion_evidence_for_ip(entries: &[AuditLogEntry], target_ip: &str, cap: usize) -> Vec<serde_json::Value> {
|
||||
pub fn filter_fusion_evidence_for_ip(
|
||||
entries: &[AuditLogEntry],
|
||||
target_ip: &str,
|
||||
cap: usize,
|
||||
) -> (Vec<serde_json::Value>, bool) {
|
||||
let mut filtered: Vec<&AuditLogEntry> = entries
|
||||
.iter()
|
||||
.filter(|entry| detail_matches_src_ip(&entry.detail, target_ip))
|
||||
.collect();
|
||||
filtered.sort_by_key(|entry| entry.id);
|
||||
filtered
|
||||
let truncated = filtered.len() > cap;
|
||||
if truncated {
|
||||
filtered.truncate(cap);
|
||||
}
|
||||
let rendered = filtered
|
||||
.into_iter()
|
||||
.take(cap)
|
||||
.map(|entry| {
|
||||
let detail: serde_json::Value = serde_json::from_str(&entry.detail).unwrap_or(serde_json::Value::Null);
|
||||
serde_json::json!({
|
||||
@ -99,7 +111,8 @@ pub fn filter_fusion_evidence_for_ip(entries: &[AuditLogEntry], target_ip: &str,
|
||||
"detail": detail,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
.collect();
|
||||
(rendered, truncated)
|
||||
}
|
||||
|
||||
fn detail_matches_src_ip(detail_json: &str, target_ip: &str) -> bool {
|
||||
@ -138,8 +151,9 @@ mod tests {
|
||||
entry(2, "10.0.0.5", "port_scan"),
|
||||
entry(3, "1.2.3.4", "exploit"),
|
||||
];
|
||||
let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
|
||||
let (got, truncated) = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
|
||||
assert_eq!(got.len(), 2);
|
||||
assert!(!truncated);
|
||||
assert_eq!(got[0]["id"], 1);
|
||||
assert_eq!(got[1]["id"], 3);
|
||||
}
|
||||
@ -152,7 +166,7 @@ mod tests {
|
||||
entry(10, "1.1.1.1", "b"),
|
||||
entry(20, "1.1.1.1", "c"),
|
||||
];
|
||||
let got = filter_fusion_evidence_for_ip(&entries, "1.1.1.1", 100);
|
||||
let (got, _) = filter_fusion_evidence_for_ip(&entries, "1.1.1.1", 100);
|
||||
let ids: Vec<i64> = got.iter().map(|v| v["id"].as_i64().unwrap()).collect();
|
||||
assert_eq!(ids, vec![10, 20, 30]);
|
||||
}
|
||||
@ -160,12 +174,23 @@ mod tests {
|
||||
#[test]
|
||||
fn filter_applies_response_cap() {
|
||||
let entries: Vec<AuditLogEntry> = (1..=10).map(|i| entry(i, "9.9.9.9", "x")).collect();
|
||||
let got = filter_fusion_evidence_for_ip(&entries, "9.9.9.9", 3);
|
||||
let (got, truncated) = filter_fusion_evidence_for_ip(&entries, "9.9.9.9", 3);
|
||||
assert_eq!(got.len(), 3);
|
||||
assert!(truncated, "10 matching rows with cap=3 must set truncated");
|
||||
let ids: Vec<i64> = got.iter().map(|v| v["id"].as_i64().unwrap()).collect();
|
||||
assert_eq!(ids, vec![1, 2, 3], "cap takes oldest, not newest");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filter_exactly_cap_is_not_truncated() {
|
||||
// Regression guard: `matches.len() == cap` with no overflow row must
|
||||
// return `truncated = false`. Earlier `>=` check mis-flagged this.
|
||||
let entries: Vec<AuditLogEntry> = (1..=3).map(|i| entry(i, "9.9.9.9", "x")).collect();
|
||||
let (got, truncated) = filter_fusion_evidence_for_ip(&entries, "9.9.9.9", 3);
|
||||
assert_eq!(got.len(), 3);
|
||||
assert!(!truncated, "exactly cap matches must NOT report truncated");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filter_drops_rows_with_unparseable_detail() {
|
||||
let good = entry(1, "1.2.3.4", "brute_force");
|
||||
@ -176,7 +201,7 @@ mod tests {
|
||||
detail: "{{not json".into(),
|
||||
created_at: "2026-04-18T10:00:02Z".into(),
|
||||
};
|
||||
let got = filter_fusion_evidence_for_ip(&[good, bad], "1.2.3.4", 100);
|
||||
let (got, _) = filter_fusion_evidence_for_ip(&[good, bad], "1.2.3.4", 100);
|
||||
assert_eq!(got.len(), 1);
|
||||
assert_eq!(got[0]["id"], 1);
|
||||
}
|
||||
@ -184,13 +209,15 @@ mod tests {
|
||||
#[test]
|
||||
fn filter_nonmatching_ip_returns_empty() {
|
||||
let entries = [entry(1, "1.2.3.4", "brute_force")];
|
||||
assert!(filter_fusion_evidence_for_ip(&entries, "5.6.7.8", 100).is_empty());
|
||||
let (got, truncated) = filter_fusion_evidence_for_ip(&entries, "5.6.7.8", 100);
|
||||
assert!(got.is_empty());
|
||||
assert!(!truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filter_preserves_detail_structure_in_response() {
|
||||
let entries = [entry(1, "1.2.3.4", "brute_force")];
|
||||
let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
|
||||
let (got, _) = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
|
||||
assert_eq!(got.len(), 1);
|
||||
let detail = &got[0]["detail"];
|
||||
assert_eq!(detail["attack_type"], "brute_force");
|
||||
|
||||
@ -2,8 +2,9 @@
|
||||
//! YAML field, an `onnx` binary field, and an optional `scaler` JSON
|
||||
//! sidecar; streams each to `models/.staging/<uuid>/` with enforced
|
||||
//! size caps, runs structural + ONNX shape validation, then atomically
|
||||
//! renames into `models/` under a process-wide mutex so concurrent
|
||||
//! uploads serialize at the rename step. A WORM `model_swap` audit
|
||||
//! renames into `models/` under a process-wide gate (AtomicBool, not a
|
||||
//! mutex — see `PromoteGate`): a second concurrent promote is rejected
|
||||
//! with 409 Conflict rather than queued. A WORM `model_swap` audit
|
||||
//! entry records the SHA-256 of both committed files plus a snapshot
|
||||
//! of the pre-swap state.
|
||||
//!
|
||||
|
||||
@ -379,7 +379,10 @@ fn build_event(req: DryRunRequest) -> Result<ThreatDetectedEvent, String> {
|
||||
.collect::<Result<Vec<_>, _>>()?,
|
||||
None => vec![DetectionSource::ML],
|
||||
};
|
||||
let active_source_count = req.active_source_count.unwrap_or(sources.len()).max(1);
|
||||
if sources.is_empty() {
|
||||
return Err("sources[] must contain at least one DetectionSource (send null to default to [ML])".to_string());
|
||||
}
|
||||
let active_source_count = req.active_source_count.unwrap_or(sources.len());
|
||||
let fused_confidence = req.fused_confidence.unwrap_or(req.confidence);
|
||||
Ok(ThreatDetectedEvent {
|
||||
attack_type: req.attack_type,
|
||||
|
||||
@ -13,7 +13,7 @@ use crate::interface::port::api_key::{ApiKeyListItem, ApiKeyRepo};
|
||||
use crate::interface::port::audit::{AuditLogEntry, AuditRepo};
|
||||
use crate::interface::port::db_admin::DbAdminRepo;
|
||||
use crate::interface::port::enforcement::EnforcementRepo;
|
||||
use crate::interface::port::identity::{IdentityRepo, UserGroupTuple, UserListItem, UserTuple, UserWithGroups};
|
||||
use crate::interface::port::identity::{IdentityRepo, UserGroupTuple, UserTuple, UserWithGroups};
|
||||
use crate::interface::port::setting::SettingRepo;
|
||||
use crate::interface::port::soar::{PlaybookRow, SoarExecutionRow, SoarRepo};
|
||||
use crate::interface::port::stats::StatsRepo;
|
||||
@ -662,26 +662,6 @@ impl Database {
|
||||
Ok(conn.query_row("SELECT COUNT(*) FROM users", [], |row| row.get(0))?)
|
||||
}
|
||||
|
||||
pub fn list_users(&self) -> Result<Vec<UserListItem>, Error> {
|
||||
let conn = self.conn()?;
|
||||
let mut stmt =
|
||||
conn.prepare("SELECT id, username, role, force_password_change, created_at FROM users ORDER BY id")?;
|
||||
let rows = stmt.query_map([], |row| {
|
||||
Ok((
|
||||
row.get::<_, i64>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
row.get::<_, i64>(3)? != 0,
|
||||
row.get::<_, String>(4)?,
|
||||
))
|
||||
})?;
|
||||
let mut results = Vec::new();
|
||||
for row in rows {
|
||||
results.push(row?);
|
||||
}
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
pub fn list_users_with_groups(&self) -> Result<Vec<UserWithGroups>, Error> {
|
||||
let conn = self.conn()?;
|
||||
let mut stmt = conn.prepare(
|
||||
@ -1082,6 +1062,10 @@ impl Database {
|
||||
Ok(conn.last_insert_rowid())
|
||||
}
|
||||
|
||||
/// Test-only helper: direct insert of a SOAR block rule row. Production
|
||||
/// code goes through `commit_soar_block_to_db` which atomically writes
|
||||
/// both `soar_block_rules` and `acl_rules` under a transaction.
|
||||
#[cfg(test)]
|
||||
pub fn insert_soar_block_rule(&self, source_ip: &str, playbook_id: i64, expires_at: &str) -> Result<i64, Error> {
|
||||
let conn = self.conn()?;
|
||||
conn.execute(
|
||||
@ -1271,25 +1255,6 @@ impl Database {
|
||||
Ok(conn.last_insert_rowid())
|
||||
}
|
||||
|
||||
pub fn update_playbook(&self, id: i64, row: &UpdatePlaybookRow) -> Result<bool, Error> {
|
||||
let conn = self.conn()?;
|
||||
let rows = conn.execute(
|
||||
"UPDATE playbooks SET name = ?2, trigger_event = ?3, condition_threshold = ?4, \
|
||||
condition_count = ?5, condition_window_secs = ?6, cooldown_secs = ?7, \
|
||||
updated_at = datetime('now') WHERE id = ?1",
|
||||
params![
|
||||
id,
|
||||
row.name,
|
||||
row.trigger_event,
|
||||
row.condition_threshold,
|
||||
row.condition_count,
|
||||
row.condition_window_secs,
|
||||
row.cooldown_secs
|
||||
],
|
||||
)?;
|
||||
Ok(rows > 0)
|
||||
}
|
||||
|
||||
pub fn update_playbook_enabled(&self, id: i64, enabled: bool) -> Result<bool, Error> {
|
||||
let conn = self.conn()?;
|
||||
let rows = conn.execute(
|
||||
@ -1305,24 +1270,6 @@ impl Database {
|
||||
Ok(rows > 0)
|
||||
}
|
||||
|
||||
pub fn delete_playbook_actions(&self, playbook_id: i64) -> Result<(), Error> {
|
||||
let conn = self.conn()?;
|
||||
conn.execute(
|
||||
"DELETE FROM playbook_actions WHERE playbook_id = ?1",
|
||||
params![playbook_id],
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn delete_playbook_conditions(&self, playbook_id: i64) -> Result<(), Error> {
|
||||
let conn = self.conn()?;
|
||||
conn.execute(
|
||||
"DELETE FROM playbook_conditions WHERE playbook_id = ?1",
|
||||
params![playbook_id],
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn insert_playbook_condition(
|
||||
&self,
|
||||
playbook_id: i64,
|
||||
@ -1757,9 +1704,6 @@ impl AclRepo for Database {
|
||||
) -> Result<(), Error> {
|
||||
self.delete_acl_rule(ip_version, direction, list_type, ip_address, port)
|
||||
}
|
||||
fn load_acl_rules(&self) -> Result<Vec<AclRuleTuple>, Error> {
|
||||
self.load_acl_rules()
|
||||
}
|
||||
fn has_manual_acl_rule(&self, ip_address: &str) -> Result<bool, Error> {
|
||||
self.has_manual_acl_rule(ip_address)
|
||||
}
|
||||
@ -1778,27 +1722,18 @@ impl EnforcementRepo for Database {
|
||||
fn set_rate_limit(&self, key: &str, value: u64) -> Result<(), Error> {
|
||||
self.set_rate_limit(key, value)
|
||||
}
|
||||
fn load_rate_limit_config(&self) -> Result<Vec<(String, u64)>, Error> {
|
||||
self.load_rate_limit_config()
|
||||
}
|
||||
fn insert_dns_domain(&self, domain: &str) -> Result<(), Error> {
|
||||
self.insert_dns_domain(domain)
|
||||
}
|
||||
fn delete_dns_domain(&self, domain: &str) -> Result<(), Error> {
|
||||
self.delete_dns_domain(domain)
|
||||
}
|
||||
fn load_dns_domains(&self) -> Result<Vec<String>, Error> {
|
||||
self.load_dns_domains()
|
||||
}
|
||||
fn insert_geo_country(&self, code: &str) -> Result<(), Error> {
|
||||
self.insert_geo_country(code)
|
||||
}
|
||||
fn delete_geo_country(&self, code: &str) -> Result<(), Error> {
|
||||
self.delete_geo_country(code)
|
||||
}
|
||||
fn load_geo_countries(&self) -> Result<Vec<String>, Error> {
|
||||
self.load_geo_countries()
|
||||
}
|
||||
}
|
||||
|
||||
impl SettingRepo for Database {
|
||||
@ -1841,12 +1776,6 @@ impl IdentityRepo for Database {
|
||||
fn update_user_password(&self, user_id: i64, password_hash: &str) -> Result<(), Error> {
|
||||
self.update_user_password(user_id, password_hash)
|
||||
}
|
||||
fn user_count(&self) -> Result<i64, Error> {
|
||||
self.user_count()
|
||||
}
|
||||
fn list_users(&self) -> Result<Vec<UserListItem>, Error> {
|
||||
self.list_users()
|
||||
}
|
||||
fn list_users_with_groups(&self) -> Result<Vec<UserWithGroups>, Error> {
|
||||
self.list_users_with_groups()
|
||||
}
|
||||
@ -1883,9 +1812,6 @@ impl IdentityRepo for Database {
|
||||
fn get_user_permissions(&self, user_id: i64) -> Result<Vec<String>, Error> {
|
||||
self.get_user_permissions(user_id)
|
||||
}
|
||||
fn cleanup_user_memberships(&self, user_id: i64) -> Result<(), Error> {
|
||||
self.cleanup_user_memberships(user_id)
|
||||
}
|
||||
fn get_group_member_ids(&self, group_id: i64) -> Result<Vec<i64>, Error> {
|
||||
self.get_group_member_ids(group_id)
|
||||
}
|
||||
@ -1904,63 +1830,21 @@ impl IdentityRepo for Database {
|
||||
}
|
||||
|
||||
impl SoarRepo for Database {
|
||||
fn insert_playbook(
|
||||
&self,
|
||||
name: &str,
|
||||
trigger_event: &str,
|
||||
threshold: Option<f64>,
|
||||
count: Option<i64>,
|
||||
window: Option<i64>,
|
||||
cooldown: i64,
|
||||
) -> Result<i64, Error> {
|
||||
self.insert_playbook(name, trigger_event, threshold, count, window, cooldown)
|
||||
}
|
||||
fn insert_playbook_action(
|
||||
&self,
|
||||
playbook_id: i64,
|
||||
action_order: i64,
|
||||
action_type: &str,
|
||||
params_json: &str,
|
||||
) -> Result<i64, Error> {
|
||||
self.insert_playbook_action(playbook_id, action_order, action_type, params_json)
|
||||
}
|
||||
fn load_playbooks_with_actions(&self) -> Result<Vec<PlaybookRow>, Error> {
|
||||
self.load_playbooks_with_actions()
|
||||
}
|
||||
fn update_playbook(&self, id: i64, row: &UpdatePlaybookRow) -> Result<bool, Error> {
|
||||
self.update_playbook(id, row)
|
||||
}
|
||||
fn update_playbook_enabled(&self, id: i64, enabled: bool) -> Result<bool, Error> {
|
||||
self.update_playbook_enabled(id, enabled)
|
||||
}
|
||||
fn delete_playbook(&self, id: i64) -> Result<bool, Error> {
|
||||
self.delete_playbook(id)
|
||||
}
|
||||
fn delete_playbook_actions(&self, playbook_id: i64) -> Result<(), Error> {
|
||||
self.delete_playbook_actions(playbook_id)
|
||||
}
|
||||
fn delete_playbook_conditions(&self, playbook_id: i64) -> Result<(), Error> {
|
||||
self.delete_playbook_conditions(playbook_id)
|
||||
}
|
||||
fn seed_default_playbooks(&self) -> Result<(), Error> {
|
||||
self.seed_default_playbooks()
|
||||
}
|
||||
fn insert_playbook_condition(
|
||||
&self,
|
||||
playbook_id: i64,
|
||||
condition_type: &str,
|
||||
operator: &str,
|
||||
value: &str,
|
||||
value2: Option<&str>,
|
||||
) -> Result<i64, Error> {
|
||||
self.insert_playbook_condition(playbook_id, condition_type, operator, value, value2)
|
||||
}
|
||||
fn load_all_playbook_conditions(&self) -> Result<Vec<(i64, i64, String, String, String, Option<String>)>, Error> {
|
||||
self.load_all_playbook_conditions()
|
||||
}
|
||||
fn insert_soar_block_rule(&self, source_ip: &str, playbook_id: i64, expires_at: &str) -> Result<i64, Error> {
|
||||
self.insert_soar_block_rule(source_ip, playbook_id, expires_at)
|
||||
}
|
||||
fn count_active_soar_blocks(&self) -> Result<u32, Error> {
|
||||
self.count_active_soar_blocks()
|
||||
}
|
||||
@ -2107,9 +1991,6 @@ impl AuditRepo for Database {
|
||||
fn insert_audit_log(&self, actor: &str, action: &str, detail: &str) -> Result<(), Error> {
|
||||
self.insert_audit_log(actor, action, detail)
|
||||
}
|
||||
fn list_audit_logs(&self) -> Result<Vec<AuditLogEntry>, Error> {
|
||||
self.list_audit_logs()
|
||||
}
|
||||
fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result<Vec<AuditLogEntry>, Error> {
|
||||
self.list_audit_logs_by_action(action, limit)
|
||||
}
|
||||
@ -2345,13 +2226,17 @@ mod tests {
|
||||
|
||||
let acl: &dyn AclRepo = &db;
|
||||
acl.insert_acl_rule(4, "source", "blacklist", "10.0.0.1", 443).unwrap();
|
||||
let rules = acl.load_acl_rules().unwrap();
|
||||
// load_acl_rules is an inherent Database method (not on AclRepo),
|
||||
// so go through `&db` directly for this read-back assertion.
|
||||
let rules = db.load_acl_rules().unwrap();
|
||||
assert_eq!(rules.len(), 1);
|
||||
|
||||
let identity: &dyn IdentityRepo = &db;
|
||||
assert_eq!(identity.user_count().unwrap(), 0);
|
||||
// user_count is inherent — inserts still go through the trait so
|
||||
// the vtable has something to exercise.
|
||||
assert_eq!(db.user_count().unwrap(), 0);
|
||||
identity.insert_user("test", "hash", "viewer", false).unwrap();
|
||||
assert_eq!(identity.user_count().unwrap(), 1);
|
||||
assert_eq!(db.user_count().unwrap(), 1);
|
||||
}
|
||||
|
||||
/// Happy path. Verifies `commit_soar_block_to_db` writes both
|
||||
|
||||
@ -28,7 +28,6 @@ use std::task::{Context, Poll};
|
||||
use actix_web::body::EitherBody;
|
||||
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform};
|
||||
use actix_web::http::Method;
|
||||
use actix_web::http::header::HeaderMap;
|
||||
use actix_web::{Error as ActixError, HttpResponse};
|
||||
|
||||
/// Request header carrying the CSRF token. Clients (frontend fetch /
|
||||
@ -134,13 +133,6 @@ fn is_csrf_exempt_path(path: &str) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Convenience accessor used by request-scoped code that would
|
||||
/// otherwise reach into `req.headers().get(CSRF_HEADER)` directly.
|
||||
#[allow(dead_code)]
|
||||
pub fn csrf_token_from_headers(headers: &HeaderMap) -> Option<&str> {
|
||||
headers.get(CSRF_HEADER).and_then(|v| v.to_str().ok())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@ -168,7 +168,11 @@ impl ConfigService {
|
||||
}
|
||||
}
|
||||
|
||||
// Route secret keys through SecretStore (encrypted storage)
|
||||
// Route secret keys through SecretStore (encrypted storage).
|
||||
// After writing to SecretStore, scrub the plaintext row in `settings`
|
||||
// so a legacy plaintext value from pre-SecretStore deployments cannot
|
||||
// linger — readers fall back to SecretStore when the plaintext row
|
||||
// is empty.
|
||||
if let Some(ref secrets) = self.secrets {
|
||||
for key in SECRET_KEYS {
|
||||
// Secret keys live under their parent section (e.g., smtp_password under smtp)
|
||||
@ -180,6 +184,7 @@ impl ConfigService {
|
||||
.and_then(json_value_as_string)
|
||||
{
|
||||
secrets.set_secret(key, &val)?;
|
||||
self.db.set_setting(key, "")?;
|
||||
updated.push(key.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
@ -27,7 +27,11 @@ const CLEANUP_INTERVAL_SECS: u64 = 60;
|
||||
const REPEAT_OFFENDER_WINDOW_SECS: u64 = 2 * 60 * 60;
|
||||
|
||||
/// Maximum dedup entries to prevent unbounded memory growth under sustained attack.
|
||||
const MAX_DEDUP_ENTRIES: usize = 50_000;
|
||||
/// Declared as `NonZero` at compile time so `LruCache::new` never needs a
|
||||
/// runtime unwrap — if this ever goes to zero, the const expression fails
|
||||
/// to compile, not the running server.
|
||||
// SAFETY: NonZero::new on a non-zero literal is infallible; const-evaluated.
|
||||
const MAX_DEDUP_ENTRIES: NonZero<usize> = NonZero::new(50_000).unwrap();
|
||||
|
||||
/// Actor recorded on every fusion-chain WORM entry. Stable across releases —
|
||||
/// downstream audit tooling filters on this string.
|
||||
@ -92,10 +96,11 @@ impl DetectionOrchestrator {
|
||||
comm,
|
||||
geoip,
|
||||
metrics,
|
||||
// SAFETY: NonZero::new on a non-zero literal is infallible.
|
||||
// SAFETY: NonZero::new on non-zero literals; MAX_DEDUP_ENTRIES is
|
||||
// already a NonZero const so no unwrap needed for that one.
|
||||
src_ip_counts: LruCache::new(NonZero::new(10_000).unwrap()),
|
||||
repeat_tracker: LruCache::new(NonZero::new(5_000).unwrap()),
|
||||
dedup: LruCache::new(NonZero::new(MAX_DEDUP_ENTRIES).unwrap()),
|
||||
dedup: LruCache::new(MAX_DEDUP_ENTRIES),
|
||||
dedup_window: Duration::from_secs(DEDUP_WINDOW_SECS),
|
||||
fusion_windows: FusionWindowLengths::default(),
|
||||
}
|
||||
@ -205,10 +210,7 @@ impl DetectionOrchestrator {
|
||||
let key_was_absent = self.dedup.peek(&key).is_none();
|
||||
if was_full && key_was_absent {
|
||||
self.metrics.record_eviction();
|
||||
log!(DetectionLog::FusionWindowEvicted {
|
||||
key_src: key.0.clone(),
|
||||
key_type: key.1.clone(),
|
||||
});
|
||||
log!(DetectionLog::FusionWindowEvicted(key.0.clone(), key.1.clone()));
|
||||
}
|
||||
|
||||
self.dedup.put(
|
||||
@ -279,7 +281,7 @@ impl DetectionOrchestrator {
|
||||
};
|
||||
|
||||
if let Err(e) = self.comm.publish_event(audit).await {
|
||||
log!(DetectionLog::FusionAuditPublishFailed { err: e.to_string() });
|
||||
log!(DetectionLog::FusionAuditPublishFailed(e.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
@ -344,19 +346,28 @@ impl DetectionOrchestrator {
|
||||
}
|
||||
|
||||
fn cleanup_expired(&mut self) {
|
||||
// Full scan: LRU order reflects access time, not insertion time, so
|
||||
// peek_lru + break-on-first-unexpired would skip older idle entries
|
||||
// sitting in the middle of the map. Dedup touches an entry's LRU
|
||||
// position via `get_mut` on every re-emit, which can leave an entry
|
||||
// with an older `first_emitted_at` deeper in the cache than a newly
|
||||
// inserted neighbour. A full retain is O(N) but cleanup runs every
|
||||
// 60s and `MAX_DEDUP_ENTRIES` caps N at 50_000 — one walk is cheap.
|
||||
let now = Instant::now();
|
||||
let window = self.dedup_window;
|
||||
while let Some((_, entry)) = self.dedup.peek_lru() {
|
||||
let mut expired: Vec<(String, String)> = Vec::new();
|
||||
for (key, entry) in self.dedup.iter() {
|
||||
if now
|
||||
.checked_duration_since(entry.first_emitted_at)
|
||||
.unwrap_or(Duration::ZERO)
|
||||
>= window
|
||||
{
|
||||
self.dedup.pop_lru();
|
||||
} else {
|
||||
break;
|
||||
expired.push(key.clone());
|
||||
}
|
||||
}
|
||||
for key in expired {
|
||||
self.dedup.pop(&key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -2,7 +2,7 @@ use std::collections::{BTreeMap, HashMap};
|
||||
use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use super::manifest::{LabelSpec, ModelManifest};
|
||||
use super::manifest::{AdapterKind, LabelSpec, ModelManifest};
|
||||
use crate::model::error::ml::MLError;
|
||||
use crate::model::system::config::MLInferenceConfig;
|
||||
|
||||
@ -45,11 +45,16 @@ impl MLInferenceConfig {
|
||||
|
||||
reconcile_features(&manifest, &config, manifest_path)?;
|
||||
apply_manifest_overrides(&manifest, &mut config);
|
||||
validate_for_adapter(&config, manifest.adapter)?;
|
||||
|
||||
Ok((config, manifest))
|
||||
}
|
||||
}
|
||||
|
||||
/// Shape checks that apply to every sidecar regardless of adapter kind:
|
||||
/// feature list non-empty, scaler arrays match feature count. Per-adapter
|
||||
/// output-head counts live in `validate_for_adapter` and only run via the
|
||||
/// manifest path where the adapter kind is known.
|
||||
fn validate(config: &MLInferenceConfig) -> Result<(), MLError> {
|
||||
if config.ae_feature_names.is_empty() {
|
||||
return Err(MLError::ConfigParseFailed("ae_feature_names is empty"));
|
||||
@ -60,10 +65,25 @@ fn validate(config: &MLInferenceConfig) -> Result<(), MLError> {
|
||||
if config.ae_scaler_std.len() != config.ae_feature_names.len() {
|
||||
return Err(MLError::ConfigParseFailed("scaler std length mismatch"));
|
||||
}
|
||||
if config.output_names.len() != 3 {
|
||||
return Err(MLError::ConfigParseFailed(
|
||||
"MultiTaskModel requires exactly 3 output_names (anomaly, class_probs, c2_score)",
|
||||
));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Adapter-specific output-head count check. MultiTask ships three heads
|
||||
/// (anomaly / class_probs / c2_score); single-head adapters (Classifier,
|
||||
/// Autoencoder) ship one. Mismatch here means the sidecar belongs to a
|
||||
/// different adapter kind than the manifest declares — a deployment bug
|
||||
/// rather than a runtime fault.
|
||||
fn validate_for_adapter(config: &MLInferenceConfig, adapter: AdapterKind) -> Result<(), MLError> {
|
||||
let expected = match adapter {
|
||||
AdapterKind::MultiTask => 3,
|
||||
AdapterKind::ClassifierOnly | AdapterKind::AutoencoderOnly => 1,
|
||||
};
|
||||
if config.output_names.len() != expected {
|
||||
return Err(MLError::ConfigParseFailed(match adapter {
|
||||
AdapterKind::MultiTask => "MultiTaskModel requires exactly 3 output_names (anomaly, class_probs, c2_score)",
|
||||
AdapterKind::ClassifierOnly => "ClassifierOnly adapter requires exactly 1 output_name (class_probs)",
|
||||
AdapterKind::AutoencoderOnly => "AutoencoderOnly adapter requires exactly 1 output_name (reconstruction)",
|
||||
}));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@ -145,7 +145,7 @@ impl FlowData {
|
||||
if self.fwd_packets.len() < FLOW_MAX_PACKETS_PER_DIRECTION {
|
||||
self.fwd_packets.push(packet_data.clone());
|
||||
}
|
||||
self.fwd_total_bytes += packet.payload_length as u64;
|
||||
self.fwd_total_bytes += packet.packet_length as u64;
|
||||
self.fwd_header_bytes += packet.header_length as u64;
|
||||
if self.init_win_bytes_fwd == 0 {
|
||||
self.init_win_bytes_fwd = packet.tcp_window_size;
|
||||
@ -155,7 +155,7 @@ impl FlowData {
|
||||
if self.bwd_packets.len() < FLOW_MAX_PACKETS_PER_DIRECTION {
|
||||
self.bwd_packets.push(packet_data.clone());
|
||||
}
|
||||
self.bwd_total_bytes += packet.payload_length as u64;
|
||||
self.bwd_total_bytes += packet.packet_length as u64;
|
||||
self.bwd_header_bytes += packet.header_length as u64;
|
||||
if self.init_win_bytes_bwd == 0 {
|
||||
self.init_win_bytes_bwd = packet.tcp_window_size;
|
||||
|
||||
@ -68,7 +68,12 @@ impl ModelWatcher {
|
||||
}
|
||||
sleep(Duration::from_secs(DEBOUNCE_SECS)).await;
|
||||
while rx.try_recv().is_ok() {}
|
||||
self.try_reload();
|
||||
// `try_reload` loads the manifest + sidecar + ONNX off disk, any
|
||||
// of which can block for >10ms on a cold cache — move it off the
|
||||
// tokio worker so the rest of the async runtime keeps turning.
|
||||
let inference = Arc::clone(&self.inference);
|
||||
let app_config = Arc::clone(&self.app_config);
|
||||
let _ = tokio::task::spawn_blocking(move || try_reload(&inference, &app_config)).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@ -80,7 +85,10 @@ impl ModelWatcher {
|
||||
if !is_relevant_event(&event) {
|
||||
return;
|
||||
}
|
||||
let _ = tx.blocking_send(());
|
||||
// The callback runs on notify's OS dispatch thread, which must
|
||||
// not block on a full channel — the debounce loop coalesces
|
||||
// duplicates anyway, so dropping when full is safe.
|
||||
let _ = tx.try_send(());
|
||||
}
|
||||
})
|
||||
.map_err(MLError::ModelWatcherFailed)?;
|
||||
@ -92,55 +100,57 @@ impl ModelWatcher {
|
||||
.map_err(MLError::ModelWatcherFailed)?;
|
||||
Ok(watcher)
|
||||
}
|
||||
}
|
||||
|
||||
/// Full reload: manifest presence check → config re-parse → adapter build →
|
||||
/// atomic state swap. Any failure lands the pipeline in `Error` rather
|
||||
/// than crashing.
|
||||
fn try_reload(&self) {
|
||||
let manifest_path = PathBuf::from(MODELS_DIR).join(MANIFEST_FILENAME);
|
||||
/// Full reload: manifest presence check → config re-parse → adapter build →
|
||||
/// atomic state swap. Any failure lands the pipeline in `Error` rather
|
||||
/// than crashing. Runs under `spawn_blocking` because manifest + sidecar +
|
||||
/// ONNX loads are synchronous disk I/O plus a `tract` graph solve that
|
||||
/// routinely takes >10 ms.
|
||||
fn try_reload(inference: &Inference, app_config: &AppConfig) {
|
||||
let manifest_path = PathBuf::from(MODELS_DIR).join(MANIFEST_FILENAME);
|
||||
|
||||
// Path 1 — manifest disappeared: transition to Dormant.
|
||||
if !manifest_path.exists() {
|
||||
log!(MLLog::ModelReloadStarting);
|
||||
self.inference.swap_state(ModelSourceState::Dormant);
|
||||
log!(MLLog::ModelReloadSuccess);
|
||||
return;
|
||||
}
|
||||
|
||||
// Path 2 — manifest present: re-read + rebuild adapter.
|
||||
// Path 1 — manifest disappeared: transition to Dormant.
|
||||
if !manifest_path.exists() {
|
||||
log!(MLLog::ModelReloadStarting);
|
||||
let batch_size = self.app_config.inference.inference_batch_size;
|
||||
match MLInferenceConfig::from_manifest_with_sidecar(&manifest_path) {
|
||||
Ok((config, manifest)) => match build_adapter(&manifest, Some(&manifest_path), &config, batch_size) {
|
||||
Ok(adapter) => {
|
||||
let info = ModelInfo::new(
|
||||
manifest.name.clone(),
|
||||
manifest.adapter.as_str().to_string(),
|
||||
manifest.features.len(),
|
||||
);
|
||||
self.inference.swap_state(ModelSourceState::Active { adapter, info });
|
||||
log!(MLLog::ModelReloadSuccess);
|
||||
}
|
||||
Err(e) => {
|
||||
self.record_error(e.to_string(), Some(manifest_path.clone()));
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
self.record_error(e.to_string(), Some(manifest_path.clone()));
|
||||
}
|
||||
}
|
||||
inference.swap_state(ModelSourceState::Dormant);
|
||||
log!(MLLog::ModelReloadSuccess);
|
||||
return;
|
||||
}
|
||||
|
||||
fn record_error(&self, msg: String, last_attempted_path: Option<PathBuf>) {
|
||||
log!(MLLog::ModelReloadFailed(msg.clone()));
|
||||
self.inference.swap_state(ModelSourceState::Error {
|
||||
msg,
|
||||
since: SystemTime::now(),
|
||||
last_attempted_path,
|
||||
});
|
||||
// Path 2 — manifest present: re-read + rebuild adapter.
|
||||
log!(MLLog::ModelReloadStarting);
|
||||
let batch_size = app_config.inference.inference_batch_size;
|
||||
match MLInferenceConfig::from_manifest_with_sidecar(&manifest_path) {
|
||||
Ok((config, manifest)) => match build_adapter(&manifest, Some(&manifest_path), &config, batch_size) {
|
||||
Ok(adapter) => {
|
||||
let info = ModelInfo::new(
|
||||
manifest.name.clone(),
|
||||
manifest.adapter.as_str().to_string(),
|
||||
manifest.features.len(),
|
||||
);
|
||||
inference.swap_state(ModelSourceState::Active { adapter, info });
|
||||
log!(MLLog::ModelReloadSuccess);
|
||||
}
|
||||
Err(e) => {
|
||||
record_error(inference, e.to_string(), Some(manifest_path.clone()));
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
record_error(inference, e.to_string(), Some(manifest_path.clone()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn record_error(inference: &Inference, msg: String, last_attempted_path: Option<PathBuf>) {
|
||||
log!(MLLog::ModelReloadFailed(msg.clone()));
|
||||
inference.swap_state(ModelSourceState::Error {
|
||||
msg,
|
||||
since: SystemTime::now(),
|
||||
last_attempted_path,
|
||||
});
|
||||
}
|
||||
|
||||
/// Inbound event filter. Ignore `.staging/` paths entirely; pass through
|
||||
/// `.onnx` / `.yaml` / `.yml` / `.json` changes in `models/`.
|
||||
fn is_relevant_event(event: &Event) -> bool {
|
||||
|
||||
@ -34,6 +34,15 @@ pub struct LogEntry {
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
/// `Mutex<VecDeque>` rather than a lock-free ring because `snapshot()` needs
|
||||
/// an internally-consistent view: it filters by `since_id` + severity, then
|
||||
/// clones matching entries. A seqlock / `ArrayQueue`-based design would
|
||||
/// either require a retry loop that tears across concurrent writes or would
|
||||
/// lose the snapshot API entirely (`ArrayQueue` only supports push/pop, not
|
||||
/// iteration). The write path holds the lock for one `pop_front` +
|
||||
/// `push_back` — micros under load — which the tracing subscriber can
|
||||
/// comfortably pay on the event-emit hot path. Revisit if log volume grows
|
||||
/// past ~10k events/sec per writer, not before.
|
||||
struct LogRingBuffer {
|
||||
entries: Mutex<VecDeque<LogEntry>>,
|
||||
capacity: usize,
|
||||
|
||||
@ -20,6 +20,7 @@ use url::Url;
|
||||
use crate::core::email::scheduler::SmtpClient;
|
||||
use crate::core::playbook_service::ip_version_from_str;
|
||||
use crate::core::soar::engine::SoarEngine;
|
||||
use crate::interface::port::access_control::AccessControlPort;
|
||||
use crate::interface::port::app_repo::AppRepo;
|
||||
use crate::interface::port::notification::AlertPayload;
|
||||
use crate::model::error::Error;
|
||||
@ -192,8 +193,12 @@ impl SoarEngine {
|
||||
}
|
||||
}
|
||||
|
||||
// Block IP via AccessControlPort (handles IPv4/IPv6 dispatch internally)
|
||||
if let Err(e) = self.access_control.block_ip(&event.source_ip) {
|
||||
// Block IP via AccessControlPort (handles IPv4/IPv6 dispatch internally).
|
||||
// Offloaded to `spawn_blocking` so the tokio worker is not stuck while
|
||||
// parking_lot::RwLock + aya map syscalls run — under SOAR burst with
|
||||
// `active_block_count` approaching `max_cap`, this keeps the rest of
|
||||
// the runtime responsive.
|
||||
if let Err(e) = block_ip_blocking(Arc::clone(&self.access_control), event.source_ip.clone()).await {
|
||||
self.decrement_block_count();
|
||||
return Err(e);
|
||||
}
|
||||
@ -216,7 +221,8 @@ impl SoarEngine {
|
||||
.await;
|
||||
if let Err(e) = commit_result {
|
||||
// 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) {
|
||||
let unblock_outcome = unblock_ip_blocking(Arc::clone(&self.access_control), event.source_ip.clone()).await;
|
||||
if let Err(unblock_err) = unblock_outcome {
|
||||
log!(SoarLog::EventHandlingFailed(format!(
|
||||
"CRITICAL: Failed to unblock IP {} after DB error — queueing for retry: {}",
|
||||
event.source_ip, unblock_err
|
||||
@ -540,3 +546,19 @@ async fn insert_pending_unblock_blocking(db: Arc<dyn AppRepo>, source_ip: String
|
||||
.await
|
||||
.map_err(|e| SoarError::ActionFailed("insert_pending_unblock", e))?
|
||||
}
|
||||
|
||||
/// Offload `AccessControlPort::block_ip` onto a blocking thread. The port is
|
||||
/// synchronous because its eBPF-map critical sections are tiny (microseconds),
|
||||
/// but under SOAR burst many tokio workers would contend on the same
|
||||
/// `parking_lot::RwLock` and the aya syscall itself blocks the executor.
|
||||
async fn block_ip_blocking(access_control: Arc<dyn AccessControlPort>, ip: String) -> Result<(), Error> {
|
||||
spawn_blocking(move || access_control.block_ip(&ip))
|
||||
.await
|
||||
.map_err(|e| SoarError::ActionFailed("block_ip", e))?
|
||||
}
|
||||
|
||||
async fn unblock_ip_blocking(access_control: Arc<dyn AccessControlPort>, ip: String) -> Result<(), Error> {
|
||||
spawn_blocking(move || access_control.unblock_ip(&ip))
|
||||
.await
|
||||
.map_err(|e| SoarError::ActionFailed("unblock_ip", e))?
|
||||
}
|
||||
|
||||
@ -425,9 +425,9 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn test_db() -> Arc<dyn AppRepo> {
|
||||
fn test_db() -> Arc<crate::adapter::persistence::Database> {
|
||||
use crate::adapter::persistence::Database;
|
||||
Arc::new(Database::new(":memory:").expect("Failed to create test database")) as Arc<dyn AppRepo>
|
||||
Arc::new(Database::new(":memory:").expect("Failed to create test database"))
|
||||
}
|
||||
|
||||
fn test_engine(ac: Arc<dyn AccessControlPort>) -> SoarEngine {
|
||||
@ -437,7 +437,8 @@ mod tests {
|
||||
db.set_setting("enforce_mode", "enforce").ok();
|
||||
// enforce=2
|
||||
let cache = Arc::new(AtomicU8::new(2));
|
||||
SoarEngine::new(db, ac, None, None, None, cache, None).expect("Failed to create SOAR engine")
|
||||
SoarEngine::new(db as Arc<dyn AppRepo>, ac, None, None, None, cache, None)
|
||||
.expect("Failed to create SOAR engine")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@ -527,7 +528,8 @@ mod tests {
|
||||
db.insert_soar_block_rule("192.168.1.100", 1, &expires).ok();
|
||||
|
||||
let cache = Arc::new(AtomicU8::new(2));
|
||||
let engine = SoarEngine::new(db, mock.clone(), None, None, None, cache, None).expect("Failed to create engine");
|
||||
let engine = SoarEngine::new(db as Arc<dyn AppRepo>, mock.clone(), None, None, None, cache, None)
|
||||
.expect("Failed to create engine");
|
||||
engine.recover_active_blocks().await.expect("Recovery should succeed");
|
||||
|
||||
let blocked = mock.blocked_ips.lock();
|
||||
@ -548,7 +550,8 @@ mod tests {
|
||||
db.insert_soar_block_rule("10.0.0.1", 1, &expires).ok();
|
||||
|
||||
let cache = Arc::new(AtomicU8::new(2));
|
||||
let engine = SoarEngine::new(db, mock.clone(), None, None, None, cache, None).expect("Failed to create engine");
|
||||
let engine = SoarEngine::new(db as Arc<dyn AppRepo>, mock.clone(), None, None, None, cache, None)
|
||||
.expect("Failed to create engine");
|
||||
|
||||
// Should not panic — errors are logged, not propagated
|
||||
let result = engine.recover_active_blocks().await;
|
||||
@ -649,7 +652,8 @@ mod tests {
|
||||
db.insert_admin_whitelist("1.2.3.4").ok();
|
||||
|
||||
let cache = Arc::new(AtomicU8::new(2));
|
||||
let engine = SoarEngine::new(db, mock.clone(), None, None, None, cache, None).expect("Failed to create engine");
|
||||
let engine = SoarEngine::new(db as Arc<dyn AppRepo>, mock.clone(), None, None, None, cache, None)
|
||||
.expect("Failed to create engine");
|
||||
|
||||
let event = ThreatDetectedEvent {
|
||||
source_ip: "1.2.3.4".to_string(),
|
||||
@ -711,7 +715,8 @@ mod tests {
|
||||
db.seed_default_playbooks().ok();
|
||||
|
||||
let cache = Arc::new(AtomicU8::new(0));
|
||||
let engine = SoarEngine::new(db.clone(), mock, None, None, None, cache, None).expect("Failed to create engine");
|
||||
let engine = SoarEngine::new(db.clone() as Arc<dyn AppRepo>, mock, None, None, None, cache, None)
|
||||
.expect("Failed to create engine");
|
||||
|
||||
// Should have loaded default playbooks
|
||||
let count = engine.playbooks.load().len();
|
||||
|
||||
@ -12,7 +12,6 @@ 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.
|
||||
#[allow(dead_code)]
|
||||
pub trait AccessControlAdminPort: Send + Sync {
|
||||
fn add_ipv4_list(&self, direction: FlowDirection, list_type: ListType, address: SocketAddrV4) -> Result<(), Error>;
|
||||
|
||||
|
||||
@ -8,7 +8,6 @@ pub type AclRuleTuple = (u8, String, String, String, u16);
|
||||
/// Owns ACL rules (user-managed block/allow lists) and the admin whitelist that
|
||||
/// SOAR must not block. Kept disjoint from `EnforcementRepo` (rate-limit / DNS /
|
||||
/// geo) so policy tables can evolve independently of packet-matching tables.
|
||||
#[allow(dead_code)]
|
||||
pub trait AclRepo: Send + Sync {
|
||||
fn insert_acl_rule(
|
||||
&self,
|
||||
@ -28,8 +27,6 @@ pub trait AclRepo: Send + Sync {
|
||||
port: u16,
|
||||
) -> Result<(), Error>;
|
||||
|
||||
fn load_acl_rules(&self) -> Result<Vec<AclRuleTuple>, Error>;
|
||||
|
||||
/// Returns true if a manual (non-SOAR) ACL rule exists for this IP.
|
||||
/// Used by the TTL scheduler to avoid removing an eBPF block that the user
|
||||
/// explicitly installed.
|
||||
|
||||
@ -15,14 +15,10 @@ pub struct AuditLogEntry {
|
||||
///
|
||||
/// The append-only constraint is enforced by SQLite triggers
|
||||
/// (`audit_log_no_update` / `audit_log_no_delete`), not by this trait.
|
||||
#[allow(dead_code)]
|
||||
pub trait AuditRepo: Send + Sync {
|
||||
/// Append a new audit entry. `detail` is typically a JSON blob.
|
||||
fn insert_audit_log(&self, actor: &str, action: &str, detail: &str) -> Result<(), Error>;
|
||||
|
||||
/// Read all audit entries ordered by id ASC.
|
||||
fn list_audit_logs(&self) -> Result<Vec<AuditLogEntry>, Error>;
|
||||
|
||||
/// Read audit entries whose `action` exactly matches, newest first,
|
||||
/// capped at `limit`. Drives the fusion explain endpoint, which
|
||||
/// filters on `fused_threat_emitted` rather than walking the full
|
||||
|
||||
@ -4,7 +4,6 @@ use crate::model::error::Error;
|
||||
/// Used by `DnsFilterService` (HTTP-driven CRUD). Kept separate from
|
||||
/// `DnsQueryFilter` (which is the fast-path check) to reflect their distinct
|
||||
/// call sites and latency profiles.
|
||||
#[allow(dead_code)]
|
||||
pub trait DnsFilterPort: Send + Sync {
|
||||
fn add_domain(&self, domain: &str) -> Result<(), Error>;
|
||||
fn remove_domain(&self, domain: &str) -> Result<(), Error>;
|
||||
|
||||
@ -6,19 +6,15 @@ use crate::model::error::Error;
|
||||
/// lifecycle of "data-plane policy that is not per-IP ACL". Kept disjoint from
|
||||
/// `AclRepo` so the per-packet matching rules evolve independently from the
|
||||
/// aggregate policy knobs.
|
||||
#[allow(dead_code)]
|
||||
pub trait EnforcementRepo: Send + Sync {
|
||||
// --- Rate Limit ---
|
||||
fn set_rate_limit(&self, key: &str, value: u64) -> Result<(), Error>;
|
||||
fn load_rate_limit_config(&self) -> Result<Vec<(String, u64)>, Error>;
|
||||
|
||||
// --- DNS ---
|
||||
fn insert_dns_domain(&self, domain: &str) -> Result<(), Error>;
|
||||
fn delete_dns_domain(&self, domain: &str) -> Result<(), Error>;
|
||||
fn load_dns_domains(&self) -> Result<Vec<String>, Error>;
|
||||
|
||||
// --- Geo ---
|
||||
fn insert_geo_country(&self, code: &str) -> Result<(), Error>;
|
||||
fn delete_geo_country(&self, code: &str) -> Result<(), Error>;
|
||||
fn load_geo_countries(&self) -> Result<Vec<String>, Error>;
|
||||
}
|
||||
|
||||
@ -2,7 +2,6 @@ use crate::model::error::Error;
|
||||
|
||||
/// Data-plane geo-block admin port — block / unblock / list country codes.
|
||||
/// Used by `AclService` for the `/api/acl/geo` HTTP routes.
|
||||
#[allow(dead_code)]
|
||||
pub trait GeoBlockPort: Send + Sync {
|
||||
/// Add every ISO-3166-1 alpha-2 code in `codes` to the block set.
|
||||
/// Returns the number of /24 ranges actually added (existing codes
|
||||
|
||||
@ -3,10 +3,6 @@ use crate::model::error::Error;
|
||||
/// Type alias for user record tuples: (id, username, password_hash, role, force_password_change)
|
||||
pub type UserTuple = (i64, String, String, String, bool);
|
||||
|
||||
/// Type alias for user list items: (id, username, role, force_password_change, created_at)
|
||||
#[allow(dead_code)]
|
||||
pub type UserListItem = (i64, String, String, bool, String);
|
||||
|
||||
/// Type alias for user-with-groups: (id, username, role, force_password_change, created_at, groups: Vec<(group_id, group_name)>)
|
||||
pub type UserWithGroups = (i64, String, String, bool, String, Vec<(i64, String)>);
|
||||
|
||||
@ -18,7 +14,6 @@ pub type UserGroupTuple = (i64, String, String, String, String);
|
||||
/// Kept as one aggregate because user lifecycle, group membership, permission
|
||||
/// resolution and login-attempt counters all share the `users` table lifecycle
|
||||
/// and are enforced together at login time.
|
||||
#[allow(dead_code)]
|
||||
pub trait IdentityRepo: Send + Sync {
|
||||
// --- Users ---
|
||||
fn find_user(&self, username: &str) -> Result<Option<UserTuple>, Error>;
|
||||
@ -31,8 +26,6 @@ pub trait IdentityRepo: Send + Sync {
|
||||
force_password_change: bool,
|
||||
) -> Result<i64, Error>;
|
||||
fn update_user_password(&self, user_id: i64, password_hash: &str) -> Result<(), Error>;
|
||||
fn user_count(&self) -> Result<i64, Error>;
|
||||
fn list_users(&self) -> Result<Vec<UserListItem>, Error>;
|
||||
fn list_users_with_groups(&self) -> Result<Vec<UserWithGroups>, Error>;
|
||||
fn delete_user(&self, user_id: i64) -> Result<bool, Error>;
|
||||
fn update_user_role(&self, user_id: i64, role: &str) -> Result<(), Error>;
|
||||
@ -49,7 +42,6 @@ pub trait IdentityRepo: Send + Sync {
|
||||
fn get_user_groups(&self, user_id: i64) -> Result<Vec<(i64, String, String, String)>, Error>;
|
||||
fn set_user_groups(&self, user_id: i64, group_ids: &[i64]) -> Result<(), Error>;
|
||||
fn get_user_permissions(&self, user_id: i64) -> Result<Vec<String>, Error>;
|
||||
fn cleanup_user_memberships(&self, user_id: i64) -> Result<(), Error>;
|
||||
fn get_group_member_ids(&self, group_id: i64) -> Result<Vec<i64>, Error>;
|
||||
fn get_group_members(&self, group_id: i64) -> Result<Vec<(i64, String)>, Error>;
|
||||
|
||||
|
||||
@ -5,7 +5,6 @@ use crate::model::error::Error;
|
||||
/// Split into five per-protocol knobs to match the underlying eBPF per-class
|
||||
/// counters. `RateLimitService` (HTTP CRUD) and `SoarEngine` (the
|
||||
/// adjust-rate-limit action) both depend on this port.
|
||||
#[allow(dead_code)]
|
||||
pub trait RateLimitPort: Send + Sync {
|
||||
fn set_packet_rate(&self, rate: u64) -> Result<(), Error>;
|
||||
fn set_syn_rate(&self, rate: u64) -> Result<(), Error>;
|
||||
|
||||
@ -31,48 +31,19 @@ pub type SoarExecutionRow = (i64, i64, Option<String>, String, String, String);
|
||||
/// writes that block actions depend on live in `SettingRepo` and `AclRepo`
|
||||
/// respectively; cross-aggregate atomicity is handled via
|
||||
/// `DbAdminRepo::with_transaction` + `TxRepos`.
|
||||
#[allow(dead_code)]
|
||||
pub trait SoarRepo: Send + Sync {
|
||||
// --- Playbooks ---
|
||||
fn insert_playbook(
|
||||
&self,
|
||||
name: &str,
|
||||
trigger_event: &str,
|
||||
threshold: Option<f64>,
|
||||
count: Option<i64>,
|
||||
window: Option<i64>,
|
||||
cooldown: i64,
|
||||
) -> Result<i64, Error>;
|
||||
fn insert_playbook_action(
|
||||
&self,
|
||||
playbook_id: i64,
|
||||
action_order: i64,
|
||||
action_type: &str,
|
||||
params_json: &str,
|
||||
) -> Result<i64, Error>;
|
||||
fn load_playbooks_with_actions(&self) -> Result<Vec<PlaybookRow>, Error>;
|
||||
fn update_playbook(&self, id: i64, row: &UpdatePlaybookRow) -> Result<bool, Error>;
|
||||
fn update_playbook_enabled(&self, id: i64, enabled: bool) -> Result<bool, Error>;
|
||||
fn delete_playbook(&self, id: i64) -> Result<bool, Error>;
|
||||
fn delete_playbook_actions(&self, playbook_id: i64) -> Result<(), Error>;
|
||||
fn delete_playbook_conditions(&self, playbook_id: i64) -> Result<(), Error>;
|
||||
fn seed_default_playbooks(&self) -> Result<(), Error>;
|
||||
|
||||
// --- Playbook Conditions ---
|
||||
fn insert_playbook_condition(
|
||||
&self,
|
||||
playbook_id: i64,
|
||||
condition_type: &str,
|
||||
operator: &str,
|
||||
value: &str,
|
||||
value2: Option<&str>,
|
||||
) -> Result<i64, Error>;
|
||||
/// Returns: (condition_id, playbook_id, condition_type, operator, value, value2)
|
||||
#[allow(clippy::type_complexity)]
|
||||
fn load_all_playbook_conditions(&self) -> Result<Vec<(i64, i64, String, String, String, Option<String>)>, Error>;
|
||||
|
||||
// --- Block Rules ---
|
||||
fn insert_soar_block_rule(&self, source_ip: &str, playbook_id: i64, expires_at: &str) -> Result<i64, Error>;
|
||||
fn count_active_soar_blocks(&self) -> Result<u32, Error>;
|
||||
fn get_active_soar_blocks(&self) -> Result<Vec<(i64, String, i64, String)>, Error>;
|
||||
fn get_soar_block_by_id(&self, id: i64) -> Result<Option<(i64, String, i64, String)>, Error>;
|
||||
|
||||
@ -114,7 +114,12 @@ pub fn translate(source: DetectionSource, raw_label: &str) -> CanonicalAttackTyp
|
||||
fn translate_ml(label: &str) -> CanonicalAttackType {
|
||||
match label {
|
||||
"brute force" | "brute_force" | "bruteforce" => CanonicalAttackType::BruteForce,
|
||||
"reconnaissance" | "recon" | "port_scan" | "portscan" => CanonicalAttackType::Reconnaissance,
|
||||
// `port_scan` / `portscan` map to the specific `PortScan` bucket so
|
||||
// ML agreement with Correlation's `scan` / `port_scan` collapses onto
|
||||
// the same dedup key — the precondition for cross-source fusion.
|
||||
// Generic recon labels stay on `Reconnaissance`.
|
||||
"port_scan" | "portscan" => CanonicalAttackType::PortScan,
|
||||
"reconnaissance" | "recon" => CanonicalAttackType::Reconnaissance,
|
||||
"c2 communication" | "c2" | "c2_beacon" | "command_and_control" => CanonicalAttackType::C2Beacon,
|
||||
"dns tunneling" | "dns_tunnel" | "dns tunnel" => CanonicalAttackType::DnsTunnel,
|
||||
"sql injection" | "sql_injection" | "sqli" => CanonicalAttackType::SqlInjection,
|
||||
@ -138,6 +143,11 @@ fn translate_beaconing(_label: &str) -> CanonicalAttackType {
|
||||
}
|
||||
|
||||
/// Correlation (Layer 3 graph) splits across scan / lateral / botnet.
|
||||
///
|
||||
/// Note: Correlation's scan detector looks for fanout on the port dimension,
|
||||
/// so every scan-flavored sub-tag maps to `PortScan`. ML's own `port_scan`
|
||||
/// label uses the same bucket so the dedup orchestrator fuses agreement
|
||||
/// from both sources onto one `(src_ip, port_scan)` key.
|
||||
fn translate_correlation(label: &str) -> CanonicalAttackType {
|
||||
match label {
|
||||
"scan" | "port_scan" | "reconnaissance" => CanonicalAttackType::PortScan,
|
||||
@ -154,8 +164,11 @@ fn translate_correlation(label: &str) -> CanonicalAttackType {
|
||||
fn translate_suricata(label: &str) -> CanonicalAttackType {
|
||||
// classtype strings come lowercase+trimmed from `translate`
|
||||
match label {
|
||||
// Scan / reconnaissance
|
||||
"attempted-recon" | "network-scan" | "misc-activity" => CanonicalAttackType::Reconnaissance,
|
||||
// Port scan specifically — must match ML `port_scan` + Correlation
|
||||
// `scan` on the same canonical key so cross-source fusion fires.
|
||||
"network-scan" => CanonicalAttackType::PortScan,
|
||||
// Broader recon (non-port-scan host discovery, protocol probing)
|
||||
"attempted-recon" | "misc-activity" => CanonicalAttackType::Reconnaissance,
|
||||
// Exploits / admin compromise
|
||||
"attempted-admin" | "successful-admin" | "attempted-user" | "successful-user" | "shellcode-detect"
|
||||
| "attempted-exploit" => CanonicalAttackType::Exploit,
|
||||
@ -305,13 +318,39 @@ mod tests {
|
||||
fn dedup_key_non_collision_across_sources() {
|
||||
// The central invariant: two sources hitting the SAME attack on the
|
||||
// SAME src_ip produce the SAME canonical wire string, so dedup collides.
|
||||
let ml = translate(DetectionSource::ML, "Brute Force");
|
||||
let suricata = translate(DetectionSource::Suricata, "brute-force");
|
||||
assert_eq!(
|
||||
ml.as_str(),
|
||||
suricata.as_str(),
|
||||
"cross-source dedup key MUST match for the same canonical attack",
|
||||
);
|
||||
let pairs = [
|
||||
// ML vs Suricata: brute force
|
||||
(
|
||||
translate(DetectionSource::ML, "Brute Force"),
|
||||
translate(DetectionSource::Suricata, "brute-force"),
|
||||
),
|
||||
// ML vs Correlation: port scan — historically mapped to two
|
||||
// different canonicals (Reconnaissance vs PortScan) until
|
||||
// 2026-04-19. Kept as a regression guard.
|
||||
(
|
||||
translate(DetectionSource::ML, "port_scan"),
|
||||
translate(DetectionSource::Correlation, "scan"),
|
||||
),
|
||||
// Suricata network-scan vs Correlation scan — both land on
|
||||
// `PortScan` so a Suricata scan alert fuses with Correlation's
|
||||
// graph-based scan detection on the same src_ip.
|
||||
(
|
||||
translate(DetectionSource::Suricata, "network-scan"),
|
||||
translate(DetectionSource::Correlation, "scan"),
|
||||
),
|
||||
// ML vs Suricata: C2
|
||||
(
|
||||
translate(DetectionSource::ML, "C2 Communication"),
|
||||
translate(DetectionSource::Suricata, "trojan-activity"),
|
||||
),
|
||||
];
|
||||
for (a, b) in pairs {
|
||||
assert_eq!(
|
||||
a.as_str(),
|
||||
b.as_str(),
|
||||
"cross-source dedup key MUST match for the same canonical attack ({a:?} vs {b:?})",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@ -19,8 +19,8 @@ pub struct FlowStatsEntry {
|
||||
pub last_seen_us: u64,
|
||||
}
|
||||
|
||||
// NOTE: From<&FlowData> impl moved to core/infrastructure/statistics.rs
|
||||
// to maintain the dependency rule: model/ must not import core/
|
||||
// NOTE: From<&FlowData> impl lives in infrastructure/statistics.rs so
|
||||
// model/ doesn't need to import core/ (dependency-rule invariant).
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct StatsSummary {
|
||||
|
||||
@ -34,13 +34,6 @@ pub enum EbpfHealth {
|
||||
},
|
||||
}
|
||||
|
||||
impl EbpfHealth {
|
||||
#[allow(dead_code)] // exposed to HTTP handlers in task #2; keep alongside the type
|
||||
pub fn is_healthy(&self) -> bool {
|
||||
matches!(self, EbpfHealth::Healthy)
|
||||
}
|
||||
}
|
||||
|
||||
/// Which stage of eBPF bring-up failed.
|
||||
#[derive(Debug, Clone, Copy, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
|
||||
@ -4,6 +4,7 @@ use std::sync::OnceLock;
|
||||
|
||||
use tracing::Level;
|
||||
use tracing_appender::rolling::{RollingFileAppender, Rotation};
|
||||
use tracing_subscriber::filter::Directive;
|
||||
use tracing_subscriber::filter::EnvFilter;
|
||||
use tracing_subscriber::fmt::layer as fmt_layer;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
@ -18,6 +19,12 @@ use crate::model::error::io::IOError;
|
||||
/// We erase the complex layered type by boxing the modify closure.
|
||||
static FILTER_HANDLE: OnceLock<Box<dyn FilterControl>> = OnceLock::new();
|
||||
|
||||
/// Snapshot of the per-target directives (e.g. `maxminddb=warn`) in effect
|
||||
/// at `initialize()` time. `set_level` rebuilds the filter from scratch
|
||||
/// around a new root level; reapplying these keeps any RUST_LOG overrides
|
||||
/// the operator configured for specific crates from being silently lost.
|
||||
static PRESERVED_DIRECTIVES: OnceLock<Vec<String>> = OnceLock::new();
|
||||
|
||||
/// Trait to erase the complex generic type of reload::Handle.
|
||||
trait FilterControl: Send + Sync {
|
||||
fn reload_filter(&self, filter: EnvFilter) -> Result<(), String>;
|
||||
@ -68,10 +75,31 @@ impl Logging {
|
||||
Level::INFO
|
||||
});
|
||||
|
||||
let filter = EnvFilter::from_default_env()
|
||||
.add_directive(level.into())
|
||||
// SAFETY: "maxminddb=warn" is a valid tracing directive literal
|
||||
.add_directive("maxminddb=warn".parse().unwrap_or_else(|_| unreachable!()));
|
||||
// Collect per-target directives from RUST_LOG plus our hardcoded
|
||||
// `maxminddb=warn` so `set_level` can reapply them on each rebuild
|
||||
// instead of losing them to `EnvFilter::new(level)`.
|
||||
let mut preserved: Vec<String> = env::var("RUST_LOG")
|
||||
.ok()
|
||||
.map(|raw| {
|
||||
raw.split(',')
|
||||
.map(|s| s.trim().to_string())
|
||||
.filter(|d| !d.is_empty() && d.contains('='))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
if !preserved.iter().any(|d| d == "maxminddb=warn") {
|
||||
preserved.push("maxminddb=warn".to_string());
|
||||
}
|
||||
let _ = PRESERVED_DIRECTIVES.set(preserved);
|
||||
|
||||
let mut filter = EnvFilter::from_default_env().add_directive(level.into());
|
||||
if let Some(directives) = PRESERVED_DIRECTIVES.get() {
|
||||
for d in directives {
|
||||
if let Ok(parsed) = d.parse::<Directive>() {
|
||||
filter = filter.add_directive(parsed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let (filter_layer, reload_handle) = reload::Layer::new(filter);
|
||||
|
||||
@ -99,8 +127,14 @@ impl Logging {
|
||||
)
|
||||
})?;
|
||||
|
||||
let new_filter = EnvFilter::new(parsed_level.to_string())
|
||||
.add_directive("maxminddb=warn".parse().unwrap_or_else(|_| unreachable!()));
|
||||
let mut new_filter = EnvFilter::new(parsed_level.to_string());
|
||||
if let Some(directives) = PRESERVED_DIRECTIVES.get() {
|
||||
for d in directives {
|
||||
if let Ok(parsed) = d.parse::<Directive>() {
|
||||
new_filter = new_filter.add_directive(parsed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
handle.reload_filter(new_filter)?;
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user