/mandos/release : revision 24.1.63

1

#!/usr/bin/python3 -bbI

2

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

# under the terms of the GNU General Public License as published by

13

# the Free Software Foundation, either version 3 of the License, or

14

# (at your option) any later version.

15

#

16

# Mandos is distributed in the hope that it will be useful, but

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

# You should have received a copy of the GNU General Public License

22

# along with Mandos. If not, see <http://www.gnu.org/licenses/>.

23

#

24

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

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

35

import sys

36

import argparse

37

import locale

38

import datetime

39

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

if sys.version_info.major == 2:

50

__metaclass__ = type

51

str = unicode

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

74

75

# Show warnings by default

76

if not sys.warnoptions:

77

import warnings

78

warnings.simplefilter("default")

79

80

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

81

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

82

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

83

84

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

85

86

if sys.version_info.major == 2:

87

import StringIO

88

io.StringIO = StringIO.StringIO

89

90

locale.setlocale(locale.LC_ALL, "")

91

92

version = "1.8.9"

93

94

95

def main():

96

parser = argparse.ArgumentParser()

97

add_command_line_options(parser)

98

99

options = parser.parse_args()

100

check_option_syntax(parser, options)

101

102

clientnames = options.client

103

104

if options.debug:

105

log.setLevel(logging.DEBUG)

106

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)

113

114

try:

115

all_clients = bus.get_clients_and_properties()

116

except dbus.ConnectFailed as e:

117

log.critical("Could not connect to Mandos server: %s", e)

118

sys.exit(1)

119

except dbus.Error as e:

120

log.critical(

121

"Failed to access Mandos server through D-Bus:\n%s", e)

122

sys.exit(1)

123

124

# Compile dict of (clientpath: properties) to process

125

if not clientnames:

126

clients = all_clients

127

else:

128

clients = {}

129

for name in clientnames:

130

for objpath, properties in all_clients.items():

131

if properties["Name"] == name:

132

clients[objpath] = properties

133

break

134

else:

135

log.critical("Client not found on server: %r", name)

136

sys.exit(1)

137

138

commands = commands_from_options(options)

139

140

for command in commands:

141

command.run(clients, bus)

142

143

144

def add_command_line_options(parser):

145

parser.add_argument("--version", action="version",

146

version="%(prog)s {}".format(version),

147

help="show version number and exit")

148

parser.add_argument("-a", "--all", action="store_true",

149

help="Select all clients")

150

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

151

help="Print all fields")

152

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

153

action="append_const", default=[],

154

const=command.DumpJSON(),

155

help="Dump client data in JSON format")

156

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

160

help="Enable client")

161

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

162

action="append_const", default=[],

163

const=command.Disable(),

164

help="disable client")

165

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

166

action="append_const", default=[],

167

const=command.BumpTimeout(),

168

help="Bump timeout for client")

169

start_stop_checker = parser.add_mutually_exclusive_group()

170

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

171

dest="commands",

172

action="append_const", default=[],

173

const=command.StartChecker(),

174

help="Start checker for client")

175

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

176

action="append_const", default=[],

177

const=command.StopChecker(),

178

help="Stop checker for client")

179

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

180

action="append_const", default=[],

181

const=command.IsEnabled(),

182

help="Check if client is enabled")

183

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

184

action="append_const", default=[],

185

const=command.Remove(),

186

help="Remove client")

187

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

188

action="append", default=[],

189

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

190

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

206

approve_deny_default = parser.add_mutually_exclusive_group()

207

approve_deny_default.add_argument(

208

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

209

action="append_const", default=[],

210

const=command.ApproveByDefault(),

211

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

212

approve_deny_default.add_argument(

213

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

214

action="append_const", default=[],

215

const=command.DenyByDefault(),

216

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

236

approve_deny = parser.add_mutually_exclusive_group()

237

approve_deny.add_argument(

238

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

239

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

240

help="Approve any current client request")

241

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

242

action="append_const", default=[],

243

const=command.Deny(),

244

help="Deny any current client request")

245

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

246

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

247

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

248

help="Run self-test")

249

parser.add_argument("client", nargs="*", help="Client name")

250

251

252

def string_to_delta(interval):

253

"""Parse a string and return a datetime.timedelta"""

254

255

try:

256

return rfc3339_duration_to_delta(interval)

257

except ValueError as e:

258

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

259

' '.join(e.args))

260

return parse_pre_1_6_1_interval(interval)

261

262

263

def rfc3339_duration_to_delta(duration):

264

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

265

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

283

>>> # Can not be empty:

284

>>> rfc3339_duration_to_delta("")

285

Traceback (most recent call last):

286

...

287

ValueError: Invalid RFC 3339 duration: ""

288

>>> # Must start with "P":

289

>>> rfc3339_duration_to_delta("1D")

290

Traceback (most recent call last):

291

...

292

ValueError: Invalid RFC 3339 duration: "1D"

293

>>> # Must use correct order

294

>>> rfc3339_duration_to_delta("PT1S2M")

295

Traceback (most recent call last):

296

...

297

ValueError: Invalid RFC 3339 duration: "PT1S2M"

298

>>> # Time needs time marker

299

>>> rfc3339_duration_to_delta("P1H2S")

300

Traceback (most recent call last):

301

...

302

ValueError: Invalid RFC 3339 duration: "P1H2S"

303

>>> # Weeks can not be combined with anything else

304

>>> rfc3339_duration_to_delta("P1D2W")

305

Traceback (most recent call last):

306

...

307

ValueError: Invalid RFC 3339 duration: "P1D2W"

308

>>> rfc3339_duration_to_delta("P2W2H")

309

Traceback (most recent call last):

310

...

311

ValueError: Invalid RFC 3339 duration: "P2W2H"

312

"""

313

314

# Parsing an RFC 3339 duration with regular expressions is not

315

# possible - there would have to be multiple places for the same

316

# values, like seconds. The current code, while more esoteric, is

317

# cleaner without depending on a parsing library. If Python had a

318

# built-in library for parsing we would use it, but we'd like to

319

# avoid excessive use of external libraries.

320

321

# New type for defining tokens, syntax, and semantics all-in-one

322

Token = collections.namedtuple("Token", (

323

"regexp", # To match token; if "value" is not None, must have

324

# a "group" containing digits

325

"value", # datetime.timedelta or None

326

"followers")) # Tokens valid after this token

327

# RFC 3339 "duration" tokens, syntax, and semantics; taken from

328

# the "duration" ABNF definition in RFC 3339, Appendix A.

329

token_end = Token(re.compile(r"$"), None, frozenset())

330

token_second = Token(re.compile(r"(\d+)S"),

331

datetime.timedelta(seconds=1),

332

frozenset((token_end, )))

333

token_minute = Token(re.compile(r"(\d+)M"),

334

datetime.timedelta(minutes=1),

335

frozenset((token_second, token_end)))

336

token_hour = Token(re.compile(r"(\d+)H"),

337

datetime.timedelta(hours=1),

338

frozenset((token_minute, token_end)))

339

token_time = Token(re.compile(r"T"),

340

None,

341

frozenset((token_hour, token_minute,

342

token_second)))

343

token_day = Token(re.compile(r"(\d+)D"),

344

datetime.timedelta(days=1),

345

frozenset((token_time, token_end)))

346

token_month = Token(re.compile(r"(\d+)M"),

347

datetime.timedelta(weeks=4),

348

frozenset((token_day, token_end)))

349

token_year = Token(re.compile(r"(\d+)Y"),

350

datetime.timedelta(weeks=52),

351

frozenset((token_month, token_end)))

352

token_week = Token(re.compile(r"(\d+)W"),

353

datetime.timedelta(weeks=1),

354

frozenset((token_end, )))

355

token_duration = Token(re.compile(r"P"), None,

356

frozenset((token_year, token_month,

357

token_day, token_time,

358

token_week)))

359

# Define starting values:

360

# Value so far

361

value = datetime.timedelta()

362

found_token = None

363

# Following valid tokens

364

followers = frozenset((token_duration, ))

365

# String left to parse

366

s = duration

367

# Loop until end token is found

368

while found_token is not token_end:

369

# Search for any currently valid tokens

370

for token in followers:

371

match = token.regexp.match(s)

372

if match is not None:

373

# Token found

374

if token.value is not None:

375

# Value found, parse digits

376

factor = int(match.group(1), 10)

377

# Add to value so far

378

value += factor * token.value

379

# Strip token from string

380

s = token.regexp.sub("", s, 1)

381

# Go to found token

382

found_token = token

383

# Set valid next tokens

384

followers = found_token.followers

385

break

386

else:

387

# No currently valid tokens were found

388

raise ValueError("Invalid RFC 3339 duration: \"{}\""

389

.format(duration))

390

# End token found

391

return value

392

393

394

def parse_pre_1_6_1_interval(interval):

395

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

396

and return a datetime.timedelta

397

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

412

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

413

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

414

True

415

416

"""

417

418

value = datetime.timedelta(0)

419

regexp = re.compile(r"(\d+)([dsmhw]?)")

420

421

for num, suffix in regexp.findall(interval):

422

if suffix == "d":

423

value += datetime.timedelta(int(num))

424

elif suffix == "s":

425

value += datetime.timedelta(0, int(num))

426

elif suffix == "m":

427

value += datetime.timedelta(0, 0, 0, 0, int(num))

428

elif suffix == "h":

429

value += datetime.timedelta(0, 0, 0, 0, 0, int(num))

430

elif suffix == "w":

431

value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))

432

elif suffix == "":

433

value += datetime.timedelta(0, 0, 0, int(num))

434

return value

435

436

437

def check_option_syntax(parser, options):

438

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

439

