/mandos/release : revision 237.7.615

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8 -*-

3

#

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 program is free software: you can redistribute it and/or modify

10

# it under the terms of the GNU General Public License as published by

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

11

13

# the Free Software Foundation, either version 3 of the License, or

12

14

# (at your option) any later version.

13

15

#

14

# This program is distributed in the hope that it will be useful,

15

# but WITHOUT ANY WARRANTY; without even the implied warranty of

16

# Mandos is distributed in the hope that it will be useful, but

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

# You should have received a copy of the GNU General Public License

20

# along with this program. If not, see <http://www.gnu.org/licenses/>.

21

#

22

# Contact the authors at <mandos@fukt.bsnet.se>.

23

#

24

25

from __future__ import division

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

26

35

import sys

27

import dbus

28

from optparse import OptionParser

36

import argparse

29

37

import locale

30

38

import datetime

31

39

import re

32

40

import os

33

34

locale.setlocale(locale.LC_ALL, u'')

35

36

tablewords = {

37

'Name': u'Name',

38

'Enabled': u'Enabled',

39

'Timeout': u'Timeout',

40

'LastCheckedOK': u'Last Successful Check',

41

'LastApprovalRequest': u'Last Approval Request',

42

'Created': u'Created',

43

'Interval': u'Interval',

44

'Host': u'Host',

45

'Fingerprint': u'Fingerprint',

46

'CheckerRunning': u'Check Is Running',

47

'LastEnabled': u'Last Enabled',

48

'ApprovalPending': u'Approval Is Pending',

49

'ApprovedByDefault': u'Approved By Default',

50

'ApprovalDelay': u"Approval Delay",

51

'ApprovalDuration': u"Approval Duration",

52

'Checker': u'Checker',

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

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

65

locale.setlocale(locale.LC_ALL, "")

66

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"

80

81

82

def main():

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

try:

97

bus = dbus.SystemBus()

98

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

99

dbus_busname, server_dbus_path)

100

mandos_dbus_objc = bus.get_object(dbus_busname,

101

server_dbus_path)

102

except dbus.exceptions.DBusException:

103

log.critical("Could not connect to Mandos server")

104

sys.exit(1)

105

106

mandos_serv = dbus.Interface(mandos_dbus_objc,

107

dbus_interface=server_dbus_interface)

108

mandos_serv_object_manager = dbus.Interface(

109

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

110

111

try:

112

log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,

113

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

114

with SilenceLogger("dbus.proxies"):

115

all_clients = {path: ifs_and_props[client_dbus_interface]

116

for path, ifs_and_props in

117

mandos_serv_object_manager

118

.GetManagedObjects().items()

119

if client_dbus_interface in ifs_and_props}

120

except dbus.exceptions.DBusException as e:

121

log.critical("Failed to access Mandos server through D-Bus:"

122

"\n%s", e)

123

sys.exit(1)

124

125

# Compile dict of (clients: properties) to process

126

clients = {}

127

128

if not clientnames:

129

clients = all_clients

130

else:

131

for name in clientnames:

132

for objpath, properties in all_clients.items():

133

if properties["Name"] == name:

134

clients[objpath] = properties

135

break

136

else:

137

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

138

sys.exit(1)

139

140

# Run all commands on clients

141

commands = commands_from_options(options)

142

for command in commands:

143

command.run(clients, bus, mandos_serv)

144

145

146

def add_command_line_options(parser):

147

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

148

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

149

help="show version number and exit")

150

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

151

help="Select all clients")

152

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

153

help="Print all fields")

154

parser.add_argument("-j", "--dump-json", action="store_true",

155

help="Dump client data in JSON format")

156

enable_disable = parser.add_mutually_exclusive_group()

157

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

158

help="Enable client")

159

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

160

action="store_true",

161

help="disable client")

162

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

163

help="Bump timeout for client")

164

start_stop_checker = parser.add_mutually_exclusive_group()

165

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

166

action="store_true",

167

help="Start checker for client")

168

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

169

action="store_true",

170

help="Stop checker for client")

171

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

172

help="Check if client is enabled")

173

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

174

help="Remove client")

175

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

176

help="Set checker command for client")

177

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

178

help="Set timeout for client")

179

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

180

help="Set extended timeout for client")

181

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

182

help="Set checker interval for client")

183

approve_deny_default = parser.add_mutually_exclusive_group()

184

approve_deny_default.add_argument(

185

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

186

default=None, dest="approved_by_default",

187

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

188

approve_deny_default.add_argument(

189

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

190

dest="approved_by_default",

191

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

192

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

193

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

194

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

195

help="Set duration of one client approval")

196

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

197

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

198

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

199

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

200

approve_deny = parser.add_mutually_exclusive_group()

201

approve_deny.add_argument(

202

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

203

help="Approve any current client request")

204

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

205

help="Deny any current client request")

206

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

207

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

208

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

209

help="Run self-test")

210

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

211

212

213

def string_to_delta(interval):

214

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

215

216

try:

217

return rfc3339_duration_to_delta(interval)

218

except ValueError as e:

219

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

220

' '.join(e.args))

221

return parse_pre_1_6_1_interval(interval)

222

223

224

def rfc3339_duration_to_delta(duration):

225

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

226

227

>>> rfc3339_duration_to_delta("P7D")

228

datetime.timedelta(7)

229

>>> rfc3339_duration_to_delta("PT60S")

230

datetime.timedelta(0, 60)

231

>>> rfc3339_duration_to_delta("PT60M")

232

datetime.timedelta(0, 3600)

233

>>> rfc3339_duration_to_delta("P60M")

234

datetime.timedelta(1680)

235

>>> rfc3339_duration_to_delta("PT24H")

236

datetime.timedelta(1)

237

>>> rfc3339_duration_to_delta("P1W")

238

datetime.timedelta(7)

239

>>> rfc3339_duration_to_delta("PT5M30S")

240

datetime.timedelta(0, 330)

241

>>> rfc3339_duration_to_delta("P1DT3M20S")

242

datetime.timedelta(1, 200)

243

>>> # Can not be empty:

244

>>> rfc3339_duration_to_delta("")

245

Traceback (most recent call last):

246

...

247

ValueError: Invalid RFC 3339 duration: u''

248

>>> # Must start with "P":

249

>>> rfc3339_duration_to_delta("1D")

250

Traceback (most recent call last):

251

...

252

ValueError: Invalid RFC 3339 duration: u'1D'

253

>>> # Must use correct order

254

>>> rfc3339_duration_to_delta("PT1S2M")

255

Traceback (most recent call last):

256

...

257

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

258

>>> # Time needs time marker

259

>>> rfc3339_duration_to_delta("P1H2S")

260

Traceback (most recent call last):

261

...

262

ValueError: Invalid RFC 3339 duration: u'P1H2S'

263

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

264

>>> rfc3339_duration_to_delta("P1D2W")

265

Traceback (most recent call last):

266

...

267

ValueError: Invalid RFC 3339 duration: u'P1D2W'

268

>>> rfc3339_duration_to_delta("P2W2H")

269

Traceback (most recent call last):

270

...

271

ValueError: Invalid RFC 3339 duration: u'P2W2H'

272

