11217a757e
In which qmp-shell will exclusively use the HMP passthrough feature,
this is useful for testing.
Example:
# ./qmp-shell -H qmp-sock
Welcome to the HMP shell!
Connected to QEMU 0.13.50
(QEMU) info network
VLAN 0 devices:
user.0: net=10.0.2.0, restricted=n
e1000.0: model=e1000,macaddr=52:54:00:12:34:56
Devices not on any VLAN:
(QEMU)
Signed-off-by: Luiz Capitulino <lcapitulino@redhat.com>
260 lines
7.7 KiB
Python
Executable File
260 lines
7.7 KiB
Python
Executable File
#!/usr/bin/python
|
|
#
|
|
# Low-level QEMU shell on top of QMP.
|
|
#
|
|
# 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.
|
|
#
|
|
# Usage:
|
|
#
|
|
# Start QEMU with:
|
|
#
|
|
# # qemu [...] -qmp unix:./qmp-sock,server
|
|
#
|
|
# Run the shell:
|
|
#
|
|
# $ qmp-shell ./qmp-sock
|
|
#
|
|
# Commands have the following format:
|
|
#
|
|
# < command-name > [ arg-name1=arg1 ] ... [ arg-nameN=argN ]
|
|
#
|
|
# For example:
|
|
#
|
|
# (QEMU) device_add driver=e1000 id=net1
|
|
# {u'return': {}}
|
|
# (QEMU)
|
|
|
|
import qmp
|
|
import readline
|
|
import sys
|
|
|
|
class QMPCompleter(list):
|
|
def complete(self, text, state):
|
|
for cmd in self:
|
|
if cmd.startswith(text):
|
|
if not state:
|
|
return cmd
|
|
else:
|
|
state -= 1
|
|
|
|
class QMPShellError(Exception):
|
|
pass
|
|
|
|
class QMPShellBadPort(QMPShellError):
|
|
pass
|
|
|
|
# TODO: QMPShell's interface is a bit ugly (eg. _fill_completion() and
|
|
# _execute_cmd()). Let's design a better one.
|
|
class QMPShell(qmp.QEMUMonitorProtocol):
|
|
def __init__(self, address):
|
|
qmp.QEMUMonitorProtocol.__init__(self, self.__get_address(address))
|
|
self._greeting = None
|
|
self._completer = None
|
|
|
|
def __get_address(self, arg):
|
|
"""
|
|
Figure out if the argument is in the port:host form, if it's not it's
|
|
probably a file path.
|
|
"""
|
|
addr = arg.split(':')
|
|
if len(addr) == 2:
|
|
try:
|
|
port = int(addr[1])
|
|
except ValueError:
|
|
raise QMPShellBadPort
|
|
return ( addr[0], port )
|
|
# socket path
|
|
return arg
|
|
|
|
def _fill_completion(self):
|
|
for cmd in self.cmd('query-commands')['return']:
|
|
self._completer.append(cmd['name'])
|
|
|
|
def __completer_setup(self):
|
|
self._completer = QMPCompleter()
|
|
self._fill_completion()
|
|
readline.set_completer(self._completer.complete)
|
|
readline.parse_and_bind("tab: complete")
|
|
# XXX: default delimiters conflict with some command names (eg. query-),
|
|
# clearing everything as it doesn't seem to matter
|
|
readline.set_completer_delims('')
|
|
|
|
def __build_cmd(self, cmdline):
|
|
"""
|
|
Build a QMP input object from a user provided command-line in the
|
|
following format:
|
|
|
|
< command-name > [ arg-name1=arg1 ] ... [ arg-nameN=argN ]
|
|
"""
|
|
cmdargs = cmdline.split()
|
|
qmpcmd = { 'execute': cmdargs[0], 'arguments': {} }
|
|
for arg in cmdargs[1:]:
|
|
opt = arg.split('=')
|
|
try:
|
|
value = int(opt[1])
|
|
except ValueError:
|
|
value = opt[1]
|
|
qmpcmd['arguments'][opt[0]] = value
|
|
return qmpcmd
|
|
|
|
def _execute_cmd(self, cmdline):
|
|
try:
|
|
qmpcmd = self.__build_cmd(cmdline)
|
|
except:
|
|
print 'command format: <command-name> ',
|
|
print '[arg-name1=arg1] ... [arg-nameN=argN]'
|
|
return True
|
|
resp = self.cmd_obj(qmpcmd)
|
|
if resp is None:
|
|
print 'Disconnected'
|
|
return False
|
|
print resp
|
|
return True
|
|
|
|
def connect(self):
|
|
self._greeting = qmp.QEMUMonitorProtocol.connect(self)
|
|
self.__completer_setup()
|
|
|
|
def show_banner(self, msg='Welcome to the QMP low-level shell!'):
|
|
print msg
|
|
version = self._greeting['QMP']['version']['qemu']
|
|
print 'Connected to QEMU %d.%d.%d\n' % (version['major'],version['minor'],version['micro'])
|
|
|
|
def read_exec_command(self, prompt):
|
|
"""
|
|
Read and execute a command.
|
|
|
|
@return True if execution was ok, return False if disconnected.
|
|
"""
|
|
try:
|
|
cmdline = raw_input(prompt)
|
|
except EOFError:
|
|
print
|
|
return False
|
|
if cmdline == '':
|
|
for ev in self.get_events():
|
|
print ev
|
|
self.clear_events()
|
|
return True
|
|
else:
|
|
return self._execute_cmd(cmdline)
|
|
|
|
class HMPShell(QMPShell):
|
|
def __init__(self, address):
|
|
QMPShell.__init__(self, address)
|
|
self.__cpu_index = 0
|
|
|
|
def __cmd_completion(self):
|
|
for cmd in self.__cmd_passthrough('help')['return'].split('\r\n'):
|
|
if cmd and cmd[0] != '[' and cmd[0] != '\t':
|
|
name = cmd.split()[0] # drop help text
|
|
if name == 'info':
|
|
continue
|
|
if name.find('|') != -1:
|
|
# Command in the form 'foobar|f' or 'f|foobar', take the
|
|
# full name
|
|
opt = name.split('|')
|
|
if len(opt[0]) == 1:
|
|
name = opt[1]
|
|
else:
|
|
name = opt[0]
|
|
self._completer.append(name)
|
|
self._completer.append('help ' + name) # help completion
|
|
|
|
def __info_completion(self):
|
|
for cmd in self.__cmd_passthrough('info')['return'].split('\r\n'):
|
|
if cmd:
|
|
self._completer.append('info ' + cmd.split()[1])
|
|
|
|
def __other_completion(self):
|
|
# special cases
|
|
self._completer.append('help info')
|
|
|
|
def _fill_completion(self):
|
|
self.__cmd_completion()
|
|
self.__info_completion()
|
|
self.__other_completion()
|
|
|
|
def __cmd_passthrough(self, cmdline, cpu_index = 0):
|
|
return self.cmd_obj({ 'execute': 'human-monitor-command', 'arguments':
|
|
{ 'command-line': cmdline,
|
|
'cpu-index': cpu_index } })
|
|
|
|
def _execute_cmd(self, cmdline):
|
|
if cmdline.split()[0] == "cpu":
|
|
# trap the cpu command, it requires special setting
|
|
try:
|
|
idx = int(cmdline.split()[1])
|
|
if not 'return' in self.__cmd_passthrough('info version', idx):
|
|
print 'bad CPU index'
|
|
return True
|
|
self.__cpu_index = idx
|
|
except ValueError:
|
|
print 'cpu command takes an integer argument'
|
|
return True
|
|
resp = self.__cmd_passthrough(cmdline, self.__cpu_index)
|
|
if resp is None:
|
|
print 'Disconnected'
|
|
return False
|
|
assert 'return' in resp or 'error' in resp
|
|
if 'return' in resp:
|
|
# Success
|
|
if len(resp['return']) > 0:
|
|
print resp['return'],
|
|
else:
|
|
# Error
|
|
print '%s: %s' % (resp['error']['class'], resp['error']['desc'])
|
|
return True
|
|
|
|
def show_banner(self):
|
|
QMPShell.show_banner(self, msg='Welcome to the HMP shell!')
|
|
|
|
def die(msg):
|
|
sys.stderr.write('ERROR: %s\n' % msg)
|
|
sys.exit(1)
|
|
|
|
def fail_cmdline(option=None):
|
|
if option:
|
|
sys.stderr.write('ERROR: bad command-line option \'%s\'\n' % option)
|
|
sys.stderr.write('qemu-shell [ -H ] < UNIX socket path> | < TCP address:port >\n')
|
|
sys.exit(1)
|
|
|
|
def main():
|
|
addr = ''
|
|
try:
|
|
if len(sys.argv) == 2:
|
|
qemu = QMPShell(sys.argv[1])
|
|
addr = sys.argv[1]
|
|
elif len(sys.argv) == 3:
|
|
if sys.argv[1] != '-H':
|
|
fail_cmdline(sys.argv[1])
|
|
qemu = HMPShell(sys.argv[2])
|
|
addr = sys.argv[2]
|
|
else:
|
|
fail_cmdline()
|
|
except QMPShellBadPort:
|
|
die('bad port number in command-line')
|
|
|
|
try:
|
|
qemu.connect()
|
|
except qmp.QMPConnectError:
|
|
die('Didn\'t get QMP greeting message')
|
|
except qmp.QMPCapabilitiesError:
|
|
die('Could not negotiate capabilities')
|
|
except qemu.error:
|
|
die('Could not connect to %s' % addr)
|
|
|
|
qemu.show_banner()
|
|
while qemu.read_exec_command('(QEMU) '):
|
|
pass
|
|
qemu.close()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|