勒索软件解密研究：从样本分析到密钥恢复的完整指南

┌─────────────────────────────────────────────────────────────────┐
│                典型勒索软件加密流程 (AES+RSA 混合)                │
│                                                                 │
│  受害主机                                                    │
│  ┌──────────┐                                                 │
│  │ 勒索软件  │                                                │
│  │          │                                                  │
│  │ 1. 生成随 │                                                │
│  │    机 AES │                                                │
│  │    密钥    │──▶ AES-256-CBC ──▶ 加密所有文件                 │
│  │    (256-bit)│                                              │
│  │           │                                                │
│  │ 2. RSA 公 │                                                │
│  │    钥加密  │──▶ RSA-2048 加密 ──▶ AES 密钥密文               │
│  │    AES 密钥│                                              │
│  │           │                                                │
│  │ 3. 删除原 │                                                │
│  │    AES 密钥│──▶ 本地无残留                                  │
│  └─────┬─────┘                                                │
│        │                                                      │
│        ▼ 上传到 C2                                             │
│  ┌──────────┐                                                 │
│  │ C2 服务器  │  ◀── AES 密钥密文 ──▶ RSA 私钥解密 ──▶ 原始 AES 密钥 │
│  │ (攻击者)   │                                                │
│  └──────────┘                                                │
│                                                                 │
│  关键: 没有 C2 服务器的 RSA 私钥，无法恢复 AES 密钥              │
└─────────────────────────────────────────────────────────────────┘

# 创建专用分析虚拟机
# 推荐使用 FLARE VM (Windows) 或 REMnux (Linux)

# 网络隔离配置 (防止样本外连)
# 方法一: 无网络适配器
# 方法二: 仅主机模式 + 假 C2 (HoneyCreds / FakeNet)

# FakeNet-NG 配置假 C2
git clone https://github.com/fireeye/flare-fakenet-ng.git
cd flare-fakenet-ng
python3 fakenet-ng.bat  # Windows
# 或
./fakenet-ng.sh  # Linux

# 典型 FakeNet 输出:
# [*] DNS: 192.168.1.100 -> 192.168.1.100 (所有域名解析到本机)
# [*] HTTP: 监听 80/443 -> 返回伪造页面
# [*] SMTP: 监听 25 -> 捕获外发邮件

#!/usr/bin/env python3
"""
勒索软件样本信息采集工具
"""
import hashlib
import os
import sys
import json
import pefile
import yara
from datetime import datetime
from typing import Dict, List

