/mandos/trunk : revision 1085

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

#

24

25

from __future__ import division

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

26

35

import sys

27

import dbus

28

from optparse import OptionParser

36

import argparse

29

37

import locale

30

38

import datetime

31

39

import re

32

40

import os

33

34

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

35

36

tablewords = {

37

'Name': u'Name',

38

'Enabled': u'Enabled',

39

'Timeout': u'Timeout',

40

'LastCheckedOK': u'Last Successful Check',

41

'LastApprovalRequest': u'Last Approval Request',

42

'Created': u'Created',

43

'Interval': u'Interval',

44

'Host': u'Host',

45

'Fingerprint': u'Fingerprint',

46

'CheckerRunning': u'Check Is Running',

47

'LastEnabled': u'Last Enabled',

48

'ApprovalPending': u'Approval Is Pending',

49

'ApprovedByDefault': u'Approved By Default',

50

'ApprovalDelay': u"Approval Delay",

51

'ApprovalDuration': u"Approval Duration",

52

'Checker': u'Checker',

53

}

54

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

55

domain = 'se.bsnet.fukt'

56

busname = domain + '.Mandos'

57

server_path = '/'

58

server_interface = domain + '.Mandos'

59

client_interface = domain + '.Mandos.Client'

60

version = "1.2"

61

62

def timedelta_to_milliseconds(td):

63

"Convert a datetime.timedelta object to milliseconds"

64

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

65

+ (td.seconds * 1000)

66

+ (td.microseconds // 1000))

67

68

def milliseconds_to_string(ms):

69

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

70

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

71

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

72

"hours": td.seconds // 3600,

73

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

74

"seconds": td.seconds % 60,

75

})

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

87

add_command_line_options(parser)

88

89

options = parser.parse_args()

90

91

check_option_syntax(parser, options)

92

93

clientnames = options.client

94

95

if options.debug:

96

log.setLevel(logging.DEBUG)

97

98

bus = dbus.SystemBus()

99

100

mandos_dbus_object = get_mandos_dbus_object(bus)

101

102

mandos_serv = dbus.Interface(

103

mandos_dbus_object, dbus_interface=server_dbus_interface)

104

mandos_serv_object_manager = dbus.Interface(

105

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

106

107

managed_objects = get_managed_objects(mandos_serv_object_manager)

108

109

all_clients = {}

110

for path, ifs_and_props in managed_objects.items():

111

try:

112

all_clients[path] = ifs_and_props[client_dbus_interface]

113

except KeyError:

114

pass

115

116

# Compile dict of (clientpath: properties) to process

117

if not clientnames:

118

clients = all_clients

119

else:

120

clients = {}

121

for name in clientnames:

122

for objpath, properties in all_clients.items():

123

if properties["Name"] == name:

124

clients[objpath] = properties

125

break

126

else:

127

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

128

sys.exit(1)

129

130

commands = commands_from_options(options)

131

132

for command in commands:

133

command.run(clients, bus, mandos_serv)

134

135

136

def add_command_line_options(parser):

137

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

138

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

139

help="show version number and exit")

140

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

141

help="Select all clients")

142

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

143

help="Print all fields")

144

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

145

help="Dump client data in JSON format")

146

enable_disable = parser.add_mutually_exclusive_group()

147

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

148

help="Enable client")

149

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

150

action="store_true",

151

help="disable client")

152

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

153

help="Bump timeout for client")

154

start_stop_checker = parser.add_mutually_exclusive_group()

155

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

156

action="store_true",

157

help="Start checker for client")

158

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

159

action="store_true",

160

help="Stop checker for client")

161

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

162

help="Check if client is enabled")

163

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

164

help="Remove client")

165

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

166

help="Set checker command for client")

167

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

168

help="Set timeout for client")

169

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

170

help="Set extended timeout for client")

171

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

172

help="Set checker interval for client")

173

approve_deny_default = parser.add_mutually_exclusive_group()

174

approve_deny_default.add_argument(

175

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

176

default=None, dest="approved_by_default",

177

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

178

approve_deny_default.add_argument(

179

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

180

dest="approved_by_default",

181

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

182

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

183

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

184

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

185

help="Set duration of one client approval")

186

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

187

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

188

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

189

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

190

approve_deny = parser.add_mutually_exclusive_group()

191

approve_deny.add_argument(

192

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

193

help="Approve any current client request")

194

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

195

help="Deny any current client request")

196

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

197

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

198

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

199

help="Run self-test")

200

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

76

201

77

202

78

203

def string_to_delta(interval):

79

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

80

81

>>> string_to_delta('7d')

82

datetime.timedelta(7)

83

>>> string_to_delta('60s')

84

datetime.timedelta(0, 60)

85

>>> string_to_delta('60m')

86

datetime.timedelta(0, 3600)

87

>>> string_to_delta('24h')

88

datetime.timedelta(1)

89

>>> string_to_delta(u'1w')

90

datetime.timedelta(7)

91

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

92

datetime.timedelta(0, 330)

93

"""

94

timevalue = datetime.timedelta(0)

95

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

96

97

for s in regexp.findall(interval):

98

try:

99

suffix = unicode(s[-1])

100

value = int(s[:-1])

101

if suffix == u"d":

102

delta = datetime.timedelta(value)

103

elif suffix == u"s":

104

delta = datetime.timedelta(0, value)

105

elif suffix == u"m":

106

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

107

elif suffix == u"h":

108

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

109

elif suffix == u"w":

110

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

111

else:

112

raise ValueError

113

except (ValueError, IndexError):

114

raise ValueError

115

timevalue += delta

116

return timevalue

117

118

def print_clients(clients, keywords):

119

def valuetostring(value, keyword):

120

if type(value) is dbus.Boolean:

121

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

122

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

123

u"ApprovalDuration"):

124

return milliseconds_to_string(value)

125

return unicode(value)

126

127

# Create format string to print table rows

128

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

129

max(len(tablewords[key]),

130

max(len(valuetostring(client[key],

131

key))

132

for client in

133

clients))

134

for key in keywords)

135

# Print header line

136

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

137

for client in clients:

138

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

139

for key in keywords)

140

141

def has_actions(options):

142

return any((options.enable,

143

options.disable,

144

options.bump_timeout,

145

options.start_checker,

146

options.stop_checker,

147

options.is_enabled,

148

options.remove,

149

options.checker is not None,

150

options.timeout is not None,

151

options.interval is not None,

152

options.approved_by_default is not None,

153

options.approval_delay is not None,

154

options.approval_duration is not None,

155

options.host is not None,

156

options.secret is not None,

157

options.approve,

158

options.deny))

159

160

def main():

161

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

162

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

163

help="Select all clients")

164

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

165

help="Print all fields")

166

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

167

help="Enable client")

168

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

169

help="disable client")

170

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

171

help="Bump timeout for client")

172

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

173

help="Start checker for client")

174

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

175

help="Stop checker for client")

176

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

177

help="Check if client is enabled")

178

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

179

help="Remove client")

180

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

181

help="Set checker command for client")

182

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

183

help="Set timeout for client")

184

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

185

help="Set checker interval for client")

186

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

187

dest=u"approved_by_default",

188

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

189

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

190

dest=u"approved_by_default",

191

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

192

parser.add_option("--approval-delay", type="string",

193

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

194

parser.add_option("--approval-duration", type="string",

195

help="Set duration of one client approval")

196

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

197

help="Set host for client")

198

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

199

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

200

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

201

help="Approve any current client request")

202

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

203

help="Deny any current client request")

204

options, client_names = parser.parse_args()

205

206

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

207

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

208

if options.verbose and has_actions(options):

209

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

210

' --all.')

211

if options.all and not has_actions(options):

212

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

213

214

try:

215

bus = dbus.SystemBus()

216

mandos_dbus_objc = bus.get_object(busname, server_path)

217

except dbus.exceptions.DBusException:

218

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

219

sys.exit(1)

220

221

mandos_serv = dbus.Interface(mandos_dbus_objc,

222

dbus_interface = server_interface)

223

224

#block stderr since dbus library prints to stderr

225

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

226

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

227

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

228

os.close(null)

229

try:

230

try:

231

mandos_clients = mandos_serv.GetAllClientsWithProperties()

232

finally:

233

#restore stderr

234

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

235

os.close(stderrcopy)

236

except dbus.exceptions.DBusException, e:

237

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

238

sys.exit(1)

239

240

# Compile dict of (clients: properties) to process

241

clients={}

242

243

if options.all or not client_names:

244

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

245

for path, properties in

246

mandos_clients.iteritems())

247

else:

248

for name in client_names:

249

for path, client in mandos_clients.iteritems():

250

if client['Name'] == name:

251

client_objc = bus.get_object(busname, path)

252

clients[client_objc] = client

253

break

254

else:

255

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

256

sys.exit(1)

257

258

if not has_actions(options) and clients:

259

if options.verbose:

260

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

261

'LastCheckedOK', 'Created', 'Interval',

262

'Host', 'Fingerprint', 'CheckerRunning',

263

'LastEnabled', 'ApprovalPending',

264

'ApprovedByDefault',

265

'LastApprovalRequest', 'ApprovalDelay',

266

'ApprovalDuration', 'Checker')

267

else:

268

keywords = defaultkeywords

269

270

print_clients(clients.values(), keywords)

271

else:

272

# Process each client in the list by all selected options

273

for client in clients:

274

if options.remove:

275

mandos_serv.RemoveClient(client.__dbus_object_path__)

276

if options.enable:

277

client.Enable(dbus_interface=client_interface)

278

if options.disable:

279

client.Disable(dbus_interface=client_interface)

280

if options.bump_timeout:

281

client.CheckedOK(dbus_interface=client_interface)

282

if options.start_checker:

283

client.StartChecker(dbus_interface=client_interface)

284

if options.stop_checker:

285

client.StopChecker(dbus_interface=client_interface)

286

if options.is_enabled:

287

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

288

u"Enabled",

289

dbus_interface=dbus.PROPERTIES_IFACE)

290

else 1)

291

if options.checker:

292

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

293

dbus_interface=dbus.PROPERTIES_IFACE)

294

if options.host:

295

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

296

dbus_interface=dbus.PROPERTIES_IFACE)

297

if options.interval:

298

client.Set(client_interface, u"Interval",

299

timedelta_to_milliseconds

300

(string_to_delta(options.interval)),

301

dbus_interface=dbus.PROPERTIES_IFACE)

302

if options.approval_delay:

303

client.Set(client_interface, u"ApprovalDelay",

304

timedelta_to_milliseconds

305

(string_to_delta(options.

306

approval_delay)),

307

dbus_interface=dbus.PROPERTIES_IFACE)

308

if options.approval_duration:

309

client.Set(client_interface, u"ApprovalDuration",

310

timedelta_to_milliseconds

311

(string_to_delta(options.

312

approval_duration)),

313

dbus_interface=dbus.PROPERTIES_IFACE)

314

if options.timeout:

315

client.Set(client_interface, u"Timeout",

316

timedelta_to_milliseconds

317

(string_to_delta(options.timeout)),

318

dbus_interface=dbus.PROPERTIES_IFACE)

319

if options.secret:

320

client.Set(client_interface, u"Secret",

321

dbus.ByteArray(open(options.secret,

322

u'rb').read()),

323

dbus_interface=dbus.PROPERTIES_IFACE)

324

if options.approved_by_default is not None:

325

client.Set(client_interface, u"ApprovedByDefault",

326

dbus.Boolean(options

327

.approved_by_default),

328

dbus_interface=dbus.PROPERTIES_IFACE)

329

if options.approve:

330

client.Approve(dbus.Boolean(True),

331

dbus_interface=client_interface)

332

elif options.deny:

333

client.Approve(dbus.Boolean(False),

334

dbus_interface=client_interface)

335

336

if __name__ == '__main__':

337

main()

204

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

205

206

try:

207

return rfc3339_duration_to_delta(interval)

208

except ValueError as e:

209

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

210

' '.join(e.args))

211

return parse_pre_1_6_1_interval(interval)

212

213

214

def rfc3339_duration_to_delta(duration):

215

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

216

217

>>> rfc3339_duration_to_delta("P7D")

218

datetime.timedelta(7)

219

>>> rfc3339_duration_to_delta("PT60S")

220

datetime.timedelta(0, 60)

221

>>> rfc3339_duration_to_delta("PT60M")

222

datetime.timedelta(0, 3600)

223

>>> rfc3339_duration_to_delta("P60M")

224

datetime.timedelta(1680)

225

>>> rfc3339_duration_to_delta("PT24H")

226

datetime.timedelta(1)

227

>>> rfc3339_duration_to_delta("P1W")

228

datetime.timedelta(7)

229

>>> rfc3339_duration_to_delta("PT5M30S")

230

datetime.timedelta(0, 330)

231

>>> rfc3339_duration_to_delta("P1DT3M20S")

232

datetime.timedelta(1, 200)

233

>>> # Can not be empty:

234

>>> rfc3339_duration_to_delta("")

235

Traceback (most recent call last):

236

...

237

ValueError: Invalid RFC 3339 duration: ""

238

>>> # Must start with "P":

239

>>> rfc3339_duration_to_delta("1D")

240

Traceback (most recent call last):

241

...

242

ValueError: Invalid RFC 3339 duration: "1D"

243

>>> # Must use correct order

244

>>> rfc3339_duration_to_delta("PT1S2M")

245

Traceback (most recent call last):

246

...

247

ValueError: Invalid RFC 3339 duration: "PT1S2M"

248

>>> # Time needs time marker

249

>>> rfc3339_duration_to_delta("P1H2S")

250

Traceback (most recent call last):

251

...

252

ValueError: Invalid RFC 3339 duration: "P1H2S"

253

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

254

>>> rfc3339_duration_to_delta("P1D2W")

255

Traceback (most recent call last):

256

...

257

ValueError: Invalid RFC 3339 duration: "P1D2W"

258

>>> rfc3339_duration_to_delta("P2W2H")

259

Traceback (most recent call last):

260

...

261

ValueError: Invalid RFC 3339 duration: "P2W2H"

262

"""

263

264

# Parsing an RFC 3339 duration with regular expressions is not

265

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

266

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

267

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

268

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

269

# avoid excessive use of external libraries.

270

271

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

272

Token = collections.namedtuple("Token", (

273

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

274

# a "group" containing digits

275

"value", # datetime.timedelta or None

276

"followers")) # Tokens valid after this token

277

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

278

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

279

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

280

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

281

datetime.timedelta(seconds=1),

282

frozenset((token_end, )))

283

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

284

datetime.timedelta(minutes=1),

285

frozenset((token_second, token_end)))

286

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

287

datetime.timedelta(hours=1),

288

frozenset((token_minute, token_end)))

289

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

290

None,

291

frozenset((token_hour, token_minute,

292

token_second)))

293

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

294

datetime.timedelta(days=1),

295

frozenset((token_time, token_end)))

296

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

297

datetime.timedelta(weeks=4),

298

frozenset((token_day, token_end)))

299

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

300

datetime.timedelta(weeks=52),

301

frozenset((token_month, token_end)))

302

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

303

datetime.timedelta(weeks=1),

304

frozenset((token_end, )))

305

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

306

frozenset((token_year, token_month,

307

token_day, token_time,

308

token_week)))

309

# Define starting values:

310

# Value so far

311

value = datetime.timedelta()

312

found_token = None

313

# Following valid tokens

314

followers = frozenset((token_duration, ))

315

# String left to parse

316

s = duration

317

# Loop until end token is found

318

while found_token is not token_end:

319

# Search for any currently valid tokens

320

for token in followers:

321

match = token.regexp.match(s)

322

if match is not None:

323

# Token found

324

if token.value is not None:

325

# Value found, parse digits

326

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

327

# Add to value so far

328

value += factor * token.value

329

# Strip token from string

330

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

331

# Go to found token

332

found_token = token

333

# Set valid next tokens

334

followers = found_token.followers

335

break

336

else:

337

# No currently valid tokens were found

338

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

339

.format(duration))

340

# End token found

341

return value

342

343

344

def parse_pre_1_6_1_interval(interval):

345

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

346

and return a datetime.timedelta

347

348

>>> parse_pre_1_6_1_interval('7d')

349

datetime.timedelta(7)

350

>>> parse_pre_1_6_1_interval('60s')

351

datetime.timedelta(0, 60)

352

>>> parse_pre_1_6_1_interval('60m')

353

datetime.timedelta(0, 3600)

354

>>> parse_pre_1_6_1_interval('24h')

355

datetime.timedelta(1)

356

>>> parse_pre_1_6_1_interval('1w')

357

datetime.timedelta(7)

358

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

359

datetime.timedelta(0, 330)

360

>>> parse_pre_1_6_1_interval('')

361

datetime.timedelta(0)

362

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

363

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

364

datetime.timedelta(2, 480, 18000)

365

366

"""

367

368

value = datetime.timedelta(0)

369

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

370

371

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

372

if suffix == "d":

373

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

374

elif suffix == "s":

375

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

376

elif suffix == "m":

377

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

378

elif suffix == "h":

379

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

380

elif suffix == "w":

381

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

382

elif suffix == "":

383

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

384

return value

385

386

387

def check_option_syntax(parser, options):

388

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

389

argparse"""

390

391

def has_actions(options):

392

return any((options.enable,

393

options.disable,

394

options.bump_timeout,

395

options.start_checker,

396

options.stop_checker,

397

options.is_enabled,

398

options.remove,

399

options.checker is not None,

400

options.timeout is not None,

401

options.extended_timeout is not None,

402

options.interval is not None,

403

options.approved_by_default is not None,

404

options.approval_delay is not None,

405

options.approval_duration is not None,

406

options.host is not None,

407

options.secret is not None,

408

options.approve,

409

options.deny))

410

411

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

412

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

413

if options.verbose and has_actions(options):

414

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

415

if options.dump_json and (options.verbose

416

or has_actions(options)):

417

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

418

if options.all and not has_actions(options):

419

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

420

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

421

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

422

if options.remove:

423

options.remove = False

424

if has_actions(options) and not options.deny:

425

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

426

options.remove = True

427

428

429

def get_mandos_dbus_object(bus):

430

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

431

dbus_busname, server_dbus_path)

432

with if_dbus_exception_log_with_exception_and_exit(

433

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

434

mandos_dbus_object = bus.get_object(dbus_busname,

435

server_dbus_path)

436

return mandos_dbus_object

437

438

439

@contextlib.contextmanager

440

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

441

try:

442

yield

443

except dbus.exceptions.DBusException as e:

444

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

445

sys.exit(1)

446

447

448

def get_managed_objects(object_manager):

449

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

450

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

451

with if_dbus_exception_log_with_exception_and_exit(

452

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

453

with SilenceLogger("dbus.proxies"):

454

managed_objects = object_manager.GetManagedObjects()

455

return managed_objects

456

457

458

class SilenceLogger(object):

459

"Simple context manager to silence a particular logger"

460

def __init__(self, loggername):

461

self.logger = logging.getLogger(loggername)

462

463

def __enter__(self):

464

self.logger.addFilter(self.nullfilter)

465

return self

466

467

class NullFilter(logging.Filter):

468

def filter(self, record):

469

return False

470

471

nullfilter = NullFilter()

472

473

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

474

self.logger.removeFilter(self.nullfilter)

475

476

477

def commands_from_options(options):

478

479

commands = []

480

481

if options.is_enabled:

482

commands.append(command.IsEnabled())

483

484

if options.approve:

485

commands.append(command.Approve())

486

487

if options.deny:

488

commands.append(command.Deny())

489

490

if options.remove:

491

commands.append(command.Remove())

492

493

if options.dump_json:

494

commands.append(command.DumpJSON())

495

496

if options.enable:

497

commands.append(command.Enable())

498

499

if options.disable:

500

commands.append(command.Disable())

501

502

if options.bump_timeout:

503

commands.append(command.BumpTimeout())

504

505

if options.start_checker:

506

commands.append(command.StartChecker())

507

508

if options.stop_checker:

509

commands.append(command.StopChecker())

510

511

if options.approved_by_default is not None:

512

if options.approved_by_default:

513

commands.append(command.ApproveByDefault())

514

else:

515

commands.append(command.DenyByDefault())

516

517

if options.checker is not None:

518

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

519

520

if options.host is not None:

521

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

522

523

if options.secret is not None:

524

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

525

526

if options.timeout is not None:

527

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

528

529

if options.extended_timeout:

530

commands.append(

531

command.SetExtendedTimeout(options.extended_timeout))

532

533

if options.interval is not None:

534

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

535

536

if options.approval_delay is not None:

537

commands.append(

538

command.SetApprovalDelay(options.approval_delay))

539

540

if options.approval_duration is not None:

541

commands.append(

542

command.SetApprovalDuration(options.approval_duration))

543

544

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

545

# optionally verbosely

546

if not commands:

547

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

548

549

return commands

550

551

552

class command(object):

553

"""A namespace for command classes"""

554

555

class Base(object):

556

"""Abstract base class for commands"""

557

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

558

"""Normal commands should implement run_on_one_client(),

559

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

560

override this run() method instead.

561

"""

562

self.mandos = mandos

563

for clientpath, properties in clients.items():

564

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

565

dbus_busname, str(clientpath))

566

client = bus.get_object(dbus_busname, clientpath)

567

self.run_on_one_client(client, properties)

568

569

570

class IsEnabled(Base):

571

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

572

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

573

if self.is_enabled(client, properties):

574

sys.exit(0)

575

sys.exit(1)

576

def is_enabled(self, client, properties):

577

return properties["Enabled"]

578

579

580

class Approve(Base):

581

def run_on_one_client(self, client, properties):

582

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

583

client.__dbus_object_path__,

584

client_dbus_interface)

585

client.Approve(dbus.Boolean(True),

586

dbus_interface=client_dbus_interface)

587

588

589

class Deny(Base):

590

def run_on_one_client(self, client, properties):

591

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

592

client.__dbus_object_path__,

593

client_dbus_interface)

594

client.Approve(dbus.Boolean(False),

595

dbus_interface=client_dbus_interface)

596

597

598

class Remove(Base):

599

def run_on_one_client(self, client, properties):

600

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

601

dbus_busname, server_dbus_path,

602

server_dbus_interface,

603

str(client.__dbus_object_path__))

604

self.mandos.RemoveClient(client.__dbus_object_path__)

605

606

607

class Output(Base):

608

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

609

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

610

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

611

"Fingerprint", "CheckerRunning",

612

"LastEnabled", "ApprovalPending",

613

"ApprovedByDefault", "LastApprovalRequest",

614

"ApprovalDelay", "ApprovalDuration",

615

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

616

"LastCheckerStatus")

617

618

619

class DumpJSON(Output):

620

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

621

data = {client["Name"]:

622

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

623

for key in self.all_keywords}

624

for client in clients.values()}

625

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

626

627

@staticmethod

628

def dbus_boolean_to_bool(value):

629

if isinstance(value, dbus.Boolean):

630

value = bool(value)

631

return value

632

633

634

class PrintTable(Output):

635

def __init__(self, verbose=False):

636

self.verbose = verbose

637

638

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

639

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

640

"LastCheckedOK")

641

keywords = default_keywords

642

if self.verbose:

643

keywords = self.all_keywords

644

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

645

646

class TableOfClients(object):

647

tableheaders = {

648

"Name": "Name",

649

"Enabled": "Enabled",

650

"Timeout": "Timeout",

651

"LastCheckedOK": "Last Successful Check",

652

"LastApprovalRequest": "Last Approval Request",

653

"Created": "Created",

654

"Interval": "Interval",

655

"Host": "Host",

656

"Fingerprint": "Fingerprint",

657

"KeyID": "Key ID",

658

"CheckerRunning": "Check Is Running",

659

"LastEnabled": "Last Enabled",

660

"ApprovalPending": "Approval Is Pending",

661

"ApprovedByDefault": "Approved By Default",

662

"ApprovalDelay": "Approval Delay",

663

"ApprovalDuration": "Approval Duration",

664

"Checker": "Checker",

665

"ExtendedTimeout": "Extended Timeout",

666

"Expires": "Expires",

667

"LastCheckerStatus": "Last Checker Status",

668

}

669

670

def __init__(self, clients, keywords):

671

self.clients = clients

672

self.keywords = keywords

673

674

def __str__(self):

675

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

676

677

if sys.version_info.major == 2:

678

__unicode__ = __str__

679

def __str__(self):

680

return str(self).encode(

681

locale.getpreferredencoding())

682

683

def rows(self):

684

format_string = self.row_formatting_string()

685

rows = [self.header_line(format_string)]

686

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

687

for client in self.clients)

688

return rows

689

690

def row_formatting_string(self):

691

"Format string used to format table rows"

692

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

693

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

694

*(len(self.string_from_client(client,

695

key))

696

for client in self.clients)),

697

key=key)

698

for key in self.keywords)

699

700

def string_from_client(self, client, key):

701

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

702

703

@classmethod

704

def valuetostring(cls, value, keyword):

705

if isinstance(value, dbus.Boolean):

706

return "Yes" if value else "No"

707

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

708

"ApprovalDuration", "ExtendedTimeout"):

709

return cls.milliseconds_to_string(value)

710

return str(value)

711

712

def header_line(self, format_string):

713

return format_string.format(**self.tableheaders)

714

715

def client_line(self, client, format_string):

716

return format_string.format(

717

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

718

for key in self.keywords})

719

720

@staticmethod

721

def milliseconds_to_string(ms):

722

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

723

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

724

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

725

if td.days else "",

726

hours=td.seconds // 3600,

727

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

728

seconds=td.seconds % 60))

729

730

731

class Property(Base):

732

"Abstract class for Actions for setting one client property"

733

734

def run_on_one_client(self, client, properties):

735

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

736

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

737

client.__dbus_object_path__,

738

dbus.PROPERTIES_IFACE, client_dbus_interface,

739

self.propname, self.value_to_set

740

if not isinstance(self.value_to_set,

741

dbus.Boolean)

742

else bool(self.value_to_set))

743

client.Set(client_dbus_interface, self.propname,

744

self.value_to_set,

745

dbus_interface=dbus.PROPERTIES_IFACE)

746

747

@property

748

def propname(self):

749

raise NotImplementedError()

750

751

752

class Enable(Property):

753

propname = "Enabled"

754

value_to_set = dbus.Boolean(True)

755

756

757

class Disable(Property):

758

propname = "Enabled"

759

value_to_set = dbus.Boolean(False)

760

761

762

class BumpTimeout(Property):

763

propname = "LastCheckedOK"

764

value_to_set = ""

765

766

767

class StartChecker(Property):

768

propname = "CheckerRunning"

769

value_to_set = dbus.Boolean(True)

770

771

772

class StopChecker(Property):

773

propname = "CheckerRunning"

774

value_to_set = dbus.Boolean(False)

775

776

777

class ApproveByDefault(Property):

778

propname = "ApprovedByDefault"

779

value_to_set = dbus.Boolean(True)

780

781

782

class DenyByDefault(Property):

783

propname = "ApprovedByDefault"

784

value_to_set = dbus.Boolean(False)

785

786

787

class PropertyValue(Property):

788

"Abstract class for Property recieving a value as argument"

789

def __init__(self, value):

790

self.value_to_set = value

791

792

793

class SetChecker(PropertyValue):

794

propname = "Checker"

795

796

797

class SetHost(PropertyValue):

798

propname = "Host"

799

800

801

class SetSecret(PropertyValue):

802

propname = "Secret"

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, read data from supplied file object"""

811

self._vts = value.read()

812

value.close()

813

814

815

class MillisecondsPropertyValueArgument(PropertyValue):

816

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

817

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

818

819

@property

820

def value_to_set(self):

821

return self._vts

822

823

@value_to_set.setter

824

def value_to_set(self, value):

825

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

826

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

827

828

829

class SetTimeout(MillisecondsPropertyValueArgument):

830

propname = "Timeout"

831

832

833

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

834

propname = "ExtendedTimeout"

835

836

837

class SetInterval(MillisecondsPropertyValueArgument):

838

propname = "Interval"

839

840

841

class SetApprovalDelay(MillisecondsPropertyValueArgument):

842

propname = "ApprovalDelay"

843

844

845

class SetApprovalDuration(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDuration"

847

848

849

850

class TestCaseWithAssertLogs(unittest.TestCase):

851

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

852

853

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

854

@contextlib.contextmanager

855

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

856

capturing_handler = self.CapturingLevelHandler(level)

857

old_level = logger.level

858

old_propagate = logger.propagate

859

logger.addHandler(capturing_handler)

860

logger.setLevel(level)

861

logger.propagate = False

862

try:

863

yield capturing_handler.watcher

864

finally:

865

logger.propagate = old_propagate

866

logger.removeHandler(capturing_handler)

867

logger.setLevel(old_level)

868

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

869

0)

870

871

class CapturingLevelHandler(logging.Handler):

872

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

873

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

874

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

875

def emit(self, record):

876

self.watcher.records.append(record)

877

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

878

879

LoggingWatcher = collections.namedtuple("LoggingWatcher",

880

("records",

881

"output"))

882

883

884

class Test_string_to_delta(TestCaseWithAssertLogs):

885

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

886

# rfc3339_duration_to_delta() already has more comprehensive

887

# tests, which is run by doctest.

888

889

def test_rfc3339_zero_seconds(self):

890

self.assertEqual(datetime.timedelta(),

891

string_to_delta("PT0S"))

892

893

def test_rfc3339_zero_days(self):

894

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

895

896

def test_rfc3339_one_second(self):

897

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

898

string_to_delta("PT1S"))

899

900

def test_rfc3339_two_hours(self):

901

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

902

string_to_delta("PT2H"))

903

904

def test_falls_back_to_pre_1_6_1_with_warning(self):

905

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

906

value = string_to_delta("2h")

907

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

908

909

910

class Test_check_option_syntax(unittest.TestCase):

911

def setUp(self):

912

self.parser = argparse.ArgumentParser()

913

add_command_line_options(self.parser)

914

915

def test_actions_requires_client_or_all(self):

916

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

917

options = self.parser.parse_args()

918

setattr(options, action, value)

919

with self.assertParseError():

920

self.check_option_syntax(options)

921

922

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

923

# check_option_syntax()

924

actions = {

925

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

926

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

927

# never used

928

"enable": True,

929

"disable": True,

930

"bump_timeout": True,

931

"start_checker": True,

932

"stop_checker": True,

933

"is_enabled": True,

934

"remove": True,

935

"checker": "x",

936

"timeout": datetime.timedelta(),

937

"extended_timeout": datetime.timedelta(),

938

"interval": datetime.timedelta(),

939

"approved_by_default": True,

940

"approval_delay": datetime.timedelta(),

941

"approval_duration": datetime.timedelta(),

942

"host": "x",

943

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

944

"approve": True,

945

"deny": True,

946

}

947

948

@contextlib.contextmanager

949

def assertParseError(self):

950

with self.assertRaises(SystemExit) as e:

951

with self.redirect_stderr_to_devnull():

952

yield

953

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

954

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

955

# /argparse.html#exiting-methods

956

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

957

958

@staticmethod

959

@contextlib.contextmanager

960

def redirect_stderr_to_devnull():

961

old_stderr = sys.stderr

962

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

963

sys.stderr = null

964

try:

965

yield

966

finally:

967

sys.stderr = old_stderr

968

969

def check_option_syntax(self, options):

970

check_option_syntax(self.parser, options)

971

972

def test_actions_all_conflicts_with_verbose(self):

973

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

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.all = True

977

options.verbose = True

978

with self.assertParseError():

979

self.check_option_syntax(options)

980

981

def test_actions_with_client_conflicts_with_verbose(self):

982

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

983

options = self.parser.parse_args()

984

setattr(options, action, value)

985

options.verbose = True

986

options.client = ["foo"]

987

with self.assertParseError():

988

self.check_option_syntax(options)

989

990

def test_dump_json_conflicts_with_verbose(self):

991

options = self.parser.parse_args()

992

options.dump_json = True

993

options.verbose = True

994

with self.assertParseError():

995

self.check_option_syntax(options)

996

997

def test_dump_json_conflicts_with_action(self):

998

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

999

options = self.parser.parse_args()

1000

setattr(options, action, value)

1001

options.dump_json = True

1002

with self.assertParseError():

1003

self.check_option_syntax(options)

1004

1005

def test_all_can_not_be_alone(self):

1006

options = self.parser.parse_args()

1007

options.all = True

1008

with self.assertParseError():

1009

self.check_option_syntax(options)

1010

1011

def test_all_is_ok_with_any_action(self):

1012

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

1013

options = self.parser.parse_args()

1014

setattr(options, action, value)

1015

options.all = True

1016

self.check_option_syntax(options)

1017

1018

def test_any_action_is_ok_with_one_client(self):

1019

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

1020

options = self.parser.parse_args()

1021

setattr(options, action, value)

1022

options.client = ["foo"]

1023

self.check_option_syntax(options)

1024

1025

def test_one_client_with_all_actions_except_is_enabled(self):

1026

options = self.parser.parse_args()

1027

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

1028

if action == "is_enabled":

1029

continue

1030

setattr(options, action, value)

1031

options.client = ["foo"]

1032

self.check_option_syntax(options)

1033

1034

def test_two_clients_with_all_actions_except_is_enabled(self):

1035

options = self.parser.parse_args()

1036

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

1037

if action == "is_enabled":

1038

continue

1039

setattr(options, action, value)

1040

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

1041

self.check_option_syntax(options)

1042

1043

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1044

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

1045

if action == "is_enabled":

1046

continue

1047

options = self.parser.parse_args()

1048

setattr(options, action, value)

1049

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

1050

self.check_option_syntax(options)

1051

1052

def test_is_enabled_fails_without_client(self):

1053

options = self.parser.parse_args()

1054

options.is_enabled = True

1055

with self.assertParseError():

1056

self.check_option_syntax(options)

1057

1058

def test_is_enabled_fails_with_two_clients(self):

1059

options = self.parser.parse_args()

1060

options.is_enabled = True

1061

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

1062

with self.assertParseError():

1063

self.check_option_syntax(options)

1064

1065

def test_remove_can_only_be_combined_with_action_deny(self):

1066

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

1067

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

1068

continue

1069

options = self.parser.parse_args()

1070

setattr(options, action, value)

1071

options.all = True

1072

options.remove = True

1073

with self.assertParseError():

1074

self.check_option_syntax(options)

1075

1076

1077

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1078

def test_calls_and_returns_get_object_on_bus(self):

1079

class MockBus(object):

1080

called = False

1081

def get_object(mockbus_self, busname, dbus_path):

1082

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

1083

# this MockBus instance is in "mockbus_self".

1084

self.assertEqual(dbus_busname, busname)

1085

self.assertEqual(server_dbus_path, dbus_path)

1086

mockbus_self.called = True

1087

return mockbus_self

1088

1089

mockbus = get_mandos_dbus_object(bus=MockBus())

1090

self.assertIsInstance(mockbus, MockBus)

1091

self.assertTrue(mockbus.called)

1092

1093

def test_logs_and_exits_on_dbus_error(self):

1094

class MockBusFailing(object):

1095

def get_object(self, busname, dbus_path):

1096

raise dbus.exceptions.DBusException("Test")

1097

1098

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

1099

with self.assertRaises(SystemExit) as e:

1100

bus = get_mandos_dbus_object(bus=MockBusFailing())

1101

1102

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

1103

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

1104

else:

1105

self.assertIsNotNone(e.exception.code)

1106

1107

1108

class Test_get_managed_objects(TestCaseWithAssertLogs):

1109

def test_calls_and_returns_GetManagedObjects(self):

1110

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

1111

class MockObjectManager(object):

1112

def GetManagedObjects(self):

1113

return managed_objects

1114

retval = get_managed_objects(MockObjectManager())

1115

self.assertDictEqual(managed_objects, retval)

1116

1117

def test_logs_and_exits_on_dbus_error(self):

1118

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

1119

1120

class MockObjectManagerFailing(object):

1121

def GetManagedObjects(self):

1122

dbus_logger.error("Test")

1123

raise dbus.exceptions.DBusException("Test")

1124

1125

class CountingHandler(logging.Handler):

1126

count = 0

1127

def emit(self, record):

1128

self.count += 1

1129

1130

counting_handler = CountingHandler()

1131

1132

dbus_logger.addHandler(counting_handler)

1133

1134

try:

1135

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

1136

with self.assertRaises(SystemExit) as e:

1137

get_managed_objects(MockObjectManagerFailing())

1138

finally:

1139

dbus_logger.removeFilter(counting_handler)

1140

1141

# Make sure the dbus logger was suppressed

1142

self.assertEqual(0, counting_handler.count)

1143

1144

# Test that the dbus_logger still works

1145

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

1146

dbus_logger.error("Test")

1147

1148

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

1149

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

1150

else:

1151

self.assertIsNotNone(e.exception.code)

1152

1153

1154

class Test_commands_from_options(unittest.TestCase):

1155

def setUp(self):

1156

self.parser = argparse.ArgumentParser()

1157

add_command_line_options(self.parser)

1158

1159

def test_is_enabled(self):

1160

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

1161

command.IsEnabled)

1162

1163

def assert_command_from_args(self, args, command_cls,

1164

**cmd_attrs):

1165

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

1166

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

1167

options = self.parser.parse_args(args)

1168

check_option_syntax(self.parser, options)

1169

commands = commands_from_options(options)

1170

self.assertEqual(1, len(commands))

1171

command = commands[0]

1172

self.assertIsInstance(command, command_cls)

1173

for key, value in cmd_attrs.items():

1174

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

1175

1176

def test_is_enabled_short(self):

1177

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

1178

command.IsEnabled)

1179

1180

def test_approve(self):

1181

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

1182

command.Approve)

1183

1184

def test_approve_short(self):

1185

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

1186

1187

def test_deny(self):

1188

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

1189

1190

def test_deny_short(self):

1191

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

1192

1193

def test_remove(self):

1194

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

1195

command.Remove)

1196

1197

def test_deny_before_remove(self):

1198

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

1199

"foo"])

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", "foo"], command.Remove)

1217

1218

def test_dump_json(self):

1219

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

1220

command.DumpJSON)

1221

1222

def test_enable(self):

1223

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

1224

command.Enable)

1225

1226

def test_enable_short(self):

1227

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

1228

1229

def test_disable(self):

1230

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

1231

command.Disable)

1232

1233

def test_disable_short(self):

1234

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

1235

1236

def test_bump_timeout(self):

1237

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

1238

command.BumpTimeout)

1239

1240

def test_bump_timeout_short(self):

1241

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

1242

command.BumpTimeout)

1243

1244

def test_start_checker(self):

1245

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

1246

command.StartChecker)

1247

1248

def test_stop_checker(self):

1249

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

1250

command.StopChecker)

1251

1252

def test_approve_by_default(self):

1253

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

1254

command.ApproveByDefault)

1255

1256

def test_deny_by_default(self):

1257

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

1258

command.DenyByDefault)

1259

1260

def test_checker(self):

1261

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

1262

command.SetChecker,

1263

value_to_set=":")

1264

1265

def test_checker_empty(self):

1266

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

1267

command.SetChecker,

1268

value_to_set="")

1269

1270

def test_checker_short(self):

1271

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

1272

command.SetChecker,

1273

value_to_set=":")

1274

1275

def test_host(self):

1276

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

1277

"foo"], command.SetHost,

1278

value_to_set="foo.example.org")

1279

1280

def test_host_short(self):

1281

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

1282

"foo"], command.SetHost,

1283

value_to_set="foo.example.org")

1284

1285

def test_secret_devnull(self):

1286

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

1287

"foo"], command.SetSecret,

1288

value_to_set=b"")

1289

1290

def test_secret_tempfile(self):

1291

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

1292

value = b"secret\0xyzzy\nbar"

1293

f.write(value)

1294

f.seek(0)

1295

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

1296

"foo"], command.SetSecret,

1297

value_to_set=value)

1298

1299

def test_secret_devnull_short(self):

1300

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

1301

command.SetSecret,

1302

value_to_set=b"")

1303

1304

def test_secret_tempfile_short(self):

1305

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

1306

value = b"secret\0xyzzy\nbar"

1307

f.write(value)

1308

f.seek(0)

1309

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

1310

command.SetSecret,

1311

value_to_set=value)

1312

1313

def test_timeout(self):

1314

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

1315

command.SetTimeout,

1316

value_to_set=300000)

1317

1318

def test_timeout_short(self):

1319

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

1320

command.SetTimeout,

1321

value_to_set=300000)

1322

1323

def test_extended_timeout(self):

1324

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

1325

"foo"],

1326

command.SetExtendedTimeout,

1327

value_to_set=900000)

1328

1329

def test_interval(self):

1330

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

1331

command.SetInterval,

1332

value_to_set=120000)

1333

1334

def test_interval_short(self):

1335

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

1336

command.SetInterval,

1337

value_to_set=120000)

1338

1339

def test_approval_delay(self):

1340

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

1341

"foo"],

1342

command.SetApprovalDelay,

1343

value_to_set=30000)

1344

1345

def test_approval_duration(self):

1346

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

1347

"foo"],

1348

command.SetApprovalDuration,

1349

value_to_set=1000)

1350

1351

def test_print_table(self):

1352

self.assert_command_from_args([], command.PrintTable,

1353

verbose=False)

1354

1355

def test_print_table_verbose(self):

1356

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

1357

command.PrintTable,

1358

verbose=True)

1359

1360

def test_print_table_verbose_short(self):

1361

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

1362

verbose=True)

1363

1364

1365

class TestCommand(unittest.TestCase):

1366

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

1367

1368

def setUp(self):

1369

testcase = self

1370

class MockClient(object):

1371

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

1372

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

1373

self.attributes = attributes

1374

self.attributes["Name"] = name

1375

self.calls = []

1376

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

1377

testcase.assertEqual(client_dbus_interface, interface)

1378

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1379

dbus_interface)

1380

self.attributes[propname] = value

1381

def Approve(self, approve, dbus_interface):

1382

testcase.assertEqual(client_dbus_interface,

1383

dbus_interface)

1384

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

1385

dbus_interface)))

1386

self.client = MockClient(

1387

"foo",

1388

KeyID=("92ed150794387c03ce684574b1139a65"

1389

"94a34f895daaaf09fd8ea90a27cddb12"),

1390

Secret=b"secret",

1391

Host="foo.example.org",

1392

Enabled=dbus.Boolean(True),

1393

Timeout=300000,

1394

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

1395

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

1396

Interval=120000,

1397

Fingerprint=("778827225BA7DE539C5A"

1398

"7CFA59CFF7CDBD9A5920"),

1399

CheckerRunning=dbus.Boolean(False),

1400

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

1401

ApprovalPending=dbus.Boolean(False),

1402

ApprovedByDefault=dbus.Boolean(True),

1403

LastApprovalRequest="",

1404

ApprovalDelay=0,

1405

ApprovalDuration=1000,

1406

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

1407

ExtendedTimeout=900000,

1408

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

1409

LastCheckerStatus=0)

1410

self.other_client = MockClient(

1411

"barbar",

1412

KeyID=("0558568eedd67d622f5c83b35a115f79"

1413

"6ab612cff5ad227247e46c2b020f441c"),

1414

Secret=b"secretbar",

1415

Host="192.0.2.3",

1416

Enabled=dbus.Boolean(True),

1417

Timeout=300000,

1418

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

1419

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

1420

Interval=120000,

1421

Fingerprint=("3E393AEAEFB84C7E89E2"

1422

"F547B3A107558FCA3A27"),

1423

CheckerRunning=dbus.Boolean(True),

1424

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

1425

ApprovalPending=dbus.Boolean(False),

1426

ApprovedByDefault=dbus.Boolean(False),

1427

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

1428

ApprovalDelay=30000,

1429

ApprovalDuration=93785000,

1430

Checker=":",

1431

ExtendedTimeout=900000,

1432

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

1433

LastCheckerStatus=-2)

1434

self.clients = collections.OrderedDict(

1435

[

1436

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

1437

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

1438

])

1439

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

1440

1441

@property

1442

def bus(self):

1443

class Bus(object):

1444

@staticmethod

1445

def get_object(client_bus_name, path):

1446

self.assertEqual(dbus_busname, client_bus_name)

1447

return {

1448

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

1449

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

1450

"/clients/foo": self.client,

1451

"/clients/barbar": self.other_client,

1452

}[path]

1453

return Bus()

1454

1455

1456

class TestBaseCommands(TestCommand):

1457

1458

def test_IsEnabled_exits_successfully(self):

1459

with self.assertRaises(SystemExit) as e:

1460

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

1461

if e.exception.code is not None:

1462

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

1463

else:

1464

self.assertIsNone(e.exception.code)

1465

1466

def test_IsEnabled_exits_with_failure(self):

1467

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

1468

with self.assertRaises(SystemExit) as e:

1469

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

1470

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

1471

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

1472

else:

1473

self.assertIsNotNone(e.exception.code)

1474

1475

def test_Approve(self):

1476

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

1477

for clientpath in self.clients:

1478

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

1479

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

1480

client.calls)

1481

1482

def test_Deny(self):

1483

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

1484

for clientpath in self.clients:

1485

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

1486

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

1487

client.calls)

1488

1489

def test_Remove(self):

1490

class MockMandos(object):

1491

def __init__(self):

1492

self.calls = []

1493

def RemoveClient(self, dbus_path):

1494

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

1495

mandos = MockMandos()

1496

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

1497

for clientpath in self.clients:

1498

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

1499

mandos.calls)

1500

1501

expected_json = {

1502

"foo": {

1503

"Name": "foo",

1504

"KeyID": ("92ed150794387c03ce684574b1139a65"

1505

"94a34f895daaaf09fd8ea90a27cddb12"),

1506

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

1507

"Enabled": True,

1508

"Timeout": 300000,

1509

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

1510

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

1511

"Interval": 120000,

1512

"Fingerprint": ("778827225BA7DE539C5A"

1513

"7CFA59CFF7CDBD9A5920"),

1514

"CheckerRunning": False,

1515

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

1516

"ApprovalPending": False,

1517

"ApprovedByDefault": True,

1518

"LastApprovalRequest": "",

1519

"ApprovalDelay": 0,

1520

"ApprovalDuration": 1000,

1521

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

1522

"ExtendedTimeout": 900000,

1523

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

1524

"LastCheckerStatus": 0,

1525

},

1526

"barbar": {

1527

"Name": "barbar",

1528

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1529

"6ab612cff5ad227247e46c2b020f441c"),

1530

"Host": "192.0.2.3",

1531

"Enabled": True,

1532

"Timeout": 300000,

1533

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

1534

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

1535

"Interval": 120000,

1536

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1537

"F547B3A107558FCA3A27"),

1538

"CheckerRunning": True,

1539

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

1540

"ApprovalPending": False,

1541

"ApprovedByDefault": False,

1542

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

1543

"ApprovalDelay": 30000,

1544

"ApprovalDuration": 93785000,

1545

"Checker": ":",

1546

"ExtendedTimeout": 900000,

1547

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

1548

"LastCheckerStatus": -2,

1549

},

1550

}

1551

1552

def test_DumpJSON_normal(self):

1553

with self.capture_stdout_to_buffer() as buffer:

1554

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

1555

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

1556

self.assertDictEqual(self.expected_json, json_data)

1557

1558

@staticmethod

1559

@contextlib.contextmanager

1560

def capture_stdout_to_buffer():

1561

capture_buffer = io.StringIO()

1562

old_stdout = sys.stdout

1563

sys.stdout = capture_buffer

1564

try:

1565

yield capture_buffer

1566

finally:

1567

sys.stdout = old_stdout

1568

1569

def test_DumpJSON_one_client(self):

1570

with self.capture_stdout_to_buffer() as buffer:

1571

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

1572

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

1573

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

1574

self.assertDictEqual(expected_json, json_data)

1575

1576

def test_PrintTable_normal(self):

1577

with self.capture_stdout_to_buffer() as buffer:

1578

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

1579

expected_output = "\n".join((

1580

"Name Enabled Timeout Last Successful Check",

1581

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

1582

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

1583

)) + "\n"

1584

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

1585

1586

def test_PrintTable_verbose(self):

1587

with self.capture_stdout_to_buffer() as buffer:

1588

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

1589

columns = (

1590

(

1591

"Name ",

1592

"foo ",

1593

"barbar ",

1594

),(

1595

"Enabled ",

1596

"Yes ",

1597

"Yes ",

1598

),(

1599

"Timeout ",

1600

"00:05:00 ",

1601

"00:05:00 ",

1602

),(

1603

"Last Successful Check ",

1604

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

1605

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

1606

),(

1607

"Created ",

1608

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

1609

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

1610

),(

1611

"Interval ",

1612

"00:02:00 ",

1613

"00:02:00 ",

1614

),(

1615

"Host ",

1616

"foo.example.org ",

1617

"192.0.2.3 ",

1618

),(

1619

("Key ID "

1620

" "),

1621

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1622

"ea90a27cddb12 "),

1623

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1624

"46c2b020f441c "),

1625

),(

1626

"Fingerprint ",

1627

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1628

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1629

),(

1630

"Check Is Running ",

1631

"No ",

1632

"Yes ",

1633

),(

1634

"Last Enabled ",

1635

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

1636

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

1637

),(

1638

"Approval Is Pending ",

1639

"No ",

1640

"No ",

1641

),(

1642

"Approved By Default ",

1643

"Yes ",

1644

"No ",

1645

),(

1646

"Last Approval Request ",

1647

" ",

1648

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

1649

),(

1650

"Approval Delay ",

1651

"00:00:00 ",

1652

"00:00:30 ",

1653

),(

1654

"Approval Duration ",

1655

"00:00:01 ",

1656

"1T02:03:05 ",

1657

),(

1658

"Checker ",

1659

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

1660

": ",

1661

),(

1662

"Extended Timeout ",

1663

"00:15:00 ",

1664

"00:15:00 ",

1665

),(

1666

"Expires ",

1667

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

1668

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

1669

),(

1670

"Last Checker Status",

1671

"0 ",

1672

"-2 ",

1673

)

1674

)

1675

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

1676

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

1677

for rows in columns)

1678

for line in range(num_lines))

1679

+ "\n")

1680

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

1681

1682

def test_PrintTable_one_client(self):

1683

with self.capture_stdout_to_buffer() as buffer:

1684

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

1685

expected_output = "\n".join((

1686

"Name Enabled Timeout Last Successful Check",

1687

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

1688

)) + "\n"

1689

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

1690

1691

1692

class TestPropertyCmd(TestCommand):

1693

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

1694

def runTest(self):

1695

if not hasattr(self, "command"):

1696

return

1697

values_to_get = getattr(self, "values_to_get",

1698

self.values_to_set)

1699

for value_to_set, value_to_get in zip(self.values_to_set,

1700

values_to_get):

1701

for clientpath in self.clients:

1702

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

1703

old_value = client.attributes[self.propname]

1704

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

1705

self.run_command(value_to_set, self.clients)

1706

for clientpath in self.clients:

1707

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

1708

value = client.attributes[self.propname]

1709

self.assertNotIsInstance(value, self.Unique)

1710

self.assertEqual(value_to_get, value)

1711

1712

class Unique(object):

1713

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

1714

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

1715

1716

def run_command(self, value, clients):

1717

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

1718

1719

1720

class TestEnableCmd(TestPropertyCmd):

1721

command = command.Enable

1722

propname = "Enabled"

1723

values_to_set = [dbus.Boolean(True)]

1724

1725

1726

class TestDisableCmd(TestPropertyCmd):

1727

command = command.Disable

1728

propname = "Enabled"

1729

values_to_set = [dbus.Boolean(False)]

1730

1731

1732

class TestBumpTimeoutCmd(TestPropertyCmd):

1733

command = command.BumpTimeout

1734

propname = "LastCheckedOK"

1735

values_to_set = [""]

1736

1737

1738

class TestStartCheckerCmd(TestPropertyCmd):

1739

command = command.StartChecker

1740

propname = "CheckerRunning"

1741

values_to_set = [dbus.Boolean(True)]

1742

1743

1744

class TestStopCheckerCmd(TestPropertyCmd):

1745

command = command.StopChecker

1746

propname = "CheckerRunning"

1747

values_to_set = [dbus.Boolean(False)]

1748

1749

1750

class TestApproveByDefaultCmd(TestPropertyCmd):

1751

command = command.ApproveByDefault

1752

propname = "ApprovedByDefault"

1753

values_to_set = [dbus.Boolean(True)]

1754

1755

1756

class TestDenyByDefaultCmd(TestPropertyCmd):

1757

command = command.DenyByDefault

1758

propname = "ApprovedByDefault"

1759

values_to_set = [dbus.Boolean(False)]

1760

1761

1762

class TestPropertyValueCmd(TestPropertyCmd):

1763

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

1764

1765

def runTest(self):

1766

if type(self) is TestPropertyValueCmd:

1767

return

1768

return super(TestPropertyValueCmd, self).runTest()

1769

1770

def run_command(self, value, clients):

1771

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

1772

1773

1774

class TestSetCheckerCmd(TestPropertyValueCmd):

1775

command = command.SetChecker

1776

propname = "Checker"

1777

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

1778

1779

1780

class TestSetHostCmd(TestPropertyValueCmd):

1781

command = command.SetHost

1782

propname = "Host"

1783

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

1784

1785

1786

class TestSetSecretCmd(TestPropertyValueCmd):

1787

command = command.SetSecret

1788

propname = "Secret"

1789

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

1790

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

1791

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

1792

1793

1794

class TestSetTimeoutCmd(TestPropertyValueCmd):

1795

command = command.SetTimeout

1796

propname = "Timeout"

1797

values_to_set = [datetime.timedelta(),

1798

datetime.timedelta(minutes=5),

1799

datetime.timedelta(seconds=1),

1800

datetime.timedelta(weeks=1),

1801

datetime.timedelta(weeks=52)]

1802

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

1803

1804

1805

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1806

command = command.SetExtendedTimeout

1807

propname = "ExtendedTimeout"

1808

values_to_set = [datetime.timedelta(),

1809

datetime.timedelta(minutes=5),

1810

datetime.timedelta(seconds=1),

1811

datetime.timedelta(weeks=1),

1812

datetime.timedelta(weeks=52)]

1813

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

1814

1815

1816

class TestSetIntervalCmd(TestPropertyValueCmd):

1817

command = command.SetInterval

1818

propname = "Interval"

1819

values_to_set = [datetime.timedelta(),

1820

datetime.timedelta(minutes=5),

1821

datetime.timedelta(seconds=1),

1822

datetime.timedelta(weeks=1),

1823

datetime.timedelta(weeks=52)]

1824

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

1825

1826

1827

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1828

command = command.SetApprovalDelay

1829

propname = "ApprovalDelay"

1830

values_to_set = [datetime.timedelta(),

1831

datetime.timedelta(minutes=5),

1832

datetime.timedelta(seconds=1),

1833

datetime.timedelta(weeks=1),

1834

datetime.timedelta(weeks=52)]

1835

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

1836

1837

1838

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1839

command = command.SetApprovalDuration

1840

propname = "ApprovalDuration"

1841

values_to_set = [datetime.timedelta(),

1842

datetime.timedelta(minutes=5),

1843

datetime.timedelta(seconds=1),

1844

datetime.timedelta(weeks=1),

1845

datetime.timedelta(weeks=52)]

1846

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

1847

1848

1849

1850

def should_only_run_tests():

1851

parser = argparse.ArgumentParser(add_help=False)

1852

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

1853

args, unknown_args = parser.parse_known_args()

1854

run_tests = args.check

1855

if run_tests:

1856

# Remove --check argument from sys.argv

1857

sys.argv[1:] = unknown_args

1858

return run_tests

1859

1860

# Add all tests from doctest strings

1861

def load_tests(loader, tests, none):

1862

import doctest

1863

tests.addTests(doctest.DocTestSuite())

1864

return tests

1865

1866

if __name__ == "__main__":

1867

try:

1868

if should_only_run_tests():

1869

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

1870

unittest.main()

1871

else:

1872

main()

1873

finally:

1874

logging.shutdown()