智能网联汽车与车路云一体化安全取证深度分析

┌───────┬─────────────┬───┬─────┬─────┬───────┬───────┐
│ SOF   │ 11-bit ID   │RTR│ IDE │ r0  │ DLC   │ 0-8B  │ CRC  │ ACK │ EOF │ IFS │
│ 1-bit │ 11 bits     │1  │ 1   │ 1   │ 4 bits│  data │ 15+1 │ 2   │ 7   │ 3   │
└───────┴─────────────┴───┴─────┴─────┴───────┴───────┘

攻击准备阶段
├── 1. 物理接入（OBD-II / 车载设备 / 线束）
├── 2. CAN 接口初始化（socketcan / Kvaser / PEAK）
├── 3. 总线嗅探与帧识别（被动监听）
└── 4. CAN ID 映射表构建（逆向分析）
攻击实施阶段
├── 5. 目标 CAN ID 定位
├── 6. 数据字段构造与 CRC 计算
├── 7. CAN 帧注入（cansend）
└── 8. 效果验证与迭代优化

sudo ip link set can0 up type can bitrate 500000
candump can0 -t a -c > can_log_$(date +%Y%m%d_%H%M%S).raw &
sudo cansend can0 7E0#02010C0000000000
sudo cangen can0 -g 10 -I 42A -L 8 -D i

import can
from collections import Counter, defaultdict
import statistics
import sys

def analyze_can_traffic(log_file):
    frame_counts = defaultdict(list)
    frame_times = defaultdict(list)
    all_frames = []
    
    bus = can.interface.Bus(channel='can0', interface='socketcan')
    reader = can.ASCReader(log_file)
    
    prev_timestamp = {}
    for msg in reader:
        can_id = msg.arbitration_id
        frame_counts[can_id].append(1)
        frame_times[can_id].append(msg.timestamp)
        all_frames.append(msg)
        
        if can_id in prev_timestamp:
            delta = msg.timestamp - prev_timestamp[can_id]
            if delta < 0.001:
                print(f"[ANOMALY] Rapid frame: ID=0x{can_id:03X} delta={delta:.6f}s")
        prev_timestamp[can_id] = msg.timestamp
    
    print("\n--- CAN ID Frequency Analysis ---")
    for can_id in sorted(frame_counts.keys()):
        count = len(frame_counts[can_id])
        timestamps = frame_times[can_id]
        if len(timestamps) > 10:
            intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]
            avg_interval = statistics.mean(intervals)
            std_interval = statistics.stdev(intervals) if len(intervals) > 1 else 0
            cv = std_interval / avg_interval if avg_interval > 0 else 0
            if cv < 0.05 and count > 100:
                print(f"[SUSPECT] ID=0x{can_id:03X} count={count} avg_interval={avg_interval:.4f} CV={cv:.4f}")
    
    return frame_counts

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <can_log_file.asc>")
        sys.exit(1)
    analyze_can_traffic(sys.argv[1])

sudo tcpdump -i eth0 -w vehicle_ethernet_$(date +%Y%m%d_%H%M%S).pcap \
  'port 13400 or port 34973 or port 8080' -s 0
sudo tshark -r vehicle_ethernet.pcap -Y "doip" -T fields \
  -e doip.payload_type -e doip.vin -e doip.logical_address

BSM Message Structure (SAE J2735)
├── Part I (Mandatory)
│   ├── MessageID: 0x14 (BSM)
│   ├── DSecond: 时间戳
│   ├── Latitude: 纬度 (1/10 degree)
│   ├── Longitude: 经度 (1/10 degree)
│   ├── Elevation: 海拔
│   ├── Accuracy: 位置精度
│   ├── TransmissionState: 变速箱状态
│   ├── Speed: 车速 (0.02 m/s)
│   ├── Heading: 航向角 (0.0125 degree)
│   ├── SteeringWheelAngle: 方向盘角度
│   └── BrakeSystemStatus: 制动状态
└── Part II (Optional)
    ├── VehicleSafetyExtension
    ├── PathHistory
    ├── PathPrediction
    └── Regional

sudo hackrf_transfer -r v2x_capture.raw -f 5900000000 -s 8000000 -g 40
sudo rtl_sdr -f 5.9e9 -s 2e6 -g 40 -d 0 v2x_raw.bin
python3 decode_bsm.py v2x_capture.raw --format raw --output bsm_log.json

OTA Firmware Package
├── Header
│   ├── Magic Number (4 bytes)
│   ├── Package Version
│   ├── Target ECU ID
│   ├── Min Compatible Version
│   ├── Target Version
│   └── Creation Timestamp
├── Metadata
│   ├── Build Environment Info
│   ├── File Size
│   └── Checksum (SHA-256)
├── Encrypted Payload
│   ├── AES-256-GCM Encrypted Firmware
│   └── IV / Nonce
├── Digital Signature
│   ├── Algorithm (RSA-2048-PSS / ECDSA-P256)
│   ├── Signed Hash (SHA-256)
│   └── Certificate Chain
└── Post-install Script (optional)

Full Firmware v2.0 (100 MB)
    │
    ├── Binary Diff Algorithm (bsdiff/xdelta)
    │
Delta Package (5 MB) + Reference v1.0
    │
    └── Patch Application → Full Firmware v2.0

import numpy as np
from scipy.spatial import ConvexHull
import struct
import socket

class LidarSpoofDetector:
    def __init__(self, threshold=2.0, window_size=100):
        self.threshold = threshold
        self.window_size = window_size
        self.point_history = []
        self.velocity_history = []
    
    def analyze_point_cloud(self, points):
        if len(points) < 10:
            return []
        
        anomalies = []
        centroid = np.mean(points[:, :3], axis=0)
        distances = np.linalg.norm(points[:, :3] - centroid, axis=1)
        mean_dist = np.mean(distances)
        std_dist = np.std(distances)
        
        outlier_mask = distances > mean_dist + self.threshold * std_dist
        outlier_ratio = np.sum(outlier_mask) / len(points)
        
        if outlier_ratio > 0.3:
            anomalies.append({
                'type': 'density_anomaly',
                'severity': 'high',
                'ratio': outlier_ratio
            })
        
        if len(self.point_history) >= 2:
            prev_centroid = np.mean(self.point_history[-1][:, :3], axis=0)
            curr_centroid = centroid
            velocity = np.linalg.norm(curr_centroid - prev_centroid)
            self.velocity_history.append(velocity)
            
            if len(self.velocity_history) > 10:
                vel_mean = np.mean(self.velocity_history[-10:])
                vel_std = np.std(self.velocity_history[-10:])
                if vel_std > 0 and abs(velocity - vel_mean) > 3 * vel_std:
                    anomalies.append({
                        'type': 'velocity_jump',
                        'severity': 'critical',
                        'velocity': velocity
                    })
        
        if len(points) > 50:
            angles = np.arctan2(points[:, 1], points[:, 0])
            angle_bins = np.histogram(angles, bins=36)[0]
            zero_bins = np.sum(angle_bins == 0)
            if zero_bins > 10:
                anomalies.append({
                    'type': 'angular_gap',
                    'severity': 'medium',
                    'gaps': int(zero_bins)
                })
        
        self.point_history.append(points)
        if len(self.point_history) > self.window_size:
            self.point_history.pop(0)
        
        return anomalies

adb shell ls /data/system/packages.xml
adb shell dumpsys bluetooth_manager
adb pull /sdcard/Android/data/com.android.bluetooth/files/
adb logcat -d | grep -i "can\|vehicle\|telematics"
adb shell pm list packages -f | grep -i "car\|auto\|drive"

import paho.mqtt.client as mqtt
import json
import ssl
from datetime import datetime

