高级恶意软件分析技术：从沙箱对抗到自动化逆向

// 1. IsDebuggerPresent 检测
if (IsDebuggerPresent()) {
    ExitProcess(0);  // 检测到调试器直接退出
}

// 2. NtQueryInformationProcess 检测
HANDLE hProcess = GetCurrentProcess();
DWORD isDebugged = 0;
NtQueryInformationProcess(hProcess, ProcessDebugPort,
    &isDebugged, sizeof(isDebugged), NULL);
if (isDebugged == -1) {
    // 被调试，退出
}

// 3. 时间检测 (Sleep 欺骗)
DWORD start = GetTickCount();
Sleep(1000);
DWORD elapsed = GetTickCount() - start;
if (elapsed < 1000) {
    // 沙箱可能加速了时间
}

// 4. 进程数量检测
// 检查是否存在调试工具进程
const char* dbg_tools[] = {
    "ollydbg.exe", "x64dbg.exe", "idaq.exe",
    "wireshark.exe", "procmon.exe", NULL
};

#!/usr/bin/env python3
"""
使用 Frida 绕过恶意软件反调试检测
"""
import frida
import sys


def on_message(message, data):
    if message['type'] == 'send':
        print(f"[+] {message['payload']}")
    elif message['type'] == 'error':
        print(f"[-] {message['description']}")


script = frida.get_usb_device().attach('malware.exe').create_script(r"""

// Patch IsDebuggerPresent
const IsDebuggerPresentAddr = Module.findExportByName('kernel32.dll', 'IsDebuggerPresent');
Interceptor.replace(IsDebuggerPresentAddr, new NativeCallback(function() {
    return 0;  // 始终返回 false
}, 'int', []));

// Patch NtQueryInformationProcess - ProcessDebugPort
const NtQueryAddr = Module.findExportByName('ntdll.dll', 'NtQueryInformationProcess');
Interceptor.attach(NtQueryAddr, {
    onEnter: function(args) {
        // ProcessDebugPort = 7
        if (args[2].toInt32() === 7) {
            // 将返回值设为 0 (未调试)
            Memory.writeUInt(args[3], 0);
        }
    },
    onLeave: function(retval) {
        retval.replace(0);  // STATUS_SUCCESS
    }
});

// Patch GetTickCount - 防止时间检测
const GetTickCountAddr = Module.findExportByName('kernel32.dll', 'GetTickCount');
let tickCount = 0;
Interceptor.replace(GetTickCountAddr, new NativeCallback(function() {
    tickCount += 1000;
    return tickCount;
}, 'DWORD', []));

send("Anti-debug patches applied successfully");
""")

script.on('message', on_message)
script.load()
sys.stdin.read()

#!/usr/bin/env python3
"""
IDAPython 自动化恶意软件分析脚本
自动提取 IOC、字符串、API 调用、导入函数
"""
import idc
import idaapi
import idautils
import ida_bytes
import json
from collections import defaultdict


