功能：接入Linux原生光标并完善设备接入工具

tools/tests/test_native_cursor.py

@@ -0,0 +1,63 @@
+"""Bounds/replay checks for untrusted cursor metadata before SDL allocation."""
+import base64
+import importlib.util
+import hashlib
+from pathlib import Path
+import unittest
+import tempfile
+
+spec = importlib.util.spec_from_file_location(
+    'native_cursor_client', Path(__file__).resolve().parents[2] /
+    'terminal/native-cursor/client.py')
+client = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(client)
+
+
+class CursorProtocolTests(unittest.TestCase):
+    def state(self):
+        return dict(type='cursor', version=1, lease='c'*32, seq=2, ack=0, w=2, h=2,
+                    hx=1, hy=1, x=100, y=100, sw=2560, sh=1440,
+                    rgba=base64.b64encode(bytes([255]*16)).decode())
+
+    def test_valid_and_hidden(self):
+        for pixels in (bytes([255]*16), bytes(16)):
+            s = self.state()
+            s['rgba'] = base64.b64encode(pixels).decode()
+            packet = client.validate(s, 'c'*32, [2560, 1440], 1)
+            self.assertEqual(len(packet), 88)
+            self.assertEqual(packet[64:80], pixels)
+
+    def test_bad_metadata(self):
+        for key, value in [('version', 2), ('version', True), ('ack', -1),
+                           ('ack', 2**63), ('lease', 'd'*32), ('type', 'ready'),
+                           ('seq', 1), ('seq', 2**63), ('w', 0), ('w', 129),
+                           ('h', -1), ('hx', 2), ('hy', -1), ('x', -1),
+                           ('y', 1440), ('sw', 1920), ('w', True),
+                           ('seq', 2.0), ('rgba', '***'), ('rgba', 'AAAA')]:
+            with self.subTest(key=key, value=value):
+                state = self.state()
+                state[key] = value
+                with self.assertRaises(ValueError):
+                    client.validate(state, 'c'*32, [2560, 1440], 1)
+
+    def test_missing_field(self):
+        state = self.state()
+        del state['h']
+        with self.assertRaises(ValueError):
+            client.validate(state, 'c'*32, [2560, 1440], 1)
+
+    def test_old_client_rejected_before_lease(self):
+        with tempfile.TemporaryDirectory() as directory:
+            binary = Path(directory) / 'client'
+            binary.write_bytes(b'validated client')
+            expected = hashlib.sha256(binary.read_bytes()).hexdigest()
+            client.verify_client(binary, expected)
+            binary.write_bytes(b'old client without cursor receiver')
+            with self.assertRaises(ValueError):
+                client.verify_client(binary, expected)
+            with self.assertRaises(ValueError):
+                client.verify_client(binary, None)
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/tests/test_usb_device_qmp.py

@@ -1,5 +1,7 @@
 import importlib.util
 from pathlib import Path
+import socket
+import threading
 import unittest
 
 spec = importlib.util.spec_from_file_location("usb_qmp", Path(__file__).parents[1] / "usb-device-qmp.py")
@@ -51,6 +53,24 @@ class UsbQmpTests(unittest.TestCase):
             usb.attach(q, 100, 40001, "ehci-2.0", "0951", "1666")
         self.assertEqual(q.calls[-1], ("chardev-remove", {"id": "vdi_usb_box_100"}))
 
+    def test_status_probe_distinguishes_ready_from_empty(self):
+        listener = socket.socket()
+        listener.bind(("127.0.0.1", 0))
+        listener.listen()
+        port = listener.getsockname()[1]
+
+        def serve():
+            conn, _ = listener.accept()
+            conn.sendall(b"ready\n")
+            conn.close()
+            listener.close()
+
+        thread = threading.Thread(target=serve)
+        thread.start()
+        self.assertTrue(usb.probe_status(port, timeout=1))
+        thread.join(timeout=1)
+        self.assertFalse(usb.probe_status(port, timeout=0.05))
+
 
 if __name__ == "__main__":
     unittest.main()