threats: sequence-pair chain detectors
Jaime Fournier <jaimef@linbsd.org>
165247cfcdb532208e0cfa7224faab15b93a5ec4
--- a/README.md +++ b/README.md @@ -79,5 +79,6 @@ then crypto orchestration, then I/O / async / FFI (monitors, server, storage). | `crypto::ecies` | — | ⏳ FFI-delegated; orchestration only | | `storage` (events table, store/query/filters) | `jsecmon/storage.ss` | ✅ **untyped layer** — SQLite event store on `(std db sqlite-native)` (rusqlite): secmon's schema (events + indexes + collector_state), `store-event` INSERT-OR-IGNORE dedup, and the full EventFilter WHERE builder (host/type/severity/since/until/pid/process_name LIKE/search/exclude_event_ids). `query-events` returns row hashes with `data` parsed from JSON, so detect/triage/analytics consume them directly. `make storage-check` round-trips store→query→detect→analytics (host risk 30, same as `detect-check`). | | `storage` SQL-aggregation detectors (brute_force, credential_stuffing, dns_tunnel, suspicious_cron, recon_port_scan, data_exfil) | `jsecmon/threats.ss` | ✅ **untyped layer** — secmon's `run_detections` family: the time-bucket GROUP BY/HAVING rules and the two 5-min sliding-window rules, run as SQL (json_extract) over a `(jsecmon storage)` handle. `make threats-check` reproduces secmon's six detection-rule test vectors. Remaining: the sequence/kill-chain rules (priv_escalation_chain, persistence_after_access, log_cover, lateral_after_shell, impossible_travel) + frequency/severity/off-hours aggregates. | -| `storage` sequence/chain detectors (priv_escalation, kill_chain, impossible_travel, …) | — | ⏳ next: sequence-pair + multi-phase window rules | +| `storage` sequence/chain detectors (priv_escalation_chain, persistence_after_access, log_cover, lateral_after_shell) | `jsecmon/threats.ss` | ✅ **untyped layer** — secmon's `detect_sequence_pair` family: event A then event B within a window on the same host (auth-success→priv-esc /5min, reverse-shell/webshell→persistence /1h, any-critical→log-tampering /1h, shell→lateral /1h). Reproduces secmon's chain test vectors incl. the outside-window negative. | +| `storage` impossible_travel / kill_chain / freq+severity+off-hours aggregates | — | ⏳ impossible_travel needs a geoip CSV loader; kill_chain + frequency/severity/off-hours are the remaining `detect_anomalies` aggregates | | monitors / server / ebpf / dtrace | — | ⏳ I/O+async+FFI, last | --- a/examples/threats_check.ss +++ b/examples/threats_check.ss @@ -119,6 +119,59 @@ (check " connection_count" (hash-get (hash-get (car r) "details") "connection_count") 22)) (store-close db)) +;; ── priv_escalation_chain: auth success then privilege_escalation /5min -> 1 ── +(displayln "priv_escalation_chain:") +(let ((db (fresh))) + (store-event db 1 "h1" "s1" 1000 "auth_event" "info" #f #f "auth" + (jdata "username" "admin" "success" #t "auth_type" "ssh")) + (store-event db 2 "h1" "s1" (+ 1000 120000) "privilege_escalation" "critical" #f #f "priv" + (jdata "old_uid" 1000 "new_uid" 0 "method" "sudo")) + (let ((r (run-threat-detections db "priv_escalation_chain"))) + (check "fires once" (length r) 1) + (check " severity" (hash-get (car r) "severity") "critical")) + (store-close db)) + +;; outside the 5-min window -> 0 (also confirms #t serializes to JSON true) +(let ((db (fresh))) + (store-event db 1 "h1" "s1" 1000 "auth_event" "info" #f #f "auth" + (jdata "username" "admin" "success" #t)) + (store-event db 2 "h1" "s1" (+ 1000 600000) "privilege_escalation" "critical" #f #f "priv" + (jdata "new_uid" 0)) + (check "priv_escalation outside window" (length (run-threat-detections db "priv_escalation_chain")) 0) + (store-close db)) + +;; ── persistence_after_access: reverse_shell then persistence_event /1h -> 1 ─── +(displayln "persistence_after_access:") +(let ((db (fresh))) + (store-event db 1 "h1" "s1" 1000 "reverse_shell" "critical" #f #f "revshell" + (jdata "process_name" "bash" "remote_addr" "10.0.0.99")) + (store-event db 2 "h1" "s1" (+ 1000 60000) "persistence_event" "critical" #f #f "persist" + (jdata "mechanism" "systemd" "path" "/etc/systemd/system/backdoor.service")) + (check "fires once" (length (run-threat-detections db "persistence_after_access")) 1) + (store-close db)) + +;; ── log_cover: critical event then log_tampering /1h -> 1 ───────────────────── +(displayln "log_cover:") +(let ((db (fresh))) + (store-event db 1 "h1" "s1" 1000 "reverse_shell" "critical" #f #f "revshell" + (jdata "process_name" "bash" "remote_addr" "10.0.0.1")) + (store-event db 2 "h1" "s1" (+ 1000 300000) "log_tampering" "critical" #f #f "tamper" + (jdata "tamper_type" "truncation" "log_path" "/var/log/auth.log")) + (let ((r (run-threat-detections db "log_cover"))) + (check "fires once" (length r) 1) + (check " trigger_event" (hash-get (hash-get (car r) "details") "trigger_event") "reverse_shell")) + (store-close db)) + +;; ── lateral_after_shell: reverse_shell then lateral_movement /1h -> 1 ───────── +(displayln "lateral_after_shell:") +(let ((db (fresh))) + (store-event db 1 "h1" "s1" 1000 "reverse_shell" "critical" #f #f "revshell" + (jdata "process_name" "bash" "remote_addr" "10.0.0.99")) + (store-event db 2 "h1" "s1" (+ 1000 120000) "lateral_movement" "high" #f #f "lateral" + (jdata "movement_type" "ssh" "target_host" "10.0.0.50" "target_port" 22)) + (check "fires once" (length (run-threat-detections db "lateral_after_shell")) 1) + (store-close db)) + ;; ── run-all dispatch: every rule together, sorted by time ───────────────────── (displayln "run-all + unknown-rule guard:") (let ((db (fresh))) --- a/jsecmon/threats.ss +++ b/jsecmon/threats.ss @@ -25,7 +25,9 @@ (library (jsecmon threats) (export run-threat-detections threat-rules detect-brute-force detect-credential-stuffing detect-dns-tunnel - detect-suspicious-cron detect-recon-port-scan detect-data-exfil) + detect-suspicious-cron detect-recon-port-scan detect-data-exfil + detect-priv-escalation-chain detect-persistence-after-access + detect-log-cover detect-lateral-after-shell) (import (except (chezscheme) make-hash-table hash-table? sort sort! @@ -194,14 +196,90 @@ "connection_count" count "window_start_ms" (a-num start "timestamp_ms")))))))) + ;; ── sequence-pair rules (event A then event B within a window, same host) ───── + (def hour-ms 3600000) + + ;; For each A (in order), the first B on the same host with a.ts < b.ts <= + ;; a.ts+window yields one anomaly (built by `emit`), then we move to the next A + ;; — exactly secmon's nested-loop-with-break. Both lists are ordered by host,ts. + (def (pair-sequence as bs window-ms emit) + (filter-map + (lambda (a) + (let ((ah (a-str a "host")) (at (a-num a "timestamp_ms"))) + (let loop ((bs bs)) + (and (pair? bs) + (let ((b (car bs))) + (if (and (string=? (a-str b "host") ah) + (> (a-num b "timestamp_ms") at) + (<= (a-num b "timestamp_ms") (+ at window-ms))) + (emit a b) + (loop (cdr bs)))))))) + as)) + + (def (gap-seconds a b) + (quotient (- (a-num b "timestamp_ms") (a-num a "timestamp_ms")) 1000)) + + ;; Generic: event_type A (optional json equality on data) followed by event_type + ;; B within window_ms on the same host. json-filter is (path . literal), e.g. + ;; ("$.success" . "1") -> AND json_extract(data,'$.success')=1. + (def (sequence-pair db type-a json-filter type-b window-ms rule-name severity filter) + (let ((as (filtered-query db + (str "SELECT host, timestamp_ms FROM events WHERE event_type='" type-a "'" + (if json-filter + (str " AND json_extract(data,'" (car json-filter) "')=" (cdr json-filter)) + "")) + " ORDER BY host, timestamp_ms" filter)) + (bs (filtered-query db + (str "SELECT host, timestamp_ms FROM events WHERE event_type='" type-b "'") + " ORDER BY host, timestamp_ms" filter))) + (pair-sequence as bs window-ms + (lambda (a b) + (make-anomaly rule-name (a-str a "host") severity (a-num a "timestamp_ms") + (details-hash "event_a" type-a "event_b" type-b + "gap_seconds" (gap-seconds a b))))))) + + (def (detect-priv-escalation-chain db filter) + (sequence-pair db "auth_event" (cons "$.success" "1") "privilege_escalation" + 300000 "priv_escalation_chain" "critical" filter)) + + (def (detect-persistence-after-access db filter) + (append + (sequence-pair db "reverse_shell" #f "persistence_event" hour-ms + "persistence_after_access" "critical" filter) + (sequence-pair db "webshell" #f "persistence_event" hour-ms + "persistence_after_access" "critical" filter))) + + (def (detect-lateral-after-shell db filter) + (sequence-pair db "reverse_shell" #f "lateral_movement" hour-ms + "lateral_after_shell" "critical" filter)) + + ;; Any critical event followed by log_tampering within 1h on the same host. + ;; The A-set is by severity (not event_type), so it needs its own queries. + (def (detect-log-cover db filter) + (let ((as (filtered-query db + "SELECT host, timestamp_ms, event_type FROM events WHERE severity='critical'" + " ORDER BY host, timestamp_ms" filter)) + (bs (filtered-query db + "SELECT host, timestamp_ms FROM events WHERE event_type='log_tampering'" + " ORDER BY host, timestamp_ms" filter))) + (pair-sequence as bs hour-ms + (lambda (a b) + (make-anomaly "log_cover" (a-str a "host") "critical" (a-num a "timestamp_ms") + (details-hash "trigger_event" (a-str a "event_type") + "gap_seconds" (gap-seconds a b))))))) + ;; ── dispatch ───────────────────────────────────────────────────────────────── (def threat-rules - (list (cons "brute_force" detect-brute-force) - (cons "credential_stuffing" detect-credential-stuffing) - (cons "dns_tunnel" detect-dns-tunnel) - (cons "suspicious_cron" detect-suspicious-cron) - (cons "recon_port_scan" detect-recon-port-scan) - (cons "data_exfil" detect-data-exfil))) + (list (cons "brute_force" detect-brute-force) + (cons "credential_stuffing" detect-credential-stuffing) + (cons "dns_tunnel" detect-dns-tunnel) + (cons "suspicious_cron" detect-suspicious-cron) + (cons "recon_port_scan" detect-recon-port-scan) + (cons "data_exfil" detect-data-exfil) + (cons "priv_escalation_chain" detect-priv-escalation-chain) + (cons "persistence_after_access" detect-persistence-after-access) + (cons "log_cover" detect-log-cover) + (cons "lateral_after_shell" detect-lateral-after-shell))) ;; Run every threat rule (or one by name) over the store, sorted by time — ;; mirrors secmon run_detections(rule_name, filter). filter defaults to all.