WebAssembly安全取证深度分析

wasm-objdump --details --sections suspicious.wasm

wasm2wat suspicious.wasm -o suspicious.wat
head -200 suspicious.wat

wasm-objdump -x suspicious.wasm | grep -i "export\|import"

strings suspicious.wasm | head -50

xxd suspicious.wasm | head -5

wasm-validate suspicious.wasm
echo "Validation exit code: $?"

title: Suspicious WebAssembly Module Loading Pattern
id: 9f8a7b6c-1d2e-3f4a-5b6c-7d8e9f0a1b2c
status: experimental
description: Detects suspicious WebAssembly module loading patterns indicative of cryptojacking or malicious code execution
author: x7peeps蓝队
date: 2026-07-21
tags:
  - attack.defense_evasion
  - attack.t1027
  - attack.t1496
  - attack.t1059.007
logsource:
  category: proxy
  product: web
detection:
  selection_wasm_fetch:
    cs-uri|endswith: '.wasm'
    cs-method|contains:
      - 'GET'
      - 'POST'
  selection_suspicious_wasm_source:
    cs-uri|re: '(?i)\.(wasm|wat)$'
    - cs-uri|contains:
      - 'mine'
      - 'hash'
      - 'coin'
      - 'crypto'
      - 'stratum'
      - 'pool'
  selection_websocket_upgrade:
    cs-uri|contains: 'wss://'
    cs-content-type|contains: 'websocket'
  selection_inline_wasm:
    request-body|contains: '\x00asm'
  condition: selection_wasm_fetch and (selection_suspicious_wasm_source or selection_websocket_upgrade or selection_inline_wasm)
level: high
falsepositives:
  - Legitimate WebAssembly applications (Figma, Google Earth, AutoCAD Web)
  - WebAssembly-based development tools
fields:
  - cs-uri
  - cs-ip
  - cs-user-agent
  - c-ip
  - cs-method
  - sc-status

title: WASM Cryptojacking Network Indicators
id: a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d
status: experimental
description: Detects network-level indicators of WASM-based cryptocurrency mining operations
author: x7peeps蓝队
date: 2026-07-21
tags:
  - attack.impact
  - attack.t1496
logsource:
  category: proxy
  product: web
detection:
  selection_mining_pool:
    cs-uri|contains:
      - 'stratum+tcp://'
      - 'stratum+ssl://'
      - 'mining.subscribe'
      - 'mining.authorize'
      - 'mining.notify'
  selection_wasm_mining_domain:
    cs-dst-host|contains:
      - 'coinhive'
      - 'cryptoloot'
      - 'coin-imp'
      - 'webdollar'
      - 'miner'
      - 'moneropool'
  selection_websocket_heavy:
    cs-uri|contains: 'wss://'
    sc-bytes|gt: 1048576
  condition: selection_mining_pool or selection_wasm_mining_domain or (selection_websocket_heavy and cs-uri|endswith: '.wasm')
level: critical
falsepositives:
  - Legitimate mining applications
  - Cryptocurrency exchange WebSocket connections
fields:
  - cs-uri
  - cs-dst-host
  - cs-ip
  - sc-bytes
  - cs-bytes
  - sc-status

#!/bin/bash

SCAN_DIR="${1:-.}"
REPORT_DIR="/tmp/wasm-hunt-$(date +%Y%m%d-%H%M%S)"
SUSPICIOUS_THRESHOLD=1048576

mkdir -p "$REPORT_DIR"

echo "=========================================="
echo "  WASM File Discovery & Analysis v1.0"
echo "  Scan Directory: $SCAN_DIR"
echo "  Report: $REPORT_DIR"
echo "=========================================="

echo ""
echo "[*] Step 1: Discovering WASM files..."
find "$SCAN_DIR" -type f -name "*.wasm" 2>/dev/null > "$REPORT_DIR/wasm_files.txt"
find "$SCAN_DIR" -type f -name "*.wat" 2>/dev/null >> "$REPORT_DIR/wasm_files.txt"
TOTAL=$(wc -l < "$REPORT_DIR/wasm_files.txt")
echo "[+] Found $TOTAL WASM/WAT files"

