/mandos/trunk : revision 24.1.51

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

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

81

82

def milliseconds_to_string(ms):

83

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

84

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

85

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

86

hours=td.seconds // 3600,

87

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

88

seconds=td.seconds % 60))

89

90

91

def rfc3339_duration_to_delta(duration):

92

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

93

94

>>> rfc3339_duration_to_delta("P7D")

95

datetime.timedelta(7)

96

>>> rfc3339_duration_to_delta("PT60S")

97

datetime.timedelta(0, 60)

98

>>> rfc3339_duration_to_delta("PT60M")

99

datetime.timedelta(0, 3600)

100

>>> rfc3339_duration_to_delta("P60M")

101

datetime.timedelta(1680)

102

>>> rfc3339_duration_to_delta("PT24H")

103

datetime.timedelta(1)

104

>>> rfc3339_duration_to_delta("P1W")

105

datetime.timedelta(7)

106

>>> rfc3339_duration_to_delta("PT5M30S")

107

datetime.timedelta(0, 330)

108

>>> rfc3339_duration_to_delta("P1DT3M20S")

109

datetime.timedelta(1, 200)

110

>>> # Can not be empty:

111

>>> rfc3339_duration_to_delta("")

112

Traceback (most recent call last):

113

...

114

ValueError: Invalid RFC 3339 duration: u''

115

>>> # Must start with "P":

116

>>> rfc3339_duration_to_delta("1D")

117

Traceback (most recent call last):

118

...

119

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

120

>>> # Must use correct order

121

>>> rfc3339_duration_to_delta("PT1S2M")

122

Traceback (most recent call last):

123

...

124

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

125

>>> # Time needs time marker

126

>>> rfc3339_duration_to_delta("P1H2S")

127

Traceback (most recent call last):

128

...

129

ValueError: Invalid RFC 3339 duration: u'P1H2S'

130

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

131

>>> rfc3339_duration_to_delta("P1D2W")

132

Traceback (most recent call last):

133

...

134

ValueError: Invalid RFC 3339 duration: u'P1D2W'

135

>>> rfc3339_duration_to_delta("P2W2H")

136

Traceback (most recent call last):

137

...

138

ValueError: Invalid RFC 3339 duration: u'P2W2H'

139

"""

140

141

# Parsing an RFC 3339 duration with regular expressions is not

142

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

143

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

144

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

145

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

146

# avoid excessive use of external libraries.

147

148

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

149

Token = collections.namedtuple("Token", (

150

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

151

# a "group" containing digits

152

"value", # datetime.timedelta or None

153

"followers")) # Tokens valid after this token

154

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

155

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

156

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

157

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

158

datetime.timedelta(seconds=1),

159

frozenset((token_end, )))

160

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

161

datetime.timedelta(minutes=1),

162

frozenset((token_second, token_end)))

163

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

164

datetime.timedelta(hours=1),

165

frozenset((token_minute, token_end)))

166

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

167

None,

168

frozenset((token_hour, token_minute,

169

token_second)))

170

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

171

datetime.timedelta(days=1),

172

frozenset((token_time, token_end)))

173

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

174

datetime.timedelta(weeks=4),

175

frozenset((token_day, token_end)))

176

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

177

datetime.timedelta(weeks=52),

178

frozenset((token_month, token_end)))

179

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

180

datetime.timedelta(weeks=1),

181

frozenset((token_end, )))

182

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

183

frozenset((token_year, token_month,

184

token_day, token_time,

185

token_week)))

186

# Define starting values:

187

# Value so far

188

value = datetime.timedelta()

189

found_token = None

190

# Following valid tokens

191

followers = frozenset((token_duration, ))

192

# String left to parse

193

s = duration

194

# Loop until end token is found

195

while found_token is not token_end:

196

# Search for any currently valid tokens

197

for token in followers:

198

match = token.regexp.match(s)

199

if match is not None:

200

# Token found

201

if token.value is not None:

202

# Value found, parse digits

203

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

204

# Add to value so far

205

value += factor * token.value

206

# Strip token from string

207

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

208

# Go to found token

209

found_token = token

210

# Set valid next tokens

211

followers = found_token.followers

212

break

213

else:

214

# No currently valid tokens were found

215

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

216

.format(duration))

217

# End token found

218

return value

219

220

221

def string_to_delta(interval):

222

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

223

224

try:

225

return rfc3339_duration_to_delta(interval)

226

except ValueError as e:

227

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

228

' '.join(e.args))

229

return parse_pre_1_6_1_interval(interval)

230

231

232

def parse_pre_1_6_1_interval(interval):

233

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

234

and return a datetime.timedelta

235

236

>>> parse_pre_1_6_1_interval('7d')

237

datetime.timedelta(7)

238

>>> parse_pre_1_6_1_interval('60s')

239

datetime.timedelta(0, 60)

240

>>> parse_pre_1_6_1_interval('60m')

241

datetime.timedelta(0, 3600)

242

>>> parse_pre_1_6_1_interval('24h')

243

datetime.timedelta(1)

244

>>> parse_pre_1_6_1_interval('1w')

245

datetime.timedelta(7)

246

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

247

datetime.timedelta(0, 330)

248

>>> parse_pre_1_6_1_interval('')

249

datetime.timedelta(0)

250

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

251

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

252

datetime.timedelta(2, 480, 18000)

253

254

"""

255

256

value = datetime.timedelta(0)

257

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

258

259

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

260

if suffix == "d":

261

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

262

elif suffix == "s":

263

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

264

elif suffix == "m":

265

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

266

elif suffix == "h":

267

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

268

elif suffix == "w":

269

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

270

elif suffix == "":

271

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

272

return value

273

274

275

## Classes for commands.

276

277

# Abstract classes first

278

class Command(object):

279

"""Abstract class for commands"""

280

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

281

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

282

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

283

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

284

self.mandos = mandos

285

for clientpath, properties in clients.items():

286

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

287

dbus_busname, str(clientpath))

288

client = bus.get_object(dbus_busname, clientpath)

289

self.run_on_one_client(client, properties)

290

291

class PrintCmd(Command):

292

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

293

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

294

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

295

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

296

"ApprovalPending", "ApprovedByDefault",

297

"LastApprovalRequest", "ApprovalDelay",

298

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

299

"Expires", "LastCheckerStatus")

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

304

305

class PropertyCmd(Command):

306

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

307

def run_on_one_client(self, client, properties):

308

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

309

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

310

client.__dbus_object_path__,

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

313

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

314

else bool(self.value_to_set))

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

317

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

322

class PropertyValueCmd(PropertyCmd):

323

"""Abstract class for PropertyCmd recieving a value as argument"""

324

def __init__(self, value):

325

self.value_to_set = value

326

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

"""Abstract class for PropertyValueCmd taking a value argument as

329

a datetime.timedelta() but should store it as milliseconds."""

330

@property

331

def value_to_set(self):

332

return self._vts

333

@value_to_set.setter

334

def value_to_set(self, value):

335

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

336

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

337

338

# Actual (non-abstract) command classes

339

340

class PrintTableCmd(PrintCmd):

341

def __init__(self, verbose=False):

342

self.verbose = verbose

343

344

def output(self, clients):

345

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

346

"LastCheckedOK")

347

keywords = default_keywords

348

if self.verbose:

349

keywords = self.all_keywords

350

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

351

352

class TableOfClients(object):

353

tableheaders = {

354

"Name": "Name",

355

"Enabled": "Enabled",

356

"Timeout": "Timeout",

357

"LastCheckedOK": "Last Successful Check",

358

"LastApprovalRequest": "Last Approval Request",

359

"Created": "Created",

360

"Interval": "Interval",

361

"Host": "Host",

362

"Fingerprint": "Fingerprint",

363

"KeyID": "Key ID",

364

"CheckerRunning": "Check Is Running",

365

"LastEnabled": "Last Enabled",

366

"ApprovalPending": "Approval Is Pending",

367

"ApprovedByDefault": "Approved By Default",

368

"ApprovalDelay": "Approval Delay",

369

"ApprovalDuration": "Approval Duration",

370

"Checker": "Checker",

371

"ExtendedTimeout": "Extended Timeout",

372

"Expires": "Expires",

373

"LastCheckerStatus": "Last Checker Status",

374

}

375

376

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

377

self.clients = clients

378

self.keywords = keywords

379

if tableheaders is not None:

380

self.tableheaders = tableheaders

381

382

def __str__(self):

383

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

384

385

if sys.version_info.major == 2:

386

__unicode__ = __str__

387

def __str__(self):

388

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

389

390

def rows(self):

391

format_string = self.row_formatting_string()

392

rows = [self.header_line(format_string)]

393

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

394

for client in self.clients)

395

return rows

396

397

def row_formatting_string(self):

398

"Format string used to format table rows"

399

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

400

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

401

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

402

for client in self.clients)),

403

key=key)

404

for key in self.keywords)

405

406

def string_from_client(self, client, key):

407

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

408

409

@staticmethod

410

def valuetostring(value, keyword):

411

if isinstance(value, dbus.Boolean):

412

return "Yes" if value else "No"

413

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

414

"ApprovalDuration", "ExtendedTimeout"):

415

return milliseconds_to_string(value)

416

return str(value)

417

418

def header_line(self, format_string):

419

return format_string.format(**self.tableheaders)

420

421

def client_line(self, client, format_string):

422

return format_string.format(

423

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

424

for key in self.keywords})

425

426

427

428

class DumpJSONCmd(PrintCmd):

429

def output(self, clients):

430

data = {client["Name"]:

431

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

432

for key in self.all_keywords}

433

for client in clients.values()}

434

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

435

@staticmethod

436

def dbus_boolean_to_bool(value):

437

if isinstance(value, dbus.Boolean):

438

value = bool(value)

439

return value

440

441

class IsEnabledCmd(Command):

442

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

443

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

444

if self.is_enabled(client, properties):

445

sys.exit(0)

446

sys.exit(1)

447

def is_enabled(self, client, properties):

448

return properties["Enabled"]

449

450

class RemoveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

server_dbus_path, server_dbus_interface,

454

str(client.__dbus_object_path__))

455

self.mandos.RemoveClient(client.__dbus_object_path__)

456

457

class ApproveCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_dbus_interface)

