/mandos/trunk : revision 1087

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

21

# <http://www.gnu.org/licenses/>.

22

#

22

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

23

#

23

24

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

24

#

25

#

25

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

from future_builtins import *

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

30

34

31

35

import sys

32

import dbus

33

36

import argparse

34

37

import locale

35

38

import datetime

36

39

import re

37

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

import StringIO

65

io.StringIO = StringIO.StringIO

38

66

39

67

locale.setlocale(locale.LC_ALL, "")

40

68

41

tablewords = {

42

"Name": "Name",

43

"Enabled": "Enabled",

44

"Timeout": "Timeout",

45

"LastCheckedOK": "Last Successful Check",

46

"LastApprovalRequest": "Last Approval Request",

47

"Created": "Created",

48

"Interval": "Interval",

49

"Host": "Host",

50

"Fingerprint": "Fingerprint",

51

"CheckerRunning": "Check Is Running",

52

"LastEnabled": "Last Enabled",

53

"ApprovalPending": "Approval Is Pending",

54

"ApprovedByDefault": "Approved By Default",

55

"ApprovalDelay": "Approval Delay",

56

"ApprovalDuration": "Approval Duration",

57

"Checker": "Checker",

58

"ExtendedTimeout" : "Extended Timeout"

59

}

60

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

61

domain = "se.recompile"

62

busname = domain + ".Mandos"

63

server_path = "/"

64

server_interface = domain + ".Mandos"

65

client_interface = domain + ".Mandos.Client"

66

version = "1.5.5"

67

68

def timedelta_to_milliseconds(td):

69

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

70

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

71

+ (td.seconds * 1000)

72

+ (td.microseconds // 1000))

73

74

def milliseconds_to_string(ms):

75

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

76

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

77

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

78

hours = td.seconds // 3600,

79

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

80

seconds = td.seconds % 60,

81

))

82

83

def string_to_delta(interval):

84

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

85

86

>>> string_to_delta("7d")

87

datetime.timedelta(7)

88

>>> string_to_delta("60s")

89

datetime.timedelta(0, 60)

90

>>> string_to_delta("60m")

91

datetime.timedelta(0, 3600)

92

>>> string_to_delta("24h")

93

datetime.timedelta(1)

94

>>> string_to_delta("1w")

95

datetime.timedelta(7)

96

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

97

datetime.timedelta(0, 330)

98

"""

99

value = datetime.timedelta(0)

100

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

101

102

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

103

if suffix == "d":

104

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

105

elif suffix == "s":

106

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

107

elif suffix == "m":

108

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

109

elif suffix == "h":

110

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

111

elif suffix == "w":

112

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

113

elif suffix == "":

114

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

115

return value

116

117

def print_clients(clients, keywords):

118

def valuetostring(value, keyword):

119

if type(value) is dbus.Boolean:

120

return "Yes" if value else "No"

121

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

122

"ApprovalDuration", "ExtendedTimeout"):

123

return milliseconds_to_string(value)

124

return unicode(value)

125

126

# Create format string to print table rows

127

format_string = " ".join("{{{key}:{width}}}".format(

128

width = max(len(tablewords[key]),

129

max(len(valuetostring(client[key],

130

key))

131

for client in

132

clients)),

133

key = key) for key in keywords)

134

# Print header line

135

print(format_string.format(**tablewords))

136

for client in clients:

137

print(format_string.format(**dict((key,

138

valuetostring(client[key],

139

key))

140

for key in keywords)))

141

142

def has_actions(options):

143

return any((options.enable,

144

options.disable,

145

options.bump_timeout,

146

options.start_checker,

147

options.stop_checker,

148

options.is_enabled,

149

options.remove,

150

options.checker is not None,

151

options.timeout is not None,

152

options.extended_timeout is not None,

153

options.interval is not None,

154

options.approved_by_default is not None,

155

options.approval_delay is not None,

156

options.approval_duration is not None,

157

options.host is not None,

158

options.secret is not None,

159

options.approve,

160

options.deny))

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

161

83

162

84

def main():

163

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

164

135

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

165

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

136

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

166

137

help="show version number and exit")

167

138

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

168

139

help="Select all clients")

169

140

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

170

141

help="Print all fields")

171

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

172

help="Enable client")

173

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

174

help="disable client")

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

175

150

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

176

151

help="Bump timeout for client")

177

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

178

help="Start checker for client")

179

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

180

help="Stop checker 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")

181

159

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

182

160

help="Check if client is enabled")

183

161

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

184

162

help="Remove client")

185

163

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

186

164

help="Set checker command for client")

187

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

165

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

188

166

help="Set timeout for client")

189

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

167

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

190

168

help="Set extended timeout for client")

191

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

169

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

192

170

help="Set checker interval for client")

193

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

194

default=None, dest="approved_by_default",

195

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

196

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

197

dest="approved_by_default",

198

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

199

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

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,

200

181

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

201

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

182

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

202

183

help="Set duration of one client approval")

203

184

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

204

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

185

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

186

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

205

187

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

206

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

207

help="Approve any current client request")

208

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

209

help="Deny any current client request")

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

210

198

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

211

options = parser.parse_args()

212

199

200

201

def string_to_delta(interval):

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

213

409

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

214

410

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

215

411

if options.verbose and has_actions(options):

216

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

217

" --all.")

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

218

416

if options.all and not has_actions(options):

219

417

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

220

221

try:

222

bus = dbus.SystemBus()

223

mandos_dbus_objc = bus.get_object(busname, server_path)

224

except dbus.exceptions.DBusException:

225

print("Could not connect to Mandos server",

226

file=sys.stderr)

227

sys.exit(1)

228

229

mandos_serv = dbus.Interface(mandos_dbus_objc,

230

dbus_interface = server_interface)

231

232

#block stderr since dbus library prints to stderr

233

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

234

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

235

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

236

os.close(null)

237

try:

238

try:

239

mandos_clients = mandos_serv.GetAllClientsWithProperties()

240

finally:

241

#restore stderr

242

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

243

os.close(stderrcopy)

244

except dbus.exceptions.DBusException:

245

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

246

file=sys.stderr)

247

sys.exit(1)

248

249

# Compile dict of (clients: properties) to process

250

clients={}

251

252

if options.all or not options.client:

253

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

254

for path, properties in

255

mandos_clients.iteritems())

256

else:

257

for name in options.client:

258

for path, client in mandos_clients.iteritems():

259

if client["Name"] == name:

260

client_objc = bus.get_object(busname, path)

261

clients[client_objc] = client

262

break

263

else:

264

print("Client not found on server: {0!r}"

265

.format(name), file=sys.stderr)

266

sys.exit(1)

267

268

if not has_actions(options) and clients:

269

if options.verbose:

270

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

271

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

272

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

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

self.mandos = mandos

560

for clientpath, properties in clients.items():

561

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

562

dbus_busname, str(clientpath))

563

client = bus.get_object(dbus_busname, clientpath)

564

self.run_on_one_client(client, properties)

565

566

567

class IsEnabled(Base):

568

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

569

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

570

if self.is_enabled(client, properties):

571

sys.exit(0)

572

sys.exit(1)

573

def is_enabled(self, client, properties):

574

return properties["Enabled"]

575

576

577

class Approve(Base):

578

def run_on_one_client(self, client, properties):

579

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

580

client.__dbus_object_path__,

581

client_dbus_interface)

582

client.Approve(dbus.Boolean(True),

583

dbus_interface=client_dbus_interface)

584

585

586

class Deny(Base):

587

def run_on_one_client(self, client, properties):

588

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

589

client.__dbus_object_path__,

590

client_dbus_interface)

591

client.Approve(dbus.Boolean(False),

592

dbus_interface=client_dbus_interface)

593

594

595

class Remove(Base):

596

def run_on_one_client(self, client, properties):

597

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

598

dbus_busname, server_dbus_path,

599

server_dbus_interface,

600

str(client.__dbus_object_path__))

601

self.mandos.RemoveClient(client.__dbus_object_path__)

602

603

604

class Output(Base):

605

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

606

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

607

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

608

"Fingerprint", "CheckerRunning",

273

609

"LastEnabled", "ApprovalPending",

274

"ApprovedByDefault",

275

"LastApprovalRequest", "ApprovalDelay",

276

"ApprovalDuration", "Checker",

277

"ExtendedTimeout")

278

else:

279

keywords = defaultkeywords

280

281

print_clients(clients.values(), keywords)

282

else:

283

# Process each client in the list by all selected options

284

for client in clients:

285

def set_client_prop(prop, value):

286

"""Set a Client D-Bus property"""

287

client.Set(client_interface, prop, value,

288

dbus_interface=dbus.PROPERTIES_IFACE)

289

def set_client_prop_ms(prop, value):

290

"""Set a Client D-Bus property, converted

291

from a string to milliseconds."""

292

set_client_prop(prop,

293

timedelta_to_milliseconds

294

(string_to_delta(value)))

295

if options.remove:

296

mandos_serv.RemoveClient(client.__dbus_object_path__)

297

if options.enable:

298

set_client_prop("Enabled", dbus.Boolean(True))

299

if options.disable:

300

set_client_prop("Enabled", dbus.Boolean(False))

301

if options.bump_timeout:

302

set_client_prop("LastCheckedOK", "")

303

if options.start_checker:

304

set_client_prop("CheckerRunning", dbus.Boolean(True))

305

if options.stop_checker:

306

set_client_prop("CheckerRunning", dbus.Boolean(False))

307

if options.is_enabled:

308

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

309

"Enabled",

310

dbus_interface=

311

dbus.PROPERTIES_IFACE)

312

else 1)

313

if options.checker is not None:

314

set_client_prop("Checker", options.checker)

315

if options.host is not None:

316

set_client_prop("Host", options.host)

317

if options.interval is not None:

318

set_client_prop_ms("Interval", options.interval)

319

if options.approval_delay is not None:

320

set_client_prop_ms("ApprovalDelay",

321

options.approval_delay)

322

if options.approval_duration is not None:

323

set_client_prop_ms("ApprovalDuration",

324

options.approval_duration)

325

if options.timeout is not None:

326

set_client_prop_ms("Timeout", options.timeout)

327

if options.extended_timeout is not None:

328

set_client_prop_ms("ExtendedTimeout",

329

options.extended_timeout)

330

if options.secret is not None:

331

set_client_prop("Secret",

332

dbus.ByteArray(options.secret.read()))

333

if options.approved_by_default is not None:

334

set_client_prop("ApprovedByDefault",

335

dbus.Boolean(options

336

.approved_by_default))

337

if options.approve:

338

client.Approve(dbus.Boolean(True),

339

dbus_interface=client_interface)

340

elif options.deny:

341

client.Approve(dbus.Boolean(False),

342

dbus_interface=client_interface)

610

"ApprovedByDefault", "LastApprovalRequest",

611

"ApprovalDelay", "ApprovalDuration",

612

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

613

"LastCheckerStatus")

614

615

616

class DumpJSON(Output):

617

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

618

data = {client["Name"]:

619

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

620

for key in self.all_keywords}

621

for client in clients.values()}

622

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

623

624

@staticmethod

625

def dbus_boolean_to_bool(value):

626

if isinstance(value, dbus.Boolean):

627

value = bool(value)

628

return value

629

630

631

class PrintTable(Output):

632

def __init__(self, verbose=False):

633

self.verbose = verbose

634

635

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

636

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

637

"LastCheckedOK")

638

keywords = default_keywords

639

if self.verbose:

640

keywords = self.all_keywords

641

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

642

643

class TableOfClients(object):

644

tableheaders = {

645

"Name": "Name",

646

"Enabled": "Enabled",

647

"Timeout": "Timeout",

648

"LastCheckedOK": "Last Successful Check",

649

"LastApprovalRequest": "Last Approval Request",

650

"Created": "Created",

651

"Interval": "Interval",

652

"Host": "Host",

653

"Fingerprint": "Fingerprint",

654

"KeyID": "Key ID",

655

"CheckerRunning": "Check Is Running",

656

"LastEnabled": "Last Enabled",

657

"ApprovalPending": "Approval Is Pending",

658

"ApprovedByDefault": "Approved By Default",

659

"ApprovalDelay": "Approval Delay",

660

"ApprovalDuration": "Approval Duration",

661

"Checker": "Checker",

662

"ExtendedTimeout": "Extended Timeout",

663

"Expires": "Expires",

664

"LastCheckerStatus": "Last Checker Status",

665

}

666

667

def __init__(self, clients, keywords):

668

self.clients = clients

669

self.keywords = keywords

670

671

def __str__(self):

672

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

673

674

if sys.version_info.major == 2:

675

__unicode__ = __str__

676

def __str__(self):

677

return str(self).encode(

678

locale.getpreferredencoding())

679

680

def rows(self):

681

format_string = self.row_formatting_string()

682

rows = [self.header_line(format_string)]

683

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

684

for client in self.clients)

685

return rows

686

687

def row_formatting_string(self):

688

"Format string used to format table rows"

689

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

690

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

691

*(len(self.string_from_client(client,

692

key))

693

for client in self.clients)),

694

key=key)

695

for key in self.keywords)

696

697

def string_from_client(self, client, key):

698

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

699

700

@classmethod

701

def valuetostring(cls, value, keyword):

702

if isinstance(value, dbus.Boolean):

703

return "Yes" if value else "No"

704

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

705

"ApprovalDuration", "ExtendedTimeout"):

706

return cls.milliseconds_to_string(value)

707

return str(value)

708

709

def header_line(self, format_string):

710

return format_string.format(**self.tableheaders)

711

712

def client_line(self, client, format_string):

713

return format_string.format(

714

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

715

for key in self.keywords})

716

717

@staticmethod

718

def milliseconds_to_string(ms):

719

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

720

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

721

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

722

if td.days else "",

723

hours=td.seconds // 3600,

724

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

725

seconds=td.seconds % 60))

726

727

728

class PropertySetter(Base):

729

"Abstract class for Actions for setting one client property"

730

731

def run_on_one_client(self, client, properties):

732

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

733

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

734

client.__dbus_object_path__,

735

dbus.PROPERTIES_IFACE, client_dbus_interface,

736

self.propname, self.value_to_set

737

if not isinstance(self.value_to_set,

738

dbus.Boolean)

739

else bool(self.value_to_set))

740

client.Set(client_dbus_interface, self.propname,

741

self.value_to_set,

742

dbus_interface=dbus.PROPERTIES_IFACE)

743

744

@property

745

def propname(self):

746

raise NotImplementedError()

747

748

749

class Enable(PropertySetter):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(True)

752

753

754

class Disable(PropertySetter):

755

propname = "Enabled"

756

value_to_set = dbus.Boolean(False)

757

758

759

class BumpTimeout(PropertySetter):

760

propname = "LastCheckedOK"

761

value_to_set = ""

762

763

764

class StartChecker(PropertySetter):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(True)

767

768

769

class StopChecker(PropertySetter):

770

propname = "CheckerRunning"

771

value_to_set = dbus.Boolean(False)

772

773

774

class ApproveByDefault(PropertySetter):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(True)

777

778

779

class DenyByDefault(PropertySetter):

780

propname = "ApprovedByDefault"

781

value_to_set = dbus.Boolean(False)

782

783

784

class PropertySetterValue(PropertySetter):

785

"""Abstract class for PropertySetter recieving a value as

