/mandos/trunk : revision 1030

1

#!/usr/bin/python

2

# -*- mode: python; coding: utf-8 -*-

3

4

from __future__ import division

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

5

35

import sys

6

import dbus

7

from optparse import OptionParser

36

import argparse

8

37

import locale

9

38

import datetime

10

39

import re

11

12

locale.setlocale(locale.LC_ALL, u'')

13

14

tablewords = {

15

'name': u'Name',

16

'enabled': u'Enabled',

17

'timeout': u'Timeout',

18

'last_checked_ok': u'Last Successful Check',

19

'created': u'Created',

20

'interval': u'Interval',

21

'host': u'Host',

22

'fingerprint': u'Fingerprint',

23

'checker_running': u'Check Is Running',

24

'last_enabled': u'Last Enabled',

25

'checker': u'Checker',

26

}

27

defaultkeywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

28

'checker')

29

domain = 'se.bsnet.fukt'

30

busname = domain + '.Mandos'

31

server_path = '/'

32

server_interface = domain + '.Mandos'

33

client_interface = domain + '.Mandos.Client'

34

version = "1.0.10"

35

bus = dbus.SystemBus()

36

mandos_dbus_objc = bus.get_object(busname, server_path)

37

mandos_serv = dbus.Interface(mandos_dbus_objc,

38

dbus_interface = server_interface)

39

mandos_clients = mandos_serv.GetAllClientsWithProperties()

40

41

def datetime_to_milliseconds(dt):

42

"Return the 'timeout' attribute in milliseconds"

43

return ((dt.days * 24 * 60 * 60 * 1000)

44

+ (dt.seconds * 1000)

45

+ (dt.microseconds // 1000))

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

47

import dbus

48

49

# Show warnings by default

50

if not sys.warnoptions:

51

import warnings

52

warnings.simplefilter("default")

53

54

log = logging.getLogger(sys.argv[0])

55

logging.basicConfig(level="INFO", # Show info level messages

56

format="%(message)s") # Show basic log messages

57

58

logging.captureWarnings(True) # Show warnings via the logging system

59

60

if sys.version_info.major == 2:

61

str = unicode

62

63

locale.setlocale(locale.LC_ALL, "")

64

65

domain = "se.recompile"

66

busname = domain + ".Mandos"

67

server_path = "/"

68

server_interface = domain + ".Mandos"

69

client_interface = domain + ".Mandos.Client"

70

version = "1.8.3"

71

72

73

try:

74

dbus.OBJECT_MANAGER_IFACE

75

except AttributeError:

76

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

77

46

78

47

79

def milliseconds_to_string(ms):

48

80

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

49

return "%s%02d:%02d:%02d" % (("%dT" % td.days) if td.days else "", # days

50

td.seconds // 3600, # hours

51

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

52

(td.seconds % 60)) # seconds

81

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

82

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

83

hours=td.seconds // 3600,

84

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

85

seconds=td.seconds % 60))

86

87

88

def rfc3339_duration_to_delta(duration):

89

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

90

91

>>> rfc3339_duration_to_delta("P7D")

92

datetime.timedelta(7)

93

>>> rfc3339_duration_to_delta("PT60S")

94

datetime.timedelta(0, 60)

95

>>> rfc3339_duration_to_delta("PT60M")

96

datetime.timedelta(0, 3600)

97

>>> rfc3339_duration_to_delta("P60M")

98

datetime.timedelta(1680)

99

>>> rfc3339_duration_to_delta("PT24H")

100

datetime.timedelta(1)

101

>>> rfc3339_duration_to_delta("P1W")

102

datetime.timedelta(7)

103

>>> rfc3339_duration_to_delta("PT5M30S")

104

datetime.timedelta(0, 330)

105

>>> rfc3339_duration_to_delta("P1DT3M20S")

106

datetime.timedelta(1, 200)

107

>>> # Can not be empty:

108

>>> rfc3339_duration_to_delta("")

109

Traceback (most recent call last):

110

...

111

ValueError: Invalid RFC 3339 duration: u''

112

>>> # Must start with "P":

113

>>> rfc3339_duration_to_delta("1D")

114

Traceback (most recent call last):

115

...

116

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

117

>>> # Must use correct order

118

>>> rfc3339_duration_to_delta("PT1S2M")

119

Traceback (most recent call last):

120

...

121

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

122

>>> # Time needs time marker

123

>>> rfc3339_duration_to_delta("P1H2S")

124

Traceback (most recent call last):

125

...

126

ValueError: Invalid RFC 3339 duration: u'P1H2S'

127

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

128

>>> rfc3339_duration_to_delta("P1D2W")

129

Traceback (most recent call last):

130

...

131

ValueError: Invalid RFC 3339 duration: u'P1D2W'

132

>>> rfc3339_duration_to_delta("P2W2H")

133

Traceback (most recent call last):

134

...

135

ValueError: Invalid RFC 3339 duration: u'P2W2H'

136

"""

137

138

# Parsing an RFC 3339 duration with regular expressions is not

139

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

140

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

141

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

142

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

143

# avoid excessive use of external libraries.

144

145

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

146

Token = collections.namedtuple("Token", (

147

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

148

# a "group" containing digits

149

"value", # datetime.timedelta or None

150

"followers")) # Tokens valid after this token

151

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

152

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

153

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

154

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

155

datetime.timedelta(seconds=1),

156

frozenset((token_end, )))

157

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

158

datetime.timedelta(minutes=1),

159

frozenset((token_second, token_end)))

160

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

161

datetime.timedelta(hours=1),

162

frozenset((token_minute, token_end)))

163

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

164

None,

165

frozenset((token_hour, token_minute,

166

token_second)))

167

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

168

datetime.timedelta(days=1),

169

frozenset((token_time, token_end)))

170

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

171

datetime.timedelta(weeks=4),

172

frozenset((token_day, token_end)))

173

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

174

datetime.timedelta(weeks=52),

175

frozenset((token_month, token_end)))

176

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

177

datetime.timedelta(weeks=1),

178

frozenset((token_end, )))

179

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

180

frozenset((token_year, token_month,

181

token_day, token_time,

182

token_week)))

183

# Define starting values:

184

# Value so far

185

value = datetime.timedelta()

186

found_token = None

187

# Following valid tokens

188

followers = frozenset((token_duration, ))

189

# String left to parse

190

s = duration

191

# Loop until end token is found

192

while found_token is not token_end:

193

# Search for any currently valid tokens

194

for token in followers:

195

match = token.regexp.match(s)

196

if match is not None:

197

# Token found

198

if token.value is not None:

199

# Value found, parse digits

200

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

201

# Add to value so far

202

value += factor * token.value

203

# Strip token from string

204

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

205

# Go to found token

206

found_token = token

207

# Set valid next tokens

208

followers = found_token.followers

209

break

210

else:

211

# No currently valid tokens were found

212

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

213

.format(duration))

214

# End token found

215

return value

53

216

54

217

55

218

def string_to_delta(interval):

56

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

57

58

>>> string_to_delta('7d')

219

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

220

221

try:

222

return rfc3339_duration_to_delta(interval)

223

except ValueError as e:

224

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

225

' '.join(e.args))

226

return parse_pre_1_6_1_interval(interval)

227

228

229

def parse_pre_1_6_1_interval(interval):

230

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

231

and return a datetime.timedelta

232

233

>>> parse_pre_1_6_1_interval('7d')

59

234

datetime.timedelta(7)

60

>>> string_to_delta('60s')

235

>>> parse_pre_1_6_1_interval('60s')

61

236

datetime.timedelta(0, 60)

62

>>> string_to_delta('60m')

237

>>> parse_pre_1_6_1_interval('60m')

63

238

datetime.timedelta(0, 3600)

64

>>> string_to_delta('24h')

239

>>> parse_pre_1_6_1_interval('24h')

65

240

datetime.timedelta(1)

66

>>> string_to_delta(u'1w')

241

>>> parse_pre_1_6_1_interval('1w')

67

242

datetime.timedelta(7)

68

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

243

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

69

244

datetime.timedelta(0, 330)

245

>>> parse_pre_1_6_1_interval('')

246

datetime.timedelta(0)

247

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

248

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

249

datetime.timedelta(2, 480, 18000)

250

70

251

"""

71

timevalue = datetime.timedelta(0)

72

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

73

74

for s in regexp.findall(interval):

75

try:

76

suffix = unicode(s[-1])

77

value = int(s[:-1])

78

if suffix == u"d":

79

delta = datetime.timedelta(value)

80

elif suffix == u"s":

81

delta = datetime.timedelta(0, value)

82

elif suffix == u"m":

83

delta = datetime.timedelta(0, 0, 0, 0, value)

84

elif suffix == u"h":

85

delta = datetime.timedelta(0, 0, 0, 0, 0, value)

86

elif suffix == u"w":

87

delta = datetime.timedelta(0, 0, 0, 0, 0, 0, value)

88

else:

89

raise ValueError

90

except (ValueError, IndexError):

91

raise ValueError

92

timevalue += delta

93

return timevalue

94

95

def print_clients(clients):

96

def valuetostring(value, keyword):

97

if type(value) is dbus.Boolean:

98

return u"Yes" if value else u"No"

99

if keyword in ("timeout", "interval"):

100

return milliseconds_to_string(value)

101

return unicode(value)

102

103

format_string = u' '.join(u'%%-%ds' %

104

max(len(tablewords[key]),

105

max(len(valuetostring(client[key], key))

106

for client in

107

clients))

108

for key in keywords)

109

print format_string % tuple(tablewords[key] for key in keywords)

110

for client in clients:

111

print format_string % tuple(valuetostring(client[key], key)

112

for key in keywords)

113

114

parser = OptionParser(version = "%%prog %s" % version)

115

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

116

help="Print all fields")

117

parser.add_option("-e", "--enable", action="store_true",

118

help="Enable client")

119

parser.add_option("-d", "--disable", action="store_true",

120

help="disable client")

121

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

122

help="Bump timeout for client")

123

parser.add_option("--start-checker", action="store_true",

124

help="Start checker for client")

125

parser.add_option("--stop-checker", action="store_true",

126

help="Stop checker for client")

127

parser.add_option("-V", "--is-valid", action="store_true",

128

help="Check if client is still valid")

129

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

130

help="Remove client")

131

parser.add_option("-c", "--checker", type="string",

132

help="Set checker command for client")

133

parser.add_option("-t", "--timeout", type="string",

134

help="Set timeout for client")

135

parser.add_option("-i", "--interval", type="string",

136

help="Set checker interval for client")

137

parser.add_option("-H", "--host", type="string",

138

help="Set host for client")

139

parser.add_option("-s", "--secret", type="string",

140

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

141

options, client_names = parser.parse_args()

142

143

# Compile list of clients to process

144

clients=[]

145

for name in client_names:

146

for path, client in mandos_clients.iteritems():

147

if client['name'] == name:

148

client_objc = bus.get_object(busname, path)

149

clients.append(dbus.Interface(client_objc,

150

dbus_interface

151

= client_interface))

152

break

153

else:

154

print >> sys.stderr, "Client not found on server: %r" % name

252

253

value = datetime.timedelta(0)

254

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

255

256

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

257

if suffix == "d":

258

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

259

elif suffix == "s":

260

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

261

elif suffix == "m":

262

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

263

elif suffix == "h":

264

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

265

elif suffix == "w":

266

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

267

elif suffix == "":

268

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

269

return value

270

271

272

## Classes for commands.

273

274

# Abstract classes first

275

class Command(object):

276

"""Abstract class for commands"""

277

def run(self, mandos, clients):

278

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

279

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

280

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

281

self.mandos = mandos

282

for client, properties in clients.items():

283

self.run_on_one_client(client, properties)

284

285

class PrintCmd(Command):

286

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

287

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

288

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

289

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

290

"ApprovalPending", "ApprovedByDefault",

291

"LastApprovalRequest", "ApprovalDelay",

292

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

293

"Expires", "LastCheckerStatus")

294

def run(self, mandos, clients):

295

print(self.output(clients))

296

297

class PropertyCmd(Command):

298

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

299

def run_on_one_client(self, client, properties):

300

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

301

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

302

dbus_interface=dbus.PROPERTIES_IFACE)

303

304

class ValueArgumentMixIn(object):

305

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

306

def __init__(self, value):

307

self.value_to_set = value

308

309

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

310

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

311

milliseconds."""

312

@property

313

def value_to_set(self):

314

return self._vts

315

@value_to_set.setter

316

def value_to_set(self, value):

317

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

318

self._vts = string_to_delta(value).total_seconds() * 1000

319

320

# Actual (non-abstract) command classes

321

322

class PrintTableCmd(PrintCmd):

323

def __init__(self, verbose=False):

324

self.verbose = verbose

325

326

def output(self, clients):

327

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

328

keywords = default_keywords

329

if self.verbose:

330

keywords = self.all_keywords

331

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

332

333

class TableOfClients(object):

334

tableheaders = {

335

"Name": "Name",

336

"Enabled": "Enabled",

337

"Timeout": "Timeout",

338

"LastCheckedOK": "Last Successful Check",

339

"LastApprovalRequest": "Last Approval Request",

340

"Created": "Created",

341

"Interval": "Interval",

342

"Host": "Host",

343

"Fingerprint": "Fingerprint",

344

"KeyID": "Key ID",

345

"CheckerRunning": "Check Is Running",

346

"LastEnabled": "Last Enabled",

347

"ApprovalPending": "Approval Is Pending",

348

"ApprovedByDefault": "Approved By Default",

349

"ApprovalDelay": "Approval Delay",

350

"ApprovalDuration": "Approval Duration",

351

"Checker": "Checker",

352

"ExtendedTimeout": "Extended Timeout",

353

"Expires": "Expires",

354

"LastCheckerStatus": "Last Checker Status",

355

}

356

357

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

358

self.clients = clients

359

self.keywords = keywords

360

if tableheaders is not None:

361

self.tableheaders = tableheaders

362

363

def __str__(self):

364

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

365

366

if sys.version_info.major == 2:

367

__unicode__ = __str__

368

def __str__(self):

369

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

370

371

def rows(self):

372

format_string = self.row_formatting_string()

373

rows = [self.header_line(format_string)]

374

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

375

for client in self.clients)

376

return rows

377

378

def row_formatting_string(self):

379

"Format string used to format table rows"

380

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

381

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

382

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

383

for client in self.clients)),

384

key=key)

385

for key in self.keywords)

386

387

def string_from_client(self, client, key):

388

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

389

390

@staticmethod

391

def valuetostring(value, keyword):

392

if isinstance(value, dbus.Boolean):

393

return "Yes" if value else "No"

394

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

395

"ApprovalDuration", "ExtendedTimeout"):

396

return milliseconds_to_string(value)

397

return str(value)

398

399

def header_line(self, format_string):

400

return format_string.format(**self.tableheaders)

401

402

def client_line(self, client, format_string):

403

return format_string.format(

404

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

405

for key in self.keywords})

406

407

408

409

class DumpJSONCmd(PrintCmd):

410

def output(self, clients):

411

data = {client["Name"]:

412

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

413

for key in self.all_keywords}

414

for client in clients.values()}

415

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

416

@staticmethod

417

def dbus_boolean_to_bool(value):

418

if isinstance(value, dbus.Boolean):

419

value = bool(value)

420

return value

421

422

class IsEnabledCmd(Command):

423

def run_on_one_client(self, client, properties):

424

if self.is_enabled(client, properties):

425

sys.exit(0)

155

426

sys.exit(1)

156

157

if not clients and mandos_clients.values():

158

keywords = defaultkeywords

159

if options.all:

160

keywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

161

'created', 'interval', 'host', 'fingerprint',

162

'checker_running', 'last_enabled', 'checker')

163

print_clients(mandos_clients.values())

164

165

# Process each client in the list by all selected options

166

for client in clients:

167

if options.remove:

168

mandos_serv.RemoveClient(client.__dbus_object_path__)

427

def is_enabled(self, client, properties):

428

return bool(properties["Enabled"])

429

430

class RemoveCmd(Command):

431

def run_on_one_client(self, client, properties):

432

self.mandos.RemoveClient(client.__dbus_object_path__)

433

434

class ApproveCmd(Command):

435

def run_on_one_client(self, client, properties):

436

client.Approve(dbus.Boolean(True),

437

dbus_interface=client_interface)

438

439

class DenyCmd(Command):

440

def run_on_one_client(self, client, properties):

441

client.Approve(dbus.Boolean(False),

442

dbus_interface=client_interface)

443

444

class EnableCmd(PropertyCmd):

445

property = "Enabled"

446

value_to_set = dbus.Boolean(True)

447

448

class DisableCmd(PropertyCmd):

449

property = "Enabled"

450

value_to_set = dbus.Boolean(False)

451

452

class BumpTimeoutCmd(PropertyCmd):

453

property = "LastCheckedOK"

454

value_to_set = ""

455

456

class StartCheckerCmd(PropertyCmd):

457

property = "CheckerRunning"

458

value_to_set = dbus.Boolean(True)

459

460

class StopCheckerCmd(PropertyCmd):

461

property = "CheckerRunning"

462

value_to_set = dbus.Boolean(False)

463

464

class ApproveByDefaultCmd(PropertyCmd):

465

property = "ApprovedByDefault"

466

value_to_set = dbus.Boolean(True)

467

468

class DenyByDefaultCmd(PropertyCmd):

469

property = "ApprovedByDefault"

470

value_to_set = dbus.Boolean(False)

471

472

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

473

property = "Checker"

474

475

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

476

property = "Host"

477

478

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

479

@property

480

def value_to_set(self):

481

return self._vts

482

@value_to_set.setter

483

def value_to_set(self, value):

484

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

485

self._vts = value.read()

486

value.close()

487

property = "Secret"

488

489

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

490

property = "Timeout"

491

492

class SetExtendedTimeoutCmd(PropertyCmd,

493

MillisecondsValueArgumentMixIn):

494

property = "ExtendedTimeout"

495

496

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

497

property = "Interval"

498

499

class SetApprovalDelayCmd(PropertyCmd,

500

MillisecondsValueArgumentMixIn):

501

property = "ApprovalDelay"

502

503

class SetApprovalDurationCmd(PropertyCmd,

504

MillisecondsValueArgumentMixIn):

505

property = "ApprovalDuration"

506

507

def has_actions(options):

508

return any((options.enable,

509

options.disable,

510

options.bump_timeout,

511

options.start_checker,

512

options.stop_checker,

513

options.is_enabled,

514

options.remove,

515

options.checker is not None,

516

options.timeout is not None,

517

options.extended_timeout is not None,

518

options.interval is not None,

519

options.approved_by_default is not None,

520

options.approval_delay is not None,

521

options.approval_duration is not None,

522

options.host is not None,

523

options.secret is not None,

524

options.approve,

525

options.deny))

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",

559

help="Set timeout for client")

560

parser.add_argument("--extended-timeout",

561

help="Set extended timeout for client")

562

parser.add_argument("-i", "--interval",

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",

574

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

575

parser.add_argument("--approval-duration",

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("--check", action="store_true",

588

help="Run self-test")

589

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

590

591

592

def commands_from_options(options):

593

594

commands = []

595

596

if options.dump_json:

597

commands.append(DumpJSONCmd())

598

169

599

if options.enable:

170

client.Enable()

600

commands.append(EnableCmd())

601

171

602

if options.disable:

172

client.Disable()

603

commands.append(DisableCmd())

604

173

605

if options.bump_timeout:

174

client.BumpTimeout()

606

commands.append(BumpTimeoutCmd())

607

175

608

if options.start_checker:

176

client.StartChecker()

609

commands.append(StartCheckerCmd())

610

177

611

if options.stop_checker:

178

client.StopChecker()

179

if options.is_valid:

180

sys.exit(0 if client.IsStillValid() else 1)

181

if options.checker:

182

client.SetChecker(options.checker)

183

if options.host:

184

client.SetHost(options.host)

185

if options.interval:

186

client.SetInterval(datetime_to_milliseconds

187

(string_to_delta(options.interval)))

188

if options.timeout:

189

client.SetTimeout(datetime_to_milliseconds

190

(string_to_delta(options.timeout)))

191

if options.secret:

192

client.SetSecret(dbus.ByteArray(open(options.secret, 'rb').read()))

193

612

commands.append(StopCheckerCmd())

613

614

if options.is_enabled:

615

commands.append(IsEnabledCmd())

616

617

if options.remove:

618

commands.append(RemoveCmd())

619

620

if options.checker is not None:

621

commands.append(SetCheckerCmd(options.checker))

622

623

if options.timeout is not None:

624

commands.append(SetTimeoutCmd(options.timeout))

625

626

if options.extended_timeout:

627

commands.append(

628

SetExtendedTimeoutCmd(options.extended_timeout))

629

630

if options.interval is not None:

631

commands.append(SetIntervalCmd(options.interval))

632

633

if options.approved_by_default is not None:

634

if options.approved_by_default:

635

commands.append(ApproveByDefaultCmd())

636

else:

637

commands.append(DenyByDefaultCmd())

638

639

if options.approval_delay is not None:

640

commands.append(SetApprovalDelayCmd(options.approval_delay))

641

642

if options.approval_duration is not None:

643

commands.append(

644

SetApprovalDurationCmd(options.approval_duration))

645

646

if options.host is not None:

647

commands.append(SetHostCmd(options.host))

648

649

if options.secret is not None:

650

commands.append(SetSecretCmd(options.secret))

651

652

if options.approve:

653

commands.append(ApproveCmd())

654

655

if options.deny:

656

commands.append(DenyCmd())

657

658

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

659

# optionally verbosely

660

if not commands:

661

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

662

663

return commands

664

665

666

def main():

667

parser = argparse.ArgumentParser()

668

669

add_command_line_options(parser)

670

671

options = parser.parse_args()

672

673

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

674

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

675

if options.verbose and has_actions(options):

676

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

677

if options.dump_json and (options.verbose

678

or has_actions(options)):

679

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

680

if options.all and not has_actions(options):

681

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

682

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

683

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

684

685

clientnames = options.client

686

687

try:

688

bus = dbus.SystemBus()

689

mandos_dbus_objc = bus.get_object(busname, server_path)

690

except dbus.exceptions.DBusException:

691

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

692

sys.exit(1)

693

694

mandos_serv = dbus.Interface(mandos_dbus_objc,

695

dbus_interface=server_interface)

696

mandos_serv_object_manager = dbus.Interface(

697

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

698

699

# Filter out log message from dbus module

700

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

701

class NullFilter(logging.Filter):

702

def filter(self, record):

703

return False

704

dbus_filter = NullFilter()

705

try:

706

dbus_logger.addFilter(dbus_filter)

707

mandos_clients = {path: ifs_and_props[client_interface]

708

for path, ifs_and_props in

709

mandos_serv_object_manager

710

.GetManagedObjects().items()

711

if client_interface in ifs_and_props}

712

except dbus.exceptions.DBusException as e:

713

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

714

"\n%s", e)

715

sys.exit(1)

716

finally:

717

# restore dbus logger

718

dbus_logger.removeFilter(dbus_filter)

719

720

# Compile dict of (clients: properties) to process

721

clients = {}

722

723

if not clientnames:

724

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

725

for path, properties in mandos_clients.items()}

726

else:

727

for name in clientnames:

728

for path, client in mandos_clients.items():

729

if client["Name"] == name:

730

client_objc = bus.get_object(busname, path)

731

clients[client_objc] = client

732

break

733

else:

734

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

735

sys.exit(1)

736

737

# Run all commands on clients

738

commands = commands_from_options(options)

739

for command in commands:

740

command.run(mandos_serv, clients)

741

742

743

class Test_milliseconds_to_string(unittest.TestCase):

744

def test_all(self):

745

self.assertEqual(milliseconds_to_string(93785000),

746

"1T02:03:05")

747

def test_no_days(self):

748

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

749

def test_all_zero(self):

750

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

751

def test_no_fractional_seconds(self):

752

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

753

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

754

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

755

756

class Test_string_to_delta(unittest.TestCase):

757

def test_handles_basic_rfc3339(self):

758

self.assertEqual(string_to_delta("PT0S"),

759

datetime.timedelta())

760

self.assertEqual(string_to_delta("P0D"),

761

datetime.timedelta())

762

self.assertEqual(string_to_delta("PT1S"),

763

datetime.timedelta(0, 1))

764

self.assertEqual(string_to_delta("PT2H"),

765

datetime.timedelta(0, 7200))

766

def test_falls_back_to_pre_1_6_1_with_warning(self):

767

# assertLogs only exists in Python 3.4

768

if hasattr(self, "assertLogs"):

769

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

770

value = string_to_delta("2h")

771

else:

772

class WarningFilter(logging.Filter):

773

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

774

def filter(self, record):

775

is_warning = record.levelno >= logging.WARNING

776

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

777

False)

778

return not is_warning

779

warning_filter = WarningFilter()

780

log.addFilter(warning_filter)

781

try:

782

value = string_to_delta("2h")

783

finally:

784

log.removeFilter(warning_filter)

785

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

786

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

787

788

789

class TestCmd(unittest.TestCase):

790

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

791

def setUp(self):

792

testcase = self

793

class MockClient(object):

794

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

795

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

796

self.attributes = attributes

797

self.attributes["Name"] = name

798

self.calls = []

799

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

800

testcase.assertEqual(interface, client_interface)

801

testcase.assertEqual(dbus_interface,

802

dbus.PROPERTIES_IFACE)

803

self.attributes[property] = value

804

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

805

testcase.assertEqual(interface, client_interface)

806

testcase.assertEqual(dbus_interface,

807

dbus.PROPERTIES_IFACE)

808

return self.attributes[property]

809

def Approve(self, approve, dbus_interface):

810

testcase.assertEqual(dbus_interface, client_interface)

811

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

812

dbus_interface)))

813

self.client = MockClient(

814

"foo",

815

KeyID=("92ed150794387c03ce684574b1139a65"

816

"94a34f895daaaf09fd8ea90a27cddb12"),

817

Secret=b"secret",

818

Host="foo.example.org",

819

Enabled=dbus.Boolean(True),

820

Timeout=300000,

821

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

822

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

823

Interval=120000,

824

Fingerprint=("778827225BA7DE539C5A"

825

"7CFA59CFF7CDBD9A5920"),

826

CheckerRunning=dbus.Boolean(False),

827

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

828

ApprovalPending=dbus.Boolean(False),

829

ApprovedByDefault=dbus.Boolean(True),

830

LastApprovalRequest="",

831

ApprovalDelay=0,

832

ApprovalDuration=1000,

833

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

834

ExtendedTimeout=900000,

835

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

836

LastCheckerStatus=0)

837

self.other_client = MockClient(

838

"barbar",

839

KeyID=("0558568eedd67d622f5c83b35a115f79"

840

"6ab612cff5ad227247e46c2b020f441c"),

841

Secret=b"secretbar",

842

Host="192.0.2.3",

843

Enabled=dbus.Boolean(True),

844

Timeout=300000,

845

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

846

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

847

Interval=120000,

848

Fingerprint=("3E393AEAEFB84C7E89E2"

849

"F547B3A107558FCA3A27"),

850

CheckerRunning=dbus.Boolean(True),

851

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

852

ApprovalPending=dbus.Boolean(False),

853

ApprovedByDefault=dbus.Boolean(False),

854

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

855

ApprovalDelay=30000,

856

ApprovalDuration=1000,

857

Checker=":",

858

ExtendedTimeout=900000,

859

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

860

LastCheckerStatus=-2)

861

self.clients = collections.OrderedDict(

862

[

863

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

864

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

865

])

866

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

867

868

class TestPrintTableCmd(TestCmd):

869

def test_normal(self):

870

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

871

expected_output = """

872

Name Enabled Timeout Last Successful Check

873

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

874

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

875

"""[1:-1]

876

self.assertEqual(output, expected_output)

877

def test_verbose(self):

878

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

879

expected_output = """

880

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

881

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

882

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

883

"""[1:-1]

884

self.assertEqual(output, expected_output)

885

def test_one_client(self):

886

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

887

expected_output = """

888

Name Enabled Timeout Last Successful Check

889

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

890

"""[1:-1]

891

self.assertEqual(output, expected_output)

892

893

class TestDumpJSONCmd(TestCmd):

894

def setUp(self):

895

self.expected_json = {

896

"foo": {

897

"Name": "foo",

898

"KeyID": ("92ed150794387c03ce684574b1139a65"

899

"94a34f895daaaf09fd8ea90a27cddb12"),

900

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

901

"Enabled": True,

902

"Timeout": 300000,

903

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

904

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

905

"Interval": 120000,

906

"Fingerprint": ("778827225BA7DE539C5A"

907

"7CFA59CFF7CDBD9A5920"),

908

"CheckerRunning": False,

909

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

910

"ApprovalPending": False,

911

"ApprovedByDefault": True,

912

"LastApprovalRequest": "",

913

"ApprovalDelay": 0,

914

"ApprovalDuration": 1000,

915

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

916

"ExtendedTimeout": 900000,

917

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

918

"LastCheckerStatus": 0,

919

},

920

"barbar": {

921

"Name": "barbar",

922

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

923

"6ab612cff5ad227247e46c2b020f441c"),

924

"Host": "192.0.2.3",

925

"Enabled": True,

926

"Timeout": 300000,

927

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

928

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

929

"Interval": 120000,

930

"Fingerprint": ("3E393AEAEFB84C7E89E2"

931

"F547B3A107558FCA3A27"),

932

"CheckerRunning": True,

933

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

934

"ApprovalPending": False,

935

"ApprovedByDefault": False,

936

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

937

"ApprovalDelay": 30000,

938

"ApprovalDuration": 1000,

939

"Checker": ":",

940

"ExtendedTimeout": 900000,

941

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

942

"LastCheckerStatus": -2,

943

},

944

}

945

return super(TestDumpJSONCmd, self).setUp()

946

def test_normal(self):

947

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

948

self.assertDictEqual(json_data, self.expected_json)

949

def test_one_client(self):

950

clients = self.one_client

951

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

952

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

953

self.assertDictEqual(json_data, expected_json)

954

955

class TestIsEnabledCmd(TestCmd):

956

def test_is_enabled(self):

957

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

958

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

959

def test_is_enabled_run_exits_successfully(self):

960

with self.assertRaises(SystemExit) as e:

961

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

962

if e.exception.code is not None:

963

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

964

else:

965

self.assertIsNone(e.exception.code)

966

def test_is_enabled_run_exits_with_failure(self):

967

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

968

with self.assertRaises(SystemExit) as e:

969

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

970

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

971

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

972

else:

973

self.assertIsNotNone(e.exception.code)

974

975

class TestRemoveCmd(TestCmd):

976

def test_remove(self):

977

class MockMandos(object):

978

def __init__(self):

979

self.calls = []

980

def RemoveClient(self, dbus_path):

981

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

982

mandos = MockMandos()

983

super(TestRemoveCmd, self).setUp()

984

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

985

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

986

for client in self.clients:

987

self.assertIn(("RemoveClient",

988

(client.__dbus_object_path__,)),

989

mandos.calls)

990

991

class TestApproveCmd(TestCmd):

992

def test_approve(self):

993

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

994

for client in self.clients:

995

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

996

client.calls)

997

998

class TestDenyCmd(TestCmd):

999

def test_deny(self):

1000

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

1001

for client in self.clients:

1002

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

1003

client.calls)

1004

1005

class TestEnableCmd(TestCmd):

1006

def test_enable(self):

1007

for client in self.clients:

1008

client.attributes["Enabled"] = False

1009

1010

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

1011

1012

for client in self.clients:

1013

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

1014

1015

class TestDisableCmd(TestCmd):

1016

def test_disable(self):

1017

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

1018

1019

for client in self.clients:

1020

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

1021

1022

class Unique(object):

1023

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

1024

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

1025

1026

class TestPropertyCmd(TestCmd):

1027

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

1028

def runTest(self):

1029

if not hasattr(self, "command"):

1030

return

1031

values_to_get = getattr(self, "values_to_get",

1032

self.values_to_set)

1033

for value_to_set, value_to_get in zip(self.values_to_set,

1034

values_to_get):

1035

for client in self.clients:

1036

old_value = client.attributes[self.property]

1037

self.assertNotIsInstance(old_value, Unique)

1038

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

1039

self.run_command(value_to_set, self.clients)

1040

for client in self.clients:

1041

value = client.attributes[self.property]

1042

self.assertNotIsInstance(value, Unique)

1043

self.assertEqual(value, value_to_get)

1044

def run_command(self, value, clients):

1045

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

1046

1047

class TestBumpTimeoutCmd(TestPropertyCmd):

1048

command = BumpTimeoutCmd

1049

property = "LastCheckedOK"

1050

values_to_set = [""]

1051

1052

class TestStartCheckerCmd(TestPropertyCmd):

1053

command = StartCheckerCmd

1054

property = "CheckerRunning"

1055

values_to_set = [dbus.Boolean(True)]

1056

1057

class TestStopCheckerCmd(TestPropertyCmd):

1058

command = StopCheckerCmd

1059

property = "CheckerRunning"

1060

values_to_set = [dbus.Boolean(False)]

1061

1062

class TestApproveByDefaultCmd(TestPropertyCmd):

1063

command = ApproveByDefaultCmd

1064

property = "ApprovedByDefault"

1065

values_to_set = [dbus.Boolean(True)]

1066

1067

class TestDenyByDefaultCmd(TestPropertyCmd):

1068

command = DenyByDefaultCmd

1069

property = "ApprovedByDefault"

1070

values_to_set = [dbus.Boolean(False)]

1071

1072

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1073

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

1074

ValueArgumentMixIn"""

1075

def runTest(self):

1076

if type(self) is TestValueArgumentPropertyCmd:

1077

return

1078

return super(TestValueArgumentPropertyCmd, self).runTest()

1079

def run_command(self, value, clients):

1080

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

1081

1082

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1083

command = SetCheckerCmd

1084

property = "Checker"

1085

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

1086

1087

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1088

command = SetHostCmd

1089

property = "Host"

1090

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

1091

1092

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1093

command = SetSecretCmd

1094

property = "Secret"

1095

values_to_set = [open("/dev/null", "rb"),

1096

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

1097

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

1098

1099

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1100

command = SetTimeoutCmd

1101

property = "Timeout"

1102

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1103

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

1104

1105

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1106

command = SetExtendedTimeoutCmd

1107

property = "ExtendedTimeout"

1108

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1109

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

1110

1111

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1112

command = SetIntervalCmd

1113

property = "Interval"

1114

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1115

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

1116

1117

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1118

command = SetApprovalDelayCmd

1119

property = "ApprovalDelay"

1120

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1121

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

1122

1123

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1124

command = SetApprovalDurationCmd

1125

property = "ApprovalDuration"

1126

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1127

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

1128

1129

class TestOptions(unittest.TestCase):

1130

def setUp(self):

1131

self.parser = argparse.ArgumentParser()

1132

add_command_line_options(self.parser)

1133

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

1134

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

1135

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

1136

options = self.parser.parse_args(args)

1137

commands = commands_from_options(options)

1138

self.assertEqual(len(commands), 1)

1139

command = commands[0]

1140

self.assertIsInstance(command, command_cls)

1141

for key, value in cmd_attrs.items():

1142

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

1143

def test_default_is_show_table(self):

1144

self.assert_command_from_args([], PrintTableCmd,

1145

verbose=False)

1146

def test_show_table_verbose(self):

1147

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

1148

verbose=True)

1149

def test_enable(self):

1150

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

1151

def test_disable(self):

1152

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

1153

DisableCmd)

1154

1155

1156

1157

def should_only_run_tests():

1158

parser = argparse.ArgumentParser(add_help=False)

1159

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

1160

args, unknown_args = parser.parse_known_args()

1161

run_tests = args.check

1162

if run_tests:

1163

# Remove --check argument from sys.argv

1164

sys.argv[1:] = unknown_args

1165

return run_tests

1166

1167

# Add all tests from doctest strings

1168

def load_tests(loader, tests, none):

1169

import doctest

1170

tests.addTests(doctest.DocTestSuite())

1171

return tests

1172

1173

if __name__ == "__main__":

1174

if should_only_run_tests():

1175

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

1176

unittest.main()

1177

else:

1178

main()