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 1092
- compare with another revision
- download diff
- download tarball
- view history from revision 1092
- Committer: Teddy Hogeborn
- Date: 2019-03-30 17:02:33 UTC
- Revision ID: teddy@recompile.se-20190330170233-nxu3cgb98q2g3o5j
Fix typo in intro(8mandos).
* intro.xml (INTRODUCTION): Add missing period at end of last sentence
of penultimate paragraph.
* intro.xml (INTRODUCTION): Add missing period at end of last sentence
of penultimate paragraph.
- files removed:
- initramfs-tools-conf-hook
- files modified:
- Makefile
added
removed
mandos-ctl
45
45
import io
46
46
import tempfile
47
47
import contextlib
48
import abc
48
49
49
try:
50
import pydbus
51
import gi
52
dbus_python = None
53
except ImportError:
54
import dbus as dbus_python
55
pydbus = None
56
class gi(object):
57
"""Dummy gi module, for the tests"""
58
class repository(object):
59
class GLib(object):
60
class Error(Exception):
61
pass
50
import dbus as dbus_python
62
51
63
52
# Show warnings by default
64
53
if not sys.warnoptions:
93
82
if options.debug:
94
83
log.setLevel(logging.DEBUG)
95
84
96
if pydbus is not None:
97
bus = pydbus_adapter.CachingBus(pydbus)
98
else:
99
bus = dbus_python_adapter.CachingBus(dbus_python)
85
bus = dbus_python_adapter.CachingBus(dbus_python)
100
86
101
87
try:
102
88
all_clients = bus.get_clients_and_properties()
136
122
help="Select all clients")
137
123
parser.add_argument("-v", "--verbose", action="store_true",
138
124
help="Print all fields")
139
parser.add_argument("-j", "--dump-json", dest="commands",
140
action="append_const", default=[],
141
const=command.DumpJSON(),
125
parser.add_argument("-j", "--dump-json", action="store_true",
142
126
help="Dump client data in JSON format")
143
127
enable_disable = parser.add_mutually_exclusive_group()
144
enable_disable.add_argument("-e", "--enable", dest="commands",
145
action="append_const", default=[],
146
const=command.Enable(),
128
enable_disable.add_argument("-e", "--enable", action="store_true",
147
129
help="Enable client")
148
enable_disable.add_argument("-d", "--disable", dest="commands",
149
action="append_const", default=[],
150
const=command.Disable(),
130
enable_disable.add_argument("-d", "--disable",
131
action="store_true",
151
132
help="disable client")
152
parser.add_argument("-b", "--bump-timeout", dest="commands",
153
action="append_const", default=[],
154
const=command.BumpTimeout(),
133
parser.add_argument("-b", "--bump-timeout", action="store_true",
155
134
help="Bump timeout for client")
156
135
start_stop_checker = parser.add_mutually_exclusive_group()
157
136
start_stop_checker.add_argument("--start-checker",
158
dest="commands",
159
action="append_const", default=[],
160
const=command.StartChecker(),
137
action="store_true",
161
138
help="Start checker for client")
162
start_stop_checker.add_argument("--stop-checker", dest="commands",
163
action="append_const", default=[],
164
const=command.StopChecker(),
139
start_stop_checker.add_argument("--stop-checker",
140
action="store_true",
165
141
help="Stop checker for client")
166
parser.add_argument("-V", "--is-enabled", dest="commands",
167
action="append_const", default=[],
168
const=command.IsEnabled(),
142
parser.add_argument("-V", "--is-enabled", action="store_true",
169
143
help="Check if client is enabled")
170
parser.add_argument("-r", "--remove", dest="commands",
171
action="append_const", default=[],
172
const=command.Remove(),
144
parser.add_argument("-r", "--remove", action="store_true",
173
145
help="Remove client")
174
parser.add_argument("-c", "--checker", dest="commands",
175
action="append", default=[],
176
metavar="COMMAND", type=command.SetChecker,
146
parser.add_argument("-c", "--checker",
177
147
help="Set checker command for client")
178
parser.add_argument(
179
"-t", "--timeout", dest="commands", action="append",
180
default=[], metavar="TIME",
181
type=command.SetTimeout.argparse(string_to_delta),
182
help="Set timeout for client")
183
parser.add_argument(
184
"--extended-timeout", dest="commands", action="append",
185
default=[], metavar="TIME",
186
type=command.SetExtendedTimeout.argparse(string_to_delta),
187
help="Set extended timeout for client")
188
parser.add_argument(
189
"-i", "--interval", dest="commands", action="append",
190
default=[], metavar="TIME",
191
type=command.SetInterval.argparse(string_to_delta),
192
help="Set checker interval for client")
148
parser.add_argument("-t", "--timeout", type=string_to_delta,
149
help="Set timeout for client")
150
parser.add_argument("--extended-timeout", type=string_to_delta,
151
help="Set extended timeout for client")
152
parser.add_argument("-i", "--interval", type=string_to_delta,
153
help="Set checker interval for client")
193
154
approve_deny_default = parser.add_mutually_exclusive_group()
194
155
approve_deny_default.add_argument(
195
"--approve-by-default", dest="commands",
196
action="append_const", default=[],
197
const=command.ApproveByDefault(),
156
"--approve-by-default", action="store_true",
157
default=None, dest="approved_by_default",
198
158
help="Set client to be approved by default")
199
159
approve_deny_default.add_argument(
200
"--deny-by-default", dest="commands",
201
action="append_const", default=[],
202
const=command.DenyByDefault(),
160
"--deny-by-default", action="store_false",
161
dest="approved_by_default",
203
162
help="Set client to be denied by default")
204
parser.add_argument(
205
"--approval-delay", dest="commands", action="append",
206
default=[], metavar="TIME",
207
type=command.SetApprovalDelay.argparse(string_to_delta),
208
help="Set delay before client approve/deny")
209
parser.add_argument(
210
"--approval-duration", dest="commands", action="append",
211
default=[], metavar="TIME",
212
type=command.SetApprovalDuration.argparse(string_to_delta),
213
help="Set duration of one client approval")
214
parser.add_argument("-H", "--host", dest="commands",
215
action="append", default=[], metavar="STRING",
216
type=command.SetHost,
217
help="Set host for client")
218
parser.add_argument(
219
"-s", "--secret", dest="commands", action="append",
220
default=[], metavar="FILENAME",
221
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
222
help="Set password blob (file) for client")
163
parser.add_argument("--approval-delay", type=string_to_delta,
164
help="Set delay before client approve/deny")
165
parser.add_argument("--approval-duration", type=string_to_delta,
166
help="Set duration of one client approval")
167
parser.add_argument("-H", "--host", help="Set host for client")
168
parser.add_argument("-s", "--secret",
169
type=argparse.FileType(mode="rb"),
170
help="Set password blob (file) for client")
223
171
approve_deny = parser.add_mutually_exclusive_group()
224
172
approve_deny.add_argument(
225
"-A", "--approve", dest="commands", action="append_const",
226
default=[], const=command.Approve(),
173
"-A", "--approve", action="store_true",
227
174
help="Approve any current client request")
228
approve_deny.add_argument("-D", "--deny", dest="commands",
229
action="append_const", default=[],
230
const=command.Deny(),
175
approve_deny.add_argument("-D", "--deny", action="store_true",
231
176
help="Deny any current client request")
232
177
parser.add_argument("--debug", action="store_true",
233
178
help="Debug mode (show D-Bus commands)")
424
369
"""Apply additional restrictions on options, not expressible in
425
370
argparse"""
426
371
427
def has_commands(options, commands=None):
428
if commands is None:
429
commands = (command.Enable,
430
command.Disable,
431
command.BumpTimeout,
432
command.StartChecker,
433
command.StopChecker,
434
command.IsEnabled,
435
command.Remove,
436
command.SetChecker,
437
command.SetTimeout,
438
command.SetExtendedTimeout,
439
command.SetInterval,
440
command.ApproveByDefault,
441
command.DenyByDefault,
442
command.SetApprovalDelay,
443
command.SetApprovalDuration,
444
command.SetHost,
445
command.SetSecret,
446
command.Approve,
447
command.Deny)
448
return any(isinstance(cmd, commands)
449
for cmd in options.commands)
372
def has_actions(options):
373
return any((options.enable,
374
options.disable,
375
options.bump_timeout,
376
options.start_checker,
377
options.stop_checker,
378
options.is_enabled,
379
options.remove,
380
options.checker is not None,
381
options.timeout is not None,
382
options.extended_timeout is not None,
383
options.interval is not None,
384
options.approved_by_default is not None,
385
options.approval_delay is not None,
386
options.approval_duration is not None,
387
options.host is not None,
388
options.secret is not None,
389
options.approve,
390
options.deny))
450
391
451
if has_commands(options) and not (options.client or options.all):
392
if has_actions(options) and not (options.client or options.all):
452
393
parser.error("Options require clients names or --all.")
453
if options.verbose and has_commands(options):
394
if options.verbose and has_actions(options):
454
395
parser.error("--verbose can only be used alone.")
455
if (has_commands(options, (command.DumpJSON,))
456
and (options.verbose or len(options.commands) > 1)):
396
if options.dump_json and (options.verbose
397
or has_actions(options)):
457
398
parser.error("--dump-json can only be used alone.")
458
if options.all and not has_commands(options):
399
if options.all and not has_actions(options):
459
400
parser.error("--all requires an action.")
460
if (has_commands(options, (command.IsEnabled,))
461
and len(options.client) > 1):
401
if options.is_enabled and len(options.client) > 1:
462
402
parser.error("--is-enabled requires exactly one client")
463
if (len(options.commands) > 1
464
and has_commands(options, (command.Remove,))
465
and not has_commands(options, (command.Deny,))):
466
parser.error("--remove can only be combined with --deny")
403
if options.remove:
404
options.remove = False
405
if has_actions(options) and not options.deny:
406
parser.error("--remove can only be combined with --deny")
407
options.remove = True
408
467
409
468
410
469
411
class dbus(object):
571
513
for key, subval in value.items()}
572
514
return value
573
515
574
def set_client_property(self, objectpath, key, value):
575
if key == "Secret":
576
if not isinstance(value, bytes):
577
value = value.encode("utf-8")
578
value = self.dbus_python.ByteArray(value)
579
return self.set_property(self.busname, objectpath,
580
self.client_interface, key,
581
value)
582
516
583
517
class SilenceLogger(object):
584
518
"Simple context manager to silence a particular logger"
615
549
return new_object
616
550
617
551
618
class pydbus_adapter(object):
619
class SystemBus(dbus.MandosBus):
620
def __init__(self, module=pydbus):
621
self.pydbus = module
622
self.bus = self.pydbus.SystemBus()
623
624
@contextlib.contextmanager
625
def convert_exception(self, exception_class=dbus.Error):
626
try:
627
yield
628
except gi.repository.GLib.Error as e:
629
# This does what "raise from" would do
630
exc = exception_class(*e.args)
631
exc.__cause__ = e
632
raise exc
633
634
def call_method(self, methodname, busname, objectpath,
635
interface, *args):
636
proxy_object = self.get(busname, objectpath)
637
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
638
interface, methodname,
639
", ".join(repr(a) for a in args))
640
method = getattr(proxy_object[interface], methodname)
641
with self.convert_exception():
642
return method(*args)
643
644
def get(self, busname, objectpath):
645
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
646
busname, objectpath)
647
with self.convert_exception(dbus.ConnectFailed):
648
if sys.version_info.major <= 2:
649
with warnings.catch_warnings():
650
warnings.filterwarnings(
651
"ignore", "", DeprecationWarning,
652
r"^xml\.etree\.ElementTree$")
653
return self.bus.get(busname, objectpath)
654
else:
655
return self.bus.get(busname, objectpath)
656
657
def set_property(self, busname, objectpath, interface, key,
658
value):
659
proxy_object = self.get(busname, objectpath)
660
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
661
objectpath, self.properties_iface, interface,
662
key, value)
663
setattr(proxy_object[interface], key, value)
664
665
class CachingBus(SystemBus):
666
"""A caching layer for pydbus_adapter.SystemBus"""
667
def __init__(self, *args, **kwargs):
668
self.object_cache = {}
669
super(pydbus_adapter.CachingBus,
670
self).__init__(*args, **kwargs)
671
def get(self, busname, objectpath):
672
try:
673
return self.object_cache[(busname, objectpath)]
674
except KeyError:
675
new_object = (super(pydbus_adapter.CachingBus, self)
676
.get(busname, objectpath))
677
self.object_cache[(busname, objectpath)] = new_object
678
return new_object
679
680
681
552
def commands_from_options(options):
682
553
683
commands = list(options.commands)
684
685
def find_cmd(cmd, commands):
686
i = 0
687
for i, c in enumerate(commands):
688
if isinstance(c, cmd):
689
return i
690
return i+1
691
692
# If command.Remove is present, move any instances of command.Deny
693
# to occur ahead of command.Remove.
694
index_of_remove = find_cmd(command.Remove, commands)
695
before_remove = commands[:index_of_remove]
696
after_remove = commands[index_of_remove:]
697
cleaned_after = []
698
for cmd in after_remove:
699
if isinstance(cmd, command.Deny):
700
before_remove.append(cmd)
554
commands = []
555
556
if options.is_enabled:
557
commands.append(command.IsEnabled())
558
559
if options.approve:
560
commands.append(command.Approve())
561
562
if options.deny:
563
commands.append(command.Deny())
564
565
if options.remove:
566
commands.append(command.Remove())
567
568
if options.dump_json:
569
commands.append(command.DumpJSON())
570
571
if options.enable:
572
commands.append(command.Enable())
573
574
if options.disable:
575
commands.append(command.Disable())
576
577
if options.bump_timeout:
578
commands.append(command.BumpTimeout())
579
580
if options.start_checker:
581
commands.append(command.StartChecker())
582
583
if options.stop_checker:
584
commands.append(command.StopChecker())
585
586
if options.approved_by_default is not None:
587
if options.approved_by_default:
588
commands.append(command.ApproveByDefault())
701
589
else:
702
cleaned_after.append(cmd)
703
if cleaned_after != after_remove:
704
commands = before_remove + cleaned_after
590
commands.append(command.DenyByDefault())
591
592
if options.checker is not None:
593
commands.append(command.SetChecker(options.checker))
594
595
if options.host is not None:
596
commands.append(command.SetHost(options.host))
597
598
if options.secret is not None:
599
commands.append(command.SetSecret(options.secret))
600
601
if options.timeout is not None:
602
commands.append(command.SetTimeout(options.timeout))
603
604
if options.extended_timeout:
605
commands.append(
606
command.SetExtendedTimeout(options.extended_timeout))
607
608
if options.interval is not None:
609
commands.append(command.SetInterval(options.interval))
610
611
if options.approval_delay is not None:
612
commands.append(
613
command.SetApprovalDelay(options.approval_delay))
614
615
if options.approval_duration is not None:
616
commands.append(
617
command.SetApprovalDuration(options.approval_duration))
705
618
706
619
# If no command option has been given, show table of clients,
707
620
# optionally verbosely
922
835
def __init__(self, value):
923
836
self.value_to_set = value
924
837
925
@classmethod
926
def argparse(cls, argtype):
927
def cmdtype(arg):
928
return cls(argtype(arg))
929
return cmdtype
930
838
931
839
class SetChecker(PropertySetterValue):
932
840
propname = "Checker"
1058
966
1059
967
def test_actions_requires_client_or_all(self):
1060
968
for action, value in self.actions.items():
1061
args = self.actionargs(action, value)
969
options = self.parser.parse_args()
970
setattr(options, action, value)
1062
971
with self.assertParseError():
1063
self.parse_args(args)
972
self.check_option_syntax(options)
1064
973
1065
# This mostly corresponds to the definition from has_commands() in
974
# This mostly corresponds to the definition from has_actions() in
1066
975
# check_option_syntax()
1067
976
actions = {
1068
"--enable": None,
1069
"--disable": None,
1070
"--bump-timeout": None,
1071
"--start-checker": None,
1072
"--stop-checker": None,
1073
"--is-enabled": None,
1074
"--remove": None,
1075
"--checker": "x",
1076
"--timeout": "PT0S",
1077
"--extended-timeout": "PT0S",
1078
"--interval": "PT0S",
1079
"--approve-by-default": None,
1080
"--deny-by-default": None,
1081
"--approval-delay": "PT0S",
1082
"--approval-duration": "PT0S",
1083
"--host": "hostname",
1084
"--secret": "/dev/null",
1085
"--approve": None,
1086
"--deny": None,
977
# The actual values set here are not that important, but we do
978
# at least stick to the correct types, even though they are
979
# never used
980
"enable": True,
981
"disable": True,
982
"bump_timeout": True,
983
"start_checker": True,
984
"stop_checker": True,
985
"is_enabled": True,
986
"remove": True,
987
"checker": "x",
988
"timeout": datetime.timedelta(),
989
"extended_timeout": datetime.timedelta(),
990
"interval": datetime.timedelta(),
991
"approved_by_default": True,
992
"approval_delay": datetime.timedelta(),
993
"approval_duration": datetime.timedelta(),
994
"host": "x",
995
"secret": io.BytesIO(b"x"),
996
"approve": True,
997
"deny": True,
1087
998
}
1088
999
1089
@staticmethod
1090
def actionargs(action, value, *args):
1091
if value is not None:
1092
return [action, value] + list(args)
1093
else:
1094
return [action] + list(args)
1095
1096
1000
@contextlib.contextmanager
1097
1001
def assertParseError(self):
1098
1002
with self.assertRaises(SystemExit) as e:
1103
1007
# /argparse.html#exiting-methods
1104
1008
self.assertEqual(2, e.exception.code)
1105
1009
1106
def parse_args(self, args):
1107
options = self.parser.parse_args(args)
1108
check_option_syntax(self.parser, options)
1109
1110
1010
@staticmethod
1111
1011
@contextlib.contextmanager
1112
1012
def redirect_stderr_to_devnull():
1123
1023
1124
1024
def test_actions_all_conflicts_with_verbose(self):
1125
1025
for action, value in self.actions.items():
1126
args = self.actionargs(action, value, "--all",
1127
"--verbose")
1026
options = self.parser.parse_args()
1027
setattr(options, action, value)
1028
options.all = True
1029
options.verbose = True
1128
1030
with self.assertParseError():
1129
self.parse_args(args)
1031
self.check_option_syntax(options)
1130
1032
1131
1033
def test_actions_with_client_conflicts_with_verbose(self):
1132
1034
for action, value in self.actions.items():
1133
args = self.actionargs(action, value, "--verbose",
1134
"client")
1035
options = self.parser.parse_args()
1036
setattr(options, action, value)
1037
options.verbose = True
1038
options.client = ["client"]
1135
1039
with self.assertParseError():
1136
self.parse_args(args)
1040
self.check_option_syntax(options)
1137
1041
1138
1042
def test_dump_json_conflicts_with_verbose(self):
1139
args = ["--dump-json", "--verbose"]
1043
options = self.parser.parse_args()
1044
options.dump_json = True
1045
options.verbose = True
1140
1046
with self.assertParseError():
1141
self.parse_args(args)
1047
self.check_option_syntax(options)
1142
1048
1143
1049
def test_dump_json_conflicts_with_action(self):
1144
1050
for action, value in self.actions.items():
1145
args = self.actionargs(action, value, "--dump-json")
1051
options = self.parser.parse_args()
1052
setattr(options, action, value)
1053
options.dump_json = True
1146
1054
with self.assertParseError():
1147
self.parse_args(args)
1055
self.check_option_syntax(options)
1148
1056
1149
1057
def test_all_can_not_be_alone(self):
1150
args = ["--all"]
1058
options = self.parser.parse_args()
1059
options.all = True
1151
1060
with self.assertParseError():
1152
self.parse_args(args)
1061
self.check_option_syntax(options)
1153
1062
1154
1063
def test_all_is_ok_with_any_action(self):
1155
1064
for action, value in self.actions.items():
1156
args = self.actionargs(action, value, "--all")
1157
self.parse_args(args)
1065
options = self.parser.parse_args()
1066
setattr(options, action, value)
1067
options.all = True
1068
self.check_option_syntax(options)
1158
1069
1159
1070
def test_any_action_is_ok_with_one_client(self):
1160
1071
for action, value in self.actions.items():
1161
args = self.actionargs(action, value, "client")
1162
self.parse_args(args)
1072
options = self.parser.parse_args()
1073
setattr(options, action, value)
1074
options.client = ["client"]
1075
self.check_option_syntax(options)
1163
1076
1164
1077
def test_one_client_with_all_actions_except_is_enabled(self):
1078
options = self.parser.parse_args()
1165
1079
for action, value in self.actions.items():
1166
if action == "--is-enabled":
1080
if action == "is_enabled":
1167
1081
continue
1168
args = self.actionargs(action, value, "client")
1169
self.parse_args(args)
1082
setattr(options, action, value)
1083
options.client = ["client"]
1084
self.check_option_syntax(options)
1170
1085
1171
1086
def test_two_clients_with_all_actions_except_is_enabled(self):
1087
options = self.parser.parse_args()
1172
1088
for action, value in self.actions.items():
1173
if action == "--is-enabled":
1089
if action == "is_enabled":
1174
1090
continue
1175
args = self.actionargs(action, value, "client1",
1176
"client2")
1177
self.parse_args(args)
1091
setattr(options, action, value)
1092
options.client = ["client1", "client2"]
1093
self.check_option_syntax(options)
1178
1094
1179
1095
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1180
1096
for action, value in self.actions.items():
1181
if action == "--is-enabled":
1097
if action == "is_enabled":
1182
1098
continue
1183
args = self.actionargs(action, value, "client1",
1184
"client2")
1185
self.parse_args(args)
1099
options = self.parser.parse_args()
1100
setattr(options, action, value)
1101
options.client = ["client1", "client2"]
1102
self.check_option_syntax(options)
1186
1103
1187
1104
def test_is_enabled_fails_without_client(self):
1188
args = ["--is-enabled"]
1105
options = self.parser.parse_args()
1106
options.is_enabled = True
1189
1107
with self.assertParseError():
1190
self.parse_args(args)
1108
self.check_option_syntax(options)
1191
1109
1192
1110
def test_is_enabled_fails_with_two_clients(self):
1193
args = ["--is-enabled", "client1", "client2"]
1111
options = self.parser.parse_args()
1112
options.is_enabled = True
1113
options.client = ["client1", "client2"]
1194
1114
with self.assertParseError():
1195
self.parse_args(args)
1115
self.check_option_syntax(options)
1196
1116
1197
1117
def test_remove_can_only_be_combined_with_action_deny(self):
1198
1118
for action, value in self.actions.items():
1199
if action in {"--remove", "--deny"}:
1119
if action in {"remove", "deny"}:
1200
1120
continue
1201
args = self.actionargs(action, value, "--all",
1202
"--remove")
1121
options = self.parser.parse_args()
1122
setattr(options, action, value)
1123
options.all = True
1124
options.remove = True
1203
1125
with self.assertParseError():
1204
self.parse_args(args)
1126
self.check_option_syntax(options)
1205
1127
1206
1128
1207
1129
class Test_dbus_exceptions(unittest.TestCase):
1320
1242
@staticmethod
1321
1243
def get_object(busname, objectpath):
1322
1244
DBusObject = collections.namedtuple(
1323
"DBusObject", ("methodname", "Set"))
1245
"DBusObject", ("methodname",))
1324
1246
def method(*args, **kwargs):
1325
1247
self.assertEqual({"dbus_interface":
1326
1248
"interface"},
1327
1249
kwargs)
1328
1250
return func(*args)
1329
def set_property(interface, key, value,
1330
dbus_interface=None):
1331
self.assertEqual(
1332
"org.freedesktop.DBus.Properties",
1333
dbus_interface)
1334
self.assertEqual("Secret", key)
1335
return func(interface, key, value,
1336
dbus_interface=dbus_interface)
1337
return DBusObject(methodname=method,
1338
Set=set_property)
1251
return DBusObject(methodname=method)
1339
1252
class Boolean(object):
1340
1253
def __init__(self, value):
1341
1254
self.value = bool(value)
1347
1260
pass
1348
1261
class Dictionary(dict):
1349
1262
pass
1350
class ByteArray(bytes):
1351
pass
1352
1263
return mock_dbus_python
1353
1264
1354
1265
def call_method(self, bus, methodname, busname, objectpath,
1531
1442
# Make sure the dbus logger was suppressed
1532
1443
self.assertEqual(0, counting_handler.count)
1533
1444
1534
def test_Set_Secret_sends_bytearray(self):
1535
ret = [None]
1536
def func(*args, **kwargs):
1537
ret[0] = (args, kwargs)
1538
mock_dbus_python = self.MockDBusPython_func(func)
1539
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1540
bus.set_client_property("objectpath", "Secret", "value")
1541
expected_call = (("se.recompile.Mandos.Client", "Secret",
1542
mock_dbus_python.ByteArray(b"value")),
1543
{"dbus_interface":
1544
"org.freedesktop.DBus.Properties"})
1545
self.assertEqual(expected_call, ret[0])
1546
if sys.version_info.major == 2:
1547
self.assertIsInstance(ret[0][0][-1],
1548
mock_dbus_python.ByteArray)
1549
1550
1445
def test_get_object_converts_to_correct_exception(self):
1551
1446
bus = dbus_python_adapter.SystemBus(
1552
1447
self.fake_dbus_python_raises_exception_on_connect)
1621
1516
self.assertIs(obj1, obj1b)
1622
1517
1623
1518
1624
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1625
1626
def Stub_pydbus_func(self, func):
1627
class stub_pydbus(object):
1628
"""stub pydbus module"""
1629
class SystemBus(object):
1630
@staticmethod
1631
def get(busname, objectpath):
1632
DBusObject = collections.namedtuple(
1633
"DBusObject", ("methodname",))
1634
return {"interface":
1635
DBusObject(methodname=func)}
1636
return stub_pydbus
1637
1638
def call_method(self, bus, methodname, busname, objectpath,
1639
interface, *args):
1640
with self.assertLogs(log, logging.DEBUG):
1641
return bus.call_method(methodname, busname, objectpath,
1642
interface, *args)
1643
1644
def test_call_method_returns(self):
1645
expected_method_return = Unique()
1646
method_args = (Unique(), Unique())
1647
def func(*args):
1648
self.assertEqual(len(method_args), len(args))
1649
for marg, arg in zip(method_args, args):
1650
self.assertIs(marg, arg)
1651
return expected_method_return
1652
stub_pydbus = self.Stub_pydbus_func(func)
1653
bus = pydbus_adapter.SystemBus(stub_pydbus)
1654
ret = self.call_method(bus, "methodname", "busname",
1655
"objectpath", "interface",
1656
*method_args)
1657
self.assertIs(ret, expected_method_return)
1658
1659
def test_call_method_handles_exception(self):
1660
dbus_logger = logging.getLogger("dbus.proxies")
1661
1662
def func():
1663
raise gi.repository.GLib.Error()
1664
1665
stub_pydbus = self.Stub_pydbus_func(func)
1666
bus = pydbus_adapter.SystemBus(stub_pydbus)
1667
1668
with self.assertRaises(dbus.Error) as e:
1669
self.call_method(bus, "methodname", "busname",
1670
"objectpath", "interface")
1671
1672
self.assertNotIsInstance(e, dbus.ConnectFailed)
1673
1674
def test_get_converts_to_correct_exception(self):
1675
bus = pydbus_adapter.SystemBus(
1676
self.fake_pydbus_raises_exception_on_connect)
1677
with self.assertRaises(dbus.ConnectFailed):
1678
self.call_method(bus, "methodname", "busname",
1679
"objectpath", "interface")
1680
1681
class fake_pydbus_raises_exception_on_connect(object):
1682
"""fake dbus-python module"""
1683
@classmethod
1684
def SystemBus(cls):
1685
def get(busname, objectpath):
1686
raise gi.repository.GLib.Error()
1687
Bus = collections.namedtuple("Bus", ["get"])
1688
return Bus(get=get)
1689
1690
def test_set_property_uses_setattr(self):
1691
class Object(object):
1692
pass
1693
obj = Object()
1694
class pydbus_spy(object):
1695
class SystemBus(object):
1696
@staticmethod
1697
def get(busname, objectpath):
1698
return {"interface": obj}
1699
bus = pydbus_adapter.SystemBus(pydbus_spy)
1700
value = Unique()
1701
bus.set_property("busname", "objectpath", "interface", "key",
1702
value)
1703
self.assertIs(value, obj.key)
1704
1705
def test_get_suppresses_xml_deprecation_warning(self):
1706
if sys.version_info.major >= 3:
1707
return
1708
class stub_pydbus_get(object):
1709
class SystemBus(object):
1710
@staticmethod
1711
def get(busname, objectpath):
1712
warnings.warn_explicit(
1713
"deprecated", DeprecationWarning,
1714
"xml.etree.ElementTree", 0)
1715
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1716
with warnings.catch_warnings(record=True) as w:
1717
warnings.simplefilter("always")
1718
bus.get("busname", "objectpath")
1719
self.assertEqual(0, len(w))
1720
1721
1722
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1723
class stub_pydbus(object):
1724
"""stub pydbus module"""
1725
class SystemBus(object):
1726
@staticmethod
1727
def get(busname, objectpath):
1728
return Unique()
1729
1730
def setUp(self):
1731
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1732
1733
def test_returns_distinct_objectpaths(self):
1734
obj1 = self.bus.get("busname", "objectpath1")
1735
self.assertIsInstance(obj1, Unique)
1736
obj2 = self.bus.get("busname", "objectpath2")
1737
self.assertIsInstance(obj2, Unique)
1738
self.assertIsNot(obj1, obj2)
1739
1740
def test_returns_distinct_busnames(self):
1741
obj1 = self.bus.get("busname1", "objectpath")
1742
self.assertIsInstance(obj1, Unique)
1743
obj2 = self.bus.get("busname2", "objectpath")
1744
self.assertIsInstance(obj2, Unique)
1745
self.assertIsNot(obj1, obj2)
1746
1747
def test_returns_distinct_both(self):
1748
obj1 = self.bus.get("busname1", "objectpath")
1749
self.assertIsInstance(obj1, Unique)
1750
obj2 = self.bus.get("busname2", "objectpath")
1751
self.assertIsInstance(obj2, Unique)
1752
self.assertIsNot(obj1, obj2)
1753
1754
def test_returns_same(self):
1755
obj1 = self.bus.get("busname", "objectpath")
1756
self.assertIsInstance(obj1, Unique)
1757
obj2 = self.bus.get("busname", "objectpath")
1758
self.assertIsInstance(obj2, Unique)
1759
self.assertIs(obj1, obj2)
1760
1761
def test_returns_same_old(self):
1762
obj1 = self.bus.get("busname1", "objectpath1")
1763
self.assertIsInstance(obj1, Unique)
1764
obj2 = self.bus.get("busname2", "objectpath2")
1765
self.assertIsInstance(obj2, Unique)
1766
obj1b = self.bus.get("busname1", "objectpath1")
1767
self.assertIsInstance(obj1b, Unique)
1768
self.assertIsNot(obj1, obj2)
1769
self.assertIsNot(obj2, obj1b)
1770
self.assertIs(obj1, obj1b)
1771
1772
1773
1519
class Test_commands_from_options(unittest.TestCase):
1774
1520
1775
1521
def setUp(self):
2324
2070
def runTest(self):
2325
2071
if not hasattr(self, "command"):
2326
2072
return # Abstract TestCase class
2327
2328
if hasattr(self, "values_to_set"):
2329
cmd_args = [(value,) for value in self.values_to_set]
2330
values_to_get = getattr(self, "values_to_get",
2331
self.values_to_set)
2332
else:
2333
cmd_args = [() for x in range(len(self.values_to_get))]
2334
values_to_get = self.values_to_get
2335
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2336
for clientpath in self.bus.clients:
2337
self.bus.clients[clientpath][self.propname] = (
2338
Unique())
2339
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2340
for clientpath in self.bus.clients:
2341
value = (self.bus.clients[clientpath]
2342
[self.propname])
2073
values_to_get = getattr(self, "values_to_get",
2074
self.values_to_set)
2075
for value_to_set, value_to_get in zip(self.values_to_set,
2076
values_to_get):
2077
for clientpath in self.bus.clients:
2078
self.bus.clients[clientpath][self.propname] = Unique()
2079
self.run_command(value_to_set, self.bus.clients)
2080
for clientpath in self.bus.clients:
2081
value = self.bus.clients[clientpath][self.propname]
2343
2082
self.assertNotIsInstance(value, Unique)
2344
2083
self.assertEqual(value_to_get, value)
2345
2084
2085
def run_command(self, value, clients):
2086
self.command().run(clients, self.bus)
2087
2346
2088
2347
2089
class TestEnableCmd(TestPropertySetterCmd):
2348
2090
command = command.Enable
2349
2091
propname = "Enabled"
2350
values_to_get = [True]
2092
values_to_set = [True]
2351
2093
2352
2094
2353
2095
class TestDisableCmd(TestPropertySetterCmd):
2354
2096
command = command.Disable
2355
2097
propname = "Enabled"
2356
values_to_get = [False]
2098
values_to_set = [False]
2357
2099
2358
2100
2359
2101
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2360
2102
command = command.BumpTimeout
2361
2103
propname = "LastCheckedOK"
2362
values_to_get = [""]
2104
values_to_set = [""]
2363
2105
2364
2106
2365
2107
class TestStartCheckerCmd(TestPropertySetterCmd):
2366
2108
command = command.StartChecker
2367
2109
propname = "CheckerRunning"
2368
values_to_get = [True]
2110
values_to_set = [True]
2369
2111
2370
2112
2371
2113
class TestStopCheckerCmd(TestPropertySetterCmd):
2372
2114
command = command.StopChecker
2373
2115
propname = "CheckerRunning"
2374
values_to_get = [False]
2116
values_to_set = [False]
2375
2117
2376
2118
2377
2119
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2378
2120
command = command.ApproveByDefault
2379
2121
propname = "ApprovedByDefault"
2380
values_to_get = [True]
2122
values_to_set = [True]
2381
2123
2382
2124
2383
2125
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2384
2126
command = command.DenyByDefault
2385
2127
propname = "ApprovedByDefault"
2386
values_to_get = [False]
2387
2388
2389
class TestSetCheckerCmd(TestPropertySetterCmd):
2128
values_to_set = [False]
2129
2130
2131
class TestPropertySetterValueCmd(TestPropertySetterCmd):
2132
"""Abstract class for tests of PropertySetterValueCmd classes"""
2133
2134
def run_command(self, value, clients):
2135
self.command(value).run(clients, self.bus)
2136
2137
2138
class TestSetCheckerCmd(TestPropertySetterValueCmd):
2390
2139
command = command.SetChecker
2391
2140
propname = "Checker"
2392
2141
values_to_set = ["", ":", "fping -q -- %s"]
2393
2142
2394
2143
2395
class TestSetHostCmd(TestPropertySetterCmd):
2144
class TestSetHostCmd(TestPropertySetterValueCmd):
2396
2145
command = command.SetHost
2397
2146
propname = "Host"
2398
2147
values_to_set = ["192.0.2.3", "client.example.org"]
2399
2148
2400
2149
2401
class TestSetSecretCmd(TestPropertySetterCmd):
2150
class TestSetSecretCmd(TestPropertySetterValueCmd):
2402
2151
command = command.SetSecret
2403
2152
propname = "Secret"
2404
2153
values_to_set = [io.BytesIO(b""),
2406
2155
values_to_get = [f.getvalue() for f in values_to_set]
2407
2156
2408
2157
2409
class TestSetTimeoutCmd(TestPropertySetterCmd):
2158
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
2410
2159
command = command.SetTimeout
2411
2160
propname = "Timeout"
2412
2161
values_to_set = [datetime.timedelta(),
2417
2166
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2418
2167
2419
2168
2420
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
2169
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
2421
2170
command = command.SetExtendedTimeout
2422
2171
propname = "ExtendedTimeout"
2423
2172
values_to_set = [datetime.timedelta(),
2428
2177
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2429
2178
2430
2179
2431
class TestSetIntervalCmd(TestPropertySetterCmd):
2180
class TestSetIntervalCmd(TestPropertySetterValueCmd):
2432
2181
command = command.SetInterval
2433
2182
propname = "Interval"
2434
2183
values_to_set = [datetime.timedelta(),
2439
2188
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2440
2189
2441
2190
2442
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
2191
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
2443
2192
command = command.SetApprovalDelay
2444
2193
propname = "ApprovalDelay"
2445
2194
values_to_set = [datetime.timedelta(),
2450
2199
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2451
2200
2452
2201
2453
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
2202
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
2454
2203
command = command.SetApprovalDuration
2455
2204
propname = "ApprovalDuration"
2456
2205
values_to_set = [datetime.timedelta(),
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0