/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: 2024-09-15 20:21:49 UTC
  • Revision ID: teddy@recompile.se-20240915202149-447ayubypn08tpki
Fix lintian warning "build-depends-on-obsolete-package"

Apparently, pkg-config is deprecated in favor of pkgconf:
<https://bugs.debian.org/1020799>,
<http://pkgconf-migration.debian.net/>

Therefore, prefer pkgconf in the build dependencies.

* debian/control (Source: mandos/Build-Depends): Change "pkg-config"
  to "pkgconf | pkg-config".

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
 
#!/usr/bin/python
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 --test" (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
 
#
4
 
# Mandos Monitor - Control and monitor the Mandos server
5
 
#
6
 
# Copyright © 2008-2019 Teddy Hogeborn
7
 
# Copyright © 2008-2019 Björn Påhlsson
 
1
#!/usr/bin/python3 -bbI
 
2
# -*- coding: utf-8; lexical-binding: t -*-
 
3
#
 
4
# Mandos Control - Control or query the Mandos server
 
5
#
 
6
# Copyright © 2008-2022 Teddy Hogeborn
 
7
# Copyright © 2008-2022 Björn Påhlsson
8
8
#
9
9
# This file is part of Mandos.
10
10
#
23
23
#
24
24
# Contact the authors at <mandos@recompile.se>.
25
25
#
26
 
 
27
26
from __future__ import (division, absolute_import, print_function,
28
27
                        unicode_literals)
29
28
 
33
32
    pass
34
33
 
35
34
import sys
 
35
import unittest
36
36
import argparse
 
37
import logging
 
38
import os
37
39
import locale
38
40
import datetime
39
41
import re
40
 
import os
41
42
import collections
42
43
import json
43
 
import unittest
44
 
import logging
45
44
import io
46
45
import tempfile
47
46
import contextlib
48
47
 
 
48
if sys.version_info.major == 2:
 
49
    __metaclass__ = type
 
50
    str = unicode
 
51
    input = raw_input
 
52
 
 
53
 
 
54
class gi:
 
55
    """Dummy gi module, for the tests"""
 
56
    class repository:
 
57
        class GLib:
 
58
            class Error(Exception):
 
59
                pass
 
60
 
 
61
 
 
62
dbussy = None
 
63
ravel = None
 
64
dbus_python = None
 
65
pydbus = None
 
66
 
49
67
try:
50
 
    import pydbus
51
 
    import gi
52
 
    dbus_python = None
 
68
    import dbussy
 
69
    import ravel
53
70
except ImportError:
54
 
    import dbus as dbus_python
55
 
    pydbus = None
56
 
    class gi(object):
57
 
        """Dummy gi module, for the tests"""
58
 
        class repository(object):
59
 
            class GLib(object):
60
 
                class Error(Exception):
61
 
                    pass
 
71
    try:
 
72
        import pydbus
 
73
        import gi
 
74
    except ImportError:
 
75
        import dbus as dbus_python
 
76
 
62
77
 
63
78
# Show warnings by default
64
79
if not sys.warnoptions:
65
80
    import warnings
66
81
    warnings.simplefilter("default")
67
82
 
68
 
log = logging.getLogger(sys.argv[0])
69
 
logging.basicConfig(level="INFO", # Show info level messages
 
83
log = logging.getLogger(os.path.basename(sys.argv[0]))
 
84
logging.basicConfig(level="INFO",         # Show info level messages
70
85
                    format="%(message)s") # Show basic log messages
71
86
 
72
87
logging.captureWarnings(True)   # Show warnings via the logging system
73
88
 
74
89
if sys.version_info.major == 2:
75
 
    str = unicode
76
90
    import StringIO
77
91
    io.StringIO = StringIO.StringIO
78
92
 
79
93
locale.setlocale(locale.LC_ALL, "")
80
94
 
81
 
version = "1.8.4"
 
95
version = "1.8.17"
82
96
 
83
97
 
84
98
def main():
91
105
    clientnames = options.client
92
106
 
93
107
    if options.debug:
94
 
        log.setLevel(logging.DEBUG)
 
108
        logging.getLogger("").setLevel(logging.DEBUG)
95
109
 
96
 
    if pydbus is not None:
 
110
    if dbussy is not None and ravel is not None:
 
111
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
112
    elif pydbus is not None:
97
113
        bus = pydbus_adapter.CachingBus(pydbus)
98
114
    else:
99
115
        bus = dbus_python_adapter.CachingBus(dbus_python)
243
259
        return rfc3339_duration_to_delta(interval)
244
260
    except ValueError as e:
245
261
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
246
 
                    ' '.join(e.args))
 
262
                    " ".join(e.args))
247
263
    return parse_pre_1_6_1_interval(interval)
248
264
 
249
265
 
250
266
def rfc3339_duration_to_delta(duration):
251
267
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
252
268
 
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)
 
269
    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
 
270
    True
 
271
    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
 
272
    True
 
273
    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
 
274
    True
 
275
    >>> # 60 months
 
276
    >>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
 
277
    True
 
278
    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
 
279
    True
 
280
    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
 
281
    True
 
282
    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
 
283
    True
 
284
    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
 
285
    True
269
286
    >>> # Can not be empty:
