/mandos/trunk : revision 24.1.75

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.values()))

298

def output(self, clients):

299

raise NotImplementedError()

300

301

class PropertyCmd(Command):

302

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

303

def run_on_one_client(self, client, properties):

304

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

305

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

306

client.__dbus_object_path__,

307

dbus.PROPERTIES_IFACE, client_interface,

308

self.propname, self.value_to_set

309

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

310

else bool(self.value_to_set))

311

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

312

dbus_interface=dbus.PROPERTIES_IFACE)

313

@property

314

def propname(self):

315

raise NotImplementedError()

316

317

class ValueArgumentMixIn(object):

318

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

319

def __init__(self, value):

320

self.value_to_set = value

321

322

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

323

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

324

milliseconds."""

325

@property

326

def value_to_set(self):

327

return self._vts

328

@value_to_set.setter

329

def value_to_set(self, value):

330

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

331

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

332

333

# Actual (non-abstract) command classes

334

335

class PrintTableCmd(PrintCmd):

336

def __init__(self, verbose=False):

337

self.verbose = verbose

338

339

def output(self, clients):

340

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

341

keywords = default_keywords

342

if self.verbose:

343

keywords = self.all_keywords

344

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

345

346

class TableOfClients(object):

347

tableheaders = {

348

"Name": "Name",

349

"Enabled": "Enabled",

350

"Timeout": "Timeout",

351

"LastCheckedOK": "Last Successful Check",

352

"LastApprovalRequest": "Last Approval Request",

353

"Created": "Created",

354

"Interval": "Interval",

355

"Host": "Host",

356

"Fingerprint": "Fingerprint",

357

"KeyID": "Key ID",

358

"CheckerRunning": "Check Is Running",

359

"LastEnabled": "Last Enabled",

360

"ApprovalPending": "Approval Is Pending",

361

"ApprovedByDefault": "Approved By Default",

362

"ApprovalDelay": "Approval Delay",

363

"ApprovalDuration": "Approval Duration",

364

"Checker": "Checker",

365

"ExtendedTimeout": "Extended Timeout",

366

"Expires": "Expires",

367

"LastCheckerStatus": "Last Checker Status",

368

}

369

370

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

371

self.clients = clients

372

self.keywords = keywords

373

if tableheaders is not None:

374

self.tableheaders = tableheaders

375

376

def __str__(self):

377

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

378

379

if sys.version_info.major == 2:

380

__unicode__ = __str__

381

def __str__(self):

382

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

383

384

def rows(self):

385

format_string = self.row_formatting_string()

386

rows = [self.header_line(format_string)]

387

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

388

for client in self.clients)

389

return rows

390

391

def row_formatting_string(self):

392

"Format string used to format table rows"

393

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

394

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

395

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

396

for client in self.clients)),

397

key=key)

398

for key in self.keywords)

399

400

def string_from_client(self, client, key):

401

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

402

403

@staticmethod

404

def valuetostring(value, keyword):

405

if isinstance(value, dbus.Boolean):

406

return "Yes" if value else "No"

407

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

408

"ApprovalDuration", "ExtendedTimeout"):

409

return milliseconds_to_string(value)

410

return str(value)

411

412

def header_line(self, format_string):

413

return format_string.format(**self.tableheaders)

414

415

def client_line(self, client, format_string):

416

return format_string.format(

417

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

418

for key in self.keywords})

419

420

421

422

class DumpJSONCmd(PrintCmd):

423

def output(self, clients):

424

data = {client["Name"]:

425

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

426

for key in self.all_keywords}

427

for client in clients.values()}

428

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

429

@staticmethod

430

def dbus_boolean_to_bool(value):

431

if isinstance(value, dbus.Boolean):

432

value = bool(value)

433

return value

434

435

class IsEnabledCmd(Command):

436

def run_on_one_client(self, client, properties):

437

if self.is_enabled(client, properties):

438

sys.exit(0)

439

sys.exit(1)

440

def is_enabled(self, client, properties):

441

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

442

client.__dbus_object_path__,

443

dbus.PROPERTIES_IFACE, client_interface,

444

"Enabled")

445

return bool(client.Get(client_interface, "Enabled",

446

dbus_interface=dbus.PROPERTIES_IFACE))

447

448

class RemoveCmd(Command):

449

def run_on_one_client(self, client, properties):

450

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

451

server_path, server_interface,

452

str(client.__dbus_object_path__))

453

self.mandos.RemoveClient(client.__dbus_object_path__)

454

455

class ApproveCmd(Command):

456

def run_on_one_client(self, client, properties):

457

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

458

client.__dbus_object_path__, client_interface)

459

client.Approve(dbus.Boolean(True),

460

dbus_interface=client_interface)

461

462

class DenyCmd(Command):

463

def run_on_one_client(self, client, properties):

464

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

465

client.__dbus_object_path__, client_interface)

466

client.Approve(dbus.Boolean(False),

467

dbus_interface=client_interface)

468

469

class EnableCmd(PropertyCmd):

470

propname = "Enabled"

471

value_to_set = dbus.Boolean(True)

472

473

class DisableCmd(PropertyCmd):

474

propname = "Enabled"

475

value_to_set = dbus.Boolean(False)

476

477

class BumpTimeoutCmd(PropertyCmd):

478

propname = "LastCheckedOK"

479

value_to_set = ""

480

481

class StartCheckerCmd(PropertyCmd):

482

propname = "CheckerRunning"

483

value_to_set = dbus.Boolean(True)

484

485

class StopCheckerCmd(PropertyCmd):

486

propname = "CheckerRunning"

487

value_to_set = dbus.Boolean(False)

488

489

class ApproveByDefaultCmd(PropertyCmd):

490

propname = "ApprovedByDefault"

491

value_to_set = dbus.Boolean(True)

492

493

class DenyByDefaultCmd(PropertyCmd):

494

propname = "ApprovedByDefault"

495

value_to_set = dbus.Boolean(False)

496

497

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

498

propname = "Checker"

499

500

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

501

propname = "Host"

502

503

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

504

propname = "Secret"

505

@property

506

def value_to_set(self):

507

return self._vts

508

@value_to_set.setter

509

def value_to_set(self, value):

510

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

511

self._vts = value.read()

512

value.close()

513

514

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

515

propname = "Timeout"

516

517

class SetExtendedTimeoutCmd(PropertyCmd,

518

MillisecondsValueArgumentMixIn):

519

propname = "ExtendedTimeout"

520

521

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

522

propname = "Interval"

523

524

class SetApprovalDelayCmd(PropertyCmd,

525

MillisecondsValueArgumentMixIn):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(PropertyCmd,

529

MillisecondsValueArgumentMixIn):

530

propname = "ApprovalDuration"

531

532

def add_command_line_options(parser):

533

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

534

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

535

help="show version number and exit")

536

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

537

help="Select all clients")

538

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

539

help="Print all fields")

540

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

541

help="Dump client data in JSON format")

542

enable_disable = parser.add_mutually_exclusive_group()

543

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

544

help="Enable client")

545

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

546

action="store_true",

547

help="disable client")

548

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

549

help="Bump timeout for client")

550

start_stop_checker = parser.add_mutually_exclusive_group()

551

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

552

action="store_true",

553

help="Start checker for client")

554

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

555

action="store_true",

556

help="Stop checker for client")

557

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

558

help="Check if client is enabled")

559

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

560

help="Remove client")

561

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

562

help="Set checker command for client")

563

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

564

help="Set timeout for client")

565

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

566

help="Set extended timeout for client")

567

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

568

help="Set checker interval for client")

569

approve_deny_default = parser.add_mutually_exclusive_group()

570

approve_deny_default.add_argument(

571

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

572

default=None, dest="approved_by_default",

573

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

574

approve_deny_default.add_argument(

575

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

576

dest="approved_by_default",

577

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

578

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

579

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

580

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

581

help="Set duration of one client approval")

582

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

583

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

584

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

585

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

586

approve_deny = parser.add_mutually_exclusive_group()

587

approve_deny.add_argument(

588

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

589

help="Approve any current client request")

590

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

591

help="Deny any current client request")

592

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

593

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

594

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

595

help="Run self-test")

596

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

597

598

599

def commands_from_options(options):

600

601

commands = []

602

603

if options.dump_json:

604

commands.append(DumpJSONCmd())

605

606

if options.enable:

607

commands.append(EnableCmd())

608

609

if options.disable:

610

commands.append(DisableCmd())

611

612

if options.bump_timeout:

613

commands.append(BumpTimeoutCmd())

614

615

if options.start_checker:

616

commands.append(StartCheckerCmd())

617

618

if options.stop_checker:

619

commands.append(StopCheckerCmd())

620

621

if options.is_enabled:

622

commands.append(IsEnabledCmd())

623

624

if options.checker is not None:

625

commands.append(SetCheckerCmd(options.checker))

626

627

if options.timeout is not None:

628

commands.append(SetTimeoutCmd(options.timeout))

629

630

if options.extended_timeout:

631

commands.append(

632

SetExtendedTimeoutCmd(options.extended_timeout))

633

634

if options.interval is not None:

635

commands.append(SetIntervalCmd(options.interval))

636

637

if options.approved_by_default is not None:

638

if options.approved_by_default:

639

commands.append(ApproveByDefaultCmd())

640

else:

641

commands.append(DenyByDefaultCmd())

642

643

if options.approval_delay is not None:

644

commands.append(SetApprovalDelayCmd(options.approval_delay))

645

646

if options.approval_duration is not None:

647

commands.append(

648

SetApprovalDurationCmd(options.approval_duration))

649

650

if options.host is not None:

651

commands.append(SetHostCmd(options.host))

652

653

if options.secret is not None:

654

commands.append(SetSecretCmd(options.secret))

655

656

if options.approve:

657

commands.append(ApproveCmd())

658

659

if options.deny:

660

commands.append(DenyCmd())

661

662

if options.remove:

663

commands.append(RemoveCmd())

664

665

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

666

# optionally verbosely

667

if not commands:

668

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

669

670

return commands

671

672

673

def check_option_syntax(parser, options):

674

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

675

argparse"""

676

677

def has_actions(options):

678

return any((options.enable,

679

options.disable,

680

options.bump_timeout,

681

options.start_checker,

682

options.stop_checker,

683

options.is_enabled,

684

options.remove,

685

options.checker is not None,

686

options.timeout is not None,

687

options.extended_timeout is not None,

688

options.interval is not None,

689

options.approved_by_default is not None,

690

options.approval_delay is not None,

691

options.approval_duration is not None,

692

options.host is not None,

693

options.secret is not None,

694

options.approve,

695

options.deny))

696

697

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

698

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

699

if options.verbose and has_actions(options):

700

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

701

if options.dump_json and (options.verbose

702

or has_actions(options)):

703

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

704

if options.all and not has_actions(options):

705

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

706

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

707

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

708

if options.remove:

709

options.remove = False

710

if has_actions(options) and not options.deny:

711

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

712

options.remove = True

713

714

715

def main():

716

parser = argparse.ArgumentParser()

717

718

add_command_line_options(parser)

719

720

options = parser.parse_args()

721

722

check_option_syntax(parser, options)

723

724

clientnames = options.client

725

726

if options.debug:

727

log.setLevel(logging.DEBUG)

728

729

try:

730

bus = dbus.SystemBus()

731

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

732

server_path)

733

mandos_dbus_objc = bus.get_object(busname, server_path)

