aboutsummaryrefslogtreecommitdiff
path: root/python/qemu/qmp.py
blob: 9223307ed81b349bff76a9119f75b6051d80875b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
""" QEMU Monitor Protocol Python class """
# Copyright (C) 2009, 2010 Red Hat Inc.
#
# Authors:
#  Luiz Capitulino <lcapitulino@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2.  See
# the COPYING file in the top-level directory.

import errno
import json
import logging
import socket
from types import TracebackType
from typing import (
    Any,
    Dict,
    List,
    Optional,
    TextIO,
    Tuple,
    Type,
    Union,
    cast,
)


# QMPMessage is a QMP Message of any kind.
# e.g. {'yee': 'haw'}
#
# QMPReturnValue is the inner value of return values only.
# {'return': {}} is the QMPMessage,
# {} is the QMPReturnValue.
QMPMessage = Dict[str, Any]
QMPReturnValue = Dict[str, Any]

InternetAddrT = Tuple[str, str]
UnixAddrT = str
SocketAddrT = Union[InternetAddrT, UnixAddrT]


class QMPError(Exception):
    """
    QMP base exception
    """


class QMPConnectError(QMPError):
    """
    QMP connection exception
    """


class QMPCapabilitiesError(QMPError):
    """
    QMP negotiate capabilities exception
    """


class QMPTimeoutError(QMPError):
    """
    QMP timeout exception
    """


class QMPProtocolError(QMPError):
    """
    QMP protocol error; unexpected response
    """


class QMPResponseError(QMPError):
    """
    Represents erroneous QMP monitor reply
    """
    def __init__(self, reply: QMPMessage):
        try:
            desc = reply['error']['desc']
        except KeyError:
            desc = reply
        super().__init__(desc)
        self.reply = reply


class QEMUMonitorProtocol:
    """
    Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then
    allow to handle commands and events.
    """

    #: Logger object for debugging messages
    logger = logging.getLogger('QMP')

    def __init__(self, address: SocketAddrT,
                 server: bool = False,
                 nickname: Optional[str] = None):
        """
        Create a QEMUMonitorProtocol class.

        @param address: QEMU address, can be either a unix socket path (string)
                        or a tuple in the form ( address, port ) for a TCP
                        connection
        @param server: server mode listens on the socket (bool)
        @raise OSError on socket connection errors
        @note No connection is established, this is done by the connect() or
              accept() methods
        """
        self.__events: List[QMPMessage] = []
        self.__address = address
        self.__sock = self.__get_sock()
        self.__sockfile: Optional[TextIO] = None
        self._nickname = nickname
        if self._nickname:
            self.logger = logging.getLogger('QMP').getChild(self._nickname)
        if server:
            self.__sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.__sock.bind(self.__address)
            self.__sock.listen(1)

    def __get_sock(self) -> socket.socket:
        if isinstance(self.__address, tuple):
            family = socket.AF_INET
        else:
            family = socket.AF_UNIX
        return socket.socket(family, socket.SOCK_STREAM)

    def __negotiate_capabilities(self) -> QMPMessage:
        greeting = self.__json_read()
        if greeting is None or "QMP" not in greeting:
            raise QMPConnectError
        # Greeting seems ok, negotiate capabilities
        resp = self.cmd('qmp_capabilities')
        if resp and "return" in resp:
            return greeting
        raise QMPCapabilitiesError

    def __json_read(self, only_event: bool = False) -> Optional[QMPMessage]:
        assert self.__sockfile is not None
        while True:
            data = self.__sockfile.readline()
            if not data:
                return None
            # By definition, any JSON received from QMP is a QMPMessage,
            # and we are asserting only at static analysis time that it
            # has a particular shape.
            resp: QMPMessage = json.loads(data)
            if 'event' in resp:
                self.logger.debug("<<< %s", resp)
                self.__events.append(resp)
                if not only_event:
                    continue
            return resp

    def __get_events(self, wait: Union[bool, float] = False) -> None:
        """
        Check for new events in the stream and cache them in __events.

        @param wait (bool): block until an event is available.
        @param wait (float): If wait is a float, treat it as a timeout value.

        @raise QMPTimeoutError: If a timeout float is provided and the timeout
                                period elapses.
        @raise QMPConnectError: If wait is True but no events could be
                                retrieved or if some other error occurred.
        """

        # Check for new events regardless and pull them into the cache:
        self.__sock.setblocking(False)
        try:
            self.__json_read()
        except OSError as err:
            if err.errno == errno.EAGAIN:
                # No data available
                pass
        self.__sock.setblocking(True)

        # Wait for new events, if needed.
        # if wait is 0.0, this means "no wait" and is also implicitly false.
        if not self.__events and wait:
            if isinstance(wait, float):
                self.__sock.settimeout(wait)
            try:
                ret = self.__json_read(only_event=True)
            except socket.timeout:
                raise QMPTimeoutError("Timeout waiting for event")
            except:
                raise QMPConnectError("Error while reading from socket")
            if ret is None:
                raise QMPConnectError("Error while reading from socket")
            self.__sock.settimeout(None)

    def __enter__(self) -> 'QEMUMonitorProtocol':
        # Implement context manager enter function.
        return self

    def __exit__(self,
                 # pylint: disable=duplicate-code
                 # see https://github.com/PyCQA/pylint/issues/3619
                 exc_type: Optional[Type[BaseException]],
                 exc_val: Optional[BaseException],
                 exc_tb: Optional[TracebackType]) -> None:
        # Implement context manager exit function.
        self.close()

    def connect(self, negotiate: bool = True) -> Optional[QMPMessage]:
        """
        Connect to the QMP Monitor and perform capabilities negotiation.

        @return QMP greeting dict, or None if negotiate is false
        @raise OSError on socket connection errors
        @raise QMPConnectError if the greeting is not received
        @raise QMPCapabilitiesError if fails to negotiate capabilities
        """
        self.__sock.connect(self.__address)
        self.__sockfile = self.__sock.makefile(mode='r')
        if negotiate:
            return self.__negotiate_capabilities()
        return None

    def accept(self, timeout: float = 15.0) -> QMPMessage:
        """
        Await connection from QMP Monitor and perform capabilities negotiation.

        @param timeout: timeout in seconds (nonnegative float number, or
                        None). The value passed will set the behavior of the
                        underneath QMP socket as described in [1].
                        Default value is set to 15.0.
        @return QMP greeting dict
        @raise OSError on socket connection errors
        @raise QMPConnectError if the greeting is not received
        @raise QMPCapabilitiesError if fails to negotiate capabilities

        [1]
        https://docs.python.org/3/library/socket.html#socket.socket.settimeout
        """
        self.__sock.settimeout(timeout)
        self.__sock, _ = self.__sock.accept()
        self.__sockfile = self.__sock.makefile(mode='r')
        return self.__negotiate_capabilities()

    def cmd_obj(self, qmp_cmd: QMPMessage) -> QMPMessage:
        """
        Send a QMP command to the QMP Monitor.

        @param qmp_cmd: QMP command to be sent as a Python dict
        @return QMP response as a Python dict
        """
        self.logger.debug(">>> %s", qmp_cmd)
        self.__sock.sendall(json.dumps(qmp_cmd).encode('utf-8'))
        resp = self.__json_read()
        if resp is None:
            raise QMPConnectError("Unexpected empty reply from server")
        self.logger.debug("<<< %s", resp)
        return resp

    def cmd(self, name: str,
            args: Optional[Dict[str, Any]] = None,
            cmd_id: Optional[Any] = None) -> QMPMessage:
        """
        Build a QMP command and send it to the QMP Monitor.

        @param name: command name (string)
        @param args: command arguments (dict)
        @param cmd_id: command id (dict, list, string or int)
        """
        qmp_cmd: QMPMessage = {'execute': name}
        if args:
            qmp_cmd['arguments'] = args
        if cmd_id:
            qmp_cmd['id'] = cmd_id
        return self.cmd_obj(qmp_cmd)

    def command(self, cmd: str, **kwds: Any) -> QMPReturnValue:
        """
        Build and send a QMP command to the monitor, report errors if any
        """
        ret = self.cmd(cmd, kwds)
        if 'error' in ret:
            raise QMPResponseError(ret)
        if 'return' not in ret:
            raise QMPProtocolError(
                "'return' key not found in QMP response '{}'".format(str(ret))
            )
        return cast(QMPReturnValue, ret['return'])

    def pull_event(self,
                   wait: Union[bool, float] = False) -> Optional[QMPMessage]:
        """
        Pulls a single event.

        @param wait (bool): block until an event is available.
        @param wait (float): If wait is a float, treat it as a timeout value.

        @raise QMPTimeoutError: If a timeout float is provided and the timeout
                                period elapses.
        @raise QMPConnectError: If wait is True but no events could be
                                retrieved or if some other error occurred.

        @return The first available QMP event, or None.
        """
        self.__get_events(wait)

        if self.__events:
            return self.__events.pop(0)
        return None

    def get_events(self, wait: bool = False) -> List[QMPMessage]:
        """
        Get a list of available QMP events.

        @param wait (bool): block until an event is available.
        @param wait (float): If wait is a float, treat it as a timeout value.

        @raise QMPTimeoutError: If a timeout float is provided and the timeout
                                period elapses.
        @raise QMPConnectError: If wait is True but no events could be
                                retrieved or if some other error occurred.

        @return The list of available QMP events.
        """
        self.__get_events(wait)
        return self.__events

    def clear_events(self) -> None:
        """
        Clear current list of pending events.
        """
        self.__events = []

    def close(self) -> None:
        """
        Close the socket and socket file.
        """
        if self.__sock:
            self.__sock.close()
        if self.__sockfile:
            self.__sockfile.close()

    def settimeout(self, timeout: float) -> None:
        """
        Set the socket timeout.

        @param timeout (float): timeout in seconds, or None.
        @note This is a wrap around socket.settimeout
        """
        self.__sock.settimeout(timeout)

    def get_sock_fd(self) -> int:
        """
        Get the socket file descriptor.

        @return The file descriptor number.
        """
        return self.__sock.fileno()

    def is_scm_available(self) -> bool:
        """
        Check if the socket allows for SCM_RIGHTS.

        @return True if SCM_RIGHTS is available, otherwise False.
        """
        return self.__sock.family == socket.AF_UNIX