scsi: esp: check buffer length before reading scsi command
[qemu.git] / scripts / qtest.py
1 # QEMU qtest library
2 #
3 # Copyright (C) 2015 Red Hat Inc.
4 #
5 # Authors:
6 # Fam Zheng <famz@redhat.com>
7 #
8 # This work is licensed under the terms of the GNU GPL, version 2. See
9 # the COPYING file in the top-level directory.
10 #
11 # Based on qmp.py.
12 #
13
14 import errno
15 import socket
16
17 class QEMUQtestProtocol(object):
18 def __init__(self, address, server=False):
19 """
20 Create a QEMUQtestProtocol object.
21
22 @param address: QEMU address, can be either a unix socket path (string)
23 or a tuple in the form ( address, port ) for a TCP
24 connection
25 @param server: server mode, listens on the socket (bool)
26 @raise socket.error on socket connection errors
27 @note No connection is established, this is done by the connect() or
28 accept() methods
29 """
30 self._address = address
31 self._sock = self._get_sock()
32 if server:
33 self._sock.bind(self._address)
34 self._sock.listen(1)
35
36 def _get_sock(self):
37 if isinstance(self._address, tuple):
38 family = socket.AF_INET
39 else:
40 family = socket.AF_UNIX
41 return socket.socket(family, socket.SOCK_STREAM)
42
43 def connect(self):
44 """
45 Connect to the qtest socket.
46
47 @raise socket.error on socket connection errors
48 """
49 self._sock.connect(self._address)
50
51 def accept(self):
52 """
53 Await connection from QEMU.
54
55 @raise socket.error on socket connection errors
56 """
57 self._sock, _ = self._sock.accept()
58
59 def cmd(self, qtest_cmd):
60 """
61 Send a qtest command on the wire.
62
63 @param qtest_cmd: qtest command text to be sent
64 """
65 self._sock.sendall(qtest_cmd + "\n")
66
67 def close(self):
68 self._sock.close()
69
70 def settimeout(self, timeout):
71 self._sock.settimeout(timeout)