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 1055
- compare with another revision
- download diff
- download tarball
- view history from revision 1055
- Committer: Teddy Hogeborn
- Date: 2019-03-12 20:13:34 UTC
- Revision ID: teddy@recompile.se-20190312201334-my3htrprewjosuw5
mandos-ctl: Refactor
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
* mandos-ctl: Reorder everything into logical order; put main() first,
and put every subsequent definition as soon as possible
after its first use, except superclasses which need to
be placed before the classes inheriting from them.
Reorder all tests to match.
- files modified:
- mandos-ctl
added
removed
mandos-ctl
42
42
import json
43
43
import unittest
44
44
import logging
45
import io
46
import tempfile
47
import contextlib
45
48
46
49
import dbus
47
50
61
64
62
65
locale.setlocale(locale.LC_ALL, "")
63
66
64
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
65
domain = "se.recompile"
66
busname = domain + ".Mandos"
67
server_path = "/"
68
server_interface = domain + ".Mandos"
69
client_interface = domain + ".Mandos.Client"
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
70
73
version = "1.8.3"
71
74
72
75
76
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
77
80
78
81
79
def milliseconds_to_string(ms):
80
td = datetime.timedelta(0, 0, 0, ms)
81
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
82
.format(days="{}T".format(td.days) if td.days else "",
83
hours=td.seconds // 3600,
84
minutes=(td.seconds % 3600) // 60,
85
seconds=td.seconds % 60))
82
def main():
83
parser = argparse.ArgumentParser()
84
85
add_command_line_options(parser)
86
87
options = parser.parse_args()
88
89
check_option_syntax(parser, options)
90
91
clientnames = options.client
92
93
if options.debug:
94
log.setLevel(logging.DEBUG)
95
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
137
if not clientnames:
138
clients = {objpath: properties
139
for objpath, properties in mandos_clients.items()}
140
else:
141
for name in clientnames:
142
for objpath, properties in mandos_clients.items():
143
if properties["Name"] == name:
144
clients[objpath] = properties
145
break
146
else:
147
log.critical("Client not found on server: %r", name)
148
sys.exit(1)
149
150
# Run all commands on clients
151
commands = commands_from_options(options)
152
for command in commands:
153
command.run(clients, bus, mandos_serv)
154
155
156
def add_command_line_options(parser):
157
parser.add_argument("--version", action="version",
158
version="%(prog)s {}".format(version),
159
help="show version number and exit")
160
parser.add_argument("-a", "--all", action="store_true",
161
help="Select all clients")
162
parser.add_argument("-v", "--verbose", action="store_true",
163
help="Print all fields")
164
parser.add_argument("-j", "--dump-json", action="store_true",
165
help="Dump client data in JSON format")
166
enable_disable = parser.add_mutually_exclusive_group()
167
enable_disable.add_argument("-e", "--enable", action="store_true",
168
help="Enable client")
169
enable_disable.add_argument("-d", "--disable",
170
action="store_true",
171
help="disable client")
172
parser.add_argument("-b", "--bump-timeout", action="store_true",
173
help="Bump timeout for client")
174
start_stop_checker = parser.add_mutually_exclusive_group()
175
start_stop_checker.add_argument("--start-checker",
176
action="store_true",
177
help="Start checker for client")
178
start_stop_checker.add_argument("--stop-checker",
179
action="store_true",
180
help="Stop checker for client")
181
parser.add_argument("-V", "--is-enabled", action="store_true",
182
help="Check if client is enabled")
183
parser.add_argument("-r", "--remove", action="store_true",
184
help="Remove client")
185
parser.add_argument("-c", "--checker",
186
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")
193
approve_deny_default = parser.add_mutually_exclusive_group()
194
approve_deny_default.add_argument(
195
"--approve-by-default", action="store_true",
196
default=None, dest="approved_by_default",
197
help="Set client to be approved by default")
198
approve_deny_default.add_argument(
199
"--deny-by-default", action="store_false",
200
dest="approved_by_default",
201
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")
210
approve_deny = parser.add_mutually_exclusive_group()
211
approve_deny.add_argument(
212
"-A", "--approve", action="store_true",
213
help="Approve any current client request")
214
approve_deny.add_argument("-D", "--deny", action="store_true",
215
help="Deny any current client request")
216
parser.add_argument("--debug", action="store_true",
217
help="Debug mode (show D-Bus commands)")
218
parser.add_argument("--check", action="store_true",
219
help="Run self-test")
220
parser.add_argument("client", nargs="*", help="Client name")
221
222
223
def string_to_delta(interval):
224
"""Parse a string and return a datetime.timedelta"""
225
226
try:
227
return rfc3339_duration_to_delta(interval)
228
except ValueError as e:
229
log.warning("%s - Parsing as pre-1.6.1 interval instead",
230
' '.join(e.args))
231
return parse_pre_1_6_1_interval(interval)
86
232
87
233
88
234
def rfc3339_duration_to_delta(duration):
215
361
return value
216
362
217
363
218
def string_to_delta(interval):
219
"""Parse a string and return a datetime.timedelta"""
220
221
try:
222
return rfc3339_duration_to_delta(interval)
223
except ValueError as e:
224
log.warning("%s - Parsing as pre-1.6.1 interval instead",
225
' '.join(e.args))
226
return parse_pre_1_6_1_interval(interval)
227
228
229
364
def parse_pre_1_6_1_interval(interval):
230
365
"""Parse an interval string as documented by Mandos before 1.6.1,
231
366
and return a datetime.timedelta
269
404
return value
270
405
271
406
272
class TableOfClients(object):
273
tablewords = {
274
"Name": "Name",
275
"Enabled": "Enabled",
276
"Timeout": "Timeout",
277
"LastCheckedOK": "Last Successful Check",
278
"LastApprovalRequest": "Last Approval Request",
279
"Created": "Created",
280
"Interval": "Interval",
281
"Host": "Host",
282
"Fingerprint": "Fingerprint",
283
"KeyID": "Key ID",
284
"CheckerRunning": "Check Is Running",
285
"LastEnabled": "Last Enabled",
286
"ApprovalPending": "Approval Is Pending",
287
"ApprovedByDefault": "Approved By Default",
288
"ApprovalDelay": "Approval Delay",
289
"ApprovalDuration": "Approval Duration",
290
"Checker": "Checker",
291
"ExtendedTimeout": "Extended Timeout",
292
"Expires": "Expires",
293
"LastCheckerStatus": "Last Checker Status",
294
}
295
296
def __init__(self, clients, keywords, tablewords=None):
297
self.clients = clients
298
self.keywords = keywords
299
if tablewords is not None:
300
self.tablewords = tablewords
301
302
def __str__(self):
303
return "\n".join(self.rows())
304
305
if sys.version_info.major == 2:
306
__unicode__ = __str__
307
def __str__(self):
308
return str(self).encode(locale.getpreferredencoding())
309
310
def rows(self):
311
format_string = self.row_formatting_string()
312
rows = [self.header_line(format_string)]
313
rows.extend(self.client_line(client, format_string)
314
for client in self.clients)
315
return rows
316
317
def row_formatting_string(self):
318
"Format string used to format table rows"
319
return " ".join("{{{key}:{width}}}".format(
320
width=max(len(self.tablewords[key]),
321
*(len(self.string_from_client(client, key))
322
for client in self.clients)),
323
key=key)
324
for key in self.keywords)
325
326
def string_from_client(self, client, key):
327
return self.valuetostring(client[key], key)
328
329
@staticmethod
330
def valuetostring(value, keyword):
331
if isinstance(value, dbus.Boolean):
332
return "Yes" if value else "No"
333
if keyword in ("Timeout", "Interval", "ApprovalDelay",
334
"ApprovalDuration", "ExtendedTimeout"):
335
return milliseconds_to_string(value)
336
return str(value)
337
338
def header_line(self, format_string):
339
return format_string.format(**self.tablewords)
340
341
def client_line(self, client, format_string):
342
return format_string.format(
343
**{key: self.string_from_client(client, key)
344
for key in self.keywords})
345
346
347
def has_actions(options):
348
return any((options.enable,
349
options.disable,
350
options.bump_timeout,
351
options.start_checker,
352
options.stop_checker,
353
options.is_enabled,
354
options.remove,
355
options.checker is not None,
356
options.timeout is not None,
357
options.extended_timeout is not None,
358
options.interval is not None,
359
options.approved_by_default is not None,
360
options.approval_delay is not None,
361
options.approval_duration is not None,
362
options.host is not None,
363
options.secret is not None,
364
options.approve,
365
options.deny))
366
367
368
def main():
369
parser = argparse.ArgumentParser()
370
parser.add_argument("--version", action="version",
371
version="%(prog)s {}".format(version),
372
help="show version number and exit")
373
parser.add_argument("-a", "--all", action="store_true",
374
help="Select all clients")
375
parser.add_argument("-v", "--verbose", action="store_true",
376
help="Print all fields")
377
parser.add_argument("-j", "--dump-json", action="store_true",
378
help="Dump client data in JSON format")
379
parser.add_argument("-e", "--enable", action="store_true",
380
help="Enable client")
381
parser.add_argument("-d", "--disable", action="store_true",
382
help="disable client")
383
parser.add_argument("-b", "--bump-timeout", action="store_true",
384
help="Bump timeout for client")
385
parser.add_argument("--start-checker", action="store_true",
386
help="Start checker for client")
387
parser.add_argument("--stop-checker", action="store_true",
388
help="Stop checker for client")
389
parser.add_argument("-V", "--is-enabled", action="store_true",
390
help="Check if client is enabled")
391
parser.add_argument("-r", "--remove", action="store_true",
392
help="Remove client")
393
parser.add_argument("-c", "--checker",
394
help="Set checker command for client")
395
parser.add_argument("-t", "--timeout",
396
help="Set timeout for client")
397
parser.add_argument("--extended-timeout",
398
help="Set extended timeout for client")
399
parser.add_argument("-i", "--interval",
400
help="Set checker interval for client")
401
parser.add_argument("--approve-by-default", action="store_true",
402
default=None, dest="approved_by_default",
403
help="Set client to be approved by default")
404
parser.add_argument("--deny-by-default", action="store_false",
405
dest="approved_by_default",
406
help="Set client to be denied by default")
407
parser.add_argument("--approval-delay",
408
help="Set delay before client approve/deny")
409
parser.add_argument("--approval-duration",
410
help="Set duration of one client approval")
411
parser.add_argument("-H", "--host", help="Set host for client")
412
parser.add_argument("-s", "--secret",
413
type=argparse.FileType(mode="rb"),
414
help="Set password blob (file) for client")
415
parser.add_argument("-A", "--approve", action="store_true",
416
help="Approve any current client request")
417
parser.add_argument("-D", "--deny", action="store_true",
418
help="Deny any current client request")
419
parser.add_argument("--check", action="store_true",
420
help="Run self-test")
421
parser.add_argument("client", nargs="*", help="Client name")
422
options = parser.parse_args()
407
def check_option_syntax(parser, options):
408
"""Apply additional restrictions on options, not expressible in
409
argparse"""
410
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))
423
430
424
431
if has_actions(options) and not (options.client or options.all):
425
432
parser.error("Options require clients names or --all.")
430
437
parser.error("--dump-json can only be used alone.")
431
438
if options.all and not has_actions(options):
432
439
parser.error("--all requires an action.")
433
434
try:
435
bus = dbus.SystemBus()
436
mandos_dbus_objc = bus.get_object(busname, server_path)
437
except dbus.exceptions.DBusException:
438
log.critical("Could not connect to Mandos server")
440
if options.is_enabled and len(options.client) > 1:
441
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
447
448
449
def commands_from_options(options):
450
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())
486
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))
514
515
# If no command option has been given, show table of clients,
516
# optionally verbosely
517
if not commands:
518
commands.append(PrintTableCmd(verbose=options.verbose))
519
520
return commands
521
522
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)
439
542
sys.exit(1)
440
441
mandos_serv = dbus.Interface(mandos_dbus_objc,
442
dbus_interface=server_interface)
443
mandos_serv_object_manager = dbus.Interface(
444
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
445
446
# block stderr since dbus library prints to stderr
447
null = os.open(os.path.devnull, os.O_RDWR)
448
stderrcopy = os.dup(sys.stderr.fileno())
449
os.dup2(null, sys.stderr.fileno())
450
os.close(null)
451
try:
543
def is_enabled(self, client, properties):
544
return properties["Enabled"]
545
546
547
class ApproveCmd(Command):
548
def run_on_one_client(self, client, properties):
549
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
550
client.__dbus_object_path__, client_dbus_interface)
551
client.Approve(dbus.Boolean(True),
552
dbus_interface=client_dbus_interface)
553
554
555
class DenyCmd(Command):
556
def run_on_one_client(self, client, properties):
557
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
558
client.__dbus_object_path__, client_dbus_interface)
559
client.Approve(dbus.Boolean(False),
560
dbus_interface=client_dbus_interface)
561
562
563
class RemoveCmd(Command):
564
def run_on_one_client(self, client, properties):
565
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
566
server_dbus_path, server_dbus_interface,
567
str(client.__dbus_object_path__))
568
self.mandos.RemoveClient(client.__dbus_object_path__)
569
570
571
class PrintCmd(Command):
572
"""Abstract class for commands printing client details"""
573
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
574
"Created", "Interval", "Host", "KeyID",
575
"Fingerprint", "CheckerRunning", "LastEnabled",
576
"ApprovalPending", "ApprovedByDefault",
577
"LastApprovalRequest", "ApprovalDelay",
578
"ApprovalDuration", "Checker", "ExtendedTimeout",
579
"Expires", "LastCheckerStatus")
580
def run(self, clients, bus=None, mandos=None):
581
print(self.output(clients.values()))
582
def output(self, clients):
583
raise NotImplementedError()
584
585
586
class DumpJSONCmd(PrintCmd):
587
def output(self, clients):
588
data = {client["Name"]:
589
{key: self.dbus_boolean_to_bool(client[key])
590
for key in self.all_keywords}
591
for client in clients.values()}
592
return json.dumps(data, indent=4, separators=(',', ': '))
593
@staticmethod
594
def dbus_boolean_to_bool(value):
595
if isinstance(value, dbus.Boolean):
596
value = bool(value)
597
return value
598
599
600
class PrintTableCmd(PrintCmd):
601
def __init__(self, verbose=False):
602
self.verbose = verbose
603
604
def output(self, clients):
605
default_keywords = ("Name", "Enabled", "Timeout",
606
"LastCheckedOK")
607
keywords = default_keywords
608
if self.verbose:
609
keywords = self.all_keywords
610
return str(self.TableOfClients(clients, keywords))
611
612
class TableOfClients(object):
613
tableheaders = {
614
"Name": "Name",
615
"Enabled": "Enabled",
616
"Timeout": "Timeout",
617
"LastCheckedOK": "Last Successful Check",
618
"LastApprovalRequest": "Last Approval Request",
619
"Created": "Created",
620
"Interval": "Interval",
621
"Host": "Host",
622
"Fingerprint": "Fingerprint",
623
"KeyID": "Key ID",
624
"CheckerRunning": "Check Is Running",
625
"LastEnabled": "Last Enabled",
626
"ApprovalPending": "Approval Is Pending",
627
"ApprovedByDefault": "Approved By Default",
628
"ApprovalDelay": "Approval Delay",
629
"ApprovalDuration": "Approval Duration",
630
"Checker": "Checker",
631
"ExtendedTimeout": "Extended Timeout",
632
"Expires": "Expires",
633
"LastCheckerStatus": "Last Checker Status",
634
}
635
636
def __init__(self, clients, keywords, tableheaders=None):
637
self.clients = clients
638
self.keywords = keywords
639
if tableheaders is not None:
640
self.tableheaders = tableheaders
641
642
def __str__(self):
643
return "\n".join(self.rows())
644
645
if sys.version_info.major == 2:
646
__unicode__ = __str__
647
def __str__(self):
648
return str(self).encode(locale.getpreferredencoding())
649
650
def rows(self):
651
format_string = self.row_formatting_string()
652
rows = [self.header_line(format_string)]
653
rows.extend(self.client_line(client, format_string)
654
for client in self.clients)
655
return rows
656
657
def row_formatting_string(self):
658
"Format string used to format table rows"
659
return " ".join("{{{key}:{width}}}".format(
660
width=max(len(self.tableheaders[key]),
661
*(len(self.string_from_client(client, key))
662
for client in self.clients)),
663
key=key)
664
for key in self.keywords)
665
666
def string_from_client(self, client, key):
667
return self.valuetostring(client[key], key)
668
669
@staticmethod
670
def valuetostring(value, keyword):
671
if isinstance(value, dbus.Boolean):
672
return "Yes" if value else "No"
673
if keyword in ("Timeout", "Interval", "ApprovalDelay",
674
"ApprovalDuration", "ExtendedTimeout"):
675
return milliseconds_to_string(value)
676
return str(value)
677
678
def header_line(self, format_string):
679
return format_string.format(**self.tableheaders)
680
681
def client_line(self, client, format_string):
682
return format_string.format(
683
**{key: self.string_from_client(client, key)
684
for key in self.keywords})
685
686
687
def milliseconds_to_string(ms):
688
td = datetime.timedelta(0, 0, 0, ms)
689
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
690
.format(days="{}T".format(td.days) if td.days else "",
691
hours=td.seconds // 3600,
692
minutes=(td.seconds % 3600) // 60,
693
seconds=td.seconds % 60))
694
695
696
class PropertyCmd(Command):
697
"""Abstract class for Actions for setting one client property"""
698
def run_on_one_client(self, client, properties):
699
"""Set the Client's D-Bus property"""
700
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
701
client.__dbus_object_path__,
702
dbus.PROPERTIES_IFACE, client_dbus_interface,
703
self.propname, self.value_to_set
704
if not isinstance(self.value_to_set, dbus.Boolean)
705
else bool(self.value_to_set))
706
client.Set(client_dbus_interface, self.propname,
707
self.value_to_set,
708
dbus_interface=dbus.PROPERTIES_IFACE)
709
@property
710
def propname(self):
711
raise NotImplementedError()
712
713
714
class EnableCmd(PropertyCmd):
715
propname = "Enabled"
716
value_to_set = dbus.Boolean(True)
717
718
719
class DisableCmd(PropertyCmd):
720
propname = "Enabled"
721
value_to_set = dbus.Boolean(False)
722
723
724
class BumpTimeoutCmd(PropertyCmd):
725
propname = "LastCheckedOK"
726
value_to_set = ""
727
728
729
class StartCheckerCmd(PropertyCmd):
730
propname = "CheckerRunning"
731
value_to_set = dbus.Boolean(True)
732
733
734
class StopCheckerCmd(PropertyCmd):
735
propname = "CheckerRunning"
736
value_to_set = dbus.Boolean(False)
737
738
739
class ApproveByDefaultCmd(PropertyCmd):
740
propname = "ApprovedByDefault"
741
value_to_set = dbus.Boolean(True)
742
743
744
class DenyByDefaultCmd(PropertyCmd):
745
propname = "ApprovedByDefault"
746
value_to_set = dbus.Boolean(False)
747
748
749
class PropertyValueCmd(PropertyCmd):
750
"""Abstract class for PropertyCmd recieving a value as argument"""
751
def __init__(self, value):
752
self.value_to_set = value
753
754
755
class SetCheckerCmd(PropertyValueCmd):
756
propname = "Checker"
757
758
759
class SetHostCmd(PropertyValueCmd):
760
propname = "Host"
761
762
763
class SetSecretCmd(PropertyValueCmd):
764
propname = "Secret"
765
@property
766
def value_to_set(self):
767
return self._vts
768
@value_to_set.setter
769
def value_to_set(self, value):
770
"""When setting, read data from supplied file object"""
771
self._vts = value.read()
772
value.close()
773
774
775
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
776
"""Abstract class for PropertyValueCmd taking a value argument as
777
a datetime.timedelta() but should store it as milliseconds."""
778
@property
779
def value_to_set(self):
780
return self._vts
781
@value_to_set.setter
782
def value_to_set(self, value):
783
"""When setting, convert value from a datetime.timedelta"""
784
self._vts = int(round(value.total_seconds() * 1000))
785
786
787
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
788
propname = "Timeout"
789
790
791
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
792
propname = "ExtendedTimeout"
793
794
795
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
796
propname = "Interval"
797
798
799
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
800
propname = "ApprovalDelay"
801
802
803
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
804
propname = "ApprovalDuration"
805
806
807
808
class Test_string_to_delta(unittest.TestCase):
809
def test_handles_basic_rfc3339(self):
810
self.assertEqual(string_to_delta("PT0S"),
811
datetime.timedelta())
812
self.assertEqual(string_to_delta("P0D"),
813
datetime.timedelta())
814
self.assertEqual(string_to_delta("PT1S"),
815
datetime.timedelta(0, 1))
816
self.assertEqual(string_to_delta("PT2H"),
817
datetime.timedelta(0, 7200))
818
def test_falls_back_to_pre_1_6_1_with_warning(self):
819
# assertLogs only exists in Python 3.4
820
if hasattr(self, "assertLogs"):
821
with self.assertLogs(log, logging.WARNING):
822
value = string_to_delta("2h")
823
else:
824
class WarningFilter(logging.Filter):
825
"""Don't show, but record the presence of, warnings"""
826
def filter(self, record):
827
is_warning = record.levelno >= logging.WARNING
828
self.found = is_warning or getattr(self, "found",
829
False)
830
return not is_warning
831
warning_filter = WarningFilter()
832
log.addFilter(warning_filter)
833
try:
834
value = string_to_delta("2h")
835
finally:
836
log.removeFilter(warning_filter)
837
self.assertTrue(getattr(warning_filter, "found", False))
838
self.assertEqual(value, datetime.timedelta(0, 7200))
839
840
841
class Test_check_option_syntax(unittest.TestCase):
842
# This mostly corresponds to the definition from has_actions() in
843
# check_option_syntax()
844
actions = {
845
# The actual values set here are not that important, but we do
846
# at least stick to the correct types, even though they are
847
# never used
848
"enable": True,
849
"disable": True,
850
"bump_timeout": True,
851
"start_checker": True,
852
"stop_checker": True,
853
"is_enabled": True,
854
"remove": True,
855
"checker": "x",
856
"timeout": datetime.timedelta(),
857
"extended_timeout": datetime.timedelta(),
858
"interval": datetime.timedelta(),
859
"approved_by_default": True,
860
"approval_delay": datetime.timedelta(),
861
"approval_duration": datetime.timedelta(),
862
"host": "x",
863
"secret": io.BytesIO(b"x"),
864
"approve": True,
865
"deny": True,
866
}
867
868
def setUp(self):
869
self.parser = argparse.ArgumentParser()
870
add_command_line_options(self.parser)
871
872
@contextlib.contextmanager
873
def assertParseError(self):
874
with self.assertRaises(SystemExit) as e:
875
with self.temporarily_suppress_stderr():
876
yield
877
# Exit code from argparse is guaranteed to be "2". Reference:
878
# https://docs.python.org/3/library
879
# /argparse.html#exiting-methods
880
self.assertEqual(e.exception.code, 2)
881
882
@staticmethod
883
@contextlib.contextmanager
884
def temporarily_suppress_stderr():
885
null = os.open(os.path.devnull, os.O_RDWR)
886
stderrcopy = os.dup(sys.stderr.fileno())
887
os.dup2(null, sys.stderr.fileno())
888
os.close(null)
452
889
try:
453
mandos_clients = {path: ifs_and_props[client_interface]
454
for path, ifs_and_props in
455
mandos_serv_object_manager
456
.GetManagedObjects().items()
457
if client_interface in ifs_and_props}
890
yield
458
891
finally:
459
892
# restore stderr
460
893
os.dup2(stderrcopy, sys.stderr.fileno())
461
894
os.close(stderrcopy)
462
except dbus.exceptions.DBusException as e:
463
log.critical("Failed to access Mandos server through D-Bus:"
464
"\n%s", e)
465
sys.exit(1)
466
467
# Compile dict of (clients: properties) to process
468
clients = {}
469
470
if options.all or not options.client:
471
clients = {bus.get_object(busname, path): properties
472
for path, properties in mandos_clients.items()}
473
else:
474
for name in options.client:
475
for path, client in mandos_clients.items():
476
if client["Name"] == name:
477
client_objc = bus.get_object(busname, path)
478
clients[client_objc] = client
479
break
480
else:
481
log.critical("Client not found on server: %r", name)
482
sys.exit(1)
483
484
if not has_actions(options) and clients:
485
if options.verbose or options.dump_json:
486
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
487
"Created", "Interval", "Host", "KeyID",
488
"Fingerprint", "CheckerRunning",
489
"LastEnabled", "ApprovalPending",
490
"ApprovedByDefault", "LastApprovalRequest",
491
"ApprovalDelay", "ApprovalDuration",
492
"Checker", "ExtendedTimeout", "Expires",
493
"LastCheckerStatus")
494
else:
495
keywords = defaultkeywords
496
497
if options.dump_json:
498
json.dump({client["Name"]: {key:
499
bool(client[key])
500
if isinstance(client[key],
501
dbus.Boolean)
502
else client[key]
503
for key in keywords}
504
for client in clients.values()},
505
fp=sys.stdout, indent=4,
506
separators=(',', ': '))
507
print()
508
else:
509
print(TableOfClients(clients.values(), keywords))
510
else:
511
# Process each client in the list by all selected options
512
for client in clients:
513
514
def set_client_prop(prop, value):
515
"""Set a Client D-Bus property"""
516
client.Set(client_interface, prop, value,
517
dbus_interface=dbus.PROPERTIES_IFACE)
518
519
def set_client_prop_ms(prop, value):
520
"""Set a Client D-Bus property, converted
521
from a string to milliseconds."""
522
set_client_prop(prop,
523
string_to_delta(value).total_seconds()
524
* 1000)
525
526
if options.remove:
527
mandos_serv.RemoveClient(client.__dbus_object_path__)
528
if options.enable:
529
set_client_prop("Enabled", dbus.Boolean(True))
530
if options.disable:
531
set_client_prop("Enabled", dbus.Boolean(False))
532
if options.bump_timeout:
533
set_client_prop("LastCheckedOK", "")
534
if options.start_checker:
535
set_client_prop("CheckerRunning", dbus.Boolean(True))
536
if options.stop_checker:
537
set_client_prop("CheckerRunning", dbus.Boolean(False))
538
if options.is_enabled:
539
if client.Get(client_interface, "Enabled",
540
dbus_interface=dbus.PROPERTIES_IFACE):
541
sys.exit(0)
542
else:
543
sys.exit(1)
544
if options.checker is not None:
545
set_client_prop("Checker", options.checker)
546
if options.host is not None:
547
set_client_prop("Host", options.host)
548
if options.interval is not None:
549
set_client_prop_ms("Interval", options.interval)
550
if options.approval_delay is not None:
551
set_client_prop_ms("ApprovalDelay",
552
options.approval_delay)
553
if options.approval_duration is not None:
554
set_client_prop_ms("ApprovalDuration",
555
options.approval_duration)
556
if options.timeout is not None:
557
set_client_prop_ms("Timeout", options.timeout)
558
if options.extended_timeout is not None:
559
set_client_prop_ms("ExtendedTimeout",
560
options.extended_timeout)
561
if options.secret is not None:
562
set_client_prop("Secret",
563
dbus.ByteArray(options.secret.read()))
564
if options.approved_by_default is not None:
565
set_client_prop("ApprovedByDefault",
566
dbus.Boolean(options
567
.approved_by_default))
568
if options.approve:
569
client.Approve(dbus.Boolean(True),
570
dbus_interface=client_interface)
571
elif options.deny:
572
client.Approve(dbus.Boolean(False),
573
dbus_interface=client_interface)
574
575
895
896
def check_option_syntax(self, options):
897
check_option_syntax(self.parser, options)
898
899
def test_actions_requires_client_or_all(self):
900
for action, value in self.actions.items():
901
options = self.parser.parse_args()
902
setattr(options, action, value)
903
with self.assertParseError():
904
self.check_option_syntax(options)
905
906
def test_actions_conflicts_with_verbose(self):
907
for action, value in self.actions.items():
908
options = self.parser.parse_args()
909
setattr(options, action, value)
910
options.verbose = True
911
with self.assertParseError():
912
self.check_option_syntax(options)
913
914
def test_dump_json_conflicts_with_verbose(self):
915
options = self.parser.parse_args()
916
options.dump_json = True
917
options.verbose = True
918
with self.assertParseError():
919
self.check_option_syntax(options)
920
921
def test_dump_json_conflicts_with_action(self):
922
for action, value in self.actions.items():
923
options = self.parser.parse_args()
924
setattr(options, action, value)
925
options.dump_json = True
926
with self.assertParseError():
927
self.check_option_syntax(options)
928
929
def test_all_can_not_be_alone(self):
930
options = self.parser.parse_args()
931
options.all = True
932
with self.assertParseError():
933
self.check_option_syntax(options)
934
935
def test_all_is_ok_with_any_action(self):
936
for action, value in self.actions.items():
937
options = self.parser.parse_args()
938
setattr(options, action, value)
939
options.all = True
940
self.check_option_syntax(options)
941
942
def test_is_enabled_fails_without_client(self):
943
options = self.parser.parse_args()
944
options.is_enabled = True
945
with self.assertParseError():
946
self.check_option_syntax(options)
947
948
def test_is_enabled_works_with_one_client(self):
949
options = self.parser.parse_args()
950
options.is_enabled = True
951
options.client = ["foo"]
952
self.check_option_syntax(options)
953
954
def test_is_enabled_fails_with_two_clients(self):
955
options = self.parser.parse_args()
956
options.is_enabled = True
957
options.client = ["foo", "barbar"]
958
with self.assertParseError():
959
self.check_option_syntax(options)
960
961
def test_remove_can_only_be_combined_with_action_deny(self):
962
for action, value in self.actions.items():
963
if action in {"remove", "deny"}:
964
continue
965
options = self.parser.parse_args()
966
setattr(options, action, value)
967
options.all = True
968
options.remove = True
969
with self.assertParseError():
970
self.check_option_syntax(options)
971
972
973
class Test_command_from_options(unittest.TestCase):
974
def setUp(self):
975
self.parser = argparse.ArgumentParser()
976
add_command_line_options(self.parser)
977
def assert_command_from_args(self, args, command_cls,
978
**cmd_attrs):
979
"""Assert that parsing ARGS should result in an instance of
980
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
981
options = self.parser.parse_args(args)
982
check_option_syntax(self.parser, options)
983
commands = commands_from_options(options)
984
self.assertEqual(len(commands), 1)
985
command = commands[0]
986
self.assertIsInstance(command, command_cls)
987
for key, value in cmd_attrs.items():
988
self.assertEqual(getattr(command, key), value)
989
def test_print_table(self):
990
self.assert_command_from_args([], PrintTableCmd,
991
verbose=False)
992
993
def test_print_table_verbose(self):
994
self.assert_command_from_args(["--verbose"], PrintTableCmd,
995
verbose=True)
996
997
def test_print_table_verbose_short(self):
998
self.assert_command_from_args(["-v"], PrintTableCmd,
999
verbose=True)
1000
1001
def test_enable(self):
1002
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1003
1004
def test_enable_short(self):
1005
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1006
1007
def test_disable(self):
1008
self.assert_command_from_args(["--disable", "foo"],
1009
DisableCmd)
1010
1011
def test_disable_short(self):
1012
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1013
1014
def test_bump_timeout(self):
1015
self.assert_command_from_args(["--bump-timeout", "foo"],
1016
BumpTimeoutCmd)
1017
1018
def test_bump_timeout_short(self):
1019
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1020
1021
def test_start_checker(self):
1022
self.assert_command_from_args(["--start-checker", "foo"],
1023
StartCheckerCmd)
1024
1025
def test_stop_checker(self):
1026
self.assert_command_from_args(["--stop-checker", "foo"],
1027
StopCheckerCmd)
1028
1029
def test_remove(self):
1030
self.assert_command_from_args(["--remove", "foo"],
1031
RemoveCmd)
1032
1033
def test_remove_short(self):
1034
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1035
1036
def test_checker(self):
1037
self.assert_command_from_args(["--checker", ":", "foo"],
1038
SetCheckerCmd, value_to_set=":")
1039
1040
def test_checker_empty(self):
1041
self.assert_command_from_args(["--checker", "", "foo"],
1042
SetCheckerCmd, value_to_set="")
1043
1044
def test_checker_short(self):
1045
self.assert_command_from_args(["-c", ":", "foo"],
1046
SetCheckerCmd, value_to_set=":")
1047
1048
def test_timeout(self):
1049
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1050
SetTimeoutCmd,
1051
value_to_set=300000)
1052
1053
def test_timeout_short(self):
1054
self.assert_command_from_args(["-t", "PT5M", "foo"],
1055
SetTimeoutCmd,
1056
value_to_set=300000)
1057
1058
def test_extended_timeout(self):
1059
self.assert_command_from_args(["--extended-timeout", "PT15M",
1060
"foo"],
1061
SetExtendedTimeoutCmd,
1062
value_to_set=900000)
1063
1064
def test_interval(self):
1065
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1066
SetIntervalCmd,
1067
value_to_set=120000)
1068
1069
def test_interval_short(self):
1070
self.assert_command_from_args(["-i", "PT2M", "foo"],
1071
SetIntervalCmd,
1072
value_to_set=120000)
1073
1074
def test_approve_by_default(self):
1075
self.assert_command_from_args(["--approve-by-default", "foo"],
1076
ApproveByDefaultCmd)
1077
1078
def test_deny_by_default(self):
1079
self.assert_command_from_args(["--deny-by-default", "foo"],
1080
DenyByDefaultCmd)
1081
1082
def test_approval_delay(self):
1083
self.assert_command_from_args(["--approval-delay", "PT30S",
1084
"foo"], SetApprovalDelayCmd,
1085
value_to_set=30000)
1086
1087
def test_approval_duration(self):
1088
self.assert_command_from_args(["--approval-duration", "PT1S",
1089
"foo"], SetApprovalDurationCmd,
1090
value_to_set=1000)
1091
1092
def test_host(self):
1093
self.assert_command_from_args(["--host", "foo.example.org",
1094
"foo"], SetHostCmd,
1095
value_to_set="foo.example.org")
1096
1097
def test_host_short(self):
1098
self.assert_command_from_args(["-H", "foo.example.org",
1099
"foo"], SetHostCmd,
1100
value_to_set="foo.example.org")
1101
1102
def test_secret_devnull(self):
1103
self.assert_command_from_args(["--secret", os.path.devnull,
1104
"foo"], SetSecretCmd,
1105
value_to_set=b"")
1106
1107
def test_secret_tempfile(self):
1108
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1109
value = b"secret\0xyzzy\nbar"
1110
f.write(value)
1111
f.seek(0)
1112
self.assert_command_from_args(["--secret", f.name,
1113
"foo"], SetSecretCmd,
1114
value_to_set=value)
1115
1116
def test_secret_devnull_short(self):
1117
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1118
SetSecretCmd, value_to_set=b"")
1119
1120
def test_secret_tempfile_short(self):
1121
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1122
value = b"secret\0xyzzy\nbar"
1123
f.write(value)
1124
f.seek(0)
1125
self.assert_command_from_args(["-s", f.name, "foo"],
1126
SetSecretCmd,
1127
value_to_set=value)
1128
1129
def test_approve(self):
1130
self.assert_command_from_args(["--approve", "foo"],
1131
ApproveCmd)
1132
1133
def test_approve_short(self):
1134
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1135
1136
def test_deny(self):
1137
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1138
1139
def test_deny_short(self):
1140
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1141
1142
def test_dump_json(self):
1143
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1144
1145
def test_is_enabled(self):
1146
self.assert_command_from_args(["--is-enabled", "foo"],
1147
IsEnabledCmd)
1148
1149
def test_is_enabled_short(self):
1150
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1151
1152
def test_deny_before_remove(self):
1153
options = self.parser.parse_args(["--deny", "--remove",
1154
"foo"])
1155
check_option_syntax(self.parser, options)
1156
commands = commands_from_options(options)
1157
self.assertEqual(len(commands), 2)
1158
self.assertIsInstance(commands[0], DenyCmd)
1159
self.assertIsInstance(commands[1], RemoveCmd)
1160
1161
def test_deny_before_remove_reversed(self):
1162
options = self.parser.parse_args(["--remove", "--deny",
1163
"--all"])
1164
check_option_syntax(self.parser, options)
1165
commands = commands_from_options(options)
1166
self.assertEqual(len(commands), 2)
1167
self.assertIsInstance(commands[0], DenyCmd)
1168
self.assertIsInstance(commands[1], RemoveCmd)
1169
1170
1171
class TestCmd(unittest.TestCase):
1172
"""Abstract class for tests of command classes"""
1173
def setUp(self):
1174
testcase = self
1175
class MockClient(object):
1176
def __init__(self, name, **attributes):
1177
self.__dbus_object_path__ = "/clients/{}".format(name)
1178
self.attributes = attributes
1179
self.attributes["Name"] = name
1180
self.calls = []
1181
def Set(self, interface, propname, value, dbus_interface):
1182
testcase.assertEqual(interface, client_dbus_interface)
1183
testcase.assertEqual(dbus_interface,
1184
dbus.PROPERTIES_IFACE)
1185
self.attributes[propname] = value
1186
def Get(self, interface, propname, dbus_interface):
1187
testcase.assertEqual(interface, client_dbus_interface)
1188
testcase.assertEqual(dbus_interface,
1189
dbus.PROPERTIES_IFACE)
1190
return self.attributes[propname]
1191
def Approve(self, approve, dbus_interface):
1192
testcase.assertEqual(dbus_interface,
1193
client_dbus_interface)
1194
self.calls.append(("Approve", (approve,
1195
dbus_interface)))
1196
self.client = MockClient(
1197
"foo",
1198
KeyID=("92ed150794387c03ce684574b1139a65"
1199
"94a34f895daaaf09fd8ea90a27cddb12"),
1200
Secret=b"secret",
1201
Host="foo.example.org",
1202
Enabled=dbus.Boolean(True),
1203
Timeout=300000,
1204
LastCheckedOK="2019-02-03T00:00:00",
1205
Created="2019-01-02T00:00:00",
1206
Interval=120000,
1207
Fingerprint=("778827225BA7DE539C5A"
1208
"7CFA59CFF7CDBD9A5920"),
1209
CheckerRunning=dbus.Boolean(False),
1210
LastEnabled="2019-01-03T00:00:00",
1211
ApprovalPending=dbus.Boolean(False),
1212
ApprovedByDefault=dbus.Boolean(True),
1213
LastApprovalRequest="",
1214
ApprovalDelay=0,
1215
ApprovalDuration=1000,
1216
Checker="fping -q -- %(host)s",
1217
ExtendedTimeout=900000,
1218
Expires="2019-02-04T00:00:00",
1219
LastCheckerStatus=0)
1220
self.other_client = MockClient(
1221
"barbar",
1222
KeyID=("0558568eedd67d622f5c83b35a115f79"
1223
"6ab612cff5ad227247e46c2b020f441c"),
1224
Secret=b"secretbar",
1225
Host="192.0.2.3",
1226
Enabled=dbus.Boolean(True),
1227
Timeout=300000,
1228
LastCheckedOK="2019-02-04T00:00:00",
1229
Created="2019-01-03T00:00:00",
1230
Interval=120000,
1231
Fingerprint=("3E393AEAEFB84C7E89E2"
1232
"F547B3A107558FCA3A27"),
1233
CheckerRunning=dbus.Boolean(True),
1234
LastEnabled="2019-01-04T00:00:00",
1235
ApprovalPending=dbus.Boolean(False),
1236
ApprovedByDefault=dbus.Boolean(False),
1237
LastApprovalRequest="2019-01-03T00:00:00",
1238
ApprovalDelay=30000,
1239
ApprovalDuration=1000,
1240
Checker=":",
1241
ExtendedTimeout=900000,
1242
Expires="2019-02-05T00:00:00",
1243
LastCheckerStatus=-2)
1244
self.clients = collections.OrderedDict(
1245
[
1246
("/clients/foo", self.client.attributes),
1247
("/clients/barbar", self.other_client.attributes),
1248
])
1249
self.one_client = {"/clients/foo": self.client.attributes}
1250
@property
1251
def bus(self):
1252
class Bus(object):
1253
@staticmethod
1254
def get_object(client_bus_name, path):
1255
self.assertEqual(client_bus_name, dbus_busname)
1256
return {
1257
"/clients/foo": self.client,
1258
"/clients/barbar": self.other_client,
1259
}[path]
1260
return Bus()
1261
1262
1263
class TestIsEnabledCmd(TestCmd):
1264
def test_is_enabled(self):
1265
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1266
properties)
1267
for client, properties
1268
in self.clients.items()))
1269
def test_is_enabled_run_exits_successfully(self):
1270
with self.assertRaises(SystemExit) as e:
1271
IsEnabledCmd().run(self.one_client)
1272
if e.exception.code is not None:
1273
self.assertEqual(e.exception.code, 0)
1274
else:
1275
self.assertIsNone(e.exception.code)
1276
def test_is_enabled_run_exits_with_failure(self):
1277
self.client.attributes["Enabled"] = dbus.Boolean(False)
1278
with self.assertRaises(SystemExit) as e:
1279
IsEnabledCmd().run(self.one_client)
1280
if isinstance(e.exception.code, int):
1281
self.assertNotEqual(e.exception.code, 0)
1282
else:
1283
self.assertIsNotNone(e.exception.code)
1284
1285
1286
class TestApproveCmd(TestCmd):
1287
def test_approve(self):
1288
ApproveCmd().run(self.clients, self.bus)
1289
for clientpath in self.clients:
1290
client = self.bus.get_object(dbus_busname, clientpath)
1291
self.assertIn(("Approve", (True, client_dbus_interface)),
1292
client.calls)
1293
1294
1295
class TestDenyCmd(TestCmd):
1296
def test_deny(self):
1297
DenyCmd().run(self.clients, self.bus)
1298
for clientpath in self.clients:
1299
client = self.bus.get_object(dbus_busname, clientpath)
1300
self.assertIn(("Approve", (False, client_dbus_interface)),
1301
client.calls)
1302
1303
class TestRemoveCmd(TestCmd):
1304
def test_remove(self):
1305
class MockMandos(object):
1306
def __init__(self):
1307
self.calls = []
1308
def RemoveClient(self, dbus_path):
1309
self.calls.append(("RemoveClient", (dbus_path,)))
1310
mandos = MockMandos()
1311
super(TestRemoveCmd, self).setUp()
1312
RemoveCmd().run(self.clients, self.bus, mandos)
1313
self.assertEqual(len(mandos.calls), 2)
1314
for clientpath in self.clients:
1315
self.assertIn(("RemoveClient", (clientpath,)),
1316
mandos.calls)
1317
1318
1319
class TestDumpJSONCmd(TestCmd):
1320
def setUp(self):
1321
self.expected_json = {
1322
"foo": {
1323
"Name": "foo",
1324
"KeyID": ("92ed150794387c03ce684574b1139a65"
1325
"94a34f895daaaf09fd8ea90a27cddb12"),
1326
"Host": "foo.example.org",
1327
"Enabled": True,
1328
"Timeout": 300000,
1329
"LastCheckedOK": "2019-02-03T00:00:00",
1330
"Created": "2019-01-02T00:00:00",
1331
"Interval": 120000,
1332
"Fingerprint": ("778827225BA7DE539C5A"
1333
"7CFA59CFF7CDBD9A5920"),
1334
"CheckerRunning": False,
1335
"LastEnabled": "2019-01-03T00:00:00",
1336
"ApprovalPending": False,
1337
"ApprovedByDefault": True,
1338
"LastApprovalRequest": "",
1339
"ApprovalDelay": 0,
1340
"ApprovalDuration": 1000,
1341
"Checker": "fping -q -- %(host)s",
1342
"ExtendedTimeout": 900000,
1343
"Expires": "2019-02-04T00:00:00",
1344
"LastCheckerStatus": 0,
1345
},
1346
"barbar": {
1347
"Name": "barbar",
1348
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1349
"6ab612cff5ad227247e46c2b020f441c"),
1350
"Host": "192.0.2.3",
1351
"Enabled": True,
1352
"Timeout": 300000,
1353
"LastCheckedOK": "2019-02-04T00:00:00",
1354
"Created": "2019-01-03T00:00:00",
1355
"Interval": 120000,
1356
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1357
"F547B3A107558FCA3A27"),
1358
"CheckerRunning": True,
1359
"LastEnabled": "2019-01-04T00:00:00",
1360
"ApprovalPending": False,
1361
"ApprovedByDefault": False,
1362
"LastApprovalRequest": "2019-01-03T00:00:00",
1363
"ApprovalDelay": 30000,
1364
"ApprovalDuration": 1000,
1365
"Checker": ":",
1366
"ExtendedTimeout": 900000,
1367
"Expires": "2019-02-05T00:00:00",
1368
"LastCheckerStatus": -2,
1369
},
1370
}
1371
return super(TestDumpJSONCmd, self).setUp()
1372
def test_normal(self):
1373
json_data = json.loads(DumpJSONCmd().output(self.clients))
1374
self.assertDictEqual(json_data, self.expected_json)
1375
def test_one_client(self):
1376
clients = self.one_client
1377
json_data = json.loads(DumpJSONCmd().output(clients))
1378
expected_json = {"foo": self.expected_json["foo"]}
1379
self.assertDictEqual(json_data, expected_json)
1380
1381
1382
class TestPrintTableCmd(TestCmd):
1383
def test_normal(self):
1384
output = PrintTableCmd().output(self.clients.values())
1385
expected_output = "\n".join((
1386
"Name Enabled Timeout Last Successful Check",
1387
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1388
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1389
))
1390
self.assertEqual(output, expected_output)
1391
def test_verbose(self):
1392
output = PrintTableCmd(verbose=True).output(
1393
self.clients.values())
1394
columns = (
1395
(
1396
"Name ",
1397
"foo ",
1398
"barbar ",
1399
),(
1400
"Enabled ",
1401
"Yes ",
1402
"Yes ",
1403
),(
1404
"Timeout ",
1405
"00:05:00 ",
1406
"00:05:00 ",
1407
),(
1408
"Last Successful Check ",
1409
"2019-02-03T00:00:00 ",
1410
"2019-02-04T00:00:00 ",
1411
),(
1412
"Created ",
1413
"2019-01-02T00:00:00 ",
1414
"2019-01-03T00:00:00 ",
1415
),(
1416
"Interval ",
1417
"00:02:00 ",
1418
"00:02:00 ",
1419
),(
1420
"Host ",
1421
"foo.example.org ",
1422
"192.0.2.3 ",
1423
),(
1424
("Key ID "
1425
" "),
1426
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1427
"ea90a27cddb12 "),
1428
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1429
"46c2b020f441c "),
1430
),(
1431
"Fingerprint ",
1432
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1433
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1434
),(
1435
"Check Is Running ",
1436
"No ",
1437
"Yes ",
1438
),(
1439
"Last Enabled ",
1440
"2019-01-03T00:00:00 ",
1441
"2019-01-04T00:00:00 ",
1442
),(
1443
"Approval Is Pending ",
1444
"No ",
1445
"No ",
1446
),(
1447
"Approved By Default ",
1448
"Yes ",
1449
"No ",
1450
),(
1451
"Last Approval Request ",
1452
" ",
1453
"2019-01-03T00:00:00 ",
1454
),(
1455
"Approval Delay ",
1456
"00:00:00 ",
1457
"00:00:30 ",
1458
),(
1459
"Approval Duration ",
1460
"00:00:01 ",
1461
"00:00:01 ",
1462
),(
1463
"Checker ",
1464
"fping -q -- %(host)s ",
1465
": ",
1466
),(
1467
"Extended Timeout ",
1468
"00:15:00 ",
1469
"00:15:00 ",
1470
),(
1471
"Expires ",
1472
"2019-02-04T00:00:00 ",
1473
"2019-02-05T00:00:00 ",
1474
),(
1475
"Last Checker Status",
1476
"0 ",
1477
"-2 ",
1478
)
1479
)
1480
num_lines = max(len(rows) for rows in columns)
1481
expected_output = "\n".join("".join(rows[line]
1482
for rows in columns)
1483
for line in range(num_lines))
1484
self.assertEqual(output, expected_output)
1485
def test_one_client(self):
1486
output = PrintTableCmd().output(self.one_client.values())
1487
expected_output = """
1488
Name Enabled Timeout Last Successful Check
1489
foo Yes 00:05:00 2019-02-03T00:00:00
1490
"""[1:-1]
1491
self.assertEqual(output, expected_output)
1492
1493
576
1494
class Test_milliseconds_to_string(unittest.TestCase):
577
1495
def test_all(self):
578
1496
self.assertEqual(milliseconds_to_string(93785000),
586
1504
self.assertEqual(milliseconds_to_string(900), "00:00:00")
587
1505
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
588
1506
589
class Test_string_to_delta(unittest.TestCase):
590
def test_handles_basic_rfc3339(self):
591
self.assertEqual(string_to_delta("PT2H"),
592
datetime.timedelta(0, 7200))
593
def test_falls_back_to_pre_1_6_1_with_warning(self):
594
# assertLogs only exists in Python 3.4
595
if hasattr(self, "assertLogs"):
596
with self.assertLogs(log, logging.WARNING):
597
value = string_to_delta("2h")
598
else:
599
value = string_to_delta("2h")
600
self.assertEqual(value, datetime.timedelta(0, 7200))
601
602
class Test_TableOfClients(unittest.TestCase):
603
def setUp(self):
604
self.tablewords = {
605
"Attr1": "X",
606
"AttrTwo": "Yy",
607
"AttrThree": "Zzz",
608
"Bool": "A D-BUS Boolean",
609
"NonDbusBoolean": "A Non-D-BUS Boolean",
610
"Integer": "An Integer",
611
"Timeout": "Timedelta 1",
612
"Interval": "Timedelta 2",
613
"ApprovalDelay": "Timedelta 3",
614
"ApprovalDuration": "Timedelta 4",
615
"ExtendedTimeout": "Timedelta 5",
616
"String": "A String",
617
}
618
self.keywords = ["Attr1", "AttrTwo"]
619
self.clients = [
620
{
621
"Attr1": "x1",
622
"AttrTwo": "y1",
623
"AttrThree": "z1",
624
"Bool": dbus.Boolean(False),
625
"NonDbusBoolean": False,
626
"Integer": 0,
627
"Timeout": 0,
628
"Interval": 1000,
629
"ApprovalDelay": 2000,
630
"ApprovalDuration": 3000,
631
"ExtendedTimeout": 4000,
632
"String": "",
633
},
634
{
635
"Attr1": "x2",
636
"AttrTwo": "y2",
637
"AttrThree": "z2",
638
"Bool": dbus.Boolean(True),
639
"NonDbusBoolean": True,
640
"Integer": 1,
641
"Timeout": 93785000,
642
"Interval": 93786000,
643
"ApprovalDelay": 93787000,
644
"ApprovalDuration": 93788000,
645
"ExtendedTimeout": 93789000,
646
"String": "A huge string which will not fit," * 10,
647
},
648
]
649
def test_short_header(self):
650
text = str(TableOfClients(self.clients, self.keywords,
651
self.tablewords))
652
expected_text = """
653
X Yy
654
x1 y1
655
x2 y2
656
"""[1:-1]
657
self.assertEqual(text, expected_text)
658
def test_booleans(self):
659
keywords = ["Bool", "NonDbusBoolean"]
660
text = str(TableOfClients(self.clients, keywords,
661
self.tablewords))
662
expected_text = """
663
A D-BUS Boolean A Non-D-BUS Boolean
664
No False
665
Yes True
666
"""[1:-1]
667
self.assertEqual(text, expected_text)
668
def test_milliseconds_detection(self):
669
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
670
"ApprovalDuration", "ExtendedTimeout"]
671
text = str(TableOfClients(self.clients, keywords,
672
self.tablewords))
673
expected_text = """
674
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
675
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
676
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
677
"""[1:-1]
678
self.assertEqual(text, expected_text)
679
def test_empty_and_long_string_values(self):
680
keywords = ["String"]
681
text = str(TableOfClients(self.clients, keywords,
682
self.tablewords))
683
expected_text = """
684
A String
685
686
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
687
"""[1:-1]
688
self.assertEqual(text, expected_text)
1507
1508
class Unique(object):
1509
"""Class for objects which exist only to be unique objects, since
1510
unittest.mock.sentinel only exists in Python 3.3"""
1511
1512
1513
class TestPropertyCmd(TestCmd):
1514
"""Abstract class for tests of PropertyCmd classes"""
1515
def runTest(self):
1516
if not hasattr(self, "command"):
1517
return
1518
values_to_get = getattr(self, "values_to_get",
1519
self.values_to_set)
1520
for value_to_set, value_to_get in zip(self.values_to_set,
1521
values_to_get):
1522
for clientpath in self.clients:
1523
client = self.bus.get_object(dbus_busname, clientpath)
1524
old_value = client.attributes[self.propname]
1525
self.assertNotIsInstance(old_value, Unique)
1526
client.attributes[self.propname] = Unique()
1527
self.run_command(value_to_set, self.clients)
1528
for clientpath in self.clients:
1529
client = self.bus.get_object(dbus_busname, clientpath)
1530
value = client.attributes[self.propname]
1531
self.assertNotIsInstance(value, Unique)
1532
self.assertEqual(value, value_to_get)
1533
def run_command(self, value, clients):
1534
self.command().run(clients, self.bus)
1535
1536
1537
class TestEnableCmd(TestCmd):
1538
def test_enable(self):
1539
for clientpath in self.clients:
1540
client = self.bus.get_object(dbus_busname, clientpath)
1541
client.attributes["Enabled"] = False
1542
1543
EnableCmd().run(self.clients, self.bus)
1544
1545
for clientpath in self.clients:
1546
client = self.bus.get_object(dbus_busname, clientpath)
1547
self.assertTrue(client.attributes["Enabled"])
1548
1549
1550
class TestDisableCmd(TestCmd):
1551
def test_disable(self):
1552
DisableCmd().run(self.clients, self.bus)
1553
for clientpath in self.clients:
1554
client = self.bus.get_object(dbus_busname, clientpath)
1555
self.assertFalse(client.attributes["Enabled"])
1556
1557
1558
class TestBumpTimeoutCmd(TestPropertyCmd):
1559
command = BumpTimeoutCmd
1560
propname = "LastCheckedOK"
1561
values_to_set = [""]
1562
1563
1564
class TestStartCheckerCmd(TestPropertyCmd):
1565
command = StartCheckerCmd
1566
propname = "CheckerRunning"
1567
values_to_set = [dbus.Boolean(True)]
1568
1569
1570
class TestStopCheckerCmd(TestPropertyCmd):
1571
command = StopCheckerCmd
1572
propname = "CheckerRunning"
1573
values_to_set = [dbus.Boolean(False)]
1574
1575
1576
class TestApproveByDefaultCmd(TestPropertyCmd):
1577
command = ApproveByDefaultCmd
1578
propname = "ApprovedByDefault"
1579
values_to_set = [dbus.Boolean(True)]
1580
1581
1582
class TestDenyByDefaultCmd(TestPropertyCmd):
1583
command = DenyByDefaultCmd
1584
propname = "ApprovedByDefault"
1585
values_to_set = [dbus.Boolean(False)]
1586
1587
1588
class TestPropertyValueCmd(TestPropertyCmd):
1589
"""Abstract class for tests of PropertyValueCmd classes"""
1590
def runTest(self):
1591
if type(self) is TestPropertyValueCmd:
1592
return
1593
return super(TestPropertyValueCmd, self).runTest()
1594
def run_command(self, value, clients):
1595
self.command(value).run(clients, self.bus)
1596
1597
1598
class TestSetCheckerCmd(TestPropertyValueCmd):
1599
command = SetCheckerCmd
1600
propname = "Checker"
1601
values_to_set = ["", ":", "fping -q -- %s"]
1602
1603
1604
class TestSetHostCmd(TestPropertyValueCmd):
1605
command = SetHostCmd
1606
propname = "Host"
1607
values_to_set = ["192.0.2.3", "foo.example.org"]
1608
1609
1610
class TestSetSecretCmd(TestPropertyValueCmd):
1611
command = SetSecretCmd
1612
propname = "Secret"
1613
values_to_set = [io.BytesIO(b""),
1614
io.BytesIO(b"secret\0xyzzy\nbar")]
1615
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1616
1617
1618
class TestSetTimeoutCmd(TestPropertyValueCmd):
1619
command = SetTimeoutCmd
1620
propname = "Timeout"
1621
values_to_set = [datetime.timedelta(),
1622
datetime.timedelta(minutes=5),
1623
datetime.timedelta(seconds=1),
1624
datetime.timedelta(weeks=1),
1625
datetime.timedelta(weeks=52)]
1626
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1627
1628
1629
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1630
command = SetExtendedTimeoutCmd
1631
propname = "ExtendedTimeout"
1632
values_to_set = [datetime.timedelta(),
1633
datetime.timedelta(minutes=5),
1634
datetime.timedelta(seconds=1),
1635
datetime.timedelta(weeks=1),
1636
datetime.timedelta(weeks=52)]
1637
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1638
1639
1640
class TestSetIntervalCmd(TestPropertyValueCmd):
1641
command = SetIntervalCmd
1642
propname = "Interval"
1643
values_to_set = [datetime.timedelta(),
1644
datetime.timedelta(minutes=5),
1645
datetime.timedelta(seconds=1),
1646
datetime.timedelta(weeks=1),
1647
datetime.timedelta(weeks=52)]
1648
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1649
1650
1651
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1652
command = SetApprovalDelayCmd
1653
propname = "ApprovalDelay"
1654
values_to_set = [datetime.timedelta(),
1655
datetime.timedelta(minutes=5),
1656
datetime.timedelta(seconds=1),
1657
datetime.timedelta(weeks=1),
1658
datetime.timedelta(weeks=52)]
1659
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1660
1661
1662
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1663
command = SetApprovalDurationCmd
1664
propname = "ApprovalDuration"
1665
values_to_set = [datetime.timedelta(),
1666
datetime.timedelta(minutes=5),
1667
datetime.timedelta(seconds=1),
1668
datetime.timedelta(weeks=1),
1669
datetime.timedelta(weeks=52)]
1670
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
689
1671
690
1672
691
1673
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0