parse_result.py

import xml.etree.ElementTree as ET
import os
import sys
import argparse
from datetime import datetime
try:
    import javalang
except ImportError:
    print("Warning: javalang library not found. Source code extraction may be limited.")
    javalang = None

# 加载Source-Sink分类配置
def load_sink_categories(file_path):
    categories = {}
    current_category = "其他漏洞"
    
    if not os.path.exists(file_path):
        print(f"Warning: SourcesAndSinks file not found at {file_path}")
        return categories
        
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                    
                if line.startswith('#'):
                    # 提取注释作为分类，去除 # 和可能的说明
                    comment = line.lstrip('#').strip()
                    # 去除括号及后面的内容（通常是说明）
                    if '（' in comment:
                        comment = comment.split('（')[0]
                    elif '(' in comment:
                        comment = comment.split('(')[0]
                    
                    # 忽略Source的注释头
                    if "Source" in comment or "数据源点" in comment or "检测规则" in comment:
                        continue
                        
                    current_category = comment.strip()
                    
                elif '-> _SINK_' in line:
                    # 提取方法签名
                    parts = line.split('->')
                    if len(parts) > 0:
                        method_sig = parts[0].strip()
                        categories[method_sig] = current_category
    except Exception as e:
        print(f"Error loading sink categories: {e}")
        
    return categories

# 漏洞类型分类
def classify_vulnerability(sink_method, sink_def, sink_categories=None):
    vuln_type = '其他漏洞'
    severity = '低'
    # 1. 优先使用配置文件中的分类 (Source-Sink映射)
    if sink_categories and sink_def in sink_categories:
        vuln_type = sink_categories[sink_def]
    else:
        # 如果找不到映射，不再尝试任何启发式分类，直接标记为其他漏洞
        # 这样确保所有分类都来自Source-Sink文件的定义
        pass
    # 3. 确定风险等级
    if any(k in vuln_type for k in ['命令执行', '反射', '文件读写', 'Intent重定向']):
        severity = '高'
    elif any(k in vuln_type for k in ['广播漏洞', 'LaunchAnyWhere', '沙箱文件', 'WebView', 'Provider']):
        severity = '中'
    return vuln_type, severity

# 生成修复建议
def generate_fix_advice(vuln_type):
    advice_map = {
        'Provider漏洞': [
            '对Uri的lastPathSegment进行严格验证',
            '限制可访问的文件路径范围',
            '实施文件权限检查',
            '使用应用专用目录而非外部存储'
        ],
        'Intent重定向': [
            '对deeplink参数进行白名单验证',
            '限制可重定向的URL范围',
            '实施Intent校验机制',
            '使用显式Intent而非隐式Intent'
        ],
        'LaunchAnyWhere漏洞': [
            '使用显式Intent指定组件',
            '添加Intent校验逻辑',
            '限制组件的导出性',
            '实施权限检查'
        ],
        'WebView漏洞': [
            '避免使用addJavascriptInterface',
            '启用JavaScript时要谨慎',
            '设置合理的WebSettings配置',
            '使用WebView安全最佳实践'
        ],
        '文件读写漏洞': [
            '避免使用外部存储存储敏感数据',
            '验证文件路径以防止路径遍历',
            '使用应用私有目录',
            '对写入的文件内容进行加密'
        ],
        '广播漏洞': [
            '限制广播接收者的导出属性',
            '对接收到的Intent进行校验',
            '使用LocalBroadcastManager',
            '设置广播权限'
        ],
        '沙箱文件篡改漏洞': [
            '避免在SharedPreferences中存储敏感明文',
            '使用EncryptedSharedPreferences',
            '正确设置文件访问模式（MODE_PRIVATE）',
            '定期清理敏感数据'
        ],
        'Settings Provider篡改漏洞': [
            '避免修改系统全局设置',
            '检查修改设置的权限',
            '验证写入设置的值',
            '遵循Android安全最佳实践'
        ],
        '命令执行漏洞': [
            '避免使用Runtime.exec等危险方法',
            '对输入参数进行严格验证',
            '使用安全的替代方案',
            '实施最小权限原则'
        ],
        '反射导致RCE漏洞': [
            '避免使用反射执行外部可控代码',
            '对反射调用的目标类和方法进行白名单验证',
            '限制反射调用的权限',
            '使用非反射的替代实现'
        ],
        '其他漏洞': [
            '对输入进行严格验证',
            '实施最小权限原则',
            '遵循安全编码最佳实践',
            '定期进行安全测试'
        ]
    }
    return advice_map.get(vuln_type, advice_map['其他漏洞'])

