云原生时代，身份就是新的防火墙

预计阅读时间：15 分钟

当工作负载的平均寿命从"几个月"缩短到"几分钟"，当服务间调用数量从几十涨到几千，基于网络边界的安全模型就彻底失效了。你没法给一个存活 30 秒的 Job 逐一配置防火墙规则，也没法靠 IP 白名单管理几百个互相调动的微服务。云原生架构下，身份（Identity）取代网络边界，成为最核心的安全防线——这是近年来各大安全白皮书反复论证的结论，也是每个正在做微服务和 K8s 迁移的团队必须正视的现实。

传统模型为什么扛不住

经典的企业安全思路是：内网可信，外网不可信，用 VPN、防火墙、VPC 把"可信区域"圈起来。这套逻辑在云原生面前有三个硬伤：

工作负载短命。 一个 Kubernetes Job 或 Serverless 函数可能只跑几秒到几分钟。传统方式靠人工审批、静态配置角色和权限，根本来不及。等你把 ServiceAccount 的权限批下来，Pod 已经结束了。

服务间通信爆炸。 微服务架构下，调用链路不再是"前端→后端→数据库"三层，而是几十个服务互相调用。每个调用都需要认证和授权，IP 地址随时可能因为滚动更新而变化，基于 IP 的访问控制变得不可靠。

零信任要求逐请求验证。 零信任的核心不是"不信任任何人"，而是"每次请求都要验证身份和权限，不管请求来自内网还是外网"。这意味着身份系统必须支持高频、自动化的令牌签发与校验。

云原生 IAM 的核心模型

云原生场景下的 IAM，需要满足几个关键能力：

能力	说明
工作负载身份	每个服务、Pod、Job 拥有独立身份，不依赖 IP
短期令牌	Token 有效期几分钟到几小时，自动轮换，不靠长期密钥
细粒度授权	基于属性（ABAC）或关系（ReBAC）做权限判定，而非粗粒度角色
可观测性	每次访问决策有日志，可审计、可回溯
策略即代码	授权策略用代码定义，随应用一起版本管理和部署

这五个能力叠加起来，才能支撑零信任在云原生环境落地。

工作负载身份：从 SPIFFE 到 Kubernetes ServiceAccount

工作负载身份是云原生 IAM 的基石。目前业界有两个主流实现路径：

SPIFFE/SPIRE 提了一套标准化的工作负载身份框架。每个工作负载获得一个 SPIFFE ID（格式如 spiffe://example.org/ns/default/sa/my-service），通过 SPIRE Agent 自动获取短期 X.509 SVID 证书，服务间用 mTLS 互相验证身份。

Kubernetes原生方案 则利用 ServiceAccount + OIDC Token。K8s 1.30+ 默认启用 TokenRequest API，Pod 可以自动获取绑定受众（audience）的短期 JWT，有效期可精确到分钟级。

下面用一个实际可运行的例子，演示 K8s 原生工作负载身份的配置和验证。

实践：为微服务配置细粒度工作负载身份

假设你有一个订单服务 order-service，需要调用库存服务 inventory-service 的 API。我们用 ServiceAccount + TokenRequest + OPA 策略来实现身份认证和授权。

第一步：创建专用 ServiceAccount 并绑定最小权限

# order-service-sa.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: order-service
  namespace: production
automountServiceAccountToken: false  # 不自动挂载默认 token，改用 TokenRequest
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: order-service-inventory-reader
  namespace: production
