/mandos/release

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-09-04 21:17:42 UTC
  • Revision ID: teddy@recompile.se-20190904211742-9luzszv5yt5h91pw
Tags: version-1.8.9-2
* debian/changelog (1.8.9-2): Add new entry.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
 
#!/usr/bin/python
2
 
# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-
 
1
#!/usr/bin/python3 -bb
 
2
# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
46
46
import tempfile
47
47
import contextlib
48
48
 
 
49
if sys.version_info.major == 2:
 
50
    __metaclass__ = type
 
51
 
49
52
try:
50
53
    import pydbus
51
54
    import gi
53
56
except ImportError:
54
57
    import dbus as dbus_python
55
58
    pydbus = None
56
 
    class gi(object):
 
59
    class gi:
57
60
        """Dummy gi module, for the tests"""
58
 
        class repository(object):
59
 
            class GLib(object):
 
61
        class repository:
 
62
            class GLib:
60
63
                class Error(Exception):
61
64
                    pass
62
65
 
78
81
 
79
82
locale.setlocale(locale.LC_ALL, "")
80
83
 
81
 
version = "1.8.3"
 
84
version = "1.8.9"
82
85
 
83
86
 
84
87
def main():
250
253
def rfc3339_duration_to_delta(duration):
251
254
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
252
255
 
253
 
    >>> rfc3339_duration_to_delta("P7D")
254
 
    datetime.timedelta(7)
255
 
    >>> rfc3339_duration_to_delta("PT60S")
256
 
    datetime.timedelta(0, 60)
257
 
    >>> rfc3339_duration_to_delta("PT60M")
258
 
    datetime.timedelta(0, 3600)
259
 
    >>> rfc3339_duration_to_delta("P60M")
260
 
    datetime.timedelta(1680)
261
 
    >>> rfc3339_duration_to_delta("PT24H")
262
 
    datetime.timedelta(1)
263
 
    >>> rfc3339_duration_to_delta("P1W")
264
 
    datetime.timedelta(7)
265
 
    >>> rfc3339_duration_to_delta("PT5M30S")
266
 
    datetime.timedelta(0, 330)
267
 
    >>> rfc3339_duration_to_delta("P1DT3M20S")
268
 
    datetime.timedelta(1, 200)
 
256
    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
 
257
    True
 
258
    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
 
259
    True
 
260
    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
 
261
    True
 
262
    >>> # 60 months
 
263
    >>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
 
264
    True
 
265
    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
 
266
    True
 
267
    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
 
268
    True
 
269
    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
 
270
    True
 
271
    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
 
272
    True
269
273
    >>> # Can not be empty:
270
274
    >>> rfc3339_duration_to_delta("")
271
275
    Traceback (most recent call last):
381
385
    """Parse an interval string as documented by Mandos before 1.6.1,
382
386
    and return a datetime.timedelta
383
387
 
384
 
    >>> parse_pre_1_6_1_interval('7d')
385
 
    datetime.timedelta(7)
386
 
    >>> parse_pre_1_6_1_interval('60s')
387
 
    datetime.timedelta(0, 60)
388
 
    >>> parse_pre_1_6_1_interval('60m')
389
 
    datetime.timedelta(0, 3600)
390
 
    >>> parse_pre_1_6_1_interval('24h')
391
 
    datetime.timedelta(1)
392
 
    >>> parse_pre_1_6_1_interval('1w')
393
 
    datetime.timedelta(7)
394
 
    >>> parse_pre_1_6_1_interval('5m 30s')
395
 
    datetime.timedelta(0, 330)
396
 
    >>> parse_pre_1_6_1_interval('')
397
 
    datetime.timedelta(0)
 
388
    >>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
 
389
    True
 
390
    >>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
 
391
    True
 
392
    >>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
 
393
    True
 
394
    >>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
 
395
    True
 
396
    >>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
 
397
    True
 
398
    >>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
 
399
    True
 
400
    >>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
 
401
    True
398
402
    >>> # Ignore unknown characters, allow any order and repetitions
399
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
 
    datetime.timedelta(2, 480, 18000)
 
403
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
 
404
    True
401
405
 
402
406
    """
403
407
 
466
470
        parser.error("--remove can only be combined with --deny")
467
471
 
468
472
 
469
 
class dbus(object):
 
473
class dbus:
470
474
 
471
 
    class SystemBus(object):
 
475
    class SystemBus:
472
476
 
473
477
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
478
        def get_managed_objects(self, busname, objectpath):
520
524
        pass
521
525
 
522
526
 
523
 
class dbus_python_adapter(object):
 
