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
121
122
123
124
125
126
127
128
129
130
131
132
133
|
"""
Casting in the Android application, pinned by reading the source.
The relay is a port of meshbay-client/src/cast-relay.js and its mitigations are
the same ones; the JVM tests (CastRelayTest) run it on loopback. What is held
here is the wiring around it: the same bridge surface as the desktop, order
kept where order is meaning, the receiver pointed at nothing but the relay, and
what keeps a cast alive with the screen off switched on only while one runs.
"""
import re
from pathlib import Path
import pytest
PACKAGES = Path(__file__).resolve().parents[2]
MAIN = PACKAGES / "meshbay-android" / "app" / "src" / "main"
SRC = MAIN / "kotlin" / "org" / "meshbay" / "client"
CAST = SRC / "cast"
SHIM = MAIN / "assets" / "bridge" / "meshbay-bridge.js"
DESKTOP = PACKAGES / "meshbay-client" / "src"
pytestmark = pytest.mark.skipif(not CAST.exists(), reason="android sources not present")
def _read(path: Path) -> str:
return path.read_text(encoding="utf-8")
def test_the_cast_channels_are_the_desktops():
def channels(text: str, call: str) -> set[str]:
return set(re.findall(rf"{call}\('(cast:[\w:-]+)'", text))
preload = channels(_read(DESKTOP / "preload.js"), r"ipcRenderer\.invoke")
shim = channels(_read(SHIM), "call")
kt = _read(CAST / "CastChannels.kt")
native = set(re.findall(r'^\s*"(cast:[\w:-]+)" ->', kt, flags=re.M))
assert preload and shim == preload, shim ^ preload
assert native == preload, native ^ preload
def test_no_cast_object_where_casting_cannot_work():
"""`platform.cast.available` is whether the object exists."""
shim = _read(SHIM)
assert "...(CAST ? { cast: {" in shim
assert "lanCast: CAST" in shim
assert "const CAST = ${cast.control.available()}" in _read(SRC / "MainActivity.kt")
def test_the_relay_keeps_the_desktop_mitigations():
relay = _read(CAST / "CastRelay.kt")
desktop = _read(DESKTOP / "cast-relay.js")
for name in ("RING_CAP", "PORT_BASE", "PORT_COUNT"):
js = re.search(rf"const {name} = (\d+);", desktop).group(1)
assert re.search(rf"const val {name} = {js}\b", relay), name
assert "InetSocketAddress(address, PORT_BASE + i)" in relay, "bound to the LAN address"
# Code, not comments: the comment saying "never 0.0.0.0" is not a bind.
code = re.sub(r"/\*.*?\*/", "", relay, flags=re.S)
code = re.sub(r"(?m)//.*$", "", code)
assert "0.0.0.0" not in code
assert 'query["t"] != token' in relay
# The preflight before the token: a refusal there reads as a network error.
serve = relay.split("private fun serve(", 1)[1]
assert serve.index('method == "OPTIONS"') < serve.index('query["t"] != token')
def test_what_a_receiver_has_not_read_waits_on_disk():
"""The desktop drops a fragment once 8 MB wait for a receiver; a fragment
is 5 to 10 MB at a film's bitrate, so the television froze for its length
(measured: three dropped in the first fifteen seconds). Here the lead waits
in a spool file per receiver, deleted with it, and is dropped only past a
disk bound."""
relay = _read(CAST / "CastRelay.kt")
assert "BACKPRESSURE_HIGH" not in relay
assert "RandomAccessFile(file, \"rw\").channel" in relay
assert "c.waiting() > spoolLimit()" in relay
assert "SPOOL_MAX_BYTES = 2L * 1024 * 1024 * 1024" in relay
assert "file.delete()" in relay
assert 'java.io.File(context.cacheDir, "cast-relay")' in _read(CAST / "CastChannels.kt")
def test_the_backlog_is_bounded_in_bytes():
"""A fragment is a segment, megabytes at a film's bitrate: 64 of them killed
the application with an OutOfMemoryError on the emulator."""
relay = _read(CAST / "CastRelay.kt")
assert "RING_MAX_BYTES" in relay and "ringBytes > RING_MAX_BYTES" in relay
def test_the_relay_is_on_wifi_never_the_mobile_network():
channels = _read(CAST / "CastChannels.kt")
lan = channels.split("private fun lanAddress()", 1)[1]
assert "TRANSPORT_WIFI" in lan and "TRANSPORT_ETHERNET" in lan
assert "TRANSPORT_CELLULAR" not in lan
def test_the_receiver_is_pointed_at_the_relay_and_nothing_else():
channels = _read(CAST / "CastChannels.kt")
check = channels.split("private fun relayUrl(", 1)[1].split("\n }", 1)[0]
assert "asked != ours" in check and "throw Refused" in check
assert channels.count("relayUrl(o)") == 2, "connect and reload both go through the check"
def test_relay_calls_keep_their_order():
"""The page does not await each push; on a pool they could overtake."""
channels = _read(CAST / "CastChannels.kt")
assert '"cast:start", "cast:push", "cast:subtitle", "cast:finish", "cast:stop"' in channels
bridge = _read(SRC / "bridge" / "Bridge.kt")
assert "Executors.newSingleThreadExecutor()" in bridge
assert "(if (ordered) serial else work).execute" in bridge
assert "fun ordered(frame: BinaryFrame) = frame.channel == BinaryFrame.CAST_PUSH" in _read(
SRC / "bridge" / "Channels.kt")
def test_screen_off_survival_is_switched_on_only_while_casting():
activity = _read(SRC / "MainActivity.kt")
casting = activity.split("private fun casting(on: Boolean)", 1)[1].split("\n }", 1)[0]
assert "web.keepVisible = on" in casting
assert "startForegroundService(service) else stopService(service)" in casting
assert activity.count("keepVisible =") == 1, "the page is kept visible from one place only"
view = _read(SRC / "shell" / "ShellWebView.kt")
assert "var keepVisible = false" in view
manifest = _read(MAIN / "AndroidManifest.xml")
assert 'android:name=".cast.CastService"' in manifest
assert 'android:foregroundServiceType="mediaPlayback"' in manifest
def test_the_receiver_is_given_what_the_desktop_gives_it():
control = _read(CAST / "CastControl.kt")
assert "DEFAULT_MEDIA_RECEIVER_APPLICATION_ID" in control
assert "MediaInfo.STREAM_TYPE_LIVE" in control
assert "TEXT_TRACK_ID = 1L" in control
assert "SCAN_DURATION_MS = 6000L" in control
assert re.search(r"const SCAN_DURATION_MS = 6000;", _read(DESKTOP / "cast-chromecast.js"))
|