自动驾驶系统安全取证深度分析

# 使用 CANoe 导出自动驾驶传感器同步日志
# 前提：已通过 Vector CAN 接口连接车辆 OBD-II 端口
canconvert sensor_sync_canoe.asc sensor_sync.csv --layout "timestamp,channel,id,dlc,data"

# 提取传感器时间戳同步信息
awk -F',' 'NR>1 && $4==8 {printf "TS=%s CH=%s ID=0x%s DL=%s DATA=%s%s%s%s%s%s%s%s\n", $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12}' sensor_sync.csv | head -50

# 使用 Velodyne 官方工具捕获原始 LiDAR 点云数据
# 前提：LiDAR 通过以太网连接，IP 为 192.168.1.201
velodyne_capture --ip 192.168.1.201 --port 2368 --output raw_lidar.pcap --duration 60

# 使用 tshark 分析 LiDAR 数据流特征
tshark -r raw_lidar.pcap -Y "udp.port == 2368" -T fields \
  -e frame.time_relative -e udp.length -e data.len \
  -e udp.srcport -e udp.dstport > lidar_flow.csv

# 检测 LiDAR 数据流中的异常包间隔（可能的注入信号）
python3 -c "
import csv
intervals = []
with open('lidar_flow.csv') as f:
    reader = csv.DictReader(f, delimiter='\t')
    prev = None
    for row in reader:
        ts = float(row['frame.time_relative'])
        if prev is not None:
            intervals.append(ts - prev)
        prev = ts
avg = sum(intervals) / len(intervals)
anomalies = [(i, v) for i, v in enumerate(intervals) if abs(v - avg) > 3 * (sum((x-avg)**2 for x in intervals)/len(intervals))**0.5]
print(f'平均包间隔: {avg*1000:.2f}ms, 异常包数: {len(anomalies)}')
for idx, val in anomalies[:20]:
    print(f'  异常帧 #{idx}: 间隔 {val*1000:.2f}ms (偏差 {(val-avg)/avg*100:.1f}%)')
"

# 使用 OpenCV 分析行车记录仪视频中的异常帧
python3 << 'EOF'
import cv2
import numpy as np
import sys

cap = cv2.VideoCapture(sys.argv[1] if len(sys.argv) > 1 else "dashcam_recording.mp4")
fps = cap.get(cv2.CAP_PROP_FPS)
frame_count = 0
anomaly_frames = []
prev_hist = None

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break
    frame_count += 1
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    hist = cv2.calcHist([gray], [0], None, [256], [0, 256])
    hist = cv2.normalize(hist, hist).flatten()
    if prev_hist is not None:
        diff = cv2.compareHist(prev_hist, hist, cv2.HISTCMP_BHATTACHARYYA)
        if diff > 0.4:
            ts = frame_count / fps
            anomaly_frames.append((frame_count, ts, diff))
            cv2.imwrite(f'anomaly_frame_{frame_count}.jpg', frame)
    prev_hist = hist

cap.release()
print(f'总帧数: {frame_count}, 异常帧数: {len(anomaly_frames)}')
for fc, ts, d in anomaly_frames[:20]:
    print(f'  帧 #{fc} ({ts:.2f}s): 直方图差异 = {d:.4f}')
EOF

# 使用 candump 实时捕获 CAN 总线流量
# 前提：Linux 系统已加载 CAN 驱动（如 SocketCAN）
sudo ip link set can0 up type can bitrate 500000
candump can0 -t a -c 1000 > can_capture_$(date +%Y%m%d_%H%M%S).log

# 分析 CAN 消息频率分布（检测异常高频注入）
cat can_capture_*.log | awk '{print $2}' | sort | uniq -c | sort -rn | head -30

# 使用 cansniffer 实时可视化 CAN 总线消息变化
cansniffer can0 -c

# 使用 CANalyzat0r 进行 CAN 总线安全审计
# 启动 CANalyzat0r 控制台
python3 canalyzat0r.py --interface can0

# 批量分析 CAN 日志中的异常消息
python3 << 'PYEOF'
import can
import collections
import statistics

def analyze_can_log(log_file):
    msg_counts = collections.Counter()
    msg_data_patterns = collections.defaultdict(list)
    msg_intervals = collections.defaultdict(list)
    prev_time = {}
    
    with open(log_file, 'r') as f:
        for line in f:
            parts = line.strip().split()
            if len(parts) < 3:
                continue
            ts = float(parts[0].strip('()'))
            msg_id = parts[1]
            data = parts[2]
            
            msg_counts[msg_id] += 1
            msg_data_patterns[msg_id].append(data)
            
            if msg_id in prev_time:
                msg_intervals[msg_id].append(ts - prev_time[msg_id])
            prev_time[msg_id] = ts
    
    print(f"{'消息ID':<12} {'计数':>8} {'频率Hz':>8} {'数据唯一值':>10} {'间隔均值ms':>12} {'异常标记':>8}")
    print("-" * 75)
    
    for msg_id in sorted(msg_counts.keys()):
        count = msg_counts[msg_id]
        intervals = msg_intervals.get(msg_id, [])
        avg_interval = statistics.mean(intervals) if intervals else 0
        freq = 1.0 / avg_interval if avg_interval > 0 else 0
        unique_data = len(set(msg_data_patterns[msg_id]))
        flag = ""
        
        if freq > 100:
            flag = "⚠️高频"
        if unique_data == 1:
            flag = "⚠️静态"
        if len(intervals) > 10:
            cv = statistics.stdev(intervals) / avg_interval if avg_interval > 0 else 0
            if cv > 0.5:
                flag = "⚠️抖动"
        
        print(f"0x{msg_id:<8} {count:>8} {freq:>8.1f} {unique_data:>10} {avg_interval*1000:>10.2f}ms {flag:>8}")

analyze_can_log("can_capture_20260720.log")
PYEOF

