From 1ed97fb07515f29bd46551db5842f272e0e89543 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Fri, 28 Aug 2026 13:53:02 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20audit=20batches=203=E2=80=934=20?= =?UTF-8?q?=E2=80=94=20TLS=20verify,=20WS,=20OFFSET,=20B-tree,=20NULL=20eq?= =?UTF-8?q?uality?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close the remaining 2026-08 findings: peer TLS on leader forward, disttxn SO_ERROR, compaction catalog order, OFFSET without LIMIT, window aggregates, WebSocket mask/size/auth, SCRAM timing and cbind, B-tree leaf left-max separators, and SQL three-valued NULL comparisons. --- BUG_AUDIT_2026-08.md | 53 +++--- CHANGELOG.md | 12 +- PLAN.md | 8 +- src/barabadb/core/config.nim | 27 +++- src/barabadb/core/disttxn.nim | 28 +++- src/barabadb/core/httpserver.nim | 9 ++ src/barabadb/core/server.nim | 37 +++-- src/barabadb/core/websocket.nim | 129 ++++++++++----- src/barabadb/protocol/auth.nim | 15 +- src/barabadb/protocol/scram.nim | 14 ++ src/barabadb/query/exec/eval.nim | 41 ++++- src/barabadb/query/exec/lower.nim | 17 +- src/barabadb/query/exec/plan_exec.nim | 6 + src/barabadb/query/exec/rls.nim | 14 +- src/barabadb/query/exec/window.nim | 58 +++++++ src/barabadb/query/parser.nim | 5 + src/barabadb/storage/btree.nim | 82 ++++++++-- src/barabadb/storage/compaction.nim | 11 +- src/baradadb.nim | 47 ++++-- tests/bugfix_test.nim | 225 ++++++++++++++++++++++++++ tests/test_all.nim | 15 ++ 21 files changed, 718 insertions(+), 135 deletions(-) diff --git a/BUG_AUDIT_2026-08.md b/BUG_AUDIT_2026-08.md index 3f321cb..d705bca 100644 --- a/BUG_AUDIT_2026-08.md +++ b/BUG_AUDIT_2026-08.md @@ -3,7 +3,7 @@ > Дата: 2026-08-02 > Метод: 4 паралелни одит-агента по слоеве (Storage / Query / Core / Protocol), всеки чете всички файлове в обхвата си и проверява находките срещу реалния код. > Обхват: **само нови дефекти** — 80-те вече оправени в `BUGS.md` / `BUG_AUDIT.md` / `BARADB_CLIENT_BUGS.md` са изключени. -> **Общо: ~28 находки | Поправени: 17 (батч 1: 5 + батч 2: 12, вкл. hygiene) | Остават: 12** +> **Общо: ~28 находки | Поправени: 28 (батч 1: 5 + батч 2: 12 + батч 3: 10 + батч 4: 2) | Остават: 0** --- @@ -36,36 +36,33 @@ **Верификация (батч 2):** `baradadb` build чист; `tests/bugfix_test.nim` (вкл. batch-2 suite) и `tests/test_all.nim` (501 OK) минават без `[FAILED]`. `tests/prop_test.nim` B-Tree suite OK (H10 *не* е в този батч — naive left-max fix чупи interleaved remove). +## Поправени — батч 3 (10) + +| # | Severity | Проблем | Файл | Fix | +|---|----------|---------|------|-----| +| H2 | 🟠 HIGH | **TLS client връзките не верифицираха сертификата** | `core/server.nim`, `core/config.nim` | Отделен `tlsClient` контекст; CA auto-enable verify; production fail-closed | +| M2 | 🟡 MEDIUM | **disttxn refused-connect wedge** | `core/disttxn.nim` | `getsockopt(SO_ERROR)` + try/except около RPC | +| M7 | 🟡 MEDIUM | **Compaction unlink преди catalog load** | `storage/compaction.nim`, `baradadb.nim` | Unlink след load + MANIFEST | +| M8 | 🟡 MEDIUM | **`OFFSET` без `LIMIT` → 0 реда** | `query/exec/lower.nim`, `plan_exec.nim` | `limitCount = -1` unlimited; clamp negative | +| M9 | 🟡 MEDIUM | **Window SUM/AVG/COUNT/MIN/MAX → NULL** | `query/exec/window.nim` | Frame aggregates | +| M10 | 🟡 MEDIUM | **WebSocket unmasked client frames** | `core/websocket.nim` | Protocol error / close | +| M11 | 🟡 MEDIUM | **WebSocket unbounded buffer** | `core/websocket.nim` | 1 MiB frame / 4 MiB message / 125-byte control | +| M12 | 🟡 MEDIUM | **SUBSCRIBE без table auth** | `core/websocket.nim`, `httpserver.nim` | `canSubscribe` + `hasPrivilegeFor` SELECT | +| L1 | 🟢 LOW | **SCRAM timing user enumeration** | `protocol/auth.nim` | Dummy nonce+encode work за unknown users | +| L2 | 🟢 LOW | **SCRAM `c=` не се верифицира** | `protocol/auth.nim`, `scram.nim` | `c=` must match gs2 header (`biws` за `n,,`) | + +## Поправени — батч 4 (2) + +| # | Severity | Проблем | Файл | Fix | +|---|----------|---------|------|-----| +| H10 | 🟠 HIGH | **B-tree `remove` separator convention** — `splitChild`/search са left-max (`key > sep → right`); `removeRec` копираше first key на дясното дете (right-min). Naive left-max върху *internal* ключове чупи `prop_test`. | `storage/btree.nim` | Leaf borrow/remove пишат left-max; merge underflow се качва нагоре; invariant `max(left) <= sep` (boundary duplicates са позволени, `next` ги събира) | +| L4 | 🟢 LOW | **NULL equality** — `NULL = NULL` / `col = NULL` бяха true през string sentinel | `query/exec/eval.nim` | Сравнения, LIKE, IN, NOT, AND/OR: NULL operand → unknown (`\N`); `IS NULL` непроменен | + --- -## Остават (12) +## Остават (0) -### 🟠 HIGH (2) - -| # | Проблем | Файл | Предложен fix | -|---|---------|------|---------------| -| H2 | **TLS client връзките между възли не верифицират сертификата** — `forwardQueryToLeader` ползва `verifyMode = CVerifyNone` → MITM на клъстър линка. Raft client dials са със същия default (`raftTlsVerifyPeer: false`). | `core/server.nim:70` | Verify peer cert срещу CA при client handshake (fail-closed при enabled TLS) | -| H10 | **B-tree `remove` separator convention** — audit: `splitChild` left-max vs `removeRec` right-min. Naive left-max rewrite of separators/borrows **fails** `prop_test` interleaved insert/remove; needs careful multi-level fix + more targeted repro first. | `storage/btree.nim:377` | Repro + full-tree separator invariant; keep borrow/merge/search consistent | - -### 🟡 MEDIUM (7) - -| # | Проблем | Файл | Предложен fix | -|---|---------|------|---------------| -| M2 | **disttxn `connectWithTimeout` без SO_ERROR + uncaught RPC** — refused connect е "writable" → връща true; `sendDistTxnRpc` няма try/except → OSError wedge-ва 2PC състояние. (BUG-042 fix-нат в replication, не тук.) | `core/disttxn.nim:88` | `getsockopt(SO_ERROR)` + try/except около per-participant RPC | -| M7 | **Compaction unlink-ва input-ите преди output-ът да е loadable в каталога** — verifySSTable вече е преди unlink; остава catalog re-load ordering в caller. | `storage/compaction.nim` / LSM apply | Load/verify output в каталога ПРЕДИ unlink на input-ите | -| M8 | **`OFFSET n` без `LIMIT` връща 0 реда; negative `LIMIT` чупи** — `limitCount = 0` е sentinel и за "няма limit", и за "LIMIT 0"; `sourceRows[start.. sep → right`); merge underflow rebalances up the tree; `checkInvariants` guards `max(left) <= sep` (`storage/btree.nim`) +- **NULL comparison three-valued logic (LOW)** — `NULL = NULL` / `col = NULL` / `col != x` yield unknown (`\N`) so `WHERE` excludes them; `IS NULL` is unchanged (`query/exec/eval.nim`) ### Removed @@ -35,7 +45,7 @@ All notable changes to BaraDB are documented in this file. ### Added -- Deep audit report `BUG_AUDIT_2026-08.md` (~28 findings; 17 fixed across batches 1–2, ~12 tracked) +- Deep audit report `BUG_AUDIT_2026-08.md` (~28 findings; all fixed across batches 1–4) --- diff --git a/PLAN.md b/PLAN.md index a0a7ea6..225cd2f 100644 --- a/PLAN.md +++ b/PLAN.md @@ -165,7 +165,11 @@ **Батч 2 — поправени (12):** semi-sync partial-ack (H3), COUNT/SUM/AVG(DISTINCT) (H6), UNION/INTERSECT/EXCEPT (H7), MERGE THEN DELETE (H8), WAL torn-record recovery (H9), MVCC write iteration (M1), checkpoint lock leak (M3), flushUnsafe order (M4), compaction empty-key (M5), rewriteLive atomic replace (M6), mmap overflow (L3), stray `protocol/scram` ELF. -**Остават (~12):** вж. `BUG_AUDIT_2026-08.md` — TLS peer verify (H2), B-tree separator (H10, needs careful repro), disttxn SO_ERROR (M2), compaction catalog order (M7), OFFSET-без-LIMIT (M8), window агрегати (M9), WebSocket (M10–M12), SCRAM (L1–L2), NULL equality (L4). +**Батч 3 — поправени (10):** TLS peer verify (H2), disttxn SO_ERROR (M2), compaction catalog order (M7), OFFSET-без-LIMIT (M8), window агрегати (M9), WebSocket (M10–M12), SCRAM (L1–L2). + +**Батч 4 — поправени (2):** B-tree leaf left-max separators (H10), NULL three-valued comparisons (L4). + +**Остават (0):** вж. `BUG_AUDIT_2026-08.md`. --- @@ -181,7 +185,7 @@ | **Този план** — Сесии 10, 11, 12 | ✅ Завършен | | Raft C3a/C3b + DDL/forward/compact/metrics (2026-07-30) | ✅ Завършен на `main` — `docs/superpowers/specs/2026-07-30-raft-cluster-status.md` | | **Production GA v1.2.0** (single-node) | ✅ `docs/superpowers/plans/2026-07-30-production-ga.md` | -| **Сесия 13** — Stabilization & Deep Audit (2026-08) | 🔄 В процес — батч 1+2 (17 поправки); остават ~12; `BUG_AUDIT_2026-08.md` | +| **Сесия 13** — Stabilization & Deep Audit (2026-08) | ✅ Батч 1–4 (28 поправки); `BUG_AUDIT_2026-08.md` | --- diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 068cd27..bae7060 100644 --- a/src/barabadb/core/config.nim +++ b/src/barabadb/core/config.nim @@ -14,6 +14,8 @@ type tlsEnabled*: bool certFile*: string keyFile*: string + tlsCaFile*: string + tlsVerifyPeer*: bool idleTimeoutMs*: int queryTimeoutMs*: int slowQueryThresholdMs*: int @@ -67,6 +69,8 @@ proc defaultConfig*(): BaraConfig = tlsEnabled: false, certFile: "", keyFile: "", + tlsCaFile: "", + tlsVerifyPeer: false, idleTimeoutMs: 300_000, queryTimeoutMs: 30_000, slowQueryThresholdMs: 1_000, @@ -134,6 +138,8 @@ proc loadConfigFromJson*(path: string, cfg: var BaraConfig) = if s.hasKey("enabled"): cfg.tlsEnabled = s["enabled"].getBool() if s.hasKey("cert_file"): cfg.certFile = s["cert_file"].getStr() if s.hasKey("key_file"): cfg.keyFile = s["key_file"].getStr() + if s.hasKey("ca_file"): cfg.tlsCaFile = s["ca_file"].getStr() + if s.hasKey("verify_peer"): cfg.tlsVerifyPeer = s["verify_peer"].getBool() if j.hasKey("auth"): let s = j["auth"] if s.hasKey("enabled"): cfg.authEnabled = s["enabled"].getBool() @@ -177,6 +183,13 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) = cfg.tlsEnabled = parseEnvBool(getEnv("BARADB_TLS_ENABLED", ""), cfg.tlsEnabled) cfg.certFile = getEnv("BARADB_CERT_FILE", cfg.certFile) cfg.keyFile = getEnv("BARADB_KEY_FILE", cfg.keyFile) + cfg.tlsCaFile = getEnv("BARADB_TLS_CA_FILE", cfg.tlsCaFile) + let tlsVerifyEnv = getEnv("BARADB_TLS_VERIFY_PEER", "") + if tlsVerifyEnv.len > 0: + cfg.tlsVerifyPeer = parseEnvBool(tlsVerifyEnv, cfg.tlsVerifyPeer) + elif cfg.tlsCaFile.len > 0: + # CA present and verify flag unset → verify (fail-closed for MITM). + cfg.tlsVerifyPeer = true cfg.idleTimeoutMs = parseEnvInt(getEnv("BARADB_IDLE_TIMEOUT_MS", ""), cfg.idleTimeoutMs) cfg.queryTimeoutMs = parseEnvInt(getEnv("BARADB_QUERY_TIMEOUT_MS", ""), cfg.queryTimeoutMs) cfg.slowQueryThresholdMs = parseEnvInt(getEnv("BARADB_SLOW_QUERY_THRESHOLD_MS", ""), cfg.slowQueryThresholdMs) @@ -233,7 +246,11 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) = cfg.raftTlsCertFile = getEnv("BARADB_RAFT_TLS_CERT_FILE", cfg.raftTlsCertFile) cfg.raftTlsKeyFile = getEnv("BARADB_RAFT_TLS_KEY_FILE", cfg.raftTlsKeyFile) cfg.raftTlsCaFile = getEnv("BARADB_RAFT_TLS_CA_FILE", cfg.raftTlsCaFile) - cfg.raftTlsVerifyPeer = parseEnvBool(getEnv("BARADB_RAFT_TLS_VERIFY_PEER", ""), cfg.raftTlsVerifyPeer) + let raftVerifyEnv = getEnv("BARADB_RAFT_TLS_VERIFY_PEER", "") + if raftVerifyEnv.len > 0: + cfg.raftTlsVerifyPeer = parseEnvBool(raftVerifyEnv, cfg.raftTlsVerifyPeer) + elif cfg.raftTlsCaFile.len > 0: + cfg.raftTlsVerifyPeer = true # Optional: client (SQL) addresses for leader write forwarding. # Same id@host:port shape as BARADB_RAFT_PEERS, but ports are BARADB_PORT values. let clientPeersEnv = getEnv("BARADB_RAFT_CLIENT_PEERS", "") @@ -292,6 +309,14 @@ proc validateProductionConfig*(cfg: BaraConfig) = if cfg.jwtSecret in ["change-me", "change-me-to-random-32-char-string", "secret", "default"]: raise newException(ValueError, "Production refuses insecure JWT secret placeholder. Set a strong BARADB_JWT_SECRET.") + if cfg.tlsEnabled: + if not cfg.tlsVerifyPeer or cfg.tlsCaFile.len == 0: + raise newException(ValueError, + "Production TLS requires peer verification. Set BARADB_TLS_VERIFY_PEER=true and BARADB_TLS_CA_FILE.") + if cfg.raftTlsEnabled: + if not cfg.raftTlsVerifyPeer or cfg.raftTlsCaFile.len == 0: + raise newException(ValueError, + "Production raft TLS requires peer verification. Set BARADB_RAFT_TLS_VERIFY_PEER=true and BARADB_RAFT_TLS_CA_FILE.") proc getEffectiveJwtSecret*(cfg: BaraConfig): string = if cfg.jwtSecret.len > 0: diff --git a/src/barabadb/core/disttxn.nim b/src/barabadb/core/disttxn.nim index 88f8ac3..7d9e66d 100644 --- a/src/barabadb/core/disttxn.nim +++ b/src/barabadb/core/disttxn.nim @@ -5,6 +5,8 @@ import std/monotimes import std/net import std/strutils import std/nativesockets +when defined(posix): + import std/posix type DistTxnState* = enum @@ -89,6 +91,13 @@ proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int): var fds = @[sock.getFd] if selectWrite(fds, timeoutMs) <= 0: return false + when defined(posix): + # selectWrite reports a refused connect as writable; SO_ERROR tells the truth. + var err: cint = 0 + var errLen = SockLen(sizeof(err)) + discard posix.getsockopt(sock.getFd, 1'i32, 4'i32, addr err, addr errLen) + if err != 0: + return false sock.getFd.setBlocking(true) return true @@ -96,15 +105,18 @@ proc sendDistTxnRpc(host: string, port: int, txnId: uint64, action: string, time ## Send 2PC RPC to participant node via TCP text protocol. ## Protocol: "DISTTXN \n" where action = PREPARE|COMMIT|ROLLBACK ## Response: "OK\n" or "ERR \n" - var sock = newSocket() - defer: sock.close() - if not connectWithTimeout(sock, host, Port(port), timeoutMs): + try: + var sock = newSocket() + defer: sock.close() + if not connectWithTimeout(sock, host, Port(port), timeoutMs): + return false + let msg = "DISTTXN " & $txnId & " " & action & "\n" + sock.send(msg) + var response = "" + sock.readLine(response) + return response.strip() == "OK" + except CatchableError: return false - let msg = "DISTTXN " & $txnId & " " & action & "\n" - sock.send(msg) - var response = "" - sock.readLine(response) - return response.strip() == "OK" type ParticipantInfo = object diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index e717831..10d3b0d 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -19,6 +19,7 @@ import ../storage/gate import ../core/mvcc import ../protocol/wire import ../core/websocket +import ../query/exec/rls import jwt as jwtlib import ../protocol/auth import ../protocol/ratelimit @@ -55,6 +56,14 @@ proc newHttpServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): ctx.txnManager = newTxnManager() let secret = config.getEffectiveJwtSecret() let ws = newWsServer(config, secret) + block: + let wsRef {.cursor.} = ws + let ctxRef {.cursor.} = ctx + wsRef.canSubscribe = proc(username, table: string): bool {.gcsafe.} = + if username.len == 0: + return false + {.cast(gcsafe).}: + return hasPrivilegeFor(ctxRef, username, table, "SELECT") let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient) ctx.onChange = proc(ev: ChangeEvent) = let msg = $ev.kind & " " & ev.table diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index e1cb551..b2f44fd 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -1,6 +1,7 @@ ## BaraDB Server — async TCP server with wire protocol import std/asyncdispatch import std/asyncnet +import std/os import std/strutils import std/sequtils import std/tables @@ -50,6 +51,10 @@ type clusterMembership*: ClusterMembership gossipProtocol*: GossipProtocol tls*: TLSContext + ## Dedicated client-role TLS context for follower→leader forwarding. + ## Must not reuse `tls` with verifyPeer — OpenSSL contexts are role-agnostic + ## and enabling verify on the server context would break inbound handshakes. + tlsClient*: TLSContext rateLimiter*: RateLimiter activeConnections*: int activeConnectionsLock*: Lock @@ -66,9 +71,22 @@ proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Ser let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx)) ctx.txnManager = newTxnManager() var tls: TLSContext = nil + var tlsClient: TLSContext = nil if config.tlsEnabled and config.certFile.len > 0 and config.keyFile.len > 0: - let tlsConfig = newTLSConfig(config.certFile, config.keyFile) - tls = newTLSContext(tlsConfig) + if config.tlsVerifyPeer and config.tlsCaFile.len == 0: + raise newException(ValueError, + "tlsVerifyPeer is true but CA file is missing. Set BARADB_TLS_CA_FILE") + if config.tlsVerifyPeer and config.tlsCaFile.len > 0 and + not fileExists(config.tlsCaFile): + raise newException(ValueError, + "BARADB_TLS_VERIFY_PEER=true but CA file missing: " & config.tlsCaFile) + tls = newTLSContext(newTLSConfig(config.certFile, config.keyFile)) + if config.tlsVerifyPeer: + tlsClient = newTLSContext(newTLSConfig( + config.certFile, config.keyFile, + caFile = config.tlsCaFile, verifyPeer = true)) + else: + tlsClient = tls # Initialize sharding / gossip. Server fields own the refs; locals used inside # callback closures are {.cursor.} so ARC does not form uncollectable cycles @@ -85,6 +103,7 @@ proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Ser clusterMembership: nil, gossipProtocol: newGossipProtocol(localId, config.address, config.port, gossipPort = gossipPort), tls: tls, + tlsClient: tlsClient, rateLimiter: rl) result.clusterMembership = newClusterMembership(result.shardRouter, localId) initLock(result.activeConnectionsLock) @@ -227,12 +246,12 @@ proc forwardQueryToLeader*(host: string, port: int, query: string, timeoutMs: int = 5000): Future[(bool, QueryResult, string)] {.async.} = ## Proxy a write/DDL to the known leader's SQL port. Used by followers when ## BARADB_RAFT_CLIENT_PEERS maps leader id → host:clientPort. - ## `tls` is the local server's client-port TLS context: when the wire port + ## `tls` is a *client-role* context (Server.tlsClient). When the wire port ## serves TLS, the leader's does too, so the forwarding dial must complete a - ## client handshake. The context is reused as-is (verifyMode stays - ## CVerifyNone — do NOT enable verifyPeer on the reused context); OpenSSL - ## contexts are role-agnostic in Nim's stdlib, wrapConnectedSocket with - ## handshakeAsClient sets the role. + ## client handshake. Peer verification is honoured when that context was + ## built with verifyPeer (BARADB_TLS_VERIFY_PEER + BARADB_TLS_CA_FILE). + ## Do NOT pass Server.tls (the inbound/server context) with verifyPeer + ## flipped on — OpenSSL contexts are role-agnostic in Nim's stdlib. var sock: AsyncSocket = nil try: sock = newAsyncSocket() @@ -806,7 +825,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs, raftPeerClientAddrs=server.config.raftPeerClientAddrs, - forwardTls=server.tls) + forwardTls=server.tlsClient) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: @@ -830,7 +849,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs, raftPeerClientAddrs=server.config.raftPeerClientAddrs, - forwardTls=server.tls) + forwardTls=server.tlsClient) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: diff --git a/src/barabadb/core/websocket.nim b/src/barabadb/core/websocket.nim index d95947a..b9df845 100644 --- a/src/barabadb/core/websocket.nim +++ b/src/barabadb/core/websocket.nim @@ -15,18 +15,25 @@ else: import config import jwt as jwtlib +const + ## RFC 6455: reject oversized frames/messages to bound memory. + MaxWsFrameBytes* = 1 * 1024 * 1024 + MaxWsMessageBytes* = 4 * 1024 * 1024 + MaxWsControlPayload* = 125 + type - WsFrame = object - fin: bool - opcode: uint8 - masked: bool - payloadLen: uint64 - maskKey: array[4, byte] - payload: string + WsFrame* = object + fin*: bool + opcode*: uint8 + masked*: bool + payloadLen*: uint64 + maskKey*: array[4, byte] + payload*: string WsClient* = ref object socket: AsyncSocket id: int + username: string subscriptions: HashSet[string] WsServer* = ref object @@ -35,6 +42,8 @@ type running: bool config*: BaraConfig secretKey*: string + ## Table-level read authorization for SUBSCRIBE. Nil + authEnabled → deny. + canSubscribe*: proc(username, table: string): bool {.closure, gcsafe.} onInsert*: proc (table, key, value: string) {.closure.} onDelete*: proc (table, key: string) {.closure.} @@ -46,20 +55,19 @@ proc newWsServer*(cfg: BaraConfig = defaultConfig(), secret: string = ""): WsSer # WebSocket frame encoding/decoding (RFC 6455) # ---------------------------------------------------------------------- -proc encodeFrame(opcode: uint8, payload: string): string = +proc encodeFrame*(opcode: uint8, payload: string, masked = false, + maskKey: array[4, byte] = [0'u8, 0, 0, 0]): string = result = "" - let isMasked = false var b0 = 0x80'u8 or opcode result.add(char(b0)) - var b1 = 0'u8 - if not isMasked: - if payload.len < 126: - b1 = uint8(payload.len) - elif payload.len <= 65535: - b1 = 126 - else: - b1 = 127 + var b1 = if masked: 0x80'u8 else: 0'u8 + if payload.len < 126: + b1 = b1 or uint8(payload.len) + elif payload.len <= 65535: + b1 = b1 or 126 + else: + b1 = b1 or 127 result.add(char(b1)) if payload.len >= 126 and payload.len <= 65535: @@ -71,9 +79,17 @@ proc encodeFrame(opcode: uint8, payload: string): string = for i in countdown(7, 0): result.add(char((len64 shr (i * 8)) and 0xFF)) - result.add(payload) + if masked: + for i in 0..3: + result.add(char(maskKey[i])) + for i, c in payload: + result.add(char(byte(c) xor maskKey[i mod 4])) + else: + result.add(payload) -proc decodeFrame(data: string): (WsFrame, int) = +proc decodeFrame*(data: string): (WsFrame, int) = + ## Returns (frame, consumed). consumed == 0 → need more bytes; + ## consumed < 0 → protocol error (close the connection). if data.len < 2: return (WsFrame(), 0) @@ -84,6 +100,10 @@ proc decodeFrame(data: string): (WsFrame, int) = frame.opcode = b0 and 0x0F frame.masked = (b1 and 0x80) != 0 + # RFC 6455 §5.1 — client-to-server frames MUST be masked. + if not frame.masked: + return (WsFrame(), -1) + var len = uint64(b1 and 0x7F) var offset = 2 @@ -98,23 +118,24 @@ proc decodeFrame(data: string): (WsFrame, int) = len = (len shl 8) or uint64(uint8(data[2 + i])) offset = 10 - if frame.masked: - if data.len < offset + 4: return (WsFrame(), 0) - for i in 0..3: - frame.maskKey[i] = byte(data[offset + i]) - offset += 4 + let isControl = frame.opcode == 0x8 or frame.opcode == 0x9 or frame.opcode == 0xA + if isControl and (not frame.fin or len > uint64(MaxWsControlPayload)): + return (WsFrame(), -1) + if len > uint64(MaxWsFrameBytes): + return (WsFrame(), -1) + + if data.len < offset + 4: return (WsFrame(), 0) + for i in 0..3: + frame.maskKey[i] = byte(data[offset + i]) + offset += 4 if uint64(data.len) < uint64(offset) + len: - return (Wsframe(), 0) + return (WsFrame(), 0) - if len > uint64(high(int) - 1): - return (Wsframe(), 0) let plen = int(len) - if frame.masked: - for i in 0.. 128: + return false + for c in table: + if c notin {'a'..'z', 'A'..'Z', '0'..'9', '_'}: + return false + return true + proc subscribe*(client: WsClient, table: string) = client.subscriptions.incl(table) @@ -201,9 +230,11 @@ proc broadcastToTable*(server: WsServer, table: string, msg: string) {.async.} = # WebSocket client handler # ---------------------------------------------------------------------- -proc handleWsClient(server: WsServer, client: AsyncSocket, id: int) {.async.} = +proc handleWsClient(server: WsServer, client: AsyncSocket, id: int, + username: string = "") {.async.} = echo "WebSocket client ", id, " connected" - var wsClient = WsClient(socket: client, id: id, subscriptions: initHashSet[string]()) + var wsClient = WsClient(socket: client, id: id, username: username, + subscriptions: initHashSet[string]()) server.clients[id] = wsClient var buf = "" @@ -212,12 +243,22 @@ proc handleWsClient(server: WsServer, client: AsyncSocket, id: int) {.async.} = let chunk = await client.recv(4096) if chunk.len == 0: break + if buf.len + chunk.len > MaxWsMessageBytes: + let closeF = encodeFrame(0x8, "") + try: await client.send(closeF) except CatchableError: discard + break buf.add(chunk) while buf.len >= 2: let (frame, consumed) = decodeFrame(buf) if consumed == 0: break + if consumed < 0: + let closeF = encodeFrame(0x8, "") + try: await client.send(closeF) except CatchableError: discard + client.close() + server.clients.del(id) + return case frame.opcode of 0x8: # close @@ -231,9 +272,18 @@ proc handleWsClient(server: WsServer, client: AsyncSocket, id: int) {.async.} = let msg = frame.payload if msg.startsWith("SUBSCRIBE "): let table = msg[10..^1].strip() - wsClient.subscribe(table) - let ack = encodeFrame(0x1, "OK subscribed to " & table) - await client.send(ack) + var allowed = validSubscribeTable(table) + if allowed and server.config.authEnabled: + if username.len == 0 or server.canSubscribe == nil or + not server.canSubscribe(username, table): + allowed = false + if allowed: + wsClient.subscribe(table) + let ack = encodeFrame(0x1, "OK subscribed to " & table) + await client.send(ack) + else: + let nack = encodeFrame(0x1, "ERR subscribe denied for " & table) + await client.send(nack) elif msg.startsWith("UNSUBSCRIBE "): let table = msg[12..^1].strip() wsClient.unsubscribe(table) @@ -285,6 +335,7 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} = return # Auth check + var username = "" if server.config.authEnabled: let authHeader = headers.getOrDefault("authorization", "") if authHeader.len == 0 or not authHeader.startsWith("Bearer "): @@ -305,6 +356,8 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} = await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n") client.close() return + if "sub" in token.claims: + username = token.claims["sub"].node.str except CatchableError: await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n") client.close() @@ -321,7 +374,7 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} = await client.send(response) inc server.nextId - asyncCheck server.handleWsClient(client, server.nextId) + asyncCheck server.handleWsClient(client, server.nextId, username) proc setTcpNoDelay(sock: AsyncSocket) = ## Enable TCP_NODELAY using the correct protocol level (IPPROTO_TCP). diff --git a/src/barabadb/protocol/auth.nim b/src/barabadb/protocol/auth.nim index 6187593..714291e 100644 --- a/src/barabadb/protocol/auth.nim +++ b/src/barabadb/protocol/auth.nim @@ -221,10 +221,17 @@ proc registerScramUser*(am: AuthManager, username, password: string, let cred = createScramCredential(password, iterationCount = iterationCount) am.scramUsers[username] = cred +proc dummyScramStartWork() = + ## Match known-user startScram work (urandom nonce + base64) so unknown + ## users cannot be enumerated by timing. + discard generateNonce() + discard encode("0123456789abcdef0123456789abcdef") + proc startScram*(am: AuthManager, clientFirstMessage: string): string = ## Start SCRAM authentication. Returns server-first-message. - let (_, username, clientNonce) = parseClientFirst(clientFirstMessage) - if username notin am.scramUsers: + let (gs2, username, clientNonce) = parseClientFirst(clientFirstMessage) + if username notin am.scramUsers or gs2 notin ["n", "y"]: + dummyScramStartWork() raise newException(ValueError, "Authentication failed") let cred = am.scramUsers[username] @@ -239,6 +246,7 @@ proc startScram*(am: AuthManager, clientFirstMessage: string): string = var state = ScramServerState( username: username, + gs2Flag: gs2, clientFirstMessageBare: clientFirstMessageBare, serverFirstMessage: serverFirst, authMessage: authMessage, @@ -264,6 +272,9 @@ proc finishScram*(am: AuthManager, clientFinalMessage: string): (bool, string) = var state = am.scramSessions[nonce] am.scramSessions.del(nonce) + if stripB64Padding(cbind) != expectedChannelBinding(state.gs2Flag): + return (false, "e=channel-bindings-dont-match") + # Update authMessage with client-final-message-without-proof let clientFinalWithoutProof = "c=" & cbind & ",r=" & nonce state.authMessage = state.authMessage & "," & clientFinalWithoutProof diff --git a/src/barabadb/protocol/scram.nim b/src/barabadb/protocol/scram.nim index f1f59b4..d510ee0 100644 --- a/src/barabadb/protocol/scram.nim +++ b/src/barabadb/protocol/scram.nim @@ -19,6 +19,7 @@ type ScramServerState* = object username*: string + gs2Flag*: string clientFirstMessageBare*: string serverFirstMessage*: string authMessage*: string @@ -189,6 +190,19 @@ proc createScramCredential*(password: string, salt: string = "", # SCRAM message parsing / building # --------------------------------------------------------------------------- +proc expectedChannelBinding*(gs2Flag: string): string = + ## RFC 5802 cbind-input with no cbind-data is the gs2-header: flag + "," + authzid + ",". + ## Authzid is unsupported, so the header is `n,,` or `y,,`. + let header = gs2Flag & ",," + result = encode(header) + while result.endsWith("="): + result.setLen(result.len - 1) + +proc stripB64Padding*(s: string): string = + result = s + while result.endsWith("="): + result.setLen(result.len - 1) + proc parseClientFirst*(msg: string): (string, string, string) = ## Parse client-first-message: gs2-header,username,nonce ## Returns: (gs2_header, username, nonce) diff --git a/src/barabadb/query/exec/eval.nim b/src/barabadb/query/exec/eval.nim index 82134bc..87279d7 100644 --- a/src/barabadb/query/exec/eval.nim +++ b/src/barabadb/query/exec/eval.nim @@ -429,6 +429,8 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex let right = evalExprOld(expr.binRight, row, ctx) case expr.binOp of irEq: + # SQL three-valued logic: any NULL operand → unknown, not true. + if isNull(left) or isNull(right): return "\\N" if left == right: return "true" # Try numeric comparison try: @@ -436,6 +438,7 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex except CatchableError: discard return "false" of irNeq: + if isNull(left) or isNull(right): return "\\N" # Numeric-first so `!=` is the exact complement of `=` (irEq): string # inequality alone would make `5 != 5.0` true while `5 = 5.0` is true. try: @@ -443,26 +446,37 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex except CatchableError: return if left != right: "true" else: "false" of irLt: + if isNull(left) or isNull(right): return "\\N" try: return if parseFloat(left) < parseFloat(right): "true" else: "false" except CatchableError: return if left < right: "true" else: "false" of irLte: + if isNull(left) or isNull(right): return "\\N" try: return if parseFloat(left) <= parseFloat(right): "true" else: "false" except CatchableError: return if left <= right: "true" else: "false" of irGt: + if isNull(left) or isNull(right): return "\\N" try: return if parseFloat(left) > parseFloat(right): "true" else: "false" except CatchableError: return if left > right: "true" else: "false" of irGte: + if isNull(left) or isNull(right): return "\\N" try: return if parseFloat(left) >= parseFloat(right): "true" else: "false" except CatchableError: return if left >= right: "true" else: "false" of irAnd: - if left == "true" and right == "true": return "true" - return "false" + # false AND x = false; unknown AND true/unknown = unknown; else both true. + let lNull = isNull(left) + let rNull = isNull(right) + let lTrue = left == "true" + let rTrue = right == "true" + if (not lNull and not lTrue) or (not rNull and not rTrue): return "false" + if lNull or rNull: return "\\N" + return "true" of irOr: if left == "true" or right == "true": return "true" + if isNull(left) or isNull(right): return "\\N" return "false" of irAdd, irSub, irMul, irDiv, irMod, irPow: let v = evalExpr(expr, stringTableToValueRow(row), ctx) @@ -475,6 +489,7 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex of vkString: return v.strVal else: return "\\N" of irLike: + if isNull(left) or isNull(right): return "\\N" proc escapeRe(s: string): string = result = "" for ch in s: @@ -490,6 +505,7 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex except CatchableError: discard return "false" of irILike: + if isNull(left) or isNull(right): return "\\N" proc escapeRe(s: string): string = result = "" for ch in s: @@ -505,8 +521,10 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex except CatchableError: discard return "false" of irIn: + if isNull(left): return "\\N" if expr.binRight.kind == irekSubquery: let subRows = requireExecutePlanHook()(ctx, expr.binRight.subqueryPlan) + var sawNull = false for row in subRows: # Compare against the first non-internal column only (SQL semantics) var firstVal = "" @@ -516,8 +534,14 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex firstVal = valueToString(v) found = true break - if found and firstVal == left: return "true" + if not found: continue + if isNull(firstVal): + sawNull = true + continue + if firstVal == left: return "true" + if sawNull: return "\\N" return "false" + if isNull(right): return "\\N" try: let lv = parseFloat(left) let rv = parseFloat(right) @@ -525,8 +549,10 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex except CatchableError: discard return if left == right: "true" else: "false" of irNotIn: + if isNull(left): return "\\N" if expr.binRight.kind == irekSubquery: let subRows = requireExecutePlanHook()(ctx, expr.binRight.subqueryPlan) + var sawNull = false for row in subRows: # Compare against the first non-internal column only (SQL semantics) var firstVal = "" @@ -536,8 +562,14 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex firstVal = valueToString(v) found = true break - if found and firstVal == left: return "false" + if not found: continue + if isNull(firstVal): + sawNull = true + continue + if firstVal == left: return "false" + if sawNull: return "\\N" return "true" + if isNull(right): return "\\N" try: let lv = parseFloat(left) let rv = parseFloat(right) @@ -662,6 +694,7 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex case expr.unOp of irNot: let v = evalExprOld(expr.unExpr, row, ctx) + if isNull(v): return "\\N" return if v == "true": "false" else: "true" of irIsNull: let v = evalExprOld(expr.unExpr, row, ctx) diff --git a/src/barabadb/query/exec/lower.nim b/src/barabadb/query/exec/lower.nim index c4be989..9dc47b5 100644 --- a/src/barabadb/query/exec/lower.nim +++ b/src/barabadb/query/exec/lower.nim @@ -411,9 +411,18 @@ proc lowerSelect*(node: Node): IRPlan = if node.selLimit != nil or node.selOffset != nil: let limitPlan = IRPlan(kind: irpkLimit) limitPlan.limitSource = result - limitPlan.limitCount = if node.selLimit != nil and node.selLimit.limitExpr.kind == nkIntLit: - node.selLimit.limitExpr.intVal else: 0 - limitPlan.limitOffset = if node.selOffset != nil and node.selOffset.offsetExpr.kind == nkIntLit: - node.selOffset.offsetExpr.intVal else: 0 + # limitCount: -1 = unlimited (OFFSET without LIMIT). LIMIT 0 is empty. + # Negative LIMIT/OFFSET are clamped so slicing cannot IndexDefect. + if node.selLimit != nil: + if node.selLimit.limitExpr.kind == nkIntLit: + limitPlan.limitCount = max(0'i64, node.selLimit.limitExpr.intVal) + else: + limitPlan.limitCount = 0 + else: + limitPlan.limitCount = -1 + if node.selOffset != nil and node.selOffset.offsetExpr.kind == nkIntLit: + limitPlan.limitOffset = max(0'i64, node.selOffset.offsetExpr.intVal) + else: + limitPlan.limitOffset = 0 result = limitPlan diff --git a/src/barabadb/query/exec/plan_exec.nim b/src/barabadb/query/exec/plan_exec.nim index fffb3db..e07ee4f 100644 --- a/src/barabadb/query/exec/plan_exec.nim +++ b/src/barabadb/query/exec/plan_exec.nim @@ -255,12 +255,18 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = of irpkLimit: let sourceRows = executePlan(ctx, plan.limitSource) var start = int(plan.limitOffset) + if start < 0: start = 0 if start > sourceRows.len: start = sourceRows.len + if plan.limitCount < 0: + # OFFSET without LIMIT — return the remainder. + return sourceRows[start ..< sourceRows.len] if plan.limitCount == 0: return @[] var endIdx = start + int(plan.limitCount) if endIdx > sourceRows.len: endIdx = sourceRows.len + if endIdx < start: + endIdx = start return sourceRows[start.. 0: + for i in fStart .. fEnd: + let s = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[i]], ctx)) + if not isNull(s) and s.len > 0: + inc cnt + result[rowIdx] = $cnt + continue + if expr.wfArgs.len == 0: + result[rowIdx] = "\\N" + continue + var sum = 0.0 + var cnt = 0 + var minF = 0.0 + var maxF = 0.0 + var minS = "" + var maxS = "" + var allNumeric = true + for i in fStart .. fEnd: + let s = valueToString(evalExpr(expr.wfArgs[0], rows[sortedIdxs[i]], ctx)) + if isNull(s) or s.len == 0: continue + inc cnt + if cnt == 1 or s < minS: minS = s + if cnt == 1 or s > maxS: maxS = s + try: + let f = parseFloat(s) + sum += f + if cnt == 1: + minF = f + maxF = f + else: + if f < minF: minF = f + if f > maxF: maxF = f + except CatchableError: + allNumeric = false + if cnt == 0: + result[rowIdx] = "\\N" + else: + case wfName + of "sum": + result[rowIdx] = if allNumeric: $sum else: "\\N" + of "avg": + result[rowIdx] = if allNumeric: $(sum / float(cnt)) else: "\\N" + of "min": + result[rowIdx] = if allNumeric: $minF else: minS + of "max": + result[rowIdx] = if allNumeric: $maxF else: maxS + else: + result[rowIdx] = "\\N" else: # Unknown window function — fill with null for rowIdx in sortedIdxs: diff --git a/src/barabadb/query/parser.nim b/src/barabadb/query/parser.nim index f3bd137..1b61497 100644 --- a/src/barabadb/query/parser.nim +++ b/src/barabadb/query/parser.nim @@ -194,6 +194,11 @@ proc parsePrimary(p: var Parser): Node = discard p.expect(tkWhere) node.funcFilter = p.parseExpr() discard p.expect(tkRParen) + # Window aggregate: SUM/AVG/COUNT/MIN/MAX(...) OVER (...) + if p.peek().kind == tkOver: + let overClause = p.parseOverClause() + return Node(kind: nkWindowExpr, winFunc: funcName.toLower(), winArgs: args, + winOver: overClause, line: tok.line, col: tok.col) return node of tkCase: discard p.advance() diff --git a/src/barabadb/storage/btree.nim b/src/barabadb/storage/btree.nim index 3b5d915..c22be45 100644 --- a/src/barabadb/storage/btree.nim +++ b/src/barabadb/storage/btree.nim @@ -175,6 +175,61 @@ proc scan*[K, V](btree: BTreeIndex[K, V], startKey, endKey: K): seq[(K, seq[V])] finally: release(btree.lock) +proc subtreeMinMax[K, V](node: BTreeNode[K, V]): (bool, K, K) = + ## Inclusive min/max of keys stored in this subtree's leaves. + if node == nil: + return (false, default(K), default(K)) + if node.isLeaf: + if node.keys.len == 0: + return (false, default(K), default(K)) + return (true, node.keys[0], node.keys[^1]) + var have = false + var mn, mx: K + for c in node.children: + let (ok, a, b) = subtreeMinMax(c) + if ok: + if not have: + mn = a + mx = b + have = true + else: + if a < mn: mn = a + if b > mx: mx = b + return (have, mn, mx) + +proc collectSeparatorErrors[K, V](node: BTreeNode[K, V], errors: var seq[string]) = + ## Search uses `key > separator → right child`, so every key in the left + ## subtree must be <= sep (otherwise it is routed right and missed — no + ## prev-leaf pointer). Boundary duplicates are allowed: the same key may be + ## max(left) and min(right); get/remove walk `next` to collect them. + if node == nil or node.isLeaf: + return + for i in 0..= node.children.len: + errors.add("internal node has fewer children than keys+1") + break + let (lok, _, lmax) = subtreeMinMax(node.children[i]) + let (rok, rmin, _) = subtreeMinMax(node.children[i + 1]) + let sep = node.keys[i] + if lok and lmax > sep: + errors.add("separator[" & $i & "]=" & $sep & + " < max(left)=" & $lmax & " (search would miss left keys)") + if lok and rok and rmin < lmax: + errors.add("separator[" & $i & "]=" & $sep & + " leaf order inverted: max(left)=" & $lmax & " > min(right)=" & $rmin) + collectSeparatorErrors(node.children[i], errors) + if node.children.len > 0: + collectSeparatorErrors(node.children[^1], errors) + +proc checkInvariants*[K, V](btree: BTreeIndex[K, V]): seq[string] = + ## Returns a list of separator/search-routing violations (empty = healthy). + acquire(btree.lock) + try: + result = @[] + collectSeparatorErrors(btree.root, result) + finally: + release(btree.lock) + proc len*[K, V](btree: BTreeIndex[K, V]): int = acquire(btree.lock) try: @@ -209,7 +264,9 @@ proc borrowFromLeft[K, V](node: BTreeNode[K, V], parent: BTreeNode[K, V], parent node.values.insert(borrowVal, 0) sibling.keys.setLen(sibling.keys.len - 1) sibling.values.setLen(sibling.values.len - 1) - parent.keys[parentIdx - 1] = node.keys[0] + # Search is `key > sep → right` (left-max). After lending, sep is the + # left sibling's new max — not the borrowed key now sitting in `node`. + parent.keys[parentIdx - 1] = sibling.keys[^1] else: # Borrow from internal sibling let borrowKey = sibling.keys[^1] @@ -231,7 +288,8 @@ proc borrowFromRight[K, V](node: BTreeNode[K, V], parent: BTreeNode[K, V], paren node.values.add(borrowVal) sibling.keys.delete(0) sibling.values.delete(0) - parent.keys[parentIdx] = sibling.keys[0] + # Borrowed key is now left's max; search must keep it on the left. + parent.keys[parentIdx] = node.keys[^1] else: let borrowKey = sibling.keys[0] let borrowChild = sibling.children[0] @@ -325,10 +383,14 @@ proc rebalanceAfterDelete[K, V](node: BTreeNode[K, V], root: var BTreeNode[K, V] mergeWithLeft(node, parent, parentIdx) elif hasRight: mergeWithRight(node, parent, parentIdx) + else: + return - # Recursively rebalance parent if it fell below minimum + # Merge dropped a separator from parent — rebalance up the tree. if parent == root and parent.keys.len == 0 and parent.children.len == 1: root = parent.children[0] + elif parent.keys.len < minKeysForLeaf(parent, order): + rebalanceAfterDelete(parent, root, order) proc remove*[K, V](btree: var BTreeIndex[K, V], key: K, value: V) = acquire(btree.lock) @@ -368,15 +430,17 @@ proc remove*[K, V](btree: var BTreeIndex[K, V], key: K, value: V) = else: # Internal node: recurse into child let child = node.children[i] - let oldFirstKey = if child.keys.len > 0: child.keys[0] else: default(K) let found = removeRec(child, root, order) if found: - # Update separator if child's first key changed. - # Separator node.keys[i-1] represents child's first key (for i > 0). - if i > 0 and child.keys.len > 0 and child.keys[0] != oldFirstKey: - node.keys[i - 1] = child.keys[0] - # Rebalance the child if needed + # Rebalance first — merge/borrow rewrite parent separators. rebalanceAfterDelete(child, root, order) + # Leaf children only: refresh left-max separator. Skip if merge + # already unlinked this child. Internal keys are promoted + # separators, not copies of child.keys[^1]; copying those (the + # naive H10 rewrite) makes sep < max(left) and misses keys. + if i < node.children.len and node.children[i] == child and + child.isLeaf and child.keys.len > 0 and i < node.keys.len: + node.keys[i] = child.keys[^1] return found if removeRec(btree.root, btree.root, btree.order): diff --git a/src/barabadb/storage/compaction.nim b/src/barabadb/storage/compaction.nim index 6ec15db..542e482 100644 --- a/src/barabadb/storage/compaction.nim +++ b/src/barabadb/storage/compaction.nim @@ -158,20 +158,15 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult = createdAt: tables[^1].createdAt, ) - # Verify output SSTable before deleting sources + # Verify output SSTable before mutating the catalog. Input files stay on + # disk until the caller loads the output and writes MANIFEST (crash-safe + # order: output durable + catalog updated, then unlink inputs). let (ok, msg) = verifySSTable(outputPath) if not ok: echo "[ERROR] Compaction output verification failed: ", msg try: removeFile(outputPath) except CatchableError: discard return CompactionResult() - # Remove old SSTable files - for t in tables: - try: - removeFile(t.path) - except CatchableError as e: - echo "[WARN] Failed to remove old SSTable: ", t.path, ": ", e.msg - # Update level arrays var newTables: seq[SSTableMeta] = @[] for t in cs.levels[level]: diff --git a/src/baradadb.nim b/src/baradadb.nim index a7fbe15..ff15ebc 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -41,27 +41,18 @@ proc newCompactionManager*(db: LSMTree): CompactionManager = result.strategy.rebuildFromLSM(db) proc applyCompactionResult(db: LSMTree, result: compaction.CompactionResult) = - ## Apply compaction output under the caller's lock: update sstables + MANIFEST. - ## On Linux, compact may already have unlinked inputs; we still close our mmaps. + ## Crash-safe apply: load output while inputs still exist, swap catalog, + ## write MANIFEST, then unlink inputs. A crash before MANIFEST leaves the + ## old set intact (orphan output is ignored); a crash after MANIFEST leaves + ## at worst unlinked-but-closed input files. if result.outputTables.len == 0: return - var newSSTables: seq[SSTable] = @[] - var removedPaths = initTable[string, bool]() - for t in result.inputTables: - removedPaths[t.path] = true - for sst in db.sstables.mitems: - if sst.path notin removedPaths: - newSSTables.add(sst) - else: - # Drop mmap after compact unlinked the path (fd remains valid until close) - sst.close() - + var loaded: seq[SSTable] = @[] for meta in result.outputTables: try: var sst = loadSSTable(meta.path) let name = splitFile(meta.path).name - # Prefer numeric id from filename; otherwise allocate let parsed = try: parseInt(name) except CatchableError: -1 if parsed >= 0: sst.id = parsed @@ -69,11 +60,30 @@ proc applyCompactionResult(db: LSMTree, result: compaction.CompactionResult) = sst.id = db.nextSSTableId inc db.nextSSTableId sst.level = meta.level - newSSTables.add(sst) + loaded.add(sst) db.nextSSTableId = max(db.nextSSTableId, sst.id + 1) except CatchableError as e: warn("Compaction output SSTable failed to load: " & meta.path & " — " & e.msg) + for s in loaded.mitems: + s.close() + for m in result.outputTables: + try: removeFile(m.path) except CatchableError: discard + return + var removedPaths = initTable[string, bool]() + for t in result.inputTables: + removedPaths[t.path] = true + + var newSSTables: seq[SSTable] = @[] + var toDrop: seq[SSTable] = @[] + for sst in db.sstables.mitems: + if sst.path notin removedPaths: + newSSTables.add(sst) + else: + toDrop.add(sst) + + for sst in loaded: + newSSTables.add(sst) newSSTables.sort(proc(a, b: SSTable): int = cmp(a.id, b.id)) db.sstables = newSSTables db.needsCompaction = db.countL0() >= L0CompactionTrigger @@ -84,6 +94,13 @@ proc applyCompactionResult(db: LSMTree, result: compaction.CompactionResult) = except CatchableError as e: warn("Failed to write MANIFEST after compaction: " & e.msg) + for sst in toDrop.mitems: + try: + removeFile(sst.path) + except CatchableError as e: + warn("Failed to remove compacted SSTable: " & sst.path & " — " & e.msg) + sst.close() + proc compact*(cm: CompactionManager) = # Gate first (cross-thread), then per-DB write lock withStorageGate: diff --git a/tests/bugfix_test.nim b/tests/bugfix_test.nim index 7204d72..11a7bf8 100644 --- a/tests/bugfix_test.nim +++ b/tests/bugfix_test.nim @@ -8,7 +8,15 @@ import ../src/barabadb/query/exec/dml import ../src/barabadb/core/types import ../src/barabadb/core/config import ../src/barabadb/core/replication +import ../src/barabadb/core/disttxn +import ../src/barabadb/core/websocket +import ../src/barabadb/protocol/auth +import ../src/barabadb/protocol/scram import ../src/barabadb/storage/lsm +import ../src/barabadb/storage/compaction +import ../src/barabadb/storage/btree +import std/random +import std/sets const testDir = "/tmp/baradb_bugfix_test" @@ -721,3 +729,220 @@ suite "Query correctness — audit batch 2": rm.connectReplica("r1") let lsn = rm.writeLsn(@[1'u8, 2, 3]) check lsn == 0 + + +suite "Audit batch 3 — remaining 2026-08 findings": + + test "TLS CA file auto-enables peer verify": + putEnv("BARADB_TLS_CA_FILE", "/tmp/ca.crt") + defer: delEnv("BARADB_TLS_CA_FILE") + var cfg = defaultConfig() + loadConfigFromEnv(cfg) + check cfg.tlsCaFile == "/tmp/ca.crt" + check cfg.tlsVerifyPeer == true + + test "explicit BARADB_TLS_VERIFY_PEER=false wins over CA auto-enable": + putEnv("BARADB_TLS_CA_FILE", "/tmp/ca.crt") + putEnv("BARADB_TLS_VERIFY_PEER", "false") + defer: + delEnv("BARADB_TLS_CA_FILE") + delEnv("BARADB_TLS_VERIFY_PEER") + var cfg = defaultConfig() + loadConfigFromEnv(cfg) + check cfg.tlsVerifyPeer == false + + test "production TLS without verify is rejected": + putEnv("BARADB_ENV", "production") + defer: delEnv("BARADB_ENV") + var cfg = defaultConfig() + cfg.authEnabled = true + cfg.jwtSecret = "a".repeat(32) + cfg.tlsEnabled = true + var msg = "" + try: + validateProductionConfig(cfg) + except ValueError as e: + msg = e.msg + check "TLS" in msg or "verify" in msg.toLower() + + test "OFFSET without LIMIT returns remaining rows": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')")) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'b')")) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (3, 'c')")) + let r = executeQuery(ctx, parse("SELECT id FROM users ORDER BY id OFFSET 1")) + check r.success + check r.rows.len == 2 + check valueToString(r.rows[0]["id"]) == "2" + check valueToString(r.rows[1]["id"]) == "3" + + test "LIMIT 0 returns no rows": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')")) + let r = executeQuery(ctx, parse("SELECT id FROM users LIMIT 0")) + check r.success + check r.rows.len == 0 + + test "negative LIMIT is clamped to empty, not IndexDefect": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')")) + let r = executeQuery(ctx, parse("SELECT id FROM users LIMIT -5")) + check r.success + check r.rows.len == 0 + + test "WebSocket decode rejects unmasked client frames": + let raw = encodeFrame(0x1, "SUBSCRIBE t", masked = false) + let (_, n) = decodeFrame(raw) + check n < 0 + + test "WebSocket decode accepts masked client frames": + let raw = encodeFrame(0x1, "SUBSCRIBE t", masked = true, + maskKey = [1'u8, 2, 3, 4]) + let (frame, n) = decodeFrame(raw) + check n > 0 + check frame.masked + check frame.payload == "SUBSCRIBE t" + + test "WebSocket decode rejects oversized frame length": + var raw = newString(10) + raw[0] = char(0x81) # FIN + text + raw[1] = char(0xFF) # MASK + 127 + # 8-byte length = 2 GiB + raw[2] = char(0) + raw[3] = char(0) + raw[4] = char(0) + raw[5] = char(0) + raw[6] = char(0x80) + raw[7] = char(0) + raw[8] = char(0) + raw[9] = char(0) + let (_, n) = decodeFrame(raw) + check n < 0 + + test "disttxn prepare against refused port fails closed": + var tm = newDistTxnManager() + let txn = tm.beginTransaction("coord") + txn.addParticipant("n1", "127.0.0.1", 1) + check txn.prepare() == false + check txn.isAborted + + test "compact leaves input files on disk for catalog apply": + let testDir = "/tmp/baradb_bugfix_compact_order" + removeDir(testDir) + var db = newLSMTree(testDir, 128) + defer: + db.close() + removeDir(testDir) + for round in 0 ..< L0CompactionTrigger: + db.put("r" & $round, cast[seq[byte]]("v" & $round)) + db.flush() + var cs = newCompactionStrategy(testDir) + cs.rebuildFromLSM(db) + let cr = cs.compact(0) + check cr.outputTables.len == 1 + check fileExists(cr.outputTables[0].path) + for t in cr.inputTables: + check fileExists(t.path) + + test "SCRAM unknown user fails without leaking existence": + var am = newAuthManager() + am.registerScramUser("alice", "wonderland") + var msg = "" + try: + discard am.startScram("n,,n=eve,r=abcnonceabcnonceabcn") + except ValueError as e: + msg = e.msg + check msg == "Authentication failed" + + test "SCRAM rejects mismatched channel binding": + var am = newAuthManager() + am.registerScramUser("alice", "wonderland") + let clientNonce = generateNonce() + let clientFirst = "n,,n=alice,r=" & clientNonce + let serverFirst = am.startScram(clientFirst) + var combinedNonce = "" + for part in serverFirst.split(","): + if part.startsWith("r="): combinedNonce = part[2..^1] + let (ok, err) = am.finishScram("c=AAAA,r=" & combinedNonce & ",p=AA") + check ok == false + check err == "e=channel-bindings-dont-match" + + test "SCRAM expected cbind for gs2 n is biws": + check expectedChannelBinding("n") == "biws" + + +suite "Audit batch 4 — B-tree separator (H10) and NULL equality (L4)": + + test "B-tree separators stay valid after interleaved insert/remove": + ## Search routes with `key > sep → right`. After delete, removeRec used to + ## copy the right child's first key into the separator (right-min), which + ## breaks max(left) <= sep < min(right). + var rng = initRand(20260828) + var btree = newBTreeIndex[int, string](order = 5) + var tracker = initTable[int, seq[string]]() + for i in 0..<400: + let k = rng.rand(0..80) + if rng.rand(0..2) < 2: + let v = "v" & $i + btree.insert(k, v) + if k notin tracker: tracker[k] = @[] + tracker[k].add(v) + else: + if k in tracker and tracker[k].len > 0: + let v = tracker[k][0] + btree.remove(k, v) + tracker[k].del(0) + if tracker[k].len == 0: tracker.del(k) + if i mod 50 == 49: + check btree.checkInvariants().len == 0 + for k, vals in tracker: + check btree.get(k).toHashSet == vals.toHashSet + + test "B-tree sequential fill + prefix delete keeps separators": + var btree = newBTreeIndex[int, string](order = 4) + for i in 0..<60: + btree.insert(i, "v" & $i) + for i in 0..<30: + btree.remove(i, "v" & $i) + check btree.checkInvariants().len == 0 + for i in 30..<60: + check btree.get(i) == @["v" & $i] + for i in 0..<30: + check btree.get(i).len == 0 + + test "NULL = NULL is unknown, not true": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, NULL)")) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')")) + let eqNull = executeQuery(ctx, parse("SELECT id FROM users WHERE name = NULL")) + check eqNull.success + check eqNull.rows.len == 0 + let isNull = executeQuery(ctx, parse("SELECT id FROM users WHERE name IS NULL")) + check isNull.success + check isNull.rows.len == 1 + check valueToString(isNull.rows[0]["id"]) == "1" + + test "NULL != value is unknown so WHERE excludes the row": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, NULL)")) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (2, 'bob')")) + let r = executeQuery(ctx, parse("SELECT id FROM users WHERE name != 'bob'")) + check r.success + check r.rows.len == 0 + + test "SELECT NULL = NULL yields NULL, IS NULL is true": + var ctx = setupCtx() + defer: teardown(ctx) + discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')")) + let eq = executeQuery(ctx, parse("SELECT (NULL = NULL) AS x FROM users")) + check eq.success + check eq.rows.len == 1 + check valueToString(eq.rows[0]["x"]) == "\\N" + let isn = executeQuery(ctx, parse("SELECT (NULL IS NULL) AS x FROM users")) + check isn.success + check valueToString(isn.rows[0]["x"]) == "true" diff --git a/tests/test_all.nim b/tests/test_all.nim index 3d4b040..66fdcec 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -4298,6 +4298,21 @@ suite "Window Functions": if row["name"] == "Bob": check row["last_sal"] == "90000" + test "SUM/AVG/COUNT window aggregates over a running frame": + ## Default frame is UNBOUNDED PRECEDING .. CURRENT ROW (lower.nim). + let r = qexec.executeQuery(ctx, parse( + "SELECT name, salary, SUM(salary) OVER (ORDER BY salary) AS running, COUNT(*) OVER (ORDER BY salary) AS cnt FROM employees")) + check r.success + check r.rows.len == 5 + for row in r.rows: + if row["name"] == "Charlie": + check parseFloat($row["running"]) == 70000.0 + check $row["cnt"] == "1" + if row["name"] == "Eve": + # 70000+75000+80000+90000+95000 + check parseFloat($row["running"]) == 410000.0 + check $row["cnt"] == "5" + suite "GROUP BY Aggregates": var db: LSMTree var ctx: qexec.ExecutionContext