/mandos/trunk : revision 1274

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

26

from __future__ import (division, absolute_import, print_function,

28

27

unicode_literals)

29

28

33

32

pass

34

33

35

34

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

37

39

import locale

38

40

import datetime

39

41

import re

40

import os

41

42

import collections

42

43

import json

43

import unittest

44

import logging

45

44

import io

46

45

import tempfile

47

46

import contextlib

48

import abc

49

50

import dbus as dbus_python

47

48

if sys.version_info.major == 2:

49

__metaclass__ = type

50

str = unicode

51

input = raw_input

52

53

class gi:

54

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

55

class repository:

56

class GLib:

57

class Error(Exception):

58

pass

59

dbussy = None

60

ravel = None

61

dbus_python = None

62

pydbus = None

63

64

try:

65

import dbussy

66

import ravel

67

except ImportError:

68

try:

69

import pydbus

70

import gi

71

except ImportError:

72

import dbus as dbus_python

73

51

74

52

75

# Show warnings by default

53

76

if not sys.warnoptions:

54

77

import warnings

55

78

warnings.simplefilter("default")

56

79

57

log = logging.getLogger(sys.argv[0])

80

log = logging.getLogger(os.path.basename(sys.argv[0]))

58

81

logging.basicConfig(level="INFO", # Show info level messages

59

82

format="%(message)s") # Show basic log messages

60

83

61

84

logging.captureWarnings(True) # Show warnings via the logging system

62

85

63

86

if sys.version_info.major == 2:

64

str = unicode

65

87

import StringIO

66

88

io.StringIO = StringIO.StringIO

67

89

68

90

locale.setlocale(locale.LC_ALL, "")

69

91

70

version = "1.8.3"

92

version = "1.8.16"

71

93

72

94

73

95

def main():

80

102

clientnames = options.client

81

103

82

104

if options.debug:

83

log.setLevel(logging.DEBUG)

105

logging.getLogger("").setLevel(logging.DEBUG)

84

106

85

bus = dbus_python_adapter.CachingBus(dbus_python)

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

109

elif pydbus is not None:

110

bus = pydbus_adapter.CachingBus(pydbus)

111

else:

112

bus = dbus_python_adapter.CachingBus(dbus_python)

86

113

87

114

try:

88

115

all_clients = bus.get_clients_and_properties()

122

149

help="Select all clients")

123

150

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

124

151

help="Print all fields")

125

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

152

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

153

action="append_const", default=[],

154

const=command.DumpJSON(),

126

155

help="Dump client data in JSON format")

127

156

enable_disable = parser.add_mutually_exclusive_group()

128

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

157

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

158

action="append_const", default=[],

159

const=command.Enable(),

129

160

help="Enable client")

130

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

131

action="store_true",

161

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

162

action="append_const", default=[],

163

const=command.Disable(),

132

164

help="disable client")

133

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

165

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

166

action="append_const", default=[],

167

const=command.BumpTimeout(),

134

168

help="Bump timeout for client")

135

169

start_stop_checker = parser.add_mutually_exclusive_group()

136

170

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

137

action="store_true",

171

dest="commands",

172

action="append_const", default=[],

173

const=command.StartChecker(),

138

174

help="Start checker for client")

139

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

140

action="store_true",

175

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

176

action="append_const", default=[],

177

const=command.StopChecker(),

141

178

help="Stop checker for client")

142

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

179

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

180

action="append_const", default=[],

181

const=command.IsEnabled(),

143

182

help="Check if client is enabled")

144

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

183

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

184

action="append_const", default=[],

185

const=command.Remove(),

145

186

help="Remove client")

146

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

187

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

188

action="append", default=[],

189

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

147

190

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

191

parser.add_argument(

192

"-t", "--timeout", dest="commands", action="append",

193

default=[], metavar="TIME",

194

type=command.SetTimeout.argparse(string_to_delta),

195

help="Set timeout for client")

196

parser.add_argument(

197

"--extended-timeout", dest="commands", action="append",

198

default=[], metavar="TIME",

199

type=command.SetExtendedTimeout.argparse(string_to_delta),

200

help="Set extended timeout for client")

201

parser.add_argument(

202

"-i", "--interval", dest="commands", action="append",

203

default=[], metavar="TIME",

204

type=command.SetInterval.argparse(string_to_delta),

205

help="Set checker interval for client")

154

206

approve_deny_default = parser.add_mutually_exclusive_group()

155

207

approve_deny_default.add_argument(

156

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

157

default=None, dest="approved_by_default",

208

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

209

action="append_const", default=[],

210

const=command.ApproveByDefault(),

158

211

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

159

212

approve_deny_default.add_argument(

160

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

161

dest="approved_by_default",

213

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

214

action="append_const", default=[],

215

const=command.DenyByDefault(),

162

216

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

217

parser.add_argument(

218

"--approval-delay", dest="commands", action="append",

219

default=[], metavar="TIME",

220

type=command.SetApprovalDelay.argparse(string_to_delta),

221

help="Set delay before client approve/deny")

222

parser.add_argument(

223

"--approval-duration", dest="commands", action="append",

224

default=[], metavar="TIME",

225

type=command.SetApprovalDuration.argparse(string_to_delta),

226

help="Set duration of one client approval")

227

parser.add_argument("-H", "--host", dest="commands",

228

action="append", default=[], metavar="STRING",

229

type=command.SetHost,

230

help="Set host for client")

231

parser.add_argument(

232

"-s", "--secret", dest="commands", action="append",

233

default=[], metavar="FILENAME",

234

type=command.SetSecret.argparse(argparse.FileType(mode="rb")),

235

help="Set password blob (file) for client")

171

236

approve_deny = parser.add_mutually_exclusive_group()

172

237

approve_deny.add_argument(

173

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

238

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

239

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

174

240

help="Approve any current client request")

175

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

241

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

242

action="append_const", default=[],

243

const=command.Deny(),

176

244

help="Deny any current client request")

177

245

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

178

246

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

188

256

return rfc3339_duration_to_delta(interval)

189

257

except ValueError as e:

190

258

log.warning("%s - Parsing as pre-1.6.1 interval instead",

191

' '.join(e.args))

259

" ".join(e.args))

192

260

return parse_pre_1_6_1_interval(interval)

193

261

194

262

195

263

def rfc3339_duration_to_delta(duration):

196

264

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

197

265

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)

266

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

267

True

268

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

269

True

270

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

271

True

272

>>> # 60 months

273

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

274

True

275

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

276

True

277

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

278

True

279

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

280

True

281

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

282

True

214

283

>>> # Can not be empty:

215

284

>>> rfc3339_duration_to_delta("")

216

285

Traceback (most recent call last):

326

395

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

327

396

and return a datetime.timedelta

328

397

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)

398

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

399

True

400

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

401

True

402

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

403

True

404

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

405

True

406

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

407

True

408

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

409

True

410

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

411

True

343

412

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

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

413

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

414

True

346

415

347

416

"""

348

417

369

438

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

370

439

argparse"""

371

440

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

441

def has_commands(options, commands=None):

442

if commands is None:

443

commands = (command.Enable,

444

command.Disable,

445

command.BumpTimeout,

446

command.StartChecker,

447

command.StopChecker,

448

command.IsEnabled,

449

command.Remove,

450

command.SetChecker,

451

command.SetTimeout,

452

command.SetExtendedTimeout,

453

command.SetInterval,

454

command.ApproveByDefault,

455

command.DenyByDefault,

456

command.SetApprovalDelay,

457

command.SetApprovalDuration,

458

command.SetHost,

459

command.SetSecret,

460

command.Approve,

461

command.Deny)

462

return any(isinstance(cmd, commands)

463

for cmd in options.commands)

391

464

392

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

465

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

393

466

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

394

if options.verbose and has_actions(options):

467

if options.verbose and has_commands(options):

395

468

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

469

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

470

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

398

471

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

399

if options.all and not has_actions(options):

472

if options.all and not has_commands(options):

400

473

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

401

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

474

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

475

and len(options.client) > 1):

402

476

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

409

410

411

class dbus(object):

412

413

class SystemBus(object):

477

if (len(options.commands) > 1

478

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

479

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

480

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

481

482

483

class dbus:

484

485

class SystemBus:

414

486

415

487

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

416

488

def get_managed_objects(self, busname, objectpath):

425

497

self.properties_iface, interface, key,

426

498

value)

427

499

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

428

504

429

505

class MandosBus(SystemBus):

430

506

busname_domain = "se.recompile"

462

538

pass

463

539

464

540

465

class dbus_python_adapter(object):

541

class dbus_python_adapter:

466

542

467

543

class SystemBus(dbus.MandosBus):

468

544

"""Use dbus-python"""

513

589

for key, subval in value.items()}

514

590

return value

515

591

592

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

593

if key == "Secret":

594

if not isinstance(value, bytes):

595

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

596

value = self.dbus_python.ByteArray(value)

597

return self.set_property(self.busname, objectpath,

598

self.client_interface, key,

599

value)

516

600

517

class SilenceLogger(object):

601

class SilenceLogger:

518

602

"Simple context manager to silence a particular logger"

519

603

def __init__(self, loggername):

520

604

self.logger = logging.getLogger(loggername)

549

633

return new_object

550

634

551

635

636

class pydbus_adapter:

637

class SystemBus(dbus.MandosBus):

638

def __init__(self, module=pydbus):

639

self.pydbus = module

640

self.bus = self.pydbus.SystemBus()

641

642

@contextlib.contextmanager

643

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

644

try:

645

yield

646

except gi.repository.GLib.Error as e:

647

# This does what "raise from" would do

648

exc = exception_class(*e.args)

649

exc.__cause__ = e

650

raise exc

651

652

def call_method(self, methodname, busname, objectpath,

653

interface, *args):

654

proxy_object = self.get(busname, objectpath)

655

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

656

interface, methodname,

657

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

658

method = getattr(proxy_object[interface], methodname)

659

with self.convert_exception():

660

return method(*args)

661

662

def get(self, busname, objectpath):

663

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

664

busname, objectpath)

665

with self.convert_exception(dbus.ConnectFailed):

666

if sys.version_info.major <= 2:

667

with warnings.catch_warnings():

668

warnings.filterwarnings(

669

"ignore", "", DeprecationWarning,

670

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

671

return self.bus.get(busname, objectpath)

672

else:

673

return self.bus.get(busname, objectpath)

674

675

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

676

value):

677

proxy_object = self.get(busname, objectpath)

678

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

679

objectpath, self.properties_iface, interface,

680

key, value)

681

setattr(proxy_object[interface], key, value)

682

683

class CachingBus(SystemBus):

684

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

685

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

686

self.object_cache = {}

687

super(pydbus_adapter.CachingBus,

688

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

689

def get(self, busname, objectpath):

690

try:

691

return self.object_cache[(busname, objectpath)]

692

except KeyError:

693

new_object = (super(pydbus_adapter.CachingBus, self)

694

.get(busname, objectpath))

695

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

696

return new_object

697

698

699

class dbussy_adapter:

700

class SystemBus(dbus.SystemBus):

701

"""Use DBussy"""

702

703

def __init__(self, dbussy, ravel):

704

self.dbussy = dbussy

705

self.ravel = ravel

706

self.bus = ravel.system_bus()

707

708

@contextlib.contextmanager

709

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

710

try:

711

yield

712

except self.dbussy.DBusError as e:

713

# This does what "raise from" would do

714

exc = exception_class(*e.args)

715

exc.__cause__ = e

716

raise exc

717

718

def call_method(self, methodname, busname, objectpath,

719

interface, *args):

720

proxy_object = self.get_object(busname, objectpath)

721

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

722

interface, methodname,

723

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

724

iface = proxy_object.get_interface(interface)

725

method = getattr(iface, methodname)

726

with self.convert_exception(dbus.Error):

727

value = method(*args)

728

# DBussy returns values either as an empty list or as a

729

# list of one element with the return value

730

if value:

731

return self.type_filter(value[0])

732

733

def get_object(self, busname, objectpath):

734

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

735

busname, objectpath)

736

with self.convert_exception(dbus.ConnectFailed):

737

return self.bus[busname][objectpath]

738

739

def type_filter(self, value):

740

"""Convert the most bothersome types to Python types"""

741

# A D-Bus Variant value is represented as the Python type

742

# Tuple[dbussy.DBUS.Signature, Any]

743

if isinstance(value, tuple):

744

if (len(value) == 2

745

and isinstance(value[0],

746

self.dbussy.DBUS.Signature)):

747

return self.type_filter(value[1])

748

elif isinstance(value, self.dbussy.DBUS.ObjectPath):

749

return str(value)

750

# Also recurse into dictionaries

751

elif isinstance(value, dict):

752

return {self.type_filter(key):

753

self.type_filter(subval)

754

for key, subval in value.items()}

755

return value

756

757

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

758

value):

759

proxy_object = self.get_object(busname, objectpath)

760

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

761

objectpath, self.properties_iface, interface,

762

key, value)

763

if key == "Secret":

764

# DBussy wants a Byte Array to be a sequence of

765

# values, not a byte string

766

value = tuple(value)

767

setattr(proxy_object.get_interface(interface), key, value)

768

769

class MandosBus(SystemBus, dbus.MandosBus):

770

pass

771

772

class CachingBus(MandosBus):

773

"""A caching layer for dbussy_adapter.MandosBus"""

774

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

775

self.object_cache = {}

776

super(dbussy_adapter.CachingBus, self).__init__(*args,

777

**kwargs)

778

def get_object(self, busname, objectpath):

779

try:

780

return self.object_cache[(busname, objectpath)]

781

except KeyError:

782

new_object = super(

783

dbussy_adapter.CachingBus,

784

self).get_object(busname, objectpath)

785

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

786

return new_object

787

788

552

789

def commands_from_options(options):

553

790

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

791

commands = list(options.commands)

792

793

def find_cmd(cmd, commands):

794

i = 0

795

for i, c in enumerate(commands):

796

if isinstance(c, cmd):

797

return i

798

return i+1

799

800

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

801

# to occur ahead of command.Remove.

802

index_of_remove = find_cmd(command.Remove, commands)

803

before_remove = commands[:index_of_remove]

804

after_remove = commands[index_of_remove:]

805

cleaned_after = []

806

for cmd in after_remove:

807

if isinstance(cmd, command.Deny):

808

before_remove.append(cmd)

589

809

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

810

cleaned_after.append(cmd)

811

if cleaned_after != after_remove:

812

commands = before_remove + cleaned_after

618

813

619

814

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

620

815

# optionally verbosely

624

819

return commands

625

820

626

821

627

class command(object):

822

class command:

628

823

"""A namespace for command classes"""

629

824

630

class Base(object):

825

class Base:

631

826

"""Abstract base class for commands"""

632

827

def run(self, clients, bus=None):

633

828

"""Normal commands should implement run_on_one_client(),

681

876

{key: properties[key]

682

877

for key in self.all_keywords}

683

878

for properties in clients.values()}

684

print(json.dumps(data, indent=4, separators=(',', ': ')))

879

print(json.dumps(data, indent=4, separators=(",", ": ")))

685

880

686

881

687

882

class PrintTable(Output):

696

891

keywords = self.all_keywords

697

892

print(self.TableOfClients(clients.values(), keywords))

698

893

699

class TableOfClients(object):

894

class TableOfClients:

700

895

tableheaders = {

701

896

"Name": "Name",

702

897

"Enabled": "Enabled",

835

1030

def __init__(self, value):

836

1031

self.value_to_set = value

837

1032

1033

@classmethod

1034

def argparse(cls, argtype):

1035

def cmdtype(arg):

1036

return cls(argtype(arg))

1037

return cmdtype

838

1038

839

1039

class SetChecker(PropertySetterValue):

840

1040

propname = "Checker"

928

1128

"output"))

929

1129

930

1130

931

class Unique(object):

1131

class Unique:

932

1132

"""Class for objects which exist only to be unique objects, since

933

1133

unittest.mock.sentinel only exists in Python 3.3"""

934

1134

936

1136

class Test_string_to_delta(TestCaseWithAssertLogs):

937

1137

# Just test basic RFC 3339 functionality here, the doc string for

938

1138

# rfc3339_duration_to_delta() already has more comprehensive

939

# tests, which is run by doctest.

1139

# tests, which are run by doctest.

940

1140

941

1141

def test_rfc3339_zero_seconds(self):

942

1142

self.assertEqual(datetime.timedelta(),

966

1166

967

1167

def test_actions_requires_client_or_all(self):

968

1168

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

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1169

args = self.actionargs(action, value)

971

1170

with self.assertParseError():

972

self.check_option_syntax(options)

1171

self.parse_args(args)

973

1172

974

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

1173

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

975

1174

# check_option_syntax()

976

1175

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,

1176

"--enable": None,

1177

"--disable": None,

1178

"--bump-timeout": None,

1179

"--start-checker": None,

1180

"--stop-checker": None,

1181

"--is-enabled": None,

1182

"--remove": None,

1183

"--checker": "x",

1184

"--timeout": "PT0S",

1185

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

1186

"--interval": "PT0S",

1187

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

1188

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

1189

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

1190

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

1191

"--host": "hostname",

1192

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

1193

"--approve": None,

1194

"--deny": None,

998

1195

}

999

1196

1197

@staticmethod

1198

def actionargs(action, value, *args):

1199

if value is not None:

1200

return [action, value] + list(args)

1201

else:

1202

return [action] + list(args)

1203

1000

1204

@contextlib.contextmanager

1001

1205

def assertParseError(self):

1002

1206

with self.assertRaises(SystemExit) as e:

1007

1211

# /argparse.html#exiting-methods

1008

1212

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

1009

1213

1214

def parse_args(self, args):

1215

options = self.parser.parse_args(args)

1216

check_option_syntax(self.parser, options)

1217

1010

1218

@staticmethod

1011

1219

@contextlib.contextmanager

1012

1220

def redirect_stderr_to_devnull():

1023

1231

1024

1232

def test_actions_all_conflicts_with_verbose(self):

1025

1233

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

1234

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

1235

"--verbose")

1030

1236

with self.assertParseError():

1031

self.check_option_syntax(options)

1237

self.parse_args(args)

1032

1238

1033

1239

def test_actions_with_client_conflicts_with_verbose(self):

1034

1240

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

1241

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

1242

"client")

1039

1243

with self.assertParseError():

1040

self.check_option_syntax(options)

1244

self.parse_args(args)

1041

1245

1042

1246

def test_dump_json_conflicts_with_verbose(self):

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1247

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

1046

1248

with self.assertParseError():

1047

self.check_option_syntax(options)

1249

self.parse_args(args)

1048

1250

1049

1251

def test_dump_json_conflicts_with_action(self):

1050

1252

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1253

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

1054

1254

with self.assertParseError():

1055

self.check_option_syntax(options)

1255

self.parse_args(args)

1056

1256

1057

1257

def test_all_can_not_be_alone(self):

1058

options = self.parser.parse_args()

1059

options.all = True

1258

args = ["--all"]

1060

1259

with self.assertParseError():

1061

self.check_option_syntax(options)

1260

self.parse_args(args)

1062

1261

1063

1262

def test_all_is_ok_with_any_action(self):

1064

1263

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)

1264

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

1265

self.parse_args(args)

1069

1266

1070

1267

def test_any_action_is_ok_with_one_client(self):

1071

1268

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)

1269

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

1270

self.parse_args(args)

1076

1271

1077

1272

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1079

1273

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

1080

if action == "is_enabled":

1274

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

1081

1275

continue

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1276

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

1277

self.parse_args(args)

1085

1278

1086

1279

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1088

1280

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

1089

if action == "is_enabled":

1281

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

1090

1282

continue

1091

setattr(options, action, value)

1092

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

1093

self.check_option_syntax(options)

1283

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

1284

"client2")

1285

self.parse_args(args)

1094

1286

1095

1287

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1096

1288

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

1097

if action == "is_enabled":

1289

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

1098

1290

continue

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

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

1102

self.check_option_syntax(options)

1291

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

1292

"client2")

1293

self.parse_args(args)

1103

1294

1104

1295

def test_is_enabled_fails_without_client(self):

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1296

args = ["--is-enabled"]

1107

1297

with self.assertParseError():

1108

self.check_option_syntax(options)

1298

self.parse_args(args)

1109

1299

1110

1300

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

1301

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

1114

1302

with self.assertParseError():

1115

self.check_option_syntax(options)

1303

self.parse_args(args)

1116

1304

1117

1305

def test_remove_can_only_be_combined_with_action_deny(self):

1118

1306

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

1119

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

1307

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

1120

1308

continue

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1309

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

1310

"--remove")

1125

1311

with self.assertParseError():

1126

self.check_option_syntax(options)

1312

self.parse_args(args)

1127

1313

1128

1314

1129

1315

class Test_dbus_exceptions(unittest.TestCase):

1232

1418

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1233

1419

1234

1420

def MockDBusPython_func(self, func):

1235

class mock_dbus_python(object):

1421

class mock_dbus_python:

1236

1422

"""mock dbus-python module"""

1237

class exceptions(object):

1423

class exceptions:

1238

1424

"""Pseudo-namespace"""

1239

1425

class DBusException(Exception):

1240

1426

pass

1241

class SystemBus(object):

1427

class SystemBus:

1242

1428

@staticmethod

1243

1429

def get_object(busname, objectpath):

1244

1430

DBusObject = collections.namedtuple(

1245

"DBusObject", ("methodname",))

1431

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

1246

1432

def method(*args, **kwargs):

1247

1433

self.assertEqual({"dbus_interface":

1248

1434

"interface"},

1249

1435

kwargs)

1250

1436

return func(*args)

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

1437

def set_property(interface, key, value,

1438

dbus_interface=None):

1439

self.assertEqual(

1440

"org.freedesktop.DBus.Properties",

1441

dbus_interface)

1442

self.assertEqual("Secret", key)

1443

return func(interface, key, value,

1444

dbus_interface=dbus_interface)

1445

return DBusObject(methodname=method,

1446

Set=set_property)

1447

class Boolean:

1253

1448

def __init__(self, value):

1254

1449

self.value = bool(value)

1255

1450

def __bool__(self):

1260

1455

pass

1261

1456

class Dictionary(dict):

1262

1457

pass

1458

class ByteArray(bytes):

1459

pass

1263

1460

return mock_dbus_python

1264

1461

1265

1462

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

1437

1634

finally:

1438

1635

dbus_logger.removeFilter(counting_handler)

1439

1636

1440

self.assertNotIsInstance(e, dbus.ConnectFailed)

1637

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1441

1638

1442

1639

# Make sure the dbus logger was suppressed

1443

1640

self.assertEqual(0, counting_handler.count)

1444

1641

1642

def test_Set_Secret_sends_bytearray(self):

1643

ret = [None]

1644

def func(*args, **kwargs):

1645

ret[0] = (args, kwargs)

1646

mock_dbus_python = self.MockDBusPython_func(func)

1647

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1648

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

1649

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

1650

mock_dbus_python.ByteArray(b"value")),

1651

{"dbus_interface":

1652

"org.freedesktop.DBus.Properties"})

1653

self.assertEqual(expected_call, ret[0])

1654

if sys.version_info.major == 2:

1655

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

1656

mock_dbus_python.ByteArray)

1657

1445

1658

def test_get_object_converts_to_correct_exception(self):

1446

1659

bus = dbus_python_adapter.SystemBus(

1447

1660

self.fake_dbus_python_raises_exception_on_connect)

1449

1662

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

1450

1663

"objectpath", "interface")

1451

1664

1452

class fake_dbus_python_raises_exception_on_connect(object):

1665

class fake_dbus_python_raises_exception_on_connect:

1453

1666

"""fake dbus-python module"""

1454

class exceptions(object):

1667

class exceptions:

1455

1668

"""Pseudo-namespace"""

1456

1669

class DBusException(Exception):

1457

1670

pass

1465

1678

1466

1679

1467

1680

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1468

class mock_dbus_python(object):

1681

class mock_dbus_python:

1469

1682

"""mock dbus-python modules"""

1470

class SystemBus(object):

1683

class SystemBus:

1471

1684

@staticmethod

1472

1685

def get_object(busname, objectpath):

1473

1686

return Unique()

1516

1729

self.assertIs(obj1, obj1b)

1517

1730

1518

1731

1732

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1733

1734

def Stub_pydbus_func(self, func):

1735

class stub_pydbus:

1736

"""stub pydbus module"""

1737

class SystemBus:

1738

@staticmethod

1739

def get(busname, objectpath):

1740

DBusObject = collections.namedtuple(

1741

"DBusObject", ("methodname",))

1742

return {"interface":

1743

DBusObject(methodname=func)}

1744

return stub_pydbus

1745

1746

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

1747

interface, *args):

1748

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

1749

return bus.call_method(methodname, busname, objectpath,

1750

interface, *args)

1751

1752

def test_call_method_returns(self):

1753

expected_method_return = Unique()

1754

method_args = (Unique(), Unique())

1755

def func(*args):

1756

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

1757

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

1758

self.assertIs(marg, arg)

1759

return expected_method_return

1760

stub_pydbus = self.Stub_pydbus_func(func)

1761

bus = pydbus_adapter.SystemBus(stub_pydbus)

1762

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

1763

"objectpath", "interface",

1764

*method_args)

1765

self.assertIs(ret, expected_method_return)

1766

1767

def test_call_method_handles_exception(self):

1768

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

1769

1770

def func():

1771

raise gi.repository.GLib.Error()

1772

1773

stub_pydbus = self.Stub_pydbus_func(func)

1774

bus = pydbus_adapter.SystemBus(stub_pydbus)

1775

1776

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

1777

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

1778

"objectpath", "interface")

1779

1780

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

1781

1782

def test_get_converts_to_correct_exception(self):

1783

bus = pydbus_adapter.SystemBus(

1784

self.fake_pydbus_raises_exception_on_connect)

1785

with self.assertRaises(dbus.ConnectFailed):

1786

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

1787

"objectpath", "interface")

1788

1789

class fake_pydbus_raises_exception_on_connect:

1790

"""fake dbus-python module"""

1791

@classmethod

1792

def SystemBus(cls):

1793

def get(busname, objectpath):

1794

raise gi.repository.GLib.Error()

1795

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

1796

return Bus(get=get)

1797

1798

def test_set_property_uses_setattr(self):

1799

class Object:

1800

pass

1801

obj = Object()

1802

class pydbus_spy:

1803

class SystemBus:

1804

@staticmethod

1805

def get(busname, objectpath):

1806

return {"interface": obj}

1807

bus = pydbus_adapter.SystemBus(pydbus_spy)

1808

value = Unique()

1809

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

1810

value)

1811

self.assertIs(value, obj.key)

1812

1813

def test_get_suppresses_xml_deprecation_warning(self):

1814

if sys.version_info.major >= 3:

1815

return

1816

class stub_pydbus_get:

1817

class SystemBus:

1818

@staticmethod

1819

def get(busname, objectpath):

1820

warnings.warn_explicit(

1821

"deprecated", DeprecationWarning,

1822

"xml.etree.ElementTree", 0)

1823

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1824

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

1825

warnings.simplefilter("always")

1826

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

1827

self.assertEqual(0, len(w))

1828

1829

1830

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1831

class stub_pydbus:

1832

"""stub pydbus module"""

1833

class SystemBus:

1834

@staticmethod

1835

def get(busname, objectpath):

1836

return Unique()

1837

1838

def setUp(self):

1839

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

1840

1841

def test_returns_distinct_objectpaths(self):

1842

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

1843

self.assertIsInstance(obj1, Unique)

1844

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

1845

self.assertIsInstance(obj2, Unique)

1846

self.assertIsNot(obj1, obj2)

1847

1848

def test_returns_distinct_busnames(self):

1849

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

1850

self.assertIsInstance(obj1, Unique)

1851

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

1852

self.assertIsInstance(obj2, Unique)

1853

self.assertIsNot(obj1, obj2)

1854

1855

def test_returns_distinct_both(self):

1856

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

1857

self.assertIsInstance(obj1, Unique)

1858

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

1859

self.assertIsInstance(obj2, Unique)

1860

self.assertIsNot(obj1, obj2)

1861

1862

def test_returns_same(self):

1863

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

1864

self.assertIsInstance(obj1, Unique)

1865

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

1866

self.assertIsInstance(obj2, Unique)

1867

self.assertIs(obj1, obj2)

1868

1869

def test_returns_same_old(self):

1870

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

1871

self.assertIsInstance(obj1, Unique)

1872

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

1873

self.assertIsInstance(obj2, Unique)

1874

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

1875

self.assertIsInstance(obj1b, Unique)

1876

self.assertIsNot(obj1, obj2)

1877

self.assertIsNot(obj2, obj1b)

1878

self.assertIs(obj1, obj1b)

1879

1880

1881

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1882

1883

class dummy_dbussy:

1884

class DBUS:

1885

class ObjectPath(str):

1886

pass

1887

class DBusError(Exception):

1888

pass

1889

1890

def fake_ravel_func(self, func):

1891

class fake_ravel:

1892

@staticmethod

1893

def system_bus():

1894

class DBusInterfaceProxy:

1895

@staticmethod

1896

def methodname(*args):

1897

return [func(*args)]

1898

class DBusObject:

1899

@staticmethod

1900

def get_interface(interface):

1901

if interface == "interface":

1902

return DBusInterfaceProxy()

1903

return {"busname": {"objectpath": DBusObject()}}

1904

return fake_ravel

1905

1906

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

1907

interface, *args):

1908

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

1909

return bus.call_method(methodname, busname, objectpath,

1910

interface, *args)

1911

1912

def test_call_method_returns(self):

1913

expected_method_return = Unique()

1914

method_args = (Unique(), Unique())

1915

def func(*args):

1916

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

1917

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

1918

self.assertIs(marg, arg)

1919

return expected_method_return

1920

fake_ravel = self.fake_ravel_func(func)

1921

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1922

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

1923

"objectpath", "interface",

1924

*method_args)

1925

self.assertIs(ret, expected_method_return)

1926

1927

def test_call_method_filters_objectpath(self):

1928

def func():

1929

return method_return

1930

fake_ravel = self.fake_ravel_func(func)

1931

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1932

method_return = (self.dummy_dbussy.DBUS

1933

.ObjectPath("objectpath"))

1934

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

1935

"objectpath", "interface")

1936

self.assertEqual("objectpath", ret)

1937

self.assertNotIsInstance(ret,

1938

self.dummy_dbussy.DBUS.ObjectPath)

1939

1940

def test_call_method_filters_objectpaths_in_dict(self):

1941

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1942

def func():

1943

return method_return

1944

fake_ravel = self.fake_ravel_func(func)

1945

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1946

method_return = {

1947

ObjectPath("objectpath_key_1"):

1948

ObjectPath("objectpath_value_1"),

1949

ObjectPath("objectpath_key_2"):

1950

ObjectPath("objectpath_value_2"),

1951

}

1952

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

1953

"objectpath", "interface")

1954

expected_method_return = {str(key): str(value)

1955

for key, value in

1956

method_return.items()}

1957

for key, value in ret.items():

1958

self.assertNotIsInstance(key, ObjectPath)

1959

self.assertNotIsInstance(value, ObjectPath)

1960

self.assertEqual(expected_method_return, ret)

1961

self.assertIsInstance(ret, dict)

1962

1963

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1964

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1965

def func():

1966

return method_return

1967

fake_ravel = self.fake_ravel_func(func)

1968

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

1969

method_return = {

1970

ObjectPath("key1"): {

1971

ObjectPath("key11"): ObjectPath("value11"),

1972

ObjectPath("key12"): ObjectPath("value12"),

1973

},

1974

ObjectPath("key2"): {

1975

ObjectPath("key21"): ObjectPath("value21"),

1976

ObjectPath("key22"): ObjectPath("value22"),

1977

},

1978

}

1979

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

1980

"objectpath", "interface")

1981

expected_method_return = {

1982

"key1": {"key11": "value11",

1983

"key12": "value12"},

1984

"key2": {"key21": "value21",

1985

"key22": "value22"},

1986

}

1987

self.assertEqual(expected_method_return, ret)

1988

for key, value in ret.items():

1989

self.assertIsInstance(value, dict)

1990

self.assertEqual(expected_method_return[key], value)

1991

self.assertNotIsInstance(key, ObjectPath)

1992

for inner_key, inner_value in value.items():

1993

self.assertIsInstance(value, dict)

1994

self.assertEqual(

1995

expected_method_return[key][inner_key],

1996

inner_value)

1997

self.assertNotIsInstance(key, ObjectPath)

1998

1999

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

2000

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

2001

def func():

2002

return method_return

2003

fake_ravel = self.fake_ravel_func(func)

2004

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

2005

method_return = {

2006

ObjectPath("key1"): {

2007

ObjectPath("key2"): {

2008

ObjectPath("key3"): ObjectPath("value"),

2009

},

2010

},

2011

}

2012

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

2013

"objectpath", "interface")

2014

expected_method_return = {"key1": {"key2": {"key3": "value"}}}

2015

self.assertEqual(expected_method_return, ret)

2016

self.assertIsInstance(ret, dict)

2017

self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)

2018

self.assertIsInstance(ret["key1"], dict)

2019

self.assertNotIsInstance(next(iter(ret["key1"].keys())),

2020

ObjectPath)

2021

self.assertIsInstance(ret["key1"]["key2"], dict)

2022

self.assertNotIsInstance(

2023

next(iter(ret["key1"]["key2"].keys())),

2024

ObjectPath)

2025

self.assertEqual("value", ret["key1"]["key2"]["key3"])

2026

self.assertNotIsInstance(ret["key1"]["key2"]["key3"],

2027

self.dummy_dbussy.DBUS.ObjectPath)

2028

2029

def test_call_method_handles_exception(self):

2030

def func():

2031

raise self.dummy_dbussy.DBusError()

2032

2033

fake_ravel = self.fake_ravel_func(func)

2034

bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)

2035

2036

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

2037

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

2038

"objectpath", "interface")

2039

2040

self.assertNotIsInstance(e.exception, dbus.ConnectFailed)

2041

2042

def test_get_object_converts_to_correct_exception(self):

2043

class fake_ravel_raises_exception_on_connect:

2044

@staticmethod

2045

def system_bus():

2046

class Bus:

2047

@staticmethod

2048

def __getitem__(key):

2049

if key == "objectpath":

2050

raise self.dummy_dbussy.DBusError()

2051

raise Exception(key)

2052

return {"busname": Bus()}

2053

def func():

2054

raise self.dummy_dbussy.DBusError()

2055

bus = dbussy_adapter.SystemBus(

2056

self.dummy_dbussy,

2057

fake_ravel_raises_exception_on_connect)

2058

with self.assertRaises(dbus.ConnectFailed):

2059

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

2060

"objectpath", "interface")

2061

2062

1519

2063

class Test_commands_from_options(unittest.TestCase):

1520

2064

1521

2065

def setUp(self):

1526

2070

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

1527

2071

command.IsEnabled)

1528

2072

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

2073

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

2074

clients=None, **cmd_attrs):

1531

2075

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

1532

2076

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

1533

2077

options = self.parser.parse_args(args)

1534

2078

check_option_syntax(self.parser, options)

1535

2079

commands = commands_from_options(options)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

2080

self.assertEqual(length, len(commands))

2081

for command in commands:

2082

if isinstance(command, command_cls):

2083

break

2084

else:

2085

self.assertIsInstance(command, command_cls)

2086

if clients is not None:

2087

self.assertEqual(clients, options.client)

1539

2088

for key, value in cmd_attrs.items():

1540

2089

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

1541

2090

2091

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

2092

for cmd in commands:

2093

self.assert_command_from_args(args, cmd,

2094

length=len(commands),

2095

clients=clients)

2096

1542

2097

def test_is_enabled_short(self):

1543

2098

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

1544

2099

command.IsEnabled)

1735

2290

verbose=True)

1736

2291

1737

2292

2293

def test_manual_page_example_1(self):

2294

self.assert_command_from_args("",

2295

command.PrintTable,

2296

clients=[],

2297

verbose=False)

2298

2299

def test_manual_page_example_2(self):

2300

self.assert_command_from_args(

2301

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

2302

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

2303

"foo2.example.org"],

2304

verbose=True)

2305

2306

def test_manual_page_example_3(self):

2307

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

2308

command.Enable,

2309

clients=[])

2310

2311

def test_manual_page_example_4(self):

2312

self.assert_commands_from_args(

2313

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

2314

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

2315

[command.SetTimeout, command.SetInterval],

2316

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

2317

2318

def test_manual_page_example_5(self):

2319

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

2320

command.Approve,

2321

clients=[])

2322

2323

1738

2324

class TestCommand(unittest.TestCase):

1739

2325

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

1740

2326

1853

2439

busname = "se.recompile.Mandos"

1854

2440

client_interface = "se.recompile.Mandos.Client"

1855

2441

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

2442

self.assertTrue(self.bus.clients)

1856

2443

for clientpath in self.bus.clients:

1857

2444

self.assertIn(("Approve", busname, clientpath,

1858

2445

client_interface, (True,)), self.bus.calls)

1861

2448

busname = "se.recompile.Mandos"

1862

2449

client_interface = "se.recompile.Mandos.Client"

1863

2450

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

2451

self.assertTrue(self.bus.clients)

1864

2452

for clientpath in self.bus.clients:

1865

2453

self.assertIn(("Approve", busname, clientpath,

1866

2454

client_interface, (False,)),

1867

2455

self.bus.calls)

1868

2456

1869

2457

def test_Remove(self):

2458

busname = "se.recompile.Mandos"

2459

server_path = "/"

2460

server_interface = "se.recompile.Mandos"

2461

orig_clients = self.bus.clients.copy()

1870

2462

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

1871

for clientpath in self.bus.clients:

1872

self.assertIn(("RemoveClient", dbus_busname,

1873

dbus_server_path, dbus_server_interface,

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

1874

2467

(clientpath,)), self.bus.calls)

1875

2468

1876

2469

expected_json = {

2070

2663

def runTest(self):

2071

2664

if not hasattr(self, "command"):

2072

2665

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]

2666

2667

if hasattr(self, "values_to_set"):

2668

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

2669

values_to_get = getattr(self, "values_to_get",

2670

self.values_to_set)

2671

else:

2672

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

2673

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2675

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

2676

for clientpath in self.bus.clients:

2677

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

2678

Unique())

2679

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

2680

self.assertTrue(self.bus.clients)

2681

for clientpath in self.bus.clients:

2682

value = (self.bus.clients[clientpath]

2683

[self.propname])

2082

2684

self.assertNotIsInstance(value, Unique)

2083

2685

self.assertEqual(value_to_get, value)

2084

2686

2085

def run_command(self, value, clients):

2086

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

2087

2088

2687

2089

2688

class TestEnableCmd(TestPropertySetterCmd):

2090

2689

command = command.Enable

2091

2690

propname = "Enabled"

2092

values_to_set = [True]

2691

values_to_get = [True]

2093

2692

2094

2693

2095

2694

class TestDisableCmd(TestPropertySetterCmd):

2096

2695

command = command.Disable

2097

2696

propname = "Enabled"

2098

values_to_set = [False]

2697

values_to_get = [False]

2099

2698

2100

2699

2101

2700

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2102

2701

command = command.BumpTimeout

2103

2702

propname = "LastCheckedOK"

2104

values_to_set = [""]

2703

values_to_get = [""]

2105

2704

2106

2705

2107

2706

class TestStartCheckerCmd(TestPropertySetterCmd):

2108

2707

command = command.StartChecker

2109

2708

propname = "CheckerRunning"

2110

values_to_set = [True]

2709

values_to_get = [True]

2111

2710

2112

2711

2113

2712

class TestStopCheckerCmd(TestPropertySetterCmd):

2114

2713

command = command.StopChecker

2115

2714

propname = "CheckerRunning"

2116

values_to_set = [False]

2715

values_to_get = [False]

2117

2716

2118

2717

2119

2718

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2120

2719

command = command.ApproveByDefault

2121

2720

propname = "ApprovedByDefault"

2122

values_to_set = [True]

2721

values_to_get = [True]

2123

2722

2124

2723

2125

2724

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2126

2725

command = command.DenyByDefault

2127

2726

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

2727

values_to_get = [False]

2728

2729

2730

class TestSetCheckerCmd(TestPropertySetterCmd):

2139

2731

command = command.SetChecker

2140

2732

propname = "Checker"

2141

2733

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

2142

2734

2143

2735

2144

class TestSetHostCmd(TestPropertySetterValueCmd):

2736

class TestSetHostCmd(TestPropertySetterCmd):

2145

2737

command = command.SetHost

2146

2738

propname = "Host"

2147

2739

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

2148

2740

2149

2741

2150

class TestSetSecretCmd(TestPropertySetterValueCmd):

2742

class TestSetSecretCmd(TestPropertySetterCmd):

2151

2743

command = command.SetSecret

2152

2744

propname = "Secret"

2153

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

2154

io.BytesIO(b"secret\0xyzzy\nbar")]

2155

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

2156

2157

2158

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

2745

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

2746

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

2747

io.BytesIO(b"secret\0xyzzy\nbar")]

2748

self.values_to_get = [f.getvalue() for f in

2749

self.values_to_set]

2750

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

2751

2752

2753

class TestSetTimeoutCmd(TestPropertySetterCmd):

2159

2754

command = command.SetTimeout

2160

2755

propname = "Timeout"

2161

2756

values_to_set = [datetime.timedelta(),

2166

2761

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

2167

2762

2168

2763

2169

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

2764

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2170

2765

command = command.SetExtendedTimeout

2171

2766

propname = "ExtendedTimeout"

2172

2767

values_to_set = [datetime.timedelta(),

2177

2772

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

2178

2773

2179

2774

2180

class TestSetIntervalCmd(TestPropertySetterValueCmd):

2775

class TestSetIntervalCmd(TestPropertySetterCmd):

2181

2776

command = command.SetInterval

2182

2777

propname = "Interval"

2183

2778

values_to_set = [datetime.timedelta(),

2188

2783

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

2189

2784

2190

2785

2191

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

2786

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2192

2787

command = command.SetApprovalDelay

2193

2788

propname = "ApprovalDelay"

2194

2789

values_to_set = [datetime.timedelta(),

2199

2794

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

2200

2795

2201

2796

2202

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

2797

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2203

2798

command = command.SetApprovalDuration

2204

2799

propname = "ApprovalDuration"

2205

2800

values_to_set = [datetime.timedelta(),

2211

2806

2212

2807

2213

2808

2214

def should_only_run_tests():

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2215

2811

parser = argparse.ArgumentParser(add_help=False)

2216

parser.add_argument("--check", action='store_true')

2812

parser.add_argument("--check", action="store_true")

2813

parser.add_argument("--prefix", )

2217

2814

args, unknown_args = parser.parse_known_args()

2218

run_tests = args.check

2219

if run_tests:

2220

# Remove --check argument from sys.argv

2815

if args.check:

2816

# Remove test options from sys.argv

2221

2817

sys.argv[1:] = unknown_args

2222

return run_tests

2818

return args

2223

2819

2224

2820

# Add all tests from doctest strings

2225

2821

def load_tests(loader, tests, none):

2228

2824

return tests

2229

2825

2230

2826

if __name__ == "__main__":

2827

options = parse_test_args()

2231

2828

try:

2232

if should_only_run_tests():

2233

# Call using ./tdd-python-script --check [--verbose]

2234

unittest.main()

2829

if options.check:

2830

extra_test_prefix = options.prefix

2831

if extra_test_prefix is not None:

2832

if not (unittest.main(argv=[""], exit=False)

2833

.result.wasSuccessful()):

2834

sys.exit(1)

2835

class ExtraTestLoader(unittest.TestLoader):

2836

testMethodPrefix = extra_test_prefix

2837

# Call using ./scriptname --check [--verbose]

2838

unittest.main(argv=[""], testLoader=ExtraTestLoader())

2839

else:

2840

unittest.main(argv=[""])

2235

2841

else:

2236

2842

main()

2237

2843

finally:

2238

2844

logging.shutdown()

2845

2846

# Local Variables:

2847

# run-tests:

2848

# (lambda (&optional extra)

2849

# (if (not (funcall run-tests-in-test-buffer default-directory

2850

# extra))

2851

# (funcall show-test-buffer-in-test-window)

2852

# (funcall remove-test-window)

2853

# (if extra (message "Extra tests run successfully!"))))

2854

# run-tests-in-test-buffer:

2855

# (lambda (dir &optional extra)

2856

# (with-current-buffer (get-buffer-create "*Test*")

2857

# (setq buffer-read-only nil

2858

# default-directory dir)

2859

# (erase-buffer)

2860

# (compilation-mode))

2861

# (let ((process-result

2862

# (let ((inhibit-read-only t))

2863

# (process-file-shell-command

2864

# (funcall get-command-line extra) nil "*Test*"))))

2865

# (and (numberp process-result)

2866

# (= process-result 0))))

2867

# get-command-line:

2868

# (lambda (&optional extra)

2869

# (let ((quoted-script

2870

# (shell-quote-argument (funcall get-script-name))))

2871

# (format

2872

# (concat "%s --check" (if extra " --prefix=atest" ""))

2873

# quoted-script)))

2874

# get-script-name:

2875

# (lambda ()

2876

# (if (fboundp 'file-local-name)

2877

# (file-local-name (buffer-file-name))

2878

# (or (file-remote-p (buffer-file-name) 'localname)

2879

# (buffer-file-name))))

2880

# remove-test-window:

2881

# (lambda ()

2882

# (let ((test-window (get-buffer-window "*Test*")))

2883

# (if test-window (delete-window test-window))))

2884

# show-test-buffer-in-test-window:

2885

# (lambda ()

2886

# (when (not (get-buffer-window-list "*Test*"))

2887

# (setq next-error-last-buffer (get-buffer "*Test*"))

2888

# (let* ((side (if (>= (window-width) 146) 'right 'bottom))

2889

# (display-buffer-overriding-action

2890

# `((display-buffer-in-side-window) (side . ,side)

2891

# (window-height . fit-window-to-buffer)

2892

# (window-width . fit-window-to-buffer))))

2893

# (display-buffer "*Test*"))))

2894

# eval:

2895

# (progn

2896

# (let* ((run-extra-tests (lambda () (interactive)

2897

# (funcall run-tests t)))

2898

# (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t

2899

# (outer-keymap `(keymap (3 . ,inner-keymap)))) ; C-c

2900

# (setq minor-mode-overriding-map-alist

2901

# (cons `(run-tests . ,outer-keymap)

2902

# minor-mode-overriding-map-alist)))

2903

# (add-hook 'after-save-hook run-tests 90 t))

2904

# End: