To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 994
- compare with another revision
- download diff
- download tarball
- view history from revision 994
- Committer: Teddy Hogeborn
- Date: 2019-03-02 01:28:12 UTC
- Revision ID: teddy@recompile.se-20190302012812-3ymo5liufbe9jr6o
mandos-ctl: Add tests for table_rows_of_clients()
* mandos-ctl (class Test_table_rows_of_clients): New.
* mandos-ctl (class Test_table_rows_of_clients): New.
- files modified:
- mandos-ctl
added
removed
mandos-ctl
61
61
62
62
locale.setlocale(locale.LC_ALL, "")
63
63
64
tablewords = {
65
"Name": "Name",
66
"Enabled": "Enabled",
67
"Timeout": "Timeout",
68
"LastCheckedOK": "Last Successful Check",
69
"LastApprovalRequest": "Last Approval Request",
70
"Created": "Created",
71
"Interval": "Interval",
72
"Host": "Host",
73
"Fingerprint": "Fingerprint",
74
"KeyID": "Key ID",
75
"CheckerRunning": "Check Is Running",
76
"LastEnabled": "Last Enabled",
77
"ApprovalPending": "Approval Is Pending",
78
"ApprovedByDefault": "Approved By Default",
79
"ApprovalDelay": "Approval Delay",
80
"ApprovalDuration": "Approval Duration",
81
"Checker": "Checker",
82
"ExtendedTimeout": "Extended Timeout",
83
"Expires": "Expires",
84
"LastCheckerStatus": "Last Checker Status",
85
}
86
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
64
87
domain = "se.recompile"
65
88
busname = domain + ".Mandos"
66
89
server_path = "/"
215
238
216
239
217
240
def string_to_delta(interval):
218
"""Parse a string and return a datetime.timedelta"""
241
"""Parse a string and return a datetime.timedelta
242
"""
219
243
220
244
try:
221
245
return rfc3339_duration_to_delta(interval)
226
250
227
251
228
252
def parse_pre_1_6_1_interval(interval):
229
"""Parse an interval string as documented by Mandos before 1.6.1,
230
and return a datetime.timedelta
231
253
"""Parse an interval string as documented by Mandos before 1.6.1, and
254
return a datetime.timedelta
232
255
>>> parse_pre_1_6_1_interval('7d')
233
256
datetime.timedelta(7)
234
257
>>> parse_pre_1_6_1_interval('60s')
268
291
return value
269
292
270
293
271
## Classes for commands.
272
273
# Abstract classes first
274
class Command(object):
275
"""Abstract class for commands"""
276
def run(self, mandos, clients):
277
"""Normal commands should implement run_on_one_client(), but
278
commands which want to operate on all clients at the same time
279
can override this run() method instead."""
280
self.mandos = mandos
281
for client, properties in clients.items():
282
self.run_on_one_client(client, properties)
283
284
class PrintCmd(Command):
285
"""Abstract class for commands printing client details"""
286
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
287
"Created", "Interval", "Host", "KeyID",
288
"Fingerprint", "CheckerRunning", "LastEnabled",
289
"ApprovalPending", "ApprovedByDefault",
290
"LastApprovalRequest", "ApprovalDelay",
291
"ApprovalDuration", "Checker", "ExtendedTimeout",
292
"Expires", "LastCheckerStatus")
293
def run(self, mandos, clients):
294
print(self.output(clients))
295
296
class PropertyCmd(Command):
297
"""Abstract class for Actions for setting one client property"""
298
def run_on_one_client(self, client, properties):
299
"""Set the Client's D-Bus property"""
300
client.Set(client_interface, self.property, self.value_to_set,
301
dbus_interface=dbus.PROPERTIES_IFACE)
302
303
class ValueArgumentMixIn(object):
304
"""Mixin class for commands taking a value as argument"""
305
def __init__(self, value):
306
self.value_to_set = value
307
308
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
309
"""Mixin class for commands taking a value argument as
310
milliseconds."""
311
@property
312
def value_to_set(self):
313
return self._vts
314
@value_to_set.setter
315
def value_to_set(self, value):
316
"""When setting, convert value to a datetime.timedelta"""
317
self._vts = string_to_delta(value).total_seconds() * 1000
318
319
# Actual (non-abstract) command classes
320
321
class PrintTableCmd(PrintCmd):
322
def __init__(self, verbose=False):
323
self.verbose = verbose
324
325
def output(self, clients):
326
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
327
keywords = default_keywords
328
if self.verbose:
329
keywords = self.all_keywords
330
return str(self.TableOfClients(clients.values(), keywords))
331
332
class TableOfClients(object):
333
tableheaders = {
334
"Name": "Name",
335
"Enabled": "Enabled",
336
"Timeout": "Timeout",
337
"LastCheckedOK": "Last Successful Check",
338
"LastApprovalRequest": "Last Approval Request",
339
"Created": "Created",
340
"Interval": "Interval",
341
"Host": "Host",
342
"Fingerprint": "Fingerprint",
343
"KeyID": "Key ID",
344
"CheckerRunning": "Check Is Running",
345
"LastEnabled": "Last Enabled",
346
"ApprovalPending": "Approval Is Pending",
347
"ApprovedByDefault": "Approved By Default",
348
"ApprovalDelay": "Approval Delay",
349
"ApprovalDuration": "Approval Duration",
350
"Checker": "Checker",
351
"ExtendedTimeout": "Extended Timeout",
352
"Expires": "Expires",
353
"LastCheckerStatus": "Last Checker Status",
354
}
355
356
def __init__(self, clients, keywords, tableheaders=None):
357
self.clients = clients
358
self.keywords = keywords
359
if tableheaders is not None:
360
self.tableheaders = tableheaders
361
362
def __str__(self):
363
return "\n".join(self.rows())
364
365
if sys.version_info.major == 2:
366
__unicode__ = __str__
367
def __str__(self):
368
return str(self).encode(locale.getpreferredencoding())
369
370
def rows(self):
371
format_string = self.row_formatting_string()
372
rows = [self.header_line(format_string)]
373
rows.extend(self.client_line(client, format_string)
374
for client in self.clients)
375
return rows
376
377
def row_formatting_string(self):
378
"Format string used to format table rows"
379
return " ".join("{{{key}:{width}}}".format(
380
width=max(len(self.tableheaders[key]),
381
*(len(self.string_from_client(client, key))
382
for client in self.clients)),
383
key=key)
384
for key in self.keywords)
385
386
def string_from_client(self, client, key):
387
return self.valuetostring(client[key], key)
388
389
@staticmethod
390
def valuetostring(value, keyword):
391
if isinstance(value, dbus.Boolean):
392
return "Yes" if value else "No"
393
if keyword in ("Timeout", "Interval", "ApprovalDelay",
394
"ApprovalDuration", "ExtendedTimeout"):
395
return milliseconds_to_string(value)
396
return str(value)
397
398
def header_line(self, format_string):
399
return format_string.format(**self.tableheaders)
400
401
def client_line(self, client, format_string):
402
return format_string.format(
403
**{key: self.string_from_client(client, key)
404
for key in self.keywords})
405
406
407
408
class DumpJSONCmd(PrintCmd):
409
def output(self, clients):
410
data = {client["Name"]:
411
{key: self.dbus_boolean_to_bool(client[key])
412
for key in self.all_keywords}
413
for client in clients.values()}
414
return json.dumps(data, indent=4, separators=(',', ': '))
415
@staticmethod
416
def dbus_boolean_to_bool(value):
294
def print_clients(clients, keywords):
295
print('\n'.join(table_rows_of_clients(clients, keywords)))
296
297
def table_rows_of_clients(clients, keywords):
298
def valuetostring(value, keyword):
417
299
if isinstance(value, dbus.Boolean):
418
value = bool(value)
419
return value
420
421
class IsEnabledCmd(Command):
422
def run_on_one_client(self, client, properties):
423
if self.is_enabled(client, properties):
424
sys.exit(0)
425
sys.exit(1)
426
def is_enabled(self, client, properties):
427
return bool(properties["Enabled"])
428
429
class RemoveCmd(Command):
430
def run_on_one_client(self, client, properties):
431
self.mandos.RemoveClient(client.__dbus_object_path__)
432
433
class ApproveCmd(Command):
434
def run_on_one_client(self, client, properties):
435
client.Approve(dbus.Boolean(True),
436
dbus_interface=client_interface)
437
438
class DenyCmd(Command):
439
def run_on_one_client(self, client, properties):
440
client.Approve(dbus.Boolean(False),
441
dbus_interface=client_interface)
442
443
class EnableCmd(PropertyCmd):
444
property = "Enabled"
445
value_to_set = dbus.Boolean(True)
446
447
class DisableCmd(PropertyCmd):
448
property = "Enabled"
449
value_to_set = dbus.Boolean(False)
450
451
class BumpTimeoutCmd(PropertyCmd):
452
property = "LastCheckedOK"
453
value_to_set = ""
454
455
class StartCheckerCmd(PropertyCmd):
456
property = "CheckerRunning"
457
value_to_set = dbus.Boolean(True)
458
459
class StopCheckerCmd(PropertyCmd):
460
property = "CheckerRunning"
461
value_to_set = dbus.Boolean(False)
462
463
class ApproveByDefaultCmd(PropertyCmd):
464
property = "ApprovedByDefault"
465
value_to_set = dbus.Boolean(True)
466
467
class DenyByDefaultCmd(PropertyCmd):
468
property = "ApprovedByDefault"
469
value_to_set = dbus.Boolean(False)
470
471
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
472
property = "Checker"
473
474
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
475
property = "Host"
476
477
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
478
property = "Secret"
479
480
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
481
property = "Timeout"
482
483
class SetExtendedTimeoutCmd(PropertyCmd,
484
MillisecondsValueArgumentMixIn):
485
property = "ExtendedTimeout"
486
487
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
488
property = "Interval"
489
490
class SetApprovalDelayCmd(PropertyCmd,
491
MillisecondsValueArgumentMixIn):
492
property = "ApprovalDelay"
493
494
class SetApprovalDurationCmd(PropertyCmd,
495
MillisecondsValueArgumentMixIn):
496
property = "ApprovalDuration"
300
return "Yes" if value else "No"
301
if keyword in ("Timeout", "Interval", "ApprovalDelay",
302
"ApprovalDuration", "ExtendedTimeout"):
303
return milliseconds_to_string(value)
304
return str(value)
305
306
# Create format string to print table rows
307
format_string = " ".join("{{{key}:{width}}}".format(
308
width=max(len(tablewords[key]),
309
max(len(valuetostring(client[key], key))
310
for client in clients)),
311
key=key)
312
for key in keywords)
313
# Start with header line
314
rows = [format_string.format(**tablewords)]
315
for client in clients:
316
rows.append(format_string
317
.format(**{key: valuetostring(client[key], key)
318
for key in keywords}))
319
return rows
320
497
321
498
322
def has_actions(options):
499
323
return any((options.enable,
515
339
options.approve,
516
340
options.deny))
517
341
518
def add_command_line_options(parser):
342
343
def main():
344
parser = argparse.ArgumentParser()
519
345
parser.add_argument("--version", action="version",
520
346
version="%(prog)s {}".format(version),
521
347
help="show version number and exit")
525
351
help="Print all fields")
526
352
parser.add_argument("-j", "--dump-json", action="store_true",
527
353
help="Dump client data in JSON format")
528
enable_disable = parser.add_mutually_exclusive_group()
529
enable_disable.add_argument("-e", "--enable", action="store_true",
530
help="Enable client")
531
enable_disable.add_argument("-d", "--disable",
532
action="store_true",
533
help="disable client")
354
parser.add_argument("-e", "--enable", action="store_true",
355
help="Enable client")
356
parser.add_argument("-d", "--disable", action="store_true",
357
help="disable client")
534
358
parser.add_argument("-b", "--bump-timeout", action="store_true",
535
359
help="Bump timeout for client")
536
start_stop_checker = parser.add_mutually_exclusive_group()
537
start_stop_checker.add_argument("--start-checker",
538
action="store_true",
539
help="Start checker for client")
540
start_stop_checker.add_argument("--stop-checker",
541
action="store_true",
542
help="Stop checker for client")
360
parser.add_argument("--start-checker", action="store_true",
361
help="Start checker for client")
362
parser.add_argument("--stop-checker", action="store_true",
363
help="Stop checker for client")
543
364
parser.add_argument("-V", "--is-enabled", action="store_true",
544
365
help="Check if client is enabled")
545
366
parser.add_argument("-r", "--remove", action="store_true",
552
373
help="Set extended timeout for client")
553
374
parser.add_argument("-i", "--interval",
554
375
help="Set checker interval for client")
555
approve_deny_default = parser.add_mutually_exclusive_group()
556
approve_deny_default.add_argument(
557
"--approve-by-default", action="store_true",
558
default=None, dest="approved_by_default",
559
help="Set client to be approved by default")
560
approve_deny_default.add_argument(
561
"--deny-by-default", action="store_false",
562
dest="approved_by_default",
563
help="Set client to be denied by default")
376
parser.add_argument("--approve-by-default", action="store_true",
377
default=None, dest="approved_by_default",
378
help="Set client to be approved by default")
379
parser.add_argument("--deny-by-default", action="store_false",
380
dest="approved_by_default",
381
help="Set client to be denied by default")
564
382
parser.add_argument("--approval-delay",
565
383
help="Set delay before client approve/deny")
566
384
parser.add_argument("--approval-duration",
569
387
parser.add_argument("-s", "--secret",
570
388
type=argparse.FileType(mode="rb"),
571
389
help="Set password blob (file) for client")
572
approve_deny = parser.add_mutually_exclusive_group()
573
approve_deny.add_argument(
574
"-A", "--approve", action="store_true",
575
help="Approve any current client request")
576
approve_deny.add_argument("-D", "--deny", action="store_true",
577
help="Deny any current client request")
390
parser.add_argument("-A", "--approve", action="store_true",
391
help="Approve any current client request")
392
parser.add_argument("-D", "--deny", action="store_true",
393
help="Deny any current client request")
578
394
parser.add_argument("--check", action="store_true",
579
395
help="Run self-test")
580
396
parser.add_argument("client", nargs="*", help="Client name")
581
582
583
def commands_from_options(options):
584
585
commands = []
586
587
if options.dump_json:
588
commands.append(DumpJSONCmd())
589
590
if options.enable:
591
commands.append(EnableCmd())
592
593
if options.disable:
594
commands.append(DisableCmd())
595
596
if options.bump_timeout:
597
commands.append(BumpTimeoutCmd())
598
599
if options.start_checker:
600
commands.append(StartCheckerCmd())
601
602
if options.stop_checker:
603
commands.append(StopCheckerCmd())
604
605
if options.is_enabled:
606
commands.append(IsEnabledCmd())
607
608
if options.remove:
609
commands.append(RemoveCmd())
610
611
if options.checker is not None:
612
commands.append(SetCheckerCmd())
613
614
if options.timeout is not None:
615
commands.append(SetTimeoutCmd(options.timeout))
616
617
if options.extended_timeout:
618
commands.append(
619
SetExtendedTimeoutCmd(options.extended_timeout))
620
621
if options.interval is not None:
622
command.append(SetIntervalCmd(options.interval))
623
624
if options.approved_by_default is not None:
625
if options.approved_by_default:
626
command.append(ApproveByDefaultCmd())
627
else:
628
command.append(DenyByDefaultCmd())
629
630
if options.approval_delay is not None:
631
command.append(SetApprovalDelayCmd(options.approval_delay))
632
633
if options.approval_duration is not None:
634
command.append(
635
SetApprovalDurationCmd(options.approval_duration))
636
637
if options.host is not None:
638
command.append(SetHostCmd(options.host))
639
640
if options.secret is not None:
641
command.append(SetSecretCmd(options.secret))
642
643
if options.approve:
644
commands.append(ApproveCmd())
645
646
if options.deny:
647
commands.append(DenyCmd())
648
649
# If no command option has been given, show table of clients,
650
# optionally verbosely
651
if not commands:
652
commands.append(PrintTableCmd(verbose=options.verbose))
653
654
return commands
655
656
657
def main():
658
parser = argparse.ArgumentParser()
659
660
add_command_line_options(parser)
661
662
397
options = parser.parse_args()
663
398
664
399
if has_actions(options) and not (options.client or options.all):
670
405
parser.error("--dump-json can only be used alone.")
671
406
if options.all and not has_actions(options):
672
407
parser.error("--all requires an action.")
673
if options.is_enabled and len(options.client) > 1:
674
parser.error("--is-enabled requires exactly one client")
675
676
clientnames = options.client
677
408
678
409
try:
679
410
bus = dbus.SystemBus()
687
418
mandos_serv_object_manager = dbus.Interface(
688
419
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
689
420
690
# Filter out log message from dbus module
691
dbus_logger = logging.getLogger("dbus.proxies")
692
class NullFilter(logging.Filter):
693
def filter(self, record):
694
return False
695
dbus_filter = NullFilter()
421
# block stderr since dbus library prints to stderr
422
null = os.open(os.path.devnull, os.O_RDWR)
423
stderrcopy = os.dup(sys.stderr.fileno())
424
os.dup2(null, sys.stderr.fileno())
425
os.close(null)
696
426
try:
697
dbus_logger.addFilter(dbus_filter)
698
mandos_clients = {path: ifs_and_props[client_interface]
699
for path, ifs_and_props in
700
mandos_serv_object_manager
701
.GetManagedObjects().items()
702
if client_interface in ifs_and_props}
427
try:
428
mandos_clients = {path: ifs_and_props[client_interface]
429
for path, ifs_and_props in
430
mandos_serv_object_manager
431
.GetManagedObjects().items()
432
if client_interface in ifs_and_props}
433
finally:
434
# restore stderr
435
os.dup2(stderrcopy, sys.stderr.fileno())
436
os.close(stderrcopy)
703
437
except dbus.exceptions.DBusException as e:
704
438
log.critical("Failed to access Mandos server through D-Bus:"
705
439
"\n%s", e)
706
440
sys.exit(1)
707
finally:
708
# restore dbus logger
709
dbus_logger.removeFilter(dbus_filter)
710
441
711
442
# Compile dict of (clients: properties) to process
712
443
clients = {}
713
444
714
if not clientnames:
445
if options.all or not options.client:
715
446
clients = {bus.get_object(busname, path): properties
716
447
for path, properties in mandos_clients.items()}
717
448
else:
718
for name in clientnames:
449
for name in options.client:
719
450
for path, client in mandos_clients.items():
720
451
if client["Name"] == name:
721
452
client_objc = bus.get_object(busname, path)
725
456
log.critical("Client not found on server: %r", name)
726
457
sys.exit(1)
727
458
728
# Run all commands on clients
729
commands = commands_from_options(options)
730
for command in commands:
731
command.run(mandos_serv, clients)
459
if not has_actions(options) and clients:
460
if options.verbose or options.dump_json:
461
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
462
"Created", "Interval", "Host", "KeyID",
463
"Fingerprint", "CheckerRunning",
464
"LastEnabled", "ApprovalPending",
465
"ApprovedByDefault", "LastApprovalRequest",
466
"ApprovalDelay", "ApprovalDuration",
467
"Checker", "ExtendedTimeout", "Expires",
468
"LastCheckerStatus")
469
else:
470
keywords = defaultkeywords
471
472
if options.dump_json:
473
json.dump({client["Name"]: {key:
474
bool(client[key])
475
if isinstance(client[key],
476
dbus.Boolean)
477
else client[key]
478
for key in keywords}
479
for client in clients.values()},
480
fp=sys.stdout, indent=4,
481
separators=(',', ': '))
482
print()
483
else:
484
print_clients(clients.values(), keywords)
485
else:
486
# Process each client in the list by all selected options
487
for client in clients:
488
489
def set_client_prop(prop, value):
490
"""Set a Client D-Bus property"""
491
client.Set(client_interface, prop, value,
492
dbus_interface=dbus.PROPERTIES_IFACE)
493
494
def set_client_prop_ms(prop, value):
495
"""Set a Client D-Bus property, converted
496
from a string to milliseconds."""
497
set_client_prop(prop,
498
string_to_delta(value).total_seconds()
499
* 1000)
500
501
if options.remove:
502
mandos_serv.RemoveClient(client.__dbus_object_path__)
503
if options.enable:
504
set_client_prop("Enabled", dbus.Boolean(True))
505
if options.disable:
506
set_client_prop("Enabled", dbus.Boolean(False))
507
if options.bump_timeout:
508
set_client_prop("LastCheckedOK", "")
509
if options.start_checker:
510
set_client_prop("CheckerRunning", dbus.Boolean(True))
511
if options.stop_checker:
512
set_client_prop("CheckerRunning", dbus.Boolean(False))
513
if options.is_enabled:
514
if client.Get(client_interface, "Enabled",
515
dbus_interface=dbus.PROPERTIES_IFACE):
516
sys.exit(0)
517
else:
518
sys.exit(1)
519
if options.checker is not None:
520
set_client_prop("Checker", options.checker)
521
if options.host is not None:
522
set_client_prop("Host", options.host)
523
if options.interval is not None:
524
set_client_prop_ms("Interval", options.interval)
525
if options.approval_delay is not None:
526
set_client_prop_ms("ApprovalDelay",
527
options.approval_delay)
528
if options.approval_duration is not None:
529
set_client_prop_ms("ApprovalDuration",
530
options.approval_duration)
531
if options.timeout is not None:
532
set_client_prop_ms("Timeout", options.timeout)
533
if options.extended_timeout is not None:
534
set_client_prop_ms("ExtendedTimeout",
535
options.extended_timeout)
536
if options.secret is not None:
537
set_client_prop("Secret",
538
dbus.ByteArray(options.secret.read()))
539
if options.approved_by_default is not None:
540
set_client_prop("ApprovedByDefault",
541
dbus.Boolean(options
542
.approved_by_default))
543
if options.approve:
544
client.Approve(dbus.Boolean(True),
545
dbus_interface=client_interface)
546
elif options.deny:
547
client.Approve(dbus.Boolean(False),
548
dbus_interface=client_interface)
732
549
733
550
734
551
class Test_milliseconds_to_string(unittest.TestCase):
746
563
747
564
class Test_string_to_delta(unittest.TestCase):
748
565
def test_handles_basic_rfc3339(self):
749
self.assertEqual(string_to_delta("PT0S"),
750
datetime.timedelta())
751
self.assertEqual(string_to_delta("P0D"),
752
datetime.timedelta())
753
self.assertEqual(string_to_delta("PT1S"),
754
datetime.timedelta(0, 1))
755
566
self.assertEqual(string_to_delta("PT2H"),
756
567
datetime.timedelta(0, 7200))
757
568
def test_falls_back_to_pre_1_6_1_with_warning(self):
760
571
with self.assertLogs(log, logging.WARNING):
761
572
value = string_to_delta("2h")
762
573
else:
763
class WarningFilter(logging.Filter):
764
"""Don't show, but record the presence of, warnings"""
765
def filter(self, record):
766
is_warning = record.levelno >= logging.WARNING
767
self.found = is_warning or getattr(self, "found",
768
False)
769
return not is_warning
770
warning_filter = WarningFilter()
771
log.addFilter(warning_filter)
772
try:
773
value = string_to_delta("2h")
774
finally:
775
log.removeFilter(warning_filter)
776
self.assertTrue(getattr(warning_filter, "found", False))
574
value = string_to_delta("2h")
777
575
self.assertEqual(value, datetime.timedelta(0, 7200))
778
576
779
780
class TestCmd(unittest.TestCase):
781
"""Abstract class for tests of command classes"""
782
def setUp(self):
783
testcase = self
784
class MockClient(object):
785
def __init__(self, name, **attributes):
786
self.__dbus_object_path__ = "objpath_{}".format(name)
787
self.attributes = attributes
788
self.attributes["Name"] = name
789
self.calls = []
790
def Set(self, interface, property, value, dbus_interface):
791
testcase.assertEqual(interface, client_interface)
792
testcase.assertEqual(dbus_interface,
793
dbus.PROPERTIES_IFACE)
794
self.attributes[property] = value
795
def Get(self, interface, property, dbus_interface):
796
testcase.assertEqual(interface, client_interface)
797
testcase.assertEqual(dbus_interface,
798
dbus.PROPERTIES_IFACE)
799
return self.attributes[property]
800
def Approve(self, approve, dbus_interface):
801
testcase.assertEqual(dbus_interface, client_interface)
802
self.calls.append(("Approve", (approve,
803
dbus_interface)))
804
self.client = MockClient(
805
"foo",
806
KeyID=("92ed150794387c03ce684574b1139a65"
807
"94a34f895daaaf09fd8ea90a27cddb12"),
808
Secret=b"secret",
809
Host="foo.example.org",
810
Enabled=dbus.Boolean(True),
811
Timeout=300000,
812
LastCheckedOK="2019-02-03T00:00:00",
813
Created="2019-01-02T00:00:00",
814
Interval=120000,
815
Fingerprint=("778827225BA7DE539C5A"
816
"7CFA59CFF7CDBD9A5920"),
817
CheckerRunning=dbus.Boolean(False),
818
LastEnabled="2019-01-03T00:00:00",
819
ApprovalPending=dbus.Boolean(False),
820
ApprovedByDefault=dbus.Boolean(True),
821
LastApprovalRequest="",
822
ApprovalDelay=0,
823
ApprovalDuration=1000,
824
Checker="fping -q -- %(host)s",
825
ExtendedTimeout=900000,
826
Expires="2019-02-04T00:00:00",
827
LastCheckerStatus=0)
828
self.other_client = MockClient(
829
"barbar",
830
KeyID=("0558568eedd67d622f5c83b35a115f79"
831
"6ab612cff5ad227247e46c2b020f441c"),
832
Secret=b"secretbar",
833
Host="192.0.2.3",
834
Enabled=dbus.Boolean(True),
835
Timeout=300000,
836
LastCheckedOK="2019-02-04T00:00:00",
837
Created="2019-01-03T00:00:00",
838
Interval=120000,
839
Fingerprint=("3E393AEAEFB84C7E89E2"
840
"F547B3A107558FCA3A27"),
841
CheckerRunning=dbus.Boolean(True),
842
LastEnabled="2019-01-04T00:00:00",
843
ApprovalPending=dbus.Boolean(False),
844
ApprovedByDefault=dbus.Boolean(False),
845
LastApprovalRequest="2019-01-03T00:00:00",
846
ApprovalDelay=30000,
847
ApprovalDuration=1000,
848
Checker=":",
849
ExtendedTimeout=900000,
850
Expires="2019-02-05T00:00:00",
851
LastCheckerStatus=-2)
852
self.clients = collections.OrderedDict(
853
[
854
(self.client, self.client.attributes),
855
(self.other_client, self.other_client.attributes),
856
])
857
self.one_client = {self.client: self.client.attributes}
858
859
class TestPrintTableCmd(TestCmd):
860
def test_normal(self):
861
output = PrintTableCmd().output(self.clients)
862
expected_output = """
863
Name Enabled Timeout Last Successful Check
864
foo Yes 00:05:00 2019-02-03T00:00:00
865
barbar Yes 00:05:00 2019-02-04T00:00:00
866
"""[1:-1]
867
self.assertEqual(output, expected_output)
868
def test_verbose(self):
869
output = PrintTableCmd(verbose=True).output(self.clients)
870
expected_output = """
871
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
872
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
873
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
874
"""[1:-1]
875
self.assertEqual(output, expected_output)
876
def test_one_client(self):
877
output = PrintTableCmd().output(self.one_client)
878
expected_output = """
879
Name Enabled Timeout Last Successful Check
880
foo Yes 00:05:00 2019-02-03T00:00:00
881
"""[1:-1]
882
self.assertEqual(output, expected_output)
883
884
class TestDumpJSONCmd(TestCmd):
885
def setUp(self):
886
self.expected_json = {
887
"foo": {
888
"Name": "foo",
889
"KeyID": ("92ed150794387c03ce684574b1139a65"
890
"94a34f895daaaf09fd8ea90a27cddb12"),
891
"Host": "foo.example.org",
892
"Enabled": True,
893
"Timeout": 300000,
894
"LastCheckedOK": "2019-02-03T00:00:00",
895
"Created": "2019-01-02T00:00:00",
896
"Interval": 120000,
897
"Fingerprint": ("778827225BA7DE539C5A"
898
"7CFA59CFF7CDBD9A5920"),
899
"CheckerRunning": False,
900
"LastEnabled": "2019-01-03T00:00:00",
901
"ApprovalPending": False,
902
"ApprovedByDefault": True,
903
"LastApprovalRequest": "",
904
"ApprovalDelay": 0,
905
"ApprovalDuration": 1000,
906
"Checker": "fping -q -- %(host)s",
907
"ExtendedTimeout": 900000,
908
"Expires": "2019-02-04T00:00:00",
909
"LastCheckerStatus": 0,
910
},
911
"barbar": {
912
"Name": "barbar",
913
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
914
"6ab612cff5ad227247e46c2b020f441c"),
915
"Host": "192.0.2.3",
916
"Enabled": True,
917
"Timeout": 300000,
918
"LastCheckedOK": "2019-02-04T00:00:00",
919
"Created": "2019-01-03T00:00:00",
920
"Interval": 120000,
921
"Fingerprint": ("3E393AEAEFB84C7E89E2"
922
"F547B3A107558FCA3A27"),
923
"CheckerRunning": True,
924
"LastEnabled": "2019-01-04T00:00:00",
925
"ApprovalPending": False,
926
"ApprovedByDefault": False,
927
"LastApprovalRequest": "2019-01-03T00:00:00",
928
"ApprovalDelay": 30000,
929
"ApprovalDuration": 1000,
930
"Checker": ":",
931
"ExtendedTimeout": 900000,
932
"Expires": "2019-02-05T00:00:00",
933
"LastCheckerStatus": -2,
934
},
577
class Test_table_rows_of_clients(unittest.TestCase):
578
def setUp(self):
579
global tablewords
580
self.old_tablewords = tablewords
581
tablewords = {
582
"Attr1": "X",
583
"AttrTwo": "Yy",
584
"AttrThree": "Zzz",
585
"Bool": "A D-BUS Boolean",
586
"NonDbusBoolean": "A Non-D-BUS Boolean",
587
"Integer": "An Integer",
588
"Timeout": "Timedelta 1",
589
"Interval": "Timedelta 2",
590
"ApprovalDelay": "Timedelta 3",
591
"ApprovalDuration": "Timedelta 4",
592
"ExtendedTimeout": "Timedelta 5",
593
"String": "A String",
935
594
}
936
return super(TestDumpJSONCmd, self).setUp()
937
def test_normal(self):
938
json_data = json.loads(DumpJSONCmd().output(self.clients))
939
self.assertDictEqual(json_data, self.expected_json)
940
def test_one_client(self):
941
clients = self.one_client
942
json_data = json.loads(DumpJSONCmd().output(clients))
943
expected_json = {"foo": self.expected_json["foo"]}
944
self.assertDictEqual(json_data, expected_json)
945
946
class TestIsEnabledCmd(TestCmd):
947
def test_is_enabled(self):
948
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
949
for client, properties in self.clients.items()))
950
def test_is_enabled_run_exits_successfully(self):
951
with self.assertRaises(SystemExit) as e:
952
IsEnabledCmd().run(None, self.one_client)
953
if e.exception.code is not None:
954
self.assertEqual(e.exception.code, 0)
955
else:
956
self.assertIsNone(e.exception.code)
957
def test_is_enabled_run_exits_with_failure(self):
958
self.client.attributes["Enabled"] = dbus.Boolean(False)
959
with self.assertRaises(SystemExit) as e:
960
IsEnabledCmd().run(None, self.one_client)
961
if isinstance(e.exception.code, int):
962
self.assertNotEqual(e.exception.code, 0)
963
else:
964
self.assertIsNotNone(e.exception.code)
965
966
class TestRemoveCmd(TestCmd):
967
def test_remove(self):
968
class MockMandos(object):
969
def __init__(self):
970
self.calls = []
971
def RemoveClient(self, dbus_path):
972
self.calls.append(("RemoveClient", (dbus_path,)))
973
mandos = MockMandos()
974
super(TestRemoveCmd, self).setUp()
975
RemoveCmd().run(mandos, self.clients)
976
self.assertEqual(len(mandos.calls), 2)
977
for client in self.clients:
978
self.assertIn(("RemoveClient",
979
(client.__dbus_object_path__,)),
980
mandos.calls)
981
982
class TestApproveCmd(TestCmd):
983
def test_approve(self):
984
ApproveCmd().run(None, self.clients)
985
for client in self.clients:
986
self.assertIn(("Approve", (True, client_interface)),
987
client.calls)
988
989
class TestDenyCmd(TestCmd):
990
def test_deny(self):
991
DenyCmd().run(None, self.clients)
992
for client in self.clients:
993
self.assertIn(("Approve", (False, client_interface)),
994
client.calls)
995
996
class TestEnableCmd(TestCmd):
997
def test_enable(self):
998
for client in self.clients:
999
client.attributes["Enabled"] = False
1000
1001
EnableCmd().run(None, self.clients)
1002
1003
for client in self.clients:
1004
self.assertTrue(client.attributes["Enabled"])
1005
1006
class TestDisableCmd(TestCmd):
1007
def test_disable(self):
1008
DisableCmd().run(None, self.clients)
1009
1010
for client in self.clients:
1011
self.assertFalse(client.attributes["Enabled"])
1012
1013
class Unique(object):
1014
"""Class for objects which exist only to be unique objects, since
1015
unittest.mock.sentinel only exists in Python 3.3"""
1016
1017
class TestPropertyCmd(TestCmd):
1018
"""Abstract class for tests of PropertyCmd classes"""
1019
def runTest(self):
1020
if not hasattr(self, "command"):
1021
return
1022
values_to_get = getattr(self, "values_to_get",
1023
self.values_to_set)
1024
for value_to_set, value_to_get in zip(self.values_to_set,
1025
values_to_get):
1026
for client in self.clients:
1027
old_value = client.attributes[self.property]
1028
self.assertNotIsInstance(old_value, Unique)
1029
client.attributes[self.property] = Unique()
1030
self.run_command(value_to_set, self.clients)
1031
for client in self.clients:
1032
value = client.attributes[self.property]
1033
self.assertNotIsInstance(value, Unique)
1034
self.assertEqual(value, value_to_get)
1035
def run_command(self, value, clients):
1036
self.command().run(None, clients)
1037
1038
class TestBumpTimeoutCmd(TestPropertyCmd):
1039
command = BumpTimeoutCmd
1040
property = "LastCheckedOK"
1041
values_to_set = [""]
1042
1043
class TestStartCheckerCmd(TestPropertyCmd):
1044
command = StartCheckerCmd
1045
property = "CheckerRunning"
1046
values_to_set = [dbus.Boolean(True)]
1047
1048
class TestStopCheckerCmd(TestPropertyCmd):
1049
command = StopCheckerCmd
1050
property = "CheckerRunning"
1051
values_to_set = [dbus.Boolean(False)]
1052
1053
class TestApproveByDefaultCmd(TestPropertyCmd):
1054
command = ApproveByDefaultCmd
1055
property = "ApprovedByDefault"
1056
values_to_set = [dbus.Boolean(True)]
1057
1058
class TestDenyByDefaultCmd(TestPropertyCmd):
1059
command = DenyByDefaultCmd
1060
property = "ApprovedByDefault"
1061
values_to_set = [dbus.Boolean(False)]
1062
1063
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1064
"""Abstract class for tests of PropertyCmd classes using the
1065
ValueArgumentMixIn"""
1066
def runTest(self):
1067
if type(self) is TestValueArgumentPropertyCmd:
1068
return
1069
return super(TestValueArgumentPropertyCmd, self).runTest()
1070
def run_command(self, value, clients):
1071
self.command(value).run(None, clients)
1072
1073
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1074
command = SetCheckerCmd
1075
property = "Checker"
1076
values_to_set = ["", ":", "fping -q -- %s"]
1077
1078
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1079
command = SetHostCmd
1080
property = "Host"
1081
values_to_set = ["192.0.2.3", "foo.example.org"]
1082
1083
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1084
command = SetSecretCmd
1085
property = "Secret"
1086
values_to_set = [b"", b"secret"]
1087
1088
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1089
command = SetTimeoutCmd
1090
property = "Timeout"
1091
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1092
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1093
1094
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1095
command = SetExtendedTimeoutCmd
1096
property = "ExtendedTimeout"
1097
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1098
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1099
1100
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1101
command = SetIntervalCmd
1102
property = "Interval"
1103
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1105
1106
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1107
command = SetApprovalDelayCmd
1108
property = "ApprovalDelay"
1109
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
values_to_get = [0, 300000, 1000, 120000, 31449600000]
1111
1112
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1113
command = SetApprovalDurationCmd
1114
property = "ApprovalDuration"
1115
values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
values_to_get = [0, 300000, 1000, 120000, 31449600000]
595
self.keywords = ["Attr1", "AttrTwo"]
596
self.clients = [
597
{
598
"Attr1": "x1",
599
"AttrTwo": "y1",
600
"AttrThree": "z1",
601
"Bool": dbus.Boolean(False),
602
"NonDbusBoolean": False,
603
"Integer": 0,
604
"Timeout": 0,
605
"Interval": 1000,
606
"ApprovalDelay": 2000,
607
"ApprovalDuration": 3000,
608
"ExtendedTimeout": 4000,
609
"String": "",
610
},
611
{
612
"Attr1": "x2",
613
"AttrTwo": "y2",
614
"AttrThree": "z2",
615
"Bool": dbus.Boolean(True),
616
"NonDbusBoolean": True,
617
"Integer": 1,
618
"Timeout": 93785000,
619
"Interval": 93786000,
620
"ApprovalDelay": 93787000,
621
"ApprovalDuration": 93788000,
622
"ExtendedTimeout": 93789000,
623
"String": "A huge string which will not fit," * 10,
624
},
625
]
626
def tearDown(self):
627
global tablewords
628
tablewords = self.old_tablewords
629
def test_short_header(self):
630
rows = table_rows_of_clients(self.clients, self.keywords)
631
expected_rows = [
632
"X Yy",
633
"x1 y1",
634
"x2 y2"]
635
self.assertEqual(rows, expected_rows)
636
def test_booleans(self):
637
keywords = ["Bool", "NonDbusBoolean"]
638
rows = table_rows_of_clients(self.clients, keywords)
639
expected_rows = [
640
"A D-BUS Boolean A Non-D-BUS Boolean",
641
"No False ",
642
"Yes True ",
643
]
644
self.assertEqual(rows, expected_rows)
645
def test_milliseconds_detection(self):
646
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
647
"ApprovalDuration", "ExtendedTimeout"]
648
rows = table_rows_of_clients(self.clients, keywords)
649
expected_rows = ("""
650
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
651
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
652
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
653
"""
654
).splitlines()[1:]
655
self.assertEqual(rows, expected_rows)
656
def test_empty_and_long_string_values(self):
657
keywords = ["String"]
658
rows = table_rows_of_clients(self.clients, keywords)
659
expected_rows = ("""
660
A String
661
662
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
663
"""
664
).splitlines()[1:]
665
self.assertEqual(rows, expected_rows)
1117
666
1118
667
1119
668
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0