/mandos/release : revision 237.7.743

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

class MandosBus(SystemBus):

501

busname_domain = "se.recompile"

502

busname = busname_domain + ".Mandos"

503

server_path = "/"

504

server_interface = busname_domain + ".Mandos"

505

client_interface = busname_domain + ".Mandos.Client"

506

del busname_domain

507

508

def get_clients_and_properties(self):

509

managed_objects = self.get_managed_objects(

510

self.busname, self.server_path)

511

return {objpath: properties[self.client_interface]

512

for objpath, properties in managed_objects.items()

513

if self.client_interface in properties}

514

515

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

516

return self.set_property(self.busname, objectpath,

517

self.client_interface, key,

518

value)

519

520

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

521

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

522

self.client_interface, *args)

523

524

def call_server_method(self, method, *args):

525

return self.call_method(method, self.busname,

526

self.server_path,

527

self.server_interface, *args)

528

529

class Error(Exception):

530

pass

531

532

class ConnectFailed(Error):

533

pass

534

535

536

class dbus_python_adapter:

537

538

class SystemBus(dbus.MandosBus):

539

"""Use dbus-python"""

540

541

def __init__(self, module=dbus_python):

542

self.dbus_python = module

543

self.bus = self.dbus_python.SystemBus()

544

545

@contextlib.contextmanager

546

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

547

try:

548

yield

549

except self.dbus_python.exceptions.DBusException as e:

550

# This does what "raise from" would do

551

exc = exception_class(*e.args)

552

exc.__cause__ = e

553

raise exc

554

555

def call_method(self, methodname, busname, objectpath,

556

interface, *args):

557

proxy_object = self.get_object(busname, objectpath)

558

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

559

interface, methodname,

560

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

561

method = getattr(proxy_object, methodname)

562

with self.convert_exception():

563

with dbus_python_adapter.SilenceLogger(

564

"dbus.proxies"):

565

value = method(*args, dbus_interface=interface)

566

return self.type_filter(value)

567

568

def get_object(self, busname, objectpath):

569

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

570

busname, objectpath)

571

with self.convert_exception(dbus.ConnectFailed):

572

return self.bus.get_object(busname, objectpath)

573

574

def type_filter(self, value):

575

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

576

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

577

return bool(value)

578

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

579

return str(value)

580

# Also recurse into dictionaries

581

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

582

return {self.type_filter(key):

583

self.type_filter(subval)

584

for key, subval in value.items()}

585

return value

586

587

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

588

if key == "Secret":

589

if not isinstance(value, bytes):

590

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

591

value = self.dbus_python.ByteArray(value)

592

return self.set_property(self.busname, objectpath,

593

self.client_interface, key,

594

value)

595

596

class SilenceLogger:

597

"Simple context manager to silence a particular logger"

598

def __init__(self, loggername):

599

self.logger = logging.getLogger(loggername)

600

601

def __enter__(self):

602

self.logger.addFilter(self.nullfilter)

603

604

class NullFilter(logging.Filter):

605

def filter(self, record):

606

return False

607

608

nullfilter = NullFilter()

609

610

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

611

self.logger.removeFilter(self.nullfilter)

612

613

614

class CachingBus(SystemBus):

615

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

616

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

617

self.object_cache = {}

618

super(dbus_python_adapter.CachingBus,

619

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

620

def get_object(self, busname, objectpath):

621

try:

622

return self.object_cache[(busname, objectpath)]

623

except KeyError:

624

new_object = super(

625

dbus_python_adapter.CachingBus,

626

self).get_object(busname, objectpath)

627

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

628

return new_object

629

630

631

class pydbus_adapter:

632

class SystemBus(dbus.MandosBus):

633

def __init__(self, module=pydbus):

634

self.pydbus = module

635

self.bus = self.pydbus.SystemBus()

636

637

@contextlib.contextmanager

638

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

639

try:

640

yield

641

except gi.repository.GLib.Error as e:

642

# This does what "raise from" would do

643

exc = exception_class(*e.args)

644

exc.__cause__ = e

645

raise exc

646

647

def call_method(self, methodname, busname, objectpath,

648

interface, *args):

649

proxy_object = self.get(busname, objectpath)

650

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

651

interface, methodname,

652

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

653

method = getattr(proxy_object[interface], methodname)

654

with self.convert_exception():

655

return method(*args)

656

657

def get(self, busname, objectpath):

658

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

659

busname, objectpath)

660

with self.convert_exception(dbus.ConnectFailed):

661

if sys.version_info.major <= 2:

662

with warnings.catch_warnings():

663

warnings.filterwarnings(

664

"ignore", "", DeprecationWarning,

665

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

666

return self.bus.get(busname, objectpath)

667

else:

668

return self.bus.get(busname, objectpath)

669

670

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

671

value):

672

proxy_object = self.get(busname, objectpath)

673

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

674

objectpath, self.properties_iface, interface,

675

key, value)

676

setattr(proxy_object[interface], key, value)

677

678

class CachingBus(SystemBus):

679

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

680

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

681

self.object_cache = {}

682

super(pydbus_adapter.CachingBus,

683

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

684

def get(self, busname, objectpath):

685

try:

686

return self.object_cache[(busname, objectpath)]

687

except KeyError:

688

new_object = (super(pydbus_adapter.CachingBus, self)

689

.get(busname, objectpath))

690

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

691

return new_object

692

693

694

class dbussy_adapter:

695

class SystemBus(dbus.SystemBus):

696

"""Use DBussy"""

697

698

def __init__(self, dbussy, ravel):

699

self.dbussy = dbussy

700

self.ravel = ravel

701

self.bus = ravel.system_bus()

702

703

@contextlib.contextmanager

704

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

705

try:

706

yield

707

except self.dbussy.DBusError as e:

708

# This does what "raise from" would do

709

exc = exception_class(*e.args)

710

exc.__cause__ = e

711

raise exc

712

713

def call_method(self, methodname, busname, objectpath,

714

interface, *args):

715

proxy_object = self.get_object(busname, objectpath)

716

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

717

interface, methodname,

718

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

719

iface = proxy_object.get_interface(interface)

720

method = getattr(iface, methodname)

721

with self.convert_exception(dbus.Error):

722

value = method(*args)

723

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

724

# tuple: (signature, value)

725

if value:

726

return self.type_filter(value[0])

727

728

def get_object(self, busname, objectpath):

729

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

730

busname, objectpath)

731

with self.convert_exception(dbus.ConnectFailed):

732

return self.bus[busname][objectpath]

733

734

def type_filter(self, value):

735

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

736

if isinstance(value, tuple):

737

if (len(value) == 2

738

and isinstance(value[0],

739

self.dbussy.DBUS.Signature)):

740

return self.type_filter(value[1])

741

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

742

return str(value)

743

# Also recurse into dictionaries

744

elif isinstance(value, dict):

745

return {self.type_filter(key):

746

self.type_filter(subval)

747

for key, subval in value.items()}

748

return value

749

750

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

751

value):

752

proxy_object = self.get_object(busname, objectpath)

753

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

754

objectpath, self.properties_iface, interface,

755

key, value)

756

if key == "Secret":

757

# DBussy wants a Byte Array to be a sequence of

758

# values, not a byte string

759

value = tuple(value)

760

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

761

762

class MandosBus(SystemBus, dbus.MandosBus):

763

pass

764

765

class CachingBus(MandosBus):

766

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

767

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

768

self.object_cache = {}

769

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

770

**kwargs)

771

def get_object(self, busname, objectpath):

772

try:

773

return self.object_cache[(busname, objectpath)]

774

except KeyError:

775

new_object = super(

776

dbussy_adapter.CachingBus,

777

self).get_object(busname, objectpath)

778

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

779

return new_object

780

781

782

def commands_from_options(options):

783

784

commands = list(options.commands)

785

786

def find_cmd(cmd, commands):

787

i = 0

788

for i, c in enumerate(commands):

789

if isinstance(c, cmd):

790

return i

791

return i+1

792

793

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

794

# to occur ahead of command.Remove.

795

index_of_remove = find_cmd(command.Remove, commands)

796

before_remove = commands[:index_of_remove]

797

after_remove = commands[index_of_remove:]

798

cleaned_after = []

799

for cmd in after_remove:

800

if isinstance(cmd, command.Deny):

801

before_remove.append(cmd)

802

else:

803

cleaned_after.append(cmd)

804

if cleaned_after != after_remove:

805

commands = before_remove + cleaned_after

806

807

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

808

# optionally verbosely

809

if not commands:

810

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

811

812

return commands

813

814

815

class command:

816

"""A namespace for command classes"""

817

818

class Base:

819

"""Abstract base class for commands"""

820

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

821

"""Normal commands should implement run_on_one_client(),

822

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

823

override this run() method instead.

824

"""

825

self.bus = bus

826

for client, properties in clients.items():

827

self.run_on_one_client(client, properties)

828

829

830

class IsEnabled(Base):

831

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

832

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

833

if properties["Enabled"]:

834

sys.exit(0)

835

sys.exit(1)

836

837

838

class Approve(Base):

839

def run_on_one_client(self, client, properties):

840

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

841

842

843

class Deny(Base):

844

def run_on_one_client(self, client, properties):

845

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

846

847

848

class Remove(Base):

849

def run(self, clients, bus):

850

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

851

bus.call_server_method("RemoveClient", clientpath)

852

853

854

class Output(Base):

855

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

856

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

857

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

858

"Fingerprint", "CheckerRunning",

859

"LastEnabled", "ApprovalPending",

860

"ApprovedByDefault", "LastApprovalRequest",

861

"ApprovalDelay", "ApprovalDuration",

862

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

863

"LastCheckerStatus")

864

865

866

class DumpJSON(Output):

867

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

868

data = {properties["Name"]:

869

{key: properties[key]

870

for key in self.all_keywords}

871

for properties in clients.values()}

872

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

873

874

875

class PrintTable(Output):

876

def __init__(self, verbose=False):

877

self.verbose = verbose

878

879

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

880

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

881

"LastCheckedOK")

882

keywords = default_keywords

883

if self.verbose:

884

keywords = self.all_keywords

885

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

886

887

class TableOfClients:

888

tableheaders = {

889

"Name": "Name",

890

"Enabled": "Enabled",

891

"Timeout": "Timeout",

892

"LastCheckedOK": "Last Successful Check",

893

"LastApprovalRequest": "Last Approval Request",

894

"Created": "Created",

895

"Interval": "Interval",

896

"Host": "Host",

897

"Fingerprint": "Fingerprint",

898

"KeyID": "Key ID",

899

"CheckerRunning": "Check Is Running",

900

"LastEnabled": "Last Enabled",

901

"ApprovalPending": "Approval Is Pending",

902

"ApprovedByDefault": "Approved By Default",

903

"ApprovalDelay": "Approval Delay",

904

"ApprovalDuration": "Approval Duration",

905

"Checker": "Checker",

906

"ExtendedTimeout": "Extended Timeout",

907

"Expires": "Expires",

908

"LastCheckerStatus": "Last Checker Status",

909

}

910

911

def __init__(self, clients, keywords):

912

self.clients = clients

913

self.keywords = keywords

914

915

def __str__(self):

916

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

917

918

if sys.version_info.major == 2:

919

__unicode__ = __str__

920

def __str__(self):

921

return str(self).encode(

922

locale.getpreferredencoding())

923

924

def rows(self):

925

format_string = self.row_formatting_string()

926

rows = [self.header_line(format_string)]

927

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

928

for client in self.clients)

929

return rows

930

931

def row_formatting_string(self):

932

"Format string used to format table rows"

933

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

934

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

935

*(len(self.string_from_client(client,

936

key))

937

for client in self.clients)),

938

key=key)

939

for key in self.keywords)

940

941

def string_from_client(self, client, key):

942

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

943

944

@classmethod

945

def valuetostring(cls, value, keyword):

946

if isinstance(value, bool):

947

return "Yes" if value else "No"

948

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

949

"ApprovalDuration", "ExtendedTimeout"):

950

return cls.milliseconds_to_string(value)

951

return str(value)

952

953

def header_line(self, format_string):

954

return format_string.format(**self.tableheaders)

955

956

def client_line(self, client, format_string):

957

return format_string.format(

958

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

959

for key in self.keywords})

960

961

@staticmethod

962

def milliseconds_to_string(ms):

963

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

964

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

965

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

966

if td.days else "",

967

hours=td.seconds // 3600,

968

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

969

seconds=td.seconds % 60))

970

971

972

class PropertySetter(Base):

973

"Abstract class for Actions for setting one client property"

974

975

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

976

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

977

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

978

self.value_to_set)

979

980

@property

981

def propname(self):

982

raise NotImplementedError()

983

984

985

class Enable(PropertySetter):

986

propname = "Enabled"

987

value_to_set = True

988

989

990

class Disable(PropertySetter):

991

propname = "Enabled"

992

value_to_set = False

993

994

995

class BumpTimeout(PropertySetter):

996

propname = "LastCheckedOK"

997

value_to_set = ""

998

999

1000

class StartChecker(PropertySetter):

1001

propname = "CheckerRunning"

1002

value_to_set = True

1003

1004

1005

class StopChecker(PropertySetter):

1006

propname = "CheckerRunning"

1007

value_to_set = False

1008

1009

1010

class ApproveByDefault(PropertySetter):

1011

propname = "ApprovedByDefault"

1012

value_to_set = True

1013

1014

1015

class DenyByDefault(PropertySetter):

1016

propname = "ApprovedByDefault"

1017

value_to_set = False

1018

1019

1020

class PropertySetterValue(PropertySetter):

1021

"""Abstract class for PropertySetter recieving a value as

1022

constructor argument instead of a class attribute."""

1023

def __init__(self, value):

1024

self.value_to_set = value

1025

1026

@classmethod

1027

def argparse(cls, argtype):

1028

def cmdtype(arg):

1029

return cls(argtype(arg))

1030

return cmdtype

1031

1032

class SetChecker(PropertySetterValue):

1033

propname = "Checker"

1034

1035

1036

class SetHost(PropertySetterValue):

1037

propname = "Host"

1038

1039

1040

class SetSecret(PropertySetterValue):

1041

propname = "Secret"

1042

1043

@property

1044

def value_to_set(self):

1045

return self._vts

1046

1047

@value_to_set.setter

1048

def value_to_set(self, value):

1049

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

1050

self._vts = value.read()

1051

value.close()

1052

1053

1054

class PropertySetterValueMilliseconds(PropertySetterValue):

1055

"""Abstract class for PropertySetterValue taking a value

1056

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

1057

milliseconds."""

1058

1059

@property

1060

def value_to_set(self):

1061

return self._vts

1062

1063

@value_to_set.setter

1064

def value_to_set(self, value):

1065

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

1066

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

1067

1068

1069

class SetTimeout(PropertySetterValueMilliseconds):

1070

propname = "Timeout"

1071

1072

1073

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1074

propname = "ExtendedTimeout"

1075

1076

1077

class SetInterval(PropertySetterValueMilliseconds):

1078

propname = "Interval"

1079

1080

1081

class SetApprovalDelay(PropertySetterValueMilliseconds):

1082

propname = "ApprovalDelay"

1083

1084

1085

class SetApprovalDuration(PropertySetterValueMilliseconds):

1086

propname = "ApprovalDuration"

1087

1088

1089

1090

class TestCaseWithAssertLogs(unittest.TestCase):

1091

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

1092

1093

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

1094

@contextlib.contextmanager

1095

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

1096

capturing_handler = self.CapturingLevelHandler(level)

1097

old_level = logger.level

1098

old_propagate = logger.propagate

1099

logger.addHandler(capturing_handler)

1100

logger.setLevel(level)

1101

logger.propagate = False

1102

try:

1103

yield capturing_handler.watcher

1104

finally:

1105

logger.propagate = old_propagate

1106

logger.removeHandler(capturing_handler)

1107

logger.setLevel(old_level)

1108

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

1109

0)

