1#!/usr/bin/env python3 2""" 3QEMU Object Model FUSE filesystem tool 4 5This script offers a simple FUSE filesystem within which the QOM tree 6may be browsed, queried and edited using traditional shell tooling. 7 8This script requires the 'fusepy' python package. 9 10 11usage: qom-fuse [-h] [--socket SOCKET] <mount> 12 13Mount a QOM tree as a FUSE filesystem 14 15positional arguments: 16 <mount> Mount point 17 18optional arguments: 19 -h, --help show this help message and exit 20 --socket SOCKET, -s SOCKET 21 QMP socket path or address (addr:port). May also be 22 set via QMP_SOCKET environment variable. 23""" 24## 25# Copyright IBM, Corp. 2012 26# Copyright (C) 2020 Red Hat, Inc. 27# 28# Authors: 29# Anthony Liguori <aliguori@us.ibm.com> 30# Markus Armbruster <armbru@redhat.com> 31# 32# This work is licensed under the terms of the GNU GPL, version 2 or later. 33# See the COPYING file in the top-level directory. 34## 35 36import argparse 37from errno import ENOENT, EPERM 38import os 39import stat 40import sys 41from typing import Dict 42 43import fuse 44from fuse import FUSE, FuseOSError, Operations 45 46 47sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python')) 48from qemu.qmp import QMPResponseError 49from qemu.qmp.qom_common import QOMCommand 50 51 52fuse.fuse_python_api = (0, 2) 53 54 55class QOMFuse(QOMCommand, Operations): 56 """ 57 QOMFuse implements both fuse.Operations and QOMCommand. 58 59 Operations implements the FS, and QOMCommand implements the CLI command. 60 """ 61 name = 'fuse' 62 help = 'Mount a QOM tree as a FUSE filesystem' 63 fuse: FUSE 64 65 @classmethod 66 def configure_parser(cls, parser: argparse.ArgumentParser) -> None: 67 super().configure_parser(parser) 68 parser.add_argument( 69 'mount', 70 metavar='<mount>', 71 action='store', 72 help="Mount point", 73 ) 74 75 def __init__(self, args: argparse.Namespace): 76 super().__init__(args) 77 self.mount = args.mount 78 self.ino_map: Dict[str, int] = {} 79 self.ino_count = 1 80 81 def run(self) -> int: 82 print(f"Mounting QOMFS to '{self.mount}'", file=sys.stderr) 83 self.fuse = FUSE(self, self.mount, foreground=True) 84 return 0 85 86 def get_ino(self, path): 87 """Get an inode number for a given QOM path.""" 88 if path in self.ino_map: 89 return self.ino_map[path] 90 self.ino_map[path] = self.ino_count 91 self.ino_count += 1 92 return self.ino_map[path] 93 94 def is_object(self, path): 95 """Is the given QOM path an object?""" 96 try: 97 self.qmp.command('qom-list', path=path) 98 return True 99 except QMPResponseError: 100 return False 101 102 def is_property(self, path): 103 """Is the given QOM path a property?""" 104 path, prop = path.rsplit('/', 1) 105 if path == '': 106 path = '/' 107 try: 108 for item in self.qmp.command('qom-list', path=path): 109 if item['name'] == prop: 110 return True 111 return False 112 except QMPResponseError: 113 return False 114 115 def is_link(self, path): 116 """Is the given QOM path a link?""" 117 path, prop = path.rsplit('/', 1) 118 if path == '': 119 path = '/' 120 try: 121 for item in self.qmp.command('qom-list', path=path): 122 if item['name'] == prop: 123 if item['type'].startswith('link<'): 124 return True 125 return False 126 return False 127 except QMPResponseError: 128 return False 129 130 def read(self, path, size, offset, fh): 131 if not self.is_property(path): 132 return -ENOENT 133 134 path, prop = path.rsplit('/', 1) 135 if path == '': 136 path = '/' 137 try: 138 data = self.qmp.command('qom-get', path=path, property=prop) 139 data += '\n' # make values shell friendly 140 except QMPResponseError as err: 141 raise FuseOSError(EPERM) from err 142 143 if offset > len(data): 144 return '' 145 146 return bytes(data[offset:][:size], encoding='utf-8') 147 148 def readlink(self, path): 149 if not self.is_link(path): 150 return False 151 path, prop = path.rsplit('/', 1) 152 prefix = '/'.join(['..'] * (len(path.split('/')) - 1)) 153 return prefix + str(self.qmp.command('qom-get', path=path, 154 property=prop)) 155 156 def getattr(self, path, fh=None): 157 if self.is_link(path): 158 value = { 159 'st_mode': 0o755 | stat.S_IFLNK, 160 'st_ino': self.get_ino(path), 161 'st_dev': 0, 162 'st_nlink': 2, 163 'st_uid': 1000, 164 'st_gid': 1000, 165 'st_size': 4096, 166 'st_atime': 0, 167 'st_mtime': 0, 168 'st_ctime': 0 169 } 170 elif self.is_object(path): 171 value = { 172 'st_mode': 0o755 | stat.S_IFDIR, 173 'st_ino': self.get_ino(path), 174 'st_dev': 0, 175 'st_nlink': 2, 176 'st_uid': 1000, 177 'st_gid': 1000, 178 'st_size': 4096, 179 'st_atime': 0, 180 'st_mtime': 0, 181 'st_ctime': 0 182 } 183 elif self.is_property(path): 184 value = { 185 'st_mode': 0o644 | stat.S_IFREG, 186 'st_ino': self.get_ino(path), 187 'st_dev': 0, 188 'st_nlink': 1, 189 'st_uid': 1000, 190 'st_gid': 1000, 191 'st_size': 4096, 192 'st_atime': 0, 193 'st_mtime': 0, 194 'st_ctime': 0 195 } 196 else: 197 raise FuseOSError(ENOENT) 198 return value 199 200 def readdir(self, path, fh): 201 yield '.' 202 yield '..' 203 for item in self.qmp.command('qom-list', path=path): 204 yield str(item['name']) 205 206 207if __name__ == '__main__': 208 sys.exit(QOMFuse.entry_point()) 209