Day-138-依赖漏洞管理

# Day 152: 依赖漏洞管理

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

---

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

---

## 清单 目录

1. [依赖漏洞管理概述](#依赖漏洞管理概述)
2. [漏洞生命周期管理](#漏洞生命周期管理)
3. [漏洞评估方法](#漏洞评估方法)
4. [修复策略与流程](#修复策略与流程)
5. [应急响应机制](#应急响应机制)
6. [供应商风险管理](#供应商风险管理)
7. [自动化管理实践](#自动化管理实践)
8. [度量与改进](#度量与改进)
9. [实战案例演练](#实战案例演练)
10. [总结与思考](#总结与思考)
11. [参考资料](#参考资料)

---

## 依赖漏洞管理概述

### 什么是依赖漏洞管理

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

**定义与目标**：
```
依赖漏洞管理 (Dependency Vulnerability Management) 是对应用程序中第三方依赖的安全漏洞进行持续识别、评估、修复和监控的过程。

核心目标：
1. 持续识别
   - 新依赖自动扫描
   - 新漏洞及时通知
   - 传递依赖追踪

2. 风险评估
   - 漏洞严重程度
   - 可利用性分析
   - 业务影响评估

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

4. 持续监控
   - 漏洞状态监控
   - 依赖更新监控
   - 趋势分析报告
```

**管理挑战**：
```
现代应用依赖管理挑战:

1. 依赖数量多
   - 平均应用：150+ 直接依赖
   - 传递依赖：500+ 间接依赖
   - 人工无法管理

2. 漏洞发现快
   - Log4j: 1 天内全球知晓
   - 平均漏洞披露：每天 50+ CVE
   - 需要快速响应

3. 修复复杂度高
   - 版本兼容性
   - 传递依赖冲突
   - 测试验证成本

4. 供应链复杂
   - 多层供应商
   - 传递信任
   - 难以追溯
```

### 管理框架

**管理流程**：
```
┌─────────────────────────────────────────────────────────┐
│              依赖漏洞管理流程                            │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  发现 (Discover)                                        │
│  ├── 依赖清单建立                                       │
│  ├── 漏洞扫描                                           │
│  └── 持续监控                                           │
│                                                         │
│  评估 (Assess)                                          │
│  ├── 严重程度评估                                       │
│  ├── 可利用性分析                                       │
│  └── 业务影响评估                                       │
│                                                         │
│  修复 (Remediate)                                       │
│  ├── 修复方案制定                                       │
│  ├── 修复实施                                           │
│  └── 修复验证                                           │
│                                                         │
│  监控 (Monitor)                                         │
│  ├── 状态跟踪                                           │
│  ├── 趋势分析                                           │
│  └── 持续改进                                           │
│                                                         │
└─────────────────────────────────────────────────────────┘
```

---

## 漏洞生命周期管理

### 漏洞生命周期

> ▎ 漏洞不是发现就修复，是有生命周期的。周期不管理，修复就是混乱的。

**漏洞生命周期阶段**：
```python
# 漏洞生命周期管理
class VulnerabilityLifecycle:
    """漏洞生命周期管理"""
    
    # 生命周期阶段
    STAGES = {
        'DISCOVERED': {
            'name': '已发现',
            'description': '漏洞被扫描发现',
            'actions': ['initial_assessment', 'assign_owner']
        },
        'ASSESSING': {
            'name': '评估中',
            'description': '正在评估风险和影响',
            'actions': ['risk_assessment', 'impact_analysis']
        },
        'PLANNED': {
            'name': '已计划',
            'description': '修复计划已制定',
            'actions': ['schedule_fix', 'allocate_resources']
        },
        'IN_PROGRESS': {
            'name': '修复中',
            'description': '正在实施修复',
            'actions': ['update_dependency', 'run_tests']
        },
        'VERIFIED': {
            'name': '已验证',
            'description': '修复已验证',
            'actions': ['verify_fix', 'document']
        },
        'CLOSED': {
            'name': '已关闭',
            'description': '漏洞已关闭',
            'actions': ['archive', 'lessons_learned']
        },
        'ACCEPTED': {
            'name': '已接受',
            'description': '风险已接受 (不修复)',
            'actions': ['risk_acceptance', 'compensating_controls']
        },
        'MITIGATED': {
            'name': '已缓解',
            'description': '风险已缓解 (临时措施)',
            'actions': ['implement_workaround', 'monitor']
        }
    }
    
    def __init__(self):
        self.vulnerabilities = {}
        self.metrics = {
            'mttd': [],  # Mean Time to Detect
            'mttr': [],  # Mean Time to Remediate
            'open_count': 0,
            'closed_count': 0
        }
    
    def create_vulnerability(self, vuln_data):
        """创建漏洞记录"""
        vuln_id = self.generate_vuln_id()
        
        vuln = {
            'id': vuln_id,
            'cve': vuln_data.get('cve'),
            'component': vuln_data.get('component'),
            'version': vuln_data.get('version'),
            'severity': vuln_data.get('severity'),
            'cvss_score': vuln_data.get('cvss_score'),
            'description': vuln_data.get('description'),
            'stage': 'DISCOVERED',
            'created_at': datetime.utcnow(),
            'updated_at': datetime.utcnow(),
            'owner': None,
            'due_date': self.calculate_due_date(vuln_data.get('severity')),
            'timeline': [
                {
                    'stage': 'DISCOVERED',
                    'timestamp': datetime.utcnow(),
                    'note': 'Vulnerability discovered'
                }
            ]
        }
        
        self.vulnerabilities[vuln_id] = vuln
        self.metrics['open_count'] += 1
        
        return vuln_id
    
    def transition_stage(self, vuln_id, new_stage, 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 from {old_stage} to {new_stage}")
        
        # 更新阶段
        vuln['stage'] = new_stage
        vuln['updated_at'] = datetime.utcnow()
        vuln['timeline'].append({
            'stage': new_stage,
            'timestamp': datetime.utcnow(),
            'note': note,
            'from_stage': old_stage
        })
        
        # 更新指标
        if new_stage == 'CLOSED':
            self.metrics['closed_count'] += 1
            self.metrics['open_count'] -= 1
            mttr = (datetime.utcnow() - vuln['created_at']).days
            self.metrics['mttr'].append(mttr)
        
        return True
    
    def is_valid_transition(self, from_stage, to_stage):
        """验证阶段转换是否合法"""
        valid_transitions = {
            'DISCOVERED': ['ASSESSING', 'ACCEPTED'],
            'ASSESSING': ['PLANNED', 'ACCEPTED', 'MITIGATED'],
            'PLANNED': ['IN_PROGRESS', 'ACCEPTED'],
            'IN_PROGRESS': ['VERIFIED', 'PLANNED'],
            'VERIFIED': ['CLOSED'],
            'MITIGATED': ['PLANNED', 'CLOSED', 'ACCEPTED'],
            'ACCEPTED': ['PLANNED'],  # 重新考虑修复
            'CLOSED': []  # 终态
        }
        
        return to_stage in valid_transitions.get(from_stage, [])
    
    def calculate_due_date(self, severity):
        """计算修复截止日期"""
        from datetime import timedelta
        
        sla = {
            'CRITICAL': 7,   # 7 天
            'HIGH': 30,      # 30 天
            'MEDIUM': 90,    # 90 天
            'LOW': 180       # 180 天
        }
        
        days = sla.get(severity, 90)
        return datetime.utcnow() + timedelta(days=days)
    
    def get_overdue_vulnerabilities(self):
        """获取逾期漏洞"""
        overdue = []
        now = datetime.utcnow()
        
        for vuln_id, vuln in self.vulnerabilities.items():
            if vuln['stage'] not in ['CLOSED', 'ACCEPTED']:
                if now > vuln['due_date']:
                    overdue.append({
                        'id': vuln_id,
                        'cve': vuln['cve'],
                        'severity': vuln['severity'],
                        'days_overdue': (now - vuln['due_date']).days
                    })
        
        return overdue
    
    def generate_metrics_report(self):
        """生成指标报告"""
        report = {
            'summary': {
                'total_vulnerabilities': len(self.vulnerabilities),
                'open': self.metrics['open_count'],
                'closed': self.metrics['closed_count'],
                'overdue': len(self.get_overdue_vulnerabilities())
            },
            'by_severity': self.count_by_severity(),
            'by_stage': self.count_by_stage(),
            'mttd_days': sum(self.metrics['mttd']) / len(self.metrics['mttd']) if self.metrics['mttd'] else 0,
            'mttr_days': sum(self.metrics['mttr']) / len(self.metrics['mttr']) if self.metrics['mttr'] else 0
        }
        
        return report
    
    def count_by_severity(self):
        """按严重程度统计"""
        counts = {'CRITICAL': 0, 'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}
        for vuln in self.vulnerabilities.values():
            if vuln['stage'] not in ['CLOSED', 'ACCEPTED']:
                severity = vuln['severity']
                if severity in counts:
                    counts[severity] += 1
        return counts
    
    def count_by_stage(self):
        """按阶段统计"""
        counts = {}
        for vuln in self.vulnerabilities.values():
            stage = vuln['stage']
            if stage not in counts:
                counts[stage] = 0
            counts[stage] += 1
        return counts
    
    def generate_vuln_id(self):
        """生成漏洞 ID"""
        import uuid
        return f"VULN-{uuid.uuid4().hex[:8].upper()}"
```

---

## 漏洞评估方法

### 风险评估模型

> ▎ 风险评估不是拍脑袋，是有模型的。模型不科学，优先级就是随意的。

**风险评估因素**：
```python
# 漏洞风险评估
class VulnerabilityRiskAssessment:
    """漏洞风险评估"""
    
    def __init__(self):
        # 评估因素权重
        self.weights = {
            'cvss_score': 0.30,      # CVSS 分数
            'exploitability': 0.25,  # 可利用性
            'exposure': 0.20,        # 暴露程度
            'business_impact': 0.15, # 业务影响
            'remediation_effort': 0.10  # 修复难度
        }
    
    def assess_risk(self, vulnerability: Dict, context: Dict) -> Dict:
        """评估漏洞风险"""
        scores = {
            'cvss': self.score_cvss(vulnerability),
            'exploitability': self.score_exploitability(vulnerability),
            'exposure': self.score_exposure(vulnerability, context),
            'business_impact': self.score_business_impact(vulnerability, context),
            'remediation_effort': self.score_remediation_effort(vulnerability)
        }
        
        # 计算加权风险分数
        risk_score = 0
        for factor, weight in self.weights.items():
            risk_score += scores[factor] * weight
        
        # 确定风险等级
        risk_level = self.get_risk_level(risk_score)
        
        return {
            'risk_score': risk_score,
            'risk_level': risk_level,
            'factor_scores': scores,
            'priority': self.get_priority(risk_level),
            'sla_days': self.get_sla(risk_level)
        }
    
    def score_cvss(self, vulnerability: Dict) -> float:
        """CVSS 分数评分 (0-100)"""
        cvss = vulnerability.get('cvss_score', 0)
        return cvss * 10  # 转换为 0-100
    
    def score_exploitability(self, vulnerability: Dict) -> float:
        """可利用性评分 (0-100)"""
        score = 50  # 默认
        
        # 有公开利用代码
        if vulnerability.get('exploit_available', False):
            score += 30
        
        # 在野利用
        if vulnerability.get('exploit_in_the_wild', False):
            score += 20
        
        # 有 PoC
        if vulnerability.get('poc_available', False):
            score += 15
        
        # 自动化攻击工具
        if vulnerability.get('automated_exploit', False):
            score += 10
        
        return min(score, 100)
    
    def score_exposure(self, vulnerability: Dict, context: Dict) -> float:
        """暴露程度评分 (0-100)"""
        score = 50  # 默认
        
        # 直接依赖 vs 传递依赖
        if vulnerability.get('direct_dependency', True):
            score += 20
        
        # 是否在生产环境使用
        if context.get('in_production', False):
            score += 20
        
        # 是否暴露给外部
        if context.get('externally_accessible', False):
            score += 20
        
        # 使用频率
        usage_frequency = context.get('usage_frequency', 'medium')
        if usage_frequency == 'high':
            score += 10
        
        return min(score, 100)
    
    def score_business_impact(self, vulnerability: Dict, context: Dict) -> float:
        """业务影响评分 (0-100)"""
        score = 50  # 默认
        
        # 影响核心业务
        if context.get('core_business', False):
            score += 25
        
        # 影响敏感数据
        if vulnerability.get('affects_confidentiality', False):
            score += 20
        
        # 合规影响
        if context.get('compliance_impact', False):
            score += 20
        
        # 声誉影响
        if context.get('reputation_impact', False):
            score += 15
        
        return min(score, 100)
    
    def score_remediation_effort(self, vulnerability: Dict) -> float:
        """修复难度评分 (0-100)"""
        # 注意：修复难度越高，优先级应该越高 (因为风险持续时间长)
        score = 50  # 默认
        
        # 有可用修复版本
        if vulnerability.get('fixed_version'):
            score -= 30  # 容易修复
        
        # 需要重大变更
        if vulnerability.get('breaking_changes', False):
            score += 25
        
        # 需要代码修改
        if vulnerability.get('requires_code_changes', False):
            score += 20
        
        # 测试复杂度高
        if vulnerability.get('complex_testing', False):
            score += 15
        
        return min(max(score, 0), 100)
    
    def get_risk_level(self, risk_score: float) -> str:
        """获取风险等级"""
        if risk_score >= 80:
            return 'CRITICAL'
        elif risk_score >= 60:
            return 'HIGH'
        elif risk_score >= 40:
            return 'MEDIUM'
        else:
            return 'LOW'
    
    def get_priority(self, risk_level: str) -> str:
        """获取优先级"""
        priority_map = {
            'CRITICAL': 'P0',
            'HIGH': 'P1',
            'MEDIUM': 'P2',
            'LOW': 'P3'
        }
        return priority_map.get(risk_level, 'P3')
    
    def get_sla(self, risk_level: str) -> int:
        """获取修复 SLA (天)"""
        sla_map = {
            'CRITICAL': 7,
            'HIGH': 30,
            'MEDIUM': 90,
            'LOW': 180
        }
        return sla_map.get(risk_level, 90)
```

### 上下文感知评估

```python
# 上下文感知评估
class ContextAwareAssessment:
    """上下文感知的漏洞评估"""
    
    def __init__(self):
        self.app_context = {}
    
    def set_application_context(self, context: Dict):
        """设置应用上下文"""
        self.app_context = {
            'environment': context.get('environment', 'development'),
            'exposure': context.get('exposure', 'internal'),
            'data_sensitivity': context.get('data_sensitivity', 'internal'),
            'traffic_volume': context.get('traffic_volume', 'low'),
            'compliance_requirements': context.get('compliance', []),
            'business_criticality': context.get('business_criticality', 'medium')
        }
    
    def assess_vulnerability(self, vulnerability: Dict) -> Dict:
        """基于上下文评估漏洞"""
        # 基础评分
        base_score = vulnerability.get('cvss_score', 0) * 10
        
        # 环境调整
        if self.app_context['environment'] == 'production':
            base_score *= 1.5
        elif self.app_context['environment'] == 'staging':
            base_score *= 1.2
        # development 不调整
        
        # 暴露程度调整
        if self.app_context['exposure'] == 'public':
            base_score *= 1.5
        elif self.app_context['exposure'] == 'partner':
            base_score *= 1.2
        # internal 不调整
        
        # 数据敏感度调整
        if self.app_context['data_sensitivity'] == 'pii':
            base_score *= 1.5
        elif self.app_context['data_sensitivity'] == 'confidential':
            base_score *= 1.3
        # internal 不调整
        
        # 合规要求调整
        if 'PCI-DSS' in self.app_context['compliance_requirements']:
            if vulnerability.get('affects_payment', False):
                base_score *= 1.5
        
        if 'HIPAA' in self.app_context['compliance_requirements']:
            if vulnerability.get('affects_health_data', False):
                base_score *= 1.5
        
        if 'GDPR' in self.app_context['compliance_requirements']:
            if vulnerability.get('affects_personal_data', False):
                base_score *= 1.3
        
        # 业务关键性调整
        criticality_multiplier = {
            'critical': 1.5,
            'high': 1.3,
            'medium': 1.0,
            'low': 0.8
        }
        base_score *= criticality_multiplier.get(
            self.app_context['business_criticality'], 1.0
        )
        
        # 限制在 0-100
        adjusted_score = min(max(base_score, 0), 100)
        
        return {
            'original_score': vulnerability.get('cvss_score', 0) * 10,
            'adjusted_score': adjusted_score,
            'risk_level': self.score_to_level(adjusted_score),
            'context_factors': self.app_context
        }
    
    def score_to_level(self, score: float) -> str:
        """分数转等级"""
        if score >= 80:
            return 'CRITICAL'
        elif score >= 60:
            return 'HIGH'
        elif score >= 40:
            return 'MEDIUM'
        else:
            return 'LOW'
```

---

## 修复策略与流程

### 修复策略

> ▎ 修复不是只有升级，是有多种策略的。策略不灵活，管理就是僵化的。

**修复策略选项**：
```python
# 修复策略管理
class RemediationStrategy:
    """修复策略管理"""
    
    STRATEGIES = {
        'UPGRADE': {
            'name': '升级修复',
            'description': '升级到修复版本',
            'effort': 'low',
            'risk': 'low',
            'recommended': True
        },
        'PATCH': {
            'name': '补丁修复',
            'description': '应用供应商补丁',
            'effort': 'low',
            'risk': 'low',
            'recommended': True
        },
        'WORKAROUND': {
            'name': '临时缓解',
            'description': '实施临时缓解措施',
            'effort': 'medium',
            'risk': 'medium',
            'recommended': False
        },
        'REPLACE': {
            'name': '替换组件',
            'description': '替换为替代组件',
            'effort': 'high',
            'risk': 'high',
            'recommended': False
        },
        'REMOVE': {
            'name': '移除依赖',
            'description': '移除有漏洞的依赖',
            'effort': 'medium',
            'risk': 'medium',
            'recommended': False
        },
        'ACCEPT': {
            'name': '接受风险',
            'description': '接受风险不修复',
            'effort': 'low',
            'risk': 'high',
            'recommended': False
        }
    }
    
    def select_strategy(self, vulnerability: Dict, context: Dict) -> Dict:
        """选择修复策略"""
        # 评估各策略的适用性
        strategy_scores = {}
        
        for strategy_id, strategy in self.STRATEGIES.items():
            score = self.evaluate_strategy(vulnerability, context, strategy)
            strategy_scores[strategy_id] = score
        
        # 选择最高分策略
        best_strategy = max(strategy_scores, key=strategy_scores.get)
        
        return {
            'recommended_strategy': best_strategy,
            'strategy_details': self.STRATEGIES[best_strategy],
            'all_scores': strategy_scores,
            'implementation_plan': self.create_implementation_plan(
                best_strategy, vulnerability, context
            )
        }
    
    def evaluate_strategy(self, vulnerability: Dict, context: Dict, 
                         strategy: Dict) -> float:
        """评估策略适用性"""
        score = 50  # 基础分
        
        strategy_id = list(self.STRATEGIES.keys())[
            list(self.STRATEGIES.values()).index(strategy)
        ]
        
        # 升级策略评估
        if strategy_id == 'UPGRADE':
            if vulnerability.get('fixed_version'):
                score += 30
            if not vulnerability.get('breaking_changes', False):
                score += 20
            if self.is_compatible_upgrade(vulnerability, context):
                score += 20
        
        # 补丁策略评估
        elif strategy_id == 'PATCH':
            if vulnerability.get('patch_available'):
                score += 30
            if vulnerability.get('vendor_supported'):
                score += 20
        
        # 临时缓解评估
        elif strategy_id == 'WORKAROUND':
            if vulnerability.get('workaround_available'):
                score += 25
            if context.get('urgent', False):
                score += 20  # 紧急情况优先临时方案
        
        # 替换策略评估
        elif strategy_id == 'REPLACE':
            if self.has_alternative(vulnerability):
                score += 20
            if vulnerability.get('abandoned', False):
                score += 30  # 已废弃组件优先替换
        
        # 移除策略评估
        elif strategy_id == 'REMOVE':
            if not vulnerability.get('critical_dependency', True):
                score += 30
            if context.get('unused', False):
                score += 40  # 未使用依赖优先移除
        
        # 接受风险评估
        elif strategy_id == 'ACCEPT':
            if vulnerability.get('severity') == 'LOW':
                score += 20
            if context.get('internal_only', False):
                score += 20
            if not vulnerability.get('exploit_available', False):
                score += 15
        
        return score
    
    def is_compatible_upgrade(self, vulnerability: Dict, context: Dict) -> bool:
        """检查升级是否兼容"""
        current_version = vulnerability.get('version', '0.0.0')
        fixed_version = vulnerability.get('fixed_version', '0.0.0')
        
        # 语义版本检查
        try:
            from packaging import version
            current = version.parse(current_version)
            fixed = version.parse(fixed_version)
            
            # 主版本不同可能有破坏性变更
            if current.major != fixed.major:
                return False
            
            return True
        except:
            return True
    
    def has_alternative(self, vulnerability: Dict) -> bool:
        """检查是否有替代组件"""
        alternatives = {
            'log4j': ['logback', 'slf4j'],
            'requests': ['httpx', 'urllib3'],
            'lodash': ['underscore', 'ramda'],
            'moment': ['dayjs', 'date-fns']
        }
        
        component = vulnerability.get('component', '').lower()
        return component in alternatives
    
    def create_implementation_plan(self, strategy_id: str, 
                                   vulnerability: Dict, context: Dict) -> Dict:
        """创建实施计划"""
        plan = {
            'strategy': strategy_id,
            'steps': [],
            'estimated_effort': 'unknown',
            'testing_required': True,
            'rollback_plan': True
        }
        
        if strategy_id == 'UPGRADE':
            plan['steps'] = [
                '1. 备份当前依赖配置',
                '2. 更新依赖版本',
                '3. 运行测试套件',
                '4. 验证功能正常',
                '5. 部署到生产',
                '6. 监控异常'
            ]
            plan['estimated_effort'] = '2-4 小时'
        
        elif strategy_id == 'WORKAROUND':
            plan['steps'] = [
                '1. 实施缓解措施',
                '2. 验证缓解有效',
                '3. 文档化临时方案',
                '4. 计划永久修复',
                '5. 设置修复期限'
            ]
            plan['estimated_effort'] = '1-2 天'
            plan['permanent_fix_required'] = True
        
        elif strategy_id == 'ACCEPT':
            plan['steps'] = [
                '1. 记录风险接受原因',
                '2. 获取管理层批准',
                '3. 实施补偿控制',
                '4. 设置审查日期',
                '5. 持续监控'
            ]
            plan['estimated_effort'] = '1 天'
            plan['approval_required'] = True
            plan['review_date'] = datetime.utcnow() + timedelta(days=90)
        
        return plan
```

### 修复流程

```python
# 修复流程管理
class RemediationWorkflow:
    """修复流程管理"""
    
    def __init__(self):
        self.workflow_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_remediation_ticket(self, vulnerability: Dict) -> Dict:
        """创建修复工单"""
        ticket = {
            'id': self.generate_ticket_id(),
            'vulnerability_id': vulnerability.get('id'),
            'cve': vulnerability.get('cve'),
            'component': vulnerability.get('component'),
            'current_version': vulnerability.get('version'),
            'fixed_version': vulnerability.get('fixed_version'),
            'severity': vulnerability.get('severity'),
            'state': 'OPEN',
            'assignee': None,
            'created_at': datetime.utcnow(),
            'due_date': self.calculate_due_date(vulnerability.get('severity')),
            'steps': [],
            'comments': []
        }
        
        return ticket
    
    def transition_ticket(self, ticket_id: str, new_state: str, 
                         comment: str = '') -> bool:
        """转换工单状态"""
        ticket = self.get_ticket(ticket_id)
        if not ticket:
            return False
        
        current_state = ticket['state']
        
        # 验证状态转换
        if new_state not in self.workflow_states.get(current_state, []):
            raise ValueError(f"Invalid state transition: {current_state} -> {new_state}")
        
        # 更新状态
        ticket['state'] = new_state
        ticket['updated_at'] = datetime.utcnow()
        ticket['comments'].append({
            'timestamp': datetime.utcnow(),
            'comment': f"State changed from {current_state} to {new_state}: {comment}"
        })
        
        return True
    
    def add_step(self, ticket_id: str, step: Dict) -> bool:
        """添加修复步骤"""
        ticket = self.get_ticket(ticket_id)
        if not ticket:
            return False
        
        step['created_at'] = datetime.utcnow()
        step['status'] = 'pending'
        ticket['steps'].append(step)
        
        return True
    
    def complete_step(self, ticket_id: str, step_index: int, 
                     result: str) -> bool:
        """完成修复步骤"""
        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_ticket(self, ticket_id: str) -> Dict:
        """获取工单"""
        # 实际实现从数据库获取
        pass
    
    def generate_ticket_id(self) -> str:
        """生成工单 ID"""
        import uuid
        return f"REM-{uuid.uuid4().hex[:8].upper()}"
    
    def calculate_due_date(self, severity: str) -> datetime:
        """计算截止日期"""
        from datetime import timedelta
        
        sla = {
            'CRITICAL': 7,
            'HIGH': 30,
            'MEDIUM': 90,
            'LOW': 180
        }
        
        days = sla.get(severity, 90)
        return datetime.utcnow() + timedelta(days=days)
    
    def get_overdue_tickets(self) -> List[Dict]:
        """获取逾期工单"""
        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:
    """应急响应管理"""
    
    def __init__(self):
        # 紧急漏洞标准
        self.emergency_criteria = {
            'active_exploitation': True,      # 在野利用
            'critical_severity': True,         # 严重级别
            'wide_impact': True,               # 影响广泛
            'no_workaround': True              # 无缓解措施
        }
        
        # 响应团队
        self.response_team = {
            'incident_commander': None,
            'security_lead': None,
            'engineering_lead': None,
            'communications_lead': None
        }
        
        # 响应流程
        self.response_phases = [
            'DETECTION',    # 检测
            'ASSESSMENT',   # 评估
            'CONTAINMENT',  # 遏制
            'REMEDIATION',  # 修复
            'RECOVERY',     # 恢复
            'LESSONS'       # 总结
        ]
    
    def declare_emergency(self, vulnerability: Dict) -> Dict:
        """宣布紧急状态"""
        # 检查是否符合紧急标准
        if not self.is_emergency(vulnerability):
            raise ValueError("Vulnerability does not meet emergency criteria")
        
        # 创建应急响应记录
        emergency = {
            'id': self.generate_emergency_id(),
            'vulnerability': vulnerability,
            'declared_at': datetime.utcnow(),
            'phase': 'DETECTION',
            'team': self.response_team.copy(),
            'timeline': [
                {
                    'phase': 'DETECTION',
                    'timestamp': datetime.utcnow(),
                    'action': 'Emergency declared'
                }
            ],
            'actions': [],
            'communications': []
        }
        
        # 通知响应团队
        self.notify_team(emergency)
        
        return emergency
    
    def is_emergency(self, vulnerability: Dict) -> bool:
        """判断是否为紧急漏洞"""
        # 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: str, new_phase: str, 
                     actions: List[str]) -> bool:
        """推进响应阶段"""
        emergency = self.get_emergency(emergency_id)
        if not emergency:
            return False
        
        current_phase = emergency['phase']
        current_phase_index = self.response_phases.index(current_phase)
        new_phase_index = self.response_phases.index(new_phase)
        
        # 只能推进到下一阶段
        if new_phase_index != current_phase_index + 1:
            raise ValueError(f"Invalid phase transition: {current_phase} -> {new_phase}")
        
        # 更新阶段
        emergency['phase'] = new_phase
        emergency['timeline'].append({
            'phase': new_phase,
            'timestamp': datetime.utcnow(),
            'action': f"Advanced to {new_phase} phase"
        })
        
        # 记录行动
        for action in actions:
            emergency['actions'].append({
                'phase': new_phase,
                'action': action,
                'timestamp': datetime.utcnow()
            })
        
        return True
    
    def create_communication(self, emergency_id: str, audience: str, 
                            message: str) -> Dict:
        """创建沟通记录"""
        emergency = self.get_emergency(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: str) -> str:
        """获取沟通渠道"""
        channels = {
            'internal_team': 'slack',
            'engineering': 'email',
            'management': 'email',
            'customers': 'status_page',
            'public': 'blog'
        }
        return channels.get(audience, 'email')
    
    def generate_emergency_id(self) -> str:
        """生成应急 ID"""
        import uuid
        return f"EMG-{uuid.uuid4().hex[:8].upper()}"
    
    def generate_comm_id(self) -> str:
        """生成沟通 ID"""
        import uuid
        return f"COMM-{uuid.uuid4().hex[:6].upper()}"
    
    def notify_team(self, emergency: Dict):
        """通知响应团队"""
        # 实际实现发送通知
        pass
    
    def get_emergency(self, emergency_id: str) -> Dict:
        """获取应急记录"""
        # 实际实现从数据库获取
        pass
    
    def generate_post_mortem(self, emergency_id: str) -> Dict:
        """生成事后报告"""
        emergency = self.get_emergency(emergency_id)
        if not emergency:
            return None
        
        post_mortem = {
            'emergency_id': emergency_id,
            'vulnerability': emergency['vulnerability'],
            'timeline': emergency['timeline'],
            'actions_taken': emergency['actions'],
            'communications': emergency['communications'],
            'what_went_well': [],
            'what_went_wrong': [],
            'improvements': [],
            'action_items': []
        }
        
        return post_mortem
```

---

## 供应商风险管理

### 供应商评估

> ▎ 供应商不是选了就用，是持续评估。评估不充分，风险就是隐藏的。

**供应商风险评估**：
```python
# 供应商风险管理
class VendorRiskManagement:
    """供应商风险管理"""
    
    def __init__(self):
        # 评估维度
        self.assessment_dimensions = {
            'security_practices': 0.25,    # 安全实践
            'vulnerability_response': 0.25, # 漏洞响应
            'transparency': 0.20,          # 透明度
            'maintenance': 0.15,           # 维护情况
            'community': 0.15              # 社区健康
        }
    
    def assess_vendor(self, vendor_info: Dict) -> Dict:
        """评估供应商风险"""
        scores = {
            'security_practices': self.assess_security_practices(vendor_info),
            'vulnerability_response': self.assess_vulnerability_response(vendor_info),
            'transparency': self.assess_transparency(vendor_info),
            'maintenance': self.assess_maintenance(vendor_info),
            'community': self.assess_community(vendor_info)
        }
        
        # 计算加权总分
        total_score = 0
        for dimension, weight in self.assessment_dimensions.items():
            total_score += scores[dimension] * weight
        
        # 确定风险等级
        risk_level = self.get_risk_level(total_score)
        
        return {
            'vendor': vendor_info.get('name'),
            'total_score': total_score,
            'risk_level': risk_level,
            'dimension_scores': scores,
            'recommendations': self.get_recommendations(scores)
        }
    
    def assess_security_practices(self, vendor_info: Dict) -> float:
        """评估安全实践"""
        score = 50  # 基础分
        
        # 有安全团队
        if vendor_info.get('has_security_team'):
            score += 20
        
        # 有安全政策
        if vendor_info.get('security_policy'):
            score += 15
        
        # 有漏洞披露政策
        if vendor_info.get('vulnerability_disclosure_policy'):
            score += 15
        
        # 有安全审计
        if vendor_info.get('security_audit'):
            score += 10
        
        # 有 bug bounty
        if vendor_info.get('bug_bounty'):
            score += 10
        
        return min(score, 100)
    
    def assess_vulnerability_response(self, vendor_info: Dict) -> float:
        """评估漏洞响应"""
        score = 50  # 基础分
        
        # 响应时间快
        avg_response_days = vendor_info.get('avg_response_days', 30)
        if avg_response_days <= 7:
            score += 30
        elif avg_response_days <= 30:
            score += 20
        elif avg_response_days <= 90:
            score += 10
        
        # 修复时间快
        avg_fix_days = vendor_info.get('avg_fix_days', 90)
        if avg_fix_days <= 30:
            score += 20
        elif avg_fix_days <= 90:
            score += 10
        
        # 有 CVE 分配
        if vendor_info.get('cve_assignment'):
            score += 15
        
        # 有安全公告
        if vendor_info.get('security_advisories'):
            score += 15
        
        return min(score, 100)
    
    def assess_transparency(self, vendor_info: Dict) -> float:
        """评估透明度"""
        score = 50  # 基础分
        
        # 公开源代码
        if vendor_info.get('open_source'):
            score += 20
        
        # 有变更日志
        if vendor_info.get('changelog'):
            score += 15
        
        # 有安全文档
        if vendor_info.get('security_documentation'):
            score += 15
        
        # 公开已知问题
        if vendor_info.get('public_known_issues'):
            score += 20
        
        return min(score, 100)
    
    def assess_maintenance(self, vendor_info: Dict) -> float:
        """评估维护情况"""
        score = 50  # 基础分
        
        # 最近有更新
        last_update_days = vendor_info.get('last_update_days', 365)
        if last_update_days <= 30:
            score += 25
        elif last_update_days <= 90:
            score += 20
        elif last_update_days <= 180:
            score += 15
        elif last_update_days <= 365:
            score += 10
        
        # 版本发布规律
        if vendor_info.get('regular_releases'):
            score += 15
        
        # 支持多版本
        if vendor_info.get('multi_version_support'):
            score += 15
        
        # LTS 版本
        if vendor_info.get('lts_version'):
            score += 10
        
        return min(score, 100)
    
    def assess_community(self, vendor_info: Dict) -> float:
        """评估社区健康"""
        score = 50  # 基础分
        
        # 贡献者数量
        contributors = vendor_info.get('contributors', 0)
        if contributors >= 100:
            score += 20
        elif contributors >= 50:
            score += 15
        elif contributors >= 10:
            score += 10
        
        # 下载量
        downloads = vendor_info.get('downloads', 0)
        if downloads >= 1000000:
            score += 15
        elif downloads >= 100000:
            score += 10
        
        # 问题响应
        issue_response_rate = vendor_info.get('issue_response_rate', 0)
        if issue_response_rate >= 0.8:
            score += 15
        elif issue_response_rate >= 0.5:
            score += 10
        
        return min(score, 100)
    
    def get_risk_level(self, score: float) -> str:
        """获取风险等级"""
        if score >= 80:
            return 'LOW'
        elif score >= 60:
            return 'MEDIUM'
        elif score >= 40:
            return 'HIGH'
        else:
            return 'CRITICAL'
    
    def get_recommendations(self, scores: Dict) -> List[str]:
        """获取改进建议"""
        recommendations = []
        
        if scores['security_practices'] < 60:
            recommendations.append('建议供应商建立安全团队和安全政策')
        
        if scores['vulnerability_response'] < 60:
            recommendations.append('建议供应商改进漏洞响应流程')
        
        if scores['transparency'] < 60:
            recommendations.append('建议供应商提高透明度')
        
        if scores['maintenance'] < 60:
            recommendations.append('建议评估供应商维护能力，考虑替代方案')
        
        if scores['community'] < 60:
            recommendations.append('建议评估社区健康度，关注项目可持续性')
        
        return recommendations
```

---

## 自动化管理实践

### 自动化工作流

```yaml
# GitHub Actions 自动化工作流
# .github/workflows/vulnerability-management.yml

name: Vulnerability Management

on:
  schedule:
    - cron: '0 2 * * *'  # 每天凌晨 2 点
  push:
    paths:
      - '**/requirements*.txt'
      - '**/package*.json'
      - '**/pom.xml'
      - '**/go.mod'

jobs:
  scan:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Run SCA Scan
      run: |
        pip install safety snyk
        safety check --json --output safety-report.json
        snyk test --json --output snyk-report.json
    
    - name: Process Findings
      run: |
        python scripts/process_vulnerabilities.py
    
    - name: Create Issues
      if: ${{ env.NEW_VULNERABILITIES > 0 }}
      uses: actions/github-script@v6
      with:
        script: |
          const vulns = require('./new-vulnerabilities.json');
          for (const vuln of vulns) {
            await github.rest.issues.create({
              owner: context.repo.owner,
              repo: context.repo.repo,
              title: `Security: ${vuln.cve} in ${vuln.component}`,
              body: `
## Vulnerability Details
- CVE: ${vuln.cve}
- Component: ${vuln.component}
- Current Version: ${vuln.current_version}
- Fixed Version: ${vuln.fixed_version}
- Severity: ${vuln.severity}

## Recommended Action
Update to version ${vuln.fixed_version} or later.

## References
- ${vuln.references?.[0] || 'N/A'}
              `,
              labels: ['security', 'vulnerability', vuln.severity.toLowerCase()]
            });
          }
    
    - name: Notify Security Team
      if: ${{ env.CRITICAL_VULNERABILITIES > 0 }}
      run: |
        curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
          -H 'Content-Type: application/json' \
          -d '{
            "text": " Critical vulnerabilities detected!",
            "attachments": [{
              "color": "danger",
              "fields": [{
                "title": "Count",
                "value": "${{ env.CRITICAL_VULNERABILITIES }}",
                "short": true
              }]
            }]
          }'

auto-fix:
    runs-on: ubuntu-latest
    needs: scan
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Run Dependabot
      uses: dependabot/fetch-metadata@v1
      with:
        github-token: ${{ secrets.GITHUB_TOKEN }}
    
    - name: Auto-merge Security Updates
      uses: pascalgn/automerge-action@v0.15.0
      env:
        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        MERGE_METHOD: squash
        MERGE_LABELS: 'security,dependencies'
```

---

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

---

## 总结与思考

### 核心要点回顾

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

**依赖漏洞管理框架**：
```
1. 生命周期管理
   - 发现 → 评估 → 修复 → 验证
   - 阶段转换
   - 指标跟踪

2. 风险评估
   - CVSS 基础
   - 上下文感知
   - 业务影响

3. 修复策略
   - 升级/补丁
   - 缓解/替换
   - 接受风险

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

**关键成功因素**：
```
1. 自动化
   - 自动扫描
   - 自动工单
   - 自动通知

2. 流程化
   - 明确 SLA
   - 阶段管理
   - 责任到人

3. 持续化
   - 定期扫描
   - 趋势分析
   - 持续改进
```

### 实战建议

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

**管理建设**：
```
1. 工具选型
   - SCA 工具
   - 工单系统
   - 通知渠道

2. 流程建立
   - 修复 SLA
   - 应急流程
   - 审批流程

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

---

## 参考资料

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

- OWASP Dependency Management
  https://owasp.org/www-project-dependency-check/

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

### 工具资源
```
- Snyk
  https://snyk.io/

- Dependabot
  https://github.com/dependabot

- OWASP Dependency-Check
  https://owasp.org/www-project-dependency-check/
```

### 漏洞数据库
```
- NVD
  https://nvd.nist.gov/

- OSV
  https://osv.dev/

- GitHub Advisory Database
  https://github.com/advisories
```

### 书籍推荐
```
- 《Vulnerability Management Best Practices》
- 《Dependency Management for Security》
- 《Security Operations Center》
```

---

**标记 明日预告**：Day 153 - 安全测试自动化

> ▎ 依赖管理是运营实践，测试自动化是效率提升——明天看安全测试自动化。

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

---

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

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