MCP 网关 · 技术文档

cd mcp-agent

# 首次：准备环境变量（可选，不准备也能跑，会走内置默认值）
cp backend/.env.example backend/.env

./start.sh              # 启动：后端 :8080 + 前端 :5173
./start.sh --stop       # 停止已在运行的实例（含残留端口占用）
./start.sh --force      # 端口被占用时自动停旧实例再启动
./start.sh --no-seed    # 跳过 seed（确认库结构已就绪时用）
./start.sh -h           # 帮助

════════════════════════ MCP 网关已启动 ════════════════════════
  ▶ 管理后台   http://127.0.0.1:5173
                 登录 Token: mgw_admin_dev_token (生产请通过 ADMIN_TOKEN 环境变量修改)
  ▶ MCP 端点   http://127.0.0.1:8080/mcp
                 鉴权头: Authorization: Bearer <mgw_live_…>
  ▶ 健康检查   http://127.0.0.1:8080/healthz
  按 Ctrl+C 一键停止  ·  或另开终端执行 ./start.sh --stop

# ── 终端 A：后端 ─────────────────────────────
cd backend
npm install                       # 首次，会编译 better-sqlite3
cp .env.example .env              # 按需填写 MGW_SECRET / ADMIN_TOKEN
set -a && source .env && set +a   # 手动喂入环境变量
npm run seed                      # 建库 + 内置角色 / settings / 首个 Key
npm run start:dev                 # → http://127.0.0.1:8080

# ── 终端 B：前端 ─────────────────────────────
cd frontend
npm install
npm run dev                       # → http://127.0.0.1:5173

KEY="mgw_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
BASE="http://127.0.0.1:8080/mcp"

# 1) initialize —— 协商协议版本，拿到 Mcp-Session-Id
curl -i -X POST "$BASE" \
  -H "Authorization: Bearer $KEY" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc":"2.0","id":1,"method":"initialize",
    "params":{
      "protocolVersion":"2025-06-18",
      "capabilities":{},
      "clientInfo":{"name":"curl","version":"0"}
    }
  }'
# → 响应头含 mcp-session-id: <uuid>

SID="<上一步返回的 mcp-session-id>"

# 2) notifications/initialized —— 握手完成通知（无响应体）
curl -X POST "$BASE" \
  -H "Authorization: Bearer $KEY" -H "mcp-session-id: $SID" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc":"2.0","method":"notifications/initialized"}'

# 3) tools/list —— 只返回该 Key 角色「有权看到」的工具
curl -X POST "$BASE" \
  -H "Authorization: Bearer $KEY" -H "mcp-session-id: $SID" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list"}'

# 4) tools/call —— 走完整治理链
curl -X POST "$BASE" \
  -H "Authorization: Bearer $KEY" -H "mcp-session-id: $SID" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc":"2.0","id":3,"method":"tools/call",
    "params":{"name":"wiki__search","arguments":{"q":"MCP"}}
  }'

# 5) 订阅工具变更通知（SSE 长连接，25s 心跳）
curl -N "$BASE" -H "Authorization: Bearer $KEY" -H "mcp-session-id: $SID"

{
  "mcpServers": {
    "corp-gateway": {
      "type": "streamable-http",
      "url": "http://127.0.0.1:8080/mcp",
      "headers": { "Authorization": "Bearer mgw_live_xxxxxxxxxxxx" }
    }
  }
}

PRAGMA journal_mode = WAL;      -- 写前日志：读写并发，读不阻塞写
PRAGMA busy_timeout  = 5000;    -- 写锁竞争时最多等 5s，而非立刻 SQLITE_BUSY
PRAGMA foreign_keys  = ON;      -- SQLite 默认关闭外键约束，必须显式打开
PRAGMA synchronous   = NORMAL;  -- WAL 模式下的推荐档位：性能与崩溃安全性的平衡点

async routeCall(
  fullName: string,                              // 'wiki__search'
  args: Record<string, unknown>,                 // 模型给的参数
  caller: {                                      // 由 mcp.controller 鉴权后注入
    keyId: number;
    role: string;
    userName: string | null;
    team: string | null;
    profile: string | null;
  },
  ctx: {                                         // 请求上下文
    requestId: string;
    sessionId: string | null;
    callerIp: string | null;
    clientInfo: string | null;
  },
): Promise<ToolCallResult>

const entry = this.registry.get(fullName);
if (!entry || !entry.enabled) {
  return this.fail(caller, ctx, fullName, null,
    ErrorCode.TOOL_NOT_FOUND, '工具不存在或已下线', 0);
}
const u = await this.upstreamRepo.findOne({ where: { name: entry.upstream } });
if (!u || !u.enabled) {
  return this.fail(caller, ctx, fullName, entry.upstream,
    ErrorCode.UPSTREAM_DOWN, '上游服务暂不可用', 0);
}

// 与 tools/list 共用同一份 allowedSet —— 「可见即可调」契约
const allNames = this.registry.allToolNames();
const allowed  = this.rbac.buildAllowedSet(caller.role, allNames, this.registry.version);

if (!allowed.has(fullName)) {
  if (this.rbac.getMode() === 'observe') {
    // F1 影子模式：放行，但打一条 rbac_would_forbid 审计
    this.audit.log({
      event: 'rbac_would_forbid',
      tool: fullName,
      upstream: entry.upstream,
      errorCode: ErrorCode.WOULD_FORBID,
      resultOk: 1,                       // 注意：实际是成功的
      ...caller, ...ctx,
    });
    // 不 return，继续往下走限流 / 熔断 / 转发
  } else {
    return this.fail(caller, ctx, fullName, entry.upstream,
      ErrorCode.FORBIDDEN, '无权调用该工具', 0, 'client-error');
  }
}

// 工具级覆写 QPM 优先于全局默认
const ov   = await this.overrideRepo.findOne({ where: { fullName } });
const rate = ov?.rateQpm ?? undefined;          // undefined → 走 settings 默认 30
const rl   = this.rateLimit.take(caller.keyId, fullName, rate, rate);

if (!rl.ok) {
  return this.fail(caller, ctx, fullName, entry.upstream,
    ErrorCode.RATE_LIMITED, '调用过于频繁，请稍后重试', 0, 'client-error',
    { retryAfterS: rl.retryAfterS });
}

const st = this.breaker.allow(entry.upstream);   // 'closed' | 'half-open' | 'open'
if (st === 'open') {
  return this.fail(caller, ctx, fullName, entry.upstream,
    ErrorCode.UPSTREAM_DOWN, '上游服务暂不可用', 0, 'upstream-error');
}
// half-open：breaker 内部用 probing 布尔位保证只有一个请求作为探针放行

const limit = u.concurrency ?? defaultConcurrency;     // settings: 20
// stdio 上游是本地子进程，并发能力有限，强制收敛到 ≤2
const effLimit = u.transport === 'stdio' ? Math.min(limit, 2) : limit;

const acq = await this.concurrency.acquire(entry.upstream, effLimit, 50);
if (!acq.ok) {
  return this.fail(caller, ctx, fullName, entry.upstream,
    ErrorCode.UPSTREAM_BUSY, '上游繁忙，请稍后重试', 0, 'upstream-error',
    { queueMs: acq.waitedMs });
}

try {
  // … GATE 6 转发 …
} finally {
  acq.release();          // 必须在 finally 里释放，否则 slot 泄漏
}

const timeout = u.timeoutMs ?? defaultTimeout;         // settings: 20000
const t0 = Date.now();
try {
  const raw = await this.upstreamManager.callTool(entry.upstream, entry.originalName, args, timeout);
  const durationMs = Date.now() - t0;

  this.breaker.onSuccess(entry.upstream);
  this.metrics.toolCall(entry.upstream, 'ok', durationMs);
  this.audit.log({ event: 'tool_call', resultOk: 1, durationMs, queueMs: acq.waitedMs, … });

  return { content: raw.content, isError: false };
} catch (e) {
  const ge = toGatewayError(e);                        // 归一化
  this.breaker.onFailure(entry.upstream, ge.kind);     // client-error 直接 return null
  this.metrics.toolCall(entry.upstream, ge.code, Date.now() - t0);
  return this.fail(caller, ctx, fullName, entry.upstream, ge.code, ge.safeMessage,
                   Date.now() - t0, ge.kind, ge.meta);
}

// tools/list 走这里
listTools(caller) {
  const allowed = this.rbac.buildAllowedSet(caller.role, this.registry.allToolNames(), this.registry.version);
  return this.registry.listFor(allowed);        // 只返回 allowed 命中的
}

// tools/call 走 GATE 2，用的是同一个 buildAllowedSet()
// 缓存键 = role + registryVersion + rulesVersion + 60s TTL
// → 同一次会话内，两者结果必然一致

interface RegistryEntry {
  fullName: string;        // 'wiki__search'
  upstream: string;        // 'wiki'
  originalName: string;    // 'search' —— 转发给上游时用这个名字
  def: {                   // 暴露给模型的 MCP 工具定义
    name: string;
    title?: string;
    description?: string;
    inputSchema?: object;
  };
  rawDescription: string;  // 上游原始描述（覆写前的原值，供后台对比/还原）
  enabled: boolean;        // 可被 tool_overrides 单独下线
  sensitivity: 'normal' | 'sensitive';  // 继承自上游，决定是否存全量参数
}

private _version = 0;
get version(): number { return this._version; }

/** 工具集发生任何变化都调它：单调递增，永不回退 */
private bump(): void { this._version++; }

// 类型层面：用 typeof import 拿到类型，不产生运行时 require
type ClientCtor = typeof import('@modelcontextprotocol/sdk/client/index.js');

// 运行时：三个入口分别动态 import
const { Client }                 = await import('@modelcontextprotocol/sdk/client/index.js');
const { StdioClientTransport }   = await import('@modelcontextprotocol/sdk/client/stdio.js');
const { StreamableHTTPClientTransport } =
                                   await import('@modelcontextprotocol/sdk/client/streamableHttp.js');

const STDIO_ENV_ALLOW_KEYS = new Set<string>([
  'PATH', 'HOME', 'USER', 'SHELL', 'LANG', 'TMPDIR', 'PWD', 'TZ', 'NODE_ENV',
]);
const STDIO_ENV_ALLOW_PREFIXES = ['LC_'];

function buildChildEnv(custom: Record<string, string>): NodeJS.ProcessEnv {
  const out: NodeJS.ProcessEnv = {};
  for (const [k, v] of Object.entries(process.env)) {
    if (STDIO_ENV_ALLOW_KEYS.has(k) || STDIO_ENV_ALLOW_PREFIXES.some((p) => k.startsWith(p))) {
      out[k] = v;
    }
  }
  // 只叠加管理员显式配置的、解密后的自定义变量
  return { ...out, ...custom };
}

/** 永久拒绝，无论白名单怎么写 */
const ALWAYS_DENY = ['169.254.0.0/16', '127.0.0.0/8'];

export function ipToLong(ip: string): number;      // '10.0.0.1' → 167772161
export function inCidr(ip: string, cidr: string): boolean;

export function checkUpstreamUrl(
  rawUrl: string,
  allowedCidrs: string[],
  resolvedIp?: string,      // 已做 DNS 解析时传入，避免 TOCTOU
): { ok: true } | { ok: false; reason: string };

// 每 30s 遍历所有 enabled 上游
setInterval(() => this.healthCheckAll(), 30_000);

