12 procpath = '/proc/' + str(pid) + '/cmdline'
13 with open(procpath, 'r') as f:
14 cmdline = " ".join(f.readline().split('\0'))
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()
27 if args.system and args.session:
31 # Fetch data from the bus driver
34 bus = dbus.SystemBus()
36 bus = dbus.SessionBus()
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")
44 match_rules = stats_iface.GetAllMatchRules()
46 print("GetConnectionMatchRules failed: did you enable the Stats interface?")
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)
56 rules = dict((k_rules,
58 'wkn': [k for k, v in owners.items() if v == k_rules],
60 'cmd': cmds[k_rules] or "",
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]
69 ) for k_rules, v_rules in match_rules.items())
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'
78 # Print the match rules
80 # print all match rules without analysing them
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))
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:
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))
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))