服务网格与零信任安全：Istio、OPA、SPIFFE 与合规治理

# 关键认知：Envoy 的 root cert 是一个"证书集合"，不是单个。
# istiod 下发的 ROOTCA 里可以同时包含：
#   - 旧根 CA  (仍信任)
#   - 新根 CA  (新签发的证书由它签发)
# 轮换分三步：双根共存 → 新证书铺满 → 撤旧根

# ① 在 Vault 启用 pkii 引擎，并配置允许 istiod 申请中间 CA
vault secrets enable -path=pki-istio pki
vault write pki-istio/root/generate/internal \
  common_name="istio-root-ca" \
  ttl=87600h                # 根 CA 有效期 10 年

# ② 生成一张"中间 CA 签名请求(CSR)"交给 istiod 一侧
vault write pki-istio/intermediate/generate/exported \
  common_name="istio-intermediate-ca" \
  | tee istio-int.json      # 里面含 csr 和私钥占位

# ③ Istio 的 MeshConfig 指向外部 CA（Vault 签好的中间 CA + 根）
# 通过 istioctl 安装时提供 cert 链与私钥
apiVersion: install.istio.io/v1alpha1
kind: IstioOperator
spec:
  components:
    pilot:
      k8s:
        env:
          - name: EXTERNAL_CA
            value: "ISTIOD_RA_KUBERNETES_API"   # 或自定义 RA 对接 Vault
  values:
    global:
      caAddress: "istiod.istio-system.svc:15012"

# 在 Vault 中开启 CRL 自动发布
vault write pki-istio/config/urls \
  issuing_certificates="http://vault:8200/v1/pki-istio/ca" \
  crl_distribution_points="http://vault:8200/v1/pki-istio/crl" \
  ocsp_servers="http://vault:8200/v1/pki-istio/ocsp"

# 分发策略要点（面试可答）：
# 1) 控制面下发：把 CRL 定期同步到 ConfigMap，再由 istiod 注入 Envoy 的信任上下文。
# 2) 频率与体积：CRL 可能很大，5k+ 工作负载时改为 OCSP 或"短生命周期证书(几分钟)"
#    让吊销天然失效——这是 Istio 的默认哲学：证书有效期短(默认1h)，不依赖 CRL。
# 3) 真正吊销入口：在 Vault 执行 `vault write pki-istio/revoke serial_number=...`

# ① 要求对端做 mTLS，并校验其 SPIFFE ID 前缀
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: ingress-mtls
  namespace: istio-system
spec:
  mtls:
    mode: STRICT
---
# ② 校验 JWT：iss 必须是我们的 OIDC，aud 必须匹配网关
apiVersion: security.istio.io/v1beta1
kind: RequestAuthentication
metadata:
  name: jwt-check
  namespace: istio-system
spec:
  selector:
    matchLabels:
      istio: ingressgateway
  jwtRules:
    - issuer: "https://auth.example.com"
      jwksUri: "https://auth.example.com/.well-known/jwks.json"
---
# ③ 授权：既要有合法 JWT，又要来自可信 SPIFFE ID（双重）
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: double-authz
  namespace: istio-system
spec:
  selector:
    matchLabels:
      istio: ingressgateway
  action: ALLOW
  rules:
    - from:
        - source:
            principals: ["spiffe://example.com/ns/partner/sa/client"]  # SPIFFE ID
      when:
        - key: request.auth.claims[scope]
          values: ["api:read"]                                       # JWT claim

# ① 定义两个子集，分别对应轻/重接口，熔断阈值不同
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: orders-dr
spec:
  host: orders.svc.cluster.local
  subsets:
    - name: heavy            # 重接口：阈值更严
      labels: { tier: orders }
      trafficPolicy:
        outlierDetection:
          consecutive5xxErrors: 3
          interval: 30s
          baseEjectionTime: 60s
          maxEjectionPercent: 50
    - name: light           # 轻接口：容忍度高
      labels: { tier: orders }
      trafficPolicy:
        outlierDetection:
          consecutive5xxErrors: 20
          interval: 30s
          baseEjectionTime: 30s
---
# ② VirtualService 按 path 把流量分到不同子集，实现"按 path 熔断"
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: orders-vs
spec:
  hosts: ["orders.svc.cluster.local"]
  http:
    - match: [{ uri: { prefix: "/v1/search" } }]
      route: [{ destination: { host: orders, subset: heavy } }]
    - match: [{ uri: { prefix: "/v1/health" } }]
      route: [{ destination: { host: orders, subset: light } }]

# 通过 EnvoyFilter 改写过载响应，注入友好重试头
apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: custom-overload-reply
  namespace: istio-system
spec:
  workloadSelector:
    labels: { app: orders }
  configPatches:
    - applyTo: NETWORK_FILTER
      match: { listener: { filterChain: { filter: { name: "envoy.filters.network.http_connection_manager" } } } }
      patch:
        operation: MERGE
        value:
          typed_config:
            "@type": "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"
            local_reply_config:
              mappers:
                - filter: { status_code_filter: { comparison: { op: EQ, value: { default_value: 503 } } } }
                  headers_to_add:
                    - header: { key: "Retry-After", value: "2" }
                    - header: { key: "X-Circuit-State", value: "open" }
                  body: { inline_string: "{\"error\":\"circuit_open\",\"retry_after\":2}" }

-- redis_global_token_bucket.lua
-- KEYS[1] = 桶的 key（如 rl:orders:search）
-- ARGV[1] = 容量   ARGV[2] = 每秒 refill 速率   ARGV[3] = 本次请求令牌数   ARGV[4] = 当前时间戳(ms)
local key   = KEYS[1]
local cap   = tonumber(ARGV[1])
local rate  = tonumber(ARGV[2])
local cost  = tonumber(ARGV[3])
local now   = tonumber(ARGV[4])

local data = redis.call("HMGET", key, "tokens", "ts")
local tokens = tonumber(data[1]) or cap
local ts     = tonumber(data[2]) or now

-- 按时间比例补充令牌（毫秒）
local delta = (now - ts) / 1000 * rate
tokens = math.min(cap, tokens + delta)

local allowed = 0
if tokens >= cost then
  tokens = tokens - cost
  allowed = 1
end

redis.call("HMSET", key, "tokens", tokens, "ts", now)
-- 设置过期，避免空桶无限堆积
redis.call("PEXPIRE", key, math.ceil(cap / rate * 1000) + 1000)
return { allowed, math.floor(tokens) }

# 在 Envoy 侧通过 ratelimit 服务（或直接在 sidecar 里用 lua filter）调用上面的脚本：
# 这里给出"本地验证脚本逻辑"的等价测试（用 redis-cli 模拟一次扣减）
redis-cli --eval redis_global_token_bucket.lua rl:orders:search \
  , 100 10 1 "$(date +%s)000"
# 返回值 [1, 99] 表示：放行(1)，剩余 99 令牌

# ① 定义模板：deny 时返回详细 message
apiVersion: templates.gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: k8spodlabelregex
spec:
  crd:
    spec:
      names: { kind: K8sPodLabelRegex }
      validation:
        openAPIV3Schema:
          type: object
          properties:
            label:    { type: string }
            pattern:  { type: string }
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8spodlabelregex
        violation[{"msg": msg}] {
          label := input.parameters.label
          pat   := input.parameters.pattern
          not input.review.object.metadata.labels[label]
          msg := sprintf("缺少必需标签 %v", [label])
        }
        violation[{"msg": msg}] {
          label := input.parameters.label
          pat   := input.parameters.pattern
          v := input.review.object.metadata.labels[label]
          not regex.match(pat, v)
          msg := sprintf("标签 %v 的值 %v 不匹配正则 %v", [label, v, pat])
        }        
---
# ② 实例化约束
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sPodLabelRegex
metadata:
  name: require-team-label
spec:
  match:
    kinds: [{ apiGroups: [""], kinds: ["Pod"] }]
  parameters:
    label: "team"
    pattern: "^(pay|search|risk)-[a-z]+$"

# ① 一键把"生产命名空间"从约束中排除，先止血
kubectl patch k8spodlabelregex require-team-label --type=merge \
  -p '{"spec":{"match":{"excludedNamespaces":["prod"]}}}'

# ② 或直接把动作降级为 dryrun
kubectl patch k8spodlabelregex require-team-label --type=merge \
  -p '{"spec":{"enforcementAction":"dryrun"}}'

# ③ 回滚：用 git 里的历史版本重放
kubectl apply -f policies/archive/require-team-label-v2.yaml

# ① 全局默认低采样（10%），省钱
apiVersion: telemetry.istio.io/v1alpha1
kind: Telemetry
metadata:
  name: mesh-tracing
  namespace: istio-system
spec:
  tracing:
    - providers: [{ name: otel }]
      randomSamplingPercentage: 10
      customTags:
        tenant:
          requestHeader:
            name: x-tenant-id        # 把租户 ID 带进 span
---
# ② 对"风控租户"单独高采样（100%），因为它的请求最该被看
apiVersion: telemetry.istio.io/v1alpha1
kind: Telemetry
metadata:
  name: risk-tenant-tracing
  namespace: risk
spec:
  workloadSelector:
    labels: { app: risk-engine }
  tracing:
    - providers: [{ name: otel }]
      randomSamplingPercentage: 100

# otel-collector-config.yaml（节选）
processors:
  tail_sampling:
    decision_wait: 10s            # 等 10s 收集完 span 再决策
    num_traces: 50000
    policies:
      - name: errors-policy       # 策略1：有错误的全采
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: slow-policy         # 策略2：超过 500ms 的采
        type: latency
        latency: { threshold_ms: 500 }
      - name: tenant-risk         # 策略3：风控租户全采
        type: string_attribute
        string_attribute: { key: tenant, values: ["risk"], enabled_regex: false }
      - name: default-policy      # 兜底：其余 5%
        type: probabilistic
        probabilistic: { sampling_percentage: 5 }

# ① 看 Collector 导出耗时（Agent 侧）
kubectl exec -n observability deploy/otel-collector -- \
  curl -s localhost:8889/metrics | grep exporter_send_duration

# ② 看 Kafka 消费积压（后端侧）
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group trace-ingest
# 若 LAG 持续上涨 → 后端消费跟不上，扩 consumer；
# 若 LAG 为 0 但端到端延迟高 → 是 Agent 批处理/网络问题，调小 batch_timeout。

# squid.conf 关键配置：peek（看 SNI）然后 splice（放行不解密）
acl step1 at_step SslBump1
acl step2 at_step SslBump2
ssl_bump peek step1          # 看 ClientHello 里的 SNI
ssl_bump peek step2          # 看 ServerHello
ssl_bump splice all          # 不解密，直接透传
# 把看到的 SNI 写进访问日志，实现审计
access_log /var/log/squid/access.log squid

# Istio 让所有出口流量走 Egress Gateway，再转给 Squid
apiVersion: networking.istio.io/v1beta1
kind: ServiceEntry
metadata:
  name: external-api
spec:
  hosts: ["api.vendor.com"]
  ports: [{ number: 443, name: https, protocol: HTTPS }]
  location: MESH_EXTERNAL
  resolution: DNS
---
apiVersion: networking.istio.io/v1beta1
kind: Gateway
metadata:
  name: egress-gw
spec:
  servers:
    - port: { number: 443, name: https, protocol: HTTPS }
      hosts: ["*"]
  selector: { istio: egressgateway }

# 用 Tetragon 观察某 Pod 的对外连接（节选自带 tracer）
kubectl exec -n kube-system ds/tetragon -- tetra getevents \
  -o json | jq 'select(.process_exec.process) | .process_exec.process.exec_id'
# 观察 connect 事件，记录 dst 端口/地址，用于生成网络策略与 Seccomp。

# 用 Cilium EgressGatewayPolicy 把出口流量绑定到一组 IP
apiVersion: cilium.io/v2
kind: CiliumEgressGatewayPolicy
metadata:
  name: vendor-egress
spec:
  destinationCIDRs:
    - "203.0.113.0/24"        # 第三方网段
  selectors:
    - podSelector:
        matchLabels: { app: orders }
  egressGateway:
    nodeSelector:
      matchLabels: { egress-pool: primary }   # 当前出口池

# 切换到备用池（primary -> backup），并通知 SRE
kubectl patch ciliumegressgatewaypolicy vendor-egress --type=merge \
  -p '{"spec":{"egressGateway":{"nodeSelector":{"matchLabels":{"egress-pool":"backup"}}}}}'

# 通知：用 kubectl 触发一个告警事件（实际可接 Alertmanager/飞书）
kubectl annotate --overwrite pod -n sre alert=1 \
  "message=Egress IP 被封禁，已切换至 backup 池 @sre-oncall"

