dns: add i2pdns bridge for .i2p resolution

3proxy/socks_router.py

@@ -1,10 +1,11 @@
 #!/usr/bin/env python3
 import asyncio
 import ipaddress
+import json
 import os
 import signal
 import struct
-import sys
+import time
 from datetime import datetime
 from typing import Optional, Tuple
 
@@ -38,6 +39,13 @@ LISTEN_PORT = int(os.getenv("ROUTER_LISTEN_PORT", "1080"))
 UPSTREAM_TOR = env_endpoint("ROUTER_UPSTREAM_TOR", "10.5.0.7:9050")
 UPSTREAM_I2P = env_endpoint("ROUTER_UPSTREAM_I2P", "10.5.0.2:4447")
 UPSTREAM_YGG = env_endpoint("ROUTER_UPSTREAM_YGG", "127.0.0.1:1085")
+I2P_MAP_FILE = os.getenv("ROUTER_I2P_MAP_FILE", "").strip()
+I2P_POOL_CIDR = os.getenv("ROUTER_I2P_POOL_CIDR", "172.31.0.0/16")
+I2P_POOL = ipaddress.ip_network(I2P_POOL_CIDR)
+
+_i2p_map_mtime = -1.0
+_i2p_map_last_check = 0.0
+_i2p_ip_to_host = {}
 
 BLOCKLIST = [
     item.strip().lower()
@@ -59,6 +67,59 @@ def is_blocked_domain(hostname: str) -> bool:
     return False
 
 
+def _reload_i2p_map_if_needed() -> None:
+    global _i2p_map_mtime, _i2p_map_last_check, _i2p_ip_to_host
+
+    if not I2P_MAP_FILE:
+        return
+
+    now = time.monotonic()
+    if now - _i2p_map_last_check < 1.0:
+        return
+    _i2p_map_last_check = now
+
+    try:
+        mtime = os.path.getmtime(I2P_MAP_FILE)
+    except OSError:
+        _i2p_ip_to_host = {}
+        _i2p_map_mtime = -1.0
+        return
+
+    if mtime == _i2p_map_mtime:
+        return
+
+    try:
+        with open(I2P_MAP_FILE, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+        mapping = data.get("ip_to_host", {}) if isinstance(data, dict) else {}
+        if isinstance(mapping, dict):
+            _i2p_ip_to_host = {str(k): str(v) for k, v in mapping.items()}
+        else:
+            _i2p_ip_to_host = {}
+        _i2p_map_mtime = mtime
+    except Exception:
+        _i2p_ip_to_host = {}
+
+
+def lookup_i2p_hostname_for_virtual_ip(ip_str: str) -> Optional[str]:
+    if not I2P_MAP_FILE:
+        return None
+
+    try:
+        ip = ipaddress.ip_address(ip_str)
+    except ValueError:
+        return None
+
+    if ip.version != 4 or ip not in I2P_POOL:
+        return None
+
+    _reload_i2p_map_if_needed()
+    host = _i2p_ip_to_host.get(str(ip))
+    if host and host.endswith(".i2p"):
+        return host
+    return None
+
+
 def pick_upstream(host: str, atyp: int) -> Optional[Tuple[str, int]]:
     if atyp == ATYP_DOMAIN:
         dhost = host.rstrip(".").lower()
@@ -155,14 +216,30 @@ async def handle_client(client_reader: asyncio.StreamReader, client_writer: asyn
 
         dst_port = struct.unpack("!H", await read_exact(client_reader, 2))[0]
 
-        upstream = pick_upstream(dst_host, atyp)
+        connect_atyp = atyp
+        connect_host = dst_host
+        connect_raw_addr = raw_addr
+
+        if atyp == ATYP_IPV4:
+            mapped_i2p = lookup_i2p_hostname_for_virtual_ip(dst_host)
+            if mapped_i2p:
+                connect_atyp = ATYP_DOMAIN
+                connect_host = mapped_i2p
+                connect_raw_addr = None
+
+        upstream = pick_upstream(connect_host, connect_atyp)
         if upstream is None:
             await send_reply(client_writer, REP_CONNECTION_NOT_ALLOWED)
             log(f"deny peer={peer} dst={dst_host}:{dst_port}")
             return
 
         up_host, up_port = upstream
-        log(f"route peer={peer} dst={dst_host}:{dst_port} via={up_host}:{up_port}")
+        if connect_host != dst_host:
+            log(
+                f"route peer={peer} dst={dst_host}:{dst_port} mapped={connect_host} via={up_host}:{up_port}"
+            )
+        else:
+            log(f"route peer={peer} dst={dst_host}:{dst_port} via={up_host}:{up_port}")
 
         try:
             upstream_reader, upstream_writer = await asyncio.open_connection(up_host, up_port)
@@ -179,19 +256,19 @@ async def handle_client(client_reader: asyncio.StreamReader, client_writer: asyn
             return
 
         connect_req = bytearray(b"\x05\x01\x00")
-        connect_req.append(atyp)
-        if atyp == ATYP_DOMAIN:
+        connect_req.append(connect_atyp)
+        if connect_atyp == ATYP_DOMAIN:
             try:
-                host_bytes = dst_host.encode("idna")
+                host_bytes = connect_host.encode("idna")
             except UnicodeError:
-                host_bytes = dst_host.encode("ascii", errors="ignore")
+                host_bytes = connect_host.encode("ascii", errors="ignore")
             if len(host_bytes) > 255:
                 await send_reply(client_writer, REP_HOST_UNREACHABLE)
                 return
             connect_req.append(len(host_bytes))
             connect_req.extend(host_bytes)
         else:
-            connect_req.extend(raw_addr)
+            connect_req.extend(connect_raw_addr)
         connect_req.extend(struct.pack("!H", dst_port))
 
         upstream_writer.write(connect_req)