199 lines
8.1 KiB
Python
199 lines
8.1 KiB
Python
"""Fixed Bluetooth privilege boundary. Never accepts serials, paths or SDK calls."""
|
|
|
|
import json
|
|
import os
|
|
import pwd
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from layout import CODE, trusted, write # noqa: E402
|
|
from runtime.http import Server # noqa: E402
|
|
from runtime.identity import read_binding # noqa: E402
|
|
from runtime.lifecycle import acquisition # noqa: E402
|
|
|
|
ROOT = Path("/var/lib/mission-core-x4-recovery")
|
|
SOCKET = Path("/run/mission-core-x4-recovery/driver.sock")
|
|
IDENTIFIER = re.compile(r"instax4_[0-9a-f]{32}")
|
|
|
|
|
|
def connected():
|
|
values = []
|
|
for path in Path("/sys/bus/usb/devices").iterdir():
|
|
with suppress(OSError, ValueError):
|
|
values.append(read_binding(path.name))
|
|
return values
|
|
|
|
|
|
class WakeService:
|
|
def __init__(self, root=ROOT):
|
|
self.root = root
|
|
self.lock = threading.RLock()
|
|
self.advertiser = threading.Lock()
|
|
self.running = {}
|
|
|
|
def load(self, ident):
|
|
path = self.root / (ident + ".json")
|
|
if not path.exists():
|
|
return {"enabled": False, "attempts": []}
|
|
return json.loads(trusted(path).read_text())
|
|
|
|
def save(self, ident, value):
|
|
write(self.root / (ident + ".json"), json.dumps(value).encode(), 0o600)
|
|
|
|
def registry(self):
|
|
"""Remember exact USB identities without enabling any recovery policy."""
|
|
with self.lock:
|
|
observed = connected()
|
|
known = {
|
|
p.stem for p in self.root.glob("instax4_*.json") if IDENTIFIER.fullmatch(p.stem)
|
|
}
|
|
for binding in observed:
|
|
ident = binding.device_id
|
|
if (
|
|
sum(x.device_id == ident for x in observed) != 1
|
|
or not re.fullmatch(r"[A-Za-z0-9]{7,64}", binding.serial)
|
|
or (ident not in known and len(known) >= 500)
|
|
):
|
|
continue
|
|
value = self.load(ident)
|
|
if "serial" not in value:
|
|
value["serial"] = binding.serial
|
|
self.save(ident, value)
|
|
known.add(ident)
|
|
return {"items": sorted(ident for ident in known if self.load(ident).get("serial"))}
|
|
|
|
def control(self, ident, enabled):
|
|
with self.lock:
|
|
value = self.load(ident)
|
|
if not enabled and "serial" not in value:
|
|
return {"enabled": False}
|
|
if enabled:
|
|
matches = [x for x in connected() if x.device_id == ident]
|
|
if len(matches) != 1 or not re.fullmatch(r"[A-Za-z0-9]{7,64}", matches[0].serial):
|
|
raise ValueError("Connect the exact X4 before enabling recovery")
|
|
if len(list(self.root.glob("instax4_*.json"))) >= 500 and "serial" not in value:
|
|
raise ValueError("Recovery inventory is full")
|
|
value["serial"] = matches[0].serial
|
|
value["enabled"] = enabled
|
|
self.save(ident, value)
|
|
if not enabled and ident in self.running:
|
|
self.running[ident].terminate()
|
|
return {"enabled": enabled}
|
|
|
|
def wake(self, ident, attempt, manual=False):
|
|
if not self.advertiser.acquire(blocking=False):
|
|
return {"state": "busy"}
|
|
try:
|
|
# Installation cannot replace code/state while a wake is in flight.
|
|
with acquisition():
|
|
return self.wake_locked(ident, attempt, manual)
|
|
finally:
|
|
self.advertiser.release()
|
|
|
|
def wake_locked(self, ident, attempt, manual=False):
|
|
with self.lock:
|
|
value = self.load(ident)
|
|
if (not manual and not value["enabled"]) or "serial" not in value:
|
|
return {"state": "disabled"}
|
|
serial = value["serial"]
|
|
if any(x.device_id == ident for x in connected()):
|
|
return {"state": "connected"}
|
|
# A radio suffix is weaker than full USB identity. Refuse known collisions.
|
|
for path in self.root.glob("instax4_*.json"):
|
|
other = json.loads(trusted(path).read_text()).get("serial", "")
|
|
if other and other != serial and other[-6:] == serial[-6:]:
|
|
return {"state": "identity_conflict"}
|
|
now = time.time()
|
|
attempts = [x for x in value["attempts"] if now - x["at"] < 600]
|
|
if any(x["id"] == attempt for x in attempts):
|
|
return {"state": "already_attempted"}
|
|
if len(attempts) >= 3:
|
|
return {"state": "exhausted"}
|
|
value["attempts"] = attempts + [{"id": attempt, "at": now}]
|
|
# Persist before advertising; a crash cannot silently replay the attempt.
|
|
self.save(ident, value)
|
|
target = {
|
|
"device_id": ident,
|
|
"serial": serial,
|
|
"bluetooth_address": None,
|
|
"wakeup_enabled": True,
|
|
}
|
|
with tempfile.TemporaryFile(dir=self.root) as output:
|
|
process = subprocess.Popen(
|
|
["/usr/bin/python3", "-I", str(CODE / "ble_wake.py")],
|
|
stdin=subprocess.PIPE,
|
|
stdout=output,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
self.running[ident] = process
|
|
process.stdin.write(json.dumps(target).encode())
|
|
process.stdin.close()
|
|
# Release state lock during hardware wait so disable can cancel.
|
|
self.lock.release()
|
|
try:
|
|
try:
|
|
process.wait(timeout=45)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
output.seek(0)
|
|
raw = output.read(1024 * 1024 + 1)
|
|
finally:
|
|
self.lock.acquire()
|
|
self.running.pop(ident, None)
|
|
state = "unavailable"
|
|
try:
|
|
if len(raw) <= 1024 * 1024:
|
|
report = json.loads(raw)
|
|
write(self.root / (ident + ".last-wake"), raw, 0o600)
|
|
if process.returncode == 0 and report.get("sdk_usb_returned"):
|
|
state = "connected"
|
|
except (ValueError, OSError):
|
|
pass
|
|
return {"state": state if manual or self.load(ident)["enabled"] else "disabled"}
|
|
|
|
def dispatch(self, method, route, value, _headers):
|
|
if method == "GET" and route == "/health":
|
|
return {"ready": True}
|
|
if method != "POST" or not isinstance(value, dict):
|
|
raise ValueError("Unsupported recovery request")
|
|
if route == "/registry" and not value:
|
|
return self.registry()
|
|
ident = value.get("device_id")
|
|
if not isinstance(ident, str) or not IDENTIFIER.fullmatch(ident):
|
|
raise ValueError("Invalid recovery identity")
|
|
if route == "/control" and set(value) == {"device_id", "enabled"}:
|
|
if type(value["enabled"]) is not bool:
|
|
raise ValueError("Recovery requires an explicit boolean")
|
|
return self.control(ident, value["enabled"])
|
|
if route in ("/wake", "/wake-once") and set(value) == {"device_id", "attempt"}:
|
|
if not isinstance(value["attempt"], str) or not re.fullmatch(
|
|
r"[0-9a-f]{32}", value["attempt"]
|
|
):
|
|
raise ValueError("Invalid wake attempt")
|
|
return self.wake(ident, value["attempt"], manual=route == "/wake-once")
|
|
raise ValueError("Unsupported recovery request")
|
|
|
|
|
|
def main():
|
|
if os.geteuid() or sys.argv[1:]:
|
|
raise RuntimeError("Use the installed fixed wake service")
|
|
trusted(ROOT, True)
|
|
if SOCKET.exists():
|
|
SOCKET.unlink()
|
|
allowed = {0, pwd.getpwnam("mission-core-insta360").pw_uid}
|
|
with Server(SOCKET, WakeService().dispatch, allowed) as server:
|
|
SOCKET.chmod(0o660)
|
|
server.serve_forever(poll_interval=0.5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|