/mandos/trunk : revision 1051

44

import logging

45

import io

46

import tempfile

47

import contextlib

47

48

49

import dbus

49

50

275

276

# Abstract classes first

276

277

class Command(object):

277

278

"""Abstract class for commands"""

278

def run(self, mandos, clients):

279

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

279

280

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

280

281

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

281

282

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

282

283

self.mandos = mandos

283

for client, properties in clients.items():

284

for clientpath, properties in clients.items():

285

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

286

busname, str(clientpath))

287

client = bus.get_object(busname, clientpath)

284

288

self.run_on_one_client(client, properties)

285

289

286

290

class PrintCmd(Command):

292

296

"LastApprovalRequest", "ApprovalDelay",

293

297

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

294

298

"Expires", "LastCheckerStatus")

295

def run(self, mandos, clients):

296

print(self.output(clients))

299

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

300

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

301

def output(self, clients):

302

raise NotImplementedError()

297

303

298

304

class PropertyCmd(Command):

299

305

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

300

306

def run_on_one_client(self, client, properties):

301

307

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

302

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

308

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

309

client.__dbus_object_path__,

310

dbus.PROPERTIES_IFACE, client_interface,

311

self.propname, self.value_to_set

312

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

313

else bool(self.value_to_set))

314

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

303

315

dbus_interface=dbus.PROPERTIES_IFACE)

316

@property

317

def propname(self):

318

raise NotImplementedError()

304

319

305

class ValueArgumentMixIn(object):

306

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

320

class PropertyValueCmd(PropertyCmd):

321

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

307

322

def __init__(self, value):

308

323

self.value_to_set = value

309

324

310

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

311

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

312

