Day-128-威胁建模实战

# Day 142: 威胁建模实战

> 应用安全系列第 35 天 | 预计阅读时间：50 分钟 | 难度：★★★★☆

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 威胁建模实战 - Day 142             │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ 威胁建模不是纸上谈兵，是实战分析。分析不落地，对策就是空的。今天深入威胁建模实战。

---

## 清单 目录

1. [威胁建模概述](#威胁建模概述)
2. [威胁建模方法对比](#威胁建模方法对比)
3. [STRIDE 实战](#stride 实战)
4. [PASTA 实战](#pasta 实战)
5. [LINDDUN 实战](#linddun 实战)
6. [数据流图绘制](#数据流图绘制)
7. [威胁识别技术](#威胁识别技术)
8. [风险评估方法](#风险评估方法)
9. [对策制定与验证](#对策制定与验证)
10. [实战案例演练](#实战案例演练)
11. [总结与思考](#总结与思考)
12. [参考资料](#参考资料)

---

## 威胁建模概述

### 什么是威胁建模

> ▎ 威胁建模不是猜攻击，是系统分析。分析不系统，防护就是零散的。

**定义与目标**：
```
威胁建模 (Threat Modeling) 是一种结构化的方法，用于识别、评估和缓解系统中的安全威胁。

核心目标：
1. 识别威胁
   - 系统性识别
   - 全面覆盖
   - 优先级排序

2. 评估风险
   - 可能性评估
   - 影响评估
   - 风险等级

3. 制定对策
   - 缓解措施
   - 实施优先级
   - 验证有效性
```

**威胁建模价值**：
```
投资回报：
- 早期发现威胁：修复成本降低 10 倍
- 系统性分析：遗漏威胁减少 50%
- 优先级排序：资源投入更精准
- 团队共识：安全理解更一致

实际收益：
- 安全漏洞减少：40-60%
- 安全返工减少：30-50%
- 安全审查时间：减少 25-40%
- 团队安全意识：显著提升
```

### 威胁建模时机

**最佳时机**：
```
┌─────────────────────────────────────────────────────────┐
│              威胁建模时机矩阵                            │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  项目阶段          威胁建模活动         投入产出比      │
│  ───────────────────────────────────────────────────    │
│  需求分析          初步威胁识别         ★★★★★          │
│  架构设计          详细威胁建模         ★★★★★          │
│  开发阶段          变更影响分析         ★★★★☆          │
│  测试阶段          威胁验证测试         ★★★☆☆          │
│  运营阶段          新威胁分析           ★★★☆☆          │
│                                                         │
│  关键触发点：                                           │
│  ✓ 新项目启动                                           │
│  ✓ 架构重大变更                                         │
│  ✓ 新增敏感数据                                         │
│  ✓ 安全事件发生                                         │
│  ✓ 合规要求变化                                         │
│                                                         │
└─────────────────────────────────────────────────────────┘
```

---

## 威胁建模方法对比

### 主流方法

> ▎ 方法不是选最好的，是选最合适的。方法不合适，效果就是打折的。

**方法对比表**：
```
┌─────────────────┬──────────────┬──────────────┬──────────────┬──────────────┐
│     方法        │   适用场景   │   复杂度     │   输出       │   学习曲线   │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   STRIDE        │  系统设计    │  中          │  威胁列表    │  低          │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   PASTA         │  业务风险    │  高          │  风险评级    │  高          │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   LINDDUN       │  隐私保护    │  中          │  隐私威胁    │  中          │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   TRIKE         │  风险管理    │  高          │  风险模型    │  高          │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   VAST          │  敏捷开发    │  低          │  威胁图谱    │  低          │
├─────────────────┼──────────────┼──────────────┼──────────────┼──────────────┤
│   OCTAVE        │  组织风险    │  高          │  风险评估    │  高          │
└─────────────────┴──────────────┴──────────────┴──────────────┴──────────────┘
```

### 方法选择指南

**选择决策树**：
```python
# 威胁建模方法选择
def select_threat_modeling_method(project_context):
    """根据项目上下文选择威胁建模方法"""
    
    # 1. 是否有隐私合规要求？
    if project_context.get('privacy_required', False):
        return 'LINDDUN'  # 隐私-focused
    
    # 2. 是否需要业务风险对齐？
    if project_context.get('business_risk_focus', False):
        return 'PASTA'  # 业务风险-focused
    
    # 3. 是否是敏捷/DevOps 环境？
    if project_context.get('agile_environment', False):
        return 'VAST'  # 敏捷-friendly
    
    # 4. 是否是系统设计阶段？
    if project_context.get('phase') == 'design':
        return 'STRIDE'  # 设计-focused
    
    # 5. 是否是组织级风险评估？
    if project_context.get('organizational_level', False):
        return 'OCTAVE'  # 组织级
    
    # 默认推荐
    return 'STRIDE'  # 最常用

# 项目上下文示例
project_contexts = [
    {
        'name': '电商平台',
        'phase': 'design',
        'privacy_required': True,
        'business_risk_focus': True,
        'agile_environment': True,
        'recommended': 'STRIDE + LINDDUN'  # 组合使用
    },
    {
        'name': '医疗系统',
        'phase': 'design',
        'privacy_required': True,
        'business_risk_focus': False,
        'agile_environment': False,
        'recommended': 'LINDDUN'  # 隐私优先
    },
    {
        'name': '金融科技',
        'phase': 'design',
        'privacy_required': True,
        'business_risk_focus': True,
        'agile_environment': False,
        'recommended': 'PASTA'  # 业务风险优先
    }
]
```

---

## STRIDE 实战

### STRIDE 详解

> ▎ STRIDE 不是 6 个字母，是 6 类威胁。类别不清楚，识别就是随机的。

**STRIDE 威胁分类**：
```
S - Spoofing (欺骗)
├── 定义：攻击者冒充合法实体
├── 目标：身份认证系统
├── 示例：
│   ├── 凭证窃取
│   ├── 会话劫持
│   ├── DNS 欺骗
│   └── IP 欺骗
└── 对策：强认证、双向认证、证书验证

T - Tampering (篡改)
├── 定义：恶意修改数据或代码
├── 目标：数据完整性
├── 示例：
│   ├── 数据篡改
│   ├── 代码注入
│   ├── 配置修改
│   └── 日志篡改
└── 对策：完整性检查、数字签名、版本控制

R - Repudiation (抵赖)
├── 定义：否认已执行的操作
├── 目标：可追溯性
├── 示例：
│   ├── 否认交易
│   ├── 否认访问
│   ├── 否认修改
│   └── 否认发送
└── 对策：审计日志、数字签名、时间戳

I - Information Disclosure (信息泄露)
├── 定义：敏感信息被未授权访问
├── 目标：数据机密性
├── 示例：
│   ├── 数据泄露
│   ├── 配置泄露
│   ├── 日志泄露
│   └── 元数据泄露
└── 对策：加密、访问控制、数据最小化

D - Denial of Service (拒绝服务)
├── 定义：使服务不可用
├── 目标：服务可用性
├── 示例：
│   ├── DDoS 攻击
│   ├── 资源耗尽
│   ├── 配置破坏
│   └── 依赖破坏
└── 对策：限流、冗余、弹性设计

E - Elevation of Privilege (权限提升)
├── 定义：获取未授权的权限
├── 目标：访问控制
├── 示例：
│   ├── 越权访问
│   ├── 提权攻击
│   ├── 权限绕过
│   └── 角色提升
└── 对策：最小权限、访问控制、职责分离
```

### STRIDE 实战演练

**实战步骤**：
```python
# STRIDE 威胁建模实战
class STRIDEThreatModeling:
    def __init__(self):
        self.stride_categories = {
            'spoofing': {
                'description': '身份欺骗威胁',
                'questions': [
                    '系统如何验证用户身份？',
                    '认证凭证如何保护？',
                    '如何防止会话劫持？',
                    '如何防止重放攻击？'
                ],
                'threats': [],
                'mitigations': []
            },
            'tampering': {
                'description': '数据篡改威胁',
                'questions': [
                    '数据完整性如何保证？',
                    '如何检测数据篡改？',
                    '代码完整性如何保护？',
                    '配置如何保护？'
                ],
                'threats': [],
                'mitigations': []
            },
            'repudiation': {
                'description': '行为抵赖威胁',
                'questions': [
                    '关键操作是否记录日志？',
                    '日志是否防篡改？',
                    '是否有数字签名？',
                    '时间是否同步？'
                ],
                'threats': [],
                'mitigations': []
            },
            'information_disclosure': {
                'description': '信息泄露威胁',
                'questions': [
                    '敏感数据如何加密？',
                    '谁可以访问什么数据？',
                    '数据传输如何保护？',
                    '日志是否包含敏感信息？'
                ],
                'threats': [],
                'mitigations': []
            },
            'denial_of_service': {
                'description': '拒绝服务威胁',
                'questions': [
                    '系统如何处理大流量？',
                    '资源使用如何限制？',
                    '是否有冗余设计？',
                    '如何快速恢复？'
                ],
                'threats': [],
                'mitigations': []
            },
            'elevation_of_privilege': {
                'description': '权限提升威胁',
                'questions': [
                    '权限如何分配？',
                    '如何防止越权？',
                    '管理接口如何保护？',
                    '如何审计权限使用？'
                ],
                'threats': [],
                'mitigations': []
            }
        }
    
    def model_system(self, system_design):
        """对系统进行 STRIDE 建模"""
        # 1. 分解系统
        components = self.decompose_system(system_design)
        
        # 2. 识别信任边界
        trust_boundaries = self.identify_trust_boundaries(components)
        
        # 3. 创建数据流图
        dfd = self.create_dfd(components, trust_boundaries)
        
        # 4. STRIDE 分析
        for component in components:
            self.analyze_component(component, dfd)
        
        # 5. 风险评估
        threats = self.assess_risks()
        
        # 6. 制定对策
        self.develop_mitigations(threats)
        
        return {
            'components': components,
            'trust_boundaries': trust_boundaries,
            'dfd': dfd,
            'threats': threats,
            'mitigations': self.all_mitigations
        }
    
    def decompose_system(self, system_design):
        """分解系统为组件"""
        components = []
        
        # 外部实体
        for entity in system_design.get('external_entities', []):
            components.append({
                'id': f"EE-{len(components)+1}",
                'type': 'external_entity',
                'name': entity['name'],
                'trust_level': 'untrusted',
                'description': entity.get('description', '')
            })
        
        # 进程
        for process in system_design.get('processes', []):
            components.append({
                'id': f"P-{len(components)+1}",
                'type': 'process',
                'name': process['name'],
                'trust_level': 'partial',
                'description': process.get('description', '')
            })
        
        # 数据存储
        for store in system_design.get('data_stores', []):
            components.append({
                'id': f"DS-{len(components)+1}",
                'type': 'data_store',
                'name': store['name'],
                'trust_level': 'partial',
                'data_classification': store.get('classification', 'internal')
            })
        
        return components
    
    def analyze_component(self, component, dfd):
        """分析组件威胁"""
        comp_type = component['type']
        
        if comp_type == 'external_entity':
            self.analyze_external_entity(component)
        elif comp_type == 'process':
            self.analyze_process(component, dfd)
        elif comp_type == 'data_store':
            self.analyze_data_store(component)
    
    def analyze_external_entity(self, entity):
        """分析外部实体威胁"""
        # Spoofing
        self.stride_categories['spoofing']['threats'].append({
            'id': f"SPOOF-{entity['id']}",
            'component': entity['name'],
            'description': f"攻击者可能冒充{entity['name']}",
            'risk_level': '高',
            'stride': 'S'
        })
    
    def analyze_process(self, process, dfd):
        """分析进程威胁"""
        # Tampering
        self.stride_categories['tampering']['threats'].append({
            'id': f"TAMPER-{process['id']}",
            'component': process['name'],
            'description': f"{process['name']}的输入可能被篡改",
            'risk_level': '高',
            'stride': 'T'
        })
        
        # Repudiation
        self.stride_categories['repudiation']['threats'].append({
            'id': f"REPUD-{process['id']}",
            'component': process['name'],
            'description': f"{process['name']}的操作可能被否认",
            'risk_level': '中',
            'stride': 'R'
        })
    
    def analyze_data_store(self, store):
        """分析数据存储威胁"""
        # Information Disclosure
        risk = '高' if store['data_classification'] == 'sensitive' else '中'
        self.stride_categories['information_disclosure']['threats'].append({
            'id': f"DISC-{store['id']}",
            'component': store['name'],
            'description': f"{store['name']}的数据可能被未授权访问",
            'risk_level': risk,
            'stride': 'I'
        })
        
        # Tampering
        self.stride_categories['tampering']['threats'].append({
            'id': f"TAMPER-{store['id']}",
            'component': store['name'],
            'description': f"{store['name']}的数据可能被篡改",
            'risk_level': '高',
            'stride': 'T'
        })
    
    def assess_risks(self):
        """风险评估"""
        all_threats = []
        
        for category, data in self.stride_categories.items():
            for threat in data['threats']:
                # 计算风险分数
                threat['risk_score'] = self.calculate_risk_score(threat)
                threat['priority'] = self.get_priority(threat['risk_score'])
                all_threats.append(threat)
        
        # 按风险排序
        all_threats.sort(key=lambda x: x['risk_score'], reverse=True)
        
        return all_threats
    
    def calculate_risk_score(self, threat):
        """计算风险分数"""
        likelihood_map = {'高': 3, '中': 2, '低': 1}
        likelihood = likelihood_map.get(threat.get('risk_level', '中'), 2)
        impact = likelihood_map.get(threat.get('risk_level', '中'), 2)
        return likelihood * impact
    
    def get_priority(self, risk_score):
        """获取优先级"""
        if risk_score >= 6:
            return 'P0'
        elif risk_score >= 4:
            return 'P1'
        else:
            return 'P2'
    
    def develop_mitigations(self, threats):
        """制定对策"""
        self.all_mitigations = []
        
        for threat in threats:
            if threat['priority'] in ['P0', 'P1']:
                mitigation = self.get_mitigation(threat)
                self.all_mitigations.append({
                    'threat_id': threat['id'],
                    'threat': threat['description'],
                    'mitigation': mitigation,
                    'status': 'planned',
                    'priority': threat['priority']
                })
    
    def get_mitigation(self, threat):
        """获取对策"""
        stride = threat.get('stride', '')
        
        mitigations = {
            'S': '实施强认证机制，如 MFA、证书认证',
            'T': '实施完整性验证，如数字签名、哈希校验',
            'R': '实施审计日志，确保操作可追溯',
            'I': '实施加密和访问控制，保护数据机密性',
            'D': '实施限流、冗余和弹性设计',
            'E': '实施最小权限和访问控制'
        }
        
        return mitigations.get(stride, '需要进一步分析')

# 使用示例
system_design = {
    'external_entities': [
        {'name': '用户', 'description': '系统最终用户'},
        {'name': '第三方 API', 'description': '外部服务接口'}
    ],
    'processes': [
        {'name': '认证服务', 'description': '处理用户认证'},
        {'name': '数据服务', 'description': '处理数据 CRUD'}
    ],
    'data_stores': [
        {'name': '用户数据库', 'classification': 'sensitive'},
        {'name': '日志存储', 'classification': 'internal'}
    ]
}

tm = STRIDEThreatModeling()
result = tm.model_system(system_design)
```

---

## PASTA 实战

### PASTA 七阶段

> ▎ PASTA 不是简单流程，是业务对齐。业务不对齐，安全就是孤立的。

**PASTA 流程**：
```
PASTA (Process for Attack Simulation and Threat Analysis):

阶段 1: 定义目标 (Define Objectives)
├── 业务目标
├── 安全目标
├── 合规要求
└── 风险偏好

阶段 2: 定义范围 (Define Scope)
├── 应用范围
├── 技术范围
├── 数据范围
└── 接口范围

阶段 3: 应用分解 (Application Decomposition)
├── 组件识别
├── 数据流分析
├── 信任边界
└── 依赖关系

阶段 4: 威胁分析 (Threat Analysis)
├── 威胁识别
├── 威胁分类
├── 威胁场景
└── 攻击树

阶段 5: 脆弱性分析 (Vulnerability Analysis)
├── 脆弱性识别
├── 脆弱性评估
├── 弱点映射
└── 控制差距

阶段 6: 攻击建模 (Attack Modeling)
├── 攻击路径
├── 攻击场景
├── 攻击模拟
└── 影响分析

阶段 7: 风险与影响分析 (Risk & Impact Analysis)
├── 风险评估
├── 影响分析
├── 风险排序
└── 对策建议
```

### PASTA 实战

**阶段 1-2 实现**：
```python
# PASTA 阶段 1-2: 定义目标和范围
class PASTAStage12:
    def __init__(self):
        self.objectives = {}
        self.scope = {}
    
    def define_objectives(self, business_context):
        """阶段 1: 定义目标"""
        self.objectives = {
            'business_objectives': self.extract_business_objectives(business_context),
            'security_objectives': self.define_security_objectives(business_context),
            'compliance_requirements': self.identify_compliance(business_context),
            'risk_appetite': self.assess_risk_appetite(business_context)
        }
        
        return self.objectives
    
    def extract_business_objectives(self, context):
        """提取业务目标"""
        return [
            {
                'id': 'BO-001',
                'description': '保护用户数据安全',
                'priority': '高',
                'metrics': ['数据泄露事件数', '加密覆盖率']
            },
            {
                'id': 'BO-002',
                'description': '确保服务高可用',
                'priority': '高',
                'metrics': ['可用性%', 'MTTR']
            },
            {
                'id': 'BO-003',
                'description': '满足合规要求',
                'priority': '高',
                'metrics': ['合规审计通过率']
            }
        ]
    
    def define_security_objectives(self, context):
        """定义安全目标"""
        return [
            {
                'id': 'SO-001',
                'description': '防止未授权数据访问',
                'related_business': 'BO-001',
                'controls': ['认证', '授权', '加密']
            },
            {
                'id': 'SO-002',
                'description': '检测和响应安全事件',
                'related_business': 'BO-002',
                'controls': ['监控', '告警', '响应']
            }
        ]
    
    def identify_compliance(self, context):
        """识别合规要求"""
        return [
            {
                'standard': 'GDPR',
                'requirements': ['数据保护', '用户权利', ' breach 通知'],
                'applicable': True
            },
            {
                'standard': 'PCI DSS',
                'requirements': ['支付数据安全', '访问控制', '加密'],
                'applicable': context.get('handles_payments', False)
            },
            {
                'standard': '等保 2.0',
                'requirements': ['安全设计', '安全测试', '安全运维'],
                'applicable': context.get('china_operation', False)
            }
        ]
    
    def assess_risk_appetite(self, context):
        """评估风险偏好"""
        return {
            'overall': 'medium',  # low/medium/high
            'by_category': {
                'data_breach': 'low',
                'service_outage': 'low',
                'compliance_violation': 'low',
                'reputation_damage': 'medium',
                'financial_loss': 'medium'
            }
        }
    
    def define_scope(self, system_context):
        """阶段 2: 定义范围"""
        self.scope = {
            'application_scope': self.define_application_scope(system_context),
            'technical_scope': self.define_technical_scope(system_context),
            'data_scope': self.define_data_scope(system_context),
            'interface_scope': self.define_interface_scope(system_context)
        }
        
        return self.scope
    
    def define_application_scope(self, context):
        """定义应用范围"""
        return {
            'in_scope': [
                'Web 应用',
                '移动应用',
                'API 服务',
                '管理后台'
            ],
            'out_of_scope': [
                '第三方系统',
                '遗留系统 (计划退役)'
            ],
            'boundaries': {
                'start': '用户请求入口',
                'end': '数据持久化'
            }
        }
    
    def define_technical_scope(self, context):
        """定义技术范围"""
        return {
            'technologies': [
                'React 前端',
                'Node.js 后端',
                'PostgreSQL 数据库',
                'Redis 缓存',
                'Kubernetes 容器'
            ],
            'infrastructure': {
                'cloud_provider': 'AWS',
                'regions': ['us-east-1', 'eu-west-1'],
                'network': 'VPC with private subnets'
            }
        }
    
    def define_data_scope(self, context):
        """定义数据范围"""
        return {
            'data_types': [
                {'type': '用户个人信息', 'classification': 'sensitive'},
                {'type': '认证凭证', 'classification': 'confidential'},
                {'type': '交易数据', 'classification': 'confidential'},
                {'type': '日志数据', 'classification': 'internal'}
            ],
            'data_flows': [
                '用户 → Web 应用 → API → 数据库',
                'Web 应用 → Redis → 缓存数据',
                'API → 第三方服务 → 外部数据'
            ]
        }
    
    def define_interface_scope(self, context):
        """定义接口范围"""
        return {
            'internal_interfaces': [
                'Web 应用 ↔ API 服务',
                'API 服务 ↔ 数据库',
                'API 服务 ↔ 缓存'
            ],
            'external_interfaces': [
                '用户 ↔ Web 应用',
                'API ↔ 第三方服务',
                '管理后台 ↔ 内部网络'
            ]
        }
```

---

## LINDDUN 实战

### LINDDUN 隐私威胁

> ▎ LINDDUN 不是 STRIDE 的变种，是隐私专用。隐私不专用，保护就是泛泛的。

**LINDDUN 分类**：
```
L - Linkability (关联性)
├── 定义：能够关联不同数据点识别个人
├── 示例：跨服务用户追踪、数据关联分析
└── 对策：数据隔离、假名化

I - Identifiability (可识别性)
├── 定义：能够识别特定个人
├── 示例：直接标识符、准标识符
└── 对策：匿名化、去标识化

N - Non-repudiation (不可抵赖性)
├── 定义：无法否认特定行为
├── 示例：数字签名、审计日志
└── 对策：选择性披露、隐私保护日志

D - Detectability (可检测性)
├── 定义：能够检测个人存在或行为
├── 示例：流量分析、行为分析
└── 对策：流量混淆、行为模糊

D - Disclosure (信息泄露)
├── 定义：个人数据被未授权访问
├── 示例：数据泄露、配置错误
└── 对策：加密、访问控制

U - Unawareness (不知情)
├── 定义：个人不知道数据被收集使用
├── 示例：隐蔽收集、隐藏条款
└── 对策：透明通知、明确同意

N - Non-compliance (不合规)
├── 定义：不符合隐私法规要求
├── 示例：超范围收集、无用户权利
└── 对策：合规审查、权利实现
```

### LINDDUN 实战

```python
# LINDDUN 隐私威胁建模
class LINDDUNThreatModeling:
    def __init__(self):
        self.linddun_categories = {
            'linkability': {
                'questions': [
                    '数据能否关联到同一用户？',
                    '跨服务是否有统一标识？',
                    '是否有用户行为追踪？'
                ],
                'threats': [],
                'mitigations': ['数据隔离', '假名化', '限制关联']
            },
            'identifiability': {
                'questions': [
                    '数据是否包含直接标识符？',
                    '数据是否包含准标识符？',
                    '能否通过组合识别个人？'
                ],
                'threats': [],
                'mitigations': ['匿名化', '去标识化', '数据最小化']
            },
            'non_repudiation': {
                'questions': [
                    '用户行为是否被完整记录？',
                    '记录是否可关联到个人？',
                    '用户能否否认特定行为？'
                ],
                'threats': [],
                'mitigations': ['选择性记录', '隐私保护日志']
            },
            'detectability': {
                'questions': [
                    '能否检测用户在线状态？',
                    '能否分析用户行为模式？',
                    '流量是否可分析？'
                ],
                'threats': [],
                'mitigations': ['流量混淆', '行为模糊', '元数据保护']
            },
            'disclosure': {
                'questions': [
                    '个人数据如何存储？',
                    '谁可以访问个人数据？',
                    '数据传输如何保护？'
                ],
                'threats': [],
                'mitigations': ['加密', '访问控制', '数据脱敏']
            },
            'unawareness': {
                'questions': [
                    '用户是否知道数据收集？',
                    '用户是否同意数据使用？',
                    '隐私政策是否清晰？'
                ],
                'threats': [],
                'mitigations': ['透明通知', '明确同意', '隐私政策']
            },
            'non_compliance': {
                'questions': [
                    '是否符合 GDPR 要求？',
                    '是否实现用户权利？',
                    '是否有数据保护官？'
                ],
                'threats': [],
                'mitigations': ['合规审查', '权利实现', 'DPO 任命']
            }
        }
    
    def model_privacy_threats(self, system_design):
        """建模隐私威胁"""
        # 1. 识别个人数据
        personal_data = self.identify_personal_data(system_design)
        
        # 2. 分析数据流
        data_flows = self.analyze_data_flows(system_design)
        
        # 3. LINDDUN 分析
        for category in self.linddun_categories.values():
            self.analyze_privacy_threats(category, personal_data, data_flows)
        
        # 4. 风险评估
        threats = self.assess_privacy_risks()
        
        # 5. 制定对策
        mitigations = self.develop_privacy_mitigations(threats)
        
        return {
            'personal_data': personal_data,
            'data_flows': data_flows,
            'threats': threats,
            'mitigations': mitigations
        }
    
    def identify_personal_data(self, system_design):
        """识别个人数据"""
        return {
            'direct_identifiers': [
                '姓名',
                '身份证号',
                '手机号',
                '邮箱',
                '用户 ID'
            ],
            'quasi_identifiers': [
                '出生日期',
                '性别',
                '邮编',
                '职业'
            ],
            'sensitive_data': [
                '健康信息',
                '生物特征',
                '政治观点',
                '宗教信仰'
            ],
            'behavioral_data': [
                '浏览历史',
                '购买记录',
                '位置信息'
            ]
        }
    
    def analyze_data_flows(self, system_design):
        """分析数据流"""
        return [
            {
                'id': 'DF-001',
                'source': '用户',
                'destination': 'Web 应用',
                'data_types': ['用户名', '密码', '个人信息'],
                'purpose': '用户注册'
            },
            {
                'id': 'DF-002',
                'source': 'Web 应用',
                'destination': '数据库',
                'data_types': ['用户数据', '会话数据'],
                'purpose': '数据持久化'
            },
            {
                'id': 'DF-003',
                'source': 'API 服务',
                'destination': '第三方服务',
                'data_types': ['用户 ID', '行为数据'],
                'purpose': '数据分析'
            }
        ]
    
    def analyze_privacy_threats(self, category, personal_data, data_flows):
        """分析隐私威胁"""
        # 针对每个数据流分析威胁
        for flow in data_flows:
            for question in category['questions']:
                # 简化实现
                threat = {
                    'category': list(self.linddun_categories.keys())[
                        list(self.linddun_categories.values()).index(category)
                    ],
                    'data_flow': flow['id'],
                    'data_types': flow['data_types'],
                    'description': f"基于问题：{question}",
                    'risk_level': self.assess_privacy_risk_level(flow, category)
                }
                category['threats'].append(threat)
    
    def assess_privacy_risk_level(self, data_flow, category):
        """评估隐私风险等级"""
        sensitive_types = ['sensitive_data', 'direct_identifiers']
        
        for data_type in data_flow['data_types']:
            if data_type in sensitive_types:
                return '高'
        
        return '中'
    
    def assess_privacy_risks(self):
        """评估隐私风险"""
        all_threats = []
        
        for category_name, category in self.linddun_categories.items():
            for threat in category['threats']:
                threat['priority'] = self.get_privacy_priority(threat)
                all_threats.append(threat)
        
        return all_threats
    
    def get_privacy_priority(self, threat):
        """获取隐私威胁优先级"""
        if threat['risk_level'] == '高':
            return 'P0'
        return 'P1'
    
    def develop_privacy_mitigations(self, threats):
        """制定隐私对策"""
        mitigations = []
        
        for threat in threats:
            if threat['priority'] in ['P0', 'P1']:
                category = self.linddun_categories.get(threat['category'], {})
                mitigation_options = category.get('mitigations', [])
                
                mitigations.append({
                    'threat': threat['description'],
                    'mitigations': mitigation_options[:2],  # 前两个对策
                    'priority': threat['priority']
                })
        
        return mitigations
```

---

## 数据流图绘制

### DFD 要素

> ▎ DFD 不是画图，是理解系统。系统不理解，威胁就是盲目的。

**DFD 符号**：
```
外部实体 (External Entity):
┌─────────┐
│  用户   │  ← 矩形，表示系统外的实体
└─────────┘

进程 (Process):
  ┌─────┐
 ( 处理 )  ← 圆角矩形，表示数据处理
  └─────┘

数据存储 (Data Store):
 ┌──────────┐
 │ 数据库   │  ← 开口矩形，表示数据存储
 └──────────┘

数据流 (Data Flow):
      ─────→   ← 箭头，表示数据流向

信任边界 (Trust Boundary):
  ═══════════  ← 虚线，表示信任边界
```

### DFD 绘制实战

```python
# DFD 绘制辅助
class DFDCreator:
    def __init__(self):
        self.elements = {
            'external_entities': [],
            'processes': [],
            'data_stores': [],
            'data_flows': [],
            'trust_boundaries': []
        }
    
    def create_dfd(self, system_description):
        """创建 DFD"""
        # 解析系统描述
        self.parse_system(system_description)
        
        # 生成 DFD
        dfd = {
            'diagram': self.generate_diagram(),
            'elements': self.elements,
            'analysis': self.analyze_dfd()
        }
        
        return dfd
    
    def parse_system(self, description):
        """解析系统"""
        # 识别外部实体
        self.elements['external_entities'] = [
            {'id': 'EE1', 'name': '用户', 'description': '系统使用者'},
            {'id': 'EE2', 'name': '管理员', 'description': '系统管理者'},
            {'id': 'EE3', 'name': '第三方服务', 'description': '外部 API'}
        ]
        
        # 识别进程
        self.elements['processes'] = [
            {'id': 'P1', 'name': '认证处理', 'description': '处理用户认证'},
            {'id': 'P2', 'name': '数据处理', 'description': '处理业务数据'},
            {'id': 'P3', 'name': '日志记录', 'description': '记录系统日志'}
        ]
        
        # 识别数据存储
        self.elements['data_stores'] = [
            {'id': 'DS1', 'name': '用户数据库', 'classification': 'sensitive'},
            {'id': 'DS2', 'name': '业务数据库', 'classification': 'internal'},
            {'id': 'DS3', 'name': '日志存储', 'classification': 'internal'}
        ]
        
        # 识别数据流
        self.elements['data_flows'] = [
            {'id': 'DF1', 'from': 'EE1', 'to': 'P1', 'data': '认证凭证'},
            {'id': 'DF2', 'from': 'P1', 'to': 'DS1', 'data': '用户信息'},
            {'id': 'DF3', 'from': 'P1', 'to': 'P2', 'data': '认证结果'},
            {'id': 'DF4', 'from': 'P2', 'to': 'DS2', 'data': '业务数据'},
            {'id': 'DF5', 'from': 'P2', 'to': 'P3', 'data': '操作日志'},
            {'id': 'DF6', 'from': 'P3', 'to': 'DS3', 'data': '日志记录'}
        ]
        
        # 识别信任边界
        self.elements['trust_boundaries'] = [
            {'id': 'TB1', 'description': '互联网边界', 'crosses': ['DF1']},
            {'id': 'TB2', 'description': '应用 - 数据边界', 'crosses': ['DF2', 'DF4', 'DF6']}
        ]
    
    def generate_diagram(self):
        """生成 DFD 图示"""
        # 这里生成文本表示的 DFD
        # 实际实现可以使用 graphviz 等工具
        diagram = """
        ┌─────────┐     认证凭证     ┌─────────┐
        │  用户   │ ───────────────→ │ 认证处理 │
        └─────────┘                  └────┬────┘
                                          │ 用户信息
                                          ↓
                                     ┌──────────┐
                                     │ 用户数据库 │
                                     └──────────┘
        """
        return diagram
    
    def analyze_dfd(self):
        """分析 DFD"""
        analysis = {
            'trust_boundary_crossings': len(self.elements['trust_boundaries']),
            'data_stores_with_sensitive_data': sum(
                1 for ds in self.elements['data_stores']
                if ds.get('classification') == 'sensitive'
            ),
            'external_data_flows': sum(
                1 for df in self.elements['data_flows']
                if df['from'].startswith('EE') or df['to'].startswith('EE')
            ),
            'recommendations': self.generate_recommendations()
        }
        
        return analysis
    
    def generate_recommendations(self):
        """生成建议"""
        return [
            '信任边界 crossings 需要加密',
            '敏感数据存储需要加密',
            '外部数据流需要验证',
            '所有进程需要审计日志'
        ]
```

---

## 实战案例演练

### 电商系统威胁建模

**案例背景**：
```
系统：电商平台
功能：用户注册、商品浏览、下单支付、订单管理
数据：用户信息、支付信息、订单数据
合规：GDPR、PCI DSS、等保 2.0
```

**威胁建模输出**：
```python
# 电商系统威胁建模示例
ecommerce_threat_model = {
    'system_overview': {
        'name': '电商平台',
        'components': ['Web 前端', 'API 服务', '支付服务', '数据库'],
        'data_types': ['用户信息', '支付信息', '订单数据']
    },
    'top_threats': [
        {
            'id': 'T-001',
            'category': 'STRIDE',
            'type': 'Spoofing',
            'description': '攻击者冒充用户进行购物',
            'component': '认证服务',
            'risk_score': 9,
            'priority': 'P0',
            'mitigation': '实施 MFA、会话管理'
        },
        {
            'id': 'T-002',
            'category': 'STRIDE',
            'type': 'Information Disclosure',
            'description': '支付信息泄露',
            'component': '支付服务',
            'risk_score': 10,
            'priority': 'P0',
            'mitigation': 'PCI DSS 合规、端到端加密'
        },
        {
            'id': 'T-003',
            'category': 'STRIDE',
            'type': 'Tampering',
            'description': '订单数据被篡改',
            'component': '订单服务',
            'risk_score': 8,
            'priority': 'P0',
            'mitigation': '数据完整性校验、审计日志'
        },
        {
            'id': 'T-004',
            'category': 'LINDDUN',
            'type': 'Linkability',
            'description': '用户行为跨平台追踪',
            'component': '数据分析',
            'risk_score': 6,
            'priority': 'P1',
            'mitigation': '数据隔离、用户同意'
        },
        {
            'id': 'T-005',
            'category': 'STRIDE',
            'type': 'Denial of Service',
            'description': 'DDoS 攻击导致服务不可用',
            'component': 'Web 前端',
            'risk_score': 7,
            'priority': 'P1',
            'mitigation': 'DDoS 防护、弹性伸缩'
        }
    ],
    'mitigation_plan': [
        {
            'threat_id': 'T-001',
            'action': '部署 MFA',
            'owner': '安全团队',
            'deadline': '2 周',
            'status': 'planned'
        },
        {
            'threat_id': 'T-002',
            'action': '实施令牌化',
            'owner': '支付团队',
            'deadline': '4 周',
            'status': 'planned'
        },
        {
            'threat_id': 'T-003',
            'action': '添加数字签名',
            'owner': '后端团队',
            'deadline': '3 周',
            'status': 'planned'
        }
    ]
}
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。威胁建模不是一次活动，是持续实践。实践不停，安全才有保障。

---

## 总结与思考

### 核心要点回顾

> ▎ 复盘四步法：回顾目标、评估结果、分析原因、总结经验。别跳过——这是闭环。

**威胁建模框架**：
```
1. 方法选择
   - STRIDE: 系统设计
   - PASTA: 业务风险
   - LINDDUN: 隐私保护

2. 系统分解
   - 组件识别
   - 数据流图
   - 信任边界

3. 威胁识别
   - 分类分析
   - 场景构建
   - 攻击树

4. 风险评估
   - 可能性
   - 影响
   - 优先级

5. 对策制定
   - 缓解措施
   - 实施计划
   - 验证测试
```

**关键成功因素**：
```
1. 团队协作
   - 多角色参与
   - 知识共享
   - 共识建立

2. 持续更新
   - 变更触发
   - 定期审查
   - 新威胁分析

3. 落地实施
   - 对策跟踪
   - 验证测试
   - 效果评估
```

### 深入思考问题

> ▎ 只动手不动脑，那是码农。动手又动脑，才是工程师。

**威胁建模挑战**：
```
1. 时间压力
   - 项目进度紧
   - 建模耗时
   - 平衡方案：轻量级建模

2. 专业知识
   - 安全知识不足
   - 威胁理解有限
   - 平衡方案：培训 + 模板

3. 落地困难
   - 对策实施难
   - 资源有限
   - 平衡方案：优先级 + 分阶段
```

### 实战建议

> ▎ 我给你指了路，走不走是你的事。机会给了，抓不抓得住看你。

**入门建议**：
```
1. 从简单开始
   - 小系统试点
   - 使用 STRIDE
   - 快速见效

2. 使用工具
   - Threat Dragon
   - Microsoft TMT
   - 降低门槛

3. 建立模板
   - 标准流程
   - 输出模板
   - 提高效率
```

**进阶建议**：
```
1. 方法组合
   - STRIDE + LINDDUN
   - 根据场景选择
   - 全面覆盖

2. 自动化
   - 自动 DFD
   - 自动威胁识别
   - 持续建模

3. 度量改进
   - 威胁发现率
   - 对策实施率
   - 持续改进
```

---

## 参考资料

### 学习资源
```
- Microsoft Threat Modeling Tool
  https://www.microsoft.com/securityengineering/sdl/threatmodeling

- OWASP Threat Modeling
  https://owasp.org/www-community/Threat_Modeling

- IriusRisk Threat Modeling
  https://www.iriusrisk.com/
```

### 工具资源
```
- OWASP Threat Dragon
  https://owasp.org/www-project-threat-dragon/

- Microsoft Threat Modeling Tool
  https://www.microsoft.com/download/details.aspx?id=49168

- pytm (Python Threat Modeling)
  https://github.com/izar/pytm
```

### 书籍推荐
```
- 《Threat Modeling: Designing for Security》
- 《The Art of Threat Modeling》
- 《Secure by Design》(威胁建模章节)
```

---

**标记 明日预告**：Day 143 - 安全需求分析

> ▎ 威胁建模是识别风险，安全需求是定义要求——明天看安全需求分析。

> 本文内容仅供学习和研究使用，请勿用于非法目的。所有实验请在隔离环境中进行。

---

*本文是 365 天信息安全技术系列的第 142 篇，应用安全部分第 35 篇，精编版本*

*应用安全核心系列继续！*