/mandos/trunk : revision 1055

93

if options.debug:

94

log.setLevel(logging.DEBUG)

95

96

bus = dbus.SystemBus()

97

98

mandos_dbus_object = get_mandos_dbus_object(bus)

99

100

mandos_serv = dbus.Interface(

101

mandos_dbus_object, dbus_interface=server_dbus_interface)

96

try:

97

bus = dbus.SystemBus()

98

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

99

dbus_busname, server_dbus_path)

100

mandos_dbus_objc = bus.get_object(dbus_busname,

101

server_dbus_path)

102

except dbus.exceptions.DBusException:

103

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

104

sys.exit(1)

105

106

mandos_serv = dbus.Interface(mandos_dbus_objc,

107

dbus_interface=server_dbus_interface)

102

108

mandos_serv_object_manager = dbus.Interface(

103

mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

104

105

managed_objects = get_managed_objects(mandos_serv_object_manager)

106

107

all_clients = {}

108

for path, ifs_and_props in managed_objects.items():

109

try:

110

all_clients[path] = ifs_and_props[client_dbus_interface]

111

except KeyError:

112

pass

113

114

# Compile dict of (clientpath: properties) to process

109

mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)

110

111

# Filter out log message from dbus module

112

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

113

class NullFilter(logging.Filter):

114

def filter(self, record):

115

return False

116

dbus_filter = NullFilter()

117

try:

118

dbus_logger.addFilter(dbus_filter)

119

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

120

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

121

mandos_clients = {path: ifs_and_props[client_dbus_interface]

122

for path, ifs_and_props in

123

mandos_serv_object_manager

124

.GetManagedObjects().items()

125

if client_dbus_interface in ifs_and_props}

126

except dbus.exceptions.DBusException as e:

127

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

128

"\n%s", e)

129

sys.exit(1)

130

finally:

131

# restore dbus logger

132

dbus_logger.removeFilter(dbus_filter)

133

134

# Compile dict of (clients: properties) to process

135

clients = {}

136

115

137

if not clientnames:

116

clients = all_clients

138

clients = {objpath: properties

139

for objpath, properties in mandos_clients.items()}

117

140

else:

118

clients = {}

119

141

for name in clientnames:

120

for objpath, properties in all_clients.items():

142

for objpath, properties in mandos_clients.items():

121

143

if properties["Name"] == name:

122

144

clients[objpath] = properties

123

145

break

424

446

options.remove = True

425

447

426

448

427

def get_mandos_dbus_object(bus):

428

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

429

dbus_busname, server_dbus_path)

430

with if_dbus_exception_log_with_exception_and_exit(

431

"Could not connect to Mandos server: %s"):

432

mandos_dbus_object = bus.get_object(dbus_busname,

433

server_dbus_path)

434

return mandos_dbus_object

435

436

437

@contextlib.contextmanager

438

def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):

439

try:

440

yield

441

except dbus.exceptions.DBusException as e:

442

log.critical(*(args + (e,)), **kwargs)

443

sys.exit(1)

444

445

446

def get_managed_objects(object_manager):

447

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

448

server_dbus_path, dbus.OBJECT_MANAGER_IFACE)

449

with if_dbus_exception_log_with_exception_and_exit(

450

"Failed to access Mandos server through D-Bus:\n%s"):

451

with SilenceLogger("dbus.proxies"):

452

managed_objects = object_manager.GetManagedObjects()

453

return managed_objects

454

455

456

class SilenceLogger(object):

457

"Simple context manager to silence a particular logger"

458

def __init__(self, loggername):

459

self.logger = logging.getLogger(loggername)

460

461

def __enter__(self):

462

self.logger.addFilter(self.nullfilter)

463

return self

464

465

class NullFilter(logging.Filter):

466

def filter(self, record):

467

return False

468

469

nullfilter = NullFilter()

470

471

def __exit__(self, exc_type, exc_val, exc_tb):

472

self.logger.removeFilter(self.nullfilter)

473

474

475

449

def commands_from_options(options):

476

450

477

451

commands = []

594

568

self.mandos.RemoveClient(client.__dbus_object_path__)

595

569

596

570

597

class OutputCmd(Command):

598

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

571

