"""One bounded X4 wake advertisement; no camera GATT writes or power-off.""" import hashlib import json import re import signal import sys from pathlib import Path if "__file__" in globals(): sys.path.insert(0, str(Path(__file__).resolve().parent)) from ble_diagnostic import ( # noqa: E402 ADAPTER, DEVICE, PROPS, Inspector, assert_no_usb_owner, plain, timestamp, usb_snapshot, ) ADV = "org.bluez.LEAdvertisement1" ADV_MANAGER = "org.bluez.LEAdvertisingManager1" def wake_payload(target): if set(target) != {"device_id", "serial", "bluetooth_address", "wakeup_enabled"}: raise ValueError("Exact private target fields are required") serial = target["serial"] if not isinstance(serial, str) or not re.fullmatch(r"[A-Za-z0-9]{7,64}", serial): raise ValueError("Full USB serial is required, never a suffix alone") ident = "instax4_" + hashlib.sha256(serial.encode()).hexdigest()[:32] if target["device_id"] != ident or target["wakeup_enabled"] is not True: raise ValueError("USB identity and enabled camera wakeup must be confirmed") if target["bluetooth_address"] is not None and not re.fullmatch( r"(?:[0-9A-F]{2}:){5}[0-9A-F]{2}", target["bluetooth_address"] ): raise ValueError("Prior correlated Bluetooth address is required") # Company ID 0x004c is passed separately to BlueZ. X4 remote spec section1. return ( bytes.fromhex("0215094f5242495409ff0f00") + serial[-6:].encode() + bytes.fromhex("00000000e401") ) def matching_sdk(samples, expected): return any( d.get("idVendor") == "2e1a" and d.get("idProduct") == "0002" and "instax4_" + hashlib.sha256(d.get("serial", "").encode()).hexdigest()[:32] == expected for d in samples["devices"] ) def run(target, report): payload = wake_payload(target) assert_no_usb_owner(target["device_id"]) inspector = Inspector(report) advertisement = None manager = None registered = False attempted = False own_path = "/org/missioncore/x4/wake" initial = {} cleanup = [] try: objects = inspector.objects() adapters = [ (p, x) for p, x in objects.items() if x.get(ADAPTER, {}).get("Powered") and ADV_MANAGER in x ] if len(adapters) != 1: raise RuntimeError("Exactly one powered advertising adapter is required") adapter_path, adapter = adapters[0] if ( adapter[ADAPTER].get("Pairable", True) or adapter[ADAPTER].get("Discoverable", True) or adapter[ADV_MANAGER].get("ActiveInstances", -1) != 0 ): raise RuntimeError( "Require non-pairable/non-discoverable adapter with no advertisements" ) if not adapter[ADV_MANAGER].get("SupportedInstances"): raise RuntimeError("No advertising instance is available") initial = {p: plain(x[DEVICE]) for p, x in objects.items() if DEVICE in x} for device in initial.values(): if device.get("Address") == target["bluetooth_address"] and any( device.get(k) for k in ("Connected", "Paired", "Bonded") ): raise RuntimeError("Preserve existing target connection/bond") report["adapter_before"] = plain(adapter) manager = inspector.iface(adapter_path, ADV_MANAGER) import dbus.service class Advertisement(dbus.service.Object): released = False @dbus.service.method(PROPS, in_signature="s", out_signature="a{sv}") def GetAll(self, interface): if interface != ADV: return dbus.Dictionary({}, signature="sv") return dbus.Dictionary( { "Type": dbus.String("peripheral"), "ManufacturerData": dbus.Dictionary( {dbus.UInt16(0x004C): dbus.Array(payload, signature="y")}, signature="qv", ), "Discoverable": dbus.Boolean(True), "DiscoverableTimeout": dbus.UInt16(20), "Timeout": dbus.UInt16(20), # Honored only with BlueZ experimental features. Never enable # that setting here; actual on-air interval is not asserted. "MinInterval": dbus.UInt32(100), "MaxInterval": dbus.UInt32(150), }, signature="sv", ) @dbus.service.method(ADV, in_signature="", out_signature="") def Release(self): self.released = True report["advertisement_released"] = timestamp() advertisement = Advertisement(inspector.bus, own_path) result, errors = [], [] assert_no_usb_owner(target["device_id"]) report["advertisement_intent"] = { "manufacturer_id": 0x004C, "bytes": len(payload), "timeout_seconds": 20, **timestamp(), } attempted = True manager.RegisterAdvertisement( own_path, dbus.Dictionary({}, signature="sv"), reply_handler=lambda: result.append(True), error_handler=lambda e: errors.append(str(e)), timeout=5, ) inspector.pump(6, lambda: bool(result or errors)) if errors or not result: raise RuntimeError(errors[0] if errors else "Advertisement registration timed out") registered = True report["advertisement_registered"] = timestamp() inspector.pump(20, lambda: matching_sdk(usb_snapshot(), target["device_id"])) report["sdk_usb_returned"] = matching_sdk(usb_snapshot(), target["device_id"]) finally: if attempted and advertisement and not advertisement.released: done, errors = [], [] try: manager.UnregisterAdvertisement( own_path, reply_handler=lambda: done.append(True), error_handler=lambda e: errors.append(str(e)), timeout=5, ) inspector.pump(6, lambda: bool(done or errors or advertisement.released)) if (errors and registered) or not (done or errors or advertisement.released): cleanup.append("Advertisement removal did not complete") except Exception as error: cleanup.append(str(error)) # A wake beacon is connectable; disconnect only the prior correlated # X4 address when it acquired a connection during our advertisement. try: current = {p: plain(x[DEVICE]) for p, x in inspector.objects().items() if DEVICE in x} for path, device in current.items(): if device.get("Connected") and not initial.get(path, {}).get("Connected"): if attempted and device.get("Address") == target["bluetooth_address"]: inspector.iface(path, DEVICE).Disconnect(timeout=5) report["wake_connection_closed"] = True else: cleanup.append("New uncorrelated Bluetooth connection preserved") if manager: report["adapter_after"] = plain(inspector.objects().get(adapter_path, {})) except Exception as error: cleanup.append(str(error)) try: if advertisement: advertisement.remove_from_connection() except Exception as error: cleanup.append(str(error)) finally: # Private bus owner disappearance also removes its advertisement. inspector.close() report["cleanup_errors"].extend(cleanup) def main(): def cancel(*_): raise RuntimeError("Wake cancelled") signal.signal(signal.SIGTERM, cancel) report = { "schema": "missioncore.insta360.ble-wake/v1", "started": timestamp(), "vendor_writes": 0, "pair_calls": 0, "notify_subscriptions": 0, "state": "running", } try: raw = sys.stdin.buffer.read(4097) if len(raw) > 4096: raise ValueError("Private wake target exceeds limit") run(json.loads(raw), report) report["state"] = "complete" if report.get("sdk_usb_returned") else "no_sdk_return" except Exception as error: report.update(state="error", error=str(error)) if report.get("cleanup_errors"): report["state"] = "error" report["finished"] = timestamp() print(json.dumps(report), flush=True) return 0 if report["state"] == "complete" else 1 if __name__ == "__main__": raise SystemExit(main())