1 files changed, 106 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_local_network_refused.py b/packages/meshbay-hub/tests/test_local_network_refused.py
new file mode 100644
index 00000000..3d0f3041
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_local_network_refused.py
@@ -0,0 +1,106 @@
+"""
+A device that refuses the local network is named, not reported as "DataChannel closed".
+
+Android 17 makes reaching the local network a permission of the application
+(ACCESS_LOCAL_NETWORK), enforced for any application that targets it. Chrome
+does from 155, and holds the permission only once the person grants it. Until
+then the browser drops the node's addresses on the phone's own subnet: they are
+absent from getStats, no check goes to them, the node's own checks get no
+answer, ICE fails and the channel closes. On mobile data the same group opens,
+over the node's public address. Measured on a Pixel, Android 17, Chrome 155,
+2026-10-16; the addresses below are that session's.
+
+These run the shipped transport: harness/local_network_harness.mjs.
+"""
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from spa_source import transport_argv
+
+HARNESS = Path(__file__).parent / "harness" / "local_network_harness.mjs"
+
+pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")
+
+
+def _run(cfg: dict):
+    proc = subprocess.run(["node", str(HARNESS), transport_argv(), json.dumps(cfg)],
+                          capture_output=True, text=True, encoding="utf-8", timeout=120)
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def _sdp(*candidates: tuple[str, str]) -> str:
+    lines = ["v=0", "m=application 9 UDP/DTLS/SCTP webrtc-datachannel"]
+    for i, (address, kind) in enumerate(candidates):
+        lines.append(f"a=candidate:{i} 1 udp 2130706431 {address} {50000 + i} typ {kind}")
+    return "\r\n".join(lines) + "\r\n"
+
+
+PHONE_WIFI = _sdp(("192.168.1.13", "host"), ("2a01:cb15:80e8:de00:c601:3ad3:2b17:e6a6", "host"),
+                  ("90.112.206.172", "srflx"))
+DESKTOP = _sdp(("64f602b7-7667-411f-b8ae-910c4d1de6fb.local", "host"),
+               ("07b5e6a0-2b57-45ce-8748-ce2918cc6145.local", "host"),
+               ("90.112.206.172", "srflx"))
+PHONE_MOBILE = _sdp(("10.161.7.20", "host"), ("2a04:cec0:1000:2c41::1", "host"),
+                    ("92.184.100.12", "srflx"))
+PHONE_V6_ONLY = _sdp(("2a04:cec0:1000:2c41::1", "host"))
+NODE = _sdp(("192.168.1.22", "host"), ("2a01:cb15:80e8:de00:d455:bcab:c324:4d72", "host"),
+            ("192.168.200.254", "host"), ("192.168.100.254", "host"),
+            ("90.112.206.172", "srflx"))
+NODE_NO_STUN = _sdp(("192.168.1.22", "host"))
+EVERY_NODE_ADDRESS = ["192.168.1.22", "2a01:cb15:80e8:de00:d455:bcab:c324:4d72",
+                      "192.168.200.254", "192.168.100.254", "90.112.206.172"]
+# What Chrome 155 kept, refused: the node's virtual adapters, off the phone's
+# subnet, and its public address; not 192.168.1.22 nor the IPv6 of the same /64.
+KEPT_WHEN_REFUSED = ["192.168.200.254", "192.168.100.254", "90.112.206.172"]
+
+
+def test_the_measured_refusal_is_recognised():
+    assert _run({"classify": [[PHONE_WIFI, NODE, KEPT_WHEN_REFUSED]]}) == [True]
+
+
+def test_a_browser_that_pairs_every_address_is_not_refusing():
+    """Desktop Chrome 152, 155 and 157 listed all six of the node's addresses."""
+    assert _run({"classify": [[DESKTOP, NODE, EVERY_NODE_ADDRESS],
+                              [PHONE_WIFI, NODE, EVERY_NODE_ADDRESS]]}) == [False, False]
+
+
+def test_an_unreachable_node_is_not_blamed_on_the_device():
+    """On mobile data the node's LAN address is paired and simply never answers:
+    a node that is down, not a permission."""
+    assert _run({"classify": [[PHONE_MOBILE, NODE, EVERY_NODE_ADDRESS]]}) == [False]
+
+
+def test_without_an_ipv4_of_our_own_no_absence_means_anything():
+    """No IPv4 candidate here, so no IPv4 address of the node could pair."""
+    assert _run({"classify": [[PHONE_V6_ONLY, NODE, ["2a01:cb15:80e8:de00:d455:bcab:c324:4d72"]]]}) \
+        == [False]
+
+
+def test_a_node_with_no_public_address_is_judged_by_the_subnet():
+    """With nothing paired at all, only an address on our own subnet tells."""
+    other_lan = _sdp(("10.0.0.5", "host"))
+    assert _run({"classify": [[PHONE_WIFI, NODE_NO_STUN, []],
+                              [other_lan, NODE_NO_STUN, []]]}) == [True, False]
+
+
+def test_connect_fails_with_the_reason():
+    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED})
+    assert out["reason"] == "local_network_blocked", out
+
+
+def test_a_connection_that_did_get_through_is_not_a_refusal():
+    """ICE connected once: whatever failed after, the local network was reached."""
+    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
+                "iceConnects": True})
+    assert out["reason"] is None and out["statsCalls"] == 0, out
+
+
+def test_stats_that_cannot_be_read_leave_the_old_error():
+    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
+                "statsThrow": True})
+    assert out["reason"] is None and out["outcome"] == "DataChannel closed", out