/mandos/trunk : revision 1003

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-03 01:08:58 UTC
Revision ID: teddy@recompile.se-20190303010858-c2l0sr6ekvzo7rlb

mandos-ctl: Separate determining what to do and actually doing it

* mandos-ctl (defaultkeywords): Removed; value moved into
                                PrintTableCmd.
  (Command): New abstract base class for commands to be run.
  (PrintCmd, PropertyCmd): New abstract classes for commands.
  (ValueArgumentMixIn, MillisecondsValueArgumentMixIn): New mixins for
                                                        commands.
  (PrintTableCmd, DumpJSONCmd, IsEnabledCmd, RemoveCmd, ApproveCmd,
  DenyCmd, EnableCmd, DisableCmd, BumpTimeoutCmd, StartCheckerCmd,
  StopCheckerCmd, ApproveByDefaultCmd, DenyByDefaultCmd,
  SetCheckerCmd, SetTimeoutCmd, SetExtendedTimeoutCmd,
  SetApprovalDelayCmd, SetApprovalDurationCmd): New commands.
  (main): Don't look directly at options and do things; instead go
          through all options and add commands to a list, then run all
          commands on clients.

files modified:
mandos-ctl

mandos-ctl.xml

Show diffs side-by-side

added added

removed removed

mandos-ctl

import json

import unittest

import logging

import io

import tempfile

import contextlib

import dbus

locale.setlocale(locale.LC_ALL, "")

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

domain = "se.recompile"

busname = domain + ".Mandos"

server_path = "/"

server_interface = domain + ".Mandos"

client_interface = domain + ".Mandos.Client"

version = "1.8.3"

272

268

return value

273

269

274

270

271

class TableOfClients(object):

272

tablewords = {

273

"Name": "Name",

274

"Enabled": "Enabled",

275

"Timeout": "Timeout",

276

"LastCheckedOK": "Last Successful Check",

277

"LastApprovalRequest": "Last Approval Request",

278

"Created": "Created",

279

"Interval": "Interval",

280

"Host": "Host",

281

"Fingerprint": "Fingerprint",

282

"KeyID": "Key ID",

283

"CheckerRunning": "Check Is Running",

284

"LastEnabled": "Last Enabled",

285

"ApprovalPending": "Approval Is Pending",

286

"ApprovedByDefault": "Approved By Default",

287

"ApprovalDelay": "Approval Delay",

288

"ApprovalDuration": "Approval Duration",

289

"Checker": "Checker",

290

"ExtendedTimeout": "Extended Timeout",

291

"Expires": "Expires",

292

"LastCheckerStatus": "Last Checker Status",

293

}

294

295

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

296

self.clients = clients

297

self.keywords = keywords

298

if tablewords is not None:

299

self.tablewords = tablewords

300

301

def __str__(self):

302

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

303

304

if sys.version_info.major == 2:

305

__unicode__ = __str__

306

def __str__(self):

307

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

308

309

def rows(self):

310

format_string = self.row_formatting_string()

311

rows = [self.header_line(format_string)]

312

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

313

for client in self.clients)

314

return rows

315

316

def row_formatting_string(self):

317

"Format string used to format table rows"

318

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

319

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

320

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

321

for client in self.clients)),

322

key=key)

323

for key in self.keywords)

324

325

def string_from_client(self, client, key):

326

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

327

328

@staticmethod

329

def valuetostring(value, keyword):

330

if isinstance(value, dbus.Boolean):

331

return "Yes" if value else "No"

332

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

333

"ApprovalDuration", "ExtendedTimeout"):

334

return milliseconds_to_string(value)

335

return str(value)

336

337

def header_line(self, format_string):

338

return format_string.format(**self.tablewords)

339

340

def client_line(self, client, format_string):

341

return format_string.format(

342

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

343

for key in self.keywords})

344

345

275

346

## Classes for commands.

276

347

277

348

# Abstract classes first

278

349

class Command(object):

279

350

"""Abstract class for commands"""

280

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

351

def run(self, clients):

281

352

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

282

353

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

283

354

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

284

self.mandos = mandos

285

for clientpath, properties in clients.items():

286

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

287

dbus_busname, str(clientpath))

288

client = bus.get_object(dbus_busname, clientpath)

289

self.run_on_one_client(client, properties)

355

for client in clients:

356

self.run_on_one_client(client)

290

357

291

358

class PrintCmd(Command):

292

359

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

297

364

"LastApprovalRequest", "ApprovalDelay",

298

365

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

299

366

"Expires", "LastCheckerStatus")

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

367

def run(self, clients):

368

print(self.output(clients))

304

369

305

370

class PropertyCmd(Command):

306

371

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

307

def run_on_one_client(self, client, properties):

372

def run_on_one_client(self, client):

308

373

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

309

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

310

client.__dbus_object_path__,

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

313

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

314

else bool(self.value_to_set))

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

374

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

317

375

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

376

322

class PropertyValueCmd(PropertyCmd):

323

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

377

class ValueArgumentMixIn(object):

378

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

324

379

def __init__(self, value):

325

380

self.value_to_set = value

326

381

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

382

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

383

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

384

milliseconds."""

330

385

@property

331

386

def value_to_set(self):

332

387

return self._vts

333

388

@value_to_set.setter

334

389

def value_to_set(self, value):

335

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

336

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

390

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

391

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

337

392

338

393

# Actual (non-abstract) command classes

339

394

340

395

class PrintTableCmd(PrintCmd):

341

396

def __init__(self, verbose=False):

342

397

self.verbose = verbose

343

344

398

def output(self, clients):

345

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

346

"LastCheckedOK")

347

keywords = default_keywords

348

399

if self.verbose:

349

400

keywords = self.all_keywords

350

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

351

352

class TableOfClients(object):

353

tableheaders = {

354

"Name": "Name",

355

"Enabled": "Enabled",

356

"Timeout": "Timeout",

357

"LastCheckedOK": "Last Successful Check",

358

"LastApprovalRequest": "Last Approval Request",

359

"Created": "Created",

360

"Interval": "Interval",

361

"Host": "Host",

362

"Fingerprint": "Fingerprint",

363

"KeyID": "Key ID",

364

"CheckerRunning": "Check Is Running",

365

"LastEnabled": "Last Enabled",

366

"ApprovalPending": "Approval Is Pending",

367

"ApprovedByDefault": "Approved By Default",

368

"ApprovalDelay": "Approval Delay",

369

"ApprovalDuration": "Approval Duration",

370

"Checker": "Checker",

371

"ExtendedTimeout": "Extended Timeout",

372

"Expires": "Expires",

373

"LastCheckerStatus": "Last Checker Status",

374

}

375

376

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

377

self.clients = clients

378

self.keywords = keywords

379

if tableheaders is not None:

380

self.tableheaders = tableheaders

381

382

def __str__(self):

383

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

384

385

if sys.version_info.major == 2:

386

__unicode__ = __str__

387

def __str__(self):

388

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

389

390

def rows(self):

391

format_string = self.row_formatting_string()

392

rows = [self.header_line(format_string)]

393

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

394

for client in self.clients)

395

return rows

396

397

def row_formatting_string(self):

398

"Format string used to format table rows"

399

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

400

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

401

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

402

for client in self.clients)),

403

key=key)

404

for key in self.keywords)

405

406

def string_from_client(self, client, key):

407

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

408

409

@staticmethod

410

def valuetostring(value, keyword):

411

if isinstance(value, dbus.Boolean):

412

return "Yes" if value else "No"

413

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

414

"ApprovalDuration", "ExtendedTimeout"):

415

return milliseconds_to_string(value)

416

return str(value)

417

418

def header_line(self, format_string):

419

return format_string.format(**self.tableheaders)

420

421

def client_line(self, client, format_string):

422

return format_string.format(

423

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

424

for key in self.keywords})

425

426

401

else:

402

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

403

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

427

404

428

405

class DumpJSONCmd(PrintCmd):

429

406

def output(self, clients):

439

416

return value

440

417

441

418

class IsEnabledCmd(Command):

442

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

443

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

444

if self.is_enabled(client, properties):

419

def run_on_one_client(self, client):

420

if self.is_enabled(client):

445

421

sys.exit(0)

446

422

sys.exit(1)

447

def is_enabled(self, client, properties):

448

return properties["Enabled"]

423

def is_enabled(self, client):

424

return client.Get(client_interface, "Enabled",

425

dbus_interface=dbus.PROPERTIES_IFACE)

449

426

450

427

class RemoveCmd(Command):

451

def run_on_one_client(self, client, properties):

452

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

453

server_dbus_path, server_dbus_interface,

454

str(client.__dbus_object_path__))

428

def __init__(self, mandos):

429

self.mandos = mandos

430

def run_on_one_client(self, client):

455

431

self.mandos.RemoveClient(client.__dbus_object_path__)

456

432

457

433

class ApproveCmd(Command):

458

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_dbus_interface)

434

def run_on_one_client(self, client):

461

435

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

436

dbus_interface=client_interface)

463

437

464

438

class DenyCmd(Command):

465

def run_on_one_client(self, client, properties):

466

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

467

client.__dbus_object_path__, client_dbus_interface)

439

def run_on_one_client(self, client):

468

440

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

441

dbus_interface=client_interface)

470

442

471

443

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

444

property = "Enabled"

473

445

value_to_set = dbus.Boolean(True)

474

446

475

447

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

448

property = "Enabled"

477

449

value_to_set = dbus.Boolean(False)

478

450

479

451

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

452

property = "LastCheckedOK"

481

453

value_to_set = ""

482

454

483

455

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

456

property = "CheckerRunning"

485

457

value_to_set = dbus.Boolean(True)

486

458

487

459

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

460

property = "CheckerRunning"

489

461

value_to_set = dbus.Boolean(False)

490

462

491

463

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

464

property = "ApprovedByDefault"

493

465

value_to_set = dbus.Boolean(True)

494

466

495

467

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

468

property = "ApprovedByDefault"

497

469

value_to_set = dbus.Boolean(False)

498

470

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

507

@property

508

def value_to_set(self):

509

return self._vts

510

@value_to_set.setter

511

def value_to_set(self, value):

512

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

513

self._vts = value.read()

514

value.close()

515

516

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "Timeout"

518

519

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "ExtendedTimeout"

521

522

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "Interval"

524

525

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

529

propname = "ApprovalDuration"

530

531

def add_command_line_options(parser):

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

519

def main():

520

parser = argparse.ArgumentParser()

532

521

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

533

522

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

534

523

help="show version number and exit")

559

548

help="Remove client")

560

549

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

561

550

help="Set checker command for client")

562

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

551

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

563

552

help="Set timeout for client")

564

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

553

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

565

554

help="Set extended timeout for client")

566

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

555

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

567

556

help="Set checker interval for client")

568

557

approve_deny_default = parser.add_mutually_exclusive_group()

569

558

approve_deny_default.add_argument(

574

563

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

575

564

dest="approved_by_default",

576

565

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

577

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

566

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

578

567

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

579

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

568

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

580

569

help="Set duration of one client approval")

581

570

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

582

571

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

588

577

help="Approve any current client request")

589

578

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

590

579

help="Deny any current client request")

591

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

592

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

593

580

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

594

581

help="Run self-test")

595

582

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

596

597

598

def commands_from_options(options):

583

options = parser.parse_args()

584

585

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

586

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

587

if options.verbose and has_actions(options):

588

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

589

if options.dump_json and (options.verbose

590

or has_actions(options)):

591

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

592

if options.all and not has_actions(options):

593

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

594

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

595

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

596

597

try:

598

bus = dbus.SystemBus()

599

mandos_dbus_objc = bus.get_object(busname, server_path)

600

except dbus.exceptions.DBusException:

601

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

602

sys.exit(1)

603

604

mandos_serv = dbus.Interface(mandos_dbus_objc,

605

dbus_interface=server_interface)

606

mandos_serv_object_manager = dbus.Interface(

607

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

599

608

600

609

commands = []

601

610

609

618

commands.append(DisableCmd())

610

619

611

620

if options.bump_timeout:

612

commands.append(BumpTimeoutCmd())

621

commands.append(BumpTimeoutCmd(options.bump_timeout))

613

622

614

623

if options.start_checker:

615

624

commands.append(StartCheckerCmd())

620

629

if options.is_enabled:

621

630

commands.append(IsEnabledCmd())

622

631

632

if options.remove:

633

commands.append(RemoveCmd(mandos_serv))

634

623

635

if options.checker is not None:

624

commands.append(SetCheckerCmd(options.checker))

636

commands.append(SetCheckerCmd())

625

637

626

638

if options.timeout is not None:

627

639

commands.append(SetTimeoutCmd(options.timeout))

631

643

SetExtendedTimeoutCmd(options.extended_timeout))

632

644

633

645

if options.interval is not None:

634

commands.append(SetIntervalCmd(options.interval))

646

command.append(SetIntervalCmd(options.interval))

635

647

636

648

if options.approved_by_default is not None:

637

649

if options.approved_by_default:

638

commands.append(ApproveByDefaultCmd())

650

command.append(ApproveByDefaultCmd())

639

651

else:

640

commands.append(DenyByDefaultCmd())

652

command.append(DenyByDefaultCmd())

641

653

642

654

if options.approval_delay is not None:

643

commands.append(SetApprovalDelayCmd(options.approval_delay))

655

command.append(SetApprovalDelayCmd(options.approval_delay))

644

656

645

657

if options.approval_duration is not None:

646

commands.append(

658

command.append(

647

659

SetApprovalDurationCmd(options.approval_duration))

648

660

649

661

if options.host is not None:

650

commands.append(SetHostCmd(options.host))

662

command.append(SetHostCmd(options.host))

651

663

652

664

if options.secret is not None:

653

commands.append(SetSecretCmd(options.secret))

665

command.append(SetSecretCmd(options.secret))

654

666

655

667

if options.approve:

656

668

commands.append(ApproveCmd())

658

670

if options.deny:

659

671

commands.append(DenyCmd())

660

672

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

673

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

665

674

# optionally verbosely

666

675

if not commands:

667

676

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

668

677

669

return commands

670

671

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

675

676

def has_actions(options):

677

return any((options.enable,

678

options.disable,

679

options.bump_timeout,

680

options.start_checker,

681

options.stop_checker,

682

options.is_enabled,

683

options.remove,

684

options.checker is not None,

685

options.timeout is not None,

686

options.extended_timeout is not None,

687

options.interval is not None,

688

options.approved_by_default is not None,

689

options.approval_delay is not None,

690

options.approval_duration is not None,

691

options.host is not None,

692

options.secret is not None,

693

options.approve,

694

options.deny))

695

696

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

697

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

698

if options.verbose and has_actions(options):

699

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

700

if options.dump_json and (options.verbose

701

or has_actions(options)):

702

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

703

if options.all and not has_actions(options):

704

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

705

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

706

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

723

clientnames = options.client

724

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

try:

729

bus = dbus.SystemBus()

730

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

731

dbus_busname, server_dbus_path)

732

mandos_dbus_objc = bus.get_object(dbus_busname,

733

server_dbus_path)

734

except dbus.exceptions.DBusException:

735

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

736

sys.exit(1)

737

738

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_dbus_interface)

740

mandos_serv_object_manager = dbus.Interface(

741

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

743

# Filter out log message from dbus module

744

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

745

class NullFilter(logging.Filter):

746

def filter(self, record):

747

return False

748

dbus_filter = NullFilter()

749

try:

750

dbus_logger.addFilter(dbus_filter)

751

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

752

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_dbus_interface]

754

for path, ifs_and_props in

755

mandos_serv_object_manager

756

.GetManagedObjects().items()

757

if client_dbus_interface in ifs_and_props}

678

# block stderr since dbus library prints to stderr

679

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

680

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

681

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

682

os.close(null)

683

try:

684

try:

685

mandos_clients = {path: ifs_and_props[client_interface]

686

for path, ifs_and_props in

687

mandos_serv_object_manager

688

.GetManagedObjects().items()

689

if client_interface in ifs_and_props}

690

finally:

691

# restore stderr

692

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

693

os.close(stderrcopy)

758

694

except dbus.exceptions.DBusException as e:

759

695

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

760

696

"\n%s", e)

761

697

sys.exit(1)

762

finally:

763

# restore dbus logger

764

dbus_logger.removeFilter(dbus_filter)

765

698

766

699

# Compile dict of (clients: properties) to process

767

700

clients = {}

768

701

769

if not clientnames:

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

702

if options.all or not options.client:

703

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

704

for path, properties in mandos_clients.items()}

772

705

else:

773

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

706

for name in options.client:

707

for path, client in mandos_clients.items():

708

if client["Name"] == name:

709

client_objc = bus.get_object(busname, path)

710

clients[client_objc] = client

777

711

break

778

712

else:

779

713

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

780

714

sys.exit(1)

781

715

782

716

# Run all commands on clients

783

commands = commands_from_options(options)

784

717

for command in commands:

785

command.run(clients, bus, mandos_serv)

718

command.run(clients)

786

719

787

720

788

721

class Test_milliseconds_to_string(unittest.TestCase):

800

733

801

734

class Test_string_to_delta(unittest.TestCase):

802

735

def test_handles_basic_rfc3339(self):

803

self.assertEqual(string_to_delta("PT0S"),

804

datetime.timedelta())

805

self.assertEqual(string_to_delta("P0D"),

806

datetime.timedelta())

807

self.assertEqual(string_to_delta("PT1S"),

808

datetime.timedelta(0, 1))

809

736

self.assertEqual(string_to_delta("PT2H"),

810

737

datetime.timedelta(0, 7200))

811

738

def test_falls_back_to_pre_1_6_1_with_warning(self):

814

741

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

815

742

value = string_to_delta("2h")

816

743

else:

817

class WarningFilter(logging.Filter):

818

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

819

def filter(self, record):

820

is_warning = record.levelno >= logging.WARNING

821

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

822

False)

823

return not is_warning

824

warning_filter = WarningFilter()

825

log.addFilter(warning_filter)

826

try:

827

value = string_to_delta("2h")

828

finally:

829

log.removeFilter(warning_filter)

830

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

744

value = string_to_delta("2h")

831

745

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

832

746

833

834

class TestCmd(unittest.TestCase):

835

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

836

def setUp(self):

837

testcase = self

838

class MockClient(object):

839

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

840

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

841

self.attributes = attributes

842

self.attributes["Name"] = name

843

self.calls = []

844

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

845

testcase.assertEqual(interface, client_dbus_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_dbus_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

854

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

857

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

858

dbus_interface)))

859

self.client = MockClient(

860

"foo",

861

KeyID=("92ed150794387c03ce684574b1139a65"

862

"94a34f895daaaf09fd8ea90a27cddb12"),

863

Secret=b"secret",

864

Host="foo.example.org",

865

Enabled=dbus.Boolean(True),

866

Timeout=300000,

867

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

868

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

869

Interval=120000,

870

Fingerprint=("778827225BA7DE539C5A"

871

"7CFA59CFF7CDBD9A5920"),

872

CheckerRunning=dbus.Boolean(False),

873

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

874

ApprovalPending=dbus.Boolean(False),

875

ApprovedByDefault=dbus.Boolean(True),

876

LastApprovalRequest="",

877

ApprovalDelay=0,

878

ApprovalDuration=1000,

879

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

880

ExtendedTimeout=900000,

881

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

882

LastCheckerStatus=0)

883

self.other_client = MockClient(

884

"barbar",

885

KeyID=("0558568eedd67d622f5c83b35a115f79"

886

"6ab612cff5ad227247e46c2b020f441c"),

887

Secret=b"secretbar",

888

Host="192.0.2.3",

889

Enabled=dbus.Boolean(True),

890

Timeout=300000,

891

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

892

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

893

Interval=120000,

894

Fingerprint=("3E393AEAEFB84C7E89E2"

895

"F547B3A107558FCA3A27"),

896

CheckerRunning=dbus.Boolean(True),

897

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

898

ApprovalPending=dbus.Boolean(False),

899

ApprovedByDefault=dbus.Boolean(False),

900

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

901

ApprovalDelay=30000,

902

ApprovalDuration=1000,

903

Checker=":",

904

ExtendedTimeout=900000,

905

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

906

LastCheckerStatus=-2)

907

self.clients = collections.OrderedDict(

908

[

909

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

910

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

911

])

912

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

913

@property

914

def bus(self):

915

class Bus(object):

916

@staticmethod

917

def get_object(client_bus_name, path):

918

self.assertEqual(client_bus_name, dbus_busname)

919

return {

920

"/clients/foo": self.client,

921

"/clients/barbar": self.other_client,

922

}[path]

923

return Bus()

924

925

class TestPrintTableCmd(TestCmd):

926

def test_normal(self):

927

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

928

expected_output = "\n".join((

929

"Name Enabled Timeout Last Successful Check",

930

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

931

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

932

))

933

self.assertEqual(output, expected_output)

934

def test_verbose(self):

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

expected_output = "\n".join((

938

# First line (headers)

939

"Name Enabled Timeout Last Successful Check Created "

940

" Interval Host Key ID "

941

" Fingerprint "

942

" Check Is Running Last Enabl"

943

"ed Approval Is Pending Approved By Default Last A"

944

"pproval Request Approval Delay Approval Duration Checker"

945

" Extended Timeout Expires Last "

946

"Checker Status",

947

# Second line (client "foo")

948

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

949

"T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684"

950

"574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7"

951

"DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03"

952

"T00:00:00 No Yes "

953

" 00:00:00 00:00:01 fping -"

954

"q -- %(host)s 00:15:00 2019-02-04T00:00:00 0 "

955

" ",

956

# Third line (client "barbar")

957

"barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03"

958

"T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c8"

959

"3b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB8"

960

"4C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04"

961

"T00:00:00 No No 2019-0"

962

"1-03T00:00:00 00:00:30 00:00:01 : "

963

" 00:15:00 2019-02-05T00:00:00 -2 "

964

" ",

965

))

966

self.assertEqual(output, expected_output)

967

def test_one_client(self):

968

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

969

expected_output = """

970

Name Enabled Timeout Last Successful Check

971

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

972

"""[1:-1]

973

self.assertEqual(output, expected_output)

974

975

class TestDumpJSONCmd(TestCmd):

976

def setUp(self):

977

self.expected_json = {

978

"foo": {

979

"Name": "foo",

980

"KeyID": ("92ed150794387c03ce684574b1139a65"

981

"94a34f895daaaf09fd8ea90a27cddb12"),

982

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

983

"Enabled": True,

984

"Timeout": 300000,

985

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

986

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

987

"Interval": 120000,

988

"Fingerprint": ("778827225BA7DE539C5A"

989

"7CFA59CFF7CDBD9A5920"),

990

"CheckerRunning": False,

991

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

992

"ApprovalPending": False,

993

"ApprovedByDefault": True,

994

"LastApprovalRequest": "",

995

"ApprovalDelay": 0,

996

"ApprovalDuration": 1000,

997

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

998

"ExtendedTimeout": 900000,

999

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

1000

"LastCheckerStatus": 0,

1001

1002

"barbar": {

1003

"Name": "barbar",

1004

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1005

"6ab612cff5ad227247e46c2b020f441c"),

1006

"Host": "192.0.2.3",

1007

"Enabled": True,

1008

"Timeout": 300000,

1009

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

1010

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

1011

"Interval": 120000,

1012

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1013

"F547B3A107558FCA3A27"),

1014

"CheckerRunning": True,

1015

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

1016

"ApprovalPending": False,

1017

"ApprovedByDefault": False,

1018

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

1019

"ApprovalDelay": 30000,

1020

"ApprovalDuration": 1000,

1021

"Checker": ":",

1022

"ExtendedTimeout": 900000,

1023

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

1024

"LastCheckerStatus": -2,

1025

747

class Test_TableOfClients(unittest.TestCase):

748

def setUp(self):

749

self.tablewords = {

750

"Attr1": "X",

751

"AttrTwo": "Yy",

752

"AttrThree": "Zzz",

753

"Bool": "A D-BUS Boolean",

754

"NonDbusBoolean": "A Non-D-BUS Boolean",

755

"Integer": "An Integer",

756

"Timeout": "Timedelta 1",

757

"Interval": "Timedelta 2",

758

"ApprovalDelay": "Timedelta 3",

759

"ApprovalDuration": "Timedelta 4",

760

"ExtendedTimeout": "Timedelta 5",

761

"String": "A String",

1026

762

}

1027

return super(TestDumpJSONCmd, self).setUp()

1028

def test_normal(self):

1029

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

1030

self.assertDictEqual(json_data, self.expected_json)

1031

def test_one_client(self):

1032

clients = self.one_client

1033

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

1034

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

1035

self.assertDictEqual(json_data, expected_json)

1036

1037

class TestIsEnabledCmd(TestCmd):

1038

def test_is_enabled(self):

1039

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

1040

properties)

1041

for client, properties

1042

in self.clients.items()))

1043

def test_is_enabled_run_exits_successfully(self):

1044

with self.assertRaises(SystemExit) as e:

1045

IsEnabledCmd().run(self.one_client)

1046

if e.exception.code is not None:

1047

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

1048

else:

1049

self.assertIsNone(e.exception.code)

1050

def test_is_enabled_run_exits_with_failure(self):

1051

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

1052

with self.assertRaises(SystemExit) as e:

1053

IsEnabledCmd().run(self.one_client)

1054

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

1055

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

1056

else:

1057

self.assertIsNotNone(e.exception.code)

1058

1059

class TestRemoveCmd(TestCmd):

1060

def test_remove(self):

1061

class MockMandos(object):

1062

def __init__(self):

1063

self.calls = []

1064

def RemoveClient(self, dbus_path):

1065

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

1066

mandos = MockMandos()

1067

super(TestRemoveCmd, self).setUp()

1068

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

1069

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

1070

for clientpath in self.clients:

1071

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

1072

mandos.calls)

1073

1074

class TestApproveCmd(TestCmd):

1075

def test_approve(self):

1076

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

1077

for clientpath in self.clients:

1078

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

1079

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

1080

client.calls)

1081

1082

class TestDenyCmd(TestCmd):

1083

def test_deny(self):

1084

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

1085

for clientpath in self.clients:

1086

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

1087

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

1088

client.calls)

1089

1090

class TestEnableCmd(TestCmd):

1091

def test_enable(self):

1092

for clientpath in self.clients:

1093

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

1094

client.attributes["Enabled"] = False

1095

1096

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

1097

1098

for clientpath in self.clients:

1099

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

1100

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

1101

1102

class TestDisableCmd(TestCmd):

1103

def test_disable(self):

1104

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

1105

for clientpath in self.clients:

1106

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

1107

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

1108

1109

class Unique(object):

1110

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

1111

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

1112

1113

class TestPropertyCmd(TestCmd):

1114

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

1115

def runTest(self):

1116

if not hasattr(self, "command"):

1117

return

1118

values_to_get = getattr(self, "values_to_get",

1119

self.values_to_set)

1120

for value_to_set, value_to_get in zip(self.values_to_set,

1121

values_to_get):

1122

for clientpath in self.clients:

1123

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

1124

old_value = client.attributes[self.propname]

1125

self.assertNotIsInstance(old_value, Unique)

1126

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

1127

self.run_command(value_to_set, self.clients)

1128

for clientpath in self.clients:

1129

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

1130

value = client.attributes[self.propname]

1131

self.assertNotIsInstance(value, Unique)

1132

self.assertEqual(value, value_to_get)

1133

def run_command(self, value, clients):

1134

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

1135

1136

class TestBumpTimeoutCmd(TestPropertyCmd):

1137

command = BumpTimeoutCmd

1138

propname = "LastCheckedOK"

1139

values_to_set = [""]

1140

1141

class TestStartCheckerCmd(TestPropertyCmd):

1142

command = StartCheckerCmd

1143

propname = "CheckerRunning"

1144

values_to_set = [dbus.Boolean(True)]

1145

1146

class TestStopCheckerCmd(TestPropertyCmd):

1147

command = StopCheckerCmd

1148

propname = "CheckerRunning"

1149

values_to_set = [dbus.Boolean(False)]

1150

1151

class TestApproveByDefaultCmd(TestPropertyCmd):

1152

command = ApproveByDefaultCmd

1153

propname = "ApprovedByDefault"

1154

values_to_set = [dbus.Boolean(True)]

1155

1156

class TestDenyByDefaultCmd(TestPropertyCmd):

1157

command = DenyByDefaultCmd

1158

propname = "ApprovedByDefault"

1159

values_to_set = [dbus.Boolean(False)]

1160

1161

class TestPropertyValueCmd(TestPropertyCmd):

1162

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

1163

def runTest(self):

1164

if type(self) is TestPropertyValueCmd:

1165

return

1166

return super(TestPropertyValueCmd, self).runTest()

1167

def run_command(self, value, clients):

1168

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

1169

1170

class TestSetCheckerCmd(TestPropertyValueCmd):

1171

command = SetCheckerCmd

1172

propname = "Checker"

1173

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

1174

1175

class TestSetHostCmd(TestPropertyValueCmd):

1176

command = SetHostCmd

1177

propname = "Host"

1178

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

1179

1180

class TestSetSecretCmd(TestPropertyValueCmd):

1181

command = SetSecretCmd

1182

propname = "Secret"

1183

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

1184

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

1185

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

1186

1187

class TestSetTimeoutCmd(TestPropertyValueCmd):

1188

command = SetTimeoutCmd

1189

propname = "Timeout"

1190

values_to_set = [datetime.timedelta(),

1191

datetime.timedelta(minutes=5),

1192

datetime.timedelta(seconds=1),

1193

datetime.timedelta(weeks=1),

1194

datetime.timedelta(weeks=52)]

1195

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

1196

1197

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1198

command = SetExtendedTimeoutCmd

1199

propname = "ExtendedTimeout"

1200

values_to_set = [datetime.timedelta(),

1201

datetime.timedelta(minutes=5),

1202

datetime.timedelta(seconds=1),

1203

datetime.timedelta(weeks=1),

1204

datetime.timedelta(weeks=52)]

1205

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

1206

1207

class TestSetIntervalCmd(TestPropertyValueCmd):

1208

command = SetIntervalCmd

1209

propname = "Interval"

1210

values_to_set = [datetime.timedelta(),

1211

datetime.timedelta(minutes=5),

1212

datetime.timedelta(seconds=1),

1213

datetime.timedelta(weeks=1),

1214

datetime.timedelta(weeks=52)]

1215

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

1216

1217

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1218

command = SetApprovalDelayCmd

1219

propname = "ApprovalDelay"

1220

values_to_set = [datetime.timedelta(),

1221

datetime.timedelta(minutes=5),

1222

datetime.timedelta(seconds=1),

1223

datetime.timedelta(weeks=1),

1224

datetime.timedelta(weeks=52)]

1225

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

1226

1227

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1228

command = SetApprovalDurationCmd

1229

propname = "ApprovalDuration"

1230

values_to_set = [datetime.timedelta(),

1231

datetime.timedelta(minutes=5),

1232

datetime.timedelta(seconds=1),

1233

datetime.timedelta(weeks=1),

1234

datetime.timedelta(weeks=52)]

1235

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

1236

1237

class Test_command_from_options(unittest.TestCase):

1238

def setUp(self):

1239

self.parser = argparse.ArgumentParser()

1240

add_command_line_options(self.parser)

1241

def assert_command_from_args(self, args, command_cls,

1242

**cmd_attrs):

1243

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

1244

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

1245

options = self.parser.parse_args(args)

1246

check_option_syntax(self.parser, options)

1247

commands = commands_from_options(options)

1248

self.assertEqual(len(commands), 1)

1249

command = commands[0]

1250

self.assertIsInstance(command, command_cls)

1251

for key, value in cmd_attrs.items():

1252

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

1253

def test_print_table(self):

1254

self.assert_command_from_args([], PrintTableCmd,

1255

verbose=False)

1256

1257

def test_print_table_verbose(self):

1258

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

1259

verbose=True)

1260

1261

def test_print_table_verbose_short(self):

1262

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

1263

verbose=True)

1264

1265

def test_enable(self):

1266

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

1267

1268

def test_enable_short(self):

1269

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

1270

1271

def test_disable(self):

1272

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

1273

DisableCmd)

1274

1275

def test_disable_short(self):

1276

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

1277

1278

def test_bump_timeout(self):

1279

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

1280

BumpTimeoutCmd)

1281

1282

def test_bump_timeout_short(self):

1283

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

1284

1285

def test_start_checker(self):

1286

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

1287

StartCheckerCmd)

1288

1289

def test_stop_checker(self):

1290

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

1291

StopCheckerCmd)

1292

1293

def test_remove(self):

1294

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

1295

RemoveCmd)

1296

1297

def test_remove_short(self):

1298

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

1299

1300

def test_checker(self):

1301

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

1302

SetCheckerCmd, value_to_set=":")

1303

1304

def test_checker_empty(self):

1305

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

1306

SetCheckerCmd, value_to_set="")

1307

1308

def test_checker_short(self):

1309

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

1310

SetCheckerCmd, value_to_set=":")

1311

1312

def test_timeout(self):

1313

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

1314

SetTimeoutCmd,

1315

value_to_set=300000)

1316

1317

def test_timeout_short(self):

1318

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

1319

SetTimeoutCmd,

1320

value_to_set=300000)

1321

1322

def test_extended_timeout(self):

1323

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

1324

"foo"],

1325

SetExtendedTimeoutCmd,

1326

value_to_set=900000)

1327

1328

def test_interval(self):

1329

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

1330

SetIntervalCmd,

1331

value_to_set=120000)

1332

1333

def test_interval_short(self):

1334

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

1335

SetIntervalCmd,

1336

value_to_set=120000)

1337

1338

def test_approve_by_default(self):

1339

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

1340

ApproveByDefaultCmd)

1341

1342

def test_deny_by_default(self):

1343

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

1344

DenyByDefaultCmd)

1345

1346

def test_approval_delay(self):

1347

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

1348

"foo"], SetApprovalDelayCmd,

1349

value_to_set=30000)

1350

1351

def test_approval_duration(self):

1352

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

1353

"foo"], SetApprovalDurationCmd,

1354

value_to_set=1000)

1355

1356

def test_host(self):

1357

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

1358

"foo"], SetHostCmd,

1359

value_to_set="foo.example.org")

1360

1361

def test_host_short(self):

1362

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

1363

"foo"], SetHostCmd,

1364

value_to_set="foo.example.org")

1365

1366

def test_secret_devnull(self):

1367

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

1368

"foo"], SetSecretCmd,

1369

value_to_set=b"")

1370

1371

def test_secret_tempfile(self):

1372

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

1373

value = b"secret\0xyzzy\nbar"

1374

f.write(value)

1375

f.seek(0)

1376

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

1377

"foo"], SetSecretCmd,

1378

value_to_set=value)

1379

1380

def test_secret_devnull_short(self):

1381

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

1382

SetSecretCmd, value_to_set=b"")

1383

1384

def test_secret_tempfile_short(self):

1385

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

1386

value = b"secret\0xyzzy\nbar"

1387

f.write(value)

1388

f.seek(0)

1389

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

1390

SetSecretCmd,

1391

value_to_set=value)

1392

1393

def test_approve(self):

1394

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

1395

ApproveCmd)

1396

1397

def test_approve_short(self):

1398

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

1399

1400

def test_deny(self):

1401

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

1402

1403

def test_deny_short(self):

1404

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

1405

1406

def test_dump_json(self):

1407

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

1408

1409

def test_is_enabled(self):

1410

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

1411

IsEnabledCmd)

1412

1413

def test_is_enabled_short(self):

1414

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

1415

1416

def test_deny_before_remove(self):

1417

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

1418

"foo"])

1419

check_option_syntax(self.parser, options)

1420

commands = commands_from_options(options)

1421

self.assertEqual(len(commands), 2)

1422

self.assertIsInstance(commands[0], DenyCmd)

1423

self.assertIsInstance(commands[1], RemoveCmd)

1424

1425

def test_deny_before_remove_reversed(self):

1426

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

1427

"--all"])

1428

check_option_syntax(self.parser, options)

1429

commands = commands_from_options(options)

1430

self.assertEqual(len(commands), 2)

1431

self.assertIsInstance(commands[0], DenyCmd)

1432

self.assertIsInstance(commands[1], RemoveCmd)

1433

1434

1435

class Test_check_option_syntax(unittest.TestCase):

1436

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

1437

# check_option_syntax()

1438

actions = {

1439

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

1440

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

1441

# never used

1442

"enable": True,

1443

"disable": True,

1444

"bump_timeout": True,

1445

"start_checker": True,

1446

"stop_checker": True,

1447

"is_enabled": True,

1448

"remove": True,

1449

"checker": "x",

1450

"timeout": datetime.timedelta(),

1451

"extended_timeout": datetime.timedelta(),

1452

"interval": datetime.timedelta(),

1453

"approved_by_default": True,

1454

"approval_delay": datetime.timedelta(),

1455

"approval_duration": datetime.timedelta(),

1456

"host": "x",

1457

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

1458

"approve": True,

1459

"deny": True,

1460

}

1461

1462

def setUp(self):

1463

self.parser = argparse.ArgumentParser()

1464

add_command_line_options(self.parser)

1465

1466

@contextlib.contextmanager

1467

def assertParseError(self):

1468

with self.assertRaises(SystemExit) as e:

1469

with self.temporarily_suppress_stderr():

1470

yield

1471

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

1472

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

1473

# /argparse.html#exiting-methods

1474

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

1475

1476

@staticmethod

1477

@contextlib.contextmanager

1478

def temporarily_suppress_stderr():

1479

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

1480

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

1481

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

1482

os.close(null)

1483

try:

1484

yield

1485

finally:

1486

# restore stderr

1487

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

1488

os.close(stderrcopy)

1489

1490

def check_option_syntax(self, options):

1491

check_option_syntax(self.parser, options)

1492

1493

def test_actions_requires_client_or_all(self):

1494

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

1495

options = self.parser.parse_args()

1496

setattr(options, action, value)

1497

with self.assertParseError():

1498

self.check_option_syntax(options)

1499

1500

def test_actions_conflicts_with_verbose(self):

1501

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

1502

options = self.parser.parse_args()

1503

setattr(options, action, value)

1504

options.verbose = True

1505

with self.assertParseError():

1506

self.check_option_syntax(options)

1507

1508

def test_dump_json_conflicts_with_verbose(self):

1509

options = self.parser.parse_args()

1510

options.dump_json = True

1511

options.verbose = True

1512

with self.assertParseError():

1513

self.check_option_syntax(options)

1514

1515

def test_dump_json_conflicts_with_action(self):

1516

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

1517

options = self.parser.parse_args()

1518

setattr(options, action, value)

1519

options.dump_json = True

1520

with self.assertParseError():

1521

self.check_option_syntax(options)

1522

1523

def test_all_can_not_be_alone(self):

1524

options = self.parser.parse_args()

1525

options.all = True

1526

with self.assertParseError():

1527

self.check_option_syntax(options)

1528

1529

def test_all_is_ok_with_any_action(self):

1530

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

1531

options = self.parser.parse_args()

1532

setattr(options, action, value)

1533

options.all = True

1534

self.check_option_syntax(options)

1535

1536

def test_is_enabled_fails_without_client(self):

1537

options = self.parser.parse_args()

1538

options.is_enabled = True

1539

with self.assertParseError():

1540

self.check_option_syntax(options)

1541

1542

def test_is_enabled_works_with_one_client(self):

1543

options = self.parser.parse_args()

1544

options.is_enabled = True

1545

options.client = ["foo"]

1546

self.check_option_syntax(options)

1547

1548

def test_is_enabled_fails_with_two_clients(self):

1549

options = self.parser.parse_args()

1550

options.is_enabled = True

1551

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

1552

with self.assertParseError():

1553

self.check_option_syntax(options)

1554

1555

def test_remove_can_only_be_combined_with_action_deny(self):

1556

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

1557

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

1558

continue

1559

options = self.parser.parse_args()

1560

setattr(options, action, value)

1561

options.all = True

1562

options.remove = True

1563

with self.assertParseError():

1564

self.check_option_syntax(options)

763

self.keywords = ["Attr1", "AttrTwo"]

764

self.clients = [

765

{

766

"Attr1": "x1",

767

"AttrTwo": "y1",

768

"AttrThree": "z1",

769

"Bool": dbus.Boolean(False),

770

"NonDbusBoolean": False,

771

"Integer": 0,

772

"Timeout": 0,

773

"Interval": 1000,

774

"ApprovalDelay": 2000,

775

"ApprovalDuration": 3000,

776

"ExtendedTimeout": 4000,

777

"String": "",

778

779

{

780

"Attr1": "x2",

781

"AttrTwo": "y2",

782

"AttrThree": "z2",

783

"Bool": dbus.Boolean(True),

784

"NonDbusBoolean": True,

785

"Integer": 1,

786

"Timeout": 93785000,

787

"Interval": 93786000,

788

"ApprovalDelay": 93787000,

789

"ApprovalDuration": 93788000,

790

"ExtendedTimeout": 93789000,

791

"String": "A huge string which will not fit," * 10,

792

793

]

794

def test_short_header(self):

795

text = str(TableOfClients(self.clients, self.keywords,

796

self.tablewords))

797

expected_text = """

798

X Yy

799

x1 y1

800

x2 y2

801

"""[1:-1]

802

self.assertEqual(text, expected_text)

803

def test_booleans(self):

804

keywords = ["Bool", "NonDbusBoolean"]

805

text = str(TableOfClients(self.clients, keywords,

806

self.tablewords))

807

expected_text = """

808

A D-BUS Boolean A Non-D-BUS Boolean

809

No False

810

Yes True

811

"""[1:-1]

812

self.assertEqual(text, expected_text)

813

def test_milliseconds_detection(self):

814

keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",

815

"ApprovalDuration", "ExtendedTimeout"]

816

text = str(TableOfClients(self.clients, keywords,

817

self.tablewords))

818

expected_text = """

819

An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5

820

0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04

821

1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09

822

"""[1:-1]

823

self.assertEqual(text, expected_text)

824

def test_empty_and_long_string_values(self):

825

keywords = ["String"]

826

text = str(TableOfClients(self.clients, keywords,

827

self.tablewords))

828

expected_text = """

829

A String

830

831

A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,

832

"""[1:-1]

833

self.assertEqual(text, expected_text)

1565

834

1566

835

1567

836

Older »