1110

1111

class CapturingLevelHandler(logging.Handler):

1112

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

1113

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

1114

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

1115

def emit(self, record):

1116

self.watcher.records.append(record)

1117

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

1118

1119

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1120

("records",

1121

"output"))

1122

1123

1124

class Unique:

1125

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

1126

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

1127

1128

1129

class Test_string_to_delta(TestCaseWithAssertLogs):

1130

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

1131

# rfc3339_duration_to_delta() already has more comprehensive

1132

# tests, which are run by doctest.

1133

1134

def test_rfc3339_zero_seconds(self):

1135

self.assertEqual(datetime.timedelta(),

1136

string_to_delta("PT0S"))

1137

1138

def test_rfc3339_zero_days(self):

1139

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

1140

1141

def test_rfc3339_one_second(self):

1142

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

1143

string_to_delta("PT1S"))

1144

1145

def test_rfc3339_two_hours(self):

1146

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

1147

string_to_delta("PT2H"))

1148

1149

def test_falls_back_to_pre_1_6_1_with_warning(self):

1150

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

1151

value = string_to_delta("2h")

1152

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

1153

1154

1155

class Test_check_option_syntax(unittest.TestCase):

1156

def setUp(self):

1157

self.parser = argparse.ArgumentParser()

1158

add_command_line_options(self.parser)

1159

1160

def test_actions_requires_client_or_all(self):

1161

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

1162

args = self.actionargs(action, value)

1163

with self.assertParseError():

1164

self.parse_args(args)

1165

1166

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

1167

# check_option_syntax()

1168

actions = {

1169

"--enable": None,

1170

"--disable": None,

1171

"--bump-timeout": None,

1172

"--start-checker": None,

1173

"--stop-checker": None,

1174

"--is-enabled": None,

1175

"--remove": None,

1176

"--checker": "x",

1177

"--timeout": "PT0S",

1178

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

1179

"--interval": "PT0S",

1180

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

1181

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

1182

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

1183

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

1184

"--host": "hostname",

1185

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

1186

"--approve": None,

1187

"--deny": None,

1188

}

1189

1190

@staticmethod

1191

def actionargs(action, value, *args):

1192

if value is not None:

1193

return [action, value] + list(args)

1194

else:

1195

return [action] + list(args)

1196

1197

@contextlib.contextmanager

1198

def assertParseError(self):

1199

with self.assertRaises(SystemExit) as e:

1200

with self.redirect_stderr_to_devnull():

1201

yield

1202

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

1203

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

1204

# /argparse.html#exiting-methods

1205

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

1206

1207

def parse_args(self, args):

1208

options = self.parser.parse_args(args)

1209

check_option_syntax(self.parser, options)

1210

1211

@staticmethod

1212

@contextlib.contextmanager

1213

def redirect_stderr_to_devnull():

1214

old_stderr = sys.stderr

1215

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

1216

sys.stderr = null

1217

try:

1218

yield

1219

finally:

1220

sys.stderr = old_stderr

1221

1222

def check_option_syntax(self, options):

1223

check_option_syntax(self.parser, options)

1224

1225

def test_actions_all_conflicts_with_verbose(self):

1226

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

1227

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

1228

"--verbose")

1229

with self.assertParseError():

1230

self.parse_args(args)

1231

1232

def test_actions_with_client_conflicts_with_verbose(self):

1233

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

1234

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

1235

"client")

1236

with self.assertParseError():

1237

self.parse_args(args)

1238

1239

def test_dump_json_conflicts_with_verbose(self):

1240

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

1241

with self.assertParseError():

1242

self.parse_args(args)

1243

1244

def test_dump_json_conflicts_with_action(self):

1245

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

1246

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

1247

with self.assertParseError():

1248

self.parse_args(args)

1249

1250

def test_all_can_not_be_alone(self):

1251

args = ["--all"]

1252

with self.assertParseError():

1253

self.parse_args(args)

1254

1255

def test_all_is_ok_with_any_action(self):

1256

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

1257

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

1258

self.parse_args(args)

1259

1260

def test_any_action_is_ok_with_one_client(self):

1261

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

1262

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

1263

self.parse_args(args)

1264

1265

def test_one_client_with_all_actions_except_is_enabled(self):

1266

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

1267

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

1268

continue

1269

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

1270

self.parse_args(args)

1271

1272

def test_two_clients_with_all_actions_except_is_enabled(self):

1273

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

1274

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

1275

continue

1276

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

1277

"client2")

1278

self.parse_args(args)

1279

1280

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1281

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

1282

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

1283

continue

1284

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

1285

"client2")

1286

self.parse_args(args)

1287

1288

def test_is_enabled_fails_without_client(self):

1289

args = ["--is-enabled"]

1290

with self.assertParseError():

1291

self.parse_args(args)

1292

1293

def test_is_enabled_fails_with_two_clients(self):

1294

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

1295

with self.assertParseError():

1296

self.parse_args(args)

1297

1298

def test_remove_can_only_be_combined_with_action_deny(self):

1299

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

1300

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

1301

continue

1302

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

1303

"--remove")

1304

with self.assertParseError():

1305

self.parse_args(args)

1306

1307

1308

class Test_dbus_exceptions(unittest.TestCase):

1309

1310

def test_dbus_ConnectFailed_is_Error(self):

1311

with self.assertRaises(dbus.Error):

1312

raise dbus.ConnectFailed()

1313

1314

1315

class Test_dbus_MandosBus(unittest.TestCase):

1316

1317

class MockMandosBus(dbus.MandosBus):

1318

def __init__(self):

1319

self._name = "se.recompile.Mandos"

1320

self._server_path = "/"

1321

self._server_interface = "se.recompile.Mandos"

1322

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

1323

self.calls = []

1324

self.call_method_return = Unique()

1325

1326

def call_method(self, methodname, busname, objectpath,

1327

interface, *args):

1328

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

1329

interface, args))

1330

return self.call_method_return

1331

1332

def setUp(self):

1333

self.bus = self.MockMandosBus()

1334

1335

def test_set_client_property(self):

1336

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

1337

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

1338

"org.freedesktop.DBus.Properties",

1339

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

1340

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

1341

1342

def test_call_client_method(self):

1343

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

1344

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

1345

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

1346

self.bus._client_interface, ())

1347

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

1348

1349

def test_call_client_method_with_args(self):

1350

args = (Unique(), Unique())

1351

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

1352

*args)

1353

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

1354

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

1355

self.bus._client_interface,

1356

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

1357

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

1358

1359

def test_get_clients_and_properties(self):

1360

managed_objects = {

1361

"objectpath": {

1362

self.bus._client_interface: {

1363

"key": "value",

1364

"bool": True,

1365

},

1366

"irrelevant_interface": {

1367

"key": "othervalue",

1368

"bool": False,

1369

},

1370

},

1371

"other_objectpath": {

1372

"other_irrelevant_interface": {

1373

"key": "value 3",

1374

"bool": None,

1375

},

1376

},

1377

}

1378

expected_clients_and_properties = {

1379

"objectpath": {

1380

"key": "value",

1381

"bool": True,

1382

}

1383

}

1384

self.bus.call_method_return = managed_objects

1385

ret = self.bus.get_clients_and_properties()

1386

self.assertDictEqual(expected_clients_and_properties, ret)

1387

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

1388

self.bus._server_path,

1389

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

1390

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

1391

1392

def test_call_server_method(self):

1393

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

1394

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

1395

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

1396

self.bus._server_path,

1397

self.bus._server_interface, ())

1398

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

1399

1400

def test_call_server_method_with_args(self):

1401

args = (Unique(), Unique())

1402

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

1403

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

1404

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

1405

self.bus._server_path,

1406

self.bus._server_interface,

1407

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

1408

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

1409

1410

1411

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1412

1413

def MockDBusPython_func(self, func):

1414

class mock_dbus_python:

1415

"""mock dbus-python module"""

1416

class exceptions:

1417

"""Pseudo-namespace"""

1418

class DBusException(Exception):

1419

pass

1420

class SystemBus:

1421

@staticmethod

1422

def get_object(busname, objectpath):

1423

DBusObject = collections.namedtuple(

1424

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

1425

def method(*args, **kwargs):

1426

self.assertEqual({"dbus_interface":

1427

"interface"},

1428

kwargs)

1429

return func(*args)

1430

def set_property(interface, key, value,

1431

dbus_interface=None):

1432

self.assertEqual(

1433

"org.freedesktop.DBus.Properties",

1434

dbus_interface)

1435

self.assertEqual("Secret", key)

1436

return func(interface, key, value,

1437

dbus_interface=dbus_interface)

1438

return DBusObject(methodname=method,

1439

Set=set_property)

1440

class Boolean:

1441

def __init__(self, value):

1442

self.value = bool(value)

1443

def __bool__(self):

1444

return self.value

1445

if sys.version_info.major == 2:

1446

__nonzero__ = __bool__

1447

class ObjectPath(str):

1448

pass

1449

class Dictionary(dict):

1450

pass

1451

class ByteArray(bytes):

1452

pass

1453

return mock_dbus_python

1454

1455

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

1456

interface, *args):

1457

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

1458

return bus.call_method(methodname, busname, objectpath,

1459

interface, *args)

1460

1461

def test_call_method_returns(self):

1462

expected_method_return = Unique()

1463

method_args = (Unique(), Unique())

1464

def func(*args):

1465

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

1466

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

1467

self.assertIs(marg, arg)

1468

return expected_method_return

1469

mock_dbus_python = self.MockDBusPython_func(func)

1470

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1471

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

1472

"objectpath", "interface",

1473

*method_args)

1474

self.assertIs(ret, expected_method_return)

1475

1476

def test_call_method_filters_bool_true(self):

1477

def func():

1478

return method_return

1479

mock_dbus_python = self.MockDBusPython_func(func)

1480

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1481

method_return = mock_dbus_python.Boolean(True)

1482

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

1483

"objectpath", "interface")

1484

self.assertTrue(ret)

1485

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1486

1487

def test_call_method_filters_bool_false(self):

1488

def func():

1489

return method_return

1490

mock_dbus_python = self.MockDBusPython_func(func)

1491

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1492

method_return = mock_dbus_python.Boolean(False)

1493

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

1494

"objectpath", "interface")

1495

self.assertFalse(ret)

1496

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1497

1498

def test_call_method_filters_objectpath(self):

1499

def func():

1500

return method_return

1501

mock_dbus_python = self.MockDBusPython_func(func)

1502

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1503

method_return = mock_dbus_python.ObjectPath("objectpath")

1504

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

1505

"objectpath", "interface")

1506

self.assertEqual("objectpath", ret)

1507

self.assertIsNot("objectpath", ret)

1508

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1509

1510

def test_call_method_filters_booleans_in_dict(self):

1511

def func():

1512

return method_return

1513

mock_dbus_python = self.MockDBusPython_func(func)

1514

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1515

method_return = mock_dbus_python.Dictionary(

1516

{mock_dbus_python.Boolean(True):

1517

mock_dbus_python.Boolean(False),

1518

mock_dbus_python.Boolean(False):

1519

mock_dbus_python.Boolean(True)})

1520

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

1521

"objectpath", "interface")

1522

expected_method_return = {True: False,

1523

False: True}

1524

self.assertEqual(expected_method_return, ret)

1525

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1526

1527

def test_call_method_filters_objectpaths_in_dict(self):

1528

def func():

1529

return method_return

1530

mock_dbus_python = self.MockDBusPython_func(func)

1531

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1532

method_return = mock_dbus_python.Dictionary(

1533

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1534

mock_dbus_python.ObjectPath("objectpath_value_1"),

1535

mock_dbus_python.ObjectPath("objectpath_key_2"):

1536

mock_dbus_python.ObjectPath("objectpath_value_2")})

1537

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

1538

"objectpath", "interface")

1539

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

1540

for key, value in

1541

method_return.items()}

1542

self.assertEqual(expected_method_return, ret)

1543

self.assertIsInstance(ret, dict)

1544

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1545

1546

def test_call_method_filters_dict_in_dict(self):

1547

def func():

1548

return method_return

1549

mock_dbus_python = self.MockDBusPython_func(func)

1550

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1551

method_return = mock_dbus_python.Dictionary(

1552

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

1553

"key12": "value12"}),

1554

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

1555

"key22": "value22"})})

1556

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

1557

"objectpath", "interface")

1558

expected_method_return = {

1559

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

1560

"key12": "value12"},

1561

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

1562

"key22": "value22"},

1563

}

1564

self.assertEqual(expected_method_return, ret)

1565

self.assertIsInstance(ret, dict)

1566

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1567

for key, value in ret.items():

1568

self.assertIsInstance(value, dict)

1569

self.assertEqual(expected_method_return[key], value)

1570

self.assertNotIsInstance(value,

1571

mock_dbus_python.Dictionary)

1572

1573

def test_call_method_filters_dict_three_deep(self):

1574

def func():

1575

return method_return

1576

mock_dbus_python = self.MockDBusPython_func(func)

1577

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1578

method_return = mock_dbus_python.Dictionary(

1579

{"key1":

1580

mock_dbus_python.Dictionary(

1581

{"key2":

1582

mock_dbus_python.Dictionary(

1583

{"key3":

1584

mock_dbus_python.Boolean(True),

1585

}),

1586

}),

1587

})

1588

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

1589

"objectpath", "interface")

1590

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

1591

self.assertEqual(expected_method_return, ret)

1592

self.assertIsInstance(ret, dict)

1593

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1594

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

1595

self.assertNotIsInstance(ret["key1"],

1596

mock_dbus_python.Dictionary)

1597

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

1598

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

1599

mock_dbus_python.Dictionary)

1600

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

1601

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

1602

mock_dbus_python.Boolean)

1603

1604

def test_call_method_handles_exception(self):

1605

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

1606

1607

def func():

1608

dbus_logger.error("Test")

1609

raise mock_dbus_python.exceptions.DBusException()

1610

1611

mock_dbus_python = self.MockDBusPython_func(func)

1612

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1613

1614

class CountingHandler(logging.Handler):

1615

count = 0

1616

def emit(self, record):

1617

self.count += 1

1618

1619

counting_handler = CountingHandler()

1620

1621

dbus_logger.addHandler(counting_handler)

1622

1623

try:

1624

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

1625

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

1626

"objectpath", "interface")

1627

finally:

1628

dbus_logger.removeFilter(counting_handler)

1629

1630

self.assertNotIsInstance(e, dbus.ConnectFailed)

1631

1632

# Make sure the dbus logger was suppressed

1633

self.assertEqual(0, counting_handler.count)

1634

1635

def test_Set_Secret_sends_bytearray(self):

1636

ret = [None]

1637

def func(*args, **kwargs):

1638

ret[0] = (args, kwargs)

1639

mock_dbus_python = self.MockDBusPython_func(func)

1640

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1641

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

1642

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

1643

mock_dbus_python.ByteArray(b"value")),

1644

{"dbus_interface":

1645

"org.freedesktop.DBus.Properties"})

1646

self.assertEqual(expected_call, ret[0])

1647

if sys.version_info.major == 2:

1648

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

1649

mock_dbus_python.ByteArray)

1650

1651

def test_get_object_converts_to_correct_exception(self):

1652

bus = dbus_python_adapter.SystemBus(

1653

self.fake_dbus_python_raises_exception_on_connect)

1654

with self.assertRaises(dbus.ConnectFailed):

1655

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

1656

"objectpath", "interface")

1657

1658

class fake_dbus_python_raises_exception_on_connect:

1659

"""fake dbus-python module"""

1660

class exceptions:

1661

"""Pseudo-namespace"""

1662

class DBusException(Exception):

1663

pass

1664

1665

@classmethod

1666

def SystemBus(cls):

1667

def get_object(busname, objectpath):

1668

raise cls.exceptions.DBusException()

1669

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

1670

return Bus(get_object=get_object)

1671

1672

1673

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1674

class mock_dbus_python:

1675

"""mock dbus-python modules"""

1676

class SystemBus:

1677

@staticmethod

1678

def get_object(busname, objectpath):

1679

return Unique()

1680

1681

def setUp(self):

1682

self.bus = dbus_python_adapter.CachingBus(

1683

self.mock_dbus_python)

1684

1685

def test_returns_distinct_objectpaths(self):

1686

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

1687

self.assertIsInstance(obj1, Unique)

1688

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

1689

self.assertIsInstance(obj2, Unique)

1690

self.assertIsNot(obj1, obj2)

1691

1692

def test_returns_distinct_busnames(self):

1693

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

1694

self.assertIsInstance(obj1, Unique)

1695

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

1696

self.assertIsInstance(obj2, Unique)

1697

self.assertIsNot(obj1, obj2)

1698

1699

def test_returns_distinct_both(self):

1700

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

1701

self.assertIsInstance(obj1, Unique)

1702

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

1703

self.assertIsInstance(obj2, Unique)

1704

self.assertIsNot(obj1, obj2)

1705

1706

def test_returns_same(self):

1707

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

1708

self.assertIsInstance(obj1, Unique)

1709

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

1710

self.assertIsInstance(obj2, Unique)

1711

self.assertIs(obj1, obj2)

1712

1713

def test_returns_same_old(self):

1714

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

1715

self.assertIsInstance(obj1, Unique)

1716

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

1717

self.assertIsInstance(obj2, Unique)

1718

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

1719

self.assertIsInstance(obj1b, Unique)

1720

self.assertIsNot(obj1, obj2)

1721

self.assertIsNot(obj2, obj1b)

1722

self.assertIs(obj1, obj1b)

1723

1724

1725

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1726

1727

def Stub_pydbus_func(self, func):

1728

class stub_pydbus:

1729

"""stub pydbus module"""

1730

class SystemBus:

1731

@staticmethod

1732

def get(busname, objectpath):

1733

DBusObject = collections.namedtuple(

1734

"DBusObject", ("methodname",))

1735

return {"interface":

1736

DBusObject(methodname=func)}

1737

return stub_pydbus

1738

1739

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

1740

interface, *args):

1741

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

1742

return bus.call_method(methodname, busname, objectpath,

1743

interface, *args)

1744

1745

def test_call_method_returns(self):

1746

expected_method_return = Unique()

1747

method_args = (Unique(), Unique())

1748

def func(*args):

1749

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

1750

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

1751

self.assertIs(marg, arg)

1752

return expected_method_return

1753

stub_pydbus = self.Stub_pydbus_func(func)

1754

bus = pydbus_adapter.SystemBus(stub_pydbus)

1755

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

1756

"objectpath", "interface",

1757

*method_args)

1758

self.assertIs(ret, expected_method_return)

1759

1760

def test_call_method_handles_exception(self):

1761

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

1762

1763

def func():

1764

raise gi.repository.GLib.Error()

1765

1766

stub_pydbus = self.Stub_pydbus_func(func)

1767

bus = pydbus_adapter.SystemBus(stub_pydbus)

1768

1769

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

1770

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

1771

"objectpath", "interface")

1772

1773

self.assertNotIsInstance(e, dbus.ConnectFailed)

1774

1775

def test_get_converts_to_correct_exception(self):

1776

bus = pydbus_adapter.SystemBus(

1777

self.fake_pydbus_raises_exception_on_connect)

1778

with self.assertRaises(dbus.ConnectFailed):

1779

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

1780

"objectpath", "interface")

1781

1782

class fake_pydbus_raises_exception_on_connect:

1783

"""fake dbus-python module"""

1784

@classmethod

1785

def SystemBus(cls):

1786

def get(busname, objectpath):

1787

raise gi.repository.GLib.Error()

1788

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

1789

return Bus(get=get)

1790

1791

def test_set_property_uses_setattr(self):

1792

class Object:

1793

pass

1794

obj = Object()

1795

class pydbus_spy:

1796

class SystemBus:

1797

@staticmethod

1798

def get(busname, objectpath):

1799

return {"interface": obj}

1800

bus = pydbus_adapter.SystemBus(pydbus_spy)

1801

value = Unique()

1802

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

1803

value)

1804

self.assertIs(value, obj.key)

1805

1806

def test_get_suppresses_xml_deprecation_warning(self):

1807

if sys.version_info.major >= 3:

1808

return

1809

class stub_pydbus_get:

1810

class SystemBus:

1811

@staticmethod

1812

def get(busname, objectpath):

1813

warnings.warn_explicit(

1814

"deprecated", DeprecationWarning,

1815

"xml.etree.ElementTree", 0)

1816

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1817

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

1818

warnings.simplefilter("always")

1819

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

1820

self.assertEqual(0, len(w))

1821

1822

1823

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1824

class stub_pydbus:

1825

"""stub pydbus module"""

1826

class SystemBus:

1827

@staticmethod

1828

def get(busname, objectpath):

1829

return Unique()

1830

1831

def setUp(self):

1832

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

1833

1834

def test_returns_distinct_objectpaths(self):

1835

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

1836

self.assertIsInstance(obj1, Unique)

1837

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

1838

self.assertIsInstance(obj2, Unique)

1839

self.assertIsNot(obj1, obj2)

1840

1841

def test_returns_distinct_busnames(self):

1842

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

1843

self.assertIsInstance(obj1, Unique)

1844

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

1845

self.assertIsInstance(obj2, Unique)

1846

self.assertIsNot(obj1, obj2)

1847

1848

def test_returns_distinct_both(self):

1849

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

1850

self.assertIsInstance(obj1, Unique)

1851

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

1852

self.assertIsInstance(obj2, Unique)

1853

self.assertIsNot(obj1, obj2)

1854

1855

def test_returns_same(self):

1856

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

1857

self.assertIsInstance(obj1, Unique)

1858

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

1859

self.assertIsInstance(obj2, Unique)

1860

self.assertIs(obj1, obj2)

1861

1862

def test_returns_same_old(self):

1863

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

1864

self.assertIsInstance(obj1, Unique)

1865

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

1866

self.assertIsInstance(obj2, Unique)

1867

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

1868

self.assertIsInstance(obj1b, Unique)

1869

self.assertIsNot(obj1, obj2)

1870

self.assertIsNot(obj2, obj1b)

1871

self.assertIs(obj1, obj1b)

1872

1873

1874

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1875

1876

class dummy_dbussy:

1877

class DBUS:

1878

class ObjectPath(str):

1879

pass

1880

class DBusError(Exception):

1881

pass

1882

1883

def fake_ravel_func(self, func):

1884

class fake_ravel:

1885

@staticmethod

1886

def system_bus():

1887

class DBusInterfaceProxy:

1888

@staticmethod

1889

def methodname(*args):

1890

return [func(*args)]

1891

class DBusObject:

1892

@staticmethod

1893

def get_interface(interface):

1894

if interface == "interface":

1895

return DBusInterfaceProxy()

1896

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

1897

return fake_ravel

1898

1899

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

1900

interface, *args):

1901

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

1902

return bus.call_method(methodname, busname, objectpath,

1903

interface, *args)

1904

1905

def test_call_method_returns(self):

1906

expected_method_return = Unique()

1907

method_args = (Unique(), Unique())

1908

def func(*args):

1909

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

1910

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

1911

self.assertIs(marg, arg)

1912

return expected_method_return

1913

fake_ravel = self.fake_ravel_func(func)

1914

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

1915

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

1916

"objectpath", "interface",

1917

*method_args)

1918

self.assertIs(ret, expected_method_return)

1919

1920

def test_call_method_filters_objectpath(self):

1921

def func():

1922

return method_return

1923

fake_ravel = self.fake_ravel_func(func)

1924

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

1925

method_return = (self.dummy_dbussy.DBUS

1926

.ObjectPath("objectpath"))

1927

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

1928

"objectpath", "interface")

1929

self.assertEqual("objectpath", ret)

1930

self.assertNotIsInstance(ret,

1931

self.dummy_dbussy.DBUS.ObjectPath)

1932

1933

def test_call_method_filters_objectpaths_in_dict(self):

1934

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1935

def func():

1936

return method_return

1937

fake_ravel = self.fake_ravel_func(func)

1938

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

1939

method_return = {

1940

ObjectPath("objectpath_key_1"):

1941

ObjectPath("objectpath_value_1"),

1942

ObjectPath("objectpath_key_2"):

1943

ObjectPath("objectpath_value_2"),

1944

}

1945

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

1946

"objectpath", "interface")

1947

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

1948

for key, value in

1949

method_return.items()}

1950

for key, value in ret.items():

1951

self.assertNotIsInstance(key, ObjectPath)

1952

self.assertNotIsInstance(value, ObjectPath)

1953

self.assertEqual(expected_method_return, ret)

1954

self.assertIsInstance(ret, dict)

1955

1956

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1957

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1958

def func():

1959

return method_return

1960

fake_ravel = self.fake_ravel_func(func)

1961

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

1962

method_return = {

1963

ObjectPath("key1"): {

1964

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

1965

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

1966

},

1967

ObjectPath("key2"): {

1968

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

1969

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

1970

},

1971

}

1972

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

1973

"objectpath", "interface")

1974

expected_method_return = {

1975

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

1976

"key12": "value12"},

1977

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

1978

"key22": "value22"},

1979

}

1980

self.assertEqual(expected_method_return, ret)

1981

for key, value in ret.items():

1982

self.assertIsInstance(value, dict)

1983

self.assertEqual(expected_method_return[key], value)

1984

self.assertNotIsInstance(key, ObjectPath)

1985

for inner_key, inner_value in value.items():

1986

self.assertIsInstance(value, dict)

1987

self.assertEqual(

1988

expected_method_return[key][inner_key],

1989

inner_value)

1990

self.assertNotIsInstance(key, ObjectPath)

1991

1992

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

1993

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1994

def func():

1995

return method_return

1996

fake_ravel = self.fake_ravel_func(func)

1997

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

1998

method_return = {

1999

ObjectPath("key1"): {

2000

ObjectPath("key2"): {

2001

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

2002

},

2003

},

2004

}

2005

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

2006

"objectpath", "interface")

2007

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

2008

self.assertEqual(expected_method_return, ret)

2009

self.assertIsInstance(ret, dict)

2010

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

2011

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

2012

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

2013

ObjectPath)

2014

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

2015

self.assertNotIsInstance(

2016

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

2017

ObjectPath)

2018

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

2019

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

2020

self.dummy_dbussy.DBUS.ObjectPath)

2021

2022

def test_call_method_handles_exception(self):

2023

def func():

2024

raise self.dummy_dbussy.DBusError()

2025

2026

fake_ravel = self.fake_ravel_func(func)

2027

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

2028

2029

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

2030

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

2031

"objectpath", "interface")

2032

2033

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

2034

2035

def test_get_object_converts_to_correct_exception(self):

2036

class fake_ravel_raises_exception_on_connect:

2037

@staticmethod

2038

def system_bus():

2039

class Bus:

2040

@staticmethod

2041

def __getitem__(key):

2042

if key == "objectpath":

2043

raise self.dummy_dbussy.DBusError()

2044

raise Exception(key)

2045

return {"busname": Bus()}

2046

def func():

2047

raise self.dummy_dbussy.DBusError()

2048

bus = dbussy_adapter.SystemBus(

2049

self.dummy_dbussy,

2050

fake_ravel_raises_exception_on_connect)

2051

with self.assertRaises(dbus.ConnectFailed):

2052

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

2053

"objectpath", "interface")

2054

2055

2056

class Test_commands_from_options(unittest.TestCase):

2057

2058

def setUp(self):

2059

self.parser = argparse.ArgumentParser()

2060

add_command_line_options(self.parser)

2061

2062

def test_is_enabled(self):

2063

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

2064

command.IsEnabled)

2065

2066

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

2067

clients=None, **cmd_attrs):

2068

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

2069

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

2070

options = self.parser.parse_args(args)

2071

check_option_syntax(self.parser, options)

2072

commands = commands_from_options(options)

2073

self.assertEqual(length, len(commands))

2074

for command in commands:

2075

if isinstance(command, command_cls):

2076

break

2077

else:

2078

self.assertIsInstance(command, command_cls)

2079

if clients is not None:

2080

self.assertEqual(clients, options.client)

2081

for key, value in cmd_attrs.items():

2082

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

2083

2084

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

2085

for cmd in commands:

2086

self.assert_command_from_args(args, cmd,

2087

length=len(commands),

2088

clients=clients)

2089

2090

def test_is_enabled_short(self):

2091

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

2092

command.IsEnabled)

2093

2094

def test_approve(self):

2095

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

2096

command.Approve)

2097

2098

def test_approve_short(self):

2099

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

2100

command.Approve)

2101

2102

def test_deny(self):

2103

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

2104

command.Deny)

2105

2106

def test_deny_short(self):

2107

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

2108

2109

def test_remove(self):

2110

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

2111

command.Remove)

2112

2113

def test_deny_before_remove(self):

2114

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

2115

"client"])

2116

check_option_syntax(self.parser, options)

2117

commands = commands_from_options(options)

2118

self.assertEqual(2, len(commands))

2119

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

2120

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

2121

2122

def test_deny_before_remove_reversed(self):

2123

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

2124

"--all"])

2125

check_option_syntax(self.parser, options)

2126

commands = commands_from_options(options)

2127

self.assertEqual(2, len(commands))

2128

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

2129

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

2130

2131

def test_remove_short(self):

2132

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

2133

command.Remove)

2134

2135

def test_dump_json(self):

2136

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

2137

command.DumpJSON)

2138

2139

def test_enable(self):

2140

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

2141

command.Enable)

2142

2143

def test_enable_short(self):

2144

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

2145

command.Enable)

2146

2147

def test_disable(self):

2148

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

2149

command.Disable)

2150

2151

def test_disable_short(self):

2152

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

2153

command.Disable)

2154

2155

def test_bump_timeout(self):

2156

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

2157

command.BumpTimeout)

2158

2159

def test_bump_timeout_short(self):

2160

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

2161

command.BumpTimeout)

2162

2163

def test_start_checker(self):

2164

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

2165

command.StartChecker)

2166

2167

def test_stop_checker(self):

2168

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

2169

command.StopChecker)

2170

2171

def test_approve_by_default(self):

2172

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

2173

"client"],

2174

command.ApproveByDefault)

2175

2176

def test_deny_by_default(self):

2177

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

2178

command.DenyByDefault)

2179

2180

def test_checker(self):

2181

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

2182

command.SetChecker,

2183

value_to_set=":")

2184

2185

def test_checker_empty(self):

2186

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

2187

command.SetChecker,

2188

value_to_set="")

2189

2190

def test_checker_short(self):

2191

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

2192

command.SetChecker,

2193

value_to_set=":")

2194

2195

def test_host(self):

2196

self.assert_command_from_args(

2197

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

2198

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

2199

2200

def test_host_short(self):

2201

self.assert_command_from_args(

2202

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

2203

value_to_set="client.example.org")

2204

2205

def test_secret_devnull(self):

2206

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

2207

"client"], command.SetSecret,

2208

value_to_set=b"")

2209

2210

def test_secret_tempfile(self):

2211

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

2212

value = b"secret\0xyzzy\nbar"

2213

f.write(value)

2214

f.seek(0)

2215

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

2216

"client"],

2217

command.SetSecret,

2218

value_to_set=value)

2219

2220

def test_secret_devnull_short(self):

2221

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

2222

"client"], command.SetSecret,

2223

value_to_set=b"")

2224

2225

def test_secret_tempfile_short(self):

2226

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

2227

value = b"secret\0xyzzy\nbar"

2228

f.write(value)

2229

f.seek(0)

2230

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

2231

command.SetSecret,

2232

value_to_set=value)

2233

2234

def test_timeout(self):

2235

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

2236

command.SetTimeout,

2237

value_to_set=300000)

2238

2239

def test_timeout_short(self):

2240

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

2241

command.SetTimeout,

2242

value_to_set=300000)

2243

2244

def test_extended_timeout(self):

2245

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

2246

"client"],

2247

command.SetExtendedTimeout,

2248

value_to_set=900000)

2249

2250

def test_interval(self):

2251

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

2252

"client"], command.SetInterval,

2253

value_to_set=120000)

2254

2255

def test_interval_short(self):

2256

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

2257

command.SetInterval,

2258

value_to_set=120000)

2259

2260

def test_approval_delay(self):

2261

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

2262

"client"],

2263

command.SetApprovalDelay,

2264

value_to_set=30000)

2265

2266

def test_approval_duration(self):

2267

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

2268

"client"],

2269

command.SetApprovalDuration,

2270

value_to_set=1000)

2271

2272

def test_print_table(self):

2273

self.assert_command_from_args([], command.PrintTable,

2274

verbose=False)

2275

2276

def test_print_table_verbose(self):

2277

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

2278

command.PrintTable,

2279

verbose=True)

2280

2281

def test_print_table_verbose_short(self):

2282

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

2283

verbose=True)

2284

2285

2286

def test_manual_page_example_1(self):

2287

self.assert_command_from_args("",

2288

command.PrintTable,

2289

clients=[],

2290

verbose=False)

2291

2292

def test_manual_page_example_2(self):

2293

self.assert_command_from_args(

2294

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

2295

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

2296

"foo2.example.org"],

2297

verbose=True)

2298

2299

def test_manual_page_example_3(self):

2300

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

2301

command.Enable,

2302

clients=[])

2303

2304

def test_manual_page_example_4(self):

2305

self.assert_commands_from_args(

2306

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

2307

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

2308

[command.SetTimeout, command.SetInterval],

2309

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

2310

2311

def test_manual_page_example_5(self):

2312

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

2313

command.Approve,

2314

clients=[])

2315

2316

2317

class TestCommand(unittest.TestCase):

2318

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

2319

2320

class FakeMandosBus(dbus.MandosBus):

2321

def __init__(self, testcase):

2322

self.client_properties = {

2323

"Name": "foo",

2324

"KeyID": ("92ed150794387c03ce684574b1139a65"

2325

"94a34f895daaaf09fd8ea90a27cddb12"),

2326

"Secret": b"secret",

2327

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

2328

"Enabled": True,

2329

"Timeout": 300000,

2330

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

2331

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

2332

"Interval": 120000,

2333

"Fingerprint": ("778827225BA7DE539C5A"

2334

"7CFA59CFF7CDBD9A5920"),

2335

"CheckerRunning": False,

2336

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

2337

"ApprovalPending": False,

2338

"ApprovedByDefault": True,

2339

"LastApprovalRequest": "",

2340

"ApprovalDelay": 0,

2341

"ApprovalDuration": 1000,

2342

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

2343

"ExtendedTimeout": 900000,

2344

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

2345

"LastCheckerStatus": 0,

2346

}

2347

self.other_client_properties = {

2348

"Name": "barbar",

2349

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2350

"6ab612cff5ad227247e46c2b020f441c"),

2351

"Secret": b"secretbar",

2352

"Host": "192.0.2.3",

2353

"Enabled": True,

2354

"Timeout": 300000,

2355

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

2356

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

2357

"Interval": 120000,

2358

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2359

"F547B3A107558FCA3A27"),

2360

"CheckerRunning": True,

2361

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

2362

"ApprovalPending": False,

2363

"ApprovedByDefault": False,

2364

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

2365

"ApprovalDelay": 30000,

2366

"ApprovalDuration": 93785000,

2367

"Checker": ":",

2368

"ExtendedTimeout": 900000,

2369

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

2370

"LastCheckerStatus": -2,

2371

}

2372

self.clients = collections.OrderedDict(

2373

[

2374

("client_objectpath", self.client_properties),

2375

("other_client_objectpath",

2376

self.other_client_properties),

2377

])

2378

self.one_client = {"client_objectpath":

2379

self.client_properties}

2380

self.testcase = testcase

2381

self.calls = []

2382

2383

def call_method(self, methodname, busname, objectpath,

2384

interface, *args):

2385

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

2386

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

2387

interface, args))

2388

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

2389

if methodname == "Set":

2390

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

2391

interface, key, value = args

2392

self.testcase.assertEqual(

2393

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

2394

self.clients[objectpath][key] = value

2395

return

2396

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

2397

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

2398

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

2399

clientpath = args[0]

2400

del self.clients[clientpath]

2401

return

2402

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

2403

if methodname == "Approve":

2404

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

2405

return

2406

raise ValueError()

2407

2408

def setUp(self):

2409

self.bus = self.FakeMandosBus(self)

2410

2411

2412

class TestBaseCommands(TestCommand):

2413

2414

def test_IsEnabled_exits_successfully(self):

2415

with self.assertRaises(SystemExit) as e:

2416

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

2417

if e.exception.code is not None:

2418

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

2419

else:

2420

self.assertIsNone(e.exception.code)

2421

2422

def test_IsEnabled_exits_with_failure(self):

2423

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

2424

with self.assertRaises(SystemExit) as e:

2425

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

2426

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

2427

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

2428

else:

2429

self.assertIsNotNone(e.exception.code)

2430

2431

def test_Approve(self):

2432

busname = "se.recompile.Mandos"

2433

client_interface = "se.recompile.Mandos.Client"

2434

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

2435

for clientpath in self.bus.clients:

2436

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

2437

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

2438

2439

def test_Deny(self):

2440

busname = "se.recompile.Mandos"

2441

client_interface = "se.recompile.Mandos.Client"

2442

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

2443

for clientpath in self.bus.clients:

2444

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

2445

client_interface, (False,)),

2446

self.bus.calls)

2447

2448

def test_Remove(self):

2449

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

2450

for clientpath in self.bus.clients:

2451

self.assertIn(("RemoveClient", dbus_busname,

2452

dbus_server_path, dbus_server_interface,

2453

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

2454

2455

expected_json = {

2456

"foo": {

2457

"Name": "foo",

2458

"KeyID": ("92ed150794387c03ce684574b1139a65"

2459

"94a34f895daaaf09fd8ea90a27cddb12"),

2460

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

2461

"Enabled": True,

2462

"Timeout": 300000,

2463

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

2464

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

2465

"Interval": 120000,

2466

"Fingerprint": ("778827225BA7DE539C5A"

2467

"7CFA59CFF7CDBD9A5920"),

2468

"CheckerRunning": False,

2469

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

2470

"ApprovalPending": False,

2471

"ApprovedByDefault": True,

2472

"LastApprovalRequest": "",

2473

"ApprovalDelay": 0,

2474

"ApprovalDuration": 1000,

2475

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

2476

"ExtendedTimeout": 900000,

2477

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

2478

"LastCheckerStatus": 0,

2479

},

2480

"barbar": {

2481

"Name": "barbar",

2482

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2483

"6ab612cff5ad227247e46c2b020f441c"),

2484

"Host": "192.0.2.3",

2485

"Enabled": True,

2486

"Timeout": 300000,

2487

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

2488

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

2489

"Interval": 120000,

2490

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2491

"F547B3A107558FCA3A27"),

2492

"CheckerRunning": True,

2493

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

2494

"ApprovalPending": False,

2495

"ApprovedByDefault": False,

2496

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

2497

"ApprovalDelay": 30000,

2498

"ApprovalDuration": 93785000,

2499

"Checker": ":",

2500

"ExtendedTimeout": 900000,

2501

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

2502

"LastCheckerStatus": -2,

2503

},

2504

}

2505

2506

def test_DumpJSON_normal(self):

2507

with self.capture_stdout_to_buffer() as buffer:

2508

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

2509

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

2510

self.assertDictEqual(self.expected_json, json_data)

2511

2512

@staticmethod

2513

@contextlib.contextmanager

2514

def capture_stdout_to_buffer():

2515

capture_buffer = io.StringIO()

2516

old_stdout = sys.stdout

2517

sys.stdout = capture_buffer

2518

try:

2519

yield capture_buffer

2520

finally:

2521

sys.stdout = old_stdout

2522

2523

def test_DumpJSON_one_client(self):

2524

with self.capture_stdout_to_buffer() as buffer:

2525

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

2526

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

2527

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

2528

self.assertDictEqual(expected_json, json_data)

2529

2530

def test_PrintTable_normal(self):

2531

with self.capture_stdout_to_buffer() as buffer:

2532

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

2533

expected_output = "\n".join((

2534

"Name Enabled Timeout Last Successful Check",

2535

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

2536

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

2537

)) + "\n"

2538

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

2539

2540

def test_PrintTable_verbose(self):

2541

with self.capture_stdout_to_buffer() as buffer:

2542

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

2543

columns = (

2544

(

2545

"Name ",

2546

"foo ",

2547

"barbar ",

2548

),(

2549

"Enabled ",

2550

"Yes ",

2551

"Yes ",

2552

),(

2553

"Timeout ",

2554

"00:05:00 ",

2555

"00:05:00 ",

2556

),(

2557

"Last Successful Check ",

2558

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

2559

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

2560

),(

2561

"Created ",

2562

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

2563

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

2564

),(

2565

"Interval ",

2566

"00:02:00 ",

2567

"00:02:00 ",

2568

),(

2569

"Host ",

2570

"foo.example.org ",

2571

"192.0.2.3 ",

2572

),(

2573

("Key ID "

2574

" "),

2575

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2576

"ea90a27cddb12 "),

2577

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2578

"46c2b020f441c "),

2579

),(

2580

"Fingerprint ",

2581

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2582

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2583

),(

2584

"Check Is Running ",

2585

"No ",

2586

"Yes ",

2587

),(

2588

"Last Enabled ",

2589

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

2590

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

2591

),(

2592

"Approval Is Pending ",

2593

"No ",

2594

"No ",

2595

),(

2596

"Approved By Default ",

2597

"Yes ",

2598

"No ",

2599

),(

2600

"Last Approval Request ",

2601

" ",

2602

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

2603

),(

2604

"Approval Delay ",

2605

"00:00:00 ",

2606

"00:00:30 ",

2607

),(

2608

"Approval Duration ",

2609

"00:00:01 ",

2610

"1T02:03:05 ",

2611

),(

2612

"Checker ",

2613

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

2614

": ",

2615

),(

2616

"Extended Timeout ",

2617

"00:15:00 ",

2618

"00:15:00 ",

2619

),(

2620

"Expires ",

2621

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

2622

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

2623

),(

2624

"Last Checker Status",

2625

"0 ",

2626

"-2 ",

2627

)

2628

)

2629

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

2630

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

2631

for rows in columns)

2632

for line in range(num_lines))

2633

+ "\n")

2634

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

2635

2636

def test_PrintTable_one_client(self):

2637

with self.capture_stdout_to_buffer() as buffer:

2638

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

2639

expected_output = "\n".join((

2640

"Name Enabled Timeout Last Successful Check",

2641

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

2642

)) + "\n"

2643

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

2644

2645

2646

class TestPropertySetterCmd(TestCommand):

2647

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

2648

2649

def runTest(self):

2650

if not hasattr(self, "command"):

2651

return # Abstract TestCase class

2652

2653

if hasattr(self, "values_to_set"):

2654

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

2655

values_to_get = getattr(self, "values_to_get",

2656

self.values_to_set)

2657

else:

2658

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

2659

values_to_get = self.values_to_get

2660

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

2661

for clientpath in self.bus.clients:

2662

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

2663

Unique())

2664

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

2665

for clientpath in self.bus.clients:

2666

value = (self.bus.clients[clientpath]

2667

[self.propname])

2668

self.assertNotIsInstance(value, Unique)

2669

self.assertEqual(value_to_get, value)

2670

2671

2672

class TestEnableCmd(TestPropertySetterCmd):

2673

command = command.Enable

2674

propname = "Enabled"

2675

values_to_get = [True]

2676

2677

2678

class TestDisableCmd(TestPropertySetterCmd):

2679

command = command.Disable

2680

propname = "Enabled"

2681

values_to_get = [False]

2682

2683

2684

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2685

command = command.BumpTimeout

2686

propname = "LastCheckedOK"

2687

values_to_get = [""]

2688

2689

2690

class TestStartCheckerCmd(TestPropertySetterCmd):

2691

command = command.StartChecker

2692

propname = "CheckerRunning"

2693

values_to_get = [True]

2694

2695

2696

class TestStopCheckerCmd(TestPropertySetterCmd):

2697

command = command.StopChecker

2698

propname = "CheckerRunning"

2699

values_to_get = [False]

2700

2701

2702

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2703

command = command.ApproveByDefault

2704

propname = "ApprovedByDefault"

2705

values_to_get = [True]

2706

2707

2708

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2709

command = command.DenyByDefault

2710

propname = "ApprovedByDefault"

2711

values_to_get = [False]

2712

2713

2714

class TestSetCheckerCmd(TestPropertySetterCmd):

2715

command = command.SetChecker

2716

propname = "Checker"

2717

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

2718

2719

2720

class TestSetHostCmd(TestPropertySetterCmd):

2721

command = command.SetHost

2722

propname = "Host"

2723

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

2724

2725

2726

class TestSetSecretCmd(TestPropertySetterCmd):

2727

command = command.SetSecret

2728

propname = "Secret"

2729

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

2730

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

2731

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

2732

2733

2734

class TestSetTimeoutCmd(TestPropertySetterCmd):

2735

command = command.SetTimeout

2736

propname = "Timeout"

2737

values_to_set = [datetime.timedelta(),

2738

datetime.timedelta(minutes=5),

2739

datetime.timedelta(seconds=1),

2740

datetime.timedelta(weeks=1),

2741

datetime.timedelta(weeks=52)]

2742

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

2743

2744

2745

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2746

command = command.SetExtendedTimeout

2747

propname = "ExtendedTimeout"

2748

values_to_set = [datetime.timedelta(),

2749

datetime.timedelta(minutes=5),

2750

datetime.timedelta(seconds=1),

2751

datetime.timedelta(weeks=1),

2752

datetime.timedelta(weeks=52)]

2753

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

2754

2755

2756

class TestSetIntervalCmd(TestPropertySetterCmd):

2757

command = command.SetInterval

2758

propname = "Interval"

2759

values_to_set = [datetime.timedelta(),

2760

datetime.timedelta(minutes=5),

2761

datetime.timedelta(seconds=1),

2762

datetime.timedelta(weeks=1),

2763

datetime.timedelta(weeks=52)]

2764

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

2765

2766

2767

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2768

command = command.SetApprovalDelay

2769

propname = "ApprovalDelay"

2770

values_to_set = [datetime.timedelta(),

2771

datetime.timedelta(minutes=5),

2772

datetime.timedelta(seconds=1),

2773

datetime.timedelta(weeks=1),

2774

datetime.timedelta(weeks=52)]

2775

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

2776

2777

2778

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2779

command = command.SetApprovalDuration

2780

propname = "ApprovalDuration"

2781

values_to_set = [datetime.timedelta(),

2782

datetime.timedelta(minutes=5),

2783

datetime.timedelta(seconds=1),

2784

datetime.timedelta(weeks=1),

2785

datetime.timedelta(weeks=52)]

2786

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

2787

2788

2789

2790

def should_only_run_tests():

2791

parser = argparse.ArgumentParser(add_help=False)

2792

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

2793

args, unknown_args = parser.parse_known_args()

2794

run_tests = args.check

2795

if run_tests:

2796

# Remove --check argument from sys.argv

2797

sys.argv[1:] = unknown_args

2798

return run_tests

2799

2800

# Add all tests from doctest strings

2801

def load_tests(loader, tests, none):

2802

import doctest

2803

tests.addTests(doctest.DocTestSuite())

2804

return tests

2805

2806

if __name__ == "__main__":

2807

try:

2808

if should_only_run_tests():

2809

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

2810

unittest.main()

2811

else:

2812

main()

2813

finally:

2814

logging.shutdown()