class PrintCmd(Command):

572

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

599

573

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

600

574

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

601

575

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

603

577

"LastApprovalRequest", "ApprovalDelay",

604

578

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

605

579

"Expires", "LastCheckerStatus")

606

607

580

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

608

581

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

609

610

582

def output(self, clients):

611

583

raise NotImplementedError()

612

584

613

585

614

class DumpJSONCmd(OutputCmd):

586

class DumpJSONCmd(PrintCmd):

615

587

def output(self, clients):

616

588

data = {client["Name"]:

617

589

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

618

590

for key in self.all_keywords}

619

for client in clients}

591

for client in clients.values()}

620

592

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

621

622

593

@staticmethod

623

594

def dbus_boolean_to_bool(value):

624

595

if isinstance(value, dbus.Boolean):

626

597

return value

627

598

628

599

629

class PrintTableCmd(OutputCmd):

600

class PrintTableCmd(PrintCmd):

630

601

def __init__(self, verbose=False):

631

602

self.verbose = verbose

632

603

662

633

"LastCheckerStatus": "Last Checker Status",

663

634

}

664

635

665

def __init__(self, clients, keywords):

636

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

666

637

self.clients = clients

667

638

self.keywords = keywords

639

if tableheaders is not None:

640

self.tableheaders = tableheaders

668

641

669

642

def __str__(self):

670

643

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

693

666

def string_from_client(self, client, key):

694

667

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

695

668

696

@classmethod

697

def valuetostring(cls, value, keyword):

669

@staticmethod

670

def valuetostring(value, keyword):

698

671

if isinstance(value, dbus.Boolean):

699

672

return "Yes" if value else "No"

700

673

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

701

674

"ApprovalDuration", "ExtendedTimeout"):

702

return cls.milliseconds_to_string(value)

675

return milliseconds_to_string(value)

703

676

return str(value)

704

677

705

678

def header_line(self, format_string):

710

683

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

711

684

for key in self.keywords})

712

685

713

@staticmethod

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

686

687

def milliseconds_to_string(ms):

688

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

689

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

690

.format(days="{}T".format(td.days) if td.days else "",

691

hours=td.seconds // 3600,

692

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

693

seconds=td.seconds % 60))

722

694

723

695

724

696

class PropertyCmd(Command):

725

697

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

726

727

698

def run_on_one_client(self, client, properties):

728

699

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

729

700

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

735

706

client.Set(client_dbus_interface, self.propname,

736

707

self.value_to_set,

737

708

dbus_interface=dbus.PROPERTIES_IFACE)

738

739

709

@property

740

710

def propname(self):

741

711

raise NotImplementedError()

792

762

793

763

class SetSecretCmd(PropertyValueCmd):

794

764

propname = "Secret"

795

796

765

@property

797

766

def value_to_set(self):

798

767

return self._vts

799

800

768

@value_to_set.setter

801

769

def value_to_set(self, value):

802

770

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

807

775

class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):

808

776

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

809

777

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

810

811

778

@property

812

779

def value_to_set(self):

813

780

return self._vts

814

815

781

@value_to_set.setter

816

782

def value_to_set(self, value):

817

783

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

839

805

840

806

841

807

842

class TestCaseWithAssertLogs(unittest.TestCase):

843

"""unittest.TestCase.assertLogs only exists in Python 3.4"""

844

845

if not hasattr(unittest.TestCase, "assertLogs"):

846

@contextlib.contextmanager

847

def assertLogs(self, logger, level=logging.INFO):

848

capturing_handler = self.CapturingLevelHandler(level)

849

old_level = logger.level

850

old_propagate = logger.propagate

851

logger.addHandler(capturing_handler)

852

logger.setLevel(level)

853

logger.propagate = False

854

try:

855

yield capturing_handler.watcher

856

finally:

857

logger.propagate = old_propagate

858

logger.removeHandler(capturing_handler)

859

logger.setLevel(old_level)

860

self.assertGreater(len(capturing_handler.watcher.records),

861

0)

862

863

class CapturingLevelHandler(logging.Handler):

864

def __init__(self, level, *args, **kwargs):

865

logging.Handler.__init__(self, *args, **kwargs)

866