# 使用 Wireshark/tshark 分析车载以太网流量
# 捕获 SOME/IP (Scalable service-Oriented MiddlewarE over IP) 协议流量
tshark -r vehicle_ethernet.pcap -Y "someip" -T fields \
  -e frame.time -e ip.src -e ip.dst -e someip.service_id \
  -e someip.method_id -e someip.msg_type -e someip.return_code \
  -e tcp.len -e udp.length > someip_analysis.csv

# 检测 ARP 欺骗（车载以太网常见攻击）
tshark -r vehicle_ethernet.pcap -Y "arp" -T fields \
  -e frame.time -e arp.src.hw_mac -e arp.src.proto_ipv4 \
  -e arp.dst.proto_ipv4 -e arp.opcode | \
  sort | uniq -c | sort -rn | head -20

# 提取 DoIP (Diagnostics over IP) 诊断流量
tshark -r vehicle_ethernet.pcap -Y "doip" -T fields \
  -e frame.time -e doip.version -e doip.payload_type \
  -e doip.source_address -e doip.target_address \
  -e doip.length > doip_diag.csv

# 使用 USRP 软件无线电平台捕获 C-V2X PC5 信号
# 前提：已安装 GNU Radio 和 C-V2X 解码模块
gnuradio-companion cv2x_capture.grc --args "type=usrp,addr=192.168.10.2"

# 分析捕获的 V2X 消息（SAE J2735 BSM 格式）
python3 << 'PYEOF'
import json
import struct
import hashlib
from collections import defaultdict

def parse_bsm_payload(hex_data):
    result = {}
    if len(hex_data) < 38:
        return result
    msg_count = int(hex_data[0:2], 16)
    result['msg_count'] = msg_count
    temp_id = hex_data[2:8]
    result['temp_id'] = temp_id
    lat = struct.unpack('>i', bytes.fromhex(hex_data[8:16]))[0]
    result['latitude'] = lat / 1e7
    lon = struct.unpack('>i', bytes.fromhex(hex_data[16:24]))[0]
    result['longitude'] = lon / 1e7
    speed_raw = struct.unpack('>H', bytes.fromhex(hex_data[24:28]))[0]
    result['speed_mps'] = speed_raw * 0.02
    heading = struct.unpack('>H', bytes.fromhex(hex_data[28:32]))[0]
    result['heading'] = heading * 0.0125
    return result

vehicles = defaultdict(list)

with open("cv2x_raw.hex", "r") as f:
    for line in f:
        parts = line.strip().split()
        if len(parts) < 3:
            continue
        ts, rssi, hex_data = parts[0], parts[1], parts[2]
        bsm = parse_bsm_payload(hex_data)
        if bsm:
            vid = bsm['temp_id']
            vehicles[vid].append({
                'timestamp': ts,
                'rssi': float(rssi),
                'lat': bsm['latitude'],
                'lon': bsm['longitude'],
                'speed': bsm['speed_mps']
            })

print(f"检测到 {len(vehicles)} 个唯一车辆")
for vid, msgs in vehicles.items():
    lats = [m['lat'] for m in msgs]
    speeds = [m['speed'] for m in msgs]
    rssis = [m['rssi'] for m in msgs]
    lat_jump = max(abs(lats[i] - lats[i-1]) for i in range(1, len(lats))) if len(lats) > 1 else 0
    speed_max = max(speeds) if speeds else 0
    rssi_std = (sum((r - sum(rssis)/len(rssis))**2 for r in rssis) / len(rssis))**0.5 if len(rssis) > 1 else 0
    
    flags = []
    if lat_jump > 0.001:
        flags.append("⚠️位置跳变")
    if speed_max > 50:
        flags.append("⚠️超速")
    if rssi_std > 15:
        flags.append("⚠️信号异常")
    
    print(f"  车辆 {vid}: {len(msgs)} 条消息, 位置跳变 {lat_jump*111000:.1f}m, RSSI标准差 {rssi_std:.1f}dB {' '.join(flags)}")
PYEOF

# 使用 SDR 工具扫描 V2X 频段活动
# 扫描 5.85-5.925 GHz ITS 频段
hackrf_transfer -r v2x_capture.raw -f 5890000000 -s 20000000 -g 40 -l 32 -n 200000000

# 使用 inspectrum 可视化 V2X 信号特征
inspectrum v2x_capture.raw

# 分析 V2X 消息的数字证书链
openssl cms -verify -in v2x_message.der -inform DER -CAfile its_root_ca.pem \
  -out signed_content.bin -outform DER 2>&1 | head -20

# 检查 V2X 证书吊销状态（CRL/OCSP）
openssl ocsp -issuer its_ca.pem -cert vehicle_cert.pem \
  -url http://crl.its-safety.org/ocsp -resp_text 2>/dev/null | grep -E "HTTP/1|Cert Status|This Update"

# 提取 OTA 更新包进行完整性验证
# 前提：已获取 OTA 更新包（通常为 .zip/.tar.gz 格式）
unzip -o ota_update_v2.3.1.zip -d ota_extracted/

# 验证更新包数字签名
gpg --verify ota_update.sig ota_update.bin 2>&1
openssl dgst -sha256 -verify ota_public_key.pem -signature ota_update.sig ota_update.bin

# 计算固件哈希与官方发布值比对
sha256sum ota_update.bin
cat ota_update.sha256sum

# 提取 ELF 格式固件进行逆向分析
binwalk -e ota_extracted/firmware.bin
strings ota_extracted/firmware.bin.elf | grep -iE "password|secret|key|token|api" | head -30

# 检查固件中的已知漏洞组件
cve-bin-tool ota_extracted/firmware.bin --format csv -o firmware_vuln_report.csv 2>/dev/null

# 使用 UDS (Unified Diagnostic Services) 通过 DoIP 读取 ECU 固件
# 前提：已连接至车辆以太网诊断端口
python3 << 'PYEOF'
import socket
import struct

DOIP_PORT = 13400
ECU_ADDRESS = 0x0010
TARGET_ADDRESS = 0x0E00

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("192.168.1.100", DOIP_PORT))

def send_doip(payload_type, target, source, data):
    length = 4 + len(data)
    header = struct.pack('>BBHII', 0x01, 0x00, payload_type, length, target, source)
    sock.send(header + data)

def send_uds(service, sub_func=None, data=b''):
    uds_payload = bytes([service])
    if sub_func is not None:
        uds_payload += bytes([sub_func])
    uds_payload += data
    send_doip(0x8001, TARGET_ADDRESS, ECU_ADDRESS, uds_payload)
    resp = sock.recv(4096)
    return resp[13:]

send_uds(0x10, 0x03)

resp = send_uds(0x34, 0x00, struct.pack('>IH', 0x00000000, 0x0400))
print(f"RequestDownload 响应: {resp.hex()}")

chunk_size = 0x400
total_size = 0x40000
firmware_data = b''

for offset in range(0, total_size, chunk_size):
    resp = send_uds(0x36, bytes([offset // 0x100 & 0xFF]), 
                     struct.pack('>H', min(chunk_size, total_size - offset)))
    if resp[0] == 0x76:
        firmware_data += resp[3:]

with open('ecu_firmware_dump.bin', 'wb') as f:
    f.write(firmware_data)
print(f"固件提取完成: {len(firmware_data)} 字节")
sock.close()
PYEOF

# 分析固件签名算法与密钥参数
python3 << 'PYEOF'
import hashlib
import struct
import os

with open('ecu_firmware_dump.bin', 'rb') as f:
    data = f.read()

signatures = {
    'RSA-2048': b'\x30\x82\x01\x22',
    'ECDSA-P256': b'\x30\x46\x02\x21',
    'Ed25519': b'\x40\x20',
}

print("固件头部信息:")
print(f"  文件大小: {len(data)} 字节")
print(f"  MD5: {hashlib.md5(data).hexdigest()}")
print(f"  SHA-256: {hashlib.sha256(data).hexdigest()}")

magic_patterns = {
    b'PK': 'ZIP 归档',
    b'\x1f\x8b': 'GZIP 压缩',
    b'hsqs': 'SquashFS 文件系统',
    b'UBI#': 'UBIFS 文件系统',
    b'\x7fELF': 'ELF 可执行文件',
    b'\xd0\xcf\x11\xe0': 'MSI 固件包',
    b'LUFA': 'AVR USB 固件',
}

for magic, desc in magic_patterns.items():
    offset = data.find(magic)
    if offset >= 0:
        print(f"  检测到格式: {desc} @ 偏移 0x{offset:x}")

for name, sig in signatures.items():
    count = data.count(sig)
    if count > 0:
        print(f"  检测到签名格式: {name} ({count} 处)")

key_patterns = [b'BEGIN PUBLIC KEY', b'BEGIN CERTIFICATE', b'-----BEGIN']
for pat in key_patterns:
    offset = data.find(pat)
    if offset >= 0:
        end = data.find(b'-----END', offset)
        block = data[offset:end+50] if end >= 0 else data[offset:offset+200]
        print(f"  检测到密钥材料 @ 偏移 0x{offset:x}: {block[:60]}...")
PYEOF

# 解析自动驾驶感知日志（假设为 ROS bag 格式）
# 安装: pip install rosbag
python3 << 'PYEOF'
import rosbag
import json
from collections import defaultdict

def analyze_perception_log(bag_file):
    bag = rosbag.Bag(bag_file)
    detections = []
    sensor_msgs = defaultdict(int)
    anomaly_scores = []
    
    for topic, msg, t in bag.read_messages(topics=[
        '/perception/objects',
        '/sensors/camera/front',
        '/sensors/lidar/points',
        '/sensors/radar/tracks'
    ]):
        ts = t.to_sec()
        sensor_msgs[topic] += 1
        
        if topic == '/perception/objects':
            obj_count = len(msg.objects) if hasattr(msg, 'objects') else 0
            confidences = [obj.confidence for obj in msg.objects] if hasattr(msg, 'objects') else []
            avg_conf = sum(confidences) / len(confidences) if confidences else 0
            
            detections.append({
                'timestamp': ts,
                'object_count': obj_count,
                'avg_confidence': avg_conf,
                'min_confidence': min(confidences) if confidences else 0,
                'max_confidence': max(confidences) if confidences else 0
            })
    
    print(f"感知日志时间范围: {bag.get_start_time():.2f} - {bag.get_end_time():.2f}")
    print(f"各话题消息数:")
    for topic, count in sensor_msgs.items():
        print(f"  {topic}: {count}")
    
    if detections:
        confs = [d['avg_confidence'] for d in detections]
        avg_all = sum(confs) / len(confs)
        low_conf_dets = [d for d in detections if d['avg_confidence'] < avg_all * 0.5]
        zero_dets = [d for d in detections if d['object_count'] == 0]
        
        print(f"\n目标检测统计:")
        print(f"  总帧数: {len(detections)}")
        print(f"  平均置信度: {avg_all:.3f}")
        print(f"  低置信度帧数: {len(low_conf_dets)} ({len(low_conf_dets)/len(detections)*100:.1f}%)")
        print(f"  零检测帧数: {len(zero_dets)} ({len(zero_dets)/len(detections)*100:.1f}%)")
        
        if len(detections) > 100:
            window_size = 50
            moving_avg = []
            for i in range(window_size, len(confs)):
                moving_avg.append(sum(confs[i-window_size:i]) / window_size)
            sudden_drops = []
            for i in range(1, len(moving_avg)):
                if moving_avg[i-1] - moving_avg[i] > 0.3:
                    sudden_drops.append((i + window_size, moving_avg[i-1], moving_avg[i]))
            if sudden_drops:
                print(f"\n检测到置信度突然下降事件:")
                for idx, prev, curr in sudden_drops[:10]:
                    print(f"  帧 #{idx}: {prev:.3f} → {curr:.3f} (下降 {(prev-curr)*100:.1f}%)")

analyze_perception_log("vehicle_log_20260720.bag")
PYEOF

# 提取和分析决策规划层日志
python3 << 'PYEOF'
import json
import csv
from datetime import datetime

def analyze_planning_decisions(log_file):
    with open(log_file) as f:
        entries = [json.loads(line) for line in f]
    
    print(f"决策日志条目数: {len(entries)}")
    
    decision_types = {}
    critical_events = []
    
    for entry in entries:
        ts = entry.get('timestamp', 0)
        dtype = entry.get('decision_type', 'unknown')
        decision_types[dtype] = decision_types.get(dtype, 0) + 1
        
        if dtype in ['emergency_brake', 'emergency_lane_change', 'collision_avoidance']:
            critical_events.append(entry)
        
        speed_cmd = entry.get('speed_command', 0)
        steering_cmd = entry.get('steering_command', 0)
        if abs(steering_cmd) > 0.8:
            critical_events.append({**entry, 'anomaly': 'excessive_steering'})
    
    print("\n决策类型分布:")
    for dtype, count in sorted(decision_types.items(), key=lambda x: -x[1]):
        print(f"  {dtype}: {count} ({count/len(entries)*100:.1f}%)")
    
    print(f"\n关键事件数: {len(critical_events)}")
    for event in critical_events[:15]:
        ts_str = datetime.fromtimestamp(event['timestamp']).strftime('%H:%M:%S.%f')[:-3]
        anomaly = event.get('anomaly', event.get('decision_type', ''))
        print(f"  [{ts_str}] {anomaly} - 速度: {event.get('speed_command', 'N/A')} m/s, 转向: {event.get('steering_command', 'N/A')} rad")
    
    return critical_events

analyze_planning_decisions("planning_log.jsonl")
PYEOF

# 将 CAN 总线控制指令与决策规划日志进行时序关联
python3 << 'PYEOF'
import can
import json
from datetime import datetime

CRITICAL_CAN_IDS = {
    '0x0A0': 'Steering_Angle_Cmd',
    '0x0A1': 'Brake_Pressure_Cmd', 
    '0x0A2': 'Throttle_Position_Cmd',
    '0x0A3': 'Gear_Selection_Cmd',
    '0x120': 'Vehicle_Speed_Feedback',
    '0x121': 'Steering_Angle_Feedback',
}

def correlate_can_planning(can_log, planning_log, tolerance_ms=50):
    can_entries = []
    with open(can_log) as f:
        for line in f:
            parts = line.strip().split()
            if len(parts) >= 3 and parts[1] in CRITICAL_CAN_IDS:
                ts = float(parts[0].strip('()'))
                msg_id = parts[1]
                data = parts[2]
                can_entries.append({'timestamp': ts, 'id': msg_id, 'data': data})
    
    with open(planning_log) as f:
        plan_entries = [json.loads(line) for line in f]
    
    mismatches = []
    for plan in plan_entries:
        plan_ts = plan['timestamp']
        closest_can = min(can_entries, key=lambda x: abs(x['timestamp'] - plan_ts), default=None)
        if closest_can and abs(closest_can['timestamp'] - plan_ts) * 1000 < tolerance_ms:
            pass
    
    print(f"CAN 控制指令数: {len(can_entries)}")
    print(f"规划决策数: {len(plan_entries)}")
    
    for ce in can_entries[:50]:
        ts_str = datetime.fromtimestamp(ce['timestamp']).strftime('%H:%M:%S.%f')[:-3]
        name = CRITICAL_CAN_IDS.get(ce['id'], 'Unknown')
        print(f"  [{ts_str}] {ce['id']} ({name}): {ce['data']}")

correlate_can_planning("can_critical.log", "planning_log.jsonl")
PYEOF

# 分析 IVI 系统蓝牙 HCI 日志
# 前提：已通过 hcidump 或 Ellisys 抓取蓝牙 HCI 数据
btmon -r ivi_bluetooth_hci.snoop > bluetooth_analysis.log 2>/dev/null

# 提取蓝牙连接事件和配对信息
grep -E "(Link_Key|PIN_Code|Connection_Complete|Authentication)" bluetooth_analysis.log | head -50

# 分析 IVI Wi-Fi 接入点日志
tshark -r ivi_wifi_capture.pcap -Y "wlan.fc.type_subtype == 0x00 || wlan.fc.type_subtype == 0x0b" \
  -T fields -e frame.time -e wlan.sa -e wlan.bssid -e wlan_mgt.ssid \
  -e wlan_mgt.rsn.akm -e wlan.rssi > ivi_wifi_assoc.csv

# 检测 Evil Twin 攻击（同一 SSID 多个 BSSID）
awk -F'\t' 'NR>1 {print $4"|"$3}' ivi_wifi_assoc.csv | sort | uniq -c | sort -rn | head -10

# 分析 IVI Android 系统日志（ADB 连接）
adb shell dumpsys bluetooth_manager > bt_state.txt
adb logcat -d -b events | grep -iE "bluetooth|wifi|usb|nfc" > ivi_connectivity.log

# 分析 T-BOX 通信流量
# 提取 T-BOX 与 TSP 服务器之间的 MQTT/HTTP 流量
tshark -r tbox_traffic.pcap -Y "tcp.port == 8883 || tcp.port == 443" \
  -T fields -e frame.time -e ip.src -e ip.dst -e tcp.port \
  -e tls.handshake.extensions_server_name \
  -e mqtt.topic -e mqtt.msg > tbox_comms.csv

# 解析 T-BOX 固件中的安全配置
strings tbox_firmware.bin | grep -iE "api_key|secret|token|password|endpoint" | head -30

# 提取 T-BOX 中的 AT 指令序列（4G/5G 通信）
strings tbox_firmware.bin | grep -E "^AT[+|]" | sort -u > tbox_at_commands.txt

# 自动驾驶安全事件时间线重建
python3 << 'PYEOF'
import json
import csv
from datetime import datetime, timedelta

def build_attack_timeline(event_sources):
    timeline = []
    
    for source_name, source_data in event_sources.items():
        for entry in source_data:
            timeline.append({
                'timestamp': entry['timestamp'],
                'source': source_name,
                'event_type': entry.get('type', 'unknown'),
                'severity': entry.get('severity', 'info'),
                'detail': entry.get('detail', '')
            })
    
    timeline.sort(key=lambda x: x['timestamp'])
    
    print("=" * 90)
    print(f"{'时间':<25} {'来源':<20} {'事件类型':<20} {'严重性':<5} {'详情'}")
    print("=" * 90)
    
    prev_ts = None
    for event in timeline:
        ts = datetime.fromtimestamp(event['timestamp']).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
        delta = ""
        if prev_ts:
            diff = event['timestamp'] - prev_ts
            if diff > 0.1:
                delta = f" (+{diff*1000:.0f}ms)"
        print(f"{ts}{delta:<10} {event['source']:<20} {event['event_type']:<20} {event['severity']:<5} {event['detail']}")
        prev_ts = event['timestamp']
    
    return timeline

sources = {
    'CAN': [{'timestamp': 1721431200.000, 'type': 'msg_frequency', 'severity': 'info', 'detail': 'CAN总线正常运行,负载率42%'},
            {'timestamp': 1721431200.500, 'type': 'new_msg_id', 'severity': 'high', 'detail': '检测到未注册消息ID 0x7FF'},
            {'timestamp': 1721431201.000, 'type': 'injection', 'severity': 'critical', 'detail': '0x0A0转向指令跳变: 0.0→0.85rad'}],
    'LiDAR': [{'timestamp': 1721431200.300, 'type': 'point_cloud_anomaly', 'severity': 'medium', 'detail': '前方区域点云密度异常增加300%'},
              {'timestamp': 1721431201.200, 'type': 'ghost_object', 'severity': 'high', 'detail': '检测到疑似虚假障碍物 @ 80m'}],
    'V2X':  [{'timestamp': 1721431199.800, 'type': 'rssi_anomaly', 'severity': 'medium', 'detail': 'RSU#47信号强度异常波动 ±20dB'},
             {'timestamp': 1721431200.800, 'type': 'replay_detected', 'severity': 'high', 'detail': 'BSM消息时间戳回退检测'}],
    'Decision': [{'timestamp': 1721431201.500, 'type': 'emergency_brake', 'severity': 'critical', 'detail': '紧急制动触发,原因: 检测到前方障碍物'}]
}

build_attack_timeline(sources)
PYEOF

# 多源证据关联分析脚本
python3 << 'PYEOF'
import json
from collections import defaultdict

def cross_source_correlation(sensor_log, can_log, decision_log, time_window_ms=100):
    with open(sensor_log) as f:
        sensor_events = [json.loads(l) for l in f]
    with open(can_log) as f:
        can_events = [json.loads(l) for l in f]
    with open(decision_log) as f:
        decision_events = [json.loads(l) for l in f]
    
    all_events = []
    for e in sensor_events:
        all_events.append({**e, 'category': 'sensor'})
    for e in can_events:
        all_events.append({**e, 'category': 'can'})
    for e in decision_events:
        all_events.append({**e, 'category': 'decision'})
    
    all_events.sort(key=lambda x: x['timestamp'])
    
    attack_chains = []
    current_chain = []
    
    for i, event in enumerate(all_events):
        if event.get('severity') in ['high', 'critical']:
            chain = [event]
            for j in range(i+1, min(i+20, len(all_events))):
                if all_events[j]['timestamp'] - event['timestamp'] < time_window_ms / 1000:
                    if all_events[j]['category'] != event['category']:
                        chain.append(all_events[j])
            if len(chain) >= 2:
                attack_chains.append(chain)
    
    print(f"总事件数: {len(all_events)}")
    print(f"检测到的潜在攻击链: {len(attack_chains)}")
    for idx, chain in enumerate(attack_chains[:5]):
        print(f"\n攻击链 #{idx+1} ({len(chain)} 个关联事件):")
        for event in chain:
            ts = event['timestamp']
            cat = event['category']
            etype = event.get('type', 'unknown')
            detail = event.get('detail', '')
            print(f"  [{cat}] {etype}: {detail}")

cross_source_correlation('sensor_events.jsonl', 'can_events.jsonl', 'decision_events.jsonl')
PYEOF

title: 自动驾驶系统 - CAN总线未注册消息ID注入检测
id: ad-can-injection-unreg-msg-id
status: production
description: 检测CAN总线中出现的未注册消息ID，可能指示CAN注入攻击
references:
  - https://www.autosec.io
  - https://attack.mitre.org/techniques/T1565/
author: x7peeps-blue-team
date: 2026/07/20
tags:
  - attack.data_manipulation
  - attack.t1565
  - automotive
  - autonomous_driving
logsource:
  product: automotive
  service: can_bus
detection:
  selection_msg_id:
    can_id|re: '(0x7[fF]{2}|0x[0-9a-fA-F]{3})'
  filter_known:
    can_id:
      - '0x0A0'
      - '0x0A1'
      - '0x0A2'
      - '0x0A3'
      - '0x120'
      - '0x121'
      - '0x200'
      - '0x201'
      - '0x300'
  condition: selection_msg_id and not filter_known
  timeframe: 5s
  level: high
falsepositives:
  - 新ECU上线后的首次通信
  - 诊断请求（需排除UDS端口）
fields:
  - can_id
  - timestamp
  - data
  - source_node

---
title: 自动驾驶系统 - LiDAR点云异常密度检测
id: ad-lidar-anomaly-density
status: production
description: 检测LiDAR点云中局部区域密度异常增加，可能指示虚假目标注入攻击
references:
  - https://www.usenix.org/conference/usenixsecurity22/presentation-shamir
author: x7peeps-blue-team
date: 2026/07/20
tags:
  - attack.data_manipulation
  - attack.t1565
  - automotive
  - lidar_security
logsource:
  product: automotive
  service: lidar
detection:
  selection_density:
    point_cloud_density_deviation|gte: 250
  selection_range:
    target_distance_m|lte: 150
  condition: selection_density and selection_range
  timeframe: 10s
  level: critical
falsepositives:
  - 隧道入口处的反射异常
  - 大型金属表面的多径反射
fields:
  - point_cloud_density_deviation
  - target_distance_m
  - region_id
  - timestamp

---
title: 自动驾驶系统 - V2X BSM消息重放检测
id: ad-v2x-bsm-replay-detection
status: production
description: 检测V2X BSM消息中的时间戳回退或重复，可能指示重放攻击
references:
  - https://attack.mitre.org/techniques/T1557/
author: x7peeps-blue-team
date: 2026/07/20
tags:
  - attack.interception
  - attack.t1557
  - automotive
  - v2x_security
logsource:
  product: automotive
  service: v2x
detection:
  selection_replay:
    msg_type: 'BSM'
    timestamp_regression|gt: 0
  selection_duplicate:
    msg_type: 'BSM'
    duplicate_count|gt: 1
  condition: selection_replay or selection_duplicate
  timeframe: 30s
  level: high
falsepositives:
  - GPS时间校准导致的短暂回退
fields:
  - vehicle_id
  - msg_type
  - timestamp_regression
  - duplicate_count
  - source_rssi

#!/bin/bash
ADS_HUNT_DIR="${1:-/var/log/automotive}"
REPORT_FILE="ads_forensics_report_$(date +%Y%m%d_%H%M%S).html"

echo "<html><head><title>自动驾驶安全取证报告</title></head><body>" > "$REPORT_FILE"
echo "<h1>自动驾驶系统安全取证分析报告</h1>" >> "$REPORT_FILE"
echo "<p>生成时间: $(date '+%Y-%m-%d %H:%M:%S')</p>" >> "$REPORT_FILE"

echo "<h2>1. CAN总线异常检测</h2>" >> "$REPORT_FILE"
if [ -f "$ADS_HUNT_DIR/can_bus.log" ]; then
    echo "<pre>" >> "$REPORT_FILE"
    echo "=== CAN消息频率统计 ===" >> "$REPORT_FILE"
    awk '{print $2}' "$ADS_HUNT_DIR/can_bus.log" | sort | uniq -c | sort -rn | head -30 >> "$REPORT_FILE"
    echo "" >> "$REPORT_FILE"
    echo "=== 未注册消息ID（高频异常） ===" >> "$REPORT_FILE"
    awk '{print $2}' "$ADS_HUNT_DIR/can_bus.log" | sort | uniq -c | sort -rn | \
        awk '$1 > 1000 && $2 !~ /^(0x0A0|0x0A1|0x0A2|0x0A3|0x120|0x121|0x200|0x201|0x300)$/ {print}' >> "$REPORT_FILE"
    echo "</pre>" >> "$REPORT_FILE"
fi

echo "<h2>2. 传感器一致性检查</h2>" >> "$REPORT_FILE"
if [ -f "$ADS_HUNT_DIR/sensor_sync.log" ]; then
    echo "<pre>" >> "$REPORT_FILE"
    echo "=== 传感器时间戳同步偏差 ===" >> "$REPORT_FILE"
    awk -F'\t' 'NR>1 {deviation=$4-$3; if(deviation>5) print "警告: 帧"$1" Camera-LiDAR偏差="deviation"ms"}' \
        "$ADS_HUNT_DIR/sensor_sync.log" | head -20 >> "$REPORT_FILE"
    echo "</pre>" >> "$REPORT_FILE"
fi

echo "<h2>3. V2X通信安全检查</h2>" >> "$REPORT_FILE"
if [ -f "$ADS_HUNT_DIR/v2x_traffic.log" ]; then
    echo "<pre>" >> "$REPORT_FILE"
    echo "=== V2X消息异常RSSI检测 ===" >> "$REPORT_FILE"
    awk -F'\t' 'NR>1 && ($5 < -90 || $5 > -20) {print "异常RSSI: 车辆"$1" 信号="$5"dBm 时间="$2}' \
        "$ADS_HUNT_DIR/v2x_traffic.log" | head -20 >> "$REPORT_FILE"
    echo "" >> "$REPORT_FILE"
    echo "=== V2X BSM消息频率分析 ===" >> "$REPORT_FILE"
    awk -F'\t' 'NR>1 {count[$1]++} END {for(v in count) if(count[v]>100) print "高频车辆: "v" 消息数="count[v]}' \
        "$ADS_HUNT_DIR/v2x_traffic.log" >> "$REPORT_FILE"
    echo "</pre>" >> "$REPORT_FILE"
fi

echo "<h2>4. OTA更新完整性检查</h2>" >> "$REPORT_FILE"
if [ -d "$ADS_HUNT_DIR/ota_updates" ]; then
    echo "<pre>" >> "$REPORT_FILE"
    for ota_pkg in "$ADS_HUNT_DIR/ota_updates"/*.bin; do
        if [ -f "$ota_pkg" ]; then
            echo "文件: $(basename $ota_pkg)" >> "$REPORT_FILE"
            echo "  SHA256: $(sha256sum $ota_pkg | awk '{print $1}')" >> "$REPORT_FILE"
            if [ -f "${ota_pkg}.sig" ]; then
                openssl dgst -sha256 -verify "$ADS_HUNT_DIR/ota_public_key.pem" \
                    -signature "${ota_pkg}.sig" "$ota_pkg" >> "$REPORT_FILE" 2>&1
            else
                echo "  签名文件: 未找到!" >> "$REPORT_FILE"
            fi
            echo "" >> "$REPORT_FILE"
        fi
    done
    echo "</pre>" >> "$REPORT_FILE"
fi

echo "<h2>5. IVI系统连接审计</h2>" >> "$REPORT_FILE"
if [ -f "$ADS_HUNT_DIR/ivi_connectivity.log" ]; then
    echo "<pre>" >> "$REPORT_FILE"
    echo "=== 蓝牙连接事件 ===" >> "$REPORT_FILE"
    grep -iE "bluetooth.*connect|pair|bond" "$ADS_HUNT_DIR/ivi_connectivity.log" | tail -30 >> "$REPORT_FILE"
    echo "" >> "$REPORT_FILE"
    echo "=== Wi-Fi关联事件 ===" >> "$REPORT_FILE"
    grep -iE "wifi.*assoc|connect|disconnect" "$ADS_HUNT_DIR/ivi_connectivity.log" | tail -30 >> "$REPORT_FILE"
    echo "</pre>" >> "$REPORT_FILE"
fi

echo "</body></html>" >> "$REPORT_FILE"
echo "取证报告已生成: $REPORT_FILE"

#!/usr/bin/env python3
import json
import sys
import statistics
from collections import defaultdict, Counter
from datetime import datetime

class AutonomousDrivingForensics:
    def __init__(self, can_log, sensor_log, v2x_log):
        self.can_log = can_log
        self.sensor_log = sensor_log
        self.v2x_log = v2x_log
        self.findings = []
    
    def detect_can_injection(self):
        msg_counts = Counter()
        msg_intervals = defaultdict(list)
        prev_ts = {}
        
        with open(self.can_log) as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) < 3:
                    continue
                ts = float(parts[0].strip('()'))
                msg_id = parts[1]
                msg_counts[msg_id] += 1
                if msg_id in prev_ts:
                    msg_intervals[msg_id].append(ts - prev_ts[msg_id])
                prev_ts[msg_id] = ts
        
        known_ids = {'0x0A0', '0x0A1', '0x0A2', '0x0A3', '0x120', '0x121', '0x200', '0x201', '0x300'}
        for msg_id, count in msg_counts.items():
            if msg_id not in known_ids and count > 10:
                self.findings.append({
                    'type': 'CAN_INJECTION_SUSPECT',
                    'severity': 'HIGH',
                    'detail': f'未注册消息ID {msg_id} 出现 {count} 次',
                    'evidence': '🟡'
                })
        
        for msg_id, intervals in msg_intervals.items():
            if len(intervals) > 10:
                avg = statistics.mean(intervals)
                if avg > 0:
                    cv = statistics.stdev(intervals) / avg if len(intervals) > 1 else 0
                    if cv > 0.5:
                        self.findings.append({
                            'type': 'CAN_TIMING_ANOMALY',
                            'severity': 'MEDIUM',
                            'detail': f'消息 {msg_id} 时间间隔变异系数 {cv:.3f} (正常<0.1)',
                            'evidence': '🟡'
                        })
    
    def detect_lidar_spoofing(self):
        try:
            with open(self.sensor_log) as f:
                sensor_data = [json.loads(line) for line in f]
        except FileNotFoundError:
            return
        
        densities = [d.get('point_density', 0) for d in sensor_data if 'point_density' in d]
        if not densities:
            return
        
        avg_density = statistics.mean(densities)
        std_density = statistics.stdev(densities) if len(densities) > 1 else 0
        
        for entry in sensor_data:
            pd = entry.get('point_density', 0)
            if std_density > 0 and (pd - avg_density) / std_density > 3:
                self.findings.append({
                    'type': 'LIDAR_SPOOFING_SUSPECT',
                    'severity': 'CRITICAL',
                    'detail': f'LiDAR点云密度异常: {pd:.0f} (均值 {avg_density:.0f}, 偏差 {(pd-avg_density)/std_density:.1f}σ)',
                    'evidence': '🔴'
                })
    
    def detect_v2x_anomalies(self):
        try:
            with open(self.v2x_log) as f:
                v2x_data = [json.loads(line) for line in f]
        except FileNotFoundError:
            return
        
        vehicle_msgs = defaultdict(list)
        for entry in v2x_data:
            vid = entry.get('vehicle_id', 'unknown')
            vehicle_msgs[vid].append(entry)
        
        for vid, msgs in vehicle_msgs.items():
            if len(msgs) < 5:
                continue
            rssis = [m.get('rssi', -50) for m in msgs]
            if len(rssis) > 1:
                rssi_std = statistics.stdev(rssis)
                if rssi_std > 15:
                    self.findings.append({
                        'type': 'V2X_RSSI_ANOMALY',
                        'severity': 'MEDIUM',
                        'detail': f'车辆 {vid} RSSI波动异常: σ={rssi_std:.1f}dB',
                        'evidence': '🟡'
                    })
            
            timestamps = [m.get('timestamp', 0) for m in msgs]
            for i in range(1, len(timestamps)):
                if timestamps[i] < timestamps[i-1]:
                    self.findings.append({
                        'type': 'V2X_REPLAY_ATTACK',
                        'severity': 'HIGH',
                        'detail': f'车辆 {vid} BSM时间戳回退: {timestamps[i-1]:.3f} → {timestamps[i]:.3f}',
                        'evidence': '🔴'
                    })
                    break
    
    def generate_report(self):
        self.detect_can_injection()
        self.detect_lidar_spoofing()
        self.detect_v2x_anomalies()
        
        severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
        self.findings.sort(key=lambda x: severity_order.get(x['severity'], 99))
        
        print("=" * 80)
        print("  自动驾驶系统安全取证分析报告")
        print(f"  生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 80)
        
        critical = sum(1 for f in self.findings if f['severity'] == 'CRITICAL')
        high = sum(1 for f in self.findings if f['severity'] == 'HIGH')
        medium = sum(1 for f in self.findings if f['severity'] == 'MEDIUM')
        
        print(f"\n  发现总数: {len(self.findings)}")
        print(f"  🔴 CRITICAL: {critical}  🟡 HIGH: {high}  ⚠️ MEDIUM: {medium}")
        print("-" * 80)
        
        for i, finding in enumerate(self.findings, 1):
            print(f"\n  [{finding['evidence']}] #{i} {finding['type']}")
            print(f"      严重性: {finding['severity']}")
            print(f"      详情: {finding['detail']}")

if __name__ == '__main__':
    can_log = sys.argv[1] if len(sys.argv) > 1 else 'can_bus.log'
    sensor_log = sys.argv[2] if len(sys.argv) > 2 else 'sensor_data.jsonl'
    v2x_log = sys.argv[3] if len(sys.argv) > 3 else 'v2x_data.jsonl'
    
    forensics = AutonomousDrivingForensics(can_log, sensor_log, v2x_log)
    forensics.generate_report()

rule Autonomous_Driving_Firmware_Tamper {
    meta:
        description = "检测自动驾驶ECU固件中的已知后门特征"
        author = "x7peeps-blue-team"
        date = "2026-07-20"
        reference = "https://attack.mitre.org/techniques/T1195/"
        severity = "critical"
    
    strings:
        $debug_shell = "busybox sh" ascii
        $debug_shell2 = "/bin/sh -i" ascii
        $backdoor_port = { 00 50 56 } ascii
        $hardcoded_key = { 48 65 6C 6C 6F 53 65 63 72 65 74 4B 65 79 }
        $reverse_shell = "nc -e /bin/sh" ascii
        $reverse_shell2 = "bash -i >& /dev/tcp" ascii
        $ssh_tunnel = "ssh -R" ascii
        $can_inject_lib = "libsocketcan" ascii nocase
        $suspicious_ota_endpoint = "http://" ascii
    
    condition:
        uint32(0) == 0x464C457F and 3 of them
}

rule V2X_Spoofing_Toolkit {
    meta:
        description = "检测V2X通信欺骗工具特征"
        author = "x7peeps-blue-team"
        date = "2026-07-20"
        severity = "high"
    
    strings:
        $sdr_init = "uhd_find_devices" ascii
        $bsm_gen = "generate_bsm" ascii
        $fake_rsu = "fake_rsu" ascii nocase
        $spoof_v2x = "spoof_vehicle" ascii nocase
        $dsrc_tx = "transmit_dsrc" ascii
        $c-v2x_pcap = "pc5_interface" ascii
    
    condition:
        2 of them
}

rule CAN_Bus_Attack_Tool {
    meta:
        description = "检测CAN总线攻击工具特征"
        author = "x7peeps-blue-team"
        date = "2026-07-20"
        severity = "high"
    
    strings:
        $cantools = "cansend" ascii
        $caninject = "caninject" ascii nocase
        $cankiller = "cankiller" ascii nocase
        $socketcan_setup = "ip link set can" ascii
        $can_jammer = { 43 41 4E 2D 4A 41 4D 4D 45 52 }
        $uds_reprogram = { 34 00 }
        $uds_security_bypass = { 27 03 }
    
    condition:
        (2 of ($cantools, $caninject, $cankiller)) or $can_jammer or
        ($uds_reprogram and $uds_security_bypass)
}

# 网络层 IOC
192.168.100.55          # IVI 异常外联 IP
192.168.200.10:5000     # SOME/IP 调试服务端点
aa:bb:cc:dd:ee:f0       # 异常蓝牙 MAC 地址

# 主机层 IOC
/tmp/.debug_module.so   # 恶意调试模块文件路径
/lib/sensor_override.so # 感知篡改动态库
/var/log/.uds_trace     # 隐藏的 UDS 跟踪日志

# 固件层 IOC
SHA256: e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  # 篡改后固件哈希

# 事后取证: 分析行车记录仪视频中的对抗样本攻击痕迹
python3 << 'PYEOF'
import cv2
import numpy as np

def detect_adversarial_patches(video_path, roi_regions=None):
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    suspicious_frames = []
    
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        ts = cap.get(cv2.CAP_PROP_POS_MSEC) / 1000.0
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        
        edges = cv2.Canny(gray, 50, 150)
        contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        
        for contour in contours:
            area = cv2.contourArea(contour)
            if 100 < area < 5000:
                x, y, w, h = cv2.boundingRect(contour)
                roi = gray[y:y+h, x:x+w]
                if roi.size > 0:
                    local_std = np.std(roi)
                    local_mean = np.mean(roi)
                    if local_std > 80 and local_mean > 200:
                        suspicious_frames.append({
                            'timestamp': ts,
                            'position': (x, y, w, h),
                            'std_dev': local_std,
                            'brightness': local_mean
                        })
                        cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 2)
                        cv2.putText(frame, f'PATCH? {local_std:.0f}', (x, y-5),
                                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1)
    
    cap.release()
    print(f"分析完成: 检测到 {len(suspicious_frames)} 个可疑对抗贴纸区域")
    for sf in suspicious_frames[:20]:
        print(f"  帧时间 {sf['timestamp']:.2f}s: 位置={sf['position']}, 亮度σ={sf['std_dev']:.1f}")
    return suspicious_frames

detect_adversarial_patches("dashcam_evidence.mp4")
PYEOF

# 物理层 IOC
对抗贴纸特征: 高对比度黑白图案，面积 3-10cm²
常见图案类型: 渐变条纹、棋盘格变体、数字扰动图案
投影设备特征: 亮度 > 5000lm, 投影角度 15-45°

# 数字层 IOC
感知输出异常: 目标检测置信度在特定 ROI 区域持续为零
融合层不一致: Camera 检测结果与 LiDAR 点云在 10m 范围内出现矛盾