093_数字取证高级技术：高级恶意软件分析与动态行为监控实战指南——从静态分析到内存取证的全面威胁识别方法

物理隔离：使用独立的物理机器进行分析
虚拟隔离：使用虚拟机技术，支持快照和恢复
网络隔离：设置虚拟网络，控制对外连接
沙箱环境：使用专业沙箱解决方案如Cuckoo Sandbox
容器隔离：使用Docker等容器技术隔离分析进程

Windows分析环境：
- Windows 10/11企业版（禁用自动更新）
- 4-8GB内存
- 50-100GB磁盘空间
- 快照功能启用
- 网络配置为仅主机模式或内部网络

Linux分析环境：
- Ubuntu/Debian LTS版本
- 4GB内存
- 50GB磁盘空间
- 网络隔离配置

DOS头(MZ Header)
DOS存根(DOS Stub)
PE文件头(PE Header)
文件头(File Header)
可选头(Optional Header)
节表(Section Table)
节数据(Section Data)
导入表(Import Table)
导出表(Export Table)
资源表(Resource Table)
延迟加载表(Delay Load Table)
TLS表(Thread Local Storage)

import pefile

def analyze_pe_file(filename):
    try:
        pe = pefile.PE(filename)
        
        # 提取基本信息
        basic_info = {
            'filename': filename,
            'md5': pe.get_imphash(),
            'compilation_time': pe.FILE_HEADER.TimeDateStamp,
            'subsystem': pe.OPTIONAL_HEADER.Subsystem,
            'entry_point': hex(pe.OPTIONAL_HEADER.AddressOfEntryPoint),
            'image_base': hex(pe.OPTIONAL_HEADER.ImageBase),
            'sections': [],
            'imports': {},
            'exports': [],
            'resources': []
        }
        
        # 分析节表
        for section in pe.sections:
            section_info = {
                'name': section.Name.decode('utf-8').strip('\x00'),
                'virtual_address': hex(section.VirtualAddress),
                'virtual_size': hex(section.Misc_VirtualSize),
                'raw_size': hex(section.SizeOfRawData),
                'characteristics': hex(section.Characteristics),
                'entropy': section.get_entropy()
            }
            basic_info['sections'].append(section_info)
        
        # 分析导入函数
        if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'):
            for entry in pe.DIRECTORY_ENTRY_IMPORT:
                dll_name = entry.dll.decode('utf-8')
                basic_info['imports'][dll_name] = []
                for imp in entry.imports:
                    if imp.name:
                        basic_info['imports'][dll_name].append(imp.name.decode('utf-8'))
                    else:
                        basic_info['imports'][dll_name].append(f'ordinal_{imp.ordinal}')
        
        # 分析导出函数
        if hasattr(pe, 'DIRECTORY_ENTRY_EXPORT'):
            for exp in pe.DIRECTORY_ENTRY_EXPORT.symbols:
                if exp.name:
                    basic_info['exports'].append(exp.name.decode('utf-8'))
        
        # 分析资源
        if hasattr(pe, 'DIRECTORY_ENTRY_RESOURCE'):
            for resource_type in pe.DIRECTORY_ENTRY_RESOURCE.entries:
                if hasattr(resource_type, 'directory'):
                    for resource_id in resource_type.directory.entries:
                        if hasattr(resource_id, 'directory'):
                            for resource_lang in resource_id.directory.entries:
                                basic_info['resources'].append({
                                    'type': resource_type.name,
                                    'id': resource_id.name,
                                    'lang': resource_lang.name
                                })
        
        return basic_info
        
    except Exception as e:
        print(f"Error analyzing PE file: {e}")
        return None

# 使用示例
result = analyze_pe_file('suspicious_file.exe')
print(result)

