403 lines
16 KiB
Python
403 lines
16 KiB
Python
"""Bounded X4 BLE discovery/GATT inspection; optional allowlisted option reads.
|
|
|
|
Executed by the versioned diagnostic artifact or the installed model package.
|
|
Default: standard Device Information only. Explicit read-options admits command8
|
|
for identity, then USB mode/wakeup only after exact serial identity matches.
|
|
No pairing, camera setters, adapter power, trust, routing or service edits.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
if "__file__" in globals():
|
|
# Installed entry also works with python -I. Pyz injects its verified codec.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
BLUEZ = "org.bluez"
|
|
ADAPTER = "org.bluez.Adapter1"
|
|
DEVICE = "org.bluez.Device1"
|
|
SERVICE = "org.bluez.GattService1"
|
|
CHAR = "org.bluez.GattCharacteristic1"
|
|
PROPS = "org.freedesktop.DBus.Properties"
|
|
MANAGER = "org.freedesktop.DBus.ObjectManager"
|
|
UUID_BASE = "-0000-1000-8000-00805f9b34fb"
|
|
|
|
|
|
def timestamp():
|
|
return {"utc": datetime.now(UTC).isoformat(), "monotonic": time.monotonic()}
|
|
|
|
|
|
def camera_name(value):
|
|
return isinstance(value, str) and re.fullmatch(r"X4 [A-Za-z0-9]{6}", value) is not None
|
|
|
|
|
|
def usb_snapshot():
|
|
devices = []
|
|
for path in Path("/sys/bus/usb/devices").glob("*"):
|
|
try:
|
|
if (path / "product").read_text().strip() != "Insta360 X4":
|
|
continue
|
|
item = {"port": path.name}
|
|
for key in ("idVendor", "idProduct", "serial", "speed"):
|
|
if (path / key).exists():
|
|
item[key] = (path / key).read_text().strip()
|
|
devices.append(item)
|
|
except OSError:
|
|
continue
|
|
return {"devices": devices, **timestamp()}
|
|
|
|
|
|
def assert_no_usb_owner(expected_id):
|
|
for device in usb_snapshot()["devices"]:
|
|
serial = device.get("serial", "")
|
|
ident = "instax4_" + hashlib.sha256(serial.encode()).hexdigest()[:32]
|
|
if (device.get("idVendor"), device.get("idProduct")) == ("2e1a", "0002") and (
|
|
ident == expected_id or not serial
|
|
):
|
|
raise RuntimeError("Target USB SDK mode is present; preserve its current owner")
|
|
|
|
|
|
def plain(value):
|
|
if isinstance(value, (bytes, bytearray)):
|
|
return list(value)
|
|
if isinstance(value, dict):
|
|
return {str(k): plain(v) for k, v in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [plain(v) for v in value]
|
|
if isinstance(value, str):
|
|
return str(value)
|
|
if isinstance(value, int):
|
|
return int(value)
|
|
return value
|
|
|
|
|
|
class Inspector:
|
|
def __init__(self, report):
|
|
# System packages declared by the artifact manifest and Debian package.
|
|
# Missing prerequisites are an error, never an implicit pip/apt repair.
|
|
import dbus
|
|
from dbus.mainloop.glib import DBusGMainLoop
|
|
from gi.repository import GLib
|
|
|
|
DBusGMainLoop(set_as_default=True)
|
|
self.dbus, self.glib = dbus, GLib
|
|
self.bus = dbus.SystemBus(private=True)
|
|
self.manager = self.iface("/", MANAGER)
|
|
self.report, self.fresh = report, {}
|
|
self.adapter = None
|
|
self.scanning = False
|
|
self.owned_connection = None
|
|
self.owned_notify = None
|
|
|
|
def iface(self, path, name):
|
|
return self.dbus.Interface(self.bus.get_object(BLUEZ, path), name)
|
|
|
|
def objects(self):
|
|
return self.manager.GetManagedObjects(timeout=5)
|
|
|
|
def pump(self, duration, predicate=None):
|
|
deadline = time.monotonic() + duration
|
|
next_usb = 0
|
|
context = self.glib.MainContext.default()
|
|
while time.monotonic() < deadline:
|
|
if time.monotonic() >= next_usb:
|
|
self.report.setdefault("usb_samples", []).append(usb_snapshot())
|
|
next_usb = time.monotonic() + 1
|
|
for _ in range(100):
|
|
if not context.pending():
|
|
break
|
|
context.iteration(False)
|
|
if predicate is not None and predicate():
|
|
return True
|
|
time.sleep(0.05)
|
|
return False
|
|
|
|
def remember(self, path, props):
|
|
if not camera_name(str(props.get("Name", ""))):
|
|
return
|
|
self.fresh[str(path)] = {"observed": timestamp(), "properties": plain(props)}
|
|
|
|
def added(self, path, interfaces):
|
|
if DEVICE in interfaces:
|
|
self.remember(path, interfaces[DEVICE])
|
|
|
|
def changed(self, interface, changes, invalidated, path=None):
|
|
if interface != DEVICE or not {"RSSI", "ManufacturerData", "ServiceData", "Name"} & set(
|
|
changes
|
|
):
|
|
return
|
|
props = self.objects().get(path, {}).get(DEVICE, {})
|
|
self.remember(path, props)
|
|
|
|
def scan(self, duration=20):
|
|
if duration not in {20, 60}:
|
|
raise ValueError("Discovery duration must be 20 or 60 seconds")
|
|
objects = self.objects()
|
|
adapters = [(str(p), x[ADAPTER]) for p, x in objects.items() if ADAPTER in x]
|
|
self.report["adapters"] = [{"path": p, "properties": plain(x)} for p, x in adapters]
|
|
powered = [(p, x) for p, x in adapters if bool(x.get("Powered"))]
|
|
if len(powered) != 1:
|
|
raise RuntimeError("Exactly one powered BLE adapter is required")
|
|
path, props = powered[0]
|
|
self.adapter = self.iface(path, ADAPTER)
|
|
self.bus.add_signal_receiver(
|
|
self.added, dbus_interface=MANAGER, signal_name="InterfacesAdded", bus_name=BLUEZ
|
|
)
|
|
self.bus.add_signal_receiver(
|
|
self.changed,
|
|
dbus_interface=PROPS,
|
|
signal_name="PropertiesChanged",
|
|
bus_name=BLUEZ,
|
|
path_keyword="path",
|
|
)
|
|
self.adapter.SetDiscoveryFilter(
|
|
self.dbus.Dictionary(
|
|
{
|
|
"Transport": "le",
|
|
"Pattern": "X4 ",
|
|
"DuplicateData": False,
|
|
},
|
|
signature="sv",
|
|
),
|
|
timeout=5,
|
|
)
|
|
self.adapter.StartDiscovery(timeout=5)
|
|
self.scanning = True
|
|
self.report["scan_started"] = timestamp()
|
|
self.pump(duration)
|
|
self.adapter.StopDiscovery(timeout=5)
|
|
self.scanning = False
|
|
self.report["scan_finished"] = timestamp()
|
|
self.report["candidates"] = self.fresh
|
|
return self.fresh
|
|
|
|
def inspect(self, address, expected_id, read_options=False):
|
|
assert_no_usb_owner(expected_id)
|
|
candidates = [
|
|
(p, x)
|
|
for p, x in self.fresh.items()
|
|
if str(x["properties"].get("Address", "")).upper() == address
|
|
]
|
|
if len(candidates) != 1:
|
|
raise RuntimeError("Target must have exactly one fresh X4 advertisement")
|
|
path, _ = candidates[0]
|
|
before = self.objects()[path][DEVICE]
|
|
if before.get("Connected") or before.get("Paired") or before.get("Bonded"):
|
|
raise RuntimeError("Existing connection or pairing is outside initial inspection")
|
|
self.report["target"] = {"path": path, "before": plain(before)}
|
|
device = self.iface(path, DEVICE)
|
|
self.owned_connection = path
|
|
device.Connect(timeout=20)
|
|
ready = self.pump(
|
|
10, lambda: bool(self.objects().get(path, {}).get(DEVICE, {}).get("ServicesResolved"))
|
|
)
|
|
if not ready:
|
|
raise RuntimeError("GATT service discovery timed out")
|
|
objects = self.objects()
|
|
admitted = {
|
|
str(p): plain(x)
|
|
for p, x in objects.items()
|
|
if str(p).startswith(path + "/") and (SERVICE in x or CHAR in x)
|
|
}
|
|
self.report["gatt"] = admitted
|
|
self.report["standard_reads"] = []
|
|
for p, interfaces in objects.items():
|
|
char = interfaces.get(CHAR, {})
|
|
service = objects.get(char.get("Service"), {}).get(SERVICE, {})
|
|
if (
|
|
str(service.get("Device")) != path
|
|
or str(service.get("UUID")) != "0000180a" + UUID_BASE
|
|
or "read" not in char.get("Flags", [])
|
|
):
|
|
continue
|
|
uuid = str(char.get("UUID"))
|
|
if uuid not in {
|
|
"00002a24" + UUID_BASE,
|
|
"00002a25" + UUID_BASE,
|
|
"00002a26" + UUID_BASE,
|
|
"00002a29" + UUID_BASE,
|
|
}:
|
|
continue
|
|
value = bytes(
|
|
self.iface(p, CHAR).ReadValue(self.dbus.Dictionary({}, signature="sv"), timeout=5)
|
|
)
|
|
if len(value) > 256:
|
|
raise RuntimeError("Device information exceeds limit")
|
|
item = {"uuid": uuid, "value": value.decode("utf-8", errors="strict"), **timestamp()}
|
|
self.report["standard_reads"].append(item)
|
|
if uuid == "00002a25" + UUID_BASE:
|
|
ident = (
|
|
"instax4_" + hashlib.sha256(value.decode().strip().encode()).hexdigest()[:32]
|
|
)
|
|
self.report["identity_match"] = ident == expected_id
|
|
self.report["target"]["after"] = plain(self.objects().get(path, {}).get(DEVICE, {}))
|
|
if read_options:
|
|
self.read_options(path, expected_id)
|
|
|
|
def read_options(self, path, expected_id):
|
|
from ble_options import PacketReader, decode_options_response, get_options_packet
|
|
|
|
objects = self.objects()
|
|
services = [
|
|
p
|
|
for p, x in objects.items()
|
|
if SERVICE in x
|
|
and str(x[SERVICE].get("Device")) == path
|
|
and str(x[SERVICE].get("UUID")) == "0000be80" + UUID_BASE
|
|
]
|
|
if len(services) != 1:
|
|
raise RuntimeError("Exact BE80 service is unavailable")
|
|
chars = {
|
|
str(x[CHAR].get("UUID")): (p, x[CHAR])
|
|
for p, x in objects.items()
|
|
if CHAR in x and x[CHAR].get("Service") == services[0]
|
|
}
|
|
write_path, write_props = chars["0000be81" + UUID_BASE]
|
|
notify_path, notify_props = chars["0000be82" + UUID_BASE]
|
|
if notify_props.get("Notifying") or "notify" not in notify_props.get("Flags", []):
|
|
raise RuntimeError("Notify characteristic is already owned or unsupported")
|
|
flags = write_props.get("Flags", [])
|
|
write_type = "request" if "write" in flags else "command"
|
|
if write_type == "command" and "write-without-response" not in flags:
|
|
raise RuntimeError("BE81 has no supported write method")
|
|
reader, received, errors = PacketReader(), [], []
|
|
self.report["notifications"] = []
|
|
self.report["requests"] = []
|
|
total = 0
|
|
|
|
def notification(interface, changed, invalidated):
|
|
nonlocal total
|
|
if interface != CHAR or "Value" not in changed or errors:
|
|
return
|
|
raw = bytes(changed["Value"])
|
|
total += len(raw)
|
|
if total > 32768 or len(self.report["notifications"]) >= 64:
|
|
errors.append("Notification bound exceeded")
|
|
return
|
|
self.report["notifications"].append({"hex": raw.hex(), **timestamp()})
|
|
try:
|
|
received.extend(reader.feed(raw))
|
|
self.report["heartbeat_packets"] = reader.heartbeats
|
|
except ValueError as error:
|
|
errors.append(str(error))
|
|
|
|
self.bus.add_signal_receiver(
|
|
notification,
|
|
dbus_interface=PROPS,
|
|
signal_name="PropertiesChanged",
|
|
bus_name=BLUEZ,
|
|
path=notify_path,
|
|
)
|
|
# Temporary CCCD write by BlueZ. Never a camera provisioning operation.
|
|
self.owned_notify = notify_path
|
|
self.iface(notify_path, CHAR).StartNotify(timeout=5)
|
|
self.report["notify_subscriptions"] += 1
|
|
|
|
def query(options, message_id):
|
|
assert_no_usb_owner(expected_id)
|
|
raw = get_options_packet(options, message_id)
|
|
self.report["requests"].append(
|
|
{"hex": raw.hex(), "options": options, "message_id": message_id, **timestamp()}
|
|
)
|
|
# Count intent before the effect; timeout is not a retry signal.
|
|
self.report["vendor_writes"] += 1
|
|
self.iface(write_path, CHAR).WriteValue(
|
|
self.dbus.Array(raw, signature="y"),
|
|
self.dbus.Dictionary({"type": write_type}, signature="sv"),
|
|
timeout=5,
|
|
)
|
|
self.pump(
|
|
5, lambda: bool(errors or any(x["message_id"] == message_id for x in received))
|
|
)
|
|
if errors:
|
|
raise RuntimeError(errors[0])
|
|
matches = [x for x in received if x["message_id"] == message_id]
|
|
if len(matches) != 1:
|
|
raise RuntimeError("Missing or ambiguous response; no retry")
|
|
return decode_options_response(matches[0], message_id, options)
|
|
|
|
identity = query([15, 48], 1)
|
|
self.report["identity_options"] = identity
|
|
serial = identity[15]["value"]
|
|
model = identity[48]["value"]
|
|
if not isinstance(serial, str) or not serial.strip() or model not in {"X4", "Insta360 X4"}:
|
|
raise RuntimeError("Camera did not return full X4 identity")
|
|
ident = "instax4_" + hashlib.sha256(serial.strip().encode()).hexdigest()[:32]
|
|
self.report["identity_match"] = ident == expected_id
|
|
if not self.report["identity_match"]:
|
|
raise RuntimeError("BLE camera does not match the expected USB instance")
|
|
self.report["usb_options"] = query([95, 97], 2)
|
|
|
|
def close(self):
|
|
errors = []
|
|
if self.owned_notify:
|
|
try:
|
|
self.iface(self.owned_notify, CHAR).StopNotify(timeout=5)
|
|
except Exception as error:
|
|
errors.append(str(error))
|
|
if self.owned_connection:
|
|
try:
|
|
self.iface(self.owned_connection, DEVICE).Disconnect(timeout=5)
|
|
except Exception as error:
|
|
errors.append(str(error))
|
|
if self.scanning:
|
|
try:
|
|
self.adapter.StopDiscovery(timeout=5)
|
|
except Exception as error:
|
|
errors.append(str(error))
|
|
self.bus.close()
|
|
self.report["cleanup_errors"] = errors
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--address")
|
|
parser.add_argument("--expected-device-id")
|
|
parser.add_argument("--read-options", action="store_true")
|
|
parser.add_argument("--scan-seconds", type=int, choices=(20, 60), default=20)
|
|
args = parser.parse_args()
|
|
if args.address and (
|
|
not re.fullmatch(r"(?:[0-9A-F]{2}:){5}[0-9A-F]{2}", args.address)
|
|
or not re.fullmatch(r"instax4_[0-9a-f]{32}", args.expected_device_id or "")
|
|
):
|
|
parser.error("Inspection requires an exact address and expected device ID")
|
|
if args.read_options and not args.address:
|
|
parser.error("Option reads require an explicit target")
|
|
report = {
|
|
"schema": "missioncore.insta360.ble-diagnostic/v1",
|
|
"started": timestamp(),
|
|
"vendor_writes": 0,
|
|
"pair_calls": 0,
|
|
"notify_subscriptions": 0,
|
|
"identity_match": None,
|
|
"state": "running",
|
|
}
|
|
inspector = None
|
|
try:
|
|
inspector = Inspector(report)
|
|
inspector.scan(args.scan_seconds)
|
|
if args.address:
|
|
inspector.inspect(args.address, args.expected_device_id, args.read_options)
|
|
report["state"] = "complete"
|
|
except Exception as error:
|
|
report.update(state="error", error=str(error))
|
|
finally:
|
|
if inspector:
|
|
inspector.close()
|
|
if report.get("cleanup_errors"):
|
|
report["state"] = "error"
|
|
report.setdefault("error", "Diagnostic cleanup was incomplete")
|
|
report["finished"] = timestamp()
|
|
print(json.dumps(report), flush=True)
|
|
return 0 if report["state"] == "complete" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|