/mandos/trunk : revision 1068

1

#!/usr/bin/python

2

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

3

4

from __future__ import division

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

5

35

import sys

6

import dbus

7

from optparse import OptionParser

36

import argparse

8

37

import locale

9

38

import datetime

10

39

import re

11

40

import os

12

13

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

14

15

tablewords = {

16

'Name': u'Name',

17

'Enabled': u'Enabled',

18

'Timeout': u'Timeout',

19

'LastCheckedOK': u'Last Successful Check',

20

'Created': u'Created',

21

'Interval': u'Interval',

22

'Host': u'Host',

23

'Fingerprint': u'Fingerprint',

24

'CheckerRunning': u'Check Is Running',

25

'LastEnabled': u'Last Enabled',

26

'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

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

try:

106

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

107

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

108

with SilenceLogger("dbus.proxies"):

109

all_clients = {path: ifs_and_props[client_dbus_interface]

110

for path, ifs_and_props in

111

mandos_serv_object_manager

112

.GetManagedObjects().items()

113

if client_dbus_interface in ifs_and_props}

114

except dbus.exceptions.DBusException as e:

115

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

116

"\n%s", e)

117

sys.exit(1)

118

119

# Compile dict of (clients: properties) to process

120

clients = {}

121

122

if not clientnames:

123

clients = all_clients

124

else:

125

for name in clientnames:

126

for objpath, properties in all_clients.items():

127

if properties["Name"] == name:

128

clients[objpath] = properties

129

break

130

else:

131

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

132

sys.exit(1)

133

134

# Run all commands on clients

135

commands = commands_from_options(options)

136

for command in commands:

137

command.run(clients, bus, mandos_serv)

138

139

140

def add_command_line_options(parser):

141

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

142

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

143

help="show version number and exit")

144

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

145

help="Select all clients")

146

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

147

help="Print all fields")

148

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

149

help="Dump client data in JSON format")

150

enable_disable = parser.add_mutually_exclusive_group()

151

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

152

help="Enable client")

153

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

154

action="store_true",

155

help="disable client")

156

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

157

help="Bump timeout for client")

158

start_stop_checker = parser.add_mutually_exclusive_group()

159

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

160

action="store_true",

161

help="Start checker for client")

162

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

163

action="store_true",

164

help="Stop checker for client")

165

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

166

help="Check if client is enabled")

167

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

168

help="Remove client")

169

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

170

help="Set checker command for client")

171

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

172

help="Set timeout for client")

173

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

174

help="Set extended timeout for client")

175

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

176

help="Set checker interval for client")

177

approve_deny_default = parser.add_mutually_exclusive_group()

178

approve_deny_default.add_argument(

179

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

180

default=None, dest="approved_by_default",

181

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

182

approve_deny_default.add_argument(

183

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

184

dest="approved_by_default",

185

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

186

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

187

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

188

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

189

help="Set duration of one client approval")

190

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

191

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

192

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

193

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

194

approve_deny = parser.add_mutually_exclusive_group()

195

approve_deny.add_argument(

196

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

197

help="Approve any current client request")

198

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

199

help="Deny any current client request")

200

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

201

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

202

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

203

help="Run self-test")

204

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

205

206

207

def string_to_delta(interval):

208

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

209

210

try:

211

return rfc3339_duration_to_delta(interval)

212

except ValueError as e:

213

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

214

' '.join(e.args))

215

return parse_pre_1_6_1_interval(interval)

216

217

218

def rfc3339_duration_to_delta(duration):

219

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

220

221

>>> rfc3339_duration_to_delta("P7D")

222

datetime.timedelta(7)

223

>>> rfc3339_duration_to_delta("PT60S")

224

datetime.timedelta(0, 60)

225

>>> rfc3339_duration_to_delta("PT60M")

226

datetime.timedelta(0, 3600)

227

>>> rfc3339_duration_to_delta("P60M")

228

datetime.timedelta(1680)

229

>>> rfc3339_duration_to_delta("PT24H")

230

datetime.timedelta(1)

231

>>> rfc3339_duration_to_delta("P1W")

232

datetime.timedelta(7)

233

>>> rfc3339_duration_to_delta("PT5M30S")

234

datetime.timedelta(0, 330)

235

>>> rfc3339_duration_to_delta("P1DT3M20S")

236

datetime.timedelta(1, 200)

237

>>> # Can not be empty:

238

>>> rfc3339_duration_to_delta("")

239

Traceback (most recent call last):

240

...

241

ValueError: Invalid RFC 3339 duration: u''

242

>>> # Must start with "P":

243

>>> rfc3339_duration_to_delta("1D")

244

Traceback (most recent call last):

245

...

246

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

247

>>> # Must use correct order

248

>>> rfc3339_duration_to_delta("PT1S2M")

249

Traceback (most recent call last):

250

...

251

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

252

>>> # Time needs time marker

253

>>> rfc3339_duration_to_delta("P1H2S")

254

Traceback (most recent call last):

255

...

256

ValueError: Invalid RFC 3339 duration: u'P1H2S'

257

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

258

>>> rfc3339_duration_to_delta("P1D2W")

259

Traceback (most recent call last):

260

...

261

ValueError: Invalid RFC 3339 duration: u'P1D2W'

262

>>> rfc3339_duration_to_delta("P2W2H")

263

Traceback (most recent call last):

264

...

265

ValueError: Invalid RFC 3339 duration: u'P2W2H'

266

"""

267

268

# Parsing an RFC 3339 duration with regular expressions is not

269

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

270

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

271

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

272

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

273

# avoid excessive use of external libraries.

274

275

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

276

Token = collections.namedtuple("Token", (

277

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

278

# a "group" containing digits

279

"value", # datetime.timedelta or None

280

"followers")) # Tokens valid after this token

281

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

282

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

283

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

284

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

285

datetime.timedelta(seconds=1),

286

frozenset((token_end, )))

287

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

288

datetime.timedelta(minutes=1),

289

frozenset((token_second, token_end)))

290

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

291

datetime.timedelta(hours=1),

292

frozenset((token_minute, token_end)))

293

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

294

None,

295

frozenset((token_hour, token_minute,

296

token_second)))

297

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

298

datetime.timedelta(days=1),

299

frozenset((token_time, token_end)))

300

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

301

datetime.timedelta(weeks=4),

302

frozenset((token_day, token_end)))

303

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

304

datetime.timedelta(weeks=52),

305

frozenset((token_month, token_end)))

306

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

307

datetime.timedelta(weeks=1),

308

frozenset((token_end, )))

309

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

310

frozenset((token_year, token_month,

311

token_day, token_time,

312

token_week)))

313

# Define starting values:

314

# Value so far

315

value = datetime.timedelta()

316

found_token = None

317

# Following valid tokens

318

followers = frozenset((token_duration, ))

319

# String left to parse

320

s = duration

321

# Loop until end token is found

322

while found_token is not token_end:

323

# Search for any currently valid tokens

324

for token in followers:

325

match = token.regexp.match(s)

326

if match is not None:

327

# Token found

328

if token.value is not None:

329

# Value found, parse digits

330

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

331

# Add to value so far

332

value += factor * token.value

333

# Strip token from string

334

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

335

# Go to found token

336

found_token = token

337

# Set valid next tokens

338

followers = found_token.followers

339

break

340

else:

341

# No currently valid tokens were found

342

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

343

.format(duration))

344

# End token found

345

return value

346

347

348

def parse_pre_1_6_1_interval(interval):

349

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

350

and return a datetime.timedelta

351

352

>>> parse_pre_1_6_1_interval('7d')

353

datetime.timedelta(7)

354

>>> parse_pre_1_6_1_interval('60s')

355

datetime.timedelta(0, 60)

356

>>> parse_pre_1_6_1_interval('60m')

357

datetime.timedelta(0, 3600)

358

>>> parse_pre_1_6_1_interval('24h')

359

datetime.timedelta(1)

360

>>> parse_pre_1_6_1_interval('1w')

361

datetime.timedelta(7)

362

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

363

datetime.timedelta(0, 330)

364

>>> parse_pre_1_6_1_interval('')

365

datetime.timedelta(0)

366

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

367

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

368

datetime.timedelta(2, 480, 18000)

369

370

"""

371

372

value = datetime.timedelta(0)

373

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

374

375

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

376

if suffix == "d":

377

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

378

elif suffix == "s":

379

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

380

elif suffix == "m":

381

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

382

elif suffix == "h":

383

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

384

elif suffix == "w":

385

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

386

elif suffix == "":

387

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

388

return value

389

390

391

def check_option_syntax(parser, options):

392

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

393

argparse"""

394

395

def has_actions(options):

396

return any((options.enable,

397

options.disable,

398

options.bump_timeout,

399

options.start_checker,

400

options.stop_checker,

401

options.is_enabled,

402

options.remove,

403

options.checker is not None,

404

options.timeout is not None,

405

options.extended_timeout is not None,

406

options.interval is not None,

407

options.approved_by_default is not None,

408

options.approval_delay is not None,

409

options.approval_duration is not None,

410

options.host is not None,

411

options.secret is not None,

412

options.approve,

413

options.deny))

414

415

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

416

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

417

if options.verbose and has_actions(options):

418

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

419

if options.dump_json and (options.verbose

420

or has_actions(options)):

421

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

422

if options.all and not has_actions(options):

423

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

424

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

425

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

426

if options.remove:

427

options.remove = False

428

if has_actions(options) and not options.deny:

429

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

430

options.remove = True

431

432

433

def get_mandos_dbus_object(bus):

434

try:

435

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

436

dbus_busname, server_dbus_path)

437

mandos_dbus_object = bus.get_object(dbus_busname,

438

server_dbus_path)

439

except dbus.exceptions.DBusException:

440

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

441

sys.exit(1)

442

443

return mandos_dbus_object

444

445

446

class SilenceLogger(object):

447

"Simple context manager to silence a particular logger"

448

def __init__(self, loggername):

449

self.logger = logging.getLogger(loggername)

450

451

def __enter__(self):

452

self.logger.addFilter(self.nullfilter)

453

return self

454

455

class NullFilter(logging.Filter):

456

def filter(self, record):

457

return False

458

459

nullfilter = NullFilter()

460

461

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

462

self.logger.removeFilter(self.nullfilter)

463

464

465

def commands_from_options(options):

466

467

commands = []

468

469

if options.is_enabled:

470

commands.append(IsEnabledCmd())

471

472

if options.approve:

473

commands.append(ApproveCmd())

474

475

if options.deny:

476

commands.append(DenyCmd())

477

478

if options.remove:

479

commands.append(RemoveCmd())

480

481

if options.dump_json:

482

commands.append(DumpJSONCmd())

483

484

if options.enable:

485

commands.append(EnableCmd())

486

487

if options.disable:

488

commands.append(DisableCmd())

489

490

if options.bump_timeout:

491

commands.append(BumpTimeoutCmd())

492

493

if options.start_checker:

494

commands.append(StartCheckerCmd())

495

496

if options.stop_checker:

497

commands.append(StopCheckerCmd())

498

499

if options.approved_by_default is not None:

500

if options.approved_by_default:

501

commands.append(ApproveByDefaultCmd())

502

else:

503

commands.append(DenyByDefaultCmd())

504

505

if options.checker is not None:

506

commands.append(SetCheckerCmd(options.checker))

507

508

if options.host is not None:

509

commands.append(SetHostCmd(options.host))

510

511

if options.secret is not None:

512

commands.append(SetSecretCmd(options.secret))

513

514

if options.timeout is not None:

515

commands.append(SetTimeoutCmd(options.timeout))

516

517

if options.extended_timeout:

518

commands.append(

519

SetExtendedTimeoutCmd(options.extended_timeout))

520

521

if options.interval is not None:

522

commands.append(SetIntervalCmd(options.interval))

523

524

if options.approval_delay is not None:

525

commands.append(SetApprovalDelayCmd(options.approval_delay))

526

527

if options.approval_duration is not None:

528

commands.append(

529

SetApprovalDurationCmd(options.approval_duration))

530

531

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

532

# optionally verbosely

533

if not commands:

534

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

535

536

return commands

537

538

539

class Command(object):

540

"""Abstract class for commands"""

541

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

542

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

543

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

544

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

545

self.mandos = mandos

546

for clientpath, properties in clients.items():

547

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

548

dbus_busname, str(clientpath))

549

client = bus.get_object(dbus_busname, clientpath)

550

self.run_on_one_client(client, properties)

551

552

553

class IsEnabledCmd(Command):

554

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

555

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

556

if self.is_enabled(client, properties):

557

sys.exit(0)

558

sys.exit(1)

559

def is_enabled(self, client, properties):

560

return properties["Enabled"]

561

562

563

class ApproveCmd(Command):

564

def run_on_one_client(self, client, properties):

565

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

566

client.__dbus_object_path__, client_dbus_interface)

567

client.Approve(dbus.Boolean(True),

568

dbus_interface=client_dbus_interface)

569

570

571

class DenyCmd(Command):

572

def run_on_one_client(self, client, properties):

573

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

574

client.__dbus_object_path__, client_dbus_interface)

575

client.Approve(dbus.Boolean(False),

576

dbus_interface=client_dbus_interface)

577

578

579

class RemoveCmd(Command):

580

def run_on_one_client(self, client, properties):

581

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

582

server_dbus_path, server_dbus_interface,

583

str(client.__dbus_object_path__))

584

self.mandos.RemoveClient(client.__dbus_object_path__)

585

586

587

class OutputCmd(Command):

588

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

589

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

590

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

591

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

592

"ApprovalPending", "ApprovedByDefault",

593

"LastApprovalRequest", "ApprovalDelay",

594

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

595

"Expires", "LastCheckerStatus")

596

597

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

598

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

599

600

def output(self, clients):

601

raise NotImplementedError()

602

603

604

class DumpJSONCmd(OutputCmd):

605

def output(self, clients):

606

data = {client["Name"]:

607

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

608

for key in self.all_keywords}

609

for client in clients}

610

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

611

612

@staticmethod

613

def dbus_boolean_to_bool(value):

614

if isinstance(value, dbus.Boolean):

615

value = bool(value)

616

return value

617

618

619

class PrintTableCmd(OutputCmd):

620

def __init__(self, verbose=False):

621

self.verbose = verbose

622

623

def output(self, clients):

624

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

625

"LastCheckedOK")

626

keywords = default_keywords

627

if self.verbose:

628

keywords = self.all_keywords

629

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

630

631

class TableOfClients(object):

632

tableheaders = {

633

"Name": "Name",

634

"Enabled": "Enabled",

635

"Timeout": "Timeout",

636

"LastCheckedOK": "Last Successful Check",

637

"LastApprovalRequest": "Last Approval Request",

638

"Created": "Created",

639

"Interval": "Interval",

640

"Host": "Host",

641

"Fingerprint": "Fingerprint",

642

"KeyID": "Key ID",

643

"CheckerRunning": "Check Is Running",

644

"LastEnabled": "Last Enabled",

645

"ApprovalPending": "Approval Is Pending",

646

"ApprovedByDefault": "Approved By Default",

647

"ApprovalDelay": "Approval Delay",

648

"ApprovalDuration": "Approval Duration",

649

"Checker": "Checker",

650

"ExtendedTimeout": "Extended Timeout",

651

"Expires": "Expires",

652

"LastCheckerStatus": "Last Checker Status",

653

}

654

655

def __init__(self, clients, keywords):

656

self.clients = clients

657

self.keywords = keywords

658

659

def __str__(self):

660

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

661

662

if sys.version_info.major == 2:

663

__unicode__ = __str__

664

def __str__(self):

665

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

666

667

def rows(self):

668

format_string = self.row_formatting_string()

669

rows = [self.header_line(format_string)]

670

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

671

for client in self.clients)

672

return rows

673

674

def row_formatting_string(self):

675

"Format string used to format table rows"

676

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

677

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

678

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

679

for client in self.clients)),

680

key=key)

681

for key in self.keywords)

682

683

def string_from_client(self, client, key):

684

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

685

686

@classmethod

687

def valuetostring(cls, value, keyword):

688

if isinstance(value, dbus.Boolean):

689

return "Yes" if value else "No"

690

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

691

"ApprovalDuration", "ExtendedTimeout"):

692

return cls.milliseconds_to_string(value)

693

return str(value)

694

695

def header_line(self, format_string):

696

return format_string.format(**self.tableheaders)

697

698

def client_line(self, client, format_string):

699

return format_string.format(

700

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

701

for key in self.keywords})

702

703

@staticmethod

704

def milliseconds_to_string(ms):

705

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

706

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

707

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

708

if td.days else "",

709

hours=td.seconds // 3600,

710

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

711

seconds=td.seconds % 60))

712

713

714

class PropertyCmd(Command):

715

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

716

717

def run_on_one_client(self, client, properties):

718

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

719

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

720

client.__dbus_object_path__,

721

dbus.PROPERTIES_IFACE, client_dbus_interface,

722

self.propname, self.value_to_set

723

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

724

else bool(self.value_to_set))

725

client.Set(client_dbus_interface, self.propname,

726

self.value_to_set,

727

dbus_interface=dbus.PROPERTIES_IFACE)

728

729

@property

730

def propname(self):

731

raise NotImplementedError()

732

733

734

class EnableCmd(PropertyCmd):

735

propname = "Enabled"

736

value_to_set = dbus.Boolean(True)

737

738

739

class DisableCmd(PropertyCmd):

740

propname = "Enabled"

741

value_to_set = dbus.Boolean(False)

742

743

744

class BumpTimeoutCmd(PropertyCmd):

745

propname = "LastCheckedOK"

746

value_to_set = ""

747

748

749

class StartCheckerCmd(PropertyCmd):

750

propname = "CheckerRunning"

751

value_to_set = dbus.Boolean(True)

752

753

754

class StopCheckerCmd(PropertyCmd):

755

propname = "CheckerRunning"

756

value_to_set = dbus.Boolean(False)

757

758

759

class ApproveByDefaultCmd(PropertyCmd):

760

propname = "ApprovedByDefault"

761

value_to_set = dbus.Boolean(True)

762

763

764

class DenyByDefaultCmd(PropertyCmd):

765

propname = "ApprovedByDefault"

766

value_to_set = dbus.Boolean(False)

767

768

769

class PropertyValueCmd(PropertyCmd):

770

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

771

def __init__(self, value):

772

self.value_to_set = value

773

774

775

class SetCheckerCmd(PropertyValueCmd):

776

propname = "Checker"

777

778

779

class SetHostCmd(PropertyValueCmd):

780

propname = "Host"

781

782

783

class SetSecretCmd(PropertyValueCmd):

784

propname = "Secret"

785

786

@property

787

def value_to_set(self):

788

return self._vts

789

790

@value_to_set.setter

791

def value_to_set(self, value):

792

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

793

self._vts = value.read()

794

value.close()

795

796

797

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

798

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

799

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

800

801

@property

802

def value_to_set(self):

803

return self._vts

804

805

@value_to_set.setter

806

def value_to_set(self, value):

807

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

808

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

809

810

811

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

812

propname = "Timeout"

813

814

815

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

816

propname = "ExtendedTimeout"

817

818

819

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

820

propname = "Interval"

821

822

823

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

824

propname = "ApprovalDelay"

825

826

827

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

828

propname = "ApprovalDuration"

829

830

831

832

class Test_string_to_delta(unittest.TestCase):

833

def test_handles_basic_rfc3339(self):

834

self.assertEqual(string_to_delta("PT0S"),

835

datetime.timedelta())

836

self.assertEqual(string_to_delta("P0D"),

837

datetime.timedelta())

838

self.assertEqual(string_to_delta("PT1S"),

839

datetime.timedelta(0, 1))

840

self.assertEqual(string_to_delta("PT2H"),

841

datetime.timedelta(0, 7200))

842

843

def test_falls_back_to_pre_1_6_1_with_warning(self):

844

# assertLogs only exists in Python 3.4

845

if hasattr(self, "assertLogs"):

846

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

847

value = string_to_delta("2h")

848

else:

849

class WarningFilter(logging.Filter):

850

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

851

def filter(self, record):

852

is_warning = record.levelno >= logging.WARNING

853

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

854

False)

855

return not is_warning

856

warning_filter = WarningFilter()

857

log.addFilter(warning_filter)

858

try:

859

value = string_to_delta("2h")

860

finally:

861

log.removeFilter(warning_filter)

862

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

863

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

864

865

866

class Test_check_option_syntax(unittest.TestCase):

867

def setUp(self):

868

self.parser = argparse.ArgumentParser()

869

add_command_line_options(self.parser)

870

871

def test_actions_requires_client_or_all(self):

872

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

873

options = self.parser.parse_args()

874

setattr(options, action, value)

875

with self.assertParseError():

876

self.check_option_syntax(options)

877

878

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

879

# check_option_syntax()

880

actions = {

881

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

882

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

883

# never used

884

"enable": True,

885

"disable": True,

886

"bump_timeout": True,

887

"start_checker": True,

888

"stop_checker": True,

889

"is_enabled": True,

890

"remove": True,

891

"checker": "x",

892

"timeout": datetime.timedelta(),

893

"extended_timeout": datetime.timedelta(),

894

"interval": datetime.timedelta(),

895

"approved_by_default": True,

896

"approval_delay": datetime.timedelta(),

897

"approval_duration": datetime.timedelta(),

898

"host": "x",

899

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

900

"approve": True,

901

"deny": True,

27

902

}

28

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

29

domain = 'se.bsnet.fukt'

30

busname = domain + '.Mandos'

31

server_path = '/'

32

server_interface = domain + '.Mandos'

33

client_interface = domain + '.Mandos.Client'

34

version = "1.0.14"

35

36

def timedelta_to_milliseconds(td):

37

"Convert a datetime.timedelta object to milliseconds"

38

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

39

+ (td.seconds * 1000)

40

+ (td.microseconds // 1000))

41

42

def milliseconds_to_string(ms):

43

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

44

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

45

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

46

"hours": td.seconds // 3600,

47

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

48

"seconds": td.seconds % 60,

49

})

50

51

52

def string_to_delta(interval):

53

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

54

55

>>> string_to_delta('7d')

56

datetime.timedelta(7)

57

>>> string_to_delta('60s')

58

datetime.timedelta(0, 60)

59

>>> string_to_delta('60m')

60

datetime.timedelta(0, 3600)

61

>>> string_to_delta('24h')

62

datetime.timedelta(1)

63

>>> string_to_delta(u'1w')

64

datetime.timedelta(7)

65

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

66

datetime.timedelta(0, 330)

67

"""

