Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894//! Sync 1.1 verification — signatures, repository-chain state, and repair//! authentication. Live commits are checked as signed MST diffs. Full-repo//! repair heads use the separate unbounded complete-CAR path; the asynchronous//! repair coordinator consumes the explicit `needs_resync` outcome.//!//! semantics (upstream-faithful: no event is served or archived//! unverified when the DID resolves — stream's consumers inherit its//! diligence, and verification blocks rather than passing through)://! - cache miss → resolve synchronously; parallelism comes from the//! ingest pipeline's verify workers, not from skipping work//! - resolution failure → reject and count. Until transparent resync is//! implemented, availability must not turn into unverified archive data.//! - 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)//!const std = @import("std");const zat = @import("zat");const lru = @import("../storage/lru.zig");const metrics = @import("../runtime/metrics.zig");const meta_store = @import("../storage/meta_store.zig");const repos = @import("backfill/repos.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);const max_commit_blocks_bytes = 2_000_000;const max_firehose_message_bytes = 2 * 1024 * 1024;
const CachedKey = struct { key_type: zat.multicodec.KeyType, raw: [33]u8, len: u8,};
pub const Outcome = enum { valid, /// event rev is at or behind the durable/provisional repo head replay, /// signature failed against a freshly resolved key — drop the event invalid_signature, /// could not resolve the DID — reject; authoritative repair cannot be /// authenticated without the same identity material unverified, /// input-integrity failure which an authoritative fetch cannot repair: /// future rev, size violation, or outer/inner field mismatch invalid_commit, /// Chain state cannot be advanced from this event alone. Upstream /// performs authoritative getRepo here; Stream's coordinator consumes /// this outcome and withholds the triggering event from the archive. needs_resync,};
pub const ChainUpdate = struct { rev: [13]u8, data: [128]u8 = undefined, data_len: u8,
pub fn dataSlice(self: *const ChainUpdate) []const u8 { return self.data[0..self.data_len]; }};
pub const Result = struct { outcome: Outcome = .valid, /// Exact signed post-state proven by this commit. The writer stages it /// against the event's local archive boundary for durable checkpointing. chain: ?ChainUpdate = null, hosting: ?HostingUpdate = null,};
const CommitFailureClass = enum { signature, invalid, resync, fatal };
/// Match Atmos's default PolicyResync boundary. Failures in decoding,/// inversion, duplicate paths, and op-CID proof are repairable by replacing/// local state from authoritative getRepo. Envelope/inner disagreement is a/// producer-integrity failure and bypasses repair; allocation failure must/// remain fatal rather than being disguised as a network repair.fn classifyCommitFailure(err: anyerror) CommitFailureClass { return switch (err) { error.SignatureVerificationFailed => .signature, error.FieldMismatch => .invalid, error.OutOfMemory => .fatal, else => .resync, };}
pub const HostingUpdate = struct { active: bool, seq: i64, status: [32]u8 = undefined, status_len: u8 = 0, time: [64]u8 = undefined, time_len: u8 = 0,
pub fn statusSlice(self: *const HostingUpdate) []const u8 { return self.status[0..self.status_len]; }
pub fn timeSlice(self: *const HostingUpdate) []const u8 { return self.time[0..self.time_len]; }};
pub fn stageChain(writes: *meta_store.Batch, did: []const u8, state: ChainUpdate) !void { var key_buf: [512]u8 = undefined; const key = try chainKey(&key_buf, did); var value: [1 + 13 + 1 + 128]u8 = undefined; value[0] = 1; @memcpy(value[1..14], &state.rev); value[14] = state.data_len; @memcpy(value[15..][0..state.data_len], state.dataSlice()); writes.put(key, value[0 .. 15 + state.data_len]);}
pub fn stageHosting(writes: *meta_store.Batch, did: []const u8, state: HostingUpdate) !void { var key_buf: [512]u8 = undefined; const key = try hostingKey(&key_buf, did); var value: [1 + 1 + 8 + 1 + 32 + 1 + 64]u8 = undefined; value[0] = 1; value[1] = @intFromBool(state.active); std.mem.writeInt(i64, value[2..10], state.seq, .little); value[10] = state.status_len; @memcpy(value[11..][0..state.status_len], state.statusSlice()); const time_at = 11 + state.status_len; value[time_at] = state.time_len; @memcpy(value[time_at + 1 ..][0..state.time_len], state.timeSlice()); writes.put(key, value[0 .. time_at + 1 + state.time_len]);}
fn chainKey(buf: []u8, did: []const u8) ![]const u8 { return std.fmt.bufPrint(buf, "verify/chain/{s}", .{did});}
fn hostingKey(buf: []u8, did: []const u8) ![]const u8 { return std.fmt.bufPrint(buf, "verify/host/{s}", .{did});}
fn decodeChain(value: []const u8) !ChainUpdate { if (value.len < 15 or value[0] != 1) return error.InvalidVerifierState; const data_len: usize = value[14]; if (data_len > 128 or value.len != 15 + data_len) return error.InvalidVerifierState; var state: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(data_len) }; @memcpy(&state.rev, value[1..14]); @memcpy(state.data[0..data_len], value[15..]); return state;}
fn decodeHosting(value: []const u8) !HostingUpdate { if (value.len < 12 or value[0] != 1 or value[1] > 1) return error.InvalidVerifierState; const status_len: usize = value[10]; if (status_len > 32 or 11 + status_len >= value.len) return error.InvalidVerifierState; const time_at = 11 + status_len; const time_len: usize = value[time_at]; if (time_len > 64 or value.len != time_at + 1 + time_len) return error.InvalidVerifierState; var state: HostingUpdate = .{ .active = value[1] == 1, .seq = std.mem.readInt(i64, value[2..10], .little), .status_len = @intCast(status_len), .time_len = @intCast(time_len), }; @memcpy(state.status[0..status_len], value[11..time_at]); @memcpy(state.time[0..time_len], value[time_at + 1 ..]); return state;}
pub const Verifier = struct { allocator: Allocator, io: Io, /// thread-safe LRU (internal mutex); resolution I/O runs on the /// CALLER's resolver — each verify worker owns one, so misses resolve /// in parallel (duplicate concurrent resolves of one DID are rare and /// harmless). LRU eviction replaces the old clear-when-full reset, /// which would have caused a resolve stampede at real-network scale. cache: lru.LruCache(CachedKey), stats: ?*metrics.Stats = null, meta: *meta_store.Store, chain_mu: Io.Mutex = .init, /// Provisional verified heads. Durability is staged by the ordered /// writer only after corresponding archive rows are assigned. chains: std.StringHashMapUnmanaged(ChainUpdate) = .empty, hosting: std.StringHashMapUnmanaged(HostingUpdate) = .empty,
/// plc_url: base URL of the PLC directory. for the simulator this is its /// http address (it serves GET /did:...); production is plc.directory. plc_url: []const u8,
pub fn init(allocator: Allocator, io: Io, plc_url: []const u8, meta: *meta_store.Store) Verifier { return .{ .allocator = allocator, .io = io, .plc_url = plc_url, .meta = meta, .cache = lru.LruCache(CachedKey).init(allocator, 250_000, io), }; }
/// one per verify worker; owns a keep-alive connection pub fn makeResolver(self: *Verifier) zat.DidResolver { var r = zat.DidResolver.init(self.io, self.allocator); r.plc_url = self.plc_url; return r; }
pub fn deinit(self: *Verifier) void { self.cache.deinit(); var it = self.chains.iterator(); while (it.next()) |entry| self.allocator.free(entry.key_ptr.*); self.chains.deinit(self.allocator); var hosting_it = self.hosting.iterator(); while (hosting_it.next()) |entry| self.allocator.free(entry.key_ptr.*); self.hosting.deinit(self.allocator); }
/// verify the commit signature in `blocks` (the event's CAR diff) /// against `did`'s signing key. pub fn verifyCommit(self: *Verifier, resolver: *zat.DidResolver, commit: zat.firehose.CommitEvent) !Result { // Upstream bypasses resync policy for these input-integrity gates. const rev = zat.Tid.parse(commit.rev) orelse return self.rejectInvalid(); const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); if (rev.timestamp() > @as(u64, @intCast(@max(now_us, 0))) + 5 * 60 * std.time.us_per_s) return self.rejectInvalid(); if (commit.blocks.len > max_commit_blocks_bytes or commit.ops.len > 200) return self.rejectInvalid();
// key lookup/resolution under the lock; signature crypto outside it // so verify workers actually run in parallel const prior = try self.chainFor(commit.repo); var legacy = false; if (prior) |state| { if (std.mem.order(u8, commit.rev, &state.rev) != .gt) return .{ .outcome = .replay }; if (commit.prev_data) |claimed| { if (!std.mem.eql(u8, claimed.raw, state.dataSlice())) return self.needsResync(); } else if (isLegacyCommit(commit)) { // Atmos defaults to LegacyAccept: Sync-1.0-shaped commits // cannot prove continuity, but still pass signature and // post-state op-CID verification and advance the chain. legacy = true; } else return self.needsResync(); }
const key = self.lockedGetKey(resolver, commit.repo, false) orelse { self.count("verify_unverified"); return .{ .outcome = .unverified }; }; if (checkCommit(self.allocator, commit, key, if (legacy) null else prior)) |chain| { self.count("verify_valid"); try self.rememberChain(commit.repo, chain); return .{ .chain = chain }; } else |err| switch (classifyCommitFailure(err)) { .signature => {}, .invalid => return self.rejectInvalid(), .resync => return self.needsResync(), .fatal => return err, } // key may have rotated: re-resolve once and retry const fresh = self.lockedGetKey(resolver, commit.repo, true) orelse { self.count("verify_unverified"); return .{ .outcome = .unverified }; }; if (checkCommit(self.allocator, commit, fresh, if (legacy) null else prior)) |chain| { self.count("verify_valid"); try self.rememberChain(commit.repo, chain); return .{ .chain = chain }; } else |err| switch (classifyCommitFailure(err)) { .signature => {}, .invalid => return self.rejectInvalid(), .resync => return self.needsResync(), .fatal => return err, } self.count("verify_invalid"); return .{ .outcome = .invalid_signature }; }
pub fn verifySync(self: *Verifier, resolver: *zat.DidResolver, sync_event: zat.firehose.SyncEvent) !Result { // Atmos deliberately leaves an unparseable #sync rev to the ingest // gate. A divergent sync must first repair chain state from getRepo; // only then does the gate reject the entire original directive. if (zat.Tid.parse(sync_event.rev)) |rev| { const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); if (rev.timestamp() > @as(u64, @intCast(@max(now_us, 0))) + 5 * 60 * std.time.us_per_s) return self.rejectInvalid(); }
const prior = try self.chainFor(sync_event.did) orelse return .{ .outcome = .needs_resync }; if (std.mem.order(u8, sync_event.rev, &prior.rev) != .gt) return .{ .outcome = .replay }; if (sync_event.blocks.len == 0) return .{ .outcome = .needs_resync };
var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const loaded = zat.loadCommitFromCAR(arena.allocator(), sync_event.blocks) catch return .{ .outcome = .needs_resync }; if (!std.mem.eql(u8, loaded.commit.data_cid, prior.dataSlice())) return .{ .outcome = .needs_resync }; // Atmos checks envelope/inner identity only on its no-op fast path. // A divergent data root is repaired authoritatively first; the // downstream ingest gate then owns malformed envelope fields. if (!std.mem.eql(u8, loaded.commit.did, sync_event.did) or !std.mem.eql(u8, loaded.commit.rev, sync_event.rev)) return self.rejectInvalid();
const key = self.lockedGetKey(resolver, sync_event.did, false) orelse return .{ .outcome = .unverified }; if (!checkSyncSignature(arena.allocator(), sync_event, key)) { const fresh = self.lockedGetKey(resolver, sync_event.did, true) orelse return .{ .outcome = .unverified }; if (!checkSyncSignature(arena.allocator(), sync_event, fresh)) { self.count("verify_invalid"); return .{ .outcome = .invalid_signature }; } }
var update = prior; @memcpy(&update.rev, sync_event.rev); try self.rememberChain(sync_event.did, update); self.count("verify_valid"); return .{ .chain = update }; }
/// Verify the authoritative full-repo commit returned by getRepo. This is /// deliberately separate from bootstrap: ordinary network backfill /// mirrors upstream and does not seed verifier state, while a transparent /// repair MUST authenticate the fetched head before it can replace the /// chain. The returned update is provisional only; the repair coordinator /// applies it under the per-DID FSM and the archive writer promotes it at /// the matching durable row boundary. pub fn verifyFetchedRepo( self: *Verifier, resolver: *zat.DidResolver, did: []const u8, prepared: *const repos.PreparedRepo, ) !Result { const key = self.lockedGetKey(resolver, did, false) orelse { self.count("verify_unverified"); return .{ .outcome = .unverified }; }; const update = checkFetchedRepo(did, prepared, key) catch |err| switch (err) { error.SignatureVerificationFailed => retry: { const fresh = self.lockedGetKey(resolver, did, true) orelse { self.count("verify_unverified"); return .{ .outcome = .unverified }; }; break :retry checkFetchedRepo(did, prepared, fresh) catch |fresh_err| { if (fresh_err == error.SignatureVerificationFailed) { self.count("verify_invalid"); return .{ .outcome = .invalid_signature }; } return self.rejectInvalid(); }; }, else => return self.rejectInvalid(), }; self.count("verify_valid"); return .{ .chain = update }; }
pub fn onAccount(self: *Verifier, account: zat.firehose.AccountEvent) !Result { if (zat.Did.parse(account.did) == null) return self.rejectInvalid(); if (try self.hostingFor(account.did)) |prior| { if (account.seq <= prior.seq) return .{ .outcome = .replay }; } const status = if (account.status) |value| @tagName(value) else ""; if (status.len > 32 or account.time.len > 64) return self.rejectInvalid(); var update: HostingUpdate = .{ .active = account.active, .seq = account.seq, .status_len = @intCast(status.len), .time_len = @intCast(account.time.len), }; @memcpy(update.status[0..status.len], status); @memcpy(update.time[0..account.time.len], account.time); try self.rememberHosting(account.did, update); return .{ .hosting = update }; }
fn checkSyncSignature(allocator: Allocator, sync_event: zat.firehose.SyncEvent, key: CachedKey) bool { const public_key = zat.multicodec.PublicKey{ .key_type = key.key_type, .raw = key.raw[0..key.len], }; const result = zat.verifyCommitCar(allocator, sync_event.blocks, public_key, .{ .verify_mst = false, .expected_did = sync_event.did, .max_car_size = max_firehose_message_bytes, }) catch return false; return std.mem.eql(u8, result.commit_rev, sync_event.rev); }
fn rejectInvalid(self: *Verifier) Result { self.count("verify_invalid"); return .{ .outcome = .invalid_commit }; }
fn needsResync(self: *Verifier) Result { // This is a verification failure, even though the coordinator can // repair it asynchronously. Keep the failure visible to operators. self.count("verify_invalid"); return .{ .outcome = .needs_resync }; }
fn lockedGetKey(self: *Verifier, resolver: *zat.DidResolver, did: []const u8, force_resolve: bool) ?CachedKey { if (!force_resolve) { if (self.cache.get(did)) |k| { self.count("verify_cache_hits"); return k; } self.count("verify_cache_misses"); } else { self.evict(did); } return self.resolveAndCache(resolver, did); }
/// resolve on the caller's own resolver, outside the cache lock fn resolveAndCache(self: *Verifier, resolver: *zat.DidResolver, did: []const u8) ?CachedKey { // did:web fetches attacker-named hosts, but zat's resolver routes // those through network_safety (private/loopback rejection, DoH // preflight, no redirects, checked-connection dial) — lesson #10 // is handled at the SDK layer const doc_or_null: ?zat.DidDocument = doc: { const parsed = zat.Did.parse(did) orelse break :doc null; break :doc resolver.resolve(parsed) catch |err| { log.debug("DID resolve failed for {s}: {s}", .{ did, @errorName(err) }); break :doc null; }; }; var doc = doc_or_null orelse 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);
self.cache.put(did, cached) catch {}; return cached; }
fn count(self: *Verifier, comptime field: []const u8) void { if (self.stats) |st| _ = @field(st, field).fetchAdd(1, .monotonic); }
/// rotation signal: #identity events invalidate the cached key pub fn evict(self: *Verifier, did: []const u8) void { _ = self.cache.remove(did); }
fn checkCommit(allocator: Allocator, commit: zat.firehose.CommitEvent, key: CachedKey, prior: ?ChainUpdate) !ChainUpdate { const public_key = zat.multicodec.PublicKey{ .key_type = key.key_type, .raw = key.raw[0..key.len], }; var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const ops = try commit.toMstOperations(arena.allocator()); const result = zat.verifyCommitDiff( arena.allocator(), commit.blocks, ops, if (prior) |*state| state.dataSlice() else null, public_key, .{ .expected_did = null, .max_car_size = max_commit_blocks_bytes, }, ) catch |err| switch (err) { // Match upstream's default LenientInversion carve-out narrowly: // the envelope already proved prevData == our stored head, and // inversion completed but produced a different root. Signature, // outer/inner fields, and post-state op CIDs remain mandatory. error.PrevDataMismatch => if (prior != null) try zat.verifyCommitDiff( arena.allocator(), commit.blocks, ops, prior.?.dataSlice(), public_key, .{ .expected_did = null, .skip_inversion = true, .max_car_size = max_commit_blocks_bytes, }, ) else return err, else => return err, }; // The signed inner commit and relay envelope must identify the same // revision and CAR root. A valid signature over relabeled outer // fields is still an invalid firehose event. if (!std.mem.eql(u8, result.commit_did, commit.repo) or !std.mem.eql(u8, result.commit_rev, commit.rev)) return error.FieldMismatch; const outer_cid = commit.commit orelse return error.FieldMismatch; if (!std.mem.eql(u8, result.commit_cid, outer_cid.raw)) return error.FieldMismatch; try checkOpCids(arena.allocator(), commit, result.data_cid); if (result.data_cid.len > 128) return error.FieldMismatch; var update: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(result.data_cid.len), }; @memcpy(&update.rev, result.commit_rev); @memcpy(update.data[0..result.data_cid.len], result.data_cid); return update; }
fn checkFetchedRepo( did: []const u8, prepared: *const repos.PreparedRepo, key: CachedKey, ) !ChainUpdate { const public_key = zat.multicodec.PublicKey{ .key_type = key.key_type, .raw = key.raw[0..key.len], }; // prepareRepo already indexed the CAR and proved MST/record // completeness. Authenticate those retained commit bytes directly; // reparsing and re-walking multi-gigabyte repositories here would be // both semantically redundant and operationally unacceptable. try prepared.authenticate(did, public_key); if (prepared.rev.len != 13 or zat.Tid.parse(prepared.rev) == null) return error.InvalidCommit; if (prepared.data_cid.len > 128) return error.InvalidCommit; var update: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(prepared.data_cid.len), }; @memcpy(&update.rev, prepared.rev); @memcpy(update.data[0..prepared.data_cid.len], prepared.data_cid); return update; }
fn checkOpCids(allocator: Allocator, commit: zat.firehose.CommitEvent, data_cid: []const u8) !void { if (commit.ops.len == 0) return; const loaded = try zat.loadCommitFromCAR(allocator, commit.blocks); var tree = zat.mst.Mst.loadFromBlocks(allocator, loaded.repo_car, data_cid) catch return error.OpCidMismatch; for (commit.ops) |op| { const path = if (op.path.len > 0) op.path else try std.fmt.allocPrint(allocator, "{s}/{s}", .{ op.collection, op.rkey }); const found = tree.getLazy(path) catch return error.OpCidMismatch; switch (op.action) { .create, .update => { const claimed = op.cid orelse return error.OpCidMismatch; const actual = found orelse return error.OpCidMismatch; if (!std.mem.eql(u8, claimed.raw, actual.raw)) return error.OpCidMismatch; }, .delete => if (found != null) return error.OpCidMismatch, } } }
pub fn chainFor(self: *Verifier, did: []const u8) !?ChainUpdate { self.chain_mu.lockUncancelable(self.io); if (self.chains.get(did)) |state| { self.chain_mu.unlock(self.io); return state; } self.chain_mu.unlock(self.io);
var key_buf: [512]u8 = undefined; const key = try chainKey(&key_buf, did); const encoded = try self.meta.getAlloc(self.allocator, key) orelse return null; defer self.allocator.free(encoded); const state = try decodeChain(encoded); try self.rememberChain(did, state); return state; }
pub fn rememberChain(self: *Verifier, did: []const u8, state: ChainUpdate) !void { self.chain_mu.lockUncancelable(self.io); defer self.chain_mu.unlock(self.io); if (self.chains.getPtr(did)) |existing| { existing.* = state; return; } const owned = try self.allocator.dupe(u8, did); errdefer self.allocator.free(owned); try self.chains.put(self.allocator, owned, state); }
pub fn hostingFor(self: *Verifier, did: []const u8) !?HostingUpdate { self.chain_mu.lockUncancelable(self.io); if (self.hosting.get(did)) |state| { self.chain_mu.unlock(self.io); return state; } self.chain_mu.unlock(self.io); var key_buf: [512]u8 = undefined; const key = try hostingKey(&key_buf, did); const encoded = try self.meta.getAlloc(self.allocator, key) orelse return null; defer self.allocator.free(encoded); const state = try decodeHosting(encoded); try self.rememberHosting(did, state); return state; }
fn rememberHosting(self: *Verifier, did: []const u8, state: HostingUpdate) !void { self.chain_mu.lockUncancelable(self.io); defer self.chain_mu.unlock(self.io); if (self.hosting.getPtr(did)) |existing| { existing.* = state; return; } const owned = try self.allocator.dupe(u8, did); errdefer self.allocator.free(owned); try self.hosting.put(self.allocator, owned, state); }};
fn isLegacyCommit(commit: zat.firehose.CommitEvent) bool { if (commit.prev_data != null) return false; for (commit.ops) |op| switch (op.action) { .update, .delete => if (op.prev != null) return false, .create => {}, }; return true;}
test "durable verifier chain reloads from metadata store" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(threaded.io(), &path_buf);
var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var state: ChainUpdate = .{ .rev = undefined, .data_len = 4 }; @memcpy(&state.rev, "3k2abd0000000"); @memcpy(state.data[0..4], "data"); var writes = meta.batch(); stageChain(&writes, "did:plc:test", state) catch unreachable; var hosting: HostingUpdate = .{ .active = false, .seq = 42, .status_len = 8, .time_len = 4 }; @memcpy(hosting.status[0..8], "deactive"); @memcpy(hosting.time[0..4], "time"); try stageHosting(&writes, "did:plc:test", hosting); try writes.commitDurable(); writes.deinit();
var verifier = Verifier.init(testing.allocator, threaded.io(), "http://localhost", &meta); defer verifier.deinit(); const loaded = (try verifier.chainFor("did:plc:test")).?; try testing.expectEqualSlices(u8, &state.rev, &loaded.rev); try testing.expectEqualSlices(u8, state.dataSlice(), loaded.dataSlice()); const loaded_hosting = (try verifier.hostingFor("did:plc:test")).?; try testing.expect(!loaded_hosting.active); try testing.expectEqual(@as(i64, 42), loaded_hosting.seq); try testing.expectEqualStrings("deactive", loaded_hosting.statusSlice()); try testing.expectEqualStrings("time", loaded_hosting.timeSlice());}
test "op CID proof rejects an envelope that disagrees with post-state MST" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const allocator = arena.allocator();
const record_cid = try zat.cbor.Cid.forDagCbor(allocator, "record"); const wrong_cid = try zat.cbor.Cid.forDagCbor(allocator, "wrong"); var tree = zat.mst.Mst.init(allocator); try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid); const data_cid = try tree.rootCid(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(7)); const did = try keypair.did(allocator); const signed = try zat.signCommit(allocator, .{ .did = did, .rev = "3k2abcdefghij", .data = data_cid, }, &keypair); var blocks: std.ArrayList(zat.car.Block) = .empty; try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); try tree.collectBlocks(&blocks); const car_bytes = try zat.car.writeAlloc(allocator, .{ .roots = &.{signed.cid}, .blocks = blocks.items, }); const base: zat.firehose.CommitEvent = .{ .seq = 1, .repo = did, .rev = "3k2abcdefghij", .time = "2026-07-15T00:00:00Z", .commit = signed.cid, .blocks = car_bytes, .ops = &.{.{ .action = .create, .path = "app.bsky.feed.post/3k2abcdefghij", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .cid = record_cid, }}, }; try Verifier.checkOpCids(allocator, base, data_cid.raw); var tampered = base; tampered.ops = &.{.{ .action = .create, .path = "app.bsky.feed.post/3k2abcdefghij", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .cid = wrong_cid, }}; try testing.expectError(error.OpCidMismatch, Verifier.checkOpCids(allocator, tampered, data_cid.raw)); try testing.expectEqual(CommitFailureClass.resync, classifyCommitFailure(error.OpCidMismatch));}
test "commit block limit is upstream decimal two million bytes" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(threaded.io(), &path_buf); var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var verifier = Verifier.init(testing.allocator, threaded.io(), "http://offline.invalid", &meta); defer verifier.deinit(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(29)); const public_key = try keypair.publicKey(); var cached: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(public_key.len) }; @memcpy(cached.raw[0..public_key.len], &public_key); try verifier.cache.put("did:plc:size-boundary", cached); const blocks = try testing.allocator.alloc(u8, max_commit_blocks_bytes + 1); defer testing.allocator.free(blocks); @memset(blocks, 0); const base: zat.firehose.CommitEvent = .{ .seq = 1, .repo = "did:plc:size-boundary", .rev = "3k2abcdefghij", .time = "2026-07-16T00:00:00Z", .blocks = blocks[0..max_commit_blocks_bytes], .ops = &.{}, }; const at_limit = try verifier.verifyCommit(undefined, base); try testing.expectEqual(Outcome.needs_resync, at_limit.outcome); var over = base; over.blocks = blocks; const rejected = try verifier.verifyCommit(undefined, over); try testing.expectEqual(Outcome.invalid_commit, rejected.outcome);}
test "signed chain failures follow upstream repair policy and legacy default" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const allocator = arena.allocator();
var tree = zat.mst.Mst.init(allocator); const data_cid = try tree.rootCid(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(11)); const did = try keypair.did(allocator); const signed = try zat.signCommit(allocator, .{ .did = did, .rev = "3k2abcdefghij", .data = data_cid, }, &keypair); var blocks: std.ArrayList(zat.car.Block) = .empty; try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); try tree.collectBlocks(&blocks); const car_bytes = try zat.car.writeAlloc(allocator, .{ .roots = &.{signed.cid}, .blocks = blocks.items, }); const public_key = try keypair.publicKey(); var cached: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(public_key.len), }; @memcpy(cached.raw[0..public_key.len], &public_key);
const event: zat.firehose.CommitEvent = .{ .seq = 1, .repo = did, .rev = "3k2abcdefghij", .time = "2026-07-15T00:00:00Z", .commit = signed.cid, .blocks = car_bytes, .ops = &.{}, }; var wrong_prior: ChainUpdate = .{ .rev = undefined, .data_len = 4 }; @memcpy(&wrong_prior.rev, "3k2abc0000000"); @memcpy(wrong_prior.data[0..4], "nope");
// The real signed CAR cannot invert to the deliberately wrong prior. The // verifier's production caller permits this exact case only after the // envelope's prevData has matched the stored head (Atmos's default // lenient-inversion carve-out), so it remains authenticated and accepted. const lenient = try Verifier.checkCommit(allocator, event, cached, wrong_prior); try testing.expectEqualSlices(u8, data_cid.raw, lenient.dataSlice()); // Without that caller-side prevData proof, the same mismatch is the // recoverable failure class consumed by the repair coordinator. try testing.expectEqual(CommitFailureClass.resync, classifyCommitFailure(error.PrevDataMismatch));
// With no prevData, Atmos's default LegacyAccept still authenticates the // exact same signed CAR and its post-state before advancing. const legacy = try Verifier.checkCommit(allocator, event, cached, null); try testing.expectEqualStrings(event.rev, &legacy.rev); try testing.expectEqualSlices(u8, data_cid.raw, legacy.dataSlice()); try testing.expectEqual(CommitFailureClass.invalid, classifyCommitFailure(error.FieldMismatch));
// Exercise the public verifier boundary too: a syntactically valid event // whose prevData disagrees with the stored chain is queued for repair // before resolution or signature work, never mislabeled permanently bad. var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(threaded.io(), &path_buf); var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var verifier = Verifier.init(testing.allocator, threaded.io(), "http://offline.invalid", &meta); defer verifier.deinit(); var current: ChainUpdate = .{ .rev = undefined, .data_len = @intCast(data_cid.raw.len) }; @memcpy(¤t.rev, "3k2abc0000000"); @memcpy(current.data[0..data_cid.raw.len], data_cid.raw); try verifier.rememberChain(did, current); const wrong_prev = try zat.cbor.Cid.forDagCbor(allocator, "wrong prior"); var broken = event; broken.prev_data = wrong_prev; const routed = try verifier.verifyCommit(undefined, broken); try testing.expectEqual(Outcome.needs_resync, routed.outcome);}
test "full-repo repair authenticates the signed complete repository head" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const allocator = arena.allocator();
const record = "signed repair record"; const record_cid = try zat.cbor.Cid.forDagCbor(allocator, record); var tree = zat.mst.Mst.init(allocator); try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid); const data_cid = try tree.rootCid(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(9)); const did = try keypair.did(allocator); const signed = try zat.signCommit(allocator, .{ .did = did, .rev = "3k2abcdefghij", .data = data_cid, }, &keypair); var blocks: std.ArrayList(zat.car.Block) = .empty; try blocks.append(allocator, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); try tree.collectBlocks(&blocks); const incomplete_car = try zat.car.writeAlloc(allocator, .{ .roots = &.{signed.cid}, .blocks = blocks.items, }); try blocks.append(allocator, .{ .cid_raw = record_cid.raw, .data = record }); const car_bytes = try zat.car.writeAlloc(allocator, .{ .roots = &.{signed.cid}, .blocks = blocks.items, }); const public_key = try keypair.publicKey(); var cached: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(public_key.len), }; @memcpy(cached.raw[0..public_key.len], &public_key);
var prepared = try repos.prepareRepo(allocator, car_bytes); const update = try Verifier.checkFetchedRepo(did, &prepared, cached); try testing.expectEqualStrings("3k2abcdefghij", &update.rev); try testing.expectEqualSlices(u8, data_cid.raw, update.dataSlice());
try testing.expectError( error.MissingBlock, repos.prepareRepo(allocator, incomplete_car), );
const wrong_keypair = try zat.Keypair.fromSecretKey(.p256, @splat(7)); const wrong_public_key = try wrong_keypair.publicKey(); var wrong: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(wrong_public_key.len) }; @memcpy(wrong.raw[0..wrong_public_key.len], &wrong_public_key); try testing.expectError(error.SignatureVerificationFailed, Verifier.checkFetchedRepo(did, &prepared, wrong)); try testing.expectError(error.InvalidCommit, Verifier.checkFetchedRepo("did:plc:wrongrepo", &prepared, cached));}