/mandos/trunk : revision 1035

44

import logging

45

import io

46

import tempfile

47

import contextlib

48

47

49

48

import dbus

50

49

64

63

65

64

locale.setlocale(locale.LC_ALL, "")

66

65

67

dbus_busname_domain = "se.recompile"

68

dbus_busname = dbus_busname_domain + ".Mandos"

69

server_dbus_path = "/"

70

server_dbus_interface = dbus_busname_domain + ".Mandos"

71

client_dbus_interface = dbus_busname_domain + ".Mandos.Client"

72

del dbus_busname_domain

66

domain = "se.recompile"

67

busname = domain + ".Mandos"

68

server_path = "/"

69

server_interface = domain + ".Mandos"

70

client_interface = domain + ".Mandos.Client"

73

71

version = "1.8.3"

74

72

75

73

277

275

# Abstract classes first

278

276

class Command(object):

279

277

"""Abstract class for commands"""

280

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

278

def run(self, mandos, clients):

281

279

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

282

280

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

283

281

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

284

282

self.mandos = mandos

285

for clientpath, properties in clients.items():

286

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

287

dbus_busname, str(clientpath))

288

client = bus.get_object(dbus_busname, clientpath)

283

for client, properties in clients.items():

289

284

self.run_on_one_client(client, properties)

290

285

291

286

class PrintCmd(Command):

297

292

"LastApprovalRequest", "ApprovalDelay",

298

293

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

299

294

"Expires", "LastCheckerStatus")

300

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

301

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

302

def output(self, clients):

303

raise NotImplementedError()

295

def run(self, mandos, clients):

296

print(self.output(clients))

304

297

305

298

class PropertyCmd(Command):

306

299

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

307

300

def run_on_one_client(self, client, properties):

308

301

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

309

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

310

client.__dbus_object_path__,

311

dbus.PROPERTIES_IFACE, client_dbus_interface,

312

self.propname, self.value_to_set

313

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

314

else bool(self.value_to_set))

315

client.Set(client_dbus_interface, self.propname,

316

self.value_to_set,

302

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

317

303

dbus_interface=dbus.PROPERTIES_IFACE)

318

@property

319

def propname(self):

320

raise NotImplementedError()

321

304

322

class PropertyValueCmd(PropertyCmd):

323

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

305

class ValueArgumentMixIn(object):

306

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

324

307

def __init__(self, value):

325

308

self.value_to_set = value

326

309

327

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

328

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

329

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

310

class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):

311

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

312

