网络协议滥用与流量取证深度分析

采集（Capture）→ 保留（Preservation）→ 分析（Analysis）→ 报告（Reporting）

Wireshark/tshark    → 交互式分析 & 命令行批处理
Zeek (Bro)          → 协议解析 & 连接日志
Suricata            → IDS/IPS & 流量元数据提取
Arkime (Moloch)     → 全流量存储 & 检索
tcpdump             → 轻量级抓包
NetworkMiner        → 被动流量解析 & 文件提取
ChopShop            → 协议解码框架
Xplico              → 开源取证分析平台

tshark -r capture.pcap -Y "dns" -T fields \
  -e frame.time -e ip.src -e ip.dst \
  -e dns.qry.name -e dns.qry.type \
  -e dns.a -E separator="|"

tshark -r capture.pcap -Y "http.request" -T fields \
  -e frame.time -e ip.src -e ip.dst \
  -e http.host -e http.request.uri \
  -e http.user_agent

tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields \
  -e frame.time -e ip.src -e ip.dst \
  -e tls.handshake.extensions_server_name

tcpdump -i eth0 -w /data/capture_%Y%m%d_%H%M.pcap \
  -G 3600 -W 168 \
  -s 0 \
  'not (port 53 or port 123 or port 161)'

tcpdump -i any -w suspicious.pcap \
  'host 10.0.0.0/8 and not net 10.0.0.0/24'

conn.log      → 连接元数据（五元组、时长、字节数）
dns.log       → DNS 查询/响应
http.log      → HTTP 请求/响应元数据
ssl.log       → TLS 握手元数据
files.log     → 文件传输记录
notice.log    → 异常告警

chronyc sources -v
ntpq -p
timedatectl status

sha256sum capture.pcap > capture.pcap.sha256
sha256sum -c capture.pcap.sha256

editcap -F pcapng capture.pcap capture_clean.pcapng
capinfos capture.pcap

每主机日均 DNS 查询量
查询类型分布（A/AAAA/MX/TXT/SRV/PTR 比例）
平均子域名长度
平均查询响应时间
Top-N 被查询域名
NXDOMAIN 比率

cat dns.log | jq -r 'select(.qtype_name == "A") | .id.orig_h' | \
  sort | uniq -c | sort -rn | head -20

cat dns.log | jq -r '.query' | \
  awk -F'.' '{print NF-1}' | sort | uniq -c | sort -rn

import math
from collections import Counter

def shannon_entropy(data):
    if not data:
        return 0
    counter = Counter(data)
    entropy = 0.0
    for count in counter.values():
        prob = count / len(data)
        entropy -= prob * math.log2(prob)
    return entropy

def analyze_dns_query(query):
    parts = query.rstrip('.').split('.')
    if len(parts) <= 2:
        return 0.0
    subdomain = '.'.join(parts[:-2])
    return shannon_entropy(subdomain)

queries = [
    "aGVsbG8gd29ybGQ.tunnel.evil.com",
    "www.google.com",
    "c3RyaW5n.data.exfil.attacker.com",
    "mail.office365.com"
]

for q in queries:
    ent = analyze_dns_query(q)
    flag = " [ANOMALY]" if ent > 3.5 else ""
    print(f"{q:50s} entropy={ent:.4f}{flag}")

module DNSAnomaly;

export {
    redef enum Notice::Type += {
        DNS_High_Entropy_Query,
        DNS_Excessive_Query_Volume,
        DNS_Unusual_Record_Type
    };
    const entropy_threshold = 3.5 &redef;
    const query_volume_threshold = 100 &redef;
    const suspicious_types: set[string] = {"NULL", "TXT", "CNAME", "MX"} &redef;
}

global query_counts: table[addr] of count &create_expire=1hr;
global txt_counts: table[addr] of count &create_expire=1hr;

event dns_request(c: connection, msg: dns_msg, query: string, qtype: count, qclass: count)
    {
    local orig = c$id$orig_h;
    query_counts[orig] = (orig in query_counts) ? query_counts[orig] + 1 : 1;

    if (query_counts[orig] > query_volume_threshold)
        {
        NOTICE([$note=DNS_Excessive_Query_Volume,
                $conn=c,
                $msg=fmt("Host %s made %d DNS queries in 1hr", orig, query_counts[orig])]);
        }

    local qtype_name = dns_lookup_type_name(qtype);
    if (qtype_name in suspicious_types)
        {
        txt_counts[orig] = (orig in txt_counts) ? txt_counts[orig] + 1 : 1;
        }
    }

import statistics

def detect_beacon(timestamps, threshold_cv=0.1):
    if len(timestamps) < 5:
        return False
    intervals = [timestamps[i+1] - timestamps[i]
                 for i in range(len(timestamps)-1)]
    if not intervals:
        return False
    mean = statistics.mean(intervals)
    stdev = statistics.stdev(intervals) if len(intervals) > 1 else 0
    cv = stdev / mean if mean > 0 else float('inf')
    return cv < threshold_cv

timestamps = [1000, 1060, 1120, 1180, 1240, 1300, 1360]
print(f"Beacon detected: {detect_beacon(timestamps)}")

tshark -r capture.pcap -Y "dns" -T fields \
  -e dns.qry.name -e dns.a -e dns.resp.type | \
  awk -F'\t' '{print $1}' | \
  grep -E '\.(duckdns\.org|no-ip\.com|dynu\.com|freedns\.com)$'

tshark -r capture.pcap -Y "tcp.port == 443 && tls.handshake.type == 1" \
  -T fields -e tls.handshake.extensions_server_name | \
  grep -iE '(dns\.google|cloudflare-dns\.com|dns\.quad9)'

tshark -r capture.pcap -Y "udp.port == 853" -c 10

import re

def detect_encoding(subdomain):
    subdomain = subdomain.lower().rstrip('=')
    if re.match(r'^[0-9a-f]+$', subdomain) and len(subdomain) % 2 == 0:
        return "Hex"
    if re.match(r'^[a-z2-7]+$', subdomain) and len(subdomain) >= 16:
        return "Base32"
    if re.match(r'^[a-z0-9+/]+=*$', subdomain, re.IGNORECASE):
        return "Base64"
    return "Unknown/Raw"

samples = [
    "68656c6c6f776f726c64",
    "NBSWY3DPEB2W64TMMQ",
    "aGVsbG8gd29ybGQ=",
    "random-looking-subdomain"
]
for s in samples:
    print(f"{s:35s} -> {detect_encoding(s)}")

dig @target_dns evil.com A +norecurse
dig @target_dns evil.com A +recurse

tshark -r capture.pcap -Y "dns.flags.response == 1 && dns.flags.authoritative == 0" \
  -T fields -e dns.qry.name -e dns.a -e dns.resp.ttl

def detect_rebind(dns_responses, domain):
    ips = set()
    for resp in dns_responses:
        if resp['domain'] == domain:
            for ip in resp.get('answers', []):
                ips.add(ip)
    if len(ips) > 1:
        has_internal = any(ip.startswith(('10.', '172.16.', '192.168.')) for ip in ips)
        has_external = any(not ip.startswith(('10.', '172.16.', '192.168.')) for ip in ips)
        if has_internal and has_external:
            return True, ips
    return False, ips