786

constructor argument instead of a class attribute."""

787

def __init__(self, value):

788

self.value_to_set = value

789

790

791

class SetChecker(PropertySetterValue):

792

propname = "Checker"

793

794

795

class SetHost(PropertySetterValue):

796

propname = "Host"

797

798

799

class SetSecret(PropertySetterValue):

800

propname = "Secret"

801

802

@property

803

def value_to_set(self):

804

return self._vts

805

806

@value_to_set.setter

807

def value_to_set(self, value):

808

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

809

self._vts = value.read()

810

value.close()

811

812

813

class PropertySetterValueMilliseconds(PropertySetterValue):

814

"""Abstract class for PropertySetterValue taking a value

815

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

816

milliseconds."""

817

818

@property

819

def value_to_set(self):

820

return self._vts

821

822

@value_to_set.setter

823

def value_to_set(self, value):

824

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

825

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

826

827

828

class SetTimeout(PropertySetterValueMilliseconds):

829

propname = "Timeout"

830

831

832

class SetExtendedTimeout(PropertySetterValueMilliseconds):

833

propname = "ExtendedTimeout"

834

835

836

class SetInterval(PropertySetterValueMilliseconds):

837

propname = "Interval"

838

839

840

class SetApprovalDelay(PropertySetterValueMilliseconds):

841

propname = "ApprovalDelay"

842

843

844

class SetApprovalDuration(PropertySetterValueMilliseconds):

845

propname = "ApprovalDuration"

846

847

848

849

class TestCaseWithAssertLogs(unittest.TestCase):

850

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

851

852

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

853

@contextlib.contextmanager

854

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

855

capturing_handler = self.CapturingLevelHandler(level)

856

old_level = logger.level

857

old_propagate = logger.propagate

858

logger.addHandler(capturing_handler)

859

logger.setLevel(level)

860

logger.propagate = False

861

try:

862

yield capturing_handler.watcher

863

finally:

864

logger.propagate = old_propagate

865

logger.removeHandler(capturing_handler)

866

logger.setLevel(old_level)

867

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

868

0)

869

870

class CapturingLevelHandler(logging.Handler):

871

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

872

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

873

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

874

def emit(self, record):

875

self.watcher.records.append(record)

876

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

877

878

LoggingWatcher = collections.namedtuple("LoggingWatcher",

879

("records",

880

"output"))

881

882

883

class Test_string_to_delta(TestCaseWithAssertLogs):

884

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

885

# rfc3339_duration_to_delta() already has more comprehensive

886

# tests, which is run by doctest.

887

888

def test_rfc3339_zero_seconds(self):

889

self.assertEqual(datetime.timedelta(),

890

string_to_delta("PT0S"))

891

892

def test_rfc3339_zero_days(self):

893

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

894

895

def test_rfc3339_one_second(self):

896

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

897

string_to_delta("PT1S"))

898

899

def test_rfc3339_two_hours(self):

900

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

901

string_to_delta("PT2H"))

902

903

def test_falls_back_to_pre_1_6_1_with_warning(self):

904

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

905

value = string_to_delta("2h")

906

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

907

908

909

class Test_check_option_syntax(unittest.TestCase):

910

def setUp(self):

911

self.parser = argparse.ArgumentParser()

912

add_command_line_options(self.parser)

913

914

def test_actions_requires_client_or_all(self):

915

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

916

options = self.parser.parse_args()

917

setattr(options, action, value)

918

with self.assertParseError():

919

self.check_option_syntax(options)

920

921

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

922

# check_option_syntax()

923

actions = {

924

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

925

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

926

# never used

927

"enable": True,

928

"disable": True,

929

"bump_timeout": True,

930

"start_checker": True,

931

"stop_checker": True,

932

"is_enabled": True,

933

"remove": True,

934

"checker": "x",

935

"timeout": datetime.timedelta(),

936

"extended_timeout": datetime.timedelta(),

937

"interval": datetime.timedelta(),

938

"approved_by_default": True,

939

"approval_delay": datetime.timedelta(),

940

"approval_duration": datetime.timedelta(),

941

"host": "x",

942

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

943

"approve": True,

944

"deny": True,

945

}

946

947

@contextlib.contextmanager

948

def assertParseError(self):

949

with self.assertRaises(SystemExit) as e:

950

with self.redirect_stderr_to_devnull():

951

yield

952

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

953

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

954

# /argparse.html#exiting-methods

955

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

956

957

@staticmethod

958

@contextlib.contextmanager

959

def redirect_stderr_to_devnull():

960

old_stderr = sys.stderr

961

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

962

sys.stderr = null

963

try:

964

yield

965

finally:

966

sys.stderr = old_stderr

967

968

def check_option_syntax(self, options):

969

check_option_syntax(self.parser, options)

970

971

def test_actions_all_conflicts_with_verbose(self):

972

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

973

options = self.parser.parse_args()

974

setattr(options, action, value)

975

options.all = True

976

options.verbose = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

def test_actions_with_client_conflicts_with_verbose(self):

981

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

982

options = self.parser.parse_args()

983

setattr(options, action, value)

984

options.verbose = True

985

options.client = ["client"]

986

with self.assertParseError():

987

self.check_option_syntax(options)

988

989

def test_dump_json_conflicts_with_verbose(self):

990

options = self.parser.parse_args()

991

options.dump_json = True

992

options.verbose = True

993

with self.assertParseError():

994

self.check_option_syntax(options)

995

996

def test_dump_json_conflicts_with_action(self):

997

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

998

options = self.parser.parse_args()

999

setattr(options, action, value)

1000

options.dump_json = True

1001

with self.assertParseError():

1002

self.check_option_syntax(options)

1003

1004

def test_all_can_not_be_alone(self):

1005

options = self.parser.parse_args()

1006

options.all = True

1007

with self.assertParseError():

1008

self.check_option_syntax(options)

1009

1010

def test_all_is_ok_with_any_action(self):

1011

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

1012

options = self.parser.parse_args()

1013

setattr(options, action, value)

1014

options.all = True

1015

self.check_option_syntax(options)

1016

1017

def test_any_action_is_ok_with_one_client(self):

1018

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

1019

options = self.parser.parse_args()

1020

setattr(options, action, value)

1021

options.client = ["client"]

1022

self.check_option_syntax(options)

1023

1024

def test_one_client_with_all_actions_except_is_enabled(self):

1025

options = self.parser.parse_args()

1026

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

1027

if action == "is_enabled":

1028

continue

1029

setattr(options, action, value)

1030

options.client = ["client"]

1031

self.check_option_syntax(options)

1032

1033

def test_two_clients_with_all_actions_except_is_enabled(self):

1034

options = self.parser.parse_args()

1035

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

1036

if action == "is_enabled":

1037

continue

1038

setattr(options, action, value)

1039

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

1040

self.check_option_syntax(options)

1041

1042

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1043

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

1044

if action == "is_enabled":

1045

continue

1046

options = self.parser.parse_args()

1047

setattr(options, action, value)

1048

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

1049

self.check_option_syntax(options)

1050

1051

def test_is_enabled_fails_without_client(self):

1052

options = self.parser.parse_args()

1053

options.is_enabled = True

1054

with self.assertParseError():

1055

self.check_option_syntax(options)

1056

1057

def test_is_enabled_fails_with_two_clients(self):

1058

options = self.parser.parse_args()

1059

options.is_enabled = True

1060

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

1061

with self.assertParseError():

1062

self.check_option_syntax(options)

1063

1064

def test_remove_can_only_be_combined_with_action_deny(self):

1065

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

1066

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

1067

continue

1068

options = self.parser.parse_args()

1069

setattr(options, action, value)

1070

options.all = True

1071

options.remove = True

1072

with self.assertParseError():

1073

self.check_option_syntax(options)

1074

1075

1076

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1077

def test_calls_and_returns_get_object_on_bus(self):

1078

class MockBus(object):

1079

called = False

1080

def get_object(mockbus_self, busname, dbus_path):

1081

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

1082

# this MockBus instance is in "mockbus_self".

1083

self.assertEqual(dbus_busname, busname)

1084

self.assertEqual(server_dbus_path, dbus_path)

1085

mockbus_self.called = True

1086

return mockbus_self

1087

1088

mockbus = get_mandos_dbus_object(bus=MockBus())

1089

self.assertIsInstance(mockbus, MockBus)

1090

self.assertTrue(mockbus.called)

1091

1092

def test_logs_and_exits_on_dbus_error(self):

1093

class FailingBusStub(object):

1094

def get_object(self, busname, dbus_path):

1095

raise dbus.exceptions.DBusException("Test")

1096

1097

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

1098

with self.assertRaises(SystemExit) as e:

1099

bus = get_mandos_dbus_object(bus=FailingBusStub())

1100

1101

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

1102

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

1103

else:

1104

self.assertIsNotNone(e.exception.code)

1105

1106

1107

class Test_get_managed_objects(TestCaseWithAssertLogs):

1108

def test_calls_and_returns_GetManagedObjects(self):

1109

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

1110

class ObjectManagerStub(object):

1111

def GetManagedObjects(self):

1112

return managed_objects

1113

retval = get_managed_objects(ObjectManagerStub())

1114

self.assertDictEqual(managed_objects, retval)

1115

1116

def test_logs_and_exits_on_dbus_error(self):

1117

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

1118

1119

class ObjectManagerFailingStub(object):

1120

def GetManagedObjects(self):

1121

dbus_logger.error("Test")

1122

raise dbus.exceptions.DBusException("Test")

1123

1124

class CountingHandler(logging.Handler):

1125

count = 0

1126

def emit(self, record):

1127

self.count += 1

1128

1129

counting_handler = CountingHandler()

1130

1131

dbus_logger.addHandler(counting_handler)

1132

1133

try:

1134

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

1135

with self.assertRaises(SystemExit) as e:

1136

get_managed_objects(ObjectManagerFailingStub())

1137

finally:

1138

dbus_logger.removeFilter(counting_handler)

1139

1140

# Make sure the dbus logger was suppressed

1141

self.assertEqual(0, counting_handler.count)

1142

1143

# Test that the dbus_logger still works

1144

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

1145

dbus_logger.error("Test")

1146

1147

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

1148

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

1149

else:

1150

self.assertIsNotNone(e.exception.code)

1151

1152

1153

class Test_commands_from_options(unittest.TestCase):

1154

def setUp(self):

1155

self.parser = argparse.ArgumentParser()

1156

add_command_line_options(self.parser)

1157

1158

def test_is_enabled(self):

1159

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

1160

command.IsEnabled)

1161

1162

def assert_command_from_args(self, args, command_cls,

1163

**cmd_attrs):

1164

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

1165

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

1166

options = self.parser.parse_args(args)

1167

check_option_syntax(self.parser, options)

1168

commands = commands_from_options(options)

1169

self.assertEqual(1, len(commands))

1170

command = commands[0]

1171

self.assertIsInstance(command, command_cls)

1172

for key, value in cmd_attrs.items():

1173

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

1174

1175

def test_is_enabled_short(self):

1176

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

1177

command.IsEnabled)

1178

1179

def test_approve(self):

1180

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

1181

command.Approve)

1182

1183

def test_approve_short(self):

1184

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

1185

command.Approve)

1186

1187

def test_deny(self):

1188

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

1189

command.Deny)

1190

1191

def test_deny_short(self):

1192

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

1193

1194

def test_remove(self):

1195

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

1196

command.Remove)

1197

1198

def test_deny_before_remove(self):

1199

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

1200

"client"])

1201

check_option_syntax(self.parser, options)

1202

commands = commands_from_options(options)

1203

self.assertEqual(2, len(commands))

1204

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

1205

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

1206

1207

def test_deny_before_remove_reversed(self):

1208

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

1209

"--all"])

1210

check_option_syntax(self.parser, options)

1211

commands = commands_from_options(options)

1212

self.assertEqual(2, len(commands))

1213

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

1214

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

1215

1216

def test_remove_short(self):

1217

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

1218

command.Remove)

1219

1220

def test_dump_json(self):

1221

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

1222

command.DumpJSON)

1223

1224

def test_enable(self):

1225

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

1226

command.Enable)

1227

1228

def test_enable_short(self):

1229

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

1230

command.Enable)

1231

1232

def test_disable(self):

1233

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

1234

command.Disable)

1235

1236

def test_disable_short(self):

1237

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

1238

command.Disable)

1239

1240

def test_bump_timeout(self):

1241

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

1242

command.BumpTimeout)

1243

1244

def test_bump_timeout_short(self):

1245

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

1246

command.BumpTimeout)

1247

1248

def test_start_checker(self):

1249

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

1250

command.StartChecker)

1251

1252

def test_stop_checker(self):

1253

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

1254

command.StopChecker)

1255

1256

def test_approve_by_default(self):

1257

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

1258

"client"],

1259

command.ApproveByDefault)

1260

1261

def test_deny_by_default(self):

1262

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

1263

command.DenyByDefault)

1264

1265

def test_checker(self):

1266

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

1267

command.SetChecker,

1268

value_to_set=":")

1269

1270

def test_checker_empty(self):

1271

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

1272

command.SetChecker,

1273

value_to_set="")

1274

1275

def test_checker_short(self):

1276

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

1277

command.SetChecker,

1278

value_to_set=":")

1279

1280

def test_host(self):

1281

self.assert_command_from_args(

1282

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

1283

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

1284

1285

def test_host_short(self):

1286

self.assert_command_from_args(

1287

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

1288

value_to_set="client.example.org")

1289

1290

def test_secret_devnull(self):

1291

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

1292

"client"], command.SetSecret,

1293

value_to_set=b"")

1294

1295

def test_secret_tempfile(self):

1296

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

1297

value = b"secret\0xyzzy\nbar"

1298

f.write(value)

1299

f.seek(0)

1300

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

1301

"client"],

1302

command.SetSecret,

1303

value_to_set=value)

1304

1305

def test_secret_devnull_short(self):

1306

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

1307

"client"], command.SetSecret,

1308

value_to_set=b"")

1309

1310

def test_secret_tempfile_short(self):

1311

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

1312

value = b"secret\0xyzzy\nbar"

1313

f.write(value)

1314

f.seek(0)

1315

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

1316

command.SetSecret,

1317

value_to_set=value)

1318

1319

def test_timeout(self):

1320

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

1321

command.SetTimeout,

1322

value_to_set=300000)

1323

1324

def test_timeout_short(self):

1325

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

1326

command.SetTimeout,

1327

value_to_set=300000)

1328

1329

def test_extended_timeout(self):

1330

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

1331

"client"],

1332

command.SetExtendedTimeout,

1333

value_to_set=900000)

1334

1335

def test_interval(self):

1336

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

1337

"client"], command.SetInterval,

1338

value_to_set=120000)

1339

1340

def test_interval_short(self):

1341

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

1342

command.SetInterval,

1343

value_to_set=120000)

1344

1345

def test_approval_delay(self):

1346

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

1347

"client"],

1348

command.SetApprovalDelay,

1349

value_to_set=30000)

1350

1351

def test_approval_duration(self):

1352

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

1353

"client"],

1354

command.SetApprovalDuration,

1355

value_to_set=1000)

1356

1357

def test_print_table(self):

1358

self.assert_command_from_args([], command.PrintTable,

1359

verbose=False)

1360

1361

def test_print_table_verbose(self):

1362

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

1363

command.PrintTable,

1364

verbose=True)

1365

1366

def test_print_table_verbose_short(self):

1367

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

1368

verbose=True)

1369

1370

1371

class TestCommand(unittest.TestCase):

1372

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

1373

1374

def setUp(self):

1375

testcase = self

1376

class MockClient(object):

1377

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

1378

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

1379

self.attributes = attributes

1380

self.attributes["Name"] = name

1381

self.calls = []

1382

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

1383

testcase.assertEqual(client_dbus_interface, interface)

1384

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1385

dbus_interface)

1386

self.attributes[propname] = value

1387

def Approve(self, approve, dbus_interface):

1388

testcase.assertEqual(client_dbus_interface,

1389

dbus_interface)

1390

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

1391

dbus_interface)))

1392

self.client = MockClient(

1393

"foo",

1394

KeyID=("92ed150794387c03ce684574b1139a65"

1395

"94a34f895daaaf09fd8ea90a27cddb12"),

1396

Secret=b"secret",

1397

Host="foo.example.org",

1398

Enabled=dbus.Boolean(True),

1399

Timeout=300000,

1400

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

1401

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

1402

Interval=120000,

1403

Fingerprint=("778827225BA7DE539C5A"

1404

"7CFA59CFF7CDBD9A5920"),

1405

CheckerRunning=dbus.Boolean(False),

1406

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

1407

ApprovalPending=dbus.Boolean(False),

1408

ApprovedByDefault=dbus.Boolean(True),

1409

LastApprovalRequest="",

1410

ApprovalDelay=0,

1411

ApprovalDuration=1000,

1412

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

1413

ExtendedTimeout=900000,

1414

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

1415

LastCheckerStatus=0)

1416

self.other_client = MockClient(

1417

"barbar",

1418

KeyID=("0558568eedd67d622f5c83b35a115f79"

1419

"6ab612cff5ad227247e46c2b020f441c"),

1420

Secret=b"secretbar",

1421

Host="192.0.2.3",

1422

Enabled=dbus.Boolean(True),

1423

Timeout=300000,

1424

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

1425

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

1426

Interval=120000,

1427

Fingerprint=("3E393AEAEFB84C7E89E2"

1428

"F547B3A107558FCA3A27"),

1429

CheckerRunning=dbus.Boolean(True),

1430

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

1431

ApprovalPending=dbus.Boolean(False),

1432

ApprovedByDefault=dbus.Boolean(False),

1433

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

1434

ApprovalDelay=30000,

1435

ApprovalDuration=93785000,

1436

Checker=":",

1437

ExtendedTimeout=900000,

1438

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

1439

LastCheckerStatus=-2)

1440

self.clients = collections.OrderedDict(

1441

[

1442

(self.client.__dbus_object_path__,

1443

self.client.attributes),

1444

(self.other_client.__dbus_object_path__,

1445

self.other_client.attributes),

1446

])

1447

self.one_client = {self.client.__dbus_object_path__:

1448

self.client.attributes}

1449

1450

@property

1451

def bus(self):

1452

class MockBus(object):

1453

@staticmethod

1454

def get_object(client_bus_name, path):

1455

self.assertEqual(dbus_busname, client_bus_name)

1456

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

1457

# MockBus instance, since this is a static method!

1458

if path == self.client.__dbus_object_path__:

1459

return self.client

1460

elif path == self.other_client.__dbus_object_path__:

1461

return self.other_client

1462

return MockBus()

1463

1464

1465

class TestBaseCommands(TestCommand):

1466

1467

def test_IsEnabled_exits_successfully(self):

1468

with self.assertRaises(SystemExit) as e:

1469

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

1470

if e.exception.code is not None:

1471

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

1472

else:

1473

self.assertIsNone(e.exception.code)

1474

1475

def test_IsEnabled_exits_with_failure(self):

1476

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

1477

with self.assertRaises(SystemExit) as e:

1478

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

1479

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

1480

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

1481

else:

1482

self.assertIsNotNone(e.exception.code)

1483

1484

def test_Approve(self):

1485

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

1486

for clientpath in self.clients:

1487

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

1488

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

1489

client.calls)

1490

1491

def test_Deny(self):

1492

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

1493

for clientpath in self.clients:

1494

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

1495

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

1496

client.calls)

1497

1498

def test_Remove(self):

1499

class MandosSpy(object):

1500

def __init__(self):

1501

self.calls = []

1502

def RemoveClient(self, dbus_path):

1503

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

1504

mandos = MandosSpy()

1505

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

1506

for clientpath in self.clients:

1507

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

1508

mandos.calls)

1509

1510

expected_json = {

1511

"foo": {

1512

"Name": "foo",

1513

"KeyID": ("92ed150794387c03ce684574b1139a65"

1514

"94a34f895daaaf09fd8ea90a27cddb12"),

1515

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

1516

"Enabled": True,

1517

"Timeout": 300000,

1518

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

1519

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

1520

"Interval": 120000,

1521

"Fingerprint": ("778827225BA7DE539C5A"

1522

"7CFA59CFF7CDBD9A5920"),

1523

"CheckerRunning": False,

1524

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

1525

"ApprovalPending": False,

1526

"ApprovedByDefault": True,

1527

"LastApprovalRequest": "",

1528

"ApprovalDelay": 0,

1529

"ApprovalDuration": 1000,

1530

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

1531

"ExtendedTimeout": 900000,

1532

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

1533

"LastCheckerStatus": 0,

1534

},

1535

"barbar": {

1536

"Name": "barbar",

1537

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1538

"6ab612cff5ad227247e46c2b020f441c"),

1539

"Host": "192.0.2.3",

1540

"Enabled": True,

1541

"Timeout": 300000,

1542

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

1543

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

1544

"Interval": 120000,

1545

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1546

"F547B3A107558FCA3A27"),

1547

"CheckerRunning": True,

1548

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

1549

"ApprovalPending": False,

1550

"ApprovedByDefault": False,

1551

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

1552

"ApprovalDelay": 30000,

1553

"ApprovalDuration": 93785000,

1554

"Checker": ":",

1555

"ExtendedTimeout": 900000,

1556

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

1557

"LastCheckerStatus": -2,

1558

},

1559

}

1560

1561

def test_DumpJSON_normal(self):

1562

with self.capture_stdout_to_buffer() as buffer:

1563

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

1564

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

1565

self.assertDictEqual(self.expected_json, json_data)

1566

1567

@staticmethod

1568

@contextlib.contextmanager

1569

def capture_stdout_to_buffer():

1570

capture_buffer = io.StringIO()

1571

old_stdout = sys.stdout

1572

sys.stdout = capture_buffer

1573

try:

1574

yield capture_buffer

1575

finally:

1576

sys.stdout = old_stdout

1577

1578

def test_DumpJSON_one_client(self):

1579

with self.capture_stdout_to_buffer() as buffer:

1580

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

1581

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

1582

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

1583

self.assertDictEqual(expected_json, json_data)

1584

1585

def test_PrintTable_normal(self):

1586

with self.capture_stdout_to_buffer() as buffer:

1587

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

1588

expected_output = "\n".join((

1589

"Name Enabled Timeout Last Successful Check",

1590

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

1591

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

1592

)) + "\n"

1593

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

1594

1595

def test_PrintTable_verbose(self):

1596

with self.capture_stdout_to_buffer() as buffer:

1597

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

1598

columns = (

1599

(

1600

"Name ",

1601

"foo ",

1602

"barbar ",

1603

),(

1604

"Enabled ",

1605

"Yes ",

1606

"Yes ",

1607

),(

1608

"Timeout ",

1609

"00:05:00 ",

1610

"00:05:00 ",

1611

),(

1612

"Last Successful Check ",

1613

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

1614

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

1615

),(

1616

"Created ",

1617

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

1618

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

1619

),(

1620

"Interval ",

1621

"00:02:00 ",

1622

"00:02:00 ",

1623

),(

1624

"Host ",

1625

"foo.example.org ",

1626

"192.0.2.3 ",

1627

),(

1628

("Key ID "

1629

" "),

1630

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1631

"ea90a27cddb12 "),

1632

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1633

"46c2b020f441c "),

1634

),(

1635

"Fingerprint ",

1636

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1637

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1638

),(

1639

"Check Is Running ",

1640

"No ",

1641

"Yes ",

1642

),(

1643

"Last Enabled ",

1644

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

1645

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

1646

),(

1647

"Approval Is Pending ",

1648

"No ",

1649

"No ",

1650

),(

1651

"Approved By Default ",

1652

"Yes ",

1653

"No ",

1654

),(

1655

"Last Approval Request ",

1656

" ",

1657

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

1658

),(

1659

"Approval Delay ",

1660

"00:00:00 ",

1661

"00:00:30 ",

1662

),(

1663

"Approval Duration ",

1664

"00:00:01 ",

1665

"1T02:03:05 ",

1666

),(

1667

"Checker ",

1668

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

1669

": ",

1670

),(

1671

"Extended Timeout ",

1672

"00:15:00 ",

1673

"00:15:00 ",

1674

),(

1675

"Expires ",

1676

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

1677

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

1678

),(

1679

"Last Checker Status",

1680

"0 ",

1681

"-2 ",

1682

)

1683

)

1684

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

1685

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

1686

for rows in columns)

1687

for line in range(num_lines))

1688

+ "\n")

1689

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

1690

1691

def test_PrintTable_one_client(self):

1692

with self.capture_stdout_to_buffer() as buffer:

1693

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

1694

expected_output = "\n".join((

1695

"Name Enabled Timeout Last Successful Check",

1696

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

1697

)) + "\n"

1698

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

1699

1700

1701

class TestPropertySetterCmd(TestCommand):

1702

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

1703

def runTest(self):

1704

if not hasattr(self, "command"):

1705

return

1706

values_to_get = getattr(self, "values_to_get",

1707

self.values_to_set)

1708

for value_to_set, value_to_get in zip(self.values_to_set,

1709

values_to_get):

1710

for clientpath in self.clients:

1711

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

1712

old_value = client.attributes[self.propname]

1713

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

1714

self.run_command(value_to_set, self.clients)

1715

for clientpath in self.clients:

1716

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

1717

value = client.attributes[self.propname]

1718

self.assertNotIsInstance(value, self.Unique)

1719

self.assertEqual(value_to_get, value)

1720

1721

class Unique(object):

1722

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

1723

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

1724

1725

def run_command(self, value, clients):

1726

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

1727

1728

1729

class TestEnableCmd(TestPropertySetterCmd):

1730

command = command.Enable

1731

propname = "Enabled"

1732

values_to_set = [dbus.Boolean(True)]

1733

1734

1735

class TestDisableCmd(TestPropertySetterCmd):

1736

command = command.Disable

1737

propname = "Enabled"

1738

values_to_set = [dbus.Boolean(False)]

1739

1740

1741

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1742

command = command.BumpTimeout

1743

propname = "LastCheckedOK"

1744

values_to_set = [""]

1745

1746

1747

class TestStartCheckerCmd(TestPropertySetterCmd):

1748

command = command.StartChecker

1749

propname = "CheckerRunning"

1750

values_to_set = [dbus.Boolean(True)]

1751

1752

1753

class TestStopCheckerCmd(TestPropertySetterCmd):

1754

command = command.StopChecker

1755

propname = "CheckerRunning"

1756

values_to_set = [dbus.Boolean(False)]

1757

1758

1759

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1760

command = command.ApproveByDefault

1761

propname = "ApprovedByDefault"

1762

values_to_set = [dbus.Boolean(True)]

1763

1764

1765

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1766

command = command.DenyByDefault

1767

propname = "ApprovedByDefault"

1768

values_to_set = [dbus.Boolean(False)]

1769

1770

1771

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1772

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

1773

1774

def runTest(self):

1775

if type(self) is TestPropertySetterValueCmd:

1776

return

1777

return super(TestPropertySetterValueCmd, self).runTest()

1778

1779

def run_command(self, value, clients):

1780

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

1781

1782

1783

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1784

command = command.SetChecker

1785

propname = "Checker"

1786

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

1787

1788

1789

class TestSetHostCmd(TestPropertySetterValueCmd):

1790

command = command.SetHost

1791

propname = "Host"

1792

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

1793

1794

1795

class TestSetSecretCmd(TestPropertySetterValueCmd):

1796

command = command.SetSecret

1797

propname = "Secret"

1798

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

1799

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

1800

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

1801

1802

1803

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1804

command = command.SetTimeout

1805

propname = "Timeout"

1806

values_to_set = [datetime.timedelta(),

1807

datetime.timedelta(minutes=5),

1808

datetime.timedelta(seconds=1),

1809

datetime.timedelta(weeks=1),

1810

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1815

command = command.SetExtendedTimeout

1816

propname = "ExtendedTimeout"

1817

values_to_set = [datetime.timedelta(),

1818

datetime.timedelta(minutes=5),

1819

datetime.timedelta(seconds=1),

1820

datetime.timedelta(weeks=1),

1821

datetime.timedelta(weeks=52)]

1822

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

1823

1824

1825

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1826

command = command.SetInterval

1827

propname = "Interval"

1828

values_to_set = [datetime.timedelta(),

1829

datetime.timedelta(minutes=5),

1830

datetime.timedelta(seconds=1),

1831

datetime.timedelta(weeks=1),

1832

datetime.timedelta(weeks=52)]

1833

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

1834

1835

1836

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1837

command = command.SetApprovalDelay

1838

propname = "ApprovalDelay"

1839

values_to_set = [datetime.timedelta(),

1840

datetime.timedelta(minutes=5),

1841

datetime.timedelta(seconds=1),

1842

datetime.timedelta(weeks=1),

1843

datetime.timedelta(weeks=52)]

1844

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

1845

1846

1847

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1848

command = command.SetApprovalDuration

1849

propname = "ApprovalDuration"

1850

values_to_set = [datetime.timedelta(),

1851

datetime.timedelta(minutes=5),

1852

datetime.timedelta(seconds=1),

1853

datetime.timedelta(weeks=1),

1854

datetime.timedelta(weeks=52)]

1855

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

1856

1857

1858

1859

def should_only_run_tests():

1860

parser = argparse.ArgumentParser(add_help=False)

1861

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

1862

args, unknown_args = parser.parse_known_args()

1863

run_tests = args.check

1864

if run_tests:

1865

# Remove --check argument from sys.argv

1866

sys.argv[1:] = unknown_args

1867

return run_tests

1868

1869

# Add all tests from doctest strings

1870

def load_tests(loader, tests, none):

1871

import doctest

1872

tests.addTests(doctest.DocTestSuite())

1873

return tests

343

1874

344

1875

if __name__ == "__main__":

345

main()

1876

try:

1877

if should_only_run_tests():

1878

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

1879

unittest.main()

1880

else:

1881

main()

1882

finally:

1883

logging.shutdown()