"""

273

274

# Parsing an RFC 3339 duration with regular expressions is not

275

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

276

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

277

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

278

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

279

# avoid excessive use of external libraries.

280

281

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

282

Token = collections.namedtuple("Token", (

283

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

284

# a "group" containing digits

285

"value", # datetime.timedelta or None

286

"followers")) # Tokens valid after this token

287

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

288

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

289

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

290

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

291

datetime.timedelta(seconds=1),

292

frozenset((token_end, )))

293

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

294

datetime.timedelta(minutes=1),

295

frozenset((token_second, token_end)))

296

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

297

datetime.timedelta(hours=1),

298

frozenset((token_minute, token_end)))

299

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

300

None,

301

frozenset((token_hour, token_minute,

302

token_second)))

303

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

304

datetime.timedelta(days=1),

305

frozenset((token_time, token_end)))

306

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

307

datetime.timedelta(weeks=4),

308

frozenset((token_day, token_end)))

309

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

310

datetime.timedelta(weeks=52),

311

frozenset((token_month, token_end)))

312

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

313

datetime.timedelta(weeks=1),

314

frozenset((token_end, )))

315

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

316

frozenset((token_year, token_month,

317

token_day, token_time,

318

token_week)))

319

# Define starting values:

320

# Value so far

321

value = datetime.timedelta()

322

found_token = None

323

# Following valid tokens

324

followers = frozenset((token_duration, ))

325

# String left to parse

326

s = duration

327

# Loop until end token is found

328

while found_token is not token_end:

329

# Search for any currently valid tokens

330

for token in followers:

331

match = token.regexp.match(s)

332

if match is not None:

333

# Token found

334

if token.value is not None:

335

# Value found, parse digits

336

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

337

# Add to value so far

338

value += factor * token.value

339

# Strip token from string

340

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

341

# Go to found token

342

found_token = token

343

# Set valid next tokens

344

followers = found_token.followers

345

break

346

else:

347

# No currently valid tokens were found

348

raise ValueError("Invalid RFC 3339 duration: {!r}"

349

.format(duration))

350

# End token found

351

return value

352

353

354

def parse_pre_1_6_1_interval(interval):

355

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

356

and return a datetime.timedelta

357

358

>>> parse_pre_1_6_1_interval('7d')

359

datetime.timedelta(7)

360

>>> parse_pre_1_6_1_interval('60s')

361

datetime.timedelta(0, 60)

362

>>> parse_pre_1_6_1_interval('60m')

363

datetime.timedelta(0, 3600)

364

>>> parse_pre_1_6_1_interval('24h')

365

datetime.timedelta(1)

366

>>> parse_pre_1_6_1_interval('1w')

367

datetime.timedelta(7)

368

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

369

datetime.timedelta(0, 330)

370

>>> parse_pre_1_6_1_interval('')

371

datetime.timedelta(0)

372

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

373

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

374

datetime.timedelta(2, 480, 18000)

375

376

"""

377

378

value = datetime.timedelta(0)

379

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

380

381

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

382

if suffix == "d":

383

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

384

elif suffix == "s":

385

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

386

elif suffix == "m":

387

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

388

elif suffix == "h":

389

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

390

elif suffix == "w":

391

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

392

elif suffix == "":

393

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

394

return value

395

396

397

def check_option_syntax(parser, options):

398

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

399

argparse"""

400

401

def has_actions(options):

402

return any((options.enable,

403

options.disable,

404

options.bump_timeout,

405

options.start_checker,

406

options.stop_checker,

407

options.is_enabled,

408

options.remove,

409

options.checker is not None,

410

options.timeout is not None,

411

options.extended_timeout is not None,

412

options.interval is not None,

413

options.approved_by_default is not None,

414

options.approval_delay is not None,

415

options.approval_duration is not None,

416

options.host is not None,

417

options.secret is not None,

418

options.approve,

419

options.deny))

420

421

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

422

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

423

if options.verbose and has_actions(options):

424

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

425

if options.dump_json and (options.verbose

426

or has_actions(options)):

427

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

428

if options.all and not has_actions(options):

429

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

430

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

431

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

432

if options.remove:

433

options.remove = False

434

if has_actions(options) and not options.deny:

435

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

436

options.remove = True

437

438

439

class SilenceLogger(object):

440

"Simple context manager to silence a particular logger"

441

def __init__(self, loggername):

442

self.logger = logging.getLogger(loggername)

443

444

def __enter__(self):

445

self.logger.addFilter(self.nullfilter)

446

return self

447

448

class NullFilter(logging.Filter):

449

def filter(self, record):

450

return False

451

452

nullfilter = NullFilter()

453

454

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

455

self.logger.removeFilter(self.nullfilter)

456

457

458

def commands_from_options(options):

459

460

commands = []

461

462

if options.is_enabled:

463

commands.append(IsEnabledCmd())

464

465

if options.approve:

466

commands.append(ApproveCmd())

467

468

if options.deny:

469

commands.append(DenyCmd())

470

471

if options.remove:

472

commands.append(RemoveCmd())

473

474

if options.dump_json:

475

commands.append(DumpJSONCmd())

476

477

if options.enable:

478

commands.append(EnableCmd())

479

480

if options.disable:

481

commands.append(DisableCmd())

482

483

if options.bump_timeout:

484

commands.append(BumpTimeoutCmd())

485

486

if options.start_checker:

487

commands.append(StartCheckerCmd())

488

489

if options.stop_checker:

490

commands.append(StopCheckerCmd())

491

492

if options.approved_by_default is not None:

493

if options.approved_by_default:

494

commands.append(ApproveByDefaultCmd())

495

else:

496

commands.append(DenyByDefaultCmd())

497

498

if options.checker is not None:

499

commands.append(SetCheckerCmd(options.checker))

500

501

if options.host is not None:

502

commands.append(SetHostCmd(options.host))

503

504

if options.secret is not None:

505

commands.append(SetSecretCmd(options.secret))

506

507

if options.timeout is not None:

508

commands.append(SetTimeoutCmd(options.timeout))

509

510

if options.extended_timeout:

511

commands.append(

512

SetExtendedTimeoutCmd(options.extended_timeout))

513

514

if options.interval is not None:

515

commands.append(SetIntervalCmd(options.interval))

516

517

if options.approval_delay is not None:

518

commands.append(SetApprovalDelayCmd(options.approval_delay))

519

520

if options.approval_duration is not None:

521

commands.append(

522

SetApprovalDurationCmd(options.approval_duration))

523

524

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

525

# optionally verbosely

526

if not commands:

527

commands.append(PrintTableCmd(verbose=options.verbose))

528

529

return commands

530

531

532

class Command(object):

533

"""Abstract class for commands"""

534

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

535

"""Normal commands should implement run_on_one_client(), but

536

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

537

