diff --git a/.changeset/auth-materialization-bridge.md b/.changeset/auth-materialization-bridge.md new file mode 100644 index 00000000..bc9f2471 --- /dev/null +++ b/.changeset/auth-materialization-bridge.md @@ -0,0 +1,9 @@ +--- +'@treecrdt/interface': minor +'@treecrdt/sync-protocol': minor +'@treecrdt/sync-sqlite': minor +'@treecrdt/auth': minor +'@treecrdt/wa-sqlite': minor +--- + +Surface verified auth signer and authored-at claims on materialization events for remotely applied ops. diff --git a/.changeset/local-auth-signer.md b/.changeset/local-auth-signer.md new file mode 100644 index 00000000..4145b67b --- /dev/null +++ b/.changeset/local-auth-signer.md @@ -0,0 +1,6 @@ +--- +'@treecrdt/interface': minor +'@treecrdt/wa-sqlite': minor +--- + +Allow local write auth sessions to annotate materialization events with signer public keys. diff --git a/docs/sync/v0/messages.proto b/docs/sync/v0/messages.proto index de06f3de..301518e5 100644 --- a/docs/sync/v0/messages.proto +++ b/docs/sync/v0/messages.proto @@ -118,6 +118,15 @@ message OpAuth { // Optional reference to an authorization proof (e.g. token id / hash). bytes proof_ref = 3; + + // Optional signed claims about the operation. + OpAuthClaims claims = 4; +} + +// Signed claims carried alongside an operation signature. +message OpAuthClaims { + // Author's claimed wall-clock authoring time in unix milliseconds. + optional uint64 authored_at_ms = 1; } // Push-based subscription for live updates. diff --git a/packages/sync-protocol/material/postgres/src/proof-material/postgres.ts b/packages/sync-protocol/material/postgres/src/proof-material/postgres.ts index 06945c4d..2874a794 100644 --- a/packages/sync-protocol/material/postgres/src/proof-material/postgres.ts +++ b/packages/sync-protocol/material/postgres/src/proof-material/postgres.ts @@ -40,6 +40,7 @@ CREATE TABLE IF NOT EXISTS treecrdt_sync_op_auth ( op_ref BYTEA NOT NULL, sig BYTEA NOT NULL, proof_ref BYTEA, + claims_json TEXT, created_at_ms BIGINT NOT NULL, PRIMARY KEY (doc_id, op_ref) ); @@ -62,6 +63,7 @@ CREATE TABLE IF NOT EXISTS treecrdt_sync_pending_ops ( op BYTEA NOT NULL, sig BYTEA NOT NULL, proof_ref BYTEA, + claims_json TEXT, reason TEXT NOT NULL, message TEXT, created_at_ms BIGINT NOT NULL, @@ -69,10 +71,28 @@ CREATE TABLE IF NOT EXISTS treecrdt_sync_pending_ops ( ); `; +const OP_AUTH_MIGRATIONS_SQL = ` +ALTER TABLE treecrdt_sync_op_auth ADD COLUMN IF NOT EXISTS claims_json TEXT; +`; + +const PENDING_MIGRATIONS_SQL = ` +ALTER TABLE treecrdt_sync_pending_ops ADD COLUMN IF NOT EXISTS claims_json TEXT; +`; + function toBytes(value: Uint8Array | Buffer): Uint8Array { return Uint8Array.from(value); } +function encodeOpAuthClaims(auth: OpAuth): string | null { + return auth.claims ? JSON.stringify(auth.claims) : null; +} + +function decodeOpAuthClaims(json: string | null): OpAuth['claims'] | undefined { + if (!json) return undefined; + const value = JSON.parse(json) as { authoredAtMs?: unknown }; + return typeof value.authoredAtMs === 'number' ? { authoredAtMs: value.authoredAtMs } : undefined; +} + function dedupeLatestByKey(values: T[], keyOf: (value: T) => string): T[] { const deduped = new Map(); for (const value of values) { @@ -88,19 +108,20 @@ function insertOpAuthSql(entryCount: number): string { const rows: string[] = []; for (let i = 0; i < entryCount; i += 1) { - const base = i * 5; + const base = i * 6; rows.push( - `($${base + 1}, $${base + 2}::bytea, $${base + 3}::bytea, $${base + 4}::bytea, $${base + 5})`, + `($${base + 1}, $${base + 2}::bytea, $${base + 3}::bytea, $${base + 4}::bytea, $${base + 5}, $${base + 6})`, ); } return ` -INSERT INTO treecrdt_sync_op_auth (doc_id, op_ref, sig, proof_ref, created_at_ms) +INSERT INTO treecrdt_sync_op_auth (doc_id, op_ref, sig, proof_ref, claims_json, created_at_ms) VALUES ${rows.join(',\n')} ON CONFLICT (doc_id, op_ref) DO UPDATE SET sig = EXCLUDED.sig, proof_ref = EXCLUDED.proof_ref, + claims_json = EXCLUDED.claims_json, created_at_ms = EXCLUDED.created_at_ms `; } @@ -114,7 +135,7 @@ function selectOpAuthByRefsSql(opRefCount: number): string { (_value, index) => `$${index + 2}::bytea`, ).join(', '); return ` -SELECT op_ref, sig, proof_ref +SELECT op_ref, sig, proof_ref, claims_json FROM treecrdt_sync_op_auth WHERE doc_id = $1 AND op_ref IN (${placeholders}) `; @@ -130,6 +151,7 @@ export function createOpAuthStore(opts: { return { init: async () => { await pool.query(OP_AUTH_SCHEMA_SQL); + await pool.query(OP_AUTH_MIGRATIONS_SQL); }, forDoc: (docId: string): SyncOpAuthStore => ({ storeOpAuth: async (entries) => { @@ -138,7 +160,14 @@ export function createOpAuthStore(opts: { const params: Array = []; for (const entry of deduped) { - params.push(docId, entry.opRef, entry.auth.sig, entry.auth.proofRef ?? null, nowMs()); + params.push( + docId, + entry.opRef, + entry.auth.sig, + entry.auth.proofRef ?? null, + encodeOpAuthClaims(entry.auth), + nowMs(), + ); } await pool.query(insertOpAuthSql(deduped.length), params); @@ -151,6 +180,7 @@ export function createOpAuthStore(opts: { op_ref: Buffer; sig: Buffer; proof_ref: Buffer | null; + claims_json: string | null; }>(selectOpAuthByRefsSql(opRefs.length), [docId, ...opRefs]); const byOpRefHex = new Map(); @@ -158,7 +188,12 @@ export function createOpAuthStore(opts: { const opRef = toBytes(row.op_ref); const sig = toBytes(row.sig); const proofRef = row.proof_ref ? toBytes(row.proof_ref) : undefined; - byOpRefHex.set(bytesToHex(opRef), { sig, ...(proofRef ? { proofRef } : {}) }); + const claims = decodeOpAuthClaims(row.claims_json); + byOpRefHex.set(bytesToHex(opRef), { + sig, + ...(proofRef ? { proofRef } : {}), + ...(claims ? { claims } : {}), + }); } return opRefs.map((opRef) => byOpRefHex.get(bytesToHex(opRef)) ?? null); @@ -196,20 +231,21 @@ function insertPendingOpsSql(entryCount: number): string { const rows: string[] = []; for (let i = 0; i < entryCount; i += 1) { - const base = i * 8; + const base = i * 9; rows.push( - `($${base + 1}, $${base + 2}::bytea, $${base + 3}::bytea, $${base + 4}::bytea, $${base + 5}::bytea, $${base + 6}, $${base + 7}, $${base + 8})`, + `($${base + 1}, $${base + 2}::bytea, $${base + 3}::bytea, $${base + 4}::bytea, $${base + 5}::bytea, $${base + 6}, $${base + 7}, $${base + 8}, $${base + 9})`, ); } return ` -INSERT INTO treecrdt_sync_pending_ops (doc_id, op_ref, op, sig, proof_ref, reason, message, created_at_ms) +INSERT INTO treecrdt_sync_pending_ops (doc_id, op_ref, op, sig, proof_ref, claims_json, reason, message, created_at_ms) VALUES ${rows.join(',\n')} ON CONFLICT (doc_id, op_ref) DO UPDATE SET op = EXCLUDED.op, sig = EXCLUDED.sig, proof_ref = EXCLUDED.proof_ref, + claims_json = EXCLUDED.claims_json, reason = EXCLUDED.reason, message = EXCLUDED.message, created_at_ms = EXCLUDED.created_at_ms @@ -272,10 +308,12 @@ export function createPendingOpsStore(opts: { return { init: async () => { await pool.query(PENDING_SCHEMA_SQL); + await pool.query(PENDING_MIGRATIONS_SQL); }, forDoc: (docId: string): SyncPendingOpsStore => ({ init: async () => { await pool.query(PENDING_SCHEMA_SQL); + await pool.query(PENDING_MIGRATIONS_SQL); }, storePendingOps: async (entries) => { if (entries.length === 0) return; @@ -291,6 +329,7 @@ export function createPendingOpsStore(opts: { encodeTreecrdtSyncV0Operation(entry.op), entry.auth.sig, entry.auth.proofRef ?? null, + encodeOpAuthClaims(entry.auth), entry.reason, entry.message ?? null, nowMs(), @@ -304,11 +343,12 @@ export function createPendingOpsStore(opts: { op: Buffer; sig: Buffer; proof_ref: Buffer | null; + claims_json: string | null; reason: string; message: string | null; }>( ` -SELECT op, sig, proof_ref, reason, message +SELECT op, sig, proof_ref, claims_json, reason, message FROM treecrdt_sync_pending_ops WHERE doc_id = $1 ORDER BY created_at_ms ASC, op_ref ASC @@ -316,20 +356,24 @@ ORDER BY created_at_ms ASC, op_ref ASC [docId], ); - return res.rows.map((row) => ({ - op: decodeTreecrdtSyncV0Operation(toBytes(row.op)), - auth: { - sig: toBytes(row.sig), - ...(row.proof_ref ? { proofRef: toBytes(row.proof_ref) } : {}), - }, - reason: - row.reason === 'missing_context' - ? 'missing_context' - : (() => { - throw new Error(`unexpected pending reason: ${row.reason}`); - })(), - ...(row.message ? { message: row.message } : {}), - })); + return res.rows.map((row) => { + const claims = decodeOpAuthClaims(row.claims_json); + return { + op: decodeTreecrdtSyncV0Operation(toBytes(row.op)), + auth: { + sig: toBytes(row.sig), + ...(row.proof_ref ? { proofRef: toBytes(row.proof_ref) } : {}), + ...(claims ? { claims } : {}), + }, + reason: + row.reason === 'missing_context' + ? 'missing_context' + : (() => { + throw new Error(`unexpected pending reason: ${row.reason}`); + })(), + ...(row.message ? { message: row.message } : {}), + }; + }); }, listPendingOpRefs: async () => { const res = await pool.query<{ op_ref: Buffer }>( diff --git a/packages/sync-protocol/material/sqlite/src/backend.ts b/packages/sync-protocol/material/sqlite/src/backend.ts index e0c25ebc..0a7f7696 100644 --- a/packages/sync-protocol/material/sqlite/src/backend.ts +++ b/packages/sync-protocol/material/sqlite/src/backend.ts @@ -1,8 +1,9 @@ import type { Operation } from '@treecrdt/interface'; +import type { MaterializationSource, WriteOptions } from '@treecrdt/interface/engine'; import { bytesToHex, hexToBytes } from '@treecrdt/interface/ids'; import type { SqliteRunner } from '@treecrdt/interface/sqlite'; import { deriveOpRefV0 } from '@treecrdt/sync-protocol'; -import type { Filter, OpRef, SyncBackend } from '@treecrdt/sync-protocol'; +import type { Filter, OpRef, SyncApplyOpsMetadata, SyncBackend } from '@treecrdt/sync-protocol'; import { createPendingOpsStore } from './proof-material/index.js'; @@ -18,10 +19,38 @@ export type TreecrdtSyncBackendClient = { ops: { all?: () => Promise; get: (opRefs: OpRef[]) => Promise; - appendMany: (ops: Operation[]) => Promise; + appendMany: (ops: Operation[], opts?: WriteOptions) => Promise; }; }; +function materializationSourcesForOps( + ops: readonly Operation[], + metadata?: SyncApplyOpsMetadata, +): readonly MaterializationSource[] | undefined { + const verified = metadata?.verified; + if (!verified) return undefined; + + const sources = ops.map((op, i): MaterializationSource => { + const verifiedOp = verified[i]; + const authoredAtMs = verifiedOp?.claims?.authoredAtMs; + return { + operation: { + id: { + replica: op.meta.id.replica, + counter: op.meta.id.counter, + }, + lamport: op.meta.lamport, + }, + ...(verifiedOp?.signer + ? { signer: { publicKey: Uint8Array.from(verifiedOp.signer.publicKey) } } + : {}), + ...(authoredAtMs !== undefined ? { time: { authoredAtMs } } : {}), + }; + }); + + return sources.some((source) => source.signer || source.time) ? sources : undefined; +} + async function maybeLoadNodePayloadWriterOpRef(opts: { runner: SqliteRunner; docId: string; @@ -167,9 +196,13 @@ export function createTreecrdtSyncBackendFromClient( getOpsByOpRefs: async (opRefs) => client.ops.get(opRefs), - applyOps: async (ops) => { + applyOps: async (ops, metadata) => { if (ops.length === 0) return; - await client.ops.appendMany(ops); + const materializationSources = materializationSourcesForOps(ops, metadata); + await client.ops.appendMany( + ops, + materializationSources ? { materializationSources } : undefined, + ); }, ...(pending diff --git a/packages/sync-protocol/material/sqlite/src/proof-material/sqlite.ts b/packages/sync-protocol/material/sqlite/src/proof-material/sqlite.ts index 81a6066c..40a17e99 100644 --- a/packages/sync-protocol/material/sqlite/src/proof-material/sqlite.ts +++ b/packages/sync-protocol/material/sqlite/src/proof-material/sqlite.ts @@ -29,6 +29,26 @@ function hexToBytesStrict(hex: string, expectedLen: number, field: string): Uint return bytes; } +function encodeOpAuthClaims(auth: OpAuth): string | null { + return auth.claims ? JSON.stringify(auth.claims) : null; +} + +function decodeOpAuthClaims(json: string | null): OpAuth['claims'] | undefined { + if (!json) return undefined; + const value = JSON.parse(json) as { authoredAtMs?: unknown }; + return typeof value.authoredAtMs === 'number' ? { authoredAtMs: value.authoredAtMs } : undefined; +} + +async function ensureAuthClaimsColumn(runner: SqliteRunner, table: string): Promise { + const text = await runner.getText( + `SELECT COALESCE(json_group_array(name), '[]') FROM pragma_table_info('${table}')`, + ); + const names = text ? (JSON.parse(text) as string[]) : []; + if (!names.includes('claims_json')) { + await runner.exec(`ALTER TABLE ${table} ADD COLUMN claims_json TEXT`); + } +} + export type SqlitePendingOpsStore = SyncPendingOpsStore; export type SqliteOpAuthStore = SyncOpAuthStore & { init: () => Promise; @@ -44,6 +64,7 @@ CREATE TABLE IF NOT EXISTS treecrdt_sync_pending_ops ( op BLOB NOT NULL, -- protobuf bytes (sync/v0 Operation) sig BLOB NOT NULL, -- 64 bytes (Ed25519) proof_ref BLOB, -- 16 bytes (token id), nullable + claims_json TEXT, -- JSON-encoded signed OpAuth claims, nullable reason TEXT NOT NULL, -- e.g. "missing_context" message TEXT, created_at_ms INTEGER NOT NULL, @@ -57,6 +78,7 @@ CREATE TABLE IF NOT EXISTS treecrdt_sync_op_auth ( op_ref BLOB NOT NULL, -- 16 bytes sig BLOB NOT NULL, -- 64 bytes (Ed25519) proof_ref BLOB, -- 16 bytes (token id), nullable + claims_json TEXT, -- JSON-encoded signed OpAuth claims, nullable created_at_ms INTEGER NOT NULL, PRIMARY KEY (doc_id, op_ref) ); @@ -81,8 +103,8 @@ export function createPendingOpsStore(opts: { const insertSql = ` INSERT OR REPLACE INTO treecrdt_sync_pending_ops - (doc_id, op_ref, op, sig, proof_ref, reason, message, created_at_ms) -VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + (doc_id, op_ref, op, sig, proof_ref, claims_json, reason, message, created_at_ms) +VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) RETURNING 1 `; @@ -93,6 +115,7 @@ FROM ( 'op_hex', hex(op), 'sig_hex', hex(sig), 'proof_ref_hex', CASE WHEN proof_ref IS NULL THEN NULL ELSE hex(proof_ref) END, + 'claims_json', claims_json, 'reason', reason, 'message', message ) AS obj @@ -123,6 +146,7 @@ RETURNING 1 return { init: async () => { await opts.runner.exec(PENDING_SCHEMA_SQL); + await ensureAuthClaimsColumn(opts.runner, 'treecrdt_sync_pending_ops'); }, storePendingOps: async (pending) => { @@ -139,6 +163,7 @@ RETURNING 1 opBytes, p.auth.sig, proofRef, + encodeOpAuthClaims(p.auth), p.reason, message, nowMs(), @@ -153,6 +178,7 @@ RETURNING 1 op_hex: string; sig_hex: string; proof_ref_hex: string | null; + claims_json: string | null; reason: string; message: string | null; }>; @@ -165,6 +191,7 @@ RETURNING 1 const proofRef = r.proof_ref_hex ? hexToBytesStrict(r.proof_ref_hex, 16, 'pending proof_ref') : undefined; + const claims = decodeOpAuthClaims(r.claims_json); if (r.reason !== 'missing_context') { throw new Error(`unexpected pending reason: ${r.reason}`); @@ -172,7 +199,7 @@ RETURNING 1 return { op, - auth: { sig, ...(proofRef ? { proofRef } : {}) }, + auth: { sig, ...(proofRef ? { proofRef } : {}), ...(claims ? { claims } : {}) }, reason: 'missing_context', ...(r.message ? { message: r.message } : {}), } satisfies PendingOp; @@ -204,8 +231,8 @@ export function createOpAuthStore(opts: { const insertSql = ` INSERT OR REPLACE INTO treecrdt_sync_op_auth - (doc_id, op_ref, sig, proof_ref, created_at_ms) -VALUES (?1, ?2, ?3, ?4, ?5) + (doc_id, op_ref, sig, proof_ref, claims_json, created_at_ms) +VALUES (?1, ?2, ?3, ?4, ?5, ?6) RETURNING 1 `; @@ -219,7 +246,8 @@ FROM ( SELECT json_object( 'op_ref_hex', hex(op_ref), 'sig_hex', hex(sig), - 'proof_ref_hex', CASE WHEN proof_ref IS NULL THEN NULL ELSE hex(proof_ref) END + 'proof_ref_hex', CASE WHEN proof_ref IS NULL THEN NULL ELSE hex(proof_ref) END, + 'claims_json', claims_json ) AS obj FROM treecrdt_sync_op_auth WHERE doc_id = ?1 AND op_ref IN (${placeholders}) @@ -230,6 +258,7 @@ FROM ( return { init: async () => { await opts.runner.exec(OP_AUTH_SCHEMA_SQL); + await ensureAuthClaimsColumn(opts.runner, 'treecrdt_sync_op_auth'); }, storeOpAuth: async (entries) => { @@ -237,7 +266,14 @@ FROM ( for (const e of entries) { const proofRef = e.auth.proofRef ?? null; - await opts.runner.getText(insertSql, [opts.docId, e.opRef, e.auth.sig, proofRef, nowMs()]); + await opts.runner.getText(insertSql, [ + opts.docId, + e.opRef, + e.auth.sig, + proofRef, + encodeOpAuthClaims(e.auth), + nowMs(), + ]); } }, @@ -251,6 +287,7 @@ FROM ( op_ref_hex: string; sig_hex: string; proof_ref_hex: string | null; + claims_json: string | null; }>; const byHex = new Map(); @@ -260,7 +297,12 @@ FROM ( const proofRef = r.proof_ref_hex ? hexToBytesStrict(r.proof_ref_hex, 16, 'op_auth proof_ref') : undefined; - byHex.set(opRefHex, { sig, ...(proofRef ? { proofRef } : {}) }); + const claims = decodeOpAuthClaims(r.claims_json); + byHex.set(opRefHex, { + sig, + ...(proofRef ? { proofRef } : {}), + ...(claims ? { claims } : {}), + }); } return opRefs.map((ref) => byHex.get(bytesToHex(ref)) ?? null); diff --git a/packages/sync-protocol/material/sqlite/tests/backend.test.ts b/packages/sync-protocol/material/sqlite/tests/backend.test.ts index 931cca42..cf236c08 100644 --- a/packages/sync-protocol/material/sqlite/tests/backend.test.ts +++ b/packages/sync-protocol/material/sqlite/tests/backend.test.ts @@ -1,5 +1,6 @@ import { expect, test } from 'vitest'; +import type { Operation } from '@treecrdt/interface'; import { bytesToHex, nodeIdToBytes16 } from '@treecrdt/interface/ids'; import type { SqliteRunner } from '@treecrdt/interface/sqlite'; import { deriveOpRefV0, type OpRef } from '@treecrdt/sync-protocol'; @@ -14,6 +15,55 @@ function replicaFromLabel(label: string): Uint8Array { return out; } +test('backend applyOps forwards verified metadata as materialization sources', async () => { + const docId = 'doc-backend-materialization-metadata'; + const replica = replicaFromLabel('writer'); + const authoredAtMs = 1_700_000_000_456; + const op: Operation = { + meta: { id: { replica, counter: 7 }, lamport: 12 }, + kind: { + type: 'insert', + parent: '00'.repeat(16), + node: '44'.repeat(16), + orderKey: new Uint8Array([1]), + }, + }; + let captured: unknown; + + const backend = createTreecrdtSyncBackendFromClient( + { + opRefs: { + all: async () => [], + children: async () => [], + }, + ops: { + get: async () => [], + appendMany: async (_ops, opts) => { + captured = opts; + }, + }, + }, + docId, + ); + + await backend.applyOps([op], { + verified: [{ signer: { publicKey: replica }, claims: { authoredAtMs } }], + }); + + expect(captured).toEqual({ + materializationSources: [ + { + operation: { + id: { replica, counter: 7 }, + lamport: 12, + }, + signer: { publicKey: replica }, + time: { authoredAtMs }, + }, + ], + }); +}); + test("backend children(parent) includes the parent's latest payload opRef", async () => { const docId = 'doc-backend-payload-root'; const parentHex = '11'.repeat(16); diff --git a/packages/sync-protocol/protocol/src/auth.ts b/packages/sync-protocol/protocol/src/auth.ts index a6b2447b..cae62ffc 100644 --- a/packages/sync-protocol/protocol/src/auth.ts +++ b/packages/sync-protocol/protocol/src/auth.ts @@ -1,4 +1,4 @@ -import type { Capability, Hello, HelloAck, OpAuth } from './types.js'; +import type { Capability, Hello, HelloAck, OpAuth, SyncVerifiedOpMetadata } from './types.js'; import type { Filter } from './types.js'; export type SyncOpPurpose = 'reconcile' | 'subscribe' | 'reprocess_pending'; @@ -9,6 +9,8 @@ export type SyncAuthOpDisposition = export type SyncAuthVerifyOpsResult = { dispositions: SyncAuthOpDisposition[]; + /** Verified metadata aligned with the ops passed to verifyOps. */ + verified?: readonly (SyncVerifiedOpMetadata | undefined)[]; }; export type SyncAuthOpsContext = { diff --git a/packages/sync-protocol/protocol/src/gen/sync/v0/messages_pb.ts b/packages/sync-protocol/protocol/src/gen/sync/v0/messages_pb.ts index 3dd3fc0e..66b2ae90 100644 --- a/packages/sync-protocol/protocol/src/gen/sync/v0/messages_pb.ts +++ b/packages/sync-protocol/protocol/src/gen/sync/v0/messages_pb.ts @@ -18,7 +18,7 @@ import type { Message } from '@bufbuild/protobuf'; export const file_sync_v0_messages: GenFile = /*@__PURE__*/ fileDesc( - 'ChZzeW5jL3YwL21lc3NhZ2VzLnByb3RvEhB0cmVlY3JkdC5zeW5jLnYwIikKCkNhcGFiaWxpdHkSDAoEbmFtZRgBIAEoCRINCgV2YWx1ZRgCIAEoCSJCCgpGaWx0ZXJTcGVjEgoKAmlkGAEgASgJEigKBmZpbHRlchgCIAEoCzIYLnRyZWVjcmR0LnN5bmMudjAuRmlsdGVyIn8KBUhlbGxvEjIKDGNhcGFiaWxpdGllcxgBIAMoCzIcLnRyZWVjcmR0LnN5bmMudjAuQ2FwYWJpbGl0eRItCgdmaWx0ZXJzGAIgAygLMhwudHJlZWNyZHQuc3luYy52MC5GaWx0ZXJTcGVjEhMKC21heF9sYW1wb3J0GAMgASgEIloKDlJlamVjdGVkRmlsdGVyEgoKAmlkGAEgASgJEisKBnJlYXNvbhgCIAEoDjIbLnRyZWVjcmR0LnN5bmMudjAuRXJyb3JDb2RlEg8KB21lc3NhZ2UYAyABKAkiqQEKCEhlbGxvQWNrEjIKDGNhcGFiaWxpdGllcxgBIAMoCzIcLnRyZWVjcmR0LnN5bmMudjAuQ2FwYWJpbGl0eRIYChBhY2NlcHRlZF9maWx0ZXJzGAIgAygJEjoKEHJlamVjdGVkX2ZpbHRlcnMYAyADKAsyIC50cmVlY3JkdC5zeW5jLnYwLlJlamVjdGVkRmlsdGVyEhMKC21heF9sYW1wb3J0GAQgASgEIkIKDVJpYmx0Q29kZXdvcmQSDQoFY291bnQYASABKBESDwoHa2V5X3N1bRgCIAEoDBIRCgl2YWx1ZV9zdW0YAyABKAwiewoOUmlibHRDb2Rld29yZHMSEQoJZmlsdGVyX2lkGAEgASgJEg0KBXJvdW5kGAIgASgNEhMKC3N0YXJ0X2luZGV4GAMgASgEEjIKCWNvZGV3b3JkcxgEIAMoCzIfLnRyZWVjcmR0LnN5bmMudjAuUmlibHRDb2Rld29yZCI4CglSaWJsdE1vcmUSGgoSY29kZXdvcmRzX3JlY2VpdmVkGAEgASgEEg8KB2NyZWRpdHMYAiABKA0ijgEKDFJpYmx0RGVjb2RlZBIvCg5zZW5kZXJfbWlzc2luZxgBIAMoCzIXLnRyZWVjcmR0LnN5bmMudjAuT3BSZWYSMQoQcmVjZWl2ZXJfbWlzc2luZxgCIAMoCzIXLnRyZWVjcmR0LnN5bmMudjAuT3BSZWYSGgoSY29kZXdvcmRzX3JlY2VpdmVkGAMgASgEIlQKC1JpYmx0RmFpbGVkEjQKBnJlYXNvbhgBIAEoDjIkLnRyZWVjcmR0LnN5bmMudjAuUmlibHRGYWlsdXJlUmVhc29uEg8KB21lc3NhZ2UYAiABKAkiywEKC1JpYmx0U3RhdHVzEhEKCWZpbHRlcl9pZBgBIAEoCRINCgVyb3VuZBgCIAEoDRIxCgdkZWNvZGVkGAMgASgLMh4udHJlZWNyZHQuc3luYy52MC5SaWJsdERlY29kZWRIABIvCgZmYWlsZWQYBCABKAsyHS50cmVlY3JkdC5zeW5jLnYwLlJpYmx0RmFpbGVkSAASKwoEbW9yZRgFIAEoCzIbLnRyZWVjcmR0LnN5bmMudjAuUmlibHRNb3JlSABCCQoHcGF5bG9hZCJ9CghPcHNCYXRjaBIRCglmaWx0ZXJfaWQYASABKAkSKAoDb3BzGAIgAygLMhsudHJlZWNyZHQuc3luYy52MC5PcGVyYXRpb24SJgoEYXV0aBgEIAMoCzIYLnRyZWVjcmR0LnN5bmMudjAuT3BBdXRoEgwKBGRvbmUYAyABKAgiLgoGT3BBdXRoEgsKA3NpZxgCIAEoDBIRCglwcm9vZl9yZWYYAyABKAxKBAgBEAIiTgoJU3Vic2NyaWJlEhcKD3N1YnNjcmlwdGlvbl9pZBgBIAEoCRIoCgZmaWx0ZXIYAiABKAsyGC50cmVlY3JkdC5zeW5jLnYwLkZpbHRlciJACgxTdWJzY3JpYmVBY2sSFwoPc3Vic2NyaXB0aW9uX2lkGAEgASgJEhcKD2N1cnJlbnRfbGFtcG9ydBgCIAEoBCImCgtVbnN1YnNjcmliZRIXCg9zdWJzY3JpcHRpb25faWQYASABKAkicwoJU3luY0Vycm9yEikKBGNvZGUYASABKA4yGy50cmVlY3JkdC5zeW5jLnYwLkVycm9yQ29kZRIPCgdtZXNzYWdlGAIgASgJEhEKCWZpbHRlcl9pZBgDIAEoCRIXCg9zdWJzY3JpcHRpb25faWQYBCABKAkqewoSUmlibHRGYWlsdXJlUmVhc29uEiQKIFJJQkxUX0ZBSUxVUkVfUkVBU09OX1VOU1BFQ0lGSUVEEAASGgoWTUFYX0NPREVXT1JEU19FWENFRURFRBABEhEKDURFQ09ERV9GQUlMRUQQAhIQCgxPVVRfT0ZfT1JERVIQAyrJAQoJRXJyb3JDb2RlEhoKFkVSUk9SX0NPREVfVU5TUEVDSUZJRUQQABIXChNVTlNVUFBPUlRFRF9WRVJTSU9OEAESGAoURklMVEVSX05PVF9TVVBQT1JURUQQAhIUChBUT09fTUFOWV9GSUxURVJTEAMSIAocUkVDT05DSUxJQVRJT05fREVDT0RFX0ZBSUxFRBAEEhAKDFJBVEVfTElNSVRFRBAFEhEKDURPQ19OT1RfRk9VTkQQBhIQCgxVTkFVVEhPUklaRUQQB2IGcHJvdG8z', + 'ChZzeW5jL3YwL21lc3NhZ2VzLnByb3RvEhB0cmVlY3JkdC5zeW5jLnYwIikKCkNhcGFiaWxpdHkSDAoEbmFtZRgBIAEoCRINCgV2YWx1ZRgCIAEoCSJCCgpGaWx0ZXJTcGVjEgoKAmlkGAEgASgJEigKBmZpbHRlchgCIAEoCzIYLnRyZWVjcmR0LnN5bmMudjAuRmlsdGVyIn8KBUhlbGxvEjIKDGNhcGFiaWxpdGllcxgBIAMoCzIcLnRyZWVjcmR0LnN5bmMudjAuQ2FwYWJpbGl0eRItCgdmaWx0ZXJzGAIgAygLMhwudHJlZWNyZHQuc3luYy52MC5GaWx0ZXJTcGVjEhMKC21heF9sYW1wb3J0GAMgASgEIloKDlJlamVjdGVkRmlsdGVyEgoKAmlkGAEgASgJEisKBnJlYXNvbhgCIAEoDjIbLnRyZWVjcmR0LnN5bmMudjAuRXJyb3JDb2RlEg8KB21lc3NhZ2UYAyABKAkiqQEKCEhlbGxvQWNrEjIKDGNhcGFiaWxpdGllcxgBIAMoCzIcLnRyZWVjcmR0LnN5bmMudjAuQ2FwYWJpbGl0eRIYChBhY2NlcHRlZF9maWx0ZXJzGAIgAygJEjoKEHJlamVjdGVkX2ZpbHRlcnMYAyADKAsyIC50cmVlY3JkdC5zeW5jLnYwLlJlamVjdGVkRmlsdGVyEhMKC21heF9sYW1wb3J0GAQgASgEIkIKDVJpYmx0Q29kZXdvcmQSDQoFY291bnQYASABKBESDwoHa2V5X3N1bRgCIAEoDBIRCgl2YWx1ZV9zdW0YAyABKAwiewoOUmlibHRDb2Rld29yZHMSEQoJZmlsdGVyX2lkGAEgASgJEg0KBXJvdW5kGAIgASgNEhMKC3N0YXJ0X2luZGV4GAMgASgEEjIKCWNvZGV3b3JkcxgEIAMoCzIfLnRyZWVjcmR0LnN5bmMudjAuUmlibHRDb2Rld29yZCI4CglSaWJsdE1vcmUSGgoSY29kZXdvcmRzX3JlY2VpdmVkGAEgASgEEg8KB2NyZWRpdHMYAiABKA0ijgEKDFJpYmx0RGVjb2RlZBIvCg5zZW5kZXJfbWlzc2luZxgBIAMoCzIXLnRyZWVjcmR0LnN5bmMudjAuT3BSZWYSMQoQcmVjZWl2ZXJfbWlzc2luZxgCIAMoCzIXLnRyZWVjcmR0LnN5bmMudjAuT3BSZWYSGgoSY29kZXdvcmRzX3JlY2VpdmVkGAMgASgEIlQKC1JpYmx0RmFpbGVkEjQKBnJlYXNvbhgBIAEoDjIkLnRyZWVjcmR0LnN5bmMudjAuUmlibHRGYWlsdXJlUmVhc29uEg8KB21lc3NhZ2UYAiABKAkiywEKC1JpYmx0U3RhdHVzEhEKCWZpbHRlcl9pZBgBIAEoCRINCgVyb3VuZBgCIAEoDRIxCgdkZWNvZGVkGAMgASgLMh4udHJlZWNyZHQuc3luYy52MC5SaWJsdERlY29kZWRIABIvCgZmYWlsZWQYBCABKAsyHS50cmVlY3JkdC5zeW5jLnYwLlJpYmx0RmFpbGVkSAASKwoEbW9yZRgFIAEoCzIbLnRyZWVjcmR0LnN5bmMudjAuUmlibHRNb3JlSABCCQoHcGF5bG9hZCJ9CghPcHNCYXRjaBIRCglmaWx0ZXJfaWQYASABKAkSKAoDb3BzGAIgAygLMhsudHJlZWNyZHQuc3luYy52MC5PcGVyYXRpb24SJgoEYXV0aBgEIAMoCzIYLnRyZWVjcmR0LnN5bmMudjAuT3BBdXRoEgwKBGRvbmUYAyABKAgiXgoGT3BBdXRoEgsKA3NpZxgCIAEoDBIRCglwcm9vZl9yZWYYAyABKAwSLgoGY2xhaW1zGAQgASgLMh4udHJlZWNyZHQuc3luYy52MC5PcEF1dGhDbGFpbXNKBAgBEAIiPgoMT3BBdXRoQ2xhaW1zEhsKDmF1dGhvcmVkX2F0X21zGAEgASgESACIAQFCEQoPX2F1dGhvcmVkX2F0X21zIk4KCVN1YnNjcmliZRIXCg9zdWJzY3JpcHRpb25faWQYASABKAkSKAoGZmlsdGVyGAIgASgLMhgudHJlZWNyZHQuc3luYy52MC5GaWx0ZXIiQAoMU3Vic2NyaWJlQWNrEhcKD3N1YnNjcmlwdGlvbl9pZBgBIAEoCRIXCg9jdXJyZW50X2xhbXBvcnQYAiABKAQiJgoLVW5zdWJzY3JpYmUSFwoPc3Vic2NyaXB0aW9uX2lkGAEgASgJInMKCVN5bmNFcnJvchIpCgRjb2RlGAEgASgOMhsudHJlZWNyZHQuc3luYy52MC5FcnJvckNvZGUSDwoHbWVzc2FnZRgCIAEoCRIRCglmaWx0ZXJfaWQYAyABKAkSFwoPc3Vic2NyaXB0aW9uX2lkGAQgASgJKnsKElJpYmx0RmFpbHVyZVJlYXNvbhIkCiBSSUJMVF9GQUlMVVJFX1JFQVNPTl9VTlNQRUNJRklFRBAAEhoKFk1BWF9DT0RFV09SRFNfRVhDRUVERUQQARIRCg1ERUNPREVfRkFJTEVEEAISEAoMT1VUX09GX09SREVSEAMqyQEKCUVycm9yQ29kZRIaChZFUlJPUl9DT0RFX1VOU1BFQ0lGSUVEEAASFwoTVU5TVVBQT1JURURfVkVSU0lPThABEhgKFEZJTFRFUl9OT1RfU1VQUE9SVEVEEAISFAoQVE9PX01BTllfRklMVEVSUxADEiAKHFJFQ09OQ0lMSUFUSU9OX0RFQ09ERV9GQUlMRUQQBBIQCgxSQVRFX0xJTUlURUQQBRIRCg1ET0NfTk9UX0ZPVU5EEAYSEAoMVU5BVVRIT1JJWkVEEAdiBnByb3RvMw', [file_sync_v0_filters, file_sync_v0_ops, file_sync_v0_types], ); @@ -421,6 +421,13 @@ export type OpAuth = Message<'treecrdt.sync.v0.OpAuth'> & { * @generated from field: bytes proof_ref = 3; */ proofRef: Uint8Array; + + /** + * Optional signed claims about the operation. + * + * @generated from field: treecrdt.sync.v0.OpAuthClaims claims = 4; + */ + claims?: OpAuthClaims; }; /** @@ -431,6 +438,28 @@ export const OpAuthSchema: GenMessage = /*@__PURE__*/ messageDesc(file_sync_v0_messages, 12); +/** + * Signed claims carried alongside an operation signature. + * + * @generated from message treecrdt.sync.v0.OpAuthClaims + */ +export type OpAuthClaims = Message<'treecrdt.sync.v0.OpAuthClaims'> & { + /** + * Author's claimed wall-clock authoring time in unix milliseconds. + * + * @generated from field: optional uint64 authored_at_ms = 1; + */ + authoredAtMs?: bigint; +}; + +/** + * Describes the message treecrdt.sync.v0.OpAuthClaims. + * Use `create(OpAuthClaimsSchema)` to create a new message. + */ +export const OpAuthClaimsSchema: GenMessage = + /*@__PURE__*/ + messageDesc(file_sync_v0_messages, 13); + /** * Push-based subscription for live updates. * @@ -458,7 +487,7 @@ export type Subscribe = Message<'treecrdt.sync.v0.Subscribe'> & { */ export const SubscribeSchema: GenMessage = /*@__PURE__*/ - messageDesc(file_sync_v0_messages, 13); + messageDesc(file_sync_v0_messages, 14); /** * Acknowledges a Subscribe request. @@ -483,7 +512,7 @@ export type SubscribeAck = Message<'treecrdt.sync.v0.SubscribeAck'> & { */ export const SubscribeAckSchema: GenMessage = /*@__PURE__*/ - messageDesc(file_sync_v0_messages, 14); + messageDesc(file_sync_v0_messages, 15); /** * Stops a previously-established subscription. @@ -503,7 +532,7 @@ export type Unsubscribe = Message<'treecrdt.sync.v0.Unsubscribe'> & { */ export const UnsubscribeSchema: GenMessage = /*@__PURE__*/ - messageDesc(file_sync_v0_messages, 15); + messageDesc(file_sync_v0_messages, 16); /** * @generated from message treecrdt.sync.v0.SyncError @@ -536,7 +565,7 @@ export type SyncError = Message<'treecrdt.sync.v0.SyncError'> & { */ export const SyncErrorSchema: GenMessage = /*@__PURE__*/ - messageDesc(file_sync_v0_messages, 16); + messageDesc(file_sync_v0_messages, 17); /** * @generated from enum treecrdt.sync.v0.RibltFailureReason diff --git a/packages/sync-protocol/protocol/src/in-memory.ts b/packages/sync-protocol/protocol/src/in-memory.ts index 4a945132..a104642e 100644 --- a/packages/sync-protocol/protocol/src/in-memory.ts +++ b/packages/sync-protocol/protocol/src/in-memory.ts @@ -2,7 +2,7 @@ import type { SyncPeerOptions } from './sync.js'; import { SyncPeer } from './sync.js'; import type { DuplexTransport, WireCodec } from './transport/index.js'; import { createInMemoryDuplex, wrapDuplexTransportWithCodec } from './transport/index.js'; -import type { Filter, OpRef, SyncBackend, SyncMessage } from './types.js'; +import type { Filter, OpRef, SyncApplyOpsMetadata, SyncBackend, SyncMessage } from './types.js'; export type FlushableSyncBackend = SyncBackend & { flush: () => Promise }; @@ -12,7 +12,7 @@ export function makeQueuedSyncBackend(opts: { maxLamportFromOps: (ops: Op[]) => number; listOpRefs: (filter: Filter) => Promise; getOpsByOpRefs: (opRefs: OpRef[]) => Promise; - applyOps: (ops: Op[]) => Promise; + applyOps: (ops: Op[], metadata?: SyncApplyOpsMetadata) => Promise; }): FlushableSyncBackend { let maxLamportValue = opts.initialMaxLamport; let lastApply = Promise.resolve(); @@ -22,11 +22,11 @@ export function makeQueuedSyncBackend(opts: { maxLamport: async () => BigInt(maxLamportValue), listOpRefs: opts.listOpRefs, getOpsByOpRefs: opts.getOpsByOpRefs, - applyOps: async (ops: Op[]) => { + applyOps: async (ops: Op[], metadata?: SyncApplyOpsMetadata) => { if (ops.length === 0) return; const nextMax = opts.maxLamportFromOps(ops); if (nextMax > maxLamportValue) maxLamportValue = nextMax; - lastApply = lastApply.then(() => opts.applyOps(ops)); + lastApply = lastApply.then(() => opts.applyOps(ops, metadata)); await lastApply; }, flush: async () => lastApply, diff --git a/packages/sync-protocol/protocol/src/protobuf.ts b/packages/sync-protocol/protocol/src/protobuf.ts index 1fc9007c..3663a9a5 100644 --- a/packages/sync-protocol/protocol/src/protobuf.ts +++ b/packages/sync-protocol/protocol/src/protobuf.ts @@ -14,6 +14,7 @@ import type { Hello, HelloAck, OpAuth, + OpAuthClaims, OpRef, OpsBatch, RibltCodewords, @@ -355,16 +356,32 @@ function toProtoOpAuth(auth: OpAuth) { return create(OpAuthSchema, { sig: auth.sig, ...(auth.proofRef ? { proofRef: auth.proofRef } : {}), + ...(auth.claims ? { claims: toProtoOpAuthClaims(auth.claims) } : {}), }); } +function toProtoOpAuthClaims(claims: OpAuthClaims) { + return { + ...(claims.authoredAtMs !== undefined ? { authoredAtMs: BigInt(claims.authoredAtMs) } : {}), + }; +} + +function fromProtoOpAuthClaims(claims: any): OpAuthClaims | undefined { + if (!claims) return undefined; + const out: OpAuthClaims = {}; + if (claims.authoredAtMs !== undefined) out.authoredAtMs = Number(claims.authoredAtMs); + return Object.keys(out).length > 0 ? out : undefined; +} + function fromProtoOpAuth(auth: any): OpAuth { const sig = auth.sig as Uint8Array | undefined; const proofRef = auth.proofRef as Uint8Array | undefined; + const claims = fromProtoOpAuthClaims(auth.claims); if (!sig) throw new Error('OpAuth.sig missing'); return { sig, ...(proofRef && proofRef.length > 0 ? { proofRef } : {}), + ...(claims ? { claims } : {}), }; } diff --git a/packages/sync-protocol/protocol/src/sync.ts b/packages/sync-protocol/protocol/src/sync.ts index 1ec62114..0c72e5f2 100644 --- a/packages/sync-protocol/protocol/src/sync.ts +++ b/packages/sync-protocol/protocol/src/sync.ts @@ -5,7 +5,12 @@ import { isAnyAuthCapability, isAuthCapability, } from './auth-capabilities.js'; -import type { SyncAuth, SyncAuthVerifyOpsResult, SyncOpPurpose } from './auth.js'; +import type { + SyncAuth, + SyncAuthOpDisposition, + SyncAuthVerifyOpsResult, + SyncOpPurpose, +} from './auth.js'; import { buildInitiatorHelloCapabilities, capabilitySetFingerprint, @@ -31,6 +36,7 @@ import type { RibltStatus, Subscribe, SubscribeAck, + SyncVerifiedOpMetadata, SyncBackend, SyncMessage, Unsubscribe, @@ -65,6 +71,31 @@ function deferred(): Pending { return { promise, resolve, reject }; } +function parseVerifyOpsResult( + verifyRes: void | SyncAuthVerifyOpsResult, + opCount: number, +): { + dispositions?: readonly SyncAuthOpDisposition[]; + verified?: readonly (SyncVerifiedOpMetadata | undefined)[]; +} { + if (verifyRes === undefined) return {}; + const dispositions = + (verifyRes as SyncAuthVerifyOpsResult)?.dispositions ?? + (() => { + throw new Error('verifyOps must return void or { dispositions: [...] }'); + })(); + if (dispositions.length !== opCount) { + throw new Error(`verifyOps returned ${dispositions.length} dispositions for ${opCount} ops`); + } + + const verified = (verifyRes as SyncAuthVerifyOpsResult).verified; + if (verified && verified.length !== opCount) { + throw new Error(`verifyOps returned ${verified.length} metadata entries for ${opCount} ops`); + } + + return { dispositions, verified }; +} + type ResponderSession = { filter: Filter; round: number; @@ -1708,18 +1739,7 @@ export class SyncPeer { purpose, filterId: batch.filterId, }); - const dispositions = - verifyRes === undefined - ? undefined - : ((verifyRes as SyncAuthVerifyOpsResult)?.dispositions ?? - (() => { - throw new Error('verifyOps must return void or { dispositions: [...] }'); - })()); - if (dispositions && dispositions.length !== batch.ops.length) { - throw new Error( - `verifyOps returned ${dispositions.length} dispositions for ${batch.ops.length} ops`, - ); - } + const { dispositions, verified } = parseVerifyOpsResult(verifyRes, batch.ops.length); if (auth && auth.length > 0) { await this.auth?.onVerifiedOps?.(batch.ops, auth, { docId: this.backend.docId, @@ -1730,12 +1750,14 @@ export class SyncPeer { const pending: PendingOp[] = []; const allowedOps: Op[] = []; + const allowedMetadata: (SyncVerifiedOpMetadata | undefined)[] = []; for (let i = 0; i < batch.ops.length; i += 1) { const op = batch.ops[i]!; const d = dispositions?.[i]; if (!d || d.status === 'allow') { allowedOps.push(op); + allowedMetadata.push(verified?.[i]); continue; } if (d.status !== 'pending_context') { @@ -1761,7 +1783,7 @@ export class SyncPeer { await this.backend.storePendingOps(pending); } - await this.backend.applyOps(allowedOps); + await this.backend.applyOps(allowedOps, verified ? { verified: allowedMetadata } : undefined); if (allowedOps.length > 0) void this.notifyLocalUpdate(allowedOps); await this.reprocessPendingOps(); @@ -1825,17 +1847,11 @@ export class SyncPeer { continue; } - const dispositions = - res === undefined - ? undefined - : ((res as SyncAuthVerifyOpsResult)?.dispositions ?? - (() => { - throw new Error('verifyOps must return void or { dispositions: [...] }'); - })()); + const { dispositions, verified } = parseVerifyOpsResult(res, 1); const d = dispositions?.[0]; if (d && d.status === 'pending_context') continue; - await this.backend.applyOps([p.op]); + await this.backend.applyOps([p.op], verified ? { verified } : undefined); await this.backend.deletePendingOps!([p.op]); progress = true; appliedOps.push(p.op); diff --git a/packages/sync-protocol/protocol/src/types.ts b/packages/sync-protocol/protocol/src/types.ts index c1974ff5..f06dfda9 100644 --- a/packages/sync-protocol/protocol/src/types.ts +++ b/packages/sync-protocol/protocol/src/types.ts @@ -71,6 +71,23 @@ export type RibltStatus = { export type OpAuth = { sig: Bytes; proofRef?: Bytes; + claims?: OpAuthClaims; +}; + +export type OpAuthClaims = { + authoredAtMs?: number; +}; + +export type SyncVerifiedOpMetadata = { + signer?: { + publicKey: Bytes; + }; + claims?: OpAuthClaims; +}; + +export type SyncApplyOpsMetadata = { + /** Verified metadata aligned with the `ops` argument. */ + verified?: readonly (SyncVerifiedOpMetadata | undefined)[]; }; export type PendingOpReason = 'missing_context'; @@ -132,7 +149,7 @@ export interface SyncBackend { maxLamport(): Promise; listOpRefs(filter: Filter): Promise; getOpsByOpRefs(opRefs: OpRef[]): Promise; - applyOps(ops: Op[]): Promise; + applyOps(ops: Op[], metadata?: SyncApplyOpsMetadata): Promise; /** * Optional: persist ops that were structurally valid (signatures/capabilities) diff --git a/packages/sync-protocol/protocol/tests/helpers/pending-proof-material-contract.ts b/packages/sync-protocol/protocol/tests/helpers/pending-proof-material-contract.ts index 999622b3..9968c351 100644 --- a/packages/sync-protocol/protocol/tests/helpers/pending-proof-material-contract.ts +++ b/packages/sync-protocol/protocol/tests/helpers/pending-proof-material-contract.ts @@ -43,12 +43,14 @@ function makePending( sigFill: number, proofFill?: number, message?: string, + authoredAtMs?: number, ): PendingOp { return { op: makeInsertOp(replicaFill, counter), auth: { sig: new Uint8Array(64).fill(sigFill), ...(proofFill === undefined ? {} : { proofRef: new Uint8Array(16).fill(proofFill) }), + ...(authoredAtMs === undefined ? {} : { claims: { authoredAtMs } }), }, reason: 'missing_context', ...(message ? { message } : {}), @@ -71,6 +73,7 @@ function normalizePending(value: PendingOp) { kind, sigHex: bytesToHex(value.auth.sig), proofRefHex: value.auth.proofRef ? bytesToHex(value.auth.proofRef) : null, + claims: value.auth.claims ?? null, reason: value.reason, message: value.message ?? null, }; @@ -95,7 +98,7 @@ export function definePendingProofMaterialStoreContract( try { const docId = `doc-pending-${randomUUID()}`; const pending = await harness.createPendingStore(docId); - const first = makePending(7, 1, 9, 4, 'need ancestry'); + const first = makePending(7, 1, 9, 4, 'need ancestry', 1_700_000_000_001); const second = makePending(7, 2, 8); await pending.init(); diff --git a/packages/sync-protocol/protocol/tests/helpers/proof-material-contract.ts b/packages/sync-protocol/protocol/tests/helpers/proof-material-contract.ts index 64993025..5dc2ed30 100644 --- a/packages/sync-protocol/protocol/tests/helpers/proof-material-contract.ts +++ b/packages/sync-protocol/protocol/tests/helpers/proof-material-contract.ts @@ -24,10 +24,11 @@ function makeOpRef(fill: number): OpRef { return new Uint8Array(16).fill(fill); } -function makeOpAuth(sigFill: number, proofFill?: number): OpAuth { +function makeOpAuth(sigFill: number, proofFill?: number, authoredAtMs?: number): OpAuth { return { sig: new Uint8Array(64).fill(sigFill), ...(proofFill === undefined ? {} : { proofRef: new Uint8Array(16).fill(proofFill) }), + ...(authoredAtMs === undefined ? {} : { claims: { authoredAtMs } }), }; } @@ -39,6 +40,7 @@ function normalizeOpAuth(value: OpAuth): OpAuth { return { sig: Uint8Array.from(value.sig), ...(value.proofRef ? { proofRef: Uint8Array.from(value.proofRef) } : {}), + ...(value.claims ? { claims: { ...value.claims } } : {}), }; } @@ -57,7 +59,7 @@ export function defineProofMaterialStoreContract( const refA = makeOpRef(1); const refB = makeOpRef(2); const missing = makeOpRef(3); - const authA = makeOpAuth(9, 4); + const authA = makeOpAuth(9, 4, 1_700_000_000_001); const authB = makeOpAuth(7); await opAuth.storeOpAuth([ diff --git a/packages/sync-protocol/protocol/tests/helpers/replay-only-auth-contract.ts b/packages/sync-protocol/protocol/tests/helpers/replay-only-auth-contract.ts index 8e5e3ace..ed9310a3 100644 --- a/packages/sync-protocol/protocol/tests/helpers/replay-only-auth-contract.ts +++ b/packages/sync-protocol/protocol/tests/helpers/replay-only-auth-contract.ts @@ -99,6 +99,7 @@ function normalizeOpAuth(value: OpAuth): OpAuth { return { sig: Uint8Array.from(value.sig), ...(value.proofRef ? { proofRef: Uint8Array.from(value.proofRef) } : {}), + ...(value.claims ? { claims: { ...value.claims } } : {}), }; } @@ -138,6 +139,7 @@ export function defineReplayOnlyAuthStoreContract( const authEntry: OpAuth = { sig: new Uint8Array(64).fill(9), proofRef: new Uint8Array(16).fill(3), + claims: { authoredAtMs: 1_700_000_000_001 }, }; const authorAuth: SyncAuth = { diff --git a/packages/sync-protocol/protocol/tests/smoke.test.ts b/packages/sync-protocol/protocol/tests/smoke.test.ts index 3e720b0b..1a18036d 100644 --- a/packages/sync-protocol/protocol/tests/smoke.test.ts +++ b/packages/sync-protocol/protocol/tests/smoke.test.ts @@ -11,7 +11,13 @@ import { SyncPeer } from '../dist/sync.js'; import { createInMemoryConnectedPeers } from '../dist/in-memory.js'; import { wrapDuplexTransportWithCodec } from '../dist/transport/index.js'; import type { DuplexTransport } from '../dist/transport/index.js'; -import type { Filter, OpRef, SyncBackend, SyncMessage } from '../dist/types.js'; +import type { + Filter, + OpRef, + SyncApplyOpsMetadata, + SyncBackend, + SyncMessage, +} from '../dist/types.js'; function opRefFor(docId: string, replica: string, counter: number): OpRef { const h = createHash('sha256'); @@ -334,6 +340,15 @@ class CountingListBackend extends MemoryBackend { } } +class MetadataCaptureBackend extends MemoryBackend { + lastApplyMetadata: SyncApplyOpsMetadata | undefined; + + override async applyOps(ops: Operation[], metadata?: SyncApplyOpsMetadata): Promise { + this.lastApplyMetadata = metadata; + await super.applyOps(ops); + } +} + test('syncOnce does not starve macrotask transports', async () => { const docId = 'doc-sync-macrotask'; const root = '0'.repeat(32); @@ -365,6 +380,63 @@ test('syncOnce does not starve macrotask transports', async () => { }); }); +test('syncOnce passes verified auth metadata to applyOps', async () => { + const docId = 'doc-sync-auth-metadata'; + const root = '0'.repeat(32); + const authoredAtMs = 1_700_000_000_123; + + const a = new MemoryBackend(docId); + const b = new MetadataCaptureBackend(docId); + const op = makeOp(replicas.a, 1, 1, { + type: 'insert', + parent: root, + node: nodeIdFromInt(13), + orderKey: orderKeyFromPosition(0), + }); + + await a.applyOps([op]); + + const { + peerA: pa, + transportA: ta, + detach, + } = createInMemoryConnectedPeers({ + backendA: a, + backendB: b, + codec: treecrdtSyncV0ProtobufCodec, + peerAOptions: { + auth: { + helloCapabilities: async () => [{ name: 'auth.capability', value: 'writer-token' }], + signOps: async (ops) => + ops.map(() => ({ sig: new Uint8Array([1]), claims: { authoredAtMs } })), + }, + }, + peerBOptions: { + auth: { + helloCapabilities: async () => [{ name: 'auth.capability', value: 'reader-token' }], + onHello: async () => [{ name: 'auth.capability', value: 'reader-token' }], + verifyOps: async (ops, auth) => ({ + dispositions: ops.map(() => ({ status: 'allow' as const })), + verified: ops.map((verifiedOp, i) => ({ + signer: { publicKey: verifiedOp.meta.id.replica }, + claims: auth?.[i]?.claims, + })), + }), + }, + }, + }); + + try { + await pa.syncOnce(ta, { all: {} }, { maxCodewords: 10_000, codewordsPerMessage: 256 }); + + expect(b.hasOp(replicaHex.a, 1)).toBe(true); + expect(b.lastApplyMetadata?.verified?.[0]?.signer?.publicKey).toEqual(replicas.a); + expect(b.lastApplyMetadata?.verified?.[0]?.claims).toEqual({ authoredAtMs }); + } finally { + detach(); + } +}); + test('syncOnce paces outbound codewords until delayed ribltStatus arrives', async () => { const docId = 'doc-sync-delayed-status'; const root = '0'.repeat(32); @@ -570,6 +642,36 @@ test('syncOnce protobuf roundtrips ribltStatus.more', () => { expect(treecrdtSyncV0ProtobufCodec.decode(treecrdtSyncV0ProtobufCodec.encode(msg))).toEqual(msg); }); +test('syncOnce protobuf roundtrips op auth claims', () => { + const op = makeOp(replicas.a, 1, 1, { + type: 'insert', + parent: '0'.repeat(32), + node: nodeIdFromInt(1), + orderKey: orderKeyFromPosition(0), + }); + const msg: SyncMessage = { + v: 0, + docId: 'doc-sync-op-auth-claims-codec', + payload: { + case: 'opsBatch', + value: { + filterId: 'all', + ops: [op], + auth: [ + { + sig: new Uint8Array(64).fill(9), + proofRef: new Uint8Array(16).fill(3), + claims: { authoredAtMs: 1_700_000_000_123 }, + }, + ], + done: true, + }, + }, + }; + + expect(treecrdtSyncV0ProtobufCodec.decode(treecrdtSyncV0ProtobufCodec.encode(msg))).toEqual(msg); +}); + test('syncOnce waits for ribltStatus.more before sending another codeword batch', async () => { const docId = 'doc-sync-more-flow-control'; const root = '0'.repeat(32); diff --git a/packages/treecrdt-auth/src/internal/cose-cwt-auth.ts b/packages/treecrdt-auth/src/internal/cose-cwt-auth.ts index 7ab96a8e..abc17b14 100644 --- a/packages/treecrdt-auth/src/internal/cose-cwt-auth.ts +++ b/packages/treecrdt-auth/src/internal/cose-cwt-auth.ts @@ -21,6 +21,7 @@ import { type OpRef, type SyncAuth, type SyncCapabilityMaterialStore, + type SyncVerifiedOpMetadata, } from '@treecrdt/sync-protocol'; import { base64urlDecode, base64urlEncode } from '../base64url.js'; @@ -45,7 +46,12 @@ import { type CapabilityGrant, type TreecrdtCapabilityRevocationCheckContext, } from './capability.js'; -import { signTreecrdtOpV1, verifyTreecrdtOpV1 } from './op-sig.js'; +import { + signTreecrdtOpV1, + signTreecrdtOpV2, + verifyTreecrdtOpV1, + verifyTreecrdtOpV2, +} from './op-sig.js'; import { getField } from './claims.js'; import { capAllowsNode, @@ -88,7 +94,11 @@ export type TreecrdtCoseCwtAuthOptions = { ) => boolean | Promise; allowUnsigned?: boolean; requireProofRef?: boolean; + /** Include a signed `claims.authoredAtMs` value on locally authored ops. Defaults to true. */ + includeAuthoredAt?: boolean; now?: () => number; + /** Wall-clock source for authoredAtMs claims. */ + nowMs?: () => number; }; export type TreecrdtCoseCwtParseRevocationCheckContext = @@ -109,8 +119,10 @@ export type TreecrdtCoseCwtRevocationCheckContext = export function createTreecrdtCoseCwtAuth(opts: TreecrdtCoseCwtAuthOptions): SyncAuth { const now = opts.now ?? (() => Math.floor(Date.now() / 1000)); + const nowMs = opts.nowMs ?? (() => Date.now()); const allowUnsigned = opts.allowUnsigned ?? false; const requireProofRef = opts.requireProofRef ?? false; + const includeAuthoredAt = opts.includeAuthoredAt ?? true; const localTokens = opts.localCapabilityTokens ?? []; const localTokenIds = localTokens.map((t) => deriveTokenIdV1(t)); @@ -650,12 +662,24 @@ export function createTreecrdtCoseCwtAuth(opts: TreecrdtCoseCwtAuthOptions): Syn }); proofRef = selected.tokenId; } - const sig = await signTreecrdtOpV1({ - docId: ctx.docId, - op, - privateKey: opts.localPrivateKey, - }); - const entry: OpAuth = { sig, ...(proofRef ? { proofRef } : {}) }; + const claims = includeAuthoredAt ? { authoredAtMs: nowMs() } : undefined; + const sig = claims + ? await signTreecrdtOpV2({ + docId: ctx.docId, + op, + privateKey: opts.localPrivateKey, + claims, + }) + : await signTreecrdtOpV1({ + docId: ctx.docId, + op, + privateKey: opts.localPrivateKey, + }); + const entry: OpAuth = { + sig, + ...(proofRef ? { proofRef } : {}), + ...(claims ? { claims } : {}), + }; opAuthByOpRefHex.set(opRefHex, entry); out[i] = entry; } @@ -691,6 +715,7 @@ export function createTreecrdtCoseCwtAuth(opts: TreecrdtCoseCwtAuthOptions): Syn const dispositions: Array< { status: 'allow' } | { status: 'pending_context'; message?: string } > = []; + const verified: Array = []; const toPersist: Array<{ opRef: OpRef; auth: OpAuth }> = []; for (let i = 0; i < ops.length; i += 1) { const op = ops[i]!; @@ -755,13 +780,25 @@ export function createTreecrdtCoseCwtAuth(opts: TreecrdtCoseCwtAuthOptions): Syn }); if (scopeRes === 'deny') throw new Error('capability does not allow op'); - const ok = await verifyTreecrdtOpV1({ - docId: ctx.docId, - op, - signature: a.sig, - publicKey: replica, - }); + const ok = a.claims + ? await verifyTreecrdtOpV2({ + docId: ctx.docId, + op, + signature: a.sig, + publicKey: replica, + claims: a.claims, + }) + : await verifyTreecrdtOpV1({ + docId: ctx.docId, + op, + signature: a.sig, + publicKey: replica, + }); if (!ok) throw new Error('invalid op signature'); + verified.push({ + signer: { publicKey: Uint8Array.from(replica) }, + ...(a.claims ? { claims: a.claims } : {}), + }); const opRef = deriveOpRefV0(ctx.docId, { replica, counter: op.meta.id.counter }); opAuthByOpRefHex.set(bytesToHex(opRef), a); if (opts.opAuthStore) toPersist.push({ opRef, auth: a }); @@ -781,13 +818,14 @@ export function createTreecrdtCoseCwtAuth(opts: TreecrdtCoseCwtAuthOptions): Syn await ensureOpAuthStoreReady(); await opts.opAuthStore.storeOpAuth(toPersist); } - return { dispositions }; + return { dispositions, verified }; } if (opts.opAuthStore && toPersist.length > 0) { await ensureOpAuthStoreReady(); await opts.opAuthStore.storeOpAuth(toPersist); } + return { dispositions, verified }; }, }; } diff --git a/packages/treecrdt-auth/src/internal/op-sig.ts b/packages/treecrdt-auth/src/internal/op-sig.ts index 5c947f73..b5b9f7fa 100644 --- a/packages/treecrdt-auth/src/internal/op-sig.ts +++ b/packages/treecrdt-auth/src/internal/op-sig.ts @@ -7,8 +7,13 @@ import { signEd25519, verifyEd25519 } from '../ed25519.js'; import { concatBytes, u32be, u64be, u8 } from './bytes.js'; const OP_SIG_V1_DOMAIN = utf8ToBytes('treecrdt/op-sig/v1'); +const OP_SIG_V2_DOMAIN = utf8ToBytes('treecrdt/op-sig/v2'); -export function encodeTreecrdtOpSigInputV1(opts: { docId: string; op: Operation }): Uint8Array { +export type TreecrdtOpAuthClaimsV1 = { + authoredAtMs?: number; +}; + +function encodeTreecrdtOpFields(opts: { docId: string; op: Operation }): Uint8Array { const docIdBytes = utf8ToBytes(opts.docId); const replicaBytes = replicaIdToBytes(opts.op.meta.id.replica); @@ -80,8 +85,6 @@ export function encodeTreecrdtOpSigInputV1(opts: { docId: string; op: Operation } return concatBytes( - OP_SIG_V1_DOMAIN, - u8(0), u32be(docIdBytes.length), docIdBytes, u32be(replicaBytes.length), @@ -93,6 +96,36 @@ export function encodeTreecrdtOpSigInputV1(opts: { docId: string; op: Operation ); } +function encodeTreecrdtOpAuthClaimsV1(claims: TreecrdtOpAuthClaimsV1): Uint8Array { + const parts: Uint8Array[] = []; + if (claims.authoredAtMs === undefined) { + parts.push(u8(0)); + } else { + if (!Number.isSafeInteger(claims.authoredAtMs) || claims.authoredAtMs < 0) { + throw new Error(`authoredAtMs must be a non-negative safe integer: ${claims.authoredAtMs}`); + } + parts.push(u8(1), u64be(claims.authoredAtMs)); + } + return concatBytes(...parts); +} + +export function encodeTreecrdtOpSigInputV1(opts: { docId: string; op: Operation }): Uint8Array { + return concatBytes(OP_SIG_V1_DOMAIN, u8(0), encodeTreecrdtOpFields(opts)); +} + +export function encodeTreecrdtOpSigInputV2(opts: { + docId: string; + op: Operation; + claims: TreecrdtOpAuthClaimsV1; +}): Uint8Array { + return concatBytes( + OP_SIG_V2_DOMAIN, + u8(0), + encodeTreecrdtOpFields(opts), + encodeTreecrdtOpAuthClaimsV1(opts.claims), + ); +} + export async function signTreecrdtOpV1(opts: { docId: string; op: Operation; @@ -102,6 +135,20 @@ export async function signTreecrdtOpV1(opts: { return signEd25519(msg, opts.privateKey); } +export async function signTreecrdtOpV2(opts: { + docId: string; + op: Operation; + privateKey: Uint8Array; + claims: TreecrdtOpAuthClaimsV1; +}): Promise { + const msg = encodeTreecrdtOpSigInputV2({ + docId: opts.docId, + op: opts.op, + claims: opts.claims, + }); + return signEd25519(msg, opts.privateKey); +} + export async function verifyTreecrdtOpV1(opts: { docId: string; op: Operation; @@ -111,3 +158,18 @@ export async function verifyTreecrdtOpV1(opts: { const msg = encodeTreecrdtOpSigInputV1({ docId: opts.docId, op: opts.op }); return await verifyEd25519(opts.signature, msg, opts.publicKey); } + +export async function verifyTreecrdtOpV2(opts: { + docId: string; + op: Operation; + signature: Uint8Array; + publicKey: Uint8Array; + claims: TreecrdtOpAuthClaimsV1; +}): Promise { + const msg = encodeTreecrdtOpSigInputV2({ + docId: opts.docId, + op: opts.op, + claims: opts.claims, + }); + return await verifyEd25519(opts.signature, msg, opts.publicKey); +} diff --git a/packages/treecrdt-auth/src/session.ts b/packages/treecrdt-auth/src/session.ts index d801e988..fc686713 100644 --- a/packages/treecrdt-auth/src/session.ts +++ b/packages/treecrdt-auth/src/session.ts @@ -80,6 +80,7 @@ export type TreecrdtAuthSessionOptions = Omit< export type TreecrdtAuthSession = { syncAuth: SyncAuth; + signer: { publicKey: Uint8Array }; readonly ready: Promise>; getState: () => TreecrdtAuthSessionState; refresh: () => Promise>; @@ -195,6 +196,7 @@ export function createTreecrdtAuthSession(opts: TreecrdtAuthSessionOptions): Tre return { syncAuth, + signer: { publicKey: Uint8Array.from(local.publicKey) }, get ready() { return ready; }, diff --git a/packages/treecrdt-auth/src/treecrdt-auth.ts b/packages/treecrdt-auth/src/treecrdt-auth.ts index e003b167..350a0df5 100644 --- a/packages/treecrdt-auth/src/treecrdt-auth.ts +++ b/packages/treecrdt-auth/src/treecrdt-auth.ts @@ -13,9 +13,13 @@ export type { export { encodeTreecrdtOpSigInputV1, + encodeTreecrdtOpSigInputV2, signTreecrdtOpV1, + signTreecrdtOpV2, verifyTreecrdtOpV1, + verifyTreecrdtOpV2, } from './internal/op-sig.js'; +export type { TreecrdtOpAuthClaimsV1 } from './internal/op-sig.js'; export type { TreecrdtScopeEvaluator } from './internal/scope.js'; diff --git a/packages/treecrdt-auth/tests/auth.test.ts b/packages/treecrdt-auth/tests/auth.test.ts index 336f7375..510edafd 100644 --- a/packages/treecrdt-auth/tests/auth.test.ts +++ b/packages/treecrdt-auth/tests/auth.test.ts @@ -335,6 +335,85 @@ test('auth: signOps selects proof_ref per op when multiple tokens exist', async ); }); +test('auth: signOps signs authoredAt claims', async () => { + const docId = 'doc-auth-authored-at'; + const root = '0'.repeat(32); + const authoredAtMs = 1_700_000_000_123; + + const issuerSk = ed25519Utils.randomSecretKey(); + const issuerPk = await getPublicKey(issuerSk); + const writerSk = ed25519Utils.randomSecretKey(); + const writerPk = await getPublicKey(writerSk); + const verifierSk = ed25519Utils.randomSecretKey(); + const verifierPk = await getPublicKey(verifierSk); + + const token = issueTreecrdtCapabilityTokenV1({ + issuerPrivateKey: issuerSk, + subjectPublicKey: writerPk, + docId, + actions: ['write_structure'], + }); + + const authWriter = createTreecrdtCoseCwtAuth({ + issuerPublicKeys: [issuerPk], + localPrivateKey: writerSk, + localPublicKey: writerPk, + localCapabilityTokens: [token], + requireProofRef: true, + nowMs: () => authoredAtMs, + }); + + const authVerifier = createTreecrdtCoseCwtAuth({ + issuerPublicKeys: [issuerPk], + localPrivateKey: verifierSk, + localPublicKey: verifierPk, + requireProofRef: true, + }); + + const helloCaps = await authWriter.helloCapabilities?.({ docId }); + await authVerifier.onHello?.( + { capabilities: helloCaps ?? [], filters: [], maxLamport: 0n }, + { docId }, + ); + + const op = makeOp(writerPk, 1, 1, { + type: 'insert', + parent: root, + node: nodeIdFromInt(1), + orderKey: orderKeyFromPosition(0), + }); + + const ctx = { docId, purpose: 'reconcile' as const, filterId: 'all' }; + const auth = await authWriter.signOps?.([op], ctx); + expect(auth?.[0]?.claims).toEqual({ authoredAtMs }); + + const verifyRes = await authVerifier.verifyOps?.([op], auth, ctx); + expect(verifyRes?.dispositions).toEqual([{ status: 'allow' }]); + expect(verifyRes?.verified?.[0]?.signer?.publicKey).toEqual(writerPk); + expect(verifyRes?.verified?.[0]?.claims).toEqual({ authoredAtMs }); + + await expect( + authVerifier.verifyOps?.( + [op], + [{ ...auth![0]!, claims: { authoredAtMs: authoredAtMs + 1 } }], + ctx, + ), + ).rejects.toThrow(/invalid op signature/i); + + await expect( + authVerifier.verifyOps?.( + [op], + [ + { + sig: auth![0]!.sig, + ...(auth![0]!.proofRef ? { proofRef: auth![0]!.proofRef } : {}), + }, + ], + ctx, + ), + ).rejects.toThrow(/invalid op signature/i); +}); + test('auth ignores foreign peer capability tokens during hello and still verifies known authors', async () => { const docId = 'doc-auth-ignore-foreign-hello-cap'; const root = '0'.repeat(32); @@ -411,7 +490,7 @@ test('auth ignores foreign peer capability tokens during hello and still verifie ).resolves.toEqual([true]); await expect( authVerifier.verifyOps?.([op], auth, { docId, purpose: 'reconcile', filterId: 'all' }), - ).resolves.toBeUndefined(); + ).resolves.toMatchObject({ dispositions: [{ status: 'allow' }] }); }); test('auth re-advertises trusted author capability tokens from the capability store after restart', async () => { @@ -509,7 +588,7 @@ test('auth re-advertises trusted author capability tokens from the capability st const signed = await authWriter.signOps?.([op], { docId, purpose: 'reconcile', filterId: 'all' }); await expect( authJoiner.verifyOps?.([op], signed, { docId, purpose: 'reconcile', filterId: 'all' }), - ).resolves.toBeUndefined(); + ).resolves.toMatchObject({ dispositions: [{ status: 'allow' }] }); }); test('auth: replayed author capability tokens do not widen peer filter scope', async () => { diff --git a/packages/treecrdt-auth/tests/session.test.ts b/packages/treecrdt-auth/tests/session.test.ts index bc2137fd..1672ebdb 100644 --- a/packages/treecrdt-auth/tests/session.test.ts +++ b/packages/treecrdt-auth/tests/session.test.ts @@ -38,6 +38,7 @@ test('auth session warms sync auth and exposes ready state', async () => { expect(session.getState().status).toBe('loading'); await session.ready; expect(session.getState().status).toBe('ready'); + expect(session.signer.publicKey).toEqual(testKey(5)); }); test('auth session advertises async local identity chain without app-side wrappers', async () => { diff --git a/packages/treecrdt-sqlite-node/src/index.ts b/packages/treecrdt-sqlite-node/src/index.ts index 1673de85..8cb5e7f5 100644 --- a/packages/treecrdt-sqlite-node/src/index.ts +++ b/packages/treecrdt-sqlite-node/src/index.ts @@ -1,6 +1,7 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { + annotateOutcomeSources, createTreecrdtSqliteAdapter, createTreecrdtSqliteWriter, decodeSqliteNodeIds, @@ -221,11 +222,17 @@ export function createTreecrdtClient( ops: { append: async (op, writeOpts?: WriteOptions) => { const outcome = await adapter.appendOp(op, nodeIdToBytes16, encodeReplica); - materialized.emitOutcome(outcome, writeOpts?.writeId); + materialized.emitOutcome( + annotateOutcomeSources(outcome, writeOpts?.materializationSources), + writeOpts?.writeId, + ); }, appendMany: async (ops, writeOpts?: WriteOptions) => { const outcome = await adapter.appendOps!(ops, nodeIdToBytes16, encodeReplica); - materialized.emitOutcome(outcome, writeOpts?.writeId); + materialized.emitOutcome( + annotateOutcomeSources(outcome, writeOpts?.materializationSources), + writeOpts?.writeId, + ); }, all: () => opsSinceImpl(0), since: opsSinceImpl, diff --git a/packages/treecrdt-sqlite-node/tests/conformance.test.ts b/packages/treecrdt-sqlite-node/tests/conformance.test.ts index 15813a20..9c57b883 100644 --- a/packages/treecrdt-sqlite-node/tests/conformance.test.ts +++ b/packages/treecrdt-sqlite-node/tests/conformance.test.ts @@ -7,6 +7,8 @@ import { runTreecrdtEngineConformanceScenario, treecrdtEngineConformanceScenarios, } from '@treecrdt/engine-conformance'; +import type { Operation } from '@treecrdt/interface'; +import type { MaterializationEvent } from '@treecrdt/interface/engine'; import { createTreecrdtClient, defaultExtensionPath, @@ -68,9 +70,11 @@ test('sqlite auth-aware local write rolls back on auth failure', async () => { test('sqlite auth-aware local write emits materialization after auth succeeds', async () => { const client = await createNodeEngine({ docId: 'sqlite-auth-local-success' }); - const events: unknown[] = []; + const events: MaterializationEvent[] = []; const unsubscribe = client.onMaterialized((event) => events.push(event)); + const signerPublicKey = Uint8Array.from({ length: 32 }, (_, i) => i + 1); const authSession = { + signer: { publicKey: signerPublicKey }, authorizeLocalOps: vi.fn(async () => { expect(events).toHaveLength(0); }), @@ -85,6 +89,7 @@ test('sqlite auth-aware local write emits materialization after auth succeeds', expect(op.kind.type).toBe('insert'); expect(authSession.authorizeLocalOps).toHaveBeenCalledTimes(1); expect(events).toHaveLength(1); + expect(events[0]!.changes[0]!.source?.signer?.publicKey).toEqual(signerPublicKey); expect(await client.tree.exists(node)).toBe(true); expect(await client.ops.all()).toHaveLength(1); } finally { @@ -93,6 +98,46 @@ test('sqlite auth-aware local write emits materialization after auth succeeds', } }); +test('sqlite append emits provided materialization source metadata', async () => { + const client = await createNodeEngine({ docId: 'sqlite-append-source-metadata' }); + const events: MaterializationEvent[] = []; + const unsubscribe = client.onMaterialized((event) => events.push(event)); + const node = nodeIdFromInt(12); + const signerPublicKey = Uint8Array.from({ length: 32 }, (_, i) => 255 - i); + const authoredAtMs = 1_700_000_000_789; + const op: Operation = { + meta: { id: { replica, counter: 1 }, lamport: 1 }, + kind: { + type: 'insert', + parent: root, + node, + orderKey: new Uint8Array([1]), + }, + }; + + try { + await client.ops.append(op, { + materializationSources: [ + { + operation: { + id: { replica: op.meta.id.replica, counter: op.meta.id.counter }, + lamport: op.meta.lamport, + }, + signer: { publicKey: signerPublicKey }, + time: { authoredAtMs }, + }, + ], + }); + + expect(events).toHaveLength(1); + expect(events[0]!.changes[0]!.source?.signer?.publicKey).toEqual(signerPublicKey); + expect(events[0]!.changes[0]!.source?.time).toEqual({ authoredAtMs }); + } finally { + unsubscribe(); + await client.close(); + } +}); + for (const scenario of treecrdtEngineConformanceScenarios()) { test(`sqlite engine conformance (node): ${scenario.name}`, async () => { let persistentDir: string | null = null; diff --git a/packages/treecrdt-ts/src/engine.ts b/packages/treecrdt-ts/src/engine.ts index e32c8c20..e342e775 100644 --- a/packages/treecrdt-ts/src/engine.ts +++ b/packages/treecrdt-ts/src/engine.ts @@ -1,6 +1,15 @@ import type { Operation, ReplicaId } from './index.js'; import type { SqliteTreeChildRow, SqliteTreeRow, TreecrdtSqlitePlacement } from './sqlite.js'; +export type MaterializationSigner = { + publicKey: Uint8Array; +}; + +export type MaterializationTime = { + /** Signed wall-clock authoring claim, when the auth layer verified one. */ + authoredAtMs?: number; +}; + export type MaterializationSource = { /** * Operation that caused the visible change. @@ -15,10 +24,10 @@ export type MaterializationSource = { /** Lamport timestamp assigned to the operation. */ lamport: number; }; - /** Future auth metadata for the operation signer, when available. */ - signer?: { - publicKey: Uint8Array; - }; + /** Auth signer metadata for the operation, when available. */ + signer?: MaterializationSigner; + /** Authenticated timing metadata for the operation, when available. */ + time?: MaterializationTime; }; type ChangeSource = { @@ -112,10 +121,24 @@ export function createMaterializationDispatcher(): MaterializationDispatcher { export type WriteOptions = { writeId?: string; + /** + * Optional source metadata aligned with the ops passed to append/appendMany. + * + * This lets sync/auth adapters preserve verified op metadata on materialization events without + * putting app metadata into TreeCRDT payloads. + */ + materializationSources?: readonly MaterializationSource[]; }; export type LocalWriteAuthSession = { + /** + * Authorizes local ops before they are exposed as committed. + * + * If the session has signer metadata available, expose it on `signer` so SQLite writers can + * annotate materialization events. + */ authorizeLocalOps: (ops: readonly Operation[]) => Promise; + signer?: MaterializationSigner; }; export type LocalWriteOptions = { diff --git a/packages/treecrdt-ts/src/sqlite.ts b/packages/treecrdt-ts/src/sqlite.ts index 3768169e..2a408672 100644 --- a/packages/treecrdt-ts/src/sqlite.ts +++ b/packages/treecrdt-ts/src/sqlite.ts @@ -4,9 +4,12 @@ import type { LocalWriteOptions, MaterializationEvent, MaterializationOutcome, + MaterializationSigner, MaterializationSource, + WriteOptions, } from './engine.js'; import { + bytesToHex, decodeNodeId, decodeReplicaId, hexToBytes, @@ -85,6 +88,76 @@ function emitLocalOutcome( emit?.({ ...outcome, ...(writeId ? { writeIds: [writeId] } : {}) }); } +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} + +function decodeMaterializationSigner(raw: unknown): MaterializationSigner | undefined { + if (!isRecord(raw)) return undefined; + const publicKey = raw.publicKey; + if (publicKey instanceof Uint8Array) return { publicKey: Uint8Array.from(publicKey) }; + if (Array.isArray(publicKey)) return { publicKey: Uint8Array.from(publicKey) }; + return undefined; +} + +function localWriteAuthSigner( + authSession: NonNullable, +): MaterializationSigner | undefined { + return decodeMaterializationSigner(authSession.signer); +} + +function materializationSourceKey(source: MaterializationSource): string { + return `${bytesToHex(source.operation.id.replica)}:${source.operation.id.counter}`; +} + +export function annotateOutcomeSources( + outcome: MaterializationOutcome, + sources?: WriteOptions['materializationSources'], +): MaterializationOutcome { + if (!sources || sources.length === 0) return outcome; + const byOpId = new Map(); + for (const source of sources) byOpId.set(materializationSourceKey(source), source); + if (byOpId.size === 0) return outcome; + + return { + ...outcome, + changes: outcome.changes.map((change) => { + const source = change.source; + if (!source) return change; + const verifiedSource = byOpId.get(materializationSourceKey(source)); + if (!verifiedSource) return change; + return { + ...change, + source: { + ...source, + ...(verifiedSource.signer ? { signer: verifiedSource.signer } : {}), + ...(verifiedSource.time ? { time: verifiedSource.time } : {}), + }, + }; + }), + }; +} + +function annotateOutcomeSigner( + outcome: MaterializationOutcome, + op: Operation, + signer?: MaterializationSigner, +): MaterializationOutcome { + if (!signer) return outcome; + return annotateOutcomeSources(outcome, [ + { + operation: { + id: { + replica: replicaIdToBytes(op.meta.id.replica), + counter: op.meta.id.counter, + }, + lamport: op.meta.lamport, + }, + signer, + }, + ]); +} + const ROOT_NODE_BYTES = nodeIdToBytes16(ROOT_NODE_ID_HEX); function buildAppendOp( @@ -639,7 +712,11 @@ export function createTreecrdtSqliteWriter( await authSession.authorizeLocalOps([op]); await sqliteExec(runner, `RELEASE ${savepoint}`); released = true; - emitLocalOutcome(outcome, opts.onMaterialized, writeOpts?.writeId); + emitLocalOutcome( + annotateOutcomeSigner(outcome, op, localWriteAuthSigner(authSession)), + opts.onMaterialized, + writeOpts?.writeId, + ); return op; } catch (err) { if (!released) { diff --git a/packages/treecrdt-wa-sqlite/src/client.ts b/packages/treecrdt-wa-sqlite/src/client.ts index d8446288..076ff3b4 100644 --- a/packages/treecrdt-wa-sqlite/src/client.ts +++ b/packages/treecrdt-wa-sqlite/src/client.ts @@ -1,6 +1,7 @@ import { clearOpfsStorage, detectOpfsSupport } from './opfs.js'; import type { Operation, ReplicaId } from '@treecrdt/interface'; import { + annotateOutcomeSources, createTreecrdtSqliteWriter, decodeSqliteNodeIds, decodeSqliteOpRefs, @@ -855,11 +856,17 @@ function makeTreecrdtClientFromCall(opts: { ops: { append: async (op, writeOpts?: WriteOptions) => { const outcome = await call('append', [op]); - materialized.emitOutcome(outcome, writeOpts?.writeId); + materialized.emitOutcome( + annotateOutcomeSources(outcome, writeOpts?.materializationSources), + writeOpts?.writeId, + ); }, appendMany: async (ops, writeOpts?: WriteOptions) => { const outcome = await call('appendMany', [ops]); - materialized.emitOutcome(outcome, writeOpts?.writeId); + materialized.emitOutcome( + annotateOutcomeSources(outcome, writeOpts?.materializationSources), + writeOpts?.writeId, + ); }, all: () => opsSinceImpl(0), since: opsSinceImpl,