argparse"""

440

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)

464

465

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

466

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

467

if options.verbose and has_commands(options):

468

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

469

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

470

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

471

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

472

if options.all and not has_commands(options):

473

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

474

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

475

and len(options.client) > 1):

476

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:

486

487

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

def get_managed_objects(self, busname, objectpath):

489

return self.call_method("GetManagedObjects", busname,

490

objectpath,

491

self.object_manager_iface)

492

493

properties_iface = "org.freedesktop.DBus.Properties"

494

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

495

value):

496

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

497

self.properties_iface, interface, key,

498

value)

499

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

505

class MandosBus(SystemBus):

506

busname_domain = "se.recompile"

507

busname = busname_domain + ".Mandos"

508

server_path = "/"

509

server_interface = busname_domain + ".Mandos"

510

client_interface = busname_domain + ".Mandos.Client"

511

del busname_domain

512

513

def get_clients_and_properties(self):

514

managed_objects = self.get_managed_objects(

515

self.busname, self.server_path)

516

return {objpath: properties[self.client_interface]

517

for objpath, properties in managed_objects.items()

518

if self.client_interface in properties}

519

520

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

521

return self.set_property(self.busname, objectpath,

522

self.client_interface, key,

523

value)

524

525

def call_client_method(self, objectpath, method, *args):

526

return self.call_method(method, self.busname, objectpath,

527

self.client_interface, *args)

528

529

def call_server_method(self, method, *args):

530

return self.call_method(method, self.busname,

531

self.server_path,

532

self.server_interface, *args)

533

534

class Error(Exception):

535

pass

536

537

class ConnectFailed(Error):

538

pass

539

540

541

class dbus_python_adapter:

542

543

class SystemBus(dbus.MandosBus):

544

"""Use dbus-python"""

545

546

def __init__(self, module=dbus_python):

547

self.dbus_python = module

548

self.bus = self.dbus_python.SystemBus()

549

550

@contextlib.contextmanager

551

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

552

try:

553

yield

554

except self.dbus_python.exceptions.DBusException as e:

555

# This does what "raise from" would do

556

exc = exception_class(*e.args)

557

exc.__cause__ = e

558

raise exc

559

560

def call_method(self, methodname, busname, objectpath,

561

interface, *args):

562

proxy_object = self.get_object(busname, objectpath)

563

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

564

interface, methodname,

565

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

566

method = getattr(proxy_object, methodname)

567

with self.convert_exception():

568

with dbus_python_adapter.SilenceLogger(

569

"dbus.proxies"):

570

value = method(*args, dbus_interface=interface)

571

return self.type_filter(value)

572

573

def get_object(self, busname, objectpath):

574

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

575

busname, objectpath)

576

with self.convert_exception(dbus.ConnectFailed):

577

return self.bus.get_object(busname, objectpath)

578

579

def type_filter(self, value):

580

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

581

if isinstance(value, self.dbus_python.Boolean):

582

return bool(value)

583

if isinstance(value, self.dbus_python.ObjectPath):

584

return str(value)

585

# Also recurse into dictionaries

586

if isinstance(value, self.dbus_python.Dictionary):

587

return {self.type_filter(key):

588

self.type_filter(subval)

589

for key, subval in value.items()}

590

return value

591

592

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

593

if key == "Secret":

594

if not isinstance(value, bytes):

595

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

596

value = self.dbus_python.ByteArray(value)

597

return self.set_property(self.busname, objectpath,

598

self.client_interface, key,

599

value)

600

601

class SilenceLogger:

602

"Simple context manager to silence a particular logger"

603

def __init__(self, loggername):

604

self.logger = logging.getLogger(loggername)

605

606

def __enter__(self):

607

self.logger.addFilter(self.nullfilter)

608

609

class NullFilter(logging.Filter):

610

def filter(self, record):

611

return False

612

613

nullfilter = NullFilter()

614

615

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

616

self.logger.removeFilter(self.nullfilter)

617

618

619

class CachingBus(SystemBus):

620

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

621

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

622

self.object_cache = {}

623

super(dbus_python_adapter.CachingBus,

624

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

625

def get_object(self, busname, objectpath):

626

try:

627

return self.object_cache[(busname, objectpath)]

628

except KeyError:

629

new_object = super(

630

dbus_python_adapter.CachingBus,

631

self).get_object(busname, objectpath)

632

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

633

return new_object

634

635

636

class pydbus_adapter:

637

class SystemBus(dbus.MandosBus):

638

def __init__(self, module=pydbus):

639

self.pydbus = module

640

self.bus = self.pydbus.SystemBus()

641

642

@contextlib.contextmanager

643

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

644

try:

645

yield

646

except gi.repository.GLib.Error as e:

647

# This does what "raise from" would do

648

exc = exception_class(*e.args)

649

exc.__cause__ = e

650

raise exc

651

652

def call_method(self, methodname, busname, objectpath,

653

interface, *args):

654

proxy_object = self.get(busname, objectpath)

655

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

656

interface, methodname,

657

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

658

method = getattr(proxy_object[interface], methodname)

659

with self.convert_exception():

660

return method(*args)

661

662

def get(self, busname, objectpath):

663

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

664

busname, objectpath)

665

with self.convert_exception(dbus.ConnectFailed):

666

if sys.version_info.major <= 2:

667

with warnings.catch_warnings():

668

warnings.filterwarnings(

669

"ignore", "", DeprecationWarning,

670

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

671

return self.bus.get(busname, objectpath)

672

else:

673

return self.bus.get(busname, objectpath)

674

675

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

676

value):

677

proxy_object = self.get(busname, objectpath)

678

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

679

objectpath, self.properties_iface, interface,

680

key, value)

681

setattr(proxy_object[interface], key, value)

682

683

class CachingBus(SystemBus):

684

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

685

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

686

self.object_cache = {}

687

super(pydbus_adapter.CachingBus,

688

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

689

def get(self, busname, objectpath):

690

try:

691

return self.object_cache[(busname, objectpath)]

692

except KeyError:

693

new_object = (super(pydbus_adapter.CachingBus, self)

694

.get(busname, objectpath))

695

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

696

return new_object

697

698

699

class dbussy_adapter:

700

class SystemBus(dbus.SystemBus):

701

"""Use DBussy"""

702

703

def __init__(self, dbussy, ravel):

704

self.dbussy = dbussy

705

self.ravel = ravel

706

self.bus = ravel.system_bus()

707

708

@contextlib.contextmanager

709

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

710

try:

711

yield

712

except self.dbussy.DBusError as e:

713

# This does what "raise from" would do

714

exc = exception_class(*e.args)

715

exc.__cause__ = e

716

raise exc

717

718

def call_method(self, methodname, busname, objectpath,

719

interface, *args):

720

proxy_object = self.get_object(busname, objectpath)

721

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

722

interface, methodname,

723

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

724

iface = proxy_object.get_interface(interface)

725

method = getattr(iface, methodname)

726

with self.convert_exception(dbus.Error):

727

value = method(*args)

728

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

729

# tuple: (signature, 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

if isinstance(value, tuple):

742

if (len(value) == 2

743

and isinstance(value[0],

744

self.dbussy.DBUS.Signature)):

745

return self.type_filter(value[1])

746

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

747

return str(value)

748

# Also recurse into dictionaries

749

elif isinstance(value, dict):

750

return {self.type_filter(key):

751

self.type_filter(subval)

752

for key, subval in value.items()}

753

return value

754

755

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

756

value):

757

proxy_object = self.get_object(busname, objectpath)

758

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

759

objectpath, self.properties_iface, interface,

760

key, value)

761

if key == "Secret":

762

# DBussy wants a Byte Array to be a sequence of

763

# values, not a byte string

764

value = tuple(value)

765

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

766

767

class MandosBus(SystemBus, dbus.MandosBus):

768

pass

769

770

class CachingBus(MandosBus):

771

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

772

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

773

self.object_cache = {}

774

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

775

**kwargs)

776

def get_object(self, busname, objectpath):

777

try:

778

return self.object_cache[(busname, objectpath)]

779

except KeyError:

780

new_object = super(

781

dbussy_adapter.CachingBus,

782

self).get_object(busname, objectpath)

783

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

784

return new_object

785

786

787

def commands_from_options(options):

788

789

commands = list(options.commands)

790

791

def find_cmd(cmd, commands):

792

i = 0

793

for i, c in enumerate(commands):

794

if isinstance(c, cmd):

795

return i

796

return i+1

797

798

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

799

# to occur ahead of command.Remove.

800

index_of_remove = find_cmd(command.Remove, commands)

801

before_remove = commands[:index_of_remove]

802

after_remove = commands[index_of_remove:]

803

cleaned_after = []

804

for cmd in after_remove:

805

if isinstance(cmd, command.Deny):

806

before_remove.append(cmd)

807

else:

808

cleaned_after.append(cmd)

809

if cleaned_after != after_remove:

810

commands = before_remove + cleaned_after

811

812

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

813

# optionally verbosely

814

if not commands:

815

commands.append(command.PrintTable(verbose=options.verbose))

816

817

return commands

818

819

820

class command:

821

"""A namespace for command classes"""

822

823

class Base:

824

"""Abstract base class for commands"""

825

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

826

"""Normal commands should implement run_on_one_client(),

827

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

828

override this run() method instead.

829

