/mandos/trunk : revision 1093

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

54

class gi:

55

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

56

class repository:

57

class GLib:

58

class Error(Exception):

59

pass

60

61

62

dbussy = None

63

ravel = None

64

dbus_python = None

65

pydbus = None

66

67

try:

68

import dbussy

69

import ravel

70

except ImportError:

71

try:

72

import pydbus

73

import gi

74

except ImportError:

75

import dbus as dbus_python

76

48

import abc

49

50

import dbus as dbus_python

77

51

78

52

# Show warnings by default

79

53

if not sys.warnoptions:

80

54

import warnings

81

55

warnings.simplefilter("default")

82

56

83

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

84

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

57

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

58

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

85

59

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

86

60

87

61

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

88

62

89

63

if sys.version_info.major == 2:

64

str = unicode

90

65

import StringIO

91

66

io.StringIO = StringIO.StringIO

92

67

93

68

locale.setlocale(locale.LC_ALL, "")

94

69

95

version = "1.8.16"

70

version = "1.8.3"

96

71

97

72

98

73

def main():

105

80

clientnames = options.client

106

81

107

82

if options.debug:

108

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

83

log.setLevel(logging.DEBUG)

109

84

110

if dbussy is not None and ravel is not None:

111

bus = dbussy_adapter.CachingBus(dbussy, ravel)

112

elif pydbus is not None:

113

bus = pydbus_adapter.CachingBus(pydbus)

114

else:

115

bus = dbus_python_adapter.CachingBus(dbus_python)

85

bus = dbus_python_adapter.CachingBus(dbus_python)

116

86

117

87

try:

118

88

all_clients = bus.get_clients_and_properties()

152

122

help="Select all clients")

153

123

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

154

124

help="Print all fields")

155

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

156

action="append_const", default=[],

157

const=command.DumpJSON(),

125

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

158

126

help="Dump client data in JSON format")

159

127

enable_disable = parser.add_mutually_exclusive_group()

160

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

161

action="append_const", default=[],

162

const=command.Enable(),

128

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

163

129

help="Enable client")

164

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

165

action="append_const", default=[],

166

const=command.Disable(),

130

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

131

action="store_true",

167

132

help="disable client")

168

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

169

action="append_const", default=[],

170

const=command.BumpTimeout(),

133

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

171

134

help="Bump timeout for client")

172

135

start_stop_checker = parser.add_mutually_exclusive_group()

173

136

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

174

dest="commands",

175

action="append_const", default=[],

176

const=command.StartChecker(),

137

action="store_true",

177

138

help="Start checker for client")

178

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

179

action="append_const", default=[],

180

const=command.StopChecker(),

139

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

140

action="store_true",

181

141

help="Stop checker for client")

182

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

183

action="append_const", default=[],

184

const=command.IsEnabled(),

142

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

185

143

help="Check if client is enabled")

186

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

187

action="append_const", default=[],

188

const=command.Remove(),

144

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

189

145

help="Remove client")

190

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

191

action="append", default=[],

192

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

146

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

193

147

help="Set checker command for client")

194

parser.add_argument(

195

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

196

default=[], metavar="TIME",

197

type=command.SetTimeout.argparse(string_to_delta),

198

help="Set timeout for client")

199

parser.add_argument(

200

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

201

default=[], metavar="TIME",

202

type=command.SetExtendedTimeout.argparse(string_to_delta),

203

help="Set extended timeout for client")

204

parser.add_argument(

205

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

206

default=[], metavar="TIME",

207

type=command.SetInterval.argparse(string_to_delta),

208

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

209

154

approve_deny_default = parser.add_mutually_exclusive_group()

210

155

approve_deny_default.add_argument(

211

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

212

action="append_const", default=[],

213

const=command.ApproveByDefault(),

156

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

157

default=None, dest="approved_by_default",

214

158

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

215

159

approve_deny_default.add_argument(

216

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

217

action="append_const", default=[],

218

const=command.DenyByDefault(),

160

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

161

dest="approved_by_default",

219

162

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

220

parser.add_argument(

221

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

222

default=[], metavar="TIME",

223

type=command.SetApprovalDelay.argparse(string_to_delta),

224

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

225

parser.add_argument(

226

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

227

default=[], metavar="TIME",

228

type=command.SetApprovalDuration.argparse(string_to_delta),

229

help="Set duration of one client approval")

230

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

231

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

232

type=command.SetHost,

233

help="Set host for client")

234

parser.add_argument(

235

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

236

default=[], metavar="FILENAME",

237

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

238

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

239

171

approve_deny = parser.add_mutually_exclusive_group()

240

172

approve_deny.add_argument(

241

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

242

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

173

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

243

174

help="Approve any current client request")

244

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

245

action="append_const", default=[],

246

const=command.Deny(),

175

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

247

176

help="Deny any current client request")

248

177

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

249

178

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

259

188

return rfc3339_duration_to_delta(interval)

260

189

except ValueError as e:

261

190

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

262

" ".join(e.args))

191

' '.join(e.args))

263

192

return parse_pre_1_6_1_interval(interval)

264

193

265

194

266

195

def rfc3339_duration_to_delta(duration):

267

196

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

268

197

269

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

270

True

271

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

272

True

273

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

274

True

275

>>> # 60 months

276

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

277

True

278

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

279

True

280

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

281

True

282

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

283

True

284

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

285

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)

286

214

>>> # Can not be empty:

287

215

>>> rfc3339_duration_to_delta("")

288

216

Traceback (most recent call last):

395

323

396

324

397

325

def parse_pre_1_6_1_interval(interval):

398

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

326

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

