diff --git a/core-modules/000QubesVm.py b/core-modules/000QubesVm.py index f8fdd76c..de4f8851 100644 --- a/core-modules/000QubesVm.py +++ b/core-modules/000QubesVm.py @@ -40,12 +40,12 @@ import signal from qubes import qmemman from qubes import qmemman_algo import libvirt -import warnings from qubes.qubes import dry_run,vmm from qubes.qubes import register_qubes_vm_class from qubes.qubes import QubesVmCollection,QubesException,QubesHost,QubesVmLabels from qubes.qubes import defaults,system_path,vm_files,qubes_max_qid +from qubes.storage import get_pool qmemman_present = False try: @@ -109,6 +109,7 @@ class QubesVm(object): "name": { "order": 1 }, "uuid": { "order": 0, "eval": 'uuid.UUID(value) if value else None' }, "dir_path": { "default": None, "order": 2 }, + "pool_name": { "default":"default" }, "conf_file": { "func": lambda value: self.absolute_path(value, self.name + ".conf"), @@ -198,7 +199,7 @@ class QubesVm(object): 'kernelopts', 'services', 'installed_by_rpm',\ 'uses_default_netvm', 'include_in_backups', 'debug',\ 'qrexec_timeout', 'autostart', 'uses_default_dispvm_netvm', - 'backup_content', 'backup_size', 'backup_path' ]: + 'backup_content', 'backup_size', 'backup_path', 'pool_name' ]: attrs[prop]['save'] = lambda prop=prop: str(getattr(self, prop)) # Simple paths for prop in ['conf_file', 'firewall_conf']: @@ -345,7 +346,11 @@ class QubesVm(object): self.services['qubes-update-check'] = False # Initialize VM image storage class - self.storage = defaults["storage_class"](self) + self.storage = get_pool(self.pool_name, self).getStorage() + self.dir_path = self.storage.vmdir + self.icon_path = os.path.join(self.storage.vmdir, 'icon.png') + self.conf_file = os.path.join(self.storage.vmdir, self.name + '.conf') + if hasattr(self, 'kernels_dir'): modules_path = os.path.join(self.kernels_dir, "modules.img") diff --git a/core/settings-wni-Windows_NT.py b/core/settings-wni-Windows_NT.py deleted file mode 100644 index 6e646c9d..00000000 --- a/core/settings-wni-Windows_NT.py +++ /dev/null @@ -1,45 +0,0 @@ -#!/usr/bin/python2 - -from __future__ import absolute_import -import _winreg -import os -import sys - -from qubes.storage.wni import QubesWniVmStorage - -DEFAULT_INSTALLDIR = 'c:\\program files\\Invisible Things Lab\\Qubes WNI' -DEFAULT_STOREDIR = 'c:\\qubes' - -def apply(system_path, vm_files, defaults): - system_path['qubes_base_dir'] = DEFAULT_STOREDIR - installdir = DEFAULT_INSTALLDIR - try: - reg_key = _winreg.OpenKey(_winreg.HKEY_LOCAL_MACHINE, - "Software\\Invisible Things Lab\\Qubes WNI") - installdir = _winreg.QueryValueEx(reg_key, "InstallDir")[0] - system_path['qubes_base_dir'] = \ - _winreg.QueryValueEx(reg_key, "StoreDir")[0] - except WindowsError as e: - print >>sys.stderr, \ - "WARNING: invalid installation: missing registry entries (%s)" \ - % str(e) - - system_path['config_template_pv'] = \ - os.path.join(installdir, 'vm-template.xml') - system_path['config_template_hvm'] = \ - os.path.join(installdir, 'vm-template-hvm.xml') - system_path['qubes_icon_dir'] = os.path.join(installdir, 'icons') - system_path['qubesdb_daemon_path'] = \ - os.path.join(installdir, 'bin\\qubesdb-daemon.exe') - system_path['qrexec_daemon_path'] = \ - os.path.join(installdir, 'bin\\qrexec-daemon.exe') - system_path['qrexec_client_path'] = \ - os.path.join(installdir, 'bin\\qrexec-client.exe') - system_path['qrexec_policy_dir'] = \ - os.path.join(installdir, 'qubes-rpc\\policy') - # Specific to WNI - normally VM have this file - system_path['qrexec_agent_path'] = \ - os.path.join(installdir, 'bin\\qrexec-agent.exe') - - defaults['libvirt_uri'] = 'wni:///' - defaults['storage_class'] = QubesWniVmStorage diff --git a/core/settings-xen-Linux.py b/core/settings-xen-Linux.py index de5084f5..c413e8ae 100644 --- a/core/settings-xen-Linux.py +++ b/core/settings-xen-Linux.py @@ -2,7 +2,10 @@ from __future__ import absolute_import -from qubes.storage.xen import QubesXenVmStorage +from qubes.storage.xen import XenStorage, XenPool + def apply(system_path, vm_files, defaults): - defaults['storage_class'] = QubesXenVmStorage + defaults['storage_class'] = XenStorage + defaults['pool_drivers'] = {'xen': XenPool} + defaults['pool_config'] = {'dir_path': '/var/lib/qubes/'} diff --git a/core/storage/Makefile b/core/storage/Makefile index ec59cc64..7c7af60e 100644 --- a/core/storage/Makefile +++ b/core/storage/Makefile @@ -1,5 +1,6 @@ OS ?= Linux +SYSCONFDIR ?= /etc PYTHON_QUBESPATH = $(PYTHON_SITEPATH)/qubes all: @@ -13,6 +14,8 @@ endif mkdir -p $(DESTDIR)$(PYTHON_QUBESPATH)/storage cp __init__.py $(DESTDIR)$(PYTHON_QUBESPATH)/storage cp __init__.py[co] $(DESTDIR)$(PYTHON_QUBESPATH)/storage + mkdir -p $(DESTDIR)$(SYSCONFDIR)/qubes + cp storage.conf $(DESTDIR)$(SYSCONFDIR)/qubes/ ifneq ($(BACKEND_VMM),) if [ -r $(BACKEND_VMM).py ]; then \ cp $(BACKEND_VMM).py $(DESTDIR)$(PYTHON_QUBESPATH)/storage && \ diff --git a/core/storage/README.md b/core/storage/README.md new file mode 100644 index 00000000..7258512f --- /dev/null +++ b/core/storage/README.md @@ -0,0 +1,3 @@ +# WNI File storage +Before v3.1 there existed a draft wni storage. You can find it in the git +history diff --git a/core/storage/__init__.py b/core/storage/__init__.py index 29cf8654..8e2ac720 100644 --- a/core/storage/__init__.py +++ b/core/storage/__init__.py @@ -16,22 +16,24 @@ # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software -# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. -# +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # from __future__ import absolute_import +import ConfigParser import os import os.path -import re import shutil import subprocess import sys -from qubes.qubes import vm_files,system_path,defaults -from qubes.qubes import QubesException import qubes.qubesutils +from qubes.qubes import QubesException, defaults, system_path, vm_files + +CONFIG_FILE = '/etc/qubes/storage.conf' + class QubesVmStorage(object): """ @@ -55,12 +57,6 @@ class QubesVmStorage(object): else: self.root_img_size = defaults['root_img_size'] - self.private_img = vm.absolute_path(vm_files["private_img"], None) - if self.vm.template: - self.root_img = self.vm.template.root_img - else: - self.root_img = vm.absolute_path(vm_files["root_img"], None) - self.volatile_img = vm.absolute_path(vm_files["volatile_img"], None) # For now compute this path still in QubesVm self.modules_img = modules_img @@ -199,3 +195,130 @@ class QubesVmStorage(object): print >>sys.stderr, "WARNING: Creating empty VM private image file: {0}".\ format(self.private_img) self.create_on_disk_private_img(verbose=False) + + +def dump(o): + """ Returns a string represention of the given object + + Args: + o (object): anything that response to `__module__` and `__class__` + + Given the class :class:`qubes.storage.QubesVmStorage` it returns + 'qubes.storage.QubesVmStorage' as string + """ + return o.__module__ + '.' + o.__class__.__name__ + + +def load(string): + """ Given a dotted full module string representation of a class it loads it + + Args: + string (str) i.e. 'qubes.storage.xen.QubesXenVmStorage' + + Returns: + type + + See also: + :func:`qubes.storage.dump` + """ + if not type(string) is str: + # This is a hack which allows giving a real class to a vm instead of a + # string as string_class parameter. + return string + + components = string.split(".") + module_path = ".".join(components[:-1]) + klass = components[-1:][0] + module = __import__(module_path, fromlist=[klass]) + return getattr(module, klass) + + +def get_pool(name, vm): + """ Instantiates the storage for the specified vm """ + config = _get_storage_config_parser() + + klass = _get_pool_klass(name, config) + + keys = [k for k in config.options(name) if k != 'driver' and k != 'class'] + values = [config.get(name, o) for o in keys] + config_kwargs = dict(zip(keys, values)) + + if name == 'default': + kwargs = defaults['pool_config'].copy() + kwargs.update(keys) + else: + kwargs = config_kwargs + + return klass(vm, **kwargs) + + +def pool_exists(name): + """ Check if the specified pool exists """ + try: + _get_pool_klass(name) + return True + except StoragePoolException: + return False + +def add_pool(name, **kwargs): + """ Add a storage pool to config.""" + config = _get_storage_config_parser() + config.add_section(name) + for key, value in kwargs.iteritems(): + config.set(name, key, value) + _write_config(config) + +def remove_pool(name): + """ Remove a storage pool from config file. """ + config = _get_storage_config_parser() + config.remove_section(name) + _write_config(config) + +def _write_config(config): + with open(CONFIG_FILE, 'w') as configfile: + config.write(configfile) + +def _get_storage_config_parser(): + """ Instantiates a `ConfigParaser` for specified storage config file. + + Returns: + RawConfigParser + """ + config = ConfigParser.RawConfigParser() + config.read(CONFIG_FILE) + return config + + +def _get_pool_klass(name, config=None): + """ Returns the storage klass for the specified pool. + + Args: + name: The pool name. + config: If ``config`` is not specified + `_get_storage_config_parser()` is called. + + Returns: + type: A class inheriting from `QubesVmStorage` + """ + if config is None: + config = _get_storage_config_parser() + + if not config.has_section(name): + raise StoragePoolException('Uknown storage pool ' + name) + + if config.has_option(name, 'class'): + klass = load(config.get(name, 'class')) + elif config.has_option(name, 'driver'): + pool_driver = config.get(name, 'driver') + klass = defaults['pool_drivers'][pool_driver] + else: + raise StoragePoolException('Uknown storage pool driver ' + name) + return klass + + +class StoragePoolException(QubesException): + pass + + +class Pool(object): + pass diff --git a/core/storage/storage.conf b/core/storage/storage.conf new file mode 100644 index 00000000..e9d067e5 --- /dev/null +++ b/core/storage/storage.conf @@ -0,0 +1,12 @@ +[default] ; poolname +driver=xen ; the default xen storage +; class = qubes.storage.xen.XenStorage ; class always overwrites the driver +; +; To use our own storage adapter, you need just to specify the module path and +; class name +; [pool-b] +; class = foo.bar.MyStorage +; +; [test-dummy] +; driver=dummy + diff --git a/core/storage/wni.py b/core/storage/wni.py deleted file mode 100644 index a3571765..00000000 --- a/core/storage/wni.py +++ /dev/null @@ -1,138 +0,0 @@ -#!/usr/bin/python2 -# -# The Qubes OS Project, http://www.qubes-os.org -# -# Copyright (C) 2013 Marek Marczykowski -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License -# as published by the Free Software Foundation; either version 2 -# 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 -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software -# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. -# -# - -from __future__ import absolute_import - -import sys -import os -import os.path -import win32api -import win32net -import win32netcon -import win32security -import win32profile -import pywintypes -import random - -from qubes.storage import QubesVmStorage -from qubes.qubes import QubesException,system_path - -class QubesWniVmStorage(QubesVmStorage): - """ - Class for VM storage of WNI VMs. - """ - - def __init__(self, *args, **kwargs): - super(QubesWniVmStorage, self).__init__(*args, **kwargs) - # Use the user profile as "private.img" - self.home_root = win32profile.GetProfilesDirectory() - # FIXME: the assignment below may not always be correct, - # but GetUserProfileDirectory needs a user token... - self.private_img = os.path.join(self.home_root, self._get_username()) - - # Pass paths for WNI libvirt driver - os.putenv("WNI_DRIVER_QUBESDB_PATH", system_path['qubesdb_daemon_path']) - os.putenv("WNI_DRIVER_QREXEC_AGENT_PATH", system_path['qrexec_agent_path']) - - def _get_username(self, vmname = None): - if vmname is None: - vmname = self.vm.name - return "qubes-vm-%s" % vmname - - def _get_random_password(self, vmname = None): - if vmname is None: - vmname = self.vm.name - return '%x' % random.SystemRandom().getrandombits(256) - - def get_config_params(self): - return {} - - def create_on_disk_private_img(self, verbose, source_template = None): - # FIXME: this may not always be correct - home_dir = os.path.join(self.home_root, self._get_username()) - # Create user data in information level 1 (PyUSER_INFO_1) format. - user_data = {} - user_data['name'] = self._get_username() - user_data['full_name'] = self._get_username() - # libvirt driver doesn't need to know the password anymore - user_data['password'] = self._get_random_password() - user_data['flags'] = ( - win32netcon.UF_NORMAL_ACCOUNT | - win32netcon.UF_SCRIPT | - win32netcon.UF_DONT_EXPIRE_PASSWD | - win32netcon.UF_PASSWD_CANT_CHANGE - ) - user_data['priv'] = win32netcon.USER_PRIV_USER - user_data['home_dir'] = home_dir - user_data['max_storage'] = win32netcon.USER_MAXSTORAGE_UNLIMITED - # TODO: catch possible exception - win32net.NetUserAdd(None, 1, user_data) - - def create_on_disk_root_img(self, verbose, source_template = None): - pass - - def remove_from_disk(self): - try: - sid = win32security.LookupAccountName(None, self._get_username())[0] - string_sid = win32security.ConvertSidToStringSid(sid) - win32profile.DeleteProfile(string_sid) - win32net.NetUserDel(None, self._get_username()) - except pywintypes.error, details: - if details[0] == 2221: - # "The user name cannot be found." - raise IOError("User %s doesn't exist" % self._get_username()) - else: - raise - - super(QubesWniVmStorage, self).remove_from_disk() - - def rename(self, old_name, new_name): - super(QubesWniVmStorage, self).rename(old_name, new_name) - user_data = {} - user_data['name'] = self._get_username(new_name) - win32net.NetUserSetInfo(None, - self._get_username(old_name), 0, user_data) - #TODO: rename user profile - - def verify_files(self): - if not os.path.exists (self.vmdir): - raise QubesException ( - "VM directory doesn't exist: {0}".\ - format(self.vmdir)) - - try: - # TemplateVm in WNI is quite virtual, so do not require the user - if not self.vm.is_template(): - win32net.NetUserGetInfo(None, self._get_username(), 0) - except pywintypes.error, details: - if details[0] == 2221: - # "The user name cannot be found." - raise QubesException("User %s doesn't exist" % self._get_username()) - else: - raise - - def reset_volatile_storage(self, verbose = False, source_template = None): - pass - - def prepare_for_vm_startup(self, verbose = False): - if self.vm.is_template(): - raise QubesException("Starting TemplateVM is not supported") diff --git a/core/storage/xen.py b/core/storage/xen.py index 00c3b3da..1cd5dc97 100644 --- a/core/storage/xen.py +++ b/core/storage/xen.py @@ -16,40 +16,61 @@ # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software -# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. -# +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # from __future__ import absolute_import import os import os.path +import re import subprocess import sys -import re -from qubes.storage import QubesVmStorage -from qubes.qubes import QubesException, vm_files +from qubes.qubes import (QubesAppVm, QubesDisposableVm, QubesException, + QubesHVm, QubesNetVm, QubesTemplateHVm, + QubesTemplateVm, defaults, vm_files) +from qubes.storage import Pool, QubesVmStorage -class QubesXenVmStorage(QubesVmStorage): +class XenStorage(QubesVmStorage): """ Class for VM storage of Xen VMs. """ - def __init__(self, vm, **kwargs): - super(QubesXenVmStorage, self).__init__(vm, **kwargs) + def __init__(self, vm, vmdir, **kwargs): + """ Instantiate the storage. + + Args: + vm: a QubesVM + vmdir: the root directory of the pool + """ + assert vm is not None + assert vmdir is not None + + super(XenStorage, self).__init__(vm, **kwargs) self.root_dev = "xvda" self.private_dev = "xvdb" self.volatile_dev = "xvdc" self.modules_dev = "xvdd" + self.vmdir = vmdir + if self.vm.is_template(): - self.rootcow_img = os.path.join(self.vmdir, vm_files["rootcow_img"]) + self.rootcow_img = os.path.join(self.vmdir, + vm_files["rootcow_img"]) else: self.rootcow_img = None + self.private_img = os.path.join(vmdir, 'private.img') + if self.vm.template: + self.root_img = self.vm.template.root_img + else: + self.root_img = os.path.join(vmdir, 'root.img') + self.volatile_img = os.path.join(vmdir, 'volatile.img') + def _format_disk_dev(self, path, script, vdev, rw=True, type="disk", domain=None): if path is None: return '' @@ -158,7 +179,7 @@ class QubesXenVmStorage(QubesVmStorage): self.commit_template_changes() def rename(self, old_name, new_name): - super(QubesXenVmStorage, self).rename(old_name, new_name) + super(XenStorage, self).rename(old_name, new_name) old_dirpath = os.path.join(os.path.dirname(self.vmdir), old_name) if self.rootcow_img: @@ -226,11 +247,11 @@ class QubesXenVmStorage(QubesVmStorage): f_volatile.close() f_root.close() return - super(QubesXenVmStorage, self).reset_volatile_storage( + super(XenStorage, self).reset_volatile_storage( verbose=verbose, source_template=source_template) def prepare_for_vm_startup(self, verbose): - super(QubesXenVmStorage, self).prepare_for_vm_startup(verbose=verbose) + super(XenStorage, self).prepare_for_vm_startup(verbose=verbose) if self.drive is not None: (drive_type, drive_domain, drive_path) = self.drive.split(":") @@ -249,3 +270,69 @@ class QubesXenVmStorage(QubesVmStorage): raise QubesException( "VM '{}' holding '{}' does not exists".format( drive_domain, drive_path)) + + +class XenPool(Pool): + + def __init__(self, vm, dir_path): + assert vm is not None + assert dir_path is not None + + appvms_path = os.path.join(dir_path, 'appvms') + servicevms_path = os.path.join(dir_path, 'servicevms') + vm_templates_path = os.path.join(dir_path, 'vm-templates') + + self._create_dir_if_not_exists(dir_path) + self._create_dir_if_not_exists(appvms_path) + self._create_dir_if_not_exists(servicevms_path) + self._create_dir_if_not_exists(vm_templates_path) + + self.vmdir = self._vmdir_path(vm, dir_path) + self.vm = vm + self.dir_path = dir_path + + def getStorage(self): + """ Returns an instantiated ``XenStorage``. """ + return defaults['storage_class'](self.vm, vmdir=self.vmdir) + + def _vmdir_path(self, vm, pool_dir): + """ Get the vm dir depending on the type of the VM. + + The default QubesOS file storage saves the vm images in three + different directories depending on the ``QubesVM`` type: + + * ``appvms`` for ``QubesAppVm`` or ``QubesHvm`` + * ``vm-templates`` for ``QubesTemplateVm`` or ``QubesTemplateHvm`` + * ``servicevms`` for any subclass of ``QubesNetVm`` + + Args: + vm: a QubesVM + pool_dir: the root directory of the pool + + Returns: + string (str) absolute path to the directory where the vm files + are stored + """ + vm_type = type(vm) + + if vm_type in [QubesAppVm, QubesHVm]: + subdir = 'appvms' + elif vm_type in [QubesTemplateVm, QubesTemplateHVm]: + subdir = 'vm-templates' + elif issubclass(vm_type, QubesNetVm): + subdir = 'servicevms' + elif vm_type is QubesDisposableVm: + subdir = 'appvms' + return os.path.join(pool_dir, subdir, vm.template.name + '-dvm') + else: + raise QubesException(str(vm_type) + ' unknown vm type') + + return os.path.join(pool_dir, subdir, vm.name) + + def _create_dir_if_not_exists(self, path): + """ Check if a directory exists in if not it is created. + + This method does not create any parent directories. + """ + if not os.path.exists(path): + os.mkdir(path) diff --git a/qvm-tools/qvm-create b/qvm-tools/qvm-create index a014febf..6ae002e5 100755 --- a/qvm-tools/qvm-create +++ b/qvm-tools/qvm-create @@ -41,6 +41,8 @@ def main(): help="Specify the label to use for the new VM (e.g. red, yellow, green, ...)") parser.add_option ("-p", "--proxy", action="store_true", dest="proxyvm", default=False, help="Create ProxyVM") + parser.add_option ("-P", "--pool", dest="pool_name", + help="Specify which storage pool to use") parser.add_option ("-H", "--hvm", action="store_true", dest="hvm", default=False, help="Create HVM (standalone unless --template option used)") parser.add_option ("--hvm-template", action="store_true", dest="hvm_template", default=False, @@ -71,6 +73,11 @@ def main(): parser.error ("You must specify VM name!") vmname = args[0] + if options.pool_name is None: + pool_name = "default" + else : + pool_name = options.pool_name + if (options.netvm + options.proxyvm + options.hvm + options.hvm_template) > 1: parser.error ("You must specify at most one VM type switch") @@ -170,7 +177,9 @@ def main(): vmtype = "QubesAppVm" try: - vm = qvm_collection.add_new_vm(vmtype, name=vmname, template=new_vm_template, label = label) + vm=qvm_collection.add_new_vm(vmtype, name=vmname, + template=new_vm_template, label=label, + pool_name=pool_name) except QubesException as err: print >> sys.stderr, "ERROR: {0}".format(err) exit (1) diff --git a/rpm_spec/core-dom0.spec b/rpm_spec/core-dom0.spec index 86612c48..84e106ba 100644 --- a/rpm_spec/core-dom0.spec +++ b/rpm_spec/core-dom0.spec @@ -176,6 +176,7 @@ fi %files %defattr(-,root,root,-) %config(noreplace) %attr(0664,root,qubes) %{_sysconfdir}/qubes/qmemman.conf +%config(noreplace) %attr(0664,root,qubes) %{_sysconfdir}/qubes/storage.conf /usr/bin/qvm-* /usr/bin/qubes-* %dir %{python_sitearch}/qubes diff --git a/tests/Makefile b/tests/Makefile index 8523adde..f2dff357 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -27,3 +27,5 @@ endif cp regressions.py[co] $(DESTDIR)$(PYTHON_TESTSPATH) cp run.py $(DESTDIR)$(PYTHON_TESTSPATH) cp run.py[co] $(DESTDIR)$(PYTHON_TESTSPATH) + cp storage.py $(DESTDIR)$(PYTHON_TESTSPATH) + cp storage.py[co] $(DESTDIR)$(PYTHON_TESTSPATH) diff --git a/tests/__init__.py b/tests/__init__.py index 8abea4b8..8af1afbf 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -547,6 +547,7 @@ def load_tests(loader, tests, pattern): 'qubes.tests.backup', 'qubes.tests.backupcompatibility', 'qubes.tests.regressions', + 'qubes.tests.storage', ): tests.addTests(loader.loadTestsFromName(modname)) diff --git a/tests/storage.py b/tests/storage.py new file mode 100644 index 00000000..5513d383 --- /dev/null +++ b/tests/storage.py @@ -0,0 +1,285 @@ +# The Qubes OS Project, https://www.qubes-os.org/ +# +# Copyright (C) 2015 Bahtiar `kalkin-` Gadimov +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 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 +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + + +import os +import shutil + +import qubes.storage +from qubes.qubes import defaults +from qubes.storage.xen import XenPool, XenStorage +from qubes.tests import QubesTestCase, SystemTestsMixin + + +class TC_00_Storage(SystemTestsMixin, QubesTestCase): + + """ This class tests the utility methods from :mod:``qubes.storage`` """ + + def test_000_dump(self): + """ Dumps storage instance to a storage string """ + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + vm = self.qc.add_new_vm('QubesAppVm', name=vmname, + pool_name='default', template=template) + storage = vm.storage + result = qubes.storage.dump(storage) + expected = 'qubes.storage.xen.XenStorage' + self.assertEquals(result, expected) + + def test_001_load(self): + """ Loads storage driver from a storage string """ + result = qubes.storage.load('qubes.storage.xen.XenStorage') + self.assertTrue(result is XenStorage) + + def test_002_default_pool_drivers(self): + """ The only predifined pool driver is xen """ + result = defaults['pool_drivers'].keys() + expected = ["xen"] + self.assertEquals(result, expected) + + def test_003_get_pool_klass(self): + """ Expect the default pool to be `XenPool` """ + result = qubes.storage._get_pool_klass('default') + self.assertTrue(result is XenPool) + + def test_004_pool_exists_default(self): + """ Expect the default pool to exists """ + self.assertTrue(qubes.storage.pool_exists('default')) + + def test_005_pool_exists_random(self): + """ Expect this pool to not a exist """ + self.assertFalse( + qubes.storage.pool_exists('asdh312096r832598213iudhas')) + + def test_006_add_remove_pool(self): + """ Tries to adding and removing a pool. """ + pool_name = 'asdjhrp89132' + + # make sure it's really does not exist + qubes.storage.remove_pool(pool_name) + + qubes.storage.add_pool(pool_name, driver='xen') + self.assertTrue(qubes.storage.pool_exists(pool_name)) + + qubes.storage.remove_pool(pool_name) + self.assertFalse(qubes.storage.pool_exists(pool_name)) + + +class TC_00_Pool(SystemTestsMixin, QubesTestCase): + + """ This class tests some properties of the 'default' pool. """ + + def test000_default_pool_dir(self): + """ The predefined dir for the default pool should be ``/var/lib/qubes`` + + .. sealso:: + Data :data:``qubes.qubes.defaults['pool_config']``. + """ + vm = self._init_app_vm() + result = qubes.storage.get_pool("default", vm).dir_path + expected = '/var/lib/qubes/' + self.assertEquals(result, expected) + + def test001_default_storage_class(self): + """ Check when using default pool the Storage is ``XenStorage``. """ + result = self._init_app_vm().storage + self.assertIsInstance(result, XenStorage) + + def test_002_default_pool_name(self): + """ Default pool_name is 'default'. """ + vm = self._init_app_vm() + self.assertEquals(vm.pool_name, "default") + + def _init_app_vm(self): + """ Return initalised, but not created, AppVm. """ + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + return self.qc.add_new_vm('QubesAppVm', name=vmname, template=template, + pool_name='default') + + +class TC_01_Pool(SystemTestsMixin, QubesTestCase): + + """ Test the paths for the default Xen file based storage (``XenStorage``). + """ + + POOL_DIR = '/var/lib/qubes/test-pool' + APPVMS_DIR = '/var/lib/qubes/test-pool/appvms' + TEMPLATES_DIR = '/var/lib/qubes/test-pool/vm-templates' + SERVICE_DIR = '/var/lib/qubes/test-pool/servicevms' + + def setUp(self): + """ Add a test file based storage pool """ + super(TC_01_Pool, self).setUp() + qubes.storage.add_pool('test-pool', driver='xen', + dir_path=self.POOL_DIR) + + def tearDown(self): + """ Remove the file based storage pool after testing """ + super(TC_01_Pool, self).tearDown() + qubes.storage.remove_pool("test-pool") + shutil.rmtree(self.POOL_DIR, ignore_errors=True) + + def test_001_pool_exists(self): + """ Check if the storage pool was added to the storage pool config """ + self.assertTrue(qubes.storage.pool_exists('test-pool')) + + def test_002_pool_dir_create(self): + """ Check if the storage pool dir and subdirs were created """ + + # The dir should not exists before + self.assertFalse(os.path.exists(self.POOL_DIR)) + + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + self.qc.add_new_vm('QubesAppVm', name=vmname, template=template, + pool_name='test-pool') + + self.assertTrue(os.path.exists(self.POOL_DIR)) + self.assertTrue(os.path.exists(self.APPVMS_DIR)) + self.assertTrue(os.path.exists(self.SERVICE_DIR)) + self.assertTrue(os.path.exists(self.TEMPLATES_DIR)) + + def test_003_pool_dir(self): + """ Check if the vm storage pool_dir is the same as specified """ + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + vm = self.qc.add_new_vm('QubesAppVm', name=vmname, template=template, + pool_name='test-pool') + result = qubes.storage.get_pool('test-pool', vm).dir_path + self.assertEquals(self.POOL_DIR, result) + + def test_004_app_vmdir(self): + """ Check the vm storage dir for an AppVm""" + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + vm = self.qc.add_new_vm('QubesAppVm', name=vmname, template=template, + pool_name='test-pool') + + expected = os.path.join(self.APPVMS_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_005_hvm_vmdir(self): + """ Check the vm storage dir for a HVM""" + vmname = self.make_vm_name('hvm') + vm = self.qc.add_new_vm('QubesHVm', name=vmname, + pool_name='test-pool') + + expected = os.path.join(self.APPVMS_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_006_net_vmdir(self): + """ Check the vm storage dir for a Netvm""" + vmname = self.make_vm_name('hvm') + vm = self.qc.add_new_vm('QubesNetVm', name=vmname, + pool_name='test-pool') + + expected = os.path.join(self.SERVICE_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_007_proxy_vmdir(self): + """ Check the vm storage dir for a ProxyVm""" + vmname = self.make_vm_name('proxyvm') + vm = self.qc.add_new_vm('QubesProxyVm', name=vmname, + pool_name='test-pool') + + expected = os.path.join(self.SERVICE_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_008_admin_vmdir(self): + """ Check the vm storage dir for a AdminVm""" + # TODO How to test AdminVm? + pass + + def test_009_template_vmdir(self): + """ Check the vm storage dir for a TemplateVm""" + vmname = self.make_vm_name('templatevm') + vm = self.qc.add_new_vm('QubesTemplateVm', name=vmname, + pool_name='test-pool') + + expected = os.path.join(self.TEMPLATES_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_010_template_hvm_vmdir(self): + """ Check the vm storage dir for a TemplateHVm""" + vmname = self.make_vm_name('templatehvm') + vm = self.qc.add_new_vm('QubesTemplateHVm', name=vmname, + pool_name='test-pool') + + expected = os.path.join(self.TEMPLATES_DIR, vm.name) + result = vm.storage.vmdir + self.assertEquals(expected, result) + + def test_011_appvm_file_images(self): + """ Check if all the needed image files are created for an AppVm""" + + vmname = self.make_vm_name('appvm') + template = self.qc.get_default_template() + vm = self.qc.add_new_vm('QubesAppVm', name=vmname, template=template, + pool_name='test-pool') + vm.create_on_disk(verbose=False) + + expected_vmdir = os.path.join(self.APPVMS_DIR, vm.name) + self.assertEqualsAndExists(vm.storage.vmdir, expected_vmdir) + + expected_private_path = os.path.join(expected_vmdir, 'private.img') + self.assertEqualsAndExists(vm.storage.private_img, + expected_private_path) + + expected_volatile_path = os.path.join(expected_vmdir, 'volatile.img') + self.assertEqualsAndExists(vm.storage.volatile_img, + expected_volatile_path) + + def test_012_hvm_file_images(self): + """ Check if all the needed image files are created for a HVm""" + + vmname = self.make_vm_name('hvm') + vm = self.qc.add_new_vm('QubesHVm', name=vmname, + pool_name='test-pool') + vm.create_on_disk(verbose=False) + + expected_vmdir = os.path.join(self.APPVMS_DIR, vm.name) + self.assertEqualsAndExists(vm.storage.vmdir, expected_vmdir) + + expected_private_path = os.path.join(expected_vmdir, 'private.img') + self.assertEqualsAndExists(vm.storage.private_img, + expected_private_path) + + expected_root_path = os.path.join(expected_vmdir, 'root.img') + self.assertEqualsAndExists(vm.storage.root_img, expected_root_path) + + expected_volatile_path = os.path.join(expected_vmdir, 'volatile.img') + self.assertEqualsAndExists(vm.storage.volatile_img, + expected_volatile_path) + + def assertEqualsAndExists(self, result_path, expected_path): + """ Check if the ``result_path``, matches ``expected_path`` and exists. + + See also: :meth:``assertExist`` + """ + self.assertEquals(result_path, expected_path) + self.assertExist(result_path) + + def assertExist(self, path): + """ Assert that the given path exists. """ + self.assertTrue(os.path.exists(path))