/mandos/trunk : revision 79

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, clients, bus=None, mandos=None):

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 clientpath, properties in clients.items():

285

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

286

busname, str(clientpath))

287

client = bus.get_object(busname, clientpath)

288

self.run_on_one_client(client, properties)

289

290

class PrintCmd(Command):

291

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

292

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

293

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

294

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

295

"ApprovalPending", "ApprovedByDefault",

296

"LastApprovalRequest", "ApprovalDelay",

297

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

298

"Expires", "LastCheckerStatus")

299

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

300

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

301

def output(self, clients):

302

raise NotImplementedError()

303

304

class PropertyCmd(Command):

305

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

306

def run_on_one_client(self, client, properties):

307

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

308

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

309

client.__dbus_object_path__,

310

dbus.PROPERTIES_IFACE, client_interface,

311

self.propname, self.value_to_set

312

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

313

else bool(self.value_to_set))

314

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

315

dbus_interface=dbus.PROPERTIES_IFACE)

316

@property

317

def propname(self):

318

raise NotImplementedError()

319

320

class ValueArgumentMixIn(object):

321

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

322

def __init__(self, value):

323

self.value_to_set = value

324

325

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

326

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

327

milliseconds."""

328

@property

329

def value_to_set(self):

330

return self._vts

331

@value_to_set.setter

332

def value_to_set(self, value):

333

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

334

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

335

336

# Actual (non-abstract) command classes

337

338

class PrintTableCmd(PrintCmd):

339

def __init__(self, verbose=False):

340

self.verbose = verbose

341

342

def output(self, clients):

343

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

344

keywords = default_keywords

345

if self.verbose:

346

keywords = self.all_keywords

347

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

348

349

class TableOfClients(object):

350

tableheaders = {

351

"Name": "Name",

352

"Enabled": "Enabled",

353

"Timeout": "Timeout",

354

"LastCheckedOK": "Last Successful Check",

355

"LastApprovalRequest": "Last Approval Request",

356

"Created": "Created",

357

"Interval": "Interval",

358

"Host": "Host",

359

"Fingerprint": "Fingerprint",

360

"KeyID": "Key ID",

361

"CheckerRunning": "Check Is Running",

362

"LastEnabled": "Last Enabled",

363

"ApprovalPending": "Approval Is Pending",

364

"ApprovedByDefault": "Approved By Default",

365

"ApprovalDelay": "Approval Delay",

366

"ApprovalDuration": "Approval Duration",

367

"Checker": "Checker",

368

"ExtendedTimeout": "Extended Timeout",

369

"Expires": "Expires",

370

"LastCheckerStatus": "Last Checker Status",

371

}

372

373

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

374

self.clients = clients

375

self.keywords = keywords

376

if tableheaders is not None:

377

self.tableheaders = tableheaders

378

379

def __str__(self):

380

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

381

382

if sys.version_info.major == 2:

383

__unicode__ = __str__

384

def __str__(self):

385

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

386

387

def rows(self):

388

format_string = self.row_formatting_string()

389

rows = [self.header_line(format_string)]

390

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

391

for client in self.clients)

392

return rows

393

394

def row_formatting_string(self):

395

"Format string used to format table rows"

396

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

397

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

398

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

399

for client in self.clients)),

400

key=key)

401

for key in self.keywords)

402

403

def string_from_client(self, client, key):

404

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

405

406

@staticmethod

407

def valuetostring(value, keyword):

408

if isinstance(value, dbus.Boolean):

409

return "Yes" if value else "No"

410

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

411

"ApprovalDuration", "ExtendedTimeout"):

412

return milliseconds_to_string(value)

413

return str(value)

414

415

def header_line(self, format_string):

416

return format_string.format(**self.tableheaders)

417

418

def client_line(self, client, format_string):

419

return format_string.format(

420

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

421

for key in self.keywords})

422

423

424

425

class DumpJSONCmd(PrintCmd):

426

def output(self, clients):

427

data = {client["Name"]:

428

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

429

for key in self.all_keywords}

430

for client in clients.values()}

431

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

432

@staticmethod

433

def dbus_boolean_to_bool(value):

434

if isinstance(value, dbus.Boolean):

435

value = bool(value)

436

return value

437

438

class IsEnabledCmd(Command):

439

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

440

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

441

if self.is_enabled(client, properties):

442

sys.exit(0)

443

sys.exit(1)

444

def is_enabled(self, client, properties):

445

return properties["Enabled"]

446

447

class RemoveCmd(Command):

448

def run_on_one_client(self, client, properties):

449

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

450

server_path, server_interface,

451

str(client.__dbus_object_path__))

452

self.mandos.RemoveClient(client.__dbus_object_path__)

453

454

class ApproveCmd(Command):

455

def run_on_one_client(self, client, properties):

456

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

457

client.__dbus_object_path__, client_interface)

458

client.Approve(dbus.Boolean(True),

459

dbus_interface=client_interface)

460

461

class DenyCmd(Command):

462

def run_on_one_client(self, client, properties):

463

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

464

client.__dbus_object_path__, client_interface)

465

client.Approve(dbus.Boolean(False),

466

dbus_interface=client_interface)

467

468

class EnableCmd(PropertyCmd):

469

propname = "Enabled"

470

value_to_set = dbus.Boolean(True)

471

472

class DisableCmd(PropertyCmd):

473

propname = "Enabled"

474

value_to_set = dbus.Boolean(False)

475

476

class BumpTimeoutCmd(PropertyCmd):

477

propname = "LastCheckedOK"

478

value_to_set = ""

479

480

class StartCheckerCmd(PropertyCmd):

481

propname = "CheckerRunning"

482

value_to_set = dbus.Boolean(True)

483

484

class StopCheckerCmd(PropertyCmd):

485

propname = "CheckerRunning"

486

value_to_set = dbus.Boolean(False)

487

488

class ApproveByDefaultCmd(PropertyCmd):

489

propname = "ApprovedByDefault"

490

value_to_set = dbus.Boolean(True)

491

492

class DenyByDefaultCmd(PropertyCmd):

493

propname = "ApprovedByDefault"

494

value_to_set = dbus.Boolean(False)

495

496

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

497

propname = "Checker"

498

499

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

500

propname = "Host"

501

502

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

503

propname = "Secret"

504

@property

505

def value_to_set(self):

506

return self._vts

507

@value_to_set.setter

508

def value_to_set(self, value):

509

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

510

self._vts = value.read()

511

value.close()

512

513

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

514

propname = "Timeout"

515

516

class SetExtendedTimeoutCmd(PropertyCmd,

517

MillisecondsValueArgumentMixIn):

518

propname = "ExtendedTimeout"

519

520

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

521

propname = "Interval"

522

523

class SetApprovalDelayCmd(PropertyCmd,

524

MillisecondsValueArgumentMixIn):

525

propname = "ApprovalDelay"

526

527

class SetApprovalDurationCmd(PropertyCmd,

528

MillisecondsValueArgumentMixIn):

529

propname = "ApprovalDuration"

530

531

def add_command_line_options(parser):

532

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

533

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

534

help="show version number and exit")

535

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

536

help="Select all clients")

537

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

538

help="Print all fields")

539

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

540

help="Dump client data in JSON format")

541

enable_disable = parser.add_mutually_exclusive_group()

542

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

543

help="Enable client")

544

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

545

action="store_true",

546

help="disable client")

547

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

548

help="Bump timeout for client")

549

start_stop_checker = parser.add_mutually_exclusive_group()

550

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

551

action="store_true",

552

help="Start checker for client")

553

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

554

action="store_true",

555

help="Stop checker for client")

556

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

557

help="Check if client is enabled")

558

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

559

help="Remove client")

560

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

561

help="Set checker command for client")

562

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

563

help="Set timeout for client")

564

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

565

help="Set extended timeout for client")

566

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

567

help="Set checker interval for client")

568

approve_deny_default = parser.add_mutually_exclusive_group()

569

approve_deny_default.add_argument(

570

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

571

default=None, dest="approved_by_default",

572

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

573

approve_deny_default.add_argument(

574

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

575

dest="approved_by_default",

576

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

577

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

578

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

579

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

580

help="Set duration of one client approval")

581

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

582

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

583

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

584

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

585

approve_deny = parser.add_mutually_exclusive_group()

586

approve_deny.add_argument(

587

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

588

help="Approve any current client request")

589

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

590

help="Deny any current client request")

591

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

592

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

593

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

594

help="Run self-test")

595

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

596

597

598

def commands_from_options(options):

599

600

commands = []

601

602

if options.dump_json:

603

commands.append(DumpJSONCmd())

604

605

if options.enable:

606

commands.append(EnableCmd())

607

608

if options.disable:

609

commands.append(DisableCmd())

610

611

if options.bump_timeout:

612

commands.append(BumpTimeoutCmd())

613

614

if options.start_checker:

615

commands.append(StartCheckerCmd())

616

617

if options.stop_checker:

618

commands.append(StopCheckerCmd())

619

620

if options.is_enabled:

621

commands.append(IsEnabledCmd())

622

623

if options.checker is not None:

624

commands.append(SetCheckerCmd(options.checker))

625

626

if options.timeout is not None:

627

commands.append(SetTimeoutCmd(options.timeout))

628

629

if options.extended_timeout:

630

commands.append(

631

SetExtendedTimeoutCmd(options.extended_timeout))

632

633

if options.interval is not None:

634

commands.append(SetIntervalCmd(options.interval))

635

636

if options.approved_by_default is not None:

637

if options.approved_by_default:

638

commands.append(ApproveByDefaultCmd())

639

else:

640

commands.append(DenyByDefaultCmd())

641

642

if options.approval_delay is not None:

643

commands.append(SetApprovalDelayCmd(options.approval_delay))

644

645

if options.approval_duration is not None:

646

commands.append(

647

SetApprovalDurationCmd(options.approval_duration))

648

649

if options.host is not None:

650

commands.append(SetHostCmd(options.host))

651

652

if options.secret is not None:

653

commands.append(SetSecretCmd(options.secret))

654

655

if options.approve:

656

commands.append(ApproveCmd())

657

658

if options.deny:

659

commands.append(DenyCmd())

660

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

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

665

# optionally verbosely

666

if not commands:

667

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

668

669

return commands

670

671

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

676

def has_actions(options):

677

return any((options.enable,

678

options.disable,

679

options.bump_timeout,

680

options.start_checker,

681

options.stop_checker,

682

options.is_enabled,

683

options.remove,

684

options.checker is not None,

685

options.timeout is not None,

686

options.extended_timeout is not None,

687

options.interval is not None,

688

options.approved_by_default is not None,

689

options.approval_delay is not None,

690

options.approval_duration is not None,

691

options.host is not None,

692

options.secret is not None,

693

options.approve,

694

options.deny))

695

696

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

697

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

698

if options.verbose and has_actions(options):

699

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

700

if options.dump_json and (options.verbose

701

or has_actions(options)):

702

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

703

if options.all and not has_actions(options):

704

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

705

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

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

723

clientnames = options.client

724

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

try:

729

bus = dbus.SystemBus()

730

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

731

server_path)

732

mandos_dbus_objc = bus.get_object(busname, server_path)

733

except dbus.exceptions.DBusException:

734

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

735

sys.exit(1)

736

737

mandos_serv = dbus.Interface(mandos_dbus_objc,

738

dbus_interface=server_interface)

739

mandos_serv_object_manager = dbus.Interface(

740

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

741

742

# Filter out log message from dbus module

743

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

744

class NullFilter(logging.Filter):

745

def filter(self, record):

746

return False

747

dbus_filter = NullFilter()

748

try:

749

dbus_logger.addFilter(dbus_filter)

750

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

751

server_path, dbus.OBJECT_MANAGER_IFACE)

752

mandos_clients = {path: ifs_and_props[client_interface]

753

for path, ifs_and_props in

754

mandos_serv_object_manager

755

.GetManagedObjects().items()

756

if client_interface in ifs_and_props}

757

except dbus.exceptions.DBusException as e:

758

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

759

"\n%s", e)

760

sys.exit(1)

761

finally:

762

# restore dbus logger

763

dbus_logger.removeFilter(dbus_filter)

764

765

# Compile dict of (clients: properties) to process

766

clients = {}

767

768

if not clientnames:

769

clients = {objpath: properties

770

for objpath, properties in mandos_clients.items()}

771

else:

772

for name in clientnames:

773

for objpath, properties in mandos_clients.items():

774

if properties["Name"] == name:

775

clients[objpath] = properties

776

break

777

else:

778

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

779

sys.exit(1)

780

781

# Run all commands on clients

782

commands = commands_from_options(options)

783

for command in commands:

784

command.run(clients, bus, mandos_serv)

785

786

787

class Test_milliseconds_to_string(unittest.TestCase):

788

def test_all(self):

789

self.assertEqual(milliseconds_to_string(93785000),

790

"1T02:03:05")

791

def test_no_days(self):

792

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

793

def test_all_zero(self):

794

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

795

def test_no_fractional_seconds(self):

796

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

797

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

798

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

799

800

class Test_string_to_delta(unittest.TestCase):

801

def test_handles_basic_rfc3339(self):

802

self.assertEqual(string_to_delta("PT0S"),

803

datetime.timedelta())

804

self.assertEqual(string_to_delta("P0D"),

805

datetime.timedelta())

806

self.assertEqual(string_to_delta("PT1S"),

807

datetime.timedelta(0, 1))

808

self.assertEqual(string_to_delta("PT2H"),

809

datetime.timedelta(0, 7200))

810

def test_falls_back_to_pre_1_6_1_with_warning(self):

811

# assertLogs only exists in Python 3.4

812

if hasattr(self, "assertLogs"):

813

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

814

value = string_to_delta("2h")

815

else:

816

class WarningFilter(logging.Filter):

817

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

818

def filter(self, record):

819

is_warning = record.levelno >= logging.WARNING

820

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

821

False)

822

return not is_warning

823

warning_filter = WarningFilter()

824

log.addFilter(warning_filter)

825

try:

826

value = string_to_delta("2h")

827

finally:

828

log.removeFilter(warning_filter)

829

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

830

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

831

832

833

class TestCmd(unittest.TestCase):

834

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

835

def setUp(self):

836

testcase = self

837

class MockClient(object):

838

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

839

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

840

self.attributes = attributes

841

self.attributes["Name"] = name

842

self.calls = []

843

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

844

testcase.assertEqual(interface, client_interface)

845

testcase.assertEqual(dbus_interface,

846

dbus.PROPERTIES_IFACE)

847

self.attributes[propname] = value

848

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

849

testcase.assertEqual(interface, client_interface)

850

testcase.assertEqual(dbus_interface,

851

dbus.PROPERTIES_IFACE)

852

return self.attributes[propname]

853

def Approve(self, approve, dbus_interface):

854

testcase.assertEqual(dbus_interface, client_interface)

855

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

856

dbus_interface)))

857

self.client = MockClient(

858

"foo",

859

KeyID=("92ed150794387c03ce684574b1139a65"

860

"94a34f895daaaf09fd8ea90a27cddb12"),

861

Secret=b"secret",

862

Host="foo.example.org",

863

Enabled=dbus.Boolean(True),

864

Timeout=300000,

865

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

866

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

867

Interval=120000,

868

Fingerprint=("778827225BA7DE539C5A"

869

"7CFA59CFF7CDBD9A5920"),

870

CheckerRunning=dbus.Boolean(False),

871

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

872

ApprovalPending=dbus.Boolean(False),

873

ApprovedByDefault=dbus.Boolean(True),

874

LastApprovalRequest="",

875

ApprovalDelay=0,

876

ApprovalDuration=1000,

877

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

878

ExtendedTimeout=900000,

879

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

880

LastCheckerStatus=0)

881

self.other_client = MockClient(

882

"barbar",

883

KeyID=("0558568eedd67d622f5c83b35a115f79"

884

"6ab612cff5ad227247e46c2b020f441c"),

885

Secret=b"secretbar",

886

Host="192.0.2.3",

887

Enabled=dbus.Boolean(True),

888

Timeout=300000,

889

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

890

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

891

Interval=120000,

892

Fingerprint=("3E393AEAEFB84C7E89E2"

893

"F547B3A107558FCA3A27"),

894

CheckerRunning=dbus.Boolean(True),

895

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

896

ApprovalPending=dbus.Boolean(False),

897

ApprovedByDefault=dbus.Boolean(False),

898

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

899

ApprovalDelay=30000,

900

ApprovalDuration=1000,

901

Checker=":",

902

ExtendedTimeout=900000,

903

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

904

LastCheckerStatus=-2)

905

self.clients = collections.OrderedDict(

906

[

907

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

908

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

909

])

910

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

911

@property

912

def bus(self):

913

class Bus(object):

914

@staticmethod

915

def get_object(client_bus_name, path):

916

self.assertEqual(client_bus_name, busname)

917

return {

918

"/clients/foo": self.client,

919

"/clients/barbar": self.other_client,

920

}[path]

921

return Bus()

922

923

class TestPrintTableCmd(TestCmd):

924

def test_normal(self):

925

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

926

expected_output = """

927

Name Enabled Timeout Last Successful Check

928

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

929

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

930

"""[1:-1]

931

self.assertEqual(output, expected_output)

932

def test_verbose(self):

933

output = PrintTableCmd(verbose=True).output(

934

self.clients.values())

935

expected_output = """

936

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

937

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

938

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

939

"""[1:-1]

940

self.assertEqual(output, expected_output)

941

def test_one_client(self):

942

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

943

expected_output = """

944

Name Enabled Timeout Last Successful Check

945

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

946

"""[1:-1]

947

self.assertEqual(output, expected_output)

948

949

class TestDumpJSONCmd(TestCmd):

950

def setUp(self):

951

self.expected_json = {

952

"foo": {

953

"Name": "foo",

954

"KeyID": ("92ed150794387c03ce684574b1139a65"

955

"94a34f895daaaf09fd8ea90a27cddb12"),

956

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

957

"Enabled": True,

958

"Timeout": 300000,

959

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

960

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

961

"Interval": 120000,

962

"Fingerprint": ("778827225BA7DE539C5A"

963

"7CFA59CFF7CDBD9A5920"),

964

"CheckerRunning": False,

965

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

966

"ApprovalPending": False,

967

"ApprovedByDefault": True,

968

"LastApprovalRequest": "",

969

"ApprovalDelay": 0,

970

"ApprovalDuration": 1000,

971

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

972

"ExtendedTimeout": 900000,

973

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

974

"LastCheckerStatus": 0,

975

},

976

"barbar": {

977

"Name": "barbar",

978

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

979

"6ab612cff5ad227247e46c2b020f441c"),

980

"Host": "192.0.2.3",

981

"Enabled": True,

982

"Timeout": 300000,

983

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

984

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

985

"Interval": 120000,

986

"Fingerprint": ("3E393AEAEFB84C7E89E2"

987

"F547B3A107558FCA3A27"),

988

"CheckerRunning": True,

989

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

990

"ApprovalPending": False,

991

"ApprovedByDefault": False,

992

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

993

"ApprovalDelay": 30000,

994

"ApprovalDuration": 1000,

995

"Checker": ":",

996

"ExtendedTimeout": 900000,

997

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

998

"LastCheckerStatus": -2,

999

},

1000

}

1001

return super(TestDumpJSONCmd, self).setUp()

1002

def test_normal(self):

1003

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

1004

self.assertDictEqual(json_data, self.expected_json)

1005

def test_one_client(self):

1006

clients = self.one_client

1007

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

1008

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

1009

self.assertDictEqual(json_data, expected_json)

1010

1011

class TestIsEnabledCmd(TestCmd):

1012

def test_is_enabled(self):

1013

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

1014

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

1015

def test_is_enabled_run_exits_successfully(self):

1016

with self.assertRaises(SystemExit) as e:

1017

IsEnabledCmd().run(self.one_client)

1018

if e.exception.code is not None:

1019

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

1020

else:

1021

self.assertIsNone(e.exception.code)

1022

def test_is_enabled_run_exits_with_failure(self):

1023

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

1024

with self.assertRaises(SystemExit) as e:

1025

IsEnabledCmd().run(self.one_client)

1026

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

1027

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

1028

else:

1029

self.assertIsNotNone(e.exception.code)

1030

1031

class TestRemoveCmd(TestCmd):

1032

def test_remove(self):

1033

class MockMandos(object):

1034

def __init__(self):

1035

self.calls = []

1036

def RemoveClient(self, dbus_path):

1037

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

1038

mandos = MockMandos()

1039

super(TestRemoveCmd, self).setUp()

1040

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

1041

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

1042

for clientpath in self.clients:

1043

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

1044

mandos.calls)

1045

1046

class TestApproveCmd(TestCmd):

1047

def test_approve(self):

1048

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

1049

for clientpath in self.clients:

1050

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

1051

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

1052

client.calls)

1053

1054

class TestDenyCmd(TestCmd):

1055

def test_deny(self):

1056

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

1057

for clientpath in self.clients:

1058

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

1059

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

1060

client.calls)

1061

1062

class TestEnableCmd(TestCmd):

1063

def test_enable(self):

1064

for clientpath in self.clients:

1065

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

1066

client.attributes["Enabled"] = False

1067

1068

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

1069

1070

for clientpath in self.clients:

1071

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

1072

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

1073

1074

class TestDisableCmd(TestCmd):

1075

def test_disable(self):

1076

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

1077

for clientpath in self.clients:

1078

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

1079

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

1080

1081

class Unique(object):

1082

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

1083

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

1084

1085

class TestPropertyCmd(TestCmd):

1086

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

1087

def runTest(self):

1088

if not hasattr(self, "command"):

1089

return

1090

values_to_get = getattr(self, "values_to_get",

1091

self.values_to_set)

1092

for value_to_set, value_to_get in zip(self.values_to_set,

1093

values_to_get):

1094

for clientpath in self.clients:

1095

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

1096

old_value = client.attributes[self.propname]

1097

self.assertNotIsInstance(old_value, Unique)

1098

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

1099

self.run_command(value_to_set, self.clients)

1100

for clientpath in self.clients:

1101

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

1102

value = client.attributes[self.propname]

1103

self.assertNotIsInstance(value, Unique)

1104

self.assertEqual(value, value_to_get)

1105

def run_command(self, value, clients):

1106

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

1107

1108

class TestBumpTimeoutCmd(TestPropertyCmd):

1109

command = BumpTimeoutCmd

1110

propname = "LastCheckedOK"

1111

values_to_set = [""]

1112

1113

class TestStartCheckerCmd(TestPropertyCmd):

1114

command = StartCheckerCmd

1115

propname = "CheckerRunning"

1116

values_to_set = [dbus.Boolean(True)]

1117

1118

class TestStopCheckerCmd(TestPropertyCmd):

1119

command = StopCheckerCmd

1120

propname = "CheckerRunning"

1121

values_to_set = [dbus.Boolean(False)]

1122

1123

class TestApproveByDefaultCmd(TestPropertyCmd):

1124

command = ApproveByDefaultCmd

1125

propname = "ApprovedByDefault"

1126

values_to_set = [dbus.Boolean(True)]

1127

1128

class TestDenyByDefaultCmd(TestPropertyCmd):

1129

command = DenyByDefaultCmd

1130

propname = "ApprovedByDefault"

1131

values_to_set = [dbus.Boolean(False)]

1132

1133

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1134

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

1135

ValueArgumentMixIn"""

1136

def runTest(self):

1137

if type(self) is TestValueArgumentPropertyCmd:

1138

return

1139

return super(TestValueArgumentPropertyCmd, self).runTest()

1140

def run_command(self, value, clients):

1141

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

1142

1143

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1144

command = SetCheckerCmd

1145

propname = "Checker"

1146

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

1147

1148

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1149

command = SetHostCmd

1150

propname = "Host"

1151

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

1152

1153

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1154

command = SetSecretCmd

1155

propname = "Secret"

1156

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

1157

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

1158

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

1159

1160

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1161

command = SetTimeoutCmd

1162

propname = "Timeout"

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 TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1171

command = SetExtendedTimeoutCmd

1172

propname = "ExtendedTimeout"

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 TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1181

command = SetIntervalCmd

1182

propname = "Interval"

1183

values_to_set = [datetime.timedelta(),

1184

datetime.timedelta(minutes=5),

1185

datetime.timedelta(seconds=1),

1186

datetime.timedelta(weeks=1),

1187

datetime.timedelta(weeks=52)]

1188

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

1189

1190

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1191

command = SetApprovalDelayCmd

1192

propname = "ApprovalDelay"

1193

values_to_set = [datetime.timedelta(),

1194

datetime.timedelta(minutes=5),

1195

datetime.timedelta(seconds=1),

1196

datetime.timedelta(weeks=1),

1197

datetime.timedelta(weeks=52)]

1198

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

1199

1200

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1201

command = SetApprovalDurationCmd

1202

propname = "ApprovalDuration"

1203

values_to_set = [datetime.timedelta(),

1204

datetime.timedelta(minutes=5),

1205

datetime.timedelta(seconds=1),

1206

datetime.timedelta(weeks=1),

1207

datetime.timedelta(weeks=52)]

1208

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

1209

1210

class Test_command_from_options(unittest.TestCase):

1211

def setUp(self):

1212

self.parser = argparse.ArgumentParser()

1213

add_command_line_options(self.parser)

1214

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

1215

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

1216

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

1217

options = self.parser.parse_args(args)

1218

check_option_syntax(self.parser, options)

1219

commands = commands_from_options(options)

1220

self.assertEqual(len(commands), 1)

1221

command = commands[0]

1222

self.assertIsInstance(command, command_cls)

1223

for key, value in cmd_attrs.items():

1224

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

1225

def test_print_table(self):

1226

self.assert_command_from_args([], PrintTableCmd,

1227

verbose=False)

1228

1229

def test_print_table_verbose(self):

1230

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

1231

verbose=True)

1232

1233

def test_print_table_verbose_short(self):

1234

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

1235

verbose=True)

1236

1237

def test_enable(self):

1238

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

1239

