utils.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166
  1. # encoding=utf-8
  2. #
  3. # The Qubes OS Project, https://www.qubes-os.org/
  4. #
  5. # Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  6. # Copyright (C) 2013-2015 Marek Marczykowski-Górecki
  7. # <marmarek@invisiblethingslab.com>
  8. # Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  9. #
  10. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU Lesser General Public License as published by
  12. # the Free Software Foundation; either version 2.1 of the License, or
  13. # (at your option) any later version.
  14. #
  15. # This program is distributed in the hope that it will be useful,
  16. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  18. # GNU Lesser General Public License for more details.
  19. #
  20. # You should have received a copy of the GNU Lesser General Public License along
  21. # with this program; if not, write to the Free Software Foundation, Inc.,
  22. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  23. #
  24. """Various utility functions."""
  25. import os
  26. import re
  27. import qubesadmin.exc
  28. def parse_size(size):
  29. """Parse human readable size into bytes."""
  30. units = [
  31. ('K', 1000), ('KB', 1000),
  32. ('M', 1000 * 1000), ('MB', 1000 * 1000),
  33. ('G', 1000 * 1000 * 1000), ('GB', 1000 * 1000 * 1000),
  34. ('Ki', 1024), ('KiB', 1024),
  35. ('Mi', 1024 * 1024), ('MiB', 1024 * 1024),
  36. ('Gi', 1024 * 1024 * 1024), ('GiB', 1024 * 1024 * 1024),
  37. ]
  38. size = size.strip().upper()
  39. if size.isdigit():
  40. return int(size)
  41. for unit, multiplier in units:
  42. if size.endswith(unit.upper()):
  43. size = size[:-len(unit)].strip()
  44. return int(size) * multiplier
  45. raise qubesadmin.exc.QubesException("Invalid size: {0}.".format(size))
  46. def mbytes_to_kmg(size):
  47. """Convert mbytes to human readable format."""
  48. if size > 1024:
  49. return "%d GiB" % (size / 1024)
  50. return "%d MiB" % size
  51. def kbytes_to_kmg(size):
  52. """Convert kbytes to human readable format."""
  53. if size > 1024:
  54. return mbytes_to_kmg(size / 1024)
  55. return "%d KiB" % size
  56. def bytes_to_kmg(size):
  57. """Convert bytes to human readable format."""
  58. if size > 1024:
  59. return kbytes_to_kmg(size / 1024)
  60. return "%d B" % size
  61. def size_to_human(size):
  62. """Humane readable size, with 1/10 precision"""
  63. if size < 1024:
  64. return str(size)
  65. if size < 1024 * 1024:
  66. return str(round(size / 1024.0, 1)) + ' KiB'
  67. if size < 1024 * 1024 * 1024:
  68. return str(round(size / (1024.0 * 1024), 1)) + ' MiB'
  69. return str(round(size / (1024.0 * 1024 * 1024), 1)) + ' GiB'
  70. def get_entry_point_one(group, name):
  71. """Get a single entry point of given type,
  72. raise TypeError when there are multiple.
  73. """
  74. import pkg_resources
  75. epoints = tuple(pkg_resources.iter_entry_points(group, name))
  76. if not epoints:
  77. raise KeyError(name)
  78. if len(epoints) > 1:
  79. raise TypeError('more than 1 implementation of {!r} found: {}'.format(
  80. name, ', '.join('{}.{}'.format(ep.module_name, '.'.join(ep.attrs))
  81. for ep in epoints)))
  82. return epoints[0].load()
  83. UPDATES_DEFAULT_VM_DISABLE_FLAG = \
  84. '/var/lib/qubes/updates/vm-default-disable-updates'
  85. def updates_vms_status(qvm_collection):
  86. """Check whether all VMs have the same check-updates value;
  87. if yes, return it; otherwise, return None
  88. """
  89. # default value:
  90. status = not os.path.exists(UPDATES_DEFAULT_VM_DISABLE_FLAG)
  91. # check if all the VMs uses the default value
  92. for vm in qvm_collection.domains:
  93. if vm.qid == 0:
  94. continue
  95. if vm.features.get('check-updates', True) != status:
  96. # "mixed"
  97. return None
  98. return status
  99. def vm_dependencies(app, reference_vm):
  100. """Helper function that returns a list of all the places a given VM is used
  101. in. Output is a list of tuples (property_holder, property_name), with None
  102. as property_holder for global properties
  103. """
  104. result = []
  105. global_properties = ['default_dispvm', 'default_netvm', 'default_guivm',
  106. 'default_template', 'clockvm', 'updatevm',
  107. 'management_dispvm']
  108. for prop in global_properties:
  109. if reference_vm == getattr(app, prop, None):
  110. result.append((None, prop))
  111. vm_properties = ['template', 'netvm', 'guivm',
  112. 'default_dispvm', 'management_dispvm']
  113. for vm in app.domains:
  114. if vm == reference_vm:
  115. continue
  116. for prop in vm_properties:
  117. if reference_vm == getattr(vm, prop, None) and \
  118. not vm.property_is_default(prop):
  119. result.append((vm, prop))
  120. return result
  121. def encode_for_vmexec(args):
  122. """
  123. Encode an argument list for qubes.VMExec call.
  124. """
  125. def encode(part):
  126. if part.group(0) == b'-':
  127. return b'--'
  128. return '-{:02X}'.format(ord(part.group(0))).encode('ascii')
  129. parts = []
  130. for arg in args:
  131. part = re.sub(br'[^a-zA-Z0-9_.+]', encode, arg.encode('utf-8'))
  132. parts.append(part)
  133. return b'+'.join(parts).decode('ascii')