# 从方法签名中提取类名和方法名
def extract_method_info(method_signature):
    if not method_signature:
        return 'Unknown', 'unknown()', 'void'
    
    # 解析方法签名，格式为 <class_name: return_type method_name(parameters)>
    try:
        # 移除尖括号
        if method_signature.startswith('<') and method_signature.endswith('>'):
            method_signature = method_signature[1:-1]
        
        # 分割类名和方法部分
        parts = method_signature.split(': ')
        if len(parts) >= 2:
            class_name = parts[0].strip()
            method_part = parts[1].strip()
            
            # 提取方法名和返回类型
            method_components = method_part.split('(')
            if len(method_components) >= 2:
                method_decl = method_components[0].strip()
                params = method_components[1].rstrip(')')
                
                # 提取返回类型和方法名
                method_parts = method_decl.split()
                if len(method_parts) >= 2:
                    return_type = ' '.join(method_parts[:-1])
                    method_name = method_parts[-1]
                    # 简化参数类型
                    simplified_params = simplify_param_types(params)
                    return class_name, f"{method_name}({simplified_params})", return_type
                elif len(method_parts) == 1:
                    return_type = 'void'
                    method_name = method_parts[0]
                    # 简化参数类型
                    simplified_params = simplify_param_types(params)
                    return class_name, f"{method_name}({simplified_params})", return_type
    except:
        pass
    
    return 'Unknown', 'unknown()', 'void'

# 简化参数类型
def simplify_param_types(params):
    if not params:
        return ''
    
    # 简化常见类型
    simplified = params
    simplified = simplified.replace('java.lang.String', 'String')
    simplified = simplified.replace('java.lang.int', 'int')
    simplified = simplified.replace('java.lang.long', 'long')
    simplified = simplified.replace('java.lang.float', 'float')
    simplified = simplified.replace('java.lang.double', 'double')
    simplified = simplified.replace('java.lang.boolean', 'boolean')
    simplified = simplified.replace('java.lang.Object', 'Object')
    simplified = simplified.replace('android.content.Intent', 'Intent')
    simplified = simplified.replace('android.net.Uri', 'Uri')
    simplified = simplified.replace('android.os.Bundle', 'Bundle')
    simplified = simplified.replace('android.view.View', 'View')
    
    return simplified

# 简化FlowDroid语句为更可读的形式
def simplify_statement(statement):
    # 替换FlowDroid特有的语法
    simplified = statement
    
    # 替换变量表示
    simplified = simplified.replace('$r', 'var')
    simplified = simplified.replace('r0', 'this')
    simplified = simplified.replace('r1', 'param1')
    simplified = simplified.replace('r2', 'param2')
    simplified = simplified.replace('r3', 'param3')
    simplified = simplified.replace('r4', 'param4')
    simplified = simplified.replace('r5', 'param5')
    
    # 替换方法调用语法
    simplified = simplified.replace('virtualinvoke ', '')
    simplified = simplified.replace('interfaceinvoke ', '')
    simplified = simplified.replace('staticinvoke ', '')
    simplified = simplified.replace('specialinvoke ', '')
    
    # 替换类型转换
    simplified = simplified.replace('(android.widget.TextView) ', '(TextView) ')
    simplified = simplified.replace('(android.content.Intent) ', '(Intent) ')
    simplified = simplified.replace('(android.view.View) ', '(View) ')
    simplified = simplified.replace('(android.net.Uri) ', '(Uri) ')
    simplified = simplified.replace('(android.os.Bundle) ', '(Bundle) ')
    
    # 简化类名
    simplified = simplified.replace('android.content.', '')
    simplified = simplified.replace('android.widget.', '')
    simplified = simplified.replace('android.view.', '')
    simplified = simplified.replace('android.net.', '')
    simplified = simplified.replace('android.os.', '')
    simplified = simplified.replace('android.webkit.', '')
    simplified = simplified.replace('java.lang.', '')
    
    # 限制长度
    if len(simplified) > 100:
        simplified = simplified[:97] + '...'
    
    return simplified