class RansomwareAnalyzer:
    """
    勒索软件样本分析器
    """

    def __init__(self, sample_path: str):
        self.sample_path = os.path.abspath(sample_path)
        self.sample_size = os.path.getsize(sample_path)

        with open(sample_path, 'rb') as f:
            self.data = f.read()

        # 计算哈希
        self.md5 = hashlib.md5(self.data).hexdigest()
        self.sha1 = hashlib.sha1(self.data).hexdigest()
        self.sha256 = hashlib.sha256(self.data).hexdigest()

    def basic_info(self) -> Dict:
        """基础信息收集"""
        return {
            'filename': os.path.basename(self.sample_path),
            'size': self.sample_size,
            'md5': self.md5,
            'sha1': self.sha1,
            'sha256': self.sha256,
            'file_type': self._detect_file_type(),
        }

    def _detect_file_type(self) -> str:
        """检测文件类型"""
        if self.data[:2] == b'MZ':
            return 'PE Executable (Windows)'
        elif self.data[:4] == b'\x7fELF':
            return 'ELF Executable (Linux)'
        elif self.data[:4] == b'PK\x03\x04':
            return 'ZIP Archive'
        elif self.data[:6] == b'\xfd7zXZ\x00':
            return 'XZ Compressed'
        return 'Unknown'

    def string_analysis(self) -> Dict:
        """字符串分析"""
        # 提取 ASCII 字符串 (min 4 chars)
        import re
        ascii_strings = re.findall(b'[\x20-\x7e]{4,}', self.data)

        # 提取 Unicode 字符串
        unicode_strings = re.findall(
            b'(?:[\x20-\x7e]\x00){4,}', self.data)
        unicode_decoded = [s.decode('utf-16-le', errors='ignore')
                           for s in unicode_strings]

        # 分类
        suspicious_patterns = {
            'crypto_api': [],
            'file_extension': [],
            'urls': [],
            'email': [],
            'bitcoin': [],
            'ransomware_note': [],
        }

        crypto_apis = [
            b'CryptAcquireContext', b'CryptGenKey', b'CryptEncrypt',
            b'CryptDecrypt', b'AES', b'RSA', b'BCrypt',
            b'NtCreateFile', b'NtWriteFile', b'NtReadFile',
        ]

        for s in ascii_strings:
            s_lower = s.lower()
            for api in crypto_apis:
                if api.lower() in s_lower:
                    suspicious_patterns['crypto_api'].append(s.decode())

            # URL 检测
            if b'http' in s_lower or b'ftp' in s_lower:
                suspicious_patterns['urls'].append(s.decode())

            # 邮箱检测
            if b'@' in s and b'.' in s:
                suspicious_patterns['email'].append(s.decode())

            # Bitcoin 地址 (简化检测)
            if re.match(rb'^[13bc][a-km-zA-HJ-NP-Z1-9]{25,34}$', s):
                suspicious_patterns['bitcoin'].append(s.decode())

            # 文件扩展名
            if re.match(rb'\.[a-zA-Z0-9]{2,6}$', s):
                suspicious_patterns['file_extension'].append(s.decode())

        # 去重
        for key in suspicious_patterns:
            suspicious_patterns[key] = list(set(suspicious_patterns[key]))

        return {
            'total_ascii': len(ascii_strings),
            'total_unicode': len(unicode_decoded),
            'suspicious': suspicious_patterns,
        }

    def pe_analysis(self) -> Dict:
        """PE 文件深度分析"""
        if not self.data[:2] == b'MZ':
            return {'error': 'Not a PE file'}

        pe = pefile.PE(self.sample_path)
        info = {}

        # 导入表分析
        imported_apis = set()
        if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'):
            for entry in pe.DIRECTORY_ENTRY_IMPORT:
                dll = entry.dll.decode('utf-8', errors='ignore')
                for imp in entry.imports:
                    if imp.name:
                        imported_apis.add(imp.name.decode('utf-8', errors='ignore'))

        # 关键 API 分类
        crypto_apis = {'CryptAcquireContextA', 'CryptAcquireContextW',
                       'CryptGenKey', 'CryptEncrypt', 'CryptDecrypt'}
        file_apis = {'CreateFileW', 'CreateFileA', 'ReadFile', 'WriteFile',
                     'FindFirstFileW', 'FindNextFileW', 'DeleteFileW',
                     'NtSetInformationFile'}
        network_apis = {'InternetOpenA', 'HttpOpenRequestA', 'WinHttpOpen',
                        'socket', 'connect', 'send', 'recv'}
        persistence_apis = {'RegSetValueExA', 'RegSetValueExW',
                            'CreateProcessA', 'CreateProcessW'}

        imported_set = set(imported_apis)
        info['imported_crypto_apis'] = list(imported_set & crypto_apis)
        info['imported_file_apis'] = list(imported_set & file_apis)
        info['imported_network_apis'] = list(imported_set & network_apis)
        info['imported_persistence_apis'] = list(imported_set & persistence_apis)

        # 节表分析
        info['sections'] = []
        for section in pe.sections:
            name = section.Name.decode('utf-8', errors='ignore').strip('\x00')
            info['sections'].append({
                'name': name,
                'virtual_size': section.SizeOfRawData,
                'entropy': self._section_entropy(section),
                'characteristics': section.Characteristics,
            })

        # 编译时间
        info['compile_time'] = datetime.fromtimestamp(
            pe.FILE_HEADER.TimeDateStamp).isoformat()

        pe.cleanup_resources()

        return info

    def _section_entropy(self, section) -> float:
        """计算节的熵"""
        import math
        data = section.get_data()
        if len(data) == 0:
            return 0.0
        freq = [0] * 256
        for byte in data:
            freq[byte] += 1
        entropy = 0.0
        for count in freq:
            if count > 0:
                p = count / len(data)
                entropy -= p * math.log2(p)
        return round(entropy, 4)

    def yara_scan(self, rules_path: str) -> List:
        """YARA 规则扫描"""
        try:
            rules = yara.compile(filepath=rules_path)
            matches = rules.match(data=self.data)
            return [{'rule': m.rule, 'tags': m.tags,
                     'strings': [(s.identifier,
                                  b.hex().decode()[:64])
                                 for s, _, b in m.strings]}
                    for m in matches]
        except Exception as e:
            return [{'error': str(e)}]

    def generate_report(self) -> Dict:
        """生成完整分析报告"""
        report = {
            'analysis_date': datetime.now().isoformat(),
            'basic': self.basic_info(),
            'strings': self.string_analysis(),
            'pe_info': self.pe_analysis(),
        }
        return report

    def save_report(self, output_path: str):
        """保存报告"""
        report = self.generate_report()
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        print(f"[*] Report saved to: {output_path}")
        return report


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <sample.exe> [report.json]")
        sys.exit(1)

    analyzer = RansomwareAnalyzer(sys.argv[1])
    report = analyzer.save_report(
        sys.argv[2] if len(sys.argv) > 2 else "ransomware_report.json"
    )

    print("=" * 60)
    print("勒索软件样本分析报告")
    print("=" * 60)
    print(f"\nFilename: {report['basic']['filename']}")
    print(f"Size: {report['basic']['size']} bytes")
    print(f"SHA256: {report['basic']['sha256']}")
    print(f"Type: {report['basic']['file_type']}")

    pe = report['pe_info']
    if 'error' not in pe:
        print(f"\nCompile Time: {pe.get('compile_time', 'N/A')}")
        if pe.get('imported_crypto_apis'):
            print(f"\nCrypto APIs: {', '.join(pe['imported_crypto_apis'])}")
        if pe.get('imported_network_apis'):
            print(f"Network APIs: {', '.join(pe['imported_network_apis'])}")

    strings = report['strings']
    if strings.get('suspicious'):
        sus = strings['suspicious']
        if sus.get('urls'):
            print(f"\nURLs ({len(sus['urls'])}):")
            for url in sus['urls'][:10]:
                print(f"  {url}")
        if sus.get('bitcoin'):
            print(f"\nBitcoin addresses:")
            for addr in sus['bitcoin']:
                print(f"  {addr}")

