威胁建模与安全评审：从 STRIDE 分析到发布门禁的完整流程

缺陷修复成本随阶段指数增长：
  设计阶段发现：  ~1x
  编码阶段发现：  ~5x
  测试阶段发现：  ~15x
  生产事故发现：  ~50x+
  → 在设计阶段做威胁建模，是最划算的安全投入

威胁建模五步（微软 SDL 风格）：
  1. 定义范围与资产（保护什么）
  2. 画数据流图（DFD）（数据怎么流）
  3. 识别威胁（STRIDE/威胁库）（能被怎样攻击）
  4. 评估与排序（风险评分）
  5. 设计缓解措施 + 跟踪落地
      ↓
   持续迭代（每次架构变更重新建模）

# stride_analysis.py — 对每个数据流元素应用 STRIDE
ELEMENT_TYPES = {"external_entity", "process", "data_store", "data_flow"}

def analyze_element_with_stride(element: dict) -> list[dict]:
    """对单个 DFD 元素分析六类威胁。"""
    threats = []
    kind = element["type"]
    for threat in STRIDE:
        applicable = threat_applicable(kind, threat)
        if not applicable:
            continue
        threats.append({
            "threat": threat,
            "element": element["name"],
            "question": question_for(kind, threat),
            "likelihood": estimate_likelihood(element, threat),
            "impact": estimate_impact(element, threat),
        })
    return threats

def threat_applicable(kind, threat):
    """不同元素面对的威胁集不同。"""
    mapping = {
        "external_entity": {"S", "R"},
        "process":         {"S", "T", "R", "I", "D", "E"},
        "data_store":      {"T", "I", "D", "R"},
        "data_flow":       {"T", "I", "D"},
    }
    return threat in mapping.get(kind, set())

DFD 符号：
  外部实体（External Entity）   □   用户、第三方系统
  处理过程（Process）           ○   服务、函数
  数据存储（Data Store）        ══   数据库、缓存
  数据流（Data Flow）           →   请求、响应

示例：订单系统的 DFD
  用户(□) →登录→ 认证服务(○) →读取→ 用户DB(══)
   用户(□) →下单→ 订单服务(○) →写入→ 订单DB(══)
                     ↓调用
              支付网关(外部实体 □)

# dfd.py — 用结构化数据定义数据流图
DFD = {
    "assets": ["订单数据", "用户凭证", "支付信息"],
    "elements": [
        {"id": "u1", "type": "external_entity", "name": "用户"},
        {"id": "p1", "type": "process", "name": "认证服务"},
        {"id": "p2", "type": "process", "name": "订单服务"},
        {"id": "d1", "type": "data_store", "name": "用户DB"},
        {"id": "d2", "type": "data_store", "name": "订单DB"},
        {"id": "e1", "type": "external_entity", "name": "支付网关"},
    ],
    "flows": [
        {"from": "u1", "to": "p1", "data": "凭证"},
        {"from": "p1", "to": "d1", "data": "读取/写入用户"},
        {"from": "u1", "to": "p2", "data": "订单指令"},
        {"from": "p2", "to": "d2", "data": "订单写入"},
        {"from": "p2", "to": "e1", "data": "支付请求"},
        {"from": "e1", "to": "p2", "data": "支付结果"},
    ],
}

信任边界（Trust Boundary）：
  数据跨过信任边界 = 可信度发生变化 = 风险点

  用户(不可信) ──边界──▶ 认证服务(可信) ──边界──▶ 内网服务(高可信)
                          ↑ 每次跨边界都要"重新验证"

在 DFD 中标注边界，威胁建模自动聚焦"跨边界的数据流"

def find_trust_boundary_crossings(dfd, boundaries) -> list[dict]:
    """找出跨越信任边界的数据流（风险焦点）。"""
    risky = []
    for flow in dfd["flows"]:
        src_zone = zone_of(dfd, flow["from"])
        dst_zone = zone_of(dfd, flow["to"])
        if src_zone != dst_zone:
            risky.append({**flow, "boundary": f"{src_zone}→{dst_zone}"})
    return risky

# threat_library.py — 常见威胁模式库
THREAT_LIBRARY = [
    {
        "id": "T-001",
        "pattern": "untrusted_input_to_sql",
        "stride": ["T"],
        "desc": "不可信输入进入 SQL 查询（注入）",
        "detection": "参数化查询缺失",
        "mitigation": ["prepared_statements", "ORM", "input_validation"],
        "cvss_hint": 8.5,
    },
    {
        "id": "T-002",
        "pattern": "excessive_session",
        "stride": ["I"],
        "desc": "会话/凭证有效期过长或未轮换",
        "detection": "cookie 无过期、无 SSO 集成",
        "mitigation": ["short_ttl", "rotation", "mfa"],
    },
    {
        "id": "T-003",
        "pattern": "sensitive_logging",
        "stride": ["I"],
        "desc": "日志记录敏感信息（PII/凭证）",
        "detection": "日志含 body/headers",
        "mitigation": ["redaction", "log_scrubber", "no_creds_in_logs"],
    },
    # ... 完整库可达数百条
]