// 单次检查：尝试 client.listTools()，据此更新状态
private async probe(name: string): Promise<void> {
  try {
    await this.clientFor(name).listTools();
    await this.repo.update({ name }, {
      lastStatus: 'up', lastError: null, lastCheckedAt: Date.now(),
    });
    if (wasDown) {                       // 从 down 恢复 → 通知 + 熔断恢复
      this.im.upstreamRecovered(name);
      this.breaker.onUpstreamUp(name);
      await this.registry.refreshUpstream(name, tools);
    }
  } catch (e) {
    await this.repo.update({ name }, {
      lastStatus: 'down', lastError: safeMessage(e), lastCheckedAt: Date.now(),
    });
  }
}

// 每次失败翻倍，上限 5 分钟
state.lastBackoff = Math.min((state.lastBackoff ?? 1000) * 2, 300_000);
setTimeout(() => this.connect(name), state.lastBackoff);

const CRASH_WINDOW_MS = 5 * 60_000;   // 5 分钟观察窗
const CRASH_THRESHOLD = 5;            // 窗口内崩溃 5 次

private recordCrash(name: string): void {
  const s = this.crashState(name);
  s.hits = s.hits.filter((t) => Date.now() - t < CRASH_WINDOW_MS);
  s.hits.push(Date.now());
  if (s.hits.length < CRASH_THRESHOLD) return;

  // 触发：彻底禁用，停止一切重试
  s.hits = [];
  this.repo.update({ name }, { enabled: 0, lastError: 'crash_loop' });
  this.changeLog.record({
    actor: 'system', targetType: 'upstream', targetId: name,
    action: 'auto-disable',
    diffSummary: `${name} 5 分钟内崩溃 ${CRASH_THRESHOLD} 次，已自动禁用`,
  });
  this.im.stdioCrashLoop(name);
  logger.warn({ event: 'stdio_crash_loop', upstream: name });
}

async callTool(upstream: string, originalName: string,
               args: object, timeoutMs: number) {
  const client = this.clientFor(upstream);       // 未连接 → 抛 UPSTREAM_DOWN
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), timeoutMs);
  try {
    return await client.callTool(
      { name: originalName, arguments: args },   // 注意：用原名，不是全名
      undefined,
      { signal: ac.signal, timeout: timeoutMs },
    );
  } catch (e) {
    if (ac.signal.aborted) throw upstreamError(ErrorCode.UPSTREAM_TIMEOUT, '上游响应超时', String(e));
    throw e;
  } finally {
    clearTimeout(timer);
  }
}

generateApiKey(): { key: string; hash: string; prefix: string } {
  const key = 'mgw_live_' + randomBytes(24).toString('base64url');  // 32 字符随机体
  return {
    key,
    hash:   createHash('sha256').update(key).digest('hex'),          // 库里只存这个
    prefix: key.slice(0, 16),                                        // 'mgw_live_AbCdEf'
  };
}

async verify(rawKey: string): Promise<Caller | null> {
  if (!rawKey?.startsWith('mgw_live_')) return null;      // 快速失败，省一次哈希

  // 吊销黑名单缓存：命中即拒，TTL 2s
  if (this.revokedCache.has(rawKey)) return null;

  const hash = sha256(rawKey);
  const row  = await this.repo.findOne({ where: { keyHash: hash } });
  if (!row || row.revoked) { this.revokedCache.set(rawKey, true); return null; }
  if (row.expiresAt && row.expiresAt < Date.now()) return null;

  row.lastUsedAt = Date.now();                            // 异步更新，不阻塞主链路
  return { keyId: row.id, role: row.role, userName: row.userName,
           team: row.team, profile: row.profile };
}

/** 通配符 → 正则。只有 '*' 是元字符，其余全部转义 */
private toRegex(pattern: string): RegExp {
  const body = pattern.split('*').map(escapeRegExp).join('.*');
  return new RegExp('^' + body + '$', 's');    // 's' 让 '.' 也匹配换行
}

compute(role: string, fullName: string): boolean {
  const rules = this.rulesOf(role);            // 已按 sort 升序
  for (const r of rules) {
    if (this.regexOf(r.pattern).test(fullName)) {
      return r.effect === 'allow';             // ★ 首条命中即返回
    }
  }
  return false;                                // 默认拒绝（deny by default）
}

