# -*- 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