399

327

and return a datetime.timedelta

400

328

401

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

402

True

403

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

404

True

405

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

406

True

407

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

408

True

409

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

410

True

411

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

412

True

413

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

414

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)

415

343

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

416

>>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \

417

... == datetime.timedelta(2, 480, 18000)

418

True

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

419

346

420

347

"""

421

348

442

369

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

443

370

argparse"""

444

371

445

def has_commands(options, commands=None):

446

if commands is None:

447

commands = (command.Enable,

448

command.Disable,

449

command.BumpTimeout,

450

command.StartChecker,

451

command.StopChecker,

452

command.IsEnabled,

453

command.Remove,

454

command.SetChecker,

455

command.SetTimeout,

456

command.SetExtendedTimeout,

457

command.SetInterval,

458

command.ApproveByDefault,

459

command.DenyByDefault,

460

command.SetApprovalDelay,

461

command.SetApprovalDuration,

462

command.SetHost,

463

command.SetSecret,

464

command.Approve,

465

command.Deny)

466

return any(isinstance(cmd, commands)

467

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

468

391

469

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

392

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

470

393

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

471

if options.verbose and has_commands(options):

394

if options.verbose and has_actions(options):

472

395

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

473

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

474

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

475

398

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

476

if options.all and not has_commands(options):

399

if options.all and not has_actions(options):

477

400

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

478

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

479

and len(options.client) > 1):

401

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

480

402

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

481

if (len(options.commands) > 1

482

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

483

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

484

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

485

486

487

class dbus:

488

489

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

490

414

491

415

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

492

493

416

def get_managed_objects(self, busname, objectpath):

494

417

return self.call_method("GetManagedObjects", busname,

495

418

objectpath,

496

419

self.object_manager_iface)

497

420

498

421

properties_iface = "org.freedesktop.DBus.Properties"

499

500

422

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

501

423

value):

502

424

self.call_method("Set", busname, objectpath,

503

425

self.properties_iface, interface, key,

504

426

value)

505

427

506

def call_method(self, methodname, busname, objectpath,

507

interface, *args):

508

raise NotImplementedError()

509

428

510

429

class MandosBus(SystemBus):

511

430

busname_domain = "se.recompile"

543

462

pass

544

463

545

464

546

class dbus_python_adapter:

465

class dbus_python_adapter(object):

547

466

548

467

class SystemBus(dbus.MandosBus):

549

468

"""Use dbus-python"""

594

513

for key, subval in value.items()}

595

514

return value

596

515

597

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

598

if key == "Secret":

599

if not isinstance(value, bytes):

600

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

601

value = self.dbus_python.ByteArray(value)

602

return self.set_property(self.busname, objectpath,

603

self.client_interface, key,

604

value)

605

516

606

class SilenceLogger:

517

class SilenceLogger(object):

607

518

"Simple context manager to silence a particular logger"

608

609

519

def __init__(self, loggername):

610

520

self.logger = logging.getLogger(loggername)

611

521

621

531

def __exit__(self, exc_type, exc_val, exc_tb):

622

532

self.logger.removeFilter(self.nullfilter)

623

533

534

624

535

class CachingBus(SystemBus):

625

536

"""A caching layer for dbus_python_adapter.SystemBus"""

626

627

537

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

628

538

self.object_cache = {}

629

539

super(dbus_python_adapter.CachingBus,

630

540

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

631

632

541

def get_object(self, busname, objectpath):

633

542

try:

634

543

return self.object_cache[(busname, objectpath)]

636

545

new_object = super(

637

546

dbus_python_adapter.CachingBus,

638

547

self).get_object(busname, objectpath)

639

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

640

return new_object

641

642

643

class pydbus_adapter:

644

class SystemBus(dbus.MandosBus):

645

def __init__(self, module=pydbus):

646

self.pydbus = module

647

self.bus = self.pydbus.SystemBus()

648

649

@contextlib.contextmanager

650

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

651

try:

652

yield

653

except gi.repository.GLib.Error as e:

654

# This does what "raise from" would do

655

exc = exception_class(*e.args)

656

exc.__cause__ = e

657

raise exc

658

659

def call_method(self, methodname, busname, objectpath,

660

interface, *args):

661

proxy_object = self.get(busname, objectpath)

662

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

663

interface, methodname,

664

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

665

method = getattr(proxy_object[interface], methodname)

666

with self.convert_exception():

667

return method(*args)

668

669

def get(self, busname, objectpath):

670

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

671

busname, objectpath)

672

with self.convert_exception(dbus.ConnectFailed):

673

if sys.version_info.major <= 2:

674

with warnings.catch_warnings():

675

warnings.filterwarnings(

676

"ignore", "", DeprecationWarning,

677

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

678

return self.bus.get(busname, objectpath)

679

else:

680

return self.bus.get(busname, objectpath)

681

682

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

683

value):

684

proxy_object = self.get(busname, objectpath)

685

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

686

objectpath, self.properties_iface, interface,

687

key, value)

688

setattr(proxy_object[interface], key, value)

689

690

class CachingBus(SystemBus):

691

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

692

693

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

694

self.object_cache = {}

695

