/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 dbus

270

268

return value

271

269

272

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

273

346

## Classes for commands.

274

347

275

348

# Abstract classes first

276

349

class Command(object):

277

350

"""Abstract class for commands"""

278

def run(self, mandos, clients):

351

def run(self, clients):

279

352

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

280

353

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

281

354

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

282

self.mandos = mandos

283

for client, properties in clients.items():

284

self.run_on_one_client(client, properties)

355

for client in clients:

356

self.run_on_one_client(client)

285

357

286

358

class PrintCmd(Command):

287

359

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

292

364

"LastApprovalRequest", "ApprovalDelay",

293

365

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

294

366

"Expires", "LastCheckerStatus")

295

def run(self, mandos, clients):

367

def run(self, clients):

296

368

print(self.output(clients))

297

369

298

370

class PropertyCmd(Command):

299

371

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

300

def run_on_one_client(self, client, properties):

372

def run_on_one_client(self, client):

301

373

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

302

374

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

303

375

dbus_interface=dbus.PROPERTIES_IFACE)

316

388

@value_to_set.setter

317

389

def value_to_set(self, value):

318

390

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

319

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

391

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

320

392

321

393

# Actual (non-abstract) command classes

322

394

323

395

class PrintTableCmd(PrintCmd):

324

396

def __init__(self, verbose=False):

325

397

self.verbose = verbose

326

327

398

def output(self, clients):

328

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

329

keywords = default_keywords

330

399

if self.verbose:

331

400

keywords = self.all_keywords

332

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

333

334

class TableOfClients(object):

335

tableheaders = {

336

"Name": "Name",

337

"Enabled": "Enabled",

338

"Timeout": "Timeout",

339

"LastCheckedOK": "Last Successful Check",

340

"LastApprovalRequest": "Last Approval Request",

341

"Created": "Created",

342

"Interval": "Interval",

343

"Host": "Host",

344

"Fingerprint": "Fingerprint",

345

"KeyID": "Key ID",

346

"CheckerRunning": "Check Is Running",

347

"LastEnabled": "Last Enabled",

348

"ApprovalPending": "Approval Is Pending",

349

"ApprovedByDefault": "Approved By Default",

350

"ApprovalDelay": "Approval Delay",

351

"ApprovalDuration": "Approval Duration",

352

"Checker": "Checker",

353

"ExtendedTimeout": "Extended Timeout",

354

"Expires": "Expires",

355

"LastCheckerStatus": "Last Checker Status",

356

}

357

358

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

359

self.clients = clients

360

self.keywords = keywords

361

if tableheaders is not None:

362

self.tableheaders = tableheaders

363

364

def __str__(self):

365

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

366

367

if sys.version_info.major == 2:

368

__unicode__ = __str__

369

def __str__(self):

370

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

371

372

def rows(self):

373

format_string = self.row_formatting_string()

374

rows = [self.header_line(format_string)]

375

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

376

for client in self.clients)

377

return rows

378

379

def row_formatting_string(self):

380

"Format string used to format table rows"

381

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

382

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

383

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

384

for client in self.clients)),

385

key=key)

386

for key in self.keywords)

387

388

def string_from_client(self, client, key):

389

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

390

391

@staticmethod

392

def valuetostring(value, keyword):

393

if isinstance(value, dbus.Boolean):

394

return "Yes" if value else "No"

395

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

396

"ApprovalDuration", "ExtendedTimeout"):

397

return milliseconds_to_string(value)

398

return str(value)

399

400

def header_line(self, format_string):

401

return format_string.format(**self.tableheaders)

402

403

def client_line(self, client, format_string):

404

return format_string.format(

405

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

406

for key in self.keywords})

407

408

401

else:

402

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

403

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

409

404

410

405

class DumpJSONCmd(PrintCmd):

411

406

def output(self, clients):

421

416

return value

422

417

423

418

class IsEnabledCmd(Command):

424

def run_on_one_client(self, client, properties):

425

if self.is_enabled(client, properties):

419

def run_on_one_client(self, client):

420

