/mandos/trunk : revision 183

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", action="store_true",

126

help="Dump client data in JSON format")

127

enable_disable = parser.add_mutually_exclusive_group()

128

enable_disable.add_argument("-e", "--enable", action="store_true",

129

help="Enable client")

130

enable_disable.add_argument("-d", "--disable",

131

action="store_true",

132

help="disable client")

133

parser.add_argument("-b", "--bump-timeout", action="store_true",

134

help="Bump timeout for client")

135

start_stop_checker = parser.add_mutually_exclusive_group()

136

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

137

action="store_true",

138

help="Start checker for client")

139

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

140

action="store_true",

141

help="Stop checker for client")

142

parser.add_argument("-V", "--is-enabled", action="store_true",

143

help="Check if client is enabled")

144

parser.add_argument("-r", "--remove", action="store_true",

145

help="Remove client")

146

parser.add_argument("-c", "--checker",

147

help="Set checker command for client")

148

parser.add_argument("-t", "--timeout", type=string_to_delta,

149

help="Set timeout for client")

150

parser.add_argument("--extended-timeout", type=string_to_delta,

151

help="Set extended timeout for client")

152

parser.add_argument("-i", "--interval", type=string_to_delta,

153

help="Set checker interval for client")

154

approve_deny_default = parser.add_mutually_exclusive_group()

155

approve_deny_default.add_argument(

156

"--approve-by-default", action="store_true",

157

default=None, dest="approved_by_default",

158

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

159

approve_deny_default.add_argument(

160

"--deny-by-default", action="store_false",

161

dest="approved_by_default",

162

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

163

parser.add_argument("--approval-delay", type=string_to_delta,

164

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

165

parser.add_argument("--approval-duration", type=string_to_delta,

166

help="Set duration of one client approval")

167

parser.add_argument("-H", "--host", help="Set host for client")

168

parser.add_argument("-s", "--secret",

169

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

170

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

171

approve_deny = parser.add_mutually_exclusive_group()

172

approve_deny.add_argument(

173

"-A", "--approve", action="store_true",

174

help="Approve any current client request")

175

approve_deny.add_argument("-D", "--deny", action="store_true",

176

help="Deny any current client request")

177

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

178

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

179

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

180

help="Run self-test")

181

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

182

183

184

def string_to_delta(interval):

185

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

186

187

try:

188

return rfc3339_duration_to_delta(interval)

189

except ValueError as e:

190

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

191

' '.join(e.args))

192

return parse_pre_1_6_1_interval(interval)

193

194

195

def rfc3339_duration_to_delta(duration):

196

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

197

198

>>> rfc3339_duration_to_delta("P7D")

199

datetime.timedelta(7)

200

>>> rfc3339_duration_to_delta("PT60S")

201

datetime.timedelta(0, 60)

202

>>> rfc3339_duration_to_delta("PT60M")

203

datetime.timedelta(0, 3600)

204

>>> rfc3339_duration_to_delta("P60M")

205

datetime.timedelta(1680)

206

>>> rfc3339_duration_to_delta("PT24H")

207

datetime.timedelta(1)

208

>>> rfc3339_duration_to_delta("P1W")

209

datetime.timedelta(7)

210

>>> rfc3339_duration_to_delta("PT5M30S")

211

datetime.timedelta(0, 330)

212

>>> rfc3339_duration_to_delta("P1DT3M20S")

213

datetime.timedelta(1, 200)

214

>>> # Can not be empty:

215

>>> rfc3339_duration_to_delta("")

216

Traceback (most recent call last):

217

...

218

ValueError: Invalid RFC 3339 duration: ""

219

>>> # Must start with "P":

220

>>> rfc3339_duration_to_delta("1D")

221

Traceback (most recent call last):

222

...

223

ValueError: Invalid RFC 3339 duration: "1D"

224

>>> # Must use correct order

225

>>> rfc3339_duration_to_delta("PT1S2M")

226

Traceback (most recent call last):

227

...

228

ValueError: Invalid RFC 3339 duration: "PT1S2M"

229

>>> # Time needs time marker

230

>>> rfc3339_duration_to_delta("P1H2S")

231

Traceback (most recent call last):

232

...

233

ValueError: Invalid RFC 3339 duration: "P1H2S"

234

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

235

>>> rfc3339_duration_to_delta("P1D2W")

236

Traceback (most recent call last):

237

...

238

ValueError: Invalid RFC 3339 duration: "P1D2W"

239

>>> rfc3339_duration_to_delta("P2W2H")

240

Traceback (most recent call last):

241

...

242

ValueError: Invalid RFC 3339 duration: "P2W2H"

243

"""

244

245

# Parsing an RFC 3339 duration with regular expressions is not

246

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

247

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

248

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

249

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

250

# avoid excessive use of external libraries.

251

252

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

253

Token = collections.namedtuple("Token", (

254

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

255

# a "group" containing digits

256

"value", # datetime.timedelta or None

257

"followers")) # Tokens valid after this token

258

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

259

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

260

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

261

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

262

datetime.timedelta(seconds=1),

263

frozenset((token_end, )))

264

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

265

datetime.timedelta(minutes=1),

266

frozenset((token_second, token_end)))

267

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

268

datetime.timedelta(hours=1),

269

frozenset((token_minute, token_end)))

270

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

271

None,

272

frozenset((token_hour, token_minute,

273

token_second)))

274

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

275

datetime.timedelta(days=1),

276

frozenset((token_time, token_end)))

277

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

278

datetime.timedelta(weeks=4),

279

frozenset((token_day, token_end)))

280

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

281

datetime.timedelta(weeks=52),

282

frozenset((token_month, token_end)))

283

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

284

datetime.timedelta(weeks=1),

285

frozenset((token_end, )))

286

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

287

frozenset((token_year, token_month,

288

token_day, token_time,

289

token_week)))

290

# Define starting values:

291

# Value so far

292

value = datetime.timedelta()

293

found_token = None

294

# Following valid tokens

295

followers = frozenset((token_duration, ))

296

# String left to parse

297

s = duration

298

# Loop until end token is found

299

while found_token is not token_end:

300

# Search for any currently valid tokens

301

for token in followers:

302

match = token.regexp.match(s)

303

if match is not None:

304

# Token found

305

if token.value is not None:

306

# Value found, parse digits

307

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

308

# Add to value so far

309

value += factor * token.value

310

# Strip token from string

311

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

312

# Go to found token

313

found_token = token

314

# Set valid next tokens

315

followers = found_token.followers

316

break

317

else:

318

# No currently valid tokens were found

319

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

320

.format(duration))

321

# End token found

322

return value

323

324

325

def parse_pre_1_6_1_interval(interval):

326

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

327

and return a datetime.timedelta

328

329

>>> parse_pre_1_6_1_interval('7d')

330

datetime.timedelta(7)

331

>>> parse_pre_1_6_1_interval('60s')

332

datetime.timedelta(0, 60)

333

>>> parse_pre_1_6_1_interval('60m')

334

datetime.timedelta(0, 3600)

335

>>> parse_pre_1_6_1_interval('24h')

336

datetime.timedelta(1)

337

>>> parse_pre_1_6_1_interval('1w')

338

datetime.timedelta(7)

339

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

340

datetime.timedelta(0, 330)

341

>>> parse_pre_1_6_1_interval('')

342

datetime.timedelta(0)

343

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

344

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

345

datetime.timedelta(2, 480, 18000)

346

347

"""

348

349

value = datetime.timedelta(0)

350

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

351

352

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

353

if suffix == "d":

354

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

355

elif suffix == "s":

356

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

357

elif suffix == "m":

358

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

359

elif suffix == "h":

360

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

361

elif suffix == "w":

362

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

363

elif suffix == "":

364

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

365

return value

366

367

368

def check_option_syntax(parser, options):

369

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

370

argparse"""

371

372

def has_actions(options):

373

return any((options.enable,

374

options.disable,

375

options.bump_timeout,

376

options.start_checker,

377

options.stop_checker,

378

options.is_enabled,

379

options.remove,

380

options.checker is not None,

381

options.timeout is not None,

382

options.extended_timeout is not None,

383

options.interval is not None,

384

options.approved_by_default is not None,

385

options.approval_delay is not None,

386

options.approval_duration is not None,

387

options.host is not None,

388

options.secret is not None,

389

options.approve,

390

options.deny))

391

392

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

393

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

394

if options.verbose and has_actions(options):

395

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

396

if options.dump_json and (options.verbose

397

or has_actions(options)):

398

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

399

if options.all and not has_actions(options):

400

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

401

if options.is_enabled and len(options.client) > 1:

402

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

403

if options.remove:

404

options.remove = False

405

if has_actions(options) and not options.deny:

406

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

407

options.remove = True

408

409

410

411

class dbus(object):

412

413

class SystemBus(object):

414

415

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

416

def get_managed_objects(self, busname, objectpath):

417

return self.call_method("GetManagedObjects", busname,

418

objectpath,

419

self.object_manager_iface)

420

421

properties_iface = "org.freedesktop.DBus.Properties"

422

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

423

value):

424

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

425

self.properties_iface, interface, key,

426

value)

427

428

429

class MandosBus(SystemBus):

430

busname_domain = "se.recompile"

431

busname = busname_domain + ".Mandos"

432

server_path = "/"

433

server_interface = busname_domain + ".Mandos"

434

client_interface = busname_domain + ".Mandos.Client"

435

del busname_domain

436

437

def get_clients_and_properties(self):

438

managed_objects = self.get_managed_objects(

439

self.busname, self.server_path)

440

return {objpath: properties[self.client_interface]

441

for objpath, properties in managed_objects.items()

442

if self.client_interface in properties}

443

444

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

445

return self.set_property(self.busname, objectpath,

446

self.client_interface, key,

447

value)

448

449

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

450

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

451

self.client_interface, *args)

452

453

def call_server_method(self, method, *args):

454

return self.call_method(method, self.busname,

455

self.server_path,

456

self.server_interface, *args)

457

458

class Error(Exception):

459

pass

460

461

class ConnectFailed(Error):

462

pass

463

464

465

class dbus_python_adapter(object):

466

467

class SystemBus(dbus.MandosBus):

468

"""Use dbus-python"""

469

470

def __init__(self, module=dbus_python):

471

self.dbus_python = module

472

self.bus = self.dbus_python.SystemBus()

473

474

@contextlib.contextmanager

475

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

476

try:

477

yield

478

except self.dbus_python.exceptions.DBusException as e:

479

# This does what "raise from" would do

480

exc = exception_class(*e.args)

481

exc.__cause__ = e

482

raise exc

483

484

def call_method(self, methodname, busname, objectpath,

485

interface, *args):

486

proxy_object = self.get_object(busname, objectpath)

487

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

488

interface, methodname,

489

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

490

method = getattr(proxy_object, methodname)

491

with self.convert_exception():

492

with dbus_python_adapter.SilenceLogger(

493

"dbus.proxies"):

494

value = method(*args, dbus_interface=interface)

495

return self.type_filter(value)

496

497

def get_object(self, busname, objectpath):

498

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

499

busname, objectpath)

500

with self.convert_exception(dbus.ConnectFailed):

501

return self.bus.get_object(busname, objectpath)

502

503

def type_filter(self, value):

504

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

505

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

506

return bool(value)

507

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

508

return str(value)

509

# Also recurse into dictionaries

510

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

511

return {self.type_filter(key):

512

self.type_filter(subval)

513

for key, subval in value.items()}

514

return value

515

516

517

class SilenceLogger(object):

518

"Simple context manager to silence a particular logger"

519

def __init__(self, loggername):

520

self.logger = logging.getLogger(loggername)

521

522

def __enter__(self):

523

self.logger.addFilter(self.nullfilter)

524

525

class NullFilter(logging.Filter):

526

def filter(self, record):

527

return False

528

529

nullfilter = NullFilter()

530

531

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

532

self.logger.removeFilter(self.nullfilter)

533

534

535

class CachingBus(SystemBus):

536

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

537

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

538

self.object_cache = {}

539

super(dbus_python_adapter.CachingBus,

540

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

541

def get_object(self, busname, objectpath):

542

try:

543

return self.object_cache[(busname, objectpath)]

544

except KeyError:

545

new_object = super(

546

dbus_python_adapter.CachingBus,

547

self).get_object(busname, objectpath)

548

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

549

return new_object

550

551

552

def commands_from_options(options):

553

554

commands = []

555

556

if options.is_enabled:

557

commands.append(command.IsEnabled())

558

559

if options.approve:

560

commands.append(command.Approve())

561

562

if options.deny:

563

commands.append(command.Deny())

564

565

if options.remove:

566

commands.append(command.Remove())

567

568

if options.dump_json:

569

commands.append(command.DumpJSON())

570

571

if options.enable:

572

commands.append(command.Enable())

573

574

if options.disable:

575

commands.append(command.Disable())

576

577

if options.bump_timeout:

578

commands.append(command.BumpTimeout())

579

580

if options.start_checker:

581

commands.append(command.StartChecker())

582

583

if options.stop_checker:

584

commands.append(command.StopChecker())

585

586

if options.approved_by_default is not None:

587

if options.approved_by_default:

588

commands.append(command.ApproveByDefault())

589

else:

590

commands.append(command.DenyByDefault())

591

592

if options.checker is not None:

593

commands.append(command.SetChecker(options.checker))

594

595

if options.host is not None:

596

commands.append(command.SetHost(options.host))

597

598

if options.secret is not None:

599

commands.append(command.SetSecret(options.secret))

600

601

if options.timeout is not None:

602

commands.append(command.SetTimeout(options.timeout))

603

604

if options.extended_timeout:

605

commands.append(

606

command.SetExtendedTimeout(options.extended_timeout))

607

608

if options.interval is not None:

609

commands.append(command.SetInterval(options.interval))

610

611

if options.approval_delay is not None:

612

commands.append(

613

command.SetApprovalDelay(options.approval_delay))

614

615

if options.approval_duration is not None:

616

commands.append(

617

command.SetApprovalDuration(options.approval_duration))

618

619

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

620

# optionally verbosely

621

if not commands:

622

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

623

624

return commands

625

626

627

class command(object):

628

"""A namespace for command classes"""

629

630

class Base(object):

631

"""Abstract base class for commands"""

632

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

633

"""Normal commands should implement run_on_one_client(),

634

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

635

override this run() method instead.

636

"""

637

self.bus = bus

638

for client, properties in clients.items():

639

self.run_on_one_client(client, properties)

640

641

642

class IsEnabled(Base):

643

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

644

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

645

if properties["Enabled"]:

646

sys.exit(0)

647

sys.exit(1)

648

649

650

class Approve(Base):

651

def run_on_one_client(self, client, properties):

652

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

653

654

655

class Deny(Base):

656

def run_on_one_client(self, client, properties):

657

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

658

659

660

class Remove(Base):

661

def run(self, clients, bus):

662

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

663

bus.call_server_method("RemoveClient", clientpath)

664

665

666

class Output(Base):

667

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

668

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

669

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

670

"Fingerprint", "CheckerRunning",

671

"LastEnabled", "ApprovalPending",

672

"ApprovedByDefault", "LastApprovalRequest",

673

"ApprovalDelay", "ApprovalDuration",

674

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

675

"LastCheckerStatus")

676

677

678

class DumpJSON(Output):

679

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

680

data = {properties["Name"]:

681

{key: properties[key]

682

for key in self.all_keywords}

683

for properties in clients.values()}

684

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

685

686

687

class PrintTable(Output):

688

def __init__(self, verbose=False):

689

self.verbose = verbose

690

691

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

692

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

693

"LastCheckedOK")

694

keywords = default_keywords

695

if self.verbose:

696

keywords = self.all_keywords

697

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

698

699

class TableOfClients(object):

700

tableheaders = {

701

"Name": "Name",

702

"Enabled": "Enabled",

703

"Timeout": "Timeout",

704

"LastCheckedOK": "Last Successful Check",

705

"LastApprovalRequest": "Last Approval Request",

706

"Created": "Created",

707

"Interval": "Interval",

708

"Host": "Host",

709

"Fingerprint": "Fingerprint",

710

"KeyID": "Key ID",

711

"CheckerRunning": "Check Is Running",

712

"LastEnabled": "Last Enabled",

713

"ApprovalPending": "Approval Is Pending",

714

"ApprovedByDefault": "Approved By Default",

715

"ApprovalDelay": "Approval Delay",

716

"ApprovalDuration": "Approval Duration",

717

"Checker": "Checker",

718

"ExtendedTimeout": "Extended Timeout",

719

"Expires": "Expires",

720

"LastCheckerStatus": "Last Checker Status",

721

}

722

723

def __init__(self, clients, keywords):

724

self.clients = clients

725

self.keywords = keywords

726

727

def __str__(self):

728

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

729

730

if sys.version_info.major == 2:

731

__unicode__ = __str__

732

def __str__(self):

733

return str(self).encode(

734

locale.getpreferredencoding())

735

736

def rows(self):

737

format_string = self.row_formatting_string()

738

rows = [self.header_line(format_string)]

739

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

740

for client in self.clients)

741

return rows

742

743

def row_formatting_string(self):

744

"Format string used to format table rows"

745

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

746

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

747

*(len(self.string_from_client(client,

748

key))

749

for client in self.clients)),

750

key=key)

751

for key in self.keywords)

752

753

def string_from_client(self, client, key):

754

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

755

756

@classmethod

757

def valuetostring(cls, value, keyword):

758

if isinstance(value, bool):

759

return "Yes" if value else "No"

760

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

761

"ApprovalDuration", "ExtendedTimeout"):

762

return cls.milliseconds_to_string(value)

763

return str(value)

764

765

def header_line(self, format_string):

766

return format_string.format(**self.tableheaders)

767

768

def client_line(self, client, format_string):

769

return format_string.format(

770

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

771

for key in self.keywords})

772

773

@staticmethod

774

def milliseconds_to_string(ms):

775

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

776

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

777

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

778

if td.days else "",

779

hours=td.seconds // 3600,

780

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

781

seconds=td.seconds % 60))

782

783

784

class PropertySetter(Base):

785

"Abstract class for Actions for setting one client property"

786

787

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

788

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

789

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

790

self.value_to_set)

791

792

@property

793

def propname(self):

794

raise NotImplementedError()

795

796

797

class Enable(PropertySetter):

798

propname = "Enabled"

799

value_to_set = True

800

801

802

class Disable(PropertySetter):

803

propname = "Enabled"

804

value_to_set = False

805

806

807

class BumpTimeout(PropertySetter):

808

propname = "LastCheckedOK"

809

value_to_set = ""

810

811

812

class StartChecker(PropertySetter):

813

propname = "CheckerRunning"

814

value_to_set = True

815

816

817

class StopChecker(PropertySetter):

818

propname = "CheckerRunning"

819

value_to_set = False

820

821

822

class ApproveByDefault(PropertySetter):

823

propname = "ApprovedByDefault"

824

value_to_set = True

825

826

827

class DenyByDefault(PropertySetter):

828

propname = "ApprovedByDefault"

829

value_to_set = False

830

831

832

class PropertySetterValue(PropertySetter):

833

"""Abstract class for PropertySetter recieving a value as

834

constructor argument instead of a class attribute."""

835

def __init__(self, value):

836

self.value_to_set = value

837

838

839

class SetChecker(PropertySetterValue):

840

propname = "Checker"

841

842

843

class SetHost(PropertySetterValue):

844

propname = "Host"

845

846

847

class SetSecret(PropertySetterValue):

848

propname = "Secret"

849

850

@property

851

def value_to_set(self):

852

return self._vts

853

854

@value_to_set.setter

855

def value_to_set(self, value):

856

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

857

self._vts = value.read()

858

value.close()

859

860

861

class PropertySetterValueMilliseconds(PropertySetterValue):

862

"""Abstract class for PropertySetterValue taking a value

863

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

864

milliseconds."""

865

866

@property

867

def value_to_set(self):

868

return self._vts

869

870

@value_to_set.setter

871

def value_to_set(self, value):

872

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

873

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

874

875

876

class SetTimeout(PropertySetterValueMilliseconds):

877

propname = "Timeout"

878

879

880

class SetExtendedTimeout(PropertySetterValueMilliseconds):

881

propname = "ExtendedTimeout"

882

883

884

class SetInterval(PropertySetterValueMilliseconds):

885

propname = "Interval"

886

887

888

class SetApprovalDelay(PropertySetterValueMilliseconds):

889

propname = "ApprovalDelay"

890

891

892

class SetApprovalDuration(PropertySetterValueMilliseconds):

893

propname = "ApprovalDuration"

894

895

896

897

class TestCaseWithAssertLogs(unittest.TestCase):

898

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

899

900

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

901

@contextlib.contextmanager

902

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

903

capturing_handler = self.CapturingLevelHandler(level)

904

old_level = logger.level

905

old_propagate = logger.propagate

906

logger.addHandler(capturing_handler)

907

logger.setLevel(level)

908

logger.propagate = False

909

try:

910

yield capturing_handler.watcher

911

finally:

912

logger.propagate = old_propagate

913

logger.removeHandler(capturing_handler)

914

logger.setLevel(old_level)

915

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

916

0)

917

918

class CapturingLevelHandler(logging.Handler):

919

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

920

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

921

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

922

def emit(self, record):

923

self.watcher.records.append(record)

924

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

925

926

LoggingWatcher = collections.namedtuple("LoggingWatcher",

927

("records",

928

"output"))

929

930

931

class Unique(object):

932

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

933

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

934

935

936

class Test_string_to_delta(TestCaseWithAssertLogs):

937

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

938

# rfc3339_duration_to_delta() already has more comprehensive

939

# tests, which are run by doctest.

940

941

def test_rfc3339_zero_seconds(self):

942

self.assertEqual(datetime.timedelta(),

943

string_to_delta("PT0S"))

944

945

def test_rfc3339_zero_days(self):

946

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

947

948

def test_rfc3339_one_second(self):

949

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

950

string_to_delta("PT1S"))

951

952

def test_rfc3339_two_hours(self):

953

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

954

string_to_delta("PT2H"))

955

956

def test_falls_back_to_pre_1_6_1_with_warning(self):

957

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

958

value = string_to_delta("2h")

959

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

960

961

962

class Test_check_option_syntax(unittest.TestCase):

963

def setUp(self):

964

self.parser = argparse.ArgumentParser()

965

add_command_line_options(self.parser)

966

967

def test_actions_requires_client_or_all(self):

968

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

969

options = self.parser.parse_args()

970

setattr(options, action, value)

971

with self.assertParseError():

972

self.check_option_syntax(options)

973

974

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

975

# check_option_syntax()

976

actions = {

977

# The actual values set here are not that important, but we do

978

# at least stick to the correct types, even though they are

979

# never used

980

"enable": True,

981

"disable": True,

982

"bump_timeout": True,

983

"start_checker": True,

984

"stop_checker": True,

985

"is_enabled": True,

986

"remove": True,

987

"checker": "x",

988

"timeout": datetime.timedelta(),

989

"extended_timeout": datetime.timedelta(),

990

"interval": datetime.timedelta(),

991

"approved_by_default": True,

992

"approval_delay": datetime.timedelta(),

993

"approval_duration": datetime.timedelta(),

994

"host": "x",

995

"secret": io.BytesIO(b"x"),

996

"approve": True,

997

"deny": True,

998

}

999

1000

@contextlib.contextmanager

1001

def assertParseError(self):

1002

with self.assertRaises(SystemExit) as e:

1003

with self.redirect_stderr_to_devnull():

1004

yield

1005

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

1006

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

1007

# /argparse.html#exiting-methods

1008

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

1009

1010

@staticmethod

1011

@contextlib.contextmanager

1012

def redirect_stderr_to_devnull():

1013

old_stderr = sys.stderr

1014

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

1015

sys.stderr = null

1016

try:

1017

yield

1018

finally:

1019

sys.stderr = old_stderr

1020

1021

def check_option_syntax(self, options):

1022

check_option_syntax(self.parser, options)

1023

1024

def test_actions_all_conflicts_with_verbose(self):

1025

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

1026

options = self.parser.parse_args()

1027

setattr(options, action, value)

1028

options.all = True

1029

options.verbose = True

1030

with self.assertParseError():

1031

self.check_option_syntax(options)

1032

1033

def test_actions_with_client_conflicts_with_verbose(self):

1034

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

1035

options = self.parser.parse_args()

1036

setattr(options, action, value)

1037

options.verbose = True

1038

options.client = ["client"]

1039

with self.assertParseError():

1040

self.check_option_syntax(options)

1041

1042

def test_dump_json_conflicts_with_verbose(self):

1043

options = self.parser.parse_args()

1044

options.dump_json = True

1045

options.verbose = True

1046

with self.assertParseError():

1047

self.check_option_syntax(options)

1048

1049

def test_dump_json_conflicts_with_action(self):

1050

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

1051

options = self.parser.parse_args()

1052

setattr(options, action, value)

1053

options.dump_json = True

1054

with self.assertParseError():

1055

self.check_option_syntax(options)

1056

1057

def test_all_can_not_be_alone(self):

1058

options = self.parser.parse_args()

1059

options.all = True

1060

with self.assertParseError():

1061

self.check_option_syntax(options)

1062

1063

def test_all_is_ok_with_any_action(self):

1064

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

1065

options = self.parser.parse_args()

1066

setattr(options, action, value)

1067

options.all = True

1068

self.check_option_syntax(options)

1069

1070

def test_any_action_is_ok_with_one_client(self):

1071

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

1072

options = self.parser.parse_args()

1073

setattr(options, action, value)

1074

options.client = ["client"]

1075

self.check_option_syntax(options)

1076

1077

def test_one_client_with_all_actions_except_is_enabled(self):

1078

options = self.parser.parse_args()

1079

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

1080

if action == "is_enabled":

1081

continue

1082

setattr(options, action, value)

1083

options.client = ["client"]

1084

self.check_option_syntax(options)

1085

1086

def test_two_clients_with_all_actions_except_is_enabled(self):

1087

options = self.parser.parse_args()

1088

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

1089

if action == "is_enabled":

1090

continue

1091

setattr(options, action, value)

1092

options.client = ["client1", "client2"]

1093

self.check_option_syntax(options)

1094

1095

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1096

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

1097

if action == "is_enabled":

1098

continue

1099

options = self.parser.parse_args()

1100

setattr(options, action, value)

1101

options.client = ["client1", "client2"]

1102

self.check_option_syntax(options)

1103

1104

def test_is_enabled_fails_without_client(self):

1105

options = self.parser.parse_args()

1106

options.is_enabled = True

1107

with self.assertParseError():

1108

self.check_option_syntax(options)

1109

1110

def test_is_enabled_fails_with_two_clients(self):

1111

options = self.parser.parse_args()

1112

options.is_enabled = True

1113

options.client = ["client1", "client2"]

1114

with self.assertParseError():

1115

self.check_option_syntax(options)

1116

1117

def test_remove_can_only_be_combined_with_action_deny(self):

1118

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

1119

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

1120

continue

1121

options = self.parser.parse_args()

1122

setattr(options, action, value)

1123

options.all = True

1124

options.remove = True

1125

with self.assertParseError():

1126

self.check_option_syntax(options)

1127

1128

1129

class Test_dbus_exceptions(unittest.TestCase):

1130

1131

def test_dbus_ConnectFailed_is_Error(self):

1132

with self.assertRaises(dbus.Error):

1133

raise dbus.ConnectFailed()

1134

1135

1136

class Test_dbus_MandosBus(unittest.TestCase):

1137

1138

class MockMandosBus(dbus.MandosBus):

1139

def __init__(self):

1140

self._name = "se.recompile.Mandos"

1141

self._server_path = "/"

1142

self._server_interface = "se.recompile.Mandos"

1143

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

1144

self.calls = []

1145

self.call_method_return = Unique()

1146

1147

def call_method(self, methodname, busname, objectpath,

1148

interface, *args):

1149

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

1150

interface, args))

1151

return self.call_method_return

1152

1153

def setUp(self):

1154

self.bus = self.MockMandosBus()

1155

1156

def test_set_client_property(self):

1157

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

1158

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

1159

"org.freedesktop.DBus.Properties",

1160

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

1161

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

1162

1163

def test_call_client_method(self):

1164

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

1165

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

1166

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

1167

self.bus._client_interface, ())

1168

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

1169

1170

def test_call_client_method_with_args(self):

1171

args = (Unique(), Unique())

1172

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

1173

*args)

1174

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

1175

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

1176

self.bus._client_interface,

1177

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

1178

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

1179

1180

def test_get_clients_and_properties(self):

1181

managed_objects = {

1182

"objectpath": {

1183

self.bus._client_interface: {

1184

"key": "value",

1185

"bool": True,

1186

},

1187

"irrelevant_interface": {

1188

"key": "othervalue",

1189

"bool": False,

1190

},

1191

},

1192

"other_objectpath": {

1193

"other_irrelevant_interface": {

1194

"key": "value 3",

1195

"bool": None,

1196

},

1197

},

1198

}

1199

expected_clients_and_properties = {

1200

"objectpath": {

1201

"key": "value",

1202

"bool": True,

1203

}

1204

}

1205

self.bus.call_method_return = managed_objects

1206

ret = self.bus.get_clients_and_properties()

1207

self.assertDictEqual(expected_clients_and_properties, ret)

1208

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

1209

self.bus._server_path,

1210

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

1211

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

1212

1213

def test_call_server_method(self):

1214

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

1215

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

1216

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

1217

self.bus._server_path,

1218

self.bus._server_interface, ())

1219

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

1220

1221

def test_call_server_method_with_args(self):

1222

args = (Unique(), Unique())

1223

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

1224

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

1225

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

1226

self.bus._server_path,

1227

self.bus._server_interface,

1228

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

1229

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

1230

1231

1232

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1233

1234

def MockDBusPython_func(self, func):

1235

class mock_dbus_python(object):

1236

"""mock dbus-python module"""

1237

class exceptions(object):

1238

"""Pseudo-namespace"""

1239

class DBusException(Exception):

1240

pass

1241

class SystemBus(object):

1242

@staticmethod

1243

def get_object(busname, objectpath):

1244

DBusObject = collections.namedtuple(

1245

"DBusObject", ("methodname",))

1246

def method(*args, **kwargs):

1247

self.assertEqual({"dbus_interface":

1248

"interface"},

1249

kwargs)

1250

return func(*args)

1251

return DBusObject(methodname=method)

1252

class Boolean(object):

1253

def __init__(self, value):

1254

self.value = bool(value)

1255

def __bool__(self):

1256

return self.value

1257

if sys.version_info.major == 2:

1258

__nonzero__ = __bool__

1259

class ObjectPath(str):

1260

pass

1261

class Dictionary(dict):

1262

pass

1263

return mock_dbus_python

1264

1265

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

1266

interface, *args):

1267

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

1268

return bus.call_method(methodname, busname, objectpath,

1269

interface, *args)

1270

1271

def test_call_method_returns(self):

1272

expected_method_return = Unique()

1273

method_args = (Unique(), Unique())

1274

def func(*args):

1275

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

1276

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

1277

self.assertIs(marg, arg)

1278

return expected_method_return

1279

mock_dbus_python = self.MockDBusPython_func(func)

1280

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1281

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

1282

"objectpath", "interface",

1283

*method_args)

1284

self.assertIs(ret, expected_method_return)

1285

1286

def test_call_method_filters_bool_true(self):

1287

def func():

1288

return method_return

1289

mock_dbus_python = self.MockDBusPython_func(func)

1290

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1291

method_return = mock_dbus_python.Boolean(True)

1292

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

1293

"objectpath", "interface")

1294

self.assertTrue(ret)

1295

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1296

1297

def test_call_method_filters_bool_false(self):

1298

def func():

1299

return method_return

1300

mock_dbus_python = self.MockDBusPython_func(func)

1301

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1302

method_return = mock_dbus_python.Boolean(False)

1303

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

1304

"objectpath", "interface")

1305

self.assertFalse(ret)

1306

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1307

1308

def test_call_method_filters_objectpath(self):

1309

def func():

1310

return method_return

1311

mock_dbus_python = self.MockDBusPython_func(func)

1312

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1313

method_return = mock_dbus_python.ObjectPath("objectpath")

1314

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

1315

"objectpath", "interface")

1316

self.assertEqual("objectpath", ret)

1317

self.assertIsNot("objectpath", ret)

1318

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1319

1320

def test_call_method_filters_booleans_in_dict(self):

1321

def func():

1322

return method_return

1323

mock_dbus_python = self.MockDBusPython_func(func)

1324

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1325

method_return = mock_dbus_python.Dictionary(

1326

{mock_dbus_python.Boolean(True):

1327

mock_dbus_python.Boolean(False),

1328

mock_dbus_python.Boolean(False):

1329

mock_dbus_python.Boolean(True)})

1330

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

1331

"objectpath", "interface")

1332

expected_method_return = {True: False,

1333

False: True}

1334

self.assertEqual(expected_method_return, ret)

1335

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1336

1337

def test_call_method_filters_objectpaths_in_dict(self):

1338

def func():

1339

return method_return

1340

mock_dbus_python = self.MockDBusPython_func(func)

1341

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1342

method_return = mock_dbus_python.Dictionary(

1343

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1344

mock_dbus_python.ObjectPath("objectpath_value_1"),

1345

mock_dbus_python.ObjectPath("objectpath_key_2"):

1346

mock_dbus_python.ObjectPath("objectpath_value_2")})

1347

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

1348

"objectpath", "interface")

1349

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

1350

for key, value in

1351

method_return.items()}

1352

self.assertEqual(expected_method_return, ret)

1353

self.assertIsInstance(ret, dict)

1354

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1355

1356

def test_call_method_filters_dict_in_dict(self):

1357

def func():

1358

return method_return

1359

mock_dbus_python = self.MockDBusPython_func(func)

1360

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1361

method_return = mock_dbus_python.Dictionary(

1362

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

1363

"key12": "value12"}),

1364

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

1365

"key22": "value22"})})

1366

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

1367

"objectpath", "interface")

1368

expected_method_return = {

1369

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

1370

"key12": "value12"},

1371

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

1372

"key22": "value22"},

1373

}

1374

self.assertEqual(expected_method_return, ret)

1375

self.assertIsInstance(ret, dict)

1376

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1377

for key, value in ret.items():

1378

self.assertIsInstance(value, dict)

1379

self.assertEqual(expected_method_return[key], value)

1380

self.assertNotIsInstance(value,

1381

mock_dbus_python.Dictionary)

1382

1383

def test_call_method_filters_dict_three_deep(self):

1384

def func():

1385

return method_return

1386

mock_dbus_python = self.MockDBusPython_func(func)

1387

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1388

method_return = mock_dbus_python.Dictionary(

1389

{"key1":

1390

mock_dbus_python.Dictionary(

1391

{"key2":

1392

mock_dbus_python.Dictionary(

1393

{"key3":

1394

mock_dbus_python.Boolean(True),

1395

}),

1396

}),

1397

})

1398

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

1399

"objectpath", "interface")

1400

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

1401

self.assertEqual(expected_method_return, ret)

1402

self.assertIsInstance(ret, dict)

1403

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1404

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

1405

self.assertNotIsInstance(ret["key1"],

1406

mock_dbus_python.Dictionary)

1407

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

1408

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

1409

mock_dbus_python.Dictionary)

1410

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

1411

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

1412

mock_dbus_python.Boolean)

1413

1414

def test_call_method_handles_exception(self):

1415

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

1416

1417

def func():

1418

dbus_logger.error("Test")

1419

raise mock_dbus_python.exceptions.DBusException()

1420

1421

mock_dbus_python = self.MockDBusPython_func(func)

1422

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1423

1424

class CountingHandler(logging.Handler):

1425

count = 0

1426

def emit(self, record):

1427

self.count += 1

1428

1429

counting_handler = CountingHandler()

1430

1431

dbus_logger.addHandler(counting_handler)

1432

1433

try:

1434

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

1435

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

1436

"objectpath", "interface")

1437

finally:

1438

dbus_logger.removeFilter(counting_handler)

1439

1440

self.assertNotIsInstance(e, dbus.ConnectFailed)

1441

1442

# Make sure the dbus logger was suppressed

1443

self.assertEqual(0, counting_handler.count)

1444

1445

def test_get_object_converts_to_correct_exception(self):

1446

bus = dbus_python_adapter.SystemBus(

1447

self.fake_dbus_python_raises_exception_on_connect)

1448

with self.assertRaises(dbus.ConnectFailed):

1449

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

1450

"objectpath", "interface")

1451

1452

class fake_dbus_python_raises_exception_on_connect(object):

1453

"""fake dbus-python module"""

1454

class exceptions(object):

1455

"""Pseudo-namespace"""

1456

class DBusException(Exception):

1457

pass

1458

1459

@classmethod

1460

def SystemBus(cls):

1461

def get_object(busname, objectpath):

1462

raise cls.exceptions.DBusException()

1463

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

1464

return Bus(get_object=get_object)

1465

1466

1467

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1468

class mock_dbus_python(object):

1469

"""mock dbus-python modules"""

1470

class SystemBus(object):

1471

@staticmethod

1472

def get_object(busname, objectpath):

1473

return Unique()

1474

1475

def setUp(self):

1476

self.bus = dbus_python_adapter.CachingBus(

1477

self.mock_dbus_python)

1478

1479

def test_returns_distinct_objectpaths(self):

1480

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

1481

self.assertIsInstance(obj1, Unique)

1482

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

1483

self.assertIsInstance(obj2, Unique)

1484

self.assertIsNot(obj1, obj2)

1485

1486

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

1494

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

1495

self.assertIsInstance(obj1, Unique)

1496

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

1497

self.assertIsInstance(obj2, Unique)

1498

self.assertIsNot(obj1, obj2)

1499

1500

def test_returns_same(self):

1501

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

1502

self.assertIsInstance(obj1, Unique)

1503

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

1504

self.assertIsInstance(obj2, Unique)

1505

self.assertIs(obj1, obj2)

1506

1507

def test_returns_same_old(self):

1508

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

1509

self.assertIsInstance(obj1, Unique)

1510

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

1511

self.assertIsInstance(obj2, Unique)

1512

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

1513

self.assertIsInstance(obj1b, Unique)

1514

self.assertIsNot(obj1, obj2)

1515

self.assertIsNot(obj2, obj1b)

1516

self.assertIs(obj1, obj1b)

1517

1518

1519

class Test_commands_from_options(unittest.TestCase):

1520

1521

def setUp(self):

1522

self.parser = argparse.ArgumentParser()

1523

add_command_line_options(self.parser)

1524

1525

def test_is_enabled(self):

1526

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

1527

command.IsEnabled)

1528

1529

def assert_command_from_args(self, args, command_cls,

1530

**cmd_attrs):

1531

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

1532

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

1533

options = self.parser.parse_args(args)

1534

check_option_syntax(self.parser, options)

1535

commands = commands_from_options(options)

1536

self.assertEqual(1, len(commands))

1537

command = commands[0]

1538

self.assertIsInstance(command, command_cls)

1539

for key, value in cmd_attrs.items():

1540

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

1541

1542

def test_is_enabled_short(self):

1543

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

1544

command.IsEnabled)

1545

1546

def test_approve(self):

1547

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

1548

command.Approve)

1549

1550

def test_approve_short(self):

1551

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

1552

command.Approve)

1553

1554

def test_deny(self):

1555

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

1556

command.Deny)

1557

1558

def test_deny_short(self):

1559

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

1560

1561

def test_remove(self):

1562

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

1563

command.Remove)

1564

1565

def test_deny_before_remove(self):

1566

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

1567

"client"])

1568

check_option_syntax(self.parser, options)

1569

commands = commands_from_options(options)

1570

self.assertEqual(2, len(commands))

1571

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

1572

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

1573

1574

def test_deny_before_remove_reversed(self):

1575

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

1576

"--all"])

1577

check_option_syntax(self.parser, options)

1578

commands = commands_from_options(options)

1579

self.assertEqual(2, len(commands))

1580

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

1581

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

1582

1583

def test_remove_short(self):

1584

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

1585

command.Remove)

1586

1587

def test_dump_json(self):

1588

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

1589

command.DumpJSON)

1590

1591

def test_enable(self):

1592

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

1593

command.Enable)

1594

1595

def test_enable_short(self):

1596

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

1597

command.Enable)

1598

1599

def test_disable(self):

1600

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

1601

command.Disable)

1602

1603

def test_disable_short(self):

1604

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

1605

command.Disable)

1606

1607

def test_bump_timeout(self):

1608

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

1609

command.BumpTimeout)

1610

1611

def test_bump_timeout_short(self):

1612

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

1613

command.BumpTimeout)

1614

1615

def test_start_checker(self):

1616

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

1617

command.StartChecker)

1618

1619

def test_stop_checker(self):

1620

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

1621

command.StopChecker)

1622

1623

def test_approve_by_default(self):

1624

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

1625

"client"],

1626

command.ApproveByDefault)

1627

1628

def test_deny_by_default(self):

1629

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

1630

command.DenyByDefault)

1631

1632

def test_checker(self):

1633

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

1634

command.SetChecker,

1635

value_to_set=":")

1636

1637

def test_checker_empty(self):

1638

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

1639

command.SetChecker,

1640

value_to_set="")

1641

1642

def test_checker_short(self):

1643

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

1644

command.SetChecker,

1645

value_to_set=":")

1646

1647

def test_host(self):

1648

self.assert_command_from_args(

1649

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

1650

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

1651

1652

def test_host_short(self):

1653

self.assert_command_from_args(

1654

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

1655

value_to_set="client.example.org")

1656

1657

def test_secret_devnull(self):

1658

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

1659

"client"], command.SetSecret,

1660

value_to_set=b"")

1661

1662

def test_secret_tempfile(self):

1663

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

1664

value = b"secret\0xyzzy\nbar"

1665

f.write(value)

1666

f.seek(0)

1667

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

1668

"client"],

1669

command.SetSecret,

1670

value_to_set=value)

1671

1672

def test_secret_devnull_short(self):

1673

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

1674

"client"], command.SetSecret,

1675

value_to_set=b"")

1676

1677

def test_secret_tempfile_short(self):

1678

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

1679

value = b"secret\0xyzzy\nbar"

1680

f.write(value)

1681

f.seek(0)

1682

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

1683

command.SetSecret,

1684

value_to_set=value)

1685

1686

def test_timeout(self):

1687

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

1688

command.SetTimeout,

1689

value_to_set=300000)

1690

1691

def test_timeout_short(self):

1692

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

1693

command.SetTimeout,

1694

value_to_set=300000)

1695

1696

def test_extended_timeout(self):

1697

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

1698

"client"],

1699

command.SetExtendedTimeout,

1700

value_to_set=900000)

1701

1702

def test_interval(self):

1703

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

1704

"client"], command.SetInterval,

1705

value_to_set=120000)

1706

1707

def test_interval_short(self):

1708

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

1709

command.SetInterval,

1710

value_to_set=120000)

1711

1712

def test_approval_delay(self):

1713

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

1714

"client"],

1715

command.SetApprovalDelay,

1716

value_to_set=30000)

1717

1718

def test_approval_duration(self):

1719

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

1720

"client"],

1721

command.SetApprovalDuration,

1722

value_to_set=1000)

1723

1724

def test_print_table(self):

1725

self.assert_command_from_args([], command.PrintTable,

1726

verbose=False)

1727

1728

def test_print_table_verbose(self):

1729

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

1730

command.PrintTable,

1731

verbose=True)

1732

1733

def test_print_table_verbose_short(self):

1734

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

1735

verbose=True)

1736

1737

1738

class TestCommand(unittest.TestCase):

1739

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

1740

1741

class FakeMandosBus(dbus.MandosBus):

1742

def __init__(self, testcase):

1743

self.client_properties = {

1744

"Name": "foo",

1745

"KeyID": ("92ed150794387c03ce684574b1139a65"

1746

"94a34f895daaaf09fd8ea90a27cddb12"),

1747

"Secret": b"secret",

1748

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

1749

"Enabled": True,

1750

"Timeout": 300000,

1751

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

1752

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

1753

"Interval": 120000,

1754

"Fingerprint": ("778827225BA7DE539C5A"

1755

"7CFA59CFF7CDBD9A5920"),

1756

"CheckerRunning": False,

1757

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

1758

"ApprovalPending": False,

1759

"ApprovedByDefault": True,

1760

"LastApprovalRequest": "",

1761

"ApprovalDelay": 0,

1762

"ApprovalDuration": 1000,

1763

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

1764

"ExtendedTimeout": 900000,

1765

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

1766

"LastCheckerStatus": 0,

1767

}

1768

self.other_client_properties = {

1769

"Name": "barbar",

1770

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1771

"6ab612cff5ad227247e46c2b020f441c"),

1772

"Secret": b"secretbar",

1773

"Host": "192.0.2.3",

1774

"Enabled": True,

1775

"Timeout": 300000,

1776

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

1777

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

1778

"Interval": 120000,

1779

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1780

"F547B3A107558FCA3A27"),

1781

"CheckerRunning": True,

1782

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

1783

"ApprovalPending": False,

1784

"ApprovedByDefault": False,

1785

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

1786

"ApprovalDelay": 30000,

1787

"ApprovalDuration": 93785000,

1788

"Checker": ":",

1789

"ExtendedTimeout": 900000,

1790

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

1791

"LastCheckerStatus": -2,

1792

}

1793

self.clients = collections.OrderedDict(

1794

[

1795

("client_objectpath", self.client_properties),

1796

("other_client_objectpath",

1797

self.other_client_properties),

1798

])

1799

self.one_client = {"client_objectpath":

1800

self.client_properties}

1801

self.testcase = testcase

1802

self.calls = []

1803

1804

def call_method(self, methodname, busname, objectpath,

1805

interface, *args):

1806

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

1807

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

1808

interface, args))

1809

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

1810

if methodname == "Set":

1811

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

1812

interface, key, value = args

1813

self.testcase.assertEqual(

1814

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

1815

self.clients[objectpath][key] = value

1816

return

1817

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

1818

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

1819

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

1820

clientpath = args[0]

1821

del self.clients[clientpath]

1822

return

1823

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

1824

if methodname == "Approve":

1825

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

1826

return

1827

raise ValueError()

1828

1829

def setUp(self):

1830

self.bus = self.FakeMandosBus(self)

1831

1832

1833

class TestBaseCommands(TestCommand):

1834

1835

def test_IsEnabled_exits_successfully(self):

1836

with self.assertRaises(SystemExit) as e:

1837

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

1838

if e.exception.code is not None:

1839

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

1840

else:

1841

self.assertIsNone(e.exception.code)

1842

1843

def test_IsEnabled_exits_with_failure(self):

1844

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

1845

with self.assertRaises(SystemExit) as e:

1846

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

1847

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

1848

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

1849

else:

1850

self.assertIsNotNone(e.exception.code)

1851

1852

def test_Approve(self):

1853

busname = "se.recompile.Mandos"

1854

client_interface = "se.recompile.Mandos.Client"

1855

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

1856

for clientpath in self.bus.clients:

1857

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

1858

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

1859

1860

def test_Deny(self):

1861

busname = "se.recompile.Mandos"

1862

client_interface = "se.recompile.Mandos.Client"

1863

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

1864

for clientpath in self.bus.clients:

1865

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

1866

client_interface, (False,)),

1867

self.bus.calls)

1868

1869

def test_Remove(self):

1870

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

1871

for clientpath in self.bus.clients:

1872

self.assertIn(("RemoveClient", dbus_busname,

1873

dbus_server_path, dbus_server_interface,

1874

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

1875

1876

expected_json = {

1877

"foo": {

1878

"Name": "foo",

1879

"KeyID": ("92ed150794387c03ce684574b1139a65"

1880

"94a34f895daaaf09fd8ea90a27cddb12"),

1881

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

1882

"Enabled": True,

1883

"Timeout": 300000,

1884

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

1885

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

1886

"Interval": 120000,

1887

"Fingerprint": ("778827225BA7DE539C5A"

1888

"7CFA59CFF7CDBD9A5920"),

1889

"CheckerRunning": False,

1890

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

1891

"ApprovalPending": False,

1892

"ApprovedByDefault": True,

1893

"LastApprovalRequest": "",

1894

"ApprovalDelay": 0,

1895

"ApprovalDuration": 1000,

1896

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

1897

"ExtendedTimeout": 900000,

1898

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

1899

"LastCheckerStatus": 0,

1900

},

1901

"barbar": {

1902

"Name": "barbar",

1903

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1904

"6ab612cff5ad227247e46c2b020f441c"),

1905

"Host": "192.0.2.3",

1906

"Enabled": True,

1907

"Timeout": 300000,

1908

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

1909

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

1910

"Interval": 120000,

1911

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1912

"F547B3A107558FCA3A27"),

1913

"CheckerRunning": True,

1914

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

1915

"ApprovalPending": False,

1916

"ApprovedByDefault": False,

1917

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

1918

"ApprovalDelay": 30000,

1919

"ApprovalDuration": 93785000,

1920

"Checker": ":",

1921

"ExtendedTimeout": 900000,

1922

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

1923

"LastCheckerStatus": -2,

1924

},

1925

}

1926

1927

def test_DumpJSON_normal(self):

1928

with self.capture_stdout_to_buffer() as buffer:

1929

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

1930

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

1931

self.assertDictEqual(self.expected_json, json_data)

1932

1933

@staticmethod

1934

@contextlib.contextmanager

1935

def capture_stdout_to_buffer():

1936

capture_buffer = io.StringIO()

1937

old_stdout = sys.stdout

1938

sys.stdout = capture_buffer

1939

try:

1940

yield capture_buffer

1941

finally:

1942

sys.stdout = old_stdout

1943

1944

def test_DumpJSON_one_client(self):

1945

with self.capture_stdout_to_buffer() as buffer:

1946

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

1947

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

1948

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

1949

self.assertDictEqual(expected_json, json_data)

1950

1951

def test_PrintTable_normal(self):

1952

with self.capture_stdout_to_buffer() as buffer:

1953

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

1954

expected_output = "\n".join((

1955

"Name Enabled Timeout Last Successful Check",

1956

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

1957

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

1958

)) + "\n"

1959

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

1960

1961

def test_PrintTable_verbose(self):

1962

with self.capture_stdout_to_buffer() as buffer:

1963

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

1964

columns = (

1965

(

1966

"Name ",

1967

"foo ",

1968

"barbar ",

1969

),(

1970

"Enabled ",

1971

"Yes ",

1972

"Yes ",

1973

),(

1974

"Timeout ",

1975

"00:05:00 ",

1976

"00:05:00 ",

1977

),(

1978

"Last Successful Check ",

1979

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

1980

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

1981

),(

1982

"Created ",

1983

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

1984

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

1985

),(

1986

"Interval ",

1987

"00:02:00 ",

1988

"00:02:00 ",

1989

),(

1990

"Host ",

1991

"foo.example.org ",

1992

"192.0.2.3 ",

1993

),(

1994

("Key ID "

1995

" "),

1996

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1997

"ea90a27cddb12 "),

1998

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1999

"46c2b020f441c "),

2000

),(

2001

"Fingerprint ",

2002

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2003

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2004

),(

2005

"Check Is Running ",

2006

"No ",

2007

"Yes ",

2008

),(

2009

"Last Enabled ",

2010

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

2011

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

2012

),(

2013

"Approval Is Pending ",

2014

"No ",

2015

"No ",

2016

),(

2017

"Approved By Default ",

2018

"Yes ",

2019

"No ",

2020

),(

2021

"Last Approval Request ",

2022

" ",

2023

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

2024

),(

2025

"Approval Delay ",

2026

"00:00:00 ",

2027

"00:00:30 ",

2028

),(

2029

"Approval Duration ",

2030

"00:00:01 ",

2031

"1T02:03:05 ",

2032

),(

2033

"Checker ",

2034

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

2035

": ",

2036

),(

2037

"Extended Timeout ",

2038

"00:15:00 ",

2039

"00:15:00 ",

2040

),(

2041

"Expires ",

2042

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

2043

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

2044

),(

2045

"Last Checker Status",

2046

"0 ",

2047

"-2 ",

2048

)

2049

)

2050

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

2051

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

2052

for rows in columns)

2053

for line in range(num_lines))

2054

+ "\n")

2055

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

2056

2057

def test_PrintTable_one_client(self):

2058

with self.capture_stdout_to_buffer() as buffer:

2059

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

2060

expected_output = "\n".join((

2061

"Name Enabled Timeout Last Successful Check",

2062

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

2063

)) + "\n"

2064

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

2065

2066

2067

class TestPropertySetterCmd(TestCommand):

2068

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

2069

2070

def runTest(self):

2071

if not hasattr(self, "command"):

2072

return # Abstract TestCase class

2073

2074

if hasattr(self, "values_to_set"):

2075

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

2076

values_to_get = getattr(self, "values_to_get",

2077

self.values_to_set)

2078

else:

2079

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

2080

values_to_get = self.values_to_get

2081

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

2082

for clientpath in self.bus.clients:

2083

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

2084

Unique())

2085

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

2086

for clientpath in self.bus.clients:

2087

value = (self.bus.clients[clientpath]

2088

[self.propname])

2089

self.assertNotIsInstance(value, Unique)

2090

self.assertEqual(value_to_get, value)

2091

2092

2093

class TestEnableCmd(TestPropertySetterCmd):

2094

command = command.Enable

2095

propname = "Enabled"

2096

values_to_get = [True]

2097

2098

2099

class TestDisableCmd(TestPropertySetterCmd):

2100

command = command.Disable

2101

propname = "Enabled"

2102

values_to_get = [False]

2103

2104

2105

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2106

command = command.BumpTimeout

2107

propname = "LastCheckedOK"

2108

values_to_get = [""]

2109

2110

2111

class TestStartCheckerCmd(TestPropertySetterCmd):

2112

command = command.StartChecker

2113

propname = "CheckerRunning"

2114

values_to_get = [True]

2115

2116

2117

class TestStopCheckerCmd(TestPropertySetterCmd):

2118

command = command.StopChecker

2119

propname = "CheckerRunning"

2120

values_to_get = [False]

2121

2122

2123

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2124

command = command.ApproveByDefault

2125

propname = "ApprovedByDefault"

2126

values_to_get = [True]

2127

2128

2129

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2130

command = command.DenyByDefault

2131

propname = "ApprovedByDefault"

2132

values_to_get = [False]

2133

2134

2135

class TestSetCheckerCmd(TestPropertySetterCmd):

2136

command = command.SetChecker

2137

propname = "Checker"

2138

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

2139

2140

2141

class TestSetHostCmd(TestPropertySetterCmd):

2142

command = command.SetHost

2143

propname = "Host"

2144

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

2145

2146

2147

class TestSetSecretCmd(TestPropertySetterCmd):

2148

command = command.SetSecret

2149

propname = "Secret"

2150

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

2151

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

2152

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

2153

2154

2155

class TestSetTimeoutCmd(TestPropertySetterCmd):

2156

command = command.SetTimeout

2157

propname = "Timeout"

2158

values_to_set = [datetime.timedelta(),

2159

datetime.timedelta(minutes=5),

2160

datetime.timedelta(seconds=1),

2161

datetime.timedelta(weeks=1),

2162

datetime.timedelta(weeks=52)]

2163

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

2164

2165

2166

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2167

command = command.SetExtendedTimeout

2168

propname = "ExtendedTimeout"

2169

values_to_set = [datetime.timedelta(),

2170

datetime.timedelta(minutes=5),

2171

datetime.timedelta(seconds=1),

2172

datetime.timedelta(weeks=1),

2173

datetime.timedelta(weeks=52)]

2174

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

2175

2176

2177

class TestSetIntervalCmd(TestPropertySetterCmd):

2178

command = command.SetInterval

2179

propname = "Interval"

2180

values_to_set = [datetime.timedelta(),

2181

datetime.timedelta(minutes=5),

2182

datetime.timedelta(seconds=1),

2183

datetime.timedelta(weeks=1),

2184

datetime.timedelta(weeks=52)]

2185

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

2186

2187

2188

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2189

command = command.SetApprovalDelay

2190

propname = "ApprovalDelay"

2191

values_to_set = [datetime.timedelta(),

2192

datetime.timedelta(minutes=5),

2193

datetime.timedelta(seconds=1),

2194

datetime.timedelta(weeks=1),

2195

datetime.timedelta(weeks=52)]

2196

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

2197

2198

2199

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2200

command = command.SetApprovalDuration

2201

propname = "ApprovalDuration"

2202

values_to_set = [datetime.timedelta(),

2203

datetime.timedelta(minutes=5),

2204

datetime.timedelta(seconds=1),

2205

datetime.timedelta(weeks=1),

2206

datetime.timedelta(weeks=52)]

2207

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

2208

2209

2210

2211

def should_only_run_tests():

2212

parser = argparse.ArgumentParser(add_help=False)

2213

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

2214

args, unknown_args = parser.parse_known_args()

2215

run_tests = args.check

2216

if run_tests:

2217

# Remove --check argument from sys.argv

2218

sys.argv[1:] = unknown_args

2219

return run_tests

2220

2221

# Add all tests from doctest strings

2222

def load_tests(loader, tests, none):

2223

import doctest

2224

tests.addTests(doctest.DocTestSuite())

2225

return tests

2226

2227

if __name__ == "__main__":

2228

try:

2229

if should_only_run_tests():

2230

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

2231

unittest.main()

2232

else:

2233

main()

2234

finally:

2235

logging.shutdown()