super(pydbus_adapter.CachingBus,

696

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

697

698

def get(self, busname, objectpath):

699

try:

700

return self.object_cache[(busname, objectpath)]

701

except KeyError:

702

new_object = (super(pydbus_adapter.CachingBus, self)

703

.get(busname, objectpath))

704

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

705

return new_object

706

707

708

class dbussy_adapter:

709

class SystemBus(dbus.SystemBus):

710

"""Use DBussy"""

711

712

def __init__(self, dbussy, ravel):

713

self.dbussy = dbussy

714

self.ravel = ravel

715

self.bus = ravel.system_bus()

716

717

@contextlib.contextmanager

718

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

719

try:

720

yield

721

except self.dbussy.DBusError as e:

722

# This does what "raise from" would do

723

exc = exception_class(*e.args)

724

exc.__cause__ = e

725

raise exc

726

727

def call_method(self, methodname, busname, objectpath,

728

interface, *args):

729

proxy_object = self.get_object(busname, objectpath)

730

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

731

interface, methodname,

732

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

733

iface = proxy_object.get_interface(interface)

734

method = getattr(iface, methodname)

735

with self.convert_exception(dbus.Error):

736

value = method(*args)

737

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

738

# list of one element with the return value

739

if value:

740

return self.type_filter(value[0])

741

742

def get_object(self, busname, objectpath):

743

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

744

busname, objectpath)

745

with self.convert_exception(dbus.ConnectFailed):

746

return self.bus[busname][objectpath]

747

748

def type_filter(self, value):

749

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

750

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

751

# Tuple[dbussy.DBUS.Signature, Any]

752

if isinstance(value, tuple):

753

if (len(value) == 2

754

and isinstance(value[0],

755

self.dbussy.DBUS.Signature)):

756

return self.type_filter(value[1])

757

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

758

return str(value)

759

# Also recurse into dictionaries

760

elif isinstance(value, dict):

761

return {self.type_filter(key):

762

self.type_filter(subval)

763

for key, subval in value.items()}

764

return value

765

766

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

767

value):

768

proxy_object = self.get_object(busname, objectpath)

769

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

770

objectpath, self.properties_iface, interface,

771

key, value)

772

if key == "Secret":

773

# DBussy wants a Byte Array to be a sequence of

774

# values, not a byte string

775

value = tuple(value)

776

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

777

778

class MandosBus(SystemBus, dbus.MandosBus):

779

pass

780

781

class CachingBus(MandosBus):

782

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

783

784

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

785

self.object_cache = {}

786

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

787

**kwargs)

788

789

def get_object(self, busname, objectpath):

790

try:

791

return self.object_cache[(busname, objectpath)]

792

except KeyError:

793

new_object = super(

794

dbussy_adapter.CachingBus,

795

self).get_object(busname, objectpath)

796

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

548

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

797

549

return new_object

798

550

799

551

800

552

def commands_from_options(options):

801

553

802

commands = list(options.commands)

803

804

def find_cmd(cmd, commands):

805

i = 0

806

for i, c in enumerate(commands):

807

if isinstance(c, cmd):

808

return i

809

return i+1

810

811

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

812

# to occur ahead of command.Remove.

813

index_of_remove = find_cmd(command.Remove, commands)

814

before_remove = commands[:index_of_remove]

815

after_remove = commands[index_of_remove:]

816

cleaned_after = []

817

for cmd in after_remove:

818

if isinstance(cmd, command.Deny):

819

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

820

589

else:

821

cleaned_after.append(cmd)

822

if cleaned_after != after_remove:

823

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

824

618

825

619

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

826

620

# optionally verbosely

830

624

return commands

831

625

832

626

833

class command:

627

class command(object):

834

628

"""A namespace for command classes"""

835

629

836

class Base:

630

class Base(object):

837

631

"""Abstract base class for commands"""

838

839

632

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

840

633

"""Normal commands should implement run_on_one_client(),

841

634

but commands which want to operate on all clients at the same time can

845

638

for client, properties in clients.items():

846

639

self.run_on_one_client(client, properties)

847

640

641

848

642

class IsEnabled(Base):

849

643

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

850

644

properties = next(iter(clients.values()))

852

646

sys.exit(0)

853

647

sys.exit(1)

854

648

649

855

650

class Approve(Base):

856

651

def run_on_one_client(self, client, properties):

857

652

self.bus.call_client_method(client, "Approve", True)

858

653

654

859

655

class Deny(Base):

860

656

def run_on_one_client(self, client, properties):

861

657

self.bus.call_client_method(client, "Approve", False)

862

658

659

863

660

class Remove(Base):

864

661

def run(self, clients, bus):

865

662

for clientpath in frozenset(clients.keys()):

866

663

bus.call_server_method("RemoveClient", clientpath)

867

664

665

868

666

class Output(Base):

869

667

"""Abstract class for commands outputting client details"""

870

668

all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",

876

674

"Checker", "ExtendedTimeout", "Expires",

877

675

"LastCheckerStatus")

878

676

677

879

678

class DumpJSON(Output):

880

679

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

881

680

data = {properties["Name"]:

882

681

{key: properties[key]

883

682

for key in self.all_keywords}

884

683

for properties in clients.values()}

885

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

684

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

685

886

686

887

687

class PrintTable(Output):

888

688

def __init__(self, verbose=False):

896

696

keywords = self.all_keywords

897

697

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

898

698

899

class TableOfClients:

699

class TableOfClients(object):

900

700

tableheaders = {

901

701

"Name": "Name",

902

702

"Enabled": "Enabled",

929

729

930

730

if sys.version_info.major == 2:

931

731

__unicode__ = __str__

932

933

732

def __str__(self):

934

733

return str(self).encode(

935

734

locale.getpreferredencoding())

981

780

minutes=(td.seconds % 3600) // 60,

982

781

seconds=td.seconds % 60))

983

782

783

984

784

class PropertySetter(Base):

985

785

"Abstract class for Actions for setting one client property"

986

786

993

793

def propname(self):

994

794

raise NotImplementedError()

995

795

796

996

797

class Enable(PropertySetter):

997

798

propname = "Enabled"

998

799

value_to_set = True

999

800

801

1000

802

class Disable(PropertySetter):

1001

803

propname = "Enabled"

1002

804

value_to_set = False

1003

805

806

1004

807

class BumpTimeout(PropertySetter):

1005

808

propname = "LastCheckedOK"

1006

809

value_to_set = ""

1007

810

811

1008

812

class StartChecker(PropertySetter):

1009

813

propname = "CheckerRunning"

1010

814

value_to_set = True

1011

815

816

1012

817

class StopChecker(PropertySetter):

1013

818

propname = "CheckerRunning"

1014

819

value_to_set = False

1015

820

821

1016

822

class ApproveByDefault(PropertySetter):

1017

823

propname = "ApprovedByDefault"

1018

824

value_to_set = True

1019

825

826

1020

827

class DenyByDefault(PropertySetter):

1021

828

propname = "ApprovedByDefault"

1022

829

value_to_set = False

1023

830

831

1024

832

class PropertySetterValue(PropertySetter):

1025

833

"""Abstract class for PropertySetter recieving a value as