can override this run() method instead."""

538

self.mandos = mandos

539

for clientpath, properties in clients.items():

540

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

541

dbus_busname, str(clientpath))

542

client = bus.get_object(dbus_busname, clientpath)

543

self.run_on_one_client(client, properties)

544

545

546

class IsEnabledCmd(Command):

547

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

548

client, properties = next(iter(clients.items()))

549

if self.is_enabled(client, properties):

550

sys.exit(0)

551

sys.exit(1)

552

def is_enabled(self, client, properties):

553

return properties["Enabled"]

554

555

556

class ApproveCmd(Command):

557

def run_on_one_client(self, client, properties):

558

log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,

559

client.__dbus_object_path__, client_dbus_interface)

560

client.Approve(dbus.Boolean(True),

561

dbus_interface=client_dbus_interface)

562

563

564

class DenyCmd(Command):

565

def run_on_one_client(self, client, properties):

566

log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,

567

client.__dbus_object_path__, client_dbus_interface)

568

client.Approve(dbus.Boolean(False),

569

dbus_interface=client_dbus_interface)

570

571

572

class RemoveCmd(Command):

573

def run_on_one_client(self, client, properties):

574

log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,

575

server_dbus_path, server_dbus_interface,

576

str(client.__dbus_object_path__))

577

self.mandos.RemoveClient(client.__dbus_object_path__)

578

579

580

class OutputCmd(Command):

581

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

582

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

583

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

584

"Fingerprint", "CheckerRunning", "LastEnabled",

585

"ApprovalPending", "ApprovedByDefault",

586

"LastApprovalRequest", "ApprovalDelay",

587

"ApprovalDuration", "Checker", "ExtendedTimeout",

588

"Expires", "LastCheckerStatus")

589

590

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

591

print(self.output(clients.values()))

592

593

def output(self, clients):

594

raise NotImplementedError()

595

596

597

class DumpJSONCmd(OutputCmd):

598

def output(self, clients):

599

data = {client["Name"]:

600

{key: self.dbus_boolean_to_bool(client[key])

601

for key in self.all_keywords}

602

for client in clients}

603

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

604

605

@staticmethod

606

def dbus_boolean_to_bool(value):

607

if isinstance(value, dbus.Boolean):

608

value = bool(value)

609

return value

610

611

612

class PrintTableCmd(OutputCmd):

613

def __init__(self, verbose=False):

614

self.verbose = verbose

615

616

def output(self, clients):

617

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

618

"LastCheckedOK")

619

keywords = default_keywords

620

if self.verbose:

621

keywords = self.all_keywords

622

return str(self.TableOfClients(clients, keywords))

623

624

class TableOfClients(object):

625

tableheaders = {

626

"Name": "Name",

627

"Enabled": "Enabled",

628

"Timeout": "Timeout",

629

"LastCheckedOK": "Last Successful Check",

630

"LastApprovalRequest": "Last Approval Request",

631

"Created": "Created",

632

"Interval": "Interval",

633

"Host": "Host",

634

"Fingerprint": "Fingerprint",

635

"KeyID": "Key ID",

636

"CheckerRunning": "Check Is Running",

637

"LastEnabled": "Last Enabled",

638

"ApprovalPending": "Approval Is Pending",

639

"ApprovedByDefault": "Approved By Default",

640

"ApprovalDelay": "Approval Delay",

641

"ApprovalDuration": "Approval Duration",

642

"Checker": "Checker",

643

"ExtendedTimeout": "Extended Timeout",

644

"Expires": "Expires",

645

"LastCheckerStatus": "Last Checker Status",

646

}

647

648

def __init__(self, clients, keywords):

649

self.clients = clients

650

self.keywords = keywords

651

652

def __str__(self):

653

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

654

655

if sys.version_info.major == 2:

656

__unicode__ = __str__

657

def __str__(self):

658

return str(self).encode(locale.getpreferredencoding())

659

660

def rows(self):

661

format_string = self.row_formatting_string()

662

rows = [self.header_line(format_string)]

663

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

664

for client in self.clients)

665

return rows

666

667

def row_formatting_string(self):

668

"Format string used to format table rows"

669

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

670

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

671

*(len(self.string_from_client(client, key))

672

for client in self.clients)),

673

key=key)

674

for key in self.keywords)

675

676

def string_from_client(self, client, key):

677

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

678

679

@classmethod

680

def valuetostring(cls, value, keyword):

681

if isinstance(value, dbus.Boolean):

682

return "Yes" if value else "No"

683

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

684

"ApprovalDuration", "ExtendedTimeout"):

685

return cls.milliseconds_to_string(value)

686

return str(value)

687

688

def header_line(self, format_string):

689

return format_string.format(**self.tableheaders)

690

691

def client_line(self, client, format_string):

692

return format_string.format(

693

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

694

for key in self.keywords})

695

696

@staticmethod

697

def milliseconds_to_string(ms):

698

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

699

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

700

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

701

if td.days else "",

702

hours=td.seconds // 3600,

703

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

704

seconds=td.seconds % 60))

705

706

707

class PropertyCmd(Command):

708

"""Abstract class for Actions for setting one client property"""

709

710

def run_on_one_client(self, client, properties):

711

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

712

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

713

client.__dbus_object_path__,

714

dbus.PROPERTIES_IFACE, client_dbus_interface,

715

self.propname, self.value_to_set

716

if not isinstance(self.value_to_set, dbus.Boolean)

717

else bool(self.value_to_set))

718

client.Set(client_dbus_interface, self.propname,

719

self.value_to_set,

720

dbus_interface=dbus.PROPERTIES_IFACE)

721

722

@property

723

def propname(self):

724

raise NotImplementedError()

725

726

727

class EnableCmd(PropertyCmd):

728

propname = "Enabled"

729

value_to_set = dbus.Boolean(True)

730

731

732

class DisableCmd(PropertyCmd):

733

propname = "Enabled"

734

value_to_set = dbus.Boolean(False)

735

736

737

class BumpTimeoutCmd(PropertyCmd):

738

propname = "LastCheckedOK"

739

value_to_set = ""

740

741

742

class StartCheckerCmd(PropertyCmd):

743

propname = "CheckerRunning"

744

value_to_set = dbus.Boolean(True)

745

746

747

class StopCheckerCmd(PropertyCmd):

748

propname = "CheckerRunning"

749

value_to_set = dbus.Boolean(False)

750

751

752

class ApproveByDefaultCmd(PropertyCmd):

753

propname = "ApprovedByDefault"

754

value_to_set = dbus.Boolean(True)

755

756

757

class DenyByDefaultCmd(PropertyCmd):

758

propname = "ApprovedByDefault"

759

value_to_set = dbus.Boolean(False)

760

761

762

class PropertyValueCmd(PropertyCmd):

763

"""Abstract class for PropertyCmd recieving a value as argument"""

764

def __init__(self, value):

765

self.value_to_set = value

766

767

768

class SetCheckerCmd(PropertyValueCmd):

769

propname = "Checker"

770

771

772

class SetHostCmd(PropertyValueCmd):

773

propname = "Host"

774

775

776

class SetSecretCmd(PropertyValueCmd):

777

propname = "Secret"

778

779

@property

780

def value_to_set(self):

781

return self._vts

782

783

@value_to_set.setter

784

def value_to_set(self, value):

785

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

786

self._vts = value.read()

787

value.close()

788

789

790

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

791

"""Abstract class for PropertyValueCmd taking a value argument as

792

a datetime.timedelta() but should store it as milliseconds."""

793

794

@property

795

def value_to_set(self):

796

return self._vts

797

798

@value_to_set.setter

799

def value_to_set(self, value):

800

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

801

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

802

803

804

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

805

propname = "Timeout"

806

807

808

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

809

propname = "ExtendedTimeout"

810

811

812

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

813

propname = "Interval"

814

815

816

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

817

propname = "ApprovalDelay"

818

819

820

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

821

propname = "ApprovalDuration"

822

823

824

825

class Test_string_to_delta(unittest.TestCase):

826

def test_handles_basic_rfc3339(self):

827

self.assertEqual(string_to_delta("PT0S"),

828

datetime.timedelta())

829

self.assertEqual(string_to_delta("P0D"),

830

datetime.timedelta())

831

self.assertEqual(string_to_delta("PT1S"),

832

datetime.timedelta(0, 1))

833

self.assertEqual(string_to_delta("PT2H"),

834

datetime.timedelta(0, 7200))

835

836

def test_falls_back_to_pre_1_6_1_with_warning(self):

837

# assertLogs only exists in Python 3.4

838

if hasattr(self, "assertLogs"):

839

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

840

value = string_to_delta("2h")

841

else:

842

class WarningFilter(logging.Filter):

843

"""Don't show, but record the presence of, warnings"""

844

def filter(self, record):

845

is_warning = record.levelno >= logging.WARNING

846

self.found = is_warning or getattr(self, "found",

847

False)

848

return not is_warning

849

warning_filter = WarningFilter()

850

log.addFilter(warning_filter)

851

try:

852

value = string_to_delta("2h")

853

finally:

854

log.removeFilter(warning_filter)

855

self.assertTrue(getattr(warning_filter, "found", False))

856

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

857

858

859

class Test_check_option_syntax(unittest.TestCase):

860

def setUp(self):

861

self.parser = argparse.ArgumentParser()

862

add_command_line_options(self.parser)

863

864

def test_actions_requires_client_or_all(self):

865

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

866

options = self.parser.parse_args()

867

setattr(options, action, value)

868

with self.assertParseError():

869

self.check_option_syntax(options)

870

871

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

872

# check_option_syntax()

873

actions = {

874

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

875

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

876

# never used

877

"enable": True,

878

"disable": True,

879

"bump_timeout": True,

880

"start_checker": True,

881

"stop_checker": True,

882

"is_enabled": True,

883

"remove": True,

884

"checker": "x",

885

"timeout": datetime.timedelta(),

886

"extended_timeout": datetime.timedelta(),

887

"interval": datetime.timedelta(),

888

"approved_by_default": True,

889

"approval_delay": datetime.timedelta(),

890

"approval_duration": datetime.timedelta(),

891

"host": "x",

892

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

893

"approve": True,

894

"deny": True,

53

895

}

54

defaultkeywords = ('Name', 'Enabled', 'Timeout', 'LastCheckedOK')

55

domain = 'se.bsnet.fukt'

56

busname = domain + '.Mandos'

57

server_path = '/'

58

server_interface = domain + '.Mandos'

59

client_interface = domain + '.Mandos.Client'

60

version = "1.2"

61

62

def timedelta_to_milliseconds(td):

63

"Convert a datetime.timedelta object to milliseconds"

64

