Add selected USB/IP device sharing over SSH

.github/scripts/check_usbip_transport.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""Linux-only integration: real namespaces, socat and SSH; no USB hardware.
+
+Run as root in the disposable CI runner. All keys, configuration, listeners and
+child processes belong to a temporary fixture; no systemd units are installed.
+"""
+import importlib.util
+import os
+from pathlib import Path
+import signal
+import socket
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+SOURCE = Path(__file__).resolve().parents[2] / "scripts/devices/usbip_manager.py"
+
+
+def port():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def wait_until(test, processes, log, timeout=15):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        for process in processes:
+            if process.poll() is not None:
+                raise AssertionError(f"Fixture exited ({process.returncode}): {log.read_text()}")
+        try:
+            if test():
+                return
+        except (ConnectionError, FileNotFoundError, socket.timeout):
+            pass
+        time.sleep(0.1)
+    raise AssertionError(f"Fixture did not become ready: {log.read_text()}")
+
+
+def echo(sock):
+    payload = bytes(range(256)) * 1024
+    sock.settimeout(10)
+    sock.sendall(payload)
+    received = bytearray()
+    while len(received) < len(payload):
+        chunk = sock.recv(65536)
+        if not chunk:
+            break
+        received.extend(chunk)
+    assert bytes(received) == payload, "SSH/Unix/TCP forwarding corrupted the stream"
+
+
+def main():
+    assert sys.platform == "linux" and os.geteuid() == 0, "Use the disposable Linux CI runner as root"
+    spec = importlib.util.spec_from_file_location("usbip_manager", SOURCE)
+    manager = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(manager)
+    # A mistaken direct launch must never create a listener in this namespace.
+    try:
+        manager.require_private_network()
+    except manager.USBIPError:
+        pass
+    else:
+        raise AssertionError("Expected the host namespace to be rejected")
+
+    processes = []
+    with tempfile.TemporaryDirectory(prefix="pmx-usbip-") as directory:
+        root = Path(directory)
+        log = root / "transport.log"
+        with log.open("w") as log_handle:
+            try:
+                fake_bin = root / "bin"
+                fake_bin.mkdir()
+                # The real daemon requires usbip-host and physical devices.
+                # Echo stands in only for its TCP endpoint, inside the namespace.
+                fake_daemon = fake_bin / "usbipd"
+                fake_daemon.write_text("""#!/usr/bin/python3
+import socketserver
+class Echo(socketserver.BaseRequestHandler):
+    def handle(self):
+        while True:
+            data = self.request.recv(65536)
+            if not data:
+                return
+            self.request.sendall(data)
+socketserver.ThreadingTCPServer.allow_reuse_address = True
+socketserver.ThreadingTCPServer(('0.0.0.0', 3240), Echo).serve_forever()
+""")
+                fake_daemon.chmod(0o755)
+                config = root / "config"
+                manager.atomic_json(config / "config.json", {"server": True, "exports": [], "imports": []})
+                server_run = root / "server"
+                runner = root / "server.py"
+                runner.write_text(f"""import importlib.util
+from pathlib import Path
+import subprocess
+spec = importlib.util.spec_from_file_location('usbip_manager', {str(SOURCE)!r})
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+m.CONFIG_DIR = Path({str(config)!r})
+m.SERVER_RUN_DIR = Path({str(server_run)!r})
+m.SYSFS = Path({str(root / 'no-usb-devices')!r})
+subprocess.run(['ip', 'link', 'set', 'lo', 'up'], check=True)
+m.serve()
+""")
+                server = subprocess.Popen(["unshare", "--net", sys.executable, str(runner)],
+                    env={**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"},
+                    stdout=log_handle, stderr=log_handle, start_new_session=True)
+                processes.append(server)
+                unix_path = server_run / "usbip.sock"
+                wait_until(lambda: unix_path.exists(), [server], log)
+                assert os.readlink(f"/proc/{server.pid}/ns/net") != os.readlink("/proc/self/ns/net")
+                assert stat.S_IMODE(server_run.stat().st_mode) == 0o700
+                assert stat.S_IMODE(unix_path.stat().st_mode) == 0o600
+                # Linux /proc/net reflects the caller's network namespace.
+                listening = subprocess.check_output(["ss", "-H", "-ltn", "sport", "=", ":3240"], text=True)
+                assert not listening.strip(), "Raw USB/IP TCP leaked into the host network namespace"
+                # Wait for both the proxy and its private TCP peer.
+                def unix_ready():
+                    with socket.socket(socket.AF_UNIX) as connection:
+                        connection.connect(str(unix_path))
+                        echo(connection)
+                    return True
+                wait_until(unix_ready, [server], log)
+
+                for name in ("host", "client", "wrong-client"):
+                    subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(root / name)], check=True)
+                authorized = root / "authorized_keys"
+                authorized.write_text((root / "client.pub").read_text())
+                authorized.chmod(0o600)
+                known = root / "known_hosts"
+                known.write_text("")
+                known.chmod(0o600)
+                ssh_port, local_port = port(), port()
+                ssh_config = root / "sshd_config"
+                ssh_config.write_text(f"""Port {ssh_port}
+ListenAddress 127.0.0.1
+HostKey {root / 'host'}
+PidFile {root / 'sshd.pid'}
+AuthorizedKeysFile {authorized}
+PermitRootLogin prohibit-password
+PasswordAuthentication no
+KbdInteractiveAuthentication no
+UsePAM yes
+StrictModes yes
+AllowUsers root
+AllowTcpForwarding local
+AllowStreamLocalForwarding local
+LogLevel ERROR
+""")
+                Path("/run/sshd").mkdir(exist_ok=True, mode=0o755)
+                sshd = subprocess.Popen(["/usr/sbin/sshd", "-D", "-e", "-f", str(ssh_config)],
+                                        stdout=log_handle, stderr=log_handle, start_new_session=True)
+                processes.append(sshd)
+                def ssh_ready():
+                    with socket.create_connection(("127.0.0.1", ssh_port), timeout=1):
+                        return True
+                wait_until(ssh_ready, [server, sshd], log)
+                remote = manager.validate_remote("127.0.0.1", ssh_port, str(root / "client"))
+                extra = ["-o", f"UserKnownHostsFile={known}", "-o", "GlobalKnownHostsFile=/dev/null"]
+                def check_ssh(selected):
+                    args = manager.ssh_args(selected)
+                    args[1:1] = extra
+                    return subprocess.run(args + ["--", selected["host"], "true"], capture_output=True, text=True, timeout=15)
+                unknown = check_ssh(remote)
+                assert unknown.returncode != 0 and "Host key verification failed" in unknown.stderr, unknown.stderr
+                host_public = (root / "host.pub").read_text().split()
+                known.write_text(f"[127.0.0.1]:{ssh_port} {host_public[0]} {host_public[1]}\n")
+                wrong = check_ssh(remote | {"key": str(root / "wrong-client")})
+                assert wrong.returncode != 0 and "Permission denied" in wrong.stderr, wrong.stderr
+                accepted = check_ssh(remote)
+                assert accepted.returncode == 0, accepted.stderr
+                manager.REMOTE_SOCKET = str(unix_path)
+                args = manager.tunnel_args(remote | {"local_port": local_port})
+                args[1:1] = extra
+                tunnel = subprocess.Popen(args, stdout=log_handle, stderr=log_handle, start_new_session=True)
+                processes.append(tunnel)
+                def tunnel_ready():
+                    with socket.create_connection(("127.0.0.1", local_port), timeout=1) as connection:
+                        echo(connection)
+                    return True
+                wait_until(tunnel_ready, processes, log)
+                listener = subprocess.check_output(["ss", "-H", "-ltn", "sport", "=", f":{local_port}"], text=True)
+                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, root-only Unix socket, no host USB/IP listener")
+                print("PASS: unknown host key and wrong client key rejected; valid key accepted")
+                print("PASS: 256 KiB binary stream across real SSH -> Unix socket -> socat -> private TCP")
+            finally:
+                for process in reversed(processes):
+                    if process.poll() is None:
+                        os.killpg(process.pid, signal.SIGTERM)
+                for process in reversed(processes):
+                    try:
+                        process.wait(timeout=10)
+                    except subprocess.TimeoutExpired:
+                        os.killpg(process.pid, signal.SIGKILL)
+                        process.wait(timeout=5)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/scripts/tests/test_usbip.py

@@ -0,0 +1,340 @@
+"""USB/IP ownership, device identity, SSH policy and service regression tests.
+
+No physical USB devices, live systemd units or host configuration are changed.
+The optional Linux integration lane exercises actual SSH, socat and namespaces.
+"""
+import copy
+import importlib.util
+import json
+import os
+from pathlib import Path
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[3]
+SOURCE = ROOT / "scripts/devices/usbip_manager.py"
+EXPORT = "abcdef123456"
+CLIENT = "fedcba654321"
+
+
+def load():
+    spec = importlib.util.spec_from_file_location("usbip_manager", SOURCE)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class USBIPTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.m = load()
+        for attr, path in (("CONFIG_DIR", "config"), ("RUN_DIR", "client-run"),
+                           ("SERVER_RUN_DIR", "server-run"), ("UNIT_DIR", "units"), ("SYSFS", "sysfs")):
+            setattr(self.m, attr, self.root / path)
+            getattr(self.m, attr).mkdir()
+        self.calls = []
+        self.key = self.root / "id_ed25519"
+        self.key.write_text("test-key")
+        self.key.chmod(0o600)
+        self.remote = {"host": "usb.example.test", "ssh_port": 2222, "key": str(self.key)}
+
+    def config(self, exports=None, imports=None, server=True):
+        value = {"server": server, "exports": exports or [], "imports": imports or []}
+        self.m.atomic_json(self.m.CONFIG_DIR / "config.json", value)
+        return value
+
+    def device(self, busid="1-2", serial="zigbee-123", device_class="00", interface="03", driver="usb"):
+        path = self.m.SYSFS / busid
+        path.mkdir()
+        values = {"idVendor": "10c4", "idProduct": "ea60", "serial": serial,
+                  "product": "Test Zigbee adapter", "bDeviceClass": device_class}
+        for key, value in values.items():
+            (path / key).write_text(value)
+        child = path / f"{busid}:1.0"
+        child.mkdir()
+        (child / "bInterfaceClass").write_text(interface)
+        self.set_driver(busid, driver)
+        return next(d for d in self.m.inventory() if d["busid"] == busid)
+
+    def set_driver(self, busid, driver):
+        target = self.root / "drivers" / driver
+        target.mkdir(parents=True, exist_ok=True)
+        path = self.m.SYSFS / busid / "driver"
+        path.unlink(missing_ok=True)
+        path.symlink_to(target, target_is_directory=True)
+
+    def selected(self, device):
+        return {k: device[k] for k in ("busid", "vendor", "product", "serial", "name")} | {"id": EXPORT}
+
+    def usb(self, args, timeout=20):
+        self.calls.append(args)
+        if args[:2] == ["usbip", "bind"]:
+            self.set_driver(args[-1], "usbip-host")
+        elif args[:2] == ["usbip", "unbind"]:
+            self.set_driver(args[-1], "usb")
+        return ""
+
+    def test_storage_composite_interfaces_and_hubs_are_never_exported(self):
+        for index, (base, interface) in enumerate((("08", "06"), ("09", "00"), ("00", "08"), ("", "03"))):
+            device = self.device(busid=f"1-{index + 1}", device_class=base, interface=interface)
+            self.assertTrue(device["blocked"])
+            self.config()
+            with self.assertRaises(self.m.USBIPError), patch.object(self.m, "run") as command:
+                self.m.add_export(device["busid"])
+            command.assert_not_called()
+        self.assertEqual(self.m.load_config()["exports"], [])
+
+    def test_export_intent_is_private_and_bind_is_idempotent(self):
+        device = self.device()
+        self.config()
+        selected = self.m.add_export(device["busid"])
+        owned = {}
+        with patch.object(self.m, "run", side_effect=self.usb):
+            self.assertEqual(self.m.reconcile_exports(self.m.load_config(), owned), {})
+            self.assertEqual(self.m.reconcile_exports(self.m.load_config(), owned), {})
+        self.assertEqual(self.calls, [["usbip", "bind", "--busid", "1-2"]])
+        self.assertEqual(self.m.catalog()[0]["id"], selected["id"])
+        self.assertEqual(self.m.catalog()[0]["status"], "exported")
+        for path in (self.m.CONFIG_DIR / "config.json", self.m.state_path()):
+            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+        self.assertNotEqual(self.m.RUN_DIR, self.m.SERVER_RUN_DIR)
+
+    def test_serial_moves_to_another_port_without_unbinding_a_replacement(self):
+        selected = self.selected(self.device())
+        config = self.config([selected])
+        owned = {}
+        with patch.object(self.m, "run", side_effect=self.usb):
+            self.m.reconcile_exports(config, owned)
+            shutil.rmtree(self.m.SYSFS / "1-2")
+            self.device(serial="other-device", driver="usbip-host")
+            self.device(busid="2-4")
+            self.m.reconcile_exports(config, owned)
+        self.assertEqual(owned[EXPORT]["busid"], "2-4")
+        self.assertEqual(self.calls, [["usbip", "bind", "--busid", "1-2"], ["usbip", "bind", "--busid", "2-4"]])
+        self.assertEqual(self.m.catalog()[0]["busid"], "2-4")
+
+    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")
+        self.device(busid="2-4", serial="")
+        with self.assertRaises(self.m.USBIPError):
+            self.m.resolve_device(selected, self.m.inventory())
+        self.device(busid="1-3", serial="duplicate")
+        first = self.device(busid="1-4", serial="duplicate")
+        with self.assertRaises(self.m.USBIPError):
+            self.m.resolve_device(first, self.m.inventory())
+
+    def test_foreign_exports_are_not_adopted_and_disable_only_releases_owned_devices(self):
+        own = self.selected(self.device())
+        foreign = self.device(busid="1-3", serial="foreign", driver="usbip-host")
+        config = self.config([own, self.selected(foreign) | {"id": CLIENT}])
+        owned = {}
+        with patch.object(self.m, "run", side_effect=self.usb):
+            errors = self.m.reconcile_exports(config, owned)
+            self.assertIn("outside ProxMenux", errors[CLIENT])
+            config["server"] = False
+            self.m.reconcile_exports(config, owned)
+        self.assertEqual(self.calls, [["usbip", "bind", "--busid", "1-2"], ["usbip", "unbind", "--busid", "1-2"]])
+        self.assertEqual(owned, {})
+        self.assertEqual(next(d for d in self.m.inventory() if d["busid"] == "1-3")["driver"], "usbip-host")
+
+    def test_recover_bind_intent_and_remove_selected_export(self):
+        selected = self.selected(self.device(driver="usbip-host"))
+        self.config([selected])
+        owned = {EXPORT: selected}
+        self.m.atomic_json(self.m.state_path(), owned)
+        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)
+            self.m.reconcile_exports(self.m.load_config(), owned)
+        self.assertEqual(self.calls, [["usbip", "unbind", "--busid", "1-2"]])
+
+    def test_failed_bind_is_reported_and_retains_recoverable_intent(self):
+        selected = self.selected(self.device())
+        config = self.config([selected])
+        owned = {}
+        with patch.object(self.m, "run", side_effect=self.m.USBIPError("bind failed")):
+            self.assertEqual(self.m.reconcile_exports(config, owned), {EXPORT: "bind failed"})
+        self.assertIn(EXPORT, self.m.read_json(self.m.state_path(), {}))
+        self.assertEqual(self.m.status()["server_status"]["errors"], {EXPORT: "bind failed"})
+        self.assertEqual(self.m.catalog()[0]["status"], "waiting")
+
+    def test_namespace_guard_fails_before_spawning_a_listener(self):
+        with patch.object(self.m.os, "readlink", return_value="net:[1]"), patch.object(self.m.subprocess, "Popen") as process:
+            with self.assertRaisesRegex(self.m.USBIPError, "PrivateNetwork"):
+                self.m.serve()
+        process.assert_not_called()
+
+    def test_generated_units_are_isolated_quoted_and_owned(self):
+        self.m.SCRIPT = self.root / 'with space/%and$dollar"quote.py'
+        with patch("shutil.which", return_value="/usr/sbin/modprobe"):
+            server = self.m.unit_text("server")
+            client = self.m.unit_text("client", CLIENT)
+        self.assertIn("PrivateNetwork=yes", server)
+        self.assertIn("RuntimeDirectory=proxmenux-usbip-server", server)
+        self.assertIn("RuntimeDirectoryPreserve=yes", server)
+        self.assertIn('%%and$$dollar\\"quote.py" serve', server)
+        self.assertNotIn("PrivateNetwork=yes", client)
+        self.assertNotIn("RuntimeDirectory=", client)
+        self.assertIn("vhci-hcd", client)
+        self.assertIn("Restart=on-failure", client)
+        unit = self.m.UNIT_DIR / self.m.SERVER_UNIT
+        unit.write_text("[Unit]\nDescription=Someone else's service\n")
+        with self.assertRaises(self.m.USBIPError), patch.object(self.m, "run") as command:
+            self.m.update_server(True)
+        command.assert_not_called()
+        self.assertFalse((self.m.CONFIG_DIR / "config.json").exists())
+
+    def test_failed_server_start_restores_configuration_and_removes_new_unit(self):
+        previous = self.config(server=False)
+        def fail_start(args, timeout=20):
+            self.calls.append(args)
+            if args[:3] == ["systemctl", "enable", "--now"]:
+                raise self.m.USBIPError("start failed")
+            return ""
+        with patch.object(self.m, "run", side_effect=fail_start), patch("shutil.which", return_value="/sbin/modprobe"):
+            with self.assertRaisesRegex(self.m.USBIPError, "start failed"):
+                self.m.update_server(True)
+        self.assertEqual(self.m.load_config(), previous)
+        self.assertFalse((self.m.UNIT_DIR / self.m.SERVER_UNIT).exists())
+        self.assertIn(["systemctl", "disable", "--now", self.m.SERVER_UNIT], self.calls)
+
+    def test_ssh_policy_and_argument_boundaries(self):
+        remote = self.m.validate_remote(self.remote["host"], 2222, str(self.key))
+        args = self.m.tunnel_args(remote | {"local_port": 33240})
+        for required in ("BatchMode=yes", "StrictHostKeyChecking=yes", "IdentitiesOnly=yes", "ForwardAgent=no",
+                         "PasswordAuthentication=no", "ExitOnForwardFailure=yes"):
+            self.assertIn(required, args)
+        self.assertIn("127.0.0.1:33240:/run/proxmenux-usbip-server/usbip.sock", args)
+        self.assertNotIn("StrictHostKeyChecking=no", args)
+        self.assertEqual(args[-2:], ["--", "usb.example.test"])
+        for host in ("-oProxyCommand=touch /tmp/no", "root@host", "host;id", "host\nother", "$(id)"):
+            with self.assertRaises(self.m.USBIPError):
+                self.m.validate_remote(host, 22, str(self.key))
+        self.key.chmod(0o644)
+        with self.assertRaisesRegex(self.m.USBIPError, "permissions"):
+            self.m.validate_remote("192.0.2.1", 22, str(self.key))
+
+    def test_ssh_credential_or_catalog_failure_never_saves_a_connection(self):
+        for failure in (self.m.USBIPError("Permission denied"), self.m.USBIPError("Connection timed out")):
+            with patch.object(self.m, "run", side_effect=failure):
+                with self.assertRaises(self.m.USBIPError):
+                    self.m.add_import(self.remote, EXPORT, 33240)
+            self.assertFalse((self.m.CONFIG_DIR / "config.json").exists())
+        with patch.object(self.m, "run", return_value='[{"id":"--bad","busid":"1-2"}]'):
+            with self.assertRaisesRegex(self.m.USBIPError, "valid"):
+                self.m.remote_catalog(self.remote)
+
+    def test_import_service_failure_rolls_back_and_occupied_port_is_rejected(self):
+        exports = [{"id": EXPORT, "status": "exported", "busid": "1-2"}]
+        previous = self.config()
+        with patch.object(self.m, "remote_catalog", return_value=exports):
+            with socket.socket() as busy:
+                busy.bind(("127.0.0.1", 0))
+                with self.assertRaisesRegex(self.m.USBIPError, "in use"):
+                    self.m.add_import(self.remote, EXPORT, busy.getsockname()[1])
+            with socket.socket() as unused:
+                unused.bind(("127.0.0.1", 0))
+                port = unused.getsockname()[1]
+            with patch("shutil.which", return_value="/sbin/modprobe"), patch.object(self.m, "enable_unit", side_effect=self.m.USBIPError("failed")):
+                with self.assertRaises(self.m.USBIPError):
+                    self.m.add_import(self.remote, EXPORT, port)
+        self.assertEqual(self.m.load_config(), previous)
+
+    def port(self, **overrides):
+        return {"port": 2, "status": "Port in Use", "local_busid": "3-1", "host": "127.0.0.1", "tcp_port": 33240,
+                "busid": "1-2"} | overrides
+
+    def test_real_upstream_port_output_and_unowned_port_cleanup(self):
+        output = """Imported USB devices
+====================
+Port 02: <Port in Use> at High Speed(480Mbps)
+       Test Zigbee adapter (10c4:ea60)
+       3-1 -> usbip://127.0.0.1:33240/1-2
+           -> remote bus/dev 001/002
+Port 03: <Port Error> at Full Speed(12Mbps)
+       Another adapter
+       3-2 -> usbip://192.0.2.1:3240/1-3
+"""
+        ports = self.m.parse_ports(output)
+        self.assertEqual(ports[0], self.port())
+        self.assertEqual(ports[1]["status"], "Port Error")
+        owner = self.m.RUN_DIR / f"client-{CLIENT}.json"
+        self.m.atomic_json(owner, ports[0])
+        reused = self.port(local_busid="3-7")
+        with patch.object(self.m, "imported_ports", return_value=[reused, ports[1]]), patch.object(self.m, "run") as command:
+            self.m.release_import(owner)
+        command.assert_not_called()
+        self.assertFalse(owner.exists())
+        self.m.atomic_json(owner, ports[0])
+        with patch.object(self.m, "imported_ports", return_value=ports), patch.object(self.m, "run") as command:
+            self.m.release_import(owner)
+        command.assert_called_once_with(["usbip", "detach", "--port", "2"])
+
+    def test_attach_reuses_live_port_and_recovers_error_with_new_remote_busid(self):
+        selected = self.remote | {"id": CLIENT, "local_port": 33240, "export_id": EXPORT}
+        owner = self.m.RUN_DIR / f"client-{CLIENT}.json"
+        self.m.atomic_json(owner, self.port())
+        with patch.object(self.m, "imported_ports", return_value=[self.port()]), patch.object(self.m, "remote_catalog") as remote:
+            self.assertEqual(self.m.attach_once(selected, owner)["port"], 2)
+        remote.assert_not_called()
+        ports = [self.port(status="Port Error")]
+        def usb(args, timeout=20):
+            self.calls.append(args)
+            if "detach" in args:
+                ports.clear()
+            elif "attach" in args:
+                ports.append(self.port(busid="2-4", local_busid="3-5"))
+            return ""
+        with patch.object(self.m, "imported_ports", side_effect=lambda: copy.deepcopy(ports)), patch.object(
+            self.m, "remote_catalog", return_value=[{"id": EXPORT, "status": "exported", "busid": "2-4"}]
+        ), patch.object(self.m, "run", side_effect=usb):
+            attached = self.m.attach_once(selected, owner)
+        self.assertEqual(attached["busid"], "2-4")
+        self.assertEqual(self.calls, [["usbip", "detach", "--port", "2"],
+            ["usbip", "--tcp-port", "33240", "attach", "--remote", "127.0.0.1", "--busid", "2-4"]])
+        self.assertEqual(self.m.read_json(owner, None), attached)
+
+    def test_missing_remote_or_unowned_import_is_never_detached(self):
+        selected = self.remote | {"id": CLIENT, "local_port": 33240, "export_id": EXPORT}
+        owner = self.m.RUN_DIR / f"client-{CLIENT}.json"
+        with patch.object(self.m, "remote_catalog", return_value=[]), patch.object(self.m, "run") as command:
+            with self.assertRaisesRegex(self.m.USBIPError, "unplugged"):
+                self.m.attach_once(selected, owner)
+        command.assert_not_called()
+        with patch.object(self.m, "remote_catalog", return_value=[{"id": EXPORT, "status": "exported", "busid": "1-2"}]), patch.object(
+            self.m, "imported_ports", return_value=[self.port()]
+        ), patch.object(self.m, "run") as command:
+            with self.assertRaisesRegex(self.m.USBIPError, "unowned"):
+                self.m.attach_once(selected, owner)
+        command.assert_not_called()
+
+    @unittest.skipUnless(sys.platform == "linux" and shutil.which("systemd-analyze"), "systemd verifier requires Linux")
+    def test_systemd_accepts_both_generated_service_files(self):
+        with patch("shutil.which", return_value="/bin/true"):
+            for name, content in ((self.m.SERVER_UNIT, self.m.unit_text("server")),
+                                  (self.m.unit_name(CLIENT), self.m.unit_text("client", CLIENT))):
+                path = self.m.UNIT_DIR / name
+                path.write_text(content)
+                result = subprocess.run(["systemd-analyze", "verify", str(path)], capture_output=True, text=True)
+                self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_menu_and_search_expose_server_and_client_actions(self):
+        menu = (ROOT / "scripts/menus/hw_grafics_menu.sh").read_text()
+        self.assertIn('bash "$LOCAL_SCRIPTS/devices/usbip_menu.sh"', menu)
+        index = json.loads((ROOT / "scripts/menus/search_index.json").read_text())
+        self.assertTrue(any(i["script"] == "devices/usbip_menu.sh" and "usbip" in i["keywords"] for i in index))
+        for script in (ROOT / "scripts/devices/usbip_menu.sh", ROOT / "scripts/menus/hw_grafics_menu.sh"):
+            subprocess.run(["bash", "-n", str(script)], check=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/workflows/test-usbip.yml

@@ -0,0 +1,49 @@
+name: USB/IP over SSH tests
+
+on:
+  pull_request:
+    paths:
+      - 'scripts/devices/**'
+      - 'scripts/menus/hw_grafics_menu.sh'
+      - 'scripts/menus/main_menu.sh'
+      - 'scripts/menus/search_index.json'
+      - '.github/scripts/tests/test_usbip.py'
+      - '.github/scripts/check_usbip_transport.py'
+      - '.github/workflows/test-usbip.yml'
+      - 'docs/usbip.md'
+  push:
+    branches: [main, develop]
+    paths:
+      - 'scripts/devices/**'
+      - '.github/scripts/tests/test_usbip.py'
+      - '.github/scripts/check_usbip_transport.py'
+      - '.github/workflows/test-usbip.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  usbip:
+    runs-on: ubuntu-latest
+    timeout-minutes: 8
+    strategy:
+      matrix:
+        python: ['3.11', '3.13']
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python }}
+      - 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
+      - name: Install fixture-only transport dependencies
+        run: |
+          sudo apt-get update -o APT::Update::Error-Mode=any
+          sudo apt-get install -y socat openssh-client openssh-server iproute2 util-linux
+      - name: Real SSH and private namespace transport, without USB hardware
+        run: sudo "$(command -v python3)" -B .github/scripts/check_usbip_transport.py

docs/usbip.md

@@ -0,0 +1,138 @@
+# USB/IP over SSH
+
+Open **Hardware: GPUs, Coral-TPU and USB/IP → USB/IP over SSH** to share a
+selected USB device or connect this Linux machine to a device on another host.
+The same menu is available through the main menu search (`usbip`, `zigbee`,
+`bluetooth` or `printer`). This implements the host and Linux client setup
+requested in [#350](https://github.com/MacRimi/ProxMenux/issues/350).
+
+## Server
+
+1. Enable the USB/IP server. Missing `usbip`, `socat`, `kmod` and `openssh-client`
+   packages are offered from the existing Debian repositories. Repository and
+   SSH server settings are preserved.
+2. Choose **Share a local USB device** and select a device by its bus ID, product
+   and serial number. Confirm that it will become unavailable to local programs
+   and guests using ordinary USB passthrough.
+3. Check **View USB/IP status**. The device should show `exported` and a bus ID.
+   Binding errors are reported separately rather than treated as a successful
+   export. Inspect the service log for startup errors:
+
+   ```sh
+   journalctl -u proxmenux-usbip-server.service
+   ```
+
+The server reloads saved selections every three seconds. Devices with a unique
+serial number can return on a different physical USB port after reconnecting.
+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
+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.
+
+## Transport and authentication
+
+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
+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 a filesystem Unix socket at
+`/run/proxmenux-usbip-server/usbip.sock`, inside a root-only directory with a
+0600 socket. Existing OpenSSH access forwards it to a loopback-only TCP port
+on the client. The SSH server must permit stream-local forwarding. Unknown or
+changed server host keys, bad SSH credentials and failed forwards are errors;
+the manager never disables host-key checking or changes `sshd_config`.
+
+The initial implementation uses **existing root SSH key access**. It does not
+create keys or authorize root login. Use this only between machines for which
+that access is already appropriate. Private key contents are never copied into
+the manager configuration. The local tunnel port is accessible to local users
+on the client, so the client machine must be trusted too.
+
+## Linux VM/client
+
+Run the client **inside the Linux VM that needs the device**. Connecting it to
+a Proxmox node alone does not make that device migrate with a VM. A supported
+Linux kernel must provide `vhci-hcd`; the server needs `usbip-host`. Proxmox
+kernel packages normally provide these modules, but activation fails visibly
+if they are missing. No out-of-tree kernel modules are installed.
+
+The client can use the menu when ProxMenux is installed, or the standalone
+standard-library Python manager. On a Debian Linux guest, install the packages
+and copy `scripts/devices/usbip_manager.py` from the reviewed ProxMenux checkout:
+
+```sh
+apt-get install usbip openssh-client kmod
+install -D -m 0755 usbip_manager.py /usr/local/share/proxmenux/scripts/devices/usbip_manager.py
+```
+
+First verify normal root SSH access from the client to the server using the
+intended private key and port. Verify the server host-key fingerprint through
+a trusted channel, then accept it through that normal SSH session. Automated
+connections read root's normal `known_hosts` file. SSH config aliases and
+`ProxyJump` settings are not used; enter the actual address and port.
+
+Then choose **Connect a remote USB device** and provide the server address,
+SSH port and absolute path to an existing private key (0400 or 0600). The menu
+authenticates and lists the server's configured exports before saving a client.
+Choose an unused local port for each device (33240 by default).
+
+Equivalent standalone commands:
+
+```sh
+MANAGER=/usr/local/share/proxmenux/scripts/devices/usbip_manager.py
+python3 "$MANAGER" remote --host 192.0.2.10 --ssh-port 22 --key /root/.ssh/id_ed25519
+# Substitute the export ID returned by the server:
+python3 "$MANAGER" add-import --host 192.0.2.10 --ssh-port 22 --key /root/.ssh/id_ed25519 --export-id abcdef123456 --local-port 33240
+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, so a serial-bearing dongle can move ports. 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.
+
+Only **one client at a time** can attach a physical USB device. A device already
+in use by another client may be listed as exported, but the kernel rejects a
+second attach; the client reports the error and waits. Applications must handle
+USB disconnect/reconnect themselves. A brief network interruption can still
+reset the USB device and interrupt a Zigbee coordinator, printer or Bluetooth
+session; this is not transparent USB high availability.
+
+Unprivileged LXC, Home Assistant OS and non-Linux clients require their own
+kernel/device permissions or add-on integration. This feature does not change
+LXC privileges, edit VM passthrough settings or deploy HAOS add-ons. A standard
+addon that expects raw USB/IP TCP also needs the SSH transport configured; it
+cannot directly connect to this server's isolated daemon.
+
+## Files and checks
+
+- Root-only configuration: `/etc/proxmenux/usbip/config.json` (0600).
+- Owned server unit: `proxmenux-usbip-server.service`.
+- Owned client units: `proxmenux-usbip-client-<id>.service`.
+- Server ownership/status: `/run/proxmenux-usbip-server/`.
+- Client ownership/status: `/run/proxmenux-usbip/`, independent of server cleanup.
+- Menu actions and package installations use the ProxMenux journal when present.
+
+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 encrypted SSH forwarding with an echo fixture. Physical device
+and guest-driver compatibility still requires a real deployment test.
+
+Protocol and transport references:
+[Linux USB/IP tools](https://github.com/torvalds/linux/tree/master/tools/usb/usbip),
+[Debian usbip manual](https://manpages.debian.org/trixie/usbip/usbip.8.en.html),
+[OpenSSH local socket forwarding](https://man.openbsd.org/ssh#L),
+[systemd PrivateNetwork](https://www.freedesktop.org/software/systemd/man/latest/systemd.exec.html#PrivateNetwork=).

lang/es.json

@@ -1,4 +1,34 @@
 {
+  "Hardware: GPUs, Coral-TPU and USB/IP": "Hardware: GPU, Coral-TPU y USB/IP",
+  "GPUs, Coral-TPU and USB/IP Menu": "Menú de GPU, Coral-TPU y USB/IP",
+  "USB/IP over SSH": "USB/IP mediante SSH",
+  "Run this menu as root.": "Ejecuta este menú como root.",
+  "USB/IP configuration updated. Open status to check the connection.": "Configuración de USB/IP actualizada. Abre el estado para comprobar la conexión.",
+  "Install the USB/IP dependencies from the configured Debian repositories?": "¿Instalar las dependencias de USB/IP desde los repositorios de Debian configurados?",
+  "No eligible USB devices found. Hubs and mass storage are excluded.": "No hay dispositivos USB compatibles. Se excluyen los concentradores y el almacenamiento masivo.",
+  "Select the USB device to share:": "Selecciona el dispositivo USB que quieres compartir:",
+  "The selected USB device will become unavailable to local applications and guests. Only one remote client can use it at a time. Share this device?": "El dispositivo USB dejará de estar disponible para las aplicaciones y los invitados locales. Solo puede usarlo un cliente remoto a la vez. ¿Compartirlo?",
+  "Select a saved USB/IP device:": "Selecciona un dispositivo USB/IP guardado:",
+  "The client must already trust the server SSH host key and have a root SSH key authorized there. Test normal SSH first. ProxMenux will not change SSH access or accept unknown host keys.": "El cliente debe confiar en la clave SSH del servidor y tener una clave de root autorizada en él. Comprueba primero el acceso SSH normal. ProxMenux no cambiará el acceso SSH ni aceptará claves de servidor desconocidas.",
+  "USB/IP server IP address or DNS hostname:": "Dirección IP o nombre DNS del servidor USB/IP:",
+  "Server SSH port:": "Puerto SSH del servidor:",
+  "Absolute path of the client SSH private key:": "Ruta absoluta de la clave privada SSH del cliente:",
+  "No shared devices are available on this server.": "No hay dispositivos compartidos disponibles en este servidor.",
+  "Select the remote USB device:": "Selecciona el dispositivo USB remoto:",
+  "Unused local loopback TCP port for this device:": "Puerto TCP de loopback local libre para este dispositivo:",
+  "Share selected USB devices through SSH, or connect this Linux machine to a USB/IP server.": "Comparte dispositivos USB mediante SSH o conecta este equipo Linux a un servidor USB/IP.",
+  "View USB/IP status": "Ver el estado de USB/IP",
+  "Enable USB/IP server": "Activar el servidor USB/IP",
+  "Share a local USB device": "Compartir un dispositivo USB local",
+  "Stop sharing a USB device": "Dejar de compartir un dispositivo USB",
+  "Disable USB/IP server": "Desactivar el servidor USB/IP",
+  "Connect a remote USB device": "Conectar un dispositivo USB remoto",
+  "Remove a remote USB connection": "Eliminar una conexión USB remota",
+  "Linux guest setup and migration notes": "Configuración del invitado Linux y notas sobre migración",
+  "Stop sharing this device? An attached remote client will disconnect.": "¿Dejar de compartir este dispositivo? Se desconectará el cliente remoto que lo esté usando.",
+  "Disable the server and release all USB devices managed by ProxMenux? Saved selections are kept for the next enable.": "¿Desactivar el servidor y liberar todos los dispositivos USB gestionados por ProxMenux? La selección guardada se conserva para la próxima activación.",
+  "Remove this connection and detach its USB device?": "¿Eliminar esta conexión y desconectar su dispositivo USB?",
+  "Run the client inside the Linux VM that needs the USB device. Install usbip, openssh-client and kmod, then copy usbip_manager.py to the same path as on the server. Configure key authentication and verify the server host key before adding a connection. The client uses the guest kernel and can reconnect after migration while the physical USB device stays on the server. Unprivileged LXC and Home Assistant OS need their own kernel/device or add-on setup. A physical device supports one attached client at a time.": "Ejecuta el cliente dentro de la VM Linux que necesita el USB. Instala usbip, openssh-client y kmod y copia usbip_manager.py en la misma ruta que en el servidor. Configura la autenticación por clave y verifica la clave del servidor antes de añadir la conexión. El cliente usa el kernel del invitado y puede reconectarse después de migrarlo mientras el USB físico sigue en el servidor. Los LXC sin privilegios y Home Assistant OS necesitan su propia configuración de kernel, dispositivos o complementos. Solo un cliente puede usar cada dispositivo físico a la vez.",
   "# Alternative if you prefer screen": "# Alternativa si prefieres la pantalla",
   "# Recommended: avoids disconnection during upgrade": "# Recomendado: evita la desconexión durante la actualización",
   "(Checked entries will be removed. Uncheck to keep in VM.)": "(Las entradas marcadas se eliminarán. Desmarque para mantener en VM).",

scripts/devices/usbip_manager.py

@@ -0,0 +1,677 @@
+#!/usr/bin/env python3
+"""Manage selected USB/IP exports and Linux clients over authenticated SSH.
+
+The server unit has a private network namespace. OpenSSH reaches its root-only
+filesystem Unix socket; usbipd never listens in the host's network namespace.
+Only this module's recorded devices and imported ports are released on stop.
+"""
+
+import argparse
+from contextlib import contextmanager
+import copy
+import fcntl
+import ipaddress
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import socket
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+import uuid
+
+CONFIG_DIR = Path("/etc/proxmenux/usbip")
+RUN_DIR = Path("/run/proxmenux-usbip")
+SERVER_RUN_DIR = Path("/run/proxmenux-usbip-server")
+UNIT_DIR = Path("/etc/systemd/system")
+SYSFS = Path("/sys/bus/usb/devices")
+SCRIPT = Path(__file__).resolve()
+REMOTE_SCRIPT = "/usr/local/share/proxmenux/scripts/devices/usbip_manager.py"
+REMOTE_SOCKET = "/run/proxmenux-usbip-server/usbip.sock"
+SERVER_UNIT = "proxmenux-usbip-server.service"
+MARKER = "# Managed by ProxMenux USB/IP"
+BUSID = re.compile(r"[0-9]+-[0-9]+(?:\.[0-9]+)*\Z")
+IDENTIFIER = re.compile(r"[a-f0-9]{12}\Z")
+
+
+class USBIPError(Exception):
+    pass
+
+
+def run(args, timeout=20):
+    try:
+        result = subprocess.run(args, capture_output=True, text=True, timeout=timeout,
+                                env={**os.environ, "LC_ALL": "C"})
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise USBIPError(f"Could not run {args[0]}: {type(exc).__name__}") from exc
+    if result.returncode:
+        detail = (result.stderr or result.stdout).strip()[:500]
+        raise USBIPError(f"{Path(args[0]).name} failed: {detail}")
+    return result.stdout
+
+
+def atomic_json(path, value):
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    fd, name = tempfile.mkstemp(prefix=".usbip-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w") as handle:
+            os.fchmod(handle.fileno(), 0o600)
+            json.dump(value, handle, indent=2)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(name, path)
+    finally:
+        if os.path.exists(name):
+            os.unlink(name)
+
+
+def read_json(path, default):
+    try:
+        if path.stat().st_size > 1024 * 1024:
+            raise USBIPError(f"File is too large: {path}")
+        return json.loads(path.read_text())
+    except FileNotFoundError:
+        return copy.deepcopy(default)
+    except (ValueError, OSError) as exc:
+        raise USBIPError(f"Cannot read {path}") from exc
+
+
+def load_config():
+    value = read_json(CONFIG_DIR / "config.json", {"server": False, "exports": [], "imports": []})
+    if not isinstance(value, dict) or not isinstance(value.get("server"), bool):
+        raise USBIPError("Invalid USB/IP configuration")
+    for field in ("exports", "imports"):
+        if not isinstance(value.get(field), list) or len(value[field]) > 64:
+            raise USBIPError("Invalid USB/IP configuration")
+        for item in value[field]:
+            if not isinstance(item, dict) or not isinstance(item.get("id"), str) or not IDENTIFIER.fullmatch(item["id"]):
+                raise USBIPError("Invalid USB/IP device identifier")
+    return value
+
+
+@contextmanager
+def config_lock():
+    CONFIG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
+    with (CONFIG_DIR / ".lock").open("a") as lock:
+        os.fchmod(lock.fileno(), 0o600)
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        yield
+
+
+def attribute(path, name):
+    try:
+        return (path / name).read_text().strip()[:200]
+    except (FileNotFoundError, OSError):
+        return ""
+
+
+def inventory():
+    devices = []
+    for path in sorted(SYSFS.glob("*")):
+        if not BUSID.fullmatch(path.name):
+            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
+        classes = {attribute(path, "bDeviceClass").lower()}
+        classes.update(attribute(p, "bInterfaceClass").lower() for p in path.glob(f"{path.name}:*"))
+        # Interfaces are also symlinked directly below /sys/bus/usb/devices.
+        classes.update(attribute(p, "bInterfaceClass").lower() for p in SYSFS.glob(f"{path.name}:*"))
+        blocked = "USB hubs cannot be shared" if "09" in classes else (
+            "USB mass storage cannot be shared" if "08" in classes else (
+                "USB class could not be read" if not re.fullmatch(r"[a-f0-9]{2}", attribute(path, "bDeviceClass").lower()) else ""))
+        try:
+            driver = (path / "driver").resolve(strict=True).name
+        except OSError:
+            driver = ""
+        devices.append({"busid": path.name, "vendor": vendor.lower(), "product": product.lower(),
+                        "serial": attribute(path, "serial"), "name": attribute(path, "product") or f"{vendor}:{product}",
+                        "driver": driver, "blocked": blocked})
+    return devices
+
+
+def same_device(expected, device):
+    if any(expected.get(k) != device.get(k) for k in ("vendor", "product", "serial")):
+        return False
+    return bool(expected.get("serial")) or expected.get("busid") == device.get("busid")
+
+
+def resolve_device(expected, devices):
+    matches = [d for d in devices if same_device(expected, d)]
+    if len(matches) != 1:
+        raise USBIPError("Device is missing or its identity is ambiguous")
+    if matches[0]["blocked"]:
+        raise USBIPError(matches[0]["blocked"])
+    return matches[0]
+
+
+def state_path():
+    return SERVER_RUN_DIR / "exports.json"
+
+
+def release_export(identifier, owned, devices):
+    previous = owned[identifier]
+    current = next((d for d in devices if d["busid"] == previous["busid"]), None)
+    if current and same_device(previous, current) and current["driver"] == "usbip-host":
+        run(["usbip", "unbind", "--busid", current["busid"]])
+    del owned[identifier]
+    atomic_json(state_path(), owned)
+
+
+def reconcile_exports(config, owned):
+    devices = inventory()
+    desired = {item["id"]: item for item in config["exports"]} if config["server"] else {}
+    errors = {}
+    for identifier in list(owned):
+        if identifier not in desired:
+            try:
+                release_export(identifier, owned, devices)
+            except USBIPError as exc:
+                errors[identifier] = str(exc)
+    for identifier, selected in desired.items():
+        try:
+            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)
+                previous = None
+            if device["driver"] == "usbip-host":
+                if not previous:
+                    raise USBIPError("Device is already exported outside ProxMenux; it was not adopted")
+                continue
+            # 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"]])
+        except USBIPError as exc:
+            errors[identifier] = str(exc)
+    atomic_json(SERVER_RUN_DIR / "status.json", {"errors": errors, "updated": time.time()})
+    return errors
+
+
+def catalog():
+    config, devices, owned = load_config(), inventory(), read_json(state_path(), {})
+    exports = []
+    for selected in config["exports"]:
+        row = dict(selected, status="waiting", busid=None)
+        try:
+            current = resolve_device(selected, devices)
+            if config["server"] and selected["id"] in owned and current["driver"] == "usbip-host" and (
+                owned[selected["id"]]["busid"] == current["busid"] and same_device(owned[selected["id"]], current)
+            ):
+                row.update(busid=current["busid"], status="exported")
+        except USBIPError as exc:
+            row["error"] = str(exc)
+        exports.append(row)
+    return exports
+
+
+def systemd_quote(value):
+    if any(ord(c) < 32 for c in str(value)):
+        raise USBIPError("Control characters are not allowed in systemd arguments")
+    return '"' + str(value).replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%").replace("$", "$$") + '"'
+
+
+def unit_text(role, identifier=None):
+    command = f"{systemd_quote(sys.executable)} {systemd_quote(SCRIPT)}"
+    if role == "server":
+        extra = "PrivateNetwork=yes\nRuntimeDirectory=proxmenux-usbip-server\nRuntimeDirectoryMode=0700\nRuntimeDirectoryPreserve=yes\n"
+        start = f"{command} serve"
+        module = "usbip-host"
+    else:
+        if not IDENTIFIER.fullmatch(identifier or ""):
+            raise USBIPError("Invalid import identifier")
+        start = f"{command} connect {identifier}"
+        module = "vhci-hcd"
+    import shutil
+    modprobe = shutil.which("modprobe")
+    if not modprobe:
+        raise USBIPError("modprobe is not installed")
+    return (f"{MARKER}\n[Unit]\nDescription=ProxMenux USB/IP {role} over SSH\n"
+            + ("Wants=network-online.target\nAfter=network-online.target\n" if role == "client" else "")
+            + "StartLimitIntervalSec=0\n\n[Service]\nType=simple\nUser=root\nUMask=0077\n"
+            + f"ExecStartPre={systemd_quote(modprobe)} {module}\nExecStart={start}\n"
+            + "Restart=on-failure\nRestartSec=5\nTimeoutStopSec=30\nPrivateTmp=yes\nNoNewPrivileges=yes\n"
+            + (extra if role == "server" else "")
+            + "\n[Install]\nWantedBy=multi-user.target\n")
+
+
+def unit_name(identifier):
+    if not IDENTIFIER.fullmatch(identifier):
+        raise USBIPError("Invalid import identifier")
+    return f"proxmenux-usbip-client-{identifier}.service"
+
+
+def check_unit_owned(name):
+    path = UNIT_DIR / name
+    if path.is_symlink() or (path.exists() and not path.read_text().startswith(MARKER + "\n")):
+        raise USBIPError(f"Refusing to replace a unit not owned by ProxMenux: {name}")
+    return path
+
+
+def write_unit(name, content):
+    path = check_unit_owned(name)
+    path.write_text(content)
+    path.chmod(0o644)
+
+
+def enable_unit(name, content):
+    path = check_unit_owned(name)
+    original = path.read_text() if path.exists() else None
+    try:
+        write_unit(name, content)
+        run(["systemctl", "daemon-reload"])
+        run(["systemctl", "enable", "--now", name], timeout=40)
+        run(["systemctl", "is-active", "--quiet", name])
+    except USBIPError:
+        if original is None:
+            try:
+                run(["systemctl", "disable", "--now", name], timeout=40)
+            except USBIPError:
+                pass
+            path.unlink(missing_ok=True)
+        else:
+            path.write_text(original)
+        run(["systemctl", "daemon-reload"])
+        raise
+
+
+def remove_unit(name):
+    path = check_unit_owned(name)
+    if path.exists():
+        run(["systemctl", "disable", "--now", name], timeout=40)
+        path.unlink()
+        run(["systemctl", "daemon-reload"])
+
+
+def validate_remote(host, port, key):
+    try:
+        ipaddress.ip_address(host)
+    except ValueError:
+        if len(host) > 253 or not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9.-]*[A-Za-z0-9])?", host):
+            raise USBIPError("Enter an IP address or DNS hostname without a username or SSH options")
+    if not 1 <= int(port) <= 65535:
+        raise USBIPError("Invalid SSH port")
+    key_path = Path(key)
+    if not key_path.is_absolute() or any(ord(c) < 32 for c in key):
+        raise USBIPError("SSH key must be an absolute path")
+    try:
+        mode = key_path.stat()
+        if not stat.S_ISREG(mode.st_mode) or mode.st_uid != os.geteuid() or mode.st_mode & 0o077:
+            raise USBIPError("SSH key must belong to this user and have permissions 0600 or 0400")
+    except OSError as exc:
+        raise USBIPError("SSH private key is not readable") from exc
+    return {"host": host, "ssh_port": int(port), "key": str(key_path)}
+
+
+def ssh_args(remote):
+    return ["ssh", "-F", "/dev/null", "-T", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=yes",
+            "-o", "PasswordAuthentication=no", "-o", "KbdInteractiveAuthentication=no",
+            "-o", "IdentitiesOnly=yes", "-o", "ForwardAgent=no", "-o", "ControlMaster=no",
+            "-o", "ControlPath=none", "-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=15",
+            "-o", "ServerAliveCountMax=3", "-o", "ExitOnForwardFailure=yes",
+            "-p", str(remote["ssh_port"]), "-i", remote["key"], "-l", "root"]
+
+
+def remote_catalog(remote):
+    # The remote command is fixed. All user input stays in separate local argv.
+    output = run(ssh_args(remote) + ["--", remote["host"], f"python3 {REMOTE_SCRIPT} catalog"])
+    try:
+        result = json.loads(output)
+        if not isinstance(result, list) or len(result) > 64:
+            raise ValueError()
+        for item in result:
+            if not isinstance(item, dict) or not isinstance(item.get("id"), str) or not IDENTIFIER.fullmatch(item["id"]):
+                raise ValueError()
+            if item.get("busid") is not None and (not isinstance(item["busid"], str) or not BUSID.fullmatch(item["busid"])):
+                raise ValueError()
+        return result
+    except (ValueError, TypeError) as exc:
+        raise USBIPError("Remote host did not return a valid ProxMenux USB/IP catalog") from exc
+
+
+def tunnel_args(selected):
+    return ssh_args(selected) + ["-N", "-L", f"127.0.0.1:{selected['local_port']}:{REMOTE_SOCKET}",
+                                 "--", selected["host"]]
+
+
+def parse_ports(output):
+    ports, current = [], None
+    for line in output.splitlines():
+        match = re.match(r"Port (\d+): <([^>]+)>", line)
+        if match:
+            current = {"port": int(match[1]), "status": match[2]}
+            continue
+        match = re.match(r"\s*(\S+) -> usbip://([^\s/]+):([0-9]+)/([0-9.-]+)\s*$", line)
+        if match and current is not None:
+            ports.append(dict(current, local_busid=match[1], host=match[2], tcp_port=int(match[3]), busid=match[4]))
+            current = None
+    return ports
+
+
+def imported_ports():
+    return parse_ports(run(["usbip", "port"]))
+
+
+def owns_import(saved, current):
+    return all(saved.get(k) == current.get(k) for k in ("port", "local_busid", "host", "tcp_port", "busid"))
+
+
+def release_import(path):
+    saved = read_json(path, None)
+    if saved:
+        for current in imported_ports():
+            if owns_import(saved, current):
+                run(["usbip", "detach", "--port", str(current["port"])])
+    path.unlink(missing_ok=True)
+
+
+def find_import(selected, busid):
+    return next((p for p in imported_ports() if p["host"] == "127.0.0.1" and
+                 p["tcp_port"] == selected["local_port"] and p["busid"] == busid), None)
+
+
+def attach_once(selected, owner):
+    saved = read_json(owner, None)
+    if saved:
+        existing = next((p for p in imported_ports() if owns_import(saved, p)), None)
+        if existing and existing["status"] == "Port in Use":
+            return existing
+        # A kernel port in an error state must be released before retrying.
+        release_import(owner)
+    remote = next((e for e in remote_catalog(selected) if e["id"] == selected["export_id"]), None)
+    if not remote or remote.get("status") != "exported" or not remote.get("busid"):
+        raise USBIPError("Remote device is unplugged, disabled or waiting to be exported")
+    if find_import(selected, remote["busid"]):
+        raise USBIPError("This endpoint has an unowned import; it was not adopted or detached")
+    run(["usbip", "--tcp-port", str(selected["local_port"]), "attach", "--remote", "127.0.0.1",
+         "--busid", remote["busid"]])
+    attached = find_import(selected, remote["busid"])
+    if not attached:
+        raise USBIPError("USB/IP did not report an attached port")
+    atomic_json(owner, attached)
+    return attached
+
+
+def require_private_network():
+    if os.readlink("/proc/self/ns/net") == os.readlink("/proc/1/ns/net"):
+        raise USBIPError("The USB/IP server must run in its systemd PrivateNetwork namespace")
+
+
+def stop_child(process):
+    if process and process.poll() is None:
+        process.terminate()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=5)
+
+
+@contextmanager
+def stop_event():
+    import threading
+    stopped = threading.Event()
+    old = {sig: signal.signal(sig, lambda *_: stopped.set()) for sig in (signal.SIGTERM, signal.SIGINT)}
+    try:
+        yield stopped
+    finally:
+        for sig, handler in old.items():
+            signal.signal(sig, handler)
+
+
+def serve():
+    require_private_network()
+    SERVER_RUN_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
+    SERVER_RUN_DIR.chmod(0o700)
+    socket_path = SERVER_RUN_DIR / "usbip.sock"
+    socket_path.unlink(missing_ok=True)
+    owned = read_json(state_path(), {})
+    daemon = proxy = None
+    try:
+        daemon = subprocess.Popen(["usbipd", "--ipv4", "--tcp-port", "3240"])
+        proxy = subprocess.Popen(["socat", f"UNIX-LISTEN:{socket_path},mode=0600,fork",
+                                  "TCP:127.0.0.1:3240"])
+        with stop_event() as stopped:
+            previous_errors = None
+            while not stopped.is_set():
+                if daemon.poll() is not None or proxy.poll() is not None:
+                    raise USBIPError("USB/IP daemon or Unix proxy exited")
+                config = load_config()
+                if not config["server"]:
+                    break
+                errors = reconcile_exports(config, owned)
+                if errors != previous_errors:
+                    for identifier, error in errors.items():
+                        print(f"Export {identifier}: {error}", file=sys.stderr, flush=True)
+                    previous_errors = errors
+                stopped.wait(3)
+    finally:
+        devices = inventory()
+        for identifier in list(owned):
+            try:
+                release_export(identifier, owned, devices)
+            except USBIPError as exc:
+                print(f"Could not release export {identifier}: {exc}", file=sys.stderr, flush=True)
+        stop_child(proxy)
+        stop_child(daemon)
+        socket_path.unlink(missing_ok=True)
+
+
+def connect(identifier):
+    selected = next((i for i in load_config()["imports"] if i["id"] == identifier), None)
+    if selected is None:
+        raise USBIPError("Import not found")
+    validate_remote(selected["host"], selected["ssh_port"], selected["key"])
+    RUN_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
+    owner = RUN_DIR / f"client-{identifier}.json"
+    status_path = RUN_DIR / f"client-{identifier}-status.json"
+    tunnel, last_error = None, None
+    try:
+        # Recover only this client's exact previous port after an interrupted run.
+        release_import(owner)
+        with stop_event() as stopped:
+            while not stopped.is_set():
+                try:
+                    if tunnel is None or tunnel.poll() is not None:
+                        release_import(owner)
+                        stop_child(tunnel)
+                        tunnel = subprocess.Popen(tunnel_args(selected), stdin=subprocess.DEVNULL)
+                        if stopped.wait(1):
+                            break
+                        if tunnel.poll() is not None:
+                            raise USBIPError("SSH tunnel failed; check key, host key and forwarding permissions")
+                    attached = attach_once(selected, owner)
+                    atomic_json(status_path, {"state": "attached", "port": attached["port"], "updated": time.time()})
+                    last_error = None
+                except USBIPError as exc:
+                    atomic_json(status_path, {"state": "waiting", "error": str(exc), "updated": time.time()})
+                    if str(exc) != last_error:
+                        print(str(exc), file=sys.stderr, flush=True)
+                        last_error = str(exc)
+                stopped.wait(5)
+    finally:
+        try:
+            release_import(owner)
+        finally:
+            stop_child(tunnel)
+            atomic_json(status_path, {"state": "stopped", "updated": time.time()})
+
+
+def update_server(enabled):
+    with config_lock():
+        config = load_config()
+        previous = copy.deepcopy(config)
+        check_unit_owned(SERVER_UNIT)
+        config["server"] = enabled
+        atomic_json(CONFIG_DIR / "config.json", config)
+        try:
+            if enabled:
+                enable_unit(SERVER_UNIT, unit_text("server"))
+            else:
+                remove_unit(SERVER_UNIT)
+                # Retry cleanup even if the worker had previously crashed.
+                owned = read_json(state_path(), {})
+                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:
+            if enabled:
+                atomic_json(CONFIG_DIR / "config.json", previous)
+            raise
+
+
+def add_export(busid):
+    if not BUSID.fullmatch(busid):
+        raise USBIPError("Invalid USB bus ID")
+    with config_lock():
+        config = load_config()
+        if len(config["exports"]) >= 64:
+            raise USBIPError("At most 64 exports can be configured")
+        if not config["server"]:
+            raise USBIPError("Enable the USB/IP server first")
+        device = next((d for d in inventory() if d["busid"] == busid), None)
+        if not device:
+            raise USBIPError("Device not found")
+        if device["blocked"]:
+            raise USBIPError(device["blocked"])
+        if device["driver"] == "usbip-host":
+            raise USBIPError("Device is already exported; it was not adopted")
+        if any(same_device(e, device) for e in config["exports"]):
+            raise USBIPError("Device is already configured")
+        resolve_device(device, inventory())
+        selected = {k: device[k] for k in ("busid", "vendor", "product", "serial", "name")}
+        selected["id"] = uuid.uuid4().hex[:12]
+        config["exports"].append(selected)
+        atomic_json(CONFIG_DIR / "config.json", config)
+        return selected
+
+
+def remove_export(identifier):
+    with config_lock():
+        config = load_config()
+        if not any(e["id"] == identifier for e in config["exports"]):
+            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.
+
+
+def add_import(remote, export_id, local_port):
+    if not IDENTIFIER.fullmatch(export_id):
+        raise USBIPError("Invalid remote export ID")
+    if not 1024 <= local_port <= 65535:
+        raise USBIPError("Use a local TCP port between 1024 and 65535")
+    exports = remote_catalog(remote)  # Authenticate and validate before saving anything.
+    if not any(e["id"] == export_id and e.get("status") == "exported" for e in exports):
+        raise USBIPError("Remote export is not available; enable the server and select a device first")
+    with config_lock():
+        config = load_config()
+        if len(config["imports"]) >= 64:
+            raise USBIPError("At most 64 imports can be configured")
+        if any(i["local_port"] == local_port or (i["host"] == remote["host"] and
+               i["ssh_port"] == remote["ssh_port"] and i["export_id"] == export_id) for i in config["imports"]):
+            raise USBIPError("This remote device or local port is already configured")
+        with socket.socket() as check:
+            try:
+                check.bind(("127.0.0.1", local_port))
+            except OSError as exc:
+                raise USBIPError("Local tunnel port is already in use") from exc
+        previous = copy.deepcopy(config)
+        selected = dict(remote, export_id=export_id, local_port=local_port, id=uuid.uuid4().hex[:12])
+        config["imports"].append(selected)
+        atomic_json(CONFIG_DIR / "config.json", config)
+        try:
+            enable_unit(unit_name(selected["id"]), unit_text("client", selected["id"]))
+        except USBIPError:
+            atomic_json(CONFIG_DIR / "config.json", previous)
+            raise
+        return selected
+
+
+def remove_import(identifier):
+    with config_lock():
+        config = load_config()
+        if not any(i["id"] == identifier for i in config["imports"]):
+            raise USBIPError("Import not found")
+        remove_unit(unit_name(identifier))
+        # Also recover a port left by a process killed without its finally block.
+        release_import(RUN_DIR / f"client-{identifier}.json")
+        config["imports"] = [i for i in config["imports"] if i["id"] != identifier]
+        atomic_json(CONFIG_DIR / "config.json", config)
+        (RUN_DIR / f"client-{identifier}-status.json").unlink(missing_ok=True)
+
+
+def status():
+    config = load_config()
+    result = {"server": config["server"], "exports": catalog(), "imports": []}
+    def active(unit):
+        try:
+            run(["systemctl", "is-active", "--quiet", unit])
+            return True
+        except USBIPError:
+            return False
+    result["server_active"] = active(SERVER_UNIT)
+    result["server_status"] = read_json(SERVER_RUN_DIR / "status.json", {})
+    for item in config["imports"]:
+        result["imports"].append(dict(item, active=active(unit_name(item["id"])),
+                                      status=read_json(RUN_DIR / f"client-{item['id']}-status.json", {"state": "unknown"})))
+    return result
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    for name in ("inventory", "catalog", "status", "server-enable", "server-disable", "serve"):
+        commands.add_parser(name)
+    for name in ("add-export", "remove-export", "remove-import", "connect"):
+        commands.add_parser(name).add_argument("identifier")
+    for name in ("remote", "add-import"):
+        cmd = commands.add_parser(name)
+        cmd.add_argument("--host", required=True)
+        cmd.add_argument("--ssh-port", type=int, default=22)
+        cmd.add_argument("--key", required=True)
+        if name == "add-import":
+            cmd.add_argument("--export-id", required=True)
+            cmd.add_argument("--local-port", type=int, default=33240)
+    args = parser.parse_args()
+    try:
+        if os.geteuid() != 0:
+            raise USBIPError("Run this manager as root")
+        output = None
+        if args.command == "inventory":
+            output = inventory()
+        elif args.command == "catalog":
+            output = catalog()
+        elif args.command == "status":
+            output = status()
+        elif args.command in ("server-enable", "server-disable"):
+            update_server(args.command == "server-enable")
+        elif args.command == "add-export":
+            output = add_export(args.identifier)
+        elif args.command == "remove-export":
+            remove_export(args.identifier)
+        elif args.command == "remove-import":
+            remove_import(args.identifier)
+        elif args.command == "serve":
+            serve()
+        elif args.command == "connect":
+            connect(args.identifier)
+        elif args.command in ("remote", "add-import"):
+            remote = validate_remote(args.host, args.ssh_port, args.key)
+            output = remote_catalog(remote) if args.command == "remote" else add_import(remote, args.export_id, args.local_port)
+        if output is not None:
+            print(json.dumps(output, indent=2))
+    except (USBIPError, OSError) as exc:
+        print(f"USB/IP: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/devices/usbip_menu.sh

@@ -0,0 +1,182 @@
+#!/bin/bash
+# ProxMenux - selected USB devices over authenticated SSH (GPL-3.0)
+set -o pipefail
+umask 077
+
+BASE_DIR="/usr/local/share/proxmenux"
+LOCAL_SCRIPTS="$BASE_DIR/scripts"
+MANAGER="$LOCAL_SCRIPTS/devices/usbip_manager.py"
+if [[ -f "$BASE_DIR/utils.sh" ]]; then
+    source "$BASE_DIR/utils.sh"
+    load_language
+    initialize_cache
+else
+    translate() { printf '%s' "$1"; }
+fi
+if [[ -f "$LOCAL_SCRIPTS/global/pmx_journal.sh" ]]; then
+    source "$LOCAL_SCRIPTS/global/pmx_journal.sh"
+fi
+
+if [[ $EUID -ne 0 ]]; then
+    printf '%s\n' "$(translate 'Run this menu as root.')" >&2
+    exit 1
+fi
+
+usbip_dialog() {
+    dialog --backtitle "ProxMenux" --title "$(translate 'USB/IP over SSH')" "$@"
+}
+
+usbip_action() {
+    local output result=0
+    output=$(mktemp) || return 1
+    if python3 "$MANAGER" "$@" >"$output" 2>&1; then
+        if declare -F pmx_journal_context >/dev/null; then
+            pmx_journal_context "usbip_${1//-/_}" "1.0" "usbip_menu.sh"
+            pmx_record_execution "USB/IP: $1" "python3 $MANAGER $1"
+        fi
+        usbip_dialog --msgbox "$(translate 'USB/IP configuration updated. Open status to check the connection.')" 9 72
+    else
+        usbip_dialog --textbox "$output" 20 85
+        result=1
+    fi
+    rm -f "$output"
+    return "$result"
+}
+
+usbip_dependencies() {
+    local role="$1" package output
+    local -a packages=(usbip openssh-client kmod)
+    [[ "$role" == server ]] && packages+=(socat)
+    local missing=false
+    command -v usbip >/dev/null || missing=true
+    command -v ssh >/dev/null || missing=true
+    command -v modprobe >/dev/null || missing=true
+    if [[ "$role" == server ]]; then
+        command -v usbipd >/dev/null || missing=true
+        command -v socat >/dev/null || missing=true
+    fi
+    [[ "$missing" == false ]] && return 0
+    usbip_dialog --yesno "$(translate 'Install the USB/IP dependencies from the configured Debian repositories?')" 9 72 || return 1
+    output=$(mktemp) || return 1
+    # Keep the operator's repository configuration; do not enable new sources.
+    if ! apt-get update -o APT::Update::Error-Mode=any >"$output" 2>&1; then
+        usbip_dialog --textbox "$output" 20 85
+        rm -f "$output"
+        return 1
+    fi
+    for package in "${packages[@]}"; do
+        if declare -F pmx_install_pkg >/dev/null; then
+            pmx_install_pkg "$package" >>"$output" 2>&1
+        else
+            DEBIAN_FRONTEND=noninteractive apt-get install -y "$package" >>"$output" 2>&1
+        fi
+        if [[ $? -ne 0 ]]; then
+            usbip_dialog --textbox "$output" 20 85
+            rm -f "$output"
+            return 1
+        fi
+    done
+    rm -f "$output"
+}
+
+usbip_status() {
+    local output raw
+    output=$(mktemp) || return 1
+    raw=$(mktemp) || { rm -f "$output"; return 1; }
+    if python3 "$MANAGER" status >"$raw" 2>"$output"; then
+        jq -r '"SERVER configured: \(.server)  running: \(.server_active)\n",
+          (.exports[] | "EXPORT \(.id)  \(.name)\n  \(.vendor):\(.product)  serial: \(.serial)\n  bus: \(.busid // "waiting")  state: \(.status)\n"),
+          (.server_status.errors // {} | to_entries[] | "ERROR \(.key): \(.value)"),
+          "\nCLIENTS:",
+          (.imports[] | "\(.id)  root@\(.host):\(.ssh_port)\n  export: \(.export_id)  local tunnel: 127.0.0.1:\(.local_port)\n  running: \(.active)  state: \(.status.state)  \(.status.error // "")\n")' "$raw" >"$output"
+    fi
+    usbip_dialog --textbox "$output" 26 100
+    rm -f "$output" "$raw"
+}
+
+usbip_add_export() {
+    local busid label chosen
+    local -a choices=()
+    while IFS=$'\t' read -r busid label; do
+        [[ -n "$busid" ]] && choices+=("$busid" "$label")
+    done < <(python3 "$MANAGER" inventory | jq -r '.[] | select(.blocked == "" and .driver != "usbip-host") | [.busid, "\(.name) [\(.vendor):\(.product)] serial: \(.serial)"] | @tsv')
+    if [[ ${#choices[@]} -eq 0 ]]; then
+        usbip_dialog --msgbox "$(translate 'No eligible USB devices found. Hubs and mass storage are excluded.')" 10 75
+        return
+    fi
+    chosen=$(usbip_dialog --stdout --menu "$(translate 'Select the USB device to share:')" 22 100 12 "${choices[@]}") || return
+    usbip_dialog --yesno "$(translate 'The selected USB device will become unavailable to local applications and guests. Only one remote client can use it at a time. Share this device?')\n\n$chosen" 12 80 || return
+    usbip_action add-export "$chosen"
+}
+
+usbip_select_saved() {
+    local field="$1" identifier label
+    local -a choices=()
+    while IFS=$'\t' read -r identifier label; do
+        [[ -n "$identifier" ]] && choices+=("$identifier" "$label")
+    done < <(python3 "$MANAGER" status | jq -r --arg field "$field" '.[$field][] | [.id, (.name // .host)] | @tsv')
+    [[ ${#choices[@]} -gt 0 ]] || return 1
+    usbip_dialog --stdout --menu "$(translate 'Select a saved USB/IP device:')" 20 90 10 "${choices[@]}"
+}
+
+usbip_add_import() {
+    usbip_dependencies client || return
+    local host ssh_port key local_port output error selected identifier label
+    local -a choices=()
+    usbip_dialog --msgbox "$(translate 'The client must already trust the server SSH host key and have a root SSH key authorized there. Test normal SSH first. ProxMenux will not change SSH access or accept unknown host keys.')" 12 85
+    host=$(usbip_dialog --stdout --inputbox "$(translate 'USB/IP server IP address or DNS hostname:')" 9 75) || return
+    ssh_port=$(usbip_dialog --stdout --inputbox "$(translate 'Server SSH port:')" 9 70 "22") || return
+    key=$(usbip_dialog --stdout --inputbox "$(translate 'Absolute path of the client SSH private key:')" 9 80 "/root/.ssh/id_ed25519") || return
+    output=$(mktemp) || return
+    error=$(mktemp) || { rm -f "$output"; return; }
+    if ! python3 "$MANAGER" remote --host "$host" --ssh-port "$ssh_port" --key "$key" >"$output" 2>"$error"; then
+        usbip_dialog --textbox "$error" 20 90
+        rm -f "$output" "$error"
+        return
+    fi
+    while IFS=$'\t' read -r identifier label; do
+        [[ -n "$identifier" ]] && choices+=("$identifier" "$label")
+    done < <(jq -r '.[] | select(.status == "exported") | [.id, "\(.name) [\(.busid)]"] | @tsv' "$output")
+    rm -f "$output" "$error"
+    if [[ ${#choices[@]} -eq 0 ]]; then
+        usbip_dialog --msgbox "$(translate 'No shared devices are available on this server.')" 9 72
+        return
+    fi
+    selected=$(usbip_dialog --stdout --menu "$(translate 'Select the remote USB device:')" 20 95 10 "${choices[@]}") || return
+    local_port=$(usbip_dialog --stdout --inputbox "$(translate 'Unused local loopback TCP port for this device:')" 9 75 "33240") || return
+    usbip_action add-import --host "$host" --ssh-port "$ssh_port" --key "$key" --export-id "$selected" --local-port "$local_port"
+}
+
+while true; do
+    option=$(usbip_dialog --stdout --menu "$(translate 'Share selected USB devices through SSH, or connect this Linux machine to a USB/IP server.')" 22 95 12 \
+        "1" "$(translate 'View USB/IP status')" \
+        "2" "$(translate 'Enable USB/IP server')" \
+        "3" "$(translate 'Share a local USB device')" \
+        "4" "$(translate 'Stop sharing a USB device')" \
+        "5" "$(translate 'Disable USB/IP server')" \
+        "6" "$(translate 'Connect a remote USB device')" \
+        "7" "$(translate 'Remove a remote USB connection')" \
+        "8" "$(translate 'Linux guest setup and migration notes')" \
+        "0" "$(translate 'Back')") || break
+    case "$option" in
+        1) usbip_status ;;
+        2) usbip_dependencies server && usbip_action server-enable ;;
+        3) usbip_add_export ;;
+        4)
+            selected=$(usbip_select_saved exports) || continue
+            usbip_dialog --yesno "$(translate 'Stop sharing this device? An attached remote client will disconnect.')\n\n$selected" 10 80 && usbip_action remove-export "$selected"
+            ;;
+        5)
+            usbip_dialog --yesno "$(translate 'Disable the server and release all USB devices managed by ProxMenux? Saved selections are kept for the next enable.')" 11 80 && usbip_action server-disable
+            ;;
+        6) usbip_add_import ;;
+        7)
+            selected=$(usbip_select_saved imports) || continue
+            usbip_dialog --yesno "$(translate 'Remove this connection and detach its USB device?')\n\n$selected" 10 75 && usbip_action remove-import "$selected"
+            ;;
+        8)
+            usbip_dialog --msgbox "$(translate 'Run the client inside the Linux VM that needs the USB device. Install usbip, openssh-client and kmod, then copy usbip_manager.py to the same path as on the server. Configure key authentication and verify the server host key before adding a connection. The client uses the guest kernel and can reconnect after migration while the physical USB device stays on the server. Unprivileged LXC and Home Assistant OS need their own kernel/device or add-on setup. A physical device supports one attached client at a time.')" 19 95
+            ;;
+        0) break ;;
+    esac
+done

scripts/menus/hw_grafics_menu.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 # ==========================================================
-# ProxMenux - GPU and TPU Menu
+# ProxMenux - GPU, TPU and USB/IP Menu
 # ==========================================================
 # Author      : MacRimi
 # Copyright   : (c) 2024 MacRimi
@@ -26,7 +26,7 @@ initialize_cache
 
 while true; do
     OPTION=$(dialog --colors --backtitle "ProxMenux" \
-        --title "$(translate "GPUs and Coral-TPU Menu")" \
+        --title "$(translate "GPUs, Coral-TPU and USB/IP Menu")" \
         --menu "\n$(translate "Select an option:")" 26 80 16 \
             ""  "\Zb\Z4                         HOST\Zn" \
             "1"         "$(translate "Install/Update NVIDIA Drivers (Host + LXC)")" \
@@ -43,6 +43,7 @@ while true; do
             ""  "" \
             ""  "\Z4───────────────────── $(translate "Utilities") ────────────────────────\Zn" \
             "7"         "$(translate "Manual CLI Guide (GPU/TPU)")" \
+            "8"         "$(translate "USB/IP over SSH")" \
             "0"         "$(translate "Return to Main Menu")" \
             2>&1 >/dev/tty
     ) || { exec bash "$LOCAL_SCRIPTS/menus/main_menu.sh"; }
@@ -69,6 +70,9 @@ while true; do
         7)
             bash "$LOCAL_SCRIPTS/gpu_tpu/gpu-tpu-manual-guide.sh"
             ;;
+        8)
+            bash "$LOCAL_SCRIPTS/devices/usbip_menu.sh"
+            ;;
         0)
             exec bash "$LOCAL_SCRIPTS/menus/main_menu.sh"
             ;;

scripts/menus/main_menu.sh

@@ -68,7 +68,7 @@ show_menu() {
             search "$(translate "Search options and applications")" \
             "" "" \
             1 "$(translate "Settings post-install Proxmox")" \
-            2 "$(translate "Hardware: GPUs and Coral-TPU")" \
+            2 "$(translate "Hardware: GPUs, Coral-TPU and USB/IP")" \
             3 "$(translate "Create VM from template or script")" \
             4 "$(translate "Disk Manager")" \
             5 "$(translate "Storage & Share Manager")" \

scripts/menus/search_index.json

@@ -1,4 +1,10 @@
 [
+  {
+    "title": "USB/IP over SSH",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
+    "script": "devices/usbip_menu.sh",
+    "keywords": "usbip usb ip ssh remote zigbee bluetooth printer impresora remoto"
+  },
   {
     "title": "Settings post-install Proxmox",
     "menu": "Settings post-install Proxmox",
@@ -6,8 +12,8 @@
     "keywords": ""
   },
   {
-    "title": "Hardware: GPUs and Coral-TPU",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "title": "Hardware: GPUs, Coral-TPU and USB/IP",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "menus/hw_grafics_menu.sh",
     "keywords": ""
   },
@@ -98,43 +104,43 @@
   },
   {
     "title": "Install/Update NVIDIA Drivers (Host + LXC)",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/nvidia_installer.sh",
     "keywords": "gpu nvidia driver"
   },
   {
     "title": "Install/Update Coral TPU on Host",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/install_coral.sh",
     "keywords": "coral tpu usb"
   },
   {
     "title": "Add GPU to LXC   (Intel | AMD | NVIDIA)",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/add_gpu_lxc.sh",
     "keywords": "gpu passthrough lxc igpu"
   },
   {
     "title": "Add Coral TPU to LXC",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/install_coral_lxc.sh",
     "keywords": "coral tpu lxc usb"
   },
   {
     "title": "Add GPU to VM    (Intel | AMD | NVIDIA)",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/add_gpu_vm.sh",
     "keywords": "gpu passthrough vm"
   },
   {
     "title": "Switch GPU Mode  (VM <-> LXC)",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/switch_gpu_mode.sh",
     "keywords": "gpu mode modo"
   },
   {
     "title": "Manual CLI Guide (GPU/TPU)",
-    "menu": "Hardware: GPUs and Coral-TPU",
+    "menu": "Hardware: GPUs, Coral-TPU and USB/IP",
     "script": "gpu_tpu/gpu-tpu-manual-guide.sh",
     "keywords": "gpu tpu guide guia"
   },