if self.is_enabled(client):

426

421

sys.exit(0)

427

422

sys.exit(1)

428

def is_enabled(self, client, properties):

429

return bool(properties["Enabled"])

423

def is_enabled(self, client):

424

return client.Get(client_interface, "Enabled",

425

dbus_interface=dbus.PROPERTIES_IFACE)

430

426

431

427

class RemoveCmd(Command):

432

def run_on_one_client(self, client, properties):

428

def __init__(self, mandos):

429

self.mandos = mandos

430

def run_on_one_client(self, client):

433

431

self.mandos.RemoveClient(client.__dbus_object_path__)

434

432

435

433

class ApproveCmd(Command):

436

def run_on_one_client(self, client, properties):

434

def run_on_one_client(self, client):

437

435

client.Approve(dbus.Boolean(True),

438

436

dbus_interface=client_interface)

439

437

440

438

class DenyCmd(Command):

441

def run_on_one_client(self, client, properties):

439

def run_on_one_client(self, client):

442

440

client.Approve(dbus.Boolean(False),

443

441

dbus_interface=client_interface)

444

442

477

475

property = "Host"

478

476

479

477

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

480

@property

481

def value_to_set(self):

482

return self._vts

483

@value_to_set.setter

484

def value_to_set(self, value):

485

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

486

self._vts = value.read()

487

value.close()

488

478

property = "Secret"

489

479

490

480

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

505

495

MillisecondsValueArgumentMixIn):

506

496

property = "ApprovalDuration"

507

497

508

def add_command_line_options(parser):

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

509

521

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

510

522

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

511

523

help="show version number and exit")

536

548

help="Remove client")

537

549

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

538

550

help="Set checker command for client")

539

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

551

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

540

552

help="Set timeout for client")

541

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

553

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

542

554

help="Set extended timeout for client")

543

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

555

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

544

556

help="Set checker interval for client")

545

557

approve_deny_default = parser.add_mutually_exclusive_group()

546

558

approve_deny_default.add_argument(

551

563

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

552

564

dest="approved_by_default",

553

565

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

554

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

566

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

555

567

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

556

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

568

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

557

569

help="Set duration of one client approval")

558

570

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

559

571

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

568

580

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

569

581

help="Run self-test")

570

582

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

571

572

573

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)

574

608

575

609

commands = []

576

610

584

618

commands.append(DisableCmd())

585

619

586

620

if options.bump_timeout:

587

commands.append(BumpTimeoutCmd())

621

commands.append(BumpTimeoutCmd(options.bump_timeout))

588

622

589

623

if options.start_checker:

590

624

commands.append(StartCheckerCmd())

596

630

commands.append(IsEnabledCmd())

597

631

598

632

if options.remove:

599

commands.append(RemoveCmd())

633

commands.append(RemoveCmd(mandos_serv))

600

634

601

635

if options.checker is not None:

602

commands.append(SetCheckerCmd(options.checker))

636

commands.append(SetCheckerCmd())

603

637

604

638

if options.timeout is not None:

605

639

commands.append(SetTimeoutCmd(options.timeout))

609

643

SetExtendedTimeoutCmd(options.extended_timeout))

610

644

611

645

if options.interval is not None:

612

commands.append(SetIntervalCmd(options.interval))

646

command.append(SetIntervalCmd(options.interval))

613

647

614

648

if options.approved_by_default is not None:

615

649

if options.approved_by_default:

616

commands.append(ApproveByDefaultCmd())

650

command.append(ApproveByDefaultCmd())

617

651

else:

618

commands.append(DenyByDefaultCmd())

652

command.append(DenyByDefaultCmd())

619

653

620

654

if options.approval_delay is not None:

621

commands.append(SetApprovalDelayCmd(options.approval_delay))

655

command.append(SetApprovalDelayCmd(options.approval_delay))

622

656

623

657

if options.approval_duration is not None:

624

