Trixy7/trixy_core/plugins/extensions/points/network.py

# -*- coding: utf-8 -*-
"""
Network Extension Points.

Extension Points für Verschlüsselung, Protokolle und Kompression.
"""

from abc import abstractmethod
from typing import Any, TYPE_CHECKING

from trixy_core.plugins.extensions.extension import Extension
from trixy_core.plugins.extensions.extension_point import ExtensionPoint, get_global_registry


# Extension Point IDs
NETWORK_ENCRYPTION_POINT = "network.encryption"
NETWORK_PROTOCOL_POINT = "network.protocol"
NETWORK_COMPRESSION_POINT = "network.compression"


class EncryptionExtension(Extension):
    """
    Extension für Verschlüsselungsverfahren.

    Ermöglicht Plugins, neue Verschlüsselungen hinzuzufügen:
    - Post-Quantum Kryptographie (Kyber, Dilithium)
    - Alternative Cipher (ChaCha20-Poly1305)
    - Hardware-beschleunigte Verschlüsselung
    - Spezielle Anforderungen (FIPS-konform)

    Beispiel:
        class ChaCha20Extension(EncryptionExtension):
            def __init__(self, plugin_name: str):
                super().__init__(
                    "chacha20_poly1305",
                    "ChaCha20-Poly1305",
                    plugin_name,
                    cipher="chacha20-poly1305",
                    key_size=256,
                )

            async def create_instance(self, config):
                return ChaCha20Poly1305Cipher()
    """

    def __init__(
        self,
        extension_id: str,
        name: str,
        plugin_name: str = "",
        cipher: str = "",
        key_size: int = 256,
        description: str = "",
    ) -> None:
        """
        Initialisiert die Encryption Extension.

        Args:
            extension_id: Eindeutige ID
            name: Anzeigename
            plugin_name: Plugin-Name
            cipher: Cipher-Name
            key_size: Schlüsselgröße in Bits
            description: Beschreibung
        """
        super().__init__(extension_id, name, plugin_name)
        self._info.description = description
        self._info.extension_point = NETWORK_ENCRYPTION_POINT
        self._info.metadata["cipher"] = cipher
        self._info.metadata["key_size"] = key_size

    @property
    def cipher(self) -> str:
        """Cipher-Name."""
        return self._info.metadata.get("cipher", "")

    @property
    def key_size(self) -> int:
        """Schlüsselgröße in Bits."""
        return self._info.metadata.get("key_size", 256)

    @abstractmethod
    async def create_instance(self, config: dict[str, Any]):
        """Erstellt die Cipher-Instanz."""
        pass


class ProtocolExtension(Extension):
    """
    Extension für Netzwerk-Protokolle.

    Ermöglicht Plugins, neue Protokolle hinzuzufügen:
    - HTTP/2, HTTP/3 (QUIC)
    - WebSocket-Varianten
    - gRPC
    - Custom Binary Protocols
    - IoT-Protokolle (MQTT, CoAP)

    Beispiel:
        class HTTP2Extension(ProtocolExtension):
            def __init__(self, plugin_name: str):
                super().__init__(
                    "http2",
                    "HTTP/2",
                    plugin_name,
                    protocol="http2",
                    port=443,
                )

            async def create_instance(self, config):
                return HTTP2Handler(config)
    """

    def __init__(
        self,
        extension_id: str,
        name: str,
        plugin_name: str = "",
        protocol: str = "",
        port: int = 0,
        secure: bool = True,
        description: str = "",
    ) -> None:
        """
        Initialisiert die Protocol Extension.

        Args:
            extension_id: Eindeutige ID
            name: Anzeigename
            plugin_name: Plugin-Name
            protocol: Protokoll-Name
            port: Standard-Port
            secure: Ist verschlüsselt
            description: Beschreibung
        """
        super().__init__(extension_id, name, plugin_name)
        self._info.description = description
        self._info.extension_point = NETWORK_PROTOCOL_POINT
        self._info.metadata["protocol"] = protocol
        self._info.metadata["port"] = port
        self._info.metadata["secure"] = secure

    @property
    def protocol(self) -> str:
        """Protokoll-Name."""
        return self._info.metadata.get("protocol", "")

    @property
    def port(self) -> int:
        """Standard-Port."""
        return self._info.metadata.get("port", 0)

    @property
    def is_secure(self) -> bool:
        """Ist verschlüsselt."""
        return self._info.metadata.get("secure", True)

    @abstractmethod
    async def create_instance(self, config: dict[str, Any]):
        """Erstellt die Protokoll-Handler-Instanz."""
        pass


class CompressionExtension(Extension):
    """
    Extension für Kompressionsverfahren.

    Ermöglicht Plugins, neue Kompressionen hinzuzufügen:
    - LZ4 (schnell)
    - Zstandard (gut)
    - Brotli (web-optimiert)
    - Spezielle Audio-Kompression

    Beispiel:
        class ZstdExtension(CompressionExtension):
            def __init__(self, plugin_name: str):
                super().__init__(
                    "zstd",
                    "Zstandard",
                    plugin_name,
                    algorithm="zstd",
                    level=3,
                )

            async def create_instance(self, config):
                return ZstdCompressor(level=config.get("level", 3))
    """

    def __init__(
        self,
        extension_id: str,
        name: str,
        plugin_name: str = "",
        algorithm: str = "",
        level: int = 6,
        description: str = "",
    ) -> None:
        """
        Initialisiert die Compression Extension.

        Args:
            extension_id: Eindeutige ID
            name: Anzeigename
            plugin_name: Plugin-Name
            algorithm: Algorithmus-Name
            level: Kompressionsgrad (0-9)
            description: Beschreibung
        """
        super().__init__(extension_id, name, plugin_name)
        self._info.description = description
        self._info.extension_point = NETWORK_COMPRESSION_POINT
        self._info.metadata["algorithm"] = algorithm
        self._info.metadata["level"] = level

    @property
    def algorithm(self) -> str:
        """Algorithmus-Name."""
        return self._info.metadata.get("algorithm", "")

    @property
    def level(self) -> int:
        """Kompressionsgrad."""
        return self._info.metadata.get("level", 6)

    @abstractmethod
    async def create_instance(self, config: dict[str, Any]):
        """Erstellt die Compressor-Instanz."""
        pass


def setup_network_extension_points() -> dict[str, ExtensionPoint]:
    """
    Richtet alle Network Extension Points ein.

    Returns:
        Dict mit allen erstellten Extension Points
    """
    registry = get_global_registry()

    points = {}

    # Verschlüsselung
    points[NETWORK_ENCRYPTION_POINT] = registry.create_point(
        id=NETWORK_ENCRYPTION_POINT,
        name="Verschlüsselungsverfahren",
        description="Erweiterung um neue Verschlüsselungen (ChaCha20, Post-Quantum, etc.)",
        category="network",
        allows_multiple=True,
    )

    # Protokolle
    points[NETWORK_PROTOCOL_POINT] = registry.create_point(
        id=NETWORK_PROTOCOL_POINT,
        name="Netzwerk-Protokolle",
        description="Erweiterung um neue Protokolle (HTTP/2, WebSocket, gRPC, etc.)",
        category="network",
        allows_multiple=True,
    )

    # Kompression
    points[NETWORK_COMPRESSION_POINT] = registry.create_point(
        id=NETWORK_COMPRESSION_POINT,
        name="Kompressionsverfahren",
        description="Erweiterung um neue Kompressionen (LZ4, Zstd, Brotli, etc.)",
        category="network",
        allows_multiple=True,
    )

    return points