// ============================================================ // data_gateway/index.js - 对外 API 数据网关主入口 // 暴露两个路由: // POST /api/v1/ingest/:skillId 外部数据写入(公网,需 API Key) // GET /api/v1/data/:skillId/query 内部数据读取(localhost 或 Internal Token) // ============================================================ const { logJSON } = require('../logger') const { verifyApiKey, listEnabledSkills, requireLocalOnly, requireInternalOnly } = require('./auth') const { insert, upsertLatest, queryLatest, queryList, queryByTimeRange, count } = require('./store') const { loadSkillConfig } = require('./auth') const { bindRoutes: bindExternalStorageRoutes } = require('./skills/external_storage') const { bindRoutes: bindEmailSenderRoutes } = require('./skills/email_sender') const { bindRoutes: bindPrivateClipboardRoutes } = require('./skills/private_clipboard') const setNoCache = (res) => { try { res.set('Cache-Control', 'no-store, no-cache, must-revalidate, proxy-revalidate') res.set('Pragma', 'no-cache') res.set('Expires', '0') } catch {} } // ============================================================ // 安全工具函数 // ============================================================ // 原型污染防护:移除 __proto__、constructor、prototype 等危险键 const PROTO_KEYS = new Set(['__proto__', 'constructor', 'prototype', '__defineGetter__', '__defineSetter__', '__lookupGetter__', '__lookupSetter__', '__proto__']) const sanitizeObject = (obj, depth = 0) => { if (depth > 10) return {} // 防止无限递归 if (obj === null || typeof obj !== 'object') return obj if (Array.isArray(obj)) return obj.map(item => sanitizeObject(item, depth + 1)) const clean = {} for (const [key, val] of Object.entries(obj)) { if (PROTO_KEYS.has(key)) continue clean[key] = sanitizeObject(val, depth + 1) } return clean } // 全局错误处理:隐藏详细堆栈,避免泄露路径和技术栈 const errorHandler = (err, req, res, _next) => { try { logJSON('data_gateway.error', { path: req.path, error: String(err.message || err) }, 'data_gateway') } catch {} if (!res.headersSent) { res.status(500).json({ ok: false, error: 'internal_error' }) } } // Ingest 接口速率限制:同一 IP 每分钟最多 60 次请求 const ingestRateLimit = new Map() const INGEST_MAX_PER_MIN = 60 const INGEST_WINDOW_MS = 60 * 1000 const checkIngestRate = (ip) => { const rec = ingestRateLimit.get(ip) if (!rec || Date.now() - rec.start > INGEST_WINDOW_MS) { ingestRateLimit.set(ip, { start: Date.now(), count: 1 }) return { ok: true } } rec.count++ if (rec.count > INGEST_MAX_PER_MIN) { const waitSec = Math.ceil((rec.start + INGEST_WINDOW_MS - Date.now()) / 1000) return { ok: false, retryAfter: Math.max(1, waitSec) } } return { ok: true } } // 每分钟清理一次 setInterval(() => { const now = Date.now() for (const [ip, rec] of ingestRateLimit.entries()) { if (now - rec.start > INGEST_WINDOW_MS + 10000) ingestRateLimit.delete(ip) } }, 60000) const SOURCE_MAX_LEN = 255 const DATA_MAX_BYTES = 10 * 1024 // 10KB const bindRoutes = (app) => { // ============================================================ // 外部存储 skill(第三方 OSS 读写) // 路由挂载在 /api/v1/skills/external_storage/* // ============================================================ try { bindExternalStorageRoutes(app) } catch (e) { console.error(`[data_gateway] Failed to bind external_storage routes: ${e.message}`) } // ============================================================ // 邮件发送 skill(对外发邮件) // 路由挂载在 /api/v1/ingest/email_sender/* // ============================================================ try { bindEmailSenderRoutes(app) } catch (e) { console.error(`[data_gateway] Failed to bind email_sender routes: ${e.message}`) } // ============================================================ // 私人剪贴板 skill(对外写入剪贴板) // 路由挂载在 /api/v1/ingest/private_clipboard/* // ============================================================ try { bindPrivateClipboardRoutes(app) } catch (e) { console.error(`[data_gateway] Failed to bind private_clipboard routes: ${e.message}`) } // ============================================================ // 外部写入接口(公网暴露) // POST /api/v1/ingest/:skillId // Headers: X-API-Id, X-API-Key // Body: { "source": "...", "data": { ... } } // ============================================================ app.post('/api/v1/ingest/:skillId', (req, res) => { setNoCache(res) const skillId = String(req.params.skillId || '').trim() if (!skillId) return res.status(400).json({ ok: false, error: 'missing skill id' }) const auth = verifyApiKey(req) if (!auth.ok) { try { logJSON('data_gateway.auth.fail', { skillId, error: auth.error, ip: req.ip }, 'data_gateway') } catch {} return res.status(401).json({ ok: false, error: auth.error }) } // 速率限制 const clientIp = req.ip || req.connection.remoteAddress || 'unknown' const rateCheck = checkIngestRate(clientIp) if (!rateCheck.ok) { return res.status(429).json({ ok: false, error: '请求过于频繁,请在 ' + rateCheck.retryAfter + ' 秒后重试' }) } const body = req.body || {} let source = String(body.source || '').trim() let data = body.data // source 长度限制 if (source.length > SOURCE_MAX_LEN) { return res.status(400).json({ ok: false, error: 'source 超过长度限制(' + SOURCE_MAX_LEN + ' 字符)' }) } // 如果没有 data 字段,将整个 body 视为 data(排除 source) if (data === undefined) { const bodyData = { ...body } delete bodyData.source data = Object.keys(bodyData).length > 0 ? bodyData : body } // data 类型验证:必须是对象 if (typeof data !== 'object' || data === null || Array.isArray(data)) { return res.status(400).json({ ok: false, error: 'data 必须是 JSON 对象' }) } // data 大小限制 const dataBytes = Buffer.byteLength(JSON.stringify(data), 'utf8') if (dataBytes > DATA_MAX_BYTES) { return res.status(413).json({ ok: false, error: 'data 超过大小限制(' + Math.round(DATA_MAX_BYTES / 1024) + 'KB)' }) } // 原型污染防护 data = sanitizeObject(data) const record = insert(skillId, { source, data }) try { logJSON('data_gateway.ingest.ok', { skillId, source, recordId: record.id }, 'data_gateway') } catch {} return res.json({ ok: true, id: record.id, created_at: record.created_at }) }) // ============================================================ // 内部读取接口(本机直连,或带 Internal Token 的受保护读取) // GET /api/v1/data/:skillId/query?mode=latest|list&limit=100&offset=0 // Headers: X-Internal-Token, X-Skill-Id // ============================================================ app.get('/api/v1/data/:skillId/query', requireInternalOnly, (req, res) => { setNoCache(res) const skillId = String(req.params.skillId || '').trim() if (!skillId) return res.status(400).json({ ok: false, error: 'missing skill id' }) const mode = String(req.query.mode || 'latest').trim() const limit = parseInt(req.query.limit, 10) || 100 const offset = parseInt(req.query.offset, 10) || 0 const source = req.query.source || null const start = req.query.start || null const end = req.query.end || null try { if (mode === 'list') { if (start || end) { const result = queryByTimeRange(skillId, { start, end, limit }) return res.json({ ok: true, ...result }) } const result = queryList(skillId, { limit, offset, source }) return res.json({ ok: true, ...result }) } // mode === 'latest'(默认) const row = queryLatest(skillId) if (!row) return res.json({ ok: true, rows: [], total: 0 }) return res.json({ ok: true, rows: [row], total: 1 }) } catch (e) { try { logJSON('data_gateway.query.error', { skillId, error: String(e.message || e) }, 'data_gateway') } catch {} return res.status(500).json({ ok: false, error: String(e.message || e) }) } }) // ============================================================ // 管理接口:列出所有已启用的 skill(仅 localhost) // GET /api/v1/data/skills // ============================================================ app.get('/api/v1/data/skills', requireLocalOnly, (req, res) => { setNoCache(res) const skills = listEnabledSkills() const result = skills.map(s => { const cfg = loadSkillConfig(s.id) return { id: s.id, name: s.name, mode: String(cfg && cfg.mode || 'append'), record_count: count(s.id) } }) res.json({ ok: true, skills: result }) }) // 全局错误处理:统一返回简洁错误,隐藏堆栈和路径信息 app.use(errorHandler) } module.exports = { bindRoutes, listEnabledSkills, loadSkillConfig }