From 2cd49a0d6f4332cd08ac904dabdbe3c08cc46d43 Mon Sep 17 00:00:00 2001 From: Chad Miller Date: Mon, 05 Jan 2026 21:12:32 +0000 Subject: [PATCH] feat: add DAG-CBOR encoding and fix integer overflow in CBOR Add explicit CID wrapper class for reliable DAG-CBOR tag 42 encoding instead of heuristic detection. Fix 32-bit signed integer overflow when encoding large lengths by using Math.floor instead of bitshift. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/pds.js | 710 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------------------------------------------------- test/pds.test.js | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------ 2 file(s) changed, 701 insertion(s)(+), 98 deletion(s)(-) diff --git a/src/pds.js b/src/pds.js --- a/src/pds.js +++ b/src/pds.js @@ -1,7 +1,19 @@ +// === CID WRAPPER === +// Explicit CID type for DAG-CBOR encoding (avoids fragile heuristic detection) + +class CID { + constructor(bytes) { + if (!(bytes instanceof Uint8Array)) { + throw new Error('CID must be constructed with Uint8Array') + } + this.bytes = bytes + } +} + // === CBOR ENCODING === // Minimal deterministic CBOR (RFC 8949) - sorted keys, minimal integers -function cborEncode(value) { +export function cborEncode(value) { const parts = [] function encode(val) { @@ -43,7 +55,12 @@ } else if (length < 65536) { parts.push(mt | 25, length >> 8, length & 0xff) } else if (length < 4294967296) { - parts.push(mt | 26, (length >> 24) & 0xff, (length >> 16) & 0xff, (length >> 8) & 0xff, length & 0xff) + // Use Math.floor instead of bitshift to avoid 32-bit signed integer overflow + parts.push(mt | 26, + Math.floor(length / 0x1000000) & 0xff, + Math.floor(length / 0x10000) & 0xff, + Math.floor(length / 0x100) & 0xff, + length & 0xff) } } @@ -59,7 +76,80 @@ return new Uint8Array(parts) } -function cborDecode(bytes) { +// DAG-CBOR encoder that handles CIDs with tag 42 +function cborEncodeDagCbor(value) { + const parts = [] + + function encode(val) { + if (val === null) { + parts.push(0xf6) // null + } else if (val === true) { + parts.push(0xf5) // true + } else if (val === false) { + parts.push(0xf4) // false + } else if (typeof val === 'number') { + if (Number.isInteger(val) && val >= 0) { + encodeHead(0, val) + } else if (Number.isInteger(val) && val < 0) { + encodeHead(1, -val - 1) + } + } else if (typeof val === 'string') { + const bytes = new TextEncoder().encode(val) + encodeHead(3, bytes.length) + parts.push(...bytes) + } else if (val instanceof CID) { + // CID - encode with CBOR tag 42 + 0x00 prefix + parts.push(0xd8, 42) // tag(42) + encodeHead(2, val.bytes.length + 1) // +1 for 0x00 prefix + parts.push(0x00) // multibase identity prefix + parts.push(...val.bytes) + } else if (val instanceof Uint8Array) { + // Regular byte string + encodeHead(2, val.length) + parts.push(...val) + } else if (Array.isArray(val)) { + encodeHead(4, val.length) + for (const item of val) encode(item) + } else if (typeof val === 'object') { + // DAG-CBOR: sort keys by length first, then lexicographically + const keys = Object.keys(val).filter(k => val[k] !== undefined) + keys.sort((a, b) => { + if (a.length !== b.length) return a.length - b.length + return a < b ? -1 : a > b ? 1 : 0 + }) + encodeHead(5, keys.length) + for (const key of keys) { + const keyBytes = new TextEncoder().encode(key) + encodeHead(3, keyBytes.length) + parts.push(...keyBytes) + encode(val[key]) + } + } + } + + function encodeHead(majorType, length) { + const mt = majorType << 5 + if (length < 24) { + parts.push(mt | length) + } else if (length < 256) { + parts.push(mt | 24, length) + } else if (length < 65536) { + parts.push(mt | 25, length >> 8, length & 0xff) + } else if (length < 4294967296) { + // Use Math.floor instead of bitshift to avoid 32-bit signed integer overflow + parts.push(mt | 26, + Math.floor(length / 0x1000000) & 0xff, + Math.floor(length / 0x10000) & 0xff, + Math.floor(length / 0x100) & 0xff, + length & 0xff) + } + } + + encode(value) + return new Uint8Array(parts) +} + +export function cborDecode(bytes) { let offset = 0 function read() { @@ -71,7 +161,8 @@ if (info === 24) length = bytes[offset++] else if (info === 25) { length = (bytes[offset++] << 8) | bytes[offset++] } else if (info === 26) { - length = (bytes[offset++] << 24) | (bytes[offset++] << 16) | (bytes[offset++] << 8) | bytes[offset++] + // Use multiplication instead of bitshift to avoid 32-bit signed integer overflow + length = bytes[offset++] * 0x1000000 + bytes[offset++] * 0x10000 + bytes[offset++] * 0x100 + bytes[offset++] } switch (major) { @@ -115,7 +206,7 @@ // === CID GENERATION === // dag-cbor (0x71) + sha-256 (0x12) + 32 bytes -async function createCid(bytes) { +export async function createCid(bytes) { const hash = await crypto.subtle.digest('SHA-256', bytes) const hashBytes = new Uint8Array(hash) @@ -131,12 +222,12 @@ return cid } -function cidToString(cid) { +export function cidToString(cid) { // base32lower encoding for CIDv1 return 'b' + base32Encode(cid) } -function base32Encode(bytes) { +export function base32Encode(bytes) { const alphabet = 'abcdefghijklmnopqrstuvwxyz234567' let result = '' let bits = 0 @@ -165,7 +256,7 @@ let lastTimestamp = 0 let clockId = Math.floor(Math.random() * 1024) -function createTid() { +export function createTid() { let timestamp = Date.now() * 1000 // microseconds // Ensure monotonic @@ -194,7 +285,12 @@ // === P-256 SIGNING === // Web Crypto ECDSA with P-256 curve -async function importPrivateKey(privateKeyBytes) { +export async function importPrivateKey(privateKeyBytes) { + // Validate private key length (P-256 requires exactly 32 bytes) + if (!(privateKeyBytes instanceof Uint8Array) || privateKeyBytes.length !== 32) { + throw new Error(`Invalid private key: expected 32 bytes, got ${privateKeyBytes?.length ?? 'non-Uint8Array'}`) + } + // PKCS#8 wrapper for raw P-256 private key const pkcs8Prefix = new Uint8Array([ 0x30, 0x41, 0x02, 0x01, 0x00, 0x30, 0x13, 0x06, 0x07, 0x2a, 0x86, 0x48, @@ -215,16 +311,53 @@ ) } -async function sign(privateKey, data) { +// P-256 curve order N +const P256_N = BigInt('0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551') +const P256_N_DIV_2 = P256_N / 2n + +function bytesToBigInt(bytes) { + let result = 0n + for (const byte of bytes) { + result = (result << 8n) | BigInt(byte) + } + return result +} + +function bigIntToBytes(n, length) { + const bytes = new Uint8Array(length) + for (let i = length - 1; i >= 0; i--) { + bytes[i] = Number(n & 0xffn) + n >>= 8n + } + return bytes +} + +export async function sign(privateKey, data) { const signature = await crypto.subtle.sign( { name: 'ECDSA', hash: 'SHA-256' }, privateKey, data ) - return new Uint8Array(signature) + const sig = new Uint8Array(signature) + + // Low-S normalization: if S > N/2, replace S with N - S + const r = sig.slice(0, 32) + const s = sig.slice(32, 64) + const sBigInt = bytesToBigInt(s) + + if (sBigInt > P256_N_DIV_2) { + const newS = P256_N - sBigInt + const newSBytes = bigIntToBytes(newS, 32) + const normalized = new Uint8Array(64) + normalized.set(r, 0) + normalized.set(newSBytes, 32) + return normalized + } + + return sig } -async function generateKeyPair() { +export async function generateKeyPair() { const keyPair = await crypto.subtle.generateKey( { name: 'ECDSA', namedCurve: 'P-256' }, true, @@ -265,11 +398,11 @@ return bytes } -function bytesToHex(bytes) { +export function bytesToHex(bytes) { return Array.from(bytes).map(b => b.toString(16).padStart(2, '0')).join('') } -function hexToBytes(hex) { +export function hexToBytes(hex) { const bytes = new Uint8Array(hex.length / 2) for (let i = 0; i < hex.length; i += 2) { bytes[i / 2] = parseInt(hex.substr(i, 2), 16) @@ -278,21 +411,29 @@ } // === MERKLE SEARCH TREE === -// Simple rebuild-on-write implementation +// ATProto-compliant MST implementation async function sha256(data) { const hash = await crypto.subtle.digest('SHA-256', data) return new Uint8Array(hash) } -function getKeyDepth(key) { - // Count leading zeros in hash to determine tree depth +// Cache for key depths (SHA-256 is expensive) +const keyDepthCache = new Map() + +export async function getKeyDepth(key) { + // Count leading zeros in SHA-256 hash, divide by 2 + if (keyDepthCache.has(key)) return keyDepthCache.get(key) + const keyBytes = new TextEncoder().encode(key) - // Sync hash for depth calculation (use first bytes of key as proxy) + const hash = await sha256(keyBytes) + let zeros = 0 - for (const byte of keyBytes) { - if (byte === 0) zeros += 8 - else { + for (const byte of hash) { + if (byte === 0) { + zeros += 8 + } else { + // Count leading zeros in this byte for (let i = 7; i >= 0; i--) { if ((byte >> i) & 1) break zeros++ @@ -300,7 +441,19 @@ break } } - return Math.floor(zeros / 4) + + const depth = Math.floor(zeros / 2) + keyDepthCache.set(key, depth) + return depth +} + +// Compute common prefix length between two byte arrays +function commonPrefixLen(a, b) { + const minLen = Math.min(a.length, b.length) + for (let i = 0; i < minLen; i++) { + if (a[i] !== b[i]) return i + } + return minLen } class MST { @@ -317,52 +470,86 @@ return null } - const entries = records.map(r => ({ - key: `${r.collection}/${r.rkey}`, - cid: r.cid - })) + // Build entries with pre-computed depths + const entries = [] + for (const r of records) { + const key = `${r.collection}/${r.rkey}` + entries.push({ + key, + keyBytes: new TextEncoder().encode(key), + cid: r.cid, + depth: await getKeyDepth(key) + }) + } return this.buildTree(entries, 0) } - async buildTree(entries, depth) { + async buildTree(entries, layer) { if (entries.length === 0) return null - const node = { l: null, e: [] } - let leftEntries = [] + // Separate entries for this layer vs deeper layers + const thisLayer = [] + let leftSubtree = [] for (const entry of entries) { - const keyDepth = getKeyDepth(entry.key) - - if (keyDepth > depth) { - leftEntries.push(entry) + if (entry.depth > layer) { + leftSubtree.push(entry) } else { - // Store accumulated left entries - if (leftEntries.length > 0) { - const leftCid = await this.buildTree(leftEntries, depth + 1) - if (node.e.length === 0) { - node.l = leftCid - } else { - node.e[node.e.length - 1].t = leftCid - } - leftEntries = [] + // Process accumulated left subtree + if (leftSubtree.length > 0) { + const leftCid = await this.buildTree(leftSubtree, layer + 1) + thisLayer.push({ type: 'subtree', cid: leftCid }) + leftSubtree = [] } - node.e.push({ k: entry.key, v: entry.cid, t: null }) + thisLayer.push({ type: 'entry', entry }) } } - // Handle remaining left entries - if (leftEntries.length > 0) { - const leftCid = await this.buildTree(leftEntries, depth + 1) - if (node.e.length > 0) { - node.e[node.e.length - 1].t = leftCid + // Handle remaining left subtree + if (leftSubtree.length > 0) { + const leftCid = await this.buildTree(leftSubtree, layer + 1) + thisLayer.push({ type: 'subtree', cid: leftCid }) + } + + // Build node with proper ATProto format + const node = { e: [] } + let leftCid = null + let prevKeyBytes = new Uint8Array(0) + + for (let i = 0; i < thisLayer.length; i++) { + const item = thisLayer[i] + + if (item.type === 'subtree') { + if (node.e.length === 0) { + leftCid = item.cid + } else { + // Attach to previous entry's 't' field + node.e[node.e.length - 1].t = new CID(cidToBytes(item.cid)) + } } else { - node.l = leftCid + // Entry - compute prefix compression + const keyBytes = item.entry.keyBytes + const prefixLen = commonPrefixLen(prevKeyBytes, keyBytes) + const keySuffix = keyBytes.slice(prefixLen) + + const e = { + p: prefixLen, + k: keySuffix, + v: new CID(cidToBytes(item.entry.cid)), + t: null // Always include t field (set later if subtree exists) + } + + node.e.push(e) + prevKeyBytes = keyBytes } } - // Encode and store node - const nodeBytes = cborEncode(node) + // Always include left pointer (can be null) + node.l = leftCid ? new CID(cidToBytes(leftCid)) : null + + // Encode node with proper MST CBOR format + const nodeBytes = cborEncodeMstNode(node) const nodeCid = await createCid(nodeBytes) const cidStr = cidToString(nodeCid) @@ -376,9 +563,67 @@ } } +// Special CBOR encoder for MST nodes (CIDs as raw bytes with tag 42) +function cborEncodeMstNode(node) { + const parts = [] + + function encode(val) { + if (val === null || val === undefined) { + parts.push(0xf6) // null + } else if (typeof val === 'number') { + encodeHead(0, val) // unsigned int + } else if (val instanceof CID) { + // CID - encode with CBOR tag 42 + 0x00 prefix (DAG-CBOR CID link) + parts.push(0xd8, 42) // tag 42 + encodeHead(2, val.bytes.length + 1) // +1 for 0x00 prefix + parts.push(0x00) // multibase identity prefix + parts.push(...val.bytes) + } else if (val instanceof Uint8Array) { + // Regular bytes + encodeHead(2, val.length) + parts.push(...val) + } else if (Array.isArray(val)) { + encodeHead(4, val.length) + for (const item of val) encode(item) + } else if (typeof val === 'object') { + // Sort keys for deterministic encoding (DAG-CBOR style) + // Include null values, only exclude undefined + const keys = Object.keys(val).filter(k => val[k] !== undefined) + keys.sort((a, b) => { + // DAG-CBOR: sort by length first, then lexicographically + if (a.length !== b.length) return a.length - b.length + return a < b ? -1 : a > b ? 1 : 0 + }) + encodeHead(5, keys.length) + for (const key of keys) { + // Encode key as text string + const keyBytes = new TextEncoder().encode(key) + encodeHead(3, keyBytes.length) + parts.push(...keyBytes) + // Encode value + encode(val[key]) + } + } + } + + function encodeHead(majorType, length) { + const mt = majorType << 5 + if (length < 24) { + parts.push(mt | length) + } else if (length < 256) { + parts.push(mt | 24, length) + } else if (length < 65536) { + parts.push(mt | 25, length >> 8, length & 0xff) + } + } + + encode(node) + return new Uint8Array(parts) +} + // === CAR FILE BUILDER === -function varint(n) { +export function varint(n) { const bytes = [] while (n >= 0x80) { bytes.push((n & 0x7f) | 0x80) @@ -388,13 +633,13 @@ return new Uint8Array(bytes) } -function cidToBytes(cidStr) { +export function cidToBytes(cidStr) { // Decode base32lower CID string to bytes if (!cidStr.startsWith('b')) throw new Error('expected base32lower CID') return base32Decode(cidStr.slice(1)) } -function base32Decode(str) { +export function base32Decode(str) { const alphabet = 'abcdefghijklmnopqrstuvwxyz234567' let bits = 0 let value = 0 @@ -414,12 +659,64 @@ return new Uint8Array(output) } -function buildCarFile(rootCid, blocks) { +// Encode CAR header with proper DAG-CBOR CID links +function cborEncodeCarHeader(obj) { + const parts = [] + + function encodeHead(majorType, value) { + if (value < 24) { + parts.push((majorType << 5) | value) + } else if (value < 256) { + parts.push((majorType << 5) | 24, value) + } else if (value < 65536) { + parts.push((majorType << 5) | 25, value >> 8, value & 0xff) + } + } + + function encodeCidLink(cidBytes) { + // DAG-CBOR CID link: tag(42) + byte string with 0x00 prefix + parts.push(0xd8, 42) // tag 42 + const withPrefix = new Uint8Array(cidBytes.length + 1) + withPrefix[0] = 0x00 // multibase identity prefix + withPrefix.set(cidBytes, 1) + encodeHead(2, withPrefix.length) + parts.push(...withPrefix) + } + + // Encode { roots: [...], version: 1 } + // Sort keys: "roots" (5 chars) comes after "version" (7 chars)? No - shorter first + // "roots" = 5 chars, "version" = 7 chars, so "roots" first + encodeHead(5, 2) // map with 2 entries + + // Key "roots" + const rootsKey = new TextEncoder().encode('roots') + encodeHead(3, rootsKey.length) + parts.push(...rootsKey) + + // Value: array of CID links + encodeHead(4, obj.roots.length) + for (const cid of obj.roots) { + encodeCidLink(cid) + } + + // Key "version" + const versionKey = new TextEncoder().encode('version') + encodeHead(3, versionKey.length) + parts.push(...versionKey) + + // Value: 1 + parts.push(0x01) + + return new Uint8Array(parts) +} + +export function buildCarFile(rootCid, blocks) { const parts = [] // Header: { version: 1, roots: [rootCid] } + // CIDs in header must be DAG-CBOR links (tag 42 + 0x00 prefix + CID bytes) const rootCidBytes = cidToBytes(rootCid) - const header = cborEncode({ version: 1, roots: [rootCidBytes] }) + const header = cborEncodeCarHeader({ version: 1, roots: [rootCidBytes] }) parts.push(varint(header.length)) parts.push(header) @@ -508,6 +805,60 @@ return importPrivateKey(hexToBytes(hex)) } + // Collect MST node blocks for a given root CID + collectMstBlocks(rootCidStr) { + const blocks = [] + const visited = new Set() + + const collect = (cidStr) => { + if (visited.has(cidStr)) return + visited.add(cidStr) + + const rows = this.sql.exec( + `SELECT data FROM blocks WHERE cid = ?`, cidStr + ).toArray() + if (rows.length === 0) return + + const data = new Uint8Array(rows[0].data) + blocks.push({ cid: cidStr, data }) // Keep as string, buildCarFile will convert + + // Decode and follow child CIDs (MST nodes have 'l' and 'e' with 't' subtrees) + try { + const node = cborDecode(data) + if (node.l) collect(cidToString(node.l)) + if (node.e) { + for (const entry of node.e) { + if (entry.t) collect(cidToString(entry.t)) + } + } + } catch (e) { + // Not an MST node, ignore + } + } + + collect(rootCidStr) + return blocks + } + + // Build CAR-style block bytes (without header, just block entries) + buildBlocksBytes(blocks) { + const parts = [] + for (const block of blocks) { + const cidBytes = block.cid instanceof Uint8Array ? block.cid : cidToBytes(block.cid) + const blockLen = cidBytes.length + block.data.length + // Varint encode the length + let len = blockLen + while (len >= 0x80) { + parts.push((len & 0x7f) | 0x80) + len >>= 7 + } + parts.push(len) + parts.push(...cidBytes) + parts.push(...block.data) + } + return new Uint8Array(parts) + } + async createRecord(collection, record, rkey = null) { const did = await this.getDid() if (!did) throw new Error('PDS not initialized') @@ -544,21 +895,22 @@ // Create commit const rev = createTid() + // Build commit with CIDs wrapped in CID class (for dag-cbor tag 42 encoding) const commit = { did, version: 3, - data: dataRoot, + data: new CID(cidToBytes(dataRoot)), // CID wrapped for explicit encoding rev, - prev: prevCommit?.cid || null + prev: prevCommit?.cid ? new CID(cidToBytes(prevCommit.cid)) : null } - // Sign commit - const commitBytes = cborEncode(commit) + // Sign commit (using dag-cbor encoder for CIDs) + const commitBytes = cborEncodeDagCbor(commit) const signingKey = await this.getSigningKey() const sig = await sign(signingKey, commitBytes) const signedCommit = { ...commit, sig } - const signedBytes = cborEncode(signedCommit) + const signedBytes = cborEncodeDagCbor(signedCommit) const commitCid = await createCid(signedBytes) const commitCidStr = cidToString(commitCid) @@ -574,21 +926,54 @@ commitCidStr, rev, prevCommit?.cid || null ) - // Sequence event + // Update head and rev for listRepos + await this.state.storage.put('head', commitCidStr) + await this.state.storage.put('rev', rev) + + // Collect blocks for the event (record + commit + MST nodes) + // Build a mini CAR with just the new blocks - use string CIDs + const newBlocks = [] + // Add record block + newBlocks.push({ cid: recordCidStr, data: recordBytes }) + // Add commit block + newBlocks.push({ cid: commitCidStr, data: signedBytes }) + // Add MST node blocks (get all blocks referenced by commit.data) + const mstBlocks = this.collectMstBlocks(dataRoot) + newBlocks.push(...mstBlocks) + + // Sequence event with blocks - store complete event data including rev and time + // blocks must be a full CAR file with header (roots = [commitCid]) + const eventTime = new Date().toISOString() const evt = cborEncode({ - ops: [{ action: 'create', path: `${collection}/${rkey}`, cid: recordCidStr }] + ops: [{ action: 'create', path: `${collection}/${rkey}`, cid: recordCidStr }], + blocks: buildCarFile(commitCidStr, newBlocks), // Full CAR with header + rev, // Store the actual commit revision + time: eventTime // Store the actual event time }) this.sql.exec( `INSERT INTO seq_events (did, commit_cid, evt) VALUES (?, ?, ?)`, did, commitCidStr, evt ) - // Broadcast to subscribers + // Broadcast to subscribers (both local and via default DO for relay) const evtRows = this.sql.exec( `SELECT * FROM seq_events ORDER BY seq DESC LIMIT 1` ).toArray() if (evtRows.length > 0) { this.broadcastEvent(evtRows[0]) + // Also forward to default DO for relay subscribers + if (this.env?.PDS) { + const defaultId = this.env.PDS.idFromName('default') + const defaultPds = this.env.PDS.get(defaultId) + // Convert ArrayBuffer to array for JSON serialization + const row = evtRows[0] + const evtArray = Array.from(new Uint8Array(row.evt)) + // Fire and forget but log errors + defaultPds.fetch(new Request('http://internal/forward-event', { + method: 'POST', + body: JSON.stringify({ ...row, evt: evtArray }) + })).then(r => r.json()).then(r => console.log('forward result:', r)).catch(e => console.log('forward error:', e)) + } } return { uri, cid: recordCidStr, commit: commitCidStr } @@ -596,19 +981,30 @@ formatEvent(evt) { // AT Protocol frame format: header + body + // Use DAG-CBOR encoding for body (CIDs need tag 42 + 0x00 prefix) const header = cborEncode({ op: 1, t: '#commit' }) - const body = cborEncode({ + + // Decode stored event to get ops, blocks, rev, and time + const evtData = cborDecode(new Uint8Array(evt.evt)) + const ops = evtData.ops.map(op => ({ + ...op, + cid: op.cid ? new CID(cidToBytes(op.cid)) : null // Wrap in CID class for tag 42 encoding + })) + // Get blocks from stored event (already in CAR format) + const blocks = evtData.blocks || new Uint8Array(0) + + const body = cborEncodeDagCbor({ seq: evt.seq, rebase: false, tooBig: false, repo: evt.did, - commit: cidToBytes(evt.commit_cid), - rev: createTid(), + commit: new CID(cidToBytes(evt.commit_cid)), // Wrap in CID class for tag 42 encoding + rev: evtData.rev, // Use stored rev from commit creation since: null, - blocks: new Uint8Array(0), // Simplified - real impl includes CAR slice - ops: cborDecode(new Uint8Array(evt.evt)).ops, + blocks: blocks instanceof Uint8Array ? blocks : new Uint8Array(blocks), + ops, blobs: [], - time: new Date().toISOString() + time: evtData.time // Use stored time from event creation }) // Concatenate header + body @@ -643,7 +1039,12 @@ // Handle resolution - doesn't require ?did= param if (url.pathname === '/.well-known/atproto-did') { - const did = await this.getDid() + let did = await this.getDid() + // If no DID on this instance, check registered DIDs (default instance) + if (!did) { + const registeredDids = await this.state.storage.get('registeredDids') || [] + did = registeredDids[0] + } if (!did) { return new Response('User not found', { status: 404 }) } @@ -666,6 +1067,72 @@ initialized: !!did, did: did || null }) + } + // Reset endpoint - clears all repo data but keeps identity + if (url.pathname === '/reset-repo') { + this.sql.exec(`DELETE FROM blocks`) + this.sql.exec(`DELETE FROM records`) + this.sql.exec(`DELETE FROM commits`) + this.sql.exec(`DELETE FROM seq_events`) + await this.state.storage.delete('head') + await this.state.storage.delete('rev') + return Response.json({ ok: true, message: 'repo data cleared' }) + } + // Internal endpoint to forward events for relay broadcasting + if (url.pathname === '/forward-event') { + const evt = await request.json() + // Convert evt back to proper format and broadcast + const numSockets = [...this.state.getWebSockets()].length + console.log(`forward-event: received event seq=${evt.seq}, ${numSockets} connected sockets`) + this.broadcastEvent({ + seq: evt.seq, + did: evt.did, + commit_cid: evt.commit_cid, + evt: new Uint8Array(Object.values(evt.evt)) + }) + return Response.json({ ok: true, sockets: numSockets }) + } + // Internal endpoint to register DIDs for discovery + if (url.pathname === '/register-did') { + const body = await request.json() + const registeredDids = await this.state.storage.get('registeredDids') || [] + if (!registeredDids.includes(body.did)) { + registeredDids.push(body.did) + await this.state.storage.put('registeredDids', registeredDids) + } + return Response.json({ ok: true }) + } + // Internal endpoint to get registered DIDs + if (url.pathname === '/get-registered-dids') { + const registeredDids = await this.state.storage.get('registeredDids') || [] + return Response.json({ dids: registeredDids }) + } + // Internal endpoint to get repo info (head/rev) + if (url.pathname === '/repo-info') { + const head = await this.state.storage.get('head') + const rev = await this.state.storage.get('rev') + return Response.json({ head: head || null, rev: rev || null }) + } + if (url.pathname === '/xrpc/com.atproto.server.describeServer') { + // Server DID should be did:web based on hostname, passed via header + const hostname = request.headers.get('x-hostname') || 'localhost' + return Response.json({ + did: `did:web:${hostname}`, + availableUserDomains: [`.${hostname}`], + inviteCodeRequired: false, + phoneVerificationRequired: false, + links: {}, + contact: {} + }) + } + if (url.pathname === '/xrpc/com.atproto.sync.listRepos') { + const registeredDids = await this.state.storage.get('registeredDids') || [] + // If this is the default instance, return registered DIDs + // If this is a user instance, return its own DID + const did = await this.getDid() + const repos = did ? [{ did, head: null, rev: null }] : + registeredDids.map(d => ({ did: d, head: null, rev: null })) + return Response.json({ repos }) } if (url.pathname === '/xrpc/com.atproto.repo.createRecord') { if (request.method !== 'POST') { @@ -709,6 +1176,34 @@ return Response.json({ uri, cid: row.cid, value }) } + if (url.pathname === '/xrpc/com.atproto.sync.getLatestCommit') { + const commits = this.sql.exec( + `SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1` + ).toArray() + + if (commits.length === 0) { + return Response.json({ error: 'RepoNotFound', message: 'repo not found' }, { status: 404 }) + } + + return Response.json({ cid: commits[0].cid, rev: commits[0].rev }) + } + if (url.pathname === '/xrpc/com.atproto.sync.getRepoStatus') { + const did = await this.getDid() + const commits = this.sql.exec( + `SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1` + ).toArray() + + if (commits.length === 0 || !did) { + return Response.json({ error: 'RepoNotFound', message: 'repo not found' }, { status: 404 }) + } + + return Response.json({ + did, + active: true, + status: 'active', + rev: commits[0].rev + }) + } if (url.pathname === '/xrpc/com.atproto.sync.getRepo') { const commits = this.sql.exec( `SELECT cid FROM commits ORDER BY seq DESC LIMIT 1` @@ -719,7 +1214,6 @@ } const blocks = this.sql.exec(`SELECT cid, data FROM blocks`).toArray() - // Convert ArrayBuffer data to Uint8Array const blocksForCar = blocks.map(b => ({ cid: b.cid, data: new Uint8Array(b.data) @@ -762,16 +1256,45 @@ async fetch(request, env) { const url = new URL(request.url) - // For /.well-known/atproto-did, extract DID from subdomain - // e.g., alice.atproto-pds.chad-53c.workers.dev -> look up "alice" - if (url.pathname === '/.well-known/atproto-did') { - const host = request.headers.get('Host') || '' - // For now, use the first Durable Object (single-user PDS) - // Extract handle from subdomain if present + // Endpoints that don't require ?did= param (for relay/federation) + if (url.pathname === '/.well-known/atproto-did' || + url.pathname === '/xrpc/com.atproto.server.describeServer') { const did = url.searchParams.get('did') || 'default' const id = env.PDS.idFromName(did) const pds = env.PDS.get(id) - return pds.fetch(request) + // Pass hostname for describeServer + const newReq = new Request(request.url, { + method: request.method, + headers: { ...Object.fromEntries(request.headers), 'x-hostname': url.hostname } + }) + return pds.fetch(newReq) + } + + // subscribeRepos WebSocket - route to default instance for firehose + if (url.pathname === '/xrpc/com.atproto.sync.subscribeRepos') { + const defaultId = env.PDS.idFromName('default') + const defaultPds = env.PDS.get(defaultId) + return defaultPds.fetch(request) + } + + // listRepos needs to aggregate from all registered DIDs + if (url.pathname === '/xrpc/com.atproto.sync.listRepos') { + const defaultId = env.PDS.idFromName('default') + const defaultPds = env.PDS.get(defaultId) + const regRes = await defaultPds.fetch(new Request('http://internal/get-registered-dids')) + const { dids } = await regRes.json() + + const repos = [] + for (const did of dids) { + const id = env.PDS.idFromName(did) + const pds = env.PDS.get(id) + const infoRes = await pds.fetch(new Request('http://internal/repo-info')) + const info = await infoRes.json() + if (info.head) { + repos.push({ did, head: info.head, rev: info.rev, active: true }) + } + } + return Response.json({ repos, cursor: undefined }) } const did = url.searchParams.get('did') @@ -779,15 +1302,30 @@ return new Response('missing did param', { status: 400 }) } + // On init, also register this DID with the default instance + if (url.pathname === '/init' && request.method === 'POST') { + const body = await request.json() + + // Register with default instance for discovery + const defaultId = env.PDS.idFromName('default') + const defaultPds = env.PDS.get(defaultId) + await defaultPds.fetch(new Request('http://internal/register-did', { + method: 'POST', + body: JSON.stringify({ did }) + })) + + // Forward to the actual PDS instance + const id = env.PDS.idFromName(did) + const pds = env.PDS.get(id) + return pds.fetch(new Request(request.url, { + method: 'POST', + headers: request.headers, + body: JSON.stringify(body) + })) + } + const id = env.PDS.idFromName(did) const pds = env.PDS.get(id) return pds.fetch(request) } -} - -// Export utilities for testing -export { - cborEncode, cborDecode, createCid, cidToString, base32Encode, createTid, - generateKeyPair, importPrivateKey, sign, bytesToHex, hexToBytes, - getKeyDepth, varint, base32Decode, buildCarFile } diff --git a/test/pds.test.js b/test/pds.test.js --- a/test/pds.test.js +++ b/test/pds.test.js @@ -1,7 +1,7 @@ import { test, describe } from 'node:test' import assert from 'node:assert' import { - cborEncode, cborDecode, createCid, cidToString, base32Encode, createTid, + cborEncode, cborDecode, createCid, cidToString, cidToBytes, base32Encode, createTid, generateKeyPair, importPrivateKey, sign, bytesToHex, hexToBytes, getKeyDepth, varint, base32Decode, buildCarFile } from '../src/pds.js' @@ -68,6 +68,31 @@ assert.deepStrictEqual(encoded1, encoded2) // First key should be 'a' (0x61) assert.strictEqual(encoded1[1], 0x61) + }) + + test('encodes large integers >= 2^31 without overflow', () => { + // 2^31 would overflow with bitshift operators (treated as signed 32-bit) + const twoTo31 = 2147483648 + const encoded = cborEncode(twoTo31) + const decoded = cborDecode(encoded) + assert.strictEqual(decoded, twoTo31) + + // 2^32 - 1 (max unsigned 32-bit) + const maxU32 = 4294967295 + const encoded2 = cborEncode(maxU32) + const decoded2 = cborDecode(encoded2) + assert.strictEqual(decoded2, maxU32) + }) + + test('encodes 2^31 with correct byte format', () => { + // 2147483648 = 0x80000000 + // CBOR: major type 0 (unsigned int), additional info 26 (4-byte follows) + const encoded = cborEncode(2147483648) + assert.strictEqual(encoded[0], 0x1a) // type 0 | info 26 + assert.strictEqual(encoded[1], 0x80) + assert.strictEqual(encoded[2], 0x00) + assert.strictEqual(encoded[3], 0x00) + assert.strictEqual(encoded[4], 0x00) }) }) @@ -183,40 +208,80 @@ assert.strictEqual(hex, '000ff0ffabcd') assert.deepStrictEqual(back, original) }) + + test('importPrivateKey rejects invalid key lengths', async () => { + // Too short + await assert.rejects( + () => importPrivateKey(new Uint8Array(31)), + /expected 32 bytes, got 31/ + ) + + // Too long + await assert.rejects( + () => importPrivateKey(new Uint8Array(33)), + /expected 32 bytes, got 33/ + ) + + // Empty + await assert.rejects( + () => importPrivateKey(new Uint8Array(0)), + /expected 32 bytes, got 0/ + ) + }) + + test('importPrivateKey rejects non-Uint8Array input', async () => { + // Arrays have .length but aren't Uint8Array + await assert.rejects( + () => importPrivateKey([1, 2, 3]), + /Invalid private key/ + ) + + // Strings don't work either + await assert.rejects( + () => importPrivateKey('not bytes'), + /Invalid private key/ + ) + + // null/undefined + await assert.rejects( + () => importPrivateKey(null), + /Invalid private key/ + ) + }) }) describe('MST Key Depth', () => { - test('returns a non-negative integer', () => { - const depth = getKeyDepth('app.bsky.feed.post/abc123') + test('returns a non-negative integer', async () => { + const depth = await getKeyDepth('app.bsky.feed.post/abc123') assert.strictEqual(typeof depth, 'number') assert.ok(depth >= 0) }) - test('is deterministic for same key', () => { + test('is deterministic for same key', async () => { const key = 'app.bsky.feed.post/test123' - const depth1 = getKeyDepth(key) - const depth2 = getKeyDepth(key) + const depth1 = await getKeyDepth(key) + const depth2 = await getKeyDepth(key) assert.strictEqual(depth1, depth2) }) - test('different keys can have different depths', () => { + test('different keys can have different depths', async () => { // Generate many keys and check we get some variation const depths = new Set() for (let i = 0; i < 100; i++) { - depths.add(getKeyDepth(`collection/key${i}`)) + depths.add(await getKeyDepth(`collection/key${i}`)) } // Should have at least 1 unique depth (realistically more) assert.ok(depths.size >= 1) }) - test('handles empty string', () => { - const depth = getKeyDepth('') + test('handles empty string', async () => { + const depth = await getKeyDepth('') assert.strictEqual(typeof depth, 'number') assert.ok(depth >= 0) }) - test('handles unicode strings', () => { - const depth = getKeyDepth('app.bsky.feed.post/émoji🎉') + test('handles unicode strings', async () => { + const depth = await getKeyDepth('app.bsky.feed.post/émoji🎉') assert.strictEqual(typeof depth, 'number') assert.ok(depth >= 0) }) -- tangled.sh