734

except dbus.exceptions.DBusException:

735

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

736

sys.exit(1)

737

738

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_interface)

740

mandos_serv_object_manager = dbus.Interface(

741

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

743

# Filter out log message from dbus module

744

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

745

class NullFilter(logging.Filter):

746

def filter(self, record):

747

return False

748

dbus_filter = NullFilter()

749

try:

750

dbus_logger.addFilter(dbus_filter)

751

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

752

server_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_interface]

754

for path, ifs_and_props in

755

mandos_serv_object_manager

756

.GetManagedObjects().items()

757

if client_interface in ifs_and_props}

758

except dbus.exceptions.DBusException as e:

759

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

760

"\n%s", e)

761

sys.exit(1)

762

finally:

763

# restore dbus logger

764

dbus_logger.removeFilter(dbus_filter)

765

766

# Compile dict of (clients: properties) to process

767

clients = {}

768

769

if not clientnames:

770

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

771

busname, str(path)) and False) or

772

bus.get_object(busname, path): properties

773

for path, properties in mandos_clients.items()}

774

else:

775

for name in clientnames:

776

for path, client in mandos_clients.items():

777

if client["Name"] == name:

778

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

779

busname, str(path))

780

client_objc = bus.get_object(busname, path)

781

clients[client_objc] = client

782

break

783

else:

784

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

785

sys.exit(1)

786

787

# Run all commands on clients

788

commands = commands_from_options(options)

789

for command in commands:

790

command.run(mandos_serv, clients)

791

792

793

class Test_milliseconds_to_string(unittest.TestCase):

794

def test_all(self):

795

self.assertEqual(milliseconds_to_string(93785000),

796

"1T02:03:05")

797

def test_no_days(self):

798

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

799

def test_all_zero(self):

800

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

801

def test_no_fractional_seconds(self):

802

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

803

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

804

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

805

806

class Test_string_to_delta(unittest.TestCase):

807

def test_handles_basic_rfc3339(self):

808

self.assertEqual(string_to_delta("PT0S"),

809

datetime.timedelta())

810

self.assertEqual(string_to_delta("P0D"),

811

datetime.timedelta())

812

self.assertEqual(string_to_delta("PT1S"),

813

datetime.timedelta(0, 1))

814

self.assertEqual(string_to_delta("PT2H"),

815

datetime.timedelta(0, 7200))

816

def test_falls_back_to_pre_1_6_1_with_warning(self):

817

# assertLogs only exists in Python 3.4

818

if hasattr(self, "assertLogs"):

819

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

820

value = string_to_delta("2h")

821

else:

822

class WarningFilter(logging.Filter):

823

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

824

def filter(self, record):

825

is_warning = record.levelno >= logging.WARNING

826

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

827

False)

828

return not is_warning

829

warning_filter = WarningFilter()

830

log.addFilter(warning_filter)

831

try:

832

value = string_to_delta("2h")

833

finally:

834

log.removeFilter(warning_filter)

835

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

836

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

837

838

839

class TestCmd(unittest.TestCase):

840

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

841

def setUp(self):

842

testcase = self

843

class MockClient(object):

844

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

845

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

846

self.attributes = attributes

847

self.attributes["Name"] = name

848

self.calls = []

849

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

850

testcase.assertEqual(interface, client_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

self.attributes[propname] = value

854

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

855

testcase.assertEqual(interface, client_interface)

856

testcase.assertEqual(dbus_interface,

857

dbus.PROPERTIES_IFACE)

858

return self.attributes[propname]

859

def Approve(self, approve, dbus_interface):

860

testcase.assertEqual(dbus_interface, client_interface)

861

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

862

dbus_interface)))

863

self.client = MockClient(

864

"foo",

865

KeyID=("92ed150794387c03ce684574b1139a65"

866

"94a34f895daaaf09fd8ea90a27cddb12"),

867

Secret=b"secret",

868

Host="foo.example.org",

869

Enabled=dbus.Boolean(True),

870

Timeout=300000,

871

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

872

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

873

Interval=120000,

874

Fingerprint=("778827225BA7DE539C5A"

875

"7CFA59CFF7CDBD9A5920"),

876

CheckerRunning=dbus.Boolean(False),

877

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

878

ApprovalPending=dbus.Boolean(False),

879

ApprovedByDefault=dbus.Boolean(True),

880

LastApprovalRequest="",

881

ApprovalDelay=0,

882

ApprovalDuration=1000,

883

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

884

ExtendedTimeout=900000,

885

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

886

LastCheckerStatus=0)

887

self.other_client = MockClient(

888

"barbar",

889

KeyID=("0558568eedd67d622f5c83b35a115f79"

890

"6ab612cff5ad227247e46c2b020f441c"),

891

Secret=b"secretbar",

892

Host="192.0.2.3",

893

Enabled=dbus.Boolean(True),

894

Timeout=300000,

895

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

896

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

897

Interval=120000,

898

Fingerprint=("3E393AEAEFB84C7E89E2"

899

"F547B3A107558FCA3A27"),

900

CheckerRunning=dbus.Boolean(True),

901

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

902

ApprovalPending=dbus.Boolean(False),

903

ApprovedByDefault=dbus.Boolean(False),

904

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

905

ApprovalDelay=30000,

906

ApprovalDuration=1000,

907

Checker=":",

908

ExtendedTimeout=900000,

909

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

910

LastCheckerStatus=-2)

911

self.clients = collections.OrderedDict(

912

[

913

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

914

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

915

])

916

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

917

918

class TestPrintTableCmd(TestCmd):

919

def test_normal(self):

920

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

921

expected_output = """

922

Name Enabled Timeout Last Successful Check

923

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

924

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

925

"""[1:-1]

