/mandos/trunk

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-09 15:47:52 UTC
  • mto: This revision was merged to the branch mainline in revision 1046.
  • Revision ID: teddy@recompile.se-20190309154752-hdp3hcbagkvqx91f
mandos-ctl: Refactor

* mandos-ctl (Command.run): Take additional "mandos_objmgr" parameter.
                            All callers changed.
  (PrintTableCmd.output, DumpJSONCmd.output): Take additional
         "mandos_objmgr" parameter, and "clients" is now a sequence of
         attribute dicts.  All callers changed.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
276
276
# Abstract classes first
277
277
class Command(object):
278
278
    """Abstract class for commands"""
279
 
    def run(self, mandos, clients):
 
279
    def run(self, mandos, mandos_objmgr, clients):
280
280
        """Normal commands should implement run_on_one_client(), but
281
281
        commands which want to operate on all clients at the same time
282
282
        can override this run() method instead."""
293
293
                    "LastApprovalRequest", "ApprovalDelay",
294
294
                    "ApprovalDuration", "Checker", "ExtendedTimeout",
295
295
                    "Expires", "LastCheckerStatus")
296
 
    def run(self, mandos, clients):
297
 
        print(self.output(clients))
 
296
    def run(self, mandos, mandos_objmgr, clients):
 
297
        print(self.output(mandos_objmgr, clients.values()))
298
298
 
299
299
class PropertyCmd(Command):
300
300
    """Abstract class for Actions for setting one client property"""
331
331
    def __init__(self, verbose=False):
332
332
        self.verbose = verbose
333
333
 
334
 
    def output(self, clients):
 
334
    def output(self, mandos_objmgr, clients):
335
335
        default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
336
336
        keywords = default_keywords
337
337
        if self.verbose:
338
338
            keywords = self.all_keywords
339
 
        return str(self.TableOfClients(clients.values(), keywords))
 
339
        return str(self.TableOfClients(clients, keywords))
340
340
 
341
341
    class TableOfClients(object):
