Trixy7/trixy_core/network/security/access_control.py

# -*- coding: utf-8 -*-
"""
Access Control mit Whitelist/Blacklist Unterstützung.

Bietet flexible Zugriffskontrolle basierend auf IP-Adressen,
CIDR-Bereichen und MAC-Adressen.
"""

import ipaddress
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum, auto
from typing import Any


class AccessDecision(IntEnum):
    """Entscheidung der Zugriffskontrolle."""
    ALLOW = auto()      # Zugriff erlaubt
    DENY = auto()       # Zugriff verweigert
    NEUTRAL = auto()    # Keine Entscheidung (nächste Regel prüfen)


@dataclass
class AccessRule:
    """
    Einzelne Zugriffsregel.

    Attributes:
        pattern: Das zu matchende Pattern (IP, CIDR, MAC, etc.)
        decision: Die Entscheidung bei Match
        description: Beschreibung der Regel
        expires_at: Optionales Ablaufdatum
        hit_count: Anzahl der Treffer
        created_at: Erstellungszeitpunkt
    """
    pattern: str
    decision: AccessDecision
    description: str = ""
    expires_at: datetime | None = None
    hit_count: int = 0
    created_at: datetime = field(default_factory=datetime.now)

    def is_expired(self) -> bool:
        """Prüft, ob die Regel abgelaufen ist."""
        if self.expires_at is None:
            return False
        return datetime.now() > self.expires_at

    def record_hit(self) -> None:
        """Zeichnet einen Treffer auf."""
        self.hit_count += 1


class Matcher(ABC):
    """Basisklasse für Pattern-Matcher."""

    @abstractmethod
    def matches(self, value: str) -> bool:
        """Prüft, ob der Wert dem Pattern entspricht."""
        ...


class IPMatcher(Matcher):
    """
    Matcher für einzelne IP-Adressen.

    Unterstützt IPv4 und IPv6.
    """

    def __init__(self, ip: str) -> None:
        """
        Initialisiert den IP-Matcher.

        Args:
            ip: Die zu matchende IP-Adresse
        """
        self._ip = ipaddress.ip_address(ip)

    def matches(self, value: str) -> bool:
        """Prüft, ob die IP-Adresse übereinstimmt."""
        try:
            check_ip = ipaddress.ip_address(value)
            return check_ip == self._ip
        except ValueError:
            return False


class CIDRMatcher(Matcher):
    """
    Matcher für CIDR-Bereiche (IP-Netzwerke).

    Beispiel: 192.168.1.0/24 matcht alle IPs von 192.168.1.0 bis 192.168.1.255
    """

    def __init__(self, cidr: str) -> None:
        """
        Initialisiert den CIDR-Matcher.

        Args:
            cidr: CIDR-Notation (z.B. "192.168.1.0/24")
        """
        self._network = ipaddress.ip_network(cidr, strict=False)

    def matches(self, value: str) -> bool:
        """Prüft, ob die IP im Netzwerk liegt."""
        try:
            check_ip = ipaddress.ip_address(value)
            return check_ip in self._network
        except ValueError:
            return False


class MACMatcher(Matcher):
    """
    Matcher für MAC-Adressen.

    Unterstützt verschiedene Formate:
    - AA:BB:CC:DD:EE:FF
    - AA-BB-CC-DD-EE-FF
    - AABBCCDDEEFF
    """

    # Regex für MAC-Adressen
    _MAC_PATTERN = re.compile(
        r'^([0-9A-Fa-f]{2}[:-]?){5}([0-9A-Fa-f]{2})$'
    )

    def __init__(self, mac: str) -> None:
        """
        Initialisiert den MAC-Matcher.

        Args:
            mac: Die zu matchende MAC-Adresse
        """
        self._mac = self._normalize_mac(mac)

    @staticmethod
    def _normalize_mac(mac: str) -> str:
        """Normalisiert eine MAC-Adresse zu Großbuchstaben ohne Trennzeichen."""
        return mac.upper().replace(":", "").replace("-", "")

    def matches(self, value: str) -> bool:
        """Prüft, ob die MAC-Adresse übereinstimmt."""
        if not self._MAC_PATTERN.match(value):
            return False
        return self._normalize_mac(value) == self._mac


