385
def check_option_syntax(parser, options):
386
"""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,
409
if has_actions(options) and not (options.client or options.all):
410
parser.error("Options require clients names or --all.")
411
if options.verbose and has_actions(options):
412
parser.error("--verbose can only be used alone.")
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
415
parser.error("--dump-json can only be used alone.")
416
if options.all and not has_actions(options):
417
parser.error("--all requires an action.")
418
if options.is_enabled and len(options.client) > 1:
419
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)
273
## Classes for commands.
275
# Abstract classes first
276
class Command(object):
277
"""Abstract class for commands"""
278
def run(self, mandos, clients):
279
"""Normal commands should implement run_on_one_client(), but
280
commands which want to operate on all clients at the same time
281
can override this run() method instead."""
283
for client, properties in clients.items():
284
self.run_on_one_client(client, properties)
286
class PrintCmd(Command):
287
"""Abstract class for commands printing client details"""
288
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
289
"Created", "Interval", "Host", "KeyID",
290
"Fingerprint", "CheckerRunning", "LastEnabled",
291
"ApprovalPending", "ApprovedByDefault",
292
"LastApprovalRequest", "ApprovalDelay",
293
"ApprovalDuration", "Checker", "ExtendedTimeout",
294
"Expires", "LastCheckerStatus")
295
def run(self, mandos, clients):
296
print(self.output(clients))
298
class PropertyCmd(Command):
299
"""Abstract class for Actions for setting one client property"""
300
def run_on_one_client(self, client, properties):
301
"""Set the Client's D-Bus property"""
302
client.Set(client_interface, self.property, self.value_to_set,
303
dbus_interface=dbus.PROPERTIES_IFACE)
305
class ValueArgumentMixIn(object):
306
"""Mixin class for commands taking a value as argument"""
307
def __init__(self, value):
308
self.value_to_set = value
310
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
311
"""Mixin class for commands taking a value argument as
314
def value_to_set(self):
317
def value_to_set(self, value):
318
"""When setting, convert value to a datetime.timedelta"""
319
self._vts = string_to_delta(value).total_seconds() * 1000
321
# Actual (non-abstract) command classes
323
class PrintTableCmd(PrintCmd):
324
def __init__(self, verbose=False):
325
self.verbose = verbose
327
def output(self, clients):
328
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
329
keywords = default_keywords
331
keywords = self.all_keywords
332
return str(self.TableOfClients(clients.values(), keywords))
334
class TableOfClients(object):
337
"Enabled": "Enabled",
338
"Timeout": "Timeout",
339
"LastCheckedOK": "Last Successful Check",
340
"LastApprovalRequest": "Last Approval Request",
341
"Created": "Created",
342
"Interval": "Interval",
344
"Fingerprint": "Fingerprint",
346
"CheckerRunning": "Check Is Running",
347
"LastEnabled": "Last Enabled",
348
"ApprovalPending": "Approval Is Pending",
349
"ApprovedByDefault": "Approved By Default",
350
"ApprovalDelay": "Approval Delay",
351
"ApprovalDuration": "Approval Duration",
352
"Checker": "Checker",
353
"ExtendedTimeout": "Extended Timeout",
354
"Expires": "Expires",
355
"LastCheckerStatus": "Last Checker Status",
358
def __init__(self, clients, keywords, tableheaders=None):
359
self.clients = clients
360
self.keywords = keywords
361
if tableheaders is not None:
362
self.tableheaders = tableheaders
365
return "\n".join(self.rows())
367
if sys.version_info.major == 2:
368
__unicode__ = __str__
370
return str(self).encode(locale.getpreferredencoding())
373
format_string = self.row_formatting_string()
374
rows = [self.header_line(format_string)]
375
rows.extend(self.client_line(client, format_string)
376
for client in self.clients)
379
def row_formatting_string(self):
380
"Format string used to format table rows"
381
return " ".join("{{{key}:{width}}}".format(
382
width=max(len(self.tableheaders[key]),
383
*(len(self.string_from_client(client, key))
384
for client in self.clients)),
386
for key in self.keywords)
388
def string_from_client(self, client, key):
389
return self.valuetostring(client[key], key)
392
def valuetostring(value, keyword):
393
if isinstance(value, dbus.Boolean):
394
return "Yes" if value else "No"
395
if keyword in ("Timeout", "Interval", "ApprovalDelay",
396
"ApprovalDuration", "ExtendedTimeout"):
397
return milliseconds_to_string(value)
400
def header_line(self, format_string):
401
return format_string.format(**self.tableheaders)
403
def client_line(self, client, format_string):
404
return format_string.format(
405
**{key: self.string_from_client(client, key)
406
for key in self.keywords})
410
class DumpJSONCmd(PrintCmd):
411
def output(self, clients):
412
data = {client["Name"]:
413
{key: self.dbus_boolean_to_bool(client[key])
414
for key in self.all_keywords}
415
for client in clients.values()}
416
return json.dumps(data, indent=4, separators=(',', ': '))
418
def dbus_boolean_to_bool(value):
419
if isinstance(value, dbus.Boolean):
423
class IsEnabledCmd(Command):
424
def run_on_one_client(self, client, properties):
425
if self.is_enabled(client, properties):
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)
428
def is_enabled(self, client, properties):
429
return bool(properties["Enabled"])
431
class RemoveCmd(Command):
432
def run_on_one_client(self, client, properties):
433
self.mandos.RemoveClient(client.__dbus_object_path__)
435
class ApproveCmd(Command):
436
def run_on_one_client(self, client, properties):
437
client.Approve(dbus.Boolean(True),
438
dbus_interface=client_interface)
440
class DenyCmd(Command):
441
def run_on_one_client(self, client, properties):
442
client.Approve(dbus.Boolean(False),
443
dbus_interface=client_interface)
445
class EnableCmd(PropertyCmd):
447
value_to_set = dbus.Boolean(True)
449
class DisableCmd(PropertyCmd):
451
value_to_set = dbus.Boolean(False)
453
class BumpTimeoutCmd(PropertyCmd):
454
property = "LastCheckedOK"
457
class StartCheckerCmd(PropertyCmd):
458
property = "CheckerRunning"
459
value_to_set = dbus.Boolean(True)
461
class StopCheckerCmd(PropertyCmd):
462
property = "CheckerRunning"
463
value_to_set = dbus.Boolean(False)
465
class ApproveByDefaultCmd(PropertyCmd):
466
property = "ApprovedByDefault"
467
value_to_set = dbus.Boolean(True)
469
class DenyByDefaultCmd(PropertyCmd):
470
property = "ApprovedByDefault"
471
value_to_set = dbus.Boolean(False)
473
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
476
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
479
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
481
def value_to_set(self):
484
def value_to_set(self, value):
485
"""When setting, read data from supplied file object"""
486
self._vts = value.read()
490
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
493
class SetExtendedTimeoutCmd(PropertyCmd,
494
MillisecondsValueArgumentMixIn):
495
property = "ExtendedTimeout"
497
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
498
property = "Interval"
500
class SetApprovalDelayCmd(PropertyCmd,
501
MillisecondsValueArgumentMixIn):
502
property = "ApprovalDelay"
504
class SetApprovalDurationCmd(PropertyCmd,
505
MillisecondsValueArgumentMixIn):
506
property = "ApprovalDuration"
508
def has_actions(options):
509
return any((options.enable,
511
options.bump_timeout,
512
options.start_checker,
513
options.stop_checker,
516
options.checker is not None,
517
options.timeout is not None,
518
options.extended_timeout is not None,
519
options.interval is not None,
520
options.approved_by_default is not None,
521
options.approval_delay is not None,
522
options.approval_duration is not None,
523
options.host is not None,
524
options.secret is not None,
528
def add_command_line_options(parser):
529
parser.add_argument("--version", action="version",
530
version="%(prog)s {}".format(version),
531
help="show version number and exit")
532
parser.add_argument("-a", "--all", action="store_true",
533
help="Select all clients")
534
parser.add_argument("-v", "--verbose", action="store_true",
535
help="Print all fields")
536
parser.add_argument("-j", "--dump-json", action="store_true",
537
help="Dump client data in JSON format")
538
enable_disable = parser.add_mutually_exclusive_group()
539
enable_disable.add_argument("-e", "--enable", action="store_true",
540
help="Enable client")
541
enable_disable.add_argument("-d", "--disable",
543
help="disable client")
544
parser.add_argument("-b", "--bump-timeout", action="store_true",
545
help="Bump timeout for client")
546
start_stop_checker = parser.add_mutually_exclusive_group()
547
start_stop_checker.add_argument("--start-checker",
549
help="Start checker for client")
550
start_stop_checker.add_argument("--stop-checker",
552
help="Stop checker for client")
553
parser.add_argument("-V", "--is-enabled", action="store_true",
554
help="Check if client is enabled")
555
parser.add_argument("-r", "--remove", action="store_true",
556
help="Remove client")
557
parser.add_argument("-c", "--checker",
558
help="Set checker command for client")
559
parser.add_argument("-t", "--timeout",
560
help="Set timeout for client")
561
parser.add_argument("--extended-timeout",
562
help="Set extended timeout for client")
563
parser.add_argument("-i", "--interval",
564
help="Set checker interval for client")
565
approve_deny_default = parser.add_mutually_exclusive_group()
566
approve_deny_default.add_argument(
567
"--approve-by-default", action="store_true",
568
default=None, dest="approved_by_default",
569
help="Set client to be approved by default")
570
approve_deny_default.add_argument(
571
"--deny-by-default", action="store_false",
572
dest="approved_by_default",
573
help="Set client to be denied by default")
574
parser.add_argument("--approval-delay",
575
help="Set delay before client approve/deny")
576
parser.add_argument("--approval-duration",
577
help="Set duration of one client approval")
578
parser.add_argument("-H", "--host", help="Set host for client")
579
parser.add_argument("-s", "--secret",
580
type=argparse.FileType(mode="rb"),
581
help="Set password blob (file) for client")
582
approve_deny = parser.add_mutually_exclusive_group()
583
approve_deny.add_argument(
584
"-A", "--approve", action="store_true",
585
help="Approve any current client request")
586
approve_deny.add_argument("-D", "--deny", action="store_true",
587
help="Deny any current client request")
588
parser.add_argument("--check", action="store_true",
589
help="Run self-test")
590
parser.add_argument("client", nargs="*", help="Client name")
474
593
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
597
if options.dump_json:
491
commands.append(command.DumpJSON())
598
commands.append(DumpJSONCmd())
493
600
if options.enable:
494
commands.append(command.Enable())
601
commands.append(EnableCmd())
496
603
if options.disable:
497
commands.append(command.Disable())
604
commands.append(DisableCmd())
499
606
if options.bump_timeout:
500
commands.append(command.BumpTimeout())
607
commands.append(BumpTimeoutCmd())
502
609
if options.start_checker:
503
commands.append(command.StartChecker())
610
commands.append(StartCheckerCmd())
505
612
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())
512
commands.append(command.DenyByDefault())
613
commands.append(StopCheckerCmd())
615
if options.is_enabled:
616
commands.append(IsEnabledCmd())
619
commands.append(RemoveCmd())
514
621
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))
622
commands.append(SetCheckerCmd(options.checker))
523
624
if options.timeout is not None:
524
commands.append(command.SetTimeout(options.timeout))
625
commands.append(SetTimeoutCmd(options.timeout))
526
627
if options.extended_timeout:
528
command.SetExtendedTimeout(options.extended_timeout))
629
SetExtendedTimeoutCmd(options.extended_timeout))
530
631
if options.interval is not None:
531
commands.append(command.SetInterval(options.interval))
632
commands.append(SetIntervalCmd(options.interval))
634
if options.approved_by_default is not None:
635
if options.approved_by_default:
636
commands.append(ApproveByDefaultCmd())
638
commands.append(DenyByDefaultCmd())
533
640
if options.approval_delay is not None:
535
command.SetApprovalDelay(options.approval_delay))
641
commands.append(SetApprovalDelayCmd(options.approval_delay))
537
643
if options.approval_duration is not None:
539
command.SetApprovalDuration(options.approval_duration))
645
SetApprovalDurationCmd(options.approval_duration))
647
if options.host is not None:
648
commands.append(SetHostCmd(options.host))
650
if options.secret is not None:
651
commands.append(SetSecretCmd(options.secret))
654
commands.append(ApproveCmd())
657
commands.append(DenyCmd())
541
659
# If no command option has been given, show table of clients,
542
660
# optionally verbosely
544
commands.append(command.PrintTable(verbose=options.verbose))
662
commands.append(PrintTableCmd(verbose=options.verbose))
549
class command(object):
550
"""A namespace for command classes"""
553
"""Abstract base class for commands"""
554
def run(self, clients, bus=None, mandos=None):
555
"""Normal commands should implement run_on_one_client(),
556
but commands which want to operate on all clients at the same time can
557
override this run() method instead.
560
for clientpath, properties in clients.items():
561
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
562
dbus_busname, str(clientpath))
563
client = bus.get_object(dbus_busname, clientpath)
564
self.run_on_one_client(client, properties)
567
class IsEnabled(Base):
568
def run(self, clients, bus=None, mandos=None):
569
client, properties = next(iter(clients.items()))
570
if self.is_enabled(client, properties):
573
def is_enabled(self, client, properties):
574
return properties["Enabled"]
578
def run_on_one_client(self, client, properties):
579
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
580
client.__dbus_object_path__,
581
client_dbus_interface)
582
client.Approve(dbus.Boolean(True),
583
dbus_interface=client_dbus_interface)
587
def run_on_one_client(self, client, properties):
588
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
589
client.__dbus_object_path__,
590
client_dbus_interface)
591
client.Approve(dbus.Boolean(False),
592
dbus_interface=client_dbus_interface)
596
def run_on_one_client(self, client, properties):
597
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
598
dbus_busname, server_dbus_path,
599
server_dbus_interface,
600
str(client.__dbus_object_path__))
601
self.mandos.RemoveClient(client.__dbus_object_path__)
605
"""Abstract class for commands outputting client details"""
606
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
607
"Created", "Interval", "Host", "KeyID",
608
"Fingerprint", "CheckerRunning",
609
"LastEnabled", "ApprovalPending",
610
"ApprovedByDefault", "LastApprovalRequest",
611
"ApprovalDelay", "ApprovalDuration",
612
"Checker", "ExtendedTimeout", "Expires",
616
class DumpJSON(Output):
617
def run(self, clients, bus=None, mandos=None):
618
data = {client["Name"]:
619
{key: self.dbus_boolean_to_bool(client[key])
620
for key in self.all_keywords}
621
for client in clients.values()}
622
print(json.dumps(data, indent=4, separators=(',', ': ')))
625
def dbus_boolean_to_bool(value):
626
if isinstance(value, dbus.Boolean):
631
class PrintTable(Output):
632
def __init__(self, verbose=False):
633
self.verbose = verbose
635
def run(self, clients, bus=None, mandos=None):
636
default_keywords = ("Name", "Enabled", "Timeout",
638
keywords = default_keywords
640
keywords = self.all_keywords
641
print(self.TableOfClients(clients.values(), keywords))
643
class TableOfClients(object):
646
"Enabled": "Enabled",
647
"Timeout": "Timeout",
648
"LastCheckedOK": "Last Successful Check",
649
"LastApprovalRequest": "Last Approval Request",
650
"Created": "Created",
651
"Interval": "Interval",
653
"Fingerprint": "Fingerprint",
655
"CheckerRunning": "Check Is Running",
656
"LastEnabled": "Last Enabled",
657
"ApprovalPending": "Approval Is Pending",
658
"ApprovedByDefault": "Approved By Default",
659
"ApprovalDelay": "Approval Delay",
660
"ApprovalDuration": "Approval Duration",
661
"Checker": "Checker",
662
"ExtendedTimeout": "Extended Timeout",
663
"Expires": "Expires",
664
"LastCheckerStatus": "Last Checker Status",
667
def __init__(self, clients, keywords):
668
self.clients = clients
669
self.keywords = keywords
672
return "\n".join(self.rows())
674
if sys.version_info.major == 2:
675
__unicode__ = __str__
677
return str(self).encode(
678
locale.getpreferredencoding())
681
format_string = self.row_formatting_string()
682
rows = [self.header_line(format_string)]
683
rows.extend(self.client_line(client, format_string)
684
for client in self.clients)
687
def row_formatting_string(self):
688
"Format string used to format table rows"
689
return " ".join("{{{key}:{width}}}".format(
690
width=max(len(self.tableheaders[key]),
691
*(len(self.string_from_client(client,
693
for client in self.clients)),
695
for key in self.keywords)
697
def string_from_client(self, client, key):
698
return self.valuetostring(client[key], key)
701
def valuetostring(cls, value, keyword):
702
if isinstance(value, dbus.Boolean):
703
return "Yes" if value else "No"
704
if keyword in ("Timeout", "Interval", "ApprovalDelay",
705
"ApprovalDuration", "ExtendedTimeout"):
706
return cls.milliseconds_to_string(value)
709
def header_line(self, format_string):
710
return format_string.format(**self.tableheaders)
712
def client_line(self, client, format_string):
713
return format_string.format(
714
**{key: self.string_from_client(client, key)
715
for key in self.keywords})
718
def milliseconds_to_string(ms):
719
td = datetime.timedelta(0, 0, 0, ms)
720
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
721
.format(days="{}T".format(td.days)
723
hours=td.seconds // 3600,
724
minutes=(td.seconds % 3600) // 60,
725
seconds=td.seconds % 60))
728
class PropertySetter(Base):
729
"Abstract class for Actions for setting one client property"
731
def run_on_one_client(self, client, properties):
732
"""Set the Client's D-Bus property"""
733
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
734
client.__dbus_object_path__,
735
dbus.PROPERTIES_IFACE, client_dbus_interface,
736
self.propname, self.value_to_set
737
if not isinstance(self.value_to_set,
739
else bool(self.value_to_set))
740
client.Set(client_dbus_interface, self.propname,
742
dbus_interface=dbus.PROPERTIES_IFACE)
746
raise NotImplementedError()
749
class Enable(PropertySetter):
751
value_to_set = dbus.Boolean(True)
754
class Disable(PropertySetter):
756
value_to_set = dbus.Boolean(False)
759
class BumpTimeout(PropertySetter):
760
propname = "LastCheckedOK"
764
class StartChecker(PropertySetter):
765
propname = "CheckerRunning"
766
value_to_set = dbus.Boolean(True)
769
class StopChecker(PropertySetter):
770
propname = "CheckerRunning"
771
value_to_set = dbus.Boolean(False)
774
class ApproveByDefault(PropertySetter):
775
propname = "ApprovedByDefault"
776
value_to_set = dbus.Boolean(True)
779
class DenyByDefault(PropertySetter):
780
propname = "ApprovedByDefault"
781
value_to_set = dbus.Boolean(False)
784
class PropertySetterValue(PropertySetter):
785
"""Abstract class for PropertySetter recieving a value as
786
constructor argument instead of a class attribute."""
787
def __init__(self, value):
788
self.value_to_set = value
791
class SetChecker(PropertySetterValue):
795
class SetHost(PropertySetterValue):
799
class SetSecret(PropertySetterValue):
803
def value_to_set(self):
807
def value_to_set(self, value):
808
"""When setting, read data from supplied file object"""
809
self._vts = value.read()
813
class PropertySetterValueMilliseconds(PropertySetterValue):
814
"""Abstract class for PropertySetterValue taking a value
815
argument as a datetime.timedelta() but should store it as
819
def value_to_set(self):
823
def value_to_set(self, value):
824
"When setting, convert value from a datetime.timedelta"
825
self._vts = int(round(value.total_seconds() * 1000))
828
class SetTimeout(PropertySetterValueMilliseconds):
832
class SetExtendedTimeout(PropertySetterValueMilliseconds):
833
propname = "ExtendedTimeout"
836
class SetInterval(PropertySetterValueMilliseconds):
837
propname = "Interval"
840
class SetApprovalDelay(PropertySetterValueMilliseconds):
841
propname = "ApprovalDelay"
844
class SetApprovalDuration(PropertySetterValueMilliseconds):
845
propname = "ApprovalDuration"
668
parser = argparse.ArgumentParser()
670
add_command_line_options(parser)
672
options = parser.parse_args()
674
if has_actions(options) and not (options.client or options.all):
675
parser.error("Options require clients names or --all.")
676
if options.verbose and has_actions(options):
677
parser.error("--verbose can only be used alone.")
678
if options.dump_json and (options.verbose
679
or has_actions(options)):
680
parser.error("--dump-json can only be used alone.")
681
if options.all and not has_actions(options):
682
parser.error("--all requires an action.")
683
if options.is_enabled and len(options.client) > 1:
684
parser.error("--is-enabled requires exactly one client")
686
clientnames = options.client
689
bus = dbus.SystemBus()
690
mandos_dbus_objc = bus.get_object(busname, server_path)
691
except dbus.exceptions.DBusException:
692
log.critical("Could not connect to Mandos server")
695
mandos_serv = dbus.Interface(mandos_dbus_objc,
696
dbus_interface=server_interface)
697
mandos_serv_object_manager = dbus.Interface(
698
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
700
# Filter out log message from dbus module
701
dbus_logger = logging.getLogger("dbus.proxies")
702
class NullFilter(logging.Filter):
703
def filter(self, record):
705
dbus_filter = NullFilter()
707
dbus_logger.addFilter(dbus_filter)
708
mandos_clients = {path: ifs_and_props[client_interface]
709
for path, ifs_and_props in
710
mandos_serv_object_manager
711
.GetManagedObjects().items()
712
if client_interface in ifs_and_props}
713
except dbus.exceptions.DBusException as e:
714
log.critical("Failed to access Mandos server through D-Bus:"
718
# restore dbus logger
719
dbus_logger.removeFilter(dbus_filter)
721
# Compile dict of (clients: properties) to process
725
clients = {bus.get_object(busname, path): properties
726
for path, properties in mandos_clients.items()}
728
for name in clientnames:
729
for path, client in mandos_clients.items():
730
if client["Name"] == name:
731
client_objc = bus.get_object(busname, path)
732
clients[client_objc] = client
735
log.critical("Client not found on server: %r", name)
738
# Run all commands on clients
739
commands = commands_from_options(options)
740
for command in commands:
741
command.run(mandos_serv, clients)
849
class TestCaseWithAssertLogs(unittest.TestCase):
850
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
852
if not hasattr(unittest.TestCase, "assertLogs"):
853
@contextlib.contextmanager
854
def assertLogs(self, logger, level=logging.INFO):
855
capturing_handler = self.CapturingLevelHandler(level)
856
old_level = logger.level
857
old_propagate = logger.propagate
858
logger.addHandler(capturing_handler)
859
logger.setLevel(level)
860
logger.propagate = False
862
yield capturing_handler.watcher
864
logger.propagate = old_propagate
865
logger.removeHandler(capturing_handler)
866
logger.setLevel(old_level)
867
self.assertGreater(len(capturing_handler.watcher.records),
870
class CapturingLevelHandler(logging.Handler):
871
def __init__(self, level, *args, **kwargs):
872
logging.Handler.__init__(self, *args, **kwargs)
873
self.watcher = self.LoggingWatcher([], [])
874
def emit(self, record):
875
self.watcher.records.append(record)
876
self.watcher.output.append(self.format(record))
878
LoggingWatcher = collections.namedtuple("LoggingWatcher",
883
class Test_string_to_delta(TestCaseWithAssertLogs):
884
# Just test basic RFC 3339 functionality here, the doc string for
885
# rfc3339_duration_to_delta() already has more comprehensive
886
# tests, which is run by doctest.
888
def test_rfc3339_zero_seconds(self):
889
self.assertEqual(datetime.timedelta(),
890
string_to_delta("PT0S"))
892
def test_rfc3339_zero_days(self):
893
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
895
def test_rfc3339_one_second(self):
896
self.assertEqual(datetime.timedelta(0, 1),
897
string_to_delta("PT1S"))
899
def test_rfc3339_two_hours(self):
900
self.assertEqual(datetime.timedelta(0, 7200),
901
string_to_delta("PT2H"))
744
class Test_milliseconds_to_string(unittest.TestCase):
746
self.assertEqual(milliseconds_to_string(93785000),
748
def test_no_days(self):
749
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
750
def test_all_zero(self):
751
self.assertEqual(milliseconds_to_string(0), "00:00:00")
752
def test_no_fractional_seconds(self):
753
self.assertEqual(milliseconds_to_string(400), "00:00:00")
754
self.assertEqual(milliseconds_to_string(900), "00:00:00")
755
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
757
class Test_string_to_delta(unittest.TestCase):
758
def test_handles_basic_rfc3339(self):
759
self.assertEqual(string_to_delta("PT0S"),
760
datetime.timedelta())
761
self.assertEqual(string_to_delta("P0D"),
762
datetime.timedelta())
763
self.assertEqual(string_to_delta("PT1S"),
764
datetime.timedelta(0, 1))
765
self.assertEqual(string_to_delta("PT2H"),
766
datetime.timedelta(0, 7200))
903
767
def test_falls_back_to_pre_1_6_1_with_warning(self):
904
with self.assertLogs(log, logging.WARNING):
905
value = string_to_delta("2h")
906
self.assertEqual(datetime.timedelta(0, 7200), value)
909
class Test_check_option_syntax(unittest.TestCase):
911
self.parser = argparse.ArgumentParser()
912
add_command_line_options(self.parser)
914
def test_actions_requires_client_or_all(self):
915
for action, value in self.actions.items():
916
options = self.parser.parse_args()
917
setattr(options, action, value)
918
with self.assertParseError():
919
self.check_option_syntax(options)
921
# This mostly corresponds to the definition from has_actions() in
922
# check_option_syntax()
924
# The actual values set here are not that important, but we do
925
# at least stick to the correct types, even though they are
929
"bump_timeout": True,
930
"start_checker": True,
931
"stop_checker": True,
935
"timeout": datetime.timedelta(),
936
"extended_timeout": datetime.timedelta(),
937
"interval": datetime.timedelta(),
938
"approved_by_default": True,
939
"approval_delay": datetime.timedelta(),
940
"approval_duration": datetime.timedelta(),
942
"secret": io.BytesIO(b"x"),
947
@contextlib.contextmanager
948
def assertParseError(self):
949
with self.assertRaises(SystemExit) as e:
950
with self.redirect_stderr_to_devnull():
952
# Exit code from argparse is guaranteed to be "2". Reference:
953
# https://docs.python.org/3/library
954
# /argparse.html#exiting-methods
955
self.assertEqual(2, e.exception.code)
958
@contextlib.contextmanager
959
def redirect_stderr_to_devnull():
960
old_stderr = sys.stderr
961
with contextlib.closing(open(os.devnull, "w")) as null:
768
# assertLogs only exists in Python 3.4
769
if hasattr(self, "assertLogs"):
770
with self.assertLogs(log, logging.WARNING):
771
value = string_to_delta("2h")
773
class WarningFilter(logging.Filter):
774
"""Don't show, but record the presence of, warnings"""
775
def filter(self, record):
776
is_warning = record.levelno >= logging.WARNING
777
self.found = is_warning or getattr(self, "found",
779
return not is_warning
780
warning_filter = WarningFilter()
781
log.addFilter(warning_filter)
783
value = string_to_delta("2h")
966
sys.stderr = old_stderr
968
def check_option_syntax(self, options):
969
check_option_syntax(self.parser, options)
971
def test_actions_all_conflicts_with_verbose(self):
972
for action, value in self.actions.items():
973
options = self.parser.parse_args()
974
setattr(options, action, value)
976
options.verbose = True
977
with self.assertParseError():
978
self.check_option_syntax(options)
980
def test_actions_with_client_conflicts_with_verbose(self):
981
for action, value in self.actions.items():
982
options = self.parser.parse_args()
983
setattr(options, action, value)
984
options.verbose = True
985
options.client = ["client"]
986
with self.assertParseError():
987
self.check_option_syntax(options)
989
def test_dump_json_conflicts_with_verbose(self):
990
options = self.parser.parse_args()
991
options.dump_json = True
992
options.verbose = True
993
with self.assertParseError():
994
self.check_option_syntax(options)
996
def test_dump_json_conflicts_with_action(self):
997
for action, value in self.actions.items():
998
options = self.parser.parse_args()
999
setattr(options, action, value)
1000
options.dump_json = True
1001
with self.assertParseError():
1002
self.check_option_syntax(options)
1004
def test_all_can_not_be_alone(self):
1005
options = self.parser.parse_args()
1007
with self.assertParseError():
1008
self.check_option_syntax(options)
1010
def test_all_is_ok_with_any_action(self):
1011
for action, value in self.actions.items():
1012
options = self.parser.parse_args()
1013
setattr(options, action, value)
1015
self.check_option_syntax(options)
1017
def test_any_action_is_ok_with_one_client(self):
1018
for action, value in self.actions.items():
1019
options = self.parser.parse_args()
1020
setattr(options, action, value)
1021
options.client = ["client"]
1022
self.check_option_syntax(options)
1024
def test_one_client_with_all_actions_except_is_enabled(self):
1025
options = self.parser.parse_args()
1026
for action, value in self.actions.items():
1027
if action == "is_enabled":
1029
setattr(options, action, value)
1030
options.client = ["client"]
1031
self.check_option_syntax(options)
1033
def test_two_clients_with_all_actions_except_is_enabled(self):
1034
options = self.parser.parse_args()
1035
for action, value in self.actions.items():
1036
if action == "is_enabled":
1038
setattr(options, action, value)
1039
options.client = ["client1", "client2"]
1040
self.check_option_syntax(options)
1042
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1043
for action, value in self.actions.items():
1044
if action == "is_enabled":
1046
options = self.parser.parse_args()
1047
setattr(options, action, value)
1048
options.client = ["client1", "client2"]
1049
self.check_option_syntax(options)
1051
def test_is_enabled_fails_without_client(self):
1052
options = self.parser.parse_args()
1053
options.is_enabled = True
1054
with self.assertParseError():
1055
self.check_option_syntax(options)
1057
def test_is_enabled_fails_with_two_clients(self):
1058
options = self.parser.parse_args()
1059
options.is_enabled = True
1060
options.client = ["client1", "client2"]
1061
with self.assertParseError():
1062
self.check_option_syntax(options)
1064
def test_remove_can_only_be_combined_with_action_deny(self):
1065
for action, value in self.actions.items():
1066
if action in {"remove", "deny"}:
1068
options = self.parser.parse_args()
1069
setattr(options, action, value)
1071
options.remove = True
1072
with self.assertParseError():
1073
self.check_option_syntax(options)
1076
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1077
def test_calls_and_returns_get_object_on_bus(self):
1078
class MockBus(object):
1080
def get_object(mockbus_self, busname, dbus_path):
1081
# Note that "self" is still the testcase instance,
1082
# this MockBus instance is in "mockbus_self".
1083
self.assertEqual(dbus_busname, busname)
1084
self.assertEqual(server_dbus_path, dbus_path)
1085
mockbus_self.called = True
1088
mockbus = get_mandos_dbus_object(bus=MockBus())
1089
self.assertIsInstance(mockbus, MockBus)
1090
self.assertTrue(mockbus.called)
1092
def test_logs_and_exits_on_dbus_error(self):
1093
class FailingBusStub(object):
1094
def get_object(self, busname, dbus_path):
1095
raise dbus.exceptions.DBusException("Test")
1097
with self.assertLogs(log, logging.CRITICAL):
1098
with self.assertRaises(SystemExit) as e:
1099
bus = get_mandos_dbus_object(bus=FailingBusStub())
1101
if isinstance(e.exception.code, int):
1102
self.assertNotEqual(0, e.exception.code)
1104
self.assertIsNotNone(e.exception.code)
1107
class Test_get_managed_objects(TestCaseWithAssertLogs):
1108
def test_calls_and_returns_GetManagedObjects(self):
1109
managed_objects = {"/clients/client": { "Name": "client"}}
1110
class ObjectManagerStub(object):
1111
def GetManagedObjects(self):
1112
return managed_objects
1113
retval = get_managed_objects(ObjectManagerStub())
1114
self.assertDictEqual(managed_objects, retval)
1116
def test_logs_and_exits_on_dbus_error(self):
1117
dbus_logger = logging.getLogger("dbus.proxies")
1119
class ObjectManagerFailingStub(object):
1120
def GetManagedObjects(self):
1121
dbus_logger.error("Test")
1122
raise dbus.exceptions.DBusException("Test")
1124
class CountingHandler(logging.Handler):
1126
def emit(self, record):
1129
counting_handler = CountingHandler()
1131
dbus_logger.addHandler(counting_handler)
1134
with self.assertLogs(log, logging.CRITICAL) as watcher:
1135
with self.assertRaises(SystemExit) as e:
1136
get_managed_objects(ObjectManagerFailingStub())
1138
dbus_logger.removeFilter(counting_handler)
1140
# Make sure the dbus logger was suppressed
1141
self.assertEqual(0, counting_handler.count)
1143
# Test that the dbus_logger still works
1144
with self.assertLogs(dbus_logger, logging.ERROR):
1145
dbus_logger.error("Test")
1147
if isinstance(e.exception.code, int):
1148
self.assertNotEqual(0, e.exception.code)
1150
self.assertIsNotNone(e.exception.code)
1153
class Test_commands_from_options(unittest.TestCase):
1155
self.parser = argparse.ArgumentParser()
1156
add_command_line_options(self.parser)
1158
def test_is_enabled(self):
1159
self.assert_command_from_args(["--is-enabled", "client"],
1162
def assert_command_from_args(self, args, command_cls,
1164
"""Assert that parsing ARGS should result in an instance of
1165
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1166
options = self.parser.parse_args(args)
1167
check_option_syntax(self.parser, options)
1168
commands = commands_from_options(options)
1169
self.assertEqual(1, len(commands))
1170
command = commands[0]
1171
self.assertIsInstance(command, command_cls)
1172
for key, value in cmd_attrs.items():
1173
self.assertEqual(value, getattr(command, key))
1175
def test_is_enabled_short(self):
1176
self.assert_command_from_args(["-V", "client"],
1179
def test_approve(self):
1180
self.assert_command_from_args(["--approve", "client"],
1183
def test_approve_short(self):
1184
self.assert_command_from_args(["-A", "client"],
1187
def test_deny(self):
1188
self.assert_command_from_args(["--deny", "client"],
1191
def test_deny_short(self):
1192
self.assert_command_from_args(["-D", "client"], command.Deny)
1194
def test_remove(self):
1195
self.assert_command_from_args(["--remove", "client"],
1198
def test_deny_before_remove(self):
1199
options = self.parser.parse_args(["--deny", "--remove",
1201
check_option_syntax(self.parser, options)
1202
commands = commands_from_options(options)
1203
self.assertEqual(2, len(commands))
1204
self.assertIsInstance(commands[0], command.Deny)
1205
self.assertIsInstance(commands[1], command.Remove)
1207
def test_deny_before_remove_reversed(self):
1208
options = self.parser.parse_args(["--remove", "--deny",
1210
check_option_syntax(self.parser, options)
1211
commands = commands_from_options(options)
1212
self.assertEqual(2, len(commands))
1213
self.assertIsInstance(commands[0], command.Deny)
1214
self.assertIsInstance(commands[1], command.Remove)
1216
def test_remove_short(self):
1217
self.assert_command_from_args(["-r", "client"],
1220
def test_dump_json(self):
1221
self.assert_command_from_args(["--dump-json"],
1224
def test_enable(self):
1225
self.assert_command_from_args(["--enable", "client"],
1228
def test_enable_short(self):
1229
self.assert_command_from_args(["-e", "client"],
1232
def test_disable(self):
1233
self.assert_command_from_args(["--disable", "client"],
1236
def test_disable_short(self):
1237
self.assert_command_from_args(["-d", "client"],
1240
def test_bump_timeout(self):
1241
self.assert_command_from_args(["--bump-timeout", "client"],
1242
command.BumpTimeout)
1244
def test_bump_timeout_short(self):
1245
self.assert_command_from_args(["-b", "client"],
1246
command.BumpTimeout)
1248
def test_start_checker(self):
1249
self.assert_command_from_args(["--start-checker", "client"],
1250
command.StartChecker)
1252
def test_stop_checker(self):
1253
self.assert_command_from_args(["--stop-checker", "client"],
1254
command.StopChecker)
1256
def test_approve_by_default(self):
1257
self.assert_command_from_args(["--approve-by-default",
1259
command.ApproveByDefault)
1261
def test_deny_by_default(self):
1262
self.assert_command_from_args(["--deny-by-default", "client"],
1263
command.DenyByDefault)
1265
def test_checker(self):
1266
self.assert_command_from_args(["--checker", ":", "client"],
1270
def test_checker_empty(self):
1271
self.assert_command_from_args(["--checker", "", "client"],
1275
def test_checker_short(self):
1276
self.assert_command_from_args(["-c", ":", "client"],
1280
def test_host(self):
1281
self.assert_command_from_args(
1282
["--host", "client.example.org", "client"],
1283
command.SetHost, value_to_set="client.example.org")
1285
def test_host_short(self):
1286
self.assert_command_from_args(
1287
["-H", "client.example.org", "client"], command.SetHost,
1288
value_to_set="client.example.org")
1290
def test_secret_devnull(self):
1291
self.assert_command_from_args(["--secret", os.path.devnull,
1292
"client"], command.SetSecret,
1295
def test_secret_tempfile(self):
1296
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1297
value = b"secret\0xyzzy\nbar"
1300
self.assert_command_from_args(["--secret", f.name,
1305
def test_secret_devnull_short(self):
1306
self.assert_command_from_args(["-s", os.path.devnull,
1307
"client"], command.SetSecret,
1310
def test_secret_tempfile_short(self):
1311
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1312
value = b"secret\0xyzzy\nbar"
1315
self.assert_command_from_args(["-s", f.name, "client"],
1319
def test_timeout(self):
1320
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1322
value_to_set=300000)
1324
def test_timeout_short(self):
1325
self.assert_command_from_args(["-t", "PT5M", "client"],
1327
value_to_set=300000)
1329
def test_extended_timeout(self):
1330
self.assert_command_from_args(["--extended-timeout", "PT15M",
1332
command.SetExtendedTimeout,
1333
value_to_set=900000)
1335
def test_interval(self):
1336
self.assert_command_from_args(["--interval", "PT2M",
1337
"client"], command.SetInterval,
1338
value_to_set=120000)
1340
def test_interval_short(self):
1341
self.assert_command_from_args(["-i", "PT2M", "client"],
1342
command.SetInterval,
1343
value_to_set=120000)
1345
def test_approval_delay(self):
1346
self.assert_command_from_args(["--approval-delay", "PT30S",
1348
command.SetApprovalDelay,
1351
def test_approval_duration(self):
1352
self.assert_command_from_args(["--approval-duration", "PT1S",
1354
command.SetApprovalDuration,
1357
def test_print_table(self):
1358
self.assert_command_from_args([], command.PrintTable,
1361
def test_print_table_verbose(self):
1362
self.assert_command_from_args(["--verbose"],
1366
def test_print_table_verbose_short(self):
1367
self.assert_command_from_args(["-v"], command.PrintTable,
1371
class TestCommand(unittest.TestCase):
785
log.removeFilter(warning_filter)
786
self.assertTrue(getattr(warning_filter, "found", False))
787
self.assertEqual(value, datetime.timedelta(0, 7200))
790
class TestCmd(unittest.TestCase):
1372
791
"""Abstract class for tests of command classes"""
1374
792
def setUp(self):
1376
794
class MockClient(object):
1377
795
def __init__(self, name, **attributes):
1378
self.__dbus_object_path__ = "/clients/{}".format(name)
796
self.__dbus_object_path__ = "objpath_{}".format(name)
1379
797
self.attributes = attributes
1380
798
self.attributes["Name"] = name
1382
def Set(self, interface, propname, value, dbus_interface):
1383
testcase.assertEqual(client_dbus_interface, interface)
1384
testcase.assertEqual(dbus.PROPERTIES_IFACE,
1386
self.attributes[propname] = value
800
def Set(self, interface, property, value, dbus_interface):
801
testcase.assertEqual(interface, client_interface)
802
testcase.assertEqual(dbus_interface,
803
dbus.PROPERTIES_IFACE)
804
self.attributes[property] = value
805
def Get(self, interface, property, dbus_interface):
806
testcase.assertEqual(interface, client_interface)
807
testcase.assertEqual(dbus_interface,
808
dbus.PROPERTIES_IFACE)
809
return self.attributes[property]
1387
810
def Approve(self, approve, dbus_interface):
1388
testcase.assertEqual(client_dbus_interface,
811
testcase.assertEqual(dbus_interface, client_interface)
1390
812
self.calls.append(("Approve", (approve,
1391
813
dbus_interface)))
1392
814
self.client = MockClient(
1432
854
ApprovedByDefault=dbus.Boolean(False),
1433
855
LastApprovalRequest="2019-01-03T00:00:00",
1434
856
ApprovalDelay=30000,
1435
ApprovalDuration=93785000,
857
ApprovalDuration=1000,
1437
859
ExtendedTimeout=900000,
1438
860
Expires="2019-02-05T00:00:00",
1439
861
LastCheckerStatus=-2)
1440
862
self.clients = collections.OrderedDict(
1442
(self.client.__dbus_object_path__,
1443
self.client.attributes),
1444
(self.other_client.__dbus_object_path__,
1445
self.other_client.attributes),
864
(self.client, self.client.attributes),
865
(self.other_client, self.other_client.attributes),
1447
self.one_client = {self.client.__dbus_object_path__:
1448
self.client.attributes}
1452
class MockBus(object):
1454
def get_object(client_bus_name, path):
1455
self.assertEqual(dbus_busname, client_bus_name)
1456
# Note: "self" here is the TestCmd instance, not the
1457
# MockBus instance, since this is a static method!
1458
if path == self.client.__dbus_object_path__:
1460
elif path == self.other_client.__dbus_object_path__:
1461
return self.other_client
1465
class TestBaseCommands(TestCommand):
1467
def test_IsEnabled_exits_successfully(self):
867
self.one_client = {self.client: self.client.attributes}
869
class TestPrintTableCmd(TestCmd):
870
def test_normal(self):
871
output = PrintTableCmd().output(self.clients)
872
expected_output = """
873
Name Enabled Timeout Last Successful Check
874
foo Yes 00:05:00 2019-02-03T00:00:00
875
barbar Yes 00:05:00 2019-02-04T00:00:00
877
self.assertEqual(output, expected_output)
878
def test_verbose(self):
879
output = PrintTableCmd(verbose=True).output(self.clients)
880
expected_output = """
881
Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status
882
foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0
883
barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2
885
self.assertEqual(output, expected_output)
886
def test_one_client(self):
887
output = PrintTableCmd().output(self.one_client)
888
expected_output = """
889
Name Enabled Timeout Last Successful Check
890
foo Yes 00:05:00 2019-02-03T00:00:00
892
self.assertEqual(output, expected_output)
894
class TestDumpJSONCmd(TestCmd):
896
self.expected_json = {
899
"KeyID": ("92ed150794387c03ce684574b1139a65"
900
"94a34f895daaaf09fd8ea90a27cddb12"),
901
"Host": "foo.example.org",
904
"LastCheckedOK": "2019-02-03T00:00:00",
905
"Created": "2019-01-02T00:00:00",
907
"Fingerprint": ("778827225BA7DE539C5A"
908
"7CFA59CFF7CDBD9A5920"),
909
"CheckerRunning": False,
910
"LastEnabled": "2019-01-03T00:00:00",
911
"ApprovalPending": False,
912
"ApprovedByDefault": True,
913
"LastApprovalRequest": "",
915
"ApprovalDuration": 1000,
916
"Checker": "fping -q -- %(host)s",
917
"ExtendedTimeout": 900000,
918
"Expires": "2019-02-04T00:00:00",
919
"LastCheckerStatus": 0,
923
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
924
"6ab612cff5ad227247e46c2b020f441c"),
928
"LastCheckedOK": "2019-02-04T00:00:00",
929
"Created": "2019-01-03T00:00:00",
931
"Fingerprint": ("3E393AEAEFB84C7E89E2"
932
"F547B3A107558FCA3A27"),
933
"CheckerRunning": True,
934
"LastEnabled": "2019-01-04T00:00:00",
935
"ApprovalPending": False,
936
"ApprovedByDefault": False,
937
"LastApprovalRequest": "2019-01-03T00:00:00",
938
"ApprovalDelay": 30000,
939
"ApprovalDuration": 1000,
941
"ExtendedTimeout": 900000,
942
"Expires": "2019-02-05T00:00:00",
943
"LastCheckerStatus": -2,
946
return super(TestDumpJSONCmd, self).setUp()
947
def test_normal(self):
948
json_data = json.loads(DumpJSONCmd().output(self.clients))
949
self.assertDictEqual(json_data, self.expected_json)
950
def test_one_client(self):
951
clients = self.one_client
952
json_data = json.loads(DumpJSONCmd().output(clients))
953
expected_json = {"foo": self.expected_json["foo"]}
954
self.assertDictEqual(json_data, expected_json)
956
class TestIsEnabledCmd(TestCmd):
957
def test_is_enabled(self):
958
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
959
for client, properties in self.clients.items()))
960
def test_is_enabled_run_exits_successfully(self):
1468
961
with self.assertRaises(SystemExit) as e:
1469
command.IsEnabled().run(self.one_client)
962
IsEnabledCmd().run(None, self.one_client)
1470
963
if e.exception.code is not None:
1471
self.assertEqual(0, e.exception.code)
964
self.assertEqual(e.exception.code, 0)
1473
966
self.assertIsNone(e.exception.code)
1475
def test_IsEnabled_exits_with_failure(self):
967
def test_is_enabled_run_exits_with_failure(self):
1476
968
self.client.attributes["Enabled"] = dbus.Boolean(False)
1477
969
with self.assertRaises(SystemExit) as e:
1478
command.IsEnabled().run(self.one_client)
970
IsEnabledCmd().run(None, self.one_client)
1479
971
if isinstance(e.exception.code, int):
1480
self.assertNotEqual(0, e.exception.code)
972
self.assertNotEqual(e.exception.code, 0)
1482
974
self.assertIsNotNone(e.exception.code)
1484
def test_Approve(self):
1485
command.Approve().run(self.clients, self.bus)
1486
for clientpath in self.clients:
1487
client = self.bus.get_object(dbus_busname, clientpath)
1488
self.assertIn(("Approve", (True, client_dbus_interface)),
1491
def test_Deny(self):
1492
command.Deny().run(self.clients, self.bus)
1493
for clientpath in self.clients:
1494
client = self.bus.get_object(dbus_busname, clientpath)
1495
self.assertIn(("Approve", (False, client_dbus_interface)),
1498
def test_Remove(self):
1499
class MandosSpy(object):
976
class TestRemoveCmd(TestCmd):
977
def test_remove(self):
978
class MockMandos(object):
1500
979
def __init__(self):
1502
981
def RemoveClient(self, dbus_path):
1503
982
self.calls.append(("RemoveClient", (dbus_path,)))
1504
mandos = MandosSpy()
1505
command.Remove().run(self.clients, self.bus, mandos)
1506
for clientpath in self.clients:
1507
self.assertIn(("RemoveClient", (clientpath,)),
983
mandos = MockMandos()
984
super(TestRemoveCmd, self).setUp()
985
RemoveCmd().run(mandos, self.clients)
986
self.assertEqual(len(mandos.calls), 2)
987
for client in self.clients:
988
self.assertIn(("RemoveClient",
989
(client.__dbus_object_path__,)),
1513
"KeyID": ("92ed150794387c03ce684574b1139a65"
1514
"94a34f895daaaf09fd8ea90a27cddb12"),
1515
"Host": "foo.example.org",
1518
"LastCheckedOK": "2019-02-03T00:00:00",
1519
"Created": "2019-01-02T00:00:00",
1521
"Fingerprint": ("778827225BA7DE539C5A"
1522
"7CFA59CFF7CDBD9A5920"),
1523
"CheckerRunning": False,
1524
"LastEnabled": "2019-01-03T00:00:00",
1525
"ApprovalPending": False,
1526
"ApprovedByDefault": True,
1527
"LastApprovalRequest": "",
1529
"ApprovalDuration": 1000,
1530
"Checker": "fping -q -- %(host)s",
1531
"ExtendedTimeout": 900000,
1532
"Expires": "2019-02-04T00:00:00",
1533
"LastCheckerStatus": 0,
1537
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1538
"6ab612cff5ad227247e46c2b020f441c"),
1539
"Host": "192.0.2.3",
1542
"LastCheckedOK": "2019-02-04T00:00:00",
1543
"Created": "2019-01-03T00:00:00",
1545
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1546
"F547B3A107558FCA3A27"),
1547
"CheckerRunning": True,
1548
"LastEnabled": "2019-01-04T00:00:00",
1549
"ApprovalPending": False,
1550
"ApprovedByDefault": False,
1551
"LastApprovalRequest": "2019-01-03T00:00:00",
1552
"ApprovalDelay": 30000,
1553
"ApprovalDuration": 93785000,
1555
"ExtendedTimeout": 900000,
1556
"Expires": "2019-02-05T00:00:00",
1557
"LastCheckerStatus": -2,
1561
def test_DumpJSON_normal(self):
1562
with self.capture_stdout_to_buffer() as buffer:
1563
command.DumpJSON().run(self.clients)
1564
json_data = json.loads(buffer.getvalue())
1565
self.assertDictEqual(self.expected_json, json_data)
1568
@contextlib.contextmanager
1569
def capture_stdout_to_buffer():
1570
capture_buffer = io.StringIO()
1571
old_stdout = sys.stdout
1572
sys.stdout = capture_buffer
1574
yield capture_buffer
1576
sys.stdout = old_stdout
1578
def test_DumpJSON_one_client(self):
1579
with self.capture_stdout_to_buffer() as buffer:
1580
command.DumpJSON().run(self.one_client)
1581
json_data = json.loads(buffer.getvalue())
1582
expected_json = {"foo": self.expected_json["foo"]}
1583
self.assertDictEqual(expected_json, json_data)
1585
def test_PrintTable_normal(self):
1586
with self.capture_stdout_to_buffer() as buffer:
1587
command.PrintTable().run(self.clients)
1588
expected_output = "\n".join((
1589
"Name Enabled Timeout Last Successful Check",
1590
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1591
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1593
self.assertEqual(expected_output, buffer.getvalue())
1595
def test_PrintTable_verbose(self):
1596
with self.capture_stdout_to_buffer() as buffer:
1597
command.PrintTable(verbose=True).run(self.clients)
1612
"Last Successful Check ",
1613
"2019-02-03T00:00:00 ",
1614
"2019-02-04T00:00:00 ",
1617
"2019-01-02T00:00:00 ",
1618
"2019-01-03T00:00:00 ",
1630
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1632
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1636
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1637
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1639
"Check Is Running ",
1644
"2019-01-03T00:00:00 ",
1645
"2019-01-04T00:00:00 ",
1647
"Approval Is Pending ",
1651
"Approved By Default ",
1655
"Last Approval Request ",
1657
"2019-01-03T00:00:00 ",
1663
"Approval Duration ",
1668
"fping -q -- %(host)s ",
1671
"Extended Timeout ",
1676
"2019-02-04T00:00:00 ",
1677
"2019-02-05T00:00:00 ",
1679
"Last Checker Status",
1684
num_lines = max(len(rows) for rows in columns)
1685
expected_output = ("\n".join("".join(rows[line]
1686
for rows in columns)
1687
for line in range(num_lines))
1689
self.assertEqual(expected_output, buffer.getvalue())
1691
def test_PrintTable_one_client(self):
1692
with self.capture_stdout_to_buffer() as buffer:
1693
command.PrintTable().run(self.one_client)
1694
expected_output = "\n".join((
1695
"Name Enabled Timeout Last Successful Check",
1696
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1698
self.assertEqual(expected_output, buffer.getvalue())
1701
class TestPropertySetterCmd(TestCommand):
1702
"""Abstract class for tests of command.PropertySetter classes"""
992
class TestApproveCmd(TestCmd):
993
def test_approve(self):
994
ApproveCmd().run(None, self.clients)
995
for client in self.clients:
996
self.assertIn(("Approve", (True, client_interface)),
999
class TestDenyCmd(TestCmd):
1000
def test_deny(self):
1001
DenyCmd().run(None, self.clients)
1002
for client in self.clients:
1003
self.assertIn(("Approve", (False, client_interface)),
1006
class TestEnableCmd(TestCmd):
1007
def test_enable(self):
1008
for client in self.clients:
1009
client.attributes["Enabled"] = False
1011
EnableCmd().run(None, self.clients)
1013
for client in self.clients:
1014
self.assertTrue(client.attributes["Enabled"])
1016
class TestDisableCmd(TestCmd):
1017
def test_disable(self):
1018
DisableCmd().run(None, self.clients)
1020
for client in self.clients:
1021
self.assertFalse(client.attributes["Enabled"])
1023
class Unique(object):
1024
"""Class for objects which exist only to be unique objects, since
1025
unittest.mock.sentinel only exists in Python 3.3"""
1027
class TestPropertyCmd(TestCmd):
1028
"""Abstract class for tests of PropertyCmd classes"""
1703
1029
def runTest(self):
1704
1030
if not hasattr(self, "command"):
1707
1033
self.values_to_set)
1708
1034
for value_to_set, value_to_get in zip(self.values_to_set,
1709
1035
values_to_get):
1710
for clientpath in self.clients:
1711
client = self.bus.get_object(dbus_busname, clientpath)
1712
old_value = client.attributes[self.propname]
1713
client.attributes[self.propname] = self.Unique()
1036
for client in self.clients:
1037
old_value = client.attributes[self.property]
1038
self.assertNotIsInstance(old_value, Unique)
1039
client.attributes[self.property] = Unique()
1714
1040
self.run_command(value_to_set, self.clients)
1715
for clientpath in self.clients:
1716
client = self.bus.get_object(dbus_busname, clientpath)
1717
value = client.attributes[self.propname]
1718
self.assertNotIsInstance(value, self.Unique)
1719
self.assertEqual(value_to_get, value)
1721
class Unique(object):
1722
"""Class for objects which exist only to be unique objects,
1723
since unittest.mock.sentinel only exists in Python 3.3"""
1041
for client in self.clients:
1042
value = client.attributes[self.property]
1043
self.assertNotIsInstance(value, Unique)
1044
self.assertEqual(value, value_to_get)
1725
1045
def run_command(self, value, clients):
1726
self.command().run(clients, self.bus)
1729
class TestEnableCmd(TestPropertySetterCmd):
1730
command = command.Enable
1731
propname = "Enabled"
1732
values_to_set = [dbus.Boolean(True)]
1735
class TestDisableCmd(TestPropertySetterCmd):
1736
command = command.Disable
1737
propname = "Enabled"
1738
values_to_set = [dbus.Boolean(False)]
1741
class TestBumpTimeoutCmd(TestPropertySetterCmd):
1742
command = command.BumpTimeout
1743
propname = "LastCheckedOK"
1046
self.command().run(None, clients)
1048
class TestBumpTimeoutCmd(TestPropertyCmd):
1049
command = BumpTimeoutCmd
1050
property = "LastCheckedOK"
1744
1051
values_to_set = [""]
1747
class TestStartCheckerCmd(TestPropertySetterCmd):
1748
command = command.StartChecker
1749
propname = "CheckerRunning"
1750
values_to_set = [dbus.Boolean(True)]
1753
class TestStopCheckerCmd(TestPropertySetterCmd):
1754
command = command.StopChecker
1755
propname = "CheckerRunning"
1756
values_to_set = [dbus.Boolean(False)]
1759
class TestApproveByDefaultCmd(TestPropertySetterCmd):
1760
command = command.ApproveByDefault
1761
propname = "ApprovedByDefault"
1762
values_to_set = [dbus.Boolean(True)]
1765
class TestDenyByDefaultCmd(TestPropertySetterCmd):
1766
command = command.DenyByDefault
1767
propname = "ApprovedByDefault"
1768
values_to_set = [dbus.Boolean(False)]
1771
class TestPropertySetterValueCmd(TestPropertySetterCmd):
1772
"""Abstract class for tests of PropertySetterValueCmd classes"""
1053
class TestStartCheckerCmd(TestPropertyCmd):
1054
command = StartCheckerCmd
1055
property = "CheckerRunning"
1056
values_to_set = [dbus.Boolean(True)]
1058
class TestStopCheckerCmd(TestPropertyCmd):
1059
command = StopCheckerCmd
1060
property = "CheckerRunning"
1061
values_to_set = [dbus.Boolean(False)]
1063
class TestApproveByDefaultCmd(TestPropertyCmd):
1064
command = ApproveByDefaultCmd
1065
property = "ApprovedByDefault"
1066
values_to_set = [dbus.Boolean(True)]
1068
class TestDenyByDefaultCmd(TestPropertyCmd):
1069
command = DenyByDefaultCmd
1070
property = "ApprovedByDefault"
1071
values_to_set = [dbus.Boolean(False)]
1073
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1074
"""Abstract class for tests of PropertyCmd classes using the
1075
ValueArgumentMixIn"""
1774
1076
def runTest(self):
1775
if type(self) is TestPropertySetterValueCmd:
1077
if type(self) is TestValueArgumentPropertyCmd:
1777
return super(TestPropertySetterValueCmd, self).runTest()
1079
return super(TestValueArgumentPropertyCmd, self).runTest()
1779
1080
def run_command(self, value, clients):
1780
self.command(value).run(clients, self.bus)
1783
class TestSetCheckerCmd(TestPropertySetterValueCmd):
1784
command = command.SetChecker
1785
propname = "Checker"
1081
self.command(value).run(None, clients)
1083
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1084
command = SetCheckerCmd
1085
property = "Checker"
1786
1086
values_to_set = ["", ":", "fping -q -- %s"]
1789
class TestSetHostCmd(TestPropertySetterValueCmd):
1790
command = command.SetHost
1792
values_to_set = ["192.0.2.3", "client.example.org"]
1795
class TestSetSecretCmd(TestPropertySetterValueCmd):
1796
command = command.SetSecret
1798
values_to_set = [io.BytesIO(b""),
1088
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1089
command = SetHostCmd
1091
values_to_set = ["192.0.2.3", "foo.example.org"]
1093
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1094
command = SetSecretCmd
1096
values_to_set = [open("/dev/null", "rb"),
1799
1097
io.BytesIO(b"secret\0xyzzy\nbar")]
1800
values_to_get = [f.getvalue() for f in values_to_set]
1803
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
1804
command = command.SetTimeout
1805
propname = "Timeout"
1806
values_to_set = [datetime.timedelta(),
1807
datetime.timedelta(minutes=5),
1808
datetime.timedelta(seconds=1),
1809
datetime.timedelta(weeks=1),
1810
datetime.timedelta(weeks=52)]
1811
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1814
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
1815
command = command.SetExtendedTimeout
1816
propname = "ExtendedTimeout"
1817
values_to_set = [datetime.timedelta(),
1818
datetime.timedelta(minutes=5),
1819
datetime.timedelta(seconds=1),
1820
datetime.timedelta(weeks=1),
1821
datetime.timedelta(weeks=52)]
1822
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1825
class TestSetIntervalCmd(TestPropertySetterValueCmd):
1826
command = command.SetInterval
1827
propname = "Interval"
1828
values_to_set = [datetime.timedelta(),
1829
datetime.timedelta(minutes=5),
1830
datetime.timedelta(seconds=1),
1831
datetime.timedelta(weeks=1),
1832
datetime.timedelta(weeks=52)]
1833
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1836
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
1837
command = command.SetApprovalDelay
1838
propname = "ApprovalDelay"
1839
values_to_set = [datetime.timedelta(),
1840
datetime.timedelta(minutes=5),
1841
datetime.timedelta(seconds=1),
1842
datetime.timedelta(weeks=1),
1843
datetime.timedelta(weeks=52)]
1844
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1847
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
1848
command = command.SetApprovalDuration
1849
propname = "ApprovalDuration"
1850
values_to_set = [datetime.timedelta(),
1851
datetime.timedelta(minutes=5),
1852
datetime.timedelta(seconds=1),
1853
datetime.timedelta(weeks=1),
1854
datetime.timedelta(weeks=52)]
1855
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1098
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1100
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1101
command = SetTimeoutCmd
1102
property = "Timeout"
1103
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1106
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1107
command = SetExtendedTimeoutCmd
1108
property = "ExtendedTimeout"
1109
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1112
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1113
command = SetIntervalCmd
1114
property = "Interval"
1115
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1118
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1119
command = SetApprovalDelayCmd
1120
property = "ApprovalDelay"
1121
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1122
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1124
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1125
command = SetApprovalDurationCmd
1126
property = "ApprovalDuration"
1127
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1128
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1130
class Test_command_from_options(unittest.TestCase):
1132
self.parser = argparse.ArgumentParser()
1133
add_command_line_options(self.parser)
1134
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1135
"""Assert that parsing ARGS should result in an instance of
1136
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1137
options = self.parser.parse_args(args)
1138
commands = commands_from_options(options)
1139
self.assertEqual(len(commands), 1)
1140
command = commands[0]
1141
self.assertIsInstance(command, command_cls)
1142
for key, value in cmd_attrs.items():
1143
self.assertEqual(getattr(command, key), value)
1144
def test_print_table(self):
1145
self.assert_command_from_args([], PrintTableCmd,
1148
def test_print_table_verbose(self):
1149
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1152
def test_enable(self):
1153
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1155
def test_disable(self):
1156
self.assert_command_from_args(["--disable", "foo"],
1159
def test_bump_timeout(self):
1160
self.assert_command_from_args(["--bump-timeout", "foo"],
1163
def test_start_checker(self):
1164
self.assert_command_from_args(["--start-checker", "foo"],
1167
def test_stop_checker(self):
1168
self.assert_command_from_args(["--stop-checker", "foo"],
1171
def test_remove(self):
1172
self.assert_command_from_args(["--remove", "foo"],
1175
def test_checker(self):
1176
self.assert_command_from_args(["--checker", ":", "foo"],
1177
SetCheckerCmd, value_to_set=":")
1179
def test_timeout(self):
1180
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1182
value_to_set=300000)
1184
def test_extended_timeout(self):
1185
self.assert_command_from_args(["--extended-timeout", "PT15M",
1187
SetExtendedTimeoutCmd,
1188
value_to_set=900000)
1190
def test_interval(self):
1191
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1193
value_to_set=120000)
1195
def test_approve_by_default(self):
1196
self.assert_command_from_args(["--approve-by-default", "foo"],
1197
ApproveByDefaultCmd)
1199
def test_deny_by_default(self):
1200
self.assert_command_from_args(["--deny-by-default", "foo"],
1203
def test_approval_delay(self):
1204
self.assert_command_from_args(["--approval-delay", "PT30S",
1205
"foo"], SetApprovalDelayCmd,
1208
def test_approval_duration(self):
1209
self.assert_command_from_args(["--approval-duration", "PT1S",
1210
"foo"], SetApprovalDurationCmd,
1213
def test_host(self):
1214
self.assert_command_from_args(["--host", "foo.example.org",
1216
value_to_set="foo.example.org")
1218
def test_secret_devnull(self):
1219
self.assert_command_from_args(["--secret", os.path.devnull,
1220
"foo"], SetSecretCmd,
1223
def test_secret_tempfile(self):
1224
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1225
value = b"secret\0xyzzy\nbar"
1228
self.assert_command_from_args(["--secret", f.name,
1229
"foo"], SetSecretCmd,
1232
def test_approve(self):
1233
self.assert_command_from_args(["--approve", "foo"],
1236
def test_deny(self):
1237
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1239
def test_dump_json(self):
1240
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1242
def test_is_enabled(self):
1243
self.assert_command_from_args(["--is-enabled", "foo"],