""" QEMU Guest Agent Client Usage: Start QEMU with: # qemu [...] -chardev socket,path=/tmp/qga.sock,server=on,wait=off,id=qga0 \ -device virtio-serial \ -device virtserialport,chardev=qga0,name=org.qemu.guest_agent.0 Run the script: $ qemu-ga-client --address=/tmp/qga.sock [args...] or $ export QGA_CLIENT_ADDRESS=/tmp/qga.sock $ qemu-ga-client [args...] For example: $ qemu-ga-client cat /etc/resolv.conf # Generated by NetworkManager nameserver 10.0.2.3 $ qemu-ga-client fsfreeze status thawed $ qemu-ga-client fsfreeze freeze 2 filesystems frozen See also: https://wiki.qemu.org/Features/QAPI/GuestAgent """ # Copyright (C) 2012 Ryota Ozaki # # This work is licensed under the terms of the GNU GPL, version 2. See # the COPYING file in the top-level directory. import argparse import asyncio import base64 import os import random import sys from typing import ( Any, Callable, Dict, Optional, Sequence, ) from qemu.qmp import ConnectError, SocketAddrT from qemu.qmp.legacy import QEMUMonitorProtocol # This script has not seen many patches or careful attention in quite # some time. If you would like to improve it, please review the design # carefully and add docstrings at that point in time. Until then: # pylint: disable=missing-docstring class QemuGuestAgent(QEMUMonitorProtocol): def __getattr__(self, name: str) -> Callable[..., Any]: def wrapper(**kwds: object) -> object: return self.command('guest-' + name.replace('_', '-'), **kwds) return wrapper class QemuGuestAgentClient: def __init__(self, address: SocketAddrT): self.qga = QemuGuestAgent(address) self.qga.connect(negotiate=False) def sync(self, timeout: Optional[float] = 3) -> None: # Avoid being blocked forever if not self.ping(timeout): raise EnvironmentError('Agent seems not alive') uid = random.randint(0, (1 << 32) - 1) while True: ret = self.qga.sync(id=uid) if isinstance(ret, int) and int(ret) == uid: break def __file_read_all(self, handle: int) -> bytes: eof = False data = b'' while not eof: ret = self.qga.file_read(handle=handle, count=1024) _data = base64.b64decode(ret['buf-b64']) data += _data eof = ret['eof'] return data def read(self, path: str) -> bytes: handle = self.qga.file_open(path=path) try: data = self.__file_read_all(handle) finally: self.qga.file_close(handle=handle) return data def info(self) -> str: info = self.qga.info() msgs = [] msgs.append('version: ' + info['version']) msgs.append('supported_commands:') enabled = [c['name'] for c in info['supported_commands'] if c['enabled']] msgs.append('\tenabled: ' + ', '.join(enabled)) disabled = [c['name'] for c in info['supported_commands'] if not c['enabled']] msgs.append('\tdisabled: ' + ', '.join(disabled)) return '\n'.join(msgs) @classmethod def __gen_ipv4_netmask(cls, prefixlen: int) -> str: mask = int('1' * prefixlen + '0' * (32 - prefixlen), 2) return '.'.join([str(mask >> 24), str((mask >> 16) & 0xff), str((mask >> 8) & 0xff), str(mask & 0xff)]) def ifconfig(self) -> str: nifs = self.qga.network_get_interfaces() msgs = [] for nif in nifs: msgs.append(nif['name'] + ':') if 'ip-addresses' in nif: for ipaddr in nif['ip-addresses']: if ipaddr['ip-address-type'] == 'ipv4': addr = ipaddr['ip-address'] mask = self.__gen_ipv4_netmask(int(ipaddr['prefix'])) msgs.append(f"\tinet {addr} netmask {mask}") elif ipaddr['ip-address-type'] == 'ipv6': addr = ipaddr['ip-address'] prefix = ipaddr['prefix'] msgs.append(f"\tinet6 {addr} prefixlen {prefix}") if nif['hardware-address'] != '00:00:00:00:00:00': msgs.append("\tether " + nif['hardware-address']) return '\n'.join(msgs) def ping(self, timeout: Optional[float]) -> bool: self.qga.settimeout(timeout) try: self.qga.ping() except asyncio.TimeoutError: return False return True def fsfreeze(self, cmd: str) -> object: if cmd not in ['status', 'freeze', 'thaw']: raise Exception('Invalid command: ' + cmd) # Can be int (freeze, thaw) or GuestFsfreezeStatus (status) return getattr(self.qga, 'fsfreeze' + '_' + cmd)() def fstrim(self, minimum: int) -> Dict[str, object]: # returns GuestFilesystemTrimResponse ret = getattr(self.qga, 'fstrim')(minimum=minimum) assert isinstance(ret, dict) return ret def suspend(self, mode: str) -> None: if mode not in ['disk', 'ram', 'hybrid']: raise Exception('Invalid mode: ' + mode) try: getattr(self.qga, 'suspend' + '_' + mode)() # On error exception will raise except asyncio.TimeoutError: # On success command will timed out return def shutdown(self, mode: str = 'powerdown') -> None: if mode not in ['powerdown', 'halt', 'reboot']: raise Exception('Invalid mode: ' + mode) try: self.qga.shutdown(mode=mode) except asyncio.TimeoutError: pass def _cmd_cat(client: QemuGuestAgentClient, args: Sequence[str]) -> None: if len(args) != 1: print('Invalid argument') print('Usage: cat ') sys.exit(1) print(client.read(args[0])) def _cmd_fsfreeze(client: QemuGuestAgentClient, args: Sequence[str]) -> None: usage = 'Usage: fsfreeze status|freeze|thaw' if len(args) != 1: print('Invalid argument') print(usage) sys.exit(1) if args[0] not in ['status', 'freeze', 'thaw']: print('Invalid command: ' + args[0]) print(usage) sys.exit(1) cmd = args[0] ret = client.fsfreeze(cmd) if cmd == 'status': print(ret) return assert isinstance(ret, int) verb = 'frozen' if cmd == 'freeze' else 'thawed' print(f"{ret:d} filesystems {verb}") def _cmd_fstrim(client: QemuGuestAgentClient, args: Sequence[str]) -> None: if len(args) == 0: minimum = 0 else: minimum = int(args[0]) print(client.fstrim(minimum)) def _cmd_ifconfig(client: QemuGuestAgentClient, args: Sequence[str]) -> None: assert not args print(client.ifconfig()) def _cmd_info(client: QemuGuestAgentClient, args: Sequence[str]) -> None: assert not args print(client.info()) def _cmd_ping(client: QemuGuestAgentClient, args: Sequence[str]) -> None: timeout = 3.0 if len(args) == 0 else float(args[0]) alive = client.ping(timeout) if not alive: print("Not responded in %s sec" % args[0]) sys.exit(1) def _cmd_suspend(client: QemuGuestAgentClient, args: Sequence[str]) -> None: usage = 'Usage: suspend disk|ram|hybrid' if len(args) != 1: print('Less argument') print(usage) sys.exit(1) if args[0] not in ['disk', 'ram', 'hybrid']: print('Invalid command: ' + args[0]) print(usage) sys.exit(1) client.suspend(args[0]) def _cmd_shutdown(client: QemuGuestAgentClient, args: Sequence[str]) -> None: assert not args client.shutdown() _cmd_powerdown = _cmd_shutdown def _cmd_halt(client: QemuGuestAgentClient, args: Sequence[str]) -> None: assert not args client.shutdown('halt') def _cmd_reboot(client: QemuGuestAgentClient, args: Sequence[str]) -> None: assert not args client.shutdown('reboot') commands = [m.replace('_cmd_', '') for m in dir() if '_cmd_' in m] def send_command(address: str, cmd: str, args: Sequence[str]) -> None: if not os.path.exists(address): print(f"'{address}' not found. (Is QEMU running?)") sys.exit(1) if cmd not in commands: print('Invalid command: ' + cmd) print('Available commands: ' + ', '.join(commands)) sys.exit(1) try: client = QemuGuestAgentClient(address) except ConnectError as err: print(err) if isinstance(err.exc, ConnectionError): print('(Is QEMU running?)') sys.exit(1) if cmd == 'fsfreeze' and args[0] == 'freeze': client.sync(60) elif cmd != 'ping': client.sync() globals()['_cmd_' + cmd](client, args) def main() -> None: address = os.environ.get('QGA_CLIENT_ADDRESS') parser = argparse.ArgumentParser() parser.add_argument('--address', action='store', default=address, help='Specify a ip:port pair or a unix socket path') parser.add_argument('command', choices=commands) parser.add_argument('args', nargs='*') args = parser.parse_args() if args.address is None: parser.error('address is not specified') sys.exit(1) send_command(args.address, args.command, args.args) if __name__ == '__main__': main()