/mandos/trunk : revision 24.1.162

1

#!/usr/bin/python

2

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

3

#

4

# Mandos Monitor - Control and monitor the Mandos server

5

#

6

7

8

#

9

# This 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

2

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

3

4

from __future__ import division

35

5

import sys

36

import argparse

6

import dbus

7

from optparse import OptionParser

37

8

import locale

38

9

import datetime

39

10

import re

40

import os

41

import collections

42

import json

43

import unittest

44

import logging

45

46

import dbus

47

48

# Show warnings by default

49

if not sys.warnoptions:

50

import warnings

51

warnings.simplefilter("default")

52

53

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

54

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

55

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

56

57

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

58

59

if sys.version_info.major == 2:

60

str = unicode

61

62

locale.setlocale(locale.LC_ALL, "")

63

64

domain = "se.recompile"

65

busname = domain + ".Mandos"

66

server_path = "/"

67

server_interface = domain + ".Mandos"

68

client_interface = domain + ".Mandos.Client"

69

version = "1.8.3"

70

71

11

12

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

13

14

tablewords = {

15

'name': u'Name',

16

'enabled': u'Enabled',

17

'timeout': u'Timeout',

18

'last_checked_ok': u'Last Successful Check',

19

'created': u'Created',

20

'interval': u'Interval',

21

'host': u'Host',

22

'fingerprint': u'Fingerprint',

23

'checker_running': u'Check Is Running',

24

'last_enabled': u'Last Enabled',

25

'checker': u'Checker',

26

}

27

defaultkeywords = ('name', 'enabled', 'timeout', 'last_checked_ok')

28

domain = 'se.bsnet.fukt'

29

busname = domain + '.Mandos'

30

server_path = '/'

31

server_interface = domain + '.Mandos'

32

client_interface = domain + '.Mandos.Client'

33

version = "1.0.14"

72

34

try:

73

dbus.OBJECT_MANAGER_IFACE

74

except AttributeError:

75

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

35

bus = dbus.SystemBus()

36

mandos_dbus_objc = bus.get_object(busname, server_path)

37

except dbus.exceptions.DBusException:

38

sys.exit(1)

39

40

mandos_serv = dbus.Interface(mandos_dbus_objc,

41

dbus_interface = server_interface)

42

mandos_clients = mandos_serv.GetAllClientsWithProperties()

76

43

44

def timedelta_to_milliseconds(td):

45

"Convert a datetime.timedelta object to milliseconds"

46

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

47

+ (td.seconds * 1000)

48

+ (td.microseconds // 1000))

77

49

78

50

def milliseconds_to_string(ms):

79

51

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

80

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

81

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

82

hours=td.seconds // 3600,

83

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

84

seconds=td.seconds % 60))

85

86

87

def rfc3339_duration_to_delta(duration):

88

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

89

90

>>> rfc3339_duration_to_delta("P7D")

91

datetime.timedelta(7)

92

>>> rfc3339_duration_to_delta("PT60S")

93

datetime.timedelta(0, 60)

94

>>> rfc3339_duration_to_delta("PT60M")

95

datetime.timedelta(0, 3600)

96

>>> rfc3339_duration_to_delta("P60M")

97

datetime.timedelta(1680)

98

>>> rfc3339_duration_to_delta("PT24H")

99

datetime.timedelta(1)

100

>>> rfc3339_duration_to_delta("P1W")

101

datetime.timedelta(7)

102

>>> rfc3339_duration_to_delta("PT5M30S")

103

datetime.timedelta(0, 330)

104

>>> rfc3339_duration_to_delta("P1DT3M20S")

105

datetime.timedelta(1, 200)

106

>>> # Can not be empty:

107

>>> rfc3339_duration_to_delta("")

108

Traceback (most recent call last):

109

...

110

ValueError: Invalid RFC 3339 duration: u''

111

>>> # Must start with "P":

112

>>> rfc3339_duration_to_delta("1D")

113

Traceback (most recent call last):

114

...

115

ValueError: Invalid RFC 3339 duration: u'1D'

116

>>> # Must use correct order

117

>>> rfc3339_duration_to_delta("PT1S2M")

118

Traceback (most recent call last):

119

...

120

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

121

>>> # Time needs time marker

122

>>> rfc3339_duration_to_delta("P1H2S")

123

Traceback (most recent call last):

124

...

125

ValueError: Invalid RFC 3339 duration: u'P1H2S'

126

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

127

>>> rfc3339_duration_to_delta("P1D2W")

128

Traceback (most recent call last):

129

...

130

ValueError: Invalid RFC 3339 duration: u'P1D2W'

131

>>> rfc3339_duration_to_delta("P2W2H")

132

Traceback (most recent call last):

133

...

134

ValueError: Invalid RFC 3339 duration: u'P2W2H'

135

"""

136

137

# Parsing an RFC 3339 duration with regular expressions is not

138

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

139

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

140

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

141

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

142

# avoid excessive use of external libraries.

143

144

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

145

Token = collections.namedtuple("Token", (

146

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

147

# a "group" containing digits

148

"value", # datetime.timedelta or None

149

"followers")) # Tokens valid after this token

150

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

151

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

152

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

153

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

154

datetime.timedelta(seconds=1),

155

frozenset((token_end, )))

156

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

157

datetime.timedelta(minutes=1),

158

frozenset((token_second, token_end)))

159

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

160

datetime.timedelta(hours=1),

161

frozenset((token_minute, token_end)))

162

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

163

None,

164

frozenset((token_hour, token_minute,

165

token_second)))

166

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

167

datetime.timedelta(days=1),

168

frozenset((token_time, token_end)))

169

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

170

datetime.timedelta(weeks=4),

171

frozenset((token_day, token_end)))

172

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

173

datetime.timedelta(weeks=52),

174

frozenset((token_month, token_end)))

175

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

176

datetime.timedelta(weeks=1),

177

frozenset((token_end, )))

178

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

179

frozenset((token_year, token_month,

180

token_day, token_time,

181

token_week)))

182

# Define starting values:

183

# Value so far

184

value = datetime.timedelta()

185

found_token = None

186

# Following valid tokens

187

followers = frozenset((token_duration, ))

188

# String left to parse

189

s = duration

190

# Loop until end token is found

191

while found_token is not token_end:

192

# Search for any currently valid tokens

193

for token in followers:

194

match = token.regexp.match(s)

195

if match is not None:

196

# Token found

197

if token.value is not None:

198

# Value found, parse digits

199

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

200

# Add to value so far

201

value += factor * token.value

202

# Strip token from string

203

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

204

# Go to found token

205

found_token = token

206

# Set valid next tokens

207

followers = found_token.followers

208

break

209

else:

210

# No currently valid tokens were found

211

raise ValueError("Invalid RFC 3339 duration: {!r}"

212

.format(duration))

213

# End token found

214

return value

52

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

53

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

54

"hours": td.seconds // 3600,

55

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

56

"seconds": td.seconds % 60,

57

})

215

58

216

59

217

60

def string_to_delta(interval):

218

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

219

220

try:

221

return rfc3339_duration_to_delta(interval)

222

except ValueError as e:

223

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

224

' '.join(e.args))

225

return parse_pre_1_6_1_interval(interval)

226

227

228

def parse_pre_1_6_1_interval(interval):

229

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

230

and return a datetime.timedelta

231

232

>>> parse_pre_1_6_1_interval('7d')

61

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

62

63

>>> string_to_delta('7d')

233

64

datetime.timedelta(7)

234

>>> parse_pre_1_6_1_interval('60s')

65

>>> string_to_delta('60s')

235

66

datetime.timedelta(0, 60)

236

>>> parse_pre_1_6_1_interval('60m')

67

>>> string_to_delta('60m')

237

68

datetime.timedelta(0, 3600)

238

>>> parse_pre_1_6_1_interval('24h')

69

>>> string_to_delta('24h')

239

70

datetime.timedelta(1)

240

>>> parse_pre_1_6_1_interval('1w')

71

>>> string_to_delta(u'1w')

241

72

datetime.timedelta(7)

242

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

73

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

243

74

datetime.timedelta(0, 330)

244

>>> parse_pre_1_6_1_interval('')

245

datetime.timedelta(0)

246

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

247

>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')

248

datetime.timedelta(2, 480, 18000)

249

250

75

"""

251

252

value = datetime.timedelta(0)

253

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

254

255

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

256

if suffix == "d":

257

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

258

elif suffix == "s":

259

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

260

elif suffix == "m":

261

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

262

elif suffix == "h":

263

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

264

elif suffix == "w":

265

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

266

elif suffix == "":

267

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

268

return value

269

270

271

## Classes for commands.

272

273

# Abstract classes first

274

class Command(object):

275

"""Abstract class for commands"""

276

def run(self, mandos, clients):

277

"""Normal commands should implement run_on_one_client(), but

278

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

279

can override this run() method instead."""

280

self.mandos = mandos

281

for client, properties in clients.items():

282

self.run_on_one_client(client, properties)

283

284

class PrintCmd(Command):

285

"""Abstract class for commands printing client details"""

286

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

287

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

288

"Fingerprint", "CheckerRunning", "LastEnabled",

289

"ApprovalPending", "ApprovedByDefault",

290

"LastApprovalRequest", "ApprovalDelay",

291

"ApprovalDuration", "Checker", "ExtendedTimeout",

292

"Expires", "LastCheckerStatus")

293

def run(self, mandos, clients):

294

print(self.output(clients))

295

296

class PropertyCmd(Command):

297

"""Abstract class for Actions for setting one client property"""

298

def run_on_one_client(self, client, properties):

299

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

300

client.Set(client_interface, self.property, self.value_to_set,

301

dbus_interface=dbus.PROPERTIES_IFACE)

302

303

class ValueArgumentMixIn(object):

304

"""Mixin class for commands taking a value as argument"""

305

def __init__(self, value):

306

self.value_to_set = value

307

308

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

309

