fix: handle asynchronous QMP events before greeting

hydra/qmp.py

@@ -23,6 +23,12 @@ class QMPClient:
             self.socket.settimeout(timeout)
             self.socket.connect(str(path))
             greeting = self._receive(deadline)
+            # Real QEMU can queue a SPICE event across monitor reconnects.
+            # Consume only protocol events, retaining the original deadline.
+            while 'event' in greeting:
+                if not self._async_event(greeting):
+                    raise QMPError('Invalid QMP event before greeting')
+                greeting = self._receive(deadline)
             if not isinstance(greeting.get('QMP'), dict):
                 raise QMPError('Invalid QMP greeting')
             self._command('qmp_capabilities', None, deadline)
@@ -32,6 +38,22 @@ class QMPClient:
                 raise
             raise QMPError(f'QMP connection failed: {exc}') from exc
 
+    @staticmethod
+    def _async_event(message: dict) -> bool:
+        if not {'event', 'timestamp'} <= message.keys() <= {'event', 'timestamp', 'data'}:
+            return False
+        if not isinstance(message['event'], str) or not message['event']:
+            return False
+        if 'data' in message and not isinstance(message['data'], dict):
+            return False
+        stamp = message['timestamp']
+        if (not isinstance(stamp, dict) or set(stamp) != {'seconds', 'microseconds'}
+                or any(type(value) is not int for value in stamp.values())):
+            return False
+        seconds, micros = stamp['seconds'], stamp['microseconds']
+        # QMP specifies (-1, -1) when the server cannot retrieve host time.
+        return (seconds == -1 and micros == -1) or (seconds >= 0 and 0 <= micros < 1000000)
+
     def _receive(self, deadline: float) -> dict:
         while True:
             remaining = deadline - time.monotonic()

tests/test_qmp.py

@@ -136,6 +136,83 @@ class QMPTests(unittest.TestCase):
                 client.command('query-status')
             self.assertLess(time.monotonic() - started, 0.25)
 
+    def test_spice_events_before_greeting_still_negotiate_and_query_identity(self):
+        commands = []
+        def handler(conn):
+            send(conn, {'event': 'SPICE_CONNECTED', 'data': {'server': {'family': 'unix'}},
+                        'timestamp': {'seconds': 1791919209, 'microseconds': 650048}})
+            send(conn, {'event': 'SPICE_DISCONNECTED',
+                        'timestamp': {'seconds': -1, 'microseconds': -1}})
+            greeting(conn)
+            with conn.makefile('rb') as stream:
+                for name in ('qmp_capabilities', 'query-name'):
+                    request = json.loads(stream.readline())
+                    commands.append(request['execute'])
+                    send(conn, {'return': {} if name == 'qmp_capabilities' else {'name': 'owned'},
+                                'id': request['id']})
+        self.server(handler)
+        with QMPClient(self.path, timeout=0.5) as client:
+            self.assertEqual(client.command('query-name'), {'name': 'owned'})
+        self.assertEqual(commands, ['qmp_capabilities', 'query-name'])
+
+    def test_before_greeting_rejects_replies_and_malformed_events(self):
+        messages = [
+            {}, {'return': {}}, {'error': {}},
+            {'event': 'SPICE_CONNECTED'},
+            {'event': '', 'timestamp': {'seconds': 1, 'microseconds': 0}},
+            {'event': 1, 'timestamp': {'seconds': 1, 'microseconds': 0}},
+            {'event': 'STOP', 'data': [], 'timestamp': {'seconds': 1, 'microseconds': 0}},
+            {'event': 'STOP', 'timestamp': {'seconds': True, 'microseconds': 0}},
+            {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': 1000000}},
+            {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': -1}},
+            {'event': 'STOP', 'timestamp': {'seconds': 1}},
+            {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': 0}, 'return': {}},
+        ]
+        for message in messages:
+            with self.subTest(message=message):
+                self.path.unlink(missing_ok=True)
+                server = self.server(lambda conn, message=message: (send(conn, message), greeting(conn)))
+                with self.assertRaisesRegex(QMPError, 'Invalid QMP'):
+                    QMPClient(self.path, timeout=0.2)
+                server.close()
+
+    def test_before_greeting_event_flood_cannot_extend_handshake_deadline(self):
+        def handler(conn):
+            while True:
+                send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}})
+                time.sleep(0.005)
+        self.server(handler)
+        started = time.monotonic()
+        with self.assertRaisesRegex(QMPError, 'deadline|receive'):
+            QMPClient(self.path, timeout=0.08)
+        elapsed = time.monotonic() - started
+        self.assertGreaterEqual(elapsed, 0.06)
+        self.assertLess(elapsed, 0.25)
+
+    def test_greeting_and_capabilities_share_original_handshake_deadline(self):
+        def handler(conn):
+            send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}})
+            time.sleep(0.06)
+            greeting(conn)
+            request = json.loads(conn.makefile('rb').readline())
+            time.sleep(0.06)
+            send(conn, {'return': {}, 'id': request['id']})
+        self.server(handler)
+        started = time.monotonic()
+        with self.assertRaisesRegex(QMPError, 'deadline|receive'):
+            QMPClient(self.path, timeout=0.1)
+        self.assertLess(time.monotonic() - started, 0.17)
+
+    def test_event_then_greeting_does_not_bypass_capabilities_failure(self):
+        def handler(conn):
+            send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}})
+            greeting(conn)
+            request = json.loads(conn.makefile('rb').readline())
+            send(conn, {'error': {'class': 'GenericError', 'desc': 'refused'}, 'id': request['id']})
+        self.server(handler)
+        with self.assertRaisesRegex(QMPError, 'QMP qmp_capabilities'):
+            QMPClient(self.path, timeout=0.2)
+
 
 if __name__ == '__main__':
     unittest.main()