# 识别并解密XOR加密的字符串
def detect_xor_encrypted_strings(pe_file_path):
    import re
    
    # 使用strings命令提取所有ASCII字符串
    import subprocess
    result = subprocess.run(['strings', pe_file_path], capture_output=True, text=True)
    strings_output = result.stdout.split('\n')
    
    # 查找可能的XOR密钥（假设是单字节密钥）
    possible_strings = []
    
    # 尝试所有可能的单字节XOR密钥
    for key in range(1, 256):
        possible_decrypted = []
        
        # 分析PE文件的.data节或其他可能包含加密字符串的区域
        try:
            pe = pefile.PE(pe_file_path)
            for section in pe.sections:
                if b'.data' in section.Name:
                    data = section.get_data()
                    
                    # 尝试解密数据
                    decrypted = bytearray()
                    for b in data:
                        decrypted.append(b ^ key)
                    
                    # 提取解密后的字符串
                    decrypted_strings = re.findall(rb'[a-zA-Z]{4,}', decrypted)
                    for s in decrypted_strings:
                        try:
                            possible_decrypted.append(s.decode('utf-8'))
                        except:
                            pass
        except:
            pass
        
        # 如果找到合理数量的字符串，认为这个密钥可能有效
        if len(possible_decrypted) > 5:
            possible_strings.append({
                'key': hex(key),
                'strings': possible_decrypted[:10]  # 只取前10个示例
            })
    
    return possible_strings

# 使用示例
decrypted = detect_xor_encrypted_strings('malware.exe')
for item in decrypted:
    print(f"Possible XOR key: {item['key']}")
    print(f"Decrypted strings: {item['strings'][:5]}")
    print()

# 使用IDA Pro API进行自动分析的伪代码框架

def ida_analyze_function(function_ea):
    """
    分析IDA Pro中的函数并提取关键信息
    需要在IDA Pro Python环境中运行
    """
    import idautils
    import idc
    import idaapi
    
    function_info = {
        'name': idc.get_func_name(function_ea),
        'start_ea': function_ea,
        'end_ea': idaapi.get_func(function_ea).end_ea,
        'calls': [],
        'imports': [],
        'strings': [],
        'xrefs': []
    }
    
    # 提取函数内的调用
    for head in idautils.Heads(function_ea, idaapi.get_func(function_ea).end_ea):
        if idc.print_insn_mnem(head) == 'call':
            target = idc.get_operand_value(head, 0)
            target_name = idc.get_func_name(target) or hex(target)
            function_info['calls'].append({
                'address': hex(head),
                'target': target_name
            })
        
        # 查找字符串引用
        str_ea = idc.get_operand_value(head, 0)
        if idc.isASCII(str_ea):
            string = idc.get_strlit_contents(str_ea)
            if string:
                try:
                    function_info['strings'].append({
                        'address': hex(head),
                        'string': string.decode('utf-8')
                    })
                except:
                    pass
    
    # 提取函数的交叉引用
    for xref in idautils.XrefsTo(function_ea):
        function_info['xrefs'].append({
            'from': hex(xref.frm),
            'type': xref.type
        })
    
    return function_info

# 自动生成YARA规则的示例代码
def generate_yara_rule(sample_path, rule_name, description=""):
    import hashlib
    import re
    
    # 计算文件哈希
    with open(sample_path, 'rb') as f:
        file_data = f.read()
    
    md5_hash = hashlib.md5(file_data).hexdigest()
    sha1_hash = hashlib.sha1(file_data).hexdigest()
    sha256_hash = hashlib.sha256(file_data).hexdigest()
    
    # 提取可能的字符串特征
    strings = re.findall(rb'[a-zA-Z0-9_\-\.]{6,}', file_data)
    unique_strings = list(set(strings))
    # 过滤出可能有用的字符串（避免太短或太常见的）
    potential_signature_strings = []
    for s in unique_strings:
        try:
            decoded = s.decode('utf-8')
            # 避免纯数字和常见字符串
            if not decoded.isdigit() and len(decoded) > 8 and \
               not any(common in decoded.lower() for common in ['http', 'dll', 'exe', 'com']):
                potential_signature_strings.append(decoded)
                if len(potential_signature_strings) >= 3:
                    break
        except:
            pass
    
    # 提取文件头特征
    file_header = file_data[:64].hex()
    
    # 生成YARA规则
    yara_rule = f"rule {rule_name} {{
    meta:
        description = \"{description}\"
        md5 = \"{md5_hash}\"
        sha1 = \"{sha1_hash}\"
        sha256 = \"{sha256_hash}\"
        generated_date = \"2025-01-01\"
    
    strings:
        $header = {{{file_header}}}"
    
    for i, s in enumerate(potential_signature_strings):
        yara_rule += f"\n        $s{i+1} = \"{s}\""
    
    yara_rule += "\n    
    condition:
        $header and 1 of ($s*)\n}"
    
    return yara_rule

