From 31825b2589eb6efcd19397f0c01b44df8f818574 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 7 Apr 2026 13:46:50 -0500 Subject: [PATCH] subscriber: extract prepareFrameWork + add UAF regression test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit follow-up to 1eec324 (fix UAF: dupe FrameWork.hostname per submit). the dupe-at-submit logic was inline in FrameHandler.onMessage, which made it hard to regression-test the invariant. extracted a small Subscriber method that returns a FrameWork with heap-owned data + hostname, and added a unit test that: - builds a FrameWork from a to-be-freed hostname buffer - asserts the returned slices have distinct pointers from the inputs - simulates slurper.runWorker teardown by freeing the source hostname - reads the FrameWork.hostname again — would trip the testing allocator's use-after-free detection if the dupe was elided no behavior change at the submit site. tested via zig build test. --- src/subscriber.zig | 113 ++++++++++++++++++++++++++++++++++++--------- 1 file changed, 92 insertions(+), 21 deletions(-) diff --git a/src/subscriber.zig b/src/subscriber.zig index bb4c1b8..e49675d 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -256,6 +256,35 @@ pub const Subscriber = struct { return self.shutdown.load(.acquire) or self.host_shutdown.load(.acquire); } + /// build a FrameWork that owns its data + hostname, decoupled from the + /// subscriber lifetime. returns null on OOM. + /// + /// UAF-safe contract: after this returns, the caller (or the slurper) + /// may free `self.options.hostname` and the input `data` immediately + /// without affecting the work item. the worker will free both through + /// `work.allocator` when it finishes processing. + /// + /// see `test "prepareFrameWork dupes hostname and data (UAF regression)"` + fn prepareFrameWork(self: *Subscriber, data: []const u8) ?frame_worker_mod.FrameWork { + const duped = self.allocator.dupe(u8, data) catch return null; + const hostname_dup = self.allocator.dupe(u8, self.options.hostname) catch { + self.allocator.free(duped); + return null; + }; + return .{ + .data = duped, + .host_id = self.options.host_id, + .hostname = hostname_dup, + .allocator = self.allocator, + .io = self.pool_io orelse self.io, + .bc = self.bc, + .validator = self.validator, + .persist = self.persist, + .collection_index = self.collection_index, + .resyncer = self.resyncer, + }; + } + /// run the subscriber loop. reconnects with exponential backoff. /// blocks until shutdown flag is set or host is exhausted. pub fn run(self: *Subscriber) void { @@ -477,38 +506,23 @@ const FrameHandler = struct { const d = payload.getString("repo") orelse payload.getString("did"); break :blk if (d) |s| std.hash.Wyhash.hash(0, s) else sub.options.host_id; }; - const duped = sub.allocator.dupe(u8, data) catch return; - // dupe hostname per-frame: subscriber teardown (slurper.runWorker) + // dupe data + hostname per-frame: subscriber teardown (slurper.runWorker) // frees sub.options.hostname after sub.run() returns, but FrameWorks // can still be queued in the pool. borrowing the slice would be a // use-after-free (corrupt hostnames in chain-break logs, etc.). - const hostname_dup = sub.allocator.dupe(u8, sub.options.hostname) catch { - sub.allocator.free(duped); - return; - }; + const work = sub.prepareFrameWork(data) orelse return; const t0 = nanoTimestamp(io); - if (pool.submit(did_key, .{ - .data = duped, - .host_id = sub.options.host_id, - .hostname = hostname_dup, - .allocator = sub.allocator, - .io = sub.pool_io orelse sub.io, // pool_io (Threaded) for worker-safe ops - .bc = sub.bc, - .validator = sub.validator, - .persist = sub.persist, - .collection_index = sub.collection_index, - .resyncer = sub.resyncer, - }, sub.shutdown)) { + if (pool.submit(did_key, work, sub.shutdown)) { // pool accepted — advance cursor past this frame - _ = sub.bc.stats.pool_queued_bytes.fetchAdd(duped.len, .monotonic); + _ = sub.bc.stats.pool_queued_bytes.fetchAdd(work.data.len, .monotonic); if (upstream_seq) |s| sub.last_upstream_seq = s; if (nanoTimestamp(io) - t0 > 1_000_000) { // >1ms = had to wait _ = sub.bc.stats.pool_backpressure.fetchAdd(1, .monotonic); } } else { // shutdown requested — don't advance cursor so reconnect replays this frame - sub.allocator.free(duped); - sub.allocator.free(hostname_dup); + sub.allocator.free(work.data); + sub.allocator.free(work.hostname); } return; } @@ -779,6 +793,63 @@ const FrameHandler = struct { // --- tests --- +test "prepareFrameWork dupes hostname and data (UAF regression)" { + // regression test for UAF: slurper.runWorker frees sub.options.hostname + // after sub.run() returns, but FrameWorks can still be queued in the + // frame pool with that hostname slice. prepareFrameWork must heap-dupe + // both `data` and `hostname` so the work item is independent of the + // subscriber's lifetime. + // + // the test simulates subscriber teardown by freeing the source hostname + // after the FrameWork is built, then asserts the work item is still + // intact (would trip the testing allocator's use-after-free detection + // if the dupe was skipped). + const alloc = std.testing.allocator; + + const orig_hostname = try alloc.dupe(u8, "example.pds.host"); + const data_input = try alloc.dupe(u8, "raw frame bytes"); + defer alloc.free(data_input); + + var shutdown: std.atomic.Value(bool) = .{ .raw = false }; + var sub: Subscriber = .{ + .allocator = alloc, + .io = std.testing.io, + .options = .{ .hostname = orig_hostname, .host_id = 42 }, + .bc = undefined, // not dereferenced by prepareFrameWork + .validator = undefined, + .persist = null, + .collection_index = null, + .resyncer = null, + .pool = null, + .pool_io = null, + .shutdown = &shutdown, + }; + + const work = sub.prepareFrameWork(data_input) orelse return error.OutOfMemory; + defer alloc.free(work.data); + defer alloc.free(work.hostname); + + // content is correct + try std.testing.expectEqualStrings("example.pds.host", work.hostname); + try std.testing.expectEqualStrings("raw frame bytes", work.data); + + // pointers must be distinct from caller's buffers — this is the core + // UAF invariant. if the dupe was elided, these would alias. + try std.testing.expect(work.hostname.ptr != orig_hostname.ptr); + try std.testing.expect(work.data.ptr != data_input.ptr); + + // scalar fields propagate + try std.testing.expectEqual(@as(u64, 42), work.host_id); + + // simulate subscriber teardown (slurper.runWorker frees hostname) + alloc.free(orig_hostname); + + // work item must still be readable and correct — if hostname was + // borrowed instead of duped, the testing allocator would catch the + // read-after-free above when expectEqualStrings is called again here. + try std.testing.expectEqualStrings("example.pds.host", work.hostname); +} + test "decode frame via SDK and extract fields" { const cbor = zat.cbor; -- 2.51.2