"""Mixin class for commands taking a value argument as

310

milliseconds."""

311

@property

312

def value_to_set(self):

313

return self._vts

314

@value_to_set.setter

315

def value_to_set(self, value):

316

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

317

self._vts = string_to_delta(value).total_seconds() * 1000

318

319

# Actual (non-abstract) command classes

320

321

class PrintTableCmd(PrintCmd):

322

def __init__(self, verbose=False):

323

self.verbose = verbose

324

325

def output(self, clients):

326

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

327

keywords = default_keywords

328

if self.verbose:

329

keywords = self.all_keywords

330

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

331

332

class TableOfClients(object):

333

tableheaders = {

334

"Name": "Name",

335

"Enabled": "Enabled",

336

"Timeout": "Timeout",

337

"LastCheckedOK": "Last Successful Check",

338

"LastApprovalRequest": "Last Approval Request",

339

"Created": "Created",

340

"Interval": "Interval",

341

"Host": "Host",

342

"Fingerprint": "Fingerprint",

343

"KeyID": "Key ID",

344

"CheckerRunning": "Check Is Running",

345

"LastEnabled": "Last Enabled",

346

"ApprovalPending": "Approval Is Pending",

347

"ApprovedByDefault": "Approved By Default",

348

"ApprovalDelay": "Approval Delay",

349

"ApprovalDuration": "Approval Duration",

350

"Checker": "Checker",

351

"ExtendedTimeout": "Extended Timeout",

352

"Expires": "Expires",

353

"LastCheckerStatus": "Last Checker Status",

354

}

355

356

def __init__(self, clients, keywords, tableheaders=None):

357

self.clients = clients

358

self.keywords = keywords

359

if tableheaders is not None:

360

self.tableheaders = tableheaders

361

362

def __str__(self):

363

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

364

365

if sys.version_info.major == 2:

366

__unicode__ = __str__

367

def __str__(self):

368

return str(self).encode(locale.getpreferredencoding())

369

370

def rows(self):

371

format_string = self.row_formatting_string()

372

rows = [self.header_line(format_string)]

373

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

374

for client in self.clients)

375

return rows

376

377

def row_formatting_string(self):

378

"Format string used to format table rows"

379

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

380

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

381

*(len(self.string_from_client(client, key))

382

for client in self.clients)),

383

key=key)

384

for key in self.keywords)

385

386

def string_from_client(self, client, key):

387

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

388

389

@staticmethod

390

def valuetostring(value, keyword):

391

if isinstance(value, dbus.Boolean):

392

return "Yes" if value else "No"

393

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

394

"ApprovalDuration", "ExtendedTimeout"):

395

return milliseconds_to_string(value)

396

return str(value)

397

398

def header_line(self, format_string):

399

return format_string.format(**self.tableheaders)

400

401

def client_line(self, client, format_string):

402

return format_string.format(

403

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

404

for key in self.keywords})

405

406

407

408

class DumpJSONCmd(PrintCmd):

409

def output(self, clients):

410

data = {client["Name"]:

411

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

412

for key in self.all_keywords}

413

for client in clients.values()}

414

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

415

@staticmethod

416

def dbus_boolean_to_bool(value):

417

if isinstance(value, dbus.Boolean):

418

value = bool(value)

419

return value

420

421

class IsEnabledCmd(Command):

422

def run_on_one_client(self, client, properties):

423

if self.is_enabled(client, properties):

424

sys.exit(0)

76

timevalue = datetime.timedelta(0)

77

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

78

79

for s in regexp.findall(interval):

80

try:

81

suffix = unicode(s[-1])

82

value = int(s[:-1])

83

if suffix == u"d":

84

delta = datetime.timedelta(value)

85

elif suffix == u"s":

86

delta = datetime.timedelta(0, value)

87

elif suffix == u"m":

88

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

89

elif suffix == u"h":

90

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

91

elif suffix == u"w":

92

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

93

else:

94

raise ValueError

95

except (ValueError, IndexError):

96

raise ValueError

97

timevalue += delta

98

return timevalue

99

100

def print_clients(clients):

101

def valuetostring(value, keyword):

102

if type(value) is dbus.Boolean:

103

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

104

if keyword in (u"timeout", u"interval"):

105

return milliseconds_to_string(value)

106

return unicode(value)

107

108

# Create format string to print table rows

109

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

110

max(len(tablewords[key]),

111

max(len(valuetostring(client[key], key))

112

for client in

113

clients))

114

for key in keywords)

115

# Print header line

116

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

117

for client in clients:

118

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

119

for key in keywords)

120

121

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

122

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

123

help="Print all fields")

124

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

125

help="Enable client")

126

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

127

help="disable client")

128

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

129

help="Bump timeout for client")

130

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

131

help="Start checker for client")

132

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

133

help="Stop checker for client")

134

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

135

help="Check if client is enabled")

136

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

137

help="Remove client")

138

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

139

help="Set checker command for client")

140

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

141

help="Set timeout for client")

142

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

143

help="Set checker interval for client")

144

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

145

help="Set host for client")

146

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

147

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

148

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

149

help="Approve any current client request")

150

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

151

help="Deny any current client request")

152

options, client_names = parser.parse_args()

153

154

# Compile list of clients to process

155

clients=[]

156

for name in client_names:

157

for path, client in mandos_clients.iteritems():

158

if client['name'] == name:

159

client_objc = bus.get_object(busname, path)

160

clients.append(client_objc)

161

break

162

else:

163

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

425

164

sys.exit(1)

426

def is_enabled(self, client, properties):

427

return bool(properties["Enabled"])

428

429

class RemoveCmd(Command):

430

def run_on_one_client(self, client, properties):

431

self.mandos.RemoveClient(client.__dbus_object_path__)

432

433

class ApproveCmd(Command):

434

def run_on_one_client(self, client, properties):

435

client.Approve(dbus.Boolean(True),

436

dbus_interface=client_interface)

437

438

class DenyCmd(Command):

439

def run_on_one_client(self, client, properties):

440

client.Approve(dbus.Boolean(False),

441

dbus_interface=client_interface)

442

443

class EnableCmd(PropertyCmd):

444

property = "Enabled"

445

value_to_set = dbus.Boolean(True)

446

447

class DisableCmd(PropertyCmd):

448

property = "Enabled"

449

value_to_set = dbus.Boolean(False)

450

451

class BumpTimeoutCmd(PropertyCmd):

452

property = "LastCheckedOK"

453

value_to_set = ""

454

455

class StartCheckerCmd(PropertyCmd):

456

property = "CheckerRunning"

457

value_to_set = dbus.Boolean(True)

458

459

class StopCheckerCmd(PropertyCmd):

460

property = "CheckerRunning"

461

value_to_set = dbus.Boolean(False)

462

463

class ApproveByDefaultCmd(PropertyCmd):

464

property = "ApprovedByDefault"

465

value_to_set = dbus.Boolean(True)

466

467

class DenyByDefaultCmd(PropertyCmd):

468

property = "ApprovedByDefault"

469

value_to_set = dbus.Boolean(False)

470

471

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

472

property = "Checker"

473

474

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

475

property = "Host"

476

477

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

478

property = "Secret"

479

480

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

481

property = "Timeout"

482

483

class SetExtendedTimeoutCmd(PropertyCmd,

484

MillisecondsValueArgumentMixIn):

485

property = "ExtendedTimeout"

486

487

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

488

property = "Interval"

489

490

class SetApprovalDelayCmd(PropertyCmd,

491

MillisecondsValueArgumentMixIn):

492

property = "ApprovalDelay"

493

494

class SetApprovalDurationCmd(PropertyCmd,

495

MillisecondsValueArgumentMixIn):

496

property = "ApprovalDuration"

497

498

def has_actions(options):

499

return any((options.enable,

500

options.disable,

501

options.bump_timeout,

502

options.start_checker,

503

options.stop_checker,

504

options.is_enabled,

505

options.remove,

506

options.checker is not None,

507

options.timeout is not None,

508

options.extended_timeout is not None,

509

options.interval is not None,

510

options.approved_by_default is not None,

511

options.approval_delay is not None,

512

options.approval_duration is not None,

513

options.host is not None,

514

options.secret is not None,

515

options.approve,

516

options.deny))

517

518

def add_command_line_options(parser):

519

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

520

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

521

help="show version number and exit")

522

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

523

help="Select all clients")

524

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

525

help="Print all fields")

526

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

527

help="Dump client data in JSON format")

528

enable_disable = parser.add_mutually_exclusive_group()

529

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

530

help="Enable client")

531

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

532

action="store_true",

533

help="disable client")

534

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

535

help="Bump timeout for client")

536

start_stop_checker = parser.add_mutually_exclusive_group()

537

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

538

action="store_true",

539

help="Start checker for client")

540

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

541

action="store_true",

542

help="Stop checker for client")

543

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

544

help="Check if client is enabled")

545

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

546

help="Remove client")

547

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

548

help="Set checker command for client")

549

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

550

help="Set timeout for client")

551

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

552

help="Set extended timeout for client")

553

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

554

help="Set checker interval for client")

555

approve_deny_default = parser.add_mutually_exclusive_group()

556

approve_deny_default.add_argument(

557

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

558

default=None, dest="approved_by_default",

559

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

560

approve_deny_default.add_argument(

561

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

562

dest="approved_by_default",

563

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

564

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

565

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

566

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

567

help="Set duration of one client approval")

568

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

569

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

570

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

571

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

572

approve_deny = parser.add_mutually_exclusive_group()

573

approve_deny.add_argument(

574

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

575

help="Approve any current client request")

576

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

577

help="Deny any current client request")

578

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

579

help="Run self-test")

580

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

581

582

583

def commands_from_options(options):

584

585

commands = []

586

587

if options.dump_json:

588

commands.append(DumpJSONCmd())

589

165

166

if not clients and mandos_clients.values():

167

keywords = defaultkeywords

168

if options.all:

169

keywords = ('name', 'enabled', 'timeout', 'last_checked_ok',

170

'created', 'interval', 'host', 'fingerprint',

171

'checker_running', 'last_enabled', 'checker')

172

print_clients(mandos_clients.values())

173

174

# Process each client in the list by all selected options

175

for client in clients:

176

if options.remove:

177

mandos_serv.RemoveClient(client.__dbus_object_path__)

590

178

if options.enable:

591

commands.append(EnableCmd())

592

179

client.Enable(dbus_interface=client_interface)

593

180

if options.disable:

594

commands.append(DisableCmd())

595

181

client.Disable(dbus_interface=client_interface)

596

182

if options.bump_timeout:

597

commands.append(BumpTimeoutCmd())

598

183

client.CheckedOK(dbus_interface=client_interface)

599

184

if options.start_checker:

600

commands.append(StartCheckerCmd())

601

185

client.StartChecker(dbus_interface=client_interface)

602

186

if options.stop_checker:

603

commands.append(StopCheckerCmd())

604

187

client.StopChecker(dbus_interface=client_interface)

605

188

if options.is_enabled:

606

commands.append(IsEnabledCmd())

607

608

if options.remove:

609

commands.append(RemoveCmd())

610

611

if options.checker is not None:

612

commands.append(SetCheckerCmd())

613

614

if options.timeout is not None:

615

commands.append(SetTimeoutCmd(options.timeout))

616

617

if options.extended_timeout:

618

commands.append(

619

SetExtendedTimeoutCmd(options.extended_timeout))

620

621

if options.interval is not None:

622

command.append(SetIntervalCmd(options.interval))

623

624

if options.approved_by_default is not None:

625

if options.approved_by_default:

626

command.append(ApproveByDefaultCmd())

627

else:

628

command.append(DenyByDefaultCmd())

629

630

if options.approval_delay is not None:

631

command.append(SetApprovalDelayCmd(options.approval_delay))

632

633

if options.approval_duration is not None:

634

command.append(

635

SetApprovalDurationCmd(options.approval_duration))

636

637

if options.host is not None:

638

command.append(SetHostCmd(options.host))

639

640

if options.secret is not None:

641

command.append(SetSecretCmd(options.secret))

642

189

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

190

u"enabled",

191

dbus_interface=dbus.PROPERTIES_IFACE)

192

else 1)

193

if options.checker:

194

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

195

dbus_interface=dbus.PROPERTIES_IFACE)

196

if options.host:

197

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

198

dbus_interface=dbus.PROPERTIES_IFACE)

199

if options.interval:

200

client.Set(client_interface, u"interval",

201

timedelta_to_milliseconds

202

(string_to_delta(options.interval)),

203

dbus_interface=dbus.PROPERTIES_IFACE)

204

if options.timeout:

205

client.Set(client_interface, u"timeout",

206

timedelta_to_milliseconds(string_to_delta

207

(options.timeout)),

208

dbus_interface=dbus.PROPERTIES_IFACE)

209

if options.secret:

210

client.Set(client_interface, u"secret",

211

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

212

dbus_interface=dbus.PROPERTIES_IFACE)

643

213

if options.approve:

644

commands.append(ApproveCmd())

645

214

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

646

215

if options.deny:

647

commands.append(DenyCmd())

648

649

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

650

# optionally verbosely

651

if not commands:

652

commands.append(PrintTableCmd(verbose=options.verbose))

653

654

return commands

655

656

657

def main():

658

parser = argparse.ArgumentParser()

659

660

add_command_line_options(parser)

661

662

options = parser.parse_args()

663

664

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

665

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

666

if options.verbose and has_actions(options):

667

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

668

if options.dump_json and (options.verbose

669

or has_actions(options)):

670

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

671

if options.all and not has_actions(options):

672

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

673

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

674

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

675

676

clientnames = options.client

677

678

try:

679

bus = dbus.SystemBus()

680

mandos_dbus_objc = bus.get_object(busname, server_path)

681

except dbus.exceptions.DBusException:

682

log.critical("Could not connect to Mandos server")

683

sys.exit(1)

684

685

mandos_serv = dbus.Interface(mandos_dbus_objc,

686

dbus_interface=server_interface)

687

mandos_serv_object_manager = dbus.Interface(

688

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

689

690

# Filter out log message from dbus module

691

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

692

class NullFilter(logging.Filter):

693

def filter(self, record):

694

return False

695

dbus_filter = NullFilter()

696

try:

697

dbus_logger.addFilter(dbus_filter)

698

mandos_clients = {path: ifs_and_props[client_interface]

699

for path, ifs_and_props in

700

mandos_serv_object_manager

701

.GetManagedObjects().items()

702

if client_interface in ifs_and_props}

703

except dbus.exceptions.DBusException as e:

704

log.critical("Failed to access Mandos server through D-Bus:"

705

"\n%s", e)

706

sys.exit(1)

707

finally:

708

# restore dbus logger

709

dbus_logger.removeFilter(dbus_filter)

710

711

# Compile dict of (clients: properties) to process

712

clients = {}

713

714

if not clientnames:

715

clients = {bus.get_object(busname, path): properties

716

for path, properties in mandos_clients.items()}

717

else:

718

for name in clientnames:

719

for path, client in mandos_clients.items():

720

if client["Name"] == name:

721

client_objc = bus.get_object(busname, path)

722

clients[client_objc] = client

723

break

724

else:

725

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

726

sys.exit(1)

727

728

# Run all commands on clients

729

commands = commands_from_options(options)

730

for command in commands:

731

command.run(mandos_serv, clients)

732

733

734

class Test_milliseconds_to_string(unittest.TestCase):

735

def test_all(self):

736

self.assertEqual(milliseconds_to_string(93785000),

737

"1T02:03:05")

738

def test_no_days(self):

739

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

740

def test_all_zero(self):

741

self.assertEqual(milliseconds_to_string(0), "00:00:00")

742

def test_no_fractional_seconds(self):

743

self.assertEqual(milliseconds_to_string(400), "00:00:00")

744

self.assertEqual(milliseconds_to_string(900), "00:00:00")

745

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

746

747

class Test_string_to_delta(unittest.TestCase):

748

def test_handles_basic_rfc3339(self):

749

self.assertEqual(string_to_delta("PT0S"),

750

datetime.timedelta())

751

self.assertEqual(string_to_delta("P0D"),

752

datetime.timedelta())

753

self.assertEqual(string_to_delta("PT1S"),

754

datetime.timedelta(0, 1))

755

self.assertEqual(string_to_delta("PT2H"),

756

datetime.timedelta(0, 7200))

757

def test_falls_back_to_pre_1_6_1_with_warning(self):

758

# assertLogs only exists in Python 3.4

759

if hasattr(self, "assertLogs"):

760

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

761

value = string_to_delta("2h")

762

else:

763

class WarningFilter(logging.Filter):

764

"""Don't show, but record the presence of, warnings"""

765

def filter(self, record):

766

is_warning = record.levelno >= logging.WARNING

767

self.found = is_warning or getattr(self, "found",

768

False)

769

return not is_warning

770

warning_filter = WarningFilter()

771

log.addFilter(warning_filter)

772

try:

773

value = string_to_delta("2h")

774

finally:

775

log.removeFilter(warning_filter)

776

self.assertTrue(getattr(warning_filter, "found", False))

777

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

778

779

780

class TestCmd(unittest.TestCase):

781

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

782

def setUp(self):

783

testcase = self

784

class MockClient(object):

785

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

786

self.__dbus_object_path__ = "objpath_{}".format(name)

787

self.attributes = attributes

788

self.attributes["Name"] = name

789

self.calls = []

790

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

791

testcase.assertEqual(interface, client_interface)

792

testcase.assertEqual(dbus_interface,

793

dbus.PROPERTIES_IFACE)

794

self.attributes[property] = value

795

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

796

testcase.assertEqual(interface, client_interface)

797

testcase.assertEqual(dbus_interface,

798

dbus.PROPERTIES_IFACE)

799

return self.attributes[property]

800

def Approve(self, approve, dbus_interface):

801

testcase.assertEqual(dbus_interface, client_interface)

802

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

803

dbus_interface)))

804

self.client = MockClient(

805

"foo",

806

KeyID=("92ed150794387c03ce684574b1139a65"

807

"94a34f895daaaf09fd8ea90a27cddb12"),

808

Secret=b"secret",

809

Host="foo.example.org",

810

Enabled=dbus.Boolean(True),

811

Timeout=300000,

812

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

813

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

814

Interval=120000,

815

Fingerprint=("778827225BA7DE539C5A"

816

"7CFA59CFF7CDBD9A5920"),

817

CheckerRunning=dbus.Boolean(False),

818

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

819

ApprovalPending=dbus.Boolean(False),

820

ApprovedByDefault=dbus.Boolean(True),

821

LastApprovalRequest="",

822

ApprovalDelay=0,

823

ApprovalDuration=1000,

824

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

825

ExtendedTimeout=900000,

826

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

827

LastCheckerStatus=0)

828

self.other_client = MockClient(

829

"barbar",

830

KeyID=("0558568eedd67d622f5c83b35a115f79"

831

"6ab612cff5ad227247e46c2b020f441c"),

832

Secret=b"secretbar",

833

Host="192.0.2.3",

834

Enabled=dbus.Boolean(True),

835

Timeout=300000,

836

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

837

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

838

Interval=120000,

839

Fingerprint=("3E393AEAEFB84C7E89E2"

840

"F547B3A107558FCA3A27"),

841

CheckerRunning=dbus.Boolean(True),

842

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

843

ApprovalPending=dbus.Boolean(False),

844

ApprovedByDefault=dbus.Boolean(False),

845

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

846

ApprovalDelay=30000,

847

ApprovalDuration=1000,

848

Checker=":",

849

ExtendedTimeout=900000,

850

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

851

LastCheckerStatus=-2)

852

self.clients = collections.OrderedDict(

853

[

854

(self.client, self.client.attributes),

855

(self.other_client, self.other_client.attributes),

856

])

857

self.one_client = {self.client: self.client.attributes}

858

859

class TestPrintTableCmd(TestCmd):

860

def test_normal(self):

861

output = PrintTableCmd().output(self.clients)

862

expected_output = """

863

Name Enabled Timeout Last Successful Check

864

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

865

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

866

"""[1:-1]

867

self.assertEqual(output, expected_output)

868

def test_verbose(self):

869

output = PrintTableCmd(verbose=True).output(self.clients)

870

expected_output = """

871

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

872

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

873

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

874

"""[1:-1]

875

self.assertEqual(output, expected_output)

876

def test_one_client(self):

877

output = PrintTableCmd().output(self.one_client)

878

expected_output = """

879

Name Enabled Timeout Last Successful Check

880

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

881

"""[1:-1]

882

self.assertEqual(output, expected_output)

883

884

class TestDumpJSONCmd(TestCmd):

885

def setUp(self):

886

self.expected_json = {

887

"foo": {

888

"Name": "foo",

889

"KeyID": ("92ed150794387c03ce684574b1139a65"

890

"94a34f895daaaf09fd8ea90a27cddb12"),

891

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

892

"Enabled": True,

893

"Timeout": 300000,

894

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

895

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

896

"Interval": 120000,

897

"Fingerprint": ("778827225BA7DE539C5A"

898

"7CFA59CFF7CDBD9A5920"),

899

"CheckerRunning": False,

900

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

901

"ApprovalPending": False,

902

"ApprovedByDefault": True,

903

"LastApprovalRequest": "",

904

"ApprovalDelay": 0,

905

"ApprovalDuration": 1000,

906

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

907

"ExtendedTimeout": 900000,

908

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

909

"LastCheckerStatus": 0,

910

},

911

"barbar": {

912

"Name": "barbar",

913

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

914

"6ab612cff5ad227247e46c2b020f441c"),

915

"Host": "192.0.2.3",

916

"Enabled": True,

917

"Timeout": 300000,

918

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

919

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

920

"Interval": 120000,

921

"Fingerprint": ("3E393AEAEFB84C7E89E2"

922

"F547B3A107558FCA3A27"),

923

"CheckerRunning": True,

924

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

925

"ApprovalPending": False,

926

"ApprovedByDefault": False,

927

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

928

"ApprovalDelay": 30000,

929

"ApprovalDuration": 1000,

930

"Checker": ":",

931

"ExtendedTimeout": 900000,

932

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

933

"LastCheckerStatus": -2,

934

},

935

}

936

return super(TestDumpJSONCmd, self).setUp()

937

def test_normal(self):

938

json_data = json.loads(DumpJSONCmd().output(self.clients))

939

self.assertDictEqual(json_data, self.expected_json)

940

def test_one_client(self):

941

clients = self.one_client

942

json_data = json.loads(DumpJSONCmd().output(clients))

943

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

944

self.assertDictEqual(json_data, expected_json)

945

946

class TestIsEnabledCmd(TestCmd):

947

def test_is_enabled(self):

948

self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)

949

for client, properties in self.clients.items()))

950

def test_is_enabled_run_exits_successfully(self):

951

with self.assertRaises(SystemExit) as e:

952

IsEnabledCmd().run(None, self.one_client)

953

if e.exception.code is not None:

954

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

955

else:

956

self.assertIsNone(e.exception.code)

957

def test_is_enabled_run_exits_with_failure(self):

958

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

959

with self.assertRaises(SystemExit) as e:

960

IsEnabledCmd().run(None, self.one_client)

961

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

962

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

963

else:

964

self.assertIsNotNone(e.exception.code)

965

966

class TestRemoveCmd(TestCmd):

967

def test_remove(self):

968

class MockMandos(object):

969

def __init__(self):

970

self.calls = []

971

def RemoveClient(self, dbus_path):

972

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

973

mandos = MockMandos()

974

super(TestRemoveCmd, self).setUp()

975

RemoveCmd().run(mandos, self.clients)

976

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

977

for client in self.clients:

978

self.assertIn(("RemoveClient",

979

(client.__dbus_object_path__,)),

980

mandos.calls)

981

982

class TestApproveCmd(TestCmd):

983

def test_approve(self):

984

ApproveCmd().run(None, self.clients)

985

for client in self.clients:

986

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

987

client.calls)

988

989

class TestDenyCmd(TestCmd):

990

def test_deny(self):

991

DenyCmd().run(None, self.clients)

992

for client in self.clients:

993

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

994

client.calls)

995

996

class TestEnableCmd(TestCmd):

997

def test_enable(self):

998

for client in self.clients:

999

client.attributes["Enabled"] = False

1000

1001

EnableCmd().run(None, self.clients)

1002

1003

for client in self.clients:

1004

self.assertTrue(client.attributes["Enabled"])

1005

1006

class TestDisableCmd(TestCmd):

1007

def test_disable(self):

1008

DisableCmd().run(None, self.clients)

1009

1010

for client in self.clients:

1011

self.assertFalse(client.attributes["Enabled"])

1012

1013

class Unique(object):

1014

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

1015

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

1016

1017

class TestPropertyCmd(TestCmd):

1018

"""Abstract class for tests of PropertyCmd classes"""

1019

def runTest(self):

1020

if not hasattr(self, "command"):

1021

return

1022

values_to_get = getattr(self, "values_to_get",

1023

self.values_to_set)

1024

for value_to_set, value_to_get in zip(self.values_to_set,

1025

values_to_get):

1026

for client in self.clients:

1027

old_value = client.attributes[self.property]

1028

self.assertNotIsInstance(old_value, Unique)

1029

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

1030

self.run_command(value_to_set, self.clients)

1031

for client in self.clients:

1032

value = client.attributes[self.property]

1033

self.assertNotIsInstance(value, Unique)

1034

self.assertEqual(value, value_to_get)

1035

def run_command(self, value, clients):

1036

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

1037

1038

class TestBumpTimeoutCmd(TestPropertyCmd):

1039

command = BumpTimeoutCmd

1040

property = "LastCheckedOK"

1041

values_to_set = [""]

1042

1043

class TestStartCheckerCmd(TestPropertyCmd):

1044

command = StartCheckerCmd

1045

property = "CheckerRunning"

1046

values_to_set = [dbus.Boolean(True)]

1047

1048

class TestStopCheckerCmd(TestPropertyCmd):

1049

command = StopCheckerCmd

1050

property = "CheckerRunning"

1051

values_to_set = [dbus.Boolean(False)]

1052

1053

class TestApproveByDefaultCmd(TestPropertyCmd):

1054

command = ApproveByDefaultCmd

1055

property = "ApprovedByDefault"

1056

values_to_set = [dbus.Boolean(True)]

1057

1058

class TestDenyByDefaultCmd(TestPropertyCmd):

1059

command = DenyByDefaultCmd

1060

property = "ApprovedByDefault"

1061

values_to_set = [dbus.Boolean(False)]

1062

1063

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1064

"""Abstract class for tests of PropertyCmd classes using the

1065

ValueArgumentMixIn"""

1066

def runTest(self):

1067

if type(self) is TestValueArgumentPropertyCmd:

1068

return

1069

return super(TestValueArgumentPropertyCmd, self).runTest()

1070

def run_command(self, value, clients):

1071

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

1072

1073

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1074

command = SetCheckerCmd

1075

property = "Checker"

1076

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

1077

1078

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1079

command = SetHostCmd

1080

property = "Host"

1081

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

1082

1083

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1084

command = SetSecretCmd

1085

property = "Secret"

1086

values_to_set = [b"", b"secret"]

1087

1088

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1089

command = SetTimeoutCmd

1090

property = "Timeout"

1091

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1092

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

1093

1094

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1095

command = SetExtendedTimeoutCmd

1096

property = "ExtendedTimeout"

1097

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1098

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

1099

1100

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1101

command = SetIntervalCmd

1102

property = "Interval"

1103

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1104

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

1105

1106

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1107

command = SetApprovalDelayCmd

1108

property = "ApprovalDelay"

1109

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1110

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

1111

1112

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1113

command = SetApprovalDurationCmd

1114

property = "ApprovalDuration"

1115

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1116

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

1117

1118

class TestOptions(unittest.TestCase):

1119

def setUp(self):

1120

self.parser = argparse.ArgumentParser()

1121

add_command_line_options(self.parser)

1122

def assert_command_from_args(self, args, command_cls, **cmd_attrs):

1123

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

1124

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

1125

options = self.parser.parse_args(args)

1126

commands = commands_from_options(options)

1127

self.assertEqual(len(commands), 1)

1128

command = commands[0]

1129

self.assertIsInstance(command, command_cls)

1130

for key, value in cmd_attrs.items():

1131

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

1132

def test_default_is_show_table(self):

1133

self.assert_command_from_args([], PrintTableCmd,

1134

verbose=False)

1135

def test_show_table_verbose(self):

1136

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

1137

verbose=True)

1138

def test_enable(self):

1139

self.assert_command_from_args(["--enable"], EnableCmd)

1140

def test_disable(self):

1141

self.assert_command_from_args(["--disable"], DisableCmd)

1142

1143

1144

1145

def should_only_run_tests():

1146

parser = argparse.ArgumentParser(add_help=False)

1147

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

1148

args, unknown_args = parser.parse_known_args()

1149

run_tests = args.check

1150

if run_tests:

1151

# Remove --check argument from sys.argv

1152

sys.argv[1:] = unknown_args

1153

return run_tests

1154

1155

# Add all tests from doctest strings

1156

def load_tests(loader, tests, none):

1157

import doctest

1158

tests.addTests(doctest.DocTestSuite())

1159

return tests

1160

1161

if __name__ == "__main__":

1162

if should_only_run_tests():

1163

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

1164

unittest.main()

1165

else:

1166

main()

216

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