# 使用示例
yara_rule = generate_yara_rule('malware_sample.exe', 'Trojan_Banker_XY', 'Detects XY banking trojan')
print(yara_rule)

# Cuckoo Sandbox安装与配置（Ubuntu环境）

# 安装依赖
apt-get update
apt-get install -y python3 python3-pip python3-dev libffi-dev libssl-dev python3-virtualenv python3-setuptools git mongodb redis-server postgresql libpq-dev build-essential autoconf automake libtool libjpeg-dev zlib1g-dev swig

# 创建虚拟环境
virtualenv -p python3 venv
source venv/bin/activate

# 安装Cuckoo
pip install -U pip setuptools
git clone https://github.com/cuckoosandbox/cuckoo.git
cd cuckoo
pip install -e .

# 初始化Cuckoo
sudo mkdir -p /opt/cuckoo
sudo chown $USER:$USER /opt/cuckoo
cuckoo init
cuckoo community

# 配置虚拟机网络（使用VirtualBox）
# 1. 创建仅主机网络
VBoxManage hostonlyif create
VBoxManage hostonlyif ipconfig vboxnet0 --ip 192.168.56.1 --netmask 255.255.255.0

# 2. 配置iptables规则
iptables -A FORWARD -o eth0 -i vboxnet0 -s 192.168.56.0/24 -m conntrack --ctstate NEW -j ACCEPT
iptables -A FORWARD -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
iptables -A POSTROUTING -t nat -j MASQUERADE
sysctl -w net.ipv4.ip_forward=1

# 3. 配置Cuckoo网络
cuckoo config set physicalmachine.network.host "192.168.56.1"
cuckoo config set physicalmachine.network.netmask "255.255.255.0"

# 使用PyWin32监控进程创建和终止
def monitor_process_activity():
    import win32con
    import win32api
    import win32event
    import win32process
    import win32security
    import win32service
    import win32ts
    import threading
    import time
    
    # 创建进程创建事件
    process_create_event = win32event.CreateEvent(None, 0, 0, None)
    process_terminate_event = win32event.CreateEvent(None, 0, 0, None)
    
    # 注册事件通知
    handle = win32event.WaitForMultipleObjects(
        [process_create_event, process_terminate_event],
        False,
        win32event.INFINITE
    )
    
    def get_process_info(pid):
        try:
            handle = win32api.OpenProcess(
                win32con.PROCESS_QUERY_INFORMATION | win32con.PROCESS_VM_READ,
                False,
                pid
            )
            
            # 获取进程名称
            exe_name = win32process.GetModuleFileNameEx(handle, 0)
            
            # 获取命令行参数
            cmd_line = win32process.GetCommandLine(handle)
            
            # 获取创建时间
            times = win32process.GetProcessTimes(handle)
            create_time = times['CreationTime']
            
            win32api.CloseHandle(handle)
            
            return {
                'pid': pid,
                'exe_name': exe_name,
                'command_line': cmd_line,
                'create_time': create_time
            }
        except Exception as e:
            print(f"Error getting process info for PID {pid}: {e}")
            return None
    
    # 主监控循环
    print("Monitoring process activity...")
    while True:
        result = win32event.WaitForMultipleObjects(
            [process_create_event, process_terminate_event],
            False,
            5000
        )
        
        if result == win32event.WAIT_OBJECT_0:
            # 进程创建事件
            # 注意：在实际应用中，需要使用WMI或ETW来正确监控进程创建
            print("Process created event detected")
        elif result == win32event.WAIT_OBJECT_0 + 1:
            # 进程终止事件
            print("Process terminated event detected")
        
        time.sleep(0.1)

# 使用WMI监控进程活动（更可靠的方法）
def monitor_processes_with_wmi():
    import wmi
    import datetime
    
    c = wmi.WMI()
    
    # 创建进程监控
    process_watcher = c.Win32_Process.watch_for(
        notification_type="Creation",
        delay_secs=1
    )
    
    print("Monitoring process creation events...")
    while True:
        try:
            process = process_watcher()
            print(f"\nNew process created:")
            print(f"  Process ID: {process.ProcessId}")
            print(f"  Name: {process.Name}")
            print(f"  Command Line: {process.CommandLine}")
            print(f"  Parent PID: {process.ParentProcessId}")
            print(f"  Creation Time: {datetime.datetime.now()}")
            print(f"  Path: {process.ExecutablePath}")
            
            # 获取创建该进程的用户
            try:
                owner = process.GetOwner()
                print(f"  Owner: {owner[0]}\\{owner[2]}")
            except:
                print("  Owner: N/A")
                
        except Exception as e:
            print(f"Error in process monitoring: {e}")

# 启动监控
# monitor_process_activity()
# monitor_processes_with_wmi()

# 使用Python和Scapy捕获和分析网络流量
def capture_and_analyze_traffic(interface="eth0", packet_count=100):
    from scapy.all import sniff, IP, TCP, UDP, DNS, HTTP
    import logging
    
    # 配置日志
    logging.basicConfig(filename='malware_network_analysis.log', level=logging.INFO,
                        format='%(asctime)s - %(message)s')
    
    suspicious_ips = set()
    dns_requests = []
    http_headers = []
    
    def packet_handler(packet):
        # 分析IP层
        if IP in packet:
            ip_src = packet[IP].src
            ip_dst = packet[IP].dst
            
            # 分析TCP层
            if TCP in packet:
                tcp_sport = packet[TCP].sport
                tcp_dport = packet[TCP].dport
                
                # 检查常见的C2端口
                suspicious_ports = [4444, 8080, 9000, 31337, 8443]
                if tcp_dport in suspicious_ports:
                    logging.info(f"Suspicious TCP connection: {ip_src}:{tcp_sport} -> {ip_dst}:{tcp_dport}")
                    suspicious_ips.add(ip_dst)
                    print(f"⚠️ Suspicious TCP: {ip_src}:{tcp_sport} -> {ip_dst}:{tcp_dport}")
                
                # 检查HTTP流量
                if tcp_dport == 80 or tcp_sport == 80:
                    if packet.haslayer('Raw'):
                        raw_data = packet['Raw'].load
                        try:
                            # 提取HTTP头
                            http_data = raw_data.decode('utf-8', errors='ignore')
                            if 'HTTP/' in http_data:
                                headers = http_data.split('\r\n')
                                http_headers.append({
                                    'src': ip_src,
                                    'dst': ip_dst,
                                    'headers': headers[:5]  # 只保存前5行
                                })
                                print(f"📝 HTTP Traffic: {ip_src}:{tcp_sport} -> {ip_dst}:{tcp_dport}")
                        except:
                            pass
            
            # 分析UDP层
            elif UDP in packet:
                udp_sport = packet[UDP].sport
                udp_dport = packet[UDP].dport
                
                # 检查DNS流量
                if UDP in packet and packet[UDP].dport == 53:
                    if DNS in packet:
                        dns = packet[DNS]
                        if dns.qr == 0:  # DNS查询
                            query_name = dns.qd.qname.decode('utf-8')
                            dns_requests.append({
                                'src': ip_src,
                                'query': query_name,
                                'type': dns.qd.qtype
                            })
                            print(f"🔍 DNS Query: {ip_src} -> {query_name}")
    
    print(f"Starting packet capture on {interface}...")
    print(f"Will capture {packet_count} packets or press Ctrl+C to stop")
    
    try:
        # 开始捕获
        sniff(iface=interface, prn=packet_handler, count=packet_count, store=False)
    except KeyboardInterrupt:
        print("\nCapture stopped by user")
    except Exception as e:
        print(f"Error during packet capture: {e}")
    
    # 生成报告
    print("\n--- Analysis Report ---")
    print(f"Suspicious IPs detected: {len(suspicious_ips)}")
    for ip in suspicious_ips:
        print(f"  - {ip}")
    
    print(f"\nDNS Requests captured: {len(dns_requests)}")
    for i, req in enumerate(dns_requests[:5]):  # 只显示前5个
        print(f"  {i+1}. {req['src']} -> {req['query']}")
    if len(dns_requests) > 5:
        print(f"  ... and {len(dns_requests) - 5} more DNS requests")
    
    print(f"\nHTTP Headers captured: {len(http_headers)}")
    for i, hdr in enumerate(http_headers[:3]):  # 只显示前3个
        print(f"  {i+1}. {hdr['src']} -> {hdr['dst']}")
        for line in hdr['headers']:
            print(f"    {line}")
    
    return {
        'suspicious_ips': suspicious_ips,
        'dns_requests': dns_requests,
        'http_headers': http_headers
    }