/** 检测嵌套量词，如 (a+)+ 、 (a*){2,} —— 指数级回溯的典型形态 */
const NESTED_QUANTIFIER = /\([^)]*[+*]\)[+*{]/;
const MAX_PATTERN_LEN = 200;

validatePattern(p: string): void {
  if (p.length > MAX_PATTERN_LEN) throw new BadRequestException('pattern 过长（上限 200）');
  if (NESTED_QUANTIFIER.test(p))  throw new BadRequestException('pattern 含嵌套量词，存在 ReDoS 风险');
}

interface PreviewResult {
  role: string;
  currentCount: number;      // 当前可见工具数
  nextCount: number;         // 新规则下可见工具数
  gained: string[];          // 新增可见的工具全名
  lost: string[];            // 失去可见的工具全名
  affectedKeys: KeyItem[];   // 受影响的 Key 明细
  affectedKeyCount: number;  // 受影响的 Key 数量
  recent7dCalls: number;     // 这些 Key 近 7 日总调用量
  lostCalls7d: number;       // 其中打在 lost 工具上的调用量 ← 最关键的数字
  gainedCalls7d: number;     // 其中打在 gained 工具上的调用量
  totalImpact7d: number;     // lostCalls7d + gainedCalls7d
}

buildAllowedSet(role: string, allNames: string[], registryVersion: number): Set<string> {
  const key = `${role}|${registryVersion}|${this.rulesVersion}`;
  const hit = this.cache.get(key);
  if (hit && Date.now() - hit.at < 60_000) return hit.set;   // 60s TTL 兜底

  const set = new Set<string>();
  for (const n of allNames) if (this.compute(role, n)) set.add(n);

  this.cache.set(key, { set, at: Date.now() });
  return set;
}

@Injectable()
export class AdminTokenGuard implements CanActivate {
  canActivate(ctx: ExecutionContext): boolean {
    const req = ctx.switchToHttp().getRequest();

    // 例外：SSE 端点（EventSource 无法自定义 Header），允许 ?token= 查询参数
    if (req.path.endsWith('/stream')) {
      const q = req.query?.token;
      if (q && this.crypto.safeEqual(q, this.config.adminToken)) return true;
      throw new UnauthorizedException('无效的管理令牌');
    }

    const h = req.headers['authorization'] || '';
    const token = h.startsWith('Bearer ') ? h.slice(7) : '';
    if (!token || !this.crypto.safeEqual(token, this.config.adminToken)) {
      throw new UnauthorizedException('无效的管理令牌');
    }
    return true;
  }
}

/** 主密钥派生：64 位 hex 直接用，否则 SHA-256 展开成 32 字节 */
private masterKey(secret: string): Buffer {
  return /^[0-9a-f]{64}$/i.test(secret)
    ? Buffer.from(secret, 'hex')
    : createHash('sha256').update(secret).digest();
}

encrypt(plain: string): string {
  const iv  = randomBytes(12);                              // GCM 推荐 96 bit
  const c   = createCipheriv('aes-256-gcm', this.key, iv);
  const enc = Buffer.concat([c.update(plain, 'utf8'), c.final()]);
  const tag = c.getAuthTag();                               // 16 字节认证标签
  return Buffer.concat([iv, tag, enc]).toString('base64');  // iv|tag|ct 自描述
}

decrypt(payload: string): string {
  const buf = Buffer.from(payload, 'base64');
  const iv = buf.subarray(0, 12), tag = buf.subarray(12, 28), ct = buf.subarray(28);
  const d = createDecipheriv('aes-256-gcm', this.key, iv);
  d.setAuthTag(tag);
  return Buffer.concat([d.update(ct), d.final()]).toString('utf8');
}

take(keyId: number, tool: string, rate = this.defaultQpm, burst = this.defaultBurst) {
  const id = `${keyId}|${tool}`;                    // ★ 双维度桶键
  const now = Date.now();
  let b = this.buckets.get(id);
  if (!b) { b = { tokens: burst, last: now }; this.buckets.set(id, b); }

  const capacity = burst;
  // 按经过时间线性补充令牌，上限为 capacity
  b.tokens = Math.min(capacity, b.tokens + ((now - b.last) / 60_000) * rate);
  b.last = now;

  if (b.tokens < 1) {
    const need = 1 - b.tokens;
    return { ok: false, retryAfterS: Math.ceil((need / rate) * 60) };
  }
  b.tokens -= 1;
  return { ok: true, retryAfterS: 0 };
}

private readonly failThreshold = 5;        // 窗口内失败阈值
private readonly openMs        = 60_000;   // open 持续时间
private readonly windowMs      = 60_000;   // 失败计数滑动窗口

allow(upstream: string): 'closed' | 'half-open' | 'open' {
  const s = this.state(upstream);
  const now = Date.now();

  if (s.state === 'open') {
    if (now - s.openedAt < this.openMs) return 'open';
    s.state = 'half-open';                 // 时间到，进入探测态
    s.probing = false;
  }
  if (s.state === 'half-open') {
    if (s.probing) return 'open';          // ★ 已有探针在飞，其余请求仍拒
    s.probing = true;                      // 独占探针位
    return 'half-open';
  }
  return 'closed';
}

onFailure(upstream: string, kind: 'client-error' | 'upstream-error'): BreakerTransition | null {
  if (kind === 'client-error') return null;  // ★ BAD_ARGS / FORBIDDEN 不计入熔断
  const s = this.state(upstream);
  const now = Date.now();
  s.fails = s.fails.filter((t) => now - t < this.windowMs);
  s.fails.push(now);

  if (s.state === 'half-open') {            // 探针失败 → 重新 open，重新计时
    s.state = 'open'; s.openedAt = now; s.probing = false;
    return { kind: 'opened', from: 'half-open', to: 'open', fails: s.fails.length };
  }
  if (s.fails.length >= this.failThreshold) {
    const from = s.state;
    s.state = 'open'; s.openedAt = now;
    return { kind: 'opened', from, to: 'open', fails: s.fails.length };
  }
  return null;
}

record(t: BreakerTransition, upstream: string, reason?: string): void {
  // 1) 落库 circuit_breaker_events
  this.repo.insert({
    ts: Date.now(), upstream,
    event: t.kind,                    // 'opened' | 'recovered'
    fromState: t.from, toState: t.to,
    failsInWindow: t.fails,
    openMs: t.kind === 'opened' ? this.openMs : null,
    reason: reason || '',
  });
  // 2) 推进 SSE Subject，前端熔断事件页实时刷新
  this.stream$.next({ type: 'breaker', data: { … } });
  // 3) IM 告警（带 5min 同类抑制）
  t.kind === 'opened' ? this.im.circuitOpened(upstream) : this.im.circuitClosed(upstream);
}

async acquire(upstream: string, limit: number, maxQueue = 50) {
  const s = this.state(upstream);
  const t0 = Date.now();

  if (s.active < limit) {                        // 有空位，直接拿
    s.active++;
    return { ok: true, waitedMs: 0, release: () => this.release(upstream) };
  }
  if (s.waiters.length >= maxQueue) {            // 队列满，立即拒绝
    return { ok: false, waitedMs: Date.now() - t0, release: () => {} };
  }

  // 排队等待，直到被前一个 release 唤醒
  await new Promise<void>((resolve) => s.waiters.push(resolve));
  s.active++;
  return { ok: true, waitedMs: Date.now() - t0, release: () => this.release(upstream) };
}

private release(upstream: string): void {
  const s = this.state(upstream);
  s.active = Math.max(0, s.active - 1);          // 防重复 release 打成负数
  const next = s.waiters.shift();
  if (next) next();                              // ★ 必须唤醒等待者，否则永久悬挂
}

log(entry: AuditEntry): void {           // 同步、非阻塞，调用方永不 await
  this.queue.push(this.toEntity(entry));
  if (this.queue.length >= this.flushSize) void this.flush();   // 默认 100 条
}

private async flush(): Promise<void> {
  if (!this.queue.length || this.flushing) return;
  this.flushing = true;
  const batch = this.queue.splice(0, this.queue.length);
  try {
    await this.dataSource.transaction(async (m) => {            // 单事务
      for (const e of batch) await m.getRepository(AuditLog).insert(e);
    });
  } catch (err) {
    this.writeFallback(batch);                                  // 落 JSONL 兜底
    this.im.auditFallbackGrowing(this.fallbackSize());
  } finally {
    this.flushing = false;
  }
}

// 定时器：默认每 1000ms 强制 flush 一次
setInterval(() => void this.flush(), this.flushMs);

private toEntity(e: AuditEntry): AuditLog {
  const argsText = JSON.stringify(e.args ?? {});
  const sensitive = e.sensitivity === 'sensitive';

  return {
    …e,
    // 普通工具：只存 SHA-256 前 16 位 + 脱敏后的摘要
    argsDigest: sha256(argsText).slice(0, 16),
    // 敏感工具：存脱敏后的全量 JSON（便于事后取证）
    argsFull: sensitive ? this.redact.redactDeep(e.args) : null,
    redacted: sensitive ? 1 : 0,
    resultBytes: e.resultBytes ?? null,
  };
}

const RULES: Array<{ re: RegExp; mask: (m: string) => string }> = [
  { re: /\b1[3-9]\d{9}\b/g,        mask: (m) => m.slice(0, 3) + '****' + m.slice(-4) },  // 手机号
  { re: /[\w.+-]+@[\w-]+\.[\w.]+/g, mask: (m) => m[0] + '***' + m.slice(m.indexOf('@')) },// 邮箱
  { re: /\b\d{17}[\dXx]\b/g,       mask: (m) => m.slice(0, 6) + '********' + m.slice(-4) }, // 身份证
  { re: /\b\d{16,19}\b/g,          mask: (m) => m.slice(0, 4) + '****' + m.slice(-4) },  // 银行卡
];

redactDeep(value: unknown): unknown {
  if (typeof value === 'string') return this.redactString(value);
  if (Array.isArray(value))      return value.map((v) => this.redactDeep(v));
  if (value && typeof value === 'object') {
    const out: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(value)) out[k] = this.redactDeep(v);
    return out;
  }
  return value;
}

await this.changeLog.record({
  actor: req.headers['x-admin-user'] || 'system',
  actorIp: req.ip,
  targetType: 'role_rules',        // 7 种之一
  targetId: 'support',
  action: 'update',                // 8 种之一
  beforeJson: redactDeep(before),  // 变更前完整对象（脱敏后）
  afterJson:  redactDeep(after),
  diffSummary: 'support 角色新增 deny crm__delete_record',  // 人类可读
});

async run(trigger: 'manual' | 'cron', actor = 'system'): Promise<Backup> {
  const t0 = Date.now();
  const filename = `gateway-${fmtTs(new Date())}-${trigger}.db`;
  const absPath  = join(this.backupDir, filename);

  try {
    // SQLite 在线备份 API：逐页复制，期间写入不阻塞
    this.dataSource.driver.master.backup(absPath);
    const bytes = statSync(absPath).size;

    const rec = await this.repo.save({
      ts: Date.now(), filename, absPath, bytes,
      durationMs: Date.now() - t0, trigger, status: 'ok', actor,
    });
    await this.prune();                       // 滚动保留 backup.keep 份
    return rec;
  } catch (e) {
    await this.repo.save({ …, status: 'failed', errorMsg: safeMessage(e) });
    this.im.backupFailed(trigger, safeMessage(e));   // P1-D
    throw e;
  }
}

const PROTOCOL_VERSION  = '2025-06-18';   // 网关首选
const FALLBACK_PROTOCOL = '2025-03-26';   // 客户端不认识首选版本时回落

// initialize 处理
const requested = params?.protocolVersion;
const negotiated = requested === PROTOCOL_VERSION ? PROTOCOL_VERSION : FALLBACK_PROTOCOL;

return {
  jsonrpc: '2.0', id,
  result: {
    protocolVersion: negotiated,
    capabilities: { tools: { listChanged: true } },   // ★ 声明支持工具变更推送
    serverInfo: { name: 'mcp-gateway', version: '2.0.0' },
    instructions: '…',                                 // 给模型的使用说明
  },
};
// 响应头下发会话 ID
res.setHeader('mcp-session-id', sessionId);

interface Session {
  id: string;                       // uuid，即 Mcp-Session-Id
  keyId: number;
  role: string;
  userName: string | null;
  team: string | null;
  callerIp: string | null;
  clientInfo: string | null;        // initialize 时客户端自报的 name/version
  createdAt: number;
  lastActiveAt: number;             // 每次请求刷新
  stream$: Subject<SseEvent>;       // 该会话的通知流
  res?: Response;                   // 当前挂着的 SSE 响应对象
}

@Get()
async sse(@Req() req, @Res() res: Response) {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache, no-transform',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',        // ★ 关键：告诉 Nginx 不要缓冲
  });

  const session = this.sessions.get(req.headers['mcp-session-id']);
  const sub = session.stream$.subscribe((ev) => {
    res.write(`event: message\ndata: ${JSON.stringify(ev)}\n\n`);
  });

  const heartbeat = setInterval(() => res.write(': ping\n\n'), 25_000);

  req.on('close', () => { clearInterval(heartbeat); sub.unsubscribe(); });
}

