/mandos/release : revision 24.1.79

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

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

import abc

49

50

import dbus as dbus_python

51

52

# Show warnings by default

53

if not sys.warnoptions:

54

import warnings

55

warnings.simplefilter("default")

56

57

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

58

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

59

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

60

61

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

62

63

if sys.version_info.major == 2:

64

str = unicode

65

import StringIO

66

io.StringIO = StringIO.StringIO

67

68

locale.setlocale(locale.LC_ALL, "")

69

70

version = "1.8.3"

71

72

73

def main():

74

parser = argparse.ArgumentParser()

75

add_command_line_options(parser)

76

77

options = parser.parse_args()

78

check_option_syntax(parser, options)

79

80

clientnames = options.client

81

82

if options.debug:

83

log.setLevel(logging.DEBUG)

84

85

bus = dbus_python_adapter.CachingBus(dbus_python)

86

87

try:

88

all_clients = bus.get_clients_and_properties()

89

except dbus.ConnectFailed as e:

90

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

91

sys.exit(1)

92

except dbus.Error as e:

93

log.critical(

94

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

95

sys.exit(1)

96

97

# Compile dict of (clientpath: properties) to process

98

if not clientnames:

99

clients = all_clients

100

else:

101

clients = {}

102

for name in clientnames:

103

for objpath, properties in all_clients.items():

104

if properties["Name"] == name:

105

clients[objpath] = properties

106

break

107

else:

108

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

109

sys.exit(1)

110

111

commands = commands_from_options(options)

112

113

for command in commands:

114

command.run(clients, bus)

115

116

117

def add_command_line_options(parser):

118

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

119

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

120

help="show version number and exit")

121

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

122

help="Select all clients")

123

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

124

help="Print all fields")

125

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

126

action="append_const", default=[],

127

const=command.DumpJSON(),

128

help="Dump client data in JSON format")

129

enable_disable = parser.add_mutually_exclusive_group()

130

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

131

action="append_const", default=[],

132

const=command.Enable(),

133

help="Enable client")

134

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

135

action="append_const", default=[],

136

const=command.Disable(),

137

help="disable client")

138

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

139

action="append_const", default=[],

140

const=command.BumpTimeout(),

141

help="Bump timeout for client")

142

start_stop_checker = parser.add_mutually_exclusive_group()

143

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

144

dest="commands",

145

action="append_const", default=[],

146

const=command.StartChecker(),

147

help="Start checker for client")

148

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

149

action="append_const", default=[],

150

const=command.StopChecker(),

151

help="Stop checker for client")

152

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

153

action="append_const", default=[],

154

const=command.IsEnabled(),

155

help="Check if client is enabled")

156

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

157

action="append_const", default=[],

158

const=command.Remove(),

159

help="Remove client")

160

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

161

action="append", default=[],

162

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

163

help="Set checker command for client")

164

parser.add_argument(

165

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

166

default=[], metavar="TIME",

167

type=command.SetTimeout.argparse(string_to_delta),

168

help="Set timeout for client")

169

parser.add_argument(

170

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

171

default=[], metavar="TIME",

172

type=command.SetExtendedTimeout.argparse(string_to_delta),

173

help="Set extended timeout for client")

174

parser.add_argument(

175

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

176

default=[], metavar="TIME",

177

type=command.SetInterval.argparse(string_to_delta),

178

help="Set checker interval for client")

179

approve_deny_default = parser.add_mutually_exclusive_group()

180

approve_deny_default.add_argument(

181

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

182

action="append_const", default=[],

183

const=command.ApproveByDefault(),

184

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

185

approve_deny_default.add_argument(

186

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

187

action="append_const", default=[],

188

const=command.DenyByDefault(),

189

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

190

parser.add_argument(

191

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

192

default=[], metavar="TIME",

193

type=command.SetApprovalDelay.argparse(string_to_delta),

194

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

195

parser.add_argument(

196

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

197

default=[], metavar="TIME",

198

type=command.SetApprovalDuration.argparse(string_to_delta),

199

help="Set duration of one client approval")

200

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

201

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

202

type=command.SetHost,

203

help="Set host for client")

204

parser.add_argument(

205

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

206

default=[], metavar="FILENAME",

207

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

208

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

209

approve_deny = parser.add_mutually_exclusive_group()

210

approve_deny.add_argument(

211

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

212

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

213

help="Approve any current client request")

214

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

215

action="append_const", default=[],

216

const=command.Deny(),

217

help="Deny any current client request")

218

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

219

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

220

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

221

help="Run self-test")

222

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

223

224

225

def string_to_delta(interval):

226

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

227

228

try:

229

return rfc3339_duration_to_delta(interval)

230

except ValueError as e:

231

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

232

' '.join(e.args))

233

return parse_pre_1_6_1_interval(interval)

234

235

236

def rfc3339_duration_to_delta(duration):

237

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

238

239

>>> rfc3339_duration_to_delta("P7D")

240

datetime.timedelta(7)

241

>>> rfc3339_duration_to_delta("PT60S")

242

datetime.timedelta(0, 60)

243

>>> rfc3339_duration_to_delta("PT60M")

244

datetime.timedelta(0, 3600)

245

>>> rfc3339_duration_to_delta("P60M")

246

datetime.timedelta(1680)

247

>>> rfc3339_duration_to_delta("PT24H")

248

datetime.timedelta(1)

249

>>> rfc3339_duration_to_delta("P1W")

250

datetime.timedelta(7)

251

>>> rfc3339_duration_to_delta("PT5M30S")

252

datetime.timedelta(0, 330)

253

>>> rfc3339_duration_to_delta("P1DT3M20S")

254

datetime.timedelta(1, 200)

255

>>> # Can not be empty:

256

>>> rfc3339_duration_to_delta("")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: ""

260

>>> # Must start with "P":

261

>>> rfc3339_duration_to_delta("1D")

262

Traceback (most recent call last):

263

...

264

ValueError: Invalid RFC 3339 duration: "1D"

265

>>> # Must use correct order

266

>>> rfc3339_duration_to_delta("PT1S2M")

267

Traceback (most recent call last):

268

...

269

ValueError: Invalid RFC 3339 duration: "PT1S2M"

270

>>> # Time needs time marker

271

>>> rfc3339_duration_to_delta("P1H2S")

272

Traceback (most recent call last):

273

...

274

ValueError: Invalid RFC 3339 duration: "P1H2S"

275

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

276

>>> rfc3339_duration_to_delta("P1D2W")

277

Traceback (most recent call last):

278

...

279

ValueError: Invalid RFC 3339 duration: "P1D2W"

280

>>> rfc3339_duration_to_delta("P2W2H")

281

Traceback (most recent call last):

282

...

283

ValueError: Invalid RFC 3339 duration: "P2W2H"

284