return ((td.days * 24 * 60 * 60 * 1000)

65

+ (td.seconds * 1000)

66

+ (td.microseconds // 1000))

67

68

def milliseconds_to_string(ms):

69

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

70

return (u"%(days)s%(hours)02d:%(minutes)02d:%(seconds)02d"

71

% { "days": "%dT" % td.days if td.days else "",

72

"hours": td.seconds // 3600,

73

"minutes": (td.seconds % 3600) // 60,

74

"seconds": td.seconds % 60,

75

})

76

77

78

def string_to_delta(interval):

79

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

80

81

>>> string_to_delta('7d')

82

datetime.timedelta(7)

83

>>> string_to_delta('60s')

84

datetime.timedelta(0, 60)

85

>>> string_to_delta('60m')

86

datetime.timedelta(0, 3600)

87

>>> string_to_delta('24h')

88

datetime.timedelta(1)

89

>>> string_to_delta(u'1w')

90

datetime.timedelta(7)

91

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

92

datetime.timedelta(0, 330)

93

"""

94

timevalue = datetime.timedelta(0)

95

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

96

97

for s in regexp.findall(interval):

98

try:

99

suffix = unicode(s[-1])

100

value = int(s[:-1])

101

if suffix == u"d":

102

delta = datetime.timedelta(value)

103

elif suffix == u"s":

104

delta = datetime.timedelta(0, value)

105

elif suffix == u"m":

106

delta = datetime.timedelta(0, 0, 0, 0, value)

107

elif suffix == u"h":

108

delta = datetime.timedelta(0, 0, 0, 0, 0, value)

109

elif suffix == u"w":

110

delta = datetime.timedelta(0, 0, 0, 0, 0, 0, value)

111

else:

112

raise ValueError

113

except (ValueError, IndexError):

114

raise ValueError

115

timevalue += delta

116

return timevalue

117

118

def print_clients(clients, keywords):

119

def valuetostring(value, keyword):

120

if type(value) is dbus.Boolean:

121

return u"Yes" if value else u"No"

122

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

123

u"ApprovalDuration"):

124

return milliseconds_to_string(value)

125

return unicode(value)

126

127

# Create format string to print table rows

128

format_string = u' '.join(u'%%-%ds' %

129

max(len(tablewords[key]),

130

max(len(valuetostring(client[key],

131

key))

132

for client in

133

clients))

134

for key in keywords)

135

# Print header line

136

print format_string % tuple(tablewords[key] for key in keywords)

137

for client in clients:

138

print format_string % tuple(valuetostring(client[key], key)

139

for key in keywords)

140

141

def has_actions(options):

142

return any((options.enable,

143

options.disable,

144

options.bump_timeout,

145

options.start_checker,

146

options.stop_checker,

147

options.is_enabled,

148

options.remove,

149

options.checker is not None,

150

options.timeout is not None,

151

options.interval is not None,

152

options.approved_by_default is not None,

153

options.approval_delay is not None,

154

options.approval_duration is not None,

155

options.host is not None,

156

options.secret is not None,

157

options.approve,

158

options.deny))

159

160

def main():

161

parser = OptionParser(version = "%%prog %s" % version)

162

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

163

help="Select all clients")

164

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

165

help="Print all fields")

166

parser.add_option("-e", "--enable", action="store_true",

167

help="Enable client")

168

parser.add_option("-d", "--disable", action="store_true",

169

help="disable client")

170

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

171

help="Bump timeout for client")

172

parser.add_option("--start-checker", action="store_true",

173

help="Start checker for client")

174

parser.add_option("--stop-checker", action="store_true",

175

help="Stop checker for client")

176

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

177

help="Check if client is enabled")

178

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

179

help="Remove client")

180

parser.add_option("-c", "--checker", type="string",

181

help="Set checker command for client")

182

parser.add_option("-t", "--timeout", type="string",

183

help="Set timeout for client")

184

parser.add_option("-i", "--interval", type="string",

185

help="Set checker interval for client")

186

parser.add_option("--approve-by-default", action="store_true",

187

dest=u"approved_by_default",

188

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

189

parser.add_option("--deny-by-default", action="store_false",

190

dest=u"approved_by_default",

191

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

192

parser.add_option("--approval-delay", type="string",

193

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

194

parser.add_option("--approval-duration", type="string",

195

help="Set duration of one client approval")

196

parser.add_option("-H", "--host", type="string",

197

help="Set host for client")

198

parser.add_option("-s", "--secret", type="string",

199

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

200

parser.add_option("-A", "--approve", action="store_true",

201

help="Approve any current client request")

202

parser.add_option("-D", "--deny", action="store_true",

203

help="Deny any current client request")

204

options, client_names = parser.parse_args()

205

206

if has_actions(options) and not client_names and not options.all:

207

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

208

if options.verbose and has_actions(options):

209

parser.error('--verbose can only be used alone or with'

210

' --all.')

211

if options.all and not has_actions(options):

212

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

213

214

try:

215

bus = dbus.SystemBus()

216

mandos_dbus_objc = bus.get_object(busname, server_path)

217

except dbus.exceptions.DBusException:

218

print >> sys.stderr, "Could not connect to Mandos server"

219

sys.exit(1)

220

221

mandos_serv = dbus.Interface(mandos_dbus_objc,

222

dbus_interface = server_interface)

223

224

#block stderr since dbus library prints to stderr

896

897

@contextlib.contextmanager

898

def assertParseError(self):

899

with self.assertRaises(SystemExit) as e:

900

with self.temporarily_suppress_stderr():

901

yield

902

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

903

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

904

# /argparse.html#exiting-methods

905

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

906

907

@staticmethod

908

@contextlib.contextmanager

909

def temporarily_suppress_stderr():

225

910

null = os.open(os.path.devnull, os.O_RDWR)

226

911

stderrcopy = os.dup(sys.stderr.fileno())

227

912

os.dup2(null, sys.stderr.fileno())

228

913

os.close(null)

229

914

try:

230

try:

231

mandos_clients = mandos_serv.GetAllClientsWithProperties()

232

finally:

233

#restore stderr

234

os.dup2(stderrcopy, sys.stderr.fileno())

235

os.close(stderrcopy)

236

except dbus.exceptions.DBusException, e:

237

print >> sys.stderr, "Access denied: Accessing mandos server through dbus."

238

sys.exit(1)

239

240

# Compile dict of (clients: properties) to process

241

clients={}

242

243

if options.all or not client_names:

244

clients = dict((bus.get_object(busname, path), properties)

245

for path, properties in

246

mandos_clients.iteritems())

247

else:

248

for name in client_names:

249

for path, client in mandos_clients.iteritems():

250

if client['Name'] == name:

251

client_objc = bus.get_object(busname, path)

252

clients[client_objc] = client

253

break

254

else:

255

print >> sys.stderr, "Client not found on server: %r" % name

256

sys.exit(1)

257

258

if not has_actions(options) and clients:

259

if options.verbose:

260

keywords = ('Name', 'Enabled', 'Timeout',

261

'LastCheckedOK', 'Created', 'Interval',

262

'Host', 'Fingerprint', 'CheckerRunning',

263

'LastEnabled', 'ApprovalPending',

264

'ApprovedByDefault',

265

'LastApprovalRequest', 'ApprovalDelay',

266

'ApprovalDuration', 'Checker')

267

else:

268

keywords = defaultkeywords

269

270

print_clients(clients.values(), keywords)

271

else:

272

# Process each client in the list by all selected options

273

for client in clients:

274

if options.remove:

275

mandos_serv.RemoveClient(client.__dbus_object_path__)

276

if options.enable:

277

client.Enable(dbus_interface=client_interface)

278

if options.disable:

279

client.Disable(dbus_interface=client_interface)

280

if options.bump_timeout:

281

client.CheckedOK(dbus_interface=client_interface)

282

if options.start_checker:

283

client.StartChecker(dbus_interface=client_interface)

284

if options.stop_checker:

285

client.StopChecker(dbus_interface=client_interface)

286

if options.is_enabled:

287

sys.exit(0 if client.Get(client_interface,

288

u"Enabled",

289

dbus_interface=dbus.PROPERTIES_IFACE)

290

else 1)

291

if options.checker:

292

client.Set(client_interface, u"Checker", options.checker,

293

dbus_interface=dbus.PROPERTIES_IFACE)

294

if options.host:

295

client.Set(client_interface, u"Host", options.host,

296

dbus_interface=dbus.PROPERTIES_IFACE)

297

if options.interval:

298

client.Set(client_interface, u"Interval",

299

timedelta_to_milliseconds

300

(string_to_delta(options.interval)),

301

dbus_interface=dbus.PROPERTIES_IFACE)

302

if options.approval_delay:

303

client.Set(client_interface, u"ApprovalDelay",

304

timedelta_to_milliseconds

305

(string_to_delta(options.

306

approval_delay)),

307

dbus_interface=dbus.PROPERTIES_IFACE)

308

if options.approval_duration:

309

client.Set(client_interface, u"ApprovalDuration",

310

timedelta_to_milliseconds

311

(string_to_delta(options.

312

approval_duration)),

313

dbus_interface=dbus.PROPERTIES_IFACE)

314

if options.timeout:

315

client.Set(client_interface, u"Timeout",

316

timedelta_to_milliseconds

317

(string_to_delta(options.timeout)),

318

dbus_interface=dbus.PROPERTIES_IFACE)

319

if options.secret:

320

client.Set(client_interface, u"Secret",

321

dbus.ByteArray(open(options.secret,

322

u'rb').read()),

323

dbus_interface=dbus.PROPERTIES_IFACE)

324

if options.approved_by_default is not None:

325

client.Set(client_interface, u"ApprovedByDefault",

326

dbus.Boolean(options

327

.approved_by_default),

328

dbus_interface=dbus.PROPERTIES_IFACE)

329

if options.approve:

330

client.Approve(dbus.Boolean(True),

331

dbus_interface=client_interface)

332

elif options.deny:

333

client.Approve(dbus.Boolean(False),

334

dbus_interface=client_interface)

335

336

if __name__ == '__main__':

337

main()

915

yield

916

finally:

917

# restore stderr

918

os.dup2(stderrcopy, sys.stderr.fileno())

919

os.close(stderrcopy)

920

921

def check_option_syntax(self, options):

922

check_option_syntax(self.parser, options)

923

924

def test_actions_conflicts_with_verbose(self):

925

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

926

options = self.parser.parse_args()

927

setattr(options, action, value)

928

options.verbose = True

929

with self.assertParseError():

930

self.check_option_syntax(options)

931

932

def test_dump_json_conflicts_with_verbose(self):

933

options = self.parser.parse_args()

934

options.dump_json = True

935

options.verbose = True

936

with self.assertParseError():

937

self.check_option_syntax(options)

938

939

def test_dump_json_conflicts_with_action(self):

940

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

941

options = self.parser.parse_args()

942

setattr(options, action, value)

943

options.dump_json = True

944

with self.assertParseError():

945

self.check_option_syntax(options)

946

947

def test_all_can_not_be_alone(self):

948

options = self.parser.parse_args()

949

options.all = True

950

with self.assertParseError():

951

self.check_option_syntax(options)

952

953

def test_all_is_ok_with_any_action(self):

954

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

955

options = self.parser.parse_args()

956

setattr(options, action, value)

957

options.all = True

958

self.check_option_syntax(options)

959

960

def test_is_enabled_fails_without_client(self):

961

options = self.parser.parse_args()

962

options.is_enabled = True

963

with self.assertParseError():

964

self.check_option_syntax(options)

965

966

def test_is_enabled_works_with_one_client(self):

967

options = self.parser.parse_args()

968

options.is_enabled = True

969

options.client = ["foo"]

970

self.check_option_syntax(options)

971

972

def test_is_enabled_fails_with_two_clients(self):

973

options = self.parser.parse_args()

974

options.is_enabled = True

975

options.client = ["foo", "barbar"]

976

with self.assertParseError():

977

self.check_option_syntax(options)

978

979

def test_remove_can_only_be_combined_with_action_deny(self):

980

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

981

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

982

continue

983

options = self.parser.parse_args()

984

setattr(options, action, value)

985

options.all = True

986

options.remove = True

987

with self.assertParseError():

988

self.check_option_syntax(options)

989

990

991

class Test_SilenceLogger(unittest.TestCase):

992

loggername = "mandos-ctl.Test_SilenceLogger"

993

log = logging.getLogger(loggername)

994

log.propagate = False

995

log.addHandler(logging.NullHandler())

996

997

def setUp(self):

998

self.counting_filter = self.CountingFilter()

999

1000

class CountingFilter(logging.Filter):

1001

"Count number of records"

1002

count = 0

1003

def filter(self, record):

1004

self.count += 1

1005

return True

1006

1007

def test_should_filter_records_only_when_active(self):

1008

try:

1009

with SilenceLogger(self.loggername):

1010

self.log.addFilter(self.counting_filter)

1011

self.log.info("Filtered log message 1")

1012

self.log.info("Non-filtered message 2")

1013

self.log.info("Non-filtered message 3")

1014

finally:

1015

self.log.removeFilter(self.counting_filter)

1016

self.assertEqual(self.counting_filter.count, 2)

1017

1018

1019

class Test_commands_from_options(unittest.TestCase):

1020

def setUp(self):

1021

self.parser = argparse.ArgumentParser()

1022

add_command_line_options(self.parser)

1023

1024

def test_is_enabled(self):

1025

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

1026

IsEnabledCmd)

1027

1028

def assert_command_from_args(self, args, command_cls,

1029

**cmd_attrs):

1030

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

1031

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

1032

options = self.parser.parse_args(args)

1033

check_option_syntax(self.parser, options)

1034

commands = commands_from_options(options)

1035

self.assertEqual(len(commands), 1)

1036

command = commands[0]

1037

self.assertIsInstance(command, command_cls)

1038

for key, value in cmd_attrs.items():

1039

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

1040

1041

def test_is_enabled_short(self):

1042

self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)

1043

1044

def test_approve(self):

1045

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

1046

ApproveCmd)

1047

1048

def test_approve_short(self):

1049

self.assert_command_from_args(["-A", "foo"], ApproveCmd)

1050

1051

def test_deny(self):

1052

self.assert_command_from_args(["--deny", "foo"], DenyCmd)

1053

1054

def test_deny_short(self):

1055

self.assert_command_from_args(["-D", "foo"], DenyCmd)

1056

1057

def test_remove(self):

1058

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

1059

RemoveCmd)

1060

1061

def test_deny_before_remove(self):

1062

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

1063

"foo"])

1064

check_option_syntax(self.parser, options)

1065

commands = commands_from_options(options)

1066

self.assertEqual(len(commands), 2)

1067

self.assertIsInstance(commands[0], DenyCmd)

1068

self.assertIsInstance(commands[1], RemoveCmd)

1069

1070

def test_deny_before_remove_reversed(self):

1071

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

1072

"--all"])

1073

check_option_syntax(self.parser, options)

1074

commands = commands_from_options(options)

1075

self.assertEqual(len(commands), 2)

1076

self.assertIsInstance(commands[0], DenyCmd)

1077

self.assertIsInstance(commands[1], RemoveCmd)

1078

1079

def test_remove_short(self):

1080

self.assert_command_from_args(["-r", "foo"], RemoveCmd)

1081

1082

def test_dump_json(self):

1083

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

1084

1085

def test_enable(self):

1086

self.assert_command_from_args(["--enable", "foo"], EnableCmd)

1087

1088

def test_enable_short(self):

1089

self.assert_command_from_args(["-e", "foo"], EnableCmd)

1090

1091

def test_disable(self):

1092

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

1093

DisableCmd)

1094

1095

def test_disable_short(self):

1096

self.assert_command_from_args(["-d", "foo"], DisableCmd)

1097

1098

def test_bump_timeout(self):

1099

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

1100

BumpTimeoutCmd)

1101

1102

def test_bump_timeout_short(self):

1103

self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)

1104

1105

def test_start_checker(self):

1106

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

1107

StartCheckerCmd)

1108

1109

def test_stop_checker(self):

1110

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

1111

StopCheckerCmd)

1112

1113

def test_approve_by_default(self):

1114

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

1115

ApproveByDefaultCmd)

1116

1117

def test_deny_by_default(self):

1118

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

1119

DenyByDefaultCmd)

1120

1121

def test_checker(self):

1122

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

1123

SetCheckerCmd, value_to_set=":")

1124

1125

def test_checker_empty(self):

1126

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

1127

SetCheckerCmd, value_to_set="")

1128

1129

def test_checker_short(self):

1130

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

1131

SetCheckerCmd, value_to_set=":")

1132

1133

def test_host(self):

1134

self.assert_command_from_args(["--host", "foo.example.org",

1135

"foo"], SetHostCmd,

1136

value_to_set="foo.example.org")

1137

1138

def test_host_short(self):

1139

self.assert_command_from_args(["-H", "foo.example.org",

1140

"foo"], SetHostCmd,

1141

value_to_set="foo.example.org")

1142

1143

def test_secret_devnull(self):

1144

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

1145

"foo"], SetSecretCmd,

1146

value_to_set=b"")

1147

1148

def test_secret_tempfile(self):

1149

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

1150

value = b"secret\0xyzzy\nbar"

1151

f.write(value)

1152

f.seek(0)

1153

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

1154

"foo"], SetSecretCmd,

1155

value_to_set=value)

1156

1157

def test_secret_devnull_short(self):

1158

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

1159

SetSecretCmd, value_to_set=b"")

1160

1161

def test_secret_tempfile_short(self):

1162

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

1163

value = b"secret\0xyzzy\nbar"

1164

f.write(value)

1165

f.seek(0)

1166

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

1167

SetSecretCmd,

1168

value_to_set=value)

1169

1170

def test_timeout(self):

1171

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

1172

SetTimeoutCmd,

1173

value_to_set=300000)

1174

1175

def test_timeout_short(self):

1176

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

1177

SetTimeoutCmd,

1178

value_to_set=300000)

1179

1180

def test_extended_timeout(self):

1181

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

1182

"foo"],

1183

SetExtendedTimeoutCmd,

1184

value_to_set=900000)

1185

1186

def test_interval(self):

1187

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

1188

SetIntervalCmd,

1189

value_to_set=120000)

1190

1191

def test_interval_short(self):

1192

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

1193

SetIntervalCmd,

1194

value_to_set=120000)

1195

1196

def test_approval_delay(self):

1197

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

1198

"foo"], SetApprovalDelayCmd,

1199

value_to_set=30000)

1200

1201

def test_approval_duration(self):

1202

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

1203

"foo"], SetApprovalDurationCmd,

1204

value_to_set=1000)

1205

1206

def test_print_table(self):

1207

self.assert_command_from_args([], PrintTableCmd,

1208

verbose=False)

1209

1210

def test_print_table_verbose(self):

1211

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

1212

verbose=True)

1213

1214

def test_print_table_verbose_short(self):

1215

self.assert_command_from_args(["-v"], PrintTableCmd,

1216

verbose=True)

1217

1218

1219

class TestCmd(unittest.TestCase):

1220

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

1221

1222

def setUp(self):

1223

testcase = self

1224

class MockClient(object):

1225

def __init__(self, name, **attributes):

1226

self.__dbus_object_path__ = "/clients/{}".format(name)

1227

self.attributes = attributes

1228

self.attributes["Name"] = name

1229

self.calls = []

1230

def Set(self, interface, propname, value, dbus_interface):

1231

testcase.assertEqual(interface, client_dbus_interface)

1232

testcase.assertEqual(dbus_interface,

1233

dbus.PROPERTIES_IFACE)

1234

self.attributes[propname] = value

1235

def Get(self, interface, propname, dbus_interface):

1236

testcase.assertEqual(interface, client_dbus_interface)

1237

testcase.assertEqual(dbus_interface,

1238

dbus.PROPERTIES_IFACE)

1239

return self.attributes[propname]

1240

def Approve(self, approve, dbus_interface):

1241

testcase.assertEqual(dbus_interface,

1242

client_dbus_interface)

1243

self.calls.append(("Approve", (approve,

1244

dbus_interface)))

1245

self.client = MockClient(

1246

"foo",

1247

KeyID=("92ed150794387c03ce684574b1139a65"

1248

"94a34f895daaaf09fd8ea90a27cddb12"),

1249

Secret=b"secret",

1250

Host="foo.example.org",

1251

Enabled=dbus.Boolean(True),

1252

Timeout=300000,

1253

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

1254

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

1255

Interval=120000,

1256

Fingerprint=("778827225BA7DE539C5A"

1257

"7CFA59CFF7CDBD9A5920"),

1258

CheckerRunning=dbus.Boolean(False),

1259

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

1260

ApprovalPending=dbus.Boolean(False),

1261

ApprovedByDefault=dbus.Boolean(True),

1262

LastApprovalRequest="",

1263

ApprovalDelay=0,

1264

ApprovalDuration=1000,

1265

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

1266

ExtendedTimeout=900000,

1267

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

1268

LastCheckerStatus=0)

1269

self.other_client = MockClient(

1270

"barbar",

1271

KeyID=("0558568eedd67d622f5c83b35a115f79"

1272

"6ab612cff5ad227247e46c2b020f441c"),

1273

Secret=b"secretbar",

1274

Host="192.0.2.3",

1275

Enabled=dbus.Boolean(True),

1276

Timeout=300000,

1277

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

1278

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

1279

Interval=120000,

1280

Fingerprint=("3E393AEAEFB84C7E89E2"

1281

"F547B3A107558FCA3A27"),

1282

CheckerRunning=dbus.Boolean(True),

1283

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

1284

ApprovalPending=dbus.Boolean(False),

1285

ApprovedByDefault=dbus.Boolean(False),

1286

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

1287

ApprovalDelay=30000,

1288

ApprovalDuration=93785000,

1289

Checker=":",

1290

ExtendedTimeout=900000,

1291

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

1292

LastCheckerStatus=-2)

1293

self.clients = collections.OrderedDict(

1294

[

1295

("/clients/foo", self.client.attributes),

1296

("/clients/barbar", self.other_client.attributes),

1297

])

1298

self.one_client = {"/clients/foo": self.client.attributes}

1299

1300

@property

1301

def bus(self):

1302

class Bus(object):

1303

@staticmethod

1304

def get_object(client_bus_name, path):

1305

self.assertEqual(client_bus_name, dbus_busname)

1306

return {

1307

# Note: "self" here is the TestCmd instance, not

1308

# the Bus instance, since this is a static method!

1309

"/clients/foo": self.client,

1310

"/clients/barbar": self.other_client,

1311

}[path]

1312

return Bus()

1313

1314

1315

class TestIsEnabledCmd(TestCmd):

1316

def test_is_enabled(self):

1317

self.assertTrue(all(IsEnabledCmd().is_enabled(client,

1318

properties)

1319

for client, properties

1320

in self.clients.items()))

1321

1322

def test_is_enabled_run_exits_successfully(self):

1323

with self.assertRaises(SystemExit) as e:

1324

IsEnabledCmd().run(self.one_client)

1325

if e.exception.code is not None:

1326

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

1327

else:

1328

self.assertIsNone(e.exception.code)

1329

1330

def test_is_enabled_run_exits_with_failure(self):

1331

self.client.attributes["Enabled"] = dbus.Boolean(False)

1332

with self.assertRaises(SystemExit) as e:

1333

IsEnabledCmd().run(self.one_client)

1334

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

1335

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

1336

else:

1337

self.assertIsNotNone(e.exception.code)

1338

1339

1340

class TestApproveCmd(TestCmd):

1341

def test_approve(self):

1342

ApproveCmd().run(self.clients, self.bus)

1343

for clientpath in self.clients:

1344

client = self.bus.get_object(dbus_busname, clientpath)

1345

self.assertIn(("Approve", (True, client_dbus_interface)),

1346

client.calls)

1347

1348

1349

class TestDenyCmd(TestCmd):

1350

def test_deny(self):

1351

DenyCmd().run(self.clients, self.bus)

1352

for clientpath in self.clients:

1353

client = self.bus.get_object(dbus_busname, clientpath)

1354

self.assertIn(("Approve", (False, client_dbus_interface)),

1355

client.calls)

1356

1357

1358

class TestRemoveCmd(TestCmd):

1359

def test_remove(self):

1360

class MockMandos(object):

1361

def __init__(self):

1362

self.calls = []

1363

def RemoveClient(self, dbus_path):

1364

self.calls.append(("RemoveClient", (dbus_path,)))

1365

mandos = MockMandos()

1366

super(TestRemoveCmd, self).setUp()

1367

RemoveCmd().run(self.clients, self.bus, mandos)

1368

self.assertEqual(len(mandos.calls), 2)

1369

for clientpath in self.clients:

1370

self.assertIn(("RemoveClient", (clientpath,)),

1371

mandos.calls)

1372

1373

1374

class TestDumpJSONCmd(TestCmd):

1375

def setUp(self):

1376

self.expected_json = {

1377

"foo": {

1378

"Name": "foo",

1379

"KeyID": ("92ed150794387c03ce684574b1139a65"

1380

"94a34f895daaaf09fd8ea90a27cddb12"),

1381

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

1382

"Enabled": True,

1383

"Timeout": 300000,

1384

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

1385

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

1386

"Interval": 120000,

1387

"Fingerprint": ("778827225BA7DE539C5A"

1388

"7CFA59CFF7CDBD9A5920"),

1389

"CheckerRunning": False,

1390

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

1391

"ApprovalPending": False,

1392

"ApprovedByDefault": True,

1393

"LastApprovalRequest": "",

1394

"ApprovalDelay": 0,

1395

"ApprovalDuration": 1000,

1396

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

1397

"ExtendedTimeout": 900000,

1398

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

1399

"LastCheckerStatus": 0,

1400

},

1401

"barbar": {

1402

"Name": "barbar",

1403

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1404

"6ab612cff5ad227247e46c2b020f441c"),

1405

"Host": "192.0.2.3",

1406

"Enabled": True,

1407

"Timeout": 300000,

1408

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

1409

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

1410

"Interval": 120000,

1411

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1412

"F547B3A107558FCA3A27"),

1413

"CheckerRunning": True,

1414

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

1415

"ApprovalPending": False,

1416

"ApprovedByDefault": False,

1417

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

1418

"ApprovalDelay": 30000,

1419

"ApprovalDuration": 93785000,

1420

"Checker": ":",

1421

"ExtendedTimeout": 900000,

1422

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

1423

"LastCheckerStatus": -2,

1424

},

1425

}

1426

return super(TestDumpJSONCmd, self).setUp()

1427

1428

def test_normal(self):

1429

output = DumpJSONCmd().output(self.clients.values())

1430

json_data = json.loads(output)

1431

self.assertDictEqual(json_data, self.expected_json)

1432

1433

def test_one_client(self):

1434

output = DumpJSONCmd().output(self.one_client.values())

1435

json_data = json.loads(output)

1436

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

1437

self.assertDictEqual(json_data, expected_json)

1438

1439

1440

class TestPrintTableCmd(TestCmd):

1441

def test_normal(self):

1442

output = PrintTableCmd().output(self.clients.values())

1443

expected_output = "\n".join((

1444

"Name Enabled Timeout Last Successful Check",

1445

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

1446

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

1447

))

1448

self.assertEqual(output, expected_output)

1449

1450

def test_verbose(self):

1451

output = PrintTableCmd(verbose=True).output(

1452

self.clients.values())

1453

columns = (

1454

(

1455

"Name ",

1456

"foo ",

1457

"barbar ",

1458

),(

1459

"Enabled ",

1460

"Yes ",

1461

"Yes ",

1462

),(

1463

"Timeout ",

1464

"00:05:00 ",

1465

"00:05:00 ",

1466

),(

1467

"Last Successful Check ",

1468

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

1469

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

1470

),(

1471

"Created ",

1472

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

1473

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

1474

),(

1475

"Interval ",

1476

"00:02:00 ",

1477

"00:02:00 ",

1478

),(

1479

"Host ",

1480

"foo.example.org ",

1481

"192.0.2.3 ",

1482

),(

1483

("Key ID "

1484

" "),

1485

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1486

"ea90a27cddb12 "),

1487

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1488

"46c2b020f441c "),

1489

),(

1490

"Fingerprint ",

1491

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1492

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1493

),(

1494

"Check Is Running ",

1495

"No ",

1496

"Yes ",

1497

),(

1498

"Last Enabled ",

1499

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

1500

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

1501

),(

1502

"Approval Is Pending ",

1503

"No ",

1504

"No ",

1505

),(

1506

"Approved By Default ",

1507

"Yes ",

1508

"No ",

1509

),(

1510

"Last Approval Request ",

1511

" ",

1512

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

1513

),(

1514

"Approval Delay ",

1515

"00:00:00 ",

1516

"00:00:30 ",

1517

),(

1518

"Approval Duration ",

1519

"00:00:01 ",

1520

"1T02:03:05 ",

1521

),(

1522

"Checker ",

1523

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

1524

": ",

1525

),(

1526

"Extended Timeout ",

1527

"00:15:00 ",

1528

"00:15:00 ",

1529

),(

1530

"Expires ",

1531

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

1532

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

1533

),(

1534

"Last Checker Status",

1535

"0 ",

1536

"-2 ",

1537

)

1538

)

1539

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

1540

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

1541

for rows in columns)

1542

for line in range(num_lines))

1543

self.assertEqual(output, expected_output)

1544

1545

def test_one_client(self):

1546

output = PrintTableCmd().output(self.one_client.values())

1547

expected_output = "\n".join((

1548

"Name Enabled Timeout Last Successful Check",

1549

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

1550

))

1551

self.assertEqual(output, expected_output)

1552

1553

1554

class TestPropertyCmd(TestCmd):

1555

"""Abstract class for tests of PropertyCmd classes"""

1556

def runTest(self):

1557

if not hasattr(self, "command"):

1558

return

1559

values_to_get = getattr(self, "values_to_get",

1560

self.values_to_set)

1561

for value_to_set, value_to_get in zip(self.values_to_set,

1562

values_to_get):

1563

for clientpath in self.clients:

1564

client = self.bus.get_object(dbus_busname, clientpath)

1565

old_value = client.attributes[self.propname]

1566

self.assertNotIsInstance(old_value, self.Unique)

1567

client.attributes[self.propname] = self.Unique()

1568

self.run_command(value_to_set, self.clients)

1569

for clientpath in self.clients:

1570

client = self.bus.get_object(dbus_busname, clientpath)

1571

value = client.attributes[self.propname]

1572

self.assertNotIsInstance(value, self.Unique)

1573

self.assertEqual(value, value_to_get)

1574

1575

class Unique(object):

1576

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

1577

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

1578

1579

def run_command(self, value, clients):

1580

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

1581

1582

1583

class TestEnableCmd(TestPropertyCmd):

1584

command = EnableCmd

1585

propname = "Enabled"

1586

values_to_set = [dbus.Boolean(True)]

1587

1588

1589

class TestDisableCmd(TestPropertyCmd):

1590

command = DisableCmd

1591

propname = "Enabled"

1592

values_to_set = [dbus.Boolean(False)]

1593

1594

1595

class TestBumpTimeoutCmd(TestPropertyCmd):

1596

command = BumpTimeoutCmd

1597

propname = "LastCheckedOK"

1598

values_to_set = [""]

1599

1600

1601

class TestStartCheckerCmd(TestPropertyCmd):

1602

command = StartCheckerCmd

1603

propname = "CheckerRunning"

1604

values_to_set = [dbus.Boolean(True)]

1605

1606

1607

class TestStopCheckerCmd(TestPropertyCmd):

1608

command = StopCheckerCmd

1609

propname = "CheckerRunning"

1610

values_to_set = [dbus.Boolean(False)]

1611

1612

1613

class TestApproveByDefaultCmd(TestPropertyCmd):

1614

command = ApproveByDefaultCmd

1615

propname = "ApprovedByDefault"

1616

values_to_set = [dbus.Boolean(True)]

1617

1618

1619

class TestDenyByDefaultCmd(TestPropertyCmd):

1620

command = DenyByDefaultCmd

1621

propname = "ApprovedByDefault"

1622

values_to_set = [dbus.Boolean(False)]

1623

1624

1625

class TestPropertyValueCmd(TestPropertyCmd):

1626

"""Abstract class for tests of PropertyValueCmd classes"""

1627

1628

def runTest(self):

1629

if type(self) is TestPropertyValueCmd:

1630

return

1631

return super(TestPropertyValueCmd, self).runTest()

1632

1633

def run_command(self, value, clients):

1634

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

1635

1636

1637

class TestSetCheckerCmd(TestPropertyValueCmd):

1638

command = SetCheckerCmd

1639

propname = "Checker"

1640

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

1641

1642

1643

class TestSetHostCmd(TestPropertyValueCmd):

1644

command = SetHostCmd

1645

propname = "Host"

1646

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

1647

1648

1649

class TestSetSecretCmd(TestPropertyValueCmd):

1650

command = SetSecretCmd

1651

propname = "Secret"

1652

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

1653

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

1654

values_to_get = [b"", b"secret\0xyzzy\nbar"]

1655

1656

1657

class TestSetTimeoutCmd(TestPropertyValueCmd):

1658

command = SetTimeoutCmd

1659

propname = "Timeout"

1660

values_to_set = [datetime.timedelta(),

1661

datetime.timedelta(minutes=5),

1662

datetime.timedelta(seconds=1),

1663

datetime.timedelta(weeks=1),

1664

datetime.timedelta(weeks=52)]

1665

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1666

1667

1668

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1669

command = SetExtendedTimeoutCmd

1670

propname = "ExtendedTimeout"

1671

values_to_set = [datetime.timedelta(),

1672

datetime.timedelta(minutes=5),

1673

datetime.timedelta(seconds=1),

1674

datetime.timedelta(weeks=1),

1675

datetime.timedelta(weeks=52)]

1676

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1677

1678

1679

class TestSetIntervalCmd(TestPropertyValueCmd):

1680

command = SetIntervalCmd

1681

propname = "Interval"

1682

values_to_set = [datetime.timedelta(),

1683

datetime.timedelta(minutes=5),

1684

datetime.timedelta(seconds=1),

1685

datetime.timedelta(weeks=1),

1686

datetime.timedelta(weeks=52)]

1687

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1688

1689

1690

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1691

command = SetApprovalDelayCmd

1692

propname = "ApprovalDelay"

1693

values_to_set = [datetime.timedelta(),

1694

datetime.timedelta(minutes=5),

1695

datetime.timedelta(seconds=1),

1696

datetime.timedelta(weeks=1),

1697

datetime.timedelta(weeks=52)]

1698

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1699

1700

1701

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1702

command = SetApprovalDurationCmd

1703

propname = "ApprovalDuration"

1704

values_to_set = [datetime.timedelta(),

1705

datetime.timedelta(minutes=5),

1706

datetime.timedelta(seconds=1),

1707

datetime.timedelta(weeks=1),

1708

datetime.timedelta(weeks=52)]

1709

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1710

1711

1712

1713

def should_only_run_tests():

1714

parser = argparse.ArgumentParser(add_help=False)

1715

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

1716

args, unknown_args = parser.parse_known_args()

1717

run_tests = args.check

1718

if run_tests:

1719

# Remove --check argument from sys.argv

1720

sys.argv[1:] = unknown_args

1721

return run_tests

1722

1723

# Add all tests from doctest strings

1724

def load_tests(loader, tests, none):

1725

import doctest

1726

tests.addTests(doctest.DocTestSuite())

1727

return tests

1728

1729

if __name__ == "__main__":

1730

try:

1731

if should_only_run_tests():

1732

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

1733

unittest.main()

1734

else:

1735

main()

1736

finally:

1737

logging.shutdown()