# 清理Java语句，使其更像标准Java
def clean_java_statement(statement):
    # 替换一些常见的FlowDroid特定语法
    cleaned = statement
    
    # 修复方法调用语法
    cleaned = cleaned.replace('vaparam', 'var')
    cleaned = cleaned.replace('param.', '.')
    cleaned = cleaned.replace('this.', 'this.')
    
    # 确保语句以分号结尾
    if not cleaned.endswith(';') and not cleaned.endswith('...'):
        cleaned += ';'
    
    return cleaned

# 查找Java源文件
def find_java_source_file(class_name, source_dir):
    if not source_dir or not os.path.exists(source_dir):
        return None
    try:
        target_name = class_name.split('.')[-1] + '.java'
        for root, dirs, files in os.walk(source_dir):
            if target_name in files:
                return os.path.join(root, target_name)
    except:
        pass
    return None

# 读取源代码指定行
def read_source_code(file_path, line_num):
    try:
        if not os.path.exists(file_path):
            return None
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
            if 0 < int(line_num) <= len(lines):
                start = max(0, int(line_num) - 3)
                end = min(len(lines), int(line_num) + 2)
                return "".join([f"{'-> ' if (i+1)==int(line_num) else '   '}{i+1}: {lines[i]}" for i in range(start, end)])
    except:
        pass
    return None

# 提取真实源代码
def extract_real_source_code(file_path, method_signature, target_line=None):
    if not file_path or not os.path.exists(file_path):
        return None
        
    if not javalang:
        return read_source_code(file_path, target_line) if target_line else None

    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            
        tree = javalang.parse.parse(content)
        class_name, method_part, _ = extract_method_info(method_signature)
        method_name = method_part.split('(')[0]
        
        target_method = None
        for path, node in tree.filter(javalang.tree.MethodDeclaration):
            if node.name == method_name:
                target_method = node
                break
        
        if not target_method and method_name == '<init>':
             for path, node in tree.filter(javalang.tree.ConstructorDeclaration):
                target_method = node
                break

        if target_method:
            start_line = target_method.position.line
            lines = content.split('\n')
            open_braces = 0
            end_line = start_line
            found_start = False
            for i in range(start_line - 1, len(lines)):
                line = lines[i]
                open_braces += line.count('{')
                open_braces -= line.count('}')
                if open_braces > 0: found_start = True
                if found_start and open_braces == 0:
                    end_line = i + 1
                    break
            
            code_snippet = []
            for i in range(start_line - 1, end_line):
                prefix = "-> " if target_line and (i + 1) == int(target_line) else "   "
                code_snippet.append(f"{prefix}{i+1}: {lines[i]}")
            return "".join([line + "\n" for line in code_snippet])
            
        if target_line:
            return read_source_code(file_path, target_line)
            
    except Exception as e:
        if target_line:
            return read_source_code(file_path, target_line)
            
    return None