const allowed = this.rbac.buildAllowedSet(caller.role, this.registry.allToolNames(), this.registry.version);
const tools   = this.registry.listFor(allowed);   // 已应用覆写、已过滤 disabled

return {
  result: {
    tools: tools.map((t) => ({
      name: t.fullName,                  // ★ 暴露全名，不是原名
      title: t.def.title,
      description: t.def.description,     // 可能已被 tool_overrides 覆写
      inputSchema: t.def.inputSchema,
    })),
  },
};

const CallParams = z.object({
  name: z.string().min(1).max(200),
  arguments: z.record(z.unknown()).optional().default({}),
});

// common/zod-exception.filter.ts —— 全局过滤器
@Catch(ZodError)
export class ZodExceptionFilter implements ExceptionFilter {
  catch(err: ZodError, host: ArgumentsHost) {
    const first = err.issues[0];
    const msg = first
      ? `${first.path.join('.')}: ${first.message}`
      : '参数校验失败';
    host.switchToHttp().getResponse()
        .status(400).json({ statusCode: 400, message: msg, error: 'Bad Request' });
  }
}

-- 先数总量
SELECT COUNT(*) AS n FROM audit_logs
 WHERE upstream = ? AND ts BETWEEN ? AND ? AND duration_ms IS NOT NULL;

-- 再取第 floor(n * 0.95) 条（已按耗时升序）
SELECT duration_ms FROM audit_logs
 WHERE upstream = ? AND ts BETWEEN ? AND ? AND duration_ms IS NOT NULL
 ORDER BY duration_ms
 LIMIT 1 OFFSET ?;   -- ? = floor(n * 0.95)

// frontend/src/main.ts
import { createApp } from 'vue';
import { createPinia } from 'pinia';
import App from './App.vue';
import { router } from './router';
import 'virtual:uno.css';        // UnoCSS 生成的原子类
import './styles/base.css';     // 全局观感（背景光晕/滚动条/字体）

createApp(App).use(createPinia()).use(router).mount('#app');

const routes: RouteRecordRaw[] = [
  { path: '/login', name: 'login', component: () => import('../views/Login.vue'),
    meta: { public: true } },
  { path: '/', component: () => import('../layouts/AdminLayout.vue'), children: [
      { path: '', redirect: '/dashboard' },
      { path: 'dashboard',   name: 'dashboard',   component: () => import('../views/Dashboard.vue'),
        meta: { title: '仪表盘',     icon: 'gauge'  } },
      // … 其余 9 页 …
  ]},
  { path: '/:pathMatch(.*)*', redirect: '/dashboard' },
];

router.beforeEach((to) => {
  const auth = useAuthStore();
  if (to.meta.public) {
    if (to.name === 'login' && auth.authenticated) return { name: 'dashboard' };
    return true;
  }
  if (!auth.authenticated) return { name: 'login', query: { redirect: to.fullPath } };
  return true;
});

/** 侧边栏导航项从路由表派生，避免两处维护 */
export const navItems = routes[1].children!
  .filter((r) => r.meta?.title)
  .map((r) => ({ name: r.name, path: `/${r.path}`, title: r.meta!.title, icon: r.meta!.icon }));

const TOKEN_KEY = 'mgw_admin_token';
const USER_KEY  = 'mgw_admin_user';

export class ApiError extends Error {
  constructor(public status: number, message: string, public body?: unknown) { super(message); }
}

export const tokenStore = {
  get token() { return localStorage.getItem(TOKEN_KEY) || ''; },
  get user()  { return localStorage.getItem(USER_KEY)  || ''; },
  set(token: string, user: string) { /* 写两个键 */ },
  clear() { /* 删两个键 */ },
};

let onUnauthorized: (() => void) | null = null;
export function setUnauthorizedHandler(fn: () => void) { onUnauthorized = fn; }

async function request<T>(method: string, path: string, body?: unknown): Promise<T> {
  const res = await fetch(`/admin/api${path}`, {
    method,
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${tokenStore.token}`,
      'x-admin-user': tokenStore.user,
    },
    body: body ? JSON.stringify(body) : undefined,
  });
  if (res.status === 401) { tokenStore.clear(); onUnauthorized?.(); }
  if (!res.ok) throw new ApiError(res.status, (await res.json().catch(() => ({}))).message || res.statusText);
  return res.status === 204 ? (undefined as T) : res.json();
}

export const http = {
  get:  <T>(p: string)          => request<T>('GET', p),
  post: <T>(p: string, b?: any) => request<T>('POST', p, b),
  put:  <T>(p: string, b?: any) => request<T>('PUT', p, b),
  del:  <T>(p: string)          => request<T>('DELETE', p),
};

/** CSV 导出：blob + a.download，避免浏览器直接打开乱码文本 */
export async function downloadCsv(path: string, filename: string) { … }

export const api = {
  upstreams:   { list, create, update, remove, probe, refresh },
  tools:       { list, detail, setDescription, setEnabled, setRatelimit },
  permissions: { roles, createRole, deleteRole, rules, saveRules, rollback, preview, matrix },
  keys:        { issue, list, revoke, rotate },
  audit:       { list, detail, exportCsv, changes },
  stats:       { overview, timeseries },
  system:      { health, metrics, registry, registryRefresh, settings, saveSettings },
  breaker:     { status, events, summary, streamUrl },
  backup:      { list, summary, run, downloadUrl },
};

export const useAuthStore = defineStore('auth', () => {
  const token = ref(tokenStore.token);
  const user  = ref(tokenStore.user);
  const authenticated = computed(() => !!token.value);

  function login(t: string, u: string) { tokenStore.set(t, u); token.value = t; user.value = u; }
  function logout() { tokenStore.clear(); token.value = ''; user.value = ''; }

  return { token, user, authenticated, login, logout };
});

// frontend/vite.config.ts
export default defineConfig({
  plugins: [vue(), UnoCSS()],
  server: {
    port: 5173,
    proxy: {
      '/admin/api': { target: 'http://127.0.0.1:8080', changeOrigin: true },
      '/mcp':       { target: 'http://127.0.0.1:8080', changeOrigin: true },
      '/healthz':   { target: 'http://127.0.0.1:8080', changeOrigin: true },
    },
  },
  build: { outDir: 'dist', sourcemap: false },
});

// frontend/uno.config.ts
theme: {
  colors: {
    ink: {
      900: '#08080a',   // 页面底
      800: '#0d0d10',   // 深面板
      700: '#141418',   // 卡片
      600: '#1c1c22',   // 悬浮/激活
    },
  },
},

:root {
  color-scheme: dark;
  --mgw-bg: #08080a;
  --mgw-font: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC',
              'Hiragino Sans GB', 'Microsoft YaHei', 'Helvetica Neue', Helvetica, Arial, sans-serif;
  --mgw-mono: 'SF Mono', 'JetBrains Mono', 'Fira Code', Menlo, Consolas, 'Liberation Mono', monospace;
}

/* 三层径向光晕 */
.mgw-backdrop {
  position: fixed; inset: 0; z-index: -1; pointer-events: none;
  background:
    radial-gradient(60rem 40rem at 12%  -8%, rgba(255,255,255,.055), transparent 60%),
    radial-gradient(50rem 40rem at 108% 12%, rgba(255,255,255,.035), transparent 60%),
    radial-gradient(40rem 30rem at 50% 120%, rgba(255,255,255,.028), transparent 60%);
}
/* 48px 网格 + 径向渐隐遮罩 */
.mgw-backdrop::after {
  content: ''; position: absolute; inset: 0;
  background-image:
    linear-gradient(rgba(255,255,255,.028) 1px, transparent 1px),
    linear-gradient(90deg, rgba(255,255,255,.028) 1px, transparent 1px);
  background-size: 48px 48px;
  mask-image: radial-gradient(ellipse 90% 70% at 50% 30%, #000 20%, transparent 78%);
}

body { overflow: hidden; }               /* 滚动交给布局内部容器 */
.mono { font-family: var(--mgw-mono); }
::-webkit-scrollbar { width: 8px; height: 8px; }
::-webkit-scrollbar-thumb { background: rgba(255,255,255,.12); border-radius: 8px; }
::selection { background: #fff; color: #000; }
:focus-visible { outline: 1px solid rgba(255,255,255,.45); outline-offset: 2px; }
.fade-enter-active, .fade-leave-active { transition: opacity .15s ease; }
.fade-enter-from, .fade-leave-to { opacity: 0; }
.mask-enter-active, .mask-leave-active { transition: opacity .2s ease; }
.mask-enter-from, .mask-leave-to { opacity: 0; }

// backend/src/config/config.service.ts
const str  = (env, jsonVal, def: string)  => env ?? (jsonVal != null ? String(jsonVal) : def);
const num  = (env, jsonVal, def: number)  => Number(env ?? jsonVal ?? def);
const list = (env, jsonVal, def: string[]) => {
  if (env != null && env.trim() !== '') return env.split(',').map(s => s.trim()).filter(Boolean);
  if (Array.isArray(jsonVal))           return jsonVal.map(s => String(s).trim()).filter(Boolean);
  return def;
};

// gateway.json 不存在或解析失败 → 返回 {}，只打一条 warn，绝不阻断启动
function readJsonConfig(): Record<string, any> { … }

{
  "port": 8080,
  "host": "127.0.0.1",
  "publicOrigin": "http://127.0.0.1:8080",
  "allowedOrigins": ["http://localhost:5173", "http://127.0.0.1:5173"],
  "outboundAllowedCidrs": ["10.0.0.0/8", "192.168.0.0/16"],
  "maxSessions": 300,
  "sessionIdleTimeoutS": 1800,
  "stdioMax": 8,
  "registryRefreshIntervalS": 300,
  "auditFlushMs": 1000,
  "auditFlushSize": 100,
  "dataDir": "./data"
}

{
  "status": "ok",                  // ok | degraded | down
  "uptime_s": 12843,               // 进程运行秒数
  "version": "2.0.0",
  "upstreams": { "total": 4, "up": 3, "down": 1 },
  "registry": { "tools": 87, "version": 23 },
  "sessions": 12,
  "audit_queue": 0,                // 持续 > 0 说明写入跟不上
  "el_p99_ms": 3.2                 // 事件循环延迟 p99
}

/** 1) 排除探针端点，避免自我观测污染业务指标 */
function isExcluded(path: string): boolean {
  return path === '/metrics' || path === '/metrics.prom' || path === '/healthz'
      || path.startsWith('/metrics/');
}

/** 2) 路由归一化：用 baseUrl + route.path，而非原始 URL */
function normalizeRoute(req): string {
  // '/admin/api/keys/42/revoke' → '/admin/api/keys/:id/revoke'
  return req.baseUrl + (req.route?.path ?? '') || '__unmatched__';
}

/** 3) status 归类到百位，控制标签基数 */
const statusLabel = `${Math.floor(res.statusCode / 100) * 100}`;   // '200' / '404' / '500'

// 4) SSE 长连接跳过延时埋点
if (res.getHeader('content-type')?.includes('text/event-stream')) return;

const RESOLUTION_MS      = 20;         // 采样精度
const SAMPLE_INTERVAL_MS = 5_000;       // 每 5s 读一次
const ALERT_P99_MS       = 200;         // p99 超过 200ms 告警
const ALERT_COOLDOWN_MS  = 5 * 60_000;  // 5 分钟冷却，防告警风暴

const history = monitorEventLoopDelay({ resolution: RESOLUTION_MS });
history.enable();

setInterval(() => {
  const p99 = history.percentile(99) / 1e6;      // ns → ms
  const p50 = history.percentile(50) / 1e6;
  healthReporter.setEventLoop({ p50, p99 });     // 供 /healthz 的 el_p99_ms
  history.reset();                               // ★ 必须重置，否则被历史数据钝化

  if (p99 > ALERT_P99_MS && now - lastAlert > ALERT_COOLDOWN_MS) {
    lastAlert = now;
    im.eventLoopSlow(p99);
    logger.warn({ event: 'event_loop_slow', p99_ms: p99 });
  }
}, SAMPLE_INTERVAL_MS);

export const logger = {
  info:  (o: object) => write('info', o),
  warn:  (o: object) => write('warn', o),
  error: (o: object) => write('error', o),
  debug: (o: object) => { if (process.env.NODE_ENV !== 'production') write('debug', o); },
};

function write(level: string, o: object): void {
  process.stderr.write(JSON.stringify({
    level, ts: Date.now(), ...o,
  }) + '\n');
}

# 工具调用错误率（5 分钟窗口）
sum(rate(mgw_tool_calls_total{status!="ok"}[5m]))
  / sum(rate(mgw_tool_calls_total[5m]))

# 按上游的 P95 延时
histogram_quantile(0.95,
  sum by (le, upstream) (rate(mgw_tool_call_duration_ms_bucket[5m])))

# 当前处于熔断态的上游
mgw_upstream_breaker_state == 2

# 审计写入是否跟不上（持续大于 0 即告警）
mgw_audit_queue_depth > 0

# HTTP 5xx 比例
sum(rate(mgw_http_requests_total{status="500"}[5m]))
  / sum(rate(mgw_http_requests_total[5m]))

export class GatewayToolError extends Error {
  constructor(
    public readonly code: ErrorCodeType,
    public readonly safeMessage: string,
    public readonly rawError?: string,
    public readonly kind: 'client-error' | 'upstream-error' = 'upstream-error',
    public readonly meta: Record<string, unknown> = {},
  ) {
    super(`[${code}] ${safeMessage}`);      // ★ 固定格式
  }

  /** 客户端问题不计入熔断 */
  get isClientError(): boolean { return this.kind === 'client-error'; }
}

upstream mgw_backend {
    server 127.0.0.1:8080;
    keepalive 32;
}

server {
    listen 443 ssl http2;
    server_name mcp.example.com;

    # ── 管理后台静态资源（frontend/dist）──
    root /opt/mcp-gateway/frontend/dist;
    index index.html;
    location / {
        try_files $uri $uri/ /index.html;   # hash 路由其实不需要，但留着更稳
    }

    # ── MCP 端点：必须支持 SSE ──
    location /mcp {
        proxy_pass http://mgw_backend;
        proxy_http_version 1.1;
        proxy_set_header Host              $host;
        proxy_set_header X-Real-IP         $remote_addr;
        proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # ★ SSE 三件套：关缓冲、长超时、保持连接
        proxy_buffering        off;
        proxy_cache            off;
        proxy_read_timeout     3600s;
        proxy_send_timeout     3600s;
        proxy_set_header       Connection '';
        chunked_transfer_encoding off;
    }

    # ── 管理 API ──
    location /admin/api/ {
        proxy_pass http://mgw_backend;
        proxy_set_header Host            $host;
        proxy_set_header X-Real-IP       $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        client_max_body_size 8m;
    }

    # ── 熔断事件 SSE（同样要关缓冲）──
    location ~ ^/admin/api/breaker/events/stream {
        proxy_pass http://mgw_backend;
        proxy_http_version 1.1;
        proxy_buffering off;
        proxy_read_timeout 3600s;
        proxy_set_header Connection '';
        access_log off;              # ★ token 在 query 里，不要记进访问日志
    }

    # ── 探针端点：仅限内网 ──
    location ~ ^/(healthz|metrics|metrics\.prom)$ {
        allow 10.0.0.0/8;
        allow 127.0.0.1;
        deny  all;
        proxy_pass http://mgw_backend;
    }
}

# /etc/systemd/system/mcp-gateway.service
[Unit]
Description=MCP Gateway
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=mcpgw
Group=mcpgw
WorkingDirectory=/opt/mcp-gateway/backend

# .env.example 格式可直接被 EnvironmentFile 消费
EnvironmentFile=/etc/mcp-gateway.env
Environment=NODE_ENV=production

ExecStart=/usr/bin/node dist/main
Restart=always
RestartSec=3

# ── 加固 ──
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/mcp-gateway/backend/data
LimitNOFILE=65536

# 优雅停机：给审计队列排空的时间
TimeoutStopSec=20
KillSignal=SIGTERM

[Install]
WantedBy=multi-user.target

# 1) 先看当前占用
sqlite3 backend/data/gateway.db \
  "SELECT COUNT(*) AS rows, page_count*page_size/1024/1024 AS mb FROM audit_logs, pragma_page_count(), pragma_page_size();"

# 2) 导出超期数据留档（走 Admin API 更安全，带鉴权与脱敏）
curl -H "Authorization: Bearer $ADMIN_TOKEN" \
  "http://127.0.0.1:8080/admin/api/audit/export?from=0&to=$(( $(date +%s%3N) - 180*86400000 ))" \
  -o audit-archive-$(date +%Y%m%d).csv

# 3) 停服务后归档并回收空间（DELETE 不会自动缩小文件）
sqlite3 backend/data/gateway.db "DELETE FROM audit_logs WHERE ts < $(( $(date +%s%3N) - 180*86400000 ));"
sqlite3 backend/data/gateway.db "VACUUM;"