/mandos/trunk : revision 1045.1.1

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, mandos_objmgr, 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, mandos_objmgr, clients):

297

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

298

299

class PropertyCmd(Command):

300

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

301

def run_on_one_client(self, client, properties):

302

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

303

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

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

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

308

else bool(self.value_to_set))

309

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

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

312

class ValueArgumentMixIn(object):

313

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

314

def __init__(self, value):

315

self.value_to_set = value

316

317

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

318

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

319

milliseconds."""

320

@property

321

def value_to_set(self):

322

return self._vts

323

@value_to_set.setter

324

def value_to_set(self, value):

325

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

326

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

327

328

# Actual (non-abstract) command classes

329

330

class PrintTableCmd(PrintCmd):

331

def __init__(self, verbose=False):

332

self.verbose = verbose

333

334

def output(self, mandos_objmgr, clients):

335

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

336

keywords = default_keywords

337

if self.verbose:

338

keywords = self.all_keywords

339

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

340

341

class TableOfClients(object):

342

tableheaders = {

343

"Name": "Name",

344

"Enabled": "Enabled",

345

"Timeout": "Timeout",

346

"LastCheckedOK": "Last Successful Check",

347

"LastApprovalRequest": "Last Approval Request",

348

"Created": "Created",

349

"Interval": "Interval",

350

"Host": "Host",

351

"Fingerprint": "Fingerprint",

352

"KeyID": "Key ID",

353

"CheckerRunning": "Check Is Running",

354

"LastEnabled": "Last Enabled",

355

"ApprovalPending": "Approval Is Pending",

356

"ApprovedByDefault": "Approved By Default",

357

"ApprovalDelay": "Approval Delay",

358

"ApprovalDuration": "Approval Duration",

359

"Checker": "Checker",

360

"ExtendedTimeout": "Extended Timeout",

361

"Expires": "Expires",

362

"LastCheckerStatus": "Last Checker Status",

363

}

364

365

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

366

self.clients = clients

367

self.keywords = keywords

368

if tableheaders is not None:

369

self.tableheaders = tableheaders

370

371

def __str__(self):

372

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

373

374

if sys.version_info.major == 2:

375

__unicode__ = __str__

376

def __str__(self):

377

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

378

379

def rows(self):

380

format_string = self.row_formatting_string()

381

rows = [self.header_line(format_string)]

382

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

383

for client in self.clients)

384

return rows

385

386

def row_formatting_string(self):

387

"Format string used to format table rows"

388

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

389

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

390

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

391

for client in self.clients)),

392

key=key)

393

for key in self.keywords)

394

395

def string_from_client(self, client, key):

396

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

397

398

@staticmethod

399

def valuetostring(value, keyword):

400

if isinstance(value, dbus.Boolean):

401

return "Yes" if value else "No"

402

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

403

"ApprovalDuration", "ExtendedTimeout"):

404

return milliseconds_to_string(value)

405

return str(value)

406

407

def header_line(self, format_string):

408

return format_string.format(**self.tableheaders)

409

410

def client_line(self, client, format_string):

411

return format_string.format(

412

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

413

for key in self.keywords})

414

415

416

417

class DumpJSONCmd(PrintCmd):

418

def output(self, mandos_objmgr, clients):

419

data = {client["Name"]:

420

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

421

for key in self.all_keywords}

422

for client in clients.values()}

423

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

424

@staticmethod

425

def dbus_boolean_to_bool(value):

426

if isinstance(value, dbus.Boolean):

427

value = bool(value)

428

return value

429

430

class IsEnabledCmd(Command):

431

def run_on_one_client(self, client, properties):

432

if self.is_enabled(client, properties):

433

sys.exit(0)

434

sys.exit(1)

435

def is_enabled(self, client, properties):

436

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

437

client.__dbus_object_path__,

438

dbus.PROPERTIES_IFACE, client_interface,

439

"Enabled")

440

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

441

dbus_interface=dbus.PROPERTIES_IFACE))

442

443

class RemoveCmd(Command):

444

def run_on_one_client(self, client, properties):

445

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

446

server_path, server_interface,

447

str(client.__dbus_object_path__))

448

self.mandos.RemoveClient(client.__dbus_object_path__)

449

450

class ApproveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

client.__dbus_object_path__, client_interface)

454

client.Approve(dbus.Boolean(True),

455

dbus_interface=client_interface)

456

457

class DenyCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_interface)

461

client.Approve(dbus.Boolean(False),

462

dbus_interface=client_interface)

463

464

class EnableCmd(PropertyCmd):

465

property = "Enabled"

466

value_to_set = dbus.Boolean(True)

467

468

class DisableCmd(PropertyCmd):

469

property = "Enabled"

470

value_to_set = dbus.Boolean(False)

471

472

class BumpTimeoutCmd(PropertyCmd):

473

property = "LastCheckedOK"

474

value_to_set = ""

475

476

class StartCheckerCmd(PropertyCmd):

477

property = "CheckerRunning"

478

value_to_set = dbus.Boolean(True)

479

480

class StopCheckerCmd(PropertyCmd):

481

property = "CheckerRunning"

482

value_to_set = dbus.Boolean(False)

483

484

class ApproveByDefaultCmd(PropertyCmd):

485

property = "ApprovedByDefault"

486

value_to_set = dbus.Boolean(True)

487

488

class DenyByDefaultCmd(PropertyCmd):

489

property = "ApprovedByDefault"

490

value_to_set = dbus.Boolean(False)

491

492

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

493

property = "Checker"

494

495

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

496

property = "Host"

497

498

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

499

@property

500

def value_to_set(self):

501

return self._vts

502

@value_to_set.setter

503

def value_to_set(self, value):

504

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

505

self._vts = value.read()

506

value.close()

507

property = "Secret"

508

509

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

510

property = "Timeout"

511

512

class SetExtendedTimeoutCmd(PropertyCmd,

513

MillisecondsValueArgumentMixIn):

514

property = "ExtendedTimeout"

515

516

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

517

property = "Interval"

518

519

class SetApprovalDelayCmd(PropertyCmd,

520

MillisecondsValueArgumentMixIn):

521

property = "ApprovalDelay"

522

523

class SetApprovalDurationCmd(PropertyCmd,

524

MillisecondsValueArgumentMixIn):

525

property = "ApprovalDuration"

526

527

def add_command_line_options(parser):

528

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

529

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

530

help="show version number and exit")

531

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

532

help="Select all clients")

533

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

534

help="Print all fields")

535

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

536

help="Dump client data in JSON format")

537

enable_disable = parser.add_mutually_exclusive_group()

538

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

539

help="Enable client")

540

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

541

action="store_true",

542

help="disable client")

543

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

544

help="Bump timeout for client")

545

start_stop_checker = parser.add_mutually_exclusive_group()

546

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

547

action="store_true",

548

help="Start checker for client")

549

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

550

action="store_true",

551

help="Stop checker for client")

552

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

553

help="Check if client is enabled")

554

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

555

help="Remove client")

556

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

557

help="Set checker command for client")

558

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

559

help="Set timeout for client")

560

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

561

help="Set extended timeout for client")

562

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

563

help="Set checker interval for client")

564

approve_deny_default = parser.add_mutually_exclusive_group()

565

approve_deny_default.add_argument(

566

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

567

default=None, dest="approved_by_default",

568

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

569

approve_deny_default.add_argument(

570

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

571

dest="approved_by_default",

572

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

573

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

574

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

575

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

576

help="Set duration of one client approval")

577

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

578

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

579

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

580

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

581

approve_deny = parser.add_mutually_exclusive_group()

582

approve_deny.add_argument(

583

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

584

help="Approve any current client request")

585

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

586

help="Deny any current client request")

587

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

588

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

589

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

590

help="Run self-test")

591

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

592

593

594

def commands_from_options(options):

595

596

commands = []

597

598

if options.dump_json:

599

commands.append(DumpJSONCmd())

600

601

if options.enable:

602

commands.append(EnableCmd())

603

604

if options.disable:

605

commands.append(DisableCmd())

606

607

if options.bump_timeout:

608

commands.append(BumpTimeoutCmd())

609

610

if options.start_checker:

611

commands.append(StartCheckerCmd())

612

613

if options.stop_checker:

614

commands.append(StopCheckerCmd())

615

616

if options.is_enabled:

617

commands.append(IsEnabledCmd())

618

619

if options.checker is not None:

620

commands.append(SetCheckerCmd(options.checker))

621

622

if options.timeout is not None:

623

commands.append(SetTimeoutCmd(options.timeout))

624

625

if options.extended_timeout:

626

commands.append(

627

SetExtendedTimeoutCmd(options.extended_timeout))

628

629

if options.interval is not None:

630

commands.append(SetIntervalCmd(options.interval))

631

632

if options.approved_by_default is not None:

633

if options.approved_by_default:

634

commands.append(ApproveByDefaultCmd())

635

else:

636

commands.append(DenyByDefaultCmd())

637

638

if options.approval_delay is not None:

639

commands.append(SetApprovalDelayCmd(options.approval_delay))

640

641

if options.approval_duration is not None:

642

commands.append(

643

SetApprovalDurationCmd(options.approval_duration))

644

645

if options.host is not None:

646

commands.append(SetHostCmd(options.host))

647

648

if options.secret is not None:

649

commands.append(SetSecretCmd(options.secret))

650

651

if options.approve:

652

commands.append(ApproveCmd())

653

654

if options.deny:

655

commands.append(DenyCmd())

656

657

if options.remove:

658

commands.append(RemoveCmd())

659

660

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

661

# optionally verbosely

662

if not commands:

663

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

664

665

return commands

666

667

668

def check_option_syntax(parser, options):

669

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

670

argparse"""

671

672

def has_actions(options):

673

return any((options.enable,

674

options.disable,

675

options.bump_timeout,

676

options.start_checker,

677

options.stop_checker,

678

options.is_enabled,

679

options.remove,

680

options.checker is not None,

681

options.timeout is not None,

682

options.extended_timeout is not None,

683

options.interval is not None,

684

options.approved_by_default is not None,

685

options.approval_delay is not None,

686

options.approval_duration is not None,

687

options.host is not None,

688

options.secret is not None,

689

options.approve,

690

options.deny))

691

692

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

693

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

694

if options.verbose and has_actions(options):

695

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

696

if options.dump_json and (options.verbose

697

or has_actions(options)):

698

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

699

if options.all and not has_actions(options):

700

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

701

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

702

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

703

if options.remove:

704

options.remove = False

705

if has_actions(options) and not options.deny:

706

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

707

options.remove = True

708

709

710

def main():

711

parser = argparse.ArgumentParser()

712

713

add_command_line_options(parser)

714

715

options = parser.parse_args()

716

717

check_option_syntax(parser, options)

718

719

clientnames = options.client

720

721

if options.debug:

722

log.setLevel(logging.DEBUG)

723

724

try:

725

bus = dbus.SystemBus()

726

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

727

server_path)

728

mandos_dbus_objc = bus.get_object(busname, server_path)

729

except dbus.exceptions.DBusException:

730

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

731

sys.exit(1)

732

733

mandos_serv = dbus.Interface(mandos_dbus_objc,

734

dbus_interface=server_interface)

735

mandos_serv_object_manager = dbus.Interface(

736

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

737

738

# Filter out log message from dbus module

739

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

740

class NullFilter(logging.Filter):

741

def filter(self, record):

742

return False

743

dbus_filter = NullFilter()

744

try:

745

dbus_logger.addFilter(dbus_filter)

746

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

747

server_path, dbus.OBJECT_MANAGER_IFACE)

748

mandos_clients = {path: ifs_and_props[client_interface]

749

for path, ifs_and_props in

750

mandos_serv_object_manager

751

.GetManagedObjects().items()

752

if client_interface in ifs_and_props}

753

except dbus.exceptions.DBusException as e:

754

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

755

"\n%s", e)

756

sys.exit(1)

757

finally:

758

# restore dbus logger

759

dbus_logger.removeFilter(dbus_filter)

760

761

# Compile dict of (clients: properties) to process

762

clients = {}

763

764

if not clientnames:

765

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

766

for path, properties in mandos_clients.items()}

767

else:

768

for name in clientnames:

769

for path, client in mandos_clients.items():

770

if client["Name"] == name:

771

client_objc = bus.get_object(busname, path)

772

clients[client_objc] = client

773

break

774

else:

775

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

776

sys.exit(1)

777

778

# Run all commands on clients

779

commands = commands_from_options(options)

780

for command in commands:

781

command.run(mandos_serv, mandos_serv_object_manager, clients)

782

783

784

class Test_milliseconds_to_string(unittest.TestCase):

785

def test_all(self):

786

self.assertEqual(milliseconds_to_string(93785000),

787

"1T02:03:05")

788

def test_no_days(self):

789

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

790

def test_all_zero(self):

791

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

792

def test_no_fractional_seconds(self):

793

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

794

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

795

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

796

797

class Test_string_to_delta(unittest.TestCase):

798

def test_handles_basic_rfc3339(self):

799

self.assertEqual(string_to_delta("PT0S"),

800

datetime.timedelta())

801

self.assertEqual(string_to_delta("P0D"),

802

datetime.timedelta())

803

self.assertEqual(string_to_delta("PT1S"),

804

datetime.timedelta(0, 1))

805

self.assertEqual(string_to_delta("PT2H"),

806

datetime.timedelta(0, 7200))

807

def test_falls_back_to_pre_1_6_1_with_warning(self):

808

# assertLogs only exists in Python 3.4

809

if hasattr(self, "assertLogs"):

810

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

811

value = string_to_delta("2h")

812

else:

813

class WarningFilter(logging.Filter):

814

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

815

def filter(self, record):

816

is_warning = record.levelno >= logging.WARNING

817

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

818

False)

819

return not is_warning

820

warning_filter = WarningFilter()

821

log.addFilter(warning_filter)

822

try:

823

value = string_to_delta("2h")

824

finally:

825

log.removeFilter(warning_filter)

826

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

827

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

828

829

830

class TestCmd(unittest.TestCase):

831

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

832

def setUp(self):

833

testcase = self

834

class MockClient(object):

835

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

836

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

837

self.attributes = attributes

838

self.attributes["Name"] = name

839

self.calls = []

840

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

841

testcase.assertEqual(interface, client_interface)

842

testcase.assertEqual(dbus_interface,

843

dbus.PROPERTIES_IFACE)

844

self.attributes[property] = value

845

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

846

testcase.assertEqual(interface, client_interface)

847

testcase.assertEqual(dbus_interface,

848

dbus.PROPERTIES_IFACE)

849

return self.attributes[property]

850

def Approve(self, approve, dbus_interface):

851

testcase.assertEqual(dbus_interface, client_interface)

852

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

853

dbus_interface)))

854

self.client = MockClient(

855

"foo",

856

KeyID=("92ed150794387c03ce684574b1139a65"

857

"94a34f895daaaf09fd8ea90a27cddb12"),

858

Secret=b"secret",

859

Host="foo.example.org",

860

Enabled=dbus.Boolean(True),

861

Timeout=300000,

862

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

863

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

864

Interval=120000,

865

Fingerprint=("778827225BA7DE539C5A"

866

"7CFA59CFF7CDBD9A5920"),

867

CheckerRunning=dbus.Boolean(False),

868

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

869

ApprovalPending=dbus.Boolean(False),

870

ApprovedByDefault=dbus.Boolean(True),

871

LastApprovalRequest="",

872

ApprovalDelay=0,

873

ApprovalDuration=1000,

874

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

875

ExtendedTimeout=900000,

876

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

877

LastCheckerStatus=0)

878

self.other_client = MockClient(

879

"barbar",

880

KeyID=("0558568eedd67d622f5c83b35a115f79"

881

"6ab612cff5ad227247e46c2b020f441c"),

882

Secret=b"secretbar",

883

Host="192.0.2.3",

884

Enabled=dbus.Boolean(True),

885

Timeout=300000,

886

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

887

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

888

Interval=120000,

889

Fingerprint=("3E393AEAEFB84C7E89E2"

890

"F547B3A107558FCA3A27"),

891

CheckerRunning=dbus.Boolean(True),

892

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

893

ApprovalPending=dbus.Boolean(False),

894

ApprovedByDefault=dbus.Boolean(False),

895

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

896

ApprovalDelay=30000,

897

ApprovalDuration=1000,

898

Checker=":",

899

ExtendedTimeout=900000,

900

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

901

LastCheckerStatus=-2)

902

self.clients = collections.OrderedDict(

903

[

904

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

905

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

906

])

907

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

908

909

class TestPrintTableCmd(TestCmd):

910

def test_normal(self):

911

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

912

expected_output = """

913

Name Enabled Timeout Last Successful Check

914

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

915

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

916

"""[1:-1]

917

self.assertEqual(output, expected_output)

918

def test_verbose(self):

919

output = PrintTableCmd(verbose=True).output(

920

None, self.clients.values())

921

expected_output = """

922

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

923

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

924

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

925

"""[1:-1]

926

self.assertEqual(output, expected_output)

927

def test_one_client(self):

928

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

929

expected_output = """

930

Name Enabled Timeout Last Successful Check

931

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

932

"""[1:-1]

933

self.assertEqual(output, expected_output)

934

