/mandos/trunk : revision 1073

To get this branch, use:

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

« back to all changes in this revision

Viewing changes to mandos-ctl

Committer: Teddy Hogeborn
Date: 2019-03-16 04:38:35 UTC
Revision ID: teddy@recompile.se-20190316043835-xlmz9xse3bh1u5u6

mandos-ctl: Refactor tests

* mandos-ctl (Test_get_managed_objects): Remove some useless
"@staticmethod"s.

files modified:
TODO

mandos

mandos-ctl

Show diffs side-by-side

added added

removed removed

mandos-ctl

if sys.version_info.major == 2:

str = unicode

import StringIO

io.StringIO = StringIO.StringIO

locale.setlocale(locale.LC_ALL, "")

def main():

parser = argparse.ArgumentParser()

add_command_line_options(parser)

options = parser.parse_args()

check_option_syntax(parser, options)

clientnames = options.client

125

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

126

sys.exit(1)

127

128

# Run all commands on clients

128

129

commands = commands_from_options(options)

129

130

for command in commands:

131

command.run(clients, bus, mandos_serv)

132

232

>>> rfc3339_duration_to_delta("")

233

Traceback (most recent call last):

234

...

235

ValueError: Invalid RFC 3339 duration: ""

235

ValueError: Invalid RFC 3339 duration: u''

236

>>> # Must start with "P":

237

>>> rfc3339_duration_to_delta("1D")

238

Traceback (most recent call last):

239

...

240

ValueError: Invalid RFC 3339 duration: "1D"

240

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

241

>>> # Must use correct order

242

>>> rfc3339_duration_to_delta("PT1S2M")

243

Traceback (most recent call last):

244

...

245

ValueError: Invalid RFC 3339 duration: "PT1S2M"

245

ValueError: Invalid RFC 3339 duration: u'PT1S2M'

246

>>> # Time needs time marker

247

>>> rfc3339_duration_to_delta("P1H2S")

248

Traceback (most recent call last):

249

...

250

ValueError: Invalid RFC 3339 duration: "P1H2S"

250

ValueError: Invalid RFC 3339 duration: u'P1H2S'

251

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

252

>>> rfc3339_duration_to_delta("P1D2W")

253

Traceback (most recent call last):

254

...

255

ValueError: Invalid RFC 3339 duration: "P1D2W"

255

ValueError: Invalid RFC 3339 duration: u'P1D2W'

256

>>> rfc3339_duration_to_delta("P2W2H")

257

Traceback (most recent call last):

258

...

259

ValueError: Invalid RFC 3339 duration: "P2W2H"

259

ValueError: Invalid RFC 3339 duration: u'P2W2H'

260

"""

261

262

# Parsing an RFC 3339 duration with regular expressions is not

333

break

334

else:

335

# No currently valid tokens were found

336

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

336

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

337

.format(duration))

338

# End token found

339

return value

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

463

464

465

class NullFilter(logging.Filter):

465

466

def filter(self, record):

476

477

commands = []

477

478

479

if options.is_enabled:

479

commands.append(command.IsEnabled())

480

commands.append(IsEnabledCmd())

480

481

482

if options.approve:

482

commands.append(command.Approve())

483

commands.append(ApproveCmd())

483

484

485

if options.deny:

485

commands.append(command.Deny())

486

commands.append(DenyCmd())

486

487

488

if options.remove:

488

commands.append(command.Remove())

489

commands.append(RemoveCmd())

489

490

491

if options.dump_json:

491

commands.append(command.DumpJSON())

492

commands.append(DumpJSONCmd())

492

493

494

if options.enable:

494

commands.append(command.Enable())

495

commands.append(EnableCmd())

495

496

497

if options.disable:

497

commands.append(command.Disable())

498

commands.append(DisableCmd())

498

499

500

if options.bump_timeout:

500

commands.append(command.BumpTimeout())

501

commands.append(BumpTimeoutCmd())

501

502

503

if options.start_checker:

503

commands.append(command.StartChecker())

504

commands.append(StartCheckerCmd())

504

505

506

if options.stop_checker:

506

commands.append(command.StopChecker())

507

commands.append(StopCheckerCmd())

507

508

509

if options.approved_by_default is not None:

509

510

if options.approved_by_default:

510

commands.append(command.ApproveByDefault())

511

commands.append(ApproveByDefaultCmd())

511

512

else:

512

commands.append(command.DenyByDefault())

513

commands.append(DenyByDefaultCmd())

513

514

515

if options.checker is not None:

515

commands.append(command.SetChecker(options.checker))

516

commands.append(SetCheckerCmd(options.checker))

516

517

518

if options.host is not None:

518

commands.append(command.SetHost(options.host))

519

commands.append(SetHostCmd(options.host))

519

520

521

if options.secret is not None:

521

commands.append(command.SetSecret(options.secret))

522

commands.append(SetSecretCmd(options.secret))

522

523

524

if options.timeout is not None:

524

commands.append(command.SetTimeout(options.timeout))

525

commands.append(SetTimeoutCmd(options.timeout))

525

526

527

if options.extended_timeout:

527

528

commands.append(

528

command.SetExtendedTimeout(options.extended_timeout))

529

SetExtendedTimeoutCmd(options.extended_timeout))

529

530

531

if options.interval is not None:

531

commands.append(command.SetInterval(options.interval))

532

commands.append(SetIntervalCmd(options.interval))

532

533

534

if options.approval_delay is not None:

534

commands.append(

535

command.SetApprovalDelay(options.approval_delay))

535

commands.append(SetApprovalDelayCmd(options.approval_delay))

536

537

if options.approval_duration is not None:

538

commands.append(

539

command.SetApprovalDuration(options.approval_duration))

539

SetApprovalDurationCmd(options.approval_duration))

540

541

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

542

# optionally verbosely

543

if not commands:

544

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

544

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

545

546

return commands

547

548

549

class command(object):

550

"""A namespace for command classes"""

551

552

class Base(object):

553

"""Abstract base class for commands"""

554

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

555

"""Normal commands should implement run_on_one_client(),

556

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

557

override this run() method instead.

558