461

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

463

464

class DenyCmd(Command):

465

def run_on_one_client(self, client, properties):

466

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

467

client.__dbus_object_path__, client_dbus_interface)

468

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

470

471

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

473

value_to_set = dbus.Boolean(True)

474

475

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

477

value_to_set = dbus.Boolean(False)

478

479

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

481

value_to_set = ""

482

483

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

485

value_to_set = dbus.Boolean(True)

486

487

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

489

value_to_set = dbus.Boolean(False)

490

491

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

493

value_to_set = dbus.Boolean(True)

494

495

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

497

value_to_set = dbus.Boolean(False)

498

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

507

@property

508

def value_to_set(self):

509

return self._vts

510

@value_to_set.setter

511

def value_to_set(self, value):

512

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

513

self._vts = value.read()

514

value.close()

515

516

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "Timeout"

518

519

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "ExtendedTimeout"

521

522

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "Interval"

524

525

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

529

propname = "ApprovalDuration"

530

531

def add_command_line_options(parser):

532

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

533

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

534

help="show version number and exit")

535

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

536

help="Select all clients")

537

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

538

help="Print all fields")

539

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

540

help="Dump client data in JSON format")

541

enable_disable = parser.add_mutually_exclusive_group()

542

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

543

help="Enable client")

544

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

545

action="store_true",

546

help="disable client")

547

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

548

help="Bump timeout for client")

549

start_stop_checker = parser.add_mutually_exclusive_group()

550

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

551

action="store_true",

552

help="Start checker for client")

553

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

554

action="store_true",

555

help="Stop checker for client")

556

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

557

help="Check if client is enabled")

558

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

559

help="Remove client")

560

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

561

help="Set checker command for client")

562

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

563

help="Set timeout for client")

564

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

565

help="Set extended timeout for client")

566

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

567

help="Set checker interval for client")

568

approve_deny_default = parser.add_mutually_exclusive_group()

569

approve_deny_default.add_argument(

570

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

571

default=None, dest="approved_by_default",

572

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

573

approve_deny_default.add_argument(

574

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

575

dest="approved_by_default",

576

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

577

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

578

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

579

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

580

help="Set duration of one client approval")

581

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

582

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

583

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

584

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

585

approve_deny = parser.add_mutually_exclusive_group()

586

approve_deny.add_argument(

587

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

588

help="Approve any current client request")

589

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

590

help="Deny any current client request")

591

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

592

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

593

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

594

help="Run self-test")

595

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

596

597

598

def commands_from_options(options):

599

600

commands = []

601

602

if options.dump_json:

603

commands.append(DumpJSONCmd())

604

605

if options.enable:

606

commands.append(EnableCmd())

607

608

if options.disable:

609

commands.append(DisableCmd())

610

611

if options.bump_timeout:

612

commands.append(BumpTimeoutCmd())

613

614

if options.start_checker:

615

commands.append(StartCheckerCmd())

616

617

if options.stop_checker:

618

commands.append(StopCheckerCmd())

619

620

if options.is_enabled:

621

commands.append(IsEnabledCmd())

622

623

if options.checker is not None:

624

commands.append(SetCheckerCmd(options.checker))

625

626

if options.timeout is not None:

627

commands.append(SetTimeoutCmd(options.timeout))

628

629

if options.extended_timeout:

630

commands.append(

631

SetExtendedTimeoutCmd(options.extended_timeout))

632

633

if options.interval is not None:

634

commands.append(SetIntervalCmd(options.interval))

635

636

if options.approved_by_default is not None:

637

if options.approved_by_default:

638

commands.append(ApproveByDefaultCmd())

639

else:

640

commands.append(DenyByDefaultCmd())

641

642

if options.approval_delay is not None:

643

commands.append(SetApprovalDelayCmd(options.approval_delay))

644

645

if options.approval_duration is not None:

646

commands.append(

647

SetApprovalDurationCmd(options.approval_duration))

648

649

if options.host is not None:

650

commands.append(SetHostCmd(options.host))

651

652

if options.secret is not None:

653

commands.append(SetSecretCmd(options.secret))

654

655

if options.approve:

656

commands.append(ApproveCmd())

657

658

if options.deny:

659

commands.append(DenyCmd())

660

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

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

665

# optionally verbosely

666

if not commands:

667

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

668

669

return commands

670

671

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

676

def has_actions(options):

677

return any((options.enable,

678

options.disable,

679

options.bump_timeout,

680

options.start_checker,

681

options.stop_checker,

682

options.is_enabled,

683

options.remove,

684

options.checker is not None,

685

options.timeout is not None,

686

options.extended_timeout is not None,

687

options.interval is not None,

688

options.approved_by_default is not None,

689

options.approval_delay is not None,

690

options.approval_duration is not None,

691

options.host is not None,

692

options.secret is not None,

693

options.approve,

694

options.deny))

695

696

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

697

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

698

if options.verbose and has_actions(options):

699

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

700

if options.dump_json and (options.verbose

701

or has_actions(options)):

702

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

703

if options.all and not has_actions(options):

704

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

705

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

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

723

clientnames = options.client

724

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

try:

729

bus = dbus.SystemBus()

730

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

731

dbus_busname, server_dbus_path)

732

mandos_dbus_objc = bus.get_object(dbus_busname,

733

server_dbus_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_dbus_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()", dbus_busname,

752

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_dbus_interface]

754

for path, ifs_and_props in

755

mandos_serv_object_manager

756

.GetManagedObjects().items()

757

if client_dbus_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 = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

772

else:

773

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

777

break

778

else:

779

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

780

sys.exit(1)

781

782

# Run all commands on clients

783

commands = commands_from_options(options)

784

for command in commands:

785

command.run(clients, bus, mandos_serv)

786

787

788

class Test_milliseconds_to_string(unittest.TestCase):

789

def test_all(self):

790

self.assertEqual(milliseconds_to_string(93785000),

791

"1T02:03:05")

792

def test_no_days(self):

793

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

794

def test_all_zero(self):

795

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

796

def test_no_fractional_seconds(self):

797

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

798

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

799

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

800

801

class Test_string_to_delta(unittest.TestCase):

802

def test_handles_basic_rfc3339(self):

803

self.assertEqual(string_to_delta("PT0S"),

804

datetime.timedelta())

805

self.assertEqual(string_to_delta("P0D"),

806

datetime.timedelta())

807

self.assertEqual(string_to_delta("PT1S"),

808

datetime.timedelta(0, 1))

809

self.assertEqual(string_to_delta("PT2H"),

810

datetime.timedelta(0, 7200))

811

def test_falls_back_to_pre_1_6_1_with_warning(self):

812

# assertLogs only exists in Python 3.4

813

if hasattr(self, "assertLogs"):

814

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

815

value = string_to_delta("2h")

816

else:

817

class WarningFilter(logging.Filter):

818

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

819

def filter(self, record):

820

is_warning = record.levelno >= logging.WARNING

821

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

822

False)

823

return not is_warning

824

warning_filter = WarningFilter()

825

log.addFilter(warning_filter)

826

try:

827

value = string_to_delta("2h")

828

finally:

829

log.removeFilter(warning_filter)

830

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

831

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

832

833

834

class TestCmd(unittest.TestCase):

835

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

836

def setUp(self):

837

testcase = self

838

class MockClient(object):

839

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

840

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

841

self.attributes = attributes

842

self.attributes["Name"] = name

843

self.calls = []

844

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

845

testcase.assertEqual(interface, client_dbus_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_dbus_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

854

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

857

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

858

dbus_interface)))

859

self.client = MockClient(

860

"foo",

861

KeyID=("92ed150794387c03ce684574b1139a65"

862

"94a34f895daaaf09fd8ea90a27cddb12"),

863

Secret=b"secret",

864

Host="foo.example.org",

865

Enabled=dbus.Boolean(True),

866

Timeout=300000,

867

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

868

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

869

Interval=120000,

870

Fingerprint=("778827225BA7DE539C5A"

871

"7CFA59CFF7CDBD9A5920"),

872

CheckerRunning=dbus.Boolean(False),

873

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

874

ApprovalPending=dbus.Boolean(False),

875

ApprovedByDefault=dbus.Boolean(True),

876

LastApprovalRequest="",

877

ApprovalDelay=0,

878

ApprovalDuration=1000,

879

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

880

ExtendedTimeout=900000,

881

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

882

LastCheckerStatus=0)

883

self.other_client = MockClient(

884

"barbar",

885

KeyID=("0558568eedd67d622f5c83b35a115f79"

886

"6ab612cff5ad227247e46c2b020f441c"),

887

Secret=b"secretbar",

888

Host="192.0.2.3",

889

Enabled=dbus.Boolean(True),

890

Timeout=300000,

891

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

892

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

893

Interval=120000,

894

Fingerprint=("3E393AEAEFB84C7E89E2"

895

"F547B3A107558FCA3A27"),

896

CheckerRunning=dbus.Boolean(True),

897

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

898

ApprovalPending=dbus.Boolean(False),

899

ApprovedByDefault=dbus.Boolean(False),

900

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

901

ApprovalDelay=30000,

902

ApprovalDuration=1000,

903

Checker=":",

904

ExtendedTimeout=900000,

905

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

906

LastCheckerStatus=-2)

907

self.clients = collections.OrderedDict(

908

[

909

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

910

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

911

])

912

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

913

@property

914

def bus(self):

915

class Bus(object):

916

@staticmethod

917

def get_object(client_bus_name, path):

918

self.assertEqual(client_bus_name, dbus_busname)

919

return {

920

"/clients/foo": self.client,

921

"/clients/barbar": self.other_client,

922

}[path]

923

return Bus()

924

925

class TestPrintTableCmd(TestCmd):

926

def test_normal(self):

927

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

928

expected_output = "\n".join((

929

"Name Enabled Timeout Last Successful Check",

930

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

931

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

932

))

933

self.assertEqual(output, expected_output)

934

def test_verbose(self):

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

columns = (

938

(

939

"Name ",

940

"foo ",

941

"barbar ",

942

),(

943

"Enabled ",

944

"Yes ",

945

"Yes ",

946

),(

947

"Timeout ",

948

"00:05:00 ",

949

"00:05:00 ",

950

),(

951

"Last Successful Check ",

952

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

953

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

954

),(

955

"Created ",

956

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

957

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

958

),(

959

"Interval ",

960

"00:02:00 ",

961

"00:02:00 ",

962

),(

963

"Host ",

964

"foo.example.org ",

965

"192.0.2.3 ",

966

),(

967

("Key ID "

968

" "),

969

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

970

"ea90a27cddb12 "),

971

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

972

"46c2b020f441c "),

973

),(

974

"Fingerprint ",

975

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

976

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

977

),(

978

"Check Is Running ",

979

"No ",

980

"Yes ",

981

),(

982

"Last Enabled ",

983

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

984

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

985

),(

986

"Approval Is Pending ",

987

"No ",

988

"No ",

989

),(

990

"Approved By Default ",

991

"Yes ",

992

"No ",

993

),(

994

"Last Approval Request ",

995

" ",

996

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

997

),(

998

"Approval Delay ",

999

"00:00:00 ",

1000

"00:00:30 ",

1001

),(

1002

"Approval Duration ",

1003

"00:00:01 ",

1004

"00:00:01 ",

1005

),(

1006

"Checker ",

1007

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

1008

": ",

1009

),(

1010

"Extended Timeout ",

1011

"00:15:00 ",

1012

"00:15:00 ",

1013

),(

1014

"Expires ",

1015

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

1016

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

1017

),(

1018

"Last Checker Status",

1019

"0 ",

1020

"-2 ",

1021

)

1022

)

1023

num_lines = max(len(rows) for rows in columns)

1024

expected_output = "\n".join("".join(rows[line]

1025

for rows in columns)

1026

for line in range(num_lines))

1027

self.assertEqual(output, expected_output)

1028

def test_one_client(self):

1029

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

1030

expected_output = """

1031

Name Enabled Timeout Last Successful Check

1032

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

1033

"""[1:-1]

1034

self.assertEqual(output, expected_output)

1035

1036

class TestDumpJSONCmd(TestCmd):

1037

def setUp(self):

1038

self.expected_json = {

1039

"foo": {

1040

"Name": "foo",

1041

"KeyID": ("92ed150794387c03ce684574b1139a65"

1042

"94a34f895daaaf09fd8ea90a27cddb12"),

1043

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

1044

"Enabled": True,

1045

"Timeout": 300000,

1046

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

1047

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

1048

"Interval": 120000,

1049

"Fingerprint": ("778827225BA7DE539C5A"

1050

"7CFA59CFF7CDBD9A5920"),

1051

"CheckerRunning": False,

1052

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

1053

"ApprovalPending": False,

1054

"ApprovedByDefault": True,

1055

"LastApprovalRequest": "",

1056

"ApprovalDelay": 0,

1057

"ApprovalDuration": 1000,

1058

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

1059

"ExtendedTimeout": 900000,

1060

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

1061

"LastCheckerStatus": 0,

1062

},

1063

"barbar": {

1064

"Name": "barbar",

1065

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1066

"6ab612cff5ad227247e46c2b020f441c"),

1067

"Host": "192.0.2.3",

1068

"Enabled": True,

1069

"Timeout": 300000,

1070

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

1071

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

1072

"Interval": 120000,

1073

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1074

"F547B3A107558FCA3A27"),

1075

"CheckerRunning": True,

1076

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

1077

"ApprovalPending": False,

1078

"ApprovedByDefault": False,

1079

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

1080

"ApprovalDelay": 30000,

1081

"ApprovalDuration": 1000,

1082

"Checker": ":",

1083

"ExtendedTimeout": 900000,

1084

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

1085

"LastCheckerStatus": -2,

1086

},

1087

}

1088

return super(TestDumpJSONCmd, self).setUp()

1089

def test_normal(self):

1090

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

1091

self.assertDictEqual(json_data, self.expected_json)

1092

def test_one_client(self):

1093

clients = self.one_client

1094

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

1095

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

1096

self.assertDictEqual(json_data, expected_json)

1097

1098

class TestIsEnabledCmd(TestCmd):

1099

def test_is_enabled(self):

1100

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

1101

properties)

1102

for client, properties

1103

in self.clients.items()))

1104

def test_is_enabled_run_exits_successfully(self):

1105

with self.assertRaises(SystemExit) as e:

1106

IsEnabledCmd().run(self.one_client)

1107

if e.exception.code is not None:

1108

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

1109

else:

1110

self.assertIsNone(e.exception.code)

1111

def test_is_enabled_run_exits_with_failure(self):

1112

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

1113

with self.assertRaises(SystemExit) as e:

1114

IsEnabledCmd().run(self.one_client)

1115

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

1116

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

1117

else:

1118

self.assertIsNotNone(e.exception.code)

1119

1120

class TestRemoveCmd(TestCmd):

1121

def test_remove(self):

1122

class MockMandos(object):

1123

def __init__(self):

1124

self.calls = []

1125

def RemoveClient(self, dbus_path):

1126

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

1127

mandos = MockMandos()

1128

super(TestRemoveCmd, self).setUp()

1129

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

1130

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

1131

for clientpath in self.clients:

1132

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

1133

mandos.calls)

1134

1135

class TestApproveCmd(TestCmd):

1136

def test_approve(self):

1137

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

1138

for clientpath in self.clients:

1139

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

1140

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

1141

client.calls)

1142

1143

class TestDenyCmd(TestCmd):

1144

def test_deny(self):

1145

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

1146

for clientpath in self.clients:

1147

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

1148

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

1149

client.calls)

1150

1151

class TestEnableCmd(TestCmd):

1152

def test_enable(self):

1153

for clientpath in self.clients:

1154

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

1155

client.attributes["Enabled"] = False

1156

1157

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

1158

1159

for clientpath in self.clients:

1160

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

1161

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

1162

1163

class TestDisableCmd(TestCmd):

1164

def test_disable(self):

1165

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

1166

for clientpath in self.clients:

1167

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

1168

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

1169

1170

class Unique(object):

1171

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

1172

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

1173

1174

class TestPropertyCmd(TestCmd):

1175

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

1176

def runTest(self):

1177

if not hasattr(self, "command"):

1178

return

1179

values_to_get = getattr(self, "values_to_get",

1180

self.values_to_set)

1181

for value_to_set, value_to_get in zip(self.values_to_set,

1182

values_to_get):

1183

for clientpath in self.clients:

1184

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

1185

old_value = client.attributes[self.propname]

1186

self.assertNotIsInstance(old_value, Unique)

1187

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

1188

self.run_command(value_to_set, self.clients)

1189

for clientpath in self.clients:

1190

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

1191

value = client.attributes[self.propname]

1192

self.assertNotIsInstance(value, Unique)

1193

self.assertEqual(value, value_to_get)

1194

def run_command(self, value, clients):

1195

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

1196

1197

class TestBumpTimeoutCmd(TestPropertyCmd):

1198

command = BumpTimeoutCmd

1199

propname = "LastCheckedOK"

1200

values_to_set = [""]

1201

1202

class TestStartCheckerCmd(TestPropertyCmd):

1203

command = StartCheckerCmd

1204

propname = "CheckerRunning"

1205

values_to_set = [dbus.Boolean(True)]

1206

1207

class TestStopCheckerCmd(TestPropertyCmd):

1208

command = StopCheckerCmd

1209

propname = "CheckerRunning"

1210

values_to_set = [dbus.Boolean(False)]

1211

1212

class TestApproveByDefaultCmd(TestPropertyCmd):

1213

command = ApproveByDefaultCmd

1214

propname = "ApprovedByDefault"

1215

values_to_set = [dbus.Boolean(True)]

1216

1217

class TestDenyByDefaultCmd(TestPropertyCmd):

1218

command = DenyByDefaultCmd

1219

propname = "ApprovedByDefault"

1220

values_to_set = [dbus.Boolean(False)]

1221

1222

class TestPropertyValueCmd(TestPropertyCmd):

1223

"""Abstract class for tests of PropertyValueCmd classes"""

1224

def runTest(self):

1225

if type(self) is TestPropertyValueCmd:

1226

return

1227

return super(TestPropertyValueCmd, self).runTest()

1228

def run_command(self, value, clients):

1229

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

1230

1231

class TestSetCheckerCmd(TestPropertyValueCmd):

1232

command = SetCheckerCmd

1233

propname = "Checker"

1234

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

1235

1236

class TestSetHostCmd(TestPropertyValueCmd):

1237

command = SetHostCmd

1238

propname = "Host"

1239

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

1240

1241

class TestSetSecretCmd(TestPropertyValueCmd):

1242

command = SetSecretCmd

1243

propname = "Secret"

1244

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

1245

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

1246

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

1247

1248

class TestSetTimeoutCmd(TestPropertyValueCmd):

1249

command = SetTimeoutCmd

1250

propname = "Timeout"

1251

values_to_set = [datetime.timedelta(),

1252

datetime.timedelta(minutes=5),

1253

datetime.timedelta(seconds=1),

1254

datetime.timedelta(weeks=1),

1255

datetime.timedelta(weeks=52)]

1256

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

1257

1258

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1259

command = SetExtendedTimeoutCmd

1260

propname = "ExtendedTimeout"

1261

values_to_set = [datetime.timedelta(),

1262

datetime.timedelta(minutes=5),

1263

datetime.timedelta(seconds=1),

1264

datetime.timedelta(weeks=1),

1265

datetime.timedelta(weeks=52)]

1266

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

1267

1268

class TestSetIntervalCmd(TestPropertyValueCmd):

1269

command = SetIntervalCmd

1270

propname = "Interval"

1271

values_to_set = [datetime.timedelta(),

1272

datetime.timedelta(minutes=5),

1273

datetime.timedelta(seconds=1),

1274

datetime.timedelta(weeks=1),

1275

datetime.timedelta(weeks=52)]

1276

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

1277

1278

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1279

command = SetApprovalDelayCmd

1280

propname = "ApprovalDelay"

1281

values_to_set = [datetime.timedelta(),

1282

datetime.timedelta(minutes=5),

1283

datetime.timedelta(seconds=1),

1284

datetime.timedelta(weeks=1),

1285

datetime.timedelta(weeks=52)]

1286

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

1287

1288

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1289

