core3.py 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2.1 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. '''Parser for core2 qubes.xml'''
  21. import xml.parsers
  22. import logging
  23. import lxml.etree
  24. import qubesadmin.backup
  25. import qubesadmin.firewall
  26. class Core3VM(qubesadmin.backup.BackupVM):
  27. '''VM object'''
  28. # pylint: disable=too-few-public-methods
  29. @property
  30. def included_in_backup(self):
  31. return self.backup_path is not None
  32. def handle_firewall_xml(self, vm, stream):
  33. '''Load new (Qubes >= 4.0) firewall XML format'''
  34. try:
  35. tree = lxml.etree.parse(stream) # pylint: disable=no-member
  36. xml_root = tree.getroot()
  37. rules = []
  38. for rule_node in xml_root.findall('./rules/rule'):
  39. rule_opts = {}
  40. for rule_opt in rule_node.findall('./properties/property'):
  41. rule_opts[rule_opt.get('name')] = rule_opt.text
  42. rules.append(qubesadmin.firewall.Rule(None, **rule_opts))
  43. vm.firewall.rules = rules
  44. except: # pylint: disable=bare-except
  45. vm.log.exception('Failed to set firewall')
  46. class Core3Qubes(qubesadmin.backup.BackupApp):
  47. '''Parsed qubes.xml'''
  48. def __init__(self, store=None):
  49. if store is None:
  50. raise ValueError("store path required")
  51. self.log = logging.getLogger('qubesadmin.backup.core3')
  52. self.labels = {}
  53. super(Core3Qubes, self).__init__(store)
  54. @staticmethod
  55. def get_property(xml_obj, prop):
  56. '''Get property of given object (XML node)
  57. Object can be any PropertyHolder serialized to XML - in practice
  58. :py:class:`BaseVM` or :py:class:`Qubes`.
  59. '''
  60. xml_prop = xml_obj.findall('./property[@name=\'{}\']'.format(prop))
  61. if not xml_prop:
  62. raise KeyError(prop)
  63. return xml_prop[0].text
  64. def load_labels(self, labels_element):
  65. '''Load labels table'''
  66. for node in labels_element.findall('label'):
  67. ident = node.get('id')
  68. assert ident is not None
  69. self.labels[ident] = node.text
  70. def load_globals(self, globals_element):
  71. '''Load global settings
  72. :param globals_element: XML element containing global settings
  73. '''
  74. for node in globals_element.findall('property'):
  75. name = node.get('name')
  76. assert name is not None
  77. self.globals[name] = node.text
  78. def import_core3_vm(self, element):
  79. '''Parse a single VM from given XML node
  80. This method load only VM properties not depending on other VMs
  81. (other than template). VM connections are set later.
  82. :param element: XML node
  83. '''
  84. vm = Core3VM()
  85. vm.klass = element.get('class')
  86. for node in element.findall('./properties/property'):
  87. name = node.get('name')
  88. assert name is not None
  89. vm.properties[name] = node.text
  90. for node in element.findall('./features/feature'):
  91. name = node.get('name')
  92. assert name is not None
  93. vm.features[name] = False if node.text is None else node.text
  94. for node in element.findall('./tags/tag'):
  95. name = node.get('name')
  96. assert name is not None
  97. vm.tags.add(name)
  98. for bus_node in element.findall('./devices'):
  99. bus_name = bus_node.get('class')
  100. assert bus_name is not None
  101. for node in bus_node.findall('./device'):
  102. backend_domain = node.get('backend-domain')
  103. ident = node.get('id')
  104. options = {}
  105. for opt_node in node.findall('./option'):
  106. opt_name = opt_node.get('name')
  107. options[opt_name] = opt_node.text
  108. vm.devices[bus_name][(backend_domain, ident)] = options
  109. # extract base properties
  110. if vm.klass == 'AdminVM':
  111. vm.name = 'dom0'
  112. else:
  113. vm.name = vm.properties.pop('name')
  114. vm.label = self.labels[vm.properties.pop('label')]
  115. vm.template = vm.properties.pop('template', None)
  116. # skip UUID and qid, will be generated during restore
  117. vm.properties.pop('uuid', None)
  118. vm.properties.pop('qid', None)
  119. if vm.features.pop('backup-content', False):
  120. vm.backup_path = vm.features.pop('backup-path', None)
  121. vm.size = vm.features.pop('backup-size', 0)
  122. self.domains[vm.name] = vm
  123. def load(self):
  124. with open(self.store) as fh:
  125. try:
  126. # pylint: disable=no-member
  127. tree = lxml.etree.parse(fh)
  128. except (EnvironmentError, # pylint: disable=broad-except
  129. xml.parsers.expat.ExpatError) as err:
  130. self.log.error(err)
  131. return False
  132. self.load_labels(tree.find('./labels'))
  133. for element in tree.findall('./domains/domain'):
  134. self.import_core3_vm(element)
  135. # and load other defaults (default netvm, updatevm etc)
  136. self.load_globals(tree.find('./properties'))