milliseconds."""

330

313

@property

331

314

def value_to_set(self):

332

315

return self._vts

333

316

@value_to_set.setter

334

317

def value_to_set(self, value):

335

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

318

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

336

319

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

337

320

338

321

# Actual (non-abstract) command classes

342

325

self.verbose = verbose

343

326

344

327

def output(self, clients):

345

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

346

"LastCheckedOK")

328

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

347

329

keywords = default_keywords

348

330

if self.verbose:

349

331

keywords = self.all_keywords

350

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

332

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

351

333

352

334

class TableOfClients(object):

353

335

tableheaders = {

439

421

return value

440

422

441

423

class IsEnabledCmd(Command):

442

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

443

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

424

def run_on_one_client(self, client, properties):

444

425

if self.is_enabled(client, properties):

445

426

sys.exit(0)

446

427

sys.exit(1)

447

428

def is_enabled(self, client, properties):

448

return properties["Enabled"]

429

return bool(properties["Enabled"])

449

430

450

431

class RemoveCmd(Command):

451

432

def run_on_one_client(self, client, properties):

452

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

453

server_dbus_path, server_dbus_interface,

454

str(client.__dbus_object_path__))

455

433

self.mandos.RemoveClient(client.__dbus_object_path__)

456

434

457

435

class ApproveCmd(Command):

458

436

def run_on_one_client(self, client, properties):

459

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

460

client.__dbus_object_path__, client_dbus_interface)

461

437

client.Approve(dbus.Boolean(True),

462

dbus_interface=client_dbus_interface)

438

dbus_interface=client_interface)

463

439

464

440

class DenyCmd(Command):

465

441

def run_on_one_client(self, client, properties):

466

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

467

client.__dbus_object_path__, client_dbus_interface)

468

442

client.Approve(dbus.Boolean(False),

469

dbus_interface=client_dbus_interface)

443

dbus_interface=client_interface)

470

444

471

445

class EnableCmd(PropertyCmd):

472

propname = "Enabled"

446

property = "Enabled"

473

447

value_to_set = dbus.Boolean(True)

474

448

475

449

class DisableCmd(PropertyCmd):

476

propname = "Enabled"

450

property = "Enabled"

477

451

value_to_set = dbus.Boolean(False)

478

452

479

453

class BumpTimeoutCmd(PropertyCmd):

480

propname = "LastCheckedOK"

454

property = "LastCheckedOK"

481

455

value_to_set = ""

482

456

483

457

class StartCheckerCmd(PropertyCmd):

484

propname = "CheckerRunning"

458

property = "CheckerRunning"

485

459

value_to_set = dbus.Boolean(True)

486

460

487

461

class StopCheckerCmd(PropertyCmd):

488

propname = "CheckerRunning"

462

property = "CheckerRunning"

489

463

value_to_set = dbus.Boolean(False)

490

464

491

465

class ApproveByDefaultCmd(PropertyCmd):

492

propname = "ApprovedByDefault"

466

property = "ApprovedByDefault"

493

467

value_to_set = dbus.Boolean(True)

494

468

495

469

class DenyByDefaultCmd(PropertyCmd):

496

propname = "ApprovedByDefault"

470

property = "ApprovedByDefault"

497

471

value_to_set = dbus.Boolean(False)

498

472

499

class SetCheckerCmd(PropertyValueCmd):

500

propname = "Checker"

501

502

class SetHostCmd(PropertyValueCmd):

503

propname = "Host"

504

505

class SetSecretCmd(PropertyValueCmd):

506

propname = "Secret"

473

class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):

474

property = "Checker"

475

476

class SetHostCmd(PropertyCmd, ValueArgumentMixIn):

477

property = "Host"

478

479

class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):

507

480

@property

508

481

def value_to_set(self):

509

482

return self._vts

512

485

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

513

486

self._vts = value.read()

514

487

value.close()

515

516

class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

517

propname = "Timeout"

518

519

class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):

520

propname = "ExtendedTimeout"

521

522

class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):

523

propname = "Interval"

524

525

class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):

526

propname = "ApprovalDelay"

527

528

class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):

529

propname = "ApprovalDuration"

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"

530

507

531

508

def add_command_line_options(parser):

532

509

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

588

565

help="Approve any current client request")

589

566

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

590

567

help="Deny any current client request")

591

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

592

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

593

568

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

594

569

help="Run self-test")

595

570

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

620

595

if options.is_enabled:

621

596

commands.append(IsEnabledCmd())

622

597

598

if options.remove:

599

commands.append(RemoveCmd())

600

623

601

if options.checker is not None:

624

602

commands.append(SetCheckerCmd(options.checker))

625

603

658

636

if options.deny:

659

637

commands.append(DenyCmd())

660

638

661

if options.remove:

662

commands.append(RemoveCmd())

663

664

639

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

665

640

# optionally verbosely

666

641

if not commands:

669

644

return commands

670

645

671

646

672

def check_option_syntax(parser, options):

673

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

674

argparse"""

647

def main():

648

parser = argparse.ArgumentParser()

649

650

add_command_line_options(parser)

651

652

options = parser.parse_args()

675

653

676

654

def has_actions(options):

677

655

return any((options.enable,

704

682

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

705

683

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

706

684

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

707

if options.remove:

708

options.remove = False

709

if has_actions(options) and not options.deny:

710

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

711

options.remove = True

712

713

714

def main():

715

parser = argparse.ArgumentParser()

716

717

add_command_line_options(parser)

718

719

options = parser.parse_args()

720

721

check_option_syntax(parser, options)

722

685

723

686

clientnames = options.client

724

687

725

if options.debug:

726

log.setLevel(logging.DEBUG)

727

728

688

try:

729

689

bus = dbus.SystemBus()

730

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

731

dbus_busname, server_dbus_path)

732

mandos_dbus_objc = bus.get_object(dbus_busname,

733

server_dbus_path)

690

mandos_dbus_objc = bus.get_object(busname, server_path)

734

691

except dbus.exceptions.DBusException:

735

692

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

736

693

sys.exit(1)

737

694

738

695

mandos_serv = dbus.Interface(mandos_dbus_objc,

739

dbus_interface=server_dbus_interface)

696

dbus_interface=server_interface)

