nik/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-08-08 06:46:11 +00:00
Code Issues Packages Projects Releases Wiki Activity
bitchat-android/tools/release_gate/mesh_lab.py
callebtc 44560b9792 Add Watch peer profiles and verification
2026-07-29 18:55:56 +02:00

927 lines
36 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""ADB-driven mesh test orchestrator for two (or more) live devices.
Drives the debug-only TestHookReceiver in the app
(intent action: com.bitchat.droid.TEST_HOOK) to perform mesh operations:
peer scanning, connect, Noise handshake, DMs, file transfer, broadcast,
announce, and raw packet injection.
Each on-device command writes a JSON result to
cache/testhook/results/<id>.json inside the app sandbox; this module polls
for it via `run-as` and returns the parsed dict.
Typical usage:
python3 tools/release_gate/mesh_lab.py setup --serial-a X --serial-b Y --apk app/build/outputs/apk/debug/app-debug.apk
python3 tools/release_gate/mesh_lab.py scenario dm --serial-a X --serial-b Y
python3 tools/release_gate/mesh_lab.py scenario all --serial-a X --serial-b Y
python3 tools/release_gate/mesh_lab.py cmd --serial X scan --extra timeout_ms=30000
"""
from __future__ import annotations
import argparse
import concurrent.futures
import hashlib
import json
import random
import shlex
import subprocess
import sys
import tempfile
import time
import uuid
from pathlib import Path
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
if str(REPOSITORY_ROOT) not in sys.path:
sys.path.insert(0, str(REPOSITORY_ROOT))
from tools.release_gate.android_lab import APPLICATION_ID, find_adb, run_adb
TEST_HOOK_ACTION = "com.bitchat.droid.TEST_HOOK"
TEST_HOOK_COMPONENT = f"{APPLICATION_ID}/com.bitchat.android.testhook.TestHookReceiver"
RESULTS_DIR = "cache/testhook/results"
DEVICE_TMP_DIR = "/data/local/tmp/meshlab"
APP_FIXTURE_DIR = f"/data/data/{APPLICATION_ID}/cache/fixtures"
WATCH_APPLICATION_ID = "com.bitchat.watch"
WATCH_TEST_HOOK_ACTION = "com.bitchat.watch.TEST_HOOK"
WATCH_TEST_HOOK_COMPONENT = f"{WATCH_APPLICATION_ID}/com.bitchat.watch.testhook.WearTestHookReceiver"
PERMISSIONS = [
"android.permission.BLUETOOTH_SCAN",
"android.permission.BLUETOOTH_CONNECT",
"android.permission.BLUETOOTH_ADVERTISE",
"android.permission.ACCESS_FINE_LOCATION",
"android.permission.ACCESS_COARSE_LOCATION",
"android.permission.POST_NOTIFICATIONS",
"android.permission.NEARBY_WIFI_DEVICES",
"android.permission.RECORD_AUDIO",
]
WATCH_PERMISSIONS = [
"android.permission.BLUETOOTH_SCAN",
"android.permission.BLUETOOTH_CONNECT",
"android.permission.BLUETOOTH_ADVERTISE",
"android.permission.POST_NOTIFICATIONS",
"android.permission.RECORD_AUDIO",
]
class MeshLabError(Exception):
pass
def _shell(serial: str, command: str) -> str:
return run_adb(serial, ["shell", command])
class Device:
"""One ADB-connected device running a debug build with the test hook."""
def __init__(
self,
serial: str,
alias: str,
package: str = APPLICATION_ID,
hook_action: str = TEST_HOOK_ACTION,
hook_component: str = TEST_HOOK_COMPONENT,
permissions: list[str] = PERMISSIONS,
activity_component: str = f"{APPLICATION_ID}/com.bitchat.android.MainActivity",
):
self.serial = serial
self.alias = alias
self.package = package
self.hook_action = hook_action
self.hook_component = hook_component
self.permissions = permissions
self.activity_component = activity_component
# -- app lifecycle ------------------------------------------------------
def install(self, apk: Path) -> None:
result = subprocess.run(
[find_adb(), "-s", self.serial, "install", "-r", "-g", str(apk)],
check=False, capture_output=True, text=True, timeout=300,
)
if result.returncode != 0 or "Success" not in result.stdout:
raise MeshLabError(f"[{self.alias}] install failed: {result.stdout} {result.stderr}")
def grant_permissions(self) -> None:
for perm in self.permissions:
subprocess.run(
[find_adb(), "-s", self.serial, "shell", "pm", "grant", self.package, perm],
check=False, capture_output=True, text=True, timeout=30,
)
def clear_app_data(self) -> None:
_shell(self.serial, f"am force-stop {self.package}")
output = _shell(self.serial, f"pm clear {self.package}")
if "Success" not in output:
raise MeshLabError(f"[{self.alias}] pm clear failed: {output}")
def force_stop(self) -> None:
_shell(self.serial, f"am force-stop {self.package}")
def launch(self) -> None:
"""Launch the app and verify it is actually top-resumed.
A background/cached process can be frozen by the system (observed on Wear OS),
which silently hangs test-hook commands; the foreground activity (and the FGS it
starts) keeps the process unfrozen.
"""
for _attempt in range(3):
_shell(self.serial, f"monkey -p {self.package} -c android.intent.category.LAUNCHER 1")
time.sleep(3)
try:
top = _shell(
self.serial,
"dumpsys activity activities | grep topResumedActivity",
)
if self.package in top:
return
except Exception:
pass
_shell(self.serial, f"am start -n {self.activity_component}")
time.sleep(3)
def wake(self) -> None:
"""Keep the screen on and the app foregrounded (full-power BLE duty cycle).
A backgrounded app drops to POWER_SAVER (1 s scan per 60 s), which makes
mesh reformation after restarts take minutes and scenarios flaky.
`svc power stayon` only applies while charging, so also stretch the
screen timeout as a fallback.
"""
_shell(self.serial, "svc power stayon true")
_shell(self.serial, "settings put system screen_off_timeout 600000")
subprocess.run(
[find_adb(), "-s", self.serial, "shell", "locksettings", "set-disabled", "true"],
check=False, capture_output=True, text=True, timeout=30,
)
_shell(self.serial, "input keyevent KEYCODE_WAKEUP")
_shell(self.serial, "wm dismiss-keyguard")
_shell(self.serial, "input keyevent 82") # dismiss non-secure keyguard
_shell(self.serial, "input swipe 500 1500 500 400") # swipe-up dismiss
def reset_bluetooth(self) -> None:
"""Cycle the BT adapter; clears zombie GATT connections from peer restarts."""
_shell(self.serial, "svc bluetooth disable")
time.sleep(2)
_shell(self.serial, "svc bluetooth enable")
time.sleep(3)
def enable_bluetooth(self) -> None:
subprocess.run(
[find_adb(), "-s", self.serial, "shell", "svc", "bluetooth", "enable"],
check=False, capture_output=True, text=True, timeout=30,
)
# -- fixtures -----------------------------------------------------------
def push_fixture(self, local: Path, name: str | None = None) -> str:
"""Stage a fixture inside the app sandbox and return its app-readable path.
adb push lands files as shell:ext_data_rw, which the app cannot read
through the FUSE Android/data mount, so the bytes are piped through
the shell into the app's own cache directory via run-as.
"""
fname = name or local.name
tmp = f"{DEVICE_TMP_DIR}/{fname}"
_shell(self.serial, f"mkdir -p {DEVICE_TMP_DIR}")
result = subprocess.run(
[find_adb(), "-s", self.serial, "push", str(local), tmp],
check=False, capture_output=True, text=True, timeout=120,
)
if result.returncode != 0:
raise MeshLabError(f"[{self.alias}] push failed: {result.stderr}")
fixture_dir = f"/data/data/{self.package}/cache/fixtures"
target = f"{fixture_dir}/{fname}"
_shell(
self.serial,
f"run-as {self.package} mkdir -p {fixture_dir} && "
f"cat {tmp} | run-as {self.package} sh -c 'cat > {target}' && rm -f {tmp}",
)
return target
def clear_incoming(self) -> None:
_shell(
self.serial,
f"run-as {self.package} rm -rf cache/files/incoming cache/images/incoming",
)
# -- test hook commands -------------------------------------------------
def cmd(self, cmd: str, timeout_ms: int = 60_000, **extras: object) -> dict:
"""Send a test-hook command and poll for its JSON result."""
cmd_id = uuid.uuid4().hex[:12]
_shell(self.serial, f"run-as {self.package} rm -f {RESULTS_DIR}/{cmd_id}.json")
args = [
"am", "broadcast", "-a", self.hook_action,
"-n", self.hook_component,
"--es", "cmd", cmd,
"--es", "id", cmd_id,
"--el", "timeout_ms", str(timeout_ms),
"--el", "overall_timeout_ms", str(timeout_ms + 30_000),
]
for key, value in extras.items():
if value is None:
continue
if isinstance(value, bool):
args += ["--ez", key, "true" if value else "false"]
elif isinstance(value, int):
# `am` stores --el as Long and --ei as Integer; on-device
# readers use getIntExtra, so int extras must go via --ei.
args += ["--ei", key, str(value)]
else:
args += ["--es", key, str(value)]
try:
_shell(self.serial, " ".join(shlex.quote(a) for a in args))
except Exception as error:
# The shell occasionally hangs even though the broadcast was delivered;
# fall through to result polling, which is the authoritative channel.
print(f"[{self.alias}] warning: broadcast send for '{cmd}' raised: {error}", file=sys.stderr)
deadline = time.monotonic() + (timeout_ms + 60_000) / 1000
while time.monotonic() < deadline:
try:
raw = _shell(self.serial, f"run-as {self.package} cat {RESULTS_DIR}/{cmd_id}.json")
if raw.strip().startswith("{"):
return json.loads(raw)
except Exception:
pass
time.sleep(1.0)
raise MeshLabError(f"[{self.alias}] timed out waiting for result of '{cmd}' ({cmd_id})")
def cmd_ok(self, cmd: str, timeout_ms: int = 60_000, **extras: object) -> dict:
result = self.cmd(cmd, timeout_ms=timeout_ms, **extras)
if result.get("status") != "ok":
raise MeshLabError(f"[{self.alias}] '{cmd}' failed: {result}")
return result
def logcat_dump(self, lines: int = 200) -> str:
return _shell(self.serial, f"logcat -d -t {lines}")
class WatchDevice(Device):
"""Pixel Watch running the com.bitchat.watch debug build.
Same test-hook protocol as the phone; different package/hook, a smaller permission
set (Bluetooth + notifications only), and wake tweaks that skip phone-only keyguard
commands. File-transfer scenarios are not supported on the watch yet (M5 deferred).
"""
def __init__(self, serial: str, alias: str = "watch"):
super().__init__(
serial,
alias,
package=WATCH_APPLICATION_ID,
hook_action=WATCH_TEST_HOOK_ACTION,
hook_component=WATCH_TEST_HOOK_COMPONENT,
permissions=WATCH_PERMISSIONS,
activity_component=f"{WATCH_APPLICATION_ID}/.MainActivity",
)
def wake(self) -> None:
# Keep the screen on while on the charging puck; otherwise Wear shows the
# charging activity on top, our app loses foreground, and the OS freezes the
# process (cached-app freezer), silently hanging test-hook commands.
_shell(self.serial, "settings put global stay_on_while_plugged_in 3")
_shell(self.serial, "svc power stayon true")
_shell(self.serial, "settings put system screen_off_timeout 600000")
_shell(self.serial, "input keyevent KEYCODE_WAKEUP")
# MARK: - fixtures
FIXTURE_SIZES = {
"small_1k.bin": 1_024,
"medium_512k.bin": 512 * 1_024,
"large_2m.bin": 2 * 1_024 * 1_024,
}
def make_fixtures(directory: Path, seed: int = 1337, names: list[str] | None = None) -> dict[str, dict]:
directory.mkdir(parents=True, exist_ok=True)
fixtures = {}
rng = random.Random(seed)
for name, size in FIXTURE_SIZES.items():
if names is not None and name not in names:
rng.randbytes(size) # keep the stream deterministic across subsets
continue
path = directory / name
data = rng.randbytes(size)
path.write_bytes(data)
fixtures[name] = {"path": path, "sha256": hashlib.sha256(data).hexdigest(), "bytes": size}
return fixtures
def make_private_media_fixtures(directory: Path, seed: int = 7331) -> dict[str, dict]:
"""Small attachment fixtures covering every private-media UI type."""
directory.mkdir(parents=True, exist_ok=True)
fixtures = {}
rng = random.Random(seed)
for name, mime in (
("voice_note.m4a", "audio/mp4"),
("image_note.jpg", "image/jpeg"),
("document_note.txt", "text/plain"),
):
path = directory / name
data = rng.randbytes(1_024)
path.write_bytes(data)
fixtures[name] = {
"path": path,
"sha256": hashlib.sha256(data).hexdigest(),
"bytes": len(data),
"mime": mime,
}
return fixtures
# MARK: - setup
def setup_pair(
a: Device,
b: Device,
apk_a: Path | None,
nickname_a: str,
nickname_b: str,
apk_b: Path | None = None,
) -> None:
if apk_b is None:
apk_b = apk_a
for device, nickname, apk in ((a, nickname_a, apk_a), (b, nickname_b, apk_b)):
device.reset_bluetooth()
device.enable_bluetooth()
if apk is not None:
device.install(apk)
device.clear_app_data()
device.grant_permissions()
device.wake()
device.launch()
device.cmd_ok("start")
device.cmd_ok("set_nickname", name=nickname)
wait_for_mutual_discovery(a, b)
def whoami(device: Device) -> dict:
return device.cmd_ok("whoami")
def wait_for_peer(device: Device, peer_id: str, timeout_s: int = 90) -> dict:
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
result = device.cmd_ok("peers")
for peer in result.get("peers", []):
if peer.get("id") == peer_id:
return peer
device.cmd_ok("announce")
time.sleep(3)
raise MeshLabError(f"[{device.alias}] peer {peer_id} not discovered within {timeout_s}s")
def wait_for_mutual_discovery(a: Device, b: Device) -> None:
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
fa = pool.submit(wait_for_peer, a, id_b)
fb = pool.submit(wait_for_peer, b, id_a)
fa.result()
fb.result()
# MARK: - scenarios
def scenario_dm(a: Device, b: Device) -> dict:
"""Handshake, then exchange DMs in both directions with content assertions."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
hs = a.cmd_ok("handshake", timeout_ms=60_000, peer=id_b)
hs_back = b.cmd_ok("handshake", timeout_ms=60_000, peer=id_a)
token_ab = f"dm-{uuid.uuid4().hex[:8]}"
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "dm_recv", 60_000, peer=id_a, contains=token_ab)
time.sleep(2)
send = pool.submit(a.cmd_ok, "dm_send", 30_000, peer=id_b, content=f"hello b {token_ab}")
recv_result, send_result = recv.result(), send.result()
assert token_ab in recv_result["content"], recv_result
token_ba = f"dm-{uuid.uuid4().hex[:8]}"
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(a.cmd_ok, "dm_recv", 60_000, peer=id_b, contains=token_ba)
time.sleep(2)
send = pool.submit(b.cmd_ok, "dm_send", 30_000, peer=id_a, content=f"hello a {token_ba}")
recv_result2, send_result2 = recv.result(), send.result()
assert token_ba in recv_result2["content"], recv_result2
return {
"handshake_a_to_b": hs, "handshake_b_to_a": hs_back,
"a_to_b": {"send": send_result, "recv": recv_result},
"b_to_a": {"send": send_result2, "recv": recv_result2},
}
def scenario_favorite_verification(a: Device, b: Device) -> dict:
"""Assert the three-state favorite exchange and local cryptographic verification."""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
identity_a = whoami(a)["identity_fingerprint"]
a.cmd_ok("handshake", timeout_ms=60_000, peer=id_b)
b.cmd_ok("handshake", timeout_ms=60_000, peer=id_a)
def wait_for_status(
device: Device,
command: str,
peer_id: str,
predicate,
timeout_s: int = 30,
) -> dict:
deadline = time.monotonic() + timeout_s
last: dict = {}
while time.monotonic() < deadline:
last = device.cmd_ok(command, peer=peer_id)
if predicate(last):
return last
time.sleep(1)
raise MeshLabError(
f"[{device.alias}] {command} did not reach the expected state: {last}"
)
# Start from a known non-favorite relationship without clearing either app's data.
a.cmd_ok("favorite_set", peer=id_b, enabled=False)
b.cmd_ok("favorite_set", peer=id_a, enabled=False)
neutral_a = wait_for_status(
a,
"favorite_status",
id_b,
lambda state: not state["is_favorite"] and not state["they_favorited_us"],
)
neutral_b = wait_for_status(
b,
"favorite_status",
id_a,
lambda state: not state["is_favorite"] and not state["they_favorited_us"],
)
# A favorites B. B must show the orange outline while its own favorite remains false.
a.cmd_ok("favorite_set", peer=id_b, enabled=True)
received_only = wait_for_status(
b,
"favorite_status",
id_a,
lambda state: (
not state["is_favorite"]
and state["they_favorited_us"]
and state["star_state"] == "outlined_orange"
),
)
# B favorites back. Both relationships become mutual and B's star becomes filled.
b.cmd_ok("favorite_set", peer=id_a, enabled=True)
mutual_b = wait_for_status(
b,
"favorite_status",
id_a,
lambda state: state["is_mutual"] and state["star_state"] == "filled",
)
mutual_a = wait_for_status(
a,
"favorite_status",
id_b,
lambda state: state["is_mutual"] and state["star_state"] == "filled",
)
# The code B displays for A must be A's SHA-256 Noise identity fingerprint.
verification_before = b.cmd_ok("verification_status", peer=id_a)
if verification_before.get("fingerprint") != identity_a:
raise MeshLabError("displayed verification fingerprint does not match peer identity")
b.cmd_ok("verification_set", peer=id_a, enabled=True)
verification_after = wait_for_status(
b,
"verification_status",
id_a,
lambda state: state["verified"],
)
return {
"neutral": {"a": neutral_a, "b": neutral_b},
"received_favorite": received_only,
"mutual": {"a": mutual_a, "b": mutual_b},
"verification": {
"fingerprint_matches_peer_identity": True,
"before": verification_before,
"after": verification_after,
},
}
def scenario_broadcast(a: Device, b: Device) -> dict:
"""Public broadcast from A received by B."""
id_a = whoami(a)["peer_id"]
token = f"bc-{uuid.uuid4().hex[:8]}"
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "msg_recv", 60_000, contains=token)
time.sleep(2)
send = pool.submit(a.cmd_ok, "broadcast_msg", 30_000, content=f"broadcast {token}")
recv_result, send_result = recv.result(), send.result()
assert recv_result["from"] == id_a, recv_result
return {"send": send_result, "recv": recv_result}
def scenario_file(
a: Device,
b: Device,
fixtures: dict[str, dict],
private: bool = False,
recipient: str | None = None,
) -> dict:
"""File transfer A -> B with sha256 integrity verification."""
id_b = whoami(b)["peer_id"]
b.clear_incoming() # avoid name-uniquified collisions across runs
results = {}
for name, fixture in fixtures.items():
remote = a.push_fixture(fixture["path"])
send_kwargs: dict[str, object] = {"path": remote}
if private:
send_kwargs["peer"] = recipient or id_b
if fixture.get("mime"):
send_kwargs["mime"] = fixture["mime"]
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "file_recv", 240_000, name_contains=name)
time.sleep(2)
send = pool.submit(a.cmd_ok, "file_send", 240_000, **send_kwargs)
recv_result, send_result = recv.result(), send.result()
digest_ok = recv_result["sha256"] == fixture["sha256"]
results[name] = {
"send": send_result, "recv": recv_result,
"expected_sha256": fixture["sha256"], "digest_match": digest_ok,
}
if not digest_ok:
raise MeshLabError(
f"file '{name}' digest mismatch: {recv_result['sha256']} != {fixture['sha256']}"
)
return results
def scenario_private_media(a: Device, b: Device) -> dict:
"""Voice, image, and generic file sends through the private-chat contact ID."""
identity = whoami(b)
noise_public_key = bytes.fromhex(identity["noise_public_key"])
conversation_id = f"contact_{hashlib.sha256(noise_public_key).hexdigest()}"
fixtures = make_private_media_fixtures(
Path(tempfile.mkdtemp(prefix="meshlab-private-media-"))
)
return scenario_file(
a,
b,
fixtures,
private=True,
recipient=conversation_id,
)
def scenario_raw(a: Device, b: Device) -> dict:
"""Raw packet injection (unsigned announce-type packet) reaches the mesh."""
payload = b"meshlab-raw-" + uuid.uuid4().hex[:8].encode()
result = a.cmd_ok("raw_send", 30_000, type="05", payload_hex=payload.hex())
return {"send": result}
# MARK: - session / identity churn scenarios
def _dm_roundtrip(a: Device, b: Device, id_a: str, id_b: str) -> dict:
"""Exchange DMs in both directions with content assertions."""
token_ab = f"dm-{uuid.uuid4().hex[:8]}"
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(b.cmd_ok, "dm_recv", 60_000, peer=id_a, contains=token_ab)
time.sleep(2)
send = pool.submit(a.cmd_ok, "dm_send", 30_000, peer=id_b, content=f"hello b {token_ab}")
recv_ab, send_ab = recv.result(), send.result()
assert token_ab in recv_ab["content"], recv_ab
token_ba = f"dm-{uuid.uuid4().hex[:8]}"
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
recv = pool.submit(a.cmd_ok, "dm_recv", 60_000, peer=id_b, contains=token_ba)
time.sleep(2)
send = pool.submit(b.cmd_ok, "dm_send", 30_000, peer=id_a, content=f"hello a {token_ba}")
recv_ba, send_ba = recv.result(), send.result()
assert token_ba in recv_ba["content"], recv_ba
return {"a_to_b": recv_ab, "b_to_a": recv_ba}
def wait_session_established(device: Device, peer_id: str, timeout_s: int = 90) -> dict:
deadline = time.monotonic() + timeout_s
last: dict = {}
while time.monotonic() < deadline:
last = device.cmd_ok("session", peer=peer_id)
if last.get("established"):
return last
time.sleep(2)
raise MeshLabError(
f"[{device.alias}] session with {peer_id} not established within {timeout_s}s (last: {last})"
)
def ensure_direct_link(a: Device, b: Device, id_a: str, id_b: str) -> None:
"""Wait for rediscovery, then force a direct GATT connection both ways.
Backgrounded devices drop to POWER_SAVER duty cycles (1 s scan per 60 s), so
passively waiting for the mesh to reform takes minutes. The explicit connect
makes restart scenarios deterministic. The address↔peer mapping is learned from
direct-link announces and can lag peer-list discovery after a restart, so the
connect attempt is retried while the peer announces.
"""
wait_for_peer(a, id_b, timeout_s=120)
wait_for_peer(b, id_a, timeout_s=120)
for device, peer, announcer in ((a, id_b, b), (b, id_a, a)):
connected = False
last: dict = {}
for _attempt in range(4):
last = device.cmd("connect", timeout_ms=45_000, peer=peer)
if last.get("status") == "ok" and last.get("direct"):
connected = True
break
# Already acceptable if the mesh formed a direct link on its own.
peers = device.cmd_ok("peers").get("peers", [])
match = next((p for p in peers if p.get("id") == peer), None)
if match and match.get("direct"):
connected = True
break
try:
announcer.cmd_ok("announce")
except MeshLabError:
pass
time.sleep(4)
if not connected:
raise MeshLabError(f"[{device.alias}] no direct link to {peer}: connect={last}")
def force_handshake(device: Device, peer_id: str, attempts: int = 5, per_attempt_s: int = 20) -> dict:
"""Retry explicit handshakes; inits can be lost while links settle."""
last: dict = {}
for _ in range(attempts):
last = device.cmd("handshake", timeout_ms=per_attempt_s * 1000, peer=peer_id)
if last.get("status") == "ok":
return last
time.sleep(2)
raise MeshLabError(f"[{device.alias}] handshake with {peer_id} failed after {attempts} attempts (last: {last})")
def scenario_session_recovery(a: Device, b: Device) -> dict:
"""Process death on B: identity must persist, in-memory Noise sessions are lost.
Expected recovery flow: A's DM sent with its stale session is dropped by B
(B has no session and no kick path on pure decrypt failure); B's outgoing DM
auto-triggers a fresh handshake; subsequent DMs must flow both ways.
"""
id_a = whoami(a)["peer_id"]
id_b = whoami(b)["peer_id"]
a.cmd_ok("handshake", 60_000, peer=id_b)
b.cmd_ok("handshake", 60_000, peer=id_a)
baseline = _dm_roundtrip(a, b, id_a, id_b)
b.force_stop()
b.wake()
b.launch()
b.cmd_ok("start")
b.cmd_ok("set_nickname", name="bob")
id_b_after = whoami(b)["peer_id"]
if id_b_after != id_b:
raise MeshLabError(f"identity changed across process death: {id_b} -> {id_b_after}")
wait_for_peer(a, id_b, timeout_s=120)
wait_for_peer(b, id_a, timeout_s=120)
ensure_direct_link(a, b, id_a, id_b)
# A -> B with A's stale session: B lost its in-memory session; drop expected.
a.cmd_ok("dm_send", 30_000, peer=id_b, content=f"stale-{uuid.uuid4().hex[:8]}")
# B -> A: no session on B, sendPrivateMessage auto-fires the re-handshake.
# The fire-and-forget handshake has no retry, so repeat the trigger, then
# fall back to explicit handshake commands if the auto-path stays stuck.
session_b: dict = {}
for _attempt in range(3):
b.cmd_ok("dm_send", 30_000, peer=id_a, content=f"trigger-{uuid.uuid4().hex[:8]}")
try:
session_b = wait_session_established(b, id_a, timeout_s=20)
break
except MeshLabError:
continue
if not session_b:
force_handshake(b, id_a)
session_b = wait_session_established(b, id_a, timeout_s=30)
session_a = wait_session_established(a, id_b)
recovered = _dm_roundtrip(a, b, id_a, id_b)
return {
"identity_preserved": True,
"baseline": baseline,
"session_a": session_a,
"session_b": session_b,
"recovered": recovered,
}
def scenario_identity_reset(a: Device, b: Device) -> dict:
"""pm clear on B mid-session: new identity, rediscovery, fresh handshake and DMs."""
id_a = whoami(a)["peer_id"]
id_b_old = whoami(b)["peer_id"]
a.cmd_ok("handshake", 60_000, peer=id_b_old)
b.cmd_ok("handshake", 60_000, peer=id_a)
_dm_roundtrip(a, b, id_a, id_b_old)
b.clear_app_data()
b.grant_permissions()
b.wake()
b.launch()
b.cmd_ok("start")
b.cmd_ok("set_nickname", name="bob")
id_b_new = whoami(b)["peer_id"]
if id_b_new == id_b_old:
raise MeshLabError("identity survived pm clear")
wait_for_peer(a, id_b_new, timeout_s=180)
ensure_direct_link(a, b, id_a, id_b_new)
force_handshake(a, id_b_new)
force_handshake(b, id_a)
recovered = _dm_roundtrip(a, b, id_a, id_b_new)
# Inspect how A treats the dead peer's stale session (evidence, not an assertion).
stale = a.cmd("session", peer=id_b_old)
return {
"old_peer_id": id_b_old,
"new_peer_id": id_b_new,
"identity_changed": True,
"recovered": recovered,
"stale_session_on_a": stale,
}
def scenario_file_oversize(a: Device, b: Device, fixtures: dict[str, dict]) -> dict:
"""Oversized broadcast file must be rejected sender-side (>256 fragments)."""
fixture = fixtures["medium_512k.bin"]
remote = a.push_fixture(fixture["path"])
send = a.cmd("file_send", timeout_ms=60_000, path=remote)
rejected = send.get("status") == "error" and "rejected" in send.get("error", "")
if not rejected:
raise MeshLabError(f"expected sender-side rejection, got: {send}")
# Receiver must not see any file appear.
recv = b.cmd("file_recv", timeout_ms=15_000, name_contains="medium_512k")
if recv.get("status") == "ok":
raise MeshLabError(f"receiver unexpectedly saved an oversized file: {recv}")
return {"send": send, "receiver_saw_file": False}
SCENARIOS = {
"dm": scenario_dm,
"favorite_verification": scenario_favorite_verification,
"broadcast": scenario_broadcast,
# Broadcast transfers are receiver-capped at 256 fragments (~120 KB); only
# the small fixture is end-to-end receivable.
"file": lambda a, b: scenario_file(
a, b,
make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")), names=["small_1k.bin"]),
),
"file_oversize": lambda a, b: scenario_file_oversize(
a, b, make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")))
),
# Private media is hard-capped at 256 fragments (PrivateMediaTransfer), so only
# the small fixture fits; larger sizes are expected to be rejected by the sender.
"file_private": lambda a, b: scenario_file(
a, b,
make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")), names=["small_1k.bin"]),
private=True,
),
"media_private": scenario_private_media,
"raw": scenario_raw,
"session_recovery": scenario_session_recovery,
"identity_reset": scenario_identity_reset,
}
# Scenarios supported when device B is a watch (file scenarios are receive-only: phone sends,
# the watch must receive with matching digests).
WATCH_SCENARIOS = [
"dm",
"favorite_verification",
"broadcast",
"raw",
"file",
"file_private",
"session_recovery",
"identity_reset",
]
def run_scenario(name: str, a: Device, b: Device, out: Path | None) -> dict:
started = time.time()
evidence: dict[str, object] = {"scenario": name, "devices": [a.alias, b.alias]}
try:
supported = WATCH_SCENARIOS if isinstance(b, WatchDevice) else list(SCENARIOS)
if name == "all":
results = {}
failures = []
for n in supported:
sub = run_scenario(n, a, b, out)
results[n] = sub.get("results", {"error": sub.get("error", "unknown")})
if sub["status"] != "pass":
failures.append(n)
evidence["results"] = results
if failures:
raise MeshLabError(f"sub-scenarios failed: {', '.join(failures)}")
elif name not in supported:
raise MeshLabError(f"scenario '{name}' is not supported on device '{b.alias}'")
else:
evidence["results"] = SCENARIOS[name](a, b)
evidence["status"] = "pass"
except (MeshLabError, AssertionError) as error:
evidence["status"] = "fail"
evidence["error"] = str(error)
evidence["logcat"] = {d.alias: d.logcat_dump() for d in (a, b)}
evidence["duration_s"] = round(time.time() - started, 1)
if out is not None:
out.mkdir(parents=True, exist_ok=True)
(out / f"{name}-evidence.json").write_text(json.dumps(evidence, indent=2, default=str))
return evidence
# MARK: - CLI
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
commands = parser.add_subparsers(dest="command", required=True)
setup = commands.add_parser("setup", help="install, grant, launch, nickname, discover")
setup.add_argument("--serial-a", required=True)
setup.add_argument("--serial-b")
setup.add_argument("--serial-watch", help="watch serial; used as device B (overrides --serial-b)")
setup.add_argument("--apk", type=Path, default=None)
setup.add_argument("--watch-apk", type=Path, default=None)
setup.add_argument("--nickname-a", default="alice")
setup.add_argument("--nickname-b", default="bob")
scenario = commands.add_parser("scenario", help="run a test scenario on two devices")
scenario.add_argument("name", choices=[*SCENARIOS.keys(), "all"])
scenario.add_argument("--serial-a", required=True)
scenario.add_argument("--serial-b")
scenario.add_argument("--serial-watch", help="watch serial; used as device B (overrides --serial-b)")
scenario.add_argument("--out", type=Path, default=None, help="evidence output directory")
raw = commands.add_parser("cmd", help="send a raw test-hook command to one device")
raw.add_argument("--serial", required=True)
raw.add_argument("cmd")
raw.add_argument("--extra", action="append", default=[], help="key=value string extra (repeatable)")
raw.add_argument("--extra-int", action="append", default=[], help="key=value int extra (repeatable)")
raw.add_argument("--timeout-ms", type=int, default=60_000)
return parser
def _resolve_devices(args: argparse.Namespace) -> tuple[Device, Device]:
"""Device A is always the phone; device B is a watch when --serial-watch is given."""
a = Device(args.serial_a, "alpha")
if getattr(args, "serial_watch", None):
return a, WatchDevice(args.serial_watch)
if not getattr(args, "serial_b", None):
raise MeshLabError("either --serial-b or --serial-watch is required")
return a, Device(args.serial_b, "beta")
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "setup":
a, b = _resolve_devices(args)
nickname_b = "watch" if isinstance(b, WatchDevice) and args.nickname_b == "bob" else args.nickname_b
setup_pair(
a, b, args.apk, args.nickname_a, nickname_b,
apk_b=args.watch_apk if isinstance(b, WatchDevice) else None,
)
print(json.dumps({"status": "ok", "step": "setup"}))
elif args.command == "scenario":
a, b = _resolve_devices(args)
evidence = run_scenario(args.name, a, b, args.out)
print(json.dumps(evidence, indent=2, default=str))
return 0 if evidence["status"] == "pass" else 1
elif args.command == "cmd":
extras: dict[str, object] = {}
extras: dict[str, object] = {}
for item in args.extra:
key, _, value = item.partition("=")
extras[key] = value
for item in args.extra_int:
key, _, value = item.partition("=")
extras[key] = int(value)
result = Device(args.serial, "device").cmd(args.cmd, timeout_ms=args.timeout_ms, **extras)
print(json.dumps(result, indent=2, default=str))
return 0 if result.get("status") == "ok" else 1
return 0
except MeshLabError as error:
print(f"mesh lab error: {error}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink