/mandos/trunk : revision 1080

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

40

import os

12

13

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

14

15

tablewords = {

16

'Name': u'Name',

17

'Enabled': u'Enabled',

18

'Timeout': u'Timeout',

19

'LastCheckedOK': u'Last Successful Check',

20

'Created': u'Created',

21

'Interval': u'Interval',

22

'Host': u'Host',

23

'Fingerprint': u'Fingerprint',

24

'CheckerRunning': u'Check Is Running',

25

'LastEnabled': u'Last Enabled',

26

'Checker': u'Checker',

27

}

28

defaultkeywords = ('Name', 'Enabled', 'Timeout', 'LastCheckedOK')

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.14"

35

36

def timedelta_to_milliseconds(td):

37

"Convert a datetime.timedelta object to milliseconds"

38

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

39

+ (td.seconds * 1000)

40

+ (td.microseconds // 1000))

41

42

def milliseconds_to_string(ms):

43

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

44

return (u"%(days)s%(hours)02d:%(minutes)02d:%(seconds)02d"

45

% { "days": "%dT" % td.days if td.days else "",

46

"hours": td.seconds // 3600,

47

"minutes": (td.seconds % 3600) // 60,

48

"seconds": td.seconds % 60,

49

})

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 main():

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

commands = commands_from_options(options)

129

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

135

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

136

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

137

help="show version number and exit")

138

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

139

help="Select all clients")

140

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

141

help="Print all fields")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

150

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

151

help="Bump timeout for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

159

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

160

help="Check if client is enabled")

161

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

162

help="Remove client")

163

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

164

help="Set checker command for client")

165

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

166

help="Set timeout for client")

167

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

168

help="Set extended timeout for client")

169

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

170

help="Set checker interval for client")

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

181

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

182

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

183

help="Set duration of one client approval")

184

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

185

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

186

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

187

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

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

198

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

50

199

51

200

52

201

def string_to_delta(interval):

53

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

54

55

>>> string_to_delta('7d')

56

datetime.timedelta(7)

57

>>> string_to_delta('60s')

58

datetime.timedelta(0, 60)

59

>>> string_to_delta('60m')

60

datetime.timedelta(0, 3600)

61

>>> string_to_delta('24h')

62

datetime.timedelta(1)

63

>>> string_to_delta(u'1w')

64

datetime.timedelta(7)

65

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

66

datetime.timedelta(0, 330)

67

"""

68

timevalue = datetime.timedelta(0)

69

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

70

71

for s in regexp.findall(interval):

72

try:

73

suffix = unicode(s[-1])

74

value = int(s[:-1])

75

if suffix == u"d":

76

delta = datetime.timedelta(value)

77

elif suffix == u"s":

78

delta = datetime.timedelta(0, value)

79

elif suffix == u"m":

80

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

81

elif suffix == u"h":

82

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

83

elif suffix == u"w":

84

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

85

else:

86

raise ValueError

87

except (ValueError, IndexError):

88

raise ValueError

89

timevalue += delta

90

return timevalue

91

92

def print_clients(clients, keywords):

93

def valuetostring(value, keyword):

94

if type(value) is dbus.Boolean:

95

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

96

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

97

return milliseconds_to_string(value)

98

return unicode(value)

99

100

# Create format string to print table rows

101

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

102

max(len(tablewords[key]),

103

max(len(valuetostring(client[key],

104

key))

105

for client in

106

clients))

107

for key in keywords)

108

# Print header line

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

def has_actions(options):

114

return any((options.enable,

115

options.disable,

116

options.bump_timeout,

117

options.start_checker,

118

options.stop_checker,

119

options.is_enabled,

120

options.remove,

121

options.checker is not None,

122

options.timeout is not None,

123

options.interval is not None,

124

options.host is not None,

125

options.secret is not None,

126

options.approve,

127

options.deny))

128

129

def main():

130

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

131

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

132

help="Select all clients")

133

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

134

help="Print all fields")

135

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

136

help="Enable client")

137

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

138

help="disable client")

139

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

140

help="Bump timeout for client")

141

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

142

help="Start checker for client")

143

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

144

help="Stop checker for client")

145

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

146

help="Check if client is enabled")

147

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

148

help="Remove client")

149

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

150

help="Set checker command for client")

151

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

152

help="Set timeout for client")

153

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

154

help="Set checker interval for client")

155

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

156

help="Set host for client")

157

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

158

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

159

parser.add_option("-A", "--approve", action="store_true",

160

help="Approve any current client request")

161

parser.add_option("-D", "--deny", action="store_true",

162

help="Deny any current client request")

163

options, client_names = parser.parse_args()

164

165

if has_actions(options) and not client_names and not options.all:

166

parser.error('Options requires clients names or --all.')

167

if options.verbose and has_actions(options):

168

parser.error('Verbose option can only be used alone or with --all.')

169

if options.all and not has_actions(options):

170

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

171

172

try:

173

bus = dbus.SystemBus()

174

mandos_dbus_objc = bus.get_object(busname, server_path)

175

except dbus.exceptions.DBusException:

176

print >> sys.stderr, "Could not connect to Mandos server"

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

213

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

214

215

>>> rfc3339_duration_to_delta("P7D")

216

datetime.timedelta(7)

217

>>> rfc3339_duration_to_delta("PT60S")

218

datetime.timedelta(0, 60)

219

>>> rfc3339_duration_to_delta("PT60M")

220

datetime.timedelta(0, 3600)

221

>>> rfc3339_duration_to_delta("P60M")

222

datetime.timedelta(1680)

223

>>> rfc3339_duration_to_delta("PT24H")

224

datetime.timedelta(1)

225

>>> rfc3339_duration_to_delta("P1W")

226

datetime.timedelta(7)

227

>>> rfc3339_duration_to_delta("PT5M30S")

228

datetime.timedelta(0, 330)

229

>>> rfc3339_duration_to_delta("P1DT3M20S")

230

datetime.timedelta(1, 200)

231

>>> # Can not be empty:

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: ""

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

ValueError: Invalid RFC 3339 duration: "1D"

241

>>> # Must use correct order

242

>>> rfc3339_duration_to_delta("PT1S2M")

243

Traceback (most recent call last):

244

...

245

ValueError: Invalid RFC 3339 duration: "PT1S2M"

246

>>> # Time needs time marker

247

>>> rfc3339_duration_to_delta("P1H2S")

248

Traceback (most recent call last):

249

...

250

ValueError: Invalid RFC 3339 duration: "P1H2S"

251

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

252

>>> rfc3339_duration_to_delta("P1D2W")

253

Traceback (most recent call last):

254

...

255

ValueError: Invalid RFC 3339 duration: "P1D2W"

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: "P2W2H"

260

"""

261

262

# Parsing an RFC 3339 duration with regular expressions is not

263

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

264

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

265

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

266

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

267

# avoid excessive use of external libraries.

268

269

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

270

Token = collections.namedtuple("Token", (

271

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

272

# a "group" containing digits

273

"value", # datetime.timedelta or None

274

"followers")) # Tokens valid after this token

275

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

276

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

277

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

278

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

279

datetime.timedelta(seconds=1),

280

frozenset((token_end, )))

281

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

282

datetime.timedelta(minutes=1),

283

frozenset((token_second, token_end)))

284

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

285

datetime.timedelta(hours=1),

286

frozenset((token_minute, token_end)))

287

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

288

None,

289

frozenset((token_hour, token_minute,

290

token_second)))

291

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

292

datetime.timedelta(days=1),

293

frozenset((token_time, token_end)))

294

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

295

datetime.timedelta(weeks=4),

296

frozenset((token_day, token_end)))

297

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

298

datetime.timedelta(weeks=52),

299

frozenset((token_month, token_end)))

300

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

301

datetime.timedelta(weeks=1),

302

frozenset((token_end, )))

303

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

304

frozenset((token_year, token_month,

305

token_day, token_time,

306

token_week)))

307

# Define starting values:

308

# Value so far

309

value = datetime.timedelta()

310

found_token = None

311

# Following valid tokens

312

followers = frozenset((token_duration, ))

313

# String left to parse

314

s = duration

315

# Loop until end token is found

316

while found_token is not token_end:

317

# Search for any currently valid tokens

318

for token in followers:

319

match = token.regexp.match(s)

320

if match is not None:

321

# Token found

322

if token.value is not None:

323

# Value found, parse digits

324

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

325

# Add to value so far

326

value += factor * token.value

327

# Strip token from string

328

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

329

# Go to found token

330

found_token = token

331

# Set valid next tokens

332

followers = found_token.followers

333

break

334

else:

335

# No currently valid tokens were found

336

raise ValueError("Invalid RFC 3339 duration: \"{}\""

337

.format(duration))

338

# End token found

339

return value

340

341

342

def parse_pre_1_6_1_interval(interval):

343

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

344

and return a datetime.timedelta

345

346

>>> parse_pre_1_6_1_interval('7d')

347

datetime.timedelta(7)

348

>>> parse_pre_1_6_1_interval('60s')

349

datetime.timedelta(0, 60)

350

>>> parse_pre_1_6_1_interval('60m')

351

datetime.timedelta(0, 3600)

352

>>> parse_pre_1_6_1_interval('24h')

353

datetime.timedelta(1)

354

>>> parse_pre_1_6_1_interval('1w')

355

datetime.timedelta(7)

356

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

357

datetime.timedelta(0, 330)

358

>>> parse_pre_1_6_1_interval('')

359

datetime.timedelta(0)

360

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

361

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

362

datetime.timedelta(2, 480, 18000)

363

364

"""

365

366

value = datetime.timedelta(0)

367

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

368

369

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

370

if suffix == "d":

371

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

372

elif suffix == "s":

373

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

374

elif suffix == "m":

375

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

376

elif suffix == "h":

377

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

378

elif suffix == "w":

379

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

380

elif suffix == "":

381

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

382

return value

383

384

385

def check_option_syntax(parser, options):

386

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

387

argparse"""

388

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

408

409

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

410

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

411

if options.verbose and has_actions(options):

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

416

if options.all and not has_actions(options):

417

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

418

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

419

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

"Could not connect to Mandos server: %s"):

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

log.critical(*(args + (e,)), **kwargs)

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

"Failed to access Mandos server through D-Bus:\n%s"):

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

def __exit__(self, exc_type, exc_val, exc_tb):

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(command.IsEnabled())

481

482

if options.approve:

483

commands.append(command.Approve())

484

485

if options.deny:

486

commands.append(command.Deny())

487

488

if options.remove:

489

commands.append(command.Remove())

490

491

if options.dump_json:

492

commands.append(command.DumpJSON())

493

494

if options.enable:

495

commands.append(command.Enable())

496

497

if options.disable:

498

commands.append(command.Disable())

499

500

if options.bump_timeout:

501

commands.append(command.BumpTimeout())

502

503

if options.start_checker:

504

commands.append(command.StartChecker())

505

506

if options.stop_checker:

507

commands.append(command.StopChecker())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(command.ApproveByDefault())

512

else:

513

commands.append(command.DenyByDefault())

514

515

if options.checker is not None:

516

commands.append(command.SetChecker(options.checker))

517

518

if options.host is not None:

519

commands.append(command.SetHost(options.host))

520

521

if options.secret is not None:

522

commands.append(command.SetSecret(options.secret))

523

524

if options.timeout is not None:

525

commands.append(command.SetTimeout(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

command.SetExtendedTimeout(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(command.SetInterval(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(

536

command.SetApprovalDelay(options.approval_delay))

537

538

if options.approval_duration is not None:

539

commands.append(

540

command.SetApprovalDuration(options.approval_duration))

541

542

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

543

# optionally verbosely

544

if not commands:

545

commands.append(command.PrintTable(verbose=options.verbose))

546

547

return commands

548

549

550

class command(object):

551

"""A namespace for command classes"""

552

553

class Base(object):

554

"""Abstract base class for commands"""

555

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

556

"""Normal commands should implement run_on_one_client(),

557

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

558

override this run() method instead.

559

"""

560

self.mandos = mandos

561

for clientpath, properties in clients.items():

562

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

563

dbus_busname, str(clientpath))

564

client = bus.get_object(dbus_busname, clientpath)

565

self.run_on_one_client(client, properties)

566

567

568

class IsEnabled(Base):

569

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

570

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

571

if self.is_enabled(client, properties):

572

sys.exit(0)

177

573

sys.exit(1)

178

179

mandos_serv = dbus.Interface(mandos_dbus_objc,

180

dbus_interface = server_interface)

181

182

#block stderr since dbus library prints to stderr

574

def is_enabled(self, client, properties):

575

return properties["Enabled"]

576

577

578

class Approve(Base):

579

def run_on_one_client(self, client, properties):

580

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

581

client.__dbus_object_path__,

582

client_dbus_interface)

583

client.Approve(dbus.Boolean(True),

584

dbus_interface=client_dbus_interface)

585

586

587

class Deny(Base):

588

def run_on_one_client(self, client, properties):

589

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

590

client.__dbus_object_path__,

591

client_dbus_interface)

592

client.Approve(dbus.Boolean(False),

593

dbus_interface=client_dbus_interface)

594

595

596

class Remove(Base):

597

def run_on_one_client(self, client, properties):

598

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

599

dbus_busname, server_dbus_path,

600

server_dbus_interface,

601

str(client.__dbus_object_path__))

602

self.mandos.RemoveClient(client.__dbus_object_path__)

603

604

605

class Output(Base):

606

"""Abstract class for commands outputting client details"""

607

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

608

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

609

"Fingerprint", "CheckerRunning",

610

"LastEnabled", "ApprovalPending",

611

"ApprovedByDefault", "LastApprovalRequest",

612

"ApprovalDelay", "ApprovalDuration",

613

"Checker", "ExtendedTimeout", "Expires",

614

"LastCheckerStatus")

615

616

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

617

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

618

619

def output(self, clients):

620

raise NotImplementedError()

621

622

623

class DumpJSON(Output):

624

def output(self, clients):

625

data = {client["Name"]:

626

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

627

for key in self.all_keywords}

628

for client in clients}

629

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

630

631

@staticmethod

632

def dbus_boolean_to_bool(value):

633

if isinstance(value, dbus.Boolean):

634

value = bool(value)

635

return value

636

637

638

class PrintTable(Output):

639

def __init__(self, verbose=False):

640

self.verbose = verbose

641

642

def output(self, clients):

643

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

644

"LastCheckedOK")

645

keywords = default_keywords

646

if self.verbose:

647

keywords = self.all_keywords

648

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

649

650

class TableOfClients(object):

651

tableheaders = {

652

"Name": "Name",

653

"Enabled": "Enabled",

654

"Timeout": "Timeout",

655

"LastCheckedOK": "Last Successful Check",

656

"LastApprovalRequest": "Last Approval Request",

657

"Created": "Created",

658

"Interval": "Interval",

659

"Host": "Host",

660

"Fingerprint": "Fingerprint",

661

"KeyID": "Key ID",

662

"CheckerRunning": "Check Is Running",

663

"LastEnabled": "Last Enabled",

664

"ApprovalPending": "Approval Is Pending",

665

"ApprovedByDefault": "Approved By Default",

666

"ApprovalDelay": "Approval Delay",

667

"ApprovalDuration": "Approval Duration",

668

"Checker": "Checker",

669

"ExtendedTimeout": "Extended Timeout",

670

"Expires": "Expires",

671

"LastCheckerStatus": "Last Checker Status",

672

}

673

674

def __init__(self, clients, keywords):

675

self.clients = clients

676

self.keywords = keywords

677

678

def __str__(self):

679

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

680

681

if sys.version_info.major == 2:

682

__unicode__ = __str__

683

def __str__(self):

684

return str(self).encode(

685

locale.getpreferredencoding())

686

687

def rows(self):

688

format_string = self.row_formatting_string()

689

rows = [self.header_line(format_string)]

690

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

691

for client in self.clients)

692

return rows

693

694

def row_formatting_string(self):

695

"Format string used to format table rows"

696

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

697

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

698

*(len(self.string_from_client(client,

699

key))

700

for client in self.clients)),

701

key=key)

702

for key in self.keywords)

703

704

def string_from_client(self, client, key):

705

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

706

707

@classmethod

708

def valuetostring(cls, value, keyword):

709

if isinstance(value, dbus.Boolean):

710

return "Yes" if value else "No"

711

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

712

"ApprovalDuration", "ExtendedTimeout"):

713

return cls.milliseconds_to_string(value)

714

return str(value)

715

716

def header_line(self, format_string):

717

return format_string.format(**self.tableheaders)

718

719

def client_line(self, client, format_string):

720

return format_string.format(

721

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

722

for key in self.keywords})

723

724

@staticmethod

725

def milliseconds_to_string(ms):

726

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

727

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

728

.format(days="{}T".format(td.days)

729

if td.days else "",

730

hours=td.seconds // 3600,

731

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

732

seconds=td.seconds % 60))

733

734

735

class Property(Base):

736

"Abstract class for Actions for setting one client property"

737

738

def run_on_one_client(self, client, properties):

739

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

740

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

741

client.__dbus_object_path__,

742

dbus.PROPERTIES_IFACE, client_dbus_interface,

743

self.propname, self.value_to_set

744

if not isinstance(self.value_to_set,

745

dbus.Boolean)

746

else bool(self.value_to_set))

747

client.Set(client_dbus_interface, self.propname,

748

self.value_to_set,

749

dbus_interface=dbus.PROPERTIES_IFACE)

750

751

@property

752

def propname(self):

753

raise NotImplementedError()

754

755

756

class Enable(Property):

757

propname = "Enabled"

758

value_to_set = dbus.Boolean(True)

759

760

761

class Disable(Property):

762

propname = "Enabled"

763

value_to_set = dbus.Boolean(False)

764

765

766

class BumpTimeout(Property):

767

propname = "LastCheckedOK"

768

value_to_set = ""

769

770

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

793

def __init__(self, value):

794

self.value_to_set = value

795

796

797

class SetChecker(PropertyValue):

798

propname = "Checker"

799

800

801

class SetHost(PropertyValue):

802

propname = "Host"

803

804

805

class SetSecret(PropertyValue):

806

propname = "Secret"

807

808

@property

809

def value_to_set(self):

810

return self._vts

811

812

@value_to_set.setter

813

def value_to_set(self, value):

814

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

815

self._vts = value.read()

816

value.close()

817

818

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

822

823

@property

824

def value_to_set(self):

825

return self._vts

826

827

@value_to_set.setter

828

def value_to_set(self, value):

829

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

830

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

831

832

833

class SetTimeout(MillisecondsPropertyValueArgument):

834

propname = "Timeout"

835

836

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

838

propname = "ExtendedTimeout"

839

840

841

class SetInterval(MillisecondsPropertyValueArgument):

842

propname = "Interval"

843

844

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDelay"

847

848

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

850

propname = "ApprovalDuration"

851

852

853

854

class TestCaseWithAssertLogs(unittest.TestCase):

855

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

856

857

if not hasattr(unittest.TestCase, "assertLogs"):

858

@contextlib.contextmanager

859

def assertLogs(self, logger, level=logging.INFO):

860

capturing_handler = self.CapturingLevelHandler(level)

861

old_level = logger.level

862

old_propagate = logger.propagate

863

logger.addHandler(capturing_handler)

864

logger.setLevel(level)

865

logger.propagate = False

866

try:

867

yield capturing_handler.watcher

868

finally:

869

logger.propagate = old_propagate

870

logger.removeHandler(capturing_handler)

871

logger.setLevel(old_level)

872

self.assertGreater(len(capturing_handler.watcher.records),

873

0)

874

875

class CapturingLevelHandler(logging.Handler):

876

def __init__(self, level, *args, **kwargs):

877

logging.Handler.__init__(self, *args, **kwargs)

878

self.watcher = self.LoggingWatcher([], [])

879

def emit(self, record):

880

self.watcher.records.append(record)

881

self.watcher.output.append(self.format(record))

882

883

LoggingWatcher = collections.namedtuple("LoggingWatcher",

884

("records",

885

"output"))

886

887

888

class Test_string_to_delta(TestCaseWithAssertLogs):

889

# Just test basic RFC 3339 functionality here, the doc string for

890

# rfc3339_duration_to_delta() already has more comprehensive

891

# tests, which is run by doctest.

892

893

def test_rfc3339_zero_seconds(self):

894

self.assertEqual(string_to_delta("PT0S"),

895

datetime.timedelta())

896

897

def test_rfc3339_zero_days(self):

898

self.assertEqual(string_to_delta("P0D"),

899

datetime.timedelta())

900

901

def test_rfc3339_one_second(self):

902

self.assertEqual(string_to_delta("PT1S"),

903

datetime.timedelta(0, 1))

904

905

def test_rfc3339_two_hours(self):

906

self.assertEqual(string_to_delta("PT2H"),

907

datetime.timedelta(0, 7200))

908

909

def test_falls_back_to_pre_1_6_1_with_warning(self):

910

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

911

value = string_to_delta("2h")

912

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

913

914

915

class Test_check_option_syntax(unittest.TestCase):

916

def setUp(self):

917

self.parser = argparse.ArgumentParser()

918

add_command_line_options(self.parser)

919

920

def test_actions_requires_client_or_all(self):

921

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

922

options = self.parser.parse_args()

923

setattr(options, action, value)

924

with self.assertParseError():

925

self.check_option_syntax(options)

926

927

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

928

# check_option_syntax()

929

actions = {

930

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

931

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

932

# never used

933

"enable": True,

934

"disable": True,

935

"bump_timeout": True,

936

"start_checker": True,

937

"stop_checker": True,

938

"is_enabled": True,

939

"remove": True,

940

"checker": "x",

941

"timeout": datetime.timedelta(),

942

"extended_timeout": datetime.timedelta(),

943

"interval": datetime.timedelta(),

944

"approved_by_default": True,

945

"approval_delay": datetime.timedelta(),

946

"approval_duration": datetime.timedelta(),

947

"host": "x",

948

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

949

"approve": True,

950

"deny": True,

951

}

952

953

@contextlib.contextmanager

954

def assertParseError(self):

955

with self.assertRaises(SystemExit) as e:

956

with self.redirect_stderr_to_devnull():

957

yield

958

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

959

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

960

# /argparse.html#exiting-methods

961

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

962

963

@staticmethod

964

@contextlib.contextmanager

965

def redirect_stderr_to_devnull():

183

966

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

184

967

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

185

968

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

186

969

os.close(null)

187

970

try:

188

try:

189

mandos_clients = mandos_serv.GetAllClientsWithProperties()

190

finally:

191

#restore stderr

192

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

193

os.close(stderrcopy)

194

except dbus.exceptions.DBusException, e:

195

print >> sys.stderr, "Access denied: Accessing mandos server through dbus."

196

sys.exit(1)

197

198

# Compile list of clients to process

199

clients=[]

200

201

if options.all or not client_names:

202

clients = (bus.get_object(busname, path) for path in mandos_clients.iterkeys())

203

else:

204

for name in client_names:

205

for path, client in mandos_clients.iteritems():

206

if client['Name'] == name:

207

client_objc = bus.get_object(busname, path)

208

clients.append(client_objc)

209

break

210

else:

211

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

212

sys.exit(1)

213

214

if not has_actions(options) and clients:

215

if options.verbose:

216

keywords = ('Name', 'Enabled', 'Timeout', 'LastCheckedOK',

217

'Created', 'Interval', 'Host', 'Fingerprint',

218

'CheckerRunning', 'LastEnabled', 'Checker')

219

else:

220

keywords = defaultkeywords

221

222

print_clients(mandos_clients.values(), keywords)

223

else:

224

# Process each client in the list by all selected options

225

for client in clients:

226

if options.remove:

227

mandos_serv.RemoveClient(client.__dbus_object_path__)

228

if options.enable:

229

client.Enable(dbus_interface=client_interface)

230

if options.disable:

231

client.Disable(dbus_interface=client_interface)

232

if options.bump_timeout:

233

client.CheckedOK(dbus_interface=client_interface)

234

if options.start_checker:

235

client.StartChecker(dbus_interface=client_interface)

236

if options.stop_checker:

237

client.StopChecker(dbus_interface=client_interface)

238

if options.is_enabled:

239

sys.exit(0 if client.Get(client_interface,

240

u"Enabled",

241

dbus_interface=dbus.PROPERTIES_IFACE)

242

else 1)

243

if options.checker:

244

client.Set(client_interface, u"Checker", options.checker,

245

dbus_interface=dbus.PROPERTIES_IFACE)

246

if options.host:

247

client.Set(client_interface, u"Host", options.host,

248

dbus_interface=dbus.PROPERTIES_IFACE)

249

if options.interval:

250

client.Set(client_interface, u"Interval",

251

timedelta_to_milliseconds

252

(string_to_delta(options.interval)),

253

dbus_interface=dbus.PROPERTIES_IFACE)

254

if options.timeout:

255

client.Set(client_interface, u"Timeout",

256

timedelta_to_milliseconds(string_to_delta

257

(options.timeout)),

258

dbus_interface=dbus.PROPERTIES_IFACE)

259

if options.secret:

260

client.Set(client_interface, u"Secret",

261

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

262

dbus_interface=dbus.PROPERTIES_IFACE)

263

if options.approve:

264

client.Approve(dbus.Boolean(True), dbus_interface=client_interface)

265

if options.deny:

266

client.Approve(dbus.Boolean(False), dbus_interface=client_interface)

267

268

if __name__ == '__main__':

269

main()

971

yield

972

finally:

973

# restore stderr

974

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

975

os.close(stderrcopy)

976

977

def check_option_syntax(self, options):

978

check_option_syntax(self.parser, options)

979

980

def test_actions_all_conflicts_with_verbose(self):

981

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

982

options = self.parser.parse_args()

983

setattr(options, action, value)

984

options.all = True

985

options.verbose = True

986

with self.assertParseError():

987

self.check_option_syntax(options)

988

989

def test_actions_with_client_conflicts_with_verbose(self):

990

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

991

options = self.parser.parse_args()

992

setattr(options, action, value)

993

options.verbose = True

994

options.client = ["foo"]

995

with self.assertParseError():

996

self.check_option_syntax(options)

997

998

def test_dump_json_conflicts_with_verbose(self):

999

options = self.parser.parse_args()

1000

options.dump_json = True

1001

options.verbose = True

1002

with self.assertParseError():

1003

self.check_option_syntax(options)

1004

1005

def test_dump_json_conflicts_with_action(self):

1006

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

1007

options = self.parser.parse_args()

1008

setattr(options, action, value)

1009

options.dump_json = True

1010

with self.assertParseError():

1011

self.check_option_syntax(options)

1012

1013

def test_all_can_not_be_alone(self):

1014

options = self.parser.parse_args()

1015

options.all = True

1016

with self.assertParseError():

1017

self.check_option_syntax(options)

1018

1019

def test_all_is_ok_with_any_action(self):

1020

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

1021

options = self.parser.parse_args()

1022

setattr(options, action, value)

1023

options.all = True

1024

self.check_option_syntax(options)

1025

1026

def test_any_action_is_ok_with_one_client(self):

1027

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

1028

options = self.parser.parse_args()

1029

setattr(options, action, value)

1030

options.client = ["foo"]

1031

self.check_option_syntax(options)

1032

1033

def test_one_client_with_all_actions_except_is_enabled(self):

1034

options = self.parser.parse_args()

1035

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

1036

if action == "is_enabled":

1037

continue

1038

setattr(options, action, value)

1039

options.client = ["foo"]

1040

self.check_option_syntax(options)

1041

1042

def test_two_clients_with_all_actions_except_is_enabled(self):

1043

options = self.parser.parse_args()

1044

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

1045

if action == "is_enabled":

1046

continue

1047

setattr(options, action, value)

1048

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

1049

self.check_option_syntax(options)

1050

1051

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1052

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

1053

if action == "is_enabled":

1054

continue

1055

options = self.parser.parse_args()

1056

setattr(options, action, value)

1057

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

1058

self.check_option_syntax(options)

1059

1060

def test_is_enabled_fails_without_client(self):

1061

options = self.parser.parse_args()

1062

options.is_enabled = True

1063

with self.assertParseError():

1064

self.check_option_syntax(options)

1065

1066

def test_is_enabled_fails_with_two_clients(self):

1067

options = self.parser.parse_args()

1068

options.is_enabled = True

1069

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

1070

with self.assertParseError():

1071

self.check_option_syntax(options)

1072

1073

def test_remove_can_only_be_combined_with_action_deny(self):

1074

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

1075

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

1076

continue

1077

options = self.parser.parse_args()

1078

setattr(options, action, value)

1079

options.all = True

1080

options.remove = True

1081

with self.assertParseError():

1082

self.check_option_syntax(options)

1083

1084

1085

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1086

def test_calls_and_returns_get_object_on_bus(self):

1087

class MockBus(object):

1088

called = False

1089

def get_object(mockbus_self, busname, dbus_path):

1090

# Note that "self" is still the testcase instance,

1091

# this MockBus instance is in "mockbus_self".

1092

self.assertEqual(busname, dbus_busname)

1093

self.assertEqual(dbus_path, server_dbus_path)

1094

mockbus_self.called = True

1095

return mockbus_self

1096

1097

mockbus = get_mandos_dbus_object(bus=MockBus())

1098

self.assertIsInstance(mockbus, MockBus)

1099

self.assertTrue(mockbus.called)

1100

1101

def test_logs_and_exits_on_dbus_error(self):

1102

class MockBusFailing(object):

1103

def get_object(self, busname, dbus_path):

1104

raise dbus.exceptions.DBusException("Test")

1105

1106

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

1107

with self.assertRaises(SystemExit) as e:

1108

bus = get_mandos_dbus_object(bus=MockBusFailing())

1109

1110

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

1111

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

1112

else:

1113

self.assertIsNotNone(e.exception.code)

1114

1115

1116

class Test_get_managed_objects(TestCaseWithAssertLogs):

1117

def test_calls_and_returns_GetManagedObjects(self):

1118

managed_objects = {"/clients/foo": { "Name": "foo"}}

1119

class MockObjectManager(object):

1120

def GetManagedObjects(self):

1121

return managed_objects

1122

retval = get_managed_objects(MockObjectManager())

1123

self.assertDictEqual(managed_objects, retval)

1124

1125

def test_logs_and_exits_on_dbus_error(self):

1126

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

1127

1128

class MockObjectManagerFailing(object):

1129

def GetManagedObjects(self):

1130

dbus_logger.error("Test")

1131

raise dbus.exceptions.DBusException("Test")

1132

1133

class CountingHandler(logging.Handler):

1134

count = 0

1135

def emit(self, record):

1136

self.count += 1

1137

1138

counting_handler = CountingHandler()

1139

1140

dbus_logger.addHandler(counting_handler)

1141

1142

try:

1143

with self.assertLogs(log, logging.CRITICAL) as watcher:

1144

with self.assertRaises(SystemExit) as e:

1145

get_managed_objects(MockObjectManagerFailing())

1146

finally:

1147

dbus_logger.removeFilter(counting_handler)

1148

1149

# Make sure the dbus logger was suppressed

1150

self.assertEqual(counting_handler.count, 0)

1151

1152

# Test that the dbus_logger still works

1153

with self.assertLogs(dbus_logger, logging.ERROR):

1154

dbus_logger.error("Test")

1155

1156

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

1157

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

1158

else:

1159

self.assertIsNotNone(e.exception.code)

1160

1161

1162

class Test_commands_from_options(unittest.TestCase):

1163

def setUp(self):

1164

self.parser = argparse.ArgumentParser()

1165

add_command_line_options(self.parser)

1166

1167

def test_is_enabled(self):

1168

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

1169

command.IsEnabled)

1170

1171

def assert_command_from_args(self, args, command_cls,

1172

**cmd_attrs):

1173

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

1174

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

1175

options = self.parser.parse_args(args)

1176

check_option_syntax(self.parser, options)

1177

commands = commands_from_options(options)

1178

self.assertEqual(len(commands), 1)

1179

command = commands[0]

1180

self.assertIsInstance(command, command_cls)

1181

for key, value in cmd_attrs.items():

1182

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

1183

1184

def test_is_enabled_short(self):

1185

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

1186

command.IsEnabled)

1187

1188

def test_approve(self):

1189

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

1190

command.Approve)

1191

1192

def test_approve_short(self):

1193

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

1194

1195

def test_deny(self):

1196

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

1197

1198

def test_deny_short(self):

1199

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

1200

1201

def test_remove(self):

1202

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

1203

command.Remove)

1204

1205

def test_deny_before_remove(self):

1206

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

1207

"foo"])

1208

check_option_syntax(self.parser, options)

1209

commands = commands_from_options(options)

1210

self.assertEqual(len(commands), 2)

1211

self.assertIsInstance(commands[0], command.Deny)

1212

self.assertIsInstance(commands[1], command.Remove)

1213

1214

def test_deny_before_remove_reversed(self):

1215

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

1216

"--all"])

1217

check_option_syntax(self.parser, options)

1218

commands = commands_from_options(options)

1219

self.assertEqual(len(commands), 2)

1220

self.assertIsInstance(commands[0], command.Deny)

1221

self.assertIsInstance(commands[1], command.Remove)

1222

1223

def test_remove_short(self):

1224

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

1225

1226

def test_dump_json(self):

1227

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

1228

command.DumpJSON)

1229

1230

def test_enable(self):

1231

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

1232

command.Enable)

1233

1234

def test_enable_short(self):

1235

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

1236

1237

def test_disable(self):

1238

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

1239

command.Disable)

1240

1241

def test_disable_short(self):

1242

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

1243

1244

def test_bump_timeout(self):

1245

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

1246

command.BumpTimeout)

1247

1248

def test_bump_timeout_short(self):

1249

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

1250

command.BumpTimeout)

1251

1252

def test_start_checker(self):

1253

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

1254

command.StartChecker)

1255

1256

def test_stop_checker(self):

1257

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

1258

command.StopChecker)

1259

1260

def test_approve_by_default(self):

1261

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

1262

command.ApproveByDefault)

1263

1264

def test_deny_by_default(self):

1265

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

1266

command.DenyByDefault)

1267

1268

def test_checker(self):

1269

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

1270

command.SetChecker,

1271

value_to_set=":")

1272

1273

def test_checker_empty(self):

1274

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

1275

command.SetChecker,

1276

value_to_set="")

1277

1278

def test_checker_short(self):

1279

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

1280

command.SetChecker,

1281

value_to_set=":")

1282

1283

def test_host(self):

1284

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

1285

"foo"], command.SetHost,

1286

value_to_set="foo.example.org")

1287

1288

def test_host_short(self):

1289

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

1290

"foo"], command.SetHost,

1291

value_to_set="foo.example.org")

1292

1293

def test_secret_devnull(self):

1294

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

1295

"foo"], command.SetSecret,

1296

value_to_set=b"")

1297

1298

def test_secret_tempfile(self):

1299

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

1300

value = b"secret\0xyzzy\nbar"

1301

f.write(value)

1302

f.seek(0)

1303

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

1304

"foo"], command.SetSecret,

1305

value_to_set=value)

1306

1307

def test_secret_devnull_short(self):

1308

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

1309

command.SetSecret,

1310

value_to_set=b"")

1311

1312

def test_secret_tempfile_short(self):

1313

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

1314

value = b"secret\0xyzzy\nbar"

1315

f.write(value)

1316

f.seek(0)

1317

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

1318

command.SetSecret,

1319

value_to_set=value)

1320

1321

def test_timeout(self):

1322

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

1323

command.SetTimeout,

1324

value_to_set=300000)

1325

1326

def test_timeout_short(self):

1327

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

1328

command.SetTimeout,

1329

value_to_set=300000)

1330

1331

def test_extended_timeout(self):

1332

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

1333

"foo"],

1334

command.SetExtendedTimeout,

1335

value_to_set=900000)

1336

1337

def test_interval(self):

1338

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

1339

command.SetInterval,

1340

value_to_set=120000)

1341

1342

def test_interval_short(self):

1343

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

1344

command.SetInterval,

1345

value_to_set=120000)

1346

1347

def test_approval_delay(self):

1348

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

1349

"foo"],

1350

command.SetApprovalDelay,

1351

value_to_set=30000)

1352

1353

def test_approval_duration(self):

1354

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

1355

"foo"],

1356

command.SetApprovalDuration,

1357

value_to_set=1000)

1358

1359

def test_print_table(self):

1360

self.assert_command_from_args([], command.PrintTable,

1361

verbose=False)

1362

1363

def test_print_table_verbose(self):

1364

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

1365

command.PrintTable,

1366

verbose=True)

1367

1368

def test_print_table_verbose_short(self):

1369

self.assert_command_from_args(["-v"], command.PrintTable,

1370

verbose=True)

1371

1372

1373

class TestCommand(unittest.TestCase):

1374

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

1375

1376

def setUp(self):

1377

testcase = self

1378

class MockClient(object):

1379

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

1380

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

1381

self.attributes = attributes

1382

self.attributes["Name"] = name

1383

self.calls = []

1384

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

1385

testcase.assertEqual(interface, client_dbus_interface)

1386

testcase.assertEqual(dbus_interface,

1387

dbus.PROPERTIES_IFACE)

1388

self.attributes[propname] = value

1389

def Approve(self, approve, dbus_interface):

1390

testcase.assertEqual(dbus_interface,

1391

client_dbus_interface)

1392

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

1393

dbus_interface)))

1394

self.client = MockClient(

1395

"foo",

1396

KeyID=("92ed150794387c03ce684574b1139a65"

1397

"94a34f895daaaf09fd8ea90a27cddb12"),

1398

Secret=b"secret",

1399

Host="foo.example.org",

1400

Enabled=dbus.Boolean(True),

1401

Timeout=300000,

1402

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

1403

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

1404

Interval=120000,

1405

Fingerprint=("778827225BA7DE539C5A"

1406

"7CFA59CFF7CDBD9A5920"),

1407

CheckerRunning=dbus.Boolean(False),

1408

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

1409

ApprovalPending=dbus.Boolean(False),

1410

ApprovedByDefault=dbus.Boolean(True),

1411

LastApprovalRequest="",

1412

ApprovalDelay=0,

1413

ApprovalDuration=1000,

1414

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

1415

ExtendedTimeout=900000,

1416

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

1417

LastCheckerStatus=0)

1418

self.other_client = MockClient(

1419

"barbar",

1420

KeyID=("0558568eedd67d622f5c83b35a115f79"

1421

"6ab612cff5ad227247e46c2b020f441c"),

1422

Secret=b"secretbar",

1423

Host="192.0.2.3",

1424

Enabled=dbus.Boolean(True),

1425

Timeout=300000,

1426

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

1427

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

1428

Interval=120000,

1429

Fingerprint=("3E393AEAEFB84C7E89E2"

1430

"F547B3A107558FCA3A27"),

1431

CheckerRunning=dbus.Boolean(True),

1432

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

1433

ApprovalPending=dbus.Boolean(False),

1434

ApprovedByDefault=dbus.Boolean(False),

1435

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

1436

ApprovalDelay=30000,

1437

ApprovalDuration=93785000,

1438

Checker=":",

1439

ExtendedTimeout=900000,

1440

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

1441

LastCheckerStatus=-2)

1442

self.clients = collections.OrderedDict(

1443

[

1444

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

1445

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

1446

])

1447

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

1448

1449

@property

1450

def bus(self):

1451

class Bus(object):

1452

@staticmethod

1453

def get_object(client_bus_name, path):

1454

self.assertEqual(client_bus_name, dbus_busname)

1455

return {

1456

# Note: "self" here is the TestCmd instance, not

1457

# the Bus instance, since this is a static method!

1458

"/clients/foo": self.client,

1459

"/clients/barbar": self.other_client,

1460

}[path]

1461

return Bus()

1462

1463

1464

class TestBaseCommands(TestCommand):

1465

1466

def test_IsEnabled(self):

1467

self.assertTrue(all(command.IsEnabled().is_enabled(client,

1468

properties)

1469

for client, properties

1470

in self.clients.items()))

1471

1472

def test_IsEnabled_run_exits_successfully(self):

1473

with self.assertRaises(SystemExit) as e:

1474

command.IsEnabled().run(self.one_client)

1475

if e.exception.code is not None:

1476

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

1477

else:

1478

self.assertIsNone(e.exception.code)

1479

1480

def test_IsEnabled_run_exits_with_failure(self):

1481

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

1482

with self.assertRaises(SystemExit) as e:

1483

command.IsEnabled().run(self.one_client)

1484

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

1485

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

1486

else:

1487

self.assertIsNotNone(e.exception.code)

1488

1489

def test_Approve(self):

1490

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

1491

for clientpath in self.clients:

1492

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

1493

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

1494

client.calls)

1495

1496

def test_Deny(self):

1497

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

1498

for clientpath in self.clients:

1499

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

1500

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

1501

client.calls)

1502

1503

def test_Remove(self):

1504

class MockMandos(object):

1505

def __init__(self):

1506

self.calls = []

1507

def RemoveClient(self, dbus_path):

1508

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

1509

mandos = MockMandos()

1510

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

1511

for clientpath in self.clients:

1512

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

1513

mandos.calls)

1514

1515

expected_json = {

1516

"foo": {

1517

"Name": "foo",

1518

"KeyID": ("92ed150794387c03ce684574b1139a65"

1519

"94a34f895daaaf09fd8ea90a27cddb12"),

1520

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

1521

"Enabled": True,

1522

"Timeout": 300000,

1523

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

1524

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

1525

"Interval": 120000,

1526

"Fingerprint": ("778827225BA7DE539C5A"

1527

"7CFA59CFF7CDBD9A5920"),

1528

"CheckerRunning": False,

1529

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

1530

"ApprovalPending": False,

1531

"ApprovedByDefault": True,

1532

"LastApprovalRequest": "",

1533

"ApprovalDelay": 0,

1534

"ApprovalDuration": 1000,

1535

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

1536

"ExtendedTimeout": 900000,

1537

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

1538

"LastCheckerStatus": 0,

1539

},

1540

"barbar": {

1541

"Name": "barbar",

1542

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1543

"6ab612cff5ad227247e46c2b020f441c"),

1544

"Host": "192.0.2.3",

1545

"Enabled": True,

1546

"Timeout": 300000,

1547

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

1548

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

1549

"Interval": 120000,

1550

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1551

"F547B3A107558FCA3A27"),

1552

"CheckerRunning": True,

1553

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

1554

"ApprovalPending": False,

1555

"ApprovedByDefault": False,

1556

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

1557

"ApprovalDelay": 30000,

1558

"ApprovalDuration": 93785000,

1559

"Checker": ":",

1560

"ExtendedTimeout": 900000,

1561

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

1562

"LastCheckerStatus": -2,

1563

},

1564

}

1565

1566

def test_DumpJSON_normal(self):

1567

output = command.DumpJSON().output(self.clients.values())

1568

json_data = json.loads(output)

1569

self.assertDictEqual(json_data, self.expected_json)

1570

1571

def test_DumpJSON_one_client(self):

1572

output = command.DumpJSON().output(self.one_client.values())

1573

json_data = json.loads(output)

1574

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

1575

self.assertDictEqual(json_data, expected_json)

1576

1577

def test_PrintTable_normal(self):

1578

output = command.PrintTable().output(self.clients.values())

1579

expected_output = "\n".join((

1580

"Name Enabled Timeout Last Successful Check",

1581

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

1582

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

1583

))

1584

self.assertEqual(output, expected_output)

1585

1586

def test_PrintTable_verbose(self):

1587

output = command.PrintTable(verbose=True).output(

1588

self.clients.values())

1589

columns = (

1590

(

1591

"Name ",

1592

"foo ",

1593

"barbar ",

1594

),(

1595

"Enabled ",

1596

"Yes ",

1597

"Yes ",

1598

),(

1599

"Timeout ",

1600

"00:05:00 ",

1601

"00:05:00 ",

1602

),(

1603

"Last Successful Check ",

1604

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

1605

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

1606

),(

1607

"Created ",

1608

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

1609

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

1610

),(

1611

"Interval ",

1612

"00:02:00 ",

1613

"00:02:00 ",

1614

),(

1615

"Host ",

1616

"foo.example.org ",

1617

"192.0.2.3 ",

1618

),(

1619

("Key ID "

1620

" "),

1621

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1622

"ea90a27cddb12 "),

1623

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1624

"46c2b020f441c "),

1625

),(

1626

"Fingerprint ",

1627

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1628

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1629

),(

1630

"Check Is Running ",

1631

"No ",

1632

"Yes ",

1633

),(

1634

"Last Enabled ",

1635

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

1636

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

1637

),(

1638

"Approval Is Pending ",

1639

"No ",

1640

"No ",

1641

),(

1642

"Approved By Default ",

1643

"Yes ",

1644

"No ",

1645

),(

1646

"Last Approval Request ",

1647

" ",

1648

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

1649

),(

1650

"Approval Delay ",

1651

"00:00:00 ",

1652

"00:00:30 ",

1653

),(

1654

"Approval Duration ",

1655

"00:00:01 ",

1656

"1T02:03:05 ",

1657

),(

1658

"Checker ",

1659

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

1660

": ",

1661

),(

1662

"Extended Timeout ",

1663

"00:15:00 ",

1664

"00:15:00 ",

1665

),(

1666

"Expires ",

1667

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

1668

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

1669

),(

1670

"Last Checker Status",

1671

"0 ",

1672

"-2 ",

1673

)

1674

)

1675

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

1676

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

1677

for rows in columns)

1678

for line in range(num_lines))

1679

self.assertEqual(output, expected_output)

1680

1681

def test_PrintTable_one_client(self):

1682

output = command.PrintTable().output(self.one_client.values())

1683

expected_output = "\n".join((

1684

"Name Enabled Timeout Last Successful Check",

1685

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

1686

))

1687

self.assertEqual(output, expected_output)

1688

1689

1690

class TestPropertyCmd(TestCommand):

1691

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

1692

def runTest(self):

1693

if not hasattr(self, "command"):

1694

return

1695

values_to_get = getattr(self, "values_to_get",

1696

self.values_to_set)

1697

for value_to_set, value_to_get in zip(self.values_to_set,

1698

values_to_get):

1699

for clientpath in self.clients:

1700

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

1701

old_value = client.attributes[self.propname]

1702

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

1703

self.run_command(value_to_set, self.clients)

1704

for clientpath in self.clients:

1705

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

1706

value = client.attributes[self.propname]

1707

self.assertNotIsInstance(value, self.Unique)

1708

self.assertEqual(value, value_to_get)

1709

1710

class Unique(object):

1711

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

1712

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

1713

1714

def run_command(self, value, clients):

1715

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

1716

1717

1718

class TestEnableCmd(TestPropertyCmd):

1719

command = command.Enable

1720

propname = "Enabled"

1721

values_to_set = [dbus.Boolean(True)]

1722

1723

1724

class TestDisableCmd(TestPropertyCmd):

1725

command = command.Disable

1726

propname = "Enabled"

1727

values_to_set = [dbus.Boolean(False)]

1728

1729

1730

class TestBumpTimeoutCmd(TestPropertyCmd):

1731

command = command.BumpTimeout

1732

propname = "LastCheckedOK"

1733

values_to_set = [""]

1734

1735

1736

class TestStartCheckerCmd(TestPropertyCmd):

1737

command = command.StartChecker

1738

propname = "CheckerRunning"

1739

values_to_set = [dbus.Boolean(True)]

1740

1741

1742

class TestStopCheckerCmd(TestPropertyCmd):

1743

command = command.StopChecker

1744

propname = "CheckerRunning"

1745

values_to_set = [dbus.Boolean(False)]

1746

1747

1748

class TestApproveByDefaultCmd(TestPropertyCmd):

1749

command = command.ApproveByDefault

1750

propname = "ApprovedByDefault"

1751

values_to_set = [dbus.Boolean(True)]

1752

1753

1754

class TestDenyByDefaultCmd(TestPropertyCmd):

1755

command = command.DenyByDefault

1756

propname = "ApprovedByDefault"

1757

values_to_set = [dbus.Boolean(False)]

1758

1759

1760

class TestPropertyValueCmd(TestPropertyCmd):

1761

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

1762

1763

def runTest(self):

1764

if type(self) is TestPropertyValueCmd:

1765

return

1766

return super(TestPropertyValueCmd, self).runTest()

1767

1768

def run_command(self, value, clients):

1769

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

1770

1771

1772

class TestSetCheckerCmd(TestPropertyValueCmd):

1773

command = command.SetChecker

1774

propname = "Checker"

1775

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

1776

1777

1778

class TestSetHostCmd(TestPropertyValueCmd):

1779

command = command.SetHost

1780

propname = "Host"

1781

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

1782

1783

1784

class TestSetSecretCmd(TestPropertyValueCmd):

1785

command = command.SetSecret

1786

propname = "Secret"

1787

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

1788

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

1789

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

1790

1791

1792

class TestSetTimeoutCmd(TestPropertyValueCmd):

1793

command = command.SetTimeout

1794

propname = "Timeout"

1795

values_to_set = [datetime.timedelta(),

1796

datetime.timedelta(minutes=5),

1797

datetime.timedelta(seconds=1),

1798

datetime.timedelta(weeks=1),

1799

datetime.timedelta(weeks=52)]

1800

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

1801

1802

1803

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1804

command = command.SetExtendedTimeout

1805

propname = "ExtendedTimeout"

1806

values_to_set = [datetime.timedelta(),

1807

datetime.timedelta(minutes=5),

1808

datetime.timedelta(seconds=1),

1809

datetime.timedelta(weeks=1),

1810

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

class TestSetIntervalCmd(TestPropertyValueCmd):

1815

command = command.SetInterval

1816

propname = "Interval"

1817

values_to_set = [datetime.timedelta(),

1818

datetime.timedelta(minutes=5),

1819

datetime.timedelta(seconds=1),

1820

datetime.timedelta(weeks=1),

1821

datetime.timedelta(weeks=52)]

1822

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

1823

1824

1825

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1826

command = command.SetApprovalDelay

1827

propname = "ApprovalDelay"

1828

values_to_set = [datetime.timedelta(),

1829

datetime.timedelta(minutes=5),

1830

datetime.timedelta(seconds=1),

1831

datetime.timedelta(weeks=1),

1832

datetime.timedelta(weeks=52)]

1833

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

1834

1835

1836

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1837

command = command.SetApprovalDuration

1838

propname = "ApprovalDuration"

1839

values_to_set = [datetime.timedelta(),

1840

datetime.timedelta(minutes=5),

1841

datetime.timedelta(seconds=1),

1842

datetime.timedelta(weeks=1),

1843

datetime.timedelta(weeks=52)]

1844

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

1845

1846

1847

1848

def should_only_run_tests():

1849

parser = argparse.ArgumentParser(add_help=False)

1850

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

1851

args, unknown_args = parser.parse_known_args()

1852

run_tests = args.check

1853

if run_tests:

1854

# Remove --check argument from sys.argv

1855

sys.argv[1:] = unknown_args

1856

return run_tests

1857

1858

# Add all tests from doctest strings

1859

def load_tests(loader, tests, none):

1860

import doctest

1861

tests.addTests(doctest.DocTestSuite())

1862

return tests

1863

1864

if __name__ == "__main__":

1865

try:

1866

if should_only_run_tests():

1867

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

1868

unittest.main()

1869

else:

1870

main()

1871

finally:

1872

logging.shutdown()