527
class dbus_python_adapter:
524
528
 
525
529
    class SystemBus(dbus.MandosBus):
526
530
        """Use dbus-python"""
580
584
                                     self.client_interface, key,
581
585
                                     value)
582
586
 
583
 
    class SilenceLogger(object):
 
587
    class SilenceLogger:
584
588
        "Simple context manager to silence a particular logger"
585
589
        def __init__(self, loggername):
586
590
            self.logger = logging.getLogger(loggername)
615
619
                return new_object
616
620
 
617
621
 
618
 
class pydbus_adapter(object):
 
622
class pydbus_adapter:
619
623
    class SystemBus(dbus.MandosBus):
620
624
        def __init__(self, module=pydbus):
621
625
            self.pydbus = module
711
715
    return commands
712
716
 
713
717
 
714
 
class command(object):
 
718
class command:
715
719
    """A namespace for command classes"""
716
720
 
717
 
    class Base(object):
 
721
    class Base:
718
722
        """Abstract base class for commands"""
719
723
        def run(self, clients, bus=None):
720
724
            """Normal commands should implement run_on_one_client(),
783
787
                keywords = self.all_keywords
784
788
            print(self.TableOfClients(clients.values(), keywords))
785
789
 
786
 
        class TableOfClients(object):
 
790
        class TableOfClients:
787
791
            tableheaders = {
788
792
                "Name": "Name",
789
793
                "Enabled": "Enabled",
1020
1024
                                                     "output"))
1021
1025
 
1022
1026
 
1023
 
class Unique(object):
 
1027
class Unique:
1024
1028
    """Class for objects which exist only to be unique objects, since
1025
1029
unittest.mock.sentinel only exists in Python 3.3"""
1026
1030
 
1310
1314
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1311
1315
 
1312
1316
    def MockDBusPython_func(self, func):
1313
 
        class mock_dbus_python(object):
 
1317
        class mock_dbus_python:
1314
1318
            """mock dbus-python module"""
1315
 
            class exceptions(object):
 
1319
            class exceptions:
1316
1320
                """Pseudo-namespace"""
1317
1321
                class DBusException(Exception):
1318
1322
                    pass
1319
 
            class SystemBus(object):
 
1323
            class SystemBus:
1320
1324
                @staticmethod
1321
1325
                def get_object(busname, objectpath):
1322
1326
                    DBusObject = collections.namedtuple(
1336
1340
                                    dbus_interface=dbus_interface)
1337
1341
                    return DBusObject(methodname=method,
1338
1342
                                      Set=set_property)
1339
 
            class Boolean(object):
 
1343
            class Boolean:
1340
1344
                def __init__(self, value):
1341
1345
                    self.value = bool(value)
1342
1346
                def __bool__(self):
1554
1558
            self.call_method(bus, "methodname", "busname",
1555
1559
                             "objectpath", "interface")
1556
1560
 
1557
 
    class fake_dbus_python_raises_exception_on_connect(object):
 
1561
    class fake_dbus_python_raises_exception_on_connect:
1558
1562
        """fake dbus-python module"""
1559
 
        class exceptions(object):
 
1563
        class exceptions:
1560
1564
            """Pseudo-namespace"""
1561
1565
            class DBusException(Exception):
1562
1566
                pass
1570
1574
 
1571
1575
 
1572
1576
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1573
 
    class mock_dbus_python(object):
 
1577
    class mock_dbus_python:
1574
1578
        """mock dbus-python modules"""
1575
 
        class SystemBus(object):
 
1579
        class SystemBus:
1576
1580
            @staticmethod
1577
1581
            def get_object(busname, objectpath):
1578
1582
                return Unique()
1624
1628
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1625
1629
 
1626
1630
    def Stub_pydbus_func(self, func):
1627
 
        class stub_pydbus(object):
 
1631
        class stub_pydbus:
1628
1632
            """stub pydbus module"""
1629
 
            class SystemBus(object):
 
1633
            class SystemBus:
1630
1634
                @staticmethod
1631
1635
                def get(busname, objectpath):
1632
1636
                    DBusObject = collections.namedtuple(
1678
1682
            self.call_method(bus, "methodname", "busname",
1679
1683
                             "objectpath", "interface")
1680
1684
 
1681
 
    class fake_pydbus_raises_exception_on_connect(object):
 
1685
    class fake_pydbus_raises_exception_on_connect:
1682
1686
        """fake dbus-python module"""
1683
1687
        @classmethod
1684
1688
        def SystemBus(cls):
1688
1692
            return Bus(get=get)
1689
1693
 
1690
1694
    def test_set_property_uses_setattr(self):
1691
 
        class Object(object):
 
1695
        class Object:
1692
1696
            pass
1693
1697
        obj = Object()
1694
 
        class pydbus_spy(object):
1695
 
            class SystemBus(object):
 
1698
        class pydbus_spy:
 
1699
            class SystemBus:
1696
1700
                @staticmethod
1697
1701
                def get(busname, objectpath):
1698
1702
                    return {"interface": obj}
1705
1709
    def test_get_suppresses_xml_deprecation_warning(self):
1706
1710
        if sys.version_info.major >= 3:
1707
1711
            return
1708
 
        class stub_pydbus_get(object):
1709
 
            class SystemBus(object):
 
1712
        class stub_pydbus_get:
 
1713
            class SystemBus:
1710
1714
                @staticmethod
1711
1715
                def get(busname, objectpath):
1712
1716
                    warnings.warn_explicit(
1720
1724
 
1721
1725
 
1722
1726
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1723
 
    class stub_pydbus(object):
 
1727
    class stub_pydbus:
1724
1728
        """stub pydbus module"""
1725
 
        class SystemBus(object):
 
1729
        class SystemBus:
1726
1730
            @staticmethod
1727
1731
            def get(busname, objectpath):
1728
1732
                return Unique()
1780
1784
        self.assert_command_from_args(["--is-enabled", "client"],
1781
1785
                                      command.IsEnabled)
1782
1786
 
1783
 
    def assert_command_from_args(self, args, command_cls,
1784
 
                                 **cmd_attrs):
 
1787
    def assert_command_from_args(self, args, command_cls, length=1,
 
1788
                                 clients=None, **cmd_attrs):
1785
1789
        """Assert that parsing ARGS should result in an instance of
1786
1790
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1787
1791
        options = self.parser.parse_args(args)
1788
1792
        check_option_syntax(self.parser, options)
1789
1793
        commands = commands_from_options(options)
1790
 
        self.assertEqual(1, len(commands))
1791
 
        command = commands[0]
1792
 
        self.assertIsInstance(command, command_cls)
 
1794
        self.assertEqual(length, len(commands))
 
1795
        for command in commands:
 
1796
            if isinstance(command, command_cls):
 
1797
                break
 
1798
        else:
 
1799
            self.assertIsInstance(command, command_cls)
 
1800
        if clients is not None:
 
1801
            self.assertEqual(clients, options.client)
1793
1802
        for key, value in cmd_attrs.items():
1794
1803
            self.assertEqual(value, getattr(command, key))
1795
1804
 
 
1805
    def assert_commands_from_args(self, args, commands, clients=None):
 
1806
        for cmd in commands:
 
1807
            self.assert_command_from_args(args, cmd,
 
1808
                                          length=len(commands),
 
1809
                                          clients=clients)
 
1810
 
1796
1811
    def test_is_enabled_short(self):
1797
1812
        self.assert_command_from_args(["-V", "client"],
1798
1813
                                      command.IsEnabled)
1989
2004
                                      verbose=True)
1990
2005
 
1991
2006
 
 
2007
    def test_manual_page_example_1(self):
 
2008
        self.assert_command_from_args("",
 
2009
                                      command.PrintTable,
 
2010
                                      clients=[],
 
2011
                                      verbose=False)
 
2012
 
 
2013
    def test_manual_page_example_2(self):
 
2014
        self.assert_command_from_args(
 
2015
            "--verbose foo1.example.org foo2.example.org".split(),
 
2016
            command.PrintTable, clients=["foo1.example.org",
 
2017
                                         "foo2.example.org"],
 
2018
            verbose=True)
 
2019
 
 
2020
    def test_manual_page_example_3(self):
 
2021
        self.assert_command_from_args("--enable --all".split(),
 
2022
                                      command.Enable,
 
2023
                                      clients=[])
 
2024
 
 
2025
    def test_manual_page_example_4(self):
 
2026
        self.assert_commands_from_args(
 
2027
            ("--timeout=PT5M --interval=PT1M foo1.example.org"
 
2028
             " foo2.example.org").split(),
 
2029
            [command.SetTimeout, command.SetInterval],
 
2030
            clients=["foo1.example.org", "foo2.example.org"])
 
2031
 
 
2032
    def test_manual_page_example_5(self):
 
2033
        self.assert_command_from_args("--approve --all".split(),
 
2034
                                      command.Approve,
 
2035
                                      clients=[])
 
2036
 
 
2037
 
1992
2038
class TestCommand(unittest.TestCase):
1993
2039
    """Abstract class for tests of command classes"""
1994
2040