"""

285

286

# Parsing an RFC 3339 duration with regular expressions is not

287

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

288

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

289

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

290

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

291

# avoid excessive use of external libraries.

292

293

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

294

Token = collections.namedtuple("Token", (

295

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

296

# a "group" containing digits

297

"value", # datetime.timedelta or None

298

"followers")) # Tokens valid after this token

299

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

300

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

301

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

302

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

303

datetime.timedelta(seconds=1),

304

frozenset((token_end, )))

305

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

306

datetime.timedelta(minutes=1),

307

frozenset((token_second, token_end)))

308

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

309

datetime.timedelta(hours=1),

310

frozenset((token_minute, token_end)))

311

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

312

None,

313

frozenset((token_hour, token_minute,

314

token_second)))

315

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

316

datetime.timedelta(days=1),

317

frozenset((token_time, token_end)))

318

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

319

datetime.timedelta(weeks=4),

320

frozenset((token_day, token_end)))

321

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

322

datetime.timedelta(weeks=52),

323

frozenset((token_month, token_end)))

324

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

325

datetime.timedelta(weeks=1),

326

frozenset((token_end, )))

327

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

328

frozenset((token_year, token_month,

329

token_day, token_time,

330

token_week)))

331

# Define starting values:

332

# Value so far

333

value = datetime.timedelta()

334

found_token = None

335

# Following valid tokens

336

followers = frozenset((token_duration, ))

337

# String left to parse

338

s = duration

339

# Loop until end token is found

340

while found_token is not token_end:

341

# Search for any currently valid tokens

342

for token in followers:

343

match = token.regexp.match(s)

344

if match is not None:

345

# Token found

346

if token.value is not None:

347

# Value found, parse digits

348

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

349

# Add to value so far

350

value += factor * token.value

351

# Strip token from string

352

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

353

# Go to found token

354

found_token = token

355

# Set valid next tokens

356

followers = found_token.followers

357

break

358

else:

359

# No currently valid tokens were found

360

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

361

.format(duration))

362

# End token found

363

return value

364

365

366

def parse_pre_1_6_1_interval(interval):

367

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

368

and return a datetime.timedelta

369

370

>>> parse_pre_1_6_1_interval('7d')

371

datetime.timedelta(7)

372

>>> parse_pre_1_6_1_interval('60s')

373

datetime.timedelta(0, 60)

374

>>> parse_pre_1_6_1_interval('60m')

375

datetime.timedelta(0, 3600)

376

>>> parse_pre_1_6_1_interval('24h')

377

datetime.timedelta(1)

378

>>> parse_pre_1_6_1_interval('1w')

379

datetime.timedelta(7)

380

>>> parse_pre_1_6_1_interval('5m 30s')

381

datetime.timedelta(0, 330)

382

>>> parse_pre_1_6_1_interval('')

383

datetime.timedelta(0)

384

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

385

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

386

datetime.timedelta(2, 480, 18000)

387

388

"""

389

390

value = datetime.timedelta(0)

391

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

392

393

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

394

if suffix == "d":

395

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

396

elif suffix == "s":

397

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

398

elif suffix == "m":

399

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

400

elif suffix == "h":

401

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

402

elif suffix == "w":

403

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

404

elif suffix == "":

405

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

406

return value

407

408

409

def check_option_syntax(parser, options):

410

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

411

argparse"""

412

413

def has_commands(options, commands=None):

414

if commands is None:

415

commands = (command.Enable,

416

command.Disable,

417

command.BumpTimeout,

418

command.StartChecker,

419

command.StopChecker,

420

command.IsEnabled,

421

command.Remove,

422

command.SetChecker,

423

command.SetTimeout,

424

command.SetExtendedTimeout,

425

command.SetInterval,

426

command.ApproveByDefault,

427

command.DenyByDefault,

428

command.SetApprovalDelay,

429

command.SetApprovalDuration,

430

command.SetHost,

431

command.SetSecret,

432

command.Approve,

433

command.Deny)

434

return any(isinstance(cmd, commands)

435

for cmd in options.commands)

436

437

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

438

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

439

if options.verbose and has_commands(options):

440

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

441

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

442

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

443

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

444

if options.all and not has_commands(options):

445

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

446

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

447

and len(options.client) > 1):

448

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

449

if (len(options.commands) > 1

450

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

451

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

452

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

453

454

455

class dbus(object):

456

457

class SystemBus(object):

458

459

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

460

def get_managed_objects(self, busname, objectpath):

461

return self.call_method("GetManagedObjects", busname,

462

objectpath,

463

self.object_manager_iface)

464

465

properties_iface = "org.freedesktop.DBus.Properties"

466

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

467

value):

468

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

469

self.properties_iface, interface, key,

470

value)

471

472

473

class MandosBus(SystemBus):

474

busname_domain = "se.recompile"

475

busname = busname_domain + ".Mandos"

476

server_path = "/"

477

server_interface = busname_domain + ".Mandos"

478

client_interface = busname_domain + ".Mandos.Client"

479

del busname_domain

480

481

def get_clients_and_properties(self):

482

managed_objects = self.get_managed_objects(

483

self.busname, self.server_path)

484

return {objpath: properties[self.client_interface]

485

for objpath, properties in managed_objects.items()

486

if self.client_interface in properties}

487

488

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

489

return self.set_property(self.busname, objectpath,

490

self.client_interface, key,

491

value)

492

493

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

494

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

495

self.client_interface, *args)

496

497

def call_server_method(self, method, *args):

498

return self.call_method(method, self.busname,

499

self.server_path,

500

self.server_interface, *args)

501

502

class Error(Exception):

503

pass

504

505

class ConnectFailed(Error):

506

pass

507

508

509

class dbus_python_adapter(object):

510

511

class SystemBus(dbus.MandosBus):

512

"""Use dbus-python"""

513

514

def __init__(self, module=dbus_python):

515

self.dbus_python = module

516

self.bus = self.dbus_python.SystemBus()

517

518

@contextlib.contextmanager

519

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

520

try:

521

yield

522

except self.dbus_python.exceptions.DBusException as e:

523

# This does what "raise from" would do

524

exc = exception_class(*e.args)

525

exc.__cause__ = e

526

raise exc

527

528

def call_method(self, methodname, busname, objectpath,

529

interface, *args):

530

proxy_object = self.get_object(busname, objectpath)

531

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

532

interface, methodname,

533

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

534

method = getattr(proxy_object, methodname)

535

with self.convert_exception():

536

with dbus_python_adapter.SilenceLogger(

537

"dbus.proxies"):

538

value = method(*args, dbus_interface=interface)

539

return self.type_filter(value)

540

541

def get_object(self, busname, objectpath):

542

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

543

busname, objectpath)

544

with self.convert_exception(dbus.ConnectFailed):

545

return self.bus.get_object(busname, objectpath)

546

547

def type_filter(self, value):

548

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

549

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

550

return bool(value)

551

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

552

return str(value)

553

# Also recurse into dictionaries

554

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

555

return {self.type_filter(key):

556

self.type_filter(subval)

557

for key, subval in value.items()}

558

return value

559

560

561

class SilenceLogger(object):

562

"Simple context manager to silence a particular logger"

563

def __init__(self, loggername):

564

self.logger = logging.getLogger(loggername)

565

566

def __enter__(self):

567

self.logger.addFilter(self.nullfilter)

568

569

class NullFilter(logging.Filter):

570

def filter(self, record):

571

return False

572

573

nullfilter = NullFilter()

574

575

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

576

self.logger.removeFilter(self.nullfilter)

577

578

579

class CachingBus(SystemBus):

580

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

581

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

582

self.object_cache = {}

583

super(dbus_python_adapter.CachingBus,

584

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

585

def get_object(self, busname, objectpath):

586

try:

587

return self.object_cache[(busname, objectpath)]

588

except KeyError:

589

new_object = super(

590

dbus_python_adapter.CachingBus,

591

self).get_object(busname, objectpath)

592

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

593

return new_object

594

595

596

def commands_from_options(options):

597

598

commands = list(options.commands)

599

600

def find_cmd(cmd, commands):

601

i = 0

602

for i, c in enumerate(commands):

603

if isinstance(c, cmd):

604

return i

605

return i+1

606

607

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

608

# to occur ahead of command.Remove.

609

index_of_remove = find_cmd(command.Remove, commands)

610

before_remove = commands[:index_of_remove]

611

after_remove = commands[index_of_remove:]

612

cleaned_after = []

613

for cmd in after_remove:

614

if isinstance(cmd, command.Deny):

615

before_remove.append(cmd)

616

else:

617

cleaned_after.append(cmd)

618

if cleaned_after != after_remove:

619

commands = before_remove + cleaned_after

620

621

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

622

# optionally verbosely

623

if not commands:

624

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

625

626

return commands

627

628

629

class command(object):

630

"""A namespace for command classes"""

631

632

class Base(object):

633

"""Abstract base class for commands"""

634

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

635

"""Normal commands should implement run_on_one_client(),

636

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

637

override this run() method instead.

638

"""

639

self.bus = bus

640

for client, properties in clients.items():

641

self.run_on_one_client(client, properties)

642

643

644

class IsEnabled(Base):

645

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

646

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

647

if properties["Enabled"]:

648

sys.exit(0)

649

sys.exit(1)

650

651

652

class Approve(Base):

653

def run_on_one_client(self, client, properties):

654

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

655

656

657

class Deny(Base):

658

def run_on_one_client(self, client, properties):

659

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

660

661

662

class Remove(Base):

663

def run(self, clients, bus):

664

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

665

bus.call_server_method("RemoveClient", clientpath)

666

667

668

class Output(Base):

669

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

670

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

671

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

672

"Fingerprint", "CheckerRunning",

673

"LastEnabled", "ApprovalPending",

674

"ApprovedByDefault", "LastApprovalRequest",

675

"ApprovalDelay", "ApprovalDuration",

676

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

677

"LastCheckerStatus")

678

679

680

class DumpJSON(Output):

681

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

682

data = {properties["Name"]:

683

{key: properties[key]

684

for key in self.all_keywords}

685

for properties in clients.values()}

686

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

687

688

689

class PrintTable(Output):

690

def __init__(self, verbose=False):

691

self.verbose = verbose

692

693

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

694

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

695

"LastCheckedOK")

696

keywords = default_keywords

697

if self.verbose:

698

keywords = self.all_keywords

699

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

700

701

class TableOfClients(object):

702

tableheaders = {

703

"Name": "Name",

704

"Enabled": "Enabled",

705

"Timeout": "Timeout",

706

"LastCheckedOK": "Last Successful Check",

707

"LastApprovalRequest": "Last Approval Request",

708

"Created": "Created",

709

"Interval": "Interval",

710

"Host": "Host",

711

"Fingerprint": "Fingerprint",

712

"KeyID": "Key ID",

713

"CheckerRunning": "Check Is Running",

714

"LastEnabled": "Last Enabled",

715

"ApprovalPending": "Approval Is Pending",

716

"ApprovedByDefault": "Approved By Default",

717

"ApprovalDelay": "Approval Delay",

718

"ApprovalDuration": "Approval Duration",

719

"Checker": "Checker",

720

"ExtendedTimeout": "Extended Timeout",

721

"Expires": "Expires",

722

"LastCheckerStatus": "Last Checker Status",

723

}

724

725

def __init__(self, clients, keywords):

726

self.clients = clients

727

self.keywords = keywords

728

729

def __str__(self):

730

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

731

732

if sys.version_info.major == 2:

733

__unicode__ = __str__

734

def __str__(self):

735

return str(self).encode(

736

locale.getpreferredencoding())

737

738

def rows(self):

739

format_string = self.row_formatting_string()

740

rows = [self.header_line(format_string)]

741

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

742

for client in self.clients)

743

return rows

744

745

def row_formatting_string(self):

746

"Format string used to format table rows"

747

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

748

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

749

*(len(self.string_from_client(client,

750

key))

751

for client in self.clients)),

752

key=key)

753

for key in self.keywords)

754

755

def string_from_client(self, client, key):

756

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

757

758

@classmethod

759

def valuetostring(cls, value, keyword):

760

if isinstance(value, bool):

761

return "Yes" if value else "No"

762

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

763

"ApprovalDuration", "ExtendedTimeout"):

764

return cls.milliseconds_to_string(value)

765

return str(value)

766

767

def header_line(self, format_string):

768

return format_string.format(**self.tableheaders)

769

770

def client_line(self, client, format_string):

771

return format_string.format(

772

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

773

for key in self.keywords})

774

775

@staticmethod

776

def milliseconds_to_string(ms):

777

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

778

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

779

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

780

if td.days else "",

781

hours=td.seconds // 3600,

782

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

783

seconds=td.seconds % 60))

784

785

786

class PropertySetter(Base):

787

"Abstract class for Actions for setting one client property"

788

789

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

790

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

791

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

792

self.value_to_set)

793

794

@property

795

def propname(self):

796

raise NotImplementedError()

797

798

799

class Enable(PropertySetter):

800

propname = "Enabled"

801

value_to_set = True

802

803

804

class Disable(PropertySetter):

805

propname = "Enabled"

806

value_to_set = False

807

808

809

class BumpTimeout(PropertySetter):

810

propname = "LastCheckedOK"

811

value_to_set = ""

812

813

814

class StartChecker(PropertySetter):

815

propname = "CheckerRunning"

816

value_to_set = True

817

818

819

class StopChecker(PropertySetter):

820

propname = "CheckerRunning"

821

value_to_set = False

822

823

824

class ApproveByDefault(PropertySetter):

825

propname = "ApprovedByDefault"

826

value_to_set = True

827

828

829

class DenyByDefault(PropertySetter):

830

propname = "ApprovedByDefault"

831

value_to_set = False

832

833

834

class PropertySetterValue(PropertySetter):

835

"""Abstract class for PropertySetter recieving a value as

836

constructor argument instead of a class attribute."""

837

def __init__(self, value):

838

self.value_to_set = value

839

840

@classmethod

841

def argparse(cls, argtype):

842

def cmdtype(arg):

843

return cls(argtype(arg))

844

return cmdtype

845

846

class SetChecker(PropertySetterValue):

847

propname = "Checker"

848

849

850

class SetHost(PropertySetterValue):

851

propname = "Host"

852

853

854

class SetSecret(PropertySetterValue):

855

propname = "Secret"

856

857

@property

858

def value_to_set(self):

859

return self._vts

860

861

@value_to_set.setter

862

def value_to_set(self, value):

863

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

864

self._vts = value.read()

865

value.close()

866

867

868

class PropertySetterValueMilliseconds(PropertySetterValue):

869

"""Abstract class for PropertySetterValue taking a value

870

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

871

milliseconds."""

872

873

@property

874

def value_to_set(self):

875

return self._vts

876

877

@value_to_set.setter

878

def value_to_set(self, value):

879

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

880

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

881

882

883

class SetTimeout(PropertySetterValueMilliseconds):

884

propname = "Timeout"

885

886

887

class SetExtendedTimeout(PropertySetterValueMilliseconds):

888

propname = "ExtendedTimeout"

889

890

891

class SetInterval(PropertySetterValueMilliseconds):

892

propname = "Interval"

893

894

895

class SetApprovalDelay(PropertySetterValueMilliseconds):

896

propname = "ApprovalDelay"

897

898

899

class SetApprovalDuration(PropertySetterValueMilliseconds):

900

propname = "ApprovalDuration"

901

902

903

904

class TestCaseWithAssertLogs(unittest.TestCase):

905

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

906

907

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

908

@contextlib.contextmanager

909

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

910

capturing_handler = self.CapturingLevelHandler(level)

911

old_level = logger.level

912

old_propagate = logger.propagate

913

logger.addHandler(capturing_handler)

914

logger.setLevel(level)

915

logger.propagate = False

916

try:

917

yield capturing_handler.watcher

918

finally:

919

logger.propagate = old_propagate

920

logger.removeHandler(capturing_handler)

921

logger.setLevel(old_level)

922

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

923

0)

924

925

class CapturingLevelHandler(logging.Handler):

926

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

927

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

928

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

929

def emit(self, record):

930

self.watcher.records.append(record)

931

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

932

933

LoggingWatcher = collections.namedtuple("LoggingWatcher",

934

("records",

935

"output"))

936

937

938

class Unique(object):

939

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

940

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

941

942

943

class Test_string_to_delta(TestCaseWithAssertLogs):

944

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

945

# rfc3339_duration_to_delta() already has more comprehensive

946

# tests, which are run by doctest.

947

948

def test_rfc3339_zero_seconds(self):

949

self.assertEqual(datetime.timedelta(),

950

string_to_delta("PT0S"))

951

952

def test_rfc3339_zero_days(self):

953

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

954

955

def test_rfc3339_one_second(self):

956

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

957

string_to_delta("PT1S"))

958

959

def test_rfc3339_two_hours(self):

960

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

961

string_to_delta("PT2H"))

962

963

def test_falls_back_to_pre_1_6_1_with_warning(self):

964

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

965

value = string_to_delta("2h")

966

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

967

968

969

class Test_check_option_syntax(unittest.TestCase):

970

def setUp(self):

971

self.parser = argparse.ArgumentParser()

972

add_command_line_options(self.parser)

973

974

def test_actions_requires_client_or_all(self):

975

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

976

args = self.actionargs(action, value)

977

with self.assertParseError():

978

self.parse_args(args)

979

980

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

981

# check_option_syntax()

982

actions = {

983

"--enable": None,

984

"--disable": None,

985

"--bump-timeout": None,

986

"--start-checker": None,

987

"--stop-checker": None,

988

"--is-enabled": None,

989

"--remove": None,

990

"--checker": "x",

991

"--timeout": "PT0S",

992

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

993

"--interval": "PT0S",

994

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

995

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

996

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

997

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

998

"--host": "hostname",

999

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

1000

"--approve": None,

1001

"--deny": None,

1002

}

1003

1004

@staticmethod

1005

def actionargs(action, value, *args):

1006

if value is not None:

1007

return [action, value] + list(args)

1008

else:

1009

return [action] + list(args)

1010

1011

@contextlib.contextmanager

1012

def assertParseError(self):

1013

with self.assertRaises(SystemExit) as e:

1014

with self.redirect_stderr_to_devnull():

1015

yield

1016

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

1017

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

1018

# /argparse.html#exiting-methods

1019

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

1020

1021

def parse_args(self, args):

1022

options = self.parser.parse_args(args)

1023

check_option_syntax(self.parser, options)

1024

1025

@staticmethod

1026

@contextlib.contextmanager

1027

def redirect_stderr_to_devnull():

1028

old_stderr = sys.stderr

1029

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

1030

sys.stderr = null

1031

try:

1032

yield

1033

finally:

1034

sys.stderr = old_stderr

1035

1036

def check_option_syntax(self, options):

1037

check_option_syntax(self.parser, options)

1038

1039

def test_actions_all_conflicts_with_verbose(self):

1040

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

1041

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

1042

"--verbose")

1043

with self.assertParseError():

1044

self.parse_args(args)

1045

1046

def test_actions_with_client_conflicts_with_verbose(self):

1047

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

1048

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

1049

"client")

1050

with self.assertParseError():

1051

self.parse_args(args)

1052

1053

def test_dump_json_conflicts_with_verbose(self):

1054

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

1055

with self.assertParseError():

1056

self.parse_args(args)

1057

1058

def test_dump_json_conflicts_with_action(self):

1059

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

1060

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

1061

with self.assertParseError():

1062

self.parse_args(args)

1063

1064

def test_all_can_not_be_alone(self):

1065

args = ["--all"]

1066

with self.assertParseError():

1067

self.parse_args(args)

1068

1069

def test_all_is_ok_with_any_action(self):

1070

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

1071

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

1072

self.parse_args(args)

1073

1074

def test_any_action_is_ok_with_one_client(self):

1075

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

1076

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

1077

self.parse_args(args)

1078

1079

def test_one_client_with_all_actions_except_is_enabled(self):

1080

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

1081

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

1082

continue

1083

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

1084

self.parse_args(args)

1085

1086

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

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

1088

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

1089

continue

1090

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

1091

"client2")

1092

self.parse_args(args)

1093

1094

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1095

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

1096

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

1097

continue

1098

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

1099

"client2")

1100

self.parse_args(args)

1101

1102

def test_is_enabled_fails_without_client(self):

1103

args = ["--is-enabled"]

1104

with self.assertParseError():

1105

self.parse_args(args)

1106

1107

def test_is_enabled_fails_with_two_clients(self):

1108

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

1109

with self.assertParseError():

1110

self.parse_args(args)

1111

1112

def test_remove_can_only_be_combined_with_action_deny(self):

1113

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

1114

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

1115

continue

1116

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

1117

"--remove")

1118

with self.assertParseError():

1119

self.parse_args(args)

1120

1121

1122

class Test_dbus_exceptions(unittest.TestCase):

1123

1124

def test_dbus_ConnectFailed_is_Error(self):

1125

with self.assertRaises(dbus.Error):

1126

raise dbus.ConnectFailed()

1127

1128

1129

class Test_dbus_MandosBus(unittest.TestCase):

1130

1131

class MockMandosBus(dbus.MandosBus):

1132

def __init__(self):

1133

self._name = "se.recompile.Mandos"

1134

self._server_path = "/"

1135

self._server_interface = "se.recompile.Mandos"

1136

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

1137

self.calls = []

1138

self.call_method_return = Unique()

1139

1140

def call_method(self, methodname, busname, objectpath,

1141

interface, *args):

1142

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

1143

interface, args))

1144

return self.call_method_return

1145

1146

def setUp(self):

1147

self.bus = self.MockMandosBus()

1148

1149

def test_set_client_property(self):

1150

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

1151

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

1152

"org.freedesktop.DBus.Properties",

1153

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

1154

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

1155

1156

def test_call_client_method(self):

1157

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

1158

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

1159

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

1160

self.bus._client_interface, ())

1161

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

1162

1163

def test_call_client_method_with_args(self):

1164

args = (Unique(), Unique())

1165

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

1166

*args)

1167

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

1168

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

1169

self.bus._client_interface,

1170

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

1171

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

1172

1173

def test_get_clients_and_properties(self):

1174

managed_objects = {

1175

"objectpath": {

1176

self.bus._client_interface: {

1177

"key": "value",

1178

"bool": True,

1179

},

1180

"irrelevant_interface": {

1181

"key": "othervalue",

1182

"bool": False,

1183

},

1184

},

1185

"other_objectpath": {

1186

"other_irrelevant_interface": {

1187

"key": "value 3",

1188

"bool": None,

1189

},

1190

},

1191

}

1192

expected_clients_and_properties = {

1193

"objectpath": {

1194

"key": "value",

1195

"bool": True,

1196

}

1197

}

1198

self.bus.call_method_return = managed_objects

1199

ret = self.bus.get_clients_and_properties()

1200

self.assertDictEqual(expected_clients_and_properties, ret)

1201

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

1202

self.bus._server_path,

1203

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

1204

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

1205

1206

def test_call_server_method(self):

1207

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

1208

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

1209

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

1210

self.bus._server_path,

1211

self.bus._server_interface, ())

1212

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

1213

1214

def test_call_server_method_with_args(self):

1215

args = (Unique(), Unique())

1216

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

1217

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

1218

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

1219

self.bus._server_path,

1220

self.bus._server_interface,

1221

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

1222

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

1223

1224

1225

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1226

1227

def MockDBusPython_func(self, func):

1228

class mock_dbus_python(object):

1229

"""mock dbus-python module"""

1230

class exceptions(object):

1231

"""Pseudo-namespace"""

1232

class DBusException(Exception):

1233

pass

1234

class SystemBus(object):

1235

@staticmethod

1236

def get_object(busname, objectpath):

1237

DBusObject = collections.namedtuple(

1238

"DBusObject", ("methodname",))

1239

def method(*args, **kwargs):

1240

self.assertEqual({"dbus_interface":

1241

"interface"},

1242

kwargs)

1243

return func(*args)

1244

return DBusObject(methodname=method)

1245

class Boolean(object):

1246

def __init__(self, value):

1247

self.value = bool(value)

1248

def __bool__(self):

1249

return self.value

1250

if sys.version_info.major == 2:

1251

__nonzero__ = __bool__

1252

class ObjectPath(str):

1253

pass

1254

class Dictionary(dict):

1255

pass

1256

return mock_dbus_python

1257

1258

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

1259

interface, *args):

1260

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

1261

return bus.call_method(methodname, busname, objectpath,

1262

interface, *args)

1263

1264

def test_call_method_returns(self):

1265

expected_method_return = Unique()

1266

method_args = (Unique(), Unique())

1267

def func(*args):

1268

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

1269

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

1270

self.assertIs(marg, arg)

1271

return expected_method_return

1272

mock_dbus_python = self.MockDBusPython_func(func)

1273

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1274

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

1275

"objectpath", "interface",

1276

*method_args)

1277

self.assertIs(ret, expected_method_return)

1278

1279

def test_call_method_filters_bool_true(self):

1280

def func():

1281

return method_return

1282

mock_dbus_python = self.MockDBusPython_func(func)

1283

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1284

method_return = mock_dbus_python.Boolean(True)

1285

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

1286

"objectpath", "interface")

1287

self.assertTrue(ret)

1288

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1289

1290

def test_call_method_filters_bool_false(self):

1291

def func():

1292

return method_return

1293

mock_dbus_python = self.MockDBusPython_func(func)

1294

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1295

method_return = mock_dbus_python.Boolean(False)

1296

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

1297

"objectpath", "interface")

1298

self.assertFalse(ret)

1299

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1300

1301

def test_call_method_filters_objectpath(self):

1302

def func():

1303

return method_return

1304

mock_dbus_python = self.MockDBusPython_func(func)

1305

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1306

method_return = mock_dbus_python.ObjectPath("objectpath")

1307

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

1308

"objectpath", "interface")

1309

self.assertEqual("objectpath", ret)

1310

self.assertIsNot("objectpath", ret)

1311

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1312

1313

def test_call_method_filters_booleans_in_dict(self):

1314

def func():

1315

return method_return

1316

mock_dbus_python = self.MockDBusPython_func(func)

1317

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1318

method_return = mock_dbus_python.Dictionary(

1319

{mock_dbus_python.Boolean(True):

1320

mock_dbus_python.Boolean(False),

1321

mock_dbus_python.Boolean(False):

1322

mock_dbus_python.Boolean(True)})

1323

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

1324

"objectpath", "interface")

1325

expected_method_return = {True: False,

1326

False: True}

1327

self.assertEqual(expected_method_return, ret)

1328

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1329

1330

def test_call_method_filters_objectpaths_in_dict(self):

1331

def func():

1332

return method_return

1333

mock_dbus_python = self.MockDBusPython_func(func)

1334

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1335

method_return = mock_dbus_python.Dictionary(

1336

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1337

mock_dbus_python.ObjectPath("objectpath_value_1"),

1338

mock_dbus_python.ObjectPath("objectpath_key_2"):

1339

mock_dbus_python.ObjectPath("objectpath_value_2")})

1340

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

1341

"objectpath", "interface")

1342

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

1343

for key, value in

1344

method_return.items()}

1345

self.assertEqual(expected_method_return, ret)

1346

self.assertIsInstance(ret, dict)

1347

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1348

1349

def test_call_method_filters_dict_in_dict(self):

1350

def func():

1351

return method_return

1352

mock_dbus_python = self.MockDBusPython_func(func)

1353

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1354

method_return = mock_dbus_python.Dictionary(

1355

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

1356

"key12": "value12"}),

1357

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

1358

"key22": "value22"})})

1359

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

1360

"objectpath", "interface")

1361

expected_method_return = {

1362

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

1363

"key12": "value12"},

1364

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

1365

"key22": "value22"},

1366

}

1367

self.assertEqual(expected_method_return, ret)

1368

self.assertIsInstance(ret, dict)

1369

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1370

for key, value in ret.items():

1371

self.assertIsInstance(value, dict)

1372

self.assertEqual(expected_method_return[key], value)

1373

self.assertNotIsInstance(value,

1374

mock_dbus_python.Dictionary)

1375

1376

def test_call_method_filters_dict_three_deep(self):

1377

def func():

1378

return method_return

1379

mock_dbus_python = self.MockDBusPython_func(func)

1380

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1381

method_return = mock_dbus_python.Dictionary(

1382

{"key1":

1383

mock_dbus_python.Dictionary(

1384

{"key2":

1385

mock_dbus_python.Dictionary(

1386

{"key3":

1387

mock_dbus_python.Boolean(True),

1388

}),

1389

}),

1390

})

1391

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

1392

"objectpath", "interface")

1393

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

1394

self.assertEqual(expected_method_return, ret)

1395

self.assertIsInstance(ret, dict)

1396

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1397

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

1398

self.assertNotIsInstance(ret["key1"],

1399

mock_dbus_python.Dictionary)

1400

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

1401

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

1402

mock_dbus_python.Dictionary)

1403

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

1404

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

1405

mock_dbus_python.Boolean)

1406

1407

def test_call_method_handles_exception(self):

1408

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

1409

1410

def func():

1411

dbus_logger.error("Test")

1412

raise mock_dbus_python.exceptions.DBusException()

1413

1414

mock_dbus_python = self.MockDBusPython_func(func)

1415

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1416

1417

class CountingHandler(logging.Handler):

1418

count = 0

1419

def emit(self, record):

1420

self.count += 1

1421

1422

counting_handler = CountingHandler()

1423

1424

dbus_logger.addHandler(counting_handler)

1425

1426

try:

1427

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

1428

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

1429

"objectpath", "interface")

1430

finally:

1431

dbus_logger.removeFilter(counting_handler)

1432

1433

self.assertNotIsInstance(e, dbus.ConnectFailed)

1434

1435

# Make sure the dbus logger was suppressed

1436

self.assertEqual(0, counting_handler.count)

1437

1438

def test_get_object_converts_to_correct_exception(self):

1439

bus = dbus_python_adapter.SystemBus(

1440

self.fake_dbus_python_raises_exception_on_connect)

1441

with self.assertRaises(dbus.ConnectFailed):

1442

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

1443

"objectpath", "interface")

1444

1445

class fake_dbus_python_raises_exception_on_connect(object):

1446

"""fake dbus-python module"""

1447

class exceptions(object):

1448

"""Pseudo-namespace"""

1449

class DBusException(Exception):

1450

pass

1451

1452

@classmethod

1453

def SystemBus(cls):

1454

def get_object(busname, objectpath):

1455

raise cls.exceptions.DBusException()

1456

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

1457

return Bus(get_object=get_object)

1458

1459

1460

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1461

class mock_dbus_python(object):

1462

"""mock dbus-python modules"""

1463

class SystemBus(object):

1464

@staticmethod

1465

def get_object(busname, objectpath):

1466

return Unique()

1467

1468

def setUp(self):

1469

self.bus = dbus_python_adapter.CachingBus(

1470

self.mock_dbus_python)

1471

1472

def test_returns_distinct_objectpaths(self):

1473

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

1474

self.assertIsInstance(obj1, Unique)

1475

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

1476

self.assertIsInstance(obj2, Unique)

1477

self.assertIsNot(obj1, obj2)

1478

1479

def test_returns_distinct_busnames(self):

1480

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

1481

self.assertIsInstance(obj1, Unique)

1482

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

1483

self.assertIsInstance(obj2, Unique)

1484

self.assertIsNot(obj1, obj2)

1485

1486

def test_returns_distinct_both(self):

1487

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

1488

self.assertIsInstance(obj1, Unique)

1489

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

1490

self.assertIsInstance(obj2, Unique)

1491

self.assertIsNot(obj1, obj2)

1492

1493

def test_returns_same(self):

1494

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

1495

self.assertIsInstance(obj1, Unique)

1496

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

1497

self.assertIsInstance(obj2, Unique)

1498

self.assertIs(obj1, obj2)

1499

1500

def test_returns_same_old(self):

1501

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

1502

self.assertIsInstance(obj1, Unique)

1503

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

1504

self.assertIsInstance(obj2, Unique)

1505

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

1506

self.assertIsInstance(obj1b, Unique)

1507

self.assertIsNot(obj1, obj2)

1508

self.assertIsNot(obj2, obj1b)

1509

self.assertIs(obj1, obj1b)

1510

1511

1512

class Test_commands_from_options(unittest.TestCase):

1513

1514

def setUp(self):

1515

self.parser = argparse.ArgumentParser()

1516

add_command_line_options(self.parser)

1517

1518

def test_is_enabled(self):

1519

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

1520

command.IsEnabled)

1521

1522

def assert_command_from_args(self, args, command_cls,

1523

**cmd_attrs):

1524

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

1525

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

1526

options = self.parser.parse_args(args)

1527

check_option_syntax(self.parser, options)

1528

commands = commands_from_options(options)

1529

self.assertEqual(1, len(commands))

1530

command = commands[0]

1531

self.assertIsInstance(command, command_cls)

1532

for key, value in cmd_attrs.items():

1533

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

1534

1535

def test_is_enabled_short(self):

1536

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

1537

command.IsEnabled)

1538

1539

def test_approve(self):

1540

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

1541

command.Approve)

1542

1543

def test_approve_short(self):

1544

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

1545

command.Approve)

1546

1547

def test_deny(self):

1548

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

1549

command.Deny)

1550

1551

def test_deny_short(self):

1552

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

1553

1554

def test_remove(self):

1555

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

1556

command.Remove)

1557

1558

def test_deny_before_remove(self):

1559

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

1560

"client"])

1561

check_option_syntax(self.parser, options)

1562

commands = commands_from_options(options)

1563

self.assertEqual(2, len(commands))

1564

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

1565

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

1566

1567

def test_deny_before_remove_reversed(self):

1568

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

1569

"--all"])

1570

check_option_syntax(self.parser, options)

1571

commands = commands_from_options(options)

1572

self.assertEqual(2, len(commands))

1573

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

1574

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

1575

1576

def test_remove_short(self):

1577

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

1578

command.Remove)

1579

1580

def test_dump_json(self):

1581

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

1582

command.DumpJSON)

1583

1584

def test_enable(self):

1585

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

1586

command.Enable)

1587

1588

def test_enable_short(self):

1589

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

1590

command.Enable)

1591

1592

def test_disable(self):

1593

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

1594

command.Disable)

1595

1596

def test_disable_short(self):

1597

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

1598

command.Disable)

1599

1600

def test_bump_timeout(self):

1601

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

1602

command.BumpTimeout)

1603

1604

def test_bump_timeout_short(self):

1605

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

1606

command.BumpTimeout)

1607

1608

def test_start_checker(self):

1609

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

1610

command.StartChecker)

1611

1612

def test_stop_checker(self):

1613

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

1614

command.StopChecker)

1615

1616

def test_approve_by_default(self):

1617

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

1618

"client"],

1619

command.ApproveByDefault)

1620

1621

def test_deny_by_default(self):

1622

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

1623

command.DenyByDefault)

1624

1625

def test_checker(self):

1626

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

1627

command.SetChecker,

1628

value_to_set=":")

1629

1630

def test_checker_empty(self):

1631

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

1632

command.SetChecker,

1633

value_to_set="")

1634

1635

def test_checker_short(self):

1636

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

1637

command.SetChecker,

1638

value_to_set=":")

1639

1640

def test_host(self):

1641

self.assert_command_from_args(

1642

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

1643

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

1644

1645

def test_host_short(self):

1646

self.assert_command_from_args(

1647

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

1648

value_to_set="client.example.org")

1649

1650

def test_secret_devnull(self):

1651

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

1652

"client"], command.SetSecret,

1653

value_to_set=b"")

1654

1655

def test_secret_tempfile(self):

1656

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

1657

value = b"secret\0xyzzy\nbar"

1658

f.write(value)

1659

f.seek(0)

1660

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

1661

"client"],

1662

command.SetSecret,

1663

value_to_set=value)

1664

1665

def test_secret_devnull_short(self):

1666

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

1667

"client"], command.SetSecret,

1668

value_to_set=b"")

1669

1670

def test_secret_tempfile_short(self):

1671

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

1672

value = b"secret\0xyzzy\nbar"

1673

f.write(value)

1674

f.seek(0)

1675

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

1676

command.SetSecret,

1677

value_to_set=value)

1678

1679

def test_timeout(self):

1680

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

1681

command.SetTimeout,

1682

value_to_set=300000)

1683

1684

def test_timeout_short(self):

1685

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

1686

command.SetTimeout,

1687

value_to_set=300000)

1688

1689

def test_extended_timeout(self):

1690

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

1691

"client"],

1692

command.SetExtendedTimeout,

1693

value_to_set=900000)

1694

1695

def test_interval(self):

1696

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

1697

"client"], command.SetInterval,

1698

value_to_set=120000)

1699

1700

def test_interval_short(self):

1701

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

1702

command.SetInterval,

1703

value_to_set=120000)

1704

1705

def test_approval_delay(self):

1706

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

1707

"client"],

1708

command.SetApprovalDelay,

1709

value_to_set=30000)

1710

1711

def test_approval_duration(self):

1712

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

1713

"client"],

1714

command.SetApprovalDuration,

1715

value_to_set=1000)

1716

1717

def test_print_table(self):

1718

self.assert_command_from_args([], command.PrintTable,

1719

verbose=False)

1720

1721

def test_print_table_verbose(self):

1722

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

1723

command.PrintTable,

1724

verbose=True)

1725

1726

def test_print_table_verbose_short(self):

1727

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

1728

verbose=True)

1729

1730

1731

class TestCommand(unittest.TestCase):

1732

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

1733

1734

class FakeMandosBus(dbus.MandosBus):

1735

def __init__(self, testcase):

1736

self.client_properties = {

1737

"Name": "foo",

1738

"KeyID": ("92ed150794387c03ce684574b1139a65"

1739

"94a34f895daaaf09fd8ea90a27cddb12"),

1740

"Secret": b"secret",

1741

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

1742

"Enabled": True,

1743

"Timeout": 300000,

1744

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

1745

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

1746

"Interval": 120000,

1747

"Fingerprint": ("778827225BA7DE539C5A"

1748

"7CFA59CFF7CDBD9A5920"),

1749

"CheckerRunning": False,

1750

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

1751

"ApprovalPending": False,

1752

"ApprovedByDefault": True,

1753

"LastApprovalRequest": "",

1754

"ApprovalDelay": 0,

1755

"ApprovalDuration": 1000,

1756

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

1757

"ExtendedTimeout": 900000,

1758

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

1759

"LastCheckerStatus": 0,

1760

}

1761

self.other_client_properties = {

1762

"Name": "barbar",

1763

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1764

"6ab612cff5ad227247e46c2b020f441c"),

1765

"Secret": b"secretbar",

1766

"Host": "192.0.2.3",

1767

"Enabled": True,

1768

"Timeout": 300000,

1769

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

1770

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

1771

"Interval": 120000,

1772

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1773

"F547B3A107558FCA3A27"),

1774

"CheckerRunning": True,

1775

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

1776

"ApprovalPending": False,

1777

"ApprovedByDefault": False,

1778

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

1779

"ApprovalDelay": 30000,

1780

"ApprovalDuration": 93785000,

1781

"Checker": ":",

1782

"ExtendedTimeout": 900000,

1783

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

1784

"LastCheckerStatus": -2,

1785

}

1786

self.clients = collections.OrderedDict(

1787

[

1788

("client_objectpath", self.client_properties),

1789

("other_client_objectpath",

1790

self.other_client_properties),

1791

])

1792

self.one_client = {"client_objectpath":

1793

self.client_properties}

1794

self.testcase = testcase

1795

self.calls = []

1796

1797

def call_method(self, methodname, busname, objectpath,

1798

interface, *args):

1799

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

1800

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

1801

interface, args))

1802

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

1803

if methodname == "Set":

1804

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

1805

interface, key, value = args

1806

self.testcase.assertEqual(

1807

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

1808

self.clients[objectpath][key] = value

1809

return

1810

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

1811

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

1812

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

1813

clientpath = args[0]

1814

del self.clients[clientpath]

1815

return

1816

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

1817

if methodname == "Approve":

1818

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

1819

return

1820

raise ValueError()

1821

1822

def setUp(self):

1823

self.bus = self.FakeMandosBus(self)

1824

1825

1826

class TestBaseCommands(TestCommand):

1827

1828

def test_IsEnabled_exits_successfully(self):

1829

with self.assertRaises(SystemExit) as e:

1830

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

1831

if e.exception.code is not None:

1832

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

1833

else:

1834

self.assertIsNone(e.exception.code)

1835

1836

def test_IsEnabled_exits_with_failure(self):

1837

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

1838

with self.assertRaises(SystemExit) as e:

1839

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

1840

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

1841

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

1842

else:

1843

self.assertIsNotNone(e.exception.code)

1844

1845

def test_Approve(self):

1846

busname = "se.recompile.Mandos"

1847

client_interface = "se.recompile.Mandos.Client"

1848

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

1849

for clientpath in self.bus.clients:

1850

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

1851

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

1852

1853

def test_Deny(self):

1854

busname = "se.recompile.Mandos"

1855

client_interface = "se.recompile.Mandos.Client"

1856

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

1857

for clientpath in self.bus.clients:

1858

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

1859

client_interface, (False,)),

1860

self.bus.calls)

1861

1862

def test_Remove(self):

1863

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

1864

for clientpath in self.bus.clients:

1865

self.assertIn(("RemoveClient", dbus_busname,

1866

dbus_server_path, dbus_server_interface,

1867

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

1868

1869

expected_json = {

1870

"foo": {

1871

"Name": "foo",

1872

"KeyID": ("92ed150794387c03ce684574b1139a65"

1873

"94a34f895daaaf09fd8ea90a27cddb12"),

1874

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

1875

"Enabled": True,

1876

"Timeout": 300000,

1877

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

1878

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

1879

"Interval": 120000,

1880

"Fingerprint": ("778827225BA7DE539C5A"

1881

"7CFA59CFF7CDBD9A5920"),

1882

"CheckerRunning": False,

1883

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

1884

"ApprovalPending": False,

1885

"ApprovedByDefault": True,

1886

"LastApprovalRequest": "",

1887

"ApprovalDelay": 0,

1888

"ApprovalDuration": 1000,

1889

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

1890

"ExtendedTimeout": 900000,

1891

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

1892

"LastCheckerStatus": 0,

1893

},

1894

"barbar": {

1895

"Name": "barbar",

1896

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1897

"6ab612cff5ad227247e46c2b020f441c"),

1898

"Host": "192.0.2.3",

1899

"Enabled": True,

1900

"Timeout": 300000,

1901

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

1902

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

1903

"Interval": 120000,

1904

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1905

"F547B3A107558FCA3A27"),

1906

"CheckerRunning": True,

1907

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

1908

"ApprovalPending": False,

1909

"ApprovedByDefault": False,

1910

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

1911

"ApprovalDelay": 30000,

1912

"ApprovalDuration": 93785000,

1913

"Checker": ":",

1914

"ExtendedTimeout": 900000,

1915

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

1916

"LastCheckerStatus": -2,

1917

},

1918

}

1919

1920

def test_DumpJSON_normal(self):

1921

with self.capture_stdout_to_buffer() as buffer:

1922

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

1923

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

1924

self.assertDictEqual(self.expected_json, json_data)

1925

1926

@staticmethod

1927

@contextlib.contextmanager

1928

def capture_stdout_to_buffer():

1929

capture_buffer = io.StringIO()

1930

old_stdout = sys.stdout

1931

sys.stdout = capture_buffer

1932

try:

1933

yield capture_buffer

1934

finally:

1935

sys.stdout = old_stdout

1936

1937

def test_DumpJSON_one_client(self):

1938

with self.capture_stdout_to_buffer() as buffer:

1939

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

1940

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

1941

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

1942

self.assertDictEqual(expected_json, json_data)

1943

1944

def test_PrintTable_normal(self):

1945

with self.capture_stdout_to_buffer() as buffer:

1946

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

1947

expected_output = "\n".join((

1948

"Name Enabled Timeout Last Successful Check",

1949

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

1950

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

1951

)) + "\n"

1952

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

1953

1954

def test_PrintTable_verbose(self):

1955

with self.capture_stdout_to_buffer() as buffer:

1956

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

1957

columns = (

1958

(

1959

"Name ",

1960

"foo ",

1961

"barbar ",

1962

),(

1963

"Enabled ",

1964

"Yes ",

1965

"Yes ",

1966

),(

1967

"Timeout ",

1968

"00:05:00 ",

1969

"00:05:00 ",

1970

),(

1971

"Last Successful Check ",

1972

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

1973

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

1974

),(

1975

"Created ",

1976

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

1977

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

1978

),(

1979

"Interval ",

1980

"00:02:00 ",

1981

"00:02:00 ",

1982

),(

1983

"Host ",

1984

"foo.example.org ",

1985

"192.0.2.3 ",

1986

),(

1987

("Key ID "

1988

" "),

1989

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1990

"ea90a27cddb12 "),

1991

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1992

"46c2b020f441c "),

1993

),(

1994

"Fingerprint ",

1995

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1996

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1997

),(

1998

"Check Is Running ",

1999

"No ",

2000

"Yes ",

2001

),(

2002

"Last Enabled ",

2003

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

2004

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

2005

),(

2006

"Approval Is Pending ",

2007

"No ",

2008

"No ",

2009

),(

2010

"Approved By Default ",

2011

"Yes ",

2012

"No ",

2013

),(

2014

"Last Approval Request ",

2015

" ",

2016

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

2017

),(

2018

"Approval Delay ",

2019

"00:00:00 ",

2020

"00:00:30 ",

2021

),(

2022

"Approval Duration ",

2023

"00:00:01 ",

2024

"1T02:03:05 ",

2025

),(

2026

"Checker ",

2027

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

2028

": ",

2029

),(

2030

"Extended Timeout ",

2031

"00:15:00 ",

2032

"00:15:00 ",

2033

),(

2034

"Expires ",

2035

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

2036

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

2037

),(

2038

"Last Checker Status",

2039

"0 ",

2040

"-2 ",

2041

)

2042

)

2043

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

2044

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

2045

for rows in columns)

2046

for line in range(num_lines))

2047

+ "\n")

2048

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

2049

2050

def test_PrintTable_one_client(self):

2051

with self.capture_stdout_to_buffer() as buffer:

2052

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

2053

expected_output = "\n".join((

2054

"Name Enabled Timeout Last Successful Check",

2055

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

2056

)) + "\n"

2057

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

2058

2059

2060

class TestPropertySetterCmd(TestCommand):

2061

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

2062

2063

def runTest(self):

2064

if not hasattr(self, "command"):

2065

return # Abstract TestCase class

2066

2067

if hasattr(self, "values_to_set"):

2068

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

2069

values_to_get = getattr(self, "values_to_get",

2070

self.values_to_set)

2071

else:

2072

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

2073

values_to_get = self.values_to_get

2074

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

2075

for clientpath in self.bus.clients:

2076

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

2077

Unique())

2078

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

2079

for clientpath in self.bus.clients:

2080

value = (self.bus.clients[clientpath]

2081

[self.propname])

2082

self.assertNotIsInstance(value, Unique)

2083

self.assertEqual(value_to_get, value)

2084

2085

2086

class TestEnableCmd(TestPropertySetterCmd):

2087

command = command.Enable

2088

propname = "Enabled"

2089

values_to_get = [True]

2090

2091

2092

class TestDisableCmd(TestPropertySetterCmd):

2093

command = command.Disable

2094

propname = "Enabled"

2095

values_to_get = [False]

2096

2097

2098

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2099

command = command.BumpTimeout

2100

propname = "LastCheckedOK"

2101

values_to_get = [""]

2102

2103

2104

class TestStartCheckerCmd(TestPropertySetterCmd):

2105

command = command.StartChecker

2106

propname = "CheckerRunning"

2107

values_to_get = [True]

2108

2109

2110

class TestStopCheckerCmd(TestPropertySetterCmd):

2111

command = command.StopChecker

2112

propname = "CheckerRunning"

2113

values_to_get = [False]

2114

2115

2116

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2117

command = command.ApproveByDefault

2118

propname = "ApprovedByDefault"

2119

values_to_get = [True]

2120

2121

2122

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2123

command = command.DenyByDefault

2124

propname = "ApprovedByDefault"

2125

values_to_get = [False]

2126

2127

2128

class TestSetCheckerCmd(TestPropertySetterCmd):

2129

command = command.SetChecker

2130

propname = "Checker"

2131

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

2132

2133

2134

class TestSetHostCmd(TestPropertySetterCmd):

2135

command = command.SetHost

2136

propname = "Host"

2137

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

2138

2139

2140

class TestSetSecretCmd(TestPropertySetterCmd):

2141

command = command.SetSecret

2142

propname = "Secret"

2143

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

2144

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

2145

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

2146

2147

2148

class TestSetTimeoutCmd(TestPropertySetterCmd):

2149

command = command.SetTimeout

2150

propname = "Timeout"

2151

values_to_set = [datetime.timedelta(),

2152

datetime.timedelta(minutes=5),

2153

datetime.timedelta(seconds=1),

2154

datetime.timedelta(weeks=1),

2155

datetime.timedelta(weeks=52)]

2156

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

2157

2158

2159

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2160

command = command.SetExtendedTimeout

2161

propname = "ExtendedTimeout"

2162

values_to_set = [datetime.timedelta(),

2163

datetime.timedelta(minutes=5),

2164

datetime.timedelta(seconds=1),

2165

datetime.timedelta(weeks=1),

2166

datetime.timedelta(weeks=52)]

2167

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

2168

2169

2170

class TestSetIntervalCmd(TestPropertySetterCmd):

2171

command = command.SetInterval

2172

propname = "Interval"

2173

values_to_set = [datetime.timedelta(),

2174

datetime.timedelta(minutes=5),

2175

datetime.timedelta(seconds=1),

2176

datetime.timedelta(weeks=1),

2177

datetime.timedelta(weeks=52)]

2178

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

2179

2180

2181

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2182

command = command.SetApprovalDelay

2183

propname = "ApprovalDelay"

2184

values_to_set = [datetime.timedelta(),

2185

datetime.timedelta(minutes=5),

2186

datetime.timedelta(seconds=1),

2187

datetime.timedelta(weeks=1),

2188

datetime.timedelta(weeks=52)]

2189

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

2190

2191

2192

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2193

command = command.SetApprovalDuration

2194

propname = "ApprovalDuration"

2195

values_to_set = [datetime.timedelta(),

2196

datetime.timedelta(minutes=5),

2197

datetime.timedelta(seconds=1),

2198

datetime.timedelta(weeks=1),

2199

datetime.timedelta(weeks=52)]

2200

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

2201

2202

2203

2204

def should_only_run_tests():

2205

parser = argparse.ArgumentParser(add_help=False)

2206

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

2207

args, unknown_args = parser.parse_known_args()

2208

run_tests = args.check

2209

if run_tests:

2210

# Remove --check argument from sys.argv

2211

sys.argv[1:] = unknown_args

2212

return run_tests

2213

2214

# Add all tests from doctest strings

2215

def load_tests(loader, tests, none):

2216

import doctest

2217

tests.addTests(doctest.DocTestSuite())

2218

return tests

2219

2220

if __name__ == "__main__":

2221

try:

2222

if should_only_run_tests():

2223

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

2224

unittest.main()

2225

else:

2226

main()

2227

finally:

2228

logging.shutdown()