# 使用Frida进行API钩子示例（JavaScript代码，在Python中执行）
frida_script = """
// 目标进程中的API钩子代码

// 监控文件创建API
Interceptor.attach(Module.findExportByName(null, 'CreateFileW'), {
    onEnter: function(args) {
        // args[0] 是文件路径
        const filePath = Memory.readUtf16String(args[0]);
        send({ 
            type: 'file_create',
            path: filePath,
            handle: args[0].toString(),
            flags: args[1].toString(16),
            access: args[2].toString(16)
        });
    },
    onLeave: function(retval) {
        // retval 是返回的文件句柄
        send({ 
            type: 'file_create_result',
            success: !retval.isNull()
        });
    }
});

// 监控注册表写入API
Interceptor.attach(Module.findExportByName(null, 'RegSetValueExW'), {
    onEnter: function(args) {
        // args[0] 是注册表键句柄
        // args[1] 是值名
        const valueName = Memory.readUtf16String(args[1]);
        send({ 
            type: 'registry_write',
            value_name: valueName,
            value_type: args[2].toString(),
            value_size: args[4].toString()
        });
    }
});

// 监控进程创建API
Interceptor.attach(Module.findExportByName(null, 'CreateProcessW'), {
    onEnter: function(args) {
        // args[0] 是应用程序名称
        // args[1] 是命令行
        const appName = args[0].isNull() ? null : Memory.readUtf16String(args[0]);
        const commandLine = args[1].isNull() ? null : Memory.readUtf16String(args[1]);
        send({ 
            type: 'process_create',
            app_name: appName,
            command_line: commandLine
        });
    }
});

// 监控网络连接API
Interceptor.attach(Module.findExportByName(null, 'connect'), {
    onEnter: function(args) {
        // args[1] 是sockaddr结构体
        const sockaddr = args[1];
        const family = Memory.readUshort(sockaddr);
        
        // 仅处理IPv4连接
        if (family === 2) {
            const port = Memory.readUshort(sockaddr.add(2));
            const addr = Array.from(new Uint8Array(Memory.readByteArray(sockaddr.add(4), 4)))
                .map(b => b.toString()).join('.');
            
            send({ 
                type: 'network_connect',
                address: addr,
                port: port,
                socket: args[0].toString()
            });
        }
    }
});
"""

# Python代码，用于加载并运行Frida脚本
def monitor_with_frida(target_process):
    import frida
    import sys
    
    def on_message(message, data):
        if message['type'] == 'send':
            payload = message['payload']
            if payload['type'] == 'file_create':
                print(f"[FILE] Creating: {payload['path']} (Flags: {payload['flags']})")
            elif payload['type'] == 'registry_write':
                print(f"[REG] Writing to: {payload['value_name']} (Type: {payload['value_type']})")
            elif payload['type'] == 'process_create':
                print(f"[PROC] Creating: {payload['app_name'] or 'N/A'}")
                print(f"        Command: {payload['command_line']}")
            elif payload['type'] == 'network_connect':
                print(f"[NET] Connecting to: {payload['address']}:{payload['port']}")
        elif message['type'] == 'error':
            print(f"Error: {message}")
    
    try:
        # 附加到目标进程
        session = frida.attach(target_process)
        script = session.create_script(frida_script)
        script.on('message', on_message)
        script.load()
        
        print(f"Attached to process {target_process}. Monitoring API calls...")
        print("Press Ctrl+C to stop monitoring")
        
        # 保持运行
        sys.stdin.read()
        
        # 清理
        session.detach()
        
    except Exception as e:
        print(f"Error: {e}")

# 使用示例
# monitor_with_frida("suspicious_process.exe")
# 或者使用PID: monitor_with_frida(1234)