Something went wrong. Try again.
Integration tests
Something went wrong. Try again.
3.5 kB · 104 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105"""Extract NTCP2 connection info from a Java I2P router's router.info file.
Reads the binary RouterInfo, parses it, and writes the NTCP2 host, port,and static public key to a JSON file.
Usage: python extract_java_ri.py --ri-file /shared/router.info \ --output /shared/java_ntcp2_info.json"""
import argparseimport base64import jsonimport traceback
from i2p_data.router import RouterInfo
def extract_ntcp2_info(ri_path, output_path): """Parse router.info and extract NTCP2 address details.""" try: with open(ri_path, "rb") as f: data = f.read()
print(f"RouterInfo file: {len(data)} bytes", flush=True)
ri = RouterInfo.from_bytes(data) print(f"Parsed RouterInfo: {ri}", flush=True) print(f" Published: {ri.published}", flush=True) print(f" Addresses: {len(ri.addresses)}", flush=True) print(f" Options: {ri.options}", flush=True)
# Find NTCP2 address ntcp2_addr = None for addr in ri.addresses: print(f" Address: {addr} options={addr.options}", flush=True) if addr.transport == "NTCP2": ntcp2_addr = addr break
if ntcp2_addr is None: # Some newer Java routers use "NTCP" for NTCP2 for addr in ri.addresses: if "NTCP" in addr.transport.upper(): ntcp2_addr = addr break
if ntcp2_addr is None: result = { "status": "error", "error": "No NTCP2 address found in RouterInfo", "transports": [a.transport for a in ri.addresses], } else: # The "s" option contains the X25519 static key (base64) static_key_b64 = ntcp2_addr.options.get("s", "") host = ntcp2_addr.get_host() or "127.0.0.1" port = ntcp2_addr.get_port() or 12345
if static_key_b64: # I2P uses modified base64: ~ instead of /, - instead of + std_b64 = static_key_b64.replace("~", "/").replace("-", "+") # Ensure proper padding padding = 4 - len(std_b64) % 4 if padding != 4: std_b64 += "=" * padding static_key = base64.b64decode(std_b64) else: static_key = b""
result = { "status": "ok", "host": host, "port": port, "static_key_b64": static_key_b64, "static_key_hex": static_key.hex() if static_key else "", "static_key_len": len(static_key), "transport": ntcp2_addr.transport, "all_options": ntcp2_addr.options, "signature_valid": ri.verify(), }
with open(output_path, "w") as f: json.dump(result, f, indent=2) print(f"Result: {json.dumps(result, indent=2)}", flush=True)
except Exception as e: result = {"status": "error", "error": str(e)} traceback.print_exc() with open(output_path, "w") as f: json.dump(result, f) print(f"Error: {e}", flush=True)
def main(): parser = argparse.ArgumentParser() parser.add_argument("--ri-file", required=True) parser.add_argument("--output", required=True) args = parser.parse_args() extract_ntcp2_info(args.ri_file, args.output)
if __name__ == "__main__": main()