1026

834

constructor argument instead of a class attribute."""

1027

1028

835

def __init__(self, value):

1029

836

self.value_to_set = value

1030

837

1031

@classmethod

1032

def argparse(cls, argtype):

1033

def cmdtype(arg):

1034

return cls(argtype(arg))

1035

return cmdtype

1036

838

1037

839

class SetChecker(PropertySetterValue):

1038

840

propname = "Checker"

1039

841

842

1040

843

class SetHost(PropertySetterValue):

1041

844

propname = "Host"

1042

845

846

1043

847

class SetSecret(PropertySetterValue):

1044

848

propname = "Secret"

1045

849

1053

857

self._vts = value.read()

1054

858

value.close()

1055

859

860

1056

861

class PropertySetterValueMilliseconds(PropertySetterValue):

1057

862

"""Abstract class for PropertySetterValue taking a value

1058

863

argument as a datetime.timedelta() but should store it as

1067

872

"When setting, convert value from a datetime.timedelta"

1068

873

self._vts = int(round(value.total_seconds() * 1000))

1069

874

875

1070

876

class SetTimeout(PropertySetterValueMilliseconds):

1071

877

propname = "Timeout"

1072

878

879

1073

880

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1074

881

propname = "ExtendedTimeout"

1075

882

883

1076

884

class SetInterval(PropertySetterValueMilliseconds):

1077

885

propname = "Interval"

1078

886

887

1079

888

class SetApprovalDelay(PropertySetterValueMilliseconds):

1080

889

propname = "ApprovalDelay"

1081

890

891

1082

892

class SetApprovalDuration(PropertySetterValueMilliseconds):

1083

893

propname = "ApprovalDuration"

1084

894

1118

928

"output"))

1119

929

1120

930

1121

class Unique:

931

class Unique(object):

1122

932

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

1123

933

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

1124

934

1156

966

1157

967

def test_actions_requires_client_or_all(self):

1158

968

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

1159

args = self.actionargs(action, value)

969

options = self.parser.parse_args()

970

setattr(options, action, value)

1160

971

with self.assertParseError():

1161

self.parse_args(args)

972

self.check_option_syntax(options)

1162

973

1163

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

974

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

1164

975

# check_option_syntax()

1165

976

actions = {

1166

"--enable": None,

1167

"--disable": None,

1168

"--bump-timeout": None,

1169

"--start-checker": None,

1170

"--stop-checker": None,

1171

"--is-enabled": None,

1172

"--remove": None,

1173

"--checker": "x",

1174

"--timeout": "PT0S",

1175

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

1176

"--interval": "PT0S",

1177

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

1178

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

1179

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

1180

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

1181

"--host": "hostname",

1182

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

1183

"--approve": None,

1184

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

1185

998

}

1186

999

1187

@staticmethod

1188

def actionargs(action, value, *args):

1189

if value is not None:

1190

return [action, value] + list(args)

1191

else:

1192

return [action] + list(args)

1193

1194

1000

@contextlib.contextmanager

1195

1001

def assertParseError(self):

1196

1002

with self.assertRaises(SystemExit) as e:

1201

1007

# /argparse.html#exiting-methods

1202

1008

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

1203

1009

1204

def parse_args(self, args):

1205

options = self.parser.parse_args(args)

1206

check_option_syntax(self.parser, options)

1207

1208

1010

@staticmethod

1209

1011

@contextlib.contextmanager

1210

1012

def redirect_stderr_to_devnull():

1221

1023

1222

1024

def test_actions_all_conflicts_with_verbose(self):

1223

1025

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

1224

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

1225

"--verbose")

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1226

1030

with self.assertParseError():

1227

self.parse_args(args)

1031

self.check_option_syntax(options)

1228

1032

1229

1033

def test_actions_with_client_conflicts_with_verbose(self):

1230

1034

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

1231

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

1232

"client")

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1233

1039

with self.assertParseError():

1234

self.parse_args(args)

1040

self.check_option_syntax(options)

1235

1041

1236

1042

def test_dump_json_conflicts_with_verbose(self):

1237

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

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1238

1046

with self.assertParseError():

1239

self.parse_args(args)

1047

self.check_option_syntax(options)

1240

1048

1241

1049

def test_dump_json_conflicts_with_action(self):

1242

1050

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

1243

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1244

1054

with self.assertParseError():

1245

self.parse_args(args)

1055

self.check_option_syntax(options)

1246

1056

1247

1057

def test_all_can_not_be_alone(self):

1248

