To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1088
- compare with another revision
- download diff
- download tarball
- view history from revision 1088
- Committer: Teddy Hogeborn
- Date: 2019-03-23 14:59:54 UTC
- Revision ID: teddy@recompile.se-20190323145954-latqk2bg6xt989d5
mandos: Refactor; use classes, not instances, as namespaces
* mandos (Avahi): Rename to "avahi".
(avahi.string_array_to_txt_array): Make static method.
(GnuTLS): Rename to "gnutls".
(gnutls.__init__): Remove; move version check up to class body.
(gnutls.Error): Don't use class name anymore.
* mandos (Avahi): Rename to "avahi".
(avahi.string_array_to_txt_array): Make static method.
(GnuTLS): Rename to "gnutls".
(gnutls.__init__): Remove; move version check up to class body.
(gnutls.Error): Don't use class name anymore.
- files modified:
- intro.xml
added
removed
mandos-ctl
46
46
import tempfile
47
47
import contextlib
48
48
49
try:
50
import pydbus
51
import gi
52
dbus_python = None
53
except ImportError:
54
import dbus as dbus_python
55
pydbus = None
56
class gi(object):
57
"""Dummy gi module, for the tests"""
58
class repository(object):
59
class GLib(object):
60
class Error(Exception):
61
pass
49
import dbus
62
50
63
51
# Show warnings by default
64
52
if not sys.warnoptions:
78
66
79
67
locale.setlocale(locale.LC_ALL, "")
80
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
81
75
version = "1.8.3"
82
76
83
77
78
try:
79
dbus.OBJECT_MANAGER_IFACE
80
except AttributeError:
81
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
82
83
84
84
def main():
85
85
parser = argparse.ArgumentParser()
86
86
add_command_line_options(parser)
93
93
if options.debug:
94
94
log.setLevel(logging.DEBUG)
95
95
96
if pydbus is not None:
97
bus = pydbus_adapter.CachingBus(pydbus)
98
else:
99
bus = dbus_python_adapter.CachingBus(dbus_python)
100
101
try:
102
all_clients = bus.get_clients_and_properties()
103
except dbus.ConnectFailed as e:
104
log.critical("Could not connect to Mandos server: %s", e)
105
sys.exit(1)
106
except dbus.Error as e:
107
log.critical(
108
"Failed to access Mandos server through D-Bus:\n%s", e)
109
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
110
113
111
114
# Compile dict of (clientpath: properties) to process
112
115
if not clientnames:
125
128
commands = commands_from_options(options)
126
129
127
130
for command in commands:
128
command.run(clients, bus)
131
command.run(clients, bus, mandos_serv)
129
132
130
133
131
134
def add_command_line_options(parser):
136
139
help="Select all clients")
137
140
parser.add_argument("-v", "--verbose", action="store_true",
138
141
help="Print all fields")
139
parser.add_argument("-j", "--dump-json", dest="commands",
140
action="append_const", default=[],
141
const=command.DumpJSON(),
142
parser.add_argument("-j", "--dump-json", action="store_true",
142
143
help="Dump client data in JSON format")
143
144
enable_disable = parser.add_mutually_exclusive_group()
144
enable_disable.add_argument("-e", "--enable", dest="commands",
145
action="append_const", default=[],
146
const=command.Enable(),
145
enable_disable.add_argument("-e", "--enable", action="store_true",
147
146
help="Enable client")
148
enable_disable.add_argument("-d", "--disable", dest="commands",
149
action="append_const", default=[],
150
const=command.Disable(),
147
enable_disable.add_argument("-d", "--disable",
148
action="store_true",
151
149
help="disable client")
152
parser.add_argument("-b", "--bump-timeout", dest="commands",
153
action="append_const", default=[],
154
const=command.BumpTimeout(),
150
parser.add_argument("-b", "--bump-timeout", action="store_true",
155
151
help="Bump timeout for client")
156
152
start_stop_checker = parser.add_mutually_exclusive_group()
157
153
start_stop_checker.add_argument("--start-checker",
158
dest="commands",
159
action="append_const", default=[],
160
const=command.StartChecker(),
154
action="store_true",
161
155
help="Start checker for client")
162
start_stop_checker.add_argument("--stop-checker", dest="commands",
163
action="append_const", default=[],
164
const=command.StopChecker(),
156
start_stop_checker.add_argument("--stop-checker",
157
action="store_true",
165
158
help="Stop checker for client")
166
parser.add_argument("-V", "--is-enabled", dest="commands",
167
action="append_const", default=[],
168
const=command.IsEnabled(),
159
parser.add_argument("-V", "--is-enabled", action="store_true",
169
160
help="Check if client is enabled")
170
parser.add_argument("-r", "--remove", dest="commands",
171
action="append_const", default=[],
172
const=command.Remove(),
161
parser.add_argument("-r", "--remove", action="store_true",
173
162
help="Remove client")
174
parser.add_argument("-c", "--checker", dest="commands",
175
action="append", default=[],
176
metavar="COMMAND", type=command.SetChecker,
163
parser.add_argument("-c", "--checker",
177
164
help="Set checker command for client")
178
parser.add_argument(
179
"-t", "--timeout", dest="commands", action="append",
180
default=[], metavar="TIME",
181
type=command.SetTimeout.argparse(string_to_delta),
182
help="Set timeout for client")
183
parser.add_argument(
184
"--extended-timeout", dest="commands", action="append",
185
default=[], metavar="TIME",
186
type=command.SetExtendedTimeout.argparse(string_to_delta),
187
help="Set extended timeout for client")
188
parser.add_argument(
189
"-i", "--interval", dest="commands", action="append",
190
default=[], metavar="TIME",
191
type=command.SetInterval.argparse(string_to_delta),
192
help="Set checker interval for client")
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")
193
171
approve_deny_default = parser.add_mutually_exclusive_group()
194
172
approve_deny_default.add_argument(
195
"--approve-by-default", dest="commands",
196
action="append_const", default=[],
197
const=command.ApproveByDefault(),
173
"--approve-by-default", action="store_true",
174
default=None, dest="approved_by_default",
198
175
help="Set client to be approved by default")
199
176
approve_deny_default.add_argument(
200
"--deny-by-default", dest="commands",
201
action="append_const", default=[],
202
const=command.DenyByDefault(),
177
"--deny-by-default", action="store_false",
178
dest="approved_by_default",
203
179
help="Set client to be denied by default")
204
parser.add_argument(
205
"--approval-delay", dest="commands", action="append",
206
default=[], metavar="TIME",
207
type=command.SetApprovalDelay.argparse(string_to_delta),
208
help="Set delay before client approve/deny")
209
parser.add_argument(
210
"--approval-duration", dest="commands", action="append",
211
default=[], metavar="TIME",
212
type=command.SetApprovalDuration.argparse(string_to_delta),
213
help="Set duration of one client approval")
214
parser.add_argument("-H", "--host", dest="commands",
215
action="append", default=[], metavar="STRING",
216
type=command.SetHost,
217
help="Set host for client")
218
parser.add_argument(
219
"-s", "--secret", dest="commands", action="append",
220
default=[], metavar="FILENAME",
221
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
222
help="Set password blob (file) for client")
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")
223
188
approve_deny = parser.add_mutually_exclusive_group()
224
189
approve_deny.add_argument(
225
"-A", "--approve", dest="commands", action="append_const",
226
default=[], const=command.Approve(),
190
"-A", "--approve", action="store_true",
227
191
help="Approve any current client request")
228
approve_deny.add_argument("-D", "--deny", dest="commands",
229
action="append_const", default=[],
230
const=command.Deny(),
192
approve_deny.add_argument("-D", "--deny", action="store_true",
231
193
help="Deny any current client request")
232
194
parser.add_argument("--debug", action="store_true",
233
195
help="Debug mode (show D-Bus commands)")
424
386
"""Apply additional restrictions on options, not expressible in
425
387
argparse"""
426
388
427
def has_commands(options, commands=None):
428
if commands is None:
429
commands = (command.Enable,
430
command.Disable,
431
command.BumpTimeout,
432
command.StartChecker,
433
command.StopChecker,
434
command.IsEnabled,
435
command.Remove,
436
command.SetChecker,
437
command.SetTimeout,
438
command.SetExtendedTimeout,
439
command.SetInterval,
440
command.ApproveByDefault,
441
command.DenyByDefault,
442
command.SetApprovalDelay,
443
command.SetApprovalDuration,
444
command.SetHost,
445
command.SetSecret,
446
command.Approve,
447
command.Deny)
448
return any(isinstance(cmd, commands)
449
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))
450
408
451
if has_commands(options) and not (options.client or options.all):
409
if has_actions(options) and not (options.client or options.all):
452
410
parser.error("Options require clients names or --all.")
453
if options.verbose and has_commands(options):
411
if options.verbose and has_actions(options):
454
412
parser.error("--verbose can only be used alone.")
455
if (has_commands(options, (command.DumpJSON,))
456
and (options.verbose or len(options.commands) > 1)):
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
457
415
parser.error("--dump-json can only be used alone.")
458
if options.all and not has_commands(options):
416
if options.all and not has_actions(options):
459
417
parser.error("--all requires an action.")
460
if (has_commands(options, (command.IsEnabled,))
461
and len(options.client) > 1):
418
if options.is_enabled and len(options.client) > 1:
462
419
parser.error("--is-enabled requires exactly one client")
463
if (len(options.commands) > 1
464
and has_commands(options, (command.Remove,))
465
and not has_commands(options, (command.Deny,))):
466
parser.error("--remove can only be combined with --deny")
467
468
469
class dbus(object):
470
471
class SystemBus(object):
472
473
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
def get_managed_objects(self, busname, objectpath):
475
return self.call_method("GetManagedObjects", busname,
476
objectpath,
477
self.object_manager_iface)
478
479
properties_iface = "org.freedesktop.DBus.Properties"
480
def set_property(self, busname, objectpath, interface, key,
481
value):
482
self.call_method("Set", busname, objectpath,
483
self.properties_iface, interface, key,
484
value)
485
486
487
class MandosBus(SystemBus):
488
busname_domain = "se.recompile"
489
busname = busname_domain + ".Mandos"
490
server_path = "/"
491
server_interface = busname_domain + ".Mandos"
492
client_interface = busname_domain + ".Mandos.Client"
493
del busname_domain
494
495
def get_clients_and_properties(self):
496
managed_objects = self.get_managed_objects(
497
self.busname, self.server_path)
498
return {objpath: properties[self.client_interface]
499
for objpath, properties in managed_objects.items()
500
if self.client_interface in properties}
501
502
def set_client_property(self, objectpath, key, value):
503
return self.set_property(self.busname, objectpath,
504
self.client_interface, key,
505
value)
506
507
def call_client_method(self, objectpath, method, *args):
508
return self.call_method(method, self.busname, objectpath,
509
self.client_interface, *args)
510
511
def call_server_method(self, method, *args):
512
return self.call_method(method, self.busname,
513
self.server_path,
514
self.server_interface, *args)
515
516
class Error(Exception):
517
pass
518
519
class ConnectFailed(Error):
520
pass
521
522
523
class dbus_python_adapter(object):
524
525
class SystemBus(dbus.MandosBus):
526
"""Use dbus-python"""
527
528
def __init__(self, module=dbus_python):
529
self.dbus_python = module
530
self.bus = self.dbus_python.SystemBus()
531
532
@contextlib.contextmanager
533
def convert_exception(self, exception_class=dbus.Error):
534
try:
535
yield
536
except self.dbus_python.exceptions.DBusException as e:
537
# This does what "raise from" would do
538
exc = exception_class(*e.args)
539
exc.__cause__ = e
540
raise exc
541
542
def call_method(self, methodname, busname, objectpath,
543
interface, *args):
544
proxy_object = self.get_object(busname, objectpath)
545
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
546
interface, methodname,
547
", ".join(repr(a) for a in args))
548
method = getattr(proxy_object, methodname)
549
with self.convert_exception():
550
with dbus_python_adapter.SilenceLogger(
551
"dbus.proxies"):
552
value = method(*args, dbus_interface=interface)
553
return self.type_filter(value)
554
555
def get_object(self, busname, objectpath):
556
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
557
busname, objectpath)
558
with self.convert_exception(dbus.ConnectFailed):
559
return self.bus.get_object(busname, objectpath)
560
561
def type_filter(self, value):
562
"""Convert the most bothersome types to Python types"""
563
if isinstance(value, self.dbus_python.Boolean):
564
return bool(value)
565
if isinstance(value, self.dbus_python.ObjectPath):
566
return str(value)
567
# Also recurse into dictionaries
568
if isinstance(value, self.dbus_python.Dictionary):
569
return {self.type_filter(key):
570
self.type_filter(subval)
571
for key, subval in value.items()}
572
return value
573
574
575
class SilenceLogger(object):
576
"Simple context manager to silence a particular logger"
577
def __init__(self, loggername):
578
self.logger = logging.getLogger(loggername)
579
580
def __enter__(self):
581
self.logger.addFilter(self.nullfilter)
582
583
class NullFilter(logging.Filter):
584
def filter(self, record):
585
return False
586
587
nullfilter = NullFilter()
588
589
def __exit__(self, exc_type, exc_val, exc_tb):
590
self.logger.removeFilter(self.nullfilter)
591
592
593
class CachingBus(SystemBus):
594
"""A caching layer for dbus_python_adapter.SystemBus"""
595
def __init__(self, *args, **kwargs):
596
self.object_cache = {}
597
super(dbus_python_adapter.CachingBus,
598
self).__init__(*args, **kwargs)
599
def get_object(self, busname, objectpath):
600
try:
601
return self.object_cache[(busname, objectpath)]
602
except KeyError:
603
new_object = super(
604
dbus_python_adapter.CachingBus,
605
self).get_object(busname, objectpath)
606
self.object_cache[(busname, objectpath)] = new_object
607
return new_object
608
609
610
class pydbus_adapter(object):
611
class SystemBus(dbus.MandosBus):
612
def __init__(self, module=pydbus):
613
self.pydbus = module
614
self.bus = self.pydbus.SystemBus()
615
616
@contextlib.contextmanager
617
def convert_exception(self, exception_class=dbus.Error):
618
try:
619
yield
620
except gi.repository.GLib.Error as e:
621
# This does what "raise from" would do
622
exc = exception_class(*e.args)
623
exc.__cause__ = e
624
raise exc
625
626
def call_method(self, methodname, busname, objectpath,
627
interface, *args):
628
proxy_object = self.get(busname, objectpath)
629
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
630
interface, methodname,
631
", ".join(repr(a) for a in args))
632
method = getattr(proxy_object[interface], methodname)
633
with self.convert_exception():
634
return method(*args)
635
636
def get(self, busname, objectpath):
637
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
638
busname, objectpath)
639
with self.convert_exception(dbus.ConnectFailed):
640
if sys.version_info.major <= 2:
641
with warnings.catch_warnings():
642
warnings.filterwarnings(
643
"ignore", "", DeprecationWarning,
644
r"^xml\.etree\.ElementTree$")
645
return self.bus.get(busname, objectpath)
646
else:
647
return self.bus.get(busname, objectpath)
648
649
def set_property(self, busname, objectpath, interface, key,
650
value):
651
proxy_object = self.get(busname, objectpath)
652
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
653
objectpath, self.properties_iface, interface,
654
key, value)
655
setattr(proxy_object[interface], key, value)
656
657
class CachingBus(SystemBus):
658
"""A caching layer for pydbus_adapter.SystemBus"""
659
def __init__(self, *args, **kwargs):
660
self.object_cache = {}
661
super(pydbus_adapter.CachingBus,
662
self).__init__(*args, **kwargs)
663
def get(self, busname, objectpath):
664
try:
665
return self.object_cache[(busname, objectpath)]
666
except KeyError:
667
new_object = (super(pydbus_adapter.CachingBus, self)
668
.get(busname, objectpath))
669
self.object_cache[(busname, objectpath)] = new_object
670
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)
671
472
672
473
673
474
def commands_from_options(options):
674
475
675
commands = list(options.commands)
676
677
def find_cmd(cmd, commands):
678
i = 0
679
for i, c in enumerate(commands):
680
if isinstance(c, cmd):
681
return i
682
return i+1
683
684
# If command.Remove is present, move any instances of command.Deny
685
# to occur ahead of command.Remove.
686
index_of_remove = find_cmd(command.Remove, commands)
687
before_remove = commands[:index_of_remove]
688
after_remove = commands[index_of_remove:]
689
cleaned_after = []
690
for cmd in after_remove:
691
if isinstance(cmd, command.Deny):
692
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())
693
511
else:
694
cleaned_after.append(cmd)
695
if cleaned_after != after_remove:
696
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))
697
540
698
541
# If no command option has been given, show table of clients,
699
542
# optionally verbosely
708
551
709
552
class Base(object):
710
553
"""Abstract base class for commands"""
711
def run(self, clients, bus=None):
554
def run(self, clients, bus=None, mandos=None):
712
555
"""Normal commands should implement run_on_one_client(),
713
556
but commands which want to operate on all clients at the same time can
714
557
override this run() method instead.
715
558
"""
716
self.bus = bus
717
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)
718
564
self.run_on_one_client(client, properties)
719
565
720
566
721
567
class IsEnabled(Base):
722
def run(self, clients, bus=None):
723
properties = next(iter(clients.values()))
724
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):
725
571
sys.exit(0)
726
572
sys.exit(1)
573
def is_enabled(self, client, properties):
574
return properties["Enabled"]
727
575
728
576
729
577
class Approve(Base):
730
578
def run_on_one_client(self, client, properties):
731
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)
732
584
733
585
734
586
class Deny(Base):
735
587
def run_on_one_client(self, client, properties):
736
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)
737
593
738
594
739
595
class Remove(Base):
740
def run(self, clients, bus):
741
for clientpath in frozenset(clients.keys()):
742
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__)
743
602
744
603
745
604
class Output(Base):
755
614
756
615
757
616
class DumpJSON(Output):
758
def run(self, clients, bus=None):
759
data = {properties["Name"]:
760
{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])
761
620
for key in self.all_keywords}
762
for properties in clients.values()}
621
for client in clients.values()}
763
622
print(json.dumps(data, indent=4, separators=(',', ': ')))
764
623
624
@staticmethod
625
def dbus_boolean_to_bool(value):
626
if isinstance(value, dbus.Boolean):
627
value = bool(value)
628
return value
629
765
630
766
631
class PrintTable(Output):
767
632
def __init__(self, verbose=False):
768
633
self.verbose = verbose
769
634
770
def run(self, clients, bus=None):
635
def run(self, clients, bus=None, mandos=None):
771
636
default_keywords = ("Name", "Enabled", "Timeout",
772
637
"LastCheckedOK")
773
638
keywords = default_keywords
834
699
835
700
@classmethod
836
701
def valuetostring(cls, value, keyword):
837
if isinstance(value, bool):
702
if isinstance(value, dbus.Boolean):
838
703
return "Yes" if value else "No"
839
704
if keyword in ("Timeout", "Interval", "ApprovalDelay",
840
705
"ApprovalDuration", "ExtendedTimeout"):
863
728
class PropertySetter(Base):
864
729
"Abstract class for Actions for setting one client property"
865
730
866
def run_on_one_client(self, client, properties=None):
731
def run_on_one_client(self, client, properties):
867
732
"""Set the Client's D-Bus property"""
868
self.bus.set_client_property(client, self.propname,
869
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)
870
743
871
744
@property
872
745
def propname(self):
875
748
876
749
class Enable(PropertySetter):
877
750
propname = "Enabled"
878
value_to_set = True
751
value_to_set = dbus.Boolean(True)
879
752
880
753
881
754
class Disable(PropertySetter):
882
755
propname = "Enabled"
883
value_to_set = False
756
value_to_set = dbus.Boolean(False)
884
757
885
758
886
759
class BumpTimeout(PropertySetter):
890
763
891
764
class StartChecker(PropertySetter):
892
765
propname = "CheckerRunning"
893
value_to_set = True
766
value_to_set = dbus.Boolean(True)
894
767
895
768
896
769
class StopChecker(PropertySetter):
897
770
propname = "CheckerRunning"
898
value_to_set = False
771
value_to_set = dbus.Boolean(False)
899
772
900
773
901
774
class ApproveByDefault(PropertySetter):
902
775
propname = "ApprovedByDefault"
903
value_to_set = True
776
value_to_set = dbus.Boolean(True)
904
777
905
778
906
779
class DenyByDefault(PropertySetter):
907
780
propname = "ApprovedByDefault"
908
value_to_set = False
781
value_to_set = dbus.Boolean(False)
909
782
910
783
911
784
class PropertySetterValue(PropertySetter):
914
787
def __init__(self, value):
915
788
self.value_to_set = value
916
789
917
@classmethod
918
def argparse(cls, argtype):
919
def cmdtype(arg):
920
return cls(argtype(arg))
921
return cmdtype
922
790
923
791
class SetChecker(PropertySetterValue):
924
792
propname = "Checker"
1012
880
"output"))
1013
881
1014
882
1015
class Unique(object):
1016
"""Class for objects which exist only to be unique objects, since
1017
unittest.mock.sentinel only exists in Python 3.3"""
1018
1019
1020
883
class Test_string_to_delta(TestCaseWithAssertLogs):
1021
884
# Just test basic RFC 3339 functionality here, the doc string for
1022
885
# rfc3339_duration_to_delta() already has more comprehensive
1023
# tests, which are run by doctest.
886
# tests, which is run by doctest.
1024
887
1025
888
def test_rfc3339_zero_seconds(self):
1026
889
self.assertEqual(datetime.timedelta(),
1050
913
1051
914
def test_actions_requires_client_or_all(self):
1052
915
for action, value in self.actions.items():
1053
args = self.actionargs(action, value)
916
options = self.parser.parse_args()
917
setattr(options, action, value)
1054
918
with self.assertParseError():
1055
self.parse_args(args)
919
self.check_option_syntax(options)
1056
920
1057
# This mostly corresponds to the definition from has_commands() in
921
# This mostly corresponds to the definition from has_actions() in
1058
922
# check_option_syntax()
1059
923
actions = {
1060
"--enable": None,
1061
"--disable": None,
1062
"--bump-timeout": None,
1063
"--start-checker": None,
1064
"--stop-checker": None,
1065
"--is-enabled": None,
1066
"--remove": None,
1067
"--checker": "x",
1068
"--timeout": "PT0S",
1069
"--extended-timeout": "PT0S",
1070
"--interval": "PT0S",
1071
"--approve-by-default": None,
1072
"--deny-by-default": None,
1073
"--approval-delay": "PT0S",
1074
"--approval-duration": "PT0S",
1075
"--host": "hostname",
1076
"--secret": "/dev/null",
1077
"--approve": None,
1078
"--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,
1079
945
}
1080
946
1081
@staticmethod
1082
def actionargs(action, value, *args):
1083
if value is not None:
1084
return [action, value] + list(args)
1085
else:
1086
return [action] + list(args)
1087
1088
947
@contextlib.contextmanager
1089
948
def assertParseError(self):
1090
949
with self.assertRaises(SystemExit) as e:
1095
954
# /argparse.html#exiting-methods
1096
955
self.assertEqual(2, e.exception.code)
1097
956
1098
def parse_args(self, args):
1099
options = self.parser.parse_args(args)
1100
check_option_syntax(self.parser, options)
1101
1102
957
@staticmethod
1103
958
@contextlib.contextmanager
1104
959
def redirect_stderr_to_devnull():
1115
970
1116
971
def test_actions_all_conflicts_with_verbose(self):
1117
972
for action, value in self.actions.items():
1118
args = self.actionargs(action, value, "--all",
1119
"--verbose")
973
options = self.parser.parse_args()
974
setattr(options, action, value)
975
options.all = True
976
options.verbose = True
1120
977
with self.assertParseError():
1121
self.parse_args(args)
978
self.check_option_syntax(options)
1122
979
1123
980
def test_actions_with_client_conflicts_with_verbose(self):
1124
981
for action, value in self.actions.items():
1125
args = self.actionargs(action, value, "--verbose",
1126
"client")
982
options = self.parser.parse_args()
983
setattr(options, action, value)
984
options.verbose = True
985
options.client = ["client"]
1127
986
with self.assertParseError():
1128
self.parse_args(args)
987
self.check_option_syntax(options)
1129
988
1130
989
def test_dump_json_conflicts_with_verbose(self):
1131
args = ["--dump-json", "--verbose"]
990
options = self.parser.parse_args()
991
options.dump_json = True
992
options.verbose = True
1132
993
with self.assertParseError():
1133
self.parse_args(args)
994
self.check_option_syntax(options)
1134
995
1135
996
def test_dump_json_conflicts_with_action(self):
1136
997
for action, value in self.actions.items():
1137
args = self.actionargs(action, value, "--dump-json")
998
options = self.parser.parse_args()
999
setattr(options, action, value)
1000
options.dump_json = True
1138
1001
with self.assertParseError():
1139
self.parse_args(args)
1002
self.check_option_syntax(options)
1140
1003
1141
1004
def test_all_can_not_be_alone(self):
1142
args = ["--all"]
1005
options = self.parser.parse_args()
1006
options.all = True
1143
1007
with self.assertParseError():
1144
self.parse_args(args)
1008
self.check_option_syntax(options)
1145
1009
1146
1010
def test_all_is_ok_with_any_action(self):
1147
1011
for action, value in self.actions.items():
1148
args = self.actionargs(action, value, "--all")
1149
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)
1150
1016
1151
1017
def test_any_action_is_ok_with_one_client(self):
1152
1018
for action, value in self.actions.items():
1153
args = self.actionargs(action, value, "client")
1154
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)
1155
1023
1156
1024
def test_one_client_with_all_actions_except_is_enabled(self):
1025
options = self.parser.parse_args()
1157
1026
for action, value in self.actions.items():
1158
if action == "--is-enabled":
1027
if action == "is_enabled":
1159
1028
continue
1160
args = self.actionargs(action, value, "client")
1161
self.parse_args(args)
1029
setattr(options, action, value)
1030
options.client = ["client"]
1031
self.check_option_syntax(options)
1162
1032
1163
1033
def test_two_clients_with_all_actions_except_is_enabled(self):
1034
options = self.parser.parse_args()
1164
1035
for action, value in self.actions.items():
1165
if action == "--is-enabled":
1036
if action == "is_enabled":
1166
1037
continue
1167
args = self.actionargs(action, value, "client1",
1168
"client2")
1169
self.parse_args(args)
1038
setattr(options, action, value)
1039
options.client = ["client1", "client2"]
1040
self.check_option_syntax(options)
1170
1041
1171
1042
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1172
1043
for action, value in self.actions.items():
1173
if action == "--is-enabled":
1044
if action == "is_enabled":
1174
1045
continue
1175
args = self.actionargs(action, value, "client1",
1176
"client2")
1177
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)
1178
1050
1179
1051
def test_is_enabled_fails_without_client(self):
1180
args = ["--is-enabled"]
1052
options = self.parser.parse_args()
1053
options.is_enabled = True
1181
1054
with self.assertParseError():
1182
self.parse_args(args)
1055
self.check_option_syntax(options)
1183
1056
1184
1057
def test_is_enabled_fails_with_two_clients(self):
1185
args = ["--is-enabled", "client1", "client2"]
1058
options = self.parser.parse_args()
1059
options.is_enabled = True
1060
options.client = ["client1", "client2"]
1186
1061
with self.assertParseError():
1187
self.parse_args(args)
1062
self.check_option_syntax(options)
1188
1063
1189
1064
def test_remove_can_only_be_combined_with_action_deny(self):
1190
1065
for action, value in self.actions.items():
1191
if action in {"--remove", "--deny"}:
1066
if action in {"remove", "deny"}:
1192
1067
continue
1193
args = self.actionargs(action, value, "--all",
1194
"--remove")
1068
options = self.parser.parse_args()
1069
setattr(options, action, value)
1070
options.all = True
1071
options.remove = True
1195
1072
with self.assertParseError():
1196
self.parse_args(args)
1197
1198
1199
class Test_dbus_exceptions(unittest.TestCase):
1200
1201
def test_dbus_ConnectFailed_is_Error(self):
1202
with self.assertRaises(dbus.Error):
1203
raise dbus.ConnectFailed()
1204
1205
1206
class Test_dbus_MandosBus(unittest.TestCase):
1207
1208
class MockMandosBus(dbus.MandosBus):
1209
def __init__(self):
1210
self._name = "se.recompile.Mandos"
1211
self._server_path = "/"
1212
self._server_interface = "se.recompile.Mandos"
1213
self._client_interface = "se.recompile.Mandos.Client"
1214
self.calls = []
1215
self.call_method_return = Unique()
1216
1217
def call_method(self, methodname, busname, objectpath,
1218
interface, *args):
1219
self.calls.append((methodname, busname, objectpath,
1220
interface, args))
1221
return self.call_method_return
1222
1223
def setUp(self):
1224
self.bus = self.MockMandosBus()
1225
1226
def test_set_client_property(self):
1227
self.bus.set_client_property("objectpath", "key", "value")
1228
expected_call = ("Set", self.bus._name, "objectpath",
1229
"org.freedesktop.DBus.Properties",
1230
(self.bus._client_interface, "key", "value"))
1231
self.assertIn(expected_call, self.bus.calls)
1232
1233
def test_call_client_method(self):
1234
ret = self.bus.call_client_method("objectpath", "methodname")
1235
self.assertIs(self.bus.call_method_return, ret)
1236
expected_call = ("methodname", self.bus._name, "objectpath",
1237
self.bus._client_interface, ())
1238
self.assertIn(expected_call, self.bus.calls)
1239
1240
def test_call_client_method_with_args(self):
1241
args = (Unique(), Unique())
1242
ret = self.bus.call_client_method("objectpath", "methodname",
1243
*args)
1244
self.assertIs(self.bus.call_method_return, ret)
1245
expected_call = ("methodname", self.bus._name, "objectpath",
1246
self.bus._client_interface,
1247
(args[0], args[1]))
1248
self.assertIn(expected_call, self.bus.calls)
1249
1250
def test_get_clients_and_properties(self):
1251
managed_objects = {
1252
"objectpath": {
1253
self.bus._client_interface: {
1254
"key": "value",
1255
"bool": True,
1256
},
1257
"irrelevant_interface": {
1258
"key": "othervalue",
1259
"bool": False,
1260
},
1261
},
1262
"other_objectpath": {
1263
"other_irrelevant_interface": {
1264
"key": "value 3",
1265
"bool": None,
1266
},
1267
},
1268
}
1269
expected_clients_and_properties = {
1270
"objectpath": {
1271
"key": "value",
1272
"bool": True,
1273
}
1274
}
1275
self.bus.call_method_return = managed_objects
1276
ret = self.bus.get_clients_and_properties()
1277
self.assertDictEqual(expected_clients_and_properties, ret)
1278
expected_call = ("GetManagedObjects", self.bus._name,
1279
self.bus._server_path,
1280
"org.freedesktop.DBus.ObjectManager", ())
1281
self.assertIn(expected_call, self.bus.calls)
1282
1283
def test_call_server_method(self):
1284
ret = self.bus.call_server_method("methodname")
1285
self.assertIs(self.bus.call_method_return, ret)
1286
expected_call = ("methodname", self.bus._name,
1287
self.bus._server_path,
1288
self.bus._server_interface, ())
1289
self.assertIn(expected_call, self.bus.calls)
1290
1291
def test_call_server_method_with_args(self):
1292
args = (Unique(), Unique())
1293
ret = self.bus.call_server_method("methodname", *args)
1294
self.assertIs(self.bus.call_method_return, ret)
1295
expected_call = ("methodname", self.bus._name,
1296
self.bus._server_path,
1297
self.bus._server_interface,
1298
(args[0], args[1]))
1299
self.assertIn(expected_call, self.bus.calls)
1300
1301
1302
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1303
1304
def MockDBusPython_func(self, func):
1305
class mock_dbus_python(object):
1306
"""mock dbus-python module"""
1307
class exceptions(object):
1308
"""Pseudo-namespace"""
1309
class DBusException(Exception):
1310
pass
1311
class SystemBus(object):
1312
@staticmethod
1313
def get_object(busname, objectpath):
1314
DBusObject = collections.namedtuple(
1315
"DBusObject", ("methodname",))
1316
def method(*args, **kwargs):
1317
self.assertEqual({"dbus_interface":
1318
"interface"},
1319
kwargs)
1320
return func(*args)
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1323
def __init__(self, value):
1324
self.value = bool(value)
1325
def __bool__(self):
1326
return self.value
1327
if sys.version_info.major == 2:
1328
__nonzero__ = __bool__
1329
class ObjectPath(str):
1330
pass
1331
class Dictionary(dict):
1332
pass
1333
return mock_dbus_python
1334
1335
def call_method(self, bus, methodname, busname, objectpath,
1336
interface, *args):
1337
with self.assertLogs(log, logging.DEBUG):
1338
return bus.call_method(methodname, busname, objectpath,
1339
interface, *args)
1340
1341
def test_call_method_returns(self):
1342
expected_method_return = Unique()
1343
method_args = (Unique(), Unique())
1344
def func(*args):
1345
self.assertEqual(len(method_args), len(args))
1346
for marg, arg in zip(method_args, args):
1347
self.assertIs(marg, arg)
1348
return expected_method_return
1349
mock_dbus_python = self.MockDBusPython_func(func)
1350
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1351
ret = self.call_method(bus, "methodname", "busname",
1352
"objectpath", "interface",
1353
*method_args)
1354
self.assertIs(ret, expected_method_return)
1355
1356
def test_call_method_filters_bool_true(self):
1357
def func():
1358
return method_return
1359
mock_dbus_python = self.MockDBusPython_func(func)
1360
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1361
method_return = mock_dbus_python.Boolean(True)
1362
ret = self.call_method(bus, "methodname", "busname",
1363
"objectpath", "interface")
1364
self.assertTrue(ret)
1365
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1366
1367
def test_call_method_filters_bool_false(self):
1368
def func():
1369
return method_return
1370
mock_dbus_python = self.MockDBusPython_func(func)
1371
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1372
method_return = mock_dbus_python.Boolean(False)
1373
ret = self.call_method(bus, "methodname", "busname",
1374
"objectpath", "interface")
1375
self.assertFalse(ret)
1376
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1377
1378
def test_call_method_filters_objectpath(self):
1379
def func():
1380
return method_return
1381
mock_dbus_python = self.MockDBusPython_func(func)
1382
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1383
method_return = mock_dbus_python.ObjectPath("objectpath")
1384
ret = self.call_method(bus, "methodname", "busname",
1385
"objectpath", "interface")
1386
self.assertEqual("objectpath", ret)
1387
self.assertIsNot("objectpath", ret)
1388
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1389
1390
def test_call_method_filters_booleans_in_dict(self):
1391
def func():
1392
return method_return
1393
mock_dbus_python = self.MockDBusPython_func(func)
1394
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1395
method_return = mock_dbus_python.Dictionary(
1396
{mock_dbus_python.Boolean(True):
1397
mock_dbus_python.Boolean(False),
1398
mock_dbus_python.Boolean(False):
1399
mock_dbus_python.Boolean(True)})
1400
ret = self.call_method(bus, "methodname", "busname",
1401
"objectpath", "interface")
1402
expected_method_return = {True: False,
1403
False: True}
1404
self.assertEqual(expected_method_return, ret)
1405
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1406
1407
def test_call_method_filters_objectpaths_in_dict(self):
1408
def func():
1409
return method_return
1410
mock_dbus_python = self.MockDBusPython_func(func)
1411
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1412
method_return = mock_dbus_python.Dictionary(
1413
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1414
mock_dbus_python.ObjectPath("objectpath_value_1"),
1415
mock_dbus_python.ObjectPath("objectpath_key_2"):
1416
mock_dbus_python.ObjectPath("objectpath_value_2")})
1417
ret = self.call_method(bus, "methodname", "busname",
1418
"objectpath", "interface")
1419
expected_method_return = {str(key): str(value)
1420
for key, value in
1421
method_return.items()}
1422
self.assertEqual(expected_method_return, ret)
1423
self.assertIsInstance(ret, dict)
1424
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1425
1426
def test_call_method_filters_dict_in_dict(self):
1427
def func():
1428
return method_return
1429
mock_dbus_python = self.MockDBusPython_func(func)
1430
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1431
method_return = mock_dbus_python.Dictionary(
1432
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1433
"key12": "value12"}),
1434
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1435
"key22": "value22"})})
1436
ret = self.call_method(bus, "methodname", "busname",
1437
"objectpath", "interface")
1438
expected_method_return = {
1439
"key1": {"key11": "value11",
1440
"key12": "value12"},
1441
"key2": {"key21": "value21",
1442
"key22": "value22"},
1443
}
1444
self.assertEqual(expected_method_return, ret)
1445
self.assertIsInstance(ret, dict)
1446
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1447
for key, value in ret.items():
1448
self.assertIsInstance(value, dict)
1449
self.assertEqual(expected_method_return[key], value)
1450
self.assertNotIsInstance(value,
1451
mock_dbus_python.Dictionary)
1452
1453
def test_call_method_filters_dict_three_deep(self):
1454
def func():
1455
return method_return
1456
mock_dbus_python = self.MockDBusPython_func(func)
1457
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1458
method_return = mock_dbus_python.Dictionary(
1459
{"key1":
1460
mock_dbus_python.Dictionary(
1461
{"key2":
1462
mock_dbus_python.Dictionary(
1463
{"key3":
1464
mock_dbus_python.Boolean(True),
1465
}),
1466
}),
1467
})
1468
ret = self.call_method(bus, "methodname", "busname",
1469
"objectpath", "interface")
1470
expected_method_return = {"key1": {"key2": {"key3": True}}}
1471
self.assertEqual(expected_method_return, ret)
1472
self.assertIsInstance(ret, dict)
1473
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1474
self.assertIsInstance(ret["key1"], dict)
1475
self.assertNotIsInstance(ret["key1"],
1476
mock_dbus_python.Dictionary)
1477
self.assertIsInstance(ret["key1"]["key2"], dict)
1478
self.assertNotIsInstance(ret["key1"]["key2"],
1479
mock_dbus_python.Dictionary)
1480
self.assertTrue(ret["key1"]["key2"]["key3"])
1481
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1482
mock_dbus_python.Boolean)
1483
1484
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):
1485
1117
dbus_logger = logging.getLogger("dbus.proxies")
1486
1118
1487
def func():
1488
dbus_logger.error("Test")
1489
raise mock_dbus_python.exceptions.DBusException()
1490
1491
mock_dbus_python = self.MockDBusPython_func(func)
1492
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")
1493
1123
1494
1124
class CountingHandler(logging.Handler):
1495
1125
count = 0
1501
1131
dbus_logger.addHandler(counting_handler)
1502
1132
1503
1133
try:
1504
with self.assertRaises(dbus.Error) as e:
1505
self.call_method(bus, "methodname", "busname",
1506
"objectpath", "interface")
1134
with self.assertLogs(log, logging.CRITICAL) as watcher:
1135
with self.assertRaises(SystemExit) as e:
1136
get_managed_objects(ObjectManagerFailingStub())
1507
1137
finally:
1508
1138
dbus_logger.removeFilter(counting_handler)
1509
1139
1510
self.assertNotIsInstance(e, dbus.ConnectFailed)
1511
1512
1140
# Make sure the dbus logger was suppressed
1513
1141
self.assertEqual(0, counting_handler.count)
1514
1142
1515
def test_get_object_converts_to_correct_exception(self):
1516
bus = dbus_python_adapter.SystemBus(
1517
self.fake_dbus_python_raises_exception_on_connect)
1518
with self.assertRaises(dbus.ConnectFailed):
1519
self.call_method(bus, "methodname", "busname",
1520
"objectpath", "interface")
1521
1522
class fake_dbus_python_raises_exception_on_connect(object):
1523
"""fake dbus-python module"""
1524
class exceptions(object):
1525
"""Pseudo-namespace"""
1526
class DBusException(Exception):
1527
pass
1528
1529
@classmethod
1530
def SystemBus(cls):
1531
def get_object(busname, objectpath):
1532
raise cls.exceptions.DBusException()
1533
Bus = collections.namedtuple("Bus", ["get_object"])
1534
return Bus(get_object=get_object)
1535
1536
1537
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1538
class mock_dbus_python(object):
1539
"""mock dbus-python modules"""
1540
class SystemBus(object):
1541
@staticmethod
1542
def get_object(busname, objectpath):
1543
return Unique()
1544
1545
def setUp(self):
1546
self.bus = dbus_python_adapter.CachingBus(
1547
self.mock_dbus_python)
1548
1549
def test_returns_distinct_objectpaths(self):
1550
obj1 = self.bus.get_object("busname", "objectpath1")
1551
self.assertIsInstance(obj1, Unique)
1552
obj2 = self.bus.get_object("busname", "objectpath2")
1553
self.assertIsInstance(obj2, Unique)
1554
self.assertIsNot(obj1, obj2)
1555
1556
def test_returns_distinct_busnames(self):
1557
obj1 = self.bus.get_object("busname1", "objectpath")
1558
self.assertIsInstance(obj1, Unique)
1559
obj2 = self.bus.get_object("busname2", "objectpath")
1560
self.assertIsInstance(obj2, Unique)
1561
self.assertIsNot(obj1, obj2)
1562
1563
def test_returns_distinct_both(self):
1564
obj1 = self.bus.get_object("busname1", "objectpath")
1565
self.assertIsInstance(obj1, Unique)
1566
obj2 = self.bus.get_object("busname2", "objectpath")
1567
self.assertIsInstance(obj2, Unique)
1568
self.assertIsNot(obj1, obj2)
1569
1570
def test_returns_same(self):
1571
obj1 = self.bus.get_object("busname", "objectpath")
1572
self.assertIsInstance(obj1, Unique)
1573
obj2 = self.bus.get_object("busname", "objectpath")
1574
self.assertIsInstance(obj2, Unique)
1575
self.assertIs(obj1, obj2)
1576
1577
def test_returns_same_old(self):
1578
obj1 = self.bus.get_object("busname1", "objectpath1")
1579
self.assertIsInstance(obj1, Unique)
1580
obj2 = self.bus.get_object("busname2", "objectpath2")
1581
self.assertIsInstance(obj2, Unique)
1582
obj1b = self.bus.get_object("busname1", "objectpath1")
1583
self.assertIsInstance(obj1b, Unique)
1584
self.assertIsNot(obj1, obj2)
1585
self.assertIsNot(obj2, obj1b)
1586
self.assertIs(obj1, obj1b)
1587
1588
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1590
1591
def Stub_pydbus_func(self, func):
1592
class stub_pydbus(object):
1593
"""stub pydbus module"""
1594
class SystemBus(object):
1595
@staticmethod
1596
def get(busname, objectpath):
1597
DBusObject = collections.namedtuple(
1598
"DBusObject", ("methodname",))
1599
return {"interface":
1600
DBusObject(methodname=func)}
1601
return stub_pydbus
1602
1603
def call_method(self, bus, methodname, busname, objectpath,
1604
interface, *args):
1605
with self.assertLogs(log, logging.DEBUG):
1606
return bus.call_method(methodname, busname, objectpath,
1607
interface, *args)
1608
1609
def test_call_method_returns(self):
1610
expected_method_return = Unique()
1611
method_args = (Unique(), Unique())
1612
def func(*args):
1613
self.assertEqual(len(method_args), len(args))
1614
for marg, arg in zip(method_args, args):
1615
self.assertIs(marg, arg)
1616
return expected_method_return
1617
stub_pydbus = self.Stub_pydbus_func(func)
1618
bus = pydbus_adapter.SystemBus(stub_pydbus)
1619
ret = self.call_method(bus, "methodname", "busname",
1620
"objectpath", "interface",
1621
*method_args)
1622
self.assertIs(ret, expected_method_return)
1623
1624
def test_call_method_handles_exception(self):
1625
dbus_logger = logging.getLogger("dbus.proxies")
1626
1627
def func():
1628
raise gi.repository.GLib.Error()
1629
1630
stub_pydbus = self.Stub_pydbus_func(func)
1631
bus = pydbus_adapter.SystemBus(stub_pydbus)
1632
1633
with self.assertRaises(dbus.Error) as e:
1634
self.call_method(bus, "methodname", "busname",
1635
"objectpath", "interface")
1636
1637
self.assertNotIsInstance(e, dbus.ConnectFailed)
1638
1639
def test_get_converts_to_correct_exception(self):
1640
bus = pydbus_adapter.SystemBus(
1641
self.fake_pydbus_raises_exception_on_connect)
1642
with self.assertRaises(dbus.ConnectFailed):
1643
self.call_method(bus, "methodname", "busname",
1644
"objectpath", "interface")
1645
1646
class fake_pydbus_raises_exception_on_connect(object):
1647
"""fake dbus-python module"""
1648
@classmethod
1649
def SystemBus(cls):
1650
def get(busname, objectpath):
1651
raise gi.repository.GLib.Error()
1652
Bus = collections.namedtuple("Bus", ["get"])
1653
return Bus(get=get)
1654
1655
def test_set_property_uses_setattr(self):
1656
class Object(object):
1657
pass
1658
obj = Object()
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1661
@staticmethod
1662
def get(busname, objectpath):
1663
return {"interface": obj}
1664
bus = pydbus_adapter.SystemBus(pydbus_spy)
1665
value = Unique()
1666
bus.set_property("busname", "objectpath", "interface", "key",
1667
value)
1668
self.assertIs(value, obj.key)
1669
1670
def test_get_suppresses_xml_deprecation_warning(self):
1671
if sys.version_info.major >= 3:
1672
return
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1675
@staticmethod
1676
def get(busname, objectpath):
1677
warnings.warn_explicit(
1678
"deprecated", DeprecationWarning,
1679
"xml.etree.ElementTree", 0)
1680
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1681
with warnings.catch_warnings(record=True) as w:
1682
warnings.simplefilter("always")
1683
bus.get("busname", "objectpath")
1684
self.assertEqual(0, len(w))
1685
1686
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1688
class stub_pydbus(object):
1689
"""stub pydbus module"""
1690
class SystemBus(object):
1691
@staticmethod
1692
def get(busname, objectpath):
1693
return Unique()
1694
1695
def setUp(self):
1696
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1697
1698
def test_returns_distinct_objectpaths(self):
1699
obj1 = self.bus.get("busname", "objectpath1")
1700
self.assertIsInstance(obj1, Unique)
1701
obj2 = self.bus.get("busname", "objectpath2")
1702
self.assertIsInstance(obj2, Unique)
1703
self.assertIsNot(obj1, obj2)
1704
1705
def test_returns_distinct_busnames(self):
1706
obj1 = self.bus.get("busname1", "objectpath")
1707
self.assertIsInstance(obj1, Unique)
1708
obj2 = self.bus.get("busname2", "objectpath")
1709
self.assertIsInstance(obj2, Unique)
1710
self.assertIsNot(obj1, obj2)
1711
1712
def test_returns_distinct_both(self):
1713
obj1 = self.bus.get("busname1", "objectpath")
1714
self.assertIsInstance(obj1, Unique)
1715
obj2 = self.bus.get("busname2", "objectpath")
1716
self.assertIsInstance(obj2, Unique)
1717
self.assertIsNot(obj1, obj2)
1718
1719
def test_returns_same(self):
1720
obj1 = self.bus.get("busname", "objectpath")
1721
self.assertIsInstance(obj1, Unique)
1722
obj2 = self.bus.get("busname", "objectpath")
1723
self.assertIsInstance(obj2, Unique)
1724
self.assertIs(obj1, obj2)
1725
1726
def test_returns_same_old(self):
1727
obj1 = self.bus.get("busname1", "objectpath1")
1728
self.assertIsInstance(obj1, Unique)
1729
obj2 = self.bus.get("busname2", "objectpath2")
1730
self.assertIsInstance(obj2, Unique)
1731
obj1b = self.bus.get("busname1", "objectpath1")
1732
self.assertIsInstance(obj1b, Unique)
1733
self.assertIsNot(obj1, obj2)
1734
self.assertIsNot(obj2, obj1b)
1735
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)
1736
1151
1737
1152
1738
1153
class Test_commands_from_options(unittest.TestCase):
1739
1740
1154
def setUp(self):
1741
1155
self.parser = argparse.ArgumentParser()
1742
1156
add_command_line_options(self.parser)
1957
1371
class TestCommand(unittest.TestCase):
1958
1372
"""Abstract class for tests of command classes"""
1959
1373
1960
class FakeMandosBus(dbus.MandosBus):
1961
def __init__(self, testcase):
1962
self.client_properties = {
1963
"Name": "foo",
1964
"KeyID": ("92ed150794387c03ce684574b1139a65"
1965
"94a34f895daaaf09fd8ea90a27cddb12"),
1966
"Secret": b"secret",
1967
"Host": "foo.example.org",
1968
"Enabled": True,
1969
"Timeout": 300000,
1970
"LastCheckedOK": "2019-02-03T00:00:00",
1971
"Created": "2019-01-02T00:00:00",
1972
"Interval": 120000,
1973
"Fingerprint": ("778827225BA7DE539C5A"
1974
"7CFA59CFF7CDBD9A5920"),
1975
"CheckerRunning": False,
1976
"LastEnabled": "2019-01-03T00:00:00",
1977
"ApprovalPending": False,
1978
"ApprovedByDefault": True,
1979
"LastApprovalRequest": "",
1980
"ApprovalDelay": 0,
1981
"ApprovalDuration": 1000,
1982
"Checker": "fping -q -- %(host)s",
1983
"ExtendedTimeout": 900000,
1984
"Expires": "2019-02-04T00:00:00",
1985
"LastCheckerStatus": 0,
1986
}
1987
self.other_client_properties = {
1988
"Name": "barbar",
1989
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1990
"6ab612cff5ad227247e46c2b020f441c"),
1991
"Secret": b"secretbar",
1992
"Host": "192.0.2.3",
1993
"Enabled": True,
1994
"Timeout": 300000,
1995
"LastCheckedOK": "2019-02-04T00:00:00",
1996
"Created": "2019-01-03T00:00:00",
1997
"Interval": 120000,
1998
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1999
"F547B3A107558FCA3A27"),
2000
"CheckerRunning": True,
2001
"LastEnabled": "2019-01-04T00:00:00",
2002
"ApprovalPending": False,
2003
"ApprovedByDefault": False,
2004
"LastApprovalRequest": "2019-01-03T00:00:00",
2005
"ApprovalDelay": 30000,
2006
"ApprovalDuration": 93785000,
2007
"Checker": ":",
2008
"ExtendedTimeout": 900000,
2009
"Expires": "2019-02-05T00:00:00",
2010
"LastCheckerStatus": -2,
2011
}
2012
self.clients = collections.OrderedDict(
2013
[
2014
("client_objectpath", self.client_properties),
2015
("other_client_objectpath",
2016
self.other_client_properties),
2017
])
2018
self.one_client = {"client_objectpath":
2019
self.client_properties}
2020
self.testcase = testcase
2021
self.calls = []
2022
2023
def call_method(self, methodname, busname, objectpath,
2024
interface, *args):
2025
self.testcase.assertEqual("se.recompile.Mandos", busname)
2026
self.calls.append((methodname, busname, objectpath,
2027
interface, args))
2028
if interface == "org.freedesktop.DBus.Properties":
2029
if methodname == "Set":
2030
self.testcase.assertEqual(3, len(args))
2031
interface, key, value = args
2032
self.testcase.assertEqual(
2033
"se.recompile.Mandos.Client", interface)
2034
self.clients[objectpath][key] = value
2035
return
2036
elif interface == "se.recompile.Mandos":
2037
self.testcase.assertEqual("RemoveClient", methodname)
2038
self.testcase.assertEqual(1, len(args))
2039
clientpath = args[0]
2040
del self.clients[clientpath]
2041
return
2042
elif interface == "se.recompile.Mandos.Client":
2043
if methodname == "Approve":
2044
self.testcase.assertEqual(1, len(args))
2045
return
2046
raise ValueError()
2047
2048
1374
def setUp(self):
2049
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()
2050
1463
2051
1464
2052
1465
class TestBaseCommands(TestCommand):
2053
1466
2054
1467
def test_IsEnabled_exits_successfully(self):
2055
1468
with self.assertRaises(SystemExit) as e:
2056
command.IsEnabled().run(self.bus.one_client)
1469
command.IsEnabled().run(self.one_client)
2057
1470
if e.exception.code is not None:
2058
1471
self.assertEqual(0, e.exception.code)
2059
1472
else:
2060
1473
self.assertIsNone(e.exception.code)
2061
1474
2062
1475
def test_IsEnabled_exits_with_failure(self):
2063
self.bus.client_properties["Enabled"] = False
1476
self.client.attributes["Enabled"] = dbus.Boolean(False)
2064
1477
with self.assertRaises(SystemExit) as e:
2065
command.IsEnabled().run(self.bus.one_client)
1478
command.IsEnabled().run(self.one_client)
2066
1479
if isinstance(e.exception.code, int):
2067
1480
self.assertNotEqual(0, e.exception.code)
2068
1481
else:
2069
1482
self.assertIsNotNone(e.exception.code)
2070
1483
2071
1484
def test_Approve(self):
2072
busname = "se.recompile.Mandos"
2073
client_interface = "se.recompile.Mandos.Client"
2074
command.Approve().run(self.bus.clients, self.bus)
2075
for clientpath in self.bus.clients:
2076
self.assertIn(("Approve", busname, clientpath,
2077
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)
2078
1490
2079
1491
def test_Deny(self):
2080
busname = "se.recompile.Mandos"
2081
client_interface = "se.recompile.Mandos.Client"
2082
command.Deny().run(self.bus.clients, self.bus)
2083
for clientpath in self.bus.clients:
2084
self.assertIn(("Approve", busname, clientpath,
2085
client_interface, (False,)),
2086
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)
2087
1497
2088
1498
def test_Remove(self):
2089
command.Remove().run(self.bus.clients, self.bus)
2090
for clientpath in self.bus.clients:
2091
self.assertIn(("RemoveClient", dbus_busname,
2092
dbus_server_path, dbus_server_interface,
2093
(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)
2094
1509
2095
1510
expected_json = {
2096
1511
"foo": {
2145
1560
2146
1561
def test_DumpJSON_normal(self):
2147
1562
with self.capture_stdout_to_buffer() as buffer:
2148
command.DumpJSON().run(self.bus.clients)
1563
command.DumpJSON().run(self.clients)
2149
1564
json_data = json.loads(buffer.getvalue())
2150
1565
self.assertDictEqual(self.expected_json, json_data)
2151
1566
2162
1577
2163
1578
def test_DumpJSON_one_client(self):
2164
1579
with self.capture_stdout_to_buffer() as buffer:
2165
command.DumpJSON().run(self.bus.one_client)
1580
command.DumpJSON().run(self.one_client)
2166
1581
json_data = json.loads(buffer.getvalue())
2167
1582
expected_json = {"foo": self.expected_json["foo"]}
2168
1583
self.assertDictEqual(expected_json, json_data)
2169
1584
2170
1585
def test_PrintTable_normal(self):
2171
1586
with self.capture_stdout_to_buffer() as buffer:
2172
command.PrintTable().run(self.bus.clients)
1587
command.PrintTable().run(self.clients)
2173
1588
expected_output = "\n".join((
2174
1589
"Name Enabled Timeout Last Successful Check",
2175
1590
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2179
1594
2180
1595
def test_PrintTable_verbose(self):
2181
1596
with self.capture_stdout_to_buffer() as buffer:
2182
command.PrintTable(verbose=True).run(self.bus.clients)
1597
command.PrintTable(verbose=True).run(self.clients)
2183
1598
columns = (
2184
1599
(
2185
1600
"Name ",
2275
1690
2276
1691
def test_PrintTable_one_client(self):
2277
1692
with self.capture_stdout_to_buffer() as buffer:
2278
command.PrintTable().run(self.bus.one_client)
1693
command.PrintTable().run(self.one_client)
2279
1694
expected_output = "\n".join((
2280
1695
"Name Enabled Timeout Last Successful Check",
2281
1696
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2285
1700
2286
1701
class TestPropertySetterCmd(TestCommand):
2287
1702
"""Abstract class for tests of command.PropertySetter classes"""
2288
2289
1703
def runTest(self):
2290
1704
if not hasattr(self, "command"):
2291
return # Abstract TestCase class
2292
2293
if hasattr(self, "values_to_set"):
2294
cmd_args = [(value,) for value in self.values_to_set]
2295
values_to_get = getattr(self, "values_to_get",
2296
self.values_to_set)
2297
else:
2298
cmd_args = [() for x in range(len(self.values_to_get))]
2299
values_to_get = self.values_to_get
2300
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2301
for clientpath in self.bus.clients:
2302
self.bus.clients[clientpath][self.propname] = (
2303
Unique())
2304
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2305
for clientpath in self.bus.clients:
2306
value = (self.bus.clients[clientpath]
2307
[self.propname])
2308
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)
2309
1719
self.assertEqual(value_to_get, value)
2310
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
2311
1728
2312
1729
class TestEnableCmd(TestPropertySetterCmd):
2313
1730
command = command.Enable
2314
1731
propname = "Enabled"
2315
values_to_get = [True]
1732
values_to_set = [dbus.Boolean(True)]
2316
1733
2317
1734
2318
1735
class TestDisableCmd(TestPropertySetterCmd):
2319
1736
command = command.Disable
2320
1737
propname = "Enabled"
2321
values_to_get = [False]
1738
values_to_set = [dbus.Boolean(False)]
2322
1739
2323
1740
2324
1741
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2325
1742
command = command.BumpTimeout
2326
1743
propname = "LastCheckedOK"
2327
values_to_get = [""]
1744
values_to_set = [""]
2328
1745
2329
1746
2330
1747
class TestStartCheckerCmd(TestPropertySetterCmd):
2331
1748
command = command.StartChecker
2332
1749
propname = "CheckerRunning"
2333
values_to_get = [True]
1750
values_to_set = [dbus.Boolean(True)]
2334
1751
2335
1752
2336
1753
class TestStopCheckerCmd(TestPropertySetterCmd):
2337
1754
command = command.StopChecker
2338
1755
propname = "CheckerRunning"
2339
values_to_get = [False]
1756
values_to_set = [dbus.Boolean(False)]
2340
1757
2341
1758
2342
1759
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2343
1760
command = command.ApproveByDefault
2344
1761
propname = "ApprovedByDefault"
2345
values_to_get = [True]
1762
values_to_set = [dbus.Boolean(True)]
2346
1763
2347
1764
2348
1765
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2349
1766
command = command.DenyByDefault
2350
1767
propname = "ApprovedByDefault"
2351
values_to_get = [False]
2352
2353
2354
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):
2355
1784
command = command.SetChecker
2356
1785
propname = "Checker"
2357
1786
values_to_set = ["", ":", "fping -q -- %s"]
2358
1787
2359
1788
2360
class TestSetHostCmd(TestPropertySetterCmd):
1789
class TestSetHostCmd(TestPropertySetterValueCmd):
2361
1790
command = command.SetHost
2362
1791
propname = "Host"
2363
1792
values_to_set = ["192.0.2.3", "client.example.org"]
2364
1793
2365
1794
2366
class TestSetSecretCmd(TestPropertySetterCmd):
1795
class TestSetSecretCmd(TestPropertySetterValueCmd):
2367
1796
command = command.SetSecret
2368
1797
propname = "Secret"
2369
1798
values_to_set = [io.BytesIO(b""),
2371
1800
values_to_get = [f.getvalue() for f in values_to_set]
2372
1801
2373
1802
2374
class TestSetTimeoutCmd(TestPropertySetterCmd):
1803
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
2375
1804
command = command.SetTimeout
2376
1805
propname = "Timeout"
2377
1806
values_to_set = [datetime.timedelta(),
2382
1811
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2383
1812
2384
1813
2385
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
1814
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
2386
1815
command = command.SetExtendedTimeout
2387
1816
propname = "ExtendedTimeout"
2388
1817
values_to_set = [datetime.timedelta(),
2393
1822
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2394
1823
2395
1824
2396
class TestSetIntervalCmd(TestPropertySetterCmd):
1825
class TestSetIntervalCmd(TestPropertySetterValueCmd):
2397
1826
command = command.SetInterval
2398
1827
propname = "Interval"
2399
1828
values_to_set = [datetime.timedelta(),
2404
1833
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2405
1834
2406
1835
2407
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
1836
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
2408
1837
command = command.SetApprovalDelay
2409
1838
propname = "ApprovalDelay"
2410
1839
values_to_set = [datetime.timedelta(),
2415
1844
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2416
1845
2417
1846
2418
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
1847
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
2419
1848
command = command.SetApprovalDuration
2420
1849
propname = "ApprovalDuration"
2421
1850
values_to_set = [datetime.timedelta(),
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0