command = SetApprovalDurationCmd

1290

propname = "ApprovalDuration"

1291

values_to_set = [datetime.timedelta(),

1292

datetime.timedelta(minutes=5),

1293

datetime.timedelta(seconds=1),

1294

datetime.timedelta(weeks=1),

1295

datetime.timedelta(weeks=52)]

1296

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

1297

1298

class Test_command_from_options(unittest.TestCase):

1299

def setUp(self):

1300

self.parser = argparse.ArgumentParser()

1301

add_command_line_options(self.parser)

1302

def assert_command_from_args(self, args, command_cls,

1303

**cmd_attrs):

1304

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

1305

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

1306

options = self.parser.parse_args(args)

1307

check_option_syntax(self.parser, options)

1308

commands = commands_from_options(options)

1309

self.assertEqual(len(commands), 1)

1310

command = commands[0]

1311

self.assertIsInstance(command, command_cls)

1312

for key, value in cmd_attrs.items():

1313

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

1314

def test_print_table(self):

1315

self.assert_command_from_args([], PrintTableCmd,

1316

verbose=False)

1317

1318

def test_print_table_verbose(self):

1319

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

1320

verbose=True)

1321

1322

def test_print_table_verbose_short(self):

1323

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

1324

verbose=True)

1325

1326

def test_enable(self):

1327

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

1328

1329

def test_enable_short(self):

1330

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

1331

1332

def test_disable(self):

1333

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

1334

DisableCmd)

1335

1336

def test_disable_short(self):

1337

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

1338

1339

def test_bump_timeout(self):

1340

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

1341

BumpTimeoutCmd)

1342

1343

def test_bump_timeout_short(self):

1344

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

1345

1346

def test_start_checker(self):

1347

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

1348

StartCheckerCmd)

1349

1350

def test_stop_checker(self):

1351

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

1352

StopCheckerCmd)

1353

1354

def test_remove(self):

1355

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

1356

RemoveCmd)

1357

1358

def test_remove_short(self):

1359

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

1360

1361

def test_checker(self):

1362

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

1363

SetCheckerCmd, value_to_set=":")

1364

1365

def test_checker_empty(self):

1366

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

1367

SetCheckerCmd, value_to_set="")

1368

1369

def test_checker_short(self):

1370

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

1371

SetCheckerCmd, value_to_set=":")

1372

1373

def test_timeout(self):

1374

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

1375

SetTimeoutCmd,

1376

value_to_set=300000)

1377

1378

def test_timeout_short(self):

1379

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

1380

SetTimeoutCmd,

1381

value_to_set=300000)

1382

1383

def test_extended_timeout(self):

1384

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

1385

"foo"],

1386

SetExtendedTimeoutCmd,

1387

value_to_set=900000)

1388

1389

def test_interval(self):

1390

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

1391

SetIntervalCmd,

1392

value_to_set=120000)

1393

1394

def test_interval_short(self):

1395

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

1396

SetIntervalCmd,

1397

value_to_set=120000)

1398

1399

def test_approve_by_default(self):

1400

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

1401

ApproveByDefaultCmd)

1402

1403

def test_deny_by_default(self):

1404

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

1405

DenyByDefaultCmd)

1406

1407

def test_approval_delay(self):

1408

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

1409

"foo"], SetApprovalDelayCmd,

1410

value_to_set=30000)

1411

1412

def test_approval_duration(self):

1413

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

1414

"foo"], SetApprovalDurationCmd,

1415

value_to_set=1000)

1416

1417

def test_host(self):

1418

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

1419

"foo"], SetHostCmd,

1420

value_to_set="foo.example.org")

1421

1422

def test_host_short(self):

1423

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

1424

"foo"], SetHostCmd,

1425

value_to_set="foo.example.org")

1426

1427

def test_secret_devnull(self):

1428

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

1429

"foo"], SetSecretCmd,

1430

value_to_set=b"")

1431

1432

def test_secret_tempfile(self):

1433

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

1434

value = b"secret\0xyzzy\nbar"

1435

f.write(value)

1436

f.seek(0)

1437

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

1438

"foo"], SetSecretCmd,

1439

value_to_set=value)

1440

1441

def test_secret_devnull_short(self):

1442

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

1443

SetSecretCmd, value_to_set=b"")

1444

1445

def test_secret_tempfile_short(self):

1446

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

1447

value = b"secret\0xyzzy\nbar"

1448

f.write(value)

1449

f.seek(0)

1450

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

1451

SetSecretCmd,

1452

value_to_set=value)

1453

1454

def test_approve(self):

1455

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

1456

ApproveCmd)

1457

1458

def test_approve_short(self):

1459

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

1460

1461

def test_deny(self):

1462

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

1463

1464

def test_deny_short(self):

1465

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

1466

1467

def test_dump_json(self):

1468

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

1469

1470

def test_is_enabled(self):

1471

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

1472

IsEnabledCmd)

1473

1474

def test_is_enabled_short(self):

1475

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

1476

1477

def test_deny_before_remove(self):

1478

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

1479

"foo"])

1480

check_option_syntax(self.parser, options)

1481

commands = commands_from_options(options)

1482

self.assertEqual(len(commands), 2)

1483

