diff --git a/hydra/qmp.py b/hydra/qmp.py index eb0f687..70b2487 100644 --- a/hydra/qmp.py +++ b/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() diff --git a/tests/test_qmp.py b/tests/test_qmp.py index 4488986..6c300b5 100644 --- a/tests/test_qmp.py +++ b/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()