"""Java reference implementation interop tests — v2 with real NTCP2 protocol. Uses the real NTCP2 handshake (AES-CBC obfuscated Noise_XK) and real transport (SipHash-obfuscated frame lengths) instead of the simplified handshake. Also tests reseed from live servers and SU3 parsing. Requirements: - podman available on the host (for container-based tests) - docker.io/geti2p/i2p:latest image pulled (for container-based tests) - Network access (for reseed test) Extracted from the monolith (i2p-python/tests/test_integration/test_java_interop_v2.py). """ import asyncio import base64 import hashlib import json import os import shutil import subprocess import tempfile import time import unittest import pytest from i2p_data.router import RouterInfo from i2p_data.su3 import SU3File from i2p_crypto.x25519 import X25519DH from i2p_transport.ntcp2_real_server import NTCP2RealConnector from i2p_transport.ntcp2_blocks import BLOCK_DATETIME, NTCP2Block, datetime_block # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _run(cmd, **kwargs): """Run a command and return stdout.""" result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=120, **kwargs ) if result.returncode != 0: raise RuntimeError( f"Command failed: {cmd}\nstdout: {result.stdout}\nstderr: {result.stderr}" ) return result.stdout.strip() def _run_quiet(cmd, **kwargs): """Run a command, ignoring errors.""" subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60, **kwargs) def _i2p_b64_decode(s: str) -> bytes: """Decode I2P modified base64 (~ instead of /, - instead of +).""" std_b64 = s.replace("~", "/").replace("-", "+") padding = 4 - len(std_b64) % 4 if padding != 4: std_b64 += "=" * padding return base64.b64decode(std_b64) def _podman_available() -> bool: """Return True if podman CLI is available.""" try: _run("podman --version") return True except (FileNotFoundError, RuntimeError): return False def _java_image_available() -> bool: """Return True if geti2p/i2p:latest image is available.""" result = subprocess.run( "podman image exists docker.io/geti2p/i2p:latest", shell=True, capture_output=True, ) return result.returncode == 0 POD_NAME = "i2p-interop-v2-test" # --------------------------------------------------------------------------- # Shared Java router fixture (class-level) # --------------------------------------------------------------------------- class _JavaRouterMixin: """Mixin that starts a Java I2P router in podman for interop tests.""" _shared_dir: str _i2p_config_dir: str @classmethod def _check_prerequisites(cls): if not _podman_available(): raise unittest.SkipTest("podman not available") if not _java_image_available(): raise unittest.SkipTest( "geti2p/i2p:latest image not available " "(run: podman pull docker.io/geti2p/i2p:latest)" ) def _setup_pod(self): self._shared_dir = tempfile.mkdtemp(prefix="i2p-interop-v2-") os.chmod(self._shared_dir, 0o777) self._i2p_config_dir = os.path.join(self._shared_dir, "i2p_config") os.makedirs(self._i2p_config_dir, exist_ok=True) os.chmod(self._i2p_config_dir, 0o777) _run_quiet(f"podman pod rm -f {POD_NAME}") _run(f"podman pod create --name {POD_NAME}") def _teardown_pod(self): _run_quiet(f"podman pod rm -f {POD_NAME}") shutil.rmtree(self._shared_dir, ignore_errors=True) def _start_java_router(self): """Start the Java I2P router container and return config dir.""" _run( f"podman run -d --pod {POD_NAME} --name java-router-v2 " f"-v {self._i2p_config_dir}:/i2p/.i2p:Z " f"-e JVM_XMX=256m " f"docker.io/geti2p/i2p:latest" ) def _wait_for_router_info(self, timeout=90) -> bytes: """Wait for router.info to appear and return its raw bytes.""" ri_path = os.path.join(self._i2p_config_dir, "router.info") deadline = time.time() + timeout while time.time() < deadline: if os.path.exists(ri_path) and os.path.getsize(ri_path) > 100: time.sleep(3) # Let it finish writing with open(ri_path, "rb") as f: return f.read() time.sleep(1) # Dump container logs for debugging logs = subprocess.run( f"podman logs java-router-v2 2>&1", shell=True, capture_output=True, text=True, ) print(f"\n--- java-router-v2 logs ---\n{logs.stdout}\n{logs.stderr}") raise TimeoutError(f"router.info not ready after {timeout}s") def _extract_ntcp2_info(self, ri_bytes: bytes): """Parse RouterInfo and return NTCP2 address details. Returns dict with keys: host, port, static_key (32 bytes), iv (16 bytes), ri_hash (32 bytes), transport. Raises ValueError if no NTCP2 address found. """ ri = RouterInfo.from_bytes(ri_bytes) ri_hash = hashlib.sha256(ri_bytes).digest() # Find NTCP2 address ntcp2_addr = None for addr in ri.addresses: if "NTCP" in addr.transport.upper(): ntcp2_addr = addr break if ntcp2_addr is None: transports = [a.transport for a in ri.addresses] raise ValueError( f"No NTCP2 address in RouterInfo (transports: {transports})" ) opts = ntcp2_addr.options s_b64 = opts.get("s", "") i_b64 = opts.get("i", "") if not s_b64: raise ValueError("NTCP2 address has no 's' (static key) option") if not i_b64: raise ValueError("NTCP2 address has no 'i' (IV) option") static_key = _i2p_b64_decode(s_b64) iv = _i2p_b64_decode(i_b64) return { "host": ntcp2_addr.get_host() or "127.0.0.1", "port": ntcp2_addr.get_port() or 12345, "static_key": static_key, "iv": iv, "ri_hash": ri_hash, "ri_bytes": ri_bytes, "transport": ntcp2_addr.transport, "all_options": opts, } # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- @pytest.mark.slow @pytest.mark.podman class TestJavaInteropV2(unittest.TestCase, _JavaRouterMixin): """Interop tests using the real NTCP2 handshake against a Java I2P router.""" @classmethod def setUpClass(cls): cls._check_prerequisites() def setUp(self): self._setup_pod() def tearDown(self): self._teardown_pod() # -- test 1: real handshake -- def test_real_handshake_with_java(self): """Start Java router, extract NTCP2 info, attempt real handshake.""" self._start_java_router() # Wait for router.info try: ri_bytes = self._wait_for_router_info(timeout=90) except TimeoutError as e: self.skipTest(f"Java router did not start in time: {e}") return # Extract NTCP2 info try: info = self._extract_ntcp2_info(ri_bytes) except ValueError as e: self.skipTest(f"Could not extract NTCP2 info: {e}") return print( f"Java NTCP2: {info['transport']} at {info['host']}:{info['port']}", flush=True, ) print(f" static_key: {info['static_key'].hex()}", flush=True) print(f" iv: {info['iv'].hex()}", flush=True) # Give Java router time to start NTCP2 listener time.sleep(5) # Generate our own keypair and minimal RouterInfo our_static = X25519DH.generate_keypair() our_ri_bytes = ri_bytes # re-use peer's RI as a stand-in async def _do_handshake(): connector = NTCP2RealConnector() conn = await asyncio.wait_for( connector.connect( host=info["host"], port=info["port"], our_static_key=our_static, our_ri_bytes=our_ri_bytes, peer_static_pub=info["static_key"], peer_ri_hash=info["ri_hash"], peer_iv=info["iv"], ), timeout=30.0, ) # Try to read some frames — Java router usually sends DateTime frames = [] try: for _ in range(3): frame_blocks = await asyncio.wait_for( conn.recv_frame(), timeout=5.0, ) for blk in frame_blocks: frames.append({ "block_type": blk.block_type, "data_len": len(blk.data), }) print( f"Received block: type={blk.block_type}, " f"data_len={len(blk.data)}", flush=True, ) except asyncio.TimeoutError: print("No more frames (timeout)", flush=True) except Exception as e: print(f"Frame read error: {e}", flush=True) await conn.close() return frames try: frames = asyncio.run(_do_handshake()) print( f"SUCCESS: Real NTCP2 handshake completed, " f"got {len(frames)} blocks", flush=True, ) dt_blocks = [f for f in frames if f["block_type"] == BLOCK_DATETIME] if dt_blocks: print("Got DateTime block(s) from Java router!", flush=True) except asyncio.TimeoutError: self.skipTest("Handshake timed out (expected — Java may reject us)") except ConnectionRefusedError: self.skipTest("Connection refused (Java NTCP2 listener not ready)") except OSError as e: self.skipTest(f"Connection error: {e}") except Exception as e: print(f"INTEROP NOTE: Handshake incomplete: {e}", flush=True) self.skipTest(f"Handshake incomplete (expected): {e}") # -- test 2: parse RouterInfo with NTCP2 options -- def test_parse_java_router_info_v2(self): """Parse Java RouterInfo and verify NTCP2 's' and 'i' options.""" self._start_java_router() try: ri_bytes = self._wait_for_router_info(timeout=90) except TimeoutError as e: self.skipTest(f"Java router did not start in time: {e}") return try: info = self._extract_ntcp2_info(ri_bytes) except ValueError as e: self.skipTest(f"Could not extract NTCP2 info: {e}") return # Verify "s" is 32 bytes (X25519 public key) self.assertEqual( len(info["static_key"]), 32, f"Expected 32-byte X25519 static key, got {len(info['static_key'])} bytes", ) # Verify "i" is 16 bytes (AES-CBC IV) self.assertEqual( len(info["iv"]), 16, f"Expected 16-byte IV, got {len(info['iv'])} bytes", ) # Verify host and port are present self.assertIsNotNone(info["host"]) self.assertIsNotNone(info["port"]) self.assertGreater(info["port"], 0) # Verify ri_hash is 32 bytes (SHA-256) self.assertEqual(len(info["ri_hash"]), 32) # Verify the raw options dict contains both keys self.assertIn("s", info["all_options"]) self.assertIn("i", info["all_options"]) print( f"Java RouterInfo parsed successfully:", flush=True, ) print(f" Transport: {info['transport']}", flush=True) print(f" Host: {info['host']}", flush=True) print(f" Port: {info['port']}", flush=True) print(f" Key (hex): {info['static_key'].hex()}", flush=True) print(f" IV (hex): {info['iv'].hex()}", flush=True) print(f" RI hash: {info['ri_hash'].hex()}", flush=True) @pytest.mark.slow @pytest.mark.network class TestReseedFromLiveServer(unittest.TestCase): """Test fetching and parsing a real SU3 reseed file from the network.""" # Reseed servers to try (in order) RESEED_URLS = [ "https://reseed.stormycloud.org", "https://reseed.diva.exchange", "https://reseed.memcpy.io", "https://reseed.i2pgit.org", "https://i2p.novg.net", ] def _fetch_su3(self, url: str) -> bytes: """Fetch SU3 from a single reseed server.""" import urllib.request import urllib.error full_url = url.rstrip("/") + "/i2pseeds.su3?netid=2" req = urllib.request.Request( full_url, headers={"User-Agent": "Wget/1.11.4"}, ) with urllib.request.urlopen(req, timeout=30) as resp: return resp.read(1_048_576) def test_reseed_from_live_server(self): """Fetch a real SU3 file, parse it, and verify NTCP2 RouterInfos.""" import urllib.error su3_data = None last_error = None for url in self.RESEED_URLS: try: print(f"Trying reseed from {url}...", flush=True) su3_data = self._fetch_su3(url) print(f"Got {len(su3_data)} bytes from {url}", flush=True) break except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e: last_error = e print(f" Failed: {e}", flush=True) continue except Exception as e: last_error = e print(f" Unexpected error: {e}", flush=True) continue if su3_data is None: self.skipTest( f"Could not fetch SU3 from any reseed server " f"(last error: {last_error})" ) return # Parse SU3 header su3 = SU3File.from_bytes(su3_data) # Verify it is a reseed file self.assertTrue( su3.is_reseed(), f"Expected reseed content type (3), got {su3.content_type}", ) self.assertEqual(su3.file_type, SU3File.TYPE_ZIP) self.assertEqual(su3.magic, SU3File.MAGIC) print(f"SU3 parsed:", flush=True) print(f" Version: {su3.version}", flush=True) print(f" Signer: {su3.signer_id}", flush=True) print(f" Content type: {su3.content_type} (reseed={su3.is_reseed()})", flush=True) print(f" File type: {su3.file_type}", flush=True) print(f" Sig type: {su3.sig_type_code}", flush=True) # Extract RouterInfos from the ZIP ri_bytes_list = su3.extract_routerinfos() self.assertGreater( len(ri_bytes_list), 0, "Expected at least one RouterInfo in reseed bundle", ) print(f" RouterInfos: {len(ri_bytes_list)}", flush=True) # Parse each RI and check for NTCP2 addresses ntcp2_count = 0 parse_errors = 0 for i, ri_bytes in enumerate(ri_bytes_list): try: ri = RouterInfo.from_bytes(ri_bytes) for addr in ri.addresses: if "NTCP" in addr.transport.upper(): opts = addr.options if "s" in opts and "i" in opts: s_key = _i2p_b64_decode(opts["s"]) iv = _i2p_b64_decode(opts["i"]) if len(s_key) == 32 and len(iv) == 16: ntcp2_count += 1 break except Exception as e: parse_errors += 1 if parse_errors <= 3: print(f" Parse error on RI #{i}: {e}", flush=True) print( f" NTCP2 addresses: {ntcp2_count}/{len(ri_bytes_list)} " f"(parse errors: {parse_errors})", flush=True, ) self.assertGreater( ntcp2_count, 0, f"Expected at least some RouterInfos with NTCP2 addresses, " f"got 0 out of {len(ri_bytes_list)} " f"({parse_errors} parse errors)", ) if __name__ == "__main__": unittest.main()