commands.append(

658

command.append(

625

659

SetApprovalDurationCmd(options.approval_duration))

626

660

627

661

if options.host is not None:

628

commands.append(SetHostCmd(options.host))

662

command.append(SetHostCmd(options.host))

629

663

630

664

if options.secret is not None:

631

commands.append(SetSecretCmd(options.secret))

665

command.append(SetSecretCmd(options.secret))

632

666

633

667

if options.approve:

634

668

commands.append(ApproveCmd())

641

675

if not commands:

642

676

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

643

677

644

return commands

645

646

647

def main():

648

parser = argparse.ArgumentParser()

649

650

add_command_line_options(parser)

651

652

options = parser.parse_args()

653

654

def has_actions(options):

655

return any((options.enable,

656

options.disable,

657

options.bump_timeout,

658

options.start_checker,

659

options.stop_checker,

660

options.is_enabled,

661

options.remove,

662

options.checker is not None,

663

options.timeout is not None,

664

options.extended_timeout is not None,

665

options.interval is not None,

666

options.approved_by_default is not None,

667

options.approval_delay is not None,

668

options.approval_duration is not None,

669

options.host is not None,

670

options.secret is not None,

671

options.approve,

672

options.deny))

673

674

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

675

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

676

if options.verbose and has_actions(options):

677

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

678

if options.dump_json and (options.verbose

679

or has_actions(options)):

680

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

681

if options.all and not has_actions(options):

682

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

683

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

684

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

685

686

clientnames = options.client

687

688

try:

689

bus = dbus.SystemBus()

690

mandos_dbus_objc = bus.get_object(busname, server_path)

691

except dbus.exceptions.DBusException:

692

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

693

sys.exit(1)

694

695

mandos_serv = dbus.Interface(mandos_dbus_objc,

696

dbus_interface=server_interface)

697

mandos_serv_object_manager = dbus.Interface(

698

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

699

700

# Filter out log message from dbus module

701

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

702

class NullFilter(logging.Filter):

703

def filter(self, record):

704

return False

705

dbus_filter = NullFilter()

706

try:

707

dbus_logger.addFilter(dbus_filter)

708

mandos_clients = {path: ifs_and_props[client_interface]

709

for path, ifs_and_props in

710

mandos_serv_object_manager

711

.GetManagedObjects().items()

712

if client_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)

713

694

except dbus.exceptions.DBusException as e:

714

695

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

715

696

"\n%s", e)

716

697

sys.exit(1)

717

finally:

718

# restore dbus logger

719

dbus_logger.removeFilter(dbus_filter)

720

698

721

699

# Compile dict of (clients: properties) to process

722

700

clients = {}

723

701

724

if not clientnames:

702

if options.all or not options.client:

725

703

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

726

704

for path, properties in mandos_clients.items()}

727

705

else:

728

for name in clientnames:

706

for name in options.client:

729

707

for path, client in mandos_clients.items():

730

708

if client["Name"] == name:

731

709

client_objc = bus.get_object(busname, path)

736

714

sys.exit(1)

737

715

738

716

# Run all commands on clients

739

commands = commands_from_options(options)

740

717

for command in commands:

741

command.run(mandos_serv, clients)

718

command.run(clients)

742

719

743

720

744

721

class Test_milliseconds_to_string(unittest.TestCase):

756

733

757

734

class Test_string_to_delta(unittest.TestCase):

758

735

def test_handles_basic_rfc3339(self):

759

self.assertEqual(string_to_delta("PT0S"),

760

datetime.timedelta())

761

self.assertEqual(string_to_delta("P0D"),

762

datetime.timedelta())

763

self.assertEqual(string_to_delta("PT1S"),

764

datetime.timedelta(0, 1))

765

736

self.assertEqual(string_to_delta("PT2H"),

766

737

datetime.timedelta(0, 7200))

767

738

def test_falls_back_to_pre_1_6_1_with_warning(self):

770

741

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

771

742

value = string_to_delta("2h")

772

743

else:

773

class WarningFilter(logging.Filter):

774

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

775

def filter(self, record):

776

is_warning = record.levelno >= logging.WARNING

777

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

778

False)

779

return not is_warning

780

