automotive-cybersecurity

Automotive Cybersecurity

Intrusion Detection Prevention

Intrusion Detection & Prevention Skill

Overview

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

Core Competencies

IDS/IPS Architecture

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

Detection Techniques

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

CAN Bus Intrusion Detection System

CAN IDS Implementation

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

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

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

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

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

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

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

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

        start_time = time.time()
        message_count = 0

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

            self._update_statistics(msg)
            message_count += 1

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

        # Compute baseline statistics
        self._compute_baseline()

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

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

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

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

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

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

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

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

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

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

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

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

        from collections import Counter
        import math

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

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

        return entropy

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

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

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

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

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

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

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

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

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

        # Update statistics
        self._update_statistics(msg)

        return anomalies

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

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

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

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

                message_count += 1

                # Detect anomalies
                anomalies = self.detect_anomalies(msg)

                if anomalies:
                    anomaly_count += len(anomalies)

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

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

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

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

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

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

        # Log alert
        self.alerts.append(anomaly)

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

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

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

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

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


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

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

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

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

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

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

    # Learn baseline
    ids.learn_baseline(duration_seconds=60)

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

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

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

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

if __name__ == "__main__":
    demo_can_ids()