/mandos/trunk : revision 1089

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

import StringIO

65

io.StringIO = StringIO.StringIO

66

67

locale.setlocale(locale.LC_ALL, "")

68

69

dbus_busname_domain = "se.recompile"

70

dbus_busname = dbus_busname_domain + ".Mandos"

71

server_dbus_path = "/"

72

server_dbus_interface = dbus_busname_domain + ".Mandos"

73

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

74

del dbus_busname_domain

75

version = "1.8.3"

76

77

78

try:

79

dbus.OBJECT_MANAGER_IFACE

80

except AttributeError:

81

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

82

83

84

def main():

85

parser = argparse.ArgumentParser()

86

add_command_line_options(parser)

87

88

options = parser.parse_args()

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"

177

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

183

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

184

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

185

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

186

os.close(null)

187

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

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

464

class NullFilter(logging.Filter):

465

def filter(self, record):

466

return False

467

468

nullfilter = NullFilter()

469

470

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

471

self.logger.removeFilter(self.nullfilter)

472

473

474

def commands_from_options(options):

475

476

commands = []

477

478

if options.is_enabled:

479

commands.append(command.IsEnabled())

480

481

if options.approve:

482

commands.append(command.Approve())

483

484

if options.deny:

485

commands.append(command.Deny())

486

487

if options.remove:

488

commands.append(command.Remove())

489

490

if options.dump_json:

491

commands.append(command.DumpJSON())

492

493

if options.enable:

494

commands.append(command.Enable())

495

496

if options.disable:

497

commands.append(command.Disable())

498

499

if options.bump_timeout:

500

commands.append(command.BumpTimeout())

501

502

if options.start_checker:

503

commands.append(command.StartChecker())

504

505

if options.stop_checker:

506

commands.append(command.StopChecker())

507

508

if options.approved_by_default is not None:

509

if options.approved_by_default:

510

commands.append(command.ApproveByDefault())

511

else:

512

commands.append(command.DenyByDefault())

513

514

if options.checker is not None:

515

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

516

517

if options.host is not None:

518

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

519

520

if options.secret is not None:

521

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

522

523

if options.timeout is not None:

524

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

525

526

if options.extended_timeout:

527

commands.append(

528

command.SetExtendedTimeout(options.extended_timeout))

529

530

if options.interval is not None:

531

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

532

533

if options.approval_delay is not None:

534

commands.append(

535

command.SetApprovalDelay(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

command.SetApprovalDuration(options.approval_duration))

540

541

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

542

# optionally verbosely

543

if not commands:

544

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

545

546

return commands

547

548

549

class command(object):

550

"""A namespace for command classes"""

551

552

class Base(object):

553

"""Abstract base class for commands"""

554

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

555

"""Normal commands should implement run_on_one_client(),

556

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

557

override this run() method instead.

558

"""

559

for clientpath, properties in clients.items():

560

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

561

dbus_busname, str(clientpath))

562

client = bus.get_object(dbus_busname, clientpath)

563

self.run_on_one_client(client, properties)

564

565

566

class IsEnabled(Base):

567

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

568

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

569

if self.is_enabled(client, properties):

570

sys.exit(0)

571

sys.exit(1)

572

def is_enabled(self, client, properties):

573

return properties["Enabled"]

574

575

576

class Approve(Base):

577

def run_on_one_client(self, client, properties):

578

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

579

client.__dbus_object_path__,

580

client_dbus_interface)

581

client.Approve(dbus.Boolean(True),

582

dbus_interface=client_dbus_interface)

583

584

585

class Deny(Base):

586

def run_on_one_client(self, client, properties):

587

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

588

client.__dbus_object_path__,

589

client_dbus_interface)

590

client.Approve(dbus.Boolean(False),

591

dbus_interface=client_dbus_interface)

592

593

594

class Remove(Base):

595

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

596

for clientpath in clients.keys():

597

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

598

dbus_busname, server_dbus_path,

599

server_dbus_interface, clientpath)

600

mandos.RemoveClient(clientpath)

601

602

603

class Output(Base):

604

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

605

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

606

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

607

"Fingerprint", "CheckerRunning",

608

"LastEnabled", "ApprovalPending",

609

"ApprovedByDefault", "LastApprovalRequest",

610

"ApprovalDelay", "ApprovalDuration",

611

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

612

"LastCheckerStatus")

613

614

615

class DumpJSON(Output):

616

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

617

data = {client["Name"]:

618

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

619

for key in self.all_keywords}

620

for client in clients.values()}

621

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

622

623

@staticmethod

624

def dbus_boolean_to_bool(value):

625

if isinstance(value, dbus.Boolean):

626

value = bool(value)

627

return value

628

629

630

class PrintTable(Output):

631

def __init__(self, verbose=False):

632

self.verbose = verbose

633

634

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

635

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

636

"LastCheckedOK")

637

keywords = default_keywords

638

if self.verbose:

639

keywords = self.all_keywords

640

print(self.TableOfClients(clients.values(), keywords))

641

642

class TableOfClients(object):

643

tableheaders = {

644

"Name": "Name",

645

"Enabled": "Enabled",

646

"Timeout": "Timeout",

647

"LastCheckedOK": "Last Successful Check",

648

"LastApprovalRequest": "Last Approval Request",

649

"Created": "Created",

650

"Interval": "Interval",

651

"Host": "Host",

652

"Fingerprint": "Fingerprint",

653

"KeyID": "Key ID",

654

"CheckerRunning": "Check Is Running",

655

"LastEnabled": "Last Enabled",

656

"ApprovalPending": "Approval Is Pending",

657

"ApprovedByDefault": "Approved By Default",

658

"ApprovalDelay": "Approval Delay",

659

"ApprovalDuration": "Approval Duration",

660

"Checker": "Checker",

661

"ExtendedTimeout": "Extended Timeout",

662

"Expires": "Expires",

663

"LastCheckerStatus": "Last Checker Status",

664

}

665

666

def __init__(self, clients, keywords):

667

self.clients = clients

668

self.keywords = keywords

669

670

def __str__(self):

671

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

672

673

if sys.version_info.major == 2:

674

__unicode__ = __str__

675

def __str__(self):

676

return str(self).encode(

677

locale.getpreferredencoding())

678

679

def rows(self):

680

format_string = self.row_formatting_string()

681

rows = [self.header_line(format_string)]

682

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

683

for client in self.clients)

684

return rows

685

686

def row_formatting_string(self):

687

"Format string used to format table rows"

688

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

689

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

690

*(len(self.string_from_client(client,

691

key))

692

for client in self.clients)),

693

key=key)

694

for key in self.keywords)

695

696

def string_from_client(self, client, key):

697

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

698

699

@classmethod

700

def valuetostring(cls, value, keyword):

701

if isinstance(value, dbus.Boolean):

702

return "Yes" if value else "No"

703

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

704

"ApprovalDuration", "ExtendedTimeout"):

705

return cls.milliseconds_to_string(value)

706

return str(value)

707

708

def header_line(self, format_string):

709

return format_string.format(**self.tableheaders)

710

711

def client_line(self, client, format_string):

712

return format_string.format(

713

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

714

for key in self.keywords})

715

716

@staticmethod

717

def milliseconds_to_string(ms):

718

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

719

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

720

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

721

if td.days else "",

722

hours=td.seconds // 3600,

723

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

724

seconds=td.seconds % 60))

725

726

727

class PropertySetter(Base):

728

"Abstract class for Actions for setting one client property"

729

730

def run_on_one_client(self, client, properties):

731

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

732

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

733

client.__dbus_object_path__,

734

dbus.PROPERTIES_IFACE, client_dbus_interface,

735

self.propname, self.value_to_set

736

if not isinstance(self.value_to_set,

737

dbus.Boolean)

738

else bool(self.value_to_set))

739

client.Set(client_dbus_interface, self.propname,

740

self.value_to_set,

741

dbus_interface=dbus.PROPERTIES_IFACE)

742

743

@property

744

def propname(self):

745

raise NotImplementedError()

746

747

748

class Enable(PropertySetter):

749

propname = "Enabled"

750

value_to_set = dbus.Boolean(True)

751

752

753

class Disable(PropertySetter):

754

propname = "Enabled"

755

value_to_set = dbus.Boolean(False)

756

757

758

class BumpTimeout(PropertySetter):

759

propname = "LastCheckedOK"

760

value_to_set = ""

761

762

763

class StartChecker(PropertySetter):

764

propname = "CheckerRunning"

765

value_to_set = dbus.Boolean(True)

766

767

768

class StopChecker(PropertySetter):

769

propname = "CheckerRunning"

770

value_to_set = dbus.Boolean(False)

771

772

773

class ApproveByDefault(PropertySetter):

774

propname = "ApprovedByDefault"

775

value_to_set = dbus.Boolean(True)

776

777

778

class DenyByDefault(PropertySetter):

779

propname = "ApprovedByDefault"

780

value_to_set = dbus.Boolean(False)

781

782

783

class PropertySetterValue(PropertySetter):

784

"""Abstract class for PropertySetter recieving a value as

785

constructor argument instead of a class attribute."""

786

def __init__(self, value):

787

self.value_to_set = value

788

789

790

class SetChecker(PropertySetterValue):

791

propname = "Checker"

792

793

794

class SetHost(PropertySetterValue):

795

propname = "Host"

796

797

798

class SetSecret(PropertySetterValue):

799

propname = "Secret"

800

801

@property

802

def value_to_set(self):

803

return self._vts

804

805

@value_to_set.setter

806

def value_to_set(self, value):

807

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

808

self._vts = value.read()

809

value.close()

810

811

812

class PropertySetterValueMilliseconds(PropertySetterValue):

813

"""Abstract class for PropertySetterValue taking a value

814

argument as a datetime.timedelta() but should store it as

815

milliseconds."""

816

817

@property

818

def value_to_set(self):

819

return self._vts

820

821

@value_to_set.setter

822

def value_to_set(self, value):

823

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

824

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

825

826

827

class SetTimeout(PropertySetterValueMilliseconds):

828

propname = "Timeout"

829

830

831

class SetExtendedTimeout(PropertySetterValueMilliseconds):

832

