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