code-analysis/Report-Generation/cppcheck_test_generator/verification.py

"""
验证和测试模块
"""
import subprocess
import time
import json
from pathlib import Path
from typing import List, Optional

from .models import CppcheckIssue


def verify_single_test(cpp_file: Path, timeout: int = 30, project_root: Optional[Path] = None, include_dirs: List[str] = None) -> dict:
    """验证单个测试用例"""
    result = {
        "file": cpp_file.name,
        "compiled": False,
        "executed": False,
        "exit_code": None,
        "output": "",
        "error": "",
        "duration": 0,
        "timeout": False,
        "vulnerability_confirmed": False,
        "vulnerability_type": "unknown"
    }
    
    exe_file = cpp_file.with_suffix(".exe")
    
    try:
        # 编译
        start_time = time.time()
        compile_cmd = [
            "g++", "-std=c++17", "-O0", "-g", "-Wall", "-Wextra", "-pedantic"
        ]
        
        # 添加项目相关的编译选项
        if project_root:
            compile_cmd.extend(["-I", str(project_root)])
        
        if include_dirs:
            for include_dir in include_dirs:
                compile_cmd.extend(["-I", include_dir])
        
        compile_cmd.extend(["-o", str(exe_file), str(cpp_file)])
        
        compile_result = subprocess.run(
            compile_cmd, 
            capture_output=True, 
            text=True, 
            timeout=timeout
        )
        
        result["compiled"] = (compile_result.returncode == 0)
        result["duration"] = time.time() - start_time
        
        if not result["compiled"]:
            result["error"] = compile_result.stderr
            return result
        
        # 执行
        if exe_file.exists():
            start_time = time.time()
            try:
                execute_result = subprocess.run(
                    [str(exe_file)], 
                    capture_output=True, 
                    text=True, 
                    timeout=timeout
                )
                
                result["executed"] = True
                result["exit_code"] = execute_result.returncode
                result["output"] = execute_result.stdout
                result["error"] = execute_result.stderr
                result["duration"] = time.time() - start_time
                
                # 分析漏洞类型
                result["vulnerability_type"] = analyze_vulnerability_type(cpp_file.name, result)
                result["vulnerability_confirmed"] = determine_vulnerability_confirmed(result)
                
            except subprocess.TimeoutExpired:
                result["timeout"] = True
                result["error"] = f"执行超时（{timeout}秒）"
            except Exception as e:
                result["error"] = f"执行异常: {str(e)}"
    
    except subprocess.TimeoutExpired:
        result["timeout"] = True
        result["error"] = f"编译超时（{timeout}秒）"
    except Exception as e:
        result["error"] = f"编译异常: {str(e)}"
    finally:
        # 清理
        if exe_file.exists():
            exe_file.unlink()
    
    return result


def analyze_vulnerability_type(filename: str, result: dict) -> str:
    """分析漏洞类型"""
    filename_lower = filename.lower()
    
    if "uninitvar" in filename_lower:
        return "未初始化变量"
    elif "memleak" in filename_lower:
        return "内存泄漏"
    elif "nullpointer" in filename_lower:
        return "空指针解引用"
    elif "arrayindex" in filename_lower:
        return "数组越界"
    elif "buffer" in filename_lower:
        return "缓冲区溢出"
    elif "useafterfree" in filename_lower:
        return "释放后使用"
    elif "doublefree" in filename_lower:
        return "重复释放"
    else:
        return "未知类型"


def determine_vulnerability_confirmed(result: dict) -> bool:
    """判断漏洞是否被确认"""
    if not result["compiled"] or not result["executed"]:
        return False
    
    # 根据漏洞类型和程序行为判断
    vuln_type = result["vulnerability_type"]
    exit_code = result["exit_code"]
    output = result["output"]
    error = result["error"]
    
    if "未初始化变量" in vuln_type:
        # 未初始化变量：程序应该能正常运行，但输出随机值
        return exit_code == 0 and "buffer" in output.lower()
    
    elif "内存泄漏" in vuln_type:
        # 内存泄漏：程序应该能正常运行
        return exit_code == 0
    
    elif "空指针解引用" in vuln_type:
        # 空指针解引用：程序应该崩溃
        return exit_code != 0 or "segmentation fault" in error.lower()
    
    elif "数组越界" in vuln_type:
        # 数组越界：程序可能崩溃
        return exit_code != 0 or "segmentation fault" in error.lower()
    
    elif "缓冲区溢出" in vuln_type:
        # 缓冲区溢出：程序可能崩溃
        return exit_code != 0 or "stack smashing" in error.lower()
    
    else:
        # 默认：程序能正常运行就认为漏洞存在
        return exit_code == 0