propname = "ExtendedTimeout"

833

834

835

class SetInterval(PropertySetterValueMilliseconds):

836

propname = "Interval"

837

838

839

class SetApprovalDelay(PropertySetterValueMilliseconds):

840

propname = "ApprovalDelay"

841

842

843

class SetApprovalDuration(PropertySetterValueMilliseconds):

844

propname = "ApprovalDuration"

845

846

847

848

class TestCaseWithAssertLogs(unittest.TestCase):

849

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

850

851

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

852

@contextlib.contextmanager

853

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

854

capturing_handler = self.CapturingLevelHandler(level)

855

old_level = logger.level

856

old_propagate = logger.propagate

857

logger.addHandler(capturing_handler)

858

logger.setLevel(level)

859

logger.propagate = False

860

try:

861

yield capturing_handler.watcher

862

finally:

863

logger.propagate = old_propagate

864

logger.removeHandler(capturing_handler)

865

logger.setLevel(old_level)

866

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

867

0)

868

869

class CapturingLevelHandler(logging.Handler):

870

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

871

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

872

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

873

def emit(self, record):

874

self.watcher.records.append(record)

875

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

876

877

LoggingWatcher = collections.namedtuple("LoggingWatcher",

878

("records",

879

"output"))

880

881

882

class Test_string_to_delta(TestCaseWithAssertLogs):

883

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

884

# rfc3339_duration_to_delta() already has more comprehensive

885

# tests, which is run by doctest.

886

887

def test_rfc3339_zero_seconds(self):

888

self.assertEqual(datetime.timedelta(),

889

string_to_delta("PT0S"))

890

891

def test_rfc3339_zero_days(self):

892

self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))

893

894

def test_rfc3339_one_second(self):

895

self.assertEqual(datetime.timedelta(0, 1),

896

string_to_delta("PT1S"))

897

898

def test_rfc3339_two_hours(self):

899

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

900

string_to_delta("PT2H"))

901

902

def test_falls_back_to_pre_1_6_1_with_warning(self):

903

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

904

value = string_to_delta("2h")

905

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

906

907

908

class Test_check_option_syntax(unittest.TestCase):

909

def setUp(self):

910

self.parser = argparse.ArgumentParser()

911

add_command_line_options(self.parser)

912

913

def test_actions_requires_client_or_all(self):

914

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

915

options = self.parser.parse_args()

916

setattr(options, action, value)

917

with self.assertParseError():

918

self.check_option_syntax(options)

919

920

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

921

# check_option_syntax()

922

actions = {

923

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

924

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

925

# never used

926

"enable": True,

927

"disable": True,

928

"bump_timeout": True,

929

"start_checker": True,

930

"stop_checker": True,

931

"is_enabled": True,

932

"remove": True,

933

"checker": "x",

934

"timeout": datetime.timedelta(),

935

"extended_timeout": datetime.timedelta(),

936

"interval": datetime.timedelta(),

937

"approved_by_default": True,

938

"approval_delay": datetime.timedelta(),

939

"approval_duration": datetime.timedelta(),

940

"host": "x",

941

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

942

"approve": True,

943

"deny": True,

944

}

945

946

@contextlib.contextmanager

947

def assertParseError(self):

948

with self.assertRaises(SystemExit) as e:

949

with self.redirect_stderr_to_devnull():

950

yield

951

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

952

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

953

# /argparse.html#exiting-methods

954

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

955

956

@staticmethod

957

@contextlib.contextmanager

958

def redirect_stderr_to_devnull():

959

old_stderr = sys.stderr

960

with contextlib.closing(open(os.devnull, "w")) as null:

961

sys.stderr = null

962

try:

963

yield

964

finally:

965

sys.stderr = old_stderr

966

967

def check_option_syntax(self, options):

968

check_option_syntax(self.parser, options)

969

970

def test_actions_all_conflicts_with_verbose(self):

971

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

972

options = self.parser.parse_args()

973

setattr(options, action, value)

974

options.all = True

975

options.verbose = True

976

with self.assertParseError():

977

self.check_option_syntax(options)

978

979

def test_actions_with_client_conflicts_with_verbose(self):

980

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

981

options = self.parser.parse_args()

982

setattr(options, action, value)

983

options.verbose = True

984

options.client = ["client"]

985

with self.assertParseError():

986

self.check_option_syntax(options)

987

988

def test_dump_json_conflicts_with_verbose(self):

989

options = self.parser.parse_args()

990

options.dump_json = True

991

options.verbose = True

992

with self.assertParseError():

993

self.check_option_syntax(options)

994

995

def test_dump_json_conflicts_with_action(self):

996

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

997

options = self.parser.parse_args()

998

setattr(options, action, value)

999

options.dump_json = True

1000

with self.assertParseError():

1001

self.check_option_syntax(options)

1002

1003

def test_all_can_not_be_alone(self):

1004

options = self.parser.parse_args()

1005

options.all = True

1006

with self.assertParseError():

1007

self.check_option_syntax(options)

1008

1009

def test_all_is_ok_with_any_action(self):

1010

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

1011

options = self.parser.parse_args()

1012

setattr(options, action, value)

1013

options.all = True

1014

self.check_option_syntax(options)

1015

1016

def test_any_action_is_ok_with_one_client(self):

1017

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

1018

options = self.parser.parse_args()

1019

setattr(options, action, value)

1020

options.client = ["client"]

1021

self.check_option_syntax(options)

1022

1023

def test_one_client_with_all_actions_except_is_enabled(self):

1024

options = self.parser.parse_args()

1025

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

1026

if action == "is_enabled":

1027

continue

1028

setattr(options, action, value)

1029

options.client = ["client"]

1030

self.check_option_syntax(options)

1031

1032

def test_two_clients_with_all_actions_except_is_enabled(self):

1033

options = self.parser.parse_args()

1034

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

1035

if action == "is_enabled":

1036

continue

1037

setattr(options, action, value)

1038

options.client = ["client1", "client2"]

1039

self.check_option_syntax(options)

1040

1041

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1042

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

1043

if action == "is_enabled":

1044

continue

1045

options = self.parser.parse_args()

1046

setattr(options, action, value)

1047

options.client = ["client1", "client2"]

1048

self.check_option_syntax(options)

1049

1050

def test_is_enabled_fails_without_client(self):

1051

options = self.parser.parse_args()

1052

options.is_enabled = True

1053

with self.assertParseError():

1054

self.check_option_syntax(options)

1055

1056

def test_is_enabled_fails_with_two_clients(self):

1057

options = self.parser.parse_args()

1058

options.is_enabled = True

1059

options.client = ["client1", "client2"]

1060

with self.assertParseError():

1061

self.check_option_syntax(options)

1062

1063

def test_remove_can_only_be_combined_with_action_deny(self):

1064

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

1065

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

1066

continue

1067

options = self.parser.parse_args()

1068

setattr(options, action, value)

1069

options.all = True

1070

options.remove = True

1071

with self.assertParseError():

1072

self.check_option_syntax(options)

1073

1074

1075

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1076

def test_calls_and_returns_get_object_on_bus(self):

1077

class MockBus(object):

1078

called = False

1079

def get_object(mockbus_self, busname, dbus_path):

1080

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

1081

# this MockBus instance is in "mockbus_self".

1082

self.assertEqual(dbus_busname, busname)

1083

self.assertEqual(server_dbus_path, dbus_path)

1084

mockbus_self.called = True

1085

return mockbus_self

1086

1087

mockbus = get_mandos_dbus_object(bus=MockBus())

1088

self.assertIsInstance(mockbus, MockBus)

1089

self.assertTrue(mockbus.called)

1090

1091

def test_logs_and_exits_on_dbus_error(self):

1092

class FailingBusStub(object):

1093

def get_object(self, busname, dbus_path):

1094

raise dbus.exceptions.DBusException("Test")

1095

1096

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

1097

with self.assertRaises(SystemExit) as e:

1098

bus = get_mandos_dbus_object(bus=FailingBusStub())

1099

1100

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

1101

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

1102

else:

1103

self.assertIsNotNone(e.exception.code)

1104

1105

1106

class Test_get_managed_objects(TestCaseWithAssertLogs):

1107

def test_calls_and_returns_GetManagedObjects(self):

1108

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

1109

class ObjectManagerStub(object):

1110

def GetManagedObjects(self):

1111

return managed_objects

1112

retval = get_managed_objects(ObjectManagerStub())

1113

self.assertDictEqual(managed_objects, retval)

1114

1115

def test_logs_and_exits_on_dbus_error(self):

1116

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

1117

1118

class ObjectManagerFailingStub(object):

1119

def GetManagedObjects(self):

1120

dbus_logger.error("Test")

1121

raise dbus.exceptions.DBusException("Test")

1122

1123

class CountingHandler(logging.Handler):

1124

count = 0

1125

def emit(self, record):

1126

self.count += 1

1127

1128

counting_handler = CountingHandler()

1129

1130

dbus_logger.addHandler(counting_handler)

1131

1132

try:

1133

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

1134

with self.assertRaises(SystemExit) as e:

1135

get_managed_objects(ObjectManagerFailingStub())

1136

finally:

1137

dbus_logger.removeFilter(counting_handler)

1138

1139

# Make sure the dbus logger was suppressed

1140

self.assertEqual(0, counting_handler.count)

1141

1142

# Test that the dbus_logger still works

1143

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

1144

dbus_logger.error("Test")

1145

1146

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

1147

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

1148

else:

1149

self.assertIsNotNone(e.exception.code)

1150

1151

1152

class Test_commands_from_options(unittest.TestCase):

1153

def setUp(self):

1154

self.parser = argparse.ArgumentParser()

1155

add_command_line_options(self.parser)

1156

1157

def test_is_enabled(self):

1158

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

1159

command.IsEnabled)

1160

1161

def assert_command_from_args(self, args, command_cls,

1162

**cmd_attrs):

1163

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

1164

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

1165

options = self.parser.parse_args(args)

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(1, len(commands))

1169

command = commands[0]

1170

self.assertIsInstance(command, command_cls)

1171

for key, value in cmd_attrs.items():

1172

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

1173

1174

def test_is_enabled_short(self):

1175

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

1176

command.IsEnabled)

1177

1178

def test_approve(self):

1179

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

1180

command.Approve)

1181

1182

def test_approve_short(self):

1183

self.assert_command_from_args(["-A", "client"],

1184

command.Approve)

1185

1186

def test_deny(self):

1187

self.assert_command_from_args(["--deny", "client"],

1188

command.Deny)

1189

1190

def test_deny_short(self):

1191

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

1192

1193

def test_remove(self):

1194

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

1195

command.Remove)

1196

1197

def test_deny_before_remove(self):

1198

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

1199

"client"])

1200

check_option_syntax(self.parser, options)

1201

commands = commands_from_options(options)

1202

self.assertEqual(2, len(commands))

1203

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

1204

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

1205

1206

def test_deny_before_remove_reversed(self):

1207

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

1208

"--all"])

1209

check_option_syntax(self.parser, options)

1210

commands = commands_from_options(options)

1211

self.assertEqual(2, len(commands))

1212

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

1213

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

1214

1215

def test_remove_short(self):

1216

self.assert_command_from_args(["-r", "client"],

1217

command.Remove)

1218

1219

def test_dump_json(self):

1220

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

1221

command.DumpJSON)

1222

1223

def test_enable(self):

1224

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

1225

command.Enable)

1226

1227

def test_enable_short(self):

1228

self.assert_command_from_args(["-e", "client"],

1229

command.Enable)

1230

1231

def test_disable(self):

1232

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

1233

command.Disable)

1234

1235

def test_disable_short(self):

1236

self.assert_command_from_args(["-d", "client"],

1237

command.Disable)

1238

1239

def test_bump_timeout(self):

1240

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

1241

command.BumpTimeout)

1242

1243

def test_bump_timeout_short(self):

1244

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

1245

command.BumpTimeout)

1246

1247

def test_start_checker(self):

1248

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

1249

command.StartChecker)

1250

1251

def test_stop_checker(self):

1252

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

1253

command.StopChecker)

1254

1255

def test_approve_by_default(self):

1256

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

1257

"client"],

1258

command.ApproveByDefault)

1259

1260

def test_deny_by_default(self):

1261

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

1262

command.DenyByDefault)

1263

1264

def test_checker(self):

1265

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

1266

command.SetChecker,

1267

value_to_set=":")

1268

1269

def test_checker_empty(self):

1270

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

1271

command.SetChecker,

1272

value_to_set="")

1273

1274

def test_checker_short(self):

1275

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

1276

command.SetChecker,

1277

value_to_set=":")

1278

1279

def test_host(self):

1280

self.assert_command_from_args(

1281

["--host", "client.example.org", "client"],

1282

command.SetHost, value_to_set="client.example.org")

1283

1284

def test_host_short(self):

1285

self.assert_command_from_args(

1286

["-H", "client.example.org", "client"], command.SetHost,

1287

value_to_set="client.example.org")

1288

1289

def test_secret_devnull(self):

1290

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

1291

"client"], command.SetSecret,

1292

value_to_set=b"")

1293

1294

def test_secret_tempfile(self):

1295

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

1296

value = b"secret\0xyzzy\nbar"

1297

f.write(value)

1298

f.seek(0)

1299

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

1300

"client"],

1301

command.SetSecret,

1302

value_to_set=value)

1303

1304

def test_secret_devnull_short(self):

1305

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

1306

"client"], command.SetSecret,

1307

value_to_set=b"")

1308

1309

def test_secret_tempfile_short(self):

1310

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

1311

value = b"secret\0xyzzy\nbar"

1312

f.write(value)

1313

f.seek(0)

1314

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

1315

command.SetSecret,

1316

value_to_set=value)

1317

1318

def test_timeout(self):

1319

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

1320

command.SetTimeout,

1321

value_to_set=300000)

1322

1323

def test_timeout_short(self):

1324

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

1325

command.SetTimeout,

1326

value_to_set=300000)

1327

1328

def test_extended_timeout(self):

1329

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

1330

"client"],

1331

command.SetExtendedTimeout,

1332

value_to_set=900000)

1333

1334

def test_interval(self):

1335

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

1336

"client"], command.SetInterval,

1337

value_to_set=120000)

1338

1339

def test_interval_short(self):

1340

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

1341

command.SetInterval,

1342

value_to_set=120000)

1343

1344

def test_approval_delay(self):

1345

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

1346

"client"],

1347

command.SetApprovalDelay,

1348

value_to_set=30000)

1349

1350

def test_approval_duration(self):

1351

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

1352

"client"],

1353

command.SetApprovalDuration,

1354

value_to_set=1000)

1355

1356

def test_print_table(self):

1357

self.assert_command_from_args([], command.PrintTable,

1358

verbose=False)

1359

1360

def test_print_table_verbose(self):

1361

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

1362

command.PrintTable,

1363

verbose=True)

1364

1365

def test_print_table_verbose_short(self):

1366

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

1367

verbose=True)

1368

1369

1370

class TestCommand(unittest.TestCase):

1371

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

1372

1373

def setUp(self):

1374

testcase = self

1375

class MockClient(object):

1376

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

1377

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

1378

self.attributes = attributes

1379

self.attributes["Name"] = name

1380

self.calls = []

1381

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

1382

testcase.assertEqual(client_dbus_interface, interface)

1383

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1384

dbus_interface)

1385

self.attributes[propname] = value

1386

def Approve(self, approve, dbus_interface):

1387

testcase.assertEqual(client_dbus_interface,

1388

dbus_interface)

1389

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

1390

dbus_interface)))

1391

self.client = MockClient(

1392

"foo",

1393

KeyID=("92ed150794387c03ce684574b1139a65"

1394

"94a34f895daaaf09fd8ea90a27cddb12"),

1395

Secret=b"secret",

1396

Host="foo.example.org",

1397

Enabled=dbus.Boolean(True),

1398

Timeout=300000,

1399

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

1400

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

1401

Interval=120000,

1402

Fingerprint=("778827225BA7DE539C5A"

1403

"7CFA59CFF7CDBD9A5920"),

1404

CheckerRunning=dbus.Boolean(False),

1405

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

1406

ApprovalPending=dbus.Boolean(False),

1407

ApprovedByDefault=dbus.Boolean(True),

1408

LastApprovalRequest="",

1409

ApprovalDelay=0,

1410

ApprovalDuration=1000,

1411

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

1412

ExtendedTimeout=900000,

1413

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

1414

LastCheckerStatus=0)

1415

self.other_client = MockClient(

1416

"barbar",

1417

KeyID=("0558568eedd67d622f5c83b35a115f79"

1418

"6ab612cff5ad227247e46c2b020f441c"),

1419

Secret=b"secretbar",

1420

Host="192.0.2.3",

1421

Enabled=dbus.Boolean(True),

1422

Timeout=300000,

1423

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

1424

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

1425

Interval=120000,

1426

Fingerprint=("3E393AEAEFB84C7E89E2"

1427

"F547B3A107558FCA3A27"),

1428

CheckerRunning=dbus.Boolean(True),

1429

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

1430

ApprovalPending=dbus.Boolean(False),

1431

ApprovedByDefault=dbus.Boolean(False),

1432

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

1433

ApprovalDelay=30000,

1434

ApprovalDuration=93785000,

1435

Checker=":",

1436

ExtendedTimeout=900000,

1437

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

1438

LastCheckerStatus=-2)

1439

self.clients = collections.OrderedDict(

1440

[

1441

(self.client.__dbus_object_path__,

1442

self.client.attributes),

1443

(self.other_client.__dbus_object_path__,

1444

self.other_client.attributes),

1445

])

1446

self.one_client = {self.client.__dbus_object_path__:

1447

self.client.attributes}

1448

1449

@property

1450

def bus(self):

1451

class MockBus(object):

1452

@staticmethod

1453

def get_object(client_bus_name, path):

1454

self.assertEqual(dbus_busname, client_bus_name)

1455

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

1456

# MockBus instance, since this is a static method!

1457

if path == self.client.__dbus_object_path__:

1458

return self.client

1459

elif path == self.other_client.__dbus_object_path__:

1460

return self.other_client

1461

return MockBus()

1462

1463

1464

class TestBaseCommands(TestCommand):

1465

1466

def test_IsEnabled_exits_successfully(self):

1467

with self.assertRaises(SystemExit) as e:

1468

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

1469

if e.exception.code is not None:

1470

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

1471

else:

1472

self.assertIsNone(e.exception.code)

1473

1474

def test_IsEnabled_exits_with_failure(self):

1475

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

1476

with self.assertRaises(SystemExit) as e:

1477

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

1478

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

1479

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

1480

else:

1481

self.assertIsNotNone(e.exception.code)

1482

1483

def test_Approve(self):

1484

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

1485

for clientpath in self.clients:

1486

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

1487

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

1488

client.calls)

1489

1490

def test_Deny(self):

1491

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

1492

for clientpath in self.clients:

1493

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

1494

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

1495

client.calls)

1496

1497

def test_Remove(self):

1498

class MandosSpy(object):

1499

def __init__(self):

1500

self.calls = []

1501

def RemoveClient(self, dbus_path):

1502

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

1503

mandos = MandosSpy()

1504

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

1505

for clientpath in self.clients:

1506

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

1507

mandos.calls)

1508

1509

expected_json = {

1510

"foo": {

1511

"Name": "foo",

1512

"KeyID": ("92ed150794387c03ce684574b1139a65"

1513

"94a34f895daaaf09fd8ea90a27cddb12"),

1514

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

1515

"Enabled": True,

1516

"Timeout": 300000,

1517

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

1518

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

1519

"Interval": 120000,

1520

"Fingerprint": ("778827225BA7DE539C5A"

1521

"7CFA59CFF7CDBD9A5920"),

1522

"CheckerRunning": False,

1523

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

1524

"ApprovalPending": False,

1525

"ApprovedByDefault": True,

1526

"LastApprovalRequest": "",

1527

"ApprovalDelay": 0,

1528

"ApprovalDuration": 1000,

1529

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

1530

"ExtendedTimeout": 900000,

1531

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

1532

"LastCheckerStatus": 0,

1533

},

1534

"barbar": {

1535

"Name": "barbar",

1536

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1537

"6ab612cff5ad227247e46c2b020f441c"),

1538

"Host": "192.0.2.3",

1539

"Enabled": True,

1540

"Timeout": 300000,

1541

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

1542

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

1543

"Interval": 120000,

1544

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1545

"F547B3A107558FCA3A27"),

1546

"CheckerRunning": True,

1547

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

1548

"ApprovalPending": False,

1549

"ApprovedByDefault": False,

1550

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

1551

"ApprovalDelay": 30000,

1552

"ApprovalDuration": 93785000,

1553

"Checker": ":",

1554

"ExtendedTimeout": 900000,

1555

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

1556

"LastCheckerStatus": -2,

1557

},

1558

}

1559

1560

def test_DumpJSON_normal(self):

1561

with self.capture_stdout_to_buffer() as buffer:

1562

command.DumpJSON().run(self.clients)

1563

json_data = json.loads(buffer.getvalue())

1564

self.assertDictEqual(self.expected_json, json_data)

1565

1566

@staticmethod

1567

@contextlib.contextmanager

1568

def capture_stdout_to_buffer():

1569

capture_buffer = io.StringIO()

1570

old_stdout = sys.stdout

1571

sys.stdout = capture_buffer

1572

try:

1573

yield capture_buffer

1574

finally:

1575

sys.stdout = old_stdout

1576

1577

def test_DumpJSON_one_client(self):

1578

with self.capture_stdout_to_buffer() as buffer:

1579

command.DumpJSON().run(self.one_client)

1580

json_data = json.loads(buffer.getvalue())

1581

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

1582

self.assertDictEqual(expected_json, json_data)

1583

1584

def test_PrintTable_normal(self):

1585

with self.capture_stdout_to_buffer() as buffer:

1586

command.PrintTable().run(self.clients)

1587

expected_output = "\n".join((

1588

"Name Enabled Timeout Last Successful Check",

1589

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

1590

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

1591

)) + "\n"

1592

self.assertEqual(expected_output, buffer.getvalue())

1593

1594

def test_PrintTable_verbose(self):

1595

with self.capture_stdout_to_buffer() as buffer:

1596

command.PrintTable(verbose=True).run(self.clients)

1597

columns = (

1598

(

1599

"Name ",

1600

"foo ",

1601

"barbar ",

1602

),(

1603

"Enabled ",

1604

"Yes ",

1605

"Yes ",

1606

),(

1607

"Timeout ",

1608

"00:05:00 ",

1609

"00:05:00 ",

1610

),(

1611

"Last Successful Check ",

1612

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

1613

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

1614

),(

1615

"Created ",

1616

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

1617

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

1618

),(

1619

"Interval ",

1620

"00:02:00 ",

1621

"00:02:00 ",

1622

),(

1623

"Host ",

1624

"foo.example.org ",

1625

"192.0.2.3 ",

1626

),(

1627

("Key ID "

1628

" "),

1629

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1630

"ea90a27cddb12 "),

1631

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1632

"46c2b020f441c "),

1633

),(

1634

"Fingerprint ",

1635

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1636

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1637

),(

1638

"Check Is Running ",

1639

"No ",

1640

"Yes ",

1641

),(

1642

"Last Enabled ",

1643

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

1644

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

1645

),(

1646

"Approval Is Pending ",

1647

"No ",

1648

"No ",

1649

),(

1650

"Approved By Default ",

1651

"Yes ",

1652

"No ",

1653

),(

1654

"Last Approval Request ",

1655

" ",

1656

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

1657

),(

1658

"Approval Delay ",

1659

"00:00:00 ",

1660

"00:00:30 ",

1661

),(

1662

"Approval Duration ",

1663

"00:00:01 ",

1664

"1T02:03:05 ",

1665

),(

1666

"Checker ",

1667

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

1668

": ",

1669

),(

1670

"Extended Timeout ",

1671

"00:15:00 ",

1672

"00:15:00 ",

1673

),(

1674

"Expires ",

1675

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

1676

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

1677

),(

1678

"Last Checker Status",

1679

"0 ",

1680

"-2 ",

1681

)

1682

)

1683

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

1684

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

1685

for rows in columns)

1686

for line in range(num_lines))

1687

+ "\n")

1688

self.assertEqual(expected_output, buffer.getvalue())

1689

1690

def test_PrintTable_one_client(self):

1691

with self.capture_stdout_to_buffer() as buffer:

1692

command.PrintTable().run(self.one_client)

1693

expected_output = "\n".join((

1694

"Name Enabled Timeout Last Successful Check",

1695

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

1696

)) + "\n"

1697

self.assertEqual(expected_output, buffer.getvalue())

1698

1699

1700

class TestPropertySetterCmd(TestCommand):

1701

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

1702

def runTest(self):

1703

if not hasattr(self, "command"):

1704

return

1705

values_to_get = getattr(self, "values_to_get",

1706

self.values_to_set)

1707

for value_to_set, value_to_get in zip(self.values_to_set,

1708

values_to_get):

1709

for clientpath in self.clients:

1710

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

1711

old_value = client.attributes[self.propname]

1712

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

1713

self.run_command(value_to_set, self.clients)

1714

for clientpath in self.clients:

1715

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

1716

value = client.attributes[self.propname]

1717

self.assertNotIsInstance(value, self.Unique)

1718

self.assertEqual(value_to_get, value)

1719

1720

class Unique(object):

1721

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

1722

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

1723

1724

def run_command(self, value, clients):

1725

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

1726

1727

1728

class TestEnableCmd(TestPropertySetterCmd):

1729

command = command.Enable

1730

propname = "Enabled"

1731

values_to_set = [dbus.Boolean(True)]

1732

1733

1734

class TestDisableCmd(TestPropertySetterCmd):

1735

command = command.Disable

1736

propname = "Enabled"

1737

values_to_set = [dbus.Boolean(False)]

1738

1739

1740

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1741

command = command.BumpTimeout

1742

propname = "LastCheckedOK"

1743

values_to_set = [""]

1744

1745

1746

class TestStartCheckerCmd(TestPropertySetterCmd):

1747

command = command.StartChecker

1748

propname = "CheckerRunning"

1749

values_to_set = [dbus.Boolean(True)]

1750

1751

1752

class TestStopCheckerCmd(TestPropertySetterCmd):

1753

command = command.StopChecker

1754

propname = "CheckerRunning"

1755

values_to_set = [dbus.Boolean(False)]

1756

1757

1758

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1759

command = command.ApproveByDefault

1760

propname = "ApprovedByDefault"

1761

values_to_set = [dbus.Boolean(True)]

1762

1763

1764

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1765

command = command.DenyByDefault

1766

propname = "ApprovedByDefault"

1767

values_to_set = [dbus.Boolean(False)]

1768

1769

1770

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1771

"""Abstract class for tests of PropertySetterValueCmd classes"""

1772

1773

def runTest(self):

1774

if type(self) is TestPropertySetterValueCmd:

1775

return

1776

return super(TestPropertySetterValueCmd, self).runTest()

1777

1778

def run_command(self, value, clients):

1779

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

1780

1781

1782

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1783

command = command.SetChecker

1784

propname = "Checker"

1785

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

1786

1787

1788

class TestSetHostCmd(TestPropertySetterValueCmd):

1789

command = command.SetHost

1790

propname = "Host"

1791

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

1792

1793

1794

class TestSetSecretCmd(TestPropertySetterValueCmd):

1795

command = command.SetSecret

1796

propname = "Secret"

1797

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

1798

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

1799

values_to_get = [f.getvalue() for f in values_to_set]

1800

1801

1802

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1803

command = command.SetTimeout

1804

propname = "Timeout"

1805

values_to_set = [datetime.timedelta(),

1806

datetime.timedelta(minutes=5),

1807

datetime.timedelta(seconds=1),

1808

datetime.timedelta(weeks=1),

1809

datetime.timedelta(weeks=52)]

1810

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1811

1812

1813

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1814

command = command.SetExtendedTimeout

1815

propname = "ExtendedTimeout"

1816

values_to_set = [datetime.timedelta(),

1817

datetime.timedelta(minutes=5),

1818

datetime.timedelta(seconds=1),

1819

datetime.timedelta(weeks=1),

1820

datetime.timedelta(weeks=52)]

1821

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1822

1823

1824

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1825

command = command.SetInterval

1826

propname = "Interval"

1827

values_to_set = [datetime.timedelta(),

1828

datetime.timedelta(minutes=5),

1829

datetime.timedelta(seconds=1),

1830

datetime.timedelta(weeks=1),

1831

datetime.timedelta(weeks=52)]

1832

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1833

1834

1835

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1836

command = command.SetApprovalDelay

1837

propname = "ApprovalDelay"

1838

values_to_set = [datetime.timedelta(),

1839

datetime.timedelta(minutes=5),

1840

datetime.timedelta(seconds=1),

1841

datetime.timedelta(weeks=1),

1842

datetime.timedelta(weeks=52)]

1843

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1844

1845

1846

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1847

command = command.SetApprovalDuration

1848

propname = "ApprovalDuration"

1849

values_to_set = [datetime.timedelta(),

1850

datetime.timedelta(minutes=5),

1851

datetime.timedelta(seconds=1),

1852

datetime.timedelta(weeks=1),

1853

datetime.timedelta(weeks=52)]

1854

values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]

1855

1856

1857

1858

def should_only_run_tests():

1859

parser = argparse.ArgumentParser(add_help=False)

1860

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

1861

args, unknown_args = parser.parse_known_args()

1862

run_tests = args.check

1863

if run_tests:

1864

# Remove --check argument from sys.argv

1865

sys.argv[1:] = unknown_args

1866

return run_tests

1867

1868

# Add all tests from doctest strings

1869

def load_tests(loader, tests, none):

1870

import doctest

1871

tests.addTests(doctest.DocTestSuite())

1872

return tests

1873

1874

if __name__ == "__main__":

1875

try:

1876

if should_only_run_tests():

1877

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

1878

unittest.main()

1879

else:

1880

main()

1881

finally:

1882

logging.shutdown()