/mandos/release : revision 237.7.630

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

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

23

#

22

24

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

23

#

25

#

24

26

25

27

from __future__ import (division, absolute_import, print_function,

26

28

unicode_literals)

27

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

28

35

import sys

29

import dbus

30

36

import argparse

31

37

import locale

32

38

import datetime

33

39

import re

34

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

35

64

36

65

locale.setlocale(locale.LC_ALL, "")

37

66

38

tablewords = {

39

"Name": "Name",

40

"Enabled": "Enabled",

41

"Timeout": "Timeout",

42

"LastCheckedOK": "Last Successful Check",

43

"LastApprovalRequest": "Last Approval Request",

44

"Created": "Created",

45

"Interval": "Interval",

46

"Host": "Host",

47

"Fingerprint": "Fingerprint",

48

"CheckerRunning": "Check Is Running",

49

"LastEnabled": "Last Enabled",

50

"ApprovalPending": "Approval Is Pending",

51

"ApprovedByDefault": "Approved By Default",

52

"ApprovalDelay": "Approval Delay",

53

"ApprovalDuration": "Approval Duration",

54

"Checker": "Checker",

55

"ExtendedTimeout" : "Extended Timeout"

56

}

57

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

58

domain = "se.recompile"

59

busname = domain + ".Mandos"

60

server_path = "/"

61

server_interface = domain + ".Mandos"

62

client_interface = domain + ".Mandos.Client"

63

version = "1.5.3"

64

65

def timedelta_to_milliseconds(td):

66

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

67

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

68

+ (td.seconds * 1000)

69

+ (td.microseconds // 1000))

70

71

def milliseconds_to_string(ms):

72

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

73

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

74

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

75

hours = td.seconds // 3600,

76

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

77

seconds = td.seconds % 60,

78

))

79

80

def string_to_delta(interval):

81

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

82

83

>>> string_to_delta("7d")

84

datetime.timedelta(7)

85

>>> string_to_delta("60s")

86

datetime.timedelta(0, 60)

87

>>> string_to_delta("60m")

88

datetime.timedelta(0, 3600)

89

>>> string_to_delta("24h")

90

datetime.timedelta(1)

91

>>> string_to_delta("1w")

92

datetime.timedelta(7)

93

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

94

datetime.timedelta(0, 330)

95

"""

96

value = datetime.timedelta(0)

97

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

98

99

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

100

if suffix == "d":

101

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

102

elif suffix == "s":

103

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

104

elif suffix == "m":

105

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

106

elif suffix == "h":

107

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

108

elif suffix == "w":

109

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

110

elif suffix == "":

111

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

112

return value

113

114

def print_clients(clients, keywords):

115

def valuetostring(value, keyword):

116

if type(value) is dbus.Boolean:

117

return "Yes" if value else "No"

118

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

119

"ApprovalDuration"):

120

return milliseconds_to_string(value)

121

return unicode(value)

122

123

# Create format string to print table rows

124

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

125

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

126

max(len(valuetostring(client[key],

127

key))

128

for client in

129

clients)),

130

key = key) for key in keywords)

131

# Print header line

132

print(format_string.format(**tablewords))

133

for client in clients:

134

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

135

valuetostring(client[key],

136

key))

137

for key in keywords)))

138

139

def has_actions(options):

140

return any((options.enable,

141

options.disable,

142

options.bump_timeout,

143

options.start_checker,

144

options.stop_checker,

145

options.is_enabled,

146

options.remove,

147

options.checker is not None,

148

options.timeout is not None,

149

options.extended_timeout is not None,

150

options.interval is not None,

151

options.approved_by_default is not None,

152

options.approval_delay is not None,

153

options.approval_duration is not None,

154

options.host is not None,

155

options.secret is not None,

156

options.approve,

157

options.deny))

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

73

version = "1.8.3"

74

75

76

try:

77

dbus.OBJECT_MANAGER_IFACE

78

except AttributeError:

79

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

80

158

81

159

82

def main():

160

83

parser = argparse.ArgumentParser()

84

85

add_command_line_options(parser)

86

87

options = parser.parse_args()

88

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

commands = commands_from_options(options)

129

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

161

135

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

162

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

136

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

163

137

help="show version number and exit")

164

138

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

165

139

help="Select all clients")

166

140

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

167

141

help="Print all fields")

168

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

169

help="Enable client")

170

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

171

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

172

150

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

173

151

help="Bump timeout for client")

174

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

175

help="Start checker for client")

176

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

177

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

178

159

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

179

160

help="Check if client is enabled")

180

161

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

181

162

help="Remove client")

182

163

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

183

164

help="Set checker command for client")

184

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

165

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

185

166

help="Set timeout for client")

186

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

167

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

187

168

help="Set extended timeout for client")

188

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

169

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

189

170

help="Set checker interval for client")

190

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

191

default=None, dest="approved_by_default",

192

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

193

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

194

dest="approved_by_default",

195

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

196

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,

197

181

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

198

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

182

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

199

183

help="Set duration of one client approval")

200

184

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

201

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

185

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

186

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

202

187

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

203

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

204

help="Approve any current client request")

205

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

206

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

207

198

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

208

options = parser.parse_args()

209

210

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

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

409

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

211

410

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

212

411

if options.verbose and has_actions(options):

213

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

214

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

215

416

if options.all and not has_actions(options):

216

417

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

217

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

218

439

try:

219

bus = dbus.SystemBus()

220

mandos_dbus_objc = bus.get_object(busname, server_path)

221

except dbus.exceptions.DBusException:

222

print("Could not connect to Mandos server",

223

file=sys.stderr)

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

224

443

sys.exit(1)

225

226

mandos_serv = dbus.Interface(mandos_dbus_objc,

227

dbus_interface = server_interface)

228

229

#block stderr since dbus library prints to stderr

230

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

231

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

232

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

233

os.close(null)

234

try:

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

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

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

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

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(command.IsEnabled())

481

482

if options.approve:

483

commands.append(command.Approve())

484

485

if options.deny:

486

commands.append(command.Deny())

487

488

if options.remove:

489

commands.append(command.Remove())

490

491

if options.dump_json:

492

commands.append(command.DumpJSON())

493

494

if options.enable:

495

commands.append(command.Enable())

496

497

if options.disable:

498

commands.append(command.Disable())

499

500

if options.bump_timeout:

501

commands.append(command.BumpTimeout())

502

503

if options.start_checker:

504

commands.append(command.StartChecker())

505

506

if options.stop_checker:

507

commands.append(command.StopChecker())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(command.ApproveByDefault())

512

else:

513

commands.append(command.DenyByDefault())

514

515

if options.checker is not None:

516

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

517

518

if options.host is not None:

519

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

520

521

if options.secret is not None:

522

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

523

524

if options.timeout is not None:

525

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

526

527

if options.extended_timeout:

528

commands.append(

529

command.SetExtendedTimeout(options.extended_timeout))

530

531

if options.interval is not None:

532

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

533

534

if options.approval_delay is not None:

535

commands.append(

536

command.SetApprovalDelay(options.approval_delay))

537

538

if options.approval_duration is not None:

539

commands.append(

540

command.SetApprovalDuration(options.approval_duration))

541

542

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

543

# optionally verbosely

544

if not commands:

545

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

546

547

return commands

548

549

550

class command(object):

551

"""A namespace for command classes"""

552

553

class Base(object):

554

"""Abstract base class for commands"""

555

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

556

"""Normal commands should implement run_on_one_client(),

557

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

558

override this run() method instead.

559

"""

560

self.mandos = mandos

561

for clientpath, properties in clients.items():

562

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

563

dbus_busname, str(clientpath))

564

client = bus.get_object(dbus_busname, clientpath)

565

self.run_on_one_client(client, properties)

566

567

568

class IsEnabled(Base):

569

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

570

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

571

if self.is_enabled(client, properties):

572

sys.exit(0)

573

sys.exit(1)

574

def is_enabled(self, client, properties):

575

return properties["Enabled"]

576

577

578

class Approve(Base):

579

def run_on_one_client(self, client, properties):

580

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

581

client.__dbus_object_path__,

582

client_dbus_interface)

583

client.Approve(dbus.Boolean(True),

584

dbus_interface=client_dbus_interface)

585

586

587

class Deny(Base):

588

def run_on_one_client(self, client, properties):

589

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

590

client.__dbus_object_path__,

591

client_dbus_interface)

592

client.Approve(dbus.Boolean(False),

593

dbus_interface=client_dbus_interface)

594

595

596

class Remove(Base):

597

def run_on_one_client(self, client, properties):

598

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

599

dbus_busname, server_dbus_path,

600

server_dbus_interface,

601

str(client.__dbus_object_path__))

602

self.mandos.RemoveClient(client.__dbus_object_path__)

603

604

605

class Output(Base):

606

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

607

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

608

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

609

"Fingerprint", "CheckerRunning",

610

"LastEnabled", "ApprovalPending",

611

"ApprovedByDefault", "LastApprovalRequest",

612

"ApprovalDelay", "ApprovalDuration",

613

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

614

"LastCheckerStatus")

615

616

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

617

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

618

619

def output(self, clients):

620

raise NotImplementedError()

621

622

623

class DumpJSON(Output):

624

def output(self, clients):

625

data = {client["Name"]:

626

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

627

for key in self.all_keywords}

628

for client in clients}

629

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

630

631

@staticmethod

632

def dbus_boolean_to_bool(value):

633

if isinstance(value, dbus.Boolean):

634

value = bool(value)

635

return value

636

637

638

class PrintTable(Output):

639

def __init__(self, verbose=False):

640

self.verbose = verbose

641

642

def output(self, clients):

643

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

644

"LastCheckedOK")

645

keywords = default_keywords

646

if self.verbose:

647

keywords = self.all_keywords

648

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

649

650

class TableOfClients(object):

651

tableheaders = {

652

"Name": "Name",

653

"Enabled": "Enabled",

654

"Timeout": "Timeout",

655

"LastCheckedOK": "Last Successful Check",

656

"LastApprovalRequest": "Last Approval Request",

657

"Created": "Created",

658

"Interval": "Interval",

659

"Host": "Host",

660

"Fingerprint": "Fingerprint",

661

"KeyID": "Key ID",

662

"CheckerRunning": "Check Is Running",

663

"LastEnabled": "Last Enabled",

664

"ApprovalPending": "Approval Is Pending",

665

"ApprovedByDefault": "Approved By Default",

666

"ApprovalDelay": "Approval Delay",

667

"ApprovalDuration": "Approval Duration",

668

"Checker": "Checker",

669

"ExtendedTimeout": "Extended Timeout",

670

"Expires": "Expires",

671

"LastCheckerStatus": "Last Checker Status",

672

}

673

674

def __init__(self, clients, keywords):

675

self.clients = clients

676

self.keywords = keywords

677

678

def __str__(self):

679

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

680

681

if sys.version_info.major == 2:

682

__unicode__ = __str__

683

def __str__(self):

684

return str(self).encode(

685

locale.getpreferredencoding())

686

687

def rows(self):

688

format_string = self.row_formatting_string()

689

rows = [self.header_line(format_string)]

690

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

691

for client in self.clients)

692

return rows

693

694

def row_formatting_string(self):

695

"Format string used to format table rows"

696

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

697

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

698

*(len(self.string_from_client(client,

699

key))

700

for client in self.clients)),

701

key=key)

702

for key in self.keywords)

703

704

def string_from_client(self, client, key):

705

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

706

707

@classmethod

708

def valuetostring(cls, value, keyword):

709

if isinstance(value, dbus.Boolean):

710

return "Yes" if value else "No"

711

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

712

"ApprovalDuration", "ExtendedTimeout"):

713

return cls.milliseconds_to_string(value)

714

return str(value)

715

716

def header_line(self, format_string):

717

return format_string.format(**self.tableheaders)

718

719

def client_line(self, client, format_string):

720

return format_string.format(

721

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

722

for key in self.keywords})

723

724

@staticmethod

725

def milliseconds_to_string(ms):

726

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

727

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

728

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

729

if td.days else "",

730

hours=td.seconds // 3600,

731

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

732

seconds=td.seconds % 60))

733

734

735

class Property(Base):

736

"Abstract class for Actions for setting one client property"

737

738

def run_on_one_client(self, client, properties):

739

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

740

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

741

client.__dbus_object_path__,

742

dbus.PROPERTIES_IFACE, client_dbus_interface,

743

self.propname, self.value_to_set

744

if not isinstance(self.value_to_set,

745

dbus.Boolean)

746

else bool(self.value_to_set))

747

client.Set(client_dbus_interface, self.propname,

748

self.value_to_set,

749

dbus_interface=dbus.PROPERTIES_IFACE)

750

751

@property

752

def propname(self):

753

raise NotImplementedError()

754

755

756

class Enable(Property):

757

propname = "Enabled"

758

value_to_set = dbus.Boolean(True)

759

760

761

class Disable(Property):

762

propname = "Enabled"

763

value_to_set = dbus.Boolean(False)

764

765

766

class BumpTimeout(Property):

767

propname = "LastCheckedOK"

768

value_to_set = ""

769

770

771

class StartChecker(Property):

772

propname = "CheckerRunning"

773

value_to_set = dbus.Boolean(True)

774

775

776

class StopChecker(Property):

777

propname = "CheckerRunning"

778

value_to_set = dbus.Boolean(False)

779

780

781

class ApproveByDefault(Property):

782

propname = "ApprovedByDefault"

783

value_to_set = dbus.Boolean(True)

784

785

786

class DenyByDefault(Property):

787

propname = "ApprovedByDefault"

788

value_to_set = dbus.Boolean(False)

789

790

791

class PropertyValue(Property):

792

"Abstract class for Property recieving a value as argument"

793

def __init__(self, value):

794

self.value_to_set = value

795

796

797

class SetChecker(PropertyValue):

798

propname = "Checker"

799

800

801

class SetHost(PropertyValue):

802

propname = "Host"

803

804

805

class SetSecret(PropertyValue):

806

propname = "Secret"

807

808

@property

809

def value_to_set(self):

810

return self._vts

811

812

@value_to_set.setter

813

def value_to_set(self, value):

814

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

815

self._vts = value.read()

816

value.close()

817

818

819

class MillisecondsPropertyValueArgument(PropertyValue):

820

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

821

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

822

823

@property

824

def value_to_set(self):

825

return self._vts

826

827

@value_to_set.setter

828

def value_to_set(self, value):

829

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

830

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

831

832

833

class SetTimeout(MillisecondsPropertyValueArgument):

834

propname = "Timeout"

835

836

837

class SetExtendedTimeout(MillisecondsPropertyValueArgument):

838

propname = "ExtendedTimeout"

839

840

841

class SetInterval(MillisecondsPropertyValueArgument):

842

propname = "Interval"

843

844

845

class SetApprovalDelay(MillisecondsPropertyValueArgument):

846

propname = "ApprovalDelay"

847

848

849

class SetApprovalDuration(MillisecondsPropertyValueArgument):

850

propname = "ApprovalDuration"

851

852

853

854

class TestCaseWithAssertLogs(unittest.TestCase):

855

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

856

857

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

858

@contextlib.contextmanager

859

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

860

capturing_handler = self.CapturingLevelHandler(level)

861

old_level = logger.level

862

old_propagate = logger.propagate

863

logger.addHandler(capturing_handler)

864

logger.setLevel(level)

865

logger.propagate = False

866

try:

867

yield capturing_handler.watcher

868

finally:

869

logger.propagate = old_propagate

870

logger.removeHandler(capturing_handler)

871

logger.setLevel(old_level)

872

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

873

0)

874

875

class CapturingLevelHandler(logging.Handler):

876

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

877

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

878

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

879

def emit(self, record):

880

self.watcher.records.append(record)

881

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

882

883

LoggingWatcher = collections.namedtuple("LoggingWatcher",

884

("records",

885

"output"))

886

887

888

class Test_string_to_delta(TestCaseWithAssertLogs):

889

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

890

# rfc3339_duration_to_delta() already has more comprehensive

891

# tests, which is run by doctest.

892

893

def test_rfc3339_zero_seconds(self):

894

self.assertEqual(datetime.timedelta(),

895

string_to_delta("PT0S"))

896

897

def test_rfc3339_zero_days(self):

898

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

899

900

def test_rfc3339_one_second(self):

901

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

902

string_to_delta("PT1S"))

903

904

def test_rfc3339_two_hours(self):

905

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

906

string_to_delta("PT2H"))

907

908

def test_falls_back_to_pre_1_6_1_with_warning(self):

909

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

910

value = string_to_delta("2h")

911

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

912

913

914

class Test_check_option_syntax(unittest.TestCase):

915

def setUp(self):

916

self.parser = argparse.ArgumentParser()

917

add_command_line_options(self.parser)

918

919

def test_actions_requires_client_or_all(self):

920

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

921

options = self.parser.parse_args()

922

setattr(options, action, value)

923

with self.assertParseError():

924

self.check_option_syntax(options)

925

926

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

927

# check_option_syntax()

928

actions = {

929

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

930

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

931

# never used

932

"enable": True,

933

"disable": True,

934

"bump_timeout": True,

935

"start_checker": True,

936

"stop_checker": True,

937

"is_enabled": True,

938

"remove": True,

939

"checker": "x",

940

"timeout": datetime.timedelta(),

941

"extended_timeout": datetime.timedelta(),

942

"interval": datetime.timedelta(),

943

"approved_by_default": True,

944

"approval_delay": datetime.timedelta(),

945

"approval_duration": datetime.timedelta(),

946

"host": "x",

947

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

948

"approve": True,

949

"deny": True,

950

}

951

952

@contextlib.contextmanager

953

def assertParseError(self):

954

with self.assertRaises(SystemExit) as e:

955

with self.redirect_stderr_to_devnull():

956

yield

957

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

958

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

959

# /argparse.html#exiting-methods

960

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

961

962

@staticmethod

963

@contextlib.contextmanager

964

def redirect_stderr_to_devnull():

965

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

966

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

967

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

968

os.close(null)

235

969

try:

236

mandos_clients = mandos_serv.GetAllClientsWithProperties()

970

yield

237

971

finally:

238

#restore stderr

972

# restore stderr

239

973

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

240

974

os.close(stderrcopy)

241

except dbus.exceptions.DBusException:

242

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

243

file=sys.stderr)

244

sys.exit(1)

245

246

# Compile dict of (clients: properties) to process

247

clients={}

248

249

if options.all or not options.client:

250

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

251

for path, properties in

252

mandos_clients.iteritems())

253

else:

254

for name in options.client:

255

for path, client in mandos_clients.iteritems():

256

if client["Name"] == name:

257

client_objc = bus.get_object(busname, path)

258

clients[client_objc] = client

259

break

260

else:

261

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

262

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

263

sys.exit(1)

264

265

if not has_actions(options) and clients:

266

if options.verbose:

267

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

268

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

269

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

270

"LastEnabled", "ApprovalPending",

271

"ApprovedByDefault",

272

"LastApprovalRequest", "ApprovalDelay",

273

"ApprovalDuration", "Checker",

274

"ExtendedTimeout")

275

else:

276

keywords = defaultkeywords

277

278

print_clients(clients.values(), keywords)

279

else:

280

# Process each client in the list by all selected options

281

for client in clients:

282

if options.remove:

283

mandos_serv.RemoveClient(client.__dbus_object_path__)

284

if options.enable:

285

client.Set(client_interface, "Enabled",

286

dbus.Boolean(True),

287

dbus_interface=dbus.PROPERTIES_IFACE)

288

if options.disable:

289

client.Set(client_interface, "Enabled",

290

dbus.Boolean(False),

291

dbus_interface=dbus.PROPERTIES_IFACE)

292

if options.bump_timeout:

293

client.Set(client_interface, "LastCheckedOK", "",

294

dbus_interface=dbus.PROPERTIES_IFACE)

295

if options.start_checker:

296

client.Set(client_interface, "CheckerRunning",

297

dbus.Boolean(True),

298

dbus_interface=dbus.PROPERTIES_IFACE)

299

if options.stop_checker:

300

client.Set(client_interface, "CheckerRunning",

301

dbus.Boolean(False),

302

dbus_interface=dbus.PROPERTIES_IFACE)

303

if options.is_enabled:

304

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

305

"Enabled",

306

dbus_interface=

307

dbus.PROPERTIES_IFACE)

308

else 1)

309

if options.checker is not None:

310

client.Set(client_interface, "Checker",

311

options.checker,

312

dbus_interface=dbus.PROPERTIES_IFACE)

313

if options.host is not None:

314

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

315

dbus_interface=dbus.PROPERTIES_IFACE)

316

if options.interval is not None:

317

client.Set(client_interface, "Interval",

318

timedelta_to_milliseconds

319

(string_to_delta(options.interval)),

320

dbus_interface=dbus.PROPERTIES_IFACE)

321

if options.approval_delay is not None:

322

client.Set(client_interface, "ApprovalDelay",

323

timedelta_to_milliseconds

324

(string_to_delta(options.

325

approval_delay)),

326

dbus_interface=dbus.PROPERTIES_IFACE)

327

if options.approval_duration is not None:

328

client.Set(client_interface, "ApprovalDuration",

329

timedelta_to_milliseconds

330

(string_to_delta(options.

331

approval_duration)),

332

dbus_interface=dbus.PROPERTIES_IFACE)

333

if options.timeout is not None:

334

client.Set(client_interface, "Timeout",

335

timedelta_to_milliseconds

336

(string_to_delta(options.timeout)),

337

dbus_interface=dbus.PROPERTIES_IFACE)

338

if options.extended_timeout is not None:

339

client.Set(client_interface, "ExtendedTimeout",

340

timedelta_to_milliseconds

341

(string_to_delta(options.extended_timeout)),

342

dbus_interface=dbus.PROPERTIES_IFACE)

343

if options.secret is not None:

344

client.Set(client_interface, "Secret",

345

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

346

dbus_interface=dbus.PROPERTIES_IFACE)

347

if options.approved_by_default is not None:

348

client.Set(client_interface, "ApprovedByDefault",

349

dbus.Boolean(options

350

.approved_by_default),

351

dbus_interface=dbus.PROPERTIES_IFACE)

352

if options.approve:

353

client.Approve(dbus.Boolean(True),

354

dbus_interface=client_interface)

355

elif options.deny:

356

client.Approve(dbus.Boolean(False),

357

dbus_interface=client_interface)

975

976

def check_option_syntax(self, options):

977

check_option_syntax(self.parser, options)

978

979

def test_actions_all_conflicts_with_verbose(self):

980

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

981

options = self.parser.parse_args()

982

setattr(options, action, value)

983

options.all = True

984

options.verbose = True

985

with self.assertParseError():

986

self.check_option_syntax(options)

987

988

def test_actions_with_client_conflicts_with_verbose(self):

989

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

990

options = self.parser.parse_args()

991

setattr(options, action, value)

992

options.verbose = True

993

options.client = ["foo"]

994

with self.assertParseError():

995

self.check_option_syntax(options)

996

997

def test_dump_json_conflicts_with_verbose(self):

998

options = self.parser.parse_args()

999

options.dump_json = True

1000

options.verbose = True

1001

with self.assertParseError():

1002

self.check_option_syntax(options)

1003

1004

def test_dump_json_conflicts_with_action(self):

1005

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

1006

options = self.parser.parse_args()

1007

setattr(options, action, value)

1008

options.dump_json = True

1009

with self.assertParseError():

1010

self.check_option_syntax(options)

1011

1012

def test_all_can_not_be_alone(self):

1013

options = self.parser.parse_args()

1014

options.all = True

1015

with self.assertParseError():

1016

self.check_option_syntax(options)

1017

1018

def test_all_is_ok_with_any_action(self):

1019

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

1020

options = self.parser.parse_args()

1021

setattr(options, action, value)

1022

options.all = True

1023

self.check_option_syntax(options)

1024

1025

def test_any_action_is_ok_with_one_client(self):

1026

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

1027

options = self.parser.parse_args()

1028

setattr(options, action, value)

1029

options.client = ["foo"]

1030

self.check_option_syntax(options)

1031

1032

def test_one_client_with_all_actions_except_is_enabled(self):

1033

options = self.parser.parse_args()

1034

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

1035

if action == "is_enabled":

1036

continue

1037

setattr(options, action, value)

1038

options.client = ["foo"]

1039

self.check_option_syntax(options)

1040

1041

def test_two_clients_with_all_actions_except_is_enabled(self):

1042

options = self.parser.parse_args()

1043

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

1044

if action == "is_enabled":

1045

continue

1046

setattr(options, action, value)

1047

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

1048

self.check_option_syntax(options)

1049

1050

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1051

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

1052

if action == "is_enabled":

1053

continue

1054

options = self.parser.parse_args()

1055

setattr(options, action, value)

1056

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

1057

self.check_option_syntax(options)

1058

1059

def test_is_enabled_fails_without_client(self):

1060

options = self.parser.parse_args()

1061

options.is_enabled = True

1062

with self.assertParseError():

1063

self.check_option_syntax(options)

1064

1065

def test_is_enabled_fails_with_two_clients(self):

1066

options = self.parser.parse_args()

1067

options.is_enabled = True

1068

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

1069

with self.assertParseError():

1070

self.check_option_syntax(options)

1071

1072

def test_remove_can_only_be_combined_with_action_deny(self):

1073

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

1074

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

1075

continue

1076

options = self.parser.parse_args()

1077

setattr(options, action, value)

1078

options.all = True

1079

options.remove = True

1080

with self.assertParseError():

1081

self.check_option_syntax(options)

1082

1083

1084

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1085

def test_calls_and_returns_get_object_on_bus(self):

1086

class MockBus(object):

1087

called = False

1088

def get_object(mockbus_self, busname, dbus_path):

1089

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

1090

# this MockBus instance is in "mockbus_self".

1091

self.assertEqual(dbus_busname, busname)

1092

self.assertEqual(server_dbus_path, dbus_path)

1093

mockbus_self.called = True

1094

return mockbus_self

1095

1096

mockbus = get_mandos_dbus_object(bus=MockBus())

1097

self.assertIsInstance(mockbus, MockBus)

1098

self.assertTrue(mockbus.called)

1099

1100

def test_logs_and_exits_on_dbus_error(self):

1101

class MockBusFailing(object):

1102

def get_object(self, busname, dbus_path):

1103

raise dbus.exceptions.DBusException("Test")

1104

1105

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

1106

with self.assertRaises(SystemExit) as e:

1107

bus = get_mandos_dbus_object(bus=MockBusFailing())

1108

1109

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

1110

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

1111

else:

1112

self.assertIsNotNone(e.exception.code)

1113

1114

1115

class Test_get_managed_objects(TestCaseWithAssertLogs):

1116

def test_calls_and_returns_GetManagedObjects(self):

1117

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

1118

class MockObjectManager(object):

1119

def GetManagedObjects(self):

1120

return managed_objects

1121

retval = get_managed_objects(MockObjectManager())

1122

self.assertDictEqual(managed_objects, retval)

1123

1124

def test_logs_and_exits_on_dbus_error(self):

1125

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

1126

1127

class MockObjectManagerFailing(object):

1128

def GetManagedObjects(self):

1129

dbus_logger.error("Test")

1130

raise dbus.exceptions.DBusException("Test")

1131

1132

class CountingHandler(logging.Handler):

1133

count = 0

1134

def emit(self, record):

1135

self.count += 1

1136

1137

counting_handler = CountingHandler()

1138

1139

dbus_logger.addHandler(counting_handler)

1140

1141

try:

1142

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

1143

with self.assertRaises(SystemExit) as e:

1144

get_managed_objects(MockObjectManagerFailing())

1145

finally:

1146

dbus_logger.removeFilter(counting_handler)

1147

1148

# Make sure the dbus logger was suppressed

1149

self.assertEqual(0, counting_handler.count)

1150

1151

# Test that the dbus_logger still works

1152

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

1153

dbus_logger.error("Test")

1154

1155

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

1156

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

1157

else:

1158

self.assertIsNotNone(e.exception.code)

1159

1160

1161

class Test_commands_from_options(unittest.TestCase):

1162

def setUp(self):

1163

self.parser = argparse.ArgumentParser()

1164

add_command_line_options(self.parser)

1165

1166

def test_is_enabled(self):

1167

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

1168

command.IsEnabled)

1169

1170

def assert_command_from_args(self, args, command_cls,

1171

**cmd_attrs):

1172

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

1173

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

1174

options = self.parser.parse_args(args)

1175

check_option_syntax(self.parser, options)

1176

commands = commands_from_options(options)

1177

self.assertEqual(1, len(commands))

1178

command = commands[0]

1179

self.assertIsInstance(command, command_cls)

1180

for key, value in cmd_attrs.items():

1181

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

1182

1183

def test_is_enabled_short(self):

1184

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

1185

command.IsEnabled)

1186

1187

def test_approve(self):

1188

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

1189

command.Approve)

1190

1191

def test_approve_short(self):

1192

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

1193

1194

def test_deny(self):

1195

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

1196

1197

def test_deny_short(self):

1198

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

1199

1200

def test_remove(self):

1201

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

1202

command.Remove)

1203

1204

def test_deny_before_remove(self):

1205

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

1206

"foo"])

1207

check_option_syntax(self.parser, options)

1208

commands = commands_from_options(options)

1209

self.assertEqual(2, len(commands))

1210

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

1211

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

1212

1213

def test_deny_before_remove_reversed(self):

1214

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

1215

"--all"])

1216

check_option_syntax(self.parser, options)

1217

commands = commands_from_options(options)

1218

self.assertEqual(2, len(commands))

1219

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

1220

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

1221

1222

def test_remove_short(self):

1223

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

1224

1225

def test_dump_json(self):

1226

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

1227

command.DumpJSON)

1228

1229

def test_enable(self):

1230

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

1231

command.Enable)

1232

1233

def test_enable_short(self):

1234

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

1235

1236

def test_disable(self):

1237

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

1238

command.Disable)

1239

1240

def test_disable_short(self):

1241

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

1242

1243

def test_bump_timeout(self):

1244

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

1245

command.BumpTimeout)

1246

1247

def test_bump_timeout_short(self):

1248

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

1249

command.BumpTimeout)

1250

1251

def test_start_checker(self):

1252

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

1253

command.StartChecker)

1254

1255

def test_stop_checker(self):

1256

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

1257

command.StopChecker)

1258

1259

def test_approve_by_default(self):

1260

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

1261

command.ApproveByDefault)

1262

1263

def test_deny_by_default(self):

1264

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

1265

command.DenyByDefault)

1266

1267

def test_checker(self):

1268

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

1269

command.SetChecker,

1270

value_to_set=":")

1271

1272

def test_checker_empty(self):

1273

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

1274

command.SetChecker,

1275

value_to_set="")

1276

1277

def test_checker_short(self):

1278

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

1279

command.SetChecker,

1280

value_to_set=":")

1281

1282

def test_host(self):

1283

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

1284

"foo"], command.SetHost,

1285

value_to_set="foo.example.org")

1286

1287

def test_host_short(self):

1288

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

1289

"foo"], command.SetHost,

1290

value_to_set="foo.example.org")

1291

1292

def test_secret_devnull(self):

1293

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

1294

"foo"], command.SetSecret,

1295

value_to_set=b"")

1296

1297

def test_secret_tempfile(self):

1298

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

1299

value = b"secret\0xyzzy\nbar"

1300

f.write(value)

1301

f.seek(0)

1302

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

1303

"foo"], command.SetSecret,

1304

value_to_set=value)

1305

1306

def test_secret_devnull_short(self):

1307

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

1308

command.SetSecret,

1309

value_to_set=b"")

1310

1311

def test_secret_tempfile_short(self):

1312

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

1313

value = b"secret\0xyzzy\nbar"

1314

f.write(value)

1315

f.seek(0)

1316

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

1317

command.SetSecret,

1318

value_to_set=value)

1319

1320

def test_timeout(self):

1321

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

1322

command.SetTimeout,

1323

value_to_set=300000)

1324

1325

def test_timeout_short(self):

1326

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

1327

command.SetTimeout,

1328

value_to_set=300000)

1329

1330

def test_extended_timeout(self):

1331

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

1332

"foo"],

1333

command.SetExtendedTimeout,

1334

value_to_set=900000)

1335

1336

def test_interval(self):

1337

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

1338

command.SetInterval,

1339

value_to_set=120000)

1340

1341

def test_interval_short(self):

1342

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

1343

command.SetInterval,

1344

value_to_set=120000)

1345

1346

def test_approval_delay(self):

1347

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

1348

"foo"],

1349

command.SetApprovalDelay,

1350

value_to_set=30000)

1351

1352

def test_approval_duration(self):

1353

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

1354

"foo"],

1355

command.SetApprovalDuration,

1356

value_to_set=1000)

1357

1358

def test_print_table(self):

1359

self.assert_command_from_args([], command.PrintTable,

1360

verbose=False)

1361

1362

def test_print_table_verbose(self):

1363

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

1364

command.PrintTable,

1365

verbose=True)

1366

1367

def test_print_table_verbose_short(self):

1368

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

1369

verbose=True)

1370

1371

1372

class TestCommand(unittest.TestCase):

1373

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

1374

1375

def setUp(self):

1376

testcase = self

1377

class MockClient(object):

1378

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

1379

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

1380

self.attributes = attributes

1381

self.attributes["Name"] = name

1382

self.calls = []

1383

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

1384

testcase.assertEqual(client_dbus_interface, interface)

1385

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1386

dbus_interface)

1387

self.attributes[propname] = value

1388

def Approve(self, approve, dbus_interface):

1389

testcase.assertEqual(client_dbus_interface,

1390

dbus_interface)

1391

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

1392

dbus_interface)))

1393

self.client = MockClient(

1394

"foo",

1395

KeyID=("92ed150794387c03ce684574b1139a65"

1396

"94a34f895daaaf09fd8ea90a27cddb12"),

1397

Secret=b"secret",

1398

Host="foo.example.org",

1399

Enabled=dbus.Boolean(True),

1400

Timeout=300000,

1401

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

1402

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

1403

Interval=120000,

1404

Fingerprint=("778827225BA7DE539C5A"

1405

"7CFA59CFF7CDBD9A5920"),

1406

CheckerRunning=dbus.Boolean(False),

1407

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

1408

ApprovalPending=dbus.Boolean(False),

1409

ApprovedByDefault=dbus.Boolean(True),

1410

LastApprovalRequest="",

1411

ApprovalDelay=0,

1412

ApprovalDuration=1000,

1413

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

1414

ExtendedTimeout=900000,

1415

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

1416

LastCheckerStatus=0)

1417

self.other_client = MockClient(

1418

"barbar",

1419

KeyID=("0558568eedd67d622f5c83b35a115f79"

1420

"6ab612cff5ad227247e46c2b020f441c"),

1421

Secret=b"secretbar",

1422

Host="192.0.2.3",

1423

Enabled=dbus.Boolean(True),

1424

Timeout=300000,

1425

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

1426

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

1427

Interval=120000,

1428

Fingerprint=("3E393AEAEFB84C7E89E2"

1429

"F547B3A107558FCA3A27"),

1430

CheckerRunning=dbus.Boolean(True),

1431

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

1432

ApprovalPending=dbus.Boolean(False),

1433

ApprovedByDefault=dbus.Boolean(False),

1434

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

1435

ApprovalDelay=30000,

1436

ApprovalDuration=93785000,

1437

Checker=":",

1438

ExtendedTimeout=900000,

1439

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

1440

LastCheckerStatus=-2)

1441

self.clients = collections.OrderedDict(

1442

[

1443

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

1444

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

1445

])

1446

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

1447

1448

@property

1449

def bus(self):

1450

class Bus(object):

1451

@staticmethod

1452

def get_object(client_bus_name, path):

1453

self.assertEqual(dbus_busname, client_bus_name)

1454

return {

1455

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

1456

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

1457

"/clients/foo": self.client,

1458

"/clients/barbar": self.other_client,

1459

}[path]

1460

return Bus()

1461

1462

1463

class TestBaseCommands(TestCommand):

1464

1465

def test_IsEnabled_exits_successfully(self):

1466

with self.assertRaises(SystemExit) as e:

1467

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

1468

if e.exception.code is not None:

1469

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

1470

else:

1471

self.assertIsNone(e.exception.code)

1472

1473

def test_IsEnabled_exits_with_failure(self):

1474

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

1475

with self.assertRaises(SystemExit) as e:

1476

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

1477

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

1478

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

1479

else:

1480

self.assertIsNotNone(e.exception.code)

1481

1482

def test_Approve(self):

1483

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

1484

for clientpath in self.clients:

1485

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

1486

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

1487

client.calls)

1488

1489

def test_Deny(self):

1490

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

1491

for clientpath in self.clients:

1492

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

1493

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

1494

client.calls)

1495

1496

def test_Remove(self):

1497

class MockMandos(object):

1498

def __init__(self):

1499

self.calls = []

1500

def RemoveClient(self, dbus_path):

1501

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

1502

mandos = MockMandos()

1503

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

1504

for clientpath in self.clients:

1505

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

1506

mandos.calls)

1507

1508

expected_json = {

1509

"foo": {

1510

"Name": "foo",

1511

"KeyID": ("92ed150794387c03ce684574b1139a65"

1512

"94a34f895daaaf09fd8ea90a27cddb12"),

1513

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

1514

"Enabled": True,

1515

"Timeout": 300000,

1516

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

1517

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

1518

"Interval": 120000,

1519

"Fingerprint": ("778827225BA7DE539C5A"

1520

"7CFA59CFF7CDBD9A5920"),

1521

"CheckerRunning": False,

1522

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

1523

"ApprovalPending": False,

1524

"ApprovedByDefault": True,

1525

"LastApprovalRequest": "",

1526

"ApprovalDelay": 0,

1527

"ApprovalDuration": 1000,

1528

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

1529

"ExtendedTimeout": 900000,

1530

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

1531

"LastCheckerStatus": 0,

1532

},

1533

"barbar": {

1534

"Name": "barbar",

1535

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1536

"6ab612cff5ad227247e46c2b020f441c"),

1537

"Host": "192.0.2.3",

1538

"Enabled": True,

1539

"Timeout": 300000,

1540

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

1541

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

1542

"Interval": 120000,

1543

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1544

"F547B3A107558FCA3A27"),

1545

"CheckerRunning": True,

1546

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

1547

"ApprovalPending": False,

1548

"ApprovedByDefault": False,

1549

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

1550

"ApprovalDelay": 30000,

1551

"ApprovalDuration": 93785000,

1552

"Checker": ":",

1553

"ExtendedTimeout": 900000,

1554

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

1555

"LastCheckerStatus": -2,

1556

},

1557

}

1558

1559

def test_DumpJSON_normal(self):

1560

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

1561

json_data = json.loads(output)

1562

self.assertDictEqual(self.expected_json, json_data)

1563

1564

def test_DumpJSON_one_client(self):

1565

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

1566

json_data = json.loads(output)

1567

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

1568

self.assertDictEqual(expected_json, json_data)

1569

1570

def test_PrintTable_normal(self):

1571

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

1572

expected_output = "\n".join((

1573

"Name Enabled Timeout Last Successful Check",

1574

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

1575

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

1576

))

1577

self.assertEqual(expected_output, output)

1578

1579

def test_PrintTable_verbose(self):

1580

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

1581

self.clients.values())

1582

columns = (

1583

(

1584

"Name ",

1585

"foo ",

1586

"barbar ",

1587

),(

1588

"Enabled ",

1589

"Yes ",

1590

"Yes ",

1591

),(

1592

"Timeout ",

1593

"00:05:00 ",

1594

"00:05:00 ",

1595

),(

1596

"Last Successful Check ",

1597

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

1598

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

1599

),(

1600

"Created ",

1601

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

1602

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

1603

),(

1604

"Interval ",

1605

"00:02:00 ",

1606

"00:02:00 ",

1607

),(

1608

"Host ",

1609

"foo.example.org ",

1610

"192.0.2.3 ",

1611

),(

1612

("Key ID "

1613

" "),

1614

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1615

"ea90a27cddb12 "),

1616

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1617

"46c2b020f441c "),

1618

),(

1619

"Fingerprint ",

1620

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1621

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1622

),(

1623

"Check Is Running ",

1624

"No ",

1625

"Yes ",

1626

),(

1627

"Last Enabled ",

1628

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

1629

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

1630

),(

1631

"Approval Is Pending ",

1632

"No ",

1633

"No ",

1634

),(

1635

"Approved By Default ",

1636

"Yes ",

1637

"No ",

1638

),(

1639

"Last Approval Request ",

1640

" ",

1641

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

1642

),(

1643

"Approval Delay ",

1644

"00:00:00 ",

1645

"00:00:30 ",

1646

),(

1647

"Approval Duration ",

1648

"00:00:01 ",

1649

"1T02:03:05 ",

1650

),(

1651

"Checker ",

1652

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

1653

": ",

1654

),(

1655

"Extended Timeout ",

1656

"00:15:00 ",

1657

"00:15:00 ",

1658

),(

1659

"Expires ",

1660

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

1661

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

1662

),(

1663

"Last Checker Status",

1664

"0 ",

1665

"-2 ",

1666

)

1667

)

1668

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

1669

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

1670

for rows in columns)

1671

for line in range(num_lines))

1672

self.assertEqual(expected_output, output)

1673

1674

def test_PrintTable_one_client(self):

1675

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

1676

expected_output = "\n".join((

1677

"Name Enabled Timeout Last Successful Check",

1678

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

1679

))

1680

self.assertEqual(expected_output, output)

1681

1682

1683

class TestPropertyCmd(TestCommand):

1684

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

1685

def runTest(self):

1686

if not hasattr(self, "command"):

1687

return

1688

values_to_get = getattr(self, "values_to_get",

1689

self.values_to_set)

1690

for value_to_set, value_to_get in zip(self.values_to_set,

1691

values_to_get):

1692

for clientpath in self.clients:

1693

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

1694

old_value = client.attributes[self.propname]

1695

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

1696

self.run_command(value_to_set, self.clients)

1697

for clientpath in self.clients:

1698

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

1699

value = client.attributes[self.propname]

1700

self.assertNotIsInstance(value, self.Unique)

1701

self.assertEqual(value_to_get, value)

1702

1703

class Unique(object):

1704

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

1705

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

1706

1707

def run_command(self, value, clients):

1708

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

1709

1710

1711

class TestEnableCmd(TestPropertyCmd):

1712

command = command.Enable

1713

propname = "Enabled"

1714

values_to_set = [dbus.Boolean(True)]

1715

1716

1717

class TestDisableCmd(TestPropertyCmd):

1718

command = command.Disable

1719

propname = "Enabled"

1720

values_to_set = [dbus.Boolean(False)]

1721

1722

1723

class TestBumpTimeoutCmd(TestPropertyCmd):

1724

command = command.BumpTimeout

1725

propname = "LastCheckedOK"

1726

values_to_set = [""]

1727

1728

1729

class TestStartCheckerCmd(TestPropertyCmd):

1730

command = command.StartChecker

1731

propname = "CheckerRunning"

1732

values_to_set = [dbus.Boolean(True)]

1733

1734

1735

class TestStopCheckerCmd(TestPropertyCmd):

1736

command = command.StopChecker

1737

propname = "CheckerRunning"

1738

values_to_set = [dbus.Boolean(False)]

1739

1740

1741

class TestApproveByDefaultCmd(TestPropertyCmd):

1742

command = command.ApproveByDefault

1743

propname = "ApprovedByDefault"

1744

values_to_set = [dbus.Boolean(True)]

1745

1746

1747

class TestDenyByDefaultCmd(TestPropertyCmd):

1748

command = command.DenyByDefault

1749

propname = "ApprovedByDefault"

1750

values_to_set = [dbus.Boolean(False)]

1751

1752

1753

class TestPropertyValueCmd(TestPropertyCmd):

1754

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

1755

1756

def runTest(self):

1757

if type(self) is TestPropertyValueCmd:

1758

return

1759

return super(TestPropertyValueCmd, self).runTest()

1760

1761

def run_command(self, value, clients):

1762

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

1763

1764

1765

class TestSetCheckerCmd(TestPropertyValueCmd):

1766

command = command.SetChecker

1767

propname = "Checker"

1768

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

1769

1770

1771

class TestSetHostCmd(TestPropertyValueCmd):

1772

command = command.SetHost

1773

propname = "Host"

1774

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

1775

1776

1777

class TestSetSecretCmd(TestPropertyValueCmd):

1778

command = command.SetSecret

1779

propname = "Secret"

1780

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

1781

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

1782

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

1783

1784

1785

class TestSetTimeoutCmd(TestPropertyValueCmd):

1786

command = command.SetTimeout

1787

propname = "Timeout"

1788

values_to_set = [datetime.timedelta(),

1789

datetime.timedelta(minutes=5),

1790

datetime.timedelta(seconds=1),

1791

datetime.timedelta(weeks=1),

1792

datetime.timedelta(weeks=52)]

1793

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

1794

1795

1796

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1797

command = command.SetExtendedTimeout

1798

propname = "ExtendedTimeout"

1799

values_to_set = [datetime.timedelta(),

1800

datetime.timedelta(minutes=5),

1801

datetime.timedelta(seconds=1),

1802

datetime.timedelta(weeks=1),

1803

datetime.timedelta(weeks=52)]

1804

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

1805

1806

1807

class TestSetIntervalCmd(TestPropertyValueCmd):

1808

command = command.SetInterval

1809

propname = "Interval"

1810

values_to_set = [datetime.timedelta(),

1811

datetime.timedelta(minutes=5),

1812

datetime.timedelta(seconds=1),

1813

datetime.timedelta(weeks=1),

1814

datetime.timedelta(weeks=52)]

1815

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

1816

1817

1818

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1819

command = command.SetApprovalDelay

1820

propname = "ApprovalDelay"

1821

values_to_set = [datetime.timedelta(),

1822

datetime.timedelta(minutes=5),

1823

datetime.timedelta(seconds=1),

1824

datetime.timedelta(weeks=1),

1825

datetime.timedelta(weeks=52)]

1826

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

1827

1828

1829

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1830

command = command.SetApprovalDuration

1831

propname = "ApprovalDuration"

1832

values_to_set = [datetime.timedelta(),

1833

datetime.timedelta(minutes=5),

1834

datetime.timedelta(seconds=1),

1835

datetime.timedelta(weeks=1),

1836

datetime.timedelta(weeks=52)]

1837

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

1838

1839

1840

1841

def should_only_run_tests():

1842

parser = argparse.ArgumentParser(add_help=False)

1843

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

1844

args, unknown_args = parser.parse_known_args()

1845

run_tests = args.check

1846

if run_tests:

1847

# Remove --check argument from sys.argv

1848

sys.argv[1:] = unknown_args

1849

return run_tests

1850

1851

# Add all tests from doctest strings

1852

def load_tests(loader, tests, none):

1853

import doctest

1854

tests.addTests(doctest.DocTestSuite())

1855

return tests

358

1856

359

1857

if __name__ == "__main__":

360

main()

1858

try:

1859

if should_only_run_tests():

1860

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

1861

unittest.main()

1862

else:

1863

main()

1864

finally:

1865

logging.shutdown()