To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/release
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 237.7.603
- compare with another revision
- download diff
- download tarball
- view history from revision 237.7.603
- Committer: Teddy Hogeborn
- Date: 2019-03-12 20:13:34 UTC
- mto: This revision was merged to the branch mainline in revision 382.
- Revision ID: teddy@recompile.se-20190312201334-my3htrprewjosuw5
mandos-ctl: Refactor
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
- files modified:
- mandos-ctl
added
removed
mandos-ctl
93
93
if options.debug:
94
94
log.setLevel(logging.DEBUG)
95
95
96
bus = dbus.SystemBus()
97
98
mandos_dbus_object = get_mandos_dbus_object(bus)
99
100
mandos_serv = dbus.Interface(
101
mandos_dbus_object, dbus_interface=server_dbus_interface)
96
try:
97
bus = dbus.SystemBus()
98
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
99
dbus_busname, server_dbus_path)
100
mandos_dbus_objc = bus.get_object(dbus_busname,
101
server_dbus_path)
102
except dbus.exceptions.DBusException:
103
log.critical("Could not connect to Mandos server")
104
sys.exit(1)
105
106
mandos_serv = dbus.Interface(mandos_dbus_objc,
107
dbus_interface=server_dbus_interface)
102
108
mandos_serv_object_manager = dbus.Interface(
103
mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
104
105
managed_objects = get_managed_objects(mandos_serv_object_manager)
106
107
all_clients = {}
108
for path, ifs_and_props in managed_objects.items():
109
try:
110
all_clients[path] = ifs_and_props[client_dbus_interface]
111
except KeyError:
112
pass
113
114
# Compile dict of (clientpath: properties) to process
109
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
110
111
# Filter out log message from dbus module
112
dbus_logger = logging.getLogger("dbus.proxies")
113
class NullFilter(logging.Filter):
114
def filter(self, record):
115
return False
116
dbus_filter = NullFilter()
117
try:
118
dbus_logger.addFilter(dbus_filter)
119
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
120
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
121
mandos_clients = {path: ifs_and_props[client_dbus_interface]
122
for path, ifs_and_props in
123
mandos_serv_object_manager
124
.GetManagedObjects().items()
125
if client_dbus_interface in ifs_and_props}
126
except dbus.exceptions.DBusException as e:
127
log.critical("Failed to access Mandos server through D-Bus:"
128
"\n%s", e)
129
sys.exit(1)
130
finally:
131
# restore dbus logger
132
dbus_logger.removeFilter(dbus_filter)
133
134
# Compile dict of (clients: properties) to process
135
clients = {}
136
115
137
if not clientnames:
116
clients = all_clients
138
clients = {objpath: properties
139
for objpath, properties in mandos_clients.items()}
117
140
else:
118
clients = {}
119
141
for name in clientnames:
120
for objpath, properties in all_clients.items():
142
for objpath, properties in mandos_clients.items():
121
143
if properties["Name"] == name:
122
144
clients[objpath] = properties
123
145
break
125
147
log.critical("Client not found on server: %r", name)
126
148
sys.exit(1)
127
149
150
# Run all commands on clients
128
151
commands = commands_from_options(options)
129
130
152
for command in commands:
131
153
command.run(clients, bus, mandos_serv)
132
154
232
254
>>> rfc3339_duration_to_delta("")
233
255
Traceback (most recent call last):
234
256
...
235
ValueError: Invalid RFC 3339 duration: ""
257
ValueError: Invalid RFC 3339 duration: u''
236
258
>>> # Must start with "P":
237
259
>>> rfc3339_duration_to_delta("1D")
238
260
Traceback (most recent call last):
239
261
...
240
ValueError: Invalid RFC 3339 duration: "1D"
262
ValueError: Invalid RFC 3339 duration: u'1D'
241
263
>>> # Must use correct order
242
264
>>> rfc3339_duration_to_delta("PT1S2M")
243
265
Traceback (most recent call last):
244
266
...
245
ValueError: Invalid RFC 3339 duration: "PT1S2M"
267
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
246
268
>>> # Time needs time marker
247
269
>>> rfc3339_duration_to_delta("P1H2S")
248
270
Traceback (most recent call last):
249
271
...
250
ValueError: Invalid RFC 3339 duration: "P1H2S"
272
ValueError: Invalid RFC 3339 duration: u'P1H2S'
251
273
>>> # Weeks can not be combined with anything else
252
274
>>> rfc3339_duration_to_delta("P1D2W")
253
275
Traceback (most recent call last):
254
276
...
255
ValueError: Invalid RFC 3339 duration: "P1D2W"
277
ValueError: Invalid RFC 3339 duration: u'P1D2W'
256
278
>>> rfc3339_duration_to_delta("P2W2H")
257
279
Traceback (most recent call last):
258
280
...
259
ValueError: Invalid RFC 3339 duration: "P2W2H"
281
ValueError: Invalid RFC 3339 duration: u'P2W2H'
260
282
"""
261
283
262
284
# Parsing an RFC 3339 duration with regular expressions is not
333
355
break
334
356
else:
335
357
# No currently valid tokens were found
336
raise ValueError("Invalid RFC 3339 duration: \"{}\""
358
raise ValueError("Invalid RFC 3339 duration: {!r}"
337
359
.format(duration))
338
360
# End token found
339
361
return value
424
446
options.remove = True
425
447
426
448
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,
433
server_dbus_path)
434
return mandos_dbus_object
435
436
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
439
try:
440
yield
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
443
sys.exit(1)
444
445
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
454
455
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
460
461
def __enter__(self):
462
self.logger.addFilter(self.nullfilter)
463
return self
464
465
class NullFilter(logging.Filter):
466
def filter(self, record):
467
return False
468
469
nullfilter = NullFilter()
470
471
def __exit__(self, exc_type, exc_val, exc_tb):
472
self.logger.removeFilter(self.nullfilter)
473
474
475
449
def commands_from_options(options):
476
450
477
451
commands = []
478
452
479
453
if options.is_enabled:
480
commands.append(command.IsEnabled())
454
commands.append(IsEnabledCmd())
481
455
482
456
if options.approve:
483
commands.append(command.Approve())
457
commands.append(ApproveCmd())
484
458
485
459
if options.deny:
486
commands.append(command.Deny())
460
commands.append(DenyCmd())
487
461
488
462
if options.remove:
489
commands.append(command.Remove())
463
commands.append(RemoveCmd())
490
464
491
465
if options.dump_json:
492
commands.append(command.DumpJSON())
466
commands.append(DumpJSONCmd())
493
467
494
468
if options.enable:
495
commands.append(command.Enable())
469
commands.append(EnableCmd())
496
470
497
471
if options.disable:
498
commands.append(command.Disable())
472
commands.append(DisableCmd())
499
473
500
474
if options.bump_timeout:
501
commands.append(command.BumpTimeout())
475
commands.append(BumpTimeoutCmd())
502
476
503
477
if options.start_checker:
504
commands.append(command.StartChecker())
478
commands.append(StartCheckerCmd())
505
479
506
480
if options.stop_checker:
507
commands.append(command.StopChecker())
481
commands.append(StopCheckerCmd())
508
482
509
483
if options.approved_by_default is not None:
510
484
if options.approved_by_default:
511
commands.append(command.ApproveByDefault())
485
commands.append(ApproveByDefaultCmd())
512
486
else:
513
commands.append(command.DenyByDefault())
487
commands.append(DenyByDefaultCmd())
514
488
515
489
if options.checker is not None:
516
commands.append(command.SetChecker(options.checker))
490
commands.append(SetCheckerCmd(options.checker))
517
491
518
492
if options.host is not None:
519
commands.append(command.SetHost(options.host))
493
commands.append(SetHostCmd(options.host))
520
494
521
495
if options.secret is not None:
522
commands.append(command.SetSecret(options.secret))
496
commands.append(SetSecretCmd(options.secret))
523
497
524
498
if options.timeout is not None:
525
commands.append(command.SetTimeout(options.timeout))
499
commands.append(SetTimeoutCmd(options.timeout))
526
500
527
501
if options.extended_timeout:
528
502
commands.append(
529
command.SetExtendedTimeout(options.extended_timeout))
503
SetExtendedTimeoutCmd(options.extended_timeout))
530
504
531
505
if options.interval is not None:
532
commands.append(command.SetInterval(options.interval))
506
commands.append(SetIntervalCmd(options.interval))
533
507
534
508
if options.approval_delay is not None:
535
commands.append(
536
command.SetApprovalDelay(options.approval_delay))
509
commands.append(SetApprovalDelayCmd(options.approval_delay))
537
510
538
511
if options.approval_duration is not None:
539
512
commands.append(
540
command.SetApprovalDuration(options.approval_duration))
513
SetApprovalDurationCmd(options.approval_duration))
541
514
542
515
# If no command option has been given, show table of clients,
543
516
# optionally verbosely
544
517
if not commands:
545
commands.append(command.PrintTable(verbose=options.verbose))
518
commands.append(PrintTableCmd(verbose=options.verbose))
546
519
547
520
return commands
548
521
549
522
550
class command(object):
551
"""A namespace for command classes"""
552
553
class Base(object):
554
"""Abstract base class for commands"""
555
def run(self, clients, bus=None, mandos=None):
556
"""Normal commands should implement run_on_one_client(),
557
but commands which want to operate on all clients at the same time can
558
override this run() method instead.
559
"""
560
self.mandos = mandos
561
for clientpath, properties in clients.items():
562
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
563
dbus_busname, str(clientpath))
564
client = bus.get_object(dbus_busname, clientpath)
565
self.run_on_one_client(client, properties)
566
567
568
class IsEnabled(Base):
569
def run(self, clients, bus=None, mandos=None):
570
client, properties = next(iter(clients.items()))
571
if self.is_enabled(client, properties):
572
sys.exit(0)
573
sys.exit(1)
574
def is_enabled(self, client, properties):
575
return properties["Enabled"]
576
577
578
class Approve(Base):
579
def run_on_one_client(self, client, properties):
580
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
581
client.__dbus_object_path__,
582
client_dbus_interface)
583
client.Approve(dbus.Boolean(True),
584
dbus_interface=client_dbus_interface)
585
586
587
class Deny(Base):
588
def run_on_one_client(self, client, properties):
589
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
590
client.__dbus_object_path__,
591
client_dbus_interface)
592
client.Approve(dbus.Boolean(False),
593
dbus_interface=client_dbus_interface)
594
595
596
class Remove(Base):
597
def run_on_one_client(self, client, properties):
598
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
599
dbus_busname, server_dbus_path,
600
server_dbus_interface,
601
str(client.__dbus_object_path__))
602
self.mandos.RemoveClient(client.__dbus_object_path__)
603
604
605
class Output(Base):
606
"""Abstract class for commands outputting client details"""
607
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
608
"Created", "Interval", "Host", "KeyID",
609
"Fingerprint", "CheckerRunning",
610
"LastEnabled", "ApprovalPending",
611
"ApprovedByDefault", "LastApprovalRequest",
612
"ApprovalDelay", "ApprovalDuration",
613
"Checker", "ExtendedTimeout", "Expires",
614
"LastCheckerStatus")
615
616
def run(self, clients, bus=None, mandos=None):
617
print(self.output(clients.values()))
618
619
def output(self, clients):
620
raise NotImplementedError()
621
622
623
class DumpJSON(Output):
624
def output(self, clients):
625
data = {client["Name"]:
626
{key: self.dbus_boolean_to_bool(client[key])
627
for key in self.all_keywords}
628
for client in clients}
629
return json.dumps(data, indent=4, separators=(',', ': '))
523
class Command(object):
524
"""Abstract class for commands"""
525
def run(self, clients, bus=None, mandos=None):
526
"""Normal commands should implement run_on_one_client(), but
527
commands which want to operate on all clients at the same time
528
can override this run() method instead."""
529
self.mandos = mandos
530
for clientpath, properties in clients.items():
531
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
532
dbus_busname, str(clientpath))
533
client = bus.get_object(dbus_busname, clientpath)
534
self.run_on_one_client(client, properties)
535
536
537
class IsEnabledCmd(Command):
538
def run(self, clients, bus=None, mandos=None):
539
client, properties = next(iter(clients.items()))
540
if self.is_enabled(client, properties):
541
sys.exit(0)
542
sys.exit(1)
543
def is_enabled(self, client, properties):
544
return properties["Enabled"]
545
546
547
class ApproveCmd(Command):
548
def run_on_one_client(self, client, properties):
549
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
550
client.__dbus_object_path__, client_dbus_interface)
551
client.Approve(dbus.Boolean(True),
552
dbus_interface=client_dbus_interface)
553
554
555
class DenyCmd(Command):
556
def run_on_one_client(self, client, properties):
557
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
558
client.__dbus_object_path__, client_dbus_interface)
559
client.Approve(dbus.Boolean(False),
560
dbus_interface=client_dbus_interface)
561
562
563
class RemoveCmd(Command):
564
def run_on_one_client(self, client, properties):
565
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
566
server_dbus_path, server_dbus_interface,
567
str(client.__dbus_object_path__))
568
self.mandos.RemoveClient(client.__dbus_object_path__)
569
570
571
class PrintCmd(Command):
572
"""Abstract class for commands printing client details"""
573
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
574
"Created", "Interval", "Host", "KeyID",
575
"Fingerprint", "CheckerRunning", "LastEnabled",
576
"ApprovalPending", "ApprovedByDefault",
577
"LastApprovalRequest", "ApprovalDelay",
578
"ApprovalDuration", "Checker", "ExtendedTimeout",
579
"Expires", "LastCheckerStatus")
580
def run(self, clients, bus=None, mandos=None):
581
print(self.output(clients.values()))
582
def output(self, clients):
583
raise NotImplementedError()
584
585
586
class DumpJSONCmd(PrintCmd):
587
def output(self, clients):
588
data = {client["Name"]:
589
{key: self.dbus_boolean_to_bool(client[key])
590
for key in self.all_keywords}
591
for client in clients.values()}
592
return json.dumps(data, indent=4, separators=(',', ': '))
593
@staticmethod
594
def dbus_boolean_to_bool(value):
595
if isinstance(value, dbus.Boolean):
596
value = bool(value)
597
return value
598
599
600
class PrintTableCmd(PrintCmd):
601
def __init__(self, verbose=False):
602
self.verbose = verbose
603
604
def output(self, clients):
605
default_keywords = ("Name", "Enabled", "Timeout",
606
"LastCheckedOK")
607
keywords = default_keywords
608
if self.verbose:
609
keywords = self.all_keywords
610
return str(self.TableOfClients(clients, keywords))
611
612
class TableOfClients(object):
613
tableheaders = {
614
"Name": "Name",
615
"Enabled": "Enabled",
616
"Timeout": "Timeout",
617
"LastCheckedOK": "Last Successful Check",
618
"LastApprovalRequest": "Last Approval Request",
619
"Created": "Created",
620
"Interval": "Interval",
621
"Host": "Host",
622
"Fingerprint": "Fingerprint",
623
"KeyID": "Key ID",
624
"CheckerRunning": "Check Is Running",
625
"LastEnabled": "Last Enabled",
626
"ApprovalPending": "Approval Is Pending",
627
"ApprovedByDefault": "Approved By Default",
628
"ApprovalDelay": "Approval Delay",
629
"ApprovalDuration": "Approval Duration",
630
"Checker": "Checker",
631
"ExtendedTimeout": "Extended Timeout",
632
"Expires": "Expires",
633
"LastCheckerStatus": "Last Checker Status",
634
}
635
636
def __init__(self, clients, keywords, tableheaders=None):
637
self.clients = clients
638
self.keywords = keywords
639
if tableheaders is not None:
640
self.tableheaders = tableheaders
641
642
def __str__(self):
643
return "\n".join(self.rows())
644
645
if sys.version_info.major == 2:
646
__unicode__ = __str__
647
def __str__(self):
648
return str(self).encode(locale.getpreferredencoding())
649
650
def rows(self):
651
format_string = self.row_formatting_string()
652
rows = [self.header_line(format_string)]
653
rows.extend(self.client_line(client, format_string)
654
for client in self.clients)
655
return rows
656
657
def row_formatting_string(self):
658
"Format string used to format table rows"
659
return " ".join("{{{key}:{width}}}".format(
660
width=max(len(self.tableheaders[key]),
661
*(len(self.string_from_client(client, key))
662
for client in self.clients)),
663
key=key)
664
for key in self.keywords)
665
666
def string_from_client(self, client, key):
667
return self.valuetostring(client[key], key)
630
668
631
669
@staticmethod
632
def dbus_boolean_to_bool(value):
670
def valuetostring(value, keyword):
633
671
if isinstance(value, dbus.Boolean):
634
value = bool(value)
635
return value
636
637
638
class PrintTable(Output):
639
def __init__(self, verbose=False):
640
self.verbose = verbose
641
642
def output(self, clients):
643
default_keywords = ("Name", "Enabled", "Timeout",
644
"LastCheckedOK")
645
keywords = default_keywords
646
if self.verbose:
647
keywords = self.all_keywords
648
return str(self.TableOfClients(clients, keywords))
649
650
class TableOfClients(object):
651
tableheaders = {
652
"Name": "Name",
653
"Enabled": "Enabled",
654
"Timeout": "Timeout",
655
"LastCheckedOK": "Last Successful Check",
656
"LastApprovalRequest": "Last Approval Request",
657
"Created": "Created",
658
"Interval": "Interval",
659
"Host": "Host",
660
"Fingerprint": "Fingerprint",
661
"KeyID": "Key ID",
662
"CheckerRunning": "Check Is Running",
663
"LastEnabled": "Last Enabled",
664
"ApprovalPending": "Approval Is Pending",
665
"ApprovedByDefault": "Approved By Default",
666
"ApprovalDelay": "Approval Delay",
667
"ApprovalDuration": "Approval Duration",
668
"Checker": "Checker",
669
"ExtendedTimeout": "Extended Timeout",
670
"Expires": "Expires",
671
"LastCheckerStatus": "Last Checker Status",
672
}
673
674
def __init__(self, clients, keywords):
675
self.clients = clients
676
self.keywords = keywords
677
678
def __str__(self):
679
return "\n".join(self.rows())
680
681
if sys.version_info.major == 2:
682
__unicode__ = __str__
683
def __str__(self):
684
return str(self).encode(
685
locale.getpreferredencoding())
686
687
def rows(self):
688
format_string = self.row_formatting_string()
689
rows = [self.header_line(format_string)]
690
rows.extend(self.client_line(client, format_string)
691
for client in self.clients)
692
return rows
693
694
def row_formatting_string(self):
695
"Format string used to format table rows"
696
return " ".join("{{{key}:{width}}}".format(
697
width=max(len(self.tableheaders[key]),
698
*(len(self.string_from_client(client,
699
key))
700
for client in self.clients)),
701
key=key)
702
for key in self.keywords)
703
704
def string_from_client(self, client, key):
705
return self.valuetostring(client[key], key)
706
707
@classmethod
708
def valuetostring(cls, value, keyword):
709
if isinstance(value, dbus.Boolean):
710
return "Yes" if value else "No"
711
if keyword in ("Timeout", "Interval", "ApprovalDelay",
712
"ApprovalDuration", "ExtendedTimeout"):
713
return cls.milliseconds_to_string(value)
714
return str(value)
715
716
def header_line(self, format_string):
717
return format_string.format(**self.tableheaders)
718
719
def client_line(self, client, format_string):
720
return format_string.format(
721
**{key: self.string_from_client(client, key)
722
for key in self.keywords})
723
724
@staticmethod
725
def milliseconds_to_string(ms):
726
td = datetime.timedelta(0, 0, 0, ms)
727
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
728
.format(days="{}T".format(td.days)
729
if td.days else "",
730
hours=td.seconds // 3600,
731
minutes=(td.seconds % 3600) // 60,
732
seconds=td.seconds % 60))
733
734
735
class Property(Base):
736
"Abstract class for Actions for setting one client property"
737
738
def run_on_one_client(self, client, properties):
739
"""Set the Client's D-Bus property"""
740
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
741
client.__dbus_object_path__,
742
dbus.PROPERTIES_IFACE, client_dbus_interface,
743
self.propname, self.value_to_set
744
if not isinstance(self.value_to_set,
745
dbus.Boolean)
746
else bool(self.value_to_set))
747
client.Set(client_dbus_interface, self.propname,
748
self.value_to_set,
749
dbus_interface=dbus.PROPERTIES_IFACE)
750
751
@property
752
def propname(self):
753
raise NotImplementedError()
754
755
756
class Enable(Property):
757
propname = "Enabled"
758
value_to_set = dbus.Boolean(True)
759
760
761
class Disable(Property):
762
propname = "Enabled"
763
value_to_set = dbus.Boolean(False)
764
765
766
class BumpTimeout(Property):
767
propname = "LastCheckedOK"
768
value_to_set = ""
769
770
771
class StartChecker(Property):
772
propname = "CheckerRunning"
773
value_to_set = dbus.Boolean(True)
774
775
776
class StopChecker(Property):
777
propname = "CheckerRunning"
778
value_to_set = dbus.Boolean(False)
779
780
781
class ApproveByDefault(Property):
782
propname = "ApprovedByDefault"
783
value_to_set = dbus.Boolean(True)
784
785
786
class DenyByDefault(Property):
787
propname = "ApprovedByDefault"
788
value_to_set = dbus.Boolean(False)
789
790
791
class PropertyValue(Property):
792
"Abstract class for Property recieving a value as argument"
793
def __init__(self, value):
794
self.value_to_set = value
795
796
797
class SetChecker(PropertyValue):
798
propname = "Checker"
799
800
801
class SetHost(PropertyValue):
802
propname = "Host"
803
804
805
class SetSecret(PropertyValue):
806
propname = "Secret"
807
808
@property
809
def value_to_set(self):
810
return self._vts
811
812
@value_to_set.setter
813
def value_to_set(self, value):
814
"""When setting, read data from supplied file object"""
815
self._vts = value.read()
816
value.close()
817
818
819
class MillisecondsPropertyValueArgument(PropertyValue):
820
"""Abstract class for PropertyValue taking a value argument as
672
return "Yes" if value else "No"
673
if keyword in ("Timeout", "Interval", "ApprovalDelay",
674
"ApprovalDuration", "ExtendedTimeout"):
675
return milliseconds_to_string(value)
676
return str(value)
677
678
def header_line(self, format_string):
679
return format_string.format(**self.tableheaders)
680
681
def client_line(self, client, format_string):
682
return format_string.format(
683
**{key: self.string_from_client(client, key)
684
for key in self.keywords})
685
686
687
def milliseconds_to_string(ms):
688
td = datetime.timedelta(0, 0, 0, ms)
689
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
690
.format(days="{}T".format(td.days) if td.days else "",
691
hours=td.seconds // 3600,
692
minutes=(td.seconds % 3600) // 60,
693
seconds=td.seconds % 60))
694
695
696
class PropertyCmd(Command):
697
"""Abstract class for Actions for setting one client property"""
698
def run_on_one_client(self, client, properties):
699
"""Set the Client's D-Bus property"""
700
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
701
client.__dbus_object_path__,
702
dbus.PROPERTIES_IFACE, client_dbus_interface,
703
self.propname, self.value_to_set
704
if not isinstance(self.value_to_set, dbus.Boolean)
705
else bool(self.value_to_set))
706
client.Set(client_dbus_interface, self.propname,
707
self.value_to_set,
708
dbus_interface=dbus.PROPERTIES_IFACE)
709
@property
710
def propname(self):
711
raise NotImplementedError()
712
713
714
class EnableCmd(PropertyCmd):
715
propname = "Enabled"
716
value_to_set = dbus.Boolean(True)
717
718
719
class DisableCmd(PropertyCmd):
720
propname = "Enabled"
721
value_to_set = dbus.Boolean(False)
722
723
724
class BumpTimeoutCmd(PropertyCmd):
725
propname = "LastCheckedOK"
726
value_to_set = ""
727
728
729
class StartCheckerCmd(PropertyCmd):
730
propname = "CheckerRunning"
731
value_to_set = dbus.Boolean(True)
732
733
734
class StopCheckerCmd(PropertyCmd):
735
propname = "CheckerRunning"
736
value_to_set = dbus.Boolean(False)
737
738
739
class ApproveByDefaultCmd(PropertyCmd):
740
propname = "ApprovedByDefault"
741
value_to_set = dbus.Boolean(True)
742
743
744
class DenyByDefaultCmd(PropertyCmd):
745
propname = "ApprovedByDefault"
746
value_to_set = dbus.Boolean(False)
747
748
749
class PropertyValueCmd(PropertyCmd):
750
"""Abstract class for PropertyCmd recieving a value as argument"""
751
def __init__(self, value):
752
self.value_to_set = value
753
754
755
class SetCheckerCmd(PropertyValueCmd):
756
propname = "Checker"
757
758
759
class SetHostCmd(PropertyValueCmd):
760
propname = "Host"
761
762
763
class SetSecretCmd(PropertyValueCmd):
764
propname = "Secret"
765
@property
766
def value_to_set(self):
767
return self._vts
768
@value_to_set.setter
769
def value_to_set(self, value):
770
"""When setting, read data from supplied file object"""
771
self._vts = value.read()
772
value.close()
773
774
775
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
776
"""Abstract class for PropertyValueCmd taking a value argument as
821
777
a datetime.timedelta() but should store it as milliseconds."""
822
823
@property
824
def value_to_set(self):
825
return self._vts
826
827
@value_to_set.setter
828
def value_to_set(self, value):
829
"When setting, convert value from a datetime.timedelta"
830
self._vts = int(round(value.total_seconds() * 1000))
831
832
833
class SetTimeout(MillisecondsPropertyValueArgument):
834
propname = "Timeout"
835
836
837
class SetExtendedTimeout(MillisecondsPropertyValueArgument):
838
propname = "ExtendedTimeout"
839
840
841
class SetInterval(MillisecondsPropertyValueArgument):
842
propname = "Interval"
843
844
845
class SetApprovalDelay(MillisecondsPropertyValueArgument):
846
propname = "ApprovalDelay"
847
848
849
class SetApprovalDuration(MillisecondsPropertyValueArgument):
850
propname = "ApprovalDuration"
778
@property
779
def value_to_set(self):
780
return self._vts
781
@value_to_set.setter
782
def value_to_set(self, value):
783
"""When setting, convert value from a datetime.timedelta"""
784
self._vts = int(round(value.total_seconds() * 1000))
785
786
787
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
788
propname = "Timeout"
789
790
791
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
792
propname = "ExtendedTimeout"
793
794
795
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
796
propname = "Interval"
797
798
799
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
800
propname = "ApprovalDelay"
801
802
803
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
804
propname = "ApprovalDuration"
851
805
852
806
853
807
854
class TestCaseWithAssertLogs(unittest.TestCase):
855
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
856
857
if not hasattr(unittest.TestCase, "assertLogs"):
858
@contextlib.contextmanager
859
def assertLogs(self, logger, level=logging.INFO):
860
capturing_handler = self.CapturingLevelHandler(level)
861
old_level = logger.level
862
old_propagate = logger.propagate
863
logger.addHandler(capturing_handler)
864
logger.setLevel(level)
865
logger.propagate = False
866
try:
867
yield capturing_handler.watcher
868
finally:
869
logger.propagate = old_propagate
870
logger.removeHandler(capturing_handler)
871
logger.setLevel(old_level)
872
self.assertGreater(len(capturing_handler.watcher.records),
873
0)
874
875
class CapturingLevelHandler(logging.Handler):
876
def __init__(self, level, *args, **kwargs):
877
logging.Handler.__init__(self, *args, **kwargs)
878
self.watcher = self.LoggingWatcher([], [])
879
def emit(self, record):
880
self.watcher.records.append(record)
881
self.watcher.output.append(self.format(record))
882
883
LoggingWatcher = collections.namedtuple("LoggingWatcher",
884
("records",
885
"output"))
886
887
888
class Test_string_to_delta(TestCaseWithAssertLogs):
808
class Test_string_to_delta(unittest.TestCase):
889
809
def test_handles_basic_rfc3339(self):
890
810
self.assertEqual(string_to_delta("PT0S"),
891
811
datetime.timedelta())
895
815
datetime.timedelta(0, 1))
896
816
self.assertEqual(string_to_delta("PT2H"),
897
817
datetime.timedelta(0, 7200))
898
899
818
def test_falls_back_to_pre_1_6_1_with_warning(self):
900
with self.assertLogs(log, logging.WARNING):
901
value = string_to_delta("2h")
819
# assertLogs only exists in Python 3.4
820
if hasattr(self, "assertLogs"):
821
with self.assertLogs(log, logging.WARNING):
822
value = string_to_delta("2h")
823
else:
824
class WarningFilter(logging.Filter):
825
"""Don't show, but record the presence of, warnings"""
826
def filter(self, record):
827
is_warning = record.levelno >= logging.WARNING
828
self.found = is_warning or getattr(self, "found",
829
False)
830
return not is_warning
831
warning_filter = WarningFilter()
832
log.addFilter(warning_filter)
833
try:
834
value = string_to_delta("2h")
835
finally:
836
log.removeFilter(warning_filter)
837
self.assertTrue(getattr(warning_filter, "found", False))
902
838
self.assertEqual(value, datetime.timedelta(0, 7200))
903
839
904
840
905
841
class Test_check_option_syntax(unittest.TestCase):
906
def setUp(self):
907
self.parser = argparse.ArgumentParser()
908
add_command_line_options(self.parser)
909
910
def test_actions_requires_client_or_all(self):
911
for action, value in self.actions.items():
912
options = self.parser.parse_args()
913
setattr(options, action, value)
914
with self.assertParseError():
915
self.check_option_syntax(options)
916
917
842
# This mostly corresponds to the definition from has_actions() in
918
843
# check_option_syntax()
919
844
actions = {
940
865
"deny": True,
941
866
}
942
867
868
def setUp(self):
869
self.parser = argparse.ArgumentParser()
870
add_command_line_options(self.parser)
871
943
872
@contextlib.contextmanager
944
873
def assertParseError(self):
945
874
with self.assertRaises(SystemExit) as e:
946
with self.redirect_stderr_to_devnull():
875
with self.temporarily_suppress_stderr():
947
876
yield
948
877
# Exit code from argparse is guaranteed to be "2". Reference:
949
878
# https://docs.python.org/3/library
952
881
953
882
@staticmethod
954
883
@contextlib.contextmanager
955
def redirect_stderr_to_devnull():
884
def temporarily_suppress_stderr():
956
885
null = os.open(os.path.devnull, os.O_RDWR)
957
886
stderrcopy = os.dup(sys.stderr.fileno())
958
887
os.dup2(null, sys.stderr.fileno())
967
896
def check_option_syntax(self, options):
968
897
check_option_syntax(self.parser, options)
969
898
970
def test_actions_all_conflicts_with_verbose(self):
899
def test_actions_requires_client_or_all(self):
971
900
for action, value in self.actions.items():
972
901
options = self.parser.parse_args()
973
902
setattr(options, action, value)
974
options.all = True
975
options.verbose = True
976
903
with self.assertParseError():
977
904
self.check_option_syntax(options)
978
905
979
def test_actions_with_client_conflicts_with_verbose(self):
906
def test_actions_conflicts_with_verbose(self):
980
907
for action, value in self.actions.items():
981
908
options = self.parser.parse_args()
982
909
setattr(options, action, value)
983
910
options.verbose = True
984
options.client = ["foo"]
985
911
with self.assertParseError():
986
912
self.check_option_syntax(options)
987
913
1013
939
options.all = True
1014
940
self.check_option_syntax(options)
1015
941
1016
def test_any_action_is_ok_with_one_client(self):
1017
for action, value in self.actions.items():
1018
options = self.parser.parse_args()
1019
setattr(options, action, value)
1020
options.client = ["foo"]
942
def test_is_enabled_fails_without_client(self):
943
options = self.parser.parse_args()
944
options.is_enabled = True
945
with self.assertParseError():
1021
946
self.check_option_syntax(options)
1022
947
1023
def test_one_client_with_all_actions_except_is_enabled(self):
948
def test_is_enabled_works_with_one_client(self):
1024
949
options = self.parser.parse_args()
1025
for action, value in self.actions.items():
1026
if action == "is_enabled":
1027
continue
1028
setattr(options, action, value)
950
options.is_enabled = True
1029
951
options.client = ["foo"]
1030
952
self.check_option_syntax(options)
1031
953
1032
def test_two_clients_with_all_actions_except_is_enabled(self):
1033
options = self.parser.parse_args()
1034
for action, value in self.actions.items():
1035
if action == "is_enabled":
1036
continue
1037
setattr(options, action, value)
1038
options.client = ["foo", "barbar"]
1039
self.check_option_syntax(options)
1040
1041
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1042
for action, value in self.actions.items():
1043
if action == "is_enabled":
1044
continue
1045
options = self.parser.parse_args()
1046
setattr(options, action, value)
1047
options.client = ["foo", "barbar"]
1048
self.check_option_syntax(options)
1049
1050
def test_is_enabled_fails_without_client(self):
1051
options = self.parser.parse_args()
1052
options.is_enabled = True
1053
with self.assertParseError():
1054
self.check_option_syntax(options)
1055
1056
954
def test_is_enabled_fails_with_two_clients(self):
1057
955
options = self.parser.parse_args()
1058
956
options.is_enabled = True
1072
970
self.check_option_syntax(options)
1073
971
1074
972
1075
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1076
def test_calls_and_returns_get_object_on_bus(self):
1077
class MockBus(object):
1078
called = False
1079
def get_object(mockbus_self, busname, dbus_path):
1080
# Note that "self" is still the testcase instance,
1081
# this MockBus instance is in "mockbus_self".
1082
self.assertEqual(busname, dbus_busname)
1083
self.assertEqual(dbus_path, server_dbus_path)
1084
mockbus_self.called = True
1085
return mockbus_self
1086
1087
mockbus = get_mandos_dbus_object(bus=MockBus())
1088
self.assertIsInstance(mockbus, MockBus)
1089
self.assertTrue(mockbus.called)
1090
1091
def test_logs_and_exits_on_dbus_error(self):
1092
class MockBusFailing(object):
1093
def get_object(self, busname, dbus_path):
1094
raise dbus.exceptions.DBusException("Test")
1095
1096
with self.assertLogs(log, logging.CRITICAL):
1097
with self.assertRaises(SystemExit) as e:
1098
bus = get_mandos_dbus_object(bus=MockBusFailing())
1099
1100
if isinstance(e.exception.code, int):
1101
self.assertNotEqual(e.exception.code, 0)
1102
else:
1103
self.assertIsNotNone(e.exception.code)
1104
1105
1106
class Test_get_managed_objects(TestCaseWithAssertLogs):
1107
def test_calls_and_returns_GetManagedObjects(self):
1108
managed_objects = {"/clients/foo": { "Name": "foo"}}
1109
class MockObjectManager(object):
1110
def GetManagedObjects(self):
1111
return managed_objects
1112
retval = get_managed_objects(MockObjectManager())
1113
self.assertDictEqual(managed_objects, retval)
1114
1115
def test_logs_and_exits_on_dbus_error(self):
1116
dbus_logger = logging.getLogger("dbus.proxies")
1117
1118
class MockObjectManagerFailing(object):
1119
def GetManagedObjects(self):
1120
dbus_logger.error("Test")
1121
raise dbus.exceptions.DBusException("Test")
1122
1123
class CountingHandler(logging.Handler):
1124
count = 0
1125
def emit(self, record):
1126
self.count += 1
1127
1128
counting_handler = CountingHandler()
1129
1130
dbus_logger.addHandler(counting_handler)
1131
1132
try:
1133
with self.assertLogs(log, logging.CRITICAL) as watcher:
1134
with self.assertRaises(SystemExit) as e:
1135
get_managed_objects(MockObjectManagerFailing())
1136
finally:
1137
dbus_logger.removeFilter(counting_handler)
1138
1139
# Make sure the dbus logger was suppressed
1140
self.assertEqual(counting_handler.count, 0)
1141
1142
# Test that the dbus_logger still works
1143
with self.assertLogs(dbus_logger, logging.ERROR):
1144
dbus_logger.error("Test")
1145
1146
if isinstance(e.exception.code, int):
1147
self.assertNotEqual(e.exception.code, 0)
1148
else:
1149
self.assertIsNotNone(e.exception.code)
1150
1151
1152
class Test_commands_from_options(unittest.TestCase):
973
class Test_command_from_options(unittest.TestCase):
1153
974
def setUp(self):
1154
975
self.parser = argparse.ArgumentParser()
1155
976
add_command_line_options(self.parser)
1156
1157
def test_is_enabled(self):
1158
self.assert_command_from_args(["--is-enabled", "foo"],
1159
command.IsEnabled)
1160
1161
977
def assert_command_from_args(self, args, command_cls,
1162
978
**cmd_attrs):
1163
979
"""Assert that parsing ARGS should result in an instance of
1170
986
self.assertIsInstance(command, command_cls)
1171
987
for key, value in cmd_attrs.items():
1172
988
self.assertEqual(getattr(command, key), value)
1173
1174
def test_is_enabled_short(self):
1175
self.assert_command_from_args(["-V", "foo"],
1176
command.IsEnabled)
1177
1178
def test_approve(self):
1179
self.assert_command_from_args(["--approve", "foo"],
1180
command.Approve)
1181
1182
def test_approve_short(self):
1183
self.assert_command_from_args(["-A", "foo"], command.Approve)
1184
1185
def test_deny(self):
1186
self.assert_command_from_args(["--deny", "foo"], command.Deny)
1187
1188
def test_deny_short(self):
1189
self.assert_command_from_args(["-D", "foo"], command.Deny)
1190
1191
def test_remove(self):
1192
self.assert_command_from_args(["--remove", "foo"],
1193
command.Remove)
1194
1195
def test_deny_before_remove(self):
1196
options = self.parser.parse_args(["--deny", "--remove",
1197
"foo"])
1198
check_option_syntax(self.parser, options)
1199
commands = commands_from_options(options)
1200
self.assertEqual(len(commands), 2)
1201
self.assertIsInstance(commands[0], command.Deny)
1202
self.assertIsInstance(commands[1], command.Remove)
1203
1204
def test_deny_before_remove_reversed(self):
1205
options = self.parser.parse_args(["--remove", "--deny",
1206
"--all"])
1207
check_option_syntax(self.parser, options)
1208
commands = commands_from_options(options)
1209
self.assertEqual(len(commands), 2)
1210
self.assertIsInstance(commands[0], command.Deny)
1211
self.assertIsInstance(commands[1], command.Remove)
1212
1213
def test_remove_short(self):
1214
self.assert_command_from_args(["-r", "foo"], command.Remove)
1215
1216
def test_dump_json(self):
1217
self.assert_command_from_args(["--dump-json"],
1218
command.DumpJSON)
989
def test_print_table(self):
990
self.assert_command_from_args([], PrintTableCmd,
991
verbose=False)
992
993
def test_print_table_verbose(self):
994
self.assert_command_from_args(["--verbose"], PrintTableCmd,
995
verbose=True)
996
997
def test_print_table_verbose_short(self):
998
self.assert_command_from_args(["-v"], PrintTableCmd,
999
verbose=True)
1219
1000
1220
1001
def test_enable(self):
1221
self.assert_command_from_args(["--enable", "foo"],
1222
command.Enable)
1002
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1223
1003
1224
1004
def test_enable_short(self):
1225
self.assert_command_from_args(["-e", "foo"], command.Enable)
1005
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1226
1006
1227
1007
def test_disable(self):
1228
1008
self.assert_command_from_args(["--disable", "foo"],
1229
command.Disable)
1009
DisableCmd)
1230
1010
1231
1011
def test_disable_short(self):
1232
self.assert_command_from_args(["-d", "foo"], command.Disable)
1012
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1233
1013
1234
1014
def test_bump_timeout(self):
1235
1015
self.assert_command_from_args(["--bump-timeout", "foo"],
1236
command.BumpTimeout)
1016
BumpTimeoutCmd)
1237
1017
1238
1018
def test_bump_timeout_short(self):
1239
self.assert_command_from_args(["-b", "foo"],
1240
command.BumpTimeout)
1019
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1241
1020
1242
1021
def test_start_checker(self):
1243
1022
self.assert_command_from_args(["--start-checker", "foo"],
1244
command.StartChecker)
1023
StartCheckerCmd)
1245
1024
1246
1025
def test_stop_checker(self):
1247
1026
self.assert_command_from_args(["--stop-checker", "foo"],
1248
command.StopChecker)
1249
1250
def test_approve_by_default(self):
1251
self.assert_command_from_args(["--approve-by-default", "foo"],
1252
command.ApproveByDefault)
1253
1254
def test_deny_by_default(self):
1255
self.assert_command_from_args(["--deny-by-default", "foo"],
1256
command.DenyByDefault)
1027
StopCheckerCmd)
1028
1029
def test_remove(self):
1030
self.assert_command_from_args(["--remove", "foo"],
1031
RemoveCmd)
1032
1033
def test_remove_short(self):
1034
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1257
1035
1258
1036
def test_checker(self):
1259
1037
self.assert_command_from_args(["--checker", ":", "foo"],
1260
command.SetChecker,
1261
value_to_set=":")
1038
SetCheckerCmd, value_to_set=":")
1262
1039
1263
1040
def test_checker_empty(self):
1264
1041
self.assert_command_from_args(["--checker", "", "foo"],
1265
command.SetChecker,
1266
value_to_set="")
1042
SetCheckerCmd, value_to_set="")
1267
1043
1268
1044
def test_checker_short(self):
1269
1045
self.assert_command_from_args(["-c", ":", "foo"],
1270
command.SetChecker,
1271
value_to_set=":")
1272
1273
def test_host(self):
1274
self.assert_command_from_args(["--host", "foo.example.org",
1275
"foo"], command.SetHost,
1276
value_to_set="foo.example.org")
1277
1278
def test_host_short(self):
1279
self.assert_command_from_args(["-H", "foo.example.org",
1280
"foo"], command.SetHost,
1281
value_to_set="foo.example.org")
1282
1283
def test_secret_devnull(self):
1284
self.assert_command_from_args(["--secret", os.path.devnull,
1285
"foo"], command.SetSecret,
1286
value_to_set=b"")
1287
1288
def test_secret_tempfile(self):
1289
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1290
value = b"secret\0xyzzy\nbar"
1291
f.write(value)
1292
f.seek(0)
1293
self.assert_command_from_args(["--secret", f.name,
1294
"foo"], command.SetSecret,
1295
value_to_set=value)
1296
1297
def test_secret_devnull_short(self):
1298
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1299
command.SetSecret,
1300
value_to_set=b"")
1301
1302
def test_secret_tempfile_short(self):
1303
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1304
value = b"secret\0xyzzy\nbar"
1305
f.write(value)
1306
f.seek(0)
1307
self.assert_command_from_args(["-s", f.name, "foo"],
1308
command.SetSecret,
1309
value_to_set=value)
1046
SetCheckerCmd, value_to_set=":")
1310
1047
1311
1048
def test_timeout(self):
1312
1049
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1313
command.SetTimeout,
1050
SetTimeoutCmd,
1314
1051
value_to_set=300000)
1315
1052
1316
1053
def test_timeout_short(self):
1317
1054
self.assert_command_from_args(["-t", "PT5M", "foo"],
1318
command.SetTimeout,
1055
SetTimeoutCmd,
1319
1056
value_to_set=300000)
1320
1057
1321
1058
def test_extended_timeout(self):
1322
1059
self.assert_command_from_args(["--extended-timeout", "PT15M",
1323
1060
"foo"],
1324
command.SetExtendedTimeout,
1061
SetExtendedTimeoutCmd,
1325
1062
value_to_set=900000)
1326
1063
1327
1064
def test_interval(self):
1328
1065
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1329
command.SetInterval,
1066
SetIntervalCmd,
1330
1067
value_to_set=120000)
1331
1068
1332
1069
def test_interval_short(self):
1333
1070
self.assert_command_from_args(["-i", "PT2M", "foo"],
1334
command.SetInterval,
1071
SetIntervalCmd,
1335
1072
value_to_set=120000)
1336
1073
1074
def test_approve_by_default(self):
1075
self.assert_command_from_args(["--approve-by-default", "foo"],
1076
ApproveByDefaultCmd)
1077
1078
def test_deny_by_default(self):
1079
self.assert_command_from_args(["--deny-by-default", "foo"],
1080
DenyByDefaultCmd)
1081
1337
1082
def test_approval_delay(self):
1338
1083
self.assert_command_from_args(["--approval-delay", "PT30S",
1339
"foo"],
1340
command.SetApprovalDelay,
1084
"foo"], SetApprovalDelayCmd,
1341
1085
value_to_set=30000)
1342
1086
1343
1087
def test_approval_duration(self):
1344
1088
self.assert_command_from_args(["--approval-duration", "PT1S",
1345
"foo"],
1346
command.SetApprovalDuration,
1089
"foo"], SetApprovalDurationCmd,
1347
1090
value_to_set=1000)
1348
1091
1349
def test_print_table(self):
1350
self.assert_command_from_args([], command.PrintTable,
1351
verbose=False)
1352
1353
def test_print_table_verbose(self):
1354
self.assert_command_from_args(["--verbose"],
1355
command.PrintTable,
1356
verbose=True)
1357
1358
def test_print_table_verbose_short(self):
1359
self.assert_command_from_args(["-v"], command.PrintTable,
1360
verbose=True)
1361
1362
1363
class TestCommand(unittest.TestCase):
1092
def test_host(self):
1093
self.assert_command_from_args(["--host", "foo.example.org",
1094
"foo"], SetHostCmd,
1095
value_to_set="foo.example.org")
1096
1097
def test_host_short(self):
1098
self.assert_command_from_args(["-H", "foo.example.org",
1099
"foo"], SetHostCmd,
1100
value_to_set="foo.example.org")
1101
1102
def test_secret_devnull(self):
1103
self.assert_command_from_args(["--secret", os.path.devnull,
1104
"foo"], SetSecretCmd,
1105
value_to_set=b"")
1106
1107
def test_secret_tempfile(self):
1108
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1109
value = b"secret\0xyzzy\nbar"
1110
f.write(value)
1111
f.seek(0)
1112
self.assert_command_from_args(["--secret", f.name,
1113
"foo"], SetSecretCmd,
1114
value_to_set=value)
1115
1116
def test_secret_devnull_short(self):
1117
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1118
SetSecretCmd, value_to_set=b"")
1119
1120
def test_secret_tempfile_short(self):
1121
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1122
value = b"secret\0xyzzy\nbar"
1123
f.write(value)
1124
f.seek(0)
1125
self.assert_command_from_args(["-s", f.name, "foo"],
1126
SetSecretCmd,
1127
value_to_set=value)
1128
1129
def test_approve(self):
1130
self.assert_command_from_args(["--approve", "foo"],
1131
ApproveCmd)
1132
1133
def test_approve_short(self):
1134
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1135
1136
def test_deny(self):
1137
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1138
1139
def test_deny_short(self):
1140
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1141
1142
def test_dump_json(self):
1143
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1144
1145
def test_is_enabled(self):
1146
self.assert_command_from_args(["--is-enabled", "foo"],
1147
IsEnabledCmd)
1148
1149
def test_is_enabled_short(self):
1150
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1151
1152
def test_deny_before_remove(self):
1153
options = self.parser.parse_args(["--deny", "--remove",
1154
"foo"])
1155
check_option_syntax(self.parser, options)
1156
commands = commands_from_options(options)
1157
self.assertEqual(len(commands), 2)
1158
self.assertIsInstance(commands[0], DenyCmd)
1159
self.assertIsInstance(commands[1], RemoveCmd)
1160
1161
def test_deny_before_remove_reversed(self):
1162
options = self.parser.parse_args(["--remove", "--deny",
1163
"--all"])
1164
check_option_syntax(self.parser, options)
1165
commands = commands_from_options(options)
1166
self.assertEqual(len(commands), 2)
1167
self.assertIsInstance(commands[0], DenyCmd)
1168
self.assertIsInstance(commands[1], RemoveCmd)
1169
1170
1171
class TestCmd(unittest.TestCase):
1364
1172
"""Abstract class for tests of command classes"""
1365
1366
1173
def setUp(self):
1367
1174
testcase = self
1368
1175
class MockClient(object):
1429
1236
ApprovedByDefault=dbus.Boolean(False),
1430
1237
LastApprovalRequest="2019-01-03T00:00:00",
1431
1238
ApprovalDelay=30000,
1432
ApprovalDuration=93785000,
1239
ApprovalDuration=1000,
1433
1240
Checker=":",
1434
1241
ExtendedTimeout=900000,
1435
1242
Expires="2019-02-05T00:00:00",
1440
1247
("/clients/barbar", self.other_client.attributes),
1441
1248
])
1442
1249
self.one_client = {"/clients/foo": self.client.attributes}
1443
1444
1250
@property
1445
1251
def bus(self):
1446
1252
class Bus(object):
1448
1254
def get_object(client_bus_name, path):
1449
1255
self.assertEqual(client_bus_name, dbus_busname)
1450
1256
return {
1451
# Note: "self" here is the TestCmd instance, not
1452
# the Bus instance, since this is a static method!
1453
1257
"/clients/foo": self.client,
1454
1258
"/clients/barbar": self.other_client,
1455
1259
}[path]
1456
1260
return Bus()
1457
1261
1458
1262
1459
class TestBaseCommands(TestCommand):
1460
1263
class TestIsEnabledCmd(TestCmd):
1461
1264
def test_is_enabled(self):
1462
self.assertTrue(all(command.IsEnabled().is_enabled(client,
1265
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1463
1266
properties)
1464
1267
for client, properties
1465
1268
in self.clients.items()))
1466
1467
1269
def test_is_enabled_run_exits_successfully(self):
1468
1270
with self.assertRaises(SystemExit) as e:
1469
command.IsEnabled().run(self.one_client)
1271
IsEnabledCmd().run(self.one_client)
1470
1272
if e.exception.code is not None:
1471
1273
self.assertEqual(e.exception.code, 0)
1472
1274
else:
1473
1275
self.assertIsNone(e.exception.code)
1474
1475
1276
def test_is_enabled_run_exits_with_failure(self):
1476
1277
self.client.attributes["Enabled"] = dbus.Boolean(False)
1477
1278
with self.assertRaises(SystemExit) as e:
1478
command.IsEnabled().run(self.one_client)
1279
IsEnabledCmd().run(self.one_client)
1479
1280
if isinstance(e.exception.code, int):
1480
1281
self.assertNotEqual(e.exception.code, 0)
1481
1282
else:
1482
1283
self.assertIsNotNone(e.exception.code)
1483
1284
1285
1286
class TestApproveCmd(TestCmd):
1484
1287
def test_approve(self):
1485
command.Approve().run(self.clients, self.bus)
1288
ApproveCmd().run(self.clients, self.bus)
1486
1289
for clientpath in self.clients:
1487
1290
client = self.bus.get_object(dbus_busname, clientpath)
1488
1291
self.assertIn(("Approve", (True, client_dbus_interface)),
1489
1292
client.calls)
1490
1293
1294
1295
class TestDenyCmd(TestCmd):
1491
1296
def test_deny(self):
1492
command.Deny().run(self.clients, self.bus)
1297
DenyCmd().run(self.clients, self.bus)
1493
1298
for clientpath in self.clients:
1494
1299
client = self.bus.get_object(dbus_busname, clientpath)
1495
1300
self.assertIn(("Approve", (False, client_dbus_interface)),
1496
1301
client.calls)
1497
1302
1303
class TestRemoveCmd(TestCmd):
1498
1304
def test_remove(self):
1499
1305
class MockMandos(object):
1500
1306
def __init__(self):
1502
1308
def RemoveClient(self, dbus_path):
1503
1309
self.calls.append(("RemoveClient", (dbus_path,)))
1504
1310
mandos = MockMandos()
1505
super(TestBaseCommands, self).setUp()
1506
command.Remove().run(self.clients, self.bus, mandos)
1311
super(TestRemoveCmd, self).setUp()
1312
RemoveCmd().run(self.clients, self.bus, mandos)
1507
1313
self.assertEqual(len(mandos.calls), 2)
1508
1314
for clientpath in self.clients:
1509
1315
self.assertIn(("RemoveClient", (clientpath,)),
1510
1316
mandos.calls)
1511
1317
1512
expected_json = {
1513
"foo": {
1514
"Name": "foo",
1515
"KeyID": ("92ed150794387c03ce684574b1139a65"
1516
"94a34f895daaaf09fd8ea90a27cddb12"),
1517
"Host": "foo.example.org",
1518
"Enabled": True,
1519
"Timeout": 300000,
1520
"LastCheckedOK": "2019-02-03T00:00:00",
1521
"Created": "2019-01-02T00:00:00",
1522
"Interval": 120000,
1523
"Fingerprint": ("778827225BA7DE539C5A"
1524
"7CFA59CFF7CDBD9A5920"),
1525
"CheckerRunning": False,
1526
"LastEnabled": "2019-01-03T00:00:00",
1527
"ApprovalPending": False,
1528
"ApprovedByDefault": True,
1529
"LastApprovalRequest": "",
1530
"ApprovalDelay": 0,
1531
"ApprovalDuration": 1000,
1532
"Checker": "fping -q -- %(host)s",
1533
"ExtendedTimeout": 900000,
1534
"Expires": "2019-02-04T00:00:00",
1535
"LastCheckerStatus": 0,
1536
},
1537
"barbar": {
1538
"Name": "barbar",
1539
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1540
"6ab612cff5ad227247e46c2b020f441c"),
1541
"Host": "192.0.2.3",
1542
"Enabled": True,
1543
"Timeout": 300000,
1544
"LastCheckedOK": "2019-02-04T00:00:00",
1545
"Created": "2019-01-03T00:00:00",
1546
"Interval": 120000,
1547
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1548
"F547B3A107558FCA3A27"),
1549
"CheckerRunning": True,
1550
"LastEnabled": "2019-01-04T00:00:00",
1551
"ApprovalPending": False,
1552
"ApprovedByDefault": False,
1553
"LastApprovalRequest": "2019-01-03T00:00:00",
1554
"ApprovalDelay": 30000,
1555
"ApprovalDuration": 93785000,
1556
"Checker": ":",
1557
"ExtendedTimeout": 900000,
1558
"Expires": "2019-02-05T00:00:00",
1559
"LastCheckerStatus": -2,
1560
},
1561
}
1562
1318
1563
def test_DumpJSON_normal(self):
1564
output = command.DumpJSON().output(self.clients.values())
1565
json_data = json.loads(output)
1319
class TestDumpJSONCmd(TestCmd):
1320
def setUp(self):
1321
self.expected_json = {
1322
"foo": {
1323
"Name": "foo",
1324
"KeyID": ("92ed150794387c03ce684574b1139a65"
1325
"94a34f895daaaf09fd8ea90a27cddb12"),
1326
"Host": "foo.example.org",
1327
"Enabled": True,
1328
"Timeout": 300000,
1329
"LastCheckedOK": "2019-02-03T00:00:00",
1330
"Created": "2019-01-02T00:00:00",
1331
"Interval": 120000,
1332
"Fingerprint": ("778827225BA7DE539C5A"
1333
"7CFA59CFF7CDBD9A5920"),
1334
"CheckerRunning": False,
1335
"LastEnabled": "2019-01-03T00:00:00",
1336
"ApprovalPending": False,
1337
"ApprovedByDefault": True,
1338
"LastApprovalRequest": "",
1339
"ApprovalDelay": 0,
1340
"ApprovalDuration": 1000,
1341
"Checker": "fping -q -- %(host)s",
1342
"ExtendedTimeout": 900000,
1343
"Expires": "2019-02-04T00:00:00",
1344
"LastCheckerStatus": 0,
1345
},
1346
"barbar": {
1347
"Name": "barbar",
1348
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1349
"6ab612cff5ad227247e46c2b020f441c"),
1350
"Host": "192.0.2.3",
1351
"Enabled": True,
1352
"Timeout": 300000,
1353
"LastCheckedOK": "2019-02-04T00:00:00",
1354
"Created": "2019-01-03T00:00:00",
1355
"Interval": 120000,
1356
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1357
"F547B3A107558FCA3A27"),
1358
"CheckerRunning": True,
1359
"LastEnabled": "2019-01-04T00:00:00",
1360
"ApprovalPending": False,
1361
"ApprovedByDefault": False,
1362
"LastApprovalRequest": "2019-01-03T00:00:00",
1363
"ApprovalDelay": 30000,
1364
"ApprovalDuration": 1000,
1365
"Checker": ":",
1366
"ExtendedTimeout": 900000,
1367
"Expires": "2019-02-05T00:00:00",
1368
"LastCheckerStatus": -2,
1369
},
1370
}
1371
return super(TestDumpJSONCmd, self).setUp()
1372
def test_normal(self):
1373
json_data = json.loads(DumpJSONCmd().output(self.clients))
1566
1374
self.assertDictEqual(json_data, self.expected_json)
1567
1568
def test_DumpJSON_one_client(self):
1569
output = command.DumpJSON().output(self.one_client.values())
1570
json_data = json.loads(output)
1375
def test_one_client(self):
1376
clients = self.one_client
1377
json_data = json.loads(DumpJSONCmd().output(clients))
1571
1378
expected_json = {"foo": self.expected_json["foo"]}
1572
1379
self.assertDictEqual(json_data, expected_json)
1573
1380
1574
def test_PrintTable_normal(self):
1575
output = command.PrintTable().output(self.clients.values())
1381
1382
class TestPrintTableCmd(TestCmd):
1383
def test_normal(self):
1384
output = PrintTableCmd().output(self.clients.values())
1576
1385
expected_output = "\n".join((
1577
1386
"Name Enabled Timeout Last Successful Check",
1578
1387
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1579
1388
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1580
1389
))
1581
1390
self.assertEqual(output, expected_output)
1582
1583
def test_PrintTable_verbose(self):
1584
output = command.PrintTable(verbose=True).output(
1391
def test_verbose(self):
1392
output = PrintTableCmd(verbose=True).output(
1585
1393
self.clients.values())
1586
1394
columns = (
1587
1395
(
1650
1458
),(
1651
1459
"Approval Duration ",
1652
1460
"00:00:01 ",
1653
"1T02:03:05 ",
1461
"00:00:01 ",
1654
1462
),(
1655
1463
"Checker ",
1656
1464
"fping -q -- %(host)s ",
1674
1482
for rows in columns)
1675
1483
for line in range(num_lines))
1676
1484
self.assertEqual(output, expected_output)
1677
1678
def test_PrintTable_one_client(self):
1679
output = command.PrintTable().output(self.one_client.values())
1680
expected_output = "\n".join((
1681
"Name Enabled Timeout Last Successful Check",
1682
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1683
))
1485
def test_one_client(self):
1486
output = PrintTableCmd().output(self.one_client.values())
1487
expected_output = """
1488
Name Enabled Timeout Last Successful Check
1489
foo Yes 00:05:00 2019-02-03T00:00:00
1490
"""[1:-1]
1684
1491
self.assertEqual(output, expected_output)
1685
1492
1686
1493
1687
class TestPropertyCmd(TestCommand):
1688
"""Abstract class for tests of command.Property classes"""
1494
class Test_milliseconds_to_string(unittest.TestCase):
1495
def test_all(self):
1496
self.assertEqual(milliseconds_to_string(93785000),
1497
"1T02:03:05")
1498
def test_no_days(self):
1499
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
1500
def test_all_zero(self):
1501
self.assertEqual(milliseconds_to_string(0), "00:00:00")
1502
def test_no_fractional_seconds(self):
1503
self.assertEqual(milliseconds_to_string(400), "00:00:00")
1504
self.assertEqual(milliseconds_to_string(900), "00:00:00")
1505
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
1506
1507
1508
class Unique(object):
1509
"""Class for objects which exist only to be unique objects, since
1510
unittest.mock.sentinel only exists in Python 3.3"""
1511
1512
1513
class TestPropertyCmd(TestCmd):
1514
"""Abstract class for tests of PropertyCmd classes"""
1689
1515
def runTest(self):
1690
1516
if not hasattr(self, "command"):
1691
1517
return
1696
1522
for clientpath in self.clients:
1697
1523
client = self.bus.get_object(dbus_busname, clientpath)
1698
1524
old_value = client.attributes[self.propname]
1699
self.assertNotIsInstance(old_value, self.Unique)
1700
client.attributes[self.propname] = self.Unique()
1525
self.assertNotIsInstance(old_value, Unique)
1526
client.attributes[self.propname] = Unique()
1701
1527
self.run_command(value_to_set, self.clients)
1702
1528
for clientpath in self.clients:
1703
1529
client = self.bus.get_object(dbus_busname, clientpath)
1704
1530
value = client.attributes[self.propname]
1705
self.assertNotIsInstance(value, self.Unique)
1531
self.assertNotIsInstance(value, Unique)
1706
1532
self.assertEqual(value, value_to_get)
1707
1708
class Unique(object):
1709
"""Class for objects which exist only to be unique objects,
1710
since unittest.mock.sentinel only exists in Python 3.3"""
1711
1712
1533
def run_command(self, value, clients):
1713
1534
self.command().run(clients, self.bus)
1714
1535
1715
1536
1716
class TestEnableCmd(TestPropertyCmd):
1717
command = command.Enable
1718
propname = "Enabled"
1719
values_to_set = [dbus.Boolean(True)]
1720
1721
1722
class TestDisableCmd(TestPropertyCmd):
1723
command = command.Disable
1724
propname = "Enabled"
1725
values_to_set = [dbus.Boolean(False)]
1537
class TestEnableCmd(TestCmd):
1538
def test_enable(self):
1539
for clientpath in self.clients:
1540
client = self.bus.get_object(dbus_busname, clientpath)
1541
client.attributes["Enabled"] = False
1542
1543
EnableCmd().run(self.clients, self.bus)
1544
1545
for clientpath in self.clients:
1546
client = self.bus.get_object(dbus_busname, clientpath)
1547
self.assertTrue(client.attributes["Enabled"])
1548
1549
1550
class TestDisableCmd(TestCmd):
1551
def test_disable(self):
1552
DisableCmd().run(self.clients, self.bus)
1553
for clientpath in self.clients:
1554
client = self.bus.get_object(dbus_busname, clientpath)
1555
self.assertFalse(client.attributes["Enabled"])
1726
1556
1727
1557
1728
1558
class TestBumpTimeoutCmd(TestPropertyCmd):
1729
command = command.BumpTimeout
1559
command = BumpTimeoutCmd
1730
1560
propname = "LastCheckedOK"
1731
1561
values_to_set = [""]
1732
1562
1733
1563
1734
1564
class TestStartCheckerCmd(TestPropertyCmd):
1735
command = command.StartChecker
1565
command = StartCheckerCmd
1736
1566
propname = "CheckerRunning"
1737
1567
values_to_set = [dbus.Boolean(True)]
1738
1568
1739
1569
1740
1570
class TestStopCheckerCmd(TestPropertyCmd):
1741
command = command.StopChecker
1571
command = StopCheckerCmd
1742
1572
propname = "CheckerRunning"
1743
1573
values_to_set = [dbus.Boolean(False)]
1744
1574
1745
1575
1746
1576
class TestApproveByDefaultCmd(TestPropertyCmd):
1747
command = command.ApproveByDefault
1577
command = ApproveByDefaultCmd
1748
1578
propname = "ApprovedByDefault"
1749
1579
values_to_set = [dbus.Boolean(True)]
1750
1580
1751
1581
1752
1582
class TestDenyByDefaultCmd(TestPropertyCmd):
1753
command = command.DenyByDefault
1583
command = DenyByDefaultCmd
1754
1584
propname = "ApprovedByDefault"
1755
1585
values_to_set = [dbus.Boolean(False)]
1756
1586
1757
1587
1758
1588
class TestPropertyValueCmd(TestPropertyCmd):
1759
1589
"""Abstract class for tests of PropertyValueCmd classes"""
1760
1761
1590
def runTest(self):
1762
1591
if type(self) is TestPropertyValueCmd:
1763
1592
return
1764
1593
return super(TestPropertyValueCmd, self).runTest()
1765
1766
1594
def run_command(self, value, clients):
1767
1595
self.command(value).run(clients, self.bus)
1768
1596
1769
1597
1770
1598
class TestSetCheckerCmd(TestPropertyValueCmd):
1771
command = command.SetChecker
1599
command = SetCheckerCmd
1772
1600
propname = "Checker"
1773
1601
values_to_set = ["", ":", "fping -q -- %s"]
1774
1602
1775
1603
1776
1604
class TestSetHostCmd(TestPropertyValueCmd):
1777
command = command.SetHost
1605
command = SetHostCmd
1778
1606
propname = "Host"
1779
1607
values_to_set = ["192.0.2.3", "foo.example.org"]
1780
1608
1781
1609
1782
1610
class TestSetSecretCmd(TestPropertyValueCmd):
1783
command = command.SetSecret
1611
command = SetSecretCmd
1784
1612
propname = "Secret"
1785
1613
values_to_set = [io.BytesIO(b""),
1786
1614
io.BytesIO(b"secret\0xyzzy\nbar")]
1788
1616
1789
1617
1790
1618
class TestSetTimeoutCmd(TestPropertyValueCmd):
1791
command = command.SetTimeout
1619
command = SetTimeoutCmd
1792
1620
propname = "Timeout"
1793
1621
values_to_set = [datetime.timedelta(),
1794
1622
datetime.timedelta(minutes=5),
1799
1627
1800
1628
1801
1629
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1802
command = command.SetExtendedTimeout
1630
command = SetExtendedTimeoutCmd
1803
1631
propname = "ExtendedTimeout"
1804
1632
values_to_set = [datetime.timedelta(),
1805
1633
datetime.timedelta(minutes=5),
1810
1638
1811
1639
1812
1640
class TestSetIntervalCmd(TestPropertyValueCmd):
1813
command = command.SetInterval
1641
command = SetIntervalCmd
1814
1642
propname = "Interval"
1815
1643
values_to_set = [datetime.timedelta(),
1816
1644
datetime.timedelta(minutes=5),
1821
1649
1822
1650
1823
1651
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1824
command = command.SetApprovalDelay
1652
command = SetApprovalDelayCmd
1825
1653
propname = "ApprovalDelay"
1826
1654
values_to_set = [datetime.timedelta(),
1827
1655
datetime.timedelta(minutes=5),
1832
1660
1833
1661
1834
1662
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1835
command = command.SetApprovalDuration
1663
command = SetApprovalDurationCmd
1836
1664
propname = "ApprovalDuration"
1837
1665
values_to_set = [datetime.timedelta(),
1838
1666
datetime.timedelta(minutes=5),
1860
1688
return tests
1861
1689
1862
1690
if __name__ == "__main__":
1863
try:
1864
if should_only_run_tests():
1865
# Call using ./tdd-python-script --check [--verbose]
1866
unittest.main()
1867
else:
1868
main()
1869
finally:
1870
logging.shutdown()
1691
if should_only_run_tests():
1692
# Call using ./tdd-python-script --check [--verbose]
1693
unittest.main()
1694
else:
1695
main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0