/mandos/trunk : revision 72

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

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

844

# check_option_syntax()

845

actions = {

846

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

847

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

848

# never used

849

"enable": True,

850

"disable": True,

851

"bump_timeout": True,

852

"start_checker": True,

853

"stop_checker": True,

854

"is_enabled": True,

855

"remove": True,

856

"checker": "x",

857

"timeout": datetime.timedelta(),

858

"extended_timeout": datetime.timedelta(),

859

"interval": datetime.timedelta(),

860

"approved_by_default": True,

861

"approval_delay": datetime.timedelta(),

862

"approval_duration": datetime.timedelta(),

863

"host": "x",

864

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

865

"approve": True,

866

"deny": True,

867

}

868

869

def setUp(self):

870

self.parser = argparse.ArgumentParser()

871

add_command_line_options(self.parser)

872

873

@contextlib.contextmanager

874

def assertParseError(self):

875

with self.assertRaises(SystemExit) as e:

876

with self.temporarily_suppress_stderr():

877

yield

878

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

879

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

880

# /argparse.html#exiting-methods

881

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

882

883

@staticmethod

884

@contextlib.contextmanager

885

def temporarily_suppress_stderr():

886

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

887

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

888

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

889

os.close(null)

890

try:

891

yield

892

finally:

893

# restore stderr

894

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

895

os.close(stderrcopy)

896

897

def check_option_syntax(self, options):

898

check_option_syntax(self.parser, options)

899

900

def test_actions_requires_client_or_all(self):

901

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

902

options = self.parser.parse_args()

903

setattr(options, action, value)

904

with self.assertParseError():

905

self.check_option_syntax(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_command_from_options(unittest.TestCase):

975

def setUp(self):

976

self.parser = argparse.ArgumentParser()

977

add_command_line_options(self.parser)

978

def assert_command_from_args(self, args, command_cls,

979

**cmd_attrs):

980

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

981

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

982

options = self.parser.parse_args(args)

983

check_option_syntax(self.parser, options)

984

commands = commands_from_options(options)

985

self.assertEqual(len(commands), 1)

986

command = commands[0]

987

self.assertIsInstance(command, command_cls)

988

for key, value in cmd_attrs.items():

989

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

990

def test_print_table(self):

991

self.assert_command_from_args([], PrintTableCmd,

992

verbose=False)

993

994

def test_print_table_verbose(self):

995

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

996

verbose=True)

997

998

def test_print_table_verbose_short(self):

999

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

1000

verbose=True)

1001

1002

def test_enable(self):

1003

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

1004

1005

def test_enable_short(self):

1006

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

1007

1008

def test_disable(self):

1009

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

1010

DisableCmd)

1011

1012

def test_disable_short(self):

1013

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

1014

1015

def test_bump_timeout(self):

1016

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

1017

BumpTimeoutCmd)

1018

1019

def test_bump_timeout_short(self):

1020

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

1021

1022

def test_start_checker(self):

1023

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

1024

StartCheckerCmd)

1025

1026

def test_stop_checker(self):

1027

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

1028

StopCheckerCmd)

1029

1030

def test_remove(self):

1031

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

1032

RemoveCmd)

1033

1034

def test_remove_short(self):

1035

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

1036

1037

def test_checker(self):

1038

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

1039

SetCheckerCmd, value_to_set=":")

1040

1041

def test_checker_empty(self):

1042

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

1043

SetCheckerCmd, value_to_set="")

1044

1045

def test_checker_short(self):

1046

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

1047

SetCheckerCmd, value_to_set=":")

1048

1049

def test_timeout(self):

1050

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

1051

SetTimeoutCmd,

1052

value_to_set=300000)

1053

1054

def test_timeout_short(self):

1055

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

1056

SetTimeoutCmd,

1057

value_to_set=300000)

1058

1059

def test_extended_timeout(self):

1060

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

1061

"foo"],

1062

SetExtendedTimeoutCmd,

1063

value_to_set=900000)

1064

1065

def test_interval(self):

1066

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

1067

SetIntervalCmd,

1068

value_to_set=120000)

1069

1070

def test_interval_short(self):

1071

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

1072

SetIntervalCmd,

1073

value_to_set=120000)

1074

1075

def test_approve_by_default(self):

1076

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

1077

ApproveByDefaultCmd)

1078

1079

def test_deny_by_default(self):

1080

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

1081

DenyByDefaultCmd)

1082

1083

def test_approval_delay(self):

1084

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

1085

"foo"], SetApprovalDelayCmd,

1086

value_to_set=30000)

1087

1088

def test_approval_duration(self):

1089

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

1090

"foo"], SetApprovalDurationCmd,

1091

value_to_set=1000)

1092

1093

def test_host(self):

1094

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

1095

"foo"], SetHostCmd,

1096

value_to_set="foo.example.org")

1097

1098

def test_host_short(self):

1099

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

1100

"foo"], SetHostCmd,

1101

value_to_set="foo.example.org")

1102

1103

def test_secret_devnull(self):

1104

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

1105

"foo"], SetSecretCmd,

1106

value_to_set=b"")

1107

1108

def test_secret_tempfile(self):

1109

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

1110

value = b"secret\0xyzzy\nbar"

1111

f.write(value)

1112

f.seek(0)

1113

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

1114

"foo"], SetSecretCmd,

1115

value_to_set=value)

1116

1117

def test_secret_devnull_short(self):

1118

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

1119

SetSecretCmd, value_to_set=b"")

1120

1121

def test_secret_tempfile_short(self):

1122

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

1123

value = b"secret\0xyzzy\nbar"

1124

f.write(value)

1125

f.seek(0)

1126

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

1127

SetSecretCmd,

1128

value_to_set=value)

1129

1130

def test_approve(self):

1131

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

1132

ApproveCmd)

1133

1134

def test_approve_short(self):

1135

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

1136

1137

def test_deny(self):

1138

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

1139

1140

def test_deny_short(self):

1141

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

1142

1143

def test_dump_json(self):

1144

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

1145

1146

def test_is_enabled(self):

1147

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

1148

IsEnabledCmd)

1149

1150

def test_is_enabled_short(self):

1151

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

1152

1153

def test_deny_before_remove(self):

1154

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

1155

"foo"])

1156

check_option_syntax(self.parser, options)

1157

commands = commands_from_options(options)

1158

self.assertEqual(len(commands), 2)

1159

self.assertIsInstance(commands[0], DenyCmd)

1160

self.assertIsInstance(commands[1], RemoveCmd)

1161

1162

def test_deny_before_remove_reversed(self):

1163

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

1164

"--all"])

1165

check_option_syntax(self.parser, options)

1166

commands = commands_from_options(options)

1167

self.assertEqual(len(commands), 2)

1168

self.assertIsInstance(commands[0], DenyCmd)

1169

self.assertIsInstance(commands[1], RemoveCmd)

1170

1171

1172

class TestCmd(unittest.TestCase):

1173

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

1174

def setUp(self):

1175

testcase = self

1176

class MockClient(object):

1177

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

1178

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

1179

self.attributes = attributes

1180

self.attributes["Name"] = name

1181

self.calls = []

1182

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

1183

testcase.assertEqual(interface, client_dbus_interface)

1184

testcase.assertEqual(dbus_interface,

1185

dbus.PROPERTIES_IFACE)

1186

self.attributes[propname] = value

1187

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

1188

testcase.assertEqual(interface, client_dbus_interface)

1189

testcase.assertEqual(dbus_interface,

1190

dbus.PROPERTIES_IFACE)

1191

return self.attributes[propname]

1192

def Approve(self, approve, dbus_interface):

1193

testcase.assertEqual(dbus_interface,

1194

client_dbus_interface)

1195

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

1196

dbus_interface)))

1197

self.client = MockClient(

1198

"foo",

1199

KeyID=("92ed150794387c03ce684574b1139a65"

1200

"94a34f895daaaf09fd8ea90a27cddb12"),

1201

Secret=b"secret",

1202

Host="foo.example.org",

1203

Enabled=dbus.Boolean(True),

1204

Timeout=300000,

1205

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

1206

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

1207

Interval=120000,

1208

Fingerprint=("778827225BA7DE539C5A"

1209

"7CFA59CFF7CDBD9A5920"),

1210

CheckerRunning=dbus.Boolean(False),

1211

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

1212

ApprovalPending=dbus.Boolean(False),

1213

ApprovedByDefault=dbus.Boolean(True),

1214

LastApprovalRequest="",

1215

ApprovalDelay=0,

1216

ApprovalDuration=1000,

1217

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

1218

ExtendedTimeout=900000,

1219

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

1220

LastCheckerStatus=0)

