zhuyuan-ice-gl-zy001/QODER-CN-DEV-20260807/state-wake-20260812/bridge/bridge-node.mjs
2026-08-20 00:01:53 +08:00

495 lines
20 KiB
JavaScript
Executable file
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env node
/**
* 光湖桥 · 独立参考实现 · guanghu-bridge/v1
*
* 定位:光湖桥通信层的独立运行时 + 协议参考实现(P3 最小闭环)。
* 不依赖任何桌面壳,不依赖第三方包,仅用 Node 内置模块。
*
* 分层(对齐 HLP-CURRENT-ARCH-001):
* 传输层 TCP + NDJSON 帧(可选 TLS 后续接)
* 会话层 X25519 密钥交换 + HKDF + AES-256-GCM,中转只见密文
* 消息层 信封 schema=guanghu.guanghu-bridge-message/v1
* 事实层 中转节点将信封指纹沉淀 facts/*.ndjson(耐久层入口,Git 可挂)
*
* 运行方式:
* node bridge-node.mjs relay --port 9701 --state-dir <dir>
* node bridge-node.mjs node --id <节点号> --relay host:port --state-dir <dir>
* node bridge-node.mjs selftest [--workdir <dir>]
*/
import net from 'node:net';
import fs from 'node:fs';
import path from 'node:path';
import readline from 'node:readline';
import process from 'node:process';
import {
generateKeyPairSync, diffieHellman, hkdfSync, randomBytes,
createCipheriv, createDecipheriv, createHash, createHmac,
createPublicKey, createPrivateKey, randomUUID,
} from 'node:crypto';
export const SCHEMA = 'guanghu.guanghu-bridge-message/v1';
const MAX_LINE = 1024 * 1024;
// ────────────────────────── 密钥与加解密 ──────────────────────────
function ensureDir(p) { fs.mkdirSync(p, { recursive: true }); return p; }
function loadOrCreateIdentity(nodeId, stateDir) {
const dir = ensureDir(path.join(stateDir, 'identities'));
const file = path.join(dir, `${nodeId}.key.json`);
if (fs.existsSync(file)) {
const j = JSON.parse(fs.readFileSync(file, 'utf8'));
return {
nodeId,
privateKey: createPrivateKey({ key: Buffer.from(j.priv_der_b64, 'base64'), format: 'der', type: 'pkcs8' }),
pubB64: j.pub_der_b64,
};
}
const { publicKey, privateKey } = generateKeyPairSync('x25519');
const pubDer = publicKey.export({ type: 'spki', format: 'der' });
const privDer = privateKey.export({ type: 'pkcs8', format: 'der' });
fs.writeFileSync(file, JSON.stringify({
schema: 'guanghu.guanghu-bridge-identity/v1',
node_id: nodeId,
pub_der_b64: pubDer.toString('base64'),
priv_der_b64: privDer.toString('base64'),
created_at: new Date().toISOString(),
}, null, 2));
return { nodeId, privateKey, pubB64: pubDer.toString('base64') };
}
function importPub(b64) {
return createPublicKey({ key: Buffer.from(b64, 'base64'), format: 'der', type: 'spki' });
}
function sharedIkm(privKey, peerPubB64) {
return diffieHellman({ privateKey: privKey, publicKey: importPub(peerPubB64) });
}
function deriveKeys(ikm) {
return {
msg: Buffer.from(hkdfSync('sha256', ikm, 'guanghu-bridge-v1', 'msg', 32)),
receipt: Buffer.from(hkdfSync('sha256', ikm, 'guanghu-bridge-v1', 'receipt', 32)),
};
}
function seal(key, plaintext, aad) {
const iv = randomBytes(12);
const c = createCipheriv('aes-256-gcm', key, iv);
c.setAAD(Buffer.from(aad));
const ct = Buffer.concat([c.update(plaintext, 'utf8'), c.final()]);
return { iv_b64: iv.toString('base64'), ct_b64: ct.toString('base64'), tag_b64: c.getAuthTag().toString('base64') };
}
function open(key, payload, aad) {
const d = createDecipheriv('aes-256-gcm', key, Buffer.from(payload.iv_b64, 'base64'));
d.setAAD(Buffer.from(aad));
d.setAuthTag(Buffer.from(payload.tag_b64, 'base64'));
return Buffer.concat([d.update(Buffer.from(payload.ct_b64, 'base64')), d.final()]).toString('utf8');
}
function signReceipt(key, msgId, from, to, ts) {
return createHmac('sha256', key).update(`${msgId}|${from}|${to}|${ts}`).digest('base64');
}
// ────────────────────────── NDJSON 帧 ──────────────────────────
function attachFraming(socket, onLine) {
let buf = '';
socket.on('data', (chunk) => {
buf += chunk.toString('utf8');
let idx;
while ((idx = buf.indexOf('\n')) >= 0) {
const line = buf.slice(0, idx).trim();
buf = buf.slice(idx + 1);
if (!line) continue;
if (line.length > MAX_LINE) { socket.destroy(); return; }
try { onLine(JSON.parse(line)); } catch { /* 坏帧丢弃,不断连 */ }
}
});
return (obj) => { if (!socket.destroyed) socket.write(JSON.stringify(obj) + '\n'); };
}
// ────────────────────────── 中转(会合点)──────────────────────────
export function startRelay({ port = 0, stateDir, log = () => {} }) {
ensureDir(stateDir);
const offlineDir = ensureDir(path.join(stateDir, 'offline'));
const factsDir = ensureDir(path.join(stateDir, 'facts'));
const registryFile = path.join(stateDir, 'relay-registry.json');
const registry = fs.existsSync(registryFile) ? JSON.parse(fs.readFileSync(registryFile, 'utf8')) : {};
const online = new Map(); // node_id -> send()
// 会合点自身身份:用户自有节点的公钥可信任,用于离线信封混合加密
const relayIdentity = loadOrCreateIdentity('__RELAY__', stateDir);
const saveRegistry = () => fs.writeFileSync(registryFile, JSON.stringify(registry, null, 2));
const factsFile = () => path.join(factsDir, `${new Date().toISOString().slice(0, 10)}.ndjson`);
function sinkFact(env) {
// 耐久事实层入口:只沉淀信封指纹与元数据,不落密文副本
fs.appendFileSync(factsFile(), JSON.stringify({
schema: SCHEMA, kind: 'FACT_SINK', ts: new Date().toISOString(),
msg_id: env.id, from: env.from, to: env.to,
envelope_sha256: createHash('sha256').update(JSON.stringify(env)).digest('hex'),
}) + '\n');
}
function reencryptForRecipient(env) {
// 离线信封原以会合点公钥加密;收件人上线后转成收件人公钥加密再投递,
// 中转仅在信封生命周期内短暂见到明文
if (!env.hybrid) return env;
const entry = registry[env.to];
if (!entry || !entry.pub) return null;
try {
const inKeys = deriveKeys(sharedIkm(relayIdentity.privateKey, env.from_pub));
const text = open(inKeys.msg, env.payload, env.id);
const outKeys = deriveKeys(sharedIkm(relayIdentity.privateKey, entry.pub));
const converted = { ...env, hybrid: false };
delete converted.from_pub;
converted.sealed_by = 'RELAY'; // 收件人据此选解密钥
converted.relay_pub = relayIdentity.pubB64;
converted.payload = seal(outKeys.msg, text, env.id);
return converted;
} catch (e) { log(`relay: 离线信封 ${env.id} 转加密失败(${e.message})`); return null; }
}
function deliver(env) {
const send = online.get(env.to);
if (send) {
const finalEnv = env.hybrid ? reencryptForRecipient(env) : env;
if (!finalEnv) {
fs.appendFileSync(path.join(offlineDir, `${env.to}.ndjson`), JSON.stringify(env) + '\n');
return false;
}
send(finalEnv); log(`relay: 投递 ${env.id} → ${env.to}(在线${env.hybrid ? '·转加密' : ''})`); return true;
}
fs.appendFileSync(path.join(offlineDir, `${env.to}.ndjson`), JSON.stringify(env) + '\n');
log(`relay: ${env.to} 离线,${env.id} 入离线队列`);
return false;
}
function flushOffline(nodeId) {
const file = path.join(offlineDir, `${nodeId}.ndjson`);
if (!fs.existsSync(file)) return 0;
const lines = fs.readFileSync(file, 'utf8').split('\n').filter(Boolean);
fs.unlinkSync(file);
let n = 0;
for (const line of lines) {
try { if (deliver(JSON.parse(line))) n++; } catch { /* 坏行跳过 */ }
}
return n;
}
const server = net.createServer((socket) => {
let peer = null;
const send = attachFraming(socket, (msg) => {
if (msg.type === 'HELLO') {
peer = msg.node_id;
registry[peer] = { pub: msg.pub, last_seen: new Date().toISOString() };
saveRegistry();
online.set(peer, send);
send({ schema: SCHEMA, type: 'HELLO_ACK', node_id: peer });
const n = flushOffline(peer);
log(`relay: 节点 ${peer} 接入${n ? `,补发 ${n} 条离线消息` : ''}`);
return;
}
if (!peer) return;
if (msg.type === 'SEND' && msg.envelope) {
const env = msg.envelope;
sinkFact(env); // 实时是光,事实层是湖底
deliver(env);
send({ schema: SCHEMA, type: 'RELAY_ACK', id: env.id });
return;
}
if (msg.type === 'RECEIPT' || msg.type === 'MSG') {
// RECEIPT 与转发类信封同样按 to 路由
deliver(msg);
return;
}
if (msg.type === 'GET_PUB') {
// 收件人未登记时回发会合点公钥(混合加密降级),发送端不阻塞
const entry = registry[msg.to];
send({
schema: SCHEMA, type: 'PUB', node_id: msg.to,
pub: entry ? entry.pub : relayIdentity.pubB64,
hybrid: !entry,
});
return;
}
});
socket.on('close', () => { if (peer && online.get(peer) === send) online.delete(peer); });
});
return new Promise((resolve) => {
server.listen(port, '127.0.0.1', () => {
const actual = server.address().port;
log(`relay: 会合点启动 127.0.0.1:${actual}`);
resolve({
port: actual, stateDir, registry,
relayPubB64: relayIdentity.pubB64,
onlineNodes: () => [...online.keys()],
close: () => new Promise((r) => server.close(r)),
});
});
});
}
// ────────────────────────── 节点 ──────────────────────────
export function startNode({ id, relayHost, relayPort, stateDir, log = () => {}, onMessage = () => {} }) {
ensureDir(stateDir);
const identity = loadOrCreateIdentity(id, stateDir);
const receiptsDir = ensureDir(path.join(stateDir, 'receipts', id));
const pubCache = new Map(); // node_id -> {pub, hybrid}
const inbox = [];
let sendFn = null;
const socket = net.connect(relayPort, relayHost);
async function ensurePeerPub(peerId) {
// 收到离线补发/陌生对端消息时,缓存里可能只有混合占位公钥,先拉真实公钥
let entry = pubCache.get(peerId);
if (entry && !entry.hybrid) return entry;
sendFn({ schema: SCHEMA, type: 'GET_PUB', to: peerId });
for (let i = 0; i < 100; i++) {
await new Promise((r) => setTimeout(r, 20));
entry = pubCache.get(peerId);
if (entry && !entry.hybrid) return entry;
}
return null;
}
sendFn = attachFraming(socket, (msg) => {
if (msg.type === 'HELLO_ACK') { log(`node ${id}: 已接入会合点`); return; }
if (msg.type === 'PUB') { if (msg.pub) pubCache.set(msg.node_id, { pub: msg.pub, hybrid: !!msg.hybrid }); return; }
if (msg.type === 'MSG') {
void (async () => {
const senderEntry = await ensurePeerPub(msg.from);
// sealed_by=RELAY:离线补发信封由会合点转加密,用会合点公钥解密;
// 其余用发件人公钥解密
const decPub = msg.sealed_by === 'RELAY' ? msg.relay_pub : (senderEntry && senderEntry.pub);
let text = null;
if (decPub) {
try {
const decKeys = deriveKeys(sharedIkm(identity.privateKey, decPub));
text = open(decKeys.msg, msg.payload, msg.id);
// 回送达回执:签名键始终用与发件人的共享密钥,保证发件人可验
if (senderEntry) {
const rKeys = deriveKeys(sharedIkm(identity.privateKey, senderEntry.pub));
const ts = new Date().toISOString();
sendFn({
schema: SCHEMA, type: 'RECEIPT', id: randomUUID(), msg_id: msg.id,
from: id, to: msg.from, ts,
sig: signReceipt(rKeys.receipt, msg.id, id, msg.from, ts),
});
}
} catch (e) { log(`node ${id}: 解密失败 ${msg.id}(${e.message})`); }
}
inbox.push({ msg_id: msg.id, from: msg.from, text, ts: msg.ts });
onMessage({ msg_id: msg.id, from: msg.from, text });
})();
return;
}
if (msg.type === 'RECEIPT') {
// 校验并落回执
void (async () => {
const entry = await ensurePeerPub(msg.from);
let verified = false;
if (entry) {
const keys = deriveKeys(sharedIkm(identity.privateKey, entry.pub));
verified = signReceipt(keys.receipt, msg.msg_id, msg.from, msg.to, msg.ts) === msg.sig;
}
fs.writeFileSync(path.join(receiptsDir, `${msg.msg_id}.json`), JSON.stringify({
schema: 'guanghu.guanghu-bridge-delivery-receipt/v1',
msg_id: msg.msg_id, from: msg.from, to: msg.to, ts: msg.ts,
signature_verified: verified,
}, null, 2));
log(`node ${id}: 收到送达回执 ${msg.msg_id}(签名${verified ? '验证通过' : '未通过'})`);
})();
return;
}
});
socket.on('connect', () => {
sendFn({ schema: SCHEMA, type: 'HELLO', node_id: id, pub: identity.pubB64 });
});
async function resolvePub(to) {
if (pubCache.has(to)) return pubCache.get(to);
sendFn({ schema: SCHEMA, type: 'GET_PUB', to });
for (let i = 0; i < 50 && !pubCache.has(to); i++) await new Promise((r) => setTimeout(r, 20));
return pubCache.get(to) || null;
}
async function sendText(to, text) {
const entry = await resolvePub(to);
if (!entry) throw new Error(`会合点未应答 ${to} 的公钥解析`);
const keys = deriveKeys(sharedIkm(identity.privateKey, entry.pub));
const env = {
schema: SCHEMA, type: 'MSG', id: randomUUID(),
from: id, to, ts: new Date().toISOString(),
payload: null,
// hybrid=true:收件人未登记,信封暂以会合点公钥加密,投递前转加密
hybrid: entry.hybrid ? true : undefined,
from_pub: entry.hybrid ? identity.pubB64 : undefined,
};
env.payload = seal(keys.msg, text, env.id); // aad 绑定信封 id,防篡改
sendFn({ schema: SCHEMA, type: 'SEND', envelope: env });
log(`node ${id}: 发出 ${env.id} → ${to}${entry.hybrid ? '(离线·混合加密)' : ''}`);
return env.id;
}
return {
id, inbox,
sendText,
receiptsDir,
close: () => new Promise((r) => { socket.end(); socket.on('close', r); setTimeout(r, 200); }),
};
}
// ────────────────────────── 自检 ──────────────────────────
export async function selftest(workdir) {
const logs = [];
const log = (s) => logs.push(s);
const root = ensureDir(workdir);
const relay = await startRelay({ port: 0, stateDir: path.join(root, 'relay'), log });
const nodeA = await startNode({
id: 'NODE-A', relayHost: '127.0.0.1', relayPort: relay.port,
stateDir: path.join(root, 'node-a'), log, onMessage: () => {},
});
await new Promise((r) => setTimeout(r, 150));
// 场景 1:B 离线时 A 发消息 → 离线队列
const m1 = await nodeA.sendText('NODE-B', '冰朔,这是离线补发的第一条。');
await new Promise((r) => setTimeout(r, 150));
// 场景 2:B 接入 → 补发 → 解密 → 回执回到 A
const received = [];
const nodeB = await startNode({
id: 'NODE-B', relayHost: '127.0.0.1', relayPort: relay.port,
stateDir: path.join(root, 'node-b'), log,
onMessage: (m) => received.push(m),
});
await new Promise((r) => setTimeout(r, 400));
// 场景 3:双在线直发
const m2 = await nodeA.sendText('NODE-B', '第二条,双在线直发。');
await new Promise((r) => setTimeout(r, 400));
// 校验
const facts = fs.readdirSync(path.join(root, 'relay', 'facts')).length > 0;
const receiptFiles = fs.readdirSync(path.join(root, 'node-a', 'receipts', 'NODE-A'));
const ok =
received.length === 2 &&
received[0].text === '冰朔,这是离线补发的第一条。' &&
received[1].text === '第二条,双在线直发。' &&
facts && receiptFiles.length === 2;
await nodeA.close(); await nodeB.close(); await relay.close();
const report = {
schema: SCHEMA, kind: 'SELFTEST_REPORT', ts: new Date().toISOString(),
result: ok ? 'PASS' : 'FAIL',
checks: {
offline_queue_and_redelivery: received[0]?.text === '冰朔,这是离线补发的第一条。',
online_direct_delivery: received[1]?.text === '第二条,双在线直发。',
e2e_decrypt_ok: received.every((r) => r.text !== null),
fact_sink_written: facts,
delivery_receipts_signed: receiptFiles.length === 2,
},
msg_ids: [m1, m2],
};
fs.writeFileSync(path.join(root, 'SELFTEST-REPORT.json'), JSON.stringify(report, null, 2));
return report;
}
// ────────────────────────── stdio 桥接 ──────────────────────────
function socketStdioBridge(node, emit) {
// 回执事件:监听回执目录新增文件(简化:由 onMessage 之外补一层轮询不优雅,
// 改为包一层:直接读 node 内部不可行,用定时器扫描回执目录)
const seen = new Set();
const scan = () => {
try {
for (const f of fs.readdirSync(node.receiptsDir)) {
if (seen.has(f)) continue;
seen.add(f);
try {
const r = JSON.parse(fs.readFileSync(path.join(node.receiptsDir, f), 'utf8'));
emit({ event: 'receipt', msg_id: r.msg_id, from: r.from, verified: r.signature_verified });
} catch { /* 半写入文件下轮再读 */ }
}
} catch { /* 目录尚未建立 */ }
};
const timer = setInterval(scan, 300);
const rl = readline.createInterface({ input: process.stdin });
rl.on('line', async (line) => {
let cmd;
try { cmd = JSON.parse(line); } catch { return; }
if (cmd.cmd === 'send' && cmd.to && typeof cmd.text === 'string') {
try { const msgId = await node.sendText(cmd.to, cmd.text); emit({ event: 'sent', msg_id: msgId, to: cmd.to }); }
catch (e) { emit({ event: 'error', message: e.message }); }
} else if (cmd.cmd === 'quit') {
clearInterval(timer); process.exit(0);
}
});
rl.on('close', () => { clearInterval(timer); process.exit(0); });
emit({ event: 'connected', node_id: node.id });
}
// ────────────────────────── CLI ──────────────────────────
function parseArgs(argv) {
const args = {};
for (let i = 0; i < argv.length; i++) {
if (!argv[i].startsWith('--')) continue;
const key = argv[i].slice(2);
if (i + 1 < argv.length && !argv[i + 1].startsWith('--')) { args[key] = argv[i + 1]; i++; }
else args[key] = true; // 无值开关(如 --stdio)
}
return args;
}
const [mode, ...rest] = process.argv.slice(2);
if (mode === 'relay') {
const a = parseArgs(rest);
startRelay({ port: Number(a.port || 9701), stateDir: a['state-dir'] || './bridge-state/relay', log: console.log });
} else if (mode === 'node') {
const a = parseArgs(rest);
const [host, port] = (a.relay || '127.0.0.1:9701').split(':');
const stdio = process.argv.includes('--stdio');
const emit = (obj) => { if (stdio) process.stdout.write(JSON.stringify(obj) + '\n'); };
const node = startNode({
id: a.id, relayHost: host, relayPort: Number(port),
stateDir: a['state-dir'] || `./bridge-state/${a.id}`,
log: stdio ? ((s) => console.error(s)) : console.log,
onMessage: (m) => {
if (stdio) emit({ event: 'message', msg_id: m.msg_id, from: m.from, text: m.text });
else console.log(`〔收到 ${m.from}〕${m.text}`);
},
});
if (stdio) {
// stdio JSON 模式:供 Electron/Tauri 壳以子进程集成(日志走 stderr)
socketStdioBridge(node, emit);
} else {
const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
rl.on('line', async (line) => {
const mm = line.match(/^send\s+(\S+)\s+(.+)$/);
if (mm) { try { await node.sendText(mm[1], mm[2]); } catch (e) { console.error(`发送失败:${e.message}`); } }
else if (line.trim() === 'quit') process.exit(0);
else console.log('用法:send <对端节点号> <文本>');
});
rl.on('close', () => process.exit(0)); // stdin 结束即退出,不挂死
}
} else if (mode === 'selftest') {
const a = parseArgs(rest);
const report = await selftest(a.workdir || './bridge-selftest');
console.log(JSON.stringify(report, null, 2));
process.exit(report.result === 'PASS' ? 0 : 1);
} else if (mode) {
console.log('用法:bridge-node.mjs relay|node|selftest');
}