"""

830

self.bus = bus

831

for client, properties in clients.items():

832

self.run_on_one_client(client, properties)

833

834

835

class IsEnabled(Base):

836

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

837

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

838

if properties["Enabled"]:

839

sys.exit(0)

840

sys.exit(1)

841

842

843

class Approve(Base):

844

def run_on_one_client(self, client, properties):

845

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

846

847

848

class Deny(Base):

849

def run_on_one_client(self, client, properties):

850

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

851

852

853

class Remove(Base):

854

def run(self, clients, bus):

855

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

856

bus.call_server_method("RemoveClient", clientpath)

857

858

859

class Output(Base):

860

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

861

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

862

"Created", "Interval", "Host", "KeyID",

863

"Fingerprint", "CheckerRunning",

864

"LastEnabled", "ApprovalPending",

865

"ApprovedByDefault", "LastApprovalRequest",

866

"ApprovalDelay", "ApprovalDuration",

867

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

868

"LastCheckerStatus")

869

870

871

class DumpJSON(Output):

872

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

873

data = {properties["Name"]:

874

{key: properties[key]

875

for key in self.all_keywords}

876

for properties in clients.values()}

877

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

878

879

880

class PrintTable(Output):

881

def __init__(self, verbose=False):

882

self.verbose = verbose

883

884

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

885

default_keywords = ("Name", "Enabled", "Timeout",

886

"LastCheckedOK")

887

keywords = default_keywords

888

if self.verbose:

889

keywords = self.all_keywords

890

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

891

892

class TableOfClients:

893

tableheaders = {

894

"Name": "Name",

895

"Enabled": "Enabled",

896

"Timeout": "Timeout",

897

"LastCheckedOK": "Last Successful Check",

898

"LastApprovalRequest": "Last Approval Request",

899

"Created": "Created",

900

"Interval": "Interval",

901

"Host": "Host",

902

"Fingerprint": "Fingerprint",

903

"KeyID": "Key ID",

904

"CheckerRunning": "Check Is Running",

905

"LastEnabled": "Last Enabled",

906

"ApprovalPending": "Approval Is Pending",

907

"ApprovedByDefault": "Approved By Default",

908

"ApprovalDelay": "Approval Delay",

909

"ApprovalDuration": "Approval Duration",

910

"Checker": "Checker",

911

"ExtendedTimeout": "Extended Timeout",

912

"Expires": "Expires",

913

"LastCheckerStatus": "Last Checker Status",

914

}

915

916

def __init__(self, clients, keywords):

917

self.clients = clients

918

self.keywords = keywords

919

920

def __str__(self):

921

return "\n".join(self.rows())

922

923

if sys.version_info.major == 2:

924

__unicode__ = __str__

925

def __str__(self):

926

return str(self).encode(

927

locale.getpreferredencoding())

928

929

def rows(self):

930

format_string = self.row_formatting_string()

931

rows = [self.header_line(format_string)]

932

rows.extend(self.client_line(client, format_string)

933

for client in self.clients)

934

return rows

935

936

def row_formatting_string(self):

937

"Format string used to format table rows"

938

return " ".join("{{{key}:{width}}}".format(

939

width=max(len(self.tableheaders[key]),

940

*(len(self.string_from_client(client,

941

key))

942

for client in self.clients)),

943

key=key)

944

for key in self.keywords)

945

946

def string_from_client(self, client, key):

947

return self.valuetostring(client[key], key)

948

949

@classmethod

950

def valuetostring(cls, value, keyword):

951

if isinstance(value, bool):

952

return "Yes" if value else "No"

953

if keyword in ("Timeout", "Interval", "ApprovalDelay",

954

"ApprovalDuration", "ExtendedTimeout"):

955

return cls.milliseconds_to_string(value)

956

return str(value)

957

958

def header_line(self, format_string):

959

return format_string.format(**self.tableheaders)

960

961

def client_line(self, client, format_string):

962

return format_string.format(

963

**{key: self.string_from_client(client, key)

964

for key in self.keywords})

965

966

@staticmethod

967

def milliseconds_to_string(ms):

968

td = datetime.timedelta(0, 0, 0, ms)

969

return ("{days}{hours:02}:{minutes:02}:{seconds:02}"

970

.format(days="{}T".format(td.days)

971

if td.days else "",

972

hours=td.seconds // 3600,

973

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

974

seconds=td.seconds % 60))

975

976

977

class PropertySetter(Base):

978

"Abstract class for Actions for setting one client property"

979

980

def run_on_one_client(self, client, properties=None):

981

"""Set the Client's D-Bus property"""

982

self.bus.set_client_property(client, self.propname,

983

self.value_to_set)

984

985

@property

986

def propname(self):

987

raise NotImplementedError()

988

989

990

class Enable(PropertySetter):

991

propname = "Enabled"

992

value_to_set = True

993

994

995

class Disable(PropertySetter):

996

propname = "Enabled"

997

value_to_set = False

998

999

1000

class BumpTimeout(PropertySetter):

1001

propname = "LastCheckedOK"

1002

value_to_set = ""

1003

1004

1005

class StartChecker(PropertySetter):

1006

propname = "CheckerRunning"

1007

value_to_set = True

1008

1009

1010

class StopChecker(PropertySetter):

1011

propname = "CheckerRunning"

1012

value_to_set = False

1013

1014

1015

class ApproveByDefault(PropertySetter):

1016

propname = "ApprovedByDefault"

1017

value_to_set = True

1018

1019

1020

class DenyByDefault(PropertySetter):

1021

propname = "ApprovedByDefault"

1022

value_to_set = False

1023

1024

1025

class PropertySetterValue(PropertySetter):

1026

"""Abstract class for PropertySetter recieving a value as

1027

constructor argument instead of a class attribute."""

1028

def __init__(self, value):

1029

self.value_to_set = value

1030

1031

@classmethod

1032

def argparse(cls, argtype):

1033

def cmdtype(arg):

1034

return cls(argtype(arg))

1035

return cmdtype

1036

1037

class SetChecker(PropertySetterValue):

1038

propname = "Checker"

1039

1040

1041

class SetHost(PropertySetterValue):

1042

propname = "Host"

1043

1044

1045

class SetSecret(PropertySetterValue):

1046

propname = "Secret"

1047

1048

@property

1049

def value_to_set(self):

1050

return self._vts

1051

1052

@value_to_set.setter

1053

def value_to_set(self, value):

1054

"""When setting, read data from supplied file object"""

1055

self._vts = value.read()

1056

value.close()

1057

1058

1059

class PropertySetterValueMilliseconds(PropertySetterValue):

1060

"""Abstract class for PropertySetterValue taking a value

1061

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

1062

milliseconds."""

1063

1064

@property

1065

def value_to_set(self):

1066

return self._vts

1067

1068

@value_to_set.setter

1069

def value_to_set(self, value):

1070

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

1071

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

1072

1073

1074

class SetTimeout(PropertySetterValueMilliseconds):

1075

propname = "Timeout"

1076

1077

1078

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1079

propname = "ExtendedTimeout"

1080

1081

1082

class SetInterval(PropertySetterValueMilliseconds):

1083

propname = "Interval"

1084

1085

1086

class SetApprovalDelay(PropertySetterValueMilliseconds):

1087

propname = "ApprovalDelay"

1088

1089

1090

class SetApprovalDuration(PropertySetterValueMilliseconds):

1091

propname = "ApprovalDuration"

1092

1093

1094

1095

class TestCaseWithAssertLogs(unittest.TestCase):

1096

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

1097

1098

if not hasattr(unittest.TestCase, "assertLogs"):

1099

@contextlib.contextmanager

1100

def assertLogs(self, logger, level=logging.INFO):

1101

capturing_handler = self.CapturingLevelHandler(level)

1102

old_level = logger.level

1103

old_propagate = logger.propagate

1104

logger.addHandler(capturing_handler)

1105

logger.setLevel(level)

1106

logger.propagate = False

1107

try:

1108

yield capturing_handler.watcher

1109

finally:

1110

logger.propagate = old_propagate

1111

logger.removeHandler(capturing_handler)

1112

logger.setLevel(old_level)

1113

self.assertGreater(len(capturing_handler.watcher.records),

1114

0)

1115

1116

class CapturingLevelHandler(logging.Handler):

1117

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

1118

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

1119

self.watcher = self.LoggingWatcher([], [])

1120

def emit(self, record):

1121

self.watcher.records.append(record)

1122

self.watcher.output.append(self.format(record))

1123

1124

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1125

("records",

1126

"output"))

1127

1128

1129

class Unique:

1130

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

1131

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

1132

1133

1134

class Test_string_to_delta(TestCaseWithAssertLogs):

1135

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

1136

# rfc3339_duration_to_delta() already has more comprehensive

1137

# tests, which are run by doctest.

1138

1139

def test_rfc3339_zero_seconds(self):

1140

self.assertEqual(datetime.timedelta(),

1141

string_to_delta("PT0S"))

1142

1143

def test_rfc3339_zero_days(self):

1144

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

1145

1146

def test_rfc3339_one_second(self):

1147

self.assertEqual(datetime.timedelta(0, 1),

1148

string_to_delta("PT1S"))

1149

1150

def test_rfc3339_two_hours(self):

1151

self.assertEqual(datetime.timedelta(0, 7200),

1152

string_to_delta("PT2H"))

1153

1154

def test_falls_back_to_pre_1_6_1_with_warning(self):

1155

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

1156

value = string_to_delta("2h")

1157

self.assertEqual(datetime.timedelta(0, 7200), value)

1158

1159

1160

class Test_check_option_syntax(unittest.TestCase):

1161

def setUp(self):

1162

self.parser = argparse.ArgumentParser()

1163

add_command_line_options(self.parser)

1164

1165

def test_actions_requires_client_or_all(self):

1166

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

1167

args = self.actionargs(action, value)

1168

with self.assertParseError():

1169

self.parse_args(args)

1170

1171

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

1172

# check_option_syntax()

1173

actions = {

1174

"--enable": None,

1175

"--disable": None,

1176

"--bump-timeout": None,

1177

"--start-checker": None,

1178

"--stop-checker": None,

1179

"--is-enabled": None,

1180

"--remove": None,

1181

"--checker": "x",

1182

"--timeout": "PT0S",

1183

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

1184

"--interval": "PT0S",

1185

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

1186

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

1187

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

1188

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

1189

"--host": "hostname",

1190

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

1191

"--approve": None,

1192

"--deny": None,

1193

}

1194

1195

@staticmethod

1196

def actionargs(action, value, *args):

1197

if value is not None:

1198

return [action, value] + list(args)

1199

else:

1200

return [action] + list(args)

1201

1202

@contextlib.contextmanager

1203

def assertParseError(self):

1204

with self.assertRaises(SystemExit) as e:

1205

with self.redirect_stderr_to_devnull():

1206

yield

1207

# Exit code from argparse is guaranteed to be "2". Reference:

1208

# https://docs.python.org/3/library

1209

# /argparse.html#exiting-methods

1210

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

1211

1212

def parse_args(self, args):

1213

options = self.parser.parse_args(args)

1214

check_option_syntax(self.parser, options)

1215

1216

@staticmethod

1217

@contextlib.contextmanager

1218

def redirect_stderr_to_devnull():

1219

old_stderr = sys.stderr

1220

with contextlib.closing(open(os.devnull, "w")) as null:

1221

sys.stderr = null

1222

try:

1223

yield

1224

finally:

1225

sys.stderr = old_stderr

1226

1227

def check_option_syntax(self, options):

1228

check_option_syntax(self.parser, options)

1229

1230