milliseconds."""

325

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

326

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

327

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

313

328

@property

314

329

def value_to_set(self):

315

330

return self._vts

316

331

@value_to_set.setter

317

332

def value_to_set(self, value):

318

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

319

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

333

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

334

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

320

335

321

336

# Actual (non-abstract) command classes

322

337

329

344

keywords = default_keywords

330

345

if self.verbose:

331

346

keywords = self.all_keywords

332

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

347

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

333

348

334

349

class TableOfClients(object):

335

350

tableheaders = {

421

436

return value

422

437

423

438

class IsEnabledCmd(Command):

424

def run_on_one_client(self, client, properties):

439

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

440

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

425

441

if self.is_enabled(client, properties):

426

442

sys.exit(0)

427

443

sys.exit(1)

428

444

def is_enabled(self, client, properties):

429

return bool(properties["Enabled"])

445

return properties["Enabled"]

430

446

431

447

class RemoveCmd(Command):

432

448

def run_on_one_client(self, client, properties):

449

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

450

server_path, server_interface,

451

str(client.__dbus_object_path__))

433

452

self.mandos.RemoveClient(client.__dbus_object_path__)

434

453

435

454

class ApproveCmd(Command):

436

455

def run_on_one_client(self, client, properties):

456

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

457

client.__dbus_object_path__, client_interface)

437

458

client.Approve(dbus.Boolean(True),

438

459

dbus_interface=client_interface)

439

460

440

461

class DenyCmd(Command):

441

462

def run_on_one_client(self, client, properties):

463

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

464

client.__dbus_object_path__, client_interface)

442

465

client.Approve(dbus.Boolean(False),

443

466

dbus_interface=client_interface)

444

467

445

468

class EnableCmd(PropertyCmd):

446

property = "Enabled"

469

propname = "Enabled"

447

470

value_to_set = dbus.Boolean(True)

448

471

449

472

class DisableCmd(PropertyCmd):

450

property = "Enabled"

473

propname = "Enabled"

451

474

value_to_set = dbus.Boolean(False)

452

475

453

476

class BumpTimeoutCmd(PropertyCmd):

454

property = "LastCheckedOK"

477

propname = "LastCheckedOK"

455

478

value_to_set = ""

456

479

457

480

class StartCheckerCmd(PropertyCmd):

458

property = "CheckerRunning"

481

propname = "CheckerRunning"

459

482

value_to_set = dbus.Boolean(True)

460

483

461

484

class StopCheckerCmd(PropertyCmd):

462

property = "CheckerRunning"

485

propname = "CheckerRunning"

463

486

value_to_set = dbus.Boolean(False)

464

487

465

488

class ApproveByDefaultCmd(PropertyCmd):

466

property = "ApprovedByDefault"

489

propname = "ApprovedByDefault"

467

490

value_to_set = dbus.Boolean(True)

468

491

469

492

class DenyByDefaultCmd(PropertyCmd):

470

property = "ApprovedByDefault"

493

propname = "ApprovedByDefault"

471

494

value_to_set = dbus.Boolean(False)

472

495

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

474

property = "Checker"

475

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

477

property = "Host"

478

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

496

class SetCheckerCmd(PropertyValueCmd):

497

propname = "Checker"

498

499

class SetHostCmd(PropertyValueCmd):

500

propname = "Host"

501

502

class SetSecretCmd(PropertyValueCmd):

503

propname = "Secret"

480

504

@property

481

505

def value_to_set(self):

482

506

return self._vts

485

509

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

486

510

self._vts = value.read()

487

511

value.close()

488

property = "Secret"

489

490

class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

491

property = "Timeout"

492

493

class SetExtendedTimeoutCmd(PropertyCmd,

494

MillisecondsValueArgumentMixIn):

495

property = "ExtendedTimeout"

496

497

class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):

498

property = "Interval"

499

500

class SetApprovalDelayCmd(PropertyCmd,

501

MillisecondsValueArgumentMixIn):

502

property = "ApprovalDelay"

503

504

class SetApprovalDurationCmd(PropertyCmd,

505

MillisecondsValueArgumentMixIn):

506

property = "ApprovalDuration"

507

508

def has_actions(options):

509

return any((options.enable,

510

options.disable,

511

options.bump_timeout,

512

options.start_checker,

513

options.stop_checker,

514

options.is_enabled,

515

options.remove,

516

options.checker is not None,

517

options.timeout is not None,

518

options.extended_timeout is not None,

519

options.interval is not None,

520

options.approved_by_default is not None,

521

options.approval_delay is not None,

522

options.approval_duration is not None,

523

options.host is not None,

524

options.secret is not None,

525

options.approve,

526

options.deny))

512

513

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

514

propname = "Timeout"

515

516

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "ExtendedTimeout"

518

519

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "Interval"

521

522

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "ApprovalDelay"

524

525

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDuration"

527

528

def add_command_line_options(parser):

529

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

556

help="Remove client")

557

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

558

help="Set checker command for client")

559

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

559

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

560

help="Set timeout for client")

561

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

561

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

562

help="Set extended timeout for client")

563

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

563

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

564

help="Set checker interval for client")

565

approve_deny_default = parser.add_mutually_exclusive_group()

566

approve_deny_default.add_argument(

571

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

572

dest="approved_by_default",

573

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

574

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

574

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

575

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

576

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

576

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

577

help="Set duration of one client approval")

578

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

579

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

585

help="Approve any current client request")

586

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

587

help="Deny any current client request")

588

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

589

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

588

590

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

589

591

help="Run self-test")

590

592

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

615

617

if options.is_enabled:

616

618

commands.append(IsEnabledCmd())

617

619

618

if options.remove:

619

commands.append(RemoveCmd())

620

621

620

if options.checker is not None:

622

621

commands.append(SetCheckerCmd(options.checker))

623

622

656

655

if options.deny:

657

656

commands.append(DenyCmd())

658

657

658

if options.remove:

659

commands.append(RemoveCmd())

660

659

661

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

660

662

# optionally verbosely

661

663

if not commands:

664

666

return commands

665

667

666

668

667

def main():

668

parser = argparse.ArgumentParser()

669

670

add_command_line_options(parser)

671

672

options = parser.parse_args()

669

def check_option_syntax(parser, options):

670

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

671

argparse"""

672

673

def has_actions(options):

674

return any((options.enable,

675

options.disable,

676

options.bump_timeout,

677

options.start_checker,

678

options.stop_checker,

679

options.is_enabled,

680

options.remove,

681

options.checker is not None,

682

options.timeout is not None,

683

options.extended_timeout is not None,

684

options.interval is not None,

685

options.approved_by_default is not None,

686

options.approval_delay is not None,

687

options.approval_duration is not None,

688

options.host is not None,

689

options.secret is not None,

690

options.approve,

691

options.deny))

673

692

674

693

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

675

694

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

682

701

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

683

702

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

684

703

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

704

if options.remove:

705

options.remove = False

706

if has_actions(options) and not options.deny:

707

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

708

options.remove = True

709

710

711

def main():

712

parser = argparse.ArgumentParser()

713

714

add_command_line_options(parser)

715

716

options = parser.parse_args()

717

718

check_option_syntax(parser, options)

685

719

686

720

clientnames = options.client

687

721

722

if options.debug:

723

log.setLevel(logging.DEBUG)

724

688

725

try:

689

726

bus = dbus.SystemBus()

727

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

728

server_path)

690

729

mandos_dbus_objc = bus.get_object(busname, server_path)

691

730

except dbus.exceptions.DBusException:

692

731

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

705

744

dbus_filter = NullFilter()

706

745

try:

707

746

dbus_logger.addFilter(dbus_filter)

747

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

748

server_path, dbus.OBJECT_MANAGER_IFACE)

708

749

mandos_clients = {path: ifs_and_props[client_interface]

709

750

for path, ifs_and_props in

710

751

mandos_serv_object_manager

722

763

clients = {}

723

764

724

765

if not clientnames:

725

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

726

for path, properties in mandos_clients.items()}

766

clients = {objpath: properties

767

for objpath, properties in mandos_clients.items()}

727

768

else:

728

769

for name in clientnames:

729

for path, client in mandos_clients.items():

730

if client["Name"] == name:

731

client_objc = bus.get_object(busname, path)

732

clients[client_objc] = client

770

for objpath, properties in mandos_clients.items():

771

if properties["Name"] == name:

772

clients[objpath] = properties

733

773

break

734

774

else:

735

775

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

738

778

# Run all commands on clients

739

779

commands = commands_from_options(options)

740

780

for command in commands:

741

command.run(mandos_serv, clients)

781

command.run(clients, bus, mandos_serv)

742

782

743

783

744

784

class Test_milliseconds_to_string(unittest.TestCase):

793

833

testcase = self

794

834

class MockClient(object):

795

835

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

796

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

836

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

797

837

self.attributes = attributes

798

838

self.attributes["Name"] = name

799

839

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]

840

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

841

testcase.assertEqual(interface, client_interface)

842

testcase.assertEqual(dbus_interface,

843

dbus.PROPERTIES_IFACE)

844

self.attributes[propname] = value

845

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

846

testcase.assertEqual(interface, client_interface)

847

testcase.assertEqual(dbus_interface,

848

dbus.PROPERTIES_IFACE)

849

return self.attributes[propname]

810

850

def Approve(self, approve, dbus_interface):

811

851

testcase.assertEqual(dbus_interface, client_interface)

812

852

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

861

901

LastCheckerStatus=-2)

862

902

self.clients = collections.OrderedDict(

863

903

[

864

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

865

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

904

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

905

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

866

906

])

867

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

907

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

908

@property

909

def bus(self):

910

class Bus(object):

911

@staticmethod

912

def get_object(client_bus_name, path):

913

self.assertEqual(client_bus_name, busname)

914

return {

915

"/clients/foo": self.client,

916

"/clients/barbar": self.other_client,

917

}[path]

918

return Bus()

868

919

869

920

class TestPrintTableCmd(TestCmd):

870

921

def test_normal(self):

871

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

922

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

872

923

expected_output = """

873

924

Name Enabled Timeout Last Successful Check

874

925

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

876

927

"""[1:-1]

877

928

self.assertEqual(output, expected_output)

878

929

def test_verbose(self):

879

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

930

output = PrintTableCmd(verbose=True).output(

931

self.clients.values())

880

932

expected_output = """

881

933

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

934

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

884

936

"""[1:-1]

885

937

self.assertEqual(output, expected_output)

886

938

def test_one_client(self):

887

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

939

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

888

940

expected_output = """

889

941

Name Enabled Timeout Last Successful Check

890

942

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

959

1011

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

960

1012

def test_is_enabled_run_exits_successfully(self):

961

1013

with self.assertRaises(SystemExit) as e:

962

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

1014

IsEnabledCmd().run(self.one_client)

963

1015

if e.exception.code is not None:

964

1016

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

965

1017

else:

967

1019

def test_is_enabled_run_exits_with_failure(self):

968

1020

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

969

1021

with self.assertRaises(SystemExit) as e:

970

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

1022

IsEnabledCmd().run(self.one_client)

971

1023

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

972

1024

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

973

1025

else:

982

1034

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

983

1035

mandos = MockMandos()

984

1036

super(TestRemoveCmd, self).setUp()

985

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

1037

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

986

1038

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

987

for client in self.clients:

988

self.assertIn(("RemoveClient",

989

(client.__dbus_object_path__,)),

1039

for clientpath in self.clients:

1040

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

990

1041

mandos.calls)

991

1042

992

1043

class TestApproveCmd(TestCmd):

993

1044

def test_approve(self):

994

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

995

for client in self.clients:

1045

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

1046

for clientpath in self.clients:

1047

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

996

1048

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

997

1049

client.calls)

998

1050

999

1051

class TestDenyCmd(TestCmd):

1000

1052

def test_deny(self):

1001

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

1002

for client in self.clients:

1053

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

1054

for clientpath in self.clients:

1055

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

1003

1056

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

1004

1057

client.calls)

1005

1058

1006

1059

class TestEnableCmd(TestCmd):

1007

1060

def test_enable(self):

1008

for client in self.clients:

1061

for clientpath in self.clients:

1062

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

1009

1063

client.attributes["Enabled"] = False

1010

1064

1011

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

1065

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

1012

1066

1013

for client in self.clients:

1067

for clientpath in self.clients:

1068

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

1014

1069

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

1015

1070

1016

1071

class TestDisableCmd(TestCmd):

1017

1072

def test_disable(self):

1018

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

1019

1020

for client in self.clients:

1073

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

1074

for clientpath in self.clients:

1075

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

1021

1076

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

1022

1077

1023

1078

class Unique(object):

1033

1088

self.values_to_set)

1034

1089

for value_to_set, value_to_get in zip(self.values_to_set,

1035

1090

values_to_get):

1036

for client in self.clients:

1037

old_value = client.attributes[self.property]

1091

for clientpath in self.clients:

1092

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

1093

old_value = client.attributes[self.propname]

1038

1094

self.assertNotIsInstance(old_value, Unique)

1039

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

1095

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

1040

1096

self.run_command(value_to_set, self.clients)

1041

for client in self.clients:

1042

value = client.attributes[self.property]

1097

for clientpath in self.clients:

1098

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

1099

value = client.attributes[self.propname]

1043

1100

self.assertNotIsInstance(value, Unique)

1044

1101

self.assertEqual(value, value_to_get)

1045

1102

def run_command(self, value, clients):

1046

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

1103

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

1047

1104

1048

1105

class TestBumpTimeoutCmd(TestPropertyCmd):

1049

1106

command = BumpTimeoutCmd

1050

property = "LastCheckedOK"

1107

propname = "LastCheckedOK"

1051

1108

values_to_set = [""]

1052

1109

1053

1110

class TestStartCheckerCmd(TestPropertyCmd):

1054

1111

command = StartCheckerCmd

1055

property = "CheckerRunning"

1112

propname = "CheckerRunning"

1056

1113

values_to_set = [dbus.Boolean(True)]

1057

1114

1058

1115

class TestStopCheckerCmd(TestPropertyCmd):

1059

1116

command = StopCheckerCmd

1060

property = "CheckerRunning"

1117

propname = "CheckerRunning"

1061

1118

values_to_set = [dbus.Boolean(False)]

1062

1119

1063

1120

class TestApproveByDefaultCmd(TestPropertyCmd):

1064

1121

command = ApproveByDefaultCmd

1065

property = "ApprovedByDefault"

1122

propname = "ApprovedByDefault"

1066

1123

values_to_set = [dbus.Boolean(True)]

1067

1124

1068

1125

class TestDenyByDefaultCmd(TestPropertyCmd):

1069

1126

command = DenyByDefaultCmd

1070

property = "ApprovedByDefault"

1127

propname = "ApprovedByDefault"

1071

1128

values_to_set = [dbus.Boolean(False)]

1072

1129

1073

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1074

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

1075

ValueArgumentMixIn"""

1130

class TestPropertyValueCmd(TestPropertyCmd):

1131

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

1076

1132

def runTest(self):

1077

if type(self) is TestValueArgumentPropertyCmd:

1133

if type(self) is TestPropertyValueCmd:

1078

1134

return

1079

return super(TestValueArgumentPropertyCmd, self).runTest()

1135

return super(TestPropertyValueCmd, self).runTest()

1080

1136

def run_command(self, value, clients):

1081

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

1137

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

1082

1138

1083

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1139

class TestSetCheckerCmd(TestPropertyValueCmd):

1084

1140

command = SetCheckerCmd

1085

property = "Checker"

1141

propname = "Checker"

1086

1142

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

1087

1143

1088

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1144

class TestSetHostCmd(TestPropertyValueCmd):

1089

1145

command = SetHostCmd

1090

property = "Host"

1146

propname = "Host"

1091

1147

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

1092

1148

1093

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1149

class TestSetSecretCmd(TestPropertyValueCmd):

1094

1150

command = SetSecretCmd

1095

property = "Secret"

1096

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

1151

propname = "Secret"

1152

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

1097

1153

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

1098

1154

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

1099

1155

1100

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1156

class TestSetTimeoutCmd(TestPropertyValueCmd):

1101

1157

command = SetTimeoutCmd

1102

property = "Timeout"

1103

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

1104

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

1158

propname = "Timeout"

1159

values_to_set = [datetime.timedelta(),

1160

datetime.timedelta(minutes=5),

1161

datetime.timedelta(seconds=1),

1162

datetime.timedelta(weeks=1),

1163

datetime.timedelta(weeks=52)]

1164

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

1105

1165

1106

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1166

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1107

1167

command = SetExtendedTimeoutCmd

1108

property = "ExtendedTimeout"

1109

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

1110

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

1168

propname = "ExtendedTimeout"

1169

values_to_set = [datetime.timedelta(),

1170

datetime.timedelta(minutes=5),

1171

datetime.timedelta(seconds=1),

1172

datetime.timedelta(weeks=1),

1173

datetime.timedelta(weeks=52)]

1174

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

1111

1175

1112

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1176

class TestSetIntervalCmd(TestPropertyValueCmd):

1113

1177

command = SetIntervalCmd

1114

property = "Interval"

1115

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

1116

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

1178

propname = "Interval"

1179

values_to_set = [datetime.timedelta(),

1180

datetime.timedelta(minutes=5),

1181

datetime.timedelta(seconds=1),

1182

datetime.timedelta(weeks=1),

1183

datetime.timedelta(weeks=52)]

1184

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

1117

1185

1118

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1186

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1119

1187

command = SetApprovalDelayCmd

1120

property = "ApprovalDelay"

1121

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

1122

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

1188

propname = "ApprovalDelay"

1189

values_to_set = [datetime.timedelta(),

1190

datetime.timedelta(minutes=5),

1191

datetime.timedelta(seconds=1),

1192

datetime.timedelta(weeks=1),

1193

datetime.timedelta(weeks=52)]

1194

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

1123

1195

1124

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1196

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1125

1197

command = SetApprovalDurationCmd

1126

property = "ApprovalDuration"

1127

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

1128

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

1198

propname = "ApprovalDuration"

1199

values_to_set = [datetime.timedelta(),

1200

datetime.timedelta(minutes=5),

1201

datetime.timedelta(seconds=1),

1202

datetime.timedelta(weeks=1),

1203

datetime.timedelta(weeks=52)]

1204

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

1129

1205

1130

1206

class Test_command_from_options(unittest.TestCase):

1131

1207

def setUp(self):

1135

1211

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

1136

1212

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

1137

1213

options = self.parser.parse_args(args)

1214

check_option_syntax(self.parser, options)

1138

1215

commands = commands_from_options(options)

1139

1216

self.assertEqual(len(commands), 1)

1140

1217

command = commands[0]

1149

1226

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

1150

1227

verbose=True)

1151

1228

1229

def test_print_table_verbose_short(self):

1230

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

1231

verbose=True)

1232

1152

1233

def test_enable(self):

1153

1234

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

1154

1235

1236

def test_enable_short(self):

1237

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

1238

1155

1239

def test_disable(self):

1156

1240

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

1157

1241

DisableCmd)

1158

1242

1243

def test_disable_short(self):

1244

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

1245

1159

1246

def test_bump_timeout(self):

1160

1247

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

1161

1248

BumpTimeoutCmd)

1162

1249

1250

def test_bump_timeout_short(self):

1251

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

1252

1163

1253

def test_start_checker(self):

1164

1254

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

1165

1255

StartCheckerCmd)

1172

1262

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

1173

1263

RemoveCmd)

1174

1264

1265

def test_remove_short(self):

1266

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

1267

1175

1268

def test_checker(self):

1176

1269

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

1177

1270

SetCheckerCmd, value_to_set=":")

1180

1273

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

1181

1274

SetCheckerCmd, value_to_set="")

1182

1275

1276

def test_checker_short(self):

1277

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

1278

SetCheckerCmd, value_to_set=":")

1279

1183

1280

def test_timeout(self):

1184

1281

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

1185

1282

SetTimeoutCmd,

1186

1283

value_to_set=300000)

1187

1284

1285

def test_timeout_short(self):

1286

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

1287

SetTimeoutCmd,

1288

value_to_set=300000)

1289

1188

1290

def test_extended_timeout(self):

1189

1291

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

1190

1292

"foo"],

1196

1298

SetIntervalCmd,

1197

1299

value_to_set=120000)

1198

1300

1301

def test_interval_short(self):

1302

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

1303

SetIntervalCmd,

1304

value_to_set=120000)

1305

1199

1306

def test_approve_by_default(self):

1200

1307

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

1201

1308

ApproveByDefaultCmd)

1219

1326

"foo"], SetHostCmd,

1220

1327

value_to_set="foo.example.org")

1221

1328

1329

def test_host_short(self):

1330

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

1331

"foo"], SetHostCmd,

1332

value_to_set="foo.example.org")

1333

1222

1334

def test_secret_devnull(self):

1223

1335

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

1224

1336

"foo"], SetSecretCmd,

1233

1345

"foo"], SetSecretCmd,

1234

1346

value_to_set=value)

1235

1347

1348

def test_secret_devnull_short(self):

1349

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

1350

SetSecretCmd, value_to_set=b"")

1351

1352

def test_secret_tempfile_short(self):

1353

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

1354

value = b"secret\0xyzzy\nbar"

1355

f.write(value)

1356

f.seek(0)

1357

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

1358

SetSecretCmd,

1359

value_to_set=value)

1360

1236

1361

def test_approve(self):

1237

1362

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

1238

1363

ApproveCmd)

1239

1364

1365

def test_approve_short(self):

1366

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

1367

1240

1368

def test_deny(self):

1241

1369

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

1242

1370

1371

def test_deny_short(self):

1372

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

1373

1243

1374

def test_dump_json(self):

1244

1375

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

1245

1376

1247

1378

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

1248

1379

IsEnabledCmd)

1249

1380

1381

def test_is_enabled_short(self):

1382

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

1383

1384

def test_deny_before_remove(self):

1385

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

1386

check_option_syntax(self.parser, options)

1387

commands = commands_from_options(options)

1388

self.assertEqual(len(commands), 2)

1389

self.assertIsInstance(commands[0], DenyCmd)

1390

self.assertIsInstance(commands[1], RemoveCmd)

1391

1392

def test_deny_before_remove_reversed(self):

1393

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

1394

check_option_syntax(self.parser, options)

1395

commands = commands_from_options(options)

1396

self.assertEqual(len(commands), 2)

1397

self.assertIsInstance(commands[0], DenyCmd)

1398

self.assertIsInstance(commands[1], RemoveCmd)

1399

1400

1401

class Test_check_option_syntax(unittest.TestCase):

1402

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

1403

# check_option_syntax()

1404

actions = {

1405

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

1406

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

1407

# never used

1408

"enable": True,

1409

"disable": True,

1410

"bump_timeout": True,

1411

"start_checker": True,

1412

"stop_checker": True,

1413

"is_enabled": True,

1414

"remove": True,

1415

"checker": "x",

1416

"timeout": datetime.timedelta(),

1417

"extended_timeout": datetime.timedelta(),

1418

"interval": datetime.timedelta(),

1419

"approved_by_default": True,

1420

"approval_delay": datetime.timedelta(),

1421

"approval_duration": datetime.timedelta(),

1422

"host": "x",

1423

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

1424

"approve": True,

1425

"deny": True,

1426

}

1427

1428

def setUp(self):

1429

self.parser = argparse.ArgumentParser()

1430

add_command_line_options(self.parser)

1431

1432

@contextlib.contextmanager

1433

def assertParseError(self):

1434

with self.assertRaises(SystemExit) as e:

1435

with self.temporarily_suppress_stderr():

1436

yield

1437

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

1438

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

1439

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

1440

1441

@staticmethod

1442

@contextlib.contextmanager

1443

def temporarily_suppress_stderr():

1444

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

1445

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

1446

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

1447

os.close(null)

1448

try:

1449

yield

1450

finally:

1451

# restore stderr

1452

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

1453

os.close(stderrcopy)

1454

1455

def check_option_syntax(self, options):

1456

check_option_syntax(self.parser, options)

1457

1458

def test_actions_requires_client_or_all(self):

1459

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

1460

options = self.parser.parse_args()

1461

setattr(options, action, value)

1462

with self.assertParseError():

1463

self.check_option_syntax(options)

1464

1465

def test_actions_conflicts_with_verbose(self):

1466

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

1467

options = self.parser.parse_args()

1468

setattr(options, action, value)

1469

options.verbose = True

1470

with self.assertParseError():

1471

self.check_option_syntax(options)

1472

1473

def test_dump_json_conflicts_with_verbose(self):

1474

options = self.parser.parse_args()

1475

options.dump_json = True

1476

options.verbose = True

1477

with self.assertParseError():

1478

self.check_option_syntax(options)

1479

1480

def test_dump_json_conflicts_with_action(self):

1481

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

1482

options = self.parser.parse_args()

1483

setattr(options, action, value)

1484

options.dump_json = True

1485

with self.assertParseError():

1486

self.check_option_syntax(options)

1487

1488

def test_all_can_not_be_alone(self):

1489

options = self.parser.parse_args()

1490

options.all = True

1491

with self.assertParseError():

1492

self.check_option_syntax(options)

1493

1494

def test_all_is_ok_with_any_action(self):

1495

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

1496

options = self.parser.parse_args()

1497

setattr(options, action, value)

1498

options.all = True

1499

self.check_option_syntax(options)

1500

1501

def test_is_enabled_fails_without_client(self):

1502

options = self.parser.parse_args()

1503

options.is_enabled = True

1504

with self.assertParseError():

1505

self.check_option_syntax(options)

1506

1507

def test_is_enabled_works_with_one_client(self):

1508

options = self.parser.parse_args()

1509

options.is_enabled = True

1510

options.client = ["foo"]

1511

self.check_option_syntax(options)

1512

1513

def test_is_enabled_fails_with_two_clients(self):

1514

options = self.parser.parse_args()

1515

options.is_enabled = True

1516

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

1517

with self.assertParseError():

1518

self.check_option_syntax(options)

1519

1520

def test_remove_can_only_be_combined_with_action_deny(self):

1521

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

1522

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

1523

continue

1524

options = self.parser.parse_args()

1525

setattr(options, action, value)

1526

options.all = True

1527

options.remove = True

1528

with self.assertParseError():

1529

self.check_option_syntax(options)

1530

1250

1531

1251

1532

1252

1533

def should_only_run_tests():