68

timevalue = datetime.timedelta(0)

69

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

70

71

for s in regexp.findall(interval):

72

try:

73

suffix = unicode(s[-1])

74

value = int(s[:-1])

75

if suffix == u"d":

76

delta = datetime.timedelta(value)

77

elif suffix == u"s":

78

delta = datetime.timedelta(0, value)

79

elif suffix == u"m":

80

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

81

elif suffix == u"h":

82

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

83

elif suffix == u"w":

84

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

85

else:

86

raise ValueError

87

except (ValueError, IndexError):

88

raise ValueError

89

timevalue += delta

90

return timevalue

91

92

def print_clients(clients, keywords):

93

def valuetostring(value, keyword):

94

if type(value) is dbus.Boolean:

95

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

96

if keyword in (u"Timeout", u"Interval"):

97

return milliseconds_to_string(value)

98

return unicode(value)

99

100

# Create format string to print table rows

101

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

102

max(len(tablewords[key]),

103

max(len(valuetostring(client[key],

104

key))

105

for client in

106

clients))

107

for key in keywords)

108

# Print header line

109

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

110

for client in clients:

111

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

112

for key in keywords)

113

114

def has_actions(options):

115

return any((options.enable,

116

options.disable,

117

options.bump_timeout,

118

options.start_checker,

119

options.stop_checker,

120

options.is_enabled,

121

options.remove,

122

options.checker is not None,

123

options.timeout is not None,

124

options.interval is not None,

125

options.host is not None,

126

options.secret is not None,

127

options.approve,

128

options.deny))