class MalwareAutoAnalyzer:
    """恶意软件自动分析器"""
    
    def __init__(self):
        self.results = {
            'strings': [],
            'api_calls': [],
            'imports': [],
            'ioc': {
                'ip_addresses': [],
                'domains': [],
                'urls': [],
                'mutexes': [],
                'registry_keys': [],
                'file_paths': [],
            },
            'crypto_functions': [],
        }
    
    def extract_strings(self) -> list:
        """提取所有字符串"""
        for seg_ea in idautils.Segments():
            seg_name = idc.get_segm_name(seg_ea)
            if seg_name != '.rdata' and seg_name != '.data':
                continue
            
            for head in idautils.Heads(seg_ea, idc.get_segm_end(seg_ea)):
                if idc.get_str_type(head) != 0:
                    s = idc.get_strlit_contents(head)
                    if s:
                        s_str = s.decode('utf-8', errors='ignore')
                        self.results['strings'].append({
                            'address': hex(head),
                            'string': s_str,
                        })
        return self.results['strings']
    
    def extract_api_calls(self) -> list:
        """提取所有 API 调用"""
        for ea in idautils.Functions():
            func_name = idc.get_func_name(ea)
            
            for head in idautils.Heads(ea, idc.get_func_attr(ea, idc.FUNCATTR_END)):
                mnem = idc.print_insn_mnem(head)
                if mnem in ('call', 'jmp'):
                    op = idc.print_operand(head, 0)
                    # 过滤 API 调用
                    if 'sub_' not in op and 'loc_' not in op:
                        self.results['api_calls'].append({
                            'function': func_name,
                            'address': hex(head),
                            'call': op,
                        })
        
        return self.results['api_calls']
    
    def extract_ioc(self) -> dict:
        """提取 IOC"""
        import re
        
        strings = [s['string'] for s in self.results['strings']]
        
        # IP 地址
        ip_pattern = re.compile(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b')
        for s in strings:
            for match in ip_pattern.finditer(s):
                ip = match.group(0)
                if not ip.startswith('127.') and not ip.startswith('0.'):
                    self.results['ioc']['ip_addresses'].append(ip)
        
        # 域名
        domain_pattern = re.compile(r'[a-zA-Z0-9][-a-zA-Z0-9]*\.[a-zA-Z]{2,}')
        for s in strings:
            for match in domain_pattern.finditer(s):
                domain = match.group(0)
                if '.' in domain and len(domain) > 5:
                    self.results['ioc']['domains'].append(domain)
        
        # URL
        url_pattern = re.compile(r'https?://[^\s<>"{}|\\^`\[\]]+')
        for s in strings:
            for match in url_pattern.finditer(s):
                self.results['ioc']['urls'].append(match.group(0))
        
        # Mutex
        mutex_pattern = re.compile(r'Mutex[._-][a-zA-Z0-9_-]+')
        for s in strings:
            for match in mutex_pattern.finditer(s):
                self.results['ioc']['mutexes'].append(match.group(0))
        
        return self.results['ioc']
    
    def detect_crypto(self) -> list:
        """检测加密相关函数"""
        crypto_keywords = [
            'CryptEncrypt', 'CryptDecrypt', 'CryptGenKey',
            'AesEncrypt', 'RsaEncrypt', 'AES_', 'RSA_',
            'ChaCha20', 'Salsa20', 'SHA256', 'SHA512',
            'MD5', 'HMAC',
        ]
        
        for func_ea in idautils.Functions():
            func_name = idc.get_func_name(func_ea).lower()
            for keyword in crypto_keywords:
                if keyword.lower() in func_name:
                    self.results['crypto_functions'].append({
                        'function': func_name,
                        'address': hex(func_ea),
                        'keyword': keyword,
                    })
        
        return self.results['crypto_functions']
    
    def run(self) -> dict:
        """执行完整分析"""
        print("[*] Extracting strings...")
        self.extract_strings()
        
        print("[*] Extracting API calls...")
        self.extract_api_calls()
        
        print("[*] Extracting IOC...")
        self.extract_ioc()
        
        print("[*] Detecting crypto functions...")
        self.detect_crypto()
        
        return self.results


if __name__ == '__main__':
    analyzer = MalwareAutoAnalyzer()
    results = analyzer.run()
    
    with open('analysis_results.json', 'w') as f:
        json.dump(results, f, indent=2)
    
    print(f"\n[+] Analysis complete. Results saved to analysis_results.json")
    print(f"    Strings: {len(results['strings'])}")
    print(f"    API calls: {len(results['api_calls'])}")
    print(f"    IOC IPs: {len(results['ioc']['ip_addresses'])}")
    print(f"    IOC domains: {len(results['ioc']['domains'])}")
    print(f"    Crypto functions: {len(results['crypto_functions'])}")

#!/usr/bin/env python3
"""
自动化恶意软件分析流水线
集成多个工具实现端到端分析
"""
import subprocess
import json
import hashlib
import os
from datetime import datetime


class MalwareAnalysisPipeline:
    """恶意软件自动分析流水线"""
    
    def __init__(self, sample_path: str):
        self.sample_path = sample_path
        self.sha256 = self._calc_hash()
        self.results = {}
    
    def _calc_hash(self) -> str:
        with open(self.sample_path, 'rb') as f:
            return hashlib.sha256(f.read()).hexdigest()
    
    def run_static_analysis(self) -> dict:
        """静态分析"""
        results = {}
        
        # 1. strings 提取
        strings_output = subprocess.check_output(
            ['strings', '-n', '8', self.sample_path],
            text=True
        )
        results['strings'] = strings_output.split('\n')[:100]
        
        # 2. PEiD 检测
        peid_output = subprocess.check_output(
            ['peid', self.sample_path], text=True
        )
        results['packer'] = peid_output
        
        # 3. ExeinfoPE
        results['file_info'] = {
            'size': os.path.getsize(self.sample_path),
            'sha256': self.sha256,
        }
        
        return results
    
    def run_yara_scan(self, rules_dir: str = '/usr/share/yara/rules') -> list:
        """YARA 规则扫描"""
        yara_cmd = [
            'yara', '-r', rules_dir, self.sample_path
        ]
        output = subprocess.check_output(yara_cmd, text=True)
        return output.strip().split('\n') if output.strip() else []
    
    def run_dynamic_analysis(self) -> dict:
        """动态分析 (需要沙箱环境)"""
        return {
            'status': 'requires_sandbox_environment',
            'recommendation': 'Use Cuckoo Sandbox or REMnux for dynamic analysis',
        }
    
    def run(self) -> dict:
        """执行完整流水线"""
        print(f"[*] Analyzing sample: {self.sample_path}")
        print(f"[*] SHA256: {self.sha256}")
        
        print("\n[*] Running static analysis...")
        self.results['static'] = self.run_static_analysis()
        
        print("[*] Running YARA scan...")
        self.results['yara'] = self.run_yara_scan()
        
        print("[*] Running dynamic analysis (placeholder)...")
        self.results['dynamic'] = self.run_dynamic_analysis()
        
        # 保存结果
        output_file = f"analysis_{self.sha256}.json"
        with open(output_file, 'w') as f:
            json.dump(self.results, f, indent=2)
        
        print(f"\n[+] Analysis complete. Results saved to {output_file}")
        return self.results


if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <malware_sample>")
        sys.exit(1)
    
    pipeline = MalwareAnalysisPipeline(sys.argv[1])
    pipeline.run()