graph.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121
  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 library is free software; you can redistribute it and/or
  9. # modify it under the terms of the GNU Lesser General Public
  10. # License as published by the Free Software Foundation; either
  11. # version 2.1 of the License, or (at your option) any later version.
  12. #
  13. # This library 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 GNU
  16. # Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public
  19. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  20. import argparse
  21. import json
  22. import os
  23. import sys
  24. import qubespolicy
  25. parser = argparse.ArgumentParser(description='Graph qrexec policy')
  26. parser.add_argument('--include-ask', action='store_true',
  27. help='Include `ask` action in graph')
  28. parser.add_argument('--source', action='store', nargs='+',
  29. help='Limit graph to calls from *source*')
  30. parser.add_argument('--target', action='store', nargs='+',
  31. help='Limit graph to calls to *target*')
  32. parser.add_argument('--service', action='store', nargs='+',
  33. help='Limit graph to *service*')
  34. parser.add_argument('--output', action='store',
  35. help='Write to *output* instead of stdout')
  36. parser.add_argument('--policy-dir', action='store',
  37. default=qubespolicy.POLICY_DIR,
  38. help='Look for policy in *policy-dir*')
  39. parser.add_argument('--system-info', action='store',
  40. help='Load system information from file instead of querying qubesd')
  41. parser.add_argument('--skip-labels', action='store_true',
  42. help='Do not include service names on the graph, also deduplicate '
  43. 'connections.')
  44. def handle_single_action(args, action):
  45. '''Get single policy action and output (or not) a line to add'''
  46. if args.skip_labels:
  47. service = ''
  48. else:
  49. service = action.service
  50. target = action.target or action.original_target
  51. # handle forced target=
  52. if action.rule.override_target:
  53. target = action.rule.override_target
  54. if args.target and target not in args.target:
  55. return ''
  56. if action.action == qubespolicy.Action.ask:
  57. if args.include_ask:
  58. return ' "{}" -> "{}" [label="{}" color=orange];\n'.format(
  59. action.source, target, service)
  60. elif action.action == qubespolicy.Action.allow:
  61. return ' "{}" -> "{}" [label="{}" color=red];\n'.format(
  62. action.source, target, service)
  63. return ''
  64. def main(args=None):
  65. args = parser.parse_args(args)
  66. output = sys.stdout
  67. if args.output:
  68. output = open(args.output, 'w')
  69. if args.system_info:
  70. with open(args.system_info) as f_system_info:
  71. system_info = json.load(f_system_info)
  72. else:
  73. system_info = qubespolicy.get_system_info()
  74. sources = list(system_info['domains'].keys())
  75. if args.source:
  76. sources = args.source
  77. targets = list(system_info['domains'].keys())
  78. targets.append('$dispvm')
  79. targets.extend('$dispvm:' + dom for dom in system_info['domains']
  80. if system_info['domains'][dom]['template_for_dispvms'])
  81. connections = set()
  82. output.write('digraph g {\n')
  83. for service in os.listdir(args.policy_dir):
  84. if os.path.isdir(os.path.join(args.policy_dir, service)):
  85. continue
  86. if args.service and service not in args.service and \
  87. not any(service.startswith(srv + '+') for srv in args.service):
  88. continue
  89. policy = qubespolicy.Policy(service, args.policy_dir)
  90. for source in sources:
  91. for target in targets:
  92. try:
  93. action = policy.evaluate(system_info, source, target)
  94. line = handle_single_action(args, action)
  95. if line in connections:
  96. continue
  97. if line:
  98. output.write(line)
  99. connections.add(line)
  100. except qubespolicy.AccessDenied:
  101. continue
  102. output.write('}\n')
  103. if args.output:
  104. output.close()
  105. if __name__ == '__main__':
  106. sys.exit(main())