129

130

def main():

131

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

132

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

133

help="Select all clients")

134

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

135

help="Print all fields")

136

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

137

help="Enable client")

138

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

139

help="disable client")

140

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

141

help="Bump timeout for client")

142

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

143

help="Start checker for client")

144

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

145

help="Stop checker for client")

146

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

147

help="Check if client is enabled")

148

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

149

help="Remove client")

150

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

151

help="Set checker command for client")

152

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

153

help="Set timeout for client")

154

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

155

help="Set checker interval for client")

156

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

157

help="Set host for client")

158

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

159

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

160

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

161

help="Approve any current client request")

162

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

163

help="Deny any current client request")

164

options, client_names = parser.parse_args()

165

166

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

167

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

168

if options.verbose and has_actions(options):

169

parser.error('Verbose option can only be used alone or with --all.')

170

if options.all and not has_actions(options):

171

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

172

173

try:

174

bus = dbus.SystemBus()

175

mandos_dbus_objc = bus.get_object(busname, server_path)

176

except dbus.exceptions.DBusException:

177

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

178

sys.exit(1)

179

180

mandos_serv = dbus.Interface(mandos_dbus_objc,

181

dbus_interface = server_interface)

182

183

#block stderr since dbus library prints to stderr

903

904

@contextlib.contextmanager

905

def assertParseError(self):

906

with self.assertRaises(SystemExit) as e:

907

with self.temporarily_suppress_stderr():

908

yield

909

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

910

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

911

# /argparse.html#exiting-methods

912

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

913

914

@staticmethod

915

@contextlib.contextmanager

916

def temporarily_suppress_stderr():

184

917

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

185

918

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

186

919

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

187

920

os.close(null)

188

921

try:

922

yield

923

finally:

924

# restore stderr

925

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

926

os.close(stderrcopy)

927

928

def check_option_syntax(self, options):

929

check_option_syntax(self.parser, options)

930

931

def test_actions_conflicts_with_verbose(self):

932

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

933

options = self.parser.parse_args()

934

setattr(options, action, value)

935

options.verbose = True

936

with self.assertParseError():

937

self.check_option_syntax(options)

938

939

def test_dump_json_conflicts_with_verbose(self):

940

options = self.parser.parse_args()

941

options.dump_json = True

942

options.verbose = True

943

with self.assertParseError():

944

self.check_option_syntax(options)

945

946

def test_dump_json_conflicts_with_action(self):

947

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

948

options = self.parser.parse_args()

949

setattr(options, action, value)

950

options.dump_json = True

951

with self.assertParseError():

952

self.check_option_syntax(options)

953

954

def test_all_can_not_be_alone(self):

955

options = self.parser.parse_args()

956

options.all = True

957

with self.assertParseError():

958

self.check_option_syntax(options)

959

960

def test_all_is_ok_with_any_action(self):

961

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

962

options = self.parser.parse_args()

963

setattr(options, action, value)

964

options.all = True

965

self.check_option_syntax(options)

966

967

def test_is_enabled_fails_without_client(self):

968

options = self.parser.parse_args()

969

options.is_enabled = True

970

with self.assertParseError():

971

self.check_option_syntax(options)

972

973

def test_is_enabled_works_with_one_client(self):

974

options = self.parser.parse_args()

975

options.is_enabled = True

976

options.client = ["foo"]

977

self.check_option_syntax(options)

978

979

def test_is_enabled_fails_with_two_clients(self):

980

options = self.parser.parse_args()

981

options.is_enabled = True

982

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

983

with self.assertParseError():

984

self.check_option_syntax(options)

985

986

def test_remove_can_only_be_combined_with_action_deny(self):

987

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

988

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

989

continue

990

options = self.parser.parse_args()

991

setattr(options, action, value)

992

options.all = True

993

options.remove = True

994

with self.assertParseError():

995

self.check_option_syntax(options)

996

997

998

class Test_get_mandos_dbus_object(unittest.TestCase):

999

def test_calls_and_returns_get_object_on_bus(self):

1000

class MockBus(object):

1001

called = False

1002

def get_object(mockbus_self, busname, dbus_path):

1003

# Note that "self" is still the testcase instance,

1004

# this MockBus instance is in "mockbus_self".

1005

self.assertEqual(busname, dbus_busname)

1006

self.assertEqual(dbus_path, server_dbus_path)

1007

mockbus_self.called = True

1008

return mockbus_self

1009

1010

mockbus = get_mandos_dbus_object(bus=MockBus())

1011

self.assertIsInstance(mockbus, MockBus)

1012

self.assertTrue(mockbus.called)

1013

1014

def test_logs_and_exits_on_dbus_error(self):

1015

class MockBusFailing(object):

1016

def get_object(self, busname, dbus_path):

1017

raise dbus.exceptions.DBusException("Test")

1018

1019

# assertLogs only exists in Python 3.4

1020

if hasattr(self, "assertLogs"):

1021

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

1022

with self.assertRaises(SystemExit) as e:

1023

bus = get_mandos_dbus_object(bus=MockBus())

1024

else:

1025

critical_filter = self.CriticalFilter()

1026

log.addFilter(critical_filter)

189

1027

try:

190

mandos_clients = mandos_serv.GetAllClientsWithProperties()

1028

with self.assertRaises(SystemExit) as e:

1029

get_mandos_dbus_object(bus=MockBusFailing())

191

1030

finally:

192

#restore stderr

193

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

194

os.close(stderrcopy)

195

except dbus.exceptions.DBusException, e:

196

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

197

sys.exit(1)

198

199

# Compile dict of (clients: properties) to process

200

clients={}

201

202

if options.all or not client_names:

203

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

204

for path, properties in

205

mandos_clients.iteritems())

206

else:

207

for name in client_names:

208

for path, client in mandos_clients.iteritems():

209

if client['Name'] == name:

210

client_objc = bus.get_object(busname, path)

211

clients[client_objc] = client

212

break

213

else:

214

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

215

sys.exit(1)

216

217

if not has_actions(options) and clients:

218

if options.verbose:

219

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

220

'Created', 'Interval', 'Host', 'Fingerprint',

221

'CheckerRunning', 'LastEnabled', 'Checker')

222

else:

223

keywords = defaultkeywords

224

225

print_clients(clients.values(), keywords)

226

else:

227

# Process each client in the list by all selected options

228

for client in clients:

229

if options.remove:

230

mandos_serv.RemoveClient(client.__dbus_object_path__)

231

if options.enable:

232

client.Enable(dbus_interface=client_interface)

233

if options.disable:

234

client.Disable(dbus_interface=client_interface)

235

if options.bump_timeout:

236

client.CheckedOK(dbus_interface=client_interface)

237

if options.start_checker:

238

client.StartChecker(dbus_interface=client_interface)

239

if options.stop_checker:

240

client.StopChecker(dbus_interface=client_interface)

241

if options.is_enabled:

242

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

243

u"Enabled",

244

dbus_interface=dbus.PROPERTIES_IFACE)

245

else 1)

246

if options.checker:

247

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

248

dbus_interface=dbus.PROPERTIES_IFACE)

249

if options.host:

250

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

251

dbus_interface=dbus.PROPERTIES_IFACE)

252

if options.interval:

253

client.Set(client_interface, u"Interval",

254

timedelta_to_milliseconds

255

(string_to_delta(options.interval)),

256

dbus_interface=dbus.PROPERTIES_IFACE)

257

if options.timeout:

258

client.Set(client_interface, u"Timeout",

259

timedelta_to_milliseconds(string_to_delta

260

(options.timeout)),

261

dbus_interface=dbus.PROPERTIES_IFACE)

262

if options.secret:

263

client.Set(client_interface, u"Secret",

264

dbus.ByteArray(open(options.secret, u'rb').read()),

265

dbus_interface=dbus.PROPERTIES_IFACE)

266

if options.approve:

267

client.Approve(dbus.Boolean(True), dbus_interface=client_interface)

268

if options.deny:

269

client.Approve(dbus.Boolean(False), dbus_interface=client_interface)

270

271

if __name__ == '__main__':

272

main()

1031

log.removeFilter(critical_filter)

1032

self.assertTrue(critical_filter.found)

1033

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

1034

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

1035

else:

1036

self.assertIsNotNone(e.exception.code)

1037

1038

class CriticalFilter(logging.Filter):

1039

"""Don't show, but register, critical messages"""

1040

found = False

1041

def filter(self, record):

1042

is_critical = record.levelno >= logging.CRITICAL

1043

self.found = is_critical or self.found

1044

return not is_critical

1045

1046

1047

class Test_SilenceLogger(unittest.TestCase):

1048

loggername = "mandos-ctl.Test_SilenceLogger"

1049

log = logging.getLogger(loggername)

1050

log.propagate = False

1051

log.addHandler(logging.NullHandler())

1052

1053

def setUp(self):

1054

self.counting_filter = self.CountingFilter()

1055

1056

class CountingFilter(logging.Filter):

1057

"Count number of records"

1058

count = 0

1059

def filter(self, record):

1060

self.count += 1

1061

return True

1062

1063

def test_should_filter_records_only_when_active(self):

1064

try:

1065

with SilenceLogger(self.loggername):

1066

self.log.addFilter(self.counting_filter)

1067

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

1068

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

1069

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

1070

finally:

1071

self.log.removeFilter(self.counting_filter)

1072

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

1073

1074

1075

class Test_commands_from_options(unittest.TestCase):

1076

def setUp(self):

1077

self.parser = argparse.ArgumentParser()

1078

add_command_line_options(self.parser)

1079

1080

def test_is_enabled(self):

1081

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

1082

IsEnabledCmd)

1083

1084

def assert_command_from_args(self, args, command_cls,

1085

**cmd_attrs):

1086

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

1087

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

1088

options = self.parser.parse_args(args)

1089

check_option_syntax(self.parser, options)

1090

commands = commands_from_options(options)

1091

self.assertEqual(len(commands), 1)

1092

command = commands[0]

1093

self.assertIsInstance(command, command_cls)

1094

for key, value in cmd_attrs.items():

1095

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

1096

1097

def test_is_enabled_short(self):

1098

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

1099

1100

def test_approve(self):

1101

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

1102

ApproveCmd)

1103

1104

def test_approve_short(self):

1105

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

1106

1107

def test_deny(self):

1108

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

1109

1110

def test_deny_short(self):

1111

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

1112

1113

def test_remove(self):

1114

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

1115

RemoveCmd)

1116

1117

def test_deny_before_remove(self):

1118

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

1119

"foo"])

1120

check_option_syntax(self.parser, options)

1121

commands = commands_from_options(options)

1122

self.assertEqual(len(commands), 2)

1123

self.assertIsInstance(commands[0], DenyCmd)

1124

self.assertIsInstance(commands[1], RemoveCmd)

1125

1126

def test_deny_before_remove_reversed(self):

1127

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

1128

"--all"])

1129

check_option_syntax(self.parser, options)

1130

commands = commands_from_options(options)

1131

self.assertEqual(len(commands), 2)

1132

self.assertIsInstance(commands[0], DenyCmd)

1133

self.assertIsInstance(commands[1], RemoveCmd)

1134

1135

def test_remove_short(self):

1136

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

1137

1138

def test_dump_json(self):

1139

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

1140

1141

def test_enable(self):

1142

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

1143

1144

def test_enable_short(self):

1145

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

1146

1147

def test_disable(self):

1148

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

1149

DisableCmd)

1150

1151

def test_disable_short(self):

1152

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

1153

1154

def test_bump_timeout(self):

1155

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

1156

BumpTimeoutCmd)

1157

1158

def test_bump_timeout_short(self):

1159

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

1160

1161

def test_start_checker(self):

1162

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

1163

StartCheckerCmd)

1164

1165

def test_stop_checker(self):

1166

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

1167

StopCheckerCmd)

1168

1169

def test_approve_by_default(self):

1170

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

1171

ApproveByDefaultCmd)

1172

1173

def test_deny_by_default(self):

1174

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

1175

DenyByDefaultCmd)

1176

1177

def test_checker(self):

1178

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

1179

SetCheckerCmd, value_to_set=":")

1180

1181

def test_checker_empty(self):

1182

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

1183

SetCheckerCmd, value_to_set="")

1184

1185

def test_checker_short(self):

1186

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

1187

SetCheckerCmd, value_to_set=":")

1188

1189

def test_host(self):

1190

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

1191

"foo"], SetHostCmd,

1192

value_to_set="foo.example.org")

1193

1194

def test_host_short(self):

1195

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

1196

"foo"], SetHostCmd,

1197

value_to_set="foo.example.org")

1198

1199

def test_secret_devnull(self):

1200

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

1201

"foo"], SetSecretCmd,

1202

value_to_set=b"")

1203

1204

def test_secret_tempfile(self):

1205

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

1206

value = b"secret\0xyzzy\nbar"

1207

f.write(value)

1208

f.seek(0)

1209

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

1210

"foo"], SetSecretCmd,

1211

value_to_set=value)

1212

1213

def test_secret_devnull_short(self):

1214

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

1215

SetSecretCmd, value_to_set=b"")

1216

1217

def test_secret_tempfile_short(self):

1218

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

1219

value = b"secret\0xyzzy\nbar"

1220

f.write(value)

1221

f.seek(0)

1222

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

1223

SetSecretCmd,

1224

value_to_set=value)

1225

1226

def test_timeout(self):

1227

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

1228

SetTimeoutCmd,

1229

value_to_set=300000)

1230

1231

def test_timeout_short(self):

1232

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

1233

SetTimeoutCmd,

1234

value_to_set=300000)

1235

1236

def test_extended_timeout(self):

1237

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

1238

"foo"],

1239

SetExtendedTimeoutCmd,

1240

value_to_set=900000)

1241

1242

def test_interval(self):

1243

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

1244

SetIntervalCmd,

1245

value_to_set=120000)

1246

1247

def test_interval_short(self):

1248

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

1249

SetIntervalCmd,

1250

value_to_set=120000)

1251

1252

def test_approval_delay(self):

1253

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

1254

"foo"], SetApprovalDelayCmd,

1255

value_to_set=30000)

1256

1257

def test_approval_duration(self):

1258

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

1259

"foo"], SetApprovalDurationCmd,

1260

value_to_set=1000)

1261

1262

def test_print_table(self):

1263

self.assert_command_from_args([], PrintTableCmd,

1264

verbose=False)

1265

1266

def test_print_table_verbose(self):

1267

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

1268

verbose=True)

1269

1270

def test_print_table_verbose_short(self):

1271

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

1272

verbose=True)

1273

1274

1275

class TestCmd(unittest.TestCase):

1276

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

1277

1278

def setUp(self):

1279

testcase = self

1280

class MockClient(object):

1281

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

1282

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

1283

self.attributes = attributes

1284

self.attributes["Name"] = name

1285

self.calls = []

1286

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

1287

testcase.assertEqual(interface, client_dbus_interface)

1288

testcase.assertEqual(dbus_interface,

1289

dbus.PROPERTIES_IFACE)

1290

self.attributes[propname] = value

1291

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

1292

testcase.assertEqual(interface, client_dbus_interface)

1293

testcase.assertEqual(dbus_interface,

1294

dbus.PROPERTIES_IFACE)

1295

return self.attributes[propname]

1296

def Approve(self, approve, dbus_interface):

1297

testcase.assertEqual(dbus_interface,

1298

client_dbus_interface)

1299

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

1300

dbus_interface)))

1301

self.client = MockClient(

1302

"foo",

1303

KeyID=("92ed150794387c03ce684574b1139a65"

1304

"94a34f895daaaf09fd8ea90a27cddb12"),

1305

Secret=b"secret",

1306

Host="foo.example.org",

1307

Enabled=dbus.Boolean(True),

1308

Timeout=300000,

1309

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

1310

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

1311

Interval=120000,

1312

Fingerprint=("778827225BA7DE539C5A"

1313

"7CFA59CFF7CDBD9A5920"),

1314

CheckerRunning=dbus.Boolean(False),

1315

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

1316

ApprovalPending=dbus.Boolean(False),

1317

ApprovedByDefault=dbus.Boolean(True),

1318

LastApprovalRequest="",

1319

ApprovalDelay=0,

1320

ApprovalDuration=1000,

1321

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

1322

ExtendedTimeout=900000,

1323

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

1324

LastCheckerStatus=0)

1325

self.other_client = MockClient(

1326

"barbar",

1327

KeyID=("0558568eedd67d622f5c83b35a115f79"

1328

"6ab612cff5ad227247e46c2b020f441c"),

1329

Secret=b"secretbar",

1330

Host="192.0.2.3",

1331

Enabled=dbus.Boolean(True),

1332

Timeout=300000,

1333

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

1334

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

1335

Interval=120000,

1336

Fingerprint=("3E393AEAEFB84C7E89E2"

1337

"F547B3A107558FCA3A27"),

1338

CheckerRunning=dbus.Boolean(True),

1339

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

1340

ApprovalPending=dbus.Boolean(False),

1341

ApprovedByDefault=dbus.Boolean(False),

1342

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

1343

ApprovalDelay=30000,

1344

ApprovalDuration=93785000,

1345

Checker=":",

1346

ExtendedTimeout=900000,

1347

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

1348

LastCheckerStatus=-2)

1349

self.clients = collections.OrderedDict(

1350

[

1351

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

1352

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

1353

])

1354

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

1355

1356

@property

1357

def bus(self):

1358

class Bus(object):

1359

@staticmethod

1360

def get_object(client_bus_name, path):

1361

self.assertEqual(client_bus_name, dbus_busname)

1362

return {

1363

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

1364

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

1365

"/clients/foo": self.client,

1366

"/clients/barbar": self.other_client,

1367

}[path]

1368

return Bus()

1369

1370

1371

class TestIsEnabledCmd(TestCmd):

1372

def test_is_enabled(self):

1373

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

1374

properties)

1375

for client, properties

1376

in self.clients.items()))

1377

1378

def test_is_enabled_run_exits_successfully(self):

1379

with self.assertRaises(SystemExit) as e:

1380

IsEnabledCmd().run(self.one_client)

1381

if e.exception.code is not None:

1382

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

1383

else:

1384

self.assertIsNone(e.exception.code)

1385

1386

def test_is_enabled_run_exits_with_failure(self):

1387

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

1388

with self.assertRaises(SystemExit) as e:

1389

IsEnabledCmd().run(self.one_client)

1390

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

1391

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

1392

else:

1393

self.assertIsNotNone(e.exception.code)

1394

1395

1396

class TestApproveCmd(TestCmd):

1397

def test_approve(self):

1398

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

1399

for clientpath in self.clients:

1400

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

1401

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

1402

client.calls)

1403

1404

1405

class TestDenyCmd(TestCmd):

1406

def test_deny(self):

1407

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

1408

for clientpath in self.clients:

1409

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

1410

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

1411

client.calls)

1412

1413

1414

class TestRemoveCmd(TestCmd):

1415

def test_remove(self):

1416

class MockMandos(object):

1417

def __init__(self):

1418

self.calls = []

1419

def RemoveClient(self, dbus_path):

1420

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

1421

mandos = MockMandos()

1422

super(TestRemoveCmd, self).setUp()

1423

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

1424

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

1425

for clientpath in self.clients:

1426

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

1427

mandos.calls)

1428

1429

1430

class TestDumpJSONCmd(TestCmd):

1431

def setUp(self):

1432

self.expected_json = {

1433

"foo": {

1434

"Name": "foo",

1435

"KeyID": ("92ed150794387c03ce684574b1139a65"

1436

"94a34f895daaaf09fd8ea90a27cddb12"),

1437

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

1438

"Enabled": True,

1439

"Timeout": 300000,

1440

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

1441

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

1442

"Interval": 120000,

1443

"Fingerprint": ("778827225BA7DE539C5A"

1444

"7CFA59CFF7CDBD9A5920"),

1445

"CheckerRunning": False,

1446

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

1447

"ApprovalPending": False,

1448

"ApprovedByDefault": True,

1449

"LastApprovalRequest": "",

1450

"ApprovalDelay": 0,

1451

"ApprovalDuration": 1000,

1452

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

1453

"ExtendedTimeout": 900000,

1454

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

1455

"LastCheckerStatus": 0,

1456

},

1457

"barbar": {

1458

"Name": "barbar",

1459

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1460

"6ab612cff5ad227247e46c2b020f441c"),

1461

"Host": "192.0.2.3",

1462

"Enabled": True,

1463

"Timeout": 300000,

1464

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

1465

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

1466

"Interval": 120000,

1467

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1468

"F547B3A107558FCA3A27"),

1469

"CheckerRunning": True,

1470

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

1471

"ApprovalPending": False,

1472

"ApprovedByDefault": False,

1473

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

1474

"ApprovalDelay": 30000,

1475

"ApprovalDuration": 93785000,

1476

"Checker": ":",

1477

"ExtendedTimeout": 900000,

1478

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

1479

"LastCheckerStatus": -2,

1480

},

1481

}

1482

return super(TestDumpJSONCmd, self).setUp()

1483

1484

def test_normal(self):

1485

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

1486

json_data = json.loads(output)

1487

self.assertDictEqual(json_data, self.expected_json)

1488

1489

def test_one_client(self):

1490

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

1491

json_data = json.loads(output)

1492

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

1493

self.assertDictEqual(json_data, expected_json)

1494

1495

1496

class TestPrintTableCmd(TestCmd):

1497

def test_normal(self):

1498

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

1499

expected_output = "\n".join((

1500

"Name Enabled Timeout Last Successful Check",

1501

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

1502

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

1503

))

1504

self.assertEqual(output, expected_output)

1505

1506

def test_verbose(self):

1507

output = PrintTableCmd(verbose=True).output(

1508

self.clients.values())

1509

columns = (

1510

(

1511

"Name ",

1512

"foo ",

1513

"barbar ",

1514

),(

1515

"Enabled ",

1516

"Yes ",

1517

"Yes ",

1518

),(

1519

"Timeout ",

1520

"00:05:00 ",

1521

"00:05:00 ",

1522

),(

1523

"Last Successful Check ",

1524

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

1525

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

1526

),(

1527

"Created ",

1528

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

1529

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

1530

),(

1531

"Interval ",

1532

"00:02:00 ",

1533

"00:02:00 ",

1534

),(

1535

"Host ",

1536

"foo.example.org ",

1537

"192.0.2.3 ",

1538

),(

1539

("Key ID "

1540

" "),

1541

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1542

"ea90a27cddb12 "),

1543

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1544

"46c2b020f441c "),

1545

),(

1546

"Fingerprint ",

1547

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1548

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1549

),(

1550

"Check Is Running ",

1551

"No ",

1552

"Yes ",

1553

),(

1554

"Last Enabled ",

1555

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

1556

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

1557

),(

1558

"Approval Is Pending ",

1559

"No ",

1560

"No ",

1561

),(

1562

"Approved By Default ",

1563

"Yes ",

1564

"No ",

1565

),(

1566

"Last Approval Request ",

1567

" ",

1568

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

1569

),(

1570

"Approval Delay ",

1571

"00:00:00 ",

1572

"00:00:30 ",

1573

),(

1574

"Approval Duration ",

1575

"00:00:01 ",

1576

"1T02:03:05 ",

1577

),(

1578

"Checker ",

1579

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

1580

": ",

1581

),(

1582

"Extended Timeout ",

1583

"00:15:00 ",

1584

"00:15:00 ",

1585

),(

1586

"Expires ",

1587

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

1588

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

1589

),(

1590

"Last Checker Status",

1591

"0 ",

1592

"-2 ",

1593

)

1594

)

1595

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

1596

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

1597

for rows in columns)

1598

for line in range(num_lines))

1599

self.assertEqual(output, expected_output)

1600

1601

def test_one_client(self):

1602

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

1603

expected_output = "\n".join((

1604

"Name Enabled Timeout Last Successful Check",

1605

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

1606

))

1607

self.assertEqual(output, expected_output)

1608

1609

1610

class TestPropertyCmd(TestCmd):

1611

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

1612

def runTest(self):

1613

if not hasattr(self, "command"):

1614

return

1615

values_to_get = getattr(self, "values_to_get",

1616

self.values_to_set)

1617

for value_to_set, value_to_get in zip(self.values_to_set,

1618

values_to_get):

1619

for clientpath in self.clients:

1620

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

1621

old_value = client.attributes[self.propname]

1622

self.assertNotIsInstance(old_value, self.Unique)

1623

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

1624

self.run_command(value_to_set, self.clients)

1625

for clientpath in self.clients:

1626

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

1627

value = client.attributes[self.propname]

1628

self.assertNotIsInstance(value, self.Unique)

1629

self.assertEqual(value, value_to_get)

1630

1631

class Unique(object):

1632

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

1633

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

1634

1635

def run_command(self, value, clients):

1636

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

1637

1638

1639

class TestEnableCmd(TestPropertyCmd):

1640

command = EnableCmd

1641

propname = "Enabled"

1642

values_to_set = [dbus.Boolean(True)]

1643

1644

1645

class TestDisableCmd(TestPropertyCmd):

1646

command = DisableCmd

1647

propname = "Enabled"

1648

values_to_set = [dbus.Boolean(False)]

1649

1650

1651

class TestBumpTimeoutCmd(TestPropertyCmd):

1652

command = BumpTimeoutCmd

1653

propname = "LastCheckedOK"

1654

values_to_set = [""]

1655

1656

1657

class TestStartCheckerCmd(TestPropertyCmd):

1658

command = StartCheckerCmd

1659

propname = "CheckerRunning"

1660

values_to_set = [dbus.Boolean(True)]

1661

1662

1663

class TestStopCheckerCmd(TestPropertyCmd):

1664

command = StopCheckerCmd

1665

propname = "CheckerRunning"

1666

values_to_set = [dbus.Boolean(False)]

1667

1668

1669

class TestApproveByDefaultCmd(TestPropertyCmd):

1670

command = ApproveByDefaultCmd

1671

propname = "ApprovedByDefault"

1672

values_to_set = [dbus.Boolean(True)]

1673

1674

1675

class TestDenyByDefaultCmd(TestPropertyCmd):

1676

command = DenyByDefaultCmd

1677

propname = "ApprovedByDefault"

1678

values_to_set = [dbus.Boolean(False)]

1679

1680

1681

class TestPropertyValueCmd(TestPropertyCmd):

1682

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

1683

1684

def runTest(self):

1685

if type(self) is TestPropertyValueCmd:

1686

return

1687

return super(TestPropertyValueCmd, self).runTest()

1688

1689

def run_command(self, value, clients):

1690

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

1691

1692

1693

class TestSetCheckerCmd(TestPropertyValueCmd):

1694

command = SetCheckerCmd

1695

propname = "Checker"

1696

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

1697

1698

1699

class TestSetHostCmd(TestPropertyValueCmd):

1700

command = SetHostCmd

1701

propname = "Host"

1702

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

1703

1704

1705

class TestSetSecretCmd(TestPropertyValueCmd):

1706

command = SetSecretCmd

1707

propname = "Secret"

1708

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

1709

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

1710

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

1711

1712

1713

class TestSetTimeoutCmd(TestPropertyValueCmd):

1714

command = SetTimeoutCmd

1715

propname = "Timeout"

1716

values_to_set = [datetime.timedelta(),

1717

datetime.timedelta(minutes=5),

1718

datetime.timedelta(seconds=1),

1719

datetime.timedelta(weeks=1),

1720

datetime.timedelta(weeks=52)]

1721

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

1722

1723

1724

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1725

command = SetExtendedTimeoutCmd

1726

propname = "ExtendedTimeout"

1727

values_to_set = [datetime.timedelta(),

1728

datetime.timedelta(minutes=5),

1729

datetime.timedelta(seconds=1),

1730

datetime.timedelta(weeks=1),

1731

datetime.timedelta(weeks=52)]

1732

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

1733

1734

1735

class TestSetIntervalCmd(TestPropertyValueCmd):

1736

command = SetIntervalCmd

1737

propname = "Interval"

1738

values_to_set = [datetime.timedelta(),

1739

datetime.timedelta(minutes=5),

1740

datetime.timedelta(seconds=1),

1741

datetime.timedelta(weeks=1),

1742

datetime.timedelta(weeks=52)]

1743

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

1744

1745

1746

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1747

command = SetApprovalDelayCmd

1748

propname = "ApprovalDelay"

1749

values_to_set = [datetime.timedelta(),

1750

datetime.timedelta(minutes=5),

1751

datetime.timedelta(seconds=1),

1752

datetime.timedelta(weeks=1),

1753

datetime.timedelta(weeks=52)]

1754

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

1755

1756

1757

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1758

command = SetApprovalDurationCmd

1759

propname = "ApprovalDuration"

1760

values_to_set = [datetime.timedelta(),

1761

datetime.timedelta(minutes=5),

1762

datetime.timedelta(seconds=1),

1763

datetime.timedelta(weeks=1),

1764

datetime.timedelta(weeks=52)]

1765

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

1766

1767

1768

1769

def should_only_run_tests():

1770

parser = argparse.ArgumentParser(add_help=False)

1771

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

1772

args, unknown_args = parser.parse_known_args()

1773

run_tests = args.check

1774

if run_tests:

1775

# Remove --check argument from sys.argv

1776

sys.argv[1:] = unknown_args

1777

return run_tests

1778

1779

# Add all tests from doctest strings

1780

def load_tests(loader, tests, none):

1781

import doctest

1782

tests.addTests(doctest.DocTestSuite())

1783

return tests

1784

1785

if __name__ == "__main__":

1786

try:

1787

if should_only_run_tests():

1788

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

1789

unittest.main()

1790

else:

1791

main()

1792

finally:

1793

logging.shutdown()