# Automotive Zonal > Automotive Zonal expertise. Covers 5 topics: Automotive Ethernet, Network Security Zonal, Service Oriented Communication, Zonal Architecture Design, Zone Controller Development. - Skill: `pangzhenying2025/automotive-zonal` (Agent Skill) - Install (CLI): `npx skillmds@latest add pangzhenying2025/automotive-zonal` - Raw SKILL.md: https://api.skillmd.com/api/skills/pangzhenying2025/automotive-zonal/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Security - Author: pangzhenying2025 (https://skillmd.com/u/pangzhenying2025) - Updated: 2026-09-22 - Page: https://skillmd.com/skills/pangzhenying2025/automotive-zonal --- # Automotive Zonal ## Automotive Ethernet # Automotive Ethernet - TSN & AVB **Category:** automotive-zonal **Version:** 1.0.0 **Maturity:** production **Complexity:** advanced ## Overview Expert knowledge in automotive Ethernet technologies including Time-Sensitive Networking (TSN), Audio Video Bridging (AVB), physical layer standards (100BASE-T1, 1000BASE-T1), switch configuration, VLAN management, and Quality of Service (QoS) for deterministic vehicle networks. ## Core Competencies ### 1. Physical Layer Standards #### 100BASE-T1 (IEEE 802.3bw) ```c // 100BASE-T1 PHY Configuration typedef struct { uint8_t standard; // IEEE 802.3bw uint16_t data_rate_mbps; // 100 Mbps full-duplex uint8_t wire_pairs; // 1 twisted pair uint16_t max_length_m; // 15 meters (typ), 40m (max) float voltage_p2p; // 2.4V peak-to-peak uint8_t encoding; // PAM3 (3-level) bool pma_master; // Master/Slave negotiation } BASE100_T1_Config; // Example configuration for zone controller BASE100_T1_Config zcu_phy = { .standard = IEEE_802_3bw, .data_rate_mbps = 100, .wire_pairs = 1, .max_length_m = 15, .voltage_p2p = 2.4, .encoding = PAM3, .pma_master = true // Zone controller is master }; ``` #### 1000BASE-T1 (IEEE 802.3bp) ```c // 1000BASE-T1 PHY Configuration (for cameras, ADAS) typedef struct { uint8_t standard; // IEEE 802.3bp uint16_t data_rate_mbps; // 1000 Mbps full-duplex uint8_t wire_pairs; // 1 unshielded twisted pair uint16_t max_length_m; // 15m (standard), 40m (extended) uint8_t encoding; // PAM3 bool automotive_grade; // AEC-Q100 qualified float temp_range[2]; // -40°C to +125°C } BASE1000_T1_Config; // Camera link configuration BASE1000_T1_Config camera_phy = { .standard = IEEE_802_3bp, .data_rate_mbps = 1000, .wire_pairs = 1, .max_length_m = 15, .encoding = PAM3, .automotive_grade = true, .temp_range = {-40.0, 125.0} }; ``` #### 10BASE-T1S (IEEE 802.3cg) - Multidrop Bus ```c // 10BASE-T1S for low-cost sensor networks typedef struct { uint8_t standard; // IEEE 802.3cg uint16_t data_rate_mbps; // 10 Mbps half-duplex uint8_t topology; // Multidrop bus uint8_t max_nodes; // 8 nodes per segment uint16_t max_length_m; // 25 meters uint8_t collision_detection; // CSMA/CD bool plca_mode; // Physical Layer Collision Avoidance } BASE10_T1S_Config; // Sensor bus configuration BASE10_T1S_Config sensor_bus = { .standard = IEEE_802_3cg, .data_rate_mbps = 10, .topology = MULTIDROP_BUS, .max_nodes = 8, .max_length_m = 25, .collision_detection = CSMA_CD, .plca_mode = true // Enables deterministic access }; ``` ### 2. Time-Sensitive Networking (TSN) #### IEEE 802.1 TSN Standards **Key Standards:** - **802.1AS** - Precision Time Protocol (gPTP) for time synchronization - **802.1Qbv** - Time-Aware Shaper (TAS) for scheduled traffic - **802.1Qav** - Credit-Based Shaper (CBS) for AVB streams - **802.1Qbu** - Frame Preemption for low-latency - **802.1Qci** - Per-Stream Filtering and Policing - **802.1CB** - Frame Replication and Elimination for Reliability (FRER) ```python # TSN Configuration Example class TSNSwitchConfig: def __init__(self): self.gptp_domain = 0 # Time domain for sync self.sync_interval_ms = 125 # gPTP sync every 125ms self.time_aware_shaper = True self.frame_preemption = True self.stream_reservation = True def configure_tas_schedule(self): """ Configure Time-Aware Shaper (802.1Qbv) for deterministic scheduling. Divides time into repeating cycles with gates for each priority queue. """ # 1ms cycle time (1,000,000 ns) cycle_time_ns = 1_000_000 schedule = { 'cycle_time_ns': cycle_time_ns, 'gates': [ # Time slot 0-100μs: Priority 7 (Safety-critical) { 'start_ns': 0, 'duration_ns': 100_000, 'open_gates': [7], # Only priority 7 queue open 'traffic_class': 'Safety' }, # Time slot 100-300μs: Priority 6 (ADAS) { 'start_ns': 100_000, 'duration_ns': 200_000, 'open_gates': [6], 'traffic_class': 'Control' }, # Time slot 300-800μs: Priority 4-5 (Video streams) { 'start_ns': 300_000, 'duration_ns': 500_000, 'open_gates': [4, 5], 'traffic_class': 'AVB' }, # Time slot 800-1000μs: Priority 0-3 (Best effort) { 'start_ns': 800_000, 'duration_ns': 200_000, 'open_gates': [0, 1, 2, 3], 'traffic_class': 'BestEffort' } ] } return schedule def configure_stream_reservation(self, stream_id, bandwidth_mbps, latency_us): """ Configure stream reservation for AVB/TSN streams (802.1Qat/Qcc). Args: stream_id: Unique stream identifier bandwidth_mbps: Required bandwidth in Mbps latency_us: Maximum latency in microseconds """ stream_config = { 'stream_id': stream_id, 'talker_mac': '00:11:22:33:44:55', 'listener_mac': ['00:11:22:33:44:66'], 'vlan_id': 100, 'priority': 6, # SR Class A 'max_frame_size': 1522, 'max_interval_frames': 1, 'bandwidth_mbps': bandwidth_mbps, 'max_latency_us': latency_us, 'redundancy': 'FRER' # Frame Replication } return stream_config ``` #### gPTP Time Synchronization (802.1AS) ```c // gPTP Time Synchronization Configuration typedef struct { uint8_t domain_number; // 0 for automotive uint32_t sync_interval_ns; // 125ms = 125,000,000 ns uint32_t pdelay_interval_ns; // Peer delay measurement int8_t clock_class; // 248 for automotive grandmaster int8_t clock_accuracy; // 0xFE (unknown) uint16_t offset_scaled_log_var; // Variance of clock uint8_t priority1; // 248 uint8_t priority2; // 248 bool as_capable; // TSN-capable port } gPTP_Config_t; // Grandmaster clock (gateway) gPTP_Config_t grandmaster = { .domain_number = 0, .sync_interval_ns = 125000000, // 125ms .pdelay_interval_ns = 1000000000, // 1 second .clock_class = 248, // Automotive default application-specific .clock_accuracy = 0xFE, .offset_scaled_log_var = 0x4E5D, .priority1 = 248, .priority2 = 248, .as_capable = true }; // Typical time sync accuracy: ±500ns between nodes ``` ### 3. VLAN Configuration ```python class VLANManager: """ Manage VLANs for traffic segregation in zonal architecture. """ def __init__(self): self.vlans = { 100: {'name': 'Safety', 'priority': 7, 'color': 'RED'}, 200: {'name': 'ADAS', 'priority': 6, 'color': 'ORANGE'}, 300: {'name': 'Infotainment', 'priority': 5, 'color': 'YELLOW'}, 400: {'name': 'Body', 'priority': 4, 'color': 'GREEN'}, 500: {'name': 'Diagnostics', 'priority': 3, 'color': 'BLUE'}, 999: {'name': 'Management', 'priority': 7, 'color': 'PURPLE'} } def configure_switch_ports(self): """ Configure switch ports with VLAN memberships. """ port_config = { 'port_1': { # Gateway uplink 'mode': 'trunk', 'allowed_vlans': [100, 200, 300, 400, 500, 999], 'native_vlan': 999, 'pvid': 999 }, 'port_2': { # Front-left ZCU 'mode': 'trunk', 'allowed_vlans': [100, 400], # Safety + Body 'native_vlan': 400, 'pvid': 400 }, 'port_3': { # Front camera (ADAS) 'mode': 'access', 'vlan': 200, # ADAS VLAN only 'pvid': 200 }, 'port_4': { # Rear camera (infotainment) 'mode': 'access', 'vlan': 300, # Infotainment VLAN 'pvid': 300 }, 'port_5': { # Diagnostic connector (OBD-II) 'mode': 'access', 'vlan': 500, 'pvid': 500 } } return port_config ``` ### 4. Quality of Service (QoS) #### Priority Mapping (IEEE 802.1Q) ```c // 8 priority levels (0-7) typedef enum { PRIORITY_0_BEST_EFFORT = 0, // Background PRIORITY_1_BACKGROUND = 1, // Backup data PRIORITY_2_EXCELLENT_EFFORT = 2, // Business-critical PRIORITY_3_CRITICAL_APPS = 3, // Call signaling PRIORITY_4_VIDEO = 4, // Streaming video PRIORITY_5_VOICE = 5, // Interactive voice/video PRIORITY_6_CONTROL = 6, // Control plane (ADAS) PRIORITY_7_NETWORK_CONTROL = 7 // Safety-critical } EthernetPriority_t; // Traffic class mapping typedef struct { EthernetPriority_t priority; uint8_t traffic_class; uint16_t max_latency_us; char description[32]; } QoS_Mapping_t; QoS_Mapping_t qos_table[] = { {PRIORITY_7_NETWORK_CONTROL, 7, 100, "Safety (ABS, ESC)"}, {PRIORITY_6_CONTROL, 6, 500, "ADAS (Braking, Steering)"}, {PRIORITY_5_VOICE, 5, 2000, "Camera streams"}, {PRIORITY_4_VIDEO, 4, 10000, "Infotainment video"}, {PRIORITY_3_CRITICAL_APPS, 3, 20000, "Diagnostics"}, {PRIORITY_2_EXCELLENT_EFFORT, 2, 50000, "SW updates"}, {PRIORITY_1_BACKGROUND, 1, 100000, "Telemetry"}, {PRIORITY_0_BEST_EFFORT, 0, 1000000, "General data"} }; ``` #### Credit-Based Shaper (802.1Qav) ```python def configure_cbs(port, stream_class): """ Configure Credit-Based Shaper for AVB traffic (SR Class A/B). Args: port: Ethernet port number stream_class: 'A' for Class A (2ms), 'B' for Class B (50ms) """ if stream_class == 'A': config = { 'idle_slope': 0x3FFF, # 75% of link bandwidth 'send_slope': -0x2AAA, # -25% of link bandwidth 'hi_credit': 0x186A0, # 100,000 credits 'lo_credit': -0x186A0, # -100,000 credits 'priority': 6 } elif stream_class == 'B': config = { 'idle_slope': 0x1FFF, # 50% of link bandwidth 'send_slope': -0x1FFF, # -50% of link bandwidth 'hi_credit': 0xC350, # 50,000 credits 'lo_credit': -0xC350, # -50,000 credits 'priority': 5 } return config ``` ### 5. Automotive Ethernet Switch Configuration ```yaml # Example switch configuration (YAML) switch: model: "NXP SJA1110" ports: 10 tsn_capable: true global_config: gptp_domain: 0 management_vlan: 999 port_1: # Uplink to gateway speed: "1000BASE-T1" mode: "trunk" vlans: [100, 200, 300, 400, 500, 999] tsn: tas_enabled: true frame_preemption: true port_2: # Front-left zone controller speed: "100BASE-T1" mode: "trunk" vlans: [100, 400] tsn: tas_enabled: true port_3: # Front camera speed: "1000BASE-T1" mode: "access" vlan: 200 qos: priority: 6 cbs_enabled: true stream_reservation: true port_4: # Rear camera speed: "1000BASE-T1" mode: "access" vlan: 200 qos: priority: 6 cbs_enabled: true ``` ## Network Performance Targets | Traffic Type | Priority | Max Latency | Jitter | Packet Loss | |--------------|----------|-------------|--------|-------------| | Safety (ABS, ESC) | 7 | <100 μs | <10 μs | 0% | | ADAS Control | 6 | <500 μs | <50 μs | <10^-9 | | Camera Streams | 5-6 | <2 ms | <100 μs | <10^-6 | | Infotainment | 4 | <10 ms | <1 ms | <10^-4 | | Diagnostics | 3 | <50 ms | N/A | <10^-3 | | Best Effort | 0-2 | <1 s | N/A | <10^-2 | ## Tools & Testing **Network Analyzers:** - **Vector VN5600** - TSN-capable network interface - **Wireshark with Automotive plugins** - Packet capture and analysis - **Ixia/Keysight IxNetwork** - TSN traffic generation and testing **Configuration Tools:** - **NXP SJA1110 Config Tool** - Switch configuration - **Vector CANoe.Ethernet** - Network simulation - **Marvell TSN Studio** - TSN stream configuration ## References - IEEE 802.1 TSN Task Group Standards - SAE J3161 On-Board Ethernet Communication - OPEN Alliance BroadR-Reach Specification - AUTOSAR Ethernet Communication Specification --- ## Network Security Zonal # Network Security for Zonal Architecture **Category:** automotive-zonal **Version:** 1.0.0 **Maturity:** production **Complexity:** advanced ## Overview Expert knowledge in securing automotive Ethernet networks in zonal architectures. Covers MACsec (IEEE 802.1AE), IPsec, firewall rules for zone controllers, intrusion detection systems (IDS), secure gateway design, and IDPS deployment for vehicle networks. ## Core Competencies ### 1. MACsec (IEEE 802.1AE) - Layer 2 Encryption ```c // MACsec Configuration for Automotive Ethernet typedef struct { uint8_t enabled; uint8_t cipher_suite; // AES-GCM-128 or AES-GCM-256 uint8_t confidentiality; // Encrypt payload uint8_t integrity; // ICV (Integrity Check Value) uint32_t pn; // Packet Number (anti-replay) uint8_t key[32]; // 128-bit or 256-bit key uint8_t sci[8]; // Secure Channel Identifier } MACSec_Config_t; // Example: MACsec between gateway and zone controller MACSec_Config_t macsec_link = { .enabled = 1, .cipher_suite = AES_GCM_256, .confidentiality = 1, // Encrypt .integrity = 1, // 16-byte ICV .pn = 0x00000001, // Initial packet number .key = {0x2b, 0x7e, 0x15, ...}, // 256-bit key from key management .sci = {0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77} }; ``` **MACsec Frame Structure:** ``` ┌──────────────────────────────────────────────┐ │ Ethernet Header (14 bytes) │ ├──────────────────────────────────────────────┤ │ SecTAG (8 bytes) │ │ - TCI/AN (1 byte): Version, encrypted flag │ │ - SL (1 byte): Short Length │ │ - PN (4 bytes): Packet Number │ │ - SCI (8 bytes): Secure Channel ID │ ├──────────────────────────────────────────────┤ │ Encrypted Payload │ ├──────────────────────────────────────────────┤ │ ICV (16 bytes): Integrity Check Value │ └──────────────────────────────────────────────┘ ``` **Performance Impact:** - Latency overhead: ~100-200 μs - Throughput reduction: ~5-10% (due to encryption) - CPU overhead: ~15% on zone controller ### 2. IPsec for End-to-End Security ```python class IPsecTunnel: """ IPsec tunnel configuration for secure SOME/IP communication. """ def __init__(self): self.mode = 'ESP' # Encapsulating Security Payload self.encryption = 'AES-256-CBC' self.authentication = 'HMAC-SHA256' self.pfs = True # Perfect Forward Secrecy def configure_tunnel(self, local_ip, remote_ip): """ Configure IPsec tunnel between two ECUs. Args: local_ip: Local zone controller IP remote_ip: Remote zone controller/gateway IP """ config = { 'src': local_ip, 'dst': remote_ip, 'protocol': 'ESP', # ESP for encryption + auth 'spi': 0x1234, # Security Parameter Index 'encryption': { 'algorithm': 'AES-256-CBC', 'key': self._generate_key(256) }, 'authentication': { 'algorithm': 'HMAC-SHA256', 'key': self._generate_key(256) }, 'lifetime': 3600, # Rekey every hour 'anti_replay': True, 'window_size': 64 } return config def _generate_key(self, bits): """Generate cryptographic key (placeholder - use proper KMS).""" import secrets return secrets.token_bytes(bits // 8) # Example: Secure tunnel from FL zone to gateway tunnel = IPsecTunnel() config = tunnel.configure_tunnel( local_ip='192.168.1.10', # FL Zone Controller remote_ip='192.168.1.1' # Gateway ) ``` ### 3. Firewall Rules for Zone Controllers ```python class ZonalFirewall: """ Stateful firewall for zone controller Ethernet interface. """ def __init__(self, zone_id): self.zone_id = zone_id self.rules = [] self.default_policy = 'DROP' # Deny by default def add_rule(self, rule): """ Add firewall rule. Rule format: { 'src_ip': '192.168.1.0/24', 'dst_ip': '192.168.2.10', 'protocol': 'UDP', 'dst_port': 30500, 'action': 'ALLOW'/'DROP'/'REJECT', 'priority': 100 } """ self.rules.append(rule) # Sort by priority self.rules.sort(key=lambda x: x['priority']) def evaluate_packet(self, packet): """ Evaluate packet against firewall rules. Returns: 'ALLOW', 'DROP', or 'REJECT' """ for rule in self.rules: if self._match_rule(packet, rule): return rule['action'] return self.default_policy def _match_rule(self, packet, rule): """Check if packet matches rule.""" # Check source IP if not self._ip_in_subnet(packet['src_ip'], rule.get('src_ip', 'any')): return False # Check destination IP if not self._ip_in_subnet(packet['dst_ip'], rule.get('dst_ip', 'any')): return False # Check protocol if rule.get('protocol', 'any') != 'any' and packet['protocol'] != rule['protocol']: return False # Check port if rule.get('dst_port') and packet.get('dst_port') != rule['dst_port']: return False return True # Example: Firewall rules for FL Zone Controller fw = ZonalFirewall(zone_id='FL_ZONE') # Allow SOME/IP from gateway fw.add_rule({ 'src_ip': '192.168.1.1', # Gateway 'dst_ip': '192.168.1.10', # FL Zone 'protocol': 'UDP', 'dst_port': 30500, # SOME/IP service port 'action': 'ALLOW', 'priority': 100 }) # Allow diagnostic access (DoIP) fw.add_rule({ 'src_ip': '192.168.5.0/24', # Diagnostic VLAN 'dst_ip': '192.168.1.10', 'protocol': 'TCP', 'dst_port': 13400, # DoIP port 'action': 'ALLOW', 'priority': 200 }) # Block all other incoming traffic fw.add_rule({ 'src_ip': 'any', 'dst_ip': '192.168.1.10', 'action': 'DROP', 'priority': 1000 }) ``` ### 4. Intrusion Detection System (IDS) ```python class AutomotiveIDS: """ Intrusion Detection System for automotive Ethernet networks. Detects anomalies and attacks specific to vehicle networks. """ def __init__(self): self.baseline = {} # Normal traffic patterns self.alerts = [] def detect_anomalies(self, traffic_sample): """ Detect network anomalies. Detection methods: - Signature-based: Known attack patterns - Anomaly-based: Deviation from baseline - Behavior-based: Unusual communication patterns """ alerts = [] # 1. Port scan detection if self._detect_port_scan(traffic_sample): alerts.append({ 'type': 'PORT_SCAN', 'severity': 'HIGH', 'description': 'Port scan detected from ' + traffic_sample['src_ip'] }) # 2. DoS detection (flooding) if self._detect_dos(traffic_sample): alerts.append({ 'type': 'DOS_ATTACK', 'severity': 'CRITICAL', 'description': 'Potential DoS attack detected' }) # 3. Unusual SOME/IP service access if self._detect_unauthorized_service_access(traffic_sample): alerts.append({ 'type': 'UNAUTHORIZED_ACCESS', 'severity': 'HIGH', 'description': 'Unauthorized SOME/IP service access' }) # 4. ARP spoofing detection if self._detect_arp_spoofing(traffic_sample): alerts.append({ 'type': 'ARP_SPOOFING', 'severity': 'CRITICAL', 'description': 'ARP spoofing detected' }) # 5. Replay attack detection (abnormal packet rate) if self._detect_replay(traffic_sample): alerts.append({ 'type': 'REPLAY_ATTACK', 'severity': 'MEDIUM', 'description': 'Potential replay attack detected' }) return alerts def _detect_port_scan(self, traffic): """ Detect port scanning: - Multiple connection attempts to different ports - From same source IP in short time window """ src_ip = traffic.get('src_ip') unique_ports = traffic.get('unique_dst_ports', []) time_window = traffic.get('time_window_sec', 0) # More than 20 ports in 10 seconds = port scan if len(unique_ports) > 20 and time_window < 10: return True return False def _detect_dos(self, traffic): """ Detect Denial of Service: - Packet rate > 10x baseline - Same message repeated at high rate """ pkt_rate = traffic.get('packets_per_second', 0) baseline_rate = self.baseline.get('avg_pkt_rate', 100) if pkt_rate > baseline_rate * 10: return True return False def _detect_unauthorized_service_access(self, traffic): """ Detect unauthorized SOME/IP service access: - Access to service not in whitelist - Access from unauthorized client """ service_id = traffic.get('someip_service_id') client_ip = traffic.get('src_ip') authorized_services = { 0x1234: ['192.168.1.1'], # Battery service - gateway only 0x5678: ['192.168.1.1', '192.168.1.10'] # Body service - gateway + FL zone } if service_id in authorized_services: if client_ip not in authorized_services[service_id]: return True return False def _detect_arp_spoofing(self, traffic): """ Detect ARP spoofing: - Different MAC for same IP - Gratuitous ARP with conflicting info """ if traffic.get('protocol') == 'ARP': ip = traffic.get('ip') mac = traffic.get('mac') known_mac = self.baseline.get('ip_to_mac', {}).get(ip) if known_mac and known_mac != mac: return True # MAC changed for this IP return False def _detect_replay(self, traffic): """ Detect replay attacks: - Same packet repeated (duplicate sequence numbers) - Packet rate anomaly for specific message """ msg_id = traffic.get('msg_id') pkt_count = traffic.get('pkt_count', 0) baseline_count = self.baseline.get('msg_counts', {}).get(msg_id, 1) if pkt_count > baseline_count * 5: return True return False ``` ### 5. Secure Gateway Design ```c // Secure gateway functionality typedef struct { uint8_t zone_count; struct { uint8_t zone_id; uint32_t ip_address; uint8_t vlan_id; bool macsec_enabled; bool ipsec_enabled; bool firewall_enabled; } zones[8]; struct { bool ids_enabled; bool ips_enabled; // Intrusion Prevention uint16_t alert_threshold; char siem_server[64]; // SIEM logging } security; } SecureGateway_t; // Example gateway configuration SecureGateway_t gateway = { .zone_count = 4, .zones = { {.zone_id = 1, .ip_address = 0xC0A80110, .vlan_id = 100, .macsec_enabled = true, .ipsec_enabled = false, .firewall_enabled = true}, // FL Zone {.zone_id = 2, .ip_address = 0xC0A80120, .vlan_id = 100, .macsec_enabled = true, .ipsec_enabled = false, .firewall_enabled = true}, // FR Zone {.zone_id = 3, .ip_address = 0xC0A80130, .vlan_id = 200, .macsec_enabled = true, .ipsec_enabled = true, .firewall_enabled = true}, // ADAS Zone (extra IPsec) {.zone_id = 4, .ip_address = 0xC0A80140, .vlan_id = 100, .macsec_enabled = true, .ipsec_enabled = false, .firewall_enabled = true} // RL Zone }, .security = { .ids_enabled = true, .ips_enabled = true, // Block detected attacks automatically .alert_threshold = 10, // Alert after 10 suspicious events .siem_server = "192.168.99.10" } }; ``` ## Security Architecture Layers ``` ┌────────────────────────────────────────────┐ │ Layer 7: Application Security │ │ - SOME/IP authentication │ │ - Service access control │ └────────────────────────────────────────────┘ ┌────────────────────────────────────────────┐ │ Layer 4-6: Transport/Session Security │ │ - IPsec (ESP): End-to-end encryption │ │ - TLS 1.3: For diagnostic protocols │ └────────────────────────────────────────────┘ ┌────────────────────────────────────────────┐ │ Layer 3: Network Security │ │ - Firewall: Packet filtering │ │ - IDS/IPS: Anomaly detection │ └────────────────────────────────────────────┘ ┌────────────────────────────────────────────┐ │ Layer 2: Data Link Security │ │ - MACsec (IEEE 802.1AE): Link encryption │ │ - 802.1X: Port-based authentication │ └────────────────────────────────────────────┘ ``` ## Key Management ```python class VehicleKeyManagement: """ Key management for MACsec and IPsec. Supports both static provisioning and dynamic key exchange. """ def __init__(self): self.keys = {} self.key_lifetime_hours = 24 # Rotate every 24 hours def provision_static_key(self, zone_id, key_type, key_material): """ Provision static key during manufacturing. Args: zone_id: Zone controller ID key_type: 'MACSEC' or 'IPSEC' key_material: 256-bit key """ self.keys[zone_id] = { 'type': key_type, 'key': key_material, 'provisioned_at': time.time(), 'valid_until': time.time() + (self.key_lifetime_hours * 3600) } def rotate_keys(self): """ Automatic key rotation every 24 hours. Uses Diffie-Hellman key exchange for new keys. """ for zone_id, key_info in self.keys.items(): if time.time() > key_info['valid_until']: # Generate new key new_key = self._dh_key_exchange(zone_id) self.keys[zone_id]['key'] = new_key self.keys[zone_id]['valid_until'] = time.time() + (self.key_lifetime_hours * 3600) def _dh_key_exchange(self, zone_id): """Diffie-Hellman key exchange (simplified).""" # In production, use proper DH or ECDH import secrets return secrets.token_bytes(32) # 256-bit key ``` ## Performance Impact | Security Feature | Latency Overhead | CPU Overhead | Throughput Impact | |------------------|------------------|--------------|-------------------| | MACsec (AES-128) | +100 μs | +10% | -5% | | MACsec (AES-256) | +150 μs | +15% | -8% | | IPsec (ESP) | +200 μs | +20% | -10% | | Firewall | +50 μs | +5% | -2% | | IDS (passive) | +10 μs | +8% | 0% | | IPS (active) | +100 μs | +15% | -5% | ## Tools & Testing - **Wireshark with MACsec plugin** - Decrypt and analyze MACsec traffic - **Scapy** - Craft attack packets for security testing - **Suricata** - Open-source IDS/IPS engine - **Kali Linux** - Penetration testing toolkit - **CANalyze** - Automotive-specific security testing ## References - IEEE 802.1AE (MACsec) Standard - ISO/SAE 21434 Cybersecurity Engineering - UNECE R155 Cybersecurity Regulation - AUTOSAR Secure Communication Specification --- ## Service Oriented Communication # Service-Oriented Communication - SOME/IP & DDS **Category:** automotive-zonal **Version:** 1.0.0 **Maturity:** production **Complexity:** advanced ## Overview Expert knowledge in service-oriented middleware for automotive zonal architectures. Covers SOME/IP (Scalable Service-Oriented Middleware over IP), DDS (Data Distribution Service), service discovery, publish-subscribe patterns, event-driven architecture, and method invocations over Ethernet. ## Core Competencies ### 1. SOME/IP (AUTOSAR Standard) #### Protocol Overview **SOME/IP = Scalable Service-Oriented Middleware over IP** - Used in AUTOSAR Adaptive Platform - Transport: UDP or TCP over IPv4/IPv6 - Serialization: SOME/IP binary format - Discovery: SOME/IP-SD (Service Discovery) ```c // SOME/IP Message Header (16 bytes) typedef struct __attribute__((packed)) { uint32_t message_id; // Service ID (16 bits) + Method ID (16 bits) uint32_t length; // Payload length + 8 uint32_t request_id; // Client ID (16 bits) + Session ID (16 bits) uint8_t protocol_version; // 0x01 uint8_t interface_version; // Service interface version uint8_t message_type; // REQUEST=0x00, RESPONSE=0x80, ERROR=0x81, NOTIFICATION=0x02 uint8_t return_code; // E_OK=0x00, E_NOT_OK=0x01, etc. } SOMEIP_Header_t; // Example: Request message SOMEIP_Header_t request = { .message_id = 0x12340001, // Service 0x1234, Method 0x0001 .length = 24, // Header (16) + Payload (8) .request_id = 0x00010001, // Client 0x0001, Session 0x0001 .protocol_version = 0x01, .interface_version = 0x01, .message_type = 0x00, // REQUEST .return_code = 0x00 // E_OK }; ``` #### Service Definition (FIDL) ```fidl // Franca IDL (FIDL) - SOME/IP Service Definition package org.genivi.battery interface BatteryManagementService { version { major 1 minor 0 } // Methods (Request/Response) method GetBatteryStatus { out { UInt8 stateOfCharge // 0-100% Float voltage // Volts Float current // Amps Int8 temperature // Celsius } } method SetChargingLimit { in { UInt8 targetSoC // Target SOC % } out { Boolean success } } // Events (Notifications) broadcast BatteryAlarm { out { UInt16 alarmCode String description } } // Attributes (Getter/Setter/Notification) attribute UInt8 stateOfCharge readonly // Error codes enumeration BatteryError { OK = 0 INVALID_PARAMETER = 1 HARDWARE_FAULT = 2 COMMUNICATION_ERROR = 3 } } ``` #### SOME/IP Service Implementation ```cpp #include #include using namespace v1::org::genivi::battery; class BatteryClient { public: BatteryClient() { runtime_ = CommonAPI::Runtime::get(); proxy_ = runtime_->buildProxy( "local", "BatteryService"); // Wait for service availability while (!proxy_->isAvailable()) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } // Subscribe to battery alarms proxy_->getBatteryAlarmEvent().subscribe( [](uint16_t alarmCode, std::string description) { std::cout << "Alarm " << alarmCode << ": " << description << std::endl; }); } void getBatteryStatus() { // Synchronous method call CommonAPI::CallStatus callStatus; uint8_t soc; float voltage, current; int8_t temperature; proxy_->GetBatteryStatus(callStatus, soc, voltage, current, temperature); if (callStatus == CommonAPI::CallStatus::SUCCESS) { std::cout << "SOC: " << (int)soc << "%" << std::endl; std::cout << "Voltage: " << voltage << "V" << std::endl; std::cout << "Current: " << current << "A" << std::endl; std::cout << "Temperature: " << (int)temperature << "°C" << std::endl; } } void setChargingLimit(uint8_t targetSoC) { // Asynchronous method call with callback proxy_->SetChargingLimitAsync( targetSoC, [](const CommonAPI::CallStatus& status, bool success) { if (status == CommonAPI::CallStatus::SUCCESS && success) { std::cout << "Charging limit set successfully" << std::endl; } }); } private: std::shared_ptr runtime_; std::shared_ptr> proxy_; }; ``` #### SOME/IP-SD (Service Discovery) ```python class SOMEIPServiceDiscovery: """ SOME/IP Service Discovery (SOME/IP-SD) implementation. Uses UDP multicast (224.244.224.245:30490) for service advertisement. """ def __init__(self): self.multicast_group = '224.244.224.245' self.multicast_port = 30490 self.services = {} def offer_service(self, service_id, instance_id, endpoint): """ Offer a service via SOME/IP-SD. Args: service_id: Service identifier (16-bit) instance_id: Instance identifier (16-bit) endpoint: (IP, port, protocol) protocol='UDP' or 'TCP' """ offer_message = { 'message_type': 'OfferService', 'service_id': service_id, 'instance_id': instance_id, 'major_version': 1, 'minor_version': 0, 'ttl': 3, # Time-to-live in seconds (0xFFFFFF = infinite) 'endpoint': { 'ipv4': endpoint[0], 'port': endpoint[1], 'protocol': endpoint[2] # UDP or TCP } } # Send cyclic offers (every 1 second) # Until service is stopped return offer_message def find_service(self, service_id, instance_id=None): """ Find a service via SOME/IP-SD. Args: service_id: Service to find instance_id: Specific instance (None = any) Returns: List of available service endpoints """ find_message = { 'message_type': 'FindService', 'service_id': service_id, 'instance_id': instance_id if instance_id else 0xFFFF, # ANY 'major_version': 0xFF, # ANY 'minor_version': 0xFFFFFFFF # ANY } # Wait for OfferService responses # Return list of endpoints return [] # Example usage: sd = SOMEIPServiceDiscovery() # Offer battery service sd.offer_service( service_id=0x1234, instance_id=0x0001, endpoint=('192.168.1.10', 30500, 'UDP') ) # Find battery service endpoints = sd.find_service(service_id=0x1234) ``` ### 2. DDS (Data Distribution Service) #### DDS Quality of Service (QoS) ```python from dataclasses import dataclass from enum import Enum class ReliabilityKind(Enum): BEST_EFFORT = 0 # UDP-like, lossy RELIABLE = 1 # TCP-like, guaranteed delivery class DurabilityKind(Enum): VOLATILE = 0 # Only for live data TRANSIENT_LOCAL = 1 # Store last value for late joiners TRANSIENT = 2 # Persist across processes PERSISTENT = 3 # Persist to disk @dataclass class DDSQoS: """DDS Quality of Service configuration.""" reliability: ReliabilityKind durability: DurabilityKind history_depth: int # Number of samples to keep max_blocking_time_ms: int # Max time to block writer latency_budget_ms: int # Hint for latency optimization lifespan_ms: int # Sample validity duration # Example QoS profiles for different use cases # Safety-critical real-time data (ESC, ABS) SAFETY_QOS = DDSQoS( reliability=ReliabilityKind.RELIABLE, durability=DurabilityKind.VOLATILE, history_depth=1, # Only latest value matters max_blocking_time_ms=10, latency_budget_ms=5, lifespan_ms=100 ) # Sensor data (high-rate, best-effort) SENSOR_QOS = DDSQoS( reliability=ReliabilityKind.BEST_EFFORT, durability=DurabilityKind.VOLATILE, history_depth=5, max_blocking_time_ms=0, # Non-blocking latency_budget_ms=1, lifespan_ms=50 ) # Configuration data (late-joiner support) CONFIG_QOS = DDSQoS( reliability=ReliabilityKind.RELIABLE, durability=DurabilityKind.TRANSIENT_LOCAL, history_depth=1, max_blocking_time_ms=1000, latency_budget_ms=100, lifespan_ms=0 # No expiration ) ``` #### DDS Topic Definition (IDL) ```idl // OMG IDL for DDS topics module automotive { module battery { struct BatteryStatus { unsigned long timestamp; // Unix timestamp (ms) octet stateOfCharge; // 0-100% float voltage; // Volts float current; // Amps char temperature; // Celsius boolean charging; }; struct BatteryAlarm { unsigned long timestamp; unsigned short alarmCode; string<256> description; octet severity; // 0=Info, 1=Warning, 2=Error, 3=Critical }; }; }; ``` #### DDS Publisher/Subscriber (C++) ```cpp #include #include "BatteryStatus.hpp" using namespace automotive::battery; class BatteryPublisher { public: BatteryPublisher() { // Create DDS participant (one per application) participant_ = dds::domain::DomainParticipant(0); // Create topic topic_ = dds::topic::Topic( participant_, "BatteryStatusTopic"); // Create publisher with QoS dds::pub::qos::PublisherQos pub_qos; publisher_ = dds::pub::Publisher(participant_, pub_qos); // Create data writer dds::pub::qos::DataWriterQos writer_qos; writer_qos << SAFETY_QOS; // Use safety QoS profile writer_ = dds::pub::DataWriter(publisher_, topic_, writer_qos); } void publishStatus(uint8_t soc, float voltage, float current, int8_t temp) { BatteryStatus status; status.timestamp(std::chrono::system_clock::now().time_since_epoch().count()); status.stateOfCharge(soc); status.voltage(voltage); status.current(current); status.temperature(temp); status.charging(current > 0); writer_.write(status); } private: dds::domain::DomainParticipant participant_; dds::topic::Topic topic_; dds::pub::Publisher publisher_; dds::pub::DataWriter writer_; }; class BatterySubscriber { public: BatterySubscriber() { participant_ = dds::domain::DomainParticipant(0); topic_ = dds::topic::Topic( participant_, "BatteryStatusTopic"); dds::sub::qos::SubscriberQos sub_qos; subscriber_ = dds::sub::Subscriber(participant_, sub_qos); dds::sub::qos::DataReaderQos reader_qos …(truncated)