217 lines
8.5 KiB
Python
217 lines
8.5 KiB
Python
"""One bounded X4 wake advertisement; no camera GATT writes or power-off."""
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
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 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():
|
|
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())
|