performing-threat-hunting-with-yara-rules

使用 YARA 规则进行威胁狩猎

使用 YARA 规则扫描文件、目录和内存转储，以识别 恶意软件家族、可疑模式和 IOC 匹配项。

适用场景

• 主动在网络共享、端点和电子邮件附件中搜寻未知恶意软件变种
• 扫描隔离目录或沙箱输出，以对恶意软件家族进行分类
• 搜索进程内存转储，查找注入的代码或仅存在于内存中的载荷
• 根据收集的大型样本库验证威胁情报 IOC
• 对事件响应工件进行分类，快速识别已知恶意软件家族
• 构建自动化检测流水线，在新文件进入时对其进行扫描

请勿用于实时端点防护（请改用 EDR Agent）；YARA 扫描最适合用于批量狩猎、分类和收集后的分析，这些场景可以接受扫描延迟。

前置条件

• 已安装 YARA 4.x（Debian/Ubuntu 上使用 apt install yara，macOS 上使用 brew install yara）
• Python 3.8+ 和 yara-python（pip install yara-python）
• 用于自动生成规则的 yarGen（git clone https://github.com/Neo23x0/yarGen）
• 用于扫描的恶意软件样本库或可疑文件（来自恶意软件样本库、VT 或事件工件）
• 可选：用于 PE 头分析的 pefile，用于内存提取的 malduck
• 威胁情报 YARA 规则集（例如 YARA-Rules 社区仓库、Florian Roth 的 signature-base）

工作流

步骤 1：安装 YARA 和 Python 绑定

# Linux
sudo apt update && sudo apt install -y yara

# Python bindings
pip install yara-python

# Verify installation
yara --version
python3 -c "import yara; print(yara.YARA_VERSION)"

步骤 2：编写基础 YARA 规则

编写基于字符串、十六进制模式和文件元数据进行匹配的规则：

// File: rules/emotet_loader.yar
rule Emotet_Loader_2026 {
    meta:
        author = "Threat Intel Team"
        description = "Detects Emotet first-stage loader DLL"
        date = "2026-01-20"
        reference = "https://attack.mitre.org/software/S0367/"
        mitre_attack = "T1059.001, T1055.001"
        severity = "critical"

    strings:
        // Emotet export function name patterns
        $export1 = "DllRegisterServer" ascii
        $export2 = "RunDLL" ascii nocase

        // Obfuscated string decryption routine
        $decrypt_loop = { 8B 45 ?? 33 45 ?? 89 45 ?? 8B 4D ?? 03 4D ?? }

        // PowerShell download cradle in embedded script
        $ps_cradle = /powershell[^\n]{0,50}-e(nc|ncodedcommand)/i

        // Known C2 URI patterns
        $uri1 = "/wp-content/uploads/" ascii
        $uri2 = "/wp-admin/css/" ascii
        $uri3 = "/wp-includes/" ascii

        // PE characteristics
        $mz = "MZ" at 0

    condition:
        $mz and
        filesize < 2MB and
        (
            ($export1 and $decrypt_loop) or
            ($ps_cradle and any of ($uri*)) or
            (2 of ($uri*) and $decrypt_loop)
        )
}

步骤 3：使用模块编写高级规则

使用 YARA 模块检查 PE 头，并基于数学计算进行熵检查：

import "pe"
import "math"

rule Suspicious_Packed_Executable {
    meta:
        author = "Threat Hunting Team"
        description = "Detects PE files with high entropy sections indicating packing or encryption"
        severity = "medium"

    condition:
        pe.is_pe and
        pe.number_of_sections > 0 and
        for any section in pe.sections : (
            math.entropy(section.offset, section.size) > 7.2 and
            section.size > 1024
        ) and
        pe.imports("kernel32.dll", "VirtualAlloc") and
        pe.imports("kernel32.dll", "VirtualProtect")
}

rule Suspicious_UPX_Modified {
    meta:
        description = "Detects UPX-packed binaries with tampered section names"
        severity = "medium"

    strings:
        $upx_magic = { 55 50 58 21 }  // UPX!

    condition:
        pe.is_pe and
        $upx_magic and
        not (
            pe.sections[0].name == "UPX0" and
            pe.sections[1].name == "UPX1"
        )
}

步骤 4：使用 yara-python 扫描文件和目录

import yara
import os
import json
from datetime import datetime
from pathlib import Path

def compile_rules(rule_paths):
    """Compile YARA rules from one or more .yar files."""
    rule_files = {}
    for i, path in enumerate(rule_paths):
        namespace = Path(path).stem
        rule_files[namespace] = path
    return yara.compile(filepaths=rule_files)

def scan_directory(rules, target_dir, recursive=True):
    """Scan a directory for matches and return structured results."""
    results = []
    scan_count = 0
    error_count = 0

    for root, dirs, files in os.walk(target_dir):
        for filename in files:
            filepath = os.path.join(root, filename)
            scan_count += 1
            try:
                matches = rules.match(filepath, timeout=60)
                if matches:
                    for match in matches:
                        result = {
                            "file": filepath,
                            "rule": match.rule,
                            "namespace": match.namespace,
                            "tags": match.tags,
                            "meta": match.meta,
                            "strings": [],
                            "scan_time": datetime.utcnow().isoformat()
                        }
                        for offset, identifier, data in match.strings:
                            result["strings"].append({
                                "offset": hex(offset),
                                "identifier": identifier,
                                "data": data.hex() if isinstance(data, bytes) else data
                            })
                        results.append(result)
                        print(f"  MATCH: {match.rule} -> {filepath}")
            except yara.TimeoutError:
                error_count += 1
                print(f"  TIMEOUT scanning {filepath}")
            except yara.Error as e:
                error_count += 1

        if not recursive:
            break

    print(f"\nScan complete: {scan_count} files scanned, "
          f"{len(results)} matches, {error_count} errors")
    return results

# Compile and scan
rules = compile_rules([
    "rules/emotet_loader.yar",
    "rules/suspicious_packed.yar"
])

matches = scan_directory(rules, "/mnt/evidence/collected_samples/")

# Export results
with open("yara_scan_results.json", "w") as f:
    json.dump(matches, f, indent=2)

步骤 5：扫描进程内存转储

搜寻仅存在于运行中进程内存中的指标：

import yara

def scan_memory_dump(rules, dump_path):
    """Scan a process memory dump for YARA matches."""
    matches = rules.match(dump_path, timeout=120)

    for match in matches:
        print(f"Rule: {match.rule}")
        print(f"  Severity: {match.meta.get('severity', 'unknown')}")
        for offset, identifier, data in match.strings:
            # Show context around the match
            print(f"  String {identifier} at offset {hex(offset)}")
            if len(data) <= 64:
                print(f"    Data: {data.hex()}")

    return matches

# Rules targeting in-memory artifacts
memory_rules = yara.compile(source="""
rule Cobalt_Strike_Beacon_Memory {
    meta:
        description = "Detects Cobalt Strike beacon in process memory"
        severity = "critical"
    strings:
        $config_start = { 2E 2F 2E 2F 2E 2C }
        $sleep_mask = { 48 8B 44 24 ?? 48 89 44 24 ?? 48 8B 44 24 }
        $named_pipe = "\\\\\\\\.\\\\pipe\\\\msagent_" ascii
        $watermark = { 00 00 00 00 00 00 ?? ?? 00 00 }
    condition:
        2 of them
}
""")

scan_memory_dump(memory_rules, "/mnt/evidence/lsass_dump.dmp")

步骤 6：使用 yarGen 自动生成规则

使用 yarGen 从恶意软件样本中提取独特字符串并生成规则：

# Clone and set up yarGen
git clone https://github.com/Neo23x0/yarGen.git
cd yarGen
pip install -r requirements.txt

# Download the string databases (run once)
python3 yarGen.py --update

# Generate rules from a directory of malware samples
python3 yarGen.py \
    -m /mnt/evidence/malware_samples/ \
    -o generated_rules.yar \
    --excludegood \
    -p "AutoGen" \
    -a "Threat Hunting Team" \
    --score 50

# Generate rules for a single sample with maximum detail
python3 yarGen.py \
    -m /mnt/evidence/malware_samples/suspicious.exe \
    -o single_sample_rule.yar \
    --opcodes \
    --debug

步骤 7：集成社区规则集

从公开的威胁情报仓库下载并合并规则：

# Clone Florian Roth's signature-base (large community rule set)
git clone https://github.com/Neo23x0/signature-base.git

# Clone YARA-Rules community repository
git clone https://github.com/Yara-Rules/rules.git yara-community-rules

# Clone ReversingLabs YARA rules
git clone https://github.com/reversinglabs/reversinglabs-yara-rules.git

import yara
from pathlib import Path

def load_rule_directory(rule_dir, extensions=(".yar", ".yara")):
    """Load all YARA rules from a directory tree."""
    rule_files = {}
    for ext in extensions:
        for rule_file in Path(rule_dir).rglob(f"*{ext}"):
            namespace = rule_file.stem
            # Avoid namespace collisions
            if namespace in rule_files:
                namespace = f"{rule_file.parent.name}_{namespace}"
            rule_files[namespace] = str(rule_file)

    print(f"Loading {len(rule_files)} rule files from {rule_dir}")
    try:
        compiled = yara.compile(filepaths=rule_files)
        return compiled
    except yara.SyntaxError as e:
        print(f"Syntax error in rules: {e}")
        # Fall back to loading rules one by one, skipping broken ones
        valid_rules = {}
        for ns, path in rule_files.items():
            try:
                yara.compile(filepath=path)
                valid_rules[ns] = path
            except yara.SyntaxError:
                print(f"  Skipping broken rule: {path}")
        return yara.compile(filepaths=valid_rules)

# Load and scan with community rules
community_rules = load_rule_directory("signature-base/yara/")
matches = community_rules.match("/mnt/evidence/suspicious_file.exe", timeout=120)

for m in matches:
    print(f"Matched: {m.rule} (namespace: {m.namespace})")

步骤 8：构建持续狩猎流水线

使用文件系统监控，在新文件到达时自动进行扫描：

import yara
import time
import json
import hashlib
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class YaraHuntingHandler(FileSystemEventHandler):
    def __init__(self, rules, alert_file="yara_alerts.jsonl"):
        self.rules = rules
        self.alert_file = alert_file
        self.scanned_hashes = set()

    def on_created(self, event):
        if event.is_directory:
            return
        self._scan_file(event.src_path)

    def _scan_file(self, filepath):
        # Deduplicate by file hash
        try:
            file_hash = hashlib.sha256(Path(filepath).read_bytes()).hexdigest()
        except (PermissionError, FileNotFoundError):
            return

        if file_hash in self.scanned_hashes:
            return
        self.scanned_hashes.add(file_hash)

        matches = self.rules.match(filepath, timeout=60)
        if matches:
            alert = {
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "file": filepath,
                "sha256": file_hash,
                "matches": [
                    {"rule": m.rule, "severity": m.meta.get("severity", "unknown")}
                    for m in matches
                ]
            }
            with open(self.alert_file, "a") as f:
                f.write(json.dumps(alert) + "\n")
            print(f"ALERT: {filepath} matched {len(matches)} rules")

# Set up continuous monitoring
rules = yara.compile(filepaths={"hunting": "rules/all_hunting_rules.yar"})
handler = YaraHuntingHandler(rules)
observer = Observer()
observer.schedule(handler, path="/mnt/quarantine/", recursive=True)
observer.start()
print("YARA hunting pipeline active. Monitoring /mnt/quarantine/ ...")

验证

• 编译所有自定义规则，确保没有语法错误：yara -w rules/*.yar /dev/null
• 确认规则能匹配测试样本库中的已知恶意软件样本（真阳性验证）
• 验证规则不会匹配常见系统文件组成的良性软件样本库（误报测试）
• 测试扫描性能：单个文件的扫描应在超时阈值内完成
• 验证 yarGen 输出的规则能够编译，并能在输入样本上产生有意义的匹配结果
• 检查过滤后，社区规则集是否能够加载且没有严重的语法错误
• 确认将测试文件放入后，持续狩猎流水线会生成 JSONL 格式的告警
• 将 YARA 匹配结果与 VirusTotal 或沙箱结果交叉核对，以验证检测准确性