distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""I2PControl RPC method implementations."""
2from __future__ import annotations
3
4
5class I2PControlMethods:
6 """RPC method implementations.
7
8 Each method takes a *params* dict and returns a result dict.
9 Uses a mock/stub router context for now.
10 """
11
12 def __init__(self, router_context: dict | None = None) -> None:
13 self._context: dict = router_context or {}
14 self._config: dict[str, str] = {}
15
16 # ----- router methods ---------------------------------------------------
17
18 def router_info(self, params: dict) -> dict:
19 """Return router info: version, status, uptime, etc."""
20 return {
21 "version": "0.1.0",
22 "status": self._context.get("status", "running"),
23 "uptime": self._context.get("uptime", 0),
24 "network_status": self._context.get("network_status", "ok"),
25 }
26
27 def router_status(self, params: dict) -> dict:
28 """Return router status state."""
29 return {"status": self._context.get("status", "running")}
30
31 # ----- tunnel methods ---------------------------------------------------
32
33 def tunnel_list(self, params: dict) -> dict:
34 """Return list of active tunnels."""
35 return {"tunnels": self._context.get("tunnels", [])}
36
37 # ----- peer methods -----------------------------------------------------
38
39 def peer_list(self, params: dict) -> dict:
40 """Return list of known peers, with optional limit."""
41 limit = params.get("limit", 50)
42 peers = self._context.get("peers", [])
43 return {"peers": peers[:limit]}
44
45 # ----- netdb methods ----------------------------------------------------
46
47 def netdb_stats(self, params: dict) -> dict:
48 """Return network database statistics."""
49 return {
50 "total_routers": self._context.get("total_routers", 0),
51 "total_leasesets": self._context.get("total_leasesets", 0),
52 }
53
54 # ----- bandwidth --------------------------------------------------------
55
56 def bw_status(self, params: dict) -> dict:
57 """Return bandwidth status."""
58 return {
59 "inbound_bps": self._context.get("inbound_bps", 0),
60 "outbound_bps": self._context.get("outbound_bps", 0),
61 }
62
63 # ----- config -----------------------------------------------------------
64
65 def config_get(self, params: dict) -> dict:
66 """Get a configuration value by key."""
67 key = params.get("key", "")
68 return {"key": key, "value": self._config.get(key)}
69
70 def config_set(self, params: dict) -> dict:
71 """Set a configuration value."""
72 key = params.get("key", "")
73 value = params.get("value", "")
74 self._config[key] = value
75 return {"key": key, "value": value, "success": True}
76
77 # ----- dispatch ---------------------------------------------------------
78
79 def dispatch(self, method: str, params: dict) -> dict:
80 """Dispatch a method call by name.
81
82 Dot-separated method names (e.g. ``router.info``) are converted to
83 underscores (``router_info``) for lookup.
84
85 Raises ``ValueError`` for unknown methods.
86 """
87 handler_name = method.replace(".", "_")
88 handler = getattr(self, handler_name, None)
89 if handler is None or handler_name.startswith("_") or handler_name == "dispatch":
90 raise ValueError(f"Unknown method: {method}")
91 return handler(params)