/mandos/trunk : revision 134

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

domain = "se.recompile"

68

busname = domain + ".Mandos"

69

server_path = "/"

70

server_interface = domain + ".Mandos"

71

client_interface = domain + ".Mandos.Client"

72

version = "1.8.3"

73

74

75

try:

76

dbus.OBJECT_MANAGER_IFACE

77

except AttributeError:

78

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

79

80

81

def milliseconds_to_string(ms):

82

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

83

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

84

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

85

hours=td.seconds // 3600,

86

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

87

seconds=td.seconds % 60))

88

89

90

def rfc3339_duration_to_delta(duration):

91

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

92

93

>>> rfc3339_duration_to_delta("P7D")

94

datetime.timedelta(7)

95

>>> rfc3339_duration_to_delta("PT60S")

96

datetime.timedelta(0, 60)

97

>>> rfc3339_duration_to_delta("PT60M")

98

datetime.timedelta(0, 3600)

99

>>> rfc3339_duration_to_delta("P60M")

100

datetime.timedelta(1680)

101

>>> rfc3339_duration_to_delta("PT24H")

102

datetime.timedelta(1)

103

>>> rfc3339_duration_to_delta("P1W")

104

datetime.timedelta(7)

105

>>> rfc3339_duration_to_delta("PT5M30S")

106

datetime.timedelta(0, 330)

107

>>> rfc3339_duration_to_delta("P1DT3M20S")

108

datetime.timedelta(1, 200)

109

>>> # Can not be empty:

110

>>> rfc3339_duration_to_delta("")

111

Traceback (most recent call last):

112

...

113

ValueError: Invalid RFC 3339 duration: u''

114

>>> # Must start with "P":

115

>>> rfc3339_duration_to_delta("1D")

116

Traceback (most recent call last):

117

...

118

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

119

>>> # Must use correct order

120

>>> rfc3339_duration_to_delta("PT1S2M")

121

Traceback (most recent call last):

122

...

123

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

124

>>> # Time needs time marker

125

>>> rfc3339_duration_to_delta("P1H2S")

126

Traceback (most recent call last):

127

...

128

ValueError: Invalid RFC 3339 duration: u'P1H2S'

129

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

130

>>> rfc3339_duration_to_delta("P1D2W")

131

Traceback (most recent call last):

132

...

133

ValueError: Invalid RFC 3339 duration: u'P1D2W'

134

>>> rfc3339_duration_to_delta("P2W2H")

135

Traceback (most recent call last):

136

...

137

ValueError: Invalid RFC 3339 duration: u'P2W2H'

138

"""

139

140

# Parsing an RFC 3339 duration with regular expressions is not

141

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

142

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

143

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

144

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

145

# avoid excessive use of external libraries.

146

147

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

148

Token = collections.namedtuple("Token", (

149

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

150

# a "group" containing digits

151

"value", # datetime.timedelta or None

152

"followers")) # Tokens valid after this token

153

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

154

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

155

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

156

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

157

datetime.timedelta(seconds=1),

158

frozenset((token_end, )))

159

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

160

datetime.timedelta(minutes=1),

161

frozenset((token_second, token_end)))

162

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

163

datetime.timedelta(hours=1),

164

frozenset((token_minute, token_end)))

165

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

166

None,

167

frozenset((token_hour, token_minute,

168

token_second)))

169

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

170

datetime.timedelta(days=1),

171

frozenset((token_time, token_end)))

172

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

173

datetime.timedelta(weeks=4),

174

frozenset((token_day, token_end)))

175

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

176

datetime.timedelta(weeks=52),

177

frozenset((token_month, token_end)))

178

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

179

datetime.timedelta(weeks=1),

180

frozenset((token_end, )))

181

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

182

frozenset((token_year, token_month,

183

token_day, token_time,

184

token_week)))

185

# Define starting values:

186

# Value so far

187

value = datetime.timedelta()

188

found_token = None

189

# Following valid tokens

190

followers = frozenset((token_duration, ))

191

# String left to parse

192

s = duration

193

# Loop until end token is found

194

while found_token is not token_end:

195

# Search for any currently valid tokens

196

for token in followers:

197

match = token.regexp.match(s)

198

if match is not None:

199

# Token found

200

if token.value is not None:

201

# Value found, parse digits

202

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

203

# Add to value so far

204

value += factor * token.value

205

# Strip token from string

206

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

207

# Go to found token

208

found_token = token

209

# Set valid next tokens

210

followers = found_token.followers

211

break

212

else:

213

# No currently valid tokens were found

214

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

215

.format(duration))

216

# End token found

217

return value

218

219

220

def string_to_delta(interval):

221

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

222

223

try:

224

return rfc3339_duration_to_delta(interval)

225

except ValueError as e:

226

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

227

' '.join(e.args))

228

return parse_pre_1_6_1_interval(interval)

229

230

231

def parse_pre_1_6_1_interval(interval):

232

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

233

and return a datetime.timedelta

234

235

>>> parse_pre_1_6_1_interval('7d')

236

datetime.timedelta(7)

237

>>> parse_pre_1_6_1_interval('60s')

238

datetime.timedelta(0, 60)

239

>>> parse_pre_1_6_1_interval('60m')

240

datetime.timedelta(0, 3600)

241

>>> parse_pre_1_6_1_interval('24h')

242

datetime.timedelta(1)

243

>>> parse_pre_1_6_1_interval('1w')

244

datetime.timedelta(7)

245

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

246

datetime.timedelta(0, 330)

247

>>> parse_pre_1_6_1_interval('')

248

datetime.timedelta(0)

249

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

250

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

251

datetime.timedelta(2, 480, 18000)

252

253

"""

