Trixy7/trixy_core/network/security/ddos.py

# -*- coding: utf-8 -*-
"""
DDoS-Schutz für Trixy Network.

Bietet Erkennung und Schutz vor verschiedenen DDoS-Angriffen:
- Connection Flooding
- SYN Floods
- Slowloris-Angriffe
- Amplification-Angriffe
"""

import asyncio
import time
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum, auto
from typing import Any, Callable


class ThreatLevel(IntEnum):
    """Bedrohungsstufe."""
    NONE = auto()       # Keine Bedrohung
    LOW = auto()        # Geringe Bedrohung
    MEDIUM = auto()     # Mittlere Bedrohung
    HIGH = auto()       # Hohe Bedrohung
    CRITICAL = auto()   # Kritische Bedrohung


@dataclass
class DDoSConfig:
    """
    Konfiguration für DDoS-Schutz.

    Attributes:
        max_connections_per_ip: Maximale gleichzeitige Verbindungen pro IP
        max_connection_rate: Maximale neue Verbindungen pro Sekunde pro IP
        max_global_connections: Maximale globale Verbindungen
        max_global_connection_rate: Maximale globale Verbindungsrate
        slowloris_timeout: Timeout für Slowloris-Erkennung
        suspicious_threshold: Schwelle für verdächtige Aktivität
        ban_duration: Dauer der automatischen Sperre in Sekunden
        monitoring_window: Zeitfenster für Analyse in Sekunden
    """
    max_connections_per_ip: int = 10
    max_connection_rate: float = 5.0  # pro Sekunde
    max_global_connections: int = 1000
    max_global_connection_rate: float = 100.0
    slowloris_timeout: float = 30.0
    suspicious_threshold: int = 5
    ban_duration: float = 600.0  # 10 Minuten
    monitoring_window: float = 60.0  # 1 Minute


@dataclass
class ConnectionInfo:
    """Informationen über eine einzelne Verbindung."""
    ip: str
    connected_at: float
    last_activity: float
    bytes_received: int = 0
    bytes_sent: int = 0
    requests_count: int = 0
    state: str = "connected"


@dataclass
class IPStatistics:
    """Statistiken für eine IP-Adresse."""
    ip: str
    connections: list[str] = field(default_factory=list)  # Connection-IDs
    connection_times: deque = field(default_factory=lambda: deque(maxlen=100))
    suspicious_events: int = 0
    last_suspicious: float | None = None
    banned_until: float | None = None
    total_connections: int = 0
    total_bytes_received: int = 0


class ConnectionTracker:
    """
    Verfolgt alle aktiven Verbindungen.

    Überwacht Verbindungen pro IP und global für DDoS-Erkennung.
    """

    def __init__(self, config: DDoSConfig | None = None) -> None:
        """
        Initialisiert den Connection-Tracker.

        Args:
            config: DDoS-Konfiguration
        """
        self._config = config or DDoSConfig()
        self._connections: dict[str, ConnectionInfo] = {}
        self._ip_stats: dict[str, IPStatistics] = {}
        self._connection_counter: int = 0
        self._lock = asyncio.Lock()

    def _get_or_create_ip_stats(self, ip: str) -> IPStatistics:
        """Gibt oder erstellt IP-Statistiken."""
        if ip not in self._ip_stats:
            self._ip_stats[ip] = IPStatistics(ip=ip)
        return self._ip_stats[ip]

    def register_connection(self, ip: str) -> tuple[str, bool]:
        """
        Registriert eine neue Verbindung.

        Args:
            ip: IP-Adresse der Verbindung

        Returns:
            Tuple aus (connection_id, erlaubt)
        """
        stats = self._get_or_create_ip_stats(ip)
        now = time.monotonic()

        # Prüfe Sperre
        if stats.banned_until and now < stats.banned_until:
            return "", False

        # Prüfe Verbindungslimit pro IP
        if len(stats.connections) >= self._config.max_connections_per_ip:
            stats.suspicious_events += 1
            stats.last_suspicious = now
            return "", False

        # Prüfe globales Limit
        if len(self._connections) >= self._config.max_global_connections:
            return "", False

        # Prüfe Verbindungsrate
        stats.connection_times.append(now)
        recent_connections = sum(
            1 for t in stats.connection_times
            if now - t < 1.0  # Letzter Sekunde
        )
        if recent_connections > self._config.max_connection_rate:
            stats.suspicious_events += 1
            stats.last_suspicious = now
            return "", False

        # Verbindung erlauben
        self._connection_counter += 1
        conn_id = f"conn_{self._connection_counter}"

        self._connections[conn_id] = ConnectionInfo(
            ip=ip,
            connected_at=now,
            last_activity=now
        )
        stats.connections.append(conn_id)
        stats.total_connections += 1

        return conn_id, True

    def unregister_connection(self, conn_id: str) -> None:
        """
        Entfernt eine Verbindung.

        Args:
            conn_id: Verbindungs-ID
        """
        if conn_id not in self._connections:
            return

        conn = self._connections.pop(conn_id)

        if conn.ip in self._ip_stats:
            stats = self._ip_stats[conn.ip]
            if conn_id in stats.connections:
                stats.connections.remove(conn_id)
            stats.total_bytes_received += conn.bytes_received

    def update_activity(
        self,
        conn_id: str,
        bytes_received: int = 0,
        bytes_sent: int = 0
    ) -> None:
        """
        Aktualisiert die Aktivität einer Verbindung.

        Args:
            conn_id: Verbindungs-ID
            bytes_received: Empfangene Bytes
            bytes_sent: Gesendete Bytes
        """
        if conn_id not in self._connections:
            return

        conn = self._connections[conn_id]
        conn.last_activity = time.monotonic()
        conn.bytes_received += bytes_received
        conn.bytes_sent += bytes_sent
        conn.requests_count += 1

    def get_connection_count(self, ip: str | None = None) -> int:
        """
        Gibt die Anzahl aktiver Verbindungen zurück.

        Args:
            ip: Optionale IP-Filterung

        Returns:
            Anzahl Verbindungen
        """
        if ip is None:
            return len(self._connections)

        if ip in self._ip_stats:
            return len(self._ip_stats[ip].connections)

        return 0

    def is_banned(self, ip: str) -> bool:
        """Prüft, ob eine IP gesperrt ist."""
        if ip not in self._ip_stats:
            return False

        stats = self._ip_stats[ip]
        if stats.banned_until is None:
            return False

        if time.monotonic() > stats.banned_until:
            stats.banned_until = None
            return False

        return True

    def ban_ip(self, ip: str, duration: float | None = None) -> None:
        """
        Sperrt eine IP-Adresse.

        Args:
            ip: Die zu sperrende IP
            duration: Sperrdauer (Standard: config.ban_duration)
        """
        stats = self._get_or_create_ip_stats(ip)
        duration = duration or self._config.ban_duration
        stats.banned_until = time.monotonic() + duration

        # Alle Verbindungen dieser IP schließen
        for conn_id in list(stats.connections):
            self.unregister_connection(conn_id)

    def unban_ip(self, ip: str) -> bool:
        """
        Entsperrt eine IP-Adresse.

        Args:
            ip: Die zu entsperrende IP

        Returns:
            True wenn IP gesperrt war
        """
        if ip not in self._ip_stats:
            return False

        stats = self._ip_stats[ip]
        was_banned = stats.banned_until is not None
        stats.banned_until = None
        stats.suspicious_events = 0

        return was_banned

    def get_statistics(self) -> dict[str, Any]:
        """Gibt globale Statistiken zurück."""
        now = time.monotonic()
        banned_count = sum(
            1 for s in self._ip_stats.values()
            if s.banned_until and now < s.banned_until
        )

        return {
            "total_connections": len(self._connections),
            "unique_ips": len(self._ip_stats),
            "banned_ips": banned_count,
            "max_connections": self._config.max_global_connections
        }


