"""LocalHTTPServer — internal HTTP handler for proxy.i2p. Serves health pages, addressbook add forms, and error page CSS when the HTTP proxy receives a request for the reserved proxy.i2p hostname. Ported from net.i2p.i2ptunnel.localServer.LocalHTTPServer. """ from __future__ import annotations import secrets from html import escape class LocalHTTPHandler: """Handles HTTP requests to proxy.i2p.""" def __init__(self) -> None: self._nonces: set[str] = set() def _generate_nonce(self) -> str: """Generate a CSRF nonce for forms.""" nonce = secrets.token_hex(16) self._nonces.add(nonce) # Keep only last 100 nonces if len(self._nonces) > 100: self._nonces = set(list(self._nonces)[-100:]) return nonce def _validate_nonce(self, nonce: str) -> bool: """Validate and consume a CSRF nonce.""" if nonce in self._nonces: self._nonces.discard(nonce) return True return False def handle_get(self, path: str) -> tuple[int, str]: """Handle a GET request. Returns (status_code, html_body).""" if path == "/": return self._health_page() if path.startswith("/add"): return self._add_form(path) return 404, "
The proxy is running.
" "" ) def _add_form(self, path: str) -> tuple[int, str]: """Serve the addressbook add form.""" # Parse query string params: dict[str, str] = {} if "?" in path: qs = path.split("?", 1)[1] for pair in qs.split("&"): if "=" in pair: k, v = pair.split("=", 1) params[k] = v host = escape(params.get("host", "")) dest = escape(params.get("dest", "")) nonce = self._generate_nonce() return 200, ( "Invalid nonce.
" host = form_data.get("host", "") dest = form_data.get("dest", "") if not host or not dest: return 400, "Missing host or dest.
" return 200, ( "{escape(host)} has been added to your addressbook.
" "" ) return 404, "