254

255

value = datetime.timedelta(0)

256

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

257

258

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

259

if suffix == "d":

260

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

261

elif suffix == "s":

262

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

263

elif suffix == "m":

264

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

265

elif suffix == "h":

266

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

267

elif suffix == "w":

268

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

269

elif suffix == "":

270

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

271

return value

272

273

274

## Classes for commands.

275

276

# Abstract classes first

277

class Command(object):

278

"""Abstract class for commands"""

279

def run(self, mandos, clients):

280

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

281

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

282

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

283

self.mandos = mandos

284

for client, properties in clients.items():

285

self.run_on_one_client(client, properties)

286

287

class PrintCmd(Command):

288

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

289

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

290

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

291

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

292

"ApprovalPending", "ApprovedByDefault",

293

"LastApprovalRequest", "ApprovalDelay",

294

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

295

"Expires", "LastCheckerStatus")

296

def run(self, mandos, clients):

297

print(self.output(clients))

298

299

class PropertyCmd(Command):

300

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

301

def run_on_one_client(self, client, properties):

302

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

303

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

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

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

308

else bool(self.value_to_set))

309

client.Set(client_interface, self.property, self.value_to_set,

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

312

class ValueArgumentMixIn(object):

313

"""Mixin class for commands taking a value as argument"""

314

def __init__(self, value):

315

self.value_to_set = value

316

317

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

318

"""Mixin class for commands taking a value argument as

319

milliseconds."""

320

@property

321

def value_to_set(self):

322

return self._vts

323

@value_to_set.setter

324

def value_to_set(self, value):

325

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

326

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

327

328

# Actual (non-abstract) command classes

329

330

class PrintTableCmd(PrintCmd):

331

def __init__(self, verbose=False):

332

self.verbose = verbose

333

334

def output(self, clients):

335

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

336

keywords = default_keywords

337

if self.verbose:

338

keywords = self.all_keywords

339

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

340

341

class TableOfClients(object):

342

tableheaders = {

343

"Name": "Name",

344

"Enabled": "Enabled",

345

"Timeout": "Timeout",

346

"LastCheckedOK": "Last Successful Check",

347

"LastApprovalRequest": "Last Approval Request",

348

"Created": "Created",

349

"Interval": "Interval",

350

"Host": "Host",

351

"Fingerprint": "Fingerprint",

352

"KeyID": "Key ID",

353

"CheckerRunning": "Check Is Running",

354

"LastEnabled": "Last Enabled",

355

"ApprovalPending": "Approval Is Pending",

356

"ApprovedByDefault": "Approved By Default",

357

"ApprovalDelay": "Approval Delay",

358

"ApprovalDuration": "Approval Duration",

359

"Checker": "Checker",

360

"ExtendedTimeout": "Extended Timeout",

361

"Expires": "Expires",

362

"LastCheckerStatus": "Last Checker Status",

363

}

364

365

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

366

self.clients = clients

367

self.keywords = keywords

368

if tableheaders is not None:

369

self.tableheaders = tableheaders

370

371

def __str__(self):

372

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

373

374

if sys.version_info.major == 2:

375

__unicode__ = __str__

376

def __str__(self):

377

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

378

379

def rows(self):

380

format_string = self.row_formatting_string()

381

rows = [self.header_line(format_string)]

382

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

383

for client in self.clients)

384

return rows

385

386

def row_formatting_string(self):

387

"Format string used to format table rows"

388

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

389

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

390

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

391

for client in self.clients)),

392

key=key)

393

for key in self.keywords)

394

395

def string_from_client(self, client, key):

396

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

397

398

@staticmethod

399

def valuetostring(value, keyword):

400

if isinstance(value, dbus.Boolean):

401

return "Yes" if value else "No"

402

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

403

"ApprovalDuration", "ExtendedTimeout"):

404

return milliseconds_to_string(value)

405

return str(value)

406

407

def header_line(self, format_string):

408

return format_string.format(**self.tableheaders)

409

410

def client_line(self, client, format_string):

411

return format_string.format(

412

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

413

for key in self.keywords})

414

415

416

417

class DumpJSONCmd(PrintCmd):

418

def output(self, clients):

419

data = {client["Name"]:

420

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

421

for key in self.all_keywords}

422

for client in clients.values()}

423

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

424

@staticmethod

425

def dbus_boolean_to_bool(value):

426

if isinstance(value, dbus.Boolean):

427

value = bool(value)

428

return value

429

430

class IsEnabledCmd(Command):

431

def run_on_one_client(self, client, properties):

432

if self.is_enabled(client, properties):

433

sys.exit(0)

434

sys.exit(1)

435

def is_enabled(self, client, properties):

436

return bool(properties["Enabled"])

437

438

class RemoveCmd(Command):

439

def run_on_one_client(self, client, properties):

440

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

441

server_path, server_interface,

442

str(client.__dbus_object_path__))

443

self.mandos.RemoveClient(client.__dbus_object_path__)

444

445

class ApproveCmd(Command):

446

def run_on_one_client(self, client, properties):

447

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

448

client.__dbus_object_path__, client_interface)

449

client.Approve(dbus.Boolean(True),

450

dbus_interface=client_interface)

451

452

class DenyCmd(Command):

453

def run_on_one_client(self, client, properties):

454

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

455

client.__dbus_object_path__, client_interface)

456

client.Approve(dbus.Boolean(False),

457

dbus_interface=client_interface)

458

459

class EnableCmd(PropertyCmd):

460

property = "Enabled"

461

value_to_set = dbus.Boolean(True)

462

463

class DisableCmd(PropertyCmd):

464

property = "Enabled"

465

value_to_set = dbus.Boolean(False)

466

467

class BumpTimeoutCmd(PropertyCmd):

468

property = "LastCheckedOK"

469

value_to_set = ""

470

471

class StartCheckerCmd(PropertyCmd):

472

property = "CheckerRunning"

473

value_to_set = dbus.Boolean(True)

474

475

class StopCheckerCmd(PropertyCmd):

476

property = "CheckerRunning"

477

value_to_set = dbus.Boolean(False)

478

479

class ApproveByDefaultCmd(PropertyCmd):

480

property = "ApprovedByDefault"

481

value_to_set = dbus.Boolean(True)

482

483

class DenyByDefaultCmd(PropertyCmd):

484

property = "ApprovedByDefault"

485

value_to_set = dbus.Boolean(False)

486

487

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

488

property = "Checker"

489

490

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

491

property = "Host"

492

493

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

494

@property

495

def value_to_set(self):

496

return self._vts

497

@value_to_set.setter

498

def value_to_set(self, value):

499

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

500

self._vts = value.read()

501

value.close()

502

property = "Secret"

503

504

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

505

property = "Timeout"

506

507

class SetExtendedTimeoutCmd(PropertyCmd,

508

MillisecondsValueArgumentMixIn):

509

property = "ExtendedTimeout"

510

511

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

512

property = "Interval"

513

514

class SetApprovalDelayCmd(PropertyCmd,

515

MillisecondsValueArgumentMixIn):

516

property = "ApprovalDelay"

517

518

class SetApprovalDurationCmd(PropertyCmd,

519

MillisecondsValueArgumentMixIn):

520

property = "ApprovalDuration"

521

522

def add_command_line_options(parser):

523

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

524

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

525

help="show version number and exit")

526

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

527

help="Select all clients")

528

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

529

help="Print all fields")

530

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

531

help="Dump client data in JSON format")

532

enable_disable = parser.add_mutually_exclusive_group()

533

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

534

help="Enable client")

535

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

536

action="store_true",

537

help="disable client")

538

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

539

help="Bump timeout for client")

540

start_stop_checker = parser.add_mutually_exclusive_group()

541

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

542

action="store_true",

543

help="Start checker for client")

544

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

545

action="store_true",

546

help="Stop checker for client")

547

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

548

help="Check if client is enabled")

549

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

550

help="Remove client")

551

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

552

help="Set checker command for client")

553

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

554

help="Set timeout for client")

555

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

556

help="Set extended timeout for client")

557

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

558

help="Set checker interval for client")

559

approve_deny_default = parser.add_mutually_exclusive_group()

560

approve_deny_default.add_argument(

561

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

562

default=None, dest="approved_by_default",

563

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

564

approve_deny_default.add_argument(

565

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

566

dest="approved_by_default",

567

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

568

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

569

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

570

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

571

help="Set duration of one client approval")

572

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

573

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

574

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

575

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

576

approve_deny = parser.add_mutually_exclusive_group()

577

approve_deny.add_argument(

578

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

579

help="Approve any current client request")

580

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

581

help="Deny any current client request")

582

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

583

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

584

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

585

help="Run self-test")

586

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

587

588

589

def commands_from_options(options):

590

591

commands = []

592

593

if options.dump_json:

594

commands.append(DumpJSONCmd())

595

596

if options.enable:

597

commands.append(EnableCmd())

598

599

if options.disable:

600

commands.append(DisableCmd())

601

602

if options.bump_timeout:

603

commands.append(BumpTimeoutCmd())

604

605

if options.start_checker:

606

commands.append(StartCheckerCmd())

607

608

if options.stop_checker:

609

commands.append(StopCheckerCmd())

610

611

if options.is_enabled:

612

commands.append(IsEnabledCmd())

613

614

if options.remove:

615

commands.append(RemoveCmd())

616

617

if options.checker is not None:

618

commands.append(SetCheckerCmd(options.checker))

619

620

if options.timeout is not None:

621

commands.append(SetTimeoutCmd(options.timeout))

622

623

if options.extended_timeout:

624

commands.append(

625

SetExtendedTimeoutCmd(options.extended_timeout))

626

627

if options.interval is not None:

628

commands.append(SetIntervalCmd(options.interval))

629

630

if options.approved_by_default is not None:

631

if options.approved_by_default:

632

commands.append(ApproveByDefaultCmd())

633

else:

634

commands.append(DenyByDefaultCmd())

635

636

if options.approval_delay is not None:

637

commands.append(SetApprovalDelayCmd(options.approval_delay))

638

639

if options.approval_duration is not None:

640

commands.append(

641

SetApprovalDurationCmd(options.approval_duration))

642

643

if options.host is not None:

644

commands.append(SetHostCmd(options.host))

645

646

if options.secret is not None:

647

commands.append(SetSecretCmd(options.secret))

648

649

if options.approve:

650

commands.append(ApproveCmd())

651

652

if options.deny:

653

commands.append(DenyCmd())

654

655

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

656

# optionally verbosely

657

if not commands:

658

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

659

660

return commands

661

662

663

def check_option_syntax(parser, options):

664

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

665

argparse"""

666

667

def has_actions(options):

668

return any((options.enable,

669

options.disable,

670

options.bump_timeout,

671

options.start_checker,

672

options.stop_checker,

673

options.is_enabled,

674

options.remove,

675

options.checker is not None,

676

options.timeout is not None,

677

options.extended_timeout is not None,

678

options.interval is not None,

679

options.approved_by_default is not None,

680

options.approval_delay is not None,

681

options.approval_duration is not None,

682

options.host is not None,

683

options.secret is not None,

684

options.approve,

685

options.deny))

686

687

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

688

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

689

if options.verbose and has_actions(options):

690

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

691

if options.dump_json and (options.verbose

692

or has_actions(options)):

693

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

694

if options.all and not has_actions(options):

695

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

696

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

697

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

698

699

700

def main():

701

parser = argparse.ArgumentParser()

702

703

add_command_line_options(parser)

704

705

options = parser.parse_args()

706

707

check_option_syntax(parser, options)

708

709

clientnames = options.client

710

711

if options.debug:

712

log.setLevel(logging.DEBUG)

713

714

try:

715

bus = dbus.SystemBus()

716

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

717

server_path)

718

mandos_dbus_objc = bus.get_object(busname, server_path)

719

except dbus.exceptions.DBusException:

720

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

721

sys.exit(1)

722

723

mandos_serv = dbus.Interface(mandos_dbus_objc,

724

dbus_interface=server_interface)

725

mandos_serv_object_manager = dbus.Interface(

726

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

727

728

# Filter out log message from dbus module

729

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

730

class NullFilter(logging.Filter):

731

def filter(self, record):

732

return False

733

dbus_filter = NullFilter()

734

try:

735

dbus_logger.addFilter(dbus_filter)

736

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

737

server_path, dbus.OBJECT_MANAGER_IFACE)

738

mandos_clients = {path: ifs_and_props[client_interface]

739

for path, ifs_and_props in

740

mandos_serv_object_manager

741

.GetManagedObjects().items()

742

if client_interface in ifs_and_props}

743

except dbus.exceptions.DBusException as e:

744

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

745

"\n%s", e)

746

sys.exit(1)

747

finally:

748

# restore dbus logger

749

dbus_logger.removeFilter(dbus_filter)

750

751

# Compile dict of (clients: properties) to process

752

clients = {}

753

754

if not clientnames:

755

clients = {bus.get_object(busname, path): properties

756

for path, properties in mandos_clients.items()}

757

else:

758

for name in clientnames:

759

for path, client in mandos_clients.items():

760

if client["Name"] == name:

761

client_objc = bus.get_object(busname, path)

762

clients[client_objc] = client

763

break

764

else:

765

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

766

sys.exit(1)

767

768

# Run all commands on clients

769

commands = commands_from_options(options)

770

for command in commands:

771

command.run(mandos_serv, clients)

772

773

774

class Test_milliseconds_to_string(unittest.TestCase):

775

def test_all(self):

776

self.assertEqual(milliseconds_to_string(93785000),

777

"1T02:03:05")

778

def test_no_days(self):

779

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

780

def test_all_zero(self):

781

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

782

def test_no_fractional_seconds(self):

783

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

784

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

785

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

786

787

class Test_string_to_delta(unittest.TestCase):

788

def test_handles_basic_rfc3339(self):

789

self.assertEqual(string_to_delta("PT0S"),

790

datetime.timedelta())

791

self.assertEqual(string_to_delta("P0D"),

792

datetime.timedelta())

793

self.assertEqual(string_to_delta("PT1S"),

794

datetime.timedelta(0, 1))

795

self.assertEqual(string_to_delta("PT2H"),

796

datetime.timedelta(0, 7200))

797

def test_falls_back_to_pre_1_6_1_with_warning(self):

798

# assertLogs only exists in Python 3.4

799

if hasattr(self, "assertLogs"):

800

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

801

value = string_to_delta("2h")

802

else:

803

class WarningFilter(logging.Filter):

804

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

805

def filter(self, record):

806

is_warning = record.levelno >= logging.WARNING

807

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

808

False)

809

return not is_warning

810

warning_filter = WarningFilter()

811

log.addFilter(warning_filter)

812

try:

813

value = string_to_delta("2h")

814

finally:

815

log.removeFilter(warning_filter)

816

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

817

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

818

819

820

class TestCmd(unittest.TestCase):

821

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

822

def setUp(self):

823

testcase = self

824

class MockClient(object):

825

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

826

self.__dbus_object_path__ = "objpath_{}".format(name)

827

self.attributes = attributes

828

self.attributes["Name"] = name

829

self.calls = []

830

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

831

testcase.assertEqual(interface, client_interface)

832

testcase.assertEqual(dbus_interface,

833

dbus.PROPERTIES_IFACE)

834

self.attributes[property] = value

835

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

836

testcase.assertEqual(interface, client_interface)

837

testcase.assertEqual(dbus_interface,

838

dbus.PROPERTIES_IFACE)

839

return self.attributes[property]

840

def Approve(self, approve, dbus_interface):

841

testcase.assertEqual(dbus_interface, client_interface)

842

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

843

dbus_interface)))

844

self.client = MockClient(

845

"foo",

846

KeyID=("92ed150794387c03ce684574b1139a65"

847

"94a34f895daaaf09fd8ea90a27cddb12"),

848

Secret=b"secret",

849

Host="foo.example.org",

850

Enabled=dbus.Boolean(True),

851

Timeout=300000,

852

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

853

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

854

Interval=120000,

855

Fingerprint=("778827225BA7DE539C5A"

856

"7CFA59CFF7CDBD9A5920"),

857

CheckerRunning=dbus.Boolean(False),

858

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

859

ApprovalPending=dbus.Boolean(False),

860

ApprovedByDefault=dbus.Boolean(True),

861

LastApprovalRequest="",

862

ApprovalDelay=0,

863

ApprovalDuration=1000,

864

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

865

ExtendedTimeout=900000,

866

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

867

LastCheckerStatus=0)

868

self.other_client = MockClient(

869

"barbar",

870

KeyID=("0558568eedd67d622f5c83b35a115f79"

871

"6ab612cff5ad227247e46c2b020f441c"),

872

Secret=b"secretbar",

873

Host="192.0.2.3",

874

Enabled=dbus.Boolean(True),

875

Timeout=300000,

876

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

877

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

878

Interval=120000,

879

Fingerprint=("3E393AEAEFB84C7E89E2"

880

"F547B3A107558FCA3A27"),

881

CheckerRunning=dbus.Boolean(True),

882

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

883

ApprovalPending=dbus.Boolean(False),

884

ApprovedByDefault=dbus.Boolean(False),

885

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

886

ApprovalDelay=30000,

887

ApprovalDuration=1000,

888

Checker=":",

889

ExtendedTimeout=900000,

890

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

891

LastCheckerStatus=-2)

892

self.clients = collections.OrderedDict(

893

[

894

(self.client, self.client.attributes),

895

(self.other_client, self.other_client.attributes),

896

])

897

self.one_client = {self.client: self.client.attributes}

898

899

class TestPrintTableCmd(TestCmd):

900

def test_normal(self):

901

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

902

expected_output = """

903

Name Enabled Timeout Last Successful Check

904

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

905

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

906

"""[1:-1]

907

self.assertEqual(output, expected_output)

908

def test_verbose(self):

909

output = PrintTableCmd(verbose=True).output(self.clients)

910

expected_output = """

911

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

912

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

913

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

914

"""[1:-1]

915

self.assertEqual(output, expected_output)

916

def test_one_client(self):

917

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

918

expected_output = """

919

Name Enabled Timeout Last Successful Check

920

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

921

"""[1:-1]

922

self.assertEqual(output, expected_output)

923

924

class TestDumpJSONCmd(TestCmd):

925

def setUp(self):

926

self.expected_json = {

927

"foo": {

928

"Name": "foo",

929

"KeyID": ("92ed150794387c03ce684574b1139a65"

930

"94a34f895daaaf09fd8ea90a27cddb12"),

931

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

932

"Enabled": True,

933

"Timeout": 300000,

934

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

935

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

936

"Interval": 120000,

937

"Fingerprint": ("778827225BA7DE539C5A"

938

"7CFA59CFF7CDBD9A5920"),

939

"CheckerRunning": False,

940

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

941

"ApprovalPending": False,

942

"ApprovedByDefault": True,

943

"LastApprovalRequest": "",

944

"ApprovalDelay": 0,

945

"ApprovalDuration": 1000,

946

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

947

"ExtendedTimeout": 900000,

948

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

949

"LastCheckerStatus": 0,

950

},

951

"barbar": {

952

"Name": "barbar",

953

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

954

"6ab612cff5ad227247e46c2b020f441c"),

955

"Host": "192.0.2.3",

956

"Enabled": True,

957

"Timeout": 300000,

958

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

959

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

960

"Interval": 120000,

961

"Fingerprint": ("3E393AEAEFB84C7E89E2"

962

"F547B3A107558FCA3A27"),

963

"CheckerRunning": True,

964

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

965

"ApprovalPending": False,

966

"ApprovedByDefault": False,

967

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

968

"ApprovalDelay": 30000,

969

"ApprovalDuration": 1000,

970

"Checker": ":",

971

"ExtendedTimeout": 900000,

972

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

973

"LastCheckerStatus": -2,

974

},

975

}

976

return super(TestDumpJSONCmd, self).setUp()

977

def test_normal(self):

978

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

979

self.assertDictEqual(json_data, self.expected_json)

980

def test_one_client(self):

981

clients = self.one_client

982

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

983

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

984

self.assertDictEqual(json_data, expected_json)

985

986

class TestIsEnabledCmd(TestCmd):

987

def test_is_enabled(self):

988

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

989

for client, properties in self.clients.items()))

990

def test_is_enabled_run_exits_successfully(self):

991

with self.assertRaises(SystemExit) as e:

992

IsEnabledCmd().run(None, self.one_client)

993

if e.exception.code is not None:

994

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

995

else:

996

self.assertIsNone(e.exception.code)

997

def test_is_enabled_run_exits_with_failure(self):

998

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

999

with self.assertRaises(SystemExit) as e:

1000

IsEnabledCmd().run(None, self.one_client)

1001

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

1002

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

1003

else:

1004

self.assertIsNotNone(e.exception.code)

1005

1006

class TestRemoveCmd(TestCmd):

1007

def test_remove(self):

1008

class MockMandos(object):

1009

def __init__(self):

1010

self.calls = []

1011

def RemoveClient(self, dbus_path):

1012

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

1013

mandos = MockMandos()

1014

super(TestRemoveCmd, self).setUp()

1015

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

1016

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

1017

for client in self.clients:

1018

self.assertIn(("RemoveClient",

1019

(client.__dbus_object_path__,)),

1020

mandos.calls)

1021

1022

class TestApproveCmd(TestCmd):

1023

def test_approve(self):

1024

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

1025

for client in self.clients:

1026

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

1027

client.calls)

1028

1029

class TestDenyCmd(TestCmd):

1030

def test_deny(self):

1031

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

1032

for client in self.clients:

1033

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

1034

client.calls)

1035

1036

class TestEnableCmd(TestCmd):

1037

def test_enable(self):

1038

for client in self.clients:

1039

client.attributes["Enabled"] = False

1040

1041

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

1042

1043

for client in self.clients:

1044

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

1045

1046

class TestDisableCmd(TestCmd):

1047

def test_disable(self):

1048

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

1049

1050

for client in self.clients:

1051

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

1052

1053

class Unique(object):

1054

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

1055

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

1056

1057

class TestPropertyCmd(TestCmd):

1058

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

1059

def runTest(self):

1060

if not hasattr(self, "command"):

1061

return

1062

values_to_get = getattr(self, "values_to_get",

1063

self.values_to_set)

1064

for value_to_set, value_to_get in zip(self.values_to_set,

1065

values_to_get):

1066

for client in self.clients:

1067

old_value = client.attributes[self.property]

1068

self.assertNotIsInstance(old_value, Unique)

1069

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

1070

self.run_command(value_to_set, self.clients)

1071

for client in self.clients:

1072

value = client.attributes[self.property]

1073

self.assertNotIsInstance(value, Unique)

1074

self.assertEqual(value, value_to_get)

1075

def run_command(self, value, clients):

1076

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

1077

1078

class TestBumpTimeoutCmd(TestPropertyCmd):

1079

command = BumpTimeoutCmd

1080

property = "LastCheckedOK"

1081

values_to_set = [""]

1082

1083

class TestStartCheckerCmd(TestPropertyCmd):

1084

command = StartCheckerCmd

1085

property = "CheckerRunning"

1086

values_to_set = [dbus.Boolean(True)]

1087

1088

class TestStopCheckerCmd(TestPropertyCmd):

1089

command = StopCheckerCmd

1090

property = "CheckerRunning"

1091

values_to_set = [dbus.Boolean(False)]

1092

1093

class TestApproveByDefaultCmd(TestPropertyCmd):

1094

command = ApproveByDefaultCmd

1095

property = "ApprovedByDefault"

1096

values_to_set = [dbus.Boolean(True)]

1097

1098

class TestDenyByDefaultCmd(TestPropertyCmd):

1099

command = DenyByDefaultCmd

1100

property = "ApprovedByDefault"

1101

values_to_set = [dbus.Boolean(False)]

1102

1103

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1104

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

1105

ValueArgumentMixIn"""

1106

def runTest(self):

1107

if type(self) is TestValueArgumentPropertyCmd:

1108

return

1109

return super(TestValueArgumentPropertyCmd, self).runTest()

1110

def run_command(self, value, clients):

1111

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

1112

1113

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1114

command = SetCheckerCmd

1115

property = "Checker"

1116

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

1117

1118

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1119

command = SetHostCmd

1120

property = "Host"

1121

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

1122

1123

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1124

command = SetSecretCmd

1125

property = "Secret"

1126

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

1127

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

1128

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

1129

1130

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1131

command = SetTimeoutCmd

1132

property = "Timeout"

1133

values_to_set = [datetime.timedelta(),

1134

datetime.timedelta(minutes=5),

1135

datetime.timedelta(seconds=1),

1136

datetime.timedelta(weeks=1),

1137

datetime.timedelta(weeks=52)]

1138

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

1139

1140

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1141

command = SetExtendedTimeoutCmd

1142

property = "ExtendedTimeout"

1143

values_to_set = [datetime.timedelta(),

1144

datetime.timedelta(minutes=5),

1145

datetime.timedelta(seconds=1),

1146

datetime.timedelta(weeks=1),

1147

datetime.timedelta(weeks=52)]

1148

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

1149

1150

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1151

command = SetIntervalCmd

1152

property = "Interval"

1153

values_to_set = [datetime.timedelta(),

1154

datetime.timedelta(minutes=5),

1155

datetime.timedelta(seconds=1),

1156

datetime.timedelta(weeks=1),

1157

datetime.timedelta(weeks=52)]

1158

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

1159

1160

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1161

command = SetApprovalDelayCmd

1162

property = "ApprovalDelay"

1163

values_to_set = [datetime.timedelta(),

1164

datetime.timedelta(minutes=5),

1165

datetime.timedelta(seconds=1),

1166

datetime.timedelta(weeks=1),

1167

datetime.timedelta(weeks=52)]

1168

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

1169

1170

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1171

command = SetApprovalDurationCmd

1172

property = "ApprovalDuration"

1173

values_to_set = [datetime.timedelta(),

1174

datetime.timedelta(minutes=5),

1175

datetime.timedelta(seconds=1),

1176

datetime.timedelta(weeks=1),

1177

datetime.timedelta(weeks=52)]

1178

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

1179

1180

class Test_command_from_options(unittest.TestCase):

1181

def setUp(self):

1182

self.parser = argparse.ArgumentParser()

1183

add_command_line_options(self.parser)

1184

def assert_command_from_args(self, args, command_cls, **cmd_attrs):

1185

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

1186

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

1187

options = self.parser.parse_args(args)

1188

check_option_syntax(self.parser, options)

1189

commands = commands_from_options(options)

1190

self.assertEqual(len(commands), 1)

1191

command = commands[0]

1192

self.assertIsInstance(command, command_cls)

1193

for key, value in cmd_attrs.items():

1194

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

1195

def test_print_table(self):

1196

self.assert_command_from_args([], PrintTableCmd,

1197

verbose=False)

1198

1199

def test_print_table_verbose(self):

1200

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

1201

verbose=True)

1202

1203

def test_print_table_verbose_short(self):

1204

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

1205

verbose=True)

1206

1207

def test_enable(self):

1208

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

1209

1210

def test_enable_short(self):

1211

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

1212

1213

def test_disable(self):

1214

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

1215

DisableCmd)

1216

1217

def test_disable_short(self):

1218

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

1219

1220

def test_bump_timeout(self):

1221

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

1222

BumpTimeoutCmd)

1223

1224

def test_bump_timeout_short(self):

1225

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

1226

1227

def test_start_checker(self):

1228

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

1229

StartCheckerCmd)

1230

1231

def test_stop_checker(self):

1232

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

1233

StopCheckerCmd)

1234

1235

def test_remove(self):

1236

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

1237

RemoveCmd)

1238

1239

def test_remove_short(self):

1240

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

1241

1242

def test_checker(self):

1243

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

1244

SetCheckerCmd, value_to_set=":")

1245

1246

def test_checker_empty(self):

1247

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

1248

SetCheckerCmd, value_to_set="")

1249

1250

def test_checker_short(self):

1251

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

1252

SetCheckerCmd, value_to_set=":")

1253

1254

def test_timeout(self):

1255

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

1256

SetTimeoutCmd,

1257

value_to_set=300000)

1258

1259

def test_timeout_short(self):

1260

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

1261

SetTimeoutCmd,

1262

value_to_set=300000)

1263

1264

def test_extended_timeout(self):

1265

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

1266

"foo"],

1267

SetExtendedTimeoutCmd,

1268

value_to_set=900000)

1269

1270

def test_interval(self):

1271

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

1272

SetIntervalCmd,

1273

value_to_set=120000)

1274

1275

def test_interval_short(self):

1276

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

1277

SetIntervalCmd,

1278

value_to_set=120000)

1279

1280

def test_approve_by_default(self):

1281

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

1282

ApproveByDefaultCmd)

1283

1284

def test_deny_by_default(self):

1285

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

1286

DenyByDefaultCmd)

1287

1288

def test_approval_delay(self):

1289

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

1290

"foo"], SetApprovalDelayCmd,

1291

value_to_set=30000)

1292

1293

def test_approval_duration(self):

1294

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

1295

"foo"], SetApprovalDurationCmd,

1296

value_to_set=1000)

1297

1298

def test_host(self):

1299

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

1300

"foo"], SetHostCmd,

1301

value_to_set="foo.example.org")

1302

1303

def test_host_short(self):

1304

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

1305

"foo"], SetHostCmd,

1306

value_to_set="foo.example.org")

1307

1308

def test_secret_devnull(self):

1309

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

1310

"foo"], SetSecretCmd,

1311

value_to_set=b"")

1312

1313

def test_secret_tempfile(self):

1314

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

1315

value = b"secret\0xyzzy\nbar"

1316

f.write(value)

1317

f.seek(0)

1318

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

1319

"foo"], SetSecretCmd,

1320

value_to_set=value)

1321

