From c1fc8393ab1329985818a3ebed4fab1ceee3de90 Mon Sep 17 00:00:00 2001 From: flamboh Date: Thu, 20 Aug 2026 17:49:30 -0700 Subject: [PATCH 01/14] feat(netflow-db): scaffold singularity and feed subcommands --- tools/netflow-db/src/feed.rs | 56 +++++++++++++++++++ tools/netflow-db/src/lib.rs | 2 + tools/netflow-db/src/main.rs | 83 +++++++++++++++++++++++++++-- tools/netflow-db/src/singularity.rs | 39 ++++++++++++++ 4 files changed, 175 insertions(+), 5 deletions(-) create mode 100644 tools/netflow-db/src/feed.rs create mode 100644 tools/netflow-db/src/singularity.rs diff --git a/tools/netflow-db/src/feed.rs b/tools/netflow-db/src/feed.rs new file mode 100644 index 0000000..214b81c --- /dev/null +++ b/tools/netflow-db/src/feed.rs @@ -0,0 +1,56 @@ +//! Continuous Singularity alert feed over an nfcapd capture tree. +//! +//! `netflow-db feed ` is a long-running process: it polls the +//! dataset's capture tree for newly completed five-minute buckets, unions the +//! distinct addresses across the dataset's members for each window, scores +//! them with [`crate::singularity`], and appends threshold-crossing addresses +//! to a rolling alert database (`alerts.sqlite` beside the dataset's product +//! database). Rows older than the retention window are pruned on each pass. +//! +//! The alert database is an ephemeral rolling buffer owned by this module, +//! not a pipeline product database: it carries no dataset identity or +//! coverage semantics and is safe to delete at any time. + +use std::path::PathBuf; + +/// Configuration for one `netflow-db feed` run. +#[derive(Debug)] +pub struct FeedOptions { + /// Dataset id resolved through the datasets registry. + pub dataset_id: String, + /// Registry path override; defaults to standard `datasets.json` discovery. + pub registry_path: Option, + /// Alert database path; defaults to `alerts.sqlite` beside the dataset's + /// configured `db_path`. + pub database_path: Option, + /// nfdump executable (the pinned fork supporting the atlantis contract). + pub nfdump: String, + /// Seconds between capture-tree scans. + pub poll_seconds: u64, + /// Days of alerts to retain. + pub retention_days: u32, + /// Cap on recorded alerts per tail per window. + pub max_per_tail: u32, + /// Alpha at or above which an address alerts; `None` uses the calibrated + /// default. + pub threshold_high: Option, + /// Alpha at or below which an address alerts; `None` uses the calibrated + /// default. + pub threshold_low: Option, + /// Also process historical windows this far back, e.g. `"36h"` or `"7d"`. + pub backfill: Option, + /// Process available windows once and exit instead of polling. + pub once: bool, +} + +#[derive(Debug, thiserror::Error)] +pub enum FeedError { + #[error("feed failure: {0}")] + Other(String), +} + +/// Run the feed until interrupted (or once, with [`FeedOptions::once`]). +pub fn run(options: FeedOptions) -> Result<(), FeedError> { + let _ = options; + todo!("feed loop: implemented by the feed task") +} diff --git a/tools/netflow-db/src/lib.rs b/tools/netflow-db/src/lib.rs index 45e0db6..abfbbb4 100644 --- a/tools/netflow-db/src/lib.rs +++ b/tools/netflow-db/src/lib.rs @@ -10,6 +10,7 @@ pub mod config; pub mod coverage; pub mod domain; pub mod export; +pub mod feed; pub mod ingest; pub mod maad; pub(crate) mod nfdump; @@ -20,5 +21,6 @@ pub mod prepare; pub mod provenance; pub mod publish; pub mod registry; +pub mod singularity; pub mod storage; pub mod verify; diff --git a/tools/netflow-db/src/main.rs b/tools/netflow-db/src/main.rs index 41c1bc6..cc83ec3 100644 --- a/tools/netflow-db/src/main.rs +++ b/tools/netflow-db/src/main.rs @@ -11,13 +11,14 @@ use clap::{Args, Parser, Subcommand, ValueEnum}; use netflow_db::{ compare::{CompareOptions, compare_databases}, export::{ExtractRequest, extract_window, validate_extract_plan}, - maad, + feed, maad, operations::{ UgrAssetKind, scrape_ugr16_urls, select_web_verification_window, verify_web_routes, }, prepare::{PrepareOptions, prepare_archive}, registry::DatasetRegistry, storage::{backup_database, promote_database}, + singularity, verify::{VerifyOptions, verify_database}, }; @@ -48,6 +49,10 @@ enum Command { VerifyWebRoutes(WebVerifyArgs), /// Compute MAAD JSON from IPv4 addresses, one per line. Maad(MaadArgs), + /// Score IPv4 addresses (one per line) by Singularity alpha, as CSV. + Singularity(SingularityArgs), + /// Maintain a rolling Singularity alert feed over live five-minute captures. + Feed(FeedArgs), /// Print the persisted pipeline contract version. ContractVersion, } @@ -235,6 +240,47 @@ struct MaadArgs { input: Option, } +#[derive(Debug, Args)] +struct SingularityArgs { + /// Read addresses from this file instead of standard input. + input: Option, +} + +#[derive(Debug, Args)] +struct FeedArgs { + /// Dataset id from the datasets registry. + dataset: String, + /// Registry path override (defaults to datasets.json discovery). + #[arg(long)] + datasets: Option, + /// Alert database path (defaults to alerts.sqlite beside the dataset's database). + #[arg(long)] + database_path: Option, + #[arg(long, default_value = "nfdump")] + nfdump: String, + /// Seconds between capture-tree scans. + #[arg(long, default_value_t = 30)] + poll_seconds: u64, + /// Days of alerts to retain. + #[arg(long, default_value_t = 7)] + retention_days: u32, + /// Maximum alerts recorded per tail per window. + #[arg(long, default_value_t = 20)] + max_per_tail: u32, + /// Alpha at or above which an address alerts (defaults to the calibrated value). + #[arg(long)] + threshold_high: Option, + /// Alpha at or below which an address alerts (defaults to the calibrated value). + #[arg(long)] + threshold_low: Option, + /// Also process historical windows this far back (e.g. "36h", "7d"). + #[arg(long)] + backfill: Option, + /// Process available windows once and exit instead of polling. + #[arg(long)] + once: bool, +} + fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) @@ -262,6 +308,20 @@ fn main() -> Result<()> { Command::ScrapeUgr16(args) => run_scrape(args)?, Command::VerifyWebRoutes(args) => run_web_verify(args)?, Command::Maad(args) => run_maad(args)?, + Command::Singularity(args) => run_singularity(args)?, + Command::Feed(args) => feed::run(feed::FeedOptions { + dataset_id: args.dataset, + registry_path: args.datasets, + database_path: args.database_path, + nfdump: args.nfdump, + poll_seconds: args.poll_seconds, + retention_days: args.retention_days, + max_per_tail: args.max_per_tail, + threshold_high: args.threshold_high, + threshold_low: args.threshold_low, + backfill: args.backfill, + once: args.once, + })?, Command::ContractVersion => println!("{}", netflow_db::PIPELINE_CONTRACT_VERSION), } Ok(()) @@ -511,7 +571,22 @@ fn run_web_verify(args: WebVerifyArgs) -> Result<()> { } fn run_maad(args: MaadArgs) -> Result<()> { - let input: Box = match args.input { + let addresses = read_ipv4_lines(args.input)?; + maad::write_json(&maad::compute(addresses), io::stdout().lock())?; + io::stdout().flush()?; + Ok(()) +} + +fn run_singularity(args: SingularityArgs) -> Result<()> { + let addresses = read_ipv4_lines(args.input)?; + singularity::write_csv(&singularity::score(addresses), io::stdout().lock())?; + io::stdout().flush()?; + Ok(()) +} + +/// Read IPv4 addresses, one per line, from a file or standard input. +fn read_ipv4_lines(input: Option) -> Result> { + let input: Box = match input { Some(path) => Box::new(BufReader::new( File::open(&path).with_context(|| format!("unable to open {}", path.display()))?, )), @@ -530,9 +605,7 @@ fn run_maad(args: MaadArgs) -> Result<()> { .with_context(|| format!("invalid IPv4 address {value:?}"))?, ); } - maad::write_json(&maad::compute(addresses), io::stdout().lock())?; - io::stdout().flush()?; - Ok(()) + Ok(addresses) } fn parse_boundary(raw: &str, timezone: &str) -> Result { diff --git a/tools/netflow-db/src/singularity.rs b/tools/netflow-db/src/singularity.rs new file mode 100644 index 0000000..6de2bba --- /dev/null +++ b/tools/netflow-db/src/singularity.rs @@ -0,0 +1,39 @@ +//! Per-address Singularity scoring, a port of MAAD's `Singularities.hs`. +//! +//! For each distinct IPv4 address `x`, `alpha(x)` is the OLS slope of +//! `-log2(mu_l(x) / n)` against prefix length `l`, where `mu_l(x)` counts the +//! distinct addresses sharing `x`'s `/l` prefix and `n` is the total distinct +//! address count. Prefix levels stop at the first isolated prefix +//! (`mu == 1`), matching the reference in `vendor/maad/Singularities.hs`. +//! +//! High alpha marks an address in a sparse, isolated region of address +//! space; low alpha marks one that stays inside a dense cluster across many +//! prefix levels. Both tails are anomalous. + +use std::io; +use std::net::Ipv4Addr; + +/// Fitted singularity exponent for one distinct address. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct AddressScore { + pub address: Ipv4Addr, + pub alpha: f64, + pub intercept: f64, + pub r_squared: f64, + /// Number of prefix levels used in the regression. + pub prefix_levels: u8, +} + +/// Score every distinct address in `addresses` (duplicates are ignored). +/// Returns scores sorted by ascending alpha, ties broken by address, matching +/// the reference ordering. +pub fn score(addresses: Vec) -> Vec { + let _ = addresses; + todo!("singularity port: implemented by the scoring task") +} + +/// Write scores as CSV with an `addr,alpha,intercept,r2,n_levels` header. +pub fn write_csv(scores: &[AddressScore], output: impl io::Write) -> io::Result<()> { + let _ = (scores, output); + todo!("singularity port: implemented by the scoring task") +} From 8683354f870b8871479ece7129cd4fe4a053a3dc Mon Sep 17 00:00:00 2001 From: flamboh Date: Thu, 20 Aug 2026 18:13:39 -0700 Subject: [PATCH 02/14] feat(netflow-db): implement singularity scoring and alert feed --- tools/netflow-db/src/feed.rs | 899 +++++++++++++++++++++- tools/netflow-db/src/singularity.rs | 264 ++++++- tools/netflow-db/tests/singularity_cli.rs | 44 ++ 3 files changed, 1193 insertions(+), 14 deletions(-) create mode 100644 tools/netflow-db/tests/singularity_cli.rs diff --git a/tools/netflow-db/src/feed.rs b/tools/netflow-db/src/feed.rs index 214b81c..0a965b5 100644 --- a/tools/netflow-db/src/feed.rs +++ b/tools/netflow-db/src/feed.rs @@ -11,7 +11,66 @@ //! not a pipeline product database: it carries no dataset identity or //! coverage semantics and is safe to delete at any time. -use std::path::PathBuf; +use std::{ + collections::BTreeSet, + fs, + net::{IpAddr, Ipv4Addr}, + path::{Path, PathBuf}, + time::{Duration, Instant}, +}; + +use jiff::Timestamp; +use rusqlite::{Connection, TransactionBehavior, params}; + +use crate::{ + domain::{AddressSide, CanonicalBucket, FlowSelection, IpVersion, Scope, Visibility}, + ingest, + registry::{self, DatasetRegistry}, + singularity, +}; + +const WINDOW_SECONDS: i64 = 300; +const GRACE_SECONDS: i64 = 10 * 60; +const SECONDS_PER_HOUR: i64 = 60 * 60; +const SECONDS_PER_DAY: i64 = 24 * SECONDS_PER_HOUR; + +// This mirrors pipeline.rs's private DEFAULT_TIMEZONE. Registry datasets do +// not carry a timezone, and registry-driven pipeline runs use this default. +const TIMEZONE: &str = "America/Los_Angeles"; + +// TODO: These are uncalibrated placeholders. A future calibration pass will +// replace them with thresholds derived from representative datasets. +const DEFAULT_THRESHOLD_HIGH: f64 = 3.5; +const DEFAULT_THRESHOLD_LOW: f64 = 0.4; + +const ALERT_SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS feed_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS windows ( + window_start INTEGER PRIMARY KEY, + window_end INTEGER NOT NULL, + member_files INTEGER NOT NULL, + address_count INTEGER NOT NULL, + alert_count INTEGER NOT NULL, + alpha_min REAL, + alpha_max REAL, + alpha_median REAL, + processed_at INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS alerts ( + window_start INTEGER NOT NULL REFERENCES windows(window_start) ON DELETE CASCADE, + address TEXT NOT NULL, + alpha REAL NOT NULL, + tail TEXT NOT NULL CHECK (tail IN ('high','low')), + rank INTEGER NOT NULL, + r2 REAL NOT NULL, + prefix_levels INTEGER NOT NULL, + PRIMARY KEY (window_start, tail, rank) +); +CREATE INDEX IF NOT EXISTS alerts_address ON alerts(address, window_start); +"#; /// Configuration for one `netflow-db feed` run. #[derive(Debug)] @@ -31,11 +90,9 @@ pub struct FeedOptions { pub retention_days: u32, /// Cap on recorded alerts per tail per window. pub max_per_tail: u32, - /// Alpha at or above which an address alerts; `None` uses the calibrated - /// default. + /// Alpha at or above which an address alerts; `None` uses the module default. pub threshold_high: Option, - /// Alpha at or below which an address alerts; `None` uses the calibrated - /// default. + /// Alpha at or below which an address alerts; `None` uses the module default. pub threshold_low: Option, /// Also process historical windows this far back, e.g. `"36h"` or `"7d"`. pub backfill: Option, @@ -45,12 +102,836 @@ pub struct FeedOptions { #[derive(Debug, thiserror::Error)] pub enum FeedError { - #[error("feed failure: {0}")] - Other(String), + #[error(transparent)] + Registry(#[from] registry::RegistryError), + #[error("alert database error: {0}")] + Database(#[from] rusqlite::Error), + #[error("feed I/O error: {0}")] + Io(#[from] std::io::Error), + #[error("invalid feed configuration: {0}")] + InvalidConfig(String), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum AlertTail { + High, + Low, +} + +impl AlertTail { + const fn as_str(self) -> &'static str { + match self { + Self::High => "high", + Self::Low => "low", + } + } +} + +#[derive(Clone, Debug, PartialEq)] +struct SelectedAlert { + address: Ipv4Addr, + alpha: f64, + tail: AlertTail, + rank: u32, + r2: f64, + prefix_levels: u8, +} + +#[derive(Debug)] +struct MemberFile { + member_id: String, + path: PathBuf, +} + +#[derive(Debug)] +struct WindowReadiness { + existing_files: Vec, + processable: bool, +} + +struct FeedContext<'a> { + root_path: &'a Path, + members: &'a [String], + nfdump: &'a str, + selection: &'a FlowSelection, + threshold_high: f64, + threshold_low: f64, + max_per_tail: u32, + retention_days: u32, } /// Run the feed until interrupted (or once, with [`FeedOptions::once`]). pub fn run(options: FeedOptions) -> Result<(), FeedError> { - let _ = options; - todo!("feed loop: implemented by the feed task") + let repository_root = std::env::current_dir()?; + let registry = match &options.registry_path { + Some(path) => DatasetRegistry::load(path, &repository_root)?, + None => DatasetRegistry::load_default(&repository_root)?, + }; + let dataset = registry.get(&options.dataset_id)?.clone(); + let members = dataset + .logical_sources()? + .into_iter() + .flat_map(|source| source.members) + .collect::>() + .into_iter() + .collect::>(); + if members.is_empty() { + return Err(FeedError::InvalidConfig(format!( + "dataset {:?} has no capture members", + dataset.dataset_id + ))); + } + + let threshold_high = options.threshold_high.unwrap_or(DEFAULT_THRESHOLD_HIGH); + let threshold_low = options.threshold_low.unwrap_or(DEFAULT_THRESHOLD_LOW); + if !threshold_high.is_finite() || !threshold_low.is_finite() { + return Err(FeedError::InvalidConfig( + "alert thresholds must be finite numbers".into(), + )); + } + let backfill_seconds = options + .backfill + .as_deref() + .map(parse_backfill_duration) + .transpose()?; + + let database_path = options.database_path.clone().unwrap_or_else(|| { + dataset + .db_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .join("alerts.sqlite") + }); + if let Some(parent) = database_path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent)?; + } + + let mut connection = open_alert_database(&database_path)?; + upsert_feed_meta( + &connection, + &dataset.dataset_id, + threshold_high, + threshold_low, + options.max_per_tail, + )?; + + let startup_time = Timestamp::now().as_second(); + // Windows older than the retention cutoff would be pruned in the same + // pass that processed them, so a backfill deeper than retention is + // clamped rather than wasted. + let retention_floor = retention_cutoff(startup_time, options.retention_days); + let requested_start = backfill_seconds + .map(|duration| startup_time.saturating_sub(duration)) + .unwrap_or(startup_time); + if requested_start < retention_floor { + tracing::warn!( + retention_days = options.retention_days, + "backfill exceeds retention; clamping scan start to the retention cutoff" + ); + } + let initial_scan_start = align_window_at_or_after(requested_start.max(retention_floor)); + let selection = FlowSelection::default(); + let context = FeedContext { + root_path: &dataset.root_path, + members: &members, + nfdump: &options.nfdump, + selection: &selection, + threshold_high, + threshold_low, + max_per_tail: options.max_per_tail, + retention_days: options.retention_days, + }; + + loop { + let now = Timestamp::now().as_second(); + process_pass(&mut connection, &context, initial_scan_start, now)?; + prune_windows(&connection, retention_cutoff(now, options.retention_days))?; + + if options.once { + return Ok(()); + } + std::thread::sleep(Duration::from_secs(options.poll_seconds)); + } +} + +fn open_alert_database(path: &Path) -> Result { + let connection = Connection::open(path)?; + init_alert_schema(&connection)?; + Ok(connection) +} + +fn init_alert_schema(connection: &Connection) -> Result<(), FeedError> { + connection.busy_timeout(Duration::from_millis(crate::storage::BUSY_TIMEOUT_MS))?; + connection.pragma_update(None, "foreign_keys", "ON")?; + connection.pragma_update(None, "journal_mode", "WAL")?; + connection.execute_batch(ALERT_SCHEMA)?; + Ok(()) +} + +fn upsert_feed_meta( + connection: &Connection, + dataset_id: &str, + threshold_high: f64, + threshold_low: f64, + max_per_tail: u32, +) -> Result<(), FeedError> { + const UPSERT: &str = " + INSERT INTO feed_meta (key, value) VALUES (?1, ?2) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + "; + for (key, value) in [ + ("schema_version", "1".to_owned()), + ("dataset_id", dataset_id.to_owned()), + ("threshold_high", threshold_high.to_string()), + ("threshold_low", threshold_low.to_string()), + ("max_per_tail", max_per_tail.to_string()), + ] { + connection.execute(UPSERT, params![key, value])?; + } + Ok(()) +} + +fn process_pass( + connection: &mut Connection, + context: &FeedContext<'_>, + initial_scan_start: i64, + now: i64, +) -> Result<(), FeedError> { + let mut window_start = + next_scan_start(connection, initial_scan_start, now, context.retention_days)?; + + loop { + let window_end = window_start.checked_add(WINDOW_SECONDS).ok_or_else(|| { + FeedError::InvalidConfig("candidate window timestamp overflowed".into()) + })?; + if window_end > now { + break; + } + if window_exists(connection, window_start)? { + window_start = window_end; + continue; + } + + let readiness = window_readiness(context.root_path, context.members, window_start, now)?; + if !readiness.processable { + break; + } + + let started_at = Instant::now(); + let mut member_files = 0_i64; + let mut addresses = BTreeSet::new(); + for member_file in readiness.existing_files { + match ingest::read_nfcapd_bucket( + &member_file.path, + &member_file.member_id, + context.selection, + context.nfdump, + TIMEZONE, + ) { + Ok(bucket) => { + member_files += 1; + collect_total_ipv4_addresses(bucket, &mut addresses); + } + Err(error) => tracing::warn!( + path = %member_file.path.display(), + member = %member_file.member_id, + error = %error, + "failed to read feed capture file" + ), + } + } + + if member_files == 0 { + tracing::warn!( + window_start, + window_end, + "no member files could be read for feed window" + ); + window_start = window_end; + continue; + } + + let scores = singularity::score(addresses.into_iter().collect()); + let alerts = select_alerts( + &scores, + context.threshold_high, + context.threshold_low, + context.max_per_tail, + ); + let (alpha_min, alpha_max, alpha_median) = alpha_summary(&scores); + write_window( + connection, + &WindowRecord { + window_start, + window_end, + member_files, + scores: &scores, + alerts: &alerts, + alpha_min, + alpha_max, + alpha_median, + processed_at: now, + }, + )?; + prune_windows(connection, retention_cutoff(now, context.retention_days))?; + + tracing::info!( + window_start, + window_end, + address_count = scores.len(), + alert_count = alerts.len(), + duration_ms = started_at.elapsed().as_millis() as u64, + "processed feed window" + ); + window_start = window_end; + } + + Ok(()) +} + +fn collect_total_ipv4_addresses(bucket: CanonicalBucket, addresses: &mut BTreeSet) { + let total_ipv4_scope = Scope::new(IpVersion::V4, Visibility::All, Visibility::All); + for scoped in bucket.addresses { + if scoped.scope != total_ipv4_scope + || !matches!( + scoped.address_side, + AddressSide::Source | AddressSide::Destination + ) + { + continue; + } + for address in scoped.addresses.iter() { + if let IpAddr::V4(address) = address { + addresses.insert(*address); + } + } + } +} + +/// One processed window's row data, written transactionally by [`write_window`]. +struct WindowRecord<'a> { + window_start: i64, + window_end: i64, + member_files: i64, + scores: &'a [singularity::AddressScore], + alerts: &'a [SelectedAlert], + alpha_min: Option, + alpha_max: Option, + alpha_median: Option, + processed_at: i64, +} + +fn write_window(connection: &mut Connection, record: &WindowRecord) -> Result<(), FeedError> { + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + transaction.execute( + "INSERT INTO windows ( + window_start, window_end, member_files, address_count, alert_count, + alpha_min, alpha_max, alpha_median, processed_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", + params![ + record.window_start, + record.window_end, + record.member_files, + record.scores.len() as i64, + record.alerts.len() as i64, + record.alpha_min, + record.alpha_max, + record.alpha_median, + record.processed_at, + ], + )?; + for alert in record.alerts { + transaction.execute( + "INSERT INTO alerts ( + window_start, address, alpha, tail, rank, r2, prefix_levels + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + record.window_start, + alert.address.to_string(), + alert.alpha, + alert.tail.as_str(), + i64::from(alert.rank), + alert.r2, + i64::from(alert.prefix_levels), + ], + )?; + } + transaction.commit()?; + Ok(()) +} + +fn prune_windows(connection: &Connection, cutoff: i64) -> Result<(), FeedError> { + connection.execute( + "DELETE FROM windows WHERE window_start < ?1", + params![cutoff], + )?; + Ok(()) +} + +fn window_readiness( + root: &Path, + members: &[String], + window_start: i64, + now: i64, +) -> Result { + let mut existing_files = Vec::with_capacity(members.len()); + for member in members { + let path = expected_nfcapd_path(root, member, window_start)?; + if path.exists() { + existing_files.push(MemberFile { + member_id: member.clone(), + path, + }); + } + } + let all_present = existing_files.len() == members.len(); + let window_end = window_start + .checked_add(WINDOW_SECONDS) + .ok_or_else(|| FeedError::InvalidConfig("candidate window timestamp overflowed".into()))?; + let past_grace = now.saturating_sub(window_end) > GRACE_SECONDS; + Ok(WindowReadiness { + existing_files, + processable: all_present || past_grace, + }) +} + +fn expected_nfcapd_path( + root: &Path, + member: &str, + window_start: i64, +) -> Result { + let timestamp = Timestamp::from_second(window_start) + .and_then(|timestamp| timestamp.in_tz(TIMEZONE)) + .map_err(|error| { + FeedError::InvalidConfig(format!( + "invalid feed window timestamp {window_start}: {error}" + )) + })?; + Ok(root + .join(member) + .join(timestamp.strftime("%Y").to_string()) + .join(timestamp.strftime("%m").to_string()) + .join(timestamp.strftime("%d").to_string()) + .join(format!("nfcapd.{}", timestamp.strftime("%Y%m%d%H%M")))) +} + +fn next_scan_start( + connection: &Connection, + initial_scan_start: i64, + now: i64, + retention_days: u32, +) -> Result { + let newest = connection.query_row("SELECT MAX(window_start) FROM windows", [], |row| { + row.get::<_, Option>(0) + })?; + match newest { + Some(window_start) => { + let after_newest = window_start.checked_add(WINDOW_SECONDS).ok_or_else(|| { + FeedError::InvalidConfig("processed window timestamp overflowed".into()) + })?; + Ok(after_newest.max(align_window_at_or_after(retention_cutoff( + now, + retention_days, + )))) + } + None => Ok(initial_scan_start), + } +} + +fn window_exists(connection: &Connection, window_start: i64) -> Result { + Ok(connection.query_row( + "SELECT EXISTS(SELECT 1 FROM windows WHERE window_start = ?1)", + params![window_start], + |row| row.get(0), + )?) +} + +fn retention_cutoff(now: i64, retention_days: u32) -> i64 { + now.saturating_sub(i64::from(retention_days) * SECONDS_PER_DAY) +} + +fn align_window_at_or_after(timestamp: i64) -> i64 { + let remainder = timestamp.rem_euclid(WINDOW_SECONDS); + if remainder == 0 { + timestamp + } else { + timestamp.saturating_add(WINDOW_SECONDS - remainder) + } +} + +fn parse_backfill_duration(value: &str) -> Result { + let Some((unit_index, unit)) = value.char_indices().last() else { + return Err(invalid_backfill(value)); + }; + let amount = &value[..unit_index]; + if amount.is_empty() || !amount.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(invalid_backfill(value)); + } + let amount = amount.parse::().map_err(|_| invalid_backfill(value))?; + let unit_seconds = match unit { + 'h' => SECONDS_PER_HOUR, + 'd' => SECONDS_PER_DAY, + _ => return Err(invalid_backfill(value)), + }; + amount.checked_mul(unit_seconds).ok_or_else(|| { + FeedError::InvalidConfig(format!("backfill duration {value:?} is too large")) + }) +} + +fn invalid_backfill(value: &str) -> FeedError { + FeedError::InvalidConfig(format!( + "invalid backfill duration {value:?}; expected an integer followed by 'h' or 'd'" + )) +} + +fn alpha_summary(scores: &[singularity::AddressScore]) -> (Option, Option, Option) { + if scores.is_empty() { + return (None, None, None); + } + let mut alphas = scores.iter().map(|score| score.alpha).collect::>(); + alphas.sort_by(f64::total_cmp); + let median_index = alphas.len() / 2; + let median = if alphas.len() % 2 == 0 { + alphas[median_index - 1] / 2.0 + alphas[median_index] / 2.0 + } else { + alphas[median_index] + }; + ( + alphas.first().copied(), + alphas.last().copied(), + Some(median), + ) +} + +fn select_alerts( + scores: &[singularity::AddressScore], + threshold_high: f64, + threshold_low: f64, + max_per_tail: u32, +) -> Vec { + let mut high = scores + .iter() + .filter(|score| score.alpha >= threshold_high) + .collect::>(); + high.sort_by(|left, right| { + right + .alpha + .total_cmp(&left.alpha) + .then_with(|| left.address.cmp(&right.address)) + }); + + let mut low = scores + .iter() + .filter(|score| score.alpha <= threshold_low) + .collect::>(); + low.sort_by(|left, right| { + left.alpha + .total_cmp(&right.alpha) + .then_with(|| left.address.cmp(&right.address)) + }); + + let limit = max_per_tail as usize; + high.into_iter() + .take(limit) + .enumerate() + .map(|(index, score)| selected_alert(score, AlertTail::High, index)) + .chain( + low.into_iter() + .take(limit) + .enumerate() + .map(|(index, score)| selected_alert(score, AlertTail::Low, index)), + ) + .collect() +} + +fn selected_alert( + score: &singularity::AddressScore, + tail: AlertTail, + index: usize, +) -> SelectedAlert { + SelectedAlert { + address: score.address, + alpha: score.alpha, + tail, + rank: index as u32 + 1, + r2: score.r_squared, + prefix_levels: score.prefix_levels, + } +} + +#[cfg(test)] +mod tests { + use std::fs; + + use tempfile::tempdir; + + use super::*; + + #[test] + fn schema_initialization_is_idempotent() { + let temporary = tempdir().unwrap(); + let database_path = temporary.path().join("alerts.sqlite"); + + { + let connection = Connection::open(&database_path).unwrap(); + init_alert_schema(&connection).unwrap(); + init_alert_schema(&connection).unwrap(); + } + let connection = Connection::open(&database_path).unwrap(); + init_alert_schema(&connection).unwrap(); + + let tables = connection + .prepare( + "SELECT name FROM sqlite_master + WHERE type = 'table' AND name IN ('feed_meta', 'windows', 'alerts') + ORDER BY name", + ) + .unwrap() + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(tables, vec!["alerts", "feed_meta", "windows"]); + + let alert_columns = connection + .prepare("PRAGMA table_info(alerts)") + .unwrap() + .query_map([], |row| row.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!( + alert_columns, + vec![ + "window_start", + "address", + "alpha", + "tail", + "rank", + "r2", + "prefix_levels" + ] + ); + } + + #[test] + fn pruning_removes_old_windows_and_their_alerts() { + let temporary = tempdir().unwrap(); + let connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap(); + insert_window(&connection, 100); + insert_window(&connection, 1_000); + for window_start in [100, 1_000] { + connection + .execute( + "INSERT INTO alerts + (window_start, address, alpha, tail, rank, r2, prefix_levels) + VALUES (?1, '192.0.2.1', 4.0, 'high', 1, 0.9, 12)", + params![window_start], + ) + .unwrap(); + } + + prune_windows(&connection, 1_000).unwrap(); + + let windows = query_i64_column(&connection, "SELECT window_start FROM windows"); + let alerts = query_i64_column(&connection, "SELECT window_start FROM alerts"); + assert_eq!(windows, vec![1_000]); + assert_eq!(alerts, vec![1_000]); + } + + #[test] + fn parses_hour_and_day_backfills() { + assert_eq!( + parse_backfill_duration("36h").unwrap(), + 36 * SECONDS_PER_HOUR + ); + assert_eq!(parse_backfill_duration("7d").unwrap(), 7 * SECONDS_PER_DAY); + } + + #[test] + fn rejects_malformed_backfills() { + for value in ["", "36", "h", "1.5h", "-2d", "7w", " 7d"] { + assert!( + parse_backfill_duration(value).is_err(), + "accepted {value:?}" + ); + } + } + + #[test] + fn window_readiness_obeys_file_completeness_and_grace_period() { + let temporary = tempdir().unwrap(); + let members = vec!["alpha".to_owned(), "beta".to_owned()]; + let complete_start = 1_700_000_100; + for member in &members { + create_empty_capture(temporary.path(), member, complete_start); + } + + let complete = window_readiness( + temporary.path(), + &members, + complete_start, + complete_start + WINDOW_SECONDS, + ) + .unwrap(); + assert!(complete.processable); + assert_eq!(complete.existing_files.len(), 2); + + let partial_start = complete_start + WINDOW_SECONDS; + create_empty_capture(temporary.path(), &members[0], partial_start); + let before_grace = window_readiness( + temporary.path(), + &members, + partial_start, + partial_start + WINDOW_SECONDS + GRACE_SECONDS, + ) + .unwrap(); + assert!(!before_grace.processable); + assert_eq!(before_grace.existing_files.len(), 1); + + let after_grace = window_readiness( + temporary.path(), + &members, + partial_start, + partial_start + WINDOW_SECONDS + GRACE_SECONDS + 1, + ) + .unwrap(); + assert!(after_grace.processable); + assert_eq!(after_grace.existing_files.len(), 1); + } + + #[cfg(unix)] + #[test] + fn process_pass_skips_missing_window_and_processes_later_window() { + let temporary = tempdir().unwrap(); + let root = temporary.path().join("captures"); + let members = vec!["alpha".to_owned(), "beta".to_owned()]; + let missing_start = 1_700_000_100; + let populated_start = missing_start + WINDOW_SECONDS; + for member in &members { + create_empty_capture(&root, member, populated_start); + } + + let decoder = temporary.path().join("fake-nfdump"); + write_fake_nfdump(&decoder, ""); + let mut connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap(); + let selection = FlowSelection::default(); + let context = FeedContext { + root_path: &root, + members: &members, + nfdump: decoder.to_str().unwrap(), + selection: &selection, + threshold_high: DEFAULT_THRESHOLD_HIGH, + threshold_low: DEFAULT_THRESHOLD_LOW, + max_per_tail: 0, + retention_days: 1, + }; + let now = missing_start + WINDOW_SECONDS + GRACE_SECONDS + 1; + + process_pass(&mut connection, &context, missing_start, now).unwrap(); + + assert!(!window_exists(&connection, missing_start).unwrap()); + assert!(window_exists(&connection, populated_start).unwrap()); + } + + #[test] + fn resume_starts_after_the_newest_window_and_respects_retention_floor() { + let temporary = tempdir().unwrap(); + let connection = open_alert_database(&temporary.path().join("alerts.sqlite")).unwrap(); + insert_window(&connection, 900); + + assert_eq!(next_scan_start(&connection, 0, 2_000, 7).unwrap(), 1_200); + + connection.execute("DELETE FROM windows", []).unwrap(); + insert_window(&connection, 0); + let now = 10 * SECONDS_PER_DAY; + let retention_floor = align_window_at_or_after(now - SECONDS_PER_DAY); + assert_eq!( + next_scan_start(&connection, 0, now, 1).unwrap(), + retention_floor + ); + } + + #[test] + fn alert_selection_caps_and_ranks_both_tails_deterministically() { + let scores = vec![ + score_fixture([192, 0, 2, 4], 5.0), + score_fixture([192, 0, 2, 2], 5.0), + score_fixture([192, 0, 2, 3], 4.0), + score_fixture([192, 0, 2, 9], -0.2), + score_fixture([192, 0, 2, 7], 0.1), + score_fixture([192, 0, 2, 8], 0.3), + score_fixture([192, 0, 2, 6], 1.0), + ]; + + let alerts = select_alerts(&scores, 3.5, 0.4, 2); + assert_eq!(alerts.len(), 4); + assert_eq!( + alerts + .iter() + .map(|alert| (alert.tail, alert.rank, alert.address, alert.alpha)) + .collect::>(), + vec![ + (AlertTail::High, 1, Ipv4Addr::new(192, 0, 2, 2), 5.0), + (AlertTail::High, 2, Ipv4Addr::new(192, 0, 2, 4), 5.0), + (AlertTail::Low, 1, Ipv4Addr::new(192, 0, 2, 9), -0.2), + (AlertTail::Low, 2, Ipv4Addr::new(192, 0, 2, 7), 0.1), + ] + ); + } + + fn insert_window(connection: &Connection, window_start: i64) { + connection + .execute( + "INSERT INTO windows ( + window_start, window_end, member_files, address_count, alert_count, + alpha_min, alpha_max, alpha_median, processed_at + ) VALUES (?1, ?2, 1, 1, 1, 1.0, 1.0, 1.0, ?1)", + params![window_start, window_start + WINDOW_SECONDS], + ) + .unwrap(); + } + + fn query_i64_column(connection: &Connection, query: &str) -> Vec { + connection + .prepare(query) + .unwrap() + .query_map([], |row| row.get(0)) + .unwrap() + .collect::, _>>() + .unwrap() + } + + fn create_empty_capture(root: &Path, member: &str, window_start: i64) { + let path = expected_nfcapd_path(root, member, window_start).unwrap(); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, []).unwrap(); + } + + #[cfg(unix)] + fn write_fake_nfdump(executable: &std::path::Path, setup: &str) { + use std::os::unix::fs::PermissionsExt; + + let stream = executable.with_extension("stream"); + fs::write(&stream, crate::nfdump::ONE_V4_TEST_STREAM).unwrap(); + fs::write( + executable, + format!("#!/bin/sh\n{setup}\ncat '{}'\n", stream.display()), + ) + .unwrap(); + fs::set_permissions(executable, fs::Permissions::from_mode(0o755)).unwrap(); + } + + fn score_fixture(address: [u8; 4], alpha: f64) -> singularity::AddressScore { + singularity::AddressScore { + address: Ipv4Addr::from(address), + alpha, + intercept: 0.0, + r_squared: 0.95, + prefix_levels: 12, + } + } } diff --git a/tools/netflow-db/src/singularity.rs b/tools/netflow-db/src/singularity.rs index 6de2bba..5732a43 100644 --- a/tools/netflow-db/src/singularity.rs +++ b/tools/netflow-db/src/singularity.rs @@ -24,16 +24,270 @@ pub struct AddressScore { pub prefix_levels: u8, } +#[derive(Clone, Copy, Debug, Default)] +struct RunningOls { + count: f64, + sum_x: f64, + sum_y: f64, + sum_xx: f64, + sum_xy: f64, + sum_yy: f64, +} + +impl RunningOls { + fn add(self, x: f64, y: f64) -> Self { + Self { + count: self.count + 1.0, + sum_x: self.sum_x + x, + sum_y: self.sum_y + y, + sum_xx: self.sum_xx + x * x, + sum_xy: self.sum_xy + x * y, + sum_yy: self.sum_yy + y * y, + } + } + + /// Fits the accumulated points without branches for degenerate inputs. + /// + /// Zero or one point, and a constant ordinate across two or more points, + /// leave at least one zero denominator. IEEE-754 division therefore + /// produces `NaN`, matching MAAD's rank-deficient OLS behavior. These + /// values are intentional and must not be replaced with fallback scores. + fn fit(self) -> Regression { + let mean_x = self.sum_x / self.count; + let mean_y = self.sum_y / self.count; + let sxx = self.sum_xx - self.count * mean_x * mean_x; + let sxy = self.sum_xy - self.count * mean_x * mean_y; + let syy = self.sum_yy - self.count * mean_y * mean_y; + let alpha = sxy / sxx; + + Regression { + alpha, + intercept: mean_y - alpha * mean_x, + r_squared: (sxy * sxy) / (sxx * syy), + } + } +} + +#[derive(Clone, Copy, Debug)] +struct Regression { + alpha: f64, + intercept: f64, + r_squared: f64, +} + /// Score every distinct address in `addresses` (duplicates are ignored). /// Returns scores sorted by ascending alpha, ties broken by address, matching /// the reference ordering. pub fn score(addresses: Vec) -> Vec { - let _ = addresses; - todo!("singularity port: implemented by the scoring task") + let mut addresses: Vec = addresses.into_iter().map(u32::from).collect(); + addresses.sort_unstable(); + addresses.dedup(); + + if addresses.is_empty() { + return Vec::new(); + } + + let mut scores = Vec::with_capacity(addresses.len()); + visit_prefix( + &addresses, + 0, + RunningOls::default(), + &mut scores, + addresses.len() as f64, + ); + scores.sort_by(|left, right| { + left.alpha + .total_cmp(&right.alpha) + .then_with(|| left.address.cmp(&right.address)) + }); + scores +} + +fn visit_prefix( + addresses: &[u32], + level: u8, + running: RunningOls, + scores: &mut Vec, + total: f64, +) { + if addresses.len() == 1 { + let regression = running.fit(); + scores.push(AddressScore { + address: Ipv4Addr::from(addresses[0]), + alpha: regression.alpha, + intercept: regression.intercept, + r_squared: regression.r_squared, + prefix_levels: level, + }); + return; + } + + let ordinate = -((addresses.len() as f64) / total).log2(); + let running = running.add(f64::from(level), ordinate); + + if level < 32 { + let bit = 31 - u32::from(level); + let split = addresses.partition_point(|address| (address >> bit) & 1 == 0); + + if split > 0 { + visit_prefix(&addresses[..split], level + 1, running, scores, total); + } + if split < addresses.len() { + visit_prefix(&addresses[split..], level + 1, running, scores, total); + } + } } /// Write scores as CSV with an `addr,alpha,intercept,r2,n_levels` header. -pub fn write_csv(scores: &[AddressScore], output: impl io::Write) -> io::Result<()> { - let _ = (scores, output); - todo!("singularity port: implemented by the scoring task") +pub fn write_csv(scores: &[AddressScore], mut output: impl io::Write) -> io::Result<()> { + writeln!(output, "addr,alpha,intercept,r2,n_levels")?; + for score in scores { + writeln!( + output, + "{},{},{},{},{}", + score.address, score.alpha, score.intercept, score.r_squared, score.prefix_levels + )?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + const EPSILON: f64 = 1e-9; + + #[test] + fn scores_hand_verified_prefix_fixture() { + let scores = score(vec![ + Ipv4Addr::new(0, 0, 0, 0), + Ipv4Addr::new(32, 0, 0, 0), + Ipv4Addr::new(128, 0, 0, 0), + Ipv4Addr::new(192, 0, 0, 0), + ]); + + let expected = [ + (Ipv4Addr::new(0, 0, 0, 0), 0.5, 1.0 / 6.0, 0.75, 3), + (Ipv4Addr::new(32, 0, 0, 0), 0.5, 1.0 / 6.0, 0.75, 3), + (Ipv4Addr::new(128, 0, 0, 0), 1.0, 0.0, 1.0, 2), + (Ipv4Addr::new(192, 0, 0, 0), 1.0, 0.0, 1.0, 2), + ]; + + assert_eq!(scores.len(), expected.len()); + for (actual, (address, alpha, intercept, r_squared, prefix_levels)) in + scores.iter().zip(expected) + { + assert_eq!(actual.address, address); + assert_close(actual.alpha, alpha); + assert_close(actual.intercept, intercept); + assert_close(actual.r_squared, r_squared); + assert_eq!(actual.prefix_levels, prefix_levels); + } + } + + #[test] + fn ignores_duplicate_addresses() { + let address = Ipv4Addr::new(10, 0, 0, 1); + let duplicate_only = score(vec![address, address]); + assert_eq!(duplicate_only.len(), 1); + assert_eq!(duplicate_only[0].address, address); + + let deduped = vec![ + Ipv4Addr::new(0, 0, 0, 0), + Ipv4Addr::new(32, 0, 0, 0), + Ipv4Addr::new(128, 0, 0, 0), + Ipv4Addr::new(192, 0, 0, 0), + ]; + let mut with_duplicate = deduped.clone(); + with_duplicate.insert(2, Ipv4Addr::new(32, 0, 0, 0)); + + assert_eq!(score(with_duplicate), score(deduped)); + } + + #[test] + fn empty_input_has_no_scores() { + assert!(score(Vec::new()).is_empty()); + } + + #[test] + fn single_address_has_degenerate_regression() { + let address = Ipv4Addr::new(203, 0, 113, 7); + let scores = score(vec![address]); + + assert_eq!(scores.len(), 1); + assert_eq!(scores[0].address, address); + assert_eq!(scores[0].prefix_levels, 0); + assert!(scores[0].alpha.is_nan()); + assert!(scores[0].intercept.is_nan()); + assert!(scores[0].r_squared.is_nan()); + } + + #[test] + fn sorts_three_alpha_groups_before_breaking_ties_by_address() { + let scores = score(vec![ + Ipv4Addr::new(192, 0, 0, 0), + Ipv4Addr::new(16, 0, 0, 0), + Ipv4Addr::new(128, 0, 0, 0), + Ipv4Addr::new(0, 0, 0, 0), + Ipv4Addr::new(32, 0, 0, 0), + ]); + + assert_eq!( + scores.iter().map(|entry| entry.address).collect::>(), + vec![ + Ipv4Addr::new(32, 0, 0, 0), + Ipv4Addr::new(0, 0, 0, 0), + Ipv4Addr::new(16, 0, 0, 0), + Ipv4Addr::new(128, 0, 0, 0), + Ipv4Addr::new(192, 0, 0, 0), + ] + ); + assert!(scores[0].alpha < scores[1].alpha); + assert_close(scores[1].alpha, scores[2].alpha); + assert!(scores[2].alpha < scores[3].alpha); + assert_close(scores[3].alpha, scores[4].alpha); + } + + #[test] + fn writes_scores_as_csv() { + let scores = score(vec![ + Ipv4Addr::new(0, 0, 0, 0), + Ipv4Addr::new(32, 0, 0, 0), + Ipv4Addr::new(128, 0, 0, 0), + Ipv4Addr::new(192, 0, 0, 0), + ]); + let mut csv = Vec::new(); + + write_csv(&scores, &mut csv).unwrap(); + + let csv = String::from_utf8(csv).unwrap(); + let mut lines = csv.lines(); + assert_eq!(lines.next(), Some("addr,alpha,intercept,r2,n_levels")); + + let expected = [ + ("0.0.0.0", 0.5, 1.0 / 6.0, 0.75, 3), + ("32.0.0.0", 0.5, 1.0 / 6.0, 0.75, 3), + ("128.0.0.0", 1.0, 0.0, 1.0, 2), + ("192.0.0.0", 1.0, 0.0, 1.0, 2), + ]; + for (line, (address, alpha, intercept, r_squared, prefix_levels)) in + lines.by_ref().zip(expected) + { + let fields: Vec<_> = line.split(',').collect(); + assert_eq!(fields.len(), 5); + assert_eq!(fields[0], address); + assert_close(fields[1].parse().unwrap(), alpha); + assert_close(fields[2].parse().unwrap(), intercept); + assert_close(fields[3].parse().unwrap(), r_squared); + assert_eq!(fields[4].parse::().unwrap(), prefix_levels); + } + assert_eq!(lines.next(), None); + } + + fn assert_close(actual: f64, expected: f64) { + assert!( + (actual - expected).abs() <= EPSILON, + "actual={actual:?}, expected={expected:?}" + ); + } } diff --git a/tools/netflow-db/tests/singularity_cli.rs b/tools/netflow-db/tests/singularity_cli.rs new file mode 100644 index 0000000..966624c --- /dev/null +++ b/tools/netflow-db/tests/singularity_cli.rs @@ -0,0 +1,44 @@ +use std::{fs, process::Command}; + +const EPSILON: f64 = 1e-9; +const FIXTURE: &str = "0.0.0.0\n32.0.0.0\n128.0.0.0\n192.0.0.0\n"; + +#[test] +fn singularity_file_input_emits_scores_as_csv() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("addresses.txt"); + fs::write(&input, FIXTURE).unwrap(); + + let output = Command::new(env!("CARGO_BIN_EXE_netflow-db")) + .args(["singularity"]) + .arg(&input) + .output() + .unwrap(); + assert!( + output.status.success(), + "stdout={}\nstderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).unwrap(); + let mut lines = stdout.lines(); + assert_eq!(lines.next(), Some("addr,alpha,intercept,r2,n_levels")); + + let rows: Vec<_> = lines.collect(); + assert_eq!(rows.len(), 4); + let first: Vec<_> = rows[0].split(',').collect(); + assert_eq!(first.len(), 5); + assert_eq!(first[0], "0.0.0.0"); + assert_close(first[1].parse().unwrap(), 0.5); + assert_close(first[2].parse().unwrap(), 1.0 / 6.0); + assert_close(first[3].parse().unwrap(), 0.75); + assert_eq!(first[4], "3"); +} + +fn assert_close(actual: f64, expected: f64) { + assert!( + (actual - expected).abs() <= EPSILON, + "actual={actual:?}, expected={expected:?}" + ); +} From af37f524adb3d06d2d5f508d1ae166b129117d40 Mon Sep 17 00:00:00 2001 From: flamboh Date: Thu, 20 Aug 2026 18:14:07 -0700 Subject: [PATCH 03/14] feat(web): alert feed page and API --- apps/web/src/lib/server/alerts.ts | 370 ++++++++++++++++++ apps/web/src/lib/types/types.ts | 32 ++ apps/web/src/routes/+layout.svelte | 5 + apps/web/src/routes/alerts/+page.svelte | 453 ++++++++++++++++++++++ apps/web/src/routes/alerts/+page.ts | 40 ++ apps/web/src/routes/api/alerts/+server.ts | 61 +++ apps/web/tests/lib/server/alerts.test.ts | 277 +++++++++++++ apps/web/tests/routes/api-alerts.test.ts | 126 ++++++ 8 files changed, 1364 insertions(+) create mode 100644 apps/web/src/lib/server/alerts.ts create mode 100644 apps/web/src/routes/alerts/+page.svelte create mode 100644 apps/web/src/routes/alerts/+page.ts create mode 100644 apps/web/src/routes/api/alerts/+server.ts create mode 100644 apps/web/tests/lib/server/alerts.test.ts create mode 100644 apps/web/tests/routes/api-alerts.test.ts diff --git a/apps/web/src/lib/server/alerts.ts b/apps/web/src/lib/server/alerts.ts new file mode 100644 index 0000000..38cf672 --- /dev/null +++ b/apps/web/src/lib/server/alerts.ts @@ -0,0 +1,370 @@ +import { env as privateEnv } from '$env/dynamic/private'; +import type Database from 'better-sqlite3'; +import type { + AlertFeedAlert, + AlertFeedWindow, + AlertsFeedResponse, + AlertTail +} from '$lib/types/types'; + +type QueryParam = string | number | boolean | null | Uint8Array; + +type SqliteClient = Database.Database; + +type LocalDbIdentity = { + device: number; + inode: number; +}; + +type LocalDbCacheEntry = { + db: SqliteClient; + identity: LocalDbIdentity; +}; + +type FeedMetaRow = { + key: string; + value: string; +}; + +type WindowRow = { + windowStart: number; + windowEnd: number; + addressCount: number; + alertCount: number; +}; + +type LatestWindowRow = { + windowStart: number; + processedAt: number; +}; + +const DEFAULT_LIMIT_WINDOWS = 24; +const MAX_LIMIT_WINDOWS = 288; +const REQUIRED_TABLES = ['alerts', 'feed_meta', 'windows'] as const; +const REQUIRED_META_KEYS = [ + 'schema_version', + 'dataset_id', + 'threshold_high', + 'threshold_low', + 'max_per_tail' +] as const; + +const localDbCache = new Map(); + +export type AlertsFeedOptions = { + platform?: App.Platform; + tail?: AlertTail; + limitWindows?: number; + before?: number; +}; + +function absentFeed(): AlertsFeedResponse { + return { feed: { present: false }, windows: [] }; +} + +function getEnv(name: string): string | undefined { + return globalThis.process?.env?.[name]?.trim() || privateEnv[name]?.trim() || undefined; +} + +function shouldUseD1(platform?: App.Platform): boolean { + return getEnv('ATLANTIS_DB_DRIVER') !== 'sqlite' && Boolean(platform?.env.DB); +} + +async function resolvePath(value: string): Promise { + if (value === ':memory:') { + return value; + } + + const path = await import('node:path'); + return path.isAbsolute(value) ? value : path.resolve(process.cwd(), value); +} + +async function discoverLocalSqlitePaths(): Promise { + const configured = getEnv('LOCAL_SQLITE_PATH') ?? getEnv('DATABASE_PATH'); + if (configured) { + return [await resolvePath(configured)]; + } + + const fs = await import('node:fs/promises'); + const path = await import('node:path'); + const roots = [path.resolve(process.cwd(), 'data'), path.resolve(process.cwd(), '../../data')]; + const dbPaths = new Set(); + + for (const root of roots) { + let entries: import('node:fs').Dirent[]; + try { + entries = await fs.readdir(root, { withFileTypes: true }); + } catch { + continue; + } + + for (const entry of entries) { + if (!entry.isDirectory()) { + continue; + } + + const dbPath = path.join(root, entry.name, 'netflow.sqlite'); + try { + const stat = await fs.stat(dbPath); + if (stat.isFile()) { + dbPaths.add(dbPath); + } + } catch { + // Not every data directory is a web dataset. + } + } + } + + return [...dbPaths].sort(); +} + +async function openReadonlyClient(dbPath: string): Promise { + const betterSqlite3 = await import(/* @vite-ignore */ 'better-sqlite3'); + const sqlite = new betterSqlite3.default(dbPath, { readonly: true, fileMustExist: true }); + sqlite.pragma('query_only = ON'); + sqlite.pragma('busy_timeout = 60000'); + return sqlite; +} + +async function resolveDatasetDirectory(datasetId: string): Promise { + const path = await import('node:path'); + for (const dbPath of await discoverLocalSqlitePaths()) { + if (dbPath === ':memory:') { + continue; + } + + let db: SqliteClient | undefined; + try { + db = await openReadonlyClient(dbPath); + const row = db.prepare('SELECT id FROM datasets WHERE id = ? LIMIT 1').get(datasetId) as + | { id: string } + | undefined; + if (row) { + return path.dirname(dbPath); + } + } catch { + // A broken candidate cannot contain a usable local dataset. + } finally { + db?.close(); + } + } + + return null; +} + +async function localDbIdentity(dbPath: string): Promise { + const fs = await import('node:fs/promises'); + const stat = await fs.stat(dbPath); + return { device: stat.dev, inode: stat.ino }; +} + +function sameLocalDbIdentity(left: LocalDbIdentity, right: LocalDbIdentity): boolean { + return left.device === right.device && left.inode === right.inode; +} + +function evictLocalDb(dbPath: string): void { + const existing = localDbCache.get(dbPath); + if (!existing) { + return; + } + + localDbCache.delete(dbPath); + existing.db.close(); +} + +async function createLocalDb(dbPath: string): Promise { + for (let attempt = 0; attempt < 3; attempt += 1) { + const identityBeforeOpen = await localDbIdentity(dbPath); + const db = await openReadonlyClient(dbPath); + try { + const identityAfterOpen = await localDbIdentity(dbPath); + if (sameLocalDbIdentity(identityBeforeOpen, identityAfterOpen)) { + return { db, identity: identityAfterOpen }; + } + } catch (error) { + db.close(); + throw error; + } + db.close(); + } + + throw new Error(`Local alerts database kept changing while opening: ${dbPath}`); +} + +async function getLocalDb(dbPath: string): Promise { + try { + const identity = await localDbIdentity(dbPath); + const existing = localDbCache.get(dbPath); + if (existing && sameLocalDbIdentity(existing.identity, identity)) { + return existing.db; + } + + evictLocalDb(dbPath); + const entry = await createLocalDb(dbPath); + localDbCache.set(dbPath, entry); + return entry.db; + } catch (error) { + evictLocalDb(dbPath); + throw error; + } +} + +function clampLimitWindows(limitWindows: number | undefined): number { + if (limitWindows === undefined || !Number.isFinite(limitWindows)) { + return DEFAULT_LIMIT_WINDOWS; + } + return Math.min(MAX_LIMIT_WINDOWS, Math.max(1, Math.trunc(limitWindows))); +} + +function readFeedMetadata( + db: SqliteClient, + datasetId: string +): { thresholdHigh: number; thresholdLow: number } { + const tables = db + .prepare( + `SELECT name FROM sqlite_master WHERE type = 'table' AND name IN (${REQUIRED_TABLES.map(() => '?').join(', ')})` + ) + .all(...REQUIRED_TABLES) as { name: string }[]; + if (new Set(tables.map((row) => row.name)).size !== REQUIRED_TABLES.length) { + throw new Error('Alerts database is missing required tables'); + } + db.prepare( + `SELECT + window_start, + window_end, + member_files, + address_count, + alert_count, + alpha_min, + alpha_max, + alpha_median, + processed_at + FROM windows + LIMIT 0` + ).all(); + db.prepare( + `SELECT window_start, address, alpha, tail, rank, r2, prefix_levels + FROM alerts + LIMIT 0` + ).all(); + + const rows = db + .prepare( + `SELECT key, value FROM feed_meta WHERE key IN (${REQUIRED_META_KEYS.map(() => '?').join(', ')})` + ) + .all(...REQUIRED_META_KEYS) as FeedMetaRow[]; + const metadata = new Map(rows.map((row) => [row.key, row.value])); + const thresholdHigh = Number(metadata.get('threshold_high')); + const thresholdLow = Number(metadata.get('threshold_low')); + const maxPerTail = Number(metadata.get('max_per_tail')); + + if ( + metadata.get('schema_version') !== '1' || + metadata.get('dataset_id') !== datasetId || + !Number.isFinite(thresholdHigh) || + !Number.isFinite(thresholdLow) || + !Number.isSafeInteger(maxPerTail) || + maxPerTail < 1 + ) { + throw new Error('Alerts database metadata is invalid'); + } + + return { thresholdHigh, thresholdLow }; +} + +function readAlertsForWindow( + db: SqliteClient, + windowStart: number, + tail: AlertTail | undefined +): AlertFeedAlert[] { + const whereTail = tail ? 'AND tail = ?' : ''; + const params: QueryParam[] = tail ? [windowStart, tail] : [windowStart]; + return db + .prepare( + ` + SELECT address, alpha, tail, rank, r2 + FROM alerts + WHERE window_start = ? + ${whereTail} + ORDER BY CASE tail WHEN 'high' THEN 0 ELSE 1 END, rank ASC + ` + ) + .all(...params) as AlertFeedAlert[]; +} + +function readWindows( + db: SqliteClient, + options: Pick +): AlertFeedWindow[] { + const limitWindows = clampLimitWindows(options.limitWindows); + const beforeClause = options.before === undefined ? '' : 'WHERE window_start < ?'; + const params: QueryParam[] = + options.before === undefined ? [limitWindows] : [options.before, limitWindows]; + const rows = db + .prepare( + ` + SELECT + window_start AS windowStart, + window_end AS windowEnd, + address_count AS addressCount, + alert_count AS alertCount + FROM windows + ${beforeClause} + ORDER BY window_start DESC + LIMIT ? + ` + ) + .all(...params) as WindowRow[]; + + return rows.map((row) => ({ + ...row, + alerts: readAlertsForWindow(db, row.windowStart, options.tail) + })); +} + +export async function getAlertsFeedForDataset( + datasetId: string, + options: AlertsFeedOptions = {} +): Promise { + if (shouldUseD1(options.platform)) { + return absentFeed(); + } + + let alertsDbPath: string | undefined; + try { + const datasetDirectory = await resolveDatasetDirectory(datasetId); + if (!datasetDirectory) { + return absentFeed(); + } + + const path = await import('node:path'); + alertsDbPath = path.join(datasetDirectory, 'alerts.sqlite'); + const db = await getLocalDb(alertsDbPath); + const metadata = readFeedMetadata(db, datasetId); + const latestWindow = db + .prepare( + ` + SELECT window_start AS windowStart, processed_at AS processedAt + FROM windows + ORDER BY window_start DESC + LIMIT 1 + ` + ) + .get() as LatestWindowRow | undefined; + + return { + feed: { + present: true, + latestWindowStart: latestWindow?.windowStart ?? null, + latestProcessedAt: latestWindow?.processedAt ?? null, + thresholds: { high: metadata.thresholdHigh, low: metadata.thresholdLow } + }, + windows: readWindows(db, options) + }; + } catch { + if (alertsDbPath) { + evictLocalDb(alertsDbPath); + } + return absentFeed(); + } +} diff --git a/apps/web/src/lib/types/types.ts b/apps/web/src/lib/types/types.ts index b12d518..f73a2a5 100644 --- a/apps/web/src/lib/types/types.ts +++ b/apps/web/src/lib/types/types.ts @@ -11,6 +11,38 @@ export interface DatasetSummariesResponse { error: string | null; } +export type AlertTail = 'high' | 'low'; + +export interface AlertFeedAlert { + address: string; + alpha: number; + tail: AlertTail; + rank: number; + r2: number; +} + +export interface AlertFeedWindow { + windowStart: number; + windowEnd: number; + addressCount: number; + alertCount: number; + alerts: AlertFeedAlert[]; +} + +export type AlertFeedStatus = + | { present: false } + | { + present: true; + latestWindowStart: number | null; + latestProcessedAt: number | null; + thresholds: { high: number; low: number }; + }; + +export interface AlertsFeedResponse { + feed: AlertFeedStatus; + windows: AlertFeedWindow[]; +} + export type CoverageState = 'complete' | 'partial' | 'unknown'; export interface BucketCoverage { diff --git a/apps/web/src/routes/+layout.svelte b/apps/web/src/routes/+layout.svelte index bf47244..7a2a892 100644 --- a/apps/web/src/routes/+layout.svelte +++ b/apps/web/src/routes/+layout.svelte @@ -34,6 +34,11 @@ class="text-gray-500 hover:text-gray-900 dark:text-gray-400 dark:hover:text-gray-100" >Files + Alerts + + {#if copyError} + + {/if} + + {:else} +
+
+ {#each TAIL_OPTIONS as option (option.value)} + + {/each} +
+ + +
+ + {#if loading} +

Updating alerts…

+ {/if} + + {#if feedResponse.windows.length === 0} +
+ No windows have been processed yet. +
+ {:else} +
+ {#each feedResponse.windows as window (window.windowStart)} +
+
+
+

+ {formatTime(window.windowStart)}–{formatTime(window.windowEnd)} +

+

+ {formatDate(window.windowStart)} +

+
+

+ {countFormatter.format(window.addressCount)} addresses · {countFormatter.format( + window.alertCount + )} alerts +

+
+ + {#if window.alerts.length === 0} +
no alerts
+ {:else} +
+ {#each window.alerts as alert (`${alert.tail}:${alert.rank}`)} +
+ + {alert.address} + + + α {alert.alpha.toFixed(3)} + + + {alert.tail === 'high' ? 'High' : 'Low'} + + + r² {alert.r2.toFixed(2)} + +
+ {/each} +
+ {/if} +
+ {/each} +
+ + {#if hasOlder && oldestWindow} +
+ +
+ {/if} + {/if} + {/if} + diff --git a/apps/web/src/routes/alerts/+page.ts b/apps/web/src/routes/alerts/+page.ts new file mode 100644 index 0000000..48f3268 --- /dev/null +++ b/apps/web/src/routes/alerts/+page.ts @@ -0,0 +1,40 @@ +import { error } from '@sveltejs/kit'; +import type { PageLoad } from './$types'; +import { loadDatasetSummariesFromFetch, resolveDefaultDatasetId } from '$lib/datasets'; +import type { AlertsFeedResponse } from '$lib/types/types'; + +type ErrorResponse = { + data: null; + error: string; +}; + +export const load: PageLoad = async ({ fetch, url }) => { + try { + const datasets = await loadDatasetSummariesFromFetch(fetch); + const requestedDataset = url.searchParams.get('dataset')?.trim() || ''; + const selectedDataset = + requestedDataset && datasets.some((dataset) => dataset.datasetId === requestedDataset) + ? requestedDataset + : resolveDefaultDatasetId(datasets); + + let alerts: AlertsFeedResponse = { feed: { present: false }, windows: [] }; + if (selectedDataset) { + const response = await fetch( + `/api/alerts?dataset=${encodeURIComponent(selectedDataset)}&limitWindows=24` + ); + const payload = (await response.json()) as AlertsFeedResponse | ErrorResponse; + if (!response.ok || 'error' in payload) { + throw new Error('error' in payload ? payload.error : 'Failed to load alerts feed'); + } + alerts = payload; + } + + return { + datasets, + selectedDataset, + alerts + }; + } catch (err) { + throw error(500, err instanceof Error ? err.message : 'Failed to load alerts feed'); + } +}; diff --git a/apps/web/src/routes/api/alerts/+server.ts b/apps/web/src/routes/api/alerts/+server.ts new file mode 100644 index 0000000..b923ab3 --- /dev/null +++ b/apps/web/src/routes/api/alerts/+server.ts @@ -0,0 +1,61 @@ +import { json } from '@sveltejs/kit'; +import type { RequestHandler } from './$types'; +import { getAlertsFeedForDataset } from '$lib/server/alerts'; +import { getRequestedDataset } from '$lib/server/datasets'; +import type { AlertsFeedResponse, AlertTail } from '$lib/types/types'; + +type ErrorResponse = { + data: null; + error: string; +}; + +function parseInteger(value: string | null): number | undefined | null { + if (value === null) { + return undefined; + } + const normalized = value.trim(); + if (!/^-?\d+$/.test(normalized)) { + return null; + } + + const parsed = Number(normalized); + return Number.isSafeInteger(parsed) ? parsed : null; +} + +function errorResponse(message: string, status: number): Response { + const response: ErrorResponse = { data: null, error: message }; + return json(response, { status }); +} + +export const GET: RequestHandler = async ({ url, platform }) => { + const tailParam = url.searchParams.get('tail'); + if (tailParam !== null && tailParam !== 'high' && tailParam !== 'low') { + return errorResponse('Invalid tail parameter', 400); + } + const tail = tailParam as AlertTail | null; + + const parsedLimitWindows = parseInteger(url.searchParams.get('limitWindows')); + if (parsedLimitWindows === null) { + return errorResponse('Invalid limitWindows parameter', 400); + } + const limitWindows = Math.min(288, Math.max(1, parsedLimitWindows ?? 24)); + + const before = parseInteger(url.searchParams.get('before')); + if (before === null) { + return errorResponse('Invalid before parameter', 400); + } + + try { + const dataset = await getRequestedDataset(url, platform); + const response: AlertsFeedResponse = await getAlertsFeedForDataset(dataset, { + platform, + tail: tail ?? undefined, + limitWindows, + before + }); + return json(response); + } catch (error) { + console.error('Failed to load alerts feed:', error); + return errorResponse('Failed to load alerts feed', 500); + } +}; diff --git a/apps/web/tests/lib/server/alerts.test.ts b/apps/web/tests/lib/server/alerts.test.ts new file mode 100644 index 0000000..60b87ff --- /dev/null +++ b/apps/web/tests/lib/server/alerts.test.ts @@ -0,0 +1,277 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import Database from 'better-sqlite3'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +const ALERT_SCHEMA = ` + CREATE TABLE feed_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE windows ( + window_start INTEGER PRIMARY KEY, + window_end INTEGER NOT NULL, + member_files INTEGER NOT NULL, + address_count INTEGER NOT NULL, + alert_count INTEGER NOT NULL, + alpha_min REAL, + alpha_max REAL, + alpha_median REAL, + processed_at INTEGER NOT NULL + ); + CREATE TABLE alerts ( + window_start INTEGER NOT NULL REFERENCES windows(window_start) ON DELETE CASCADE, + address TEXT NOT NULL, + alpha REAL NOT NULL, + tail TEXT NOT NULL CHECK (tail IN ('high', 'low')), + rank INTEGER NOT NULL, + r2 REAL NOT NULL, + prefix_levels INTEGER NOT NULL, + PRIMARY KEY (window_start, tail, rank) + ); + CREATE INDEX alerts_address ON alerts(address, window_start); +`; + +type Fixture = { + directory: string; + netflowPath: string; + alertsPath: string; +}; + +async function loadAlertsModule() { + vi.resetModules(); + return import('../../../src/lib/server/alerts'); +} + +function createDatasetFixture(datasetId = 'alpha'): Fixture { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'alerts-test-')); + const netflowPath = path.join(directory, 'netflow.sqlite'); + const db = new Database(netflowPath); + db.exec(` + CREATE TABLE datasets ( + id TEXT PRIMARY KEY NOT NULL, + label TEXT NOT NULL, + default_start_date TEXT NOT NULL, + source_mode TEXT DEFAULT 'static' NOT NULL, + discovery_mode TEXT DEFAULT 'static' NOT NULL, + sort_order INTEGER DEFAULT 0 NOT NULL + ); + `); + db.prepare( + `INSERT INTO datasets ( + id, label, default_start_date, source_mode, discovery_mode, sort_order + ) VALUES (?, ?, '2025-03-01', 'static', 'live', 0)` + ).run(datasetId, 'Alpha Label'); + db.close(); + + return { + directory, + netflowPath, + alertsPath: path.join(directory, 'alerts.sqlite') + }; +} + +function seedAlertsDatabase(fixture: Fixture, windowCount = 2): void { + const db = new Database(fixture.alertsPath); + db.exec(ALERT_SCHEMA); + const insertMeta = db.prepare('INSERT INTO feed_meta (key, value) VALUES (?, ?)'); + for (const [key, value] of [ + ['schema_version', '1'], + ['dataset_id', 'alpha'], + ['threshold_high', '3.5'], + ['threshold_low', '0.4'], + ['max_per_tail', '25'] + ] as const) { + insertMeta.run(key, value); + } + + const insertWindow = db.prepare(` + INSERT INTO windows ( + window_start, + window_end, + member_files, + address_count, + alert_count, + alpha_min, + alpha_max, + alpha_median, + processed_at + ) VALUES (?, ?, 3, ?, ?, NULL, NULL, NULL, ?) + `); + const seedWindows = db.transaction(() => { + for (let index = 0; index < windowCount; index += 1) { + const windowStart = 1_700_000_000 + index * 300; + const isLatestFixtureWindow = index === 1; + insertWindow.run( + windowStart, + windowStart + 300, + 48_000 + index, + isLatestFixtureWindow ? 3 : index === 0 ? 1 : 0, + windowStart + 320 + ); + } + }); + seedWindows(); + + if (windowCount >= 1) { + db.prepare( + `INSERT INTO alerts ( + window_start, address, alpha, tail, rank, r2, prefix_levels + ) VALUES (?, '9.9.9.9', 0.21, 'low', 1, 0.91, 24)` + ).run(1_700_000_000); + } + if (windowCount >= 2) { + const insertAlert = db.prepare(` + INSERT INTO alerts ( + window_start, address, alpha, tail, rank, r2, prefix_levels + ) VALUES (?, ?, ?, ?, ?, ?, 24) + `); + const latestWindowStart = 1_700_000_300; + insertAlert.run(latestWindowStart, '1.1.1.2', 3.7, 'high', 2, 0.94); + insertAlert.run(latestWindowStart, '2.2.2.2', 0.2, 'low', 1, 0.89); + insertAlert.run(latestWindowStart, '1.1.1.1', 3.9, 'high', 1, 0.98); + } + db.close(); +} + +describe('alerts server helper', () => { + const originalCwd = process.cwd(); + + afterEach(() => { + process.chdir(originalCwd); + vi.unstubAllEnvs(); + }); + + it('returns feed metadata, reverse-chronological windows, and tail-rank ordered alerts', async () => { + const fixture = createDatasetFixture(); + seedAlertsDatabase(fixture); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({ + feed: { + present: true, + latestWindowStart: 1_700_000_300, + latestProcessedAt: 1_700_000_620, + thresholds: { high: 3.5, low: 0.4 } + }, + windows: [ + { + windowStart: 1_700_000_300, + windowEnd: 1_700_000_600, + addressCount: 48_001, + alertCount: 3, + alerts: [ + { address: '1.1.1.1', alpha: 3.9, tail: 'high', rank: 1, r2: 0.98 }, + { address: '1.1.1.2', alpha: 3.7, tail: 'high', rank: 2, r2: 0.94 }, + { address: '2.2.2.2', alpha: 0.2, tail: 'low', rank: 1, r2: 0.89 } + ] + }, + { + windowStart: 1_700_000_000, + windowEnd: 1_700_000_300, + addressCount: 48_000, + alertCount: 1, + alerts: [{ address: '9.9.9.9', alpha: 0.21, tail: 'low', rank: 1, r2: 0.91 }] + } + ] + }); + }); + + it('returns absent when a discovered dataset has no alerts database', async () => { + const workspace = fs.mkdtempSync(path.join(os.tmpdir(), 'alerts-discovery-')); + const datasetDirectory = path.join(workspace, 'data', 'alpha'); + fs.mkdirSync(datasetDirectory, { recursive: true }); + const fixture = createDatasetFixture(); + fs.renameSync(fixture.netflowPath, path.join(datasetDirectory, 'netflow.sqlite')); + process.chdir(workspace); + const alerts = await loadAlertsModule(); + + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({ + feed: { present: false }, + windows: [] + }); + }); + + it('does not throw when a configured dataset database has no sibling alerts database', async () => { + const fixture = createDatasetFixture(); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({ + feed: { present: false }, + windows: [] + }); + }); + + it('opens a feed file that appears after an earlier absent result', async () => { + const fixture = createDatasetFixture(); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({ + feed: { present: false } + }); + seedAlertsDatabase(fixture); + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({ + feed: { present: true }, + windows: [{ windowStart: 1_700_000_300 }, { windowStart: 1_700_000_000 }] + }); + }); + + it('evicts a cached feed handle when its file disappears', async () => { + const fixture = createDatasetFixture(); + seedAlertsDatabase(fixture); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toMatchObject({ + feed: { present: true } + }); + fs.unlinkSync(fixture.alertsPath); + await expect(alerts.getAlertsFeedForDataset('alpha')).resolves.toEqual({ + feed: { present: false }, + windows: [] + }); + }); + + it('filters nested alerts by tail without dropping windows or changing stored counts', async () => { + const fixture = createDatasetFixture(); + seedAlertsDatabase(fixture); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + const result = await alerts.getAlertsFeedForDataset('alpha', { tail: 'high' }); + expect(result.windows).toMatchObject([ + { + windowStart: 1_700_000_300, + alertCount: 3, + alerts: [ + { tail: 'high', rank: 1 }, + { tail: 'high', rank: 2 } + ] + }, + { windowStart: 1_700_000_000, alertCount: 1, alerts: [] } + ]); + }); + + it('uses before as an exclusive window-start cursor', async () => { + const fixture = createDatasetFixture(); + seedAlertsDatabase(fixture); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + const result = await alerts.getAlertsFeedForDataset('alpha', { before: 1_700_000_300 }); + expect(result.windows.map((window) => window.windowStart)).toEqual([1_700_000_000]); + }); + + it('clamps window limits to the inclusive range from 1 through 288', async () => { + const fixture = createDatasetFixture(); + seedAlertsDatabase(fixture, 289); + vi.stubEnv('LOCAL_SQLITE_PATH', fixture.netflowPath); + const alerts = await loadAlertsModule(); + + const lower = await alerts.getAlertsFeedForDataset('alpha', { limitWindows: 0 }); + const upper = await alerts.getAlertsFeedForDataset('alpha', { limitWindows: 999 }); + expect(lower.windows).toHaveLength(1); + expect(upper.windows).toHaveLength(288); + }); +}); diff --git a/apps/web/tests/routes/api-alerts.test.ts b/apps/web/tests/routes/api-alerts.test.ts new file mode 100644 index 0000000..f3d7300 --- /dev/null +++ b/apps/web/tests/routes/api-alerts.test.ts @@ -0,0 +1,126 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { getAlertsFeedForDataset } from '$lib/server/alerts'; +import { getRequestedDataset } from '$lib/server/datasets'; +import { GET } from '../../src/routes/api/alerts/+server'; + +vi.mock('$lib/server/alerts', () => ({ + getAlertsFeedForDataset: vi.fn() +})); + +vi.mock('$lib/server/datasets', () => ({ + getRequestedDataset: vi.fn() +})); + +const PRESENT_RESPONSE = { + feed: { + present: true as const, + latestWindowStart: 1_700_000_300, + latestProcessedAt: 1_700_000_620, + thresholds: { high: 3.5, low: 0.4 } + }, + windows: [ + { + windowStart: 1_700_000_300, + windowEnd: 1_700_000_600, + addressCount: 48_001, + alertCount: 1, + alerts: [{ address: '1.1.1.1', alpha: 3.9, tail: 'high' as const, rank: 1, r2: 0.98 }] + } + ] +}; + +function eventFor(query = '') { + return { + url: new URL(`http://localhost/api/alerts${query}`), + platform: undefined + } as never; +} + +describe('/api/alerts GET', () => { + beforeEach(() => { + vi.mocked(getRequestedDataset).mockReset().mockResolvedValue('alpha'); + vi.mocked(getAlertsFeedForDataset).mockReset().mockResolvedValue(PRESENT_RESPONSE); + }); + + it('returns the exact present-feed response and default query options', async () => { + const response = await GET(eventFor('?dataset=alpha')); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual(PRESENT_RESPONSE); + expect(getRequestedDataset).toHaveBeenCalledWith( + new URL('http://localhost/api/alerts?dataset=alpha'), + undefined + ); + expect(getAlertsFeedForDataset).toHaveBeenCalledWith('alpha', { + platform: undefined, + tail: undefined, + limitWindows: 24, + before: undefined + }); + }); + + it('passes a valid tail filter through and rejects an invalid tail', async () => { + const validResponse = await GET(eventFor('?tail=low')); + expect(validResponse.status).toBe(200); + expect(getAlertsFeedForDataset).toHaveBeenLastCalledWith( + 'alpha', + expect.objectContaining({ tail: 'low' }) + ); + + const invalidResponse = await GET(eventFor('?tail=middle')); + expect(invalidResponse.status).toBe(400); + await expect(invalidResponse.json()).resolves.toEqual({ + data: null, + error: 'Invalid tail parameter' + }); + }); + + it.each([ + ['?limitWindows=0', 1], + ['?limitWindows=999', 288] + ])('clamps %s before passing the window limit to the data layer', async (query, expected) => { + const response = await GET(eventFor(query)); + + expect(response.status).toBe(200); + expect(getAlertsFeedForDataset).toHaveBeenCalledWith( + 'alpha', + expect.objectContaining({ limitWindows: expected }) + ); + }); + + it('rejects a non-numeric window limit', async () => { + const response = await GET(eventFor('?limitWindows=many')); + + expect(response.status).toBe(400); + await expect(response.json()).resolves.toEqual({ + data: null, + error: 'Invalid limitWindows parameter' + }); + }); + + it('parses an exclusive before cursor and rejects an invalid cursor', async () => { + const validResponse = await GET(eventFor('?before=1700000300')); + expect(validResponse.status).toBe(200); + expect(getAlertsFeedForDataset).toHaveBeenLastCalledWith( + 'alpha', + expect.objectContaining({ before: 1_700_000_300 }) + ); + + const invalidResponse = await GET(eventFor('?before=1700000300.5')); + expect(invalidResponse.status).toBe(400); + await expect(invalidResponse.json()).resolves.toEqual({ + data: null, + error: 'Invalid before parameter' + }); + }); + + it('returns an absent feed as a normal 200 response', async () => { + const absentResponse = { feed: { present: false as const }, windows: [] }; + vi.mocked(getAlertsFeedForDataset).mockResolvedValue(absentResponse); + + const response = await GET(eventFor()); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual(absentResponse); + }); +}); From efff45b0169ca3ae6e9cd93262c88892c25078d6 Mon Sep 17 00:00:00 2001 From: flamboh Date: Thu, 20 Aug 2026 21:20:34 -0700 Subject: [PATCH 04/14] feat(netflow-db): calibrate singularity alert thresholds --- docs/agent/singularity-calibration.md | 227 ++++++++++++++++++++++++++ tools/netflow-db/src/feed.rs | 10 +- tools/netflow-db/src/main.rs | 2 +- 3 files changed, 234 insertions(+), 5 deletions(-) create mode 100644 docs/agent/singularity-calibration.md diff --git a/docs/agent/singularity-calibration.md b/docs/agent/singularity-calibration.md new file mode 100644 index 0000000..8cc0c5d --- /dev/null +++ b/docs/agent/singularity-calibration.md @@ -0,0 +1,227 @@ +# Rust Singularity conformance and threshold calibration + +## Summary + +I compared the Rust port with the prebuilt Haskell reference on 24 real five-minute windows and calibrated the Rust alert feed on 200 separate windows. The conformance verdict is PASS. Every compared address and `n_levels` value matched, and no numeric value violated the requested tolerance. + +The calibration supports `threshold_high = 2` and `threshold_low = 0.3`. At the feed's 20-per-tail cap, this pair produces a mean of 19.7 and median of 20.0 recorded alerts per window. The middle 80% spans 14.0 to 25.0. + +## Environment and extraction contract + +The Rust binary was `netflow-db 0.1.0`, built from commit `a30a51f41793ec88c99e07bf4bb2aa8078345192` with the repository's Rust 1.97.1 toolchain. Its SHA-256 was `a3258b36c29021a6056cdbb53633313d3200884ce8a36e61b62f8ec6e9f1cac6`. The prebuilt Haskell binary SHA-256 was `ea5b0ccca94355cafbebe6f3c2a9ca3e8441b2cb11406a93b8f67e4971f81d69`; I did not rebuild it. Capture reads used nfdump 1.7.6-release on Linux 6.18.35 x86_64. + +For each timestamp, the extractor reads both `cc_ir1_gw` and `oh_ir1_gw` files with `nfdump -q -o 'csv:%sa,%da'`, rejects fields containing `:`, combines source and destination addresses, and runs one locale-fixed external `sort -u`. The standard CSV probe reported `srcAddr` and `dstAddr` as columns 4 and 6. A 10,000-record validation found 5,149 unique IPv4 addresses in both the standard ten-column output and the custom two-column projection, with byte-identical sorted lists. + +The capture inventory contained 111,272 paired timestamps from 2025-06-01 00:00 through 2026-06-30 13:15. The selector excluded one cc-only and two oh-only timestamps. Filename hours are treated as America/Los_Angeles local time, matching the feed's configured timezone. + +## Part 1: Rust versus Haskell conformance + +### Method + +The deterministic selector chose 24 windows across all 13 available months. I kept all 24 candidates, instead of stopping at 20, so May and June 2026 remained represented. It used 03:00, 09:00, 15:00, and 21:00 hours six times each, with 16 weekday and 8 weekend windows. Both implementations received the exact same sorted address file. I joined results by address, compared `n_levels` with exact integer equality, and applied `abs(a-b) <= max(1e-9 * max(abs(a),abs(b)), 1e-12)` to alpha, intercept, and r2. + +The deviation columns below report `abs(a-b) / max(abs(a),abs(b),1e-12)`. Near zero, that display value can exceed 1e-9 while the absolute 1e-12 tolerance still passes. The A/I/R violations column removes that ambiguity. + +| Timestamp | Addresses | Max rel alpha | Max rel intercept | Max rel r2 | A/I/R violations | Level mismatches | Rust/Haskell only | +| ------------ | --------- | ------------- | ----------------- | ---------- | ---------------- | ---------------- | ----------------- | +| 202506080355 | 296,779 | 7.270e-15 | 1.474e-10 | 6.947e-15 | 0/0/0 | 0 | 0/0 | +| 202506250900 | 309,275 | 3.911e-15 | 4.524e-09 | 6.901e-15 | 0/0/0 | 0 | 0/0 | +| 202507111555 | 299,657 | 3.817e-15 | 5.471e-10 | 6.190e-15 | 0/0/0 | 0 | 0/0 | +| 202507272155 | 294,684 | 4.441e-15 | 6.094e-11 | 1.168e-14 | 0/0/0 | 0 | 0/0 | +| 202508140300 | 302,023 | 5.717e-15 | 1.033e-10 | 8.630e-15 | 0/0/0 | 0 | 0/0 | +| 202508290955 | 301,098 | 4.400e-15 | 1.066e-10 | 6.494e-15 | 0/0/0 | 0 | 0/0 | +| 202509141555 | 291,695 | 4.422e-15 | 2.007e-10 | 7.696e-15 | 0/0/0 | 0 | 0/0 | +| 202510022100 | 330,546 | 4.318e-15 | 1.182e-09 | 7.520e-15 | 0/0/0 | 0 | 0/0 | +| 202510170355 | 294,497 | 7.737e-15 | 2.417e-10 | 8.917e-15 | 0/0/0 | 0 | 0/0 | +| 202511020955 | 309,703 | 4.559e-15 | 7.636e-11 | 7.037e-15 | 0/0/0 | 0 | 0/0 | +| 202511171555 | 316,316 | 7.450e-15 | 1.004e-10 | 7.095e-15 | 0/0/0 | 0 | 0/0 | +| 202512082100 | 318,123 | 5.220e-15 | 7.459e-10 | 7.741e-15 | 0/0/0 | 0 | 0/0 | +| 202512210355 | 290,362 | 4.610e-15 | 2.848e-09 | 8.392e-15 | 0/0/0 | 0 | 0/0 | +| 202601090900 | 373,695 | 4.797e-15 | 1.187e-09 | 8.681e-15 | 0/0/0 | 0 | 0/0 | +| 202601261500 | 321,101 | 5.734e-15 | 1.404e-08 | 7.690e-15 | 0/0/0 | 0 | 0/0 | +| 202602082155 | 305,401 | 6.569e-15 | 1.324e-10 | 7.569e-15 | 0/0/0 | 0 | 0/0 | +| 202602270300 | 311,010 | 5.386e-15 | 1.380e-10 | 9.091e-15 | 0/0/0 | 0 | 0/0 | +| 202603160900 | 335,478 | 4.635e-15 | 1.410e-10 | 6.593e-15 | 0/0/0 | 0 | 0/0 | +| 202603291555 | 329,895 | 5.980e-15 | 5.910e-11 | 8.342e-15 | 0/0/0 | 0 | 0/0 | +| 202604172100 | 343,949 | 4.615e-15 | 1.824e-10 | 7.257e-15 | 0/0/0 | 0 | 0/0 | +| 202605040300 | 348,968 | 4.265e-15 | 2.754e-10 | 6.557e-15 | 0/0/0 | 0 | 0/0 | +| 202605170955 | 346,383 | 7.401e-15 | 1.073e-08 | 7.783e-15 | 0/0/0 | 0 | 0/0 | +| 202606051555 | 338,304 | 5.021e-15 | 1.737e-10 | 7.039e-15 | 0/0/0 | 0 | 0/0 | +| 202606222100 | 324,410 | 4.557e-15 | 1.269e-10 | 7.463e-15 | 0/0/0 | 0 | 0/0 | + +Overall maxima were 7.737e-15 for alpha, 1.404e-08 for intercept, and 1.168e-14 for r2. The total mismatch or tolerance-violation count was 0. The largest tested window was 202601090900 with 373,695 addresses. + +The largest displayed intercept ratio came from two near-zero values at 72.5.189.239: Rust -1.642975968608e-07, Haskell -1.642975945543e-07. Their absolute difference was 2.306e-15, versus an allowed 1.000e-12. This is the expected absolute-tolerance case, not a numerical failure. + +### Haskell wall-clock budget + +No window was skipped, replaced, sampled down, or timed out. The firm budget was 600 seconds per Haskell run. Observed Haskell time had a median of 11.7 seconds and a maximum of 16.0 seconds. The 299,472-address format probe took 9.1 seconds in Haskell and 3.7 seconds in Rust, so full-window checks were comfortably tractable. + +## Part 2: Rust threshold calibration + +### Sampling and statistics + +The calibration used 200 unique paired windows from 202506010000 through 202606301100. Coverage included 13 months and all 24 hours, with 8 or 9 samples per hour, 14 to 16 per month, 56 weekend windows, and 144 weekday windows. The selector chose the nearest available pair around evenly spaced date-bin midpoints and permuted requested hours with `(index * 5) % 24`. Percentiles use Type 7 linear interpolation at rank `(n - 1) * p`. + +All 200 windows completed. Rust emitted 0 non-finite alpha values across the sample. + +Across-window distribution of each per-window statistic: + +| Statistic | Mean | SD | Min | p10 | Median | p90 | Max | +| ----------- | ------- | ------ | ------- | ------- | ------- | ------- | ------- | +| n_addresses | 318,709 | 21,833 | 274,916 | 296,414 | 314,254 | 349,859 | 396,341 | +| min | 0.199 | 0.024 | 0.170 | 0.178 | 0.191 | 0.229 | 0.277 | +| p0_1 | 0.364 | 0.010 | 0.348 | 0.353 | 0.362 | 0.379 | 0.396 | +| p1 | 0.394 | 0.009 | 0.371 | 0.383 | 0.392 | 0.408 | 0.421 | +| p5 | 0.439 | 0.009 | 0.406 | 0.425 | 0.439 | 0.449 | 0.458 | +| median | 0.511 | 0.002 | 0.508 | 0.509 | 0.511 | 0.514 | 0.519 | +| p95 | 1.021 | 0.006 | 1.006 | 1.013 | 1.021 | 1.028 | 1.055 | +| p99 | 1.190 | 0.004 | 1.174 | 1.184 | 1.190 | 1.195 | 1.202 | +| p99_9 | 1.379 | 0.012 | 1.345 | 1.363 | 1.380 | 1.392 | 1.410 | +| max | 3.194 | 1.155 | 2.013 | 2.149 | 2.265 | 4.507 | 4.917 | + +Eight representative windows: + +| Timestamp | Addresses | Min | p0.1 | Median | p99.9 | Max | +| ------------ | --------- | ----- | ----- | ------ | ----- | ----- | +| 202506010000 | 298,018 | 0.180 | 0.352 | 0.510 | 1.354 | 4.457 | +| 202507272015 | 300,483 | 0.177 | 0.360 | 0.510 | 1.392 | 4.457 | +| 202509222130 | 315,015 | 0.181 | 0.359 | 0.510 | 1.377 | 4.509 | +| 202511161745 | 331,615 | 0.203 | 0.363 | 0.513 | 1.373 | 2.193 | +| 202601131800 | 341,425 | 0.231 | 0.368 | 0.514 | 1.365 | 4.539 | +| 202603091415 | 316,560 | 0.198 | 0.361 | 0.512 | 1.385 | 2.239 | +| 202605051545 | 360,662 | 0.200 | 0.371 | 0.516 | 1.363 | 2.214 | +| 202606301100 | 349,565 | 0.190 | 0.369 | 0.514 | 1.357 | 2.209 | + +### Tail stability and time effects + +The table compares overall spread with date, hour, weekend, and month groupings. Date uses Pearson correlation against capture time. Hour and month cells show the groups with the lowest and highest means. Weekday/weekend is shown in that order. + +| Metric | Overall SD | Overall p10 to p90 | Date r | Hourly mean low/high | Weekday/weekend mean | Monthly mean low/high | +| ------ | ---------- | ------------------ | ------ | ----------------------- | -------------------- | ---------------------------- | +| min | 0.024 | 0.178 to 0.229 | 0.052 | 0:00 0.186; 2:00 0.225 | 0.201/0.193 | 2026-03 0.190; 2025-12 0.207 | +| p0_1 | 0.010 | 0.353 to 0.379 | 0.423 | 2:00 0.359; 5:00 0.370 | 0.366/0.361 | 2025-08 0.356; 2026-05 0.374 | +| p99_9 | 0.012 | 1.363 to 1.392 | -0.360 | 9:00 1.373; 21:00 1.386 | 1.377/1.383 | 2026-05 1.368; 2025-08 1.387 | +| max | 1.155 | 2.149 to 4.507 | -0.607 | 4:00 2.724; 18:00 3.663 | 3.247/3.057 | 2025-11 2.178; 2025-06 4.509 | + +The center of the alpha distribution is tight: the per-window median has mean 0.511 and SD 0.002. The p0.1 and p99.9 tails have SDs of 0.010 and 0.012. Min and max are much noisier because a single address controls each value. The table keeps network growth and calendar effects separate from that single-address churn instead of treating every max swing as a feed-wide shift. + +There is a measured date shift, but it is small in absolute alpha terms. Address count rose with date at r=0.697; median alpha also rose at r=0.719, while its SD stayed 0.002. The p0.1 mean rose from a monthly low of 0.356 in August 2025 to 0.374 in May 2026, with date r=0.423. The p99.9 mean moved the other way, from 1.387 in August 2025 to 1.368 in May 2026, with r=-0.360. + +Time-of-day and weekday effects were smaller. Hourly p0.1 means ranged from 0.359 to 0.370, and hourly p99.9 means ranged from 1.373 to 1.386. Weekday versus weekend means differed by 0.005 for p0.1 and 0.006 for p99.9. Max alpha was the unstable statistic: SD 1.155, p10 to p90 2.149 to 4.507, and date r=-0.607. That reflects individual recurring addresses entering or leaving the tail, not a broad distribution shift. + +High alpha marks sparse, isolated address-space regions. Low alpha marks addresses that remain inside dense prefix clusters. Both tails therefore remain in the recommendation. + +### Initial candidate grid + +Counts here use the requested strict comparisons. Capped values apply the feed's 20-per-tail limit before combining tails. + +| High | Low | Raw mean | Raw median | Capped mean | Capped median | +| ---- | --- | -------- | ---------- | ----------- | ------------- | +| 2.5 | 0.3 | 6.5 | 6.0 | 5.7 | 6.0 | +| 2.5 | 0.5 | 129,919 | 128,360 | 20.4 | 20.0 | +| 2.5 | 0.7 | 245,417 | 241,792 | 20.4 | 20.0 | +| 3.0 | 0.3 | 6.5 | 6.0 | 5.7 | 6.0 | +| 3.0 | 0.5 | 129,919 | 128,360 | 20.4 | 20.0 | +| 3.0 | 0.7 | 245,417 | 241,792 | 20.4 | 20.0 | +| 3.5 | 0.3 | 6.5 | 6.0 | 5.7 | 6.0 | +| 3.5 | 0.5 | 129,919 | 128,360 | 20.4 | 20.0 | +| 3.5 | 0.7 | 245,417 | 241,792 | 20.4 | 20.0 | + +The initial high range of 2.5 to 3.5 is too conservative on this feed. It contributes almost no high-tail volume, leaving the low tail to determine the result. A high threshold of 2.0 restores a useful sparse-address signal without pinning the high tail at its cap. + +### Recommendation + +Use `threshold_high = 2` and `threshold_low = 0.3` as fixed global constants. Before caps, high 2.0 produces mean/median counts of 14.7/14.0 and low 0.3 produces 6.1/5.0. Combined raw mean/median is 20.8/20.0. After the 20-per-tail caps, combined mean/median is 19.7/20.0. + +The saved grid uses strict `>` and `<` as requested. The Rust feed uses inclusive `>=` and `<=`. Exactly 0 sampled address scores landed on either recommended constant, so inclusive feed-visible mean/median remains 19.7/20.0. + +I prefer 2.0 and 0.3 because they are round, give both tails room to contribute, and put a typical window inside the requested 5 to 30 recorded-alert range. The p10 to p90 capped range is 14 to 25. One May 2026 window had 177 raw low-tail crossings and 194 raw combined crossings; the per-tail caps reduced it to 37 recorded alerts. None of the 200 selected windows was quiet enough to produce zero at this pair, but an empty or genuinely low-traffic window can still do so. + +### Repeat offenders + +I selected eight evenly spaced calibration windows and retained every address beyond the recommended strict thresholds. + +| Timestamp | High | Low | Combined | +| ------------ | ---- | --- | -------- | +| 202506010000 | 5 | 4 | 9 | +| 202507272015 | 17 | 6 | 23 | +| 202509222130 | 16 | 6 | 22 | +| 202511161745 | 15 | 6 | 21 | +| 202601131800 | 18 | 4 | 22 | +| 202603091415 | 17 | 5 | 22 | +| 202605051545 | 19 | 7 | 26 | +| 202606301100 | 13 | 6 | 19 | + +Pairwise overlap across all 28 window pairs: + +| Tail | Mean Jaccard | Median Jaccard | Jaccard range | Mean intersection | Intersection range | +| -------- | ------------ | -------------- | -------------- | ----------------- | ------------------ | +| high | 0.008 | 0.000 | 0.000 to 0.050 | 0.2 | 0 to 1 | +| low | 0.487 | 0.444 | 0.300 to 0.833 | 3.5 | 3 to 5 | +| combined | 0.102 | 0.101 | 0.065 to 0.148 | 3.7 | 3 to 5 | + +Addresses recurring in at least two of the eight windows, limited to the ten most frequent: + +| Tail | Address | Windows | Min alpha | Max alpha | +| ---- | --------------- | ------- | --------- | --------- | +| low | 224.0.0.1 | 8/8 | 0.267 | 0.273 | +| low | 224.0.0.13 | 8/8 | 0.289 | 0.296 | +| low | 224.0.0.2 | 8/8 | 0.267 | 0.273 | +| low | 72.4.181.18 | 5/8 | 0.215 | 0.231 | +| high | 239.255.255.250 | 4/8 | 4.457 | 4.539 | +| low | 72.5.64.18 | 3/8 | 0.198 | 0.214 | +| low | 60.247.96.22 | 2/8 | 0.190 | 0.200 | + +The two tails behave differently. High-tail membership mostly churned, with mean Jaccard 0.008 and pairwise intersections of at most one address. The only recurring high address was 239.255.255.250, present in four of eight windows. The low tail had a stable core: 224.0.0.1, 224.0.0.2, and 224.0.0.13 appeared in all eight windows, and low-tail mean Jaccard was 0.487. The feed should therefore expect repeat multicast-style low alerts alongside a changing high tail. + +## How to reproduce + +All scripts, raw outputs, address lists, logs, and summaries are under `/tmp/singularity-calibration/`. Nothing from this run was written to tracked repository files. The Cargo build wrote only the permitted `target/` output. + +Key audit files: + +- `inventory/conformance_candidates.txt` and `inventory/calibration_timestamps.txt` contain the selected timestamps. +- `inventory/*_coverage.csv` records dates, hours, day types, and both capture paths. +- `data/conformance_per_window/` and `data/calibration_per_window/` contain every per-window CSV summary. +- `data/conformance_windows.csv` and `data/calibration_windows.csv` are the consolidated tables. +- `data/conformance_worst_deviations.csv` records the address and absolute difference behind each displayed maximum. +- `data/threshold_grid.csv`, `data/threshold_tail_sensitivity.csv`, and `data/threshold_pair_sensitivity.csv` contain the threshold scan. +- `data/repeat_*.csv` contains the repeat-offender sets and overlaps. +- `work/conformance/` retains Rust and Haskell CSVs plus exact address lists. `work/calibration/` retains all Rust CSVs and address lists. + +Run these commands from a fresh `/tmp/singularity-calibration/` layout after restoring the scripts: + +```sh +cd /home/obo/.t3/worktrees/netflow-analysis/t3code-5e5d2a1c +cargo build --release -p atlantis-netflow-db +/tmp/singularity-calibration/scripts/select_windows.py +/tmp/singularity-calibration/scripts/run_conformance.py /tmp/singularity-calibration/inventory/conformance_candidates.txt target/release/netflow-db --target 24 --workers 2 --timeout-seconds 600 +/tmp/singularity-calibration/scripts/run_calibration.py /tmp/singularity-calibration/inventory/calibration_timestamps.txt target/release/netflow-db --workers 4 +/tmp/singularity-calibration/scripts/find_worst_deviations.py +/tmp/singularity-calibration/scripts/augment_candidate_pairs.py /tmp/singularity-calibration/data/calibration_windows.csv /tmp/singularity-calibration/data/calibration_per_window +/tmp/singularity-calibration/scripts/scan_thresholds.py /tmp/singularity-calibration/work/calibration/scores /tmp/singularity-calibration/data/threshold_grid.csv +/tmp/singularity-calibration/scripts/analyze_calibration.py /tmp/singularity-calibration/data/calibration_windows.csv /tmp/singularity-calibration/data/threshold_grid.csv /tmp/singularity-calibration/data +/tmp/singularity-calibration/scripts/analyze_repeat_offenders.py /tmp/singularity-calibration/work/calibration/scores /tmp/singularity-calibration/inventory/calibration_timestamps.txt /tmp/singularity-calibration/data --high 2.0 --low 0.3 --windows 8 +/tmp/singularity-calibration/scripts/render_report.py +``` + +## Final concise summary + +Conformance verdict: PASS across 24 real windows and 7,633,352 address comparisons. Maximum displayed deviations were 7.737e-15 alpha, 1.404e-08 intercept, and 1.168e-14 r2. Tolerance violations, level mismatches, and missing addresses were all zero. + +Recommended constants: `threshold_high = 2` and `threshold_low = 0.3`. High 2.0 brings the sparse-address tail back into the feed, while low 0.3 keeps the dense-cluster tail selective. At the 20-per-tail cap, expected alerts per window are mean 19.7, median 20.0, with p10 to p90 of 14.0 to 25.0. + +Sensitivity around the recommendation: + +| High | Low | Raw mean | Raw median | Raw p10 to p90 | Capped mean | Capped median | Capped p10 to p90 | Zero windows | +| ---- | ----- | -------- | ---------- | -------------- | ----------- | ------------- | ----------------- | ------------ | +| 1.75 | 0.3 | 63.2 | 62.0 | 51.0 to 74.1 | 25.2 | 25.0 | 24.0 to 27.0 | 0.0% | +| 1.9 | 0.3 | 34.3 | 33.5 | 26.0 to 41.0 | 25.1 | 25.0 | 24.0 to 27.0 | 0.0% | +| 2 | 0.25 | 16.8 | 17.0 | 11.0 to 23.0 | 16.5 | 17.0 | 11.0 to 22.0 | 0.0% | +| 2 | 0.275 | 19.2 | 19.0 | 13.0 to 25.0 | 18.6 | 19.0 | 13.0 to 24.0 | 0.0% | +| 2 | 0.3 | 20.8 | 20.0 | 14.0 to 26.0 | 19.7 | 20.0 | 14.0 to 25.0 | 0.0% | +| 2 | 0.325 | 22.0 | 21.0 | 15.0 to 27.0 | 20.9 | 21.0 | 15.0 to 26.0 | 0.0% | +| 2 | 0.35 | 41.2 | 27.0 | 17.9 to 40.1 | 26.4 | 26.0 | 17.9 to 36.1 | 0.0% | +| 2.1 | 0.3 | 11.9 | 11.0 | 7.0 to 15.1 | 11.1 | 11.0 | 7.0 to 15.1 | 0.0% | +| 2.25 | 0.3 | 6.7 | 6.0 | 4.0 to 8.0 | 6.0 | 6.0 | 4.0 to 8.0 | 0.0% | diff --git a/tools/netflow-db/src/feed.rs b/tools/netflow-db/src/feed.rs index 0a965b5..71994f6 100644 --- a/tools/netflow-db/src/feed.rs +++ b/tools/netflow-db/src/feed.rs @@ -38,10 +38,12 @@ const SECONDS_PER_DAY: i64 = 24 * SECONDS_PER_HOUR; // not carry a timezone, and registry-driven pipeline runs use this default. const TIMEZONE: &str = "America/Los_Angeles"; -// TODO: These are uncalibrated placeholders. A future calibration pass will -// replace them with thresholds derived from representative datasets. -const DEFAULT_THRESHOLD_HIGH: f64 = 3.5; -const DEFAULT_THRESHOLD_LOW: f64 = 0.4; +// Calibrated on 200 uOregon five-minute windows spanning 2025-06 through +// 2026-06 after conformance against the Haskell reference; a typical window +// records ~20 alerts (p10-p90: 14-25) across both tails at these values. +// Methodology and sensitivity: docs/agent/singularity-calibration.md. +const DEFAULT_THRESHOLD_HIGH: f64 = 2.0; +const DEFAULT_THRESHOLD_LOW: f64 = 0.3; const ALERT_SCHEMA: &str = r#" CREATE TABLE IF NOT EXISTS feed_meta ( diff --git a/tools/netflow-db/src/main.rs b/tools/netflow-db/src/main.rs index cc83ec3..e047d5b 100644 --- a/tools/netflow-db/src/main.rs +++ b/tools/netflow-db/src/main.rs @@ -17,8 +17,8 @@ use netflow_db::{ }, prepare::{PrepareOptions, prepare_archive}, registry::DatasetRegistry, - storage::{backup_database, promote_database}, singularity, + storage::{backup_database, promote_database}, verify::{VerifyOptions, verify_database}, }; From 10cc36fa85acb7785c0a5a936b1bf3f86a2ddf14 Mon Sep 17 00:00:00 2001 From: flamboh Date: Thu, 20 Aug 2026 21:20:49 -0700 Subject: [PATCH 05/14] feat(web): explain alert tails with badge tooltips --- apps/web/src/routes/alerts/+page.svelte | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/web/src/routes/alerts/+page.svelte b/apps/web/src/routes/alerts/+page.svelte index 8a512a6..70a8cf6 100644 --- a/apps/web/src/routes/alerts/+page.svelte +++ b/apps/web/src/routes/alerts/+page.svelte @@ -417,6 +417,9 @@ α {alert.alpha.toFixed(3)} Date: Thu, 20 Aug 2026 23:55:09 -0700 Subject: [PATCH 06/14] feat(netflow-db): accept minute backfill durations --- tools/netflow-db/src/feed.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tools/netflow-db/src/feed.rs b/tools/netflow-db/src/feed.rs index 71994f6..69baf8d 100644 --- a/tools/netflow-db/src/feed.rs +++ b/tools/netflow-db/src/feed.rs @@ -574,6 +574,7 @@ fn parse_backfill_duration(value: &str) -> Result { } let amount = amount.parse::().map_err(|_| invalid_backfill(value))?; let unit_seconds = match unit { + 'm' => 60, 'h' => SECONDS_PER_HOUR, 'd' => SECONDS_PER_DAY, _ => return Err(invalid_backfill(value)), @@ -585,7 +586,7 @@ fn parse_backfill_duration(value: &str) -> Result { fn invalid_backfill(value: &str) -> FeedError { FeedError::InvalidConfig(format!( - "invalid backfill duration {value:?}; expected an integer followed by 'h' or 'd'" + "invalid backfill duration {value:?}; expected an integer followed by 'm', 'h', or 'd'" )) } @@ -745,7 +746,8 @@ mod tests { } #[test] - fn parses_hour_and_day_backfills() { + fn parses_minute_hour_and_day_backfills() { + assert_eq!(parse_backfill_duration("15m").unwrap(), 15 * 60); assert_eq!( parse_backfill_duration("36h").unwrap(), 36 * SECONDS_PER_HOUR From 22ec066b75a9f345285d52796dfb8d6075401077 Mon Sep 17 00:00:00 2001 From: flamboh Date: Fri, 21 Aug 2026 00:25:41 -0700 Subject: [PATCH 07/14] feat(web): restructure alerts into address-centric feed --- apps/web/src/lib/server/alerts.ts | 194 +++++++--- apps/web/src/lib/types/types.ts | 29 +- apps/web/src/routes/alerts/+page.svelte | 451 +++++++++++++--------- apps/web/src/routes/alerts/+page.ts | 9 +- apps/web/src/routes/api/alerts/+server.ts | 33 +- apps/web/tests/lib/server/alerts.test.ts | 126 ++++-- apps/web/tests/routes/api-alerts.test.ts | 356 +++++++++++++---- 7 files changed, 804 insertions(+), 394 deletions(-) diff --git a/apps/web/src/lib/server/alerts.ts b/apps/web/src/lib/server/alerts.ts index 38cf672..df9fd1f 100644 --- a/apps/web/src/lib/server/alerts.ts +++ b/apps/web/src/lib/server/alerts.ts @@ -1,8 +1,9 @@ import { env as privateEnv } from '$env/dynamic/private'; import type Database from 'better-sqlite3'; import type { - AlertFeedAlert, - AlertFeedWindow, + AlertFeedAddress, + AlertHorizon, + AlertSort, AlertsFeedResponse, AlertTail } from '$lib/types/types'; @@ -26,20 +27,26 @@ type FeedMetaRow = { value: string; }; -type WindowRow = { +type LatestWindowRow = { windowStart: number; windowEnd: number; addressCount: number; - alertCount: number; + processedAt: number; }; -type LatestWindowRow = { - windowStart: number; - processedAt: number; +type AddressRow = AlertFeedAddress & { + totalAddresses: number; }; -const DEFAULT_LIMIT_WINDOWS = 24; -const MAX_LIMIT_WINDOWS = 288; +const HORIZON_SECONDS: Record = { + '1h': 60 * 60, + '6h': 6 * 60 * 60, + '24h': 24 * 60 * 60, + '7d': 7 * 24 * 60 * 60 +}; +const DEFAULT_HORIZON: AlertHorizon = '24h'; +const DEFAULT_LIMIT = 100; +const MAX_LIMIT = 500; const REQUIRED_TABLES = ['alerts', 'feed_meta', 'windows'] as const; const REQUIRED_META_KEYS = [ 'schema_version', @@ -54,12 +61,18 @@ const localDbCache = new Map(); export type AlertsFeedOptions = { platform?: App.Platform; tail?: AlertTail; - limitWindows?: number; - before?: number; + horizon?: AlertHorizon; + sort?: AlertSort; + limit?: number; }; -function absentFeed(): AlertsFeedResponse { - return { feed: { present: false }, windows: [] }; +function absentFeed(horizonSeconds: number): AlertsFeedResponse { + return { + feed: { present: false }, + horizonSeconds, + totalAddresses: 0, + addresses: [] + }; } function getEnv(name: string): string | undefined { @@ -209,11 +222,11 @@ async function getLocalDb(dbPath: string): Promise { } } -function clampLimitWindows(limitWindows: number | undefined): number { - if (limitWindows === undefined || !Number.isFinite(limitWindows)) { - return DEFAULT_LIMIT_WINDOWS; +function clampLimit(limit: number | undefined): number { + if (limit === undefined || !Number.isFinite(limit)) { + return DEFAULT_LIMIT; } - return Math.min(MAX_LIMIT_WINDOWS, Math.max(1, Math.trunc(limitWindows))); + return Math.min(MAX_LIMIT, Math.max(1, Math.trunc(limit))); } function readFeedMetadata( @@ -272,99 +285,154 @@ function readFeedMetadata( return { thresholdHigh, thresholdLow }; } -function readAlertsForWindow( +function readAddresses( db: SqliteClient, - windowStart: number, - tail: AlertTail | undefined -): AlertFeedAlert[] { - const whereTail = tail ? 'AND tail = ?' : ''; - const params: QueryParam[] = tail ? [windowStart, tail] : [windowStart]; - return db - .prepare( - ` - SELECT address, alpha, tail, rank, r2 - FROM alerts - WHERE window_start = ? - ${whereTail} - ORDER BY CASE tail WHEN 'high' THEN 0 ELSE 1 END, rank ASC - ` - ) - .all(...params) as AlertFeedAlert[]; -} + options: Pick, + horizonStart: number, + thresholdHigh: number, + thresholdLow: number +): { totalAddresses: number; addresses: AlertFeedAddress[] } { + const whereTail = options.tail ? 'AND tail = ?' : ''; + const sortOrder = + options.sort === 'recent' + ? 'lastSeen DESC, severity DESC, address ASC' + : 'severity DESC, address ASC'; + const limit = clampLimit(options.limit); + const params: QueryParam[] = [thresholdHigh, thresholdLow, horizonStart]; + if (options.tail) { + params.push(options.tail); + } + params.push(limit); -function readWindows( - db: SqliteClient, - options: Pick -): AlertFeedWindow[] { - const limitWindows = clampLimitWindows(options.limitWindows); - const beforeClause = options.before === undefined ? '' : 'WHERE window_start < ?'; - const params: QueryParam[] = - options.before === undefined ? [limitWindows] : [options.before, limitWindows]; const rows = db .prepare( ` + WITH scoped AS ( + SELECT + address, + alpha, + tail, + rank, + r2, + window_start, + CASE tail + WHEN 'high' THEN alpha - ? + ELSE ? - alpha + END AS severity + FROM alerts + WHERE window_start >= ? + ${whereTail} + ), + ranked AS ( + SELECT + address, + alpha, + tail, + r2, + window_start, + severity, + ROW_NUMBER() OVER ( + PARTITION BY address + ORDER BY severity DESC, window_start DESC, rank ASC + ) AS peak_rank, + MAX(window_start) OVER (PARTITION BY address) AS last_seen, + MIN(window_start) OVER (PARTITION BY address) AS first_seen, + COUNT(*) OVER (PARTITION BY address) AS times_flagged + FROM scoped + ) SELECT - window_start AS windowStart, - window_end AS windowEnd, - address_count AS addressCount, - alert_count AS alertCount - FROM windows - ${beforeClause} - ORDER BY window_start DESC + address, + tail, + alpha AS peakAlpha, + window_start AS peakWindowStart, + r2 AS peakR2, + last_seen AS lastSeen, + first_seen AS firstSeen, + times_flagged AS timesFlagged, + COUNT(*) OVER () AS totalAddresses + FROM ranked + WHERE peak_rank = 1 + ORDER BY ${sortOrder} LIMIT ? ` ) - .all(...params) as WindowRow[]; - - return rows.map((row) => ({ - ...row, - alerts: readAlertsForWindow(db, row.windowStart, options.tail) - })); + .all(...params) as AddressRow[]; + + return { + totalAddresses: rows[0]?.totalAddresses ?? 0, + addresses: rows.map((row) => ({ + address: row.address, + tail: row.tail, + peakAlpha: row.peakAlpha, + peakWindowStart: row.peakWindowStart, + peakR2: row.peakR2, + lastSeen: row.lastSeen, + firstSeen: row.firstSeen, + timesFlagged: row.timesFlagged + })) + }; } export async function getAlertsFeedForDataset( datasetId: string, options: AlertsFeedOptions = {} ): Promise { + const horizonSeconds = HORIZON_SECONDS[options.horizon ?? DEFAULT_HORIZON]; if (shouldUseD1(options.platform)) { - return absentFeed(); + return absentFeed(horizonSeconds); } let alertsDbPath: string | undefined; try { const datasetDirectory = await resolveDatasetDirectory(datasetId); if (!datasetDirectory) { - return absentFeed(); + return absentFeed(horizonSeconds); } const path = await import('node:path'); alertsDbPath = path.join(datasetDirectory, 'alerts.sqlite'); const db = await getLocalDb(alertsDbPath); - const metadata = readFeedMetadata(db, datasetId); + const { thresholdHigh, thresholdLow } = readFeedMetadata(db, datasetId); const latestWindow = db .prepare( ` - SELECT window_start AS windowStart, processed_at AS processedAt + SELECT + window_start AS windowStart, + window_end AS windowEnd, + address_count AS addressCount, + processed_at AS processedAt FROM windows ORDER BY window_start DESC LIMIT 1 ` ) .get() as LatestWindowRow | undefined; + const result = latestWindow + ? readAddresses( + db, + options, + latestWindow.windowEnd - horizonSeconds, + thresholdHigh, + thresholdLow + ) + : { totalAddresses: 0, addresses: [] }; return { feed: { present: true, latestWindowStart: latestWindow?.windowStart ?? null, + latestWindowEnd: latestWindow?.windowEnd ?? null, + latestAddressCount: latestWindow?.addressCount ?? null, latestProcessedAt: latestWindow?.processedAt ?? null, - thresholds: { high: metadata.thresholdHigh, low: metadata.thresholdLow } + thresholds: { high: thresholdHigh, low: thresholdLow } }, - windows: readWindows(db, options) + horizonSeconds, + ...result }; } catch { if (alertsDbPath) { evictLocalDb(alertsDbPath); } - return absentFeed(); + return absentFeed(horizonSeconds); } } diff --git a/apps/web/src/lib/types/types.ts b/apps/web/src/lib/types/types.ts index f73a2a5..961023c 100644 --- a/apps/web/src/lib/types/types.ts +++ b/apps/web/src/lib/types/types.ts @@ -13,20 +13,19 @@ export interface DatasetSummariesResponse { export type AlertTail = 'high' | 'low'; -export interface AlertFeedAlert { +export type AlertHorizon = '1h' | '6h' | '24h' | '7d'; + +export type AlertSort = 'extreme' | 'recent'; + +export interface AlertFeedAddress { address: string; - alpha: number; tail: AlertTail; - rank: number; - r2: number; -} - -export interface AlertFeedWindow { - windowStart: number; - windowEnd: number; - addressCount: number; - alertCount: number; - alerts: AlertFeedAlert[]; + peakAlpha: number; + peakWindowStart: number; + peakR2: number; + lastSeen: number; + firstSeen: number; + timesFlagged: number; } export type AlertFeedStatus = @@ -34,13 +33,17 @@ export type AlertFeedStatus = | { present: true; latestWindowStart: number | null; + latestWindowEnd: number | null; + latestAddressCount: number | null; latestProcessedAt: number | null; thresholds: { high: number; low: number }; }; export interface AlertsFeedResponse { feed: AlertFeedStatus; - windows: AlertFeedWindow[]; + horizonSeconds: number; + totalAddresses: number; + addresses: AlertFeedAddress[]; } export type CoverageState = 'complete' | 'partial' | 'unknown'; diff --git a/apps/web/src/routes/alerts/+page.svelte b/apps/web/src/routes/alerts/+page.svelte index 70a8cf6..b4b47fc 100644 --- a/apps/web/src/routes/alerts/+page.svelte +++ b/apps/web/src/routes/alerts/+page.svelte @@ -2,18 +2,20 @@ import { afterNavigate } from '$app/navigation'; import { untrack } from 'svelte'; import type { PageProps } from './$types'; - import type { AlertsFeedResponse, AlertTail } from '$lib/types/types'; + import type { AlertHorizon, AlertsFeedResponse, AlertSort, AlertTail } from '$lib/types/types'; type TailSelection = 'all' | AlertTail; type ErrorResponse = { data: null; error: string }; - const PAGE_SIZE = 24; + const PAGE_SIZE = 100; + const MAX_LIMIT = 500; const REFRESH_INTERVAL_MS = 30_000; const LIVE_WINDOW_AGE_MS = 15 * 60_000; + const NEW_ADDRESS_AGE_SECONDS = 15 * 60; const CONTROL_GROUP_CLASS = 'dark:border-dark-border dark:bg-dark-subtle grid w-full gap-0.5 rounded-md border border-gray-200 bg-gray-50 p-1 sm:w-fit'; const CONTROL_BUTTON_CLASS = - 'flex min-h-7 items-center justify-center rounded px-2.5 py-0.5 text-center text-xs font-medium transition-colors focus:ring-2 focus:ring-blue-500 focus:outline-none sm:min-w-20'; + 'flex min-h-7 items-center justify-center rounded px-2.5 py-0.5 text-center text-xs font-medium transition-colors focus:ring-2 focus:ring-blue-500 focus:outline-none'; const CONTROL_BUTTON_INACTIVE_CLASS = 'text-gray-700 hover:text-gray-900 dark:text-gray-300 dark:hover:text-gray-100'; const CONTROL_BUTTON_ACTIVE_CLASS = 'bg-blue-600 text-white shadow-sm'; @@ -22,15 +24,20 @@ { value: 'high', label: 'High α' }, { value: 'low', label: 'Low α' } ]; + const HORIZON_OPTIONS: Array<{ value: AlertHorizon; label: string }> = [ + { value: '1h', label: '1h' }, + { value: '6h', label: '6h' }, + { value: '24h', label: '24h' }, + { value: '7d', label: '7d' } + ]; + const SORT_OPTIONS: Array<{ value: AlertSort; label: string }> = [ + { value: 'extreme', label: 'Most extreme' }, + { value: 'recent', label: 'Recent' } + ]; const timeFormatter = new Intl.DateTimeFormat(undefined, { hour: '2-digit', minute: '2-digit' }); - const dateFormatter = new Intl.DateTimeFormat(undefined, { - year: 'numeric', - month: 'short', - day: 'numeric' - }); const dateTimeFormatter = new Intl.DateTimeFormat(undefined, { dateStyle: 'medium', timeStyle: 'short' @@ -41,11 +48,13 @@ let activeDataset = $state(untrack(() => data.selectedDataset)); let feedResponse = $state(untrack(() => data.alerts)); let selectedTail = $state('all'); + let selectedHorizon = $state('24h'); + let selectedSort = $state('extreme'); + let limit = $state(PAGE_SIZE); let autoRefresh = $state(true); let now = $state(Date.now()); let loading = $state(false); - let loadingOlder = $state(false); - let hasOlder = $state(untrack(() => data.alerts.windows.length === PAGE_SIZE)); + let loadingMore = $state(false); let fetchError = $state(''); let copied = $state(false); let copyError = $state(''); @@ -55,49 +64,83 @@ data.datasets.find((dataset) => dataset.datasetId === data.selectedDataset)?.label ?? data.selectedDataset ); - const newestWindow = $derived(feedResponse.windows[0]); - const oldestWindow = $derived(feedResponse.windows.at(-1)); const feedCommand = $derived(`netflow-db feed ${data.selectedDataset || ''}`); + const canShowMore = $derived( + feedResponse.addresses.length < feedResponse.totalAddresses && limit < MAX_LIMIT + ); const statusText = $derived.by(() => { if (!feedResponse.feed.present) { return 'Feed not running'; } - if (!newestWindow) { + if ( + feedResponse.feed.latestWindowStart === null || + feedResponse.feed.latestWindowEnd === null || + feedResponse.feed.latestAddressCount === null + ) { return 'Feed idle · no windows processed'; } - const windowAge = now - newestWindow.windowEnd * 1000; + const windowStart = feedResponse.feed.latestWindowStart; + const windowEnd = feedResponse.feed.latestWindowEnd; + const windowAge = now - windowEnd * 1000; if (windowAge >= 0 && windowAge < LIVE_WINDOW_AGE_MS) { - return `Live · last window ${formatTime(newestWindow.windowStart)}–${formatTime(newestWindow.windowEnd)} · ${countFormatter.format(newestWindow.addressCount)} addresses scored`; + return `Live · last window ${formatTime(windowStart)}–${formatTime(windowEnd)} · ${countFormatter.format(feedResponse.feed.latestAddressCount)} addresses scored`; } - return `Feed idle · last window ${dateTimeFormatter.format(new Date(newestWindow.windowEnd * 1000))}`; + return `Feed idle · last window ${dateTimeFormatter.format(new Date(windowEnd * 1000))}`; }); function formatTime(timestamp: number): string { return timeFormatter.format(new Date(timestamp * 1000)); } - function formatDate(timestamp: number): string { - return dateFormatter.format(new Date(timestamp * 1000)); + function formatRelativeTime(timestamp: number): string { + const ageMs = Math.max(0, now - timestamp * 1000); + if (ageMs < 60_000) { + return 'just now'; + } + if (ageMs < 60 * 60_000) { + return `${Math.floor(ageMs / 60_000)} min ago`; + } + if (ageMs < 24 * 60 * 60_000) { + return `${Math.floor(ageMs / (60 * 60_000))} h ago`; + } + return dateTimeFormatter.format(new Date(timestamp * 1000)); + } + + function isNewAddress(firstSeen: number): boolean { + return ( + feedResponse.feed.present && + feedResponse.feed.latestWindowStart !== null && + firstSeen >= feedResponse.feed.latestWindowStart - NEW_ADDRESS_AGE_SECONDS + ); } - function feedUrl(tail: TailSelection, before?: number): string { + function feedUrl( + tail: TailSelection, + horizon: AlertHorizon, + sort: AlertSort, + requestLimit: number + ): string { const params = [ `dataset=${encodeURIComponent(data.selectedDataset)}`, - `limitWindows=${PAGE_SIZE}` + `horizon=${horizon}`, + `sort=${sort}`, + `limit=${requestLimit}` ]; if (tail !== 'all') { params.push(`tail=${tail}`); } - if (before !== undefined) { - params.push(`before=${before}`); - } return `/api/alerts?${params.join('&')}`; } - async function requestFeed(tail: TailSelection, before?: number): Promise { - const response = await fetch(feedUrl(tail, before)); + async function requestFeed( + tail: TailSelection, + horizon: AlertHorizon, + sort: AlertSort, + requestLimit: number + ): Promise { + const response = await fetch(feedUrl(tail, horizon, sort, requestLimit)); const payload = (await response.json()) as AlertsFeedResponse | ErrorResponse; if (!response.ok || 'error' in payload) { throw new Error('error' in payload ? payload.error : 'Failed to load alerts feed'); @@ -105,59 +148,37 @@ return payload; } - function mergeWindows( - latest: AlertsFeedResponse, - current: AlertsFeedResponse - ): AlertsFeedResponse { - if (!latest.feed.present) { - return latest; - } - - return { - feed: latest.feed, - windows: mergeWindowLists(latest.windows, current.windows) - }; - } - - function mergeWindowLists( - preferred: AlertsFeedResponse['windows'], - additional: AlertsFeedResponse['windows'] - ): AlertsFeedResponse['windows'] { - const windows = [...preferred]; - for (const window of additional) { - if (!windows.some((candidate) => candidate.windowStart === window.windowStart)) { - windows.push(window); - } - } - return windows.sort((left, right) => right.windowStart - left.windowStart); - } - - async function loadFirstPage(tail: TailSelection, preserveOlder: boolean): Promise { + async function loadFeed( + tail: TailSelection, + horizon: AlertHorizon, + sort: AlertSort, + requestLimit: number, + showLoading: boolean + ): Promise { if (!data.selectedDataset) { - return; + return false; } const generation = ++requestGeneration; - if (!preserveOlder) { + if (showLoading) { loading = true; } try { - const nextResponse = await requestFeed(tail); + const nextResponse = await requestFeed(tail, horizon, sort, requestLimit); if (generation !== requestGeneration) { - return; + return false; } - feedResponse = preserveOlder ? mergeWindows(nextResponse, feedResponse) : nextResponse; - if (!preserveOlder) { - hasOlder = nextResponse.windows.length === PAGE_SIZE; - } + feedResponse = nextResponse; fetchError = ''; + return true; } catch (error) { if (generation === requestGeneration) { fetchError = error instanceof Error ? error.message : 'Failed to load alerts feed'; } + return false; } finally { - if (generation === requestGeneration && !preserveOlder) { + if (generation === requestGeneration && showLoading) { loading = false; } } @@ -168,50 +189,51 @@ return; } - loadingOlder = false; selectedTail = tail; - await loadFirstPage(tail, false); + limit = PAGE_SIZE; + await loadFeed(tail, selectedHorizon, selectedSort, limit, true); + } + + async function selectHorizon(horizon: AlertHorizon): Promise { + if (horizon === selectedHorizon) { + return; + } + + selectedHorizon = horizon; + limit = PAGE_SIZE; + await loadFeed(selectedTail, horizon, selectedSort, limit, true); + } + + async function selectSort(sort: AlertSort): Promise { + if (sort === selectedSort) { + return; + } + + selectedSort = sort; + limit = PAGE_SIZE; + await loadFeed(selectedTail, selectedHorizon, sort, limit, true); } async function refreshFeed(): Promise { - if (loading || loadingOlder) { + if (loading || loadingMore) { return; } - await loadFirstPage(selectedTail, true); + await loadFeed(selectedTail, selectedHorizon, selectedSort, limit, false); } - async function loadOlderWindows(): Promise { - if (!oldestWindow || loading || loadingOlder || !hasOlder) { + async function showMore(): Promise { + if (!canShowMore || loading || loadingMore) { return; } - const generation = ++requestGeneration; - loadingOlder = true; + const nextLimit = Math.min(MAX_LIMIT, limit + PAGE_SIZE); + loadingMore = true; try { - const olderResponse = await requestFeed(selectedTail, oldestWindow.windowStart); - if (generation !== requestGeneration) { - return; - } - - if (!olderResponse.feed.present) { - feedResponse = olderResponse; - hasOlder = false; - } else { - feedResponse = { - feed: olderResponse.feed, - windows: mergeWindowLists(feedResponse.windows, olderResponse.windows) - }; - hasOlder = olderResponse.windows.length === PAGE_SIZE; - } - fetchError = ''; - } catch (error) { - if (generation === requestGeneration) { - fetchError = error instanceof Error ? error.message : 'Failed to load older windows'; + if (await loadFeed(selectedTail, selectedHorizon, selectedSort, nextLimit, false)) { + limit = nextLimit; } } finally { - if (generation === requestGeneration) { - loadingOlder = false; - } + loadingMore = false; } } @@ -235,9 +257,12 @@ activeDataset = data.selectedDataset; feedResponse = data.alerts; selectedTail = 'all'; + selectedHorizon = '24h'; + selectedSort = 'extreme'; + limit = PAGE_SIZE; + now = Date.now(); loading = false; - loadingOlder = false; - hasOlder = data.alerts.windows.length === PAGE_SIZE; + loadingMore = false; fetchError = ''; copied = false; copyError = ''; @@ -334,30 +359,83 @@ {/if} {:else} -
-
- {#each TAIL_OPTIONS as option (option.value)} - + {/each} +
+
+ +
+ Horizon +
+ {#each HORIZON_OPTIONS as option (option.value)} + + {/each} +
+
+ +
+ Sort +
- {option.label} - - {/each} + {#each SORT_OPTIONS as option (option.value)} + + {/each} +
+
{#if isNewAddress(alert.firstSeen)} new diff --git a/apps/web/tests/routes/api-alerts.test.ts b/apps/web/tests/routes/api-alerts.test.ts index a87bb56..d720c79 100644 --- a/apps/web/tests/routes/api-alerts.test.ts +++ b/apps/web/tests/routes/api-alerts.test.ts @@ -117,6 +117,9 @@ function seedAlerts(fixture: Fixture): void { ) VALUES (?, ?, ?, ?, ?, ?, 24) `); insertAlert.run(113_899, 'outside-24h', 12, 'high', 1, 0.5); + // history for high-address outside every horizon: firstSeen must be + // retention-wide (113_899) while in-horizon aggregates ignore this row. + insertAlert.run(113_899, 'high-address', 2.05, 'high', 2, 0.5); insertAlert.run(196_600, 'old-high', 6, 'high', 1, 0.6); insertAlert.run(196_700, 'repeat', 2.5, 'high', 1, 0.7); insertAlert.run(199_700, 'repeat', 4.5, 'high', 1, 0.91); @@ -188,7 +191,8 @@ describe('/api/alerts GET', () => { peakWindowStart: LATEST_WINDOW_START, peakR2: 0.93, lastSeen: LATEST_WINDOW_START, - firstSeen: LATEST_WINDOW_START, + // retention-wide, not horizon-scoped: the 113_899 history row + firstSeen: 113_899, timesFlagged: 1 }, { From 6a1e1a5b7031d4cb30452e90aca842e0d7f07880 Mon Sep 17 00:00:00 2001 From: flamboh Date: Fri, 21 Aug 2026 00:39:04 -0700 Subject: [PATCH 09/14] feat(web): show peak and latest alpha per alert address --- apps/web/src/lib/server/alerts.ts | 17 ++++++++++++-- apps/web/src/lib/types/types.ts | 2 ++ apps/web/src/routes/alerts/+page.svelte | 28 +++++++++++++++++------- apps/web/tests/lib/server/alerts.test.ts | 4 ++++ apps/web/tests/routes/api-alerts.test.ts | 4 ++++ 5 files changed, 45 insertions(+), 10 deletions(-) diff --git a/apps/web/src/lib/server/alerts.ts b/apps/web/src/lib/server/alerts.ts index f13f2b2..a7e9892 100644 --- a/apps/web/src/lib/server/alerts.ts +++ b/apps/web/src/lib/server/alerts.ts @@ -335,9 +335,20 @@ function readAddresses( PARTITION BY address ORDER BY severity DESC, window_start DESC, rank ASC ) AS peak_rank, + ROW_NUMBER() OVER ( + PARTITION BY address + ORDER BY window_start DESC, rank ASC + ) AS recent_rank, MAX(window_start) OVER (PARTITION BY address) AS last_seen, COUNT(*) OVER (PARTITION BY address) AS times_flagged FROM scoped + ), + enriched AS ( + SELECT + ranked.*, + MAX(CASE WHEN recent_rank = 1 THEN alpha END) + OVER (PARTITION BY address) AS latest_alpha + FROM ranked ) SELECT address, @@ -345,15 +356,16 @@ function readAddresses( alpha AS peakAlpha, window_start AS peakWindowStart, r2 AS peakR2, + latest_alpha AS latestAlpha, last_seen AS lastSeen, -- Novelty is judged against the whole retained history, not the -- horizon or tail filter: "new" means never flagged before at all. (SELECT MIN(history.window_start) FROM alerts AS history - WHERE history.address = ranked.address) AS firstSeen, + WHERE history.address = enriched.address) AS firstSeen, times_flagged AS timesFlagged, COUNT(*) OVER () AS totalAddresses - FROM ranked + FROM enriched WHERE peak_rank = 1 ORDER BY ${sortOrder} LIMIT ? @@ -369,6 +381,7 @@ function readAddresses( peakAlpha: row.peakAlpha, peakWindowStart: row.peakWindowStart, peakR2: row.peakR2, + latestAlpha: row.latestAlpha, lastSeen: row.lastSeen, firstSeen: row.firstSeen, timesFlagged: row.timesFlagged diff --git a/apps/web/src/lib/types/types.ts b/apps/web/src/lib/types/types.ts index 961023c..07ff854 100644 --- a/apps/web/src/lib/types/types.ts +++ b/apps/web/src/lib/types/types.ts @@ -23,6 +23,8 @@ export interface AlertFeedAddress { peakAlpha: number; peakWindowStart: number; peakR2: number; + /** Alpha of the address's most recent crossing within the horizon. */ + latestAlpha: number; lastSeen: number; firstSeen: number; timesFlagged: number; diff --git a/apps/web/src/routes/alerts/+page.svelte b/apps/web/src/routes/alerts/+page.svelte index eaad681..764de81 100644 --- a/apps/web/src/routes/alerts/+page.svelte +++ b/apps/web/src/routes/alerts/+page.svelte @@ -365,11 +365,11 @@
- Tail + Alpha
{#each TAIL_OPTIONS as option (option.value)}
- {#if loading} -

Updating alerts…

- {/if} - {#if feedResponse.addresses.length === 0}
{:else}
{#each feedResponse.addresses as alert (alert.address)} @@ -484,9 +481,24 @@
- + peak α {alert.peakAlpha.toFixed(3)} + + latest α {alert.latestAlpha.toFixed(3)} + { peakAlpha: 3.9, peakWindowStart: 1_700_000_300, peakR2: 0.98, + latestAlpha: 3.9, lastSeen: 1_700_000_300, firstSeen: 1_700_000_300, timesFlagged: 1 @@ -174,6 +175,7 @@ describe('alerts server helper', () => { peakAlpha: 3.7, peakWindowStart: 1_700_000_300, peakR2: 0.94, + latestAlpha: 3.7, lastSeen: 1_700_000_300, firstSeen: 1_700_000_300, timesFlagged: 1 @@ -184,6 +186,7 @@ describe('alerts server helper', () => { peakAlpha: 0.2, peakWindowStart: 1_700_000_300, peakR2: 0.89, + latestAlpha: 0.2, lastSeen: 1_700_000_300, firstSeen: 1_700_000_300, timesFlagged: 1 @@ -194,6 +197,7 @@ describe('alerts server helper', () => { peakAlpha: 0.21, peakWindowStart: 1_700_000_000, peakR2: 0.91, + latestAlpha: 0.21, lastSeen: 1_700_000_000, firstSeen: 1_700_000_000, timesFlagged: 1 diff --git a/apps/web/tests/routes/api-alerts.test.ts b/apps/web/tests/routes/api-alerts.test.ts index d720c79..95868a4 100644 --- a/apps/web/tests/routes/api-alerts.test.ts +++ b/apps/web/tests/routes/api-alerts.test.ts @@ -180,6 +180,7 @@ describe('/api/alerts GET', () => { peakAlpha: 4.5, peakWindowStart: 199_700, peakR2: 0.91, + latestAlpha: 2.2, lastSeen: LATEST_WINDOW_START, firstSeen: 196_700, timesFlagged: 3 @@ -190,6 +191,7 @@ describe('/api/alerts GET', () => { peakAlpha: 2.8, peakWindowStart: LATEST_WINDOW_START, peakR2: 0.93, + latestAlpha: 2.8, lastSeen: LATEST_WINDOW_START, // retention-wide, not horizon-scoped: the 113_899 history row firstSeen: 113_899, @@ -201,6 +203,7 @@ describe('/api/alerts GET', () => { peakAlpha: 0, peakWindowStart: 199_700, peakR2: 0.87, + latestAlpha: 0, lastSeen: 199_700, firstSeen: 199_700, timesFlagged: 1 @@ -211,6 +214,7 @@ describe('/api/alerts GET', () => { peakAlpha: 0.1, peakWindowStart: LATEST_WINDOW_START, peakR2: 0.95, + latestAlpha: 0.1, lastSeen: LATEST_WINDOW_START, firstSeen: 199_700, timesFlagged: 2 From 45aea5911cd93838b1f842bd3172da8f5db36acb Mon Sep 17 00:00:00 2001 From: flamboh Date: Fri, 21 Aug 2026 00:45:20 -0700 Subject: [PATCH 10/14] feat(web): tabular alert rows and visible thresholds --- apps/web/src/routes/alerts/+page.svelte | 163 +++++++++++++++--------- 1 file changed, 104 insertions(+), 59 deletions(-) diff --git a/apps/web/src/routes/alerts/+page.svelte b/apps/web/src/routes/alerts/+page.svelte index 764de81..1f748e6 100644 --- a/apps/web/src/routes/alerts/+page.svelte +++ b/apps/web/src/routes/alerts/+page.svelte @@ -435,6 +435,27 @@ {/each}
+ + {#if feedResponse.feed.present && feedResponse.feed.thresholds} +
+ Thresholds +
+ + α ≥ {feedResponse.feed.thresholds.high} + + + α ≤ {feedResponse.feed.thresholds.low} + +
+
+ {/if}