def match_threat_library(component_desc: str) -> list[dict]:
    """按组件特征匹配威胁库中的成熟模式。"""
    return [t for t in THREAT_LIBRARY
            if any(kw in component_desc.lower() for kw in t["keywords"])]

目标：窃取用户支付信息
├── AND 需要同时达成
│   ├── 获取数据库访问
│   │   ├── OR 途径
│   │   │   ├── 利用 SQL 注入
│   │   │   ├── 泄露 DBA 凭证
│   │   │   └── 利用 RCE 漏洞
│   └── 绕过 DLP/审计
└── AND 需要同时达成
    ├── 截获传输中的流量（无 TLS/弱 TLS）
    └── 解密（弱加密/密钥泄露）

# attack_tree.py — 攻击树建模
class AttackTree:
    def __init__(self, root, children=None, operator="OR"):
        self.root = root
        self.children = children or []
        self.operator = operator   # OR: 任一子路径即可达成

    def all_paths(self):
        """枚举所有攻击路径（叶子组合）。"""
        if not self.children:
            return [[self.root]]
        paths = []
        for child in self.children:
            for sub in child.all_paths():
                paths.append([self.root] + sub)
        return paths

def evaluate_tree(tree, likelihood_fn, min_cost=0) -> list[dict]:
    """对每条攻击路径评估可行性与成本。"""
    findings = []
    for path in tree.all_paths():
        lik = max(likelihood_fn(step) for step in path)   # OR 取最大
        cost = sum(cost_fn(step) for step in path)
        if lik > min_cost:
            findings.append({"path": path, "likelihood": lik, "cost": cost})
    return sorted(findings, key=lambda f: f["likelihood"], reverse=True)

# risk_rating.py — 风险评分与排序
def score_risk(likelihood: int, impact: int) -> str:
    """
    likelihood/impact ∈ 1-5。
    High: >= 12 或 (5,3) 以上
    Medium: 6-11
    Low: <= 5
    """
    score = likelihood * impact
    if score >= 12 or (likelihood >= 4 and impact >= 4):
        return "HIGH"
    if score >= 6:
        return "MEDIUM"
    return "LOW"

def prioritize_threats(threats: list[dict]) -> list[dict]:
    """按风险等级 + 缓解成本排序，输出处理顺序。"""
    for t in threats:
        t["score"] = score_risk(t["likelihood"], t["impact"])
        t["priority"] = {
            "HIGH": 0, "MEDIUM": 1, "LOW": 2
        }[t["score"]]
    # 高优优先；同级按缓解成本升序
    return sorted(threats, key=lambda t: (t["priority"], t.get("cost", 0)))

处理威胁的四种选择：
  · 缓解（Mitigate）：实施控制，降低风险
  · 转移（Transfer）：购买保险/依赖第三方控制
  · 规避（Avoid）：改变设计，去掉风险路径
  · 接受（Accept）：明确接受残余风险（需管理层批准）

门禁原则：HIGH 风险必须缓解或显式接受，不允许默认忽略

@dataclass
class ResidualRisk:
    threat_id: str
    mitigation: str
    residual_level: str    # HIGH/MEDIUM/LOW
    accepted_by: str = ""
    accepted_at: str = ""
    review_due: str = ""

def validate_no_unaccepted_high(risks) -> bool:
    """发布门禁：不允许存在未处理的 HIGH 残余风险。"""
    unaccepted_high = [r for r in risks
                       if r.residual_level == "HIGH" and not r.accepted_by]
    if unaccepted_high:
        raise SecurityGateFailure(
            f"存在未接受的 HIGH 风险: {[r.threat_id for r in unaccepted_high]}")
    return True

# threat_model.py — 用代码定义 DFD 并自动识别威胁
from pytm import TM, ExternalEntity, Process, DataStore, DataFlow, Boundary

tm = TM("订单系统威胁模型")
user = ExternalEntity(tm, "用户")
auth = Process(tm, "认证服务")
orders = Process(tm, "订单服务")
db = DataStore(tm, "订单数据库")
gw = ExternalEntity(tm, "支付网关")

login_flow = DataFlow(tm, "凭证", source=user, destination=auth)
order_flow = DataFlow(tm, "下单", source=user, destination=orders)
write_flow = DataFlow(tm, "写订单", source=orders, destination=db)
pay_flow = DataFlow(tm, "支付", source=orders, destination=gw)

tm.process()   # 自动对每个元素应用 STRIDE 生成威胁清单
# 输出：威胁列表 + 每个威胁的建议缓解

威胁建模输出 → 安全评审输入：
  · 威胁清单（按风险排序）
  · 缓解措施建议
  · 需要安全团队评审的 HIGH 项
  · 待补强的架构决策记录

# review_checklist.py — 分层安全评审清单
AUTHN_CHECKLIST = [
    "所有敏感接口要求认证",
    "密码存储用 bcrypt/argon2（非 MD5/SHA）",
    "MFA 应用于特权操作",
    "会话有超时与轮换",
    "禁用默认/弱凭证",
]

AUTHZ_CHECKLIST = [
    "服务端强制授权（不只前端隐藏按钮）",
    "对象级授权（Ownership 校验）",
    "角色最小化，无通配权限",
    "越权测试覆盖（水平/垂直）",
]

DATA_CHECKLIST = [
    "传输加密（TLS 1.2+）",
    "静态敏感数据加密",
    "日志脱敏（无 PII/凭证）",
    "备份加密与访问控制",
]

INPUT_CHECKLIST = [
    "输入校验（白名单优先）",
    "参数化查询（防注入）",
    "输出编码（防 XSS）",
    "上传文件安全（见文件上传专题）",
]

def run_checklist(component, checklist) -> dict:
    return {item: assess_item(component, item) for item in checklist}

# security_gate.py — 发布门禁
def run_security_gate(feature: dict, findings: list[dict]) -> dict:
    """发布前安全门禁：HIGH 必堵、MEDIUM 有截止日。"""
    highs = [f for f in findings if f["level"] == "HIGH"]
    mediums = [f for f in findings if f["level"] == "MEDIUM"]

    if highs:
        return {"verdict": "BLOCK", "reason": f"{len(highs)} 个 HIGH 未解决"}
    if any(m["due"] > now() for m in mediums):
        return {"verdict": "BLOCK", "reason": "存在过期未处理的 MEDIUM"}
    return {"verdict": "PASS", "deferred": mediums}

def register_gate_check(gate_fn):
    """在 CI 中注册安全门禁检查项。"""
    def wrapper():
        try:
            gate_fn()
            return {"check": gate_fn.__name__, "status": "PASS"}
        except SecurityGateFailure as e:
            return {"check": gate_fn.__name__, "status": "FAIL",
                    "reason": str(e)}
    return wrapper

# .github/workflows/security-gate.yml
name: Security Review Gate
on: [pull_request]
jobs:
  threat-model-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: 运行威胁模型门禁
        run: |
          python scripts/run_threat_gate.py \
            --model threat_models/order_system.py \
            --max-high 0 --max-medium-due 7
      - name: 运行安全评审清单
        run: |
          python scripts/run_review_checklist.py \
            --component order-service

评审触发时机：
  · 新系统立项（必须）
  · 架构变更（必须）
  · 新增外部集成/信任边界（必须）
  · 敏感功能上线（必须）
  · 常规功能（轻量 checklist）

评审频率：
  · 新项目：设计期 1 次 + 上线前 1 次
  · 存量系统：季度/半年复审 + 事件驱动复审

def create_review_record(threats, mitigations, decisions) -> dict:
    """安全评审记录：可追踪、可审计。"""
    return {
        "id": uuid4(),
        "date": now(),
        "assets": DFD["assets"],
        "threats": [t["id"] for t in threats],
        "mitigations": mitigations,
        "decisions": decisions,     # 缓解/转移/规避/接受
        "residual": compute_residual(threats, mitigations),
        "reviewer": current_user(),
    }

# 评审记录入库，供下次复审与合规审计

def version_threat_model(model, change_note) -> None:
    """威胁模型随代码库版本化，变更可追溯。"""
    # 存 threat_models/<component>_v<version>.json
    save_model(model)
    audit_log("threat_model", action="UPDATE",
              component=model["component"], note=change_note)

安全左移全景：
  设计期：威胁建模 → 安全设计评审 → 威胁清单
  编码期：SAST 扫描 + 安全编码规范 + 依赖扫描
  测试期：DAST + 渗透测试 + 威胁用例
  发布期：安全门禁（HIGH 清零）+ 配置扫描
  运行期：RASP + 监控 + 事件响应
  → 威胁建模是这条链的"第一公里"，决定后续投入方向