# 典型 Cobalt Strike Malleable Profile
set sleeptime "30000";
set jitter    "10";
set useragent "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36";

http-get {
    set uri "/s/update";
    client {
        header "Accept" "text/html,application/xhtml+xml";
        metadata {
            base64;
            prepend "session=";
            header "Cookie";
        }
    }
    server {
        header "Content-Type" "application/octet-stream";
        output {
            base64;
            print;
        }
    }
}

tshark -r capture.pcap -Y "http.request" -T fields \
  -e http.host -e http.request.uri -e http.user_agent \
  -e http.cookie -e http.content_type | \
  grep -E "(submit\.php|pixel\.gif|__cf|/visit)"

tshark -r capture.pcap -Y "http.request.method == CONNECT" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e http.request.uri -e http.request.host

tshark -r capture.pcap -Y "http.request.method == CONNECT" \
  -T fields -e http.request.uri | \
  awk -F: '{print $2}' | sort | uniq -c | sort -rn | head -20

def detect_long_polling(http_logs, duration_threshold=30):
    sessions = {}
    for log in http_logs:
        key = (log['src_ip'], log['dst_ip'])
        if key not in sessions:
            sessions[key] = []
        sessions[key].append(log)

    anomalies = []
    for key, reqs in sessions.items():
        reqs.sort(key=lambda x: x['timestamp'])
        for i in range(len(reqs) - 1):
            gap = reqs[i+1]['timestamp'] - reqs[i]['timestamp']
            if gap > duration_threshold:
                anomalies.append({
                    'session': key,
                    'gap_seconds': gap,
                    'uri': reqs[i]['uri']
                })
    return anomalies

tshark -r capture.pcap -Y "http.transfer_encoding == chunked" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e http.request.uri -e http.content_length

tshark -r capture.pcap -Y "tls.handshake.type == 1" \
  -T fields -e tls.handshake.extensions_server_name \
  -e ip.src -e ip.dst | sort | uniq -c | sort -rn | head -30

tshark -r capture.pcap -Y "tls.handshake.type == 11" \
  -T fields -e tls.handshake.cert_hash \
  -e x509sat.printableString \
  -e x509af.issuer

tshark -r capture.pcap -Y "tls" \
  -T fields -e tls.cert.hash | sort -u

tshark -r capture.pcap -Y "tls.handshake.type == 1" \
  -T fields \
  -e tls.handshake.extensions_support_groups \
  -e tls.handshake.extensions_ec_point_formats \
  -e tls.handshake.ciphersuite

def detect_exfiltration(connections, threshold_bytes=10_000_000):
    upload_stats = {}
    for conn in connections:
        key = conn['dst_ip']
        if key not in upload_stats:
            upload_stats[key] = {'total': 0, 'count': 0}
        upload_stats[key]['total'] += conn['bytes_sent']
        upload_stats[key]['count'] += 1

    anomalies = []
    for ip, stats in upload_stats.items():
        avg = stats['total'] / stats['count'] if stats['count'] else 0
        if stats['total'] > threshold_bytes:
            anomalies.append({
                'dst_ip': ip,
                'total_uploaded': stats['total'],
                'avg_per_conn': avg
            })
    return sorted(anomalies, key=lambda x: x['total_uploaded'], reverse=True)

tshark -r capture.pcap -Y "http.request.method == POST" \
  -T fields -e http.request.uri -e http.content_length \
  -e http.content_type | \
  awk -F'\t' '$2 > 100000 {print}' | sort -t$'\t' -k2 -rn

tshark -r capture.pcap -Y "http.request.method == POST && http.content_type == application/x-www-form-urlencoded" \
  -T fields -e http.request.uri -e http.file_data | \
  grep -iE '(cmd|exec|shell|pass|eval|system|assert)'

import json
import re
from collections import defaultdict
from datetime import datetime

class HTTPAnomalyDetector:
    def __init__(self):
        self.uri_counts = defaultdict(int)
        self.ua_counts = defaultdict(int)
        self.host_connections = defaultdict(set)
        self.post_sizes = defaultdict(list)
        self.session_tracker = defaultdict(list)

    def process_request(self, timestamp, src_ip, dst_ip, method,
                        uri, host, user_agent, content_length, content_type):
        self.uri_counts[uri] += 1
        self.ua_counts[user_agent] += 1
        self.host_connections[host].add(src_ip)

        if method == "POST" and content_length:
            self.post_sizes[host].append(int(content_length))

        self.session_tracker[(src_ip, dst_ip)].append({
            'timestamp': timestamp,
            'method': method,
            'uri': uri,
            'host': host
        })

    def detect_anomalies(self):
        results = []

        for uri, count in self.uri_counts.items():
            if re.match(r'^/[a-z]{1,3}\.(php|asp|jsp)$', uri) and count > 10:
                results.append({
                    'type': 'WEBSHELL_URI',
                    'severity': 'HIGH',
                    'uri': uri,
                    'count': count
                })

        for ua, count in self.ua_counts.items():
            if not ua or ua == "-" or len(ua) < 10:
                results.append({
                    'type': 'SUSPICIOUS_UA',
                    'severity': 'MEDIUM',
                    'user_agent': ua,
                    'count': count
                })

        for host, sizes in self.post_sizes.items():
            if sizes:
                avg_size = sum(sizes) / len(sizes)
                max_size = max(sizes)
                if max_size > avg_size * 10 and max_size > 1_000_000:
                    results.append({
                        'type': 'DATA_EXFILTRATION',
                        'severity': 'HIGH',
                        'host': host,
                        'max_post_size': max_size,
                        'avg_post_size': avg_size
                    })

        for (src, dst), reqs in self.session_tracker.items():
            if len(reqs) > 50:
                methods = [r['method'] for r in reqs]
                post_ratio = methods.count('POST') / len(methods)
                if post_ratio > 0.8:
                    results.append({
                        'type': 'C2_BEACON_PATTERN',
                        'severity': 'HIGH',
                        'src': src,
                        'dst': dst,
                        'request_count': len(reqs),
                        'post_ratio': post_ratio
                    })

        return results

detector = HTTPAnomalyDetector()
print("[*] HTTP Anomaly Detector initialized")

tshark -r capture.pcap -Y "smb2.cmd == 5" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e smb2.share_name -e smb2.filename

tshark -r capture.pcap -Y "smb2.cmd == 5 && smb2.share_name contains ADMIN" \
  -T fields -e ip.src -e ip.dst -e smb2.share_name

tshark -r capture.pcap -Y "smb2.cmd == 4 && smb2.filename contains pipe" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e smb2.filename | sort | uniq -c | sort -rn

tshark -r capture.pcap -Y "dcerpc" \
  -T fields -e dcerpc.cn_ctx.uuid | sort | uniq -c | sort -rn

tshark -r capture.pcap -Y "dcerpc.cn_ctx.uuid == 3305fd00-01f5-11a3-8c89-00805f8e962a" \
  -T fields -e frame.time -e ip.src -e ip.dst

