Because fusepy does not have type hints, add some targeted warning suppressions. Namely, we need to allow subclassing something of an unknown type (in qom_fuse.py), and we need to allow missing imports (recorded against fuse itself) because mypy will be unable to import fusepy (even when installed) as it has no types nor type stubs available. Note: Until now, it was possible to run invocations like 'mypy qemu/' from ./python and have that work. However, these targeted suppressions require that you run 'mypy -p qemu/' instead. The correct, canonical invocation is recorded in ./python/tests/mypy.sh and all of the various CI invocations always use this correct form. Signed-off-by: John Snow <jsnow@redhat.com> Message-id: 20210603003719.1321369-16-jsnow@redhat.com Signed-off-by: John Snow <jsnow@redhat.com>
215 lines
6 KiB
Python
Executable file
215 lines
6 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
QEMU Object Model FUSE filesystem tool
|
|
|
|
This script offers a simple FUSE filesystem within which the QOM tree
|
|
may be browsed, queried and edited using traditional shell tooling.
|
|
|
|
This script requires the 'fusepy' python package.
|
|
|
|
|
|
usage: qom-fuse [-h] [--socket SOCKET] <mount>
|
|
|
|
Mount a QOM tree as a FUSE filesystem
|
|
|
|
positional arguments:
|
|
<mount> Mount point
|
|
|
|
optional arguments:
|
|
-h, --help show this help message and exit
|
|
--socket SOCKET, -s SOCKET
|
|
QMP socket path or address (addr:port). May also be
|
|
set via QMP_SOCKET environment variable.
|
|
"""
|
|
##
|
|
# Copyright IBM, Corp. 2012
|
|
# Copyright (C) 2020 Red Hat, Inc.
|
|
#
|
|
# Authors:
|
|
# Anthony Liguori <aliguori@us.ibm.com>
|
|
# Markus Armbruster <armbru@redhat.com>
|
|
#
|
|
# This work is licensed under the terms of the GNU GPL, version 2 or later.
|
|
# See the COPYING file in the top-level directory.
|
|
##
|
|
|
|
import argparse
|
|
from errno import ENOENT, EPERM
|
|
import os
|
|
import stat
|
|
import sys
|
|
from typing import (
|
|
IO,
|
|
Dict,
|
|
Iterator,
|
|
Mapping,
|
|
Optional,
|
|
Union,
|
|
)
|
|
|
|
import fuse
|
|
from fuse import FUSE, FuseOSError, Operations
|
|
|
|
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
|
|
from qemu.qmp import QMPResponseError
|
|
from qemu.qmp.qom_common import QOMCommand
|
|
|
|
|
|
fuse.fuse_python_api = (0, 2)
|
|
|
|
|
|
class QOMFuse(QOMCommand, Operations):
|
|
"""
|
|
QOMFuse implements both fuse.Operations and QOMCommand.
|
|
|
|
Operations implements the FS, and QOMCommand implements the CLI command.
|
|
"""
|
|
name = 'fuse'
|
|
help = 'Mount a QOM tree as a FUSE filesystem'
|
|
fuse: FUSE
|
|
|
|
@classmethod
|
|
def configure_parser(cls, parser: argparse.ArgumentParser) -> None:
|
|
super().configure_parser(parser)
|
|
parser.add_argument(
|
|
'mount',
|
|
metavar='<mount>',
|
|
action='store',
|
|
help="Mount point",
|
|
)
|
|
|
|
def __init__(self, args: argparse.Namespace):
|
|
super().__init__(args)
|
|
self.mount = args.mount
|
|
self.ino_map: Dict[str, int] = {}
|
|
self.ino_count = 1
|
|
|
|
def run(self) -> int:
|
|
print(f"Mounting QOMFS to '{self.mount}'", file=sys.stderr)
|
|
self.fuse = FUSE(self, self.mount, foreground=True)
|
|
return 0
|
|
|
|
def get_ino(self, path: str) -> int:
|
|
"""Get an inode number for a given QOM path."""
|
|
if path in self.ino_map:
|
|
return self.ino_map[path]
|
|
self.ino_map[path] = self.ino_count
|
|
self.ino_count += 1
|
|
return self.ino_map[path]
|
|
|
|
def is_object(self, path: str) -> bool:
|
|
"""Is the given QOM path an object?"""
|
|
try:
|
|
self.qom_list(path)
|
|
return True
|
|
except QMPResponseError:
|
|
return False
|
|
|
|
def is_property(self, path: str) -> bool:
|
|
"""Is the given QOM path a property?"""
|
|
path, prop = path.rsplit('/', 1)
|
|
if path == '':
|
|
path = '/'
|
|
try:
|
|
for item in self.qom_list(path):
|
|
if item.name == prop:
|
|
return True
|
|
return False
|
|
except QMPResponseError:
|
|
return False
|
|
|
|
def is_link(self, path: str) -> bool:
|
|
"""Is the given QOM path a link?"""
|
|
path, prop = path.rsplit('/', 1)
|
|
if path == '':
|
|
path = '/'
|
|
try:
|
|
for item in self.qom_list(path):
|
|
if item.name == prop and item.link:
|
|
return True
|
|
return False
|
|
except QMPResponseError:
|
|
return False
|
|
|
|
def read(self, path: str, size: int, offset: int, fh: IO[bytes]) -> bytes:
|
|
if not self.is_property(path):
|
|
raise FuseOSError(ENOENT)
|
|
|
|
path, prop = path.rsplit('/', 1)
|
|
if path == '':
|
|
path = '/'
|
|
try:
|
|
data = str(self.qmp.command('qom-get', path=path, property=prop))
|
|
data += '\n' # make values shell friendly
|
|
except QMPResponseError as err:
|
|
raise FuseOSError(EPERM) from err
|
|
|
|
if offset > len(data):
|
|
return b''
|
|
|
|
return bytes(data[offset:][:size], encoding='utf-8')
|
|
|
|
def readlink(self, path: str) -> Union[bool, str]:
|
|
if not self.is_link(path):
|
|
return False
|
|
path, prop = path.rsplit('/', 1)
|
|
prefix = '/'.join(['..'] * (len(path.split('/')) - 1))
|
|
return prefix + str(self.qmp.command('qom-get', path=path,
|
|
property=prop))
|
|
|
|
def getattr(self, path: str,
|
|
fh: Optional[IO[bytes]] = None) -> Mapping[str, object]:
|
|
if self.is_link(path):
|
|
value = {
|
|
'st_mode': 0o755 | stat.S_IFLNK,
|
|
'st_ino': self.get_ino(path),
|
|
'st_dev': 0,
|
|
'st_nlink': 2,
|
|
'st_uid': 1000,
|
|
'st_gid': 1000,
|
|
'st_size': 4096,
|
|
'st_atime': 0,
|
|
'st_mtime': 0,
|
|
'st_ctime': 0
|
|
}
|
|
elif self.is_object(path):
|
|
value = {
|
|
'st_mode': 0o755 | stat.S_IFDIR,
|
|
'st_ino': self.get_ino(path),
|
|
'st_dev': 0,
|
|
'st_nlink': 2,
|
|
'st_uid': 1000,
|
|
'st_gid': 1000,
|
|
'st_size': 4096,
|
|
'st_atime': 0,
|
|
'st_mtime': 0,
|
|
'st_ctime': 0
|
|
}
|
|
elif self.is_property(path):
|
|
value = {
|
|
'st_mode': 0o644 | stat.S_IFREG,
|
|
'st_ino': self.get_ino(path),
|
|
'st_dev': 0,
|
|
'st_nlink': 1,
|
|
'st_uid': 1000,
|
|
'st_gid': 1000,
|
|
'st_size': 4096,
|
|
'st_atime': 0,
|
|
'st_mtime': 0,
|
|
'st_ctime': 0
|
|
}
|
|
else:
|
|
raise FuseOSError(ENOENT)
|
|
return value
|
|
|
|
def readdir(self, path: str, fh: IO[bytes]) -> Iterator[str]:
|
|
yield '.'
|
|
yield '..'
|
|
for item in self.qom_list(path):
|
|
yield item.name
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(QOMFuse.entry_point())
|