/mandos/trunk : revision 1074

To get this branch, use:

bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

Committer: Teddy Hogeborn
Date: 2019-03-16 04:51:11 UTC
Revision ID: teddy@recompile.se-20190316045111-63jv8d18i2tcakuw

mandos-ctl: Add comment

* mandos-ctl
(Test_get_managed_objects.test_logs_and_exits_on_dbus_error): Add a
comment.

files modified:
mandos-ctl

mandos-ctl.xml

Show diffs side-by-side

added added

removed removed

mandos-ctl

locale.setlocale(locale.LC_ALL, "")

domain = "se.recompile"

busname = domain + ".Mandos"

server_path = "/"

server_interface = domain + ".Mandos"

client_interface = domain + ".Mandos.Client"

dbus_busname_domain = "se.recompile"

dbus_busname = dbus_busname_domain + ".Mandos"

server_dbus_path = "/"

server_dbus_interface = dbus_busname_domain + ".Mandos"

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

del dbus_busname_domain

version = "1.8.3"

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

def milliseconds_to_string(ms):

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

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

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

hours=td.seconds // 3600,

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

seconds=td.seconds % 60))

def main():

parser = argparse.ArgumentParser()

add_command_line_options(parser)

options = parser.parse_args()

check_option_syntax(parser, options)

clientnames = options.client

if options.debug:

log.setLevel(logging.DEBUG)

bus = dbus.SystemBus()

mandos_dbus_object = get_mandos_dbus_object(bus)

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

102

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

115

if not clientnames:

116

clients = all_clients

117

else:

118

clients = {}

119

for name in clientnames:

120

for objpath, properties in all_clients.items():

121

if properties["Name"] == name:

122

clients[objpath] = properties

123

break

124

else:

125

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

126

sys.exit(1)

127

128

# Run all commands on clients

129

commands = commands_from_options(options)

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

133

134

def add_command_line_options(parser):

135

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

136

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

137

help="show version number and exit")

138

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

139

help="Select all clients")

140

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

141

help="Print all fields")

142

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

143

help="Dump client data in JSON format")

144

enable_disable = parser.add_mutually_exclusive_group()

145

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

146

help="Enable client")

147

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

148

action="store_true",

149

help="disable client")

150

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

151

help="Bump timeout for client")

152

start_stop_checker = parser.add_mutually_exclusive_group()

153

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

154

action="store_true",

155

help="Start checker for client")

156

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

157

action="store_true",

158

help="Stop checker for client")

159

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

160

help="Check if client is enabled")

161

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

162

help="Remove client")

163

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

164

help="Set checker command for client")

165

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

166

help="Set timeout for client")

167

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

168

help="Set extended timeout for client")

169

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

170

help="Set checker interval for client")

171

approve_deny_default = parser.add_mutually_exclusive_group()

172

approve_deny_default.add_argument(

173

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

174

default=None, dest="approved_by_default",

175

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

176

approve_deny_default.add_argument(

177

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

178

dest="approved_by_default",

179

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

180

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

181

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

182

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

183

help="Set duration of one client approval")

184

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

185

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

186

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

187

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

188

approve_deny = parser.add_mutually_exclusive_group()

189

approve_deny.add_argument(

190

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

191

help="Approve any current client request")

192

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

193

help="Deny any current client request")

194

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

195

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

196

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

197

help="Run self-test")

198

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

199

200

201

def string_to_delta(interval):

202

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

203

204

try:

205

return rfc3339_duration_to_delta(interval)

206

except ValueError as e:

207

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

208

' '.join(e.args))

209

return parse_pre_1_6_1_interval(interval)

210

211

212

def rfc3339_duration_to_delta(duration):

217

339

return value

218

340

219

341

220

def string_to_delta(interval):

221

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

222

223

try:

224

return rfc3339_duration_to_delta(interval)

225

except ValueError as e:

226

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

227

' '.join(e.args))

228

return parse_pre_1_6_1_interval(interval)

229

230

231

342

def parse_pre_1_6_1_interval(interval):

232

343

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

233

344

and return a datetime.timedelta

271

382

return value

272

383

273

384

274

## Classes for commands.

275

276

# Abstract classes first

385

def check_option_syntax(parser, options):

386

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

387

argparse"""

388

389

def has_actions(options):

390

return any((options.enable,

391

options.disable,

392

options.bump_timeout,

393

options.start_checker,

394

options.stop_checker,

395

options.is_enabled,

396

options.remove,

397

options.checker is not None,

398

options.timeout is not None,

399

options.extended_timeout is not None,

400

options.interval is not None,

401

options.approved_by_default is not None,

402

options.approval_delay is not None,

403

options.approval_duration is not None,

404

options.host is not None,

405

options.secret is not None,

406

options.approve,

407

options.deny))

408

409

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

410

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

411

if options.verbose and has_actions(options):

412

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

413

if options.dump_json and (options.verbose

414

or has_actions(options)):

415

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

416

if options.all and not has_actions(options):

417

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

418

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

419

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

420

if options.remove:

421

options.remove = False

422

if has_actions(options) and not options.deny:

423

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

424

options.remove = True

425

426

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

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

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

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

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

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

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

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

472

self.logger.removeFilter(self.nullfilter)

473

474

475

def commands_from_options(options):

476

477

commands = []

478

479

if options.is_enabled:

480

commands.append(IsEnabledCmd())

481

482

if options.approve:

483

commands.append(ApproveCmd())

484

485

if options.deny:

486

commands.append(DenyCmd())

487

488

if options.remove:

489

commands.append(RemoveCmd())

490

491

if options.dump_json:

492

commands.append(DumpJSONCmd())

493

494

if options.enable:

495

commands.append(EnableCmd())

496

497

if options.disable:

498

commands.append(DisableCmd())

499

500

if options.bump_timeout:

501

commands.append(BumpTimeoutCmd())

502

503

if options.start_checker:

504

commands.append(StartCheckerCmd())

505

506

if options.stop_checker:

507

commands.append(StopCheckerCmd())

508

509

if options.approved_by_default is not None:

510

if options.approved_by_default:

511

commands.append(ApproveByDefaultCmd())

512

else:

513

commands.append(DenyByDefaultCmd())

514

515

if options.checker is not None:

516

commands.append(SetCheckerCmd(options.checker))

517

518

if options.host is not None:

519

commands.append(SetHostCmd(options.host))

520

521

if options.secret is not None:

522

commands.append(SetSecretCmd(options.secret))

523

524

if options.timeout is not None:

525

commands.append(SetTimeoutCmd(options.timeout))

526

527

if options.extended_timeout:

528

commands.append(

529

SetExtendedTimeoutCmd(options.extended_timeout))

530

531

if options.interval is not None:

532

commands.append(SetIntervalCmd(options.interval))

533

534

if options.approval_delay is not None:

535

commands.append(SetApprovalDelayCmd(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

SetApprovalDurationCmd(options.approval_duration))

540

541

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

542

# optionally verbosely

543

if not commands:

544

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

545

546

return commands

547

548

277

549

class Command(object):

278

550

"""Abstract class for commands"""

279

def run(self, mandos, clients):

551

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

280

552

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

281

553

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

282

554

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

283

555

self.mandos = mandos

284

for client, properties in clients.items():

556

for clientpath, properties in clients.items():

557

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

558

dbus_busname, str(clientpath))

559

client = bus.get_object(dbus_busname, clientpath)

285

560

self.run_on_one_client(client, properties)

286

561

287

class PrintCmd(Command):

288

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

562

563

class IsEnabledCmd(Command):

564

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

565

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

566

if self.is_enabled(client, properties):

567

sys.exit(0)

568

sys.exit(1)

569

def is_enabled(self, client, properties):

570

return properties["Enabled"]

571

572

573

class ApproveCmd(Command):

574

def run_on_one_client(self, client, properties):

575

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

576

client.__dbus_object_path__, client_dbus_interface)

577

client.Approve(dbus.Boolean(True),

578

dbus_interface=client_dbus_interface)

579

580

581

class DenyCmd(Command):

582

def run_on_one_client(self, client, properties):

583

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

584

client.__dbus_object_path__, client_dbus_interface)

585

client.Approve(dbus.Boolean(False),

586

dbus_interface=client_dbus_interface)

587

588

589

class RemoveCmd(Command):

590

def run_on_one_client(self, client, properties):

591

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

592

server_dbus_path, server_dbus_interface,

593

str(client.__dbus_object_path__))

594

self.mandos.RemoveClient(client.__dbus_object_path__)

595

596

597

class OutputCmd(Command):

598

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

289

599

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

290

600

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

291

601

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

293

603

"LastApprovalRequest", "ApprovalDelay",

294

604

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

295

605

"Expires", "LastCheckerStatus")

296

def run(self, mandos, clients):

297

print(self.output(clients))

298

299

class PropertyCmd(Command):

300

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

301

def run_on_one_client(self, client, properties):

302

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

303

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

304

dbus_interface=dbus.PROPERTIES_IFACE)

305

306

class ValueArgumentMixIn(object):

307

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

308

def __init__(self, value):

309

self.value_to_set = value

310

311

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

312

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

313

milliseconds."""

314

@property

315

def value_to_set(self):

316

return self._vts

317

@value_to_set.setter

318

def value_to_set(self, value):

319

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

320

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

321

322

# Actual (non-abstract) command classes

323

324

class PrintTableCmd(PrintCmd):

606

607

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

608

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

609

610

def output(self, clients):

611

raise NotImplementedError()

612

613

614

class DumpJSONCmd(OutputCmd):

615

def output(self, clients):

616

data = {client["Name"]:

617

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

618

for key in self.all_keywords}

619

for client in clients}

620

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

621

622

@staticmethod

623

def dbus_boolean_to_bool(value):

624

if isinstance(value, dbus.Boolean):

625

value = bool(value)

626

return value

627

628

629

class PrintTableCmd(OutputCmd):

325

630

def __init__(self, verbose=False):

326

631

self.verbose = verbose

327

632

328

633

def output(self, clients):

329

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

634

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

635

"LastCheckedOK")

330

636

keywords = default_keywords

331

637

if self.verbose:

332

638

keywords = self.all_keywords

333

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

639

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

334

640

335

641

class TableOfClients(object):

336

642

tableheaders = {

356

662

"LastCheckerStatus": "Last Checker Status",

357

663

}

358

664

359

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

665

def __init__(self, clients, keywords):

360

666

self.clients = clients

361

667

self.keywords = keywords

362

if tableheaders is not None:

363

self.tableheaders = tableheaders

364

668

365

669

def __str__(self):

366

670

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

389

693

def string_from_client(self, client, key):

390

694

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

391

695

392

@staticmethod

393

def valuetostring(value, keyword):

696

@classmethod

697

def valuetostring(cls, value, keyword):

394

698

if isinstance(value, dbus.Boolean):

395

699

return "Yes" if value else "No"

396

700

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

397

701

"ApprovalDuration", "ExtendedTimeout"):

398

return milliseconds_to_string(value)

702

return cls.milliseconds_to_string(value)

399

703

return str(value)

400

704

401

705

def header_line(self, format_string):

406

710

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

407

711

for key in self.keywords})

408

712

409

410

411

class DumpJSONCmd(PrintCmd):

412

def output(self, clients):

413

data = {client["Name"]:

414

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

415

for key in self.all_keywords}

416

for client in clients.values()}

417

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

418

@staticmethod

419

def dbus_boolean_to_bool(value):

420

if isinstance(value, dbus.Boolean):

421

value = bool(value)

422

return value

423

424

class IsEnabledCmd(Command):

425

def run_on_one_client(self, client, properties):

426

if self.is_enabled(client, properties):

427

sys.exit(0)

428

sys.exit(1)

429

def is_enabled(self, client, properties):

430

return bool(properties["Enabled"])

431

432

class RemoveCmd(Command):

433

def run_on_one_client(self, client, properties):

434

self.mandos.RemoveClient(client.__dbus_object_path__)

435

436

class ApproveCmd(Command):

437

def run_on_one_client(self, client, properties):

438

client.Approve(dbus.Boolean(True),

439

dbus_interface=client_interface)

440

441

class DenyCmd(Command):

442

def run_on_one_client(self, client, properties):

443

client.Approve(dbus.Boolean(False),

444

dbus_interface=client_interface)

713

@staticmethod

714

def milliseconds_to_string(ms):

715

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

716

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

717

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

718

if td.days else "",

719

hours=td.seconds // 3600,

720

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

721

seconds=td.seconds % 60))

722

723

724

class PropertyCmd(Command):

725

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

726

727

def run_on_one_client(self, client, properties):

728

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

729

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

730

client.__dbus_object_path__,

731

dbus.PROPERTIES_IFACE, client_dbus_interface,

732

self.propname, self.value_to_set

733

if not isinstance(self.value_to_set, dbus.Boolean)

734

else bool(self.value_to_set))

735

client.Set(client_dbus_interface, self.propname,

736

self.value_to_set,

737

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

@property

740

def propname(self):

741

raise NotImplementedError()

742

445

743

446

744

class EnableCmd(PropertyCmd):

447

property = "Enabled"

745

propname = "Enabled"

448

746

value_to_set = dbus.Boolean(True)

449

747

748

450

749

class DisableCmd(PropertyCmd):

451

property = "Enabled"

750

propname = "Enabled"

452

751

value_to_set = dbus.Boolean(False)

453

752

753

454

754

class BumpTimeoutCmd(PropertyCmd):

455

property = "LastCheckedOK"

755

propname = "LastCheckedOK"

456

756

value_to_set = ""

457

757

758

458

759

class StartCheckerCmd(PropertyCmd):

459

property = "CheckerRunning"

760

propname = "CheckerRunning"

460

761

value_to_set = dbus.Boolean(True)

461

762

763

462

764

class StopCheckerCmd(PropertyCmd):

463

property = "CheckerRunning"

765

propname = "CheckerRunning"

464

766

value_to_set = dbus.Boolean(False)

465

767

768

466

769

class ApproveByDefaultCmd(PropertyCmd):

467

property = "ApprovedByDefault"

770

propname = "ApprovedByDefault"

468

771

value_to_set = dbus.Boolean(True)

469

772

773

470

774

class DenyByDefaultCmd(PropertyCmd):

471

property = "ApprovedByDefault"

775

propname = "ApprovedByDefault"

472

776

value_to_set = dbus.Boolean(False)

473

777

474

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

475

property = "Checker"

476

477

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

478

property = "Host"

479

480

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

778

779

class PropertyValueCmd(PropertyCmd):

780

"""Abstract class for PropertyCmd recieving a value as argument"""

781

def __init__(self, value):

782

self.value_to_set = value

783

784

785

class SetCheckerCmd(PropertyValueCmd):

786

propname = "Checker"

787

788

789

class SetHostCmd(PropertyValueCmd):

790

propname = "Host"

791

792

793

class SetSecretCmd(PropertyValueCmd):

794

propname = "Secret"

795

481

796

@property

482

797

def value_to_set(self):

483

798

return self._vts

799

484

800

@value_to_set.setter

485

801

def value_to_set(self, value):

486

802

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

487

803

self._vts = value.read()

488

804

value.close()

489

property = "Secret"

490

491

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

492

property = "Timeout"

493

494

class SetExtendedTimeoutCmd(PropertyCmd,

495

MillisecondsValueArgumentMixIn):

496

property = "ExtendedTimeout"

497

498

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

499

property = "Interval"

500

501

class SetApprovalDelayCmd(PropertyCmd,

502

MillisecondsValueArgumentMixIn):

503

property = "ApprovalDelay"

504

505

class SetApprovalDurationCmd(PropertyCmd,

506

MillisecondsValueArgumentMixIn):

507

property = "ApprovalDuration"

508

509

def add_command_line_options(parser):

510

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

511

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

512

help="show version number and exit")

513

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

514

help="Select all clients")

515

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

516

help="Print all fields")

517

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

518

help="Dump client data in JSON format")

519

enable_disable = parser.add_mutually_exclusive_group()

520

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

521

help="Enable client")

522

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

523

action="store_true",

524

help="disable client")

525

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

526

help="Bump timeout for client")

527

start_stop_checker = parser.add_mutually_exclusive_group()

528

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

529

action="store_true",

530

help="Start checker for client")

531

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

532

action="store_true",

533

help="Stop checker for client")

534

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

535

help="Check if client is enabled")

536

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

537

help="Remove client")

538

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

539

help="Set checker command for client")

540

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

541

help="Set timeout for client")

542

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

543

help="Set extended timeout for client")

544

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

545

help="Set checker interval for client")

546

approve_deny_default = parser.add_mutually_exclusive_group()

547

approve_deny_default.add_argument(

548

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

549

default=None, dest="approved_by_default",

550

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

551

approve_deny_default.add_argument(

552

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

553

dest="approved_by_default",

554

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

555

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

556

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

557

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

558

help="Set duration of one client approval")

559

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

560

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

561

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

562

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

563

approve_deny = parser.add_mutually_exclusive_group()

564

approve_deny.add_argument(

565

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

566

help="Approve any current client request")

567

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

568

help="Deny any current client request")

569

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

570

help="Run self-test")

571

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

572

573

574

def commands_from_options(options):

575

576

commands = []

577

578

if options.dump_json:

579

commands.append(DumpJSONCmd())

580

581

if options.enable:

582

commands.append(EnableCmd())

583

584

if options.disable:

585

commands.append(DisableCmd())

586

587

if options.bump_timeout:

588

commands.append(BumpTimeoutCmd())

589

590

if options.start_checker:

591

commands.append(StartCheckerCmd())

592

593

if options.stop_checker:

594

commands.append(StopCheckerCmd())

595

596

if options.is_enabled:

597

commands.append(IsEnabledCmd())

598

599

if options.remove:

600

commands.append(RemoveCmd())

601

602

if options.checker is not None:

603

commands.append(SetCheckerCmd(options.checker))

604

605

if options.timeout is not None:

606

commands.append(SetTimeoutCmd(options.timeout))

607

608

if options.extended_timeout:

609

commands.append(

610

SetExtendedTimeoutCmd(options.extended_timeout))

611

612

if options.interval is not None:

613

commands.append(SetIntervalCmd(options.interval))

614

615

if options.approved_by_default is not None:

616

if options.approved_by_default:

617

commands.append(ApproveByDefaultCmd())

618

else:

619

commands.append(DenyByDefaultCmd())

620

621

if options.approval_delay is not None:

622

commands.append(SetApprovalDelayCmd(options.approval_delay))

623

624

if options.approval_duration is not None:

625

commands.append(

626

SetApprovalDurationCmd(options.approval_duration))

627

628

if options.host is not None:

629

commands.append(SetHostCmd(options.host))

630

631

if options.secret is not None:

632

commands.append(SetSecretCmd(options.secret))

633

634

if options.approve:

635

commands.append(ApproveCmd())

636

637

if options.deny:

638

commands.append(DenyCmd())

639

640

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

641

# optionally verbosely

642

if not commands:

643

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

644

645

return commands

646

647

648

def check_option_syntax(parser, options):

649

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

650