342
342
        tableheaders = {
415
415
 
416
416
 
417
417
class DumpJSONCmd(PrintCmd):
418
 
    def output(self, clients):
 
418
    def output(self, mandos_objmgr, clients):
419
419
        data = {client["Name"]:
420
420
                {key: self.dbus_boolean_to_bool(client[key])
421
421
                 for key in self.all_keywords}
433
433
            sys.exit(0)
434
434
        sys.exit(1)
435
435
    def is_enabled(self, client, properties):
436
 
        return bool(properties["Enabled"])
 
436
        log.debug("D-Bus: %s:%s:%s.Get(%r, %r)", busname,
 
437
                  client.__dbus_object_path__,
 
438
                  dbus.PROPERTIES_IFACE, client_interface,
 
439
                  "Enabled")
 
440
        return bool(client.Get(client_interface, "Enabled",
 
441
                               dbus_interface=dbus.PROPERTIES_IFACE))
437
442
 
438
443
class RemoveCmd(Command):
439
444
    def run_on_one_client(self, client, properties):
773
778
    # Run all commands on clients
774
779
    commands = commands_from_options(options)
775
780
    for command in commands:
776
 
        command.run(mandos_serv, clients)
 
781
        command.run(mandos_serv, mandos_serv_object_manager, clients)
777
782
 
778
783
 
779
784
class Test_milliseconds_to_string(unittest.TestCase):
903
908
 
904
909
class TestPrintTableCmd(TestCmd):
905
910
    def test_normal(self):
906
 
        output = PrintTableCmd().output(self.clients)
 
911
        output = PrintTableCmd().output(None, self.clients.values())
907
912
        expected_output = """
908
913
Name   Enabled Timeout  Last Successful Check
909
914
foo    Yes     00:05:00 2019-02-03T00:00:00  
911
916
"""[1:-1]
912
917
        self.assertEqual(output, expected_output)
913
918
    def test_verbose(self):
914
 
        output = PrintTableCmd(verbose=True).output(self.clients)
 
919
        output = PrintTableCmd(verbose=True).output(
 
920
            None, self.clients.values())
915
921
        expected_output = """
916
922
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
917
923
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                  
919
925
"""[1:-1]
920
926
        self.assertEqual(output, expected_output)
921
927
    def test_one_client(self):
922
 
        output = PrintTableCmd().output(self.one_client)
 
928
        output = PrintTableCmd().output(None, self.one_client.values())
923
929
        expected_output = """
924
930
Name Enabled Timeout  Last Successful Check
925
931
foo  Yes     00:05:00 2019-02-03T00:00:00  
980
986
        }
981
987
        return super(TestDumpJSONCmd, self).setUp()
982
988
    def test_normal(self):
983
 
        json_data = json.loads(DumpJSONCmd().output(self.clients))
 
989
        json_data = json.loads(DumpJSONCmd().output(None,
 
990
                                                    self.clients))
984
991
        self.assertDictEqual(json_data, self.expected_json)
985
992
    def test_one_client(self):
986
993
        clients = self.one_client
987
 
        json_data = json.loads(DumpJSONCmd().output(clients))
 
994
        json_data = json.loads(DumpJSONCmd().output(None, clients))
988
995
        expected_json = {"foo": self.expected_json["foo"]}
989
996
        self.assertDictEqual(json_data, expected_json)
990
997
 
994
1001
                            for client, properties in self.clients.items()))
995
1002
    def test_is_enabled_run_exits_successfully(self):
996
1003
        with self.assertRaises(SystemExit) as e:
997
 
            IsEnabledCmd().run(None, self.one_client)
 
1004
            IsEnabledCmd().run(None, None, self.one_client)
998
1005
        if e.exception.code is not None:
999
1006
            self.assertEqual(e.exception.code, 0)
1000
1007
        else:
1002
1009
    def test_is_enabled_run_exits_with_failure(self):
1003
1010
        self.client.attributes["Enabled"] = dbus.Boolean(False)
1004
1011
        with self.assertRaises(SystemExit) as e:
1005
 
            IsEnabledCmd().run(None, self.one_client)
 
1012
            IsEnabledCmd().run(None, None, self.one_client)
1006
1013
        if isinstance(e.exception.code, int):
1007
1014
            self.assertNotEqual(e.exception.code, 0)
1008
1015
        else:
1017
1024
                self.calls.append(("RemoveClient", (dbus_path,)))
1018
1025
        mandos = MockMandos()
1019
1026
        super(TestRemoveCmd, self).setUp()
1020
 
        RemoveCmd().run(mandos, self.clients)
 
1027
        RemoveCmd().run(mandos, None, self.clients)
1021
1028
        self.assertEqual(len(mandos.calls), 2)
1022
1029
        for client in self.clients:
1023
1030
            self.assertIn(("RemoveClient",
1026
1033
 
1027
1034
class TestApproveCmd(TestCmd):
1028
1035
    def test_approve(self):
1029
 
        ApproveCmd().run(None, self.clients)
 
1036
        ApproveCmd().run(None, None, self.clients)
1030
1037
        for client in self.clients:
1031
1038
            self.assertIn(("Approve", (True, client_interface)),
1032
1039
                          client.calls)
1033
1040
 
1034
1041
class TestDenyCmd(TestCmd):
1035
1042
    def test_deny(self):
1036
 
        DenyCmd().run(None, self.clients)
 
1043
        DenyCmd().run(None, None, self.clients)
1037
1044
        for client in self.clients:
1038
1045
            self.assertIn(("Approve", (False, client_interface)),
1039
1046
                          client.calls)
1043
1050
        for client in self.clients:
1044
1051
            client.attributes["Enabled"] = False
1045
1052
 
1046
 
        EnableCmd().run(None, self.clients)
 
1053
        EnableCmd().run(None, None, self.clients)
1047
1054
 
1048
1055
        for client in self.clients:
1049
1056
            self.assertTrue(client.attributes["Enabled"])
1050
1057
 
1051
1058
class TestDisableCmd(TestCmd):
1052
1059
    def test_disable(self):
1053
 
        DisableCmd().run(None, self.clients)
 
1060
        DisableCmd().run(None, None, self.clients)
1054
1061
 
1055
1062
        for client in self.clients:
1056
1063
            self.assertFalse(client.attributes["Enabled"])
1078
1085
                self.assertNotIsInstance(value, Unique)
1079
1086
                self.assertEqual(value, value_to_get)
1080
1087
    def run_command(self, value, clients):
1081
 
        self.command().run(None, clients)
 
1088
        self.command().run(None, None, clients)
1082
1089
 
1083
1090
class TestBumpTimeoutCmd(TestPropertyCmd):
1084
1091
    command = BumpTimeoutCmd
1113
1120
            return
1114
1121
        return super(TestValueArgumentPropertyCmd, self).runTest()
1115
1122
    def run_command(self, value, clients):
1116
 
        self.command(value).run(None, clients)
 
1123
        self.command(value).run(None, None, clients)
1117
1124
 
1118
1125
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1119
1126
    command = SetCheckerCmd