Trixy7/trixy_core/network/tls/config.py

# -*- coding: utf-8 -*-
"""
TLS-Konfiguration für sichere Verbindungen.

Definiert TLS-Versionen, Cipher-Suites und Konfigurationsoptionen.
"""

import ssl
from dataclasses import dataclass, field
from enum import IntEnum, auto
from pathlib import Path
from typing import Any


class TLSVersion(IntEnum):
    """Unterstützte TLS-Versionen."""
    TLS_1_2 = auto()
    TLS_1_3 = auto()

    def to_ssl_version(self) -> int:
        """Konvertiert zu ssl.TLSVersion."""
        if self == TLSVersion.TLS_1_2:
            return ssl.TLSVersion.TLSv1_2
        return ssl.TLSVersion.TLSv1_3


class CipherSuite(IntEnum):
    """Cipher-Suite-Profile."""
    MODERN = auto()      # Nur moderne, sichere Ciphers
    INTERMEDIATE = auto()  # Balance zwischen Sicherheit und Kompatibilität
    COMPATIBLE = auto()   # Maximale Kompatibilität (weniger sicher)

    def get_ciphers(self) -> str:
        """Gibt die OpenSSL Cipher-String zurück."""
        if self == CipherSuite.MODERN:
            return (
                "TLS_AES_256_GCM_SHA384:"
                "TLS_CHACHA20_POLY1305_SHA256:"
                "TLS_AES_128_GCM_SHA256:"
                "ECDHE+AESGCM:"
                "DHE+AESGCM"
            )
        elif self == CipherSuite.INTERMEDIATE:
            return (
                "ECDHE+AESGCM:"
                "DHE+AESGCM:"
                "ECDHE+AES:"
                "DHE+AES:"
                "!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5"
            )
        else:  # COMPATIBLE
            return (
                "ECDHE+AESGCM:"
                "DHE+AESGCM:"
                "ECDHE+AES:"
                "DHE+AES:"
                "AES256-SHA:"
                "AES128-SHA:"
                "!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5"
            )