#!/usr/bin/env python3
"""
从勒索软件样本中提取嵌入的 RSA 私钥
"""
import re
import binascii
import struct
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives.asymmetric.utils import (
    decode_dss_signature, encode_dss_signature
)
from cryptography.hazmat.primitives import hashes


class RSAKeyExtractor:
    """
    RSA 私钥提取器 - 用于分析有缺陷实现的勒索软件
    """

    # PEM 格式的 RSA 私钥标记
    PEM_PRIVATE_KEY_PATTERN = re.compile(
        rb'-----BEGIN (?:RSA )?PRIVATE KEY-----.*?-----END (?:RSA )?PRIVATE KEY-----',
        re.DOTALL
    )

    # 原始 RSA 参数模式 (连续的大整数)
    # RSA-2048: 512-byte modulus + 512-byte exponent + ...
    RSA_PARAM_PATTERN = re.compile(
        rb'(?:[\x00-\xff]{64,}){4,}',
    )

    def __init__(self, sample_path: str):
        with open(sample_path, 'rb') as f:
            self.data = f.read()

    def extract_pem_keys(self) -> list:
        """提取 PEM 格式的 RSA 私钥"""
        keys = []
        for match in self.PEM_PRIVATE_KEY_PATTERN.finditer(self.data):
            pem_data = match.group(0)
            try:
                private_key = serialization.load_pem_private_key(
                    pem_data, password=None, backend=default_backend()
                )
                keys.append({
                    'format': 'PEM',
                    'offset': match.start(),
                    'key': private_key,
                    'key_size': private_key.key_size,
                })
                print(f"[+] Found PEM RSA key at offset 0x{match.start():x}")
                print(f"    Key size: {private_key.key_size} bits")
            except Exception as e:
                print(f"[-] Failed to parse PEM key at 0x{match.start():x}: {e}")

        return keys

    def extract_raw_rsa_params(self) -> list:
        """
        从二进制数据中提取原始 RSA 参数
        适用于没有 PEM 封装的原始密钥数据
        """
        keys = []

        # RSA-2048 密钥结构 (典型格式):
        # [modulus: 256 bytes] [public_exp: 4 bytes] [private_exp: 256 bytes]
        # [p: 128 bytes] [q: 128 bytes] [dp: 128 bytes] [dq: 128 bytes]
        # [qinv: 128 bytes]

        # 扫描常见的 RSA 公钥指数
        common_exponents = [
            b'\x00\x01\x00\x01',  # 65537 (最常见)
            b'\x00\x00\x00\x03',  # 3
            b'\x00\x00\x00\x11',  # 17
        ]

        for exp_bytes in common_exponents:
            idx = 0
            while idx < len(self.data):
                try:
                    idx = self.data.index(exp_bytes, idx)

                    # 检查前面是否有 256 字节的 modulus
                    if idx >= 256:
                        modulus_bytes = self.data[idx - 256:idx]
                        modulus = int.from_bytes(modulus_bytes, 'big')

                        if modulus > 0 and modulus.bit_length() >= 2048:
                            # 检查后面是否有 256 字节的 private exponent
                            after_exp = idx + 4
                            if after_exp + 256 <= len(self.data):
                                priv_exp_bytes = self.data[after_exp:after_exp + 256]
                                priv_exp = int.from_bytes(priv_exp_bytes, 'big')

                                if priv_exp > 0:
                                    keys.append({
                                        'format': 'raw_params',
                                        'offset': idx - 256,
                                        'modulus': modulus,
                                        'public_exponent': int.from_bytes(exp_bytes, 'big'),
                                        'private_exponent': priv_exp,
                                        'key_size': modulus.bit_length(),
                                    })
                                    print(f"[+] Raw RSA key at 0x{idx - 256:x}")
                                    print(f"    Modulus: {modulus.bit_length()} bits")

                    idx += 1
                except ValueError:
                    break

        return keys

    def extract_xor_obfuscated_keys(self, xor_key: int = None) -> list:
        """
        尝试 XOR 解混淆后提取密钥
        某些勒索软件会用短 XOR key 混淆密钥数据
        """
        keys = []

        if xor_key is None:
            # 暴力 XOR key
            test_targets = [
                b'-----BEGIN',
                b'MII',  # Base64 of DER ASN.1
                b'\x30\x82',  # DER SEQUENCE header
            ]
            keys_to_try = range(1, 256)
        else:
            test_targets = [b'-----BEGIN', b'MII', b'\x30\x82']
            keys_to_try = [xor_key]

        for key_byte in keys_to_try:
            decrypted = bytes([b ^ key_byte for b in self.data[:4096]])
            for target in test_targets:
                if target in decrypted:
                    offset = decrypted.index(target)
                    full_decrypted = bytes(
                        [b ^ key_byte for b in self.data]
                    )
                    print(f"[+] Possible XOR key: 0x{key_byte:02x}")

                    # 在解混淆数据中查找 PEM 密钥
                    for match in self.PEM_PRIVATE_KEY_PATTERN.finditer(
                            full_decrypted):
                        try:
                            private_key = serialization.load_pem_private_key(
                                match.group(0), password=None,
                                backend=default_backend()
                            )
                            keys.append({
                                'format': 'XOR_decrypted_PEM',
                                'xor_key': key_byte,
                                'offset': match.start(),
                                'key': private_key,
                            })
                        except Exception:
                            pass

        return keys

    def save_key(self, key_info: dict, output_path: str):
        """保存提取的密钥到文件"""
        if 'key' in key_info:
            pem = key_info['key'].private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            )
        else:
            # 原始参数格式
            pem = f"""# Raw RSA Parameters (offset: 0x{key_info['offset']:x})
# Modulus ({key_info['key_size']} bits):
{key_info['modulus']:x}
# Public Exponent:
{key_info['public_exponent']:x}
# Private Exponent:
{key_info['private_exponent']:x}
"""
            pem = pem.encode()

        with open(output_path, 'wb') as f:
            f.write(pem)
        print(f"[+] Key saved to: {output_path}")


if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <sample.exe> [output_key.pem]")
        sys.exit(1)

    extractor = RSAKeyExtractor(sys.argv[1])

    print("[*] Scanning for PEM-formatted RSA keys...")
    pem_keys = extractor.extract_pem_keys()

    print("\n[*] Scanning for raw RSA parameters...")
    raw_keys = extractor.extract_raw_rsa_params()

    print("\n[*] Scanning for XOR-obfuscated keys...")
    xor_keys = extractor.extract_xor_obfuscated_keys()

    all_keys = pem_keys + raw_keys + xor_keys
    if all_keys:
        print(f"\n[+] Total keys found: {len(all_keys)}")
        output = sys.argv[2] if len(sys.argv) > 2 else "extracted_key.pem"
        extractor.save_key(all_keys[0], output)
    else:
        print("\n[-] No embedded RSA keys found")
        print("    This sample may use proper key generation (no embedded keys)")

"""
IDAPython 密钥提取脚本
在 IDA Pro 中运行，自动定位和提取嵌入的 RSA 密钥
"""
import idc
import idaapi
import idautils
import ida_bytes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
import re
import binascii


class IDARSAKeyExtractor:
    """
    IDA Pro RSA 密钥自动提取器
    """

    def __init__(self):
        self.dbg_ida = idaapi.get_inf_structure()

    def scan_all_segments(self) -> list:
        """扫描所有段中的密钥数据"""
        keys = []

        for seg_ea in idautils.Segments():
            seg_name = idc.get_segm_name(seg_ea)
            seg_start = idc.get_segm_start(seg_ea)
            seg_end = idc.get_segm_end(seg_ea)

            # 提取段数据
            seg_data = ida_bytes.get_bytes(seg_start, seg_end - seg_start)

            # PEM 密钥检测
            pem_pattern = re.compile(
                rb'-----BEGIN (?:RSA )?PRIVATE KEY-----'
                rb'.*?'
                rb'-----END (?:RSA )?PRIVATE KEY-----',
                re.DOTALL
            )

            for match in pem_pattern.finditer(seg_data):
                key_offset = seg_start + match.start()
                print(f"[+] PEM key in {seg_name} at 0x{key_offset:08x}")

                try:
                    key = serialization.load_pem_private_key(
                        match.group(0), password=None,
                        backend=default_backend()
                    )
                    keys.append({
                        'segment': seg_name,
                        'ea': key_offset,
                        'key': key,
                        'key_size': key.key_size,
                    })
                except Exception as e:
                    print(f"  [-] Parse error: {e}")

            # DER 格式的 ASN.1 SEQUENCE 检测
            # RSA 私钥 DER 以 30 82 XX XX 开头
            der_pattern = re.compile(rb'\x30\x82[\x01-\x03][\x00-\xff]')
            for match in der_pattern.finditer(seg_data):
                key_offset = seg_start + match.start()

                # 尝试提取可能的 DER 数据
                der_len = int.from_bytes(seg_data[match.start() + 2:
                                                   match.start() + 4], 'big')
                if der_len > 0 and der_len < 8192:
                    der_data = seg_data[match.start():match.start() + 2 + der_len]

                    try:
                        key = serialization.load_der_private_key(
                            der_data, password=None,
                            backend=default_backend()
                        )
                        print(f"[+] DER key in {seg_name} at 0x{key_offset:08x}")
                        keys.append({
                            'segment': seg_name,
                            'ea': key_offset,
                            'key': key,
                            'key_size': key.key_size,
                            'format': 'DER',
                        })
                    except Exception:
                        pass

        return keys

    def find_crypto_api_calls(self) -> list:
        """查找加密 API 调用，定位密钥使用位置"""
        crypto_apis = [
            'CryptAcquireContext', 'CryptGenKey', 'CryptImportKey',
            'CryptEncrypt', 'CryptDecrypt', 'CryptDeriveKey',
            'BCryptOpenAlgorithmProvider', 'BCryptGenerateSymmetricKey',
            'BCryptEncrypt', 'BCryptDecrypt',
        ]

        calls = []
        for ea in idautils.Functions():
            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)
                    for api in crypto_apis:
                        if api.lower() in op.lower():
                            calls.append({
                                'ea': head,
                                'api': api,
                                'instruction': idc.GetDisasm(head),
                            })
        return calls

    def extract_key_usage_context(self, api_call_ea: int, window: int = 50) -> dict:
        """
        提取加密 API 调用的上下文，定位密钥数据引用
        """
        context = {'calls': [], 'data_refs': []}

        # 向前追溯 window 条指令
        ea = api_call_ea
        for _ in range(window):
            ea = idc.prev_head(ea)
            disasm = idc.GetDisasm(ea)
            context['calls'].append({
                'ea': ea,
                'disasm': disasm,
            })

            # 检查数据引用
            for ref in idautils.DataRefsFrom(ea):
                ref_name = idc.get_name(ref)
                if ref_name:
                    context['data_refs'].append({
                        'ref_ea': ref,
                        'name': ref_name,
                    })

        return context

    def run(self):
        """执行完整分析"""
        print("=" * 60)
        print("IDARSAKeyExtractor - 勒索软件 RSA 密钥提取器")
        print("=" * 60)

        print("\n[*] Scanning segments for embedded keys...")
        keys = self.scan_all_segments()

        print(f"\n[*] Searching for crypto API calls...")
        calls = self.find_crypto_api_calls()
        print(f"    Found {len(calls)} crypto API calls")

        for call in calls[:20]:
            print(f"  0x{call['ea']:08x}: {call['api']}")
            ctx = self.extract_key_usage_context(call['ea'])
            if ctx['data_refs']:
                for ref in ctx['data_refs']:
                    print(f"    -> {ref['name']} @ 0x{ref['ref_ea']:08x}")

        if keys:
            print(f"\n[+] Found {len(keys)} embedded keys!")
            for key in keys:
                output_path = f"extracted_key_{key['segment']}_0x{key['ea']:08x}.pem"
                pem = key['key'].private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption()
                )
                with open(output_path, 'wb') as f:
                    f.write(pem)
                print(f"    Saved: {output_path}")
        else:
            print("\n[-] No embedded RSA keys found")
            print("    Possible reasons:")
            print("    1. Key generated at runtime (from RNG)")
            print("    2. Key received from C2 server")
            print("    3. Key is obfuscated (XOR/base64/etc)")


if __name__ == "__main__":
    extractor = IDARSAKeyExtractor()
    extractor.run()

# Volatility 3 分析流程

# Step 1: 确认内存镜像
vol3 -f ransomware_memory.raw imageinfo
# 输出:
# Volatility 3 Framework 2.5.2
# Progress:  100.00		PDB scanning finished
# LayerUnion: Attempting to stack a CachingLayer on top of the requested 1 layers
# Determining profile based on KDBG scan...
# Profile suggestion: Win10x64_19041

# Step 2: 列出进程
vol3 -f ransomware_memory.raw -o ./output windows.pslist
# 找到可疑进程 (如 svchost.exe 异常位置)

# Step 3: 提取可疑进程内存
vol3 -f ransomware_memory.raw -o ./output windows.dump mempid <PID>
# 输出: ransomware_<PID>.dmp

# Step 4: 在进程内存中搜索 RSA 密钥
python3 << 'PYEOF'
"""
从内存 dump 中提取 RSA 密钥
"""
import re
import sys

def extract_keys_from_memory(dump_path: str):
    with open(dump_path, 'rb') as f:
        data = f.read()

    print(f"[*] Analyzing memory dump: {len(data)} bytes")

    # 1. 搜索 PEM 格式密钥
    pem_pattern = re.compile(
        rb'-----BEGIN (?:RSA )?PRIVATE KEY-----'
        rb'.*?'
        rb'-----END (?:RSA )?PRIVATE KEY-----',
        re.DOTALL
    )

    for match in pem_pattern.finditer(data):
        offset = match.start()
        print(f"[+] PEM key at 0x{offset:08x}")
        with open(f"pem_key_0x{offset:08x}.pem", "wb") as f:
            f.write(match.group(0))

    # 2. 搜索 Base64 编码的密钥 (无 PEM header)
    b64_pattern = re.compile(
        rb'[A-Za-z0-9+/]{100,}={0,2}',
    )

    for match in b64_pattern.finditer(data):
        encoded = match.group(0)
        try:
            import base64
            decoded = base64.b64decode(encoded)
            if b'\x30\x82' in decoded[:4]:  # DER header
                offset = match.start()
                print(f"[+] DER key (base64) at 0x{offset:08x}")
                with open(f"der_key_0x{offset:08x}.der", "wb") as f:
                    f.write(decoded)
        except Exception:
            pass

    # 3. 搜索 AES 密钥 (128/192/256 bit)
    # AES-256 密钥在内存中的特征模式
    # 通常在加密操作后，密钥以连续的非重复字节存在
    # 简单启发式: 寻找 32 字节高熵区域

    def entropy(data):
        import math
        if len(data) == 0:
            return 0
        freq = [0] * 256
        for b in data:
            freq[b] += 1
        return -sum((c/len(data)) * math.log2(c/len(data)) for c in freq if c > 0)

    # 扫描 32 字节窗口
    key_candidates = []
    for i in range(0, len(data) - 32, 16):
        window = data[i:i + 32]
        e = entropy(window)
        if e > 7.0:  # 高熵阈值
            key_candidates.append((i, e, window.hex()[:32]))

    if key_candidates:
        print(f"\n[*] High-entropy 32-byte candidates (possible AES-256 keys):")
        for offset, ent, preview in sorted(key_candidates,
                                            key=lambda x: -x[1])[:10]:
            print(f"  0x{offset:08x} (entropy: {ent:.2f}): {preview}...")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <memory_dump>")
        sys.exit(1)
    extract_keys_from_memory(sys.argv[1])
PYEOF

# Step 5: 使用 yarascan 插件搜索
vol3 -f ransomware_memory.raw -o ./output windows.vadyarascan \
    --yara-rules ransomware_signatures.yar

# Step 6: malfind 检测注入代码
vol3 -f ransomware_memory.raw windows.malfind \
    --pid <ransomware_pid> -D ./injected_code/

# Step 7: cmdscan 查看命令行参数
vol3 -f ransomware_memory.raw windows.cmdscan

# Step 8: netscan 查看网络连接 (C2 信息)
vol3 -f ransomware_memory.raw windows.netscan

#!/usr/bin/env python3
"""
NoMoreRansom.org 密钥数据库查询工具
自动化检查受害文件是否可使用已知密钥解密
"""
import requests
import hashlib
import os
import json
from typing import Optional, Dict

class NoMoreRansomChecker:
    """
    查询 NoMoreRansom.org 的解密工具数据库
    """

    BASE_URL = "https://www.nomoreransom.org"
    API_URL = "https://www.nomoreransom.org/api/crypto-sherlock/v1"

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'RansomwareAnalyzer/1.0 (Security Research)'
        })

    def identify_ransomware(self, encrypted_file_path: str,
                            ransom_note_path: str = None) -> Optional[Dict]:
        """
        识别勒索软件家族
        通过文件扩展名、勒索信内容等特征匹配
        """
        results = {
            'extensions_match': [],
            'content_match': [],
            'hash_match': [],
        }

        # 已知勒索软件扩展名数据库 (简化版)
        ransomware_extensions = {
            '.lockbit': 'LockBit 3.0',
            '.alphv': 'BlackCat (ALPHV)',
            '.hive': 'Hive',
            '.conti': 'Conti',
            '.wncry': 'WannaCry',
            '.encrypted': 'REvil/Sodinokibi',
            '.cry': 'Generic',
            '.locked': 'Generic',
            '.crypto': 'CryptoLocker',
            '.zepto': 'Zepto/CryptoWall',
            '.toad': 'Tox',
        }

        # 文件扩展名匹配
        _, ext = os.path.splitext(encrypted_file_path)
        ext_lower = ext.lower()
        if ext_lower in ransomware_extensions:
            results['extensions_match'].append({
                'family': ransomware_extensions[ext_lower],
                'extension': ext_lower,
                'confidence': 'medium'
            })

        # 勒索信内容匹配
        if ransom_note_path and os.path.exists(ransom_note_path):
            with open(ransom_note_path, 'r', errors='ignore') as f:
                note_content = f.read().lower()

            note_signatures = {
                'your files are encrypted': 'Generic',
                'lockbit': 'LockBit',
                'satan': 'Satan Ransomware',
                'you have been infected': 'TeslaCrypt',
                'bitcoin': 'Generic ransomware',
                'decrypt': 'Generic ransomware',
                'pay': 'Generic ransomware',
            }

            for sig, family in note_signatures.items():
                if sig in note_content:
                    results['content_match'].append({
                        'family': family,
                        'signature': sig,
                        'confidence': 'low' if family == 'Generic' else 'medium',
                    })

        # 文件哈希匹配
        with open(encrypted_file_path, 'rb') as f:
            data = f.read()
            sha256 = hashlib.sha256(data).hexdigest()

        # 这里应该查询 NoMoreRansom 的哈希数据库
        results['hash_match'].append({
            'sha256': sha256,
            'query_status': 'check_against_nomoreransom_db',
        })

        return results

    def search_decryption_tools(self, family: str) -> list:
        """
        搜索特定勒索软件家族的解密工具
        """
        # NoMoreRansom 提供解密工具的列表页面
        # 实际使用时需要解析页面或调用官方 API

        tools_database = {
            'WannaCry': {
                'available': True,
                'url': 'https://www.nomoreransom.org/en/wannacry-decryption-tool.html',
                'requirements': 'Encrypted file + original file pair'
            },
            'LockBit': {
                'available': True,  # 执法部门合作后获得密钥
                'url': 'https://www.nomoreransom.org/en/lockbit-decryption-tool.html',
                'requirements': 'Encrypted file'
            },
            'Hive': {
                'available': True,  # 执法部门获取密钥数据库
                'url': 'https://www.nomoreransom.org/en/hive-decryption-tool.html',
                'requirements': 'Encrypted file'
            },
            'BlackCat': {
                'available': False,  # 截至 2024 年初无解密工具
                'url': None,
                'requirements': None
            },
        }

        results = []
        for name, info in tools_database.items():
            if family.lower() in name.lower():
                results.append({
                    'name': name,
                    'available': info['available'],
                    'url': info['url'],
                    'requirements': info['requirements'],
                })

        return results

    def check_decryption_feasibility(self, encrypted_file: str,
                                      ransom_note: str = None) -> Dict:
        """
        综合评估解密可行性
        """
        print("=" * 60)
        print("勒索软件解密可行性评估")
        print("=" * 60)

        # 1. 识别勒索软件
        print("\n[*] 正在识别勒索软件家族...")
        identification = self.identify_ransomware(encrypted_file, ransom_note)

        all_matches = (identification['extensions_match'] +
                       identification['content_match'])

        if not all_matches:
            print("[-] 无法识别勒索软件家族")
            return {'feasibility': 'unknown'}

        for match in all_matches:
            print(f"    匹配: {match['family']} (置信度: {match['confidence']})")

        # 2. 查询解密工具
        print("\n[*] 查询解密工具可用性...")
        families = set(m['family'] for m in all_matches)
        tools = []
        for family in families:
            family_tools = self.search_decryption_tools(family)
            tools.extend(family_tools)

        for tool in tools:
            status = "可用" if tool['available'] else "不可用"
            print(f"    {tool['name']}: 解密工具{status}")

        # 3. 综合评估
        available_tools = [t for t in tools if t['available']]
        if available_tools:
            feasibility = 'high'
            recommendation = "使用 NoMoreRansom 提供的解密工具"
        else:
            feasibility = 'low'
            recommendation = "无公开解密工具，建议保留加密文件等待未来突破"

        return {
            'identified_families': list(families),
            'tools': tools,
            'feasibility': feasibility,
            'recommendation': recommendation,
        }


if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <encrypted_file> [ransom_note]")
        sys.exit(1)

    checker = NoMoreRansomChecker()
    result = checker.check_decryption_feasibility(
        sys.argv[1],
        sys.argv[2] if len(sys.argv) > 2 else None
    )
    print(f"\n[*] 评估结果: {result['feasibility']}")
    print(f"[*] 建议: {result['recommendation']}")

┌────────────────────────────────────────────────────────────┐
│                  勒索软件事件响应流程                         │
│                                                            │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐              │
│  │ 发现感染   │───▶│ 隔离主机   │───▶│ 证据保全   │              │
│  │ (异常加密)  │    │ (断网/快照) │    │ (内存/磁盘) │              │
│  └──────────┘    └──────────┘    └──────────┘              │
│       │                                              │
│       ▼                                              │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐              │
│  │ 样本分析   │───▶│ 密钥恢复   │───▶│ 批量解密   │              │
│  │ (逆向/内存) │    │ (尝试所有) │    │ (使用密钥) │              │
│  └──────────┘    └──────────┘    └──────────┘              │
│       │                                              │
│       ▼                                              │
│  ┌────────────────────────────────────┐               │
│  │  恢复决策:                          │               │
│  │  ├─ 密钥恢复成功 → 批量解密文件      │               │
│  │  ├─ 有备份 → 从备份恢复             │               │
│  │  └─ 无密钥无备份 → 保留文件，等待     │               │
│  └────────────────────────────────────┘               │
└────────────────────────────────────────────────────┘

# 关键防御命令清单

# 1. 禁用 VSS (卷影复制) 的勒索软件对抗
# 大多数勒索软件会首先删除卷影副本
# 使用独立备份系统 (不可变备份)

# 2. 应用白名单策略
# Windows AppLocker 配置
New-AppLockerPolicy -RuleExecutionLevel Enforce -RuleType Path

# 3. 网络分段
# 阻止横向传播
iptables -A FORWARD -p tcp --dport 445 -j DROP  # SMB
iptables -A FORWARD -p tcp --dport 3389 -j DROP  # RDP

# 4. 端点保护
# 部署 EDR 并配置勒索软件专用检测规则
# 示例: CrowdStrike Falcon / SentinelOne / Microsoft Defender

# 5. 文件系统保护
# Windows Controlled Folder Access
Set-MpPreference -EnableControlledFolderAccess Enabled
Add-MpPreference -ControlledFolderAccessPaths "C:\Users\Documents"

# 6. 备份策略
# 3-2-1 备份规则
# - 3 份副本
# - 2 种不同介质
# - 1 份离线/异地