diff --git a/.github/scripts/check_usbip_kernel.py b/.github/scripts/check_usbip_kernel.py new file mode 100644 index 00000000..a167432e --- /dev/null +++ b/.github/scripts/check_usbip_kernel.py @@ -0,0 +1,301 @@ +"""Disposable QEMU-only real-kernel USB/IP regression lab. Never use on a host.""" + +import contextlib, importlib.util, os, select, socket, subprocess, sys, tempfile, termios, time, tty +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "scripts/devices")) +import usbip_manager as m +from usbip_protocol import Gate + + +def run(*args): + result = subprocess.run(args, text=True, capture_output=True) + assert result.returncode == 0, (args, result.stdout, result.stderr) + return result.stdout + + +def put(p, v): + Path(p).write_text(v) + + +def wait(test): + for _ in range(100): + value = test() + if value: + return value + time.sleep(0.05) + raise AssertionError("condition timed out") + + +def gadget(name, serial, udc): + g = Path("/sys/kernel/config/usb_gadget") / name + g.mkdir() + for f, v in [("idVendor", "0x1d6b"), ("idProduct", "0x0104")]: + put(g / f, v) + (g / "strings/0x409").mkdir() + for f, v in [ + ("serialnumber", serial), + ("manufacturer", "Regression lab"), + ("product", name), + ]: + put(g / "strings/0x409" / f, v) + (g / "configs/c.1").mkdir() + (g / "functions/acm.usb0").mkdir() + (g / "configs/c.1/acm.usb0").symlink_to(g / "functions/acm.usb0") + put(g / "UDC", udc) + return g + + +def find(serial): + return next((d for d in m.inventory() if d["serial"] == serial), None) + + +def chosen(serial, id): + return dict(wait(lambda: find(serial)), id=id) + + +def control(port, busid): + import struct + + with socket.create_connection(("127.0.0.1", port)) as s: + s.sendall( + struct.pack("!HHI", 0x111, 0x8003, 0) + busid.encode().ljust(32, b"\0") + ) + reply = s.recv(8) + assert len(reply) == 8 + return struct.unpack("!HHI", reply)[2] + + +def main(): + assert sys.argv[1:] == [ + "--disposable-vm" + ], "Requires --disposable-vm on a fresh throwaway QEMU VM" + assert ( + os.geteuid() == 0 and "QEMU" in Path("/sys/class/dmi/id/sys_vendor").read_text() + ) + assert not Path("/sys/kernel/config/usb_gadget").exists() or not list( + Path("/sys/kernel/config/usb_gadget").iterdir() + ), "Start with a fresh disposable VM" + + run("modprobe", "dummy_hcd", "num=3") + run("modprobe", "libcomposite") + run("modprobe", "usbip-host") + run("modprobe", "vhci-hcd") + udcs = sorted(p.name for p in Path("/sys/class/udc").iterdir()) + assert len(udcs) == 3 + + a_udc = udcs[0] + a = gadget("selected-a", "original-a", udcs[0]) + b = gadget("selected-b", "original-b", udcs[1]) + time.sleep(0.5) + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + m.CONFIG_DIR = root / "config" + m.SERVER_RUN_DIR = root / "run" + m.RUN_DIR = root / "client" + selected_a = chosen("original-a", "abcdef123456") + selected_b = chosen("original-b", "fedcba654321") + config = {"server": True, "exports": [selected_a, selected_b], "imports": []} + m.atomic_json(m.CONFIG_DIR / "config.json", config) + owned = {} + leases = {} + daemon = subprocess.Popen( + ["usbipd", "-4"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + gate = None + proxy = None + try: + assert not (errors := m.reconcile_exports(config, owned, leases)), errors + + def claim(name): + for selected in config["exports"]: + id = selected["id"] + lease = leases.get(id) + record = owned.get(id) + if lease and lease.ready and record and record["busid"] == name: + return dict(record, id=id) + + gate = Gate(root / "gate.sock", 0, claim) + proxy = subprocess.Popen( + [ + "socat", + "TCP-LISTEN:33240,bind=127.0.0.1,reuseaddr,fork", + "UNIX-CONNECT:" + str(root / "gate.sock"), + ] + ) + time.sleep(0.2) + # A native independent client remains attached throughout A's physical cycle. + print( + run( + "usbip", + "--tcp-port", + "33240", + "attach", + "-r", + "127.0.0.1", + "-b", + selected_b["busid"], + ), + flush=True, + ) + + def serial_transfer(gadget=b, serial="original-b"): + # Gadget B's function tty and remote vhci serial port, not local A's tty. + gadget_tty = ( + "/dev/ttyGS" + + (gadget / "functions/acm.usb0/port_num").read_text().strip() + ) + client = wait( + lambda: next( + ( + p + for p in Path("/sys/class/tty").glob("ttyACM*") + if "vhci_hcd" in str(p.resolve()) + and any( + (q / "serial").exists() + and (q / "serial").read_text().strip() == serial + for q in p.resolve().parents + ) + ), + None, + ) + ) + with contextlib.ExitStack() as stack: + fds = [ + os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) + for path in (gadget_tty, "/dev/" + client.name) + ] + for fd in fds: + stack.callback(os.close, fd) + tty.setraw(fd) + termios.tcflush(fd, termios.TCIOFLUSH) + payload = b"independent-b-still-transferring\n" + os.write(fds[0], payload) + assert select.select([fds[1]], [], [], 3)[0] + assert os.read(fds[1], len(payload)) == payload + + serial_transfer() + for iteration in range(20): + put(a / "UDC", "\n") + time.sleep(0.03) + m.reconcile_exports(config, owned, leases) + assert selected_a["busid"] not in m.MATCH_BUSID.read_text().split() + put(a / "UDC", a_udc) + wait(lambda: find("original-a")) + time.sleep(0.25) + assert not ( + errors := m.reconcile_exports(config, owned, leases) + ), errors + assert leases[selected_a["id"]].ready + serial_transfer() + print( + "PASS: 20 unplug/replug cycles; stale slot removed and original exported; independent native serial transfer survives", + flush=True, + ) + put(a / "UDC", "\n") + time.sleep(0.05) + m.reconcile_exports(config, owned, leases) + a_udc = udcs[2] + put(a / "UDC", a_udc) + wait(lambda: find("original-a")) + time.sleep(0.25) + assert not (errors := m.reconcile_exports(config, owned, leases)), errors + assert owned[selected_a["id"]]["busid"] != selected_a["busid"] + serial_transfer() + print( + "PASS: moving original to another USB port preserves independent serial transfer", + flush=True, + ) + # An unselected replacement returns to a local driver and cannot import. + put(a / "UDC", "\n") + time.sleep(0.05) + put(a / "strings/0x409/serialnumber", "unselected-replacement") + put(a / "UDC", a_udc) + replacement = wait(lambda: find("unselected-replacement")) + errors = m.reconcile_exports(config, owned, leases) + assert selected_a["id"] in errors + assert find("unselected-replacement")["driver"] == "usb" + assert control(33240, replacement["busid"]) != 0 + # Also bind that unselected replacement in a separate native setup: the + # daemon can see it, but a direct import through our gate must still fail. + run("usbip", "bind", "-b", replacement["busid"]) + assert replacement["busid"] in run("usbip", "list", "-r", "127.0.0.1") + assert control(33240, replacement["busid"]) != 0 + m.reconcile_exports(config, owned, leases) + assert find("unselected-replacement")["driver"] == "usbip-host" + run("usbip", "unbind", "-b", replacement["busid"]) + serial_transfer() + print( + "PASS: same-port replacement stays local, native import rejected, independent serial survives", + flush=True, + ) + # Same original then can be selected, imported and exchange native URBs. + put(a / "UDC", "\n") + time.sleep(0.1) + put(a / "strings/0x409/serialnumber", "original-a") + put(a / "UDC", a_udc) + wait(lambda: find("original-a")) + time.sleep(0.25) + assert not (errors := m.reconcile_exports(config, owned, leases)), errors + print( + run( + "usbip", + "--tcp-port", + "33240", + "attach", + "-r", + "127.0.0.1", + "-b", + find("original-a")["busid"], + ), + flush=True, + ) + wait(lambda: len(m.imported_ports()) == 2) + serial_transfer(a, "original-a") + serial_transfer() + for port in m.imported_ports(): + if port["busid"] == find("original-a")["busid"]: + run("usbip", "detach", "-p", str(port["port"])) + # Simulate losing process-owned descriptors before a present-device + # release. The manager must acquire and verify fresh usbfs pins. + leases.pop(selected_a["id"]).close() + config["exports"] = [selected_b] + assert not (errors := m.reconcile_exports(config, owned, leases)), errors + assert find("original-a")["driver"] == "usb" + serial_transfer() + print( + "PASS: original reimport and present-device release restore local driver", + flush=True, + ) + finally: + if proxy: + proxy.terminate() + proxy.wait() + if gate: + gate.terminate() + for port in m.imported_ports(): + with contextlib.suppress(Exception): + run("usbip", "detach", "-p", str(port["port"])) + config["exports"] = [] + print("cleanup", m.reconcile_exports(config, owned, leases), flush=True) + daemon.terminate() + daemon.wait() + for lease in leases.values(): + lease.close() + print("PASS: no Oops/BUG in kernel log", flush=True) + for g in (a, b): + if (g / "UDC").read_text().strip(): + put(g / "UDC", "\n") + (g / "configs/c.1/acm.usb0").unlink() + (g / "functions/acm.usb0").rmdir() + (g / "configs/c.1").rmdir() + (g / "strings/0x409").rmdir() + g.rmdir() + assert not any( + x in run("dmesg") for x in ["BUG:", "Oops:", "general protection fault"] + ) + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/check_usbip_services.py b/.github/scripts/check_usbip_services.py new file mode 100644 index 00000000..35d7e7fc --- /dev/null +++ b/.github/scripts/check_usbip_services.py @@ -0,0 +1,153 @@ +"""Opt-in fresh-QEMU lab: actual generated units, restricted SSH and native USB URBs.""" + +import contextlib, os, select, shutil, sys, termios, time, tty +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "scripts/devices")) +from check_usbip_kernel import m, run, put, gadget, find + + +def wait(test, seconds=30): + end = time.monotonic() + seconds + while time.monotonic() < end: + value = test() + if value: + return value + time.sleep(0.1) + raise AssertionError("timeout") + + +def transfer(g, serial): + client = wait( + lambda: next( + ( + p + for p in Path("/sys/class/tty").glob("ttyACM*") + if "vhci_hcd" in str(p.resolve()) + and any( + (q / "serial").exists() + and (q / "serial").read_text().strip() == serial + for q in p.resolve().parents + ) + ), + None, + ) + ) + with contextlib.ExitStack() as stack: + fds = [ + os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) + for path in ( + "/dev/ttyGS" + (g / "functions/acm.usb0/port_num").read_text().strip(), + "/dev/" + client.name, + ) + ] + for fd in fds: + stack.callback(os.close, fd) + tty.setraw(fd) + termios.tcflush(fd, termios.TCIOFLUSH) + for src, dst in ((fds[0], fds[1]), (fds[1], fds[0])): + payload = b"actual-systemd-ssh-usbip-serial\n" + os.write(src, payload) + assert select.select([dst], [], [], 5)[0] + assert os.read(dst, len(payload)) == payload + + +def main(): + assert sys.argv[1:] == [ + "--disposable-vm" + ], "Requires --disposable-vm on a fresh throwaway QEMU VM" + assert not Path("/etc/proxmenux/usbip").exists() + assert "QEMU" in Path("/sys/class/dmi/id/sys_vendor").read_text() + for name in ("usbip_manager.py", "usbip_protocol.py", "usbip_lifetime.py"): + target = Path("/usr/local/share/proxmenux/scripts/devices") / name + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy(m.SCRIPT.parent / name, target) + run("modprobe", "dummy_hcd", "num=3") + run("modprobe", "libcomposite") + run("modprobe", "usbip-host") + run("modprobe", "vhci-hcd") + udcs = sorted(p.name for p in Path("/sys/class/udc").iterdir()) + assert len(udcs) == 3 + assert not list(Path("/sys/kernel/config/usb_gadget").iterdir()) + a = gadget("service-a", "service-a", udcs[0]) + b = gadget("service-b", "service-b", udcs[1]) + time.sleep(0.5) + key = Path("/root/service-usbip-key") + run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(key)) + m.authorize_key(str(key) + ".pub") + public = Path("/etc/ssh/ssh_host_ed25519_key.pub").read_text().split() + Path("/root/.ssh/known_hosts").write_text( + "127.0.0.1 " + public[0] + " " + public[1] + "\n" + ) + imports = [] + try: + m.update_server(True) + sa = m.add_export(wait(lambda: find("service-a"))["busid"]) + sb = m.add_export(wait(lambda: find("service-b"))["busid"]) + wait(lambda: len([e for e in m.catalog() if e["status"] == "exported"]) == 2) + remote = m.validate_remote("127.0.0.1", 22, str(key)) + for selected, port in ((sa, 33240), (sb, 33241)): + imports.append(m.add_import(remote, selected["id"], port)) + wait(lambda: len(m.imported_ports()) == 2) + transfer(a, "service-a") + transfer(b, "service-b") + print( + "PASS: actual generated server/client systemd units and restricted SSH carry bidirectional native serial on two USBs", + flush=True, + ) + for cycle in range(3): + put(a / "UDC", "\n") + time.sleep(0.2) + transfer(b, "service-b") + put(a / "UDC", udcs[0]) + wait(lambda: find("service-a")) + time.sleep(0.5) + # An old vhci port may still exist; wait for the new guest driver and verify data. + wait( + lambda: next( + ( + e + for e in m.catalog() + if e["id"] == sa["id"] and e["status"] == "exported" + ), + None, + ) + ) + time.sleep(6) + transfer(a, "service-a") + transfer(b, "service-b") + print( + "PASS: three physical reconnects automatically reattach through restricted SSH; second serial connection survives", + flush=True, + ) + m.remove_export(sa["id"]) + assert find("service-a")["driver"] == "usb" + transfer(b, "service-b") + print( + "PASS: stop-sharing waits for real release and returns local driver while independent client keeps transferring", + flush=True, + ) + for item in imports: + m.remove_import(item["id"]) + imports = [] + m.update_server(False) + assert not m.read_json(m.state_path(), {}) + assert find("service-b")["driver"] == "usb" + print( + "PASS: disable stops listeners, removes owned bindings, leaves both local USB drivers restored", + flush=True, + ) + finally: + for item in imports: + with contextlib.suppress(Exception): + m.remove_import(item["id"]) + with contextlib.suppress(Exception): + m.update_server(False) + print( + run("journalctl", "-u", m.SERVER_UNIT, "--no-pager", "-n", "20"), flush=True + ) + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/check_usbip_transport.py b/.github/scripts/check_usbip_transport.py index 91099d32..fce6506f 100644 --- a/.github/scripts/check_usbip_transport.py +++ b/.github/scripts/check_usbip_transport.py @@ -41,16 +41,18 @@ def wait_until(test, processes, log, timeout=15): def echo(sock): - payload = bytes(range(256)) * 1024 + # Exercise the production protocol gate through the real SSH stream. Empty + # fixture selection must produce an empty native DEVLIST, never raw echo. + import struct sock.settimeout(10) - sock.sendall(payload) + sock.sendall(struct.pack("!HHI", 0x0111, 0x8005, 0)) received = bytearray() - while len(received) < len(payload): - chunk = sock.recv(65536) + while len(received) < 12: + chunk = sock.recv(12 - len(received)) if not chunk: break received.extend(chunk) - assert bytes(received) == payload, "SSH/Unix/TCP forwarding corrupted the stream" + assert bytes(received) == struct.pack("!HHII", 0x0111, 5, 0, 0), received def main(): @@ -83,6 +85,8 @@ def main(): fixture_source = root / "usbip_manager.py" fixture_source.write_bytes(SOURCE.read_bytes()) fixture_source.chmod(0o644) + for name in ("usbip_protocol.py", "usbip_lifetime.py"): + (root / name).write_bytes((SOURCE.parent / name).read_bytes()) subprocess.run(["usermod", "--home", str(root), username], check=True) log = root / "transport.log" with log.open("w") as log_handle: @@ -96,11 +100,15 @@ def main(): import socketserver class Echo(socketserver.BaseRequestHandler): def handle(self): - while True: - data = self.request.recv(65536) - if not data: + import struct + data = b'' + while len(data) < 8: + block = self.request.recv(8 - len(data)) + if not block: return - self.request.sendall(data) + data += block + if data == struct.pack('!HHI', 0x0111, 0x8005, 0): + self.request.sendall(struct.pack('!HHII', 0x0111, 5, 0, 0)) socketserver.ThreadingTCPServer.allow_reuse_address = True socketserver.ThreadingTCPServer(('0.0.0.0', 3240), Echo).serve_forever() """) @@ -236,7 +244,7 @@ LogLevel VERBOSE assert f"127.0.0.1:{local_port}" in listener and f"0.0.0.0:{local_port}" not in listener, listener print("PASS: private namespace, dedicated-group Unix socket, no host USB/IP listener") print("PASS: unknown/wrong keys, shell commands and arbitrary forwarding rejected; dedicated key accepted") - print("PASS: 256 KiB binary stream across real SSH -> Unix socket -> socat -> private TCP") + print("PASS: native USB/IP DEVLIST across real SSH -> Unix protocol gate -> private TCP") finally: if tunnel: tunnel.terminate() diff --git a/.github/scripts/tests/test_usbip.py b/.github/scripts/tests/test_usbip.py index 69bd1fc5..226621bd 100644 --- a/.github/scripts/tests/test_usbip.py +++ b/.github/scripts/tests/test_usbip.py @@ -137,6 +137,30 @@ class USBIPTests(unittest.TestCase): self.assertEqual(self.calls, [["usbip", "bind", "--busid", "1-2"]]) self.assertIsNone(self.m.catalog()[0]["busid"]) + def test_retained_slot_is_rejected_before_detaching_reconnected_local_usb(self): + device = self.device() + selected = self.selected(device) + config = self.config([selected]) + self.m.MATCH_BUSID.slots[device["busid"]] = "REMOV" + with patch.object(self.m, "run", side_effect=self.usb): + errors = self.m.reconcile_exports(config, {EXPORT: selected}, {}) + self.assertIn("local driver", errors[EXPORT]) + self.assertEqual(self.calls, []) + self.assertEqual(self.m.inventory()[0]["driver"], "usb") + + def test_imported_usb_does_not_shadow_physical_original(self): + device = self.device() + imported = self.m.SYSFS / "2-1" + imported.symlink_to(self.m.SYSFS / "1-2", target_is_directory=True) + target = self.root / "vhci_hcd.0" / "2-1" + target.parent.mkdir() + shutil.copytree(self.m.SYSFS / "1-2", target, symlinks=True) + imported.unlink() + imported.symlink_to(target, target_is_directory=True) + devices = self.m.inventory() + self.assertEqual(len(devices), 1) + self.assertEqual(self.m.resolve_device(device, devices)["busid"], "1-2") + def test_no_serial_requires_same_busid_and_ambiguous_serial_is_rejected(self): selected = self.selected(self.device(serial="")) shutil.rmtree(self.m.SYSFS / "1-2") @@ -170,7 +194,8 @@ class USBIPTests(unittest.TestCase): with patch.object(self.m, "run", side_effect=self.usb): self.m.reconcile_exports(self.m.load_config(), owned) self.assertEqual(self.calls, []) - self.m.remove_export(EXPORT) + with patch.object(self.m.time, "sleep", side_effect=lambda _: self.m.reconcile_exports(self.m.load_config(), owned)): + self.m.remove_export(EXPORT) self.m.reconcile_exports(self.m.load_config(), owned) self.assertEqual(self.calls, [["usbip", "unbind", "--busid", "1-2"]]) diff --git a/.github/scripts/tests/test_usbip_lifetime.py b/.github/scripts/tests/test_usbip_lifetime.py new file mode 100644 index 00000000..774c2954 --- /dev/null +++ b/.github/scripts/tests/test_usbip_lifetime.py @@ -0,0 +1,112 @@ +"""Absent-slot recovery must require a verified pin and completed kernel removal.""" + +import importlib.util +import errno +from pathlib import Path +import socket +import tempfile +import unittest +from unittest.mock import Mock, patch + +SOURCE = Path(__file__).resolve().parents[3] / "scripts/devices/usbip_lifetime.py" +spec = importlib.util.spec_from_file_location("usbip_lifetime", SOURCE) +l = importlib.util.module_from_spec(spec) +spec.loader.exec_module(l) + + +class LifetimeTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + lease = l.Lease.__new__(l.Lease) + lease.path = self.root / "1-2" + lease.devpath = "/devices/test/usb1/1-2" + lease.bus = 1 + lease.number = 3 + lease.sequence = 100 + lease.ready = True + lease.removed = False + lease.fds = [] + lease.events = Mock() + lease.events.recvmsg.side_effect = BlockingIOError() + self.lease = lease + self.match = Mock() + self.match.parent = self.root + self.match.read_text.side_effect = lambda: ( + "" if (self.root / "rebind").exists() else "1-2" + ) + + def event(self, **changes): + values = { + "ACTION": "remove", + "DEVTYPE": "usb_device", + "DEVPATH": self.lease.devpath, + "BUSNUM": "001", + "DEVNUM": "003", + "SEQNUM": "101", + } | changes + return b"\0".join(f"{k}={v}".encode() for k, v in values.items()) + b"\0" + + def test_no_rebind_without_a_verified_lifetime_pin(self): + self.lease.ready = False + self.lease.removed = True + with self.assertRaises(l.LifetimeError): + self.lease.release(self.match, Mock()) + self.assertFalse((self.root / "rebind").exists()) + + def test_rebind_requires_completed_removal_not_only_missing_sysfs(self): + with patch.object( + self.lease, "validate", side_effect=FileNotFoundError() + ), patch.object(l.time, "monotonic", side_effect=[0, 3, 4]): + with self.assertRaisesRegex(l.LifetimeError, "Waiting"): + self.lease.release(self.match, Mock()) + self.assertFalse((self.root / "rebind").exists()) + + def test_unrelated_or_userspace_events_do_not_authorize_slot_recovery(self): + packets = [ + (self.event(SEQNUM="99"), [], 0, (0, 1)), + (self.event(DEVNUM="004"), [], 0, (0, 1)), + (self.event(DEVPATH="/devices/other"), [], 0, (0, 1)), + (self.event(), [], 0, (12345, 1)), + (self.event(DEVTYPE="usb_interface"), [], 0, (0, 1)), + BlockingIOError(), + ] + self.lease.events.recvmsg.side_effect = packets + self.assertFalse(self.lease.poll_removed()) + + def test_lost_or_truncated_events_invalidate_recovery(self): + for fault in ( + OSError(errno.ENOBUFS, "overflow"), + (self.event(), [], socket.MSG_TRUNC, (0, 1)), + ): + self.lease.ready = True + self.lease.events.recvmsg.side_effect = [fault] + with self.assertRaises(l.LifetimeError): + self.lease.poll_removed() + self.assertFalse(self.lease.ready) + self.assertFalse((self.root / "rebind").exists()) + + def test_pinned_completed_removal_cleans_only_its_slot(self): + self.lease.events.recvmsg.side_effect = [ + (self.event(), [], 0, (0, 1)), + BlockingIOError(), + ] + unbind = Mock() + self.lease.release(self.match, unbind) + unbind.assert_not_called() + self.assertEqual((self.root / "rebind").read_text(), "1-2") + self.assertFalse(self.lease.ready) + + def test_present_device_uses_native_unbind_not_absent_recovery(self): + self.match.read_text.return_value = "" + self.match.read_text.side_effect = None + unbind = Mock() + with patch.object(self.lease, "validate"): + self.lease.release(self.match, unbind) + unbind.assert_called_once_with("1-2") + self.assertFalse((self.root / "rebind").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/tests/test_usbip_protocol.py b/.github/scripts/tests/test_usbip_protocol.py new file mode 100644 index 00000000..daa6f67a --- /dev/null +++ b/.github/scripts/tests/test_usbip_protocol.py @@ -0,0 +1,303 @@ +"""Real socket control/URB tests with disposable sysfs-shaped device objects.""" + +import importlib.util +import os +from pathlib import Path +import socket +import socketserver +import struct +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[3] +spec = importlib.util.spec_from_file_location( + "usbip_protocol", ROOT / "scripts/devices/usbip_protocol.py" +) +p = importlib.util.module_from_spec(spec) +spec.loader.exec_module(p) + + +def descriptor( + name="2-1", bus=2, number=3, vendor=0x1D6B, product=0x0104, interfaces=1 +): + return p.DEVICE.pack( + b"/sys/devices/disposable", + name.encode(), + bus, + number, + 3, + vendor, + product, + 0x0100, + 0, + 0, + 0, + 1, + 1, + interfaces, + ) + + +def exact(connection, count): + result = bytearray() + while len(result) < count: + block = connection.recv(count - len(result)) + if not block: + break + result.extend(block) + return bytes(result) + + +class GateTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory(prefix="usbip-gate-") + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.sysfs = self.root / "sysfs" + self.sysfs.mkdir() + self.device("original") + self.selected = { + "2-1": { + "id": "abcdef123456", + "vendor": "1d6b", + "product": "0104", + "serial": "selected-serial", + } + } + self.calls = [] + self.reply = descriptor() + self.listed = [ + (descriptor(), b"\x02\x02\x01\0"), + (descriptor("3-1", bus=3, number=4), b"\x02\x02\x01\0"), + ] + self.hook = lambda: None + self.header_override = None + fixture = self + + class Handler(socketserver.BaseRequestHandler): + def handle(self): + try: + header = exact(self.request, 8) + if len(header) != 8: + return + _, code, _ = p.HEADER.unpack(header) + request = header + ( + exact(self.request, 32) if code == p.IMPORT else b"" + ) + fixture.calls.append(request) + fixture.hook() + self.request.sendall( + fixture.header_override + or p.HEADER.pack(p.VERSION, code & 0x7FFF, 0) + ) + if code == p.DEVLIST: + self.request.sendall(struct.pack("!I", len(fixture.listed))) + for raw, interfaces in fixture.listed: + self.request.sendall(raw + interfaces) + return + self.request.sendall(fixture.reply) + while True: + block = self.request.recv(65536) + if not block: + self.request.shutdown(socket.SHUT_WR) + return + self.request.sendall(block) + except OSError: + return + + class Backend(socketserver.ThreadingMixIn, socketserver.TCPServer): + daemon_threads = True + + self.backend = Backend(("127.0.0.1", 0), Handler) + self.backend_thread = threading.Thread( + target=self.backend.serve_forever, + kwargs={"poll_interval": 0.05}, + daemon=True, + ) + self.backend_thread.start() + self.addCleanup(self.backend.server_close) + self.addCleanup(self.backend.shutdown) + # Production calls this as root; these protocol tests do not alter users. + with patch.object(p.os, "chown"): + self.gate = p.Gate( + self.root / "gate.sock", + os.getgid(), + lambda name: self.selected.get(name), + self.sysfs, + self.backend.server_address, + max_connections=2, + ) + self.addCleanup(self.gate.terminate) + + def device(self, directory, serial="selected-serial", number=3): + device = self.root / directory + device.mkdir() + for name, value in { + "idVendor": "1d6b", + "idProduct": "0104", + "serial": serial, + "busnum": "2", + "devnum": str(number), + "bDeviceClass": "00", + }.items(): + (device / name).write_text(value) + interface = device / "2-1:1.0" + interface.mkdir() + (interface / "bInterfaceClass").write_text("02") + (device / "driver").symlink_to("/sys/bus/usb/drivers/usbip-host") + link = self.sysfs / "2-1" + link.unlink(missing_ok=True) + link.symlink_to(device, target_is_directory=True) + return device + + def connect(self): + connection = socket.socket(socket.AF_UNIX) + connection.settimeout(3) + connection.connect(str(self.gate.path)) + self.addCleanup(connection.close) + return connection + + def request(self, name=b"2-1\0", code=p.IMPORT): + return p.HEADER.pack(p.VERSION, code, 0) + ( + name.ljust(32, b"\0") if code == p.IMPORT else b"" + ) + + def reject(self, request): + connection = self.connect() + connection.sendall(request) + result = exact(connection, 8) + self.assertTrue(not result or p.HEADER.unpack(result)[2] != 0, result) + self.assertEqual(connection.recv(1), b"") + return result + + def test_fragmented_selected_import_and_large_binary_full_duplex(self): + connection = self.connect() + for value in self.request(): + connection.sendall(bytes([value])) + self.assertEqual( + exact(connection, 320), p.HEADER.pack(p.VERSION, 3, 0) + self.reply + ) + payload = bytes(range(256)) * 4096 + writer = threading.Thread(target=connection.sendall, args=(payload,)) + writer.start() + self.assertEqual(exact(connection, len(payload)), payload) + writer.join(3) + self.assertFalse(writer.is_alive()) + connection.shutdown(socket.SHUT_WR) + self.assertEqual(connection.recv(1), b"") + + def test_foreign_device_is_rejected_before_native_backend_import(self): + self.reject(self.request(b"3-1\0")) + self.assertEqual(self.calls, []) + + def test_malformed_busids_never_reach_backend(self): + for value in (b"2-1\0x", b"../../etc\0", b"2-1\xff\0", b"x" * 32): + with self.subTest(value=value): + self.reject(self.request(value)) + self.assertEqual(self.calls, []) + + def test_unsupported_control_headers_never_reach_backend(self): + for header in ( + p.HEADER.pack(0x0110, p.IMPORT, 0), + p.HEADER.pack(p.VERSION, 0x8001, 0), + p.HEADER.pack(p.VERSION, p.IMPORT, 1), + ): + self.reject(header) + self.assertEqual(self.calls, []) + + def test_devlist_filters_unselected_export_and_preserves_interface_records(self): + connection = self.connect() + connection.sendall(self.request(code=p.DEVLIST)) + reply = exact(connection, 12) + self.assertEqual(reply, p.HEADER.pack(p.VERSION, 5, 0) + struct.pack("!I", 1)) + self.assertEqual(exact(connection, 316), self.reply + b"\x02\x02\x01\0") + self.assertEqual(connection.recv(1), b"") + + def test_changed_selection_during_import_receives_no_success_payload(self): + self.hook = self.selected.clear + self.reject(self.request()) + self.assertEqual(len(self.calls), 1) + + def test_mutated_callback_dictionary_cannot_rewrite_saved_selection(self): + self.hook = lambda: self.selected["2-1"].update(id="fedcba654321") + self.reject(self.request()) + + def test_replaced_sysfs_object_cannot_use_identical_reply_metadata(self): + self.hook = lambda: self.device("replacement") + self.reject(self.request()) + + def test_driver_change_during_import_is_rejected(self): + def replace(): + driver = self.root / "original/driver" + driver.unlink() + driver.symlink_to("/sys/bus/usb/drivers/usb") + + self.hook = replace + self.reject(self.request()) + + def test_descriptor_identity_and_devnum_must_match_pinned_object(self): + for reply in ( + descriptor(number=4), + descriptor(vendor=0x05AC), + descriptor("3-1"), + descriptor(bus=3), + ): + self.reply = reply + self.reject(self.request()) + + def test_composite_storage_or_hub_device_is_rejected_before_backend(self): + for name in ("bDeviceClass", "2-1:1.0/bInterfaceClass"): + path = self.root / "original" / name + old = path.read_text() + for value in ("08", "09"): + path.write_text(value) + self.reject(self.request()) + path.write_text(old) + self.assertEqual(self.calls, []) + + def test_unicode_serial_is_valid_without_changing_busid_encoding(self): + serial = "selected-ñ-设备" + (self.root / "original/serial").write_text(serial) + self.selected["2-1"]["serial"] = serial + connection = self.connect() + connection.sendall(self.request()) + self.assertEqual( + exact(connection, 320), p.HEADER.pack(p.VERSION, 3, 0) + self.reply + ) + + def test_backend_response_version_opcode_and_status_are_validated(self): + for header in ( + p.HEADER.pack(0x0110, 3, 0), + p.HEADER.pack(p.VERSION, 5, 0), + p.HEADER.pack(p.VERSION, 3, 0xFFFFFFFF), + ): + self.header_override = header + self.reject(self.request()) + + def test_incomplete_header_times_out_but_accepted_idle_stream_does_not(self): + with patch.object(p, "CONTROL_SECONDS", 0.1): + partial = self.connect() + partial.sendall(b"\x01") + self.assertEqual(partial.recv(1), b"") + connection = self.connect() + connection.sendall(self.request()) + self.assertEqual(len(exact(connection, 320)), 320) + time.sleep(0.2) + connection.sendall(b"idle-still-works") + self.assertEqual(exact(connection, 16), b"idle-still-works") + + def test_stop_closes_idle_backend_and_client_and_removes_listener(self): + connection = self.connect() + connection.sendall(self.request()) + self.assertEqual(len(exact(connection, 320)), 320) + self.gate.terminate() + self.assertEqual(connection.recv(1), b"") + self.assertFalse(self.gate.path.exists()) + self.assertIsNotNone(self.gate.poll()) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/workflows/test-usbip.yml b/.github/workflows/test-usbip.yml index a8af9c25..db17b601 100644 --- a/.github/workflows/test-usbip.yml +++ b/.github/workflows/test-usbip.yml @@ -7,7 +7,7 @@ on: - 'scripts/menus/hw_grafics_menu.sh' - 'scripts/menus/main_menu.sh' - 'scripts/menus/search_index.json' - - '.github/scripts/tests/test_usbip.py' + - '.github/scripts/tests/test_usbip*.py' - '.github/scripts/check_usbip_transport.py' - '.github/workflows/test-usbip.yml' - 'docs/usbip.md' @@ -15,7 +15,7 @@ on: branches: [main, develop] paths: - 'scripts/devices/**' - - '.github/scripts/tests/test_usbip.py' + - '.github/scripts/tests/test_usbip*.py' - '.github/scripts/check_usbip_transport.py' - '.github/workflows/test-usbip.yml' workflow_dispatch: @@ -41,7 +41,7 @@ jobs: - name: Device ownership, reconnect and systemd validation env: PYTHONDONTWRITEBYTECODE: '1' - run: python3 -B -m unittest discover -s .github/scripts/tests -p test_usbip.py -v + run: python3 -B -m unittest discover -s .github/scripts/tests -p 'test_usbip*.py' -v - name: Install fixture-only transport dependencies run: | sudo apt-get update -o APT::Update::Error-Mode=any diff --git a/docs/usbip.md b/docs/usbip.md index 32255d80..193f809d 100644 --- a/docs/usbip.md +++ b/docs/usbip.md @@ -22,22 +22,33 @@ requested in [#350](https://github.com/MacRimi/ProxMenux/issues/350). journalctl -u proxmenux-usbip-server.service ``` -The server reloads saved selections every three seconds. A live exported device's -match is immediately marked for removal so the kernel rejects a later automatic -probe on that port. Physical unplug/replug remains an **unresolved limitation**: -upstream Linux can retain the match-table slot even after a successful `del` -write. The manager verifies disappearance before forgetting ownership or binding -a new port, and otherwise fails closed with a visible cleanup error. It never -calls `rebind` on an absent device or reloads a module affecting foreign exports. -Automatic reconnect after physical USB removal is therefore not yet guaranteed; -this PR needs maintainer review and a safe kernel-supported cleanup policy. +The server reloads saved selections every three seconds. It keeps open usbfs +references to each original USB device and its USB ancestors before binding. +After binding it marks the live match REMOV and verifies the original is still +bound. Replacement devices cannot use that match for automatic USB/IP probing. +On physical removal, the worker waits for the original device's completed kernel +remove event, uses the still-pinned original reference to clear that one retained +slot, and verifies disappearance before binding again. Other exports are left +running. Reconnection at the same port, movement to another port and normal +present-device release do not require reloading a shared module. + +If the worker is killed, a bind cannot be verified, removal events are lost, or +an older installation left a retained slot, that slot has no proven lifetime +reference. The worker reports an error and refuses to detach a new local device +or attempt absent-slot recovery. Stop sharing while the original remains present; +if that cannot clear the slot, a host reboot is the supported recovery. This +exceptional recovery interrupts the host, so schedule it yourself; the manager +never reboots it or reloads USB modules automatically. Graceful service stop +releases managed bindings before dropping the references. + Without a serial number, the vendor/product and original bus ID must match; the manager will not guess which identical dongle to bind. Duplicate serial numbers are rejected as ambiguous. USB hubs and mass-storage devices, including composite devices with a storage interface, are excluded. -**Stop sharing** removes one saved selection and releases its binding on the -next reconciliation. **Disable server** stops the service and releases its +**Stop sharing** removes one saved selection and waits for confirmed release. +A cleanup error or timeout is reported instead of claiming that eject succeeded. +**Disable server** stops the service and releases its managed exports, retaining selections for the next enable. Devices already bound by another USB/IP setup are not adopted or unbound by this manager. Native USB/IP unbind reprobes the device so its normal local driver can return. @@ -46,14 +57,21 @@ Native USB/IP unbind reprobes the device so its normal local driver can return. USB/IP has no built-in authentication or encryption. The generated server unit uses systemd `PrivateNetwork=yes`: `usbipd --ipv4 --tcp-port 3240` and its -`socat` proxy run in an isolated network namespace. There is **no USB/IP TCP +Python protocol gate run in an isolated network namespace. There is **no USB/IP TCP listener in the host network namespace**, and this feature does not open a firewall port. Running the manager's `serve` command outside a private network namespace is rejected before spawning a listener. -The proxy exposes `/run/proxmenux-usbip-server/usbip.sock` to the dedicated -`proxmenux-usbip` group (0660), inside a root-owned 0750 directory. Root alone -can read or change the managed device/configuration state. A separate group-readable +The gate exposes `/run/proxmenux-usbip-server/usbip.sock` to the dedicated +`proxmenux-usbip` group (0660), inside a root-owned 0750 directory. For each native +DEVLIST and IMPORT, the gate checks the saved selection, +verified ownership, current USB identity, driver and sysfs incarnation. It checks +again before releasing a successful import reply; an unselected export bound by +another setup cannot bypass selection by requesting its bus ID directly. Control +requests have bounded time, memory and connection counts. Accepted URB streams +retain full-duplex transfer, half-close and backpressure without an idle timeout. +Root alone can read or change the managed device/configuration state. +A separate group-readable catalog contains only the selected USB device metadata. Before enabling the server, choose **Authorize a client SSH public key** and @@ -124,7 +142,8 @@ python3 "$MANAGER" status The client has an enabled systemd service per device. It reconnects the SSH tunnel and reattaches the device after a network interruption or restart. On each new attach it resolves the **saved export ID** to the server's current bus -ID; physical port movement is subject to the kernel cleanup limitation above. The client keeps an ownership +ID; physical port movement is resolved only after confirmed cleanup of the old +slot. The client keeps an ownership record of the exact imported kernel port, local bus ID and remote endpoint; cleanup detaches only a matching owned import. Removing a connection disables its service, releases its device and removes its saved configuration. @@ -154,8 +173,40 @@ cannot directly connect to this server's isolated daemon. Validation includes fake sysfs/USB commands for identity and ownership, service rollback and quoting, malformed catalogs, credential failures and reconnects. The Linux CI also verifies generated units and exercises actual namespace -isolation and restricted SSH sessions with an echo fixture. Physical device -and guest-driver compatibility still requires a real deployment test. +isolation and restricted SSH sessions with native USB/IP control messages. +Protocol socket tests cover binary transfers, malformed requests, selection +changes, foreign exports and replaced device objects. Lifetime tests require a +verified pin and a matching completed kernel removal event before slot recovery. + +The kernel regression lab is deliberately opt-in and requires a fresh, disposable +QEMU Linux VM with usbip, socat and the dummy_hcd/libcomposite/usbip-host/vhci-hcd +modules. Never run it on a production host. In that VM, from the checkout: + +```sh +sudo unshare --net sh -c 'ip link set lo up; python3 -B .github/scripts/check_usbip_kernel.py --disposable-vm' +``` + +It creates two virtual ACM gadgets, performs 20 physical unplug/replug cycles, +rejects an unselected same-port replacement, reimports the original and confirms +local-driver restoration. An independent native USB/IP serial transfer must keep +working during the cycles. The lab passed on Debian 13 / Linux 6.12.115; actual +hardware and guest-driver compatibility still depends on the deployment. + +The companion service lab must run in a separate fresh disposable QEMU VM: + +```sh +sudo python3 -B .github/scripts/check_usbip_services.py --disposable-vm +``` + +It installs the actual generated server/client units and a dedicated SSH key +inside that VM, transfers native serial data in both directions through restricted +SSH, checks three automatic reconnects with an independent connection alive, and +verifies synchronous stop-sharing and disable cleanup. Destroy the VM afterwards. + +The lifetime contract is based on [usbfs open/release references](https://github.com/torvalds/linux/blob/v6.12/drivers/usb/core/devio.c), +[USB/IP match/rebind](https://github.com/torvalds/linux/blob/v6.12/drivers/usb/usbip/stub_main.c) +and [completed stub disconnect](https://github.com/torvalds/linux/blob/v6.12/drivers/usb/usbip/stub_dev.c). +A sysfs pathname alone is never sufficient authority for absent-slot rebind. Protocol and transport references: [Linux USB/IP tools](https://github.com/torvalds/linux/tree/master/tools/usb/usbip), diff --git a/scripts/devices/usbip_lifetime.py b/scripts/devices/usbip_lifetime.py new file mode 100644 index 00000000..cfbfe3e2 --- /dev/null +++ b/scripts/devices/usbip_lifetime.py @@ -0,0 +1,189 @@ +"""Keep kernel USB references until a managed, disarmed match is released. + +usbfs descriptors hold usb_device references (including its USB parents). +A kernel remove uevent confirms disconnect finished before absent-slot rebind. +A pathname or a sysfs directory descriptor alone is not a USB lifetime pin. +Lost events, unverified binds and ownership recovered without pins fail closed. +""" + +import os +from pathlib import Path +import socket +import stat +import time + + +class LifetimeError(RuntimeError): + pass + + +class Lease: + def __init__(self, sysfs, busid, expected): + self.path = Path(sysfs) / busid + self.fds = [] + self.ready = False + self.removed = False + self.events = socket.socket( + socket.AF_NETLINK, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC, 15 + ) + try: + self.events.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) + self.events.bind((0, 1)) + self.events.setblocking(False) + self.directory = os.open( + self.path, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC + ) + self.fds.append(self.directory) + self.object = self._object(os.fstat(self.directory)) + self.devpath = str(self.path.resolve()).removeprefix("/sys") + self.bus = int(self.read("busnum")) + self.number = int(self.read("devnum")) + self.identity = { + key: self.read(name) + for key, name in [ + ("vendor", "idVendor"), + ("product", "idProduct"), + ("serial", "serial"), + ] + } + if self.identity != {key: expected[key] for key in self.identity}: + raise LifetimeError("USB changed before its lifetime could be pinned") + # Keep every USB ancestor alive too: do_rebind locks the parent. + current = self.path.resolve() + while (current / "busnum").exists(): + bus = int((current / "busnum").read_text()) + number = int((current / "devnum").read_text()) + fd = os.open( + f"/dev/bus/usb/{bus:03}/{number:03}", + os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, + ) + self.fds.append(fd) + info = os.fstat(fd) + if not stat.S_ISCHR(info.st_mode) or os.major(info.st_rdev) != 189: + raise LifetimeError("USB lifetime reference is not a usbfs device") + if os.minor(info.st_rdev) != (bus - 1) * 128 + number - 1: + raise LifetimeError("USB lifetime reference changed") + current = current.parent + self.validate() + # Ignore older queued events, including a previous incarnation with + # a reused port and USB device number. + self.sequence = int(Path("/sys/kernel/uevent_seqnum").read_text()) + except BaseException: + self.close() + raise + + @staticmethod + def _object(info): + return info.st_dev, info.st_ino + + def read(self, name): + try: + fd = os.open( + name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, dir_fd=self.directory + ) + except FileNotFoundError: + if name == "serial": + return "" + raise + try: + return os.read(fd, 1024).decode().strip() + finally: + os.close(fd) + + def validate(self): + if ( + self._object(os.stat(self.path)) != self.object + or int(self.read("busnum")) != self.bus + or int(self.read("devnum")) != self.number + or any( + self.read(name) != self.identity[key] + for key, name in [ + ("vendor", "idVendor"), + ("product", "idProduct"), + ("serial", "serial"), + ] + ) + ): + raise LifetimeError("Pinned USB was removed or replaced") + + def confirm(self, disarm): + # REMOV prevents a replacement from overwriting the table's udev pointer. + # Verify the pinned original is still bound AFTER that transition. + disarm(self.path.name) + self.validate() + if Path(os.readlink("driver", dir_fd=self.directory)).name != "usbip-host": + raise LifetimeError("Pinned USB was not bound by USB/IP") + self.read("usbip_status") + self.ready = True + + def poll_removed(self): + while True: + try: + packet, _, flags, sender = self.events.recvmsg(65536) + except BlockingIOError: + return self.removed + except OSError as exc: + self.ready = False + raise LifetimeError( + "USB removal events were lost; restart the host before reusing this slot" + ) from exc + if flags & socket.MSG_TRUNC: + self.ready = False + raise LifetimeError("Truncated USB removal event") + if sender[0] != 0: + continue + fields = dict( + value.split(b"=", 1) for value in packet.split(b"\0") if b"=" in value + ) + if ( + fields.get(b"ACTION") == b"remove" + and fields.get(b"DEVTYPE") == b"usb_device" + and fields.get(b"DEVPATH") == self.devpath.encode() + and fields.get(b"BUSNUM") == f"{self.bus:03}".encode() + and fields.get(b"DEVNUM") == f"{self.number:03}".encode() + and int(fields.get(b"SEQNUM", b"0")) > self.sequence + ): + self.removed = True + + def release(self, match, unbind): + if not self.ready: + raise LifetimeError( + "USB bind was not verified; restart the host before reusing this slot" + ) + self.poll_removed() + if not self.removed: + try: + self.validate() + except (OSError, LifetimeError): + # Enumeration may already have removed sysfs while disconnect is + # still running. Wait for its completed USB-device remove event. + deadline = time.monotonic() + 2 + while not self.poll_removed() and time.monotonic() < deadline: + time.sleep(0.02) + if not self.removed: + raise LifetimeError( + "Waiting for completed USB removal; no slot recovery was attempted" + ) + else: + unbind(self.path.name) + if self.path.name not in match.read_text().split(): + self.close() + return + # Physical disconnect can race the native unbind. Only a remove + # event allows the pinned absent path below. + if not self.poll_removed(): + raise LifetimeError("USB/IP unbind did not clear its match") + # Unlike a stale pathname, these still-open usbfs files pin the exact + # udev and parent referenced by this verified REMOV slot. No newer probe + # can overwrite it. Removal notification is emitted after stub_disconnect. + (match.parent / "rebind").write_text(self.path.name) + if self.path.name in match.read_text().split(): + raise LifetimeError("Pinned USB/IP match cleanup was not confirmed") + self.close() + + def close(self): + for fd in self.fds: + os.close(fd) + self.fds.clear() + self.events.close() + self.ready = False diff --git a/scripts/devices/usbip_manager.py b/scripts/devices/usbip_manager.py index 3e36dcb1..87c56109 100644 --- a/scripts/devices/usbip_manager.py +++ b/scripts/devices/usbip_manager.py @@ -123,6 +123,10 @@ def inventory(): for path in sorted(SYSFS.glob("*")): if not BUSID.fullmatch(path.name): continue + # Imported devices cannot be re-exported by usbip-host. Exclude them + # before identity matching so a local client cannot shadow its source. + if any(parent.name.startswith("vhci_hcd") for parent in path.resolve().parents): + continue vendor, product = attribute(path, "idVendor"), attribute(path, "idProduct") if not re.fullmatch(r"[a-fA-F0-9]{4}", vendor) or not re.fullmatch(r"[a-fA-F0-9]{4}", product): continue @@ -172,11 +176,33 @@ def disarm_match(busid): raise USBIPError("Could not disarm the USB/IP match") from exc -def release_export(identifier, owned, devices): +def release_export(identifier, owned, devices, leases=None): previous = owned[identifier] current = next((d for d in devices if d["busid"] == previous["busid"]), None) if not BUSID.fullmatch(previous.get("busid", "")): raise USBIPError("Invalid owned USB bus ID") + if leases is not None: + # A restart may have lost the old pins. Recover only a currently present, + # identity-matching, bound object; never pin a replacement by pathname. + if (current and same_device(previous, current) and current["driver"] == "usbip-host" + and (identifier not in leases or not leases[identifier].ready)): + from usbip_lifetime import Lease + recovered = Lease(SYSFS, current["busid"], current) + try: + recovered.confirm(disarm_match) + except BaseException: + recovered.close() + raise + old = leases.pop(identifier, None) + if old: + old.close() + leases[identifier] = recovered + if identifier in leases: + leases[identifier].release(MATCH_BUSID, lambda busid: run(["usbip", "unbind", "--busid", busid])) + del leases[identifier] + del owned[identifier] + atomic_json(state_path(), owned) + return if current and same_device(previous, current) and current["driver"] == "usbip-host": run(["usbip", "unbind", "--busid", current["busid"]]) else: @@ -194,18 +220,19 @@ def release_export(identifier, owned, devices): atomic_json(state_path(), owned) -def reconcile_exports(config, owned): +def reconcile_exports(config, owned, leases=None): devices = inventory() desired = {item["id"]: item for item in config["exports"]} if config["server"] else {} errors = {} for identifier in list(owned): previous = owned[identifier] current = next((d for d in devices if d["busid"] == previous["busid"]), None) - if identifier not in desired or current is None or not same_device(previous, current): - try: - release_export(identifier, owned, devices) - except USBIPError as exc: - errors[identifier] = str(exc) + try: + removed = leases is not None and identifier in leases and leases[identifier].poll_removed() + if identifier not in desired or current is None or not same_device(previous, current) or removed: + release_export(identifier, owned, devices, leases) + except (USBIPError, OSError, RuntimeError) as exc: + errors[identifier] = str(exc) for identifier, selected in desired.items(): if identifier in errors: continue @@ -213,21 +240,33 @@ def reconcile_exports(config, owned): device = resolve_device(selected, devices) previous = owned.get(identifier) if previous and (previous["busid"] != device["busid"] or not same_device(previous, device)): - release_export(identifier, owned, devices) + release_export(identifier, owned, devices, leases) previous = None if device["driver"] == "usbip-host": if not previous: raise USBIPError("Device is already exported outside ProxMenux; it was not adopted") + if leases is not None and (identifier not in leases or not leases[identifier].ready): + raise USBIPError("Recovered export has no lifetime pin; stop sharing while the original USB remains connected, or restart the host") disarm_match(device["busid"]) continue + if device["busid"] in MATCH_BUSID.read_text().split(): + raise USBIPError("Kernel retained this USB/IP slot; cleanup must finish before binding so the local driver is preserved") + if leases is not None: + from usbip_lifetime import Lease + if identifier in leases: + leases.pop(identifier).close() + leases[identifier] = Lease(SYSFS, device["busid"], device) # Record intent before binding, so a crash cannot orphan our export. owned[identifier] = {k: device[k] for k in ("busid", "vendor", "product", "serial")} atomic_json(state_path(), owned) run(["usbip", "bind", "--busid", device["busid"]]) # ALLOC -> REMOV rejects future probes without interrupting this # live export. Do not leave a reusable automatic port match armed. - disarm_match(device["busid"]) - except USBIPError as exc: + if leases is None: + disarm_match(device["busid"]) + else: + leases[identifier].confirm(disarm_match) + except (USBIPError, OSError, ValueError, RuntimeError) as exc: errors[identifier] = str(exc) atomic_json(SERVER_RUN_DIR / "status.json", {"errors": errors, "updated": time.time()}) return errors @@ -235,9 +274,13 @@ def reconcile_exports(config, owned): def catalog(): config, devices, owned = load_config(), inventory(), read_json(state_path(), {}) + errors = read_json(SERVER_RUN_DIR / "status.json", {}).get("errors", {}) exports = [] for selected in config["exports"]: row = dict(selected, status="waiting", busid=None) + if selected["id"] in errors: + exports.append(dict(row, error=errors[selected["id"]])) + continue try: current = resolve_device(selected, devices) if config["server"] and selected["id"] in owned and current["driver"] == "usbip-host" and ( @@ -601,11 +644,25 @@ def serve(): socket_path = SERVER_RUN_DIR / "usbip.sock" socket_path.unlink(missing_ok=True) owned = read_json(state_path(), {}) + leases = {} + from usbip_protocol import Gate + def claim(busid): + # Re-read selection for every import, including deselection between polls. + config = load_config() + if not config["server"]: + return None + for selected in config["exports"]: + identifier = selected["id"] + lease = leases.get(identifier) + previous = owned.get(identifier) + if (lease and lease.ready and previous and previous["busid"] == busid + and same_device(selected, previous)): + return dict(previous, id=identifier) + return None daemon = proxy = None try: daemon = subprocess.Popen(["usbipd", "--ipv4", "--tcp-port", "3240"]) - proxy = subprocess.Popen(["socat", f"UNIX-LISTEN:{socket_path},mode=0660,group={account.pw_gid},fork", - "TCP:127.0.0.1:3240"]) + proxy = Gate(socket_path, account.pw_gid, claim, SYSFS) with stop_event() as stopped: previous_errors = None while not stopped.is_set(): @@ -614,7 +671,7 @@ def serve(): config = load_config() if not config["server"]: break - errors = reconcile_exports(config, owned) + errors = reconcile_exports(config, owned, leases) publish_catalog() if errors != previous_errors: for identifier, error in errors.items(): @@ -622,13 +679,15 @@ def serve(): previous_errors = errors stopped.wait(3) finally: + stop_child(proxy) devices = inventory() for identifier in list(owned): try: - release_export(identifier, owned, devices) - except USBIPError as exc: + release_export(identifier, owned, devices, leases) + except (USBIPError, OSError, RuntimeError) as exc: print(f"Could not release export {identifier}: {exc}", file=sys.stderr, flush=True) - stop_child(proxy) + for lease in leases.values(): + lease.close() stop_child(daemon) socket_path.unlink(missing_ok=True) (SERVER_RUN_DIR / "catalog.json").unlink(missing_ok=True) @@ -689,7 +748,7 @@ def update_server(enabled): remove_unit(SERVER_UNIT) # Retry cleanup even if the worker had previously crashed. owned = read_json(state_path(), {}) - errors = reconcile_exports(config, owned) + errors = reconcile_exports(config, owned, {}) if errors: raise USBIPError("Server stopped, but some device bindings could not be released; check status and retry disable") except USBIPError: @@ -731,7 +790,15 @@ def remove_export(identifier): raise USBIPError("Export not found") config["exports"] = [e for e in config["exports"] if e["id"] != identifier] atomic_json(CONFIG_DIR / "config.json", config) - # The server loop releases only its own binding within three seconds. + requested = time.time() + deadline = time.monotonic() + 10 + while identifier in read_json(state_path(), {}): + status = read_json(SERVER_RUN_DIR / "status.json", {}) + if status.get("updated", 0) >= requested and identifier in status.get("errors", {}): + raise USBIPError("Selection removed, but release is incomplete: " + status["errors"][identifier]) + if time.monotonic() >= deadline: + raise USBIPError("Selection removed; USB release has not been confirmed. Check server status") + time.sleep(.1) def add_import(remote, export_id, local_port): diff --git a/scripts/devices/usbip_protocol.py b/scripts/devices/usbip_protocol.py new file mode 100644 index 00000000..1ad40605 --- /dev/null +++ b/scripts/devices/usbip_protocol.py @@ -0,0 +1,410 @@ +"""Authorize USB/IP control requests before exposing the native daemon. + +The Unix listener runs in the server's private network namespace. Root supplies +current selection/ownership; the forced-command SSH account only reaches this +listener. An accepted import becomes an opaque, bounded, full-duplex URB stream. +""" + +from contextlib import contextmanager +import os +from pathlib import Path +import re +import selectors +import socket +import socketserver +import struct +import threading +import time + +VERSION = 0x0111 +IMPORT = 0x8003 +DEVLIST = 0x8005 +HEADER = struct.Struct("!HHI") +DEVICE = struct.Struct("!256s32sIIIHHHBBBBBB") +BUSID = re.compile(r"[0-9]+-[0-9]+(?:\.[0-9]+)*\Z") +CONTROL_SECONDS = 5 +MAX_DEVICES = 256 +MAX_BUFFER = 65536 + + +class ProtocolError(Exception): + pass + + +def receive(connection, count, deadline): + result = bytearray() + while len(result) < count: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ProtocolError("USB/IP control request timed out") + connection.settimeout(remaining) + block = connection.recv(count - len(result)) + if not block: + raise ProtocolError("Incomplete USB/IP control request") + result.extend(block) + return bytes(result) + + +def busid(value): + """One canonical NUL-terminated 32-byte Linux bus identifier.""" + if len(value) != 32 or b"\0" not in value: + raise ProtocolError("Invalid USB/IP bus identifier") + name, padding = value.split(b"\0", 1) + if any(padding): + raise ProtocolError("Invalid USB/IP bus identifier padding") + try: + name = name.decode("ascii") + except UnicodeDecodeError as exc: + raise ProtocolError("Non-ASCII USB/IP bus identifier") from exc + if not BUSID.fullmatch(name): + raise ProtocolError("Invalid USB/IP bus identifier") + return name + + +class PinnedDevice: + """Pin the sysfs object used for identity checks across the backend import. + + This does not claim to hold usb_device's kernel lifetime. It detects a + removed/replaced sysfs object; USBFS lifetime pins belong to the owner. + """ + + def __init__(self, root, name, expected): + self.path = Path(root) / name + self.fd = os.open(self.path, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC) + try: + self.object = self._object(os.fstat(self.fd)) + self.expected = { + key: str(expected[key]) for key in ("vendor", "product", "serial") + } + self.identity = self._identity() + self.validate() + except BaseException: + os.close(self.fd) + raise + + @staticmethod + def _object(info): + return info.st_dev, info.st_ino + + def read(self, name): + descriptor = os.open( + name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, dir_fd=self.fd + ) + try: + value = os.read(descriptor, 513) + if len(value) > 512: + raise ProtocolError("Oversized USB identity attribute") + return value.decode("utf-8").strip() + finally: + os.close(descriptor) + + def _identity(self): + try: + serial = self.read("serial") + except FileNotFoundError: + serial = "" + classes = [int(self.read("bDeviceClass"), 16)] + for name in os.listdir(self.fd): + if name.startswith(self.path.name + ":"): + classes.append(int(self.read(name + "/bInterfaceClass"), 16)) + if any(value in (8, 9) for value in classes): + raise ProtocolError("USB hubs and mass storage cannot be shared") + return { + "vendor": self.read("idVendor").lower(), + "product": self.read("idProduct").lower(), + "serial": serial, + "busnum": int(self.read("busnum")), + "devnum": int(self.read("devnum")), + "classes": tuple(sorted(classes)), + } + + def validate(self, descriptor=None): + if ( + self.object != self._object(os.stat(self.path)) + or self._identity() != self.identity + ): + raise ProtocolError("USB device was removed or replaced") + if any(self.identity[key] != value for key, value in self.expected.items()): + raise ProtocolError("USB device identity is not selected") + if Path(os.readlink("driver", dir_fd=self.fd)).name != "usbip-host": + raise ProtocolError("Selected USB device is no longer exported") + if descriptor is not None: + values = DEVICE.unpack(descriptor) + if ( + busid(values[1]) != self.path.name + or values[2] != self.identity["busnum"] + or values[3] != self.identity["devnum"] + or values[5] != int(self.identity["vendor"], 16) + or values[6] != int(self.identity["product"], 16) + ): + raise ProtocolError( + "USB/IP reply does not describe the selected device" + ) + + def close(self): + os.close(self.fd) + + +def relay(client, backend, stopped): + """Preserve half-close and backpressure without an idle-session timeout.""" + peers = (client, backend) + queued = [bytearray(), bytearray()] + reading = [True, True] + write_closed = [False, False] + for connection in peers: + connection.setblocking(False) + with selectors.DefaultSelector() as selector: + while not stopped.is_set(): + for index, connection in enumerate(peers): + other = 1 - index + if not reading[other] and not queued[index] and not write_closed[index]: + connection.shutdown(socket.SHUT_WR) + write_closed[index] = True + events = ( + selectors.EVENT_READ + if reading[index] and len(queued[other]) < MAX_BUFFER + else 0 + ) | (selectors.EVENT_WRITE if queued[index] else 0) + try: + selector.get_key(connection) + except KeyError: + if events: + selector.register(connection, events, index) + else: + if events: + selector.modify(connection, events, index) + else: + selector.unregister(connection) + if not selector.get_map(): + return + for key, events in selector.select(timeout=1): + index = key.data + other = 1 - index + if events & selectors.EVENT_READ: + try: + block = peers[index].recv(MAX_BUFFER - len(queued[other])) + except BlockingIOError: + continue + if block: + queued[other].extend(block) + else: + reading[index] = False + if events & selectors.EVENT_WRITE: + try: + sent = peers[index].send(queued[index]) + except BlockingIOError: + continue + if sent == 0: + raise ProtocolError("USB/IP connection closed during transfer") + del queued[index][:sent] + + +class Gate: + """A bounded Unix server with the process-like lifecycle used by serve().""" + + def __init__( + self, + path, + group, + claim, + sysfs="/sys/bus/usb/devices", + backend=("127.0.0.1", 3240), + max_connections=64, + ): + self.path = Path(path) + self.claim = claim + self.sysfs = Path(sysfs) + self.backend = backend + self.stopped = threading.Event() + self.lock = threading.Lock() + self.connections = set() + self.slots = threading.BoundedSemaphore(max_connections) + gate = self + + class Handler(socketserver.BaseRequestHandler): + def handle(self): + gate.handle(self.request) + + class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): + daemon_threads = True + + def process_request(self, request, address): + if not gate.slots.acquire(blocking=False): + self.shutdown_request(request) + return + try: + super().process_request(request, address) + except BaseException: + gate.slots.release() + raise + + def process_request_thread(self, request, address): + try: + super().process_request_thread(request, address) + finally: + gate.slots.release() + + self.server = Server(str(self.path), Handler) + try: + os.chown(self.path, 0, group) + self.path.chmod(0o660) + self.thread = threading.Thread( + target=self.server.serve_forever, + kwargs={"poll_interval": 0.1}, + daemon=True, + ) + self.thread.start() + except BaseException: + self.server.server_close() + self.path.unlink(missing_ok=True) + raise + + @contextmanager + def pin(self, name): + expected = self.selection(name) + if not expected: + raise ProtocolError("USB device is not selected for sharing") + device = PinnedDevice(self.sysfs, name, expected) + try: + yield device, expected + finally: + device.close() + + def _still_selected(self, name, expected): + if self.selection(name) != expected: + raise ProtocolError("USB sharing selection changed") + + def selection(self, name): + value = self.claim(name) + if not value: + return None + # Freeze the authorization snapshot even when a callback reuses its + # dictionary; mutable ownership changes must not rewrite this lease. + return {key: str(value[key]) for key in ("id", "vendor", "product", "serial")} + + @contextmanager + def tracked(self, connection, close=True): + with self.lock: + if self.stopped.is_set(): + connection.close() + raise ProtocolError("USB/IP server is stopping") + self.connections.add(connection) + try: + yield connection + finally: + with self.lock: + self.connections.discard(connection) + if close: + connection.close() + + def _backend(self, request, deadline): + connection = socket.create_connection( + self.backend, timeout=max(0.001, deadline - time.monotonic()) + ) + try: + connection.sendall(request) + return connection + except BaseException: + connection.close() + raise + + @staticmethod + def response(connection, code, deadline): + header = receive(connection, HEADER.size, deadline) + version, received_code, status = HEADER.unpack(header) + if version != VERSION or received_code != (code & 0x7FFF) or status > 5: + raise ProtocolError("Invalid USB/IP backend response") + return header, status + + def handle(self, client): + code = None + try: + with self.tracked(client, close=False): + deadline = time.monotonic() + CONTROL_SECONDS + request = receive(client, HEADER.size, deadline) + version, code, status = HEADER.unpack(request) + if version != VERSION or code not in (DEVLIST, IMPORT) or status: + code = None + raise ProtocolError("Unsupported USB/IP control request") + if code == IMPORT: + raw_name = receive(client, 32, deadline) + name = busid(raw_name) + with self.pin(name) as (device, expected): + with self.tracked( + self._backend(request + raw_name, deadline) + ) as backend: + header, failed = self.response(backend, code, deadline) + if failed: + code = None + client.sendall(header) + return + descriptor = receive(backend, DEVICE.size, deadline) + self._still_selected(name, expected) + device.validate(descriptor) + code = ( + None # Data mode: never append another control header. + ) + client.sendall(header + descriptor) + relay(client, backend, self.stopped) + else: + with self.tracked(self._backend(request, deadline)) as backend: + header, failed = self.response(backend, code, deadline) + if failed: + code = None + client.sendall(header) + return + count = struct.unpack("!I", receive(backend, 4, deadline))[0] + if count > MAX_DEVICES: + raise ProtocolError( + "USB/IP backend device count is too large" + ) + allowed = [] + for _ in range(count): + descriptor = receive(backend, DEVICE.size, deadline) + values = DEVICE.unpack(descriptor) + interfaces = receive(backend, 4 * values[-1], deadline) + name = busid(values[1]) + try: + with self.pin(name) as (device, expected): + self._still_selected(name, expected) + device.validate(descriptor) + allowed.append(descriptor + interfaces) + except ( + ProtocolError, + OSError, + ValueError, + KeyError, + UnicodeError, + ): + continue + code = None + client.sendall( + header + struct.pack("!I", len(allowed)) + b"".join(allowed) + ) + except (ProtocolError, OSError, ValueError, KeyError, UnicodeError): + if code in (DEVLIST, IMPORT): + try: + client.sendall(HEADER.pack(VERSION, code & 0x7FFF, 1)) + except OSError: + pass + finally: + client.close() + + def poll(self): + return None if not self.stopped.is_set() and self.thread.is_alive() else 1 + + def terminate(self): + self.stopped.set() + self.server.shutdown() + with self.lock: + for connection in self.connections: + try: + connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self.server.server_close() + self.path.unlink(missing_ok=True) + + def wait(self, timeout=5): + self.thread.join(timeout) + return 0