self.watcher = self.LoggingWatcher([], [])

867

def emit(self, record):

868

self.watcher.records.append(record)

869

self.watcher.output.append(self.format(record))

870

871

LoggingWatcher = collections.namedtuple("LoggingWatcher",

872

("records",

873

"output"))

874

875

class Test_string_to_delta(TestCaseWithAssertLogs):

808

class Test_string_to_delta(unittest.TestCase):

876

809

def test_handles_basic_rfc3339(self):

877

810

self.assertEqual(string_to_delta("PT0S"),

878

811

datetime.timedelta())

882

815

datetime.timedelta(0, 1))

883

816

self.assertEqual(string_to_delta("PT2H"),

884

817

datetime.timedelta(0, 7200))

885

886

818

def test_falls_back_to_pre_1_6_1_with_warning(self):

887

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

888

value = string_to_delta("2h")

819

# assertLogs only exists in Python 3.4

820

if hasattr(self, "assertLogs"):

821

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

822

value = string_to_delta("2h")

823

else:

824

class WarningFilter(logging.Filter):

825

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

826

def filter(self, record):

827

is_warning = record.levelno >= logging.WARNING

828

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

829

False)

830

return not is_warning

831

warning_filter = WarningFilter()

832

log.addFilter(warning_filter)

833

try:

834

value = string_to_delta("2h")

835

finally:

836

log.removeFilter(warning_filter)

837

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

889

838

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

890

839

891

840

892

841

class Test_check_option_syntax(unittest.TestCase):

893

def setUp(self):

894

self.parser = argparse.ArgumentParser()

895

add_command_line_options(self.parser)

896

897

def test_actions_requires_client_or_all(self):

898

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

899

options = self.parser.parse_args()

900

setattr(options, action, value)

901

with self.assertParseError():

902

self.check_option_syntax(options)

903

904

842

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

905

843

# check_option_syntax()

906

844

actions = {

927

865

"deny": True,

928

866

}

929

867

868

def setUp(self):

869

self.parser = argparse.ArgumentParser()

870

add_command_line_options(self.parser)

871

930

872

@contextlib.contextmanager

931

873

def assertParseError(self):

932

874

with self.assertRaises(SystemExit) as e:

954

896

def check_option_syntax(self, options):

955

897

check_option_syntax(self.parser, options)

956

898

957

def test_actions_all_conflicts_with_verbose(self):

899

def test_actions_requires_client_or_all(self):

958

900

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

959

901

options = self.parser.parse_args()

960

902

setattr(options, action, value)

961

options.all = True

962

options.verbose = True

963

903

with self.assertParseError():

964

904

self.check_option_syntax(options)

965

905

966

def test_actions_with_client_conflicts_with_verbose(self):

906

def test_actions_conflicts_with_verbose(self):

967

907

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

968

908

options = self.parser.parse_args()

969

909

setattr(options, action, value)

970

910

options.verbose = True

971

options.client = ["foo"]

972

911

with self.assertParseError():

973

912

self.check_option_syntax(options)

974

913

1000

939

options.all = True

1001

940

self.check_option_syntax(options)

1002

941

1003

def test_any_action_is_ok_with_one_client(self):

1004

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

1005

options = self.parser.parse_args()

1006

setattr(options, action, value)

1007

options.client = ["foo"]

1008

self.check_option_syntax(options)

1009

1010

def test_actions_except_is_enabled_are_ok_with_two_clients(self):

1011

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

1012

if action == "is_enabled":

1013

continue

1014

options = self.parser.parse_args()

1015

setattr(options, action, value)

1016

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

1017

self.check_option_syntax(options)

1018

1019

942

def test_is_enabled_fails_without_client(self):

1020

943

options = self.parser.parse_args()

1021

944

options.is_enabled = True

1022

945

with self.assertParseError():

1023

946

self.check_option_syntax(options)

1024

947

948

def test_is_enabled_works_with_one_client(self):

949

options = self.parser.parse_args()

950

options.is_enabled = True

951

options.client = ["foo"]

952

self.check_option_syntax(options)

953

1025

954

def test_is_enabled_fails_with_two_clients(self):

1026

955

options = self.parser.parse_args()

1027

956

options.is_enabled = True

