lou
128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""抓包验收分析: 软路由 WAN 口 tcpdump 抓的 pcap, 验证伪装/套壳效果。
|
|
|
|
用法: .venv/bin/python tools/analyze_capture.py <capture.pcap>
|
|
|
|
输出:
|
|
1. 总览(包数/字节/协议分布)
|
|
2. 出站 SYN 特征统计(TTL/window/wscale/TS/options) - L1 伪装器生效证据
|
|
3. 云通道(203.0.113.10:8444)流量统计 - 套壳证据
|
|
4. 明文 HTTP 出站(应 0, 全被套壳)与 DNS 出站(应 0, 走通道)检查
|
|
5. 目标端口/目标 IP Top
|
|
|
|
"出站"判定: 源 IP 在私有网段(192.168.0.0/16 等)视为内网侧。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
from scapy.layers.inet import IP, TCP, UDP
|
|
from scapy.utils import rdpcap
|
|
|
|
CLOUD_IP = "203.0.113.10"
|
|
CLOUD_PORT = 8444
|
|
LAN_NETS = ("192.168.", "172.16.", "10.")
|
|
|
|
|
|
def _is_lan(ip: str) -> bool:
|
|
return any(ip.startswith(p) for p in LAN_NETS)
|
|
|
|
|
|
def _tcp_options(tcp: TCP) -> dict[str, object]:
|
|
"""scapy option kind 是字符串名(MSS/SAckOK/Timestamp/NOP/WScale/EOL)。"""
|
|
opts: dict[str, object] = {}
|
|
if not tcp.options:
|
|
return opts
|
|
for entry in tcp.options:
|
|
if isinstance(entry, tuple) and len(entry) >= 1 and isinstance(entry[0], str):
|
|
opts[entry[0]] = entry[1] if len(entry) > 1 else None
|
|
return opts
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2:
|
|
print("usage: python tools/analyze_capture.py <capture.pcap>", file=sys.stderr)
|
|
return 2
|
|
path = Path(sys.argv[1])
|
|
if not path.exists():
|
|
print(f"文件不存在: {path}", file=sys.stderr)
|
|
return 2
|
|
pkts = rdpcap(str(path))
|
|
total = len(pkts)
|
|
bytes_total = sum(len(bytes(p)) for p in pkts)
|
|
print(f"== 总览: {total} 包 / {bytes_total/1024:.1f} KB")
|
|
|
|
proto: Counter[int] = Counter()
|
|
out_syn: list[tuple[int, int, int, bool, list[str], int, str]] = []
|
|
cloud_bytes = 0
|
|
cloud_pkts = 0
|
|
plain_http_out = 0
|
|
dns_out = 0
|
|
dst_ports: Counter[int] = Counter()
|
|
dst_ips: Counter[str] = Counter()
|
|
for p in pkts:
|
|
if IP not in p:
|
|
continue
|
|
ip = p[IP]
|
|
proto[ip.proto] += 1
|
|
src, dst = ip.src, ip.dst
|
|
outbound = _is_lan(src) and not _is_lan(dst)
|
|
if dst == CLOUD_IP or src == CLOUD_IP:
|
|
cloud_pkts += 1
|
|
cloud_bytes += len(p)
|
|
if TCP in p:
|
|
tcp = p[TCP]
|
|
dst_ports[tcp.dport] += 1
|
|
dst_ips[dst] += 1
|
|
if outbound and tcp.flags & 0x02 and not tcp.flags & 0x10:
|
|
opts = _tcp_options(tcp)
|
|
ws = opts.get("WScale")
|
|
wscale_v = ws if isinstance(ws, int) else -1
|
|
kinds = list(opts.keys())
|
|
out_syn.append(
|
|
(ip.ttl, tcp.window, wscale_v, "Timestamp" in opts, kinds, len(p), dst)
|
|
)
|
|
if outbound and tcp.dport == 80:
|
|
plain_http_out += 1
|
|
elif UDP in p:
|
|
udp = p[UDP]
|
|
dst_ports[udp.dport] += 1
|
|
if outbound and udp.dport == 53:
|
|
dns_out += 1
|
|
|
|
print("\n== 出站 SYN 特征(L1 伪装器生效证据)")
|
|
if out_syn:
|
|
ttl_c = Counter(s[0] for s in out_syn)
|
|
win_c = Counter(s[1] for s in out_syn)
|
|
ws_c = Counter(s[2] for s in out_syn)
|
|
ts_c = Counter(s[3] for s in out_syn)
|
|
print(f"SYN 总数: {len(out_syn)}")
|
|
print(f"TTL 分布: {dict(ttl_c)} (期望 128 统一)")
|
|
print(f"window 分布: {dict(win_c)} (期望 64240 统一)")
|
|
print(f"wscale 分布: {dict(ws_c)} (期望 8 统一)")
|
|
print(f"带 TS option: {dict(ts_c)} (期望全部 False - 已剥)")
|
|
seen_orders: Counter[tuple[str, ...]] = Counter(tuple(s[4]) for s in out_syn)
|
|
print(f"options 顺序(前 3 种): {seen_orders.most_common(3)}")
|
|
else:
|
|
print("无出站 SYN(抓包期间无新连接?)")
|
|
|
|
print("\n== 云通道(套壳证据)")
|
|
print(f"到/从 {CLOUD_IP}:{CLOUD_PORT} 的包: {cloud_pkts} 个 / {cloud_bytes/1024:.1f} KB")
|
|
|
|
print("\n== 明文泄漏检查(期望全 0)")
|
|
print(f"明文 HTTP 出站(dport 80): {plain_http_out}")
|
|
print(f"公网 DNS 出站(dport 53): {dns_out}")
|
|
|
|
print("\n== 目标端口 Top 8")
|
|
for port, n in dst_ports.most_common(8):
|
|
print(f" {port}: {n}")
|
|
print("== 目标 IP Top 8")
|
|
for ip, n in dst_ips.most_common(8):
|
|
print(f" {ip}: {n}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|