test_mls_broadcast.py raw
1 """Diagnostic: trace MLS broadcast event delivery through the WASM relay-proxy.
2
3 Publishes a kind 1059 event to a local relay and traces whether the browser's
4 relay-proxy receives and dispatches it. Uses native mlstest binary for signing.
5
6 Run: ./test/run.sh -k mls_broadcast --headed -s
7 """
8
9 import json
10 import os
11 import signal
12 import subprocess
13 import time
14
15 import pytest
16 from selenium import webdriver
17 from selenium.webdriver.common.by import By
18 from selenium.webdriver.firefox.options import Options
19 from selenium.webdriver.firefox.service import Service
20 from selenium.webdriver.support.ui import WebDriverWait
21
22 from conftest import Helpers, RELAY_BIN, STATIC_DIR
23
24 PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
25 MLSTEST_BIN = os.path.join(PROJECT_ROOT, "mlstest")
26 MLS_DATA_DIR = "/tmp/smesh-mlsbcast-test"
27 MLS_PORT = 23336
28
29
30 def _start_relay():
31 if not os.path.exists(RELAY_BIN):
32 pytest.skip(f"Relay binary not found: {RELAY_BIN}")
33 if os.path.exists(MLS_DATA_DIR):
34 subprocess.run(["rm", "-rf", MLS_DATA_DIR], check=True)
35 os.makedirs(MLS_DATA_DIR, exist_ok=True)
36
37 env = os.environ.copy()
38 env["ORLY_DATA_DIR"] = MLS_DATA_DIR
39 env["ORLY_LISTEN"] = "127.0.0.1"
40 env["ORLY_PORT"] = str(MLS_PORT)
41 env["ORLY_STATIC_DIR"] = STATIC_DIR
42 env["ORLY_HTTP_GUARD_ENABLED"] = "false"
43 env["ORLY_MARMOT_OPEN"] = "true"
44
45 proc = subprocess.Popen(
46 [RELAY_BIN], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE
47 )
48
49 import urllib.request
50 url = f"http://127.0.0.1:{MLS_PORT}"
51 for _ in range(30):
52 try:
53 urllib.request.urlopen(url, timeout=1)
54 break
55 except Exception:
56 time.sleep(0.2)
57 else:
58 proc.kill()
59 pytest.fail("Relay failed to start")
60
61 return {"proc": proc, "url": url, "ws": f"ws://127.0.0.1:{MLS_PORT}"}
62
63
64 def _stop_relay(info):
65 proc = info["proc"]
66 if proc.poll() is not None:
67 stderr = proc.stderr.read().decode(errors="replace")
68 if stderr:
69 print(f"\n=== RELAY STDERR ===\n{stderr[-3000:]}\n=== END ===\n")
70 return
71 proc.send_signal(signal.SIGTERM)
72 try:
73 proc.wait(timeout=5)
74 except subprocess.TimeoutExpired:
75 proc.kill()
76
77
78 def _make_browser(headed):
79 opts = Options()
80 if not headed:
81 opts.add_argument("--headless")
82 opts.set_preference("xpinstall.signatures.required", False)
83 opts.set_preference("extensions.autoDisableScopes", 0)
84 opts.set_preference("devtools.console.stdout.content", True)
85 svc = Service(log_output="/dev/null")
86 driver = webdriver.Firefox(options=opts, service=svc)
87 driver.set_page_load_timeout(60)
88 driver.set_script_timeout(30)
89 driver.implicitly_wait(3)
90 return driver
91
92
93 def _bech32_encode(hrp, data_bytes):
94 CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
95
96 def convertbits(data, frombits, tobits, pad=True):
97 acc, bits, ret = 0, 0, []
98 maxv = (1 << tobits) - 1
99 for value in data:
100 acc = (acc << frombits) | value
101 bits += frombits
102 while bits >= tobits:
103 bits -= tobits
104 ret.append((acc >> bits) & maxv)
105 if pad and bits:
106 ret.append((acc << (tobits - bits)) & maxv)
107 return ret
108
109 def polymod(values):
110 GEN = [0x3b6a57b2, 0x26508e6d, 0x1ea119fa, 0x3d4233dd, 0x2a1462b3]
111 chk = 1
112 for v in values:
113 b = chk >> 25
114 chk = ((chk & 0x1ffffff) << 5) ^ v
115 for i in range(5):
116 chk ^= GEN[i] if ((b >> i) & 1) else 0
117 return chk
118
119 def create_checksum(hrp, data):
120 values = [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] + data
121 p = polymod(values + [0, 0, 0, 0, 0, 0]) ^ 1
122 return [(p >> 5 * (5 - i)) & 31 for i in range(6)]
123
124 data5 = convertbits(list(data_bytes), 8, 5)
125 checksum = create_checksum(hrp, data5)
126 return hrp + "1" + "".join(CHARSET[d] for d in data5 + checksum)
127
128
129 def _install_console_interceptor(driver):
130 driver.execute_script("""
131 if (!window.__bcast_logs) {
132 window.__bcast_logs = [];
133 var origLog = console.log;
134 var origWarn = console.warn;
135 var origErr = console.error;
136 console.log = function() {
137 var msg = Array.prototype.slice.call(arguments).join(' ');
138 window.__bcast_logs.push(msg);
139 origLog.apply(console, arguments);
140 };
141 console.warn = function() {
142 var msg = '[WARN] ' + Array.prototype.slice.call(arguments).join(' ');
143 window.__bcast_logs.push(msg);
144 origWarn.apply(console, arguments);
145 };
146 console.error = function() {
147 var msg = '[ERROR] ' + Array.prototype.slice.call(arguments).join(' ');
148 window.__bcast_logs.push(msg);
149 origErr.apply(console, arguments);
150 };
151 }
152 """)
153
154
155 def _get_logs(driver):
156 return driver.execute_script("return window.__bcast_logs || [];")
157
158
159 def _clear_logs(driver):
160 driver.execute_script("window.__bcast_logs = [];")
161
162
163 def _setup_user(driver, relay_url):
164 """Load app, create vault with default identity, return hex pubkey."""
165 driver.get(relay_url)
166 h = Helpers(driver)
167 h.wait_wasm_ready()
168 h.wait_signer_ready()
169 pk = h.ensure_identity()
170 assert pk and len(pk) == 64, f"Failed to get pubkey: {pk}"
171 return pk
172
173
174 @pytest.fixture
175 def bcast_env(request):
176 headed = request.config.getoption("--headed", default=False)
177
178 if not os.path.exists(MLSTEST_BIN):
179 pytest.skip(f"mlstest binary not found: {MLSTEST_BIN}")
180
181 relay = _start_relay()
182 driver = _make_browser(headed)
183
184 yield {"relay": relay, "driver": driver}
185
186 driver.quit()
187 _stop_relay(relay)
188
189
190 def test_mls_broadcast_trace(bcast_env):
191 """Trace kind 1059 broadcast delivery through the WASM relay-proxy."""
192 relay = bcast_env["relay"]
193 driver = bcast_env["driver"]
194 url = relay["url"]
195 ws_url = relay["ws"]
196
197 # 1. Setup identity
198 pk = _setup_user(driver, url)
199 print(f"\nUser pubkey: {pk}")
200
201 # 2. Install console interceptor
202 _install_console_interceptor(driver)
203
204 # 3. Navigate to DM with a dummy peer to trigger MLS subscription
205 dummy_pk_bytes = bytes(range(1, 33))
206 dummy_npub = _bech32_encode("npub", dummy_pk_bytes)
207 print(f"Navigating to /msg/{dummy_npub}")
208 driver.get(f"{url}/msg/{dummy_npub}")
209 time.sleep(3)
210
211 # Re-install interceptor after navigation
212 _install_console_interceptor(driver)
213
214 # 4. Wait for MLS subscription to be established
215 # The MLS worker boots, sends M_INIT, which triggers MLS_SUB to relay-proxy
216 print("Waiting for MLS subscription...")
217 time.sleep(5)
218
219 # 5. Check logs so far
220 logs_before = _get_logs(driver)
221 print(f"\n=== Console logs after MLS setup ({len(logs_before)} entries) ===")
222 for log in logs_before:
223 if any(k in log for k in ["MLS", "mls", "relay-proxy", "M_INIT", "M_SUBSCRIBE", "M_INCOMING"]):
224 print(f" {log}")
225 print("=== end setup logs ===\n")
226
227 # 6. Clear logs, publish kind 1059 from native tool
228 _clear_logs(driver)
229 print(f"Publishing kind 1059 with #p={pk} via mlstest...")
230 result = subprocess.run(
231 [MLSTEST_BIN, ws_url, pk],
232 capture_output=True, text=True, timeout=15
233 )
234 print(f"mlstest stdout: {result.stdout.strip()}")
235 print(f"mlstest stderr: {result.stderr.strip()}")
236 assert result.returncode == 0, f"mlstest failed: {result.stderr}"
237
238 # 7. Wait for broadcast to propagate
239 time.sleep(3)
240
241 # 8. Read console logs and check each stage
242 logs_after = _get_logs(driver)
243 print(f"\n=== Console logs after publish ({len(logs_after)} entries) ===")
244 for log in logs_after:
245 print(f" {log}")
246 print("=== end publish logs ===\n")
247
248 # 9. Diagnostic analysis
249 has_mls_event = any("[wasm-host] relay-proxy -> MLS worker: M_INCOMING" in l for l in logs_after)
250 has_mls_dropped = any("MLS_EVENT dropped" in l for l in logs_after)
251 has_event_any = any("EVENT" in l for l in logs_after)
252
253 print("=== DIAGNOSTIC RESULTS ===")
254 print(f" MLS_EVENT received by supervisor and forwarded: {has_mls_event}")
255 print(f" MLS_EVENT dropped (worker not booted): {has_mls_dropped}")
256 print(f" Any EVENT-related log: {has_event_any}")
257
258 if has_mls_event:
259 print(" VERDICT: relay-proxy -> supervisor -> MLS worker chain is WORKING")
260 print(" The broadcast delivery is functional. Bug is downstream in MLS worker.")
261 elif has_mls_dropped:
262 print(" VERDICT: relay-proxy sent MLS_EVENT but MLS worker was not booted")
263 print(" Fix: ensure _ensureMlsWorker() is called before MLS subscriptions")
264 elif has_event_any:
265 print(" VERDICT: relay-proxy received events but did not dispatch MLS_EVENT")
266 print(" Check: mlsSubIDs map, subscription ID matching, wireConn onEvent handler")
267 else:
268 print(" VERDICT: no event-related logs at all")
269 print(" Possible causes:")
270 print(" - relay-proxy WebSocket did not receive the broadcast")
271 print(" - relay-proxy MLS subscription was not created")
272 print(" - relay subscription uses wrong URL (ws vs wss mismatch)")
273 print(" Check relay stderr for broadcast diagnostics")
274
275 # 10. Also check relay stderr for broadcast diagnostic logs
276 relay_proc = relay["proc"]
277 # Don't wait for the process, just peek at any available stderr
278 import select
279 if hasattr(relay_proc.stderr, 'fileno'):
280 r, _, _ = select.select([relay_proc.stderr], [], [], 0.5)
281 if r:
282 stderr_peek = relay_proc.stderr.read1(4096).decode(errors="replace")
283 print(f"\n=== Relay stderr peek ===\n{stderr_peek}\n=== end ===")
284
285 print("=== END DIAGNOSTIC ===")
286
287 # The test passes regardless - it's diagnostic, not assertive
288 # The output tells us where the chain breaks
289