/mandos/trunk : revision 24.1.51

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 OutputCmd(Command):

572

"""Abstract class for commands outputting 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(OutputCmd):

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(OutputCmd):

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

@classmethod

670

def valuetostring(cls, 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 cls.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

@staticmethod

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)

691

if td.days else "",

692

hours=td.seconds // 3600,

693

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

694

seconds=td.seconds % 60))

695

696

697

class PropertyCmd(Command):

698

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

699

def run_on_one_client(self, client, properties):

700

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

701

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

702

client.__dbus_object_path__,

703

dbus.PROPERTIES_IFACE, client_dbus_interface,

704

self.propname, self.value_to_set

705

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

706

else bool(self.value_to_set))

707

client.Set(client_dbus_interface, self.propname,

708

self.value_to_set,

709

dbus_interface=dbus.PROPERTIES_IFACE)

710

@property

711

def propname(self):

712

raise NotImplementedError()

713

714

715

class EnableCmd(PropertyCmd):

716

propname = "Enabled"

717

value_to_set = dbus.Boolean(True)

718

719

720

class DisableCmd(PropertyCmd):

721

propname = "Enabled"

722

value_to_set = dbus.Boolean(False)

723

724

725

class BumpTimeoutCmd(PropertyCmd):

726

propname = "LastCheckedOK"

727

value_to_set = ""

728

729

730

class StartCheckerCmd(PropertyCmd):

731

propname = "CheckerRunning"

732

value_to_set = dbus.Boolean(True)

733

734

735

class StopCheckerCmd(PropertyCmd):

736

propname = "CheckerRunning"

737

value_to_set = dbus.Boolean(False)

738

739

740

class ApproveByDefaultCmd(PropertyCmd):

741

propname = "ApprovedByDefault"

742

value_to_set = dbus.Boolean(True)

743

744

745

class DenyByDefaultCmd(PropertyCmd):

746

propname = "ApprovedByDefault"

747

value_to_set = dbus.Boolean(False)

748

749

750

class PropertyValueCmd(PropertyCmd):

751

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

752

def __init__(self, value):

753

self.value_to_set = value

754

755

756

class SetCheckerCmd(PropertyValueCmd):

757

propname = "Checker"

758

759

760

class SetHostCmd(PropertyValueCmd):

761

propname = "Host"

762

763

764

class SetSecretCmd(PropertyValueCmd):

765

propname = "Secret"

766

@property

767

def value_to_set(self):

768

return self._vts

769

@value_to_set.setter

770

def value_to_set(self, value):

771

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

772

self._vts = value.read()

773

value.close()

774

775

776

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

777

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

778

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

779

@property

780

def value_to_set(self):

781

return self._vts

782

@value_to_set.setter

783

def value_to_set(self, value):

784

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

785

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

786

787

788

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

789

propname = "Timeout"

790

791

792

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

793

propname = "ExtendedTimeout"

794

795

796

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

797

propname = "Interval"

798

799

800

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

801

propname = "ApprovalDelay"

802

803

804

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

805

propname = "ApprovalDuration"

806

807

808

809

class Test_string_to_delta(unittest.TestCase):

810

def test_handles_basic_rfc3339(self):

811

self.assertEqual(string_to_delta("PT0S"),

812

datetime.timedelta())

813

self.assertEqual(string_to_delta("P0D"),

814

datetime.timedelta())

815

self.assertEqual(string_to_delta("PT1S"),

816

datetime.timedelta(0, 1))

817

self.assertEqual(string_to_delta("PT2H"),

818

datetime.timedelta(0, 7200))

819

def test_falls_back_to_pre_1_6_1_with_warning(self):

820

# assertLogs only exists in Python 3.4

821

if hasattr(self, "assertLogs"):

822

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

823

value = string_to_delta("2h")

824

else:

825

class WarningFilter(logging.Filter):

826

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

827

def filter(self, record):

828

is_warning = record.levelno >= logging.WARNING

829

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

830

False)

831

return not is_warning

832

warning_filter = WarningFilter()

833

log.addFilter(warning_filter)

834

try:

835

value = string_to_delta("2h")

836

finally:

837

log.removeFilter(warning_filter)

838

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

839

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

840

841

842

class Test_check_option_syntax(unittest.TestCase):

843

def setUp(self):

844

self.parser = argparse.ArgumentParser()

845

add_command_line_options(self.parser)

846

847

def test_actions_requires_client_or_all(self):

848

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

849

options = self.parser.parse_args()

850

setattr(options, action, value)

851

with self.assertParseError():

852

self.check_option_syntax(options)

853

854

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

855

# check_option_syntax()

856

actions = {

857

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

858

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

859

# never used

860

"enable": True,

861

"disable": True,

862

"bump_timeout": True,

863

"start_checker": True,

864

"stop_checker": True,

865

"is_enabled": True,

866

"remove": True,

867

"checker": "x",

868

"timeout": datetime.timedelta(),

869

"extended_timeout": datetime.timedelta(),

870

"interval": datetime.timedelta(),

871

"approved_by_default": True,

872

"approval_delay": datetime.timedelta(),

873

"approval_duration": datetime.timedelta(),

874

"host": "x",

875

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

876

"approve": True,

877

"deny": True,

878

}

879

880

@contextlib.contextmanager

881

def assertParseError(self):

882

with self.assertRaises(SystemExit) as e:

883

with self.temporarily_suppress_stderr():

884

yield

885

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

886

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

887

# /argparse.html#exiting-methods

888

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

889

890

@staticmethod

891

@contextlib.contextmanager

892

def temporarily_suppress_stderr():

893

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

894

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

895

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

896

os.close(null)

897

try:

898

yield

899

finally:

900

# restore stderr

901

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

902

os.close(stderrcopy)

903

904

def check_option_syntax(self, options):

905

check_option_syntax(self.parser, options)

906

907

def test_actions_conflicts_with_verbose(self):

908

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

909

options = self.parser.parse_args()

910

setattr(options, action, value)

911

options.verbose = True

912

with self.assertParseError():

913

self.check_option_syntax(options)

914

915

def test_dump_json_conflicts_with_verbose(self):

916

options = self.parser.parse_args()

917

options.dump_json = True

918

options.verbose = True

919

with self.assertParseError():

920

self.check_option_syntax(options)

921

922

def test_dump_json_conflicts_with_action(self):

923

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

924

options = self.parser.parse_args()

925

setattr(options, action, value)

926

options.dump_json = True

927

with self.assertParseError():

928

self.check_option_syntax(options)

929

930

def test_all_can_not_be_alone(self):

931

options = self.parser.parse_args()

932

options.all = True

933

with self.assertParseError():

934

self.check_option_syntax(options)

935

936

def test_all_is_ok_with_any_action(self):

937

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

938

options = self.parser.parse_args()

939

setattr(options, action, value)

940

options.all = True

941

self.check_option_syntax(options)

942

943

def test_is_enabled_fails_without_client(self):

944

options = self.parser.parse_args()

945

options.is_enabled = True

946

with self.assertParseError():

947

self.check_option_syntax(options)

948

949

def test_is_enabled_works_with_one_client(self):

950

options = self.parser.parse_args()

951

options.is_enabled = True

952

options.client = ["foo"]

953

self.check_option_syntax(options)

954

955

def test_is_enabled_fails_with_two_clients(self):

956

options = self.parser.parse_args()

957

options.is_enabled = True

958

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

959

with self.assertParseError():

960

self.check_option_syntax(options)

961

962

def test_remove_can_only_be_combined_with_action_deny(self):

963

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

964

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

965

continue

966

options = self.parser.parse_args()

967

setattr(options, action, value)

968

options.all = True

969

options.remove = True

970

with self.assertParseError():

971

self.check_option_syntax(options)

972

973

974

class Test_commands_from_options(unittest.TestCase):

975

def setUp(self):

976

self.parser = argparse.ArgumentParser()

977

add_command_line_options(self.parser)

978

979

def test_is_enabled(self):

980

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

981

IsEnabledCmd)

982

983

def assert_command_from_args(self, args, command_cls,

984

**cmd_attrs):

985

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

986

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

987

options = self.parser.parse_args(args)

988

check_option_syntax(self.parser, options)

989

commands = commands_from_options(options)

990

self.assertEqual(len(commands), 1)

991

command = commands[0]

992

self.assertIsInstance(command, command_cls)

993

for key, value in cmd_attrs.items():

994

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

995

996

def test_is_enabled_short(self):

997

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

998

999

def test_approve(self):

1000

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

1001

ApproveCmd)

1002

1003

def test_approve_short(self):

1004

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

1005

1006

def test_deny(self):

1007

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

1008

1009

def test_deny_short(self):

1010

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

1011

1012

def test_remove(self):

1013

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

1014

RemoveCmd)

1015

1016

def test_deny_before_remove(self):

1017

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

1018

"foo"])

1019

check_option_syntax(self.parser, options)

1020

commands = commands_from_options(options)

1021

self.assertEqual(len(commands), 2)

1022

self.assertIsInstance(commands[0], DenyCmd)

1023

self.assertIsInstance(commands[1], RemoveCmd)

1024

1025

def test_deny_before_remove_reversed(self):

1026

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

1027

"--all"])

1028

check_option_syntax(self.parser, options)

1029

commands = commands_from_options(options)

1030

self.assertEqual(len(commands), 2)

1031

self.assertIsInstance(commands[0], DenyCmd)

1032

self.assertIsInstance(commands[1], RemoveCmd)

1033

1034

def test_remove_short(self):

1035

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

1036

1037

def test_dump_json(self):

1038

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

1039

1040

def test_enable(self):

1041

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

1042

1043

def test_enable_short(self):

1044

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

1045

1046

def test_disable(self):

1047

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

1048

DisableCmd)

1049

1050

def test_disable_short(self):

1051

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

1052

1053

def test_bump_timeout(self):

1054

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

1055

BumpTimeoutCmd)

1056

1057

def test_bump_timeout_short(self):

1058

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

1059

1060

def test_start_checker(self):

1061

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

1062

StartCheckerCmd)

1063

1064

def test_stop_checker(self):

1065

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

1066

StopCheckerCmd)

1067

1068

def test_approve_by_default(self):

1069

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

1070

ApproveByDefaultCmd)

1071

1072

def test_deny_by_default(self):

1073

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

1074

DenyByDefaultCmd)

1075

1076

def test_checker(self):

1077

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

1078

SetCheckerCmd, value_to_set=":")

1079

1080

def test_checker_empty(self):

1081

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

1082

SetCheckerCmd, value_to_set="")

1083

1084

def test_checker_short(self):

1085

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

1086

SetCheckerCmd, value_to_set=":")

1087

1088

def test_host(self):

1089

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

1090

"foo"], SetHostCmd,

1091

value_to_set="foo.example.org")

1092

1093

def test_host_short(self):

1094

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

1095

"foo"], SetHostCmd,

1096

value_to_set="foo.example.org")

1097

1098

def test_secret_devnull(self):

1099

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

1100

"foo"], SetSecretCmd,

1101

value_to_set=b"")

1102

1103

def test_secret_tempfile(self):

1104

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

1105

value = b"secret\0xyzzy\nbar"

1106

f.write(value)

1107

f.seek(0)

1108

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

1109

"foo"], SetSecretCmd,

1110

value_to_set=value)

1111

1112

def test_secret_devnull_short(self):

1113

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

1114

SetSecretCmd, value_to_set=b"")

1115

1116

def test_secret_tempfile_short(self):

1117

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

1118

value = b"secret\0xyzzy\nbar"

1119

f.write(value)

1120

f.seek(0)

1121

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

1122

SetSecretCmd,

1123

value_to_set=value)

1124

1125

def test_timeout(self):

1126

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

1127

SetTimeoutCmd,

1128

value_to_set=300000)

1129

1130

def test_timeout_short(self):

1131

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

1132

SetTimeoutCmd,

1133

value_to_set=300000)

1134

1135

def test_extended_timeout(self):

1136

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

1137

"foo"],

1138

SetExtendedTimeoutCmd,

1139

value_to_set=900000)

1140

1141

def test_interval(self):

1142

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

1143

SetIntervalCmd,

1144

value_to_set=120000)

1145

1146

def test_interval_short(self):

1147

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

1148

SetIntervalCmd,

1149

value_to_set=120000)

1150

1151

def test_approval_delay(self):

1152

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

1153

"foo"], SetApprovalDelayCmd,

1154

value_to_set=30000)

1155

1156

def test_approval_duration(self):

1157

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

1158

"foo"], SetApprovalDurationCmd,

1159

value_to_set=1000)

1160

1161

def test_print_table(self):

1162

self.assert_command_from_args([], PrintTableCmd,

1163

verbose=False)

1164

1165

def test_print_table_verbose(self):

1166

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

1167

verbose=True)

1168

1169

def test_print_table_verbose_short(self):

1170

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

1171

verbose=True)

1172

1173

1174

class TestCmd(unittest.TestCase):

1175

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

1176

def setUp(self):

1177

testcase = self

1178

class MockClient(object):

1179

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

1180

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

1181

self.attributes = attributes

1182

self.attributes["Name"] = name

1183

self.calls = []

1184

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

1185

testcase.assertEqual(interface, client_dbus_interface)

1186

testcase.assertEqual(dbus_interface,

1187

dbus.PROPERTIES_IFACE)

1188

self.attributes[propname] = value

1189

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

1190

testcase.assertEqual(interface, client_dbus_interface)

1191

testcase.assertEqual(dbus_interface,

1192

dbus.PROPERTIES_IFACE)

1193

return self.attributes[propname]

1194

def Approve(self, approve, dbus_interface):

1195

testcase.assertEqual(dbus_interface,

1196

client_dbus_interface)

1197

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

1198

dbus_interface)))

1199

self.client = MockClient(

1200

"foo",

1201

KeyID=("92ed150794387c03ce684574b1139a65"

1202

"94a34f895daaaf09fd8ea90a27cddb12"),

1203

Secret=b"secret",

1204

Host="foo.example.org",

1205

Enabled=dbus.Boolean(True),

1206

Timeout=300000,

1207

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

1208

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

1209

Interval=120000,

1210

Fingerprint=("778827225BA7DE539C5A"

1211

"7CFA59CFF7CDBD9A5920"),

1212

CheckerRunning=dbus.Boolean(False),

1213

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

1214

ApprovalPending=dbus.Boolean(False),

1215

ApprovedByDefault=dbus.Boolean(True),

1216

LastApprovalRequest="",

1217

ApprovalDelay=0,

1218

ApprovalDuration=1000,

1219

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

1220

ExtendedTimeout=900000,

1221

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

1222

LastCheckerStatus=0)

1223

self.other_client = MockClient(

1224

"barbar",

1225

KeyID=("0558568eedd67d622f5c83b35a115f79"

1226

"6ab612cff5ad227247e46c2b020f441c"),

1227

Secret=b"secretbar",

1228

Host="192.0.2.3",

1229

Enabled=dbus.Boolean(True),

1230

Timeout=300000,

1231

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

1232

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

1233

Interval=120000,

1234

Fingerprint=("3E393AEAEFB84C7E89E2"

1235

"F547B3A107558FCA3A27"),

1236

CheckerRunning=dbus.Boolean(True),

1237

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

1238

ApprovalPending=dbus.Boolean(False),

1239

ApprovedByDefault=dbus.Boolean(False),

1240

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

1241

ApprovalDelay=30000,

1242

ApprovalDuration=93785000,

1243

Checker=":",

1244

ExtendedTimeout=900000,

1245

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

1246

LastCheckerStatus=-2)

1247

self.clients = collections.OrderedDict(

1248

[

1249

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

1250

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

1251

])

1252

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

1253

@property

1254

def bus(self):

1255

class Bus(object):

1256

@staticmethod

1257

def get_object(client_bus_name, path):

1258

self.assertEqual(client_bus_name, dbus_busname)

1259

return {

1260

"/clients/foo": self.client,

1261

"/clients/barbar": self.other_client,

1262

}[path]

1263

return Bus()

1264

1265

1266

class TestIsEnabledCmd(TestCmd):

1267

def test_is_enabled(self):

1268

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

1269

properties)

1270

for client, properties

1271

in self.clients.items()))

1272

def test_is_enabled_run_exits_successfully(self):

1273

with self.assertRaises(SystemExit) as e:

1274

IsEnabledCmd().run(self.one_client)

1275

if e.exception.code is not None:

1276

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

1277

else:

1278

self.assertIsNone(e.exception.code)

1279

def test_is_enabled_run_exits_with_failure(self):

1280

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

1281

with self.assertRaises(SystemExit) as e:

1282

IsEnabledCmd().run(self.one_client)

1283

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

1284

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

1285

else:

1286

self.assertIsNotNone(e.exception.code)

1287

1288

1289

class TestApproveCmd(TestCmd):

1290

def test_approve(self):

1291

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

1292

for clientpath in self.clients:

1293

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

1294

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

1295

client.calls)

1296

1297

1298

class TestDenyCmd(TestCmd):

1299

def test_deny(self):

1300

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

1301

for clientpath in self.clients:

1302

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

1303

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

1304

client.calls)

1305

1306

class TestRemoveCmd(TestCmd):

1307

def test_remove(self):

1308

class MockMandos(object):

1309

def __init__(self):

1310

self.calls = []

1311

def RemoveClient(self, dbus_path):

1312

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

1313

mandos = MockMandos()

1314

super(TestRemoveCmd, self).setUp()

1315

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

1316

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

1317

for clientpath in self.clients:

1318

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

1319

mandos.calls)

1320

1321

1322

class TestDumpJSONCmd(TestCmd):

1323

def setUp(self):

1324

self.expected_json = {

1325

"foo": {

1326

"Name": "foo",

1327

"KeyID": ("92ed150794387c03ce684574b1139a65"

1328

"94a34f895daaaf09fd8ea90a27cddb12"),

1329

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

1330

"Enabled": True,

1331

"Timeout": 300000,

1332

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

1333

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

1334

"Interval": 120000,

1335

"Fingerprint": ("778827225BA7DE539C5A"

1336

"7CFA59CFF7CDBD9A5920"),

1337

"CheckerRunning": False,

1338

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

1339

"ApprovalPending": False,

1340

"ApprovedByDefault": True,

1341

"LastApprovalRequest": "",

1342

"ApprovalDelay": 0,

1343

"ApprovalDuration": 1000,

1344

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

1345

"ExtendedTimeout": 900000,

1346

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

1347

"LastCheckerStatus": 0,

1348

},

1349

"barbar": {

1350

"Name": "barbar",

1351

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1352

"6ab612cff5ad227247e46c2b020f441c"),

1353

"Host": "192.0.2.3",

1354

"Enabled": True,

1355

"Timeout": 300000,

1356

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

1357

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

1358

"Interval": 120000,

1359

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1360

"F547B3A107558FCA3A27"),

1361

"CheckerRunning": True,

1362

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

1363

"ApprovalPending": False,

1364

"ApprovedByDefault": False,

1365

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

1366

"ApprovalDelay": 30000,

1367

"ApprovalDuration": 93785000,

1368

"Checker": ":",

1369

"ExtendedTimeout": 900000,

1370

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

1371

"LastCheckerStatus": -2,

1372

},

1373

}

1374

return super(TestDumpJSONCmd, self).setUp()

1375

def test_normal(self):

1376

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

1377

self.assertDictEqual(json_data, self.expected_json)

1378

def test_one_client(self):

1379

clients = self.one_client

1380

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

1381

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

1382

self.assertDictEqual(json_data, expected_json)

1383

1384

1385

class TestPrintTableCmd(TestCmd):

1386

def test_normal(self):

1387

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

1388

expected_output = "\n".join((

1389

"Name Enabled Timeout Last Successful Check",

1390

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

1391

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

1392

))

1393

self.assertEqual(output, expected_output)

1394

def test_verbose(self):

1395

output = PrintTableCmd(verbose=True).output(

1396

self.clients.values())

1397

columns = (

1398

(

1399

"Name ",

1400

"foo ",

1401

"barbar ",

1402

),(

1403

"Enabled ",

1404

"Yes ",

1405

"Yes ",

1406

),(

1407

"Timeout ",

1408

"00:05:00 ",

1409

"00:05:00 ",

1410

),(

1411

"Last Successful Check ",

1412

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

1413

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

1414

),(

1415

"Created ",

1416

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

1417

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

1418

),(

1419

"Interval ",

1420

"00:02:00 ",

1421

"00:02:00 ",

1422

),(

1423

"Host ",

1424

"foo.example.org ",

1425

"192.0.2.3 ",

1426

),(

1427

("Key ID "

1428

" "),

1429

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1430

"ea90a27cddb12 "),

1431

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1432

"46c2b020f441c "),

1433

),(

1434

"Fingerprint ",

1435

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1436

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1437

),(

1438

"Check Is Running ",

1439

"No ",

1440

"Yes ",

1441

),(

1442

"Last Enabled ",

1443

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

1444

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

1445

),(

1446

"Approval Is Pending ",

1447

"No ",

1448

"No ",

1449

),(

1450

"Approved By Default ",

1451

"Yes ",

1452

"No ",

1453

),(

1454

"Last Approval Request ",

1455

" ",

1456

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

1457

),(

1458

"Approval Delay ",

1459

"00:00:00 ",

1460

"00:00:30 ",

1461

),(

1462

"Approval Duration ",

1463

"00:00:01 ",

1464

"1T02:03:05 ",

1465

),(

1466

"Checker ",

1467

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

1468

": ",

1469

),(

1470

"Extended Timeout ",

1471

"00:15:00 ",

1472

"00:15:00 ",

1473

),(

1474

"Expires ",

1475

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

1476

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

1477

),(

1478

"Last Checker Status",

1479

"0 ",

1480

"-2 ",

1481

)

1482

)

1483

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

1484

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

1485

for rows in columns)

1486

for line in range(num_lines))

1487

self.assertEqual(output, expected_output)

1488

def test_one_client(self):

1489

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

1490

expected_output = "\n".join((

1491

"Name Enabled Timeout Last Successful Check",

1492

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

1493

))

1494

self.assertEqual(output, expected_output)

1495

1496

1497

class Unique(object):

1498

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

1499

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

1500

1501

1502

class TestPropertyCmd(TestCmd):

1503

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

1504

def runTest(self):

1505

if not hasattr(self, "command"):

1506

return

1507

values_to_get = getattr(self, "values_to_get",

1508

self.values_to_set)

1509

for value_to_set, value_to_get in zip(self.values_to_set,

1510

values_to_get):

1511

for clientpath in self.clients:

1512

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

1513

old_value = client.attributes[self.propname]

1514

self.assertNotIsInstance(old_value, Unique)

1515

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

1516

self.run_command(value_to_set, self.clients)

1517

for clientpath in self.clients:

1518

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

1519

value = client.attributes[self.propname]

1520

self.assertNotIsInstance(value, Unique)

1521

self.assertEqual(value, value_to_get)

1522

def run_command(self, value, clients):

1523

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

1524

1525

1526

class TestEnableCmd(TestPropertyCmd):

1527

command = EnableCmd

1528

propname = "Enabled"

1529

values_to_set = [dbus.Boolean(True)]

1530

1531

1532

class TestDisableCmd(TestPropertyCmd):

1533

command = DisableCmd

1534

propname = "Enabled"

1535

values_to_set = [dbus.Boolean(False)]

1536

1537

1538

class TestBumpTimeoutCmd(TestPropertyCmd):

1539

command = BumpTimeoutCmd

1540

propname = "LastCheckedOK"

1541

values_to_set = [""]

1542

1543

1544

class TestStartCheckerCmd(TestPropertyCmd):

1545

command = StartCheckerCmd

1546

propname = "CheckerRunning"

1547

values_to_set = [dbus.Boolean(True)]

1548

1549

1550

class TestStopCheckerCmd(TestPropertyCmd):

1551

command = StopCheckerCmd

1552

propname = "CheckerRunning"

1553

values_to_set = [dbus.Boolean(False)]

1554

1555

1556

class TestApproveByDefaultCmd(TestPropertyCmd):

1557

command = ApproveByDefaultCmd

1558

propname = "ApprovedByDefault"

1559

values_to_set = [dbus.Boolean(True)]

1560

1561

1562

class TestDenyByDefaultCmd(TestPropertyCmd):

1563

command = DenyByDefaultCmd

1564

propname = "ApprovedByDefault"

1565

values_to_set = [dbus.Boolean(False)]

1566

1567

1568

class TestPropertyValueCmd(TestPropertyCmd):

1569

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

1570

def runTest(self):

1571

if type(self) is TestPropertyValueCmd:

1572

return

1573

return super(TestPropertyValueCmd, self).runTest()

1574

def run_command(self, value, clients):

1575

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

1576

1577

1578

class TestSetCheckerCmd(TestPropertyValueCmd):

1579

command = SetCheckerCmd

1580

propname = "Checker"

1581

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

1582

1583

1584

class TestSetHostCmd(TestPropertyValueCmd):

1585

command = SetHostCmd

1586

propname = "Host"

1587

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

1588

1589

1590

class TestSetSecretCmd(TestPropertyValueCmd):

1591

command = SetSecretCmd

1592

propname = "Secret"

1593

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

1594

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

1595

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

1596

1597

1598

class TestSetTimeoutCmd(TestPropertyValueCmd):

1599

command = SetTimeoutCmd

1600

propname = "Timeout"

1601

values_to_set = [datetime.timedelta(),

1602

datetime.timedelta(minutes=5),

1603

datetime.timedelta(seconds=1),

1604

datetime.timedelta(weeks=1),

1605

datetime.timedelta(weeks=52)]

1606

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

1607

1608

1609

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1610

command = SetExtendedTimeoutCmd

1611

propname = "ExtendedTimeout"

1612

values_to_set = [datetime.timedelta(),

1613

datetime.timedelta(minutes=5),

1614

datetime.timedelta(seconds=1),

1615

datetime.timedelta(weeks=1),

1616

datetime.timedelta(weeks=52)]

1617

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

1618

1619

1620

class TestSetIntervalCmd(TestPropertyValueCmd):

1621

command = SetIntervalCmd

1622

propname = "Interval"

1623

values_to_set = [datetime.timedelta(),

1624

datetime.timedelta(minutes=5),

1625

datetime.timedelta(seconds=1),

1626

datetime.timedelta(weeks=1),

1627

datetime.timedelta(weeks=52)]

1628

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

1629

1630

1631

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1632

command = SetApprovalDelayCmd

1633

propname = "ApprovalDelay"

1634

values_to_set = [datetime.timedelta(),

1635

datetime.timedelta(minutes=5),

1636

datetime.timedelta(seconds=1),

1637

datetime.timedelta(weeks=1),

1638

datetime.timedelta(weeks=52)]

1639

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

1640

1641

1642

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1643

command = SetApprovalDurationCmd

1644

propname = "ApprovalDuration"

1645

values_to_set = [datetime.timedelta(),

1646

datetime.timedelta(minutes=5),

1647

datetime.timedelta(seconds=1),

1648

datetime.timedelta(weeks=1),

1649

datetime.timedelta(weeks=52)]

1650

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

1651

1652

1653

1654

def should_only_run_tests():

1655

parser = argparse.ArgumentParser(add_help=False)

1656

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

1657

args, unknown_args = parser.parse_known_args()

1658

run_tests = args.check

1659

if run_tests:

1660

# Remove --check argument from sys.argv

1661

sys.argv[1:] = unknown_args

1662

return run_tests

1663

1664

# Add all tests from doctest strings

1665

def load_tests(loader, tests, none):

1666

import doctest

1667

tests.addTests(doctest.DocTestSuite())

1668

return tests

1669

1670

if __name__ == "__main__":

1671

if should_only_run_tests():

1672

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

1673

unittest.main()

1674

else:

1675

main()