warning_filter = WarningFilter()

781

log.addFilter(warning_filter)

782

try:

783

value = string_to_delta("2h")

784

finally:

785

log.removeFilter(warning_filter)

786

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

744

value = string_to_delta("2h")

787

745

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

788

746

789

790

class TestCmd(unittest.TestCase):

791

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

792

def setUp(self):

793

testcase = self

794

class MockClient(object):

795

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

796

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

797

self.attributes = attributes

798

self.attributes["Name"] = name

799

self.calls = []

800

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

801

testcase.assertEqual(interface, client_interface)

802

testcase.assertEqual(dbus_interface,

803

dbus.PROPERTIES_IFACE)

804

self.attributes[property] = value

805

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

806

testcase.assertEqual(interface, client_interface)

807

testcase.assertEqual(dbus_interface,

808

dbus.PROPERTIES_IFACE)

809

return self.attributes[property]

810

def Approve(self, approve, dbus_interface):

811

testcase.assertEqual(dbus_interface, client_interface)

812

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

813

dbus_interface)))

814

self.client = MockClient(

815

"foo",

816

KeyID=("92ed150794387c03ce684574b1139a65"

817

"94a34f895daaaf09fd8ea90a27cddb12"),

818

Secret=b"secret",

819

Host="foo.example.org",

820

Enabled=dbus.Boolean(True),

821

Timeout=300000,

822

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

823

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

824

Interval=120000,

825

Fingerprint=("778827225BA7DE539C5A"

826

"7CFA59CFF7CDBD9A5920"),

827

CheckerRunning=dbus.Boolean(False),

828

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

829

ApprovalPending=dbus.Boolean(False),

830

ApprovedByDefault=dbus.Boolean(True),

831

LastApprovalRequest="",

832

ApprovalDelay=0,

833

ApprovalDuration=1000,

834

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

835

ExtendedTimeout=900000,

836

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

837

LastCheckerStatus=0)

838

self.other_client = MockClient(

839

"barbar",

840

KeyID=("0558568eedd67d622f5c83b35a115f79"

841

"6ab612cff5ad227247e46c2b020f441c"),

842

Secret=b"secretbar",

843

Host="192.0.2.3",

844

Enabled=dbus.Boolean(True),

845

Timeout=300000,

846

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

847

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

848

Interval=120000,

849

Fingerprint=("3E393AEAEFB84C7E89E2"

850

"F547B3A107558FCA3A27"),

851

CheckerRunning=dbus.Boolean(True),

852

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

853

ApprovalPending=dbus.Boolean(False),

854

ApprovedByDefault=dbus.Boolean(False),

855

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

856

ApprovalDelay=30000,

857

ApprovalDuration=1000,

858

Checker=":",

859

ExtendedTimeout=900000,

860

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

861

LastCheckerStatus=-2)

862

self.clients = collections.OrderedDict(

863

[

864

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

865

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

866

])

867

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

868

869

class TestPrintTableCmd(TestCmd):

870

def test_normal(self):

871

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

872

expected_output = """

873

Name Enabled Timeout Last Successful Check

874

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

875

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

876

"""[1:-1]

877

self.assertEqual(output, expected_output)

878

def test_verbose(self):

879

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

880

expected_output = """

881

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

882

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

883

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

884

"""[1:-1]

885

self.assertEqual(output, expected_output)

886

def test_one_client(self):

887

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

888

expected_output = """

889

Name Enabled Timeout Last Successful Check

890

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

891

"""[1:-1]

892

self.assertEqual(output, expected_output)

893

894

class TestDumpJSONCmd(TestCmd):

895

def setUp(self):

896

