Trixy7/tests/test_network_tls.py

# -*- coding: utf-8 -*-
"""
Tests für TLS/SSL Module.

Testet TLS-Konfiguration, Zertifikats-Manager und Generator.
"""

import os
import ssl
import tempfile
from pathlib import Path

import pytest

from trixy_core.network.tls.config import (
    TLSConfig,
    TLSVersion,
    CipherSuite,
)
from trixy_core.network.tls.manager import (
    TLSManager,
    CertificateInfo,
    create_development_certificates,
)
from trixy_core.network.tls.generator import (
    CertificateGenerator,
    CertificateRequest,
    KeyType,
)


# =============================================================================
# TLS Config Tests
# =============================================================================

class TestTLSVersion:
    """Tests für TLSVersion."""

    def test_tls_1_2_to_ssl(self):
        """TLS 1.2 zu ssl.TLSVersion."""
        assert TLSVersion.TLS_1_2.to_ssl_version() == ssl.TLSVersion.TLSv1_2

    def test_tls_1_3_to_ssl(self):
        """TLS 1.3 zu ssl.TLSVersion."""
        assert TLSVersion.TLS_1_3.to_ssl_version() == ssl.TLSVersion.TLSv1_3


class TestCipherSuite:
    """Tests für CipherSuite."""

    def test_modern_ciphers(self):
        """Moderne Cipher-Suites."""
        ciphers = CipherSuite.MODERN.get_ciphers()
        assert "TLS_AES_256_GCM_SHA384" in ciphers
        assert "ECDHE" in ciphers

    def test_intermediate_ciphers(self):
        """Intermediate Cipher-Suites."""
        ciphers = CipherSuite.INTERMEDIATE.get_ciphers()
        assert "ECDHE" in ciphers
        assert "!aNULL" in ciphers  # Keine anonymen Ciphers

    def test_compatible_ciphers(self):
        """Compatible Cipher-Suites."""
        ciphers = CipherSuite.COMPATIBLE.get_ciphers()
        assert "AES" in ciphers


class TestTLSConfig:
    """Tests für TLSConfig."""

    def test_default_config(self):
        """Standard-Konfiguration."""
        config = TLSConfig()
        assert config.enabled is False
        assert config.min_version == TLSVersion.TLS_1_2
        assert config.cipher_suite == CipherSuite.MODERN

    def test_config_with_paths(self):
        """Konfiguration mit Pfaden."""
        config = TLSConfig(
            enabled=True,
            cert_file="/path/to/cert.crt",
            key_file="/path/to/key.key",
        )
        assert isinstance(config.cert_file, Path)
        assert config.cert_file == Path("/path/to/cert.crt")

    def test_validation_disabled(self):
        """Validierung bei deaktiviertem TLS."""
        config = TLSConfig(enabled=False)
        valid, error = config.is_valid()
        assert valid is True
        assert error == ""

    def test_validation_missing_cert(self):
        """Validierung ohne Zertifikat."""
        config = TLSConfig(enabled=True, key_file="/path/to/key")
        valid, error = config.is_valid()
        assert valid is False
        assert "Zertifikat" in error

    def test_validation_missing_key(self):
        """Validierung ohne Schlüssel."""
        config = TLSConfig(enabled=True, cert_file="/path/to/cert")
        valid, error = config.is_valid()
        assert valid is False
        assert "Schlüssel" in error

    def test_validation_file_not_found(self):
        """Validierung mit nicht existierender Datei."""
        config = TLSConfig(
            enabled=True,
            cert_file="/nonexistent/cert.crt",
            key_file="/nonexistent/key.key",
        )
        valid, error = config.is_valid()
        assert valid is False
        assert "nicht gefunden" in error

    def test_serialization(self):
        """Serialisierung und Deserialisierung."""
        config = TLSConfig(
            enabled=True,
            cert_file="/path/to/cert.crt",
            key_file="/path/to/key.key",
            min_version=TLSVersion.TLS_1_3,
            cipher_suite=CipherSuite.INTERMEDIATE,
            verify_client=True,
        )

        data = config.to_dict()
        restored = TLSConfig.from_dict(data)

        assert restored.enabled == config.enabled
        assert restored.min_version == config.min_version
        assert restored.cipher_suite == config.cipher_suite
        assert restored.verify_client == config.verify_client


# =============================================================================
# Certificate Request Tests
# =============================================================================

class TestCertificateRequest:
    """Tests für CertificateRequest."""

    def test_default_request(self):
        """Standard-Anforderung."""
        request = CertificateRequest(common_name="localhost")
        assert request.common_name == "localhost"
        assert request.organization == "Trixy"
        assert request.key_type == KeyType.RSA_2048
        assert request.days_valid == 365

    def test_subject_string(self):
        """Subject-String generieren."""
        request = CertificateRequest(
            common_name="trixy.local",
            organization="Test Org",
            country="DE",
        )
        subject = request.get_subject_string()
        assert "/CN=trixy.local" in subject
        assert "/O=Test Org" in subject
        assert "/C=DE" in subject

    def test_san_extension(self):
        """SAN-Extension generieren."""
        request = CertificateRequest(
            common_name="trixy.local",
            san_dns=["localhost", "trixy.local"],
            san_ip=["127.0.0.1", "::1"],
        )
        san = request.get_san_extension()
        assert "DNS:localhost" in san
        assert "DNS:trixy.local" in san
        assert "IP:127.0.0.1" in san
        assert "IP:::1" in san

    def test_san_default_cn(self):
        """SAN Default = CN wenn leer."""
        request = CertificateRequest(common_name="example.com")
        san = request.get_san_extension()
        assert "DNS:example.com" in san


# =============================================================================
# Certificate Generator Tests
# =============================================================================

class TestCertificateGenerator:
    """Tests für CertificateGenerator."""

    def test_openssl_availability(self):
        """OpenSSL-Verfügbarkeit prüfen."""
        generator = CertificateGenerator()
        # Sollte auf den meisten Systemen True sein
        result = generator.is_openssl_available()
        assert isinstance(result, bool)

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_generate_self_signed(self):
        """Selbstsigniertes Zertifikat generieren."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(
                common_name="test.local",
                organization="Test",
                days_valid=30,
                san_dns=["test.local", "localhost"],
                san_ip=["127.0.0.1"],
            )

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

            assert cert_path.exists()
            assert key_path.exists()

            # Key-Berechtigungen prüfen
            key_mode = os.stat(key_path).st_mode & 0o777
            assert key_mode == 0o600

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_generate_different_key_types(self):
        """Verschiedene Schlüsseltypen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()

            for key_type in [KeyType.RSA_2048, KeyType.RSA_4096]:
                request = CertificateRequest(
                    common_name=f"test-{key_type.name}.local",
                    key_type=key_type,
                )

                cert_path, key_path = generator.generate_self_signed(
                    request=request,
                    output_dir=Path(tmpdir),
                    cert_filename=f"cert_{key_type.name}.crt",
                    key_filename=f"key_{key_type.name}.key",
                )

                assert cert_path.exists()
                assert key_path.exists()

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_generate_ecdsa(self):
        """ECDSA-Schlüssel generieren."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(
                common_name="ecdsa.local",
                key_type=KeyType.ECDSA_P256,
            )

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

            assert cert_path.exists()
            assert key_path.exists()


# =============================================================================
# TLS Manager Tests
# =============================================================================

class TestTLSManager:
    """Tests für TLSManager."""

    def test_disabled_tls(self):
        """TLS deaktiviert."""
        config = TLSConfig(enabled=False)
        manager = TLSManager(config)

        assert manager.is_enabled is False
        assert manager.get_server_context() is None
        assert manager.get_client_context() is None

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_create_server_context(self):
        """Server-Kontext erstellen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            # Zertifikat generieren
            generator = CertificateGenerator()
            request = CertificateRequest(common_name="localhost")
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            # Manager erstellen
            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
            )
            manager = TLSManager(config)

            context = manager.get_server_context()
            assert context is not None
            assert isinstance(context, ssl.SSLContext)

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_certificate_info(self):
        """Zertifikats-Info abrufen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(
                common_name="test.local",
                organization="Test Org",
            )
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
            )
            manager = TLSManager(config)

            info = manager.get_certificate_info()
            assert info is not None
            assert info.fingerprint_sha256 != ""

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_reload(self):
        """Kontexte neu laden."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(common_name="localhost")
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
            )
            manager = TLSManager(config)

            # Ersten Kontext erstellen
            ctx1 = manager.get_server_context()

            # Reload
            manager.reload()

            # Neuen Kontext erstellen
            ctx2 = manager.get_server_context()

            # Sollten verschiedene Objekte sein
            assert ctx1 is not ctx2

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_get_status(self):
        """Status abrufen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(common_name="localhost")
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
            )
            manager = TLSManager(config)

            status = manager.get_status()
            assert status["enabled"] is True
            assert status["min_version"] == "TLS_1_2"


# =============================================================================
# Development Certificates Helper Tests
# =============================================================================

class TestDevelopmentCertificates:
    """Tests für Entwicklungs-Zertifikate."""

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_create_development_certificates(self):
        """Entwicklungszertifikate erstellen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            cert_path, key_path = create_development_certificates(
                output_dir=Path(tmpdir),
                common_name="dev.local",
                days_valid=7,
            )

            assert cert_path.exists()
            assert key_path.exists()

            # Zertifikat-Inhalt prüfen
            cert_content = cert_path.read_text()
            assert "BEGIN CERTIFICATE" in cert_content

            # Schlüssel-Inhalt prüfen
            key_content = key_path.read_text()
            assert "PRIVATE KEY" in key_content


# =============================================================================
# SSL Context Creation Tests
# =============================================================================

class TestSSLContextCreation:
    """Tests für SSL-Context-Erstellung."""

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_server_context_options(self):
        """Server-Kontext-Optionen prüfen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(common_name="localhost")
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
                min_version=TLSVersion.TLS_1_2,
                cipher_suite=CipherSuite.MODERN,
            )

            context = config.create_server_context()

            # Minimale Version prüfen
            assert context.minimum_version == ssl.TLSVersion.TLSv1_2

            # Sicherheitsoptionen prüfen
            # Hinweis: OP_NO_SSLv2 ist 0 in modernem Python (SSLv2 komplett entfernt)
            assert context.options & ssl.OP_NO_SSLv3
            assert context.options & ssl.OP_NO_TLSv1
            assert context.options & ssl.OP_NO_TLSv1_1

    @pytest.mark.skipif(
        not CertificateGenerator().is_openssl_available(),
        reason="OpenSSL nicht verfügbar"
    )
    def test_client_context_options(self):
        """Client-Kontext-Optionen prüfen."""
        with tempfile.TemporaryDirectory() as tmpdir:
            generator = CertificateGenerator()
            request = CertificateRequest(common_name="localhost")
            cert_path, key_path = generator.generate_self_signed(
                request=request,
                output_dir=Path(tmpdir),
            )

            config = TLSConfig(
                enabled=True,
                cert_file=cert_path,
                key_file=key_path,
                ca_file=cert_path,  # Selbstsigniert = CA
                check_hostname=False,
            )

            context = config.create_client_context()

            assert context.check_hostname is False
            assert context.minimum_version >= ssl.TLSVersion.TLSv1_2


# =============================================================================
# Edge Cases
# =============================================================================

class TestEdgeCases:
    """Tests für Randfälle."""

    def test_invalid_openssl_path(self):
        """Ungültiger OpenSSL-Pfad."""
        generator = CertificateGenerator(openssl_path="/nonexistent/openssl")
        assert generator.is_openssl_available() is False

    def test_empty_config(self):
        """Leere Konfiguration."""
        config = TLSConfig()
        data = config.to_dict()
        restored = TLSConfig.from_dict(data)
        assert restored.enabled is False

    def test_path_string_conversion(self):
        """String zu Path Konvertierung."""
        config = TLSConfig(
            enabled=True,
            cert_file="relative/path/cert.crt",
            key_file="relative/path/key.key",
        )
        assert isinstance(config.cert_file, Path)
        assert isinstance(config.key_file, Path)