Source code for qubes.ext.services

# -*- encoding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
#                               <marmarek@invisiblethingslab.com>
#
# 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/>.

"""Extension responsible for qvm-service framework"""

import os
import qubes.ext
import qubes.config


[docs] class ServicesExtension(qubes.ext.Extension): """This extension export features with 'service.' prefix to QubesDB in /qubes-service/ tree. """ @staticmethod def add_dom0_service(vm, service): try: os.makedirs( qubes.config.system_path['dom0_services_dir'], exist_ok=True) service = '{}/{}'.format( qubes.config.system_path['dom0_services_dir'], service) if not os.path.exists(service): os.mknod(service) except PermissionError: vm.log.warning("Cannot write to {}".format( qubes.config.system_path['dom0_services_dir'])) @staticmethod def remove_dom0_service(vm, service): try: service = '{}/{}'.format( qubes.config.system_path['dom0_services_dir'], service) if os.path.exists(service): os.remove(service) except PermissionError: vm.log.warning("Cannot write to {}".format( qubes.config.system_path['dom0_services_dir'])) @qubes.ext.handler('domain-qdb-create') def on_domain_qdb_create(self, vm, event): """Actually export features""" # pylint: disable=unused-argument for feature, value in vm.features.items(): if not feature.startswith('service.'): continue service = feature[len('service.'):] # forcefully convert to '0' or '1' vm.untrusted_qdb.write('/qubes-service/{}'.format(service), str(int(bool(value)))) # always set meminfo-writer according to maxmem vm.untrusted_qdb.write('/qubes-service/meminfo-writer', '1' if vm.maxmem > 0 else '0') @qubes.ext.handler('domain-feature-set:*') def on_domain_feature_set(self, vm, event, feature, value, oldvalue=None): """Update /qubes-service/ QubesDB tree in runtime""" # pylint: disable=unused-argument # TODO: remove this compatibility hack in Qubes 4.1 if feature == 'service.meminfo-writer': # if someone try to enable meminfo-writer ... if value: # ... reset maxmem to default vm.maxmem = qubes.property.DEFAULT else: # otherwise, set to 0 vm.maxmem = 0 # in any case, remove the entry, as it does not indicate memory # balancing state anymore del vm.features['service.meminfo-writer'] if not feature.startswith('service.'): return service = feature[len('service.'):] # qubesdb keys are limited to 63 bytes, and "/qubes-service/" is # 15 bytes. That leaves 48 for the service name. if len(service) > 48: raise qubes.exc.QubesValueError( 'Service name must not exceed 48 bytes') # The empty string is not a valid file name. if not service: raise qubes.exc.QubesValueError('Empty service name not allowed') # Require service names to start with an ASCII letter. This implicitly # rejects names which start with '-' (which could be interpreted as an # option) or are '.' or '..'. if not (('a' <= service[0] <= 'z') or ('A' <= service[0] <= 'Z')): raise qubes.exc.QubesValueError( 'Service name must start with an ASCII letter') if not vm.is_running(): return # forcefully convert to '0' or '1' vm.untrusted_qdb.write('/qubes-service/' + service, str(int(bool(value)))) if vm.name == "dom0": if str(int(bool(value))) == "1": self.add_dom0_service(vm, service) else: self.remove_dom0_service(vm, service) @qubes.ext.handler('domain-feature-delete:*') def on_domain_feature_delete(self, vm, event, feature): """Update /qubes-service/ QubesDB tree in runtime""" # pylint: disable=unused-argument if not vm.is_running(): return if not feature.startswith('service.'): return service = feature[len('service.'):] # this one is excluded from user control if service == 'meminfo-writer': return vm.untrusted_qdb.rm('/qubes-service/{}'.format(service)) if vm.name == "dom0": self.remove_dom0_service(vm, service) @qubes.ext.handler('domain-load') def on_domain_load(self, vm, event): """Migrate meminfo-writer service into maxmem""" # pylint: disable=unused-argument if 'service.meminfo-writer' in vm.features: # if was set to false, force maxmem=0 # otherwise, simply ignore as the default is fine if not vm.features['service.meminfo-writer']: vm.maxmem = 0 del vm.features['service.meminfo-writer'] if vm.name == "dom0": for feature, value in vm.features.items(): if not feature.startswith('service.'): continue service = feature[len('service.'):] if str(int(bool(value))) == "1": self.add_dom0_service(vm, service) else: self.remove_dom0_service(vm, service) @qubes.ext.handler('features-request') def supported_services(self, vm, event, untrusted_features): """Handle advertisement of supported services""" # pylint: disable=unused-argument if getattr(vm, 'template', None): vm.log.warning( 'Ignoring qubes.FeaturesRequest from template-based VM') return new_supported_services = set() for requested_service in untrusted_features: if not requested_service.startswith('supported-service.'): continue if untrusted_features[requested_service] == '1': # only allow to advertise service as supported, lack of entry # means service is not supported new_supported_services.add(requested_service) del untrusted_features # if no service is supported, ignore the whole thing - do not clear # all services in case of empty request (manual or such) if not new_supported_services: return old_supported_services = set( feat for feat in vm.features if feat.startswith('supported-service.') and vm.features[feat]) for feature in new_supported_services.difference( old_supported_services): vm.features[feature] = True if feature == 'supported-service.apparmor' and \ not 'apparmor' in vm.features: vm.features['apparmor'] = True for feature in old_supported_services.difference( new_supported_services): del vm.features[feature] if feature == 'supported-service.apparmor' and \ 'apparmor' in vm.features and vm.features['apparmor'] == '1': del vm.features['apparmor']