core-admin/qubes/tests/storage_callback.py

#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2020 David Hobach <david@hobach.de>
#
# This library 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 library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, see <https://www.gnu.org/licenses/>.
#
''' Tests for the callback storage driver.

    They are mostly based upon the lvm storage driver tests.
'''
# pylint: disable=line-too-long

import os
import json
import subprocess
import qubes.tests
import qubes.tests.storage
import qubes.tests.storage_lvm
from qubes.tests.storage_lvm import skipUnlessLvmPoolExists
from qubes.storage.callback import CallbackPool, CallbackVolume

CB_CONF = '/etc/qubes_callback.json'
LOG_BIN = '/tmp/testCbLogArgs'

CB_DATA = {'utest-callback-01': {
                'bdriver': 'lvm_thin',
                'bdriver_args': {
                     'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0],
                     'thin_pool':    qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1]
                },
                'description': 'For unit testing of the callback pool driver.'
            },
            'utest-callback-02': {
                'bdriver': 'lvm_thin',
                'bdriver_args': {
                     'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0],
                     'thin_pool':    qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1]
                },
                'cmd': LOG_BIN,
                'description': 'For unit testing of the callback pool driver.'
            },
            'utest-callback-03': {
                'bdriver': 'lvm_thin',
                'bdriver_args': {
                     'volume_group': qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0],
                     'thin_pool':    qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[1]
                },
                'cmd': 'exit 1',
                'pre_sinit': LOG_BIN + ' pre_sinit',
                'pre_setup': LOG_BIN + ' pre_setup',
                'pre_volume_create': LOG_BIN + ' pre_volume_create',
                'post_volume_create': LOG_BIN + ' post_volume_create',
                'pre_volume_import_data': LOG_BIN + ' pre_volume_import_data',
                'post_volume_import_data_end': LOG_BIN + ' post_volume_import_data_end',
                'post_volume_remove': LOG_BIN + ' post_volume_remove',
                'post_destroy': '-',
                'description': 'For unit testing of the callback pool driver.'
            },
            'testing-fail-missing-all': {
            },
            'testing-fail-missing-bdriver-args': {
                'bdriver': 'file',
                'description': 'For unit testing of the callback pool driver.'
            },
            'testing-fail-incorrect-bdriver': {
                'bdriver': 'nonexisting-bdriver',
                'bdriver_args': {
                     'foo': 'bar',
                     'bla': 'blub'
                },
                'cmd': 'echo foo',
                'description': 'For unit testing of the callback pool driver.'
            },
          }

class CallbackBase:
    ''' Mixin base class for callback tests. Has no base class. '''
    conf_id = None
    pool_name = 'test-callback'

    @classmethod
    def setUpClass(cls, conf_id='utest-callback-01'):
        conf = {'name': cls.pool_name,
                'driver': 'callback',
                'conf_id': conf_id}
        cls.conf_id = conf_id

        assert not(os.path.exists(CB_CONF)), '%s must NOT exist. Please delete it, if you do not need it.' % CB_CONF

        sudo = [] if os.getuid() == 0 else ['sudo']
        subprocess.run(sudo + ['install', '-m', '666', '/dev/null', CB_CONF], check=True)

        with open(CB_CONF, 'w') as outfile:
            json.dump(CB_DATA, outfile)
        super().setUpClass(pool_class=CallbackPool, volume_class=CallbackVolume, pool_conf=conf)

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()

        sudo = [] if os.getuid() == 0 else ['sudo']
        subprocess.run(sudo + ['rm', '-f', CB_CONF], check=True)

    def setUp(self, init_pool=True):
        super().setUp(init_pool=init_pool)
        if init_pool:
            #tests from other pools will assume that they're fully initialized after calling __init__()
            self.loop.run_until_complete(self.pool._assert_initialized())

    def test_000_000_callback_test_init(self):
        ''' Check whether the test init did work. '''
        if hasattr(self, 'pool'):
            self.assertIsInstance(self.pool, CallbackPool)
            self.assertEqual(self.pool.backend_class, qubes.storage.lvm.ThinPool)
        self.assertTrue(os.path.isfile(CB_CONF))

@skipUnlessLvmPoolExists
class TC_00_CallbackPool(CallbackBase, qubes.tests.storage_lvm.TC_00_ThinPool):
    pass

@skipUnlessLvmPoolExists
class TC_01_CallbackPool(CallbackBase, qubes.tests.storage_lvm.TC_01_ThinPool):
    pass

@skipUnlessLvmPoolExists
class TC_02_cb_StorageHelpers(CallbackBase, qubes.tests.storage_lvm.TC_02_StorageHelpers):
    pass

class LoggingCallbackBase(CallbackBase):
    ''' Mixin base class that sets up LOG_BIN and removes `LoggingCallbackBase.test_log`, if needed. '''
    test_log = '/tmp/cb_tests.log'
    test_log_expected = None #dict: class + test name --> test index (int, 0..x) --> expected _additional_ log content
    volume_name = 'volume_name'
    xml_path = '/tmp/qubes-test-callback.xml'

    @classmethod
    def setUpClass(cls, conf_id=None, log_expected=None):
        script = """#!/bin/bash
i=1
for arg in "$@" ; do
    echo "$i: $arg" >> "LOG_OUT"
    (( i++))
done
exit 0
"""
        script = script.replace('LOG_OUT', cls.test_log)
        with open(LOG_BIN, 'w') as f:
            f.write(script)
        os.chmod(LOG_BIN, 0o775)

        cls.test_log_expected = log_expected
        super().setUpClass(conf_id=conf_id)

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        os.remove(LOG_BIN)

    def setUp(self, init_pool=False):
        assert not(os.path.exists(self.test_log)), '%s must NOT exist. Please delete it, if you do not need it.' % self.test_log
        self.maxDiff = None

        xml = """
        <qubes>
          <labels>
            <label color="0x000000" id="label-8">black</label>
          </labels>
          <pools>
            <pool dir_path="/var/lib/qubes" driver="file" name="varlibqubes" revisions_to_keep="1"/>
            <pool dir_path="/var/lib/qubes/vm-kernels" driver="linux-kernel" name="linux-kernel"/>
            <pool conf_id="CONF_ID" driver="callback" name="POOL_NAME"/>
          </pools>
          <properties>
            <property name="clockvm"></property>
            <property name="default_pool_kernel">linux-kernel</property>
            <property name="default_template"></property>
            <property name="updatevm"></property>
          </properties>
          <domains>
            <domain id="domain-0" class="AdminVM">
              <properties>
                <property name="label">black</property>
              </properties>
              <features/>
              <tags/>
            </domain>
          </domains>
        </qubes>
        """
        xml = xml.replace('CONF_ID', self.conf_id)
        xml = xml.replace('POOL_NAME', self.pool_name)
        with open(self.xml_path, 'w') as f:
            f.write(xml)
        self.app = qubes.Qubes(self.xml_path,
            clockvm=None,
            updatevm=None,
            offline_mode=True,
        )
        os.environ['QUBES_XML_PATH'] = self.xml_path
        super().setUp(init_pool=init_pool)

    def tearDown(self):
        super().tearDown()
        os.unlink(self.app.store)
        self.app.close()
        del self.app
        for attr in dir(self):
            if isinstance(getattr(self, attr), qubes.vm.BaseVM):
                delattr(self, attr)

        if os.path.exists(self.test_log):
            os.remove(self.test_log)

        if os.path.exists(self.xml_path):
            os.remove(self.xml_path)

    def assertLogContent(self, expected):
        ''' Assert that the log matches the given string.
        :param expected: Expected content of the log file (String).
        '''
        try:
            with open(self.test_log, 'r') as f:
                found = f.read()
        except FileNotFoundError:
            found = ''
        if expected != '':
            expected = expected + '\n'
        self.assertEqual(found, expected)

    def assertLog(self, test_name, ind=0):
        ''' Assert that the log matches the expected status.
        :param test_name: Name of the test.
        :param ind: Index inside `test_log_expected` to check against (Integer starting at 0).
        '''
        d = self.test_log_expected[str(self.__class__) + test_name]
        expected = []
        for i in range(ind+1):
            expected = expected + [d[i]]
        expected = filter(None, expected)
        self.assertLogContent('\n'.join(expected))

    def test_001_callbacks(self):
        ''' create a lvm pool with additional callbacks '''
        config = {
            'name': self.volume_name,
            'pool': self.pool_name,
            'save_on_stop': True,
            'rw': True,
            'revisions_to_keep': 2,
            'size': qubes.config.defaults['root_img_size'],
        }
        new_size = 2 * qubes.config.defaults['root_img_size']

        test_name = 'test_001_callbacks'
        self.assertLog(test_name, 0)
        self.init_pool()
        self.assertFalse(self.created_pool)
        self.assertIsInstance(self.pool, CallbackPool)
        self.assertLog(test_name, 1)
        vm = qubes.tests.storage.TestVM(self)
        volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
        self.assertLog(test_name, 2)
        self.loop.run_until_complete(volume.create())
        self.assertLog(test_name, 3)
        self.loop.run_until_complete(volume.import_data(new_size))
        self.assertLog(test_name, 4)
        self.loop.run_until_complete(volume.import_data_end(True))
        self.assertLog(test_name, 5)
        self.assertEqual(volume.size, new_size)
        self.loop.run_until_complete(volume.remove())
        self.assertLog(test_name, 6)

@skipUnlessLvmPoolExists
class TC_91_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBase):
    ''' Tests for the actual callback functionality.
        conf_id = utest-callback-02
    '''

    @classmethod
    def setUpClass(cls):
        conf_id = 'utest-callback-02'
        name = cls.pool_name
        bdriver = (CB_DATA[conf_id])['bdriver']
        ctor_params = json.dumps(CB_DATA[conf_id], sort_keys=True, indent=2)
        vname = cls.volume_name
        vid = '{0}/vm-test-inst-appvm-{1}'.format(qubes.tests.storage_lvm.DEFAULT_LVM_POOL.split('/')[0], vname)
        vsize = 2 * qubes.config.defaults['root_img_size']
        log_expected = \
            {str(cls) + 'test_001_callbacks':
                {0: '',
                 1: '',
                 2: '',
                 3: '1: {0}\n2: {1}\n3: pre_sinit\n4: {2}\n5: {3}\n1: {0}\n2: {1}\n3: pre_volume_create\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n1: {0}\n2: {1}\n3: post_volume_create\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None'.format(name, bdriver, conf_id, ctor_params, vname, vid),
                 4: '1: {0}\n2: {1}\n3: pre_volume_import_data\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n9: {6}'.format(name, bdriver, conf_id, ctor_params, vname, vid, vsize),
                 5: '1: {0}\n2: {1}\n3: post_volume_import_data_end\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None\n9: {6}'.format(name, bdriver, conf_id, ctor_params, vname, vid, True),
                 6: '1: {0}\n2: {1}\n3: post_volume_remove\n4: {2}\n5: {3}\n6: {4}\n7: {5}\n8: None'.format(name, bdriver, conf_id, ctor_params, vname, vid),
                 }
            }
        super().setUpClass(conf_id=conf_id, log_expected=log_expected)

@skipUnlessLvmPoolExists
class TC_92_CallbackPool(LoggingCallbackBase, qubes.tests.storage_lvm.ThinPoolBase):
    ''' Tests for the actual callback functionality.
        conf_id = utest-callback-03
    '''

    @classmethod
    def setUpClass(cls):
        log_expected = \
            {str(cls) + 'test_001_callbacks':
                {0: '',
                 1: '',
                 2: '',
                 3: '1: pre_sinit\n1: pre_volume_create\n1: post_volume_create',
                 4: '1: pre_volume_import_data',
                 5: '1: post_volume_import_data_end',
                 6: '1: post_volume_remove',
                 }
            }
        super().setUpClass(conf_id='utest-callback-03', log_expected=log_expected)

    def test_002_failing_callback(self):
        ''' Make sure that we check the exit code of executed callbacks. '''
        config = {
            'name': self.volume_name,
            'pool': self.pool_name,
            'save_on_stop': True,
            'rw': True,
            'revisions_to_keep': 2,
            'size': qubes.config.defaults['root_img_size'],
        }
        self.init_pool()
        vm = qubes.tests.storage.TestVM(self)
        volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
        with self.assertRaises(subprocess.CalledProcessError) as cm:
            #should trigger the `exit 1` of `cmd`
            self.loop.run_until_complete(volume.start())
        self.assertTrue('exit status 1' in str(cm.exception))

    def test_003_errors(self):
        ''' Make sure we error out on common user & dev mistakes. '''
        #missing conf_id
        with self.assertRaises(qubes.storage.StoragePoolException):
            cb = CallbackPool(name='some-name', conf_id='')

        #invalid conf_id
        with self.assertRaises(qubes.storage.StoragePoolException):
            cb = CallbackPool(name='some-name', conf_id='nonexisting-id')

        #incorrect backend driver
        with self.assertRaises(qubes.storage.StoragePoolException):
            cb = CallbackPool(name='some-name', conf_id='testing-fail-incorrect-bdriver')

        #missing config entries
        with self.assertRaises(qubes.storage.StoragePoolException):
            cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-all')

        #missing bdriver args
        with self.assertRaises(TypeError):
            cb = CallbackPool(name='some-name', conf_id='testing-fail-missing-bdriver-args')

class TC_93_CallbackPool(qubes.tests.QubesTestCase):
    def test_001_missing_conf(self):
        ''' A missing config file must cause errors. '''
        with self.assertRaises(FileNotFoundError):
            cb = CallbackPool(name='some-name', conf_id='nonexisting-id')