# require_team_label.rego
package k8spodlabelregex
violation[{"msg": msg}] {
  not input.review.object.metadata.labels.team
  msg := "missing team label"
}

# require_team_label_test.rego
package k8spodlabelregex
test_missing_label {
  input := {"review": {"object": {"metadata": {}}}}
  results := violation with input as input
  count(results) == 1
}
test_has_label {
  input := {"review": {"object": {"metadata": {"labels": {"team": "pay-core"}}}}}
  results := violation with input as input
  count(results) == 0
}

# 跑测试
opa test . -v

# 把模板、约束、待测资源放一起跑
gator test \
  --templates policies/templates/ \
  --constraints policies/constraints/ \
  --objects   samples/bad-pod.yaml
# 输出：哪个约束、哪条 violation、msg 详情——CI 里据此 fail。

# ① 把策略目录打成 bundle
opa build -b policies/ -o bundle.tar.gz

# ② 起一个 bundle server（或用 OCI 仓库）
opa run -s -b bundle.tar.gz   # 暴露 /v1/bundles/<name>

# ③ Gatekeeper 侧引用远程 bundle（而非内联 CRD）
kubectl apply -f - <<'EOF'
apiVersion: config.gatekeeper.sh/v1alpha1
kind: Config
metadata: { name: config }
spec:
  sync: { syncOnly: [{ group: "", version: "v1", kind: "Pod" }] }
  validation:
    providers:
      - name: <your-bundle-server>
        url: http://bundle-server:8181/v1/bundles/policies
EOF

# 跑 CIS 1.8 检查（以 master 节点为例）
kube-bench run --targets master --version 1.8 \
  --benchmark cis-1.8
# 输出 [FAIL] 项 + 修复建议；可加 --json 让 CI 解析并生成修复 PR。

# 示例：把"kubelet 匿名认证"自动修复脚本化
# 修复前先备份，再改 kubelet 配置并重启（生产用滚动而非硬重启）
cp /etc/kubernetes/kubelet.conf /backup/kubelet.conf.$(date +%s)
sed -i 's/--anonymous-auth=true/--anonymous-auth=false/' \
  /etc/kubernetes/kubelet.conf
systemctl restart kubelet

# 用 CronJob 定时跑 kube-bench 并上报（节选）
apiVersion: batch/v1
kind: CronJob
metadata: { name: kube-bench-weekly }
spec:
  schedule: "0 3 * * 0"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
            - name: kube-bench
              image: aquasec/kube-bench:latest
              args: ["run","--targets","master,node,etcd,policies","--json"]
          restartPolicy: Never

# 一键修复：关闭匿名认证 + 记录审计注解
kubectl get node -o name | while read n; do
  ssh "$n" "sed -i 's/anonymous-auth=true/anonymous-auth=false/' \
    /etc/kubernetes/kubelet.conf && systemctl restart kubelet"
  # 审计：在 CMDB/工单里记录 节点+操作人+时间
  echo "$(date -Iseconds) fixed anonymous-auth on $n by sre-remediation" \
    >> /var/log/security-audit.log
done

# ① 启用 transform 引擎并定义 FPE 转换
vault secrets enable transform
vault write transform/transformation/ccn-fpe \
  type=fpe \
  template=ccn \
  tweak_source=internal \
  allowed_roles=ccn-role

# ② 定义模板：只保留最后 4 位可见，其余格式保持加密
vault write transform/template/ccn \
  type=regex \
  pattern='(\d{4}-\d{4}-\d{4}-)(\d{4})' \
  alphabet=builtin/numeric

# ③ 加密 / 解密示例
vault write transform/encode/ccn-role \
  transformation=ccn-fpe value="4111-1111-1111-1111"
# 输出仍是 16 位数字，格式合法但值已变

apiVersion: v1
kind: ConfigMap
metadata:
  name: benthos-redact
data:
  config.yaml: |
    input:
      kafka:                       # 从 Kafka 读原始日志
        addresses: [ "kafka:9092" ]
        topics: [ "app-logs" ]
    pipeline:
      processors:
        - awk: |                   # 正则把卡号打码
            {
              gsub(/[0-9]{4}-[0-9]{4}-[0-9]{4}-[0-9]{4}/, "****-****-****-****")
              gsub(/1[3-9][0-9]{9}/, "***")   # 手机号
              print
            }
    output:
      kafka:                       # 脱敏后写回另一主题
        addresses: [ "kafka:9092" ]
        topic: "app-logs-redacted"    

# 以 Sidecar 形式注入到业务 Pod，实时拦截日志流
kubectl apply -f benthos-redact-cm.yaml

# 伪流程：拉取 S3 历史日志 → 用新 Benthos 配置重处理 → 覆盖写回
aws s3 sync s3://logs-raw/2024-01/ /tmp/raw/
benthos -c new-redact.yaml <<< "$(cat /tmp/raw/*.log)" > /tmp/redacted/
aws s3 sync /tmp/redacted/ s3://logs-raw/2024-01/ --delete
# 注意：写回前确认新规则不会破坏不可篡改(WORM)存储的合规要求。

# CI 步骤：扫描并阻断（CVE 严重/高危且不在白名单则非零退出）
trivy image --exit-code 1 --severity CRITICAL,HIGH \
  --ignorefile .trivyignore \     # 白名单：已知可接受 CVE
  registry.example.com/orders:1.2.3

# .trivyignore 示例（明确记录"为何豁免"）
# CVE-2023-1234 仅影响我们不使用的 CLI 子命令，本期接受
CVE-2023-1234

# Harbor 项目策略（图形化等价的 API 表达）
# 在 Harbor UI: 项目 -> 配置 -> 漏洞扫描 -> 阻止高风险漏洞 = 开启
# 严重/高危 = 阻断；中/低 = 仅告警

# ① 用 Cosign 对镜像签名（密钥来自 KMS/硬件）
cosign sign --key kms://hashicorp-vault/transit/cosign \
  registry.example.com/orders:1.2.3

# ② 生成构建来源证明（SLSA/In-Toto 格式）
cosign attest --key kms://hashicorp-vault/transit/cosign \
  --type slsaprovenance \
  --predicate ./provenance.json \
  registry.example.com/orders:1.2.3

# ③ 验证：签名 + 证明都过才允许准入（接 Kyverno/Gatekeeper）
cosign verify --key kms://... registry.example.com/orders:1.2.3
cosign verify-attestation --key kms://... registry.example.com/orders:1.2.3

# Kueue 队列 + 优先级（节选）
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata: { name: scan-queue, namespace: ci }
spec:
  clusterQueue: scan-cq
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: Workload
metadata: { name: scan-prod-123, namespace: ci }
spec:
  priority: 100            # 生产高优先级
  queueName: scan-queue
  podSets:
    - name: scanner
      template:
        spec:
          containers:
            - name: trivy
              image: aquasec/trivy:latest
              args: ["image","registry.example.com/orders:1.2.3"]

# apiserver 审计策略（节选）
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  - level: Metadata          # 只记 who/what/when，不记 requestObject/responseObject
    verbs: ["create","update","delete","patch"]
    omitStages: ["RequestReceived"]
  - level: None              # 只读 get/list/watch 不审计，降量
    verbs: ["get","list","watch"]

# 给 kube-apiserver 挂上该策略
# --audit-policy-file=/etc/kubernetes/audit-policy.yaml
# --audit-log-path=/var/log/k8s-audit.log

{
  "policy": {
    "phases": {
      "hot":   { "actions": { "rollover": { "max_age": "7d", "max_size": "50gb" } } },
      "warm":  { "actions": { "readonly": {} } },
      "cold":  { "actions": { "freeze": {} } },
      "delete": { "min_age": "365d", "actions": { "delete": {} } }
    }
  }
}

# 在对象存储侧开启 WORM（以阿里云 OSS 合规保留策略为例）
ossutil legal-hold --mode COMPLIANCE oss://audit-logs/
# 开启后该桶对象在保留期内不可删除/覆盖，满足等保留存要求。

# ① 对目标 Pod 做 checkpoint（需启用 K8s 1.25+ 的 checkpoint API 或 CRIU 直连）
kubectl alpha checkpoint pod orders-xyz -n prod
# 生成 checkpoint 归档（含内存页、寄存器、打开的文件描述符）

# ② 在隔离环境还原，离线取证（不连生产网络）
criu restore --images-dir /checkpoints/orders-xyz/ \
  --restore-detached
# 还原后可 gdb 附加、dump 堆、查未落盘的敏感状态。