def test_actions_all_conflicts_with_verbose(self):

1231

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

1232

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

1233

"--verbose")

1234

with self.assertParseError():

1235

self.parse_args(args)

1236

1237

def test_actions_with_client_conflicts_with_verbose(self):

1238

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

1239

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

1240

"client")

1241

with self.assertParseError():

1242

self.parse_args(args)

1243

1244

def test_dump_json_conflicts_with_verbose(self):

1245

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

1246

with self.assertParseError():

1247

self.parse_args(args)

1248

1249

def test_dump_json_conflicts_with_action(self):

1250

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

1251

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

1252

with self.assertParseError():

1253

self.parse_args(args)

1254

1255

def test_all_can_not_be_alone(self):

1256

args = ["--all"]

1257

with self.assertParseError():

1258

self.parse_args(args)

1259

1260

def test_all_is_ok_with_any_action(self):

1261

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

1262

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

1263

self.parse_args(args)

1264

1265

def test_any_action_is_ok_with_one_client(self):

1266

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

1267

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

1268

self.parse_args(args)

1269

1270

def test_one_client_with_all_actions_except_is_enabled(self):

1271

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

1272

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

1273

continue

1274

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

1275

self.parse_args(args)

1276

1277

def test_two_clients_with_all_actions_except_is_enabled(self):

1278

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

1279

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

1280

continue

1281

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

1282

"client2")

1283

self.parse_args(args)

1284

1285

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1286

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

1287

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

1288

continue

1289

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

1290

"client2")

1291

self.parse_args(args)

1292

1293

def test_is_enabled_fails_without_client(self):

1294

args = ["--is-enabled"]

1295

with self.assertParseError():

1296

self.parse_args(args)

1297

1298

def test_is_enabled_fails_with_two_clients(self):

1299

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

1300

with self.assertParseError():

1301

self.parse_args(args)

1302

1303

def test_remove_can_only_be_combined_with_action_deny(self):

1304

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

1305

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

1306

continue

1307

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

1308

"--remove")

1309

with self.assertParseError():

1310

self.parse_args(args)

1311

1312

1313

class Test_dbus_exceptions(unittest.TestCase):

1314

1315

def test_dbus_ConnectFailed_is_Error(self):

1316

with self.assertRaises(dbus.Error):

1317

raise dbus.ConnectFailed()

1318

1319

1320

class Test_dbus_MandosBus(unittest.TestCase):

1321

1322

class MockMandosBus(dbus.MandosBus):

1323

def __init__(self):

1324

self._name = "se.recompile.Mandos"

1325

self._server_path = "/"

1326

self._server_interface = "se.recompile.Mandos"

1327

self._client_interface = "se.recompile.Mandos.Client"

1328

self.calls = []

1329

self.call_method_return = Unique()

1330

1331

def call_method(self, methodname, busname, objectpath,

1332

interface, *args):

1333

self.calls.append((methodname, busname, objectpath,

1334

interface, args))

1335

return self.call_method_return

1336

1337

def setUp(self):

1338

self.bus = self.MockMandosBus()

1339

1340

def test_set_client_property(self):

1341

self.bus.set_client_property("objectpath", "key", "value")

1342

expected_call = ("Set", self.bus._name, "objectpath",

1343

"org.freedesktop.DBus.Properties",

1344

(self.bus._client_interface, "key", "value"))

1345

self.assertIn(expected_call, self.bus.calls)

1346

1347

def test_call_client_method(self):

1348

ret = self.bus.call_client_method("objectpath", "methodname")

1349

self.assertIs(self.bus.call_method_return, ret)

1350

expected_call = ("methodname", self.bus._name, "objectpath",

1351

self.bus._client_interface, ())

1352

self.assertIn(expected_call, self.bus.calls)

1353

1354

def test_call_client_method_with_args(self):

1355

args = (Unique(), Unique())

1356

ret = self.bus.call_client_method("objectpath", "methodname",

1357

*args)

1358

self.assertIs(self.bus.call_method_return, ret)

1359

expected_call = ("methodname", self.bus._name, "objectpath",

1360

self.bus._client_interface,

1361

(args[0], args[1]))

1362

self.assertIn(expected_call, self.bus.calls)

1363

1364

def test_get_clients_and_properties(self):

1365

managed_objects = {

1366

"objectpath": {

1367

self.bus._client_interface: {

1368

"key": "value",

1369

"bool": True,

1370

},

1371

"irrelevant_interface": {

1372

"key": "othervalue",

1373

"bool": False,

1374

},

1375

},

1376

"other_objectpath": {

1377

"other_irrelevant_interface": {

1378

"key": "value 3",

1379

"bool": None,

1380

},

1381

},

1382

}

1383

expected_clients_and_properties = {

1384

"objectpath": {

1385

"key": "value",

1386

"bool": True,

1387

}

1388

}

1389

self.bus.call_method_return = managed_objects

1390

ret = self.bus.get_clients_and_properties()

1391

self.assertDictEqual(expected_clients_and_properties, ret)

1392

expected_call = ("GetManagedObjects", self.bus._name,

1393

self.bus._server_path,

1394

"org.freedesktop.DBus.ObjectManager", ())

1395

self.assertIn(expected_call, self.bus.calls)

1396

1397

def test_call_server_method(self):

1398

ret = self.bus.call_server_method("methodname")

1399

self.assertIs(self.bus.call_method_return, ret)

1400

expected_call = ("methodname", self.bus._name,

1401

self.bus._server_path,

1402

self.bus._server_interface, ())

1403

self.assertIn(expected_call, self.bus.calls)

1404

1405

def test_call_server_method_with_args(self):

1406

args = (Unique(), Unique())

1407

ret = self.bus.call_server_method("methodname", *args)

1408

self.assertIs(self.bus.call_method_return, ret)

1409

expected_call = ("methodname", self.bus._name,

1410

self.bus._server_path,

1411

self.bus._server_interface,

1412

(args[0], args[1]))

1413

self.assertIn(expected_call, self.bus.calls)

1414

1415

1416

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1417

1418

def MockDBusPython_func(self, func):

1419

class mock_dbus_python:

1420

"""mock dbus-python module"""

1421

class exceptions:

1422

"""Pseudo-namespace"""

1423

class DBusException(Exception):

1424

pass

1425

class SystemBus:

1426

@staticmethod

1427

def get_object(busname, objectpath):

1428

DBusObject = collections.namedtuple(

1429

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

1430

def method(*args, **kwargs):

1431

self.assertEqual({"dbus_interface":

1432

"interface"},

1433

kwargs)

1434

return func(*args)

1435

def set_property(interface, key, value,

1436

dbus_interface=None):

1437

self.assertEqual(

1438

"org.freedesktop.DBus.Properties",

1439

dbus_interface)

1440

self.assertEqual("Secret", key)

1441

return func(interface, key, value,

1442

dbus_interface=dbus_interface)

1443

return DBusObject(methodname=method,

1444

Set=set_property)

1445

class Boolean:

1446

def __init__(self, value):

1447

self.value = bool(value)

1448

def __bool__(self):

1449

return self.value

1450

if sys.version_info.major == 2:

1451

__nonzero__ = __bool__

1452

class ObjectPath(str):

1453

pass

1454

class Dictionary(dict):

1455

pass

1456

class ByteArray(bytes):

1457

pass

1458

return mock_dbus_python

1459

1460

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

1461

interface, *args):

1462

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

1463

return bus.call_method(methodname, busname, objectpath,

1464

interface, *args)

1465

1466

def test_call_method_returns(self):

1467

expected_method_return = Unique()

1468

method_args = (Unique(), Unique())

1469

def func(*args):

1470

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

1471

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

1472

self.assertIs(marg, arg)

1473

return expected_method_return

1474

mock_dbus_python = self.MockDBusPython_func(func)

1475

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1476

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

1477

"objectpath", "interface",

1478

*method_args)

1479

self.assertIs(ret, expected_method_return)

1480

1481

def test_call_method_filters_bool_true(self):

1482

def func():

1483

return method_return

1484

mock_dbus_python = self.MockDBusPython_func(func)

1485

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1486

method_return = mock_dbus_python.Boolean(True)

1487

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

1488

"objectpath", "interface")

1489

self.assertTrue(ret)

1490

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1491

1492

def test_call_method_filters_bool_false(self):

1493

def func():

1494

return method_return

1495

mock_dbus_python = self.MockDBusPython_func(func)

1496

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1497

method_return = mock_dbus_python.Boolean(False)

1498

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

1499

"objectpath", "interface")

1500

self.assertFalse(ret)

1501

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1502

1503

def test_call_method_filters_objectpath(self):

1504

def func():

1505

return method_return

1506

mock_dbus_python = self.MockDBusPython_func(func)

1507

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1508

method_return = mock_dbus_python.ObjectPath("objectpath")

1509

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

1510

"objectpath", "interface")

1511

self.assertEqual("objectpath", ret)

1512

self.assertIsNot("objectpath", ret)

1513

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1514

1515

def test_call_method_filters_booleans_in_dict(self):

1516

def func():

1517

return method_return

1518

mock_dbus_python = self.MockDBusPython_func(func)

1519

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1520

method_return = mock_dbus_python.Dictionary(

1521

{mock_dbus_python.Boolean(True):

1522

mock_dbus_python.Boolean(False),

1523

mock_dbus_python.Boolean(False):

1524

mock_dbus_python.Boolean(True)})

1525

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

1526

"objectpath", "interface")

1527

expected_method_return = {True: False,

1528

False: True}

1529

self.assertEqual(expected_method_return, ret)

1530

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1531

1532

def test_call_method_filters_objectpaths_in_dict(self):

1533

def func():

1534

return method_return

1535

mock_dbus_python = self.MockDBusPython_func(func)

1536

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1537

method_return = mock_dbus_python.Dictionary(

1538

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1539

mock_dbus_python.ObjectPath("objectpath_value_1"),

1540

mock_dbus_python.ObjectPath("objectpath_key_2"):

1541

mock_dbus_python.ObjectPath("objectpath_value_2")})

1542

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

1543

"objectpath", "interface")

1544

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

1545

for key, value in

1546

method_return.items()}

1547

self.assertEqual(expected_method_return, ret)

1548

