Day-140-漏洞管理与响应

# Day 154: 漏洞管理与响应

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

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 漏洞管理与响应 - Day 154           │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ 漏洞管理不是修完就结束，是持续运营。运营不持续，漏洞就是反复的。今天深入漏洞管理与响应。

---

## 清单 目录

1. [漏洞管理概述](#漏洞管理概述)
2. [漏洞管理流程](#漏洞管理流程)
3. [漏洞分类分级](#漏洞分类分级)
4. [漏洞修复流程](#漏洞修复流程)
5. [应急响应机制](#应急响应机制)
6. [漏洞披露管理](#漏洞披露管理)
7. [度量与报告](#度量与报告)
8. [工具与平台](#工具与平台)
9. [实战案例演练](#实战案例演练)
10. [总结与思考](#总结与思考)
11. [参考资料](#参考资料)

---

## 漏洞管理概述

### 什么是漏洞管理

> ▎ 漏洞管理不是修 bug，是管风险。风险不管控，漏洞就是持续的。

**定义与目标**：
```
漏洞管理 (Vulnerability Management) 是对组织内所有安全漏洞进行持续识别、评估、修复和监控的系统化过程。

核心目标：
1. 全面识别
   - 所有资产漏洞
   - 所有来源漏洞
   - 持续发现

2. 风险评估
   - 漏洞严重程度
   - 业务影响
   - 修复优先级

3. 修复跟踪
   - 修复计划
   - 进度跟踪
   - 效果验证

4. 持续监控
   - 新漏洞发现
   - 修复状态
   - 趋势分析
```

**管理挑战**：
```
漏洞管理挑战:

1. 漏洞数量多
   - 每天 50+ 新 CVE
   - 平均组织 10000+ 漏洞
   - 人工无法处理

2. 修复资源有限
   - 安全团队人手不足
   - 开发资源紧张
   - 修复 backlog 积压

3. 优先级难定
   - 所有漏洞都重要？
   - 业务影响难评估
   - 修复成本不一

4. 修复验证难
   - 修复是否有效
   - 是否引入新问题
   - 回归测试成本
```

### 管理框架

**漏洞管理生命周期**：
```
┌─────────────────────────────────────────────────────────┐
│              漏洞管理生命周期                            │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  发现 (Identify)                                        │
│  ├── 资产发现                                           │
│  ├── 漏洞扫描                                           │
│  └── 威胁情报                                           │
│                                                         │
│  评估 (Assess)                                          │
│  ├── 风险评估                                           │
│  ├── 优先级排序                                         │
│  └── 修复方案                                           │
│                                                         │
│  修复 (Remediate)                                       │
│  ├── 修复实施                                           │
│  ├── 验证测试                                           │
│  └── 文档记录                                           │
│                                                         │
│  报告 (Report)                                          │
│  ├── 状态报告                                           │
│  ├── 趋势分析                                           │
│  └── 合规报告                                           │
│                                                         │
│  改进 (Improve)                                         │
│  ├── 流程优化                                           │
│  ├── 工具改进                                           │
│  └── 能力提升                                           │
│                                                         │
└─────────────────────────────────────────────────────────┘
```

---

## 漏洞管理流程

### 标准流程

> ▎ 漏洞管理不是随意修，是有流程的。流程不规范，管理就是混乱的。

**流程定义**：
```python
# 漏洞管理流程
class VulnerabilityManagementProcess:
    """漏洞管理流程"""
    
    # 流程阶段
    STAGES = {
        'DISCOVERY': {
            'name': '发现',
            'description': '发现新漏洞',
            'inputs': ['扫描结果', '威胁情报', '外部报告'],
            'outputs': ['漏洞记录'],
            'owner': '安全团队'
        },
        'TRIAGE': {
            'name': '分诊',
            'description': '初步评估和分类',
            'inputs': ['漏洞记录'],
            'outputs': ['分类结果', '初步优先级'],
            'owner': '安全团队'
        },
        'ASSESSMENT': {
            'name': '评估',
            'description': '详细风险评估',
            'inputs': ['分类结果'],
            'outputs': ['风险评分', '修复优先级'],
            'owner': '安全团队'
        },
        'ASSIGNMENT': {
            'name': '分配',
            'description': '分配给修复团队',
            'inputs': ['风险评分', '优先级'],
            'outputs': ['工单', '负责人'],
            'owner': '安全团队'
        },
        'REMEDIATION': {
            'name': '修复',
            'description': '实施修复',
            'inputs': ['工单'],
            'outputs': ['修复代码', '修复验证'],
            'owner': '开发团队'
        },
        'VERIFICATION': {
            'name': '验证',
            'description': '验证修复效果',
            'inputs': ['修复代码'],
            'outputs': ['验证结果'],
            'owner': '安全团队'
        },
        'CLOSURE': {
            'name': '关闭',
            'description': '关闭漏洞记录',
            'inputs': ['验证结果'],
            'outputs': ['关闭记录', '经验总结'],
            'owner': '安全团队'
        }
    }
    
    def __init__(self):
        self.vulnerabilities = {}
        self.workflow = {}
        self.metrics = {}
    
    def create_vulnerability(self, vuln_data):
        """创建漏洞记录"""
        vuln_id = self.generate_vuln_id()
        
        vuln = {
            'id': vuln_id,
            'source': vuln_data.get('source'),
            'type': vuln_data.get('type'),
            'asset': vuln_data.get('asset'),
            'cve': vuln_data.get('cve'),
            'severity': vuln_data.get('severity'),
            'cvss_score': vuln_data.get('cvss_score'),
            'description': vuln_data.get('description'),
            'stage': 'DISCOVERY',
            'created_at': datetime.utcnow(),
            'updated_at': datetime.utcnow(),
            'owner': None,
            'due_date': None,
            'timeline': [
                {
                    'stage': 'DISCOVERY',
                    'timestamp': datetime.utcnow(),
                    'actor': 'system',
                    'note': 'Vulnerability discovered'
                }
            ]
        }
        
        self.vulnerabilities[vuln_id] = vuln
        
        return vuln_id
    
    def transition(self, vuln_id, new_stage, actor, note=''):
        """转换阶段"""
        if vuln_id not in self.vulnerabilities:
            raise ValueError(f"Vulnerability {vuln_id} not found")
        
        vuln = self.vulnerabilities[vuln_id]
        old_stage = vuln['stage']
        
        # 验证阶段转换
        if not self.is_valid_transition(old_stage, new_stage):
            raise ValueError(f"Invalid transition: {old_stage} -> {new_stage}")
        
        # 更新阶段
        vuln['stage'] = new_stage
        vuln['updated_at'] = datetime.utcnow()
        vuln['timeline'].append({
            'stage': new_stage,
            'timestamp': datetime.utcnow(),
            'actor': actor,
            'note': note,
            'from_stage': old_stage
        })
        
        # 阶段特定处理
        self.handle_stage_transition(vuln, old_stage, new_stage)
        
        return True
    
    def is_valid_transition(self, from_stage, to_stage):
        """验证阶段转换"""
        valid_transitions = {
            'DISCOVERY': ['TRIAGE'],
            'TRIAGE': ['ASSESSMENT', 'CLOSED'],
            'ASSESSMENT': ['ASSIGNMENT', 'ACCEPTED'],
            'ASSIGNMENT': ['REMEDIATION', 'DEFERRED'],
            'REMEDIATION': ['VERIFICATION', 'ASSIGNMENT'],
            'VERIFICATION': ['CLOSURE', 'REMEDIATION'],
            'CLOSURE': ['REOPENED'],
            'ACCEPTED': ['ASSIGNMENT'],
            'DEFERRED': ['ASSIGNMENT'],
            'REOPENED': ['TRIAGE']
        }
        
        return to_stage in valid_transitions.get(from_stage, [])
    
    def handle_stage_transition(self, vuln, from_stage, to_stage):
        """处理阶段转换"""
        if to_stage == 'TRIAGE':
            # 初步分类
            vuln['category'] = self.categorize_vulnerability(vuln)
        
        elif to_stage == 'ASSESSMENT':
            # 风险评估
            vuln['risk_score'] = self.assess_risk(vuln)
            vuln['priority'] = self.get_priority(vuln['risk_score'])
            vuln['due_date'] = self.calculate_due_date(vuln['priority'])
        
        elif to_stage == 'ASSIGNMENT':
            # 分配负责人
            vuln['owner'] = self.assign_owner(vuln)
            self.create_ticket(vuln)
        
        elif to_stage == 'CLOSURE':
            # 关闭处理
            vuln['closed_at'] = datetime.utcnow()
            vuln['time_to_remediate'] = (
                vuln['closed_at'] - vuln['created_at']
            ).days
    
    def categorize_vulnerability(self, vuln):
        """分类漏洞"""
        # 按类型分类
        if vuln.get('cve'):
            return 'cve'
        elif vuln.get('type') == 'dependency':
            return 'dependency'
        elif vuln.get('type') == 'configuration':
            return 'configuration'
        else:
            return 'other'
    
    def assess_risk(self, vuln):
        """风险评估"""
        # CVSS 基础分
        cvss = vuln.get('cvss_score', 0)
        
        # 资产重要性调整
        asset_criticality = vuln.get('asset_criticality', 'medium')
        asset_multiplier = {'critical': 1.5, 'high': 1.3, 'medium': 1.0, 'low': 0.8}
        
        # 暴露程度调整
        exposure = vuln.get('exposure', 'internal')
        exposure_multiplier = {'public': 1.5, 'partner': 1.2, 'internal': 1.0}
        
        risk_score = cvss * 10 * \
                     asset_multiplier.get(asset_criticality, 1.0) * \
                     exposure_multiplier.get(exposure, 1.0)
        
        return min(risk_score, 100)
    
    def get_priority(self, risk_score):
        """获取优先级"""
        if risk_score >= 80:
            return 'P0'
        elif risk_score >= 60:
            return 'P1'
        elif risk_score >= 40:
            return 'P2'
        else:
            return 'P3'
    
    def calculate_due_date(self, priority):
        """计算截止日期"""
        from datetime import timedelta
        
        sla = {
            'P0': 7,    # 7 天
            'P1': 30,   # 30 天
            'P2': 90,   # 90 天
            'P3': 180   # 180 天
        }
        
        days = sla.get(priority, 90)
        return datetime.utcnow() + timedelta(days=days)
    
    def assign_owner(self, vuln):
        """分配负责人"""
        # 根据资产类型分配
        asset_type = vuln.get('asset_type', 'application')
        
        ownership_map = {
            'application': 'app-team',
            'infrastructure': 'infra-team',
            'network': 'network-team',
            'cloud': 'cloud-team'
        }
        
        return ownership_map.get(asset_type, 'security-team')
    
    def create_ticket(self, vuln):
        """创建工单"""
        # 集成工单系统 (Jira, ServiceNow 等)
        pass
    
    def generate_vuln_id(self):
        """生成漏洞 ID"""
        import uuid
        return f"VULN-{uuid.uuid4().hex[:8].upper()}"
```

---

## 漏洞分类分级

### 分类体系

> ▎ 漏洞不是都一样，是有类别的。类别不清楚，管理就是混乱的。

**漏洞分类**：
```python
# 漏洞分类体系
class VulnerabilityClassification:
    """漏洞分类体系"""
    
    # 按来源分类
    BY_SOURCE = {
        'SCANNER': {
            'name': '扫描发现',
            'tools': ['Nessus', 'Qualys', 'OpenVAS'],
            'reliability': 'high'
        },
        'SAST': {
            'name': '静态分析',
            'tools': ['Bandit', 'Semgrep', 'Fortify'],
            'reliability': 'medium'
        },
        'DAST': {
            'name': '动态分析',
            'tools': ['ZAP', 'Burp', 'Nessus'],
            'reliability': 'high'
        },
        'DEPENDENCY': {
            'name': '依赖扫描',
            'tools': ['Snyk', 'Safety', 'Dependabot'],
            'reliability': 'high'
        },
        'EXTERNAL': {
            'name': '外部报告',
            'sources': ['CVE', '厂商公告', '白帽子'],
            'reliability': 'varies'
        },
        'INTERNAL': {
            'name': '内部发现',
            'sources': ['员工报告', '监控发现'],
            'reliability': 'varies'
        }
    }
    
    # 按类型分类
    BY_TYPE = {
        'INJECTION': {
            'name': '注入漏洞',
            'examples': ['SQL 注入', '命令注入', 'XSS'],
            'severity_typical': 'high'
        },
        'AUTHENTICATION': {
            'name': '认证漏洞',
            'examples': ['弱密码', '会话固定', '凭证泄露'],
            'severity_typical': 'high'
        },
        'AUTHORIZATION': {
            'name': '授权漏洞',
            'examples': ['越权访问', '权限提升'],
            'severity_typical': 'high'
        },
        'CONFIGURATION': {
            'name': '配置漏洞',
            'examples': ['默认密码', '开放端口', '错误配置'],
            'severity_typical': 'medium'
        },
        'DEPENDENCY': {
            'name': '依赖漏洞',
            'examples': ['CVE 漏洞', '过时组件'],
            'severity_typical': 'varies'
        },
        'DATA': {
            'name': '数据漏洞',
            'examples': ['敏感数据泄露', '未加密存储'],
            'severity_typical': 'high'
        }
    }
    
    # 按严重程度分类
    BY_SEVERITY = {
        'CRITICAL': {
            'cvss_range': (9.0, 10.0),
            'sla_days': 7,
            'description': '可能被立即利用，造成严重损害'
        },
        'HIGH': {
            'cvss_range': (7.0, 8.9),
            'sla_days': 30,
            'description': '可能被利用，造成较大损害'
        },
        'MEDIUM': {
            'cvss_range': (4.0, 6.9),
            'sla_days': 90,
            'description': '可能被利用，造成中等损害'
        },
        'LOW': {
            'cvss_range': (0.1, 3.9),
            'sla_days': 180,
            'description': '利用难度大，损害较小'
        },
        'INFO': {
            'cvss_range': (0, 0),
            'sla_days': None,
            'description': '信息类问题，无需修复'
        }
    }
    
    def classify(self, vulnerability):
        """分类漏洞"""
        classification = {
            'source': self.classify_by_source(vulnerability),
            'type': self.classify_by_type(vulnerability),
            'severity': self.classify_by_severity(vulnerability)
        }
        
        return classification
    
    def classify_by_source(self, vulnerability):
        """按来源分类"""
        source = vulnerability.get('source', 'UNKNOWN')
        return self.BY_SOURCE.get(source, {
            'name': '未知来源',
            'reliability': 'unknown'
        })
    
    def classify_by_type(self, vulnerability):
        """按类型分类"""
        vuln_type = vulnerability.get('type', 'OTHER')
        return self.BY_TYPE.get(vuln_type, {
            'name': '其他类型',
            'severity_typical': 'medium'
        })
    
    def classify_by_severity(self, vulnerability):
        """按严重程度分类"""
        cvss = vulnerability.get('cvss_score', 0)
        
        for severity, info in self.BY_SEVERITY.items():
            min_cvss, max_cvss = info['cvss_range']
            if min_cvss <= cvss <= max_cvss:
                return {
                    'level': severity,
                    **info
                }
        
        return self.BY_SEVERITY['INFO']
```

### 优先级排序

```python
# 漏洞优先级排序
class VulnerabilityPrioritization:
    """漏洞优先级排序"""
    
    def __init__(self):
        # 优先级因素权重
        self.weights = {
            'cvss': 0.25,           # CVSS 分数
            'exploitability': 0.25, # 可利用性
            'asset_value': 0.20,    # 资产价值
            'exposure': 0.15,       # 暴露程度
            'threat_intel': 0.15    # 威胁情报
        }
    
    def calculate_priority_score(self, vulnerability, context):
        """计算优先级分数"""
        scores = {
            'cvss': self.score_cvss(vulnerability),
            'exploitability': self.score_exploitability(vulnerability),
            'asset_value': self.score_asset_value(vulnerability, context),
            'exposure': self.score_exposure(vulnerability, context),
            'threat_intel': self.score_threat_intel(vulnerability)
        }
        
        # 加权计算
        total_score = 0
        for factor, weight in self.weights.items():
            total_score += scores[factor] * weight
        
        return {
            'total_score': total_score,
            'priority': self.score_to_priority(total_score),
            'factor_scores': scores
        }
    
    def score_cvss(self, vulnerability):
        """CVSS 评分"""
        cvss = vulnerability.get('cvss_score', 0)
        return cvss * 10  # 0-100
    
    def score_exploitability(self, vulnerability):
        """可利用性评分"""
        score = 50  # 基础分
        
        # 有公开利用代码
        if vulnerability.get('exploit_available'):
            score += 30
        
        # 在野利用
        if vulnerability.get('exploit_in_the_wild'):
            score += 20
        
        # 自动化工具
        if vulnerability.get('automated_exploit'):
            score += 10
        
        return min(score, 100)
    
    def score_asset_value(self, vulnerability, context):
        """资产价值评分"""
        asset_criticality = context.get('asset_criticality', 'medium')
        
        scores = {
            'critical': 100,
            'high': 75,
            'medium': 50,
            'low': 25
        }
        
        return scores.get(asset_criticality, 50)
    
    def score_exposure(self, vulnerability, context):
        """暴露程度评分"""
        exposure = context.get('exposure', 'internal')
        
        scores = {
            'public': 100,
            'partner': 70,
            'internal': 40,
            'isolated': 20
        }
        
        return scores.get(exposure, 40)
    
    def score_threat_intel(self, vulnerability):
        """威胁情报评分"""
        score = 50  # 基础分
        
        # 活跃威胁
        if vulnerability.get('active_threat'):
            score += 30
        
        # 针对本行业
        if vulnerability.get('targets_industry'):
            score += 20
        
        return min(score, 100)
    
    def score_to_priority(self, score):
        """分数转优先级"""
        if score >= 80:
            return 'P0'
        elif score >= 60:
            return 'P1'
        elif score >= 40:
            return 'P2'
        else:
            return 'P3'
```

---

## 漏洞修复流程

### 修复流程

> ▎ 漏洞修复不是开发自己修，是有流程的。流程不清晰，修复就是低效的。

**修复工作流**：
```python
# 漏洞修复工作流
class RemediationWorkflow:
    """漏洞修复工作流"""
    
    def __init__(self):
        self.states = {
            'OPEN': ['IN_PROGRESS', 'DEFERRED', 'ACCEPTED'],
            'IN_PROGRESS': ['VERIFIED', 'BLOCKED'],
            'BLOCKED': ['IN_PROGRESS', 'OPEN'],
            'VERIFIED': ['CLOSED', 'REOPENED'],
            'CLOSED': ['REOPENED'],
            'DEFERRED': ['OPEN', 'ACCEPTED'],
            'ACCEPTED': ['OPEN', 'CLOSED']
        }
    
    def create_ticket(self, vulnerability):
        """创建修复工单"""
        ticket = {
            'id': self.generate_ticket_id(),
            'vulnerability_id': vulnerability['id'],
            'cve': vulnerability.get('cve'),
            'title': f"{vulnerability.get('cve') or 'Security Issue'} in {vulnerability.get('asset')}",
            'description': vulnerability.get('description'),
            'severity': vulnerability.get('severity'),
            'priority': vulnerability.get('priority'),
            'assignee': vulnerability.get('owner'),
            'reporter': 'security-team',
            'state': 'OPEN',
            'created_at': datetime.utcnow(),
            'due_date': vulnerability.get('due_date'),
            'sla_days': self.get_sla_days(vulnerability.get('severity')),
            'steps': [],
            'comments': []
        }
        
        # 添加修复步骤模板
        ticket['steps'] = self.get_remediation_steps(vulnerability)
        
        return ticket
    
    def get_remediation_steps(self, vulnerability):
        """获取修复步骤模板"""
        vuln_type = vulnerability.get('type', 'unknown')
        
        templates = {
            'dependency': [
                {'name': '确认受影响版本', 'status': 'pending'},
                {'name': '查找修复版本', 'status': 'pending'},
                {'name': '更新依赖', 'status': 'pending'},
                {'name': '运行测试', 'status': 'pending'},
                {'name': '验证修复', 'status': 'pending'}
            ],
            'configuration': [
                {'name': '确认错误配置', 'status': 'pending'},
                {'name': '制定修复方案', 'status': 'pending'},
                {'name': '实施修复', 'status': 'pending'},
                {'name': '验证配置', 'status': 'pending'}
            ],
            'code': [
                {'name': '分析漏洞原因', 'status': 'pending'},
                {'name': '编写修复代码', 'status': 'pending'},
                {'name': '代码审查', 'status': 'pending'},
                {'name': '运行测试', 'status': 'pending'},
                {'name': '部署验证', 'status': 'pending'}
            ]
        }
        
        return templates.get(vuln_type, templates['code'])
    
    def update_ticket(self, ticket_id, updates):
        """更新工单"""
        ticket = self.get_ticket(ticket_id)
        if not ticket:
            return False
        
        # 状态转换
        if 'state' in updates:
            new_state = updates['state']
            if new_state not in self.states.get(ticket['state'], []):
                raise ValueError(f"Invalid state transition")
            
            ticket['state'] = new_state
            ticket['comments'].append({
                'timestamp': datetime.utcnow(),
                'author': 'system',
                'comment': f"State changed to {new_state}"
            })
        
        # 其他更新
        for key, value in updates.items():
            if key != 'state':
                ticket[key] = value
        
        ticket['updated_at'] = datetime.utcnow()
        
        return True
    
    def complete_step(self, ticket_id, step_index, result):
        """完成修复步骤"""
        ticket = self.get_ticket(ticket_id)
        if not ticket:
            return False
        
        if step_index >= len(ticket['steps']):
            return False
        
        ticket['steps'][step_index]['status'] = 'completed'
        ticket['steps'][step_index]['result'] = result
        ticket['steps'][step_index]['completed_at'] = datetime.utcnow()
        
        return True
    
    def get_sla_days(self, severity):
        """获取 SLA 天数"""
        sla = {
            'CRITICAL': 7,
            'HIGH': 30,
            'MEDIUM': 90,
            'LOW': 180
        }
        return sla.get(severity, 90)
    
    def generate_ticket_id(self):
        """生成工单 ID"""
        import uuid
        return f"REM-{uuid.uuid4().hex[:8].upper()}"
    
    def get_ticket(self, ticket_id):
        """获取工单"""
        # 实际实现从数据库获取
        pass
    
    def get_overdue_tickets(self):
        """获取逾期工单"""
        overdue = []
        now = datetime.utcnow()
        
        for ticket in self.all_tickets:
            if ticket['state'] not in ['CLOSED', 'ACCEPTED']:
                if now > ticket['due_date']:
                    overdue.append({
                        'id': ticket['id'],
                        'cve': ticket['cve'],
                        'severity': ticket['severity'],
                        'days_overdue': (now - ticket['due_date']).days
                    })
        
        return overdue
```

---

## 应急响应机制

### 紧急漏洞响应

> ▎ 应急响应不是临时抱佛脚，是有预案的。预案不充分，响应就是混乱的。

**应急响应流程**：
```python
# 应急响应管理
class EmergencyResponse:
    """应急响应管理"""
    
    # 紧急漏洞标准
    EMERGENCY_CRITERIA = {
        'active_exploitation': True,      # 在野利用
        'critical_severity': True,         # 严重级别
        'wide_impact': True,               # 影响广泛
        'no_workaround': True              # 无缓解措施
    }
    
    # 响应团队角色
    ROLES = {
        'incident_commander': '事件指挥官',
        'security_lead': '安全负责人',
        'engineering_lead': '工程负责人',
        'communications_lead': '沟通负责人',
        'legal_lead': '法务负责人'
    }
    
    def __init__(self):
        self.emergencies = {}
        self.response_team = {}
    
    def declare_emergency(self, vulnerability):
        """宣布紧急状态"""
        if not self.is_emergency(vulnerability):
            raise ValueError("不符合紧急标准")
        
        emergency = {
            'id': self.generate_emergency_id(),
            'vulnerability': vulnerability,
            'declared_at': datetime.utcnow(),
            'phase': 'DETECTION',
            'team': self.response_team.copy(),
            'timeline': [],
            'actions': [],
            'communications': []
        }
        
        self.emergencies[emergency['id']] = emergency
        
        # 通知响应团队
        self.notify_team(emergency)
        
        return emergency
    
    def is_emergency(self, vulnerability):
        """判断是否为紧急情况"""
        # Log4j 级别
        if vulnerability.get('cve') == 'CVE-2021-44228':
            return True
        
        # 在野利用 + 严重
        if vulnerability.get('exploit_in_the_wild') and \
           vulnerability.get('severity') == 'CRITICAL':
            return True
        
        # 核心基础设施
        if vulnerability.get('affects_critical_infrastructure'):
            return True
        
        return False
    
    def advance_phase(self, emergency_id, new_phase, actions):
        """推进响应阶段"""
        emergency = self.emergencies.get(emergency_id)
        if not emergency:
            return False
        
        phases = ['DETECTION', 'ASSESSMENT', 'CONTAINMENT', 
                  'REMEDIATION', 'RECOVERY', 'LESSONS']
        
        current_index = phases.index(emergency['phase'])
        new_index = phases.index(new_phase)
        
        if new_index != current_index + 1:
            raise ValueError("只能推进到下一阶段")
        
        emergency['phase'] = new_phase
        emergency['timeline'].append({
            'phase': new_phase,
            'timestamp': datetime.utcnow(),
            'action': f"Advanced to {new_phase}"
        })
        
        for action in actions:
            emergency['actions'].append({
                'phase': new_phase,
                'action': action,
                'timestamp': datetime.utcnow()
            })
        
        return True
    
    def create_communication(self, emergency_id, audience, message):
        """创建沟通记录"""
        emergency = self.emergencies.get(emergency_id)
        if not emergency:
            return None
        
        comm = {
            'id': self.generate_comm_id(),
            'emergency_id': emergency_id,
            'audience': audience,
            'message': message,
            'sent_at': datetime.utcnow(),
            'channel': self.get_channel(audience)
        }
        
        emergency['communications'].append(comm)
        
        return comm
    
    def get_channel(self, audience):
        """获取沟通渠道"""
        channels = {
            'internal_team': 'slack',
            'engineering': 'email',
            'management': 'email',
            'customers': 'status_page',
            'public': 'blog'
        }
        return channels.get(audience, 'email')
    
    def generate_emergency_id(self):
        """生成应急 ID"""
        import uuid
        return f"EMG-{uuid.uuid4().hex[:8].upper()}"
    
    def generate_comm_id(self):
        """生成沟通 ID"""
        import uuid
        return f"COMM-{uuid.uuid4().hex[:6].upper()}"
    
    def notify_team(self, emergency):
        """通知响应团队"""
        # 实际实现发送通知
        pass
    
    def generate_post_mortem(self, emergency_id):
        """生成事后报告"""
        emergency = self.emergencies.get(emergency_id)
        if not emergency:
            return None
        
        return {
            'emergency_id': emergency_id,
            'vulnerability': emergency['vulnerability'],
            'timeline': emergency['timeline'],
            'actions_taken': emergency['actions'],
            'what_went_well': [],
            'what_went_wrong': [],
            'improvements': [],
            'action_items': []
        }
```

---

## 度量与报告

### 关键指标

> ▎ 度量不是为了好看，是为了改进。改进不基于数据，管理就是盲目的。

**关键指标定义**：
```python
# 漏洞管理指标
class VulnerabilityMetrics:
    """漏洞管理指标"""
    
    def __init__(self):
        self.vulnerabilities = []
    
    def calculate_metrics(self, time_period='30d'):
        """计算指标"""
        return {
            'discovery': self.discovery_metrics(time_period),
            'remediation': self.remediation_metrics(time_period),
            'backlog': self.backlog_metrics(),
            'sla': self.sla_metrics(time_period),
            'trend': self.trend_metrics(time_period)
        }
    
    def discovery_metrics(self, time_period):
        """发现指标"""
        return {
            'total_discovered': self.count_discovered(time_period),
            'by_severity': self.count_by_severity(time_period),
            'by_source': self.count_by_source(time_period),
            'discovery_rate': self.discovery_rate(time_period)
        }
    
    def remediation_metrics(self, time_period):
        """修复指标"""
        return {
            'total_remediated': self.count_remediated(time_period),
            'remediation_rate': self.remediation_rate(time_period),
            'mttr': self.calculate_mttr(time_period),
            'first_time_fix_rate': self.first_time_fix_rate(time_period)
        }
    
    def backlog_metrics(self):
        """积压指标"""
        return {
            'total_open': self.count_open(),
            'by_severity': self.count_open_by_severity(),
            'by_age': self.count_open_by_age(),
            'overdue': self.count_overdue()
        }
    
    def sla_metrics(self, time_period):
        """SLA 指标"""
        return {
            'sla_compliance': self.sla_compliance_rate(time_period),
            'by_severity': self.sla_compliance_by_severity(time_period),
            'average_days_to_remediate': self.avg_days_to_remediate(time_period)
        }
    
    def trend_metrics(self, time_period):
        """趋势指标"""
        return {
            'vulnerability_trend': self.vulnerability_trend(time_period),
            'remediation_trend': self.remediation_trend(time_period),
            'backlog_trend': self.backlog_trend(time_period)
        }
    
    # 具体计算方法
    def count_discovered(self, time_period):
        """统计发现数量"""
        cutoff = self.get_cutoff_date(time_period)
        return len([v for v in self.vulnerabilities 
                   if v['created_at'] >= cutoff])
    
    def count_remediated(self, time_period):
        """统计修复数量"""
        cutoff = self.get_cutoff_date(time_period)
        return len([v for v in self.vulnerabilities 
                   if v.get('closed_at') and v['closed_at'] >= cutoff])
    
    def calculate_mttr(self, time_period):
        """计算平均修复时间"""
        cutoff = self.get_cutoff_date(time_period)
        
        remediated = [v for v in self.vulnerabilities 
                     if v.get('closed_at') and v['closed_at'] >= cutoff]
        
        if not remediated:
            return 0
        
        total_days = sum(
            (v['closed_at'] - v['created_at']).days 
            for v in remediated
        )
        
        return total_days / len(remediated)
    
    def sla_compliance_rate(self, time_period):
        """计算 SLA 合规率"""
        cutoff = self.get_cutoff_date(time_period)
        
        remediated = [v for v in self.vulnerabilities 
                     if v.get('closed_at') and v['closed_at'] >= cutoff]
        
        if not remediated:
            return 100.0
        
        compliant = len([v for v in remediated 
                        if v['closed_at'] <= v.get('due_date', datetime.max)])
        
        return (compliant / len(remediated)) * 100
    
    def get_cutoff_date(self, time_period):
        """获取截止日期"""
        from datetime import timedelta
        
        if time_period == '7d':
            return datetime.utcnow() - timedelta(days=7)
        elif time_period == '30d':
            return datetime.utcnow() - timedelta(days=30)
        elif time_period == '90d':
            return datetime.utcnow() - timedelta(days=90)
        else:
            return datetime.utcnow() - timedelta(days=30)
```

### 报告模板

```markdown
# 漏洞管理月度报告

## 执行摘要

### 关键指标
| 指标 | 本月 | 上月 | 变化 |
|------|------|------|------|
| 新发现漏洞 | 150 | 180 | -17% |
| 已修复漏洞 | 120 | 100 | +20% |
| 平均修复时间 | 25 天 | 30 天 | -17% |
| SLA 合规率 | 85% | 80% | +5% |

### 风险概览
| 严重程度 | 新增 | 修复 | 积压 | 逾期 |
|----------|------|------|------|------|
| Critical | 5 | 5 | 2 | 1 |
| High | 25 | 20 | 30 | 5 |
| Medium | 60 | 50 | 100 | 15 |
| Low | 60 | 45 | 150 | 20 |

## 详细分析

### 漏洞发现趋势
[图表：月度漏洞发现趋势]

### 修复效率分析
[图表：MTTR 趋势]

### SLA 合规分析
[图表：SLA 合规率]

### TOP 漏洞类型
1. 依赖漏洞 (40%)
2. 配置漏洞 (25%)
3. 注入漏洞 (15%)
4. 其他 (20%)

## 重点漏洞

### CVE-2024-XXXX
- 发现日期：2024-01-15
- 严重程度：Critical
- 状态：已修复
- 修复时间：5 天

### CVE-2024-YYYY
- 发现日期：2024-01-20
- 严重程度：High
- 状态：修复中
- 预计修复：2024-02-10

## 改进建议

1. 加快 Critical 漏洞修复速度
2. 减少逾期漏洞数量
3. 提高依赖更新频率

## 下月目标

1. SLA 合规率达到 90%
2. 平均修复时间降低到 20 天
3. 逾期漏洞减少 50%
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。漏洞管理不是修修补补，是体系运营。运营不体系，安全就是救火的。

---

## 总结与思考

### 核心要点回顾

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

**漏洞管理框架**：
```
1. 流程管理
   - 发现 → 评估 → 修复 → 验证
   - 阶段转换
   - 工单跟踪

2. 分类分级
   - 来源分类
   - 类型分类
   - 严重程度

3. 优先级排序
   - CVSS 基础
   - 可利用性
   - 业务影响

4. 应急响应
   - 紧急漏洞
   - 响应流程
   - 事后总结
```

**关键成功因素**：
```
1. 流程化
   - 明确流程
   - 责任到人
   - SLA 定义

2. 自动化
   - 自动发现
   - 自动分派
   - 自动报告

3. 持续化
   - 持续监控
   - 持续改进
   - 持续运营
```

### 实战建议

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

**管理建设**：
```
1. 流程建立
   - 定义流程
   - 明确角色
   - 制定 SLA

2. 工具选型
   - 漏洞管理平台
   - 工单系统
   - 报告工具

3. 持续运营
   - 定期报告
   - 指标跟踪
   - 持续改进
```

---

## 参考资料

### 学习资源
```
- NIST Vulnerability Management Guide
  https://csrc.nist.gov/publications

- SANS Vulnerability Management
  https://www.sans.org/white-papers/vulnerability-management/

- FIRST Vulnerability Management
  https://www.first.org/
```

### 工具资源
```
- DefectDojo
  https://www.defectdojo.org/

- Jira Security
  https://www.atlassian.com/software/jira/features/security

- ServiceNow VRM
  https://www.servicenow.com/products/vulnerability-response.html
```

### 标准规范
```
- ISO 27001 (Vulnerability Management)
- NIST SP 800-40 (Vulnerability Management Guide)
```

### 书籍推荐
```
- 《Vulnerability Management Best Practices》
- 《Security Operations Center》
- 《Incident Response & Vulnerability Management》
```

---

**标记 明日预告**：Day 155 - 应用安全总结与展望

> ▎ 漏洞管理是运营实践，应用安全总结是阶段回顾——明天看应用安全总结与展望。

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

---

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

*应用安全核心系列 (Day 107-154) 完成！共 48 篇文章！*