IAM 与零信任身份管理

# Access Token：短时效（5-15 分钟），无状态
# Refresh Token：长时效，存储在 HttpOnly Cookie 或安全存储

# Token 验证中间件（Python/FastAPI）
from jose import jwt, JWTError
from fastapi import HTTPException, Depends
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def verify_token(credentials: HTTPBearer = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(
            token,
            key=PUBLIC_KEY,
            algorithms=["ES256"],
            audience="api.example.com",
            issuer="https://auth.example.com",
        )
        
        # 检查 Token 是否在撤销列表（Redis）
        jti = payload.get("jti")
        if await redis.exists(f"revoked:{jti}"):
            raise HTTPException(status_code=401, detail="Token revoked")
        
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# RBAC 模型
roles:
  admin:
    permissions:
      - user:*
      - post:*
      - setting:*
  
  editor:
    permissions:
      - post:read
      - post:write
      - post:publish
  
  viewer:
    permissions:
      - post:read

# 用户-角色映射
user_roles:
  alice: [admin]
  bob: [editor]
  carol: [viewer]

# ABAC 策略引擎
class ABACEngine:
    def evaluate(self, subject, resource, action, environment):
        """
        subject: { user, roles, department, clearance }
        resource: { owner, classification, department }
        action: read | write | delete
        environment: { time, location, device_trust }
        """
        policy = {
            "allow": [
                # 规则 1：数据所有者可以读写自己的数据
                {
                    "subject.user": resource.owner,
                    "action": ["read", "write"],
                },
                # 规则 2：同部门成员在工作时间可以读取
                {
                    "subject.department": resource.department,
                    "resource.classification": "internal",
                    "action": "read",
                    "environment.time": {"between": ["09:00", "18:00"]},
                },
                # 规则 3：高管可以访问高密级数据
                {
                    "subject.clearance": {"gte": resource.classification},
                    "action": "read",
                },
            ],
            "deny": [
                # 拒绝规则覆盖允许规则
                {
                    "environment.device_trust": {"lt": 0.5},
                },
            ]
        }
        
        # 先评估 deny 规则
        for rule in policy["deny"]:
            if self.match(rule, subject, resource, action, environment):
                return False
        
        # 再评估 allow 规则
        for rule in policy["allow"]:
            if self.match(rule, subject, resource, action, environment):
                return True
        
        return False  # 默认拒绝

# Google Zanzibar 风格的 ReBAC
# 关系如：document:readme#owner@user:alice

relations = {
    "document:readme": {
        "owner": ["user:alice"],
        "editor": ["user:bob", "group:engineering#member"],
        "viewer": ["group:company#member"],
    }
}

# 检查权限：alice 是否是 readme 的 owner？
check("document:readme", "owner", "user:alice")  # ✅ True

# 检查权限：bob 是否可以 viewer readme？（viewer ⊆ editor ⊆ owner）
check("document:readme", "viewer", "user:bob")   # ✅ True（继承 editor）

class DeviceTrustScore:
    def calculate(self, device) -> float:
        score = 1.0
        
        # 设备合规性检查
        if not device.encrypted:
            score *= 0.3
        if not device.os_updated:
            score *= 0.5
        if device.jailbroken:
            score *= 0.0  # 完全不可信
        
        # 位置检查
        if device.location not in self.allowed_countries:
            score *= 0.4
        
        # 行为分析
        if device.anomaly_score > 0.8:
            score *= 0.3
        
        return score

# 动态访问决策
if device_trust < 0.5:
    require_step_up_auth()  # 要求额外认证
elif device_trust < 0.3:
    deny_access()           # 拒绝访问
else:
    allow_access()          # 正常访问