Day-144-漏洞挖掘方法论

# Day 158: 漏洞挖掘方法论

> 漏洞与攻防系列第 3 天 | 预计阅读时间：60 分钟 | 难度：★★★★★

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 漏洞挖掘方法论 - Day 158           │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ 漏洞挖掘不是盲目测试，是系统方法。方法不系统，挖掘就是碰运气的。今天深入漏洞挖掘方法论。

---

## 清单 目录

1. [漏洞挖掘概述](#漏洞挖掘概述)
2. [信息收集方法](#信息收集方法)
3. [威胁建模方法](#威胁建模方法)
4. [漏洞扫描技术](#漏洞扫描技术)
5. [模糊测试技术](#模糊测试技术)
6. [代码审计技术](#代码审计技术)
7. [渗透测试流程](#渗透测试流程)
8. [漏洞验证方法](#漏洞验证方法)
9. [漏洞报告编写](#漏洞报告编写)
10. [总结与思考](#总结与思考)
11. [参考资料](#参考资料)

---

## 漏洞挖掘概述

### 什么是漏洞挖掘

> ▎ 漏洞挖掘不是黑客行为，是安全研究。研究不系统，发现就是偶然的。

**定义与价值**：
```
漏洞挖掘 (Vulnerability Discovery) 是通过系统化方法发现软件或系统中安全漏洞的过程。

核心价值:
1. 主动发现
   - 在攻击者之前发现
   - 提前修复漏洞
   - 降低安全风险

2. 能力提升
   - 理解攻击手法
   - 提升防护能力
   - 培养安全思维

3. 安全改进
   - 发现设计缺陷
   - 改进安全架构
   - 提升整体安全
```

**挖掘方法分类**：
```
┌─────────────────────────────────────────────────────────┐
│              漏洞挖掘方法分类                            │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  白盒测试                                               │
│  ├── 有源代码访问权限                                   │
│  ├── 代码审计                                           │
│  ├── 静态分析                                           │
│  └── 覆盖率高                                           │
│                                                         │
│  灰盒测试                                               │
│  ├── 部分内部知识                                       │
│  ├── 架构了解                                           │
│  ├── 有限代码访问                                       │
│  └── 平衡效率与覆盖                                     │
│                                                         │
│  黑盒测试                                               │
│  ├── 无内部知识                                         │
│  ├── 外部测试                                           │
│  ├── 模拟真实攻击                                       │
│  └── 发现运行时问题                                     │
│                                                         │
└─────────────────────────────────────────────────────────┘

最佳实践：白盒 + 灰盒 + 黑盒组合使用
```

### 挖掘流程

**标准流程**：
```python
# 漏洞挖掘流程
class VulnerabilityDiscoveryProcess:
    """漏洞挖掘标准流程"""
    
    # 流程阶段
    phases = {
        'reconnaissance': {
            'name': '信息收集',
            'description': '收集目标信息',
            'duration': '20%',
            'tools': ['Nmap', 'Whois', 'Shodan'],
            'outputs': ['目标清单', '资产清单', '技术栈信息']
        },
        
        'threat_modeling': {
            'name': '威胁建模',
            'description': '分析潜在威胁',
            'duration': '15%',
            'tools': ['STRIDE', '攻击树'],
            'outputs': ['威胁列表', '攻击场景', '优先级']
        },
        
        'scanning': {
            'name': '漏洞扫描',
            'description': '自动化扫描漏洞',
            'duration': '15%',
            'tools': ['Nessus', 'ZAP', 'Burp'],
            'outputs': ['漏洞列表', '风险评级']
        },
        
        'exploitation': {
            'name': '漏洞利用',
            'description': '验证漏洞可利用性',
            'duration': '25%',
            'tools': ['Metasploit', '自定义脚本'],
            'outputs': ['利用代码', '影响证明']
        },
        
        'reporting': {
            'name': '报告编写',
            'description': '编写漏洞报告',
            'duration': '15%',
            'tools': ['报告模板', '截图工具'],
            'outputs': ['漏洞报告', '修复建议']
        },
        
        'remediation': {
            'name': '修复验证',
            'description': '验证修复效果',
            'duration': '10%',
            'tools': ['复测工具'],
            'outputs': ['修复验证报告']
        }
    }
    
    def execute_process(self, target):
        """执行挖掘流程"""
        results = {
            'target': target,
            'phases': {},
            'findings': [],
            'summary': {}
        }
        
        # 执行各阶段
        for phase_name, phase_info in self.phases.items():
            phase_result = self.execute_phase(phase_name, target)
            results['phases'][phase_name] = phase_result
            results['findings'].extend(phase_result.get('findings', []))
        
        # 汇总结果
        results['summary'] = self.summarize_findings(results['findings'])
        
        return results
    
    def execute_phase(self, phase_name, target):
        """执行单个阶段"""
        # 实际实现调用各阶段的具体方法
        pass
    
    def summarize_findings(self, findings):
        """汇总发现"""
        summary = {
            'total': len(findings),
            'critical': len([f for f in findings if f['severity'] == 'critical']),
            'high': len([f for f in findings if f['severity'] == 'high']),
            'medium': len([f for f in findings if f['severity'] == 'medium']),
            'low': len([f for f in findings if f['severity'] == 'low'])
        }
        return summary
```

---

## 信息收集方法

### 被动收集

> ▎ 信息收集不是盲目搜索，是有目标的。目标不明确，收集就是浪费时间的。

**被动收集技术**：
```python
# 被动信息收集
class PassiveReconnaissance:
    """被动信息收集"""
    
    def __init__(self, target):
        self.target = target
        self.results = {}
    
    def collect_all(self):
        """执行所有被动收集"""
        self.results = {
            'domain_info': self.get_domain_info(),
            'dns_records': self.get_dns_records(),
            'subdomains': self.find_subdomains(),
            'technologies': self.identify_technologies(),
            'exposed_ports': self.scan_ports(),
            'web_content': self.analyze_web_content(),
            'employee_info': self.find_employee_info(),
            'documents': self.find_documents()
        }
        return self.results
    
    def get_domain_info(self):
        """获取域名信息"""
        import whois
        
        try:
            domain_info = whois.whois(self.target)
            return {
                'registrar': domain_info.registrar,
                'creation_date': domain_info.creation_date,
                'expiration_date': domain_info.expiration_date,
                'name_servers': domain_info.name_servers,
                'status': domain_info.status
            }
        except Exception as e:
            return {'error': str(e)}
    
    def get_dns_records(self):
        """获取 DNS 记录"""
        import dns.resolver
        
        records = {}
        record_types = ['A', 'AAAA', 'MX', 'NS', 'TXT', 'SOA', 'CNAME']
        
        for rtype in record_types:
            try:
                answers = dns.resolver.resolve(self.target, rtype)
                records[rtype] = [str(rdata) for rdata in answers]
            except:
                records[rtype] = []
        
        return records
    
    def find_subdomains(self):
        """发现子域名"""
        subdomains = set()
        
        # 使用证书透明度
        subdomains.update(self.ct_subdomain_enumeration())
        
        # 使用搜索引擎
        subdomains.update(self.search_engine_enumeration())
        
        # 使用 DNS 枚举
        subdomains.update(self.dns_enumeration())
        
        return list(subdomains)
    
    def ct_subdomain_enumeration(self):
        """证书透明度子域名枚举"""
        import requests
        
        subdomains = set()
        url = f"https://crt.sh/?q=%.{self.target}&output=json"
        
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                data = response.json()
                for entry in data:
                    name = entry.get('name_value', '')
                    subdomains.update(name.split('\n'))
        except:
            pass
        
        return subdomains
    
    def search_engine_enumeration(self):
        """搜索引擎枚举"""
        subdomains = set()
        
        # Google dorks
        dorks = [
            f'site:*.{self.target}',
            f'site:{self.target} -www',
        ]
        
        # 实际实现需要调用搜索引擎 API
        # 这里仅示例
        
        return subdomains
    
    def dns_enumeration(self):
        """DNS 枚举"""
        import dns.resolver
        
        subdomains = set()
        common_prefixes = ['www', 'mail', 'ftp', 'admin', 'api', 'dev', 'test', 'staging']
        
        for prefix in common_prefixes:
            try:
                domain = f"{prefix}.{self.target}"
                dns.resolver.resolve(domain, 'A')
                subdomains.add(domain)
            except:
                pass
        
        return subdomains
    
    def identify_technologies(self):
        """识别技术栈"""
        import requests
        
        technologies = {
            'server': None,
            'framework': None,
            'language': None,
            'cms': None
        }
        
        try:
            response = requests.get(f"https://{self.target}", timeout=10)
            
            # 从响应头识别
            server = response.headers.get('Server', '')
            if server:
                technologies['server'] = server
            
            # 从 HTML 内容识别
            content = response.text
            
            if 'wordpress' in content.lower():
                technologies['cms'] = 'WordPress'
            elif 'drupal' in content.lower():
                technologies['cms'] = 'Drupal'
            
            if 'react' in content.lower():
                technologies['framework'] = 'React'
            elif 'angular' in content.lower():
                technologies['framework'] = 'Angular'
            elif 'vue' in content.lower():
                technologies['framework'] = 'Vue'
            
        except:
            pass
        
        return technologies
    
    def scan_ports(self):
        """端口扫描"""
        import nmap
        
        nm = nmap.PortScanner()
        nm.scan(self.target, '1-1000')
        
        exposed_ports = {}
        for host in nm.all_hosts():
            for proto in nm[host].all_protocols():
                exposed_ports[host] = nm[host][proto].keys()
        
        return exposed_ports
    
    def analyze_web_content(self):
        """分析 Web 内容"""
        import requests
        from bs4 import BeautifulSoup
        
        content_info = {
            'forms': [],
            'inputs': [],
            'links': [],
            'comments': []
        }
        
        try:
            response = requests.get(f"https://{self.target}", timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # 提取表单
            for form in soup.find_all('form'):
                content_info['forms'].append({
                    'action': form.get('action', ''),
                    'method': form.get('method', 'GET'),
                    'inputs': [input.get('name', '') for input in form.find_all('input')]
                })
            
            # 提取链接
            for link in soup.find_all('a', href=True):
                content_info['links'].append(link['href'])
            
            # 提取注释
            comments = soup.find_all(string=lambda text: isinstance(text, str) and text.strip().startswith('<!--'))
            content_info['comments'] = [str(c) for c in comments[:10]]  # 限制数量
            
        except:
            pass
        
        return content_info
    
    def find_employee_info(self):
        """查找员工信息"""
        # 实际实现需要调用 LinkedIn API 等
        # 这里仅示例
        return {
            'count': 0,
            'roles': []
        }
    
    def find_documents(self):
        """查找公开文档"""
        documents = []
        
        # 搜索 PDF、DOC 等文档
        # 实际实现需要调用搜索引擎
        
        return documents
```

### 主动收集

**主动收集技术**：
```python
# 主动信息收集
class ActiveReconnaissance:
    """主动信息收集"""
    
    def __init__(self, target):
        self.target = target
    
    def directory_enumeration(self):
        """目录枚举"""
        import requests
        
        directories = []
        wordlist = [
            'admin', 'login', 'api', 'backup', 'config',
            'database', 'test', 'dev', 'staging', 'prod'
        ]
        
        for dir in wordlist:
            url = f"https://{self.target}/{dir}"
            try:
                response = requests.get(url, timeout=5)
                if response.status_code == 200:
                    directories.append(url)
            except:
                pass
        
        return directories
    
    def parameter_discovery(self):
        """参数发现"""
        import requests
        
        parameters = []
        common_params = [
            'id', 'user', 'username', 'password', 'email',
            'page', 'sort', 'order', 'search', 'query'
        ]
        
        for param in common_params:
            url = f"https://{self.target}/"
            try:
                response = requests.get(url, params={param: 'test'})
                if response.status_code == 200:
                    parameters.append(param)
            except:
                pass
        
        return parameters
    
    def technology_fingerprinting(self):
        """技术指纹识别"""
        import requests
        
        fingerprints = {
            'server': None,
            'framework': None,
            'language': None,
            'database': None
        }
        
        # 通过错误页面识别
        error_pages = self.trigger_error_pages()
        
        # 通过特定文件识别
        specific_files = self.check_specific_files()
        
        return fingerprints
    
    def trigger_error_pages(self):
        """触发错误页面"""
        # 访问不存在的页面获取错误信息
        pass
    
    def check_specific_files(self):
        """检查特定文件"""
        files = [
            '/robots.txt',
            '/sitemap.xml',
            '/.git/config',
            '/.env',
            '/wp-config.php.bak'
        ]
        
        found = []
        import requests
        
        for file in files:
            url = f"https://{self.target}{file}"
            try:
                response = requests.get(url, timeout=5)
                if response.status_code == 200:
                    found.append(file)
            except:
                pass
        
        return found
```

---

## 威胁建模方法

### STRIDE 威胁建模

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

**STRIDE 方法**：
```python
# STRIDE 威胁建模
class STRIDEThreatModeling:
    """STRIDE 威胁建模"""
    
    stride_categories = {
        'Spoofing': {
            'description': '身份欺骗',
            'questions': [
                '系统如何验证用户身份？',
                '认证凭证如何保护？',
                '如何防止会话劫持？'
            ],
            'mitigations': [
                '强认证机制',
                '多因素认证',
                '安全会话管理'
            ]
        },
        'Tampering': {
            'description': '数据篡改',
            'questions': [
                '数据完整性如何保证？',
                '如何检测数据篡改？',
                '配置如何保护？'
            ],
            'mitigations': [
                '数字签名',
                '完整性校验',
                '访问控制'
            ]
        },
        'Repudiation': {
            'description': '行为抵赖',
            'questions': [
                '关键操作是否记录日志？',
                '日志是否防篡改？',
                '是否有时间同步？'
            ],
            'mitigations': [
                '审计日志',
                '数字签名',
                '时间戳服务'
            ]
        },
        'Information Disclosure': {
            'description': '信息泄露',
            'questions': [
                '敏感数据如何加密？',
                '谁可以访问什么数据？',
                '数据传输如何保护？'
            ],
            'mitigations': [
                '数据加密',
                '访问控制',
                '数据最小化'
            ]
        },
        'Denial of Service': {
            'description': '拒绝服务',
            'questions': [
                '系统如何处理大流量？',
                '资源使用如何限制？',
                '如何快速恢复？'
            ],
            'mitigations': [
                '速率限制',
                '冗余设计',
                '弹性伸缩'
            ]
        },
        'Elevation of Privilege': {
            'description': '权限提升',
            'questions': [
                '权限如何分配？',
                '如何防止越权？',
                '如何审计权限使用？'
            ],
            'mitigations': [
                '最小权限',
                '访问控制',
                '权限审计'
            ]
        }
    }
    
    def model_system(self, system_description):
        """对系统进行 STRIDE 建模"""
        threats = []
        
        for category, info in self.stride_categories.items():
            category_threats = self.analyze_category(category, info, system_description)
            threats.extend(category_threats)
        
        # 按风险排序
        threats.sort(key=lambda x: x['risk_score'], reverse=True)
        
        return threats
    
    def analyze_category(self, category, info, system_description):
        """分析类别威胁"""
        threats = []
        
        for question in info['questions']:
            # 分析问题答案
            answer = self.answer_question(question, system_description)
            
            if not answer['secure']:
                threats.append({
                    'category': category,
                    'question': question,
                    'finding': answer['finding'],
                    'risk_score': answer['risk_score'],
                    'mitigation': info['mitigations'][0]
                })
        
        return threats
    
    def answer_question(self, question, system_description):
        """回答问题"""
        # 实际实现需要分析系统描述
        # 这里仅示例
        return {
            'secure': False,
            'finding': '未实施相关安全措施',
            'risk_score': 8
        }
```

### 攻击树分析

**攻击树方法**：
```python
# 攻击树分析
class AttackTreeAnalysis:
    """攻击树分析"""
    
    def __init__(self, goal):
        self.goal = goal  # 攻击目标
        self.tree = {
            'goal': goal,
            'children': []
        }
    
    def build_tree(self):
        """构建攻击树"""
        # 分解攻击目标
        self.tree['children'] = [
            {
                'name': '获取认证凭证',
                'type': 'OR',  # OR 节点：任一子节点成功即可
                'children': [
                    {'name': '暴力破解密码', 'type': 'leaf'},
                    {'name': '钓鱼攻击', 'type': 'leaf'},
                    {'name': '凭证泄露利用', 'type': 'leaf'}
                ]
            },
            {
                'name': '绕过认证',
                'type': 'OR',
                'children': [
                    {'name': 'SQL 注入', 'type': 'leaf'},
                    {'name': '会话劫持', 'type': 'leaf'},
                    {'name': '认证绕过漏洞', 'type': 'leaf'}
                ]
            },
            {
                'name': '提升权限',
                'type': 'OR',
                'children': [
                    {'name': '越权访问', 'type': 'leaf'},
                    {'name': '权限配置错误', 'type': 'leaf'},
                    {'name': '内核漏洞利用', 'type': 'leaf'}
                ]
            }
        ]
        
        return self.tree
    
    def analyze_path(self, node, path=[]):
        """分析攻击路径"""
        paths = []
        current_path = path + [node['name']]
        
        if node.get('type') == 'leaf':
            paths.append(current_path)
        else:
            for child in node.get('children', []):
                paths.extend(self.analyze_path(child, current_path))
        
        return paths
    
    def get_all_paths(self):
        """获取所有攻击路径"""
        paths = []
        for child in self.tree['children']:
            paths.extend(self.analyze_path(child))
        return paths
    
    def calculate_difficulty(self, path):
        """计算路径难度"""
        # 基于路径中的步骤数量和技术难度
        difficulty = len(path) * 2  # 简化计算
        return difficulty
    
    def prioritize_paths(self):
        """优先排序攻击路径"""
        paths = self.get_all_paths()
        
        scored_paths = []
        for path in paths:
            difficulty = self.calculate_difficulty(path)
            scored_paths.append({
                'path': path,
                'difficulty': difficulty,
                'priority': 'high' if difficulty < 5 else 'medium'
            })
        
        scored_paths.sort(key=lambda x: x['difficulty'])
        
        return scored_paths
```

---

## 漏洞扫描技术

### 自动化扫描

> ▎ 漏洞扫描不是跑工具，是系统检测。检测不系统，漏洞就是遗漏的。

**扫描器使用**：
```python
# 自动化漏洞扫描
class AutomatedVulnerabilityScanner:
    """自动化漏洞扫描"""
    
    def __init__(self, target):
        self.target = target
        self.scanners = {
            'web': self.web_scanner,
            'network': self.network_scanner,
            'dependency': self.dependency_scanner
        }
    
    def run_all_scans(self):
        """运行所有扫描"""
        results = {}
        
        for scanner_name, scanner_func in self.scanners.items():
            results[scanner_name] = scanner_func()
        
        return results
    
    def web_scanner(self):
        """Web 应用扫描"""
        # 使用 OWASP ZAP
        import requests
        
        zap_url = "http://localhost:8080"
        api_key = "zap-api-key"
        
        # 启动爬虫
        spider_params = {
            'url': f"https://{self.target}",
            'apikey': api_key
        }
        response = requests.get(f"{zap_url}/JSON/spider/action/scan/", params=spider_params)
        spider_id = response.json().get('scan')
        
        # 等待爬虫完成
        while True:
            status = requests.get(f"{zap_url}/JSON/spider/view/status/", params={'scanId': spider_id})
            if int(status.json().get('status', 0)) == 100:
                break
        
        # 启动主动扫描
        scan_params = {
            'url': f"https://{self.target}",
            'apikey': api_key
        }
        response = requests.get(f"{zap_url}/JSON/ascan/action/scan/", params=scan_params)
        scan_id = response.json().get('scan')
        
        # 等待扫描完成
        while True:
            status = requests.get(f"{zap_url}/JSON/ascan/view/status/", params={'scanId': scan_id})
            if int(status.json().get('status', 0)) == 100:
                break
        
        # 获取告警
        alerts = requests.get(f"{zap_url}/JSON/core/view/alerts/", params={'apikey': api_key})
        
        return alerts.json().get('alerts', [])
    
    def network_scanner(self):
        """网络扫描"""
        import nmap
        
        nm = nmap.PortScanner()
        nm.scan(self.target, '1-65535', '-sV -sC -O')
        
        results = []
        for host in nm.all_hosts():
            for proto in nm[host].all_protocols():
                for port in nm[host][proto]:
                    results.append({
                        'host': host,
                        'port': port,
                        'protocol': proto,
                        'state': nm[host][proto][port]['state'],
                        'service': nm[host][proto][port].get('name', ''),
                        'version': nm[host][proto][port].get('version', '')
                    })
        
        return results
    
    def dependency_scanner(self):
        """依赖扫描"""
        import subprocess
        
        # Python 依赖扫描
        result = subprocess.run(
            ['safety', 'check', '--json'],
            capture_output=True,
            text=True
        )
        
        import json
        vulnerabilities = json.loads(result.stdout)
        
        return vulnerabilities
```

### 手动测试

**手动测试技术**：
```python
# 手动漏洞测试
class ManualVulnerabilityTesting:
    """手动漏洞测试"""
    
    def test_sql_injection(self, url, parameters):
        """SQL 注入测试"""
        payloads = [
            "' OR '1'='1",
            "1; DROP TABLE users--",
            "' UNION SELECT NULL--",
            "1' AND '1'='1",
            "' OR 1=1--"
        ]
        
        findings = []
        
        for param in parameters:
            for payload in payloads:
                test_url = f"{url}?{param}={payload}"
                
                import requests
                response = requests.get(test_url, timeout=10)
                
                # 检测 SQL 错误
                sql_errors = [
                    'SQL syntax', 'mysql_fetch', 'ORA-',
                    'PostgreSQL', 'SQLite', 'syntax error'
                ]
                
                for error in sql_errors:
                    if error.lower() in response.text.lower():
                        findings.append({
                            'type': 'SQL Injection',
                            'url': url,
                            'parameter': param,
                            'payload': payload,
                            'evidence': error
                        })
                        break
        
        return findings
    
    def test_xss(self, url, parameters):
        """XSS 测试"""
        payloads = [
            '<script>alert(1)</script>',
            '<img src=x onerror=alert(1)>',
            '<svg onload=alert(1)>',
            "javascript:alert(1)",
            "'-alert(1)-'"
        ]
        
        findings = []
        
        for param in parameters:
            for payload in payloads:
                test_url = f"{url}?{param}={payload}"
                
                import requests
                response = requests.get(test_url, timeout=10)
                
                # 检测 XSS
                if payload in response.text:
                    # 检查是否编码
                    if not self.is_encoded(response.text, payload):
                        findings.append({
                            'type': 'XSS',
                            'url': url,
                            'parameter': param,
                            'payload': payload
                        })
        
        return findings
    
    def is_encoded(self, content, payload):
        """检查是否编码"""
        from html import escape
        encoded = escape(payload)
        return encoded in content
    
    def test_csrf(self, url, forms):
        """CSRF 测试"""
        findings = []
        
        for form in forms:
            # 检查是否有 CSRF token
            has_csrf_token = False
            
            for input_field in form.get('inputs', []):
                if 'csrf' in input_field.lower() or 'token' in input_field.lower():
                    has_csrf_token = True
                    break
            
            if not has_csrf_token and form.get('method', 'GET').upper() == 'POST':
                findings.append({
                    'type': 'CSRF',
                    'url': form.get('action', url),
                    'evidence': 'Missing CSRF token'
                })
        
        return findings
    
    def test_authentication(self, url):
        """认证测试"""
        findings = []
        
        import requests
        
        # 测试弱密码
        weak_credentials = [
            ('admin', 'admin'),
            ('admin', 'password'),
            ('admin', '123456'),
            ('root', 'root'),
            ('test', 'test')
        ]
        
        for username, password in weak_credentials:
            response = requests.post(
                f"{url}/login",
                data={'username': username, 'password': password}
            )
            
            if 'welcome' in response.text.lower() or 'success' in response.text.lower():
                findings.append({
                    'type': 'Weak Credentials',
                    'url': url,
                    'credentials': f"{username}:{password}"
                })
        
        # 测试暴力破解防护
        for i in range(10):
            response = requests.post(
                f"{url}/login",
                data={'username': 'test', 'password': 'wrong'}
            )
        
        # 检查是否有限制
        if response.status_code == 200:
            findings.append({
                'type': 'No Brute Force Protection',
                'url': url,
                'evidence': 'No account lockout after 10 failed attempts'
            })
        
        return findings
```

---

## 模糊测试技术

### 模糊测试原理

> ▎ 模糊测试不是乱输入，是系统测试。测试不系统，漏洞就是遗漏的。

**模糊测试方法**：
```python
# 模糊测试
class FuzzTesting:
    """模糊测试"""
    
    def __init__(self, target):
        self.target = target
        self.test_cases = self.generate_test_cases()
    
    def generate_test_cases(self):
        """生成测试用例"""
        test_cases = {
            'boundary_values': self.generate_boundary_values(),
            'special_characters': self.generate_special_characters(),
            'long_strings': self.generate_long_strings(),
            'format_strings': self.generate_format_strings(),
            'sql_injection': self.generate_sql_payloads(),
            'xss_payloads': self.generate_xss_payloads()
        }
        return test_cases
    
    def generate_boundary_values(self):
        """生成边界值"""
        return [
            0, 1, -1,  # 整数边界
            127, 128, -128,  # 8 位边界
            255, 256,  # 8 位无符号边界
            32767, 32768,  # 16 位边界
            2147483647, 2147483648,  # 32 位边界
            '',  # 空字符串
            'a' * 1000,  # 长字符串
        ]
    
    def generate_special_characters(self):
        """生成特殊字符"""
        return [
            '<', '>', '"', "'",  # HTML 特殊字符
            ';', '|', '&', '$', '`',  # Shell 特殊字符
            '\\', '/',  # 路径字符
            '%00', '%0a', '%0d',  # 控制字符
            '..',  # 路径遍历
        ]
    
    def generate_long_strings(self, lengths=[100, 1000, 10000, 100000]):
        """生成长字符串"""
        strings = []
        for length in lengths:
            strings.append('A' * length)
        return strings
    
    def generate_format_strings(self):
        """生成格式化字符串"""
        return [
            '%s', '%x', '%n', '%p',
            '%1$ s', '%2$ x', '%3$ n',
            '{0}', '{1}', '{2}',
        ]
    
    def generate_sql_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 generate_xss_payloads(self):
        """生成 XSS payload"""
        return [
            '<script>alert(1)</script>',
            '<img src=x onerror=alert(1)>',
            '<svg onload=alert(1)>',
            'javascript:alert(1)',
            "'-alert(1)-'",
        ]
    
    def fuzz_parameter(self, url, parameter):
        """模糊测试参数"""
        findings = []
        
        import requests
        
        for test_case in self.test_cases['boundary_values']:
            try:
                response = requests.get(url, params={parameter: test_case}, timeout=5)
                
                if response.status_code >= 500:
                    findings.append({
                        'type': 'Server Error',
                        'url': url,
                        'parameter': parameter,
                        'payload': test_case,
                        'status_code': response.status_code
                    })
            except:
                pass
        
        for test_case in self.test_cases['special_characters']:
            try:
                response = requests.get(url, params={parameter: test_case}, timeout=5)
                
                # 检测 SQL 错误
                sql_errors = ['SQL syntax', 'mysql_fetch', 'ORA-']
                for error in sql_errors:
                    if error.lower() in response.text.lower():
                        findings.append({
                            'type': 'SQL Injection',
                            'url': url,
                            'parameter': parameter,
                            'payload': test_case
                        })
                        break
            except:
                pass
        
        return findings
    
    def fuzz_form(self, form_action, form_fields):
        """模糊测试表单"""
        findings = []
        
        import requests
        
        for field in form_fields:
            for test_case in self.test_cases['boundary_values'] + self.test_cases['special_characters']:
                try:
                    data = {field: test_case}
                    response = requests.post(form_action, data=data, timeout=5)
                    
                    if response.status_code >= 500:
                        findings.append({
                            'type': 'Server Error',
                            'url': form_action,
                            'field': field,
                            'payload': test_case
                        })
                except:
                    pass
        
        return findings
```

---

## 代码审计技术

### 静态分析

> ▎ 代码审计不是随便看，是系统审查。审查不系统，漏洞就是隐藏的。

**代码审计方法**：
```python
# 代码审计
class CodeAudit:
    """代码审计"""
    
    def __init__(self, codebase_path):
        self.codebase_path = codebase_path
        self.findings = []
    
    def audit_all(self):
        """执行全面审计"""
        self.findings = []
        
        # 收集所有源代码文件
        source_files = self.collect_source_files()
        
        # 审计每个文件
        for file in source_files:
            file_findings = self.audit_file(file)
            self.findings.extend(file_findings)
        
        return self.findings
    
    def collect_source_files(self):
        """收集源代码文件"""
        import os
        
        source_files = []
        extensions = ['.py', '.java', '.js', '.php', '.c', '.cpp']
        
        for root, dirs, files in os.walk(self.codebase_path):
            # 跳过特定目录
            if any(skip in root for skip in ['node_modules', 'venv', '.git', '__pycache__']):
                continue
            
            for file in files:
                if any(file.endswith(ext) for ext in extensions):
                    source_files.append(os.path.join(root, file))
        
        return source_files
    
    def audit_file(self, file_path):
        """审计单个文件"""
        findings = []
        
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            lines = content.split('\n')
        
        # 检查常见漏洞模式
        findings.extend(self.check_sql_injection(file_path, content, lines))
        findings.extend(self.check_xss(file_path, content, lines))
        findings.extend(self.check_command_injection(file_path, content, lines))
        findings.extend(self.check_hardcoded_secrets(file_path, content, lines))
        findings.extend(self.check_insecure_crypto(file_path, content, lines))
        
        return findings
    
    def check_sql_injection(self, file_path, content, lines):
        """检查 SQL 注入"""
        findings = []
        
        # 危险模式
        dangerous_patterns = [
            r'execute\s*\(\s*["\'].*%.*["\'].*%',  # % 格式化
            r'execute\s*\(\s*f["\'].*\{.*\}',  # f-string
            r'execute\s*\(\s*["\'].*\.format\s*\(',  # .format()
            r'execute\s*\(\s*["\'].*\+',  # 字符串拼接
        ]
        
        import re
        for i, line in enumerate(lines, 1):
            for pattern in dangerous_patterns:
                if re.search(pattern, line):
                    findings.append({
                        'type': 'SQL Injection',
                        'file': file_path,
                        'line': i,
                        'code': line.strip(),
                        'severity': 'high'
                    })
        
        return findings
    
    def check_xss(self, file_path, content, lines):
        """检查 XSS"""
        findings = []
        
        # 危险模式
        dangerous_patterns = [
            r'return\s+.*request\.(args|form|json)',  # 直接返回用户输入
            r'render.*\{.*request\.',  # 模板中直接使用用户输入
            r'Markup\s*\(\s*request\.',  # Markup 包装用户输入
        ]
        
        import re
        for i, line in enumerate(lines, 1):
            for pattern in dangerous_patterns:
                if re.search(pattern, line):
                    findings.append({
                        'type': 'XSS',
                        'file': file_path,
                        'line': i,
                        'code': line.strip(),
                        'severity': 'high'
                    })
        
        return findings
    
    def check_command_injection(self, file_path, content, lines):
        """检查命令注入"""
        findings = []
        
        dangerous_patterns = [
            r'os\.system\s*\(\s*f["\']',  # f-string 命令
            r'os\.system\s*\(\s*.*\+',  # 字符串拼接命令
            r'subprocess\..*shell\s*=\s*True',  # shell=True
            r'eval\s*\(\s*request\.',  # eval 用户输入
            r'exec\s*\(\s*request\.',  # exec 用户输入
        ]
        
        import re
        for i, line in enumerate(lines, 1):
            for pattern in dangerous_patterns:
                if re.search(pattern, line):
                    findings.append({
                        'type': 'Command Injection',
                        'file': file_path,
                        'line': i,
                        'code': line.strip(),
                        'severity': 'critical'
                    })
        
        return findings
    
    def check_hardcoded_secrets(self, file_path, content, lines):
        """检查硬编码凭证"""
        findings = []
        
        dangerous_patterns = [
            r'password\s*=\s*["\'][^"\']+["\']',  # 硬编码密码
            r'secret\s*=\s*["\'][^"\']+["\']',  # 硬编码密钥
            r'api_key\s*=\s*["\'][^"\']+["\']',  # 硬编码 API 密钥
            r'token\s*=\s*["\'][^"\']+["\']',  # 硬编码令牌
        ]
        
        import re
        for i, line in enumerate(lines, 1):
            for pattern in dangerous_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    # 排除示例和测试代码
                    if 'example' not in line.lower() and 'test' not in line.lower():
                        findings.append({
                            'type': 'Hardcoded Secret',
                            'file': file_path,
                            'line': i,
                            'code': line.strip(),
                            'severity': 'high'
                        })
        
        return findings
    
    def check_insecure_crypto(self, file_path, content, lines):
        """检查不安全加密"""
        findings = []
        
        dangerous_patterns = [
            r'MD5\s*\(',  # MD5
            r'SHA1\s*\(',  # SHA1
            r'DES\s*\(',  # DES
            r'RC4\s*\(',  # RC4
            r'random\.random\s*\(',  # 非安全随机数
        ]
        
        import re
        for i, line in enumerate(lines, 1):
            for pattern in dangerous_patterns:
                if re.search(pattern, line):
                    findings.append({
                        'type': 'Insecure Cryptography',
                        'file': file_path,
                        'line': i,
                        'code': line.strip(),
                        'severity': 'medium'
                    })
        
        return findings
```

---

## 渗透测试流程

### PTES 标准

**渗透测试执行标准**：
```
PTES (Penetration Testing Execution Standard):

阶段 1: 前期交互
├── 确定测试范围
├── 确定测试规则
├── 获得书面授权
└── 定义应急联系人

阶段 2: 情报收集
├── 被动收集
├── 主动收集
├── 社会工程
└── 物理安全评估

阶段 3: 威胁建模
├── 识别资产
├── 识别威胁
├── 识别脆弱性
└── 确定攻击向量

阶段 4: 漏洞分析
├── 自动化扫描
├── 手动验证
├── 漏洞评级
└── 利用可行性分析

阶段 5: 渗透攻击
├── 获取访问
├── 权限提升
├── 横向移动
└── 维持访问

阶段 6: 后渗透攻击
├── 数据发现
├── 数据收集
├── 影响分析
└── 业务影响评估

阶段 7: 报告
├── 执行摘要
├── 技术细节
├── 修复建议
└── 附录
```

---

## 漏洞验证方法

### 验证流程

```python
# 漏洞验证
class VulnerabilityVerification:
    """漏洞验证"""
    
    def verify_vulnerability(self, finding):
        """验证漏洞"""
        verification = {
            'vulnerability': finding,
            'verified': False,
            'evidence': [],
            'impact': None,
            'reproduction_steps': []
        }
        
        # 重现漏洞
        reproduction = self.reproduce(finding)
        
        if reproduction['success']:
            verification['verified'] = True
            verification['evidence'] = reproduction['evidence']
            verification['impact'] = reproduction['impact']
            verification['reproduction_steps'] = reproduction['steps']
        
        return verification
    
    def reproduce(self, finding):
        """重现漏洞"""
        # 根据漏洞类型执行不同的重现方法
        if finding['type'] == 'SQL Injection':
            return self.reproduce_sql_injection(finding)
        elif finding['type'] == 'XSS':
            return self.reproduce_xss(finding)
        # ... 其他类型
        
        return {'success': False}
    
    def reproduce_sql_injection(self, finding):
        """重现 SQL 注入"""
        import requests
        
        url = finding['url']
        parameter = finding['parameter']
        payload = finding['payload']
        
        test_url = f"{url}?{parameter}={payload}"
        response = requests.get(test_url, timeout=10)
        
        # 验证
        sql_errors = ['SQL syntax', 'mysql_fetch', 'ORA-']
        evidence = []
        
        for error in sql_errors:
            if error.lower() in response.text.lower():
                evidence.append(f"SQL error detected: {error}")
        
        return {
            'success': len(evidence) > 0,
            'evidence': evidence,
            'impact': 'Database compromise possible',
            'steps': [
                f"1. Access URL: {url}",
                f"2. Modify parameter '{parameter}' to: {payload}",
                f"3. Observe SQL error in response"
            ]
        }
```

---

## 漏洞报告编写

### 报告结构

```markdown
# 漏洞报告模板

## 执行摘要

### 测试概述
- 测试目标：[目标系统]
- 测试时间：[开始时间] - [结束时间]
- 测试类型：[黑盒/灰盒/白盒]
- 测试范围：[范围描述]

### 风险概览
| 严重程度 | 数量 | 占比 |
|----------|------|------|
| Critical | X    | X%   |
| High     | X    | X%   |
| Medium   | X    | X%   |
| Low      | X    | X%   |

### 关键发现
1. [关键漏洞 1]
2. [关键漏洞 2]
3. [关键漏洞 3]

## 详细发现

### 漏洞 1: [漏洞名称]

#### 漏洞信息
- CVE 编号：[如有]
- CWE 编号：[CWE-XXX]
- 严重程度：[Critical/High/Medium/Low]
- CVSS 分数：[X.X]

#### 漏洞描述
[详细描述漏洞]

#### 影响分析
[描述漏洞可能造成的影响]

#### 复现步骤
1. [步骤 1]
2. [步骤 2]
3. [步骤 3]

#### 证据
[截图、响应等]

#### 修复建议
[具体修复建议]

#### 参考链接
- [相关文档]
- [相关 CVE]

## 附录

### 测试工具
- [工具列表]

### 术语表
- [术语解释]
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。漏洞挖掘不是碰运气，是系统方法。方法不系统，发现就是偶然的。

---

## 总结与思考

### 核心要点回顾

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

**漏洞挖掘框架**：
```
1. 信息收集
   - 被动收集
   - 主动收集
   - 社会工程

2. 威胁建模
   - STRIDE
   - 攻击树
   - 攻击面分析

3. 漏洞扫描
   - 自动化扫描
   - 手动测试
   - 模糊测试

4. 代码审计
   - 静态分析
   - 动态分析
   - 人工审查

5. 渗透测试
   - PTES 标准
   - 漏洞验证
   - 报告编写
```

**关键成功因素**：
```
1. 系统化方法
   - 标准流程
   - 完整覆盖
   - 可重复性

2. 工具 + 人工
   - 工具提效率
   - 人工找深度
   - 两者互补

3. 持续学习
   - 新技术学习
   - 新漏洞研究
   - 经验分享
```

---

## 参考资料

### 学习资源
```
- PTES Standard
  http://www.pentest-standard.org/

- OWASP Testing Guide
  https://owasp.org/www-project-web-security-testing-guide/

- NIST SP 800-115
  https://csrc.nist.gov/publications/detail/sp/800-115/final
```

### 工具资源
```
- Metasploit
  https://www.metasploit.com/

- Burp Suite
  https://portswigger.net/burp

- OWASP ZAP
  https://www.zaproxy.org/
```

### 书籍推荐
```
- 《The Web Application Hacker's Handbook》
- 《Penetration Testing: A Hands-On Introduction》
- 《The Hacker Playbook》
```

---

**标记 明日预告**：Day 159 - 模糊测试技术

> ▎ 漏洞挖掘是方法论，模糊测试是具体技术——明天深入模糊测试技术。

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

---

*本文是 365 天信息安全技术系列的第 158 篇，漏洞与攻防系列第 3 篇，精编版本*

*漏洞与攻防系列 (Day 156-165) 继续！*