Vehicle Security Operations Center (VSOC)

#!/usr/bin/env python3
# VSOC fleet threat intelligence: detect coordinated attack campaigns
from collections import defaultdict
from datetime import datetime, timedelta

def detect_coordinated_attack(events: list, window_hours=48,
                               min_vins=50) -> list:
    # Identify attack signatures appearing on >= min_vins vehicles within window
    campaigns = []
    sig_to_vins = defaultdict(set)

    cutoff = datetime.utcnow() - timedelta(hours=window_hours)
    for ev in events:
        if ev["timestamp"] >= cutoff:
            sig_to_vins[ev["attack_signature"]].add(ev["vin"])

    for sig, vins in sig_to_vins.items():
        if len(vins) >= min_vins:
            campaigns.append({
                "signature":  sig,
                "vin_count":  len(vins),
                "first_seen": min(ev["timestamp"] for ev in events
                                  if ev["attack_signature"]==sig),
                "severity":   "CRITICAL" if len(vins) >= 100 else "HIGH",
                "actions": [
                    "Generate IoC report for PSIRT",
                    "Share signature with AUTO-ISAC",
                    "Evaluate fleet-wide OTA countermeasure",
                ],
            })
            print(f"CAMPAIGN DETECTED: {sig[:50]} on {len(vins)} vehicles")

    return campaigns

# example events format:
# [{"vin":"WBA001","timestamp":datetime(...),"attack_signature":"SecOC_fail|EngTorque|0xDEAD"}]