Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-05 20:32:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Authorize USB/IP imports and retain USB references through hotplug cleanup

Browse Source
This commit is contained in:
Jonatan Castro
2026-11-04 11:05:19 +01:00
parent d3005bbaff
commit 007cf78ec2
11 changed files with 1669 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/scripts/check_usbip_kernel.py
+301
View File
@@ -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()
.github/scripts/check_usbip_services.py
+153
View File
@@ -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()
.github/scripts/check_usbip_transport.py
+18 -10
View File
@@ -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()
.github/scripts/tests/test_usbip.py
+25
View File
@@ -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,6 +194,7 @@ 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, [])
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"]])
.github/scripts/tests/test_usbip_lifetime.py
+112
View File
@@ -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()
.github/scripts/tests/test_usbip_protocol.py
+303
View File
@@ -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()
.github/workflows/test-usbip.yml
+3 -3
View File
@@ -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
docs/usbip.md
+69 -18
View File
@@ -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),
scripts/devices/usbip_lifetime.py
+189
View File
@@ -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
scripts/devices/usbip_manager.py
+82 -15
View File
@@ -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,17 +220,18 @@ 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:
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:
@@ -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.
if leases is None:
disarm_match(device["busid"])
except USBIPError as exc:
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):
scripts/devices/usbip_protocol.py
+410
View File
@@ -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