MCP 云基础设施与 IaC 工具：plan/apply 分离与爆炸半径控制

# 观察类（安全）: 查资源、看指标、读日志、列成本
# 计划类（安全）: 生成变更计划（plan/diff），不改任何东西
# 变更类（危险）: 创建/修改/删除资源（apply/scale/restart）
# 破坏类（极危）: 删库、删集群、改 IAM、改网络
# 工具设计的第一原则: 让模型在"观察+计划"层自由，在"变更"层受限

# L1 低危变更: 打标签、改描述、加注解（可逆、无影响）
# L2 中危变更: 扩缩容、重启、更新镜像（有影响但可控）
# L3 高危变更: 改网络/安全组、改 IAM、删无状态资源
# L4 极危变更: 删数据、删集群、改计费、动生产数据库
# 分级决定: 谁可用、是否需审批、是否需 dry-run、是否限流

// 每个工具声明风险级，网关据此套用策略
const TOOL_RISK: Record<string, RiskLevel> = {
  list_resources: "none",
  estimate_cost: "none",
  scale_deployment: "medium",
  update_security_group: "high",
  delete_database: "critical",
};

function guard(tool: string, ctx: CallContext) {
  const risk = TOOL_RISK[tool] ?? "high";   // 未标注一律按 high
  if (risk === "critical" && !ctx.flags.allowCritical) {
    throw new McpError(-32003, "critical operation requires approval");
  }
  if (risk === "high" && !ctx.approvalToken) {
    throw new McpError(-32003, "high-risk operation requires dry-run + approval");
  }
}

# terraform_plan:
#   输入: 工作目录/变量/目标
#   输出: 计划摘要（+N ~M -K）、完整 plan 文件 ID、风险摘要
#   副作用: 无（只读 state 与 provider API）
# terraform_apply:
#   输入: plan 文件 ID（必须是本次生成的、未过期的）
#   输出: 执行结果
#   副作用: 改变真实资源
# 关键: apply 只接受 plan_id，不接受"重新生成计划"

async def terraform_plan(workspace: str, var_file: str | None = None,
                         targets: list[str] | None = None) -> dict:
    ws = resolve_workspace(workspace)             # 白名单校验
    plan = await tf.plan(ws, var_file=var_file, targets=targets, lock_timeout="30s")
    # 解析计划，做风险摘要（尤其是销毁操作）
    summary = {
        "add": plan.count("add"),
        "change": plan.count("change"),
        "destroy": plan.count("destroy"),
        "destroys": plan.resource_addresses("destroy"),  # 将被销毁的资源清单
    }
    # 计划文件短期有效，绑定会话
    plan_id = plan_store.put(ws, plan.file, ttl_seconds=900,
                             bind_to=current_session())
    return {
        "plan_id": plan_id,
        "summary": summary,
        "risk": classify_plan_risk(summary),      # none/low/high
        "expires_in": 900,
    }

async function terraformApply(planId: string, ctx: CallContext) {
  const entry = planStore.get(planId);
  if (!entry) throw new McpError(-32602, "unknown or expired plan_id");
  if (entry.bind_to !== ctx.sessionId) {
    throw new McpError(-32003, "plan belongs to another session");   // 防跨会话盗用
  }
  const risk = classifyPlanRisk(entry.summary);
  if (risk === "high" && !(await approvals.isGranted(ctx, planId))) {
    throw new McpError(-32003, "high-risk plan requires approval");
  }
  audit.record({ op: "tf_apply", plan_id: planId, summary: entry.summary });
  return tf.apply(entry.workspace, entry.file);
}

ALLOWED_NAMESPACES = {"dev", "staging", "ai-sandbox"}
DENIED_NAMESPACES = {"kube-system", "prod", "monitoring"}

def assert_namespace(ns: str) -> None:
    if ns in DENIED_NAMESPACES:
        raise McpError(-32003, f"namespace {ns} is protected")
    if ns not in ALLOWED_NAMESPACES:
        raise McpError(-32003, f"namespace {ns} is out of scope")

def assert_scope(manifest: dict) -> None:
    ns = manifest.get("metadata", {}).get("namespace", "default")
    assert_namespace(ns)
    # 禁止跨集群资源（ClusterRole 等）
    if manifest.get("kind") in {"ClusterRole", "ClusterRoleBinding", "CustomResourceDefinition"}:
        raise McpError(-32003, "cluster-scoped resources are not allowed")

# K8s 原生 dry-run: 校验但不落地
kubectl apply -f manifest.yaml --dry-run=server -o yaml
# 先 dry-run 校验，再真实 apply —— 让模型"先看会发生什么"
kubectl apply -f manifest.yaml --dry-run=server \
  && kubectl apply -f manifest.yaml

# 1) 不暴露通用"执行任意 API 调用"的工具（等于给万能钥匙）
# 2) 按场景封装: "查 EC2 实例"、"看 S3 桶大小"、"列 RDS"
# 3) 每个工具只映射到少数几个只读或低危 API
# 4) 参数做白名单: 只允许安全的过滤条件
# 反面教材: 一个 aws_call(service, action, params) 工具 = 无限制

# 只封装只读的 Describe/List 类 API
READONLY_ACTIONS = {
    "ec2:DescribeInstances", "ec2:DescribeSecurityGroups",
    "s3:ListBuckets", "s3:GetBucketLocation",
    "rds:DescribeDBInstances", "cloudwatch:GetMetricData",
    "ce:GetCostAndUsage",       # 成本查询
}

