/mandos/trunk : revision 1069

1

#!/usr/bin/python

2

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

3

#

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 program is free software: you can redistribute it and/or modify

10

# it under the terms of the GNU General Public License as published by

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

11

13

# the Free Software Foundation, either version 3 of the License, or

12

14

# (at your option) any later version.

13

15

#

14

# This program is distributed in the hope that it will be useful,

15

# but WITHOUT ANY WARRANTY; without even the implied warranty of

16

# Mandos is distributed in the hope that it will be useful, but

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

# You should have received a copy of the GNU General Public License

20

# along with this program. If not, see <http://www.gnu.org/licenses/>.

21

#

22

# Contact the authors at <mandos@fukt.bsnet.se>.

23

#

22

# along with Mandos. If not, see <http://www.gnu.org/licenses/>.

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

24

26

25

27

from __future__ import (division, absolute_import, print_function,

26

28

unicode_literals)

27

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

28

35

import sys

29

import dbus

30

36

import argparse

31

37

import locale

32

38

import datetime

33

39

import re

34

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

35

64

36

65

locale.setlocale(locale.LC_ALL, "")

37

66

38

tablewords = {

39

"Name": "Name",

40

"Enabled": "Enabled",

41

"Timeout": "Timeout",

42

"LastCheckedOK": "Last Successful Check",

43

"LastApprovalRequest": "Last Approval Request",

44

"Created": "Created",

45

"Interval": "Interval",

46

"Host": "Host",

47

"Fingerprint": "Fingerprint",

48

"CheckerRunning": "Check Is Running",

49

"LastEnabled": "Last Enabled",

50

"ApprovalPending": "Approval Is Pending",

51

"ApprovedByDefault": "Approved By Default",

52

"ApprovalDelay": "Approval Delay",

53

"ApprovalDuration": "Approval Duration",

54

"Checker": "Checker",

55

"ExtendedTimeout" : "Extended Timeout"

56

}

57

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

58

domain = "se.recompile"

59

busname = domain + ".Mandos"

60

server_path = "/"

61

server_interface = domain + ".Mandos"

62

client_interface = domain + ".Mandos.Client"

63

version = "1.3.1"

64

65

def timedelta_to_milliseconds(td):

66

"""Convert a datetime.timedelta object to milliseconds"""

67

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

68

+ (td.seconds * 1000)

69

+ (td.microseconds // 1000))

70

71

def milliseconds_to_string(ms):

72

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

73

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

74

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

75

"hours": td.seconds // 3600,

76

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

77

"seconds": td.seconds % 60,

78

})

79

80

def string_to_delta(interval):

81

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

82

83

>>> string_to_delta("7d")

84

datetime.timedelta(7)

85

>>> string_to_delta("60s")

86

datetime.timedelta(0, 60)

87

>>> string_to_delta("60m")

88

datetime.timedelta(0, 3600)

89

>>> string_to_delta("24h")

90

datetime.timedelta(1)

91

>>> string_to_delta("1w")

92

datetime.timedelta(7)

93

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

94

datetime.timedelta(0, 330)

95

"""

96

timevalue = datetime.timedelta(0)

97

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

98

99

for s in regexp.findall(interval):

100

try:

101

suffix = unicode(s[-1])

102

value = int(s[:-1])

103

if suffix == "d":

104

delta = datetime.timedelta(value)

105

elif suffix == "s":

106

delta = datetime.timedelta(0, value)

107

elif suffix == "m":

108

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

109

elif suffix == "h":

110

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

111

elif suffix == "w":

112

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

113

else:

114

raise ValueError

115

except (ValueError, IndexError):

116

raise ValueError

117

timevalue += delta

118

return timevalue

119

120

def print_clients(clients, keywords):

121

def valuetostring(value, keyword):

122

if type(value) is dbus.Boolean:

123

return "Yes" if value else "No"

124

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

125

"ApprovalDuration"):

126

return milliseconds_to_string(value)

127

return unicode(value)

128

129

# Create format string to print table rows

130

format_string = " ".join("%%-%ds" %

131

max(len(tablewords[key]),

132

max(len(valuetostring(client[key],

133

key))

134

for client in

135

clients))

136

for key in keywords)

137

# Print header line

138

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

139

for client in clients:

140

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

141

for key in keywords))

142

143

def has_actions(options):

144

return any((options.enable,

145

options.disable,

146

options.bump_timeout,

147

options.start_checker,

148

options.stop_checker,

149

options.is_enabled,

150

options.remove,

151

options.checker is not None,

152

options.timeout is not None,

153

options.extended_timeout is not None,

154

options.interval is not None,

155

options.approved_by_default is not None,

156

options.approval_delay is not None,

157

options.approval_duration is not None,

158

options.host is not None,

159

options.secret is not None,

160

options.approve,

161

options.deny))

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

162

81

163

82

def main():

164

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

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

106

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

107

try:

108

with SilenceLogger("dbus.proxies"):

109

managed_objects = (mandos_serv_object_manager

110

.GetManagedObjects())

111

except dbus.exceptions.DBusException as e:

112

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

113

"\n%s", e)

114

sys.exit(1)

115

116

all_clients = {}

117

for path, ifs_and_props in managed_objects.items():

118

try:

119

all_clients[path] = ifs_and_props[client_dbus_interface]

120

except KeyError:

121

pass

122

123

# Compile dict of (clientpath: properties) to process

124

if not clientnames:

125

clients = all_clients

126

else:

127

clients = {}

128

for name in clientnames:

129

for objpath, properties in all_clients.items():

130

if properties["Name"] == name:

131

clients[objpath] = properties

132

break

133

else:

134

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

135

sys.exit(1)

136

137

# Run all commands on clients

138

commands = commands_from_options(options)

139

for command in commands:

140

command.run(clients, bus, mandos_serv)

141

142

143

def add_command_line_options(parser):

165

144

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

166

version = "%%prog %s" % version,

145

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

167

146

help="show version number and exit")

168

147

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

169

148

help="Select all clients")

170

149

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

171

150

help="Print all fields")

172

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

173

help="Enable client")

174

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

175

help="disable client")

151

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

152

help="Dump client data in JSON format")

153

enable_disable = parser.add_mutually_exclusive_group()

154

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

155

help="Enable client")

156

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

157

action="store_true",

158

help="disable client")

176

159

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

177

160

help="Bump timeout for client")

178

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

179

help="Start checker for client")

180

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

181

help="Stop checker for client")

161

start_stop_checker = parser.add_mutually_exclusive_group()

162

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

163

action="store_true",

164

help="Start checker for client")

165

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

166

action="store_true",

167

help="Stop checker for client")

182

168

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

183

169

help="Check if client is enabled")

184

170

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

185

171

help="Remove client")

186

172

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

187

173

help="Set checker command for client")

188

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

174

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

189

175

help="Set timeout for client")

190

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

176

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

191

177

help="Set extended timeout for client")

192

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

178

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

193

179

help="Set checker interval for client")

194

parser.add_argument("--approve-by-default", action="store_true",

195

default=None, dest="approved_by_default",

196

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

197

parser.add_argument("--deny-by-default", action="store_false",

198

dest="approved_by_default",

199

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

200

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

180

approve_deny_default = parser.add_mutually_exclusive_group()

181

approve_deny_default.add_argument(

182

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

183

default=None, dest="approved_by_default",

184

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

185

approve_deny_default.add_argument(

186

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

187

dest="approved_by_default",

188

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

189

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

201

190

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

202

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

191

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

203

192

help="Set duration of one client approval")

204

193

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

205

parser.add_argument("-s", "--secret", type=file,

194

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

195

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

206

196

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

207

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

208

help="Approve any current client request")

209

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

210

help="Deny any current client request")

197

approve_deny = parser.add_mutually_exclusive_group()

198

approve_deny.add_argument(

199

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

200

help="Approve any current client request")

201

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

202

help="Deny any current client request")

203

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

204

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

205

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

206

help="Run self-test")

211

207

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

212

options = parser.parse_args()

213

214

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

208

209

210

def string_to_delta(interval):

211

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

212

213

try:

214

return rfc3339_duration_to_delta(interval)

215

except ValueError as e:

216

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

217

' '.join(e.args))

218

return parse_pre_1_6_1_interval(interval)

219

220

221

def rfc3339_duration_to_delta(duration):

222

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

223

224

>>> rfc3339_duration_to_delta("P7D")

225

datetime.timedelta(7)

226

>>> rfc3339_duration_to_delta("PT60S")

227

datetime.timedelta(0, 60)

228

>>> rfc3339_duration_to_delta("PT60M")

229

datetime.timedelta(0, 3600)

230

>>> rfc3339_duration_to_delta("P60M")

231

datetime.timedelta(1680)

232

>>> rfc3339_duration_to_delta("PT24H")

233

datetime.timedelta(1)

234

>>> rfc3339_duration_to_delta("P1W")

235

datetime.timedelta(7)

236

>>> rfc3339_duration_to_delta("PT5M30S")

237

datetime.timedelta(0, 330)

238

>>> rfc3339_duration_to_delta("P1DT3M20S")

239

datetime.timedelta(1, 200)

240

>>> # Can not be empty:

241

>>> rfc3339_duration_to_delta("")

242

Traceback (most recent call last):

243

...

244

ValueError: Invalid RFC 3339 duration: u''

245

>>> # Must start with "P":

246

>>> rfc3339_duration_to_delta("1D")

247

Traceback (most recent call last):

248

...

249

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

250

>>> # Must use correct order

251

>>> rfc3339_duration_to_delta("PT1S2M")

252

Traceback (most recent call last):

253

...

254

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

255

>>> # Time needs time marker

256

>>> rfc3339_duration_to_delta("P1H2S")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: u'P1H2S'

260

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

261

>>> rfc3339_duration_to_delta("P1D2W")

262

Traceback (most recent call last):

263

...

264

ValueError: Invalid RFC 3339 duration: u'P1D2W'

265

>>> rfc3339_duration_to_delta("P2W2H")

266

Traceback (most recent call last):

267

...

268

ValueError: Invalid RFC 3339 duration: u'P2W2H'

269

"""

270

271

# Parsing an RFC 3339 duration with regular expressions is not

272

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

273

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

274

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

275

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

276

# avoid excessive use of external libraries.

277

278

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

279

Token = collections.namedtuple("Token", (

280

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

281

# a "group" containing digits

282

"value", # datetime.timedelta or None

283

"followers")) # Tokens valid after this token

284

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

285

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

286

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

287

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

288

datetime.timedelta(seconds=1),

289

frozenset((token_end, )))

290

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

291

datetime.timedelta(minutes=1),

292

frozenset((token_second, token_end)))

293

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

294

datetime.timedelta(hours=1),

295

frozenset((token_minute, token_end)))

296

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

297

None,

298

frozenset((token_hour, token_minute,

299

token_second)))

300

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

301

datetime.timedelta(days=1),

302

frozenset((token_time, token_end)))

303

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

304

datetime.timedelta(weeks=4),

305

frozenset((token_day, token_end)))

306

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

307

datetime.timedelta(weeks=52),

308

frozenset((token_month, token_end)))

309

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

310

datetime.timedelta(weeks=1),

311

frozenset((token_end, )))

312

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

313

frozenset((token_year, token_month,

314

token_day, token_time,

315

token_week)))

316

# Define starting values:

317

# Value so far

318

value = datetime.timedelta()

319

found_token = None

320

# Following valid tokens

321

followers = frozenset((token_duration, ))

322

# String left to parse

323

s = duration

324

# Loop until end token is found

325

while found_token is not token_end:

326

# Search for any currently valid tokens

327

for token in followers:

328

match = token.regexp.match(s)

329

if match is not None:

330

# Token found

331

if token.value is not None:

332

# Value found, parse digits

333

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

334

# Add to value so far

335

value += factor * token.value

336

# Strip token from string

337

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

338

# Go to found token

339

found_token = token

340

# Set valid next tokens

341

followers = found_token.followers

342

break

343

else:

344

# No currently valid tokens were found

345

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

346

.format(duration))

347

# End token found

348

return value

349

350

351

def parse_pre_1_6_1_interval(interval):

352

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

353

and return a datetime.timedelta

354

355

>>> parse_pre_1_6_1_interval('7d')

356

datetime.timedelta(7)

357

>>> parse_pre_1_6_1_interval('60s')

358

datetime.timedelta(0, 60)

359

>>> parse_pre_1_6_1_interval('60m')

360

datetime.timedelta(0, 3600)

361

>>> parse_pre_1_6_1_interval('24h')

362

datetime.timedelta(1)

363

>>> parse_pre_1_6_1_interval('1w')

364

datetime.timedelta(7)

365

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

366

datetime.timedelta(0, 330)

367

>>> parse_pre_1_6_1_interval('')

368

datetime.timedelta(0)

369

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

370

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

371

datetime.timedelta(2, 480, 18000)

372

373

"""

374

375

value = datetime.timedelta(0)

376

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

377

378

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

379

if suffix == "d":

380

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

381

elif suffix == "s":

382

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

383

elif suffix == "m":

384

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

385

elif suffix == "h":

386

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

387

elif suffix == "w":

388

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

389

elif suffix == "":

390

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

391

return value

392

393

394

def check_option_syntax(parser, options):

395

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

396

argparse"""

397

398

def has_actions(options):

399

return any((options.enable,

400

options.disable,

401

options.bump_timeout,

402

options.start_checker,

403

options.stop_checker,

404

options.is_enabled,

405

options.remove,

406

options.checker is not None,

407

options.timeout is not None,

408

options.extended_timeout is not None,

409

options.interval is not None,

410

options.approved_by_default is not None,

411

options.approval_delay is not None,

412

options.approval_duration is not None,

413

options.host is not None,

414

options.secret is not None,

415

options.approve,

416

options.deny))

417

418

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

215

419

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

216

420

if options.verbose and has_actions(options):

217

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

218

" --all.")

421

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

422

if options.dump_json and (options.verbose

423

or has_actions(options)):

424

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

219

425

if options.all and not has_actions(options):

220

426

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

221

427

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

428

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

429

if options.remove:

430

options.remove = False

431

if has_actions(options) and not options.deny:

432

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

433

options.remove = True

434

435

436

def get_mandos_dbus_object(bus):

222

437

try:

223

bus = dbus.SystemBus()

224

mandos_dbus_objc = bus.get_object(busname, server_path)

438

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

439

dbus_busname, server_dbus_path)

440

mandos_dbus_object = bus.get_object(dbus_busname,

441

server_dbus_path)

225

442

except dbus.exceptions.DBusException:

226

print("Could not connect to Mandos server",

227

file=sys.stderr)

228

sys.exit(1)

229

230

mandos_serv = dbus.Interface(mandos_dbus_objc,

231

dbus_interface = server_interface)

232

233

#block stderr since dbus library prints to stderr

234

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

235

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

236

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

237

os.close(null)

238

try:

443

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

444

sys.exit(1)

445

446

return mandos_dbus_object

447

448

449

class SilenceLogger(object):

450

"Simple context manager to silence a particular logger"

451

def __init__(self, loggername):

452

self.logger = logging.getLogger(loggername)

453

454

def __enter__(self):

455

self.logger.addFilter(self.nullfilter)

456

return self

457

458

class NullFilter(logging.Filter):

459

def filter(self, record):

460

return False

461

462

nullfilter = NullFilter()

463

464

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

465

self.logger.removeFilter(self.nullfilter)

466

467

468

def commands_from_options(options):

469

470

commands = []

471

472

if options.is_enabled:

473

commands.append(IsEnabledCmd())

474

475

if options.approve:

476

commands.append(ApproveCmd())

477

478

if options.deny:

479

commands.append(DenyCmd())

480

481

if options.remove:

482

commands.append(RemoveCmd())

483

484

if options.dump_json:

485

commands.append(DumpJSONCmd())

486

487

if options.enable:

488

commands.append(EnableCmd())

489

490

if options.disable:

491

commands.append(DisableCmd())

492

493

if options.bump_timeout:

494

commands.append(BumpTimeoutCmd())

495

496

if options.start_checker:

497

commands.append(StartCheckerCmd())

498

499

if options.stop_checker:

500

commands.append(StopCheckerCmd())

501

502

if options.approved_by_default is not None:

503

if options.approved_by_default:

504

commands.append(ApproveByDefaultCmd())

505

else:

506

commands.append(DenyByDefaultCmd())

507

508

if options.checker is not None:

509

commands.append(SetCheckerCmd(options.checker))

510

511

if options.host is not None:

512

commands.append(SetHostCmd(options.host))

513

514

if options.secret is not None:

515

commands.append(SetSecretCmd(options.secret))

516

517

if options.timeout is not None:

518

commands.append(SetTimeoutCmd(options.timeout))

519

520

if options.extended_timeout:

521

commands.append(

522

SetExtendedTimeoutCmd(options.extended_timeout))

523

524

if options.interval is not None:

525

commands.append(SetIntervalCmd(options.interval))

526

527

if options.approval_delay is not None:

528

commands.append(SetApprovalDelayCmd(options.approval_delay))

529

530

if options.approval_duration is not None:

531

commands.append(

532

SetApprovalDurationCmd(options.approval_duration))

533

534

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

535

# optionally verbosely

536

if not commands:

537

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

538

539

return commands

540

541

542

class Command(object):

543

"""Abstract class for commands"""

544

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

545

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

546

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

547

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

548

self.mandos = mandos

549

for clientpath, properties in clients.items():

550

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

551

dbus_busname, str(clientpath))

552

client = bus.get_object(dbus_busname, clientpath)

553

self.run_on_one_client(client, properties)

554

555

556

class IsEnabledCmd(Command):

557

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

558

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

559

if self.is_enabled(client, properties):

560

sys.exit(0)

561

sys.exit(1)

562

def is_enabled(self, client, properties):

563

return properties["Enabled"]

564

565

566

class ApproveCmd(Command):

567

def run_on_one_client(self, client, properties):

568

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

569

client.__dbus_object_path__, client_dbus_interface)

570

client.Approve(dbus.Boolean(True),

571

dbus_interface=client_dbus_interface)

572

573

574

class DenyCmd(Command):

575

def run_on_one_client(self, client, properties):

576

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

577

client.__dbus_object_path__, client_dbus_interface)

578

client.Approve(dbus.Boolean(False),

579

dbus_interface=client_dbus_interface)

580

581

582

class RemoveCmd(Command):

583

def run_on_one_client(self, client, properties):

584

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

585

server_dbus_path, server_dbus_interface,

586

str(client.__dbus_object_path__))

587

self.mandos.RemoveClient(client.__dbus_object_path__)

588

589

590

class OutputCmd(Command):

591

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

592

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

593

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

594

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

595

"ApprovalPending", "ApprovedByDefault",

596

"LastApprovalRequest", "ApprovalDelay",

597

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

598

"Expires", "LastCheckerStatus")

599

600

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

601

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

602

603

def output(self, clients):

604

raise NotImplementedError()

605

606

607

class DumpJSONCmd(OutputCmd):

608

def output(self, clients):

609

data = {client["Name"]:

610

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

611

for key in self.all_keywords}

612

for client in clients}

613

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

614

615

@staticmethod

616

def dbus_boolean_to_bool(value):

617

if isinstance(value, dbus.Boolean):

618

value = bool(value)

619

return value

620

621

622

class PrintTableCmd(OutputCmd):

623

def __init__(self, verbose=False):

624

self.verbose = verbose

625

626

def output(self, clients):

627

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

628

"LastCheckedOK")

629

keywords = default_keywords

630

if self.verbose:

631

keywords = self.all_keywords

632

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

633

634

class TableOfClients(object):

635

tableheaders = {

636

"Name": "Name",

637

"Enabled": "Enabled",

638

"Timeout": "Timeout",

639

"LastCheckedOK": "Last Successful Check",

640

"LastApprovalRequest": "Last Approval Request",

641

"Created": "Created",

642

"Interval": "Interval",

643

"Host": "Host",

644

"Fingerprint": "Fingerprint",

645

"KeyID": "Key ID",

646

"CheckerRunning": "Check Is Running",

647

"LastEnabled": "Last Enabled",

648

"ApprovalPending": "Approval Is Pending",

649

"ApprovedByDefault": "Approved By Default",

650

"ApprovalDelay": "Approval Delay",

651

"ApprovalDuration": "Approval Duration",

652

"Checker": "Checker",

653

"ExtendedTimeout": "Extended Timeout",

654

"Expires": "Expires",

655

"LastCheckerStatus": "Last Checker Status",

656

}

657

658

def __init__(self, clients, keywords):

659

self.clients = clients

660

self.keywords = keywords

661

662

def __str__(self):

663

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

664

665

if sys.version_info.major == 2:

666

__unicode__ = __str__

667

def __str__(self):

668

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

669

670

def rows(self):

671

format_string = self.row_formatting_string()

672

rows = [self.header_line(format_string)]

673

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

674

for client in self.clients)

675

return rows

676

677

def row_formatting_string(self):

678

"Format string used to format table rows"

679

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

680

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

681

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

682

for client in self.clients)),

683

key=key)

684

for key in self.keywords)

685

686

def string_from_client(self, client, key):

687

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

688

689

@classmethod

690

def valuetostring(cls, value, keyword):

691

if isinstance(value, dbus.Boolean):

692

return "Yes" if value else "No"

693

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

694

"ApprovalDuration", "ExtendedTimeout"):

695

return cls.milliseconds_to_string(value)

696

return str(value)

697

698

def header_line(self, format_string):

699

return format_string.format(**self.tableheaders)

700

701

def client_line(self, client, format_string):

702

return format_string.format(

703

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

704

for key in self.keywords})

705

706

@staticmethod

707

def milliseconds_to_string(ms):

708

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

709

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

710

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

711

if td.days else "",

712

hours=td.seconds // 3600,

713

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

714

seconds=td.seconds % 60))

715

716

717

class PropertyCmd(Command):

718

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

719

720

def run_on_one_client(self, client, properties):

721

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

722

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

723

client.__dbus_object_path__,

724

dbus.PROPERTIES_IFACE, client_dbus_interface,

725

self.propname, self.value_to_set

726

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

727

else bool(self.value_to_set))

728

client.Set(client_dbus_interface, self.propname,

729

self.value_to_set,

730

dbus_interface=dbus.PROPERTIES_IFACE)

731

732

@property

733

def propname(self):

734

raise NotImplementedError()

735

736

737

class EnableCmd(PropertyCmd):

738

propname = "Enabled"

739

value_to_set = dbus.Boolean(True)

740

741

742

class DisableCmd(PropertyCmd):

743

propname = "Enabled"

744

value_to_set = dbus.Boolean(False)

745

746

747

class BumpTimeoutCmd(PropertyCmd):

748

propname = "LastCheckedOK"

749

value_to_set = ""

750

751

752

class StartCheckerCmd(PropertyCmd):

753

propname = "CheckerRunning"

754

value_to_set = dbus.Boolean(True)

755

756

757

class StopCheckerCmd(PropertyCmd):

758

propname = "CheckerRunning"

759

value_to_set = dbus.Boolean(False)

760

761

762

class ApproveByDefaultCmd(PropertyCmd):

763

propname = "ApprovedByDefault"

764

value_to_set = dbus.Boolean(True)

765

766

767

class DenyByDefaultCmd(PropertyCmd):

768

propname = "ApprovedByDefault"

769

value_to_set = dbus.Boolean(False)

770

771

772

class PropertyValueCmd(PropertyCmd):

773

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

774

def __init__(self, value):

775

self.value_to_set = value

776

777

778

class SetCheckerCmd(PropertyValueCmd):

779

propname = "Checker"

780

781

782

class SetHostCmd(PropertyValueCmd):

783

propname = "Host"

784

785

786

class SetSecretCmd(PropertyValueCmd):

787

propname = "Secret"

788

789

@property

790

def value_to_set(self):

791

return self._vts

792

793

@value_to_set.setter

794

def value_to_set(self, value):

795

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

796

self._vts = value.read()

797

value.close()

798

799

800

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

801

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

802

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

803

804

@property

805

def value_to_set(self):

806

return self._vts

807

808

@value_to_set.setter

809

def value_to_set(self, value):

810

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

811

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

812

813

814

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

815

propname = "Timeout"

816

817

818

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

819

propname = "ExtendedTimeout"

820

821

822

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

823

propname = "Interval"

824

825

826

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

827

propname = "ApprovalDelay"

828

829

830

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

831

propname = "ApprovalDuration"

832

833

834

835

class Test_string_to_delta(unittest.TestCase):

836

def test_handles_basic_rfc3339(self):

837

self.assertEqual(string_to_delta("PT0S"),

838

datetime.timedelta())

839

self.assertEqual(string_to_delta("P0D"),

840

datetime.timedelta())

841

self.assertEqual(string_to_delta("PT1S"),

842

datetime.timedelta(0, 1))

843

self.assertEqual(string_to_delta("PT2H"),

844

datetime.timedelta(0, 7200))

845

846

def test_falls_back_to_pre_1_6_1_with_warning(self):

847

# assertLogs only exists in Python 3.4

848

if hasattr(self, "assertLogs"):

849

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

850

value = string_to_delta("2h")

851

else:

852

class WarningFilter(logging.Filter):

853

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

854

def filter(self, record):

855

is_warning = record.levelno >= logging.WARNING

856

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

857

False)

858

return not is_warning

859

warning_filter = WarningFilter()

860

log.addFilter(warning_filter)

861

try:

862

value = string_to_delta("2h")

863

finally:

864

log.removeFilter(warning_filter)

865

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

866

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

867

868

869

class Test_check_option_syntax(unittest.TestCase):

870

def setUp(self):

871

self.parser = argparse.ArgumentParser()

872

add_command_line_options(self.parser)

873

874

def test_actions_requires_client_or_all(self):

875

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

876

options = self.parser.parse_args()

877

setattr(options, action, value)

878

with self.assertParseError():

879

self.check_option_syntax(options)

880

881

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

882

# check_option_syntax()

883

actions = {

884

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

885

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

886

# never used

887

"enable": True,

888

"disable": True,

889

"bump_timeout": True,

890

"start_checker": True,

891

"stop_checker": True,

892

"is_enabled": True,

893

"remove": True,

894

"checker": "x",

895

"timeout": datetime.timedelta(),

896

"extended_timeout": datetime.timedelta(),

897

"interval": datetime.timedelta(),

898

"approved_by_default": True,

899

"approval_delay": datetime.timedelta(),

900

"approval_duration": datetime.timedelta(),

901

"host": "x",

902

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

903

"approve": True,

904

"deny": True,

905

}

906

907

@contextlib.contextmanager

908

def assertParseError(self):

909

with self.assertRaises(SystemExit) as e:

910

with self.temporarily_suppress_stderr():

911

yield

912

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

913

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

914

# /argparse.html#exiting-methods

915

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

916

917

@staticmethod

918

@contextlib.contextmanager

919

def temporarily_suppress_stderr():

920

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

921

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

922

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

923

os.close(null)

239

924

try:

240

mandos_clients = mandos_serv.GetAllClientsWithProperties()

925

yield

241

926

finally:

242

#restore stderr

927

# restore stderr

243

928

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

244

929

os.close(stderrcopy)

245

except dbus.exceptions.DBusException, e:

246

print("Access denied: Accessing mandos server through dbus.",

247

file=sys.stderr)

248

sys.exit(1)

249

250

# Compile dict of (clients: properties) to process

251

clients={}

252

253

if options.all or not options.client:

254

clients = dict((bus.get_object(busname, path), properties)

255

for path, properties in

256

mandos_clients.iteritems())

257

else:

258

for name in options.client:

259

for path, client in mandos_clients.iteritems():

260

if client["Name"] == name:

261

client_objc = bus.get_object(busname, path)

262

clients[client_objc] = client

263

break

264

else:

265

print("Client not found on server: %r" % name,

266

file=sys.stderr)

267

sys.exit(1)

268

269

if not has_actions(options) and clients:

270

if options.verbose:

271

keywords = ("Name", "Enabled", "Timeout",

272

"LastCheckedOK", "Created", "Interval",

273

"Host", "Fingerprint", "CheckerRunning",

274

"LastEnabled", "ApprovalPending",

275

"ApprovedByDefault",

276

"LastApprovalRequest", "ApprovalDelay",

277

"ApprovalDuration", "Checker",

278

"ExtendedTimeout")

279

else:

280

keywords = defaultkeywords

281

282

print_clients(clients.values(), keywords)

283

else:

284

# Process each client in the list by all selected options

285

for client in clients:

286

if options.remove:

287

mandos_serv.RemoveClient(client.__dbus_object_path__)

288

if options.enable:

289

client.Enable(dbus_interface=client_interface)

290

if options.disable:

291

client.Disable(dbus_interface=client_interface)

292

if options.bump_timeout:

293

client.CheckedOK(dbus_interface=client_interface)

294

if options.start_checker:

295

client.StartChecker(dbus_interface=client_interface)

296

if options.stop_checker:

297

client.StopChecker(dbus_interface=client_interface)

298

if options.is_enabled:

299

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

300

"Enabled",

301

dbus_interface=

302

dbus.PROPERTIES_IFACE)

303

else 1)

304

if options.checker:

305

client.Set(client_interface, "Checker",

306

options.checker,

307

dbus_interface=dbus.PROPERTIES_IFACE)

308

if options.host:

309

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

310

dbus_interface=dbus.PROPERTIES_IFACE)

311

if options.interval:

312

client.Set(client_interface, "Interval",

313

timedelta_to_milliseconds

314

(string_to_delta(options.interval)),

315

dbus_interface=dbus.PROPERTIES_IFACE)

316

if options.approval_delay:

317

client.Set(client_interface, "ApprovalDelay",

318

timedelta_to_milliseconds

319

(string_to_delta(options.

320

approval_delay)),

321

dbus_interface=dbus.PROPERTIES_IFACE)

322

if options.approval_duration:

323

client.Set(client_interface, "ApprovalDuration",

324

timedelta_to_milliseconds

325

(string_to_delta(options.

326

approval_duration)),

327

dbus_interface=dbus.PROPERTIES_IFACE)

328

if options.timeout:

329

client.Set(client_interface, "Timeout",

330

timedelta_to_milliseconds

331

(string_to_delta(options.timeout)),

332

dbus_interface=dbus.PROPERTIES_IFACE)

333

if options.extended_timeout:

334

client.Set(client_interface, "ExtendedTimeout",

335

timedelta_to_milliseconds

336

(string_to_delta(options.extended_timeout)),

337

dbus_interface=dbus.PROPERTIES_IFACE)

338

if options.secret:

339

client.Set(client_interface, "Secret",

340

dbus.ByteArray(open(options.secret,

341

"rb").read()),

342

dbus_interface=dbus.PROPERTIES_IFACE)

343

if options.approved_by_default is not None:

344

client.Set(client_interface, "ApprovedByDefault",

345

dbus.Boolean(options

346

.approved_by_default),

347

dbus_interface=dbus.PROPERTIES_IFACE)

348

if options.approve:

349

client.Approve(dbus.Boolean(True),

350

dbus_interface=client_interface)

351

elif options.deny:

352

client.Approve(dbus.Boolean(False),

353

dbus_interface=client_interface)

930

931

def check_option_syntax(self, options):

932

check_option_syntax(self.parser, options)

933

934

def test_actions_conflicts_with_verbose(self):

935

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

936

options = self.parser.parse_args()

937

setattr(options, action, value)

938

options.verbose = True

939

with self.assertParseError():

940

self.check_option_syntax(options)

941

942

def test_dump_json_conflicts_with_verbose(self):

943

options = self.parser.parse_args()

944

options.dump_json = True

945

options.verbose = True

946

with self.assertParseError():

947

self.check_option_syntax(options)

948

949

def test_dump_json_conflicts_with_action(self):

950

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

951

options = self.parser.parse_args()

952

setattr(options, action, value)

953

options.dump_json = True

954

with self.assertParseError():

955

self.check_option_syntax(options)

956

957

def test_all_can_not_be_alone(self):

958

options = self.parser.parse_args()

959

options.all = True

960

with self.assertParseError():

961

self.check_option_syntax(options)

962

963

def test_all_is_ok_with_any_action(self):

964

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

965

options = self.parser.parse_args()

966

setattr(options, action, value)

967

options.all = True

968

self.check_option_syntax(options)

969

970

def test_is_enabled_fails_without_client(self):

971

options = self.parser.parse_args()

972

options.is_enabled = True

973

with self.assertParseError():

974

self.check_option_syntax(options)

975

976

def test_is_enabled_works_with_one_client(self):

977

options = self.parser.parse_args()

978

options.is_enabled = True

979

options.client = ["foo"]

980

self.check_option_syntax(options)

981

982

def test_is_enabled_fails_with_two_clients(self):

983

options = self.parser.parse_args()

984

options.is_enabled = True

985

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

986

with self.assertParseError():

987

self.check_option_syntax(options)

988

989

def test_remove_can_only_be_combined_with_action_deny(self):

990

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

991

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

992

continue

993

options = self.parser.parse_args()

994

setattr(options, action, value)

995

options.all = True

996

options.remove = True

997

with self.assertParseError():

998

self.check_option_syntax(options)

999

1000

1001

class Test_get_mandos_dbus_object(unittest.TestCase):

1002

def test_calls_and_returns_get_object_on_bus(self):

1003

class MockBus(object):

1004

called = False

1005

def get_object(mockbus_self, busname, dbus_path):

1006

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

1007

# this MockBus instance is in "mockbus_self".

1008

self.assertEqual(busname, dbus_busname)

1009

self.assertEqual(dbus_path, server_dbus_path)

1010

mockbus_self.called = True

1011

return mockbus_self

1012

1013

mockbus = get_mandos_dbus_object(bus=MockBus())

1014

self.assertIsInstance(mockbus, MockBus)

1015

self.assertTrue(mockbus.called)

1016

1017

def test_logs_and_exits_on_dbus_error(self):

1018

class MockBusFailing(object):

1019

def get_object(self, busname, dbus_path):

1020

raise dbus.exceptions.DBusException("Test")

1021

1022

# assertLogs only exists in Python 3.4

1023

if hasattr(self, "assertLogs"):

1024

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

1025

with self.assertRaises(SystemExit) as e:

1026

bus = get_mandos_dbus_object(bus=MockBus())

1027

else:

1028

critical_filter = self.CriticalFilter()

1029

log.addFilter(critical_filter)

1030

try:

1031

with self.assertRaises(SystemExit) as e:

1032

get_mandos_dbus_object(bus=MockBusFailing())

1033

finally:

1034

log.removeFilter(critical_filter)

1035

self.assertTrue(critical_filter.found)

1036

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

1037

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

1038

else:

1039

self.assertIsNotNone(e.exception.code)

1040

1041

class CriticalFilter(logging.Filter):

1042

"""Don't show, but register, critical messages"""

1043

found = False

1044

def filter(self, record):

1045

is_critical = record.levelno >= logging.CRITICAL

1046

self.found = is_critical or self.found

1047

return not is_critical

1048

1049

1050

class Test_SilenceLogger(unittest.TestCase):

1051

loggername = "mandos-ctl.Test_SilenceLogger"

1052

log = logging.getLogger(loggername)

1053

log.propagate = False

1054

log.addHandler(logging.NullHandler())

1055

1056

def setUp(self):

1057

self.counting_filter = self.CountingFilter()

1058

1059

class CountingFilter(logging.Filter):

1060

"Count number of records"

1061

count = 0

1062

def filter(self, record):

1063

self.count += 1

1064

return True

1065

1066

def test_should_filter_records_only_when_active(self):

1067

try:

1068

with SilenceLogger(self.loggername):

1069

self.log.addFilter(self.counting_filter)

1070

self.log.info("Filtered log message 1")

1071

self.log.info("Non-filtered message 2")

1072

self.log.info("Non-filtered message 3")

1073

finally:

1074

self.log.removeFilter(self.counting_filter)

1075

self.assertEqual(self.counting_filter.count, 2)

1076

1077

1078

class Test_commands_from_options(unittest.TestCase):

1079

def setUp(self):

1080

self.parser = argparse.ArgumentParser()

1081

add_command_line_options(self.parser)

1082

1083

def test_is_enabled(self):

1084

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

1085

IsEnabledCmd)

1086

1087

def assert_command_from_args(self, args, command_cls,

1088

**cmd_attrs):

1089

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

1090

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

1091

options = self.parser.parse_args(args)

1092

check_option_syntax(self.parser, options)

1093

commands = commands_from_options(options)

1094

self.assertEqual(len(commands), 1)

1095

command = commands[0]

1096

self.assertIsInstance(command, command_cls)

1097

for key, value in cmd_attrs.items():

1098

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

1099

1100

def test_is_enabled_short(self):

1101

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

1102

1103

def test_approve(self):

1104

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

1105

ApproveCmd)

1106

1107

def test_approve_short(self):

1108

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

1109

1110

def test_deny(self):

1111

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

1112

1113

def test_deny_short(self):

1114

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

1115

1116

def test_remove(self):

1117

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

1118

RemoveCmd)

1119

1120

def test_deny_before_remove(self):

1121

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

1122

"foo"])

1123

check_option_syntax(self.parser, options)

1124

commands = commands_from_options(options)

1125

self.assertEqual(len(commands), 2)

1126

self.assertIsInstance(commands[0], DenyCmd)

1127

self.assertIsInstance(commands[1], RemoveCmd)

1128

1129

def test_deny_before_remove_reversed(self):

1130

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

1131

"--all"])

1132

check_option_syntax(self.parser, options)

1133

commands = commands_from_options(options)

1134

self.assertEqual(len(commands), 2)

1135

self.assertIsInstance(commands[0], DenyCmd)

1136

self.assertIsInstance(commands[1], RemoveCmd)

1137

1138

def test_remove_short(self):

1139

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

1140

1141

def test_dump_json(self):

1142

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

1143

1144

def test_enable(self):

1145

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

1146

1147

def test_enable_short(self):

1148

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

1149

1150

def test_disable(self):

1151

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

1152

DisableCmd)

1153

1154

def test_disable_short(self):

1155

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

1156

1157

def test_bump_timeout(self):

1158

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

1159

BumpTimeoutCmd)

1160

1161

def test_bump_timeout_short(self):

1162

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

1163

1164

def test_start_checker(self):

1165

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

1166

StartCheckerCmd)

1167

1168

def test_stop_checker(self):

1169

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

1170

StopCheckerCmd)

1171

1172

def test_approve_by_default(self):

1173

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

1174

ApproveByDefaultCmd)

1175

1176

def test_deny_by_default(self):

1177

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

1178

DenyByDefaultCmd)

1179

1180

def test_checker(self):

1181

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

1182

SetCheckerCmd, value_to_set=":")

1183

1184

def test_checker_empty(self):

1185

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

1186

SetCheckerCmd, value_to_set="")

1187

1188

def test_checker_short(self):

1189

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

1190

SetCheckerCmd, value_to_set=":")

1191

1192

def test_host(self):

1193

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

1194

"foo"], SetHostCmd,

1195

value_to_set="foo.example.org")

1196

1197

def test_host_short(self):

1198

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

1199

"foo"], SetHostCmd,

1200

value_to_set="foo.example.org")

1201

1202

def test_secret_devnull(self):

1203

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

1204

"foo"], SetSecretCmd,

1205

value_to_set=b"")

1206

1207

def test_secret_tempfile(self):

1208

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

1209

value = b"secret\0xyzzy\nbar"

1210

f.write(value)

1211

f.seek(0)

1212

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

1213

"foo"], SetSecretCmd,

1214

value_to_set=value)

1215

1216

def test_secret_devnull_short(self):

1217

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

1218

SetSecretCmd, value_to_set=b"")

1219

1220

def test_secret_tempfile_short(self):

1221

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

1222

value = b"secret\0xyzzy\nbar"

1223

f.write(value)

1224

f.seek(0)

1225

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

1226

SetSecretCmd,

1227

value_to_set=value)

1228

1229

def test_timeout(self):

1230

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

1231

SetTimeoutCmd,

1232

value_to_set=300000)

1233

1234

def test_timeout_short(self):

1235

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

1236

SetTimeoutCmd,

1237

value_to_set=300000)

1238

1239

def test_extended_timeout(self):

1240

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

1241

"foo"],

1242

SetExtendedTimeoutCmd,

1243

value_to_set=900000)

1244

1245

def test_interval(self):

1246

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

1247

SetIntervalCmd,

1248

value_to_set=120000)

1249

1250

def test_interval_short(self):

1251

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

1252

SetIntervalCmd,

1253

value_to_set=120000)

1254

1255

def test_approval_delay(self):

1256

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

1257

"foo"], SetApprovalDelayCmd,

1258

value_to_set=30000)

1259

1260

def test_approval_duration(self):

1261

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

1262

"foo"], SetApprovalDurationCmd,

1263

value_to_set=1000)

1264

1265

def test_print_table(self):

1266

self.assert_command_from_args([], PrintTableCmd,

1267

verbose=False)

1268

1269

def test_print_table_verbose(self):

1270

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

1271

verbose=True)

1272

1273

def test_print_table_verbose_short(self):

1274

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

1275

verbose=True)

1276

1277

1278

class TestCmd(unittest.TestCase):

1279

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

1280

1281

def setUp(self):

1282

testcase = self

1283

class MockClient(object):

1284

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

1285

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

1286

self.attributes = attributes

1287

self.attributes["Name"] = name

1288

self.calls = []

1289

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

1290

testcase.assertEqual(interface, client_dbus_interface)

1291

testcase.assertEqual(dbus_interface,

1292

dbus.PROPERTIES_IFACE)

1293

self.attributes[propname] = value

1294

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

1295

testcase.assertEqual(interface, client_dbus_interface)

1296

testcase.assertEqual(dbus_interface,

1297

dbus.PROPERTIES_IFACE)

1298

return self.attributes[propname]

1299

def Approve(self, approve, dbus_interface):

1300

testcase.assertEqual(dbus_interface,

1301

client_dbus_interface)

1302

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

1303

dbus_interface)))

1304

self.client = MockClient(

1305

"foo",

1306

KeyID=("92ed150794387c03ce684574b1139a65"

1307

"94a34f895daaaf09fd8ea90a27cddb12"),

1308

Secret=b"secret",

1309

Host="foo.example.org",

1310

Enabled=dbus.Boolean(True),

1311

Timeout=300000,

1312

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

1313

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

1314

Interval=120000,

1315

Fingerprint=("778827225BA7DE539C5A"

1316

"7CFA59CFF7CDBD9A5920"),

1317

CheckerRunning=dbus.Boolean(False),

1318

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

1319

ApprovalPending=dbus.Boolean(False),

1320

ApprovedByDefault=dbus.Boolean(True),

1321

LastApprovalRequest="",

1322

ApprovalDelay=0,

1323

ApprovalDuration=1000,

1324

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

1325

ExtendedTimeout=900000,

1326

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

1327

LastCheckerStatus=0)

1328

self.other_client = MockClient(

1329

"barbar",

1330

KeyID=("0558568eedd67d622f5c83b35a115f79"

1331

"6ab612cff5ad227247e46c2b020f441c"),

1332

Secret=b"secretbar",

1333

Host="192.0.2.3",

1334

Enabled=dbus.Boolean(True),

1335

Timeout=300000,

1336

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

1337

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

1338

Interval=120000,

1339

Fingerprint=("3E393AEAEFB84C7E89E2"

1340

"F547B3A107558FCA3A27"),

1341

CheckerRunning=dbus.Boolean(True),

1342

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

1343

ApprovalPending=dbus.Boolean(False),

1344

ApprovedByDefault=dbus.Boolean(False),

1345

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

1346

ApprovalDelay=30000,

1347

ApprovalDuration=93785000,

1348

Checker=":",

1349

ExtendedTimeout=900000,

1350

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

1351

LastCheckerStatus=-2)

1352

self.clients = collections.OrderedDict(

1353

[

1354

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

1355

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

1356

])

1357

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

1358

1359

@property

1360

def bus(self):

1361

class Bus(object):

1362

@staticmethod

1363

def get_object(client_bus_name, path):

1364

self.assertEqual(client_bus_name, dbus_busname)

1365

return {

1366

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

1367

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

1368

"/clients/foo": self.client,

1369

"/clients/barbar": self.other_client,

1370

}[path]

1371

return Bus()

1372

1373

1374

class TestIsEnabledCmd(TestCmd):

1375

def test_is_enabled(self):

1376

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

1377

properties)

1378

for client, properties

1379

in self.clients.items()))

1380

1381

def test_is_enabled_run_exits_successfully(self):

1382

with self.assertRaises(SystemExit) as e:

1383

IsEnabledCmd().run(self.one_client)

1384

if e.exception.code is not None:

1385

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

1386

else:

1387

self.assertIsNone(e.exception.code)

1388

1389

def test_is_enabled_run_exits_with_failure(self):

1390

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

1391

with self.assertRaises(SystemExit) as e:

1392

IsEnabledCmd().run(self.one_client)

1393

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

1394

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

1395

else:

1396

self.assertIsNotNone(e.exception.code)

1397

1398

1399

class TestApproveCmd(TestCmd):

1400

def test_approve(self):

1401

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

1402

for clientpath in self.clients:

1403

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

1404

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

1405

client.calls)

1406

1407

1408

class TestDenyCmd(TestCmd):

1409

def test_deny(self):

1410

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

1411

for clientpath in self.clients:

1412

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

1413

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

1414

client.calls)

1415

1416

1417

class TestRemoveCmd(TestCmd):

1418

def test_remove(self):

1419

class MockMandos(object):

1420

def __init__(self):

1421

self.calls = []

1422

def RemoveClient(self, dbus_path):

1423

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

1424

mandos = MockMandos()

1425

super(TestRemoveCmd, self).setUp()

1426

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

1427

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

1428

for clientpath in self.clients:

1429

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

1430

mandos.calls)

1431

1432

1433

class TestDumpJSONCmd(TestCmd):

1434

def setUp(self):

1435

self.expected_json = {

1436

"foo": {

1437

"Name": "foo",

1438

"KeyID": ("92ed150794387c03ce684574b1139a65"

1439

"94a34f895daaaf09fd8ea90a27cddb12"),

1440

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

1441

"Enabled": True,

1442

"Timeout": 300000,

1443

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

1444

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

1445

"Interval": 120000,

1446

"Fingerprint": ("778827225BA7DE539C5A"

1447

"7CFA59CFF7CDBD9A5920"),

1448

"CheckerRunning": False,

1449

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

1450

"ApprovalPending": False,

1451

"ApprovedByDefault": True,

1452

"LastApprovalRequest": "",

1453

"ApprovalDelay": 0,

1454

"ApprovalDuration": 1000,

1455

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

1456

"ExtendedTimeout": 900000,

1457

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

1458

"LastCheckerStatus": 0,

1459

},

1460

"barbar": {

1461

"Name": "barbar",

1462

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1463

"6ab612cff5ad227247e46c2b020f441c"),

1464

"Host": "192.0.2.3",

1465

"Enabled": True,

1466

"Timeout": 300000,

1467

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

1468

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

1469

"Interval": 120000,

1470

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1471

"F547B3A107558FCA3A27"),

1472

"CheckerRunning": True,

1473

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

1474

"ApprovalPending": False,

1475

"ApprovedByDefault": False,

1476

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

1477

"ApprovalDelay": 30000,

1478

"ApprovalDuration": 93785000,

1479

"Checker": ":",

1480

"ExtendedTimeout": 900000,

1481

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

1482

"LastCheckerStatus": -2,

1483

},

1484

}

1485

return super(TestDumpJSONCmd, self).setUp()

1486

1487

def test_normal(self):

1488

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

1489

json_data = json.loads(output)

1490

self.assertDictEqual(json_data, self.expected_json)

1491

1492

def test_one_client(self):

1493

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

1494

json_data = json.loads(output)

1495

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

1496

self.assertDictEqual(json_data, expected_json)

1497

1498

1499

class TestPrintTableCmd(TestCmd):

1500

def test_normal(self):

1501

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

1502

expected_output = "\n".join((

1503

"Name Enabled Timeout Last Successful Check",

1504

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

1505

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

1506

))

1507

self.assertEqual(output, expected_output)

1508

1509

def test_verbose(self):

1510

output = PrintTableCmd(verbose=True).output(

1511

self.clients.values())

1512

columns = (

1513

(

1514

"Name ",

1515

"foo ",

1516

"barbar ",

1517

),(

1518

"Enabled ",

1519

"Yes ",

1520

"Yes ",

1521

),(

1522

"Timeout ",

1523

"00:05:00 ",

1524

"00:05:00 ",

1525

),(

1526

"Last Successful Check ",

1527

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

1528

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

1529

),(

1530

"Created ",

1531

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

1532

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

1533

),(

1534

"Interval ",

1535

"00:02:00 ",

1536

"00:02:00 ",

1537

),(

1538

"Host ",

1539

"foo.example.org ",

1540

"192.0.2.3 ",

1541

),(

1542

("Key ID "

1543

" "),

1544

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1545

"ea90a27cddb12 "),

1546

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1547

"46c2b020f441c "),

1548

),(

1549

"Fingerprint ",

1550

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1551

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1552

),(

1553

"Check Is Running ",

1554

"No ",

1555

"Yes ",

1556

),(

1557

"Last Enabled ",

1558

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

1559

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

1560

),(

1561

"Approval Is Pending ",

1562

"No ",

1563

"No ",

1564

),(

1565

"Approved By Default ",

1566

"Yes ",

1567

"No ",

1568

),(

1569

"Last Approval Request ",

1570

" ",

1571

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

1572

),(

1573

"Approval Delay ",

1574

"00:00:00 ",

1575

"00:00:30 ",

1576

),(

1577

"Approval Duration ",

1578

"00:00:01 ",

1579

"1T02:03:05 ",

1580

),(

1581

"Checker ",

1582

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

1583

": ",

1584

),(

1585

"Extended Timeout ",

1586

"00:15:00 ",

1587

"00:15:00 ",

1588

),(

1589

"Expires ",

1590

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

1591

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

1592

),(

1593

"Last Checker Status",

1594

"0 ",

1595

"-2 ",

1596

)

1597

)

1598

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

1599

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

1600

for rows in columns)

1601

for line in range(num_lines))

1602

self.assertEqual(output, expected_output)

1603

1604

def test_one_client(self):

1605

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

1606

expected_output = "\n".join((

1607

"Name Enabled Timeout Last Successful Check",

1608

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

1609

))

1610

self.assertEqual(output, expected_output)

1611

1612

1613

class TestPropertyCmd(TestCmd):

1614

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

1615

def runTest(self):

1616

if not hasattr(self, "command"):

1617

return

1618

values_to_get = getattr(self, "values_to_get",

1619

self.values_to_set)

1620

for value_to_set, value_to_get in zip(self.values_to_set,

1621

values_to_get):

1622

for clientpath in self.clients:

1623

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

1624

old_value = client.attributes[self.propname]

1625

self.assertNotIsInstance(old_value, self.Unique)

1626

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

1627

self.run_command(value_to_set, self.clients)

1628

for clientpath in self.clients:

1629

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

1630

value = client.attributes[self.propname]

1631

self.assertNotIsInstance(value, self.Unique)

1632

self.assertEqual(value, value_to_get)

1633

1634

class Unique(object):

1635

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

1636

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

1637

1638

def run_command(self, value, clients):

1639

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

1640

1641

1642

class TestEnableCmd(TestPropertyCmd):

1643

command = EnableCmd

1644

propname = "Enabled"

1645

values_to_set = [dbus.Boolean(True)]

1646

1647

1648

class TestDisableCmd(TestPropertyCmd):

1649

command = DisableCmd

1650

propname = "Enabled"

1651

values_to_set = [dbus.Boolean(False)]

1652

1653

1654

class TestBumpTimeoutCmd(TestPropertyCmd):

1655

command = BumpTimeoutCmd

1656

propname = "LastCheckedOK"

1657

values_to_set = [""]

1658

1659

1660

class TestStartCheckerCmd(TestPropertyCmd):

1661

command = StartCheckerCmd

1662

propname = "CheckerRunning"

1663

values_to_set = [dbus.Boolean(True)]

1664

1665

1666

class TestStopCheckerCmd(TestPropertyCmd):

1667

command = StopCheckerCmd

1668

propname = "CheckerRunning"

1669

values_to_set = [dbus.Boolean(False)]

1670

1671

1672

class TestApproveByDefaultCmd(TestPropertyCmd):

1673

command = ApproveByDefaultCmd

1674

propname = "ApprovedByDefault"

1675

values_to_set = [dbus.Boolean(True)]

1676

1677

1678

class TestDenyByDefaultCmd(TestPropertyCmd):

1679

command = DenyByDefaultCmd

1680

propname = "ApprovedByDefault"

1681

values_to_set = [dbus.Boolean(False)]

1682

1683

1684

class TestPropertyValueCmd(TestPropertyCmd):

1685

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

1686

1687

def runTest(self):

1688

if type(self) is TestPropertyValueCmd:

1689

return

1690

return super(TestPropertyValueCmd, self).runTest()

1691

1692

def run_command(self, value, clients):

1693

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

1694

1695

1696

class TestSetCheckerCmd(TestPropertyValueCmd):

1697

command = SetCheckerCmd

1698

propname = "Checker"

1699

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

1700

1701

1702

class TestSetHostCmd(TestPropertyValueCmd):

1703

command = SetHostCmd

1704

propname = "Host"

1705

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

1706

1707

1708

class TestSetSecretCmd(TestPropertyValueCmd):

1709

command = SetSecretCmd

1710

propname = "Secret"

1711

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

1712

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

1713

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

1714

1715

1716

class TestSetTimeoutCmd(TestPropertyValueCmd):

1717

command = SetTimeoutCmd

1718

propname = "Timeout"

1719

values_to_set = [datetime.timedelta(),

1720

datetime.timedelta(minutes=5),

1721

datetime.timedelta(seconds=1),

1722

datetime.timedelta(weeks=1),

1723

datetime.timedelta(weeks=52)]

1724

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

1725

1726

1727

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1728

command = SetExtendedTimeoutCmd

1729

propname = "ExtendedTimeout"

1730

values_to_set = [datetime.timedelta(),

1731

datetime.timedelta(minutes=5),

1732

datetime.timedelta(seconds=1),

1733

datetime.timedelta(weeks=1),

1734

datetime.timedelta(weeks=52)]

1735

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

1736

1737

1738

class TestSetIntervalCmd(TestPropertyValueCmd):

1739

command = SetIntervalCmd

1740

propname = "Interval"

1741

values_to_set = [datetime.timedelta(),

1742

datetime.timedelta(minutes=5),

1743

datetime.timedelta(seconds=1),

1744

datetime.timedelta(weeks=1),

1745

datetime.timedelta(weeks=52)]

1746

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

1747

1748

1749

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1750

command = SetApprovalDelayCmd

1751

propname = "ApprovalDelay"

1752

values_to_set = [datetime.timedelta(),

1753

datetime.timedelta(minutes=5),

1754

datetime.timedelta(seconds=1),

1755

datetime.timedelta(weeks=1),

1756

datetime.timedelta(weeks=52)]

1757

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

1758

1759

1760

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1761

command = SetApprovalDurationCmd

1762

propname = "ApprovalDuration"

1763

values_to_set = [datetime.timedelta(),

1764

datetime.timedelta(minutes=5),

1765

datetime.timedelta(seconds=1),

1766

datetime.timedelta(weeks=1),

1767

datetime.timedelta(weeks=52)]

1768

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

1769

1770

1771

1772

def should_only_run_tests():

1773

parser = argparse.ArgumentParser(add_help=False)

1774

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

1775

args, unknown_args = parser.parse_known_args()

1776

run_tests = args.check

1777

if run_tests:

1778

# Remove --check argument from sys.argv

1779

sys.argv[1:] = unknown_args

1780

return run_tests

1781

1782

# Add all tests from doctest strings

1783

def load_tests(loader, tests, none):

1784

import doctest

1785

tests.addTests(doctest.DocTestSuite())

1786

return tests

354

1787

355

1788

if __name__ == "__main__":

356

main()

1789

try:

1790

if should_only_run_tests():

1791

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

1792

unittest.main()

1793

else:

1794

main()

1795

finally:

1796

logging.shutdown()