"""Naming service — resolve .i2p hostnames to Destinations.""" from abc import ABC, abstractmethod from i2p_data.data_helper import from_base64 class LookupResult: """Result of a hostname lookup.""" def __init__(self, success: bool, destination_data: bytes | None = None, error_code: int | None = None): self.success = success self.destination_data = destination_data self.error_code = error_code class NamingService(ABC): """Abstract base for hostname resolution.""" @abstractmethod def lookup(self, hostname: str) -> LookupResult: """Resolve a hostname to a Destination.""" @abstractmethod def list_all(self) -> dict[str, bytes]: """List all known hostname→destination mappings.""" class HostsTxtService(NamingService): """File-based hosts.txt resolver. Format: hostname=base64destination (one per line). Comments (#) and blank lines are ignored. Lookups are case-insensitive. """ def __init__(self, entries: dict[str, bytes]): self._entries = entries @classmethod def from_string(cls, text: str) -> "HostsTxtService": entries: dict[str, bytes] = {} for line in text.splitlines(): line = line.strip() if not line or line.startswith("#"): continue if "=" not in line: continue hostname, b64 = line.split("=", 1) hostname = hostname.strip().lower() try: dest_data = from_base64(b64.strip()) except Exception: continue entries[hostname] = dest_data return cls(entries) @classmethod def from_file(cls, path: str) -> "HostsTxtService": with open(path, "r") as f: return cls.from_string(f.read()) def lookup(self, hostname: str) -> LookupResult: key = hostname.lower() dest = self._entries.get(key) if dest is not None: return LookupResult(success=True, destination_data=dest) return LookupResult(success=False, error_code=1) def list_all(self) -> dict[str, bytes]: return dict(self._entries)