Something went wrong. Try again.
Integration tests
Something went wrong. Try again.
17 kB · 481 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482"""Java reference implementation interop tests — v2 with real NTCP2 protocol.
Uses the real NTCP2 handshake (AES-CBC obfuscated Noise_XK) and realtransport (SipHash-obfuscated frame lengths) instead of the simplifiedhandshake.
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 asyncioimport base64import hashlibimport jsonimport osimport shutilimport subprocessimport tempfileimport timeimport unittest
import pytest
from i2p_data.router import RouterInfofrom i2p_data.su3 import SU3Filefrom i2p_crypto.x25519 import X25519DHfrom i2p_transport.ntcp2_real_server import NTCP2RealConnectorfrom 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.podmanclass 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.networkclass 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()