/mandos/trunk : revision 1092

1

#!/usr/bin/python

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --test" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

2

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

45

import io

46

import tempfile

47

import contextlib

48

import abc

48

49

try:

50

import pydbus

51

import gi

52

dbus_python = None

53

except ImportError:

54

import dbus as dbus_python

55

pydbus = None

56

class gi(object):

57

"""Dummy gi module, for the tests"""

58

class repository(object):

59

class GLib(object):

60

class Error(Exception):

61

pass

50

import dbus as dbus_python

62

51

63

52

# Show warnings by default

64

53

if not sys.warnoptions:

78

67

79

68

locale.setlocale(locale.LC_ALL, "")

80

69

81

version = "1.8.4"

70

version = "1.8.3"

82

71

83

72

84

73

def main():

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):

1780

1526

self.assert_command_from_args(["--is-enabled", "client"],

1781

1527

command.IsEnabled)

1782

1528

1783

def assert_command_from_args(self, args, command_cls, length=1,

1784

clients=None, **cmd_attrs):

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

1785

1531

"""Assert that parsing ARGS should result in an instance of

1786

1532

COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""

1787

1533

options = self.parser.parse_args(args)

1788

1534

check_option_syntax(self.parser, options)

1789

1535

commands = commands_from_options(options)

1790

self.assertEqual(length, len(commands))

1791

for command in commands:

1792

if isinstance(command, command_cls):

1793

break

1794

else:

1795

self.assertIsInstance(command, command_cls)

1796

if clients is not None:

1797

self.assertEqual(clients, options.client)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

1798

1539

for key, value in cmd_attrs.items():

1799

1540

self.assertEqual(value, getattr(command, key))

1800

1541

1801

def assert_commands_from_args(self, args, commands, clients=None):

1802

for cmd in commands:

1803

self.assert_command_from_args(args, cmd,

1804

length=len(commands),

1805

clients=clients)

1806

1807

1542

def test_is_enabled_short(self):

1808

1543

self.assert_command_from_args(["-V", "client"],

1809

1544

command.IsEnabled)

2000

1735

verbose=True)

2001

1736

2002

1737

2003

def test_manual_page_example_1(self):

2004

self.assert_command_from_args("--verbose".split(),

2005

command.PrintTable,

2006

clients=[],

2007

verbose=True)

2008

2009

def test_manual_page_example_2(self):

2010

self.assert_command_from_args(

2011

"--verbose foo1.example.org foo2.example.org".split(),

2012

command.PrintTable, clients=["foo1.example.org",

2013

"foo2.example.org"],

2014

verbose=True)

2015

2016

def test_manual_page_example_3(self):

2017

self.assert_command_from_args("--enable --all".split(),

2018

command.Enable,

2019

clients=[])

2020

2021

def test_manual_page_example_4(self):

2022

self.assert_commands_from_args(

2023

("--timeout=PT5M --interval=PT1M foo1.example.org"

2024

" foo2.example.org").split(),

2025

[command.SetTimeout, command.SetInterval],

2026

clients=["foo1.example.org", "foo2.example.org"])

2027

2028

def test_manual_page_example_5(self):

2029

self.assert_command_from_args("--approve --all".split(),

2030

command.Approve,

2031

clients=[])

2032

2033

2034

1738

class TestCommand(unittest.TestCase):

2035

1739

"""Abstract class for tests of command classes"""

2036

1740

2366

2070

def runTest(self):

2367

2071

if not hasattr(self, "command"):

2368

2072

return # Abstract TestCase class

2369

2370

if hasattr(self, "values_to_set"):

2371

cmd_args = [(value,) for value in self.values_to_set]

2372

values_to_get = getattr(self, "values_to_get",

2373

self.values_to_set)

2374

else:

2375

cmd_args = [() for x in range(len(self.values_to_get))]

2376

values_to_get = self.values_to_get

2377

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2378

for clientpath in self.bus.clients:

2379

self.bus.clients[clientpath][self.propname] = (

2380

Unique())

2381

self.command(*cmd_arg).run(self.bus.clients, self.bus)

2382

for clientpath in self.bus.clients:

2383

value = (self.bus.clients[clientpath]

2384

[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]

2385

2082

self.assertNotIsInstance(value, Unique)

2386

2083

self.assertEqual(value_to_get, value)

2387

2084

2085

def run_command(self, value, clients):

2086

self.command().run(clients, self.bus)

2087

2388

2088

2389

2089

class TestEnableCmd(TestPropertySetterCmd):

2390

2090

command = command.Enable

2391

2091

propname = "Enabled"

2392

values_to_get = [True]

2092

values_to_set = [True]

2393

2093

2394

2094

2395

2095

class TestDisableCmd(TestPropertySetterCmd):

2396

2096

command = command.Disable

2397

2097

propname = "Enabled"

2398

values_to_get = [False]

2098

values_to_set = [False]

2399

2099

2400

2100

2401

2101

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2402

2102

command = command.BumpTimeout

2403

2103

propname = "LastCheckedOK"

2404

values_to_get = [""]

2104

values_to_set = [""]

2405

2105

2406

2106

2407

2107

class TestStartCheckerCmd(TestPropertySetterCmd):

2408

2108

command = command.StartChecker

2409

2109

propname = "CheckerRunning"

2410

values_to_get = [True]

2110

values_to_set = [True]

2411

2111

2412

2112

2413

2113

class TestStopCheckerCmd(TestPropertySetterCmd):

2414

2114

command = command.StopChecker

2415

2115

propname = "CheckerRunning"

2416

values_to_get = [False]

2116

values_to_set = [False]

2417

2117

2418

2118

2419

2119

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2420

2120

command = command.ApproveByDefault

2421

2121

propname = "ApprovedByDefault"

2422

values_to_get = [True]

2122

values_to_set = [True]

2423

2123

2424

2124

2425

2125

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2426

2126

command = command.DenyByDefault

2427

2127

propname = "ApprovedByDefault"

2428

values_to_get = [False]

2429

2430

2431

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):

2432

2139

command = command.SetChecker

2433

2140

propname = "Checker"

2434

2141

values_to_set = ["", ":", "fping -q -- %s"]

2435

2142

2436

2143

2437

class TestSetHostCmd(TestPropertySetterCmd):

2144

class TestSetHostCmd(TestPropertySetterValueCmd):

2438

2145

command = command.SetHost

2439

2146

propname = "Host"

2440

2147

values_to_set = ["192.0.2.3", "client.example.org"]

2441

2148

2442

2149

2443

class TestSetSecretCmd(TestPropertySetterCmd):

2150

class TestSetSecretCmd(TestPropertySetterValueCmd):

2444

2151

command = command.SetSecret

2445

2152

propname = "Secret"

2446

2153

values_to_set = [io.BytesIO(b""),

2448

2155

values_to_get = [f.getvalue() for f in values_to_set]

2449

2156

2450

2157

2451

class TestSetTimeoutCmd(TestPropertySetterCmd):

2158

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

2452

2159

command = command.SetTimeout

2453

2160

propname = "Timeout"

2454

2161

values_to_set = [datetime.timedelta(),

2459

2166

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2460

2167

2461

2168

2462

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2169

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

2463

2170

command = command.SetExtendedTimeout

2464

2171

propname = "ExtendedTimeout"

2465

2172

values_to_set = [datetime.timedelta(),

2470

2177

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2471

2178

2472

2179

2473

class TestSetIntervalCmd(TestPropertySetterCmd):

2180

class TestSetIntervalCmd(TestPropertySetterValueCmd):

2474

2181

command = command.SetInterval

2475

2182

propname = "Interval"

2476

2183

values_to_set = [datetime.timedelta(),

2481

2188

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2482

2189

2483

2190

2484

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2191

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

2485

2192

command = command.SetApprovalDelay

2486

2193

propname = "ApprovalDelay"

2487

2194

values_to_set = [datetime.timedelta(),

2492

2199

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

2493

2200

2494

2201

2495

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2202

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

2496

2203

command = command.SetApprovalDuration

2497

2204

propname = "ApprovalDuration"

2498

2205

values_to_set = [datetime.timedelta(),