"""

559

self.mandos = mandos

560

for clientpath, properties in clients.items():

561

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

562

dbus_busname, str(clientpath))

563

client = bus.get_object(dbus_busname, clientpath)

564

self.run_on_one_client(client, properties)

565

566

567

class IsEnabled(Base):

568

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

569

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

570

if self.is_enabled(client, properties):

571

sys.exit(0)

572

sys.exit(1)

573

def is_enabled(self, client, properties):

574

return properties["Enabled"]

575

576

577

class Approve(Base):

578

def run_on_one_client(self, client, properties):

579

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

580

client.__dbus_object_path__,

581

client_dbus_interface)

582

client.Approve(dbus.Boolean(True),

583

dbus_interface=client_dbus_interface)

584

585

586

class Deny(Base):

587

def run_on_one_client(self, client, properties):

588

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

589

client.__dbus_object_path__,

590

client_dbus_interface)

591

client.Approve(dbus.Boolean(False),

592

dbus_interface=client_dbus_interface)

593

594

595

class Remove(Base):

596

def run_on_one_client(self, client, properties):

597

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

598

dbus_busname, server_dbus_path,

599

server_dbus_interface,

600

str(client.__dbus_object_path__))

601

self.mandos.RemoveClient(client.__dbus_object_path__)

602

603

604

class Output(Base):

605

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

606

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

607

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

608

"Fingerprint", "CheckerRunning",

609

"LastEnabled", "ApprovalPending",

610

"ApprovedByDefault", "LastApprovalRequest",

611

"ApprovalDelay", "ApprovalDuration",

612

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

613

"LastCheckerStatus")

614

615

616

class DumpJSON(Output):

617

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

618

data = {client["Name"]:

619

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

620

for key in self.all_keywords}

621

for client in clients.values()}

622

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

549

class Command(object):

550

"""Abstract class for commands"""

551

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

552

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

553

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

554

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

555

self.mandos = mandos

556

for clientpath, properties in clients.items():

557

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

558

dbus_busname, str(clientpath))

559

client = bus.get_object(dbus_busname, clientpath)

560

self.run_on_one_client(client, properties)

561

562

563

class IsEnabledCmd(Command):

564

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

565

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

566

if self.is_enabled(client, properties):

567

sys.exit(0)

568

sys.exit(1)

569

def is_enabled(self, client, properties):

570

return properties["Enabled"]

571

572

573

class ApproveCmd(Command):

574

def run_on_one_client(self, client, properties):

575

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

576

client.__dbus_object_path__, client_dbus_interface)

577

client.Approve(dbus.Boolean(True),

578

dbus_interface=client_dbus_interface)

579

580

581

class DenyCmd(Command):

582

def run_on_one_client(self, client, properties):

583

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

584

client.__dbus_object_path__, client_dbus_interface)

585

client.Approve(dbus.Boolean(False),

586

dbus_interface=client_dbus_interface)

587

588

589

class RemoveCmd(Command):

590

def run_on_one_client(self, client, properties):

591

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

592

server_dbus_path, server_dbus_interface,

593

str(client.__dbus_object_path__))

594

self.mandos.RemoveClient(client.__dbus_object_path__)

595

596

597

class OutputCmd(Command):

598

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

599

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

600

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

601

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

602

"ApprovalPending", "ApprovedByDefault",

603

"LastApprovalRequest", "ApprovalDelay",

604

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

605

"Expires", "LastCheckerStatus")

606

607

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

608

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

609

610

def output(self, clients):

611

raise NotImplementedError()

612

613

614

class DumpJSONCmd(OutputCmd):

615

def output(self, clients):

616

data = {client["Name"]:

617

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

618

for key in self.all_keywords}

619

for client in clients}

620

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

621

622

@staticmethod

623

def dbus_boolean_to_bool(value):

624

if isinstance(value, dbus.Boolean):

625

value = bool(value)

626

return value

627

628

629

class PrintTableCmd(OutputCmd):

630

def __init__(self, verbose=False):

631

self.verbose = verbose

632

633

def output(self, clients):

634

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

635

"LastCheckedOK")

636

keywords = default_keywords

637

if self.verbose:

638

keywords = self.all_keywords

639

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

640

641

class TableOfClients(object):

642

tableheaders = {

643

"Name": "Name",

644

"Enabled": "Enabled",

645

"Timeout": "Timeout",

646

"LastCheckedOK": "Last Successful Check",

647

"LastApprovalRequest": "Last Approval Request",

648

"Created": "Created",

649

"Interval": "Interval",

650

"Host": "Host",

651

"Fingerprint": "Fingerprint",

652

"KeyID": "Key ID",

653

"CheckerRunning": "Check Is Running",

654

"LastEnabled": "Last Enabled",

655

"ApprovalPending": "Approval Is Pending",

656

"ApprovedByDefault": "Approved By Default",

657

"ApprovalDelay": "Approval Delay",

658

"ApprovalDuration": "Approval Duration",

659

"Checker": "Checker",

660

"ExtendedTimeout": "Extended Timeout",

661

"Expires": "Expires",

662

"LastCheckerStatus": "Last Checker Status",

663

}

664

665

def __init__(self, clients, keywords):

666

self.clients = clients

667

self.keywords = keywords

668

669

def __str__(self):

670

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

671

672

if sys.version_info.major == 2:

673

__unicode__ = __str__

674

def __str__(self):

675

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

676

677

def rows(self):

678

format_string = self.row_formatting_string()

679

rows = [self.header_line(format_string)]

680

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

681

for client in self.clients)

682

return rows

683

684

def row_formatting_string(self):

685

"Format string used to format table rows"

686

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

687

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

688

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

689

for client in self.clients)),

690

key=key)

691

for key in self.keywords)

692

693

def string_from_client(self, client, key):

694

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

695

696

@classmethod

697

def valuetostring(cls, value, keyword):

698

if isinstance(value, dbus.Boolean):

699

return "Yes" if value else "No"

700

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

701

"ApprovalDuration", "ExtendedTimeout"):

702

return cls.milliseconds_to_string(value)

703

return str(value)

704

705

def header_line(self, format_string):

706

return format_string.format(**self.tableheaders)

707

708

def client_line(self, client, format_string):

709

return format_string.format(

710

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

711

for key in self.keywords})

623

712

624

713

@staticmethod

625

def dbus_boolean_to_bool(value):

626

if isinstance(value, dbus.Boolean):

627

value = bool(value)

628

return value

629

630

631

class PrintTable(Output):

632

def __init__(self, verbose=False):

633

self.verbose = verbose

634

635

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

636

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

637

"LastCheckedOK")

638

keywords = default_keywords

639

if self.verbose:

640

keywords = self.all_keywords

641

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

642

643

class TableOfClients(object):

644

tableheaders = {

645

"Name": "Name",

646

"Enabled": "Enabled",

647

"Timeout": "Timeout",

648

"LastCheckedOK": "Last Successful Check",

649

"LastApprovalRequest": "Last Approval Request",

650

"Created": "Created",

651

"Interval": "Interval",

652

"Host": "Host",

653

"Fingerprint": "Fingerprint",

654

"KeyID": "Key ID",

655

"CheckerRunning": "Check Is Running",

656

"LastEnabled": "Last Enabled",

657

"ApprovalPending": "Approval Is Pending",

658

"ApprovedByDefault": "Approved By Default",

659

"ApprovalDelay": "Approval Delay",

660

"ApprovalDuration": "Approval Duration",

661

"Checker": "Checker",

662

"ExtendedTimeout": "Extended Timeout",

663

"Expires": "Expires",

664

"LastCheckerStatus": "Last Checker Status",

665

}

666

667

def __init__(self, clients, keywords):

668

self.clients = clients

669

self.keywords = keywords

670

671

def __str__(self):

672

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

673

674

if sys.version_info.major == 2:

675

__unicode__ = __str__

676

def __str__(self):

677

return str(self).encode(

678

locale.getpreferredencoding())

679

680

def rows(self):

681

format_string = self.row_formatting_string()

682

rows = [self.header_line(format_string)]

683

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

684

for client in self.clients)

685

return rows

686

687

def row_formatting_string(self):

688

"Format string used to format table rows"

689

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

690

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

691

*(len(self.string_from_client(client,

692

key))

693

for client in self.clients)),

694

key=key)

695

for key in self.keywords)

696

697

def string_from_client(self, client, key):

698

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

699

700

@classmethod

701

def valuetostring(cls, value, keyword):

702

if isinstance(value, dbus.Boolean):

703

return "Yes" if value else "No"

704

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

705

"ApprovalDuration", "ExtendedTimeout"):

706

return cls.milliseconds_to_string(value)

707

return str(value)

708

709

def header_line(self, format_string):

710

return format_string.format(**self.tableheaders)

711

712

def client_line(self, client, format_string):

713

return format_string.format(

714

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

715

for key in self.keywords})

716

717

@staticmethod

718

def milliseconds_to_string(ms):

719

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

720

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

721

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

722

if td.days else "",

723

hours=td.seconds // 3600,

724

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

725

seconds=td.seconds % 60))

726

727

728

class PropertySetter(Base):

729

"Abstract class for Actions for setting one client property"

730

731

def run_on_one_client(self, client, properties):

732

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

733

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

734

client.__dbus_object_path__,

735

dbus.PROPERTIES_IFACE, client_dbus_interface,

736

self.propname, self.value_to_set

737

if not isinstance(self.value_to_set,

738

dbus.Boolean)

739

else bool(self.value_to_set))

740

client.Set(client_dbus_interface, self.propname,

741

self.value_to_set,

742

dbus_interface=dbus.PROPERTIES_IFACE)

743

744

@property

745

def propname(self):

746

raise NotImplementedError()

747

748

749

class Enable(PropertySetter):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(True)

752

753

754

class Disable(PropertySetter):

755

propname = "Enabled"

756

value_to_set = dbus.Boolean(False)

757

758

759

class BumpTimeout(PropertySetter):

760

propname = "LastCheckedOK"

761

value_to_set = ""

762

763

764

class StartChecker(PropertySetter):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(True)

767

768

769

class StopChecker(PropertySetter):

770

propname = "CheckerRunning"

771

value_to_set = dbus.Boolean(False)

772

773

774

class ApproveByDefault(PropertySetter):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(True)

777

778

779

class DenyByDefault(PropertySetter):

780

propname = "ApprovedByDefault"

781

value_to_set = dbus.Boolean(False)

782

783

784

class PropertySetterValue(PropertySetter):

785

"""Abstract class for PropertySetter recieving a value as

786

constructor argument instead of a class attribute."""

787

def __init__(self, value):

788

self.value_to_set = value

789

790

791

class SetChecker(PropertySetterValue):

792

propname = "Checker"

793

794

795

class SetHost(PropertySetterValue):

796

propname = "Host"

797

798

799

class SetSecret(PropertySetterValue):

800

propname = "Secret"

801

802

@property

803

def value_to_set(self):

804

return self._vts

805

806

@value_to_set.setter

807

def value_to_set(self, value):

808

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

809

self._vts = value.read()

810

value.close()

811

812

813

class PropertySetterValueMilliseconds(PropertySetterValue):

814

"""Abstract class for PropertySetterValue taking a value

815

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

816

milliseconds."""

817

818

@property

819

def value_to_set(self):

820

return self._vts

821

822

@value_to_set.setter

823

def value_to_set(self, value):

824

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

825

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

826

827

828

class SetTimeout(PropertySetterValueMilliseconds):

829

propname = "Timeout"

830

831

832

class SetExtendedTimeout(PropertySetterValueMilliseconds):

833

propname = "ExtendedTimeout"

834

835

836

class SetInterval(PropertySetterValueMilliseconds):

837

propname = "Interval"

838

839

840

class SetApprovalDelay(PropertySetterValueMilliseconds):

841

propname = "ApprovalDelay"

842

843

844

class SetApprovalDuration(PropertySetterValueMilliseconds):

845

propname = "ApprovalDuration"

714

def milliseconds_to_string(ms):

715

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

716

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

717

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

718

if td.days else "",

719

hours=td.seconds // 3600,

720

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

721

seconds=td.seconds % 60))

722

723

724

class PropertyCmd(Command):

725

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

726

727

def run_on_one_client(self, client, properties):

728

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

729

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

730

client.__dbus_object_path__,

731

dbus.PROPERTIES_IFACE, client_dbus_interface,

732

self.propname, self.value_to_set

733

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

734

else bool(self.value_to_set))

735

client.Set(client_dbus_interface, self.propname,

736

self.value_to_set,

737

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

@property

740

def propname(self):

741

raise NotImplementedError()

742

743

744

class EnableCmd(PropertyCmd):

745

propname = "Enabled"

746

value_to_set = dbus.Boolean(True)

747

748

749

class DisableCmd(PropertyCmd):

750

propname = "Enabled"

751

value_to_set = dbus.Boolean(False)

752

753

754

class BumpTimeoutCmd(PropertyCmd):

755

propname = "LastCheckedOK"

756

value_to_set = ""

757

758

759

class StartCheckerCmd(PropertyCmd):

760

propname = "CheckerRunning"

761

value_to_set = dbus.Boolean(True)

762

763

764

class StopCheckerCmd(PropertyCmd):

765

propname = "CheckerRunning"

766

value_to_set = dbus.Boolean(False)

767

768

769

class ApproveByDefaultCmd(PropertyCmd):

770

propname = "ApprovedByDefault"

771

value_to_set = dbus.Boolean(True)

772

773

774

class DenyByDefaultCmd(PropertyCmd):

775

propname = "ApprovedByDefault"

776

value_to_set = dbus.Boolean(False)

777

778

779

class PropertyValueCmd(PropertyCmd):

780

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

781

def __init__(self, value):

782

self.value_to_set = value

783

784

785

class SetCheckerCmd(PropertyValueCmd):

786

propname = "Checker"

787

788

789

class SetHostCmd(PropertyValueCmd):

790

propname = "Host"

791

792

793

class SetSecretCmd(PropertyValueCmd):

794

propname = "Secret"

795

796

@property

797

def value_to_set(self):

798

return self._vts

799

800

@value_to_set.setter

801

def value_to_set(self, value):

802

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

803

self._vts = value.read()

804

value.close()

805

806

807

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

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

809

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

810

811

@property

812

def value_to_set(self):

813

return self._vts

814

815

@value_to_set.setter

816

def value_to_set(self, value):

817

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

818

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

819

820

821

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

822

propname = "Timeout"

823

824

825

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

826

propname = "ExtendedTimeout"

827

828

829

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

830

propname = "Interval"

831

832

833

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

834

propname = "ApprovalDelay"

835

836

837

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

838

propname = "ApprovalDuration"

846

839

847

840

848

841

879

872

("records",

880

873

"output"))

881

874

882

883

875

class Test_string_to_delta(TestCaseWithAssertLogs):

884

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

885

# rfc3339_duration_to_delta() already has more comprehensive

886

# tests, which is run by doctest.

887

888

def test_rfc3339_zero_seconds(self):

889

self.assertEqual(datetime.timedelta(),

890

string_to_delta("PT0S"))

891

892

def test_rfc3339_zero_days(self):

893

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

894

895

def test_rfc3339_one_second(self):

896

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

897

string_to_delta("PT1S"))

898

899

def test_rfc3339_two_hours(self):

900

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

901

string_to_delta("PT2H"))

876

def test_handles_basic_rfc3339(self):

877

self.assertEqual(string_to_delta("PT0S"),

878

datetime.timedelta())

879

self.assertEqual(string_to_delta("P0D"),

880

datetime.timedelta())

881

self.assertEqual(string_to_delta("PT1S"),

882

datetime.timedelta(0, 1))

883

self.assertEqual(string_to_delta("PT2H"),

884

datetime.timedelta(0, 7200))

902

885

903

886

def test_falls_back_to_pre_1_6_1_with_warning(self):

904

887

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

905

888

value = string_to_delta("2h")

906

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

889

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

907

890

908

891

909

892

class Test_check_option_syntax(unittest.TestCase):

947

930

@contextlib.contextmanager

948

931

def assertParseError(self):

949

932

with self.assertRaises(SystemExit) as e:

950

with self.redirect_stderr_to_devnull():

933

with self.temporarily_suppress_stderr():

951

934

yield

952

935

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

953

936

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

954

937

# /argparse.html#exiting-methods

955

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

938

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

956

939

957

940

@staticmethod

958

941

@contextlib.contextmanager

959

def redirect_stderr_to_devnull():

960

old_stderr = sys.stderr

961

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

962

sys.stderr = null

963

try:

964

yield

965

finally:

966

sys.stderr = old_stderr

942

def temporarily_suppress_stderr():

943

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

944

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

945

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

946

os.close(null)

947

try:

948

yield

949

finally:

950

# restore stderr

951

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

952

os.close(stderrcopy)

967

953

968

954

def check_option_syntax(self, options):

969

955

check_option_syntax(self.parser, options)

970

956

971

def test_actions_all_conflicts_with_verbose(self):

972

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

973

options = self.parser.parse_args()

974

setattr(options, action, value)

975

options.all = True

976

options.verbose = True

977

with self.assertParseError():

978

self.check_option_syntax(options)

979

980

def test_actions_with_client_conflicts_with_verbose(self):

981

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

982

options = self.parser.parse_args()

983

setattr(options, action, value)

984

options.verbose = True

985

options.client = ["client"]

957

def test_actions_conflicts_with_verbose(self):

958

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

959

options = self.parser.parse_args()

960

setattr(options, action, value)

961

options.verbose = True

986

962

with self.assertParseError():

987

963

self.check_option_syntax(options)

988

964

1014

990

options.all = True

1015

991

self.check_option_syntax(options)

1016

992

1017

def test_any_action_is_ok_with_one_client(self):

1018

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

1019

options = self.parser.parse_args()

1020

setattr(options, action, value)

1021

options.client = ["client"]

1022

self.check_option_syntax(options)

1023

1024

def test_one_client_with_all_actions_except_is_enabled(self):

1025

options = self.parser.parse_args()

1026

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

1027

if action == "is_enabled":

1028

continue

1029

setattr(options, action, value)

1030

options.client = ["client"]

1031

self.check_option_syntax(options)

1032

1033

def test_two_clients_with_all_actions_except_is_enabled(self):

1034

options = self.parser.parse_args()

1035

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

1036

if action == "is_enabled":

1037

continue

1038

setattr(options, action, value)

1039

options.client = ["client1", "client2"]

1040

self.check_option_syntax(options)

1041

1042

def test_two_clients_are_ok_with_actions_except_is_enabled(self):

1043

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

1044

if action == "is_enabled":

1045

continue

1046

options = self.parser.parse_args()

1047

setattr(options, action, value)

1048

options.client = ["client1", "client2"]

1049

self.check_option_syntax(options)

1050

1051

993

def test_is_enabled_fails_without_client(self):

1052

994

options = self.parser.parse_args()

1053

995

options.is_enabled = True

1054

996

with self.assertParseError():

1055

997

self.check_option_syntax(options)

1056

998

999

def test_is_enabled_works_with_one_client(self):

1000

options = self.parser.parse_args()

1001

options.is_enabled = True

1002

options.client = ["foo"]

1003

self.check_option_syntax(options)

1004

1057

1005

def test_is_enabled_fails_with_two_clients(self):

1058

1006

options = self.parser.parse_args()

1059

1007

options.is_enabled = True

1060

options.client = ["client1", "client2"]

1008

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

1061

1009

with self.assertParseError():

1062

1010

self.check_option_syntax(options)

1063

1011

1080

1028

def get_object(mockbus_self, busname, dbus_path):

1081

1029

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

1082

1030

# this MockBus instance is in "mockbus_self".

1083

self.assertEqual(dbus_busname, busname)

1084

self.assertEqual(server_dbus_path, dbus_path)

1031

self.assertEqual(busname, dbus_busname)

1032

self.assertEqual(dbus_path, server_dbus_path)

1085

1033

mockbus_self.called = True

1086

1034

return mockbus_self

1087

1035

1090

1038

self.assertTrue(mockbus.called)

1091

1039

1092

1040

def test_logs_and_exits_on_dbus_error(self):

1093

class FailingBusStub(object):

1041

class MockBusFailing(object):

1094

1042

def get_object(self, busname, dbus_path):

1095

1043

raise dbus.exceptions.DBusException("Test")

1096

1044

1097

1045

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

1098

1046

with self.assertRaises(SystemExit) as e:

1099

bus = get_mandos_dbus_object(bus=FailingBusStub())

1047

bus = get_mandos_dbus_object(bus=MockBusFailing())

1100

1048

1101

1049

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

1102

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

1050

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

1103

1051

else:

1104

1052

self.assertIsNotNone(e.exception.code)

1105

1053

1106

1054

1107

1055

class Test_get_managed_objects(TestCaseWithAssertLogs):

1108

1056

def test_calls_and_returns_GetManagedObjects(self):

1109

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

1110

class ObjectManagerStub(object):

1057

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

1058

class MockObjectManager(object):

1111

1059

def GetManagedObjects(self):

1112

1060

return managed_objects

1113

retval = get_managed_objects(ObjectManagerStub())

1061

retval = get_managed_objects(MockObjectManager())

1114

1062

self.assertDictEqual(managed_objects, retval)

1115

1063

1116

1064

def test_logs_and_exits_on_dbus_error(self):

1117

1065

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

1118

1066

1119

class ObjectManagerFailingStub(object):

1067

class MockObjectManagerFailing(object):

1120

1068

def GetManagedObjects(self):

1121

1069

dbus_logger.error("Test")

1122

1070

raise dbus.exceptions.DBusException("Test")

1133

1081

try:

1134

1082

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

1135

1083

with self.assertRaises(SystemExit) as e:

1136

get_managed_objects(ObjectManagerFailingStub())

1084

get_managed_objects(MockObjectManagerFailing())

1137

1085

finally:

1138

1086

dbus_logger.removeFilter(counting_handler)

1139

1140

# Make sure the dbus logger was suppressed

1141

self.assertEqual(0, counting_handler.count)

1087

self.assertEqual(counting_handler.count, 0)

1142

1088

1143

1089

# Test that the dbus_logger still works

1144

1090

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

1145

1091

dbus_logger.error("Test")

1146

1092

1147

1093

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

1148

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

1094

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

1149

1095

else:

1150

1096

self.assertIsNotNone(e.exception.code)

1151

1097

1156

1102

add_command_line_options(self.parser)

1157

1103

1158

1104

def test_is_enabled(self):

1159

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

1160

command.IsEnabled)

1105

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

1106

IsEnabledCmd)

1161

1107

1162

1108

def assert_command_from_args(self, args, command_cls,

1163

1109

**cmd_attrs):

1166

1112

options = self.parser.parse_args(args)

1167

1113

check_option_syntax(self.parser, options)

1168

1114

commands = commands_from_options(options)

1169

self.assertEqual(1, len(commands))

1115

self.assertEqual(len(commands), 1)

1170

1116

command = commands[0]

1171

1117

self.assertIsInstance(command, command_cls)

1172

1118

for key, value in cmd_attrs.items():

1173

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

1119

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

1174

1120

1175

1121

def test_is_enabled_short(self):

1176

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

1177

command.IsEnabled)

1122

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

1178

1123

1179

1124

def test_approve(self):

1180

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

1181

command.Approve)

1125

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

1126

ApproveCmd)

1182

1127

1183

1128

def test_approve_short(self):

1184

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

1185

command.Approve)

1129

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

1186

1130

1187

1131

def test_deny(self):

1188

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

1189

command.Deny)

1132

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

1190

1133

1191

1134

def test_deny_short(self):

1192

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

1135

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

1193

1136

1194

1137

def test_remove(self):

1195

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

1196

command.Remove)

1138

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

1139

RemoveCmd)

1197

1140

1198

1141

def test_deny_before_remove(self):

1199

1142

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

1200

"client"])

1143

"foo"])

1201

1144

check_option_syntax(self.parser, options)

1202

1145

commands = commands_from_options(options)

1203

self.assertEqual(2, len(commands))

1204

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

1205

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

1146

self.assertEqual(len(commands), 2)

1147

self.assertIsInstance(commands[0], DenyCmd)

1148

self.assertIsInstance(commands[1], RemoveCmd)

1206

1149

1207

1150

def test_deny_before_remove_reversed(self):

1208

1151

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

1209

1152

"--all"])

1210

1153

check_option_syntax(self.parser, options)

1211

1154

commands = commands_from_options(options)

1212

self.assertEqual(2, len(commands))

1213

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

1214

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

1155

self.assertEqual(len(commands), 2)

1156

self.assertIsInstance(commands[0], DenyCmd)

1157

self.assertIsInstance(commands[1], RemoveCmd)

1215

1158

1216

1159

def test_remove_short(self):

1217

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

1218

command.Remove)

1160

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

1219

1161

1220

1162

def test_dump_json(self):

1221

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

1222

command.DumpJSON)

1163

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

1223

1164

1224

1165

def test_enable(self):

1225

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

1226

command.Enable)

1166

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

1227

1167

1228

1168

def test_enable_short(self):

1229

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

1230

command.Enable)

1169

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

1231

1170

1232

1171

def test_disable(self):

1233

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

1234

command.Disable)

1172

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

1173

DisableCmd)

1235

1174

1236

1175

def test_disable_short(self):

1237

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

1238

command.Disable)

1176

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

1239

1177

1240

1178

def test_bump_timeout(self):

1241

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

1242

command.BumpTimeout)

1179

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

1180

BumpTimeoutCmd)

1243

1181

1244

1182

def test_bump_timeout_short(self):

1245

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

1246

command.BumpTimeout)

1183

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

1247

1184

1248

1185

def test_start_checker(self):

1249

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

1250

command.StartChecker)

1186

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

1187

StartCheckerCmd)

1251

1188

1252

1189

def test_stop_checker(self):

1253

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

1254

command.StopChecker)

1190

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

1191

StopCheckerCmd)

1255

1192

1256

1193

def test_approve_by_default(self):

1257

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

1258

"client"],

1259

command.ApproveByDefault)

1194

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

1195

ApproveByDefaultCmd)

1260

1196

1261

1197

def test_deny_by_default(self):

1262

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

1263

command.DenyByDefault)

1198

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

1199

DenyByDefaultCmd)

1264

1200

1265

1201

def test_checker(self):

1266

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

1267

command.SetChecker,

1268

value_to_set=":")

1202

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

1203

SetCheckerCmd, value_to_set=":")

1269

1204

1270

1205

def test_checker_empty(self):

1271

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

1272

command.SetChecker,

1273

value_to_set="")

1206

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

1207

SetCheckerCmd, value_to_set="")

1274

1208

1275

1209

def test_checker_short(self):

1276

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

1277

command.SetChecker,

1278

value_to_set=":")

1210

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

1211

SetCheckerCmd, value_to_set=":")

1279

1212

1280

1213

def test_host(self):

1281

self.assert_command_from_args(

1282

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

1283

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

1214

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

1215

"foo"], SetHostCmd,

1216

value_to_set="foo.example.org")

1284

1217

1285

1218

def test_host_short(self):

1286

self.assert_command_from_args(

1287

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

1288

value_to_set="client.example.org")

1219

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

1220

"foo"], SetHostCmd,

1221

value_to_set="foo.example.org")

1289

1222

1290

1223

def test_secret_devnull(self):

1291

1224

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

1292

"client"], command.SetSecret,

1225

"foo"], SetSecretCmd,

1293

1226

value_to_set=b"")

1294

1227

1295

1228

def test_secret_tempfile(self):

1298

1231

f.write(value)

1299

1232

f.seek(0)

1300

1233

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

1301

"client"],

1302

command.SetSecret,

1234

"foo"], SetSecretCmd,

1303

1235

value_to_set=value)

1304

1236

1305

1237

def test_secret_devnull_short(self):

1306

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

1307

"client"], command.SetSecret,

1308

value_to_set=b"")

1238

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

1239

SetSecretCmd, value_to_set=b"")

1309

1240

1310

1241

def test_secret_tempfile_short(self):

1311

1242

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

1312

1243

value = b"secret\0xyzzy\nbar"

1313

1244

f.write(value)

1314

1245

f.seek(0)

1315

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

1316

command.SetSecret,

1246

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

1247

SetSecretCmd,

1317

1248

value_to_set=value)

1318

1249

1319

1250

def test_timeout(self):

1320

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

1321

command.SetTimeout,

1251

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

1252

SetTimeoutCmd,

1322

1253

value_to_set=300000)

1323

1254

1324

1255

def test_timeout_short(self):

1325

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

1326

command.SetTimeout,

1256

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

1257

SetTimeoutCmd,

1327

1258

value_to_set=300000)

1328

1259

1329

1260

def test_extended_timeout(self):

1330

1261

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

1331

"client"],

1332

command.SetExtendedTimeout,

1262

"foo"],

1263

SetExtendedTimeoutCmd,

1333

1264

value_to_set=900000)

1334

1265

1335

1266

def test_interval(self):

1336

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

1337

"client"], command.SetInterval,

1267

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

1268

SetIntervalCmd,

1338

1269

value_to_set=120000)

1339

1270

1340

1271

def test_interval_short(self):

1341

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

1342

command.SetInterval,

1272

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

1273

SetIntervalCmd,

1343

1274

value_to_set=120000)

1344

1275

1345

1276

def test_approval_delay(self):

1346

1277

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

1347

"client"],

1348

command.SetApprovalDelay,

1278

"foo"], SetApprovalDelayCmd,

1349

1279

value_to_set=30000)

1350

1280

1351

1281

def test_approval_duration(self):

1352

1282

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

1353

"client"],

1354

command.SetApprovalDuration,

1283

"foo"], SetApprovalDurationCmd,

1355

1284

value_to_set=1000)

1356

1285

1357

1286

def test_print_table(self):

1358

self.assert_command_from_args([], command.PrintTable,

1287

self.assert_command_from_args([], PrintTableCmd,

1359

1288

verbose=False)

1360

1289

1361

1290

def test_print_table_verbose(self):

1362

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

1363

command.PrintTable,

1291

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

1364

1292

verbose=True)

1365

1293

1366

1294

def test_print_table_verbose_short(self):

1367

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

1295

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

1368

1296

verbose=True)

1369

1297

1370

1298

1371

class TestCommand(unittest.TestCase):

1299

class TestCmd(unittest.TestCase):

1372

1300

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

1373

1301

1374

1302

def setUp(self):

1380

1308

self.attributes["Name"] = name

1381

1309

self.calls = []

1382

1310

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

1383

testcase.assertEqual(client_dbus_interface, interface)

1384

testcase.assertEqual(dbus.PROPERTIES_IFACE,

1385

dbus_interface)

1311

testcase.assertEqual(interface, client_dbus_interface)

1312

testcase.assertEqual(dbus_interface,

1313

dbus.PROPERTIES_IFACE)

1386

1314

self.attributes[propname] = value

1315

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

1316

testcase.assertEqual(interface, client_dbus_interface)

1317

testcase.assertEqual(dbus_interface,

1318

dbus.PROPERTIES_IFACE)

1319

return self.attributes[propname]

1387

1320

def Approve(self, approve, dbus_interface):

1388

testcase.assertEqual(client_dbus_interface,

1389

dbus_interface)

1321

testcase.assertEqual(dbus_interface,

1322

client_dbus_interface)

1390

1323

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

1391

1324

dbus_interface)))

1392

1325

self.client = MockClient(

1439

1372

LastCheckerStatus=-2)

1440

1373

self.clients = collections.OrderedDict(

1441

1374

[

1442

(self.client.__dbus_object_path__,

1443

self.client.attributes),

1444

(self.other_client.__dbus_object_path__,

1445

self.other_client.attributes),

1375

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

1376

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

1446

1377

])

1447

self.one_client = {self.client.__dbus_object_path__:

1448

self.client.attributes}

1378

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

1449

1379

1450

1380

@property

1451

1381

def bus(self):

1452

class MockBus(object):

1382

class Bus(object):

1453

1383

@staticmethod

1454

1384

def get_object(client_bus_name, path):

1455

self.assertEqual(dbus_busname, client_bus_name)

1456

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

1457

# MockBus instance, since this is a static method!

1458

if path == self.client.__dbus_object_path__:

1459

return self.client

1460

elif path == self.other_client.__dbus_object_path__:

1461

return self.other_client

1462

return MockBus()

1463

1464

1465

class TestBaseCommands(TestCommand):

1466

1467

def test_IsEnabled_exits_successfully(self):

1385

self.assertEqual(client_bus_name, dbus_busname)

1386

return {

1387

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

1388

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

1389

"/clients/foo": self.client,

1390

"/clients/barbar": self.other_client,

1391

}[path]

1392

return Bus()

1393

1394

1395

class TestIsEnabledCmd(TestCmd):

1396

def test_is_enabled(self):

1397

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

1398

properties)

1399

for client, properties

1400

in self.clients.items()))

1401

1402

def test_is_enabled_run_exits_successfully(self):

1468

1403

with self.assertRaises(SystemExit) as e:

1469

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

1404

IsEnabledCmd().run(self.one_client)

1470

1405

if e.exception.code is not None:

1471

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

1406

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

1472

1407

else:

1473

1408

self.assertIsNone(e.exception.code)

1474

1409

1475

def test_IsEnabled_exits_with_failure(self):

1410

def test_is_enabled_run_exits_with_failure(self):

1476

1411

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

1477

1412

with self.assertRaises(SystemExit) as e:

1478

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

1413

IsEnabledCmd().run(self.one_client)

1479

1414

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

1480

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

1415

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

1481

1416

else:

1482

1417

self.assertIsNotNone(e.exception.code)

1483

1418

1484

def test_Approve(self):

1485

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

1419

1420

class TestApproveCmd(TestCmd):

1421

def test_approve(self):

1422

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

1486

1423

for clientpath in self.clients:

1487

1424

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

1488

1425

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

1489

1426

client.calls)

1490

1427

1491

def test_Deny(self):

1492

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

1428

1429

class TestDenyCmd(TestCmd):

1430

def test_deny(self):

1431

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

1493

1432

for clientpath in self.clients:

1494

1433

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

1495

1434

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

1496

1435

client.calls)

1497

1436

1498

def test_Remove(self):

1499

class MandosSpy(object):

1437

1438

class TestRemoveCmd(TestCmd):

1439

def test_remove(self):

1440

class MockMandos(object):

1500

1441

def __init__(self):

1501

1442

self.calls = []

1502

1443

def RemoveClient(self, dbus_path):

1503

1444

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

1504

mandos = MandosSpy()

1505

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

1445

mandos = MockMandos()

1446

super(TestRemoveCmd, self).setUp()

1447

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

1448

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

1506

1449

for clientpath in self.clients:

1507

1450

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

1508

1451

mandos.calls)

1509

1452

1510

expected_json = {

1511

"foo": {

1512

"Name": "foo",

1513

"KeyID": ("92ed150794387c03ce684574b1139a65"

1514

"94a34f895daaaf09fd8ea90a27cddb12"),

1515

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

1516

"Enabled": True,

1517

"Timeout": 300000,

1518

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

1519

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

1520

"Interval": 120000,

1521

"Fingerprint": ("778827225BA7DE539C5A"

1522

"7CFA59CFF7CDBD9A5920"),

1523

"CheckerRunning": False,

1524

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

1525

"ApprovalPending": False,

1526

"ApprovedByDefault": True,

1527

"LastApprovalRequest": "",

1528

"ApprovalDelay": 0,

1529

"ApprovalDuration": 1000,

1530

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

1531

"ExtendedTimeout": 900000,

1532

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

1533

"LastCheckerStatus": 0,

1534

1535

"barbar": {

1536

"Name": "barbar",

1537

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1538

"6ab612cff5ad227247e46c2b020f441c"),

1539

"Host": "192.0.2.3",

1540

"Enabled": True,

1541

"Timeout": 300000,

1542

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

1543

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

1544

"Interval": 120000,

1545

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1546

"F547B3A107558FCA3A27"),

1547

"CheckerRunning": True,

1548

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

1549

"ApprovalPending": False,

1550

"ApprovedByDefault": False,

1551

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

1552

"ApprovalDelay": 30000,

1553

"ApprovalDuration": 93785000,

1554

"Checker": ":",

1555

"ExtendedTimeout": 900000,

1556

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

1557

"LastCheckerStatus": -2,

1558

1559

}

1560

1561

def test_DumpJSON_normal(self):

1562

with self.capture_stdout_to_buffer() as buffer:

1563

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

1564

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

1565

self.assertDictEqual(self.expected_json, json_data)

1566

1567

@staticmethod

1568

@contextlib.contextmanager

1569

def capture_stdout_to_buffer():

1570

capture_buffer = io.StringIO()

1571

old_stdout = sys.stdout

1572

sys.stdout = capture_buffer

1573

try:

1574

yield capture_buffer

1575

finally:

1576

sys.stdout = old_stdout

1577

1578

def test_DumpJSON_one_client(self):

1579

with self.capture_stdout_to_buffer() as buffer:

1580

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

1581

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

1453

1454

class TestDumpJSONCmd(TestCmd):

1455

def setUp(self):

1456

self.expected_json = {

1457

"foo": {

1458

"Name": "foo",

1459

"KeyID": ("92ed150794387c03ce684574b1139a65"

1460

"94a34f895daaaf09fd8ea90a27cddb12"),

1461

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

1462

"Enabled": True,

1463

"Timeout": 300000,

1464

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

1465

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

1466

"Interval": 120000,

1467

"Fingerprint": ("778827225BA7DE539C5A"

1468

"7CFA59CFF7CDBD9A5920"),

1469

"CheckerRunning": False,

1470

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

1471

"ApprovalPending": False,

1472

"ApprovedByDefault": True,

1473

"LastApprovalRequest": "",

1474

"ApprovalDelay": 0,

1475

"ApprovalDuration": 1000,

1476

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

1477

"ExtendedTimeout": 900000,

1478

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

1479

"LastCheckerStatus": 0,

1480

1481

"barbar": {

1482

"Name": "barbar",

1483

"KeyID": ("0558568eedd67d622f5c83b35a115f79"

1484

"6ab612cff5ad227247e46c2b020f441c"),

1485

"Host": "192.0.2.3",

1486

"Enabled": True,

1487

"Timeout": 300000,

1488

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

1489

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

1490

"Interval": 120000,

1491

"Fingerprint": ("3E393AEAEFB84C7E89E2"

1492

"F547B3A107558FCA3A27"),

1493

"CheckerRunning": True,

1494

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

1495

"ApprovalPending": False,

1496

"ApprovedByDefault": False,

1497

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

1498

"ApprovalDelay": 30000,

1499

"ApprovalDuration": 93785000,

1500

"Checker": ":",

1501

"ExtendedTimeout": 900000,

1502

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

1503

"LastCheckerStatus": -2,

1504

1505

}

1506

return super(TestDumpJSONCmd, self).setUp()

1507

1508

def test_normal(self):

1509

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

1510

json_data = json.loads(output)

1511

self.assertDictEqual(json_data, self.expected_json)

1512

1513

def test_one_client(self):

1514

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

1515

json_data = json.loads(output)

1582

1516

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

1583

self.assertDictEqual(expected_json, json_data)

1584

1585

def test_PrintTable_normal(self):

1586

with self.capture_stdout_to_buffer() as buffer:

1587

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

1517

self.assertDictEqual(json_data, expected_json)

1518

1519

1520

class TestPrintTableCmd(TestCmd):

1521

def test_normal(self):

1522

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

1588

1523

expected_output = "\n".join((

1589

1524

"Name Enabled Timeout Last Successful Check",

1590

1525

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

1591

1526

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

1592

)) + "\n"

1593

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

1527

))

1528

self.assertEqual(output, expected_output)

1594

1529

1595

def test_PrintTable_verbose(self):

1596

with self.capture_stdout_to_buffer() as buffer:

1597

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

1530

def test_verbose(self):

1531

output = PrintTableCmd(verbose=True).output(

1532

self.clients.values())

1598

1533

columns = (

1599

1534

(

1600

1535

"Name ",

1682

1617

)

1683

1618

)

1684

1619

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

1685

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

1686

for rows in columns)

1687

for line in range(num_lines))

1688

+ "\n")

1689

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

1620

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

1621

for rows in columns)

1622

for line in range(num_lines))

1623

self.assertEqual(output, expected_output)

1690

1624

1691

def test_PrintTable_one_client(self):

1692

with self.capture_stdout_to_buffer() as buffer:

1693

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

1625

def test_one_client(self):

1626

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

1694

1627

expected_output = "\n".join((

1695

1628

"Name Enabled Timeout Last Successful Check",

1696

1629

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

1697

)) + "\n"

1698

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

1699

1700

1701

class TestPropertySetterCmd(TestCommand):

1702

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

1630

))

1631

self.assertEqual(output, expected_output)

1632

1633

1634

class TestPropertyCmd(TestCmd):

1635

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

1703

1636

def runTest(self):

1704

1637

if not hasattr(self, "command"):

1705

1638

return

1710

1643

for clientpath in self.clients:

1711

1644

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

1712

1645

old_value = client.attributes[self.propname]

1646

self.assertNotIsInstance(old_value, self.Unique)

1713

1647

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

1714

1648

self.run_command(value_to_set, self.clients)

1715

1649

for clientpath in self.clients:

1716

1650

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

1717

1651

value = client.attributes[self.propname]

1718

1652

self.assertNotIsInstance(value, self.Unique)

1719

self.assertEqual(value_to_get, value)

1653

self.assertEqual(value, value_to_get)

1720

1654

1721

1655

class Unique(object):

1722

1656

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

1726

1660

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

1727

1661

1728

1662

1729

class TestEnableCmd(TestPropertySetterCmd):

1730

command = command.Enable

1663

class TestEnableCmd(TestPropertyCmd):

1664

command = EnableCmd

1731

1665

propname = "Enabled"

1732

1666

values_to_set = [dbus.Boolean(True)]

1733

1667

1734

1668

1735

class TestDisableCmd(TestPropertySetterCmd):

1736

command = command.Disable

1669

class TestDisableCmd(TestPropertyCmd):

1670

command = DisableCmd

1737

1671

propname = "Enabled"

1738

1672

values_to_set = [dbus.Boolean(False)]

1739

1673

1740

1674

1741

class TestBumpTimeoutCmd(TestPropertySetterCmd):

1742

command = command.BumpTimeout

1675

class TestBumpTimeoutCmd(TestPropertyCmd):

1676

command = BumpTimeoutCmd

1743

1677

propname = "LastCheckedOK"

1744

1678

values_to_set = [""]

1745

1679

1746

1680

1747

class TestStartCheckerCmd(TestPropertySetterCmd):

1748

command = command.StartChecker

1749

propname = "CheckerRunning"

1750

values_to_set = [dbus.Boolean(True)]

1751

1752

1753

class TestStopCheckerCmd(TestPropertySetterCmd):

1754

command = command.StopChecker

1755

propname = "CheckerRunning"

1756

values_to_set = [dbus.Boolean(False)]

1757

1758

1759

class TestApproveByDefaultCmd(TestPropertySetterCmd):

1760

command = command.ApproveByDefault

1761

propname = "ApprovedByDefault"

1762

values_to_set = [dbus.Boolean(True)]

1763

1764

1765

class TestDenyByDefaultCmd(TestPropertySetterCmd):

1766

command = command.DenyByDefault

1767

propname = "ApprovedByDefault"

1768

values_to_set = [dbus.Boolean(False)]

1769

1770

1771

class TestPropertySetterValueCmd(TestPropertySetterCmd):

1772

"""Abstract class for tests of PropertySetterValueCmd classes"""

1681

class TestStartCheckerCmd(TestPropertyCmd):

1682

command = StartCheckerCmd

1683

propname = "CheckerRunning"

1684

values_to_set = [dbus.Boolean(True)]

1685

1686

1687

class TestStopCheckerCmd(TestPropertyCmd):

1688

command = StopCheckerCmd

1689

propname = "CheckerRunning"

1690

values_to_set = [dbus.Boolean(False)]

1691

1692

1693

class TestApproveByDefaultCmd(TestPropertyCmd):

1694

command = ApproveByDefaultCmd

1695

propname = "ApprovedByDefault"

1696

values_to_set = [dbus.Boolean(True)]

1697

1698

1699

class TestDenyByDefaultCmd(TestPropertyCmd):

1700

command = DenyByDefaultCmd

1701

propname = "ApprovedByDefault"

1702

values_to_set = [dbus.Boolean(False)]

1703

1704

1705

class TestPropertyValueCmd(TestPropertyCmd):

1706

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

1773

1707

1774

1708

def runTest(self):

1775

if type(self) is TestPropertySetterValueCmd:

1709

if type(self) is TestPropertyValueCmd:

1776

1710

return

1777

return super(TestPropertySetterValueCmd, self).runTest()

1711

return super(TestPropertyValueCmd, self).runTest()

1778

1712

1779

1713

def run_command(self, value, clients):

1780

1714

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

1781

1715

1782

1716

1783

class TestSetCheckerCmd(TestPropertySetterValueCmd):

1784

command = command.SetChecker

1717

class TestSetCheckerCmd(TestPropertyValueCmd):

1718

command = SetCheckerCmd

1785

1719

propname = "Checker"

1786

1720

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

1787

1721

1788

1722

1789

class TestSetHostCmd(TestPropertySetterValueCmd):

1790

command = command.SetHost

1723

class TestSetHostCmd(TestPropertyValueCmd):

1724

command = SetHostCmd

1791

1725

propname = "Host"

1792

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

1793

1794

1795

class TestSetSecretCmd(TestPropertySetterValueCmd):

1796

command = command.SetSecret

1726

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

1727

1728

1729

class TestSetSecretCmd(TestPropertyValueCmd):

1730

command = SetSecretCmd

1797

1731

propname = "Secret"

1798

1732

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

1799

1733

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

1800

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

1801

1802

1803

class TestSetTimeoutCmd(TestPropertySetterValueCmd):

1804

command = command.SetTimeout

1734

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

1735

1736

1737

class TestSetTimeoutCmd(TestPropertyValueCmd):

1738

command = SetTimeoutCmd

1805

1739

propname = "Timeout"

1806

1740

values_to_set = [datetime.timedelta(),

1807

1741

datetime.timedelta(minutes=5),

1808

1742

datetime.timedelta(seconds=1),

1809

1743

datetime.timedelta(weeks=1),

1810

1744

datetime.timedelta(weeks=52)]

1811

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

1812

1813

1814

class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):

1815

command = command.SetExtendedTimeout

1745

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

1746

1747

1748

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1749

command = SetExtendedTimeoutCmd

1816

1750

propname = "ExtendedTimeout"

1817

1751

values_to_set = [datetime.timedelta(),

1818

1752

datetime.timedelta(minutes=5),

1819

1753

datetime.timedelta(seconds=1),

1820

1754

datetime.timedelta(weeks=1),

1821

1755

datetime.timedelta(weeks=52)]

1822

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

1823

1824

1825

class TestSetIntervalCmd(TestPropertySetterValueCmd):

1826

command = command.SetInterval

1756

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

1757

1758

1759

class TestSetIntervalCmd(TestPropertyValueCmd):

1760

command = SetIntervalCmd

1827

1761

propname = "Interval"

1828

1762

values_to_set = [datetime.timedelta(),

1829

1763

datetime.timedelta(minutes=5),

1830

1764

datetime.timedelta(seconds=1),

1831

1765

datetime.timedelta(weeks=1),

1832

1766

datetime.timedelta(weeks=52)]

1833

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

1834

1835

1836

class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):

1837

command = command.SetApprovalDelay

1767

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

1768

1769

1770

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1771

command = SetApprovalDelayCmd

1838

1772

propname = "ApprovalDelay"

1839

1773

values_to_set = [datetime.timedelta(),

1840

1774

datetime.timedelta(minutes=5),

1841

1775

datetime.timedelta(seconds=1),

1842

1776

datetime.timedelta(weeks=1),

1843

1777

datetime.timedelta(weeks=52)]

1844

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

1845

1846

1847

class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):

1848

command = command.SetApprovalDuration

1778

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

1779

1780

1781

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1782

command = SetApprovalDurationCmd

1849

1783

propname = "ApprovalDuration"

1850

1784

values_to_set = [datetime.timedelta(),

1851

1785

datetime.timedelta(minutes=5),

1852

1786

datetime.timedelta(seconds=1),

1853

1787

datetime.timedelta(weeks=1),

1854

1788

datetime.timedelta(weeks=52)]

1855

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

1789

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

1856

1790

1857

1791

1858

1792

Older »