740

697

mandos_serv_object_manager = dbus.Interface(

741

698

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

742

699

748

705

dbus_filter = NullFilter()

749

706

try:

750

707

dbus_logger.addFilter(dbus_filter)

751

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

752

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

753

mandos_clients = {path: ifs_and_props[client_dbus_interface]

708

mandos_clients = {path: ifs_and_props[client_interface]

754

709

for path, ifs_and_props in

755

710

mandos_serv_object_manager

756

711

.GetManagedObjects().items()

757

if client_dbus_interface in ifs_and_props}

712

if client_interface in ifs_and_props}

758

713

except dbus.exceptions.DBusException as e:

759

714

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

760

715

"\n%s", e)

767

722

clients = {}

768

723

769

724

if not clientnames:

770

clients = {objpath: properties

771

for objpath, properties in mandos_clients.items()}

725

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

726

for path, properties in mandos_clients.items()}

772

727

else:

773

728

for name in clientnames:

774

for objpath, properties in mandos_clients.items():

775

if properties["Name"] == name:

776

clients[objpath] = properties

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

777

733

break

778

734

else:

779

735

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

782

738

# Run all commands on clients

783

739

commands = commands_from_options(options)

784

740

for command in commands:

785

command.run(clients, bus, mandos_serv)

741

command.run(mandos_serv, clients)

786

742

787

743

788

744

class Test_milliseconds_to_string(unittest.TestCase):

837

793

testcase = self

838

794

class MockClient(object):

839

795

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

840

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

796

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

841

797

self.attributes = attributes

842

798

self.attributes["Name"] = name

843

799

self.calls = []

844

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

845

testcase.assertEqual(interface, client_dbus_interface)

846

testcase.assertEqual(dbus_interface,

847

dbus.PROPERTIES_IFACE)

848

self.attributes[propname] = value

849

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

850

testcase.assertEqual(interface, client_dbus_interface)

851

testcase.assertEqual(dbus_interface,

852

dbus.PROPERTIES_IFACE)

853

return self.attributes[propname]

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]

854

810

def Approve(self, approve, dbus_interface):

855

testcase.assertEqual(dbus_interface,

856

client_dbus_interface)

811

testcase.assertEqual(dbus_interface, client_interface)

857

812

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

858

813

dbus_interface)))

859

814

self.client = MockClient(

906

861

LastCheckerStatus=-2)

907

862

self.clients = collections.OrderedDict(

908

863

[

909

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

910

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

864

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

865

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

911

866

])

912

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

913

@property

914

def bus(self):

915

class Bus(object):

916

@staticmethod

917

def get_object(client_bus_name, path):

918

self.assertEqual(client_bus_name, dbus_busname)

919

return {

920

"/clients/foo": self.client,

921

"/clients/barbar": self.other_client,

922

}[path]

923

return Bus()

867

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

924

868

925

869

class TestPrintTableCmd(TestCmd):

926

870

def test_normal(self):

927

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

928

expected_output = "\n".join((

929

"Name Enabled Timeout Last Successful Check",

930

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

931

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

932

))

871

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

872

expected_output = """

873

Name Enabled Timeout Last Successful Check

874

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

875

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

876

"""[1:-1]

933

877

self.assertEqual(output, expected_output)

934

878

def test_verbose(self):

935

output = PrintTableCmd(verbose=True).output(

936

self.clients.values())

937

expected_output = "\n".join((

938

# First line (headers)

939

"Name Enabled Timeout Last Successful Check Created "

940

" Interval Host Key ID "

941

" Fingerprint "

942

" Check Is Running Last Enabl"

943

"ed Approval Is Pending Approved By Default Last A"

944

"pproval Request Approval Delay Approval Duration Checker"

945

" Extended Timeout Expires Last "

946

"Checker Status",

947

# Second line (client "foo")

948

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

949

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

950

"574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7"

951

"DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03"

952

"T00:00:00 No Yes "

953

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

954

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

955

" ",

956

# Third line (client "barbar")

957

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

958

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

959

"3b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB8"

960

"4C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04"

961

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

962

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

963

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

964

" ",

965

))

879

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

880

expected_output = """

881

Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status

882

foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0

883

barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2

884

"""[1:-1]

966

885

self.assertEqual(output, expected_output)

967

886

def test_one_client(self):

968

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

887

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

969

888

expected_output = """

970

889

Name Enabled Timeout Last Successful Check

971

890

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

1036

955

1037

956

class TestIsEnabledCmd(TestCmd):

1038

957

def test_is_enabled(self):

1039

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

1040

properties)

1041

for client, properties

1042

in self.clients.items()))

958

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

959

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

1043

960

def test_is_enabled_run_exits_successfully(self):

1044

961

with self.assertRaises(SystemExit) as e:

1045

IsEnabledCmd().run(self.one_client)

962

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

1046

963

if e.exception.code is not None:

1047

964

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

1048

965

else:

1050

967

def test_is_enabled_run_exits_with_failure(self):

1051

968

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

1052

969

with self.assertRaises(SystemExit) as e:

1053

IsEnabledCmd().run(self.one_client)

970

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

1054

971

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

1055

972

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

1056

973

else:

1065

982

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

1066

983

mandos = MockMandos()

1067

984

super(TestRemoveCmd, self).setUp()

1068

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

985

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

1069

986

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

1070

for clientpath in self.clients:

1071

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

987

for client in self.clients:

988

self.assertIn(("RemoveClient",

989

(client.__dbus_object_path__,)),

1072

990

mandos.calls)

1073

991

1074

992

class TestApproveCmd(TestCmd):

1075

993

def test_approve(self):

1076

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

1077

for clientpath in self.clients:

1078

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

1079

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

994

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

995

for client in self.clients:

996

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

1080

997

client.calls)

1081

998

1082

999

class TestDenyCmd(TestCmd):

1083

1000

def test_deny(self):

1084

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

1085

for clientpath in self.clients:

1086

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

1087

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

1001

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

1002

for client in self.clients:

1003

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

1088

1004

client.calls)

1089

1005

1090

1006

class TestEnableCmd(TestCmd):

1091

1007

def test_enable(self):

1092

for clientpath in self.clients:

1093

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

1008

for client in self.clients:

1094

1009

client.attributes["Enabled"] = False

1095

1010

1096

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

1011

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

1097

1012

1098

for clientpath in self.clients:

1099

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

1013

for client in self.clients:

1100

1014

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

1101

1015

1102

1016

class TestDisableCmd(TestCmd):

1103

1017

def test_disable(self):

1104

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

1105

for clientpath in self.clients:

1106

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

1018

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

1019

1020

for client in self.clients:

1107

1021

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

1108

1022

1109

1023

class Unique(object):

1119

1033

self.values_to_set)

1120

1034

for value_to_set, value_to_get in zip(self.values_to_set,

1121

1035

values_to_get):

1122

for clientpath in self.clients:

1123

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

1124

old_value = client.attributes[self.propname]

1036

for client in self.clients:

1037

old_value = client.attributes[self.property]

1125

1038

self.assertNotIsInstance(old_value, Unique)

1126

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

1039

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

1127

1040

self.run_command(value_to_set, self.clients)

1128

for clientpath in self.clients:

1129

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

1130

value = client.attributes[self.propname]

1041

for client in self.clients:

1042

value = client.attributes[self.property]

1131

1043

self.assertNotIsInstance(value, Unique)

1132

1044

self.assertEqual(value, value_to_get)

1133

1045

def run_command(self, value, clients):

1134

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

1046

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

1135

1047

1136

1048

class TestBumpTimeoutCmd(TestPropertyCmd):

1137

1049

command = BumpTimeoutCmd

1138

propname = "LastCheckedOK"

1050

property = "LastCheckedOK"

1139

1051

values_to_set = [""]

1140

1052

1141

1053

class TestStartCheckerCmd(TestPropertyCmd):

1142

1054

command = StartCheckerCmd

1143

propname = "CheckerRunning"

1055

property = "CheckerRunning"

1144

1056

values_to_set = [dbus.Boolean(True)]

1145

1057

1146

1058

class TestStopCheckerCmd(TestPropertyCmd):

1147

1059

command = StopCheckerCmd

1148

propname = "CheckerRunning"

1060

property = "CheckerRunning"

1149

1061

values_to_set = [dbus.Boolean(False)]

1150

1062

1151

1063

class TestApproveByDefaultCmd(TestPropertyCmd):

1152

1064

command = ApproveByDefaultCmd

1153

propname = "ApprovedByDefault"

1065

property = "ApprovedByDefault"

1154

1066

values_to_set = [dbus.Boolean(True)]

1155

1067

1156

1068

class TestDenyByDefaultCmd(TestPropertyCmd):

1157

1069

command = DenyByDefaultCmd

1158

propname = "ApprovedByDefault"

1070

property = "ApprovedByDefault"

1159

1071

values_to_set = [dbus.Boolean(False)]

1160

1072

1161

class TestPropertyValueCmd(TestPropertyCmd):

1162

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

1073

class TestValueArgumentPropertyCmd(TestPropertyCmd):

1074

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

1075

ValueArgumentMixIn"""

