139
136
help="Select all clients")
140
137
parser.add_argument("-v", "--verbose", action="store_true",
141
138
help="Print all fields")
142
parser.add_argument("-j", "--dump-json", action="store_true",
139
parser.add_argument("-j", "--dump-json", dest="commands",
140
action="append_const", default=[],
141
const=command.DumpJSON(),
143
142
help="Dump client data in JSON format")
144
143
enable_disable = parser.add_mutually_exclusive_group()
145
enable_disable.add_argument("-e", "--enable", action="store_true",
144
enable_disable.add_argument("-e", "--enable", dest="commands",
145
action="append_const", default=[],
146
const=command.Enable(),
146
147
help="Enable client")
147
enable_disable.add_argument("-d", "--disable",
148
enable_disable.add_argument("-d", "--disable", dest="commands",
149
action="append_const", default=[],
150
const=command.Disable(),
149
151
help="disable client")
150
parser.add_argument("-b", "--bump-timeout", action="store_true",
152
parser.add_argument("-b", "--bump-timeout", dest="commands",
153
action="append_const", default=[],
154
const=command.BumpTimeout(),
151
155
help="Bump timeout for client")
152
156
start_stop_checker = parser.add_mutually_exclusive_group()
153
157
start_stop_checker.add_argument("--start-checker",
159
action="append_const", default=[],
160
const=command.StartChecker(),
155
161
help="Start checker for client")
156
start_stop_checker.add_argument("--stop-checker",
162
start_stop_checker.add_argument("--stop-checker", dest="commands",
163
action="append_const", default=[],
164
const=command.StopChecker(),
158
165
help="Stop checker for client")
159
parser.add_argument("-V", "--is-enabled", action="store_true",
166
parser.add_argument("-V", "--is-enabled", dest="commands",
167
action="append_const", default=[],
168
const=command.IsEnabled(),
160
169
help="Check if client is enabled")
161
parser.add_argument("-r", "--remove", action="store_true",
170
parser.add_argument("-r", "--remove", dest="commands",
171
action="append_const", default=[],
172
const=command.Remove(),
162
173
help="Remove client")
163
parser.add_argument("-c", "--checker",
174
parser.add_argument("-c", "--checker", dest="commands",
175
action="append", default=[],
176
metavar="COMMAND", type=command.SetChecker,
164
177
help="Set checker command for client")
165
parser.add_argument("-t", "--timeout", type=string_to_delta,
166
help="Set timeout for client")
167
parser.add_argument("--extended-timeout", type=string_to_delta,
168
help="Set extended timeout for client")
169
parser.add_argument("-i", "--interval", type=string_to_delta,
170
help="Set checker interval for client")
179
"-t", "--timeout", dest="commands", action="append",
180
default=[], metavar="TIME",
181
type=command.SetTimeout.argparse(string_to_delta),
182
help="Set timeout for client")
184
"--extended-timeout", dest="commands", action="append",
185
default=[], metavar="TIME",
186
type=command.SetExtendedTimeout.argparse(string_to_delta),
187
help="Set extended timeout for client")
189
"-i", "--interval", dest="commands", action="append",
190
default=[], metavar="TIME",
191
type=command.SetInterval.argparse(string_to_delta),
192
help="Set checker interval for client")
171
193
approve_deny_default = parser.add_mutually_exclusive_group()
172
194
approve_deny_default.add_argument(
173
"--approve-by-default", action="store_true",
174
default=None, dest="approved_by_default",
195
"--approve-by-default", dest="commands",
196
action="append_const", default=[],
197
const=command.ApproveByDefault(),
175
198
help="Set client to be approved by default")
176
199
approve_deny_default.add_argument(
177
"--deny-by-default", action="store_false",
178
dest="approved_by_default",
200
"--deny-by-default", dest="commands",
201
action="append_const", default=[],
202
const=command.DenyByDefault(),
179
203
help="Set client to be denied by default")
180
parser.add_argument("--approval-delay", type=string_to_delta,
181
help="Set delay before client approve/deny")
182
parser.add_argument("--approval-duration", type=string_to_delta,
183
help="Set duration of one client approval")
184
parser.add_argument("-H", "--host", help="Set host for client")
185
parser.add_argument("-s", "--secret",
186
type=argparse.FileType(mode="rb"),
187
help="Set password blob (file) for client")
205
"--approval-delay", dest="commands", action="append",
206
default=[], metavar="TIME",
207
type=command.SetApprovalDelay.argparse(string_to_delta),
208
help="Set delay before client approve/deny")
210
"--approval-duration", dest="commands", action="append",
211
default=[], metavar="TIME",
212
type=command.SetApprovalDuration.argparse(string_to_delta),
213
help="Set duration of one client approval")
214
parser.add_argument("-H", "--host", dest="commands",
215
action="append", default=[], metavar="STRING",
216
type=command.SetHost,
217
help="Set host for client")
219
"-s", "--secret", dest="commands", action="append",
220
default=[], metavar="FILENAME",
221
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
222
help="Set password blob (file) for client")
188
223
approve_deny = parser.add_mutually_exclusive_group()
189
224
approve_deny.add_argument(
190
"-A", "--approve", action="store_true",
225
"-A", "--approve", dest="commands", action="append_const",
226
default=[], const=command.Approve(),
191
227
help="Approve any current client request")
192
approve_deny.add_argument("-D", "--deny", action="store_true",
228
approve_deny.add_argument("-D", "--deny", dest="commands",
229
action="append_const", default=[],
230
const=command.Deny(),
193
231
help="Deny any current client request")
194
232
parser.add_argument("--debug", action="store_true",
195
233
help="Debug mode (show D-Bus commands)")
386
425
"""Apply additional restrictions on options, not expressible in
389
def has_actions(options):
390
return any((options.enable,
392
options.bump_timeout,
393
options.start_checker,
394
options.stop_checker,
397
options.checker is not None,
398
options.timeout is not None,
399
options.extended_timeout is not None,
400
options.interval is not None,
401
options.approved_by_default is not None,
402
options.approval_delay is not None,
403
options.approval_duration is not None,
404
options.host is not None,
405
options.secret is not None,
428
def has_commands(options, commands=None):
430
commands = (command.Enable,
433
command.StartChecker,
439
command.SetExtendedTimeout,
441
command.ApproveByDefault,
442
command.DenyByDefault,
443
command.SetApprovalDelay,
444
command.SetApprovalDuration,
449
return any(isinstance(cmd, commands)
450
for cmd in options.commands)
409
if has_actions(options) and not (options.client or options.all):
452
if has_commands(options) and not (options.client or options.all):
410
453
parser.error("Options require clients names or --all.")
411
if options.verbose and has_actions(options):
454
if options.verbose and has_commands(options):
412
455
parser.error("--verbose can only be used alone.")
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
456
if (has_commands(options, (command.DumpJSON,))
457
and (options.verbose or len(options.commands) > 1)):
415
458
parser.error("--dump-json can only be used alone.")
416
if options.all and not has_actions(options):
459
if options.all and not has_commands(options):
417
460
parser.error("--all requires an action.")
418
if options.is_enabled and len(options.client) > 1:
461
if (has_commands(options, (command.IsEnabled,))
462
and len(options.client) > 1):
419
463
parser.error("--is-enabled requires exactly one client")
421
options.remove = False
422
if has_actions(options) and not options.deny:
423
parser.error("--remove can only be combined with --deny")
424
options.remove = True
427
def get_mandos_dbus_object(bus):
428
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
429
dbus_busname, server_dbus_path)
430
with if_dbus_exception_log_with_exception_and_exit(
431
"Could not connect to Mandos server: %s"):
432
mandos_dbus_object = bus.get_object(dbus_busname,
434
return mandos_dbus_object
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
446
def get_managed_objects(object_manager):
447
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
448
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
449
with if_dbus_exception_log_with_exception_and_exit(
450
"Failed to access Mandos server through D-Bus:\n%s"):
451
with SilenceLogger("dbus.proxies"):
452
managed_objects = object_manager.GetManagedObjects()
453
return managed_objects
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
462
self.logger.addFilter(self.nullfilter)
464
class NullFilter(logging.Filter):
465
def filter(self, record):
468
nullfilter = NullFilter()
470
def __exit__(self, exc_type, exc_val, exc_tb):
471
self.logger.removeFilter(self.nullfilter)
464
if (len(options.commands) > 1
465
and has_commands(options, (command.Remove,))
466
and not has_commands(options, (command.Deny,))):
467
parser.error("--remove can only be combined with --deny")
472
class SystemBus(object):
474
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
475
def get_managed_objects(self, busname, objectpath):
476
return self.call_method("GetManagedObjects", busname,
478
self.object_manager_iface)
480
properties_iface = "org.freedesktop.DBus.Properties"
481
def set_property(self, busname, objectpath, interface, key,
483
self.call_method("Set", busname, objectpath,
484
self.properties_iface, interface, key,
488
class MandosBus(SystemBus):
489
busname_domain = "se.recompile"
490
busname = busname_domain + ".Mandos"
492
server_interface = busname_domain + ".Mandos"
493
client_interface = busname_domain + ".Mandos.Client"
496
def get_clients_and_properties(self):
497
managed_objects = self.get_managed_objects(
498
self.busname, self.server_path)
499
return {objpath: properties[self.client_interface]
500
for objpath, properties in managed_objects.items()
501
if self.client_interface in properties}
503
def set_client_property(self, objectpath, key, value):
504
return self.set_property(self.busname, objectpath,
505
self.client_interface, key,
508
def call_client_method(self, objectpath, method, *args):
509
return self.call_method(method, self.busname, objectpath,
510
self.client_interface, *args)
512
def call_server_method(self, method, *args):
513
return self.call_method(method, self.busname,
515
self.server_interface, *args)
517
class Error(Exception):
520
class ConnectFailed(Error):
524
class dbus_python_adapter(object):
526
class SystemBus(dbus.MandosBus):
527
"""Use dbus-python"""
529
def __init__(self, module=dbus_python):
530
self.dbus_python = module
531
self.bus = self.dbus_python.SystemBus()
533
@contextlib.contextmanager
534
def convert_exception(self, exception_class=dbus.Error):
537
except self.dbus_python.exceptions.DBusException as e:
538
# This does what "raise from" would do
539
exc = exception_class(*e.args)
543
def call_method(self, methodname, busname, objectpath,
545
proxy_object = self.get_object(busname, objectpath)
546
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
547
interface, methodname,
548
", ".join(repr(a) for a in args))
549
method = getattr(proxy_object, methodname)
550
with self.convert_exception():
551
with dbus_python_adapter.SilenceLogger(
553
value = method(*args, dbus_interface=interface)
554
return self.type_filter(value)
556
def get_object(self, busname, objectpath):
557
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
559
with self.convert_exception(dbus.ConnectFailed):
560
return self.bus.get_object(busname, objectpath)
562
def type_filter(self, value):
563
"""Convert the most bothersome types to Python types"""
564
if isinstance(value, self.dbus_python.Boolean):
566
if isinstance(value, self.dbus_python.ObjectPath):
568
# Also recurse into dictionaries
569
if isinstance(value, self.dbus_python.Dictionary):
570
return {self.type_filter(key):
571
self.type_filter(subval)
572
for key, subval in value.items()}
575
def set_client_property(self, objectpath, key, value):
577
if not isinstance(value, bytes):
578
value = value.encode("utf-8")
579
value = self.dbus_python.ByteArray(value)
580
return self.set_property(self.busname, objectpath,
581
self.client_interface, key,
584
class SilenceLogger(object):
585
"Simple context manager to silence a particular logger"
586
def __init__(self, loggername):
587
self.logger = logging.getLogger(loggername)
590
self.logger.addFilter(self.nullfilter)
592
class NullFilter(logging.Filter):
593
def filter(self, record):
596
nullfilter = NullFilter()
598
def __exit__(self, exc_type, exc_val, exc_tb):
599
self.logger.removeFilter(self.nullfilter)
602
class CachingBus(SystemBus):
603
"""A caching layer for dbus_python_adapter.SystemBus"""
604
def __init__(self, *args, **kwargs):
605
self.object_cache = {}
606
super(dbus_python_adapter.CachingBus,
607
self).__init__(*args, **kwargs)
608
def get_object(self, busname, objectpath):
610
return self.object_cache[(busname, objectpath)]
613
dbus_python_adapter.CachingBus,
614
self).get_object(busname, objectpath)
615
self.object_cache[(busname, objectpath)] = new_object
619
class pydbus_adapter(object):
620
class SystemBus(dbus.MandosBus):
621
def __init__(self, module=pydbus):
623
self.bus = self.pydbus.SystemBus()
625
@contextlib.contextmanager
626
def convert_exception(self, exception_class=dbus.Error):
629
except gi.repository.GLib.Error as e:
630
# This does what "raise from" would do
631
exc = exception_class(*e.args)
635
def call_method(self, methodname, busname, objectpath,
637
proxy_object = self.get(busname, objectpath)
638
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
639
interface, methodname,
640
", ".join(repr(a) for a in args))
641
method = getattr(proxy_object[interface], methodname)
642
with self.convert_exception():
645
def get(self, busname, objectpath):
646
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
648
with self.convert_exception(dbus.ConnectFailed):
649
if sys.version_info.major <= 2:
650
with warnings.catch_warnings():
651
warnings.filterwarnings(
652
"ignore", "", DeprecationWarning,
653
r"^xml\.etree\.ElementTree$")
654
return self.bus.get(busname, objectpath)
656
return self.bus.get(busname, objectpath)
658
def set_property(self, busname, objectpath, interface, key,
660
proxy_object = self.get(busname, objectpath)
661
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
662
objectpath, self.properties_iface, interface,
664
setattr(proxy_object[interface], key, value)
666
class CachingBus(SystemBus):
667
"""A caching layer for pydbus_adapter.SystemBus"""
668
def __init__(self, *args, **kwargs):
669
self.object_cache = {}
670
super(pydbus_adapter.CachingBus,
671
self).__init__(*args, **kwargs)
672
def get(self, busname, objectpath):
674
return self.object_cache[(busname, objectpath)]
676
new_object = (super(pydbus_adapter.CachingBus, self)
677
.get(busname, objectpath))
678
self.object_cache[(busname, objectpath)] = new_object
474
682
def commands_from_options(options):
478
if options.is_enabled:
479
commands.append(command.IsEnabled())
482
commands.append(command.Approve())
485
commands.append(command.Deny())
488
commands.append(command.Remove())
490
if options.dump_json:
491
commands.append(command.DumpJSON())
494
commands.append(command.Enable())
497
commands.append(command.Disable())
499
if options.bump_timeout:
500
commands.append(command.BumpTimeout())
502
if options.start_checker:
503
commands.append(command.StartChecker())
505
if options.stop_checker:
506
commands.append(command.StopChecker())
508
if options.approved_by_default is not None:
509
if options.approved_by_default:
510
commands.append(command.ApproveByDefault())
684
commands = list(options.commands)
686
def find_cmd(cmd, commands):
688
for i, c in enumerate(commands):
689
if isinstance(c, cmd):
693
# If command.Remove is present, move any instances of command.Deny
694
# to occur ahead of command.Remove.
695
index_of_remove = find_cmd(command.Remove, commands)
696
before_remove = commands[:index_of_remove]
697
after_remove = commands[index_of_remove:]
699
for cmd in after_remove:
700
if isinstance(cmd, command.Deny):
701
before_remove.append(cmd)
512
commands.append(command.DenyByDefault())
514
if options.checker is not None:
515
commands.append(command.SetChecker(options.checker))
517
if options.host is not None:
518
commands.append(command.SetHost(options.host))
520
if options.secret is not None:
521
commands.append(command.SetSecret(options.secret))
523
if options.timeout is not None:
524
commands.append(command.SetTimeout(options.timeout))
526
if options.extended_timeout:
528
command.SetExtendedTimeout(options.extended_timeout))
530
if options.interval is not None:
531
commands.append(command.SetInterval(options.interval))
533
if options.approval_delay is not None:
535
command.SetApprovalDelay(options.approval_delay))
537
if options.approval_duration is not None:
539
command.SetApprovalDuration(options.approval_duration))
703
cleaned_after.append(cmd)
704
if cleaned_after != after_remove:
705
commands = before_remove + cleaned_after
541
707
# If no command option has been given, show table of clients,
542
708
# optionally verbosely
970
1125
def test_actions_all_conflicts_with_verbose(self):
971
1126
for action, value in self.actions.items():
972
options = self.parser.parse_args()
973
setattr(options, action, value)
975
options.verbose = True
1127
args = self.actionargs(action, value, "--all",
976
1129
with self.assertParseError():
977
self.check_option_syntax(options)
1130
self.parse_args(args)
979
1132
def test_actions_with_client_conflicts_with_verbose(self):
980
1133
for action, value in self.actions.items():
981
options = self.parser.parse_args()
982
setattr(options, action, value)
983
options.verbose = True
984
options.client = ["client"]
1134
args = self.actionargs(action, value, "--verbose",
985
1136
with self.assertParseError():
986
self.check_option_syntax(options)
1137
self.parse_args(args)
988
1139
def test_dump_json_conflicts_with_verbose(self):
989
options = self.parser.parse_args()
990
options.dump_json = True
991
options.verbose = True
1140
args = ["--dump-json", "--verbose"]
992
1141
with self.assertParseError():
993
self.check_option_syntax(options)
1142
self.parse_args(args)
995
1144
def test_dump_json_conflicts_with_action(self):
996
1145
for action, value in self.actions.items():
997
options = self.parser.parse_args()
998
setattr(options, action, value)
999
options.dump_json = True
1146
args = self.actionargs(action, value, "--dump-json")
1000
1147
with self.assertParseError():
1001
self.check_option_syntax(options)
1148
self.parse_args(args)
1003
1150
def test_all_can_not_be_alone(self):
1004
options = self.parser.parse_args()
1006
1152
with self.assertParseError():
1007
self.check_option_syntax(options)
1153
self.parse_args(args)
1009
1155
def test_all_is_ok_with_any_action(self):
1010
1156
for action, value in self.actions.items():
1011
options = self.parser.parse_args()
1012
setattr(options, action, value)
1014
self.check_option_syntax(options)
1157
args = self.actionargs(action, value, "--all")
1158
self.parse_args(args)
1016
1160
def test_any_action_is_ok_with_one_client(self):
1017
1161
for action, value in self.actions.items():
1018
options = self.parser.parse_args()
1019
setattr(options, action, value)
1020
options.client = ["client"]
1021
self.check_option_syntax(options)
1162
args = self.actionargs(action, value, "client")
1163
self.parse_args(args)
1023
1165
def test_one_client_with_all_actions_except_is_enabled(self):
1024
options = self.parser.parse_args()
1025
1166
for action, value in self.actions.items():
1026
if action == "is_enabled":
1167
if action == "--is-enabled":
1028
setattr(options, action, value)
1029
options.client = ["client"]
1030
self.check_option_syntax(options)
1169
args = self.actionargs(action, value, "client")
1170
self.parse_args(args)
1032
1172
def test_two_clients_with_all_actions_except_is_enabled(self):
1033
options = self.parser.parse_args()
1034
1173
for action, value in self.actions.items():
1035
if action == "is_enabled":
1174
if action == "--is-enabled":
1037
setattr(options, action, value)
1038
options.client = ["client1", "client2"]
1039
self.check_option_syntax(options)
1176
args = self.actionargs(action, value, "client1",
1178
self.parse_args(args)
1041
1180
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1042
1181
for action, value in self.actions.items():
1043
if action == "is_enabled":
1182
if action == "--is-enabled":
1045
options = self.parser.parse_args()
1046
setattr(options, action, value)
1047
options.client = ["client1", "client2"]
1048
self.check_option_syntax(options)
1184
args = self.actionargs(action, value, "client1",
1186
self.parse_args(args)
1050
1188
def test_is_enabled_fails_without_client(self):
1051
options = self.parser.parse_args()
1052
options.is_enabled = True
1189
args = ["--is-enabled"]
1053
1190
with self.assertParseError():
1054
self.check_option_syntax(options)
1191
self.parse_args(args)
1056
1193
def test_is_enabled_fails_with_two_clients(self):
1057
options = self.parser.parse_args()
1058
options.is_enabled = True
1059
options.client = ["client1", "client2"]
1194
args = ["--is-enabled", "client1", "client2"]
1060
1195
with self.assertParseError():
1061
self.check_option_syntax(options)
1196
self.parse_args(args)
1063
1198
def test_remove_can_only_be_combined_with_action_deny(self):
1064
1199
for action, value in self.actions.items():
1065
if action in {"remove", "deny"}:
1200
if action in {"--remove", "--deny"}:
1067
options = self.parser.parse_args()
1068
setattr(options, action, value)
1070
options.remove = True
1202
args = self.actionargs(action, value, "--all",
1071
1204
with self.assertParseError():
1072
self.check_option_syntax(options)
1075
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1076
def test_calls_and_returns_get_object_on_bus(self):
1077
class MockBus(object):
1079
def get_object(mockbus_self, busname, dbus_path):
1080
# Note that "self" is still the testcase instance,
1081
# this MockBus instance is in "mockbus_self".
1082
self.assertEqual(dbus_busname, busname)
1083
self.assertEqual(server_dbus_path, dbus_path)
1084
mockbus_self.called = True
1087
mockbus = get_mandos_dbus_object(bus=MockBus())
1088
self.assertIsInstance(mockbus, MockBus)
1089
self.assertTrue(mockbus.called)
1091
def test_logs_and_exits_on_dbus_error(self):
1092
class FailingBusStub(object):
1093
def get_object(self, busname, dbus_path):
1094
raise dbus.exceptions.DBusException("Test")
1096
with self.assertLogs(log, logging.CRITICAL):
1097
with self.assertRaises(SystemExit) as e:
1098
bus = get_mandos_dbus_object(bus=FailingBusStub())
1100
if isinstance(e.exception.code, int):
1101
self.assertNotEqual(0, e.exception.code)
1103
self.assertIsNotNone(e.exception.code)
1106
class Test_get_managed_objects(TestCaseWithAssertLogs):
1107
def test_calls_and_returns_GetManagedObjects(self):
1108
managed_objects = {"/clients/client": { "Name": "client"}}
1109
class ObjectManagerStub(object):
1110
def GetManagedObjects(self):
1111
return managed_objects
1112
retval = get_managed_objects(ObjectManagerStub())
1113
self.assertDictEqual(managed_objects, retval)
1115
def test_logs_and_exits_on_dbus_error(self):
1205
self.parse_args(args)
1208
class Test_dbus_exceptions(unittest.TestCase):
1210
def test_dbus_ConnectFailed_is_Error(self):
1211
with self.assertRaises(dbus.Error):
1212
raise dbus.ConnectFailed()
1215
class Test_dbus_MandosBus(unittest.TestCase):
1217
class MockMandosBus(dbus.MandosBus):
1219
self._name = "se.recompile.Mandos"
1220
self._server_path = "/"
1221
self._server_interface = "se.recompile.Mandos"
1222
self._client_interface = "se.recompile.Mandos.Client"
1224
self.call_method_return = Unique()
1226
def call_method(self, methodname, busname, objectpath,
1228
self.calls.append((methodname, busname, objectpath,
1230
return self.call_method_return
1233
self.bus = self.MockMandosBus()
1235
def test_set_client_property(self):
1236
self.bus.set_client_property("objectpath", "key", "value")
1237
expected_call = ("Set", self.bus._name, "objectpath",
1238
"org.freedesktop.DBus.Properties",
1239
(self.bus._client_interface, "key", "value"))
1240
self.assertIn(expected_call, self.bus.calls)
1242
def test_call_client_method(self):
1243
ret = self.bus.call_client_method("objectpath", "methodname")
1244
self.assertIs(self.bus.call_method_return, ret)
1245
expected_call = ("methodname", self.bus._name, "objectpath",
1246
self.bus._client_interface, ())
1247
self.assertIn(expected_call, self.bus.calls)
1249
def test_call_client_method_with_args(self):
1250
args = (Unique(), Unique())
1251
ret = self.bus.call_client_method("objectpath", "methodname",
1253
self.assertIs(self.bus.call_method_return, ret)
1254
expected_call = ("methodname", self.bus._name, "objectpath",
1255
self.bus._client_interface,
1257
self.assertIn(expected_call, self.bus.calls)
1259
def test_get_clients_and_properties(self):
1262
self.bus._client_interface: {
1266
"irrelevant_interface": {
1267
"key": "othervalue",
1271
"other_objectpath": {
1272
"other_irrelevant_interface": {
1278
expected_clients_and_properties = {
1284
self.bus.call_method_return = managed_objects
1285
ret = self.bus.get_clients_and_properties()
1286
self.assertDictEqual(expected_clients_and_properties, ret)
1287
expected_call = ("GetManagedObjects", self.bus._name,
1288
self.bus._server_path,
1289
"org.freedesktop.DBus.ObjectManager", ())
1290
self.assertIn(expected_call, self.bus.calls)
1292
def test_call_server_method(self):
1293
ret = self.bus.call_server_method("methodname")
1294
self.assertIs(self.bus.call_method_return, ret)
1295
expected_call = ("methodname", self.bus._name,
1296
self.bus._server_path,
1297
self.bus._server_interface, ())
1298
self.assertIn(expected_call, self.bus.calls)
1300
def test_call_server_method_with_args(self):
1301
args = (Unique(), Unique())
1302
ret = self.bus.call_server_method("methodname", *args)
1303
self.assertIs(self.bus.call_method_return, ret)
1304
expected_call = ("methodname", self.bus._name,
1305
self.bus._server_path,
1306
self.bus._server_interface,
1308
self.assertIn(expected_call, self.bus.calls)
1311
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1313
def MockDBusPython_func(self, func):
1314
class mock_dbus_python(object):
1315
"""mock dbus-python module"""
1316
class exceptions(object):
1317
"""Pseudo-namespace"""
1318
class DBusException(Exception):
1320
class SystemBus(object):
1322
def get_object(busname, objectpath):
1323
DBusObject = collections.namedtuple(
1324
"DBusObject", ("methodname", "Set"))
1325
def method(*args, **kwargs):
1326
self.assertEqual({"dbus_interface":
1330
def set_property(interface, key, value,
1331
dbus_interface=None):
1333
"org.freedesktop.DBus.Properties",
1335
self.assertEqual("Secret", key)
1336
return func(interface, key, value,
1337
dbus_interface=dbus_interface)
1338
return DBusObject(methodname=method,
1340
class Boolean(object):
1341
def __init__(self, value):
1342
self.value = bool(value)
1345
if sys.version_info.major == 2:
1346
__nonzero__ = __bool__
1347
class ObjectPath(str):
1349
class Dictionary(dict):
1351
class ByteArray(bytes):
1353
return mock_dbus_python
1355
def call_method(self, bus, methodname, busname, objectpath,
1357
with self.assertLogs(log, logging.DEBUG):
1358
return bus.call_method(methodname, busname, objectpath,
1361
def test_call_method_returns(self):
1362
expected_method_return = Unique()
1363
method_args = (Unique(), Unique())
1365
self.assertEqual(len(method_args), len(args))
1366
for marg, arg in zip(method_args, args):
1367
self.assertIs(marg, arg)
1368
return expected_method_return
1369
mock_dbus_python = self.MockDBusPython_func(func)
1370
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1371
ret = self.call_method(bus, "methodname", "busname",
1372
"objectpath", "interface",
1374
self.assertIs(ret, expected_method_return)
1376
def test_call_method_filters_bool_true(self):
1378
return method_return
1379
mock_dbus_python = self.MockDBusPython_func(func)
1380
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1381
method_return = mock_dbus_python.Boolean(True)
1382
ret = self.call_method(bus, "methodname", "busname",
1383
"objectpath", "interface")
1384
self.assertTrue(ret)
1385
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1387
def test_call_method_filters_bool_false(self):
1389
return method_return
1390
mock_dbus_python = self.MockDBusPython_func(func)
1391
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1392
method_return = mock_dbus_python.Boolean(False)
1393
ret = self.call_method(bus, "methodname", "busname",
1394
"objectpath", "interface")
1395
self.assertFalse(ret)
1396
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1398
def test_call_method_filters_objectpath(self):
1400
return method_return
1401
mock_dbus_python = self.MockDBusPython_func(func)
1402
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1403
method_return = mock_dbus_python.ObjectPath("objectpath")
1404
ret = self.call_method(bus, "methodname", "busname",
1405
"objectpath", "interface")
1406
self.assertEqual("objectpath", ret)
1407
self.assertIsNot("objectpath", ret)
1408
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1410
def test_call_method_filters_booleans_in_dict(self):
1412
return method_return
1413
mock_dbus_python = self.MockDBusPython_func(func)
1414
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1415
method_return = mock_dbus_python.Dictionary(
1416
{mock_dbus_python.Boolean(True):
1417
mock_dbus_python.Boolean(False),
1418
mock_dbus_python.Boolean(False):
1419
mock_dbus_python.Boolean(True)})
1420
ret = self.call_method(bus, "methodname", "busname",
1421
"objectpath", "interface")
1422
expected_method_return = {True: False,
1424
self.assertEqual(expected_method_return, ret)
1425
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1427
def test_call_method_filters_objectpaths_in_dict(self):
1429
return method_return
1430
mock_dbus_python = self.MockDBusPython_func(func)
1431
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1432
method_return = mock_dbus_python.Dictionary(
1433
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1434
mock_dbus_python.ObjectPath("objectpath_value_1"),
1435
mock_dbus_python.ObjectPath("objectpath_key_2"):
1436
mock_dbus_python.ObjectPath("objectpath_value_2")})
1437
ret = self.call_method(bus, "methodname", "busname",
1438
"objectpath", "interface")
1439
expected_method_return = {str(key): str(value)
1441
method_return.items()}
1442
self.assertEqual(expected_method_return, ret)
1443
self.assertIsInstance(ret, dict)
1444
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1446
def test_call_method_filters_dict_in_dict(self):
1448
return method_return
1449
mock_dbus_python = self.MockDBusPython_func(func)
1450
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1451
method_return = mock_dbus_python.Dictionary(
1452
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1453
"key12": "value12"}),
1454
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1455
"key22": "value22"})})
1456
ret = self.call_method(bus, "methodname", "busname",
1457
"objectpath", "interface")
1458
expected_method_return = {
1459
"key1": {"key11": "value11",
1460
"key12": "value12"},
1461
"key2": {"key21": "value21",
1462
"key22": "value22"},
1464
self.assertEqual(expected_method_return, ret)
1465
self.assertIsInstance(ret, dict)
1466
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1467
for key, value in ret.items():
1468
self.assertIsInstance(value, dict)
1469
self.assertEqual(expected_method_return[key], value)
1470
self.assertNotIsInstance(value,
1471
mock_dbus_python.Dictionary)
1473
def test_call_method_filters_dict_three_deep(self):
1475
return method_return
1476
mock_dbus_python = self.MockDBusPython_func(func)
1477
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1478
method_return = mock_dbus_python.Dictionary(
1480
mock_dbus_python.Dictionary(
1482
mock_dbus_python.Dictionary(
1484
mock_dbus_python.Boolean(True),
1488
ret = self.call_method(bus, "methodname", "busname",
1489
"objectpath", "interface")
1490
expected_method_return = {"key1": {"key2": {"key3": True}}}
1491
self.assertEqual(expected_method_return, ret)
1492
self.assertIsInstance(ret, dict)
1493
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1494
self.assertIsInstance(ret["key1"], dict)
1495
self.assertNotIsInstance(ret["key1"],
1496
mock_dbus_python.Dictionary)
1497
self.assertIsInstance(ret["key1"]["key2"], dict)
1498
self.assertNotIsInstance(ret["key1"]["key2"],
1499
mock_dbus_python.Dictionary)
1500
self.assertTrue(ret["key1"]["key2"]["key3"])
1501
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1502
mock_dbus_python.Boolean)
1504
def test_call_method_handles_exception(self):
1116
1505
dbus_logger = logging.getLogger("dbus.proxies")
1118
class ObjectManagerFailingStub(object):
1119
def GetManagedObjects(self):
1120
dbus_logger.error("Test")
1121
raise dbus.exceptions.DBusException("Test")
1508
dbus_logger.error("Test")
1509
raise mock_dbus_python.exceptions.DBusException()
1511
mock_dbus_python = self.MockDBusPython_func(func)
1512
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1123
1514
class CountingHandler(logging.Handler):
1130
1521
dbus_logger.addHandler(counting_handler)
1133
with self.assertLogs(log, logging.CRITICAL) as watcher:
1134
with self.assertRaises(SystemExit) as e:
1135
get_managed_objects(ObjectManagerFailingStub())
1524
with self.assertRaises(dbus.Error) as e:
1525
self.call_method(bus, "methodname", "busname",
1526
"objectpath", "interface")
1137
1528
dbus_logger.removeFilter(counting_handler)
1530
self.assertNotIsInstance(e, dbus.ConnectFailed)
1139
1532
# Make sure the dbus logger was suppressed
1140
1533
self.assertEqual(0, counting_handler.count)
1142
# Test that the dbus_logger still works
1143
with self.assertLogs(dbus_logger, logging.ERROR):
1144
dbus_logger.error("Test")
1146
if isinstance(e.exception.code, int):
1147
self.assertNotEqual(0, e.exception.code)
1149
self.assertIsNotNone(e.exception.code)
1535
def test_Set_Secret_sends_bytearray(self):
1537
def func(*args, **kwargs):
1538
ret[0] = (args, kwargs)
1539
mock_dbus_python = self.MockDBusPython_func(func)
1540
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1541
bus.set_client_property("objectpath", "Secret", "value")
1542
expected_call = (("se.recompile.Mandos.Client", "Secret",
1543
mock_dbus_python.ByteArray(b"value")),
1545
"org.freedesktop.DBus.Properties"})
1546
self.assertEqual(expected_call, ret[0])
1547
if sys.version_info.major == 2:
1548
self.assertIsInstance(ret[0][0][-1],
1549
mock_dbus_python.ByteArray)
1551
def test_get_object_converts_to_correct_exception(self):
1552
bus = dbus_python_adapter.SystemBus(
1553
self.fake_dbus_python_raises_exception_on_connect)
1554
with self.assertRaises(dbus.ConnectFailed):
1555
self.call_method(bus, "methodname", "busname",
1556
"objectpath", "interface")
1558
class fake_dbus_python_raises_exception_on_connect(object):
1559
"""fake dbus-python module"""
1560
class exceptions(object):
1561
"""Pseudo-namespace"""
1562
class DBusException(Exception):
1567
def get_object(busname, objectpath):
1568
raise cls.exceptions.DBusException()
1569
Bus = collections.namedtuple("Bus", ["get_object"])
1570
return Bus(get_object=get_object)
1573
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1574
class mock_dbus_python(object):
1575
"""mock dbus-python modules"""
1576
class SystemBus(object):
1578
def get_object(busname, objectpath):
1582
self.bus = dbus_python_adapter.CachingBus(
1583
self.mock_dbus_python)
1585
def test_returns_distinct_objectpaths(self):
1586
obj1 = self.bus.get_object("busname", "objectpath1")
1587
self.assertIsInstance(obj1, Unique)
1588
obj2 = self.bus.get_object("busname", "objectpath2")
1589
self.assertIsInstance(obj2, Unique)
1590
self.assertIsNot(obj1, obj2)
1592
def test_returns_distinct_busnames(self):
1593
obj1 = self.bus.get_object("busname1", "objectpath")
1594
self.assertIsInstance(obj1, Unique)
1595
obj2 = self.bus.get_object("busname2", "objectpath")
1596
self.assertIsInstance(obj2, Unique)
1597
self.assertIsNot(obj1, obj2)
1599
def test_returns_distinct_both(self):
1600
obj1 = self.bus.get_object("busname1", "objectpath")
1601
self.assertIsInstance(obj1, Unique)
1602
obj2 = self.bus.get_object("busname2", "objectpath")
1603
self.assertIsInstance(obj2, Unique)
1604
self.assertIsNot(obj1, obj2)
1606
def test_returns_same(self):
1607
obj1 = self.bus.get_object("busname", "objectpath")
1608
self.assertIsInstance(obj1, Unique)
1609
obj2 = self.bus.get_object("busname", "objectpath")
1610
self.assertIsInstance(obj2, Unique)
1611
self.assertIs(obj1, obj2)
1613
def test_returns_same_old(self):
1614
obj1 = self.bus.get_object("busname1", "objectpath1")
1615
self.assertIsInstance(obj1, Unique)
1616
obj2 = self.bus.get_object("busname2", "objectpath2")
1617
self.assertIsInstance(obj2, Unique)
1618
obj1b = self.bus.get_object("busname1", "objectpath1")
1619
self.assertIsInstance(obj1b, Unique)
1620
self.assertIsNot(obj1, obj2)
1621
self.assertIsNot(obj2, obj1b)
1622
self.assertIs(obj1, obj1b)
1625
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1627
def Stub_pydbus_func(self, func):
1628
class stub_pydbus(object):
1629
"""stub pydbus module"""
1630
class SystemBus(object):
1632
def get(busname, objectpath):
1633
DBusObject = collections.namedtuple(
1634
"DBusObject", ("methodname",))
1635
return {"interface":
1636
DBusObject(methodname=func)}
1639
def call_method(self, bus, methodname, busname, objectpath,
1641
with self.assertLogs(log, logging.DEBUG):
1642
return bus.call_method(methodname, busname, objectpath,
1645
def test_call_method_returns(self):
1646
expected_method_return = Unique()
1647
method_args = (Unique(), Unique())
1649
self.assertEqual(len(method_args), len(args))
1650
for marg, arg in zip(method_args, args):
1651
self.assertIs(marg, arg)
1652
return expected_method_return
1653
stub_pydbus = self.Stub_pydbus_func(func)
1654
bus = pydbus_adapter.SystemBus(stub_pydbus)
1655
ret = self.call_method(bus, "methodname", "busname",
1656
"objectpath", "interface",
1658
self.assertIs(ret, expected_method_return)
1660
def test_call_method_handles_exception(self):
1661
dbus_logger = logging.getLogger("dbus.proxies")
1664
raise gi.repository.GLib.Error()
1666
stub_pydbus = self.Stub_pydbus_func(func)
1667
bus = pydbus_adapter.SystemBus(stub_pydbus)
1669
with self.assertRaises(dbus.Error) as e:
1670
self.call_method(bus, "methodname", "busname",
1671
"objectpath", "interface")
1673
self.assertNotIsInstance(e, dbus.ConnectFailed)
1675
def test_get_converts_to_correct_exception(self):
1676
bus = pydbus_adapter.SystemBus(
1677
self.fake_pydbus_raises_exception_on_connect)
1678
with self.assertRaises(dbus.ConnectFailed):
1679
self.call_method(bus, "methodname", "busname",
1680
"objectpath", "interface")
1682
class fake_pydbus_raises_exception_on_connect(object):
1683
"""fake dbus-python module"""
1686
def get(busname, objectpath):
1687
raise gi.repository.GLib.Error()
1688
Bus = collections.namedtuple("Bus", ["get"])
1691
def test_set_property_uses_setattr(self):
1692
class Object(object):
1695
class pydbus_spy(object):
1696
class SystemBus(object):
1698
def get(busname, objectpath):
1699
return {"interface": obj}
1700
bus = pydbus_adapter.SystemBus(pydbus_spy)
1702
bus.set_property("busname", "objectpath", "interface", "key",
1704
self.assertIs(value, obj.key)
1706
def test_get_suppresses_xml_deprecation_warning(self):
1707
if sys.version_info.major >= 3:
1709
class stub_pydbus_get(object):
1710
class SystemBus(object):
1712
def get(busname, objectpath):
1713
warnings.warn_explicit(
1714
"deprecated", DeprecationWarning,
1715
"xml.etree.ElementTree", 0)
1716
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1717
with warnings.catch_warnings(record=True) as w:
1718
warnings.simplefilter("always")
1719
bus.get("busname", "objectpath")
1720
self.assertEqual(0, len(w))
1723
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1724
class stub_pydbus(object):
1725
"""stub pydbus module"""
1726
class SystemBus(object):
1728
def get(busname, objectpath):
1732
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1734
def test_returns_distinct_objectpaths(self):
1735
obj1 = self.bus.get("busname", "objectpath1")
1736
self.assertIsInstance(obj1, Unique)
1737
obj2 = self.bus.get("busname", "objectpath2")
1738
self.assertIsInstance(obj2, Unique)
1739
self.assertIsNot(obj1, obj2)
1741
def test_returns_distinct_busnames(self):
1742
obj1 = self.bus.get("busname1", "objectpath")
1743
self.assertIsInstance(obj1, Unique)
1744
obj2 = self.bus.get("busname2", "objectpath")
1745
self.assertIsInstance(obj2, Unique)
1746
self.assertIsNot(obj1, obj2)
1748
def test_returns_distinct_both(self):
1749
obj1 = self.bus.get("busname1", "objectpath")
1750
self.assertIsInstance(obj1, Unique)
1751
obj2 = self.bus.get("busname2", "objectpath")
1752
self.assertIsInstance(obj2, Unique)
1753
self.assertIsNot(obj1, obj2)
1755
def test_returns_same(self):
1756
obj1 = self.bus.get("busname", "objectpath")
1757
self.assertIsInstance(obj1, Unique)
1758
obj2 = self.bus.get("busname", "objectpath")
1759
self.assertIsInstance(obj2, Unique)
1760
self.assertIs(obj1, obj2)
1762
def test_returns_same_old(self):
1763
obj1 = self.bus.get("busname1", "objectpath1")
1764
self.assertIsInstance(obj1, Unique)
1765
obj2 = self.bus.get("busname2", "objectpath2")
1766
self.assertIsInstance(obj2, Unique)
1767
obj1b = self.bus.get("busname1", "objectpath1")
1768
self.assertIsInstance(obj1b, Unique)
1769
self.assertIsNot(obj1, obj2)
1770
self.assertIsNot(obj2, obj1b)
1771
self.assertIs(obj1, obj1b)
1152
1774
class Test_commands_from_options(unittest.TestCase):
1153
1776
def setUp(self):
1154
1777
self.parser = argparse.ArgumentParser()
1155
1778
add_command_line_options(self.parser)
2004
def test_manual_page_example_1(self):
2005
self.assert_command_from_args("",
2010
def test_manual_page_example_2(self):
2011
self.assert_command_from_args(
2012
"--verbose foo1.example.org foo2.example.org".split(),
2013
command.PrintTable, clients=["foo1.example.org",
2014
"foo2.example.org"],
2017
def test_manual_page_example_3(self):
2018
self.assert_command_from_args("--enable --all".split(),
2022
def test_manual_page_example_4(self):
2023
self.assert_commands_from_args(
2024
("--timeout=PT5M --interval=PT1M foo1.example.org"
2025
" foo2.example.org").split(),
2026
[command.SetTimeout, command.SetInterval],
2027
clients=["foo1.example.org", "foo2.example.org"])
2029
def test_manual_page_example_5(self):
2030
self.assert_command_from_args("--approve --all".split(),
1370
2035
class TestCommand(unittest.TestCase):
1371
2036
"""Abstract class for tests of command classes"""
2038
class FakeMandosBus(dbus.MandosBus):
2039
def __init__(self, testcase):
2040
self.client_properties = {
2042
"KeyID": ("92ed150794387c03ce684574b1139a65"
2043
"94a34f895daaaf09fd8ea90a27cddb12"),
2044
"Secret": b"secret",
2045
"Host": "foo.example.org",
2048
"LastCheckedOK": "2019-02-03T00:00:00",
2049
"Created": "2019-01-02T00:00:00",
2051
"Fingerprint": ("778827225BA7DE539C5A"
2052
"7CFA59CFF7CDBD9A5920"),
2053
"CheckerRunning": False,
2054
"LastEnabled": "2019-01-03T00:00:00",
2055
"ApprovalPending": False,
2056
"ApprovedByDefault": True,
2057
"LastApprovalRequest": "",
2059
"ApprovalDuration": 1000,
2060
"Checker": "fping -q -- %(host)s",
2061
"ExtendedTimeout": 900000,
2062
"Expires": "2019-02-04T00:00:00",
2063
"LastCheckerStatus": 0,
2065
self.other_client_properties = {
2067
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
2068
"6ab612cff5ad227247e46c2b020f441c"),
2069
"Secret": b"secretbar",
2070
"Host": "192.0.2.3",
2073
"LastCheckedOK": "2019-02-04T00:00:00",
2074
"Created": "2019-01-03T00:00:00",
2076
"Fingerprint": ("3E393AEAEFB84C7E89E2"
2077
"F547B3A107558FCA3A27"),
2078
"CheckerRunning": True,
2079
"LastEnabled": "2019-01-04T00:00:00",
2080
"ApprovalPending": False,
2081
"ApprovedByDefault": False,
2082
"LastApprovalRequest": "2019-01-03T00:00:00",
2083
"ApprovalDelay": 30000,
2084
"ApprovalDuration": 93785000,
2086
"ExtendedTimeout": 900000,
2087
"Expires": "2019-02-05T00:00:00",
2088
"LastCheckerStatus": -2,
2090
self.clients = collections.OrderedDict(
2092
("client_objectpath", self.client_properties),
2093
("other_client_objectpath",
2094
self.other_client_properties),
2096
self.one_client = {"client_objectpath":
2097
self.client_properties}
2098
self.testcase = testcase
2101
def call_method(self, methodname, busname, objectpath,
2103
self.testcase.assertEqual("se.recompile.Mandos", busname)
2104
self.calls.append((methodname, busname, objectpath,
2106
if interface == "org.freedesktop.DBus.Properties":
2107
if methodname == "Set":
2108
self.testcase.assertEqual(3, len(args))
2109
interface, key, value = args
2110
self.testcase.assertEqual(
2111
"se.recompile.Mandos.Client", interface)
2112
self.clients[objectpath][key] = value
2114
elif interface == "se.recompile.Mandos":
2115
self.testcase.assertEqual("RemoveClient", methodname)
2116
self.testcase.assertEqual(1, len(args))
2117
clientpath = args[0]
2118
del self.clients[clientpath]
2120
elif interface == "se.recompile.Mandos.Client":
2121
if methodname == "Approve":
2122
self.testcase.assertEqual(1, len(args))
1373
2126
def setUp(self):
1375
class MockClient(object):
1376
def __init__(self, name, **attributes):
1377
self.__dbus_object_path__ = "/clients/{}".format(name)
1378
self.attributes = attributes
1379
self.attributes["Name"] = name
1381
def Set(self, interface, propname, value, dbus_interface):
1382
testcase.assertEqual(client_dbus_interface, interface)
1383
testcase.assertEqual(dbus.PROPERTIES_IFACE,
1385
self.attributes[propname] = value
1386
def Approve(self, approve, dbus_interface):
1387
testcase.assertEqual(client_dbus_interface,
1389
self.calls.append(("Approve", (approve,
1391
self.client = MockClient(
1393
KeyID=("92ed150794387c03ce684574b1139a65"
1394
"94a34f895daaaf09fd8ea90a27cddb12"),
1396
Host="foo.example.org",
1397
Enabled=dbus.Boolean(True),
1399
LastCheckedOK="2019-02-03T00:00:00",
1400
Created="2019-01-02T00:00:00",
1402
Fingerprint=("778827225BA7DE539C5A"
1403
"7CFA59CFF7CDBD9A5920"),
1404
CheckerRunning=dbus.Boolean(False),
1405
LastEnabled="2019-01-03T00:00:00",
1406
ApprovalPending=dbus.Boolean(False),
1407
ApprovedByDefault=dbus.Boolean(True),
1408
LastApprovalRequest="",
1410
ApprovalDuration=1000,
1411
Checker="fping -q -- %(host)s",
1412
ExtendedTimeout=900000,
1413
Expires="2019-02-04T00:00:00",
1414
LastCheckerStatus=0)
1415
self.other_client = MockClient(
1417
KeyID=("0558568eedd67d622f5c83b35a115f79"
1418
"6ab612cff5ad227247e46c2b020f441c"),
1419
Secret=b"secretbar",
1421
Enabled=dbus.Boolean(True),
1423
LastCheckedOK="2019-02-04T00:00:00",
1424
Created="2019-01-03T00:00:00",
1426
Fingerprint=("3E393AEAEFB84C7E89E2"
1427
"F547B3A107558FCA3A27"),
1428
CheckerRunning=dbus.Boolean(True),
1429
LastEnabled="2019-01-04T00:00:00",
1430
ApprovalPending=dbus.Boolean(False),
1431
ApprovedByDefault=dbus.Boolean(False),
1432
LastApprovalRequest="2019-01-03T00:00:00",
1433
ApprovalDelay=30000,
1434
ApprovalDuration=93785000,
1436
ExtendedTimeout=900000,
1437
Expires="2019-02-05T00:00:00",
1438
LastCheckerStatus=-2)
1439
self.clients = collections.OrderedDict(
1441
(self.client.__dbus_object_path__,
1442
self.client.attributes),
1443
(self.other_client.__dbus_object_path__,
1444
self.other_client.attributes),
1446
self.one_client = {self.client.__dbus_object_path__:
1447
self.client.attributes}
1451
class MockBus(object):
1453
def get_object(client_bus_name, path):
1454
self.assertEqual(dbus_busname, client_bus_name)
1455
# Note: "self" here is the TestCmd instance, not the
1456
# MockBus instance, since this is a static method!
1457
if path == self.client.__dbus_object_path__:
1459
elif path == self.other_client.__dbus_object_path__:
1460
return self.other_client
2127
self.bus = self.FakeMandosBus(self)
1464
2130
class TestBaseCommands(TestCommand):
1466
2132
def test_IsEnabled_exits_successfully(self):
1467
2133
with self.assertRaises(SystemExit) as e:
1468
command.IsEnabled().run(self.one_client)
2134
command.IsEnabled().run(self.bus.one_client)
1469
2135
if e.exception.code is not None:
1470
2136
self.assertEqual(0, e.exception.code)
1472
2138
self.assertIsNone(e.exception.code)
1474
2140
def test_IsEnabled_exits_with_failure(self):
1475
self.client.attributes["Enabled"] = dbus.Boolean(False)
2141
self.bus.client_properties["Enabled"] = False
1476
2142
with self.assertRaises(SystemExit) as e:
1477
command.IsEnabled().run(self.one_client)
2143
command.IsEnabled().run(self.bus.one_client)
1478
2144
if isinstance(e.exception.code, int):
1479
2145
self.assertNotEqual(0, e.exception.code)
1481
2147
self.assertIsNotNone(e.exception.code)
1483
2149
def test_Approve(self):
1484
command.Approve().run(self.clients, self.bus)
1485
for clientpath in self.clients:
1486
client = self.bus.get_object(dbus_busname, clientpath)
1487
self.assertIn(("Approve", (True, client_dbus_interface)),
2150
busname = "se.recompile.Mandos"
2151
client_interface = "se.recompile.Mandos.Client"
2152
command.Approve().run(self.bus.clients, self.bus)
2153
for clientpath in self.bus.clients:
2154
self.assertIn(("Approve", busname, clientpath,
2155
client_interface, (True,)), self.bus.calls)
1490
2157
def test_Deny(self):
1491
command.Deny().run(self.clients, self.bus)
1492
for clientpath in self.clients:
1493
client = self.bus.get_object(dbus_busname, clientpath)
1494
self.assertIn(("Approve", (False, client_dbus_interface)),
2158
busname = "se.recompile.Mandos"
2159
client_interface = "se.recompile.Mandos.Client"
2160
command.Deny().run(self.bus.clients, self.bus)
2161
for clientpath in self.bus.clients:
2162
self.assertIn(("Approve", busname, clientpath,
2163
client_interface, (False,)),
1497
2166
def test_Remove(self):
1498
class MandosSpy(object):
1501
def RemoveClient(self, dbus_path):
1502
self.calls.append(("RemoveClient", (dbus_path,)))
1503
mandos = MandosSpy()
1504
command.Remove().run(self.clients, self.bus, mandos)
1505
for clientpath in self.clients:
1506
self.assertIn(("RemoveClient", (clientpath,)),
2167
command.Remove().run(self.bus.clients, self.bus)
2168
for clientpath in self.bus.clients:
2169
self.assertIn(("RemoveClient", dbus_busname,
2170
dbus_server_path, dbus_server_interface,
2171
(clientpath,)), self.bus.calls)
1509
2173
expected_json = {
1700
2364
class TestPropertySetterCmd(TestCommand):
1701
2365
"""Abstract class for tests of command.PropertySetter classes"""
1702
2367
def runTest(self):
1703
2368
if not hasattr(self, "command"):
1705
values_to_get = getattr(self, "values_to_get",
1707
for value_to_set, value_to_get in zip(self.values_to_set,
1709
for clientpath in self.clients:
1710
client = self.bus.get_object(dbus_busname, clientpath)
1711
old_value = client.attributes[self.propname]
1712
client.attributes[self.propname] = self.Unique()
1713
self.run_command(value_to_set, self.clients)
1714
for clientpath in self.clients:
1715
client = self.bus.get_object(dbus_busname, clientpath)
1716
value = client.attributes[self.propname]
1717
self.assertNotIsInstance(value, self.Unique)
2369
return # Abstract TestCase class
2371
if hasattr(self, "values_to_set"):
2372
cmd_args = [(value,) for value in self.values_to_set]
2373
values_to_get = getattr(self, "values_to_get",
2376
cmd_args = [() for x in range(len(self.values_to_get))]
2377
values_to_get = self.values_to_get
2378
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2379
for clientpath in self.bus.clients:
2380
self.bus.clients[clientpath][self.propname] = (
2382
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2383
for clientpath in self.bus.clients:
2384
value = (self.bus.clients[clientpath]
2386
self.assertNotIsInstance(value, Unique)
1718
2387
self.assertEqual(value_to_get, value)
1720
class Unique(object):
1721
"""Class for objects which exist only to be unique objects,
1722
since unittest.mock.sentinel only exists in Python 3.3"""
1724
def run_command(self, value, clients):
1725
self.command().run(clients, self.bus)
1728
2390
class TestEnableCmd(TestPropertySetterCmd):
1729
2391
command = command.Enable
1730
2392
propname = "Enabled"
1731
values_to_set = [dbus.Boolean(True)]
2393
values_to_get = [True]
1734
2396
class TestDisableCmd(TestPropertySetterCmd):
1735
2397
command = command.Disable
1736
2398
propname = "Enabled"
1737
values_to_set = [dbus.Boolean(False)]
2399
values_to_get = [False]
1740
2402
class TestBumpTimeoutCmd(TestPropertySetterCmd):
1741
2403
command = command.BumpTimeout
1742
2404
propname = "LastCheckedOK"
1743
values_to_set = [""]
2405
values_to_get = [""]
1746
2408
class TestStartCheckerCmd(TestPropertySetterCmd):
1747
2409
command = command.StartChecker
1748
2410
propname = "CheckerRunning"
1749
values_to_set = [dbus.Boolean(True)]
2411
values_to_get = [True]
1752
2414
class TestStopCheckerCmd(TestPropertySetterCmd):
1753
2415
command = command.StopChecker
1754
2416
propname = "CheckerRunning"
1755
values_to_set = [dbus.Boolean(False)]
2417
values_to_get = [False]
1758
2420
class TestApproveByDefaultCmd(TestPropertySetterCmd):
1759
2421
command = command.ApproveByDefault
1760
2422
propname = "ApprovedByDefault"
1761
values_to_set = [dbus.Boolean(True)]
2423
values_to_get = [True]
1764
2426
class TestDenyByDefaultCmd(TestPropertySetterCmd):
1765
2427
command = command.DenyByDefault
1766
2428
propname = "ApprovedByDefault"
1767
values_to_set = [dbus.Boolean(False)]
1770
class TestPropertySetterValueCmd(TestPropertySetterCmd):
1771
"""Abstract class for tests of PropertySetterValueCmd classes"""
1774
if type(self) is TestPropertySetterValueCmd:
1776
return super(TestPropertySetterValueCmd, self).runTest()
1778
def run_command(self, value, clients):
1779
self.command(value).run(clients, self.bus)
1782
class TestSetCheckerCmd(TestPropertySetterValueCmd):
2429
values_to_get = [False]
2432
class TestSetCheckerCmd(TestPropertySetterCmd):
1783
2433
command = command.SetChecker
1784
2434
propname = "Checker"
1785
2435
values_to_set = ["", ":", "fping -q -- %s"]
1788
class TestSetHostCmd(TestPropertySetterValueCmd):
2438
class TestSetHostCmd(TestPropertySetterCmd):
1789
2439
command = command.SetHost
1790
2440
propname = "Host"
1791
2441
values_to_set = ["192.0.2.3", "client.example.org"]
1794
class TestSetSecretCmd(TestPropertySetterValueCmd):
2444
class TestSetSecretCmd(TestPropertySetterCmd):
1795
2445
command = command.SetSecret
1796
2446
propname = "Secret"
1797
2447
values_to_set = [io.BytesIO(b""),