@dataclass
class TLSConfig:
    """
    Konfiguration für TLS-Verbindungen.

    Attributes:
        enabled: TLS aktiviert
        cert_file: Pfad zum Zertifikat
        key_file: Pfad zum privaten Schlüssel
        ca_file: Optionaler Pfad zur CA-Datei
        min_version: Minimale TLS-Version
        cipher_suite: Cipher-Suite-Profil
        verify_client: Client-Zertifikat prüfen
        require_client_cert: Client-Zertifikat erforderlich
        check_hostname: Hostname prüfen
        session_cache_size: Session-Cache-Größe (0 = deaktiviert)
        session_timeout: Session-Timeout in Sekunden
    """
    enabled: bool = False
    cert_file: str | Path = ""
    key_file: str | Path = ""
    ca_file: str | Path = ""
    min_version: TLSVersion = TLSVersion.TLS_1_2
    cipher_suite: CipherSuite = CipherSuite.MODERN
    verify_client: bool = False
    require_client_cert: bool = False
    check_hostname: bool = True
    session_cache_size: int = 1024
    session_timeout: int = 86400  # 24 Stunden

    def __post_init__(self) -> None:
        """Konvertiert Pfade zu Path-Objekten."""
        if isinstance(self.cert_file, str) and self.cert_file:
            self.cert_file = Path(self.cert_file)
        if isinstance(self.key_file, str) and self.key_file:
            self.key_file = Path(self.key_file)
        if isinstance(self.ca_file, str) and self.ca_file:
            self.ca_file = Path(self.ca_file)

    def is_valid(self) -> tuple[bool, str]:
        """
        Prüft, ob die Konfiguration gültig ist.

        Returns:
            (gültig, Fehlermeldung)
        """
        if not self.enabled:
            return True, ""

        if not self.cert_file:
            return False, "Zertifikat-Datei nicht angegeben"

        if not self.key_file:
            return False, "Schlüssel-Datei nicht angegeben"

        if isinstance(self.cert_file, Path) and not self.cert_file.exists():
            return False, f"Zertifikat-Datei nicht gefunden: {self.cert_file}"

        if isinstance(self.key_file, Path) and not self.key_file.exists():
            return False, f"Schlüssel-Datei nicht gefunden: {self.key_file}"

        if self.ca_file:
            if isinstance(self.ca_file, Path) and not self.ca_file.exists():
                return False, f"CA-Datei nicht gefunden: {self.ca_file}"

        return True, ""

    def create_server_context(self) -> ssl.SSLContext:
        """
        Erstellt einen SSL-Kontext für Server.

        Returns:
            Konfigurierter SSL-Kontext

        Raises:
            ValueError: Bei ungültiger Konfiguration
            FileNotFoundError: Wenn Zertifikate nicht gefunden werden
        """
        valid, error = self.is_valid()
        if not valid:
            raise ValueError(error)

        # Kontext erstellen
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

        # Minimale TLS-Version
        context.minimum_version = self.min_version.to_ssl_version()

        # Zertifikate laden
        context.load_cert_chain(
            certfile=str(self.cert_file),
            keyfile=str(self.key_file)
        )

        # CA für Client-Verifizierung
        if self.ca_file:
            context.load_verify_locations(cafile=str(self.ca_file))

        # Client-Verifizierung
        if self.require_client_cert:
            context.verify_mode = ssl.CERT_REQUIRED
        elif self.verify_client:
            context.verify_mode = ssl.CERT_OPTIONAL
        else:
            context.verify_mode = ssl.CERT_NONE

        # Cipher-Suites
        context.set_ciphers(self.cipher_suite.get_ciphers())

        # Session-Cache
        if self.session_cache_size > 0:
            context.options |= ssl.OP_NO_TICKET  # Deaktiviere Session-Tickets
            # Session-Cache wird automatisch verwaltet

        # Sicherheitsoptionen
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_TLSv1
        context.options |= ssl.OP_NO_TLSv1_1
        context.options |= ssl.OP_SINGLE_DH_USE
        context.options |= ssl.OP_SINGLE_ECDH_USE

        return context

    def create_client_context(self) -> ssl.SSLContext:
        """
        Erstellt einen SSL-Kontext für Clients.

        Returns:
            Konfigurierter SSL-Kontext
        """
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

        # Minimale TLS-Version
        context.minimum_version = self.min_version.to_ssl_version()

        # CA laden
        if self.ca_file:
            context.load_verify_locations(cafile=str(self.ca_file))
        else:
            context.load_default_certs()

        # Client-Zertifikat
        if self.cert_file and self.key_file:
            context.load_cert_chain(
                certfile=str(self.cert_file),
                keyfile=str(self.key_file)
            )

        # Hostname-Prüfung
        context.check_hostname = self.check_hostname
        context.verify_mode = ssl.CERT_REQUIRED if self.check_hostname else ssl.CERT_OPTIONAL

        # Cipher-Suites
        context.set_ciphers(self.cipher_suite.get_ciphers())

        # Sicherheitsoptionen
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_TLSv1
        context.options |= ssl.OP_NO_TLSv1_1

        return context

    def to_dict(self) -> dict[str, Any]:
        """Serialisiert die Konfiguration."""
        return {
            "enabled": self.enabled,
            "cert_file": str(self.cert_file) if self.cert_file else "",
            "key_file": str(self.key_file) if self.key_file else "",
            "ca_file": str(self.ca_file) if self.ca_file else "",
            "min_version": self.min_version.name,
            "cipher_suite": self.cipher_suite.name,
            "verify_client": self.verify_client,
            "require_client_cert": self.require_client_cert,
            "check_hostname": self.check_hostname,
            "session_cache_size": self.session_cache_size,
            "session_timeout": self.session_timeout,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "TLSConfig":
        """Erstellt eine Konfiguration aus einem Dictionary."""
        min_version = TLSVersion[data.get("min_version", "TLS_1_2")]
        cipher_suite = CipherSuite[data.get("cipher_suite", "MODERN")]

        return cls(
            enabled=data.get("enabled", False),
            cert_file=data.get("cert_file", ""),
            key_file=data.get("key_file", ""),
            ca_file=data.get("ca_file", ""),
            min_version=min_version,
            cipher_suite=cipher_suite,
            verify_client=data.get("verify_client", False),
            require_client_cert=data.get("require_client_cert", False),
            check_hostname=data.get("check_hostname", True),
            session_cache_size=data.get("session_cache_size", 1024),
            session_timeout=data.get("session_timeout", 86400),
        )