1240

def test_enable_short(self):

1241

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

1242

1243

def test_disable(self):

1244

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

1245

DisableCmd)

1246

1247

def test_disable_short(self):

1248

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

1249

1250

def test_bump_timeout(self):

1251

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

1252

BumpTimeoutCmd)

1253

1254

def test_bump_timeout_short(self):

1255

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

1256

1257

def test_start_checker(self):

1258

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

1259

StartCheckerCmd)

1260

1261

def test_stop_checker(self):

1262

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

1263

StopCheckerCmd)

1264

1265

def test_remove(self):

1266

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

1267

RemoveCmd)

1268

1269

def test_remove_short(self):

1270

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

1271

1272

def test_checker(self):

1273

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

1274

SetCheckerCmd, value_to_set=":")

1275

1276

def test_checker_empty(self):

1277

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

1278

SetCheckerCmd, value_to_set="")

1279

1280

def test_checker_short(self):

1281

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

1282

SetCheckerCmd, value_to_set=":")

1283

1284

def test_timeout(self):

1285

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

1286

SetTimeoutCmd,

1287

value_to_set=300000)

1288

1289

def test_timeout_short(self):

1290

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

1291

SetTimeoutCmd,

1292

value_to_set=300000)

1293

1294

def test_extended_timeout(self):

1295

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

1296

"foo"],

1297

SetExtendedTimeoutCmd,

1298

value_to_set=900000)

1299

1300

def test_interval(self):

1301

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

1302

SetIntervalCmd,

1303

value_to_set=120000)

1304

1305

def test_interval_short(self):

1306

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

1307

SetIntervalCmd,

1308

value_to_set=120000)

1309

1310

def test_approve_by_default(self):

1311

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

1312

ApproveByDefaultCmd)

1313

1314

def test_deny_by_default(self):

1315

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

1316

DenyByDefaultCmd)

1317

1318

def test_approval_delay(self):

1319

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

1320

"foo"], SetApprovalDelayCmd,

1321

value_to_set=30000)

1322

1323

def test_approval_duration(self):

1324

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

1325

"foo"], SetApprovalDurationCmd,

1326

value_to_set=1000)

1327

1328

def test_host(self):

1329

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

1330

"foo"], SetHostCmd,

1331

value_to_set="foo.example.org")

1332

1333

def test_host_short(self):

1334

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

1335

"foo"], SetHostCmd,

1336

value_to_set="foo.example.org")

1337

1338

def test_secret_devnull(self):

1339

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

1340

"foo"], SetSecretCmd,

1341

value_to_set=b"")

1342

1343

def test_secret_tempfile(self):

1344

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

1345

value = b"secret\0xyzzy\nbar"

1346

f.write(value)

1347

f.seek(0)

1348

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

1349

"foo"], SetSecretCmd,

1350

value_to_set=value)

1351

1352

def test_secret_devnull_short(self):

1353

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

1354

SetSecretCmd, value_to_set=b"")

1355

1356

def test_secret_tempfile_short(self):

1357

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

1358

value = b"secret\0xyzzy\nbar"

1359

f.write(value)

1360

f.seek(0)

1361

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

1362

SetSecretCmd,

1363

value_to_set=value)

1364

1365

def test_approve(self):

1366

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

1367

ApproveCmd)

1368

1369

def test_approve_short(self):

1370

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

1371

1372

def test_deny(self):

1373

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

1374

1375

def test_deny_short(self):

1376

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

1377

1378

def test_dump_json(self):

1379

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

1380

1381

def test_is_enabled(self):

1382

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

1383

IsEnabledCmd)

1384

1385

def test_is_enabled_short(self):

1386

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

1387

1388

def test_deny_before_remove(self):

1389

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

1390

check_option_syntax(self.parser, options)

1391

commands = commands_from_options(options)

1392

self.assertEqual(len(commands), 2)

1393

self.assertIsInstance(commands[0], DenyCmd)

1394

self.assertIsInstance(commands[1], RemoveCmd)

1395

1396

def test_deny_before_remove_reversed(self):

1397

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

1398

check_option_syntax(self.parser, options)

1399

commands = commands_from_options(options)

1400

self.assertEqual(len(commands), 2)

1401

self.assertIsInstance(commands[0], DenyCmd)

1402

self.assertIsInstance(commands[1], RemoveCmd)

1403

1404

1405

class Test_check_option_syntax(unittest.TestCase):

1406

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

1407

# check_option_syntax()

1408

actions = {

1409

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

1410

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

1411

# never used

1412

"enable": True,

1413

"disable": True,

1414

"bump_timeout": True,

1415

"start_checker": True,

1416

"stop_checker": True,

1417

"is_enabled": True,

1418

"remove": True,

1419

"checker": "x",

1420

"timeout": datetime.timedelta(),

1421

"extended_timeout": datetime.timedelta(),

1422

"interval": datetime.timedelta(),

1423

"approved_by_default": True,

1424

"approval_delay": datetime.timedelta(),

1425

"approval_duration": datetime.timedelta(),

1426

"host": "x",

1427

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

1428

"approve": True,

1429

"deny": True,

1430

}

1431

1432

def setUp(self):

1433

self.parser = argparse.ArgumentParser()

1434

add_command_line_options(self.parser)

1435

1436

@contextlib.contextmanager

1437

def assertParseError(self):

1438

with self.assertRaises(SystemExit) as e:

1439

with self.temporarily_suppress_stderr():

1440

yield

1441

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

1442

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

1443

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

1444

1445

@staticmethod

1446

@contextlib.contextmanager

1447

def temporarily_suppress_stderr():

1448

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

1449

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

1450

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

1451

os.close(null)

1452

try:

1453

yield

1454

finally:

1455

# restore stderr

1456

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

1457

os.close(stderrcopy)

1458

1459

def check_option_syntax(self, options):

1460

check_option_syntax(self.parser, options)

1461

1462

def test_actions_requires_client_or_all(self):

1463

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

1464

options = self.parser.parse_args()

1465

setattr(options, action, value)

1466

with self.assertParseError():

1467

self.check_option_syntax(options)

1468

1469

def test_actions_conflicts_with_verbose(self):

1470

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

1471

options = self.parser.parse_args()

1472

setattr(options, action, value)

1473

options.verbose = True

1474

with self.assertParseError():

1475

self.check_option_syntax(options)

1476

1477

def test_dump_json_conflicts_with_verbose(self):

1478

options = self.parser.parse_args()

1479

options.dump_json = True

1480

options.verbose = True

1481

with self.assertParseError():

1482

self.check_option_syntax(options)

1483

1484

def test_dump_json_conflicts_with_action(self):

1485

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

1486

options = self.parser.parse_args()

1487

setattr(options, action, value)

1488

options.dump_json = True

1489

with self.assertParseError():

1490

self.check_option_syntax(options)

1491

1492

def test_all_can_not_be_alone(self):

1493

options = self.parser.parse_args()

1494

options.all = True

1495

with self.assertParseError():

1496

self.check_option_syntax(options)

1497

1498

def test_all_is_ok_with_any_action(self):

1499

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

1500

options = self.parser.parse_args()

1501

setattr(options, action, value)

1502

options.all = True

1503

self.check_option_syntax(options)

1504

1505

def test_is_enabled_fails_without_client(self):

1506

options = self.parser.parse_args()

1507

options.is_enabled = True

1508

with self.assertParseError():

1509

self.check_option_syntax(options)

1510

1511

def test_is_enabled_works_with_one_client(self):

1512

options = self.parser.parse_args()

1513

options.is_enabled = True

1514

options.client = ["foo"]

1515

self.check_option_syntax(options)

1516

1517

def test_is_enabled_fails_with_two_clients(self):

1518

options = self.parser.parse_args()

1519

options.is_enabled = True

1520

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

1521

with self.assertParseError():

1522

self.check_option_syntax(options)

1523

1524

def test_remove_can_only_be_combined_with_action_deny(self):

1525

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

1526

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

1527

continue

1528

options = self.parser.parse_args()

1529

setattr(options, action, value)

1530

options.all = True

1531

options.remove = True

1532

with self.assertParseError():

1533

self.check_option_syntax(options)

1534

1535

1536

1537

def should_only_run_tests():

1538

parser = argparse.ArgumentParser(add_help=False)

1539

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

1540

args, unknown_args = parser.parse_known_args()

1541

run_tests = args.check

1542

if run_tests:

1543

# Remove --check argument from sys.argv

1544

sys.argv[1:] = unknown_args

1545

return run_tests

1546

1547

# Add all tests from doctest strings

1548

def load_tests(loader, tests, none):

1549

import doctest

1550

tests.addTests(doctest.DocTestSuite())

1551

return tests

1552

1553

if __name__ == "__main__":

1554

if should_only_run_tests():

1555

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

1556

unittest.main()

1557

else:

1558

main()