1163

1076

def runTest(self):

1164

if type(self) is TestPropertyValueCmd:

1077

if type(self) is TestValueArgumentPropertyCmd:

1165

1078

return

1166

return super(TestPropertyValueCmd, self).runTest()

1079

return super(TestValueArgumentPropertyCmd, self).runTest()

1167

1080

def run_command(self, value, clients):

1168

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

1081

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

1169

1082

1170

class TestSetCheckerCmd(TestPropertyValueCmd):

1083

class TestSetCheckerCmd(TestValueArgumentPropertyCmd):

1171

1084

command = SetCheckerCmd

1172

propname = "Checker"

1085

property = "Checker"

1173

1086

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

1174

1087

1175

class TestSetHostCmd(TestPropertyValueCmd):

1088

class TestSetHostCmd(TestValueArgumentPropertyCmd):

1176

1089

command = SetHostCmd

1177

propname = "Host"

1090

property = "Host"

1178

1091

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

1179

1092

1180

class TestSetSecretCmd(TestPropertyValueCmd):

1093

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1181

1094

command = SetSecretCmd

1182

propname = "Secret"

1183

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

1095

property = "Secret"

1096

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

1184

1097

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

1185

1098

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

1186

1099

1187

class TestSetTimeoutCmd(TestPropertyValueCmd):

1100

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1188

1101

command = SetTimeoutCmd

1189

propname = "Timeout"

1102

property = "Timeout"

1190

1103

values_to_set = [datetime.timedelta(),

1191

1104

datetime.timedelta(minutes=5),

1192

1105

datetime.timedelta(seconds=1),

1194

1107

datetime.timedelta(weeks=52)]

1195

1108

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

1196

1109

1197

class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):

1110

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1198

1111

command = SetExtendedTimeoutCmd

1199

propname = "ExtendedTimeout"

1112

property = "ExtendedTimeout"

1200

1113

values_to_set = [datetime.timedelta(),

1201

1114

datetime.timedelta(minutes=5),

1202

1115

datetime.timedelta(seconds=1),

1204

1117

datetime.timedelta(weeks=52)]

1205

1118

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

1206

1119

1207

class TestSetIntervalCmd(TestPropertyValueCmd):

1120

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1208

1121

command = SetIntervalCmd

1209

propname = "Interval"

1122

property = "Interval"

1210

1123

values_to_set = [datetime.timedelta(),

1211

1124

datetime.timedelta(minutes=5),

1212

1125

datetime.timedelta(seconds=1),

1214

1127

datetime.timedelta(weeks=52)]

1215

1128

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

1216

1129

1217

class TestSetApprovalDelayCmd(TestPropertyValueCmd):

1130

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1218

1131

command = SetApprovalDelayCmd

1219

propname = "ApprovalDelay"

1132

property = "ApprovalDelay"

1220

1133

values_to_set = [datetime.timedelta(),

1221

1134

datetime.timedelta(minutes=5),

1222

1135

datetime.timedelta(seconds=1),

1224

1137

datetime.timedelta(weeks=52)]

1225

1138

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

1226

1139

1227

class TestSetApprovalDurationCmd(TestPropertyValueCmd):

1140

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1228

1141

command = SetApprovalDurationCmd

1229

propname = "ApprovalDuration"

1142

property = "ApprovalDuration"

1230

1143

values_to_set = [datetime.timedelta(),

1231

1144

datetime.timedelta(minutes=5),

1232

1145

datetime.timedelta(seconds=1),

1238

1151

def setUp(self):

1239

1152

self.parser = argparse.ArgumentParser()

1240

1153

add_command_line_options(self.parser)

1241

def assert_command_from_args(self, args, command_cls,

1242

**cmd_attrs):

1154

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

1243

1155

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

1244

1156

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

1245

1157

options = self.parser.parse_args(args)

1246

check_option_syntax(self.parser, options)

1247

1158

commands = commands_from_options(options)

1248

1159

self.assertEqual(len(commands), 1)

1249

1160

command = commands[0]

1258

1169

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

1259

1170

verbose=True)

1260

1171

1261

def test_print_table_verbose_short(self):

1262

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

1263

verbose=True)

1264

1265

1172

def test_enable(self):

1266

1173

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

1267

1174

1268

def test_enable_short(self):

1269

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

1270

1271

1175

def test_disable(self):

1272

1176

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

1273

1177

DisableCmd)

1274

1178

1275

def test_disable_short(self):

1276

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

1277

1278

1179

def test_bump_timeout(self):

1279

1180

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

1280

1181

BumpTimeoutCmd)

1281

1182

1282

def test_bump_timeout_short(self):

1283

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

1284

1285

1183

def test_start_checker(self):

1286

1184

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

1287

1185

StartCheckerCmd)

1294

1192

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

1295

1193

RemoveCmd)

1296

1194

1297

def test_remove_short(self):

1298

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

1299

1300

1195

def test_checker(self):

1301

1196

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

1302

1197

SetCheckerCmd, value_to_set=":")

1305

1200

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

1306

1201

SetCheckerCmd, value_to_set="")

1307

1202

1308

def test_checker_short(self):

1309

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

1310

SetCheckerCmd, value_to_set=":")

1311

1312

1203

def test_timeout(self):

1313

1204

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

1314

1205

SetTimeoutCmd,

1315

1206

value_to_set=300000)

1316

1207

1317

def test_timeout_short(self):

1318

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

1319

SetTimeoutCmd,

1320

value_to_set=300000)

1321

1322

1208

def test_extended_timeout(self):

1323

1209

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

1324

1210

"foo"],

1330

1216

SetIntervalCmd,

1331

1217

value_to_set=120000)

1332

1218

1333

def test_interval_short(self):

1334

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

1335

SetIntervalCmd,

1336

value_to_set=120000)

1337

1338

1219

def test_approve_by_default(self):

1339

1220

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

1340

1221

ApproveByDefaultCmd)

1358

1239

"foo"], SetHostCmd,

1359

1240

value_to_set="foo.example.org")

1360

1241

1361

def test_host_short(self):

1362

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

1363

"foo"], SetHostCmd,

1364

value_to_set="foo.example.org")

1365

1366

1242

def test_secret_devnull(self):

1367

1243

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

1368

1244

"foo"], SetSecretCmd,

1377

1253

"foo"], SetSecretCmd,

1378

1254

value_to_set=value)

1379

1255

1380

def test_secret_devnull_short(self):

1381

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

1382

SetSecretCmd, value_to_set=b"")

1383

1384

def test_secret_tempfile_short(self):

1385

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

1386

value = b"secret\0xyzzy\nbar"

1387

f.write(value)

1388

f.seek(0)

1389

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

1390

SetSecretCmd,

1391

value_to_set=value)

1392

1393

1256

def test_approve(self):

1394

1257

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

1395

1258

ApproveCmd)

1396

1259

1397

def test_approve_short(self):

1398

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

1399

1400

1260

def test_deny(self):

1401

1261

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

1402

1262

1403

def test_deny_short(self):

1404

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

1405

1406

1263

def test_dump_json(self):

1407

1264

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

1408

1265

1410

1267

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

1411

1268

IsEnabledCmd)

1412

1269

1413

def test_is_enabled_short(self):

1414

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

1415

1416

def test_deny_before_remove(self):

1417

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

1418

"foo"])

1419

check_option_syntax(self.parser, options)

1420

commands = commands_from_options(options)

1421

self.assertEqual(len(commands), 2)

1422

self.assertIsInstance(commands[0], DenyCmd)

1423

self.assertIsInstance(commands[1], RemoveCmd)

1424

1425

def test_deny_before_remove_reversed(self):

1426

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

1427

"--all"])

1428

check_option_syntax(self.parser, options)

1429

commands = commands_from_options(options)

1430

self.assertEqual(len(commands), 2)

1431

self.assertIsInstance(commands[0], DenyCmd)

1432

self.assertIsInstance(commands[1], RemoveCmd)

1433

1434

1435

class Test_check_option_syntax(unittest.TestCase):

1436

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

1437

# check_option_syntax()

1438

actions = {

1439

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

1440

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

1441

# never used

1442

"enable": True,

1443

"disable": True,

1444

"bump_timeout": True,

1445

"start_checker": True,

1446

"stop_checker": True,

1447

"is_enabled": True,

1448

"remove": True,

1449

"checker": "x",

1450

"timeout": datetime.timedelta(),

1451

"extended_timeout": datetime.timedelta(),

1452

"interval": datetime.timedelta(),

1453

"approved_by_default": True,

1454

"approval_delay": datetime.timedelta(),

1455

"approval_duration": datetime.timedelta(),

1456

"host": "x",

1457

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

1458

"approve": True,

1459

"deny": True,

1460

}

1461

1462

def setUp(self):

1463

self.parser = argparse.ArgumentParser()

1464

add_command_line_options(self.parser)

1465

1466

@contextlib.contextmanager

1467

def assertParseError(self):

1468

with self.assertRaises(SystemExit) as e:

1469

with self.temporarily_suppress_stderr():

1470

yield

1471

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

1472

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

1473

# /argparse.html#exiting-methods

1474

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

1475

1476

@staticmethod

1477

@contextlib.contextmanager

1478

def temporarily_suppress_stderr():

1479

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

1480

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

1481

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

1482

os.close(null)

1483

try:

1484

yield

1485

finally:

1486

# restore stderr

1487

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

1488

os.close(stderrcopy)

1489

1490

def check_option_syntax(self, options):

1491

check_option_syntax(self.parser, options)

1492

1493

def test_actions_requires_client_or_all(self):

1494

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

1495

options = self.parser.parse_args()

1496

setattr(options, action, value)

1497

with self.assertParseError():

1498

self.check_option_syntax(options)

1499

1500

def test_actions_conflicts_with_verbose(self):

1501

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

1502

options = self.parser.parse_args()

1503

setattr(options, action, value)

1504

options.verbose = True

1505

with self.assertParseError():

1506

self.check_option_syntax(options)

1507

1508

def test_dump_json_conflicts_with_verbose(self):

1509

options = self.parser.parse_args()

1510

options.dump_json = True

1511

options.verbose = True

1512

with self.assertParseError():

1513

self.check_option_syntax(options)

1514

1515

def test_dump_json_conflicts_with_action(self):

1516

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

1517

options = self.parser.parse_args()

1518

setattr(options, action, value)

1519

options.dump_json = True

1520

with self.assertParseError():

1521

self.check_option_syntax(options)

1522

1523

def test_all_can_not_be_alone(self):

1524

options = self.parser.parse_args()

1525

options.all = True

1526

with self.assertParseError():

1527

self.check_option_syntax(options)

1528

1529

def test_all_is_ok_with_any_action(self):

1530

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

1531

options = self.parser.parse_args()

1532

setattr(options, action, value)

1533

options.all = True

1534

self.check_option_syntax(options)

1535

1536

def test_is_enabled_fails_without_client(self):

1537

options = self.parser.parse_args()

1538

options.is_enabled = True

1539

with self.assertParseError():

1540

self.check_option_syntax(options)

1541

1542

def test_is_enabled_works_with_one_client(self):

1543

options = self.parser.parse_args()

1544

options.is_enabled = True

1545

options.client = ["foo"]

1546

self.check_option_syntax(options)

1547

1548

def test_is_enabled_fails_with_two_clients(self):

1549

options = self.parser.parse_args()

1550

options.is_enabled = True

1551

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

1552

with self.assertParseError():

1553

self.check_option_syntax(options)

1554

1555

def test_remove_can_only_be_combined_with_action_deny(self):

1556

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

1557

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

1558

continue

1559

options = self.parser.parse_args()

1560

setattr(options, action, value)

1561

options.all = True

1562

options.remove = True

1563

with self.assertParseError():

1564

self.check_option_syntax(options)

1565

1566

1270

1567

1271

1568

1272

def should_only_run_tests():