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.635
- compare with another revision
- download diff
- download tarball
- view history from revision 237.7.635
- Committer: Teddy Hogeborn
- Date: 2019-03-21 18:37:29 UTC
- mto: This revision was merged to the branch mainline in revision 382.
- Revision ID: teddy@recompile.se-20190321183729-6okcpelqgalk1p6k
mandos-ctl: Refactor
* mandos-ctl (main): Remove a couple of blank lines.
(SilenceLogger.__enter__): Remove unnecessary return statement.
(commands.Property): Rename to "PropertySetter".
(commands.PropertyValue): Rename to "PropertySetterValue".
(commands.MillisecondsPropertyValueArgument): Rename to
"PropertySetterValueMilliseconds".
(TestPropertyCmd): Rename to "TestPropertySetterCmd".
(TestPropertyValueCmd): Rename to "TestPropertySetterValueCmd".
* mandos-ctl (main): Remove a couple of blank lines.
(SilenceLogger.__enter__): Remove unnecessary return statement.
(commands.Property): Rename to "PropertySetter".
(commands.PropertyValue): Rename to "PropertySetterValue".
(commands.MillisecondsPropertyValueArgument): Rename to
"PropertySetterValueMilliseconds".
(TestPropertyCmd): Rename to "TestPropertySetterCmd".
(TestPropertyValueCmd): Rename to "TestPropertySetterValueCmd".
- files modified:
- intro.xml
added
removed
mandos-ctl
46
46
import tempfile
47
47
import contextlib
48
48
49
import dbus as dbus_python
49
import dbus
50
50
51
51
# Show warnings by default
52
52
if not sys.warnoptions:
66
66
67
67
locale.setlocale(locale.LC_ALL, "")
68
68
69
dbus_busname_domain = "se.recompile"
70
dbus_busname = dbus_busname_domain + ".Mandos"
71
server_dbus_path = "/"
72
server_dbus_interface = dbus_busname_domain + ".Mandos"
73
client_dbus_interface = dbus_busname_domain + ".Mandos.Client"
74
del dbus_busname_domain
69
75
version = "1.8.3"
70
76
71
77
78
try:
79
dbus.OBJECT_MANAGER_IFACE
80
except AttributeError:
81
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
82
83
72
84
def main():
73
85
parser = argparse.ArgumentParser()
74
86
add_command_line_options(parser)
81
93
if options.debug:
82
94
log.setLevel(logging.DEBUG)
83
95
84
bus = dbus_python_adapter.CachingBus(dbus_python)
85
86
try:
87
all_clients = bus.get_clients_and_properties()
88
except dbus.ConnectFailed as e:
89
log.critical("Could not connect to Mandos server: %s", e)
90
sys.exit(1)
91
except dbus.Error as e:
92
log.critical(
93
"Failed to access Mandos server through D-Bus:\n%s", e)
94
sys.exit(1)
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)
102
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
95
113
96
114
# Compile dict of (clientpath: properties) to process
97
115
if not clientnames:
110
128
commands = commands_from_options(options)
111
129
112
130
for command in commands:
113
command.run(clients, bus)
131
command.run(clients, bus, mandos_serv)
114
132
115
133
116
134
def add_command_line_options(parser):
121
139
help="Select all clients")
122
140
parser.add_argument("-v", "--verbose", action="store_true",
123
141
help="Print all fields")
124
parser.add_argument("-j", "--dump-json", dest="commands",
125
action="append_const", default=[],
126
const=command.DumpJSON(),
142
parser.add_argument("-j", "--dump-json", action="store_true",
127
143
help="Dump client data in JSON format")
128
144
enable_disable = parser.add_mutually_exclusive_group()
129
enable_disable.add_argument("-e", "--enable", dest="commands",
130
action="append_const", default=[],
131
const=command.Enable(),
145
enable_disable.add_argument("-e", "--enable", action="store_true",
132
146
help="Enable client")
133
enable_disable.add_argument("-d", "--disable", dest="commands",
134
action="append_const", default=[],
135
const=command.Disable(),
147
enable_disable.add_argument("-d", "--disable",
148
action="store_true",
136
149
help="disable client")
137
parser.add_argument("-b", "--bump-timeout", dest="commands",
138
action="append_const", default=[],
139
const=command.BumpTimeout(),
150
parser.add_argument("-b", "--bump-timeout", action="store_true",
140
151
help="Bump timeout for client")
141
152
start_stop_checker = parser.add_mutually_exclusive_group()
142
153
start_stop_checker.add_argument("--start-checker",
143
dest="commands",
144
action="append_const", default=[],
145
const=command.StartChecker(),
154
action="store_true",
146
155
help="Start checker for client")
147
start_stop_checker.add_argument("--stop-checker", dest="commands",
148
action="append_const", default=[],
149
const=command.StopChecker(),
156
start_stop_checker.add_argument("--stop-checker",
157
action="store_true",
150
158
help="Stop checker for client")
151
parser.add_argument("-V", "--is-enabled", dest="commands",
152
action="append_const", default=[],
153
const=command.IsEnabled(),
159
parser.add_argument("-V", "--is-enabled", action="store_true",
154
160
help="Check if client is enabled")
155
parser.add_argument("-r", "--remove", dest="commands",
156
action="append_const", default=[],
157
const=command.Remove(),
161
parser.add_argument("-r", "--remove", action="store_true",
158
162
help="Remove client")
159
parser.add_argument("-c", "--checker", dest="commands",
160
action="append", default=[],
161
metavar="COMMAND", type=command.SetChecker,
163
parser.add_argument("-c", "--checker",
162
164
help="Set checker command for client")
163
parser.add_argument(
164
"-t", "--timeout", dest="commands", action="append",
165
default=[], metavar="TIME",
166
type=command.SetTimeout.argparse(string_to_delta),
167
help="Set timeout for client")
168
parser.add_argument(
169
"--extended-timeout", dest="commands", action="append",
170
default=[], metavar="TIME",
171
type=command.SetExtendedTimeout.argparse(string_to_delta),
172
help="Set extended timeout for client")
173
parser.add_argument(
174
"-i", "--interval", dest="commands", action="append",
175
default=[], metavar="TIME",
176
type=command.SetInterval.argparse(string_to_delta),
177
help="Set checker interval for client")
165
parser.add_argument("-t", "--timeout", type=string_to_delta,
166
help="Set timeout for client")
167
parser.add_argument("--extended-timeout", type=string_to_delta,
168
help="Set extended timeout for client")
169
parser.add_argument("-i", "--interval", type=string_to_delta,
170
help="Set checker interval for client")
178
171
approve_deny_default = parser.add_mutually_exclusive_group()
179
172
approve_deny_default.add_argument(
180
"--approve-by-default", dest="commands",
181
action="append_const", default=[],
182
const=command.ApproveByDefault(),
173
"--approve-by-default", action="store_true",
174
default=None, dest="approved_by_default",
183
175
help="Set client to be approved by default")
184
176
approve_deny_default.add_argument(
185
"--deny-by-default", dest="commands",
186
action="append_const", default=[],
187
const=command.DenyByDefault(),
177
"--deny-by-default", action="store_false",
178
dest="approved_by_default",
188
179
help="Set client to be denied by default")
189
parser.add_argument(
190
"--approval-delay", dest="commands", action="append",
191
default=[], metavar="TIME",
192
type=command.SetApprovalDelay.argparse(string_to_delta),
193
help="Set delay before client approve/deny")
194
parser.add_argument(
195
"--approval-duration", dest="commands", action="append",
196
default=[], metavar="TIME",
197
type=command.SetApprovalDuration.argparse(string_to_delta),
198
help="Set duration of one client approval")
199
parser.add_argument("-H", "--host", dest="commands",
200
action="append", default=[], metavar="STRING",
201
type=command.SetHost,
202
help="Set host for client")
203
parser.add_argument(
204
"-s", "--secret", dest="commands", action="append",
205
default=[], metavar="FILENAME",
206
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
207
help="Set password blob (file) for client")
180
parser.add_argument("--approval-delay", type=string_to_delta,
181
help="Set delay before client approve/deny")
182
parser.add_argument("--approval-duration", type=string_to_delta,
183
help="Set duration of one client approval")
184
parser.add_argument("-H", "--host", help="Set host for client")
185
parser.add_argument("-s", "--secret",
186
type=argparse.FileType(mode="rb"),
187
help="Set password blob (file) for client")
208
188
approve_deny = parser.add_mutually_exclusive_group()
209
189
approve_deny.add_argument(
210
"-A", "--approve", dest="commands", action="append_const",
211
default=[], const=command.Approve(),
190
"-A", "--approve", action="store_true",
212
191
help="Approve any current client request")
213
approve_deny.add_argument("-D", "--deny", dest="commands",
214
action="append_const", default=[],
215
const=command.Deny(),
192
approve_deny.add_argument("-D", "--deny", action="store_true",
216
193
help="Deny any current client request")
217
194
parser.add_argument("--debug", action="store_true",
218
195
help="Debug mode (show D-Bus commands)")
409
386
"""Apply additional restrictions on options, not expressible in
410
387
argparse"""
411
388
412
def has_commands(options, commands=None):
413
if commands is None:
414
commands = (command.Enable,
415
command.Disable,
416
command.BumpTimeout,
417
command.StartChecker,
418
command.StopChecker,
419
command.IsEnabled,
420
command.Remove,
421
command.SetChecker,
422
command.SetTimeout,
423
command.SetExtendedTimeout,
424
command.SetInterval,
425
command.ApproveByDefault,
426
command.DenyByDefault,
427
command.SetApprovalDelay,
428
command.SetApprovalDuration,
429
command.SetHost,
430
command.SetSecret,
431
command.Approve,
432
command.Deny)
433
return any(isinstance(cmd, commands)
434
for cmd in options.commands)
389
def has_actions(options):
390
return any((options.enable,
391
options.disable,
392
options.bump_timeout,
393
options.start_checker,
394
options.stop_checker,
395
options.is_enabled,
396
options.remove,
397
options.checker is not None,
398
options.timeout is not None,
399
options.extended_timeout is not None,
400
options.interval is not None,
401
options.approved_by_default is not None,
402
options.approval_delay is not None,
403
options.approval_duration is not None,
404
options.host is not None,
405
options.secret is not None,
406
options.approve,
407
options.deny))
435
408
436
if has_commands(options) and not (options.client or options.all):
409
if has_actions(options) and not (options.client or options.all):
437
410
parser.error("Options require clients names or --all.")
438
if options.verbose and has_commands(options):
411
if options.verbose and has_actions(options):
439
412
parser.error("--verbose can only be used alone.")
440
if (has_commands(options, (command.DumpJSON,))
441
and (options.verbose or len(options.commands) > 1)):
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
442
415
parser.error("--dump-json can only be used alone.")
443
if options.all and not has_commands(options):
416
if options.all and not has_actions(options):
444
417
parser.error("--all requires an action.")
445
if (has_commands(options, (command.IsEnabled,))
446
and len(options.client) > 1):
418
if options.is_enabled and len(options.client) > 1:
447
419
parser.error("--is-enabled requires exactly one client")
448
if (len(options.commands) > 1
449
and has_commands(options, (command.Remove,))
450
and not has_commands(options, (command.Deny,))):
451
parser.error("--remove can only be combined with --deny")
452
453
454
class dbus(object):
455
456
class SystemBus(object):
457
458
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
459
def get_managed_objects(self, busname, objectpath):
460
return self.call_method("GetManagedObjects", busname,
461
objectpath,
462
self.object_manager_iface)
463
464
properties_iface = "org.freedesktop.DBus.Properties"
465
def set_property(self, busname, objectpath, interface, key,
466
value):
467
self.call_method("Set", busname, objectpath,
468
self.properties_iface, interface, key,
469
value)
470
471
472
class MandosBus(SystemBus):
473
busname_domain = "se.recompile"
474
busname = busname_domain + ".Mandos"
475
server_path = "/"
476
server_interface = busname_domain + ".Mandos"
477
client_interface = busname_domain + ".Mandos.Client"
478
del busname_domain
479
480
def get_clients_and_properties(self):
481
managed_objects = self.get_managed_objects(
482
self.busname, self.server_path)
483
return {objpath: properties[self.client_interface]
484
for objpath, properties in managed_objects.items()
485
if self.client_interface in properties}
486
487
def set_client_property(self, objectpath, key, value):
488
return self.set_property(self.busname, objectpath,
489
self.client_interface, key,
490
value)
491
492
def call_client_method(self, objectpath, method, *args):
493
return self.call_method(method, self.busname, objectpath,
494
self.client_interface, *args)
495
496
def call_server_method(self, method, *args):
497
return self.call_method(method, self.busname,
498
self.server_path,
499
self.server_interface, *args)
500
501
class Error(Exception):
502
pass
503
504
class ConnectFailed(Error):
505
pass
506
507
508
class dbus_python_adapter(object):
509
510
class SystemBus(dbus.MandosBus):
511
"""Use dbus-python"""
512
513
def __init__(self, module=dbus_python):
514
self.dbus_python = module
515
self.bus = self.dbus_python.SystemBus()
516
517
@contextlib.contextmanager
518
def convert_exception(self, exception_class=dbus.Error):
519
try:
520
yield
521
except self.dbus_python.exceptions.DBusException as e:
522
# This does what "raise from" would do
523
exc = exception_class(*e.args)
524
exc.__cause__ = e
525
raise exc
526
527
def call_method(self, methodname, busname, objectpath,
528
interface, *args):
529
proxy_object = self.get_object(busname, objectpath)
530
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
531
interface, methodname,
532
", ".join(repr(a) for a in args))
533
method = getattr(proxy_object, methodname)
534
with self.convert_exception():
535
with dbus_python_adapter.SilenceLogger(
536
"dbus.proxies"):
537
value = method(*args, dbus_interface=interface)
538
return self.type_filter(value)
539
540
def get_object(self, busname, objectpath):
541
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
542
busname, objectpath)
543
with self.convert_exception(dbus.ConnectFailed):
544
return self.bus.get_object(busname, objectpath)
545
546
def type_filter(self, value):
547
"""Convert the most bothersome types to Python types"""
548
if isinstance(value, self.dbus_python.Boolean):
549
return bool(value)
550
if isinstance(value, self.dbus_python.ObjectPath):
551
return str(value)
552
# Also recurse into dictionaries
553
if isinstance(value, self.dbus_python.Dictionary):
554
return {self.type_filter(key):
555
self.type_filter(subval)
556
for key, subval in value.items()}
557
return value
558
559
560
class SilenceLogger(object):
561
"Simple context manager to silence a particular logger"
562
def __init__(self, loggername):
563
self.logger = logging.getLogger(loggername)
564
565
def __enter__(self):
566
self.logger.addFilter(self.nullfilter)
567
568
class NullFilter(logging.Filter):
569
def filter(self, record):
570
return False
571
572
nullfilter = NullFilter()
573
574
def __exit__(self, exc_type, exc_val, exc_tb):
575
self.logger.removeFilter(self.nullfilter)
576
577
578
class CachingBus(SystemBus):
579
"""A caching layer for dbus_python_adapter.SystemBus"""
580
def __init__(self, *args, **kwargs):
581
self.object_cache = {}
582
super(dbus_python_adapter.CachingBus,
583
self).__init__(*args, **kwargs)
584
def get_object(self, busname, objectpath):
585
try:
586
return self.object_cache[(busname, objectpath)]
587
except KeyError:
588
new_object = super(
589
dbus_python_adapter.CachingBus,
590
self).get_object(busname, objectpath)
591
self.object_cache[(busname, objectpath)] = new_object
592
return new_object
420
if options.remove:
421
options.remove = False
422
if has_actions(options) and not options.deny:
423
parser.error("--remove can only be combined with --deny")
424
options.remove = True
425
426
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
464
class NullFilter(logging.Filter):
465
def filter(self, record):
466
return False
467
468
nullfilter = NullFilter()
469
470
def __exit__(self, exc_type, exc_val, exc_tb):
471
self.logger.removeFilter(self.nullfilter)
593
472
594
473
595
474
def commands_from_options(options):
596
475
597
commands = list(options.commands)
598
599
def find_cmd(cmd, commands):
600
i = 0
601
for i, c in enumerate(commands):
602
if isinstance(c, cmd):
603
return i
604
return i+1
605
606
# If command.Remove is present, move any instances of command.Deny
607
# to occur ahead of command.Remove.
608
index_of_remove = find_cmd(command.Remove, commands)
609
before_remove = commands[:index_of_remove]
610
after_remove = commands[index_of_remove:]
611
cleaned_after = []
612
for cmd in after_remove:
613
if isinstance(cmd, command.Deny):
614
before_remove.append(cmd)
476
commands = []
477
478
if options.is_enabled:
479
commands.append(command.IsEnabled())
480
481
if options.approve:
482
commands.append(command.Approve())
483
484
if options.deny:
485
commands.append(command.Deny())
486
487
if options.remove:
488
commands.append(command.Remove())
489
490
if options.dump_json:
491
commands.append(command.DumpJSON())
492
493
if options.enable:
494
commands.append(command.Enable())
495
496
if options.disable:
497
commands.append(command.Disable())
498
499
if options.bump_timeout:
500
commands.append(command.BumpTimeout())
501
502
if options.start_checker:
503
commands.append(command.StartChecker())
504
505
if options.stop_checker:
506
commands.append(command.StopChecker())
507
508
if options.approved_by_default is not None:
509
if options.approved_by_default:
510
commands.append(command.ApproveByDefault())
615
511
else:
616
cleaned_after.append(cmd)
617
if cleaned_after != after_remove:
618
commands = before_remove + cleaned_after
512
commands.append(command.DenyByDefault())
513
514
if options.checker is not None:
515
commands.append(command.SetChecker(options.checker))
516
517
if options.host is not None:
518
commands.append(command.SetHost(options.host))
519
520
if options.secret is not None:
521
commands.append(command.SetSecret(options.secret))
522
523
if options.timeout is not None:
524
commands.append(command.SetTimeout(options.timeout))
525
526
if options.extended_timeout:
527
commands.append(
528
command.SetExtendedTimeout(options.extended_timeout))
529
530
if options.interval is not None:
531
commands.append(command.SetInterval(options.interval))
532
533
if options.approval_delay is not None:
534
commands.append(
535
command.SetApprovalDelay(options.approval_delay))
536
537
if options.approval_duration is not None:
538
commands.append(
539
command.SetApprovalDuration(options.approval_duration))
619
540
620
541
# If no command option has been given, show table of clients,
621
542
# optionally verbosely
630
551
631
552
class Base(object):
632
553
"""Abstract base class for commands"""
633
def run(self, clients, bus=None):
554
def run(self, clients, bus=None, mandos=None):
634
555
"""Normal commands should implement run_on_one_client(),
635
556
but commands which want to operate on all clients at the same time can
636
557
override this run() method instead.
637
558
"""
638
self.bus = bus
639
for client, properties in clients.items():
559
self.mandos = mandos
560
for clientpath, properties in clients.items():
561
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
562
dbus_busname, str(clientpath))
563
client = bus.get_object(dbus_busname, clientpath)
640
564
self.run_on_one_client(client, properties)
641
565
642
566
643
567
class IsEnabled(Base):
644
def run(self, clients, bus=None):
645
properties = next(iter(clients.values()))
646
if properties["Enabled"]:
568
def run(self, clients, bus=None, mandos=None):
569
client, properties = next(iter(clients.items()))
570
if self.is_enabled(client, properties):
647
571
sys.exit(0)
648
572
sys.exit(1)
573
def is_enabled(self, client, properties):
574
return properties["Enabled"]
649
575
650
576
651
577
class Approve(Base):
652
578
def run_on_one_client(self, client, properties):
653
self.bus.call_client_method(client, "Approve", True)
579
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
580
client.__dbus_object_path__,
581
client_dbus_interface)
582
client.Approve(dbus.Boolean(True),
583
dbus_interface=client_dbus_interface)
654
584
655
585
656
586
class Deny(Base):
657
587
def run_on_one_client(self, client, properties):
658
self.bus.call_client_method(client, "Approve", False)
588
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
589
client.__dbus_object_path__,
590
client_dbus_interface)
591
client.Approve(dbus.Boolean(False),
592
dbus_interface=client_dbus_interface)
659
593
660
594
661
595
class Remove(Base):
662
def run(self, clients, bus):
663
for clientpath in frozenset(clients.keys()):
664
bus.call_server_method("RemoveClient", clientpath)
596
def run_on_one_client(self, client, properties):
597
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
598
dbus_busname, server_dbus_path,
599
server_dbus_interface,
600
str(client.__dbus_object_path__))
601
self.mandos.RemoveClient(client.__dbus_object_path__)
665
602
666
603
667
604
class Output(Base):
677
614
678
615
679
616
class DumpJSON(Output):
680
def run(self, clients, bus=None):
681
data = {properties["Name"]:
682
{key: properties[key]
617
def run(self, clients, bus=None, mandos=None):
618
data = {client["Name"]:
619
{key: self.dbus_boolean_to_bool(client[key])
683
620
for key in self.all_keywords}
684
for properties in clients.values()}
621
for client in clients.values()}
685
622
print(json.dumps(data, indent=4, separators=(',', ': ')))
686
623
624
@staticmethod
625
def dbus_boolean_to_bool(value):
626
if isinstance(value, dbus.Boolean):
627
value = bool(value)
628
return value
629
687
630
688
631
class PrintTable(Output):
689
632
def __init__(self, verbose=False):
690
633
self.verbose = verbose
691
634
692
def run(self, clients, bus=None):
635
def run(self, clients, bus=None, mandos=None):
693
636
default_keywords = ("Name", "Enabled", "Timeout",
694
637
"LastCheckedOK")
695
638
keywords = default_keywords
756
699
757
700
@classmethod
758
701
def valuetostring(cls, value, keyword):
759
if isinstance(value, bool):
702
if isinstance(value, dbus.Boolean):
760
703
return "Yes" if value else "No"
761
704
if keyword in ("Timeout", "Interval", "ApprovalDelay",
762
705
"ApprovalDuration", "ExtendedTimeout"):
785
728
class PropertySetter(Base):
786
729
"Abstract class for Actions for setting one client property"
787
730
788
def run_on_one_client(self, client, properties=None):
731
def run_on_one_client(self, client, properties):
789
732
"""Set the Client's D-Bus property"""
790
self.bus.set_client_property(client, self.propname,
791
self.value_to_set)
733
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
734
client.__dbus_object_path__,
735
dbus.PROPERTIES_IFACE, client_dbus_interface,
736
self.propname, self.value_to_set
737
if not isinstance(self.value_to_set,
738
dbus.Boolean)
739
else bool(self.value_to_set))
740
client.Set(client_dbus_interface, self.propname,
741
self.value_to_set,
742
dbus_interface=dbus.PROPERTIES_IFACE)
792
743
793
744
@property
794
745
def propname(self):
797
748
798
749
class Enable(PropertySetter):
799
750
propname = "Enabled"
800
value_to_set = True
751
value_to_set = dbus.Boolean(True)
801
752
802
753
803
754
class Disable(PropertySetter):
804
755
propname = "Enabled"
805
value_to_set = False
756
value_to_set = dbus.Boolean(False)
806
757
807
758
808
759
class BumpTimeout(PropertySetter):
812
763
813
764
class StartChecker(PropertySetter):
814
765
propname = "CheckerRunning"
815
value_to_set = True
766
value_to_set = dbus.Boolean(True)
816
767
817
768
818
769
class StopChecker(PropertySetter):
819
770
propname = "CheckerRunning"
820
value_to_set = False
771
value_to_set = dbus.Boolean(False)
821
772
822
773
823
774
class ApproveByDefault(PropertySetter):
824
775
propname = "ApprovedByDefault"
825
value_to_set = True
776
value_to_set = dbus.Boolean(True)
826
777
827
778
828
779
class DenyByDefault(PropertySetter):
829
780
propname = "ApprovedByDefault"
830
value_to_set = False
781
value_to_set = dbus.Boolean(False)
831
782
832
783
833
784
class PropertySetterValue(PropertySetter):
836
787
def __init__(self, value):
837
788
self.value_to_set = value
838
789
839
@classmethod
840
def argparse(cls, argtype):
841
def cmdtype(arg):
842
return cls(argtype(arg))
843
return cmdtype
844
790
845
791
class SetChecker(PropertySetterValue):
846
792
propname = "Checker"
934
880
"output"))
935
881
936
882
937
class Unique(object):
938
"""Class for objects which exist only to be unique objects, since
939
unittest.mock.sentinel only exists in Python 3.3"""
940
941
942
883
class Test_string_to_delta(TestCaseWithAssertLogs):
943
884
# Just test basic RFC 3339 functionality here, the doc string for
944
885
# rfc3339_duration_to_delta() already has more comprehensive
945
# tests, which are run by doctest.
886
# tests, which is run by doctest.
946
887
947
888
def test_rfc3339_zero_seconds(self):
948
889
self.assertEqual(datetime.timedelta(),
972
913
973
914
def test_actions_requires_client_or_all(self):
974
915
for action, value in self.actions.items():
975
args = self.actionargs(action, value)
916
options = self.parser.parse_args()
917
setattr(options, action, value)
976
918
with self.assertParseError():
977
self.parse_args(args)
919
self.check_option_syntax(options)
978
920
979
# This mostly corresponds to the definition from has_commands() in
921
# This mostly corresponds to the definition from has_actions() in
980
922
# check_option_syntax()
981
923
actions = {
982
"--enable": None,
983
"--disable": None,
984
"--bump-timeout": None,
985
"--start-checker": None,
986
"--stop-checker": None,
987
"--is-enabled": None,
988
"--remove": None,
989
"--checker": "x",
990
"--timeout": "PT0S",
991
"--extended-timeout": "PT0S",
992
"--interval": "PT0S",
993
"--approve-by-default": None,
994
"--deny-by-default": None,
995
"--approval-delay": "PT0S",
996
"--approval-duration": "PT0S",
997
"--host": "hostname",
998
"--secret": "/dev/null",
999
"--approve": None,
1000
"--deny": None,
924
# The actual values set here are not that important, but we do
925
# at least stick to the correct types, even though they are
926
# never used
927
"enable": True,
928
"disable": True,
929
"bump_timeout": True,
930
"start_checker": True,
931
"stop_checker": True,
932
"is_enabled": True,
933
"remove": True,
934
"checker": "x",
935
"timeout": datetime.timedelta(),
936
"extended_timeout": datetime.timedelta(),
937
"interval": datetime.timedelta(),
938
"approved_by_default": True,
939
"approval_delay": datetime.timedelta(),
940
"approval_duration": datetime.timedelta(),
941
"host": "x",
942
"secret": io.BytesIO(b"x"),
943
"approve": True,
944
"deny": True,
1001
945
}
1002
946
1003
@staticmethod
1004
def actionargs(action, value, *args):
1005
if value is not None:
1006
return [action, value] + list(args)
1007
else:
1008
return [action] + list(args)
1009
1010
947
@contextlib.contextmanager
1011
948
def assertParseError(self):
1012
949
with self.assertRaises(SystemExit) as e:
1017
954
# /argparse.html#exiting-methods
1018
955
self.assertEqual(2, e.exception.code)
1019
956
1020
def parse_args(self, args):
1021
options = self.parser.parse_args(args)
1022
check_option_syntax(self.parser, options)
1023
1024
957
@staticmethod
1025
958
@contextlib.contextmanager
1026
959
def redirect_stderr_to_devnull():
1037
970
1038
971
def test_actions_all_conflicts_with_verbose(self):
1039
972
for action, value in self.actions.items():
1040
args = self.actionargs(action, value, "--all",
1041
"--verbose")
973
options = self.parser.parse_args()
974
setattr(options, action, value)
975
options.all = True
976
options.verbose = True
1042
977
with self.assertParseError():
1043
self.parse_args(args)
978
self.check_option_syntax(options)
1044
979
1045
980
def test_actions_with_client_conflicts_with_verbose(self):
1046
981
for action, value in self.actions.items():
1047
args = self.actionargs(action, value, "--verbose",
1048
"client")
982
options = self.parser.parse_args()
983
setattr(options, action, value)
984
options.verbose = True
985
options.client = ["client"]
1049
986
with self.assertParseError():
1050
self.parse_args(args)
987
self.check_option_syntax(options)
1051
988
1052
989
def test_dump_json_conflicts_with_verbose(self):
1053
args = ["--dump-json", "--verbose"]
990
options = self.parser.parse_args()
991
options.dump_json = True
992
options.verbose = True
1054
993
with self.assertParseError():
1055
self.parse_args(args)
994
self.check_option_syntax(options)
1056
995
1057
996
def test_dump_json_conflicts_with_action(self):
1058
997
for action, value in self.actions.items():
1059
args = self.actionargs(action, value, "--dump-json")
998
options = self.parser.parse_args()
999
setattr(options, action, value)
1000
options.dump_json = True
1060
1001
with self.assertParseError():
1061
self.parse_args(args)
1002
self.check_option_syntax(options)
1062
1003
1063
1004
def test_all_can_not_be_alone(self):
1064
args = ["--all"]
1005
options = self.parser.parse_args()
1006
options.all = True
1065
1007
with self.assertParseError():
1066
self.parse_args(args)
1008
self.check_option_syntax(options)
1067
1009
1068
1010
def test_all_is_ok_with_any_action(self):
1069
1011
for action, value in self.actions.items():
1070
args = self.actionargs(action, value, "--all")
1071
self.parse_args(args)
1012
options = self.parser.parse_args()
1013
setattr(options, action, value)
1014
options.all = True
1015
self.check_option_syntax(options)
1072
1016
1073
1017
def test_any_action_is_ok_with_one_client(self):
1074
1018
for action, value in self.actions.items():
1075
args = self.actionargs(action, value, "client")
1076
self.parse_args(args)
1019
options = self.parser.parse_args()
1020
setattr(options, action, value)
1021
options.client = ["client"]
1022
self.check_option_syntax(options)
1077
1023
1078
1024
def test_one_client_with_all_actions_except_is_enabled(self):
1025
options = self.parser.parse_args()
1079
1026
for action, value in self.actions.items():
1080
if action == "--is-enabled":
1027
if action == "is_enabled":
1081
1028
continue
1082
args = self.actionargs(action, value, "client")
1083
self.parse_args(args)
1029
setattr(options, action, value)
1030
options.client = ["client"]
1031
self.check_option_syntax(options)
1084
1032
1085
1033
def test_two_clients_with_all_actions_except_is_enabled(self):
1034
options = self.parser.parse_args()
1086
1035
for action, value in self.actions.items():
1087
if action == "--is-enabled":
1036
if action == "is_enabled":
1088
1037
continue
1089
args = self.actionargs(action, value, "client1",
1090
"client2")
1091
self.parse_args(args)
1038
setattr(options, action, value)
1039
options.client = ["client1", "client2"]
1040
self.check_option_syntax(options)
1092
1041
1093
1042
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1094
1043
for action, value in self.actions.items():
1095
if action == "--is-enabled":
1044
if action == "is_enabled":
1096
1045
continue
1097
args = self.actionargs(action, value, "client1",
1098
"client2")
1099
self.parse_args(args)
1046
options = self.parser.parse_args()
1047
setattr(options, action, value)
1048
options.client = ["client1", "client2"]
1049
self.check_option_syntax(options)
1100
1050
1101
1051
def test_is_enabled_fails_without_client(self):
1102
args = ["--is-enabled"]
1052
options = self.parser.parse_args()
1053
options.is_enabled = True
1103
1054
with self.assertParseError():
1104
self.parse_args(args)
1055
self.check_option_syntax(options)
1105
1056
1106
1057
def test_is_enabled_fails_with_two_clients(self):
1107
args = ["--is-enabled", "client1", "client2"]
1058
options = self.parser.parse_args()
1059
options.is_enabled = True
1060
options.client = ["client1", "client2"]
1108
1061
with self.assertParseError():
1109
self.parse_args(args)
1062
self.check_option_syntax(options)
1110
1063
1111
1064
def test_remove_can_only_be_combined_with_action_deny(self):
1112
1065
for action, value in self.actions.items():
1113
if action in {"--remove", "--deny"}:
1066
if action in {"remove", "deny"}:
1114
1067
continue
1115
args = self.actionargs(action, value, "--all",
1116
"--remove")
1068
options = self.parser.parse_args()
1069
setattr(options, action, value)
1070
options.all = True
1071
options.remove = True
1117
1072
with self.assertParseError():
1118
self.parse_args(args)
1119
1120
1121
class Test_dbus_exceptions(unittest.TestCase):
1122
1123
def test_dbus_ConnectFailed_is_Error(self):
1124
with self.assertRaises(dbus.Error):
1125
raise dbus.ConnectFailed()
1126
1127
1128
class Test_dbus_MandosBus(unittest.TestCase):
1129
1130
class MockMandosBus(dbus.MandosBus):
1131
def __init__(self):
1132
self._name = "se.recompile.Mandos"
1133
self._server_path = "/"
1134
self._server_interface = "se.recompile.Mandos"
1135
self._client_interface = "se.recompile.Mandos.Client"
1136
self.calls = []
1137
self.call_method_return = Unique()
1138
1139
def call_method(self, methodname, busname, objectpath,
1140
interface, *args):
1141
self.calls.append((methodname, busname, objectpath,
1142
interface, args))
1143
return self.call_method_return
1144
1145
def setUp(self):
1146
self.bus = self.MockMandosBus()
1147
1148
def test_set_client_property(self):
1149
self.bus.set_client_property("objectpath", "key", "value")
1150
expected_call = ("Set", self.bus._name, "objectpath",
1151
"org.freedesktop.DBus.Properties",
1152
(self.bus._client_interface, "key", "value"))
1153
self.assertIn(expected_call, self.bus.calls)
1154
1155
def test_call_client_method(self):
1156
ret = self.bus.call_client_method("objectpath", "methodname")
1157
self.assertIs(self.bus.call_method_return, ret)
1158
expected_call = ("methodname", self.bus._name, "objectpath",
1159
self.bus._client_interface, ())
1160
self.assertIn(expected_call, self.bus.calls)
1161
1162
def test_call_client_method_with_args(self):
1163
args = (Unique(), Unique())
1164
ret = self.bus.call_client_method("objectpath", "methodname",
1165
*args)
1166
self.assertIs(self.bus.call_method_return, ret)
1167
expected_call = ("methodname", self.bus._name, "objectpath",
1168
self.bus._client_interface,
1169
(args[0], args[1]))
1170
self.assertIn(expected_call, self.bus.calls)
1171
1172
def test_get_clients_and_properties(self):
1173
managed_objects = {
1174
"objectpath": {
1175
self.bus._client_interface: {
1176
"key": "value",
1177
"bool": True,
1178
},
1179
"irrelevant_interface": {
1180
"key": "othervalue",
1181
"bool": False,
1182
},
1183
},
1184
"other_objectpath": {
1185
"other_irrelevant_interface": {
1186
"key": "value 3",
1187
"bool": None,
1188
},
1189
},
1190
}
1191
expected_clients_and_properties = {
1192
"objectpath": {
1193
"key": "value",
1194
"bool": True,
1195
}
1196
}
1197
self.bus.call_method_return = managed_objects
1198
ret = self.bus.get_clients_and_properties()
1199
self.assertDictEqual(expected_clients_and_properties, ret)
1200
expected_call = ("GetManagedObjects", self.bus._name,
1201
self.bus._server_path,
1202
"org.freedesktop.DBus.ObjectManager", ())
1203
self.assertIn(expected_call, self.bus.calls)
1204
1205
def test_call_server_method(self):
1206
ret = self.bus.call_server_method("methodname")
1207
self.assertIs(self.bus.call_method_return, ret)
1208
expected_call = ("methodname", self.bus._name,
1209
self.bus._server_path,
1210
self.bus._server_interface, ())
1211
self.assertIn(expected_call, self.bus.calls)
1212
1213
def test_call_server_method_with_args(self):
1214
args = (Unique(), Unique())
1215
ret = self.bus.call_server_method("methodname", *args)
1216
self.assertIs(self.bus.call_method_return, ret)
1217
expected_call = ("methodname", self.bus._name,
1218
self.bus._server_path,
1219
self.bus._server_interface,
1220
(args[0], args[1]))
1221
self.assertIn(expected_call, self.bus.calls)
1222
1223
1224
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1225
1226
def MockDBusPython_func(self, func):
1227
class mock_dbus_python(object):
1228
"""mock dbus-python module"""
1229
class exceptions(object):
1230
"""Pseudo-namespace"""
1231
class DBusException(Exception):
1232
pass
1233
class SystemBus(object):
1234
@staticmethod
1235
def get_object(busname, objectpath):
1236
DBusObject = collections.namedtuple(
1237
"DBusObject", ("methodname",))
1238
def method(*args, **kwargs):
1239
self.assertEqual({"dbus_interface":
1240
"interface"},
1241
kwargs)
1242
return func(*args)
1243
return DBusObject(methodname=method)
1244
class Boolean(object):
1245
def __init__(self, value):
1246
self.value = bool(value)
1247
def __bool__(self):
1248
return self.value
1249
if sys.version_info.major == 2:
1250
__nonzero__ = __bool__
1251
class ObjectPath(str):
1252
pass
1253
class Dictionary(dict):
1254
pass
1255
return mock_dbus_python
1256
1257
def call_method(self, bus, methodname, busname, objectpath,
1258
interface, *args):
1259
with self.assertLogs(log, logging.DEBUG):
1260
return bus.call_method(methodname, busname, objectpath,
1261
interface, *args)
1262
1263
def test_call_method_returns(self):
1264
expected_method_return = Unique()
1265
method_args = (Unique(), Unique())
1266
def func(*args):
1267
self.assertEqual(len(method_args), len(args))
1268
for marg, arg in zip(method_args, args):
1269
self.assertIs(marg, arg)
1270
return expected_method_return
1271
mock_dbus_python = self.MockDBusPython_func(func)
1272
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1273
ret = self.call_method(bus, "methodname", "busname",
1274
"objectpath", "interface",
1275
*method_args)
1276
self.assertIs(ret, expected_method_return)
1277
1278
def test_call_method_filters_bool_true(self):
1279
def func():
1280
return method_return
1281
mock_dbus_python = self.MockDBusPython_func(func)
1282
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1283
method_return = mock_dbus_python.Boolean(True)
1284
ret = self.call_method(bus, "methodname", "busname",
1285
"objectpath", "interface")
1286
self.assertTrue(ret)
1287
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1288
1289
def test_call_method_filters_bool_false(self):
1290
def func():
1291
return method_return
1292
mock_dbus_python = self.MockDBusPython_func(func)
1293
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1294
method_return = mock_dbus_python.Boolean(False)
1295
ret = self.call_method(bus, "methodname", "busname",
1296
"objectpath", "interface")
1297
self.assertFalse(ret)
1298
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1299
1300
def test_call_method_filters_objectpath(self):
1301
def func():
1302
return method_return
1303
mock_dbus_python = self.MockDBusPython_func(func)
1304
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1305
method_return = mock_dbus_python.ObjectPath("objectpath")
1306
ret = self.call_method(bus, "methodname", "busname",
1307
"objectpath", "interface")
1308
self.assertEqual("objectpath", ret)
1309
self.assertIsNot("objectpath", ret)
1310
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1311
1312
def test_call_method_filters_booleans_in_dict(self):
1313
def func():
1314
return method_return
1315
mock_dbus_python = self.MockDBusPython_func(func)
1316
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1317
method_return = mock_dbus_python.Dictionary(
1318
{mock_dbus_python.Boolean(True):
1319
mock_dbus_python.Boolean(False),
1320
mock_dbus_python.Boolean(False):
1321
mock_dbus_python.Boolean(True)})
1322
ret = self.call_method(bus, "methodname", "busname",
1323
"objectpath", "interface")
1324
expected_method_return = {True: False,
1325
False: True}
1326
self.assertEqual(expected_method_return, ret)
1327
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1328
1329
def test_call_method_filters_objectpaths_in_dict(self):
1330
def func():
1331
return method_return
1332
mock_dbus_python = self.MockDBusPython_func(func)
1333
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1334
method_return = mock_dbus_python.Dictionary(
1335
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1336
mock_dbus_python.ObjectPath("objectpath_value_1"),
1337
mock_dbus_python.ObjectPath("objectpath_key_2"):
1338
mock_dbus_python.ObjectPath("objectpath_value_2")})
1339
ret = self.call_method(bus, "methodname", "busname",
1340
"objectpath", "interface")
1341
expected_method_return = {str(key): str(value)
1342
for key, value in
1343
method_return.items()}
1344
self.assertEqual(expected_method_return, ret)
1345
self.assertIsInstance(ret, dict)
1346
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1347
1348
def test_call_method_filters_dict_in_dict(self):
1349
def func():
1350
return method_return
1351
mock_dbus_python = self.MockDBusPython_func(func)
1352
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1353
method_return = mock_dbus_python.Dictionary(
1354
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1355
"key12": "value12"}),
1356
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1357
"key22": "value22"})})
1358
ret = self.call_method(bus, "methodname", "busname",
1359
"objectpath", "interface")
1360
expected_method_return = {
1361
"key1": {"key11": "value11",
1362
"key12": "value12"},
1363
"key2": {"key21": "value21",
1364
"key22": "value22"},
1365
}
1366
self.assertEqual(expected_method_return, ret)
1367
self.assertIsInstance(ret, dict)
1368
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1369
for key, value in ret.items():
1370
self.assertIsInstance(value, dict)
1371
self.assertEqual(expected_method_return[key], value)
1372
self.assertNotIsInstance(value,
1373
mock_dbus_python.Dictionary)
1374
1375
def test_call_method_filters_dict_three_deep(self):
1376
def func():
1377
return method_return
1378
mock_dbus_python = self.MockDBusPython_func(func)
1379
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1380
method_return = mock_dbus_python.Dictionary(
1381
{"key1":
1382
mock_dbus_python.Dictionary(
1383
{"key2":
1384
mock_dbus_python.Dictionary(
1385
{"key3":
1386
mock_dbus_python.Boolean(True),
1387
}),
1388
}),
1389
})
1390
ret = self.call_method(bus, "methodname", "busname",
1391
"objectpath", "interface")
1392
expected_method_return = {"key1": {"key2": {"key3": True}}}
1393
self.assertEqual(expected_method_return, ret)
1394
self.assertIsInstance(ret, dict)
1395
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1396
self.assertIsInstance(ret["key1"], dict)
1397
self.assertNotIsInstance(ret["key1"],
1398
mock_dbus_python.Dictionary)
1399
self.assertIsInstance(ret["key1"]["key2"], dict)
1400
self.assertNotIsInstance(ret["key1"]["key2"],
1401
mock_dbus_python.Dictionary)
1402
self.assertTrue(ret["key1"]["key2"]["key3"])
1403
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1404
mock_dbus_python.Boolean)
1405
1406
def test_call_method_handles_exception(self):
1073
self.check_option_syntax(options)
1074
1075
1076
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1077
def test_calls_and_returns_get_object_on_bus(self):
1078
class MockBus(object):
1079
called = False
1080
def get_object(mockbus_self, busname, dbus_path):
1081
# Note that "self" is still the testcase instance,
1082
# this MockBus instance is in "mockbus_self".
1083
self.assertEqual(dbus_busname, busname)
1084
self.assertEqual(server_dbus_path, dbus_path)
1085
mockbus_self.called = True
1086
return mockbus_self
1087
1088
mockbus = get_mandos_dbus_object(bus=MockBus())
1089
self.assertIsInstance(mockbus, MockBus)
1090
self.assertTrue(mockbus.called)
1091
1092
def test_logs_and_exits_on_dbus_error(self):
1093
class FailingBusStub(object):
1094
def get_object(self, busname, dbus_path):
1095
raise dbus.exceptions.DBusException("Test")
1096
1097
with self.assertLogs(log, logging.CRITICAL):
1098
with self.assertRaises(SystemExit) as e:
1099
bus = get_mandos_dbus_object(bus=FailingBusStub())
1100
1101
if isinstance(e.exception.code, int):
1102
self.assertNotEqual(0, e.exception.code)
1103
else:
1104
self.assertIsNotNone(e.exception.code)
1105
1106
1107
class Test_get_managed_objects(TestCaseWithAssertLogs):
1108
def test_calls_and_returns_GetManagedObjects(self):
1109
managed_objects = {"/clients/client": { "Name": "client"}}
1110
class ObjectManagerStub(object):
1111
def GetManagedObjects(self):
1112
return managed_objects
1113
retval = get_managed_objects(ObjectManagerStub())
1114
self.assertDictEqual(managed_objects, retval)
1115
1116
def test_logs_and_exits_on_dbus_error(self):
1407
1117
dbus_logger = logging.getLogger("dbus.proxies")
1408
1118
1409
def func():
1410
dbus_logger.error("Test")
1411
raise mock_dbus_python.exceptions.DBusException()
1412
1413
mock_dbus_python = self.MockDBusPython_func(func)
1414
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1119
class ObjectManagerFailingStub(object):
1120
def GetManagedObjects(self):
1121
dbus_logger.error("Test")
1122
raise dbus.exceptions.DBusException("Test")
1415
1123
1416
1124
class CountingHandler(logging.Handler):
1417
1125
count = 0
1423
1131
dbus_logger.addHandler(counting_handler)
1424
1132
1425
1133
try:
1426
with self.assertRaises(dbus.Error) as e:
1427
self.call_method(bus, "methodname", "busname",
1428
"objectpath", "interface")
1134
with self.assertLogs(log, logging.CRITICAL) as watcher:
1135
with self.assertRaises(SystemExit) as e:
1136
get_managed_objects(ObjectManagerFailingStub())
1429
1137
finally:
1430
1138
dbus_logger.removeFilter(counting_handler)
1431
1139
1432
self.assertNotIsInstance(e, dbus.ConnectFailed)
1433
1434
1140
# Make sure the dbus logger was suppressed
1435
1141
self.assertEqual(0, counting_handler.count)
1436
1142
1437
def test_get_object_converts_to_correct_exception(self):
1438
bus = dbus_python_adapter.SystemBus(
1439
self.fake_dbus_python_raises_exception_on_connect)
1440
with self.assertRaises(dbus.ConnectFailed):
1441
self.call_method(bus, "methodname", "busname",
1442
"objectpath", "interface")
1443
1444
class fake_dbus_python_raises_exception_on_connect(object):
1445
"""fake dbus-python module"""
1446
class exceptions(object):
1447
"""Pseudo-namespace"""
1448
class DBusException(Exception):
1449
pass
1450
1451
@classmethod
1452
def SystemBus(cls):
1453
def get_object(busname, objectpath):
1454
raise cls.exceptions.DBusException()
1455
Bus = collections.namedtuple("Bus", ["get_object"])
1456
return Bus(get_object=get_object)
1457
1458
1459
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1460
class mock_dbus_python(object):
1461
"""mock dbus-python modules"""
1462
class SystemBus(object):
1463
@staticmethod
1464
def get_object(busname, objectpath):
1465
return Unique()
1466
1467
def setUp(self):
1468
self.bus = dbus_python_adapter.CachingBus(
1469
self.mock_dbus_python)
1470
1471
def test_returns_distinct_objectpaths(self):
1472
obj1 = self.bus.get_object("busname", "objectpath1")
1473
self.assertIsInstance(obj1, Unique)
1474
obj2 = self.bus.get_object("busname", "objectpath2")
1475
self.assertIsInstance(obj2, Unique)
1476
self.assertIsNot(obj1, obj2)
1477
1478
def test_returns_distinct_busnames(self):
1479
obj1 = self.bus.get_object("busname1", "objectpath")
1480
self.assertIsInstance(obj1, Unique)
1481
obj2 = self.bus.get_object("busname2", "objectpath")
1482
self.assertIsInstance(obj2, Unique)
1483
self.assertIsNot(obj1, obj2)
1484
1485
def test_returns_distinct_both(self):
1486
obj1 = self.bus.get_object("busname1", "objectpath")
1487
self.assertIsInstance(obj1, Unique)
1488
obj2 = self.bus.get_object("busname2", "objectpath")
1489
self.assertIsInstance(obj2, Unique)
1490
self.assertIsNot(obj1, obj2)
1491
1492
def test_returns_same(self):
1493
obj1 = self.bus.get_object("busname", "objectpath")
1494
self.assertIsInstance(obj1, Unique)
1495
obj2 = self.bus.get_object("busname", "objectpath")
1496
self.assertIsInstance(obj2, Unique)
1497
self.assertIs(obj1, obj2)
1498
1499
def test_returns_same_old(self):
1500
obj1 = self.bus.get_object("busname1", "objectpath1")
1501
self.assertIsInstance(obj1, Unique)
1502
obj2 = self.bus.get_object("busname2", "objectpath2")
1503
self.assertIsInstance(obj2, Unique)
1504
obj1b = self.bus.get_object("busname1", "objectpath1")
1505
self.assertIsInstance(obj1b, Unique)
1506
self.assertIsNot(obj1, obj2)
1507
self.assertIsNot(obj2, obj1b)
1508
self.assertIs(obj1, obj1b)
1143
# Test that the dbus_logger still works
1144
with self.assertLogs(dbus_logger, logging.ERROR):
1145
dbus_logger.error("Test")
1146
1147
if isinstance(e.exception.code, int):
1148
self.assertNotEqual(0, e.exception.code)
1149
else:
1150
self.assertIsNotNone(e.exception.code)
1509
1151
1510
1152
1511
1153
class Test_commands_from_options(unittest.TestCase):
1512
1513
1154
def setUp(self):
1514
1155
self.parser = argparse.ArgumentParser()
1515
1156
add_command_line_options(self.parser)
1730
1371
class TestCommand(unittest.TestCase):
1731
1372
"""Abstract class for tests of command classes"""
1732
1373
1733
class FakeMandosBus(dbus.MandosBus):
1734
def __init__(self, testcase):
1735
self.client_properties = {
1736
"Name": "foo",
1737
"KeyID": ("92ed150794387c03ce684574b1139a65"
1738
"94a34f895daaaf09fd8ea90a27cddb12"),
1739
"Secret": b"secret",
1740
"Host": "foo.example.org",
1741
"Enabled": True,
1742
"Timeout": 300000,
1743
"LastCheckedOK": "2019-02-03T00:00:00",
1744
"Created": "2019-01-02T00:00:00",
1745
"Interval": 120000,
1746
"Fingerprint": ("778827225BA7DE539C5A"
1747
"7CFA59CFF7CDBD9A5920"),
1748
"CheckerRunning": False,
1749
"LastEnabled": "2019-01-03T00:00:00",
1750
"ApprovalPending": False,
1751
"ApprovedByDefault": True,
1752
"LastApprovalRequest": "",
1753
"ApprovalDelay": 0,
1754
"ApprovalDuration": 1000,
1755
"Checker": "fping -q -- %(host)s",
1756
"ExtendedTimeout": 900000,
1757
"Expires": "2019-02-04T00:00:00",
1758
"LastCheckerStatus": 0,
1759
}
1760
self.other_client_properties = {
1761
"Name": "barbar",
1762
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1763
"6ab612cff5ad227247e46c2b020f441c"),
1764
"Secret": b"secretbar",
1765
"Host": "192.0.2.3",
1766
"Enabled": True,
1767
"Timeout": 300000,
1768
"LastCheckedOK": "2019-02-04T00:00:00",
1769
"Created": "2019-01-03T00:00:00",
1770
"Interval": 120000,
1771
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1772
"F547B3A107558FCA3A27"),
1773
"CheckerRunning": True,
1774
"LastEnabled": "2019-01-04T00:00:00",
1775
"ApprovalPending": False,
1776
"ApprovedByDefault": False,
1777
"LastApprovalRequest": "2019-01-03T00:00:00",
1778
"ApprovalDelay": 30000,
1779
"ApprovalDuration": 93785000,
1780
"Checker": ":",
1781
"ExtendedTimeout": 900000,
1782
"Expires": "2019-02-05T00:00:00",
1783
"LastCheckerStatus": -2,
1784
}
1785
self.clients = collections.OrderedDict(
1786
[
1787
("client_objectpath", self.client_properties),
1788
("other_client_objectpath",
1789
self.other_client_properties),
1790
])
1791
self.one_client = {"client_objectpath":
1792
self.client_properties}
1793
self.testcase = testcase
1794
self.calls = []
1795
1796
def call_method(self, methodname, busname, objectpath,
1797
interface, *args):
1798
self.testcase.assertEqual("se.recompile.Mandos", busname)
1799
self.calls.append((methodname, busname, objectpath,
1800
interface, args))
1801
if interface == "org.freedesktop.DBus.Properties":
1802
if methodname == "Set":
1803
self.testcase.assertEqual(3, len(args))
1804
interface, key, value = args
1805
self.testcase.assertEqual(
1806
"se.recompile.Mandos.Client", interface)
1807
self.clients[objectpath][key] = value
1808
return
1809
elif interface == "se.recompile.Mandos":
1810
self.testcase.assertEqual("RemoveClient", methodname)
1811
self.testcase.assertEqual(1, len(args))
1812
clientpath = args[0]
1813
del self.clients[clientpath]
1814
return
1815
elif interface == "se.recompile.Mandos.Client":
1816
if methodname == "Approve":
1817
self.testcase.assertEqual(1, len(args))
1818
return
1819
raise ValueError()
1820
1821
1374
def setUp(self):
1822
self.bus = self.FakeMandosBus(self)
1375
testcase = self
1376
class MockClient(object):
1377
def __init__(self, name, **attributes):
1378
self.__dbus_object_path__ = "/clients/{}".format(name)
1379
self.attributes = attributes
1380
self.attributes["Name"] = name
1381
self.calls = []
1382
def Set(self, interface, propname, value, dbus_interface):
1383
testcase.assertEqual(client_dbus_interface, interface)
1384
testcase.assertEqual(dbus.PROPERTIES_IFACE,
1385
dbus_interface)
1386
self.attributes[propname] = value
1387
def Approve(self, approve, dbus_interface):
1388
testcase.assertEqual(client_dbus_interface,
1389
dbus_interface)
1390
self.calls.append(("Approve", (approve,
1391
dbus_interface)))
1392
self.client = MockClient(
1393
"foo",
1394
KeyID=("92ed150794387c03ce684574b1139a65"
1395
"94a34f895daaaf09fd8ea90a27cddb12"),
1396
Secret=b"secret",
1397
Host="foo.example.org",
1398
Enabled=dbus.Boolean(True),
1399
Timeout=300000,
1400
LastCheckedOK="2019-02-03T00:00:00",
1401
Created="2019-01-02T00:00:00",
1402
Interval=120000,
1403
Fingerprint=("778827225BA7DE539C5A"
1404
"7CFA59CFF7CDBD9A5920"),
1405
CheckerRunning=dbus.Boolean(False),
1406
LastEnabled="2019-01-03T00:00:00",
1407
ApprovalPending=dbus.Boolean(False),
1408
ApprovedByDefault=dbus.Boolean(True),
1409
LastApprovalRequest="",
1410
ApprovalDelay=0,
1411
ApprovalDuration=1000,
1412
Checker="fping -q -- %(host)s",
1413
ExtendedTimeout=900000,
1414
Expires="2019-02-04T00:00:00",
1415
LastCheckerStatus=0)
1416
self.other_client = MockClient(
1417
"barbar",
1418
KeyID=("0558568eedd67d622f5c83b35a115f79"
1419
"6ab612cff5ad227247e46c2b020f441c"),
1420
Secret=b"secretbar",
1421
Host="192.0.2.3",
1422
Enabled=dbus.Boolean(True),
1423
Timeout=300000,
1424
LastCheckedOK="2019-02-04T00:00:00",
1425
Created="2019-01-03T00:00:00",
1426
Interval=120000,
1427
Fingerprint=("3E393AEAEFB84C7E89E2"
1428
"F547B3A107558FCA3A27"),
1429
CheckerRunning=dbus.Boolean(True),
1430
LastEnabled="2019-01-04T00:00:00",
1431
ApprovalPending=dbus.Boolean(False),
1432
ApprovedByDefault=dbus.Boolean(False),
1433
LastApprovalRequest="2019-01-03T00:00:00",
1434
ApprovalDelay=30000,
1435
ApprovalDuration=93785000,
1436
Checker=":",
1437
ExtendedTimeout=900000,
1438
Expires="2019-02-05T00:00:00",
1439
LastCheckerStatus=-2)
1440
self.clients = collections.OrderedDict(
1441
[
1442
(self.client.__dbus_object_path__,
1443
self.client.attributes),
1444
(self.other_client.__dbus_object_path__,
1445
self.other_client.attributes),
1446
])
1447
self.one_client = {self.client.__dbus_object_path__:
1448
self.client.attributes}
1449
1450
@property
1451
def bus(self):
1452
class MockBus(object):
1453
@staticmethod
1454
def get_object(client_bus_name, path):
1455
self.assertEqual(dbus_busname, client_bus_name)
1456
# Note: "self" here is the TestCmd instance, not the
1457
# MockBus instance, since this is a static method!
1458
if path == self.client.__dbus_object_path__:
1459
return self.client
1460
elif path == self.other_client.__dbus_object_path__:
1461
return self.other_client
1462
return MockBus()
1823
1463
1824
1464
1825
1465
class TestBaseCommands(TestCommand):
1826
1466
1827
1467
def test_IsEnabled_exits_successfully(self):
1828
1468
with self.assertRaises(SystemExit) as e:
1829
command.IsEnabled().run(self.bus.one_client)
1469
command.IsEnabled().run(self.one_client)
1830
1470
if e.exception.code is not None:
1831
1471
self.assertEqual(0, e.exception.code)
1832
1472
else:
1833
1473
self.assertIsNone(e.exception.code)
1834
1474
1835
1475
def test_IsEnabled_exits_with_failure(self):
1836
self.bus.client_properties["Enabled"] = False
1476
self.client.attributes["Enabled"] = dbus.Boolean(False)
1837
1477
with self.assertRaises(SystemExit) as e:
1838
command.IsEnabled().run(self.bus.one_client)
1478
command.IsEnabled().run(self.one_client)
1839
1479
if isinstance(e.exception.code, int):
1840
1480
self.assertNotEqual(0, e.exception.code)
1841
1481
else:
1842
1482
self.assertIsNotNone(e.exception.code)
1843
1483
1844
1484
def test_Approve(self):
1845
busname = "se.recompile.Mandos"
1846
client_interface = "se.recompile.Mandos.Client"
1847
command.Approve().run(self.bus.clients, self.bus)
1848
for clientpath in self.bus.clients:
1849
self.assertIn(("Approve", busname, clientpath,
1850
client_interface, (True,)), self.bus.calls)
1485
command.Approve().run(self.clients, self.bus)
1486
for clientpath in self.clients:
1487
client = self.bus.get_object(dbus_busname, clientpath)
1488
self.assertIn(("Approve", (True, client_dbus_interface)),
1489
client.calls)
1851
1490
1852
1491
def test_Deny(self):
1853
busname = "se.recompile.Mandos"
1854
client_interface = "se.recompile.Mandos.Client"
1855
command.Deny().run(self.bus.clients, self.bus)
1856
for clientpath in self.bus.clients:
1857
self.assertIn(("Approve", busname, clientpath,
1858
client_interface, (False,)),
1859
self.bus.calls)
1492
command.Deny().run(self.clients, self.bus)
1493
for clientpath in self.clients:
1494
client = self.bus.get_object(dbus_busname, clientpath)
1495
self.assertIn(("Approve", (False, client_dbus_interface)),
1496
client.calls)
1860
1497
1861
1498
def test_Remove(self):
1862
command.Remove().run(self.bus.clients, self.bus)
1863
for clientpath in self.bus.clients:
1864
self.assertIn(("RemoveClient", dbus_busname,
1865
dbus_server_path, dbus_server_interface,
1866
(clientpath,)), self.bus.calls)
1499
class MandosSpy(object):
1500
def __init__(self):
1501
self.calls = []
1502
def RemoveClient(self, dbus_path):
1503
self.calls.append(("RemoveClient", (dbus_path,)))
1504
mandos = MandosSpy()
1505
command.Remove().run(self.clients, self.bus, mandos)
1506
for clientpath in self.clients:
1507
self.assertIn(("RemoveClient", (clientpath,)),
1508
mandos.calls)
1867
1509
1868
1510
expected_json = {
1869
1511
"foo": {
1918
1560
1919
1561
def test_DumpJSON_normal(self):
1920
1562
with self.capture_stdout_to_buffer() as buffer:
1921
command.DumpJSON().run(self.bus.clients)
1563
command.DumpJSON().run(self.clients)
1922
1564
json_data = json.loads(buffer.getvalue())
1923
1565
self.assertDictEqual(self.expected_json, json_data)
1924
1566
1935
1577
1936
1578
def test_DumpJSON_one_client(self):
1937
1579
with self.capture_stdout_to_buffer() as buffer:
1938
command.DumpJSON().run(self.bus.one_client)
1580
command.DumpJSON().run(self.one_client)
1939
1581
json_data = json.loads(buffer.getvalue())
1940
1582
expected_json = {"foo": self.expected_json["foo"]}
1941
1583
self.assertDictEqual(expected_json, json_data)
1942
1584
1943
1585
def test_PrintTable_normal(self):
1944
1586
with self.capture_stdout_to_buffer() as buffer:
1945
command.PrintTable().run(self.bus.clients)
1587
command.PrintTable().run(self.clients)
1946
1588
expected_output = "\n".join((
1947
1589
"Name Enabled Timeout Last Successful Check",
1948
1590
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1952
1594
1953
1595
def test_PrintTable_verbose(self):
1954
1596
with self.capture_stdout_to_buffer() as buffer:
1955
command.PrintTable(verbose=True).run(self.bus.clients)
1597
command.PrintTable(verbose=True).run(self.clients)
1956
1598
columns = (
1957
1599
(
1958
1600
"Name ",
2048
1690
2049
1691
def test_PrintTable_one_client(self):
2050
1692
with self.capture_stdout_to_buffer() as buffer:
2051
command.PrintTable().run(self.bus.one_client)
1693
command.PrintTable().run(self.one_client)
2052
1694
expected_output = "\n".join((
2053
1695
"Name Enabled Timeout Last Successful Check",
2054
1696
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2058
1700
2059
1701
class TestPropertySetterCmd(TestCommand):
2060
1702
"""Abstract class for tests of command.PropertySetter classes"""
2061
2062
1703
def runTest(self):
2063
1704
if not hasattr(self, "command"):
2064
return # Abstract TestCase class
2065
2066
if hasattr(self, "values_to_set"):
2067
cmd_args = [(value,) for value in self.values_to_set]
2068
values_to_get = getattr(self, "values_to_get",
2069
self.values_to_set)
2070
else:
2071
cmd_args = [() for x in range(len(self.values_to_get))]
2072
values_to_get = self.values_to_get
2073
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2074
for clientpath in self.bus.clients:
2075
self.bus.clients[clientpath][self.propname] = (
2076
Unique())
2077
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2078
for clientpath in self.bus.clients:
2079
value = (self.bus.clients[clientpath]
2080
[self.propname])
2081
self.assertNotIsInstance(value, Unique)
1705
return
1706
values_to_get = getattr(self, "values_to_get",
1707
self.values_to_set)
1708
for value_to_set, value_to_get in zip(self.values_to_set,
1709
values_to_get):
1710
for clientpath in self.clients:
1711
client = self.bus.get_object(dbus_busname, clientpath)
1712
old_value = client.attributes[self.propname]
1713
client.attributes[self.propname] = self.Unique()
1714
self.run_command(value_to_set, self.clients)
1715
for clientpath in self.clients:
1716
client = self.bus.get_object(dbus_busname, clientpath)
1717
value = client.attributes[self.propname]
1718
self.assertNotIsInstance(value, self.Unique)
2082
1719
self.assertEqual(value_to_get, value)
2083
1720
1721
class Unique(object):
1722
"""Class for objects which exist only to be unique objects,
1723
since unittest.mock.sentinel only exists in Python 3.3"""
1724
1725
def run_command(self, value, clients):
1726
self.command().run(clients, self.bus)
1727
2084
1728
2085
1729
class TestEnableCmd(TestPropertySetterCmd):
2086
1730
command = command.Enable
2087
1731
propname = "Enabled"
2088
values_to_get = [True]
1732
values_to_set = [dbus.Boolean(True)]
2089
1733
2090
1734
2091
1735
class TestDisableCmd(TestPropertySetterCmd):
2092
1736
command = command.Disable
2093
1737
propname = "Enabled"
2094
values_to_get = [False]
1738
values_to_set = [dbus.Boolean(False)]
2095
1739
2096
1740
2097
1741
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2098
1742
command = command.BumpTimeout
2099
1743
propname = "LastCheckedOK"
2100
values_to_get = [""]
1744
values_to_set = [""]
2101
1745
2102
1746
2103
1747
class TestStartCheckerCmd(TestPropertySetterCmd):
2104
1748
command = command.StartChecker
2105
1749
propname = "CheckerRunning"
2106
values_to_get = [True]
1750
values_to_set = [dbus.Boolean(True)]
2107
1751
2108
1752
2109
1753
class TestStopCheckerCmd(TestPropertySetterCmd):
2110
1754
command = command.StopChecker
2111
1755
propname = "CheckerRunning"
2112
values_to_get = [False]
1756
values_to_set = [dbus.Boolean(False)]
2113
1757
2114
1758
2115
1759
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2116
1760
command = command.ApproveByDefault
2117
1761
propname = "ApprovedByDefault"
2118
values_to_get = [True]
1762
values_to_set = [dbus.Boolean(True)]
2119
1763
2120
1764
2121
1765
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2122
1766
command = command.DenyByDefault
2123
1767
propname = "ApprovedByDefault"
2124
values_to_get = [False]
2125
2126
2127
class TestSetCheckerCmd(TestPropertySetterCmd):
1768
values_to_set = [dbus.Boolean(False)]
1769
1770
1771
class TestPropertySetterValueCmd(TestPropertySetterCmd):
1772
"""Abstract class for tests of PropertySetterValueCmd classes"""
1773
1774
def runTest(self):
1775
if type(self) is TestPropertySetterValueCmd:
1776
return
1777
return super(TestPropertySetterValueCmd, self).runTest()
1778
1779
def run_command(self, value, clients):
1780
self.command(value).run(clients, self.bus)
1781
1782
1783
class TestSetCheckerCmd(TestPropertySetterValueCmd):
2128
1784
command = command.SetChecker
2129
1785
propname = "Checker"
2130
1786
values_to_set = ["", ":", "fping -q -- %s"]
2131
1787
2132
1788
2133
class TestSetHostCmd(TestPropertySetterCmd):
1789
class TestSetHostCmd(TestPropertySetterValueCmd):
2134
1790
command = command.SetHost
2135
1791
propname = "Host"
2136
1792
values_to_set = ["192.0.2.3", "client.example.org"]
2137
1793
2138
1794
2139
class TestSetSecretCmd(TestPropertySetterCmd):
1795
class TestSetSecretCmd(TestPropertySetterValueCmd):
2140
1796
command = command.SetSecret
2141
1797
propname = "Secret"
2142
1798
values_to_set = [io.BytesIO(b""),
2144
1800
values_to_get = [f.getvalue() for f in values_to_set]
2145
1801
2146
1802
2147
class TestSetTimeoutCmd(TestPropertySetterCmd):
1803
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
2148
1804
command = command.SetTimeout
2149
1805
propname = "Timeout"
2150
1806
values_to_set = [datetime.timedelta(),
2155
1811
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2156
1812
2157
1813
2158
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
1814
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
2159
1815
command = command.SetExtendedTimeout
2160
1816
propname = "ExtendedTimeout"
2161
1817
values_to_set = [datetime.timedelta(),
2166
1822
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2167
1823
2168
1824
2169
class TestSetIntervalCmd(TestPropertySetterCmd):
1825
class TestSetIntervalCmd(TestPropertySetterValueCmd):
2170
1826
command = command.SetInterval
2171
1827
propname = "Interval"
2172
1828
values_to_set = [datetime.timedelta(),
2177
1833
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2178
1834
2179
1835
2180
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
1836
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
2181
1837
command = command.SetApprovalDelay
2182
1838
propname = "ApprovalDelay"
2183
1839
values_to_set = [datetime.timedelta(),
2188
1844
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2189
1845
2190
1846
2191
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
1847
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
2192
1848
command = command.SetApprovalDuration
2193
1849
propname = "ApprovalDuration"
2194
1850
values_to_set = [datetime.timedelta(),
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0