# 生成源代码上下文
def generate_source_context(sink_method, taint_paths, source_dir=None, sink_line=None):
    context = ""
    
    # 尝试提取真实源代码
    if source_dir:
        # 1. 收集路径中涉及的所有方法
        methods_to_extract = []
        extracted_methods = set()
        
        # 优先提取污点路径中的方法（Source -> Intermediate -> Sink）
        if taint_paths:
            for taint_path in taint_paths:
                path_elements = taint_path.findall('.//PathElement')
                for element in path_elements:
                    method_sig = element.get('Method')
                    # 过滤掉 android.* 和 java.* 等系统方法，只保留应用代码
                    if method_sig and not method_sig.startswith('<android.') and not method_sig.startswith('<java.'):
                        if method_sig not in extracted_methods:
                            extracted_methods.add(method_sig)
                            methods_to_extract.append(method_sig)
        
        # 确保 Sink 方法也被提取（如果不在路径中且是应用代码）
        if sink_method and sink_method not in extracted_methods:
            if not sink_method.startswith('<android.') and not sink_method.startswith('<java.'):
                extracted_methods.add(sink_method)
                methods_to_extract.append(sink_method)
            
        # 2. 依次提取代码
        for method_sig in methods_to_extract:
            class_name, _, _ = extract_method_info(method_sig)
            source_file = find_java_source_file(class_name, source_dir)
            
            if source_file:
                # 传递 sink_line 给 extract_real_source_code，如果是 Sink 方法
                current_line = sink_line if method_sig == sink_method else None
                real_code = extract_real_source_code(source_file, method_sig, current_line)
                
                if real_code:
                    context += f"// 真实源代码: {class_name} (从反编译代码提取)\n"
                    context += f"// 方法: {method_sig}\n"
                    if current_line:
                        context += f"// 漏洞行号: {current_line}\n"
                    context += real_code
                    context += "\n" # 方法之间空行分隔

        if context:
            return context

    # 如果无法提取真实源代码，回退到伪代码生成
    class_name, method_signature, return_type = extract_method_info(sink_method)
    
    # 生成方法签名
    context = f"// {class_name}.{method_signature} 方法"
    context += "\n"
    context += f"public class {class_short_name} {{"
    context += "\n"
    context += "    // 漏洞所在方法"
    context += "\n"
    context += f"    {return_type} {method_signature} {{"
    context += "\n"
    
    # 添加污点传播路径的关键步骤
    if taint_paths:
        context += "        // 污点传播路径关键步骤"
        context += "\n"
        for taint_path in taint_paths:
            path_elements = taint_path.findall('.//PathElement')
            for element in path_elements[:5]:
                statement = element.get('Statement', 'N/A')
                # 简化语句，使其更像Java代码
                simplified_stmt = simplify_statement(statement)
                # 进一步清理语句，使其更像标准Java
                cleaned_stmt = clean_java_statement(simplified_stmt)
                context += f"        {cleaned_stmt}"
                context += "\n"
            if len(path_elements) > 5:
                context += "        // ... 省略其他步骤 ..."
                context += "\n"
    
    # 添加漏洞标记
    context += "        // 【漏洞点】此处存在安全漏洞"
    context += "\n"
    
    context += "    }"
    context += "\n"
    context += "}"
    
    return context

