Modül 5: Pratik Projeler ve Portfolio
Konular
İlerleme0%
Hafta 17-18: Proje 1 - CAN Bus Güvenlik Analiz Aracı
Proje 1: CANSecure - CAN Bus Güvenlik Analiz Aracı
🎯 Proje Hedefleri
- CAN mesajlarını yakalayacak sniffer geliştirmek
- Anomali tespiti (flood, replay, unknown ID)
- Güvenlik raporu oluşturmak
Proje Yapısı
cansecure/
├── __init__.py
├── sniffer.py # CAN mesaj yakalama
├── analyzer.py # Trafik analizi
├── anomaly.py # Anomali tespiti
├── reporter.py # Rapor oluşturma
└── main.py # CLI arayüzü
sniffer.py
#!/usr/bin/env python3
"""
CANSecure - CAN Sniffer Module
"""
import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from collections import defaultdict
import threading
@dataclass
class CANMessage:
"""CAN mesaj veri yapısı."""
timestamp: float
arbitration_id: int
data: bytes
dlc: int = field(init=False)
channel: str = "vcan0"
def __post_init__(self):
self.dlc = len(self.data)
@property
def data_hex(self) -> str:
return self.data.hex().upper()
@property
def is_uds(self) -> bool:
return 0x7E0 <= self.arbitration_id <= 0x7EF
def __str__(self) -> str:
return f"{self.timestamp:.6f} 0x{self.arbitration_id:03X} [{self.dlc}] {self.data_hex}"
class CANSniffer:
"""CAN mesaj yakalama sınıfı."""
def __init__(self, channel: str = "vcan0"):
self.channel = channel
self.messages: List[CANMessage] = []
self.running = False
self.callbacks: List[Callable[[CANMessage], None]] = []
self._lock = threading.Lock()
# İstatistikler
self.stats = {
"total_messages": 0,
"start_time": 0,
"id_counts": defaultdict(int)
}
def add_callback(self, callback: Callable[[CANMessage], None]):
"""Yeni mesaj callback'i ekle."""
self.callbacks.append(callback)
def _process_message(self, msg: CANMessage):
"""Mesajı işle."""
with self._lock:
self.messages.append(msg)
self.stats["total_messages"] += 1
self.stats["id_counts"][msg.arbitration_id] += 1
for callback in self.callbacks:
try:
callback(msg)
except Exception as e:
print(f"Callback hatası: {e}")
def simulate_traffic(self, duration: float = 5.0, rate: float = 100):
"""
Simüle edilmiş CAN trafiği oluştur.
Args:
duration: Süre (saniye)
rate: Mesaj/saniye
"""
import random
self.running = True
self.stats["start_time"] = time.time()
# Bilinen ID'ler
normal_ids = [0x100, 0x123, 0x200, 0x300, 0x456, 0x500]
uds_ids = [0x7E0, 0x7E8, 0x7E1, 0x7E9]
print(f"🔊 CAN trafiği simülasyonu başladı ({duration}s, {rate} msg/s)")
start = time.time()
msg_count = 0
while self.running and (time.time() - start) < duration:
# Rastgele ID seç
if random.random() < 0.1: # %10 UDS
arb_id = random.choice(uds_ids)
elif random.random() < 0.02: # %2 bilinmeyen (anomali)
arb_id = random.randint(0x600, 0x700)
else:
arb_id = random.choice(normal_ids)
# Veri oluştur
dlc = 8 if arb_id in uds_ids else random.randint(1, 8)
data = bytes([random.randint(0, 255) for _ in range(dlc)])
msg = CANMessage(
timestamp=time.time() - start,
arbitration_id=arb_id,
data=data,
channel=self.channel
)
self._process_message(msg)
msg_count += 1
# Rate limiting
time.sleep(1.0 / rate)
self.running = False
print(f"✅ Simülasyon tamamlandı: {msg_count} mesaj")
def stop(self):
"""Sniffing'i durdur."""
self.running = False
def get_statistics(self) -> Dict:
"""Trafik istatistiklerini döndür."""
with self._lock:
duration = time.time() - self.stats["start_time"] if self.stats["start_time"] else 0
return {
"total_messages": self.stats["total_messages"],
"unique_ids": len(self.stats["id_counts"]),
"duration_seconds": duration,
"messages_per_second": self.stats["total_messages"] / duration if duration > 0 else 0,
"top_ids": sorted(
self.stats["id_counts"].items(),
key=lambda x: x[1],
reverse=True
)[:10]
}
anomaly.py
#!/usr/bin/env python3
"""
CANSecure - Anomali Tespit Modülü
"""
from dataclasses import dataclass
from typing import List, Dict, Set
from enum import Enum
from collections import defaultdict
import time
class AnomalyType(Enum):
FLOOD = "CAN Bus Flooding"
UNKNOWN_ID = "Bilinmeyen CAN ID"
REPLAY = "Replay Saldırısı"
DIAGNOSTIC_IN_DRIVE = "Sürüş Sırasında Diagnostik"
UNUSUAL_TIMING = "Anormal Zamanlama"
@dataclass
class Anomaly:
"""Tespit edilen anomali."""
timestamp: float
anomaly_type: AnomalyType
severity: str # "LOW", "MEDIUM", "HIGH", "CRITICAL"
can_id: int
description: str
raw_data: bytes = b''
class AnomalyDetector:
"""CAN anomali tespit sınıfı."""
def __init__(self):
# Bilinen ID'ler (gerçek uygulamada veritabanından yüklenir)
self.known_ids: Set[int] = set()
self.id_intervals: Dict[int, List[float]] = defaultdict(list)
self.id_data_history: Dict[int, List[bytes]] = defaultdict(list)
self.anomalies: List[Anomaly] = []
# Eşikler
self.flood_threshold_ms = 1.0 # <1ms aralık = flood
self.replay_window_s = 5.0 # 5s içinde aynı veri = replay
# Durum
self.vehicle_state = "parked" # "parked", "driving"
self.last_timestamps: Dict[int, float] = {}
def learn_baseline(self, messages):
"""Normal trafik öğren."""
for msg in messages:
self.known_ids.add(msg.arbitration_id)
print(f"📊 {len(self.known_ids)} benzersiz CAN ID öğrenildi")
def analyze_message(self, msg) -> List[Anomaly]:
"""Tek mesajı analiz et."""
detected = []
current_time = msg.timestamp
arb_id = msg.arbitration_id
# 1. Bilinmeyen ID kontrolü
if self.known_ids and arb_id not in self.known_ids:
anomaly = Anomaly(
timestamp=current_time,
anomaly_type=AnomalyType.UNKNOWN_ID,
severity="MEDIUM",
can_id=arb_id,
description=f"Bilinmeyen CAN ID: 0x{arb_id:03X}",
raw_data=msg.data
)
detected.append(anomaly)
# 2. Flood tespiti
if arb_id in self.last_timestamps:
interval_ms = (current_time - self.last_timestamps[arb_id]) * 1000
if interval_ms < self.flood_threshold_ms:
anomaly = Anomaly(
timestamp=current_time,
anomaly_type=AnomalyType.FLOOD,
severity="HIGH",
can_id=arb_id,
description=f"Flooding tespit edildi: {interval_ms:.2f}ms aralık"
)
detected.append(anomaly)
self.last_timestamps[arb_id] = current_time
# 3. Replay tespiti
history = self.id_data_history[arb_id]
if msg.data in history:
# Aynı veri kısa süre içinde tekrarlandı
anomaly = Anomaly(
timestamp=current_time,
anomaly_type=AnomalyType.REPLAY,
severity="MEDIUM",
can_id=arb_id,
description=f"Tekrarlanan veri tespit edildi",
raw_data=msg.data
)
detected.append(anomaly)
# History'yi güncelle
history.append(msg.data)
if len(history) > 100: # Son 100 mesajı tut
history.pop(0)
# 4. Sürüş sırasında diagnostik
if self.vehicle_state == "driving" and msg.is_uds:
anomaly = Anomaly(
timestamp=current_time,
anomaly_type=AnomalyType.DIAGNOSTIC_IN_DRIVE,
severity="HIGH",
can_id=arb_id,
description="Sürüş sırasında diagnostik aktivitesi",
raw_data=msg.data
)
detected.append(anomaly)
self.anomalies.extend(detected)
return detected
def get_summary(self) -> Dict:
"""Anomali özeti."""
summary = defaultdict(int)
for a in self.anomalies:
summary[a.anomaly_type.value] += 1
return {
"total_anomalies": len(self.anomalies),
"by_type": dict(summary),
"critical_count": sum(1 for a in self.anomalies if a.severity == "CRITICAL"),
"high_count": sum(1 for a in self.anomalies if a.severity == "HIGH")
}
Demo Çalıştırma
#!/usr/bin/env python3
"""
CANSecure - Ana Demo
"""
if __name__ == "__main__":
print("
" + "=" * 60)
print("🔐 CANSECURE - CAN Bus Güvenlik Analiz Aracı")
print("=" * 60)
# Sniffer oluştur
sniffer = CANSniffer("vcan0")
detector = AnomalyDetector()
# Baseline öğren
baseline_ids = [0x100, 0x123, 0x200, 0x300, 0x456, 0x500, 0x7E0, 0x7E8]
detector.known_ids = set(baseline_ids)
# Callback ekle
def on_message(msg):
anomalies = detector.analyze_message(msg)
for a in anomalies:
severity_emoji = {
"CRITICAL": "🔴",
"HIGH": "🟠",
"MEDIUM": "🟡",
"LOW": "🟢"
}
print(f" {severity_emoji[a.severity]} [{a.anomaly_type.value}] 0x{a.can_id:03X}")
sniffer.add_callback(on_message)
# Simülasyon çalıştır
sniffer.simulate_traffic(duration=3.0, rate=50)
# Sonuçlar
print("
" + "-" * 40)
print("📊 TRAFİK İSTATİSTİKLERİ")
stats = sniffer.get_statistics()
print(f" Toplam: {stats['total_messages']} mesaj")
print(f" Benzersiz ID: {stats['unique_ids']}")
print(f" Hız: {stats['messages_per_second']:.1f} msg/s")
print("
" + "-" * 40)
print("⚠️ ANOMALİ ÖZETİ")
summary = detector.get_summary()
print(f" Toplam: {summary['total_anomalies']}")
print(f" Yüksek Öncelik: {summary['high_count']}")
for atype, count in summary['by_type'].items():
print(f" - {atype}: {count}")