distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""LocalHTTPServer — internal HTTP handler for proxy.i2p.
2
3Serves health pages, addressbook add forms, and error page CSS
4when the HTTP proxy receives a request for the reserved proxy.i2p hostname.
5
6Ported from net.i2p.i2ptunnel.localServer.LocalHTTPServer.
7"""
8
9from __future__ import annotations
10
11import secrets
12from html import escape
13
14
15class LocalHTTPHandler:
16 """Handles HTTP requests to proxy.i2p."""
17
18 def __init__(self) -> None:
19 self._nonces: set[str] = set()
20
21 def _generate_nonce(self) -> str:
22 """Generate a CSRF nonce for forms."""
23 nonce = secrets.token_hex(16)
24 self._nonces.add(nonce)
25 # Keep only last 100 nonces
26 if len(self._nonces) > 100:
27 self._nonces = set(list(self._nonces)[-100:])
28 return nonce
29
30 def _validate_nonce(self, nonce: str) -> bool:
31 """Validate and consume a CSRF nonce."""
32 if nonce in self._nonces:
33 self._nonces.discard(nonce)
34 return True
35 return False
36
37 def handle_get(self, path: str) -> tuple[int, str]:
38 """Handle a GET request. Returns (status_code, html_body)."""
39 if path == "/":
40 return self._health_page()
41
42 if path.startswith("/add"):
43 return self._add_form(path)
44
45 return 404, "<html><body><h1>404 Not Found</h1></body></html>"
46
47 def _health_page(self) -> tuple[int, str]:
48 """Serve the health/index page."""
49 return 200, (
50 "<html><head><title>I2P Proxy</title></head>"
51 "<body><h1>I2P HTTP Proxy</h1>"
52 "<p>The proxy is running.</p>"
53 "</body></html>"
54 )
55
56 def _add_form(self, path: str) -> tuple[int, str]:
57 """Serve the addressbook add form."""
58 # Parse query string
59 params: dict[str, str] = {}
60 if "?" in path:
61 qs = path.split("?", 1)[1]
62 for pair in qs.split("&"):
63 if "=" in pair:
64 k, v = pair.split("=", 1)
65 params[k] = v
66
67 host = escape(params.get("host", ""))
68 dest = escape(params.get("dest", ""))
69 nonce = self._generate_nonce()
70
71 return 200, (
72 "<html><head><title>Add to Addressbook</title></head>"
73 "<body><h1>Add to Addressbook</h1>"
74 f"<form method='POST' action='/add'>"
75 f"<input type='hidden' name='nonce' value='{nonce}'/>"
76 f"<p>Host: <input name='host' value='{host}'/></p>"
77 f"<p>Destination: <textarea name='dest'>{dest}</textarea></p>"
78 f"<p><input type='submit' value='Add'/></p>"
79 f"</form></body></html>"
80 )
81
82 def handle_post(self, path: str, form_data: dict[str, str]) -> tuple[int, str]:
83 """Handle a POST request. Returns (status_code, html_body)."""
84 if path == "/add":
85 nonce = form_data.get("nonce", "")
86 if not self._validate_nonce(nonce):
87 return 403, "<html><body><h1>403 Forbidden</h1><p>Invalid nonce.</p></body></html>"
88
89 host = form_data.get("host", "")
90 dest = form_data.get("dest", "")
91 if not host or not dest:
92 return 400, "<html><body><h1>400 Bad Request</h1><p>Missing host or dest.</p></body></html>"
93
94 return 200, (
95 "<html><body><h1>Added</h1>"
96 f"<p>{escape(host)} has been added to your addressbook.</p>"
97 "</body></html>"
98 )
99
100 return 404, "<html><body><h1>404 Not Found</h1></body></html>"