# 解析XML结果文件并生成markdown报告
def parse_flowdroid_result(input_file, output_file, source_dir=None):
    if not os.path.exists(input_file):
        print(f"错误: 文件 {input_file} 不存在")
        return 1
    
    # 加载Sink分类
    script_dir = os.path.dirname(os.path.abspath(__file__))
    sink_file = os.path.join(script_dir, 'SourcesAndSinks.txt')
    sink_categories = load_sink_categories(sink_file)
    if sink_categories:
        print(f"已加载 {len(sink_categories)} 条Sink分类规则")
    
    try:
        # 解析XML文件
        tree = ET.parse(input_file)
        root = tree.getroot()
        
        # 生成markdown报告
        markdown_content = []
        
        # 报告标题
        markdown_content.append('# 移动应用安全分析报告')
        markdown_content.append('')
        
        # 报告基本信息
        markdown_content.append('## 报告信息')
        markdown_content.append('| 项目 | 内容 |')
        markdown_content.append('|------|------|')
        markdown_content.append(f'| 分析时间 | {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} |')
        markdown_content.append('| 分析工具 | FlowDroid |')
        markdown_content.append('| 分析目标 | ovaa.apk |')
        
        # 获取终止状态
        termination_state = root.get('TerminationState')
        markdown_content.append(f'| 分析状态 | {termination_state} |')
        markdown_content.append('')
        
        # 统计结果数量
        results = root.findall('.//Result')
        markdown_content.append('## 漏洞摘要')
        markdown_content.append('| 漏洞类型 | 风险等级 | 数量 |')
        markdown_content.append('|----------|----------|------|')
        
        # 统计漏洞类型
        vuln_stats = {}
        for result in results:
            sink = result.find('.//Sink')
            if sink is not None:
                sink_method = sink.get('Method', '')
                sink_def = sink.get('MethodSourceSinkDefinition', '')
                vuln_type, severity = classify_vulnerability(sink_method, sink_def, sink_categories)
                key = (vuln_type, severity)
                vuln_stats[key] = vuln_stats.get(key, 0) + 1
        
        # 输出漏洞统计
        total_vulns = 0
        for (vuln_type, severity), count in vuln_stats.items():
            markdown_content.append(f'| {vuln_type} | {severity} | {count} |')
            total_vulns += count
        
        markdown_content.append(f'| **总计** | - | **{total_vulns}** |')
        markdown_content.append('')
        
        # 详细漏洞信息
        if results:
            markdown_content.append('## 详细漏洞信息')
            markdown_content.append('')
            
            # 遍历所有结果
            for i, result in enumerate(results, 1):
                # 提取sink信息
                sink = result.find('.//Sink')
                if sink is not None:
                    sink_method = sink.get('Method', 'N/A')
                    sink_statement = sink.get('Statement', 'N/A')
                    sink_def = sink.get('MethodSourceSinkDefinition', 'N/A')
                    sink_line = sink.get('LineNumber', None)
                    
                    # 分类漏洞
                    vuln_type, severity = classify_vulnerability(sink_method, sink_def, sink_categories)
                    
                    # 生成漏洞标题
                    severity_icon = '🔴' if severity == '高' else '🟡' if severity == '中' else '🟢'
                    markdown_content.append(f'### {severity_icon} 漏洞 #{i}: {vuln_type} ({severity}风险)')
                    markdown_content.append('')
                    
                    # 漏洞详情
                    markdown_content.append('#### 漏洞详情')
                    markdown_content.append('| 项目 | 内容 |')
                    markdown_content.append('|------|------|')
                    markdown_content.append(f'| 漏洞类型 | {vuln_type} |')
                    markdown_content.append(f'| 风险等级 | {severity} |')
                    markdown_content.append(f'| 影响方法 | {sink_method} |')
                    markdown_content.append(f'| 触发语句 | `{sink_statement}` |')
                    markdown_content.append(f'| 漏洞定义 | {sink_def} |')
                    markdown_content.append('')
                    
                    # 提取source信息
                    sources = result.findall('.//Source')
                    if sources:
                        markdown_content.append('#### 攻击路径')
                        markdown_content.append('| 序号 | 方法 | 语句 |')
                        markdown_content.append('|------|------|------|')
                        
                        for j, source in enumerate(sources, 1):
                            source_method = source.get('Method', 'N/A')
                            source_statement = source.get('Statement', 'N/A')
                            markdown_content.append(f'| {j} | {source_method} | `{source_statement}` |')
                        markdown_content.append('')
                    
                    # 提取污点传播路径
                    taint_paths = result.findall('.//TaintPath')
                    if taint_paths:
                        markdown_content.append('#### 污点传播路径')
                        markdown_content.append('```')
                        
                        for j, taint_path in enumerate(taint_paths, 1):
                            path_elements = taint_path.findall('.//PathElement')
                            markdown_content.append(f'路径 #{j}:')
                            
                            # 只显示前5个步骤
                            for k, element in enumerate(path_elements[:5], 1):
                                statement = element.get('Statement', 'N/A')
                                method = element.get('Method', 'N/A')
                                markdown_content.append(f'  {k}. {statement}')
                                markdown_content.append(f'     方法: {method}')
                            
                            if len(path_elements) > 5:
                                markdown_content.append(f'  ... 省略 {len(path_elements) - 5} 个步骤')
                            markdown_content.append('')
                        
                        markdown_content.append('```')
                        markdown_content.append('')
                    
                    # 生成源代码上下文
                    markdown_content.append('#### 源代码上下文')
                    markdown_content.append('```java')
                    if sink_statement != 'N/A':
                        markdown_content.append(f'// 触发语句: {sink_statement}')
                    source_context = generate_source_context(sink_method, taint_paths, source_dir, sink_line)
                    markdown_content.append(source_context)
                    markdown_content.append('```')
                    markdown_content.append('')
                    
                    # 生成修复建议
                    fix_advice = generate_fix_advice(vuln_type)
                    markdown_content.append('#### 修复建议')
                    markdown_content.append('1. ' + fix_advice[0])
                    markdown_content.append('2. ' + fix_advice[1])
                    markdown_content.append('3. ' + fix_advice[2])
                    markdown_content.append('4. ' + fix_advice[3])
                    markdown_content.append('')
                    markdown_content.append('---')
                    markdown_content.append('')
        
        # 安全建议
        markdown_content.append('## 整体安全建议')
        markdown_content.append('1. **输入验证**: 对所有外部输入进行严格验证')
        markdown_content.append('2. **权限管理**: 仅申请必要的权限')
        markdown_content.append('3. **数据存储**: 敏感数据应加密存储')
        markdown_content.append('4. **组件安全**: 限制组件的导出性')
        markdown_content.append('5. **WebView安全**: 谨慎使用WebView，避免启用危险设置')
        markdown_content.append('6. **Intent安全**: 使用显式Intent，实施Intent校验')
        markdown_content.append('7. **代码审计**: 定期进行安全代码审计')
        markdown_content.append('8. **依赖管理**: 及时更新第三方依赖')
        markdown_content.append('')
        
        # 性能数据
        performance_data = root.find('.//PerformanceData')
        if performance_data is not None:
            markdown_content.append('## 分析性能数据')
            markdown_content.append('| 指标 | 值 |')
            markdown_content.append('|------|------|')
            
            for entry in performance_data.findall('.//PerformanceEntry'):
                name = entry.get('Name')
                value = entry.get('Value')
                markdown_content.append(f'| {name} | {value} |')
            markdown_content.append('')
        
        # 确保输出目录存在
        output_dir = os.path.dirname(output_file)
        if output_dir and not os.path.exists(output_dir):
            os.makedirs(output_dir, exist_ok=True)
        
        # 写入报告文件
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write('\n'.join(markdown_content))
        
        print(f"安全分析报告已生成: {output_file}")
        print(f"共检测到 {total_vulns} 个漏洞")
        return 0
        
    except ET.ParseError as e:
        print(f"解析XML文件时出错: {e}")
        return 1
    except Exception as e:
        print(f"处理结果时出错: {e}")
        return 1

if __name__ == "__main__":
    # 解析命令行参数
    parser = argparse.ArgumentParser(description='解析FlowDroid结果并生成安全分析报告')
    parser.add_argument('-i', '--input', type=str, default='analysis_result.xml',
                      help='FlowDroid分析结果XML文件路径')
    parser.add_argument('-o', '--output', type=str, default='security_analysis_report.md',
                      help='生成的安全分析报告路径')
    parser.add_argument('-s', '--source', type=str, default=None,
                      help='反编译后的源代码目录')
    
    args = parser.parse_args()
    
    # 执行解析和报告生成
    exit_code = parse_flowdroid_result(args.input, args.output, args.source)
    sys.exit(exit_code)