diff --git a/CHANGELOG.md b/CHANGELOG.md index 01a8a71..3823c77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ is `0`, minor versions may carry breaking changes — they are called out here. ## [Unreleased] +### Fixed — a window on a quiet source closes +- **A window fed by a source that goes quiet closes**: a detect unit sets an + idle grace, `VEJAS_IDLE_CLOSE_SECS` (60 seconds by default, 0 turns it + off). Once a subject's type has sent nothing that long, its event time + moves on with the wall clock, less the grace, and the unit checks while it + has nothing to read. A brute force on a sparse source (VPN logons, one + application's log) was raised only when that source spoke again; it is now + raised about a grace after its window ends. The types' clocks are in the + unit's snapshot, so a restarted unit closes the windows it restored on time + too (varpulis #286). `e2e/detect` D7 covers it: three VPN failures, then + silence; the previous runtime never raises it. + ## [0.3.2] — 2026-09-23 ### Fixed — detect units, from the Varpulis engine (varpulis #284, #285) diff --git a/core/Cargo.lock b/core/Cargo.lock index 9f6dc61..779344b 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -1846,7 +1846,7 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "varpulis-core" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "chrono", "indexmap", @@ -1863,7 +1863,7 @@ dependencies = [ [[package]] name = "varpulis-dead-letter" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "chrono", "serde", @@ -1875,7 +1875,7 @@ dependencies = [ [[package]] name = "varpulis-engine" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "serde_json", "thiserror", @@ -1887,7 +1887,7 @@ dependencies = [ [[package]] name = "varpulis-hamlet" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "rustc-hash", "smallvec", @@ -1898,7 +1898,7 @@ dependencies = [ [[package]] name = "varpulis-parser" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "logos", "miette", @@ -1911,7 +1911,7 @@ dependencies = [ [[package]] name = "varpulis-pst" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "hashlink", "rustc-hash", @@ -1921,7 +1921,7 @@ dependencies = [ [[package]] name = "varpulis-runtime" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "chrono", "hashlink", @@ -1949,7 +1949,7 @@ dependencies = [ [[package]] name = "varpulis-sase" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "chrono", "rustc-hash", @@ -1961,7 +1961,7 @@ dependencies = [ [[package]] name = "varpulis-simd" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "varpulis-core", ] @@ -1969,7 +1969,7 @@ dependencies = [ [[package]] name = "varpulis-zdd" version = "0.11.0" -source = "git+https://github.com/varpulis/varpulis?rev=4e3dbb5eca2c985bfc5b4f568d1118a96123ef45#4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" +source = "git+https://github.com/varpulis/varpulis?rev=92f07b966709bacc042ed97a71e16e8dd625c84d#92f07b966709bacc042ed97a71e16e8dd625c84d" dependencies = [ "rustc-hash", ] diff --git a/core/Cargo.toml b/core/Cargo.toml index 88d68f4..082906a 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -22,7 +22,7 @@ webpki-roots = "0.26" # The Varpulis CEP engine, embedded as a library (ADR-0031): compile a VPL # program, feed it events, publish its emits. No async runtime in its tree — # its own CI (scripts/check-engine-deps.py) fails if one ever appears. -varpulis-engine = { git = "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/varpulis/varpulis", rev = "4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" } +varpulis-engine = { git = "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/varpulis/varpulis", rev = "92f07b966709bacc042ed97a71e16e8dd625c84d" } [[bin]] name = "vejas-runtime" diff --git a/core/src/main.rs b/core/src/main.rs index d7a3bd5..992d7a2 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -1374,6 +1374,15 @@ fn supervise_vjs(handle: Arc, root: PathBuf) { }); } +/// Where a detect unit publishes an emit: the subject its `.to()` names, else +/// `.detect..`. +fn emit_subject(emit: &varpulis_engine::Emit, subj_root: &str, stem: &str) -> String { + emit.sink + .as_ref() + .and_then(|s| s.topic.clone()) + .unwrap_or_else(|| format!("{subj_root}.detect.{stem}.{}", emit.stream)) +} + /// A detect unit (ADR-0031): a VPL program run by the embedded Varpulis engine. /// /// One durable pull consumer per `.from()` subject, each on a thread of its @@ -1404,6 +1413,15 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { ); let snapshot_every_acks: u64 = env::var("VEJAS_SNAPSHOT_ACKS").ok().and_then(|v| v.parse().ok()).unwrap_or(10_000).max(1); + // A window whose sources go quiet closes about this long after its end: + // once a source has sent nothing for the grace, its event time moves on + // with the wall clock (varpulis `Program::set_idle_grace`), at every + // batch and on the ticks of an idle unit. 0 turns it off: the unit is + // then judged in event time alone, like a replay. + let idle_grace = match env::var("VEJAS_IDLE_CLOSE_SECS").ok().and_then(|v| v.parse::().ok()).unwrap_or(60) { + 0 => None, + secs => Some(Duration::from_secs(secs)), + }; let mut delay = Duration::from_secs(1); 'outer: loop { if !RUNNING.load(Ordering::SeqCst) || handle.stop.load(Ordering::SeqCst) { @@ -1494,6 +1512,9 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { None => {} } } + // After the restore: a restored window's sources start counting + // their silence from now. + program.set_idle_grace(idle_grace); match resume_seq { Some(seq) => { // The server cannot move an existing consumer: re-create it after the snapshot. @@ -1588,7 +1609,28 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { // the rest without waiting, bounded. let first = match rx.recv_timeout(Duration::from_millis(500)) { Ok(m) => m, - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + // Nothing to feed: close what time has closed on + // quiet sources. Those emits stand for no message, + // so there is nothing to ack; a crash before the + // next snapshot replays and closes them again. + if idle_grace.is_some() { + match program.tick() { + Ok(emits) if !emits.is_empty() => { + for emit in &emits { + let subject = emit_subject(emit, &subj_root, &stem); + if let Err(e) = nc.publish(&subject, emit.to_payload()) { + eprintln!("[vejas] {}: publish {subject}: {e}", handle.spec.name); + } + } + metrics::observe_tick_emits(&handle.spec.name, emits.len() as u64); + } + Ok(_) => {} + Err(e) => eprintln!("[vejas] {}: tick: {e}", handle.spec.name), + } + } + continue; + } Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { return Err("the consumer reader stopped".into()) } @@ -1650,13 +1692,7 @@ fn supervise_detect(handle: Arc, _root: PathBuf) { let mut ok = true; let mut subjects = Vec::with_capacity(emits.len()); for emit in &emits { - let subject = emit - .sink - .as_ref() - .and_then(|s| s.topic.clone()) - .unwrap_or_else(|| { - format!("{subj_root}.detect.{stem}.{}", emit.stream) - }); + let subject = emit_subject(emit, &subj_root, &stem); if let Err(e) = nc.publish(&subject, emit.to_payload()) { eprintln!("[vejas] {}: publish {subject}: {e}", handle.spec.name); ok = false; diff --git a/core/src/metrics.rs b/core/src/metrics.rs index efcbdbb..0c98ba7 100644 --- a/core/src/metrics.rs +++ b/core/src/metrics.rs @@ -80,6 +80,13 @@ pub fn observe(unit: &str, ok: bool, emits: u64, secs: f64) { } } +/// Emits published by a detect unit's tick: windows that time closed while +/// nothing arrived. They count as emits, not as processed events. +pub fn observe_tick_emits(unit: &str, emits: u64) { + let mut map = units().lock().unwrap(); + map.entry(unit.to_string()).or_default().emits += emits; +} + /// One pull round of the consumer loop and how many messages it returned. /// Empty rounds count too: they are the idle long-poll cadence. pub fn observe_fetch(unit: &str, messages: usize) { diff --git a/docs/book/src/concepts/detects.md b/docs/book/src/concepts/detects.md index 147be2c..7beca2d 100644 --- a/docs/book/src/concepts/detects.md +++ b/docs/book/src/concepts/detects.md @@ -54,9 +54,12 @@ machine it connected to: one alert, on `vx.alerts.lateral`. end**: on any event of those types, not only one of the window's own. A brute force counted per address on Security events is raised by the next Security event of any kind, even when the attacker got in and stopped. A - window fed by several types waits for the slowest; one that no event - follows stays open, so a source that can go quiet should send a heartbeat - of its own type. A stream whose events arrive late declares + window fed by several types waits for the slowest. A type that sends + nothing for `VEJAS_IDLE_CLOSE_SECS` (60 by default; 0 turns it off) has + its event time move on with the wall clock, less that grace, so a brute + force on a sparse source (VPN logons) is raised about a minute after its + window ends even if the source never speaks again; the unit also checks + while it has nothing to read. A stream whose events arrive late declares `.watermark(out_of_order: 30s)` to hold its windows that much longer. - **Types.** A string `event_type` in the payload names the event type; without it, the type is the one the `.from()` binding declares for that diff --git a/e2e/detect/run.sh b/e2e/detect/run.sh index fe49069..39971a1 100755 --- a/e2e/detect/run.sh +++ b/e2e/detect/run.sh @@ -18,6 +18,9 @@ # D6 count close a count per address closes on the next event of its source # past the window, from any address: a brute force whose # attacker got in and stopped is still raised +# D7 quiet source a count on a source that goes silent still closes: past +# the idle grace (VEJAS_IDLE_CLOSE_SECS, 1 s here) the +# source's event time moves on with the wall clock # D3 vpl-check the engine's verdict on the CLI: ok / refused, with exit codes # D4 topology /topology lists the units under "detects", lang vpl, running # @@ -118,10 +121,35 @@ stream Brute = Failed .to(Bus, topic: "vxt.brute") VPL +cat > "$ROOT/detects/quiet.vpl" <<'VPL' +connector Bus = nats ( + url: "ignored-by-vejas: the bus is NATS_URL" +) + +event Vpn: + ip: str + status: str + +stream Logons = Vpn + .from(Bus, topic: "vxt.vpn") + +stream Failed = Vpn + .where(status == "failure") + +stream Brute = Failed + .partition_by(ip) + .window(5s) + .aggregate(ip: last(ip), n: count()) + .where(n >= 3) + .emit(rule: "vpn_brute_force", ip: ip, n: n) + .to(Bus, topic: "vxt.vpnbrute") +VPL + start_nats() { "$NATSD" -js -sd "$WORK/js" -a 127.0.0.1 -p "$NATS_P" > /dev/null 2>&1 & NATS_PID=$!; sleep 0.5; } start_rt() { VEJAS_ROOT="$ROOT" NATS_URL="$URL" VEJAS_STREAM=TTEST VEJAS_SUBJECT_ROOT=vxt \ VEJAS_HTTP_ADDR="127.0.0.1:$HTTP_P" VEJAS_ACK_WAIT_SECS=1 VEJAS_SNAPSHOT_SECS="${SNAP_SECS:-1}" \ + VEJAS_IDLE_CLOSE_SECS=1 \ "$BIN" >> "$WORK/rt.log" 2>&1 & RT_PID=$! until curl -sf -o /dev/null "http://127.0.0.1:$HTTP_P/healthz"; do sleep 0.05; done @@ -157,14 +185,14 @@ echo "── D4 topology" deadline=$((SECONDS+10)) while [ "$SECONDS" -lt "$deadline" ]; do topo=$(curl -s "http://127.0.0.1:$HTTP_P/topology") - [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 3 ] && break + [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 4 ] && break sleep 0.2 done -python3 - "$topo" <<'PY' && ok "three detects listed, lang vpl, running" || bad "topology: $topo" +python3 - "$topo" <<'PY' && ok "four detects listed, lang vpl, running" || bad "topology: $topo" import json, sys t=json.loads(sys.argv[1]); d=t.get("detects", []) names=sorted(x["name"] for x in d) -assert names==["detect:bruteforce","detect:lateral","detect:threshold"], names +assert names==["detect:bruteforce","detect:lateral","detect:quiet","detect:threshold"], names assert all(x["lang"]=="vpl" for x in d), d assert all(x["status"]=="running" for x in d), [x["status"] for x in d] PY @@ -243,6 +271,22 @@ assert len(alerts)==1 and alerts[0]["ip"]=="10.0.0.66" and alerts[0]["n"]==3, al PY kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.brute" 2>/dev/null +echo "── D7 a count on a source that goes quiet still closes (idle grace)" +( timeout 30 "$NATS" -s "$URL" sub vxt.vpnbrute --raw > "$WORK/vpnbrute.txt" 2>/dev/null ) & SUB_PID=$! +sleep 0.5 +for s in 00 01 02; do + "$NATS" -s "$URL" pub vxt.vpn "{\"event_type\":\"Vpn\",\"@timestamp\":\"2026-09-21T19:00:${s}Z\",\"ip\":\"10.0.0.88\",\"status\":\"failure\"}" > /dev/null 2>&1 +done +# Nothing else is ever published on the VPN subject. +deadline=$((SECONDS+15)); while [ "$SECONDS" -lt "$deadline" ]; do grep -q '10.0.0.88' "$WORK/vpnbrute.txt" && break; sleep 0.2; done +sleep 1 +python3 - "$WORK/vpnbrute.txt" <<'PY' && ok "the silent source's brute force is raised once the grace has passed, with its count" || bad "quiet source: $(cat "$WORK/vpnbrute.txt")" +import json, sys +alerts=[json.loads(l) for l in open(sys.argv[1]) if l.startswith("{")] +assert len(alerts)==1 and alerts[0]["ip"]=="10.0.0.88" and alerts[0]["n"]==3, alerts +PY +kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.vpnbrute" 2>/dev/null + echo "── D5 a stateful sequence survives kill -9 (snapshot, resume by sequence)" ( timeout 60 "$NATS" -s "$URL" sub vxt.lateral --raw > "$WORK/lateral5.txt" 2>/dev/null ) & SUB_PID=$! sleep 0.5