echo ""
echo "[*] Step 2: Analyzing WASM binary headers..."
while IFS= read -r wasm_file; do
    if [ -f "$wasm_file" ]; then
        HEADER=$(xxd -l 8 "$wasm_file" 2>/dev/null | head -1)
        FILE_SIZE=$(stat -f%z "$wasm_file" 2>/dev/null || stat --printf="%s" "$wasm_file" 2>/dev/null)

        if echo "$HEADER" | grep -q "0061 736d"; then
            echo "[WASM] $wasm_file ($FILE_SIZE bytes)"

            if [ "$FILE_SIZE" -gt "$SUSPICIOUS_THRESHOLD" ]; then
                echo "  [!] WARNING: Unusually large WASM file ($FILE_SIZE bytes)"
                echo "$wasm_file:$FILE_SIZE" >> "$REPORT_DIR/large_files.txt"
            fi

            EXPORTS=$(wasm-objdump -x "$wasm_file" 2>/dev/null | grep "EXPORT" | head -10)
            if [ -n "$EXPORTS" ]; then
                echo "  Exports: $(echo "$EXPORTS" | wc -l) entries"
                echo "$EXPORTS" >> "$REPORT_DIR/exports_$(basename "$wasm_file").txt"
            fi

            IMPORTS=$(wasm-objdump -x "$wasm_file" 2>/dev/null | grep "IMPORT" | head -10)
            if [ -n "$IMPORTS" ]; then
                echo "  Imports: $(echo "$IMPORTS" | wc -l) entries"
            fi

            STRINGS_OUT=$(strings "$wasm_file" 2>/dev/null | grep -iE "stratum|mining|pool|coin|crypto|wallet|monero|xmr" | head -5)
            if [ -n "$STRINGS_OUT" ]; then
                echo "  [!] CRITICAL: Mining-related strings detected:"
                echo "$STRINGS_OUT" | sed 's/^/    /'
                echo "$wasm_file: $STRINGS_OUT" >> "$REPORT_DIR/mining_strings.txt"
            fi
        fi
    fi
done < "$REPORT_DIR/wasm_files.txt"

echo ""
echo "[*] Step 3: Checking for WASM embedded in JavaScript files..."
grep -rl "WebAssembly.instantiate\|WebAssembly.compile\|\.wasm" "$SCAN_DIR" --include="*.js" 2>/dev/null | head -20 > "$REPORT_DIR/js_wasm_loaders.txt"
JS_COUNT=$(wc -l < "$REPORT_DIR/js_wasm_loaders.txt")
echo "[+] Found $JS_COUNT JavaScript files loading WASM modules"

echo ""
echo "[*] Step 4: Analyzing string entropy of WASM files..."
while IFS= read -r wasm_file; do
    if [ -f "$wasm_file" ]; then
        ENTROPY=$(strings "$wasm_file" 2>/dev/null | awk '{ for(i=1; i<=length($0); i++) { c=substr($0,i,1); freq[c]++ } } END { n=0; for(c in freq) { p=freq[c]/length($0); n-=p*log(p)/log(2) }; printf "%.2f", n }')
        if [ -n "$ENTROPY" ]; then
            echo "[ENTROPY] $wasm_file: $ENTROPY"
            if (( $(echo "$ENTROPY > 5.0" | bc -l 2>/dev/null || echo 0) )); then
                echo "  [!] HIGH ENTROPY - possible encrypted/obfuscated content"
                echo "$wasm_file:$ENTROPY" >> "$REPORT_DIR/high_entropy.txt"
            fi
        fi
    fi
done < "$REPORT_DIR/wasm_files.txt"

echo ""
echo "[*] Step 5: Checking Service Worker registrations..."
find "$SCAN_DIR" -path "*/sw.js" -o -path "*/service-worker.js" -o -path "*/sw*.js" 2>/dev/null | while read -r sw_file; do
    if grep -l "WebAssembly\|\.wasm\|memory\.grow" "$sw_file" 2>/dev/null; then
        echo "  [!] Service Worker with WASM: $sw_file"
        echo "$sw_file" >> "$REPORT_DIR/suspicious_sw.txt"
    fi
done

echo ""
echo "=========================================="
echo "[+] Analysis complete. Reports in: $REPORT_DIR"
echo "  wasm_files.txt        - All discovered WASM files"
echo "  large_files.txt       - Oversized WASM files"
echo "  mining_strings.txt    - Mining-related string matches"
echo "  js_wasm_loaders.txt   - JS files loading WASM"
echo "  high_entropy.txt      - High-entropy WASM files"
echo "  suspicious_sw.txt     - Suspicious Service Workers"
echo "=========================================="

#!/usr/bin/env python3
import struct
import sys
import os
import json
import hashlib
import math
from collections import Counter
from datetime import datetime


WASM_MAGIC = b'\x00asm'
SECTION_NAMES = {
    0: 'Custom', 1: 'Type', 2: 'Import', 3: 'Function',
    4: 'Table', 5: 'Memory', 6: 'Global', 7: 'Export',
    8: 'Start', 9: 'Element', 10: 'Code', 11: 'Data'
}

MINING_KEYWORDS = [
    'stratum', 'mining', 'pool', 'coin', 'crypto',
    'wallet', 'monero', 'xmr', 'hash', 'miner',
    'difficulty', 'submit', 'job', 'blob', 'nonce'
]

C2_KEYWORDS = [
    'ws://', 'wss://', 'http://', 'https://',
    'connect', 'socket', 'tunnel', 'relay', 'proxy'
]


def read_leb128(data, offset):
    result = 0
    shift = 0
    while offset < len(data):
        byte = data[offset]
        result |= (byte & 0x7f) << shift
        offset += 1
        if (byte & 0x80) == 0:
            break
        shift += 7
    return result, offset


def calculate_entropy(data):
    if not data:
        return 0.0
    counter = Counter(data)
    length = len(data)
    entropy = 0.0
    for count in counter.values():
        p = count / length
        if p > 0:
            entropy -= p * math.log2(p)
    return round(entropy, 4)


class WASMAnalyzer:
    def __init__(self, filepath):
        self.filepath = filepath
        self.file_size = os.path.getsize(filepath)
        with open(filepath, 'rb') as f:
            self.data = f.read()
        self.sections = []
        self.findings = []
        self.info = {}

    def validate_magic(self):
        if self.data[:4] != WASM_MAGIC:
            return False
        self.info['magic'] = self.data[:4].hex()
        self.info['version'] = struct.unpack('<I', self.data[4:8])[0]
        return True

    def parse_sections(self):
        offset = 8
        while offset < len(self.data):
            if offset >= len(self.data):
                break
            section_id = self.data[offset]
            offset += 1
            size, offset = read_leb128(self.data, offset)
            section_data = self.data[offset:offset + size]
            offset += size

            section_name = SECTION_NAMES.get(section_id, f'Unknown({section_id})')
            entropy = calculate_entropy(section_data)

            self.sections.append({
                'id': section_id,
                'name': section_name,
                'size': size,
                'entropy': entropy,
                'data': section_data
            })

    def extract_strings(self, min_length=4):
        strings = []
        current = []
        for byte in self.data:
            if 32 <= byte <= 126:
                current.append(chr(byte))
            else:
                if len(current) >= min_length:
                    strings.append(''.join(current))
                current = []
        if len(current) >= min_length:
            strings.append(''.join(current))
        return strings

    def check_mining_indicators(self, strings):
        found = []
        for s in strings:
            s_lower = s.lower()
            for keyword in MINING_KEYWORDS:
                if keyword in s_lower:
                    found.append({'string': s, 'keyword': keyword})
        return found

    def check_c2_indicators(self, strings):
        found = []
        for s in strings:
            for keyword in C2_KEYWORDS:
                if keyword in s.lower():
                    found.append({'string': s, 'keyword': keyword})
        return found

    def analyze_export_section(self):
        exports = []
        for section in self.sections:
            if section['id'] == 7:
                data = section['data']
                offset = 0
                count, offset = read_leb128(data, offset)
                for _ in range(count):
                    name_len, offset = read_leb128(data, offset)
                    name = data[offset:offset + name_len].decode('utf-8', errors='replace')
                    offset += name_len
                    kind = data[offset] if offset < len(data) else 0
                    offset += 1
                    idx, offset = read_leb128(data, offset)
                    exports.append({'name': name, 'kind': kind, 'index': idx})
        return exports

    def analyze_memory_section(self):
        memories = []
        for section in self.sections:
            if section['id'] == 5:
                data = section['data']
                offset = 0
                count, offset = read_leb128(data, offset)
                for _ in range(count):
                    flags = data[offset]; offset += 1
                    initial, offset = read_leb128(data, offset)
                    has_max = flags & 0x01
                    maximum = 0
                    if has_max:
                        maximum, offset = read_leb128(data, offset)
                    memories.append({
                        'initial_pages': initial,
                        'maximum_pages': maximum,
                        'initial_bytes': initial * 65536,
                        'has_max': has_max
                    })
        return memories

    def detect_obfuscation(self):
        indicators = []
        high_entropy_sections = [
            s for s in self.sections
            if s['entropy'] > 5.5 and s['size'] > 1024
        ]
        if high_entropy_sections:
            indicators.append({
                'type': 'HIGH_ENTROPY_SECTIONS',
                'severity': 'HIGH',
                'detail': f"{len(high_entropy_sections)} sections with entropy > 5.5"
            })

        code_sections = [s for s in self.sections if s['id'] == 10]
        if code_sections:
            code_size = sum(s['size'] for s in code_sections)
            ratio = code_size / self.file_size if self.file_size > 0 else 0
            if ratio > 0.9:
                indicators.append({
                    'type': 'HIGH_CODE_RATIO',
                    'severity': 'MEDIUM',
                    'detail': f"Code section is {ratio:.1%} of total file"
                })

        data_sections = [s for s in self.sections if s['id'] == 11]
        for ds in data_sections:
            if ds['size'] > 65536:
                indicators.append({
                    'type': 'LARGE_DATA_SECTION',
                    'severity': 'HIGH',
                    'detail': f"Data section is {ds['size']} bytes"
                })

        custom_sections = [s for s in self.sections if s['id'] == 0]
        for cs in custom_sections:
            if cs['entropy'] > 6.0:
                indicators.append({
                    'type': 'ENCRYPTED_CUSTOM_SECTION',
                    'severity': 'CRITICAL',
                    'detail': f"Custom section with entropy {cs['entropy']}"
                })

        return indicators

    def analyze(self):
        if not self.validate_magic():
            return {'error': 'Not a valid WASM file'}

        self.parse_sections()
        strings = self.extract_strings()
        mining = self.check_mining_indicators(strings)
        c2 = self.check_c2_indicators(strings)
        exports = self.analyze_export_section()
        memories = self.analyze_memory_section()
        obfuscation = self.detect_obfuscation()

        file_hash_md5 = hashlib.md5(self.data).hexdigest()
        file_hash_sha256 = hashlib.sha256(self.data).hexdigest()
        overall_entropy = calculate_entropy(self.data)

        result = {
            'file': self.filepath,
            'size': self.file_size,
            'md5': file_hash_md5,
            'sha256': file_hash_sha256,
            'overall_entropy': overall_entropy,
            'wasm_version': self.info.get('version', 0),
            'sections': [],
            'exports': exports,
            'memories': memories,
            'mining_indicators': mining,
            'c2_indicators': c2,
            'obfuscation_indicators': obfuscation,
            'strings_count': len(strings),
            'risk_score': 0
        }

        for s in self.sections:
            result['sections'].append({
                'name': s['name'],
                'id': s['id'],
                'size': s['size'],
                'entropy': s['entropy']
            })

        risk = 0
        risk += len(mining) * 15
        risk += len(c2) * 10
        risk += len([i for i in obfuscation if i['severity'] == 'CRITICAL']) * 25
        risk += len([i for i in obfuscation if i['severity'] == 'HIGH']) * 15
        risk += len([i for i in obfuscation if i['severity'] == 'MEDIUM']) * 8
        if overall_entropy > 6.0:
            risk += 20
        if any(m.get('initial_pages', 0) > 256 for m in memories):
            risk += 15

        result['risk_score'] = min(risk, 100)
        result['risk_level'] = (
            'CRITICAL' if risk >= 70 else
            'HIGH' if risk >= 40 else
            'MEDIUM' if risk >= 20 else
            'LOW'
        )

        return result


def main():
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <wasm_file_or_directory>")
        sys.exit(1)

    target = sys.argv[1]
    results = []

    if os.path.isfile(target):
        analyzer = WASMAnalyzer(target)
        results.append(analyzer.analyze())
    elif os.path.isdir(target):
        for root, dirs, files in os.walk(target):
            for f in files:
                if f.endswith('.wasm'):
                    filepath = os.path.join(root, f)
                    print(f"[*] Analyzing: {filepath}")
                    analyzer = WASMAnalyzer(filepath)
                    results.append(analyzer.analyze())
    else:
        print(f"[!] Target not found: {target}")
        sys.exit(1)

    for r in results:
        if 'error' in r:
            print(f"\n[!] {r['file']}: {r['error']}")
            continue

        print(f"\n{'='*60}")
        print(f"  WASM Analysis Report")
        print(f"{'='*60}")
        print(f"  File:           {r['file']}")
        print(f"  Size:           {r['size']} bytes")
        print(f"  MD5:            {r['md5']}")
        print(f"  SHA256:         {r['sha256']}")
        print(f"  Entropy:        {r['overall_entropy']}")
        print(f"  Risk Score:     {r['risk_score']}/100")
        print(f"  Risk Level:     {r['risk_level']}")
        print(f"{'='*60}")

        print(f"\n  Sections ({len(r['sections'])}):")
        for s in r['sections']:
            print(f"    [{s['id']:2d}] {s['name']:12s} size={s['size']:8d}  entropy={s['entropy']:.2f}")

        if r['exports']:
            print(f"\n  Exports ({len(r['exports'])}):")
            for e in r['exports'][:15]:
                kind_map = {0: 'Func', 1: 'Table', 2: 'Mem', 3: 'Global'}
                print(f"    {kind_map.get(e['kind'], 'Unknown'):8s} [{e['index']}] {e['name']}")
            if len(r['exports']) > 15:
                print(f"    ... and {len(r['exports'])-15} more")

        if r['memories']:
            print(f"\n  Memory:")
            for m in r['memories']:
                print(f"    Initial: {m['initial_pages']} pages ({m['initial_bytes']} bytes)")
                if m['has_max']:
                    print(f"    Maximum: {m['maximum_pages']} pages ({m['maximum_pages']*65536} bytes)")

        if r['mining_indicators']:
            print(f"\n  [!] MINING INDICATORS ({len(r['mining_indicators'])}):")
            for mi in r['mining_indicators']:
                print(f"    Keyword: {mi['keyword']} | String: {mi['string'][:80]}")

        if r['c2_indicators']:
            print(f"\n  [!] C2 INDICATORS ({len(r['c2_indicators'])}):")
            for ci in r['c2_indicators']:
                print(f"    Keyword: {ci['keyword']} | String: {ci['string'][:80]}")

        if r['obfuscation_indicators']:
            print(f"\n  [!] OBFUSCATION INDICATORS ({len(r['obfuscation_indicators'])}):")
            for oi in r['obfuscation_indicators']:
                print(f"    [{oi['severity']}] {oi['type']}: {oi['detail']}")

    output_path = "/tmp/wasm-analysis-results.json"
    with open(output_path, "w") as f:
        json.dump(results, f, indent=2, default=str)
    print(f"\n[+] Full results saved to {output_path}")


if __name__ == "__main__":
    main()

矿池域名:
  *.coinhive.com (已关闭)
  *.cryptoloot.pro
  *.cryptoloot.xy
  *.coin-imp.com
  *.webdollar.io
  *.miner.nablito.com

WASM文件特征:
  文件大小: 64KB - 256KB
  Magic: 0x0061736d
  Export函数数: 1-3
  Data Section特征: 包含大量高熵数据块（加密的矿工配置）
  Code/Data比例: > 0.7（代码占比高，典型挖矿特征）

网络指标:
  WebSocket连接: wss://域名:443
  Stratum消息模式: mining.subscribe, mining.authorize, mining.notify
  心跳间隔: 30-60秒
  请求体大小: 100-500 bytes（JSON-RPC）

浏览器存储:
  Service Worker注册: 包含.wasm加载逻辑的SW脚本
  IndexedDB: 存储矿工ID和矿池配置
  LocalStorage: mining_enabled, hashrate等键值

域名特征:
  常见伪装: jquery-cdn[.]net, googleapis-static[.]com
  域名注册时间: 攻击前1-3个月注册
  DNS托管: Cloudflare/其他CDN（隐藏真实IP）

网络指标:
  重定向链: 3-5次302重定向
  WASM请求: GET /assets/[random].wasm (Referer为水坑网站)
  User-Agent: 正常浏览器UA（无异常特征）
  请求频率: 单次请求（每用户只触发一次）

WASM模块特征:
  模块大小: 200KB-1.5MB（指纹模块约20KB）
  Export函数: 通常1个（执行入口函数）
  Import函数: 包含WebAssembly.Memory和多个JS函数导入
  Data Section: 包含加密的URL字符串和配置数据
  特征字符串: XOR加密，解密后包含漏洞利用相关的路径和参数

后门指标:
  持久化路径: %APPDATA%\Microsoft\[random].dll
  注册表键: HKCU\Software\Microsoft\Windows\CurrentVersion\Run
  C2通信: HTTPS POST，Base64编码的JSON数据
  C2间隔: 600秒（10分钟）心跳