class SuspiciousActivityDetector:
    """
    Erkennt verdächtige Aktivitätsmuster.

    Analysiert Verbindungs- und Anfragemuster für DDoS-Erkennung.
    """

    def __init__(self, config: DDoSConfig | None = None) -> None:
        """
        Initialisiert den Detektor.

        Args:
            config: DDoS-Konfiguration
        """
        self._config = config or DDoSConfig()
        self._events: deque[tuple[float, str, str]] = deque(maxlen=10000)
        self._patterns: dict[str, int] = {}

    def record_event(self, ip: str, event_type: str) -> None:
        """
        Zeichnet ein Ereignis auf.

        Args:
            ip: Quell-IP
            event_type: Art des Ereignisses
        """
        now = time.monotonic()
        self._events.append((now, ip, event_type))

        key = f"{ip}:{event_type}"
        self._patterns[key] = self._patterns.get(key, 0) + 1

    def analyze(self, ip: str) -> ThreatLevel:
        """
        Analysiert die Bedrohungsstufe für eine IP.

        Args:
            ip: Die zu analysierende IP

        Returns:
            Bedrohungsstufe
        """
        now = time.monotonic()
        window_start = now - self._config.monitoring_window

        # Zähle Events in Zeitfenster
        ip_events = [
            e for e in self._events
            if e[0] > window_start and e[1] == ip
        ]

        event_count = len(ip_events)

        if event_count == 0:
            return ThreatLevel.NONE
        elif event_count < 10:
            return ThreatLevel.LOW
        elif event_count < 50:
            return ThreatLevel.MEDIUM
        elif event_count < 100:
            return ThreatLevel.HIGH
        else:
            return ThreatLevel.CRITICAL

    def get_suspicious_ips(self, min_level: ThreatLevel = ThreatLevel.MEDIUM) -> list[tuple[str, ThreatLevel]]:
        """
        Gibt verdächtige IPs zurück.

        Args:
            min_level: Minimale Bedrohungsstufe

        Returns:
            Liste von (IP, ThreatLevel) Tupeln
        """
        now = time.monotonic()
        window_start = now - self._config.monitoring_window

        # Sammle IPs aus aktuellem Fenster
        ips: set[str] = set()
        for timestamp, ip, _ in self._events:
            if timestamp > window_start:
                ips.add(ip)

        # Analysiere jede IP
        suspicious = []
        for ip in ips:
            level = self.analyze(ip)
            if level >= min_level:
                suspicious.append((ip, level))

        # Sortiere nach Bedrohungsstufe (höchste zuerst)
        suspicious.sort(key=lambda x: x[1], reverse=True)

        return suspicious

    def cleanup(self) -> None:
        """Bereinigt alte Daten."""
        now = time.monotonic()
        window_start = now - self._config.monitoring_window * 2

        # Entferne alte Events
        while self._events and self._events[0][0] < window_start:
            self._events.popleft()