class WildcardMatcher(Matcher):
    """
    Matcher mit Wildcard-Unterstützung.

    Unterstützt * als Platzhalter für beliebige Zeichen.
    Beispiel: "192.168.*.*" matcht alle IPs im 192.168.x.x Bereich
    """

    def __init__(self, pattern: str) -> None:
        """
        Initialisiert den Wildcard-Matcher.

        Args:
            pattern: Pattern mit optionalen Wildcards (*)
        """
        # Konvertiere Wildcard-Pattern zu Regex
        escaped = re.escape(pattern)
        regex_pattern = escaped.replace(r'\*', '.*')
        self._regex = re.compile(f'^{regex_pattern}$')

    def matches(self, value: str) -> bool:
        """Prüft, ob der Wert dem Pattern entspricht."""
        return bool(self._regex.match(value))


class AccessController:
    """
    Zentrale Zugriffskontrolle mit Whitelist/Blacklist.

    Verarbeitet Regeln in folgender Reihenfolge:
    1. Blacklist (wenn Match → DENY)
    2. Whitelist (wenn Match → ALLOW)
    3. Default-Entscheidung

    Example:
        controller = AccessController(default_allow=False)

        # Whitelist: Lokales Netzwerk erlauben
        controller.whitelist("192.168.1.0/24", "Lokales Netzwerk")

        # Blacklist: Bestimmte IP sperren
        controller.blacklist("192.168.1.100", "Bekannter Angreifer")

        # Prüfung
        if controller.check_ip("192.168.1.50") == AccessDecision.ALLOW:
            print("Zugriff erlaubt")
    """

    def __init__(self, default_allow: bool = False) -> None:
        """
        Initialisiert die Zugriffskontrolle.

        Args:
            default_allow: Standardentscheidung wenn keine Regel matcht
        """
        self._default_allow = default_allow
        self._whitelist: list[tuple[Matcher, AccessRule]] = []
        self._blacklist: list[tuple[Matcher, AccessRule]] = []
        self._mac_whitelist: list[tuple[Matcher, AccessRule]] = []
        self._mac_blacklist: list[tuple[Matcher, AccessRule]] = []

    @property
    def default_decision(self) -> AccessDecision:
        """Standard-Entscheidung wenn keine Regel matcht."""
        return AccessDecision.ALLOW if self._default_allow else AccessDecision.DENY

    def _create_ip_matcher(self, pattern: str) -> Matcher:
        """Erstellt den passenden Matcher für ein IP-Pattern."""
        if "/" in pattern:
            return CIDRMatcher(pattern)
        elif "*" in pattern:
            return WildcardMatcher(pattern)
        else:
            return IPMatcher(pattern)

    def whitelist(
        self,
        pattern: str,
        description: str = "",
        expires_at: datetime | None = None
    ) -> AccessRule:
        """
        Fügt eine IP/CIDR zur Whitelist hinzu.

        Args:
            pattern: IP-Adresse, CIDR oder Wildcard-Pattern
            description: Beschreibung der Regel
            expires_at: Optionales Ablaufdatum

        Returns:
            Die erstellte Regel
        """
        rule = AccessRule(
            pattern=pattern,
            decision=AccessDecision.ALLOW,
            description=description,
            expires_at=expires_at
        )
        matcher = self._create_ip_matcher(pattern)
        self._whitelist.append((matcher, rule))
        return rule

    def blacklist(
        self,
        pattern: str,
        description: str = "",
        expires_at: datetime | None = None
    ) -> AccessRule:
        """
        Fügt eine IP/CIDR zur Blacklist hinzu.

        Args:
            pattern: IP-Adresse, CIDR oder Wildcard-Pattern
            description: Beschreibung der Regel
            expires_at: Optionales Ablaufdatum

        Returns:
            Die erstellte Regel
        """
        rule = AccessRule(
            pattern=pattern,
            decision=AccessDecision.DENY,
            description=description,
            expires_at=expires_at
        )
        matcher = self._create_ip_matcher(pattern)
        self._blacklist.append((matcher, rule))
        return rule

    def whitelist_mac(
        self,
        mac: str,
        description: str = "",
        expires_at: datetime | None = None
    ) -> AccessRule:
        """
        Fügt eine MAC-Adresse zur Whitelist hinzu.

        Args:
            mac: MAC-Adresse
            description: Beschreibung der Regel
            expires_at: Optionales Ablaufdatum

        Returns:
            Die erstellte Regel
        """
        rule = AccessRule(
            pattern=mac,
            decision=AccessDecision.ALLOW,
            description=description,
            expires_at=expires_at
        )
        matcher = MACMatcher(mac)
        self._mac_whitelist.append((matcher, rule))
        return rule

    def blacklist_mac(
        self,
        mac: str,
        description: str = "",
        expires_at: datetime | None = None
    ) -> AccessRule:
        """
        Fügt eine MAC-Adresse zur Blacklist hinzu.

        Args:
            mac: MAC-Adresse
            description: Beschreibung der Regel
            expires_at: Optionales Ablaufdatum

        Returns:
            Die erstellte Regel
        """
        rule = AccessRule(
            pattern=mac,
            decision=AccessDecision.DENY,
            description=description,
            expires_at=expires_at
        )
        matcher = MACMatcher(mac)
        self._mac_blacklist.append((matcher, rule))
        return rule

    def check_ip(self, ip: str) -> AccessDecision:
        """
        Prüft die Zugriffsberechtigung für eine IP-Adresse.

        Reihenfolge: Blacklist → Whitelist → Default

        Args:
            ip: Die zu prüfende IP-Adresse

        Returns:
            AccessDecision (ALLOW oder DENY)
        """
        # Blacklist zuerst prüfen
        for matcher, rule in self._blacklist:
            if rule.is_expired():
                continue
            if matcher.matches(ip):
                rule.record_hit()
                return AccessDecision.DENY

        # Dann Whitelist prüfen
        for matcher, rule in self._whitelist:
            if rule.is_expired():
                continue
            if matcher.matches(ip):
                rule.record_hit()
                return AccessDecision.ALLOW

        # Default-Entscheidung
        return self.default_decision

    def check_mac(self, mac: str) -> AccessDecision:
        """
        Prüft die Zugriffsberechtigung für eine MAC-Adresse.

        Args:
            mac: Die zu prüfende MAC-Adresse

        Returns:
            AccessDecision (ALLOW oder DENY)
        """
        # Blacklist zuerst prüfen
        for matcher, rule in self._mac_blacklist:
            if rule.is_expired():
                continue
            if matcher.matches(mac):
                rule.record_hit()
                return AccessDecision.DENY

        # Dann Whitelist prüfen
        for matcher, rule in self._mac_whitelist:
            if rule.is_expired():
                continue
            if matcher.matches(mac):
                rule.record_hit()
                return AccessDecision.ALLOW

        # Default-Entscheidung
        return self.default_decision

    def check(self, ip: str, mac: str | None = None) -> AccessDecision:
        """
        Kombinierte Prüfung von IP und optional MAC.

        Beide müssen erlaubt sein (wenn MAC angegeben).

        Args:
            ip: IP-Adresse
            mac: Optionale MAC-Adresse

        Returns:
            AccessDecision (ALLOW oder DENY)
        """
        ip_decision = self.check_ip(ip)

        if ip_decision == AccessDecision.DENY:
            return AccessDecision.DENY

        if mac is not None:
            mac_decision = self.check_mac(mac)
            if mac_decision == AccessDecision.DENY:
                return AccessDecision.DENY

        return AccessDecision.ALLOW

    def remove_from_whitelist(self, pattern: str) -> bool:
        """Entfernt ein Pattern von der IP-Whitelist."""
        for i, (_, rule) in enumerate(self._whitelist):
            if rule.pattern == pattern:
                self._whitelist.pop(i)
                return True
        return False

    def remove_from_blacklist(self, pattern: str) -> bool:
        """Entfernt ein Pattern von der IP-Blacklist."""
        for i, (_, rule) in enumerate(self._blacklist):
            if rule.pattern == pattern:
                self._blacklist.pop(i)
                return True
        return False

    def clear_expired(self) -> int:
        """
        Entfernt alle abgelaufenen Regeln.

        Returns:
            Anzahl entfernter Regeln
        """
        count = 0

        for rules_list in (
            self._whitelist,
            self._blacklist,
            self._mac_whitelist,
            self._mac_blacklist
        ):
            expired = [
                (m, r) for m, r in rules_list if r.is_expired()
            ]
            for item in expired:
                rules_list.remove(item)
                count += 1

        return count

    def get_statistics(self) -> dict[str, Any]:
        """
        Gibt Statistiken über die Zugriffskontrolle zurück.

        Returns:
            Dictionary mit Statistiken
        """
        return {
            "whitelist_count": len(self._whitelist),
            "blacklist_count": len(self._blacklist),
            "mac_whitelist_count": len(self._mac_whitelist),
            "mac_blacklist_count": len(self._mac_blacklist),
            "default_allow": self._default_allow,
            "whitelist_hits": sum(r.hit_count for _, r in self._whitelist),
            "blacklist_hits": sum(r.hit_count for _, r in self._blacklist),
        }

    def to_dict(self) -> dict[str, Any]:
        """
        Serialisiert die Regeln zu einem Dictionary.

        Returns:
            Dictionary mit allen Regeln
        """
        def rule_to_dict(rule: AccessRule) -> dict[str, Any]:
            return {
                "pattern": rule.pattern,
                "decision": rule.decision.name,
                "description": rule.description,
                "expires_at": rule.expires_at.isoformat() if rule.expires_at else None,
                "hit_count": rule.hit_count,
                "created_at": rule.created_at.isoformat(),
            }

        return {
            "default_allow": self._default_allow,
            "ip_whitelist": [rule_to_dict(r) for _, r in self._whitelist],
            "ip_blacklist": [rule_to_dict(r) for _, r in self._blacklist],
            "mac_whitelist": [rule_to_dict(r) for _, r in self._mac_whitelist],
            "mac_blacklist": [rule_to_dict(r) for _, r in self._mac_blacklist],
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "AccessController":
        """
        Erstellt einen AccessController aus einem Dictionary.

        Args:
            data: Serialisierte Regeln

        Returns:
            Neuer AccessController
        """
        controller = cls(default_allow=data.get("default_allow", False))

        for rule_data in data.get("ip_whitelist", []):
            expires = None
            if rule_data.get("expires_at"):
                expires = datetime.fromisoformat(rule_data["expires_at"])
            controller.whitelist(
                rule_data["pattern"],
                rule_data.get("description", ""),
                expires
            )

        for rule_data in data.get("ip_blacklist", []):
            expires = None
            if rule_data.get("expires_at"):
                expires = datetime.fromisoformat(rule_data["expires_at"])
            controller.blacklist(
                rule_data["pattern"],
                rule_data.get("description", ""),
                expires
            )

        for rule_data in data.get("mac_whitelist", []):
            expires = None
            if rule_data.get("expires_at"):
                expires = datetime.fromisoformat(rule_data["expires_at"])
            controller.whitelist_mac(
                rule_data["pattern"],
                rule_data.get("description", ""),
                expires
            )

        for rule_data in data.get("mac_blacklist", []):
            expires = None
            if rule_data.get("expires_at"):
                expires = datetime.fromisoformat(rule_data["expires_at"])
            controller.blacklist_mac(
                rule_data["pattern"],
                rule_data.get("description", ""),
                expires
            )

        return controller