renegade
/
dsh-memory
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
125 Commits
1 Branch
0 Tags
2.3 MiB
 
 
 
 
 
Tree: 1b0b78ae08
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '1b0b78ae08'
${ noResults }
dsh-memory/bridge/server.js

436 lines
19 KiB
Raw Blame History

'use strict';
// DSH memory - server.js
// 本地桥接服务：HTTP 路由 + Bearer token 鉴权 + 限流 + 审计。
// 仅绑定 127.0.0.1。零 npm 依赖。启动：node server.js（需先在 config.json 填写 token）。
// 预留 /mcp 路由分支：将来可在同一 handler 层挂 streamable-HTTP MCP 端点，
// tools/list -> GET /tools、tools/call -> POST /tool，供 ChatGPT Connector / Claude 直连复用工具层。
const http = require('http');
const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const { loadConfig, ToolError, redact, resolveWorkspace, resolveMemoryDir } = require('./guard');
const { TOOL_DEFS, runTool } = require('./tools');
const memoryModule = require('./memory');
const { McpManager } = require('./mcp-tools');
const stateDir = require('./state-dir');
const { HandoffManager, dshProbe } = require('./handoff');
const BRIDGE_DIR = __dirname;
const AUDIT_FILE = path.join(BRIDGE_DIR, 'audit.jsonl');
let config;
try {
config = loadConfig(BRIDGE_DIR);
} catch (err) {
console.error('[DSH memory] config load failed:', err.message);
process.exit(1);
}
if (!config.token || String(config.token).length < 32) {
console.error('[DSH memory] STARTUP: token missing or short, exiting');
process.exit(1);
}
const handoff = new HandoffManager(config);
const mcp = new McpManager(config);
const pendingConfirms = new Map(); // confirmToken -> {exp, tool, args, workspace}
// ---------------------------------------------------------------------------
// 工具函数
// ---------------------------------------------------------------------------
function timingSafeEqual(a, b) {
const ba = Buffer.from(String(a));
const bb = Buffer.from(String(b));
if (ba.length !== bb.length) return false;
return crypto.timingSafeEqual(ba, bb);
}
function audit(entry) {
try {
const clean = {};
for (const [k, v] of Object.entries(entry)) {
clean[k] = typeof v === 'string' ? redact(v).slice(0, 300) : v;
}
fs.appendFileSync(AUDIT_FILE, JSON.stringify({ ts: new Date().toISOString(), ...clean }) + '\n', 'utf8');
} catch {}
}
// Rate limit removed: local 127.0.0.1, single trusted client, token already checked.
function sendJson(res, status, obj) {
const body = JSON.stringify(obj);
res.writeHead(status, {
'Content-Type': 'application/json; charset=utf-8',
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Headers': 'Content-Type, Authorization',
'Access-Control-Allow-Methods': 'GET, POST, PUT, OPTIONS',
});
res.end(body);
}
const HTTP_STATUS = {
E_UNAUTHORIZED: 401,
E_TOOL_NOT_FOUND: 404,
E_ARGS_INVALID: 400,
E_ROOT_NOT_ALLOWED: 403,
E_PATH_ESCAPE: 403,
E_SYMLINK_ESCAPE: 403,
E_FORBIDDEN_PATH: 403,
E_BINARY_FILE: 415,
E_BASH_BLOCKED: 403,
E_NEEDS_CONFIRM: 403,
E_BUSY: 409,
E_TIMEOUT: 408,
E_INTERNAL: 500,
};
function readBody(req, maxBytes) {
return new Promise((resolve, reject) => {
const chunks = [];
let size = 0;
req.on('data', (c) => {
size += c.length;
if (size > maxBytes) {
reject(new ToolError('E_ARGS_INVALID', '请求体超过 10MB 上限'));
req.destroy();
return;
}
chunks.push(c);
});
req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
req.on('error', reject);
});
}
function authOk(req) {
const h = req.headers['authorization'] || '';
const m = h.match(/^Bearer\s+(.+)$/i);
if (!m) return false;
return timingSafeEqual(m[1], config.token);
}
// ---------------------------------------------------------------------------
// 路由
// ---------------------------------------------------------------------------
async function handle(req, res, pathname, body) {
switch (pathname) {
case '/health': {
const probe = dshProbe(config);
return {
ok: true,
name: 'dsh-memory',
version: '1.0.0',
workspace: config.allowedRoots[0],
memoryDir: resolveMemoryDir(config, BRIDGE_DIR),
allowedRoots: config.allowedRoots,
dsh: { binFound: probe.binFound, bin: probe.bin, nodeSource: probe.node ? probe.node.source : null },
};
}
case '/tools': {
return { ok: true, tools: TOOL_DEFS.concat(mcp.toolDefs()) };
}
case '/mcp/apps': {
return { ok: true, root: mcp.root, apps: await mcp.list() };
}
case '/mcp/rescan': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'only POST');
return { ok: true, root: mcp.root, apps: await mcp.rescan() };
}
case '/mcp/root': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'only POST');
const p = body && body.path;
if (!p || typeof p !== 'string') throw new ToolError('E_ARGS_INVALID', 'missing path');
const r = await mcp.setRoot(p);
return { ok: true, root: r.root, apps: r.apps };
}
case '/mcp/launch': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'only POST');
const app = body && body.app;
if (!app || typeof app !== 'string') throw new ToolError('E_ARGS_INVALID', 'missing app');
return { ok: true, ...(await mcp.launchGui(app)) };
}
case '/mcp/stop': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'only POST');
const app = body && body.app;
if (!app || typeof app !== 'string') throw new ToolError('E_ARGS_INVALID', 'missing app');
return { ok: true, ...(await mcp.stop(app)) };
}
case '/tool': {
const { tool, args, workspace, confirmToken } = body || {};
if (!tool || typeof tool !== 'string') throw new ToolError('E_ARGS_INVALID', '缺少 tool 字段');
const started = Date.now();
const call = async (confirmed) => {
const ctx = { config, workspace, confirmed };
return await runTool(tool, args, ctx, handoff, mcp);
};
let data;
// Если пришёл confirmToken — сразу выполняем с confirmed:true, без лишнего прохода call(false).
if (confirmToken && pendingConfirms.has(confirmToken)) {
const pending = pendingConfirms.get(confirmToken);
pendingConfirms.delete(confirmToken);
if (pending.exp < Date.now()) throw new ToolError('E_NEEDS_CONFIRM', '确认令牌已过期，请重新发起');
data = await call(true);
} else {
try {
data = await call(false);
} catch (err) {
if (err instanceof ToolError && err.code === 'E_NEEDS_CONFIRM') {
const token = crypto.randomBytes(16).toString('hex');
pendingConfirms.set(token, { exp: Date.now() + 10 * 60 * 1000, tool, args, workspace });
for (const [k, v] of pendingConfirms) if (v.exp < Date.now()) pendingConfirms.delete(k);
err.extra = { ...(err.extra || {}), confirmToken: token };
throw err;
}
throw err;
}
}
audit({ op: 'tool', tool, workspace: workspace || config.allowedRoots[0], ms: Date.now() - started, ok: true });
// Result above resultMaxBytes is dumped to .dsh-memory/outputs/ so the model can read it in chunks
let payload = data;
const dataStr = JSON.stringify(data);
const resultMaxBytes = Number(config.resultMaxBytes) || 24576;
if (dataStr.length > resultMaxBytes) {
const root = resolveWorkspace(config, workspace);
stateDir.ensureStateDir(root);
const outsDir = stateDir.outputsDir(root);
fs.mkdirSync(outsDir, { recursive: true });
const name = `${Date.now()}-${String(tool).replace(/[^a-z0-9_-]/gi, '_')}.json`;
fs.writeFileSync(path.join(outsDir, name), dataStr, 'utf8');
payload = {
preview: dataStr.slice(0, resultMaxBytes),
fullOutputFile: `.dsh-memory/outputs/${name}`,
hint: `Result exceeds ${resultMaxBytes} bytes, dumped to file. Use dsh_read to read it in chunks (or reduce limit/maxResults).`,
};
}
return { ok: true, data: payload, meta: { durationMs: Date.now() - started } };
}
case '/handoff': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', '仅支持 POST');
const { planText, workspace, taskNote, timeoutMs } = body || {};
const started = Date.now();
const r = handoff.start({ workspace, planText, taskNote, timeoutMs });
audit({ op: 'handoff-start', workspace: workspace || config.allowedRoots[0], runId: r.runId, planMs: Date.now() - started });
return { ok: true, ...r };
}
case '/memory/state': {
const memDir = resolveMemoryDir(config, BRIDGE_DIR);
if (req.method === 'GET') {
const st = memoryModule.loadState(memDir);
return { ok: true, data: st };
}
if (req.method === 'PUT') {
const b = body || {};
const st = memoryModule.loadState(memDir);
if (Array.isArray(b.entries)) st.entries = b.entries;
if (b.chats && typeof b.chats === 'object') st.chats = b.chats;
const saved = memoryModule.saveState(memDir, st);
return { ok: true, data: saved };
}
throw new ToolError('E_ARGS_INVALID', 'GET or PUT only');
}
case '/memory/query': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'POST only');
const memDir = resolveMemoryDir(config, BRIDGE_DIR);
const st = memoryModule.loadState(memDir);
const op = body && body.op ? String(body.op) : 'list';
let result;
try { result = memoryModule.query(st, body || {}); }
catch (err) { throw new ToolError('E_ARGS_INVALID', String(err && err.message)); }
if (op === 'write' || op === 'add' || op === 'replace' || op === 'remove') { memoryModule.saveState(memDir, st); }
return { ok: true, data: result };
}
case '/memory/inject': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'POST only');
const memDir = resolveMemoryDir(config, BRIDGE_DIR);
const st = memoryModule.loadState(memDir);
const chatId = body && body.chatId ? String(body.chatId) : '';
const wsKey = body && body.workspaceKey ? String(body.workspaceKey) : String(config.allowedRoots[0] || '');
const inj = memoryModule.computeInjection(st, chatId, wsKey);
const chat = (chatId && st.chats) ? st.chats[chatId] : null;
const lastFp = chat ? String(chat.lastInjFp || '') : '';
if (lastFp && lastFp === inj.fingerprint) {
return { ok: true, data: { block: '', fingerprint: inj.fingerprint, count: 0, bytes: 0, deduped: true } };
}
memoryModule.applyInjection(st, inj, chatId);
memoryModule.saveState(memDir, st);
return { ok: true, data: inj };
}
case '/memory/chats': {
const memDir = resolveMemoryDir(config, BRIDGE_DIR);
const stateFile = path.join(memDir, 'state.json');
let st = {};
try { st = JSON.parse(fs.readFileSync(stateFile, 'utf8')); } catch (e) { st = {}; }
if (!st.chats || typeof st.chats !== 'object') st.chats = {};
if (req.method === 'GET') {
var qid = ''; var qs = String(req.url || ''); var qi = qs.indexOf('chatId='); if (qi !== -1) { qid = qs.slice(qi + 7); var qa = qid.indexOf('&'); if (qa !== -1) qid = qid.slice(0, qa); } const id = qid ? String(qid) : ((body && body.chatId) ? String(body.chatId) : '');
if (id) return { ok: true, data: st.chats[id] || null };
return { ok: true, data: st.chats };
}
if (req.method === 'POST') {
const b = body || {};
const id = b.chatId ? String(b.chatId) : '';
if (!id || id.length > 64 || id.indexOf('/') !== -1) throw new ToolError('E_ARGS_INVALID', 'bad chatId');
const entry = (b.entry && typeof b.entry === 'object') ? b.entry : {};
entry.updatedAt = Date.now();
st.chats[id] = entry;
const ids = Object.keys(st.chats);
if (ids.length > 20) {
ids.sort(function (x, y) { return (st.chats[x].updatedAt || 0) - (st.chats[y].updatedAt || 0); });
const drop = ids.length - 20;
for (let i = 0; i < drop; i++) delete st.chats[ids[i]];
}
st.version = st.version || 1;
st.updatedAt = new Date().toISOString();
fs.mkdirSync(memDir, { recursive: true });
const tmp = stateFile + '.tmp';
fs.writeFileSync(tmp, JSON.stringify(st, null, 2), 'utf8');
fs.renameSync(tmp, stateFile);
return { ok: true, data: st.chats[id] };
}
throw new ToolError('E_ARGS_INVALID', 'GET or POST only');
}
case '/log/append': {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', 'only POST');
const file = body && body.file;
const lines = body && body.lines;
if (typeof file !== 'string') throw new ToolError('E_ARGS_INVALID', 'file required');
if (file.length < 7 || file.length > 44) throw new ToolError('E_ARGS_INVALID', 'file length');
if (file.slice(-6) !== '.jsonl') throw new ToolError('E_ARGS_INVALID', 'file ext');
if (file.indexOf('/') !== -1) throw new ToolError('E_ARGS_INVALID', 'file slash');
if (file.indexOf(String.fromCharCode(92)) !== -1) throw new ToolError('E_ARGS_INVALID', 'file backslash');
if (!Array.isArray(lines) || lines.length === 0) throw new ToolError('E_ARGS_INVALID', 'lines required');
if (lines.length > 200) throw new ToolError('E_ARGS_INVALID', 'too many lines');
const logDir = path.join(resolveMemoryDir(config, BRIDGE_DIR), 'chat-log');
fs.mkdirSync(logDir, { recursive: true });
const target = path.join(logDir, file);
const resolved = path.resolve(target);
const rootResolved = path.resolve(logDir) + path.sep;
if (!resolved.startsWith(rootResolved)) throw new ToolError('E_PATH_ESCAPE', 'escape');
const buf = lines.map(function (l) {
const s = typeof l === 'string' ? l : JSON.stringify(l);
return s + String.fromCharCode(10);
}).join('');
const bytes = Buffer.byteLength(buf, 'utf8');
if (bytes > 524288) throw new ToolError('E_ARGS_INVALID', 'payload too big');
fs.appendFileSync(resolved, buf, 'utf8');
try {
var __trimMax = 50 * 1024;
var __trimKeep = 25 * 1024;
var __st = fs.statSync(resolved);
if (__st.size > __trimMax) {
var __cur = fs.readFileSync(resolved, 'utf8');
var __start = __cur.length - __trimKeep;
if (__start < 0) __start = 0;
var __nl = __cur.indexOf(String.fromCharCode(10), __start);
if (__nl !== -1) {
fs.writeFileSync(resolved, __cur.slice(__nl + 1), 'utf8');
}
}
} catch (e) { }
return { ok: true, data: { file: file, lines: lines.length, bytes: bytes } };
}
default:
// /handoff/:runId 与 /handoff/:runId/cancel
const m = pathname.match(/^\/handoff\/([A-Za-z0-9-]+)(\/cancel)?$/);
if (m) {
const runId = m[1];
if (req.method === 'POST' && m[2]) {
return { ok: true, ...handoff.cancel(runId) };
}
if (req.method === 'GET') {
const run = handoff.get(runId);
if (!run) throw new ToolError('E_TOOL_NOT_FOUND', `未找到 handoff：${runId}`);
return { ok: true, run };
}
}
if (pathname === '/mcp') {
throw new ToolError('E_TOOL_NOT_FOUND', 'MCP 端点为预留分支，本期未实现（见 server.js 头注释）');
}
if (pathname === '/shutdown') {
if (req.method !== 'POST') throw new ToolError('E_ARGS_INVALID', '仅支持 POST');
const restart = !!(body && body.restart);
if (restart && process.env.DSH_SUPERVISOR !== '1') {
// No supervisor around — spawn a detached one so restart actually works.
try {
const supPath = path.join(__dirname, '..', 'scripts', 'supervisor.js');
const child = require('child_process').spawn(process.execPath, [supPath], { cwd: path.join(__dirname, '..'), detached: true, stdio: 'ignore', windowsHide: true });
child.unref();
console.log('[DSH memory] no supervisor detected, spawned detached supervisor');
} catch (e) {
console.warn('[DSH memory] failed to spawn supervisor: ' + (e && e.message));
}
}
setTimeout(() => process.exit(restart ? 42 : 0), restart ? 1500 : 300);
return { ok: true, bye: true, restart: restart };
}
throw new ToolError('E_TOOL_NOT_FOUND', `未知路径：${pathname}`);
}
}
const server = http.createServer(async (req, res) => {
const ip = req.socket.remoteAddress || '';
const u = new URL(req.url, 'http://127.0.0.1');
const pathname = u.pathname;
const started = Date.now();
if (req.method === 'OPTIONS') {
res.writeHead(204, {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Headers': 'Content-Type, Authorization',
'Access-Control-Allow-Methods': 'GET, POST, PUT, OPTIONS',
});
res.end();
return;
}
try {
if (ip !== '127.0.0.1' && ip !== '::1' && ip !== '::ffff:127.0.0.1') {
throw new ToolError('E_UNAUTHORIZED', '仅允许本机访问');
}
if (!authOk(req)) {
throw new ToolError('E_UNAUTHORIZED', 'invalid or missing bearer token');
}
let body = null;
if (req.method === 'POST' || req.method === 'PUT') {
const raw = await readBody(req, 10 * 1024 * 1024);
if (raw.trim()) {
try {
body = JSON.parse(raw);
} catch {
throw new ToolError('E_ARGS_INVALID', '请求体不是合法 JSON');
}
}
}
const result = await handle(req, res, pathname, body);
audit({ op: 'http', method: req.method, path: pathname, ms: Date.now() - started, ok: true });
sendJson(res, 200, result);
} catch (err) {
const isTool = err instanceof ToolError;
const code = isTool ? err.code : 'E_INTERNAL';
const status = HTTP_STATUS[code] || 500;
audit({ op: 'http', method: req.method, path: pathname, ms: Date.now() - started, ok: false, code, msg: String(err && err.message) });
sendJson(res, status, { ok: false, error: { code, message: String(err && err.message), ...(isTool && err.extra ? { extra: err.extra } : {}) } });
}
});
server.listen(config.port, '127.0.0.1', () => {
console.log(`[DSH memory] bridge listening on http://127.0.0.1:${config.port}`);
// ensureStateDir on startup: create .dsh-memory/ in every allowedRoot
for (const root of config.allowedRoots) {
const sd = stateDir.ensureStateDir(root);
if (sd) console.log('[DSH memory] state dir ready: ' + sd);
else console.log('[DSH memory] state dir failed for root: ' + root);
}
mcp.init().then(function () { console.log('[mcp] init done'); }).catch(function (e) { console.warn('[mcp] init failed: ' + (e && e.message)); });
console.log(`[DSH memory] allowedRoots: ${config.allowedRoots.join(' ; ')}`);
const probe = dshProbe(config);
if (probe.binFound) console.log(`[DSH memory] dsh runtime: found (${probe.bin})`);
else console.log('[DSH memory] dsh runtime: NOT FOUND (handoff disabled; set config.dshBin to enable)');
});
process.on('uncaughtException', (err) => {
console.error('[DSH memory] uncaught:', err && err.stack);
});
process.on('unhandledRejection', (err) => {
console.error('[DSH memory] unhandledRejection:', String(err));
});