argparse"""

651

652

def has_actions(options):

653

return any((options.enable,

654

options.disable,

655

options.bump_timeout,

656

options.start_checker,

657

options.stop_checker,

658

options.is_enabled,

659

options.remove,

660

options.checker is not None,

661

options.timeout is not None,

662

options.extended_timeout is not None,

663

options.interval is not None,

664

options.approved_by_default is not None,

665

options.approval_delay is not None,

666

options.approval_duration is not None,

667

options.host is not None,

668

options.secret is not None,

669

options.approve,

670

options.deny))

671

672

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

673

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

674

if options.verbose and has_actions(options):

675

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

676

if options.dump_json and (options.verbose

677

or has_actions(options)):

678

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

679

if options.all and not has_actions(options):

680

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

681

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

682

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

683

684

685

def main():

686

parser = argparse.ArgumentParser()

687

688

add_command_line_options(parser)

689

690

options = parser.parse_args()

691

692

check_option_syntax(parser, options)

693

694

clientnames = options.client

695

696

try:

697

bus = dbus.SystemBus()

698

mandos_dbus_objc = bus.get_object(busname, server_path)

699

except dbus.exceptions.DBusException:

700

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

701

sys.exit(1)

702

703

mandos_serv = dbus.Interface(mandos_dbus_objc,

704

dbus_interface=server_interface)

705

mandos_serv_object_manager = dbus.Interface(

706

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

707

708

# Filter out log message from dbus module

709

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

710

class NullFilter(logging.Filter):

711

def filter(self, record):

712

return False

713

dbus_filter = NullFilter()

714

try:

715

dbus_logger.addFilter(dbus_filter)

716

mandos_clients = {path: ifs_and_props[client_interface]

717

for path, ifs_and_props in

718

mandos_serv_object_manager

719

.GetManagedObjects().items()

720

if client_interface in ifs_and_props}

721

except dbus.exceptions.DBusException as e:

722

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

723

"\n%s", e)

724

sys.exit(1)

725

finally:

726

# restore dbus logger

727

dbus_logger.removeFilter(dbus_filter)

728

729

# Compile dict of (clients: properties) to process

730

clients = {}

731

732

if not clientnames:

733

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

734

for path, properties in mandos_clients.items()}

735

else:

736

for name in clientnames:

737

for path, client in mandos_clients.items():

738

if client["Name"] == name:

739

client_objc = bus.get_object(busname, path)

740

clients[client_objc] = client

741

break

742

else:

743

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

744

sys.exit(1)

745

746

# Run all commands on clients

747

commands = commands_from_options(options)

748

for command in commands:

749

command.run(mandos_serv, clients)

805

806

807

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

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

809

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

810

811

@property

812

def value_to_set(self):

813

return self._vts

814

815

@value_to_set.setter

816

def value_to_set(self, value):

817

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

818

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

819

820

821

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

822

propname = "Timeout"

823

824

825

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

826

propname = "ExtendedTimeout"

827

828

829

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

830

propname = "Interval"

831

832

833

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

834

propname = "ApprovalDelay"

835

836

837

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

838

propname = "ApprovalDuration"

839

750

840

751

841

752

class Test_milliseconds_to_string(unittest.TestCase):

753

def test_all(self):

754

self.assertEqual(milliseconds_to_string(93785000),

755

"1T02:03:05")

756

def test_no_days(self):

757

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

758

def test_all_zero(self):

759

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

760

def test_no_fractional_seconds(self):

761

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

762

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

763

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

764

765

class Test_string_to_delta(unittest.TestCase):

842

class TestCaseWithAssertLogs(unittest.TestCase):

843

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

844

845

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

846

@contextlib.contextmanager

847

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

848

capturing_handler = self.CapturingLevelHandler(level)

849

old_level = logger.level

850

old_propagate = logger.propagate

851

logger.addHandler(capturing_handler)

852

logger.setLevel(level)

853

logger.propagate = False

854

try:

855

yield capturing_handler.watcher

856

finally:

857

logger.propagate = old_propagate

858

logger.removeHandler(capturing_handler)

859

logger.setLevel(old_level)

860

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

861

862

863

class CapturingLevelHandler(logging.Handler):

864

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

865

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

866

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

867

def emit(self, record):

868

self.watcher.records.append(record)

869

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

870

871

LoggingWatcher = collections.namedtuple("LoggingWatcher",

872

("records",

873

"output"))

874

875

class Test_string_to_delta(TestCaseWithAssertLogs):

766

876

def test_handles_basic_rfc3339(self):

767

877

self.assertEqual(string_to_delta("PT0S"),

768

878

datetime.timedelta())

772

882

datetime.timedelta(0, 1))

773

883

self.assertEqual(string_to_delta("PT2H"),

774

884

datetime.timedelta(0, 7200))

885

775

886

def test_falls_back_to_pre_1_6_1_with_warning(self):

776

# assertLogs only exists in Python 3.4

777

if hasattr(self, "assertLogs"):

778

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

779

value = string_to_delta("2h")

780

else:

781

class WarningFilter(logging.Filter):

782

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

783

def filter(self, record):

784

is_warning = record.levelno >= logging.WARNING

785

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

786

False)

787

return not is_warning

788

warning_filter = WarningFilter()

789

log.addFilter(warning_filter)

790

try:

791

value = string_to_delta("2h")

792

finally:

793

log.removeFilter(warning_filter)

794

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

887

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

888

value = string_to_delta("2h")

795

889

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

796

890

797

891

892

class Test_check_option_syntax(unittest.TestCase):

893

def setUp(self):

894

self.parser = argparse.ArgumentParser()

895

add_command_line_options(self.parser)

896

897

def test_actions_requires_client_or_all(self):

898

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

899

options = self.parser.parse_args()

900

setattr(options, action, value)

901

with self.assertParseError():

902

self.check_option_syntax(options)

903

904

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

905

# check_option_syntax()

906

actions = {

907

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

908

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

909

# never used

910

"enable": True,

911

"disable": True,

912

"bump_timeout": True,

913

"start_checker": True,

914

"stop_checker": True,

915

"is_enabled": True,

916

"remove": True,

917

"checker": "x",

918

"timeout": datetime.timedelta(),

919

"extended_timeout": datetime.timedelta(),

920

"interval": datetime.timedelta(),

921

"approved_by_default": True,

922

"approval_delay": datetime.timedelta(),

923

"approval_duration": datetime.timedelta(),

924

"host": "x",

925

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

926

"approve": True,

927

"deny": True,

928

}

929

930

@contextlib.contextmanager

931

def assertParseError(self):

932

with self.assertRaises(SystemExit) as e:

933

with self.temporarily_suppress_stderr():

934

yield

935

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

936

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

937

# /argparse.html#exiting-methods

938

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

939

940

@staticmethod

941

@contextlib.contextmanager

942

def temporarily_suppress_stderr():

943

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

944

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

945

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

946

os.close(null)

947

try:

948

yield

949

finally:

950

# restore stderr

951

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

952

os.close(stderrcopy)

953

954

def check_option_syntax(self, options):

955

check_option_syntax(self.parser, options)

956

957

def test_actions_conflicts_with_verbose(self):

958

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

959

options = self.parser.parse_args()

960

setattr(options, action, value)

961

options.verbose = True

962

with self.assertParseError():

963

self.check_option_syntax(options)

964

965

def test_dump_json_conflicts_with_verbose(self):

966

options = self.parser.parse_args()

967

options.dump_json = True

968

options.verbose = True

969

with self.assertParseError():

970

self.check_option_syntax(options)

971

972

def test_dump_json_conflicts_with_action(self):

973

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

974

options = self.parser.parse_args()

975

setattr(options, action, value)

976

options.dump_json = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

def test_all_can_not_be_alone(self):

981

options = self.parser.parse_args()

982

options.all = True

983

with self.assertParseError():

984

self.check_option_syntax(options)

985

986

def test_all_is_ok_with_any_action(self):

987

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

988

options = self.parser.parse_args()

989

setattr(options, action, value)

990

options.all = True

991

self.check_option_syntax(options)

992

993

def test_is_enabled_fails_without_client(self):

994

options = self.parser.parse_args()

995

options.is_enabled = True

996

with self.assertParseError():

997

self.check_option_syntax(options)

998

999

def test_is_enabled_works_with_one_client(self):

1000

options = self.parser.parse_args()

1001

options.is_enabled = True

1002

options.client = ["foo"]

1003

self.check_option_syntax(options)

1004

1005

def test_is_enabled_fails_with_two_clients(self):

1006

options = self.parser.parse_args()

1007

options.is_enabled = True

1008

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

1009

with self.assertParseError():

1010

self.check_option_syntax(options)

1011

1012

def test_remove_can_only_be_combined_with_action_deny(self):

1013

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

1014

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

1015

continue

1016

options = self.parser.parse_args()

1017

setattr(options, action, value)

1018

options.all = True

1019

options.remove = True

1020

with self.assertParseError():

1021

self.check_option_syntax(options)

1022

1023

1024

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1025

def test_calls_and_returns_get_object_on_bus(self):

1026

class MockBus(object):

1027

called = False

1028

def get_object(mockbus_self, busname, dbus_path):

1029

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

1030

# this MockBus instance is in "mockbus_self".

1031

self.assertEqual(busname, dbus_busname)

1032

self.assertEqual(dbus_path, server_dbus_path)

1033

mockbus_self.called = True

1034

return mockbus_self

1035

1036

mockbus = get_mandos_dbus_object(bus=MockBus())

1037

self.assertIsInstance(mockbus, MockBus)

1038

self.assertTrue(mockbus.called)

1039

1040

def test_logs_and_exits_on_dbus_error(self):

1041

class MockBusFailing(object):

1042

def get_object(self, busname, dbus_path):

1043

raise dbus.exceptions.DBusException("Test")

1044

1045

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

1046

with self.assertRaises(SystemExit) as e:

1047

bus = get_mandos_dbus_object(bus=MockBusFailing())

1048

1049

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

1050

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

1051

else:

1052

self.assertIsNotNone(e.exception.code)

1053

1054

1055

class Test_get_managed_objects(TestCaseWithAssertLogs):

1056

def test_calls_and_returns_GetManagedObjects(self):

1057

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

1058

class MockObjectManager(object):

1059

def GetManagedObjects(self):

1060

return managed_objects

1061

retval = get_managed_objects(MockObjectManager())

1062

self.assertDictEqual(managed_objects, retval)

1063

1064

def test_logs_and_exits_on_dbus_error(self):

1065

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

1066

1067

class MockObjectManagerFailing(object):

1068

def GetManagedObjects(self):

1069

dbus_logger.error("Test")

1070

raise dbus.exceptions.DBusException("Test")

1071

1072

class CountingHandler(logging.Handler):

1073

count = 0

1074

def emit(self, record):

1075

self.count += 1

1076

1077

counting_handler = CountingHandler()

1078

1079

dbus_logger.addHandler(counting_handler)

1080

1081

try:

1082

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

1083

with self.assertRaises(SystemExit) as e:

1084

get_managed_objects(MockObjectManagerFailing())

1085

finally:

1086

dbus_logger.removeFilter(counting_handler)

1087

1088

# Make sure the dbus logger was suppressed

1089

self.assertEqual(counting_handler.count, 0)

1090

1091

# Test that the dbus_logger still works

1092

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

1093

dbus_logger.error("Test")

1094

1095

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

1096

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

1097

else:

1098

self.assertIsNotNone(e.exception.code)

1099

1100

1101

class Test_commands_from_options(unittest.TestCase):

1102

def setUp(self):

1103

self.parser = argparse.ArgumentParser()

1104

add_command_line_options(self.parser)

1105

1106

def test_is_enabled(self):

1107

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

1108

IsEnabledCmd)

1109

1110

def assert_command_from_args(self, args, command_cls,

1111

**cmd_attrs):

1112

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

1113

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

1114

options = self.parser.parse_args(args)

1115

check_option_syntax(self.parser, options)

1116

commands = commands_from_options(options)

1117

self.assertEqual(len(commands), 1)

1118

command = commands[0]

1119

self.assertIsInstance(command, command_cls)

1120

for key, value in cmd_attrs.items():

1121

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

1122

1123

def test_is_enabled_short(self):

1124

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

1125

1126

def test_approve(self):

1127

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

1128

ApproveCmd)

1129

1130

def test_approve_short(self):

1131

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

1132

1133

def test_deny(self):

1134

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

1135

1136

def test_deny_short(self):

1137

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

1138

1139

def test_remove(self):

1140

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

1141

RemoveCmd)

1142

1143

def test_deny_before_remove(self):

1144

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

1145

"foo"])

1146

check_option_syntax(self.parser, options)

1147

commands = commands_from_options(options)

1148

self.assertEqual(len(commands), 2)

1149

self.assertIsInstance(commands[0], DenyCmd)

1150

self.assertIsInstance(commands[1], RemoveCmd)

1151

1152

def test_deny_before_remove_reversed(self):

1153

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

1154

"--all"])

1155

check_option_syntax(self.parser, options)

1156

commands = commands_from_options(options)

1157

self.assertEqual(len(commands), 2)

1158

self.assertIsInstance(commands[0], DenyCmd)

1159

self.assertIsInstance(commands[1], RemoveCmd)

1160

1161

def test_remove_short(self):

1162

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

1163

1164

def test_dump_json(self):

1165

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

1166

1167

def test_enable(self):

1168

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

1169

1170

def test_enable_short(self):

1171

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

1172

1173

def test_disable(self):

1174

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

1175

DisableCmd)

1176

1177

def test_disable_short(self):

1178

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

1179

1180

def test_bump_timeout(self):

1181

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

1182

BumpTimeoutCmd)

1183

1184

def test_bump_timeout_short(self):

1185

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

1186

1187

def test_start_checker(self):

1188

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

1189

StartCheckerCmd)

1190

1191

def test_stop_checker(self):

1192

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

1193

StopCheckerCmd)

1194

1195

def test_approve_by_default(self):

1196

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

1197

ApproveByDefaultCmd)

1198

1199

def test_deny_by_default(self):

1200

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

1201

DenyByDefaultCmd)

1202

1203

def test_checker(self):

1204

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

1205

SetCheckerCmd, value_to_set=":")

1206

1207

def test_checker_empty(self):

1208

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

1209

SetCheckerCmd, value_to_set="")

1210

1211

def test_checker_short(self):

1212

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

1213

SetCheckerCmd, value_to_set=":")

1214

1215

def test_host(self):

1216

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

1217

"foo"], SetHostCmd,

1218

value_to_set="foo.example.org")

1219

1220

def test_host_short(self):

1221

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

1222

"foo"], SetHostCmd,

1223

value_to_set="foo.example.org")

1224

1225

def test_secret_devnull(self):

1226

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

1227

"foo"], SetSecretCmd,

1228

value_to_set=b"")

1229

1230

def test_secret_tempfile(self):

1231

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

1232

value = b"secret\0xyzzy\nbar"

1233

f.write(value)

1234

f.seek(0)

1235

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

1236

"foo"], SetSecretCmd,

1237

value_to_set=value)

1238

1239

def test_secret_devnull_short(self):

1240

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

1241

SetSecretCmd, value_to_set=b"")

1242

1243

def test_secret_tempfile_short(self):

1244

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

1245

value = b"secret\0xyzzy\nbar"

1246

f.write(value)

1247

f.seek(0)

1248

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

1249

SetSecretCmd,

1250

value_to_set=value)

1251

1252

def test_timeout(self):

1253

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

1254

SetTimeoutCmd,

1255

value_to_set=300000)

1256

1257

def test_timeout_short(self):

1258

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

1259

SetTimeoutCmd,

1260

value_to_set=300000)

1261

1262

def test_extended_timeout(self):

1263

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

1264

"foo"],

1265

SetExtendedTimeoutCmd,

1266

value_to_set=900000)

1267

1268

def test_interval(self):

1269

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

1270

SetIntervalCmd,

1271

value_to_set=120000)

1272

1273

def test_interval_short(self):

1274

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

1275

SetIntervalCmd,

1276

value_to_set=120000)

1277

1278

def test_approval_delay(self):

1279

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

1280

"foo"], SetApprovalDelayCmd,

1281

value_to_set=30000)

1282

1283

def test_approval_duration(self):

1284

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

1285

"foo"], SetApprovalDurationCmd,

1286

value_to_set=1000)

1287

1288

def test_print_table(self):

1289

self.assert_command_from_args([], PrintTableCmd,

1290

verbose=False)

1291

1292

def test_print_table_verbose(self):

1293

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

1294

verbose=True)

1295

1296

def test_print_table_verbose_short(self):

1297

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

1298

verbose=True)

1299

1300

798

1301

class TestCmd(unittest.TestCase):

799

1302

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

1303

800

1304

def setUp(self):

801

1305

testcase = self

802

1306

class MockClient(object):

803

1307

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

804

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

1308

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

805

1309

self.attributes = attributes

806

1310

self.attributes["Name"] = name

807

1311

self.calls = []

808

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

809

testcase.assertEqual(interface, client_interface)

810

testcase.assertEqual(dbus_interface,

811

dbus.PROPERTIES_IFACE)

812

self.attributes[property] = value

813

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

814

testcase.assertEqual(interface, client_interface)

815

testcase.assertEqual(dbus_interface,

816

dbus.PROPERTIES_IFACE)

817

return self.attributes[property]

1312

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

1313

testcase.assertEqual(interface, client_dbus_interface)

1314

testcase.assertEqual(dbus_interface,

1315

dbus.PROPERTIES_IFACE)

1316

self.attributes[propname] = value

1317

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

1318

testcase.assertEqual(interface, client_dbus_interface)

1319

testcase.assertEqual(dbus_interface,

1320

dbus.PROPERTIES_IFACE)

1321

return self.attributes[propname]

818

1322

def Approve(self, approve, dbus_interface):

819

testcase.assertEqual(dbus_interface, client_interface)

1323

testcase.assertEqual(dbus_interface,

1324

client_dbus_interface)

820

1325

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

821

1326

dbus_interface)))

822

1327

self.client = MockClient(

862

1367

ApprovedByDefault=dbus.Boolean(False),

863

1368

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

864

1369

ApprovalDelay=30000,

865

ApprovalDuration=1000,

1370

ApprovalDuration=93785000,

866

1371

Checker=":",

867

1372

ExtendedTimeout=900000,

868

1373

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

869

1374

LastCheckerStatus=-2)

870

1375

self.clients = collections.OrderedDict(

871

1376

[

872

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

873

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

1377

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

1378

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

874

1379

])

875

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

876

877

class TestPrintTableCmd(TestCmd):

878

def test_normal(self):

879

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

880

expected_output = """

881

Name Enabled Timeout Last Successful Check

882

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

883

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

884

"""[1:-1]

885

self.assertEqual(output, expected_output)

886

def test_verbose(self):

887

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

888

expected_output = """

889

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

890

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

891

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

892

"""[1:-1]

893

self.assertEqual(output, expected_output)

894

def test_one_client(self):

895

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

896

expected_output = """

897

Name Enabled Timeout Last Successful Check

898

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

899

"""[1:-1]

900

self.assertEqual(output, expected_output)

1380

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

1381

1382

@property

1383

def bus(self):

1384

class Bus(object):

1385

@staticmethod

1386

def get_object(client_bus_name, path):

1387

self.assertEqual(client_bus_name, dbus_busname)

1388

return {

1389

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

1390

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

1391

"/clients/foo": self.client,

1392

"/clients/barbar": self.other_client,

1393

}[path]

1394

return Bus()

1395

1396

1397

class TestIsEnabledCmd(TestCmd):

1398

def test_is_enabled(self):

1399

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

1400

properties)

1401

for client, properties

1402

in self.clients.items()))

1403

1404

def test_is_enabled_run_exits_successfully(self):

1405

with self.assertRaises(SystemExit) as e:

1406

IsEnabledCmd().run(self.one_client)

1407

if e.exception.code is not None:

1408

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

1409

else:

1410

self.assertIsNone(e.exception.code)

1411

1412

def test_is_enabled_run_exits_with_failure(self):

1413

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

1414

with self.assertRaises(SystemExit) as e:

1415

IsEnabledCmd().run(self.one_client)

1416

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

1417

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

1418

else:

1419

self.assertIsNotNone(e.exception.code)

1420

1421

1422

class TestApproveCmd(TestCmd):

1423

def test_approve(self):

1424

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

1425

for clientpath in self.clients:

1426

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

1427

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

1428

client.calls)

1429

1430

1431

class TestDenyCmd(TestCmd):

1432

def test_deny(self):

1433

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

1434

for clientpath in self.clients:

1435

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

1436

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

1437

client.calls)

1438

1439

1440

class TestRemoveCmd(TestCmd):

1441

def test_remove(self):

1442

class MockMandos(object):

1443

def __init__(self):

1444

self.calls = []

1445

def RemoveClient(self, dbus_path):

1446

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

1447

mandos = MockMandos()

1448

super(TestRemoveCmd, self).setUp()

1449

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

1450

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

1451

for clientpath in self.clients:

1452

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

1453

mandos.calls)

1454

901

1455

902

1456

class TestDumpJSONCmd(TestCmd):

903

1457

def setUp(self):

944

1498

"ApprovedByDefault": False,

945

1499

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

946

1500

"ApprovalDelay": 30000,

947

"ApprovalDuration": 1000,

1501

"ApprovalDuration": 93785000,

948

1502

"Checker": ":",

949

1503

"ExtendedTimeout": 900000,

950

1504

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

952

1506

953

1507

}

954

1508

return super(TestDumpJSONCmd, self).setUp()

1509

955

1510

def test_normal(self):

956

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

1511

output = DumpJSONCmd().output(self.clients.values())

1512

json_data = json.loads(output)

957

1513

self.assertDictEqual(json_data, self.expected_json)

1514

958

1515

def test_one_client(self):

959

clients = self.one_client

960

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

1516

output = DumpJSONCmd().output(self.one_client.values())

1517

json_data = json.loads(output)

961

1518

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

962

1519

self.assertDictEqual(json_data, expected_json)

963

1520

964

class TestIsEnabledCmd(TestCmd):

965

def test_is_enabled(self):

966

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

967

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

968

def test_is_enabled_run_exits_successfully(self):

969

with self.assertRaises(SystemExit) as e:

970

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

971

if e.exception.code is not None:

972

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

973

else:

974

self.assertIsNone(e.exception.code)

975

def test_is_enabled_run_exits_with_failure(self):

976

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

977

with self.assertRaises(SystemExit) as e:

978

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

979

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

980

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

981

else:

982

self.assertIsNotNone(e.exception.code)

983

984

class TestRemoveCmd(TestCmd):

985

def test_remove(self):

986

class MockMandos(object):

987

def __init__(self):

988

self.calls = []

989

def RemoveClient(self, dbus_path):

990

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

991

mandos = MockMandos()

992

super(TestRemoveCmd, self).setUp()

993

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

994

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

995

for client in self.clients:

996

self.assertIn(("RemoveClient",

997

(client.__dbus_object_path__,)),

998

mandos.calls)

999

1000

class TestApproveCmd(TestCmd):

1001

def test_approve(self):

1002

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

1003

for client in self.clients:

1004

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

1005

client.calls)

1006

1007

class TestDenyCmd(TestCmd):

1008

def test_deny(self):

1009

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

1010

for client in self.clients:

1011

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

1012

client.calls)

1013

1014

class TestEnableCmd(TestCmd):

1015

def test_enable(self):

1016

for client in self.clients:

1017

client.attributes["Enabled"] = False

1018

1019

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

1020

1021

for client in self.clients:

1022

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

1023

1024

class TestDisableCmd(TestCmd):

1025

def test_disable(self):

1026

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

1027

1028

for client in self.clients:

1029

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

1030

1031

class Unique(object):

1032

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

1033

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

1521

1522

class TestPrintTableCmd(TestCmd):

1523

def test_normal(self):

1524

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

1525

expected_output = "\n".join((

1526

"Name Enabled Timeout Last Successful Check",

1527

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

1528

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

1529

))

1530

self.assertEqual(output, expected_output)

1531

1532

def test_verbose(self):

1533

output = PrintTableCmd(verbose=True).output(

1534

self.clients.values())

1535

columns = (

1536

(

1537

"Name ",

1538

"foo ",

1539

"barbar ",

1540

),(

1541

"Enabled ",

1542

"Yes ",

1543

"Yes ",

1544

),(

1545

"Timeout ",

1546

"00:05:00 ",

1547

"00:05:00 ",

1548

),(

1549

"Last Successful Check ",

1550

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

1551

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

1552

),(

1553

"Created ",

1554

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

1555

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

1556

),(

1557

"Interval ",

1558

"00:02:00 ",

1559

"00:02:00 ",

1560

),(

1561

"Host ",

1562

"foo.example.org ",

1563

"192.0.2.3 ",

1564

),(

1565

("Key ID "

1566

" "),

1567

("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"

1568

"ea90a27cddb12 "),

1569

("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"

1570

"46c2b020f441c "),

1571

),(

1572

"Fingerprint ",

1573

"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",

1574

"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",

1575

),(

1576

"Check Is Running ",

1577

"No ",

1578

"Yes ",

1579

),(

1580

"Last Enabled ",

1581

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

1582

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

1583

),(

1584

"Approval Is Pending ",

1585

"No ",

1586

"No ",

1587

),(

1588

"Approved By Default ",

1589

"Yes ",

1590

"No ",

1591

),(

1592

"Last Approval Request ",

1593

" ",

1594

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

1595

),(

1596

"Approval Delay ",

1597

"00:00:00 ",

1598

"00:00:30 ",

1599

),(

1600

"Approval Duration ",

1601

"00:00:01 ",

1602

"1T02:03:05 ",

1603

),(

1604

"Checker ",

1605

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

1606

": ",

1607

),(

1608

"Extended Timeout ",

1609

"00:15:00 ",

1610

"00:15:00 ",

1611

),(

1612

"Expires ",

1613

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

1614

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

1615

),(

1616

"Last Checker Status",

1617

"0 ",

1618

"-2 ",

1619

)

1620

)

1621

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

1622

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

1623

for rows in columns)

1624

for line in range(num_lines))

1625

self.assertEqual(output, expected_output)

1626

1627

def test_one_client(self):

1628

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

1629

expected_output = "\n".join((

1630

"Name Enabled Timeout Last Successful Check",

1631

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

1632

))

1633

self.assertEqual(output, expected_output)

1634

1034

1635

1035

1636

class TestPropertyCmd(TestCmd):

1036

1637

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

1041

1642

self.values_to_set)

1042

1643

for value_to_set, value_to_get in zip(self.values_to_set,

1043

1644

values_to_get):

1044

for client in self.clients:

1045

old_value = client.attributes[self.property]

1046

self.assertNotIsInstance(old_value, Unique)

1047

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

1645

for clientpath in self.clients:

1646

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

1647

old_value = client.attributes[self.propname]

1648

self.assertNotIsInstance(old_value, self.Unique)

1649

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

1048

1650

self.run_command(value_to_set, self.clients)

1049

for client in self.clients:

1050

value = client.attributes[self.property]

1051

self.assertNotIsInstance(value, Unique)

1651

for clientpath in self.clients:

1652

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

1653

value = client.attributes[self.propname]

1654

self.assertNotIsInstance(value, self.Unique)

1052

1655

self.assertEqual(value, value_to_get)

1656

1657

class Unique(object):

1658

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

1659

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

1660

1053

1661

def run_command(self, value, clients):

1054

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

1662

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

1663

1664

1665

class TestEnableCmd(TestPropertyCmd):

1666

command = EnableCmd

1667

propname = "Enabled"

1668

values_to_set = [dbus.Boolean(True)]

1669

1670

1671

class TestDisableCmd(TestPropertyCmd):

1672

command = DisableCmd

1673

propname = "Enabled"

1674

values_to_set = [dbus.Boolean(False)]

1675

1055

1676

1056

1677

class TestBumpTimeoutCmd(TestPropertyCmd):

1057

1678

command = BumpTimeoutCmd

1058

property = "LastCheckedOK"

1679

propname = "LastCheckedOK"

1059

1680

values_to_set = [""]

1060

1681

1682

1061

1683

class TestStartCheckerCmd(TestPropertyCmd):

1062

1684

command = StartCheckerCmd

1063

property = "CheckerRunning"

1685

propname = "CheckerRunning"

1064

1686

values_to_set = [dbus.Boolean(True)]

1065

1687

1688

1066

1689

class TestStopCheckerCmd(TestPropertyCmd):

1067

1690

command = StopCheckerCmd

1068

property = "CheckerRunning"

1691

propname = "CheckerRunning"

1069

1692

values_to_set = [dbus.Boolean(False)]

1070

1693

1694

1071

1695

class TestApproveByDefaultCmd(TestPropertyCmd):

1072

1696

command = ApproveByDefaultCmd

1073

property = "ApprovedByDefault"

1697

propname = "ApprovedByDefault"

1074

1698

values_to_set = [dbus.Boolean(True)]

1075

1699

1700

1076

1701

class TestDenyByDefaultCmd(TestPropertyCmd):

1077

1702

command = DenyByDefaultCmd

1078

property = "ApprovedByDefault"

1703

propname = "ApprovedByDefault"

1079

1704

values_to_set = [dbus.Boolean(False)]

1080

1705

1081

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1082

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

1083

ValueArgumentMixIn"""

1706

1707

class TestPropertyValueCmd(TestPropertyCmd):

1708

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

1709

1084

1710

def runTest(self):

1085

if type(self) is TestValueArgumentPropertyCmd:

1711

if type(self) is TestPropertyValueCmd:

1086

1712

return

1087

return super(TestValueArgumentPropertyCmd, self).runTest()

1713

return super(TestPropertyValueCmd, self).runTest()

1714

1088

1715

def run_command(self, value, clients):

1089

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

1090

1091

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1716

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

1717

1718

1719

class TestSetCheckerCmd(TestPropertyValueCmd):

1092

1720

command = SetCheckerCmd

1093

property = "Checker"

1721

propname = "Checker"

1094

1722

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

1095

1723

1096

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1724

1725

class TestSetHostCmd(TestPropertyValueCmd):

1097

1726

command = SetHostCmd

1098

property = "Host"

1727

propname = "Host"

1099

1728

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

1100

1729

1101

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1730

1731

class TestSetSecretCmd(TestPropertyValueCmd):

1102

1732

command = SetSecretCmd

1103

property = "Secret"

1733

propname = "Secret"

1104

1734

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

1105

1735

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

1106

1736

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

1107

1737

1108

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1738

1739

class TestSetTimeoutCmd(TestPropertyValueCmd):

1109

1740

command = SetTimeoutCmd

1110

property = "Timeout"

1741

propname = "Timeout"

1111

1742

values_to_set = [datetime.timedelta(),

1112

1743

datetime.timedelta(minutes=5),

1113

1744

datetime.timedelta(seconds=1),

1115

1746

datetime.timedelta(weeks=52)]

1116

1747

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

1117

1748

1118

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1749

1750

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1119

1751

command = SetExtendedTimeoutCmd

1120

property = "ExtendedTimeout"

1752

propname = "ExtendedTimeout"

1121

1753

values_to_set = [datetime.timedelta(),

1122

1754

datetime.timedelta(minutes=5),

1123

1755

datetime.timedelta(seconds=1),

1125

1757

datetime.timedelta(weeks=52)]

1126

1758

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

1127

1759

1128

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1760

1761

class TestSetIntervalCmd(TestPropertyValueCmd):

1129

1762

command = SetIntervalCmd

1130

property = "Interval"

1763

propname = "Interval"

1131

1764

values_to_set = [datetime.timedelta(),

1132

1765

datetime.timedelta(minutes=5),

1133

1766

datetime.timedelta(seconds=1),

1135

1768

datetime.timedelta(weeks=52)]

1136

1769

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

1137

1770

1138

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1771

1772

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1139

1773

command = SetApprovalDelayCmd

1140

property = "ApprovalDelay"

1774

propname = "ApprovalDelay"

1141

1775

values_to_set = [datetime.timedelta(),

1142

1776

datetime.timedelta(minutes=5),

1143

1777

datetime.timedelta(seconds=1),

1145

1779

datetime.timedelta(weeks=52)]

1146

1780

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

1147

1781

1148

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1782

1783

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1149

1784

command = SetApprovalDurationCmd

1150

property = "ApprovalDuration"

1785

propname = "ApprovalDuration"

1151

1786

values_to_set = [datetime.timedelta(),

1152

1787

datetime.timedelta(minutes=5),

1153

1788

datetime.timedelta(seconds=1),

1155

1790

datetime.timedelta(weeks=52)]

1156

1791

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

1157

1792

1158

class Test_command_from_options(unittest.TestCase):

1159

def setUp(self):

1160

self.parser = argparse.ArgumentParser()

1161

add_command_line_options(self.parser)

1162

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

1163

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

1164

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

1165

options = self.parser.parse_args(args)

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 1)

1169

command = commands[0]

1170

self.assertIsInstance(command, command_cls)

1171

for key, value in cmd_attrs.items():

1172

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

1173

def test_print_table(self):

1174

self.assert_command_from_args([], PrintTableCmd,

1175

verbose=False)

1176

1177

def test_print_table_verbose(self):

1178

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

1179

verbose=True)

1180

1181

def test_print_table_verbose_short(self):

1182

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

1183

verbose=True)

1184

1185

def test_enable(self):

1186

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

1187

1188

def test_enable_short(self):

1189

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

1190

1191

def test_disable(self):

1192

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

1193

DisableCmd)

1194

1195

def test_disable_short(self):

1196

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

1197

1198

def test_bump_timeout(self):

1199

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

1200

BumpTimeoutCmd)

1201

1202

def test_bump_timeout_short(self):

1203

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

1204

1205

def test_start_checker(self):

1206

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

1207

StartCheckerCmd)

1208

1209

def test_stop_checker(self):

1210

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

1211

StopCheckerCmd)

1212

1213

def test_remove(self):

1214

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

1215

RemoveCmd)

1216

1217

def test_remove_short(self):

1218

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

1219

1220

def test_checker(self):

1221

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

1222

SetCheckerCmd, value_to_set=":")

1223

1224

def test_checker_empty(self):

1225

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

1226

SetCheckerCmd, value_to_set="")

1227

1228

def test_checker_short(self):

1229

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

1230

SetCheckerCmd, value_to_set=":")

1231

1232

def test_timeout(self):

1233

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

1234

SetTimeoutCmd,

1235

value_to_set=300000)

1236

1237

def test_timeout_short(self):

1238

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

1239

SetTimeoutCmd,

1240

value_to_set=300000)

1241

1242

def test_extended_timeout(self):

1243

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

1244

"foo"],

1245

SetExtendedTimeoutCmd,

1246

value_to_set=900000)

1247

1248

def test_interval(self):

1249

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

1250

SetIntervalCmd,

1251

value_to_set=120000)

1252

1253

def test_interval_short(self):

1254

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

1255

SetIntervalCmd,

1256

value_to_set=120000)

1257

1258

def test_approve_by_default(self):

1259

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

1260

ApproveByDefaultCmd)

1261

1262

def test_deny_by_default(self):

1263

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

1264

DenyByDefaultCmd)

1265

1266

def test_approval_delay(self):

1267

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

1268

"foo"], SetApprovalDelayCmd,

1269

value_to_set=30000)

1270

1271

def test_approval_duration(self):

1272

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

1273

"foo"], SetApprovalDurationCmd,

1274

value_to_set=1000)

1275

1276

def test_host(self):

1277

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

1278

"foo"], SetHostCmd,

1279

value_to_set="foo.example.org")

1280

1281

def test_host_short(self):

1282

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

1283

"foo"], SetHostCmd,

1284

value_to_set="foo.example.org")

1285

1286

def test_secret_devnull(self):

1287

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

1288

"foo"], SetSecretCmd,

1289

value_to_set=b"")

1290

1291

def test_secret_tempfile(self):

1292

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

1293

value = b"secret\0xyzzy\nbar"

1294

f.write(value)

1295

f.seek(0)

1296

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

1297

"foo"], SetSecretCmd,

1298

value_to_set=value)

1299

1300

def test_secret_devnull_short(self):

1301

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

1302

SetSecretCmd, value_to_set=b"")

1303

1304

def test_secret_tempfile_short(self):

1305

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

1306

value = b"secret\0xyzzy\nbar"

1307

f.write(value)

1308

f.seek(0)

1309

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

1310

SetSecretCmd,

1311

value_to_set=value)

1312

1313

def test_approve(self):

1314

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

1315

ApproveCmd)

1316

1317

def test_approve_short(self):

1318

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

1319

1320

def test_deny(self):

1321

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

1322

1323

def test_deny_short(self):

1324

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

1325

1326

def test_dump_json(self):

1327

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

1328

1329

def test_is_enabled(self):

1330

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

1331

IsEnabledCmd)

1332

1333

def test_is_enabled_short(self):

1334

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

1335

1336

1337

class Test_check_option_syntax(unittest.TestCase):

1338

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

1339

# check_option_syntax()

1340

actions = {

1341

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

1342

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

1343

# never used

1344

"enable": True,

1345

"disable": True,

1346

"bump_timeout": True,

1347

"start_checker": True,

1348

"stop_checker": True,

1349

"is_enabled": True,

1350

"remove": True,

1351

"checker": "x",

1352

"timeout": datetime.timedelta(),

1353

"extended_timeout": datetime.timedelta(),

1354

"interval": datetime.timedelta(),

1355

"approved_by_default": True,

1356

"approval_delay": datetime.timedelta(),

1357

"approval_duration": datetime.timedelta(),

1358

"host": "x",

1359

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

1360

"approve": True,

1361

"deny": True,

1362

}

1363

1364

def setUp(self):

1365

self.parser = argparse.ArgumentParser()

1366

add_command_line_options(self.parser)

1367

1368

@contextlib.contextmanager

1369

def assertParseError(self):

1370

with self.assertRaises(SystemExit) as e:

1371

with self.temporarily_suppress_stderr():

1372

yield

1373

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

1374

# https://docs.python.org/3/library/argparse.html#exiting-methods

1375

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

1376

1377

@staticmethod

1378

@contextlib.contextmanager

1379

def temporarily_suppress_stderr():

1380

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

1381

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

1382

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

1383

os.close(null)

1384

try:

1385

yield

1386

finally:

1387

# restore stderr

1388

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

1389

os.close(stderrcopy)

1390

1391

def check_option_syntax(self, options):

1392

check_option_syntax(self.parser, options)

1393

1394

def test_actions_requires_client_or_all(self):

1395

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

1396

options = self.parser.parse_args()

1397

setattr(options, action, value)

1398

with self.assertParseError():

1399

self.check_option_syntax(options)

1400

1401

def test_actions_conflicts_with_verbose(self):

1402

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

1403

options = self.parser.parse_args()

1404

setattr(options, action, value)

1405

options.verbose = True

1406

with self.assertParseError():

1407

self.check_option_syntax(options)

1408

1409

def test_dump_json_conflicts_with_verbose(self):

1410

options = self.parser.parse_args()

1411

options.dump_json = True

1412

options.verbose = True

1413

with self.assertParseError():

1414

self.check_option_syntax(options)

1415

1416

def test_dump_json_conflicts_with_action(self):

1417

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

1418

options = self.parser.parse_args()

1419

setattr(options, action, value)

1420

options.dump_json = True

1421

with self.assertParseError():

1422

self.check_option_syntax(options)

1423

1424

def test_all_can_not_be_alone(self):

1425

options = self.parser.parse_args()

1426

options.all = True

1427

with self.assertParseError():

1428

self.check_option_syntax(options)

1429

1430

def test_all_is_ok_with_any_action(self):

1431

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

1432

options = self.parser.parse_args()

1433

setattr(options, action, value)

1434

options.all = True

1435

self.check_option_syntax(options)

1436

1437

def test_is_enabled_fails_without_client(self):

1438

options = self.parser.parse_args()

1439

options.is_enabled = True

1440

with self.assertParseError():

1441

self.check_option_syntax(options)

1442

1443

def test_is_enabled_works_with_one_client(self):

1444

options = self.parser.parse_args()

1445

options.is_enabled = True

1446

options.client = ["foo"]

1447

self.check_option_syntax(options)

1448

1449

def test_is_enabled_fails_with_two_clients(self):

1450

options = self.parser.parse_args()

1451

options.is_enabled = True

1452

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

1453

with self.assertParseError():

1454

self.check_option_syntax(options)

1455

1456

1793

1457

1794

1458

1795

def should_only_run_tests():

1472

1809

return tests

1473

1810

1474

1811

if __name__ == "__main__":

1475

if should_only_run_tests():

1476

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

1477

unittest.main()

1478

else:

1479

main()

1812

try:

1813

if should_only_run_tests():

1814

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

1815

unittest.main()

1816

else:

1817

main()

1818

finally:

1819

logging.shutdown()

Older »