python/qemu/utils/qom_fuse.py

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