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
- browse files at revision 1097
- compare with another revision
- download diff
- download tarball
- view history from revision 1097
- Committer: Teddy Hogeborn
- Date: 2019-04-08 21:53:22 UTC
- Revision ID: teddy@recompile.se-20190408215322-y3hmfxzdgs9t84l1
plugin-runner: Fix minor memory leak
* plugin-runner.c (free_plugin): Even if argv[0] is NULL, as for the
pseudo-plugin for global options,
free all other arguments by simply
starting at argv[1] and freeing
plugin_node->name (which is always
argv[0]) separately.
* plugin-runner.c (free_plugin): Even if argv[0] is NULL, as for the
pseudo-plugin for global options,
free all other arguments by simply
starting at argv[1] and freeing
plugin_node->name (which is always
argv[0]) separately.
- files removed:
- debian/po/POTFILES.in
- debian/tests
- dracut-module
- files modified:
- .bzrignore
added
removed
mandos-ctl
1
#!/usr/bin/python3 -bbI
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 -*-
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*")))); -*-
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
52
49
try:
53
50
import pydbus
54
51
import gi
56
53
except ImportError:
57
54
import dbus as dbus_python
58
55
pydbus = None
59
class gi:
56
class gi(object):
60
57
"""Dummy gi module, for the tests"""
61
class repository:
62
class GLib:
58
class repository(object):
59
class GLib(object):
63
60
class Error(Exception):
64
61
pass
65
62
81
78
82
79
locale.setlocale(locale.LC_ALL, "")
83
80
84
version = "1.8.9"
81
version = "1.8.3"
85
82
86
83
87
84
def main():
253
250
def rfc3339_duration_to_delta(duration):
254
251
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
255
252
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
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)
273
269
>>> # Can not be empty:
274
270
>>> rfc3339_duration_to_delta("")
275
271
Traceback (most recent call last):
385
381
"""Parse an interval string as documented by Mandos before 1.6.1,
386
382
and return a datetime.timedelta
387
383
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
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)
402
398
>>> # Ignore unknown characters, allow any order and repetitions
403
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
404
True
399
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
datetime.timedelta(2, 480, 18000)
405
401
406
402
"""
407
403
470
466
parser.error("--remove can only be combined with --deny")
471
467
472
468
473
class dbus:
469
class dbus(object):
474
470
475
class SystemBus:
471
class SystemBus(object):
476
472
477
473
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
478
474
def get_managed_objects(self, busname, objectpath):
524
520
pass
525
521
526
522
527
class dbus_python_adapter:
523
class dbus_python_adapter(object):
528
524
529
525
class SystemBus(dbus.MandosBus):
530
526
"""Use dbus-python"""
575
571
for key, subval in value.items()}
576
572
return value
577
573
578
def set_client_property(self, objectpath, key, value):
579
if key == "Secret":
580
if not isinstance(value, bytes):
581
value = value.encode("utf-8")
582
value = self.dbus_python.ByteArray(value)
583
return self.set_property(self.busname, objectpath,
584
self.client_interface, key,
585
value)
586
574
587
class SilenceLogger:
575
class SilenceLogger(object):
588
576
"Simple context manager to silence a particular logger"
589
577
def __init__(self, loggername):
590
578
self.logger = logging.getLogger(loggername)
619
607
return new_object
620
608
621
609
622
class pydbus_adapter:
610
class pydbus_adapter(object):
623
611
class SystemBus(dbus.MandosBus):
624
612
def __init__(self, module=pydbus):
625
613
self.pydbus = module
715
703
return commands
716
704
717
705
718
class command:
706
class command(object):
719
707
"""A namespace for command classes"""
720
708
721
class Base:
709
class Base(object):
722
710
"""Abstract base class for commands"""
723
711
def run(self, clients, bus=None):
724
712
"""Normal commands should implement run_on_one_client(),
787
775
keywords = self.all_keywords
788
776
print(self.TableOfClients(clients.values(), keywords))
789
777
790
class TableOfClients:
778
class TableOfClients(object):
791
779
tableheaders = {
792
780
"Name": "Name",
793
781
"Enabled": "Enabled",
1024
1012
"output"))
1025
1013
1026
1014
1027
class Unique:
1015
class Unique(object):
1028
1016
"""Class for objects which exist only to be unique objects, since
1029
1017
unittest.mock.sentinel only exists in Python 3.3"""
1030
1018
1314
1302
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1315
1303
1316
1304
def MockDBusPython_func(self, func):
1317
class mock_dbus_python:
1305
class mock_dbus_python(object):
1318
1306
"""mock dbus-python module"""
1319
class exceptions:
1307
class exceptions(object):
1320
1308
"""Pseudo-namespace"""
1321
1309
class DBusException(Exception):
1322
1310
pass
1323
class SystemBus:
1311
class SystemBus(object):
1324
1312
@staticmethod
1325
1313
def get_object(busname, objectpath):
1326
1314
DBusObject = collections.namedtuple(
1327
"DBusObject", ("methodname", "Set"))
1315
"DBusObject", ("methodname",))
1328
1316
def method(*args, **kwargs):
1329
1317
self.assertEqual({"dbus_interface":
1330
1318
"interface"},
1331
1319
kwargs)
1332
1320
return func(*args)
1333
def set_property(interface, key, value,
1334
dbus_interface=None):
1335
self.assertEqual(
1336
"org.freedesktop.DBus.Properties",
1337
dbus_interface)
1338
self.assertEqual("Secret", key)
1339
return func(interface, key, value,
1340
dbus_interface=dbus_interface)
1341
return DBusObject(methodname=method,
1342
Set=set_property)
1343
class Boolean:
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1344
1323
def __init__(self, value):
1345
1324
self.value = bool(value)
1346
1325
def __bool__(self):
1351
1330
pass
1352
1331
class Dictionary(dict):
1353
1332
pass
1354
class ByteArray(bytes):
1355
pass
1356
1333
return mock_dbus_python
1357
1334
1358
1335
def call_method(self, bus, methodname, busname, objectpath,
1535
1512
# Make sure the dbus logger was suppressed
1536
1513
self.assertEqual(0, counting_handler.count)
1537
1514
1538
def test_Set_Secret_sends_bytearray(self):
1539
ret = [None]
1540
def func(*args, **kwargs):
1541
ret[0] = (args, kwargs)
1542
mock_dbus_python = self.MockDBusPython_func(func)
1543
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1544
bus.set_client_property("objectpath", "Secret", "value")
1545
expected_call = (("se.recompile.Mandos.Client", "Secret",
1546
mock_dbus_python.ByteArray(b"value")),
1547
{"dbus_interface":
1548
"org.freedesktop.DBus.Properties"})
1549
self.assertEqual(expected_call, ret[0])
1550
if sys.version_info.major == 2:
1551
self.assertIsInstance(ret[0][0][-1],
1552
mock_dbus_python.ByteArray)
1553
1554
1515
def test_get_object_converts_to_correct_exception(self):
1555
1516
bus = dbus_python_adapter.SystemBus(
1556
1517
self.fake_dbus_python_raises_exception_on_connect)
1558
1519
self.call_method(bus, "methodname", "busname",
1559
1520
"objectpath", "interface")
1560
1521
1561
class fake_dbus_python_raises_exception_on_connect:
1522
class fake_dbus_python_raises_exception_on_connect(object):
1562
1523
"""fake dbus-python module"""
1563
class exceptions:
1524
class exceptions(object):
1564
1525
"""Pseudo-namespace"""
1565
1526
class DBusException(Exception):
1566
1527
pass
1574
1535
1575
1536
1576
1537
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1577
class mock_dbus_python:
1538
class mock_dbus_python(object):
1578
1539
"""mock dbus-python modules"""
1579
class SystemBus:
1540
class SystemBus(object):
1580
1541
@staticmethod
1581
1542
def get_object(busname, objectpath):
1582
1543
return Unique()
1628
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1629
1590
1630
1591
def Stub_pydbus_func(self, func):
1631
class stub_pydbus:
1592
class stub_pydbus(object):
1632
1593
"""stub pydbus module"""
1633
class SystemBus:
1594
class SystemBus(object):
1634
1595
@staticmethod
1635
1596
def get(busname, objectpath):
1636
1597
DBusObject = collections.namedtuple(
1682
1643
self.call_method(bus, "methodname", "busname",
1683
1644
"objectpath", "interface")
1684
1645
1685
class fake_pydbus_raises_exception_on_connect:
1646
class fake_pydbus_raises_exception_on_connect(object):
1686
1647
"""fake dbus-python module"""
1687
1648
@classmethod
1688
1649
def SystemBus(cls):
1692
1653
return Bus(get=get)
1693
1654
1694
1655
def test_set_property_uses_setattr(self):
1695
class Object:
1656
class Object(object):
1696
1657
pass
1697
1658
obj = Object()
1698
class pydbus_spy:
1699
class SystemBus:
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1700
1661
@staticmethod
1701
1662
def get(busname, objectpath):
1702
1663
return {"interface": obj}
1709
1670
def test_get_suppresses_xml_deprecation_warning(self):
1710
1671
if sys.version_info.major >= 3:
1711
1672
return
1712
class stub_pydbus_get:
1713
class SystemBus:
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1714
1675
@staticmethod
1715
1676
def get(busname, objectpath):
1716
1677
warnings.warn_explicit(
1724
1685
1725
1686
1726
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1727
class stub_pydbus:
1688
class stub_pydbus(object):
1728
1689
"""stub pydbus module"""
1729
class SystemBus:
1690
class SystemBus(object):
1730
1691
@staticmethod
1731
1692
def get(busname, objectpath):
1732
1693
return Unique()
1784
1745
self.assert_command_from_args(["--is-enabled", "client"],
1785
1746
command.IsEnabled)
1786
1747
1787
def assert_command_from_args(self, args, command_cls, length=1,
1788
clients=None, **cmd_attrs):
1748
def assert_command_from_args(self, args, command_cls,
1749
**cmd_attrs):
1789
1750
"""Assert that parsing ARGS should result in an instance of
1790
1751
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1791
1752
options = self.parser.parse_args(args)
1792
1753
check_option_syntax(self.parser, options)
1793
1754
commands = commands_from_options(options)
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)
1755
self.assertEqual(1, len(commands))
1756
command = commands[0]
1757
self.assertIsInstance(command, command_cls)
1802
1758
for key, value in cmd_attrs.items():
1803
1759
self.assertEqual(value, getattr(command, key))
1804
1760
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
1811
1761
def test_is_enabled_short(self):
1812
1762
self.assert_command_from_args(["-V", "client"],
1813
1763
command.IsEnabled)
2004
1954
verbose=True)
2005
1955
2006
1956
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
2038
1957
class TestCommand(unittest.TestCase):
2039
1958
"""Abstract class for tests of command classes"""
2040
1959
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0