self.assertIsInstance(ret, dict)

1549

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1550

1551

def test_call_method_filters_dict_in_dict(self):

1552

def func():

1553

return method_return

1554

mock_dbus_python = self.MockDBusPython_func(func)

1555

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1556

method_return = mock_dbus_python.Dictionary(

1557

{"key1": mock_dbus_python.Dictionary({"key11": "value11",

1558

"key12": "value12"}),

1559

"key2": mock_dbus_python.Dictionary({"key21": "value21",

1560

"key22": "value22"})})

1561

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

1562

"objectpath", "interface")

1563

expected_method_return = {

1564

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

1565

"key12": "value12"},

1566

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

1567

"key22": "value22"},

1568

}

1569

self.assertEqual(expected_method_return, ret)

1570

self.assertIsInstance(ret, dict)

1571

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1572

for key, value in ret.items():

1573

self.assertIsInstance(value, dict)

1574

self.assertEqual(expected_method_return[key], value)

1575

self.assertNotIsInstance(value,

1576

mock_dbus_python.Dictionary)

1577

1578

def test_call_method_filters_dict_three_deep(self):

1579

def func():

1580

return method_return

1581

mock_dbus_python = self.MockDBusPython_func(func)

1582

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1583

method_return = mock_dbus_python.Dictionary(

1584

{"key1":

1585

mock_dbus_python.Dictionary(

1586

{"key2":

1587

mock_dbus_python.Dictionary(

1588

{"key3":

1589

mock_dbus_python.Boolean(True),

1590

}),

1591

}),

1592

})

1593

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

1594

"objectpath", "interface")

1595

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

1596

self.assertEqual(expected_method_return, ret)

1597

self.assertIsInstance(ret, dict)

1598

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1599

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

1600

self.assertNotIsInstance(ret["key1"],

1601

mock_dbus_python.Dictionary)

1602

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

1603

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

1604

mock_dbus_python.Dictionary)

1605

self.assertTrue(ret["key1"]["key2"]["key3"])

1606

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

1607

mock_dbus_python.Boolean)

1608

1609

def test_call_method_handles_exception(self):

1610

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

1611

1612

def func():

1613

dbus_logger.error("Test")

1614

raise mock_dbus_python.exceptions.DBusException()

1615

1616

mock_dbus_python = self.MockDBusPython_func(func)

1617

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1618

1619

class CountingHandler(logging.Handler):

1620

count = 0

1621

def emit(self, record):

1622

self.count += 1

1623

1624

counting_handler = CountingHandler()

1625

1626

dbus_logger.addHandler(counting_handler)

1627

1628

try:

1629

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

1630

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

1631

"objectpath", "interface")

1632

finally:

1633

dbus_logger.removeFilter(counting_handler)

1634

1635

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

1636

1637

# Make sure the dbus logger was suppressed

1638

self.assertEqual(0, counting_handler.count)

1639

1640

def test_Set_Secret_sends_bytearray(self):

1641

ret = [None]

1642

def func(*args, **kwargs):

1643

ret[0] = (args, kwargs)

1644

mock_dbus_python = self.MockDBusPython_func(func)

1645

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1646

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

1647

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

1648

mock_dbus_python.ByteArray(b"value")),

1649

{"dbus_interface":

1650

"org.freedesktop.DBus.Properties"})

1651

self.assertEqual(expected_call, ret[0])

1652

if sys.version_info.major == 2:

1653

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

1654

mock_dbus_python.ByteArray)

1655

1656

def test_get_object_converts_to_correct_exception(self):

1657

bus = dbus_python_adapter.SystemBus(

1658

self.fake_dbus_python_raises_exception_on_connect)

1659

with self.assertRaises(dbus.ConnectFailed):

1660

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

1661

"objectpath", "interface")

1662

1663

class fake_dbus_python_raises_exception_on_connect:

1664

"""fake dbus-python module"""

1665

class exceptions:

1666

"""Pseudo-namespace"""

1667

class DBusException(Exception):

1668

pass

1669

1670

@classmethod

1671

def SystemBus(cls):

1672

def get_object(busname, objectpath):

1673

raise cls.exceptions.DBusException()

1674

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

1675

return Bus(get_object=get_object)

1676

1677

1678

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1679

class mock_dbus_python:

1680

"""mock dbus-python modules"""

1681

class SystemBus:

1682

@staticmethod

1683

def get_object(busname, objectpath):

1684

return Unique()

1685

1686

def setUp(self):

1687

self.bus = dbus_python_adapter.CachingBus(

1688

self.mock_dbus_python)

1689

1690

def test_returns_distinct_objectpaths(self):

1691

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

1692

self.assertIsInstance(obj1, Unique)

1693

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

1694

self.assertIsInstance(obj2, Unique)

1695

self.assertIsNot(obj1, obj2)

1696

1697

def test_returns_distinct_busnames(self):

1698

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

1699

self.assertIsInstance(obj1, Unique)

1700

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

1701

self.assertIsInstance(obj2, Unique)

1702

self.assertIsNot(obj1, obj2)

1703

1704

def test_returns_distinct_both(self):

1705

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

1706

self.assertIsInstance(obj1, Unique)

1707

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

1708

self.assertIsInstance(obj2, Unique)

1709

self.assertIsNot(obj1, obj2)

1710

1711

def test_returns_same(self):

1712

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

1713

self.assertIsInstance(obj1, Unique)

1714

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

1715

self.assertIsInstance(obj2, Unique)

1716

self.assertIs(obj1, obj2)

1717

1718

def test_returns_same_old(self):

1719

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

1720

self.assertIsInstance(obj1, Unique)

1721

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

1722

self.assertIsInstance(obj2, Unique)

1723

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

1724

self.assertIsInstance(obj1b, Unique)

1725

self.assertIsNot(obj1, obj2)

1726

self.assertIsNot(obj2, obj1b)

1727

self.assertIs(obj1, obj1b)

1728

1729

1730

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1731

1732

def Stub_pydbus_func(self, func):

1733

class stub_pydbus:

1734

"""stub pydbus module"""

1735

class SystemBus:

1736

@staticmethod

1737

def get(busname, objectpath):

1738

DBusObject = collections.namedtuple(

1739

"DBusObject", ("methodname",))

1740

return {"interface":

1741

DBusObject(methodname=func)}

1742

return stub_pydbus

1743

1744

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

1745

interface, *args):

1746

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

1747

return bus.call_method(methodname, busname, objectpath,

1748

interface, *args)

1749

1750

def test_call_method_returns(self):

1751

expected_method_return = Unique()

1752

method_args = (Unique(), Unique())

1753

def func(*args):

1754

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

1755

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

1756

self.assertIs(marg, arg)

1757

return expected_method_return

1758

stub_pydbus = self.Stub_pydbus_func(func)

1759

bus = pydbus_adapter.SystemBus(stub_pydbus)

1760

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

1761

"objectpath", "interface",

1762

*method_args)

1763

self.assertIs(ret, expected_method_return)

1764

1765

def test_call_method_handles_exception(self):

1766

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

1767

1768

def func():

1769

raise gi.repository.GLib.Error()

1770

1771

stub_pydbus = self.Stub_pydbus_func(func)

1772

bus = pydbus_adapter.SystemBus(stub_pydbus)

1773

1774

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

1775

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

1776

"objectpath", "interface")

1777

1778

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

1779

1780

def test_get_converts_to_correct_exception(self):

1781

bus = pydbus_adapter.SystemBus(

1782

self.fake_pydbus_raises_exception_on_connect)

1783

with self.assertRaises(dbus.ConnectFailed):

1784

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

1785

"objectpath", "interface")

1786

1787

class fake_pydbus_raises_exception_on_connect:

1788

"""fake dbus-python module"""

1789

@classmethod

1790

def SystemBus(cls):

1791

def get(busname, objectpath):

1792

raise gi.repository.GLib.Error()

1793

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

1794

return Bus(get=get)

1795

1796

def test_set_property_uses_setattr(self):

1797

class Object:

1798

pass

1799

obj = Object()

1800

class pydbus_spy:

1801

class SystemBus:

1802

@staticmethod

1803

def get(busname, objectpath):

1804

return {"interface": obj}

1805

bus = pydbus_adapter.SystemBus(pydbus_spy)

1806

value = Unique()

1807

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

1808

value)

1809

self.assertIs(value, obj.key)

1810

1811

def test_get_suppresses_xml_deprecation_warning(self):

1812

if sys.version_info.major >= 3:

1813

return

1814

class stub_pydbus_get:

1815

class SystemBus:

1816

@staticmethod

1817

def get(busname, objectpath):

1818

warnings.warn_explicit(

1819

"deprecated", DeprecationWarning,

1820

"xml.etree.ElementTree", 0)

1821

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1822

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

1823

warnings.simplefilter("always")

1824

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

1825

self.assertEqual(0, len(w))

1826

1827

1828

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1829

class stub_pydbus:

1830

"""stub pydbus module"""

1831

class SystemBus:

1832

@staticmethod

1833

def get(busname, objectpath):

1834

return Unique()

1835

1836

def setUp(self):

1837

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

1838

1839

def test_returns_distinct_objectpaths(self):

1840

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

1841

self.assertIsInstance(obj1, Unique)

1842

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

1843

self.assertIsInstance(obj2, Unique)

1844

self.assertIsNot(obj1, obj2)

1845

1846

def test_returns_distinct_busnames(self):

1847

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

1848

self.assertIsInstance(obj1, Unique)

1849

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

1850

self.assertIsInstance(obj2, Unique)

1851

self.assertIsNot(obj1, obj2)

1852

1853

def test_returns_distinct_both(self):

1854

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

1855

self.assertIsInstance(obj1, Unique)

1856

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

1857

self.assertIsInstance(obj2, Unique)

1858

self.assertIsNot(obj1, obj2)

1859

1860

def test_returns_same(self):

1861

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

1862

self.assertIsInstance(obj1, Unique)

1863

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

1864

self.assertIsInstance(obj2, Unique)

1865

self.assertIs(obj1, obj2)

1866

1867

def test_returns_same_old(self):

1868

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

1869

self.assertIsInstance(obj1, Unique)

1870

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

1871

self.assertIsInstance(obj2, Unique)

1872

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

1873

self.assertIsInstance(obj1b, Unique)

1874

self.assertIsNot(obj1, obj2)

1875

self.assertIsNot(obj2, obj1b)

1876

self.assertIs(obj1, obj1b)

1877

1878

1879

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1880

1881

class dummy_dbussy:

1882

class DBUS:

1883

class ObjectPath(str):

1884

pass

1885

class DBusError(Exception):

1886

pass

1887

1888

def fake_ravel_func(self, func):

1889

class fake_ravel:

1890

@staticmethod

1891

def system_bus():

1892

class DBusInterfaceProxy:

1893

@staticmethod

1894

def methodname(*args):

1895

return [func(*args)]

1896

class DBusObject:

1897

@staticmethod

1898

def get_interface(interface):

1899

if interface == "interface":

1900

return DBusInterfaceProxy()

1901

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

1902

return fake_ravel

1903

1904

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

1905

interface, *args):

1906

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

1907

return bus.call_method(methodname, busname, objectpath,

1908

interface, *args)

1909

1910

def test_call_method_returns(self):

1911

expected_method_return = Unique()

1912

method_args = (Unique(), Unique())

1913

def func(*args):

1914

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

1915

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

1916

self.assertIs(marg, arg)

1917

return expected_method_return

1918

fake_ravel = self.fake_ravel_func(func)

1919

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

1920

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

1921

"objectpath", "interface",

1922

*method_args)

1923

self.assertIs(ret, expected_method_return)

1924

1925

def test_call_method_filters_objectpath(self):

1926

def func():

1927

return method_return

1928

fake_ravel = self.fake_ravel_func(func)

1929

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

1930

method_return = (self.dummy_dbussy.DBUS

1931

.ObjectPath("objectpath"))

1932

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

1933

"objectpath", "interface")

1934

self.assertEqual("objectpath", ret)

1935

self.assertNotIsInstance(ret,

1936

self.dummy_dbussy.DBUS.ObjectPath)

1937

1938

def test_call_method_filters_objectpaths_in_dict(self):

1939

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1940

def func():

1941

return method_return

1942

fake_ravel = self.fake_ravel_func(func)

1943

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

1944

method_return = {

1945

ObjectPath("objectpath_key_1"):

1946

ObjectPath("objectpath_value_1"),

1947

ObjectPath("objectpath_key_2"):

1948

ObjectPath("objectpath_value_2"),

1949

}

1950

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

1951

"objectpath", "interface")

1952

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

1953

for key, value in

1954

method_return.items()}

1955

for key, value in ret.items():

1956

self.assertNotIsInstance(key, ObjectPath)

1957

self.assertNotIsInstance(value, ObjectPath)

1958

self.assertEqual(expected_method_return, ret)

1959

self.assertIsInstance(ret, dict)

1960

1961

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1962

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1963

def func():

1964

return method_return

1965

fake_ravel = self.fake_ravel_func(func)

1966

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

1967

method_return = {

1968

ObjectPath("key1"): {

1969

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

1970

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

1971

},

1972

ObjectPath("key2"): {

1973

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

1974

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

1975

},

1976

}

1977

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

1978

"objectpath", "interface")

1979

expected_method_return = {

1980

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

1981

"key12": "value12"},

1982

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

1983

"key22": "value22"},

1984

}

1985

self.assertEqual(expected_method_return, ret)

1986

for key, value in ret.items():

1987

self.assertIsInstance(value, dict)

1988

self.assertEqual(expected_method_return[key], value)

1989

self.assertNotIsInstance(key, ObjectPath)

1990

for inner_key, inner_value in value.items():

1991

self.assertIsInstance(value, dict)

1992

self.assertEqual(

1993

expected_method_return[key][inner_key],

1994

inner_value)

1995

self.assertNotIsInstance(key, ObjectPath)

1996

1997

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1998

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1999

def func():

2000

return method_return

2001

fake_ravel = self.fake_ravel_func(func)

2002

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

2003

method_return = {

2004

ObjectPath("key1"): {

2005

ObjectPath("key2"): {

2006

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

2007

},

2008

},

2009

}

2010

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

2011

"objectpath", "interface")

2012

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

2013

self.assertEqual(expected_method_return, ret)

2014

self.assertIsInstance(ret, dict)

2015

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

2016

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

2017

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

2018

ObjectPath)

2019

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

2020

self.assertNotIsInstance(

2021

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

2022

ObjectPath)

2023

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

2024

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

2025

self.dummy_dbussy.DBUS.ObjectPath)

2026

2027

def test_call_method_handles_exception(self):

2028

def func():

2029

raise self.dummy_dbussy.DBusError()

2030

2031

fake_ravel = self.fake_ravel_func(func)

2032

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

2033

2034

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

2035

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

2036

"objectpath", "interface")

2037

2038

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

2039

2040

def test_get_object_converts_to_correct_exception(self):

2041

class fake_ravel_raises_exception_on_connect:

2042

@staticmethod

2043

def system_bus():

2044

class Bus:

2045

@staticmethod

2046

def __getitem__(key):

2047

if key == "objectpath":

2048

raise self.dummy_dbussy.DBusError()

2049

raise Exception(key)

2050

return {"busname": Bus()}

2051

def func():

2052

raise self.dummy_dbussy.DBusError()

2053

bus = dbussy_adapter.SystemBus(

2054

self.dummy_dbussy,

2055

fake_ravel_raises_exception_on_connect)

2056

with self.assertRaises(dbus.ConnectFailed):

2057

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

2058

"objectpath", "interface")

2059

2060

2061

class Test_commands_from_options(unittest.TestCase):

2062

2063

def setUp(self):

2064

self.parser = argparse.ArgumentParser()

2065

add_command_line_options(self.parser)

2066

2067

def test_is_enabled(self):

2068

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

2069

command.IsEnabled)

2070

2071

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

2072

clients=None, **cmd_attrs):

2073

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

2074

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

2075

options = self.parser.parse_args(args)

2076

check_option_syntax(self.parser, options)

2077

commands = commands_from_options(options)

2078

self.assertEqual(length, len(commands))

2079

for command in commands:

2080

if isinstance(command, command_cls):

2081

break

2082

else:

2083

self.assertIsInstance(command, command_cls)

2084

if clients is not None:

2085

self.assertEqual(clients, options.client)

2086

for key, value in cmd_attrs.items():

2087

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

2088

2089

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

2090

for cmd in commands:

2091

self.assert_command_from_args(args, cmd,

2092

length=len(commands),

2093

clients=clients)

2094

2095

def test_is_enabled_short(self):

2096

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

2097

command.IsEnabled)

2098

2099

def test_approve(self):

2100

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

2101

command.Approve)

2102

2103

def test_approve_short(self):

2104

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

2105

command.Approve)

2106

2107

def test_deny(self):

2108

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

2109

command.Deny)

2110

2111

def test_deny_short(self):

2112

self.assert_command_from_args(["-D", "client"], command.Deny)

2113

2114

def test_remove(self):

2115

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

2116

command.Remove)

2117

2118

def test_deny_before_remove(self):

2119

options = self.parser.parse_args(["--deny", "--remove",

2120

"client"])

2121

check_option_syntax(self.parser, options)

2122

commands = commands_from_options(options)

2123

self.assertEqual(2, len(commands))

2124

self.assertIsInstance(commands[0], command.Deny)

2125

self.assertIsInstance(commands[1], command.Remove)

2126

2127

def test_deny_before_remove_reversed(self):

2128

options = self.parser.parse_args(["--remove", "--deny",

2129

"--all"])

2130

check_option_syntax(self.parser, options)

2131

commands = commands_from_options(options)

2132

self.assertEqual(2, len(commands))

2133

self.assertIsInstance(commands[0], command.Deny)

2134

self.assertIsInstance(commands[1], command.Remove)

2135

2136

def test_remove_short(self):

2137

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

2138

command.Remove)

2139

2140

def test_dump_json(self):

2141

self.assert_command_from_args(["--dump-json"],

2142

command.DumpJSON)

2143

2144

def test_enable(self):

2145

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

2146

command.Enable)

2147

2148

def test_enable_short(self):

2149

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

2150

command.Enable)

2151

2152

def test_disable(self):

2153

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

2154

command.Disable)

2155

2156

def test_disable_short(self):

2157

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

2158

command.Disable)

2159

2160

def test_bump_timeout(self):

2161

self.assert_command_from_args(["--bump-timeout", "client"],

2162

command.BumpTimeout)

2163

2164

def test_bump_timeout_short(self):

2165

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

2166

command.BumpTimeout)

2167

2168

def test_start_checker(self):

2169

self.assert_command_from_args(["--start-checker", "client"],

2170

command.StartChecker)

2171

2172

def test_stop_checker(self):

2173

self.assert_command_from_args(["--stop-checker", "client"],

2174

command.StopChecker)

2175

2176

def test_approve_by_default(self):

2177

self.assert_command_from_args(["--approve-by-default",

2178

"client"],

2179

command.ApproveByDefault)

2180

2181

def test_deny_by_default(self):

2182

self.assert_command_from_args(["--deny-by-default", "client"],

2183

command.DenyByDefault)

2184

2185

def test_checker(self):

2186

self.assert_command_from_args(["--checker", ":", "client"],

2187

command.SetChecker,

2188

value_to_set=":")

2189

2190

def test_checker_empty(self):

2191

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

2192

command.SetChecker,

2193

value_to_set="")

2194

2195

def test_checker_short(self):

2196

self.assert_command_from_args(["-c", ":", "client"],

2197

command.SetChecker,

2198

value_to_set=":")

2199

2200

def test_host(self):

2201

self.assert_command_from_args(

2202

["--host", "client.example.org", "client"],

2203

command.SetHost, value_to_set="client.example.org")

2204

2205

def test_host_short(self):

2206

self.assert_command_from_args(

2207

["-H", "client.example.org", "client"], command.SetHost,

2208

value_to_set="client.example.org")

2209

2210

def test_secret_devnull(self):

2211

self.assert_command_from_args(["--secret", os.path.devnull,

2212

"client"], command.SetSecret,

2213

value_to_set=b"")

2214

2215

def test_secret_tempfile(self):

2216

with tempfile.NamedTemporaryFile(mode="r+b") as f:

2217

value = b"secret\0xyzzy\nbar"

2218

f.write(value)

2219

f.seek(0)

2220

self.assert_command_from_args(["--secret", f.name,

2221

"client"],

2222

command.SetSecret,

2223

value_to_set=value)

2224

2225

def test_secret_devnull_short(self):

2226

self.assert_command_from_args(["-s", os.path.devnull,

2227

"client"], command.SetSecret,

2228

value_to_set=b"")

2229

2230

def test_secret_tempfile_short(self):

2231

with tempfile.NamedTemporaryFile(mode="r+b") as f:

2232

value = b"secret\0xyzzy\nbar"

2233

f.write(value)

2234

f.seek(0)

2235

self.assert_command_from_args(["-s", f.name, "client"],

2236

command.SetSecret,

2237

value_to_set=value)

2238

2239

def test_timeout(self):

2240

self.assert_command_from_args(["--timeout", "PT5M", "client"],

2241

command.SetTimeout,

2242

value_to_set=300000)

2243

2244

def test_timeout_short(self):

2245

self.assert_command_from_args(["-t", "PT5M", "client"],

2246

command.SetTimeout,

2247

value_to_set=300000)

2248

2249

def test_extended_timeout(self):

2250

self.assert_command_from_args(["--extended-timeout", "PT15M",

2251

"client"],

2252

command.SetExtendedTimeout,

2253

value_to_set=900000)

2254

2255

def test_interval(self):

2256

self.assert_command_from_args(["--interval", "PT2M",

2257

"client"], command.SetInterval,

2258

value_to_set=120000)

2259

2260

def test_interval_short(self):

2261

self.assert_command_from_args(["-i", "PT2M", "client"],

2262

command.SetInterval,

2263

value_to_set=120000)

2264

2265

def test_approval_delay(self):

2266

self.assert_command_from_args(["--approval-delay", "PT30S",

2267

"client"],

2268

command.SetApprovalDelay,

2269

value_to_set=30000)

2270

2271

def test_approval_duration(self):

2272

self.assert_command_from_args(["--approval-duration", "PT1S",

2273

"client"],

2274

command.SetApprovalDuration,

2275

value_to_set=1000)

2276

2277

def test_print_table(self):

2278

self.assert_command_from_args([], command.PrintTable,

2279

verbose=False)

2280

2281

def test_print_table_verbose(self):

2282

self.assert_command_from_args(["--verbose"],

2283

command.PrintTable,

2284

verbose=True)

2285

2286

def test_print_table_verbose_short(self):

2287

self.assert_command_from_args(["-v"], command.PrintTable,

2288

verbose=True)

2289

2290

2291

def test_manual_page_example_1(self):

2292

self.assert_command_from_args("",

2293

command.PrintTable,

2294

clients=[],

2295

verbose=False)

2296

2297

def test_manual_page_example_2(self):

2298

self.assert_command_from_args(

2299

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

2300

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

2301

"foo2.example.org"],

2302

verbose=True)

2303

2304

def test_manual_page_example_3(self):

2305

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

2306

command.Enable,

2307

clients=[])

2308

2309

def test_manual_page_example_4(self):

2310

self.assert_commands_from_args(

2311

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

2312

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

2313

[command.SetTimeout, command.SetInterval],

2314

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

2315

2316

def test_manual_page_example_5(self):

2317

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

2318

command.Approve,

2319

clients=[])

2320

2321

2322

class TestCommand(unittest.TestCase):

2323

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

2324

2325

class FakeMandosBus(dbus.MandosBus):

2326

def __init__(self, testcase):

2327

self.client_properties = {

2328

"Name": "foo",

2329

"KeyID": ("92ed150794387c03ce684574b1139a65"

2330

"94a34f895daaaf09fd8ea90a27cddb12"),

2331

"Secret": b"secret",

2332

"Host": "foo.example.org",

2333

"Enabled": True,

2334

"Timeout": 300000,

2335

"LastCheckedOK": "2019-02-03T00:00:00",

2336

"Created": "2019-01-02T00:00:00",

2337

"Interval": 120000,

2338

"Fingerprint": ("778827225BA7DE539C5A"

2339

"7CFA59CFF7CDBD9A5920"),

2340

"CheckerRunning": False,

2341

"LastEnabled": "2019-01-03T00:00:00",

2342

"ApprovalPending": False,

2343

"ApprovedByDefault": True,

2344

"LastApprovalRequest": "",

2345

"ApprovalDelay": 0,

2346

"ApprovalDuration": 1000,

2347

"Checker": "fping -q -- %(host)s",

2348

"ExtendedTimeout": 900000,

2349

"Expires": "2019-02-04T00:00:00",

2350

"LastCheckerStatus": 0,

2351

}

2352

self.other_client_properties = {

2353

"Name": "barbar",

2354

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2355

"6ab612cff5ad227247e46c2b020f441c"),

2356

"Secret": b"secretbar",

2357

"Host": "192.0.2.3",

2358

"Enabled": True,

2359

"Timeout": 300000,

2360

"LastCheckedOK": "2019-02-04T00:00:00",

2361

"Created": "2019-01-03T00:00:00",

2362

"Interval": 120000,

2363

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2364

"F547B3A107558FCA3A27"),

2365

"CheckerRunning": True,

2366

"LastEnabled": "2019-01-04T00:00:00",

2367

"ApprovalPending": False,

2368

"ApprovedByDefault": False,

2369

"LastApprovalRequest": "2019-01-03T00:00:00",

2370

"ApprovalDelay": 30000,

2371

"ApprovalDuration": 93785000,

2372

"Checker": ":",

2373

"ExtendedTimeout": 900000,

2374

"Expires": "2019-02-05T00:00:00",

2375

"LastCheckerStatus": -2,

2376

}

2377

self.clients = collections.OrderedDict(

2378

[

2379

("client_objectpath", self.client_properties),

2380

("other_client_objectpath",

2381

self.other_client_properties),

2382

])

2383

self.one_client = {"client_objectpath":

2384

self.client_properties}

2385

self.testcase = testcase

2386

self.calls = []

2387

2388

def call_method(self, methodname, busname, objectpath,

2389

interface, *args):

2390

self.testcase.assertEqual("se.recompile.Mandos", busname)

2391

self.calls.append((methodname, busname, objectpath,

2392

interface, args))

2393

if interface == "org.freedesktop.DBus.Properties":

2394

if methodname == "Set":

2395

self.testcase.assertEqual(3, len(args))

2396

interface, key, value = args

2397

self.testcase.assertEqual(

2398

"se.recompile.Mandos.Client", interface)

2399

self.clients[objectpath][key] = value

2400

return

2401

elif interface == "se.recompile.Mandos":

2402

self.testcase.assertEqual("RemoveClient", methodname)

2403

self.testcase.assertEqual(1, len(args))

2404

clientpath = args[0]

2405

del self.clients[clientpath]

2406

return

2407

elif interface == "se.recompile.Mandos.Client":

2408

if methodname == "Approve":

2409

self.testcase.assertEqual(1, len(args))

2410

return

2411

raise ValueError()

2412

2413

def setUp(self):

2414

self.bus = self.FakeMandosBus(self)

2415

2416

2417

class TestBaseCommands(TestCommand):

2418

2419

def test_IsEnabled_exits_successfully(self):

2420

with self.assertRaises(SystemExit) as e:

2421

command.IsEnabled().run(self.bus.one_client)

2422

if e.exception.code is not None:

2423

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

2424

else:

2425

self.assertIsNone(e.exception.code)

2426

2427

def test_IsEnabled_exits_with_failure(self):

2428

self.bus.client_properties["Enabled"] = False

2429

with self.assertRaises(SystemExit) as e:

2430

command.IsEnabled().run(self.bus.one_client)

2431

if isinstance(e.exception.code, int):

2432

self.assertNotEqual(0, e.exception.code)

2433

else:

2434

self.assertIsNotNone(e.exception.code)

2435

2436

def test_Approve(self):

2437

busname = "se.recompile.Mandos"

2438

client_interface = "se.recompile.Mandos.Client"

2439

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

2440

for clientpath in self.bus.clients:

2441

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

2442

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

2443

2444

def test_Deny(self):

2445

busname = "se.recompile.Mandos"

2446

client_interface = "se.recompile.Mandos.Client"

2447

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

2448

for clientpath in self.bus.clients:

2449

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

2450

client_interface, (False,)),

2451

self.bus.calls)

2452

2453

def test_Remove(self):

2454

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

2455

for clientpath in self.bus.clients:

2456

self.assertIn(("RemoveClient", dbus_busname,

2457

dbus_server_path, dbus_server_interface,

2458

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

2459

2460

expected_json = {

2461

"foo": {

2462

"Name": "foo",

2463

"KeyID": ("92ed150794387c03ce684574b1139a65"

2464

"94a34f895daaaf09fd8ea90a27cddb12"),

2465

"Host": "foo.example.org",

2466

"Enabled": True,

2467

"Timeout": 300000,

2468

"LastCheckedOK": "2019-02-03T00:00:00",

2469

"Created": "2019-01-02T00:00:00",

2470

"Interval": 120000,

2471

"Fingerprint": ("778827225BA7DE539C5A"

2472

"7CFA59CFF7CDBD9A5920"),

2473

"CheckerRunning": False,

2474

"LastEnabled": "2019-01-03T00:00:00",

2475

"ApprovalPending": False,

2476

"ApprovedByDefault": True,

2477

"LastApprovalRequest": "",

2478

"ApprovalDelay": 0,

2479

"ApprovalDuration": 1000,

2480

"Checker": "fping -q -- %(host)s",

2481

"ExtendedTimeout": 900000,

2482

"Expires": "2019-02-04T00:00:00",

2483

"LastCheckerStatus": 0,

2484

},

2485

"barbar": {

2486

"Name": "barbar",

2487

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2488

"6ab612cff5ad227247e46c2b020f441c"),

2489

"Host": "192.0.2.3",

2490

"Enabled": True,

2491

"Timeout": 300000,

2492

"LastCheckedOK": "2019-02-04T00:00:00",

2493

"Created": "2019-01-03T00:00:00",

2494

"Interval": 120000,

2495

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2496

"F547B3A107558FCA3A27"),

2497

"CheckerRunning": True,

2498

"LastEnabled": "2019-01-04T00:00:00",

2499

"ApprovalPending": False,

2500

"ApprovedByDefault": False,

2501

"LastApprovalRequest": "2019-01-03T00:00:00",

2502

"ApprovalDelay": 30000,

2503

"ApprovalDuration": 93785000,

2504

"Checker": ":",

2505

"ExtendedTimeout": 900000,

2506

"Expires": "2019-02-05T00:00:00",

2507

"LastCheckerStatus": -2,

2508

},

2509

}

2510

2511

def test_DumpJSON_normal(self):

2512

with self.capture_stdout_to_buffer() as buffer:

2513

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

2514

json_data = json.loads(buffer.getvalue())

2515

self.assertDictEqual(self.expected_json, json_data)

2516

2517

@staticmethod

2518

@contextlib.contextmanager

2519

def capture_stdout_to_buffer():

2520

capture_buffer = io.StringIO()

2521

old_stdout = sys.stdout

2522

sys.stdout = capture_buffer

2523

try:

2524

yield capture_buffer

2525

finally:

2526

sys.stdout = old_stdout

2527

2528

def test_DumpJSON_one_client(self):

2529

with self.capture_stdout_to_buffer() as buffer:

2530

command.DumpJSON().run(self.bus.one_client)

2531

json_data = json.loads(buffer.getvalue())

2532

expected_json = {"foo": self.expected_json["foo"]}

2533

self.assertDictEqual(expected_json, json_data)

2534

2535

def test_PrintTable_normal(self):

2536

with self.capture_stdout_to_buffer() as buffer:

2537

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

2538

expected_output = "\n".join((

2539

"Name Enabled Timeout Last Successful Check",

2540

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

2541

"barbar Yes 00:05:00 2019-02-04T00:00:00 ",

2542

)) + "\n"

2543

self.assertEqual(expected_output, buffer.getvalue())

2544

2545

def test_PrintTable_verbose(self):

2546

with self.capture_stdout_to_buffer() as buffer:

2547

command.PrintTable(verbose=True).run(self.bus.clients)

2548

columns = (

2549

(

2550

"Name ",

2551

"foo ",

2552

"barbar ",

2553

),(

2554

"Enabled ",

2555

"Yes ",

2556

"Yes ",

2557

),(

2558

"Timeout ",

2559

"00:05:00 ",

2560

"00:05:00 ",

2561

),(

2562

"Last Successful Check ",

2563

"2019-02-03T00:00:00 ",

2564

"2019-02-04T00:00:00 ",

2565

),(

2566

"Created ",

2567

"2019-01-02T00:00:00 ",

2568

"2019-01-03T00:00:00 ",

2569

),(

2570

"Interval ",

2571

"00:02:00 ",

2572

"00:02:00 ",

2573

),(

2574

"Host ",

2575

"foo.example.org ",

2576

"192.0.2.3 ",

2577

),(

2578

("Key ID "

2579

" "),

2580

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2581

"ea90a27cddb12 "),

2582

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2583

"46c2b020f441c "),

2584

),(

2585

"Fingerprint ",

2586

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2587

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2588

),(

2589

"Check Is Running ",

2590

"No ",

2591

"Yes ",

2592

),(

2593

"Last Enabled ",

2594

"2019-01-03T00:00:00 ",

2595

"2019-01-04T00:00:00 ",

2596

),(

2597

"Approval Is Pending ",

2598

"No ",

2599

"No ",

2600

),(

2601

"Approved By Default ",

2602

"Yes ",

2603

"No ",

2604

),(

2605

"Last Approval Request ",

2606

" ",

2607

"2019-01-03T00:00:00 ",

2608

),(

2609

"Approval Delay ",

2610

"00:00:00 ",

2611

"00:00:30 ",

2612

),(

2613

"Approval Duration ",

2614

"00:00:01 ",

2615

"1T02:03:05 ",

2616

),(

2617

"Checker ",

2618

"fping -q -- %(host)s ",

2619

": ",

2620

),(

2621

"Extended Timeout ",

2622

"00:15:00 ",

2623

"00:15:00 ",

2624

),(

2625

"Expires ",

2626

"2019-02-04T00:00:00 ",

2627

"2019-02-05T00:00:00 ",

2628

),(

2629

"Last Checker Status",

2630

"0 ",

2631

"-2 ",

2632

)

2633

)

2634

num_lines = max(len(rows) for rows in columns)

2635

expected_output = ("\n".join("".join(rows[line]

2636

for rows in columns)

2637

for line in range(num_lines))

2638

+ "\n")

2639

self.assertEqual(expected_output, buffer.getvalue())

2640

2641

def test_PrintTable_one_client(self):

2642

with self.capture_stdout_to_buffer() as buffer:

2643

command.PrintTable().run(self.bus.one_client)

2644

expected_output = "\n".join((

2645

"Name Enabled Timeout Last Successful Check",

2646

"foo Yes 00:05:00 2019-02-03T00:00:00 ",

2647

)) + "\n"

2648

self.assertEqual(expected_output, buffer.getvalue())

2649

2650

2651

class TestPropertySetterCmd(TestCommand):

2652

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

2653

2654

def runTest(self):

2655

if not hasattr(self, "command"):

2656

return # Abstract TestCase class

2657

2658

if hasattr(self, "values_to_set"):

2659

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

2660

values_to_get = getattr(self, "values_to_get",

2661

self.values_to_set)

2662

else:

2663

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

2664

values_to_get = self.values_to_get

2665

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

2666

for clientpath in self.bus.clients:

2667

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

2668

Unique())

2669

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

2670

for clientpath in self.bus.clients:

2671

value = (self.bus.clients[clientpath]

2672

[self.propname])

2673

self.assertNotIsInstance(value, Unique)

2674

self.assertEqual(value_to_get, value)

2675

2676

2677

class TestEnableCmd(TestPropertySetterCmd):

2678

command = command.Enable

2679

propname = "Enabled"

2680

values_to_get = [True]

2681

2682

2683

class TestDisableCmd(TestPropertySetterCmd):

2684

command = command.Disable

2685

propname = "Enabled"

2686

values_to_get = [False]

2687

2688

2689

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2690

command = command.BumpTimeout

2691

propname = "LastCheckedOK"

2692

values_to_get = [""]

2693

2694

2695

class TestStartCheckerCmd(TestPropertySetterCmd):

2696

command = command.StartChecker

2697

propname = "CheckerRunning"

2698

values_to_get = [True]

2699

2700

2701

class TestStopCheckerCmd(TestPropertySetterCmd):

2702

command = command.StopChecker

2703

propname = "CheckerRunning"

2704

values_to_get = [False]

2705

2706

2707

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2708

command = command.ApproveByDefault

2709

propname = "ApprovedByDefault"

2710

values_to_get = [True]

2711

2712

2713

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2714

command = command.DenyByDefault

2715

propname = "ApprovedByDefault"

2716

values_to_get = [False]

2717

2718

2719

class TestSetCheckerCmd(TestPropertySetterCmd):

2720

command = command.SetChecker

2721

propname = "Checker"

2722

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

2723

2724

2725

class TestSetHostCmd(TestPropertySetterCmd):

2726

command = command.SetHost

2727

propname = "Host"

2728

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

2729

2730

2731

class TestSetSecretCmd(TestPropertySetterCmd):

2732

command = command.SetSecret

2733

propname = "Secret"

2734

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

2735

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

2736

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

2737

2738

2739

class TestSetTimeoutCmd(TestPropertySetterCmd):

2740

command = command.SetTimeout

2741

propname = "Timeout"

2742

values_to_set = [datetime.timedelta(),

2743

datetime.timedelta(minutes=5),

2744

datetime.timedelta(seconds=1),

2745

datetime.timedelta(weeks=1),

2746

datetime.timedelta(weeks=52)]

2747

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

2748

2749

2750

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2751

command = command.SetExtendedTimeout

2752

propname = "ExtendedTimeout"

2753

values_to_set = [datetime.timedelta(),

2754

datetime.timedelta(minutes=5),

2755

datetime.timedelta(seconds=1),

2756

datetime.timedelta(weeks=1),

2757

datetime.timedelta(weeks=52)]

2758

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

2759

2760

2761

class TestSetIntervalCmd(TestPropertySetterCmd):

2762

command = command.SetInterval

2763

propname = "Interval"

2764

values_to_set = [datetime.timedelta(),

2765

datetime.timedelta(minutes=5),

2766

datetime.timedelta(seconds=1),

2767

datetime.timedelta(weeks=1),

2768

datetime.timedelta(weeks=52)]

2769

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

2770

2771

2772

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2773

command = command.SetApprovalDelay

2774

propname = "ApprovalDelay"

2775

values_to_set = [datetime.timedelta(),

2776

datetime.timedelta(minutes=5),

2777

datetime.timedelta(seconds=1),

2778

datetime.timedelta(weeks=1),

2779

datetime.timedelta(weeks=52)]

2780

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

2781

2782

2783

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2784

command = command.SetApprovalDuration

2785

propname = "ApprovalDuration"

2786

values_to_set = [datetime.timedelta(),

2787

datetime.timedelta(minutes=5),

2788

datetime.timedelta(seconds=1),

2789

datetime.timedelta(weeks=1),

2790

datetime.timedelta(weeks=52)]

2791

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

2792

2793

2794

2795

def should_only_run_tests():

2796

parser = argparse.ArgumentParser(add_help=False)

2797

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

2798

args, unknown_args = parser.parse_known_args()

2799

run_tests = args.check

2800

if run_tests:

2801

# Remove --check argument from sys.argv

2802

sys.argv[1:] = unknown_args

2803

return run_tests

2804

2805

# Add all tests from doctest strings

2806

def load_tests(loader, tests, none):

2807

import doctest

2808

tests.addTests(doctest.DocTestSuite())

2809

return tests

2810

2811

if __name__ == "__main__":

2812

try:

2813

if should_only_run_tests():

2814

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

2815

unittest.main()

2816

else:

2817

main()

2818

finally:

2819

logging.shutdown()