core-admin/qubespolicy/graph.py
Marek Marczykowski-Górecki 68b6f1ec76
qubespolicy: use '@' instead of '$' for policy keywords
Using '$' is easy to misuse in shell scripts, shell commands etc. After
all this years, lets abandon this dangerous character and move to
something safer: '@'. The choice was made after reviewing specifications
of various shells on different operating systems and this is the
character that have no special meaning in none of them.

To preserve compatibility, automatically translate '$' to '@' when
loading policy files.
2018-02-19 03:33:40 +01:00

122 lines
4.3 KiB
Python

# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
# <marmarek@invisiblethingslab.com>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, see <https://www.gnu.org/licenses/>.
import argparse
import json
import os
import sys
import qubespolicy
parser = argparse.ArgumentParser(description='Graph qrexec policy')
parser.add_argument('--include-ask', action='store_true',
help='Include `ask` action in graph')
parser.add_argument('--source', action='store', nargs='+',
help='Limit graph to calls from *source*')
parser.add_argument('--target', action='store', nargs='+',
help='Limit graph to calls to *target*')
parser.add_argument('--service', action='store', nargs='+',
help='Limit graph to *service*')
parser.add_argument('--output', action='store',
help='Write to *output* instead of stdout')
parser.add_argument('--policy-dir', action='store',
default=qubespolicy.POLICY_DIR,
help='Look for policy in *policy-dir*')
parser.add_argument('--system-info', action='store',
help='Load system information from file instead of querying qubesd')
parser.add_argument('--skip-labels', action='store_true',
help='Do not include service names on the graph, also deduplicate '
'connections.')
def handle_single_action(args, action):
'''Get single policy action and output (or not) a line to add'''
if args.skip_labels:
service = ''
else:
service = action.service
target = action.target or action.original_target
# handle forced target=
if action.rule.override_target:
target = action.rule.override_target
if args.target and target not in args.target:
return ''
if action.action == qubespolicy.Action.ask:
if args.include_ask:
return ' "{}" -> "{}" [label="{}" color=orange];\n'.format(
action.source, target, service)
elif action.action == qubespolicy.Action.allow:
return ' "{}" -> "{}" [label="{}" color=red];\n'.format(
action.source, target, service)
return ''
def main(args=None):
args = parser.parse_args(args)
output = sys.stdout
if args.output:
output = open(args.output, 'w')
if args.system_info:
with open(args.system_info) as f_system_info:
system_info = json.load(f_system_info)
else:
system_info = qubespolicy.get_system_info()
sources = list(system_info['domains'].keys())
if args.source:
sources = args.source
targets = list(system_info['domains'].keys())
targets.append('@dispvm')
targets.extend('@dispvm:' + dom for dom in system_info['domains']
if system_info['domains'][dom]['template_for_dispvms'])
connections = set()
output.write('digraph g {\n')
for service in os.listdir(args.policy_dir):
if os.path.isdir(os.path.join(args.policy_dir, service)):
continue
if args.service and service not in args.service and \
not any(service.startswith(srv + '+') for srv in args.service):
continue
policy = qubespolicy.Policy(service, args.policy_dir)
for source in sources:
for target in targets:
try:
action = policy.evaluate(system_info, source, target)
line = handle_single_action(args, action)
if line in connections:
continue
if line:
output.write(line)
connections.add(line)
except qubespolicy.AccessDenied:
continue
output.write('}\n')
if args.output:
output.close()
if __name__ == '__main__':
sys.exit(main())