270
287
    >>> rfc3339_duration_to_delta("")
271
288
    Traceback (most recent call last):
378
395
 
379
396
 
380
397
def parse_pre_1_6_1_interval(interval):
381
 
    """Parse an interval string as documented by Mandos before 1.6.1,
 
398
    r"""Parse an interval string as documented by Mandos before 1.6.1,
382
399
    and return a datetime.timedelta
383
400
 
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)
 
401
    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
 
402
    True
 
403
    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
 
404
    True
 
405
    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
 
406
    True
 
407
    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
 
408
    True
 
409
    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
 
410
    True
 
411
    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
 
412
    True
 
413
    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
 
414
    True
398
415
    >>> # Ignore unknown characters, allow any order and repetitions
399
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
 
    datetime.timedelta(2, 480, 18000)
 
416
    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \
 
417
    ... == datetime.timedelta(2, 480, 18000)
 
418
    True
401
419
 
402
420
    """
403
421
 
466
484
        parser.error("--remove can only be combined with --deny")
467
485
 
468
486
 
469
 
class dbus(object):
 
487
class dbus:
470
488
 
471
 
    class SystemBus(object):
 
489
    class SystemBus:
472
490
 
473
491
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
 
492
 
474
493
        def get_managed_objects(self, busname, objectpath):
475
494
            return self.call_method("GetManagedObjects", busname,
476
495
                                    objectpath,
477
496
                                    self.object_manager_iface)
478
497
 
479
498
        properties_iface = "org.freedesktop.DBus.Properties"
 
499
 
480
500
        def set_property(self, busname, objectpath, interface, key,
481
501
                         value):
482
502
            self.call_method("Set", busname, objectpath,
483
503
                             self.properties_iface, interface, key,
484
504
                             value)
485
505
 
 
506
        def call_method(self, methodname, busname, objectpath,
 
507
                        interface, *args):
 
508
            raise NotImplementedError()
486
509
 
487
510
    class MandosBus(SystemBus):
488
511
        busname_domain = "se.recompile"
520
543
        pass
521
544
 
522
545
 
523
 
class dbus_python_adapter(object):
 
546
class dbus_python_adapter:
524
547
 
525
548
    class SystemBus(dbus.MandosBus):
526
549
        """Use dbus-python"""
580
603
                                     self.client_interface, key,
581
604
                                     value)
582
605
 
583
 
    class SilenceLogger(object):
 
606
    class SilenceLogger:
584
607
        "Simple context manager to silence a particular logger"
 
608
 
585
609
        def __init__(self, loggername):
586
610
            self.logger = logging.getLogger(loggername)
587
611
 
597
621
        def __exit__(self, exc_type, exc_val, exc_tb):
598
622
            self.logger.removeFilter(self.nullfilter)
599
623
 
600
 
 
601
624
    class CachingBus(SystemBus):
602
625
        """A caching layer for dbus_python_adapter.SystemBus"""
 
626
 
603
627
        def __init__(self, *args, **kwargs):
604
628
            self.object_cache = {}
605
629
            super(dbus_python_adapter.CachingBus,
606
630
                  self).__init__(*args, **kwargs)
 
631
 
607
632
        def get_object(self, busname, objectpath):
608
633
            try:
609
634
                return self.object_cache[(busname, objectpath)]
611
636
                new_object = super(
612
637
                    dbus_python_adapter.CachingBus,
613
638
                    self).get_object(busname, objectpath)
614
 
                self.object_cache[(busname, objectpath)]  = new_object
 
639
                self.object_cache[(busname, objectpath)] = new_object
615
640
                return new_object
616
641
 
617
642
 
618
 
class pydbus_adapter(object):
 
643
class pydbus_adapter:
619
644
    class SystemBus(dbus.MandosBus):
620
645
        def __init__(self, module=pydbus):
621
646
            self.pydbus = module
664
689
 
665
690
    class CachingBus(SystemBus):
666
691
        """A caching layer for pydbus_adapter.SystemBus"""
 
692
 
667
693
        def __init__(self, *args, **kwargs):
668
694
            self.object_cache = {}
669
695
            super(pydbus_adapter.CachingBus,
670
696
                  self).__init__(*args, **kwargs)
 
697
 
671
698
        def get(self, busname, objectpath):
672
699
            try:
673
700
                return self.object_cache[(busname, objectpath)]
674
701
            except KeyError:
675
702
                new_object = (super(pydbus_adapter.CachingBus, self)
676
703
                              .get(busname, objectpath))
677
 
                self.object_cache[(busname, objectpath)]  = new_object
 
704
                self.object_cache[(busname, objectpath)] = new_object
 
705
                return new_object
 
706
 
 
707
 
 
708
class dbussy_adapter:
 
709
    class SystemBus(dbus.SystemBus):
 
710
        """Use DBussy"""
 
711
 
 
712
        def __init__(self, dbussy, ravel):
 
713
            self.dbussy = dbussy
 
714
            self.ravel = ravel
 
715
            self.bus = ravel.system_bus()
 
716
 
 
717
        @contextlib.contextmanager
 
718
        def convert_exception(self, exception_class=dbus.Error):
 
719
            try:
 
720
                yield
 
721
            except self.dbussy.DBusError as e:
 
722
                # This does what "raise from" would do
 
723
                exc = exception_class(*e.args)
 
724
                exc.__cause__ = e
 
725
                raise exc
 
726
 
 
727
        def call_method(self, methodname, busname, objectpath,
 
728
                        interface, *args):
 
729
            proxy_object = self.get_object(busname, objectpath)
 
730
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
731
                      interface, methodname,
 
732
                      ", ".join(repr(a) for a in args))
 
733
            iface = proxy_object.get_interface(interface)
 
734
            method = getattr(iface, methodname)
 
735
            with self.convert_exception(dbus.Error):
 
736
                value = method(*args)
 
737
            # DBussy returns values either as an empty list or as a
 
738
            # list of one element with the return value
 
739
            if value:
 
740
                return self.type_filter(value[0])
 
741
 
 
742
        def get_object(self, busname, objectpath):
 
743
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
744
                      busname, objectpath)
 
745
            with self.convert_exception(dbus.ConnectFailed):
 
746
                return self.bus[busname][objectpath]
 
747
 
 
748
        def type_filter(self, value):
 
749
            """Convert the most bothersome types to Python types"""
 
750
            # A D-Bus Variant value is represented as the Python type
 
751
            # Tuple[dbussy.DBUS.Signature, Any]
 
752
            if isinstance(value, tuple):
 
753
                if (len(value) == 2
 
754
                    and isinstance(value[0],
 
755
                                   self.dbussy.DBUS.Signature)):
 
756
                    return self.type_filter(value[1])
 
757
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
 
758
                return str(value)
 
759
            # Also recurse into dictionaries
 
760
            elif isinstance(value, dict):
 
761
                return {self.type_filter(key):
 
762
                        self.type_filter(subval)
 
763
                        for key, subval in value.items()}
 
764
            return value
 
765
 
 
766
        def set_property(self, busname, objectpath, interface, key,
 
767
                         value):
 
768
            proxy_object = self.get_object(busname, objectpath)
 
769
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
770
                      objectpath, self.properties_iface, interface,
 
771
                      key, value)
 
772
            if key == "Secret":
 
773
                # DBussy wants a Byte Array to be a sequence of
 
774
                # values, not a byte string
 
775
                value = tuple(value)
 
776
            setattr(proxy_object.get_interface(interface), key, value)
 
777
 
 
778
    class MandosBus(SystemBus, dbus.MandosBus):
 
779
        pass
 
780
 
 
781
    class CachingBus(MandosBus):
 
782
        """A caching layer for dbussy_adapter.MandosBus"""
 
783
 
 
784
        def __init__(self, *args, **kwargs):
 
785
            self.object_cache = {}
 
786
            super(dbussy_adapter.CachingBus, self).__init__(*args,
 
787
                                                            **kwargs)
 
788
 
 
789
        def get_object(self, busname, objectpath):
 
790
            try:
 
791
                return self.object_cache[(busname, objectpath)]
 
792
            except KeyError:
 
793
                new_object = super(
 
794
                    dbussy_adapter.CachingBus,
 
795
                    self).get_object(busname, objectpath)
 
796
                self.object_cache[(busname, objectpath)] = new_object
678
797
                return new_object
679
798
 
680
799
 
711
830
    return commands
712
831
 
713
832
 
714
 
class command(object):
 
833
class command:
715
834
    """A namespace for command classes"""
716
835
 
717
 
    class Base(object):
 
836
    class Base:
718
837
        """Abstract base class for commands"""
 
838
 
719
839
        def run(self, clients, bus=None):
720
840
            """Normal commands should implement run_on_one_client(),
721
841
but commands which want to operate on all clients at the same time can
725
845
            for client, properties in clients.items():
726
846
                self.run_on_one_client(client, properties)
727
847
 
728
 
 
729
848
    class IsEnabled(Base):
730
849
        def run(self, clients, bus=None):
731
850
            properties = next(iter(clients.values()))
733
852
                sys.exit(0)
734
853
            sys.exit(1)
735
854
 
736
 
 
737
855
    class Approve(Base):
738
856
        def run_on_one_client(self, client, properties):
739
857
            self.bus.call_client_method(client, "Approve", True)
740
858
 
741
 
 
742
859
    class Deny(Base):
743
860
        def run_on_one_client(self, client, properties):
744
861
            self.bus.call_client_method(client, "Approve", False)
745
862
 
746
 
 
747
863
    class Remove(Base):
748
864
        def run(self, clients, bus):
749
865
            for clientpath in frozenset(clients.keys()):
750
866
                bus.call_server_method("RemoveClient", clientpath)
751
867
 
752
 
 
753
868
    class Output(Base):
754
869
        """Abstract class for commands outputting client details"""
755
870
        all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
761
876
                        "Checker", "ExtendedTimeout", "Expires",
762
877
                        "LastCheckerStatus")
763
878
 
764
 
 
765
879
    class DumpJSON(Output):
766
880
        def run(self, clients, bus=None):
767
881
            data = {properties["Name"]:
768
882
                    {key: properties[key]
769
883
                     for key in self.all_keywords}
770
884
                    for properties in clients.values()}
771
 
            print(json.dumps(data, indent=4, separators=(',', ': ')))
772
 
 
 