935

class TestDumpJSONCmd(TestCmd):

936

def setUp(self):

937

self.expected_json = {

938

"foo": {

939

"Name": "foo",

940

"KeyID": ("92ed150794387c03ce684574b1139a65"

941

"94a34f895daaaf09fd8ea90a27cddb12"),

942

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

943

"Enabled": True,

944

"Timeout": 300000,

945

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

946

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

947

"Interval": 120000,

948

"Fingerprint": ("778827225BA7DE539C5A"

949

"7CFA59CFF7CDBD9A5920"),

950

"CheckerRunning": False,

951

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

952

"ApprovalPending": False,

953

"ApprovedByDefault": True,

954

"LastApprovalRequest": "",

955

"ApprovalDelay": 0,

956

"ApprovalDuration": 1000,

957

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

958

"ExtendedTimeout": 900000,

959

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

960

"LastCheckerStatus": 0,

961

},

962

"barbar": {

963

"Name": "barbar",

964

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

965

"6ab612cff5ad227247e46c2b020f441c"),

966

"Host": "192.0.2.3",

967

"Enabled": True,

968

"Timeout": 300000,

969

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

970

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

971

"Interval": 120000,

972

"Fingerprint": ("3E393AEAEFB84C7E89E2"

973

"F547B3A107558FCA3A27"),

974

"CheckerRunning": True,

975

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

976

"ApprovalPending": False,

977

"ApprovedByDefault": False,

978

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

979

"ApprovalDelay": 30000,

980

"ApprovalDuration": 1000,

981

"Checker": ":",

982

"ExtendedTimeout": 900000,

983

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

984

"LastCheckerStatus": -2,

985

},

986

}

987

return super(TestDumpJSONCmd, self).setUp()

988

def test_normal(self):

989

json_data = json.loads(DumpJSONCmd().output(None,

990

self.clients))

991

self.assertDictEqual(json_data, self.expected_json)

992

def test_one_client(self):

993

clients = self.one_client

994

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

995

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

996

self.assertDictEqual(json_data, expected_json)

997

998

class TestIsEnabledCmd(TestCmd):

999

def test_is_enabled(self):

1000

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

1001

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

1002

def test_is_enabled_run_exits_successfully(self):

1003

with self.assertRaises(SystemExit) as e:

1004

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

1005

if e.exception.code is not None:

1006

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

1007

else:

1008

self.assertIsNone(e.exception.code)

1009

def test_is_enabled_run_exits_with_failure(self):

1010

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

1011

with self.assertRaises(SystemExit) as e:

1012

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

1013

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

1014

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

1015

else:

1016

self.assertIsNotNone(e.exception.code)

1017

1018

class TestRemoveCmd(TestCmd):

1019

def test_remove(self):

1020

class MockMandos(object):

1021

def __init__(self):

1022

self.calls = []

1023

def RemoveClient(self, dbus_path):

1024

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

1025

mandos = MockMandos()

1026

super(TestRemoveCmd, self).setUp()

1027

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

1028

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

1029

for client in self.clients:

1030

self.assertIn(("RemoveClient",

1031

(client.__dbus_object_path__,)),

1032

mandos.calls)

1033

1034

class TestApproveCmd(TestCmd):

1035

def test_approve(self):

1036

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

1037

for client in self.clients:

1038

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

1039

client.calls)

1040

1041

class TestDenyCmd(TestCmd):

1042

def test_deny(self):

1043

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

1044

for client in self.clients:

1045

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

1046

client.calls)

1047

1048

class TestEnableCmd(TestCmd):

1049

def test_enable(self):

1050

for client in self.clients:

1051

client.attributes["Enabled"] = False

1052

1053

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

1054

1055

for client in self.clients:

1056

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

1057

1058

class TestDisableCmd(TestCmd):

1059

def test_disable(self):

1060

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

1061

1062

for client in self.clients:

1063

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

1064

1065

class Unique(object):

1066

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

1067

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

1068

1069

class TestPropertyCmd(TestCmd):

1070

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

1071

def runTest(self):

1072

if not hasattr(self, "command"):

1073

return

1074

values_to_get = getattr(self, "values_to_get",

1075

self.values_to_set)

1076

for value_to_set, value_to_get in zip(self.values_to_set,

1077

values_to_get):

1078

for client in self.clients:

1079

old_value = client.attributes[self.property]

1080

self.assertNotIsInstance(old_value, Unique)

