Source code for

# encoding=utf-8
# The Qubes OS Project,
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <>
# Copyright (C) 2017 Marek Marczykowski-Górecki
#                               <>
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <>.

"""Qubes volume management"""

from __future__ import print_function

import argparse
import os
import sys

import collections

import qubesadmin
import qubesadmin.exc
import qubesadmin.utils

[docs] def prepare_table(vd_list, full=False): """ Converts a list of :py:class:`VolumeData` objects to a list of tupples for the :py:func:``. If :program:`qvm-volume` is running in a TTY, it will ommit duplicate data. :param list vd_list: List of :py:class:`VolumeData` objects. :param bool full: If set to true duplicate data is printed even when running from TTY. :returns: list of tupples """ output = [] output += [('POOL:VOLUME', 'VMNAME', 'VOLUME_NAME', 'REVERT_POSSIBLE')] for volume in sorted(vd_list): if vmname, volume_name = output += [(str(volume), vmname, volume_name, volume.revisions)] for tupple in vmname, volume_name = tupple if full or not sys.stdout.isatty(): output += [(str(volume), vmname, volume_name, volume.revisions)] else: output += [('', vmname, volume_name, volume.revisions)] else: output += [(str(volume), "")] return output
[docs] class VolumeData(object): """ Wrapper object around :py:class:``, mainly to track the domains a volume is attached to. """ # pylint: disable=too-few-public-methods def __init__(self, volume): self.pool = volume.pool self.vid = volume.vid if volume.revisions: self.revisions = 'Yes' else: self.revisions = 'No' = [] def __lt__(self, other): return (self.pool, self.vid) < (other.pool, other.vid) def __str__(self): return "{!s}:{!s}".format(self.pool, self.vid)
[docs] def info_volume(args): """ Show info about selected volume """ volume = args.volume info_items = ( 'pool', 'vid', 'rw', 'source', 'save_on_stop', 'snap_on_start', 'size', 'usage', 'revisions_to_keep', 'ephemeral') if if == 'revisions': for rev in volume.revisions: print(rev) elif == 'is_outdated': print(volume.is_outdated()) elif in info_items: value = getattr(volume, if value is None: value = '' print(value) else: raise qubesadmin.exc.StoragePoolException( 'No such property: {}'.format( else: info = collections.OrderedDict() for item in info_items: value = getattr(volume, item) if value is None: value = '' info[item] = str(value) info['is_outdated'] = str(volume.is_outdated()) revisions = volume.revisions if revisions: print('List of available revisions (for revert):') for rev in revisions: print(' ' + rev) else: print('List of available revisions (for revert): none')
[docs] def config_volume(args): """ Change property of selected volume """ volume = args.volume if not in ('rw', 'revisions_to_keep', 'ephemeral'): raise qubesadmin.exc.QubesNoSuchPropertyError( 'Invalid property: {}'.format( setattr(volume,, args.value)
[docs] def import_volume(args): """ Import a file into volume """ volume = args.volume input_path = args.input_path if input_path == '-': input_file = sys.stdin.buffer else: # pylint: disable=consider-using-with input_file = open(input_path, 'rb') try: if args.no_resize: volume.import_data(stream=input_file) else: if args.size: size = args.size else: try: size = os.stat(input_file.fileno()).st_size except OSError as e: raise qubesadmin.exc.QubesException( 'Failed to get %s file size, ' 'specify it explicitly with --size, ' 'or use --no-resize option', str(e)) volume.import_data_with_size(stream=input_file, size=size) finally: if input_path != '-': input_file.close()
[docs] def list_volumes(args): """ Called by the parser to execute the qvm-volume list subcommand. """ app = if hasattr(args, 'domains') and domains = else: domains = volumes = [v for vm in domains for v in vm.volumes.values()] if getattr(args, 'pools', None): # only specified pools volumes = [v for v in volumes if v.pool in args.pools] vd_dict = {} for volume in volumes: volume_data = VolumeData(volume) try: vd_dict[volume.pool][volume.vid] = volume_data except KeyError: vd_dict[volume.pool] = {volume.vid: volume_data} for domain in domains: # gather the domain names try: for name, volume in domain.volumes.items(): try: volume_data = vd_dict[volume.pool][volume.vid] += [(, name)] except KeyError: # Skipping volume continue except AttributeError: # Skipping domain without volumes continue if hasattr(args, 'domains') and result = [x # reduce to only VolumeData with assigned domains for p in vd_dict.values() for x in p.values() if] else: result = [x for p in vd_dict.values() for x in p.values()] prepare_table(result, full=getattr(args, 'full', False)))
[docs] def revert_volume(args): """ Revert volume to previous state """ volume = args.volume if args.revision: revision = args.revision else: revisions = volume.revisions if not revisions: raise qubesadmin.exc.StoragePoolException( 'No snapshots available') revision = volume.revisions[-1] volume.revert(revision)
[docs] def extend_volumes(args): """ Called by the parser to execute the :program:`qvm-volume extend` subcommand """ volume = args.volume size = qubesadmin.utils.parse_size(args.size) if not args.force and size < volume.size: raise qubesadmin.exc.StoragePoolException( 'For your own safety, shrinking of %s is' ' disabled (%d < %d). If you really know what you' ' are doing, resize filesystem manually first, then use `-f` ' 'option.' % (, size, volume.size)) volume.resize(size)
[docs] def init_list_parser(sub_parsers): """ Configures the parser for the :program:`qvm-volume list` subcommand """ # pylint: disable=protected-access list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'), help='list storage volumes') list_parser.add_argument('-p', '--pool', metavar="POOL_NAME", dest='pools', list_parser.add_argument( '--full', action='store_true', help='print full line for each POOL_NAME:VOLUME_ID & vm combination') vm_name_group = list_parser, required=False,, help='list volumes from specified domain(s)') list_parser._mutually_exclusive_groups.append(vm_name_group) list_parser.set_defaults(func=list_volumes)
[docs] def init_revert_parser(sub_parsers): """ Add 'revert' action related options """ revert_parser = sub_parsers.add_parser( 'revert', aliases=('rv', 'r'), help='revert volume to previous revision') revert_parser.add_argument(metavar='VM:VOLUME', dest='volume', revert_parser.add_argument( metavar='REVISION', dest='revision', help='Optional revision to revert to; ' 'if not specified, latest one is assumed', action='store', nargs='?') revert_parser.set_defaults(func=revert_volume)
[docs] def init_extend_parser(sub_parsers): """ Add 'extend' action related options """ extend_parser = sub_parsers.add_parser( "resize", aliases=('extend', ), help="resize volume for domain") extend_parser.add_argument(metavar='VM:VOLUME', dest='volume', extend_parser.add_argument('size', help='New size in bytes') extend_parser.add_argument( '--force', '-f', action='store_true', help='Force operation, even if new size is smaller than the current ' 'one') extend_parser.set_defaults(func=extend_volumes)
[docs] def init_info_parser(sub_parsers): """ Add 'info' action related options """ info_parser = sub_parsers.add_parser( 'info', aliases=('i',), help='info about volume') info_parser.add_argument(metavar='VM:VOLUME', dest='volume', info_parser.add_argument( dest='property', action='store', nargs=argparse.OPTIONAL, help='Show only this property instead of all of them; use ' '\'revisions\' to list available revisions') info_parser.set_defaults(func=info_volume)
[docs] def init_config_parser(sub_parsers): """ Add 'info' action related options """ info_parser = sub_parsers.add_parser( 'config', aliases=('c', 'set', 's'), help='set config option for a volume') info_parser.add_argument(metavar='VM:VOLUME', dest='volume', info_parser.add_argument(dest='property', action='store') info_parser.add_argument(dest='value', action='store') info_parser.set_defaults(func=config_volume)
[docs] def init_import_parser(sub_parsers): """ Add 'import' action related options """ import_parser = sub_parsers.add_parser( 'import', help='import volume data') import_parser.add_argument(metavar='VM:VOLUME', dest='volume', import_parser.add_argument('input_path', metavar='PATH', help='File path to import, use \'-\' for standard input') import_parser.add_argument('--size', action='store', type=int, help='Set volume size to this value in bytes') import_parser.add_argument('--no-resize', action='store_true', help='Do not resize volume before importing data') import_parser.set_defaults(func=import_volume)
[docs] def get_parser(): """Create :py:class:`argparse.ArgumentParser` suitable for :program:`qvm-volume`. """ parser = parser.register( 'action', 'parsers', sub_parsers = parser.add_subparsers( title='commands', description="For more information see qvm-volume command -h", dest='command') init_info_parser(sub_parsers) init_config_parser(sub_parsers) init_extend_parser(sub_parsers) init_list_parser(sub_parsers) init_revert_parser(sub_parsers) init_import_parser(sub_parsers) # default action parser.set_defaults(func=list_volumes) return parser
[docs] def main(args=None, app=None): """Main routine of :program:`qvm-volume`.""" parser = get_parser() try: args = parser.parse_args(args, app=app) args.func(args) except qubesadmin.exc.QubesException as e: parser.print_error(str(e)) return 1 return 0
if __name__ == '__main__': sys.exit(main())