self.assertIsInstance(commands[0], DenyCmd)

1484

self.assertIsInstance(commands[1], RemoveCmd)

1485

1486

def test_deny_before_remove_reversed(self):

1487

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

1488

"--all"])

1489

check_option_syntax(self.parser, options)

1490

commands = commands_from_options(options)

1491

self.assertEqual(len(commands), 2)

1492

self.assertIsInstance(commands[0], DenyCmd)

1493

self.assertIsInstance(commands[1], RemoveCmd)

1494

1495

1496

class Test_check_option_syntax(unittest.TestCase):

1497

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

1498

# check_option_syntax()

1499

actions = {

1500

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

1501

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

1502

# never used

1503

"enable": True,

1504

"disable": True,

1505

"bump_timeout": True,

1506

"start_checker": True,

1507

"stop_checker": True,

1508

"is_enabled": True,

1509

"remove": True,

1510

"checker": "x",

1511

"timeout": datetime.timedelta(),

1512

"extended_timeout": datetime.timedelta(),

1513

"interval": datetime.timedelta(),

1514

"approved_by_default": True,

1515

"approval_delay": datetime.timedelta(),

1516

"approval_duration": datetime.timedelta(),

1517

"host": "x",

1518

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

1519

"approve": True,

1520

"deny": True,

1521

}

1522

1523

def setUp(self):

1524

self.parser = argparse.ArgumentParser()

1525

add_command_line_options(self.parser)

1526

1527

@contextlib.contextmanager

1528

def assertParseError(self):

1529

with self.assertRaises(SystemExit) as e:

1530

with self.temporarily_suppress_stderr():

1531

yield

1532

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

1533

# https://docs.python.org/3/library

1534

# /argparse.html#exiting-methods

1535

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

1536

1537

@staticmethod

1538

@contextlib.contextmanager

1539

def temporarily_suppress_stderr():

1540

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

1541

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

1542

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

1543

os.close(null)

1544

try:

1545

yield

1546

finally:

1547

# restore stderr

1548

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

1549

os.close(stderrcopy)

1550

1551

def check_option_syntax(self, options):

1552

check_option_syntax(self.parser, options)

1553

1554

def test_actions_requires_client_or_all(self):

1555

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

1556

options = self.parser.parse_args()

1557

setattr(options, action, value)

1558

with self.assertParseError():

1559

self.check_option_syntax(options)

1560

1561

def test_actions_conflicts_with_verbose(self):

1562

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

1563

options = self.parser.parse_args()

1564

setattr(options, action, value)

1565

options.verbose = True

1566

with self.assertParseError():

1567

self.check_option_syntax(options)

1568

1569

def test_dump_json_conflicts_with_verbose(self):

1570

options = self.parser.parse_args()

1571

options.dump_json = True

1572

options.verbose = True

1573

with self.assertParseError():

1574

self.check_option_syntax(options)

1575

1576

def test_dump_json_conflicts_with_action(self):

1577

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

1578

options = self.parser.parse_args()

1579

setattr(options, action, value)

1580

options.dump_json = True

1581

with self.assertParseError():

1582

self.check_option_syntax(options)

1583

1584

def test_all_can_not_be_alone(self):

1585

options = self.parser.parse_args()

1586

options.all = True

1587

with self.assertParseError():

1588

self.check_option_syntax(options)

1589

1590

def test_all_is_ok_with_any_action(self):

1591

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

1592

options = self.parser.parse_args()

1593

setattr(options, action, value)

1594

options.all = True

1595

self.check_option_syntax(options)

1596

1597

def test_is_enabled_fails_without_client(self):

1598

options = self.parser.parse_args()

1599

options.is_enabled = True

1600

with self.assertParseError():

1601

self.check_option_syntax(options)

1602

1603

def test_is_enabled_works_with_one_client(self):

1604

options = self.parser.parse_args()

1605

options.is_enabled = True

1606

options.client = ["foo"]

1607

self.check_option_syntax(options)

1608

1609

def test_is_enabled_fails_with_two_clients(self):

1610

options = self.parser.parse_args()

1611

options.is_enabled = True

1612

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

1613

with self.assertParseError():

1614

self.check_option_syntax(options)

1615

1616

def test_remove_can_only_be_combined_with_action_deny(self):

1617

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

1618

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

1619

continue

1620

options = self.parser.parse_args()

1621

setattr(options, action, value)

1622

options.all = True

1623

options.remove = True

1624

with self.assertParseError():

1625

self.check_option_syntax(options)

1626

1627

1628

1629

def should_only_run_tests():

1630

parser = argparse.ArgumentParser(add_help=False)

1631

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

1632

args, unknown_args = parser.parse_known_args()

1633

run_tests = args.check

1634

if run_tests:

1635

# Remove --check argument from sys.argv

1636

sys.argv[1:] = unknown_args

1637

return run_tests

1638

1639

# Add all tests from doctest strings

1640

def load_tests(loader, tests, none):

1641

import doctest

1642

tests.addTests(doctest.DocTestSuite())

1643

return tests

1644

1645

if __name__ == "__main__":

1646

if should_only_run_tests():

1647

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

1648

unittest.main()

1649

else:

1650

main()