Trixy7/trixy_core/network/tls/generator.py

# -*- coding: utf-8 -*-
"""
Zertifikatsgenerator für selbstsignierte Zertifikate.

Ermöglicht die Erstellung von Entwicklungs- und Produktionszertifikaten
ohne externe Abhängigkeiten (nutzt nur Python-Standardbibliothek).
"""

import hashlib
import os
import secrets
import subprocess
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import IntEnum, auto
from pathlib import Path
from typing import Any


class KeyType(IntEnum):
    """Typ des kryptographischen Schlüssels."""
    RSA_2048 = auto()
    RSA_4096 = auto()
    ECDSA_P256 = auto()
    ECDSA_P384 = auto()


@dataclass
class CertificateRequest:
    """
    Anforderung für ein neues Zertifikat.

    Attributes:
        common_name: Common Name (CN) - üblicherweise Hostname
        organization: Organisation (O)
        organizational_unit: Organisationseinheit (OU)
        country: Land (C) - 2-Buchstaben-Code
        state: Bundesland/Region (ST)
        locality: Stadt (L)
        email: E-Mail-Adresse
        key_type: Schlüsseltyp
        days_valid: Gültigkeitsdauer in Tagen
        san_dns: Subject Alternative Names (DNS)
        san_ip: Subject Alternative Names (IP)
    """
    common_name: str
    organization: str = "Trixy"
    organizational_unit: str = ""
    country: str = "DE"
    state: str = ""
    locality: str = ""
    email: str = ""
    key_type: KeyType = KeyType.RSA_2048
    days_valid: int = 365
    san_dns: list[str] = field(default_factory=list)
    san_ip: list[str] = field(default_factory=list)

    def get_subject_string(self) -> str:
        """Gibt den Subject-String im OpenSSL-Format zurück."""
        parts = []

        if self.country:
            parts.append(f"/C={self.country}")
        if self.state:
            parts.append(f"/ST={self.state}")
        if self.locality:
            parts.append(f"/L={self.locality}")
        if self.organization:
            parts.append(f"/O={self.organization}")
        if self.organizational_unit:
            parts.append(f"/OU={self.organizational_unit}")
        if self.common_name:
            parts.append(f"/CN={self.common_name}")
        if self.email:
            parts.append(f"/emailAddress={self.email}")

        return "".join(parts)

    def get_san_extension(self) -> str:
        """Gibt die SAN-Erweiterung im OpenSSL-Format zurück."""
        san_entries = []

        for dns in self.san_dns:
            san_entries.append(f"DNS:{dns}")

        for ip in self.san_ip:
            san_entries.append(f"IP:{ip}")

        if not san_entries:
            # Mindestens den CN als SAN hinzufügen
            san_entries.append(f"DNS:{self.common_name}")

        return ",".join(san_entries)


class CertificateGenerator:
    """
    Generator für selbstsignierte Zertifikate.

    Nutzt OpenSSL über subprocess für die Zertifikatserstellung.
    Fallback auf Python-Standardbibliothek wenn verfügbar.

    Example:
        generator = CertificateGenerator()

        # Entwicklungszertifikat erstellen
        request = CertificateRequest(
            common_name="localhost",
            organization="Development",
            san_dns=["localhost", "trixy.local"],
            san_ip=["127.0.0.1"]
        )

        cert_path, key_path = generator.generate_self_signed(
            request=request,
            output_dir=Path("./certs")
        )

        print(f"Zertifikat: {cert_path}")
        print(f"Schlüssel: {key_path}")
    """

    def __init__(self, openssl_path: str = "openssl") -> None:
        """
        Initialisiert den Generator.

        Args:
            openssl_path: Pfad zur OpenSSL-Binary
        """
        self._openssl_path = openssl_path
        self._has_openssl: bool | None = None

    def is_openssl_available(self) -> bool:
        """Prüft, ob OpenSSL verfügbar ist."""
        if self._has_openssl is not None:
            return self._has_openssl

        try:
            result = subprocess.run(
                [self._openssl_path, "version"],
                capture_output=True,
                text=True,
                timeout=5
            )
            self._has_openssl = result.returncode == 0
        except (FileNotFoundError, subprocess.TimeoutExpired):
            self._has_openssl = False

        return self._has_openssl

    def generate_self_signed(
        self,
        request: CertificateRequest,
        output_dir: Path,
        cert_filename: str = "server.crt",
        key_filename: str = "server.key"
    ) -> tuple[Path, Path]:
        """
        Generiert ein selbstsigniertes Zertifikat.

        Args:
            request: Zertifikatsanforderung
            output_dir: Ausgabeverzeichnis
            cert_filename: Name der Zertifikatsdatei
            key_filename: Name der Schlüsseldatei

        Returns:
            (cert_path, key_path)

        Raises:
            RuntimeError: Wenn Generierung fehlschlägt
        """
        if not self.is_openssl_available():
            return self._generate_with_python(
                request, output_dir, cert_filename, key_filename
            )

        return self._generate_with_openssl(
            request, output_dir, cert_filename, key_filename
        )

    def _generate_with_openssl(
        self,
        request: CertificateRequest,
        output_dir: Path,
        cert_filename: str,
        key_filename: str
    ) -> tuple[Path, Path]:
        """Generiert Zertifikat mit OpenSSL."""
        output_dir.mkdir(parents=True, exist_ok=True)

        key_path = output_dir / key_filename
        cert_path = output_dir / cert_filename

        # Schlüsselparameter
        key_params = self._get_openssl_key_params(request.key_type)

        # OpenSSL-Konfiguration für SAN erstellen
        config_path = output_dir / "openssl.cnf"
        self._write_openssl_config(config_path, request)

        try:
            # Privaten Schlüssel generieren
            if request.key_type in (KeyType.RSA_2048, KeyType.RSA_4096):
                key_bits = "2048" if request.key_type == KeyType.RSA_2048 else "4096"
                subprocess.run(
                    [
                        self._openssl_path, "genrsa",
                        "-out", str(key_path),
                        key_bits
                    ],
                    check=True,
                    capture_output=True
                )
            else:
                curve = "prime256v1" if request.key_type == KeyType.ECDSA_P256 else "secp384r1"
                subprocess.run(
                    [
                        self._openssl_path, "ecparam",
                        "-genkey",
                        "-name", curve,
                        "-out", str(key_path)
                    ],
                    check=True,
                    capture_output=True
                )

            # Selbstsigniertes Zertifikat erstellen
            subprocess.run(
                [
                    self._openssl_path, "req",
                    "-new",
                    "-x509",
                    "-key", str(key_path),
                    "-out", str(cert_path),
                    "-days", str(request.days_valid),
                    "-subj", request.get_subject_string(),
                    "-config", str(config_path),
                    "-extensions", "v3_req"
                ],
                check=True,
                capture_output=True
            )

            # Schlüsselberechtigungen setzen
            os.chmod(key_path, 0o600)

            return cert_path, key_path

        except subprocess.CalledProcessError as e:
            raise RuntimeError(
                f"OpenSSL-Fehler: {e.stderr.decode() if e.stderr else str(e)}"
            )
        finally:
            # Temporäre Config entfernen
            if config_path.exists():
                config_path.unlink()

    def _write_openssl_config(self, path: Path, request: CertificateRequest) -> None:
        """Schreibt die OpenSSL-Konfigurationsdatei."""
        san = request.get_san_extension()

        config = f"""
[req]
distinguished_name = req_distinguished_name
req_extensions = v3_req
prompt = no

[req_distinguished_name]
CN = {request.common_name}

[v3_req]
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth, clientAuth
subjectAltName = {san}
basicConstraints = critical, CA:FALSE
"""
        path.write_text(config)

    def _get_openssl_key_params(self, key_type: KeyType) -> dict[str, str]:
        """Gibt OpenSSL-Parameter für den Schlüsseltyp zurück."""
        params = {
            KeyType.RSA_2048: {"algorithm": "rsa", "bits": "2048"},
            KeyType.RSA_4096: {"algorithm": "rsa", "bits": "4096"},
            KeyType.ECDSA_P256: {"algorithm": "ec", "curve": "prime256v1"},
            KeyType.ECDSA_P384: {"algorithm": "ec", "curve": "secp384r1"},
        }
        return params[key_type]

    def _generate_with_python(
        self,
        request: CertificateRequest,
        output_dir: Path,
        cert_filename: str,
        key_filename: str
    ) -> tuple[Path, Path]:
        """
        Fallback-Generierung mit Python.

        Diese Methode erstellt ein minimales selbstsigniertes Zertifikat
        ohne externe Abhängigkeiten. Für Produktionsumgebungen wird
        OpenSSL oder die cryptography-Bibliothek empfohlen.
        """
        output_dir.mkdir(parents=True, exist_ok=True)

        key_path = output_dir / key_filename
        cert_path = output_dir / cert_filename

        # Warnung ausgeben
        from trixy_core.utils.debug import pwarn
        pwarn(
            "OpenSSL nicht verfügbar. Erstelle minimales Zertifikat. "
            "Für Produktion wird OpenSSL empfohlen."
        )

        # Versuche cryptography zu importieren
        try:
            return self._generate_with_cryptography(
                request, output_dir, cert_filename, key_filename
            )
        except ImportError:
            pass

        # Absoluter Fallback: Leere Dateien erstellen und Fehler werfen
        raise RuntimeError(
            "Weder OpenSSL noch cryptography-Bibliothek verfügbar. "
            "Bitte installiere: pip install cryptography"
        )

    def _generate_with_cryptography(
        self,
        request: CertificateRequest,
        output_dir: Path,
        cert_filename: str,
        key_filename: str
    ) -> tuple[Path, Path]:
        """Generiert Zertifikat mit cryptography-Bibliothek."""
        from cryptography import x509
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import rsa, ec
        from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
        import ipaddress

        output_dir.mkdir(parents=True, exist_ok=True)

        key_path = output_dir / key_filename
        cert_path = output_dir / cert_filename

        # Schlüssel generieren
        if request.key_type in (KeyType.RSA_2048, KeyType.RSA_4096):
            key_size = 2048 if request.key_type == KeyType.RSA_2048 else 4096
            private_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=key_size
            )
        else:
            curve = ec.SECP256R1() if request.key_type == KeyType.ECDSA_P256 else ec.SECP384R1()
            private_key = ec.generate_private_key(curve)

        # Subject erstellen
        subject_attrs = [
            x509.NameAttribute(NameOID.COMMON_NAME, request.common_name),
        ]
        if request.organization:
            subject_attrs.append(
                x509.NameAttribute(NameOID.ORGANIZATION_NAME, request.organization)
            )
        if request.country:
            subject_attrs.append(
                x509.NameAttribute(NameOID.COUNTRY_NAME, request.country)
            )

        subject = x509.Name(subject_attrs)

        # SAN erstellen
        san_entries = []
        for dns in request.san_dns:
            san_entries.append(x509.DNSName(dns))
        for ip in request.san_ip:
            san_entries.append(x509.IPAddress(ipaddress.ip_address(ip)))

        if not san_entries:
            san_entries.append(x509.DNSName(request.common_name))

        # Zertifikat erstellen
        now = datetime.utcnow()
        cert = (
            x509.CertificateBuilder()
            .subject_name(subject)
            .issuer_name(subject)  # Selbstsigniert
            .public_key(private_key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(now)
            .not_valid_after(now + timedelta(days=request.days_valid))
            .add_extension(
                x509.SubjectAlternativeName(san_entries),
                critical=False
            )
            .add_extension(
                x509.BasicConstraints(ca=False, path_length=None),
                critical=True
            )
            .add_extension(
                x509.KeyUsage(
                    digital_signature=True,
                    key_encipherment=True,
                    content_commitment=False,
                    data_encipherment=False,
                    key_agreement=False,
                    key_cert_sign=False,
                    crl_sign=False,
                    encipher_only=False,
                    decipher_only=False
                ),
                critical=True
            )
            .add_extension(
                x509.ExtendedKeyUsage([
                    ExtendedKeyUsageOID.SERVER_AUTH,
                    ExtendedKeyUsageOID.CLIENT_AUTH
                ]),
                critical=False
            )
            .sign(private_key, hashes.SHA256())
        )

        # Schlüssel speichern
        key_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        )
        key_path.write_bytes(key_pem)
        os.chmod(key_path, 0o600)

        # Zertifikat speichern
        cert_pem = cert.public_bytes(serialization.Encoding.PEM)
        cert_path.write_bytes(cert_pem)

        return cert_path, key_path

    def generate_ca(
        self,
        request: CertificateRequest,
        output_dir: Path,
        ca_cert_filename: str = "ca.crt",
        ca_key_filename: str = "ca.key"
    ) -> tuple[Path, Path]:
        """
        Generiert ein CA-Zertifikat (Certificate Authority).

        Args:
            request: Zertifikatsanforderung
            output_dir: Ausgabeverzeichnis
            ca_cert_filename: Name der CA-Zertifikatsdatei
            ca_key_filename: Name der CA-Schlüsseldatei

        Returns:
            (ca_cert_path, ca_key_path)
        """
        if not self.is_openssl_available():
            raise RuntimeError(
                "OpenSSL erforderlich für CA-Generierung. "
                "Bitte OpenSSL installieren."
            )

        output_dir.mkdir(parents=True, exist_ok=True)

        ca_key_path = output_dir / ca_key_filename
        ca_cert_path = output_dir / ca_cert_filename

        # CA-Schlüssel generieren
        subprocess.run(
            [
                self._openssl_path, "genrsa",
                "-out", str(ca_key_path),
                "4096"
            ],
            check=True,
            capture_output=True
        )

        # CA-Zertifikat erstellen
        subprocess.run(
            [
                self._openssl_path, "req",
                "-new",
                "-x509",
                "-key", str(ca_key_path),
                "-out", str(ca_cert_path),
                "-days", str(request.days_valid),
                "-subj", request.get_subject_string(),
                "-extensions", "v3_ca",
                "-config", "-"
            ],
            input=b"""
[req]
distinguished_name = req_distinguished_name
x509_extensions = v3_ca
prompt = no

[req_distinguished_name]
CN = Trixy CA

[v3_ca]
basicConstraints = critical, CA:TRUE
keyUsage = critical, keyCertSign, cRLSign
""",
            check=True,
            capture_output=True
        )

        os.chmod(ca_key_path, 0o600)

        return ca_cert_path, ca_key_path

    def sign_certificate(
        self,
        csr_path: Path,
        ca_cert_path: Path,
        ca_key_path: Path,
        output_path: Path,
        days_valid: int = 365
    ) -> Path:
        """
        Signiert eine Zertifikatsanforderung (CSR) mit einer CA.

        Args:
            csr_path: Pfad zur CSR-Datei
            ca_cert_path: Pfad zum CA-Zertifikat
            ca_key_path: Pfad zum CA-Schlüssel
            output_path: Ausgabepfad für das signierte Zertifikat
            days_valid: Gültigkeitsdauer

        Returns:
            Pfad zum signierten Zertifikat
        """
        if not self.is_openssl_available():
            raise RuntimeError("OpenSSL erforderlich für Zertifikatssignierung")

        subprocess.run(
            [
                self._openssl_path, "x509",
                "-req",
                "-in", str(csr_path),
                "-CA", str(ca_cert_path),
                "-CAkey", str(ca_key_path),
                "-CAcreateserial",
                "-out", str(output_path),
                "-days", str(days_valid),
                "-sha256"
            ],
            check=True,
            capture_output=True
        )

        return output_path