Something went wrong. Try again.
Tunnel construction and management
Something went wrong. Try again.
1.8 kB · 56 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657"""Tunnel data processing wiring."""
from __future__ import annotations
from i2p_tunnel.crypto import TunnelLayerDecryptor, OutboundTunnelEncryptor
class TunnelCryptoRegistry: """Registry mapping tunnel IDs to their crypto keys."""
def __init__(self) -> None: self._tunnels: dict[int, tuple[bytes, bytes, bool]] = {}
def register( self, tunnel_id: int, layer_key: bytes, iv_key: bytes, is_endpoint: bool = False, ) -> None: self._tunnels[tunnel_id] = (layer_key, iv_key, is_endpoint)
def get_keys(self, tunnel_id: int) -> tuple[bytes, bytes, bool] | None: return self._tunnels.get(tunnel_id)
def remove(self, tunnel_id: int) -> None: self._tunnels.pop(tunnel_id, None)
def registered_tunnels(self) -> list[int]: return list(self._tunnels.keys())
class TunnelDataHandler: """Process tunnel data: decrypt inbound, encrypt outbound."""
def __init__(self, crypto_registry: TunnelCryptoRegistry) -> None: self._registry = crypto_registry
def handle_inbound(self, tunnel_id: int, encrypted_data: bytes) -> dict: keys = self._registry.get_keys(tunnel_id) if keys is None: return {"action": "unknown", "tunnel_id": tunnel_id}
layer_key, iv_key, is_endpoint = keys decrypted = TunnelLayerDecryptor.decrypt_layer( encrypted_data, layer_key, iv_key )
if is_endpoint: return {"action": "deliver", "data": decrypted} return {"action": "forward", "data": decrypted}
def handle_outbound( self, data: bytes, hop_keys: list[tuple[bytes, bytes]] ) -> bytes: return OutboundTunnelEncryptor.encrypt(data, hop_keys)