1322

def test_secret_devnull_short(self):

1323

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

1324

SetSecretCmd, value_to_set=b"")

1325

1326

def test_secret_tempfile_short(self):

1327

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

1328

value = b"secret\0xyzzy\nbar"

1329

f.write(value)

1330

f.seek(0)

1331

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

1332

SetSecretCmd,

1333

value_to_set=value)

1334

1335

def test_approve(self):

1336

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

1337

ApproveCmd)

1338

1339

def test_approve_short(self):

1340

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

1341

1342

def test_deny(self):

1343

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

1344

1345

def test_deny_short(self):

1346

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

1347

1348

def test_dump_json(self):

1349

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

1350

1351

def test_is_enabled(self):

1352

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

1353

IsEnabledCmd)

1354

1355

def test_is_enabled_short(self):

1356

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

1357

1358

1359

class Test_check_option_syntax(unittest.TestCase):

1360

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

1361

# check_option_syntax()

1362

actions = {

1363

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

1364

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

1365

# never used

1366

"enable": True,

1367

"disable": True,

1368

"bump_timeout": True,

1369

"start_checker": True,

1370

"stop_checker": True,

1371

"is_enabled": True,

1372

"remove": True,

1373

"checker": "x",

1374

"timeout": datetime.timedelta(),

1375

"extended_timeout": datetime.timedelta(),

1376

"interval": datetime.timedelta(),

1377

"approved_by_default": True,

1378

"approval_delay": datetime.timedelta(),

1379

"approval_duration": datetime.timedelta(),

1380

"host": "x",

1381

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

1382

"approve": True,

1383

"deny": True,

1384

}

1385

1386

def setUp(self):

1387

self.parser = argparse.ArgumentParser()

1388

add_command_line_options(self.parser)

1389

1390

@contextlib.contextmanager

1391

def assertParseError(self):

1392

with self.assertRaises(SystemExit) as e:

1393

with self.temporarily_suppress_stderr():

1394

yield

1395

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

1396

# https://docs.python.org/3/library/argparse.html#exiting-methods

1397

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

1398

1399

@staticmethod

1400

@contextlib.contextmanager

1401

def temporarily_suppress_stderr():

1402

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

1403

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

1404

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

1405

os.close(null)

1406

try:

1407

yield

1408

finally:

1409

# restore stderr

1410

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

1411

os.close(stderrcopy)

1412

1413

def check_option_syntax(self, options):

1414

check_option_syntax(self.parser, options)

1415

1416

def test_actions_requires_client_or_all(self):

1417

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

1418

options = self.parser.parse_args()

1419

setattr(options, action, value)

1420

with self.assertParseError():

1421

self.check_option_syntax(options)

1422

1423

def test_actions_conflicts_with_verbose(self):

1424

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

1425

options = self.parser.parse_args()

1426

setattr(options, action, value)

1427

options.verbose = True

1428

with self.assertParseError():

1429

self.check_option_syntax(options)

1430

1431

def test_dump_json_conflicts_with_verbose(self):

1432

options = self.parser.parse_args()

1433

options.dump_json = True

1434

options.verbose = True

1435

with self.assertParseError():

1436

self.check_option_syntax(options)

1437

1438

def test_dump_json_conflicts_with_action(self):

1439

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

1440

options = self.parser.parse_args()

1441

setattr(options, action, value)

1442

options.dump_json = True

1443

with self.assertParseError():

1444

self.check_option_syntax(options)

1445

1446

def test_all_can_not_be_alone(self):

1447

options = self.parser.parse_args()

1448

options.all = True

1449

with self.assertParseError():

1450

self.check_option_syntax(options)

1451

1452

def test_all_is_ok_with_any_action(self):

1453

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

1454

options = self.parser.parse_args()

1455

setattr(options, action, value)

1456

options.all = True

1457

self.check_option_syntax(options)

1458

1459

def test_is_enabled_fails_without_client(self):

1460

options = self.parser.parse_args()

1461

options.is_enabled = True

1462

with self.assertParseError():

1463

self.check_option_syntax(options)

1464

1465

def test_is_enabled_works_with_one_client(self):

1466

options = self.parser.parse_args()

1467

options.is_enabled = True

1468

options.client = ["foo"]

1469

self.check_option_syntax(options)

1470

1471

def test_is_enabled_fails_with_two_clients(self):

1472

options = self.parser.parse_args()

1473

options.is_enabled = True

1474

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

1475

with self.assertParseError():

1476

self.check_option_syntax(options)

1477

1478

1479

1480

def should_only_run_tests():

1481

parser = argparse.ArgumentParser(add_help=False)

1482

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

1483

args, unknown_args = parser.parse_known_args()

1484

run_tests = args.check

1485

if run_tests:

1486

# Remove --check argument from sys.argv

1487

sys.argv[1:] = unknown_args

1488

return run_tests

1489

1490

# Add all tests from doctest strings

1491

def load_tests(loader, tests, none):

1492

import doctest

1493

tests.addTests(doctest.DocTestSuite())

1494

return tests

1495

1496

if __name__ == "__main__":

1497

if should_only_run_tests():

1498

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

1499

unittest.main()

1500

else:

1501

main()