1041

970

self.check_option_syntax(options)

1042

971

1043

972

1044

class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):

1045

def test_calls_and_returns_get_object_on_bus(self):

1046

class MockBus(object):

1047

called = False

1048

def get_object(mockbus_self, busname, dbus_path):

1049

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

1050

# this MockBus instance is in "mockbus_self".

1051

self.assertEqual(busname, dbus_busname)

1052

self.assertEqual(dbus_path, server_dbus_path)

1053

mockbus_self.called = True

1054

return mockbus_self

1055

1056

mockbus = get_mandos_dbus_object(bus=MockBus())

1057

self.assertIsInstance(mockbus, MockBus)

1058

self.assertTrue(mockbus.called)

1059

1060

def test_logs_and_exits_on_dbus_error(self):

1061

class MockBusFailing(object):

1062

def get_object(self, busname, dbus_path):

1063

raise dbus.exceptions.DBusException("Test")

1064

1065

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

1066

with self.assertRaises(SystemExit) as e:

1067

bus = get_mandos_dbus_object(bus=MockBusFailing())

1068

1069

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

1070

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

1071

else:

1072

self.assertIsNotNone(e.exception.code)

1073

1074

1075

class Test_get_managed_objects(TestCaseWithAssertLogs):

1076

def test_calls_and_returns_GetManagedObjects(self):

1077

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

1078

class MockObjectManager(object):

1079

def GetManagedObjects(self):

1080

return managed_objects

1081

retval = get_managed_objects(MockObjectManager())

1082

self.assertDictEqual(managed_objects, retval)

1083

1084

def test_logs_and_exits_on_dbus_error(self):

1085

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

1086

1087

class MockObjectManagerFailing(object):

1088

def GetManagedObjects(self):

1089

dbus_logger.error("Test")

1090

raise dbus.exceptions.DBusException("Test")

1091

1092

class CountingHandler(logging.Handler):

1093

count = 0

1094

def emit(self, record):

1095

self.count += 1

1096

1097

counting_handler = CountingHandler()

1098

1099

dbus_logger.addHandler(counting_handler)

1100

1101

try:

1102

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

1103

with self.assertRaises(SystemExit) as e:

1104

get_managed_objects(MockObjectManagerFailing())

1105

finally:

1106

dbus_logger.removeFilter(counting_handler)

1107

1108

# Make sure the dbus logger was suppressed

1109

self.assertEqual(counting_handler.count, 0)

1110

1111

# Test that the dbus_logger still works

1112

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

1113

dbus_logger.error("Test")

1114

1115

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

1116

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

1117

else:

1118

self.assertIsNotNone(e.exception.code)

1119

1120

1121

class Test_commands_from_options(unittest.TestCase):

973

class Test_command_from_options(unittest.TestCase):

1122

974

def setUp(self):

1123

975

self.parser = argparse.ArgumentParser()

1124

976

add_command_line_options(self.parser)

1125

1126

def test_is_enabled(self):

1127

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

1128

IsEnabledCmd)

1129

1130

977

def assert_command_from_args(self, args, command_cls,

1131

978

**cmd_attrs):

1132

979

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

1139

986

self.assertIsInstance(command, command_cls)

1140

987

for key, value in cmd_attrs.items():

1141

988

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

1142

1143

def test_is_enabled_short(self):

1144

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

1145

1146

def test_approve(self):

1147

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

1148

ApproveCmd)

1149

1150

def test_approve_short(self):

1151

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

1152

1153

def test_deny(self):

1154

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

1155

1156

def test_deny_short(self):

1157

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

1158

1159

def test_remove(self):

1160

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

1161

RemoveCmd)

1162

1163

def test_deny_before_remove(self):

1164

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

1165

"foo"])

1166

check_option_syntax(self.parser, options)

1167

commands = commands_from_options(options)

1168

self.assertEqual(len(commands), 2)

1169

self.assertIsInstance(commands[0], DenyCmd)

1170

self.assertIsInstance(commands[1], RemoveCmd)

1171

1172

def test_deny_before_remove_reversed(self):

1173

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

1174

"--all"])

1175

check_option_syntax(self.parser, options)

1176

commands = commands_from_options(options)

1177