926

self.assertEqual(output, expected_output)

927

def test_verbose(self):

928

output = PrintTableCmd(verbose=True).output(

929

self.clients.values())

930

expected_output = """

931

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

932

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

933

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

934

"""[1:-1]

935

self.assertEqual(output, expected_output)

936

def test_one_client(self):

937

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

938

expected_output = """

939

Name Enabled Timeout Last Successful Check

940

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

941

"""[1:-1]

942

self.assertEqual(output, expected_output)

943

944

class TestDumpJSONCmd(TestCmd):

945

def setUp(self):

946

self.expected_json = {

947

"foo": {

948

"Name": "foo",

949

"KeyID": ("92ed150794387c03ce684574b1139a65"

950

"94a34f895daaaf09fd8ea90a27cddb12"),

951

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

952

"Enabled": True,

953

"Timeout": 300000,

954

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

955

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

956

"Interval": 120000,

957

"Fingerprint": ("778827225BA7DE539C5A"

958

"7CFA59CFF7CDBD9A5920"),

959

"CheckerRunning": False,

960

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

961

"ApprovalPending": False,

962

"ApprovedByDefault": True,

963

"LastApprovalRequest": "",

964

"ApprovalDelay": 0,

965

"ApprovalDuration": 1000,

966

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

967

"ExtendedTimeout": 900000,

968

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

969

"LastCheckerStatus": 0,

970

},

971

"barbar": {

972

"Name": "barbar",

973

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

974

"6ab612cff5ad227247e46c2b020f441c"),

975

"Host": "192.0.2.3",

976

"Enabled": True,

977

"Timeout": 300000,

978

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

979

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

980

"Interval": 120000,

981

"Fingerprint": ("3E393AEAEFB84C7E89E2"

982

"F547B3A107558FCA3A27"),

983

"CheckerRunning": True,

984

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

985

"ApprovalPending": False,

986

"ApprovedByDefault": False,

987

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

988

"ApprovalDelay": 30000,

989

"ApprovalDuration": 1000,

990

"Checker": ":",

991

"ExtendedTimeout": 900000,

992

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

993

"LastCheckerStatus": -2,

994

},

995

}

996

return super(TestDumpJSONCmd, self).setUp()

997

def test_normal(self):

998

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

999

self.assertDictEqual(json_data, self.expected_json)

1000

def test_one_client(self):

1001

clients = self.one_client

1002

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

1003

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

1004

self.assertDictEqual(json_data, expected_json)

1005

1006

class TestIsEnabledCmd(TestCmd):

1007

def test_is_enabled(self):

1008

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

1009

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

1010

def test_is_enabled_run_exits_successfully(self):

1011

with self.assertRaises(SystemExit) as e:

1012

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

1013

if e.exception.code is not None:

1014

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

1015

else:

1016

self.assertIsNone(e.exception.code)

1017

def test_is_enabled_run_exits_with_failure(self):

1018

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

1019

with self.assertRaises(SystemExit) as e:

1020

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

1021

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

1022

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

1023

else:

1024

self.assertIsNotNone(e.exception.code)

1025

1026

class TestRemoveCmd(TestCmd):

1027

def test_remove(self):

1028

class MockMandos(object):

1029

def __init__(self):

1030

self.calls = []

1031

def RemoveClient(self, dbus_path):

1032

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

1033

mandos = MockMandos()

1034

super(TestRemoveCmd, self).setUp()

1035

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

1036

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

1037

for client in self.clients:

1038

self.assertIn(("RemoveClient",

1039

(client.__dbus_object_path__,)),

1040

mandos.calls)

1041

1042

class TestApproveCmd(TestCmd):

1043

def test_approve(self):

1044

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

1045

for client in self.clients:

1046

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

1047

client.calls)

1048

1049

class TestDenyCmd(TestCmd):

1050

def test_deny(self):

1051

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

1052

for client in self.clients:

1053

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

1054

client.calls)

1055

1056

class TestEnableCmd(TestCmd):

1057

def test_enable(self):

1058

for client in self.clients:

1059

client.attributes["Enabled"] = False

1060

1061

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

1062

1063

for client in self.clients:

1064

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

1065

1066

class TestDisableCmd(TestCmd):

1067

def test_disable(self):

1068

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

1069

1070

for client in self.clients:

1071

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

1072

1073

class Unique(object):

1074

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

1075

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

1076

1077

class TestPropertyCmd(TestCmd):

1078

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

1079

def runTest(self):

1080

if not hasattr(self, "command"):

1081

return

1082

values_to_get = getattr(self, "values_to_get",

1083

self.values_to_set)

1084

for value_to_set, value_to_get in zip(self.values_to_set,

1085

values_to_get):

1086

for client in self.clients:

1087

old_value = client.attributes[self.propname]

1088

self.assertNotIsInstance(old_value, Unique)

1089

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

1090

self.run_command(value_to_set, self.clients)

1091

for client in self.clients:

1092

value = client.attributes[self.propname]

1093

self.assertNotIsInstance(value, Unique)

1094

self.assertEqual(value, value_to_get)

1095

def run_command(self, value, clients):

1096

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

1097

1098

class TestBumpTimeoutCmd(TestPropertyCmd):

1099

command = BumpTimeoutCmd

1100

propname = "LastCheckedOK"

1101

values_to_set = [""]

1102

1103

class TestStartCheckerCmd(TestPropertyCmd):

1104

command = StartCheckerCmd

1105

propname = "CheckerRunning"

1106

values_to_set = [dbus.Boolean(True)]

1107

1108

class TestStopCheckerCmd(TestPropertyCmd):

1109

command = StopCheckerCmd

1110

propname = "CheckerRunning"

1111

values_to_set = [dbus.Boolean(False)]

1112

1113

class TestApproveByDefaultCmd(TestPropertyCmd):

1114

command = ApproveByDefaultCmd

1115

propname = "ApprovedByDefault"

1116

values_to_set = [dbus.Boolean(True)]

1117

1118

class TestDenyByDefaultCmd(TestPropertyCmd):

1119

command = DenyByDefaultCmd

1120

propname = "ApprovedByDefault"

1121

values_to_set = [dbus.Boolean(False)]

1122

1123

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1124

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

1125

ValueArgumentMixIn"""

1126

def runTest(self):

1127

if type(self) is TestValueArgumentPropertyCmd:

1128

return

1129

return super(TestValueArgumentPropertyCmd, self).runTest()

1130

def run_command(self, value, clients):

1131

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

1132

1133

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1134

command = SetCheckerCmd

1135

propname = "Checker"

1136

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

1137

1138

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1139

command = SetHostCmd

1140

propname = "Host"

1141

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

1142

1143

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1144

command = SetSecretCmd

1145

propname = "Secret"

1146

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

1147

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

1148

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

1149

1150

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1151

command = SetTimeoutCmd

1152

propname = "Timeout"

1153

values_to_set = [datetime.timedelta(),

1154

datetime.timedelta(minutes=5),

1155

datetime.timedelta(seconds=1),

1156

datetime.timedelta(weeks=1),

1157

datetime.timedelta(weeks=52)]

1158

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

1159

1160

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1161

command = SetExtendedTimeoutCmd

1162

propname = "ExtendedTimeout"

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

1171

command = SetIntervalCmd

1172

propname = "Interval"

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

1181

command = SetApprovalDelayCmd

1182

propname = "ApprovalDelay"

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

1191

command = SetApprovalDurationCmd

1192

propname = "ApprovalDuration"

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 Test_command_from_options(unittest.TestCase):

1201

def setUp(self):

1202

self.parser = argparse.ArgumentParser()

1203

add_command_line_options(self.parser)

1204

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

1205

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

1206

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

1207

options = self.parser.parse_args(args)

1208

check_option_syntax(self.parser, options)

1209

commands = commands_from_options(options)

1210

self.assertEqual(len(commands), 1)

1211

command = commands[0]

1212

self.assertIsInstance(command, command_cls)

1213

for key, value in cmd_attrs.items():

1214

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

1215

def test_print_table(self):

1216

self.assert_command_from_args([], PrintTableCmd,

1217

verbose=False)

1218

1219

def test_print_table_verbose(self):

1220

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

1221

verbose=True)

1222

1223

def test_print_table_verbose_short(self):

1224

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

1225

verbose=True)

1226

1227

def test_enable(self):

1228

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

1229

1230

def test_enable_short(self):

1231

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

1232

1233

def test_disable(self):

1234

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

1235

DisableCmd)

1236

1237

def test_disable_short(self):

1238

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

1239

1240

def test_bump_timeout(self):

1241

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

1242

BumpTimeoutCmd)

1243

1244

def test_bump_timeout_short(self):

1245

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

1246

1247

def test_start_checker(self):

1248

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

1249

StartCheckerCmd)

1250

1251

def test_stop_checker(self):

1252

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

1253

StopCheckerCmd)

1254

1255

def test_remove(self):

1256

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

1257

RemoveCmd)

1258

1259

def test_remove_short(self):

1260

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

1261

1262

def test_checker(self):

1263

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

1264

SetCheckerCmd, value_to_set=":")

1265

1266

def test_checker_empty(self):

1267

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

1268

SetCheckerCmd, value_to_set="")

1269

1270

def test_checker_short(self):

1271

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

1272

SetCheckerCmd, value_to_set=":")

1273

1274

def test_timeout(self):

1275

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

1276

SetTimeoutCmd,

1277

value_to_set=300000)

1278

1279

def test_timeout_short(self):

1280

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

1281

SetTimeoutCmd,

1282

value_to_set=300000)

1283

1284

def test_extended_timeout(self):

1285

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

1286

"foo"],

1287

SetExtendedTimeoutCmd,

1288

value_to_set=900000)

1289

1290

def test_interval(self):

1291

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

1292

SetIntervalCmd,

1293

value_to_set=120000)

1294

1295

def test_interval_short(self):

1296

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

1297

SetIntervalCmd,

1298

value_to_set=120000)

1299

1300

def test_approve_by_default(self):

1301

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

1302

ApproveByDefaultCmd)

1303

1304

def test_deny_by_default(self):

1305

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

1306

DenyByDefaultCmd)

1307

1308

def test_approval_delay(self):

1309

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

1310

"foo"], SetApprovalDelayCmd,

1311

value_to_set=30000)

1312

1313

def test_approval_duration(self):

1314

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

1315

"foo"], SetApprovalDurationCmd,

1316

value_to_set=1000)

1317

1318

def test_host(self):

1319

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

1320

"foo"], SetHostCmd,

1321

value_to_set="foo.example.org")

1322

1323

def test_host_short(self):

1324

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

1325

"foo"], SetHostCmd,

1326

value_to_set="foo.example.org")

1327

1328

def test_secret_devnull(self):

1329

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

1330

"foo"], SetSecretCmd,

1331

value_to_set=b"")

1332

1333

def test_secret_tempfile(self):

1334

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

1335

value = b"secret\0xyzzy\nbar"

1336

f.write(value)

1337

f.seek(0)

1338

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

1339

"foo"], SetSecretCmd,

1340

value_to_set=value)

1341

1342

def test_secret_devnull_short(self):

1343

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

1344

SetSecretCmd, value_to_set=b"")

1345

1346

def test_secret_tempfile_short(self):

1347

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

1348

value = b"secret\0xyzzy\nbar"

1349

f.write(value)

1350

f.seek(0)

1351

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

1352

SetSecretCmd,

1353

value_to_set=value)

1354

1355

def test_approve(self):

1356

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

1357

ApproveCmd)

1358

1359

def test_approve_short(self):

1360

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

1361

1362

def test_deny(self):

1363

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

1364

1365

def test_deny_short(self):

1366

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

1367

1368

def test_dump_json(self):

1369

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

1370

1371

def test_is_enabled(self):

1372

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

1373

IsEnabledCmd)

1374

1375

def test_is_enabled_short(self):

1376

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

1377

1378

def test_deny_before_remove(self):

1379

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

1380

check_option_syntax(self.parser, options)

1381

commands = commands_from_options(options)

1382

self.assertEqual(len(commands), 2)

1383

self.assertIsInstance(commands[0], DenyCmd)

1384

self.assertIsInstance(commands[1], RemoveCmd)

1385

1386

def test_deny_before_remove_reversed(self):

1387

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

1388

check_option_syntax(self.parser, options)

1389

commands = commands_from_options(options)

1390

self.assertEqual(len(commands), 2)

1391

self.assertIsInstance(commands[0], DenyCmd)

1392

self.assertIsInstance(commands[1], RemoveCmd)

1393

1394

1395

class Test_check_option_syntax(unittest.TestCase):

1396

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

1397

# check_option_syntax()

1398

actions = {

1399

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

1400

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

1401

# never used

1402

"enable": True,

1403

"disable": True,

1404

"bump_timeout": True,

1405

"start_checker": True,

1406

"stop_checker": True,

1407

"is_enabled": True,

1408

"remove": True,

1409

"checker": "x",

1410

"timeout": datetime.timedelta(),

1411

"extended_timeout": datetime.timedelta(),

1412

"interval": datetime.timedelta(),

1413

"approved_by_default": True,

1414

"approval_delay": datetime.timedelta(),

1415

"approval_duration": datetime.timedelta(),

1416

"host": "x",

1417

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

1418

"approve": True,

1419

"deny": True,

1420

}

1421

1422

def setUp(self):

1423

self.parser = argparse.ArgumentParser()

1424

add_command_line_options(self.parser)

1425

1426

@contextlib.contextmanager

1427

def assertParseError(self):

1428

with self.assertRaises(SystemExit) as e:

1429

with self.temporarily_suppress_stderr():

1430

yield

1431

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

1432

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

1433

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

1434

1435

@staticmethod

1436

@contextlib.contextmanager

1437

def temporarily_suppress_stderr():

1438

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

1439

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

1440

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

1441

os.close(null)

1442

try:

1443

yield

1444

finally:

1445

# restore stderr

1446

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

1447

os.close(stderrcopy)

1448

1449

def check_option_syntax(self, options):

1450

check_option_syntax(self.parser, options)

1451

1452

def test_actions_requires_client_or_all(self):

1453

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

1454

options = self.parser.parse_args()

1455

setattr(options, action, value)

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_actions_conflicts_with_verbose(self):

1460

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

1461

options = self.parser.parse_args()

1462

setattr(options, action, value)

1463

options.verbose = True

1464

with self.assertParseError():

1465

self.check_option_syntax(options)

1466

1467

def test_dump_json_conflicts_with_verbose(self):

1468

options = self.parser.parse_args()

1469

options.dump_json = True

1470

options.verbose = True

1471

with self.assertParseError():

1472

self.check_option_syntax(options)

1473

1474

def test_dump_json_conflicts_with_action(self):

1475

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

1476

options = self.parser.parse_args()

1477

setattr(options, action, value)

1478

options.dump_json = True

1479

with self.assertParseError():

1480

self.check_option_syntax(options)

1481

1482

def test_all_can_not_be_alone(self):

1483

options = self.parser.parse_args()

1484

options.all = True

1485

with self.assertParseError():

1486

self.check_option_syntax(options)

1487

1488

def test_all_is_ok_with_any_action(self):

1489

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

1490

options = self.parser.parse_args()

1491

setattr(options, action, value)

1492

options.all = True

1493

self.check_option_syntax(options)

1494

1495

def test_is_enabled_fails_without_client(self):

1496

options = self.parser.parse_args()

1497

options.is_enabled = True

1498

with self.assertParseError():

1499

self.check_option_syntax(options)

1500

1501

def test_is_enabled_works_with_one_client(self):

1502

options = self.parser.parse_args()

1503

options.is_enabled = True

1504

options.client = ["foo"]

1505

self.check_option_syntax(options)

1506

1507

def test_is_enabled_fails_with_two_clients(self):

1508

options = self.parser.parse_args()

1509

options.is_enabled = True

1510

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

1511

with self.assertParseError():

1512

self.check_option_syntax(options)

1513

1514

def test_remove_can_only_be_combined_with_action_deny(self):

1515

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

1516

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

1517

continue

1518

options = self.parser.parse_args()

1519

setattr(options, action, value)

1520

options.all = True

1521

options.remove = True

1522

with self.assertParseError():

1523

self.check_option_syntax(options)

1524

1525

1526

1527

def should_only_run_tests():

1528

parser = argparse.ArgumentParser(add_help=False)

1529

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

1530

args, unknown_args = parser.parse_known_args()

1531

run_tests = args.check

1532

if run_tests:

1533

# Remove --check argument from sys.argv

1534

sys.argv[1:] = unknown_args

1535

return run_tests

1536

1537

# Add all tests from doctest strings

1538

def load_tests(loader, tests, none):

1539

import doctest

1540

tests.addTests(doctest.DocTestSuite())

1541

return tests

1542

1543

if __name__ == "__main__":

1544

if should_only_run_tests():

1545

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

1546

unittest.main()

1547

else:

1548

main()