Package bounded X4 BLE diagnostics and USB mode reader
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,169 @@
|
||||
"""No second query without full identity; disconnect only the owned connection."""
|
||||
|
||||
import importlib.util
|
||||
import struct
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1] / "packaging"
|
||||
|
||||
|
||||
def load(name):
|
||||
spec = importlib.util.spec_from_file_location(name, ROOT / (name + ".py"))
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
diag, codec = load("ble_diagnostic"), load("ble_options")
|
||||
sys.modules["ble_options"] = codec
|
||||
TARGET = "/org/bluez/hci0/dev_00_00_00_00_00_01"
|
||||
SERVICE = TARGET + "/service1"
|
||||
WRITE = SERVICE + "/char1"
|
||||
NOTIFY = SERVICE + "/char2"
|
||||
|
||||
|
||||
def fixture(number, value):
|
||||
if isinstance(value, bytes):
|
||||
return codec.varint(number * 8 + 2) + codec.varint(len(value)) + value
|
||||
return codec.varint(number * 8) + codec.varint(value)
|
||||
|
||||
|
||||
class Fake:
|
||||
def __init__(self, serial=b"synthetic-x4-001", notifying=False, timeout=False):
|
||||
self.inspector = diag.Inspector.__new__(diag.Inspector)
|
||||
ins = self.inspector
|
||||
ins.report = {"vendor_writes": 0, "notify_subscriptions": 0}
|
||||
ins.dbus = SimpleNamespace(Array=lambda x, **kw: x, Dictionary=lambda x, **kw: x)
|
||||
ins.bus = self
|
||||
ins.iface = lambda path, name: self
|
||||
ins.pump = lambda duration, predicate: predicate()
|
||||
ins.owned_connection, ins.owned_notify, ins.scanning = TARGET, None, False
|
||||
ins.objects = lambda: {
|
||||
SERVICE: {diag.SERVICE: {"Device": TARGET, "UUID": "0000be80" + diag.UUID_BASE}},
|
||||
WRITE: {
|
||||
diag.CHAR: {
|
||||
"Service": SERVICE,
|
||||
"UUID": "0000be81" + diag.UUID_BASE,
|
||||
"Flags": ["write"],
|
||||
}
|
||||
},
|
||||
NOTIFY: {
|
||||
diag.CHAR: {
|
||||
"Service": SERVICE,
|
||||
"UUID": "0000be82" + diag.UUID_BASE,
|
||||
"Flags": ["notify"],
|
||||
"Notifying": notifying,
|
||||
}
|
||||
},
|
||||
}
|
||||
self.serial, self.timeout = serial, timeout
|
||||
self.calls = []
|
||||
|
||||
def add_signal_receiver(self, callback, **kwargs):
|
||||
self.callback = callback
|
||||
|
||||
def StartNotify(self, **kwargs):
|
||||
self.calls.append("notify-start")
|
||||
|
||||
def StopNotify(self, **kwargs):
|
||||
self.calls.append("notify-stop")
|
||||
|
||||
def Disconnect(self, **kwargs):
|
||||
self.calls.append("disconnect")
|
||||
|
||||
def close(self):
|
||||
self.calls.append("bus-close")
|
||||
|
||||
def WriteValue(self, raw, options, **kwargs):
|
||||
_, _, _, command, _, ident, _ = struct.unpack("<IBHHBIH", raw[:16])
|
||||
self.calls.append(command)
|
||||
if self.timeout:
|
||||
raise TimeoutError("synthetic timeout")
|
||||
fields = codec.protobuf_fields(raw[16:])
|
||||
requested = [x[2] for x in fields]
|
||||
values = {15: self.serial, 48: b"Insta360 X4", 95: 1, 97: 1}
|
||||
payload = b"".join(fixture(1, n) for n in requested)
|
||||
payload += fixture(2, b"".join(fixture(n, values[n]) for n in requested))
|
||||
reply = (
|
||||
struct.pack("<IBHHBIH", 16 + len(payload), 4, 0, 200, 2, ident | 0x40000000, 0)
|
||||
+ payload
|
||||
)
|
||||
self.callback(diag.CHAR, {"Value": reply}, [])
|
||||
|
||||
|
||||
class DiagnosticTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
mock = patch.object(diag, "usb_snapshot", return_value={"devices": []})
|
||||
mock.start()
|
||||
self.addCleanup(mock.stop)
|
||||
|
||||
def test_usb_sdk_owner_blocks_radio_command(self):
|
||||
fake = Fake()
|
||||
expected = "instax4_" + diag.hashlib.sha256(fake.serial).hexdigest()[:32]
|
||||
with (
|
||||
patch.object(
|
||||
diag,
|
||||
"usb_snapshot",
|
||||
return_value={
|
||||
"devices": [
|
||||
{
|
||||
"idVendor": "2e1a",
|
||||
"idProduct": "0002",
|
||||
"serial": fake.serial.decode(),
|
||||
}
|
||||
]
|
||||
},
|
||||
),
|
||||
self.assertRaisesRegex(RuntimeError, "preserve its current owner"),
|
||||
):
|
||||
fake.inspector.read_options(TARGET, expected)
|
||||
fake.inspector.close()
|
||||
self.assertNotIn(8, fake.calls)
|
||||
|
||||
def test_full_identity_precedes_usb_query_and_cleanup(self):
|
||||
fake = Fake()
|
||||
expected = "instax4_" + diag.hashlib.sha256(fake.serial).hexdigest()[:32]
|
||||
fake.inspector.read_options(TARGET, expected)
|
||||
fake.inspector.close()
|
||||
self.assertEqual(
|
||||
fake.calls, ["notify-start", 8, 8, "notify-stop", "disconnect", "bus-close"]
|
||||
)
|
||||
self.assertTrue(fake.inspector.report["identity_match"])
|
||||
|
||||
def test_wrong_camera_cannot_reach_usb_query(self):
|
||||
fake = Fake()
|
||||
with self.assertRaisesRegex(RuntimeError, "does not match"):
|
||||
fake.inspector.read_options(TARGET, "instax4_" + "0" * 32)
|
||||
fake.inspector.close()
|
||||
self.assertEqual(fake.calls.count(8), 1)
|
||||
self.assertNotIn("usb_options", fake.inspector.report)
|
||||
|
||||
def test_ambiguous_write_timeout_never_retried(self):
|
||||
fake = Fake(timeout=True)
|
||||
with self.assertRaises(TimeoutError):
|
||||
fake.inspector.read_options(TARGET, "instax4_" + "0" * 32)
|
||||
fake.inspector.close()
|
||||
self.assertEqual(fake.calls.count(8), 1)
|
||||
self.assertEqual(fake.inspector.report["vendor_writes"], 1)
|
||||
|
||||
def test_existing_notify_owner_is_preserved(self):
|
||||
fake = Fake(notifying=True)
|
||||
with self.assertRaisesRegex(RuntimeError, "already owned"):
|
||||
fake.inspector.read_options(TARGET, "instax4_" + "0" * 32)
|
||||
fake.inspector.close()
|
||||
self.assertNotIn("notify-stop", fake.calls)
|
||||
self.assertNotIn(8, fake.calls)
|
||||
|
||||
def test_discovery_filters_exact_model_names(self):
|
||||
for name in ("X4 ABC123", "X4 123456"):
|
||||
self.assertTrue(diag.camera_name(name))
|
||||
for name in ("X4 Air 123456", "X5 123456", "X4", "Insta360", None):
|
||||
self.assertFalse(diag.camera_name(name))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Synthetic malformed/partial replies must never authorize a camera mutation."""
|
||||
|
||||
import importlib.util
|
||||
import struct
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
path = Path(__file__).resolve().parents[1] / "packaging/ble_options.py"
|
||||
spec = importlib.util.spec_from_file_location("ble_options", path)
|
||||
codec = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(codec)
|
||||
|
||||
|
||||
def field(number, value):
|
||||
if isinstance(value, bytes):
|
||||
return codec.varint(number * 8 + 2) + codec.varint(len(value)) + value
|
||||
return codec.varint(number * 8) + codec.varint(value)
|
||||
|
||||
|
||||
def reply(payload, code=200, ident=1):
|
||||
return (
|
||||
struct.pack("<IBHHBIH", 16 + len(payload), 4, 0, code, 2, 0xC0000000 | ident, 0) + payload
|
||||
)
|
||||
|
||||
|
||||
class OptionsTests(unittest.TestCase):
|
||||
def test_request_matches_sdk_header_and_proto_descriptors(self):
|
||||
# Fixed independent vector: total18, message8, protobuf2, ID1/end, option95.
|
||||
self.assertEqual(
|
||||
codec.get_options_packet([95], 1).hex(), "12000000040000080002010000800000085f"
|
||||
)
|
||||
|
||||
def test_capture_and_arbitrary_options_are_not_encodable(self):
|
||||
for options in ([], [4], [95, 95], [True], [65536]):
|
||||
with self.assertRaises(ValueError):
|
||||
codec.get_options_packet(options, 1)
|
||||
|
||||
def test_fragmented_and_combined_notifications(self):
|
||||
raw = reply(field(1, 95) + field(2, field(95, 1)))
|
||||
reader = codec.PacketReader()
|
||||
self.assertEqual(reader.feed(raw[:3]), [])
|
||||
packets = reader.feed(raw[3:] + raw)
|
||||
self.assertEqual(len(packets), 2)
|
||||
self.assertEqual(
|
||||
codec.decode_options_response(packets[0], 1, [95])[95],
|
||||
{"acknowledged": True, "present": True, "value": 1},
|
||||
)
|
||||
|
||||
def test_absent_scalar_is_not_zero(self):
|
||||
packet = codec.PacketReader().feed(reply(field(1, 95) + field(2, b"")))[0]
|
||||
self.assertEqual(
|
||||
codec.decode_options_response(packet, 1, [95])[95],
|
||||
{"acknowledged": True, "present": False, "value": None},
|
||||
)
|
||||
|
||||
def test_explicit_zero_and_packed_option_list(self):
|
||||
packet = codec.PacketReader().feed(reply(field(1, b"\x5f") + field(2, field(95, 0))))[0]
|
||||
self.assertEqual(codec.decode_options_response(packet, 1, [95])[95]["value"], 0)
|
||||
|
||||
def test_error_and_wrong_session_are_not_success(self):
|
||||
for code, ident in [(400, 1), (500, 1), (200, 2)]:
|
||||
packet = codec.PacketReader().feed(
|
||||
reply(field(1, 95) + field(2, field(95, 1)), code, ident)
|
||||
)[0]
|
||||
with self.assertRaises(ValueError):
|
||||
codec.decode_options_response(packet, 1, [95])
|
||||
|
||||
def test_unacknowledged_duplicate_and_unknown_enums_rejected(self):
|
||||
for data in (
|
||||
field(2, field(95, 1)),
|
||||
field(1, 95) + field(2, field(95, 1) * 2),
|
||||
field(1, 95) + field(2, field(95, 3)),
|
||||
field(1, 95) * 2 + field(2, field(95, 1)),
|
||||
field(1, 15) + field(2, field(95, 1)),
|
||||
):
|
||||
packet = codec.PacketReader().feed(reply(data))[0]
|
||||
with self.assertRaises(ValueError):
|
||||
codec.decode_options_response(packet, 1, [95])
|
||||
|
||||
def test_unknown_framing_rejected_without_resync(self):
|
||||
for raw in (b"\xff\x07\x40\x10", b"\x00\x10\x01\x00", codec.get_options_packet([95], 1)):
|
||||
with self.assertRaises(ValueError):
|
||||
codec.PacketReader().feed(raw)
|
||||
|
||||
def test_malformed_protobuf_rejected(self):
|
||||
for raw in (b"\x08\x80", b"\x12\x10\x00", b"\x00", b"\x08" + b"\xff" * 10):
|
||||
with self.assertRaises(ValueError):
|
||||
codec.protobuf_fields(raw)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user