Skip to content

Commit 45b0519

Browse files
steve02081504steve02081504
committed
feat(p2p): Phase 3 gossip 协议与 checkpoint 签名验证
- dag.mjs：新增 gossip_request / gossip_response MQTT action 在 requestMissingEventsGossip 缺失事件时非阻塞向联邦广播 gossip 处理请求时本地查找后响应，TTL>0 时转发（最大跳数 2） 以 NODE_ID 防环，对同一请求去重避免广播放大 - checkpoint.mjs：新增 signCheckpoint / verifyCheckpointSignature 用 canonicalStringify + Ed25519 对 checkpoint payload 签名 buildCheckpointPayload 增加 eventIdsInEpoch 供 Merkle 验证 - 新增 checkpointVerifier.mjs：新节点 checkpoint 验证流程 验证 epoch_root_hash 由 Merkle 正确计算 验证 owner_signature（如存在） 验证 epoch_chain 回溯一致性 Made-with: Cursor
1 parent 64f2a45 commit 45b0519

3 files changed

Lines changed: 226 additions & 13 deletions

File tree

‎src/public/parts/shells/chat/src/chat/checkpointVerifier.mjs‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
import { verifyCheckpointSignature } from '../../../../../../scripts/p2p/checkpoint.mjs'
2+
import { merkleRoot } from '../../../../../../scripts/p2p/dag.mjs'
3+
4+
const HASH64 = /^[0-9a-f]{64}$/iu
5+
6+
/**
7+
* @param {object} checkpoint 远端 checkpoint 对象
8+
* @param {Uint8Array} ownerPubKey 期望的群主公钥（验签用）
9+
* @returns {Promise<{ valid: boolean, reason?: string }>} 校验结果与失败原因
10+
*/
11+
export async function verifyRemoteCheckpoint(checkpoint, ownerPubKey) {
12+
if (!checkpoint || typeof checkpoint !== 'object')
13+
return { valid: false, reason: 'checkpoint missing or not an object' }
14+
15+
const ids = checkpoint.eventIdsInEpoch
16+
if (!Array.isArray(ids) || !ids.length)
17+
return { valid: false, reason: 'eventIdsInEpoch missing or empty' }
18+
if (!ids.every(id => typeof id === 'string' && HASH64.test(id)))
19+
return { valid: false, reason: 'eventIdsInEpoch contains invalid event id' }
20+
21+
const expectedRoot = merkleRoot(ids)
22+
if (checkpoint.epoch_root_hash !== expectedRoot)
23+
return { valid: false, reason: 'epoch_root_hash does not match Merkle root of eventIdsInEpoch' }
24+
25+
if (checkpoint.owner_signature != null && checkpoint.owner_signature !== '') {
26+
if (!(ownerPubKey instanceof Uint8Array) || ownerPubKey.length !== 32)
27+
return { valid: false, reason: 'ownerPubKey required for signed checkpoint' }
28+
const ok = await verifyCheckpointSignature(checkpoint, ownerPubKey)
29+
if (!ok) return { valid: false, reason: 'owner_signature verification failed' }
30+
}
31+
32+
const curEpoch = checkpoint.epoch_id
33+
if (typeof curEpoch !== 'number' || !Number.isFinite(curEpoch) || curEpoch <= 0 || curEpoch !== Math.floor(curEpoch))
34+
return { valid: false, reason: 'epoch_id invalid' }
35+
36+
const chain = checkpoint.epoch_chain
37+
if (chain != null) {
38+
if (!Array.isArray(chain))
39+
return { valid: false, reason: 'epoch_chain is not an array' }
40+
let prev = -Infinity
41+
for (let i = 0; i < chain.length; i++) {
42+
const e = chain[i]
43+
if (!e || typeof e !== 'object')
44+
return { valid: false, reason: 'epoch_chain entry invalid' }
45+
const eid = e.epoch_id
46+
const erh = e.epoch_root_hash
47+
const cid = e.checkpoint_event_id
48+
if (typeof eid !== 'number' || !Number.isFinite(eid) || eid <= 0 || eid !== Math.floor(eid))
49+
return { valid: false, reason: 'epoch_chain epoch_id invalid' }
50+
if (typeof erh !== 'string' || !HASH64.test(erh))
51+
return { valid: false, reason: 'epoch_chain epoch_root_hash invalid' }
52+
if (typeof cid !== 'string' || !HASH64.test(cid))
53+
return { valid: false, reason: 'epoch_chain checkpoint_event_id invalid' }
54+
if (eid <= prev) return { valid: false, reason: 'epoch_chain epoch_id not strictly increasing' }
55+
prev = eid
56+
}
57+
if (chain.length && curEpoch < chain[chain.length - 1].epoch_id)
58+
return { valid: false, reason: 'epoch_id regresses relative to epoch_chain tail' }
59+
}
60+
61+
return { valid: true }
62+
}

‎src/public/parts/shells/chat/src/chat/dag.mjs‎

Lines changed: 129 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import { access, mkdir, appendFile, readFile, readdir, rm, writeFile } from 'nod
44
import { dirname, join } from 'node:path'
55

66
import { geti18n } from '../../../../../../scripts/i18n.mjs'
7-
import { buildCheckpointPayload, buildFileFoldersSnapshot } from '../../../../../../scripts/p2p/checkpoint.mjs'
7+
import { buildCheckpointPayload, buildFileFoldersSnapshot, signCheckpoint } from '../../../../../../scripts/p2p/checkpoint.mjs'
88
import { DEFAULT_MAX_CATCHUP_EVENTS, EPOCH_CHAIN_MAX } from '../../../../../../scripts/p2p/constants.mjs'
99
import { pubKeyHash, sign, verify } from '../../../../../../scripts/p2p/crypto.mjs'
1010
import {
@@ -117,11 +117,30 @@ function getFederationConfig(username) {
117117
return { enabled, appId, password }
118118
}
119119

120-
/** @type {Map<string, Promise<{ room: any, sendDag: (payload: unknown, peerId: string | null) => void } | null>>} */
120+
/** @type {Map<string, Promise<{ room: any, sendDag: (payload: unknown, peerId: string | null) => void, sendGossipRequest: (payload: unknown, peerId: string | null) => void, sendGossipResponse: (payload: unknown, peerId: string | null) => void } | null>>} */
121121
const federationRoomInflight = new Map()
122-
/** @type {Map<string, { room: any, sendDag: (payload: unknown, peerId: string | null) => void } | null>} */
122+
/** @type {Map<string, { room: any, sendDag: (payload: unknown, peerId: string | null) => void, sendGossipRequest: (payload: unknown, peerId: string | null) => void, sendGossipResponse: (payload: unknown, peerId: string | null) => void } | null>} */
123123
const federationRooms = new Map()
124124

125+
const gossipRequestDedupe = new Map()
126+
const GOSSIP_DEDUPE_MS = 30_000
127+
128+
/**
129+
* @param {string} dedupeKey 去重键（请求者、wantIds、ttl）
130+
* @returns {boolean} 若为首次处理则返回 true
131+
*/
132+
function takeGossipRequestSlot(dedupeKey) {
133+
const now = Date.now()
134+
if (gossipRequestDedupe.size > 2000)
135+
for (const [k, t] of gossipRequestDedupe)
136+
if (t < now - GOSSIP_DEDUPE_MS) gossipRequestDedupe.delete(k)
137+
138+
139+
if (gossipRequestDedupe.has(dedupeKey)) return false
140+
gossipRequestDedupe.set(dedupeKey, now)
141+
return true
142+
}
143+
125144
/**
126145
* 联邦 MQTT 连接在进程内缓存使用的复合键。
127146
* @param {string} username 用户名
@@ -136,7 +155,7 @@ function federationRoomKey(username, chatId) {
136155
* 按需加入 `fount-fed-${chatId}` MQTT 房间并订阅 `dag_event`；未启用或失败时返回 null。
137156
* @param {string} username 用户名
138157
* @param {string} chatId 群组 ID
139-
* @returns {Promise<{ room: any, sendDag: (payload: unknown, peerId: string | null) => void } | null>} 房间句柄与 DAG 发送函数，或 null
158+
* @returns {Promise<{ room: any, sendDag: (payload: unknown, peerId: string | null) => void, sendGossipRequest: (payload: unknown, peerId: string | null) => void, sendGossipResponse: (payload: unknown, peerId: string | null) => void } | null>} 房间句柄与 DAG 发送函数，或 null
140159
*/
141160
async function ensureFederationRoom(username, chatId) {
142161
const { enabled, appId, password } = getFederationConfig(username)
@@ -158,7 +177,48 @@ async function ensureFederationRoom(username, chatId) {
158177
getDag((data, _peerId) => {
159178
void ingestRemoteEvent(username, chatId, data).catch(e => console.error(e))
160179
})
161-
const slot = { room, sendDag }
180+
const [sendGossipRequest, getGossipRequest] = room.makeAction('gossip_request')
181+
const [sendGossipResponse, getGossipResponse] = room.makeAction('gossip_response')
182+
getGossipRequest((data, peerId) => {
183+
void (async () => {
184+
if (!data || typeof data !== 'object') return
185+
const rawWant = /** @type {{ wantIds?: unknown, ttl?: unknown, requesterId?: unknown }} */ data.wantIds
186+
const wantIds = Array.isArray(rawWant)
187+
? [...new Set(rawWant.filter(id => typeof id === 'string' && /^[0-9a-f]{64}$/iu.test(id)))]
188+
: []
189+
if (!wantIds.length) return
190+
const ttl = Number(/** @type {{ ttl?: unknown }} */ data.ttl)
191+
const requesterId = /** @type {{ requesterId?: unknown }} */ data.requesterId
192+
if (!Number.isFinite(ttl) || typeof requesterId !== 'string' || !requesterId) return
193+
if (requesterId === NODE_ID) return
194+
const dedupeKey = `${requesterId}\0${wantIds.slice().sort().join(',')}\0${ttl}`
195+
if (!takeGossipRequestSlot(dedupeKey)) return
196+
const path = eventsPath(username, chatId)
197+
const prev = await readJsonl(path)
198+
const byId = new Map(prev.map(e => [e.id, e]))
199+
const events = wantIds.map(id => byId.get(id)).filter(Boolean)
200+
if (events.length && peerId)
201+
try {
202+
sendGossipResponse({ events, requesterId }, peerId)
203+
}
204+
catch (e) {
205+
console.error('federation: gossip_response failed', e)
206+
}
207+
208+
if (ttl > 0)
209+
try {
210+
sendGossipRequest({ wantIds, ttl: ttl - 1, requesterId }, null)
211+
}
212+
catch (e) {
213+
console.error('federation: gossip_request forward failed', e)
214+
}
215+
216+
})().catch(e => console.error(e))
217+
})
218+
getGossipResponse((data, _peerId) => {
219+
void handleGossipResponse(username, chatId, data).catch(e => console.error(e))
220+
})
221+
const slot = { room, sendDag, sendGossipRequest, sendGossipResponse }
162222
federationRooms.set(key, slot)
163223
return slot
164224
}
@@ -225,7 +285,7 @@ async function appendValidatedRemoteEvent(username, chatId, signPayload, opts =
225285
}
226286

227287
await appendJsonl(path, signPayload)
228-
await broadcastAndPersist(username, chatId, /** @type {object} */ signPayload)
288+
await broadcastAndPersist(username, chatId, /** @type {object} */ signPayload, {})
229289
return 'ok'
230290
}
231291

@@ -246,6 +306,29 @@ async function ingestRemoteEvent(username, chatId, payload) {
246306
await appendValidatedRemoteEvent(username, chatId, /** @type {object} */ signPayload, { logFailures: true })
247307
}
248308

309+
/**
310+
* @param {string} username 用户名
311+
* @param {string} chatId 群组 ID
312+
* @param {unknown} data `gossip_response` 载荷
313+
*/
314+
async function handleGossipResponse(username, chatId, data) {
315+
if (!data || typeof data !== 'object') return
316+
const requesterId = /** @type {{ requesterId?: unknown }} */ data.requesterId
317+
if (requesterId !== NODE_ID) return
318+
const rawList = /** @type {{ events?: unknown }} */ data.events
319+
if (!Array.isArray(rawList)) return
320+
for (const rawEv of rawList) {
321+
/** @type {unknown} */
322+
let ev = rawEv
323+
if (rawEv && typeof rawEv === 'object' && 'event' in /** @type {object} */ rawEv) {
324+
const wrapped = /** @type {{ event?: object }} */ rawEv.event
325+
if (wrapped && typeof wrapped === 'object') ev = wrapped
326+
}
327+
if (!ev || typeof ev !== 'object') continue
328+
await appendValidatedRemoteEvent(username, chatId, /** @type {object} */ ev, { logFailures: false })
329+
}
330+
}
331+
249332
// ─── 写入频道消息流的事件类型 ─────────────────────────────────────────────────
250333

251334
const PERSIST_MESSAGE_TYPES = new Set([
@@ -259,12 +342,13 @@ const PERSIST_MESSAGE_TYPES = new Set([
259342
* @param {string} username 用户名
260343
* @param {string} chatId 群组 ID
261344
* @param {object} signPayload 已持久化的签名事件对象
345+
* @param {{ checkpointOwnerSecretKey?: Uint8Array }} [persistOpts] 可选群主私钥供 checkpoint 签名
262346
* @returns {Promise<void>} 广播与派生持久化完成
263347
*/
264-
async function broadcastAndPersist(username, chatId, signPayload) {
348+
async function broadcastAndPersist(username, chatId, signPayload, persistOpts = {}) {
265349
broadcastEvent(chatId, { type: 'dag_event', event: signPayload })
266350
if (!PERSIST_MESSAGE_TYPES.has(signPayload.type)) {
267-
await rebuildAndSaveCheckpoint(username, chatId)
351+
await rebuildAndSaveCheckpoint(username, chatId, persistOpts)
268352
return
269353
}
270354
const ch = signPayload.channelId || signPayload.content?.channelId || 'default'
@@ -281,7 +365,7 @@ async function broadcastAndPersist(username, chatId, signPayload) {
281365
await mkdir(join(mp, '..'), { recursive: true })
282366
await appendFile(mp, `${JSON.stringify(msgLine)}\n`, 'utf8')
283367
broadcastEvent(chatId, { type: 'channel_message', channelId: ch, message: msgLine })
284-
await rebuildAndSaveCheckpoint(username, chatId)
368+
await rebuildAndSaveCheckpoint(username, chatId, persistOpts)
285369
if (signPayload.type === 'message')
286370
void maybeAutoTriggerCharReply(username, chatId, ch).catch(() => {})
287371
}
@@ -558,9 +642,10 @@ function foldPinOverlay(events) {
558642
* 重放 DAG 授权类事件并写回 `checkpoint.json`。
559643
* @param {string} username 用户名
560644
* @param {string} chatId 群组 ID
645+
* @param {{ checkpointOwnerSecretKey?: Uint8Array }} [opts] 可选群主私钥
561646
* @returns {Promise<object | null>} 新生成的检查点对象；无事件时返回 `null`
562647
*/
563-
export async function rebuildAndSaveCheckpoint(username, chatId) {
648+
export async function rebuildAndSaveCheckpoint(username, chatId, opts = {}) {
564649
const { events, state, order } = await getState(username, chatId, { forceFullReplay: true })
565650
if (!events.length) return null
566651
const last = events[events.length - 1]
@@ -608,7 +693,7 @@ export async function rebuildAndSaveCheckpoint(username, chatId) {
608693
const pins = foldPinOverlay(events)
609694
const fileIdx = Object.fromEntries(state.fileIndex ?? new Map())
610695
const fileFolders = buildFileFoldersSnapshot(state.fileIndex)
611-
const cp = buildCheckpointPayload({
696+
let cp = buildCheckpointPayload({
612697
home_node_id: home,
613698
materialized: state,
614699
epoch_id,
@@ -618,11 +703,28 @@ export async function rebuildAndSaveCheckpoint(username, chatId) {
618703
fileFolders,
619704
epoch_chain,
620705
})
706+
const sk = opts.checkpointOwnerSecretKey
707+
if (sk && await canUseSecretKeyForCheckpointSignature(state, sk))
708+
cp = await signCheckpoint(cp, sk)
621709
await mkdir(chatDir(username, chatId), { recursive: true })
622710
await writeFile(checkpointPath(username, chatId), JSON.stringify(cp, null, '\t'), 'utf8')
623711
return cp
624712
}
625713

714+
/**
715+
* @param {ReturnType<typeof emptyMaterializedState>} state 当前物化状态
716+
* @param {Uint8Array} secretKey 候选 Ed25519 私钥
717+
* @returns {Promise<boolean>} 是否可为 checkpoint 代群主签名
718+
*/
719+
async function canUseSecretKeyForCheckpointSignature(state, secretKey) {
720+
if (!secretKey || secretKey.length < 32) return false
721+
const { getPublicKey } = await import('npm:@noble/ed25519')
722+
const h = pubKeyHash(getPublicKey(secretKey.slice(0, 32)))
723+
const del = state.delegatedOwnerPubKeyHash
724+
if (del) return h === del
725+
return adminPubKeyHashes(state).has(h)
726+
}
727+
626728
// ─── DAG 事件追加 ─────────────────────────────────────────────────────────────
627729

628730
/**
@@ -726,7 +828,7 @@ export async function appendEvent(username, chatId, event, secretKey) {
726828
await validateSignature(username, chatId, body, signPayload, event, secretKey)
727829

728830
await appendJsonl(eventsPath(username, chatId), signPayload)
729-
await broadcastAndPersist(username, chatId, signPayload)
831+
await broadcastAndPersist(username, chatId, signPayload, { checkpointOwnerSecretKey: secretKey })
730832
await publishEventToFederation(username, chatId, signPayload)
731833

732834
return signPayload
@@ -1254,5 +1356,20 @@ export async function requestMissingEventsGossip(username, chatId, query = {}) {
12541356
const stillMissing = wantIds.filter(id => !byId.has(id))
12551357
const found = wantIds.length === 0 || stillMissing.length === 0
12561358

1359+
if (stillMissing.length)
1360+
void (async () => {
1361+
const slot = await ensureFederationRoom(username, chatId)
1362+
if (!slot?.sendGossipRequest) return
1363+
try {
1364+
slot.sendGossipRequest({ wantIds: [...stillMissing], ttl: 2, requesterId: NODE_ID }, null)
1365+
}
1366+
catch (e) {
1367+
console.error('federation: gossip_request failed', e)
1368+
return
1369+
}
1370+
await new Promise(resolve => setTimeout(resolve, 3000))
1371+
})().catch(e => console.error(e))
1372+
1373+
12571374
return { found, events: filled, stillMissing, mergedFromPeer }
12581375
}

‎src/scripts/p2p/checkpoint.mjs‎

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
1+
import { Buffer } from 'node:buffer'
2+
3+
import { canonicalStringify } from './canonical_json.mjs'
14
import { MEMBERS_PAGE_SIZE } from './constants.mjs'
5+
import { sign as edSign, verify as edVerify } from './crypto.mjs'
26
import { merkleRoot } from './dag.mjs'
37

48
/**
@@ -66,7 +70,7 @@ export function buildCheckpointPayload({
6670
const members_record = Object.fromEntries(
6771
[...materialized.members.entries()].map(([k, v]) => [k, {
6872
pubKeyHex: v.pubKeyHex,
69-
roles: [...(v.roles || [])],
73+
roles: [...v.roles || []],
7074
profile: v.profile,
7175
}]),
7276
)
@@ -88,11 +92,41 @@ export function buildCheckpointPayload({
8892
messageOverlay: overlay,
8993
checkpoint_event_id,
9094
epoch_id,
95+
eventIdsInEpoch,
9196
epoch_root_hash,
9297
epoch_chain,
9398
}
9499
}
95100

101+
/**
102+
* @param {object} payload buildCheckpointPayload 的返回值
103+
* @param {Uint8Array} ownerPrivKey 群主私钥
104+
* @returns {Promise<object>} 带 owner_signature 的 checkpoint
105+
*/
106+
export async function signCheckpoint(payload, ownerPrivKey) {
107+
const unsigned = { ...payload }
108+
delete unsigned.owner_signature
109+
const msg = new TextEncoder().encode(canonicalStringify(unsigned))
110+
const sig = await edSign(msg, ownerPrivKey)
111+
return { ...payload, owner_signature: Buffer.from(sig).toString('hex') }
112+
}
113+
114+
/**
115+
* @param {object} checkpoint 带 owner_signature 的 checkpoint
116+
* @param {Uint8Array} ownerPubKey 群主公钥
117+
* @returns {Promise<boolean>} 验签是否通过
118+
*/
119+
export async function verifyCheckpointSignature(checkpoint, ownerPubKey) {
120+
const sigHex = checkpoint?.owner_signature
121+
if (typeof sigHex !== 'string' || !sigHex.trim()) return false
122+
const sig = Buffer.from(sigHex.trim(), 'hex')
123+
if (sig.length !== 64) return false
124+
const unsigned = { ...checkpoint }
125+
delete unsigned.owner_signature
126+
const msg = new TextEncoder().encode(canonicalStringify(unsigned))
127+
return edVerify(new Uint8Array(sig), msg, ownerPubKey)
128+
}
129+
96130
/**
97131
* 将 channelPermissions Map 转为可 JSON 的普通对象
98132
*

0 commit comments

Comments
 (0)