class DDoSProtection:
    """
    Hauptklasse für DDoS-Schutz.

    Kombiniert Connection-Tracking und Aktivitäts-Analyse.

    Example:
        protection = DDoSProtection(DDoSConfig(
            max_connections_per_ip=10,
            max_connection_rate=5.0
        ))

        # Bei neuer Verbindung
        conn_id, allowed = await protection.on_connect("192.168.1.100")
        if not allowed:
            # Verbindung ablehnen
            return

        # Bei Aktivität
        await protection.on_activity(conn_id, bytes_received=1024)

        # Bei Trennung
        await protection.on_disconnect(conn_id)

        # Periodische Prüfung
        for ip, level in protection.get_threats():
            if level >= ThreatLevel.HIGH:
                protection.ban_ip(ip)
    """

    def __init__(self, config: DDoSConfig | None = None) -> None:
        """
        Initialisiert den DDoS-Schutz.

        Args:
            config: DDoS-Konfiguration
        """
        self._config = config or DDoSConfig()
        self._tracker = ConnectionTracker(self._config)
        self._detector = SuspiciousActivityDetector(self._config)
        self._lock = asyncio.Lock()
        self._monitoring_task: asyncio.Task | None = None
        self._callbacks: list[Callable[[str, ThreatLevel], None]] = []

    @property
    def tracker(self) -> ConnectionTracker:
        """Connection-Tracker."""
        return self._tracker

    @property
    def detector(self) -> SuspiciousActivityDetector:
        """Aktivitäts-Detektor."""
        return self._detector

    def on_threat(
        self,
        callback: Callable[[str, ThreatLevel], None]
    ) -> Callable[[str, ThreatLevel], None]:
        """
        Registriert einen Callback für Bedrohungen.

        Args:
            callback: Funktion(ip, level)

        Returns:
            Der registrierte Callback
        """
        self._callbacks.append(callback)
        return callback

    async def on_connect(self, ip: str) -> tuple[str, bool]:
        """
        Wird bei neuer Verbindung aufgerufen.

        Args:
            ip: IP-Adresse

        Returns:
            (connection_id, erlaubt)
        """
        async with self._lock:
            conn_id, allowed = self._tracker.register_connection(ip)

            if not allowed:
                self._detector.record_event(ip, "connection_denied")

                # Prüfe Bedrohungsstufe
                level = self._detector.analyze(ip)
                if level >= ThreatLevel.HIGH:
                    self._tracker.ban_ip(ip)
                    self._notify_threat(ip, level)

            return conn_id, allowed

    async def on_disconnect(self, conn_id: str) -> None:
        """
        Wird bei Verbindungstrennung aufgerufen.

        Args:
            conn_id: Verbindungs-ID
        """
        async with self._lock:
            self._tracker.unregister_connection(conn_id)

    async def on_activity(
        self,
        conn_id: str,
        bytes_received: int = 0,
        bytes_sent: int = 0
    ) -> None:
        """
        Wird bei Verbindungsaktivität aufgerufen.

        Args:
            conn_id: Verbindungs-ID
            bytes_received: Empfangene Bytes
            bytes_sent: Gesendete Bytes
        """
        async with self._lock:
            self._tracker.update_activity(conn_id, bytes_received, bytes_sent)

    def is_allowed(self, ip: str) -> bool:
        """
        Prüft, ob eine IP erlaubt ist.

        Args:
            ip: IP-Adresse

        Returns:
            True wenn nicht gesperrt
        """
        return not self._tracker.is_banned(ip)

    def ban_ip(self, ip: str, duration: float | None = None) -> None:
        """
        Sperrt eine IP.

        Args:
            ip: IP-Adresse
            duration: Optionale Sperrdauer
        """
        self._tracker.ban_ip(ip, duration)

    def unban_ip(self, ip: str) -> bool:
        """
        Entsperrt eine IP.

        Args:
            ip: IP-Adresse

        Returns:
            True wenn IP gesperrt war
        """
        return self._tracker.unban_ip(ip)

    def get_threats(
        self,
        min_level: ThreatLevel = ThreatLevel.MEDIUM
    ) -> list[tuple[str, ThreatLevel]]:
        """
        Gibt aktuelle Bedrohungen zurück.

        Args:
            min_level: Minimale Bedrohungsstufe

        Returns:
            Liste von (IP, ThreatLevel)
        """
        return self._detector.get_suspicious_ips(min_level)

    def _notify_threat(self, ip: str, level: ThreatLevel) -> None:
        """Benachrichtigt über eine Bedrohung."""
        for callback in self._callbacks:
            try:
                callback(ip, level)
            except Exception:
                pass

    async def _monitoring_loop(self) -> None:
        """Hintergrund-Monitoring-Loop."""
        while True:
            await asyncio.sleep(10.0)  # Alle 10 Sekunden

            # Bereinigung
            self._detector.cleanup()

            # Prüfe auf kritische Bedrohungen
            for ip, level in self.get_threats(ThreatLevel.HIGH):
                if level >= ThreatLevel.CRITICAL:
                    self.ban_ip(ip)
                    self._notify_threat(ip, level)

    async def start(self) -> None:
        """Startet das DDoS-Monitoring."""
        if self._monitoring_task is None:
            self._monitoring_task = asyncio.create_task(
                self._monitoring_loop()
            )

    async def stop(self) -> None:
        """Stoppt das DDoS-Monitoring."""
        if self._monitoring_task is not None:
            self._monitoring_task.cancel()
            try:
                await self._monitoring_task
            except asyncio.CancelledError:
                pass
            self._monitoring_task = None

    def get_statistics(self) -> dict[str, Any]:
        """Gibt Statistiken zurück."""
        threats = self.get_threats(ThreatLevel.LOW)

        return {
            "connections": self._tracker.get_statistics(),
            "threats": {
                "low": sum(1 for _, l in threats if l == ThreatLevel.LOW),
                "medium": sum(1 for _, l in threats if l == ThreatLevel.MEDIUM),
                "high": sum(1 for _, l in threats if l == ThreatLevel.HIGH),
                "critical": sum(1 for _, l in threats if l == ThreatLevel.CRITICAL),
            },
            "config": {
                "max_connections_per_ip": self._config.max_connections_per_ip,
                "max_connection_rate": self._config.max_connection_rate,
                "ban_duration": self._config.ban_duration,
            }
        }