diff options
Diffstat (limited to 'poc/spike4_nat.py')
| -rw-r--r-- | poc/spike4_nat.py | 26 |
1 files changed, 16 insertions, 10 deletions
diff --git a/poc/spike4_nat.py b/poc/spike4_nat.py index 44bcdca..e34a4c4 100644 --- a/poc/spike4_nat.py +++ b/poc/spike4_nat.py @@ -12,9 +12,15 @@ Tests: E. Update hub endpoint_hint """ -import asyncio, socket, struct, os, json, httpx, time, subprocess +import asyncio +import json +import os +import socket +import struct from pathlib import Path +import httpx + PASS = "✓"; FAIL = "✗"; SKIP = "–" LOCAL_PORT = 19000 @@ -148,7 +154,7 @@ async def udp_hole_punch_test(local_port: int) -> tuple[bool, str|None, str|None try: await asyncio.wait_for(recv_event.wait(), timeout=8) - except asyncio.TimeoutError: + except TimeoutError: pass finally: transport.close() @@ -192,7 +198,7 @@ async def main(): print(f" {FAIL} {nat_type}") print(f" Cloudflare STUN : {stun_ip1}:{stun_port1}") print(f" Google STUN : {stun_ip2}:{stun_port2}") - print(f" → Hole punching unreliable; TURN relay required") + print(" → Hole punching unreliable; TURN relay required") elif stun_ip1: nat_type = "Unknown (only one STUN server responded)" print(f" {SKIP} {nat_type}") @@ -202,7 +208,7 @@ async def main(): print(f" {FAIL} {nat_type}") # ── Phase C: Bidirectional UDP hole punching ─────────────────────────────── - print(f"\n[ C ] Bidirectional UDP hole punch (node → meshbay.org → echo back)") + print("\n[ C ] Bidirectional UDP hole punch (node → meshbay.org → echo back)") print(f" Starting UDP echo server on meshbay.org:{MESHBAY_ECHO_PORT} ...") udp_ok = False seen_ext_addr = None @@ -234,7 +240,7 @@ async def main(): print(f" {FAIL} Error: {e}") # ── Phase D: UPnP ───────────────────────────────────────────────────────── - print(f"\n[ D ] UPnP port mapping") + print("\n[ D ] UPnP port mapping") upnp_ip, upnp_port = try_upnp(LOCAL_PORT) if upnp_ip: print(f" {PASS} Mapped {upnp_ip}:{upnp_port}") @@ -243,7 +249,7 @@ async def main(): print(f" {FAIL} UPnP not available on this router") # ── Phase E: Update hub ──────────────────────────────────────────────────── - print(f"\n[ E ] Update endpoint_hint on hub") + print("\n[ E ] Update endpoint_hint on hub") if endpoint_hint and state.get("username"): async with httpx.AsyncClient(timeout=10) as client: r = await client.post(f"{HUB_URL}/v1/users/login", json={ @@ -270,10 +276,10 @@ async def main(): print(f" UPnP : {PASS + ' works' if upnp_ip else FAIL + ' disabled'}") print(f" endpoint_hint : {endpoint_hint or 'none'}") if udp_ok: - print(f"\n P2P UDP is functional. QUIC transport will work.") - print(f" Spike 4 COMPLETE.") + print("\n P2P UDP is functional. QUIC transport will work.") + print(" Spike 4 COMPLETE.") else: - print(f"\n P2P blocked — TURN relay needed for this configuration.") - print(f" Spike 4 COMPLETE (with finding: relay required).") + print("\n P2P blocked — TURN relay needed for this configuration.") + print(" Spike 4 COMPLETE (with finding: relay required).") asyncio.run(main()) |