功能:扩展U盘兼容并优化插拔响应和导出负载

This commit is contained in:
songyiming committed 2026-11-04 19:12:42 +08:00
1 parent b2de05c6a6
commit bb5f554820
9 files changed
+589 -73

No files matched your search

+28 -10
View File
@@ -1,12 +1,13 @@
# USB 设备透传 POC
# USB 设备透传
将盒子上指定的 U 盘经 usbredir + SSH + QEMU 接入 Windows,枚举真实 VID/PID/序列号和
USBSTOR,使用 Windows 自带驱动。这是 Moonlight 显示会话之外的设备通道,不是网络文件盘。
当前只支持明确登记的一个纯 USB 存储设备、一个盒子和一个 VM;盒子不调用 Proxmox API。
将盒子上的纯 USB 存储设备经 usbredir + SSH + QEMU 接入 Windows,枚举真实
VID/PID/序列号和 USBSTOR,使用 Windows 自带驱动。这是 Moonlight 显示会话之外的
设备通道,不是网络文件盘。默认策略允许任意一个纯 USB 存储设备;一个盒子和一个
VM 同时只导出一个设备,键盘、鼠标、Hub 和显示设备仍由盒子本地使用。
- `src/main.rs`:Rust 标准库实现设备扫描、精确匹配、本地挂载/占用保护、导出及拔插恢复。
- `etc/vdi-usb-monitor.service`:盒子服务,读取 root 管理的单设备白名单。
- `../../tools/deploy-usb-device-poc.py`:管理员部署和受限 SSH 隧道。
- `src/main.rs`:Rust 标准库实现设备扫描、存储类筛选、本地挂载/占用保护、导出及拔插恢复。
- `etc/vdi-usb-monitor.service`:盒子服务,默认使用 `--any-storage`。
- `../../tools/deploy-usb-device-poc.py`:管理员部署、受限 SSH 隧道和 QMP 自动恢复服务。
- `../../tools/usb-device-qmp.py`:在 PVE 节点操作独立 QMP 热插拔端点。
- `../../tools/test-usb-device-poc.ps1`:Windows USB 身份和文件回读校验。
@@ -17,12 +18,29 @@ cargo test --offline
cargo build --release --offline
target/release/vdi-usb-monitor list
sudo target/release/vdi-usb-monitor serve \
--vendor 0951 --product 1666 --serial YOUR_USB_SERIAL --listen 127.0.0.1:40001
--any-storage --listen 127.0.0.1:40001
```
`list` 输出 JSON。`serve` 必须指定非空序列号,只能监听 IPv4 回环地址;
拒绝键鼠、Hub、包含非存储接口的复合设备以及本地尚在使用的磁盘。
`list` 输出 JSON。`serve --any-storage` 只接受纯 USB Mass Storage 类设备,
只能监听 IPv4 回环地址;拒绝键鼠、Hub、包含非存储接口的复合设备以及本地尚在使用
的磁盘。仍可使用 `--vendor/--product/--serial` 保留旧的精确设备策略。
本地已挂载时不会强制卸载,应先关闭本地文件操作并正常卸载。
插拔检测使用 `udevadm monitor --udev --property --subsystem-match=usb` 事件,
空闲时每 500ms 扫描兜底,活动设备每秒核对当前实例;其他设备的 sysfs 扫描错误
不会再断开正在导出的 U 盘。日志记录 Unix 毫秒时间、占用检查耗时和最近 USB 事件
至启动导出的间隔;后者不包含物理枚举、网络连接或 Windows 挂载时间。
Windows 端 `tools/set-usb-experience.ps1` 支持 Apply/Restore,为当前登录用户将
存储设备自动播放设置为“每次询问”,原设置保存在 ProgramData。不会改 U 盘分区、
隐藏启动分区、关闭通知、重启 Explorer 或更改其他任务栏图标。
发行版 usbredirect 0.13 的持续 G_IO_OUT 监听可能造成 CPU 空转。盒子当前使用
官方 0.15 的私有构建,保留 `/usr/bin/usbredirect`。可使用
`tools/build-usbredirect-private.sh` 构建后,将程序安装到
`/usr/local/libexec/vdi-usbredirect`,再在 root 管理的 device.env 中配置
`VDI_USBREDIRECT_BINARY=/usr/local/libexec/vdi-usbredirect`。
未配置时监控程序仍使用 `/usr/bin/usbredirect`;更新运行程序前须安全弹出并拔下 U 盘。
完整现场、部署、限制和回退见
[`docs/usb-device-poc-20261103.md`](../../docs/usb-device-poc-20261103.md)。
@@ -1,11 +1,11 @@
[Unit]
Description=VDI approved USB storage redirection (device POC)
Description=VDI USB Mass Storage redirection
After=network.target systemd-udevd.service
[Service]
Type=simple
EnvironmentFile=/etc/vdi-usb-monitor/device.env
ExecStart=/usr/local/libexec/vdi-usb-monitor serve --vendor ${VDI_USB_VENDOR} --product ${VDI_USB_PRODUCT} --serial ${VDI_USB_SERIAL} --listen 127.0.0.1:40001
ExecStart=/usr/local/libexec/vdi-usb-monitor serve --any-storage --listen 127.0.0.1:40001
Restart=on-failure
RestartSec=3
KillMode=control-group
+230 -36
View File
@@ -1,6 +1,63 @@
//! Device-level USB POC. Never calls the hypervisor API or exports HID devices.
use std::{collections::BTreeSet, env, fs, net::SocketAddr, path::{Path, PathBuf},
process::{self, Command}, thread, time::Duration};
//! Device-level USB redirection. Never calls the hypervisor API or exports HID devices.
use std::{collections::BTreeSet, env, fs, io::{BufRead, BufReader}, net::SocketAddr,
path::{Path, PathBuf}, process::{self, Child, Command, Stdio},
sync::mpsc::{self, Receiver}, thread, time::{Duration, Instant, SystemTime, UNIX_EPOCH}};
// udev delivers completed USB changes; the timeout also covers startup and a
// stopped event monitor. The monitor never runs a command supplied by a device.
struct UsbEvents {
child: Option<Child>,
receiver: Receiver<Instant>,
last_event: Option<Instant>,
}
impl UsbEvents {
fn start() -> Self {
let (sender, receiver) = mpsc::sync_channel(64);
let child = Command::new("/usr/bin/stdbuf")
.args(["-oL", "/usr/bin/udevadm", "monitor", "--udev", "--property", "--subsystem-match=usb"])
.stdout(Stdio::piped()).stderr(Stdio::null()).spawn();
let child = match child {
Ok(mut child) => {
let stdout = child.stdout.take().unwrap();
thread::spawn(move || {
for line in BufReader::new(stdout).lines() {
match line {
Ok(line) if line.starts_with("ACTION=") => { let _ = sender.try_send(Instant::now()); }
Ok(_) => {},
Err(_) => break,
}
}
});
Some(child)
}
Err(err) => { eprintln!("USB event monitor unavailable: {err}; using fallback scan"); None }
};
Self { child, receiver, last_event: None }
}
fn wait(&mut self, timeout: Duration) {
match self.receiver.recv_timeout(timeout) {
Ok(event) => {
self.last_event = Some(event);
while let Ok(event) = self.receiver.try_recv() { self.last_event = Some(event); }
}
Err(mpsc::RecvTimeoutError::Disconnected) => thread::sleep(timeout),
Err(mpsc::RecvTimeoutError::Timeout) => {},
}
}
}
impl Drop for UsbEvents {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() { let _ = child.kill(); let _ = child.wait(); }
}
}
fn log_event(message: &str) {
let ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis();
eprintln!("usb_time_ms={ms} {message}");
}
#[derive(Debug, Clone)]
struct Device {
@@ -38,6 +95,10 @@ fn scan(root: &Path) -> Result<Vec<Device>, String> {
let mut classes: Vec<_> = entries.iter().filter(|e|
e.file_name().to_string_lossy().starts_with(&prefix))
.map(|e| read(&e.path(), "bInterfaceClass")).collect();
// Early hotplug events can arrive before every interface is visible.
// Do not mistake the first storage interface of a composite for a pure disk.
let expected = read(&p, "bNumInterfaces").parse::<usize>().unwrap_or(0);
if expected == 0 || classes.len() != expected { continue; }
classes.sort();
devices.push(Device {
path: p.canonicalize().map_err(|e| e.to_string())?,
@@ -98,38 +159,95 @@ fn hex_id(s: &str) -> bool {
}
#[derive(Debug)]
struct Policy { vendor: String, product: String, serial: String, listen: SocketAddr }
enum MatchPolicy {
AnyStorage,
Exact { vendor: String, product: String, serial: String },
}
#[derive(Debug)]
struct Policy { matcher: MatchPolicy, listen: SocketAddr }
fn parse_policy(args: &[String]) -> Result<Policy, String> {
if args.len() != 8 { return Err("serve requires --vendor XXXX --product XXXX --serial SERIAL --listen 127.0.0.1:PORT".into()); }
let mut values = std::collections::BTreeMap::new();
for pair in args.chunks(2) {
if !["--vendor", "--product", "--serial", "--listen"].contains(&pair[0].as_str())
|| values.insert(pair[0].as_str(), pair[1].as_str()).is_some() {
return Err("unknown or duplicate policy option".into());
let mut any_storage = false;
let mut vendor = None;
let mut product = None;
let mut serial = None;
let mut listen = None;
let mut i = 0;
while i < args.len() {
match args[i].as_str() {
"--any-storage" => {
if any_storage { return Err("duplicate --any-storage".into()); }
any_storage = true;
i += 1;
}
"--vendor" | "--product" | "--serial" | "--listen" => {
if i + 1 >= args.len() { return Err("missing value for policy option".into()); }
let value = args[i + 1].clone();
let target = match args[i].as_str() {
"--vendor" => &mut vendor,
"--product" => &mut product,
"--serial" => &mut serial,
_ => &mut listen,
};
if target.replace(value).is_some() { return Err("duplicate policy option".into()); }
i += 2;
}
_ => return Err("unknown policy option".into()),
}
}
let vendor = values["--vendor"].to_ascii_lowercase();
let product = values["--product"].to_ascii_lowercase();
let serial = values["--serial"].to_string();
let listen: SocketAddr = values["--listen"].parse().map_err(|_| "invalid listen address")?;
if !hex_id(&vendor) || !hex_id(&product) || serial.is_empty()
|| serial.chars().any(char::is_control) || !listen.ip().is_loopback()
|| !listen.is_ipv4() || listen.port() == 0 {
return Err("exact VID/PID/serial and an IPv4 loopback listener are required".into());
let listen: SocketAddr = listen.ok_or("--listen is required")?.parse()
.map_err(|_| "invalid listen address")?;
if !listen.ip().is_loopback() || !listen.is_ipv4() || listen.port() == 0 {
return Err("an IPv4 loopback listener with a nonzero port is required".into());
}
let matcher = if any_storage {
if vendor.is_some() || product.is_some() || serial.is_some() {
return Err("--any-storage cannot be combined with an exact device identity".into());
}
MatchPolicy::AnyStorage
} else {
let vendor = vendor.ok_or("exact mode requires --vendor")?.to_ascii_lowercase();
let product = product.ok_or("exact mode requires --product")?.to_ascii_lowercase();
let serial = serial.ok_or("exact mode requires --serial")?;
if !hex_id(&vendor) || !hex_id(&product) || serial.is_empty()
|| serial.chars().any(char::is_control) {
return Err("exact mode requires valid VID/PID and a nonempty serial".into());
}
MatchPolicy::Exact { vendor, product, serial }
};
Ok(Policy { matcher, listen })
}
fn matches_policy(device: &Device, p: &Policy) -> bool {
match &p.matcher {
MatchPolicy::AnyStorage => true,
MatchPolicy::Exact { vendor, product, serial } =>
device.vendor == vendor.as_str() && device.product == product.as_str()
&& device.serial == serial.as_str(),
}
Ok(Policy { vendor, product, serial, listen })
}
fn select<'a>(devices: &'a [Device], p: &Policy) -> Result<&'a Device, String> {
let found: Vec<_> = devices.iter().filter(|d|
d.vendor == p.vendor && d.product == p.product && d.serial == p.serial).collect();
if found.len() != 1 { return Err(format!("waiting for exactly one approved device; found {}", found.len())); }
let d = found[0];
if !storage_only(&d.device_class, &d.classes) || d.bus == 0 || d.address == 0 {
return Err("not a pure USB mass-storage device; refusing export".into());
matches_policy(d, p) && storage_only(&d.device_class, &d.classes)
&& d.bus != 0 && d.address != 0).collect();
match &p.matcher {
MatchPolicy::Exact { .. } if found.len() != 1 =>
Err(format!("waiting for exactly one approved device; found {}", found.len())),
_ => found.into_iter().next().ok_or_else(||
"waiting for a pure USB mass-storage device".into()),
}
Ok(d)
}
fn still_present(device: &Device) -> bool {
// An unrelated USB device can vanish while a full sysfs scan is running.
// Inspect only the exported instance so that does not disconnect this disk.
read(&device.path, "busnum").parse::<u16>().ok() == Some(device.bus)
&& read(&device.path, "devnum").parse::<u16>().ok() == Some(device.address)
&& read(&device.path, "idVendor") == device.vendor
&& read(&device.path, "idProduct") == device.product
&& read(&device.path, "serial") == device.serial
}
fn run() -> Result<(), String> {
@@ -144,40 +262,47 @@ fn run() -> Result<(), String> {
return Ok(());
}
if args.first().map(String::as_str) != Some("serve") {
return Err("usage: vdi-usb-monitor list | serve --vendor XXXX --product XXXX --serial SERIAL --listen 127.0.0.1:PORT".into());
return Err("usage: vdi-usb-monitor list | serve (--any-storage | --vendor XXXX --product XXXX --serial SERIAL) --listen 127.0.0.1:PORT".into());
}
let policy = parse_policy(&args[1..])?;
let redirect_binary = env::var("VDI_USBREDIRECT_BINARY").unwrap_or_else(|_| "/usr/bin/usbredirect".into());
if !Path::new(&redirect_binary).is_absolute() || !Path::new(&redirect_binary).is_file() {
return Err("VDI_USBREDIRECT_BINARY must name an installed absolute executable path".into());
}
log_event(&format!("USB exporter: {redirect_binary}"));
let mut events = UsbEvents::start();
let mut last = String::new();
loop {
let result: Result<(), String> = scan(Path::new("/sys/bus/usb/devices")).and_then(|devices| {
let device = select(&devices, &policy)?;
let checked_at = Instant::now();
check_local_use(device)?;
// Bus/address selects one physical instance, never an arbitrary VID/PID match.
let selector = format!("{}-{}", device.bus, device.address);
eprintln!("exporting approved storage {} at {}", device.bus_id, policy.listen);
let mut child = Command::new("/usr/bin/usbredirect")
log_event(&format!("exporting storage {} {:}:{:} at {}; local_check_ms={} since_latest_usb_event_ms={}",
device.bus_id, device.vendor, device.product, policy.listen,
checked_at.elapsed().as_millis(),
events.last_event.map(|time| time.elapsed().as_millis().to_string()).unwrap_or_else(|| "startup".into())));
let mut child = Command::new(&redirect_binary)
.args(["--device", &selector, "--as", &policy.listen.to_string(), "--keepalive"])
.spawn().map_err(|e| e.to_string())?;
loop {
if let Some(status) = child.try_wait().map_err(|e| e.to_string())? {
return Err(format!("export session ended: {status}; waiting before retry"));
}
thread::sleep(Duration::from_secs(1));
let present = scan(Path::new("/sys/bus/usb/devices")).ok().map(|current| {
select(&current, &policy).map(|d| d.bus == device.bus
&& d.address == device.address && d.bus_id == device.bus_id).unwrap_or(false)
}).unwrap_or(false);
if !present {
events.wait(Duration::from_secs(1));
if !still_present(device) {
let _ = child.kill();
let _ = child.wait();
log_event(&format!("storage {} disconnected", device.bus_id));
return Err("approved USB device disconnected; waiting for reinsertion".into());
}
}
});
if let Err(message) = result {
if message != last { eprintln!("{message}"); last = message; }
if message != last { log_event(&message); last = message; }
}
thread::sleep(Duration::from_secs(2));
events.wait(Duration::from_millis(500));
}
}
@@ -206,6 +331,14 @@ mod tests {
assert!(parse_policy(&policy(address)).is_err());
}
}
#[test]
fn any_storage_policy_accepts_loopback_listener() {
let p = parse_policy(&["--any-storage".into(), "--listen".into(), "127.0.0.1:40001".into()]).unwrap();
assert!(matches!(p.matcher, MatchPolicy::AnyStorage));
assert!(parse_policy(&["--any-storage".into(), "--vendor".into(), "0951".into(),
"--listen".into(), "127.0.0.1:40001".into()]).is_err());
}
#[test]
fn mounted_partition_blocks_parent_disk_export() {
let numbers = BTreeSet::from(["8:0".into(), "8:1".into()]);
@@ -226,6 +359,67 @@ mod tests {
assert!(select(&[d.clone()], &p).is_ok());
assert!(select(&[d.clone(), d], &p).is_err());
}
#[test]
fn any_storage_selects_storage_and_rejects_hid() {
let p = parse_policy(&["--any-storage".into(), "--listen".into(), "127.0.0.1:40001".into()]).unwrap();
let storage = Device {
path: PathBuf::from("/sys/test"), bus_id: "3-1".into(),
vendor: "1234".into(), product: "5678".into(), serial: "".into(),
name: "storage".into(), bus: 3, address: 5,
classes: vec!["08".into()], device_class: "00".into(),
};
let hid = Device { classes: vec!["03".into()], ..storage.clone() };
assert_eq!(select(&[hid, storage.clone()], &p).unwrap().bus_id, "3-1");
}
#[test]
fn usb_events_wake_and_coalesce_without_waiting_for_the_scan_timer() {
let (sender, receiver) = mpsc::sync_channel(8);
let first = Instant::now();
let latest = Instant::now();
sender.send(first).unwrap();
sender.send(latest).unwrap();
let mut events = UsbEvents { child: None, receiver, last_event: None };
events.wait(Duration::from_secs(30));
assert_eq!(events.last_event, Some(latest));
}
#[test]
fn active_presence_check_rejects_a_replacement_on_the_same_port() {
let root = env::temp_dir().join(format!("vdi-usb-presence-{}-{}", process::id(),
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()));
fs::create_dir(&root).unwrap();
for (name, value) in [("busnum", "3"), ("devnum", "5"), ("idVendor", "1234"),
("idProduct", "5678"), ("serial", "TEST")] {
fs::write(root.join(name), value).unwrap();
}
let device = Device { path: root.clone(), bus_id: "3-1".into(), vendor: "1234".into(),
product: "5678".into(), serial: "TEST".into(), name: "test".into(),
bus: 3, address: 5, classes: vec!["08".into()], device_class: "00".into() };
assert!(still_present(&device));
fs::write(root.join("devnum"), "6").unwrap();
assert!(!still_present(&device));
fs::remove_dir_all(root).unwrap();
}
#[test]
fn incomplete_composite_cannot_be_exported_as_storage() {
let root = env::temp_dir().join(format!("vdi-usb-enumeration-{}-{}", process::id(),
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()));
let device = root.join("3-1");
fs::create_dir_all(&device).unwrap();
for (name, value) in [("idVendor", "1234"), ("idProduct", "5678"),
("bDeviceClass", "00"), ("bNumInterfaces", "2")] {
fs::write(device.join(name), value).unwrap();
}
fs::create_dir(root.join("3-1:1.0")).unwrap();
fs::write(root.join("3-1:1.0/bInterfaceClass"), "08").unwrap();
assert!(scan(&root).unwrap().is_empty());
fs::create_dir(root.join("3-1:1.1")).unwrap();
fs::write(root.join("3-1:1.1/bInterfaceClass"), "03").unwrap();
let devices = scan(&root).unwrap();
assert_eq!(devices.len(), 1);
assert!(!storage_only(&devices[0].device_class, &devices[0].classes));
fs::remove_dir_all(root).unwrap();
}
#[test]
fn enumeration_json_escapes_control_characters() {
assert_eq!(json_string("a\n\"\\"), "\"a\\u000a\\\"\\\\\"");