self.assertEqual(len(commands), 2)

1178

self.assertIsInstance(commands[0], DenyCmd)

1179

self.assertIsInstance(commands[1], RemoveCmd)

1180

1181

def test_remove_short(self):

1182

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

1183

1184

def test_dump_json(self):

1185

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

989

def test_print_table(self):

990

self.assert_command_from_args([], PrintTableCmd,

991

verbose=False)

992

993

def test_print_table_verbose(self):

994

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

995

verbose=True)

996

997

def test_print_table_verbose_short(self):

998

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

999

verbose=True)

1186

1000

1187

1001

def test_enable(self):

1188

1002

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

1212

1026

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

1213

1027

StopCheckerCmd)

1214

1028

1215

def test_approve_by_default(self):

1216

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

1217

ApproveByDefaultCmd)

1029

def test_remove(self):

1030

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

1031

RemoveCmd)

1218

1032

1219

def test_deny_by_default(self):

1220

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

1221

DenyByDefaultCmd)

1033

def test_remove_short(self):

1034

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

1222

1035

1223

1036

def test_checker(self):

1224

1037

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

1232

1045

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

1233

1046

SetCheckerCmd, value_to_set=":")

1234

1047

1235

def test_host(self):

1236

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

1237

"foo"], SetHostCmd,

1238

value_to_set="foo.example.org")

1239

1240

def test_host_short(self):

1241

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

1242

"foo"], SetHostCmd,

1243

value_to_set="foo.example.org")

1244

1245

def test_secret_devnull(self):

1246

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

1247

"foo"], SetSecretCmd,

1248

value_to_set=b"")

1249

1250

def test_secret_tempfile(self):

1251

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

1252

value = b"secret\0xyzzy\nbar"

1253

f.write(value)

1254

f.seek(0)

1255

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

1256

"foo"], SetSecretCmd,

1257

value_to_set=value)

1258

1259

def test_secret_devnull_short(self):

1260

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

1261

SetSecretCmd, value_to_set=b"")

1262

1263

def test_secret_tempfile_short(self):

1264

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

1265

value = b"secret\0xyzzy\nbar"

1266

f.write(value)

1267

f.seek(0)

1268

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

1269

SetSecretCmd,

1270

value_to_set=value)

1271

1272

1048

def test_timeout(self):

1273

1049

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

1274

1050

SetTimeoutCmd,

1295

1071

SetIntervalCmd,

1296

1072

value_to_set=120000)

1297

1073

1074

def test_approve_by_default(self):

1075

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

1076

ApproveByDefaultCmd)

1077

1078

def test_deny_by_default(self):

1079

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

1080

DenyByDefaultCmd)

1081

1298

1082

def test_approval_delay(self):

1299

1083

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

1300

1084

"foo"], SetApprovalDelayCmd,

1305

1089

"foo"], SetApprovalDurationCmd,

1306

1090

value_to_set=1000)

1307

1091

1308

def test_print_table(self):

1309

self.assert_command_from_args([], PrintTableCmd,

1310

verbose=False)

1311

1312

def test_print_table_verbose(self):

1313

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

1314

verbose=True)

1315

1316

def test_print_table_verbose_short(self):

1317

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

1318

verbose=True)

1092

def test_host(self):

1093

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

1094

"foo"], SetHostCmd,

1095

value_to_set="foo.example.org")

1096

1097

def test_host_short(self):

1098

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

1099

"foo"], SetHostCmd,

1100

value_to_set="foo.example.org")

1101

1102

def test_secret_devnull(self):

1103

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

1104

"foo"], SetSecretCmd,

1105

value_to_set=b"")

1106

1107

def test_secret_tempfile(self):

1108

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

1109

value = b"secret\0xyzzy\nbar"

1110

f.write(value)

1111

f.seek(0)

1112

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

1113

"foo"], SetSecretCmd,

1114

value_to_set=value)

1115

1116

def test_secret_devnull_short(self):

1117

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

1118

SetSecretCmd, value_to_set=b"")

1119

1120

def test_secret_tempfile_short(self):

1121

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

1122

value = b"secret\0xyzzy\nbar"

1123

f.write(value)

1124

f.seek(0)

1125

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

1126

SetSecretCmd,

1127

value_to_set=value)

1128

1129

def test_approve(self):

1130

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

1131

ApproveCmd)

1132

1133

def test_approve_short(self):

1134

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

1135

1136

def test_deny(self):

1137

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

1138

1139

def test_deny_short(self):

1140

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

1141

1142

def test_dump_json(self):

1143

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

1144

1145

def test_is_enabled(self):

1146

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

1147

IsEnabledCmd)

1148

1149

def test_is_enabled_short(self):

1150

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

1151

1152

def test_deny_before_remove(self):

1153

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

1154

"foo"])

1155

check_option_syntax(self.parser, options)

1156

commands = commands_from_options(options)

1157

self.assertEqual(len(commands), 2)

1158

self.assertIsInstance(commands[0], DenyCmd)

1159

self.assertIsInstance(commands[1], RemoveCmd)

1160

1161

def test_deny_before_remove_reversed(self):

1162

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

1163

"--all"])

1164

check_option_syntax(self.parser, options)

1165

commands = commands_from_options(options)

1166

self.assertEqual(len(commands), 2)

1167

self.assertIsInstance(commands[0], DenyCmd)

1168

self.assertIsInstance(commands[1], RemoveCmd)

1319

1169

1320

1170

1321

1171

class TestCmd(unittest.TestCase):

1322

1172

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

1323

1324

1173

def setUp(self):

1325

1174

testcase = self

1326

1175

class MockClient(object):

1387

1236

ApprovedByDefault=dbus.Boolean(False),

1388

1237

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

1389

1238

ApprovalDelay=30000,

1390

ApprovalDuration=93785000,

1239

ApprovalDuration=1000,

1391

1240

Checker=":",

1392

1241

ExtendedTimeout=900000,

1393

1242

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

1398

1247

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

1399

1248

])

1400

1249

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

1401

1402

1250

@property

1403

1251

def bus(self):

1404

1252

class Bus(object):

1406

1254

def get_object(client_bus_name, path):

1407

1255

self.assertEqual(client_bus_name, dbus_busname)

1408

1256

return {

1409

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

1410

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

1411

1257

"/clients/foo": self.client,

1412

1258

"/clients/barbar": self.other_client,

1413

1259

}[path]

1420

1266

properties)

1421

1267

for client, properties

1422

1268

in self.clients.items()))

1423

1424

1269

def test_is_enabled_run_exits_successfully(self):

1425

1270

with self.assertRaises(SystemExit) as e:

1426

1271

IsEnabledCmd().run(self.one_client)

1428

1273

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

1429

1274

else:

1430

1275

self.assertIsNone(e.exception.code)

1431

1432

1276

def test_is_enabled_run_exits_with_failure(self):

1433

1277

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

1434

1278

with self.assertRaises(SystemExit) as e:

1456

1300

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

1457

1301

client.calls)

1458

1302

1459

1460

1303

class TestRemoveCmd(TestCmd):

1461

1304

def test_remove(self):

1462

1305

class MockMandos(object):

1518

1361

"ApprovedByDefault": False,

1519

1362

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

1520

1363

"ApprovalDelay": 30000,

1521

"ApprovalDuration": 93785000,

1364

"ApprovalDuration": 1000,

1522

1365

"Checker": ":",

1523

1366

"ExtendedTimeout": 900000,

1524

1367

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

1526

1369

},

1527

1370

}

1528

1371

return super(TestDumpJSONCmd, self).setUp()

1529

1530

1372

def test_normal(self):

1531

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

1532

json_data = json.loads(output)

1373

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

1533

1374

self.assertDictEqual(json_data, self.expected_json)

1534

1535

1375

def test_one_client(self):

1536

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

1537

json_data = json.loads(output)

1376

clients = self.one_client

1377

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

1538

1378

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

1539

1379

self.assertDictEqual(json_data, expected_json)

1540

1380

1548

1388

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

1549

1389

))

1550

1390

self.assertEqual(output, expected_output)

1551

1552

1391

def test_verbose(self):

1553

1392

output = PrintTableCmd(verbose=True).output(

1554

1393

self.clients.values())

1619

