/mandos/trunk : revision 24.1.92

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

49

import dbus as dbus_python

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

import StringIO

65

io.StringIO = StringIO.StringIO

66

67

locale.setlocale(locale.LC_ALL, "")

68

69

version = "1.8.3"

70

71

72

def main():

73

parser = argparse.ArgumentParser()

74

add_command_line_options(parser)

75

76

options = parser.parse_args()

77

check_option_syntax(parser, options)

78

79

clientnames = options.client

80

81

if options.debug:

82

log.setLevel(logging.DEBUG)

83

84

bus = dbus_python_adapter.CachingBus(dbus_python)

85

86

try:

87

all_clients = bus.get_clients_and_properties()

88

except dbus.ConnectFailed as e:

89

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

90

sys.exit(1)

91

except dbus.Error as e:

92

log.critical(

93

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

94

sys.exit(1)

95

96

# Compile dict of (clientpath: properties) to process

97

if not clientnames:

98

clients = all_clients

99

else:

100

clients = {}

101

for name in clientnames:

102

for objpath, properties in all_clients.items():

103

if properties["Name"] == name:

104

clients[objpath] = properties

105

break

106

else:

107

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

108

sys.exit(1)

109

110

commands = commands_from_options(options)

111

112

for command in commands:

113

command.run(clients, bus)

114

115

116

def add_command_line_options(parser):

117

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

118

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

119

help="show version number and exit")

120

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

121

help="Select all clients")

122

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

123

help="Print all fields")

124

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

125

action="append_const", default=[],

126

const=command.DumpJSON(),

127

help="Dump client data in JSON format")

128

enable_disable = parser.add_mutually_exclusive_group()

129

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

130

action="append_const", default=[],

131

const=command.Enable(),

132

help="Enable client")

133

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

134

action="append_const", default=[],

135

const=command.Disable(),

136

help="disable client")

137

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

138

action="append_const", default=[],

139

const=command.BumpTimeout(),

140

help="Bump timeout for client")

141

start_stop_checker = parser.add_mutually_exclusive_group()

142

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

143

dest="commands",

144

action="append_const", default=[],

145

const=command.StartChecker(),

146

help="Start checker for client")

147

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

148

action="append_const", default=[],

149

const=command.StopChecker(),

150

help="Stop checker for client")

151

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

152

action="append_const", default=[],

153

const=command.IsEnabled(),

154

help="Check if client is enabled")

155

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

156

action="append_const", default=[],

157

const=command.Remove(),

158

help="Remove client")

159

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

160

action="append", default=[],

161

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

162

help="Set checker command for client")

163

parser.add_argument(

164

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

165

default=[], metavar="TIME",

166

type=command.SetTimeout.argparse(string_to_delta),

167

help="Set timeout for client")

168

parser.add_argument(

169

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

170

default=[], metavar="TIME",

171

type=command.SetExtendedTimeout.argparse(string_to_delta),

172

help="Set extended timeout for client")

173

parser.add_argument(

174

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

175

default=[], metavar="TIME",

176

type=command.SetInterval.argparse(string_to_delta),

177

help="Set checker interval for client")

178

approve_deny_default = parser.add_mutually_exclusive_group()

179

approve_deny_default.add_argument(

180

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

181

action="append_const", default=[],

182

const=command.ApproveByDefault(),

183

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

184

approve_deny_default.add_argument(

185

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

186

action="append_const", default=[],

187

const=command.DenyByDefault(),

188

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

189

parser.add_argument(

190

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

191

default=[], metavar="TIME",

192

type=command.SetApprovalDelay.argparse(string_to_delta),

193

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

194

parser.add_argument(

195

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

196

default=[], metavar="TIME",

197

type=command.SetApprovalDuration.argparse(string_to_delta),

198

help="Set duration of one client approval")

199

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

200

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

201

type=command.SetHost,

202

help="Set host for client")

203

parser.add_argument(

204

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

205

default=[], metavar="FILENAME",

206

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

207

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

208

approve_deny = parser.add_mutually_exclusive_group()

209

approve_deny.add_argument(

210

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

211

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

212

help="Approve any current client request")

213

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

214

action="append_const", default=[],

215

const=command.Deny(),

216

help="Deny any current client request")

217

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

218

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

219

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

220

help="Run self-test")

221

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

222

223

224

def string_to_delta(interval):

225

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

226

227

try:

228

return rfc3339_duration_to_delta(interval)

229

except ValueError as e:

230

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

231

' '.join(e.args))

232

return parse_pre_1_6_1_interval(interval)

233

234

235

def rfc3339_duration_to_delta(duration):

236

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

237

238

>>> rfc3339_duration_to_delta("P7D")

239

datetime.timedelta(7)

240

>>> rfc3339_duration_to_delta("PT60S")

241

datetime.timedelta(0, 60)

242

>>> rfc3339_duration_to_delta("PT60M")

243

datetime.timedelta(0, 3600)

244

>>> rfc3339_duration_to_delta("P60M")

245

datetime.timedelta(1680)

246

>>> rfc3339_duration_to_delta("PT24H")

247

datetime.timedelta(1)

248

>>> rfc3339_duration_to_delta("P1W")

249

datetime.timedelta(7)

250

>>> rfc3339_duration_to_delta("PT5M30S")

251

datetime.timedelta(0, 330)

252

>>> rfc3339_duration_to_delta("P1DT3M20S")

253

datetime.timedelta(1, 200)

254

>>> # Can not be empty:

255

>>> rfc3339_duration_to_delta("")

256

Traceback (most recent call last):

257

...

258

ValueError: Invalid RFC 3339 duration: ""

259

>>> # Must start with "P":

260

>>> rfc3339_duration_to_delta("1D")

261

Traceback (most recent call last):

262

...

263

ValueError: Invalid RFC 3339 duration: "1D"

264

>>> # Must use correct order

265

>>> rfc3339_duration_to_delta("PT1S2M")

266

Traceback (most recent call last):

267

...

268

ValueError: Invalid RFC 3339 duration: "PT1S2M"

269

>>> # Time needs time marker

270

>>> rfc3339_duration_to_delta("P1H2S")

271

Traceback (most recent call last):

272

...

273

ValueError: Invalid RFC 3339 duration: "P1H2S"

274

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

275

>>> rfc3339_duration_to_delta("P1D2W")

276

Traceback (most recent call last):

277

...

278

ValueError: Invalid RFC 3339 duration: "P1D2W"

279

>>> rfc3339_duration_to_delta("P2W2H")

280

Traceback (most recent call last):

281

...

282

ValueError: Invalid RFC 3339 duration: "P2W2H"

283

"""

284

285

# Parsing an RFC 3339 duration with regular expressions is not

286

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

287

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

288

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

289

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

290

# avoid excessive use of external libraries.

291

292

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

293

Token = collections.namedtuple("Token", (

294

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

295

# a "group" containing digits

296

"value", # datetime.timedelta or None

297

"followers")) # Tokens valid after this token

298

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

299

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

300

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

301

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

302

datetime.timedelta(seconds=1),

303

frozenset((token_end, )))

304

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

305

datetime.timedelta(minutes=1),

306

frozenset((token_second, token_end)))

307

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

308

datetime.timedelta(hours=1),

309

frozenset((token_minute, token_end)))

310

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

311

None,

312

frozenset((token_hour, token_minute,

313

token_second)))

314

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

315

datetime.timedelta(days=1),

316

frozenset((token_time, token_end)))

317

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

318

datetime.timedelta(weeks=4),

319

frozenset((token_day, token_end)))

320

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

321

datetime.timedelta(weeks=52),

322

frozenset((token_month, token_end)))

323

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

324

datetime.timedelta(weeks=1),

325

frozenset((token_end, )))

326

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

327

frozenset((token_year, token_month,

328

token_day, token_time,

329

token_week)))

330

# Define starting values:

331

# Value so far

332

value = datetime.timedelta()

333

found_token = None

334

# Following valid tokens

335

followers = frozenset((token_duration, ))

336

# String left to parse

337

s = duration

338

# Loop until end token is found

339

while found_token is not token_end:

340

# Search for any currently valid tokens

341

for token in followers:

342

match = token.regexp.match(s)

343

if match is not None:

344

# Token found

345

if token.value is not None:

346

# Value found, parse digits

347

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

348

# Add to value so far

349

value += factor * token.value

350

# Strip token from string

351

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

352

# Go to found token

353

found_token = token

354

# Set valid next tokens

355

followers = found_token.followers

356

break

357

else:

358

# No currently valid tokens were found

359

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

360

.format(duration))

361

# End token found

362

return value

363

364

365

def parse_pre_1_6_1_interval(interval):

366

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

367

and return a datetime.timedelta

368

369

>>> parse_pre_1_6_1_interval('7d')

370

datetime.timedelta(7)

371

>>> parse_pre_1_6_1_interval('60s')

372

datetime.timedelta(0, 60)

373

>>> parse_pre_1_6_1_interval('60m')

374

datetime.timedelta(0, 3600)

375

>>> parse_pre_1_6_1_interval('24h')

376

datetime.timedelta(1)

377

>>> parse_pre_1_6_1_interval('1w')

378

datetime.timedelta(7)

379

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

380

datetime.timedelta(0, 330)

381

>>> parse_pre_1_6_1_interval('')

382

datetime.timedelta(0)

383

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

384

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

385

datetime.timedelta(2, 480, 18000)

386

387

"""

388

389

value = datetime.timedelta(0)

390

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

391

392

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

393

if suffix == "d":

394

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

395

elif suffix == "s":

396

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

397

elif suffix == "m":

398

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

399

elif suffix == "h":

400

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

401

elif suffix == "w":

402

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

403

elif suffix == "":

404

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

405

return value

406

407

408

def check_option_syntax(parser, options):

409

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

410

argparse"""

411

412

def has_commands(options, commands=None):

413

if commands is None:

414

commands = (command.Enable,

415

command.Disable,

416

command.BumpTimeout,

417

command.StartChecker,

418

command.StopChecker,

419

command.IsEnabled,

420

command.Remove,

421

command.SetChecker,

422

command.SetTimeout,

423

command.SetExtendedTimeout,

424

command.SetInterval,

425

command.ApproveByDefault,

426

command.DenyByDefault,

427

command.SetApprovalDelay,

428

command.SetApprovalDuration,

429

command.SetHost,

430

command.SetSecret,

431

command.Approve,

432

command.Deny)

433

return any(isinstance(cmd, commands)

434

for cmd in options.commands)

435

436

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

437

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

438

if options.verbose and has_commands(options):

439

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

440

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

441

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

442

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

443

if options.all and not has_commands(options):

444

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

445

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

446

and len(options.client) > 1):

447

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

448

if (len(options.commands) > 1

449

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

450

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

451

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

452

453

454

class dbus(object):

455

456

class SystemBus(object):

457

458

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

459

def get_managed_objects(self, busname, objectpath):

460

return self.call_method("GetManagedObjects", busname,

461

objectpath,

462

self.object_manager_iface)

463

464

properties_iface = "org.freedesktop.DBus.Properties"

465

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

466

value):

467

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

468

self.properties_iface, interface, key,

469

value)

470

471

472

class MandosBus(SystemBus):

473

busname_domain = "se.recompile"

474

busname = busname_domain + ".Mandos"

475

server_path = "/"

476

server_interface = busname_domain + ".Mandos"

477

client_interface = busname_domain + ".Mandos.Client"

478

del busname_domain

479

480

def get_clients_and_properties(self):

481

managed_objects = self.get_managed_objects(

482

self.busname, self.server_path)

483

return {objpath: properties[self.client_interface]

484

for objpath, properties in managed_objects.items()

485

if self.client_interface in properties}

486

487

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

488

return self.set_property(self.busname, objectpath,

489

self.client_interface, key,

490

value)

491

492

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

493

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

494

self.client_interface, *args)

495

496

def call_server_method(self, method, *args):

497

return self.call_method(method, self.busname,

498

self.server_path,

499

self.server_interface, *args)

500

501

class Error(Exception):

502

pass

503

504

class ConnectFailed(Error):

505

pass

506

507

508

class dbus_python_adapter(object):

509

510

class SystemBus(dbus.MandosBus):

511

"""Use dbus-python"""

512

513

def __init__(self, module=dbus_python):

514

self.dbus_python = module

515

self.bus = self.dbus_python.SystemBus()

516

517

@contextlib.contextmanager

518

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

519

try:

520

yield

521

except self.dbus_python.exceptions.DBusException as e:

522

# This does what "raise from" would do

523

exc = exception_class(*e.args)

524

exc.__cause__ = e

525

raise exc

526

527

def call_method(self, methodname, busname, objectpath,

528

interface, *args):

529

proxy_object = self.get_object(busname, objectpath)

530

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

531

interface, methodname,

532

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

533

method = getattr(proxy_object, methodname)

534

with self.convert_exception():

535

with dbus_python_adapter.SilenceLogger(

536

"dbus.proxies"):

537

value = method(*args, dbus_interface=interface)

538

return self.type_filter(value)

539

540

def get_object(self, busname, objectpath):

541

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

542

busname, objectpath)

543

with self.convert_exception(dbus.ConnectFailed):

544

return self.bus.get_object(busname, objectpath)

545

546

def type_filter(self, value):

547

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

548

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

549

return bool(value)

550

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

551

return str(value)

552

# Also recurse into dictionaries

553

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

554

return {self.type_filter(key):

555

self.type_filter(subval)

556

for key, subval in value.items()}

557

return value

558

559

560

class SilenceLogger(object):

561

"Simple context manager to silence a particular logger"

562

def __init__(self, loggername):

563

self.logger = logging.getLogger(loggername)

564

565

def __enter__(self):

566

self.logger.addFilter(self.nullfilter)

567

568

class NullFilter(logging.Filter):

569

def filter(self, record):

570

return False

571

572

nullfilter = NullFilter()

573

574

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

575

self.logger.removeFilter(self.nullfilter)

576

577

578

class CachingBus(SystemBus):

579

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

580

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

581

self.object_cache = {}

582

super(dbus_python_adapter.CachingBus,

583

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

584

def get_object(self, busname, objectpath):

585

try:

586

return self.object_cache[(busname, objectpath)]

587

except KeyError:

588

new_object = super(

589

dbus_python_adapter.CachingBus,

590

self).get_object(busname, objectpath)

591

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

592

return new_object

593

594

595

def commands_from_options(options):

596

597

commands = list(options.commands)

598

599

def find_cmd(cmd, commands):

600

i = 0

601

for i, c in enumerate(commands):

602

if isinstance(c, cmd):

603

return i

604

return i+1

605

606

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

607

# to occur ahead of command.Remove.

608

index_of_remove = find_cmd(command.Remove, commands)

609

before_remove = commands[:index_of_remove]

610

after_remove = commands[index_of_remove:]

611

cleaned_after = []

612

for cmd in after_remove:

613

if isinstance(cmd, command.Deny):

614

before_remove.append(cmd)

615

else:

616

cleaned_after.append(cmd)

617

if cleaned_after != after_remove:

618

commands = before_remove + cleaned_after

619

620

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

621

# optionally verbosely

622

if not commands:

623

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

624

625

return commands

626

627

628

class command(object):

629

"""A namespace for command classes"""

630

631

class Base(object):

632

"""Abstract base class for commands"""

633

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

634

"""Normal commands should implement run_on_one_client(),

635

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

636

override this run() method instead.

637

"""

638

self.bus = bus

639

for client, properties in clients.items():

640

self.run_on_one_client(client, properties)

641

642

643

class IsEnabled(Base):

644

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

645

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

646

if properties["Enabled"]:

647

sys.exit(0)

648

sys.exit(1)

649

650

651

class Approve(Base):

652

def run_on_one_client(self, client, properties):

653

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

654

655

656

class Deny(Base):

657

def run_on_one_client(self, client, properties):

658

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

659

660

661

class Remove(Base):

662

def run(self, clients, bus):

663

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

664

bus.call_server_method("RemoveClient", clientpath)

665

666

667

class Output(Base):

668

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

669

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

670

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

671

"Fingerprint", "CheckerRunning",

672

"LastEnabled", "ApprovalPending",

673

"ApprovedByDefault", "LastApprovalRequest",

674

"ApprovalDelay", "ApprovalDuration",

675

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

676

"LastCheckerStatus")

677

678

679

class DumpJSON(Output):

680

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

681

data = {properties["Name"]:

682

{key: properties[key]

683

for key in self.all_keywords}

684

for properties in clients.values()}

685

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

686

687

688

class PrintTable(Output):

689

def __init__(self, verbose=False):

690

self.verbose = verbose

691

692

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

693

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

694

"LastCheckedOK")

695

keywords = default_keywords

696

if self.verbose:

697

keywords = self.all_keywords

698

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

699

700

class TableOfClients(object):

701

tableheaders = {

702

"Name": "Name",

703

"Enabled": "Enabled",

704

"Timeout": "Timeout",

705

"LastCheckedOK": "Last Successful Check",

706

"LastApprovalRequest": "Last Approval Request",

707

"Created": "Created",

708

"Interval": "Interval",

709

"Host": "Host",

710

"Fingerprint": "Fingerprint",

711

"KeyID": "Key ID",

712

"CheckerRunning": "Check Is Running",

713

"LastEnabled": "Last Enabled",

714

"ApprovalPending": "Approval Is Pending",

715

"ApprovedByDefault": "Approved By Default",

716

"ApprovalDelay": "Approval Delay",

717

"ApprovalDuration": "Approval Duration",

718

"Checker": "Checker",

719

"ExtendedTimeout": "Extended Timeout",

720

"Expires": "Expires",

721

"LastCheckerStatus": "Last Checker Status",

722

}

723

724

def __init__(self, clients, keywords):

725

self.clients = clients

726

self.keywords = keywords

727

728

def __str__(self):

729

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

730

731

if sys.version_info.major == 2:

732

__unicode__ = __str__

733

def __str__(self):

734

return str(self).encode(

735

locale.getpreferredencoding())

736

737

def rows(self):

738

format_string = self.row_formatting_string()

739

rows = [self.header_line(format_string)]

740

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

741

for client in self.clients)

742

return rows

743

744

def row_formatting_string(self):

745

"Format string used to format table rows"

746

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

747

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

748

*(len(self.string_from_client(client,

749

key))

750

for client in self.clients)),

751

key=key)

752

for key in self.keywords)

753

754

def string_from_client(self, client, key):

755

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

756

757

@classmethod

758

def valuetostring(cls, value, keyword):

759

if isinstance(value, bool):

760

return "Yes" if value else "No"

761

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

762

"ApprovalDuration", "ExtendedTimeout"):

763

return cls.milliseconds_to_string(value)

764

return str(value)

765

766

def header_line(self, format_string):

767

return format_string.format(**self.tableheaders)

768

769

def client_line(self, client, format_string):

770

return format_string.format(

771

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

772

for key in self.keywords})

773

774

@staticmethod

775

def milliseconds_to_string(ms):

776

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

777

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

778

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

779

if td.days else "",

780

hours=td.seconds // 3600,

781

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

782

seconds=td.seconds % 60))

783

784

785

class PropertySetter(Base):

786

"Abstract class for Actions for setting one client property"

787

788

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

789

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

790

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

791

self.value_to_set)

792

793

@property

794

def propname(self):

795

raise NotImplementedError()

796

797

798

class Enable(PropertySetter):

799

propname = "Enabled"

800

value_to_set = True

801

802

803

class Disable(PropertySetter):

804

propname = "Enabled"

805

value_to_set = False

806

807

808

class BumpTimeout(PropertySetter):

809

propname = "LastCheckedOK"

810

value_to_set = ""

811

812

813

class StartChecker(PropertySetter):

814

propname = "CheckerRunning"

815

value_to_set = True

816

817

818

class StopChecker(PropertySetter):

819

propname = "CheckerRunning"

820

value_to_set = False

821

822

823

class ApproveByDefault(PropertySetter):

824

propname = "ApprovedByDefault"

825

value_to_set = True

826

827

828

class DenyByDefault(PropertySetter):

829

propname = "ApprovedByDefault"

830

value_to_set = False

831

832

833

class PropertySetterValue(PropertySetter):

834

"""Abstract class for PropertySetter recieving a value as

835

constructor argument instead of a class attribute."""

836

def __init__(self, value):

837

self.value_to_set = value

838

839

@classmethod

840

def argparse(cls, argtype):

841

def cmdtype(arg):

842

return cls(argtype(arg))

843

return cmdtype

844

845

class SetChecker(PropertySetterValue):

846

propname = "Checker"

847

848

849

class SetHost(PropertySetterValue):

850

propname = "Host"

851

852

853

class SetSecret(PropertySetterValue):

854

propname = "Secret"

855

856

@property

857

def value_to_set(self):

858

return self._vts

859

860

@value_to_set.setter

861

def value_to_set(self, value):

862

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

863

self._vts = value.read()

864

value.close()

865

866

867

class PropertySetterValueMilliseconds(PropertySetterValue):

868

"""Abstract class for PropertySetterValue taking a value

869

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

870

milliseconds."""

871

872

@property

873

def value_to_set(self):

874

return self._vts

875

876

@value_to_set.setter

877

def value_to_set(self, value):

878

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

879

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

880

881

882

class SetTimeout(PropertySetterValueMilliseconds):

883

propname = "Timeout"

884

885

886

class SetExtendedTimeout(PropertySetterValueMilliseconds):

887

propname = "ExtendedTimeout"

888

889

890

class SetInterval(PropertySetterValueMilliseconds):

891

propname = "Interval"

892

893

894

class SetApprovalDelay(PropertySetterValueMilliseconds):

895

propname = "ApprovalDelay"

896

897

898

class SetApprovalDuration(PropertySetterValueMilliseconds):

899

propname = "ApprovalDuration"

900

901

902

903

class TestCaseWithAssertLogs(unittest.TestCase):

904

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

905

906

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

907

@contextlib.contextmanager

908

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

909

capturing_handler = self.CapturingLevelHandler(level)

910

old_level = logger.level

911

old_propagate = logger.propagate

912

logger.addHandler(capturing_handler)

913

logger.setLevel(level)

914

logger.propagate = False

915

try:

916

yield capturing_handler.watcher

917

finally:

918

logger.propagate = old_propagate

919

logger.removeHandler(capturing_handler)

920

logger.setLevel(old_level)

921

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

922

0)

923

924

class CapturingLevelHandler(logging.Handler):

925

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

926

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

927

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

928

def emit(self, record):

929

self.watcher.records.append(record)

930

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

931

932

LoggingWatcher = collections.namedtuple("LoggingWatcher",

933

("records",

934

"output"))

935

936

937

class Unique(object):

938

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

939

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

940

941

942

class Test_string_to_delta(TestCaseWithAssertLogs):

943

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

944

# rfc3339_duration_to_delta() already has more comprehensive

945

# tests, which are run by doctest.

946

947

def test_rfc3339_zero_seconds(self):

948

self.assertEqual(datetime.timedelta(),

949

string_to_delta("PT0S"))

950

951

def test_rfc3339_zero_days(self):

952

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

953

954

def test_rfc3339_one_second(self):

955

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

956

string_to_delta("PT1S"))

957

958

def test_rfc3339_two_hours(self):

959

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

960

string_to_delta("PT2H"))

961

962

def test_falls_back_to_pre_1_6_1_with_warning(self):

963

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

964

value = string_to_delta("2h")

965

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

966

967

968

class Test_check_option_syntax(unittest.TestCase):

969

def setUp(self):

970

self.parser = argparse.ArgumentParser()

971

add_command_line_options(self.parser)

972

973

def test_actions_requires_client_or_all(self):

974

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

975

args = self.actionargs(action, value)

976

with self.assertParseError():

977

self.parse_args(args)

978

979

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

980

# check_option_syntax()

981

actions = {

982

"--enable": None,

983

"--disable": None,

984

"--bump-timeout": None,

985

"--start-checker": None,

986

"--stop-checker": None,

987

"--is-enabled": None,

988

"--remove": None,

989

"--checker": "x",

990

"--timeout": "PT0S",

991

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

992

"--interval": "PT0S",

993

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

994

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

995

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

996

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

997

"--host": "hostname",

998

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

999

"--approve": None,

1000

"--deny": None,

1001

}

1002

1003

@staticmethod

1004

def actionargs(action, value, *args):

1005

if value is not None:

1006

return [action, value] + list(args)

1007

else:

1008

return [action] + list(args)

1009

1010

@contextlib.contextmanager

1011

def assertParseError(self):

1012

with self.assertRaises(SystemExit) as e:

1013

with self.redirect_stderr_to_devnull():

1014

yield

1015

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

1016

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

1017

# /argparse.html#exiting-methods

1018

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

1019

1020

def parse_args(self, args):

1021

options = self.parser.parse_args(args)

1022

check_option_syntax(self.parser, options)

1023

1024

@staticmethod

1025

@contextlib.contextmanager

1026

def redirect_stderr_to_devnull():

1027

old_stderr = sys.stderr

1028

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

1029

sys.stderr = null

1030

try:

1031

yield

1032

finally:

1033

sys.stderr = old_stderr

1034

1035

def check_option_syntax(self, options):

1036

check_option_syntax(self.parser, options)

1037

1038

def test_actions_all_conflicts_with_verbose(self):

1039

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

1040

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

1041

"--verbose")

1042

with self.assertParseError():

1043

self.parse_args(args)

1044

1045

def test_actions_with_client_conflicts_with_verbose(self):

1046

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

1047

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

1048

"client")

1049

with self.assertParseError():

1050

self.parse_args(args)

1051

1052

def test_dump_json_conflicts_with_verbose(self):

1053

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

1054

with self.assertParseError():

1055

self.parse_args(args)

1056

1057

def test_dump_json_conflicts_with_action(self):

1058

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

1059

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

1060

with self.assertParseError():

1061

self.parse_args(args)

1062

1063

def test_all_can_not_be_alone(self):

1064

args = ["--all"]

1065

with self.assertParseError():

1066

self.parse_args(args)

1067

1068

def test_all_is_ok_with_any_action(self):

1069

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

1070

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

1071

self.parse_args(args)

1072

1073

def test_any_action_is_ok_with_one_client(self):

1074

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

1075

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

1076

self.parse_args(args)

1077

1078

def test_one_client_with_all_actions_except_is_enabled(self):

1079

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

1080

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

1081

continue

1082

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

1083

self.parse_args(args)

1084

1085

def test_two_clients_with_all_actions_except_is_enabled(self):

1086

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

1087

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

1088

continue

1089

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

1090

"client2")

1091

self.parse_args(args)

1092

1093

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1094

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

1095

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

1096

continue

1097

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

1098

"client2")

1099

self.parse_args(args)

1100

1101

def test_is_enabled_fails_without_client(self):

1102

args = ["--is-enabled"]

1103

with self.assertParseError():

1104

self.parse_args(args)

1105

1106

def test_is_enabled_fails_with_two_clients(self):

1107

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

1108

with self.assertParseError():

1109

self.parse_args(args)

1110

1111

def test_remove_can_only_be_combined_with_action_deny(self):

1112

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

1113

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

1114

continue

1115

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

1116

"--remove")

1117

with self.assertParseError():

1118

self.parse_args(args)

1119

1120

1121

class Test_dbus_exceptions(unittest.TestCase):

1122

1123

def test_dbus_ConnectFailed_is_Error(self):

1124

with self.assertRaises(dbus.Error):

1125

raise dbus.ConnectFailed()

1126

1127

1128

class Test_dbus_MandosBus(unittest.TestCase):

1129

1130

class MockMandosBus(dbus.MandosBus):

1131

def __init__(self):

1132

self._name = "se.recompile.Mandos"

1133

self._server_path = "/"

1134

self._server_interface = "se.recompile.Mandos"

1135

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

1136

self.calls = []

1137

self.call_method_return = Unique()

1138

1139

def call_method(self, methodname, busname, objectpath,

1140

interface, *args):

1141

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

1142

interface, args))

1143

return self.call_method_return

1144

1145

def setUp(self):

1146

self.bus = self.MockMandosBus()

1147

1148

def test_set_client_property(self):

1149

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

1150

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

1151

"org.freedesktop.DBus.Properties",

1152

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

1153

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

1154

1155

def test_call_client_method(self):

1156

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

1157

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

1158

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

1159

self.bus._client_interface, ())

1160

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

1161

1162

def test_call_client_method_with_args(self):

1163

args = (Unique(), Unique())

1164

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

1165

*args)

1166

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

1167

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

1168

self.bus._client_interface,

1169

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

1170

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

1171

1172

def test_get_clients_and_properties(self):

1173

managed_objects = {

1174

"objectpath": {

1175

self.bus._client_interface: {

1176

"key": "value",

1177

"bool": True,

1178

},

1179

"irrelevant_interface": {

1180

"key": "othervalue",

1181

"bool": False,

1182

},

1183

},

1184

"other_objectpath": {

1185

"other_irrelevant_interface": {

1186

"key": "value 3",

1187

"bool": None,

1188

},

1189

},

1190

}

1191

expected_clients_and_properties = {

1192

"objectpath": {

1193

"key": "value",

1194

"bool": True,

1195

}

1196

}

1197

self.bus.call_method_return = managed_objects

1198

ret = self.bus.get_clients_and_properties()

1199

self.assertDictEqual(expected_clients_and_properties, ret)

1200

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

1201

self.bus._server_path,

1202

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

1203

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

1204

1205

def test_call_server_method(self):

1206

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

1207

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

1208

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

1209

self.bus._server_path,

1210

self.bus._server_interface, ())

1211

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

1212

1213

def test_call_server_method_with_args(self):

1214

args = (Unique(), Unique())

1215

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

1216

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

1217

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

1218

self.bus._server_path,

1219

self.bus._server_interface,

1220

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

1221

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

1222

1223

1224

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1225

1226

def MockDBusPython_func(self, func):

1227

class mock_dbus_python(object):

1228

"""mock dbus-python module"""

1229

class exceptions(object):

1230

"""Pseudo-namespace"""

1231

class DBusException(Exception):

1232

pass

1233

class SystemBus(object):

1234

@staticmethod

1235

def get_object(busname, objectpath):

1236

DBusObject = collections.namedtuple(

1237

"DBusObject", ("methodname",))

1238

def method(*args, **kwargs):

1239

self.assertEqual({"dbus_interface":

1240

"interface"},

1241

kwargs)

1242

return func(*args)

1243

return DBusObject(methodname=method)

1244

class Boolean(object):

1245

def __init__(self, value):

1246

self.value = bool(value)

1247

def __bool__(self):

1248

return self.value

1249

if sys.version_info.major == 2:

1250

__nonzero__ = __bool__

1251

class ObjectPath(str):

1252

pass

1253

class Dictionary(dict):

1254

pass

1255

return mock_dbus_python

1256

1257

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

1258

interface, *args):

1259

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

1260

return bus.call_method(methodname, busname, objectpath,

1261

interface, *args)

1262

1263

def test_call_method_returns(self):

1264

expected_method_return = Unique()

1265

method_args = (Unique(), Unique())

1266

def func(*args):

1267

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

1268

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

1269

self.assertIs(marg, arg)

1270

return expected_method_return

1271

mock_dbus_python = self.MockDBusPython_func(func)

1272

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1273

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

1274

"objectpath", "interface",

1275

*method_args)

1276

self.assertIs(ret, expected_method_return)

1277

1278

def test_call_method_filters_bool_true(self):

1279

def func():

1280

return method_return

1281

mock_dbus_python = self.MockDBusPython_func(func)

1282

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1283

method_return = mock_dbus_python.Boolean(True)

1284

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

1285

"objectpath", "interface")

1286

self.assertTrue(ret)

1287

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1288

1289

def test_call_method_filters_bool_false(self):

1290

def func():

1291

return method_return

1292

mock_dbus_python = self.MockDBusPython_func(func)

1293

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1294

method_return = mock_dbus_python.Boolean(False)

1295

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

1296

"objectpath", "interface")

1297

self.assertFalse(ret)

1298

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1299

1300

def test_call_method_filters_objectpath(self):

1301

def func():

1302

return method_return

1303

mock_dbus_python = self.MockDBusPython_func(func)

1304

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1305

method_return = mock_dbus_python.ObjectPath("objectpath")

1306

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

1307

"objectpath", "interface")

1308

self.assertEqual("objectpath", ret)

1309

self.assertIsNot("objectpath", ret)

1310

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1311

1312

def test_call_method_filters_booleans_in_dict(self):

1313

def func():

1314

return method_return

1315

mock_dbus_python = self.MockDBusPython_func(func)

1316

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1317

method_return = mock_dbus_python.Dictionary(

1318

{mock_dbus_python.Boolean(True):

1319

mock_dbus_python.Boolean(False),

1320

mock_dbus_python.Boolean(False):

1321

mock_dbus_python.Boolean(True)})

1322

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

1323

"objectpath", "interface")

1324

expected_method_return = {True: False,

1325

False: True}

1326

self.assertEqual(expected_method_return, ret)

1327

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1328

1329

def test_call_method_filters_objectpaths_in_dict(self):

1330

def func():

1331

return method_return

1332

mock_dbus_python = self.MockDBusPython_func(func)

1333

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1334

method_return = mock_dbus_python.Dictionary(

1335

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1336

mock_dbus_python.ObjectPath("objectpath_value_1"),

1337

mock_dbus_python.ObjectPath("objectpath_key_2"):

1338

mock_dbus_python.ObjectPath("objectpath_value_2")})

1339

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

1340

"objectpath", "interface")

1341

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

1342

for key, value in

1343

method_return.items()}

1344

self.assertEqual(expected_method_return, ret)

1345

self.assertIsInstance(ret, dict)

1346

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1347

1348

def test_call_method_filters_dict_in_dict(self):

1349

def func():

1350

return method_return

1351

mock_dbus_python = self.MockDBusPython_func(func)

1352

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1353

method_return = mock_dbus_python.Dictionary(

1354

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

1355

"key12": "value12"}),

1356

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

1357

"key22": "value22"})})

1358

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

1359

"objectpath", "interface")

1360

expected_method_return = {

1361

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

1362

"key12": "value12"},

1363

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

1364

"key22": "value22"},

1365

}

1366

self.assertEqual(expected_method_return, ret)

1367

self.assertIsInstance(ret, dict)

1368

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1369

for key, value in ret.items():

1370

self.assertIsInstance(value, dict)

1371

self.assertEqual(expected_method_return[key], value)

1372

self.assertNotIsInstance(value,

1373

mock_dbus_python.Dictionary)

1374

1375

def test_call_method_filters_dict_three_deep(self):

1376

def func():

1377

return method_return

1378

mock_dbus_python = self.MockDBusPython_func(func)

1379

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1380

method_return = mock_dbus_python.Dictionary(

1381

{"key1":

1382

mock_dbus_python.Dictionary(

1383

{"key2":

1384

mock_dbus_python.Dictionary(

1385

{"key3":

1386

mock_dbus_python.Boolean(True),

1387

}),

1388

}),

1389

})

1390

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

1391

"objectpath", "interface")

1392

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

1393

self.assertEqual(expected_method_return, ret)

1394

self.assertIsInstance(ret, dict)

1395

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1396

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

1397

self.assertNotIsInstance(ret["key1"],

1398

mock_dbus_python.Dictionary)

1399

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

1400

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

1401

mock_dbus_python.Dictionary)

1402

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

1403

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

1404

mock_dbus_python.Boolean)

1405

1406

def test_call_method_handles_exception(self):

1407

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

1408

1409

def func():

1410

dbus_logger.error("Test")

1411

raise mock_dbus_python.exceptions.DBusException()

1412

1413

mock_dbus_python = self.MockDBusPython_func(func)

1414

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1415

1416

class CountingHandler(logging.Handler):

1417

count = 0

1418

def emit(self, record):

1419

self.count += 1

1420

1421

counting_handler = CountingHandler()

1422

1423

dbus_logger.addHandler(counting_handler)

1424

1425

try:

1426

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

1427

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

1428

"objectpath", "interface")

1429

finally:

1430

dbus_logger.removeFilter(counting_handler)

1431

1432

self.assertNotIsInstance(e, dbus.ConnectFailed)

1433

1434

# Make sure the dbus logger was suppressed

1435

self.assertEqual(0, counting_handler.count)

1436

1437

def test_get_object_converts_to_correct_exception(self):

1438

bus = dbus_python_adapter.SystemBus(

1439

self.fake_dbus_python_raises_exception_on_connect)

1440

with self.assertRaises(dbus.ConnectFailed):

1441

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

1442

"objectpath", "interface")

1443

1444

class fake_dbus_python_raises_exception_on_connect(object):

1445

"""fake dbus-python module"""

1446

class exceptions(object):

1447

"""Pseudo-namespace"""

1448

class DBusException(Exception):

1449

pass

1450

1451

@classmethod

1452

def SystemBus(cls):

1453

def get_object(busname, objectpath):

1454

raise cls.exceptions.DBusException()

1455

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

1456

return Bus(get_object=get_object)

1457

1458

1459

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1460

class mock_dbus_python(object):

1461

"""mock dbus-python modules"""

1462

class SystemBus(object):

1463

@staticmethod

1464

def get_object(busname, objectpath):

1465

return Unique()

1466

1467

def setUp(self):

1468

self.bus = dbus_python_adapter.CachingBus(

1469

self.mock_dbus_python)

1470

1471

def test_returns_distinct_objectpaths(self):

1472

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

1473

self.assertIsInstance(obj1, Unique)

1474

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

1475

self.assertIsInstance(obj2, Unique)

1476

self.assertIsNot(obj1, obj2)

1477

1478

def test_returns_distinct_busnames(self):

1479

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

1480

self.assertIsInstance(obj1, Unique)

1481

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

1482

self.assertIsInstance(obj2, Unique)

1483

self.assertIsNot(obj1, obj2)

1484

1485

def test_returns_distinct_both(self):

1486

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

1487

self.assertIsInstance(obj1, Unique)

1488

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

1489

self.assertIsInstance(obj2, Unique)

1490

self.assertIsNot(obj1, obj2)

1491

1492

def test_returns_same(self):

1493

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

1494

self.assertIsInstance(obj1, Unique)

1495

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

1496

self.assertIsInstance(obj2, Unique)

1497

self.assertIs(obj1, obj2)

1498

1499

def test_returns_same_old(self):

1500

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

1501

self.assertIsInstance(obj1, Unique)

1502

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

1503

self.assertIsInstance(obj2, Unique)

1504

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

1505

self.assertIsInstance(obj1b, Unique)

1506

self.assertIsNot(obj1, obj2)

1507

self.assertIsNot(obj2, obj1b)

1508

self.assertIs(obj1, obj1b)

1509

1510

1511

class Test_commands_from_options(unittest.TestCase):

1512

1513

def setUp(self):

1514

self.parser = argparse.ArgumentParser()

1515

add_command_line_options(self.parser)

1516

1517

def test_is_enabled(self):

1518

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

1519

command.IsEnabled)

1520

1521

def assert_command_from_args(self, args, command_cls,

1522

**cmd_attrs):

1523

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

1524

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

1525

options = self.parser.parse_args(args)

1526

check_option_syntax(self.parser, options)

1527

commands = commands_from_options(options)

1528

self.assertEqual(1, len(commands))

1529

command = commands[0]

1530

self.assertIsInstance(command, command_cls)

1531

for key, value in cmd_attrs.items():

1532

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

1533

1534

def test_is_enabled_short(self):

1535

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

1536

command.IsEnabled)

1537

1538

def test_approve(self):

1539

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

1540

command.Approve)

1541

1542

def test_approve_short(self):

1543

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

1544

command.Approve)

1545

1546

def test_deny(self):

1547

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

1548

command.Deny)

1549

1550

def test_deny_short(self):

1551

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

1552

1553

def test_remove(self):

1554

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

1555

command.Remove)

1556

1557

def test_deny_before_remove(self):

1558

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

1559

"client"])

1560

check_option_syntax(self.parser, options)

1561

commands = commands_from_options(options)

1562

self.assertEqual(2, len(commands))

1563

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

1564

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

1565

1566

def test_deny_before_remove_reversed(self):

1567

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

1568

"--all"])

1569

check_option_syntax(self.parser, options)

1570

commands = commands_from_options(options)

1571

self.assertEqual(2, len(commands))

1572

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

1573

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

1574

1575

def test_remove_short(self):

1576

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

1577

command.Remove)

1578

1579

def test_dump_json(self):

1580

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

1581

command.DumpJSON)

1582

1583

def test_enable(self):

1584

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

1585

command.Enable)

1586

1587

def test_enable_short(self):

1588

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

1589

command.Enable)

1590

1591

def test_disable(self):

1592

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

1593

command.Disable)

1594

1595

def test_disable_short(self):

1596

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

1597

command.Disable)

1598

1599

def test_bump_timeout(self):

1600

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

1601

command.BumpTimeout)

1602

1603

def test_bump_timeout_short(self):

1604

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

1605

command.BumpTimeout)

1606

1607

def test_start_checker(self):

1608

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

1609

command.StartChecker)

1610

1611

def test_stop_checker(self):

1612

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

1613

command.StopChecker)

1614

1615

def test_approve_by_default(self):

1616

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

1617

"client"],

1618

command.ApproveByDefault)

1619

1620

def test_deny_by_default(self):

1621

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

1622

command.DenyByDefault)

1623

1624

def test_checker(self):

1625

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

1626

command.SetChecker,

1627

value_to_set=":")

1628

1629

def test_checker_empty(self):

1630

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

1631

command.SetChecker,

1632

value_to_set="")

1633

1634

def test_checker_short(self):

1635

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

1636

command.SetChecker,

1637

value_to_set=":")

1638

1639

def test_host(self):

1640

self.assert_command_from_args(

1641

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

1642

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

1643

1644

def test_host_short(self):

1645

self.assert_command_from_args(

1646

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

1647

value_to_set="client.example.org")

1648

1649

def test_secret_devnull(self):

1650

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

1651

"client"], command.SetSecret,

1652

value_to_set=b"")

1653

1654

def test_secret_tempfile(self):

1655

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

1656

value = b"secret\0xyzzy\nbar"

1657

f.write(value)

1658

f.seek(0)

1659

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

1660

"client"],

1661

command.SetSecret,

1662

value_to_set=value)

1663

1664

def test_secret_devnull_short(self):

1665

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

1666

"client"], command.SetSecret,

1667

value_to_set=b"")

1668

1669

def test_secret_tempfile_short(self):

1670

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

1671

value = b"secret\0xyzzy\nbar"

1672

f.write(value)

1673

f.seek(0)

1674

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

1675

command.SetSecret,

1676

value_to_set=value)

1677

1678

def test_timeout(self):

1679

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

1680

command.SetTimeout,

1681

value_to_set=300000)

1682

1683

def test_timeout_short(self):

1684

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

1685

command.SetTimeout,

1686

value_to_set=300000)

1687

1688

def test_extended_timeout(self):

1689

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

1690

"client"],

1691

command.SetExtendedTimeout,

1692

value_to_set=900000)

1693

1694

def test_interval(self):

1695

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

1696

"client"], command.SetInterval,

1697

value_to_set=120000)

1698

1699

def test_interval_short(self):

1700

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

1701

command.SetInterval,

1702

value_to_set=120000)

1703

1704

def test_approval_delay(self):

1705

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

1706

"client"],

1707

command.SetApprovalDelay,

1708

value_to_set=30000)

1709

1710

def test_approval_duration(self):

1711

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

1712

"client"],

1713

command.SetApprovalDuration,

1714

value_to_set=1000)

1715

1716

def test_print_table(self):

1717

self.assert_command_from_args([], command.PrintTable,

1718

verbose=False)

1719

1720

def test_print_table_verbose(self):

1721

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

1722

command.PrintTable,

1723

verbose=True)

1724

1725

def test_print_table_verbose_short(self):

1726

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

1727

verbose=True)

1728

1729

1730

class TestCommand(unittest.TestCase):

1731

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

1732

1733

class FakeMandosBus(dbus.MandosBus):

1734

def __init__(self, testcase):

1735

self.client_properties = {

1736

"Name": "foo",

1737

"KeyID": ("92ed150794387c03ce684574b1139a65"

1738

"94a34f895daaaf09fd8ea90a27cddb12"),

1739

"Secret": b"secret",

1740

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

1741

"Enabled": True,

1742

"Timeout": 300000,

1743

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

1744

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

1745

"Interval": 120000,

1746

"Fingerprint": ("778827225BA7DE539C5A"

1747

"7CFA59CFF7CDBD9A5920"),

1748

"CheckerRunning": False,

1749

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

1750

"ApprovalPending": False,

1751

"ApprovedByDefault": True,

1752

"LastApprovalRequest": "",

1753

"ApprovalDelay": 0,

1754

"ApprovalDuration": 1000,

1755

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

1756

"ExtendedTimeout": 900000,

1757

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

1758

"LastCheckerStatus": 0,

1759

}

1760

self.other_client_properties = {

1761

"Name": "barbar",

1762

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1763

"6ab612cff5ad227247e46c2b020f441c"),

1764

"Secret": b"secretbar",

1765

"Host": "192.0.2.3",

1766

"Enabled": True,

1767

"Timeout": 300000,

1768

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

1769

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

1770

"Interval": 120000,

1771

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1772

"F547B3A107558FCA3A27"),

1773

"CheckerRunning": True,

1774

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

1775

"ApprovalPending": False,

1776

"ApprovedByDefault": False,

1777

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

1778

"ApprovalDelay": 30000,

1779

"ApprovalDuration": 93785000,

1780

"Checker": ":",

1781

"ExtendedTimeout": 900000,

1782

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

1783

"LastCheckerStatus": -2,

1784

}

1785

self.clients = collections.OrderedDict(

1786

[

1787

("client_objectpath", self.client_properties),

1788

("other_client_objectpath",

1789

self.other_client_properties),

1790

])

1791

self.one_client = {"client_objectpath":

1792

self.client_properties}

1793

self.testcase = testcase

1794

self.calls = []

1795

1796

def call_method(self, methodname, busname, objectpath,

1797

interface, *args):

1798

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

1799

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

1800

interface, args))

1801

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

1802

if methodname == "Set":

1803

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

1804

interface, key, value = args

1805

self.testcase.assertEqual(

1806

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

1807

self.clients[objectpath][key] = value

1808

return

1809

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

1810

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

1811

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

1812

clientpath = args[0]

1813

del self.clients[clientpath]

1814

return

1815

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

1816

if methodname == "Approve":

1817

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

1818

return

1819

raise ValueError()

1820

1821

def setUp(self):

1822

self.bus = self.FakeMandosBus(self)

1823

1824

1825

class TestBaseCommands(TestCommand):

1826

1827

def test_IsEnabled_exits_successfully(self):

1828

with self.assertRaises(SystemExit) as e:

1829

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

1830

if e.exception.code is not None:

1831

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

1832

else:

1833

self.assertIsNone(e.exception.code)

1834

1835

def test_IsEnabled_exits_with_failure(self):

1836

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

1837

with self.assertRaises(SystemExit) as e:

1838

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

1839

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

1840

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

1841

else:

1842

self.assertIsNotNone(e.exception.code)

1843

1844

def test_Approve(self):

1845

busname = "se.recompile.Mandos"

1846

client_interface = "se.recompile.Mandos.Client"

1847

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

1848

for clientpath in self.bus.clients:

1849

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

1850

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

1851

1852

def test_Deny(self):

1853

busname = "se.recompile.Mandos"

1854

client_interface = "se.recompile.Mandos.Client"

1855

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

1856

for clientpath in self.bus.clients:

1857

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

1858

client_interface, (False,)),

1859

self.bus.calls)

1860

1861

def test_Remove(self):

1862

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

1863

for clientpath in self.bus.clients:

1864

self.assertIn(("RemoveClient", dbus_busname,

1865

dbus_server_path, dbus_server_interface,

1866

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

1867

1868

expected_json = {

1869

"foo": {

1870

"Name": "foo",

1871

"KeyID": ("92ed150794387c03ce684574b1139a65"

1872

"94a34f895daaaf09fd8ea90a27cddb12"),

1873

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

1874

"Enabled": True,

1875

"Timeout": 300000,

1876

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

1877

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

1878

"Interval": 120000,

1879

"Fingerprint": ("778827225BA7DE539C5A"

1880

"7CFA59CFF7CDBD9A5920"),

1881

"CheckerRunning": False,

1882

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

1883

"ApprovalPending": False,

1884

"ApprovedByDefault": True,

1885

"LastApprovalRequest": "",

1886

"ApprovalDelay": 0,

1887

"ApprovalDuration": 1000,

1888

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

1889

"ExtendedTimeout": 900000,

1890

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

1891

"LastCheckerStatus": 0,

1892

},

1893

"barbar": {

1894

"Name": "barbar",

1895

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1896

"6ab612cff5ad227247e46c2b020f441c"),

1897

"Host": "192.0.2.3",

1898

"Enabled": True,

1899

"Timeout": 300000,

1900

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

1901

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

1902

"Interval": 120000,

1903

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1904

"F547B3A107558FCA3A27"),

1905

"CheckerRunning": True,

1906

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

1907

"ApprovalPending": False,

1908

"ApprovedByDefault": False,

1909

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

1910

"ApprovalDelay": 30000,

1911

"ApprovalDuration": 93785000,

1912

"Checker": ":",

1913

"ExtendedTimeout": 900000,

1914

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

1915

"LastCheckerStatus": -2,

1916

},

1917

}

1918

1919

def test_DumpJSON_normal(self):

1920

with self.capture_stdout_to_buffer() as buffer:

1921

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

1922

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

1923

self.assertDictEqual(self.expected_json, json_data)

1924

1925

@staticmethod

1926

@contextlib.contextmanager

1927

def capture_stdout_to_buffer():

1928

capture_buffer = io.StringIO()

1929

old_stdout = sys.stdout

1930

sys.stdout = capture_buffer

1931

try:

1932

yield capture_buffer

1933

finally:

1934

sys.stdout = old_stdout

1935

1936

def test_DumpJSON_one_client(self):

1937

with self.capture_stdout_to_buffer() as buffer:

1938

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

1939

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

1940

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

1941

self.assertDictEqual(expected_json, json_data)

1942

1943

def test_PrintTable_normal(self):

1944

with self.capture_stdout_to_buffer() as buffer:

1945

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

1946

expected_output = "\n".join((

1947

"Name Enabled Timeout Last Successful Check",

1948

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

1949

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

1950

)) + "\n"

1951

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

1952

1953

def test_PrintTable_verbose(self):

1954

with self.capture_stdout_to_buffer() as buffer:

1955

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

1956

columns = (

1957

(

1958

"Name ",

1959

"foo ",

1960

"barbar ",

1961

),(

1962

"Enabled ",

1963

"Yes ",

1964

"Yes ",

1965

),(

1966

"Timeout ",

1967

"00:05:00 ",

1968

"00:05:00 ",

1969

),(

1970

"Last Successful Check ",

1971

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

1972

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

1973

),(

1974

"Created ",

1975

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

1976

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

1977

),(

1978

"Interval ",

1979

"00:02:00 ",

1980

"00:02:00 ",

1981

),(

1982

"Host ",

1983

"foo.example.org ",

1984

"192.0.2.3 ",

1985

),(

1986

("Key ID "

1987

" "),

1988

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1989

"ea90a27cddb12 "),

1990

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1991

"46c2b020f441c "),

1992

),(

1993

"Fingerprint ",

1994

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1995

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1996

),(

1997

"Check Is Running ",

1998

"No ",

1999

"Yes ",

2000

),(

2001

"Last Enabled ",

2002

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

2003

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

2004

),(

2005

"Approval Is Pending ",

2006

"No ",

2007

"No ",

2008

),(

2009

"Approved By Default ",

2010

"Yes ",

2011

"No ",

2012

),(

2013

"Last Approval Request ",

2014

" ",

2015

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

2016

),(

2017

"Approval Delay ",

2018

"00:00:00 ",

2019

"00:00:30 ",

2020

),(

2021

"Approval Duration ",

2022

"00:00:01 ",

2023

"1T02:03:05 ",

2024

),(

2025

"Checker ",

2026

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

2027

": ",

2028

),(

2029

"Extended Timeout ",

2030

"00:15:00 ",

2031

"00:15:00 ",

2032

),(

2033

"Expires ",

2034

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

2035

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

2036

),(

2037

"Last Checker Status",

2038

"0 ",

2039

"-2 ",

2040

)

2041

)

2042

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

2043

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

2044

for rows in columns)

2045

for line in range(num_lines))

2046

+ "\n")

2047

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

2048

2049

def test_PrintTable_one_client(self):

2050

with self.capture_stdout_to_buffer() as buffer:

2051

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

2052

expected_output = "\n".join((

2053

"Name Enabled Timeout Last Successful Check",

2054

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

2055

)) + "\n"

2056

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

2057

2058

2059

class TestPropertySetterCmd(TestCommand):

2060

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

2061

2062

def runTest(self):

2063

if not hasattr(self, "command"):

2064

return # Abstract TestCase class

2065

2066

if hasattr(self, "values_to_set"):

2067

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

2068

values_to_get = getattr(self, "values_to_get",

2069

self.values_to_set)

2070

else:

2071

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

2072

values_to_get = self.values_to_get

2073

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

2074

for clientpath in self.bus.clients:

2075

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

2076

Unique())

2077

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

2078

for clientpath in self.bus.clients:

2079

value = (self.bus.clients[clientpath]

2080

[self.propname])

2081

self.assertNotIsInstance(value, Unique)

2082

self.assertEqual(value_to_get, value)

2083

2084

2085

class TestEnableCmd(TestPropertySetterCmd):

2086

command = command.Enable

2087

propname = "Enabled"

2088

values_to_get = [True]

2089

2090

2091

class TestDisableCmd(TestPropertySetterCmd):

2092

command = command.Disable

2093

propname = "Enabled"

2094

values_to_get = [False]

2095

2096

2097

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2098

command = command.BumpTimeout

2099

propname = "LastCheckedOK"

2100

values_to_get = [""]

2101

2102

2103

class TestStartCheckerCmd(TestPropertySetterCmd):

2104

command = command.StartChecker

2105

propname = "CheckerRunning"

2106

values_to_get = [True]

2107

2108

2109

class TestStopCheckerCmd(TestPropertySetterCmd):

2110

command = command.StopChecker

2111

propname = "CheckerRunning"

2112

values_to_get = [False]

2113

2114

2115

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2116

command = command.ApproveByDefault

2117

propname = "ApprovedByDefault"

2118

values_to_get = [True]

2119

2120

2121

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2122

command = command.DenyByDefault

2123

propname = "ApprovedByDefault"

2124

values_to_get = [False]

2125

2126

2127

class TestSetCheckerCmd(TestPropertySetterCmd):

2128

command = command.SetChecker

2129

propname = "Checker"

2130

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

2131

2132

2133

class TestSetHostCmd(TestPropertySetterCmd):

2134

command = command.SetHost

2135

propname = "Host"

2136

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

2137

2138

2139

class TestSetSecretCmd(TestPropertySetterCmd):

2140

command = command.SetSecret

2141

propname = "Secret"

2142

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

2143

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

2144

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

2145

2146

2147

class TestSetTimeoutCmd(TestPropertySetterCmd):

2148

command = command.SetTimeout

2149

propname = "Timeout"

2150

values_to_set = [datetime.timedelta(),

2151

datetime.timedelta(minutes=5),

2152

datetime.timedelta(seconds=1),

2153

datetime.timedelta(weeks=1),

2154

datetime.timedelta(weeks=52)]

2155

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

2156

2157

2158

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2159

command = command.SetExtendedTimeout

2160

propname = "ExtendedTimeout"

2161

values_to_set = [datetime.timedelta(),

2162

datetime.timedelta(minutes=5),

2163

datetime.timedelta(seconds=1),

2164

datetime.timedelta(weeks=1),

2165

datetime.timedelta(weeks=52)]

2166

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

2167

2168

2169

class TestSetIntervalCmd(TestPropertySetterCmd):

2170

command = command.SetInterval

2171

propname = "Interval"

2172

values_to_set = [datetime.timedelta(),

2173

datetime.timedelta(minutes=5),

2174

datetime.timedelta(seconds=1),

2175

datetime.timedelta(weeks=1),

2176

datetime.timedelta(weeks=52)]

2177

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

2178

2179

2180

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2181

command = command.SetApprovalDelay

2182

propname = "ApprovalDelay"

2183

values_to_set = [datetime.timedelta(),

2184

datetime.timedelta(minutes=5),

2185

datetime.timedelta(seconds=1),

2186

datetime.timedelta(weeks=1),

2187

datetime.timedelta(weeks=52)]

2188

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

2189

2190

2191

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2192

command = command.SetApprovalDuration

2193

propname = "ApprovalDuration"

2194

values_to_set = [datetime.timedelta(),

2195

datetime.timedelta(minutes=5),

2196

datetime.timedelta(seconds=1),

2197

datetime.timedelta(weeks=1),

2198

datetime.timedelta(weeks=52)]

2199

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

2200

2201

2202

2203

def should_only_run_tests():

2204

parser = argparse.ArgumentParser(add_help=False)

2205

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

2206

args, unknown_args = parser.parse_known_args()

2207

run_tests = args.check

2208

if run_tests:

2209

# Remove --check argument from sys.argv

2210

sys.argv[1:] = unknown_args

2211

return run_tests

2212

2213

# Add all tests from doctest strings

2214

def load_tests(loader, tests, none):

2215

import doctest

2216

tests.addTests(doctest.DocTestSuite())

2217

return tests

2218

2219

if __name__ == "__main__":

2220

try:

2221

if should_only_run_tests():

2222

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

2223

unittest.main()

2224

else:

2225

main()

2226

finally:

2227

logging.shutdown()