From 7e86bfca679d33694eeb7b2b338533af63d1f3c5 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 8 Jul 2026 16:09:10 -0500 Subject: [PATCH] ingest: commit signature verification + durable cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit verify: DID signing-key cache backed by zat's DidResolver (PLC url configurable — the simulator serves DID docs itself); commit signatures checked via zat.verifyCommitCar. signature mismatch evicts the key, re-resolves once (rotation), and drops the event if still bad (new drop reason invalid_signature); resolution failure passes unverified and counts. #identity events evict the cached key. upstream runs full Sync 1.1 (MST-diff verify, per-DID rev chains, getRepo resync, per-DID worker pool) — those are tracked TODOs in verify.zig's doc comment. cursor: last seen upstream seq persisted to /cursor (tmp+fsync+rename, 5s interval) and loaded at startup, so restarts resume instead of re-tailing live. flags: --plc, --data-dir, --no-verify. verified against the simulator: e2e suite green with verification on, zero invalid-signature drops on honest traffic, restart resumes from the persisted cursor. Co-Authored-By: Claude Fable 5 --- src/internal/convert.zig | 1 + src/internal/cursor.zig | 97 +++++++++++++++++++++++ src/internal/ingest.zig | 33 ++++++++ src/internal/verify.zig | 165 +++++++++++++++++++++++++++++++++++++++ src/main.zig | 26 ++++++ src/tests.zig | 2 + 6 files changed, 324 insertions(+) create mode 100644 src/internal/cursor.zig create mode 100644 src/internal/verify.zig diff --git a/src/internal/convert.zig b/src/internal/convert.zig index 9863ea1..6c1317d 100644 --- a/src/internal/convert.zig +++ b/src/internal/convert.zig @@ -13,6 +13,7 @@ const firehose = zat.firehose; const Allocator = std.mem.Allocator; pub const DropReason = enum { + invalid_signature, invalid_rev, invalid_collection, invalid_rkey, diff --git a/src/internal/cursor.zig b/src/internal/cursor.zig new file mode 100644 index 0000000..74263f2 --- /dev/null +++ b/src/internal/cursor.zig @@ -0,0 +1,97 @@ +//! durable upstream cursor — resume subscribeRepos across restarts +//! +//! upstream jetstream persists its cursor in pebble, committed in the same +//! synced batch as the data it covers (cursor ≤ last durable event). stream +//! has no durable event store yet, so the invariant degenerates to "resume +//! near where we left off": the last seen upstream seq is written to a file +//! (tmp + fsync + rename) at a fixed interval and loaded at startup. +//! at-least-once on restart within the flush window; events during downtime +//! are replayed by the relay's own backfill window when the cursor allows. + +const std = @import("std"); + +const log = std.log.scoped(.stream); + +pub const flush_interval_us: i64 = 5 * std.time.us_per_s; + +pub const Store = struct { + dir: std.Io.Dir, + last_flushed: i64 = 0, // wall clock µs of last write + pending: ?i64 = null, + + const file_name = "cursor"; + + pub fn init(io: std.Io, data_dir: []const u8) !Store { + return .{ .dir = try std.Io.Dir.cwd().createDirPathOpen(io, data_dir, .{}) }; + } + + pub fn deinit(self: *Store, io: std.Io) void { + self.dir.close(io); + } + + pub fn load(self: *Store, io: std.Io) ?i64 { + var buf: [32]u8 = undefined; + const n = self.dir.readFile(io, file_name, &buf) catch return null; + const trimmed = std.mem.trim(u8, n, " \n\t"); + return std.fmt.parseInt(i64, trimmed, 10) catch null; + } + + /// record the latest seq; writes through to disk at most once per + /// flush interval. call flush() on shutdown for the tail. + pub fn update(self: *Store, io: std.Io, seq: i64, now_us: i64) void { + self.pending = seq; + if (now_us - self.last_flushed < flush_interval_us) return; + self.flush(io); + self.last_flushed = now_us; + } + + pub fn flush(self: *Store, io: std.Io) void { + const seq = self.pending orelse return; + self.writeSeq(io, seq) catch |err| { + log.warn("cursor flush failed: {s}", .{@errorName(err)}); + return; + }; + self.pending = null; + } + + fn writeSeq(self: *Store, io: std.Io, seq: i64) !void { + var buf: [32]u8 = undefined; + const text = try std.fmt.bufPrint(&buf, "{d}\n", .{seq}); + const tmp_name = file_name ++ ".tmp"; + var f = try self.dir.createFile(io, tmp_name, .{}); + errdefer f.close(io); + try f.writeStreamingAll(io, text); + try f.sync(io); + f.close(io); + try self.dir.rename(tmp_name, self.dir, file_name, io); + } +}; + +// === tests === + +const testing = std.testing; + +test "cursor round-trip and interval flushing" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + var store: Store = .{ .dir = try tmp.dir.createDirPathOpen(io, "data", .{}) }; + defer store.deinit(io); + try testing.expect(store.load(io) == null); + + // first update flushes immediately (last_flushed = 0) + store.update(io, 41, 10 * std.time.us_per_s); + try testing.expectEqual(@as(?i64, 41), store.load(io)); + + // within the interval: buffered, not written + store.update(io, 42, 11 * std.time.us_per_s); + try testing.expectEqual(@as(?i64, 41), store.load(io)); + + // shutdown flush picks up the pending seq + store.flush(io); + try testing.expectEqual(@as(?i64, 42), store.load(io)); +} diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index b51e88e..7ab1eb9 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -7,7 +7,9 @@ const std = @import("std"); const zat = @import("zat"); const convert = @import("convert.zig"); +const cursor_mod = @import("cursor.zig"); const tail_mod = @import("tail.zig"); +const verify = @import("verify.zig"); const Io = std.Io; const Allocator = std.mem.Allocator; @@ -21,6 +23,8 @@ pub const Consumer = struct { upstream: []const u8, cursor: ?i64 = null, to_stdout: bool = false, + verifier: ?*verify.Verifier = null, + cursor_store: ?*cursor_mod.Store = null, drops: convert.Drops = .initFill(0), /// blocks forever; zat handles reconnect with backoff and host rotation. @@ -45,6 +49,23 @@ pub const Consumer = struct { const alloc = arena.allocator(); const time_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); + + if (self.verifier) |v| { + switch (event) { + .commit => |c| switch (v.verifyCommit(c.repo, c.blocks)) { + // proven-bad after a fresh key resolve: drop the event + .invalid_signature => { + self.drops.getPtr(.invalid_signature).* += 1; + log.warn("dropped commit with invalid signature: {s} seq={d}", .{ c.repo, c.seq }); + return self.noteSeq(event, time_us); + }, + .valid, .unverified => {}, + }, + // rotation signal: next commit re-resolves the key + .identity => |i| v.evict(i.did), + else => {}, + } + } var frames: std.ArrayList(convert.Frame) = .empty; const drops_before = self.drops; try convert.convert(alloc, event, time_us, &frames, &self.drops); @@ -54,6 +75,8 @@ pub const Consumer = struct { try self.tail.append(frame.kind, frame.did, frame.collection, frame.time_us, frame.json); } + self.noteSeq(event, time_us); + if (self.to_stdout and frames.items.len > 0) { var stdout_buf: [64 * 1024]u8 = undefined; var stdout = std.Io.File.stdout().writer(self.io, &stdout_buf); @@ -63,6 +86,16 @@ pub const Consumer = struct { stdout.interface.flush() catch return error.WriteFailed; } } + + fn noteSeq(self: *Consumer, event: zat.FirehoseEvent, time_us: i64) void { + if (self.cursor_store) |cs| { + if (event.seq()) |s| cs.update(self.io, s, time_us); + } + } + + pub fn close(_: *Consumer) void { + log.info("upstream closed connection", .{}); + } }; fn logNewDrops(before: *const convert.Drops, after: *const convert.Drops) void { diff --git a/src/internal/verify.zig b/src/internal/verify.zig new file mode 100644 index 0000000..15aaa9a --- /dev/null +++ b/src/internal/verify.zig @@ -0,0 +1,165 @@ +//! commit signature verification — DID signing-key cache + zat verify +//! +//! upstream jetstream (via atmos) runs full Sync 1.1: signature check with +//! DID resolution, MST-diff verification against a per-DID rev chain, and +//! getRepo resync on failure, on a per-DID worker pool. this is the first +//! slice of that: synchronous signature verification with a key cache. +//! +//! semantics (following zlay's validator where we deviate from upstream): +//! - cache miss → resolve synchronously (the simulator PLC is localhost; +//! a per-DID worker pool is a later throughput task) +//! - resolution failure → pass unverified and count (availability over +//! strictness while we have no resync path) +//! - signature mismatch → evict key, re-resolve once (rotation), retry; +//! still bad → the event is dropped by the caller +//! - #identity events evict the cached key (rotation signal) +//! +//! TODO (upstream parity): MST-diff verification (zat.verifyCommitDiff), +//! per-DID rev chain tracking, getRepo resync, worker-pool parallelism. + +const std = @import("std"); +const zat = @import("zat"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; +const log = std.log.scoped(.stream); + +const CachedKey = struct { + key_type: zat.multicodec.KeyType, + raw: [33]u8, + len: u8, +}; + +pub const Result = enum { + valid, + /// signature failed against a freshly resolved key — drop the event + invalid_signature, + /// could not resolve the DID — passed through unverified + unverified, +}; + +pub const Stats = struct { + verified: u64 = 0, + invalid: u64 = 0, + unverified: u64 = 0, + cache_hits: u64 = 0, + cache_misses: u64 = 0, +}; + +pub const Verifier = struct { + allocator: Allocator, + io: Io, + resolver: zat.DidResolver, + cache: std.StringHashMapUnmanaged(CachedKey) = .empty, + max_cache: u32 = 100_000, + stats: Stats = .{}, + + /// plc_url: base URL of the PLC directory. for the simulator this is its + /// http address (it serves GET /did:...); production is plc.directory. + pub fn init(allocator: Allocator, io: Io, plc_url: []const u8) Verifier { + return .{ + .allocator = allocator, + .io = io, + .resolver = resolver: { + var r = zat.DidResolver.init(io, allocator); + r.plc_url = plc_url; + break :resolver r; + }, + }; + } + + pub fn deinit(self: *Verifier) void { + var it = self.cache.keyIterator(); + while (it.next()) |k| self.allocator.free(k.*); + self.cache.deinit(self.allocator); + self.resolver.deinit(); + } + + /// verify the commit signature in `blocks` (the event's CAR diff) + /// against `did`'s signing key. + pub fn verifyCommit(self: *Verifier, did: []const u8, blocks: []const u8) Result { + const key = self.getKey(did, false) orelse { + self.stats.unverified += 1; + return .unverified; + }; + if (self.checkSignature(did, blocks, key)) { + self.stats.verified += 1; + return .valid; + } + // key may have rotated: re-resolve once and retry + const fresh = self.getKey(did, true) orelse { + self.stats.unverified += 1; + return .unverified; + }; + if (self.checkSignature(did, blocks, fresh)) { + self.stats.verified += 1; + return .valid; + } + self.stats.invalid += 1; + return .invalid_signature; + } + + /// rotation signal: #identity events invalidate the cached key + pub fn evict(self: *Verifier, did: []const u8) void { + if (self.cache.fetchRemove(did)) |kv| self.allocator.free(kv.key); + } + + fn checkSignature(self: *Verifier, did: []const u8, blocks: []const u8, key: CachedKey) bool { + const public_key = zat.multicodec.PublicKey{ + .key_type = key.key_type, + .raw = key.raw[0..key.len], + }; + var arena = std.heap.ArenaAllocator.init(self.allocator); + defer arena.deinit(); + _ = zat.verifyCommitCar(arena.allocator(), blocks, public_key, .{ + .verify_mst = false, + .expected_did = did, + .max_car_size = 5 * 1024 * 1024, + }) catch return false; + return true; + } + + fn getKey(self: *Verifier, did: []const u8, force_resolve: bool) ?CachedKey { + if (!force_resolve) { + if (self.cache.get(did)) |k| { + self.stats.cache_hits += 1; + return k; + } + self.stats.cache_misses += 1; + } else { + self.evict(did); + } + + const parsed = zat.Did.parse(did) orelse return null; + var doc = self.resolver.resolve(parsed) catch |err| { + log.debug("DID resolve failed for {s}: {s}", .{ did, @errorName(err) }); + return null; + }; + defer doc.deinit(); + + const vm = doc.signingKey() orelse return null; + const key_bytes = zat.multibase.decode(self.allocator, vm.public_key_multibase) catch return null; + defer self.allocator.free(key_bytes); + const public_key = zat.multicodec.parsePublicKey(key_bytes) catch return null; + if (public_key.raw.len > 33) return null; + + var cached = CachedKey{ + .key_type = public_key.key_type, + .raw = undefined, + .len = @intCast(public_key.raw.len), + }; + @memcpy(cached.raw[0..public_key.raw.len], public_key.raw); + + // crude bound: reset when full (TODO: LRU, like zlay's) + if (self.cache.count() >= self.max_cache) { + var it = self.cache.keyIterator(); + while (it.next()) |k| self.allocator.free(k.*); + self.cache.clearRetainingCapacity(); + } + const owned = self.allocator.dupe(u8, did) catch return cached; + self.cache.put(self.allocator, owned, cached) catch { + self.allocator.free(owned); + }; + return cached; + } +}; diff --git a/src/main.zig b/src/main.zig index 44837b2..2d30455 100644 --- a/src/main.zig +++ b/src/main.zig @@ -2,7 +2,9 @@ const std = @import("std"); const websocket = @import("websocket"); +const cursor_mod = @import("internal/cursor.zig"); const ingest = @import("internal/ingest.zig"); +const verify = @import("internal/verify.zig"); const server_mod = @import("internal/server.zig"); const tail_mod = @import("internal/tail.zig"); @@ -19,9 +21,14 @@ pub fn main(init: std.process.Init.Minimal) !void { const io = threaded.io(); var upstream: []const u8 = "ws://localhost:7777"; + // the simulator serves DID docs itself; production would pass + // --plc=https://plc.directory + var plc_url: []const u8 = "http://localhost:7777"; + var data_dir: []const u8 = "./data"; var cursor: ?i64 = null; var port: u16 = 6008; var to_stdout = false; + var do_verify = true; var arg_it = init.args.iterate(); _ = arg_it.next(); // program name @@ -32,6 +39,12 @@ pub fn main(init: std.process.Init.Minimal) !void { cursor = try std.fmt.parseInt(i64, arg["--cursor=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--port=")) { port = try std.fmt.parseInt(u16, arg["--port=".len..], 10); + } else if (std.mem.startsWith(u8, arg, "--plc=")) { + plc_url = arg["--plc=".len..]; + } else if (std.mem.startsWith(u8, arg, "--data-dir=")) { + data_dir = arg["--data-dir=".len..]; + } else if (std.mem.eql(u8, arg, "--no-verify")) { + do_verify = false; } else if (std.mem.eql(u8, arg, "--stdout")) { to_stdout = true; } else { @@ -59,6 +72,17 @@ pub fn main(init: std.process.Init.Minimal) !void { defer _ = server_future.cancel(io); log.info("serving /subscribe on :{d}", .{port}); + var cursor_store = try cursor_mod.Store.init(io, data_dir); + defer cursor_store.deinit(io); + if (cursor == null) { + cursor = cursor_store.load(io); + if (cursor) |c| log.info("resuming from persisted cursor {d}", .{c}); + } + + var verifier: ?verify.Verifier = if (do_verify) verify.Verifier.init(allocator, io, plc_url) else null; + defer if (verifier) |*v| v.deinit(); + if (do_verify) log.info("signature verification on (plc: {s})", .{plc_url}); + var consumer: ingest.Consumer = .{ .allocator = allocator, .io = io, @@ -66,6 +90,8 @@ pub fn main(init: std.process.Init.Minimal) !void { .upstream = upstream, .cursor = cursor, .to_stdout = to_stdout, + .verifier = if (verifier) |*v| v else null, + .cursor_store = &cursor_store, }; try consumer.run(); } diff --git a/src/tests.zig b/src/tests.zig index da04596..c066290 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -1,6 +1,8 @@ comptime { _ = @import("main.zig"); _ = @import("internal/ingest.zig"); + _ = @import("internal/cursor.zig"); + _ = @import("internal/verify.zig"); _ = @import("internal/wire.zig"); _ = @import("internal/convert.zig"); _ = @import("internal/filter.zig"); -- 2.51.2