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 1094
- compare with another revision
- download diff
- download tarball
- view history from revision 1094
- Committer: Teddy Hogeborn
- Date: 2019-03-31 04:39:25 UTC
- Revision ID: teddy@recompile.se-20190331043925-0j9pdspo3hux5nka
mandos-ctl: Refactor: Move command list generation into argparse
* mandos-ctl (add_command_line_options): Don't simply set flags for
each option; instead, add
command objects to a
"commands" list.
(check_option_syntax): Rework to deal with an already-generated
command list.
(commands_from_options): Remove all code to generate command list,
but add more code to move command.Deny
ahead of command.Remove in command list.
(command.PropertySetterValue.argparse): New; used in
add_command_line_options.
(Test_check_option_syntax): Refactor to pass actual string arguments
to parse_args() instead of modifying an
argparse.Namespace.
* mandos-ctl (add_command_line_options): Don't simply set flags for
each option; instead, add
command objects to a
"commands" list.
(check_option_syntax): Rework to deal with an already-generated
command list.
(commands_from_options): Remove all code to generate command list,
but add more code to move command.Deny
ahead of command.Remove in command list.
(command.PropertySetterValue.argparse): New; used in
add_command_line_options.
(Test_check_option_syntax): Refactor to pass actual string arguments
to parse_args() instead of modifying an
argparse.Namespace.
- files modified:
- TODO
added
removed
mandos-ctl
45
45
import io
46
46
import tempfile
47
47
import contextlib
48
import abc
48
49
49
import dbus
50
import dbus as dbus_python
50
51
51
52
# Show warnings by default
52
53
if not sys.warnoptions:
61
62
62
63
if sys.version_info.major == 2:
63
64
str = unicode
65
import StringIO
66
io.StringIO = StringIO.StringIO
64
67
65
68
locale.setlocale(locale.LC_ALL, "")
66
69
67
dbus_busname_domain = "se.recompile"
68
dbus_busname = dbus_busname_domain + ".Mandos"
69
server_dbus_path = "/"
70
server_dbus_interface = dbus_busname_domain + ".Mandos"
71
client_dbus_interface = dbus_busname_domain + ".Mandos.Client"
72
del dbus_busname_domain
73
70
version = "1.8.3"
74
71
75
72
76
try:
77
dbus.OBJECT_MANAGER_IFACE
78
except AttributeError:
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
80
81
82
73
def main():
83
74
parser = argparse.ArgumentParser()
84
85
75
add_command_line_options(parser)
86
76
87
77
options = parser.parse_args()
88
89
78
check_option_syntax(parser, options)
90
79
91
80
clientnames = options.client
93
82
if options.debug:
94
83
log.setLevel(logging.DEBUG)
95
84
96
try:
97
bus = dbus.SystemBus()
98
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
99
dbus_busname, server_dbus_path)
100
mandos_dbus_objc = bus.get_object(dbus_busname,
101
server_dbus_path)
102
except dbus.exceptions.DBusException:
103
log.critical("Could not connect to Mandos server")
104
sys.exit(1)
105
106
mandos_serv = dbus.Interface(mandos_dbus_objc,
107
dbus_interface=server_dbus_interface)
108
mandos_serv_object_manager = dbus.Interface(
109
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
110
111
# Filter out log message from dbus module
112
dbus_logger = logging.getLogger("dbus.proxies")
113
class NullFilter(logging.Filter):
114
def filter(self, record):
115
return False
116
dbus_filter = NullFilter()
117
try:
118
dbus_logger.addFilter(dbus_filter)
119
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
120
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
121
mandos_clients = {path: ifs_and_props[client_dbus_interface]
122
for path, ifs_and_props in
123
mandos_serv_object_manager
124
.GetManagedObjects().items()
125
if client_dbus_interface in ifs_and_props}
126
except dbus.exceptions.DBusException as e:
127
log.critical("Failed to access Mandos server through D-Bus:"
128
"\n%s", e)
129
sys.exit(1)
130
finally:
131
# restore dbus logger
132
dbus_logger.removeFilter(dbus_filter)
133
134
# Compile dict of (clients: properties) to process
135
clients = {}
136
85
bus = dbus_python_adapter.CachingBus(dbus_python)
86
87
try:
88
all_clients = bus.get_clients_and_properties()
89
except dbus.ConnectFailed as e:
90
log.critical("Could not connect to Mandos server: %s", e)
91
sys.exit(1)
92
except dbus.Error as e:
93
log.critical(
94
"Failed to access Mandos server through D-Bus:\n%s", e)
95
sys.exit(1)
96
97
# Compile dict of (clientpath: properties) to process
137
98
if not clientnames:
138
clients = {objpath: properties
139
for objpath, properties in mandos_clients.items()}
99
clients = all_clients
140
100
else:
101
clients = {}
141
102
for name in clientnames:
142
for objpath, properties in mandos_clients.items():
103
for objpath, properties in all_clients.items():
143
104
if properties["Name"] == name:
144
105
clients[objpath] = properties
145
106
break
147
108
log.critical("Client not found on server: %r", name)
148
109
sys.exit(1)
149
110
150
# Run all commands on clients
151
111
commands = commands_from_options(options)
112
152
113
for command in commands:
153
command.run(clients, bus, mandos_serv)
114
command.run(clients, bus)
154
115
155
116
156
117
def add_command_line_options(parser):
161
122
help="Select all clients")
162
123
parser.add_argument("-v", "--verbose", action="store_true",
163
124
help="Print all fields")
164
parser.add_argument("-j", "--dump-json", action="store_true",
125
parser.add_argument("-j", "--dump-json", dest="commands",
126
action="append_const", default=[],
127
const=command.DumpJSON(),
165
128
help="Dump client data in JSON format")
166
129
enable_disable = parser.add_mutually_exclusive_group()
167
enable_disable.add_argument("-e", "--enable", action="store_true",
130
enable_disable.add_argument("-e", "--enable", dest="commands",
131
action="append_const", default=[],
132
const=command.Enable(),
168
133
help="Enable client")
169
enable_disable.add_argument("-d", "--disable",
170
action="store_true",
134
enable_disable.add_argument("-d", "--disable", dest="commands",
135
action="append_const", default=[],
136
const=command.Disable(),
171
137
help="disable client")
172
parser.add_argument("-b", "--bump-timeout", action="store_true",
138
parser.add_argument("-b", "--bump-timeout", dest="commands",
139
action="append_const", default=[],
140
const=command.BumpTimeout(),
173
141
help="Bump timeout for client")
174
142
start_stop_checker = parser.add_mutually_exclusive_group()
175
143
start_stop_checker.add_argument("--start-checker",
176
action="store_true",
144
dest="commands",
145
action="append_const", default=[],
146
const=command.StartChecker(),
177
147
help="Start checker for client")
178
start_stop_checker.add_argument("--stop-checker",
179
action="store_true",
148
start_stop_checker.add_argument("--stop-checker", dest="commands",
149
action="append_const", default=[],
150
const=command.StopChecker(),
180
151
help="Stop checker for client")
181
parser.add_argument("-V", "--is-enabled", action="store_true",
152
parser.add_argument("-V", "--is-enabled", dest="commands",
153
action="append_const", default=[],
154
const=command.IsEnabled(),
182
155
help="Check if client is enabled")
183
parser.add_argument("-r", "--remove", action="store_true",
156
parser.add_argument("-r", "--remove", dest="commands",
157
action="append_const", default=[],
158
const=command.Remove(),
184
159
help="Remove client")
185
parser.add_argument("-c", "--checker",
160
parser.add_argument("-c", "--checker", dest="commands",
161
action="append", default=[],
162
metavar="COMMAND", type=command.SetChecker,
186
163
help="Set checker command for client")
187
parser.add_argument("-t", "--timeout", type=string_to_delta,
188
help="Set timeout for client")
189
parser.add_argument("--extended-timeout", type=string_to_delta,
190
help="Set extended timeout for client")
191
parser.add_argument("-i", "--interval", type=string_to_delta,
192
help="Set checker interval for client")
164
parser.add_argument(
165
"-t", "--timeout", dest="commands", action="append",
166
default=[], metavar="TIME",
167
type=command.SetTimeout.argparse(string_to_delta),
168
help="Set timeout for client")
169
parser.add_argument(
170
"--extended-timeout", dest="commands", action="append",
171
default=[], metavar="TIME",
172
type=command.SetExtendedTimeout.argparse(string_to_delta),
173
help="Set extended timeout for client")
174
parser.add_argument(
175
"-i", "--interval", dest="commands", action="append",
176
default=[], metavar="TIME",
177
type=command.SetInterval.argparse(string_to_delta),
178
help="Set checker interval for client")
193
179
approve_deny_default = parser.add_mutually_exclusive_group()
194
180
approve_deny_default.add_argument(
195
"--approve-by-default", action="store_true",
196
default=None, dest="approved_by_default",
181
"--approve-by-default", dest="commands",
182
action="append_const", default=[],
183
const=command.ApproveByDefault(),
197
184
help="Set client to be approved by default")
198
185
approve_deny_default.add_argument(
199
"--deny-by-default", action="store_false",
200
dest="approved_by_default",
186
"--deny-by-default", dest="commands",
187
action="append_const", default=[],
188
const=command.DenyByDefault(),
201
189
help="Set client to be denied by default")
202
parser.add_argument("--approval-delay", type=string_to_delta,
203
help="Set delay before client approve/deny")
204
parser.add_argument("--approval-duration", type=string_to_delta,
205
help="Set duration of one client approval")
206
parser.add_argument("-H", "--host", help="Set host for client")
207
parser.add_argument("-s", "--secret",
208
type=argparse.FileType(mode="rb"),
209
help="Set password blob (file) for client")
190
parser.add_argument(
191
"--approval-delay", dest="commands", action="append",
192
default=[], metavar="TIME",
193
type=command.SetApprovalDelay.argparse(string_to_delta),
194
help="Set delay before client approve/deny")
195
parser.add_argument(
196
"--approval-duration", dest="commands", action="append",
197
default=[], metavar="TIME",
198
type=command.SetApprovalDuration.argparse(string_to_delta),
199
help="Set duration of one client approval")
200
parser.add_argument("-H", "--host", dest="commands",
201
action="append", default=[], metavar="STRING",
202
type=command.SetHost,
203
help="Set host for client")
204
parser.add_argument(
205
"-s", "--secret", dest="commands", action="append",
206
default=[], metavar="FILENAME",
207
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
208
help="Set password blob (file) for client")
210
209
approve_deny = parser.add_mutually_exclusive_group()
211
210
approve_deny.add_argument(
212
"-A", "--approve", action="store_true",
211
"-A", "--approve", dest="commands", action="append_const",
212
default=[], const=command.Approve(),
213
213
help="Approve any current client request")
214
approve_deny.add_argument("-D", "--deny", action="store_true",
214
approve_deny.add_argument("-D", "--deny", dest="commands",
215
action="append_const", default=[],
216
const=command.Deny(),
215
217
help="Deny any current client request")
216
218
parser.add_argument("--debug", action="store_true",
217
219
help="Debug mode (show D-Bus commands)")
254
256
>>> rfc3339_duration_to_delta("")
255
257
Traceback (most recent call last):
256
258
...
257
ValueError: Invalid RFC 3339 duration: u''
259
ValueError: Invalid RFC 3339 duration: ""
258
260
>>> # Must start with "P":
259
261
>>> rfc3339_duration_to_delta("1D")
260
262
Traceback (most recent call last):
261
263
...
262
ValueError: Invalid RFC 3339 duration: u'1D'
264
ValueError: Invalid RFC 3339 duration: "1D"
263
265
>>> # Must use correct order
264
266
>>> rfc3339_duration_to_delta("PT1S2M")
265
267
Traceback (most recent call last):
266
268
...
267
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
269
ValueError: Invalid RFC 3339 duration: "PT1S2M"
268
270
>>> # Time needs time marker
269
271
>>> rfc3339_duration_to_delta("P1H2S")
270
272
Traceback (most recent call last):
271
273
...
272
ValueError: Invalid RFC 3339 duration: u'P1H2S'
274
ValueError: Invalid RFC 3339 duration: "P1H2S"
273
275
>>> # Weeks can not be combined with anything else
274
276
>>> rfc3339_duration_to_delta("P1D2W")
275
277
Traceback (most recent call last):
276
278
...
277
ValueError: Invalid RFC 3339 duration: u'P1D2W'
279
ValueError: Invalid RFC 3339 duration: "P1D2W"
278
280
>>> rfc3339_duration_to_delta("P2W2H")
279
281
Traceback (most recent call last):
280
282
...
281
ValueError: Invalid RFC 3339 duration: u'P2W2H'
283
ValueError: Invalid RFC 3339 duration: "P2W2H"
282
284
"""
283
285
284
286
# Parsing an RFC 3339 duration with regular expressions is not
355
357
break
356
358
else:
357
359
# No currently valid tokens were found
358
raise ValueError("Invalid RFC 3339 duration: {!r}"
360
raise ValueError("Invalid RFC 3339 duration: \"{}\""
359
361
.format(duration))
360
362
# End token found
361
363
return value
408
410
"""Apply additional restrictions on options, not expressible in
409
411
argparse"""
410
412
411
def has_actions(options):
412
return any((options.enable,
413
options.disable,
414
options.bump_timeout,
415
options.start_checker,
416
options.stop_checker,
417
options.is_enabled,
418
options.remove,
419
options.checker is not None,
420
options.timeout is not None,
421
options.extended_timeout is not None,
422
options.interval is not None,
423
options.approved_by_default is not None,
424
options.approval_delay is not None,
425
options.approval_duration is not None,
426
options.host is not None,
427
options.secret is not None,
428
options.approve,
429
options.deny))
413
def has_commands(options, commands=None):
414
if commands is None:
415
commands = (command.Enable,
416
command.Disable,
417
command.BumpTimeout,
418
command.StartChecker,
419
command.StopChecker,
420
command.IsEnabled,
421
command.Remove,
422
command.SetChecker,
423
command.SetTimeout,
424
command.SetExtendedTimeout,
425
command.SetInterval,
426
command.ApproveByDefault,
427
command.DenyByDefault,
428
command.SetApprovalDelay,
429
command.SetApprovalDuration,
430
command.SetHost,
431
command.SetSecret,
432
command.Approve,
433
command.Deny)
434
return any(isinstance(cmd, commands)
435
for cmd in options.commands)
430
436
431
if has_actions(options) and not (options.client or options.all):
437
if has_commands(options) and not (options.client or options.all):
432
438
parser.error("Options require clients names or --all.")
433
if options.verbose and has_actions(options):
439
if options.verbose and has_commands(options):
434
440
parser.error("--verbose can only be used alone.")
435
if options.dump_json and (options.verbose
436
or has_actions(options)):
441
if (has_commands(options, (command.DumpJSON,))
442
and (options.verbose or len(options.commands) > 1)):
437
443
parser.error("--dump-json can only be used alone.")
438
if options.all and not has_actions(options):
444
if options.all and not has_commands(options):
439
445
parser.error("--all requires an action.")
440
if options.is_enabled and len(options.client) > 1:
446
if (has_commands(options, (command.IsEnabled,))
447
and len(options.client) > 1):
441
448
parser.error("--is-enabled requires exactly one client")
442
if options.remove:
443
options.remove = False
444
if has_actions(options) and not options.deny:
445
parser.error("--remove can only be combined with --deny")
446
options.remove = True
449
if (len(options.commands) > 1
450
and has_commands(options, (command.Remove,))
451
and not has_commands(options, (command.Deny,))):
452
parser.error("--remove can only be combined with --deny")
453
454
455
class dbus(object):
456
457
class SystemBus(object):
458
459
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
460
def get_managed_objects(self, busname, objectpath):
461
return self.call_method("GetManagedObjects", busname,
462
objectpath,
463
self.object_manager_iface)
464
465
properties_iface = "org.freedesktop.DBus.Properties"
466
def set_property(self, busname, objectpath, interface, key,
467
value):
468
self.call_method("Set", busname, objectpath,
469
self.properties_iface, interface, key,
470
value)
471
472
473
class MandosBus(SystemBus):
474
busname_domain = "se.recompile"
475
busname = busname_domain + ".Mandos"
476
server_path = "/"
477
server_interface = busname_domain + ".Mandos"
478
client_interface = busname_domain + ".Mandos.Client"
479
del busname_domain
480
481
def get_clients_and_properties(self):
482
managed_objects = self.get_managed_objects(
483
self.busname, self.server_path)
484
return {objpath: properties[self.client_interface]
485
for objpath, properties in managed_objects.items()
486
if self.client_interface in properties}
487
488
def set_client_property(self, objectpath, key, value):
489
return self.set_property(self.busname, objectpath,
490
self.client_interface, key,
491
value)
492
493
def call_client_method(self, objectpath, method, *args):
494
return self.call_method(method, self.busname, objectpath,
495
self.client_interface, *args)
496
497
def call_server_method(self, method, *args):
498
return self.call_method(method, self.busname,
499
self.server_path,
500
self.server_interface, *args)
501
502
class Error(Exception):
503
pass
504
505
class ConnectFailed(Error):
506
pass
507
508
509
class dbus_python_adapter(object):
510
511
class SystemBus(dbus.MandosBus):
512
"""Use dbus-python"""
513
514
def __init__(self, module=dbus_python):
515
self.dbus_python = module
516
self.bus = self.dbus_python.SystemBus()
517
518
@contextlib.contextmanager
519
def convert_exception(self, exception_class=dbus.Error):
520
try:
521
yield
522
except self.dbus_python.exceptions.DBusException as e:
523
# This does what "raise from" would do
524
exc = exception_class(*e.args)
525
exc.__cause__ = e
526
raise exc
527
528
def call_method(self, methodname, busname, objectpath,
529
interface, *args):
530
proxy_object = self.get_object(busname, objectpath)
531
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
532
interface, methodname,
533
", ".join(repr(a) for a in args))
534
method = getattr(proxy_object, methodname)
535
with self.convert_exception():
536
with dbus_python_adapter.SilenceLogger(
537
"dbus.proxies"):
538
value = method(*args, dbus_interface=interface)
539
return self.type_filter(value)
540
541
def get_object(self, busname, objectpath):
542
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
543
busname, objectpath)
544
with self.convert_exception(dbus.ConnectFailed):
545
return self.bus.get_object(busname, objectpath)
546
547
def type_filter(self, value):
548
"""Convert the most bothersome types to Python types"""
549
if isinstance(value, self.dbus_python.Boolean):
550
return bool(value)
551
if isinstance(value, self.dbus_python.ObjectPath):
552
return str(value)
553
# Also recurse into dictionaries
554
if isinstance(value, self.dbus_python.Dictionary):
555
return {self.type_filter(key):
556
self.type_filter(subval)
557
for key, subval in value.items()}
558
return value
559
560
561
class SilenceLogger(object):
562
"Simple context manager to silence a particular logger"
563
def __init__(self, loggername):
564
self.logger = logging.getLogger(loggername)
565
566
def __enter__(self):
567
self.logger.addFilter(self.nullfilter)
568
569
class NullFilter(logging.Filter):
570
def filter(self, record):
571
return False
572
573
nullfilter = NullFilter()
574
575
def __exit__(self, exc_type, exc_val, exc_tb):
576
self.logger.removeFilter(self.nullfilter)
577
578
579
class CachingBus(SystemBus):
580
"""A caching layer for dbus_python_adapter.SystemBus"""
581
def __init__(self, *args, **kwargs):
582
self.object_cache = {}
583
super(dbus_python_adapter.CachingBus,
584
self).__init__(*args, **kwargs)
585
def get_object(self, busname, objectpath):
586
try:
587
return self.object_cache[(busname, objectpath)]
588
except KeyError:
589
new_object = super(
590
dbus_python_adapter.CachingBus,
591
self).get_object(busname, objectpath)
592
self.object_cache[(busname, objectpath)] = new_object
593
return new_object
447
594
448
595
449
596
def commands_from_options(options):
450
597
451
commands = []
452
453
if options.is_enabled:
454
commands.append(IsEnabledCmd())
455
456
if options.approve:
457
commands.append(ApproveCmd())
458
459
if options.deny:
460
commands.append(DenyCmd())
461
462
if options.remove:
463
commands.append(RemoveCmd())
464
465
if options.dump_json:
466
commands.append(DumpJSONCmd())
467
468
if options.enable:
469
commands.append(EnableCmd())
470
471
if options.disable:
472
commands.append(DisableCmd())
473
474
if options.bump_timeout:
475
commands.append(BumpTimeoutCmd())
476
477
if options.start_checker:
478
commands.append(StartCheckerCmd())
479
480
if options.stop_checker:
481
commands.append(StopCheckerCmd())
482
483
if options.approved_by_default is not None:
484
if options.approved_by_default:
485
commands.append(ApproveByDefaultCmd())
598
commands = list(options.commands)
599
600
def find_cmd(cmd, commands):
601
i = 0
602
for i, c in enumerate(commands):
603
if isinstance(c, cmd):
604
return i
605
return i+1
606
607
# If command.Remove is present, move any instances of command.Deny
608
# to occur ahead of command.Remove.
609
index_of_remove = find_cmd(command.Remove, commands)
610
before_remove = commands[:index_of_remove]
611
after_remove = commands[index_of_remove:]
612
cleaned_after = []
613
for cmd in after_remove:
614
if isinstance(cmd, command.Deny):
615
before_remove.append(cmd)
486
616
else:
487
commands.append(DenyByDefaultCmd())
488
489
if options.checker is not None:
490
commands.append(SetCheckerCmd(options.checker))
491
492
if options.host is not None:
493
commands.append(SetHostCmd(options.host))
494
495
if options.secret is not None:
496
commands.append(SetSecretCmd(options.secret))
497
498
if options.timeout is not None:
499
commands.append(SetTimeoutCmd(options.timeout))
500
501
if options.extended_timeout:
502
commands.append(
503
SetExtendedTimeoutCmd(options.extended_timeout))
504
505
if options.interval is not None:
506
commands.append(SetIntervalCmd(options.interval))
507
508
if options.approval_delay is not None:
509
commands.append(SetApprovalDelayCmd(options.approval_delay))
510
511
if options.approval_duration is not None:
512
commands.append(
513
SetApprovalDurationCmd(options.approval_duration))
617
cleaned_after.append(cmd)
618
if cleaned_after != after_remove:
619
commands = before_remove + cleaned_after
514
620
515
621
# If no command option has been given, show table of clients,
516
622
# optionally verbosely
517
623
if not commands:
518
commands.append(PrintTableCmd(verbose=options.verbose))
624
commands.append(command.PrintTable(verbose=options.verbose))
519
625
520
626
return commands
521
627
522
628
523
class Command(object):
524
"""Abstract class for commands"""
525
def run(self, clients, bus=None, mandos=None):
526
"""Normal commands should implement run_on_one_client(), but
527
commands which want to operate on all clients at the same time
528
can override this run() method instead."""
529
self.mandos = mandos
530
for clientpath, properties in clients.items():
531
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
532
dbus_busname, str(clientpath))
533
client = bus.get_object(dbus_busname, clientpath)
534
self.run_on_one_client(client, properties)
535
536
537
class IsEnabledCmd(Command):
538
def run(self, clients, bus=None, mandos=None):
539
client, properties = next(iter(clients.items()))
540
if self.is_enabled(client, properties):
541
sys.exit(0)
542
sys.exit(1)
543
def is_enabled(self, client, properties):
544
return properties["Enabled"]
545
546
547
class ApproveCmd(Command):
548
def run_on_one_client(self, client, properties):
549
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
550
client.__dbus_object_path__, client_dbus_interface)
551
client.Approve(dbus.Boolean(True),
552
dbus_interface=client_dbus_interface)
553
554
555
class DenyCmd(Command):
556
def run_on_one_client(self, client, properties):
557
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
558
client.__dbus_object_path__, client_dbus_interface)
559
client.Approve(dbus.Boolean(False),
560
dbus_interface=client_dbus_interface)
561
562
563
class RemoveCmd(Command):
564
def run_on_one_client(self, client, properties):
565
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
566
server_dbus_path, server_dbus_interface,
567
str(client.__dbus_object_path__))
568
self.mandos.RemoveClient(client.__dbus_object_path__)
569
570
571
class OutputCmd(Command):
572
"""Abstract class for commands outputting client details"""
573
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
574
"Created", "Interval", "Host", "KeyID",
575
"Fingerprint", "CheckerRunning", "LastEnabled",
576
"ApprovalPending", "ApprovedByDefault",
577
"LastApprovalRequest", "ApprovalDelay",
578
"ApprovalDuration", "Checker", "ExtendedTimeout",
579
"Expires", "LastCheckerStatus")
580
581
def run(self, clients, bus=None, mandos=None):
582
print(self.output(clients.values()))
583
584
def output(self, clients):
585
raise NotImplementedError()
586
587
588
class DumpJSONCmd(OutputCmd):
589
def output(self, clients):
590
data = {client["Name"]:
591
{key: self.dbus_boolean_to_bool(client[key])
592
for key in self.all_keywords}
593
for client in clients.values()}
594
return json.dumps(data, indent=4, separators=(',', ': '))
595
596
@staticmethod
597
def dbus_boolean_to_bool(value):
598
if isinstance(value, dbus.Boolean):
599
value = bool(value)
600
return value
601
602
603
class PrintTableCmd(OutputCmd):
604
def __init__(self, verbose=False):
605
self.verbose = verbose
606
607
def output(self, clients):
608
default_keywords = ("Name", "Enabled", "Timeout",
609
"LastCheckedOK")
610
keywords = default_keywords
611
if self.verbose:
612
keywords = self.all_keywords
613
return str(self.TableOfClients(clients, keywords))
614
615
class TableOfClients(object):
616
tableheaders = {
617
"Name": "Name",
618
"Enabled": "Enabled",
619
"Timeout": "Timeout",
620
"LastCheckedOK": "Last Successful Check",
621
"LastApprovalRequest": "Last Approval Request",
622
"Created": "Created",
623
"Interval": "Interval",
624
"Host": "Host",
625
"Fingerprint": "Fingerprint",
626
"KeyID": "Key ID",
627
"CheckerRunning": "Check Is Running",
628
"LastEnabled": "Last Enabled",
629
"ApprovalPending": "Approval Is Pending",
630
"ApprovedByDefault": "Approved By Default",
631
"ApprovalDelay": "Approval Delay",
632
"ApprovalDuration": "Approval Duration",
633
"Checker": "Checker",
634
"ExtendedTimeout": "Extended Timeout",
635
"Expires": "Expires",
636
"LastCheckerStatus": "Last Checker Status",
637
}
638
639
def __init__(self, clients, keywords, tableheaders=None):
640
self.clients = clients
641
self.keywords = keywords
642
if tableheaders is not None:
643
self.tableheaders = tableheaders
644
645
def __str__(self):
646
return "\n".join(self.rows())
647
648
if sys.version_info.major == 2:
649
__unicode__ = __str__
629
class command(object):
630
"""A namespace for command classes"""
631
632
class Base(object):
633
"""Abstract base class for commands"""
634
def run(self, clients, bus=None):
635
"""Normal commands should implement run_on_one_client(),
636
but commands which want to operate on all clients at the same time can
637
override this run() method instead.
638
"""
639
self.bus = bus
640
for client, properties in clients.items():
641
self.run_on_one_client(client, properties)
642
643
644
class IsEnabled(Base):
645
def run(self, clients, bus=None):
646
properties = next(iter(clients.values()))
647
if properties["Enabled"]:
648
sys.exit(0)
649
sys.exit(1)
650
651
652
class Approve(Base):
653
def run_on_one_client(self, client, properties):
654
self.bus.call_client_method(client, "Approve", True)
655
656
657
class Deny(Base):
658
def run_on_one_client(self, client, properties):
659
self.bus.call_client_method(client, "Approve", False)
660
661
662
class Remove(Base):
663
def run(self, clients, bus):
664
for clientpath in frozenset(clients.keys()):
665
bus.call_server_method("RemoveClient", clientpath)
666
667
668
class Output(Base):
669
"""Abstract class for commands outputting client details"""
670
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
671
"Created", "Interval", "Host", "KeyID",
672
"Fingerprint", "CheckerRunning",
673
"LastEnabled", "ApprovalPending",
674
"ApprovedByDefault", "LastApprovalRequest",
675
"ApprovalDelay", "ApprovalDuration",
676
"Checker", "ExtendedTimeout", "Expires",
677
"LastCheckerStatus")
678
679
680
class DumpJSON(Output):
681
def run(self, clients, bus=None):
682
data = {properties["Name"]:
683
{key: properties[key]
684
for key in self.all_keywords}
685
for properties in clients.values()}
686
print(json.dumps(data, indent=4, separators=(',', ': ')))
687
688
689
class PrintTable(Output):
690
def __init__(self, verbose=False):
691
self.verbose = verbose
692
693
def run(self, clients, bus=None):
694
default_keywords = ("Name", "Enabled", "Timeout",
695
"LastCheckedOK")
696
keywords = default_keywords
697
if self.verbose:
698
keywords = self.all_keywords
699
print(self.TableOfClients(clients.values(), keywords))
700
701
class TableOfClients(object):
702
tableheaders = {
703
"Name": "Name",
704
"Enabled": "Enabled",
705
"Timeout": "Timeout",
706
"LastCheckedOK": "Last Successful Check",
707
"LastApprovalRequest": "Last Approval Request",
708
"Created": "Created",
709
"Interval": "Interval",
710
"Host": "Host",
711
"Fingerprint": "Fingerprint",
712
"KeyID": "Key ID",
713
"CheckerRunning": "Check Is Running",
714
"LastEnabled": "Last Enabled",
715
"ApprovalPending": "Approval Is Pending",
716
"ApprovedByDefault": "Approved By Default",
717
"ApprovalDelay": "Approval Delay",
718
"ApprovalDuration": "Approval Duration",
719
"Checker": "Checker",
720
"ExtendedTimeout": "Extended Timeout",
721
"Expires": "Expires",
722
"LastCheckerStatus": "Last Checker Status",
723
}
724
725
def __init__(self, clients, keywords):
726
self.clients = clients
727
self.keywords = keywords
728
650
729
def __str__(self):
651
return str(self).encode(locale.getpreferredencoding())
652
653
def rows(self):
654
format_string = self.row_formatting_string()
655
rows = [self.header_line(format_string)]
656
rows.extend(self.client_line(client, format_string)
657
for client in self.clients)
658
return rows
659
660
def row_formatting_string(self):
661
"Format string used to format table rows"
662
return " ".join("{{{key}:{width}}}".format(
663
width=max(len(self.tableheaders[key]),
664
*(len(self.string_from_client(client, key))
665
for client in self.clients)),
666
key=key)
667
for key in self.keywords)
668
669
def string_from_client(self, client, key):
670
return self.valuetostring(client[key], key)
730
return "\n".join(self.rows())
731
732
if sys.version_info.major == 2:
733
__unicode__ = __str__
734
def __str__(self):
735
return str(self).encode(
736
locale.getpreferredencoding())
737
738
def rows(self):
739
format_string = self.row_formatting_string()
740
rows = [self.header_line(format_string)]
741
rows.extend(self.client_line(client, format_string)
742
for client in self.clients)
743
return rows
744
745
def row_formatting_string(self):
746
"Format string used to format table rows"
747
return " ".join("{{{key}:{width}}}".format(
748
width=max(len(self.tableheaders[key]),
749
*(len(self.string_from_client(client,
750
key))
751
for client in self.clients)),
752
key=key)
753
for key in self.keywords)
754
755
def string_from_client(self, client, key):
756
return self.valuetostring(client[key], key)
757
758
@classmethod
759
def valuetostring(cls, value, keyword):
760
if isinstance(value, bool):
761
return "Yes" if value else "No"
762
if keyword in ("Timeout", "Interval", "ApprovalDelay",
763
"ApprovalDuration", "ExtendedTimeout"):
764
return cls.milliseconds_to_string(value)
765
return str(value)
766
767
def header_line(self, format_string):
768
return format_string.format(**self.tableheaders)
769
770
def client_line(self, client, format_string):
771
return format_string.format(
772
**{key: self.string_from_client(client, key)
773
for key in self.keywords})
774
775
@staticmethod
776
def milliseconds_to_string(ms):
777
td = datetime.timedelta(0, 0, 0, ms)
778
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
779
.format(days="{}T".format(td.days)
780
if td.days else "",
781
hours=td.seconds // 3600,
782
minutes=(td.seconds % 3600) // 60,
783
seconds=td.seconds % 60))
784
785
786
class PropertySetter(Base):
787
"Abstract class for Actions for setting one client property"
788
789
def run_on_one_client(self, client, properties=None):
790
"""Set the Client's D-Bus property"""
791
self.bus.set_client_property(client, self.propname,
792
self.value_to_set)
793
794
@property
795
def propname(self):
796
raise NotImplementedError()
797
798
799
class Enable(PropertySetter):
800
propname = "Enabled"
801
value_to_set = True
802
803
804
class Disable(PropertySetter):
805
propname = "Enabled"
806
value_to_set = False
807
808
809
class BumpTimeout(PropertySetter):
810
propname = "LastCheckedOK"
811
value_to_set = ""
812
813
814
class StartChecker(PropertySetter):
815
propname = "CheckerRunning"
816
value_to_set = True
817
818
819
class StopChecker(PropertySetter):
820
propname = "CheckerRunning"
821
value_to_set = False
822
823
824
class ApproveByDefault(PropertySetter):
825
propname = "ApprovedByDefault"
826
value_to_set = True
827
828
829
class DenyByDefault(PropertySetter):
830
propname = "ApprovedByDefault"
831
value_to_set = False
832
833
834
class PropertySetterValue(PropertySetter):
835
"""Abstract class for PropertySetter recieving a value as
836
constructor argument instead of a class attribute."""
837
def __init__(self, value):
838
self.value_to_set = value
671
839
672
840
@classmethod
673
def valuetostring(cls, value, keyword):
674
if isinstance(value, dbus.Boolean):
675
return "Yes" if value else "No"
676
if keyword in ("Timeout", "Interval", "ApprovalDelay",
677
"ApprovalDuration", "ExtendedTimeout"):
678
return cls.milliseconds_to_string(value)
679
return str(value)
680
681
def header_line(self, format_string):
682
return format_string.format(**self.tableheaders)
683
684
def client_line(self, client, format_string):
685
return format_string.format(
686
**{key: self.string_from_client(client, key)
687
for key in self.keywords})
688
689
@staticmethod
690
def milliseconds_to_string(ms):
691
td = datetime.timedelta(0, 0, 0, ms)
692
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
693
.format(days="{}T".format(td.days)
694
if td.days else "",
695
hours=td.seconds // 3600,
696
minutes=(td.seconds % 3600) // 60,
697
seconds=td.seconds % 60))
698
699
700
class PropertyCmd(Command):
701
"""Abstract class for Actions for setting one client property"""
702
703
def run_on_one_client(self, client, properties):
704
"""Set the Client's D-Bus property"""
705
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
706
client.__dbus_object_path__,
707
dbus.PROPERTIES_IFACE, client_dbus_interface,
708
self.propname, self.value_to_set
709
if not isinstance(self.value_to_set, dbus.Boolean)
710
else bool(self.value_to_set))
711
client.Set(client_dbus_interface, self.propname,
712
self.value_to_set,
713
dbus_interface=dbus.PROPERTIES_IFACE)
714
715
@property
716
def propname(self):
717
raise NotImplementedError()
718
719
720
class EnableCmd(PropertyCmd):
721
propname = "Enabled"
722
value_to_set = dbus.Boolean(True)
723
724
725
class DisableCmd(PropertyCmd):
726
propname = "Enabled"
727
value_to_set = dbus.Boolean(False)
728
729
730
class BumpTimeoutCmd(PropertyCmd):
731
propname = "LastCheckedOK"
732
value_to_set = ""
733
734
735
class StartCheckerCmd(PropertyCmd):
736
propname = "CheckerRunning"
737
value_to_set = dbus.Boolean(True)
738
739
740
class StopCheckerCmd(PropertyCmd):
741
propname = "CheckerRunning"
742
value_to_set = dbus.Boolean(False)
743
744
745
class ApproveByDefaultCmd(PropertyCmd):
746
propname = "ApprovedByDefault"
747
value_to_set = dbus.Boolean(True)
748
749
750
class DenyByDefaultCmd(PropertyCmd):
751
propname = "ApprovedByDefault"
752
value_to_set = dbus.Boolean(False)
753
754
755
class PropertyValueCmd(PropertyCmd):
756
"""Abstract class for PropertyCmd recieving a value as argument"""
757
def __init__(self, value):
758
self.value_to_set = value
759
760
761
class SetCheckerCmd(PropertyValueCmd):
762
propname = "Checker"
763
764
765
class SetHostCmd(PropertyValueCmd):
766
propname = "Host"
767
768
769
class SetSecretCmd(PropertyValueCmd):
770
propname = "Secret"
771
772
@property
773
def value_to_set(self):
774
return self._vts
775
776
@value_to_set.setter
777
def value_to_set(self, value):
778
"""When setting, read data from supplied file object"""
779
self._vts = value.read()
780
value.close()
781
782
783
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
784
"""Abstract class for PropertyValueCmd taking a value argument as
785
a datetime.timedelta() but should store it as milliseconds."""
786
787
@property
788
def value_to_set(self):
789
return self._vts
790
791
@value_to_set.setter
792
def value_to_set(self, value):
793
"""When setting, convert value from a datetime.timedelta"""
794
self._vts = int(round(value.total_seconds() * 1000))
795
796
797
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
798
propname = "Timeout"
799
800
801
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
802
propname = "ExtendedTimeout"
803
804
805
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
806
propname = "Interval"
807
808
809
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
810
propname = "ApprovalDelay"
811
812
813
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
814
propname = "ApprovalDuration"
841
def argparse(cls, argtype):
842
def cmdtype(arg):
843
return cls(argtype(arg))
844
return cmdtype
845
846
class SetChecker(PropertySetterValue):
847
propname = "Checker"
848
849
850
class SetHost(PropertySetterValue):
851
propname = "Host"
852
853
854
class SetSecret(PropertySetterValue):
855
propname = "Secret"
856
857
@property
858
def value_to_set(self):
859
return self._vts
860
861
@value_to_set.setter
862
def value_to_set(self, value):
863
"""When setting, read data from supplied file object"""
864
self._vts = value.read()
865
value.close()
866
867
868
class PropertySetterValueMilliseconds(PropertySetterValue):
869
"""Abstract class for PropertySetterValue taking a value
870
argument as a datetime.timedelta() but should store it as
871
milliseconds."""
872
873
@property
874
def value_to_set(self):
875
return self._vts
876
877
@value_to_set.setter
878
def value_to_set(self, value):
879
"When setting, convert value from a datetime.timedelta"
880
self._vts = int(round(value.total_seconds() * 1000))
881
882
883
class SetTimeout(PropertySetterValueMilliseconds):
884
propname = "Timeout"
885
886
887
class SetExtendedTimeout(PropertySetterValueMilliseconds):
888
propname = "ExtendedTimeout"
889
890
891
class SetInterval(PropertySetterValueMilliseconds):
892
propname = "Interval"
893
894
895
class SetApprovalDelay(PropertySetterValueMilliseconds):
896
propname = "ApprovalDelay"
897
898
899
class SetApprovalDuration(PropertySetterValueMilliseconds):
900
propname = "ApprovalDuration"
815
901
816
902
817
903
818
class Test_string_to_delta(unittest.TestCase):
819
def test_handles_basic_rfc3339(self):
820
self.assertEqual(string_to_delta("PT0S"),
821
datetime.timedelta())
822
self.assertEqual(string_to_delta("P0D"),
823
datetime.timedelta())
824
self.assertEqual(string_to_delta("PT1S"),
825
datetime.timedelta(0, 1))
826
self.assertEqual(string_to_delta("PT2H"),
827
datetime.timedelta(0, 7200))
904
class TestCaseWithAssertLogs(unittest.TestCase):
905
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
906
907
if not hasattr(unittest.TestCase, "assertLogs"):
908
@contextlib.contextmanager
909
def assertLogs(self, logger, level=logging.INFO):
910
capturing_handler = self.CapturingLevelHandler(level)
911
old_level = logger.level
912
old_propagate = logger.propagate
913
logger.addHandler(capturing_handler)
914
logger.setLevel(level)
915
logger.propagate = False
916
try:
917
yield capturing_handler.watcher
918
finally:
919
logger.propagate = old_propagate
920
logger.removeHandler(capturing_handler)
921
logger.setLevel(old_level)
922
self.assertGreater(len(capturing_handler.watcher.records),
923
0)
924
925
class CapturingLevelHandler(logging.Handler):
926
def __init__(self, level, *args, **kwargs):
927
logging.Handler.__init__(self, *args, **kwargs)
928
self.watcher = self.LoggingWatcher([], [])
929
def emit(self, record):
930
self.watcher.records.append(record)
931
self.watcher.output.append(self.format(record))
932
933
LoggingWatcher = collections.namedtuple("LoggingWatcher",
934
("records",
935
"output"))
936
937
938
class Unique(object):
939
"""Class for objects which exist only to be unique objects, since
940
unittest.mock.sentinel only exists in Python 3.3"""
941
942
943
class Test_string_to_delta(TestCaseWithAssertLogs):
944
# Just test basic RFC 3339 functionality here, the doc string for
945
# rfc3339_duration_to_delta() already has more comprehensive
946
# tests, which are run by doctest.
947
948
def test_rfc3339_zero_seconds(self):
949
self.assertEqual(datetime.timedelta(),
950
string_to_delta("PT0S"))
951
952
def test_rfc3339_zero_days(self):
953
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
954
955
def test_rfc3339_one_second(self):
956
self.assertEqual(datetime.timedelta(0, 1),
957
string_to_delta("PT1S"))
958
959
def test_rfc3339_two_hours(self):
960
self.assertEqual(datetime.timedelta(0, 7200),
961
string_to_delta("PT2H"))
828
962
829
963
def test_falls_back_to_pre_1_6_1_with_warning(self):
830
# assertLogs only exists in Python 3.4
831
if hasattr(self, "assertLogs"):
832
with self.assertLogs(log, logging.WARNING):
833
value = string_to_delta("2h")
834
else:
835
class WarningFilter(logging.Filter):
836
"""Don't show, but record the presence of, warnings"""
837
def filter(self, record):
838
is_warning = record.levelno >= logging.WARNING
839
self.found = is_warning or getattr(self, "found",
840
False)
841
return not is_warning
842
warning_filter = WarningFilter()
843
log.addFilter(warning_filter)
844
try:
845
value = string_to_delta("2h")
846
finally:
847
log.removeFilter(warning_filter)
848
self.assertTrue(getattr(warning_filter, "found", False))
849
self.assertEqual(value, datetime.timedelta(0, 7200))
964
with self.assertLogs(log, logging.WARNING):
965
value = string_to_delta("2h")
966
self.assertEqual(datetime.timedelta(0, 7200), value)
850
967
851
968
852
969
class Test_check_option_syntax(unittest.TestCase):
856
973
857
974
def test_actions_requires_client_or_all(self):
858
975
for action, value in self.actions.items():
859
options = self.parser.parse_args()
860
setattr(options, action, value)
976
args = self.actionargs(action, value)
861
977
with self.assertParseError():
862
self.check_option_syntax(options)
978
self.parse_args(args)
863
979
864
# This mostly corresponds to the definition from has_actions() in
980
# This mostly corresponds to the definition from has_commands() in
865
981
# check_option_syntax()
866
982
actions = {
867
# The actual values set here are not that important, but we do
868
# at least stick to the correct types, even though they are
869
# never used
870
"enable": True,
871
"disable": True,
872
"bump_timeout": True,
873
"start_checker": True,
874
"stop_checker": True,
875
"is_enabled": True,
876
"remove": True,
877
"checker": "x",
878
"timeout": datetime.timedelta(),
879
"extended_timeout": datetime.timedelta(),
880
"interval": datetime.timedelta(),
881
"approved_by_default": True,
882
"approval_delay": datetime.timedelta(),
883
"approval_duration": datetime.timedelta(),
884
"host": "x",
885
"secret": io.BytesIO(b"x"),
886
"approve": True,
887
"deny": True,
983
"--enable": None,
984
"--disable": None,
985
"--bump-timeout": None,
986
"--start-checker": None,
987
"--stop-checker": None,
988
"--is-enabled": None,
989
"--remove": None,
990
"--checker": "x",
991
"--timeout": "PT0S",
992
"--extended-timeout": "PT0S",
993
"--interval": "PT0S",
994
"--approve-by-default": None,
995
"--deny-by-default": None,
996
"--approval-delay": "PT0S",
997
"--approval-duration": "PT0S",
998
"--host": "hostname",
999
"--secret": "/dev/null",
1000
"--approve": None,
1001
"--deny": None,
888
1002
}
889
1003
1004
@staticmethod
1005
def actionargs(action, value, *args):
1006
if value is not None:
1007
return [action, value] + list(args)
1008
else:
1009
return [action] + list(args)
1010
890
1011
@contextlib.contextmanager
891
1012
def assertParseError(self):
892
1013
with self.assertRaises(SystemExit) as e:
893
with self.temporarily_suppress_stderr():
1014
with self.redirect_stderr_to_devnull():
894
1015
yield
895
1016
# Exit code from argparse is guaranteed to be "2". Reference:
896
1017
# https://docs.python.org/3/library
897
1018
# /argparse.html#exiting-methods
898
self.assertEqual(e.exception.code, 2)
1019
self.assertEqual(2, e.exception.code)
1020
1021
def parse_args(self, args):
1022
options = self.parser.parse_args(args)
1023
check_option_syntax(self.parser, options)
899
1024
900
1025
@staticmethod
901
1026
@contextlib.contextmanager
902
def temporarily_suppress_stderr():
903
null = os.open(os.path.devnull, os.O_RDWR)
904
stderrcopy = os.dup(sys.stderr.fileno())
905
os.dup2(null, sys.stderr.fileno())
906
os.close(null)
907
try:
908
yield
909
finally:
910
# restore stderr
911
os.dup2(stderrcopy, sys.stderr.fileno())
912
os.close(stderrcopy)
1027
def redirect_stderr_to_devnull():
1028
old_stderr = sys.stderr
1029
with contextlib.closing(open(os.devnull, "w")) as null:
1030
sys.stderr = null
1031
try:
1032
yield
1033
finally:
1034
sys.stderr = old_stderr
913
1035
914
1036
def check_option_syntax(self, options):
915
1037
check_option_syntax(self.parser, options)
916
1038
917
def test_actions_conflicts_with_verbose(self):
918
for action, value in self.actions.items():
919
options = self.parser.parse_args()
920
setattr(options, action, value)
921
options.verbose = True
922
with self.assertParseError():
923
self.check_option_syntax(options)
1039
def test_actions_all_conflicts_with_verbose(self):
1040
for action, value in self.actions.items():
1041
args = self.actionargs(action, value, "--all",
1042
"--verbose")
1043
with self.assertParseError():
1044
self.parse_args(args)
1045
1046
def test_actions_with_client_conflicts_with_verbose(self):
1047
for action, value in self.actions.items():
1048
args = self.actionargs(action, value, "--verbose",
1049
"client")
1050
with self.assertParseError():
1051
self.parse_args(args)
924
1052
925
1053
def test_dump_json_conflicts_with_verbose(self):
926
options = self.parser.parse_args()
927
options.dump_json = True
928
options.verbose = True
1054
args = ["--dump-json", "--verbose"]
929
1055
with self.assertParseError():
930
self.check_option_syntax(options)
1056
self.parse_args(args)
931
1057
932
1058
def test_dump_json_conflicts_with_action(self):
933
1059
for action, value in self.actions.items():
934
options = self.parser.parse_args()
935
setattr(options, action, value)
936
options.dump_json = True
1060
args = self.actionargs(action, value, "--dump-json")
937
1061
with self.assertParseError():
938
self.check_option_syntax(options)
1062
self.parse_args(args)
939
1063
940
1064
def test_all_can_not_be_alone(self):
941
options = self.parser.parse_args()
942
options.all = True
1065
args = ["--all"]
943
1066
with self.assertParseError():
944
self.check_option_syntax(options)
1067
self.parse_args(args)
945
1068
946
1069
def test_all_is_ok_with_any_action(self):
947
1070
for action, value in self.actions.items():
948
options = self.parser.parse_args()
949
setattr(options, action, value)
950
options.all = True
951
self.check_option_syntax(options)
1071
args = self.actionargs(action, value, "--all")
1072
self.parse_args(args)
1073
1074
def test_any_action_is_ok_with_one_client(self):
1075
for action, value in self.actions.items():
1076
args = self.actionargs(action, value, "client")
1077
self.parse_args(args)
1078
1079
def test_one_client_with_all_actions_except_is_enabled(self):
1080
for action, value in self.actions.items():
1081
if action == "--is-enabled":
1082
continue
1083
args = self.actionargs(action, value, "client")
1084
self.parse_args(args)
1085
1086
def test_two_clients_with_all_actions_except_is_enabled(self):
1087
for action, value in self.actions.items():
1088
if action == "--is-enabled":
1089
continue
1090
args = self.actionargs(action, value, "client1",
1091
"client2")
1092
self.parse_args(args)
1093
1094
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1095
for action, value in self.actions.items():
1096
if action == "--is-enabled":
1097
continue
1098
args = self.actionargs(action, value, "client1",
1099
"client2")
1100
self.parse_args(args)
952
1101
953
1102
def test_is_enabled_fails_without_client(self):
954
options = self.parser.parse_args()
955
options.is_enabled = True
1103
args = ["--is-enabled"]
956
1104
with self.assertParseError():
957
self.check_option_syntax(options)
958
959
def test_is_enabled_works_with_one_client(self):
960
options = self.parser.parse_args()
961
options.is_enabled = True
962
options.client = ["foo"]
963
self.check_option_syntax(options)
1105
self.parse_args(args)
964
1106
965
1107
def test_is_enabled_fails_with_two_clients(self):
966
options = self.parser.parse_args()
967
options.is_enabled = True
968
options.client = ["foo", "barbar"]
1108
args = ["--is-enabled", "client1", "client2"]
969
1109
with self.assertParseError():
970
self.check_option_syntax(options)
1110
self.parse_args(args)
971
1111
972
1112
def test_remove_can_only_be_combined_with_action_deny(self):
973
1113
for action, value in self.actions.items():
974
if action in {"remove", "deny"}:
1114
if action in {"--remove", "--deny"}:
975
1115
continue
976
options = self.parser.parse_args()
977
setattr(options, action, value)
978
options.all = True
979
options.remove = True
1116
args = self.actionargs(action, value, "--all",
1117
"--remove")
980
1118
with self.assertParseError():
981
self.check_option_syntax(options)
1119
self.parse_args(args)
1120
1121
1122
class Test_dbus_exceptions(unittest.TestCase):
1123
1124
def test_dbus_ConnectFailed_is_Error(self):
1125
with self.assertRaises(dbus.Error):
1126
raise dbus.ConnectFailed()
1127
1128
1129
class Test_dbus_MandosBus(unittest.TestCase):
1130
1131
class MockMandosBus(dbus.MandosBus):
1132
def __init__(self):
1133
self._name = "se.recompile.Mandos"
1134
self._server_path = "/"
1135
self._server_interface = "se.recompile.Mandos"
1136
self._client_interface = "se.recompile.Mandos.Client"
1137
self.calls = []
1138
self.call_method_return = Unique()
1139
1140
def call_method(self, methodname, busname, objectpath,
1141
interface, *args):
1142
self.calls.append((methodname, busname, objectpath,
1143
interface, args))
1144
return self.call_method_return
1145
1146
def setUp(self):
1147
self.bus = self.MockMandosBus()
1148
1149
def test_set_client_property(self):
1150
self.bus.set_client_property("objectpath", "key", "value")
1151
expected_call = ("Set", self.bus._name, "objectpath",
1152
"org.freedesktop.DBus.Properties",
1153
(self.bus._client_interface, "key", "value"))
1154
self.assertIn(expected_call, self.bus.calls)
1155
1156
def test_call_client_method(self):
1157
ret = self.bus.call_client_method("objectpath", "methodname")
1158
self.assertIs(self.bus.call_method_return, ret)
1159
expected_call = ("methodname", self.bus._name, "objectpath",
1160
self.bus._client_interface, ())
1161
self.assertIn(expected_call, self.bus.calls)
1162
1163
def test_call_client_method_with_args(self):
1164
args = (Unique(), Unique())
1165
ret = self.bus.call_client_method("objectpath", "methodname",
1166
*args)
1167
self.assertIs(self.bus.call_method_return, ret)
1168
expected_call = ("methodname", self.bus._name, "objectpath",
1169
self.bus._client_interface,
1170
(args[0], args[1]))
1171
self.assertIn(expected_call, self.bus.calls)
1172
1173
def test_get_clients_and_properties(self):
1174
managed_objects = {
1175
"objectpath": {
1176
self.bus._client_interface: {
1177
"key": "value",
1178
"bool": True,
1179
},
1180
"irrelevant_interface": {
1181
"key": "othervalue",
1182
"bool": False,
1183
},
1184
},
1185
"other_objectpath": {
1186
"other_irrelevant_interface": {
1187
"key": "value 3",
1188
"bool": None,
1189
},
1190
},
1191
}
1192
expected_clients_and_properties = {
1193
"objectpath": {
1194
"key": "value",
1195
"bool": True,
1196
}
1197
}
1198
self.bus.call_method_return = managed_objects
1199
ret = self.bus.get_clients_and_properties()
1200
self.assertDictEqual(expected_clients_and_properties, ret)
1201
expected_call = ("GetManagedObjects", self.bus._name,
1202
self.bus._server_path,
1203
"org.freedesktop.DBus.ObjectManager", ())
1204
self.assertIn(expected_call, self.bus.calls)
1205
1206
def test_call_server_method(self):
1207
ret = self.bus.call_server_method("methodname")
1208
self.assertIs(self.bus.call_method_return, ret)
1209
expected_call = ("methodname", self.bus._name,
1210
self.bus._server_path,
1211
self.bus._server_interface, ())
1212
self.assertIn(expected_call, self.bus.calls)
1213
1214
def test_call_server_method_with_args(self):
1215
args = (Unique(), Unique())
1216
ret = self.bus.call_server_method("methodname", *args)
1217
self.assertIs(self.bus.call_method_return, ret)
1218
expected_call = ("methodname", self.bus._name,
1219
self.bus._server_path,
1220
self.bus._server_interface,
1221
(args[0], args[1]))
1222
self.assertIn(expected_call, self.bus.calls)
1223
1224
1225
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1226
1227
def MockDBusPython_func(self, func):
1228
class mock_dbus_python(object):
1229
"""mock dbus-python module"""
1230
class exceptions(object):
1231
"""Pseudo-namespace"""
1232
class DBusException(Exception):
1233
pass
1234
class SystemBus(object):
1235
@staticmethod
1236
def get_object(busname, objectpath):
1237
DBusObject = collections.namedtuple(
1238
"DBusObject", ("methodname",))
1239
def method(*args, **kwargs):
1240
self.assertEqual({"dbus_interface":
1241
"interface"},
1242
kwargs)
1243
return func(*args)
1244
return DBusObject(methodname=method)
1245
class Boolean(object):
1246
def __init__(self, value):
1247
self.value = bool(value)
1248
def __bool__(self):
1249
return self.value
1250
if sys.version_info.major == 2:
1251
__nonzero__ = __bool__
1252
class ObjectPath(str):
1253
pass
1254
class Dictionary(dict):
1255
pass
1256
return mock_dbus_python
1257
1258
def call_method(self, bus, methodname, busname, objectpath,
1259
interface, *args):
1260
with self.assertLogs(log, logging.DEBUG):
1261
return bus.call_method(methodname, busname, objectpath,
1262
interface, *args)
1263
1264
def test_call_method_returns(self):
1265
expected_method_return = Unique()
1266
method_args = (Unique(), Unique())
1267
def func(*args):
1268
self.assertEqual(len(method_args), len(args))
1269
for marg, arg in zip(method_args, args):
1270
self.assertIs(marg, arg)
1271
return expected_method_return
1272
mock_dbus_python = self.MockDBusPython_func(func)
1273
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1274
ret = self.call_method(bus, "methodname", "busname",
1275
"objectpath", "interface",
1276
*method_args)
1277
self.assertIs(ret, expected_method_return)
1278
1279
def test_call_method_filters_bool_true(self):
1280
def func():
1281
return method_return
1282
mock_dbus_python = self.MockDBusPython_func(func)
1283
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1284
method_return = mock_dbus_python.Boolean(True)
1285
ret = self.call_method(bus, "methodname", "busname",
1286
"objectpath", "interface")
1287
self.assertTrue(ret)
1288
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1289
1290
def test_call_method_filters_bool_false(self):
1291
def func():
1292
return method_return
1293
mock_dbus_python = self.MockDBusPython_func(func)
1294
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1295
method_return = mock_dbus_python.Boolean(False)
1296
ret = self.call_method(bus, "methodname", "busname",
1297
"objectpath", "interface")
1298
self.assertFalse(ret)
1299
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1300
1301
def test_call_method_filters_objectpath(self):
1302
def func():
1303
return method_return
1304
mock_dbus_python = self.MockDBusPython_func(func)
1305
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1306
method_return = mock_dbus_python.ObjectPath("objectpath")
1307
ret = self.call_method(bus, "methodname", "busname",
1308
"objectpath", "interface")
1309
self.assertEqual("objectpath", ret)
1310
self.assertIsNot("objectpath", ret)
1311
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1312
1313
def test_call_method_filters_booleans_in_dict(self):
1314
def func():
1315
return method_return
1316
mock_dbus_python = self.MockDBusPython_func(func)
1317
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1318
method_return = mock_dbus_python.Dictionary(
1319
{mock_dbus_python.Boolean(True):
1320
mock_dbus_python.Boolean(False),
1321
mock_dbus_python.Boolean(False):
1322
mock_dbus_python.Boolean(True)})
1323
ret = self.call_method(bus, "methodname", "busname",
1324
"objectpath", "interface")
1325
expected_method_return = {True: False,
1326
False: True}
1327
self.assertEqual(expected_method_return, ret)
1328
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1329
1330
def test_call_method_filters_objectpaths_in_dict(self):
1331
def func():
1332
return method_return
1333
mock_dbus_python = self.MockDBusPython_func(func)
1334
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1335
method_return = mock_dbus_python.Dictionary(
1336
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1337
mock_dbus_python.ObjectPath("objectpath_value_1"),
1338
mock_dbus_python.ObjectPath("objectpath_key_2"):
1339
mock_dbus_python.ObjectPath("objectpath_value_2")})
1340
ret = self.call_method(bus, "methodname", "busname",
1341
"objectpath", "interface")
1342
expected_method_return = {str(key): str(value)
1343
for key, value in
1344
method_return.items()}
1345
self.assertEqual(expected_method_return, ret)
1346
self.assertIsInstance(ret, dict)
1347
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1348
1349
def test_call_method_filters_dict_in_dict(self):
1350
def func():
1351
return method_return
1352
mock_dbus_python = self.MockDBusPython_func(func)
1353
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1354
method_return = mock_dbus_python.Dictionary(
1355
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1356
"key12": "value12"}),
1357
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1358
"key22": "value22"})})
1359
ret = self.call_method(bus, "methodname", "busname",
1360
"objectpath", "interface")
1361
expected_method_return = {
1362
"key1": {"key11": "value11",
1363
"key12": "value12"},
1364
"key2": {"key21": "value21",
1365
"key22": "value22"},
1366
}
1367
self.assertEqual(expected_method_return, ret)
1368
self.assertIsInstance(ret, dict)
1369
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1370
for key, value in ret.items():
1371
self.assertIsInstance(value, dict)
1372
self.assertEqual(expected_method_return[key], value)
1373
self.assertNotIsInstance(value,
1374
mock_dbus_python.Dictionary)
1375
1376
def test_call_method_filters_dict_three_deep(self):
1377
def func():
1378
return method_return
1379
mock_dbus_python = self.MockDBusPython_func(func)
1380
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1381
method_return = mock_dbus_python.Dictionary(
1382
{"key1":
1383
mock_dbus_python.Dictionary(
1384
{"key2":
1385
mock_dbus_python.Dictionary(
1386
{"key3":
1387
mock_dbus_python.Boolean(True),
1388
}),
1389
}),
1390
})
1391
ret = self.call_method(bus, "methodname", "busname",
1392
"objectpath", "interface")
1393
expected_method_return = {"key1": {"key2": {"key3": True}}}
1394
self.assertEqual(expected_method_return, ret)
1395
self.assertIsInstance(ret, dict)
1396
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1397
self.assertIsInstance(ret["key1"], dict)
1398
self.assertNotIsInstance(ret["key1"],
1399
mock_dbus_python.Dictionary)
1400
self.assertIsInstance(ret["key1"]["key2"], dict)
1401
self.assertNotIsInstance(ret["key1"]["key2"],
1402
mock_dbus_python.Dictionary)
1403
self.assertTrue(ret["key1"]["key2"]["key3"])
1404
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1405
mock_dbus_python.Boolean)
1406
1407
def test_call_method_handles_exception(self):
1408
dbus_logger = logging.getLogger("dbus.proxies")
1409
1410
def func():
1411
dbus_logger.error("Test")
1412
raise mock_dbus_python.exceptions.DBusException()
1413
1414
mock_dbus_python = self.MockDBusPython_func(func)
1415
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1416
1417
class CountingHandler(logging.Handler):
1418
count = 0
1419
def emit(self, record):
1420
self.count += 1
1421
1422
counting_handler = CountingHandler()
1423
1424
dbus_logger.addHandler(counting_handler)
1425
1426
try:
1427
with self.assertRaises(dbus.Error) as e:
1428
self.call_method(bus, "methodname", "busname",
1429
"objectpath", "interface")
1430
finally:
1431
dbus_logger.removeFilter(counting_handler)
1432
1433
self.assertNotIsInstance(e, dbus.ConnectFailed)
1434
1435
# Make sure the dbus logger was suppressed
1436
self.assertEqual(0, counting_handler.count)
1437
1438
def test_get_object_converts_to_correct_exception(self):
1439
bus = dbus_python_adapter.SystemBus(
1440
self.fake_dbus_python_raises_exception_on_connect)
1441
with self.assertRaises(dbus.ConnectFailed):
1442
self.call_method(bus, "methodname", "busname",
1443
"objectpath", "interface")
1444
1445
class fake_dbus_python_raises_exception_on_connect(object):
1446
"""fake dbus-python module"""
1447
class exceptions(object):
1448
"""Pseudo-namespace"""
1449
class DBusException(Exception):
1450
pass
1451
1452
@classmethod
1453
def SystemBus(cls):
1454
def get_object(busname, objectpath):
1455
raise cls.exceptions.DBusException()
1456
Bus = collections.namedtuple("Bus", ["get_object"])
1457
return Bus(get_object=get_object)
1458
1459
1460
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1461
class mock_dbus_python(object):
1462
"""mock dbus-python modules"""
1463
class SystemBus(object):
1464
@staticmethod
1465
def get_object(busname, objectpath):
1466
return Unique()
1467
1468
def setUp(self):
1469
self.bus = dbus_python_adapter.CachingBus(
1470
self.mock_dbus_python)
1471
1472
def test_returns_distinct_objectpaths(self):
1473
obj1 = self.bus.get_object("busname", "objectpath1")
1474
self.assertIsInstance(obj1, Unique)
1475
obj2 = self.bus.get_object("busname", "objectpath2")
1476
self.assertIsInstance(obj2, Unique)
1477
self.assertIsNot(obj1, obj2)
1478
1479
def test_returns_distinct_busnames(self):
1480
obj1 = self.bus.get_object("busname1", "objectpath")
1481
self.assertIsInstance(obj1, Unique)
1482
obj2 = self.bus.get_object("busname2", "objectpath")
1483
self.assertIsInstance(obj2, Unique)
1484
self.assertIsNot(obj1, obj2)
1485
1486
def test_returns_distinct_both(self):
1487
obj1 = self.bus.get_object("busname1", "objectpath")
1488
self.assertIsInstance(obj1, Unique)
1489
obj2 = self.bus.get_object("busname2", "objectpath")
1490
self.assertIsInstance(obj2, Unique)
1491
self.assertIsNot(obj1, obj2)
1492
1493
def test_returns_same(self):
1494
obj1 = self.bus.get_object("busname", "objectpath")
1495
self.assertIsInstance(obj1, Unique)
1496
obj2 = self.bus.get_object("busname", "objectpath")
1497
self.assertIsInstance(obj2, Unique)
1498
self.assertIs(obj1, obj2)
1499
1500
def test_returns_same_old(self):
1501
obj1 = self.bus.get_object("busname1", "objectpath1")
1502
self.assertIsInstance(obj1, Unique)
1503
obj2 = self.bus.get_object("busname2", "objectpath2")
1504
self.assertIsInstance(obj2, Unique)
1505
obj1b = self.bus.get_object("busname1", "objectpath1")
1506
self.assertIsInstance(obj1b, Unique)
1507
self.assertIsNot(obj1, obj2)
1508
self.assertIsNot(obj2, obj1b)
1509
self.assertIs(obj1, obj1b)
982
1510
983
1511
984
1512
class Test_commands_from_options(unittest.TestCase):
1513
985
1514
def setUp(self):
986
1515
self.parser = argparse.ArgumentParser()
987
1516
add_command_line_options(self.parser)
988
1517
989
1518
def test_is_enabled(self):
990
self.assert_command_from_args(["--is-enabled", "foo"],
991
IsEnabledCmd)
1519
self.assert_command_from_args(["--is-enabled", "client"],
1520
command.IsEnabled)
992
1521
993
1522
def assert_command_from_args(self, args, command_cls,
994
1523
**cmd_attrs):
997
1526
options = self.parser.parse_args(args)
998
1527
check_option_syntax(self.parser, options)
999
1528
commands = commands_from_options(options)
1000
self.assertEqual(len(commands), 1)
1529
self.assertEqual(1, len(commands))
1001
1530
command = commands[0]
1002
1531
self.assertIsInstance(command, command_cls)
1003
1532
for key, value in cmd_attrs.items():
1004
self.assertEqual(getattr(command, key), value)
1533
self.assertEqual(value, getattr(command, key))
1005
1534
1006
1535
def test_is_enabled_short(self):
1007
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1536
self.assert_command_from_args(["-V", "client"],
1537
command.IsEnabled)
1008
1538
1009
1539
def test_approve(self):
1010
self.assert_command_from_args(["--approve", "foo"],
1011
ApproveCmd)
1540
self.assert_command_from_args(["--approve", "client"],
1541
command.Approve)
1012
1542
1013
1543
def test_approve_short(self):
1014
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1544
self.assert_command_from_args(["-A", "client"],
1545
command.Approve)
1015
1546
1016
1547
def test_deny(self):
1017
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1548
self.assert_command_from_args(["--deny", "client"],
1549
command.Deny)
1018
1550
1019
1551
def test_deny_short(self):
1020
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1552
self.assert_command_from_args(["-D", "client"], command.Deny)
1021
1553
1022
1554
def test_remove(self):
1023
self.assert_command_from_args(["--remove", "foo"],
1024
RemoveCmd)
1555
self.assert_command_from_args(["--remove", "client"],
1556
command.Remove)
1025
1557
1026
1558
def test_deny_before_remove(self):
1027
1559
options = self.parser.parse_args(["--deny", "--remove",
1028
"foo"])
1560
"client"])
1029
1561
check_option_syntax(self.parser, options)
1030
1562
commands = commands_from_options(options)
1031
self.assertEqual(len(commands), 2)
1032
self.assertIsInstance(commands[0], DenyCmd)
1033
self.assertIsInstance(commands[1], RemoveCmd)
1563
self.assertEqual(2, len(commands))
1564
self.assertIsInstance(commands[0], command.Deny)
1565
self.assertIsInstance(commands[1], command.Remove)
1034
1566
1035
1567
def test_deny_before_remove_reversed(self):
1036
1568
options = self.parser.parse_args(["--remove", "--deny",
1037
1569
"--all"])
1038
1570
check_option_syntax(self.parser, options)
1039
1571
commands = commands_from_options(options)
1040
self.assertEqual(len(commands), 2)
1041
self.assertIsInstance(commands[0], DenyCmd)
1042
self.assertIsInstance(commands[1], RemoveCmd)
1572
self.assertEqual(2, len(commands))
1573
self.assertIsInstance(commands[0], command.Deny)
1574
self.assertIsInstance(commands[1], command.Remove)
1043
1575
1044
1576
def test_remove_short(self):
1045
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1577
self.assert_command_from_args(["-r", "client"],
1578
command.Remove)
1046
1579
1047
1580
def test_dump_json(self):
1048
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1581
self.assert_command_from_args(["--dump-json"],
1582
command.DumpJSON)
1049
1583
1050
1584
def test_enable(self):
1051
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1585
self.assert_command_from_args(["--enable", "client"],
1586
command.Enable)
1052
1587
1053
1588
def test_enable_short(self):
1054
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1589
self.assert_command_from_args(["-e", "client"],
1590
command.Enable)
1055
1591
1056
1592
def test_disable(self):
1057
self.assert_command_from_args(["--disable", "foo"],
1058
DisableCmd)
1593
self.assert_command_from_args(["--disable", "client"],
1594
command.Disable)
1059
1595
1060
1596
def test_disable_short(self):
1061
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1597
self.assert_command_from_args(["-d", "client"],
1598
command.Disable)
1062
1599
1063
1600
def test_bump_timeout(self):
1064
self.assert_command_from_args(["--bump-timeout", "foo"],
1065
BumpTimeoutCmd)
1601
self.assert_command_from_args(["--bump-timeout", "client"],
1602
command.BumpTimeout)
1066
1603
1067
1604
def test_bump_timeout_short(self):
1068
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1605
self.assert_command_from_args(["-b", "client"],
1606
command.BumpTimeout)
1069
1607
1070
1608
def test_start_checker(self):
1071
self.assert_command_from_args(["--start-checker", "foo"],
1072
StartCheckerCmd)
1609
self.assert_command_from_args(["--start-checker", "client"],
1610
command.StartChecker)
1073
1611
1074
1612
def test_stop_checker(self):
1075
self.assert_command_from_args(["--stop-checker", "foo"],
1076
StopCheckerCmd)
1613
self.assert_command_from_args(["--stop-checker", "client"],
1614
command.StopChecker)
1077
1615
1078
1616
def test_approve_by_default(self):
1079
self.assert_command_from_args(["--approve-by-default", "foo"],
1080
ApproveByDefaultCmd)
1617
self.assert_command_from_args(["--approve-by-default",
1618
"client"],
1619
command.ApproveByDefault)
1081
1620
1082
1621
def test_deny_by_default(self):
1083
self.assert_command_from_args(["--deny-by-default", "foo"],
1084
DenyByDefaultCmd)
1622
self.assert_command_from_args(["--deny-by-default", "client"],
1623
command.DenyByDefault)
1085
1624
1086
1625
def test_checker(self):
1087
self.assert_command_from_args(["--checker", ":", "foo"],
1088
SetCheckerCmd, value_to_set=":")
1626
self.assert_command_from_args(["--checker", ":", "client"],
1627
command.SetChecker,
1628
value_to_set=":")
1089
1629
1090
1630
def test_checker_empty(self):
1091
self.assert_command_from_args(["--checker", "", "foo"],
1092
SetCheckerCmd, value_to_set="")
1631
self.assert_command_from_args(["--checker", "", "client"],
1632
command.SetChecker,
1633
value_to_set="")
1093
1634
1094
1635
def test_checker_short(self):
1095
self.assert_command_from_args(["-c", ":", "foo"],
1096
SetCheckerCmd, value_to_set=":")
1636
self.assert_command_from_args(["-c", ":", "client"],
1637
command.SetChecker,
1638
value_to_set=":")
1097
1639
1098
1640
def test_host(self):
1099
self.assert_command_from_args(["--host", "foo.example.org",
1100
"foo"], SetHostCmd,
1101
value_to_set="foo.example.org")
1641
self.assert_command_from_args(
1642
["--host", "client.example.org", "client"],
1643
command.SetHost, value_to_set="client.example.org")
1102
1644
1103
1645
def test_host_short(self):
1104
self.assert_command_from_args(["-H", "foo.example.org",
1105
"foo"], SetHostCmd,
1106
value_to_set="foo.example.org")
1646
self.assert_command_from_args(
1647
["-H", "client.example.org", "client"], command.SetHost,
1648
value_to_set="client.example.org")
1107
1649
1108
1650
def test_secret_devnull(self):
1109
1651
self.assert_command_from_args(["--secret", os.path.devnull,
1110
"foo"], SetSecretCmd,
1652
"client"], command.SetSecret,
1111
1653
value_to_set=b"")
1112
1654
1113
1655
def test_secret_tempfile(self):
1116
1658
f.write(value)
1117
1659
f.seek(0)
1118
1660
self.assert_command_from_args(["--secret", f.name,
1119
"foo"], SetSecretCmd,
1661
"client"],
1662
command.SetSecret,
1120
1663
value_to_set=value)
1121
1664
1122
1665
def test_secret_devnull_short(self):
1123
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1124
SetSecretCmd, value_to_set=b"")
1666
self.assert_command_from_args(["-s", os.path.devnull,
1667
"client"], command.SetSecret,
1668
value_to_set=b"")
1125
1669
1126
1670
def test_secret_tempfile_short(self):
1127
1671
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1128
1672
value = b"secret\0xyzzy\nbar"
1129
1673
f.write(value)
1130
1674
f.seek(0)
1131
self.assert_command_from_args(["-s", f.name, "foo"],
1132
SetSecretCmd,
1675
self.assert_command_from_args(["-s", f.name, "client"],
1676
command.SetSecret,
1133
1677
value_to_set=value)
1134
1678
1135
1679
def test_timeout(self):
1136
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1137
SetTimeoutCmd,
1680
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1681
command.SetTimeout,
1138
1682
value_to_set=300000)
1139
1683
1140
1684
def test_timeout_short(self):
1141
self.assert_command_from_args(["-t", "PT5M", "foo"],
1142
SetTimeoutCmd,
1685
self.assert_command_from_args(["-t", "PT5M", "client"],
1686
command.SetTimeout,
1143
1687
value_to_set=300000)
1144
1688
1145
1689
def test_extended_timeout(self):
1146
1690
self.assert_command_from_args(["--extended-timeout", "PT15M",
1147
"foo"],
1148
SetExtendedTimeoutCmd,
1691
"client"],
1692
command.SetExtendedTimeout,
1149
1693
value_to_set=900000)
1150
1694
1151
1695
def test_interval(self):
1152
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1153
SetIntervalCmd,
1696
self.assert_command_from_args(["--interval", "PT2M",
1697
"client"], command.SetInterval,
1154
1698
value_to_set=120000)
1155
1699
1156
1700
def test_interval_short(self):
1157
self.assert_command_from_args(["-i", "PT2M", "foo"],
1158
SetIntervalCmd,
1701
self.assert_command_from_args(["-i", "PT2M", "client"],
1702
command.SetInterval,
1159
1703
value_to_set=120000)
1160
1704
1161
1705
def test_approval_delay(self):
1162
1706
self.assert_command_from_args(["--approval-delay", "PT30S",
1163
"foo"], SetApprovalDelayCmd,
1707
"client"],
1708
command.SetApprovalDelay,
1164
1709
value_to_set=30000)
1165
1710
1166
1711
def test_approval_duration(self):
1167
1712
self.assert_command_from_args(["--approval-duration", "PT1S",
1168
"foo"], SetApprovalDurationCmd,
1713
"client"],
1714
command.SetApprovalDuration,
1169
1715
value_to_set=1000)
1170
1716
1171
1717
def test_print_table(self):
1172
self.assert_command_from_args([], PrintTableCmd,
1718
self.assert_command_from_args([], command.PrintTable,
1173
1719
verbose=False)
1174
1720
1175
1721
def test_print_table_verbose(self):
1176
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1722
self.assert_command_from_args(["--verbose"],
1723
command.PrintTable,
1177
1724
verbose=True)
1178
1725
1179
1726
def test_print_table_verbose_short(self):
1180
self.assert_command_from_args(["-v"], PrintTableCmd,
1727
self.assert_command_from_args(["-v"], command.PrintTable,
1181
1728
verbose=True)
1182
1729
1183
1730
1184
class TestCmd(unittest.TestCase):
1731
class TestCommand(unittest.TestCase):
1185
1732
"""Abstract class for tests of command classes"""
1186
1733
1187
def setUp(self):
1188
testcase = self
1189
class MockClient(object):
1190
def __init__(self, name, **attributes):
1191
self.__dbus_object_path__ = "/clients/{}".format(name)
1192
self.attributes = attributes
1193
self.attributes["Name"] = name
1194
self.calls = []
1195
def Set(self, interface, propname, value, dbus_interface):
1196
testcase.assertEqual(interface, client_dbus_interface)
1197
testcase.assertEqual(dbus_interface,
1198
dbus.PROPERTIES_IFACE)
1199
self.attributes[propname] = value
1200
def Get(self, interface, propname, dbus_interface):
1201
testcase.assertEqual(interface, client_dbus_interface)
1202
testcase.assertEqual(dbus_interface,
1203
dbus.PROPERTIES_IFACE)
1204
return self.attributes[propname]
1205
def Approve(self, approve, dbus_interface):
1206
testcase.assertEqual(dbus_interface,
1207
client_dbus_interface)
1208
self.calls.append(("Approve", (approve,
1209
dbus_interface)))
1210
self.client = MockClient(
1211
"foo",
1212
KeyID=("92ed150794387c03ce684574b1139a65"
1213
"94a34f895daaaf09fd8ea90a27cddb12"),
1214
Secret=b"secret",
1215
Host="foo.example.org",
1216
Enabled=dbus.Boolean(True),
1217
Timeout=300000,
1218
LastCheckedOK="2019-02-03T00:00:00",
1219
Created="2019-01-02T00:00:00",
1220
Interval=120000,
1221
Fingerprint=("778827225BA7DE539C5A"
1222
"7CFA59CFF7CDBD9A5920"),
1223
CheckerRunning=dbus.Boolean(False),
1224
LastEnabled="2019-01-03T00:00:00",
1225
ApprovalPending=dbus.Boolean(False),
1226
ApprovedByDefault=dbus.Boolean(True),
1227
LastApprovalRequest="",
1228
ApprovalDelay=0,
1229
ApprovalDuration=1000,
1230
Checker="fping -q -- %(host)s",
1231
ExtendedTimeout=900000,
1232
Expires="2019-02-04T00:00:00",
1233
LastCheckerStatus=0)
1234
self.other_client = MockClient(
1235
"barbar",
1236
KeyID=("0558568eedd67d622f5c83b35a115f79"
1237
"6ab612cff5ad227247e46c2b020f441c"),
1238
Secret=b"secretbar",
1239
Host="192.0.2.3",
1240
Enabled=dbus.Boolean(True),
1241
Timeout=300000,
1242
LastCheckedOK="2019-02-04T00:00:00",
1243
Created="2019-01-03T00:00:00",
1244
Interval=120000,
1245
Fingerprint=("3E393AEAEFB84C7E89E2"
1246
"F547B3A107558FCA3A27"),
1247
CheckerRunning=dbus.Boolean(True),
1248
LastEnabled="2019-01-04T00:00:00",
1249
ApprovalPending=dbus.Boolean(False),
1250
ApprovedByDefault=dbus.Boolean(False),
1251
LastApprovalRequest="2019-01-03T00:00:00",
1252
ApprovalDelay=30000,
1253
ApprovalDuration=93785000,
1254
Checker=":",
1255
ExtendedTimeout=900000,
1256
Expires="2019-02-05T00:00:00",
1257
LastCheckerStatus=-2)
1258
self.clients = collections.OrderedDict(
1259
[
1260
("/clients/foo", self.client.attributes),
1261
("/clients/barbar", self.other_client.attributes),
1262
])
1263
self.one_client = {"/clients/foo": self.client.attributes}
1264
1265
@property
1266
def bus(self):
1267
class Bus(object):
1268
@staticmethod
1269
def get_object(client_bus_name, path):
1270
self.assertEqual(client_bus_name, dbus_busname)
1271
return {
1272
"/clients/foo": self.client,
1273
"/clients/barbar": self.other_client,
1274
}[path]
1275
return Bus()
1276
1277
1278
class TestIsEnabledCmd(TestCmd):
1279
def test_is_enabled(self):
1280
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1281
properties)
1282
for client, properties
1283
in self.clients.items()))
1284
1285
def test_is_enabled_run_exits_successfully(self):
1286
with self.assertRaises(SystemExit) as e:
1287
IsEnabledCmd().run(self.one_client)
1288
if e.exception.code is not None:
1289
self.assertEqual(e.exception.code, 0)
1290
else:
1291
self.assertIsNone(e.exception.code)
1292
1293
def test_is_enabled_run_exits_with_failure(self):
1294
self.client.attributes["Enabled"] = dbus.Boolean(False)
1295
with self.assertRaises(SystemExit) as e:
1296
IsEnabledCmd().run(self.one_client)
1297
if isinstance(e.exception.code, int):
1298
self.assertNotEqual(e.exception.code, 0)
1299
else:
1300
self.assertIsNotNone(e.exception.code)
1301
1302
1303
class TestApproveCmd(TestCmd):
1304
def test_approve(self):
1305
ApproveCmd().run(self.clients, self.bus)
1306
for clientpath in self.clients:
1307
client = self.bus.get_object(dbus_busname, clientpath)
1308
self.assertIn(("Approve", (True, client_dbus_interface)),
1309
client.calls)
1310
1311
1312
class TestDenyCmd(TestCmd):
1313
def test_deny(self):
1314
DenyCmd().run(self.clients, self.bus)
1315
for clientpath in self.clients:
1316
client = self.bus.get_object(dbus_busname, clientpath)
1317
self.assertIn(("Approve", (False, client_dbus_interface)),
1318
client.calls)
1319
1320
1321
class TestRemoveCmd(TestCmd):
1322
def test_remove(self):
1323
class MockMandos(object):
1324
def __init__(self):
1325
self.calls = []
1326
def RemoveClient(self, dbus_path):
1327
self.calls.append(("RemoveClient", (dbus_path,)))
1328
mandos = MockMandos()
1329
super(TestRemoveCmd, self).setUp()
1330
RemoveCmd().run(self.clients, self.bus, mandos)
1331
self.assertEqual(len(mandos.calls), 2)
1332
for clientpath in self.clients:
1333
self.assertIn(("RemoveClient", (clientpath,)),
1334
mandos.calls)
1335
1336
1337
class TestDumpJSONCmd(TestCmd):
1338
def setUp(self):
1339
self.expected_json = {
1340
"foo": {
1734
class FakeMandosBus(dbus.MandosBus):
1735
def __init__(self, testcase):
1736
self.client_properties = {
1341
1737
"Name": "foo",
1342
1738
"KeyID": ("92ed150794387c03ce684574b1139a65"
1343
1739
"94a34f895daaaf09fd8ea90a27cddb12"),
1740
"Secret": b"secret",
1344
1741
"Host": "foo.example.org",
1345
1742
"Enabled": True,
1346
1743
"Timeout": 300000,
1360
1757
"ExtendedTimeout": 900000,
1361
1758
"Expires": "2019-02-04T00:00:00",
1362
1759
"LastCheckerStatus": 0,
1363
},
1364
"barbar": {
1760
}
1761
self.other_client_properties = {
1365
1762
"Name": "barbar",
1366
1763
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1367
1764
"6ab612cff5ad227247e46c2b020f441c"),
1765
"Secret": b"secretbar",
1368
1766
"Host": "192.0.2.3",
1369
1767
"Enabled": True,
1370
1768
"Timeout": 300000,
1384
1782
"ExtendedTimeout": 900000,
1385
1783
"Expires": "2019-02-05T00:00:00",
1386
1784
"LastCheckerStatus": -2,
1387
},
1388
}
1389
return super(TestDumpJSONCmd, self).setUp()
1390
1391
def test_normal(self):
1392
json_data = json.loads(DumpJSONCmd().output(self.clients))
1393
self.assertDictEqual(json_data, self.expected_json)
1394
1395
def test_one_client(self):
1396
clients = self.one_client
1397
json_data = json.loads(DumpJSONCmd().output(clients))
1785
}
1786
self.clients = collections.OrderedDict(
1787
[
1788
("client_objectpath", self.client_properties),
1789
("other_client_objectpath",
1790
self.other_client_properties),
1791
])
1792
self.one_client = {"client_objectpath":
1793
self.client_properties}
1794
self.testcase = testcase
1795
self.calls = []
1796
1797
def call_method(self, methodname, busname, objectpath,
1798
interface, *args):
1799
self.testcase.assertEqual("se.recompile.Mandos", busname)
1800
self.calls.append((methodname, busname, objectpath,
1801
interface, args))
1802
if interface == "org.freedesktop.DBus.Properties":
1803
if methodname == "Set":
1804
self.testcase.assertEqual(3, len(args))
1805
interface, key, value = args
1806
self.testcase.assertEqual(
1807
"se.recompile.Mandos.Client", interface)
1808
self.clients[objectpath][key] = value
1809
return
1810
elif interface == "se.recompile.Mandos":
1811
self.testcase.assertEqual("RemoveClient", methodname)
1812
self.testcase.assertEqual(1, len(args))
1813
clientpath = args[0]
1814
del self.clients[clientpath]
1815
return
1816
elif interface == "se.recompile.Mandos.Client":
1817
if methodname == "Approve":
1818
self.testcase.assertEqual(1, len(args))
1819
return
1820
raise ValueError()
1821
1822
def setUp(self):
1823
self.bus = self.FakeMandosBus(self)
1824
1825
1826
class TestBaseCommands(TestCommand):
1827
1828
def test_IsEnabled_exits_successfully(self):
1829
with self.assertRaises(SystemExit) as e:
1830
command.IsEnabled().run(self.bus.one_client)
1831
if e.exception.code is not None:
1832
self.assertEqual(0, e.exception.code)
1833
else:
1834
self.assertIsNone(e.exception.code)
1835
1836
def test_IsEnabled_exits_with_failure(self):
1837
self.bus.client_properties["Enabled"] = False
1838
with self.assertRaises(SystemExit) as e:
1839
command.IsEnabled().run(self.bus.one_client)
1840
if isinstance(e.exception.code, int):
1841
self.assertNotEqual(0, e.exception.code)
1842
else:
1843
self.assertIsNotNone(e.exception.code)
1844
1845
def test_Approve(self):
1846
busname = "se.recompile.Mandos"
1847
client_interface = "se.recompile.Mandos.Client"
1848
command.Approve().run(self.bus.clients, self.bus)
1849
for clientpath in self.bus.clients:
1850
self.assertIn(("Approve", busname, clientpath,
1851
client_interface, (True,)), self.bus.calls)
1852
1853
def test_Deny(self):
1854
busname = "se.recompile.Mandos"
1855
client_interface = "se.recompile.Mandos.Client"
1856
command.Deny().run(self.bus.clients, self.bus)
1857
for clientpath in self.bus.clients:
1858
self.assertIn(("Approve", busname, clientpath,
1859
client_interface, (False,)),
1860
self.bus.calls)
1861
1862
def test_Remove(self):
1863
command.Remove().run(self.bus.clients, self.bus)
1864
for clientpath in self.bus.clients:
1865
self.assertIn(("RemoveClient", dbus_busname,
1866
dbus_server_path, dbus_server_interface,
1867
(clientpath,)), self.bus.calls)
1868
1869
expected_json = {
1870
"foo": {
1871
"Name": "foo",
1872
"KeyID": ("92ed150794387c03ce684574b1139a65"
1873
"94a34f895daaaf09fd8ea90a27cddb12"),
1874
"Host": "foo.example.org",
1875
"Enabled": True,
1876
"Timeout": 300000,
1877
"LastCheckedOK": "2019-02-03T00:00:00",
1878
"Created": "2019-01-02T00:00:00",
1879
"Interval": 120000,
1880
"Fingerprint": ("778827225BA7DE539C5A"
1881
"7CFA59CFF7CDBD9A5920"),
1882
"CheckerRunning": False,
1883
"LastEnabled": "2019-01-03T00:00:00",
1884
"ApprovalPending": False,
1885
"ApprovedByDefault": True,
1886
"LastApprovalRequest": "",
1887
"ApprovalDelay": 0,
1888
"ApprovalDuration": 1000,
1889
"Checker": "fping -q -- %(host)s",
1890
"ExtendedTimeout": 900000,
1891
"Expires": "2019-02-04T00:00:00",
1892
"LastCheckerStatus": 0,
1893
},
1894
"barbar": {
1895
"Name": "barbar",
1896
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1897
"6ab612cff5ad227247e46c2b020f441c"),
1898
"Host": "192.0.2.3",
1899
"Enabled": True,
1900
"Timeout": 300000,
1901
"LastCheckedOK": "2019-02-04T00:00:00",
1902
"Created": "2019-01-03T00:00:00",
1903
"Interval": 120000,
1904
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1905
"F547B3A107558FCA3A27"),
1906
"CheckerRunning": True,
1907
"LastEnabled": "2019-01-04T00:00:00",
1908
"ApprovalPending": False,
1909
"ApprovedByDefault": False,
1910
"LastApprovalRequest": "2019-01-03T00:00:00",
1911
"ApprovalDelay": 30000,
1912
"ApprovalDuration": 93785000,
1913
"Checker": ":",
1914
"ExtendedTimeout": 900000,
1915
"Expires": "2019-02-05T00:00:00",
1916
"LastCheckerStatus": -2,
1917
},
1918
}
1919
1920
def test_DumpJSON_normal(self):
1921
with self.capture_stdout_to_buffer() as buffer:
1922
command.DumpJSON().run(self.bus.clients)
1923
json_data = json.loads(buffer.getvalue())
1924
self.assertDictEqual(self.expected_json, json_data)
1925
1926
@staticmethod
1927
@contextlib.contextmanager
1928
def capture_stdout_to_buffer():
1929
capture_buffer = io.StringIO()
1930
old_stdout = sys.stdout
1931
sys.stdout = capture_buffer
1932
try:
1933
yield capture_buffer
1934
finally:
1935
sys.stdout = old_stdout
1936
1937
def test_DumpJSON_one_client(self):
1938
with self.capture_stdout_to_buffer() as buffer:
1939
command.DumpJSON().run(self.bus.one_client)
1940
json_data = json.loads(buffer.getvalue())
1398
1941
expected_json = {"foo": self.expected_json["foo"]}
1399
self.assertDictEqual(json_data, expected_json)
1400
1401
1402
class TestPrintTableCmd(TestCmd):
1403
def test_normal(self):
1404
output = PrintTableCmd().output(self.clients.values())
1942
self.assertDictEqual(expected_json, json_data)
1943
1944
def test_PrintTable_normal(self):
1945
with self.capture_stdout_to_buffer() as buffer:
1946
command.PrintTable().run(self.bus.clients)
1405
1947
expected_output = "\n".join((
1406
1948
"Name Enabled Timeout Last Successful Check",
1407
1949
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1408
1950
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1409
))
1410
self.assertEqual(output, expected_output)
1951
)) + "\n"
1952
self.assertEqual(expected_output, buffer.getvalue())
1411
1953
1412
def test_verbose(self):
1413
output = PrintTableCmd(verbose=True).output(
1414
self.clients.values())
1954
def test_PrintTable_verbose(self):
1955
with self.capture_stdout_to_buffer() as buffer:
1956
command.PrintTable(verbose=True).run(self.bus.clients)
1415
1957
columns = (
1416
1958
(
1417
1959
"Name ",
1499
2041
)
1500
2042
)
1501
2043
num_lines = max(len(rows) for rows in columns)
1502
expected_output = "\n".join("".join(rows[line]
1503
for rows in columns)
1504
for line in range(num_lines))
1505
self.assertEqual(output, expected_output)
2044
expected_output = ("\n".join("".join(rows[line]
2045
for rows in columns)
2046
for line in range(num_lines))
2047
+ "\n")
2048
self.assertEqual(expected_output, buffer.getvalue())
1506
2049
1507
def test_one_client(self):
1508
output = PrintTableCmd().output(self.one_client.values())
2050
def test_PrintTable_one_client(self):
2051
with self.capture_stdout_to_buffer() as buffer:
2052
command.PrintTable().run(self.bus.one_client)
1509
2053
expected_output = "\n".join((
1510
2054
"Name Enabled Timeout Last Successful Check",
1511
2055
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1512
))
1513
self.assertEqual(output, expected_output)
1514
1515
1516
class Unique(object):
1517
"""Class for objects which exist only to be unique objects, since
1518
unittest.mock.sentinel only exists in Python 3.3"""
1519
1520
1521
class TestPropertyCmd(TestCmd):
1522
"""Abstract class for tests of PropertyCmd classes"""
2056
)) + "\n"
2057
self.assertEqual(expected_output, buffer.getvalue())
2058
2059
2060
class TestPropertySetterCmd(TestCommand):
2061
"""Abstract class for tests of command.PropertySetter classes"""
2062
1523
2063
def runTest(self):
1524
2064
if not hasattr(self, "command"):
1525
return
1526
values_to_get = getattr(self, "values_to_get",
1527
self.values_to_set)
1528
for value_to_set, value_to_get in zip(self.values_to_set,
1529
values_to_get):
1530
for clientpath in self.clients:
1531
client = self.bus.get_object(dbus_busname, clientpath)
1532
old_value = client.attributes[self.propname]
1533
self.assertNotIsInstance(old_value, Unique)
1534
client.attributes[self.propname] = Unique()
1535
self.run_command(value_to_set, self.clients)
1536
for clientpath in self.clients:
1537
client = self.bus.get_object(dbus_busname, clientpath)
1538
value = client.attributes[self.propname]
2065
return # Abstract TestCase class
2066
2067
if hasattr(self, "values_to_set"):
2068
cmd_args = [(value,) for value in self.values_to_set]
2069
values_to_get = getattr(self, "values_to_get",
2070
self.values_to_set)
2071
else:
2072
cmd_args = [() for x in range(len(self.values_to_get))]
2073
values_to_get = self.values_to_get
2074
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2075
for clientpath in self.bus.clients:
2076
self.bus.clients[clientpath][self.propname] = (
2077
Unique())
2078
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2079
for clientpath in self.bus.clients:
2080
value = (self.bus.clients[clientpath]
2081
[self.propname])
1539
2082
self.assertNotIsInstance(value, Unique)
1540
self.assertEqual(value, value_to_get)
1541
def run_command(self, value, clients):
1542
self.command().run(clients, self.bus)
1543
1544
1545
class TestEnableCmd(TestPropertyCmd):
1546
command = EnableCmd
1547
propname = "Enabled"
1548
values_to_set = [dbus.Boolean(True)]
1549
1550
1551
class TestDisableCmd(TestPropertyCmd):
1552
command = DisableCmd
1553
propname = "Enabled"
1554
values_to_set = [dbus.Boolean(False)]
1555
1556
1557
class TestBumpTimeoutCmd(TestPropertyCmd):
1558
command = BumpTimeoutCmd
2083
self.assertEqual(value_to_get, value)
2084
2085
2086
class TestEnableCmd(TestPropertySetterCmd):
2087
command = command.Enable
2088
propname = "Enabled"
2089
values_to_get = [True]
2090
2091
2092
class TestDisableCmd(TestPropertySetterCmd):
2093
command = command.Disable
2094
propname = "Enabled"
2095
values_to_get = [False]
2096
2097
2098
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2099
command = command.BumpTimeout
1559
2100
propname = "LastCheckedOK"
1560
values_to_set = [""]
1561
1562
1563
class TestStartCheckerCmd(TestPropertyCmd):
1564
command = StartCheckerCmd
1565
propname = "CheckerRunning"
1566
values_to_set = [dbus.Boolean(True)]
1567
1568
1569
class TestStopCheckerCmd(TestPropertyCmd):
1570
command = StopCheckerCmd
1571
propname = "CheckerRunning"
1572
values_to_set = [dbus.Boolean(False)]
1573
1574
1575
class TestApproveByDefaultCmd(TestPropertyCmd):
1576
command = ApproveByDefaultCmd
1577
propname = "ApprovedByDefault"
1578
values_to_set = [dbus.Boolean(True)]
1579
1580
1581
class TestDenyByDefaultCmd(TestPropertyCmd):
1582
command = DenyByDefaultCmd
1583
propname = "ApprovedByDefault"
1584
values_to_set = [dbus.Boolean(False)]
1585
1586
1587
class TestPropertyValueCmd(TestPropertyCmd):
1588
"""Abstract class for tests of PropertyValueCmd classes"""
1589
1590
def runTest(self):
1591
if type(self) is TestPropertyValueCmd:
1592
return
1593
return super(TestPropertyValueCmd, self).runTest()
1594
1595
def run_command(self, value, clients):
1596
self.command(value).run(clients, self.bus)
1597
1598
1599
class TestSetCheckerCmd(TestPropertyValueCmd):
1600
command = SetCheckerCmd
2101
values_to_get = [""]
2102
2103
2104
class TestStartCheckerCmd(TestPropertySetterCmd):
2105
command = command.StartChecker
2106
propname = "CheckerRunning"
2107
values_to_get = [True]
2108
2109
2110
class TestStopCheckerCmd(TestPropertySetterCmd):
2111
command = command.StopChecker
2112
propname = "CheckerRunning"
2113
values_to_get = [False]
2114
2115
2116
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2117
command = command.ApproveByDefault
2118
propname = "ApprovedByDefault"
2119
values_to_get = [True]
2120
2121
2122
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2123
command = command.DenyByDefault
2124
propname = "ApprovedByDefault"
2125
values_to_get = [False]
2126
2127
2128
class TestSetCheckerCmd(TestPropertySetterCmd):
2129
command = command.SetChecker
1601
2130
propname = "Checker"
1602
2131
values_to_set = ["", ":", "fping -q -- %s"]
1603
2132
1604
2133
1605
class TestSetHostCmd(TestPropertyValueCmd):
1606
command = SetHostCmd
2134
class TestSetHostCmd(TestPropertySetterCmd):
2135
command = command.SetHost
1607
2136
propname = "Host"
1608
values_to_set = ["192.0.2.3", "foo.example.org"]
1609
1610
1611
class TestSetSecretCmd(TestPropertyValueCmd):
1612
command = SetSecretCmd
2137
values_to_set = ["192.0.2.3", "client.example.org"]
2138
2139
2140
class TestSetSecretCmd(TestPropertySetterCmd):
2141
command = command.SetSecret
1613
2142
propname = "Secret"
1614
2143
values_to_set = [io.BytesIO(b""),
1615
2144
io.BytesIO(b"secret\0xyzzy\nbar")]
1616
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1617
1618
1619
class TestSetTimeoutCmd(TestPropertyValueCmd):
1620
command = SetTimeoutCmd
2145
values_to_get = [f.getvalue() for f in values_to_set]
2146
2147
2148
class TestSetTimeoutCmd(TestPropertySetterCmd):
2149
command = command.SetTimeout
1621
2150
propname = "Timeout"
1622
2151
values_to_set = [datetime.timedelta(),
1623
2152
datetime.timedelta(minutes=5),
1624
2153
datetime.timedelta(seconds=1),
1625
2154
datetime.timedelta(weeks=1),
1626
2155
datetime.timedelta(weeks=52)]
1627
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1628
1629
1630
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1631
command = SetExtendedTimeoutCmd
2156
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2157
2158
2159
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
2160
command = command.SetExtendedTimeout
1632
2161
propname = "ExtendedTimeout"
1633
2162
values_to_set = [datetime.timedelta(),
1634
2163
datetime.timedelta(minutes=5),
1635
2164
datetime.timedelta(seconds=1),
1636
2165
datetime.timedelta(weeks=1),
1637
2166
datetime.timedelta(weeks=52)]
1638
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1639
1640
1641
class TestSetIntervalCmd(TestPropertyValueCmd):
1642
command = SetIntervalCmd
2167
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2168
2169
2170
class TestSetIntervalCmd(TestPropertySetterCmd):
2171
command = command.SetInterval
1643
2172
propname = "Interval"
1644
2173
values_to_set = [datetime.timedelta(),
1645
2174
datetime.timedelta(minutes=5),
1646
2175
datetime.timedelta(seconds=1),
1647
2176
datetime.timedelta(weeks=1),
1648
2177
datetime.timedelta(weeks=52)]
1649
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1650
1651
1652
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1653
command = SetApprovalDelayCmd
2178
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2179
2180
2181
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
2182
command = command.SetApprovalDelay
1654
2183
propname = "ApprovalDelay"
1655
2184
values_to_set = [datetime.timedelta(),
1656
2185
datetime.timedelta(minutes=5),
1657
2186
datetime.timedelta(seconds=1),
1658
2187
datetime.timedelta(weeks=1),
1659
2188
datetime.timedelta(weeks=52)]
1660
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1661
1662
1663
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1664
command = SetApprovalDurationCmd
2189
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2190
2191
2192
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
2193
command = command.SetApprovalDuration
1665
2194
propname = "ApprovalDuration"
1666
2195
values_to_set = [datetime.timedelta(),
1667
2196
datetime.timedelta(minutes=5),
1668
2197
datetime.timedelta(seconds=1),
1669
2198
datetime.timedelta(weeks=1),
1670
2199
datetime.timedelta(weeks=52)]
1671
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
2200
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1672
2201
1673
2202
1674
2203
1689
2218
return tests
1690
2219
1691
2220
if __name__ == "__main__":
1692
if should_only_run_tests():
1693
# Call using ./tdd-python-script --check [--verbose]
1694
unittest.main()
1695
else:
1696
main()
2221
try:
2222
if should_only_run_tests():
2223
# Call using ./tdd-python-script --check [--verbose]
2224
unittest.main()
2225
else:
2226
main()
2227
finally:
2228
logging.shutdown()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0