/mandos/trunk : revision 171

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 Control - Control or query 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.14"

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

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

106

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

109

elif pydbus is not None:

110

bus = pydbus_adapter.CachingBus(pydbus)

111

else:

112

bus = dbus_python_adapter.CachingBus(dbus_python)

113

114

try:

115

all_clients = bus.get_clients_and_properties()

116

except dbus.ConnectFailed as e:

117

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

118

sys.exit(1)

119

except dbus.Error as e:

120

log.critical(

121

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

122

sys.exit(1)

123

124

# Compile dict of (clientpath: properties) to process

125

if not clientnames:

126

clients = all_clients

127

else:

128

clients = {}

129

for name in clientnames:

130

for objpath, properties in all_clients.items():

131

if properties["Name"] == name:

132

clients[objpath] = properties

133

break

134

else:

135

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

136

sys.exit(1)

137

138

commands = commands_from_options(options)

139

140

for command in commands:

141

command.run(clients, bus)

142

143

144

def add_command_line_options(parser):

145

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

146

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

147

help="show version number and exit")

148

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

149

help="Select all clients")

150

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

151

help="Print all fields")

152

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

153

action="append_const", default=[],

154

const=command.DumpJSON(),

155

help="Dump client data in JSON format")

156

enable_disable = parser.add_mutually_exclusive_group()

157

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

158

action="append_const", default=[],

159

const=command.Enable(),

160

help="Enable client")

161

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

162

action="append_const", default=[],

163

const=command.Disable(),

164

help="disable client")

165

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

166

action="append_const", default=[],

167

const=command.BumpTimeout(),

168

help="Bump timeout for client")

169

start_stop_checker = parser.add_mutually_exclusive_group()

170

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

171

dest="commands",

172

action="append_const", default=[],

173

const=command.StartChecker(),

174

help="Start checker for client")

175

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

176

action="append_const", default=[],

177

const=command.StopChecker(),

178

help="Stop checker for client")

179

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

180

action="append_const", default=[],

181

const=command.IsEnabled(),

182

help="Check if client is enabled")

183

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

184

action="append_const", default=[],

185

const=command.Remove(),

186

help="Remove client")

187

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

188

action="append", default=[],

189

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

190

help="Set checker command for client")

191

parser.add_argument(

192

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

193

default=[], metavar="TIME",

194

type=command.SetTimeout.argparse(string_to_delta),

195

help="Set timeout for client")

196

parser.add_argument(

197

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

198

default=[], metavar="TIME",

199

type=command.SetExtendedTimeout.argparse(string_to_delta),

200

help="Set extended timeout for client")

201

parser.add_argument(

202

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

203

default=[], metavar="TIME",

204

type=command.SetInterval.argparse(string_to_delta),

205

help="Set checker interval for client")

206

approve_deny_default = parser.add_mutually_exclusive_group()

207

approve_deny_default.add_argument(

208

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

209

action="append_const", default=[],

210

const=command.ApproveByDefault(),

211

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

212

approve_deny_default.add_argument(

213

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

214

action="append_const", default=[],

215

const=command.DenyByDefault(),

216

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

217

parser.add_argument(

218

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

219

default=[], metavar="TIME",

220

type=command.SetApprovalDelay.argparse(string_to_delta),

221

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

222

parser.add_argument(

223

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

224

default=[], metavar="TIME",

225

type=command.SetApprovalDuration.argparse(string_to_delta),

226

help="Set duration of one client approval")

227

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

228

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

229

type=command.SetHost,

230

help="Set host for client")

231

parser.add_argument(

232

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

233

default=[], metavar="FILENAME",

234

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

235

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

236

approve_deny = parser.add_mutually_exclusive_group()

237

approve_deny.add_argument(

238

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

239

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

240

help="Approve any current client request")

241

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

242

action="append_const", default=[],

243

const=command.Deny(),

244

help="Deny any current client request")

245

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

246

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

247

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

248

help="Run self-test")

249

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

250

251

252

def string_to_delta(interval):

253

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

254

255

try:

256

return rfc3339_duration_to_delta(interval)

257

except ValueError as e:

258

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

259

" ".join(e.args))

260

return parse_pre_1_6_1_interval(interval)

261

262

263

def rfc3339_duration_to_delta(duration):

264

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

265

266

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

267

True

268

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

269

True

270

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

271

True

272

>>> # 60 months

273

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

274

True

275

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

276

True

277

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

278

True

279

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

280

True

281

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

282

True

283

>>> # Can not be empty:

284

>>> rfc3339_duration_to_delta("")

285

Traceback (most recent call last):

286

...

287

ValueError: Invalid RFC 3339 duration: ""

288

>>> # Must start with "P":

289

>>> rfc3339_duration_to_delta("1D")

290

Traceback (most recent call last):

291

...

292

ValueError: Invalid RFC 3339 duration: "1D"

293

>>> # Must use correct order

294

>>> rfc3339_duration_to_delta("PT1S2M")

295

Traceback (most recent call last):

296

...

297

ValueError: Invalid RFC 3339 duration: "PT1S2M"

298

>>> # Time needs time marker

299

>>> rfc3339_duration_to_delta("P1H2S")

300

Traceback (most recent call last):

301

...

302

ValueError: Invalid RFC 3339 duration: "P1H2S"

303

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

304

>>> rfc3339_duration_to_delta("P1D2W")

305

Traceback (most recent call last):

306

...

307

ValueError: Invalid RFC 3339 duration: "P1D2W"

308

>>> rfc3339_duration_to_delta("P2W2H")

309

Traceback (most recent call last):

310

...

311

ValueError: Invalid RFC 3339 duration: "P2W2H"

312

"""

313

314

# Parsing an RFC 3339 duration with regular expressions is not

315

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

316

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

317

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

318

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

319

# avoid excessive use of external libraries.

320

321

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

322

Token = collections.namedtuple("Token", (

323

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

324

# a "group" containing digits

325

"value", # datetime.timedelta or None

326

"followers")) # Tokens valid after this token

327

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

328

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

329

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

330

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

331

datetime.timedelta(seconds=1),

332

frozenset((token_end, )))

333

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

334

datetime.timedelta(minutes=1),

335

frozenset((token_second, token_end)))

336

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

337

datetime.timedelta(hours=1),

338

frozenset((token_minute, token_end)))

339

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

340

None,

341

frozenset((token_hour, token_minute,

342

token_second)))

343

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

344

datetime.timedelta(days=1),

345

frozenset((token_time, token_end)))

346

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

347

datetime.timedelta(weeks=4),

348

frozenset((token_day, token_end)))

349

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

350

datetime.timedelta(weeks=52),

351

frozenset((token_month, token_end)))

352

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

353

datetime.timedelta(weeks=1),

354

frozenset((token_end, )))

355

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

356

frozenset((token_year, token_month,

357

token_day, token_time,

358

token_week)))

359

# Define starting values:

360

# Value so far

361

value = datetime.timedelta()

362

found_token = None

363

# Following valid tokens

364

followers = frozenset((token_duration, ))

365

# String left to parse

366

s = duration

367

# Loop until end token is found

368

while found_token is not token_end:

369

# Search for any currently valid tokens

370

for token in followers:

371

match = token.regexp.match(s)

372

if match is not None:

373

# Token found

374

if token.value is not None:

375

# Value found, parse digits

376

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

377

# Add to value so far

378

value += factor * token.value

379

# Strip token from string

380

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

381

# Go to found token

382

found_token = token

383

# Set valid next tokens

384

followers = found_token.followers

385

break

386

else:

387

# No currently valid tokens were found

388

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

389

.format(duration))

390

# End token found

391

return value

392

393

394

def parse_pre_1_6_1_interval(interval):

395

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

396

and return a datetime.timedelta

397

398

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

399

True

400

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

401

True

402

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

403

True

404

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

405

True

406

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

407

True

408

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

409

True

410

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

411

True

412

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

413

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

414

True

415

416

"""

417

418

value = datetime.timedelta(0)

419

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

420

421

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

422

if suffix == "d":

423

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

424

elif suffix == "s":

425

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

426

elif suffix == "m":

427

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

428

elif suffix == "h":

429

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

430

elif suffix == "w":

431

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

432

elif suffix == "":

433

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

434

return value

435

436

437

def check_option_syntax(parser, options):

438

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

439

argparse"""

440

441

def has_commands(options, commands=None):

442

if commands is None:

443

commands = (command.Enable,

444

command.Disable,

445

command.BumpTimeout,

446

command.StartChecker,

447

command.StopChecker,

448

command.IsEnabled,

449

command.Remove,

450

command.SetChecker,

451

command.SetTimeout,

452

command.SetExtendedTimeout,

453

command.SetInterval,

454

command.ApproveByDefault,

455

command.DenyByDefault,

456

command.SetApprovalDelay,

457

command.SetApprovalDuration,

458

command.SetHost,

459

command.SetSecret,

460

command.Approve,

461

command.Deny)

462

return any(isinstance(cmd, commands)

463

for cmd in options.commands)

464

465

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

466

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

467

if options.verbose and has_commands(options):

468

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

469

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

470

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

471

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

472

if options.all and not has_commands(options):

473

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

474

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

475

and len(options.client) > 1):

476

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

477

if (len(options.commands) > 1

478

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

479

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

480

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

481

482

483

class dbus:

484

485

class SystemBus:

486

487

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

488

def get_managed_objects(self, busname, objectpath):

489

return self.call_method("GetManagedObjects", busname,

490

objectpath,

491

self.object_manager_iface)

492

493

properties_iface = "org.freedesktop.DBus.Properties"

494

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

495

value):

496

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

497

self.properties_iface, interface, key,

498

value)

499

500

def call_method(self, methodname, busname, objectpath,

501

interface, *args):

502

raise NotImplementedError()

503

504

505

class MandosBus(SystemBus):

506

busname_domain = "se.recompile"

507

busname = busname_domain + ".Mandos"

508

server_path = "/"

509

server_interface = busname_domain + ".Mandos"

510

client_interface = busname_domain + ".Mandos.Client"

511

del busname_domain

512

513

def get_clients_and_properties(self):

514

managed_objects = self.get_managed_objects(

515

self.busname, self.server_path)

516

return {objpath: properties[self.client_interface]

517

for objpath, properties in managed_objects.items()

518

if self.client_interface in properties}

519

520

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

521

return self.set_property(self.busname, objectpath,

522

self.client_interface, key,

523

value)

524

525

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

526

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

527

self.client_interface, *args)

528

529

def call_server_method(self, method, *args):

530

return self.call_method(method, self.busname,

531

self.server_path,

532

self.server_interface, *args)

533

534

class Error(Exception):

535

pass

536

537

class ConnectFailed(Error):

538

pass

539

540

541

class dbus_python_adapter:

542

543

class SystemBus(dbus.MandosBus):

544

"""Use dbus-python"""

545

546

def __init__(self, module=dbus_python):

547

self.dbus_python = module

548

self.bus = self.dbus_python.SystemBus()

549

550

@contextlib.contextmanager

551

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

552

try:

553

yield

554

except self.dbus_python.exceptions.DBusException as e:

555

# This does what "raise from" would do

556

exc = exception_class(*e.args)

557

exc.__cause__ = e

558

raise exc

559

560

def call_method(self, methodname, busname, objectpath,

561

interface, *args):

562

proxy_object = self.get_object(busname, objectpath)

563

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

564

interface, methodname,

565

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

566

method = getattr(proxy_object, methodname)

567

with self.convert_exception():

568

with dbus_python_adapter.SilenceLogger(

569

"dbus.proxies"):

570

value = method(*args, dbus_interface=interface)

571

return self.type_filter(value)

572

573

def get_object(self, busname, objectpath):

574

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

575

busname, objectpath)

576

with self.convert_exception(dbus.ConnectFailed):

577

return self.bus.get_object(busname, objectpath)

578

579

def type_filter(self, value):

580

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

581

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

582

return bool(value)

583

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

584

return str(value)

585

# Also recurse into dictionaries

586

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

587

return {self.type_filter(key):

588

self.type_filter(subval)

589

for key, subval in value.items()}

590

return value

591

592

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

593

if key == "Secret":

594

if not isinstance(value, bytes):

595

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

596

value = self.dbus_python.ByteArray(value)

597

return self.set_property(self.busname, objectpath,

598

self.client_interface, key,

599

value)

600

601

class SilenceLogger:

602

"Simple context manager to silence a particular logger"

603

def __init__(self, loggername):

604

self.logger = logging.getLogger(loggername)

605

606

def __enter__(self):

607

self.logger.addFilter(self.nullfilter)

608

609

class NullFilter(logging.Filter):

610

def filter(self, record):

611

return False

612

613

nullfilter = NullFilter()

614

615

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

616

self.logger.removeFilter(self.nullfilter)

617

618

619

class CachingBus(SystemBus):

620

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

621

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

622

self.object_cache = {}

623

super(dbus_python_adapter.CachingBus,

624

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

625

def get_object(self, busname, objectpath):

626

try:

627

return self.object_cache[(busname, objectpath)]

628

except KeyError:

629

new_object = super(

630

dbus_python_adapter.CachingBus,

631

self).get_object(busname, objectpath)

632

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

633

return new_object

634

635

636

class pydbus_adapter:

637

class SystemBus(dbus.MandosBus):

638

def __init__(self, module=pydbus):

639

self.pydbus = module

640

self.bus = self.pydbus.SystemBus()

641

642

@contextlib.contextmanager

643

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

644

try:

645

yield

646

except gi.repository.GLib.Error as e:

647

# This does what "raise from" would do

648

exc = exception_class(*e.args)

649

exc.__cause__ = e

650

raise exc

651

652

def call_method(self, methodname, busname, objectpath,

653

interface, *args):

654

proxy_object = self.get(busname, objectpath)

655

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

656

interface, methodname,

657

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

658

method = getattr(proxy_object[interface], methodname)

659

with self.convert_exception():

660

return method(*args)

661

662

def get(self, busname, objectpath):

663

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

664

busname, objectpath)

665

with self.convert_exception(dbus.ConnectFailed):

666

if sys.version_info.major <= 2:

667

with warnings.catch_warnings():

668

warnings.filterwarnings(

669

"ignore", "", DeprecationWarning,

670

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

671

return self.bus.get(busname, objectpath)

672

else:

673

return self.bus.get(busname, objectpath)

674

675

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

676

value):

677

proxy_object = self.get(busname, objectpath)

678

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

679

objectpath, self.properties_iface, interface,

680

key, value)

681

setattr(proxy_object[interface], key, value)

682

683

class CachingBus(SystemBus):

684

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

685

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

686

self.object_cache = {}

687

super(pydbus_adapter.CachingBus,

688

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

689

def get(self, busname, objectpath):

690

try:

691

return self.object_cache[(busname, objectpath)]

692

except KeyError:

693

new_object = (super(pydbus_adapter.CachingBus, self)

694

.get(busname, objectpath))

695

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

696

return new_object

697

698

699

class dbussy_adapter:

700

class SystemBus(dbus.SystemBus):

701

"""Use DBussy"""

702

703

def __init__(self, dbussy, ravel):

704

self.dbussy = dbussy

705

self.ravel = ravel

706

self.bus = ravel.system_bus()

707

708

@contextlib.contextmanager

709

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

710

try:

711

yield

712

except self.dbussy.DBusError as e:

713

# This does what "raise from" would do

714

exc = exception_class(*e.args)

715

exc.__cause__ = e

716

raise exc

717

718

def call_method(self, methodname, busname, objectpath,

719

interface, *args):

720

proxy_object = self.get_object(busname, objectpath)

721

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

722

interface, methodname,

723

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

724

iface = proxy_object.get_interface(interface)

725

method = getattr(iface, methodname)

726

with self.convert_exception(dbus.Error):

727

value = method(*args)

728

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

729

# list of one element with the return value

730

if value:

731

return self.type_filter(value[0])

732

733

def get_object(self, busname, objectpath):

734

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

735

busname, objectpath)

736

with self.convert_exception(dbus.ConnectFailed):

737

return self.bus[busname][objectpath]

738

739

def type_filter(self, value):

740

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

741

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

742

# Tuple[dbussy.DBUS.Signature, Any]

743

if isinstance(value, tuple):

744

if (len(value) == 2

745

and isinstance(value[0],

746

self.dbussy.DBUS.Signature)):

747

return self.type_filter(value[1])

748

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

749

return str(value)

750

# Also recurse into dictionaries

751

elif isinstance(value, dict):

752

return {self.type_filter(key):

753

self.type_filter(subval)

754

for key, subval in value.items()}

755

return value

756

757

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

758

value):

759

proxy_object = self.get_object(busname, objectpath)

760

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

761

objectpath, self.properties_iface, interface,

762

key, value)

763

if key == "Secret":

764

# DBussy wants a Byte Array to be a sequence of

765

# values, not a byte string

766

value = tuple(value)

767

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

768

769

class MandosBus(SystemBus, dbus.MandosBus):

770

pass

771

772

class CachingBus(MandosBus):

773

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

774

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

775

self.object_cache = {}

776

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

777

**kwargs)

778

def get_object(self, busname, objectpath):

779

try:

780

return self.object_cache[(busname, objectpath)]

781

except KeyError:

782

new_object = super(

783

dbussy_adapter.CachingBus,

784

self).get_object(busname, objectpath)

785

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

786

return new_object

787

788

789

def commands_from_options(options):

790

791

commands = list(options.commands)

792

793

def find_cmd(cmd, commands):

794

i = 0

795

for i, c in enumerate(commands):

796

if isinstance(c, cmd):

797

return i

798

return i+1

799

800

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

801

# to occur ahead of command.Remove.

802

index_of_remove = find_cmd(command.Remove, commands)

803

before_remove = commands[:index_of_remove]

804

after_remove = commands[index_of_remove:]

805

cleaned_after = []

806

for cmd in after_remove:

807

if isinstance(cmd, command.Deny):

808

before_remove.append(cmd)

809

else:

810

cleaned_after.append(cmd)

811

if cleaned_after != after_remove:

812

commands = before_remove + cleaned_after

813

814

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

815

# optionally verbosely

816

if not commands:

817

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

818

819

return commands

820

821

822

class command:

823

"""A namespace for command classes"""

824

825

class Base:

826

"""Abstract base class for commands"""

827

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

828

"""Normal commands should implement run_on_one_client(),

829

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

830

override this run() method instead.

831

"""

832

self.bus = bus

833

for client, properties in clients.items():

834

self.run_on_one_client(client, properties)

835

836

837

class IsEnabled(Base):

838

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

839

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

840

if properties["Enabled"]:

841

sys.exit(0)

842

sys.exit(1)

843

844

845

class Approve(Base):

846

def run_on_one_client(self, client, properties):

847

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

848

849

850

class Deny(Base):

851

def run_on_one_client(self, client, properties):

852

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

853

854

855

class Remove(Base):

856

def run(self, clients, bus):

857

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

858

bus.call_server_method("RemoveClient", clientpath)

859

860

861

class Output(Base):

862

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

863

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

864

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

865

"Fingerprint", "CheckerRunning",

866

"LastEnabled", "ApprovalPending",

867

"ApprovedByDefault", "LastApprovalRequest",

868

"ApprovalDelay", "ApprovalDuration",

869

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

870

"LastCheckerStatus")

871

872

873

class DumpJSON(Output):

874

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

875

data = {properties["Name"]:

876

{key: properties[key]

877

for key in self.all_keywords}

878

for properties in clients.values()}

879

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

880

881

882

class PrintTable(Output):

883

def __init__(self, verbose=False):

884

self.verbose = verbose

885

886

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

887

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

888

"LastCheckedOK")

889

keywords = default_keywords

890

if self.verbose:

891

keywords = self.all_keywords

892

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

893

894

class TableOfClients:

895

tableheaders = {

896

"Name": "Name",

897

"Enabled": "Enabled",

898

"Timeout": "Timeout",

899

"LastCheckedOK": "Last Successful Check",

900

"LastApprovalRequest": "Last Approval Request",

901

"Created": "Created",

902

"Interval": "Interval",

903

"Host": "Host",

904

"Fingerprint": "Fingerprint",

905

"KeyID": "Key ID",

906

"CheckerRunning": "Check Is Running",

907

"LastEnabled": "Last Enabled",

908

"ApprovalPending": "Approval Is Pending",

909

"ApprovedByDefault": "Approved By Default",

910

"ApprovalDelay": "Approval Delay",

911

"ApprovalDuration": "Approval Duration",

912

"Checker": "Checker",

913

"ExtendedTimeout": "Extended Timeout",

914

"Expires": "Expires",

915

"LastCheckerStatus": "Last Checker Status",

916

}

917

918

def __init__(self, clients, keywords):

919

self.clients = clients

920

self.keywords = keywords

921

922

def __str__(self):

923

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

924

925

if sys.version_info.major == 2:

926

__unicode__ = __str__

927

def __str__(self):

928

return str(self).encode(

929

locale.getpreferredencoding())

930

931

def rows(self):

932

format_string = self.row_formatting_string()

933

rows = [self.header_line(format_string)]

934

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

935

for client in self.clients)

936

return rows

937

938

def row_formatting_string(self):

939

"Format string used to format table rows"

940

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

941

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

942

*(len(self.string_from_client(client,

943

key))

944

for client in self.clients)),

945

key=key)

946

for key in self.keywords)

947

948

def string_from_client(self, client, key):

949

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

950

951

@classmethod

952

def valuetostring(cls, value, keyword):

953

if isinstance(value, bool):

954

return "Yes" if value else "No"

955

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

956

"ApprovalDuration", "ExtendedTimeout"):

957

return cls.milliseconds_to_string(value)

958

return str(value)

959

960

def header_line(self, format_string):

961

return format_string.format(**self.tableheaders)

962

963

def client_line(self, client, format_string):

964

return format_string.format(

965

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

966

for key in self.keywords})

967

968

@staticmethod

969

def milliseconds_to_string(ms):

970

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

971

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

972

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

973

if td.days else "",

974

hours=td.seconds // 3600,

975

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

976

seconds=td.seconds % 60))

977

978

979

class PropertySetter(Base):

980

"Abstract class for Actions for setting one client property"

981

982

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

983

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

984

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

985

self.value_to_set)

986

987

@property

988

def propname(self):

989

raise NotImplementedError()

990

991

992

class Enable(PropertySetter):

993

propname = "Enabled"

994

value_to_set = True

995

996

997

class Disable(PropertySetter):

998

propname = "Enabled"

999

value_to_set = False

1000

1001

1002

class BumpTimeout(PropertySetter):

1003

propname = "LastCheckedOK"

1004

value_to_set = ""

1005

1006

1007

class StartChecker(PropertySetter):

1008

propname = "CheckerRunning"

1009

value_to_set = True

1010

1011

1012

class StopChecker(PropertySetter):

1013

propname = "CheckerRunning"

1014

value_to_set = False

1015

1016

1017

class ApproveByDefault(PropertySetter):

1018

propname = "ApprovedByDefault"

1019

value_to_set = True

1020

1021

1022

class DenyByDefault(PropertySetter):

1023

propname = "ApprovedByDefault"

1024

value_to_set = False

1025

1026

1027

class PropertySetterValue(PropertySetter):

1028

"""Abstract class for PropertySetter recieving a value as

1029

constructor argument instead of a class attribute."""

1030

def __init__(self, value):

1031

self.value_to_set = value

1032

1033

@classmethod

1034

def argparse(cls, argtype):

1035

def cmdtype(arg):

1036

return cls(argtype(arg))

1037

return cmdtype

1038

1039

class SetChecker(PropertySetterValue):

1040

propname = "Checker"

1041

1042

1043

class SetHost(PropertySetterValue):

1044

propname = "Host"

1045

1046

1047

class SetSecret(PropertySetterValue):

1048

propname = "Secret"

1049

1050

@property

1051

def value_to_set(self):

1052

return self._vts

1053

1054

@value_to_set.setter

1055

def value_to_set(self, value):

1056

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

1057

self._vts = value.read()

1058

value.close()

1059

1060

1061

class PropertySetterValueMilliseconds(PropertySetterValue):

1062

"""Abstract class for PropertySetterValue taking a value

1063

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

1064

milliseconds."""

1065

1066

@property

1067

def value_to_set(self):

1068

return self._vts

1069

1070

@value_to_set.setter

1071

def value_to_set(self, value):

1072

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

1073

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

1074

1075

1076

class SetTimeout(PropertySetterValueMilliseconds):

1077

propname = "Timeout"

1078

1079

1080

class SetExtendedTimeout(PropertySetterValueMilliseconds):

1081

propname = "ExtendedTimeout"

1082

1083

1084

class SetInterval(PropertySetterValueMilliseconds):

1085

propname = "Interval"

1086

1087

1088

class SetApprovalDelay(PropertySetterValueMilliseconds):

1089

propname = "ApprovalDelay"

1090

1091

1092

class SetApprovalDuration(PropertySetterValueMilliseconds):

1093

propname = "ApprovalDuration"

1094

1095

1096

1097

class TestCaseWithAssertLogs(unittest.TestCase):

1098

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

1099

1100

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

1101

@contextlib.contextmanager

1102

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

1103

capturing_handler = self.CapturingLevelHandler(level)

1104

old_level = logger.level

1105

old_propagate = logger.propagate

1106

logger.addHandler(capturing_handler)

1107

logger.setLevel(level)

1108

logger.propagate = False

1109

try:

1110

yield capturing_handler.watcher

1111

finally:

1112

logger.propagate = old_propagate

1113

logger.removeHandler(capturing_handler)

1114

logger.setLevel(old_level)

1115

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

1116

0)

1117

1118

class CapturingLevelHandler(logging.Handler):

1119

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

1120

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

1121

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

1122

def emit(self, record):

1123

self.watcher.records.append(record)

1124

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

1125

1126

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1127

("records",

1128

"output"))

1129

1130

1131

class Unique:

1132

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

1133

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

1134

1135

1136

class Test_string_to_delta(TestCaseWithAssertLogs):

1137

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

1138

# rfc3339_duration_to_delta() already has more comprehensive

1139

# tests, which are run by doctest.

1140

1141

def test_rfc3339_zero_seconds(self):

1142

self.assertEqual(datetime.timedelta(),

1143

string_to_delta("PT0S"))

1144

1145

def test_rfc3339_zero_days(self):

1146

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

1147

1148

def test_rfc3339_one_second(self):

1149

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

1150

string_to_delta("PT1S"))

1151

1152

def test_rfc3339_two_hours(self):

1153

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

1154

string_to_delta("PT2H"))

1155

1156

def test_falls_back_to_pre_1_6_1_with_warning(self):

1157

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

1158

value = string_to_delta("2h")

1159

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

1160

1161

1162

class Test_check_option_syntax(unittest.TestCase):

1163

def setUp(self):

1164

self.parser = argparse.ArgumentParser()

1165

add_command_line_options(self.parser)

1166

1167

def test_actions_requires_client_or_all(self):

1168

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

1169

args = self.actionargs(action, value)

1170

with self.assertParseError():

1171

self.parse_args(args)

1172

1173

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

1174

# check_option_syntax()

1175

actions = {

1176

"--enable": None,

1177

"--disable": None,

1178

"--bump-timeout": None,

1179

"--start-checker": None,

1180

"--stop-checker": None,

1181

"--is-enabled": None,

1182

"--remove": None,

1183

"--checker": "x",

1184

"--timeout": "PT0S",

1185

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

1186

"--interval": "PT0S",

1187

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

1188

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

1189

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

1190

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

1191

"--host": "hostname",

1192

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

1193

"--approve": None,

1194

"--deny": None,

1195

}

1196

1197

@staticmethod

1198

def actionargs(action, value, *args):

1199

if value is not None:

1200

return [action, value] + list(args)

1201

else:

1202

return [action] + list(args)

1203

1204

@contextlib.contextmanager

1205

def assertParseError(self):

1206

with self.assertRaises(SystemExit) as e:

1207

with self.redirect_stderr_to_devnull():

1208

yield

1209

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

1210

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

1211

# /argparse.html#exiting-methods

1212

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

1213

1214

def parse_args(self, args):

1215

options = self.parser.parse_args(args)

1216

check_option_syntax(self.parser, options)

1217

1218

@staticmethod

1219

@contextlib.contextmanager

1220

def redirect_stderr_to_devnull():

1221

old_stderr = sys.stderr

1222

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

1223

sys.stderr = null

1224

try:

1225

yield

1226

finally:

1227

sys.stderr = old_stderr

1228

1229

def check_option_syntax(self, options):

1230

check_option_syntax(self.parser, options)

1231

1232

def test_actions_all_conflicts_with_verbose(self):

1233

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

1234

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

1235

"--verbose")

1236

with self.assertParseError():

1237

self.parse_args(args)

1238

1239

def test_actions_with_client_conflicts_with_verbose(self):

1240

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

1241

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

1242

"client")

1243

with self.assertParseError():

1244

self.parse_args(args)

1245

1246

def test_dump_json_conflicts_with_verbose(self):

1247

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

1248

with self.assertParseError():

1249

self.parse_args(args)

1250

1251

def test_dump_json_conflicts_with_action(self):

1252

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

1253

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

1254

with self.assertParseError():

1255

self.parse_args(args)

1256

1257

def test_all_can_not_be_alone(self):

1258

args = ["--all"]

1259

with self.assertParseError():

1260

self.parse_args(args)

1261

1262

def test_all_is_ok_with_any_action(self):

1263

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

1264

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

1265

self.parse_args(args)

1266

1267

def test_any_action_is_ok_with_one_client(self):

1268

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

1269

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

1270

self.parse_args(args)

1271

1272

def test_one_client_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, "client")

1277

self.parse_args(args)

1278

1279

def test_two_clients_with_all_actions_except_is_enabled(self):

1280

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

1281

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

1282

continue

1283

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

1284

"client2")

1285

self.parse_args(args)

1286

1287

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1288

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

1289

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

1290

continue

1291

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

1292

"client2")

1293

self.parse_args(args)

1294

1295

def test_is_enabled_fails_without_client(self):

1296

args = ["--is-enabled"]

1297

with self.assertParseError():

1298

self.parse_args(args)

1299

1300

def test_is_enabled_fails_with_two_clients(self):

1301

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

1302

with self.assertParseError():

1303

self.parse_args(args)

1304

1305

def test_remove_can_only_be_combined_with_action_deny(self):

1306

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

1307

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

1308

continue

1309

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

1310

"--remove")

1311

with self.assertParseError():

1312

self.parse_args(args)

1313

1314

1315

class Test_dbus_exceptions(unittest.TestCase):

1316

1317

def test_dbus_ConnectFailed_is_Error(self):

1318

with self.assertRaises(dbus.Error):

1319

raise dbus.ConnectFailed()

1320

1321

1322

class Test_dbus_MandosBus(unittest.TestCase):

1323

1324

class MockMandosBus(dbus.MandosBus):

1325

def __init__(self):

1326

self._name = "se.recompile.Mandos"

1327

self._server_path = "/"

1328

self._server_interface = "se.recompile.Mandos"

1329

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

1330

self.calls = []

1331

self.call_method_return = Unique()

1332

1333

def call_method(self, methodname, busname, objectpath,

1334

interface, *args):

1335

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

1336

interface, args))

1337

return self.call_method_return

1338

1339

def setUp(self):

1340

self.bus = self.MockMandosBus()

1341

1342

def test_set_client_property(self):

1343

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

1344

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

1345

"org.freedesktop.DBus.Properties",

1346

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

1347

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

1348

1349

def test_call_client_method(self):

1350

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

1351

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

1352

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

1353

self.bus._client_interface, ())

1354

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

1355

1356

def test_call_client_method_with_args(self):

1357

args = (Unique(), Unique())

1358

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

1359

*args)

1360

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

1361

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

1362

self.bus._client_interface,

1363

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

1364

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

1365

1366

def test_get_clients_and_properties(self):

1367

managed_objects = {

1368

"objectpath": {

1369

self.bus._client_interface: {

1370

"key": "value",

1371

"bool": True,

1372

},

1373

"irrelevant_interface": {

1374

"key": "othervalue",

1375

"bool": False,

1376

},

1377

},

1378

"other_objectpath": {

1379

"other_irrelevant_interface": {

1380

"key": "value 3",

1381

"bool": None,

1382

},

1383

},

1384

}

1385

expected_clients_and_properties = {

1386

"objectpath": {

1387

"key": "value",

1388

"bool": True,

1389

}

1390

}

1391

self.bus.call_method_return = managed_objects

1392

ret = self.bus.get_clients_and_properties()

1393

self.assertDictEqual(expected_clients_and_properties, ret)

1394

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

1395

self.bus._server_path,

1396

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

1397

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

1398

1399

def test_call_server_method(self):

1400

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

1401

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

1402

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

1403

self.bus._server_path,

1404

self.bus._server_interface, ())

1405

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

1406

1407

def test_call_server_method_with_args(self):

1408

args = (Unique(), Unique())

1409

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

1410

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

1411

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

1412

self.bus._server_path,

1413

self.bus._server_interface,

1414

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

1415

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

1416

1417

1418

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1419

1420

def MockDBusPython_func(self, func):

1421

class mock_dbus_python:

1422

"""mock dbus-python module"""

1423

class exceptions:

1424

"""Pseudo-namespace"""

1425

class DBusException(Exception):

1426

pass

1427

class SystemBus:

1428

@staticmethod

1429

def get_object(busname, objectpath):

1430

DBusObject = collections.namedtuple(

1431

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

1432

def method(*args, **kwargs):

1433

self.assertEqual({"dbus_interface":

1434

"interface"},

1435

kwargs)

1436

return func(*args)

1437

def set_property(interface, key, value,

1438

dbus_interface=None):

1439

self.assertEqual(

1440

"org.freedesktop.DBus.Properties",

1441

dbus_interface)

1442

self.assertEqual("Secret", key)

1443

return func(interface, key, value,

1444

dbus_interface=dbus_interface)

1445

return DBusObject(methodname=method,

1446

Set=set_property)

1447

class Boolean:

1448

def __init__(self, value):

1449

self.value = bool(value)

1450

def __bool__(self):

1451

return self.value

1452

if sys.version_info.major == 2:

1453

__nonzero__ = __bool__

1454

class ObjectPath(str):

1455

pass

1456

class Dictionary(dict):

1457

pass

1458

class ByteArray(bytes):

1459

pass

1460

return mock_dbus_python

1461

1462

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

1463

interface, *args):

1464

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

1465

return bus.call_method(methodname, busname, objectpath,

1466

interface, *args)

1467

1468

def test_call_method_returns(self):

1469

expected_method_return = Unique()

1470

method_args = (Unique(), Unique())

1471

def func(*args):

1472

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

1473

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

1474

self.assertIs(marg, arg)

1475

return expected_method_return

1476

mock_dbus_python = self.MockDBusPython_func(func)

1477

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1478

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

1479

"objectpath", "interface",

1480

*method_args)

1481

self.assertIs(ret, expected_method_return)

1482

1483

def test_call_method_filters_bool_true(self):

1484

def func():

1485

return method_return

1486

mock_dbus_python = self.MockDBusPython_func(func)

1487

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1488

method_return = mock_dbus_python.Boolean(True)

1489

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

1490

"objectpath", "interface")

1491

self.assertTrue(ret)

1492

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1493

1494

def test_call_method_filters_bool_false(self):

1495

def func():

1496

return method_return

1497

mock_dbus_python = self.MockDBusPython_func(func)

1498

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1499

method_return = mock_dbus_python.Boolean(False)

1500

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

1501

"objectpath", "interface")

1502

self.assertFalse(ret)

1503

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1504

1505

def test_call_method_filters_objectpath(self):

1506

def func():

1507

return method_return

1508

mock_dbus_python = self.MockDBusPython_func(func)

1509

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1510

method_return = mock_dbus_python.ObjectPath("objectpath")

1511

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

1512

"objectpath", "interface")

1513

self.assertEqual("objectpath", ret)

1514

self.assertIsNot("objectpath", ret)

1515

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1516

1517

def test_call_method_filters_booleans_in_dict(self):

1518

def func():

1519

return method_return

1520

mock_dbus_python = self.MockDBusPython_func(func)

1521

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1522

method_return = mock_dbus_python.Dictionary(

1523

{mock_dbus_python.Boolean(True):

1524

mock_dbus_python.Boolean(False),

1525

mock_dbus_python.Boolean(False):

1526

mock_dbus_python.Boolean(True)})

1527

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

1528

"objectpath", "interface")

1529

expected_method_return = {True: False,

1530

False: True}

1531

self.assertEqual(expected_method_return, ret)

1532

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1533

1534

def test_call_method_filters_objectpaths_in_dict(self):

1535

def func():

1536

return method_return

1537

mock_dbus_python = self.MockDBusPython_func(func)

1538

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1539

method_return = mock_dbus_python.Dictionary(

1540

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1541

mock_dbus_python.ObjectPath("objectpath_value_1"),

1542

mock_dbus_python.ObjectPath("objectpath_key_2"):

1543

mock_dbus_python.ObjectPath("objectpath_value_2")})

1544

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

1545

"objectpath", "interface")

1546

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

1547

for key, value in

1548

method_return.items()}

1549

self.assertEqual(expected_method_return, ret)

1550

self.assertIsInstance(ret, dict)

1551

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1552

1553

def test_call_method_filters_dict_in_dict(self):

1554

def func():

1555

return method_return

1556

mock_dbus_python = self.MockDBusPython_func(func)

1557

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1558

method_return = mock_dbus_python.Dictionary(

1559

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

1560

"key12": "value12"}),

1561

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

1562

"key22": "value22"})})

1563

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

1564

"objectpath", "interface")

1565

expected_method_return = {

1566

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

1567

"key12": "value12"},

1568

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

1569

"key22": "value22"},

1570

}

1571

self.assertEqual(expected_method_return, ret)

1572

self.assertIsInstance(ret, dict)

1573

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1574

for key, value in ret.items():

1575

self.assertIsInstance(value, dict)

1576

self.assertEqual(expected_method_return[key], value)

1577

self.assertNotIsInstance(value,

1578

mock_dbus_python.Dictionary)

1579

1580

def test_call_method_filters_dict_three_deep(self):

1581

def func():

1582

return method_return

1583

mock_dbus_python = self.MockDBusPython_func(func)

1584

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1585

method_return = mock_dbus_python.Dictionary(

1586

{"key1":

1587

mock_dbus_python.Dictionary(

1588

{"key2":

1589

mock_dbus_python.Dictionary(

1590

{"key3":

1591

mock_dbus_python.Boolean(True),

1592

}),

1593

}),

1594

})

1595

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

1596

"objectpath", "interface")

1597

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

1598

self.assertEqual(expected_method_return, ret)

1599

self.assertIsInstance(ret, dict)

1600

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1601

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

1602

self.assertNotIsInstance(ret["key1"],

1603

mock_dbus_python.Dictionary)

1604

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

1605

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

1606

mock_dbus_python.Dictionary)

1607

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

1608

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

1609

mock_dbus_python.Boolean)

1610

1611

def test_call_method_handles_exception(self):

1612

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

1613

1614

def func():

1615

dbus_logger.error("Test")

1616

raise mock_dbus_python.exceptions.DBusException()

1617

1618

mock_dbus_python = self.MockDBusPython_func(func)

1619

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1620

1621

class CountingHandler(logging.Handler):

1622

count = 0

1623

def emit(self, record):

1624

self.count += 1

1625

1626

counting_handler = CountingHandler()

1627

1628

dbus_logger.addHandler(counting_handler)

1629

1630

try:

1631

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

1632

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

1633

"objectpath", "interface")

1634

finally:

1635

dbus_logger.removeFilter(counting_handler)

1636

1637

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

1638

1639

# Make sure the dbus logger was suppressed

1640

self.assertEqual(0, counting_handler.count)

1641

1642

def test_Set_Secret_sends_bytearray(self):

1643

ret = [None]

1644

def func(*args, **kwargs):

1645

ret[0] = (args, kwargs)

1646

mock_dbus_python = self.MockDBusPython_func(func)

1647

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1648

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

1649

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

1650

mock_dbus_python.ByteArray(b"value")),

1651

{"dbus_interface":

1652

"org.freedesktop.DBus.Properties"})

1653

self.assertEqual(expected_call, ret[0])

1654

if sys.version_info.major == 2:

1655

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

1656

mock_dbus_python.ByteArray)

1657

1658

def test_get_object_converts_to_correct_exception(self):

1659

bus = dbus_python_adapter.SystemBus(

1660

self.fake_dbus_python_raises_exception_on_connect)

1661

with self.assertRaises(dbus.ConnectFailed):

1662

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

1663

"objectpath", "interface")

1664

1665

class fake_dbus_python_raises_exception_on_connect:

1666

"""fake dbus-python module"""

1667

class exceptions:

1668

"""Pseudo-namespace"""

1669

class DBusException(Exception):

1670

pass

1671

1672

@classmethod

1673

def SystemBus(cls):

1674

def get_object(busname, objectpath):

1675

raise cls.exceptions.DBusException()

1676

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

1677

return Bus(get_object=get_object)

1678

1679

1680

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1681

class mock_dbus_python:

1682

"""mock dbus-python modules"""

1683

class SystemBus:

1684

@staticmethod

1685

def get_object(busname, objectpath):

1686

return Unique()

1687

1688

def setUp(self):

1689

self.bus = dbus_python_adapter.CachingBus(

1690

self.mock_dbus_python)

1691

1692

def test_returns_distinct_objectpaths(self):

1693

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

1694

self.assertIsInstance(obj1, Unique)

1695

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

1696

self.assertIsInstance(obj2, Unique)

1697

self.assertIsNot(obj1, obj2)

1698

1699

def test_returns_distinct_busnames(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_distinct_both(self):

1707

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

1708

self.assertIsInstance(obj1, Unique)

1709

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

1710

self.assertIsInstance(obj2, Unique)

1711

self.assertIsNot(obj1, obj2)

1712

1713

def test_returns_same(self):

1714

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

1715

self.assertIsInstance(obj1, Unique)

1716

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

1717

self.assertIsInstance(obj2, Unique)

1718

self.assertIs(obj1, obj2)

1719

1720

def test_returns_same_old(self):

1721

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

1722

self.assertIsInstance(obj1, Unique)

1723

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

1724

self.assertIsInstance(obj2, Unique)

1725

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

1726

self.assertIsInstance(obj1b, Unique)

1727

self.assertIsNot(obj1, obj2)

1728

self.assertIsNot(obj2, obj1b)

1729

self.assertIs(obj1, obj1b)

1730

1731

1732

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1733

1734

def Stub_pydbus_func(self, func):

1735

class stub_pydbus:

1736

"""stub pydbus module"""

1737

class SystemBus:

1738

@staticmethod

1739

def get(busname, objectpath):

1740

DBusObject = collections.namedtuple(

1741

"DBusObject", ("methodname",))

1742

return {"interface":

1743

DBusObject(methodname=func)}

1744

return stub_pydbus

1745

1746

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

1747

interface, *args):

1748

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

1749

return bus.call_method(methodname, busname, objectpath,

1750

interface, *args)

1751

1752

def test_call_method_returns(self):

1753

expected_method_return = Unique()

1754

method_args = (Unique(), Unique())

1755

def func(*args):

1756

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

1757

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

1758

self.assertIs(marg, arg)

1759

return expected_method_return

1760

stub_pydbus = self.Stub_pydbus_func(func)

1761

bus = pydbus_adapter.SystemBus(stub_pydbus)

1762

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

1763

"objectpath", "interface",

1764

*method_args)

1765

self.assertIs(ret, expected_method_return)

1766

1767

def test_call_method_handles_exception(self):

1768

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

1769

1770

def func():

1771

raise gi.repository.GLib.Error()

1772

1773

stub_pydbus = self.Stub_pydbus_func(func)

1774

bus = pydbus_adapter.SystemBus(stub_pydbus)

1775

1776

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

1777

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

1778

"objectpath", "interface")

1779

1780

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

1781

1782

def test_get_converts_to_correct_exception(self):

1783

bus = pydbus_adapter.SystemBus(

1784

self.fake_pydbus_raises_exception_on_connect)

1785

with self.assertRaises(dbus.ConnectFailed):

1786

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

1787

"objectpath", "interface")

1788

1789

class fake_pydbus_raises_exception_on_connect:

1790

"""fake dbus-python module"""

1791

@classmethod

1792

def SystemBus(cls):

1793

def get(busname, objectpath):

1794

raise gi.repository.GLib.Error()

1795

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

1796

return Bus(get=get)

1797

1798

def test_set_property_uses_setattr(self):

1799

class Object:

1800

pass

1801

obj = Object()

1802

class pydbus_spy:

1803

class SystemBus:

1804

@staticmethod

1805

def get(busname, objectpath):

1806

return {"interface": obj}

1807

bus = pydbus_adapter.SystemBus(pydbus_spy)

1808

value = Unique()

1809

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

1810

value)

1811

self.assertIs(value, obj.key)

1812

1813

def test_get_suppresses_xml_deprecation_warning(self):

1814

if sys.version_info.major >= 3:

1815

return

1816

class stub_pydbus_get:

1817

class SystemBus:

1818

@staticmethod

1819

def get(busname, objectpath):

1820

warnings.warn_explicit(

1821

"deprecated", DeprecationWarning,

1822

"xml.etree.ElementTree", 0)

1823

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1824

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

1825

warnings.simplefilter("always")

1826

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

1827

self.assertEqual(0, len(w))

1828

1829

1830

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1831

class stub_pydbus:

1832

"""stub pydbus module"""

1833

class SystemBus:

1834

@staticmethod

1835

def get(busname, objectpath):

1836

return Unique()

1837

1838

def setUp(self):

1839

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

1840

1841

def test_returns_distinct_objectpaths(self):

1842

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

1843

self.assertIsInstance(obj1, Unique)

1844

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

1845

self.assertIsInstance(obj2, Unique)

1846

self.assertIsNot(obj1, obj2)

1847

1848

def test_returns_distinct_busnames(self):

1849

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

1850

self.assertIsInstance(obj1, Unique)

1851

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

1852

self.assertIsInstance(obj2, Unique)

1853

self.assertIsNot(obj1, obj2)

1854

1855

def test_returns_distinct_both(self):

1856

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

1857

self.assertIsInstance(obj1, Unique)

1858

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

1859

self.assertIsInstance(obj2, Unique)

1860

self.assertIsNot(obj1, obj2)

1861

1862

def test_returns_same(self):

1863

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

1864

self.assertIsInstance(obj1, Unique)

1865

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

1866

self.assertIsInstance(obj2, Unique)

1867

self.assertIs(obj1, obj2)

1868

1869

def test_returns_same_old(self):

1870

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

1871

self.assertIsInstance(obj1, Unique)

1872

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

1873

self.assertIsInstance(obj2, Unique)

1874

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

1875

self.assertIsInstance(obj1b, Unique)

1876

self.assertIsNot(obj1, obj2)

1877

self.assertIsNot(obj2, obj1b)

1878

self.assertIs(obj1, obj1b)

1879

1880

1881

class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):

1882

1883

class dummy_dbussy:

1884

class DBUS:

1885

class ObjectPath(str):

1886

pass

1887

class DBusError(Exception):

1888

pass

1889

1890

def fake_ravel_func(self, func):

1891

class fake_ravel:

1892

@staticmethod

1893

def system_bus():

1894

class DBusInterfaceProxy:

1895

@staticmethod

1896

def methodname(*args):

1897

return [func(*args)]

1898

class DBusObject:

1899

@staticmethod

1900

def get_interface(interface):

1901

if interface == "interface":

1902

return DBusInterfaceProxy()

1903

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

1904

return fake_ravel

1905

1906

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

1907

interface, *args):

1908

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

1909

return bus.call_method(methodname, busname, objectpath,

1910

interface, *args)

1911

1912

def test_call_method_returns(self):

1913

expected_method_return = Unique()

1914

method_args = (Unique(), Unique())

1915

def func(*args):

1916

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

1917

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

1918

self.assertIs(marg, arg)

1919

return expected_method_return

1920

fake_ravel = self.fake_ravel_func(func)

1921

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

1922

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

1923

"objectpath", "interface",

1924

*method_args)

1925

self.assertIs(ret, expected_method_return)

1926

1927

def test_call_method_filters_objectpath(self):

1928

def func():

1929

return method_return

1930

fake_ravel = self.fake_ravel_func(func)

1931

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

1932

method_return = (self.dummy_dbussy.DBUS

1933

.ObjectPath("objectpath"))

1934

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

1935

"objectpath", "interface")

1936

self.assertEqual("objectpath", ret)

1937

self.assertNotIsInstance(ret,

1938

self.dummy_dbussy.DBUS.ObjectPath)

1939

1940

def test_call_method_filters_objectpaths_in_dict(self):

1941

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1942

def func():

1943

return method_return

1944

fake_ravel = self.fake_ravel_func(func)

1945

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

1946

method_return = {

1947

ObjectPath("objectpath_key_1"):

1948

ObjectPath("objectpath_value_1"),

1949

ObjectPath("objectpath_key_2"):

1950

ObjectPath("objectpath_value_2"),

1951

}

1952

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

1953

"objectpath", "interface")

1954

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

1955

for key, value in

1956

method_return.items()}

1957

for key, value in ret.items():

1958

self.assertNotIsInstance(key, ObjectPath)

1959

self.assertNotIsInstance(value, ObjectPath)

1960

self.assertEqual(expected_method_return, ret)

1961

self.assertIsInstance(ret, dict)

1962

1963

def test_call_method_filters_objectpaths_in_dict_in_dict(self):

1964

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

1965

def func():

1966

return method_return

1967

fake_ravel = self.fake_ravel_func(func)

1968

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

1969

method_return = {

1970

ObjectPath("key1"): {

1971

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

1972

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

1973

},

1974

ObjectPath("key2"): {

1975

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

1976

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

1977

},

1978

}

1979

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

1980

"objectpath", "interface")

1981

expected_method_return = {

1982

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

1983

"key12": "value12"},

1984

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

1985

"key22": "value22"},

1986

}

1987

self.assertEqual(expected_method_return, ret)

1988

for key, value in ret.items():

1989

self.assertIsInstance(value, dict)

1990

self.assertEqual(expected_method_return[key], value)

1991

self.assertNotIsInstance(key, ObjectPath)

1992

for inner_key, inner_value in value.items():

1993

self.assertIsInstance(value, dict)

1994

self.assertEqual(

1995

expected_method_return[key][inner_key],

1996

inner_value)

1997

self.assertNotIsInstance(key, ObjectPath)

1998

1999

def test_call_method_filters_objectpaths_in_dict_three_deep(self):

2000

ObjectPath = self.dummy_dbussy.DBUS.ObjectPath

2001

def func():

2002

return method_return

2003

fake_ravel = self.fake_ravel_func(func)

2004

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

2005

method_return = {

2006

ObjectPath("key1"): {

2007

ObjectPath("key2"): {

2008

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

2009

},

2010

},

2011

}

2012

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

2013

"objectpath", "interface")

2014

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

2015

self.assertEqual(expected_method_return, ret)

2016

self.assertIsInstance(ret, dict)

2017

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

2018

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

2019

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

2020

ObjectPath)

2021

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

2022

self.assertNotIsInstance(

2023

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

2024

ObjectPath)

2025

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

2026

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

2027

self.dummy_dbussy.DBUS.ObjectPath)

2028

2029

def test_call_method_handles_exception(self):

2030

def func():

2031

raise self.dummy_dbussy.DBusError()

2032

2033

fake_ravel = self.fake_ravel_func(func)

2034

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

2035

2036

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

2037

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

2038

"objectpath", "interface")

2039

2040

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

2041

2042

def test_get_object_converts_to_correct_exception(self):

2043

class fake_ravel_raises_exception_on_connect:

2044

@staticmethod

2045

def system_bus():

2046

class Bus:

2047

@staticmethod

2048

def __getitem__(key):

2049

if key == "objectpath":

2050

raise self.dummy_dbussy.DBusError()

2051

raise Exception(key)

2052

return {"busname": Bus()}

2053

def func():

2054

raise self.dummy_dbussy.DBusError()

2055

bus = dbussy_adapter.SystemBus(

2056

self.dummy_dbussy,

2057

fake_ravel_raises_exception_on_connect)

2058

with self.assertRaises(dbus.ConnectFailed):

2059

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

2060

"objectpath", "interface")

2061

2062

2063

class Test_commands_from_options(unittest.TestCase):

2064

2065

def setUp(self):

2066

self.parser = argparse.ArgumentParser()

2067

add_command_line_options(self.parser)

2068

2069

def test_is_enabled(self):

2070

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

2071

command.IsEnabled)

2072

2073

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

2074

clients=None, **cmd_attrs):

2075

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

2076

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

2077

options = self.parser.parse_args(args)

2078

check_option_syntax(self.parser, options)

2079

commands = commands_from_options(options)

2080

self.assertEqual(length, len(commands))

2081

for command in commands:

2082

if isinstance(command, command_cls):

2083

break

2084

else:

2085

self.assertIsInstance(command, command_cls)

2086

if clients is not None:

2087

self.assertEqual(clients, options.client)

2088

for key, value in cmd_attrs.items():

2089

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

2090

2091

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

2092

for cmd in commands:

2093

self.assert_command_from_args(args, cmd,

2094

length=len(commands),

2095

clients=clients)

2096

2097

def test_is_enabled_short(self):

2098

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

2099

command.IsEnabled)

2100

2101

def test_approve(self):

2102

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

2103

command.Approve)

2104

2105

def test_approve_short(self):

2106

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

2107

command.Approve)

2108

2109

def test_deny(self):

2110

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

2111

command.Deny)

2112

2113

def test_deny_short(self):

2114

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

2115

2116

def test_remove(self):

2117

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

2118

command.Remove)

2119

2120

def test_deny_before_remove(self):

2121

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

2122

"client"])

2123

check_option_syntax(self.parser, options)

2124

commands = commands_from_options(options)

2125

self.assertEqual(2, len(commands))

2126

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

2127

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

2128

2129

def test_deny_before_remove_reversed(self):

2130

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

2131

"--all"])

2132

check_option_syntax(self.parser, options)

2133

commands = commands_from_options(options)

2134

self.assertEqual(2, len(commands))

2135

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

2136

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

2137

2138

def test_remove_short(self):

2139

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

2140

command.Remove)

2141

2142

def test_dump_json(self):

2143

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

2144

command.DumpJSON)

2145

2146

def test_enable(self):

2147

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

2148

command.Enable)

2149

2150

def test_enable_short(self):

2151

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

2152

command.Enable)

2153

2154

def test_disable(self):

2155

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

2156

command.Disable)

2157

2158

def test_disable_short(self):

2159

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

2160

command.Disable)

2161

2162

def test_bump_timeout(self):

2163

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

2164

command.BumpTimeout)

2165

2166

def test_bump_timeout_short(self):

2167

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

2168

command.BumpTimeout)

2169

2170

def test_start_checker(self):

2171

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

2172

command.StartChecker)

2173

2174

def test_stop_checker(self):

2175

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

2176

command.StopChecker)

2177

2178

def test_approve_by_default(self):

2179

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

2180

"client"],

2181

command.ApproveByDefault)

2182

2183

def test_deny_by_default(self):

2184

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

2185

command.DenyByDefault)

2186

2187

def test_checker(self):

2188

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

2189

command.SetChecker,

2190

value_to_set=":")

2191

2192

def test_checker_empty(self):

2193

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

2194

command.SetChecker,

2195

value_to_set="")

2196

2197

def test_checker_short(self):

2198

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

2199

command.SetChecker,

2200

value_to_set=":")

2201

2202

def test_host(self):

2203

self.assert_command_from_args(

2204

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

2205

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

2206

2207

def test_host_short(self):

2208

self.assert_command_from_args(

2209

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

2210

value_to_set="client.example.org")

2211

2212

def test_secret_devnull(self):

2213

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

2214

"client"], command.SetSecret,

2215

value_to_set=b"")

2216

2217

def test_secret_tempfile(self):

2218

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

2219

value = b"secret\0xyzzy\nbar"

2220

f.write(value)

2221

f.seek(0)

2222

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

2223

"client"],

2224

command.SetSecret,

2225

value_to_set=value)

2226

2227

def test_secret_devnull_short(self):

2228

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

2229

"client"], command.SetSecret,

2230

value_to_set=b"")

2231

2232

def test_secret_tempfile_short(self):

2233

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

2234

value = b"secret\0xyzzy\nbar"

2235

f.write(value)

2236

f.seek(0)

2237

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

2238

command.SetSecret,

2239

value_to_set=value)

2240

2241

def test_timeout(self):

2242

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

2243

command.SetTimeout,

2244

value_to_set=300000)

2245

2246

def test_timeout_short(self):

2247

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

2248

command.SetTimeout,

2249

value_to_set=300000)

2250

2251

def test_extended_timeout(self):

2252

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

2253

"client"],

2254

command.SetExtendedTimeout,

2255

value_to_set=900000)

2256

2257

def test_interval(self):

2258

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

2259

"client"], command.SetInterval,

2260

value_to_set=120000)

2261

2262

def test_interval_short(self):

2263

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

2264

command.SetInterval,

2265

value_to_set=120000)

2266

2267

def test_approval_delay(self):

2268

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

2269

"client"],

2270

command.SetApprovalDelay,

2271

value_to_set=30000)

2272

2273

def test_approval_duration(self):

2274

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

2275

"client"],

2276

command.SetApprovalDuration,

2277

value_to_set=1000)

2278

2279

def test_print_table(self):

2280

self.assert_command_from_args([], command.PrintTable,

2281

verbose=False)

2282

2283

def test_print_table_verbose(self):

2284

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

2285

command.PrintTable,

2286

verbose=True)

2287

2288

def test_print_table_verbose_short(self):

2289

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

2290

verbose=True)

2291

2292

2293

def test_manual_page_example_1(self):

2294

self.assert_command_from_args("",

2295

command.PrintTable,

2296

clients=[],

2297

verbose=False)

2298

2299

def test_manual_page_example_2(self):

2300

self.assert_command_from_args(

2301

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

2302

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

2303

"foo2.example.org"],

2304

verbose=True)

2305

2306

def test_manual_page_example_3(self):

2307

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

2308

command.Enable,

2309

clients=[])

2310

2311

def test_manual_page_example_4(self):

2312

self.assert_commands_from_args(

2313

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

2314

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

2315

[command.SetTimeout, command.SetInterval],

2316

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

2317

2318

def test_manual_page_example_5(self):

2319

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

2320

command.Approve,

2321

clients=[])

2322

2323

2324

class TestCommand(unittest.TestCase):

2325

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

2326

2327

class FakeMandosBus(dbus.MandosBus):

2328

def __init__(self, testcase):

2329

self.client_properties = {

2330

"Name": "foo",

2331

"KeyID": ("92ed150794387c03ce684574b1139a65"

2332

"94a34f895daaaf09fd8ea90a27cddb12"),

2333

"Secret": b"secret",

2334

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

2335

"Enabled": True,

2336

"Timeout": 300000,

2337

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

2338

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

2339

"Interval": 120000,

2340

"Fingerprint": ("778827225BA7DE539C5A"

2341

"7CFA59CFF7CDBD9A5920"),

2342

"CheckerRunning": False,

2343

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

2344

"ApprovalPending": False,

2345

"ApprovedByDefault": True,

2346

"LastApprovalRequest": "",

2347

"ApprovalDelay": 0,

2348

"ApprovalDuration": 1000,

2349

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

2350

"ExtendedTimeout": 900000,

2351

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

2352

"LastCheckerStatus": 0,

2353

}

2354

self.other_client_properties = {

2355

"Name": "barbar",

2356

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2357

"6ab612cff5ad227247e46c2b020f441c"),

2358

"Secret": b"secretbar",

2359

"Host": "192.0.2.3",

2360

"Enabled": True,

2361

"Timeout": 300000,

2362

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

2363

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

2364

"Interval": 120000,

2365

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2366

"F547B3A107558FCA3A27"),

2367

"CheckerRunning": True,

2368

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

2369

"ApprovalPending": False,

2370

"ApprovedByDefault": False,

2371

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

2372

"ApprovalDelay": 30000,

2373

"ApprovalDuration": 93785000,

2374

"Checker": ":",

2375

"ExtendedTimeout": 900000,

2376

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

2377

"LastCheckerStatus": -2,

2378

}

2379

self.clients = collections.OrderedDict(

2380

[

2381

("client_objectpath", self.client_properties),

2382

("other_client_objectpath",

2383

self.other_client_properties),

2384

])

2385

self.one_client = {"client_objectpath":

2386

self.client_properties}

2387

self.testcase = testcase

2388

self.calls = []

2389

2390

def call_method(self, methodname, busname, objectpath,

2391

interface, *args):

2392

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

2393

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

2394

interface, args))

2395

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

2396

if methodname == "Set":

2397

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

2398

interface, key, value = args

2399

self.testcase.assertEqual(

2400

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

2401

self.clients[objectpath][key] = value

2402

return

2403

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

2404

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

2405

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

2406

clientpath = args[0]

2407

del self.clients[clientpath]

2408

return

2409

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

2410

if methodname == "Approve":

2411

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

2412

return

2413

raise ValueError()

2414

2415

def setUp(self):

2416

self.bus = self.FakeMandosBus(self)

2417

2418

2419

class TestBaseCommands(TestCommand):

2420

2421

def test_IsEnabled_exits_successfully(self):

2422

with self.assertRaises(SystemExit) as e:

2423

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

2424

if e.exception.code is not None:

2425

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

2426

else:

2427

self.assertIsNone(e.exception.code)

2428

2429

def test_IsEnabled_exits_with_failure(self):

2430

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

2431

with self.assertRaises(SystemExit) as e:

2432

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

2433

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

2434

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

2435

else:

2436

self.assertIsNotNone(e.exception.code)

2437

2438

def test_Approve(self):

2439

busname = "se.recompile.Mandos"

2440

client_interface = "se.recompile.Mandos.Client"

2441

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

2442

self.assertTrue(self.bus.clients)

2443

for clientpath in self.bus.clients:

2444

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

2445

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

2446

2447

def test_Deny(self):

2448

busname = "se.recompile.Mandos"

2449

client_interface = "se.recompile.Mandos.Client"

2450

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

2451

self.assertTrue(self.bus.clients)

2452

for clientpath in self.bus.clients:

2453

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

2454

client_interface, (False,)),

2455

self.bus.calls)

2456

2457

def test_Remove(self):

2458

busname = "se.recompile.Mandos"

2459

server_path = "/"

2460

server_interface = "se.recompile.Mandos"

2461

orig_clients = self.bus.clients.copy()

2462

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

2463

self.assertFalse(self.bus.clients)

2464

for clientpath in orig_clients:

2465

self.assertIn(("RemoveClient", busname,

2466

server_path, server_interface,

2467

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

2468

2469

expected_json = {

2470

"foo": {

2471

"Name": "foo",

2472

"KeyID": ("92ed150794387c03ce684574b1139a65"

2473

"94a34f895daaaf09fd8ea90a27cddb12"),

2474

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

2475

"Enabled": True,

2476

"Timeout": 300000,

2477

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

2478

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

2479

"Interval": 120000,

2480

"Fingerprint": ("778827225BA7DE539C5A"

2481

"7CFA59CFF7CDBD9A5920"),

2482

"CheckerRunning": False,

2483

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

2484

"ApprovalPending": False,

2485

"ApprovedByDefault": True,

2486

"LastApprovalRequest": "",

2487

"ApprovalDelay": 0,

2488

"ApprovalDuration": 1000,

2489

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

2490

"ExtendedTimeout": 900000,

2491

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

2492

"LastCheckerStatus": 0,

2493

},

2494

"barbar": {

2495

"Name": "barbar",

2496

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2497

"6ab612cff5ad227247e46c2b020f441c"),

2498

"Host": "192.0.2.3",

2499

"Enabled": True,

2500

"Timeout": 300000,

2501

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

2502

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

2503

"Interval": 120000,

2504

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2505

"F547B3A107558FCA3A27"),

2506

"CheckerRunning": True,

2507

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

2508

"ApprovalPending": False,

2509

"ApprovedByDefault": False,

2510

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

2511

"ApprovalDelay": 30000,

2512

"ApprovalDuration": 93785000,

2513

"Checker": ":",

2514

"ExtendedTimeout": 900000,

2515

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

2516

"LastCheckerStatus": -2,

2517

},

2518

}

2519

2520

def test_DumpJSON_normal(self):

2521

with self.capture_stdout_to_buffer() as buffer:

2522

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

2523

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

2524

self.assertDictEqual(self.expected_json, json_data)

2525

2526

@staticmethod

2527

@contextlib.contextmanager

2528

def capture_stdout_to_buffer():

2529

capture_buffer = io.StringIO()

2530

old_stdout = sys.stdout

2531

sys.stdout = capture_buffer

2532

try:

2533

yield capture_buffer

2534

finally:

2535

sys.stdout = old_stdout

2536

2537

def test_DumpJSON_one_client(self):

2538

with self.capture_stdout_to_buffer() as buffer:

2539

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

2540

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

2541

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

2542

self.assertDictEqual(expected_json, json_data)

2543

2544

def test_PrintTable_normal(self):

2545

with self.capture_stdout_to_buffer() as buffer:

2546

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

2547

expected_output = "\n".join((

2548

"Name Enabled Timeout Last Successful Check",

2549

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

2550

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

2551

)) + "\n"

2552

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

2553

2554

def test_PrintTable_verbose(self):

2555

with self.capture_stdout_to_buffer() as buffer:

2556

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

2557

columns = (

2558

(

2559

"Name ",

2560

"foo ",

2561

"barbar ",

2562

),(

2563

"Enabled ",

2564

"Yes ",

2565

"Yes ",

2566

),(

2567

"Timeout ",

2568

"00:05:00 ",

2569

"00:05:00 ",

2570

),(

2571

"Last Successful Check ",

2572

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

2573

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

2574

),(

2575

"Created ",

2576

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

2577

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

2578

),(

2579

"Interval ",

2580

"00:02:00 ",

2581

"00:02:00 ",

2582

),(

2583

"Host ",

2584

"foo.example.org ",

2585

"192.0.2.3 ",

2586

),(

2587

("Key ID "

2588

" "),

2589

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2590

"ea90a27cddb12 "),

2591

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2592

"46c2b020f441c "),

2593

),(

2594

"Fingerprint ",

2595

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2596

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2597

),(

2598

"Check Is Running ",

2599

"No ",

2600

"Yes ",

2601

),(

2602

"Last Enabled ",

2603

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

2604

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

2605

),(

2606

"Approval Is Pending ",

2607

"No ",

2608

"No ",

2609

),(

2610

"Approved By Default ",

2611

"Yes ",

2612

"No ",

2613

),(

2614

"Last Approval Request ",

2615

" ",

2616

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

2617

),(

2618

"Approval Delay ",

2619

"00:00:00 ",

2620

"00:00:30 ",

2621

),(

2622

"Approval Duration ",

2623

"00:00:01 ",

2624

"1T02:03:05 ",

2625

),(

2626

"Checker ",

2627

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

2628

": ",

2629

),(

2630

"Extended Timeout ",

2631

"00:15:00 ",

2632

"00:15:00 ",

2633

),(

2634

"Expires ",

2635

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

2636

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

2637

),(

2638

"Last Checker Status",

2639

"0 ",

2640

"-2 ",

2641

)

2642

)

2643

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

2644

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

2645

for rows in columns)

2646

for line in range(num_lines))

2647

+ "\n")

2648

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

2649

2650

def test_PrintTable_one_client(self):

2651

with self.capture_stdout_to_buffer() as buffer:

2652

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

2653

expected_output = "\n".join((

2654

"Name Enabled Timeout Last Successful Check",

2655

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

2656

)) + "\n"

2657

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

2658

2659

2660

class TestPropertySetterCmd(TestCommand):

2661

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

2662

2663

def runTest(self):

2664

if not hasattr(self, "command"):

2665

return # Abstract TestCase class

2666

2667

if hasattr(self, "values_to_set"):

2668

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

2669

values_to_get = getattr(self, "values_to_get",

2670

self.values_to_set)

2671

else:

2672

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

2673

values_to_get = self.values_to_get

2674

self.assertTrue(values_to_get)

2675

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

2676

for clientpath in self.bus.clients:

2677

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

2678

Unique())

2679

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

2680

self.assertTrue(self.bus.clients)

2681

for clientpath in self.bus.clients:

2682

value = (self.bus.clients[clientpath]

2683

[self.propname])

2684

self.assertNotIsInstance(value, Unique)

2685

self.assertEqual(value_to_get, value)

2686

2687

2688

class TestEnableCmd(TestPropertySetterCmd):

2689

command = command.Enable

2690

propname = "Enabled"

2691

values_to_get = [True]

2692

2693

2694

class TestDisableCmd(TestPropertySetterCmd):

2695

command = command.Disable

2696

propname = "Enabled"

2697

values_to_get = [False]

2698

2699

2700

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2701

command = command.BumpTimeout

2702

propname = "LastCheckedOK"

2703

values_to_get = [""]

2704

2705

2706

class TestStartCheckerCmd(TestPropertySetterCmd):

2707

command = command.StartChecker

2708

propname = "CheckerRunning"

2709

values_to_get = [True]

2710

2711

2712

class TestStopCheckerCmd(TestPropertySetterCmd):

2713

command = command.StopChecker

2714

propname = "CheckerRunning"

2715

values_to_get = [False]

2716

2717

2718

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2719

command = command.ApproveByDefault

2720

propname = "ApprovedByDefault"

2721

values_to_get = [True]

2722

2723

2724

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2725

command = command.DenyByDefault

2726

propname = "ApprovedByDefault"

2727

values_to_get = [False]

2728

2729

2730

class TestSetCheckerCmd(TestPropertySetterCmd):

2731

command = command.SetChecker

2732

propname = "Checker"

2733

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

2734

2735

2736

class TestSetHostCmd(TestPropertySetterCmd):

2737

command = command.SetHost

2738

propname = "Host"

2739

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

2740

2741

2742

class TestSetSecretCmd(TestPropertySetterCmd):

2743

command = command.SetSecret

2744

propname = "Secret"

2745

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

2746

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

2747

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

2748

2749

2750

class TestSetTimeoutCmd(TestPropertySetterCmd):

2751

command = command.SetTimeout

2752

propname = "Timeout"

2753

values_to_set = [datetime.timedelta(),

2754

datetime.timedelta(minutes=5),

2755

datetime.timedelta(seconds=1),

2756

datetime.timedelta(weeks=1),

2757

datetime.timedelta(weeks=52)]

2758

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

2759

2760

2761

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2762

command = command.SetExtendedTimeout

2763

propname = "ExtendedTimeout"

2764

values_to_set = [datetime.timedelta(),

2765

datetime.timedelta(minutes=5),

2766

datetime.timedelta(seconds=1),

2767

datetime.timedelta(weeks=1),

2768

datetime.timedelta(weeks=52)]

2769

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

2770

2771

2772

class TestSetIntervalCmd(TestPropertySetterCmd):

2773

command = command.SetInterval

2774

propname = "Interval"

2775

values_to_set = [datetime.timedelta(),

2776

datetime.timedelta(minutes=5),

2777

datetime.timedelta(seconds=1),

2778

datetime.timedelta(weeks=1),

2779

datetime.timedelta(weeks=52)]

2780

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

2781

2782

2783

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2784

command = command.SetApprovalDelay

2785

propname = "ApprovalDelay"

2786

values_to_set = [datetime.timedelta(),

2787

datetime.timedelta(minutes=5),

2788

datetime.timedelta(seconds=1),

2789

datetime.timedelta(weeks=1),

2790

datetime.timedelta(weeks=52)]

2791

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

2792

2793

2794

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2795

command = command.SetApprovalDuration

2796

propname = "ApprovalDuration"

2797

values_to_set = [datetime.timedelta(),

2798

datetime.timedelta(minutes=5),

2799

datetime.timedelta(seconds=1),

2800

datetime.timedelta(weeks=1),

2801

datetime.timedelta(weeks=52)]

2802

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

2803

2804

2805

2806

def should_only_run_tests():

2807

parser = argparse.ArgumentParser(add_help=False)

2808

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

2809

args, unknown_args = parser.parse_known_args()

2810

run_tests = args.check

2811

if run_tests:

2812

# Remove --check argument from sys.argv

2813

sys.argv[1:] = unknown_args

2814

return run_tests

2815

2816

# Add all tests from doctest strings

2817

def load_tests(loader, tests, none):

2818

import doctest

2819

tests.addTests(doctest.DocTestSuite())

2820

return tests

2821

2822

if __name__ == "__main__":

2823

try:

2824

if should_only_run_tests():

2825

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

2826

unittest.main()

2827

else:

2828

main()

2829

finally:

2830

logging.shutdown()