tools/GetAllMatchRules.py

   1 #!/usr/bin/env python
   2
   3 import sys
   4 import argparse
   5 import dbus
   6 import time
   7
   8 def get_cmdline(pid):
   9   cmdline = ''
  10   if pid > 0:
  11     try:
  12       procpath = '/proc/' + str(pid) + '/cmdline'
  13       with open(procpath, 'r') as f:
  14         cmdline = " ".join(f.readline().split('\0'))
  15     except:
  16       pass
  17   return cmdline
  18
  19 # Parsing parameters
  20
  21 parser = argparse.ArgumentParser(description='Testing D-Bus match rules')
  22 parser.add_argument('--session', help='session bus', action="store_true")
  23 parser.add_argument('--system', help='system bus', action="store_true")
  24 parser.add_argument('--all', help='print all match rules', action="store_true")
  25 args = parser.parse_args()
  26
  27 if args.system and args.session:
  28   parser.print_help()
  29   sys.exit(1)
  30
  31 # Fetch data from the bus driver
  32
  33 if args.system:
  34   bus = dbus.SystemBus()
  35 else:
  36   bus = dbus.SessionBus()
  37
  38 remote_object = bus.get_object("org.freedesktop.DBus",
  39                                "/org/freedesktop/DBus")
  40 bus_iface = dbus.Interface(remote_object, "org.freedesktop.DBus")
  41 stats_iface = dbus.Interface(remote_object, "org.freedesktop.DBus.Debug.Stats")
  42
  43 try:
  44   match_rules = stats_iface.GetAllMatchRules()
  45 except:
  46   print("GetConnectionMatchRules failed: did you enable the Stats interface?")
  47   sys.exit(1)
  48
  49 names = bus_iface.ListNames()
  50 unique_names = [ a for a in names if a.startswith(":") ]
  51 pids = dict((name, bus_iface.GetConnectionUnixProcessID(name)) for name in unique_names)
  52 cmds = dict((name, get_cmdline(pids[name])) for name in unique_names)
  53 well_known_names = [ a for a in names if a not in unique_names ]
  54 owners = dict((wkn, bus_iface.GetNameOwner(wkn)) for wkn in well_known_names)
  55
  56 rules = dict((k_rules,
  57               dict({
  58                 'wkn': [k for k, v in owners.items() if v == k_rules],
  59                 'pid': pids[k_rules],
  60                 'cmd': cmds[k_rules] or "",
  61                 'rules': v_rules,
  62                 'warnings': dict({
  63                   'not_signal': [a for a in v_rules if "type='signal'" not in a],
  64                   'no_sender': [a for a in v_rules if "sender=" not in a],
  65                   'local': [a for a in v_rules if "org.freedesktop.DBus.Local" in a],
  66                   'NameOwnerChanged_arg0': [a for a in v_rules if "member='NameOwnerChanged'" in a and "arg0" not in a]
  67                 })
  68               })
  69              ) for k_rules, v_rules in match_rules.items())
  70
  71 warnings = dict({
  72              'not_signal': 'Match rule without selecting signals',
  73              'no_sender': 'Match rule without a sender criteria',
  74              'local': 'Match rule on the org.freedesktop.DBus.Local interface',
  75              'NameOwnerChanged_arg0': 'Match rule on NameOwnerChanged without a arg0* criteria'
  76            })
  77
  78 # Print the match rules
  79
  80 # print all match rules without analysing them
  81 if args.all:
  82   for name in rules:
  83     print("Connection %s with pid %d '%s' (%s): %d match rules, %d warnings"
  84           % (name, rules[name]['pid'], rules[name]['cmd'],
  85              ' '.join(rules[name]['wkn']), len(rules[name]['rules']),
  86              len(sum(rules[name]['warnings'].values(), []))))
  87     for rule in rules[name]['rules']:
  88       print("\t%s" % (rule))
  89     print("")
  90   sys.exit(0)
  91
  92 # analyse match rules and print only the suspicious ones
  93 for conn,data in rules.items():
  94   warnings_count = len(sum(data['warnings'].values(), []))
  95   if warnings_count == 0:
  96     continue
  97
  98   print("Connection %s with pid %d '%s' (%s): %d match rules, %d warnings"
  99         % (conn, data['pid'], data['cmd'], ' '.join(data['wkn']),
 100            len(data['rules']), warnings_count))
 101
 102   for warn_code,rule_list in [(warn_code,rule_list) \
 103                               for warn_code, rule_list \
 104                               in data['warnings'].items() \
 105                               if len(rule_list) > 0]:
 106     print("   - %s:" % (warnings[warn_code]))
 107     for rule in rule_list:
 108       print("         - %s" % (rule))
 109
 110   print("")
 111
 112 sys.exit(0)