Trixy7/trixy_core/network/tls/manager.py

# -*- coding: utf-8 -*-
"""
TLS-Manager für Zertifikatsverwaltung.

Bietet Funktionen zum Laden, Prüfen und Verwalten von Zertifikaten.
"""

import ssl
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

from trixy_core.network.tls.config import TLSConfig


@dataclass
class CertificateInfo:
    """
    Informationen über ein Zertifikat.

    Attributes:
        subject: Subject des Zertifikats
        issuer: Aussteller des Zertifikats
        serial_number: Seriennummer
        not_before: Gültig ab
        not_after: Gültig bis
        version: Zertifikats-Version
        san: Subject Alternative Names
        fingerprint_sha256: SHA-256 Fingerprint
    """
    subject: dict[str, str]
    issuer: dict[str, str]
    serial_number: int
    not_before: datetime
    not_after: datetime
    version: int
    san: list[str]
    fingerprint_sha256: str

    @property
    def common_name(self) -> str:
        """Common Name (CN) des Subjects."""
        return self.subject.get("CN", "")

    @property
    def is_valid(self) -> bool:
        """Prüft, ob das Zertifikat zeitlich gültig ist."""
        now = datetime.now()
        return self.not_before <= now <= self.not_after

    @property
    def days_until_expiry(self) -> int:
        """Tage bis zum Ablauf."""
        delta = self.not_after - datetime.now()
        return delta.days

    @property
    def is_self_signed(self) -> bool:
        """Prüft, ob das Zertifikat selbstsigniert ist."""
        return self.subject == self.issuer


class TLSManager:
    """
    Manager für TLS-Operationen und Zertifikatsverwaltung.

    Example:
        manager = TLSManager(TLSConfig(
            enabled=True,
            cert_file="certs/server.crt",
            key_file="certs/server.key"
        ))

        # Server-Kontext erstellen
        context = manager.get_server_context()

        # Zertifikat-Info abrufen
        info = manager.get_certificate_info()
        print(f"Gültig bis: {info.not_after}")

        # Prüfung auf baldigen Ablauf
        if manager.is_expiring_soon(days=30):
            print("Zertifikat läuft bald ab!")
    """

    def __init__(self, config: TLSConfig) -> None:
        """
        Initialisiert den TLS-Manager.

        Args:
            config: TLS-Konfiguration
        """
        self._config = config
        self._server_context: ssl.SSLContext | None = None
        self._client_context: ssl.SSLContext | None = None
        self._cert_info: CertificateInfo | None = None

    @property
    def config(self) -> TLSConfig:
        """Die TLS-Konfiguration."""
        return self._config

    @property
    def is_enabled(self) -> bool:
        """Prüft, ob TLS aktiviert ist."""
        return self._config.enabled

    def get_server_context(self) -> ssl.SSLContext | None:
        """
        Gibt den Server-SSL-Kontext zurück.

        Returns:
            SSL-Kontext oder None wenn TLS deaktiviert
        """
        if not self._config.enabled:
            return None

        if self._server_context is None:
            self._server_context = self._config.create_server_context()

        return self._server_context

    def get_client_context(self) -> ssl.SSLContext | None:
        """
        Gibt den Client-SSL-Kontext zurück.

        Returns:
            SSL-Kontext oder None wenn TLS deaktiviert
        """
        if not self._config.enabled:
            return None

        if self._client_context is None:
            self._client_context = self._config.create_client_context()

        return self._client_context

    def reload(self) -> None:
        """Lädt die SSL-Kontexte neu."""
        self._server_context = None
        self._client_context = None
        self._cert_info = None

    def get_certificate_info(self) -> CertificateInfo | None:
        """
        Gibt Informationen über das Server-Zertifikat zurück.

        Returns:
            CertificateInfo oder None
        """
        if not self._config.enabled or not self._config.cert_file:
            return None

        if self._cert_info is not None:
            return self._cert_info

        try:
            self._cert_info = self._load_certificate_info(
                Path(self._config.cert_file)
            )
            return self._cert_info
        except Exception:
            return None

    def _load_certificate_info(self, cert_path: Path) -> CertificateInfo:
        """
        Lädt Zertifikats-Informationen aus einer PEM-Datei.

        Args:
            cert_path: Pfad zur Zertifikatsdatei

        Returns:
            CertificateInfo
        """
        import hashlib

        # Zertifikat laden via temporären Kontext
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE

        # PEM lesen
        with open(cert_path, 'rb') as f:
            pem_data = f.read()

        # Zertifikat parsen
        from ssl import PEM_cert_to_DER_cert, DER_cert_to_PEM_cert
        import re

        # PEM-Block extrahieren
        pem_match = re.search(
            b'-----BEGIN CERTIFICATE-----(.+?)-----END CERTIFICATE-----',
            pem_data,
            re.DOTALL
        )

        if not pem_match:
            raise ValueError("Kein gültiges PEM-Zertifikat gefunden")

        # DER für Fingerprint
        import base64
        der_data = base64.b64decode(pem_match.group(1))
        fingerprint = hashlib.sha256(der_data).hexdigest().upper()
        fingerprint = ':'.join(fingerprint[i:i+2] for i in range(0, len(fingerprint), 2))

        # Zertifikat über ssl laden für Details
        # Wir nutzen einen Workaround: Socket mit getpeercert simulieren
        # Alternative: pyOpenSSL oder cryptography nutzen

        # Einfaches Parsing für grundlegende Info
        # Für vollständige Unterstützung wäre cryptography-Bibliothek nötig

        cert_info = CertificateInfo(
            subject={"CN": self._extract_cn_from_pem(pem_data)},
            issuer={"CN": self._extract_cn_from_pem(pem_data, issuer=True)},
            serial_number=0,
            not_before=datetime.now(),
            not_after=datetime.now(),
            version=3,
            san=[],
            fingerprint_sha256=fingerprint
        )

        return cert_info

    def _extract_cn_from_pem(self, pem_data: bytes, issuer: bool = False) -> str:
        """
        Extrahiert den CN aus einem PEM-Zertifikat.

        Dies ist eine vereinfachte Implementierung.
        Für vollständige Unterstützung sollte cryptography verwendet werden.

        Args:
            pem_data: PEM-Daten
            issuer: True für Issuer-CN, False für Subject-CN

        Returns:
            Common Name oder "Unknown"
        """
        # Sehr einfache Extraktion - nicht für Produktion geeignet
        # In einer echten Implementierung würde man cryptography nutzen
        import re

        pattern = rb'CN\s*=\s*([^,/\n]+)'
        matches = re.findall(pattern, pem_data)

        if matches:
            # Bei selbstsignierten Zertifikaten ist Subject = Issuer
            return matches[0].decode('utf-8', errors='replace').strip()

        return "Unknown"

    def is_expiring_soon(self, days: int = 30) -> bool:
        """
        Prüft, ob das Zertifikat bald abläuft.

        Args:
            days: Anzahl Tage

        Returns:
            True wenn Ablauf innerhalb der angegebenen Tage
        """
        info = self.get_certificate_info()
        if info is None:
            return False

        return info.days_until_expiry <= days

    def verify_chain(self, cert_path: Path, ca_path: Path | None = None) -> tuple[bool, str]:
        """
        Verifiziert die Zertifikatskette.

        Args:
            cert_path: Pfad zum Zertifikat
            ca_path: Optionaler Pfad zur CA

        Returns:
            (gültig, Fehlermeldung)
        """
        try:
            context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            context.verify_mode = ssl.CERT_REQUIRED
            context.check_hostname = False

            if ca_path:
                context.load_verify_locations(cafile=str(ca_path))
            else:
                context.load_default_certs()

            # Zertifikat laden
            context.load_cert_chain(certfile=str(cert_path))

            return True, ""

        except ssl.SSLError as e:
            return False, str(e)
        except FileNotFoundError as e:
            return False, f"Datei nicht gefunden: {e.filename}"
        except Exception as e:
            return False, str(e)

    def get_status(self) -> dict[str, Any]:
        """
        Gibt den Status des TLS-Managers zurück.

        Returns:
            Status-Dictionary
        """
        info = self.get_certificate_info()

        return {
            "enabled": self._config.enabled,
            "cert_file": str(self._config.cert_file) if self._config.cert_file else None,
            "min_version": self._config.min_version.name,
            "cipher_suite": self._config.cipher_suite.name,
            "certificate": {
                "common_name": info.common_name if info else None,
                "is_valid": info.is_valid if info else None,
                "days_until_expiry": info.days_until_expiry if info else None,
                "is_self_signed": info.is_self_signed if info else None,
                "fingerprint": info.fingerprint_sha256 if info else None,
            } if info else None
        }


def create_development_certificates(
    output_dir: Path,
    common_name: str = "localhost",
    days_valid: int = 365
) -> tuple[Path, Path]:
    """
    Erstellt selbstsignierte Entwicklungszertifikate.

    Nutzt den CertificateGenerator für die Erstellung.

    Args:
        output_dir: Ausgabeverzeichnis
        common_name: Common Name
        days_valid: Gültigkeitsdauer

    Returns:
        (cert_path, key_path)
    """
    from trixy_core.network.tls.generator import CertificateGenerator, CertificateRequest

    generator = CertificateGenerator()

    request = CertificateRequest(
        common_name=common_name,
        organization="Trixy Development",
        organizational_unit="Development",
        country="DE",
        days_valid=days_valid,
        san_dns=[common_name, "localhost"],
        san_ip=["127.0.0.1", "::1"]
    )

    cert_path, key_path = generator.generate_self_signed(
        request=request,
        output_dir=output_dir
    )

    return cert_path, key_path