680 lines
25 KiB
Python
Executable File
680 lines
25 KiB
Python
Executable File
# coding=utf-8
|
|
# The Qubes OS Project, https://www.qubes-os.org/
|
|
#
|
|
# Copyright (C) 2013-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
|
|
# Copyright (C) 2013-2017 Marek Marczykowski-Górecki
|
|
# <marmarek@invisiblethingslab.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program 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 General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License along
|
|
# with this program; if not, write to the Free Software Foundation, Inc.,
|
|
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
|
|
''' Qrexec policy parser and evaluator '''
|
|
import enum
|
|
import itertools
|
|
import json
|
|
import os
|
|
import os.path
|
|
import socket
|
|
import subprocess
|
|
|
|
# don't import 'qubes.config' please, it takes 0.3s
|
|
QREXEC_CLIENT = '/usr/lib/qubes/qrexec-client'
|
|
QUBES_RPC_MULTIPLEXER_PATH = '/usr/lib/qubes/qubes-rpc-multiplexer'
|
|
POLICY_DIR = '/etc/qubes-rpc/policy'
|
|
QUBESD_INTERNAL_SOCK = '/var/run/qubesd.internal.sock'
|
|
|
|
|
|
class AccessDenied(Exception):
|
|
''' Raised when qrexec policy denied access '''
|
|
pass
|
|
|
|
|
|
class PolicySyntaxError(AccessDenied):
|
|
''' Syntax error in qrexec policy, abort parsing '''
|
|
def __init__(self, filename, lineno, msg):
|
|
super(PolicySyntaxError, self).__init__(
|
|
'{}:{}: {}'.format(filename, lineno, msg))
|
|
|
|
|
|
class Action(enum.Enum):
|
|
''' Action as defined by policy '''
|
|
allow = 1
|
|
deny = 2
|
|
ask = 3
|
|
|
|
|
|
def verify_target_value(system_info, value):
|
|
''' Check if given value names valid target
|
|
|
|
This function check if given value is not only syntactically correct,
|
|
but also if names valid service call target (existing domain,
|
|
or valid $dispvm like keyword)
|
|
|
|
:param system_info: information about the system
|
|
:param value: value to be checked
|
|
'''
|
|
if value == '$dispvm':
|
|
return True
|
|
elif value == '$adminvm':
|
|
return True
|
|
elif value.startswith('$dispvm:'):
|
|
dispvm_base = value.split(':', 1)[1]
|
|
if dispvm_base not in system_info['domains']:
|
|
return False
|
|
dispvm_base_info = system_info['domains'][dispvm_base]
|
|
return bool(dispvm_base_info['dispvm_allowed'])
|
|
else:
|
|
return value in system_info['domains']
|
|
|
|
|
|
def verify_special_value(value, for_target=True):
|
|
'''
|
|
Verify if given special VM-specifier ('$...') is valid
|
|
|
|
:param value: value to verify
|
|
:param for_target: should classify target-only values as valid (
|
|
'$default', '$dispvm')
|
|
:return: True or False
|
|
'''
|
|
# pylint: disable=too-many-return-statements
|
|
|
|
if value.startswith('$tag:') and len(value) > len('$tag:'):
|
|
return True
|
|
elif value.startswith('$type:') and len(value) > len('$type:'):
|
|
return True
|
|
elif value == '$anyvm':
|
|
return True
|
|
elif value == '$adminvm':
|
|
return True
|
|
elif value.startswith('$dispvm:') and for_target:
|
|
return True
|
|
elif value == '$dispvm' and for_target:
|
|
return True
|
|
elif value == '$default' and for_target:
|
|
return True
|
|
return False
|
|
|
|
|
|
class PolicyRule(object):
|
|
''' A single line of policy file '''
|
|
def __init__(self, line, filename=None, lineno=None):
|
|
'''
|
|
Load a single line of qrexec policy and check its syntax.
|
|
Do not verify existence of named objects.
|
|
|
|
:raise PolicySyntaxError: when syntax error is found
|
|
|
|
:param line: a single line of actual qrexec policy (not a comment,
|
|
empty line or $include)
|
|
:param filename: name of the file from which this line is loaded
|
|
:param lineno: line number from which this line is loaded
|
|
'''
|
|
|
|
self.lineno = lineno
|
|
self.filename = filename
|
|
|
|
try:
|
|
self.source, self.target, self.full_action = line.split(maxsplit=2)
|
|
except ValueError:
|
|
raise PolicySyntaxError(filename, lineno, 'wrong number of fields')
|
|
|
|
(action, *params) = self.full_action.replace(',', ' ').split()
|
|
try:
|
|
self.action = Action[action]
|
|
except KeyError:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'invalid action: {}'.format(action))
|
|
|
|
#: alternative target, used instead of the one specified by the caller
|
|
self.override_target = None
|
|
|
|
#: alternative user, used instead of vm.default_user
|
|
self.override_user = None
|
|
|
|
#: default target when asking the user for confirmation
|
|
self.default_target = None
|
|
|
|
for param in params:
|
|
try:
|
|
param_name, value = param.split('=')
|
|
except ValueError:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'invalid action parameter syntax: {}'.format(param))
|
|
if param_name == 'target':
|
|
if self.action == Action.deny:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'target= option not allowed for deny action')
|
|
self.override_target = value
|
|
elif param_name == 'user':
|
|
if self.action == Action.deny:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'user= option not allowed for deny action')
|
|
self.override_user = value
|
|
elif param_name == 'default_target':
|
|
if self.action != Action.ask:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'default_target= option allowed only for ask action')
|
|
self.default_target = value
|
|
else:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'invalid option {} for {} action'.format(param, action))
|
|
|
|
# verify special values
|
|
if self.source.startswith('$'):
|
|
if not verify_special_value(self.source, False):
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'invalid source specification: {}'.format(self.source))
|
|
|
|
if self.target.startswith('$'):
|
|
if not verify_special_value(self.target, True):
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'invalid target specification: {}'.format(self.target))
|
|
|
|
if self.target == '$default' \
|
|
and self.action == Action.allow \
|
|
and self.override_target is None:
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'allow action for $default rule must specify target= option')
|
|
|
|
if self.override_target is not None:
|
|
if self.override_target.startswith('$') and \
|
|
not self.override_target.startswith('$dispvm') and \
|
|
self.override_target != '$adminvm':
|
|
raise PolicySyntaxError(filename, lineno,
|
|
'target= option needs to name specific target')
|
|
|
|
@staticmethod
|
|
def is_match_single(system_info, policy_value, value):
|
|
'''
|
|
Evaluate if a single value (VM name or '$default') matches policy
|
|
specification
|
|
|
|
:param system_info: information about the system
|
|
:param policy_value: value from qrexec policy (either self.source or
|
|
self.target)
|
|
:param value: value to be compared (source or target)
|
|
:return: True or False
|
|
'''
|
|
# pylint: disable=too-many-return-statements
|
|
|
|
# not specified target matches only with $default and $anyvm policy
|
|
# entry
|
|
if value == '$default' or value == '':
|
|
return policy_value in ('$default', '$anyvm')
|
|
|
|
# if specific target used, check if it's valid
|
|
# this function (is_match_single) is also used for checking call source
|
|
# values, but this isn't a problem, because it will always be a
|
|
# domain name (not $dispvm or such) - this is guaranteed by a nature
|
|
# of qrexec call
|
|
if not verify_target_value(system_info, value):
|
|
return False
|
|
|
|
# handle $adminvm keyword
|
|
if policy_value == 'dom0':
|
|
# TODO: log a warning in Qubes 4.1
|
|
policy_value = '$adminvm'
|
|
|
|
if value == 'dom0':
|
|
value = '$adminvm'
|
|
|
|
# allow any _valid_, non-dom0 target
|
|
if policy_value == '$anyvm':
|
|
return value != '$adminvm'
|
|
|
|
# exact match, including $dispvm* and $adminvm
|
|
if value == policy_value:
|
|
return True
|
|
|
|
# if $dispvm* not matched above, reject it; missing ':' is
|
|
# intentional - handle both '$dispvm' and '$dispvm:xxx'
|
|
if value.startswith('$dispvm'):
|
|
return False
|
|
|
|
# require $adminvm to be matched explicitly (not through $tag or $type)
|
|
# - if not matched already, reject it
|
|
if value == '$adminvm':
|
|
return False
|
|
|
|
# at this point, value name a specific target
|
|
domain_info = system_info['domains'][value]
|
|
|
|
if policy_value.startswith('$tag:'):
|
|
tag = policy_value.split(':', 1)[1]
|
|
return tag in domain_info['tags']
|
|
|
|
if policy_value.startswith('$type:'):
|
|
type_ = policy_value.split(':', 1)[1]
|
|
return type_ == domain_info['type']
|
|
|
|
return False
|
|
|
|
def is_match(self, system_info, source, target):
|
|
'''
|
|
Check if given (source, target) matches this policy line.
|
|
|
|
:param system_info: information about the system - available VMs,
|
|
their types, labels, tags etc. as returned by
|
|
:py:func:`app_to_system_info`
|
|
:param source: name of the source VM
|
|
:param target: name of the target VM, or None if not specified
|
|
:return: True or False
|
|
'''
|
|
|
|
if not self.is_match_single(system_info, self.source, source):
|
|
return False
|
|
if not self.is_match_single(system_info, self.target, target):
|
|
return False
|
|
return True
|
|
|
|
def expand_target(self, system_info):
|
|
'''
|
|
Return domains matching target of this policy line
|
|
|
|
:param system_info: information about the system
|
|
:return: matching domains
|
|
'''
|
|
|
|
if self.target.startswith('$tag:'):
|
|
tag = self.target.split(':', 1)[1]
|
|
for name, domain in system_info['domains'].items():
|
|
if tag in domain['tags']:
|
|
yield name
|
|
elif self.target.startswith('$type:'):
|
|
type_ = self.target.split(':', 1)[1]
|
|
for name, domain in system_info['domains'].items():
|
|
if type_ == domain['type']:
|
|
yield name
|
|
elif self.target == '$anyvm':
|
|
for name, domain in system_info['domains'].items():
|
|
if name != 'dom0':
|
|
yield name
|
|
if domain['dispvm_allowed']:
|
|
yield '$dispvm:' + name
|
|
yield '$dispvm'
|
|
elif self.target.startswith('$dispvm:'):
|
|
dispvm_base = self.target.split(':', 1)[1]
|
|
try:
|
|
if system_info['domains'][dispvm_base]['dispvm_allowed']:
|
|
yield self.target
|
|
except KeyError:
|
|
# TODO log a warning?
|
|
pass
|
|
elif self.target == '$adminvm':
|
|
yield self.target
|
|
elif self.target == '$dispvm':
|
|
yield self.target
|
|
else:
|
|
if self.target in system_info['domains']:
|
|
yield self.target
|
|
|
|
def expand_override_target(self, system_info, source):
|
|
'''
|
|
Replace '$dispvm' with specific '$dispvm:...' value, based on qrexec
|
|
call source.
|
|
|
|
:param system_info: System information
|
|
:param source: Source domain name
|
|
:return: :py:attr:`override_target` with '$dispvm' substituted
|
|
'''
|
|
if self.override_target == '$dispvm':
|
|
if system_info['domains'][source]['default_dispvm'] is None:
|
|
return None
|
|
return '$dispvm:' + system_info['domains'][source]['default_dispvm']
|
|
else:
|
|
return self.override_target
|
|
|
|
|
|
class PolicyAction(object):
|
|
''' Object representing positive policy evaluation result -
|
|
either ask or allow action '''
|
|
def __init__(self, service, source, target, rule, original_target,
|
|
targets_for_ask=None):
|
|
#: service name
|
|
self.service = service
|
|
#: calling domain
|
|
self.source = source
|
|
#: target domain the service should be connected to, None if
|
|
# not chosen yet
|
|
if targets_for_ask is None or target in targets_for_ask:
|
|
self.target = target
|
|
else:
|
|
# TODO: log a warning?
|
|
self.target = None
|
|
#: original target specified by the caller
|
|
self.original_target = original_target
|
|
#: targets for the user to choose from
|
|
self.targets_for_ask = targets_for_ask
|
|
#: policy rule from which this action is derived
|
|
self.rule = rule
|
|
if rule.action == Action.deny:
|
|
# this should be really rejected by Policy.eval()
|
|
raise AccessDenied(
|
|
'denied by policy {}:{}'.format(rule.filename, rule.lineno))
|
|
elif rule.action == Action.ask:
|
|
assert targets_for_ask is not None
|
|
elif rule.action == Action.allow:
|
|
assert targets_for_ask is None
|
|
assert target is not None
|
|
self.action = rule.action
|
|
|
|
def handle_user_response(self, response, target=None):
|
|
'''
|
|
Handle user response for the 'ask' action
|
|
|
|
:param response: whether the call was allowed or denied (bool)
|
|
:param target: target chosen by the user (if reponse==True)
|
|
:return: None
|
|
'''
|
|
assert self.action == Action.ask
|
|
if response:
|
|
assert target in self.targets_for_ask
|
|
self.target = target
|
|
self.action = Action.allow
|
|
else:
|
|
self.action = Action.deny
|
|
raise AccessDenied(
|
|
'denied by the user {}:{}'.format(self.rule.filename,
|
|
self.rule.lineno))
|
|
|
|
def execute(self, caller_ident):
|
|
''' Execute allowed service call
|
|
|
|
:param caller_ident: Service caller ident
|
|
(`process_ident,source_name, source_id`)
|
|
'''
|
|
assert self.action == Action.allow
|
|
assert self.target is not None
|
|
|
|
if self.target == '$adminvm':
|
|
self.target = 'dom0'
|
|
if self.target == 'dom0':
|
|
cmd = '{multiplexer} {service} {source} {original_target}'.format(
|
|
multiplexer=QUBES_RPC_MULTIPLEXER_PATH,
|
|
service=self.service,
|
|
source=self.source,
|
|
original_target=self.original_target)
|
|
else:
|
|
cmd = '{user}:QUBESRPC {service} {source}'.format(
|
|
user=(self.rule.override_user or 'DEFAULT'),
|
|
service=self.service,
|
|
source=self.source)
|
|
if self.target.startswith('$dispvm:'):
|
|
target = self.spawn_dispvm()
|
|
dispvm = True
|
|
else:
|
|
target = self.target
|
|
dispvm = False
|
|
self.ensure_target_running()
|
|
qrexec_opts = ['-d', target, '-c', caller_ident]
|
|
if dispvm:
|
|
qrexec_opts.append('-W')
|
|
try:
|
|
subprocess.call([QREXEC_CLIENT] + qrexec_opts + [cmd])
|
|
finally:
|
|
if dispvm:
|
|
self.cleanup_dispvm(target)
|
|
|
|
|
|
def spawn_dispvm(self):
|
|
'''
|
|
Create and start Disposable VM based on AppVM specified in
|
|
:py:attr:`target`
|
|
:return: name of new Disposable VM
|
|
'''
|
|
base_appvm = self.target.split(':', 1)[1]
|
|
dispvm_name = qubesd_call(base_appvm, 'internal.vm.Create.DispVM')
|
|
dispvm_name = dispvm_name.decode('ascii')
|
|
qubesd_call(dispvm_name, 'internal.vm.Start')
|
|
return dispvm_name
|
|
|
|
def ensure_target_running(self):
|
|
'''
|
|
Start domain if not running already
|
|
|
|
:return: None
|
|
'''
|
|
try:
|
|
qubesd_call(self.target, 'internal.vm.Start')
|
|
except QubesMgmtException as e:
|
|
if e.exc_type == 'QubesVMNotHaltedError':
|
|
pass
|
|
else:
|
|
raise
|
|
|
|
@staticmethod
|
|
def cleanup_dispvm(dispvm):
|
|
'''
|
|
Kill and remove Disposable VM
|
|
|
|
:param dispvm: name of Disposable VM
|
|
:return: None
|
|
'''
|
|
qubesd_call(dispvm, 'internal.vm.CleanupDispVM')
|
|
|
|
|
|
class Policy(object):
|
|
''' Full policy for a given service
|
|
|
|
Usage:
|
|
>>> system_info = get_system_info()
|
|
>>> policy = Policy('some-service')
|
|
>>> action = policy.evaluate(system_info, 'source-name', 'target-name')
|
|
>>> if action.action == Action.ask:
|
|
>>> # ... ask the user, see action.targets_for_ask ...
|
|
>>> action.handle_user_response(response, target_chosen_by_user)
|
|
>>> action.execute('process-ident')
|
|
|
|
'''
|
|
|
|
def __init__(self, service, policy_dir=POLICY_DIR):
|
|
policy_file = os.path.join(policy_dir, service)
|
|
if not os.path.exists(policy_file):
|
|
# fallback to policy without specific argument set (if any)
|
|
policy_file = os.path.join(policy_dir, service.split('+')[0])
|
|
|
|
#: policy storage directory
|
|
self.policy_dir = policy_dir
|
|
|
|
#: service name
|
|
self.service = service
|
|
|
|
#: list of PolicyLine objects
|
|
self.policy_rules = []
|
|
try:
|
|
self.load_policy_file(policy_file)
|
|
except OSError as e:
|
|
raise AccessDenied(
|
|
'failed to load {} file: {!s}'.format(e.filename, e))
|
|
|
|
def load_policy_file(self, path):
|
|
''' Load policy file and append rules to :py:attr:`policy_rules`
|
|
|
|
:param path: file to load
|
|
'''
|
|
with open(path) as policy_file:
|
|
for lineno, line in zip(itertools.count(start=1),
|
|
policy_file.readlines()):
|
|
line = line.strip()
|
|
if not line:
|
|
# skip empty lines
|
|
continue
|
|
if line[0] == '#':
|
|
# skip comments
|
|
continue
|
|
if line.startswith('$include:'):
|
|
include_path = line.split(':', 1)[1]
|
|
# os.path.join will leave include_path unchanged if it's
|
|
# already absolute
|
|
include_path = os.path.join(self.policy_dir, include_path)
|
|
self.load_policy_file(include_path)
|
|
else:
|
|
self.policy_rules.append(PolicyRule(line, path, lineno))
|
|
|
|
def find_matching_rule(self, system_info, source, target):
|
|
''' Find the first rule matching given arguments '''
|
|
|
|
for rule in self.policy_rules:
|
|
if rule.is_match(system_info, source, target):
|
|
return rule
|
|
raise AccessDenied('no matching rule found')
|
|
|
|
|
|
def collect_targets_for_ask(self, system_info, source):
|
|
''' Collect targets the user can choose from in 'ask' action
|
|
|
|
Word 'targets' is used intentionally instead of 'domains', because it
|
|
can also contains $dispvm like keywords.
|
|
'''
|
|
targets = set()
|
|
|
|
# iterate over rules in reversed order to easier handle 'deny'
|
|
# actions - simply remove matching domains from allowed set
|
|
for rule in reversed(self.policy_rules):
|
|
if rule.is_match_single(system_info, rule.source, source):
|
|
if rule.action == Action.deny:
|
|
targets -= set(rule.expand_target(system_info))
|
|
else:
|
|
if rule.override_target is not None:
|
|
override_target = rule.expand_override_target(
|
|
system_info, source)
|
|
if verify_target_value(system_info, override_target):
|
|
targets.add(rule.override_target)
|
|
else:
|
|
targets.update(rule.expand_target(system_info))
|
|
|
|
# expand default DispVM
|
|
if '$dispvm' in targets:
|
|
targets.remove('$dispvm')
|
|
if system_info['domains'][source]['default_dispvm'] is not None:
|
|
dispvm = '$dispvm:' + \
|
|
system_info['domains'][source]['default_dispvm']
|
|
if verify_target_value(system_info, dispvm):
|
|
targets.add(dispvm)
|
|
|
|
return targets
|
|
|
|
def evaluate(self, system_info, source, target):
|
|
''' Evaluate policy
|
|
|
|
:raise AccessDenied: when action should be denied unconditionally
|
|
|
|
:return tuple(rule, considered_targets) - where considered targets is a
|
|
list of possible targets for 'ask' action (rule.action == Action.ask)
|
|
'''
|
|
rule = self.find_matching_rule(system_info, source, target)
|
|
if rule.action == Action.deny:
|
|
raise AccessDenied(
|
|
'denied by policy {}:{}'.format(rule.filename, rule.lineno))
|
|
|
|
if rule.override_target is not None:
|
|
override_target = rule.expand_override_target(system_info, source)
|
|
if not verify_target_value(system_info, override_target):
|
|
raise AccessDenied('invalid target= value in {}:{}'.format(
|
|
rule.filename, rule.lineno))
|
|
actual_target = override_target
|
|
else:
|
|
actual_target = target
|
|
|
|
if rule.action == Action.ask:
|
|
if rule.override_target is not None:
|
|
targets = [actual_target]
|
|
else:
|
|
targets = list(
|
|
self.collect_targets_for_ask(system_info, source))
|
|
if not targets:
|
|
raise AccessDenied(
|
|
'policy define \'ask\' action at {}:{} but no target is '
|
|
'available to choose from'.format(
|
|
rule.filename, rule.lineno))
|
|
return PolicyAction(self.service, source, rule.default_target,
|
|
rule, target, targets)
|
|
elif rule.action == Action.allow:
|
|
if actual_target == '$default':
|
|
raise AccessDenied(
|
|
'policy define \'allow\' action at {}:{} but no target is '
|
|
'specified by caller or policy'.format(
|
|
rule.filename, rule.lineno))
|
|
if actual_target == '$dispvm':
|
|
if system_info['domains'][source]['default_dispvm'] is None:
|
|
raise AccessDenied(
|
|
'policy define \'allow\' action to $dispvm at {}:{} '
|
|
'but no DispVM base is set for this VM'.format(
|
|
rule.filename, rule.lineno))
|
|
actual_target = '$dispvm:' + \
|
|
system_info['domains'][source]['default_dispvm']
|
|
|
|
return PolicyAction(self.service, source,
|
|
actual_target, rule, target)
|
|
else:
|
|
# should be unreachable
|
|
raise AccessDenied(
|
|
'invalid action?! {}:{}'.format(rule.filename, rule.lineno))
|
|
|
|
|
|
class QubesMgmtException(Exception):
|
|
''' Exception returned by qubesd '''
|
|
def __init__(self, exc_type):
|
|
super(QubesMgmtException, self).__init__()
|
|
self.exc_type = exc_type
|
|
|
|
|
|
def qubesd_call(dest, method, arg=None, payload=None):
|
|
try:
|
|
client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
client_socket.connect(QUBESD_INTERNAL_SOCK)
|
|
except IOError:
|
|
# TODO:
|
|
raise
|
|
|
|
# src, method, dest, arg
|
|
for call_arg in ('dom0', method, dest, arg):
|
|
if call_arg is not None:
|
|
client_socket.sendall(call_arg.encode('ascii'))
|
|
client_socket.sendall(b'\0')
|
|
if payload is not None:
|
|
client_socket.sendall(payload)
|
|
|
|
client_socket.shutdown(socket.SHUT_WR)
|
|
|
|
return_data = client_socket.makefile('rb').read()
|
|
if return_data.startswith(b'0\x00'):
|
|
return return_data[2:]
|
|
elif return_data.startswith(b'2\x00'):
|
|
(_, exc_type, _traceback, _format_string, _args) = \
|
|
return_data.split(b'\x00', 4)
|
|
raise QubesMgmtException(exc_type.decode('ascii'))
|
|
else:
|
|
raise AssertionError(
|
|
'invalid qubesd response: {!r}'.format(return_data))
|
|
|
|
|
|
def get_system_info():
|
|
''' Get system information
|
|
|
|
This retrieve information necessary to process qrexec policy. Returned
|
|
data is nested dict structure with this structure:
|
|
|
|
- domains:
|
|
- `<domain name>`:
|
|
- tags: list of tags
|
|
- type: domain type
|
|
- dispvm_allowed: should DispVM based on this VM be allowed
|
|
- default_dispvm: name of default AppVM for DispVMs started from here
|
|
|
|
'''
|
|
|
|
system_info = qubesd_call('dom0', 'internal.GetSystemInfo')
|
|
return json.loads(system_info.decode('utf-8'))
|