async def cloud_query(service: str, action: str, params: dict) -> dict:
    full = f"{service}:{action}"
    if full not in READONLY_ACTIONS:
        raise McpError(-32003, f"action {full} is not allowed")
    return await aws_client.call(service, action, sanitize(params))

# 1) 平台原生 dry-run: K8s --dry-run=server, AWS DryRun=true
# 2) 计划式: Terraform plan（天然就是 dry-run）
# 3) 模拟式: 自建 diff 引擎，对比"变更前后"状态
# 4) 只读预检: 检查前置条件（配额、依赖、冲突）
# 不是所有操作都支持 dry-run，不支持的要有替代（如 diff 预演）

async function changeWithApproval(req: ChangeRequest, ctx: CallContext) {
  // 1) 预检 + dry-run
  const preview = await previewChange(req);
  if (preview.errors.length) {
    throw new McpError(-32602, `preflight failed: ${preview.errors.join("; ")}`);
  }
  // 2) 风险评估
  const risk = assessRisk(req, preview);
  if (risk.level === "critical") {
    const ticket = await approvals.request({
      principal: ctx.principal,
      change: req,
      preview: preview.summary,
      blast_radius: preview.blastRadius,
    });
    return { status: "pending_approval", ticket_id: ticket.id, preview: preview.summary };
  }
  // 3) 执行 + 审计
  audit.record({ op: "change", req, preview: preview.summary, risk: risk.level });
  return executeChange(req);
}

def assess_blast_radius(change: dict) -> dict:
    r = {"level": "low", "affected": []}
    if change["kind"] == "delete_namespace":
        r = {"level": "critical", "affected": ["all pods/services in ns"]}
    elif change["kind"] == "update_security_group":
        r = {"level": "high", "affected": ["network reachability"]}
    elif change["kind"] == "scale":
        r = {"level": "medium",
             "affected": [f"replicas {change['from']}→{change['to']}"]}
    r["prod"] = change.get("env") == "prod"
    return r

# 1) 一工具一角色: 不同工具用不同凭证（读工具只读、写工具限定资源）
# 2) 默认拒绝: 策略从"什么都不允许"开始加白名单
# 3) 资源限定: 不用 "*"，写到具体 ARN 或前缀
# 4) 短期凭证: 用 STS 临时凭证，不用长期 AccessKey
# 5) 会话标签: 标记调用来源，便于审计与追责
# 6) 定期轮换: 凭证有效期 + 自动轮换

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadOnlyForDiagnostics",
      "Effect": "Allow",
      "Action": ["ec2:Describe*", "cloudwatch:GetMetricData", "logs:FilterLogEvents"],
      "Resource": "*"
    },
    {
      "Sid": "ScaleOnlySandboxEKS",
      "Effect": "Allow",
      "Action": ["eks:UpdateNodegroupConfig"],
      "Resource": "arn:aws:eks:ap-northeast-1:123456789012:nodegroup/sandbox/*"
    }
  ]
}

# 凭证的"可见性"边界
# 1) 凭证只在 MCP 服务器进程内使用，绝不作为工具返回值
# 2) 工具输出里若出现 token/密钥 → 脱敏
# 3) 日志里不打印凭证
# 4) 模型无法通过任何工具"读到"凭证（没有这种工具）
# 模型不需要凭证，只需要能力——这是设计的分界线

async function scaleDeployment(req: ScaleRequest, ctx: CallContext) {
  const current = await k8s.getDeployment(req.ns, req.name);
  const delta = req.replicas - current.spec.replicas;
  const unitCost = await pricing.nodeHourlyCost(current.nodeSelector);
  const hourlyDelta = delta * unitCost;
  const monthlyDelta = hourlyDelta * 730;

  // 超预算则拒绝，并给出具体数字
  if (monthlyDelta > ctx.budget.maxMonthlyDelta) {
    throw new McpError(
      -32003,
      `scale would add $${monthlyDelta.toFixed(2)}/mo, exceeding budget ` +
      `$${ctx.budget.maxMonthlyDelta}`
    );
  }
  return { status: "ok", preview: { from: current.spec.replicas, to: req.replicas, monthlyDelta } };
}

# 每次变更前问四个问题
# 1) 影响范围: 一个实例、一个服务、一个环境、还是整个账号?
# 2) 可逆性: 能回滚吗? 回滚需要多久?
# 3) 依赖面: 谁会依赖被改的东西?（上游/下游）
# 4) 时间点: 现在是业务高峰吗? 有变更窗口吗?
# 答不上来 → 不允许自动执行

# 传统: 模型 → 云 API（直改，审计分散）
# GitOps: 模型 → Git PR（改声明）→ CI 校验 → 人审 → 合入 → 控制器落地
# 优势:
#   1) 所有变更经过 Git 评审与审计
#   2) 声明与状态一致，可回滚（revert 提交）
#   3) 模型能力限制在"改声明"，不碰真实资源
# 模型是"变更作者"，Git 是"变更通道"，GitOps 控制器是"执行者"

# 按操作类型选择通道
operations:
  declarative:            # 声明式变更走 GitOps
    examples: [deployment, configmap, ingress]
    channel: gitops_pr
  diagnostic:             # 诊断走只读工具
    examples: [get_pods, get_metrics, get_logs]
    channel: readonly_api
  imperative:             # 命令式变更走审批 API
    examples: [rollout_restart, scale]
    channel: approved_api
    require: [dry_run, approval]