core-admin/qubes/vm/__init__.py

#!/usr/bin/python2 -O

'''Qubes Virtual Machines

Main public classes
-------------------

.. autoclass:: BaseVM
   :members:
   :show-inheritance:
.. autoclass:: property
   :members:
   :show-inheritance:

Helper classes and functions
----------------------------

.. autoclass:: VMPlugin
   :members:
   :show-inheritance:

Particular VM classes
---------------------

Main types:

.. toctree::
   :maxdepth: 1

   qubesvm
   appvm
   templatevm

Special VM types:

.. toctree::
   :maxdepth: 1

   netvm
   proxyvm
   dispvm
   adminvm

HVMs:

.. toctree::
   :maxdepth: 1

   hvm
   templatehvm

'''

import ast
import collections
import functools
import sys

import dateutil.parser

import qubes.plugins

class property(object):
    '''Qubes VM property.

    This class holds one property that can be saved and loaded from qubes.xml

    :param str name: name of the property
    :param object default: default value
    :param type type: if not :py:obj:`None`, this is used to initialise value
    :param int order: order of evaluation (bigger order values are later)
    :param str doc: docstring

    '''

    def __init__(self, name, default=None, type=None, order=0, doc=None):
        self.__name__ = name
        self._default = default
        self._type = type
        self.order = order
        self.__doc__ = doc

        self._attr_name = '_qubesprop_' + self.__name__

    def __get__(self, instance, owner):
        if instance is None:
            return self

        try:
            return getattr(instance, self._attr_name)

        except AttributeError:
            if self._default is None:
                raise AttributeError('property not set')
            else:
                return self._default

    def __set__(self, instance, value):
        setattr(instance, self._attr_name,
            (self._type(value) if self._type is not None else value))

    def __repr__(self):
        return '<{} object at {:#x} name={!r} default={!r}>'.format(
            self.__class__.__name__, id(self), self.__name__, self._default)

    def __hash__(self):
        return hash(self.__name__)

    def __eq__(self, other):
        return self.__name__ == other.__name__

class VMPlugin(qubes.plugins.Plugin):
    '''Metaclass for :py:class:`.BaseVM`'''
    def __init__(cls, name, bases, dict_):
        super(VMPlugin, cls).__init__(name, bases, dict_)
        cls.__hooks__ = collections.defaultdict(list)

class BaseVM(object):
    '''Base class for all VMs

    :param xml: xml node from which to deserialise
    :type xml: :py:class:`lxml.etree._Element` or :py:obj:`None`

    This class is responsible for serialising and deserialising machines and
    provides basic framework. It contains no management logic. For that, see
    :py:class:`qubes.vm.qubesvm.QubesVM`.
    '''

    __metaclass__ = VMPlugin

    def get_props_list(self):
        '''List all properties attached to this VM'''
        props = set()
        for class_ in self.__class__.__mro__:
            props.update(prop for prop in class_.__dict__.values()
                if isinstance(prop, property))
        return sorted(props, key=lambda prop: (prop.order, prop.__name__))

    def __init__(self, xml):
        self._xml = xml

        self.services = {}
        self.devices = collections.defaultdict(list)
        self.tags = {}

        if self._xml is None:
            return

        # properties
        all_names = set(prop.__name__ for prop in self.get_props_list())
        for node in self._xml.xpath('.//property'):
            name = node.get('name')
            value = node.get('ref') or node.text

            if not name in all_names:
                raise AttributeError(
                    'No property {!r} found in {!r}'.format(
                        name, self.__class__))

            setattr(self, name, value)

        # tags
        for node in self._xml.xpath('.//tag'):
            self.tags[node.get('name')] = node.text

        # services
        for node in self._xml.xpath('.//service'):
            self.services[node.text] = bool(ast.literal_eval(node.get('enabled', 'True')))

        # devices (pci, usb, ...)
        for parent in self._xml.xpath('.//devices'):
            devclass = parent.get('class')
            for node in parent.xpath('./device'):
                self.devices[devclass].append(node.text)

        # firewall
        #TODO

    def __repr__(self):
        return '<{} object at {:#x} {}>'.format(
            self.__class__.__name__, id(self),
            ' '.join('{}={}'.format(prop.__name__, getattr(self, prop.__name__))
                for prop in self.get_props_list()))

    @classmethod
    def add_hook(cls, event, f):
        '''Add hook to entire VM class and all subclasses

        :param str event: event type
        :param callable f: function to fire on event

        Prototype of the function depends on the exact type of event. Classes
        which inherit from this class will also inherit the hook.
        '''

        cls.__hooks__[event].append(f)

    def fire_hooks(self, event, *args, **kwargs):
        '''Fire hooks associated with an event

        :param str event: event type

        *args* and *kwargs* are passed to each function
        '''

        for cls in self.__class__.__mro__:
            if not hasattr(cls, '__hooks__'): continue
            for hook in cls.__hooks__[event]:
                hook(self, *args, **kwargs)


def load(class_, D):
    cls = BaseVM[class_]
    return cls(D)

__all__ = qubes.plugins.load(__file__)