/mandos/trunk : revision 1045

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.checker is not None:

615

commands.append(SetCheckerCmd(options.checker))

616

617

if options.timeout is not None:

618

commands.append(SetTimeoutCmd(options.timeout))

619

620

if options.extended_timeout:

621

commands.append(

622

SetExtendedTimeoutCmd(options.extended_timeout))

623

624

if options.interval is not None:

625

commands.append(SetIntervalCmd(options.interval))

626

627

if options.approved_by_default is not None:

628

if options.approved_by_default:

629

commands.append(ApproveByDefaultCmd())

630

else:

631

commands.append(DenyByDefaultCmd())

632

633

if options.approval_delay is not None:

634

commands.append(SetApprovalDelayCmd(options.approval_delay))

635

636

if options.approval_duration is not None:

637

commands.append(

638

SetApprovalDurationCmd(options.approval_duration))

639

640

if options.host is not None:

641

commands.append(SetHostCmd(options.host))

642

643

if options.secret is not None:

644

commands.append(SetSecretCmd(options.secret))

645

646

if options.approve:

647

commands.append(ApproveCmd())

648

649

if options.deny:

650

commands.append(DenyCmd())

651

652

if options.remove:

653

commands.append(RemoveCmd())

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

if options.remove:

699

options.remove = False

700

if has_actions(options) and not options.deny:

701

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

702

options.remove = True

703

704

705

def main():

706

parser = argparse.ArgumentParser()

707

708

add_command_line_options(parser)

709

710

options = parser.parse_args()

711

712

check_option_syntax(parser, options)

713

714

clientnames = options.client

715

716

if options.debug:

717

log.setLevel(logging.DEBUG)

718

719

try:

720

bus = dbus.SystemBus()

721

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

722

server_path)

723

mandos_dbus_objc = bus.get_object(busname, server_path)

724

except dbus.exceptions.DBusException:

725

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

726

sys.exit(1)

727

728

mandos_serv = dbus.Interface(mandos_dbus_objc,

729

dbus_interface=server_interface)

730

mandos_serv_object_manager = dbus.Interface(

731

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

732

733

# Filter out log message from dbus module

734

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

735

class NullFilter(logging.Filter):

736

def filter(self, record):

737

return False

738

dbus_filter = NullFilter()

739

try:

740

dbus_logger.addFilter(dbus_filter)

741

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

742

server_path, dbus.OBJECT_MANAGER_IFACE)

743

mandos_clients = {path: ifs_and_props[client_interface]

744

for path, ifs_and_props in

745

mandos_serv_object_manager

746

.GetManagedObjects().items()

747

if client_interface in ifs_and_props}

748

except dbus.exceptions.DBusException as e:

749

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

750

"\n%s", e)

751

sys.exit(1)

752

finally:

753

# restore dbus logger

754

dbus_logger.removeFilter(dbus_filter)

755

756

# Compile dict of (clients: properties) to process

757

clients = {}

758

759

if not clientnames:

760

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

761

for path, properties in mandos_clients.items()}

762

else:

763

for name in clientnames:

764

for path, client in mandos_clients.items():

765

if client["Name"] == name:

766

client_objc = bus.get_object(busname, path)

767

clients[client_objc] = client

768

break

769

else:

770

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

771

sys.exit(1)

772

773

# Run all commands on clients

774

commands = commands_from_options(options)

775

for command in commands:

776

command.run(mandos_serv, clients)

777

778

779

class Test_milliseconds_to_string(unittest.TestCase):

780

def test_all(self):

781

self.assertEqual(milliseconds_to_string(93785000),

782

"1T02:03:05")

783

def test_no_days(self):

784

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

785

def test_all_zero(self):

786

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

787

def test_no_fractional_seconds(self):

788

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

789

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

790

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

791

792

class Test_string_to_delta(unittest.TestCase):

793

def test_handles_basic_rfc3339(self):

794

self.assertEqual(string_to_delta("PT0S"),

795

datetime.timedelta())

796

self.assertEqual(string_to_delta("P0D"),

797

datetime.timedelta())

798

self.assertEqual(string_to_delta("PT1S"),

799

datetime.timedelta(0, 1))

800

self.assertEqual(string_to_delta("PT2H"),

801

datetime.timedelta(0, 7200))

802

def test_falls_back_to_pre_1_6_1_with_warning(self):

803

# assertLogs only exists in Python 3.4

804

if hasattr(self, "assertLogs"):

805

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

806

value = string_to_delta("2h")

807

else:

808

class WarningFilter(logging.Filter):

809

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

810

def filter(self, record):

811

is_warning = record.levelno >= logging.WARNING

812

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

813

False)

814

return not is_warning

815

warning_filter = WarningFilter()

816

log.addFilter(warning_filter)

817

try:

818

value = string_to_delta("2h")

819

finally:

820

log.removeFilter(warning_filter)

821

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

822

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

823

824

825

class TestCmd(unittest.TestCase):

826

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

827

def setUp(self):

828

testcase = self

829

class MockClient(object):

830

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

831

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

832

self.attributes = attributes

833

self.attributes["Name"] = name

834

self.calls = []

835

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

836

testcase.assertEqual(interface, client_interface)

837

testcase.assertEqual(dbus_interface,

838

dbus.PROPERTIES_IFACE)

839

self.attributes[property] = value

840

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

841

testcase.assertEqual(interface, client_interface)

842

testcase.assertEqual(dbus_interface,

843

dbus.PROPERTIES_IFACE)

844

return self.attributes[property]

845

def Approve(self, approve, dbus_interface):

846

testcase.assertEqual(dbus_interface, client_interface)

847

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

848

dbus_interface)))

849

self.client = MockClient(

850

"foo",

851

KeyID=("92ed150794387c03ce684574b1139a65"

852

"94a34f895daaaf09fd8ea90a27cddb12"),

853

Secret=b"secret",

854

Host="foo.example.org",

855

Enabled=dbus.Boolean(True),

856

Timeout=300000,

857

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

858

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

859

Interval=120000,

860

Fingerprint=("778827225BA7DE539C5A"

861

"7CFA59CFF7CDBD9A5920"),

862

CheckerRunning=dbus.Boolean(False),

863

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

864

ApprovalPending=dbus.Boolean(False),

865

ApprovedByDefault=dbus.Boolean(True),

866

LastApprovalRequest="",

867

ApprovalDelay=0,

868

ApprovalDuration=1000,

869

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

870

ExtendedTimeout=900000,

871

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

872

LastCheckerStatus=0)

873

self.other_client = MockClient(

874

"barbar",

875

KeyID=("0558568eedd67d622f5c83b35a115f79"

876

"6ab612cff5ad227247e46c2b020f441c"),

877

Secret=b"secretbar",

878

Host="192.0.2.3",

879

Enabled=dbus.Boolean(True),

880

Timeout=300000,

881

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

882

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

883

Interval=120000,

884

Fingerprint=("3E393AEAEFB84C7E89E2"

885

"F547B3A107558FCA3A27"),

886

CheckerRunning=dbus.Boolean(True),

887

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

888

ApprovalPending=dbus.Boolean(False),

889

ApprovedByDefault=dbus.Boolean(False),

890

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

891

ApprovalDelay=30000,

892

ApprovalDuration=1000,

893

Checker=":",

894

ExtendedTimeout=900000,

895

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

896

LastCheckerStatus=-2)

897

self.clients = collections.OrderedDict(

898

[

899

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

900

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

901

])

902

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

903

904

class TestPrintTableCmd(TestCmd):

905

def test_normal(self):

906

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

907

expected_output = """

908

Name Enabled Timeout Last Successful Check

909

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

910

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

911

"""[1:-1]

912

self.assertEqual(output, expected_output)

913

def test_verbose(self):

914

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

915

expected_output = """

916

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

917

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

918

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

919

"""[1:-1]

920

self.assertEqual(output, expected_output)

921

def test_one_client(self):

922

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

923

expected_output = """

924

Name Enabled Timeout Last Successful Check

925

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

926

"""[1:-1]

927

self.assertEqual(output, expected_output)

928

929

class TestDumpJSONCmd(TestCmd):

930

def setUp(self):

931

self.expected_json = {

932

"foo": {

933

"Name": "foo",

934

"KeyID": ("92ed150794387c03ce684574b1139a65"

935

"94a34f895daaaf09fd8ea90a27cddb12"),

936

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

937

"Enabled": True,

938

"Timeout": 300000,

939

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

940

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

941

"Interval": 120000,

942

"Fingerprint": ("778827225BA7DE539C5A"

943

"7CFA59CFF7CDBD9A5920"),

944

"CheckerRunning": False,

945

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

946

"ApprovalPending": False,

947

"ApprovedByDefault": True,

948

"LastApprovalRequest": "",

949

"ApprovalDelay": 0,

950

"ApprovalDuration": 1000,

951

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

952

"ExtendedTimeout": 900000,

953

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

954

"LastCheckerStatus": 0,

955

},

956

"barbar": {

957

"Name": "barbar",

958

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

959

"6ab612cff5ad227247e46c2b020f441c"),

960

"Host": "192.0.2.3",

961

"Enabled": True,

962

"Timeout": 300000,

963

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

964

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

965

"Interval": 120000,

966

"Fingerprint": ("3E393AEAEFB84C7E89E2"

967

"F547B3A107558FCA3A27"),

968

"CheckerRunning": True,

969

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

970

"ApprovalPending": False,

971

"ApprovedByDefault": False,

972

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

973

"ApprovalDelay": 30000,

974

"ApprovalDuration": 1000,

975

"Checker": ":",

976

"ExtendedTimeout": 900000,

977

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

978

"LastCheckerStatus": -2,

979

},

980

}

981

return super(TestDumpJSONCmd, self).setUp()

982

def test_normal(self):

983

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

984

self.assertDictEqual(json_data, self.expected_json)

985

def test_one_client(self):

986

clients = self.one_client

987

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

988

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

989

self.assertDictEqual(json_data, expected_json)

990

991

class TestIsEnabledCmd(TestCmd):

992

def test_is_enabled(self):

993

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

994

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

995

def test_is_enabled_run_exits_successfully(self):

996

with self.assertRaises(SystemExit) as e:

997

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

998

if e.exception.code is not None:

999

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

1000

else:

1001

self.assertIsNone(e.exception.code)

1002

def test_is_enabled_run_exits_with_failure(self):

1003

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

1004

with self.assertRaises(SystemExit) as e:

1005

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

1006

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

1007

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

1008

else:

1009

self.assertIsNotNone(e.exception.code)

1010

1011

class TestRemoveCmd(TestCmd):

1012

def test_remove(self):

1013

class MockMandos(object):

1014

def __init__(self):

1015

self.calls = []

1016

def RemoveClient(self, dbus_path):

1017

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

1018

mandos = MockMandos()

1019

super(TestRemoveCmd, self).setUp()

1020

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

1021

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

1022

for client in self.clients:

1023

self.assertIn(("RemoveClient",

1024

(client.__dbus_object_path__,)),

1025

mandos.calls)

1026

1027

class TestApproveCmd(TestCmd):

1028

def test_approve(self):

1029

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

1030

for client in self.clients:

1031

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

1032

client.calls)

1033

1034

class TestDenyCmd(TestCmd):

1035

def test_deny(self):

1036

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

1037

for client in self.clients:

1038

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

1039

client.calls)

1040

1041

class TestEnableCmd(TestCmd):

1042

def test_enable(self):

1043

for client in self.clients:

1044

client.attributes["Enabled"] = False

1045

1046

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

1047

1048

for client in self.clients:

1049

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

1050

1051

class TestDisableCmd(TestCmd):

1052

def test_disable(self):

1053

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

1054

1055

for client in self.clients:

1056

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

1057

1058

class Unique(object):

1059

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

1060

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

1061

1062

class TestPropertyCmd(TestCmd):

1063

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

1064

def runTest(self):

1065

if not hasattr(self, "command"):

1066

return

1067

values_to_get = getattr(self, "values_to_get",

1068

self.values_to_set)

1069

for value_to_set, value_to_get in zip(self.values_to_set,

1070

values_to_get):

1071

for client in self.clients:

1072

old_value = client.attributes[self.property]

1073

self.assertNotIsInstance(old_value, Unique)

1074

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

1075

self.run_command(value_to_set, self.clients)

1076

for client in self.clients:

1077

value = client.attributes[self.property]

1078

self.assertNotIsInstance(value, Unique)

1079

self.assertEqual(value, value_to_get)

1080

def run_command(self, value, clients):

1081

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

1082

1083

class TestBumpTimeoutCmd(TestPropertyCmd):

1084

command = BumpTimeoutCmd

1085

property = "LastCheckedOK"

1086

values_to_set = [""]

1087

1088

class TestStartCheckerCmd(TestPropertyCmd):

1089

command = StartCheckerCmd

1090

property = "CheckerRunning"

1091

values_to_set = [dbus.Boolean(True)]

1092

1093

class TestStopCheckerCmd(TestPropertyCmd):

1094

command = StopCheckerCmd

1095

property = "CheckerRunning"

1096

values_to_set = [dbus.Boolean(False)]

1097

1098

class TestApproveByDefaultCmd(TestPropertyCmd):

1099

command = ApproveByDefaultCmd

1100

property = "ApprovedByDefault"

1101

values_to_set = [dbus.Boolean(True)]

1102

1103

class TestDenyByDefaultCmd(TestPropertyCmd):

1104

command = DenyByDefaultCmd

1105

property = "ApprovedByDefault"

1106

values_to_set = [dbus.Boolean(False)]

1107

1108

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1109

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

1110

ValueArgumentMixIn"""

1111

def runTest(self):

1112

if type(self) is TestValueArgumentPropertyCmd:

1113

return

1114

return super(TestValueArgumentPropertyCmd, self).runTest()

1115

def run_command(self, value, clients):

1116

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

1117

1118

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1119

command = SetCheckerCmd

1120

property = "Checker"

1121

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

1122

1123

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1124

command = SetHostCmd

1125

property = "Host"

1126

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

1127

1128

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1129

command = SetSecretCmd

1130

property = "Secret"

1131

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

1132

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

1133

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

1134

1135

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1136

command = SetTimeoutCmd

1137

property = "Timeout"

1138

values_to_set = [datetime.timedelta(),

1139

datetime.timedelta(minutes=5),

1140

datetime.timedelta(seconds=1),

1141

datetime.timedelta(weeks=1),

1142

datetime.timedelta(weeks=52)]

1143

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

1144

1145

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1146

command = SetExtendedTimeoutCmd

1147

property = "ExtendedTimeout"

1148

values_to_set = [datetime.timedelta(),

1149

datetime.timedelta(minutes=5),

1150

datetime.timedelta(seconds=1),

1151

datetime.timedelta(weeks=1),

1152

datetime.timedelta(weeks=52)]

1153

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

1154

1155

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1156

command = SetIntervalCmd

1157

property = "Interval"

1158

values_to_set = [datetime.timedelta(),

1159

datetime.timedelta(minutes=5),

1160

datetime.timedelta(seconds=1),

1161

datetime.timedelta(weeks=1),

1162

datetime.timedelta(weeks=52)]

1163

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

1164

1165

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1166

command = SetApprovalDelayCmd

1167

property = "ApprovalDelay"

1168

values_to_set = [datetime.timedelta(),

1169

datetime.timedelta(minutes=5),

1170

datetime.timedelta(seconds=1),

1171

datetime.timedelta(weeks=1),

1172

datetime.timedelta(weeks=52)]

1173

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

1174

1175

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1176

command = SetApprovalDurationCmd

1177

property = "ApprovalDuration"

1178

values_to_set = [datetime.timedelta(),

1179

datetime.timedelta(minutes=5),

1180

datetime.timedelta(seconds=1),

1181

datetime.timedelta(weeks=1),

1182

datetime.timedelta(weeks=52)]

1183

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

1184

1185

class Test_command_from_options(unittest.TestCase):

1186

def setUp(self):

1187

self.parser = argparse.ArgumentParser()

1188

add_command_line_options(self.parser)

1189

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

1190

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

1191

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

1192

options = self.parser.parse_args(args)

1193

check_option_syntax(self.parser, options)

1194

commands = commands_from_options(options)

1195

self.assertEqual(len(commands), 1)

1196

command = commands[0]

1197

self.assertIsInstance(command, command_cls)

1198

for key, value in cmd_attrs.items():

1199

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

1200

def test_print_table(self):

1201

self.assert_command_from_args([], PrintTableCmd,

1202

verbose=False)

1203

1204

def test_print_table_verbose(self):

1205

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

1206

verbose=True)

1207

1208

def test_print_table_verbose_short(self):

1209

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

1210

verbose=True)

1211

1212

def test_enable(self):

1213

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

1214

1215

def test_enable_short(self):

1216

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

1217

1218

def test_disable(self):

1219

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

1220

DisableCmd)

1221

1222

def test_disable_short(self):

1223

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

1224

1225

def test_bump_timeout(self):

1226

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

1227

BumpTimeoutCmd)

1228

1229

def test_bump_timeout_short(self):

1230

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

1231

1232

def test_start_checker(self):

1233

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

1234

StartCheckerCmd)

1235

1236

def test_stop_checker(self):

1237

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

1238

StopCheckerCmd)

1239

1240

def test_remove(self):

1241

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

1242

RemoveCmd)

1243

1244

def test_remove_short(self):

1245

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

1246

1247

def test_checker(self):

1248

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

1249

SetCheckerCmd, value_to_set=":")

1250

1251

def test_checker_empty(self):

1252

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

1253

SetCheckerCmd, value_to_set="")

1254

1255

def test_checker_short(self):

1256

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

1257

SetCheckerCmd, value_to_set=":")

1258

1259

def test_timeout(self):

1260

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

1261

SetTimeoutCmd,

1262

value_to_set=300000)

1263

1264

def test_timeout_short(self):

1265

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

1266

SetTimeoutCmd,

1267

value_to_set=300000)

1268

1269

def test_extended_timeout(self):

1270

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

1271

"foo"],

1272

SetExtendedTimeoutCmd,

1273

value_to_set=900000)

1274

1275

def test_interval(self):

1276

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

1277

SetIntervalCmd,

1278

value_to_set=120000)

1279

1280

def test_interval_short(self):

1281

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

1282

SetIntervalCmd,

1283

value_to_set=120000)

1284

1285

def test_approve_by_default(self):

1286

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

1287

ApproveByDefaultCmd)

1288

1289

def test_deny_by_default(self):

1290

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

1291

DenyByDefaultCmd)

1292

1293

def test_approval_delay(self):

1294

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

1295

"foo"], SetApprovalDelayCmd,

1296

value_to_set=30000)

1297

1298

def test_approval_duration(self):

1299

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

1300

"foo"], SetApprovalDurationCmd,

1301

value_to_set=1000)

1302

1303

def test_host(self):

1304

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

1305

"foo"], SetHostCmd,

1306

value_to_set="foo.example.org")

1307

1308

def test_host_short(self):

1309

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

1310

"foo"], SetHostCmd,

1311

value_to_set="foo.example.org")

1312

1313

def test_secret_devnull(self):

1314

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

1315

"foo"], SetSecretCmd,

1316

value_to_set=b"")

1317

1318

def test_secret_tempfile(self):

1319

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

1320

value = b"secret\0xyzzy\nbar"

1321

f.write(value)

1322

f.seek(0)

1323

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

1324

"foo"], SetSecretCmd,

1325

value_to_set=value)

1326

1327

def test_secret_devnull_short(self):

1328

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

1329

SetSecretCmd, value_to_set=b"")

1330

1331

def test_secret_tempfile_short(self):

1332

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

1333

value = b"secret\0xyzzy\nbar"

1334

f.write(value)

1335

f.seek(0)

1336

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

1337

SetSecretCmd,

1338

value_to_set=value)

1339

1340

def test_approve(self):

1341

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

1342

ApproveCmd)

1343

1344

def test_approve_short(self):

1345

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

1346

1347

def test_deny(self):

1348

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

1349

1350

def test_deny_short(self):

1351

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

1352

1353

def test_dump_json(self):

1354

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

1355

1356

def test_is_enabled(self):

1357

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

1358

IsEnabledCmd)

1359

1360

def test_is_enabled_short(self):

1361

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

1362

1363

def test_deny_before_remove(self):

1364

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

1365

check_option_syntax(self.parser, options)

1366

commands = commands_from_options(options)

1367

self.assertEqual(len(commands), 2)

1368

self.assertIsInstance(commands[0], DenyCmd)

1369

self.assertIsInstance(commands[1], RemoveCmd)

1370

1371

def test_deny_before_remove_reversed(self):

1372

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

1373

check_option_syntax(self.parser, options)

1374

commands = commands_from_options(options)

1375

self.assertEqual(len(commands), 2)

1376

self.assertIsInstance(commands[0], DenyCmd)

1377

self.assertIsInstance(commands[1], RemoveCmd)

1378

1379

1380

class Test_check_option_syntax(unittest.TestCase):

1381

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

1382

# check_option_syntax()

1383

actions = {

1384

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

1385

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

1386

# never used

1387

"enable": True,

1388

"disable": True,

1389

"bump_timeout": True,

1390

"start_checker": True,

1391

"stop_checker": True,

1392

"is_enabled": True,

1393

"remove": True,

1394

"checker": "x",

1395

"timeout": datetime.timedelta(),

1396

"extended_timeout": datetime.timedelta(),

1397

"interval": datetime.timedelta(),

1398

"approved_by_default": True,

1399

"approval_delay": datetime.timedelta(),

1400

"approval_duration": datetime.timedelta(),

1401

"host": "x",

1402

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

1403

"approve": True,

1404

"deny": True,

1405

}

1406

1407

def setUp(self):

1408

self.parser = argparse.ArgumentParser()

1409

add_command_line_options(self.parser)

1410

1411

@contextlib.contextmanager

1412

def assertParseError(self):

1413

with self.assertRaises(SystemExit) as e:

1414

with self.temporarily_suppress_stderr():

1415

yield

1416

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

1417

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

1418

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

1419

1420

@staticmethod

1421

@contextlib.contextmanager

1422

def temporarily_suppress_stderr():

1423

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

1424

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

1425

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

1426

os.close(null)

1427

try:

1428

yield

1429

finally:

1430

# restore stderr

1431

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

1432

os.close(stderrcopy)

1433

1434

def check_option_syntax(self, options):

1435

check_option_syntax(self.parser, options)

1436

1437

def test_actions_requires_client_or_all(self):

1438

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

1439

options = self.parser.parse_args()

1440

setattr(options, action, value)

1441

with self.assertParseError():

1442

self.check_option_syntax(options)

1443

1444

def test_actions_conflicts_with_verbose(self):

1445

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

1446

options = self.parser.parse_args()

1447

setattr(options, action, value)

1448

options.verbose = True

1449

with self.assertParseError():

1450

self.check_option_syntax(options)

1451

1452

def test_dump_json_conflicts_with_verbose(self):

1453

options = self.parser.parse_args()

1454

options.dump_json = True

1455

options.verbose = True

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_dump_json_conflicts_with_action(self):

1460

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

1461

options = self.parser.parse_args()

1462

setattr(options, action, value)

1463

options.dump_json = True

1464

with self.assertParseError():

1465

self.check_option_syntax(options)

1466

1467

def test_all_can_not_be_alone(self):

1468

options = self.parser.parse_args()

1469

options.all = True

1470

with self.assertParseError():

1471

self.check_option_syntax(options)

1472

1473

def test_all_is_ok_with_any_action(self):

1474

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

1475

options = self.parser.parse_args()

1476

setattr(options, action, value)

1477

options.all = True

1478

self.check_option_syntax(options)

1479

1480

def test_is_enabled_fails_without_client(self):

1481

options = self.parser.parse_args()

1482

options.is_enabled = True

1483

with self.assertParseError():

1484

self.check_option_syntax(options)

1485

1486

def test_is_enabled_works_with_one_client(self):

1487

options = self.parser.parse_args()

1488

options.is_enabled = True

1489

options.client = ["foo"]

1490

self.check_option_syntax(options)

1491

1492

def test_is_enabled_fails_with_two_clients(self):

1493

options = self.parser.parse_args()

1494

options.is_enabled = True

1495

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

1496

with self.assertParseError():

1497

self.check_option_syntax(options)

1498

1499

def test_remove_can_only_be_combined_with_action_deny(self):

1500

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

1501

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

1502

continue

1503

options = self.parser.parse_args()

1504

setattr(options, action, value)

1505

options.all = True

1506

options.remove = True

1507

with self.assertParseError():

1508

self.check_option_syntax(options)

1509

1510

1511

1512

def should_only_run_tests():

1513

parser = argparse.ArgumentParser(add_help=False)

1514

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

1515

args, unknown_args = parser.parse_known_args()

1516

run_tests = args.check

1517

if run_tests:

1518

# Remove --check argument from sys.argv

1519

sys.argv[1:] = unknown_args

1520

return run_tests

1521

1522

# Add all tests from doctest strings

1523

def load_tests(loader, tests, none):

1524

import doctest

1525

tests.addTests(doctest.DocTestSuite())

1526

return tests

1527

1528

if __name__ == "__main__":

1529

if should_only_run_tests():

1530

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

1531

unittest.main()

1532

else:

1533

main()