Files
python----/工具测试/监听usb接口.ipynb
2025-10-20 12:32:18 +08:00

383 lines
12 KiB
Plaintext

{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "f53ca1dd",
"metadata": {},
"outputs": [],
"source": [
"import wmi\n",
"import subprocess\n",
"import time\n",
"from threading import Lock\n",
"\n",
"# 配置部分\n",
"KEY_FILE = \".key\" # U盘中的密码文件\n",
"EXPECTED_PASSWORD = \"1234\" # 预设密码\n",
"DEVCON_PATH = r\"E:\\下载器\\Devcon\\x64\\devcon.exe\"\n",
"DISABLED_DEVICES = set()\n",
"device_lock = Lock()\n",
"\n",
"def disable_device(dev_id):\n",
" subprocess.run([DEVCON_PATH, \"disable\", dev_id], shell=True, capture_output=True)\n",
" with device_lock:\n",
" DISABLED_DEVICES.add(dev_id)\n",
"\n",
"def enable_device(dev_id):\n",
" subprocess.run([DEVCON_PATH, \"enable\", dev_id], shell=True, capture_output=True)\n",
" with device_lock:\n",
" if dev_id in DISABLED_DEVICES:\n",
" DISABLED_DEVICES.remove(dev_id)\n",
"\n",
"def get_usb_storage_drives():\n",
" \"\"\"获取所有可移动USB存储设备的盘符\"\"\"\n",
" c = wmi.WMI()\n",
" drives = []\n",
" for disk in c.Win32_DiskDrive(InterfaceType=\"USB\"):\n",
" for p in disk.associators(\"Win32_DiskDriveToDiskPartition\"):\n",
" for d in p.associators(\"Win32_LogicalDiskToPartition\"):\n",
" drives.append(d.DeviceID)\n",
" return drives\n",
"\n",
"def check_usb_key(drive):\n",
" \"\"\"检查U盘根目录的密钥文件\"\"\"\n",
" try:\n",
" key_path = fr\"{drive}\\{KEY_FILE}\"\n",
" with open(key_path, \"r\") as f:\n",
" return f.read().strip() == EXPECTED_PASSWORD\n",
" except:\n",
" return False\n",
"\n",
"def handle_storage_insert():\n",
" \"\"\"处理U盘插入事件\"\"\"\n",
" time.sleep(2) # 等待系统分配盘符\n",
" for drive in get_usb_storage_drives():\n",
" if check_usb_key(drive):\n",
" print(f\"在 {drive} 发现有效密钥\")\n",
" with device_lock:\n",
" for dev_id in list(DISABLED_DEVICES):\n",
" enable_device(dev_id)\n",
" return True\n",
" return False\n",
"\n",
"def monitor_devices():\n",
" c = wmi.WMI()\n",
" \n",
" # 存储设备监听线程\n",
" def storage_watcher():\n",
" storage_monitor = c.Win32_DeviceChangeEvent.watch_for(\n",
" event_type=2, # 设备插入\n",
" delay_secs=2\n",
" )\n",
" while True:\n",
" try:\n",
" storage_monitor()\n",
" if handle_storage_insert():\n",
" print(\"成功启用被禁用的键盘设备\")\n",
" except Exception as e:\n",
" print(f\"存储设备监控异常: {e}\")\n",
"\n",
" # 键盘设备监听\n",
" keyboard_monitor = c.Win32_DeviceChangeEvent.watch_for(\n",
" EventType=2,\n",
" delay_secs=1\n",
" )\n",
" \n",
" # 启动存储监控线程\n",
" import threading\n",
" threading.Thread(target=storage_watcher, daemon=True).start()\n",
"\n",
" while True:\n",
" try:\n",
" keyboard_monitor()\n",
" print(\"检测到新输入设备...\")\n",
" c = wmi.WMI()\n",
" for dev in c.Win32_PnPEntity():\n",
" if \"Keyboard\" in str(dev.Description) and \"USB\" in str(dev.Description):\n",
" dev_id = dev.DeviceID\n",
" if dev_id not in DISABLED_DEVICES:\n",
" print(f\"禁用USB键盘: {dev_id}\")\n",
" disable_device(dev_id)\n",
" except Exception as e:\n",
" print(f\"键盘监控异常: {e}\")\n",
"\n",
"# if __name__ == \"__main__\":\n",
"# # 检查管理员权限\n",
"# import ctypes, sys\n",
"# if ctypes.windll.shell32.IsUserAnAdmin() == 0:\n",
"# ctypes.windll.shell32.ShellExecuteW(None, \"runas\", sys.executable, __file__, None, 1)\n",
"# else:\n",
"# monitor_devices()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f0f8c644",
"metadata": {},
"outputs": [],
"source": [
"c = wmi.WMI()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ebc809a7",
"metadata": {},
"outputs": [],
"source": [
"watcher = c.Win32_VolumeChangeEvent.watch_for(\n",
" EventType=2 # 2表示卷载入(设备插入)\n",
")\n",
"event = watcher()\n",
"drive_letter = event.DriveName"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c2d33b3f",
"metadata": {},
"outputs": [],
"source": [
"drive_letter"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "1e6d53e6",
"metadata": {},
"outputs": [],
"source": [
"import wmi\n",
"import time\n",
"\n",
"def wait_for_usb():\n",
" c = wmi.WMI()\n",
"\n",
" # 监听USB存储设备插入事件\n",
" watcher = c.Win32_VolumeChangeEvent.watch_for(\n",
" EventType=2 # 2表示卷载入(设备插入)\n",
" )\n",
"\n",
" print(\"等待U盘插入...\")\n",
" while True:\n",
" try:\n",
" event = watcher()\n",
" drive_letter = event.DriveName\n",
" print(f\"检测到U盘插入,盘符: {drive_letter}\")\n",
" return drive_letter\n",
" except wmi.x_wmi_timed_out:\n",
" # 超时后继续等待\n",
" time.sleep(0.1)\n",
" except KeyboardInterrupt:\n",
" print(\"\\n已停止监听\")\n",
" break\n",
"\n",
"if __name__ == \"__main__\":\n",
" usb_drive = wait_for_usb()\n",
" if usb_drive:\n",
" print(f\"接下来可以使用 {usb_drive} 访问U盘文件\")\n",
" # 示例:列出U盘根目录文件\n",
" import os\n",
" print(\"根目录文件:\", os.listdir(usb_drive))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "70115805",
"metadata": {},
"outputs": [],
"source": [
"keyboard_monitor = c.Win32_DeviceChangeEvent.watch_for(\n",
" EventType=2,\n",
" delay_secs=1\n",
")\n",
"keyboard_monitor()\n",
"for dev in c.Win32_PnPEntity(fields=[\"Description\",\"DeviceID\"]):\n",
" if \"Keyboard\" in str(dev.Description):\n",
" print(f\"检测到键盘设备: {dev.Description} - ID: {dev.DeviceID}\")\n",
" os.system(rf'''E:\\下载器\\Devcon\\x64\\devcon.exe remove @\"{dev.DeviceID}\"''')"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e4475bb3",
"metadata": {},
"outputs": [],
"source": [
"\n",
" # if \"Keyboard\" in str(dev.Description) and \"USB\" in str(dev.Description):\n",
" # dev_id = dev.DeviceID\n",
" # if dev_id not in DISABLED_DEVICES:\n",
" # print(f\"禁用USB键盘: {dev_id}\")\n",
" # disable_device(dev_id)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e78487c2",
"metadata": {},
"outputs": [],
"source": [
"subprocess.run([DEVCON_PATH, \"remove\", r'''@\"HID\\VID_0C45&PID_8071&MI_00\\7&612DDAA&0&0000\"'''], shell=True, capture_output=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f7c7a31c",
"metadata": {},
"outputs": [],
"source": [
"import os \n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2787e77e",
"metadata": {},
"outputs": [],
"source": [
"for disk in c.Win32_DiskDrive():\n",
" print(f\"USB设备: {disk.InterfaceType}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "113e0a8e",
"metadata": {},
"outputs": [],
"source": [
"import tkinter as tk\n",
"\n",
"def show_auto_close_message(title, message, duration=3):\n",
" # 创建主窗口(隐藏)\n",
" root = tk.Tk()\n",
" root.withdraw()\n",
"\n",
" # 创建弹窗\n",
" popup = tk.Toplevel()\n",
" popup.title(title)\n",
" \n",
" # 设置消息内容\n",
" label = tk.Label(popup, text=message, padx=30, pady=20)\n",
" label.pack()\n",
" \n",
" # 禁止窗口缩放\n",
" popup.resizable(False, False)\n",
" \n",
" # 计算居中位置\n",
" popup.update_idletasks() # 强制更新窗口尺寸\n",
" width = popup.winfo_width()\n",
" height = popup.winfo_height()\n",
" x = (popup.winfo_screenwidth() - width) // 2\n",
" y = (popup.winfo_screenheight() - height) // 2\n",
" popup.geometry(f\"+{x}+{y}\")\n",
"\n",
" # 设置自动关闭\n",
" def close():\n",
" popup.destroy()\n",
" root.destroy() # 同时销毁主窗口\n",
"\n",
" popup.after(duration * 1000, close)\n",
" popup.mainloop()\n",
"\n",
"# 使用示例\n",
"if __name__ == \"__main__\":\n",
" show_auto_close_message(\"操作成功\", \"设备已启用!\", 3)"
]
},
{
"cell_type": "code",
"execution_count": 7,
"id": "4c8043c5",
"metadata": {},
"outputs": [
{
"ename": "KeyboardInterrupt",
"evalue": "",
"output_type": "error",
"traceback": [
"\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
"\u001b[1;31mKeyboardInterrupt\u001b[0m Traceback (most recent call last)",
"Cell \u001b[1;32mIn[7], line 33\u001b[0m\n\u001b[0;32m 30\u001b[0m popup\u001b[38;5;241m.\u001b[39mafter(\u001b[38;5;241m3000\u001b[39m, close_window)\n\u001b[0;32m 32\u001b[0m \u001b[38;5;66;03m# 启动主循环\u001b[39;00m\n\u001b[1;32m---> 33\u001b[0m \u001b[43mroot\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mmainloop\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
"File \u001b[1;32md:\\Anaconda3\\lib\\tkinter\\__init__.py:1429\u001b[0m, in \u001b[0;36mMisc.mainloop\u001b[1;34m(self, n)\u001b[0m\n\u001b[0;32m 1427\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mmainloop\u001b[39m(\u001b[38;5;28mself\u001b[39m, n\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m0\u001b[39m):\n\u001b[0;32m 1428\u001b[0m \u001b[38;5;124;03m\"\"\"Call the mainloop of Tk.\"\"\"\u001b[39;00m\n\u001b[1;32m-> 1429\u001b[0m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mtk\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mmainloop\u001b[49m\u001b[43m(\u001b[49m\u001b[43mn\u001b[49m\u001b[43m)\u001b[49m\n",
"\u001b[1;31mKeyboardInterrupt\u001b[0m: "
]
}
],
"source": [
"import tkinter as tk\n",
"\n",
"def close_window():\n",
" # 关闭弹窗和主窗口(确保程序退出)\n",
" popup.destroy()\n",
" root.destroy()\n",
"\n",
"# 创建主窗口并隐藏\n",
"root = tk.Tk()\n",
"root.withdraw()\n",
"\n",
"# 创建弹窗\n",
"popup = tk.Toplevel(root)\n",
"popup.title(\"提示\")\n",
"\n",
"# 设置弹窗大小和居中显示\n",
"window_width = 300\n",
"window_height = 200\n",
"screen_width = root.winfo_screenwidth()\n",
"screen_height = root.winfo_screenheight()\n",
"x = (screen_width - window_width) // 2\n",
"y = (screen_height - window_height) // 2\n",
"popup.geometry(f\"{window_width}x{window_height}+{x}+{y}\")\n",
"\n",
"# 添加文本标签\n",
"label = tk.Label(popup, text=\"3秒后自动关闭\", font=(\"微软雅黑\", 12))\n",
"label.pack(expand=True, pady=50)\n",
"\n",
"# 设置3秒后关闭窗口\n",
"popup.after(3000, close_window)\n",
"\n",
"# 启动主循环\n",
"root.mainloop()"
]
},
{
"cell_type": "markdown",
"id": "458bd510",
"metadata": {},
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "base",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.7"
}
},
"nbformat": 4,
"nbformat_minor": 5
}