# Clone the repository
git clone https://github.com/zhu-zhu666/S800-Vehicle-Network-Security-Testing-Framework.git
cd S800-Vehicle-Network-Security-Testing-Framework

# Install Python dependencies
pip install -r requirements.txt

# Install system dependencies (Ubuntu/Debian)
sudo apt-get update
sudo apt-get install can-utils python3-can

# Load SocketCAN kernel modules
sudo modprobe can
sudo modprobe can_raw
sudo modprobe vcan

# Setup virtual CAN interface (for testing without hardware)
sudo ip link add dev vcan0 type vcan
sudo ip link set up vcan0

# Setup physical CAN interface (example with slcan)
sudo slcand -o -c -s6 /dev/ttyUSB0 can0
sudo ip link set up can0

# Set CAN bitrate (common automotive: 500kbps)
sudo ip link set can0 type can bitrate 500000

from s800.can_analyzer import CANAnalyzer
from s800.utils import setup_interface

# Initialize analyzer
analyzer = CANAnalyzer(interface='can0')

# Start capturing traffic
analyzer.start_capture(duration=60)  # Capture for 60 seconds

# Analyze captured frames
stats = analyzer.get_statistics()
print(f"Total frames: {stats['total_frames']}")
print(f"Unique IDs: {stats['unique_ids']}")
print(f"Average data rate: {stats['avg_rate']} frames/sec")

# Export captured data
analyzer.export_pcap('capture.pcap')
analyzer.export_csv('capture.csv')

from s800.fuzzer import CANFuzzer
from s800.payloads import PayloadGenerator

# Initialize fuzzer
fuzzer = CANFuzzer(interface='can0')

# Define target CAN ID range
target_ids = range(0x100, 0x200)

# Generate mutation-based payloads
payload_gen = PayloadGenerator()
payloads = payload_gen.generate_mutations(
    base_data=[0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07],
    mutation_rate=0.3,
    count=1000
)

# Run fuzzing campaign
fuzzer.fuzz(
    can_ids=target_ids,
    payloads=payloads,
    delay=0.01,  # 10ms between frames
    monitor=True,  # Monitor for anomalies
    callback=lambda result: print(f"Sent: {result}")
)

from s800.ecu_scanner import ECUScanner
from s800.diagnostics import UDSClient

# Scan for active ECUs
scanner = ECUScanner(interface='can0')
ecus = scanner.scan_range(0x700, 0x7FF)

print(f"Found {len(ecus)} ECUs:")
for ecu in ecus:
    print(f"  ID: 0x{ecu['id']:03X}, Type: {ecu['type']}")

# Query ECU information via UDS (ISO 14229)
uds = UDSClient(interface='can0', ecu_id=0x7E0)

# Read DID (Data Identifier)
vin = uds.read_data_by_id(0xF190)  # VIN
print(f"VIN: {vin.decode()}")

# Read diagnostic trouble codes
dtcs = uds.read_dtc()
print(f"Diagnostic Trouble Codes: {dtcs}")

from s800.replay import CANReplay
import time

# Record session
recorder = CANReplay(interface='can0')
print("Recording traffic... Press Ctrl+C to stop")

try:
    recorder.start_recording()
    time.sleep(30)  # Record for 30 seconds
except KeyboardInterrupt:
    pass

recorder.stop_recording()
recorder.save_session('door_unlock_sequence.s800')

# Replay recorded session
replayer = CANReplay(interface='can0')
replayer.load_session('door_unlock_sequence.s800')

# Replay with original timing
replayer.replay(preserve_timing=True)

# Replay at faster speed
replayer.replay(speed_multiplier=2.0)

# Replay single frame repeatedly
replayer.replay_frame(index=15, count=100, interval=0.05)

from s800.mitm import CANBridge
from s800.filters import MessageFilter

# Create bridge between two CAN interfaces
bridge = CANBridge(interface_a='can0', interface_b='can1')

# Define filtering rules
def modify_speed(msg):
    """Modify speed data (example: reduce displayed speed)"""
    if msg.arbitration_id == 0x320:  # Speed message ID
        # Modify byte 3-4 (speed value)
        speed = int.from_bytes(msg.data[2:4], byteorder='big')
        new_speed = int(speed * 0.8)  # Reduce by 20%
        msg.data[2:4] = new_speed.to_bytes(2, byteorder='big')
    return msg

# Add filter
bridge.add_filter(modify_speed)

# Block specific messages
bridge.block_id(0x400)  # Block messages with ID 0x400

# Start bridging
bridge.start()

interfaces:
  primary: can0
  secondary: can1
  virtual: vcan0

capture:
  buffer_size: 10000
  auto_save: true
  output_dir: ./captures/

fuzzer:
  default_delay: 0.01
  max_iterations: 10000
  crash_detection: true
  anomaly_threshold: 5.0

scanner:
  timeout: 1.0
  retry_count: 3
  id_range:
    start: 0x000
    end: 0x7FF

uds:
  default_timeout: 2.0
  session_type: extended  # default, extended, programming
  security_access: false

logging:
  level: INFO
  file: s800.log
  console: true

from s800.config import Config

config = Config.load('config.yaml')
analyzer = CANAnalyzer(
    interface=config.interfaces['primary'],
    buffer_size=config.capture['buffer_size']
)

from s800.protocols import ProtocolDecoder

class CustomProtocolDecoder(ProtocolDecoder):
    """Decode proprietary protocol"""
    
    def decode(self, msg):
        if msg.arbitration_id == 0x300:
            return {
                'type': 'sensor_data',
                'temperature': msg.data[0] - 40,  # Offset by 40
                'pressure': int.from_bytes(msg.data[1:3], 'big') / 10,
                'status': msg.data[3]
            }
        return None

# Use custom decoder
analyzer = CANAnalyzer(interface='can0')
analyzer.add_decoder(CustomProtocolDecoder())
analyzer.start_capture()

from s800.ml import AnomalyDetector

# Train baseline model
detector = AnomalyDetector()
detector.train_from_pcap('normal_traffic.pcap')

# Real-time anomaly detection
analyzer = CANAnalyzer(interface='can0')

def check_anomaly(msg):
    if detector.is_anomaly(msg):
        print(f"ANOMALY DETECTED: ID=0x{msg.arbitration_id:03X}")
        print(f"  Data: {msg.data.hex()}")
        # Take action (log, alert, block, etc.)

analyzer.add_callback(check_anomaly)
analyzer.start_capture()

from s800.assessment import SecurityAssessment

# Run comprehensive security assessment
assessment = SecurityAssessment(interface='can0')

results = assessment.run_all_tests(
    tests=[
        'ecu_discovery',
        'uds_services',
        'authentication_bypass',
        'replay_protection',
        'fuzzing_resilience'
    ],
    report_format='html'
)

assessment.save_report('security_report.html')
print(f"Assessment complete. Score: {results['security_score']}/100")

# Capture CAN traffic
python s800.py capture -i can0 -d 60 -o capture.pcap

# Scan for ECUs
python s800.py scan -i can0 --range 0x700-0x7FF

# Fuzz CAN IDs
python s800.py fuzz -i can0 --ids 0x100-0x200 --count 1000

# Replay traffic
python s800.py replay -i can0 -f capture.pcap --speed 1.0

# UDS diagnostics
python s800.py uds -i can0 --ecu 0x7E0 --read-vin

# Run security assessment
python s800.py assess -i can0 --output report.html

# MITM with filtering
python s800.py mitm -a can0 -b can1 --block 0x400 --modify speed_reducer.py

# Export analysis
python s800.py analyze -f capture.pcap --export csv --stats

# Differential analysis
python s800.py diff baseline.pcap test.pcap --threshold 0.05

# Default CAN interface
export S800_INTERFACE=can0

# Configuration file path
export S800_CONFIG=/etc/s800/config.yaml

# Output directory
export S800_OUTPUT_DIR=/var/log/s800/

# Debug mode
export S800_DEBUG=1

# API endpoint (if using remote analysis)
export S800_API_URL=https://analysis.example.com
export S800_API_KEY=your_api_key_here

# Always capture baseline before testing
baseline = CANAnalyzer(interface='can0')
baseline.start_capture(duration=300)  # 5 minutes
baseline.save('baseline.pcap')

# Run tests
# ...

# Compare with baseline
test_capture = CANAnalyzer(interface='can0')
test_capture.start_capture(duration=60)
diff = test_capture.compare_with('baseline.pcap')
print(f"New messages: {diff['new_ids']}")

# Monitor system health during fuzzing
from s800.monitoring import SystemMonitor

monitor = SystemMonitor(interface='can0')
fuzzer = CANFuzzer(interface='can0')

monitor.add_safety_check('ecu_response', timeout=5.0)
monitor.add_safety_check('error_frames', threshold=10)

fuzzer.set_safety_monitor(monitor)
fuzzer.fuzz(can_ids=[0x100], payloads=payloads, emergency_stop=True)

# Check interface status
ip link show can0

# Verify kernel modules
lsmod | grep can

# Check dmesg for errors
dmesg | grep can

# Add user to dialout group (for USB devices)
sudo usermod -a -G dialout $USER

# Run with sudo (temporary)
sudo python s800.py capture -i can0

# Verify bitrate matches vehicle
sudo ip link set can0 type can bitrate 500000

# Check for bus-off state
candump can0 -e  # Show error frames

# Test with cansend
cansend can0 123#DEADBEEF

# Limit capture rate
analyzer = CANAnalyzer(interface='can0', filter_ids=[0x100, 0x200])

# Use hardware filtering if available
analyzer.set_hardware_filter(mask=0x700, match=0x300)