DCCONSTRUCTIONS
a3c15e11e9
Discover independent camera instances and prepare their versioned runtime from Node or remote Core. Add isolated SDK workers, camera controls, raw dual-fisheye WebRTC preview, and shared action/region loading states. Recover existing Node bindings over known Tailscale addresses after a Core LAN address change. Preserve identities and trust, pin both peers, migrate endpoints with revision checks, and require real heartbeats for online status. Fix the Python client certificate profile for Go X509 verification. Pin Design Guideline 8c53f73 and retain installer/build/acceptance history. Node 0.8.19 is installed; X4 0.1.3-3 is bundled but hardware activation is pending. Validation: qualified DG/Node builds and Go race tests; 31 fleet tests; Python-to-Go certificate interoperability and live tailnet recovery with five fresh heartbeats; prior 38 X4 tests and bounded remote WebRTC acceptance. Clean-OS, replug/power autonomy, local X4 video and long-run stability remain open.
86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
"""Read-only USB admission, before vendor enumeration or library loading."""
|
|
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
def device_id(serial):
|
|
return "instax4_" + hashlib.sha256(serial.encode()).hexdigest()[:32]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Binding:
|
|
device_id: str
|
|
port: str
|
|
bus: int
|
|
address: int
|
|
serial: str
|
|
|
|
@property
|
|
def device_path(self):
|
|
return f"/dev/bus/usb/{self.bus:03d}/{self.address:03d}"
|
|
|
|
|
|
def read_binding(port, root=Path("/sys/bus/usb/devices")):
|
|
if not isinstance(port, str) or not re.fullmatch(
|
|
r"[1-9][0-9]*-[1-9][0-9]*(?:\.[1-9][0-9]*)*", port
|
|
):
|
|
raise ValueError("Invalid camera transport binding")
|
|
path = root / port
|
|
|
|
def read(name):
|
|
return (path / name).read_text().strip()
|
|
|
|
if (read("idVendor"), read("idProduct"), read("product")) != ("2e1a", "0002", "Insta360 X4"):
|
|
raise ValueError("USB device is not the admitted X4 model")
|
|
serial = read("serial")
|
|
if not serial or len(serial) > 256 or any(ord(c) < 32 for c in serial):
|
|
raise ValueError("Camera identity is unavailable")
|
|
bus, address = int(read("busnum")), int(read("devnum"))
|
|
if not 1 <= bus <= 999 or not 1 <= address <= 127:
|
|
raise ValueError("Invalid USB bus address")
|
|
return Binding(device_id(serial), port, bus, address, serial)
|
|
|
|
|
|
def verify_isolation(binding):
|
|
"""No override switch. An ordinary host process cannot load the SDK here.
|
|
|
|
The root-owned instance service must have PrivateDevices+one BindPaths,
|
|
DevicePolicy=closed, NoNewPrivileges, no capabilities, and PrivateNetwork.
|
|
Service declarations additionally enforce cgroup access to this USB node.
|
|
"""
|
|
if os.uname().sysname != "Linux" or os.geteuid() == 0:
|
|
raise RuntimeError("X4 requires an unprivileged isolated Linux worker")
|
|
current = read_binding(binding.port)
|
|
if current != binding:
|
|
raise RuntimeError("Camera changed before SDK initialization")
|
|
status = dict(
|
|
line.split(":", 1)
|
|
for line in Path("/proc/self/status").read_text().splitlines()
|
|
if ":" in line
|
|
)
|
|
if int(status["CapEff"].strip(), 16) or status["NoNewPrivs"].strip() != "1":
|
|
raise RuntimeError("Camera process has unexpected privileges")
|
|
proof = Path("/run/mission-core-x4-control/host-net-inode")
|
|
for path in (proof.parent, proof):
|
|
info = path.lstat()
|
|
if path.is_symlink() or info.st_uid != 0 or info.st_mode & 0o022:
|
|
raise RuntimeError("Untrusted host namespace evidence")
|
|
host_network = int(proof.read_text())
|
|
if host_network <= 0 or Path("/proc/self/ns/net").stat().st_ino == host_network:
|
|
raise RuntimeError("Camera process has access to the host network")
|
|
devices = list(Path("/dev/bus/usb").glob("*/*"))
|
|
if devices != [Path(binding.device_path)]:
|
|
raise RuntimeError("Camera process can see unrelated USB devices")
|
|
info = devices[0].lstat()
|
|
expected_minor = (binding.bus - 1) * 128 + binding.address - 1
|
|
if not stat.S_ISCHR(info.st_mode) or (os.major(info.st_rdev), os.minor(info.st_rdev)) != (
|
|
189,
|
|
expected_minor,
|
|
):
|
|
raise RuntimeError("Camera device node does not match the transport")
|