sdkv.py

"""Extension responsible for VM readable hierarchical key value stores"""

import qubes.ext


def _sdkv_get_key(feature):
    branch = "sdkv."
    return feature[len(branch):] if feature.startswith(branch) else None


class SdkvExtension(qubes.ext.Extension):
    """This extension exports features prefixed with 'sdkv.' to QubesDB in
    its /sdkv/ tree.
    """

    def _set_key_value(self, vm, key, value):
        # TODO
        # Convert value to base64, chunked?
        vm.untrusted_qdb.write('/sdkv/{}'.format(key.replace('.', '/')), str(value))

    @qubes.ext.handler('domain-qdb-create')
    def on_domain_qdb_create(self, vm, _event):
        """Actually export features"""

        for feature, value in vm.features.items():
            key = _sdkv_get_key(feature)
            if key:
                self._set_key_value(vm, key, value)

    @qubes.ext.handler('domain-feature-set:*')
    def on_domain_feature_set(self, vm, _event, feature, value, oldvalue=None):
        """Update /sdkv/ QubesDB tree in runtime"""

        if not vm.is_running():
            return

        key = _sdkv_get_key(feature)
        if key:
            self._set_key_value(vm, key, value)

    @qubes.ext.handler('domain-feature-delete:*')
    def on_domain_feature_delete(self, vm, event, feature):
        """Update /sdkv/ QubesDB tree in runtime"""

        if not vm.is_running():
            return

        key = _sdkv_get_key(feature)
        if key:
            vm.untrusted_qdb.rm('/sdkv/{}'.format(key))