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 ( 42 IO, 43 Dict, 44 Iterator, 45 Mapping, 46 Optional, 47 Union, 48) 49 50import fuse 51from fuse import FUSE, FuseOSError, Operations 52 53 54sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python')) 55from qemu.qmp import QMPResponseError 56from qemu.qmp.qom_common import QOMCommand 57 58 59fuse.fuse_python_api = (0, 2) 60 61 62class QOMFuse(QOMCommand, Operations): 63 """ 64 QOMFuse implements both fuse.Operations and QOMCommand. 65 66 Operations implements the FS, and QOMCommand implements the CLI command. 67 """ 68 name = 'fuse' 69 help = 'Mount a QOM tree as a FUSE filesystem' 70 fuse: FUSE 71 72 @classmethod 73 def configure_parser(cls, parser: argparse.ArgumentParser) -> None: 74 super().configure_parser(parser) 75 parser.add_argument( 76 'mount', 77 metavar='<mount>', 78 action='store', 79 help="Mount point", 80 ) 81 82 def __init__(self, args: argparse.Namespace): 83 super().__init__(args) 84 self.mount = args.mount 85 self.ino_map: Dict[str, int] = {} 86 self.ino_count = 1 87 88 def run(self) -> int: 89 print(f"Mounting QOMFS to '{self.mount}'", file=sys.stderr) 90 self.fuse = FUSE(self, self.mount, foreground=True) 91 return 0 92 93 def get_ino(self, path: str) -> int: 94 """Get an inode number for a given QOM path.""" 95 if path in self.ino_map: 96 return self.ino_map[path] 97 self.ino_map[path] = self.ino_count 98 self.ino_count += 1 99 return self.ino_map[path] 100 101 def is_object(self, path: str) -> bool: 102 """Is the given QOM path an object?""" 103 try: 104 self.qom_list(path) 105 return True 106 except QMPResponseError: 107 return False 108 109 def is_property(self, path: str) -> bool: 110 """Is the given QOM path a property?""" 111 path, prop = path.rsplit('/', 1) 112 if path == '': 113 path = '/' 114 try: 115 for item in self.qom_list(path): 116 if item.name == prop: 117 return True 118 return False 119 except QMPResponseError: 120 return False 121 122 def is_link(self, path: str) -> bool: 123 """Is the given QOM path a link?""" 124 path, prop = path.rsplit('/', 1) 125 if path == '': 126 path = '/' 127 try: 128 for item in self.qom_list(path): 129 if item.name == prop and item.link: 130 return True 131 return False 132 except QMPResponseError: 133 return False 134 135 def read(self, path: str, size: int, offset: int, fh: IO[bytes]) -> bytes: 136 if not self.is_property(path): 137 raise FuseOSError(ENOENT) 138 139 path, prop = path.rsplit('/', 1) 140 if path == '': 141 path = '/' 142 try: 143 data = str(self.qmp.command('qom-get', path=path, property=prop)) 144 data += '\n' # make values shell friendly 145 except QMPResponseError as err: 146 raise FuseOSError(EPERM) from err 147 148 if offset > len(data): 149 return b'' 150 151 return bytes(data[offset:][:size], encoding='utf-8') 152 153 def readlink(self, path: str) -> Union[bool, str]: 154 if not self.is_link(path): 155 return False 156 path, prop = path.rsplit('/', 1) 157 prefix = '/'.join(['..'] * (len(path.split('/')) - 1)) 158 return prefix + str(self.qmp.command('qom-get', path=path, 159 property=prop)) 160 161 def getattr(self, path: str, 162 fh: Optional[IO[bytes]] = None) -> Mapping[str, object]: 163 if self.is_link(path): 164 value = { 165 'st_mode': 0o755 | stat.S_IFLNK, 166 'st_ino': self.get_ino(path), 167 'st_dev': 0, 168 'st_nlink': 2, 169 'st_uid': 1000, 170 'st_gid': 1000, 171 'st_size': 4096, 172 'st_atime': 0, 173 'st_mtime': 0, 174 'st_ctime': 0 175 } 176 elif self.is_object(path): 177 value = { 178 'st_mode': 0o755 | stat.S_IFDIR, 179 'st_ino': self.get_ino(path), 180 'st_dev': 0, 181 'st_nlink': 2, 182 'st_uid': 1000, 183 'st_gid': 1000, 184 'st_size': 4096, 185 'st_atime': 0, 186 'st_mtime': 0, 187 'st_ctime': 0 188 } 189 elif self.is_property(path): 190 value = { 191 'st_mode': 0o644 | stat.S_IFREG, 192 'st_ino': self.get_ino(path), 193 'st_dev': 0, 194 'st_nlink': 1, 195 'st_uid': 1000, 196 'st_gid': 1000, 197 'st_size': 4096, 198 'st_atime': 0, 199 'st_mtime': 0, 200 'st_ctime': 0 201 } 202 else: 203 raise FuseOSError(ENOENT) 204 return value 205 206 def readdir(self, path: str, fh: IO[bytes]) -> Iterator[str]: 207 yield '.' 208 yield '..' 209 for item in self.qom_list(path): 210 yield item.name 211 212 213if __name__ == '__main__': 214 sys.exit(QOMFuse.entry_point()) 215