rules:
- apiGroups: [""]
  resources: ["services"]
  resourceNames: ["inventory-service"]
  verbs: ["get"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: order-service-inventory-reader-binding
  namespace: production
subjects:
- kind: ServiceAccount
  name: order-service
  namespace: production
roleRef:
  kind: Role
  name: order-service-inventory-reader
  apiGroup: rbac.authorization.k8s.io

# 应用配置
kubectl apply -f order-service-sa.yaml

第二步：在 Pod 中通过 TokenRequest 获取短期 OIDC Token

# order-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: order-service-pod
  namespace: production
spec:
  serviceAccountName: order-service
  containers:
  - name: order-service
    image: order-service:latest
    env:
    - name: INVENTORY_SERVICE_URL
      value: "https://inventory-service.production.svc.cluster.local:8443"
    # 通过 projected volume 注入短期 token
    volumeMounts:
    - name: oidc-token
      mountPath: /var/run/secrets/oidc
      readOnly: true
  volumes:
  - name: oidc-token
    projected:
      sources:
      - serviceAccountToken:
          path: token
          expirationSeconds: 600    # 10 分钟有效期
          audience: "inventory-service"  # 绑定受众，防止 token 被其他服务滥用

kubectl apply -f order-pod.yaml

# 验证 token 内容（调试用，生产环境不要在日志中打印 token）
kubectl exec order-service-pod -n production -- \
  cat /var/run/secrets/oidc/token | cut -d. -f2 | base64 -d 2>/dev/null | python3 -m json.tool

解码后的 JWT payload 大致如下：

{
  "iss": "https://kubernetes.default.svc.cluster.local",
  "sub": "system:serviceaccount:production:order-service",
  "aud": "inventory-service",
  "exp": 1720000000,
  "kubernetes.io/serviceaccount/namespace": "production",
  "kubernetes.io/serviceaccount/service-account.name": "order-service"
}

关键点：aud 字段绑定了 inventory-service，只有库存服务的验证逻辑才会接受这个 token；exp 只有 10 分钟窗口，过期后 Pod 需重新请求。

第三步：在库存服务端验证请求身份

库存服务用 Python 实现一个简单的身份校验中间件：

# identity_middleware.py
import jwt
import requests
import time
from functools import wraps
from flask import request, jsonify

# K8s OIDC 发现端点（集群内部可访问）
OIDC_ISSUER = "https://kubernetes.default.svc.cluster.local"
EXPECTED_AUDIENCE = "inventory-service"

# 缓存 OIDC 公钥（JWKS），避免每次请求都拉取
_jwks_cache = {"keys": None, "expires_at": 0}

def get_jwks():
    now = time.time()
    if _jwks_cache["keys"] and now < _jks_cache["expires_at"]:
        return _jwks_cache["keys"]
    # 从集群内获取 JWKS
    resp = requests.get(
        f"{OIDC_ISSUER}/openid/v1/jwks",
        headers={"Authorization": f"Bearer {get_k8s_token()}"}
    )
    _jwks_cache["keys"] = resp.json()["keys"]
    _jwks_cache["expires_at"] = now + 3600
    return _jwks_cache["keys"]

def verify_workload_identity(token: str) -> dict:
    """验证 K8s OIDC Token，提取工作负载身份信息"""
    jwks = get_jwks()
    decoded = jwt.decode(
        token,
        key=jwks,
        algorithms=["RS256"],
        audience=EXPECTED_AUDIENCE,
        issuer=OIDC_ISSUER,
    )
    # 提取身份属性用于后续授权判定
    return {
        "service_account": decoded["sub"],
        "namespace": decoded["kubernetes.io/serviceaccount/namespace"],
        "service_name": decoded["kubernetes.io/serviceaccount/service-account.name"],
    }

def require_identity(allowed_services: list):
    """Flask 中间件：验证调用方身份是否在允许列表中"""
    def decorator(f):
        @wraps(f)
        def wrapped(*args, **kwargs):
            auth_header = request.headers.get("Authorization", "")
            if not auth_header.startswith("Bearer "):
                return jsonify({"error": "missing token"}), 401
            token = auth_header[7:]
            try:
                identity = verify_workload_identity(token)
            except jwt.ExpiredSignatureError:
                return jsonify({"error": "token expired"}), 401
            except jwt.InvalidAudienceError:
                return jsonify({"error": "invalid audience"}), 403
            except Exception as e:
                return jsonify({"error": f"invalid token: {e}"}), 401

            # 简单身份匹配（生产环境建议用 OPA/ABAC）
            sa_name = identity["service_name"]
            if sa_name not in allowed_services:
                return jsonify({
                    "error": "access denied",
                    "identity": identity,
                }), 403

            # 将身份信息注入请求上下文，供业务逻辑使用
            request.identity = identity
            return f(*args, **kwargs)
        return wrapped
    return decorator

# 使用示例
@app.route("/api/inventory/<item_id>")
@require_identity(allowed_services=["order-service", "report-service"])
def get_inventory(item_id):
    # request.identity 已注入，可用于审计日志
    log_access(request.identity, item_id)
    return jsonify({"item_id": item_id, "stock": get_stock(item_id)})

运行前需要安装依赖：

pip install flask pyjwt requests

这段代码做了三件事：验证 JWT 签名和有效期、校验受众（audience）防止 token 被挪用、匹配调用方 ServiceAccount 名称做粗粒度授权。生产环境中，最后一步应该替换为 OPA 策略引擎，实现更灵活的 ABAC 判断。

策略即代码：用 OPA 替代硬编码权限

上面中间件里的 allowed_services 列表是硬编码的，服务多了以后维护成本很高。更成熟的做法是把授权策略从代码中抽出来，交给 OPA（Open Policy Agent）统一管理。

# inventory-policy.rego - OPA 策略文件
package inventory.access

default allow = false

# 允许 order-service 读取库存
allow {
    input.identity.service_name == "order-service"
    input.action == "read"
    input.resource == "inventory"
}

# 允许 report-service 读取库存（仅工作时间段）
allow {
    input.identity.service_name == "report-service"
    input.action == "read"
    input.resource == "inventory"
    working_hours[input.identity.namespace]
}

working_hours["production"] {
    # 生产环境：UTC 1:00-23:00（对应北京 9:00-次日7:00）
    current_hour >= 1
    current_hour < 23
}

current_hour = h {
    h := time.clock(time.now_ns())[0]
}

调用方式变成：

import requests

OPA_URL = "http://opa.production.svc.cluster.local:8181/v1/data/inventory/access"

def check_opa_policy(identity: dict, action: str, resource: str) -> bool:
    payload = {
        "input": {
            "identity": identity,
            "action": action,
            "resource": resource,
        }
    }
    resp = requests.post(OPA_URL, json=payload)
    return resp.json().get("result", {}).get("allow", False)

策略文件随应用代码一起进 Git，一起走 CI/CD，变更可审计、可回滚——这就是"策略即代码"的核心价值。

短期令牌与密钥轮换

长期密钥（API Key、静态 Secret）是云原生 IAM 的最大隐患。一个泄露的长期密钥可以在几个月内被滥用，而你很难察觉。短期令牌的思路是：即使泄露，窗口也只有几分钟。

实际操作中的几个要点：

Token 有效期尽量短。 K8s TokenRequest 支持 expirationSeconds 最低 600 秒（10 分钟），对于大多数服务间调用足够。
受众绑定必须严格。 每个 token 的 audience 只指向目标服务，防止中间人截获后转用。
密钥轮换自动化。 SPIRE Agent 每 1 小时轮换 X.509 SVID；K8s OIDC Token 由 apiserver 签发，无需手动管理。
禁止自动挂载默认 token。 设置 automountServiceAccountToken: false，避免 Pod 拿到不绑定受众的长期默认 token。

# 检查集群中哪些 Pod 还在用默认长期 token
kubectl get pods -A -o json | python3 -c "
import json, sys
data = json.load(sys.stdin)
for pod in data['items']:
    for vol in pod['spec'].get('volumes', []) or []:
        if vol.get('projected'):
            for src in vol['projected'].get('sources', []) or []:
                if 'serviceAccountToken' in src:
                    sa = src['serviceAccountToken']
                    exp = sa.get('expirationSeconds', 'default(长期)')
                    aud = sa.get('audience', '未绑定')
                    print(f'{pod["metadata"]["namespace"]}/{pod["metadata"]["name"]}: exp={exp}, aud={aud}')
"

这条命令帮你快速排查：哪些 Pod 还在用不绑定受众的长期 token，哪些已经迁移到短期 TokenRequest 模式。

落地路线与取舍

从传统 IAM 迁移到云原生身份模型，不是一步到位的事。以下是按优先级排列的实践路线：

优先级 1：消灭长期密钥。 把所有静态 API Key 和 Secret 替换为短期 Token。这是投入产出比最高的一步——一个泄露的长期密钥造成的损害，远大于短期 token 过期配置不当的影响。

优先级 2：为每个服务建立独立身份。 每个微服务用独立 ServiceAccount，不共享。身份越细粒度，授权越精确，审计越清晰。

优先级 3：引入策略引擎。 当服务数量超过 10 个，硬编码权限列表就开始失控。部署 OPA 或 Cedar 等策略引擎，把授权逻辑从业务代码中剥离。

优先级 4：服务间 mTLS。 在身份认证基础上叠加传输层加密。可以用 SPIRE 自动签发证书，也可以用 service mesh（Istio/Linkerd）的 mTLS 功能。两者不冲突——SPIRE 提供身份，mesh 提供加密通道。

取舍提醒： SPIFFE/SPIRE 的学习曲线比 K8s 原生方案陡，但跨集群、跨云场景下更通用。如果你的服务全部跑在一个 K8s 集群内，先用 ServiceAccount + TokenRequest + OPA 起步，够用后再考虑 SPIRE。不要为了"架构完美"在第一天就引入全套 SPIRE——渐进式落地比一步到位更可靠。

云原生安全的核心转变是：不再问"这个请求从哪个网段来"，而是问"这个请求是谁发出的、有没有资格做这件事"。 身份系统从配角变成主角，令牌从长期变短期，策略从人工审批变成代码定义。这三步走完，你的零信任才算有了真正的骨架。