class VehicleMQTTForensics:
    def __init__(self, broker, port, ca_cert, client_cert, client_key):
        self.broker = broker
        self.port = port
        self.messages = []
        self.anomalies = []
        self.client = mqtt.Client(client_id="forensics_collector")
        self.client.tls_set(
            ca_certs=ca_cert,
            certfile=client_cert,
            keyfile=client_key,
            tls_version=ssl.PROTOCOL_TLSv1_2
        )
        self.client.on_message = self.on_message
    
    def on_message(self, client, userdata, msg):
        timestamp = datetime.utcnow().isoformat()
        try:
            payload = json.loads(msg.payload.decode())
        except (json.JSONDecodeError, UnicodeDecodeError):
            payload = msg.payload.hex()
        
        entry = {
            'timestamp': timestamp,
            'topic': msg.topic,
            'qos': msg.qos,
            'payload': payload,
            'retain': msg.retain
        }
        self.messages.append(entry)
        
        self.check_anomalies(entry)
    
    def check_anomalies(self, entry):
        topic = entry['topic']
        payload = entry['payload']
        
        if isinstance(payload, dict):
            if 'vin' in payload and 'command' in payload:
                if payload['command'] in ['unlock', 'start', 'trunk']:
                    self.anomalies.append({
                        'type': 'remote_control',
                        'severity': 'high',
                        'detail': entry
                    })
            
            if 'latitude' in payload and 'longitude' in payload:
                lat = payload['latitude']
                lon = payload['longitude']
                if abs(lat) > 90 or abs(lon) > 180:
                    self.anomalies.append({
                        'type': 'invalid_coordinates',
                        'severity': 'medium',
                        'detail': entry
                    })
    
    def analyze_topics(self):
        topic_stats = {}
        for msg in self.messages:
            topic = msg['topic']
            if topic not in topic_stats:
                topic_stats[topic] = {
                    'count': 0,
                    'first_seen': msg['timestamp'],
                    'last_seen': msg['timestamp']
                }
            topic_stats[topic]['count'] += 1
            topic_stats[topic]['last_seen'] = msg['timestamp']
        
        return topic_stats
    
    def export_evidence(self, output_file):
        evidence = {
            'collection_time': datetime.utcnow().isoformat(),
            'total_messages': len(self.messages),
            'anomalies': self.anomalies,
            'topic_stats': self.analyze_topics(),
            'messages': self.messages
        }
        with open(output_file, 'w') as f:
            json.dump(evidence, f, indent=2, ensure_ascii=False)

title: Suspicious CAN Bus Frame Injection Activity
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: stable
description: Detects CAN bus frames with characteristics indicative of injection attacks, including rapid frame bursts, unusual arbitration IDs, and abnormal data patterns
author: x7peeps Security Research
date: 2026/07/15
references:
  - https://doi.org/10.1109/TVT.2021.3059191
  - https://www.sae.org/standards/content/j2284/
tags:
  - attack.injection
  - attack.can_bus
  - car_security
logsource:
  product: automotive
  service: can_bus
detection:
  selection_rapid_burst:
    frame_rate_bpm:
      - 'gt: 500'
    can_id:
      - '0x000-0x01F'
  selection_diagnostic_abuse:
    can_id:
      - '0x7DF'
      - '0x7E0-0x7E7'
    data_byte_0:
      - '0x10'
      - '0x27'
      - '0x34'
      - '0x36'
      - '0x37'
  selection_unknown_id:
    can_id:
      - 'gt: 0x7FF'
  selection_frequency_anomaly:
    frame_interval_cv:
      - 'lt: 0.05'
    frame_count_per_second:
      - 'gt: 100'
  condition: selection_rapid_burst OR selection_diagnostic_abuse OR selection_unknown_id OR selection_frequency_anomaly
fields:
  - can_id
  - data_field
  - frame_rate_bpm
  - frame_interval_cv
  - source_ecu
falsepositives:
  - Legitimate high-frequency CAN traffic (e.g., powertrain control)
  - Authorized diagnostic sessions
level: high

title: Firmware Rollback or Integrity Tampering Detected
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
status: stable
description: Detects OTA firmware update anomalies including version rollback, signature verification failure, and hash mismatch events
author: x7peeps Security Research
date: 2026/07/15
references:
  - https://iso.org/standard/70918.html
  - https://doi.org/10.1109/ITSWC.2020.9494312
tags:
  - attack.ota
  - attack.firmware
  - car_security
logsource:
  product: automotive
  service: ota_manager
detection:
  selection_version_rollback:
    event_type:
      - 'ota_update_complete'
    installed_version:
      - 'lt: previous_version'
  selection_signature_fail:
    event_type:
      - 'ota_signature_verify'
    verify_result:
      - 'FAILED'
      - 'ERROR'
  selection_hash_mismatch:
    event_type:
      - 'ota_hash_verify'
    expected_hash:
      - 'neq: actual_hash'
  selection_unauthorized_source:
    event_type:
      - 'ota_download_start'
    source_ip:
      - 'neq: expected_ota_server_ip'
  condition: selection_version_rollback OR selection_signature_fail OR selection_hash_mismatch OR selection_unauthorized_source
fields:
  - ecu_id
  - previous_version
  - installed_version
  - verify_result
  - source_ip
  - timestamp
falsepositives:
  - Planned firmware rollback during service
  - Test environment OTA operations
level: critical

#!/bin/bash

CAN_LOG_FILE="${1:-can_traffic.asc}"
REPORT_FILE="can_forensics_report_$(date +%Y%m%d_%H%M%S).txt"
SUSPECT_THRESHOLD=500
FREQ_CV_THRESHOLD=0.05

if [ ! -f "$CAN_LOG_FILE" ]; then
    echo "Usage: $0 <can_log_file.asc>"
    exit 1
fi

echo "=== CAN Bus Forensics Analysis Report ===" > "$REPORT_FILE"
echo "Log File: $CAN_LOG_FILE" >> "$REPORT_FILE"
echo "Analysis Time: $(date -u '+%Y-%m-%d %H:%M:%S UTC')" >> "$REPORT_FILE"
echo "=========================================" >> "$REPORT_FILE"

TOTAL_FRAMES=$(wc -l < "$CAN_LOG_FILE")
echo "" >> "$REPORT_FILE"
echo "Total Frames: $TOTAL_FRAMES" >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- CAN ID Frequency Distribution ---" >> "$REPORT_FILE"
grep -oP 'ID:\s+0x([0-9A-Fa-f]+)' "$CAN_LOG_FILE" | \
    sort | uniq -c | sort -rn | head -30 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- High-Frequency CAN IDs (> ${SUSPECT_THRESHOLD} frames) ---" >> "$REPORT_FILE"
grep -oP 'ID:\s+0x([0-9A-Fa-f]+)' "$CAN_LOG_FILE" | \
    sort | uniq -c | sort -rn | \
    awk -v thresh="$SUSPECT_THRESHOLD" '$1 > thresh {print "[ALERT] ID "$2" count:"$1}' >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- UDS Diagnostic Activity Detection ---" >> "$REPORT_FILE"
UDS_COUNT=$(grep -cE 'ID:\s*0x(7DF|7E[0-7])' "$CAN_LOG_FILE")
echo "UDS Diagnostic Frames: $UDS_COUNT" >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- UDS Service Distribution ---" >> "$REPORT_FILE"
grep -E 'ID:\s*0x(7DF|7E[0-7])' "$CAN_LOG_FILE" | \
    grep -oP 'Data:\s+([0-9A-Fa-f]+)' | \
    cut -d' ' -f2 | cut -c1-2 | \
    sort | uniq -c | sort -rn >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- CAN Frame Size Anomalies ---" >> "$REPORT_FILE"
grep -E 'DLC:\s+[0-9]+' "$CAN_LOG_FILE" | \
    awk '/DLC: ([0-9]+)/{if($NF>8) print "[ALERT] Extended DLC: "$0}' | \
    head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- Burst Detection (IDs with >100 frames in 1s window) ---" >> "$REPORT_FILE"
awk '/^[0-9]+\.[0-9]+/{
    split($1, t, ".");
    ts = t[1] + t[2]/1000000;
    gsub(/.*ID:\s+0x/, "", $0);
    split($0, parts, " ");
    id = parts[1];
    if (!(id in first_ts)) {
        first_ts[id] = ts;
        burst_count[id] = 1;
    } else {
        if (ts - first_ts[id] <= 1.0) {
            burst_count[id]++;
        } else {
            if (burst_count[id] > 100) {
                printf "[ALERT] Burst: ID=0x%s count=%d in %.1fs window\n", id, burst_count[id], ts-first_ts[id] >> "'"$REPORT_FILE"'"
            }
            first_ts[id] = ts;
            burst_count[id] = 1;
        }
    }
}' "$CAN_LOG_FILE"

echo "" >> "$REPORT_FILE"
echo "--- Summary ---" >> "$REPORT_FILE"
echo "Report generated at: $(date -u '+%Y-%m-%d %H:%M:%S UTC')" >> "$REPORT_FILE"
echo "Analysis complete. Review $REPORT_FILE for findings." >> "$REPORT_FILE"

cat "$REPORT_FILE"

import struct
import json
import math
from datetime import datetime, timedelta
from collections import defaultdict

class V2XForensicsAnalyzer:
    def __init__(self):
        self.bsm_messages = []
        self.anomalies = []
        self.vehicle_tracks = defaultdict(list)
        self.certificate_stats = defaultdict(int)
        self.TIME_THRESHOLD_SEC = 5
        self.POSITION_JUMP_THRESHOLD_M = 200
        self.SPEED_LIMIT_KMH = 200
        self.DENSITY_THRESHOLD = 3.0
    
    def parse_bsm_hex(self, hex_data):
        raw = bytes.fromhex(hex_data)
        if len(raw) < 38:
            return None
        
        msg = {}
        msg['message_id'] = raw[0]
        msg['temp_id'] = struct.unpack('>I', raw[1:5])[0] & 0xFFFFFF
        msg['timestamp_sec'] = struct.unpack('>H', raw[5:7])[0]
        
        raw_lat = struct.unpack('>i', raw[7:11])[0]
        msg['latitude'] = raw_lat / 1e7
        
        raw_lon = struct.unpack('>i', raw[11:15])[0]
        msg['longitude'] = raw_lon / 1e7
        
        msg['elevation'] = struct.unpack('>h', raw[15:17])[0] * 0.5
        
        accuracy_raw = struct.unpack('>H', raw[17:19])[0]
        msg['accuracy SemiMinor'] = (accuracy_raw & 0x000F) * 0.05
        msg['accuracy SemiMajor'] = ((accuracy_raw >> 4) & 0x000F) * 0.05
        
        msg['speed_kmh'] = struct.unpack('>H', raw[20:22])[0] * 0.02 * 3.6
        msg['heading'] = struct.unpack('>H', raw[22:24])[0] * 0.0125
        
        msg['raw_hex'] = hex_data
        msg['parsed_at'] = datetime.utcnow().isoformat()
        
        return msg
    
    def validate_bsm(self, bsm):
        issues = []
        timestamp_diff = abs(
            (datetime.utcnow() - datetime.strptime(bsm['parsed_at'], '%Y-%m-%dT%H:%M:%S.%f')).total_seconds()
            - (datetime.utcnow().second - bsm['timestamp_sec'] % 60)
        )
        if timestamp_diff > self.TIME_THRESHOLD_SEC:
            issues.append({
                'type': 'timestamp_anomaly',
                'severity': 'high',
                'detail': f'Timestamp drift: {timestamp_diff:.1f}s'
            })
        
        if abs(bsm['latitude']) > 90 or abs(bsm['longitude']) > 180:
            issues.append({
                'type': 'invalid_coordinates',
                'severity': 'critical',
                'detail': f'Lat={bsm["latitude"]}, Lon={bsm["longitude"]}'
            })
        
        if bsm['speed_kmh'] > self.SPEED_LIMIT_KMH:
            issues.append({
                'type': 'speed_anomaly',
                'severity': 'medium',
                'detail': f'Speed: {bsm["speed_kmh"]:.1f} km/h'
            })
        
        return issues
    
    def check_position_consistency(self, bsm):
        temp_id = bsm['temp_id']
        track = self.vehicle_tracks[temp_id]
        
        if len(track) == 0:
            track.append(bsm)
            return []
        
        prev = track[-1]
        time_diff = (datetime.utcnow() - datetime.fromisoformat(prev['parsed_at'])).total_seconds()
        
        if time_diff <= 0 or time_diff > 10:
            track.append(bsm)
            return []
        
        lat_diff = (bsm['latitude'] - prev['latitude']) * 111320
        lon_diff = (bsm['longitude'] - prev['longitude']) * 111320 * math.cos(math.radians(bsm['latitude']))
        distance = math.sqrt(lat_diff**2 + lon_diff**2)
        
        max_distance = bsm['speed_kmh'] / 3.6 * time_diff * 2
        
        issues = []
        if distance > self.POSITION_JUMP_THRESHOLD_M:
            issues.append({
                'type': 'position_jump',
                'severity': 'critical',
                'detail': f'Distance: {distance:.1f}m in {time_diff:.2f}s'
            })
        elif max_distance > 0 and distance > max_distance:
            issues.append({
                'type': 'position_inconsistency',
                'severity': 'high',
                'detail': f'Distance {distance:.1f}m exceeds speed-based limit {max_distance:.1f}m'
            })
        
        track.append(bsm)
        if len(track) > 500:
            self.vehicle_tracks[temp_id] = track[-500:]
        
        return issues
    
    def analyze_density(self, window_sec=1):
        area_vehicles = defaultdict(int)
        now = datetime.utcnow()
        
        for temp_id, track in self.vehicle_tracks.items():
            if track:
                latest = track[-1]
                area_key = f"{round(latest['latitude'], 3)}_{round(latest['longitude'], 3)}"
                area_vehicles[area_key] += 1
        
        anomalies = []
        for area, count in area_vehicles.items():
            if count > 50:
                anomalies.append({
                    'type': 'density_anomaly',
                    'area': area,
                    'vehicle_count': count,
                    'severity': 'high'
                })
        
        return anomalies
    
    def generate_report(self):
        report = {
            'analysis_time': datetime.utcnow().isoformat(),
            'total_bsm_messages': len(self.bsm_messages),
            'tracked_vehicles': len(self.vehicle_tracks),
            'total_anomalies': len(self.anomalies),
            'anomalies': self.anomalies,
            'vehicle_summary': {
                temp_id: {
                    'message_count': len(track),
                    'latest_position': {
                        'lat': track[-1]['latitude'] if track else None,
                        'lon': track[-1]['longitude'] if track else None
                    } if track else None
                }
                for temp_id, track in self.vehicle_tracks.items()
            }
        }
        return report
    
    def process_log(self, log_file):
        with open(log_file, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                
                parts = line.split()
                if len(parts) < 3:
                    continue
                
                try:
                    hex_data = parts[2].replace('0x', '').replace(':', '')
                    bsm = self.parse_bsm_hex(hex_data)
                    if bsm is None:
                        continue
                    
                    self.bsm_messages.append(bsm)
                    
                    validation_issues = self.validate_bsm(bsm)
                    position_issues = self.check_position_consistency(bsm)
                    
                    all_issues = validation_issues + position_issues
                    
                    if all_issues:
                        for issue in all_issues:
                            issue['temp_id'] = bsm['temp_id']
                            issue['timestamp'] = bsm['parsed_at']
                            self.anomalies.append(issue)
                
                except (struct.error, ValueError, IndexError):
                    continue
        
        density_anomalies = self.analyze_density()
        self.anomalies.extend(density_anomalies)
        
        return self.generate_report()

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <v2x_log_file>")
        sys.exit(1)
    
    analyzer = V2XForensicsAnalyzer()
    report = analyzer.process_log(sys.argv[1])
    
    output_file = f"v2x_forensics_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
    with open(output_file, 'w') as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    
    print(f"Analysis complete: {report['total_bsm_messages']} messages, {report['total_anomalies']} anomalies")
    print(f"Report saved to: {output_file}")
    
    for anomaly in report['anomalies']:
        print(f"  [{anomaly['severity'].upper()}] {anomaly['type']}: {anomaly.get('detail', 'N/A')}")