Day-133-代码审计方法论

# Day 147: 代码审计方法论

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

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 代码审计方法论 - Day 147           │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ 代码审计不是随便看看，是系统分析。分析不系统，漏洞就是遗漏的。今天系统学习代码审计方法论。

---

## 清单 目录

1. [代码审计概述](#代码审计概述)
2. [审计方法论](#审计方法论)
3. [静态分析技术](#静态分析技术)
4. [动态分析技术](#动态分析技术)
5. [污点分析技术](#污点分析技术)
6. [漏洞模式识别](#漏洞模式识别)
7. [审计工具链](#审计工具链)
8. [审计报告规范](#审计报告规范)
9. [实战审计演练](#实战审计演练)
10. [总结与思考](#总结与思考)
11. [参考资料](#参考资料)

---

## 代码审计概述

### 什么是代码审计

> ▎ 代码审计不是找茬，是找漏洞。漏洞不找出来，上线就是给别人送钱。

**定义与目标**：
```
代码审计 (Code Audit) 是对源代码进行系统性安全分析的过程，目的是发现潜在的安全漏洞、逻辑错误和设计缺陷。

核心目标：
1. 发现漏洞
   - 已知漏洞模式
   - 业务逻辑漏洞
   - 设计缺陷

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

3. 提供建议
   - 修复方案
   - 最佳实践
   - 安全改进
```

**审计类型对比**：
```
┌─────────────────┬──────────────┬──────────────┬──────────────┐
│     类型        │   深度       │   覆盖率     │   时间成本   │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│   自动化扫描    │  浅          │  高          │  低          │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│   人工审计      │  深          │  中          │  高          │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│   混合审计      │  中深        │  高          │  中          │
├─────────────────┼──────────────┼──────────────┼──────────────┤
│   渗透测试      │  实战        │  低          │  中          │
└─────────────────┴──────────────┴──────────────┴──────────────┘

最佳实践：自动化扫描 + 人工审计 + 渗透测试
```

### 审计流程

**标准审计流程**：
```
阶段 1: 准备 (10%)
├── 了解项目背景
├── 确定审计范围
├── 准备审计工具
└── 建立沟通机制

阶段 2: 自动化扫描 (20%)
├── SAST 扫描
├── 依赖扫描
├── 配置扫描
└── 结果整理

阶段 3: 人工审计 (50%)
├── 架构分析
├── 关键代码审查
├── 漏洞挖掘
└── 问题验证

阶段 4: 报告与修复 (15%)
├── 报告编写
├── 客户沟通
├── 修复验证
└── 最终报告

阶段 5: 后续跟进 (5%)
├── 上线支持
├── 持续监控
└── 知识沉淀
```

---

## 审计方法论

### 审计策略

> ▎ 审计不是随机看，是有策略的。策略不明确，审计就是盲目的。

**审计策略选择**：
```python
# 审计策略模型
class AuditStrategy:
    def __init__(self):
        self.strategies = {
            'risk_based': self.risk_based_audit,      # 基于风险
            'function_based': self.function_based,    # 基于功能
            'data_flow_based': self.data_flow_based,  # 基于数据流
            'comprehensive': self.comprehensive       # 全面审计
        }
    
    def select_strategy(self, project_context):
        """选择审计策略"""
        # 高风险项目 → 基于风险
        if project_context.get('risk_level') == 'high':
            return 'risk_based'
        
        # 核心功能 → 基于功能
        if project_context.get('critical_functions'):
            return 'function_based'
        
        # 数据处理复杂 → 基于数据流
        if project_context.get('complex_data_flow'):
            return 'data_flow_based'
        
        # 默认全面审计
        return 'comprehensive'
    
    def risk_based_audit(self, codebase):
        """
        基于风险的审计
        
        优先审计高风险区域：
        1. 认证授权代码
        2. 支付相关代码
        3. 数据处理代码
        4. 外部接口代码
        """
        high_risk_areas = [
            'authentication',
            'authorization',
            'payment',
            'data_processing',
            'external_api'
        ]
        
        # 优先审计高风险区域
        for area in high_risk_areas:
            self.audit_area(codebase, area)
    
    def function_based(self, codebase):
        """
        基于功能的审计
        
        按功能模块审计：
        1. 用户管理
        2. 订单处理
        3. 支付流程
        4. 数据导出
        """
        critical_functions = [
            'user_registration',
            'user_login',
            'password_reset',
            'order_create',
            'payment_process',
            'data_export'
        ]
        
        for func in critical_functions:
            self.audit_function(codebase, func)
    
    def data_flow_based(self, codebase):
        """
        基于数据流的审计
        
        追踪敏感数据流：
        1. 用户输入 → 处理 → 存储
        2. 数据库 → 处理 → 输出
        3. 外部 API → 处理 → 内部使用
        """
        sensitive_data_types = [
            'user_input',
            'credentials',
            'payment_info',
            'personal_data',
            'file_upload'
        ]
        
        for data_type in sensitive_data_types:
            self.trace_data_flow(codebase, data_type)
```

### 审计深度

**审计深度分级**：
```
Level 1: 表面审计
├── 自动化扫描
├── 明显漏洞
├── 配置问题
└── 时间：1-2 天

Level 2: 标准审计
├── 自动化 + 人工
├── 常见漏洞
├── 逻辑问题
└── 时间：1-2 周

Level 3: 深度审计
├── 全面人工审计
├── 复杂漏洞
├── 业务逻辑
└── 时间：2-4 周

Level 4: 专家审计
├── 专家团队
├── 0day 挖掘
├── 架构问题
└── 时间：4-8 周
```

---

## 静态分析技术

### SAST 原理

> ▎ 静态分析不是万能，但没有是万万不能。工具不跑，漏洞就是隐藏的。

**SAST 工作流程**：
```python
# SAST 工作原理
class SASTEngine:
    """静态分析引擎"""
    
    def analyze(self, codebase):
        """执行静态分析"""
        # 1. 词法分析
        tokens = self.lexical_analysis(codebase)
        
        # 2. 语法分析
        ast = self.syntax_analysis(tokens)
        
        # 3. 控制流分析
        cfg = self.control_flow_analysis(ast)
        
        # 4. 数据流分析
        dfg = self.data_flow_analysis(cfg)
        
        # 5. 模式匹配
        issues = self.pattern_matching(ast, cfg, dfg)
        
        # 6. 污点分析
        taint_issues = self.taint_analysis(dfg)
        
        # 7. 结果汇总
        return self.aggregate_results(issues, taint_issues)
    
    def pattern_matching(self, ast, cfg, dfg):
        """模式匹配"""
        issues = []
        
        # 预定义漏洞模式
        vulnerability_patterns = [
            {
                'name': 'SQL Injection',
                'pattern': 'execute.*%s.*%',
                'severity': 'HIGH'
            },
            {
                'name': 'Command Injection',
                'pattern': 'os.system.*%',
                'severity': 'CRITICAL'
            },
            {
                'name': 'XSS',
                'pattern': 'render.*without escaping',
                'severity': 'MEDIUM'
            }
        ]
        
        for pattern in vulnerability_patterns:
            matches = self.search_pattern(ast, pattern['pattern'])
            for match in matches:
                issues.append({
                    'type': pattern['name'],
                    'location': match.location,
                    'severity': pattern['severity'],
                    'code': match.code
                })
        
        return issues
```

### 工具使用

**主流 SAST 工具**：
```python
# Bandit (Python)
"""
# 安装
pip install bandit

# 使用
bandit -r ./src -f json -o report.json

# 配置文件 (.bandit)
[bandit]
targets = src
exclude_dirs = tests,venv
skips = B101,B306
"""

# Semgrep (多语言)
"""
# 安装
pip install semgrep

# 使用
semgrep --config auto ./src

# 自定义规则
# .semgrep/rules.yml
rules:
  - id: sql-injection
    patterns:
      - pattern: db.execute("...%s..." % ...)
    message: "Possible SQL injection"
    severity: ERROR
    languages: [python]
"""

# SonarQube (企业级)
"""
# Docker 运行
docker run -d --name sonarqube -p 9000:9000 sonarqube

# 扫描
sonar-scanner \
  -Dsonar.projectKey=myproject \
  -Dsonar.sources=./src \
  -Dsonar.host.url=http://localhost:9000
"""

# CodeQL (GitHub)
"""
# 创建数据库
codeql database create ./db --language=python

# 执行查询
codeql database analyze ./db ./queries.ql --format=sarif
"""
```

---

## 动态分析技术

### DAST 原理

> ▎ 动态分析不是替代静态，是互补。单一分析，漏洞就是片面的。

**DAST 工作流程**：
```
1. 爬虫发现
   ├── 爬取应用
   ├── 发现端点
   └── 构建站点地图

2. 漏洞扫描
   ├── SQL 注入测试
   ├── XSS 测试
   ├── CSRF 测试
   └── 其他 OWASP Top 10

3. 认证测试
   ├── 暴力破解
   ├── 会话管理
   └── 权限测试

4. 结果验证
   ├── 误报过滤
   ├── 风险评级
   └── 报告生成
```

**工具使用**：
```python
# OWASP ZAP
"""
# Docker 运行
docker run -t owasp/zap2docker-stable zap-baseline.py \
  -t https://example.com

# 自动化扫描
zap-cli quick-scan --self-contained --scanners xss,sqli https://example.com

# API 扫描
zap-cli open-url https://example.com
zap-cli active-scan --scanners all https://example.com
zap-cli report -o report.html -f html
"""

# Burp Suite
"""
# 社区版：手动测试
# 专业版：自动化扫描

# 配置扫描
1. 设置扫描范围
2. 配置认证
3. 选择扫描策略
4. 执行扫描
5. 审查结果
"""

# SQLMap (SQL 注入)
"""
# 基本扫描
sqlmap -u "https://example.com/page?id=1"

# 深度扫描
sqlmap -u "https://example.com/page?id=1" --level=5 --risk=3

# 自动利用
sqlmap -u "https://example.com/page?id=1" --auto
"""
```

### Fuzzing 测试

```python
# 模糊测试
import requests
import random
import string

class FuzzTester:
    """模糊测试工具"""
    
    def __init__(self, target_url):
        self.target_url = target_url
        self.session = requests.Session()
    
    def fuzz_parameter(self, param_name, payloads):
        """模糊测试参数"""
        results = []
        
        for payload in payloads:
            try:
                response = self.session.get(
                    self.target_url,
                    params={param_name: payload},
                    timeout=5
                )
                
                # 检测异常
                if self.detect_anomaly(response):
                    results.append({
                        'payload': payload,
                        'status_code': response.status_code,
                        'response_size': len(response.content),
                        'anomaly': True
                    })
                
            except requests.exceptions.RequestException as e:
                results.append({
                    'payload': payload,
                    'error': str(e),
                    'anomaly': True
                })
        
        return results
    
    def get_sql_injection_payloads(self):
        """SQL 注入 payload 列表"""
        return [
            "' OR '1'='1",
            "1; DROP TABLE users--",
            "' UNION SELECT NULL--",
            "1' AND '1'='1",
            "' OR 1=1--",
            "admin'--",
            "1; WAITFOR DELAY '0:0:5'--",
        ]
    
    def get_xss_payloads(self):
        """XSS payload 列表"""
        return [
            "<script>alert(1)</script>",
            "<img src=x onerror=alert(1)>",
            "javascript:alert(1)",
            "<svg onload=alert(1)>",
            "'-alert(1)-'",
        ]
    
    def detect_anomaly(self, response):
        """检测异常响应"""
        # SQL 注入指示
        sql_errors = [
            'SQL syntax', 'mysql_fetch', 'ORA-', 
            'PostgreSQL', 'SQLite', 'syntax error'
        ]
        
        for error in sql_errors:
            if error.lower() in response.text.lower():
                return True
        
        # XSS 指示
        if '<script>alert(1)</script>' in response.text:
            return True
        
        # 服务器错误
        if response.status_code >= 500:
            return True
        
        return False

# 使用示例
fuzzer = FuzzTester('https://example.com/search')
results = fuzzer.fuzz_parameter('q', fuzzer.get_sql_injection_payloads())

for result in results:
    if result.get('anomaly'):
        print(f"Potential vulnerability: {result['payload']}")
```

---

## 污点分析技术

### 污点分析原理

> ▎ 污点分析不是跟踪数据，是跟踪风险。风险不跟踪，注入就是遗漏的。

**污点分析流程**：
```
1. 污点源识别 (Sources)
   ├── 用户输入
   ├── 外部 API
   ├── 文件读取
   └── 网络数据

2. 污点传播 (Propagation)
   ├── 变量赋值
   ├── 函数调用
   ├── 字符串操作
   └── 数据转换

3. 污点汇聚 (Sinks)
   ├── SQL 执行
   ├── 命令执行
   ├── 文件操作
   └── 输出渲染

4. 路径分析
   ├── 可达性分析
   ├── 条件分析
   └──  sanitization 检查
```

**污点分析实现**：
```python
# 简化污点分析示例
class TaintAnalyzer:
    """污点分析器"""
    
    def __init__(self):
        # 污点源
        self.sources = {
            'user_input': [
                'request.args',
                'request.form',
                'request.json',
                'input()',
                'sys.argv'
            ],
            'file_read': [
                'open().read',
                'file.read'
            ],
            'network': [
                'requests.get',
                'urllib.urlopen',
                'socket.recv'
            ]
        }
        
        # 污点汇聚点
        self.sinks = {
            'sql': [
                'execute(',
                'exec(',
                'query('
            ],
            'command': [
                'os.system(',
                'subprocess.run(',
                'eval(',
                'exec('
            ],
            'file': [
                'open(',
                'os.remove(',
                'shutil.'
            ],
            'output': [
                'render(',
                'write(',
                'print('
            ]
        }
        
        # 清理函数
        self.sanitizers = [
            'escape',
            'sanitize',
            'quote',
            'parameterize'
        ]
    
    def analyze(self, code):
        """执行污点分析"""
        issues = []
        
        # 1. 识别污点源
        tainted_vars = self.find_sources(code)
        
        # 2. 跟踪污点传播
        propagation_paths = self.track_propagation(code, tainted_vars)
        
        # 3. 检查是否到达汇聚点
        for path in propagation_paths:
            if self.reaches_sink(path):
                # 4. 检查是否有清理
                if not self.has_sanitization(path):
                    issues.append({
                        'type': 'Taint Flow',
                        'source': path['source'],
                        'sink': path['sink'],
                        'path': path,
                        'severity': 'HIGH'
                    })
        
        return issues
    
    def find_sources(self, code):
        """识别污点源"""
        tainted = {}
        
        for source_type, patterns in self.sources.items():
            for pattern in patterns:
                if pattern in code:
                    # 找到污点源
                    var_name = self.extract_variable(code, pattern)
                    tainted[var_name] = {
                        'type': source_type,
                        'location': self.find_location(code, pattern)
                    }
        
        return tainted
    
    def track_propagation(self, code, tainted_vars):
        """跟踪污点传播"""
        paths = []
        
        # 简化实现
        # 实际应构建数据流图
        for var_name, source_info in tainted_vars.items():
            # 查找变量使用
            uses = self.find_variable_uses(code, var_name)
            
            for use in uses:
                paths.append({
                    'source': source_info,
                    'use': use,
                    'variable': var_name
                })
        
        return paths
    
    def reaches_sink(self, path):
        """检查是否到达汇聚点"""
        use_location = path.get('use', {})
        code_context = use_location.get('code', '')
        
        for sink_type, patterns in self.sinks.items():
            for pattern in patterns:
                if pattern in code_context:
                    path['sink'] = {
                        'type': sink_type,
                        'location': use_location.get('location')
                    }
                    return True
        
        return False
    
    def has_sanitization(self, path):
        """检查是否有清理"""
        # 检查路径中是否有清理函数
        for sanitizer in self.sanitizers:
            if sanitizer in str(path):
                return True
        
        return False
```

---

## 漏洞模式识别

### 常见漏洞模式

> ▎ 漏洞不是随机出现，是有模式的。模式不认识，审计就是盲目的。

**SQL 注入模式**：
```python
# SQL 注入代码模式
sql_injection_patterns = [
    # 字符串拼接
    {
        'pattern': r'execute\(["\'].*%s.*["\'].*%',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'cursor.execute("SELECT * FROM users WHERE id = %s" % user_id)'
    },
    # f-string
    {
        'pattern': r'execute\(f["\'].*\{.*\}.*["\']\)',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")'
    },
    # format
    {
        'pattern': r'execute\(["\'].*\{.*\}.*["\']\.format\(',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'cursor.execute("SELECT * FROM users WHERE id = {}".format(user_id))'
    },
    # 拼接
    {
        'pattern': r'execute\(["\'].*["\'].*\+.*\)',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'cursor.execute("SELECT * FROM users WHERE id = " + user_id)'
    }
]

# 安全替代方案
safe_patterns = [
    # 参数化查询
    {
        'pattern': r'execute\(["\'].*["\'].*,.*\(',
        'description': '使用参数化查询',
        'example': 'cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))'
    }
]
```

**XSS 模式**：
```python
# XSS 代码模式
xss_patterns = [
    # 未转义输出
    {
        'pattern': r'render_template_string\(.*\{.*\}.*\)',
        'language': 'python',
        'severity': 'HIGH',
        'example': 'render_template_string("<h1>{{ user_input }}</h1>")'
    },
    # 直接返回用户输入
    {
        'pattern': r'return.*request\.(args|form|json)\[.*\]',
        'language': 'python',
        'severity': 'HIGH',
        'example': 'return request.args.get("q")'
    },
    # 未转义 HTML
    {
        'pattern': r'Markup\(.*request\.',
        'language': 'python',
        'severity': 'HIGH',
        'example': 'return Markup(request.args.get("content"))'
    }
]

# 安全替代方案
safe_xss_patterns = [
    {
        'pattern': r'escape\(.*request\.',
        'description': '使用 escape 转义',
        'example': 'return escape(request.args.get("q"))'
    },
    {
        'pattern': r'render_template\(.*\)',
        'description': '使用模板自动转义',
        'example': 'return render_template("page.html", user_input=user_input)'
    }
]
```

**命令注入模式**：
```python
# 命令注入代码模式
command_injection_patterns = [
    # shell=True
    {
        'pattern': r'subprocess\..*\(.*shell\s*=\s*True.*\)',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'subprocess.run(cmd, shell=True)'
    },
    # os.system
    {
        'pattern': r'os\.system\(.*\+.*\)',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'os.system("ping " + hostname)'
    },
    # eval/exec
    {
        'pattern': r'eval\(.*request\.',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'eval(request.args.get("expr"))'
    },
    {
        'pattern': r'exec\(.*request\.',
        'language': 'python',
        'severity': 'CRITICAL',
        'example': 'exec(request.args.get("code"))'
    }
]
```

---

## 审计工具链

### 工具集成

> ▎ 工具不是单个用，是集成用。工具不集成，效率就是低下的。

**工具链配置**：
```yaml
# .github/workflows/security-scan.yml
name: Security Scan

on: [push, pull_request]

jobs:
  security:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    # SAST - Bandit
    - name: Run Bandit
      run: |
        pip install bandit
        bandit -r ./src -f json -o bandit-report.json
      continue-on-error: true
    
    # SAST - Semgrep
    - name: Run Semgrep
      run: |
        pip install semgrep
        semgrep --config auto --json --output semgrep-report.json ./src
      continue-on-error: true
    
    # 依赖扫描 - Safety
    - name: Run Safety
      run: |
        pip install safety
        safety check --json --output safety-report.json
      continue-on-error: true
    
    # 依赖扫描 - pip-audit
    - name: Run pip-audit
      run: |
        pip install pip-audit
        pip-audit --format json --output pip-audit-report.json
      continue-on-error: true
    
    # 上传结果
    - name: Upload Reports
      uses: actions/upload-artifact@v3
      with:
        name: security-reports
        path: |
          bandit-report.json
          semgrep-report.json
          safety-report.json
          pip-audit-report.json
```

**本地审计脚本**：
```python
#!/usr/bin/env python3
"""
代码审计自动化脚本
"""

import subprocess
import json
from pathlib import Path
from datetime import datetime

class CodeAuditAutomation:
    """代码审计自动化工具"""
    
    def __init__(self, project_path):
        self.project_path = Path(project_path)
        self.reports_dir = self.project_path / 'audit-reports'
        self.reports_dir.mkdir(exist_ok=True)
    
    def run_all_scans(self):
        """执行所有扫描"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        
        results = {
            'timestamp': timestamp,
            'project': str(self.project_path),
            'scans': {}
        }
        
        # 运行 Bandit
        print("[*] Running Bandit...")
        results['scans']['bandit'] = self.run_bandit()
        
        # 运行 Semgrep
        print("[*] Running Semgrep...")
        results['scans']['semgrep'] = self.run_semgrep()
        
        # 运行 Safety
        print("[*] Running Safety...")
        results['scans']['safety'] = self.run_safety()
        
        # 运行 pip-audit
        print("[*] Running pip-audit...")
        results['scans']['pip_audit'] = self.run_pip_audit()
        
        # 汇总报告
        self.generate_summary_report(results)
        
        return results
    
    def run_bandit(self):
        """运行 Bandit"""
        cmd = ['bandit', '-r', str(self.project_path / 'src'), 
               '-f', 'json', '-o', str(self.reports_dir / 'bandit.json')]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        return {
            'status': 'completed',
            'report': str(self.reports_dir / 'bandit.json'),
            'exit_code': result.returncode
        }
    
    def run_semgrep(self):
        """运行 Semgrep"""
        cmd = ['semgrep', '--config', 'auto', '--json',
               '-o', str(self.reports_dir / 'semgrep.json'),
               str(self.project_path / 'src')]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        return {
            'status': 'completed',
            'report': str(self.reports_dir / 'semgrep.json'),
            'exit_code': result.returncode
        }
    
    def run_safety(self):
        """运行 Safety"""
        cmd = ['safety', 'check', '--json',
               '--output', str(self.reports_dir / 'safety.json')]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        return {
            'status': 'completed',
            'report': str(self.reports_dir / 'safety.json'),
            'exit_code': result.returncode
        }
    
    def run_pip_audit(self):
        """运行 pip-audit"""
        cmd = ['pip-audit', '--format', 'json',
               '--output', str(self.reports_dir / 'pip_audit.json')]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        return {
            'status': 'completed',
            'report': str(self.reports_dir / 'pip_audit.json'),
            'exit_code': result.returncode
        }
    
    def generate_summary_report(self, results):
        """生成汇总报告"""
        summary = {
            'audit_summary': {
                'timestamp': results['timestamp'],
                'project': results['project'],
                'total_issues': 0,
                'critical': 0,
                'high': 0,
                'medium': 0,
                'low': 0
            },
            'scans': results['scans']
        }
        
        # 统计问题数量
        for scan_name, scan_result in results['scans'].items():
            report_path = scan_result.get('report')
            if report_path and Path(report_path).exists():
                with open(report_path) as f:
                    report_data = json.load(f)
                    issues = self.count_issues(report_data, scan_name)
                    summary['audit_summary']['total_issues'] += issues['total']
                    summary['audit_summary']['critical'] += issues['critical']
                    summary['audit_summary']['high'] += issues['high']
                    summary['audit_summary']['medium'] += issues['medium']
                    summary['audit_summary']['low'] += issues['low']
        
        # 保存汇总报告
        summary_path = self.reports_dir / 'summary.json'
        with open(summary_path, 'w') as f:
            json.dump(summary, f, indent=2)
        
        print(f"\n[+] Audit Summary:")
        print(f"    Total Issues: {summary['audit_summary']['total_issues']}")
        print(f"    Critical: {summary['audit_summary']['critical']}")
        print(f"    High: {summary['audit_summary']['high']}")
        print(f"    Medium: {summary['audit_summary']['medium']}")
        print(f"    Low: {summary['audit_summary']['low']}")
        print(f"    Report: {summary_path}")
    
    def count_issues(self, report_data, scan_type):
        """统计问题数量"""
        counts = {'total': 0, 'critical': 0, 'high': 0, 'medium': 0, 'low': 0}
        
        if scan_type == 'bandit':
            for issue in report_data.get('results', []):
                severity = issue.get('issue_severity', 'LOW').lower()
                counts['total'] += 1
                if severity in counts:
                    counts[severity] += 1
        
        elif scan_type == 'semgrep':
            for result in report_data.get('results', []):
                severity = result.get('severity', 'INFO').lower()
                counts['total'] += 1
                if severity in counts:
                    counts[severity] += 1
        
        elif scan_type == 'safety':
            for vulnerability in report_data:
                counts['total'] += 1
                counts['high'] += 1  # 依赖漏洞通常视为高危
        
        elif scan_type == 'pip_audit':
            for entry in report_data.get('dependencies', []):
                for vuln in entry.get('vulns', []):
                    counts['total'] += 1
                    counts['high'] += 1
        
        return counts

# 使用示例
if __name__ == '__main__':
    auditor = CodeAuditAutomation('./myproject')
    auditor.run_all_scans()
```

---

## 审计报告规范

### 报告结构

> ▎ 报告不是写文章，是建证据。证据不完整，修复就是推诿的。

**审计报告模板**：
```markdown
# 代码审计报告

## 执行摘要

### 审计概况
- 项目名称：XXX
- 审计时间：2024-01-15 至 2024-01-20
- 审计范围：核心业务代码
- 审计方法：自动化扫描 + 人工审计

### 风险概览
| 严重程度 | 数量 | 占比 |
|----------|------|------|
| 严重     | 2    | 5%   |
| 高危     | 8    | 20%  |
| 中危     | 15   | 38%  |
| 低危     | 15   | 38%  |

### 关键发现
1. [严重] SQL 注入漏洞 - 用户搜索功能
2. [严重] 认证绕过 - 密码重置功能
3. [高危] XSS 漏洞 - 评论功能

## 详细发现

### 漏洞 1: SQL 注入

#### 漏洞描述
用户搜索功能存在 SQL 注入漏洞，攻击者可获取数据库任意数据。

#### 漏洞位置
- 文件：src/api/search.py
- 函数：search_users()
- 行号：45-50

#### 漏洞代码
```python
# 问题代码
def search_users(query):
    sql = f"SELECT * FROM users WHERE name LIKE '%{query}%'"
    cursor.execute(sql)
```

#### 修复建议
```python
# 修复代码
def search_users(query):
    sql = "SELECT * FROM users WHERE name LIKE %s"
    cursor.execute(sql, (f"%{query}%",))
```

#### 风险评级
- CVSS: 9.8 (Critical)
- 利用难度：低
- 业务影响：高

## 附录

### 审计工具
- Bandit v1.7.5
- Semgrep v1.0.0
- Safety v2.3.0

### 参考标准
- OWASP Top 10 2021
- CWE Top 25
- CERT 安全编码规范
```

---

## 实战审计演练

### 审计案例

**案例：电商系统审计**：
```python
# 电商系统审计检查清单
ecommerce_audit_checklist = {
    'authentication': [
        '密码是否哈希存储',
        '是否有暴力破解防护',
        '会话管理是否安全',
        'MFA 是否实现'
    ],
    'authorization': [
        '是否有越权访问',
        '权限检查是否完善',
        '管理员功能是否隔离'
    ],
    'payment': [
        '支付逻辑是否完整',
        '金额是否服务端验证',
        '订单状态是否防篡改',
        '回调是否验证签名'
    ],
    'data_protection': [
        '敏感数据是否加密',
        '日志是否脱敏',
        '数据传输是否加密'
    ],
    'input_validation': [
        '所有输入是否验证',
        '文件上传是否安全',
        'API 参数是否验证'
    ]
}

# 审计发现示例
audit_findings = [
    {
        'id': 'FIND-001',
        'category': 'SQL Injection',
        'severity': 'CRITICAL',
        'location': 'src/order/query.py:35',
        'description': '订单查询功能存在 SQL 注入',
        'evidence': 'cursor.execute(f"SELECT * FROM orders WHERE user_id = {user_id}")',
        'recommendation': '使用参数化查询',
        'fixed_code': 'cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))'
    },
    {
        'id': 'FIND-002',
        'category': 'Broken Access Control',
        'severity': 'HIGH',
        'location': 'src/order/detail.py:20',
        'description': '订单详情未验证所有权',
        'evidence': 'order = Order.query.get(order_id)  # 无所有权检查',
        'recommendation': '添加所有权验证',
        'fixed_code': 'order = Order.query.filter_by(id=order_id, user_id=current_user.id).first()'
    },
    {
        'id': 'FIND-003',
        'category': 'Insecure Direct Object Reference',
        'severity': 'HIGH',
        'location': 'src/user/profile.py:15',
        'description': '用户资料可越权访问',
        'evidence': 'user = User.query.get(request.args.get("user_id"))',
        'recommendation': '验证当前用户权限',
        'fixed_code': '''
user_id = request.args.get("user_id")
if user_id != str(current_user.id) and not current_user.is_admin:
    abort(403)
user = User.query.get(user_id)
'''
    }
]
```

---

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

---

## 总结与思考

### 核心要点回顾

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

**代码审计框架**：
```
1. 审计方法
   - 基于风险
   - 基于功能
   - 基于数据流

2. 分析技术
   - 静态分析 (SAST)
   - 动态分析 (DAST)
   - 污点分析

3. 漏洞识别
   - 模式匹配
   - 污点追踪
   - 人工审查

4. 工具链
   - Bandit/Semgrep
   - ZAP/Burp
   - 自动化集成
```

**关键成功因素**：
```
1. 工具 + 人工
   - 工具发现明显问题
   - 人工发现复杂问题
   - 两者互补

2. 流程规范
   - 标准审计流程
   - 报告模板
   - 修复跟踪

3. 持续改进
   - 经验沉淀
   - 模式库更新
   - 工具优化
```

### 实战建议

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

**审计能力建设**：
```
1. 工具建设
   - 自动化扫描
   - CI/CD 集成
   - 报告汇总

2. 知识建设
   - 漏洞模式库
   - 修复方案库
   - 案例库

3. 团队建设
   - 审计培训
   - 经验分享
   - 外部交流
```

---

## 参考资料

### 学习资源
```
- OWASP Code Review Guide
  https://owasp.org/www-project-code-review-guide/

- CERT Secure Coding Standards
  https://wiki.sei.cmu.edu/confluence/display/seccode

- CWE Top 25
  https://cwe.mitre.org/top25/
```

### 工具资源
```
- Bandit
  https://github.com/PyCQA/bandit

- Semgrep
  https://github.com/returntocorp/semgrep

- SonarQube
  https://www.sonarqube.org/
```

### 书籍推荐
```
- 《The Art of Software Security Assessment》
- 《Secure Coding Principles and Practices》
- 《Code Review for Software Security》
```

---

**标记 明日预告**：Day 148 - 静态代码分析 SAST

> ▎ 代码审计是方法论，SAST 是技术实现——明天深入静态代码分析 SAST。

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

---

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

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