args = ["--all"]

1058

options = self.parser.parse_args()

1059

options.all = True

1249

1060

with self.assertParseError():

1250

self.parse_args(args)

1061

self.check_option_syntax(options)

1251

1062

1252

1063

def test_all_is_ok_with_any_action(self):

1253

1064

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

1254

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

1255

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)

1256

1069

1257

1070

def test_any_action_is_ok_with_one_client(self):

1258

1071

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

1259

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

1260

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)

1261

1076

1262

1077

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1263

1079

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

1264

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

1080

if action == "is_enabled":

1265

1081

continue

1266

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

1267

self.parse_args(args)

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1268

1085

1269

1086

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1270

1088

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

1271

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

1089

if action == "is_enabled":

1272

1090

continue

1273

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

1274

"client2")

1275

self.parse_args(args)

1091

setattr(options, action, value)

1092

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

1093

self.check_option_syntax(options)

1276

1094

1277

1095

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1278

1096

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

1279

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

1097

if action == "is_enabled":

1280

1098

continue

1281

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

1282

"client2")

1283

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)

1284

1103

1285

1104

def test_is_enabled_fails_without_client(self):

1286

args = ["--is-enabled"]

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1287

1107

with self.assertParseError():

1288

self.parse_args(args)

1108

self.check_option_syntax(options)

1289

1109

1290

1110

def test_is_enabled_fails_with_two_clients(self):

1291

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

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

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

1292

1114

with self.assertParseError():

1293

self.parse_args(args)

1115

self.check_option_syntax(options)

1294

1116

1295

1117

def test_remove_can_only_be_combined_with_action_deny(self):

1296

1118

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

1297

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

1119

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

1298

1120

continue

1299

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

1300

"--remove")

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1301

1125

with self.assertParseError():

1302

self.parse_args(args)

1126

self.check_option_syntax(options)

1303

1127

1304

1128

1305

1129

class Test_dbus_exceptions(unittest.TestCase):

1408

1232

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1409

1233

1410

1234

def MockDBusPython_func(self, func):

1411

class mock_dbus_python:

1235

class mock_dbus_python(object):

1412

1236

"""mock dbus-python module"""

1413

class exceptions:

1237

class exceptions(object):

1414

1238

"""Pseudo-namespace"""

1415

1239

class DBusException(Exception):

1416

1240

pass

1417

class SystemBus:

1241

class SystemBus(object):

1418

1242

@staticmethod

1419

1243

def get_object(busname, objectpath):

1420

1244

DBusObject = collections.namedtuple(

1421

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

1245

"DBusObject", ("methodname",))

1422

1246

def method(*args, **kwargs):

1423

1247

self.assertEqual({"dbus_interface":

1424

1248

"interface"},

1425

1249

kwargs)

1426

1250

return func(*args)

1427

def set_property(interface, key, value,

1428

dbus_interface=None):

1429

self.assertEqual(

1430

"org.freedesktop.DBus.Properties",

1431

dbus_interface)

1432

self.assertEqual("Secret", key)

1433

return func(interface, key, value,

1434

dbus_interface=dbus_interface)

1435

return DBusObject(methodname=method,

1436

Set=set_property)

1437

class Boolean:

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

1438

1253

def __init__(self, value):

1439

1254

self.value = bool(value)

1440

1255

def __bool__(self):

1445

1260

pass

1446

1261

class Dictionary(dict):

1447

1262

pass

1448

class ByteArray(bytes):

1449

pass

1450

1263

return mock_dbus_python

1451

1264

1452

1265

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

1624

1437

finally:

1625

1438

dbus_logger.removeFilter(counting_handler)

1626

1439

1627

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

1440

self.assertNotIsInstance(e, dbus.ConnectFailed)

1628

1441

1629

1442

# Make sure the dbus logger was suppressed

1630

1443

self.assertEqual(0, counting_handler.count)

1631

1444

1632

def test_Set_Secret_sends_bytearray(self):

1633

ret = [None]

1634

def func(*args, **kwargs):

1635

ret[0] = (args, kwargs)

1636

mock_dbus_python = self.MockDBusPython_func(func)

1637

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1638

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

1639

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

1640

mock_dbus_python.ByteArray(b"value")),

1641

{"dbus_interface":

1642

"org.freedesktop.DBus.Properties"})

1643

self.assertEqual(expected_call, ret[0])

1644

if sys.version_info.major == 2:

1645

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

1646

mock_dbus_python.ByteArray)

1647

1648

1445

def test_get_object_converts_to_correct_exception(self):

1649

1446

bus = dbus_python_adapter.SystemBus(

1650

1447

self.fake_dbus_python_raises_exception_on_connect)

1652

1449

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

1653

1450

"objectpath", "interface")

1654

1451

1655

class fake_dbus_python_raises_exception_on_connect:

1452

class fake_dbus_python_raises_exception_on_connect(object):

1656

1453

"""fake dbus-python module"""

1657

class exceptions:

1454

class exceptions(object):

1658

1455

"""Pseudo-namespace"""

1659

1456

class DBusException(Exception):

1660

1457

pass

1668

1465

1669

1466

1670

1467

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1671

class mock_dbus_python:

1468

class mock_dbus_python(object):

1672

1469

"""mock dbus-python modules"""

1673

class SystemBus:

1470

class SystemBus(object):

1674

1471

@staticmethod

1675

1472

def get_object(busname, objectpath):

1676

1473

return Unique()

1719

1516

self.assertIs(obj1, obj1b)

1720

1517

1721

1518

1722

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1723

1724

def Stub_pydbus_func(self, func):

1725

class stub_pydbus:

1726

"""stub pydbus module"""

1727

class SystemBus:

1728

@staticmethod

1729

def get(busname, objectpath):

1730

DBusObject = collections.namedtuple(

1731

"DBusObject", ("methodname",))

1732

return {"interface":

1733

DBusObject(methodname=func)}

1734

return stub_pydbus

1735

1736

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

1737

interface, *args):

1738

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

1739

return bus.call_method(methodname, busname, objectpath,

1740

interface, *args)

1741

1742

def test_call_method_returns(self):

1743

expected_method_return = Unique()

1744

method_args = (Unique(), Unique())

1745

def func(*args):

1746

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

1747

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

1748

self.assertIs(marg, arg)

1749

return expected_method_return

1750

stub_pydbus = self.Stub_pydbus_func(func)

1751

bus = pydbus_adapter.SystemBus(stub_pydbus)

1752

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

1753

"objectpath", "interface",

1754

*method_args)

1755

self.assertIs(ret, expected_method_return)

1756

1757

def test_call_method_handles_exception(self):

1758

def func():

1759

raise gi.repository.GLib.Error()

1760

1761

stub_pydbus = self.Stub_pydbus_func(func)

1762

bus = pydbus_adapter.SystemBus(stub_pydbus)

1763

1764

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

1765

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

1766

"objectpath", "interface")

1767

1768

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

1769

1770

def test_get_converts_to_correct_exception(self):

1771

bus = pydbus_adapter.SystemBus(

1772

self.fake_pydbus_raises_exception_on_connect)

1773

with self.assertRaises(dbus.ConnectFailed):

1774

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

1775

"objectpath", "interface")

1776

1777

class fake_pydbus_raises_exception_on_connect:

1778

"""fake dbus-python module"""

1779

@classmethod

1780

def SystemBus(cls):

1781

def get(busname, objectpath):

1782

raise gi.repository.GLib.Error()

1783

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

1784

return Bus(get=get)

1785

1786

def test_set_property_uses_setattr(self):

1787

class Object:

1788

pass

1789

obj = Object()

1790

class pydbus_spy:

1791

class SystemBus:

1792

@staticmethod

1793

def get(busname, objectpath):

1794

return {"interface": obj}

1795

bus = pydbus_adapter.SystemBus(pydbus_spy)

1796

value = Unique()

1797

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

1798

value)

1799

self.assertIs(value, obj.key)

1800

1801

def test_get_suppresses_xml_deprecation_warning(self):

1802

if sys.version_info.major >= 3:

1803

return

1804

class stub_pydbus_get:

1805

class SystemBus:

1806

@staticmethod

1807

def get(busname, objectpath):

1808

warnings.warn_explicit(

1809

"deprecated", DeprecationWarning,

1810

"xml.etree.ElementTree", 0)

1811

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1812

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

1813

warnings.simplefilter("always")

1814

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

1815

self.assertEqual(0, len(w))

1816

1817

1818

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1819

class stub_pydbus:

1820

"""stub pydbus module"""

1821

class SystemBus:

1822

@staticmethod

1823

def get(busname, objectpath):

1824

return Unique()

1825

1826

def setUp(self):

1827

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

1828

1829

def test_returns_distinct_objectpaths(self):

1830

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

1831

self.assertIsInstance(obj1, Unique)

1832

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

1833

self.assertIsInstance(obj2, Unique)

1834

self.assertIsNot(obj1, obj2)

1835

1836

def test_returns_distinct_busnames(self):

1837

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

1838

self.assertIsInstance(obj1, Unique)

1839

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

1840

self.assertIsInstance(obj2, Unique)

1841

self.assertIsNot(obj1, obj2)

1842

1843

def test_returns_distinct_both(self):

1844

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

1845

self.assertIsInstance(obj1, Unique)

1846

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

1847

self.assertIsInstance(obj2, Unique)

1848

self.assertIsNot(obj1, obj2)

1849

1850

def test_returns_same(self):

1851

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

1852

self.assertIsInstance(obj1, Unique)

1853

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

1854

self.assertIsInstance(obj2, Unique)

1855

self.assertIs(obj1, obj2)

1856

1857

def test_returns_same_old(self):

1858

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

1859

self.assertIsInstance(obj1, Unique)

1860

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

1861

self.assertIsInstance(obj2, Unique)

1862

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

1863

self.assertIsInstance(obj1b, Unique)

1864

self.assertIsNot(obj1, obj2)

1865

self.assertIsNot(obj2, obj1b)

1866

self.assertIs(obj1, obj1b)

1867

1868

1869

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1870

1871

class dummy_dbussy:

1872

class DBUS:

1873

class ObjectPath(str):

1874

pass

1875

class DBusError(Exception):

1876

pass

1877

1878

def fake_ravel_func(self, func):

1879

class fake_ravel:

1880

@staticmethod

1881

def system_bus():

1882

class DBusInterfaceProxy:

1883

@staticmethod

1884

def methodname(*args):

1885

return [func(*args)]

1886

class DBusObject:

1887

@staticmethod

1888

def get_interface(interface):

1889

if interface == "interface":

1890

return DBusInterfaceProxy()

1891

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

1892

return fake_ravel

1893

1894

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

1895

interface, *args):

1896

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

1897

return bus.call_method(methodname, busname, objectpath,

1898

interface, *args)

1899

1900

def test_call_method_returns(self):

1901

expected_method_return = Unique()

1902

method_args = (Unique(), Unique())

1903

def func(*args):

1904

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

1905

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

1906

self.assertIs(marg, arg)

1907

return expected_method_return

1908

fake_ravel = self.fake_ravel_func(func)

1909

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

1910

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

1911

"objectpath", "interface",

1912

*method_args)

1913

self.assertIs(ret, expected_method_return)

1914

1915

def test_call_method_filters_objectpath(self):

1916

def func():

1917

return method_return

1918

fake_ravel = self.fake_ravel_func(func)

1919

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

1920

method_return = (self.dummy_dbussy.DBUS

1921

.ObjectPath("objectpath"))

1922

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

1923

"objectpath", "interface")

1924

self.assertEqual("objectpath", ret)

1925

self.assertNotIsInstance(ret,

1926

self.dummy_dbussy.DBUS.ObjectPath)

1927

1928

def test_call_method_filters_objectpaths_in_dict(self):

1929

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1930

def func():

1931

return method_return

1932

fake_ravel = self.fake_ravel_func(func)

1933

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

1934

method_return = {

1935

ObjectPath("objectpath_key_1"):

1936

ObjectPath("objectpath_value_1"),

1937

ObjectPath("objectpath_key_2"):

1938

ObjectPath("objectpath_value_2"),

1939

}

1940

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

1941

"objectpath", "interface")

1942

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

1943

for key, value in

1944

method_return.items()}

1945

for key, value in ret.items():

1946

self.assertNotIsInstance(key, ObjectPath)

1947

self.assertNotIsInstance(value, ObjectPath)

1948

self.assertEqual(expected_method_return, ret)

1949

self.assertIsInstance(ret, dict)

1950

1951

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1952

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1953

def func():

1954

return method_return

1955

fake_ravel = self.fake_ravel_func(func)

1956

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

1957

method_return = {

1958

ObjectPath("key1"): {

1959

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

1960

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

1961

},

1962

ObjectPath("key2"): {

1963

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

1964

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

1965

},

1966

}

1967

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

1968

"objectpath", "interface")

1969

expected_method_return = {

1970

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

1971

"key12": "value12"},

1972

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

1973

"key22": "value22"},

1974

}

1975

self.assertEqual(expected_method_return, ret)

1976

for key, value in ret.items():

1977

self.assertIsInstance(value, dict)

1978

self.assertEqual(expected_method_return[key], value)

1979

self.assertNotIsInstance(key, ObjectPath)

1980

for inner_key, inner_value in value.items():

1981

self.assertIsInstance(value, dict)

1982

self.assertEqual(

1983

expected_method_return[key][inner_key],

1984

inner_value)

1985

self.assertNotIsInstance(key, ObjectPath)

1986

1987

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1988

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1989

def func():

1990

return method_return

1991

fake_ravel = self.fake_ravel_func(func)

1992

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

1993

method_return = {

1994

ObjectPath("key1"): {

1995

ObjectPath("key2"): {

1996

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

1997

},

1998

},

1999

}

2000

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

2001

"objectpath", "interface")

2002

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

2003

self.assertEqual(expected_method_return, ret)

2004

self.assertIsInstance(ret, dict)

2005

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

2006

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

2007

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

2008

ObjectPath)

2009

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

2010

self.assertNotIsInstance(

2011

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

2012

ObjectPath)

2013

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

2014

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

2015

self.dummy_dbussy.DBUS.ObjectPath)

2016

2017

def test_call_method_handles_exception(self):

2018

def func():

2019

raise self.dummy_dbussy.DBusError()

2020

2021

fake_ravel = self.fake_ravel_func(func)

2022

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

2023

2024

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

2025

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

2026

"objectpath", "interface")

2027

2028

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

2029

2030

def test_get_object_converts_to_correct_exception(self):

2031

class fake_ravel_raises_exception_on_connect:

2032

@staticmethod

2033

def system_bus():

2034

class Bus:

2035

@staticmethod

2036

def __getitem__(key):

2037

if key == "objectpath":

2038

raise self.dummy_dbussy.DBusError()

2039

raise Exception(key)

2040

return {"busname": Bus()}

2041

def func():

2042

raise self.dummy_dbussy.DBusError()

2043

bus = dbussy_adapter.SystemBus(

2044

self.dummy_dbussy,

2045

fake_ravel_raises_exception_on_connect)

2046

with self.assertRaises(dbus.ConnectFailed):

2047

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

2048

"objectpath", "interface")

2049

2050

2051

1519

class Test_commands_from_options(unittest.TestCase):

2052

1520

2053

1521

def setUp(self):

2058

1526

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

2059

1527

command.IsEnabled)

2060

1528

2061

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

2062

clients=None, **cmd_attrs):

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

2063

1531

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

2064

1532

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

2065

1533

options = self.parser.parse_args(args)

2066

1534

check_option_syntax(self.parser, options)

2067

1535

commands = commands_from_options(options)

2068

self.assertEqual(length, len(commands))

2069

for command in commands:

2070

if isinstance(command, command_cls):

2071

break

2072

else:

2073

self.assertIsInstance(command, command_cls)

2074

if clients is not None:

2075

self.assertEqual(clients, options.client)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

2076

1539

for key, value in cmd_attrs.items():

2077

1540

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

2078

1541

2079

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

2080

for cmd in commands:

2081

self.assert_command_from_args(args, cmd,

2082

length=len(commands),

2083

clients=clients)

2084

2085

1542

def test_is_enabled_short(self):

2086

1543

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

2087

1544

command.IsEnabled)

2278

1735

verbose=True)

2279

1736

2280

1737

2281

def test_manual_page_example_1(self):

2282

self.assert_command_from_args("",

2283

command.PrintTable,

2284

clients=[],

2285

verbose=False)

2286

2287

def test_manual_page_example_2(self):

2288

self.assert_command_from_args(

2289

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

2290

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

2291

"foo2.example.org"],

2292

verbose=True)

2293

2294

def test_manual_page_example_3(self):

2295

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

2296

command.Enable,

2297

clients=[])

2298

2299

def test_manual_page_example_4(self):

2300

self.assert_commands_from_args(

2301

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

2302

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

2303

[command.SetTimeout, command.SetInterval],

2304

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

2305

2306

def test_manual_page_example_5(self):

2307

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

2308

command.Approve,

2309

clients=[])

2310

2311

2312

1738

class TestCommand(unittest.TestCase):

2313

1739

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

2314

1740

2427

1853

busname = "se.recompile.Mandos"

2428

1854

client_interface = "se.recompile.Mandos.Client"

2429

1855

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

2430

self.assertTrue(self.bus.clients)

2431

1856

for clientpath in self.bus.clients:

2432

1857

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

2433

1858

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

2436

1861

busname = "se.recompile.Mandos"

2437

1862

client_interface = "se.recompile.Mandos.Client"

2438

1863

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

2439

self.assertTrue(self.bus.clients)

2440

1864

for clientpath in self.bus.clients:

2441

1865

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

2442

1866

client_interface, (False,)),

2443

1867

self.bus.calls)

2444

1868

2445

1869

def test_Remove(self):

2446

busname = "se.recompile.Mandos"

2447

server_path = "/"

2448

server_interface = "se.recompile.Mandos"

2449

orig_clients = self.bus.clients.copy()

2450

1870

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

2451

self.assertFalse(self.bus.clients)

2452

for clientpath in orig_clients:

2453

self.assertIn(("RemoveClient", busname,

2454

server_path, server_interface,

1871

for clientpath in self.bus.clients:

1872

self.assertIn(("RemoveClient", dbus_busname,

1873

dbus_server_path, dbus_server_interface,

2455

1874

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

2456

1875

2457

1876

expected_json = {

2659

2078

else:

2660

2079

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

2661

2080

values_to_get = self.values_to_get

2662

self.assertTrue(values_to_get)

2663

2081

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

2664

2082

for clientpath in self.bus.clients:

2665

2083

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

2666

2084

Unique())

2667

2085

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

2668

self.assertTrue(self.bus.clients)

2669

2086

for clientpath in self.bus.clients:

2670

2087

value = (self.bus.clients[clientpath]

2671

2088

[self.propname])

2730

2147

class TestSetSecretCmd(TestPropertySetterCmd):

2731

2148

command = command.SetSecret

2732

2149

propname = "Secret"

2733

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

2734

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

2735

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

2736

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

2737

self.values_to_set]

2738

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

2150

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

2151

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

2152

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

2739

2153

2740

2154

2741

2155

class TestSetTimeoutCmd(TestPropertySetterCmd):

2794

2208

2795

2209

2796

2210

2797

def parse_test_args():

2798

# type: () -> argparse.Namespace

2211

def should_only_run_tests():

2799

2212

parser = argparse.ArgumentParser(add_help=False)

2800

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

2801

parser.add_argument("--prefix", )

2213

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

2802

2214

args, unknown_args = parser.parse_known_args()

2803

if args.check:

2804

# Remove test options from sys.argv

2215

run_tests = args.check

2216

if run_tests:

2217

# Remove --check argument from sys.argv

2805

2218

sys.argv[1:] = unknown_args

2806

return args

2219

return run_tests

2807

2220

2808

2221

# Add all tests from doctest strings

2809

2222

def load_tests(loader, tests, none):

2812

2225

return tests

2813

2226

2814

2227

if __name__ == "__main__":

2815

options = parse_test_args()

2816

2228

try:

2817

if options.check:

2818

extra_test_prefix = options.prefix

2819

if extra_test_prefix is not None:

2820

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

2821

.result.wasSuccessful()):

2822

sys.exit(1)

2823

class ExtraTestLoader(unittest.TestLoader):

2824

testMethodPrefix = extra_test_prefix

2825

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

2826

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

2827

else:

2828

unittest.main(argv=[""])

2229

if should_only_run_tests():

2230

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

2231

unittest.main()

2829

2232

else:

2830

2233

main()

2831

2234

finally:

2832

2235

logging.shutdown()

2833

2834

# Local Variables:

2835

# run-tests:

2836

# (lambda (&optional extra)

2837

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

2838

# extra))

2839

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

2840

# (funcall remove-test-window)

2841

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

2842

# run-tests-in-test-buffer:

2843

# (lambda (dir &optional extra)

2844

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

2845

# (setq buffer-read-only nil

2846

# default-directory dir)

2847

# (erase-buffer)

2848

# (compilation-mode))

2849

# (let ((process-result

2850

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

2851

# (process-file-shell-command

2852

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

2853

# (and (numberp process-result)

2854

# (= process-result 0))))

2855

# get-command-line:

2856

# (lambda (&optional extra)

2857

# (let ((quoted-script

2858

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

2859

# (format

2860

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

2861

# quoted-script)))

2862

# get-script-name:

2863

# (lambda ()

2864

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

2865

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

2866

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

2867

# (buffer-file-name))))

2868

# remove-test-window:

2869

# (lambda ()

2870

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

2871

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

2872

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

2873

# (lambda ()

2874

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

2875

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

2876

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

2877

# (display-buffer-overriding-action

2878

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

2879

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

2880

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

2881

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

2882

# eval:

2883

# (progn

2884

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

2885

# (funcall run-tests t)))

2886

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

2887

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

2888

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

2889

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

2890

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

2891

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

2892

# End: