/mandos/trunk : revision 1091

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

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

8

#

9

# This file is part of Mandos.

10

#

23

#

24

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

25

#

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

32

33

pass

33

34

35

import sys

35

import unittest

36

import argparse

37

import logging

38

import os

39

37

import locale

40

38

import datetime

41

39

import re

40

import os

42

41

import collections

43

42

import json

43

import unittest

44

import logging

44

45

import io

45

46

import tempfile

46

47

import contextlib

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

48

import abc

49

50

import dbus as dbus_python

74

51

75

52

# Show warnings by default

76

53

if not sys.warnoptions:

77

54

import warnings

78

55

warnings.simplefilter("default")

79

56

80

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

57

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

81

58

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

82

59

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

83

60

84

61

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

85

62

86

63

if sys.version_info.major == 2:

64

str = unicode

87

65

import StringIO

88

66

io.StringIO = StringIO.StringIO

89

67

90

68

locale.setlocale(locale.LC_ALL, "")

91

69

92

version = "1.8.16"

70

version = "1.8.3"

93

71

94

72

95

73

def main():

102

80

clientnames = options.client

103

81

104

82

if options.debug:

105

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

83

log.setLevel(logging.DEBUG)

106

84

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)

85

bus = dbus_python_adapter.CachingBus(dbus_python)

113

86

114

87

try:

115

88

all_clients = bus.get_clients_and_properties()

149

122

help="Select all clients")

150

123

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

151

124

help="Print all fields")

152

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

153

action="append_const", default=[],

154

const=command.DumpJSON(),

125

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

155

126

help="Dump client data in JSON format")

156

127

enable_disable = parser.add_mutually_exclusive_group()

157

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

158

action="append_const", default=[],

159

const=command.Enable(),

128

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

160

129

help="Enable client")

161

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

162

action="append_const", default=[],

163

const=command.Disable(),

130

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

131

action="store_true",

164

132

help="disable client")

165

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

166

action="append_const", default=[],

167

const=command.BumpTimeout(),

133

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

168

134

help="Bump timeout for client")

169

135

start_stop_checker = parser.add_mutually_exclusive_group()

170

136

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

171

dest="commands",

172

action="append_const", default=[],

173

const=command.StartChecker(),

137

action="store_true",

174

138

help="Start checker for client")

175

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

176

action="append_const", default=[],

177

const=command.StopChecker(),

139

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

140

action="store_true",

178

141

help="Stop checker for client")

179

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

180

action="append_const", default=[],

181

const=command.IsEnabled(),

142

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

182

143

help="Check if client is enabled")

183

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

184

action="append_const", default=[],

185

const=command.Remove(),

144

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

186

145

help="Remove client")

187

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

188

action="append", default=[],

189

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

146

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

190

147

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

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

206

154

approve_deny_default = parser.add_mutually_exclusive_group()

207

155

approve_deny_default.add_argument(

208

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

209

action="append_const", default=[],

210

const=command.ApproveByDefault(),

156

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

157

default=None, dest="approved_by_default",

211

158

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

212

159

approve_deny_default.add_argument(

213

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

214

action="append_const", default=[],

215

const=command.DenyByDefault(),

160

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

161

dest="approved_by_default",

216

162

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

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

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

236

171

approve_deny = parser.add_mutually_exclusive_group()

237

172

approve_deny.add_argument(

238

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

239

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

173

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

240

174

help="Approve any current client request")

241

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

242

action="append_const", default=[],

243

const=command.Deny(),

175

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

244

176

help="Deny any current client request")

245

177

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

246

178

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

256

188

return rfc3339_duration_to_delta(interval)

257

189

except ValueError as e:

258

190

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

259

" ".join(e.args))

191

' '.join(e.args))

260

192

return parse_pre_1_6_1_interval(interval)

261

193

262

194

263

195

def rfc3339_duration_to_delta(duration):

264

196

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

265

197

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

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)

283

214

>>> # Can not be empty:

284

215

>>> rfc3339_duration_to_delta("")

285

216

Traceback (most recent call last):

395

326

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

396

327

and return a datetime.timedelta

397

328

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

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)

412

343

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

413

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

414

True

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

415

346

416

347

"""

417

348

438

369

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

439

370

argparse"""

440

371

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)

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

464

391

465

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

392

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

466

393

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

467

if options.verbose and has_commands(options):

394

if options.verbose and has_actions(options):

468

395

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

469

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

470

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

471

398

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

472

if options.all and not has_commands(options):

399

if options.all and not has_actions(options):

473

400

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

474

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

475

and len(options.client) > 1):

401

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

476

402

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

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:

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

486

414

487

415

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

416

def get_managed_objects(self, busname, objectpath):

497

425

self.properties_iface, interface, key,

498

426

value)

499

427

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

428

505

429

class MandosBus(SystemBus):

506

430

busname_domain = "se.recompile"

538

462

pass

539

463

540

464

541

class dbus_python_adapter:

465

class dbus_python_adapter(object):

542

466

543

467

class SystemBus(dbus.MandosBus):

544

468

"""Use dbus-python"""

589

513

for key, subval in value.items()}

590

514

return value

591

515

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)

600

516

601

class SilenceLogger:

517

class SilenceLogger(object):

602

518

"Simple context manager to silence a particular logger"

603

519

def __init__(self, loggername):

604

520

self.logger = logging.getLogger(loggername)

633

549

return new_object

634

550

635

551

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

789

552

def commands_from_options(options):

790

553

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)

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

809

589

else:

810

cleaned_after.append(cmd)

811

if cleaned_after != after_remove:

812

commands = before_remove + cleaned_after

590

commands.append(command.DenyByDefault())

591

592

if options.checker is not None:

593

commands.append(command.SetChecker(options.checker))

594

595

if options.host is not None:

596

commands.append(command.SetHost(options.host))

597

598

if options.secret is not None:

599

commands.append(command.SetSecret(options.secret))

600

601

if options.timeout is not None:

602

commands.append(command.SetTimeout(options.timeout))

603

604

if options.extended_timeout:

605

commands.append(

606

command.SetExtendedTimeout(options.extended_timeout))

607

608

if options.interval is not None:

609

commands.append(command.SetInterval(options.interval))

610

611

if options.approval_delay is not None:

612

commands.append(

613

command.SetApprovalDelay(options.approval_delay))

614

615

if options.approval_duration is not None:

616

commands.append(

617

command.SetApprovalDuration(options.approval_duration))

813

618

814

619

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

815

620

# optionally verbosely

819

624

return commands

820

625

821

626

822

class command:

627

class command(object):

823

628

"""A namespace for command classes"""

824

629

825

class Base:

630

class Base(object):

826

631

"""Abstract base class for commands"""

827

632

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

828

633

"""Normal commands should implement run_on_one_client(),

876

681

{key: properties[key]

877

682

for key in self.all_keywords}

878

683

for properties in clients.values()}

879

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

684

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

880

685

881

686

882

687

class PrintTable(Output):

891

696

keywords = self.all_keywords

892

697

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

893

698

894

class TableOfClients:

699

class TableOfClients(object):

895

700

tableheaders = {

896

701

"Name": "Name",

897

702

"Enabled": "Enabled",

1030

835

def __init__(self, value):

1031

836

self.value_to_set = value

1032

837

1033

@classmethod

1034

def argparse(cls, argtype):

1035

def cmdtype(arg):

1036

return cls(argtype(arg))

1037

return cmdtype

1038

838

1039

839

class SetChecker(PropertySetterValue):

1040

840

propname = "Checker"

1128

928

"output"))

1129

929

1130

930

1131

class Unique:

931

class Unique(object):

1132

932

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

1133

933

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

1134

934

1166

966

1167

967

def test_actions_requires_client_or_all(self):

1168

968

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

1169

args = self.actionargs(action, value)

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1170

971

with self.assertParseError():

1171

self.parse_args(args)

972

self.check_option_syntax(options)

1172

973

1173

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

974

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

1174

975

# check_option_syntax()

1175

976

actions = {

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,

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,

1195

998

}

1196

999

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

1204

1000

@contextlib.contextmanager

1205

1001

def assertParseError(self):

1206

1002

with self.assertRaises(SystemExit) as e:

1211

1007

# /argparse.html#exiting-methods

1212

1008

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

1213

1009

1214

def parse_args(self, args):

1215

options = self.parser.parse_args(args)

1216

check_option_syntax(self.parser, options)

1217

1218

1010

@staticmethod

1219

1011

@contextlib.contextmanager

1220

1012

def redirect_stderr_to_devnull():

1231

1023

1232

1024

def test_actions_all_conflicts_with_verbose(self):

1233

1025

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

1234

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

1235

"--verbose")

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1236

1030

with self.assertParseError():

1237

self.parse_args(args)

1031

self.check_option_syntax(options)

1238

1032

1239

1033

def test_actions_with_client_conflicts_with_verbose(self):

1240

1034

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

1241

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

1242

"client")

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1243

1039

with self.assertParseError():

1244

self.parse_args(args)

1040

self.check_option_syntax(options)

1245

1041

1246

1042

def test_dump_json_conflicts_with_verbose(self):

1247

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

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1248

1046

with self.assertParseError():

1249

self.parse_args(args)

1047

self.check_option_syntax(options)

1250

1048

1251

1049

def test_dump_json_conflicts_with_action(self):

1252

1050

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

1253

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1254

1054

with self.assertParseError():

1255

self.parse_args(args)

1055

self.check_option_syntax(options)

1256

1056

1257

1057

def test_all_can_not_be_alone(self):

1258

args = ["--all"]

1058

options = self.parser.parse_args()

1059

options.all = True

1259

1060

with self.assertParseError():

1260

self.parse_args(args)

1061

self.check_option_syntax(options)

1261

1062

1262

1063

def test_all_is_ok_with_any_action(self):

1263

1064

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

1264

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

1265

self.parse_args(args)

1065

options = self.parser.parse_args()

1066

setattr(options, action, value)

1067

options.all = True

1068

self.check_option_syntax(options)

1266

1069

1267

1070

def test_any_action_is_ok_with_one_client(self):

1268

1071

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

1269

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

1270

self.parse_args(args)

1072

options = self.parser.parse_args()

1073

setattr(options, action, value)

1074

options.client = ["client"]

1075

self.check_option_syntax(options)

1271

1076

1272

1077

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1273

1079

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

1274

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

1080

if action == "is_enabled":

1275

1081

continue

1276

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

1277

self.parse_args(args)

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1278

1085

1279

1086

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1280

1088

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

1281

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

1089

if action == "is_enabled":

1282

1090

continue

1283

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

1284

"client2")

1285

self.parse_args(args)

1091

setattr(options, action, value)

1092

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

1093

self.check_option_syntax(options)

1286

1094

1287

1095

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1288

1096

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

1289

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

1097

if action == "is_enabled":

1290

1098

continue

1291

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

1292

"client2")

1293

self.parse_args(args)

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

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

1102

self.check_option_syntax(options)

1294

1103

1295

1104

def test_is_enabled_fails_without_client(self):

1296

args = ["--is-enabled"]

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1297

1107

with self.assertParseError():

1298

self.parse_args(args)

1108

self.check_option_syntax(options)

1299

1109

1300

1110

def test_is_enabled_fails_with_two_clients(self):

1301

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

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

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

1302

1114

with self.assertParseError():

1303

self.parse_args(args)

1115

self.check_option_syntax(options)

1304

1116

1305

1117

def test_remove_can_only_be_combined_with_action_deny(self):

1306

1118

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

1307

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

1119

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

1308

1120

continue

1309

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

1310

"--remove")

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1311

1125

with self.assertParseError():

1312

self.parse_args(args)

1126

self.check_option_syntax(options)

1313

1127

1314

1128

1315

1129

class Test_dbus_exceptions(unittest.TestCase):

1418

1232

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1419

1233

1420

1234

def MockDBusPython_func(self, func):

1421

class mock_dbus_python:

1235

class mock_dbus_python(object):

1422

1236

"""mock dbus-python module"""

1423

class exceptions:

1237

class exceptions(object):

1424

1238

"""Pseudo-namespace"""

1425

1239

class DBusException(Exception):

1426

1240

pass

1427

class SystemBus:

1241

class SystemBus(object):

1428

1242

@staticmethod

1429

1243

def get_object(busname, objectpath):

1430

1244

DBusObject = collections.namedtuple(

1431

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

1245

"DBusObject", ("methodname",))

1432

1246

def method(*args, **kwargs):

1433

1247

self.assertEqual({"dbus_interface":

1434

1248

"interface"},

1435

1249

kwargs)

1436

1250

return func(*args)

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:

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

1448

1253

def __init__(self, value):

1449

1254

self.value = bool(value)

1450

1255

def __bool__(self):

1455

1260

pass

1456

1261

class Dictionary(dict):

1457

1262

pass

1458

class ByteArray(bytes):

1459

pass

1460

1263

return mock_dbus_python

1461

1264

1462

1265

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

1634

1437

finally:

1635

1438

dbus_logger.removeFilter(counting_handler)

1636

1439

1637

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

1440

self.assertNotIsInstance(e, dbus.ConnectFailed)

1638

1441

1639

1442

# Make sure the dbus logger was suppressed

1640

1443

self.assertEqual(0, counting_handler.count)

1641

1444

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

1658

1445

def test_get_object_converts_to_correct_exception(self):

1659

1446

bus = dbus_python_adapter.SystemBus(

1660

1447

self.fake_dbus_python_raises_exception_on_connect)

1662

1449

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

1663

1450

"objectpath", "interface")

1664

1451

1665

class fake_dbus_python_raises_exception_on_connect:

1452

class fake_dbus_python_raises_exception_on_connect(object):

1666

1453

"""fake dbus-python module"""

1667

class exceptions:

1454

class exceptions(object):

1668

1455

"""Pseudo-namespace"""

1669

1456

class DBusException(Exception):

1670

1457

pass

1678

1465

1679

1466

1680

1467

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1681

class mock_dbus_python:

1468

class mock_dbus_python(object):

1682

1469

"""mock dbus-python modules"""

1683

class SystemBus:

1470

class SystemBus(object):

1684

1471

@staticmethod

1685

1472

def get_object(busname, objectpath):

1686

1473

return Unique()

1729

1516

self.assertIs(obj1, obj1b)

1730

1517

1731

1518

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

2063

1519

class Test_commands_from_options(unittest.TestCase):

2064

1520

2065

1521

def setUp(self):

2070

1526

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

2071

1527

command.IsEnabled)

2072

1528

2073

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

2074

clients=None, **cmd_attrs):

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

2075

1531

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

2076

1532

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

2077

1533

options = self.parser.parse_args(args)

2078

1534

check_option_syntax(self.parser, options)

2079

1535

commands = commands_from_options(options)

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)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

2088

1539

for key, value in cmd_attrs.items():

2089

1540

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

2090

1541

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

2097

1542

def test_is_enabled_short(self):

2098

1543

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

2099

1544

command.IsEnabled)

2290

1735

verbose=True)

2291

1736

2292

1737

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

2324

1738

class TestCommand(unittest.TestCase):

2325

1739

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

2326

1740

2439

1853

busname = "se.recompile.Mandos"

2440

1854

client_interface = "se.recompile.Mandos.Client"

2441

1855

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

2442

self.assertTrue(self.bus.clients)

2443

1856

for clientpath in self.bus.clients:

2444

1857

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

2445

1858

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

2448

1861

busname = "se.recompile.Mandos"

2449

1862

client_interface = "se.recompile.Mandos.Client"

2450

1863

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

2451

self.assertTrue(self.bus.clients)

2452

1864

for clientpath in self.bus.clients:

2453

1865

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

2454

1866

client_interface, (False,)),

2455

1867

self.bus.calls)

2456

1868

2457

1869

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

2462

1870

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

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

1871

for clientpath in self.bus.clients:

1872

self.assertIn(("RemoveClient", dbus_busname,

1873

dbus_server_path, dbus_server_interface,

2467

1874

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

2468

1875

2469

1876

expected_json = {

2663

2070

def runTest(self):

2664

2071

if not hasattr(self, "command"):

2665

2072

return # Abstract TestCase class

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

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]

2684

2082

self.assertNotIsInstance(value, Unique)

2685

2083

self.assertEqual(value_to_get, value)

2686

2084

2085

def run_command(self, value, clients):

2086

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

2087

2687

2088

2688

2089

class TestEnableCmd(TestPropertySetterCmd):

2689

2090

command = command.Enable

2690

2091

propname = "Enabled"

2691

values_to_get = [True]

2092

values_to_set = [True]

2692

2093

2693

2094

2694

2095

class TestDisableCmd(TestPropertySetterCmd):

2695

2096

command = command.Disable

2696

2097

propname = "Enabled"

2697

values_to_get = [False]

2098

values_to_set = [False]

2698

2099

2699

2100

2700

2101

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2701

2102

command = command.BumpTimeout

2702

2103

propname = "LastCheckedOK"

2703

values_to_get = [""]

2104

values_to_set = [""]

2704

2105

2705

2106

2706

2107

class TestStartCheckerCmd(TestPropertySetterCmd):

2707

2108

command = command.StartChecker

2708

2109

propname = "CheckerRunning"

2709

values_to_get = [True]

2110

values_to_set = [True]

2710

2111

2711

2112

2712

2113

class TestStopCheckerCmd(TestPropertySetterCmd):

2713

2114

command = command.StopChecker

2714

2115

propname = "CheckerRunning"

2715

values_to_get = [False]

2116

values_to_set = [False]

2716

2117

2717

2118

2718

2119

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2719

2120

command = command.ApproveByDefault

2720

2121

propname = "ApprovedByDefault"

2721

values_to_get = [True]

2122

values_to_set = [True]

2722

2123

2723

2124

2724

2125

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2725

2126

command = command.DenyByDefault

2726

2127

propname = "ApprovedByDefault"

2727

values_to_get = [False]

2728

2729

2730

class TestSetCheckerCmd(TestPropertySetterCmd):

2128

values_to_set = [False]

2129

2130

2131

class TestPropertySetterValueCmd(TestPropertySetterCmd):

2132

"""Abstract class for tests of PropertySetterValueCmd classes"""

2133

2134

def run_command(self, value, clients):

2135

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

2136

2137

2138

class TestSetCheckerCmd(TestPropertySetterValueCmd):

2731

2139

command = command.SetChecker

2732

2140

propname = "Checker"

2733

2141

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

2734

2142

2735

2143

2736

class TestSetHostCmd(TestPropertySetterCmd):

2144

class TestSetHostCmd(TestPropertySetterValueCmd):

2737

2145

command = command.SetHost

2738

2146

propname = "Host"

2739

2147

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

2740

2148

2741

2149

2742

class TestSetSecretCmd(TestPropertySetterCmd):

2150

class TestSetSecretCmd(TestPropertySetterValueCmd):

2743

2151

command = command.SetSecret

2744

2152

propname = "Secret"

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

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

2754

2159

command = command.SetTimeout

2755

2160

propname = "Timeout"

2756

2161

values_to_set = [datetime.timedelta(),

2761

2166

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

2762

2167

2763

2168

2764

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2169

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

2765

2170

command = command.SetExtendedTimeout

2766

2171

propname = "ExtendedTimeout"

2767

2172

values_to_set = [datetime.timedelta(),

2772

2177

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

2773

2178

2774

2179

2775

class TestSetIntervalCmd(TestPropertySetterCmd):

2180

class TestSetIntervalCmd(TestPropertySetterValueCmd):

2776

2181

command = command.SetInterval

2777

2182

propname = "Interval"

2778

2183

values_to_set = [datetime.timedelta(),

2783

2188

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

2784

2189

2785

2190

2786

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2191

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

2787

2192

command = command.SetApprovalDelay

2788

2193

propname = "ApprovalDelay"

2789

2194

values_to_set = [datetime.timedelta(),

2794

2199

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

2795

2200

2796

2201

2797

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2202

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

2798

2203

command = command.SetApprovalDuration

2799

2204

propname = "ApprovalDuration"

2800

2205

values_to_set = [datetime.timedelta(),

2806

2211

2807

2212

2808

2213

2809

def parse_test_args():

2810

# type: () -> argparse.Namespace

2214

def should_only_run_tests():

2811

2215

parser = argparse.ArgumentParser(add_help=False)

2812

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

2813

parser.add_argument("--prefix", )

2216

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

2814

2217

args, unknown_args = parser.parse_known_args()

2815

if args.check:

2816

# Remove test options from sys.argv

2218

run_tests = args.check

2219

if run_tests:

2220

# Remove --check argument from sys.argv

2817

2221

sys.argv[1:] = unknown_args

2818

return args

2222

return run_tests

2819

2223

2820

2224

# Add all tests from doctest strings

2821

2225

def load_tests(loader, tests, none):

2824

2228

return tests

2825

2229

2826

2230

if __name__ == "__main__":

2827

options = parse_test_args()

2828

2231

try:

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=[""])

2232

if should_only_run_tests():

2233

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

2234

unittest.main()

2841

2235

else:

2842

2236

main()

2843

2237

finally:

2844

2238

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: