/mandos/trunk : revision 24.1.23

1

#!/usr/bin/python

2

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

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

13

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

14

# (at your option) any later version.

15

#

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

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

22

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

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

35

import sys

36

import argparse

37

import locale

38

import datetime

39

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

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

# Filter out log message from dbus module

112

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

113

class NullFilter(logging.Filter):

114

def filter(self, record):

115

return False

116

dbus_filter = NullFilter()

117

try:

118

dbus_logger.addFilter(dbus_filter)

119

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

120

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

121

mandos_clients = {path: ifs_and_props[client_dbus_interface]

122

for path, ifs_and_props in

123

mandos_serv_object_manager

124

.GetManagedObjects().items()

125

if client_dbus_interface in ifs_and_props}

126

except dbus.exceptions.DBusException as e:

127

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

128

"\n%s", e)

129

sys.exit(1)

130

finally:

131

# restore dbus logger

132

dbus_logger.removeFilter(dbus_filter)

133

134

# Compile dict of (clients: properties) to process

135

clients = {}

136

137

if not clientnames:

138

clients = {objpath: properties

139

for objpath, properties in mandos_clients.items()}

140

else:

141

for name in clientnames:

142

for objpath, properties in mandos_clients.items():

143

if properties["Name"] == name:

144

clients[objpath] = properties

145

break

146

else:

147

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

148

sys.exit(1)

149

150

# Run all commands on clients

151

commands = commands_from_options(options)

152

for command in commands:

153

command.run(clients, bus, mandos_serv)

154

155

156

def add_command_line_options(parser):

157

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

158

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

159

help="show version number and exit")

160

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

161

help="Select all clients")

162

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

163

help="Print all fields")

164

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

165

help="Dump client data in JSON format")

166

enable_disable = parser.add_mutually_exclusive_group()

167

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

168

help="Enable client")

169

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

170

action="store_true",

171

help="disable client")

172

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

173

help="Bump timeout for client")

174

start_stop_checker = parser.add_mutually_exclusive_group()

175

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

176

action="store_true",

177

help="Start checker for client")

178

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

179

action="store_true",

180

help="Stop checker for client")

181

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

182

help="Check if client is enabled")

183

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

184

help="Remove client")

185

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

186

help="Set checker command for client")

187

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

188

help="Set timeout for client")

189

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

190

help="Set extended timeout for client")

191

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

192

help="Set checker interval for client")

193

approve_deny_default = parser.add_mutually_exclusive_group()

194

approve_deny_default.add_argument(

195

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

196

default=None, dest="approved_by_default",

197

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

198

approve_deny_default.add_argument(

199

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

200

dest="approved_by_default",

201

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

202

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

203

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

204

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

205

help="Set duration of one client approval")

206

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

207

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

208

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

209

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

210

approve_deny = parser.add_mutually_exclusive_group()

211

approve_deny.add_argument(

212

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

213

help="Approve any current client request")

214

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

215

help="Deny any current client request")

216

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

217

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

218

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

219

help="Run self-test")

220

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

221

222

223

def string_to_delta(interval):

224

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

225

226

try:

227

return rfc3339_duration_to_delta(interval)

228

except ValueError as e:

229

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

230

' '.join(e.args))

231

return parse_pre_1_6_1_interval(interval)

232

233

234

def rfc3339_duration_to_delta(duration):

235

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

236

237

>>> rfc3339_duration_to_delta("P7D")

238

datetime.timedelta(7)

239

>>> rfc3339_duration_to_delta("PT60S")

240

datetime.timedelta(0, 60)

241

>>> rfc3339_duration_to_delta("PT60M")

242

datetime.timedelta(0, 3600)

243

>>> rfc3339_duration_to_delta("P60M")

244

datetime.timedelta(1680)

245

>>> rfc3339_duration_to_delta("PT24H")

246

datetime.timedelta(1)

247

>>> rfc3339_duration_to_delta("P1W")

248

datetime.timedelta(7)

249

>>> rfc3339_duration_to_delta("PT5M30S")

250

datetime.timedelta(0, 330)

251

>>> rfc3339_duration_to_delta("P1DT3M20S")

252

datetime.timedelta(1, 200)

253

>>> # Can not be empty:

254

>>> rfc3339_duration_to_delta("")

255

Traceback (most recent call last):

256

...

257

ValueError: Invalid RFC 3339 duration: u''

258

>>> # Must start with "P":

259

>>> rfc3339_duration_to_delta("1D")

260

Traceback (most recent call last):

261

...

262

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

263

>>> # Must use correct order

264

>>> rfc3339_duration_to_delta("PT1S2M")

265

Traceback (most recent call last):

266

...

267

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

268

>>> # Time needs time marker

269

>>> rfc3339_duration_to_delta("P1H2S")

270

Traceback (most recent call last):

271

...

272

ValueError: Invalid RFC 3339 duration: u'P1H2S'

273

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

274

>>> rfc3339_duration_to_delta("P1D2W")

275

Traceback (most recent call last):

276

...

277

ValueError: Invalid RFC 3339 duration: u'P1D2W'

278

>>> rfc3339_duration_to_delta("P2W2H")

279

Traceback (most recent call last):

280

...

281

ValueError: Invalid RFC 3339 duration: u'P2W2H'

282

"""

283

284

# Parsing an RFC 3339 duration with regular expressions is not

285

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

286

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

287

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

288

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

289

# avoid excessive use of external libraries.

290

291

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

292

Token = collections.namedtuple("Token", (

293

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

294

# a "group" containing digits

295

"value", # datetime.timedelta or None

296

"followers")) # Tokens valid after this token

297

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

298

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

299

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

300

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

301

datetime.timedelta(seconds=1),

302

frozenset((token_end, )))

303

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

304

datetime.timedelta(minutes=1),

305

frozenset((token_second, token_end)))

306

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

307

datetime.timedelta(hours=1),

308

frozenset((token_minute, token_end)))

309

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

310

None,

311

frozenset((token_hour, token_minute,

312

token_second)))

313

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

314

datetime.timedelta(days=1),

315

frozenset((token_time, token_end)))

316

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

317

datetime.timedelta(weeks=4),

318

frozenset((token_day, token_end)))

319

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

320

datetime.timedelta(weeks=52),

321

frozenset((token_month, token_end)))

322

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

323

datetime.timedelta(weeks=1),

324

frozenset((token_end, )))

325

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

326

frozenset((token_year, token_month,

327

token_day, token_time,

328

token_week)))

329

# Define starting values:

330

# Value so far

331

value = datetime.timedelta()

332

found_token = None

333

# Following valid tokens

334

followers = frozenset((token_duration, ))

335

# String left to parse

336

s = duration

337

# Loop until end token is found

338

while found_token is not token_end:

339

# Search for any currently valid tokens

340

for token in followers:

341

match = token.regexp.match(s)

342

if match is not None:

343

# Token found

344

if token.value is not None:

345

# Value found, parse digits

346

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

347

# Add to value so far

348

value += factor * token.value

349

# Strip token from string

350

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

351

# Go to found token

352

found_token = token

353

# Set valid next tokens

354

followers = found_token.followers

355

break

356

else:

357

# No currently valid tokens were found

358

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

359

.format(duration))

360

# End token found

361

return value

362

363

364

def parse_pre_1_6_1_interval(interval):

365

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

366

and return a datetime.timedelta

367

368

>>> parse_pre_1_6_1_interval('7d')

369

datetime.timedelta(7)

370

>>> parse_pre_1_6_1_interval('60s')

371

datetime.timedelta(0, 60)

372

>>> parse_pre_1_6_1_interval('60m')

373

datetime.timedelta(0, 3600)

374

>>> parse_pre_1_6_1_interval('24h')

375

datetime.timedelta(1)

376

>>> parse_pre_1_6_1_interval('1w')

377

datetime.timedelta(7)

378

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

379

datetime.timedelta(0, 330)

380

>>> parse_pre_1_6_1_interval('')

381

datetime.timedelta(0)

382

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

383

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

384

datetime.timedelta(2, 480, 18000)

385

386

"""

387

388

value = datetime.timedelta(0)

389

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

390

391

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

392

if suffix == "d":

393

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

394

elif suffix == "s":

395

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

396

elif suffix == "m":

397

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

398

elif suffix == "h":

399

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

400

elif suffix == "w":

401

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

402

elif suffix == "":

403

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

404

return value

405

406

407

def check_option_syntax(parser, options):

408

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

409

argparse"""

410

411

def has_actions(options):

412

return any((options.enable,

413

options.disable,

414

options.bump_timeout,

415

options.start_checker,

416

options.stop_checker,

417

options.is_enabled,

418

options.remove,

419

options.checker is not None,

420

options.timeout is not None,

421

options.extended_timeout is not None,

422

options.interval is not None,

423

options.approved_by_default is not None,

424

options.approval_delay is not None,

425

options.approval_duration is not None,

426

options.host is not None,

427

options.secret is not None,

428

options.approve,

429

options.deny))

430

431

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

432

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

433

if options.verbose and has_actions(options):

434

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

435

if options.dump_json and (options.verbose

436

or has_actions(options)):

437

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

438

if options.all and not has_actions(options):

439

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

440

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

441

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

442

if options.remove:

443

options.remove = False

444

if has_actions(options) and not options.deny:

445

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

446

options.remove = True

447

448

449

def commands_from_options(options):

450

451

commands = []

452

453

if options.is_enabled:

454

commands.append(IsEnabledCmd())

455

456

if options.approve:

457

commands.append(ApproveCmd())

458

459

if options.deny:

460

commands.append(DenyCmd())

461

462

if options.remove:

463

commands.append(RemoveCmd())

464

465

if options.dump_json:

466

commands.append(DumpJSONCmd())

467

468

if options.enable:

469

commands.append(EnableCmd())

470

471

if options.disable:

472

commands.append(DisableCmd())

473

474

if options.bump_timeout:

475

commands.append(BumpTimeoutCmd())

476

477

if options.start_checker:

478

commands.append(StartCheckerCmd())

479

480

if options.stop_checker:

481

commands.append(StopCheckerCmd())

482

483

if options.approved_by_default is not None:

484

if options.approved_by_default:

485

commands.append(ApproveByDefaultCmd())

486

else:

487

commands.append(DenyByDefaultCmd())

488

489

if options.checker is not None:

490

commands.append(SetCheckerCmd(options.checker))

491

492

if options.host is not None:

493

commands.append(SetHostCmd(options.host))

494

495

if options.secret is not None:

496

commands.append(SetSecretCmd(options.secret))

497

498

if options.timeout is not None:

499

commands.append(SetTimeoutCmd(options.timeout))

500

501

if options.extended_timeout:

502

commands.append(

503

SetExtendedTimeoutCmd(options.extended_timeout))

504

505

if options.interval is not None:

506

commands.append(SetIntervalCmd(options.interval))

507

508

if options.approval_delay is not None:

509

commands.append(SetApprovalDelayCmd(options.approval_delay))

510

511

if options.approval_duration is not None:

512

commands.append(

513

SetApprovalDurationCmd(options.approval_duration))

514

515

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

516

# optionally verbosely

517

if not commands:

518

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

519

520

return commands

521

522

523

class Command(object):

524

"""Abstract class for commands"""

525

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

526

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

527

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

528

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

529

self.mandos = mandos

530

for clientpath, properties in clients.items():

531

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

532

dbus_busname, str(clientpath))

533

client = bus.get_object(dbus_busname, clientpath)

534

self.run_on_one_client(client, properties)

535

536

537

class IsEnabledCmd(Command):

538

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

539

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

540

if self.is_enabled(client, properties):

541

sys.exit(0)

542

sys.exit(1)

543

def is_enabled(self, client, properties):

544

return properties["Enabled"]

545

546

547

class ApproveCmd(Command):

548

def run_on_one_client(self, client, properties):

549

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

550

client.__dbus_object_path__, client_dbus_interface)

551

client.Approve(dbus.Boolean(True),

552

dbus_interface=client_dbus_interface)

553

554

555

class DenyCmd(Command):

556

def run_on_one_client(self, client, properties):

557

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

558

client.__dbus_object_path__, client_dbus_interface)

559

client.Approve(dbus.Boolean(False),

560

dbus_interface=client_dbus_interface)

561

562

563

class RemoveCmd(Command):

564

def run_on_one_client(self, client, properties):

565

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

566

server_dbus_path, server_dbus_interface,

567

str(client.__dbus_object_path__))

568

self.mandos.RemoveClient(client.__dbus_object_path__)

569

570

571

class PrintCmd(Command):

572

"""Abstract class for commands printing client details"""

573

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

574

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

575

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

576

"ApprovalPending", "ApprovedByDefault",

577

"LastApprovalRequest", "ApprovalDelay",

578

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

579

"Expires", "LastCheckerStatus")

580

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

581

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

582

def output(self, clients):

583

raise NotImplementedError()

584

585

586

class DumpJSONCmd(PrintCmd):

587

def output(self, clients):

588

data = {client["Name"]:

589

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

590

for key in self.all_keywords}

591

for client in clients.values()}

592

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

593

@staticmethod

594

def dbus_boolean_to_bool(value):

595

if isinstance(value, dbus.Boolean):

596

value = bool(value)

597

return value

598

599

600

class PrintTableCmd(PrintCmd):

601

def __init__(self, verbose=False):

602

self.verbose = verbose

603

604

def output(self, clients):

605

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

606

"LastCheckedOK")

607

keywords = default_keywords

608

if self.verbose:

609

keywords = self.all_keywords

610

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

611

612

class TableOfClients(object):

613

tableheaders = {

614

"Name": "Name",

615

"Enabled": "Enabled",

616

"Timeout": "Timeout",

617

"LastCheckedOK": "Last Successful Check",

618

"LastApprovalRequest": "Last Approval Request",

619

"Created": "Created",

620

"Interval": "Interval",

621

"Host": "Host",

622

"Fingerprint": "Fingerprint",

623

"KeyID": "Key ID",

624

"CheckerRunning": "Check Is Running",

625

"LastEnabled": "Last Enabled",

626

"ApprovalPending": "Approval Is Pending",

627

"ApprovedByDefault": "Approved By Default",

628

"ApprovalDelay": "Approval Delay",

629

"ApprovalDuration": "Approval Duration",

630

"Checker": "Checker",

631

"ExtendedTimeout": "Extended Timeout",

632

"Expires": "Expires",

633

"LastCheckerStatus": "Last Checker Status",

634

}

635

636

def __init__(self, clients, keywords, tableheaders=None):

637

self.clients = clients

638

self.keywords = keywords

639

if tableheaders is not None:

640

self.tableheaders = tableheaders

641

642

def __str__(self):

643

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

644

645

if sys.version_info.major == 2:

646

__unicode__ = __str__

647

def __str__(self):

648

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

649

650

def rows(self):

651

format_string = self.row_formatting_string()

652

rows = [self.header_line(format_string)]

653

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

654

for client in self.clients)

655

return rows

656

657

def row_formatting_string(self):

658

"Format string used to format table rows"

659

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

660

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

661

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

662

for client in self.clients)),

663

key=key)

664

for key in self.keywords)

665

666

def string_from_client(self, client, key):

667

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

668

669

@staticmethod

670

def valuetostring(value, keyword):

671

if isinstance(value, dbus.Boolean):

672

return "Yes" if value else "No"

673

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

674

"ApprovalDuration", "ExtendedTimeout"):

675

return milliseconds_to_string(value)

676

return str(value)

677

678

def header_line(self, format_string):

679

return format_string.format(**self.tableheaders)

680

681

def client_line(self, client, format_string):

682

return format_string.format(

683

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

684

for key in self.keywords})

685

686

687

def milliseconds_to_string(ms):

688

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

689

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

690

.format(days="{}T".format(td.days) if td.days else "",

691

hours=td.seconds // 3600,

692

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

693

seconds=td.seconds % 60))

694

695

696

class PropertyCmd(Command):

697

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

698

def run_on_one_client(self, client, properties):

699

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

700

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

701

client.__dbus_object_path__,

702

dbus.PROPERTIES_IFACE, client_dbus_interface,

703

self.propname, self.value_to_set

704

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

705

else bool(self.value_to_set))

706

client.Set(client_dbus_interface, self.propname,

707

self.value_to_set,

708

dbus_interface=dbus.PROPERTIES_IFACE)

709

@property

710

def propname(self):

711

raise NotImplementedError()

712

713

714

class EnableCmd(PropertyCmd):

715

propname = "Enabled"

716

value_to_set = dbus.Boolean(True)

717

718

719

class DisableCmd(PropertyCmd):

720

propname = "Enabled"

721

value_to_set = dbus.Boolean(False)

722

723

724

class BumpTimeoutCmd(PropertyCmd):

725

propname = "LastCheckedOK"

726

value_to_set = ""

727

728

729

class StartCheckerCmd(PropertyCmd):

730

propname = "CheckerRunning"

731

value_to_set = dbus.Boolean(True)

732

733

734

class StopCheckerCmd(PropertyCmd):

735

propname = "CheckerRunning"

736

value_to_set = dbus.Boolean(False)

737

738

739

class ApproveByDefaultCmd(PropertyCmd):

740

propname = "ApprovedByDefault"

741

value_to_set = dbus.Boolean(True)

742

743

744

class DenyByDefaultCmd(PropertyCmd):

745

propname = "ApprovedByDefault"

746

value_to_set = dbus.Boolean(False)

747

748

749

class PropertyValueCmd(PropertyCmd):

750

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

751

def __init__(self, value):

752

self.value_to_set = value

753

754

755

class SetCheckerCmd(PropertyValueCmd):

756

propname = "Checker"

757

758

759

class SetHostCmd(PropertyValueCmd):

760

propname = "Host"

761

762

763

class SetSecretCmd(PropertyValueCmd):

764

propname = "Secret"

765

@property

766

def value_to_set(self):

767

return self._vts

768

@value_to_set.setter

769

def value_to_set(self, value):

770

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

771

self._vts = value.read()

772

value.close()

773

774

775

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

776

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

777

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

778

@property

779

def value_to_set(self):

780

return self._vts

781

@value_to_set.setter

782

def value_to_set(self, value):

783

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

784

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

785

786

787

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

788

propname = "Timeout"

789

790

791

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

792

propname = "ExtendedTimeout"

793

794

795

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

796

propname = "Interval"

797

798

799

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

800

propname = "ApprovalDelay"

801

802

803

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

804

propname = "ApprovalDuration"

805

806

807

808

class Test_string_to_delta(unittest.TestCase):

809

def test_handles_basic_rfc3339(self):

810

self.assertEqual(string_to_delta("PT0S"),

811

datetime.timedelta())

812

self.assertEqual(string_to_delta("P0D"),

813

datetime.timedelta())

814

self.assertEqual(string_to_delta("PT1S"),

815

datetime.timedelta(0, 1))

816

self.assertEqual(string_to_delta("PT2H"),

817

datetime.timedelta(0, 7200))

818

def test_falls_back_to_pre_1_6_1_with_warning(self):

819

# assertLogs only exists in Python 3.4

820

if hasattr(self, "assertLogs"):

821

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

822

value = string_to_delta("2h")

823

else:

824

class WarningFilter(logging.Filter):

825

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

826

def filter(self, record):

827

is_warning = record.levelno >= logging.WARNING

828

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

829

False)

830

return not is_warning

831

warning_filter = WarningFilter()

832

log.addFilter(warning_filter)

833

try:

834

value = string_to_delta("2h")

835

finally:

836

log.removeFilter(warning_filter)

837

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

838

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

839

840

841

class Test_check_option_syntax(unittest.TestCase):

842

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

843

# check_option_syntax()

844

actions = {

845

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

846

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

847

# never used

848

"enable": True,

849

"disable": True,

850

"bump_timeout": True,

851

"start_checker": True,

852

"stop_checker": True,

853

"is_enabled": True,

854

"remove": True,

855

"checker": "x",

856

"timeout": datetime.timedelta(),

857

"extended_timeout": datetime.timedelta(),

858

"interval": datetime.timedelta(),

859

"approved_by_default": True,

860

"approval_delay": datetime.timedelta(),

861

"approval_duration": datetime.timedelta(),

862

"host": "x",

863

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

864

"approve": True,

865

"deny": True,

866

}

867

868

def setUp(self):

869

self.parser = argparse.ArgumentParser()

870

add_command_line_options(self.parser)

871

872

@contextlib.contextmanager

873

def assertParseError(self):

874

with self.assertRaises(SystemExit) as e:

875

with self.temporarily_suppress_stderr():

876

yield

877

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

878

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

879

# /argparse.html#exiting-methods

880

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

881

882

@staticmethod

883

@contextlib.contextmanager

884

def temporarily_suppress_stderr():

885

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

886

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

887

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

888

os.close(null)

889

try:

890

yield

891

finally:

892

# restore stderr

893

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

894

os.close(stderrcopy)

895

896

def check_option_syntax(self, options):

897

check_option_syntax(self.parser, options)

898

899

def test_actions_requires_client_or_all(self):

900

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

901

options = self.parser.parse_args()

902

setattr(options, action, value)

903

with self.assertParseError():

904

self.check_option_syntax(options)

905

906

def test_actions_conflicts_with_verbose(self):

907

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

908

options = self.parser.parse_args()

909

setattr(options, action, value)

910

options.verbose = True

911

with self.assertParseError():

912

self.check_option_syntax(options)

913

914

def test_dump_json_conflicts_with_verbose(self):

915

options = self.parser.parse_args()

916

options.dump_json = True

917

options.verbose = True

918

with self.assertParseError():

919

self.check_option_syntax(options)

920

921

def test_dump_json_conflicts_with_action(self):

922

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

923

options = self.parser.parse_args()

924

setattr(options, action, value)

925

options.dump_json = True

926

with self.assertParseError():

927

self.check_option_syntax(options)

928

929

def test_all_can_not_be_alone(self):

930

options = self.parser.parse_args()

931

options.all = True

932

with self.assertParseError():

933

self.check_option_syntax(options)

934

935

def test_all_is_ok_with_any_action(self):

936

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

937

options = self.parser.parse_args()

938

setattr(options, action, value)

939

options.all = True

940

self.check_option_syntax(options)

941

942

def test_is_enabled_fails_without_client(self):

943

options = self.parser.parse_args()

944

options.is_enabled = True

945

with self.assertParseError():

946

self.check_option_syntax(options)

947

948

def test_is_enabled_works_with_one_client(self):

949

options = self.parser.parse_args()

950

options.is_enabled = True

951

options.client = ["foo"]

952

self.check_option_syntax(options)

953

954

def test_is_enabled_fails_with_two_clients(self):

955

options = self.parser.parse_args()

956

options.is_enabled = True

957

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

958

with self.assertParseError():

959

self.check_option_syntax(options)

960

961

def test_remove_can_only_be_combined_with_action_deny(self):

962

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

963

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

964

continue

965

options = self.parser.parse_args()

966

setattr(options, action, value)

967

options.all = True

968

options.remove = True

969

with self.assertParseError():

970

self.check_option_syntax(options)

971

972

973

class Test_command_from_options(unittest.TestCase):

974

def setUp(self):

975

self.parser = argparse.ArgumentParser()

976

add_command_line_options(self.parser)

977

def assert_command_from_args(self, args, command_cls,

978

**cmd_attrs):

979

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

980

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

981

options = self.parser.parse_args(args)

982

check_option_syntax(self.parser, options)

983

commands = commands_from_options(options)

984

self.assertEqual(len(commands), 1)

985

command = commands[0]

986

self.assertIsInstance(command, command_cls)

987

for key, value in cmd_attrs.items():

988

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

989

def test_print_table(self):

990

self.assert_command_from_args([], PrintTableCmd,

991

verbose=False)

992

993

def test_print_table_verbose(self):

994

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

995

verbose=True)

996

997

def test_print_table_verbose_short(self):

998

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

999

verbose=True)

1000

1001

def test_enable(self):

1002

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

1003

1004

def test_enable_short(self):

1005

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

1006

1007

def test_disable(self):

1008

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

1009

DisableCmd)

1010

1011

def test_disable_short(self):

1012

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

1013

1014

def test_bump_timeout(self):

1015

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

1016

BumpTimeoutCmd)

1017

1018

def test_bump_timeout_short(self):

1019

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

1020

1021

def test_start_checker(self):

1022

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

1023

StartCheckerCmd)

1024

1025

def test_stop_checker(self):

1026

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

1027

StopCheckerCmd)

1028

1029

def test_remove(self):

1030

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

1031

RemoveCmd)

1032

1033

def test_remove_short(self):

1034

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

1035

1036

def test_checker(self):

1037

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

1038

SetCheckerCmd, value_to_set=":")

1039

1040

def test_checker_empty(self):

1041

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

1042

SetCheckerCmd, value_to_set="")

1043

1044

def test_checker_short(self):

1045

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

1046

SetCheckerCmd, value_to_set=":")

1047

1048

def test_timeout(self):

1049

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

1050

SetTimeoutCmd,

1051

value_to_set=300000)

1052

1053

def test_timeout_short(self):

1054

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

1055

SetTimeoutCmd,

1056

value_to_set=300000)

1057

1058

def test_extended_timeout(self):

1059

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

1060

"foo"],

1061

SetExtendedTimeoutCmd,

1062

value_to_set=900000)

1063

1064

def test_interval(self):

1065

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

1066

SetIntervalCmd,

1067

value_to_set=120000)

1068

1069

def test_interval_short(self):

1070

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

1071

SetIntervalCmd,

1072

value_to_set=120000)

1073

1074

def test_approve_by_default(self):

1075

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

1076

ApproveByDefaultCmd)

1077

1078

def test_deny_by_default(self):

1079

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

1080

DenyByDefaultCmd)

1081

1082

def test_approval_delay(self):

1083

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

1084

"foo"], SetApprovalDelayCmd,

1085

value_to_set=30000)

1086

1087

def test_approval_duration(self):

1088

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

1089

"foo"], SetApprovalDurationCmd,

1090

value_to_set=1000)

1091

1092

def test_host(self):

1093

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

1094

"foo"], SetHostCmd,

1095

value_to_set="foo.example.org")

1096

1097

def test_host_short(self):

1098

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

1099

"foo"], SetHostCmd,

1100

value_to_set="foo.example.org")

1101

1102

def test_secret_devnull(self):

1103

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

1104

"foo"], SetSecretCmd,

1105

value_to_set=b"")

1106

1107

def test_secret_tempfile(self):

1108

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

1109

value = b"secret\0xyzzy\nbar"

1110

f.write(value)

1111

f.seek(0)

1112

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

1113

"foo"], SetSecretCmd,

1114

value_to_set=value)

1115

1116

def test_secret_devnull_short(self):

1117

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

1118

SetSecretCmd, value_to_set=b"")

1119

1120

def test_secret_tempfile_short(self):

1121

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

1122

value = b"secret\0xyzzy\nbar"

1123

f.write(value)

1124

f.seek(0)

1125

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

1126

SetSecretCmd,

1127

value_to_set=value)

1128

1129

def test_approve(self):

1130

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

1131

ApproveCmd)

1132

1133

def test_approve_short(self):

1134

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

1135

1136

def test_deny(self):

1137

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

1138

1139

def test_deny_short(self):

1140

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

1141

1142

def test_dump_json(self):

1143

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

1144

1145

def test_is_enabled(self):

1146

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

1147

IsEnabledCmd)

1148

1149

def test_is_enabled_short(self):

1150

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

1151

1152

def test_deny_before_remove(self):

1153

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

1154

"foo"])

1155

check_option_syntax(self.parser, options)

1156

commands = commands_from_options(options)

1157

self.assertEqual(len(commands), 2)

1158

self.assertIsInstance(commands[0], DenyCmd)

1159

self.assertIsInstance(commands[1], RemoveCmd)

1160

1161

def test_deny_before_remove_reversed(self):

1162

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

1163

"--all"])

1164

check_option_syntax(self.parser, options)

1165

commands = commands_from_options(options)

1166

self.assertEqual(len(commands), 2)

1167

self.assertIsInstance(commands[0], DenyCmd)

1168

self.assertIsInstance(commands[1], RemoveCmd)

1169

1170

1171

class TestCmd(unittest.TestCase):

1172

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

1173

def setUp(self):

1174

testcase = self

1175

class MockClient(object):

1176

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

1177

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

1178

self.attributes = attributes

1179

self.attributes["Name"] = name

1180

self.calls = []

1181

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

1182

testcase.assertEqual(interface, client_dbus_interface)

1183

testcase.assertEqual(dbus_interface,

1184

dbus.PROPERTIES_IFACE)

1185

self.attributes[propname] = value

1186

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

1187

testcase.assertEqual(interface, client_dbus_interface)

1188

testcase.assertEqual(dbus_interface,

1189

dbus.PROPERTIES_IFACE)

1190

return self.attributes[propname]

1191

def Approve(self, approve, dbus_interface):

1192

testcase.assertEqual(dbus_interface,

1193

client_dbus_interface)

1194

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

1195

dbus_interface)))

1196

self.client = MockClient(

1197

"foo",

1198

KeyID=("92ed150794387c03ce684574b1139a65"

1199

"94a34f895daaaf09fd8ea90a27cddb12"),

1200

Secret=b"secret",

1201

Host="foo.example.org",

1202

Enabled=dbus.Boolean(True),

1203

Timeout=300000,

1204

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

1205

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

1206

Interval=120000,

1207

Fingerprint=("778827225BA7DE539C5A"

1208

"7CFA59CFF7CDBD9A5920"),

1209

CheckerRunning=dbus.Boolean(False),

1210

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

1211

ApprovalPending=dbus.Boolean(False),

1212

ApprovedByDefault=dbus.Boolean(True),

1213

LastApprovalRequest="",

1214

ApprovalDelay=0,

1215

ApprovalDuration=1000,

1216

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

1217

ExtendedTimeout=900000,

1218

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

1219

LastCheckerStatus=0)

1220

self.other_client = MockClient(

1221

"barbar",

1222

KeyID=("0558568eedd67d622f5c83b35a115f79"

1223

"6ab612cff5ad227247e46c2b020f441c"),

1224

Secret=b"secretbar",

1225

Host="192.0.2.3",

1226

Enabled=dbus.Boolean(True),

1227

Timeout=300000,

1228

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

1229

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

1230

Interval=120000,

1231

Fingerprint=("3E393AEAEFB84C7E89E2"

1232

"F547B3A107558FCA3A27"),

1233

CheckerRunning=dbus.Boolean(True),

1234

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

1235

ApprovalPending=dbus.Boolean(False),

1236

ApprovedByDefault=dbus.Boolean(False),

1237

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

1238

ApprovalDelay=30000,

1239

ApprovalDuration=1000,

1240

Checker=":",

1241

ExtendedTimeout=900000,

1242

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

1243

LastCheckerStatus=-2)

1244

self.clients = collections.OrderedDict(

1245

[

1246

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

1247

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

1248

])

1249

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

1250

@property

1251

def bus(self):

1252

class Bus(object):

1253

@staticmethod

1254

def get_object(client_bus_name, path):

1255

self.assertEqual(client_bus_name, dbus_busname)

1256

return {

1257

"/clients/foo": self.client,

1258

"/clients/barbar": self.other_client,

1259

}[path]

1260

return Bus()

1261

1262

1263

class TestIsEnabledCmd(TestCmd):

1264

def test_is_enabled(self):

1265

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

1266

properties)

1267

for client, properties

1268

in self.clients.items()))

1269

def test_is_enabled_run_exits_successfully(self):

1270

with self.assertRaises(SystemExit) as e:

1271

IsEnabledCmd().run(self.one_client)

1272

if e.exception.code is not None:

1273

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

1274

else:

1275

self.assertIsNone(e.exception.code)

1276

def test_is_enabled_run_exits_with_failure(self):

1277

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

1278

with self.assertRaises(SystemExit) as e:

1279

IsEnabledCmd().run(self.one_client)

1280

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

1281

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

1282

else:

1283

self.assertIsNotNone(e.exception.code)

1284

1285

1286

class TestApproveCmd(TestCmd):

1287

def test_approve(self):

1288

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

1289

for clientpath in self.clients:

1290

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

1291

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

1292

client.calls)

1293

1294

1295

class TestDenyCmd(TestCmd):

1296

def test_deny(self):

1297

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

1298

for clientpath in self.clients:

1299

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

1300

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

1301

client.calls)

1302

1303

class TestRemoveCmd(TestCmd):

1304

def test_remove(self):

1305

class MockMandos(object):

1306

def __init__(self):

1307

self.calls = []

1308

def RemoveClient(self, dbus_path):

1309

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

1310

mandos = MockMandos()

1311

super(TestRemoveCmd, self).setUp()

1312

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

1313

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

1314

for clientpath in self.clients:

1315

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

1316

mandos.calls)

1317

1318

1319

class TestDumpJSONCmd(TestCmd):

1320

def setUp(self):

1321

self.expected_json = {

1322

"foo": {

1323

"Name": "foo",

1324

"KeyID": ("92ed150794387c03ce684574b1139a65"

1325

"94a34f895daaaf09fd8ea90a27cddb12"),

1326

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

1327

"Enabled": True,

1328

"Timeout": 300000,

1329

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

1330

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

1331

"Interval": 120000,

1332

"Fingerprint": ("778827225BA7DE539C5A"

1333

"7CFA59CFF7CDBD9A5920"),

1334

"CheckerRunning": False,

1335

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

1336

"ApprovalPending": False,

1337

"ApprovedByDefault": True,

1338

"LastApprovalRequest": "",

1339

"ApprovalDelay": 0,

1340

"ApprovalDuration": 1000,

1341

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

1342

"ExtendedTimeout": 900000,

1343

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

1344

"LastCheckerStatus": 0,

1345

},

1346

"barbar": {

1347

"Name": "barbar",

1348

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1349

"6ab612cff5ad227247e46c2b020f441c"),

1350

"Host": "192.0.2.3",

1351

"Enabled": True,

1352

"Timeout": 300000,

1353

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

1354

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

1355

"Interval": 120000,

1356

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1357

"F547B3A107558FCA3A27"),

1358

"CheckerRunning": True,

1359

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

1360

"ApprovalPending": False,

1361

"ApprovedByDefault": False,

1362

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

1363

"ApprovalDelay": 30000,

1364

"ApprovalDuration": 1000,

1365

"Checker": ":",

1366

"ExtendedTimeout": 900000,

1367

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

1368

"LastCheckerStatus": -2,

1369

},

1370

}

1371

return super(TestDumpJSONCmd, self).setUp()

1372

def test_normal(self):

1373

json_data = json.loads(DumpJSONCmd().output(self.clients))

1374

self.assertDictEqual(json_data, self.expected_json)

1375

def test_one_client(self):

1376

clients = self.one_client

1377

json_data = json.loads(DumpJSONCmd().output(clients))

1378

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

1379

self.assertDictEqual(json_data, expected_json)

1380

1381

1382

class TestPrintTableCmd(TestCmd):

1383

def test_normal(self):

1384

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

1385

expected_output = "\n".join((

1386

"Name Enabled Timeout Last Successful Check",

1387

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

1388

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

1389

))

1390

self.assertEqual(output, expected_output)

1391

def test_verbose(self):

1392

output = PrintTableCmd(verbose=True).output(

1393

self.clients.values())

1394

columns = (

1395

(

1396

"Name ",

1397

"foo ",

1398

"barbar ",

1399

),(

1400

"Enabled ",

1401

"Yes ",

1402

"Yes ",

1403

),(

1404

"Timeout ",

1405

"00:05:00 ",

1406

"00:05:00 ",

1407

),(

1408

"Last Successful Check ",

1409

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

1410

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

1411

),(

1412

"Created ",

1413

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

1414

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

1415

),(

1416

"Interval ",

1417

"00:02:00 ",

1418

"00:02:00 ",

1419

),(

1420

"Host ",

1421

"foo.example.org ",

1422

"192.0.2.3 ",

1423

),(

1424

("Key ID "

1425

" "),

1426

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1427

"ea90a27cddb12 "),

1428

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1429

"46c2b020f441c "),

1430

),(

1431

"Fingerprint ",

1432

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1433

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1434

),(

1435

"Check Is Running ",

1436

"No ",

1437

"Yes ",

1438

),(

1439

"Last Enabled ",

1440

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

1441

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

1442

),(

1443

"Approval Is Pending ",

1444

"No ",

1445

"No ",

1446

),(

1447

"Approved By Default ",

1448

"Yes ",

1449

"No ",

1450

),(

1451

"Last Approval Request ",

1452

" ",

1453

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

1454

),(

1455

"Approval Delay ",

1456

"00:00:00 ",

1457

"00:00:30 ",

1458

),(

1459

"Approval Duration ",

1460

"00:00:01 ",

1461

"00:00:01 ",

1462

),(

1463

"Checker ",

1464

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

1465

": ",

1466

),(

1467

"Extended Timeout ",

1468

"00:15:00 ",

1469

"00:15:00 ",

1470

),(

1471

"Expires ",

1472

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

1473

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

1474

),(

1475

"Last Checker Status",

1476

"0 ",

1477

"-2 ",

1478

)

1479

)

1480

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

1481

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

1482

for rows in columns)

1483

for line in range(num_lines))

1484

self.assertEqual(output, expected_output)

1485

def test_one_client(self):

1486

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

1487

expected_output = """

1488

Name Enabled Timeout Last Successful Check

1489

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

1490

"""[1:-1]

1491

self.assertEqual(output, expected_output)

1492

1493

1494

class Test_milliseconds_to_string(unittest.TestCase):

1495

def test_all(self):

1496

self.assertEqual(milliseconds_to_string(93785000),

1497

"1T02:03:05")

1498

def test_no_days(self):

1499

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

1500

def test_all_zero(self):

1501

self.assertEqual(milliseconds_to_string(0), "00:00:00")

1502

def test_no_fractional_seconds(self):

1503

self.assertEqual(milliseconds_to_string(400), "00:00:00")

1504

self.assertEqual(milliseconds_to_string(900), "00:00:00")

1505

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

1506

1507

1508

class Unique(object):

1509

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

1510

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

1511

1512

1513

class TestPropertyCmd(TestCmd):

1514

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

1515

def runTest(self):

1516

if not hasattr(self, "command"):

1517

return

1518

values_to_get = getattr(self, "values_to_get",

1519

self.values_to_set)

1520

for value_to_set, value_to_get in zip(self.values_to_set,

1521

values_to_get):

1522

for clientpath in self.clients:

1523

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

1524

old_value = client.attributes[self.propname]

1525

self.assertNotIsInstance(old_value, Unique)

1526

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

1527

self.run_command(value_to_set, self.clients)

1528

for clientpath in self.clients:

1529

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

1530

value = client.attributes[self.propname]

1531

self.assertNotIsInstance(value, Unique)

1532

self.assertEqual(value, value_to_get)

1533

def run_command(self, value, clients):

1534

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

1535

1536

1537

class TestEnableCmd(TestCmd):

1538

def test_enable(self):

1539

for clientpath in self.clients:

1540

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

1541

client.attributes["Enabled"] = False

1542

1543

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

1544

1545

for clientpath in self.clients:

1546

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

1547

self.assertTrue(client.attributes["Enabled"])

1548

1549

1550

class TestDisableCmd(TestCmd):

1551

def test_disable(self):

1552

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

1553

for clientpath in self.clients:

1554

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

1555

self.assertFalse(client.attributes["Enabled"])

1556

1557

1558

class TestBumpTimeoutCmd(TestPropertyCmd):

1559

command = BumpTimeoutCmd

1560

propname = "LastCheckedOK"

1561

values_to_set = [""]

1562

1563

1564

class TestStartCheckerCmd(TestPropertyCmd):

1565

command = StartCheckerCmd

1566

propname = "CheckerRunning"

1567

values_to_set = [dbus.Boolean(True)]

1568

1569

1570

class TestStopCheckerCmd(TestPropertyCmd):

1571

command = StopCheckerCmd

1572

propname = "CheckerRunning"

1573

values_to_set = [dbus.Boolean(False)]

1574

1575

1576

class TestApproveByDefaultCmd(TestPropertyCmd):

1577

command = ApproveByDefaultCmd

1578

propname = "ApprovedByDefault"

1579

values_to_set = [dbus.Boolean(True)]

1580

1581

1582

class TestDenyByDefaultCmd(TestPropertyCmd):

1583

command = DenyByDefaultCmd

1584

propname = "ApprovedByDefault"

1585

values_to_set = [dbus.Boolean(False)]

1586

1587

1588

class TestPropertyValueCmd(TestPropertyCmd):

1589

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

1590

def runTest(self):

1591

if type(self) is TestPropertyValueCmd:

1592

return

1593

return super(TestPropertyValueCmd, self).runTest()

1594

def run_command(self, value, clients):

1595

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

1596

1597

1598

class TestSetCheckerCmd(TestPropertyValueCmd):

1599

command = SetCheckerCmd

1600

propname = "Checker"

1601

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

1602

1603

1604

class TestSetHostCmd(TestPropertyValueCmd):

1605

command = SetHostCmd

1606

propname = "Host"

1607

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

1608

1609

1610

class TestSetSecretCmd(TestPropertyValueCmd):

1611

command = SetSecretCmd

1612

propname = "Secret"

1613

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

1614

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

1615

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

1616

1617

1618

class TestSetTimeoutCmd(TestPropertyValueCmd):

1619

command = SetTimeoutCmd

1620

propname = "Timeout"

1621

values_to_set = [datetime.timedelta(),

1622

datetime.timedelta(minutes=5),

1623

datetime.timedelta(seconds=1),

1624

datetime.timedelta(weeks=1),

1625

datetime.timedelta(weeks=52)]

1626

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

1627

1628

1629

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1630

command = SetExtendedTimeoutCmd

1631

propname = "ExtendedTimeout"

1632

values_to_set = [datetime.timedelta(),

1633

datetime.timedelta(minutes=5),

1634

datetime.timedelta(seconds=1),

1635

datetime.timedelta(weeks=1),

1636

datetime.timedelta(weeks=52)]

1637

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

1638

1639

1640

class TestSetIntervalCmd(TestPropertyValueCmd):

1641

command = SetIntervalCmd

1642

propname = "Interval"

1643

values_to_set = [datetime.timedelta(),

1644

datetime.timedelta(minutes=5),

1645

datetime.timedelta(seconds=1),

1646

datetime.timedelta(weeks=1),

1647

datetime.timedelta(weeks=52)]

1648

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

1649

1650

1651

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1652

command = SetApprovalDelayCmd

1653

propname = "ApprovalDelay"

1654

values_to_set = [datetime.timedelta(),

1655

datetime.timedelta(minutes=5),

1656

datetime.timedelta(seconds=1),

1657

datetime.timedelta(weeks=1),

1658

datetime.timedelta(weeks=52)]

1659

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

1660

1661

1662

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1663

command = SetApprovalDurationCmd

1664

propname = "ApprovalDuration"

1665

values_to_set = [datetime.timedelta(),

1666

datetime.timedelta(minutes=5),

1667

datetime.timedelta(seconds=1),

1668

datetime.timedelta(weeks=1),

1669

datetime.timedelta(weeks=52)]

1670

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

1671

1672

1673

1674

def should_only_run_tests():

1675

parser = argparse.ArgumentParser(add_help=False)

1676

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

1677

args, unknown_args = parser.parse_known_args()

1678

run_tests = args.check

1679

if run_tests:

1680

# Remove --check argument from sys.argv

1681

sys.argv[1:] = unknown_args

1682

return run_tests

1683

1684

# Add all tests from doctest strings

1685

def load_tests(loader, tests, none):

1686

import doctest

1687

tests.addTests(doctest.DocTestSuite())

1688

return tests

1689

1690

if __name__ == "__main__":

1691

if should_only_run_tests():

1692

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

1693

unittest.main()

1694

else:

1695

main()