correlate: port sliding-window + sequence-pair detect_* (batch 2)
ober
212299baaa7f34ff87512af455772d1df680e3d5
--- a/Makefile +++ b/Makefile @@ -259,8 +259,10 @@ frame-check: $(SCHEME) --libdirs $(LIBDIRS) --script examples/frame_check.ss # Cross-event correlation rules (secmon storage/mod.rs detect_*), pure cores over -# pre-shaped rows: brute_force/credential_stuffing (≥5 / 10min), dns_tunnel -# (≥50 / 5min), suspicious_cron (non-root). SQL fetch + format_ts are deferred. +# pre-shaped rows. Batch 1 GROUP-BY: brute_force/credential_stuffing (≥5 / 10min), +# dns_tunnel (≥50 / 5min), suspicious_cron (non-root). Batch 2 windows/pairs: +# recon_port_scan/data_exfil (sliding window), priv_escalation/persistence/ +# lateral (sequence pair), log_cover. SQL fetch + format_ts are deferred. correlate-check: $(SCHEME) --libdirs $(LIBDIRS) --script examples/correlate_check.ss --- a/README.md +++ b/README.md @@ -128,7 +128,7 @@ then crypto orchestration, then I/O / async / FFI (monitors, server, storage). | `geoip` (CSV GeoIP/ASN, IPv4+IPv6, binary-search range lookup, is_private) | `jsecmon/geoip.ss` | ✅ **untyped layer** — full port of secmon's `src/geoip.rs`: parse `start,end,country,asn,name` CSV rows (v4 + one-`::`-expanding v6), sort-by-start + binary-search lookup, RFC1918/loopback/link-local/multicast/ULA/CGNAT → synthetic `PRIVATE`. Pure parsing + integer math + file read, so untyped. `make geoip-check` runs secmon's geoip vectors. | | `storage` impossible_travel | `jsecmon/threats.ss` | ✅ **untyped layer** — geoip-gated (reads `SECMON_GEOIP_CSV`): pair a user's consecutive successful `auth_event`s, fire `high` when the two source IPs resolve to different countries within `SECMON_TRAVEL_GAP_MIN` (default 30). Private IPs are dropped before pairing. `make geoip-check` proves the fire + the gap/same-country/private/cross-user negatives. | | `buffer::ring` (StoredEvent ring buffer) | `jsecmon/buffer.ss` | ✅ **untyped layer** — port of secmon's `src/buffer/ring.rs`: the agent's bounded in-memory event ring. FIFO list + monotonic seq numbering, priority eviction (`event_severity_u8` table, drop lowest-severity oldest-first, oldest-critical last), seq/time-range polling, FIFO delivery-ack (`clear_before`), and the little-endian header codec (`seq u64 ∥ ts i64 ∥ sev u8 ∥ payload`). Pure mechanics, so untyped — the one security step, ECIES payload encryption, is FFI-deferred: the caller hands `buffer-store!` opaque ciphertext bytes. `make buffer-check` reproduces secmon's three ring tests (store/seq, priority eviction, FIFO-oldest) + codec round-trip. | -| `storage` correlation rules (`detect_*`) | `jsecmon/correlate.ss` | ✅ **untyped layer** — secmon's anomaly detectors are "run a SQL query, then map result rows → Anomaly". The SQL fetch (with its `json_extract` / `GROUP BY … HAVING`) is the deferred storage I/O; the pure, portable part is the correlation algorithm over the fetched rows + the Anomaly description, reimplemented here over pre-shaped row lists (which is exactly what `test_detect_*` exercise). **Batch 1** = the GROUP-BY rules: `detect_brute_force` (≥5 failed-auth per user / 10-min bucket), `detect_credential_stuffing` (≥5 **distinct** users per remote / 10-min), `detect_dns_tunnel` (≥50 queries per process / 5-min), `detect_suspicious_cron` (per-row: `scheduled_task_change` by a non-root user, missing user → `""` ≠ root → kept). `timestamp_ms / N` is integer division → `quotient`, bucket anomaly ts is `bucket*N`; COUNT(*) vs COUNT(DISTINCT) become first-seen-order tallies (treat the set, not order, as significant). Each detector returns "anomaly seeds" (an Anomaly minus the calendar `format_ts` fields, which the caller adds, and minus the `attack` tags from `anomaly-rule-attack`). `make correlate-check` reproduces those four tests + the threshold/bucket-boundary/distinct-dedup edges. | +| `storage` correlation rules (`detect_*`) | `jsecmon/correlate.ss` | ✅ **untyped layer** — secmon's anomaly detectors are "run a SQL query, then map result rows → Anomaly". The SQL fetch (with its `json_extract` / `GROUP BY … HAVING`) is the deferred storage I/O; the pure, portable part is the correlation algorithm over the fetched rows + the Anomaly description, reimplemented here over pre-shaped row lists (which is exactly what `test_detect_*` exercise). **Batch 1** = the GROUP-BY rules: `detect_brute_force` (≥5 failed-auth per user / 10-min bucket), `detect_credential_stuffing` (≥5 **distinct** users per remote / 10-min), `detect_dns_tunnel` (≥50 queries per process / 5-min), `detect_suspicious_cron` (per-row: `scheduled_task_change` by a non-root user, missing user → `""` ≠ root → kept). `timestamp_ms / N` is integer division → `quotient`, bucket anomaly ts is `bucket*N`; COUNT(*) vs COUNT(DISTINCT) become first-seen-order tallies (treat the set, not order, as significant). Each detector returns "anomaly seeds" (an Anomaly minus the calendar `format_ts` fields, which the caller adds, and minus the `attack` tags from `anomaly-rule-attack`). **Batch 2** = the sliding-window / sequence-pair rules over caller-sorted rows: `detect_recon_port_scan` (≥10 distinct remote ports / 5-min, same process) and `detect_data_exfil` (≥20 outbound connections / 5-min) share a `slide` helper reproducing secmon's `while i<len { expand j; if hit { i=j } else { i+=1 } }`; `detect_sequence_pair` (event A then first B on the same host with `a.ts < b.ts ≤ a.ts+window`, one match per A) backs the three chain rules `detect_priv_escalation_chain` (auth-success→priv-esc / 5-min, caller pre-filters A to `success=1`), `detect_persistence_after_access` (reverse_shell **or** webshell→persistence / 1h — two arms appended), `detect_lateral_after_shell` (reverse_shell→lateral / 1h); and `detect_log_cover` (any critical event→log_tampering / 1h, one per critical) carries the event type into the description. `gap_seconds` is `(quotient (- b a) 1000)`. `make correlate-check` reproduces all ten `test_detect_*` (incl. `_below_threshold` and `_outside_window`) + the threshold / bucket-boundary / distinct-dedup / window-edge / cross-host / B-before-A edges (60 cases). | | `server::protocol` (length-prefixed framing) | `jsecmon/frame.ss` | ✅ **untyped layer** — the wire framing the collector↔agent transport wraps every `ProtocolMessage` in: `frame_encode` prepends a 4-byte little-endian u32 length to the payload (`to_bytes`'s framing half), `frame_read_length` reads that header as an LE u32 (`read_length`), and `frame_decode` reproduces `from_bytes`'s two guards — `< 4` bytes → `data too short`, `< 4 + declared len` → `incomplete message` — then slices out `data[4 .. 4+len]`. Pure byte mechanics, so untyped; the bincode (de)serialization of the message *body* is the Rust-specific deferred piece (the caller (de)serializes the payload `frame_decode` hands back). LE u32 verified to round-trip `0x12345678` and `0xFFFFFFFF`. `make frame-check` pins the byte layout, the encode∘decode round-trip, both error guards, and the empty/zero-length-payload edge. | | `monitor::events::SuspiciousPatterns` (process-spawn classifier) | `jsecmon/suspicious.ss` | ✅ **untyped layer** — `check_suspicious(process, parent)`: shell-from-service, attack-tool-from-service (name exact-match or exe suffix), reverse-shell command-line patterns, and crypto-miner name/cmdline patterns, in secmon's order, returning the same reason string. Pure string classification like triage. Pins two corners the Rust depends on: a missing parent short-circuits to "clean" before any check, and `str::contains` is a *literal* substring test (so `python -c.*socket` is literal, not a regex). `make suspicious-check` reproduces secmon's two events.rs tests + the other three signals + both corners. | | `monitor::network::NetworkMonitor` (connection classifier) | `jsecmon/netconn.ss` | ✅ **untyped layer** — `check_suspicious(port, addr, process)`: known reverse-shell/C2/l33t port, ephemeral port (49152..65535) that is a round multiple of 1000, and a web-server process (nginx/apache/httpd/php-fpm) connecting to a non-private address, in secmon's order with the same reason string. Pure metadata classification. secmon hides the web-server names with `obfstr!` (same scheme as `typed/obfuscate.ss`); they decode to these plaintext literals at runtime. Pins the faithfulness quirk that the "private" prefix set is literal `{127. 10. 192.168. 172.}`, so `172.` matches all of 172.x, not just RFC1918 172.16/12. `make netconn-check` reproduces secmon's two network.rs tests + the full bad-port list + the high-port and web-server rules with private-address negatives. | --- a/examples/correlate_check.ss +++ b/examples/correlate_check.ss @@ -108,7 +108,102 @@ (check "all-root -> none" (length (detect-suspicious-cron (list (list "h1" 1 "x" "root")))) 0) +;; ── recon port scan: 12 distinct ports from nmap in 5min (test_detect_…) ───── +(displayln "detect-recon-port-scan:") +(def rps (detect-recon-port-scan + (for/collect ((i (in-range 12))) + (list "h1" (+ 1000 (* i 1000)) "nmap" (+ 1000 i))))) +(check "12 distinct ports -> 1 anomaly" (length rps) 1) +(check "rule" (rule (car rps)) "recon_port_scan") +(check "severity" (sev (car rps)) "medium") +(check "distinct-ports detail" (dval (car rps) 'distinct-ports) 12) +(check "ts = first event" (ts (car rps)) 1000) +;; threshold >=10 distinct: 10 fires, 9 does not (each a distinct port) +(check "10 distinct -> fires" + (length (detect-recon-port-scan + (for/collect ((i (in-range 10))) (list "h1" (+ 1000 i) "p" i)))) 1) +(check "9 distinct -> none" + (length (detect-recon-port-scan + (for/collect ((i (in-range 9))) (list "h1" (+ 1000 i) "p" i)))) 0) +;; 20 connections but only 5 distinct ports -> not a port scan +(check "20 conns, 5 ports -> none" + (length (detect-recon-port-scan + (for/collect ((i (in-range 20))) (list "h1" (+ 1000 i) "p" (modulo i 5))))) 0) + +;; ── data exfil: 22 connections from curl in 5min (test_detect_…) ───────────── +(displayln "detect-data-exfil:") +(def dx (detect-data-exfil + (for/collect ((i (in-range 22))) (list "h1" (+ 1000 (* i 100)) "curl")))) +(check "22 conns -> 1 anomaly" (length dx) 1) +(check "rule" (rule (car dx)) "data_exfil") +(check "connection-count detail" (dval (car dx) 'connection-count) 22) +(check "20 conns -> fires" + (length (detect-data-exfil + (for/collect ((i (in-range 20))) (list "h1" (+ 1000 i) "p")))) 1) +(check "19 conns -> none" + (length (detect-data-exfil + (for/collect ((i (in-range 19))) (list "h1" (+ 1000 i) "p")))) 0) + +;; ── priv-escalation chain: auth success then priv-esc within 5min (test) ───── +(displayln "detect-priv-escalation-chain:") +(def pe (detect-priv-escalation-chain (list (list "h1" 1000)) + (list (list "h1" (+ 1000 120000))))) +(check "success then priv-esc 2min -> 1" (length pe) 1) +(check "rule" (rule (car pe)) "priv_escalation_chain") +(check "severity" (sev (car pe)) "critical") +(check "gap-seconds = 120" (dval (car pe) 'gap-seconds) 120) +;; outside the 5-min window -> none (test_…_outside_window) +(check "priv-esc 10min later -> none" + (length (detect-priv-escalation-chain (list (list "h1" 1000)) + (list (list "h1" (+ 1000 600000))))) 0) +;; window edge: b exactly at a.ts+window fires; one ms past does not +(check "B at window edge -> fires" + (length (detect-priv-escalation-chain (list (list "h1" 1000)) + (list (list "h1" (+ 1000 300000))))) 1) +(check "B one ms past edge -> none" + (length (detect-priv-escalation-chain (list (list "h1" 1000)) + (list (list "h1" (+ 1001 300000))))) 0) +;; B before A (b.ts <= a.ts) -> none (strict b.ts > a.ts) +(check "B before A -> none" + (length (detect-priv-escalation-chain (list (list "h1" 5000)) + (list (list "h1" 1000)))) 0) +;; different host -> no pairing +(check "cross-host -> none" + (length (detect-priv-escalation-chain (list (list "h1" 1000)) + (list (list "h2" 2000)))) 0) + +;; ── persistence after access: revshell|webshell then persistence/1h (test) ─── +(displayln "detect-persistence-after-access:") +(check "revshell then persistence -> 1" + (length (detect-persistence-after-access + (list (list "h1" 1000)) '() (list (list "h1" (+ 1000 60000))))) 1) +;; webshell arm fires too, with its own description +(def pw (detect-persistence-after-access + '() (list (list "h1" 1000)) (list (list "h1" (+ 1000 60000))))) +(check "webshell arm -> 1" (length pw) 1) +(check "webshell description" + (desc (car pw)) "Webshell followed by persistence on h1") + +;; ── log cover: critical event then log tampering within 1h (test) ──────────── +(displayln "detect-log-cover:") +(def lc (detect-log-cover (list (list "h1" 1000 "reverse_shell")) + (list (list "h1" (+ 1000 300000))))) +(check "crit then tamper -> 1" (length lc) 1) +(check "rule" (rule (car lc)) "log_cover") +(check "description" + (desc (car lc)) + "Critical reverse_shell event followed by log tampering on h1") +(check "gap-seconds = 300" (dval (car lc) 'gap-seconds) 300) +(check "trigger-event detail" (dval (car lc) 'trigger-event) "reverse_shell") + +;; ── lateral after shell: reverse shell then lateral move within 1h (test) ──── +(displayln "detect-lateral-after-shell:") +(def la (detect-lateral-after-shell (list (list "h1" 1000)) + (list (list "h1" (+ 1000 120000))))) +(check "revshell then lateral -> 1" (length la) 1) +(check "rule" (rule (car la)) "lateral_after_shell") + (newline) (if (= fails 0) - (displayln "OK: correlate matches secmon's storage/mod.rs detect_* (batch 1).") + (displayln "OK: correlate matches secmon's storage/mod.rs detect_* (batch 1+2).") (begin (displayln fails " FAILURES") (exit 1))) --- a/jsecmon/correlate.ss +++ b/jsecmon/correlate.ss @@ -23,6 +23,15 @@ ;;; detect-dns-tunnel rows=((host proc-name ts) …) ≥50 / 5min ;;; detect-suspicious-cron rows=((host ts summary user) …) user≠"root" ;;; +;;; Batch 2 — the sliding-window / sequence-pair rules, over rows the caller has +;;; sorted (recon/exfil by host,proc,ts; pairs by host,ts) the way the SQL does: +;;; detect-recon-port-scan rows=((host ts proc port) …) ≥10 distinct ports/5min +;;; detect-data-exfil rows=((host ts proc) …) ≥20 connections/5min +;;; detect-sequence-pair a/b=((host ts) …) first B per A in window +;;; detect-priv-escalation-chain / -persistence-after-access / -lateral-after-shell +;;; thin wrappers over detect-sequence-pair with the rule's types/window/text +;;; detect-log-cover a=((host ts type)…) b=((host ts)…) crit then tamper/1h +;;; ;;; Faithfulness notes: ;;; * SQL `timestamp_ms / N` is integer division on a positive epoch → quotient; ;;; the bucket's anomaly timestamp is bucket*N (start of the 10-/5-min window). @@ -43,7 +52,11 @@ (library (jsecmon correlate) (export detect-brute-force detect-credential-stuffing - detect-dns-tunnel detect-suspicious-cron) + detect-dns-tunnel detect-suspicious-cron + detect-recon-port-scan detect-data-exfil + detect-sequence-pair detect-priv-escalation-chain + detect-persistence-after-access detect-lateral-after-shell + detect-log-cover) (import (except (chezscheme) make-hash-table hash-table? sort sort! @@ -168,4 +181,155 @@ (cons 'host host) (cons 'details (list (cons 'user user) (cons 'summary summary))))))) - rows))) + rows)) + + ;; ── sliding window over rows sorted by (host, proc, ts) ───────────────────── + ;; Each row is (host ts proc . rest). From i, expand j over the maximal run of + ;; same host+proc with ts ≤ start.ts+window-ms; measure the run [i,j); if it + ;; meets the threshold emit (mk-seed start measure) and jump i=j, else i+=1 — + ;; exactly secmon's `while i<len { … if cond { i=j } else { i+=1 } }`. + (def (slide rows window-ms measure threshold mk-seed) + (let* ((v (list->vector rows)) (n (vector-length v))) + (let loop ((i 0) (acc '())) + (if (>= i n) + (reverse acc) + (let ((start (vector-ref v i))) + (let scan ((j i)) + (if (and (< j n) + (let ((r (vector-ref v j))) + (and (string=? (car r) (car start)) + (string=? (caddr r) (caddr start)) + (<= (cadr r) (+ (cadr start) window-ms))))) + (scan (+ j 1)) + (let ((m (measure v i j))) + (if (>= m threshold) + (loop j (cons (mk-seed start m) acc)) + (loop (+ i 1) acc)))))))))) + + (def (distinct-count v i j key-fn) + (let ((h (make-hash-table))) + (let lp ((k i)) + (when (< k j) (hash-put! h (str (key-fn (vector-ref v k))) #t) (lp (+ k 1)))) + (length (hash-keys h)))) + + ;; ── Rule 5: recon port scan — ≥10 distinct remote ports / 5min, same proc ─── + (def (detect-recon-port-scan rows) + (slide rows 300000 + (lambda (v i j) (distinct-count v i j cadddr)) ;; distinct remote_port + 10 + (lambda (start ports) + (let ((host (car start)) (ts (cadr start)) (proc (caddr start))) + (list (cons 'rule "recon_port_scan") + (cons 'description + (str "'" proc "' connected to " ports + " distinct ports in 5min on " host)) + (cons 'severity "medium") + (cons 'timestamp-ms ts) + (cons 'host host) + (cons 'details + (list (cons 'process-name proc) + (cons 'distinct-ports ports) + (cons 'window-start-ms ts)))))))) + + ;; ── Rule 6: data exfil — ≥20 outbound connections / 5min, same process ────── + (def (detect-data-exfil rows) + (slide rows 300000 + (lambda (v i j) (- j i)) ;; COUNT(*) in window + 20 + (lambda (start count) + (let ((host (car start)) (ts (cadr start)) (proc (caddr start))) + (list (cons 'rule "data_exfil") + (cons 'description + (str "'" proc "' made " count + " outbound connections in 5min on " host)) + (cons 'severity "high") + (cons 'timestamp-ms ts) + (cons 'host host) + (cons 'details + (list (cons 'process-name proc) + (cons 'connection-count count) + (cons 'window-start-ms ts)))))))) + + ;; first B-row on the same host with a.ts < b.ts ≤ a.ts+window, else #f. + (def (first-following host ts events-b window-ms) + (let loop ((bs events-b)) + (cond ((null? bs) #f) + ((let ((b (car bs))) + (and (string=? (car b) host) + (> (cadr b) ts) + (<= (cadr b) (+ ts window-ms)))) + (car bs)) + (else (loop (cdr bs)))))) + + ;; ── Generic sequence pair: event A followed by event B within window ──────── + ;; events-a / events-b are ((host ts) …); one match per A (first B, then break). + (def (detect-sequence-pair events-a events-b type-a type-b window-ms + rule-name desc-prefix severity) + (filter-map + (lambda (a) + (let* ((host (car a)) (ats (cadr a)) + (b (first-following host ats events-b window-ms))) + (and b + (let ((bts (cadr b))) + (list (cons 'rule rule-name) + (cons 'description (str desc-prefix " on " host)) + (cons 'severity severity) + (cons 'timestamp-ms ats) + (cons 'host host) + (cons 'details + (list (cons 'event-a type-a) + (cons 'event-a-time-ms ats) + (cons 'event-b type-b) + (cons 'event-b-time-ms bts) + (cons 'gap-seconds (quotient (- bts ats) 1000))))))))) + events-a)) + + ;; ── Rule 7: priv-escalation chain — auth success then priv-esc within 5min ── + ;; events-a is pre-filtered to successful auth (the SQL `$.success = 1`). + (def (detect-priv-escalation-chain auth-success-a priv-esc-b) + (detect-sequence-pair auth-success-a priv-esc-b + "auth_event" "privilege_escalation" 300000 + "priv_escalation_chain" + "Auth success followed by privilege escalation" "critical")) + + ;; ── Rule 8: persistence after access — revshell|webshell then persistence/1h ─ + (def (detect-persistence-after-access reverse-shell-a webshell-a persistence-b) + (append + (detect-sequence-pair reverse-shell-a persistence-b + "reverse_shell" "persistence_event" 3600000 + "persistence_after_access" + "Reverse shell followed by persistence" "critical") + (detect-sequence-pair webshell-a persistence-b + "webshell" "persistence_event" 3600000 + "persistence_after_access" + "Webshell followed by persistence" "critical"))) + + ;; ── Rule 10: lateral after shell — reverse shell then lateral move within 1h ─ + (def (detect-lateral-after-shell reverse-shell-a lateral-b) + (detect-sequence-pair reverse-shell-a lateral-b + "reverse_shell" "lateral_movement" 3600000 + "lateral_after_shell" + "Reverse shell followed by lateral movement" "critical")) + + ;; ── Rule 9: log cover — any critical event then log_tampering within 1h ───── + ;; events-a is ((host ts event-type) …) of critical events; b is log_tampering. + (def (detect-log-cover crit-a tamper-b) + (filter-map + (lambda (a) + (let* ((host (car a)) (ats (cadr a)) (etype (caddr a)) + (b (first-following host ats tamper-b 3600000))) + (and b + (let ((bts (cadr b))) + (list (cons 'rule "log_cover") + (cons 'description + (str "Critical " etype + " event followed by log tampering on " host)) + (cons 'severity "critical") + (cons 'timestamp-ms ats) + (cons 'host host) + (cons 'details + (list (cons 'trigger-event etype) + (cons 'trigger-time-ms ats) + (cons 'tamper-time-ms bts) + (cons 'gap-seconds (quotient (- bts ats) 1000))))))))) + crit-a)))