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.qom_list(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.qom_list(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.qom_list(path): 122 if item.name == prop and item.link: 123 return True 124 return False 125 except QMPResponseError: 126 return False 127 128 def read(self, path, size, offset, fh): 129 if not self.is_property(path): 130 return -ENOENT 131 132 path, prop = path.rsplit('/', 1) 133 if path == '': 134 path = '/' 135 try: 136 data = self.qmp.command('qom-get', path=path, property=prop) 137 data += '\n' # make values shell friendly 138 except QMPResponseError as err: 139 raise FuseOSError(EPERM) from err 140 141 if offset > len(data): 142 return '' 143 144 return bytes(data[offset:][:size], encoding='utf-8') 145 146 def readlink(self, path): 147 if not self.is_link(path): 148 return False 149 path, prop = path.rsplit('/', 1) 150 prefix = '/'.join(['..'] * (len(path.split('/')) - 1)) 151 return prefix + str(self.qmp.command('qom-get', path=path, 152 property=prop)) 153 154 def getattr(self, path, fh=None): 155 if self.is_link(path): 156 value = { 157 'st_mode': 0o755 | stat.S_IFLNK, 158 'st_ino': self.get_ino(path), 159 'st_dev': 0, 160 'st_nlink': 2, 161 'st_uid': 1000, 162 'st_gid': 1000, 163 'st_size': 4096, 164 'st_atime': 0, 165 'st_mtime': 0, 166 'st_ctime': 0 167 } 168 elif self.is_object(path): 169 value = { 170 'st_mode': 0o755 | stat.S_IFDIR, 171 'st_ino': self.get_ino(path), 172 'st_dev': 0, 173 'st_nlink': 2, 174 'st_uid': 1000, 175 'st_gid': 1000, 176 'st_size': 4096, 177 'st_atime': 0, 178 'st_mtime': 0, 179 'st_ctime': 0 180 } 181 elif self.is_property(path): 182 value = { 183 'st_mode': 0o644 | stat.S_IFREG, 184 'st_ino': self.get_ino(path), 185 'st_dev': 0, 186 'st_nlink': 1, 187 'st_uid': 1000, 188 'st_gid': 1000, 189 'st_size': 4096, 190 'st_atime': 0, 191 'st_mtime': 0, 192 'st_ctime': 0 193 } 194 else: 195 raise FuseOSError(ENOENT) 196 return value 197 198 def readdir(self, path, fh): 199 yield '.' 200 yield '..' 201 for item in self.qom_list(path): 202 yield item.name 203 204 205if __name__ == '__main__': 206 sys.exit(QOMFuse.entry_point()) 207