tshark -r capture.pcap -Y "smb2 && dcerpc" \
  -T fields -e ip.src -e ip.dst -e smb2.filename \
  -e dcerpc.cn_ctx.uuid | grep -i "msagent\|winmgmt\|wmimap"

tshark -r capture.pcap -Y "ntlmssp" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e ntlmssp.auth.domain \
  -e ntlmssp.auth.username \
  -e ntlmssp.auth.ntlmv2_response

tshark -r capture.pcap -Y "kerberos && kerberos.msg_type == 12" \
  -T fields -e ip.src -e ip.dst \
  -e kerberos.CNameString

def detect_smb_exfiltration(zeek_smb_files):
    file_transfers = {}
    for entry in zeek_smb_files:
        key = (entry['id_orig_h'], entry['id_resp_h'])
        if key not in file_transfers:
            file_transfers[key] = []
        file_transfers[key].append({
            'filename': entry.get('filename', ''),
            'size': entry.get('total_bytes', 0),
            'ts': entry['ts']
        })

    anomalies = []
    for pair, files in file_transfers.items():
        total_size = sum(f['size'] for f in files)
        if total_size > 100_000_000:
            anomalies.append({
                'src': pair[0],
                'dst': pair[1],
                'total_bytes': total_size,
                'file_count': len(files),
                'filenames': [f['filename'] for f in files[:10]]
            })
    return anomalies

cat smb_files.log | jq -r 'select(.action == "SMB::FILE_OPEN") |
  [.ts, .id.orig_h, .id.resp_h, .path, .name, .size] | @tsv'

cat smb_mapping.log | jq -r 'select(.share_type == "DISK") |
  [.id.orig_h, .id.resp_h, .path, .share_type] | @tsv'

title: SMB Admin Share Access from Unusual Source
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: 检测到从非管理员工作站访问 ADMIN$ 共享的 SMB 流量
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: smb_files
detection:
    selection:
        action: "SMB::FILE_OPEN"
        share|contains: "ADMIN"
    filter:
        id.orig_h|cidr:
            - "10.0.1.0/24"
    condition: selection and not filter
fields:
    - id.orig_h
    - id.resp_h
    - path
    - name
level: high
tags:
    - attack.lateral_movement
    - attack.t1021.002

title: Multiple SMB Named Pipe Connections
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
status: experimental
description: 检测到短时间内对多个目标建立 Named Pipe 连接
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: smb_files
detection:
    selection:
        name|contains:
            - "svcctl"
            - "atsvc"
            - "msagent"
    timeframe: 5m
    condition: selection | count(id.resp_h) by id.orig_h > 5
fields:
    - id.orig_h
    - id.resp_h
    - name
level: high
tags:
    - attack.lateral_movement
    - attack.t1021

BaseObject 查询 → 根 DSE 查询（用于发现 DC 能力）
Single Level 查询 → 枚举 OU 下的直接子对象
Subtree 查询 → 递归搜索（最常用）

1. (objectclass=domain)                          → 域信息枚举
2. (objectclass=*)  subtree base=""              → 全量对象枚举
3. (|(samtype=268435456)(samtype=268435457))     → 组枚举
4. (objectclass=gplink)                           → GPO 链接枚举
5. (objectclass=domainDNS)                        → 域信任枚举
6. (samAccountType=805306368)                     → 用户枚举
7. (samAccountType=805306369)                     → 机器账户枚举

tshark -r capture.pcap -Y "ldap" \
  -T fields -e ip.src -e ip.dst \
  -e ldap.search.baseObject \
  -e ldap.search.filter \
  -e ldap.search.scope | \
  grep -E "(objectclass=\*|samAccountType|samtype)"

tshark -r capture.pcap -Y "dcerpc.cn_ctx.uuid == e3514235-4b06-11d1-ab04-00c04fc2dcd2" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e dcerpc.cn_ctx.uuid

tshark -r capture.pcap -Y "drs" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e drs.opnum

def detect_dcsync(zeek_dce_rpc_logs, domain_controller_ips):
    dcsync_candidates = {}

    for entry in zeek_dce_rpc_logs:
        if entry.get('named_pipe') != '\\PIPE\\lsarpc':
            continue

        src_ip = entry['id.orig_h']
        dst_ip = entry['id.resp_h']

        if src_ip in domain_controller_ips:
            continue

        key = (src_ip, dst_ip)
        if key not in dcsync_candidates:
            dcsync_candidates[key] = {
                'bind_count': 0,
                'getnc_changes_count': 0,
                'first_seen': entry['ts'],
                'last_seen': entry['ts']
            }

        if entry.get('endpoint') == 'drsuuid':
            dcsync_candidates[key]['getnc_changes_count'] += 1
            dcsync_candidates[key]['last_seen'] = entry['ts']

    alerts = []
    for (src, dst), data in dcsync_candidates.items():
        if data['getnc_changes_count'] > 0:
            alerts.append({
                'severity': 'CRITICAL',
                'type': 'DCSYNC_DETECTED',
                'source': src,
                'target_dc': dst,
                'replication_requests': data['getnc_changes_count'],
                'first_seen': data['first_seen'],
                'last_seen': data['last_seen']
            })
    return alerts

*)(objectClass=*           → 通配符注入
admin)(|(password=*        → 认证绕过
*)((|mail=*)               → 信息泄露
*)(|(objectClass=*))       → 全对象枚举

tshark -r capture.pcap -Y "ldap" \
  -T fields -e ldap.search.filter | \
  grep -iE '(\*\)|\(\|\||\)\(\||\)\(\*)'

from collections import defaultdict
import re

class LDAPAnomalyDetector:
    def __init__(self):
        self.query_tracker = defaultdict(list)
        self.suspicious_filters = [
            r'objectclass=\*',
            r'samAccountType',
            r'servicePrincipalName',
            r'userAccountControl.*1\.2\.840',
            r'nTSecurityDescriptor',
            r'ms-MCS-AdmPwd'
        ]

    def process_ldap_query(self, src_ip, dst_ip, base_object,
                           search_filter, scope, attributes):
        query_record = {
            'base': base_object,
            'filter': search_filter,
            'scope': scope,
            'attrs': attributes
        }
        self.query_tracker[(src_ip, dst_ip)].append(query_record)

    def analyze(self):
        alerts = []
        for (src, dst), queries in self.query_tracker.items():
            if len(queries) > 200:
                alerts.append({
                    'type': 'EXCESSIVE_LDAP_QUERIES',
                    'severity': 'HIGH',
                    'src': src,
                    'dst': dst,
                    'count': len(queries),
                    'description': 'Possible BloodHound/AD enumeration'
                })

            for q in queries:
                for pattern in self.suspicious_filters:
                    if re.search(pattern, q['filter'], re.IGNORECASE):
                        alerts.append({
                            'type': 'SUSPICIOUS_LDAP_FILTER',
                            'severity': 'MEDIUM',
                            'src': src,
                            'dst': dst,
                            'filter': q['filter'],
                            'pattern_matched': pattern
                        })
                        break
        return alerts

应用层        → RDP 协议（T.128）
表示层        → T.125 MCS（Multi-point Communication Service）
传输层        → TPKT (RFC 1006) / TCP
安全层        → TLS / CredSSP (NLA) / RDP Standard Encryption

tshark -r capture.pcap -Y "tcp.port == 3389" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e tcp.len -e tcp.flags | head -100

tshark -r capture.pcap -Y "tpkt" \
  -T fields -e ip.src -e ip.dst -e tpkt.length | \
  awk '$3 < 30 {print}' | sort | uniq -c | sort -rn

特征 1：短时间内大量 TCP 连接建立（SYN → SYN/ACK → ACK）
特征 2：每个连接持续时间极短（< 5 秒）
特征 3：TPKT 包长度固定且很小（协商阶段）
特征 4：连接失败后快速重连（间隔 < 2 秒）
特征 5：无后续数据传输（认证失败即断开）

cat rdp.log | jq -r 'select(.cookie != null) |
  [.ts, .id.orig_h, .id.resp_h, .cookie, .result] | @tsv' | \
  sort | awk -F'\t' '{print $2}' | uniq -c | sort -rn | head -20

tshark -r capture.pcap -Y "tcp.port == 3389" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e tcp.len -e tcp.flags -e rdp.nego_req_type

tshark -r capture.pcap -Y "tcp.port == 3389 && tcp.len > 1000 && tcp.flags == 0x018" \
  -T fields -e ip.src -e ip.dst -e tcp.len

tscon <session_id> /dest:console

tshark -r capture.pcap -Y "tcp.port == 3389" \
  -T fields -e ip.src -e ip.dst -e tcp.stream | \
  sort -t$'\t' -k3 -n | uniq -c

tshark -r capture.pcap -Y "tcp.port == 3389" \
  -T fields -e ip.src -e ip.dst | \
  awk -F'\t' '{print $1}' | sort | uniq -c | sort -rn

tshark -r capture.pcap -Y "tcp.port == 8080 || tcp.port == 7000" \
  -T fields -e ip.src -e ip.dst -e tcp.len | \
  awk '$3 > 100 {print}' | head -20

Event ID 4624 (Logon Type 10) → RDP 登录成功
Event ID 4625 (Logon Type 10) → RDP 登录失败
Event ID 1149 (TerminalServices-RemoteConnectionManager) → RDP 会话建立
Event ID 25  (Microsoft-Windows-TerminalServices-LocalSessionManager) → 会话重连

%LOCALAPPDATA%\Microsoft\Terminal Server Client\Cache\
  → bmc.cache (Bitmap Cache)
  → cache*.bin (Persistent Bitmap Cache)

title: RDP Brute Force Detection
id: c3d4e5f6-a7b8-9012-cdef-234567890abc
status: experimental
description: 检测短时间内大量 RDP 连接尝试
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: rdp
detection:
    selection:
        result: "failed"
    timeframe: 10m
    condition: selection | count(id.orig_h) by id.resp_h > 20
fields:
    - id.orig_h
    - id.resp_h
    - cookie
    - result
level: high
tags:
    - attack.credential_access
    - attack.t1110

title: RDP from External Network
id: d4e5f6a7-b8c9-0123-defa-345678901bcd
status: experimental
description: 检测来自外部网络的 RDP 连接
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: rdp
detection:
    selection:
        id.orig_h|cidr:
            - "0.0.0.0/0"
    filter_internal:
        id.orig_h|cidr:
            - "10.0.0.0/8"
            - "172.16.0.0/12"
            - "192.168.0.0/16"
    condition: selection and not filter_internal
fields:
    - id.orig_h
    - id.resp_h
    - cookie
level: critical
tags:
    - attack.initial_access
    - attack.t1133

正常 ICMP 流量特征：
- Echo Request/Reply 对（ping）
- 数据包大小固定（通常 32/64/128 字节）
- 低频率（偶尔的连通性检测）
- Payload 通常为填充字节（0x00 或递增序列）

tshark -r capture.pcap -Y "icmp" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e icmp.type -e icmp.code -e data.len -e data.data | \
  awk -F'\t' '$5 > 64 {print}' | head -50

tshark -r capture.pcap -Y "icmp && data.len > 0" \
  -T fields -e data.data | head -20

from collections import defaultdict
import math
import re

class ICMPTunnelDetector:
    def __init__(self):
        self.icmp_tracker = defaultdict(list)
        self.size_anomalies = []
        self.payload_anomalies = []

    def process_icmp(self, timestamp, src_ip, dst_ip,
                     icmp_type, data_len, data_hex):
        key = (src_ip, dst_ip)
        self.icmp_tracker[key].append({
            'ts': timestamp,
            'type': icmp_type,
            'data_len': data_len,
            'data_hex': data_hex
        })

        if data_len > 64:
            self.size_anomalies.append({
                'src': src_ip, 'dst': dst_ip,
                'size': data_len, 'ts': timestamp
            })

        if data_hex and len(data_hex) > 4:
            data_bytes = bytes.fromhex(data_hex[:100])
            entropy = self._calc_entropy(data_bytes)
            if entropy > 5.0:
                self.payload_anomalies.append({
                    'src': src_ip, 'dst': dst_ip,
                    'entropy': entropy, 'ts': timestamp
                })

    def _calc_entropy(self, data):
        if not data:
            return 0.0
        freq = defaultdict(int)
        for byte in data:
            freq[byte] += 1
        entropy = 0.0
        for count in freq.values():
            p = count / len(data)
            entropy -= p * math.log2(p)
        return entropy

    def analyze(self):
        alerts = []
        for (src, dst), packets in self.icmp_tracker.items():
            if len(packets) > 100:
                alerts.append({
                    'type': 'ICMP_TUNNEL_SUSPECTED',
                    'severity': 'HIGH',
                    'src': src, 'dst': dst,
                    'packet_count': len(packets),
                    'reason': 'Excessive ICMP packet count'
                })

            echo_types = [p['type'] for p in packets]
            non_echo = [t for t in echo_types if t not in (0, 8)]
            if non_echo:
                alerts.append({
                    'type': 'ICMP_UNUSUAL_TYPE',
                    'severity': 'MEDIUM',
                    'src': src, 'dst': dst,
                    'unusual_types': list(set(non_echo))
                })

        for anomaly in self.size_anomalies:
            alerts.append({
                'type': 'ICMP_LARGE_PAYLOAD',
                'severity': 'HIGH',
                **anomaly
            })

        return alerts

tshark -r capture.pcap -Y "dhcp" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e dhcp.option.type -e dhcp.option.value | head -30

tshark -r capture.pcap -Y "dhcp && dhcp.option.length > 100" \
  -T fields -e ip.src -e ip.dst -e dhcp.option.length

tshark -r capture.pcap -Y "6to4" \
  -T fields -e frame.time -e ipv6.src -e ipv6.dst

tshark -r capture.pcap -Y "teredo" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e teredo.orig_dst

tshark -r capture.pcap -Y "isatap" \
  -T fields -e frame.time -e ipv6.src -e ipv6.dst

tshark -r capture.pcap -Y "ip.ip" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e ip.proto -e ip.ip.proto

tshark -r capture.pcap -Y "gre" \
  -T fields -e frame.time -e ip.src -e ip.dst \
  -e gre.proto

tshark -r capture.pcap -Y "http" \
  -T fields -e tcp.dstport | sort | uniq -c | sort -rn | \
  awk '$2 != 80 && $2 != 8080 && $2 != 443 && $2 != 8443 {print}'

tshark -r capture.pcap -Y "dns" \
  -T fields -e udp.dstport | sort | uniq -c | sort -rn | \
  awk '$2 != 53 {print}'

tshark -r capture.pcap -Y "ssh" \
  -T fields -e tcp.dstport | sort | uniq -c | sort -rn | \
  awk '$2 != 22 {print}'

TLS Version, Accepted Ciphers, List of Extensions,
Elliptic Curves, Elliptic Curve Point Formats

↓ 以逗号分隔，去除 SNI 扩展（ID=0）

JA3 = MD5("769,4865-4866-4867-49196-49200,...,0-5-10-11-...,23-24-25,0")

curl -s "https://ja3er.com/search/<JA3_HASH>"
curl -s "https://ja3er.com/search/4d7a2d269e3e33c5a5e8f4a7b2c1d0e9"

JA3S = MD5(TLSVersion, Cipher, List of Extensions)

t13d1515h2_8daaf615e7ed_24e7b
│   │        │         │
│   │        │         └─ HTTP/2 SETTINGS 参数哈希
│   │        └─ 扩展列表哈希（不含 SNI/ALPN）
│   └─ TLS 版本 + 密码套件数量 + SNI 存在
└─ 协议类型（t=tls, q=quic, h=http）

tshark -r capture.pcap -Y "tls.handshake.type == 11" \
  -T fields -e ip.src -e ip.dst \
  -e x509sat.printableString \
  -e x509af.issuer \
  -e x509af.subject

tshark -r capture.pcap -Y "tls" \
  -T fields -e tls.cert.hash -e x509af.issuer \
  -e x509af.subject | \
  awk -F'\t' '$2 == $3 {print "SELF-SIGNED: "$0}'

tshark -r capture.pcap -Y "tls.handshake.type == 11" \
  -T fields -e x509af.notAfter -e x509af.notBefore \
  -e x509sat.printableString

curl -s "https://crt.sh/?q=%.evil.com&output=json" | \
  jq -r '.[] | [.id, .name_value, .issuer_ca_id, .not_before, .not_after] | @tsv'

Let's Encrypt 证书特征：
- Issuer: CN=R3, O=Let's Encrypt, C=US（或类似）
- 有效期：90 天
- 免费申请，无身份验证（DV 证书）
- 攻击者常利用其频繁申请新证书

tshark -r capture.pcap -Y "tls.handshake.type == 2" \
  -T fields -e tls.handshake.version | \
  sort | uniq -c | sort -rn

tshark -r capture.pcap -Y "tls.record.version == 0x0301" \
  -T fields -e ip.src -e ip.dst | head -20

tshark -r capture.pcap -Y "tls.handshake.type == 2" \
  -T fields -e tls.handshake.ciphersuite | \
  grep -E "(0x002f|0x0035|0x003c|0x003d|0x0004|0x0005)"

import hashlib
import struct

class JA3Extractor:
    KNOWN_MALICIOUS = {
        "4d7a2d269e3e33c5a5e8f4a7b2c1d0e9": "Cobalt Strike (default)",
        "72a589da586844d7f0818ce684948eea": "Metasploit",
        "a0e9f5d64349fb13191bc781f81f42e1": "Impacket",
        "cd08e31494f9531f560d64c695b072a3": "Python requests",
        "50602055a14b56a5e97a5367437f2d73": "curl",
        "e7d705a3286e19ea42f587b344ee6865": "Go HTTP client",
    }

    def __init__(self):
        self.results = []

    def extract_from_tshark_output(self, tshark_fields):
        for line in tshark_fields:
            fields = line.strip().split('\t')
            if len(fields) < 4:
                continue

            tls_version = fields[0]
            ciphers = fields[1]
            extensions = fields[2]
            curves = fields[3] if len(fields) > 3 else ""
            point_formats = fields[4] if len(fields) > 4 else ""

            filtered_extensions = self._filter_extensions(extensions)

            ja3_str = f"{tls_version},{ciphers},{filtered_extensions},{curves},{point_formats}"
            ja3_hash = hashlib.md5(ja3_str.encode()).hexdigest()

            match = self.KNOWN_MALICIOUS.get(ja3_hash, "Unknown")
            self.results.append({
                'ja3_hash': ja3_hash,
                'ja3_string': ja3_str,
                'match': match,
                'is_malicious': ja3_hash in self.KNOWN_MALICIOUS
            })

        return self.results

    def _filter_extensions(self, extensions_str):
        if not extensions_str:
            return ""
        ext_list = extensions_str.split('-')
        filtered = [e for e in ext_list if e not in ('0', '35')]
        return '-'.join(filtered)

    def print_report(self):
        print(f"\n{'='*70}")
        print(f"JA3 Fingerprint Analysis Report")
        print(f"{'='*70}")
        for r in self.results:
            flag = " [MALICIOUS]" if r['is_malicious'] else ""
            print(f"\nJA3: {r['ja3_hash']}{flag}")
            print(f"  Match: {r['match']}")
            print(f"  String: {r['ja3_string'][:100]}...")

extractor = JA3Extractor()
print("[*] JA3 Extractor initialized")

zeek -r capture.pcap local policy/protocols/conn

cat conn.log | jq -r 'select(.proto == "tcp") |
  [.ts, .uid, .id.orig_h, .id.orig_p, .id.resp_h, .id.resp_p,
   .duration, .orig_bytes, .resp_bytes] | @tsv'

cat http.log | jq -r '[.ts, .uid, .id.orig_h, .method, .host, .uri,
  .user_agent, .status_code, .request_body_len, .response_body_len] | @tsv'

cat ssl.log | jq -r '[.ts, .uid, .id.orig_h, .server_name,
  .subject, .issuer, .ja3, .ja3s] | @tsv'

import json
from collections import defaultdict

class TrafficCorrelator:
    def __init__(self):
        self.connections = {}
        self.http_sessions = defaultdict(list)
        self.dns_sessions = defaultdict(list)
        self.ssl_sessions = defaultdict(list)
        self.timeline = []

    def load_conn_log(self, log_path):
        with open(log_path) as f:
            for line in f:
                if line.startswith('#'):
                    continue
                entry = json.loads(line)
                uid = entry.get('uid', '')
                self.connections[uid] = entry
                self.timeline.append({
                    'ts': entry['ts'],
                    'uid': uid,
                    'type': 'conn',
                    'src': entry['id.orig_h'],
                    'dst': entry['id.resp_h']
                })

    def load_http_log(self, log_path):
        with open(log_path) as f:
            for line in f:
                if line.startswith('#'):
                    continue
                entry = json.loads(line)
                uid = entry.get('uid', '')
                self.http_sessions[uid].append(entry)
                self.timeline.append({
                    'ts': entry['ts'],
                    'uid': uid,
                    'type': 'http',
                    'method': entry.get('method', ''),
                    'host': entry.get('host', ''),
                    'uri': entry.get('uri', '')
                })

    def load_dns_log(self, log_path):
        with open(log_path) as f:
            for line in f:
                if line.startswith('#'):
                    continue
                entry = json.loads(line)
                uid = entry.get('uid', '')
                self.dns_sessions[uid].append(entry)
                self.timeline.append({
                    'ts': entry['ts'],
                    'uid': uid,
                    'type': 'dns',
                    'query': entry.get('query', ''),
                    'qtype': entry.get('qtype_name', '')
                })

    def build_attack_timeline(self, suspicious_uids):
        events = []
        for uid in suspicious_uids:
            if uid in self.connections:
                conn = self.connections[uid]
                events.append({
                    'ts': conn['ts'],
                    'event': f"Connection {conn['id.orig_h']}:{conn['id.orig_p']} -> {conn['id.resp_h']}:{conn['id.resp_p']}",
                    'uid': uid
                })
            for http in self.http_sessions.get(uid, []):
                events.append({
                    'ts': http['ts'],
                    'event': f"HTTP {http.get('method','')} {http.get('host','')}{http.get('uri','')}",
                    'uid': uid
                })
            for dns in self.dns_sessions.get(uid, []):
                events.append({
                    'ts': dns['ts'],
                    'event': f"DNS {dns.get('qtype_name','')} {dns.get('query','')}",
                    'uid': uid
                })
        events.sort(key=lambda x: x['ts'])
        return events

correlator = TrafficCorrelator()
print("[*] Traffic Correlator initialized")

T+0s    DNS 查询 C2 域名（A 记录）
T+1s    TLS 握手连接 C2 服务器（JA3 匹配 Cobalt Strike）
T+2s    HTTP GET /submit.php（Beacon 首次通信）
T+60s   HTTP POST /submit.php（回传系统信息）
T+300s  DNS 查询内部主机名
T+310s  SMB 连接 10.0.1.50:445（横向移动）
T+315s  SMB Named Pipe \svcctl（PsExec 远程执行）
T+600s  DNS TXT 查询 tunnel.evil.com（DNS 隧道外泄）
T+900s  HTTP POST 大文件上传（数据外泄）

tshark -r capture.pcap -T fields \
  -e ip.src -e ip.dst -e ip.proto \
  -e tcp.srcport -e tcp.dstport \
  -e udp.srcport -e udp.dstport | \
  sort | uniq -c | sort -rn | head -20

cat *.log | jq -r 'select(.uid != null) | .uid' | sort -u | head -20

cat http.log | jq -r '.uid' | sort > /tmp/http_uids.txt
cat ssl.log | jq -r '.uid' | sort > /tmp/ssl_uids.txt
comm -12 /tmp/http_uids.txt /tmp/ssl_uids.txt

import statistics
from collections import defaultdict

class TrafficBaseline:
    def __init__(self):
        self.hourly_bytes = defaultdict(list)
        self.hourly_connections = defaultdict(list)
        self.protocol_distribution = defaultdict(int)

    def build_baseline(self, historical_data, window_days=30):
        for entry in historical_data:
            hour = entry['ts'] // 3600 % 24
            self.hourly_bytes[hour].append(entry.get('orig_bytes', 0))
            self.hourly_connections[hour].append(1)
            self.protocol_distribution[entry.get('service', 'unknown')] += 1

    def detect_anomaly(self, current_data, sigma=3):
        anomalies = []
        for hour, values in self.hourly_bytes.items():
            if not values:
                continue
            mean = statistics.mean(values)
            stdev = statistics.stdev(values) if len(values) > 1 else mean * 0.1
            current = sum(e.get('orig_bytes', 0) for e in current_data
                         if e['ts'] // 3600 % 24 == hour)
            if stdev > 0 and abs(current - mean) > sigma * stdev:
                anomalies.append({
                    'hour': hour,
                    'current_bytes': current,
                    'baseline_mean': mean,
                    'baseline_stdev': stdev,
                    'z_score': (current - mean) / stdev
                })
        return anomalies

baseline = TrafficBaseline()
print("[*] Traffic Baseline model initialized")

tshark -r capture.pcap -Y "dns" -T fields -e dns.qry.name | \
  sort -u > ioc_domains.txt

tshark -r capture.pcap -Y "http.request" -T fields \
  -e http.host -e http.request.uri | \
  sort -u > ioc_urls.txt

tshark -r capture.pcap -Y "tls.handshake.type == 11" -T fields \
  -e tls.cert.hash | sort -u > ioc_cert_hashes.txt

tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields \
  -e tls.handshake.ja3 | sort -u > ioc_ja3.txt

{
    "type": "indicator",
    "spec_version": "2.1",
    "id": "indicator--a932fcc6-e032-479c-8f9a-3c1b2c6f7eaa",
    "pattern_type": "stix",
    "pattern": "[domain-name:value = 'c2.evil.com' OR ipv4-addr:value = '198.51.100.1']",
    "valid_from": "2026-07-02T00:00:00Z",
    "labels": ["malicious-activity"],
    "confidence": 90
}

<ioc>
  <definition>
    <Indicator operator="OR">
      <IndicatorItem condition="is">
        <Context document="Network" search="Network/DNS" type="mir"/>
        <Content>c2.evil.com</Content>
      </IndicatorItem>
      <IndicatorItem condition="is">
        <Context document="Network" search="Network/IP" type="mir"/>
        <Content>198.51.100.1</Content>
      </IndicatorItem>
    </Indicator>
  </definition>
</ioc>

第一层：IOC 匹配 → 产生告警
第二层：上下文验证 → 排除已知合法用途
第三层：行为关联 → 与其他数据源交叉验证
第四层：人工审核 → 最终确认/排除

virustotal_url_check() {
    curl -s --header "x-apikey: $VT_API_KEY" \
      "https://www.virustotal.com/api/v3/domains/$1" | \
      jq '.data.attributes.last_analysis_stats'
}

for domain in $(cat ioc_domains.txt); do
    echo "Checking: $domain"
    virustotal_url_check "$domain"
done

title: DNS Tunnel via High Entropy Subdomain
id: e5f6a7b8-c9d0-1234-efab-567890abcdef
status: experimental
description: 检测 DNS 查询中包含高熵子域名的隧道行为
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: dns
detection:
    selection:
        qtype_name:
            - "TXT"
            - "NULL"
            - "CNAME"
        query|re: '[a-z0-9]{20,}\.'
    condition: selection
fields:
    - id.orig_h
    - id.resp_h
    - query
    - qtype_name
level: high
tags:
    - attack.command_and_control
    - attack.t1071.004

title: Suspicious JA3 Fingerprint Match
id: f6a7b8c9-d0e1-2345-fabc-678901bcdefa
status: experimental
description: 检测已知恶意工具的 JA3 指纹
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: ssl
detection:
    selection:
        ja3:
            - "4d7a2d269e3e33c5a5e8f4a7b2c1d0e9"
            - "72a589da586844d7f0818ce684948eea"
            - "a0e9f5d64349fb13191bc781f81f42e1"
    condition: selection
fields:
    - id.orig_h
    - id.resp_h
    - server_name
    - ja3
level: critical
tags:
    - attack.command_and_control
    - attack.t1573

title: ICMP Tunnel Detection
id: a7b8c9d0-e1f2-3456-abcd-789012cdefab
status: experimental
description: 检测 ICMP 隧道行为（大包/高频）
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: icmp
detection:
    selection_size:
        bytes|gt: 64
    selection_frequency:
        timeframe: 5m
    condition: selection_size | count() by id.orig_h > 50
fields:
    - id.orig_h
    - id.resp_h
    - bytes
level: high
tags:
    - attack.command_and_control
    - attack.t1071.004

title: LDAP Enumeration BloodHound Detection
id: b8c9d0e1-f2a3-4567-bcde-890123defabc
status: experimental
description: 检测 BloodHound/SharpHound 的 LDAP 枚举行为
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: ldap
detection:
    selection:
        search_filter|contains:
            - "samAccountType=805306368"
            - "samAccountType=805306369"
            - "servicePrincipalName="
            - "objectclass=*"
    timeframe: 5m
    condition: selection | count() by id.orig_h > 50
fields:
    - id.orig_h
    - id.resp_h
    - search_filter
level: high
tags:
    - attack.discovery
    - attack.t1087

title: SMB Lateral Movement via PsExec
id: c9d0e1f2-a3b4-5678-cdef-901234efabcd
status: experimental
description: 检测通过 SMB 进行的 PsExec 横向移动
author: Security Team
date: 2026/07/02
logsource:
    product: zeek
    service: smb_files
detection:
    selection:
        name|contains:
            - "svcctl"
            - "psexec"
        action: "SMB::FILE_OPEN"
    condition: selection
fields:
    - id.orig_h
    - id.resp_h
    - name
    - path
level: critical
tags:
    - attack.lateral_movement
    - attack.t1570

module C2Detection;

export {
    redef enum Notice::Type += {
        Suspicious_Beacon_Pattern,
        Known_Malicious_JA3,
        Unusual_DNS_Tunneling
    };
    const known_bad_ja3: set[string] = {
        "4d7a2d269e3e33c5a5e8f4a7b2c1d0e9",
        "72a589da586844d7f0818ce684948eea"
    } &redef;
}

event ssl_client_hello(c: connection, version: count, possible_ts: time,
    client_random: string, session_id: string, ciphers: vector of count)
    {
    local ja3 = ja3_calc(c);
    if (ja3 in known_bad_ja3)
        {
        NOTICE([$note=Known_Malicious_JA3,
                $conn=c,
                $msg=fmt("Known malicious JA3 detected: %s", ja3)]);
        }
    }

alert dns $HOME_NET any -> any any (
    msg:"DNS High Entropy Subdomain - Possible Tunnel";
    dns.query; pcre:"/[a-z0-9]{30,}\./i";
    threshold: type both, track by_src, count 50, seconds 300;
    sid:9000001; rev:1;
    metadata: attack_target Client_Endpoint, deployment Perimeter;
)

alert http $HOME_NET any -> $EXTERNAL_NET any (
    msg:"HTTP Possible Cobalt Strike Beacon URI";
    http.uri; content:"/submit.php";
    http.user_agent; content:"Mozilla/5.0";
    threshold: type both, track by_src, count 10, seconds 600;
    sid:9000002; rev:1;
    metadata: attack_target Client_Endpoint, deployment Perimeter;
)

alert tcp $HOME_NET any -> $HOME_NET 445 (
    msg:"SMB Admin Share Access";
    flow:to_server,established;
    content:"|00|"; depth:1;
    content:"ADMIN$"; nocase;
    sid:9000003; rev:1;
    metadata: attack_target Server_Endpoint, deployment Internal;
)

alert icmp any any -> any any (
    msg:"ICMP Large Payload - Possible Tunnel";
    dsize:>128;
    threshold: type both, track by_src, count 100, seconds 60;
    sid:9000004; rev:1;
    metadata: attack_target Client_Endpoint, deployment Perimeter;
)

alert tls $HOME_NET any -> $EXTERNAL_NET any (
    msg:"Self-Signed Certificate on Unusual Port";
    tls.cert_issuer; content:"CN=";
    tls.cert_subject; content:"CN=";
    flow:to_server,established;
    sid:9000005; rev:1;
    metadata: attack_target Client_Endpoint, deployment Perimeter;
)

rule CobaltStrike_Beacon_Metadata
{
    meta:
        description = "Detect Cobalt Strike beacon metadata pattern"
        author = "Security Team"
        date = "2026-07-02"
    strings:
        $metadata_pattern = {00 01 00 01 00 02}
        $ua_pattern = "Mozilla/5.0 (compatible; MSIE"
        $uri_submit = "/submit.php"
        $uri_pixel = "/pixel.gif"
    condition:
        any of them
}

rule DNS_Tunnel_Iodine_Pattern
{
    meta:
        description = "Detect iodine DNS tunnel traffic pattern"
        author = "Security Team"
        date = "2026-07-02"
    strings:
        $iodine_header = {69 6f 64 69 6e 65}
        $high_entropy = /[a-z0-9]{40,}/
        $null_query = {00 10 01}
    condition:
        any of them
}

rule Suspicious_UserAgent_Collection
{
    meta:
        description = "Detect known malicious user agents"
        author = "Security Team"
        date = "2026-07-02"
    strings:
        $ua1 = "Mozilla/4.0 (compatible; MSIE 6.0)"
        $ua2 = "Mozilla/4.0 (compatible; MSIE 7.0)"
        $ua3 = "User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0"
        $ua4 = "python-requests/2."
        $ua5 = "Go-http-client/1.1"
    condition:
        any of them
}

# 查找特定 JA3 指纹的会话
ja3 == 4d7a2d269e3e33c5a5e8f4a7b2c1d0e9

# 查找 DNS 隧道
protocols == dns && dns.opcode == 0 && dns.storm == 0 &&
  packets > 100 && dbytes > 50000

# 查找 SMB 横向移动
protocols == smb && port == 445 &&
  (ip.src == 10.0.0.50 || ip.dst == 10.0.0.50)

# 查找异常 TLS 证书
cert.alt == *evil* || cert.issuer.cn == *"Let's Encrypt"* &&
  cert.valid.from > 2026-06-01

import json
import subprocess
import hashlib
from collections import defaultdict
from datetime import datetime

class AutomatedTrafficHunter:
    def __init__(self, pcap_path):
        self.pcap_path = pcap_path
        self.findings = []
        self.ioc_list = {
            'domains': set(),
            'ips': set(),
            'ja3': set(),
            'cert_hashes': set(),
            'urls': set()
        }

    def run_tshark(self, display_filter, fields):
        cmd = ['tshark', '-r', self.pcap_path, '-Y', display_filter,
               '-T', 'fields']
        for f in fields:
            cmd.extend(['-e', f])
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout.strip().split('\n')

    def hunt_dns_tunneling(self):
        lines = self.run_tshark('dns',
            ['frame.time', 'ip.src', 'dns.qry.name', 'dns.qry.type'])
        for line in lines:
            parts = line.split('\t')
            if len(parts) < 4:
                continue
            query = parts[2] if len(parts) > 2 else ''
            subdomain_parts = query.split('.')
            if len(subdomain_parts) > 3:
                subdomain = '.'.join(subdomain_parts[:-2])
                if len(subdomain) > 30:
                    self.findings.append({
                        'type': 'DNS_TUNNEL_SUSPECTED',
                        'severity': 'HIGH',
                        'query': query,
                        'src': parts[1]
                    })
                    self.ioc_list['domains'].add(query)

    def hunt_c2_beacons(self):
        lines = self.run_tshark('http.request',
            ['frame.time', 'ip.src', 'ip.dst', 'http.host', 'http.request.uri'])
        host_requests = defaultdict(list)
        for line in lines:
            parts = line.split('\t')
            if len(parts) < 5:
                continue
            host = parts[3]
            host_requests[host].append({
                'ts': parts[0],
                'src': parts[1],
                'uri': parts[4]
            })

        for host, reqs in host_requests.items():
            if len(reqs) > 20:
                self.findings.append({
                    'type': 'C2_BEACON_CANDIDATE',
                    'severity': 'HIGH',
                    'host': host,
                    'request_count': len(reqs)
                })
                self.ioc_list['domains'].add(host)

    def hunt_suspicious_tls(self):
        lines = self.run_tshark('tls.handshake.type == 1',
            ['ip.src', 'ip.dst', 'tls.handshake.ja3',
             'tls.handshake.extensions_server_name'])
        known_bad = {
            '4d7a2d269e3e33c5a5e8f4a7b2c1d0e9': 'Cobalt Strike',
            '72a589da586844d7f0818ce684948eea': 'Metasploit'
        }
        for line in lines:
            parts = line.split('\t')
            if len(parts) < 3:
                continue
            ja3 = parts[2] if len(parts) > 2 else ''
            if ja3 in known_bad:
                self.findings.append({
                    'type': 'MALICIOUS_JA3',
                    'severity': 'CRITICAL',
                    'ja3': ja3,
                    'tool': known_bad[ja3],
                    'src': parts[0],
                    'dst': parts[1]
                })
                self.ioc_list['ja3'].add(ja3)

    def hunt_smb_lateral_movement(self):
        lines = self.run_tshark('smb2',
            ['frame.time', 'ip.src', 'ip.dst', 'smb2.filename'])
        for line in lines:
            parts = line.split('\t')
            if len(parts) < 4:
                continue
            filename = parts[3] if len(parts) > 3 else ''
            if any(p in filename.lower() for p in ['svcctl', 'atsvc', 'msagent']):
                self.findings.append({
                    'type': 'SMB_LATERAL_MOVEMENT',
                    'severity': 'CRITICAL',
                    'src': parts[1],
                    'dst': parts[2],
                    'pipe': filename
                })

    def generate_report(self):
        report = {
            'timestamp': datetime.now().isoformat(),
            'pcap': self.pcap_path,
            'total_findings': len(self.findings),
            'findings_by_severity': defaultdict(int),
            'findings': self.findings,
            'iocs': {k: list(v) for k, v in self.ioc_list.items()}
        }
        for f in self.findings:
            report['findings_by_severity'][f['severity']] += 1
        return report

    def execute_full_hunt(self):
        self.hunt_dns_tunneling()
        self.hunt_c2_beacons()
        self.hunt_suspicious_tls()
        self.hunt_smb_lateral_movement()
        return self.generate_report()

hunter = AutomatedTrafficHunter("capture.pcap")
print("[*] Automated Traffic Hunter initialized")

{
    "pipeline": {
        "description": "Network traffic enrichment pipeline",
        "processors": [
            {
                "grok": {
                    "field": "message",
                    "patterns": ["%{TIMESTAMP_ISO8601:timestamp} %{IP:src_ip} %{IP:dst_ip} %{NUMBER:src_port} %{NUMBER:dst_port} %{WORD:protocol}"]
                }
            },
            {
                "geoip": {
                    "field": "dst_ip",
                    "ignore_missing": true
                }
            }
        ]
    }
}

触发条件：Sigma 规则匹配
  → Step 1: 提取 IOC（IP/域名/哈希）
  → Step 2: VirusTotal/ThreatFox 查询
  → Step 3: 关联 SIEM 历史告警
  → Step 4: 查询资产管理系统（受影响主机）
  → Step 5: 自动生成工单
  → Step 6: 通知 SOC 分析师
  → Step 7: 如确认恶意 → 自动隔离（EDR/NAC）

2020-02  → 攻击者入侵 SolarWinds 构建系统
2020-03  → 植入后门代码到 Orion 平台
2020-03  → 受感染更新分发给 ~18,000 客户
2020-04  → 攻击者选择性激活约 9 个高价值目标的后门
2020-12  → FireEye/Mandiant 公开披露

{encoded_data}.appsync-api.eu-west-1.avsvmcloud.com  → C2 命令请求
{encoded_data}.appsync-api.us-east-2.avsvmcloud.com  → 数据回传
{step_number}.appsync-api.eu-west-1.avsvmcloud.com   → 阶段控制

tshark -r capture.pcap -Y "dns.qry.name contains avsvmcloud.com" \
  -T fields -e frame.time -e ip.src -e dns.qry.name

tshark -r capture.pcap -Y "dns.qry.name contains appsync-api" \
  -T fields -e dns.qry.name -e dns.qry.type | \
  sort | uniq -c | sort -rn

域名: avsvmcloud.com
SHA256: 32519b85c0b422e4656de6e6c41878e95fd95026267daab4215ee59c107d6c77 (SUNBURST)
IP: 13.59.205.66, 54.193.111.190
URI 模式: /appsync-api/{region}/avsvmcloud.com

阶段 1 - 初始访问：
  受害者 → 邮件服务器 (SMTP/IMAP)
  受害者 → 恶意 URL (HTTPS)

阶段 2 - C2 建立：
  受害者 → C2 服务器 (HTTPS, JA3 匹配)
  Beacon 间隔: 5-60 分钟

阶段 3 - 横向移动：
  受害者 → 内部主机 (SMB 445)
  Named Pipe: \svcctl, \msagent_*

阶段 4 - 数据外泄：
  受害者 → 云存储 (HTTPS, 大量上传)
  或 → 攻击者基础设施 (DNS 隧道/HTTPS)

JA3 指纹: 72a589da586844d7f0818ce684948eea (Metasploit)
域名模式: *.ddns.net, *.no-ip.com (DDNS 滥用)
证书特征: 自签名证书, CN=*.microsoft.com (仿冒)
SMB 特征: \svcctl Named Pipe + 批量连接