1221

self.other_client = MockClient(

1222

"barbar",

1223

KeyID=("0558568eedd67d622f5c83b35a115f79"

1224

"6ab612cff5ad227247e46c2b020f441c"),

1225

Secret=b"secretbar",

1226

Host="192.0.2.3",

1227

Enabled=dbus.Boolean(True),

1228

Timeout=300000,

1229

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

1230

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

1231

Interval=120000,

1232

Fingerprint=("3E393AEAEFB84C7E89E2"

1233

"F547B3A107558FCA3A27"),

1234

CheckerRunning=dbus.Boolean(True),

1235

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

1236

ApprovalPending=dbus.Boolean(False),

1237

ApprovedByDefault=dbus.Boolean(False),

1238

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

1239

ApprovalDelay=30000,

1240

ApprovalDuration=93785000,

1241

Checker=":",

1242

ExtendedTimeout=900000,

1243

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

1244

LastCheckerStatus=-2)

1245

self.clients = collections.OrderedDict(

1246

[

1247

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

1248

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

1249

])

1250

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

1251

@property

1252

def bus(self):

1253

class Bus(object):

1254

@staticmethod

1255

def get_object(client_bus_name, path):

1256

self.assertEqual(client_bus_name, dbus_busname)

1257

return {

1258

"/clients/foo": self.client,

1259

"/clients/barbar": self.other_client,

1260

}[path]

1261

return Bus()

1262

1263

1264

class TestIsEnabledCmd(TestCmd):

1265

def test_is_enabled(self):

1266

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

1267

properties)

1268

for client, properties

1269

in self.clients.items()))

1270

def test_is_enabled_run_exits_successfully(self):

1271

with self.assertRaises(SystemExit) as e:

1272

IsEnabledCmd().run(self.one_client)

1273

if e.exception.code is not None:

1274

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

1275

else:

1276

self.assertIsNone(e.exception.code)

1277

def test_is_enabled_run_exits_with_failure(self):

1278

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

1279

with self.assertRaises(SystemExit) as e:

1280

IsEnabledCmd().run(self.one_client)

1281

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

1282

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

1283

else:

1284

self.assertIsNotNone(e.exception.code)

1285

1286

1287

class TestApproveCmd(TestCmd):

1288

def test_approve(self):

1289

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

1290

for clientpath in self.clients:

1291

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

1292

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

1293

client.calls)

1294

1295

1296

class TestDenyCmd(TestCmd):

1297

def test_deny(self):

1298

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

1299

for clientpath in self.clients:

1300

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

1301

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

1302

client.calls)

1303

1304

class TestRemoveCmd(TestCmd):

1305

def test_remove(self):

1306

class MockMandos(object):

1307

def __init__(self):

1308

self.calls = []

1309

def RemoveClient(self, dbus_path):

1310

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

1311

mandos = MockMandos()

1312

super(TestRemoveCmd, self).setUp()

1313

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

1314

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

1315

for clientpath in self.clients:

1316

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

1317

mandos.calls)

1318

1319

1320

class TestDumpJSONCmd(TestCmd):

1321

def setUp(self):

1322

self.expected_json = {

1323

"foo": {

1324

"Name": "foo",

1325

"KeyID": ("92ed150794387c03ce684574b1139a65"

1326

"94a34f895daaaf09fd8ea90a27cddb12"),

1327

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

1328

"Enabled": True,

1329

"Timeout": 300000,

1330

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

1331

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

1332

"Interval": 120000,

1333

"Fingerprint": ("778827225BA7DE539C5A"

1334

"7CFA59CFF7CDBD9A5920"),

1335

"CheckerRunning": False,

1336

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

1337

"ApprovalPending": False,

1338

"ApprovedByDefault": True,

1339

"LastApprovalRequest": "",

1340

"ApprovalDelay": 0,

1341

"ApprovalDuration": 1000,

1342

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

1343

"ExtendedTimeout": 900000,

1344

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

1345

"LastCheckerStatus": 0,

1346

},

1347

"barbar": {

1348

"Name": "barbar",

1349

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1350

"6ab612cff5ad227247e46c2b020f441c"),

1351

"Host": "192.0.2.3",

1352

"Enabled": True,

1353

"Timeout": 300000,

1354

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

1355

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

1356

"Interval": 120000,

1357

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1358

"F547B3A107558FCA3A27"),

1359

"CheckerRunning": True,

1360

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

1361

"ApprovalPending": False,

1362

"ApprovedByDefault": False,

1363

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

1364

"ApprovalDelay": 30000,

1365

"ApprovalDuration": 93785000,

1366

"Checker": ":",

1367

"ExtendedTimeout": 900000,

1368

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

1369

"LastCheckerStatus": -2,

1370

},

1371

}

1372

return super(TestDumpJSONCmd, self).setUp()

1373

def test_normal(self):

1374

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

1375

self.assertDictEqual(json_data, self.expected_json)

1376

def test_one_client(self):

1377

clients = self.one_client

1378

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

1379

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

1380

self.assertDictEqual(json_data, expected_json)

1381

1382

1383

class TestPrintTableCmd(TestCmd):

1384

def test_normal(self):

1385

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

1386

expected_output = "\n".join((

1387

"Name Enabled Timeout Last Successful Check",

1388

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

1389

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

1390

))

1391

self.assertEqual(output, expected_output)

1392

def test_verbose(self):

1393

output = PrintTableCmd(verbose=True).output(

1394

self.clients.values())

1395

columns = (

1396

(

1397

"Name ",

1398

"foo ",

1399

"barbar ",

1400

),(

1401

"Enabled ",

1402

"Yes ",

1403

"Yes ",

1404

),(

1405

"Timeout ",

1406

"00:05:00 ",

1407

"00:05:00 ",

1408

),(

1409

"Last Successful Check ",

1410

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

1411

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

1412

),(

1413

"Created ",

1414

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

1415

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

1416

),(

1417

"Interval ",

1418

"00:02:00 ",

1419

"00:02:00 ",

1420

),(

1421

"Host ",

1422

"foo.example.org ",

1423

"192.0.2.3 ",

1424

),(

1425

("Key ID "

1426

" "),

1427

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1428

"ea90a27cddb12 "),

1429

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1430

"46c2b020f441c "),

1431

),(

1432

"Fingerprint ",

1433

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1434

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1435

),(

1436

"Check Is Running ",

1437

"No ",

1438

"Yes ",

1439

),(

1440

"Last Enabled ",

1441

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

1442

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

1443

),(

1444

"Approval Is Pending ",

1445

"No ",

1446

"No ",

1447

),(

1448

"Approved By Default ",

1449

"Yes ",

1450

"No ",

1451

),(

1452

"Last Approval Request ",

1453

" ",

1454

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

1455

),(

1456

"Approval Delay ",

1457

"00:00:00 ",

1458

"00:00:30 ",

1459

),(

1460

"Approval Duration ",

1461

"00:00:01 ",

1462

"1T02:03:05 ",

1463

),(

1464

"Checker ",

1465

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

1466

": ",

1467

),(

1468

"Extended Timeout ",

1469

"00:15:00 ",

1470

"00:15:00 ",

1471

),(

1472

"Expires ",

1473

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

1474

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

1475

),(

1476

"Last Checker Status",

1477

"0 ",

1478

"-2 ",

1479

)

1480

)

1481

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

1482

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

1483

for rows in columns)

1484

for line in range(num_lines))

1485

self.assertEqual(output, expected_output)

1486

def test_one_client(self):

1487

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

1488

expected_output = "\n".join((

1489

"Name Enabled Timeout Last Successful Check",

1490

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

1491

))

1492

self.assertEqual(output, expected_output)

1493

1494

1495

class Unique(object):

1496

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

1497

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

1498

1499

1500

class TestPropertyCmd(TestCmd):

1501

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

1502

def runTest(self):

1503

if not hasattr(self, "command"):

1504

return

1505

values_to_get = getattr(self, "values_to_get",

1506

self.values_to_set)

1507

for value_to_set, value_to_get in zip(self.values_to_set,

1508

values_to_get):

1509

for clientpath in self.clients:

1510

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

1511

old_value = client.attributes[self.propname]

1512

self.assertNotIsInstance(old_value, Unique)

1513

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

1514

self.run_command(value_to_set, self.clients)

1515

for clientpath in self.clients:

1516

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

1517

value = client.attributes[self.propname]

1518

self.assertNotIsInstance(value, Unique)

1519

self.assertEqual(value, value_to_get)

1520

def run_command(self, value, clients):

1521

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

1522

1523

1524

class TestEnableCmd(TestPropertyCmd):

1525

command = EnableCmd

1526

propname = "Enabled"

1527

values_to_set = [dbus.Boolean(True)]

1528

1529

1530

class TestDisableCmd(TestPropertyCmd):

1531

command = DisableCmd

1532

propname = "Enabled"

1533

values_to_set = [dbus.Boolean(False)]

1534

1535

1536

class TestBumpTimeoutCmd(TestPropertyCmd):

1537

command = BumpTimeoutCmd

1538

propname = "LastCheckedOK"

1539

values_to_set = [""]

1540

1541

1542

class TestStartCheckerCmd(TestPropertyCmd):

1543

command = StartCheckerCmd

1544

propname = "CheckerRunning"

1545

values_to_set = [dbus.Boolean(True)]

1546

1547

1548

class TestStopCheckerCmd(TestPropertyCmd):

1549

command = StopCheckerCmd

1550

propname = "CheckerRunning"

1551

values_to_set = [dbus.Boolean(False)]

1552

1553

1554

class TestApproveByDefaultCmd(TestPropertyCmd):

1555

command = ApproveByDefaultCmd

1556

propname = "ApprovedByDefault"

1557

values_to_set = [dbus.Boolean(True)]

1558

1559

1560

class TestDenyByDefaultCmd(TestPropertyCmd):

1561

command = DenyByDefaultCmd

1562

propname = "ApprovedByDefault"

1563

values_to_set = [dbus.Boolean(False)]

1564

1565

1566

class TestPropertyValueCmd(TestPropertyCmd):

1567

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

1568

def runTest(self):

1569

if type(self) is TestPropertyValueCmd:

1570

return

1571

return super(TestPropertyValueCmd, self).runTest()

1572

def run_command(self, value, clients):

1573

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

1574

1575

1576

class TestSetCheckerCmd(TestPropertyValueCmd):

1577

command = SetCheckerCmd

1578

propname = "Checker"

1579

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

1580

1581

1582

class TestSetHostCmd(TestPropertyValueCmd):

1583

command = SetHostCmd

1584

propname = "Host"

1585

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

1586

1587

1588

class TestSetSecretCmd(TestPropertyValueCmd):

1589

command = SetSecretCmd

1590

propname = "Secret"

1591

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

1592

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

1593

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

1594

1595

1596

class TestSetTimeoutCmd(TestPropertyValueCmd):

1597

command = SetTimeoutCmd

1598

propname = "Timeout"

1599

values_to_set = [datetime.timedelta(),

1600

datetime.timedelta(minutes=5),

1601

datetime.timedelta(seconds=1),

1602

datetime.timedelta(weeks=1),

1603

datetime.timedelta(weeks=52)]

1604

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

1605

1606

1607

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1608

command = SetExtendedTimeoutCmd

1609

propname = "ExtendedTimeout"

1610

values_to_set = [datetime.timedelta(),

1611

datetime.timedelta(minutes=5),

1612

datetime.timedelta(seconds=1),

1613

datetime.timedelta(weeks=1),

1614

datetime.timedelta(weeks=52)]

1615

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

1616

1617

1618

class TestSetIntervalCmd(TestPropertyValueCmd):

1619

command = SetIntervalCmd

1620

propname = "Interval"

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 TestSetApprovalDelayCmd(TestPropertyValueCmd):

1630

command = SetApprovalDelayCmd

1631

propname = "ApprovalDelay"

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 TestSetApprovalDurationCmd(TestPropertyValueCmd):

1641

command = SetApprovalDurationCmd

1642

propname = "ApprovalDuration"

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

1652

def should_only_run_tests():

1653

parser = argparse.ArgumentParser(add_help=False)

1654

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

1655

args, unknown_args = parser.parse_known_args()

1656

run_tests = args.check

1657

if run_tests:

1658

# Remove --check argument from sys.argv

1659

sys.argv[1:] = unknown_args

1660

return run_tests

1661

1662

# Add all tests from doctest strings

1663

def load_tests(loader, tests, none):

1664

import doctest

1665

tests.addTests(doctest.DocTestSuite())

1666

return tests

1667

1668

if __name__ == "__main__":

1669

if should_only_run_tests():

1670

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

1671

unittest.main()

1672

else:

1673

main()