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
|
"""
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 "BACKPRESSURE_HIGH = 8L * 1024 * 1024" in relay
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_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"))
|