1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
"""
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
def test_the_refusal_is_read_before_ice_gives_up():
"""Search stops waiting after ten seconds without progress, and ICE gives up
later: the signature is in the pairs as soon as checks begin."""
refused = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
"probeAt": 5000})
assert refused["early"] is True, refused
paired = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": EVERY_NODE_ADDRESS,
"probeAt": 5000})
assert paired["early"] is False, paired
reached = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
"iceConnects": True, "probeAt": 5000})
assert reached["early"] is False, reached
|