diff --git a/nix/nixos-module.nix b/nix/nixos-module.nix index 9938ffe..e8fc6ea 100644 --- a/nix/nixos-module.nix +++ b/nix/nixos-module.nix @@ -94,6 +94,9 @@ in ); Restart = "on-failure"; RestartSec = "5s"; + StateDirectory = "vhotplug"; + StateDirectoryMode = "0700"; + UMask = "0077"; }; }; }; diff --git a/nix/nixos-test.nix b/nix/nixos-test.nix index c22772d..f238b4d 100644 --- a/nix/nixos-test.nix +++ b/nix/nixos-test.nix @@ -48,6 +48,8 @@ pkgs.testers.runNixOSTest { machine.start() machine.wait_for_unit("vhotplug.service") machine.succeed("systemctl status vhotplug.service") + machine.succeed("test $(stat -c %a /var/lib/vhotplug) = 700") + machine.succeed("test $(systemctl show -P UMask vhotplug.service) = 0077") # Verify vhotplug is waiting for devices machine.wait_until_succeeds("journalctl -u vhotplug.service | grep -q 'Waiting for new devices'") diff --git a/nix/package.nix b/nix/package.nix index 64eaf0c..6361255 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -22,7 +22,9 @@ python3Packages.buildPythonApplication { qemu-qmp ]; - doCheck = false; + nativeCheckInputs = with python3Packages; [ + pytestCheckHook + ]; meta = with lib; { description = "Hot-plugging USB and PCI devices to virtual machines"; diff --git a/tests/test_crosvmlink.py b/tests/test_crosvmlink.py new file mode 100644 index 0000000..9713e02 --- /dev/null +++ b/tests/test_crosvmlink.py @@ -0,0 +1,138 @@ +import asyncio +from typing import Any + +import pytest + +from vhotplug.crosvmlink import CrosvmLink, CrosvmVMUnavailableError +from vhotplug.usb import USBInfo + + +class FakeProcess: + def __init__(self, stdout: str, returncode: int = 0) -> None: + self.stdout = stdout + self.returncode = returncode + + async def communicate(self) -> tuple[bytes, bytes]: + return self.stdout.encode(), b"" + + +def fake_crosvm( + monkeypatch: pytest.MonkeyPatch, + list_output: str | list[str] = "devices", + attach_output: str = "ok 2", + detach_output: str | None = None, + list_returncode: int = 0, +) -> list[tuple[str, ...]]: + calls: list[tuple[str, ...]] = [] + list_outputs = [list_output] if isinstance(list_output, str) else list(list_output) + + async def execute(*args: str, **_kwargs: Any) -> FakeProcess: + calls.append(args) + if args[1:3] == ("usb", "list"): + output = list_outputs.pop(0) if len(list_outputs) > 1 else list_outputs[0] + return FakeProcess(output, list_returncode) + if args[1:3] == ("usb", "attach"): + return FakeProcess(attach_output) + if args[1:3] == ("usb", "detach"): + return FakeProcess(detach_output if detach_output is not None else f"ok {args[3]}") + raise AssertionError(f"Unexpected Crosvm command: {args}") + + monkeypatch.setattr(asyncio, "create_subprocess_exec", execute) + return calls + + +def usb() -> USBInfo: + return USBInfo(device_node="/dev/bus/usb/001/002", vid="046d", pid="c52b") + + +def link(monkeypatch: pytest.MonkeyPatch) -> CrosvmLink: + result = CrosvmLink("/run/app-vm.sock", "/nix/store/crosvm/bin/crosvm") + monkeypatch.setattr(result, "_wait_for_boot", lambda: True) + return result + + +def test_adds_identical_device_when_port_is_not_known(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 1 046d c52b") + + port = asyncio.run(link(monkeypatch).add_usb_device(usb())) + + assert port == 2 + assert any(command[1:3] == ("usb", "attach") for command in calls) + + +def test_reuses_persisted_port(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 2 046d c52b") + + port = asyncio.run(link(monkeypatch).add_usb_device(usb(), known_port=2)) + + assert port == 2 + assert not any(command[1:3] == ("usb", "attach") for command in calls) + + +def test_removes_exact_port_among_identical_devices(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 1 046d c52b 2 046d c52b") + + asyncio.run(link(monkeypatch).remove_usb_device(usb(), known_port=2)) + + detach_commands = [command for command in calls if command[1:3] == ("usb", "detach")] + assert detach_commands == [("/nix/store/crosvm/bin/crosvm", "usb", "detach", "2", "/run/app-vm.sock")] + + +def test_refuses_to_remove_different_device_from_known_port(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 2 1050 0407") + + with pytest.raises(RuntimeError, match="different device"): + asyncio.run(link(monkeypatch).remove_usb_device(usb(), known_port=2)) + + assert not any(command[1:3] == ("usb", "detach") for command in calls) + + +def test_refuses_ambiguous_legacy_removal(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 1 046d c52b 2 046d c52b") + + with pytest.raises(RuntimeError, match="cannot be identified safely"): + asyncio.run(link(monkeypatch).remove_usb_device(usb())) + + assert not any(command[1:3] == ("usb", "detach") for command in calls) + + +def test_malformed_ok_adopts_new_port_without_retry(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, ["devices 1 1234 5678", "devices 1 1234 5678 2 046d c52b"], "ok") + + port = asyncio.run(link(monkeypatch).add_usb_device(usb())) + + assert port == 2 + assert len([command for command in calls if command[1:3] == ("usb", "attach")]) == 1 + + +def test_malformed_list_port_raises_runtime_error(monkeypatch: pytest.MonkeyPatch) -> None: + fake_crosvm(monkeypatch, "devices invalid 046d c52b") + + with pytest.raises(RuntimeError, match="invalid literal"): + asyncio.run(link(monkeypatch).remove_usb_device(usb(), known_port=2)) + + +def test_empty_detach_response_raises_runtime_error(monkeypatch: pytest.MonkeyPatch) -> None: + fake_crosvm(monkeypatch, "devices 2 046d c52b", detach_output="") + + with pytest.raises(RuntimeError, match="empty response"): + asyncio.run(link(monkeypatch).remove_usb_device(usb(), known_port=2)) + + +def test_unavailable_vm_has_distinct_error(monkeypatch: pytest.MonkeyPatch) -> None: + fake_crosvm(monkeypatch, list_returncode=1) + monkeypatch.setattr("vhotplug.crosvmlink.is_unix_socket_alive", lambda *_args: False) + + with pytest.raises(CrosvmVMUnavailableError): + asyncio.run(link(monkeypatch).remove_usb_device(usb(), known_port=2)) + + +def test_no_available_port_does_not_detach_other_devices(monkeypatch: pytest.MonkeyPatch) -> None: + calls = fake_crosvm(monkeypatch, "devices 1 1234 5678", "no_available_port") + crosvm = link(monkeypatch) + crosvm.vm_retry_count = 0 + + with pytest.raises(RuntimeError, match="timed out"): + asyncio.run(crosvm.add_usb_device(usb())) + + assert not any(command[1:3] == ("usb", "detach") for command in calls) diff --git a/tests/test_device.py b/tests/test_device.py new file mode 100644 index 0000000..1b91b19 --- /dev/null +++ b/tests/test_device.py @@ -0,0 +1,135 @@ +import asyncio +from types import SimpleNamespace +from typing import Any + +import pytest + +from vhotplug.crosvmlink import CrosvmVMUnavailableError +from vhotplug.device import _attach_device_to_vm, _remove_device_from_vm +from vhotplug.usb import USBInfo + + +class FakeState: + def __init__(self, current_vm: str | None = "app-vm") -> None: + self.current_vm = current_vm + self.removed = False + self.attached_port: int | None = None + + def get_vm_for_device(self, _dev_info: USBInfo) -> str | None: + return self.current_vm + + def get_crosvm_usb_port(self, _dev_info: USBInfo, _vm_name: str, _socket_path: str) -> int: + return 7 + + def remove_vm_for_device(self, _dev_info: USBInfo) -> None: + self.removed = True + + def set_vm_for_device(self, _dev_info: USBInfo, vm_name: str) -> None: + self.current_vm = vm_name + + def set_crosvm_usb_port( + self, + _dev_info: USBInfo, + _vm_name: str, + _socket_path: str, + port: int | None, + ) -> None: + self.attached_port = port + + def clear_disconnected(self, _dev_info: USBInfo) -> bool: + return False + + +def usb() -> USBInfo: + return USBInfo(device_node="/dev/bus/usb/001/002", sys_name="1-2.1") + + +def app_context(state: FakeState) -> Any: + config = SimpleNamespace(usb_authorization_enabled=lambda: False) + return SimpleNamespace(dev_state=state, config=config, api_server=None) + + +def test_vm_unavailable_still_clears_local_state(monkeypatch: pytest.MonkeyPatch) -> None: + state = FakeState() + + async def unavailable(*_args: Any) -> None: + raise CrosvmVMUnavailableError("VM unavailable") + + monkeypatch.setattr("vhotplug.device.vmm_remove_device", unavailable) + + asyncio.run( + _remove_device_from_vm( + app_context(state), + usb(), + {"name": "app-vm", "type": "crosvm", "socket": "/run/app-vm.sock"}, + ) + ) + + assert state.removed + + +def test_detach_refusal_preserves_local_state(monkeypatch: pytest.MonkeyPatch) -> None: + state = FakeState() + + async def refuse(*_args: Any) -> None: + raise RuntimeError("refusing to detach") + + monkeypatch.setattr("vhotplug.device.vmm_remove_device", refuse) + + with pytest.raises(RuntimeError, match="refusing"): + asyncio.run( + _remove_device_from_vm( + app_context(state), + usb(), + {"name": "app-vm", "type": "crosvm", "socket": "/run/app-vm.sock"}, + ) + ) + + assert not state.removed + + +def test_move_does_not_attach_when_old_vm_removal_fails(monkeypatch: pytest.MonkeyPatch) -> None: + state = FakeState(current_vm="old-vm") + attached = False + + async def remove(*_args: Any) -> None: + raise RuntimeError("old VM detach failed") + + async def attach(*_args: Any) -> None: + nonlocal attached + attached = True + + monkeypatch.setattr("vhotplug.device.remove_device", remove) + monkeypatch.setattr("vhotplug.device.vmm_add_device", attach) + + with pytest.raises(RuntimeError, match="old VM detach failed"): + asyncio.run( + _attach_device_to_vm( + app_context(state), + usb(), + {"name": "new-vm", "type": "crosvm", "socket": "/run/new-vm.sock"}, + ) + ) + + assert not attached + + +def test_crosvm_attach_passes_and_records_guest_port(monkeypatch: pytest.MonkeyPatch) -> None: + state = FakeState(current_vm=None) + + async def attach(*_args: Any) -> int: + assert _args[-1] == 7 + return 9 + + monkeypatch.setattr("vhotplug.device.vmm_add_device", attach) + + asyncio.run( + _attach_device_to_vm( + app_context(state), + usb(), + {"name": "app-vm", "type": "crosvm", "socket": "/run/app-vm.sock"}, + ) + ) + + assert state.current_vm == "app-vm" + assert state.attached_port == 9 diff --git a/tests/test_devicestate.py b/tests/test_devicestate.py new file mode 100644 index 0000000..a2896d8 --- /dev/null +++ b/tests/test_devicestate.py @@ -0,0 +1,117 @@ +import json +import os +from pathlib import Path + +import pytest + +from vhotplug.devicestate import DeviceState +from vhotplug.usb import USBInfo + + +def usb(device_node: str = "/dev/bus/usb/001/002", vid: str = "046d") -> USBInfo: + return USBInfo(device_node=device_node, vid=vid, pid="c52b", serial="serial", sys_name="1-2.1") + + +def test_crosvm_usb_port_is_scoped_to_vm_socket_generation(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + socket_path = tmp_path / "app-vm.sock" + socket_path.touch() + state = DeviceState(True, str(state_path)) + state.set_crosvm_usb_port(usb(), "app-vm", str(socket_path), 7) + + reloaded = DeviceState(True, str(state_path)) + assert reloaded.get_crosvm_usb_port(usb(), "app-vm", str(socket_path)) == 7 + + socket_path.unlink() + socket_path.touch() + assert reloaded.get_crosvm_usb_port(usb(), "app-vm", str(socket_path)) is None + + +def test_crosvm_usb_port_uses_topology_and_validates_identity(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + socket_path = tmp_path / "app-vm.sock" + socket_path.touch() + state = DeviceState(True, str(state_path)) + state.set_crosvm_usb_port(usb(), "app-vm", str(socket_path), 7) + + renumbered = usb("/dev/bus/usb/001/009") + assert state.get_crosvm_usb_port(renumbered, "app-vm", str(socket_path)) == 7 + assert state.get_crosvm_usb_port(usb(vid="1050"), "app-vm", str(socket_path)) is None + + +def test_crosvm_usb_port_is_removed_with_device_state(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + socket_path = tmp_path / "app-vm.sock" + socket_path.touch() + state = DeviceState(True, str(state_path)) + state.set_crosvm_usb_port(usb(), "app-vm", str(socket_path), 7) + + state.remove_vm_for_device(usb()) + + assert state.get_crosvm_usb_port(usb(), "app-vm", str(socket_path)) is None + + +def test_crosvm_usb_port_is_scoped_to_vm(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + socket_path = tmp_path / "app-vm.sock" + socket_path.touch() + state = DeviceState(True, str(state_path)) + state.set_crosvm_usb_port(usb(), "app-vm", str(socket_path), 7) + + assert state.get_crosvm_usb_port(usb(), "other-vm", str(socket_path)) is None + + +def test_invalid_state_starts_empty(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + state_path.write_text('{"selected_vms":', encoding="utf-8") + + state = DeviceState(True, str(state_path)) + + assert state.selected_vms == {} + assert state.crosvm_usb_port_map == {} + + +def test_boolean_port_is_discarded(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + state_path.write_text( + json.dumps( + { + "crosvm_usb_ports": { + "1-2.1": { + "vm": "app-vm", + "port": True, + "socket_generation": "1:2:3", + "vid": "046d", + "pid": "c52b", + "serial": None, + } + } + } + ), + encoding="utf-8", + ) + + assert DeviceState(True, str(state_path)).crosvm_usb_port_map == {} + + +def test_state_file_is_private(tmp_path: Path) -> None: + state_path = tmp_path / "vhotplug.state" + state = DeviceState(True, str(state_path)) + state.select_vm_for_device(usb(), "app-vm") + + assert os.stat(state_path).st_mode & 0o777 == 0o600 + + +def test_state_save_failure_does_not_replace_valid_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + state_path = tmp_path / "vhotplug.state" + state = DeviceState(True, str(state_path)) + state.select_vm_for_device(usb(), "app-vm") + saved = state_path.read_text(encoding="utf-8") + + def fail_replace(_self: Path, _target: Path) -> None: + raise OSError("read-only filesystem") + + monkeypatch.setattr(Path, "replace", fail_replace) + state.set_disconnected(usb()) + + assert state_path.read_text(encoding="utf-8") == saved diff --git a/tests/test_vmm.py b/tests/test_vmm.py index 557cfed..65e2ad5 100644 --- a/tests/test_vmm.py +++ b/tests/test_vmm.py @@ -1,7 +1,12 @@ +import asyncio +from types import SimpleNamespace +from typing import Any + import pytest from vhotplug.pci import PCIInfo, pci_is_nvidia_gpu -from vhotplug.vmm import vmm_args_ovmf, vmm_args_pci +from vhotplug.usb import USBInfo +from vhotplug.vmm import vmm_add_device, vmm_args_ovmf, vmm_args_pci def test_pci_is_nvidia_gpu_display_controller() -> None: @@ -47,3 +52,49 @@ def test_vmm_args_pci_uses_default_bus() -> None: "-device", "vfio-pci,host=0000:01:00.0,multifunction=on,id=vhp-pci-3", ] + + +def test_vmm_add_device_returns_crosvm_usb_port(monkeypatch: pytest.MonkeyPatch) -> None: + class FakeCrosvm: + def __init__(self, *_args: Any) -> None: + pass + + async def add_usb_device(self, _usb_info: USBInfo, known_port: int | None) -> int: + assert known_port == 3 + return 7 + + monkeypatch.setattr("vhotplug.vmm.CrosvmLink", FakeCrosvm) + app_context: Any = SimpleNamespace(config=SimpleNamespace(config={})) + + port = asyncio.run( + vmm_add_device( + app_context, + {"name": "app-vm", "type": "crosvm", "socket": "/run/app-vm.sock"}, + USBInfo(device_node="/dev/bus/usb/001/002"), + 3, + ) + ) + + assert port == 7 + + +def test_vmm_add_device_returns_none_for_qemu_usb(monkeypatch: pytest.MonkeyPatch) -> None: + class FakeQemu: + def __init__(self, *_args: Any) -> None: + pass + + async def add_usb_device(self, _usb_info: USBInfo) -> None: + pass + + monkeypatch.setattr("vhotplug.vmm.QEMULink", FakeQemu) + app_context: Any = SimpleNamespace(config=SimpleNamespace(config={})) + + port = asyncio.run( + vmm_add_device( + app_context, + {"name": "app-vm", "type": "qemu", "socket": "/run/app-vm.sock"}, + USBInfo(device_node="/dev/bus/usb/001/002"), + ) + ) + + assert port is None diff --git a/vhotplug/crosvmlink.py b/vhotplug/crosvmlink.py index 5598ad6..5748974 100644 --- a/vhotplug/crosvmlink.py +++ b/vhotplug/crosvmlink.py @@ -1,19 +1,29 @@ import asyncio import logging import socket +from typing import ClassVar -from vhotplug.misc import wait_for_unix_socket +from vhotplug.misc import is_unix_socket_alive, wait_for_unix_socket from vhotplug.pci import PCIInfo from vhotplug.usb import USBInfo logger = logging.getLogger("vhotplug") +class CrosvmVMUnavailableError(RuntimeError): + """Raised when a Crosvm control socket is unavailable.""" + + +class CrosvmUSBAttachStateError(RuntimeError): + """Raised when Crosvm attached a USB device without identifying its port.""" + + class CrosvmLink: vm_retry_count = 5 vm_retry_timeout = 1 vm_wait_after_boot = 3 vm_boot_timeout = 10 + usb_locks: ClassVar[dict[str, asyncio.Lock]] = {} def __init__(self, socket_path: str, crosvm_bin: str | None) -> None: self.socket_path = socket_path @@ -28,7 +38,11 @@ def _wait_for_boot(self) -> bool: self.socket_path, self.vm_boot_timeout, self.vm_wait_after_boot, socket.SOCK_SEQPACKET ) - async def add_usb_device(self, usb_info: USBInfo) -> None: + async def add_usb_device(self, usb_info: USBInfo, known_port: int | None = None) -> int: + async with self.usb_locks.setdefault(self.socket_path, asyncio.Lock()): + return await self._add_usb_device(usb_info, known_port) + + async def _add_usb_device(self, usb_info: USBInfo, known_port: int | None = None) -> int: dev_node = usb_info.device_node assert dev_node is not None, "Device node must be set" @@ -36,22 +50,24 @@ async def add_usb_device(self, usb_info: USBInfo) -> None: if not self._wait_for_boot(): logger.warning("VM is not booted while adding device %s", dev_node) - i = 0 - while True: + last_error: Exception | None = None + for attempt in range(self.vm_retry_count + 1): try: logger.info("Adding USB device %s to %s", dev_node, self.socket_path) - # Check if the device is already connected + # Reuse a persisted port after a daemon restart if it still + # contains the expected device. VID/PID alone is not a unique + # identity because multiple identical USB devices may exist. devices = await self.usb_list() - for _, vid, pid in devices: - if vid == usb_info.vid and pid == usb_info.pid: + for port, vid, pid in devices: + if port == known_port and vid == usb_info.vid and pid == usb_info.pid: logger.info( - "Device %s:%s is already attached to %s, skipping", - vid, - pid, + "Device %s is already attached to %s on port %s, skipping", + dev_node, self.socket_path, + port, ) - return + return port proc = await asyncio.create_subprocess_exec( self.crosvm_bin, @@ -70,6 +86,7 @@ async def add_usb_device(self, usb_info: USBInfo) -> None: stderr_str = stderr_bytes.decode() if proc.returncode != 0: + last_error = RuntimeError(f"Crosvm USB attach failed with code {proc.returncode}") logger.warning( "Failed to add device %s, error code: %s", dev_node, @@ -78,38 +95,79 @@ async def add_usb_device(self, usb_info: USBInfo) -> None: logger.warning("Out: %s", stdout_str) logger.warning("Err: %s", stderr_str) else: - r = stdout_str.split() + r = self._parse_usb_attach_response(stdout_str) if r[0] == "ok": - logger.info("Attached USB device %s, id: %s", dev_node, r[1]) - return + if len(r) == 2: + try: + port = self._parse_usb_port(r[1]) + except ValueError: + logger.warning("Crosvm returned an invalid USB port: %s", r[1]) + else: + logger.info("Attached USB device %s, id: %s", dev_node, port) + return port + port = await self._recover_attached_usb_port(usb_info, devices) + logger.info("Attached USB device %s, id: %s", dev_node, port) + return port if r[0] == "no_available_port": - # Crosvm supports attaching USB devices only after the kernel has booted - # Here, we may attempt to attach a device before that which will return no_available_port - # If we keep trying, it may eventually return I/O error and USB passthrough won't work until the VM is rebooted - # As a workaround we remove USB devices here even if it returns no_such_device - # This helps prevent I/O errors and allows USB to be successfully attached once the VM boots - logger.info("No available port, removing all devices") - devices = await self.usb_list() - try: - for index, _, _ in devices: - await self.remove_usb_device_by_id(index) - except RuntimeError as e: - logger.warning("Failed to remove: %s", str(e)) + # This can be transient while the guest xHCI driver is + # starting, or permanent when every port is occupied. + # Never detach unrelated devices to make room. + logger.info("No Crosvm USB port is available yet") else: + last_error = RuntimeError("Unexpected Crosvm USB attach response") logger.warning("Unexpected result: %s", r[0]) logger.warning("Out: %s", stdout_str) logger.warning("Err: %s", stderr_str) - except OSError as e: + except CrosvmUSBAttachStateError: + raise + except (OSError, RuntimeError, ValueError) as e: + last_error = e logger.warning("Failed to attach USB device %s: %s", dev_node, e) - if i < self.vm_retry_count: + if attempt < self.vm_retry_count: logger.info("Retrying") await asyncio.sleep(self.vm_retry_timeout) - i += 1 - else: - break - logger.error("Failed to add USB device %s after %s attempts", dev_node, i) - raise RuntimeError("Timeout") + logger.error("Failed to add USB device %s after %s attempts", dev_node, self.vm_retry_count + 1) + raise RuntimeError("Crosvm USB attach timed out") from last_error + + @staticmethod + def _parse_usb_port(value: str) -> int: + port = int(value) + if not 0 <= port <= 255: + raise ValueError(f"Invalid Crosvm USB port: {port}") + return port + + @staticmethod + def _parse_usb_attach_response(stdout: str) -> list[str]: + result = stdout.split() + if not result: + raise RuntimeError("Crosvm returned an empty USB attach response") + return result + + @staticmethod + def _single_new_usb_port(matches: list[int]) -> int: + if len(matches) != 1: + raise CrosvmUSBAttachStateError("Crosvm attached the USB device but did not report its port") + return matches[0] + + async def _recover_attached_usb_port( + self, + usb_info: USBInfo, + devices_before: list[tuple[int, str, str]], + ) -> int: + try: + devices_after = await self.usb_list() + except RuntimeError as e: + raise CrosvmUSBAttachStateError( + "Crosvm attached the USB device but its port could not be determined" + ) from e + old_ports = {port for port, _, _ in devices_before} + matches = [ + port + for port, vid, pid in devices_after + if port not in old_ports and vid == usb_info.vid and pid == usb_info.pid + ] + return self._single_new_usb_port(matches) async def remove_usb_device_by_id(self, dev_id: int) -> None: try: @@ -135,11 +193,12 @@ async def remove_usb_device_by_id(self, dev_id: int) -> None: logger.error("Err: %s", stderr_str) raise RuntimeError(proc.returncode) r = stdout_str.split() - if r[0] != "ok": - logger.error("Unexpected result: %s", r[0]) + if not r or r[0] != "ok": + result = r[0] if r else "empty response" + logger.error("Unexpected result: %s", result) logger.error("Out: %s", stdout_str) logger.error("Err: %s", stderr_str) - raise RuntimeError(r[0]) + raise RuntimeError(result) logger.info("Detached USB device %s", dev_id) return except OSError as e: @@ -165,34 +224,66 @@ async def usb_list(self) -> list[tuple[int, str, str]]: stderr_str = stderr_bytes.decode() if proc.returncode != 0: - logger.error("Failed to get USB list, error code: %s", proc.returncode) - logger.error("Out: %s", stdout_str) - logger.error("Err: %s", stderr_str) - else: - r = stdout_str.split() - if r[0] != "devices": - logger.error("Unexpected result: %s", r[0]) - logger.error("Out: %s", stdout_str) - logger.error("Err: %s", stderr_str) - else: - data = r[1:] - for i in range(0, len(data), 3): - index = int(data[i]) - vid = data[i + 1] - pid = data[i + 2] - devices.append((index, vid, pid)) - logger.debug("USB device %s: %s:%s", index, vid, pid) - - except OSError: + logger.error("Crosvm USB list failed with code %s: %s", proc.returncode, stderr_str.strip()) + if not is_unix_socket_alive(self.socket_path, socket.SOCK_SEQPACKET): + raise CrosvmVMUnavailableError("Crosvm VM is unavailable") + raise RuntimeError(f"Crosvm USB list failed with code {proc.returncode}") + + result = stdout_str.split() + if not result or result[0] != "devices" or len(result[1:]) % 3 != 0: + logger.error("Malformed Crosvm USB list response: %s", stdout_str.strip()) + raise RuntimeError("Malformed Crosvm USB list response") + + data = result[1:] + for i in range(0, len(data), 3): + index = self._parse_usb_port(data[i]) + vid = data[i + 1] + pid = data[i + 2] + devices.append((index, vid, pid)) + logger.debug("USB device %s: %s:%s", index, vid, pid) + + except (OSError, ValueError) as e: logger.exception("Failed to list USB devices") + raise RuntimeError(e) from None return devices - async def remove_usb_device(self, usb_info: USBInfo) -> None: + async def remove_usb_device(self, usb_info: USBInfo, known_port: int | None = None) -> None: + async with self.usb_locks.setdefault(self.socket_path, asyncio.Lock()): + await self._remove_usb_device(usb_info, known_port) + + async def _remove_usb_device(self, usb_info: USBInfo, known_port: int | None = None) -> None: devices = await self.usb_list() - for index, crosvm_vid, crosvm_pid in devices: - if usb_info.vid == crosvm_vid and usb_info.pid == crosvm_pid: - logger.debug("Removing %s from %s", index, self.socket_path) - await self.remove_usb_device_by_id(index) + if known_port is not None: + device = next((dev for dev in devices if dev[0] == known_port), None) + if device is None: + logger.debug("USB port %s is already empty", known_port) + return + + _, crosvm_vid, crosvm_pid = device + if usb_info.vid and usb_info.pid and (usb_info.vid != crosvm_vid or usb_info.pid != crosvm_pid): + logger.error( + "USB port %s now contains %s:%s instead of %s:%s; not detaching it", + known_port, + crosvm_vid, + crosvm_pid, + usb_info.vid, + usb_info.pid, + ) + raise RuntimeError("Crosvm USB port contains a different device; refusing to detach") + + await self.remove_usb_device_by_id(known_port) + return + + matches = [ + index + for index, crosvm_vid, crosvm_pid in devices + if usb_info.vid == crosvm_vid and usb_info.pid == crosvm_pid + ] + if len(matches) > 1: + logger.error("Multiple Crosvm USB devices match %s:%s", usb_info.vid, usb_info.pid) + raise RuntimeError("Crosvm USB device cannot be identified safely") + if matches: + await self.remove_usb_device_by_id(matches[0]) async def add_pci_device(self, _pci_info: PCIInfo) -> None: raise RuntimeError("Not implemented") diff --git a/vhotplug/device.py b/vhotplug/device.py index 38400d6..01f494c 100644 --- a/vhotplug/device.py +++ b/vhotplug/device.py @@ -9,6 +9,7 @@ from vhotplug.appcontext import AppContext from vhotplug.config import PassthroughInfo +from vhotplug.crosvmlink import CrosvmVMUnavailableError from vhotplug.evdev import EvdevInfo, evdev_test_grab, get_evdev_info, is_input_device from vhotplug.pci import ( PCIInfo, @@ -237,10 +238,7 @@ async def _attach_device_to_vm(app_context: AppContext, dev_info: USBInfo | PCII current_vm_name = app_context.dev_state.get_vm_for_device(dev_info) if current_vm_name and current_vm_name != vm_name: logger.warning("Device is attached to %s, removing...", current_vm_name) - try: - await remove_device(app_context, dev_info) - except RuntimeError as e: - logger.warning("Failed to remove: %s", e) + await remove_device(app_context, dev_info) # Setup VFIO for all PCI devices in the IOMMU group if needed if isinstance(dev_info, PCIInfo): @@ -257,7 +255,13 @@ async def _attach_device_to_vm(app_context: AppContext, dev_info: USBInfo | PCII # Attach device to the VM try: - await vmm_add_device(app_context, vm, dev_info) + vm_socket = vm.get("socket", "") + crosvm_usb_port = ( + app_context.dev_state.get_crosvm_usb_port(dev_info, vm_name, vm_socket) + if isinstance(dev_info, USBInfo) and vm.get("type") == "crosvm" + else None + ) + attached_port = await vmm_add_device(app_context, vm, dev_info, crosvm_usb_port) except Exception: if isinstance(dev_info, USBInfo) and app_context.config.usb_authorization_enabled(): logger.info("Deauthorizing %s", dev_info.friendly_name()) @@ -266,6 +270,8 @@ async def _attach_device_to_vm(app_context: AppContext, dev_info: USBInfo | PCII # Add selected VM to the state database app_context.dev_state.set_vm_for_device(dev_info, vm_name) + if isinstance(dev_info, USBInfo) and vm.get("type") == "crosvm": + app_context.dev_state.set_crosvm_usb_port(dev_info, vm_name, vm_socket, attached_port) app_context.dev_state.clear_disconnected(dev_info) if app_context.api_server: @@ -332,7 +338,17 @@ async def _remove_device_from_vm( ) -> None: """Removes device from VM, saves its state and sends a notification.""" # Remove from VM - await vmm_remove_device(app_context, vm, dev_info) + vm_name = vm.get("name", "") + vm_socket = vm.get("socket", "") + crosvm_usb_port = ( + app_context.dev_state.get_crosvm_usb_port(dev_info, vm_name, vm_socket) + if isinstance(dev_info, USBInfo) and vm.get("type") == "crosvm" + else None + ) + try: + await vmm_remove_device(app_context, vm, dev_info, crosvm_usb_port) + except CrosvmVMUnavailableError as e: + logger.warning("VM is unavailable while removing %s: %s", dev_info.friendly_name(), e) if wait: # Wait until the device is no longer in the list diff --git a/vhotplug/devicestate.py b/vhotplug/devicestate.py index a9c1dc7..9fbf001 100644 --- a/vhotplug/devicestate.py +++ b/vhotplug/devicestate.py @@ -1,7 +1,10 @@ +import contextlib import json import logging +import os import threading from pathlib import Path +from typing import Any from vhotplug.pci import PCIInfo from vhotplug.usb import USBInfo @@ -17,6 +20,10 @@ def __init__(self, persistent: bool = False, db_path: str | None = None) -> None # Runtime map of USB device_node - VM, used to know from which VM to disconnect self.usb_device_vm_map: dict[str, str] = {} + # Persistent map of USB topology - Crosvm guest binding. A guest port + # is valid only for one VM control socket generation. + self.crosvm_usb_port_map: dict[str, dict[str, Any]] = {} + # Runtime map of PCI address - VM, used to know from which VM to disconnect self.pci_device_vm_map: dict[str, str] = {} @@ -37,20 +44,80 @@ def _load(self) -> None: if self.persistent and self.db_path.exists(): try: with self.db_path.open("r", encoding="utf-8") as f: - j = json.load(f) + j = self._state_object(json.load(f)) self.selected_vms = j.get("selected_vms", {}) self.disconnected_devices = set(j.get("disconnected_devices", [])) - except OSError as e: - logger.warning("Failed to load USB state database: %s", e) + ports = j.get("crosvm_usb_ports", {}) + if isinstance(ports, dict): + for key, value in ports.items(): + if self._valid_crosvm_usb_binding(key, value): + self.crosvm_usb_port_map[key] = value + else: + logger.warning("Discarding invalid Crosvm USB binding for %s", key) + except (OSError, TypeError, ValueError) as e: + logger.warning("Failed to load state database, starting from empty state: %s", e) def _save(self) -> None: if self.persistent: - with self.db_path.open("w", encoding="utf-8") as f: - j = { - "selected_vms": self.selected_vms, - "disconnected_devices": list(self.disconnected_devices), - } - json.dump(j, f, ensure_ascii=False, indent=2) + tmp_path = self.db_path.with_name(f".{self.db_path.name}.tmp") + try: + with tmp_path.open("w", encoding="utf-8") as f: + os.chmod(tmp_path, 0o600) + j = { + "selected_vms": self.selected_vms, + "disconnected_devices": list(self.disconnected_devices), + "crosvm_usb_ports": self.crosvm_usb_port_map, + } + json.dump(j, f, ensure_ascii=False, indent=2) + f.flush() + os.fsync(f.fileno()) + tmp_path.replace(self.db_path) + except OSError as e: + logger.warning("Failed to save state database: %s", e) + with contextlib.suppress(OSError): + tmp_path.unlink(missing_ok=True) + + @staticmethod + def _state_object(value: object) -> dict[str, Any]: + if not isinstance(value, dict): + raise TypeError("State database root must be an object") + return value + + @staticmethod + def _valid_port(port: object) -> bool: + return type(port) is int and 0 <= port <= 255 + + @classmethod + def _valid_crosvm_usb_binding(cls, key: object, value: object) -> bool: + if not isinstance(key, str) or not isinstance(value, dict): + return False + return ( + cls._valid_port(value.get("port")) + and isinstance(value.get("vm"), str) + and isinstance(value.get("socket_generation"), str) + and (value.get("vid") is None or isinstance(value.get("vid"), str)) + and (value.get("pid") is None or isinstance(value.get("pid"), str)) + and (value.get("serial") is None or isinstance(value.get("serial"), str)) + ) + + @staticmethod + def _usb_key(dev_info: USBInfo) -> str | None: + return dev_info.sys_name + + @staticmethod + def _socket_generation(socket_path: str) -> str | None: + try: + stat = os.stat(socket_path) + except OSError: + return None + return f"{stat.st_dev}:{stat.st_ino}:{stat.st_ctime_ns}" + + @staticmethod + def _binding_matches_device(binding: dict[str, Any], dev_info: USBInfo) -> bool: + return all( + actual is None or binding.get(field) == actual + for field, actual in (("vid", dev_info.vid), ("pid", dev_info.pid), ("serial", dev_info.serial)) + ) def set_vm_for_device(self, dev_info: USBInfo | PCIInfo, vm_name: str) -> None: with self.lock: @@ -74,6 +141,10 @@ def remove_vm_for_device(self, dev_info: USBInfo | PCIInfo) -> None: if isinstance(dev_info, USBInfo): if dev_info.device_node in self.usb_device_vm_map: del self.usb_device_vm_map[dev_info.device_node] + key = self._usb_key(dev_info) + if key in self.crosvm_usb_port_map: + del self.crosvm_usb_port_map[key] + self._save() elif dev_info.address in self.pci_device_vm_map: del self.pci_device_vm_map[dev_info.address] @@ -115,6 +186,69 @@ def list_usb_devices(self) -> dict[str, str]: with self.lock: return dict(self.usb_device_vm_map) + def set_crosvm_usb_port( + self, + dev_info: USBInfo, + vm_name: str, + socket_path: str, + port: int | None, + ) -> None: + with self.lock: + key = self._usb_key(dev_info) + if key is None: + return + if port is None: + if self.crosvm_usb_port_map.pop(key, None) is None: + return + self._save() + return + if not self._valid_port(port): + raise ValueError(f"Invalid Crosvm USB port: {port}") + socket_generation = self._socket_generation(socket_path) + if socket_generation is None: + logger.warning("Cannot persist Crosvm USB binding: socket %s is unavailable", socket_path) + return + binding = { + "vm": vm_name, + "port": port, + "socket_generation": socket_generation, + "vid": dev_info.vid, + "pid": dev_info.pid, + "serial": dev_info.serial, + } + if self.crosvm_usb_port_map.get(key) == binding: + return + self.crosvm_usb_port_map[key] = binding + self._save() + + def get_crosvm_usb_port(self, dev_info: USBInfo, vm_name: str, socket_path: str) -> int | None: + with self.lock: + key = self._usb_key(dev_info) + if key is None: + return None + binding = self.crosvm_usb_port_map.get(key) + if binding is None: + return None + if ( + binding.get("vm") != vm_name + or binding.get("socket_generation") != self._socket_generation(socket_path) + or not self._binding_matches_device(binding, dev_info) + ): + del self.crosvm_usb_port_map[key] + self._save() + return None + port = binding.get("port") + return port if self._valid_port(port) else None + + def clear_crosvm_usb_ports(self, vm_names: list[str]) -> None: + with self.lock: + keys = [key for key, value in self.crosvm_usb_port_map.items() if value.get("vm") in vm_names] + if not keys: + return + for key in keys: + del self.crosvm_usb_port_map[key] + self._save() + def list_pci_devices(self) -> dict[str, str]: with self.lock: return dict(self.pci_device_vm_map) diff --git a/vhotplug/vhotplug.py b/vhotplug/vhotplug.py index fb02355..01c42af 100644 --- a/vhotplug/vhotplug.py +++ b/vhotplug/vhotplug.py @@ -129,6 +129,8 @@ async def monitor_loop(app_context: AppContext, file_watcher: FileWatcher, attac if vm_name: vms_restarted.append(vm_name) + app_context.dev_state.clear_crosvm_usb_ports(vms_restarted) + # Check non-USB evdev devices for restarted VMs await attach_connected_evdev(app_context) # Check PCI devices for restarted VMs diff --git a/vhotplug/vmm.py b/vhotplug/vmm.py index ed424c9..4c88daa 100644 --- a/vhotplug/vmm.py +++ b/vhotplug/vmm.py @@ -17,7 +17,12 @@ def _get_crosvm_bin(app_context: AppContext) -> str | None: return crosvm_bin if isinstance(crosvm_bin, str) else None -async def vmm_add_device(app_context: AppContext, vm: dict[str, str], dev_info: USBInfo | PCIInfo | EvdevInfo) -> None: +async def vmm_add_device( + app_context: AppContext, + vm: dict[str, str], + dev_info: USBInfo | PCIInfo | EvdevInfo, + crosvm_usb_port: int | None = None, +) -> int | None: """Attaches a device to the VM based on the VMM type and device type.""" vm_type = vm.get("type") vm_socket = vm.get("socket") @@ -33,19 +38,25 @@ async def vmm_add_device(app_context: AppContext, vm: dict[str, str], dev_info: await qemu.add_pci_device(dev_info) else: await qemu.add_evdev_device(dev_info) - elif vm_type == "crosvm": + return None + if vm_type == "crosvm": crosvm = CrosvmLink(vm_socket, _get_crosvm_bin(app_context)) if isinstance(dev_info, USBInfo): - await crosvm.add_usb_device(dev_info) - elif isinstance(dev_info, PCIInfo): + return await crosvm.add_usb_device(dev_info, crosvm_usb_port) + if isinstance(dev_info, PCIInfo): await crosvm.add_pci_device(dev_info) else: raise RuntimeError(f"Evdev passthrough is not supported by {vm_type}") - else: - raise RuntimeError(f"Unknown VM type: {vm_type}") + return None + raise RuntimeError(f"Unknown VM type: {vm_type}") -async def vmm_remove_device(app_context: AppContext, vm: dict[str, Any], dev_info: USBInfo | PCIInfo) -> None: +async def vmm_remove_device( + app_context: AppContext, + vm: dict[str, Any], + dev_info: USBInfo | PCIInfo, + crosvm_usb_port: int | None = None, +) -> None: """Removes a device from the VM based on the VMM type and device type.""" vm_type = vm.get("type") vm_socket = vm.get("socket") @@ -68,7 +79,7 @@ async def vmm_remove_device(app_context: AppContext, vm: dict[str, Any], dev_inf # Crosvm seems to automatically remove the device from the list so this code is not really used crosvm = CrosvmLink(vm_socket, _get_crosvm_bin(app_context)) if isinstance(dev_info, USBInfo): - await crosvm.remove_usb_device(dev_info) + await crosvm.remove_usb_device(dev_info, crosvm_usb_port) else: await crosvm.remove_pci_device(dev_info) else: