transport: persist replay watermark across receiver restart
ober
4053b582f1e06d73e2f082f99e968e3dd519a315
--- a/SECURITY.md +++ b/SECURITY.md @@ -44,13 +44,18 @@ must be cut from a clean checkout after: and 100,000 decoded objects. Each authenticated sender uses a monotonic sequence; receivers retain the highest sequence for every cluster/sender/recipient/ session context, so an old frame cannot become valid after cache pressure. - Session IDs must be unique for each sender lifetime and rotated on sender + session IDs must be unique for each sender lifetime and rotated on sender restart or rekey. Inbound admission is capped, and one absolute pre-auth deadline covers the TLS handshake plus the first authenticated frame, so partial headers and silent TLS ClientHello connections cannot retain slots. Established peer connections are serialized, lifetime-bounded, and registered for immediate interruption during node shutdown; reconnect backoff observes the node stop signal. + The per-context replay watermark is durably journaled under the node's + data-path and reloaded before the accept loop admits frames, so a receiver + restart cannot reopen the replay window for a still-valid sender session; + in-memory (`:memory:`) deployments rely on the session-id rotation discipline + instead. Treat this as a private cluster protocol and rotate the key and session ID together during rekeying. - DuckDB, LevelDB, epoll, TLS, and other native behavior comes from the selected --- a/lib/jerboa-db/transport.ss +++ b/lib/jerboa-db/transport.ss @@ -51,6 +51,10 @@ transport-auth-frame-self-test transport-auth-identity-self-test transport-auth-post-window-replay-self-test + transport-auth-restart-replay-self-test + transport-replay-journal-open! + transport-replay-journal-close! + transport-replay-reset-for-test! transport-active-inbound-count transport-active-accept-loop-count ;; Convenience: transport + DB connection in one call @@ -79,6 +83,9 @@ (std crypto compare) (std crypto random) (std misc channel) + (only (std security taint) check-untainted! safe-delete-file) + (only (jerboa-db storage-codec) + db-storage-encode-bytevector db-storage-decode-bytevector) (std raft) (jerboa-db replication) (jerboa-db core) @@ -300,10 +307,31 @@ ;; the highest sequence per membership/session context, so an old capture ;; never becomes valid through cache eviction. Rotate session-id whenever a ;; sender restarts or the authentication key changes. + ;; + ;; The high-water mark table is also durably journaled when a transport node + ;; arms a replay journal. A receiver restart otherwise resets the in-memory + ;; table to zero and reopens the replay window for any captured frame whose + ;; sender session is still live; loading the persisted marks before the + ;; accept loop admits frames — and writing them before a fresh sequence is + ;; admitted — keeps the watermark durable across restarts. Persistence is + ;; best-effort: a disk fault cannot wedge the inbound path because the + ;; in-memory mark still advances. (define *transport-send-sequence* 0) (define *transport-send-sequence-mutex* (make-mutex)) (define *transport-highest-sequences* (make-hash-table)) (define *transport-replay-mutex* (make-mutex)) + (define *transport-replay-journal-path* #f) + (define +transport-replay-max-entries+ 1000000) + (define +transport-replay-max-record-bytes+ (* 1024 1024)) + (define +transport-replay-max-record-objects+ 100000) + (define +transport-replay-magic+ + (let ([bv (make-bytevector 5)]) + (bytevector-u8-set! bv 0 #x4a) ;; J + (bytevector-u8-set! bv 1 #x44) ;; D + (bytevector-u8-set! bv 2 #x42) ;; B + (bytevector-u8-set! bv 3 #x52) ;; R + (bytevector-u8-set! bv 4 #x31) ;; journal format version 1 + bv)) (define (next-transport-sequence!) (with-mutex *transport-send-sequence-mutex* @@ -325,6 +353,127 @@ (if (= i 8) n (loop (+ i 1) (+ (ash n 8) (bytevector-u8-ref nonce i)))))) + (define (transport-namespace-id? v) + (or (string? v) (symbol? v) + (and (integer? v) (exact? v)))) + + (define (valid-replay-key? key) + (and (list? key) (= (length key) 4) + (transport-namespace-id? (car key)) + (transport-namespace-id? (cadr key)) + (transport-namespace-id? (caddr key)) + (transport-namespace-id? (cadddr key)))) + + (define (valid-replay-entry? entry) + (and (pair? entry) + (valid-replay-key? (car entry)) + (let ([sequence (cdr entry)]) + (and (integer? sequence) (exact? sequence) (>= sequence 0))))) + + (define (replay-journal-read path) + ;; Best-effort load of the ((cluster sender recipient session) . sequence) + ;; alist; a truncated, hostile, or legacy file yields only its valid prefix + ;; (or '() when nothing parses). The closed storage codec rejects + ;; procedures and unsupported runtime records before they reach the + ;; in-memory watermark table. The body read is bounded so a hostile + ;; oversized journal cannot OOM the receiver. + (guard (exn [#t '()]) + (call-with-port + (open-file-input-port path (file-options) (buffer-mode block)) + (lambda (port) + (let ([magic (get-bytevector-n + port (bytevector-length +transport-replay-magic+))]) + (if (and (bytevector? magic) + (= (bytevector-length magic) + (bytevector-length +transport-replay-magic+)) + (timing-safe-equal? magic +transport-replay-magic+)) + (let ([entries + (guard (exn [#t '()]) + (db-storage-decode-bytevector + (get-bytevector-n + port +transport-replay-max-record-bytes+) + +transport-replay-max-record-bytes+ + +transport-replay-max-record-objects+))]) + (if (and (list? entries) + (<= (length entries) +transport-replay-max-entries+) + (for-all valid-replay-entry? entries)) + entries + '())) + '())))))) + + (define (replay-journal-write! path entries) + ;; Atomic durable rewrite: write a temp file, then rename over the journal. + ;; A crash leaves either the old or new complete file, never a partial + ;; header. + (let ([tmp (string-append path ".tmp")] + [payload (db-storage-encode-bytevector + entries + +transport-replay-max-record-bytes+ + +transport-replay-max-record-objects+)] + [renamed? #f]) + (check-untainted! tmp 'transport-replay-journal-write!) + (dynamic-wind + void + (lambda () + (let ([port (open-file-output-port + tmp (file-options no-fail) (buffer-mode block))]) + (dynamic-wind + void + (lambda () + (put-bytevector port +transport-replay-magic+) + (put-bytevector port payload) + (flush-output-port port)) + (lambda () (close-port port)))) + (rename-file tmp path) + (set! renamed? #t)) + (lambda () + (unless renamed? + (when (file-exists? tmp) + (guard (exn [#t (void)]) (safe-delete-file tmp)))))))) + + ;; Caller holds *transport-replay-mutex*. Persistence is best-effort so a disk + ;; fault cannot wedge the inbound path; the in-memory mark still advances. + (define (replay-journal-persist!) + (let ([path *transport-replay-journal-path*]) + (when path + (guard (exn [#t + (transport-trace + "transport trace: replay journal persist failed")]) + (replay-journal-write! path + (hash->list *transport-highest-sequences*)))))) + + (define (transport-replay-journal-open! path) + ;; Reload persisted high-water marks (max per context) and arm durability. + ;; The in-memory table is preserved so a single process running multiple + ;; transport nodes keeps each recipient's marks. A new process starts with + ;; an empty table and reloads from the journal; transport-replay-reset-for-test! + ;; simulates that for self-tests. + (check-untainted! path 'transport-replay-journal-open!) + (let ([parent (path-directory path)]) + (unless (or (string=? parent "") (string=? parent ".")) + (unless (file-exists? parent) (mkdir parent)))) + (with-mutex *transport-replay-mutex* + (set! *transport-replay-journal-path* #f) + (when (file-exists? path) + (for-each + (lambda (entry) + (let ([key (car entry)] [sequence (cdr entry)]) + (when (> sequence (hash-ref *transport-highest-sequences* key 0)) + (hash-put! *transport-highest-sequences* key sequence)))) + (replay-journal-read path))) + (set! *transport-replay-journal-path* path))) + + (define (transport-replay-journal-close!) + (with-mutex *transport-replay-mutex* + (set! *transport-replay-journal-path* #f))) + + (define (transport-replay-reset-for-test!) + ;; Test-only: drop the in-memory watermark table to simulate a fresh + ;; receiver process opening its journal. Production callers never need + ;; this — a new process starts with an empty table. + (with-mutex *transport-replay-mutex* + (hash-clear! *transport-highest-sequences*))) + (define (accept-fresh-sequence! cluster-id sender-id recipient-id session-id nonce) (let ([key (list cluster-id sender-id recipient-id session-id)] [sequence (nonce-sequence nonce)]) @@ -333,6 +482,7 @@ (and (> sequence highest) (begin (hash-put! *transport-highest-sequences* key sequence) + (replay-journal-persist!) #t)))))) (define (normalize-transport-auth-key who key) @@ -497,17 +647,92 @@ session allowed?)]) (and first-valid (let loop ([i 0]) - (if (= i 8300) - (not (authenticated-frame-payload - first-frame auth-key cluster 'replay-recipient - session allowed?)) - (let* ([frame (authenticated-frame-body - payload auth-key cluster 'replay-sender - 'replay-recipient session)] - [decoded (authenticated-frame-payload - frame auth-key cluster 'replay-recipient - session allowed?)]) - (and decoded (loop (+ i 1))))))))) + (if (= i 8300) + (not (authenticated-frame-payload + first-frame auth-key cluster 'replay-recipient + session allowed?)) + (let* ([frame (authenticated-frame-body + payload auth-key cluster 'replay-sender + 'replay-recipient session)] + [decoded (authenticated-frame-payload + frame auth-key cluster 'replay-recipient + session allowed?)]) + (and decoded (loop (+ i 1))))))))) + + ;; Simulate a receiver restart by reloading the persisted watermark from a + ;; durable journal. A captured frame that was accepted before the restart + ;; must be rejected after it; a fresh in-order frame must still be accepted. + (define (transport-auth-restart-replay-self-test key) + (let* ([auth-key (normalize-transport-auth-key + 'transport-auth-restart-replay-self-test key)] + [payload (transport-safe-encode '(restart-replay-check))] + [cluster "restart-replay-cluster"] + [session "restart-replay-session"] + [sender 'restart-sender] + [recipient 'restart-recipient] + [allowed? (lambda (id) (eq? id sender))] + [dir (string-append "/tmp/jdb-replay-journal-test-" + (number->string (time-second (current-time))) + "-" (number->string (random 1000000)))] + [journal (string-append dir "/replay-watermark.jdb")] + [cleanup! + (lambda () + (transport-replay-journal-close!) + (transport-replay-reset-for-test!) + (guard (exn [#t (void)]) + (when (file-exists? journal) (safe-delete-file journal))) + (guard (exn [#t (void)]) + (when (file-exists? dir) (delete-directory dir))))]) + (guard (exn [#t (cleanup!) (raise exn)]) + (dynamic-wind + void + (lambda () + (unless (file-exists? dir) (mkdir dir)) + ;; Fresh receiver boot: open the (empty) journal and admit a frame. + ;; The high-water mark is persisted before the frame is admitted. + (transport-replay-journal-open! journal) + (let ([captured (authenticated-frame-body + payload auth-key cluster sender + recipient session)]) + (let ([first-decoded + (authenticated-frame-payload + captured auth-key cluster recipient session allowed?)]) + (unless first-decoded + (error 'transport-auth-restart-replay-self-test + "fresh frame was not accepted before restart")) + ;; Simulate a receiver restart: drop the in-memory watermark + ;; and reopen the journal. The persisted mark survives. + (transport-replay-journal-close!) + (transport-replay-reset-for-test!) + (transport-replay-journal-open! journal) + (let ([replayed (authenticated-frame-payload + captured auth-key cluster recipient + session allowed?)]) + (when replayed + (error 'transport-auth-restart-replay-self-test + "captured frame was accepted after restart")) + (let ([fresh-frame (authenticated-frame-body + payload auth-key cluster sender + recipient session)]) + (let ([fresh-decoded (authenticated-frame-payload + fresh-frame auth-key cluster + recipient session allowed?)]) + (unless fresh-decoded + (error 'transport-auth-restart-replay-self-test + "fresh in-order frame was rejected after restart"))) + ;; The fresh mark must also be persisted, so a second + ;; restart still rejects the now-stale captured frame. + (transport-replay-journal-close!) + (transport-replay-reset-for-test!) + (transport-replay-journal-open! journal) + (let ([stale-replay (authenticated-frame-payload + captured auth-key cluster recipient + session allowed?)]) + (when stale-replay + (error 'transport-auth-restart-replay-self-test + "captured frame was accepted after second restart"))) + #t))))) + cleanup!)))) (define (write-frame out-port body auth-key cluster-id sender-id recipient-id session-id) @@ -1088,6 +1313,15 @@ [inbound-registry (new-transport-connection-registry)] [accept-state (new-transport-accept-state)] [peer-mutex (make-mutex)]) + ;; Reload the persisted replay watermark before any frame is admitted so a + ;; captured authenticated frame stays rejected across a receiver restart. + ;; In-memory (:memory:) deployments skip persistence and rely on the + ;; session-id rotation discipline documented in SECURITY.md. + (when (and auth-key + (string? data-path) + (not (string=? data-path ":memory:"))) + (transport-replay-journal-open! + (string-append data-path "/replay-watermark.jdb"))) ;; Bind before starting connector threads. A bind failure therefore leaves ;; no unreachable proxy workers behind. (let-values ([(server actual-port) --- a/tests/test-transport.ss +++ b/tests/test-transport.ss @@ -247,6 +247,9 @@ (test "authenticated transport rejects replay after the former 8192-frame window" (assert-true (transport-auth-post-window-replay-self-test auth-key))) +(test "authenticated transport rejects replay across a receiver restart" + (assert-true (transport-auth-restart-replay-self-test auth-key))) + (test "authenticated startup requires explicit cluster and session identity" (let ([rejected? #f]) (guard (exn [#t (set! rejected? #t)])