/mandos/trunk : revision 1131

1

#!/usr/bin/python

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*")))); -*-

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 --check" (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 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

45

import io

46

import tempfile

47

import contextlib

48

import abc

49

48

50

import dbus as dbus_python

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

51

62

52

63

# Show warnings by default

53

64

if not sys.warnoptions:

67

78

68

79

locale.setlocale(locale.LC_ALL, "")

69

80

70

version = "1.8.3"

81

version = "1.8.4"

71

82

72

83

73

84

def main():

82

93

if options.debug:

83

94

log.setLevel(logging.DEBUG)

84

95

85

bus = dbus_python_adapter.CachingBus(dbus_python)

96

if pydbus is not None:

97

bus = pydbus_adapter.CachingBus(pydbus)

98

else:

99

bus = dbus_python_adapter.CachingBus(dbus_python)

86

100

87

101

try:

88

102

all_clients = bus.get_clients_and_properties()

122

136

help="Select all clients")

123

137

parser.add_argument("-v", "--verbose", action="store_true",

124

138

help="Print all fields")

125

parser.add_argument("-j", "--dump-json", action="store_true",

139

parser.add_argument("-j", "--dump-json", dest="commands",

140

action="append_const", default=[],

141

const=command.DumpJSON(),

126

142

help="Dump client data in JSON format")

127

143

enable_disable = parser.add_mutually_exclusive_group()

128

enable_disable.add_argument("-e", "--enable", action="store_true",

144

enable_disable.add_argument("-e", "--enable", dest="commands",

145

action="append_const", default=[],

146

const=command.Enable(),

129

147

help="Enable client")

130

enable_disable.add_argument("-d", "--disable",

131

action="store_true",

148

enable_disable.add_argument("-d", "--disable", dest="commands",

149

action="append_const", default=[],

150

const=command.Disable(),

132

151

help="disable client")

133

parser.add_argument("-b", "--bump-timeout", action="store_true",

152

parser.add_argument("-b", "--bump-timeout", dest="commands",

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

134

155

help="Bump timeout for client")

135

156

start_stop_checker = parser.add_mutually_exclusive_group()

136

157

start_stop_checker.add_argument("--start-checker",

137

action="store_true",

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

138

161

help="Start checker for client")

139

start_stop_checker.add_argument("--stop-checker",

140

action="store_true",

162

start_stop_checker.add_argument("--stop-checker", dest="commands",

163

action="append_const", default=[],

164

const=command.StopChecker(),

141

165

help="Stop checker for client")

142

parser.add_argument("-V", "--is-enabled", action="store_true",

166

parser.add_argument("-V", "--is-enabled", dest="commands",

167

action="append_const", default=[],

168

const=command.IsEnabled(),

143

169

help="Check if client is enabled")

144

parser.add_argument("-r", "--remove", action="store_true",

170

parser.add_argument("-r", "--remove", dest="commands",

171

action="append_const", default=[],

172

const=command.Remove(),

145

173

help="Remove client")

146

parser.add_argument("-c", "--checker",

174

parser.add_argument("-c", "--checker", dest="commands",

175

action="append", default=[],

176

metavar="COMMAND", type=command.SetChecker,

147

177

help="Set checker command 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")

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

154

193

approve_deny_default = parser.add_mutually_exclusive_group()

155

194

approve_deny_default.add_argument(

156

"--approve-by-default", action="store_true",

157

default=None, dest="approved_by_default",

195

"--approve-by-default", dest="commands",

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

158

198

help="Set client to be approved by default")

159

199

approve_deny_default.add_argument(

160

"--deny-by-default", action="store_false",

161

dest="approved_by_default",

200

"--deny-by-default", dest="commands",

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

162

203

help="Set client to be denied by default")

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

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

171

223

approve_deny = parser.add_mutually_exclusive_group()

172

224

approve_deny.add_argument(

173

"-A", "--approve", action="store_true",

225

"-A", "--approve", dest="commands", action="append_const",

226

default=[], const=command.Approve(),

174

227

help="Approve any current client request")

175

approve_deny.add_argument("-D", "--deny", action="store_true",

228

approve_deny.add_argument("-D", "--deny", dest="commands",

229

action="append_const", default=[],

230

const=command.Deny(),

176

231

help="Deny any current client request")

177

232

parser.add_argument("--debug", action="store_true",

178

233

help="Debug mode (show D-Bus commands)")

195

250

def rfc3339_duration_to_delta(duration):

196

251

"""Parse an RFC 3339 "duration" and return a datetime.timedelta

197

252

198

>>> rfc3339_duration_to_delta("P7D")

199

datetime.timedelta(7)

200

>>> rfc3339_duration_to_delta("PT60S")

201

datetime.timedelta(0, 60)

202

>>> rfc3339_duration_to_delta("PT60M")

203

datetime.timedelta(0, 3600)

204

>>> rfc3339_duration_to_delta("P60M")

205

datetime.timedelta(1680)

206

>>> rfc3339_duration_to_delta("PT24H")

207

datetime.timedelta(1)

208

>>> rfc3339_duration_to_delta("P1W")

209

datetime.timedelta(7)

210

>>> rfc3339_duration_to_delta("PT5M30S")

211

datetime.timedelta(0, 330)

212

>>> rfc3339_duration_to_delta("P1DT3M20S")

213

datetime.timedelta(1, 200)

253

>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

254

True

255

>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

256

True

257

>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

258

True

259

>>> # 60 months

260

>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

261

True

262

>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

263

True

264

>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

265

True

266

>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

267

True

268

>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

269

True

214

270

>>> # Can not be empty:

215

271

>>> rfc3339_duration_to_delta("")

216

272

Traceback (most recent call last):

326

382

"""Parse an interval string as documented by Mandos before 1.6.1,

327

383

and return a datetime.timedelta

328

384

329

>>> parse_pre_1_6_1_interval('7d')

330

datetime.timedelta(7)

331

>>> parse_pre_1_6_1_interval('60s')

332

datetime.timedelta(0, 60)

333

>>> parse_pre_1_6_1_interval('60m')

334

datetime.timedelta(0, 3600)

335

>>> parse_pre_1_6_1_interval('24h')

336

datetime.timedelta(1)

337

>>> parse_pre_1_6_1_interval('1w')

338

datetime.timedelta(7)

339

>>> parse_pre_1_6_1_interval('5m 30s')

340

datetime.timedelta(0, 330)

341

>>> parse_pre_1_6_1_interval('')

342

datetime.timedelta(0)

385

>>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)

386

True

387

>>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)

388

True

389

>>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)

390

True

391

>>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)

392

True

393

>>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)

394

True

395

>>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)

396

True

397

>>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)

398

True

343

399

>>> # Ignore unknown characters, allow any order and repetitions

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

400

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)

401

True

346

402

347

403

"""

348

404

369

425

"""Apply additional restrictions on options, not expressible in

370

426

argparse"""

371

427

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

428

def has_commands(options, commands=None):

429

if commands is None:

430

commands = (command.Enable,

431

command.Disable,

432

command.BumpTimeout,

433

command.StartChecker,

434

command.StopChecker,

435

command.IsEnabled,

436

command.Remove,

437

command.SetChecker,

438

command.SetTimeout,

439

command.SetExtendedTimeout,

440

command.SetInterval,

441

command.ApproveByDefault,

442

command.DenyByDefault,

443

command.SetApprovalDelay,

444

command.SetApprovalDuration,

445

command.SetHost,

446

command.SetSecret,

447

command.Approve,

448

command.Deny)

449

return any(isinstance(cmd, commands)

450

for cmd in options.commands)

391

451

392

if has_actions(options) and not (options.client or options.all):

452

if has_commands(options) and not (options.client or options.all):

393

453

parser.error("Options require clients names or --all.")

394

if options.verbose and has_actions(options):

454

if options.verbose and has_commands(options):

395

455

parser.error("--verbose can only be used alone.")

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

456

if (has_commands(options, (command.DumpJSON,))

457

and (options.verbose or len(options.commands) > 1)):

398

458

parser.error("--dump-json can only be used alone.")

399

if options.all and not has_actions(options):

459

if options.all and not has_commands(options):

400

460

parser.error("--all requires an action.")

401

if options.is_enabled and len(options.client) > 1:

461

if (has_commands(options, (command.IsEnabled,))

462

and len(options.client) > 1):

402

463

parser.error("--is-enabled requires exactly one client")

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

464

if (len(options.commands) > 1

465

and has_commands(options, (command.Remove,))

466

and not has_commands(options, (command.Deny,))):

467

parser.error("--remove can only be combined with --deny")

409

468

410

469

411

470

class dbus(object):

513

572

for key, subval in value.items()}

514

573

return value

515

574

575

def set_client_property(self, objectpath, key, value):

576

if key == "Secret":

577

if not isinstance(value, bytes):

578

value = value.encode("utf-8")

579

value = self.dbus_python.ByteArray(value)

580

return self.set_property(self.busname, objectpath,

581

self.client_interface, key,

582

value)

516

583

517

584

class SilenceLogger(object):

518

585

"Simple context manager to silence a particular logger"

549

616

return new_object

550

617

551

618

619

class pydbus_adapter(object):

620

class SystemBus(dbus.MandosBus):

621

def __init__(self, module=pydbus):

622

self.pydbus = module

623

self.bus = self.pydbus.SystemBus()

624

625

@contextlib.contextmanager

626

def convert_exception(self, exception_class=dbus.Error):

627

try:

628

yield

629

except gi.repository.GLib.Error as e:

630

# This does what "raise from" would do

631

exc = exception_class(*e.args)

632

exc.__cause__ = e

633

raise exc

634

635

def call_method(self, methodname, busname, objectpath,

636

interface, *args):

637

proxy_object = self.get(busname, objectpath)

638

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

639

interface, methodname,

640

", ".join(repr(a) for a in args))

641

method = getattr(proxy_object[interface], methodname)

642

with self.convert_exception():

643

return method(*args)

644

645

def get(self, busname, objectpath):

646

log.debug("D-Bus: Connect to: (busname=%r, path=%r)",

647

busname, objectpath)

648

with self.convert_exception(dbus.ConnectFailed):

649

if sys.version_info.major <= 2:

650

with warnings.catch_warnings():

651

warnings.filterwarnings(

652

"ignore", "", DeprecationWarning,

653

r"^xml\.etree\.ElementTree$")

654

return self.bus.get(busname, objectpath)

655

else:

656

return self.bus.get(busname, objectpath)

657

658

def set_property(self, busname, objectpath, interface, key,

659

value):

660

proxy_object = self.get(busname, objectpath)

661

log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,

662

objectpath, self.properties_iface, interface,

663

key, value)

664

setattr(proxy_object[interface], key, value)

665

666

class CachingBus(SystemBus):

667

"""A caching layer for pydbus_adapter.SystemBus"""

668

def __init__(self, *args, **kwargs):

669

self.object_cache = {}

670

super(pydbus_adapter.CachingBus,

671

self).__init__(*args, **kwargs)

672

def get(self, busname, objectpath):

673

try:

674

return self.object_cache[(busname, objectpath)]

675

except KeyError:

676

new_object = (super(pydbus_adapter.CachingBus, self)

677

.get(busname, objectpath))

678

self.object_cache[(busname, objectpath)] = new_object

679

return new_object

680

681

552

682

def commands_from_options(options):

553

683

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

684

commands = list(options.commands)

685

686

def find_cmd(cmd, commands):

687

i = 0

688

for i, c in enumerate(commands):

689

if isinstance(c, cmd):

690

return i

691

return i+1

692

693

# If command.Remove is present, move any instances of command.Deny

694

# to occur ahead of command.Remove.

695

index_of_remove = find_cmd(command.Remove, commands)

696

before_remove = commands[:index_of_remove]

697

after_remove = commands[index_of_remove:]

698

cleaned_after = []

699

for cmd in after_remove:

700

if isinstance(cmd, command.Deny):

701

before_remove.append(cmd)

589

702

else:

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

703

cleaned_after.append(cmd)

704

if cleaned_after != after_remove:

705

commands = before_remove + cleaned_after

618

706

619

707

# If no command option has been given, show table of clients,

620

708

# optionally verbosely

835

923

def __init__(self, value):

836

924

self.value_to_set = value

837

925

926

@classmethod

927

def argparse(cls, argtype):

928

def cmdtype(arg):

929

return cls(argtype(arg))

930

return cmdtype

838

931

839

932

class SetChecker(PropertySetterValue):

840

933

propname = "Checker"

966

1059

967

1060

def test_actions_requires_client_or_all(self):

968

1061

for action, value in self.actions.items():

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1062

args = self.actionargs(action, value)

971

1063

with self.assertParseError():

972

self.check_option_syntax(options)

1064

self.parse_args(args)

973

1065

974

# This mostly corresponds to the definition from has_actions() in

1066

# This mostly corresponds to the definition from has_commands() in

975

1067

# check_option_syntax()

976

1068

actions = {

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,

1069

"--enable": None,

1070

"--disable": None,

1071

"--bump-timeout": None,

1072

"--start-checker": None,

1073

"--stop-checker": None,

1074

"--is-enabled": None,

1075

"--remove": None,

1076

"--checker": "x",

1077

"--timeout": "PT0S",

1078

"--extended-timeout": "PT0S",

1079

"--interval": "PT0S",

1080

"--approve-by-default": None,

1081

"--deny-by-default": None,

1082

"--approval-delay": "PT0S",

1083

"--approval-duration": "PT0S",

1084

"--host": "hostname",

1085

"--secret": "/dev/null",

1086

"--approve": None,

1087

"--deny": None,

998

1088

}

999

1089

1090

@staticmethod

1091

def actionargs(action, value, *args):

1092

if value is not None:

1093

return [action, value] + list(args)

1094

else:

1095

return [action] + list(args)

1096

1000

1097

@contextlib.contextmanager

1001

1098

def assertParseError(self):

1002

1099

with self.assertRaises(SystemExit) as e:

1007

1104

# /argparse.html#exiting-methods

1008

1105

self.assertEqual(2, e.exception.code)

1009

1106

1107

def parse_args(self, args):

1108

options = self.parser.parse_args(args)

1109

check_option_syntax(self.parser, options)

1110

1010

1111

@staticmethod

1011

1112

@contextlib.contextmanager

1012

1113

def redirect_stderr_to_devnull():

1023

1124

1024

1125

def test_actions_all_conflicts_with_verbose(self):

1025

1126

for action, value in self.actions.items():

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1127

args = self.actionargs(action, value, "--all",

1128

"--verbose")

1030

1129

with self.assertParseError():

1031

self.check_option_syntax(options)

1130

self.parse_args(args)

1032

1131

1033

1132

def test_actions_with_client_conflicts_with_verbose(self):

1034

1133

for action, value in self.actions.items():

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1134

args = self.actionargs(action, value, "--verbose",

1135

"client")

1039

1136

with self.assertParseError():

1040

self.check_option_syntax(options)

1137

self.parse_args(args)

1041

1138

1042

1139

def test_dump_json_conflicts_with_verbose(self):

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1140

args = ["--dump-json", "--verbose"]

1046

1141

with self.assertParseError():

1047

self.check_option_syntax(options)

1142

self.parse_args(args)

1048

1143

1049

1144

def test_dump_json_conflicts_with_action(self):

1050

1145

for action, value in self.actions.items():

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1146

args = self.actionargs(action, value, "--dump-json")

1054

1147

with self.assertParseError():

1055

self.check_option_syntax(options)

1148

self.parse_args(args)

1056

1149

1057

1150

def test_all_can_not_be_alone(self):

1058

options = self.parser.parse_args()

1059

options.all = True

1151

args = ["--all"]

1060

1152

with self.assertParseError():

1061

self.check_option_syntax(options)

1153

self.parse_args(args)

1062

1154

1063

1155

def test_all_is_ok_with_any_action(self):

1064

1156

for action, value in self.actions.items():

1065

options = self.parser.parse_args()

1066

setattr(options, action, value)

1067

options.all = True

1068

self.check_option_syntax(options)

1157

args = self.actionargs(action, value, "--all")

1158

self.parse_args(args)

1069

1159

1070

1160

def test_any_action_is_ok_with_one_client(self):

1071

1161

for action, value in self.actions.items():

1072

options = self.parser.parse_args()

1073

setattr(options, action, value)

1074

options.client = ["client"]

1075

self.check_option_syntax(options)

1162

args = self.actionargs(action, value, "client")

1163

self.parse_args(args)

1076

1164

1077

1165

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1079

1166

for action, value in self.actions.items():

1080

if action == "is_enabled":

1167

if action == "--is-enabled":

1081

1168

continue

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1169

args = self.actionargs(action, value, "client")

1170

self.parse_args(args)

1085

1171

1086

1172

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1088

1173

for action, value in self.actions.items():

1089

if action == "is_enabled":

1174

if action == "--is-enabled":

1090

1175

continue

1091

setattr(options, action, value)

1092

options.client = ["client1", "client2"]

1093

self.check_option_syntax(options)

1176

args = self.actionargs(action, value, "client1",

1177

"client2")

1178

self.parse_args(args)

1094

1179

1095

1180

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1096

1181

for action, value in self.actions.items():

1097

if action == "is_enabled":

1182

if action == "--is-enabled":

1098

1183

continue

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

options.client = ["client1", "client2"]

1102

self.check_option_syntax(options)

1184

args = self.actionargs(action, value, "client1",

1185

"client2")

1186

self.parse_args(args)

1103

1187

1104

1188

def test_is_enabled_fails_without_client(self):

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1189

args = ["--is-enabled"]

1107

1190

with self.assertParseError():

1108

self.check_option_syntax(options)

1191

self.parse_args(args)

1109

1192

1110

1193

def test_is_enabled_fails_with_two_clients(self):

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

options.client = ["client1", "client2"]

1194

args = ["--is-enabled", "client1", "client2"]

1114

1195

with self.assertParseError():

1115

self.check_option_syntax(options)

1196

self.parse_args(args)

1116

1197

1117

1198

def test_remove_can_only_be_combined_with_action_deny(self):

1118

1199

for action, value in self.actions.items():

1119

if action in {"remove", "deny"}:

1200

if action in {"--remove", "--deny"}:

1120

1201

continue

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1202

args = self.actionargs(action, value, "--all",

1203

"--remove")

1125

1204

with self.assertParseError():

1126

self.check_option_syntax(options)

1205

self.parse_args(args)

1127

1206

1128

1207

1129

1208

class Test_dbus_exceptions(unittest.TestCase):

1242

1321

@staticmethod

1243

1322

def get_object(busname, objectpath):

1244

1323

DBusObject = collections.namedtuple(

1245

"DBusObject", ("methodname",))

1324

"DBusObject", ("methodname", "Set"))

1246

1325

def method(*args, **kwargs):

1247

1326

self.assertEqual({"dbus_interface":

1248

1327

"interface"},

1249

1328

kwargs)

1250

1329

return func(*args)

1251

return DBusObject(methodname=method)

1330

def set_property(interface, key, value,

1331

dbus_interface=None):

1332

self.assertEqual(

1333

"org.freedesktop.DBus.Properties",

1334

dbus_interface)

1335

self.assertEqual("Secret", key)

1336

return func(interface, key, value,

1337

dbus_interface=dbus_interface)

1338

return DBusObject(methodname=method,

1339

Set=set_property)

1252

1340

class Boolean(object):

1253

1341

def __init__(self, value):

1254

1342

self.value = bool(value)

1260

1348

pass

1261

1349

class Dictionary(dict):

1262

1350

pass

1351

class ByteArray(bytes):

1352

pass

1263

1353

return mock_dbus_python

1264

1354

1265

1355

def call_method(self, bus, methodname, busname, objectpath,

1442

1532

# Make sure the dbus logger was suppressed

1443

1533

self.assertEqual(0, counting_handler.count)

1444

1534

1535

def test_Set_Secret_sends_bytearray(self):

1536

ret = [None]

1537

def func(*args, **kwargs):

1538

ret[0] = (args, kwargs)

1539

mock_dbus_python = self.MockDBusPython_func(func)

1540

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1541

bus.set_client_property("objectpath", "Secret", "value")

1542

expected_call = (("se.recompile.Mandos.Client", "Secret",

1543

mock_dbus_python.ByteArray(b"value")),

1544

{"dbus_interface":

1545

"org.freedesktop.DBus.Properties"})

1546

self.assertEqual(expected_call, ret[0])

1547

if sys.version_info.major == 2:

1548

self.assertIsInstance(ret[0][0][-1],

1549

mock_dbus_python.ByteArray)

1550

1445

1551

def test_get_object_converts_to_correct_exception(self):

1446

1552

bus = dbus_python_adapter.SystemBus(

1447

1553

self.fake_dbus_python_raises_exception_on_connect)

1516

1622

self.assertIs(obj1, obj1b)

1517

1623

1518

1624

1625

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1626

1627

def Stub_pydbus_func(self, func):

1628

class stub_pydbus(object):

1629

"""stub pydbus module"""

1630

class SystemBus(object):

1631

@staticmethod

1632

def get(busname, objectpath):

1633

DBusObject = collections.namedtuple(

1634

"DBusObject", ("methodname",))

1635

return {"interface":

1636

DBusObject(methodname=func)}

1637

return stub_pydbus

1638

1639

def call_method(self, bus, methodname, busname, objectpath,

1640

interface, *args):

1641

with self.assertLogs(log, logging.DEBUG):

1642

return bus.call_method(methodname, busname, objectpath,

1643

interface, *args)

1644

1645

def test_call_method_returns(self):

1646

expected_method_return = Unique()

1647

method_args = (Unique(), Unique())

1648

def func(*args):

1649

self.assertEqual(len(method_args), len(args))

1650

for marg, arg in zip(method_args, args):

1651

self.assertIs(marg, arg)

1652

return expected_method_return

1653

stub_pydbus = self.Stub_pydbus_func(func)

1654

bus = pydbus_adapter.SystemBus(stub_pydbus)

1655

ret = self.call_method(bus, "methodname", "busname",

1656

"objectpath", "interface",

1657

*method_args)

1658

self.assertIs(ret, expected_method_return)

1659

1660

def test_call_method_handles_exception(self):

1661

dbus_logger = logging.getLogger("dbus.proxies")

1662

1663

def func():

1664

raise gi.repository.GLib.Error()

1665

1666

stub_pydbus = self.Stub_pydbus_func(func)

1667

bus = pydbus_adapter.SystemBus(stub_pydbus)

1668

1669

with self.assertRaises(dbus.Error) as e:

1670

self.call_method(bus, "methodname", "busname",

1671

"objectpath", "interface")

1672

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1674

1675

def test_get_converts_to_correct_exception(self):

1676

bus = pydbus_adapter.SystemBus(

1677

self.fake_pydbus_raises_exception_on_connect)

1678

with self.assertRaises(dbus.ConnectFailed):

1679

self.call_method(bus, "methodname", "busname",

1680

"objectpath", "interface")

1681

1682

class fake_pydbus_raises_exception_on_connect(object):

1683

"""fake dbus-python module"""

1684

@classmethod

1685

def SystemBus(cls):

1686

def get(busname, objectpath):

1687

raise gi.repository.GLib.Error()

1688

Bus = collections.namedtuple("Bus", ["get"])

1689

return Bus(get=get)

1690

1691

def test_set_property_uses_setattr(self):

1692

class Object(object):

1693

pass

1694

obj = Object()

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1697

@staticmethod

1698

def get(busname, objectpath):

1699

return {"interface": obj}

1700

bus = pydbus_adapter.SystemBus(pydbus_spy)

1701

value = Unique()

1702

bus.set_property("busname", "objectpath", "interface", "key",

1703

value)

1704

self.assertIs(value, obj.key)

1705

1706

def test_get_suppresses_xml_deprecation_warning(self):

1707

if sys.version_info.major >= 3:

1708

return

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1711

@staticmethod

1712

def get(busname, objectpath):

1713

warnings.warn_explicit(

1714

"deprecated", DeprecationWarning,

1715

"xml.etree.ElementTree", 0)

1716

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1717

with warnings.catch_warnings(record=True) as w:

1718

warnings.simplefilter("always")

1719

bus.get("busname", "objectpath")

1720

self.assertEqual(0, len(w))

1721

1722

1723

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1724

class stub_pydbus(object):

1725

"""stub pydbus module"""

1726

class SystemBus(object):

1727

@staticmethod

1728

def get(busname, objectpath):

1729

return Unique()

1730

1731

def setUp(self):

1732

self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)

1733

1734

def test_returns_distinct_objectpaths(self):

1735

obj1 = self.bus.get("busname", "objectpath1")

1736

self.assertIsInstance(obj1, Unique)

1737

obj2 = self.bus.get("busname", "objectpath2")

1738

self.assertIsInstance(obj2, Unique)

1739

self.assertIsNot(obj1, obj2)

1740

1741

def test_returns_distinct_busnames(self):

1742

obj1 = self.bus.get("busname1", "objectpath")

1743

self.assertIsInstance(obj1, Unique)

1744

obj2 = self.bus.get("busname2", "objectpath")

1745

self.assertIsInstance(obj2, Unique)

1746

self.assertIsNot(obj1, obj2)

1747

1748

def test_returns_distinct_both(self):

1749

obj1 = self.bus.get("busname1", "objectpath")

1750

self.assertIsInstance(obj1, Unique)

1751

obj2 = self.bus.get("busname2", "objectpath")

1752

self.assertIsInstance(obj2, Unique)

1753

self.assertIsNot(obj1, obj2)

1754

1755

def test_returns_same(self):

1756

obj1 = self.bus.get("busname", "objectpath")

1757

self.assertIsInstance(obj1, Unique)

1758

obj2 = self.bus.get("busname", "objectpath")

1759

self.assertIsInstance(obj2, Unique)

1760

self.assertIs(obj1, obj2)

1761

1762

def test_returns_same_old(self):

1763

obj1 = self.bus.get("busname1", "objectpath1")

1764

self.assertIsInstance(obj1, Unique)

1765

obj2 = self.bus.get("busname2", "objectpath2")

1766

self.assertIsInstance(obj2, Unique)

1767

obj1b = self.bus.get("busname1", "objectpath1")

1768

self.assertIsInstance(obj1b, Unique)

1769

self.assertIsNot(obj1, obj2)

1770

self.assertIsNot(obj2, obj1b)

1771

self.assertIs(obj1, obj1b)

1772

1773

1519

1774

class Test_commands_from_options(unittest.TestCase):

1520

1775

1521

1776

def setUp(self):

1526

1781

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

1527

1782

command.IsEnabled)

1528

1783

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

1784

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

1785

clients=None, **cmd_attrs):

1531

1786

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

1532

1787

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

1533

1788

options = self.parser.parse_args(args)

1534

1789

check_option_syntax(self.parser, options)

1535

1790

commands = commands_from_options(options)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

1791

self.assertEqual(length, len(commands))

1792

for command in commands:

1793

if isinstance(command, command_cls):

1794

break

1795

else:

1796

self.assertIsInstance(command, command_cls)

1797

if clients is not None:

1798

self.assertEqual(clients, options.client)

1539

1799

for key, value in cmd_attrs.items():

1540

1800

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

1541

1801

1802

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

1803

for cmd in commands:

1804

self.assert_command_from_args(args, cmd,

1805

length=len(commands),

1806

clients=clients)

1807

1542

1808

def test_is_enabled_short(self):

1543

1809

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

1544

1810

command.IsEnabled)

1735

2001

verbose=True)

1736

2002

1737

2003

2004

def test_manual_page_example_1(self):

2005

self.assert_command_from_args("",

2006

command.PrintTable,

2007

clients=[],

2008

verbose=False)

2009

2010

def test_manual_page_example_2(self):

2011

self.assert_command_from_args(

2012

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

2013

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

2014

"foo2.example.org"],

2015

verbose=True)

2016

2017

def test_manual_page_example_3(self):

2018

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

2019

command.Enable,

2020

clients=[])

2021

2022

def test_manual_page_example_4(self):

2023

self.assert_commands_from_args(

2024

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

2025

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

2026

[command.SetTimeout, command.SetInterval],

2027

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

2028

2029

def test_manual_page_example_5(self):

2030

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

2031

command.Approve,

2032

clients=[])

2033

2034

1738

2035

class TestCommand(unittest.TestCase):

1739

2036

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

1740

2037

2070

2367

def runTest(self):

2071

2368

if not hasattr(self, "command"):

2072

2369

return # Abstract TestCase class

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]

2370

2371

if hasattr(self, "values_to_set"):

2372

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

2373

values_to_get = getattr(self, "values_to_get",

2374

self.values_to_set)

2375

else:

2376

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

2377

values_to_get = self.values_to_get

2378

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

2379

for clientpath in self.bus.clients:

2380

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

2381

Unique())

2382

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

2383

for clientpath in self.bus.clients:

2384

value = (self.bus.clients[clientpath]

2385

[self.propname])

2082

2386

self.assertNotIsInstance(value, Unique)

2083

2387

self.assertEqual(value_to_get, value)

2084

2388

2085

def run_command(self, value, clients):

2086

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

2087

2088

2389

2089

2390

class TestEnableCmd(TestPropertySetterCmd):

2090

2391

command = command.Enable

2091

2392

propname = "Enabled"

2092

values_to_set = [True]

2393

values_to_get = [True]

2093

2394

2094

2395

2095

2396

class TestDisableCmd(TestPropertySetterCmd):

2096

2397

command = command.Disable

2097

2398

propname = "Enabled"

2098

values_to_set = [False]

2399

values_to_get = [False]

2099

2400

2100

2401

2101

2402

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2102

2403

command = command.BumpTimeout

2103

2404

propname = "LastCheckedOK"

2104

values_to_set = [""]

2405

values_to_get = [""]

2105

2406

2106

2407

2107

2408

class TestStartCheckerCmd(TestPropertySetterCmd):

2108

2409

command = command.StartChecker

2109

2410

propname = "CheckerRunning"

2110

values_to_set = [True]

2411

values_to_get = [True]

2111

2412

2112

2413

2113

2414

class TestStopCheckerCmd(TestPropertySetterCmd):

2114

2415

command = command.StopChecker

2115

2416

propname = "CheckerRunning"

2116

values_to_set = [False]

2417

values_to_get = [False]

2117

2418

2118

2419

2119

2420

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2120

2421

command = command.ApproveByDefault

2121

2422

propname = "ApprovedByDefault"

2122

values_to_set = [True]

2423

values_to_get = [True]

2123

2424

2124

2425

2125

2426

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2126

2427

command = command.DenyByDefault

2127

2428

propname = "ApprovedByDefault"

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

2429

values_to_get = [False]

2430

2431

2432

class TestSetCheckerCmd(TestPropertySetterCmd):

2139

2433

command = command.SetChecker

2140

2434

propname = "Checker"

2141

2435

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

2142

2436

2143

2437

2144

class TestSetHostCmd(TestPropertySetterValueCmd):

2438

class TestSetHostCmd(TestPropertySetterCmd):

2145

2439

command = command.SetHost

2146

2440

propname = "Host"

2147

2441

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

2148

2442

2149

2443

2150

class TestSetSecretCmd(TestPropertySetterValueCmd):

2444

class TestSetSecretCmd(TestPropertySetterCmd):

2151

2445

command = command.SetSecret

2152

2446

propname = "Secret"

2153

2447

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

2155

2449

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

2156

2450

2157

2451

2158

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

2452

class TestSetTimeoutCmd(TestPropertySetterCmd):

2159

2453

command = command.SetTimeout

2160

2454

propname = "Timeout"

2161

2455

values_to_set = [datetime.timedelta(),

2166

2460

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

2167

2461

2168

2462

2169

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

2463

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2170

2464

command = command.SetExtendedTimeout

2171

2465

propname = "ExtendedTimeout"

2172

2466

values_to_set = [datetime.timedelta(),

2177

2471

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

2178

2472

2179

2473

2180

class TestSetIntervalCmd(TestPropertySetterValueCmd):

2474

class TestSetIntervalCmd(TestPropertySetterCmd):

2181

2475

command = command.SetInterval

2182

2476

propname = "Interval"

2183

2477

values_to_set = [datetime.timedelta(),

2188

2482

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

2189

2483

2190

2484

2191

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

2485

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2192

2486

command = command.SetApprovalDelay

2193

2487

propname = "ApprovalDelay"

2194

2488

values_to_set = [datetime.timedelta(),

2199

2493

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

2200

2494

2201

2495

2202

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

2496

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2203

2497

command = command.SetApprovalDuration

2204

2498

propname = "ApprovalDuration"

2205

2499

values_to_set = [datetime.timedelta(),