distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Naming service — resolve .i2p hostnames to Destinations."""
2
3from abc import ABC, abstractmethod
4
5from i2p_data.data_helper import from_base64
6
7
8class LookupResult:
9 """Result of a hostname lookup."""
10
11 def __init__(self, success: bool, destination_data: bytes | None = None,
12 error_code: int | None = None):
13 self.success = success
14 self.destination_data = destination_data
15 self.error_code = error_code
16
17
18class NamingService(ABC):
19 """Abstract base for hostname resolution."""
20
21 @abstractmethod
22 def lookup(self, hostname: str) -> LookupResult:
23 """Resolve a hostname to a Destination."""
24
25 @abstractmethod
26 def list_all(self) -> dict[str, bytes]:
27 """List all known hostname→destination mappings."""
28
29
30class HostsTxtService(NamingService):
31 """File-based hosts.txt resolver.
32
33 Format: hostname=base64destination (one per line).
34 Comments (#) and blank lines are ignored.
35 Lookups are case-insensitive.
36 """
37
38 def __init__(self, entries: dict[str, bytes]):
39 self._entries = entries
40
41 @classmethod
42 def from_string(cls, text: str) -> "HostsTxtService":
43 entries: dict[str, bytes] = {}
44 for line in text.splitlines():
45 line = line.strip()
46 if not line or line.startswith("#"):
47 continue
48 if "=" not in line:
49 continue
50 hostname, b64 = line.split("=", 1)
51 hostname = hostname.strip().lower()
52 try:
53 dest_data = from_base64(b64.strip())
54 except Exception:
55 continue
56 entries[hostname] = dest_data
57 return cls(entries)
58
59 @classmethod
60 def from_file(cls, path: str) -> "HostsTxtService":
61 with open(path, "r") as f:
62 return cls.from_string(f.read())
63
64 def lookup(self, hostname: str) -> LookupResult:
65 key = hostname.lower()
66 dest = self._entries.get(key)
67 if dest is not None:
68 return LookupResult(success=True, destination_data=dest)
69 return LookupResult(success=False, error_code=1)
70
71 def list_all(self) -> dict[str, bytes]:
72 return dict(self._entries)