1458

),(

1620

1459

"Approval Duration ",

1621

1460

"00:00:01 ",

1622

"1T02:03:05 ",

1461

"00:00:01 ",

1623

1462

),(

1624

1463

"Checker ",

1625

1464

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

1643

1482

for rows in columns)

1644

1483

for line in range(num_lines))

1645

1484

self.assertEqual(output, expected_output)

1646

1647

1485

def test_one_client(self):

1648

1486

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

1649

expected_output = "\n".join((

1650

"Name Enabled Timeout Last Successful Check",

1651

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

1652

))

1487

expected_output = """

1488

Name Enabled Timeout Last Successful Check

1489

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

1490

"""[1:-1]

1653

1491

self.assertEqual(output, expected_output)

1654

1492

1655

1493

1494

class Test_milliseconds_to_string(unittest.TestCase):

1495

def test_all(self):

1496

self.assertEqual(milliseconds_to_string(93785000),

1497

"1T02:03:05")

1498

def test_no_days(self):

1499

self.assertEqual(milliseconds_to_string(7385000), "02:03:05")

1500

def test_all_zero(self):

1501

self.assertEqual(milliseconds_to_string(0), "00:00:00")

1502

def test_no_fractional_seconds(self):

1503

self.assertEqual(milliseconds_to_string(400), "00:00:00")

1504

self.assertEqual(milliseconds_to_string(900), "00:00:00")

1505

self.assertEqual(milliseconds_to_string(1900), "00:00:01")

1506

1507

1508

class Unique(object):

1509

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

1510

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

1511

1512

1656

1513

class TestPropertyCmd(TestCmd):

1657

1514

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

1658

1515

def runTest(self):

1665

1522

for clientpath in self.clients:

1666

1523

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

1667

1524

old_value = client.attributes[self.propname]

1668

self.assertNotIsInstance(old_value, self.Unique)

1669

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

1525

self.assertNotIsInstance(old_value, Unique)

1526

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

1670

1527

self.run_command(value_to_set, self.clients)

1671

1528

for clientpath in self.clients:

1672

1529

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

1673

1530

value = client.attributes[self.propname]

1674

self.assertNotIsInstance(value, self.Unique)

1531

self.assertNotIsInstance(value, Unique)

1675

1532

self.assertEqual(value, value_to_get)

1676

1677

class Unique(object):

1678

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

1679

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

1680

1681

1533

def run_command(self, value, clients):

1682

1534

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

1683

1535

1684

1536

1685

class TestEnableCmd(TestPropertyCmd):

1686

command = EnableCmd

1687

propname = "Enabled"

1688

values_to_set = [dbus.Boolean(True)]

1689

1690

1691

class TestDisableCmd(TestPropertyCmd):

1692

command = DisableCmd

1693

propname = "Enabled"

1694

values_to_set = [dbus.Boolean(False)]

1537

class TestEnableCmd(TestCmd):

1538

def test_enable(self):

1539

for clientpath in self.clients:

1540

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

1541

client.attributes["Enabled"] = False

1542

1543

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

1544

1545

for clientpath in self.clients:

1546

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

1547

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

1548

1549

1550

class TestDisableCmd(TestCmd):

1551

def test_disable(self):

1552

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

1553

for clientpath in self.clients:

1554

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

1555

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

1695

1556

1696

1557

1697

1558

class TestBumpTimeoutCmd(TestPropertyCmd):

1726

1587

1727

1588

class TestPropertyValueCmd(TestPropertyCmd):

1728

1589

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

1729

1730

1590

def runTest(self):

1731

1591

if type(self) is TestPropertyValueCmd:

1732

1592

return

1733

1593

return super(TestPropertyValueCmd, self).runTest()

1734

1735

1594

def run_command(self, value, clients):

1736

1595

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

1737

1596

1829

1688

return tests

1830

1689

1831

1690

if __name__ == "__main__":

1832

try:

1833

if should_only_run_tests():

1834

# Call using ./tdd-python-script --check [--verbose]

1835

unittest.main()

1836

else:

1837

main()

1838

finally:

1839

logging.shutdown()

1691

if should_only_run_tests():

1692

# Call using ./tdd-python-script --check [--verbose]

1693

unittest.main()

1694

else:

1695

main()