低轨卫星互联网安全取证深度分析

# 使用Satellite Tracker工具获取当前可见LEO卫星信息
# 安装: pip install orbital-trackpredict
python3 -c "
from orbital_trackpredict import predict
satellites = predict.get_visible_satellites(
    latitude=39.9042, longitude=116.4074,
    min_elevation=10,
    tle_file='stations.txt'
)
for sat in satellites:
    print(f'{sat.name}: Az={sat.azimuth:.1f}° El={sat.elevation:.1f}° Range={sat.range:.0f}km')
"

# 使用Wireshark的DVB-S2 dissector解析卫星链路流量
# 需要安装wireshark-plugin-dvbs2
tshark -r satellite_capture.pcap -Y "dvbs2" -T fields \
  -e dvbs2.modcod -e dvbs2.frame_type -e dvbs2.payload_length \
  -e dvbs2.plscodes -e dvbs2.bch_err_count 2>/dev/null

# 提取卫星链路中的IP流量进行分析
tshark -r satellite_capture.pcap -Y "ip.addr == 10.0.0.0/8 && dvbs2" \
  -T fields -e frame.time -e ip.src -e ip.dst -e tcp.port \
  -e udp.port -e dvbs2.modcod > satellite_ip_analysis.csv

# 搭建LEO卫星取证分析环境（Ubuntu 22.04+）
sudo apt update && sudo apt install -y \
  wireshark tshark \
  gnuradio \
  binwalk \
  nmap masscan \
  ghidra \
  python3-pip

pip3 install --user \
  scapy \
  pyelftools \
  capstone \
  keystone-engine \
  unicorn \
  satellite-js

# 安装GNU Radio DVB-S2模块
sudo apt install -y gr-dvbs2
# 或从源码编译
# git clone https://github.com/akosinern/gr-dvbs2.git
# cd gr-dvbs2 && mkdir build && cd build
# cmake .. && make -j$(nproc) && sudo make install && sudo ldconfig

# 使用GNU Radio捕获卫星下行信号进行DVB-S2X协议分析
# 创建GNU Radio流图捕获脚本
cat > dvbs2_capture.py << 'PYEOF'
#!/usr/bin/env python3
from gnuradio import gr, blocks, dvb
import numpy as np

class dvbs2_capture(gr.top_block):
    def __init__(self):
        gr.top_block.__init__(self, "DVB-S2X Capture")
        samp_rate = 2.4e6
        center_freq = 11720e6
        self.source = blocks.ufd_source(
            device_addr="type=b200",
            stream_args=gr.stream_args("fc32"),
            center_freq=center_freq,
            samp_rate=samp_rate,
            gain=40
        )
        self.fft_sink = blocks.probe_signal_vc(1024)
        self.file_sink = blocks.file_sink(
            gr.sizeof_gr_complex, "/tmp/satellite_capture.raw"
        )
        self.connect(self.source, self.fft_sink)
        self.connect(self.source, self.file_sink)

if __name__ == '__main__':
    tb = dvbs2_capture()
    tb.start()
    import time
    time.sleep(60)
    tb.stop()
    tb.wait()
PYEOF
python3 dvbs2_capture.py

# 分析捕获的信号频谱特征
python3 << 'PYEOF'
import numpy as np
from scipy import signal

data = np.fromfile('/tmp/satellite_capture.raw', dtype=np.complex64)
f, t, Sxx = signal.spectrogram(data, fs=2.4e6, nperseg=4096)
peak_freq = f[np.argmax(np.mean(Sxx, axis=1))]
bandwidth = f[Sxx.mean(axis=1) > Sxx.mean().max() * 0.5].ptp()
print(f"Peak frequency offset: {peak_freq:.0f} Hz")
print(f"Estimated signal bandwidth: {bandwidth:.0f} Hz")
print(f"Mean power: {10*np.log10(np.mean(np.abs(data)**2)):.1f} dBFS")
PYEOF

#!/usr/bin/env python3
import time
import struct
import hashlib
from collections import deque

class StarlinkLinkIntegrityChecker:
    def __init__(self, window_size=1000):
        self.latency_window = deque(maxlen=window_size)
        self.throughput_window = deque(maxlen=window_size)
        self.crc_error_window = deque(maxlen=window_size)
        self.alerts = []

    def check_round_trip_delay(self, timestamp_send, timestamp_recv, sat_id):
        rtt = (timestamp_recv - timestamp_send) * 1000
        self.latency_window.append(rtt)
        if len(self.latency_window) >= 100:
            mean_rtt = sum(self.latency_window) / len(self.latency_window)
            std_rtt = (sum((x - mean_rtt) ** 2 for x in self.latency_window) / len(self.latency_window)) ** 0.5
            if rtt > mean_rtt + 3 * std_rtt:
                self.alerts.append({
                    'type': 'RTT_ANOMALY',
                    'severity': 'HIGH',
                    'satellite': sat_id,
                    'rtt': rtt,
                    'baseline_mean': mean_rtt,
                    'baseline_std': std_rtt,
                    'timestamp': time.time(),
                    'attack_vector': 'T1557 - Adversary-in-the-Middle'
                })
                return True
        return False

    def check_throughput_consistency(self, bytes_transferred, interval_sec, expected_throughput_mbps):
        actual_throughput = (bytes_transferred * 8) / (interval_sec * 1e6)
        self.throughput_window.append(actual_throughput)
        if len(self.throughput_window) >= 50:
            recent_avg = sum(list(self.throughput_window)[-10:]) / 10
            if actual_throughput < expected_throughput_mbps * 0.3:
                self.alerts.append({
                    'type': 'THROUGHPUT_DEGRADATION',
                    'severity': 'MEDIUM',
                    'actual_mbps': actual_throughput,
                    'expected_mbps': expected_throughput_mbps,
                    'degradation_ratio': actual_throughput / expected_throughput_mbps,
                    'timestamp': time.time(),
                    'attack_vector': 'T1561.002 - Disk Wipe (Signal Degradation)'
                })
                return True
        return False

    def check_crc_error_rate(self, total_frames, crc_errors):
        error_rate = crc_errors / total_frames if total_frames > 0 else 0
        self.crc_error_window.append(error_rate)
        if len(self.crc_error_window) >= 20:
            baseline_rate = sum(self.crc_error_window) / len(self.crc_error_window)
            if error_rate > baseline_rate * 5 and error_rate > 0.01:
                self.alerts.append({
                    'type': 'CRC_ERROR_SPIKE',
                    'severity': 'HIGH',
                    'current_rate': error_rate,
                    'baseline_rate': baseline_rate,
                    'total_frames': total_frames,
                    'timestamp': time.time(),
                    'attack_vector': 'T1498 - Network Denial of Service'
                })
                return True
        return False

    def generate_report(self):
        report = {
            'total_alerts': len(self.alerts),
            'high_severity': len([a for a in self.alerts if a['severity'] == 'HIGH']),
            'medium_severity': len([a for a in self.alerts if a['severity'] == 'MEDIUM']),
            'alerts': self.alerts
        }
        return report

checker = StarlinkLinkIntegrityChecker()
for i in range(100):
    checker.check_round_trip_delay(time.time(), time.time() + 0.035, "SAT-001")
report = checker.generate_report()
print(f"Link integrity check complete: {report['total_alerts']} alerts")

# 阶段1：地面站网络资产发现
# 使用Nmap进行全端口扫描和服务识别
nmap -sS -sV -O -p- \
  --script=banner,ssl-cert,http-title \
  -T4 \
  -oX gateway_scan.xml \
  -oN gateway_scan.txt \
  192.168.100.0/24

# 阶段2：SNMP安全审计（卫星网管常用协议）
# 枚举SNMP社区字符串
snmpwalk -v2c -c public 192.168.100.1 1.3.6.1.2.1.1
snmpwalk -v2c -c private 192.168.100.1 1.3.6.1.2.1.1

# 检查SNMP MIB中的卫星设备信息
snmpwalk -v2c -c public 192.168.100.1 1.3.6.1.4.1 2>/dev/null | \
  grep -E "satellite|gateway|modem|frequency"

# 阶段3：NETCONF/YANG配置审计
# 卫星设备常用NETCONF管理
python3 -c "
import xml.etree.ElementTree as ET
from ncclient import manager

with manager.connect(
    host='192.168.100.1',
    port=830,
    username='admin',
    password='admin',
    hostkey_verify=False
) as m:
    config = m.get_config(source='running').data_xml
    root = ET.fromstring(config)
    for elem in root.iter():
        if 'satellite' in elem.tag.lower() or 'frequency' in elem.tag.lower():
            print(f'{elem.tag}: {elem.text}')
" 2>/dev/null

# 阶段4：Web管理界面安全测试
# 使用ffuf进行目录和参数枚举
ffuf -u http://192.168.100.10/FUZZ \
  -w /usr/share/wordlists/dirb/common.txt \
  -mc 200,301,302,403 \
  -o web_enum.json -of json

# 测试SQL注入
ffuf -u "http://192.168.100.10/api/satellite?id=FUZZ" \
  -w /usr/share/seclists/Fuzzing/special-chars.txt \
  -mr "error|warning|syntax" \
  -o sqli_test.json -of json

# 地面站Linux服务器取证脚本
#!/bin/bash
EVIDENCE_DIR="/forensics/gateway_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$EVIDENCE_DIR"/{memory,network,filesystem,logs}

# 内存取证
echo "[*] Capturing memory dump..."
sudo dd if=/proc/kcore of="$EVIDENCE_DIR/memory/kcore.raw" bs=1M 2>/dev/null
sudo cp /proc/meminfo "$EVIDENCE_DIR/memory/meminfo.txt"
# 使用LiME获取完整内存镜像
sudo insmod lime.ko "path=$EVIDENCE_DIR/memory/memdump.lime format=lime"

# 网络状态快照
echo "[*] Capturing network state..."
ss -tlnp > "$EVIDENCE_DIR/network/listening_ports.txt"
ss -tanp > "$EVIDENCE_DIR/network/all_connections.txt"
ip route > "$EVIDENCE_DIR/network/routes.txt"
ip addr > "$EVIDENCE_DIR/network/interfaces.txt"
sudo iptables -L -v -n > "$EVIDENCE_DIR/network/iptables.txt"
# 捕获ARP缓存
arp -an > "$EVIDENCE_DIR/network/arp_cache.txt"

# 进程快照
echo "[*] Capturing process information..."
ps auxwww > "$EVIDENCE_DIR/filesystem/processes.txt"
ls -la /proc/*/exe 2>/dev/null > "$EVIDENCE_DIR/filesystem/proc_exe.txt"
ls -la /proc/*/cmdline 2>/dev/null > "$EVIDENCE_DIR/filesystem/proc_cmdline.txt"

# 文件系统时间线
echo "[*] Building filesystem timeline..."
find / -xdev -type f -printf '%T+ %m %u:%g %p\n' 2>/dev/null | \
  sort > "$EVIDENCE_DIR/filesystem/timeline.txt"

# 日志收集
echo "[*] Collecting logs..."
for logfile in /var/log/syslog /var/log/auth.log /var/log/secure \
  /var/log/messages /var/log/audit/audit.log; do
  if [ -f "$logfile" ]; then
    cp "$logfile" "$EVIDENCE_DIR/logs/"
  fi
done
# 卫星网管系统日志
find /var/log/ -name "*.log" -newer /etc/hostname -exec \
  cp {} "$EVIDENCE_DIR/logs/" \; 2>/dev/null

# 计算所有取证证据的哈希值
echo "[*] Computing evidence hashes..."
find "$EVIDENCE_DIR" -type f -exec sha256sum {} \; > "$EVIDENCE_DIR/evidence_hashes.txt"
echo "[+] Forensics collection complete: $EVIDENCE_DIR"

# 方法1：通过OTA更新通道获取固件
# 使用mitmproxy拦截Starlink终端的固件更新请求
mitmproxy --mode transparent \
  --set upstream_cert=false \
  -w starlink_traffic.cap \
  --listen-port 8080 &

# 配置终端代理后监控固件更新流量
tcpdump -i eth0 port 8080 -w proxy_capture.pcap

# 从捕获的流量中提取固件包
python3 << 'PYEOF'
import re

def extract_firmware_from_pcap(pcap_path):
    with open(pcap_path, 'rb') as f:
        data = f.read()
    magic_patterns = [
        b'UBI#',        # UBI filesystem
        b'hsqs',        # SquashFS
        b'\x7fELF',     # ELF binary
        b'MTD',         # MTD partition header
    ]
    offsets = []
    for magic in magic_patterns:
        pos = 0
        while True:
            idx = data.find(magic, pos)
            if idx == -1:
                break
            offsets.append((idx, magic.decode('ascii', errors='replace')))
            pos = idx + 1
    return offsets

results = extract_firmware_from_pcap('proxy_capture.pcap')
for offset, magic in results:
    print(f"Found {magic} at offset 0x{offset:08x}")
PYEOF

# 方法2：JTAG接口固件提取
# 使用JTAGulator识别调试引脚
python3 -c "
from serial.tools.list_ports import comports
for port in comports():
    print(f'Port: {port.device}, Description: {port.description}')
"

# 使用OpenOCD连接并转储Flash
openocd -f interface/jlink.cfg \
  -c 'transport select jtag' \
  -f target/stm32f4x.cfg \
  -c 'adapter speed 1000' \
  -c init \
  -c 'reset halt' \
  -c 'flash read_bank 0 /forensics/firmware_dump.bin' \
  -c shutdown

# 使用Binwalk解包固件镜像
binwalk -e firmware_update_v2024.1.bin --directory /tmp/firmware_extract

# 提取文件系统
sudo unsquashfs -d /tmp/squashfs-root firmware.squashfs

# 查找敏感文件和配置
find /tmp/squashfs-root -type f \( \
  -name "*.conf" -o -name "*.cfg" -o -name "*.key" -o \
  -name "*.pem" -o -name "*.crt" -o -name "passwd" -o \
  -name "shadow" -o -name "*.sh" -o -name "authorized_keys" \
\) -exec ls -la {} \; 2>/dev/null

# 搜索硬编码凭证
grep -rn "password\|passwd\|secret\|key\|token\|api_key" \
  /tmp/squashfs-root/etc/ \
  --include="*.conf" --include="*.cfg" --include="*.json" \
  2>/dev/null | head -50

# 搜索嵌入的私钥
find /tmp/squashfs-root -type f -exec grep -l "BEGIN.*PRIVATE KEY" {} \; 2>/dev/null

# 使用Ghidra headless模式进行自动化逆向分析
analyzeHeadless /tmp/firmware_analysis \
  -import /tmp/squashfs-root/usr/bin/satellite_modem \
  -postScript ghidra_satellite_analysis.py \
  -scriptPath /forensics/scripts \
  -deleteProject

# Ghidra脚本：识别卫星通信相关函数
cat > /forensics/scripts/ghidra_satellite_analysis.py << 'PYEOF'
from ghidra.program.model.symbol import SymbolType
from ghidra.app.decompiler import DecompInterface

program = getCurrentProgram()
listing = program.getListing()
func_mgr = program.getFunctionManager()

keywords = [
    'encrypt', 'decrypt', 'authenticate', 'modem',
    'frequency', 'beam', 'handover', 'telemetry',
    'command', 'update', 'firmware', 'download',
    'upload', 'connect', 'disconnect', 'key_exchange'
]

functions = func_mgr.getFunctions(True)
suspicious = []
for func in functions:
    name = func.getName().lower()
    for kw in keywords:
        if kw in name:
            suspicious.append({
                'name': func.getName(),
                'address': func.getEntryPoint(),
                'size': func.getBody().getNumAddresses(),
                'keyword_match': kw
            })
            break

with open('/tmp/suspicious_functions.json', 'w') as f:
    import json
    json.dump(suspicious, f, indent=2)
print(f"Identified {len(suspicious)} suspicious functions")
PYEOF

#!/usr/bin/env python3
import numpy as np
from scipy import signal
from collections import deque
import time
import json

class SpectrumAnomalyDetector:
    def __init__(self, center_freq=12e9, bandwidth=500e6, num_bins=4096):
        self.center_freq = center_freq
        self.bandwidth = bandwidth
        self.num_bins = num_bins
        self.freq_axis = np.linspace(
            center_freq - bandwidth/2,
            center_freq + bandwidth/2,
            num_bins
        )
        self.baseline_psd = None
        self.psd_history = deque(maxlen=100)
        self.anomaly_log = []

    def calibrate_baseline(self, baseline_samples):
        psd_list = []
        for sample in baseline_samples:
            f, psd = signal.welch(sample, fs=self.bandwidth, nperseg=self.num_bins)
            psd_list.append(psd)
        self.baseline_psd = np.mean(psd_list, axis=0)
        self.baseline_std = np.std(psd_list, axis=0)
        print(f"Baseline calibrated with {len(baseline_samples)} samples")

    def detect_jamming(self, current_psd):
        if self.baseline_psd is None:
            return False
        noise_floor = np.median(self.baseline_psd)
        current_floor = np.median(current_psd)
        power_ratio = current_floor / noise_floor
        if power_ratio > 3.0:
            self.anomaly_log.append({
                'type': 'JAMMING_DETECTED',
                'power_increase_db': 10 * np.log10(power_ratio),
                'timestamp': time.time(),
                'severity': 'CRITICAL'
            })
            return True
        freq_deviation = np.abs(current_psd - self.baseline_psd)
        max_deviation_bins = np.where(freq_deviation > 3 * self.baseline_std)[0]
        if len(max_deviation_bins) > self.num_bins * 0.3:
            self.anomaly_log.append({
                'type': 'WIDEBAND_JAMMING',
                'affected_bins': len(max_deviation_bins),
                'percentage': len(max_deviation_bins) / self.num_bins * 100,
                'timestamp': time.time(),
                'severity': 'CRITICAL'
            })
            return True
        return False

    def detect_spoofing(self, current_psd, current_signal):
        if self.baseline_psd is None:
            return False
        peak_freq_idx = np.argmax(current_psd)
        baseline_peak_idx = np.argmax(self.baseline_psd)
        freq_offset = abs(self.freq_axis[peak_freq_idx] - self.freq_axis[baseline_peak_idx])
        if freq_offset > 1e6:
            self.anomaly_log.append({
                'type': 'FREQUENCY_SPOOFING',
                'freq_offset_hz': freq_offset,
                'expected_freq': self.freq_axis[baseline_peak_idx],
                'detected_freq': self.freq_axis[peak_peak_idx] if peak_freq_idx else 0,
                'timestamp': time.time(),
                'severity': 'HIGH'
            })
            return True
        _, autocorr = signal.correlate(current_signal, current_signal, mode='full'), None
        autocorr = signal.correlate(current_signal, current_signal, mode='full')
        autocorr = autocorr[len(autocorr)//2:]
        autocorr = autocorr / autocorr[0]
        periodic_peaks = []
        for i in range(1, len(autocorr) - 1):
            if autocorr[i] > autocorr[i-1] and autocorr[i] > autocorr[i+1] and autocorr[i] > 0.8:
                periodic_peaks.append(i)
        if len(periodic_peaks) > 2:
            self.anomaly_log.append({
                'type': 'SPOOFED_SIGNAL_DETECTED',
                'periodic_peaks': len(periodic_peaks),
                'timestamp': time.time(),
                'severity': 'HIGH'
            })
            return True
        return False

    def generate取证报告(self):
        report = {
            'detection_period': {
                'start': self.anomaly_log[0]['timestamp'] if self.anomaly_log else time.time(),
                'end': self.anomaly_log[-1]['timestamp'] if self.anomaly_log else time.time()
            },
            'total_anomalies': len(self.anomaly_log),
            'jamming_events': len([a for a in self.anomaly_log if 'JAMMING' in a['type']]),
            'spoofing_events': len([a for a in self.anomaly_log if 'SPOOFING' in a['type']]),
            'critical_events': len([a for a in self.anomaly_log if a['severity'] == 'CRITICAL']),
            'anomalies': self.anomaly_log
        }
        return report

detector = SpectrumAnomalyDetector(center_freq=11.72e9, bandwidth=500e6)
baseline = [np.random.randn(4096) * 0.01 for _ in range(50)]
detector.calibrate_baseline(baseline)
print("Spectrum anomaly detector initialized")

# 使用USRP进行IQ数据采集并保存为SigMF格式
python3 << 'PYEOF'
import uhd
import numpy as np
import time
import json
import hashlib

def collect_iq_evidence(output_path, duration_sec=30, sample_rate=25e6, center_freq=11.72e9):
    usrp = uhd.usrp.MultiUSRP("type=b200")
    usrp.set_rx_rate(sample_rate)
    usrp.set_rx_freq(center_freq)
    usrp.set_rx_gain(40)
    num_samples = int(sample_rate * duration_sec)
    buffer_size = 4096
    all_samples = np.zeros(num_samples, dtype=np.complex64)
    for i in range(0, num_samples, buffer_size):
        samples = usrp.recv(buffer_size, timeout=5.0)
        all_samples[i:i+len(samples)] = samples
    all_samples.tofile(output_path)
    sigmf_metadata = {
        "global": {
            "core:datatype": "cf32_le",
            "core:sample_rate": sample_rate,
            "core:version": "0.0.1",
            "core:recording": "LEO_Satellite_Evidence"
        },
        "captures": [{
            "core:sample_start": 0,
            "core:frequency": center_freq,
            "core:time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        }],
        "annotations": []
    }
    meta_path = output_path + ".sigmf-meta"
    with open(meta_path, 'w') as f:
        json.dump(sigmf_metadata, f, indent=2)
    with open(output_path, 'rb') as f:
        sha256 = hashlib.sha256(f.read()).hexdigest()
    evidence_record = {
        'file': output_path,
        'sha256': sha256,
        'sample_rate': sample_rate,
        'center_freq': center_freq,
        'duration': duration_sec,
        'collection_time': time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    }
    print(f"IQ evidence collected: {output_path}")
    print(f"SHA256: {sha256}")
    return evidence_record

evidence = collect_iq_evidence(
    "/forensics/spectrum/iq_capture_20260719.raw",
    duration_sec=30,
    sample_rate=25e6,
    center_freq=11.72e9
)
print(json.dumps(evidence, indent=2))
PYEOF

# TDOA多站定位分析
python3 << 'PYEOF'
import numpy as np

def tdoa_localize(station_coords, tdoa_measurements, c=3e8):
    n_stations = len(station_coords)
    reference = station_coords[0]
    A = np.zeros((n_stations - 1, 2))
    b = np.zeros(n_stations - 1)
    for i in range(1, n_stations):
        A[i-1, 0] = station_coords[i][0] - reference[0]
        A[i-1, 1] = station_coords[i][1] - reference[1]
        d_ref = np.linalg.norm(reference)
        d_i = np.linalg.norm(station_coords[i])
        b[i-1] = tdoa_measurements[i-1] * c + 0.5 * (d_i**2 - d_ref**2)
    x, residuals, rank, sv = np.linalg.lstsq(A, b, rcond=None)
    return x

stations = [
    (39.9042, 116.4074, 50),
    (31.2304, 121.4737, 30),
    (23.1291, 113.2644, 40)
]
tdoa = [1.2e-6, 3.5e-6]
location = tdoa_localize(stations, tdoa)
print(f"Estimated jammer location: {location[0]:.4f}°N, {location[1]:.4f}°E")
PYEOF

# 卫星网络BGP安全审计
# 监控卫星网关的BGP路由变化
sudo tcpdump -i eth0 -w bgp_capture.pcap port 179 &
BGP_PID=$!

# 使用BGPStream分析路由异常
python3 << 'PYEOF'
import pybgpstream

stream = pybgpstream.BGPStream(
    from_time="2026-07-19 00:00:00",
    until_time="2026-07-19 23:59:59",
    record_type="updates",
    collectors=["route-views2"]
)

suspicious_prefixes = []
for rec in stream.records():
    for elem in rec:
        if elem.type == 'A':
            prefix = elem.fields['prefix']
            as_path = elem.fields['as-path']
            origin_as = as_path.split()[-1]
            satellite_asns = ['AS14593', 'AS14633']
            if origin_as in satellite_asns:
                suspicious_prefixes.append({
                    'prefix': prefix,
                    'as_path': as_path,
                    'origin_as': origin_as,
                    'timestamp': elem.time,
                    'type': 'ROUTE_HIJACK_SUSPECT'
                })

if suspicious_prefixes:
    print(f"Found {len(suspicious_prefixes)} suspicious route announcements")
    for sp in suspicious_prefixes:
        print(f"  {sp['prefix']} via {sp['as_path']}")
PYEOF

# 卫星网络SNMP管理安全审计
# 枚举卫星管理MIB
snmpbulkwalk -v2c -c public 10.0.0.1 \
  1.3.6.1.4.1.44967 \
  -On 2>/dev/null | head -100

# 检查SNMPv1/v2c的不安全配置
python3 << 'PYEOF'
import subprocess
import re

community_strings = ['public', 'private', 'admin', 'satellite', 'gateway',
                     'starlink', 'operator', 'nms', 'default', 'test']

results = []
for community in community_strings:
    cmd = f'snmpget -v2c -c {community} 10.0.0.1 1.3.6.1.2.1.1.1.0'
    proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=5)
    if proc.returncode == 0 and 'No Such' not in proc.stdout:
        results.append({
            'community': community,
            'response': proc.stdout.strip(),
            'status': 'VULNERABLE',
            'risk': 'SNMP community string guessable'
        })

for r in results:
    print(f"[VULN] Community '{r['community']}' accepted: {r['response']}")
PYEOF

#!/usr/bin/env python3
import struct
import time
from collections import defaultdict

class EncryptedLinkAnalyzer:
    def __init__(self):
        self.session_stats = defaultdict(lambda: {
            'packet_count': 0,
            'byte_count': 0,
            'first_seen': None,
            'last_seen': None,
            'packet_sizes': [],
            'inter_arrival_times': [],
            'tls_versions': set(),
            'cipher_suites': set()
        })
        self.alerts = []

    def analyze_tls_handshake(self, client_hello):
        tls_version = struct.unpack('!H', client_hello[9:11])[0]
        version_map = {0x0301: 'TLS 1.0', 0x0302: 'TLS 1.1', 0x0303: 'TLS 1.2', 0x0304: 'TLS 1.3'}
        cipher_suites = []
        cs_len = struct.unpack('!H', client_hello[43:45])[0]
        for i in range(0, cs_len, 2):
            cs = struct.unpack('!H', client_hello[45+i:47+i])[0]
            cipher_suites.append(f'0x{cs:04x}')
        if tls_version < 0x0303:
            self.alerts.append({
                'type': 'TLS_DOWNGRADE',
                'severity': 'HIGH',
                'version': version_map.get(tls_version, f'0x{tls_version:04x}'),
                'cipher_suites': cipher_suites[:5],
                'attack_vector': 'T1557.001 - LLMNR/NBT-NS Poisoning',
                'timestamp': time.time()
            })
        weak_ciphers = ['0x0005', '0x000a', '0x002f', '0x0035']
        for cs in cipher_suites:
            if cs in weak_ciphers:
                self.alerts.append({
                    'type': 'WEAK_CIPHER_SUITE',
                    'severity': 'MEDIUM',
                    'cipher_suite': cs,
                    'attack_vector': 'T1557.002 - ARP Cache Poisoning',
                    'timestamp': time.time()
                })

    def analyze_ipsec_tunnel(self, spi, seq_num, payload_len):
        if hasattr(self, '_last_seq_num') and spi in self._last_seq_num:
            last_seq = self._last_seq_num[spi]
            if seq_num < last_seq:
                self.alerts.append({
                    'type': 'IPSEC_REPLAY',
                    'severity': 'CRITICAL',
                    'spi': f'0x{spi:08x}',
                    'current_seq': seq_num,
                    'last_seq': last_seq,
                    'attack_vector': 'T1557 - Adversary-in-the-Middle',
                    'timestamp': time.time()
                })
        if not hasattr(self, '_last_seq_num'):
            self._last_seq_num = {}
        self._last_seq_num[spi] = seq_num

    def detect_key_exchange_anomaly(self, key_exchange_duration_ms, key_size_bits):
        if key_exchange_duration_ms < 10:
            self.alerts.append({
                'type': 'FAST_KEY_EXCHANGE',
                'severity': 'HIGH',
                'duration_ms': key_exchange_duration_ms,
                'key_size': key_size_bits,
                'possible_cause': 'Pre-computed keys or weak randomness',
                'attack_vector': 'T1557 - Adversary-in-the-Middle',
                'timestamp': time.time()
            })
        if key_size_bits < 256:
            self.alerts.append({
                'type': 'SHORT_KEY_LENGTH',
                'severity': 'HIGH',
                'key_size': key_size_bits,
                'minimum_recommended': 256,
                'attack_vector': 'T1110 - Brute Force',
                'timestamp': time.time()
            })

    def generate_forensic_report(self):
        critical = [a for a in self.alerts if a['severity'] == 'CRITICAL']
        high = [a for a in self.alerts if a['severity'] == 'HIGH']
        return {
            'total_alerts': len(self.alerts),
            'critical_count': len(critical),
            'high_count': len(high),
            'alerts': self.alerts,
            'recommendation': 'IMMEDIATE_REVIEW' if critical else 'SCHEDULED_REVIEW'
        }

analyzer = EncryptedLinkAnalyzer()
analyzer.analyze_tls_handshake(
    bytes.fromhex('160301005a020000560301' + '00' * 70)
)
analyzer.detect_key_exchange_anomaly(5, 128)
report = analyzer.generate_forensic_report()
print(f"Encryption analysis: {report['total_alerts']} alerts found")

#!/usr/bin/env python3
import json
import time
from datetime import datetime, timedelta

class SatelliteForensicsCorrelator:
    def __init__(self):
        self.evidence_store = []
        self.correlation_rules = []

    def add_evidence(self, evidence):
        self.evidence_store.append(evidence)

    def correlate_temporal(self, window_minutes=30):
        correlated_events = []
        for i, e1 in enumerate(self.evidence_store):
            for j, e2 in enumerate(self.evidence_store):
                if i >= j:
                    continue
                t1 = datetime.fromisoformat(e1['timestamp'])
                t2 = datetime.fromisoformat(e2['timestamp'])
                time_diff = abs((t2 - t1).total_seconds()) / 60
                if time_diff <= window_minutes:
                    if e1['source_segment'] != e2['source_segment']:
                        correlated_events.append({
                            'evidence_1': e1['id'],
                            'evidence_2': e2['id'],
                            'time_diff_minutes': time_diff,
                            'segments': [e1['source_segment'], e2['source_segment']],
                            'combined_severity': self._combine_severity(e1['severity'], e2['severity']),
                            'correlation_type': 'TEMPORAL_CROSS_SEGMENT'
                        })
        return correlated_events

    def _combine_severity(self, s1, s2):
        severity_rank = {'CRITICAL': 4, 'HIGH': 3, 'MEDIUM': 2, 'LOW': 1}
        max_sev = max(severity_rank.get(s1, 0), severity_rank.get(s2, 0))
        for k, v in severity_rank.items():
            if v == max_sev:
                return k

    def build_attack_chain(self):
        sorted_evidence = sorted(self.evidence_store, key=lambda x: x['timestamp'])
        chains = []
        current_chain = []
        for evidence in sorted_evidence:
            if not current_chain:
                current_chain.append(evidence)
            else:
                last = current_chain[-1]
                time_diff = (datetime.fromisoformat(evidence['timestamp']) -
                           datetime.fromisoformat(last['timestamp'])).total_seconds()
                if time_diff < 3600 and self._is_adversary_progression(last, evidence):
                    current_chain.append(evidence)
                else:
                    if len(current_chain) >= 3:
                        chains.append(current_chain[:])
                    current_chain = [evidence]
        if len(current_chain) >= 3:
            chains.append(current_chain)
        return chains

    def _is_adversary_progression(self, prev, curr):
        progression_map = {
            'RECONNAISSANCE': ['WEAPONIZATION', 'DELIVERY'],
            'WEAPONIZATION': ['EXPLOITATION', 'DELIVERY'],
            'EXPLOITATION': ['INSTALLATION', 'COMMAND_AND_CONTROL'],
            'INSTALLATION': ['COMMAND_AND_CONTROL', 'ACTION_ON_OBJECTIVES'],
            'COMMAND_AND_CONTROL': ['ACTION_ON_OBJECTIVES', 'EXFILTRATION']
        }
        prev_killchain = prev.get('killchain_phase', '')
        curr_killchain = curr.get('killchain_phase', '')
        return curr_killchain in progression_map.get(prev_killchain, [])

correlator = SatelliteForensicsCorrelator()
correlator.add_evidence({
    'id': 'EVD-001', 'timestamp': '2026-07-19T02:15:00Z',
    'source_segment': 'SPACE', 'severity': 'HIGH',
    'type': 'BEAM_ANOMALY', 'killchain_phase': 'WEAPONIZATION'
})
correlator.add_evidence({
    'id': 'EVD-002', 'timestamp': '2026-07-19T02:20:00Z',
    'source_segment': 'GROUND', 'severity': 'CRITICAL',
    'type': 'UNAUTHORIZED_COMMAND', 'killchain_phase': 'EXPLOITATION'
})
correlator.add_evidence({
    'id': 'EVD-003', 'timestamp': '2026-07-19T02:35:00Z',
    'source_segment': 'USER', 'severity': 'HIGH',
    'type': 'DATA_EXFILTRATION', 'killchain_phase': 'EXFILTRATION'
})
chains = correlator.build_attack_chain()
print(f"Identified {len(chains)} attack chain(s)")

title: LEO Satellite Gateway - Suspicious SNMP Community String Usage
id: 8a7b3c4d-5e6f-7890-abcd-ef1234567890
status: experimental
description: Detects suspicious SNMP community string attempts against satellite gateway devices
author: x7peeps Security Research
date: 2026/07/19
tags:
  - attack.discovery
  - attack.t1046
  - satellite-security
  - leo-gateway
logsource:
  category: network
  product: snmp
detection:
  selection_weak_community:
    snmp.community:
      - 'public'
      - 'private'
      - 'admin'
      - 'satellite'
      - 'gateway'
      - 'starlink'
      - 'default'
  selection_non_business_hours:
    timestamp|time: 'T(0[0-5]|2[2-3]):'
  condition: selection_weak_community and selection_non_business_hours
  timeframe: 5m
  level: high
fields:
  - snmp.community
  - src_ip
  - dest_ip
  - timestamp
falsepositives:
  - Legitimate monitoring during maintenance windows
---
title: LEO Satellite - BGP Route Hijack Detection
id: 9c8d7e6f-5a4b-3c21-0fed-cba987654321
status: experimental
description: Detects potential BGP route hijacking affecting satellite network prefixes
author: x7peeps Security Research
date: 2026/07/19
tags:
  - attack.collection
  - attack.t1048
  - satellite-security
  - bgp-hijack
logsource:
  category: network
  product: bgp
detection:
  selection_route_change:
    event_type: 'BGP_UPDATE'
    prefix:
      - '10.0.0.0/8'
      - '172.16.0.0/12'
      - '192.168.0.0/16'
  selection_origin_as:
    origin_as:
      - 'AS0'
      - 'AS23456'
  condition: selection_route_change and selection_origin_as
  level: critical
fields:
  - prefix
  - origin_as
  - as_path
  - community
  - timestamp
falsepositives:
  - Planned AS migration events
---
title: LEO Satellite Terminal - Firmware Anomaly Detection
id: 1b2c3d4e-5f6a-7890-bcde-f12345678901
status: experimental
description: Detects anomalies in satellite terminal firmware update patterns
author: x7peeps Security Research
date: 2026/07/19
tags:
  - attack.persistence
  - attack.t1542
  - satellite-security
  - firmware-anomaly
logsource:
  category: firmware_update
  product: satellite-terminal
detection:
  selection_downgrade:
    update_type: 'FIRMWARE'
    version_comparison: 'OLDER'
  selection_unsigned:
    update_type: 'FIRMWARE'
    signature_valid: false
  selection_off_hours:
    update_type: 'FIRMWARE'
    timestamp|time: 'T(0[0-4]|2[2-3]):'
  condition: selection_downgrade or selection_unsigned or selection_off_hours
  level: critical
fields:
  - terminal_id
  - current_version
  - target_version
  - signature_valid
  - timestamp
falsepositives:
  - Authorized emergency patching

#!/bin/bash
SATELLITE_LOG_DIR="/var/log/satellite"
GATEWAY_LOG_DIR="/var/log/gateway"
REPORT_DIR="/tmp/satellite_hunt_$(date +%Y%m%d)"
mkdir -p "$REPORT_DIR"

echo "[*] LEO Satellite Security Hunt Script"
echo "[*] Scan started: $(date -u +%Y-%m-%dT%H:%M:%SZ)"
echo ""

# 1. 地面站SNMP弱口令扫描
echo "[+] Phase 1: Gateway SNMP Community String Audit"
if [ -d "$GATEWAY_LOG_DIR" ]; then
    grep -h "SNMP" "$GATEWAY_LOG_DIR"/*.log 2>/dev/null | \
    grep -iE "(public|private|admin|satellite|gateway|default)" | \
    awk '{print $1, $2, $3, $NF}' | \
    sort | uniq -c | sort -rn > "$REPORT_DIR/snmp_weak_community.txt"
    WEAK_COUNT=$(wc -l < "$REPORT_DIR/snmp_weak_community.txt")
    echo "    Found $WEAK_COUNT suspicious SNMP community string entries"
fi

# 2. 异常终端连接检测
echo "[+] Phase 2: Terminal Anomaly Detection"
if [ -d "$SATELLITE_LOG_DIR" ]; then
    grep -h "terminal_auth" "$SATELLITE_LOG_DIR"/*.log 2>/dev/null | \
    grep -E "FAILED|TIMEOUT|UNKNOWN" | \
    awk '{print $4}' | sort | uniq -c | sort -rn | \
    awk '$1 > 10 {print "[ALERT] Terminal:", $2, "failed attempts:", $1}' \
    > "$REPORT_DIR/terminal_anomalies.txt"
    ALERT_COUNT=$(grep -c "ALERT" "$REPORT_DIR/terminal_anomalies.txt" 2>/dev/null || echo 0)
    echo "    Detected $ALERT_COUNT anomalous terminal patterns"
fi

# 3. BGP路由异常检查
echo "[+] Phase 3: BGP Route Anomaly Check"
if [ -d "$SATELLITE_LOG_DIR" ]; then
    grep -h "BGP_UPDATE" "$SATELLITE_LOG_DIR"/*.log 2>/dev/null | \
    awk '{for(i=1;i<=NF;i++) if($i ~ /prefix/) print $i, $(i+1)}' | \
    grep -E "^prefix (10\.|172\.|192\.168)" | \
    sort | uniq -c | sort -rn > "$REPORT_DIR/bgp_anomalies.txt"
    BGP_COUNT=$(wc -l < "$REPORT_DIR/bgp_anomalies.txt")
    echo "    Found $BGP_COUNT suspicious BGP route changes"
fi

# 4. 固件更新异常检测
echo "[+] Phase 4: Firmware Update Anomaly Detection"
if [ -d "$SATELLITE_LOG_DIR" ]; then
    grep -h "firmware_update" "$SATELLITE_LOG_DIR"/*.log 2>/dev/null | \
    grep -iE "(downgrade|unsigned|emergency)" | \
    awk '{print $0}' > "$REPORT_DIR/firmware_anomalies.txt"
    FW_COUNT=$(wc -l < "$REPORT_DIR/firmware_anomalies.txt")
    echo "    Found $FW_COUNT anomalous firmware update events"
fi

# 5. 异常频谱活动检测
echo "[+] Phase 5: Spectrum Anomaly Detection"
if [ -d "$SATELLITE_LOG_DIR" ]; then
    grep -h "spectrum" "$SATELLITE_LOG_DIR"/*.log 2>/dev/null | \
    grep -iE "(jamming|interference|spoof|unexpected_freq)" | \
    awk '{print $1, $2, $3}' | sort -k3 | uniq -c | sort -rn \
    > "$REPORT_DIR/spectrum_anomalies.txt"
    SPEC_COUNT=$(wc -l < "$REPORT_DIR/spectrum_anomalies.txt")
    echo "    Found $SPEC_COUNT spectrum anomaly events"
fi

echo ""
echo "[*] Hunt complete. Results saved to: $REPORT_DIR"
echo "[*] Summary:"
for f in "$REPORT_DIR"/*.txt; do
    fname=$(basename "$f")
    lines=$(wc -l < "$f")
    echo "    $fname: $lines entries"
done

rule LEO_Satellite_Terminal_Backdoor {
    meta:
        description = "Detects backdoor implants in LEO satellite terminal firmware"
        author = "x7peeps Security Research"
        date = "2026-07-19"
        reference = "Satellite terminal firmware security analysis"
        severity = "critical"
        mitre_attack = "T1014,T1195.002"

    strings:
        $magic = { 4C 45 4F 53 41 54 }  
        $backdoor_cmd = "AT+SHADOWCMD" ascii
        $debug_enable = "enable_debug_console" ascii
        $telnet_start = "/bin/telnetd -l /bin/sh" ascii
        $reverse_shell = "nc -e /bin/sh" ascii
        $hidden_proc = "hide_process_from_procfs" ascii
        $firmware_mod = "modify_firmware_header" ascii
        $crypto_weak = { 44 45 53 2D 45 43 42 }  

    condition:
        $magic at 0 and (2 of ($backdoor_cmd, $debug_enable, $telnet_start, $reverse_shell, $hidden_proc, $firmware_mod) or $crypto_weak)
}

rule LEO_Gateway_BGP_Hijack_Tool {
    meta:
        description = "Detects tools used for BGP route hijacking targeting satellite networks"
        author = "x7peeps Security Research"
        date = "2026-07-19"
        reference = "Satellite network BGP security analysis"
        severity = "high"
        mitre_attack = "T1557,T1040"

    strings:
        $bgp_update = "BGP_UPDATE" ascii
        $bgp_withdraw = "BGP_WITHDRAW" ascii
        $pref_orig = "origin_as=" ascii
        $as_path_mod = "prepend_as_path" ascii
        $comm_mod = "community_attribute" ascii
        $route_hijack = "hijack_prefix" ascii
        $bgp_codec = { 00 02 }  

    condition:
        filesize < 10MB and (3 of ($bgp_update, $bgp_withdraw, $pref_orig, $as_path_mod, $comm_mod, $route_hijack) or $bgp_codec)
}

rule LEO_Satellite_Spectrum_Jammer {
    meta:
        description = "Detects software-defined radio (SDR) tools configured for satellite signal jamming"
        author = "x7peeps Security Research"
        date = "2026-07-19"
        reference = "Satellite spectrum security analysis"
        severity = "high"
        mitre_attack = "T1565.001,T1499"

    strings:
        $sdr_init = "SoapySDR" ascii
        $uhd_config = "uhd_usrp_probe" ascii
        $hackrf = "hackrf_info" ascii
        $jammer_freq = "jamming_frequency" ascii
        $power_control = "set_tx_gain" ascii
        $sweep_mode = "frequency_sweep" ascii
        $kalibrate = "kalibrate" ascii

    condition:
        (3 of ($sdr_init, $uhd_config, $hackrf, $kalibrate)) or
        (2 of ($jammer_freq, $power_control, $sweep_mode))
}