MCP 企业治理：RBAC、审批、审计与影子工具管控

# 1) 权限风险: 一个连接了云账号的 MCP 服务器 = 一把万能钥匙
# 2) 合规风险: 数据出境、PII 泄漏、操作不可审计
# 3) 影子风险: 员工私自接入未审批的服务器/工具
# 治理的目标不是"禁止"，而是"可控地放开"

// 工具级 RBAC：先看可见性，再看可执行性
interface Policy {
  effect: "allow" | "deny";
  roles: string[];
  tools: string[];          // 支持通配符 github__*
  conditions?: {
    requireApproval?: boolean;
    maxPerHour?: number;
    timeWindow?: string;    // 如 "09:00-18:00"
  };
}

function authorize(principal: Principal, tool: string): Decision {
  const rules = policies.filter(
    (p) => p.roles.includes(principal.role) && matchGlob(p.tools, tool)
  );
  // deny 优先于 allow（显式拒绝覆盖一切）
  if (rules.some((r) => r.effect === "deny")) return { allow: false };
  const allow = rules.find((r) => r.effect === "allow");
  if (!allow) return { allow: false, reason: "no matching allow rule" };
  return {
    allow: true,
    requireApproval: allow.conditions?.requireApproval ?? false,
  };
}

# 关键原则: 不可见的工具不存在
# 1) tools/list 只返回该主体被授权的工具
# 2) 模型无法"猜测"未列出的工具名（tools/call 也会被拒）
# 3) 描述里不泄漏未授权工具的存在
# 4) 按角色/部门/租户动态裁剪，而非静态全局列表
# 权限的边界要在"列表层"就收紧，而不是只靠"调用层"拦截

# 高危工具在网关侧拦截，返回"待审批"而非直接执行
async def call_tool(name: str, args: dict, principal: Principal):
    decision = authorize(principal, name)
    if not decision.allow:
        raise McpError(-32003, "forbidden")

    if decision.require_approval:
        ticket = approvals.create(
            principal=principal.id,
            tool=name,
            args=args,
            risk=assess_risk(name, args),
            approvers=resolve_approvers(name),
        )
        # 不是错误，而是"已受理"：模型知道要等人
        return {
            "status": "pending_approval",
            "ticket_id": ticket.id,
            "hint": "该操作需人工审批，审批通过后会自动执行",
        }

    return await execute(name, args)

# 模型驱动的二次确认（MCP 的 elicitation/sampling 思路）
# 1) 工具返回"确认请求"，列出将被影响的具体资源
# 2) 客户端展示给用户，用户明确确认
# 3) 模型带确认令牌再次调用，服务器校验令牌后执行
# 确认要"具体到资源": "确认删除 3 个 Pod: a/b/c"，而非"确认执行吗?"

// 审计事件追加写 + 哈希链，防篡改
interface AuditEvent {
  seq: number;
  prev_hash: string;
  payload: Record<string, unknown>;
  hash: string;
}

function appendAudit(prev: AuditEvent | null, payload: Record<string, unknown>): AuditEvent {
  const seq = (prev?.seq ?? 0) + 1;
  const prevHash = prev?.hash ?? "genesis";
  const hash = sha256(prevHash + JSON.stringify(payload));
  return { seq, prev_hash: prevHash, payload, hash };
}
// 校验时重算整条链，任一环节被改动即被发现

# 1) 留存周期: 按合规要求（金融常 5-7 年，一般 1 年）
# 2) 参数脱敏: 密码/token/PII 落库前替换为 [REDACTED]
# 3) 内容留存: 是否存参数全文取决于合规；至少存摘要 + 哈希
# 4) 导出接口: 支持按主体/时间导出，应对监管问询
# 5) 冷热分层: 近 90 天热存可查，更早转冷归档
# 审计的"可用性"和"完整性"同等重要——查不到等于没有

// 租户 ID 由网关从令牌解析，绝不接受模型传参
function buildBackendContext(ctx: CallContext) {
  const tenant = ctx.token.tenant_id;   // 来自已验证的令牌
  if (!tenant) throw new McpError(-32001, "missing tenant");
  return {
    tenant,
    // 后端据此做数据隔离；模型无法伪造
    credential: credentialVault.get(tenant, ctx.backendId),
    quota: quotaStore.scope(tenant),
  };
}

# 1) 模型传 tenant_id: 可被提示注入篡改 → 必须来自令牌
# 2) 缓存键不含租户: A 的结果被 B 命中 → 缓存键含 tenant
# 3) 连接池跨租户复用: 凭证串号 → 按租户分池
# 4) 日志混存: 越权读取 → 分租户存储 + 访问控制
# 多租户的黄金法则: 租户身份只在"可信边界"内产生，永不来自输入

# policies/
#   roles.yaml          # 角色定义
#   bindings.yaml       # 主体 → 角色绑定
#   tools.yaml          # 工具清单与风险级
#   tenants.yaml        # 租户配置
#   tests/              # 策略测试用例
# CI: 策略变更 → 语法校验 → 单元测试 → 评审 → 生效

# policies/tools.yaml
tools:
  - name: github__create_pr
    risk: medium
    allowed_roles: [developer, operator, admin]
  - name: k8s__delete_namespace
    risk: critical
    allowed_roles: [admin]
    require_approval: true
    approvers: [sre-lead]
    max_per_hour: 2
  - name: aws__terminate_instance
    risk: critical
    allowed_roles: [admin]
    require_approval: true
    deny_if:
      - "instance.env == 'prod' and principal.role != 'sre-admin'"

def test_critical_tool_requires_approval():
    d = authorize(principal(role="admin"), "k8s__delete_namespace")
    assert d.allow is True
    assert d.require_approval is True

def test_developer_cannot_delete_namespace():
    d = authorize(principal(role="developer"), "k8s__delete_namespace")
    assert d.allow is False

def test_prod_terminate_denied_for_non_sre():
    d = authorize(principal(role="admin"), "aws__terminate_instance",
                  ctx={"instance": {"env": "prod"}})
    assert d.allow is False

# 1) 员工本地: 在 IDE 里直连未审批的 server
# 2) 服务器漂移: 已审批的 server 新增了未审批的工具
# 3) 依赖漂移: 服务器升级带入了新的下游能力
# 4) 命名伪装: 工具名看似无害，实际是写操作
# 影子工具的危险在于"看不见"，治理的第一步是"看得见"

# 定期比对：注册目录 vs 实际能力，发现未登记工具
def detect_drift(registry: dict, actual: dict) -> list[str]:
    findings = []
    for ns, tools in actual.items():
        registered = set(registry.get(ns, {}).get("tools", []))
        for t in tools:
            if t not in registered:
                findings.append(f"unregistered tool: {ns}__{t}")
        # 反向：注册了但实际没有（可能被摘除）
        for t in registered - set(tools):
            findings.append(f"missing tool: {ns}__{t}")
    return findings

# 1) 恶意服务器: 以工具为名窃取凭证/数据
# 2) 依赖投毒: 服务器依赖链中的恶意包
# 3) 提示注入: 服务器返回的内容诱导模型越权
# 4) 过度索取: 申请远超功能的权限（如全盘读写）
# 5) 无维护: 停更服务器积累未修复漏洞

# CI 中的 MCP server 准入检查
stages:
  - name: sbom
    run: syft packages dir:. -o spdx-json > sbom.json
  - name: vuln_scan
    run: grype sbom.json --fail-on high
  - name: capability_review
    run: mcp-inspect tools --server ./server --assert-minimal
  - name: egress_check
    run: mcp-inspect network --server ./server --expect none

# 1) 覆盖率: 接入网关的服务器占比（目标 100%）
# 2) 影子率: 未登记工具数 / 总工具数（目标 → 0）
# 3) 审批时效: 高危操作从申请到审批的时长
# 4) 拒绝率: 被策略拒绝的调用比例（异常升高需排查）
# 5) 审计完整率: 有完整审计记录的调用占比（目标 100%）
# 6) 准入周期: 新服务器从申请到上线的时长

# 1) 每日: 影子工具漂移报告
# 2) 每周: 高危操作审批复盘
# 3) 每月: 权限回收（离职/转岗）、配额调整
# 4) 每季: 服务器准入复审、策略评审
# 5) 按需: 安全事件响应
# 治理要"有节奏"，否则会退化成"出事才管"