self.expected_json = {

897

"foo": {

898

"Name": "foo",

899

"KeyID": ("92ed150794387c03ce684574b1139a65"

900

"94a34f895daaaf09fd8ea90a27cddb12"),

901

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

902

"Enabled": True,

903

"Timeout": 300000,

904

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

905

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

906

"Interval": 120000,

907

"Fingerprint": ("778827225BA7DE539C5A"

908

"7CFA59CFF7CDBD9A5920"),

909

"CheckerRunning": False,

910

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

911

"ApprovalPending": False,

912

"ApprovedByDefault": True,

913

"LastApprovalRequest": "",

914

"ApprovalDelay": 0,

915

"ApprovalDuration": 1000,

916

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

917

"ExtendedTimeout": 900000,

918

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

919

"LastCheckerStatus": 0,

920

921

"barbar": {

922

"Name": "barbar",

923

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

924

"6ab612cff5ad227247e46c2b020f441c"),

925

"Host": "192.0.2.3",

926

"Enabled": True,

927

"Timeout": 300000,

928

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

929

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

930

"Interval": 120000,

931

"Fingerprint": ("3E393AEAEFB84C7E89E2"

932

"F547B3A107558FCA3A27"),

933

"CheckerRunning": True,

934

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

935

"ApprovalPending": False,

936

"ApprovedByDefault": False,

937

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

938

"ApprovalDelay": 30000,

939

"ApprovalDuration": 1000,

940

"Checker": ":",

941

"ExtendedTimeout": 900000,

942

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

943

"LastCheckerStatus": -2,

944

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",

945

762

}

946

return super(TestDumpJSONCmd, self).setUp()

947

def test_normal(self):

948

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

949

self.assertDictEqual(json_data, self.expected_json)

950

def test_one_client(self):

951

clients = self.one_client

952

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

953

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

954

self.assertDictEqual(json_data, expected_json)

955

956

class TestIsEnabledCmd(TestCmd):

957

def test_is_enabled(self):

958

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

959

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

960

def test_is_enabled_run_exits_successfully(self):

961

with self.assertRaises(SystemExit) as e:

962

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

963

if e.exception.code is not None:

964

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

965

else:

966

self.assertIsNone(e.exception.code)

967

def test_is_enabled_run_exits_with_failure(self):

968

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

969

with self.assertRaises(SystemExit) as e:

970

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

971

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

972

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

973

else:

974

self.assertIsNotNone(e.exception.code)

975

976

class TestRemoveCmd(TestCmd):

977

def test_remove(self):

978

class MockMandos(object):

979

def __init__(self):

980

self.calls = []

981

def RemoveClient(self, dbus_path):

982

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

983

mandos = MockMandos()

984

super(TestRemoveCmd, self).setUp()

985

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

986

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

987

for client in self.clients:

988

self.assertIn(("RemoveClient",

989

(client.__dbus_object_path__,)),

990

mandos.calls)

991

992

class TestApproveCmd(TestCmd):

993

def test_approve(self):

994

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

995

for client in self.clients:

996

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

997

client.calls)

998

999

class TestDenyCmd(TestCmd):

1000

def test_deny(self):

1001

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

1002

for client in self.clients:

1003

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

1004

client.calls)

1005

1006

class TestEnableCmd(TestCmd):

1007

def test_enable(self):

1008

for client in self.clients:

1009

client.attributes["Enabled"] = False

1010

1011

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

1012

1013

for client in self.clients:

1014

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

1015

1016

class TestDisableCmd(TestCmd):

1017

def test_disable(self):

1018

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

1019

1020

for client in self.clients:

1021

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

1022

1023

class Unique(object):

1024

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

1025

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

1026

1027

class TestPropertyCmd(TestCmd):

1028

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

1029

def runTest(self):

1030

if not hasattr(self, "command"):

1031

return

1032

values_to_get = getattr(self, "values_to_get",

1033

self.values_to_set)

1034

for value_to_set, value_to_get in zip(self.values_to_set,

1035

values_to_get):

1036

for client in self.clients:

1037

old_value = client.attributes[self.property]

1038

self.assertNotIsInstance(old_value, Unique)

1039

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

1040

self.run_command(value_to_set, self.clients)

1041

for client in self.clients:

1042

value = client.attributes[self.property]

1043

self.assertNotIsInstance(value, Unique)

1044

self.assertEqual(value, value_to_get)

1045

def run_command(self, value, clients):

1046

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

1047