def verify_test_case(test_file_path: Path, issue: CppcheckIssue) -> dict:
    """验证生成的测试用例是否能有效触发cppcheck检测"""
    verification_result = {
        'compiles': False,
        'runs': False,
        'triggers_cppcheck': False,
        'cppcheck_warnings': [],
        'compilation_errors': [],
        'runtime_errors': []
    }
    
    try:
        # 1. 尝试编译测试用例
        import tempfile
        
        # 创建临时目录
        with tempfile.TemporaryDirectory() as temp_dir:
            temp_cpp = Path(temp_dir) / "test.cpp"
            temp_exe = Path(temp_dir) / "test"
            
            # 复制测试文件到临时目录
            with open(test_file_path, 'r', encoding='utf-8') as f:
                test_content = f.read()
            
            with open(temp_cpp, 'w', encoding='utf-8') as f:
                f.write(test_content)
            
            # 尝试编译
            try:
                result = subprocess.run(
                    ['g++', '-std=c++17', '-o', str(temp_exe), str(temp_cpp)],
                    capture_output=True, text=True, timeout=30
                )
                
                if result.returncode == 0:
                    verification_result['compiles'] = True
                    
                    # 2. 尝试运行
                    try:
                        run_result = subprocess.run(
                            [str(temp_exe)],
                            capture_output=True, text=True, timeout=10
                        )
                        if run_result.returncode == 0:
                            verification_result['runs'] = True
                        else:
                            verification_result['runtime_errors'].append(run_result.stderr)
                    except subprocess.TimeoutExpired:
                        verification_result['runtime_errors'].append("Runtime timeout")
                    except Exception as e:
                        verification_result['runtime_errors'].append(str(e))
                else:
                    verification_result['compilation_errors'].append(result.stderr)
                    
            except subprocess.TimeoutExpired:
                verification_result['compilation_errors'].append("Compilation timeout")
            except Exception as e:
                verification_result['compilation_errors'].append(str(e))
            
            # 3. 使用cppcheck检查
            try:
                cppcheck_result = subprocess.run(
                    ['cppcheck', '--enable=all', '--std=c++17', str(temp_cpp)],
                    capture_output=True, text=True, timeout=30
                )
                
                if cppcheck_result.returncode != 0 or cppcheck_result.stderr:
                    # 解析cppcheck输出
                    output = cppcheck_result.stderr
                    if issue.id.lower() in output.lower():
                        verification_result['triggers_cppcheck'] = True
                    
                    # 提取警告信息
                    lines = output.split('\n')
                    for line in lines:
                        if 'warning:' in line or 'error:' in line:
                            verification_result['cppcheck_warnings'].append(line.strip())
                            
            except subprocess.TimeoutExpired:
                verification_result['cppcheck_warnings'].append("cppcheck timeout")
            except Exception as e:
                verification_result['cppcheck_warnings'].append(f"cppcheck error: {str(e)}")
                
    except Exception as e:
        verification_result['compilation_errors'].append(f"Verification error: {str(e)}")
    
    return verification_result


def auto_verify_tests(output_dir: Path, timeout: int = 30, project_root: Optional[Path] = None, include_dirs: List[str] = None) -> dict:
    """自动验证所有测试用例"""
    print("开始自动验证测试用例...")
    
    cpp_files = list(output_dir.glob("*.cpp"))
    if not cpp_files:
        print("未找到测试用例文件")
        return {"total": 0, "results": [], "summary": {}}
    
    results = []
    for i, cpp_file in enumerate(cpp_files, 1):
        print(f"验证 [{i}/{len(cpp_files)}]: {cpp_file.name}")
        result = verify_single_test(cpp_file, timeout, project_root, include_dirs)
        results.append(result)
        
        # 显示验证结果
        if result["vulnerability_confirmed"]:
            print(f"  ✓ 漏洞确认: {result['vulnerability_type']}")
        elif result["compiled"] and result["executed"]:
            print(f"  - 程序正常: {result['vulnerability_type']} (可能误报)")
        else:
            print(f"  ✗ 验证失败: {result['error']}")
    
    # 生成汇总统计
    summary = {
        "total": len(results),
        "compiled": sum(1 for r in results if r["compiled"]),
        "executed": sum(1 for r in results if r["executed"]),
        "vulnerabilities_confirmed": sum(1 for r in results if r["vulnerability_confirmed"]),
        "timeouts": sum(1 for r in results if r["timeout"]),
        "errors": sum(1 for r in results if not r["compiled"] or not r["executed"])
    }
    
    return {"total": len(results), "results": results, "summary": summary}


def generate_verification_report(output_dir: Path, verification_results: dict) -> Path:
    """生成验证结果报告"""
    report_path = output_dir / "vulnerability_verification_report.md"
    
    results = verification_results["results"]
    summary = verification_results["summary"]
    
    # 按漏洞类型分组
    vuln_groups = {}
    for result in results:
        vuln_type = result["vulnerability_type"]
        if vuln_type not in vuln_groups:
            vuln_groups[vuln_type] = []
        vuln_groups[vuln_type].append(result)
    
    # 生成报告内容
    report_content = f"""# 漏洞验证结果报告

## 验证汇总

- **总测试用例**: {summary['total']}
- **编译成功**: {summary['compiled']}
- **执行成功**: {summary['executed']}
- **漏洞确认**: {summary['vulnerabilities_confirmed']}
- **验证超时**: {summary['timeouts']}
- **验证错误**: {summary['errors']}

## 漏洞确认列表

"""
    
    # 按漏洞类型生成详细报告
    for vuln_type, vuln_results in vuln_groups.items():
        confirmed_count = sum(1 for r in vuln_results if r["vulnerability_confirmed"])
        total_count = len(vuln_results)
        
        report_content += f"### {vuln_type} ({confirmed_count}/{total_count} 确认)\n\n"
        
        for result in vuln_results:
            status = "✓ 确认" if result["vulnerability_confirmed"] else "✗ 未确认"
            report_content += f"- **{result['file']}**: {status}\n"
            
            if result["vulnerability_confirmed"]:
                report_content += f"  - 返回码: {result['exit_code']}\n"
                if result["output"]:
                    report_content += f"  - 输出: {result['output'][:100]}...\n"
            elif result["error"]:
                report_content += f"  - 错误: {result['error']}\n"
            
            report_content += "\n"
    
    # 添加修复建议
    report_content += """## 修复建议

### 确认的漏洞
以下漏洞已被验证确认，建议优先修复：

"""
    
    for vuln_type, vuln_results in vuln_groups.items():
        confirmed_results = [r for r in vuln_results if r["vulnerability_confirmed"]]
        if confirmed_results:
            report_content += f"#### {vuln_type}\n"
            for result in confirmed_results:
                report_content += f"- {result['file']}: 需要修复\n"
            report_content += "\n"
    
    report_content += """### 未确认的问题
以下问题可能是误报或需要进一步分析：

"""
    
    for vuln_type, vuln_results in vuln_groups.items():
        unconfirmed_results = [r for r in vuln_results if not r["vulnerability_confirmed"]]
        if unconfirmed_results:
            report_content += f"#### {vuln_type}\n"
            for result in unconfirmed_results:
                report_content += f"- {result['file']}: 需要进一步分析\n"
            report_content += "\n"
    
    # 写入报告文件
    report_path.write_text(report_content, encoding="utf-8")
    return report_path


def generate_json_report(output_dir: Path, verification_results: dict) -> Path:
    """生成JSON格式的详细报告"""
    json_path = output_dir / "verification_results.json"
    
    # 添加时间戳
    verification_results["timestamp"] = str(Path().cwd())
    verification_results["generated_at"] = str(Path().cwd())
    
    # 写入JSON文件
    json_path.write_text(json.dumps(verification_results, indent=2, ensure_ascii=False), encoding="utf-8")
    return json_path