From e2025a81b011b1e6dad7841e72325b0ef0161182 Mon Sep 17 00:00:00 2001 From: Conor Bronsdon <120674402+conorbronsdon@users.noreply.github.com> Date: Sun, 5 Jul 2026 23:01:12 -0700 Subject: [PATCH 1/3] fix: suppress SIGPIPE so a dead connection raises instead of killing the process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Writing to a socket whose peer has closed delivered SIGPIPE, whose default disposition terminates the whole process (exit 141) before send() could return EPIPE — so the existing `n <= 0 -> raise` error path in _send_all never ran. CPython dodges this by installing SIG_IGN for SIGPIPE at startup; a Mojo process does not. Suppress the signal at the syscall boundary, platform-gated: - Linux: pass MSG_NOSIGNAL (0x4000) in the send() flags. - macOS/BSD: set SO_NOSIGPIPE via setsockopt(2) right after socket() (there is no MSG_NOSIGNAL). Platform chosen at comptime via CompilationTarget. A dead peer now raises a catchable Error. Added test/test_connection.mojo, a self-contained regression test (stands up its own loopback peer via FFI, no Redis server) wired into CI, so it exercises the Linux MSG_NOSIGNAL path on the ubuntu runner. Against the unfixed code this test terminates the test binary with signal 13 (exit 141); with the fix it passes. Also fixes a build-blocking pixi pin: `mojo = ">=1.0.0b3"` sorts after the dev nightlies (1.0.0b3.dev…) so `pixi install` found no candidates; pinned to `>=1.0.0b3.dev0,<2`. Co-Authored-By: Claude --- .github/workflows/test.yml | 6 ++ CHANGELOG.md | 16 +++++ pixi.toml | 8 ++- src/redis/connection.mojo | 43 ++++++++++++- test/test_connection.mojo | 127 +++++++++++++++++++++++++++++++++++++ 5 files changed, 198 insertions(+), 2 deletions(-) create mode 100644 test/test_connection.mojo diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 04ebbd4..55b0523 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -28,3 +28,9 @@ jobs: # server and is run locally, not in CI, for v0.1. - name: Protocol tests run: .venv/bin/mojo run -I src test/test_resp.mojo + + # Transport regression tests — self-contained (own loopback peer, no + # Redis server). Exercises the SIGPIPE fix on Linux (MSG_NOSIGNAL): a + # write to a dead connection must raise, not kill the runner. + - name: Connection regression tests + run: .venv/bin/mojo run -I src test/test_connection.mojo diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d9f09f..3b846a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased + +### Fixed +- **Transport: suppress SIGPIPE on a dead connection.** Writing to a + socket whose peer has closed delivered `SIGPIPE`, terminating the whole + process (exit 141) before `send()` could return `EPIPE` — so the + existing error path never ran. CPython avoids this by installing + `SIG_IGN` for `SIGPIPE`; a Mojo process does not. Fixed at the syscall + boundary, platform-gated: `MSG_NOSIGNAL` on the `send()` flags on Linux, + `SO_NOSIGPIPE` via `setsockopt(2)` after `socket()` on macOS/BSD. A dead + peer now raises a catchable `Error`. Regression covered by + `test/test_connection.mojo` (self-contained, no server; runs in CI). +- **Build: `pixi install` failed to resolve.** The `mojo = ">=1.0.0b3"` + pin sorts *after* the dev nightlies (`1.0.0b3.dev…`), so the solver + found no candidates. Pinned to `>=1.0.0b3.dev0,<2`. + ## 0.1.0 — 2026-07-05 Initial release. A synchronous Redis client in pure Mojo speaking RESP2 diff --git a/pixi.toml b/pixi.toml index 369ce91..73c6cf2 100644 --- a/pixi.toml +++ b/pixi.toml @@ -9,6 +9,9 @@ version = "0.1.0" [tasks] # Protocol unit tests — pure, no server required (this is what CI runs). test = "mojo run -I src test/test_resp.mojo" +# Transport regression tests — self-contained (each test stands up its own +# throwaway loopback peer via FFI); no Redis server needed, CI-safe. +test-connection = "mojo run -I src test/test_connection.mojo" # Integration tests — require a live Redis server. Set REDIS_HOST/REDIS_PORT # (defaults 127.0.0.1:6379). e.g.: # redis-server --port 6399 --daemonize yes --save "" --appendonly no @@ -17,4 +20,7 @@ test-integration = "mojo run -I src test/test_integration.mojo" demo = "mojo run -I src examples/demo.mojo" [dependencies] -mojo = ">=1.0.0b3,<2" +# Must be `.dev0`, not `1.0.0b3`: a bare `1.0.0b3` sorts *after* the dev +# nightlies (e.g. 1.0.0b3.dev2026070506), so the solver rejects the only +# available builds and `pixi install` fails to resolve. +mojo = ">=1.0.0b3.dev0,<2" diff --git a/src/redis/connection.mojo b/src/redis/connection.mojo index e063211..78054c6 100644 --- a/src/redis/connection.mojo +++ b/src/redis/connection.mojo @@ -27,6 +27,7 @@ up yet. Sockets are blocking; there is no connect/read timeout. from std.ffi import external_call from std.memory import UnsafePointer, alloc +from std.sys.info import CompilationTarget from .resp import RespValue, ParseResult, parse_reply, encode_command @@ -35,6 +36,21 @@ comptime _SOCK_STREAM = Int32(1) comptime _SOCKADDR_IN_SIZE = 16 comptime _RECV_CHUNK = 4096 +# --- SIGPIPE suppression ----------------------------------------------------- +# Writing to a socket whose peer has closed delivers SIGPIPE, whose default +# disposition terminates the process (exit 141) *before* send() can return +# EPIPE — so the error-handling path below would never run. CPython dodges +# this by installing SIG_IGN for SIGPIPE at startup; a Mojo process does not, +# so we suppress the signal at the syscall boundary instead, and let the +# existing `n <= 0 -> raise` path surface a catchable Error. This is done +# platform-specifically because the two POSIX flavors expose different knobs: +# * Linux: pass MSG_NOSIGNAL in the send() flags (per-call). +# * macOS/BSD: no MSG_NOSIGNAL — set SO_NOSIGPIPE once via setsockopt(2) +# right after socket() (per-socket). +comptime _MSG_NOSIGNAL = Int32(0x4000) # Linux +comptime _SOL_SOCKET = Int32(0xFFFF) # macOS/BSD +comptime _SO_NOSIGPIPE = Int32(0x1022) # macOS/BSD + def _parse_ipv4(host: String) raises -> InlineArray[UInt8, 4]: """Parse a dotted-quad IPv4 address into 4 network-order bytes. @@ -111,6 +127,23 @@ struct Connection(Movable): if fd < 0: raise Error("redis: socket() failed (rc=" + String(fd) + ")") + # macOS/BSD has no MSG_NOSIGNAL, so silence SIGPIPE per-socket here. + # On Linux this is a no-op (handled per-send via MSG_NOSIGNAL below). + comptime if CompilationTarget.is_macos(): + var optval = alloc[Int32](1) + optval.init_pointee_copy(Int32(1)) + var so_rc = external_call["setsockopt", Int32]( + fd, _SOL_SOCKET, _SO_NOSIGPIPE, optval, UInt32(4) + ) + optval.free() + if so_rc < 0: + _ = external_call["close", Int32](fd) + raise Error( + "redis: setsockopt(SO_NOSIGPIPE) failed (rc=" + + String(so_rc) + + ")" + ) + # Build a Linux sockaddr_in (16 bytes): family(2) port(2, BE) # addr(4, BE) then 8 zero padding bytes. var sa = alloc[UInt8](_SOCKADDR_IN_SIZE) @@ -147,9 +180,17 @@ struct Connection(Movable): var total = len(data) var sent = 0 var ptr = data.unsafe_ptr() + # On Linux, MSG_NOSIGNAL suppresses SIGPIPE per-call; on macOS/BSD the + # signal is already off via SO_NOSIGPIPE (set in connect()), so send + # with no flags. Either way a dead peer now returns EPIPE and raises. + var send_flags: Int32 + comptime if CompilationTarget.is_linux(): + send_flags = _MSG_NOSIGNAL + else: + send_flags = Int32(0) while sent < total: var n = external_call["send", Int]( - self.fd, ptr + sent, total - sent, Int32(0) + self.fd, ptr + sent, total - sent, send_flags ) if n <= 0: raise Error("redis: send() failed (rc=" + String(n) + ")") diff --git a/test/test_connection.mojo b/test/test_connection.mojo new file mode 100644 index 0000000..b968307 --- /dev/null +++ b/test/test_connection.mojo @@ -0,0 +1,127 @@ +"""Transport-layer regression tests for `Connection`. + +Unlike `test_integration.mojo`, these need **no** Redis server: each test +stands up its own throwaway loopback TCP peer via libc FFI, so the suite is +self-contained and safe to run in CI. + +The headline test is `test_send_on_dead_connection_raises`, which pins the +SIGPIPE fix: writing to a socket whose peer has closed must surface a +*catchable* `Error`, not deliver SIGPIPE and kill the whole process. Before +the fix this test does not merely fail — it terminates the test binary with +signal 13 (exit 141), taking every other test down with it. That crash *is* +the regression signal. +""" + +from std.ffi import external_call +from std.memory import UnsafePointer, alloc +from std.time import sleep +from std.testing import assert_true, TestSuite + +from redis.connection import Connection + +comptime _AF_INET = Int32(2) +comptime _SOCK_STREAM = Int32(1) +comptime _SOL_SOCKET = Int32(0xFFFF) +comptime _SO_REUSEADDR = Int32(0x0004) + + +@fieldwise_init +struct _Listener(Copyable, Movable): + var fd: Int32 + var port: Int + + +def _spawn_loopback_listener() raises -> _Listener: + """Bind+listen on an ephemeral 127.0.0.1 port. Returns fd + chosen port. + + Nothing is accepted yet: a client `connect()` completes against the + kernel's accept queue, and the caller accepts+closes it afterwards. + """ + var fd = external_call["socket", Int32](_AF_INET, _SOCK_STREAM, Int32(0)) + if fd < 0: + raise Error("test listener: socket() failed") + + var one = alloc[Int32](1) + one.init_pointee_copy(Int32(1)) + _ = external_call["setsockopt", Int32]( + fd, _SOL_SOCKET, _SO_REUSEADDR, one, UInt32(4) + ) + one.free() + + # sockaddr_in with port 0 -> kernel picks a free ephemeral port. + var sa = alloc[UInt8](16) + for k in range(16): + (sa + k).init_pointee_copy(UInt8(0)) + (sa + 0).init_pointee_copy(UInt8(2)) # AF_INET + (sa + 4).init_pointee_copy(UInt8(127)) # 127.0.0.1 + (sa + 7).init_pointee_copy(UInt8(1)) + var brc = external_call["bind", Int32](fd, sa, UInt32(16)) + if brc < 0: + sa.free() + _ = external_call["close", Int32](fd) + raise Error("test listener: bind() failed") + + # Read back the assigned port via getsockname (BE bytes 2..3). + var alen = alloc[UInt32](1) + alen.init_pointee_copy(UInt32(16)) + var grc = external_call["getsockname", Int32](fd, sa, alen) + alen.free() + if grc < 0: + sa.free() + _ = external_call["close", Int32](fd) + raise Error("test listener: getsockname() failed") + var port = (Int((sa + 2).load()) << 8) | Int((sa + 3).load()) + sa.free() + + if external_call["listen", Int32](fd, Int32(1)) < 0: + _ = external_call["close", Int32](fd) + raise Error("test listener: listen() failed") + return _Listener(fd, port) + + +def _accept_then_close_cleanly(listen_fd: Int32) raises: + """Accept the queued connection and close it with no unread data, so the + peer sees a clean FIN (the case that yields EPIPE, hence SIGPIPE).""" + var addr = alloc[UInt8](16) + var alen = alloc[UInt32](1) + alen.init_pointee_copy(UInt32(16)) + var cfd = external_call["accept", Int32](listen_fd, addr, alen) + addr.free() + alen.free() + if cfd >= 0: + _ = external_call["close", Int32](cfd) + _ = external_call["close", Int32](listen_fd) + + +def test_send_on_dead_connection_raises() raises: + """A write to a server-closed socket must raise, not kill the process. + + Regression guard for the SIGPIPE fix (SO_NOSIGPIPE on macOS/BSD, + MSG_NOSIGNAL on Linux). Reaching the assertion at all proves the signal + was suppressed; the assertion proves the EPIPE surfaced as an `Error`. + """ + var listener = _spawn_loopback_listener() + + var conn = Connection(String("127.0.0.1"), listener.port) + conn.connect() + _accept_then_close_cleanly(listener.fd) + + # Let the peer's FIN arrive before the first write. + sleep(0.3) + + var raised = False + try: + # First write usually succeeds (buffered), peer replies RST; a + # subsequent write hits the broken pipe. Loop to cross that edge. + for _ in range(50): + conn.send_command(["PING"]) + sleep(0.02) + except: + raised = True + assert_true( + raised, "send() on a dead connection should raise a catchable Error" + ) + + +def main() raises: + TestSuite.discover_tests[__functions_in_module()]().run() From 6557e161739d533f2c3ff6656218b957d760fd1a Mon Sep 17 00:00:00 2001 From: Conor Bronsdon <120674402+conorbronsdon@users.noreply.github.com> Date: Sun, 5 Jul 2026 23:06:07 -0700 Subject: [PATCH 2/3] docs: add Coming-from-Python section + fix stale counts Adds a "Coming from Python" on-ramp table (verified against the repo's own examples/ and tests) and corrects suite/test-count accuracy issues. Co-Authored-By: Claude --- CHANGELOG.md | 2 +- README.md | 25 ++++++++++++++++++++++--- src/redis/resp.mojo | 29 +++++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b846a0..5f67e60 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,5 +33,5 @@ over a direct libc TCP socket (no third-party networking dependency). `expire`, `ttl`, `keys`, `ping`, `lpush`/`rpush`/`lpop`/`rpop`/`lrange`, `hset`/`hget`/`hgetall`, `flushdb`, an `execute` escape hatch, and `pipeline`/`execute_pipeline`. -- **Tests**: 29 protocol unit tests (no network) and 21 integration tests +- **Tests**: 34 protocol unit tests (no network) and 21 integration tests against a live server, all passing. CI runs the protocol suite only. diff --git a/README.md b/README.md index 5ea49f6..562b839 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,23 @@ def main() raises: r.close() ``` +### Coming from Python + +If you know `redis-py`, the client methods map directly: + +| Python (`redis-py`) | mojo-redis | +| ----------------------------------- | -------------------------------------- | +| `r = redis.Redis(host, port)` | `var r = Redis(host, port)` | +| `r.set("k", "v")` | `r.set("k", "v")` | +| `r.set("k", "v", ex=60)` | `r.set("k", "v", ex=60)` | +| `r.get("k")` | `r.get("k").value()` | +| `r.incr("k")` / `r.decr("k")` | `r.incr("k")` / `r.decr("k")` | +| `r.hset("h", "f", "v")` / `r.hgetall("h")` | `r.hset("h", "f", "v")` / `r.hgetall("h")` | + +One difference: `get` returns an `Optional[String]`, so call `.value()` when +the key exists (or test the optional with `Bool(...)` for a miss) rather than +getting a bare string or `None`. + ## What it handles - **RESP2 protocol**: a serializer for command arrays and a parser for @@ -117,7 +134,7 @@ Two layers: 1. **Protocol unit tests** (`test/test_resp.mojo`): pure and network-free. RESP serialize/parse round-trips, split-buffer (incremental) parsing, nil handling, nested arrays, error replies. - 29 tests. This is what CI runs. + 34 tests. This is what CI runs. ```bash pixi run test # or: mojo run -I src test/test_resp.mojo @@ -134,7 +151,7 @@ Two layers: REDIS_PORT=6399 pixi run test-integration ``` -Both suites pass: 29 protocol tests plus 21 live-Redis integration tests. +Both suites pass: 34 protocol tests plus 21 live-Redis integration tests. ## Layout @@ -152,9 +169,11 @@ examples/ ## Part of a pure-Mojo library suite -Ten pure-Mojo libraries that mirror familiar Python stdlib and PyPI APIs, +Eleven pure-Mojo libraries that mirror familiar Python stdlib and PyPI APIs, filling gaps in the native Mojo ecosystem: +- [mojo-xml](https://github.com/conorbronsdon/mojo-xml) — general-purpose XML + parsing, an ElementTree-shaped DOM (Python's `xml.etree.ElementTree`) - [mojo-feed](https://github.com/conorbronsdon/mojo-feed) — RSS, Atom, and JSON Feed parsing (Python's `feedparser`) - [mojo-captions](https://github.com/conorbronsdon/mojo-captions) — SRT and diff --git a/src/redis/resp.mojo b/src/redis/resp.mojo index f40d632..c277951 100644 --- a/src/redis/resp.mojo +++ b/src/redis/resp.mojo @@ -44,6 +44,16 @@ comptime _NINE = UInt8(ord("9")) # sibling libraries cap analogous recursion at 256, and 128 is ample here. comptime _MAX_DEPTH = 128 +# Ceilings on a single reply's advertised sizes, to bound the memory a hostile +# or buggy server can make the client buffer/allocate. A bulk-string length or +# array element count larger than these is rejected outright (raises) rather +# than driving unbounded `recv`/allocation. Without the bulk cap, a header like +# `$999999999999999999\r\n` would make the transport `recv` forever, growing +# its buffer without limit, waiting for bytes that never come. 512 MiB matches +# Redis's own default `proto-max-bulk-len`. +comptime _MAX_BULK_LEN = 512 * 1024 * 1024 # 512 MiB, per bulk string +comptime _MAX_ARRAY_COUNT = 1024 * 1024 # 1,048,576 elements, per array + struct RespValue(Copyable, Movable, Writable): """A decoded RESP2 reply value. @@ -329,6 +339,16 @@ def _parse( if length < -1: # Only `-1` (null bulk) is a legal negative length. raise Error("redis: invalid bulk-string length") + if length > _MAX_BULK_LEN: + # DoS guard: reject an oversized advertised length before it can + # drive unbounded recv/allocation (see `_MAX_BULK_LEN`). + raise Error( + "redis: bulk-string length " + + String(length) + + " exceeds cap of " + + String(_MAX_BULK_LEN) + + " bytes" + ) # A complete bulk string's body plus its trailing CRLF must already # fit in the buffer window past the header. If `length` is larger, # the reply is either not yet fully received (incomplete) or hostile @@ -349,6 +369,15 @@ def _parse( var count = _atoi(header) if count < 0: return ParseResult(True, after - offset, RespValue.nil()) + if count > _MAX_ARRAY_COUNT: + # DoS guard: an oversized element count amplifies memory even with + # tiny elements; reject before allocating (see `_MAX_ARRAY_COUNT`). + raise Error( + "redis: array element count " + + String(count) + + " exceeds cap of " + + String(_MAX_ARRAY_COUNT) + ) var items = List[ArcPointer[RespValue]]() var pos = after for _ in range(count): From 73d5853ad9950f443259f0cd9dbad3397e5ee5ef Mon Sep 17 00:00:00 2001 From: Conor Bronsdon <120674402+conorbronsdon@users.noreply.github.com> Date: Sun, 5 Jul 2026 23:18:47 -0700 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20keep=20PR=20docs-only=20=E2=80=94?= =?UTF-8?q?=20strip=20code=20swept=20in=20from=20the=20fix=20branch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SIGPIPE/DoS/pixi code + its CHANGELOG entries belong to the security fix PR; this branch is docs (Coming-from-Python + test count) only. Co-Authored-By: Claude --- .github/workflows/test.yml | 6 -- CHANGELOG.md | 16 ----- pixi.toml | 8 +-- src/redis/connection.mojo | 43 +------------ src/redis/resp.mojo | 29 --------- test/test_connection.mojo | 127 ------------------------------------- 6 files changed, 2 insertions(+), 227 deletions(-) delete mode 100644 test/test_connection.mojo diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 55b0523..04ebbd4 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -28,9 +28,3 @@ jobs: # server and is run locally, not in CI, for v0.1. - name: Protocol tests run: .venv/bin/mojo run -I src test/test_resp.mojo - - # Transport regression tests — self-contained (own loopback peer, no - # Redis server). Exercises the SIGPIPE fix on Linux (MSG_NOSIGNAL): a - # write to a dead connection must raise, not kill the runner. - - name: Connection regression tests - run: .venv/bin/mojo run -I src test/test_connection.mojo diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f67e60..cb539a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,21 +1,5 @@ # Changelog -## Unreleased - -### Fixed -- **Transport: suppress SIGPIPE on a dead connection.** Writing to a - socket whose peer has closed delivered `SIGPIPE`, terminating the whole - process (exit 141) before `send()` could return `EPIPE` — so the - existing error path never ran. CPython avoids this by installing - `SIG_IGN` for `SIGPIPE`; a Mojo process does not. Fixed at the syscall - boundary, platform-gated: `MSG_NOSIGNAL` on the `send()` flags on Linux, - `SO_NOSIGPIPE` via `setsockopt(2)` after `socket()` on macOS/BSD. A dead - peer now raises a catchable `Error`. Regression covered by - `test/test_connection.mojo` (self-contained, no server; runs in CI). -- **Build: `pixi install` failed to resolve.** The `mojo = ">=1.0.0b3"` - pin sorts *after* the dev nightlies (`1.0.0b3.dev…`), so the solver - found no candidates. Pinned to `>=1.0.0b3.dev0,<2`. - ## 0.1.0 — 2026-07-05 Initial release. A synchronous Redis client in pure Mojo speaking RESP2 diff --git a/pixi.toml b/pixi.toml index 73c6cf2..369ce91 100644 --- a/pixi.toml +++ b/pixi.toml @@ -9,9 +9,6 @@ version = "0.1.0" [tasks] # Protocol unit tests — pure, no server required (this is what CI runs). test = "mojo run -I src test/test_resp.mojo" -# Transport regression tests — self-contained (each test stands up its own -# throwaway loopback peer via FFI); no Redis server needed, CI-safe. -test-connection = "mojo run -I src test/test_connection.mojo" # Integration tests — require a live Redis server. Set REDIS_HOST/REDIS_PORT # (defaults 127.0.0.1:6379). e.g.: # redis-server --port 6399 --daemonize yes --save "" --appendonly no @@ -20,7 +17,4 @@ test-integration = "mojo run -I src test/test_integration.mojo" demo = "mojo run -I src examples/demo.mojo" [dependencies] -# Must be `.dev0`, not `1.0.0b3`: a bare `1.0.0b3` sorts *after* the dev -# nightlies (e.g. 1.0.0b3.dev2026070506), so the solver rejects the only -# available builds and `pixi install` fails to resolve. -mojo = ">=1.0.0b3.dev0,<2" +mojo = ">=1.0.0b3,<2" diff --git a/src/redis/connection.mojo b/src/redis/connection.mojo index 78054c6..e063211 100644 --- a/src/redis/connection.mojo +++ b/src/redis/connection.mojo @@ -27,7 +27,6 @@ up yet. Sockets are blocking; there is no connect/read timeout. from std.ffi import external_call from std.memory import UnsafePointer, alloc -from std.sys.info import CompilationTarget from .resp import RespValue, ParseResult, parse_reply, encode_command @@ -36,21 +35,6 @@ comptime _SOCK_STREAM = Int32(1) comptime _SOCKADDR_IN_SIZE = 16 comptime _RECV_CHUNK = 4096 -# --- SIGPIPE suppression ----------------------------------------------------- -# Writing to a socket whose peer has closed delivers SIGPIPE, whose default -# disposition terminates the process (exit 141) *before* send() can return -# EPIPE — so the error-handling path below would never run. CPython dodges -# this by installing SIG_IGN for SIGPIPE at startup; a Mojo process does not, -# so we suppress the signal at the syscall boundary instead, and let the -# existing `n <= 0 -> raise` path surface a catchable Error. This is done -# platform-specifically because the two POSIX flavors expose different knobs: -# * Linux: pass MSG_NOSIGNAL in the send() flags (per-call). -# * macOS/BSD: no MSG_NOSIGNAL — set SO_NOSIGPIPE once via setsockopt(2) -# right after socket() (per-socket). -comptime _MSG_NOSIGNAL = Int32(0x4000) # Linux -comptime _SOL_SOCKET = Int32(0xFFFF) # macOS/BSD -comptime _SO_NOSIGPIPE = Int32(0x1022) # macOS/BSD - def _parse_ipv4(host: String) raises -> InlineArray[UInt8, 4]: """Parse a dotted-quad IPv4 address into 4 network-order bytes. @@ -127,23 +111,6 @@ struct Connection(Movable): if fd < 0: raise Error("redis: socket() failed (rc=" + String(fd) + ")") - # macOS/BSD has no MSG_NOSIGNAL, so silence SIGPIPE per-socket here. - # On Linux this is a no-op (handled per-send via MSG_NOSIGNAL below). - comptime if CompilationTarget.is_macos(): - var optval = alloc[Int32](1) - optval.init_pointee_copy(Int32(1)) - var so_rc = external_call["setsockopt", Int32]( - fd, _SOL_SOCKET, _SO_NOSIGPIPE, optval, UInt32(4) - ) - optval.free() - if so_rc < 0: - _ = external_call["close", Int32](fd) - raise Error( - "redis: setsockopt(SO_NOSIGPIPE) failed (rc=" - + String(so_rc) - + ")" - ) - # Build a Linux sockaddr_in (16 bytes): family(2) port(2, BE) # addr(4, BE) then 8 zero padding bytes. var sa = alloc[UInt8](_SOCKADDR_IN_SIZE) @@ -180,17 +147,9 @@ struct Connection(Movable): var total = len(data) var sent = 0 var ptr = data.unsafe_ptr() - # On Linux, MSG_NOSIGNAL suppresses SIGPIPE per-call; on macOS/BSD the - # signal is already off via SO_NOSIGPIPE (set in connect()), so send - # with no flags. Either way a dead peer now returns EPIPE and raises. - var send_flags: Int32 - comptime if CompilationTarget.is_linux(): - send_flags = _MSG_NOSIGNAL - else: - send_flags = Int32(0) while sent < total: var n = external_call["send", Int]( - self.fd, ptr + sent, total - sent, send_flags + self.fd, ptr + sent, total - sent, Int32(0) ) if n <= 0: raise Error("redis: send() failed (rc=" + String(n) + ")") diff --git a/src/redis/resp.mojo b/src/redis/resp.mojo index c277951..f40d632 100644 --- a/src/redis/resp.mojo +++ b/src/redis/resp.mojo @@ -44,16 +44,6 @@ comptime _NINE = UInt8(ord("9")) # sibling libraries cap analogous recursion at 256, and 128 is ample here. comptime _MAX_DEPTH = 128 -# Ceilings on a single reply's advertised sizes, to bound the memory a hostile -# or buggy server can make the client buffer/allocate. A bulk-string length or -# array element count larger than these is rejected outright (raises) rather -# than driving unbounded `recv`/allocation. Without the bulk cap, a header like -# `$999999999999999999\r\n` would make the transport `recv` forever, growing -# its buffer without limit, waiting for bytes that never come. 512 MiB matches -# Redis's own default `proto-max-bulk-len`. -comptime _MAX_BULK_LEN = 512 * 1024 * 1024 # 512 MiB, per bulk string -comptime _MAX_ARRAY_COUNT = 1024 * 1024 # 1,048,576 elements, per array - struct RespValue(Copyable, Movable, Writable): """A decoded RESP2 reply value. @@ -339,16 +329,6 @@ def _parse( if length < -1: # Only `-1` (null bulk) is a legal negative length. raise Error("redis: invalid bulk-string length") - if length > _MAX_BULK_LEN: - # DoS guard: reject an oversized advertised length before it can - # drive unbounded recv/allocation (see `_MAX_BULK_LEN`). - raise Error( - "redis: bulk-string length " - + String(length) - + " exceeds cap of " - + String(_MAX_BULK_LEN) - + " bytes" - ) # A complete bulk string's body plus its trailing CRLF must already # fit in the buffer window past the header. If `length` is larger, # the reply is either not yet fully received (incomplete) or hostile @@ -369,15 +349,6 @@ def _parse( var count = _atoi(header) if count < 0: return ParseResult(True, after - offset, RespValue.nil()) - if count > _MAX_ARRAY_COUNT: - # DoS guard: an oversized element count amplifies memory even with - # tiny elements; reject before allocating (see `_MAX_ARRAY_COUNT`). - raise Error( - "redis: array element count " - + String(count) - + " exceeds cap of " - + String(_MAX_ARRAY_COUNT) - ) var items = List[ArcPointer[RespValue]]() var pos = after for _ in range(count): diff --git a/test/test_connection.mojo b/test/test_connection.mojo deleted file mode 100644 index b968307..0000000 --- a/test/test_connection.mojo +++ /dev/null @@ -1,127 +0,0 @@ -"""Transport-layer regression tests for `Connection`. - -Unlike `test_integration.mojo`, these need **no** Redis server: each test -stands up its own throwaway loopback TCP peer via libc FFI, so the suite is -self-contained and safe to run in CI. - -The headline test is `test_send_on_dead_connection_raises`, which pins the -SIGPIPE fix: writing to a socket whose peer has closed must surface a -*catchable* `Error`, not deliver SIGPIPE and kill the whole process. Before -the fix this test does not merely fail — it terminates the test binary with -signal 13 (exit 141), taking every other test down with it. That crash *is* -the regression signal. -""" - -from std.ffi import external_call -from std.memory import UnsafePointer, alloc -from std.time import sleep -from std.testing import assert_true, TestSuite - -from redis.connection import Connection - -comptime _AF_INET = Int32(2) -comptime _SOCK_STREAM = Int32(1) -comptime _SOL_SOCKET = Int32(0xFFFF) -comptime _SO_REUSEADDR = Int32(0x0004) - - -@fieldwise_init -struct _Listener(Copyable, Movable): - var fd: Int32 - var port: Int - - -def _spawn_loopback_listener() raises -> _Listener: - """Bind+listen on an ephemeral 127.0.0.1 port. Returns fd + chosen port. - - Nothing is accepted yet: a client `connect()` completes against the - kernel's accept queue, and the caller accepts+closes it afterwards. - """ - var fd = external_call["socket", Int32](_AF_INET, _SOCK_STREAM, Int32(0)) - if fd < 0: - raise Error("test listener: socket() failed") - - var one = alloc[Int32](1) - one.init_pointee_copy(Int32(1)) - _ = external_call["setsockopt", Int32]( - fd, _SOL_SOCKET, _SO_REUSEADDR, one, UInt32(4) - ) - one.free() - - # sockaddr_in with port 0 -> kernel picks a free ephemeral port. - var sa = alloc[UInt8](16) - for k in range(16): - (sa + k).init_pointee_copy(UInt8(0)) - (sa + 0).init_pointee_copy(UInt8(2)) # AF_INET - (sa + 4).init_pointee_copy(UInt8(127)) # 127.0.0.1 - (sa + 7).init_pointee_copy(UInt8(1)) - var brc = external_call["bind", Int32](fd, sa, UInt32(16)) - if brc < 0: - sa.free() - _ = external_call["close", Int32](fd) - raise Error("test listener: bind() failed") - - # Read back the assigned port via getsockname (BE bytes 2..3). - var alen = alloc[UInt32](1) - alen.init_pointee_copy(UInt32(16)) - var grc = external_call["getsockname", Int32](fd, sa, alen) - alen.free() - if grc < 0: - sa.free() - _ = external_call["close", Int32](fd) - raise Error("test listener: getsockname() failed") - var port = (Int((sa + 2).load()) << 8) | Int((sa + 3).load()) - sa.free() - - if external_call["listen", Int32](fd, Int32(1)) < 0: - _ = external_call["close", Int32](fd) - raise Error("test listener: listen() failed") - return _Listener(fd, port) - - -def _accept_then_close_cleanly(listen_fd: Int32) raises: - """Accept the queued connection and close it with no unread data, so the - peer sees a clean FIN (the case that yields EPIPE, hence SIGPIPE).""" - var addr = alloc[UInt8](16) - var alen = alloc[UInt32](1) - alen.init_pointee_copy(UInt32(16)) - var cfd = external_call["accept", Int32](listen_fd, addr, alen) - addr.free() - alen.free() - if cfd >= 0: - _ = external_call["close", Int32](cfd) - _ = external_call["close", Int32](listen_fd) - - -def test_send_on_dead_connection_raises() raises: - """A write to a server-closed socket must raise, not kill the process. - - Regression guard for the SIGPIPE fix (SO_NOSIGPIPE on macOS/BSD, - MSG_NOSIGNAL on Linux). Reaching the assertion at all proves the signal - was suppressed; the assertion proves the EPIPE surfaced as an `Error`. - """ - var listener = _spawn_loopback_listener() - - var conn = Connection(String("127.0.0.1"), listener.port) - conn.connect() - _accept_then_close_cleanly(listener.fd) - - # Let the peer's FIN arrive before the first write. - sleep(0.3) - - var raised = False - try: - # First write usually succeeds (buffered), peer replies RST; a - # subsequent write hits the broken pipe. Loop to cross that edge. - for _ in range(50): - conn.send_command(["PING"]) - sleep(0.02) - except: - raised = True - assert_true( - raised, "send() on a dead connection should raise a catchable Error" - ) - - -def main() raises: - TestSuite.discover_tests[__functions_in_module()]().run()