1048

class TestBumpTimeoutCmd(TestPropertyCmd):

1049

command = BumpTimeoutCmd

1050

property = "LastCheckedOK"

1051

values_to_set = [""]

1052

1053

class TestStartCheckerCmd(TestPropertyCmd):

1054

command = StartCheckerCmd

1055

property = "CheckerRunning"

1056

values_to_set = [dbus.Boolean(True)]

1057

1058

class TestStopCheckerCmd(TestPropertyCmd):

1059

command = StopCheckerCmd

1060

property = "CheckerRunning"

1061

values_to_set = [dbus.Boolean(False)]

1062

1063

class TestApproveByDefaultCmd(TestPropertyCmd):

1064

command = ApproveByDefaultCmd

1065

property = "ApprovedByDefault"

1066

values_to_set = [dbus.Boolean(True)]

1067

1068

class TestDenyByDefaultCmd(TestPropertyCmd):

1069

command = DenyByDefaultCmd

1070

property = "ApprovedByDefault"

1071

values_to_set = [dbus.Boolean(False)]

1072

1073

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1074

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

1075

ValueArgumentMixIn"""

1076

def runTest(self):

1077

if type(self) is TestValueArgumentPropertyCmd:

1078

return

1079

return super(TestValueArgumentPropertyCmd, self).runTest()

1080

def run_command(self, value, clients):

1081

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

1082

1083

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1084

command = SetCheckerCmd

1085

property = "Checker"

1086

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

1087

1088

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1089

command = SetHostCmd

1090

property = "Host"

1091

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

1092

1093

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

command = SetSecretCmd

1095

property = "Secret"

1096

values_to_set = [open("/dev/null", "rb"),

1097

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

1098

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

1099

1100

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1101

command = SetTimeoutCmd

1102

property = "Timeout"

1103

values_to_set = [datetime.timedelta(),

1104

datetime.timedelta(minutes=5),

1105

datetime.timedelta(seconds=1),

1106

datetime.timedelta(weeks=1),

1107

datetime.timedelta(weeks=52)]

1108

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

1109

1110

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1111

command = SetExtendedTimeoutCmd

1112

property = "ExtendedTimeout"

1113

values_to_set = [datetime.timedelta(),

1114

datetime.timedelta(minutes=5),

1115

datetime.timedelta(seconds=1),

1116

datetime.timedelta(weeks=1),

1117

datetime.timedelta(weeks=52)]

1118

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

1119

1120

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1121

command = SetIntervalCmd

1122

property = "Interval"

1123

values_to_set = [datetime.timedelta(),

1124

datetime.timedelta(minutes=5),

1125

datetime.timedelta(seconds=1),

1126

datetime.timedelta(weeks=1),

1127

datetime.timedelta(weeks=52)]

1128

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

1129

1130

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1131

command = SetApprovalDelayCmd

1132

property = "ApprovalDelay"

1133

values_to_set = [datetime.timedelta(),

1134

datetime.timedelta(minutes=5),

1135

datetime.timedelta(seconds=1),

1136

datetime.timedelta(weeks=1),

1137

datetime.timedelta(weeks=52)]

1138

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

1139

1140

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1141

command = SetApprovalDurationCmd

1142

property = "ApprovalDuration"

1143

values_to_set = [datetime.timedelta(),

1144

datetime.timedelta(minutes=5),

1145

datetime.timedelta(seconds=1),

1146

datetime.timedelta(weeks=1),

1147

datetime.timedelta(weeks=52)]

1148

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

1149

1150

class Test_command_from_options(unittest.TestCase):

1151

def setUp(self):

1152

self.parser = argparse.ArgumentParser()

1153

add_command_line_options(self.parser)

1154

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

1155

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

1156

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

1157

options = self.parser.parse_args(args)

1158

commands = commands_from_options(options)

1159

self.assertEqual(len(commands), 1)

1160

command = commands[0]

1161

self.assertIsInstance(command, command_cls)

1162

for key, value in cmd_attrs.items():

1163

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

1164

def test_print_table(self):

1165

self.assert_command_from_args([], PrintTableCmd,

1166

verbose=False)

1167

1168

def test_print_table_verbose(self):

1169

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

1170

verbose=True)

1171

1172

def test_print_table_verbose_short(self):

1173

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

1174

verbose=True)

1175

1176

def test_enable(self):

1177

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

1178

1179

def test_enable_short(self):

1180

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

1181

1182

def test_disable(self):

1183

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

1184

DisableCmd)

1185

1186

def test_disable_short(self):

1187

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

1188

1189

def test_bump_timeout(self):

1190

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

1191

BumpTimeoutCmd)

1192

1193

def test_bump_timeout_short(self):

1194

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

1195

1196

def test_start_checker(self):

1197

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

1198

StartCheckerCmd)

1199

1200

def test_stop_checker(self):

1201

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

1202

StopCheckerCmd)

1203

1204

def test_remove(self):

1205

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

1206

RemoveCmd)

1207

1208

def test_remove_short(self):

1209

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

1210

1211

def test_checker(self):

1212

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

1213

SetCheckerCmd, value_to_set=":")

1214

1215

def test_checker_empty(self):

1216

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

1217

SetCheckerCmd, value_to_set="")

1218

1219

def test_checker_short(self):

1220

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

1221

SetCheckerCmd, value_to_set=":")

1222

1223

def test_timeout(self):

1224

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

1225

SetTimeoutCmd,

1226

value_to_set=300000)

1227

1228

def test_timeout_short(self):

1229

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

1230

SetTimeoutCmd,

1231

value_to_set=300000)

1232

1233

def test_extended_timeout(self):

1234

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

1235

"foo"],

1236

SetExtendedTimeoutCmd,

1237

value_to_set=900000)

1238

1239

def test_interval(self):

1240

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

1241

SetIntervalCmd,

1242

value_to_set=120000)

1243

1244

def test_interval_short(self):

1245

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

1246

SetIntervalCmd,

1247

value_to_set=120000)

1248

1249

def test_approve_by_default(self):

1250

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

1251

ApproveByDefaultCmd)

1252

1253

def test_deny_by_default(self):

1254

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

1255

DenyByDefaultCmd)

1256

1257

def test_approval_delay(self):

1258

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

1259

"foo"], SetApprovalDelayCmd,

1260

value_to_set=30000)

1261

1262

def test_approval_duration(self):

1263

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

1264

"foo"], SetApprovalDurationCmd,

1265

value_to_set=1000)

1266

1267

def test_host(self):

1268

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

1269

"foo"], SetHostCmd,

1270

value_to_set="foo.example.org")

1271

1272

def test_host_short(self):

1273

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

1274

"foo"], SetHostCmd,

1275

value_to_set="foo.example.org")

1276

1277

def test_secret_devnull(self):

1278

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

1279

"foo"], SetSecretCmd,

1280

value_to_set=b"")

1281

1282

def test_secret_tempfile(self):

1283

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

1284

value = b"secret\0xyzzy\nbar"

1285

f.write(value)

1286

f.seek(0)

1287

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

1288

"foo"], SetSecretCmd,

1289

value_to_set=value)

1290

1291

def test_secret_devnull_short(self):

1292

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

1293

SetSecretCmd, value_to_set=b"")

1294

1295

def test_secret_tempfile_short(self):

1296

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

1297

value = b"secret\0xyzzy\nbar"

1298

f.write(value)

1299

f.seek(0)

1300

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

1301

SetSecretCmd,

1302

value_to_set=value)

1303

1304

def test_approve(self):

1305

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

1306

ApproveCmd)

1307

1308

def test_approve_short(self):

1309

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

1310

1311

def test_deny(self):

1312

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

1313

1314

def test_deny_short(self):

1315

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

1316

1317

def test_dump_json(self):

1318

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

1319

1320

def test_is_enabled(self):

1321

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

1322

IsEnabledCmd)

1323

1324

def test_is_enabled_short(self):

1325

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

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)

1326

834

1327

835

1328

836

Older »