/mandos/release : revision 237.7.625

1

#!/usr/bin/python

2

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

3

#

2

# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This program is free software: you can redistribute it and/or modify

10

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

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

# Mandos is free software: you can redistribute it and/or modify it

12

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

11

13

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

12

14

# (at your option) any later version.

13

15

#

14

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

15

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

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

16

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17

19

# GNU General Public License for more details.

18

#

20

#

19

21

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

20

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

21

#

22

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

23

#

24

25

from __future__ import division

22

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

23

#

24

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

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

26

35

import sys

27

import dbus

28

from optparse import OptionParser

36

import argparse

29

37

import locale

30

38

import datetime

31

39

import re

32

40

import os

33

34

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

35

36

tablewords = {

37

'Name': u'Name',

38

'Enabled': u'Enabled',

39

'Timeout': u'Timeout',

40

'LastCheckedOK': u'Last Successful Check',

41

'LastApprovalRequest': u'Last Approval Request',

42

'Created': u'Created',

43

'Interval': u'Interval',

44

'Host': u'Host',

45

'Fingerprint': u'Fingerprint',

46

'CheckerRunning': u'Check Is Running',

47

'LastEnabled': u'Last Enabled',

48

'ApprovalPending': u'Approval Is Pending',

49

'ApprovedByDefault': u'Approved By Default',

50

'ApprovalDelay': u"Approval Delay",

51

'ApprovalDuration': u"Approval Duration",

52

'Checker': u'Checker',

53

}

54

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

55

domain = 'se.bsnet.fukt'

56

busname = domain + '.Mandos'

57

server_path = '/'

58

server_interface = domain + '.Mandos'

59

client_interface = domain + '.Mandos.Client'

60

version = "1.2.3"

61

62

def timedelta_to_milliseconds(td):

63

"Convert a datetime.timedelta object to milliseconds"

64

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

65

+ (td.seconds * 1000)

66

+ (td.microseconds // 1000))

67

68

def milliseconds_to_string(ms):

69

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

70

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

71

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

72

"hours": td.seconds // 3600,

73

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

74

"seconds": td.seconds % 60,

75

})

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

import dbus

50

51

# Show warnings by default

52

if not sys.warnoptions:

53

import warnings

54

warnings.simplefilter("default")

55

56

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

57

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

58

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

59

60

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

61

62

if sys.version_info.major == 2:

63

str = unicode

64

65

locale.setlocale(locale.LC_ALL, "")

66

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

81

82

def main():

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

135

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

136

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

137

help="show version number and exit")

138

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

139

help="Select all clients")

140

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

141

help="Print all fields")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

150

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

151

help="Bump timeout for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

159

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

160

help="Check if client is enabled")

161

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

162

help="Remove client")

163

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

164

help="Set checker command for client")

165

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

166

help="Set timeout for client")

167

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

168

help="Set extended timeout for client")

169

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

170

help="Set checker interval for client")

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

181

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

182

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

183

help="Set duration of one client approval")

184

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

185

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

186

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

187

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

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

198

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

76

199

77

200

78

201

def string_to_delta(interval):

79

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

80

81

>>> string_to_delta('7d')

82

datetime.timedelta(7)

83

>>> string_to_delta('60s')

84

datetime.timedelta(0, 60)

85

>>> string_to_delta('60m')

86

datetime.timedelta(0, 3600)

87

>>> string_to_delta('24h')

88

datetime.timedelta(1)

89

>>> string_to_delta(u'1w')

90

datetime.timedelta(7)

91

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

92

datetime.timedelta(0, 330)

93

"""

94

timevalue = datetime.timedelta(0)

95

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

96

97

for s in regexp.findall(interval):

98

try:

99

suffix = unicode(s[-1])

100

value = int(s[:-1])

101

if suffix == u"d":

102

delta = datetime.timedelta(value)

103

elif suffix == u"s":

104

delta = datetime.timedelta(0, value)

105

elif suffix == u"m":

106

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

107

elif suffix == u"h":

108

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

109

elif suffix == u"w":

110

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

111

else:

112

raise ValueError

113

except (ValueError, IndexError):

114

raise ValueError

115

timevalue += delta

116

return timevalue

117

118

def print_clients(clients, keywords):

119

def valuetostring(value, keyword):

120

if type(value) is dbus.Boolean:

121

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

122

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

123

u"ApprovalDuration"):

124

return milliseconds_to_string(value)

125

return unicode(value)

126

127

# Create format string to print table rows

128

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

129

max(len(tablewords[key]),

130

max(len(valuetostring(client[key],

131

key))

132

for client in

133

clients))

134

for key in keywords)

135

# Print header line

136

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

137

for client in clients:

138

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

139

for key in keywords)

140

141

def has_actions(options):

142

return any((options.enable,

143

options.disable,

144

options.bump_timeout,

145

options.start_checker,

146

options.stop_checker,

147

options.is_enabled,

148

options.remove,

149

options.checker is not None,

150

options.timeout is not None,

151

options.interval is not None,

152

options.approved_by_default is not None,

153

options.approval_delay is not None,

154

options.approval_duration is not None,

155

options.host is not None,

156

options.secret is not None,

157

options.approve,

158

options.deny))

159

160

def main():

161

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

162

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

163

help="Select all clients")

164

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

165

help="Print all fields")

166

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

167

help="Enable client")

168

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

169

help="disable client")

170

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

171

help="Bump timeout for client")

172

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

173

help="Start checker for client")

174

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

175

help="Stop checker for client")

176

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

177

help="Check if client is enabled")

178

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

179

help="Remove client")

180

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

181

help="Set checker command for client")

182

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

183

help="Set timeout for client")

184

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

185

help="Set checker interval for client")

186

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

187

dest=u"approved_by_default",

188

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

189

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

190

dest=u"approved_by_default",

191

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

192

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

193

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

194

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

195

help="Set duration of one client approval")

196

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

197

help="Set host for client")

198

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

199

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

200

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

201

help="Approve any current client request")

202

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

203

help="Deny any current client request")

204

options, client_names = parser.parse_args()

205

206

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

207

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

208

if options.verbose and has_actions(options):

209

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

210

' --all.')

211

if options.all and not has_actions(options):

212

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

213

214

try:

215

bus = dbus.SystemBus()

216

mandos_dbus_objc = bus.get_object(busname, server_path)

217

except dbus.exceptions.DBusException:

218

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

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

213

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

214

215

>>> rfc3339_duration_to_delta("P7D")

216

datetime.timedelta(7)

217

>>> rfc3339_duration_to_delta("PT60S")

218

datetime.timedelta(0, 60)

219

>>> rfc3339_duration_to_delta("PT60M")

220

datetime.timedelta(0, 3600)

221

>>> rfc3339_duration_to_delta("P60M")

222

datetime.timedelta(1680)

223

>>> rfc3339_duration_to_delta("PT24H")

224

datetime.timedelta(1)

225

>>> rfc3339_duration_to_delta("P1W")

226

datetime.timedelta(7)

227

>>> rfc3339_duration_to_delta("PT5M30S")

228

datetime.timedelta(0, 330)

229

>>> rfc3339_duration_to_delta("P1DT3M20S")

230

datetime.timedelta(1, 200)

231

>>> # Can not be empty:

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: ""

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

ValueError: Invalid RFC 3339 duration: "1D"

241

>>> # Must use correct order

242

>>> rfc3339_duration_to_delta("PT1S2M")

243

Traceback (most recent call last):

244

...

245

ValueError: Invalid RFC 3339 duration: "PT1S2M"

246

>>> # Time needs time marker

247

>>> rfc3339_duration_to_delta("P1H2S")

248

Traceback (most recent call last):

249

...

250

ValueError: Invalid RFC 3339 duration: "P1H2S"

251

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

252

>>> rfc3339_duration_to_delta("P1D2W")

253

Traceback (most recent call last):

254

...

255

ValueError: Invalid RFC 3339 duration: "P1D2W"

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: "P2W2H"

260

"""

261

262

# Parsing an RFC 3339 duration with regular expressions is not

263

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

264

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

265

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

266

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

267

# avoid excessive use of external libraries.

268

269

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

270

Token = collections.namedtuple("Token", (

271

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

272

# a "group" containing digits

273

"value", # datetime.timedelta or None

274

"followers")) # Tokens valid after this token

275

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

276

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

277

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

278

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

279

datetime.timedelta(seconds=1),

280

frozenset((token_end, )))

281

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

282

datetime.timedelta(minutes=1),

283

frozenset((token_second, token_end)))

284

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

285

datetime.timedelta(hours=1),

286

frozenset((token_minute, token_end)))

287

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

288

None,

289

frozenset((token_hour, token_minute,

290

token_second)))

291

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

292

datetime.timedelta(days=1),

293

frozenset((token_time, token_end)))

294

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

295

datetime.timedelta(weeks=4),

296

frozenset((token_day, token_end)))

297

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

298

datetime.timedelta(weeks=52),

299

frozenset((token_month, token_end)))

300

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

301

datetime.timedelta(weeks=1),

302

frozenset((token_end, )))

303

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

304

frozenset((token_year, token_month,

305

token_day, token_time,

306

token_week)))

307

# Define starting values:

308

# Value so far

309

value = datetime.timedelta()

310

found_token = None

311

# Following valid tokens

312

followers = frozenset((token_duration, ))

313

# String left to parse

314

s = duration

315

# Loop until end token is found

316

while found_token is not token_end:

317

# Search for any currently valid tokens

318

for token in followers:

319

match = token.regexp.match(s)

320

if match is not None:

321

# Token found

322

if token.value is not None:

323

# Value found, parse digits

324

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

325

# Add to value so far

326

value += factor * token.value

327

# Strip token from string

328

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

329

# Go to found token

330

found_token = token

331

# Set valid next tokens

332

followers = found_token.followers

333

break

334

else:

335

# No currently valid tokens were found

336

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

337

.format(duration))

338

# End token found

339

return value

340

341

342

def parse_pre_1_6_1_interval(interval):

343

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

344

and return a datetime.timedelta

345

346

>>> parse_pre_1_6_1_interval('7d')

347

datetime.timedelta(7)

348

>>> parse_pre_1_6_1_interval('60s')

349

datetime.timedelta(0, 60)

350

>>> parse_pre_1_6_1_interval('60m')

351

datetime.timedelta(0, 3600)

352

>>> parse_pre_1_6_1_interval('24h')

353

datetime.timedelta(1)

354

>>> parse_pre_1_6_1_interval('1w')

355

datetime.timedelta(7)

356

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

357

datetime.timedelta(0, 330)

358

>>> parse_pre_1_6_1_interval('')

359

datetime.timedelta(0)

360

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

361

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

362

datetime.timedelta(2, 480, 18000)

363

364

"""

365

366

value = datetime.timedelta(0)

367

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

368

369

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

370

if suffix == "d":

371

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

372

elif suffix == "s":

373

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

374

elif suffix == "m":

375

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

376

elif suffix == "h":

377

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

378

elif suffix == "w":

379

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

380

elif suffix == "":

381

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

382

return value

383

384

385

def check_option_syntax(parser, options):

386

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

387

argparse"""

388

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

408

409

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

410

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

411

if options.verbose and has_actions(options):

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

416

if options.all and not has_actions(options):

417

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

418

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

419

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

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

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

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

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

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)

219

573

sys.exit(1)

220

221

mandos_serv = dbus.Interface(mandos_dbus_objc,

222

dbus_interface = server_interface)

223

224

#block stderr since dbus library prints to stderr

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

def test_handles_basic_rfc3339(self):

890

self.assertEqual(string_to_delta("PT0S"),

891

datetime.timedelta())

892

self.assertEqual(string_to_delta("P0D"),

893

datetime.timedelta())

894

self.assertEqual(string_to_delta("PT1S"),

895

datetime.timedelta(0, 1))

896

self.assertEqual(string_to_delta("PT2H"),

897

datetime.timedelta(0, 7200))

898

899

def test_falls_back_to_pre_1_6_1_with_warning(self):

900

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

901

value = string_to_delta("2h")

902

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

903

904

905

class Test_check_option_syntax(unittest.TestCase):

906

def setUp(self):

907

self.parser = argparse.ArgumentParser()

908

add_command_line_options(self.parser)

909

910

def test_actions_requires_client_or_all(self):

911

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

912

options = self.parser.parse_args()

913

setattr(options, action, value)

914

with self.assertParseError():

915

self.check_option_syntax(options)

916

917

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

918

# check_option_syntax()

919

actions = {

920

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

921

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

922

# never used

923

"enable": True,

924

"disable": True,

925

"bump_timeout": True,

926

"start_checker": True,

927

"stop_checker": True,

928

"is_enabled": True,

929

"remove": True,

930

"checker": "x",

931

"timeout": datetime.timedelta(),

932

"extended_timeout": datetime.timedelta(),

933

"interval": datetime.timedelta(),

934

"approved_by_default": True,

935

"approval_delay": datetime.timedelta(),

936

"approval_duration": datetime.timedelta(),

937

"host": "x",

938

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

939

"approve": True,

940

"deny": True,

941

}

942

943

@contextlib.contextmanager

944

def assertParseError(self):

945

with self.assertRaises(SystemExit) as e:

946

with self.redirect_stderr_to_devnull():

947

yield

948

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

949

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

950

# /argparse.html#exiting-methods

951

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

952

953

@staticmethod

954

@contextlib.contextmanager

955

def redirect_stderr_to_devnull():

225

956

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

226

957

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

227

958

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

228

959

os.close(null)

229

960

try:

230

try:

231

mandos_clients = mandos_serv.GetAllClientsWithProperties()

232

finally:

233

#restore stderr

234

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

235

os.close(stderrcopy)

236

except dbus.exceptions.DBusException, e:

237

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

238

sys.exit(1)

239

240

# Compile dict of (clients: properties) to process

241

clients={}

242

243

if options.all or not client_names:

244

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

245

for path, properties in

246

mandos_clients.iteritems())

247

else:

248

for name in client_names:

249

for path, client in mandos_clients.iteritems():

250

if client['Name'] == name:

251

client_objc = bus.get_object(busname, path)

252

clients[client_objc] = client

253

break

254

else:

255

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

256

sys.exit(1)

257

258

if not has_actions(options) and clients:

259

if options.verbose:

260

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

261

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

262

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

263

'LastEnabled', 'ApprovalPending',

264

'ApprovedByDefault',

265

'LastApprovalRequest', 'ApprovalDelay',

266

'ApprovalDuration', 'Checker')

267

else:

268

keywords = defaultkeywords

269

270

print_clients(clients.values(), keywords)

271

else:

272

# Process each client in the list by all selected options

273

for client in clients:

274

if options.remove:

275

mandos_serv.RemoveClient(client.__dbus_object_path__)

276

if options.enable:

277

client.Enable(dbus_interface=client_interface)

278

if options.disable:

279

client.Disable(dbus_interface=client_interface)

280

if options.bump_timeout:

281

client.CheckedOK(dbus_interface=client_interface)

282

if options.start_checker:

283

client.StartChecker(dbus_interface=client_interface)

284

if options.stop_checker:

285

client.StopChecker(dbus_interface=client_interface)

286

if options.is_enabled:

287

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

288

u"Enabled",

289

dbus_interface=dbus.PROPERTIES_IFACE)

290

else 1)

291

if options.checker:

292

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

293

dbus_interface=dbus.PROPERTIES_IFACE)

294

if options.host:

295

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

296

dbus_interface=dbus.PROPERTIES_IFACE)

297

if options.interval:

298

client.Set(client_interface, u"Interval",

299

timedelta_to_milliseconds

300

(string_to_delta(options.interval)),

301

dbus_interface=dbus.PROPERTIES_IFACE)

302

if options.approval_delay:

303

client.Set(client_interface, u"ApprovalDelay",

304

timedelta_to_milliseconds

305

(string_to_delta(options.

306

approval_delay)),

307

dbus_interface=dbus.PROPERTIES_IFACE)

308

if options.approval_duration:

309

client.Set(client_interface, u"ApprovalDuration",

310

timedelta_to_milliseconds

311

(string_to_delta(options.

312

approval_duration)),

313

dbus_interface=dbus.PROPERTIES_IFACE)

314

if options.timeout:

315

client.Set(client_interface, u"Timeout",

316

timedelta_to_milliseconds

317

(string_to_delta(options.timeout)),

318

dbus_interface=dbus.PROPERTIES_IFACE)

319

if options.secret:

320

client.Set(client_interface, u"Secret",

321

dbus.ByteArray(open(options.secret,

322

u'rb').read()),

323

dbus_interface=dbus.PROPERTIES_IFACE)

324

if options.approved_by_default is not None:

325

client.Set(client_interface, u"ApprovedByDefault",

326

dbus.Boolean(options

327

.approved_by_default),

328

dbus_interface=dbus.PROPERTIES_IFACE)

329

if options.approve:

330

client.Approve(dbus.Boolean(True),

331

dbus_interface=client_interface)

332

elif options.deny:

333

client.Approve(dbus.Boolean(False),

334

dbus_interface=client_interface)

335

336

if __name__ == '__main__':

337

main()

961

yield

962

finally:

963

# restore stderr

964

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

965

os.close(stderrcopy)

966

967

def check_option_syntax(self, options):

968

check_option_syntax(self.parser, options)

969

970

def test_actions_all_conflicts_with_verbose(self):

971

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

972

options = self.parser.parse_args()

973

setattr(options, action, value)

974

options.all = True

975

options.verbose = True

976

with self.assertParseError():

977

self.check_option_syntax(options)

978

979

def test_actions_with_client_conflicts_with_verbose(self):

980

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

981

options = self.parser.parse_args()

982

setattr(options, action, value)

983

options.verbose = True

984

options.client = ["foo"]

985

with self.assertParseError():

986

self.check_option_syntax(options)

987

988

def test_dump_json_conflicts_with_verbose(self):

989

options = self.parser.parse_args()

990

options.dump_json = True

991

options.verbose = True

992

with self.assertParseError():

993

self.check_option_syntax(options)

994

995

def test_dump_json_conflicts_with_action(self):

996

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

997

options = self.parser.parse_args()

998

setattr(options, action, value)

999

options.dump_json = True

1000

with self.assertParseError():

1001

self.check_option_syntax(options)

1002

1003

def test_all_can_not_be_alone(self):

1004

options = self.parser.parse_args()

1005

options.all = True

1006

with self.assertParseError():

1007

self.check_option_syntax(options)

1008

1009

def test_all_is_ok_with_any_action(self):

1010

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

1011

options = self.parser.parse_args()

1012

setattr(options, action, value)

1013

options.all = True

1014

self.check_option_syntax(options)

1015

1016

def test_any_action_is_ok_with_one_client(self):

1017

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

1018

options = self.parser.parse_args()

1019

setattr(options, action, value)

1020

options.client = ["foo"]

1021

self.check_option_syntax(options)

1022

1023

def test_one_client_with_all_actions_except_is_enabled(self):

1024

options = self.parser.parse_args()

1025

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

1026

if action == "is_enabled":

1027

continue

1028

setattr(options, action, value)

1029

options.client = ["foo"]

1030

self.check_option_syntax(options)

1031

1032

def test_two_clients_with_all_actions_except_is_enabled(self):

1033

options = self.parser.parse_args()

1034

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

1035

if action == "is_enabled":

1036

continue

1037

setattr(options, action, value)

1038

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

1039

self.check_option_syntax(options)

1040

1041

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1042

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

1043

if action == "is_enabled":

1044

continue

1045

options = self.parser.parse_args()

1046

setattr(options, action, value)

1047

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

1048

self.check_option_syntax(options)

1049

1050

def test_is_enabled_fails_without_client(self):

1051

options = self.parser.parse_args()

1052

options.is_enabled = True

1053

with self.assertParseError():

1054

self.check_option_syntax(options)

1055

1056

def test_is_enabled_fails_with_two_clients(self):

1057

options = self.parser.parse_args()

1058

options.is_enabled = True

1059

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

1060

with self.assertParseError():

1061

self.check_option_syntax(options)

1062

1063

def test_remove_can_only_be_combined_with_action_deny(self):

1064

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

1065

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

1066

continue

1067

options = self.parser.parse_args()

1068

setattr(options, action, value)

1069

options.all = True

1070

options.remove = True

1071

with self.assertParseError():

1072

self.check_option_syntax(options)

1073

1074

1075

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1076

def test_calls_and_returns_get_object_on_bus(self):

1077

class MockBus(object):

1078

called = False

1079

def get_object(mockbus_self, busname, dbus_path):

1080

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

1081

# this MockBus instance is in "mockbus_self".

1082

self.assertEqual(busname, dbus_busname)

1083

self.assertEqual(dbus_path, server_dbus_path)

1084

mockbus_self.called = True

1085

return mockbus_self

1086

1087

mockbus = get_mandos_dbus_object(bus=MockBus())

1088

self.assertIsInstance(mockbus, MockBus)

1089

self.assertTrue(mockbus.called)

1090

1091

def test_logs_and_exits_on_dbus_error(self):

1092

class MockBusFailing(object):

1093

def get_object(self, busname, dbus_path):

1094

raise dbus.exceptions.DBusException("Test")

1095

1096

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

1097

with self.assertRaises(SystemExit) as e:

1098

bus = get_mandos_dbus_object(bus=MockBusFailing())

1099

1100

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

1101

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

1102

else:

1103

self.assertIsNotNone(e.exception.code)

1104

1105

1106

class Test_get_managed_objects(TestCaseWithAssertLogs):

1107

def test_calls_and_returns_GetManagedObjects(self):

1108

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

1109

class MockObjectManager(object):

1110

def GetManagedObjects(self):

1111

return managed_objects

1112

retval = get_managed_objects(MockObjectManager())

1113

self.assertDictEqual(managed_objects, retval)

1114

1115

def test_logs_and_exits_on_dbus_error(self):

1116

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

1117

1118

class MockObjectManagerFailing(object):

1119

def GetManagedObjects(self):

1120

dbus_logger.error("Test")

1121

raise dbus.exceptions.DBusException("Test")

1122

1123

class CountingHandler(logging.Handler):

1124

count = 0

1125

def emit(self, record):

1126

self.count += 1

1127

1128

counting_handler = CountingHandler()

1129

1130

dbus_logger.addHandler(counting_handler)

1131

1132

try:

1133

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

1134

with self.assertRaises(SystemExit) as e:

1135

get_managed_objects(MockObjectManagerFailing())

1136

finally:

1137

dbus_logger.removeFilter(counting_handler)

1138

1139

# Make sure the dbus logger was suppressed

1140

self.assertEqual(counting_handler.count, 0)

1141

1142

# Test that the dbus_logger still works

1143

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

1144

dbus_logger.error("Test")

1145

1146

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

1147

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

1148

else:

1149

self.assertIsNotNone(e.exception.code)

1150

1151

1152

class Test_commands_from_options(unittest.TestCase):

1153

def setUp(self):

1154

self.parser = argparse.ArgumentParser()

1155

add_command_line_options(self.parser)

1156

1157

def test_is_enabled(self):

1158

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

1159

command.IsEnabled)

1160

1161

def assert_command_from_args(self, args, command_cls,

1162

**cmd_attrs):

1163

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

1164

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

1165

options = self.parser.parse_args(args)

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 1)

1169

command = commands[0]

1170

self.assertIsInstance(command, command_cls)

1171

for key, value in cmd_attrs.items():

1172

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

1173

1174

def test_is_enabled_short(self):

1175

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

1176

command.IsEnabled)

1177

1178

def test_approve(self):

1179

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

1180

command.Approve)

1181

1182

def test_approve_short(self):

1183

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

1184

1185

def test_deny(self):

1186

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

1187

1188

def test_deny_short(self):

1189

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

1190

1191

def test_remove(self):

1192

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

1193

command.Remove)

1194

1195

def test_deny_before_remove(self):

1196

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

1197

"foo"])

1198

check_option_syntax(self.parser, options)

1199

commands = commands_from_options(options)

1200

self.assertEqual(len(commands), 2)

1201

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

1202

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

1203

1204

def test_deny_before_remove_reversed(self):

1205

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

1206

"--all"])

1207

check_option_syntax(self.parser, options)

1208

commands = commands_from_options(options)

1209

self.assertEqual(len(commands), 2)

1210

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

1211

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

1212

1213

def test_remove_short(self):

1214

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

1215

1216

def test_dump_json(self):

1217

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

1218

command.DumpJSON)

1219

1220

def test_enable(self):

1221

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

1222

command.Enable)

1223

1224

def test_enable_short(self):

1225

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

1226

1227

def test_disable(self):

1228

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

1229

command.Disable)

1230

1231

def test_disable_short(self):

1232

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

1233

1234

def test_bump_timeout(self):

1235

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

1236

command.BumpTimeout)

1237

1238

def test_bump_timeout_short(self):

1239

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

1240

command.BumpTimeout)

1241

1242

def test_start_checker(self):

1243

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

1244

command.StartChecker)

1245

1246

def test_stop_checker(self):

1247

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

1248

command.StopChecker)

1249

1250

def test_approve_by_default(self):

1251

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

1252

command.ApproveByDefault)

1253

1254

def test_deny_by_default(self):

1255

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

1256

command.DenyByDefault)

1257

1258

def test_checker(self):

1259

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

1260

command.SetChecker,

1261

value_to_set=":")

1262

1263

def test_checker_empty(self):

1264

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

1265

command.SetChecker,

1266

value_to_set="")

1267

1268

def test_checker_short(self):

1269

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

1270

command.SetChecker,

1271

value_to_set=":")

1272

1273

def test_host(self):

1274

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

1275

"foo"], command.SetHost,

1276

value_to_set="foo.example.org")

1277

1278

def test_host_short(self):

1279

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

1280

"foo"], command.SetHost,

1281

value_to_set="foo.example.org")

1282

1283

def test_secret_devnull(self):

1284

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

1285

"foo"], command.SetSecret,

1286

value_to_set=b"")

1287

1288

def test_secret_tempfile(self):

1289

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

1290

value = b"secret\0xyzzy\nbar"

1291

f.write(value)

1292

f.seek(0)

1293

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

1294

"foo"], command.SetSecret,

1295

value_to_set=value)

1296

1297

def test_secret_devnull_short(self):

1298

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

1299

command.SetSecret,

1300

value_to_set=b"")

1301

1302

def test_secret_tempfile_short(self):

1303

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

1304

value = b"secret\0xyzzy\nbar"

1305

f.write(value)

1306

f.seek(0)

1307

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

1308

command.SetSecret,

1309

value_to_set=value)

1310

1311

def test_timeout(self):

1312

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

1313

command.SetTimeout,

1314

value_to_set=300000)

1315

1316

def test_timeout_short(self):

1317

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

1318

command.SetTimeout,

1319

value_to_set=300000)

1320

1321

def test_extended_timeout(self):

1322

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

1323

"foo"],

1324

command.SetExtendedTimeout,

1325

value_to_set=900000)

1326

1327

def test_interval(self):

1328

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

1329

command.SetInterval,

1330

value_to_set=120000)

1331

1332

def test_interval_short(self):

1333

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

1334

command.SetInterval,

1335

value_to_set=120000)

1336

1337

def test_approval_delay(self):

1338

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

1339

"foo"],

1340

command.SetApprovalDelay,

1341

value_to_set=30000)

1342

1343

def test_approval_duration(self):

1344

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

1345

"foo"],

1346

command.SetApprovalDuration,

1347

value_to_set=1000)

1348

1349

def test_print_table(self):

1350

self.assert_command_from_args([], command.PrintTable,

1351

verbose=False)

1352

1353

def test_print_table_verbose(self):

1354

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

1355

command.PrintTable,

1356

verbose=True)

1357

1358

def test_print_table_verbose_short(self):

1359

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

1360

verbose=True)

1361

1362

1363

class TestCommand(unittest.TestCase):

1364

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

1365

1366

def setUp(self):

1367

testcase = self

1368

class MockClient(object):

1369

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

1370

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

1371

self.attributes = attributes

1372

self.attributes["Name"] = name

1373

self.calls = []

1374

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

1375

testcase.assertEqual(interface, client_dbus_interface)

1376

testcase.assertEqual(dbus_interface,

1377

dbus.PROPERTIES_IFACE)

1378

self.attributes[propname] = value

1379

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

1380

testcase.assertEqual(interface, client_dbus_interface)

1381

testcase.assertEqual(dbus_interface,

1382

dbus.PROPERTIES_IFACE)

1383

return self.attributes[propname]

1384

def Approve(self, approve, dbus_interface):

1385

testcase.assertEqual(dbus_interface,

1386

client_dbus_interface)

1387

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

1388

dbus_interface)))

1389

self.client = MockClient(

1390

"foo",

1391

KeyID=("92ed150794387c03ce684574b1139a65"

1392

"94a34f895daaaf09fd8ea90a27cddb12"),

1393

Secret=b"secret",

1394

Host="foo.example.org",

1395

Enabled=dbus.Boolean(True),

1396

Timeout=300000,

1397

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

1398

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

1399

Interval=120000,

1400

Fingerprint=("778827225BA7DE539C5A"

1401

"7CFA59CFF7CDBD9A5920"),

1402

CheckerRunning=dbus.Boolean(False),

1403

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

1404

ApprovalPending=dbus.Boolean(False),

1405

ApprovedByDefault=dbus.Boolean(True),

1406

LastApprovalRequest="",

1407

ApprovalDelay=0,

1408

ApprovalDuration=1000,

1409

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

1410

ExtendedTimeout=900000,

1411

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

1412

LastCheckerStatus=0)

1413

self.other_client = MockClient(

1414

"barbar",

1415

KeyID=("0558568eedd67d622f5c83b35a115f79"

1416

"6ab612cff5ad227247e46c2b020f441c"),

1417

Secret=b"secretbar",

1418

Host="192.0.2.3",

1419

Enabled=dbus.Boolean(True),

1420

Timeout=300000,

1421

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

1422

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

1423

Interval=120000,

1424

Fingerprint=("3E393AEAEFB84C7E89E2"

1425

"F547B3A107558FCA3A27"),

1426

CheckerRunning=dbus.Boolean(True),

1427

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

1428

ApprovalPending=dbus.Boolean(False),

1429

ApprovedByDefault=dbus.Boolean(False),

1430

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

1431

ApprovalDelay=30000,

1432

ApprovalDuration=93785000,

1433

Checker=":",

1434

ExtendedTimeout=900000,

1435

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

1436

LastCheckerStatus=-2)

1437

self.clients = collections.OrderedDict(

1438

[

1439

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

1440

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

1441

])

1442

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

1443

1444

@property

1445

def bus(self):

1446

class Bus(object):

1447

@staticmethod

1448

def get_object(client_bus_name, path):

1449

self.assertEqual(client_bus_name, dbus_busname)

1450

return {

1451

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

1452

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

1453

"/clients/foo": self.client,

1454

"/clients/barbar": self.other_client,

1455

}[path]

1456

return Bus()

1457

1458

1459

class TestBaseCommands(TestCommand):

1460

1461

def test_is_enabled(self):

1462

self.assertTrue(all(command.IsEnabled().is_enabled(client,

1463

properties)

1464

for client, properties

1465

in self.clients.items()))

1466

1467

def test_is_enabled_run_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(e.exception.code, 0)

1472

else:

1473

self.assertIsNone(e.exception.code)

1474

1475

def test_is_enabled_run_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(e.exception.code, 0)

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 MockMandos(object):

1500

def __init__(self):

1501

self.calls = []

1502

def RemoveClient(self, dbus_path):

1503

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

1504

mandos = MockMandos()

1505

super(TestBaseCommands, self).setUp()

1506

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

1507

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

1508

for clientpath in self.clients:

1509

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

1510

mandos.calls)

1511

1512

expected_json = {

1513

"foo": {

1514

"Name": "foo",

1515

"KeyID": ("92ed150794387c03ce684574b1139a65"

1516

"94a34f895daaaf09fd8ea90a27cddb12"),

1517

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

1518

"Enabled": True,

1519

"Timeout": 300000,

1520

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

1521

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

1522

"Interval": 120000,

1523

"Fingerprint": ("778827225BA7DE539C5A"

1524

"7CFA59CFF7CDBD9A5920"),

1525

"CheckerRunning": False,

1526

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

1527

"ApprovalPending": False,

1528

"ApprovedByDefault": True,

1529

"LastApprovalRequest": "",

1530

"ApprovalDelay": 0,

1531

"ApprovalDuration": 1000,

1532

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

1533

"ExtendedTimeout": 900000,

1534

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

1535

"LastCheckerStatus": 0,

1536

},

1537

"barbar": {

1538

"Name": "barbar",

1539

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1540

"6ab612cff5ad227247e46c2b020f441c"),

1541

"Host": "192.0.2.3",

1542

"Enabled": True,

1543

"Timeout": 300000,

1544

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

1545

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

1546

"Interval": 120000,

1547

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1548

"F547B3A107558FCA3A27"),

1549

"CheckerRunning": True,

1550

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

1551

"ApprovalPending": False,

1552

"ApprovedByDefault": False,

1553

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

1554

"ApprovalDelay": 30000,

1555

"ApprovalDuration": 93785000,

1556

"Checker": ":",

1557

"ExtendedTimeout": 900000,

1558

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

1559

"LastCheckerStatus": -2,

1560

},

1561

}

1562

1563

def test_DumpJSON_normal(self):

1564

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

1565

json_data = json.loads(output)

1566

self.assertDictEqual(json_data, self.expected_json)

1567

1568

def test_DumpJSON_one_client(self):

1569

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

1570

json_data = json.loads(output)

1571

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

1572

self.assertDictEqual(json_data, expected_json)

1573

1574

def test_PrintTable_normal(self):

1575

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

1576

expected_output = "\n".join((

1577

"Name Enabled Timeout Last Successful Check",

1578

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

1579

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

1580

))

1581

self.assertEqual(output, expected_output)

1582

1583

def test_PrintTable_verbose(self):

1584

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

1585

self.clients.values())

1586

columns = (

1587

(

1588

"Name ",

1589

"foo ",

1590

"barbar ",

1591

),(

1592

"Enabled ",

1593

"Yes ",

1594

"Yes ",

1595

),(

1596

"Timeout ",

1597

"00:05:00 ",

1598

"00:05:00 ",

1599

),(

1600

"Last Successful Check ",

1601

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

1602

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

1603

),(

1604

"Created ",

1605

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

1606

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

1607

),(

1608

"Interval ",

1609

"00:02:00 ",

1610

"00:02:00 ",

1611

),(

1612

"Host ",

1613

"foo.example.org ",

1614

"192.0.2.3 ",

1615

),(

1616

("Key ID "

1617

" "),

1618

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1619

"ea90a27cddb12 "),

1620

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1621

"46c2b020f441c "),

1622

),(

1623

"Fingerprint ",

1624

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1625

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1626

),(

1627

"Check Is Running ",

1628

"No ",

1629

"Yes ",

1630

),(

1631

"Last Enabled ",

1632

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

1633

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

1634

),(

1635

"Approval Is Pending ",

1636

"No ",

1637

"No ",

1638

),(

1639

"Approved By Default ",

1640

"Yes ",

1641

"No ",

1642

),(

1643

"Last Approval Request ",

1644

" ",

1645

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

1646

),(

1647

"Approval Delay ",

1648

"00:00:00 ",

1649

"00:00:30 ",

1650

),(

1651

"Approval Duration ",

1652

"00:00:01 ",

1653

"1T02:03:05 ",

1654

),(

1655

"Checker ",

1656

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

1657

": ",

1658

),(

1659

"Extended Timeout ",

1660

"00:15:00 ",

1661

"00:15:00 ",

1662

),(

1663

"Expires ",

1664

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

1665

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

1666

),(

1667

"Last Checker Status",

1668

"0 ",

1669

"-2 ",

1670

)

1671

)

1672

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

1673

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

1674

for rows in columns)

1675

for line in range(num_lines))

1676

self.assertEqual(output, expected_output)

1677

1678

def test_PrintTable_one_client(self):

1679

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

1680

expected_output = "\n".join((

1681

"Name Enabled Timeout Last Successful Check",

1682

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

1683

))

1684

self.assertEqual(output, expected_output)

1685

1686

1687

class TestPropertyCmd(TestCommand):

1688

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

1689

def runTest(self):

1690

if not hasattr(self, "command"):

1691

return

1692

values_to_get = getattr(self, "values_to_get",

1693

self.values_to_set)

1694

for value_to_set, value_to_get in zip(self.values_to_set,

1695

values_to_get):

1696

for clientpath in self.clients:

1697

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

1698

old_value = client.attributes[self.propname]

1699

self.assertNotIsInstance(old_value, self.Unique)

1700

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

1701

self.run_command(value_to_set, self.clients)

1702

for clientpath in self.clients:

1703

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

1704

value = client.attributes[self.propname]

1705

self.assertNotIsInstance(value, self.Unique)

1706

self.assertEqual(value, value_to_get)

1707

1708

class Unique(object):

1709

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

1710

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

1711

1712

def run_command(self, value, clients):

1713

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

1714

1715

1716

class TestEnableCmd(TestPropertyCmd):

1717

command = command.Enable

1718

propname = "Enabled"

1719

values_to_set = [dbus.Boolean(True)]

1720

1721

1722

class TestDisableCmd(TestPropertyCmd):

1723

command = command.Disable

1724

propname = "Enabled"

1725

values_to_set = [dbus.Boolean(False)]

1726

1727

1728

class TestBumpTimeoutCmd(TestPropertyCmd):

1729

command = command.BumpTimeout

1730

propname = "LastCheckedOK"

1731

values_to_set = [""]

1732

1733

1734

class TestStartCheckerCmd(TestPropertyCmd):

1735

command = command.StartChecker

1736

propname = "CheckerRunning"

1737

values_to_set = [dbus.Boolean(True)]

1738

1739

1740

class TestStopCheckerCmd(TestPropertyCmd):

1741

command = command.StopChecker

1742

propname = "CheckerRunning"

1743

values_to_set = [dbus.Boolean(False)]

1744

1745

1746

class TestApproveByDefaultCmd(TestPropertyCmd):

1747

command = command.ApproveByDefault

1748

propname = "ApprovedByDefault"

1749

values_to_set = [dbus.Boolean(True)]

1750

1751

1752

class TestDenyByDefaultCmd(TestPropertyCmd):

1753

command = command.DenyByDefault

1754

propname = "ApprovedByDefault"

1755

values_to_set = [dbus.Boolean(False)]

1756

1757

1758

class TestPropertyValueCmd(TestPropertyCmd):

1759

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

1760

1761

def runTest(self):

1762

if type(self) is TestPropertyValueCmd:

1763

return

1764

return super(TestPropertyValueCmd, self).runTest()

1765

1766

def run_command(self, value, clients):

1767

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

1768

1769

1770

class TestSetCheckerCmd(TestPropertyValueCmd):

1771

command = command.SetChecker

1772

propname = "Checker"

1773

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

1774

1775

1776

class TestSetHostCmd(TestPropertyValueCmd):

1777

command = command.SetHost

1778

propname = "Host"

1779

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

1780

1781

1782

class TestSetSecretCmd(TestPropertyValueCmd):

1783

command = command.SetSecret

1784

propname = "Secret"

1785

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

1786

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

1787

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

1788

1789

1790

class TestSetTimeoutCmd(TestPropertyValueCmd):

1791

command = command.SetTimeout

1792

propname = "Timeout"

1793

values_to_set = [datetime.timedelta(),

1794

datetime.timedelta(minutes=5),

1795

datetime.timedelta(seconds=1),

1796

datetime.timedelta(weeks=1),

1797

datetime.timedelta(weeks=52)]

1798

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

1799

1800

1801

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1802

command = command.SetExtendedTimeout

1803

propname = "ExtendedTimeout"

1804

values_to_set = [datetime.timedelta(),

1805

datetime.timedelta(minutes=5),

1806

datetime.timedelta(seconds=1),

1807

datetime.timedelta(weeks=1),

1808

datetime.timedelta(weeks=52)]

1809

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

1810

1811

1812

class TestSetIntervalCmd(TestPropertyValueCmd):

1813

command = command.SetInterval

1814

propname = "Interval"

1815

values_to_set = [datetime.timedelta(),

1816

datetime.timedelta(minutes=5),

1817

datetime.timedelta(seconds=1),

1818

datetime.timedelta(weeks=1),

1819

datetime.timedelta(weeks=52)]

1820

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

1821

1822

1823

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1824

command = command.SetApprovalDelay

1825

propname = "ApprovalDelay"

1826

values_to_set = [datetime.timedelta(),

1827

datetime.timedelta(minutes=5),

1828

datetime.timedelta(seconds=1),

1829

datetime.timedelta(weeks=1),

1830

datetime.timedelta(weeks=52)]

1831

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

1832

1833

1834

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1835

command = command.SetApprovalDuration

1836

propname = "ApprovalDuration"

1837

values_to_set = [datetime.timedelta(),

1838

datetime.timedelta(minutes=5),

1839

datetime.timedelta(seconds=1),

1840

datetime.timedelta(weeks=1),

1841

datetime.timedelta(weeks=52)]

1842

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

1843

1844

1845

1846

def should_only_run_tests():

1847

parser = argparse.ArgumentParser(add_help=False)

1848

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

1849

args, unknown_args = parser.parse_known_args()

1850

run_tests = args.check

1851

if run_tests:

1852

# Remove --check argument from sys.argv

1853

sys.argv[1:] = unknown_args

1854

return run_tests

1855

1856

# Add all tests from doctest strings

1857

def load_tests(loader, tests, none):

1858

import doctest

1859

tests.addTests(doctest.DocTestSuite())

1860

return tests

1861

1862

if __name__ == "__main__":

1863

try:

1864

if should_only_run_tests():

1865

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

1866

unittest.main()

1867

else:

1868

main()

1869

finally:

1870

logging.shutdown()