/mandos/trunk : revision 1139

1

#!/usr/bin/python

2

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

3

4

from __future__ import division

2

# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This file is part of Mandos.

10

#

11

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

12

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

13

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

14

# (at your option) any later version.

15

#

16

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

17

# WITHOUT ANY WARRANTY; without even the implied warranty of

18

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19

# GNU General Public License for more details.

20

#

21

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

22

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

23

#

24

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

25

#

26

27

from __future__ import (division, absolute_import, print_function,

28

unicode_literals)

29

30

try:

31

from future_builtins import *

32

except ImportError:

33

pass

34

5

35

import sys

6

import dbus

7

from optparse import OptionParser

36

import argparse

8

37

import locale

9

38

import datetime

10

39

import re

11

40

import os

12

13

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

14

15

tablewords = {

16

'Name': u'Name',

17

'Enabled': u'Enabled',

18

'Timeout': u'Timeout',

19

'LastCheckedOK': u'Last Successful Check',

20

'Created': u'Created',

21

'Interval': u'Interval',

22

'Host': u'Host',

23

'Fingerprint': u'Fingerprint',

24

'CheckerRunning': u'Check Is Running',

25

'LastEnabled': u'Last Enabled',

26

'Checker': u'Checker',

27

}

28

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

29

domain = 'se.bsnet.fukt'

30

busname = domain + '.Mandos'

31

server_path = '/'

32

server_interface = domain + '.Mandos'

33

client_interface = domain + '.Mandos.Client'

34

version = "1.0.14"

35

36

def timedelta_to_milliseconds(td):

37

"Convert a datetime.timedelta object to milliseconds"

38

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

39

+ (td.seconds * 1000)

40

+ (td.microseconds // 1000))

41

42

def milliseconds_to_string(ms):

43

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

44

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

45

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

46

"hours": td.seconds // 3600,

47

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

48

"seconds": td.seconds % 60,

49

})

41

import collections

42

import json

43

import unittest

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

49

try:

50

import pydbus

51

import gi

52

dbus_python = None

53

except ImportError:

54

import dbus as dbus_python

55

pydbus = None

56

class gi(object):

57

"""Dummy gi module, for the tests"""

58

class repository(object):

59

class GLib(object):

60

class Error(Exception):

61

pass

62

63

# Show warnings by default

64

if not sys.warnoptions:

65

import warnings

66

warnings.simplefilter("default")

67

68

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

69

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

70

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

71

72

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

73

74

if sys.version_info.major == 2:

75

str = unicode

76

import StringIO

77

io.StringIO = StringIO.StringIO

78

79

locale.setlocale(locale.LC_ALL, "")

80

81

version = "1.8.5"

82

83

84

def main():

85

parser = argparse.ArgumentParser()

86

add_command_line_options(parser)

87

88

options = parser.parse_args()

89

check_option_syntax(parser, options)

90

91

clientnames = options.client

92

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

if pydbus is not None:

97

bus = pydbus_adapter.CachingBus(pydbus)

98

else:

99

bus = dbus_python_adapter.CachingBus(dbus_python)

100

101

try:

102

all_clients = bus.get_clients_and_properties()

103

except dbus.ConnectFailed as e:

104

log.critical("Could not connect to Mandos server: %s", e)

105

sys.exit(1)

106

except dbus.Error as e:

107

log.critical(

108

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

109

sys.exit(1)

110

111

# Compile dict of (clientpath: properties) to process

112

if not clientnames:

113

clients = all_clients

114

else:

115

clients = {}

116

for name in clientnames:

117

for objpath, properties in all_clients.items():

118

if properties["Name"] == name:

119

clients[objpath] = properties

120

break

121

else:

122

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

123

sys.exit(1)

124

125

commands = commands_from_options(options)

126

127

for command in commands:

128

command.run(clients, bus)

129

130

131

def add_command_line_options(parser):

132

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

133

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

134

help="show version number and exit")

135

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

136

help="Select all clients")

137

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

138

help="Print all fields")

139

parser.add_argument("-j", "--dump-json", dest="commands",

140

action="append_const", default=[],

141

const=command.DumpJSON(),

142

help="Dump client data in JSON format")

143

enable_disable = parser.add_mutually_exclusive_group()

144

enable_disable.add_argument("-e", "--enable", dest="commands",

145

action="append_const", default=[],

146

const=command.Enable(),

147

help="Enable client")

148

enable_disable.add_argument("-d", "--disable", dest="commands",

149

action="append_const", default=[],

150

const=command.Disable(),

151

help="disable client")

152

parser.add_argument("-b", "--bump-timeout", dest="commands",

153

action="append_const", default=[],

154

const=command.BumpTimeout(),

155

help="Bump timeout for client")

156

start_stop_checker = parser.add_mutually_exclusive_group()

157

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

158

dest="commands",

159

action="append_const", default=[],

160

const=command.StartChecker(),

161

help="Start checker for client")

162

start_stop_checker.add_argument("--stop-checker", dest="commands",

163

action="append_const", default=[],

164

const=command.StopChecker(),

165

help="Stop checker for client")

166

parser.add_argument("-V", "--is-enabled", dest="commands",

167

action="append_const", default=[],

168

const=command.IsEnabled(),

169

help="Check if client is enabled")

170

parser.add_argument("-r", "--remove", dest="commands",

171

action="append_const", default=[],

172

const=command.Remove(),

173

help="Remove client")

174

parser.add_argument("-c", "--checker", dest="commands",

175

action="append", default=[],

176

metavar="COMMAND", type=command.SetChecker,

177

help="Set checker command for client")

178

parser.add_argument(

179

"-t", "--timeout", dest="commands", action="append",

180

default=[], metavar="TIME",

181

type=command.SetTimeout.argparse(string_to_delta),

182

help="Set timeout for client")

183

parser.add_argument(

184

"--extended-timeout", dest="commands", action="append",

185

default=[], metavar="TIME",

186

type=command.SetExtendedTimeout.argparse(string_to_delta),

187

help="Set extended timeout for client")

188

parser.add_argument(

189

"-i", "--interval", dest="commands", action="append",

190

default=[], metavar="TIME",

191

type=command.SetInterval.argparse(string_to_delta),

192

help="Set checker interval for client")

193

approve_deny_default = parser.add_mutually_exclusive_group()

194

approve_deny_default.add_argument(

195

"--approve-by-default", dest="commands",

196

action="append_const", default=[],

197

const=command.ApproveByDefault(),

198

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

199

approve_deny_default.add_argument(

200

"--deny-by-default", dest="commands",

201

action="append_const", default=[],

202

const=command.DenyByDefault(),

203

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

204

parser.add_argument(

205

"--approval-delay", dest="commands", action="append",

206

default=[], metavar="TIME",

207

type=command.SetApprovalDelay.argparse(string_to_delta),

208

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

209

parser.add_argument(

210

"--approval-duration", dest="commands", action="append",

211

default=[], metavar="TIME",

212

type=command.SetApprovalDuration.argparse(string_to_delta),

213

help="Set duration of one client approval")

214

parser.add_argument("-H", "--host", dest="commands",

215

action="append", default=[], metavar="STRING",

216

type=command.SetHost,

217

help="Set host for client")

218

parser.add_argument(

219

"-s", "--secret", dest="commands", action="append",

220

default=[], metavar="FILENAME",

221

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

222

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

223

approve_deny = parser.add_mutually_exclusive_group()

224

approve_deny.add_argument(

225

"-A", "--approve", dest="commands", action="append_const",

226

default=[], const=command.Approve(),

227

help="Approve any current client request")

228

approve_deny.add_argument("-D", "--deny", dest="commands",

229

action="append_const", default=[],

230

const=command.Deny(),

231

help="Deny any current client request")

232

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

233

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

234

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

235

help="Run self-test")

236

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

50

237

51

238

52

239

def string_to_delta(interval):

53

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

54

55

>>> string_to_delta('7d')

56

datetime.timedelta(7)

57

>>> string_to_delta('60s')

58

datetime.timedelta(0, 60)

59

>>> string_to_delta('60m')

60

datetime.timedelta(0, 3600)

61

>>> string_to_delta('24h')

62

datetime.timedelta(1)

63

>>> string_to_delta(u'1w')

64

datetime.timedelta(7)

65

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

66

datetime.timedelta(0, 330)

240

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

241

242

try:

243

return rfc3339_duration_to_delta(interval)

244

except ValueError as e:

245

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

246

' '.join(e.args))

247

return parse_pre_1_6_1_interval(interval)

248

249

250

def rfc3339_duration_to_delta(duration):

251

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

252

253

>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

254

True

255

>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

256

True

257

>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

258

True

259

>>> # 60 months

260

>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

261

True

262

>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

263

True

264

>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

265

True

266

>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

267

True

268

>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

269

True

270

>>> # Can not be empty:

271

>>> rfc3339_duration_to_delta("")

272

Traceback (most recent call last):

273

...

274

ValueError: Invalid RFC 3339 duration: ""

275

>>> # Must start with "P":

276

>>> rfc3339_duration_to_delta("1D")

277

Traceback (most recent call last):

278

...

279

ValueError: Invalid RFC 3339 duration: "1D"

280

>>> # Must use correct order

281

>>> rfc3339_duration_to_delta("PT1S2M")

282

Traceback (most recent call last):

283

...

284

ValueError: Invalid RFC 3339 duration: "PT1S2M"

285

>>> # Time needs time marker

286

>>> rfc3339_duration_to_delta("P1H2S")

287

Traceback (most recent call last):

288

...

289

ValueError: Invalid RFC 3339 duration: "P1H2S"

290

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

291

>>> rfc3339_duration_to_delta("P1D2W")

292

Traceback (most recent call last):

293

...

294

ValueError: Invalid RFC 3339 duration: "P1D2W"

295

>>> rfc3339_duration_to_delta("P2W2H")

296

Traceback (most recent call last):

297

...

298

ValueError: Invalid RFC 3339 duration: "P2W2H"

67

299

"""

68

timevalue = datetime.timedelta(0)

69

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

70

71

for s in regexp.findall(interval):

72

try:

73

suffix = unicode(s[-1])

74

value = int(s[:-1])

75

if suffix == u"d":

76

delta = datetime.timedelta(value)

77

elif suffix == u"s":

78

delta = datetime.timedelta(0, value)

79

elif suffix == u"m":

80

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

81

elif suffix == u"h":

82

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

83

elif suffix == u"w":

84

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

85

else:

86

raise ValueError

87

except (ValueError, IndexError):

88

raise ValueError

89

timevalue += delta

90

return timevalue

91

92

def print_clients(clients, keywords):

93

def valuetostring(value, keyword):

94

if type(value) is dbus.Boolean:

95

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

96

if keyword in (u"Timeout", u"Interval"):

97

return milliseconds_to_string(value)

98

return unicode(value)

99

100

# Create format string to print table rows

101

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

102

max(len(tablewords[key]),

103

max(len(valuetostring(client[key],

104

key))

105

for client in

106

clients))

107

for key in keywords)

108

# Print header line

109

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

110

for client in clients:

111

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

112

for key in keywords)

113

114

def has_actions(options):

115

return any((options.enable,

116

options.disable,

117

options.bump_timeout,

118

options.start_checker,

119

options.stop_checker,

120

options.is_enabled,

121

options.remove,

122

options.checker is not None,

123

options.timeout is not None,

124

options.interval is not None,

125

options.host is not None,

126

options.secret is not None,

127

options.approve,

128

options.deny))

129

130

def main():

131

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

132

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

133

help="Select all clients")

134

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

135

help="Print all fields")

136

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

137

help="Enable client")

138

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

139

help="disable client")

140

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

141

help="Bump timeout for client")

142

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

143

help="Start checker for client")

144

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

145

help="Stop checker for client")

146

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

147

help="Check if client is enabled")

148

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

149

help="Remove client")

150

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

151

help="Set checker command for client")

152

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

153

help="Set timeout for client")

154

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

155

help="Set checker interval for client")

156

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

157

help="Set host for client")

158

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

159

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

160

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

161

help="Approve any current client request")

162

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

163

help="Deny any current client request")

164

options, client_names = parser.parse_args()

165

166

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

167

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

168

if options.verbose and has_actions(options):

169

parser.error('Verbose option can only be used alone or with --all.')

170

if options.all and not has_actions(options):

171

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

172

173

try:

174

bus = dbus.SystemBus()

175

mandos_dbus_objc = bus.get_object(busname, server_path)

176

except dbus.exceptions.DBusException:

177

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

178

sys.exit(1)

179

180

mandos_serv = dbus.Interface(mandos_dbus_objc,

181

dbus_interface = server_interface)

182

183

#block stderr since dbus library prints to stderr

184

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

185

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

186

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

187

os.close(null)

188

try:

189

try:

190

mandos_clients = mandos_serv.GetAllClientsWithProperties()

191

finally:

192

#restore stderr

193

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

194

os.close(stderrcopy)

195

except dbus.exceptions.DBusException, e:

196

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

197

sys.exit(1)

198

199

# Compile dict of (clients: properties) to process

200

clients={}

201

202

if options.all or not client_names:

203

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

204

for path, properties in

205

mandos_clients.iteritems())

300

301

# Parsing an RFC 3339 duration with regular expressions is not

302

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

303

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

304

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

305

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

306

# avoid excessive use of external libraries.

307

308

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

309

Token = collections.namedtuple("Token", (

310

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

311

# a "group" containing digits

312

"value", # datetime.timedelta or None

313

"followers")) # Tokens valid after this token

314

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

315

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

316

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

317

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

318

datetime.timedelta(seconds=1),

319

frozenset((token_end, )))

320

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

321

datetime.timedelta(minutes=1),

322

frozenset((token_second, token_end)))

323

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

324

datetime.timedelta(hours=1),

325

frozenset((token_minute, token_end)))

326

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

327

None,

328

frozenset((token_hour, token_minute,

329

token_second)))

330

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

331

datetime.timedelta(days=1),

332

frozenset((token_time, token_end)))

333

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

334

datetime.timedelta(weeks=4),

335

frozenset((token_day, token_end)))

336

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

337

datetime.timedelta(weeks=52),

338

frozenset((token_month, token_end)))

339

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

340

datetime.timedelta(weeks=1),

341

frozenset((token_end, )))

342

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

343

frozenset((token_year, token_month,

344

token_day, token_time,

345

token_week)))

346

# Define starting values:

347

# Value so far

348

value = datetime.timedelta()

349

found_token = None

350

# Following valid tokens

351

followers = frozenset((token_duration, ))

352

# String left to parse

353

s = duration

354

# Loop until end token is found

355

while found_token is not token_end:

356

# Search for any currently valid tokens

357

for token in followers:

358

match = token.regexp.match(s)

359

if match is not None:

360

# Token found

361

if token.value is not None:

362

# Value found, parse digits

363

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

364

# Add to value so far

365

value += factor * token.value

366

# Strip token from string

367

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

368

# Go to found token

369

found_token = token

370

# Set valid next tokens

371

followers = found_token.followers

372

break

206

373

else:

207

for name in client_names:

208

for path, client in mandos_clients.iteritems():

209

if client['Name'] == name:

210

client_objc = bus.get_object(busname, path)

211

clients[client_objc] = client

212

break

374

# No currently valid tokens were found

375

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

376

.format(duration))

377

# End token found

378

return value

379

380

381

def parse_pre_1_6_1_interval(interval):

382

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

383

and return a datetime.timedelta

384

385

>>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)

386

True

387

>>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)

388

True

389

>>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)

390

True

391

>>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)

392

True

393

>>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)

394

True

395

>>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)

396

True

397

>>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)

398

True

399

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

400

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)

401

True

402

403

"""

404

405

value = datetime.timedelta(0)

406

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

407

408

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

409

if suffix == "d":

410

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

411

elif suffix == "s":

412

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

413

elif suffix == "m":

414

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

415

elif suffix == "h":

416

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

417

elif suffix == "w":

418

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

419

elif suffix == "":

420

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

421

return value

422

423

424

def check_option_syntax(parser, options):

425

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

426

argparse"""

427

428

def has_commands(options, commands=None):

429

if commands is None:

430

commands = (command.Enable,

431

command.Disable,

432

command.BumpTimeout,

433

command.StartChecker,

434

command.StopChecker,

435

command.IsEnabled,

436

command.Remove,

437

command.SetChecker,

438

command.SetTimeout,

439

command.SetExtendedTimeout,

440

command.SetInterval,

441

command.ApproveByDefault,

442

command.DenyByDefault,

443

command.SetApprovalDelay,

444

command.SetApprovalDuration,

445

command.SetHost,

446

command.SetSecret,

447

command.Approve,

448

command.Deny)

449

return any(isinstance(cmd, commands)

450

for cmd in options.commands)

451

452

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

453

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

454

if options.verbose and has_commands(options):

455

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

456

if (has_commands(options, (command.DumpJSON,))

457

and (options.verbose or len(options.commands) > 1)):

458

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

459

if options.all and not has_commands(options):

460

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

461

if (has_commands(options, (command.IsEnabled,))

462

and len(options.client) > 1):

463

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

464

if (len(options.commands) > 1

465

and has_commands(options, (command.Remove,))

466

and not has_commands(options, (command.Deny,))):

467

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

468

469

470

class dbus(object):

471

472

class SystemBus(object):

473

474

object_manager_iface = "org.freedesktop.DBus.ObjectManager"

475

def get_managed_objects(self, busname, objectpath):

476

return self.call_method("GetManagedObjects", busname,

477

objectpath,

478

self.object_manager_iface)

479

480

properties_iface = "org.freedesktop.DBus.Properties"

481

def set_property(self, busname, objectpath, interface, key,

482

value):

483

self.call_method("Set", busname, objectpath,

484

self.properties_iface, interface, key,

485

value)

486

487

488

class MandosBus(SystemBus):

489

busname_domain = "se.recompile"

490

busname = busname_domain + ".Mandos"

491

server_path = "/"

492

server_interface = busname_domain + ".Mandos"

493

client_interface = busname_domain + ".Mandos.Client"

494

del busname_domain

495

496

def get_clients_and_properties(self):

497

managed_objects = self.get_managed_objects(

498

self.busname, self.server_path)

499

return {objpath: properties[self.client_interface]

500

for objpath, properties in managed_objects.items()

501

if self.client_interface in properties}

502

503

def set_client_property(self, objectpath, key, value):

504

return self.set_property(self.busname, objectpath,

505

self.client_interface, key,

506

value)

507

508

def call_client_method(self, objectpath, method, *args):

509

return self.call_method(method, self.busname, objectpath,

510

self.client_interface, *args)

511

512

def call_server_method(self, method, *args):

513

return self.call_method(method, self.busname,

514

self.server_path,

515

self.server_interface, *args)

516

517

class Error(Exception):

518

pass

519

520

class ConnectFailed(Error):

521

pass

522

523

524

class dbus_python_adapter(object):

525

526

class SystemBus(dbus.MandosBus):

527

"""Use dbus-python"""

528

529

def __init__(self, module=dbus_python):

530

self.dbus_python = module

531

self.bus = self.dbus_python.SystemBus()

532

533

@contextlib.contextmanager

534

def convert_exception(self, exception_class=dbus.Error):

535

try:

536

yield

537

except self.dbus_python.exceptions.DBusException as e:

538

# This does what "raise from" would do

539

exc = exception_class(*e.args)

540

exc.__cause__ = e

541

raise exc

542

543

def call_method(self, methodname, busname, objectpath,

544

interface, *args):

545

proxy_object = self.get_object(busname, objectpath)

546

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

547

interface, methodname,

548

", ".join(repr(a) for a in args))

549

method = getattr(proxy_object, methodname)

550

with self.convert_exception():

551

with dbus_python_adapter.SilenceLogger(

552

"dbus.proxies"):

553

value = method(*args, dbus_interface=interface)

554

return self.type_filter(value)

555

556

def get_object(self, busname, objectpath):

557

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

558

busname, objectpath)

559

with self.convert_exception(dbus.ConnectFailed):

560

return self.bus.get_object(busname, objectpath)

561

562

def type_filter(self, value):

563

"""Convert the most bothersome types to Python types"""

564

if isinstance(value, self.dbus_python.Boolean):

565

return bool(value)

566

if isinstance(value, self.dbus_python.ObjectPath):

567

return str(value)

568

# Also recurse into dictionaries

569

if isinstance(value, self.dbus_python.Dictionary):

570

return {self.type_filter(key):

571

self.type_filter(subval)

572

for key, subval in value.items()}

573

return value

574

575

def set_client_property(self, objectpath, key, value):

576

if key == "Secret":

577

if not isinstance(value, bytes):

578

value = value.encode("utf-8")

579

value = self.dbus_python.ByteArray(value)

580

return self.set_property(self.busname, objectpath,

581

self.client_interface, key,

582

value)

583

584

class SilenceLogger(object):

585

"Simple context manager to silence a particular logger"

586

def __init__(self, loggername):

587

self.logger = logging.getLogger(loggername)

588

589

def __enter__(self):

590

self.logger.addFilter(self.nullfilter)

591

592

class NullFilter(logging.Filter):

593

def filter(self, record):

594

return False

595

596

nullfilter = NullFilter()

597

598

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

599

self.logger.removeFilter(self.nullfilter)

600

601

602

class CachingBus(SystemBus):

603

"""A caching layer for dbus_python_adapter.SystemBus"""

604

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

605

self.object_cache = {}

606

super(dbus_python_adapter.CachingBus,

607

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

608

def get_object(self, busname, objectpath):

609

try:

610

return self.object_cache[(busname, objectpath)]

611

except KeyError:

612

new_object = super(

613

dbus_python_adapter.CachingBus,

614

self).get_object(busname, objectpath)

615

self.object_cache[(busname, objectpath)] = new_object

616

return new_object

617

618

619

class pydbus_adapter(object):

620

class SystemBus(dbus.MandosBus):

621

def __init__(self, module=pydbus):

622

self.pydbus = module

623

self.bus = self.pydbus.SystemBus()

624

625

@contextlib.contextmanager

626

def convert_exception(self, exception_class=dbus.Error):

627

try:

628

yield

629

except gi.repository.GLib.Error as e:

630

# This does what "raise from" would do

631

exc = exception_class(*e.args)

632

exc.__cause__ = e

633

raise exc

634

635

def call_method(self, methodname, busname, objectpath,

636

interface, *args):

637

proxy_object = self.get(busname, objectpath)

638

log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

639

interface, methodname,

640

", ".join(repr(a) for a in args))

641

method = getattr(proxy_object[interface], methodname)

642

with self.convert_exception():

643

return method(*args)

644

645

def get(self, busname, objectpath):

646

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

647

busname, objectpath)

648

with self.convert_exception(dbus.ConnectFailed):

649

if sys.version_info.major <= 2:

650

with warnings.catch_warnings():

651

warnings.filterwarnings(

652

"ignore", "", DeprecationWarning,

653

r"^xml\.etree\.ElementTree$")

654

return self.bus.get(busname, objectpath)

213

655

else:

214

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

215

sys.exit(1)

216

217

if not has_actions(options) and clients:

218

if options.verbose:

219

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

220

'Created', 'Interval', 'Host', 'Fingerprint',

221

'CheckerRunning', 'LastEnabled', 'Checker')

222

else:

223

keywords = defaultkeywords

224

225

print_clients(clients.values(), keywords)

226

else:

227

# Process each client in the list by all selected options

228

for client in clients:

229

if options.remove:

230

mandos_serv.RemoveClient(client.__dbus_object_path__)

231

if options.enable:

232

client.Enable(dbus_interface=client_interface)

233

if options.disable:

234

client.Disable(dbus_interface=client_interface)

235

if options.bump_timeout:

236

client.CheckedOK(dbus_interface=client_interface)

237

if options.start_checker:

238

client.StartChecker(dbus_interface=client_interface)

239

if options.stop_checker:

240

client.StopChecker(dbus_interface=client_interface)

241

if options.is_enabled:

242

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

243

u"Enabled",

244

dbus_interface=dbus.PROPERTIES_IFACE)

245

else 1)

246

if options.checker:

247

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

248

dbus_interface=dbus.PROPERTIES_IFACE)

249

if options.host:

250

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

251

dbus_interface=dbus.PROPERTIES_IFACE)

252

if options.interval:

253

client.Set(client_interface, u"Interval",

254

timedelta_to_milliseconds

255

(string_to_delta(options.interval)),

256

dbus_interface=dbus.PROPERTIES_IFACE)

257

if options.timeout:

258

client.Set(client_interface, u"Timeout",

259

timedelta_to_milliseconds(string_to_delta

260

(options.timeout)),

261

dbus_interface=dbus.PROPERTIES_IFACE)

262

if options.secret:

263

client.Set(client_interface, u"Secret",

264

dbus.ByteArray(open(options.secret, u'rb').read()),

265

dbus_interface=dbus.PROPERTIES_IFACE)

266

if options.approve:

267

client.Approve(dbus.Boolean(True), dbus_interface=client_interface)

268

if options.deny:

269

client.Approve(dbus.Boolean(False), dbus_interface=client_interface)

270

271

if __name__ == '__main__':

272

main()

656

return self.bus.get(busname, objectpath)

657

658

def set_property(self, busname, objectpath, interface, key,

659

value):

660

proxy_object = self.get(busname, objectpath)

661

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

662

objectpath, self.properties_iface, interface,

663

key, value)

664

setattr(proxy_object[interface], key, value)

665

666

class CachingBus(SystemBus):

667

"""A caching layer for pydbus_adapter.SystemBus"""

668

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

669

self.object_cache = {}

670

super(pydbus_adapter.CachingBus,

671

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

672

def get(self, busname, objectpath):

673

try:

674

return self.object_cache[(busname, objectpath)]

675

except KeyError:

676

new_object = (super(pydbus_adapter.CachingBus, self)

677

.get(busname, objectpath))

678

self.object_cache[(busname, objectpath)] = new_object

679

return new_object

680

681

682

def commands_from_options(options):

683

684

commands = list(options.commands)

685

686

def find_cmd(cmd, commands):

687

i = 0

688

for i, c in enumerate(commands):

689

if isinstance(c, cmd):

690

return i

691

return i+1

692

693

# If command.Remove is present, move any instances of command.Deny

694

# to occur ahead of command.Remove.

695

index_of_remove = find_cmd(command.Remove, commands)

696

before_remove = commands[:index_of_remove]

697

after_remove = commands[index_of_remove:]

698

cleaned_after = []

699

for cmd in after_remove:

700

if isinstance(cmd, command.Deny):

701

before_remove.append(cmd)

702

else:

703

cleaned_after.append(cmd)

704

if cleaned_after != after_remove:

705

commands = before_remove + cleaned_after

706

707

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

708

# optionally verbosely

709

if not commands:

710

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

711

712

return commands

713

714

715

class command(object):

716

"""A namespace for command classes"""

717

718

class Base(object):

719

"""Abstract base class for commands"""

720

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

721

"""Normal commands should implement run_on_one_client(),

722

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

723

override this run() method instead.

724

"""

725

self.bus = bus

726

for client, properties in clients.items():

727

self.run_on_one_client(client, properties)

728

729

730

class IsEnabled(Base):

731

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

732

properties = next(iter(clients.values()))

733

if properties["Enabled"]:

734

sys.exit(0)

735

sys.exit(1)

736

737

738

class Approve(Base):

739

def run_on_one_client(self, client, properties):

740

self.bus.call_client_method(client, "Approve", True)

741

742

743

class Deny(Base):

744

def run_on_one_client(self, client, properties):

745

self.bus.call_client_method(client, "Approve", False)

746

747

748

class Remove(Base):

749

def run(self, clients, bus):

750

for clientpath in frozenset(clients.keys()):

751

bus.call_server_method("RemoveClient", clientpath)

752

753

754

class Output(Base):

755

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

756

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

757

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

758

"Fingerprint", "CheckerRunning",

759

"LastEnabled", "ApprovalPending",

760

"ApprovedByDefault", "LastApprovalRequest",

761

"ApprovalDelay", "ApprovalDuration",

762

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

763

"LastCheckerStatus")

764

765

766

class DumpJSON(Output):

767

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

768

data = {properties["Name"]:

769

{key: properties[key]

770

for key in self.all_keywords}

771

for properties in clients.values()}

772

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

773

774

775

class PrintTable(Output):

776

def __init__(self, verbose=False):

777

self.verbose = verbose

778

779

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

780

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

781

"LastCheckedOK")

782

keywords = default_keywords

783

if self.verbose:

784

keywords = self.all_keywords

785

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

786

787

class TableOfClients(object):

788

tableheaders = {

789

"Name": "Name",

790

"Enabled": "Enabled",

791

"Timeout": "Timeout",

792

"LastCheckedOK": "Last Successful Check",

793

"LastApprovalRequest": "Last Approval Request",

794

"Created": "Created",

795

"Interval": "Interval",

796

"Host": "Host",

797

"Fingerprint": "Fingerprint",

798

"KeyID": "Key ID",

799

"CheckerRunning": "Check Is Running",

800

"LastEnabled": "Last Enabled",

801

"ApprovalPending": "Approval Is Pending",

802

"ApprovedByDefault": "Approved By Default",

803

"ApprovalDelay": "Approval Delay",

804

"ApprovalDuration": "Approval Duration",

805

"Checker": "Checker",

806

"ExtendedTimeout": "Extended Timeout",

807

"Expires": "Expires",

808

"LastCheckerStatus": "Last Checker Status",

809

}

810

811

def __init__(self, clients, keywords):

812

self.clients = clients

813

self.keywords = keywords

814

815

def __str__(self):

816

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

817

818

if sys.version_info.major == 2:

819

__unicode__ = __str__

820

def __str__(self):

821

return str(self).encode(

822

locale.getpreferredencoding())

823

824

def rows(self):

825

format_string = self.row_formatting_string()

826

rows = [self.header_line(format_string)]

827

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

828

for client in self.clients)

829

return rows

830

831

def row_formatting_string(self):

832

"Format string used to format table rows"

833

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

834

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

835

*(len(self.string_from_client(client,

836

key))

837

for client in self.clients)),

838

key=key)

839

for key in self.keywords)

840

841

def string_from_client(self, client, key):

842

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

843

844

@classmethod

845

def valuetostring(cls, value, keyword):

846

if isinstance(value, bool):

847

return "Yes" if value else "No"

848

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

849

"ApprovalDuration", "ExtendedTimeout"):

850

return cls.milliseconds_to_string(value)

851

return str(value)

852

853

def header_line(self, format_string):

854

return format_string.format(**self.tableheaders)

855

856

def client_line(self, client, format_string):

857

return format_string.format(

858

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

859

for key in self.keywords})

860

861

@staticmethod

862

def milliseconds_to_string(ms):

863

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

864

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

865

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

866

if td.days else "",

867

hours=td.seconds // 3600,

868

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

869

seconds=td.seconds % 60))

870

871

872

class PropertySetter(Base):

873

"Abstract class for Actions for setting one client property"

874

875

def run_on_one_client(self, client, properties=None):

876

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

877

self.bus.set_client_property(client, self.propname,

878

self.value_to_set)

879

880

@property

881

def propname(self):

882

raise NotImplementedError()

883

884

885

class Enable(PropertySetter):

886

propname = "Enabled"

887

value_to_set = True

888

889

890

class Disable(PropertySetter):

891

propname = "Enabled"

892

value_to_set = False

893

894

895

class BumpTimeout(PropertySetter):

896

propname = "LastCheckedOK"

897

value_to_set = ""

898

899

900

class StartChecker(PropertySetter):

901

propname = "CheckerRunning"

902

value_to_set = True

903

904

905

class StopChecker(PropertySetter):

906

propname = "CheckerRunning"

907

value_to_set = False

908

909

910

class ApproveByDefault(PropertySetter):

911

propname = "ApprovedByDefault"

912

value_to_set = True

913

914

915

class DenyByDefault(PropertySetter):

916

propname = "ApprovedByDefault"

917

value_to_set = False

918

919

920

class PropertySetterValue(PropertySetter):

921

"""Abstract class for PropertySetter recieving a value as

922

constructor argument instead of a class attribute."""

923

def __init__(self, value):

924

self.value_to_set = value

925

926

@classmethod

927

def argparse(cls, argtype):

928

def cmdtype(arg):

929

return cls(argtype(arg))

930

return cmdtype

931

932

class SetChecker(PropertySetterValue):

933

propname = "Checker"

934

935

936

class SetHost(PropertySetterValue):

937

propname = "Host"

938

939

940

class SetSecret(PropertySetterValue):

941

propname = "Secret"

942

943

@property

944

def value_to_set(self):

945

return self._vts

946

947

@value_to_set.setter

948

def value_to_set(self, value):

949

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

950

self._vts = value.read()

951

value.close()

952

953

954

class PropertySetterValueMilliseconds(PropertySetterValue):

955

"""Abstract class for PropertySetterValue taking a value

956

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

957

milliseconds."""

958

959

@property

960

def value_to_set(self):

961

return self._vts

962

963

@value_to_set.setter

964

def value_to_set(self, value):

965

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

966

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

967

968

969

class SetTimeout(PropertySetterValueMilliseconds):

970

propname = "Timeout"

971

972

973

class SetExtendedTimeout(PropertySetterValueMilliseconds):

974

propname = "ExtendedTimeout"

975

976

977

class SetInterval(PropertySetterValueMilliseconds):

978

propname = "Interval"

979

980

981

class SetApprovalDelay(PropertySetterValueMilliseconds):

982

propname = "ApprovalDelay"

983

984

985

class SetApprovalDuration(PropertySetterValueMilliseconds):

986

propname = "ApprovalDuration"

987

988

989

990

class TestCaseWithAssertLogs(unittest.TestCase):

991

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

992

993

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

994

@contextlib.contextmanager

995

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

996

capturing_handler = self.CapturingLevelHandler(level)

997

old_level = logger.level

998

old_propagate = logger.propagate

999

logger.addHandler(capturing_handler)

1000

logger.setLevel(level)

1001

logger.propagate = False

1002

try:

1003

yield capturing_handler.watcher

1004

finally:

1005

logger.propagate = old_propagate

1006

logger.removeHandler(capturing_handler)

1007

logger.setLevel(old_level)

1008

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

1009

0)

1010

1011

class CapturingLevelHandler(logging.Handler):

1012

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

1013

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

1014

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

1015

def emit(self, record):

1016

self.watcher.records.append(record)

1017

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

1018

1019

LoggingWatcher = collections.namedtuple("LoggingWatcher",

1020

("records",

1021

"output"))

1022

1023

1024

class Unique(object):

1025

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

1026

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

1027

1028

1029

class Test_string_to_delta(TestCaseWithAssertLogs):

1030

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

1031

# rfc3339_duration_to_delta() already has more comprehensive

1032

# tests, which are run by doctest.

1033

1034

def test_rfc3339_zero_seconds(self):

1035

self.assertEqual(datetime.timedelta(),

1036

string_to_delta("PT0S"))

1037

1038

def test_rfc3339_zero_days(self):

1039

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

1040

1041

def test_rfc3339_one_second(self):

1042

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

1043

string_to_delta("PT1S"))

1044

1045

def test_rfc3339_two_hours(self):

1046

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

1047

string_to_delta("PT2H"))

1048

1049

def test_falls_back_to_pre_1_6_1_with_warning(self):

1050

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

1051

value = string_to_delta("2h")

1052

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

1053

1054

1055

class Test_check_option_syntax(unittest.TestCase):

1056

def setUp(self):

1057

self.parser = argparse.ArgumentParser()

1058

add_command_line_options(self.parser)

1059

1060

def test_actions_requires_client_or_all(self):

1061

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

1062

args = self.actionargs(action, value)

1063

with self.assertParseError():

1064

self.parse_args(args)

1065

1066

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

1067

# check_option_syntax()

1068

actions = {

1069

"--enable": None,

1070

"--disable": None,

1071

"--bump-timeout": None,

1072

"--start-checker": None,

1073

"--stop-checker": None,

1074

"--is-enabled": None,

1075

"--remove": None,

1076

"--checker": "x",

1077

"--timeout": "PT0S",

1078

"--extended-timeout": "PT0S",

1079

"--interval": "PT0S",

1080

"--approve-by-default": None,

1081

"--deny-by-default": None,

1082

"--approval-delay": "PT0S",

1083

"--approval-duration": "PT0S",

1084

"--host": "hostname",

1085

"--secret": "/dev/null",

1086

"--approve": None,

1087

"--deny": None,

1088

}

1089

1090

@staticmethod

1091

def actionargs(action, value, *args):

1092

if value is not None:

1093

return [action, value] + list(args)

1094

else:

1095

return [action] + list(args)

1096

1097

@contextlib.contextmanager

1098

def assertParseError(self):

1099

with self.assertRaises(SystemExit) as e:

1100

with self.redirect_stderr_to_devnull():

1101

yield

1102

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

1103

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

1104

# /argparse.html#exiting-methods

1105

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

1106

1107

def parse_args(self, args):

1108

options = self.parser.parse_args(args)

1109

check_option_syntax(self.parser, options)

1110

1111

@staticmethod

1112

@contextlib.contextmanager

1113

def redirect_stderr_to_devnull():

1114

old_stderr = sys.stderr

1115

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

1116

sys.stderr = null

1117

try:

1118

yield

1119

finally:

1120

sys.stderr = old_stderr

1121

1122

def check_option_syntax(self, options):

1123

check_option_syntax(self.parser, options)

1124

1125

def test_actions_all_conflicts_with_verbose(self):

1126

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

1127

args = self.actionargs(action, value, "--all",

1128

"--verbose")

1129

with self.assertParseError():

1130

self.parse_args(args)

1131

1132

def test_actions_with_client_conflicts_with_verbose(self):

1133

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

1134

args = self.actionargs(action, value, "--verbose",

1135

"client")

1136

with self.assertParseError():

1137

self.parse_args(args)

1138

1139

def test_dump_json_conflicts_with_verbose(self):

1140

args = ["--dump-json", "--verbose"]

1141

with self.assertParseError():

1142

self.parse_args(args)

1143

1144

def test_dump_json_conflicts_with_action(self):

1145

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

1146

args = self.actionargs(action, value, "--dump-json")

1147

with self.assertParseError():

1148

self.parse_args(args)

1149

1150

def test_all_can_not_be_alone(self):

1151

args = ["--all"]

1152

with self.assertParseError():

1153

self.parse_args(args)

1154

1155

def test_all_is_ok_with_any_action(self):

1156

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

1157

args = self.actionargs(action, value, "--all")

1158

self.parse_args(args)

1159

1160

def test_any_action_is_ok_with_one_client(self):

1161

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

1162

args = self.actionargs(action, value, "client")

1163

self.parse_args(args)

1164

1165

def test_one_client_with_all_actions_except_is_enabled(self):

1166

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

1167

if action == "--is-enabled":

1168

continue

1169

args = self.actionargs(action, value, "client")

1170

self.parse_args(args)

1171

1172

def test_two_clients_with_all_actions_except_is_enabled(self):

1173

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

1174

if action == "--is-enabled":

1175

continue

1176

args = self.actionargs(action, value, "client1",

1177

"client2")

1178

self.parse_args(args)

1179

1180

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1181

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

1182

if action == "--is-enabled":

1183

continue

1184

args = self.actionargs(action, value, "client1",

1185

"client2")

1186

self.parse_args(args)

1187

1188

def test_is_enabled_fails_without_client(self):

1189

args = ["--is-enabled"]

1190

with self.assertParseError():

1191

self.parse_args(args)

1192

1193

def test_is_enabled_fails_with_two_clients(self):

1194

args = ["--is-enabled", "client1", "client2"]

1195

with self.assertParseError():

1196

self.parse_args(args)

1197

1198

def test_remove_can_only_be_combined_with_action_deny(self):

1199

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

1200

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

1201

continue

1202

args = self.actionargs(action, value, "--all",

1203

"--remove")

1204

with self.assertParseError():

1205

self.parse_args(args)

1206

1207

1208

class Test_dbus_exceptions(unittest.TestCase):

1209

1210

def test_dbus_ConnectFailed_is_Error(self):

1211

with self.assertRaises(dbus.Error):

1212

raise dbus.ConnectFailed()

1213

1214

1215

class Test_dbus_MandosBus(unittest.TestCase):

1216

1217

class MockMandosBus(dbus.MandosBus):

1218

def __init__(self):

1219

self._name = "se.recompile.Mandos"

1220

self._server_path = "/"

1221

self._server_interface = "se.recompile.Mandos"

1222

self._client_interface = "se.recompile.Mandos.Client"

1223

self.calls = []

1224

self.call_method_return = Unique()

1225

1226

def call_method(self, methodname, busname, objectpath,

1227

interface, *args):

1228

self.calls.append((methodname, busname, objectpath,

1229

interface, args))

1230

return self.call_method_return

1231

1232

def setUp(self):

1233

self.bus = self.MockMandosBus()

1234

1235

def test_set_client_property(self):

1236

self.bus.set_client_property("objectpath", "key", "value")

1237

expected_call = ("Set", self.bus._name, "objectpath",

1238

"org.freedesktop.DBus.Properties",

1239

(self.bus._client_interface, "key", "value"))

1240

self.assertIn(expected_call, self.bus.calls)

1241

1242

def test_call_client_method(self):

1243

ret = self.bus.call_client_method("objectpath", "methodname")

1244

self.assertIs(self.bus.call_method_return, ret)

1245

expected_call = ("methodname", self.bus._name, "objectpath",

1246

self.bus._client_interface, ())

1247

self.assertIn(expected_call, self.bus.calls)

1248

1249

def test_call_client_method_with_args(self):

1250

args = (Unique(), Unique())

1251

ret = self.bus.call_client_method("objectpath", "methodname",

1252

*args)

1253

self.assertIs(self.bus.call_method_return, ret)

1254

expected_call = ("methodname", self.bus._name, "objectpath",

1255

self.bus._client_interface,

1256

(args[0], args[1]))

1257

self.assertIn(expected_call, self.bus.calls)

1258

1259

def test_get_clients_and_properties(self):

1260

managed_objects = {

1261

"objectpath": {

1262

self.bus._client_interface: {

1263

"key": "value",

1264

"bool": True,

1265

},

1266

"irrelevant_interface": {

1267

"key": "othervalue",

1268

"bool": False,

1269

},

1270

},

1271

"other_objectpath": {

1272

"other_irrelevant_interface": {

1273

"key": "value 3",

1274

"bool": None,

1275

},

1276

},

1277

}

1278

expected_clients_and_properties = {

1279

"objectpath": {

1280

"key": "value",

1281

"bool": True,

1282

}

1283

}

1284

self.bus.call_method_return = managed_objects

1285

ret = self.bus.get_clients_and_properties()

1286

self.assertDictEqual(expected_clients_and_properties, ret)

1287

expected_call = ("GetManagedObjects", self.bus._name,

1288

self.bus._server_path,

1289

"org.freedesktop.DBus.ObjectManager", ())

1290

self.assertIn(expected_call, self.bus.calls)

1291

1292

def test_call_server_method(self):

1293

ret = self.bus.call_server_method("methodname")

1294

self.assertIs(self.bus.call_method_return, ret)

1295

expected_call = ("methodname", self.bus._name,

1296

self.bus._server_path,

1297

self.bus._server_interface, ())

1298

self.assertIn(expected_call, self.bus.calls)

1299

1300

def test_call_server_method_with_args(self):

1301

args = (Unique(), Unique())

1302

ret = self.bus.call_server_method("methodname", *args)

1303

self.assertIs(self.bus.call_method_return, ret)

1304

expected_call = ("methodname", self.bus._name,

1305

self.bus._server_path,

1306

self.bus._server_interface,

1307

(args[0], args[1]))

1308

self.assertIn(expected_call, self.bus.calls)

1309

1310

1311

class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):

1312

1313

def MockDBusPython_func(self, func):

1314

class mock_dbus_python(object):

1315

"""mock dbus-python module"""

1316

class exceptions(object):

1317

"""Pseudo-namespace"""

1318

class DBusException(Exception):

1319

pass

1320

class SystemBus(object):

1321

@staticmethod

1322

def get_object(busname, objectpath):

1323

DBusObject = collections.namedtuple(

1324

"DBusObject", ("methodname", "Set"))

1325

def method(*args, **kwargs):

1326

self.assertEqual({"dbus_interface":

1327

"interface"},

1328

kwargs)

1329

return func(*args)

1330

def set_property(interface, key, value,

1331

dbus_interface=None):

1332

self.assertEqual(

1333

"org.freedesktop.DBus.Properties",

1334

dbus_interface)

1335

self.assertEqual("Secret", key)

1336

return func(interface, key, value,

1337

dbus_interface=dbus_interface)

1338

return DBusObject(methodname=method,

1339

Set=set_property)

1340

class Boolean(object):

1341

def __init__(self, value):

1342

self.value = bool(value)

1343

def __bool__(self):

1344

return self.value

1345

if sys.version_info.major == 2:

1346

__nonzero__ = __bool__

1347

class ObjectPath(str):

1348

pass

1349

class Dictionary(dict):

1350

pass

1351

class ByteArray(bytes):

1352

pass

1353

return mock_dbus_python

1354

1355

def call_method(self, bus, methodname, busname, objectpath,

1356

interface, *args):

1357

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

1358

return bus.call_method(methodname, busname, objectpath,

1359

interface, *args)

1360

1361

def test_call_method_returns(self):

1362

expected_method_return = Unique()

1363

method_args = (Unique(), Unique())

1364

def func(*args):

1365

self.assertEqual(len(method_args), len(args))

1366

for marg, arg in zip(method_args, args):

1367

self.assertIs(marg, arg)

1368

return expected_method_return

1369

mock_dbus_python = self.MockDBusPython_func(func)

1370

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1371

ret = self.call_method(bus, "methodname", "busname",

1372

"objectpath", "interface",

1373

*method_args)

1374

self.assertIs(ret, expected_method_return)

1375

1376

def test_call_method_filters_bool_true(self):

1377

def func():

1378

return method_return

1379

mock_dbus_python = self.MockDBusPython_func(func)

1380

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1381

method_return = mock_dbus_python.Boolean(True)

1382

ret = self.call_method(bus, "methodname", "busname",

1383

"objectpath", "interface")

1384

self.assertTrue(ret)

1385

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1386

1387

def test_call_method_filters_bool_false(self):

1388

def func():

1389

return method_return

1390

mock_dbus_python = self.MockDBusPython_func(func)

1391

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1392

method_return = mock_dbus_python.Boolean(False)

1393

ret = self.call_method(bus, "methodname", "busname",

1394

"objectpath", "interface")

1395

self.assertFalse(ret)

1396

self.assertNotIsInstance(ret, mock_dbus_python.Boolean)

1397

1398

def test_call_method_filters_objectpath(self):

1399

def func():

1400

return method_return

1401

mock_dbus_python = self.MockDBusPython_func(func)

1402

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1403

method_return = mock_dbus_python.ObjectPath("objectpath")

1404

ret = self.call_method(bus, "methodname", "busname",

1405

"objectpath", "interface")

1406

self.assertEqual("objectpath", ret)

1407

self.assertIsNot("objectpath", ret)

1408

self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)

1409

1410

def test_call_method_filters_booleans_in_dict(self):

1411

def func():

1412

return method_return

1413

mock_dbus_python = self.MockDBusPython_func(func)

1414

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1415

method_return = mock_dbus_python.Dictionary(

1416

{mock_dbus_python.Boolean(True):

1417

mock_dbus_python.Boolean(False),

1418

mock_dbus_python.Boolean(False):

1419

mock_dbus_python.Boolean(True)})

1420

ret = self.call_method(bus, "methodname", "busname",

1421

"objectpath", "interface")

1422

expected_method_return = {True: False,

1423

False: True}

1424

self.assertEqual(expected_method_return, ret)

1425

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1426

1427

def test_call_method_filters_objectpaths_in_dict(self):

1428

def func():

1429

return method_return

1430

mock_dbus_python = self.MockDBusPython_func(func)

1431

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1432

method_return = mock_dbus_python.Dictionary(

1433

{mock_dbus_python.ObjectPath("objectpath_key_1"):

1434

mock_dbus_python.ObjectPath("objectpath_value_1"),

1435

mock_dbus_python.ObjectPath("objectpath_key_2"):

1436

mock_dbus_python.ObjectPath("objectpath_value_2")})

1437

ret = self.call_method(bus, "methodname", "busname",

1438

"objectpath", "interface")

1439

expected_method_return = {str(key): str(value)

1440

for key, value in

1441

method_return.items()}

1442

self.assertEqual(expected_method_return, ret)

1443

self.assertIsInstance(ret, dict)

1444

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1445

1446

def test_call_method_filters_dict_in_dict(self):

1447

def func():

1448

return method_return

1449

mock_dbus_python = self.MockDBusPython_func(func)

1450

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1451

method_return = mock_dbus_python.Dictionary(

1452

{"key1": mock_dbus_python.Dictionary({"key11": "value11",

1453

"key12": "value12"}),

1454

"key2": mock_dbus_python.Dictionary({"key21": "value21",

1455

"key22": "value22"})})

1456

ret = self.call_method(bus, "methodname", "busname",

1457

"objectpath", "interface")

1458

expected_method_return = {

1459

"key1": {"key11": "value11",

1460

"key12": "value12"},

1461

"key2": {"key21": "value21",

1462

"key22": "value22"},

1463

}

1464

self.assertEqual(expected_method_return, ret)

1465

self.assertIsInstance(ret, dict)

1466

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1467

for key, value in ret.items():

1468

self.assertIsInstance(value, dict)

1469

self.assertEqual(expected_method_return[key], value)

1470

self.assertNotIsInstance(value,

1471

mock_dbus_python.Dictionary)

1472

1473

def test_call_method_filters_dict_three_deep(self):

1474

def func():

1475

return method_return

1476

mock_dbus_python = self.MockDBusPython_func(func)

1477

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1478

method_return = mock_dbus_python.Dictionary(

1479

{"key1":

1480

mock_dbus_python.Dictionary(

1481

{"key2":

1482

mock_dbus_python.Dictionary(

1483

{"key3":

1484

mock_dbus_python.Boolean(True),

1485

}),

1486

}),

1487

})

1488

ret = self.call_method(bus, "methodname", "busname",

1489

"objectpath", "interface")

1490

expected_method_return = {"key1": {"key2": {"key3": True}}}

1491

self.assertEqual(expected_method_return, ret)

1492

self.assertIsInstance(ret, dict)

1493

self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)

1494

self.assertIsInstance(ret["key1"], dict)

1495

self.assertNotIsInstance(ret["key1"],

1496

mock_dbus_python.Dictionary)

1497

self.assertIsInstance(ret["key1"]["key2"], dict)

1498

self.assertNotIsInstance(ret["key1"]["key2"],

1499

mock_dbus_python.Dictionary)

1500

self.assertTrue(ret["key1"]["key2"]["key3"])

1501

self.assertNotIsInstance(ret["key1"]["key2"]["key3"],

1502

mock_dbus_python.Boolean)

1503

1504

def test_call_method_handles_exception(self):

1505

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

1506

1507

def func():

1508

dbus_logger.error("Test")

1509

raise mock_dbus_python.exceptions.DBusException()

1510

1511

mock_dbus_python = self.MockDBusPython_func(func)

1512

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1513

1514

class CountingHandler(logging.Handler):

1515

count = 0

1516

def emit(self, record):

1517

self.count += 1

1518

1519

counting_handler = CountingHandler()

1520

1521

dbus_logger.addHandler(counting_handler)

1522

1523

try:

1524

with self.assertRaises(dbus.Error) as e:

1525

self.call_method(bus, "methodname", "busname",

1526

"objectpath", "interface")

1527

finally:

1528

dbus_logger.removeFilter(counting_handler)

1529

1530

self.assertNotIsInstance(e, dbus.ConnectFailed)

1531

1532

# Make sure the dbus logger was suppressed

1533

self.assertEqual(0, counting_handler.count)

1534

1535

def test_Set_Secret_sends_bytearray(self):

1536

ret = [None]

1537

def func(*args, **kwargs):

1538

ret[0] = (args, kwargs)

1539

mock_dbus_python = self.MockDBusPython_func(func)

1540

bus = dbus_python_adapter.SystemBus(mock_dbus_python)

1541

bus.set_client_property("objectpath", "Secret", "value")

1542

expected_call = (("se.recompile.Mandos.Client", "Secret",

1543

mock_dbus_python.ByteArray(b"value")),

1544

{"dbus_interface":

1545

"org.freedesktop.DBus.Properties"})

1546

self.assertEqual(expected_call, ret[0])

1547

if sys.version_info.major == 2:

1548

self.assertIsInstance(ret[0][0][-1],

1549

mock_dbus_python.ByteArray)

1550

1551

def test_get_object_converts_to_correct_exception(self):

1552

bus = dbus_python_adapter.SystemBus(

1553

self.fake_dbus_python_raises_exception_on_connect)

1554

with self.assertRaises(dbus.ConnectFailed):

1555

self.call_method(bus, "methodname", "busname",

1556

"objectpath", "interface")

1557

1558

class fake_dbus_python_raises_exception_on_connect(object):

1559

"""fake dbus-python module"""

1560

class exceptions(object):

1561

"""Pseudo-namespace"""

1562

class DBusException(Exception):

1563

pass

1564

1565

@classmethod

1566

def SystemBus(cls):

1567

def get_object(busname, objectpath):

1568

raise cls.exceptions.DBusException()

1569

Bus = collections.namedtuple("Bus", ["get_object"])

1570

return Bus(get_object=get_object)

1571

1572

1573

class Test_dbus_python_adapter_CachingBus(unittest.TestCase):

1574

class mock_dbus_python(object):

1575

"""mock dbus-python modules"""

1576

class SystemBus(object):

1577

@staticmethod

1578

def get_object(busname, objectpath):

1579

return Unique()

1580

1581

def setUp(self):

1582

self.bus = dbus_python_adapter.CachingBus(

1583

self.mock_dbus_python)

1584

1585

def test_returns_distinct_objectpaths(self):

1586

obj1 = self.bus.get_object("busname", "objectpath1")

1587

self.assertIsInstance(obj1, Unique)

1588

obj2 = self.bus.get_object("busname", "objectpath2")

1589

self.assertIsInstance(obj2, Unique)

1590

self.assertIsNot(obj1, obj2)

1591

1592

def test_returns_distinct_busnames(self):

1593

obj1 = self.bus.get_object("busname1", "objectpath")

1594

self.assertIsInstance(obj1, Unique)

1595

obj2 = self.bus.get_object("busname2", "objectpath")

1596

self.assertIsInstance(obj2, Unique)

1597

self.assertIsNot(obj1, obj2)

1598

1599

def test_returns_distinct_both(self):

1600

obj1 = self.bus.get_object("busname1", "objectpath")

1601

self.assertIsInstance(obj1, Unique)

1602

obj2 = self.bus.get_object("busname2", "objectpath")

1603

self.assertIsInstance(obj2, Unique)

1604

self.assertIsNot(obj1, obj2)

1605

1606

def test_returns_same(self):

1607

obj1 = self.bus.get_object("busname", "objectpath")

1608

self.assertIsInstance(obj1, Unique)

1609

obj2 = self.bus.get_object("busname", "objectpath")

1610

self.assertIsInstance(obj2, Unique)

1611

self.assertIs(obj1, obj2)

1612

1613

def test_returns_same_old(self):

1614

obj1 = self.bus.get_object("busname1", "objectpath1")

1615

self.assertIsInstance(obj1, Unique)

1616

obj2 = self.bus.get_object("busname2", "objectpath2")

1617

self.assertIsInstance(obj2, Unique)

1618

obj1b = self.bus.get_object("busname1", "objectpath1")

1619

self.assertIsInstance(obj1b, Unique)

1620

self.assertIsNot(obj1, obj2)

1621

self.assertIsNot(obj2, obj1b)

1622

self.assertIs(obj1, obj1b)

1623

1624

1625

class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):

1626

1627

def Stub_pydbus_func(self, func):

1628

class stub_pydbus(object):

1629

"""stub pydbus module"""

1630

class SystemBus(object):

1631

@staticmethod

1632

def get(busname, objectpath):

1633

DBusObject = collections.namedtuple(

1634

"DBusObject", ("methodname",))

1635

return {"interface":

1636

DBusObject(methodname=func)}

1637

return stub_pydbus

1638

1639

def call_method(self, bus, methodname, busname, objectpath,

1640

interface, *args):

1641

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

1642

return bus.call_method(methodname, busname, objectpath,

1643

interface, *args)

1644

1645

def test_call_method_returns(self):

1646

expected_method_return = Unique()

1647

method_args = (Unique(), Unique())

1648

def func(*args):

1649

self.assertEqual(len(method_args), len(args))

1650

for marg, arg in zip(method_args, args):

1651

self.assertIs(marg, arg)

1652

return expected_method_return

1653

stub_pydbus = self.Stub_pydbus_func(func)

1654

bus = pydbus_adapter.SystemBus(stub_pydbus)

1655

ret = self.call_method(bus, "methodname", "busname",

1656

"objectpath", "interface",

1657

*method_args)

1658

self.assertIs(ret, expected_method_return)

1659

1660

def test_call_method_handles_exception(self):

1661

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

1662

1663

def func():

1664

raise gi.repository.GLib.Error()

1665

1666

stub_pydbus = self.Stub_pydbus_func(func)

1667

bus = pydbus_adapter.SystemBus(stub_pydbus)

1668

1669

with self.assertRaises(dbus.Error) as e:

1670

self.call_method(bus, "methodname", "busname",

1671

"objectpath", "interface")

1672

1673

self.assertNotIsInstance(e, dbus.ConnectFailed)

1674

1675

def test_get_converts_to_correct_exception(self):

1676

bus = pydbus_adapter.SystemBus(

1677

self.fake_pydbus_raises_exception_on_connect)

1678

with self.assertRaises(dbus.ConnectFailed):

1679

self.call_method(bus, "methodname", "busname",

1680

"objectpath", "interface")

1681

1682

class fake_pydbus_raises_exception_on_connect(object):

1683

"""fake dbus-python module"""

1684

@classmethod

1685

def SystemBus(cls):

1686

def get(busname, objectpath):

1687

raise gi.repository.GLib.Error()

1688

Bus = collections.namedtuple("Bus", ["get"])

1689

return Bus(get=get)

1690

1691

def test_set_property_uses_setattr(self):

1692

class Object(object):

1693

pass

1694

obj = Object()

1695

class pydbus_spy(object):

1696

class SystemBus(object):

1697

@staticmethod

1698

def get(busname, objectpath):

1699

return {"interface": obj}

1700

bus = pydbus_adapter.SystemBus(pydbus_spy)

1701

value = Unique()

1702

bus.set_property("busname", "objectpath", "interface", "key",

1703

value)

1704

self.assertIs(value, obj.key)

1705

1706

def test_get_suppresses_xml_deprecation_warning(self):

1707

if sys.version_info.major >= 3:

1708

return

1709

class stub_pydbus_get(object):

1710

class SystemBus(object):

1711

@staticmethod

1712

def get(busname, objectpath):

1713

warnings.warn_explicit(

1714

"deprecated", DeprecationWarning,

1715

"xml.etree.ElementTree", 0)

1716

bus = pydbus_adapter.SystemBus(stub_pydbus_get)

1717

with warnings.catch_warnings(record=True) as w:

1718

warnings.simplefilter("always")

1719

bus.get("busname", "objectpath")

1720

self.assertEqual(0, len(w))

1721

1722

1723

class Test_pydbus_adapter_CachingBus(unittest.TestCase):

1724

class stub_pydbus(object):

1725

"""stub pydbus module"""

1726

class SystemBus(object):

1727

@staticmethod

1728

def get(busname, objectpath):

1729

return Unique()

1730

1731

def setUp(self):

1732

self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)

1733

1734

def test_returns_distinct_objectpaths(self):

1735

obj1 = self.bus.get("busname", "objectpath1")

1736

self.assertIsInstance(obj1, Unique)

1737

obj2 = self.bus.get("busname", "objectpath2")

1738

self.assertIsInstance(obj2, Unique)

1739

self.assertIsNot(obj1, obj2)

1740

1741

def test_returns_distinct_busnames(self):

1742

obj1 = self.bus.get("busname1", "objectpath")

1743

self.assertIsInstance(obj1, Unique)

1744

obj2 = self.bus.get("busname2", "objectpath")

1745

self.assertIsInstance(obj2, Unique)

1746

self.assertIsNot(obj1, obj2)

1747

1748

def test_returns_distinct_both(self):

1749

obj1 = self.bus.get("busname1", "objectpath")

1750

self.assertIsInstance(obj1, Unique)

1751

obj2 = self.bus.get("busname2", "objectpath")

1752

self.assertIsInstance(obj2, Unique)

1753

self.assertIsNot(obj1, obj2)

1754

1755

def test_returns_same(self):

1756

obj1 = self.bus.get("busname", "objectpath")

1757

self.assertIsInstance(obj1, Unique)

1758

obj2 = self.bus.get("busname", "objectpath")

1759

self.assertIsInstance(obj2, Unique)

1760

self.assertIs(obj1, obj2)

1761

1762

def test_returns_same_old(self):

1763

obj1 = self.bus.get("busname1", "objectpath1")

1764

self.assertIsInstance(obj1, Unique)

1765

obj2 = self.bus.get("busname2", "objectpath2")

1766

self.assertIsInstance(obj2, Unique)

1767

obj1b = self.bus.get("busname1", "objectpath1")

1768

self.assertIsInstance(obj1b, Unique)

1769

self.assertIsNot(obj1, obj2)

1770

self.assertIsNot(obj2, obj1b)

1771

self.assertIs(obj1, obj1b)

1772

1773

1774

class Test_commands_from_options(unittest.TestCase):

1775

1776

def setUp(self):

1777

self.parser = argparse.ArgumentParser()

1778

add_command_line_options(self.parser)

1779

1780

def test_is_enabled(self):

1781

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

1782

command.IsEnabled)

1783

1784

def assert_command_from_args(self, args, command_cls, length=1,

1785

clients=None, **cmd_attrs):

1786

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

1787

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

1788

options = self.parser.parse_args(args)

1789

check_option_syntax(self.parser, options)

1790

commands = commands_from_options(options)

1791

self.assertEqual(length, len(commands))

1792

for command in commands:

1793

if isinstance(command, command_cls):

1794

break

1795

else:

1796

self.assertIsInstance(command, command_cls)

1797

if clients is not None:

1798

self.assertEqual(clients, options.client)

1799

for key, value in cmd_attrs.items():

1800

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

1801

1802

def assert_commands_from_args(self, args, commands, clients=None):

1803

for cmd in commands:

1804

self.assert_command_from_args(args, cmd,

1805

length=len(commands),

1806

clients=clients)

1807

1808

def test_is_enabled_short(self):

1809

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

1810

command.IsEnabled)

1811

1812

def test_approve(self):

1813

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

1814

command.Approve)

1815

1816

def test_approve_short(self):

1817

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

1818

command.Approve)

1819

1820

def test_deny(self):

1821

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

1822

command.Deny)

1823

1824

def test_deny_short(self):

1825

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

1826

1827

def test_remove(self):

1828

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

1829

command.Remove)

1830

1831

def test_deny_before_remove(self):

1832

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

1833

"client"])

1834

check_option_syntax(self.parser, options)

1835

commands = commands_from_options(options)

1836

self.assertEqual(2, len(commands))

1837

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

1838

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

1839

1840

def test_deny_before_remove_reversed(self):

1841

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

1842

"--all"])

1843

check_option_syntax(self.parser, options)

1844

commands = commands_from_options(options)

1845

self.assertEqual(2, len(commands))

1846

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

1847

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

1848

1849

def test_remove_short(self):

1850

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

1851

command.Remove)

1852

1853

def test_dump_json(self):

1854

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

1855

command.DumpJSON)

1856

1857

def test_enable(self):

1858

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

1859

command.Enable)

1860

1861

def test_enable_short(self):

1862

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

1863

command.Enable)

1864

1865

def test_disable(self):

1866

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

1867

command.Disable)

1868

1869

def test_disable_short(self):

1870

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

1871

command.Disable)

1872

1873

def test_bump_timeout(self):

1874

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

1875

command.BumpTimeout)

1876

1877

def test_bump_timeout_short(self):

1878

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

1879

command.BumpTimeout)

1880

1881

def test_start_checker(self):

1882

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

1883

command.StartChecker)

1884

1885

def test_stop_checker(self):

1886

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

1887

command.StopChecker)

1888

1889

def test_approve_by_default(self):

1890

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

1891

"client"],

1892

command.ApproveByDefault)

1893

1894

def test_deny_by_default(self):

1895

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

1896

command.DenyByDefault)

1897

1898

def test_checker(self):

1899

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

1900

command.SetChecker,

1901

value_to_set=":")

1902

1903

def test_checker_empty(self):

1904

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

1905

command.SetChecker,

1906

value_to_set="")

1907

1908

def test_checker_short(self):

1909

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

1910

command.SetChecker,

1911

value_to_set=":")

1912

1913

def test_host(self):

1914

self.assert_command_from_args(

1915

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

1916

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

1917

1918

def test_host_short(self):

1919

self.assert_command_from_args(

1920

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

1921

value_to_set="client.example.org")

1922

1923

def test_secret_devnull(self):

1924

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

1925

"client"], command.SetSecret,

1926

value_to_set=b"")

1927

1928

def test_secret_tempfile(self):

1929

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

1930

value = b"secret\0xyzzy\nbar"

1931

f.write(value)

1932

f.seek(0)

1933

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

1934

"client"],

1935

command.SetSecret,

1936

value_to_set=value)

1937

1938

def test_secret_devnull_short(self):

1939

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

1940

"client"], command.SetSecret,

1941

value_to_set=b"")

1942

1943

def test_secret_tempfile_short(self):

1944

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

1945

value = b"secret\0xyzzy\nbar"

1946

f.write(value)

1947

f.seek(0)

1948

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

1949

command.SetSecret,

1950

value_to_set=value)

1951

1952

def test_timeout(self):

1953

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

1954

command.SetTimeout,

1955

value_to_set=300000)

1956

1957

def test_timeout_short(self):

1958

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

1959

command.SetTimeout,

1960

value_to_set=300000)

1961

1962

def test_extended_timeout(self):

1963

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

1964

"client"],

1965

command.SetExtendedTimeout,

1966

value_to_set=900000)

1967

1968

def test_interval(self):

1969

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

1970

"client"], command.SetInterval,

1971

value_to_set=120000)

1972

1973

def test_interval_short(self):

1974

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

1975

command.SetInterval,

1976

value_to_set=120000)

1977

1978

def test_approval_delay(self):

1979

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

1980

"client"],

1981

command.SetApprovalDelay,

1982

value_to_set=30000)

1983

1984

def test_approval_duration(self):

1985

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

1986

"client"],

1987

command.SetApprovalDuration,

1988

value_to_set=1000)

1989

1990

def test_print_table(self):

1991

self.assert_command_from_args([], command.PrintTable,

1992

verbose=False)

1993

1994

def test_print_table_verbose(self):

1995

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

1996

command.PrintTable,

1997

verbose=True)

1998

1999

def test_print_table_verbose_short(self):

2000

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

2001

verbose=True)

2002

2003

2004

def test_manual_page_example_1(self):

2005

self.assert_command_from_args("",

2006

command.PrintTable,

2007

clients=[],

2008

verbose=False)

2009

2010

def test_manual_page_example_2(self):

2011

self.assert_command_from_args(

2012

"--verbose foo1.example.org foo2.example.org".split(),

2013

command.PrintTable, clients=["foo1.example.org",

2014

"foo2.example.org"],

2015

verbose=True)

2016

2017

def test_manual_page_example_3(self):

2018

self.assert_command_from_args("--enable --all".split(),

2019

command.Enable,

2020

clients=[])

2021

2022

def test_manual_page_example_4(self):

2023

self.assert_commands_from_args(

2024

("--timeout=PT5M --interval=PT1M foo1.example.org"

2025

" foo2.example.org").split(),

2026

[command.SetTimeout, command.SetInterval],

2027

clients=["foo1.example.org", "foo2.example.org"])

2028

2029

def test_manual_page_example_5(self):

2030

self.assert_command_from_args("--approve --all".split(),

2031

command.Approve,

2032

clients=[])

2033

2034

2035

class TestCommand(unittest.TestCase):

2036

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

2037

2038

class FakeMandosBus(dbus.MandosBus):

2039

def __init__(self, testcase):

2040

self.client_properties = {

2041

"Name": "foo",

2042

"KeyID": ("92ed150794387c03ce684574b1139a65"

2043

"94a34f895daaaf09fd8ea90a27cddb12"),

2044

"Secret": b"secret",

2045

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

2046

"Enabled": True,

2047

"Timeout": 300000,

2048

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

2049

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

2050

"Interval": 120000,

2051

"Fingerprint": ("778827225BA7DE539C5A"

2052

"7CFA59CFF7CDBD9A5920"),

2053

"CheckerRunning": False,

2054

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

2055

"ApprovalPending": False,

2056

"ApprovedByDefault": True,

2057

"LastApprovalRequest": "",

2058

"ApprovalDelay": 0,

2059

"ApprovalDuration": 1000,

2060

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

2061

"ExtendedTimeout": 900000,

2062

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

2063

"LastCheckerStatus": 0,

2064

}

2065

self.other_client_properties = {

2066

"Name": "barbar",

2067

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2068

"6ab612cff5ad227247e46c2b020f441c"),

2069

"Secret": b"secretbar",

2070

"Host": "192.0.2.3",

2071

"Enabled": True,

2072

"Timeout": 300000,

2073

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

2074

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

2075

"Interval": 120000,

2076

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2077

"F547B3A107558FCA3A27"),

2078

"CheckerRunning": True,

2079

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

2080

"ApprovalPending": False,

2081

"ApprovedByDefault": False,

2082

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

2083

"ApprovalDelay": 30000,

2084

"ApprovalDuration": 93785000,

2085

"Checker": ":",

2086

"ExtendedTimeout": 900000,

2087

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

2088

"LastCheckerStatus": -2,

2089

}

2090

self.clients = collections.OrderedDict(

2091

[

2092

("client_objectpath", self.client_properties),

2093

("other_client_objectpath",

2094

self.other_client_properties),

2095

])

2096

self.one_client = {"client_objectpath":

2097

self.client_properties}

2098

self.testcase = testcase

2099

self.calls = []

2100

2101

def call_method(self, methodname, busname, objectpath,

2102

interface, *args):

2103

self.testcase.assertEqual("se.recompile.Mandos", busname)

2104

self.calls.append((methodname, busname, objectpath,

2105

interface, args))

2106

if interface == "org.freedesktop.DBus.Properties":

2107

if methodname == "Set":

2108

self.testcase.assertEqual(3, len(args))

2109

interface, key, value = args

2110

self.testcase.assertEqual(

2111

"se.recompile.Mandos.Client", interface)

2112

self.clients[objectpath][key] = value

2113

return

2114

elif interface == "se.recompile.Mandos":

2115

self.testcase.assertEqual("RemoveClient", methodname)

2116

self.testcase.assertEqual(1, len(args))

2117

clientpath = args[0]

2118

del self.clients[clientpath]

2119

return

2120

elif interface == "se.recompile.Mandos.Client":

2121

if methodname == "Approve":

2122

self.testcase.assertEqual(1, len(args))

2123

return

2124

raise ValueError()

2125

2126

def setUp(self):

2127

self.bus = self.FakeMandosBus(self)

2128

2129

2130

class TestBaseCommands(TestCommand):

2131

2132

def test_IsEnabled_exits_successfully(self):

2133

with self.assertRaises(SystemExit) as e:

2134

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

2135

if e.exception.code is not None:

2136

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

2137

else:

2138

self.assertIsNone(e.exception.code)

2139

2140

def test_IsEnabled_exits_with_failure(self):

2141

self.bus.client_properties["Enabled"] = False

2142

with self.assertRaises(SystemExit) as e:

2143

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

2144

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

2145

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

2146

else:

2147

self.assertIsNotNone(e.exception.code)

2148

2149

def test_Approve(self):

2150

busname = "se.recompile.Mandos"

2151

client_interface = "se.recompile.Mandos.Client"

2152

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

2153

for clientpath in self.bus.clients:

2154

self.assertIn(("Approve", busname, clientpath,

2155

client_interface, (True,)), self.bus.calls)

2156

2157

def test_Deny(self):

2158

busname = "se.recompile.Mandos"

2159

client_interface = "se.recompile.Mandos.Client"

2160

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

2161

for clientpath in self.bus.clients:

2162

self.assertIn(("Approve", busname, clientpath,

2163

client_interface, (False,)),

2164

self.bus.calls)

2165

2166

def test_Remove(self):

2167

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

2168

for clientpath in self.bus.clients:

2169

self.assertIn(("RemoveClient", dbus_busname,

2170

dbus_server_path, dbus_server_interface,

2171

(clientpath,)), self.bus.calls)

2172

2173

expected_json = {

2174

"foo": {

2175

"Name": "foo",

2176

"KeyID": ("92ed150794387c03ce684574b1139a65"

2177

"94a34f895daaaf09fd8ea90a27cddb12"),

2178

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

2179

"Enabled": True,

2180

"Timeout": 300000,

2181

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

2182

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

2183

"Interval": 120000,

2184

"Fingerprint": ("778827225BA7DE539C5A"

2185

"7CFA59CFF7CDBD9A5920"),

2186

"CheckerRunning": False,

2187

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

2188

"ApprovalPending": False,

2189

"ApprovedByDefault": True,

2190

"LastApprovalRequest": "",

2191

"ApprovalDelay": 0,

2192

"ApprovalDuration": 1000,

2193

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

2194

"ExtendedTimeout": 900000,

2195

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

2196

"LastCheckerStatus": 0,

2197

},

2198

"barbar": {

2199

"Name": "barbar",

2200

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

2201

"6ab612cff5ad227247e46c2b020f441c"),

2202

"Host": "192.0.2.3",

2203

"Enabled": True,

2204

"Timeout": 300000,

2205

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

2206

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

2207

"Interval": 120000,

2208

"Fingerprint": ("3E393AEAEFB84C7E89E2"

2209

"F547B3A107558FCA3A27"),

2210

"CheckerRunning": True,

2211

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

2212

"ApprovalPending": False,

2213

"ApprovedByDefault": False,

2214

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

2215

"ApprovalDelay": 30000,

2216

"ApprovalDuration": 93785000,

2217

"Checker": ":",

2218

"ExtendedTimeout": 900000,

2219

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

2220

"LastCheckerStatus": -2,

2221

},

2222

}

2223

2224

def test_DumpJSON_normal(self):

2225

with self.capture_stdout_to_buffer() as buffer:

2226

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

2227

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

2228

self.assertDictEqual(self.expected_json, json_data)

2229

2230

@staticmethod

2231

@contextlib.contextmanager

2232

def capture_stdout_to_buffer():

2233

capture_buffer = io.StringIO()

2234

old_stdout = sys.stdout

2235

sys.stdout = capture_buffer

2236

try:

2237

yield capture_buffer

2238

finally:

2239

sys.stdout = old_stdout

2240

2241

def test_DumpJSON_one_client(self):

2242

with self.capture_stdout_to_buffer() as buffer:

2243

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

2244

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

2245

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

2246

self.assertDictEqual(expected_json, json_data)

2247

2248

def test_PrintTable_normal(self):

2249

with self.capture_stdout_to_buffer() as buffer:

2250

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

2251

expected_output = "\n".join((

2252

"Name Enabled Timeout Last Successful Check",

2253

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

2254

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

2255

)) + "\n"

2256

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

2257

2258

def test_PrintTable_verbose(self):

2259

with self.capture_stdout_to_buffer() as buffer:

2260

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

2261

columns = (

2262

(

2263

"Name ",

2264

"foo ",

2265

"barbar ",

2266

),(

2267

"Enabled ",

2268

"Yes ",

2269

"Yes ",

2270

),(

2271

"Timeout ",

2272

"00:05:00 ",

2273

"00:05:00 ",

2274

),(

2275

"Last Successful Check ",

2276

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

2277

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

2278

),(

2279

"Created ",

2280

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

2281

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

2282

),(

2283

"Interval ",

2284

"00:02:00 ",

2285

"00:02:00 ",

2286

),(

2287

"Host ",

2288

"foo.example.org ",

2289

"192.0.2.3 ",

2290

),(

2291

("Key ID "

2292

" "),

2293

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

2294

"ea90a27cddb12 "),

2295

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

2296

"46c2b020f441c "),

2297

),(

2298

"Fingerprint ",

2299

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

2300

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

2301

),(

2302

"Check Is Running ",

2303

"No ",

2304

"Yes ",

2305

),(

2306

"Last Enabled ",

2307

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

2308

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

2309

),(

2310

"Approval Is Pending ",

2311

"No ",

2312

"No ",

2313

),(

2314

"Approved By Default ",

2315

"Yes ",

2316

"No ",

2317

),(

2318

"Last Approval Request ",

2319

" ",

2320

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

2321

),(

2322

"Approval Delay ",

2323

"00:00:00 ",

2324

"00:00:30 ",

2325

),(

2326

"Approval Duration ",

2327

"00:00:01 ",

2328

"1T02:03:05 ",

2329

),(

2330

"Checker ",

2331

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

2332

": ",

2333

),(

2334

"Extended Timeout ",

2335

"00:15:00 ",

2336

"00:15:00 ",

2337

),(

2338

"Expires ",

2339

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

2340

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

2341

),(

2342

"Last Checker Status",

2343

"0 ",

2344

"-2 ",

2345

)

2346

)

2347

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

2348

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

2349

for rows in columns)

2350

for line in range(num_lines))

2351

+ "\n")

2352

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

2353

2354

def test_PrintTable_one_client(self):

2355

with self.capture_stdout_to_buffer() as buffer:

2356

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

2357

expected_output = "\n".join((

2358

"Name Enabled Timeout Last Successful Check",

2359

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

2360

)) + "\n"

2361

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

2362

2363

2364

class TestPropertySetterCmd(TestCommand):

2365

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

2366

2367

def runTest(self):

2368

if not hasattr(self, "command"):

2369

return # Abstract TestCase class

2370

2371

if hasattr(self, "values_to_set"):

2372

cmd_args = [(value,) for value in self.values_to_set]

2373

values_to_get = getattr(self, "values_to_get",

2374

self.values_to_set)

2375

else:

2376

cmd_args = [() for x in range(len(self.values_to_get))]

2377

values_to_get = self.values_to_get

2378

for value_to_get, cmd_arg in zip(values_to_get, cmd_args):

2379

for clientpath in self.bus.clients:

2380

self.bus.clients[clientpath][self.propname] = (

2381

Unique())

2382

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

2383

for clientpath in self.bus.clients:

2384

value = (self.bus.clients[clientpath]

2385

[self.propname])

2386

self.assertNotIsInstance(value, Unique)

2387

self.assertEqual(value_to_get, value)

2388

2389

2390

class TestEnableCmd(TestPropertySetterCmd):

2391

command = command.Enable

2392

propname = "Enabled"

2393

values_to_get = [True]

2394

2395

2396

class TestDisableCmd(TestPropertySetterCmd):

2397

command = command.Disable

2398

propname = "Enabled"

2399

values_to_get = [False]

2400

2401

2402

class TestBumpTimeoutCmd(TestPropertySetterCmd):

2403

command = command.BumpTimeout

2404

propname = "LastCheckedOK"

2405

values_to_get = [""]

2406

2407

2408

class TestStartCheckerCmd(TestPropertySetterCmd):

2409

command = command.StartChecker

2410

propname = "CheckerRunning"

2411

values_to_get = [True]

2412

2413

2414

class TestStopCheckerCmd(TestPropertySetterCmd):

2415

command = command.StopChecker

2416

propname = "CheckerRunning"

2417

values_to_get = [False]

2418

2419

2420

class TestApproveByDefaultCmd(TestPropertySetterCmd):

2421

command = command.ApproveByDefault

2422

propname = "ApprovedByDefault"

2423

values_to_get = [True]

2424

2425

2426

class TestDenyByDefaultCmd(TestPropertySetterCmd):

2427

command = command.DenyByDefault

2428

propname = "ApprovedByDefault"

2429

values_to_get = [False]

2430

2431

2432

class TestSetCheckerCmd(TestPropertySetterCmd):

2433

command = command.SetChecker

2434

propname = "Checker"

2435

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

2436

2437

2438

class TestSetHostCmd(TestPropertySetterCmd):

2439

command = command.SetHost

2440

propname = "Host"

2441

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

2442

2443

2444

class TestSetSecretCmd(TestPropertySetterCmd):

2445

command = command.SetSecret

2446

propname = "Secret"

2447

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

2448

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

2449

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

2450

2451

2452

class TestSetTimeoutCmd(TestPropertySetterCmd):

2453

command = command.SetTimeout

2454

propname = "Timeout"

2455

values_to_set = [datetime.timedelta(),

2456

datetime.timedelta(minutes=5),

2457

datetime.timedelta(seconds=1),

2458

datetime.timedelta(weeks=1),

2459

datetime.timedelta(weeks=52)]

2460

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

2461

2462

2463

class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):

2464

command = command.SetExtendedTimeout

2465

propname = "ExtendedTimeout"

2466

values_to_set = [datetime.timedelta(),

2467

datetime.timedelta(minutes=5),

2468

datetime.timedelta(seconds=1),

2469

datetime.timedelta(weeks=1),

2470

datetime.timedelta(weeks=52)]

2471

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

2472

2473

2474

class TestSetIntervalCmd(TestPropertySetterCmd):

2475

command = command.SetInterval

2476

propname = "Interval"

2477

values_to_set = [datetime.timedelta(),

2478

datetime.timedelta(minutes=5),

2479

datetime.timedelta(seconds=1),

2480

datetime.timedelta(weeks=1),

2481

datetime.timedelta(weeks=52)]

2482

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

2483

2484

2485

class TestSetApprovalDelayCmd(TestPropertySetterCmd):

2486

command = command.SetApprovalDelay

2487

propname = "ApprovalDelay"

2488

values_to_set = [datetime.timedelta(),

2489

datetime.timedelta(minutes=5),

2490

datetime.timedelta(seconds=1),

2491

datetime.timedelta(weeks=1),

2492

datetime.timedelta(weeks=52)]

2493

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

2494

2495

2496

class TestSetApprovalDurationCmd(TestPropertySetterCmd):

2497

command = command.SetApprovalDuration

2498

propname = "ApprovalDuration"

2499

values_to_set = [datetime.timedelta(),

2500

datetime.timedelta(minutes=5),

2501

datetime.timedelta(seconds=1),

2502

datetime.timedelta(weeks=1),

2503

datetime.timedelta(weeks=52)]

2504

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

2505

2506

2507

2508

def should_only_run_tests():

2509

parser = argparse.ArgumentParser(add_help=False)

2510

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

2511

args, unknown_args = parser.parse_known_args()

2512

run_tests = args.check

2513

if run_tests:

2514

# Remove --check argument from sys.argv

2515

sys.argv[1:] = unknown_args

2516

return run_tests

2517

2518

# Add all tests from doctest strings

2519

def load_tests(loader, tests, none):

2520

import doctest

2521

tests.addTests(doctest.DocTestSuite())

2522

return tests

2523

2524

if __name__ == "__main__":

2525

try:

2526

if should_only_run_tests():

2527

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

2528

unittest.main()

2529

else:

2530

main()

2531

finally:

2532

logging.shutdown()