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)

// 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)

// 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

// 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)

# 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)

// 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

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)

// 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)

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

# 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

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

// 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

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

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)

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

// 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

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

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)

// 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)

// 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

#include <CommonAPI/CommonAPI.hpp>
#include <v1/org/genivi/battery/BatteryManagementServiceProxy.hpp>

using namespace v1::org::genivi::battery;

class BatteryClient {
public:
    BatteryClient() {
        runtime_ = CommonAPI::Runtime::get();
        proxy_ = runtime_->buildProxy<BatteryManagementServiceProxy>(
            "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<CommonAPI::Runtime> runtime_;
    std::shared_ptr<BatteryManagementServiceProxy<>> proxy_;
};

SOME/IP-SD (Service Discovery)

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)

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)

// 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++)

#include <dds/dds.hpp>
#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<BatteryStatus>(
            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<BatteryStatus>(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<BatteryStatus> topic_;
    dds::pub::Publisher publisher_;
    dds::pub::DataWriter<BatteryStatus> writer_;
};

class BatterySubscriber {
public:
    BatterySubscriber() {
        participant_ = dds::domain::DomainParticipant(0);
        topic_ = dds::topic::Topic<BatteryStatus>(
            participant_, "BatteryStatusTopic");

        dds::sub::qos::SubscriberQos sub_qos;
        subscriber_ = dds::sub::Subscriber(participant_, sub_qos);

        dds::sub::qos::DataReaderQos reader_qos

…(truncated)