885
            print(json.dumps(data, indent=4, separators=(",", ": ")))
773
886
 
774
887
    class PrintTable(Output):
775
888
        def __init__(self, verbose=False):
783
896
                keywords = self.all_keywords
784
897
            print(self.TableOfClients(clients.values(), keywords))
785
898
 
786
 
        class TableOfClients(object):
 
899
        class TableOfClients:
787
900
            tableheaders = {
788
901
                "Name": "Name",
789
902
                "Enabled": "Enabled",
816
929
 
817
930
            if sys.version_info.major == 2:
818
931
                __unicode__ = __str__
 
932
 
819
933
                def __str__(self):
820
934
                    return str(self).encode(
821
935
                        locale.getpreferredencoding())
867
981
                                minutes=(td.seconds % 3600) // 60,
868
982
                                seconds=td.seconds % 60))
869
983
 
870
 
 
871
984
    class PropertySetter(Base):
872
985
        "Abstract class for Actions for setting one client property"
873
986
 
880
993
        def propname(self):
881
994
            raise NotImplementedError()
882
995
 
883
 
 
884
996
    class Enable(PropertySetter):
885
997
        propname = "Enabled"
886
998
        value_to_set = True
887
999
 
888
 
 
889
1000
    class Disable(PropertySetter):
890
1001
        propname = "Enabled"
891
1002
        value_to_set = False
892
1003
 
893
 
 
894
1004
    class BumpTimeout(PropertySetter):
895
1005
        propname = "LastCheckedOK"
896
1006
        value_to_set = ""
897
1007
 
898
 
 
899
1008
    class StartChecker(PropertySetter):
900
1009
        propname = "CheckerRunning"
901
1010
        value_to_set = True
902
1011
 
903
 
 
904
1012
    class StopChecker(PropertySetter):
905
1013
        propname = "CheckerRunning"
906
1014
        value_to_set = False
907
1015
 
908
 
 
909
1016
    class ApproveByDefault(PropertySetter):
910
1017
        propname = "ApprovedByDefault"
911
1018
        value_to_set = True
912
1019
 
913
 
 
914
1020
    class DenyByDefault(PropertySetter):
915
1021
        propname = "ApprovedByDefault"
916
1022
        value_to_set = False
917
1023
 
918
 
 
919
1024
    class PropertySetterValue(PropertySetter):
920
1025
        """Abstract class for PropertySetter recieving a value as
921
1026
constructor argument instead of a class attribute."""
 
1027
 
922
1028
        def __init__(self, value):
923
1029
            self.value_to_set = value
924
1030
 
931
1037
    class SetChecker(PropertySetterValue):
932
1038
        propname = "Checker"
933
1039
 
934
 
 
935
1040
    class SetHost(PropertySetterValue):
936
1041
        propname = "Host"
937
1042
 
938
 
 
939
1043
    class SetSecret(PropertySetterValue):
940
1044
        propname = "Secret"
941
1045
 
949
1053
            self._vts = value.read()
950
1054
            value.close()
951
1055
 
952
 
 
953
1056
    class PropertySetterValueMilliseconds(PropertySetterValue):
954
1057
        """Abstract class for PropertySetterValue taking a value
955
1058
argument as a datetime.timedelta() but should store it as
964
1067
            "When setting, convert value from a datetime.timedelta"
965
1068
            self._vts = int(round(value.total_seconds() * 1000))
966
1069
 
967
 
 
968
1070
    class SetTimeout(PropertySetterValueMilliseconds):
969
1071
        propname = "Timeout"
970
1072
 
971
 
 
972
1073
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
973
1074
        propname = "ExtendedTimeout"
974
1075
 
975
 
 
976
1076
    class SetInterval(PropertySetterValueMilliseconds):
977
1077
        propname = "Interval"
978
1078
 
979
 
 
980
1079
    class SetApprovalDelay(PropertySetterValueMilliseconds):
981
1080
        propname = "ApprovalDelay"
982
1081
 
983
 
 
984
1082
    class SetApprovalDuration(PropertySetterValueMilliseconds):
985
1083
        propname = "ApprovalDuration"
986
1084
 
1020
1118
                                                     "output"))
1021
1119
 
1022
1120
 
1023
 
class Unique(object):
 
1121
class Unique:
1024
1122
    """Class for objects which exist only to be unique objects, since