1081

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

1082

self.run_command(value_to_set, self.clients)

1083

for client in self.clients:

1084

value = client.attributes[self.property]

1085

self.assertNotIsInstance(value, Unique)

1086

self.assertEqual(value, value_to_get)

1087

def run_command(self, value, clients):

1088

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

1089

1090

class TestBumpTimeoutCmd(TestPropertyCmd):

1091

command = BumpTimeoutCmd

1092

property = "LastCheckedOK"

1093

values_to_set = [""]

1094

1095

class TestStartCheckerCmd(TestPropertyCmd):

1096

command = StartCheckerCmd

1097

property = "CheckerRunning"

1098

values_to_set = [dbus.Boolean(True)]

1099

1100

class TestStopCheckerCmd(TestPropertyCmd):

1101

command = StopCheckerCmd

1102

property = "CheckerRunning"

1103

values_to_set = [dbus.Boolean(False)]

1104

1105

class TestApproveByDefaultCmd(TestPropertyCmd):

1106

command = ApproveByDefaultCmd

1107

property = "ApprovedByDefault"

1108

values_to_set = [dbus.Boolean(True)]

1109

1110

class TestDenyByDefaultCmd(TestPropertyCmd):

1111

command = DenyByDefaultCmd

1112

property = "ApprovedByDefault"

1113

values_to_set = [dbus.Boolean(False)]

1114

1115

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1116

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

1117

ValueArgumentMixIn"""

1118

def runTest(self):

1119

if type(self) is TestValueArgumentPropertyCmd:

1120

return

1121

return super(TestValueArgumentPropertyCmd, self).runTest()

1122

def run_command(self, value, clients):

1123

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

1124

1125

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1126

command = SetCheckerCmd

1127

property = "Checker"

1128

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

1129

1130

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1131

command = SetHostCmd

1132

property = "Host"

1133

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

1134

1135

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1136

command = SetSecretCmd

1137

property = "Secret"

1138

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

1139

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

1140

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

1141

1142

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1143

command = SetTimeoutCmd

1144

property = "Timeout"

1145

values_to_set = [datetime.timedelta(),

1146

datetime.timedelta(minutes=5),

1147

datetime.timedelta(seconds=1),

1148

datetime.timedelta(weeks=1),

1149

datetime.timedelta(weeks=52)]

1150

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

1151

1152

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1153

command = SetExtendedTimeoutCmd

1154

property = "ExtendedTimeout"

1155

values_to_set = [datetime.timedelta(),

1156

datetime.timedelta(minutes=5),

1157

datetime.timedelta(seconds=1),

1158

datetime.timedelta(weeks=1),

1159

datetime.timedelta(weeks=52)]

1160

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

1161

1162

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1163

command = SetIntervalCmd

1164

property = "Interval"

1165

values_to_set = [datetime.timedelta(),

1166

datetime.timedelta(minutes=5),

1167

datetime.timedelta(seconds=1),

1168

datetime.timedelta(weeks=1),

1169

datetime.timedelta(weeks=52)]

1170

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

1171

1172

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1173

command = SetApprovalDelayCmd

1174

property = "ApprovalDelay"

1175

values_to_set = [datetime.timedelta(),

1176

datetime.timedelta(minutes=5),

1177

datetime.timedelta(seconds=1),

1178

datetime.timedelta(weeks=1),

1179

datetime.timedelta(weeks=52)]

1180

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

1181

1182

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1183

command = SetApprovalDurationCmd

1184

property = "ApprovalDuration"

1185

values_to_set = [datetime.timedelta(),

1186

datetime.timedelta(minutes=5),

1187

datetime.timedelta(seconds=1),

1188

datetime.timedelta(weeks=1),

1189

datetime.timedelta(weeks=52)]

1190

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

1191

1192

class Test_command_from_options(unittest.TestCase):

1193

def setUp(self):

1194

self.parser = argparse.ArgumentParser()

1195

add_command_line_options(self.parser)

1196

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

1197

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

1198

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

1199

options = self.parser.parse_args(args)

1200

check_option_syntax(self.parser, options)

1201

commands = commands_from_options(options)

1202

self.assertEqual(len(commands), 1)

1203

command = commands[0]

1204

self.assertIsInstance(command, command_cls)

1205

for key, value in cmd_attrs.items():

1206

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

1207

def test_print_table(self):

1208

self.assert_command_from_args([], PrintTableCmd,

1209

verbose=False)

1210

1211

def test_print_table_verbose(self):

1212

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

1213

verbose=True)

1214

1215

def test_print_table_verbose_short(self):

1216

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

1217

verbose=True)

1218

1219

def test_enable(self):

1220

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

1221

1222

def test_enable_short(self):

1223

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

1224

1225

def test_disable(self):

1226

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

1227

DisableCmd)

1228

1229

def test_disable_short(self):

1230

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

1231

1232

def test_bump_timeout(self):

1233

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

1234

BumpTimeoutCmd)

1235

1236

def test_bump_timeout_short(self):

1237

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

1238

1239

def test_start_checker(self):

1240

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

1241

StartCheckerCmd)

1242

1243

def test_stop_checker(self):

1244

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

1245

StopCheckerCmd)

1246

1247

def test_remove(self):

1248

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

1249

RemoveCmd)

1250

1251

def test_remove_short(self):

1252

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

1253

1254

def test_checker(self):

1255

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

1256

SetCheckerCmd, value_to_set=":")

1257

1258

def test_checker_empty(self):

1259

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

1260

SetCheckerCmd, value_to_set="")

1261

1262

def test_checker_short(self):

1263

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

1264

SetCheckerCmd, value_to_set=":")

1265

1266

def test_timeout(self):

1267

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

1268

SetTimeoutCmd,

1269

value_to_set=300000)

1270

1271

def test_timeout_short(self):

1272

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

1273

SetTimeoutCmd,

1274

value_to_set=300000)

1275

1276

def test_extended_timeout(self):

1277

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

1278

"foo"],

1279

SetExtendedTimeoutCmd,

1280

value_to_set=900000)

1281

1282

def test_interval(self):

1283

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

1284

SetIntervalCmd,

1285

value_to_set=120000)

1286

1287

def test_interval_short(self):

1288

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

1289

SetIntervalCmd,

1290

value_to_set=120000)

1291

1292

def test_approve_by_default(self):

1293

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

1294

ApproveByDefaultCmd)

1295

1296

def test_deny_by_default(self):

1297

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

1298

DenyByDefaultCmd)

1299

1300

def test_approval_delay(self):

1301

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

1302

"foo"], SetApprovalDelayCmd,

1303

value_to_set=30000)

1304

1305

def test_approval_duration(self):

1306

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

1307

"foo"], SetApprovalDurationCmd,

1308

value_to_set=1000)

1309

1310

def test_host(self):

1311

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

1312

"foo"], SetHostCmd,

1313

value_to_set="foo.example.org")

1314

1315

def test_host_short(self):

1316

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

1317

"foo"], SetHostCmd,

1318

value_to_set="foo.example.org")

1319

1320

def test_secret_devnull(self):

1321

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

1322

"foo"], SetSecretCmd,

1323

value_to_set=b"")

1324

1325

def test_secret_tempfile(self):

1326

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

1327

value = b"secret\0xyzzy\nbar"

1328

f.write(value)

1329

f.seek(0)

1330

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

1331

"foo"], SetSecretCmd,

1332

value_to_set=value)

1333

1334

def test_secret_devnull_short(self):

1335

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

1336

SetSecretCmd, value_to_set=b"")

1337

1338

def test_secret_tempfile_short(self):

1339

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

1340

value = b"secret\0xyzzy\nbar"

1341

f.write(value)

1342

f.seek(0)

1343

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

1344

SetSecretCmd,

1345

value_to_set=value)

1346

1347

def test_approve(self):

1348

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

1349

ApproveCmd)

1350

1351

def test_approve_short(self):

1352

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

1353

1354

def test_deny(self):

1355

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

1356

1357

def test_deny_short(self):

1358

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

1359

1360

def test_dump_json(self):

1361

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

1362

1363

def test_is_enabled(self):

1364

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

1365

IsEnabledCmd)

1366

1367

def test_is_enabled_short(self):

1368

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

1369

1370

def test_deny_before_remove(self):

1371

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

1372

check_option_syntax(self.parser, options)

1373

commands = commands_from_options(options)

1374

self.assertEqual(len(commands), 2)

1375

self.assertIsInstance(commands[0], DenyCmd)

1376

self.assertIsInstance(commands[1], RemoveCmd)

1377

1378

def test_deny_before_remove_reversed(self):

1379

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

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

1387

class Test_check_option_syntax(unittest.TestCase):

1388

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

1389

# check_option_syntax()

1390

actions = {

1391

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

1392

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

1393

# never used

1394

"enable": True,

1395

"disable": True,

1396

"bump_timeout": True,

1397

"start_checker": True,

1398

"stop_checker": True,

1399

"is_enabled": True,

1400

"remove": True,

1401

"checker": "x",

1402

"timeout": datetime.timedelta(),

1403

"extended_timeout": datetime.timedelta(),

1404

"interval": datetime.timedelta(),

1405

"approved_by_default": True,

1406

"approval_delay": datetime.timedelta(),

1407

"approval_duration": datetime.timedelta(),

1408

"host": "x",

1409

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

1410

"approve": True,

1411

"deny": True,

1412

}

1413

1414

def setUp(self):

1415

self.parser = argparse.ArgumentParser()

1416

add_command_line_options(self.parser)

1417

1418

@contextlib.contextmanager

1419

def assertParseError(self):

1420

with self.assertRaises(SystemExit) as e:

1421

with self.temporarily_suppress_stderr():

1422

yield

1423

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

1424

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

1425

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

1426

1427

@staticmethod

1428

@contextlib.contextmanager

1429

def temporarily_suppress_stderr():

1430

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

1431

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

1432

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

1433

os.close(null)

1434

try:

1435

yield

1436

finally:

1437

# restore stderr

1438

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

1439

os.close(stderrcopy)

1440

1441

def check_option_syntax(self, options):

1442

check_option_syntax(self.parser, options)

1443

1444

def test_actions_requires_client_or_all(self):

1445

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

1446

options = self.parser.parse_args()

1447

setattr(options, action, value)

1448

with self.assertParseError():

1449

self.check_option_syntax(options)

1450

1451

def test_actions_conflicts_with_verbose(self):

1452

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

1453

options = self.parser.parse_args()

1454

setattr(options, action, value)

1455

options.verbose = True

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_dump_json_conflicts_with_verbose(self):

1460

options = self.parser.parse_args()

1461

options.dump_json = True

1462

options.verbose = True

1463

with self.assertParseError():

1464

self.check_option_syntax(options)

1465

1466

def test_dump_json_conflicts_with_action(self):

1467

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

1468

options = self.parser.parse_args()

1469

setattr(options, action, value)

1470

options.dump_json = True

1471

with self.assertParseError():

1472

self.check_option_syntax(options)

1473

1474

def test_all_can_not_be_alone(self):

1475

options = self.parser.parse_args()

1476

options.all = True

1477

with self.assertParseError():

1478

self.check_option_syntax(options)

1479

1480

def test_all_is_ok_with_any_action(self):

1481

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

1482

options = self.parser.parse_args()

1483

setattr(options, action, value)

1484

options.all = True

1485

self.check_option_syntax(options)

1486

1487

def test_is_enabled_fails_without_client(self):

1488

options = self.parser.parse_args()

1489

options.is_enabled = True

1490

with self.assertParseError():

1491

self.check_option_syntax(options)

1492

1493

def test_is_enabled_works_with_one_client(self):

1494

options = self.parser.parse_args()

1495

options.is_enabled = True

1496

options.client = ["foo"]

1497

self.check_option_syntax(options)

1498

1499

def test_is_enabled_fails_with_two_clients(self):

1500

options = self.parser.parse_args()

1501

options.is_enabled = True

1502

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

1503

with self.assertParseError():

1504

self.check_option_syntax(options)

1505

1506

def test_remove_can_only_be_combined_with_action_deny(self):

1507

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

1508

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

1509

continue

1510

options = self.parser.parse_args()

1511

setattr(options, action, value)

1512

options.all = True

1513

options.remove = True

1514

with self.assertParseError():

1515

self.check_option_syntax(options)

1516

1517

1518

1519

def should_only_run_tests():

1520

parser = argparse.ArgumentParser(add_help=False)

1521

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

1522

args, unknown_args = parser.parse_known_args()

1523

run_tests = args.check

1524

if run_tests:

1525

# Remove --check argument from sys.argv

1526

sys.argv[1:] = unknown_args

1527

return run_tests

1528

1529

# Add all tests from doctest strings

1530

def load_tests(loader, tests, none):

1531

import doctest

1532

tests.addTests(doctest.DocTestSuite())

1533

return tests

1534

1535

if __name__ == "__main__":

1536

if should_only_run_tests():

1537

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

1538

unittest.main()

1539

else:

1540

main()