隐私增强技术(PET)与匿名网络取证深度分析

from stem import Signal
from stem.control import Controller
from stem import CircStatus
import time

def get_tor_circuits():
    with Controller.from_port(port=9051) as controller:
        controller.authenticate(password="your_password")
        for circ in controller.get_circuits():
            if circ.status == CircStatus.BUILT:
                path = " -> ".join(
                    [f"{hop[0]}({hop[1]})" for hop in circ.path]
                )
                print(f"电路ID: {circ.circuit_id} | 路径: {path}")
                print(f"  目标: {circ.purpose}")

def renew_tor_circuit():
    with Controller.from_port(port=9051) as controller:
        controller.authenticate(password="your_password")
        controller.signal(Signal.NEWNYM)
        print("新电路已建立")

def get_tor_network_info():
    with Controller.from_port(port=9051) as controller:
        controller.authenticate(password="your_password")
        info = controller.get_info("network-liveness")
        print(f"网络状态: {info}")
        info = controller.get_info("orconn-status")
        print(f"连接状态: {info}")

if __name__ == "__main__":
    get_tor_circuits()

tshark -r tor_traffic.pcap -Y "tls.handshake.type == 1" \
  -T fields -e tls.handshake.ciphersuite \
  -e tls.handshake.extensions_server_name \
  -e ip.dst | head -50

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

tshark -r tor_traffic.pcap \
  -Y "tls.handshake.extensions_server_name contains \"\"" \
  -T fields -e ip.src -e ip.dst -e tls.handshake.extensions_server_name

import requests
import json
import hashlib

def fetch_guard_nodes():
    url = "https://collector.torproject.org/metrics-consensus/"
    consensus_url = "https://metrics.torproject.org/consensus.html"

    relays_url = "https://onionoo.torproject.org/details"
    response = requests.get(relays_url, params={
        "type": "relay",
        "flag": "Guard",
        "running": "true"
    })
    data = response.json()

    guard_nodes = []
    for relay in data.get("relays", []):
        guard_info = {
            "fingerprint": relay.get("fingerprint"),
            "nickname": relay.get("nickname"),
            "address": relay.get("or_addresses", [""])[0],
            "bandwidth": relay.get("bandwidth_rate", 0),
            "flags": relay.get("flags", []),
            "country": relay.get("country", "unknown"),
            "last_seen": relay.get("last_seen"),
        }
        guard_nodes.append(guard_info)

    return guard_nodes

def analyze_guard_distribution(guards):
    country_count = {}
    bandwidth_total = 0
    for g in guards:
        country = g["country"]
        country_count[country] = country_count.get(country, 0) + 1
        bandwidth_total += g["bandwidth"]

    sorted_countries = sorted(
        country_count.items(), key=lambda x: x[1], reverse=True
    )
    print(f"Guard 节点总数: {len(guards)}")
    print(f"总带宽: {bandwidth_total / 1024 / 1024:.2f} MB/s")
    print("Top 10 国家分布:")
    for country, count in sorted_countries[:10]:
        pct = count / len(guards) * 100
        print(f"  {country}: {count} ({pct:.1f}%)")

#!/bin/bash

INTERFACE="eth0"
CAPTURE_DURATION=60
OUTPUT_FILE="tor_detection_$(date +%Y%m%d_%H%M%S).pcap"

echo "[*] 开始捕获流量，持续 ${CAPTURE_DURATION} 秒..."
tcpdump -i "$INTERFACE" -w "$OUTPUT_FILE" -G "$CAPTURE_DURATION" \
  -W 1 "tcp port 9050 or tcp port 9150"

echo "[*] 分析流量特征..."
echo "=== Tor 入口连接（9050/9150端口）==="
tshark -r "$OUTPUT_FILE" -Y "tcp.port == 9050 or tcp.port == 9150" \
  -T fields -e ip.src -e ip.dst -e tcp.srcport -e tcp.dstport | \
  awk '{print $1}' | sort | uniq -c | sort -rn | head -20

echo ""
echo "=== 可疑 TLS 连接统计 ==="
tshark -r "$OUTPUT_FILE" -Y "tls.handshake.type == 1" \
  -T fields -e ip.dst -e tls.handshake.extensions_server_name | \
  awk '{count[$1]++} END {for (ip in count) print count[ip], ip}' | \
  sort -rn | head -20

echo ""
echo "=== 固定大小包统计（Tor cell 特征）==="
tshark -r "$OUTPUT_FILE" -T fields -e frame.len -e ip.dst | \
  awk '{if ($1 == 514 || $1 == 590 || $1 == 1470) count[$2]++}
  END {for (ip in count) print count[ip], ip}' | sort -rn | head -10

echo "HELLO VERSION 3.0" | nc localhost 7656
echo "SESSION CREATE STYLE=STREAM ID=myapp SIGNING_TYPE=EdDSA" | nc localhost 7656
echo "SESSION STATUS ID=myapp" | nc localhost 7656
echo "STREAM CONNECT ID=myapp DESTINATION=<base64_dest>" | nc localhost 7656

import requests
import re
import time

class I2PServiceScanner:
    def __init__(self, i2p_proxy="http://127.0.0.1:4444"):
        self.proxies = {
            "http": i2p_proxy,
            "https": i2p_proxy
        }

    def resolve_i2p_address(self, address):
        lookup_url = f"http://127.0.0.1:4444/{address}"
        try:
            response = requests.get(lookup_url, proxies=self.proxies,
                                    timeout=30)
            return response.status_code, response.headers
        except requests.exceptions.RequestException as e:
            return None, str(e)

    def scan_i2p_service(self, address):
        result = {
            "address": address,
            "reachable": False,
            "title": "",
            "headers": {},
            "content_length": 0,
            "server_info": ""
        }
        try:
            url = f"http://{address}/"
            response = requests.get(url, proxies=self.proxies, timeout=30)
            result["reachable"] = True
            result["status_code"] = response.status_code
            result["headers"] = dict(response.headers)
            result["content_length"] = len(response.content)

            title_match = re.search(
                r"<title>(.*?)</title>", response.text, re.IGNORECASE
            )
            if title_match:
                result["title"] = title_match.group(1)

            server = response.headers.get("Server", "")
            result["server_info"] = server

        except Exception as e:
            result["error"] = str(e)

        return result

    def batch_scan(self, address_list, delay=5):
        results = []
        for addr in address_list:
            print(f"[*] 扫描: {addr}")
            result = self.scan_i2p_service(addr)
            results.append(result)
            status = "可达" if result["reachable"] else "不可达"
            print(f"    状态: {status}")
            if result["reachable"]:
                print(f"    标题: {result['title']}")
                print(f"    服务器: {result['server_info']}")
            time.sleep(delay)
        return results

tshark -r i2p_traffic.pcap -Y "udp.port == 9150" \
  -T fields -e ip.src -e ip.dst -e udp.srcport -e udp.dstport | \
  head -30

tshark -r i2p_traffic.pcap -Y "tcp.port == 4444 or tcp.port == 4445" \
  -T fields -e ip.src -e ip.dst -e tcp.dstport | \
  awk '{count[$3]++} END {for (p in count) print count[p], p}' | sort -rn

import numpy as np
from scipy.stats import pearsonr

def time_correlation_attack(sender_timestamps, receiver_timestamps,
                            time_window=1.0):
    correlation_scores = []
    for t_s in sender_timestamps:
        matched = receiver_timestamps[
            (receiver_timestamps >= t_s - time_window) &
            (receiver_timestamps <= t_s + time_window)
        ]
        if len(matched) > 0:
            min_delay = np.min(np.abs(matched - t_s))
            correlation_scores.append((t_s, min_delay))

    if not correlation_scores:
        return 0.0, []

    delays = [s[1] for s in correlation_scores]
    avg_delay = np.mean(delays)
    match_ratio = len(correlation_scores) / len(sender_timestamps)

    confidence = match_ratio * (1.0 - min(avg_delay / time_window, 1.0))
    return confidence, correlation_scores

def packet_size_correlation(sender_sizes, receiver_sizes):
    min_len = min(len(sender_sizes), len(receiver_sizes))
    if min_len == 0:
        return 0.0

    s_normalized = np.array(sender_sizes[:min_len], dtype=float)
    r_normalized = np.array(receiver_sizes[:min_len], dtype=float)

    s_normalized = (s_normalized - np.mean(s_normalized)) / (
        np.std(s_normalized) + 1e-10
    )
    r_normalized = (r_normalized - np.mean(r_normalized)) / (
        np.std(r_normalized) + 1e-10
    )

    correlation, p_value = pearsonr(s_normalized, r_normalized)
    return correlation

客户端加密流程（3层加密）：
原始数据 → [Exit密钥加密] → [Middle密钥加密] → [Guard密钥加密]

网络传输过程：
Exit节点解密Exit层 → 暴露明文数据
Middle节点解密Middle层 → 暴露Exit地址
Guard节点解密Guard层 → 暴露Middle地址

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import joblib

class TorTrafficClassifier:
    FEATURE_NAMES = [
        "total_bytes_fwd", "total_bytes_bwd",
        "fwd_packet_count", "bwd_packet_count",
        "mean_fwd_packet_size", "mean_bwd_packet_size",
        "std_fwd_packet_size", "std_bwd_packet_size",
        "flow_duration", "fwd_inter_arrival_mean",
        "bwd_inter_arrival_mean", "fwd_inter_arrival_std",
        "bwd_inter_arrival_std", "syn_count", "ack_count",
        "psh_count", "fin_count", "rst_count"
    ]

    def __init__(self):
        self.model = RandomForestClassifier(
            n_estimators=200, max_depth=30,
            min_samples_split=5, random_state=42, n_jobs=-1
        )

    def extract_features(self, packets):
        features = {}
        fwd_sizes = [p["size"] for p in packets if p["direction"] == "fwd"]
        bwd_sizes = [p["size"] for p in packets if p["direction"] == "bwd"]
        fwd_times = [p["timestamp"] for p in packets if p["direction"] == "fwd"]
        bwd_times = [p["timestamp"] for p in packets if p["direction"] == "bwd"]

        features["total_bytes_fwd"] = sum(fwd_sizes)
        features["total_bytes_bwd"] = sum(bwd_sizes)
        features["fwd_packet_count"] = len(fwd_sizes)
        features["bwd_packet_count"] = len(bwd_sizes)
        features["mean_fwd_packet_size"] = np.mean(fwd_sizes) if fwd_sizes else 0
        features["mean_bwd_packet_size"] = np.mean(bwd_sizes) if bwd_sizes else 0
        features["std_fwd_packet_size"] = np.std(fwd_sizes) if fwd_sizes else 0
        features["std_bwd_packet_size"] = np.std(bwd_sizes) if bwd_sizes else 0

        if len(packets) > 1:
            timestamps = [p["timestamp"] for p in packets]
            features["flow_duration"] = max(timestamps) - min(timestamps)
        else:
            features["flow_duration"] = 0

        fwd_intervals = np.diff(fwd_times) if len(fwd_times) > 1 else [0]
        bwd_intervals = np.diff(bwd_times) if len(bwd_times) > 1 else [0]
        features["fwd_inter_arrival_mean"] = np.mean(fwd_intervals)
        features["bwd_inter_arrival_mean"] = np.mean(bwd_intervals)
        features["fwd_inter_arrival_std"] = np.std(fwd_intervals)
        features["bwd_inter_arrival_std"] = np.std(bwd_intervals)

        flags = [p.get("flags", "") for p in packets]
        features["syn_count"] = sum(1 for f in flags if "S" in f)
        features["ack_count"] = sum(1 for f in flags if "A" in f)
        features["psh_count"] = sum(1 for f in flags if "P" in f)
        features["fin_count"] = sum(1 for f in flags if "F" in f)
        features["rst_count"] = sum(1 for f in flags if "R" in f)

        return [features[name] for name in self.FEATURE_NAMES]

    def train(self, X, y):
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        self.model.fit(X_train, y_train)
        y_pred = self.model.predict(X_test)
        print(classification_report(y_test, y_pred,
              target_names=["Normal", "Tor"]))

    def predict(self, features):
        return self.model.predict([features])[0]

    def save_model(self, path):
        joblib.dump(self.model, path)

    def load_model(self, path):
        self.model = joblib.load(path)

tshark -r vpn_traffic.pcap \
  -Y "udp.port == 51820" \
  -T fields -e ip.src -e ip.dst -e udp.srcport -e udp.dstport \
  -e frame.len -e _ws.col.Info | head -30

tshark -r vpn_traffic.pcap \
  -Y "udp.port == 51820" \
  -T fields -e frame.len | sort | uniq -c | sort -rn | head -10

#!/bin/bash

echo "=== VPN 泄漏检测脚本 ==="

echo "[1] 检测 DNS 泄漏..."
PUBLIC_DNS=$(dig +short myip.opendns.com @resolver1.opendns.com)
echo "  公网 DNS 解析结果: $PUBLIC_DNS"
LOCAL_DNS=$(dig +short example.com | head -1)
echo "  本地 DNS 解析结果: $LOCAL_DNS"
if [ "$PUBLIC_DNS" != "$LOCAL_DNS" ]; then
    echo "  [!] 可能存在 DNS 泄漏"
else
    echo "  [+] DNS 泄漏检测通过"
fi

echo ""
echo "[2] 检测 IPv6 泄漏..."
IPV6_ADDR=$(curl -6 -s https://ifconfig.co 2>/dev/null)
if [ -n "$IPV6_ADDR" ]; then
    echo "  [!] IPv6 地址暴露: $IPV6_ADDR"
else
    echo "  [+] IPv6 泄漏检测通过"
fi

echo ""
echo "[3] 检测 WebRTC 泄漏..."
echo "  请在浏览器中访问: https://browserleaks.com/webrtc"
echo "  检查 WebRTC 是否暴露真实 IP"

echo ""
echo "[4] 检查 VPN 路由配置..."
echo "  当前路由表（VPN相关）:"
route -n get default | head -5
netstat -rn | grep -E "utun|tun|ppp" | head -10

echo ""
echo "[5] 检查 VPN 进程状态..."
ps aux | grep -E "vpn|openvpn|wireguard|clash|v2ray" | grep -v grep

import requests

def analyze_transaction_heuristics(tx_data):
    findings = []
    inputs = tx_data.get("vin", [])
    outputs = tx_data.get("vout", [])
    input_addresses = [inp["addr"] for inp in inputs if "addr" in inp]

    if len(input_addresses) > 1:
        unique_addresses = set(input_addresses)
        if len(unique_addresses) > 1:
            findings.append({
                "heuristic": "共同输入启发式",
                "confidence": "高",
                "addresses": list(unique_addresses),
                "detail": f"{len(unique_addresses)}个地址在同一交易中作为输入"
            })

    if len(outputs) >= 2:
        output_addresses = [out.get("scriptPubKey", {}).get("addresses", [""])
                          for out in outputs]
        change_candidate = None
        for i, addr_list in enumerate(output_addresses):
            if addr_list and addr_list[0] in input_addresses:
                change_candidate = addr_list[0]
                break

        if change_candidate:
            findings.append({
                "heuristic": "找零地址启发式",
                "confidence": "中",
                "change_address": change_candidate,
                "detail": "疑似找零地址与输入地址匹配"
            })

    return findings

def trace_funding_flow(start_address, max_depth=5):
    api_url = f"https://blockchain.info/rawaddr/{start_address}"
    response = requests.get(api_url)
    data = response.json()

    flow_map = {
        "address": start_address,
        "total_received": data.get("total_received", 0),
        "n_tx": data.get("n_tx", 0),
        "transactions": []
    }

    for tx in data.get("txs", [])[:20]:
        tx_info = {
            "hash": tx.get("hash"),
            "time": tx.get("time"),
            "result": tx.get("result", 0),
            "fee": tx.get("fee", 0),
            "inputs": len(tx.get("inputs", [])),
            "outputs": len(tx.get("out", []))
        }
        flow_map["transactions"].append(tx_info)

    return flow_map

# 安装 Nyx
pip install nyx

# 启动 Nyx（需要 Tor ControlPort 启用）
nyx

# 在 nyx 界面中：
# Shift+S - 查看电路状态
# Shift+R - 查看中继信息
# Shift+B - 查看带宽使用
# Shift+L - 查看事件日志
# Shift+C - 创建新电路

# 安装 OnionScan
go get github.com/s-rah/onionscan

# 扫描单个隐藏服务
onionscan -verbose -torProxyAddress="127.0.0.1:9050" \
  example.onion

# 批量扫描
cat hidden_services.txt | while read onion; do
    onionscan -torProxyAddress="127.0.0.1:9050" "$onion"
    sleep 5
done

# 使用 Nmap 进行端口扫描（通过Tor）
nmap -sT -Pn -n -p 80,443,8080,8443 \
  --proxies socks4://127.0.0.1:9050 \
  --script ssl-enum-ciphers \
  target.onion

# 安装 Shadow 模拟器
git clone https://github.com/shadow/shadow.git
cd shadow && mkdir build && cd build
cmake .. -DCMAKE_INSTALL_PREFIX=/opt/shadow
make -j$(nproc) && sudo make install

# 运行 Tor 模拟场景
shadow -w /tmp/shadow-data -l /tmp/shadow-data/shadow.log \
  tor-example.yaml

tshark -r tor.pcap -Y "tor" -V | head -200

tshark -r tor.pcap -Y "tor.control" \
  -T fields -e tor.control.command -e tor.control.arg | head -50

tshark -r tor.pcap \
  -Y "tls.handshake.extensions_server_name" \
  -T fields -e frame.time_relative -e ip.src -e ip.dst \
  -e tls.handshake.extensions_server_name -e frame.len | \
  awk -F'\t' '{print $1, $2, $3, $4, $5}' | head -50

tshark -r tor.pcap -q -z conv,tcp | sort -k7 -rn | head -20

from scapy.all import *
from collections import defaultdict, Counter
import datetime

class TorTrafficAnalyzer:
    TOR_CELL_SIZE = 514
    TOR_PORTS = {9050, 9051, 9150, 9151, 4444, 4445}

    def __init__(self, pcap_file):
        self.pcap_file = pcap_file
        self.flows = defaultdict(list)
        self.cell_count = 0
        self.circuit_info = {}

    def analyze(self):
        packets = rdpcap(self.pcap_file)
        stats = {
            "total_packets": len(packets),
            "tor_related": 0,
            "tls_handshakes": 0,
            "cell_sized_packets": 0,
            "unique_sources": set(),
            "unique_destinations": set(),
            "port_distribution": Counter(),
            "size_distribution": Counter(),
            "time_series": []
        }

        for pkt in packets:
            if not pkt.haslayer(IP):
                continue
            ip_src = pkt[IP].src
            ip_dst = pkt[IP].dst

            stats["unique_sources"].add(ip_src)
            stats["unique_destinations"].add(ip_dst)

            if pkt.haslayer(TCP):
                tcp = pkt[TCP]
                stats["port_distribution"][tcp.dport] += 1
                stats["port_distribution"][tcp.sport] += 1

                if tcp.dport in self.TOR_PORTS or \
                   tcp.sport in self.TOR_PORTS:
                    stats["tor_related"] += 1

                pkt_len = len(pkt)
                stats["size_distribution"][pkt_len] += 1

                if abs(pkt_len - self.TOR_CELL_SIZE) < 20:
                    stats["cell_sized_packets"] += 1

            if pkt.haslayer(Raw):
                payload = bytes(pkt[Raw].load)
                if len(payload) > 2 and payload[0:3] == b'\x16\x03\x01' or \
                   payload[0:3] == b'\x16\x03\x03':
                    stats["tls_handshakes"] += 1

        stats["unique_sources"] = len(stats["unique_sources"])
        stats["unique_destinations"] = len(stats["unique_destinations"])

        return stats

    def generate_report(self, stats):
        report = []
        report.append("=" * 60)
        report.append("Tor 流量分析报告")
        report.append("=" * 60)
        report.append(f"数据包总数: {stats['total_packets']}")
        report.append(f"Tor 相关流量: {stats['tor_related']}")
        report.append(f"TLS 握手数: {stats['tls_handshakes']}")
        report.append(f"Tor cell 大小包: {stats['cell_sized_packets']}")
        report.append(f"源IP数量: {stats['unique_sources']}")
        report.append(f"目标IP数量: {stats['unique_destinations']}")

        report.append("\n端口分布 Top 10:")
        for port, count in stats["port_distribution"].most_common(10):
            report.append(f"  端口 {port}: {count} 个包")

        tor_probability = (
            stats["tor_related"] / max(stats["total_packets"], 1) * 100
        )
        cell_probability = (
            stats["cell_sized_packets"] / max(stats["total_packets"], 1) * 100
        )
        report.append(f"\nTor 流量概率评估: {tor_probability:.1f}%")
        report.append(f"Tor cell 匹配率: {cell_probability:.1f}%")

        return "\n".join(report)

title: Tor Browser 或 Tor 进程启动检测
id: 7a1b2c3d-4e5f-6789-abcd-ef0123456789
status: stable
description: 检测 Tor Browser 或 tor 进程的启动行为
author: Security Analyst
date: 2026/08/01
tags:
  - attack.defense_evasion
  - attack.command_and_control
  - T1090.003
logsource:
  category: process_creation
  product: windows
detection:
  selection_tor_exe:
    - Image|endswith:
      - '\tor.exe'
      - '\torbrowser.exe'
      - '\firefox.exe'
    - ParentImage|endswith:
      - '\tor.exe'
      - '\torbrowser.exe'
  selection_tor_browser:
    Image|endswith:
      - '\Tor Browser\Browser\TorBrowser\Tor\tor.exe'
    CommandLine|contains:
      - 'tor'
  filter_legitimate:
    Image|contains:
      - '\tor-project\'
  condition: selection_tor_exe or selection_tor_browser
  condition: not filter_legitimate
level: high

---
title: Tor 网络流量异常检测
id: 8b2c3d4e-5f6a-7890-bcde-f01234567890
status: stable
description: 检测异常的 Tor 网络流量模式
author: Security Analyst
date: 2026/08/01
tags:
  - attack.command_and_control
  - T1090.003
logsource:
  product: zeek
  service: conn
detection:
  selection_tor_ports:
    id.orig_p|contains:
      - '9050'
      - '9051'
      - '9150'
      - '9151'
    id.resp_p|contains:
      - '80'
      - '443'
  selection_high_volume:
    orig_bytes|gt: 1000000
    duration|gt: 300
  condition: selection_tor_ports and selection_high_volume
level: medium

#!/bin/bash

TOR_CONSENSUS_DIR="/var/lib/tor/consensus"
LOG_DIR="/var/log/tor_analysis"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)

mkdir -p "$LOG_DIR"

echo "[*] Tor 节点扫描与流量统计"
echo "[*] 时间: $TIMESTAMP"

echo "[1] 获取 Tor 网络共识文档..."
curl -s "https://collector.torproject.org/metrics-consensus/consensus-$(date +%Y-%m-%d)" \
  -o "$LOG_DIR/consensus_$TIMESTAMP.txt"

CONSENSUS="$LOG_DIR/consensus_$TIMESTAMP.txt"
if [ -f "$CONSENSUS" ]; then
    TOTAL_RELAYS=$(grep -c "^r " "$CONSENSUS")
    GUARD_COUNT=$(grep -c "Guard" "$CONSENSUS" 2>/dev/null || echo 0)
    EXIT_COUNT=$(grep -c "Exit" "$CONSENSUS" 2>/dev/null || echo 0)
    STABLE_COUNT=$(grep -c "Stable" "$CONSENSUS" 2>/dev/null || echo 0)

    echo "  总中继数: $TOTAL_RELAYS"
    echo "  Guard 节点: $GUARD_COUNT"
    echo "  Exit 节点: $EXIT_COUNT"
    echo "  Stable 节点: $STABLE_COUNT"
fi

echo ""
echo "[2] 检测本地 Tor 进程..."
TOR_PROCS=$(ps aux | grep -i "[t]or" | grep -v grep)
if [ -n "$TOR_PROCS" ]; then
    echo "  发现 Tor 进程:"
    echo "$TOR_PROCS" | awk '{print "    PID:" $2 " CMD:" $11}'
else
    echo "  未发现 Tor 进程"
fi

echo ""
echo "[3] Tor 相关端口检查..."
for port in 9050 9051 9150 9151; do
    CONNS=$(lsof -i :"$port" 2>/dev/null | grep -v COMMAND | wc -l)
    if [ "$CONNS" -gt 0 ]; then
        echo "  端口 $port: $CONNS 个连接"
        lsof -i :"$port" 2>/dev/null | grep -v COMMAND | \
          awk '{print "    " $1 " PID:" $2 " " $3}'
    fi
done

echo ""
echo "[4] 网络流量统计（Tor特征）..."
TCPDUMP_FILE="$LOG_DIR/tor_capture_$TIMESTAMP.pcap"
timeout 10 tcpdump -i any -w "$TCPDUMP_FILE" \
  "tcp portrange 9050-9150 or tcp port 443" 2>/dev/null &

if [ -f "$TCPDUMP_FILE" ]; then
    echo "  捕获数据包统计:"
    tshark -r "$TCPDUMP_FILE" -q -z conv,tcp 2>/dev/null | \
      head -20
fi

echo ""
echo "[5] 生成分析报告..."
cat > "$LOG_DIR/report_$TIMESTAMP.txt" <<EOF
Tor 节点扫描报告 - $TIMESTAMP
============================
总中继数: $TOTAL_RELAYS
Guard 节点: $GUARD_COUNT
Exit 节点: $EXIT_COUNT
Stable 节点: $STABLE_COUNT
本地Tor进程: $(ps aux | grep -i "[t]or" | grep -v grep | wc -l)
EOF

echo "  报告已保存到: $LOG_DIR/report_$TIMESTAMP.txt"
echo ""
echo "[*] 扫描完成"

import subprocess
import json
import csv
import os
from datetime import datetime

class TorForensicsSuite:
    def __init__(self, output_dir="./tor_forensics"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def check_tor_processes(self):
        findings = []
        try:
            result = subprocess.run(
                ["ps", "aux"], capture_output=True, text=True
            )
            for line in result.stdout.split("\n"):
                if "tor" in line.lower() and "grep" not in line:
                    parts = line.split()
                    findings.append({
                        "pid": parts[1] if len(parts) > 1 else "unknown",
                        "user": parts[0] if len(parts) > 0 else "unknown",
                        "command": " ".join(parts[10:]) if len(parts) > 10 else "",
                        "cpu": parts[2] if len(parts) > 2 else "0",
                        "memory": parts[3] if len(parts) > 3 else "0"
                    })
        except Exception as e:
            findings.append({"error": str(e)})
        return findings

    def check_tor_config(self, torrc_path="/etc/tor/torrc"):
        config = {}
        if os.path.exists(torrc_path):
            with open(torrc_path, "r") as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith("#"):
                        parts = line.split(None, 1)
                        if len(parts) == 2:
                            config[parts[0]] = parts[1]
        return config

    def analyze_tor_connections(self):
        connections = []
        try:
            result = subprocess.run(
                ["lsof", "-i", "-n", "-P"],
                capture_output=True, text=True
            )
            tor_ports = ["9050", "9051", "9150", "9151"]
            for line in result.stdout.split("\n"):
                for port in tor_ports:
                    if f":{port}" in line:
                        parts = line.split()
                        if len(parts) >= 9:
                            connections.append({
                                "process": parts[0],
                                "pid": parts[1],
                                "user": parts[2],
                                "fd": parts[3],
                                "type": parts[4],
                                "device": parts[5],
                                "size_off": parts[6],
                                "node": parts[7],
                                "name": parts[8]
                            })
        except Exception as e:
            connections.append({"error": str(e)})
        return connections

    def check_tor_browser_artifacts(self):
        artifacts = {
            "windows": [
                os.path.expanduser("~/AppData/Roaming/Tor Browser"),
                os.path.expanduser("~/Desktop/Tor Browser"),
            ],
            "linux": [
                os.path.expanduser("~/.tor-browser"),
                "/usr/share/tor-browser",
                os.path.expanduser("~/tor-browser_en-US"),
            ],
            "darwin": [
                os.path.expanduser("~/Library/Application Support/TorBrowser"),
                "/Applications/Tor Browser.app",
            ]
        }
        found = []
        import platform
        system = platform.system().lower()
        paths = artifacts.get(system, [])
        for path in paths:
            if os.path.exists(path):
                found.append({
                    "path": path,
                    "exists": True,
                    "modified": datetime.fromtimestamp(
                        os.path.getmtime(path)
                    ).isoformat()
                })
        return found

    def generate_forensics_report(self):
        report = {
            "timestamp": datetime.now().isoformat(),
            "tor_processes": self.check_tor_processes(),
            "tor_config": self.check_tor_config(),
            "tor_connections": self.analyze_tor_connections(),
            "tor_browser_artifacts": self.check_tor_browser_artifacts()
        }

        report_path = os.path.join(
            self.output_dir,
            f"tor_forensics_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        )
        with open(report_path, "w") as f:
            json.dump(report, f, indent=2, ensure_ascii=False)

        print(f"取证报告已生成: {report_path}")
        print(f"发现 Tor 进程: {len(report['tor_processes'])} 个")
        print(f"Tor 连接数: {len(report['tor_connections'])} 个")
        print(f"Tor Browser 工件: {len(report['tor_browser_artifacts'])} 个")

        return report

if __name__ == "__main__":
    suite = TorForensicsSuite()
    report = suite.generate_forensics_report()

# Tor 出口节点 IP（调查期间捕获）
Exit Node IPs: [已收录于威胁情报平台]

# Rclone 配置哈希
SHA256: 3b82f7d4e5c6a8b9d0e1f2a3b4c5d6e7f8a9b0c1

# DarkSide 勒索信特征
Ransom Note: readMe.txt
BTC Address: 1HWq7h1XFRh1QCNfMevh3a5G2N4V2h1H6Y

# C2 通信指标
Protocol: Tor Hidden Service (.onion)
Communication: SMTP over Tor (Port 587)

# AlphaBay 暗网地址（已下线）
AlphaBay Hidden Service: alphaBay2sj2g2qfbu.onion (已失效)

# 管理员关联信息
Email: pimp_alex_91@hotmail.com
Real Name: Alexandre Cazes
Citizenship: Canadian

# 比特币地址（部分，来自法庭文件）
Funding Address: 1CBssRwbRm7Kf9d6Jf9N1t8d6y2bJ1n9yM

# 服务器位置
Primary: 多伦多，加拿大
Secondary: 阿姆斯特丹，荷兰