distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""I2P tunnel manager — manages named tunnels.
2
3Ported from net.i2p.i2ptunnel.TunnelControllerGroup.
4"""
5
6from __future__ import annotations
7
8from typing import Any
9
10
11class I2PTunnelManager:
12 """Manages named I2P tunnels (HTTP proxies, SOCKS proxies, server tunnels)."""
13
14 def __init__(self) -> None:
15 self._tunnels: dict[str, Any] = {}
16
17 def add_tunnel(self, name: str, tunnel: Any) -> None:
18 """Add a named tunnel. Raises ValueError if name already exists."""
19 if name in self._tunnels:
20 raise ValueError(f"Tunnel '{name}' already exists")
21 self._tunnels[name] = tunnel
22
23 def remove_tunnel(self, name: str) -> bool:
24 """Remove a tunnel by name. Returns True if removed, False if not found."""
25 if name in self._tunnels:
26 del self._tunnels[name]
27 return True
28 return False
29
30 def get_tunnel(self, name: str) -> Any | None:
31 """Get a tunnel by name, or None if not found."""
32 return self._tunnels.get(name)
33
34 def list_tunnels(self) -> list[str]:
35 """List all tunnel names."""
36 return list(self._tunnels.keys())
37
38 @property
39 def tunnel_count(self) -> int:
40 """Number of managed tunnels."""
41 return len(self._tunnels)