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