1025
1123
unittest.mock.sentinel only exists in Python 3.3"""
1026
1124
 
1310
1408
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1311
1409
 
1312
1410
    def MockDBusPython_func(self, func):
1313
 
        class mock_dbus_python(object):
 
1411
        class mock_dbus_python:
1314
1412
            """mock dbus-python module"""
1315
 
            class exceptions(object):
 
1413
            class exceptions:
1316
1414
                """Pseudo-namespace"""
1317
1415
                class DBusException(Exception):
1318
1416
                    pass
1319
 
            class SystemBus(object):
 
1417
            class SystemBus:
1320
1418
                @staticmethod
1321
1419
                def get_object(busname, objectpath):
1322
1420
                    DBusObject = collections.namedtuple(
1336
1434
                                    dbus_interface=dbus_interface)
1337
1435
                    return DBusObject(methodname=method,
1338
1436
                                      Set=set_property)
1339
 
            class Boolean(object):
 
1437
            class Boolean:
1340
1438
                def __init__(self, value):
1341
1439
                    self.value = bool(value)
1342
1440
                def __bool__(self):
1526
1624
        finally:
1527
1625
            dbus_logger.removeFilter(counting_handler)
1528
1626
 
1529
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1627
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1530
1628
 
1531
1629
        # Make sure the dbus logger was suppressed
1532
1630
        self.assertEqual(0, counting_handler.count)
1554
1652
            self.call_method(bus, "methodname", "busname",
1555
1653
                             "objectpath", "interface")
1556
1654
 
1557
 
    class fake_dbus_python_raises_exception_on_connect(object):
 
1655
    class fake_dbus_python_raises_exception_on_connect:
1558
1656
        """fake dbus-python module"""
1559
 
        class exceptions(object):
 
1657
        class exceptions:
1560
1658
            """Pseudo-namespace"""
1561
1659
            class DBusException(Exception):
1562
1660
                pass
1570
1668
 
1571
1669
 
1572
1670
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1573
 
    class mock_dbus_python(object):
 
1671
    class mock_dbus_python:
1574
1672
        """mock dbus-python modules"""
1575
 
        class SystemBus(object):
 
1673
        class SystemBus:
1576
1674
            @staticmethod
1577
1675
            def get_object(busname, objectpath):
1578
1676
                return Unique()
1624
1722
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1625
1723
 
1626
1724
    def Stub_pydbus_func(self, func):
1627
 
        class stub_pydbus(object):
 
1725
        class stub_pydbus:
1628
1726
            """stub pydbus module"""
1629
 
            class SystemBus(object):
 
1727
            class SystemBus:
1630
1728
                @staticmethod
1631
1729
                def get(busname, objectpath):
1632
1730
                    DBusObject = collections.namedtuple(
1657
1755
        self.assertIs(ret, expected_method_return)
1658
1756
 
1659
1757
    def test_call_method_handles_exception(self):
1660
 
        dbus_logger = logging.getLogger("dbus.proxies")
1661
 
 
1662
1758
        def func():
1663
1759
            raise gi.repository.GLib.Error()
1664
1760
 
1669
1765
            self.call_method(bus, "methodname", "busname",
1670
1766
                             "objectpath", "interface")
1671
1767
 
1672
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1768
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1673
1769
 
1674
1770
    def test_get_converts_to_correct_exception(self):
1675
1771
        bus = pydbus_adapter.SystemBus(
1678
1774
            self.call_method(bus, "methodname", "busname",
1679
1775
                             "objectpath", "interface")
1680
1776
 
1681
 
    class fake_pydbus_raises_exception_on_connect(object):
 
1777
    class fake_pydbus_raises_exception_on_connect:
1682
1778
        """fake dbus-python module"""
1683
1779
        @classmethod
1684
1780
        def SystemBus(cls):
1688
1784
            return Bus(get=get)
1689
1785
 
1690
1786
    def test_set_property_uses_setattr(self):
1691
 
        class Object(object):
 
1787
        class Object:
1692
1788
            pass
1693
1789
        obj = Object()
1694
 
        class pydbus_spy(object):
1695
 
            class SystemBus(object):
 
1790
        class pydbus_spy:
 
1791
            class SystemBus:
1696
1792
                @staticmethod
1697
1793
                def get(busname, objectpath):
1698
1794
                    return {"interface": obj}
1705
1801
    def test_get_suppresses_xml_deprecation_warning(self):
1706
1802
        if sys.version_info.major >= 3:
1707
1803
            return
1708
 
        class stub_pydbus_get(object):
1709
 
            class SystemBus(object):
 
1804
        class stub_pydbus_get:
 
1805
            class SystemBus:
1710
1806
                @staticmethod
1711
1807
                def get(busname, objectpath):
1712
1808
                    warnings.warn_explicit(
1720
1816
 
1721
1817
 
1722
1818
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1723
 
    class stub_pydbus(object):
 
1819
    class stub_pydbus:
1724
1820
        """stub pydbus module"""
1725
 
        class SystemBus(object):
 
1821
        class SystemBus:
1726
1822
            @staticmethod
1727
1823
            def get(busname, objectpath):
1728
1824
                return Unique()
1770
1866
        self.assertIs(obj1, obj1b)
1771
1867
 
1772
1868
 
 
1869
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
 
1870
 
 
1871
    class dummy_dbussy:
 
1872
        class DBUS:
 
1873
            class ObjectPath(str):
 
1874
                pass
 
1875
        class DBusError(Exception):
 
1876
            pass
 
1877
 
 
1878
    def fake_ravel_func(self, func):
 
1879
        class fake_ravel:
 
1880
            @staticmethod
 
1881
            def system_bus():
 
1882
                class DBusInterfaceProxy:
 
1883
                    @staticmethod
 
1884
                    def methodname(*args):
 
1885
                        return [func(*args)]
 
1886
                class DBusObject:
 
1887
                    @staticmethod
 
1888
                    def get_interface(interface):
 
1889
                        if interface == "interface":
 
1890
                            return DBusInterfaceProxy()
 
1891
                return {"busname": {"objectpath": DBusObject()}}
 
1892
        return fake_ravel
 
1893
 
 
1894
    def call_method(self, bus, methodname, busname, objectpath,
 
1895
                    interface, *args):
 
1896
        with self.assertLogs(log, logging.DEBUG):
 
1897
            return bus.call_method(methodname, busname, objectpath,
 
1898
                                   interface, *args)
 
1899
 
 
1900
    def test_call_method_returns(self):
 
1901
        expected_method_return = Unique()
 
1902
        method_args = (Unique(), Unique())
 
1903
        def func(*args):
 
1904
            self.assertEqual(len(method_args), len(args))
 
1905
            for marg, arg in zip(method_args, args):
 
1906
                self.assertIs(marg, arg)
 
1907
            return expected_method_return
 
1908
        fake_ravel = self.fake_ravel_func(func)
 
1909
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1910
        ret = self.call_method(bus, "methodname", "busname",
 
1911
                               "objectpath", "interface",
 
1912
                               *method_args)
 
1913
        self.assertIs(ret, expected_method_return)
 
1914
 
 
1915
    def test_call_method_filters_objectpath(self):
 
1916
        def func():
 
1917
            return method_return
 
1918
        fake_ravel = self.fake_ravel_func(func)
 
1919
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1920
        method_return = (self.dummy_dbussy.DBUS
 
1921
                         .ObjectPath("objectpath"))
 
1922
        ret = self.call_method(bus, "methodname", "busname",
 
1923
                               "objectpath", "interface")
 
1924
        self.assertEqual("objectpath", ret)
 
1925
        self.assertNotIsInstance(ret,
 
1926
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
1927
 
 
1928
    def test_call_method_filters_objectpaths_in_dict(self):
 
1929
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1930
        def func():
 
1931
            return method_return
 
1932
        fake_ravel = self.fake_ravel_func(func)
 
1933
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1934
        method_return = {
 
1935
            ObjectPath("objectpath_key_1"):
 
1936
            ObjectPath("objectpath_value_1"),
 
1937
            ObjectPath("objectpath_key_2"):
 
1938
            ObjectPath("objectpath_value_2"),
 
1939
        }
 
1940
        ret = self.call_method(bus, "methodname", "busname",
 
1941
                               "objectpath", "interface")
 
1942
        expected_method_return = {str(key): str(value)
 
1943
                                  for key, value in
 
1944
                                  method_return.items()}
 
1945
        for key, value in ret.items():
 
1946
            self.assertNotIsInstance(key, ObjectPath)
 
1947
            self.assertNotIsInstance(value, ObjectPath)
 
1948
        self.assertEqual(expected_method_return, ret)
 
1949
        self.assertIsInstance(ret, dict)
 
1950
 
 
1951
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
 
1952
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1953
        def func():
 
1954
            return method_return
 
1955
        fake_ravel = self.fake_ravel_func(func)
 
1956
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1957
        method_return = {
 
1958
            ObjectPath("key1"): {
 
1959
                ObjectPath("key11"): ObjectPath("value11"),
 
1960
                ObjectPath("key12"): ObjectPath("value12"),
 
1961
            },
 
1962
            ObjectPath("key2"): {
 
1963
                ObjectPath("key21"): ObjectPath("value21"),
 
1964
                ObjectPath("key22"): ObjectPath("value22"),
 
1965
            },
 
1966
        }
 
1967
        ret = self.call_method(bus, "methodname", "busname",
 
1968
                               "objectpath", "interface")
 
1969
        expected_method_return = {
 
1970
            "key1": {"key11": "value11",
 
1971
                     "key12": "value12"},
 
1972
            "key2": {"key21": "value21",
 
1973
                     "key22": "value22"},
 
1974
        }
 
1975
        self.assertEqual(expected_method_return, ret)
 
1976
        for key, value in ret.items():
 
1977
            self.assertIsInstance(value, dict)
 
1978
            self.assertEqual(expected_method_return[key], value)
 
1979
            self.assertNotIsInstance(key, ObjectPath)
 
1980
            for inner_key, inner_value in value.items():
 
1981
                self.assertIsInstance(value, dict)
 
1982
                self.assertEqual(
 
1983
                    expected_method_return[key][inner_key],
 
1984
                    inner_value)
 
1985
                self.assertNotIsInstance(key, ObjectPath)
 
1986
 
 
1987
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
 
1988
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1989
        def func():
 
1990
            return method_return
 
1991
        fake_ravel = self.fake_ravel_func(func)
 
1992
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1993
        method_return = {
 
1994
            ObjectPath("key1"): {
 
1995
                ObjectPath("key2"): {
 
1996
                    ObjectPath("key3"): ObjectPath("value"),
 
1997
                },
 
1998
            },
 
1999
        }
 
2000
        ret = self.call_method(bus, "methodname", "busname",
 
2001
                               "objectpath", "interface")
 
2002
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
 
2003
        self.assertEqual(expected_method_return, ret)
 
2004
        self.assertIsInstance(ret, dict)
 
2005
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
 
2006
        self.assertIsInstance(ret["key1"], dict)
 
2007
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
 
2008
                                 ObjectPath)
 
2009
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
2010
        self.assertNotIsInstance(
 
2011
            next(iter(ret["key1"]["key2"].keys())),
 
2012
            ObjectPath)
 
2013
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
 
2014
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
2015
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
2016
 
 
2017
    def test_call_method_handles_exception(self):
 
2018
        def func():
 
2019
            raise self.dummy_dbussy.DBusError()
 
2020
 
 
2021
        fake_ravel = self.fake_ravel_func(func)
 
2022
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2023
 
 
2024
        with self.assertRaises(dbus.Error) as e:
 
2025
            self.call_method(bus, "methodname", "busname",
 
2026
                             "objectpath", "interface")
 
2027
 
 
2028
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
2029
 
 
2030
    def test_get_object_converts_to_correct_exception(self):
 
2031
        class fake_ravel_raises_exception_on_connect:
 
2032
            @staticmethod
 
2033
            def system_bus():
 
2034
                class Bus:
 
2035
                    @staticmethod
 
2036
                    def __getitem__(key):
 
2037
                        if key == "objectpath":
 
2038
                            raise self.dummy_dbussy.DBusError()
 
2039
                        raise Exception(key)
 
2040
                return {"busname": Bus()}
 
2041
        def func():
 
2042
            raise self.dummy_dbussy.DBusError()
 
2043
        bus = dbussy_adapter.SystemBus(
 
2044
            self.dummy_dbussy,
 
2045
            fake_ravel_raises_exception_on_connect)
 
2046
        with self.assertRaises(dbus.ConnectFailed):
 
2047
            self.call_method(bus, "methodname", "busname",
 
2048
                             "objectpath", "interface")
 
2049
 
 
2050
 
1773
2051
class Test_commands_from_options(unittest.TestCase):
1774
2052
 
1775
2053
    def setUp(self):
2001
2279
 
2002
2280
 
2003
2281
    def test_manual_page_example_1(self):
2004
 
        self.assert_command_from_args("--verbose".split(),
 
2282
        self.assert_command_from_args("",
2005
2283
                                      command.PrintTable,
2006
2284
                                      clients=[],
2007
 
                                      verbose=True)
 
2285
                                      verbose=False)
2008
2286
 
2009
2287
    def test_manual_page_example_2(self):
2010
2288
        self.assert_command_from_args(
2149
2427
        busname = "se.recompile.Mandos"
2150
2428
        client_interface = "se.recompile.Mandos.Client"
2151
2429
        command.Approve().run(self.bus.clients, self.bus)
 
2430
        self.assertTrue(self.bus.clients)
2152
2431
        for clientpath in self.bus.clients:
2153
2432
            self.assertIn(("Approve", busname, clientpath,
2154
2433
                           client_interface, (True,)), self.bus.calls)
2157
2436
        busname = "se.recompile.Mandos"
2158
2437
        client_interface = "se.recompile.Mandos.Client"
2159
2438
        command.Deny().run(self.bus.clients, self.bus)
 
2439
        self.assertTrue(self.bus.clients)
2160
2440
        for clientpath in self.bus.clients:
2161
2441
            self.assertIn(("Approve", busname, clientpath,
2162
2442
                           client_interface, (False,)),
2163
2443
                          self.bus.calls)
2164
2444
 
2165
2445
    def test_Remove(self):
 
2446
        busname = "se.recompile.Mandos"
 
2447
        server_path = "/"
 
2448
        server_interface = "se.recompile.Mandos"
 
2449
        orig_clients = self.bus.clients.copy()
2166
2450
        command.Remove().run(self.bus.clients, self.bus)
2167
 
        for clientpath in self.bus.clients:
2168
 
            self.assertIn(("RemoveClient", dbus_busname,
2169
 
                           dbus_server_path, dbus_server_interface,
 
2451
        self.assertFalse(self.bus.clients)
 
2452
        for clientpath in orig_clients:
 
2453
            self.assertIn(("RemoveClient", busname,
 
2454
                           server_path, server_interface,
2170
2455
                           (clientpath,)), self.bus.calls)
2171
2456
 
2172
2457
    expected_json = {
2374
2659
        else:
2375
2660
            cmd_args = [() for x in range(len(self.values_to_get))]
2376
2661
            values_to_get = self.values_to_get
 
2662
        self.assertTrue(values_to_get)
2377
2663
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2378
2664
            for clientpath in self.bus.clients:
2379
2665
                self.bus.clients[clientpath][self.propname] = (
2380
2666
                    Unique())
2381
2667
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
 
2668
            self.assertTrue(self.bus.clients)
2382
2669
            for clientpath in self.bus.clients:
2383
2670
                value = (self.bus.clients[clientpath]
2384
2671
                         [self.propname])
2443
2730
class TestSetSecretCmd(TestPropertySetterCmd):
2444
2731
    command = command.SetSecret
2445
2732
    propname = "Secret"
2446
 
    values_to_set = [io.BytesIO(b""),
2447
 
                     io.BytesIO(b"secret\0xyzzy\nbar")]
2448
 
    values_to_get = [f.getvalue() for f in values_to_set]
 
2733
    def __init__(self, *args, **kwargs):
 
2734
        self.values_to_set = [io.BytesIO(b""),
 
2735
                              io.BytesIO(b"secret\0xyzzy\nbar")]
 
2736
        self.values_to_get = [f.getvalue() for f in
 
2737
                              self.values_to_set]
 
2738
        super(TestSetSecretCmd, self).__init__(*args, **kwargs)
2449
2739
 
2450
2740
 
2451
2741
class TestSetTimeoutCmd(TestPropertySetterCmd):
2504
2794
 
2505
2795
 
2506
2796
 
2507
 
def should_only_run_tests():
 
2797
def parse_test_args():
 
2798
    # type: () -> argparse.Namespace
2508
2799
    parser = argparse.ArgumentParser(add_help=False)
2509
 
    parser.add_argument("--check", action='store_true')
 
2800
    parser.add_argument("--check", action="store_true")
 
2801
    parser.add_argument("--prefix", )
2510
2802
    args, unknown_args = parser.parse_known_args()
2511
 
    run_tests = args.check
2512
 
    if run_tests:
2513
 
        # Remove --check argument from sys.argv
 
2803
    if args.check:
 
2804
        # Remove test options from sys.argv
2514
2805
        sys.argv[1:] = unknown_args
2515
 
    return run_tests
 
2806
    return args
2516
2807
 
2517
2808
# Add all tests from doctest strings
2518
2809
def load_tests(loader, tests, none):
2521
2812
    return tests
2522
2813
 
2523
2814
if __name__ == "__main__":
 
2815
    options = parse_test_args()
2524
2816
    try:
2525
 
        if should_only_run_tests():
2526
 
            # Call using ./tdd-python-script --check [--verbose]
2527
 
            unittest.main()
 
2817
        if options.check:
 
2818
            extra_test_prefix = options.prefix
 
2819
            if extra_test_prefix is not None:
 
2820
                if not (unittest.main(argv=[""], exit=False)
 
2821
                        .result.wasSuccessful()):
 
2822
                    sys.exit(1)
 
2823
                class ExtraTestLoader(unittest.TestLoader):
 
2824
                    testMethodPrefix = extra_test_prefix
 
2825
                # Call using ./scriptname --check [--verbose]
 
2826
                unittest.main(argv=[""], testLoader=ExtraTestLoader())
 
2827
            else:
 
2828
                unittest.main(argv=[""])
2528
2829
        else:
2529
2830
            main()
2530
2831
    finally:
2531
2832
        logging.shutdown()
 
2833
 
 
2834
# Local Variables:
 
2835
# run-tests:
 
2836
# (lambda (&optional extra)
 
2837
#   (if (not (funcall run-tests-in-test-buffer default-directory
 
2838
#             extra))
 
2839
#       (funcall show-test-buffer-in-test-window)
 
2840
#     (funcall remove-test-window)
 
2841
#     (if extra (message "Extra tests run successfully!"))))
 
2842
# run-tests-in-test-buffer:
 
2843
# (lambda (dir &optional extra)
 
2844
#   (with-current-buffer (get-buffer-create "*Test*")
 
2845
#     (setq buffer-read-only nil
 
2846
#           default-directory dir)
 
2847
#     (erase-buffer)
 
2848
#     (compilation-mode))
 
2849
#   (let ((process-result
 
2850
#          (let ((inhibit-read-only t))
 
2851
#            (process-file-shell-command
 
2852
#             (funcall get-command-line extra) nil "*Test*"))))
 
2853
#     (and (numberp process-result)
 
2854
#          (= process-result 0))))
 
2855
# get-command-line:
 
2856
# (lambda (&optional extra)
 
2857
#   (let ((quoted-script
 
2858
#          (shell-quote-argument (funcall get-script-name))))
 
2859
#     (format
 
2860
#      (concat "%s --check" (if extra " --prefix=atest" ""))
 
2861
#      quoted-script)))
 
2862
# get-script-name:
 
2863
# (lambda ()
 
2864
#   (if (fboundp 'file-local-name)
 
2865
#       (file-local-name (buffer-file-name))
 
2866
#     (or (file-remote-p (buffer-file-name) 'localname)
 
2867
#         (buffer-file-name))))
 
2868
# remove-test-window:
 
2869
# (lambda ()
 
2870
#   (let ((test-window (get-buffer-window "*Test*")))
 
2871
#     (if test-window (delete-window test-window))))
 
2872
# show-test-buffer-in-test-window:
 
2873
# (lambda ()
 
2874
#   (when (not (get-buffer-window-list "*Test*"))
 
2875
#     (setq next-error-last-buffer (get-buffer "*Test*"))
 
2876
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
 
2877
#            (display-buffer-overriding-action
 
2878
#             `((display-buffer-in-side-window) (side . ,side)
 
2879
#               (window-height . fit-window-to-buffer)
 
2880
#               (window-width . fit-window-to-buffer))))
 
2881
#       (display-buffer "*Test*"))))
 
2882
# eval:
 
2883
# (progn
 
2884
#   (let* ((run-extra-tests (lambda () (interactive)
 
2885
#                             (funcall run-tests t)))
 
2886
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
 
2887
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
 
2888
#     (setq minor-mode-overriding-map-alist
 
2889
#           (cons `(run-tests . ,outer-keymap)
 
2890
#                 minor-mode-overriding-map-alist)))
 
2891
#   (add-hook 'after-save-hook run-tests 90 t))
 
2892
# End: