/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 at recompile
  • Date: 2020-02-08 14:01:38 UTC
  • Revision ID: teddy@recompile.se-20200208140138-h3ett0amdelfpt90
mandos: If possible, use shlex.quote() instead of re.escape()

* mandos: Make sure shlex.quote() exists (falling back to assigning
  re.escape to it).
  (Client.start_checker): Use shlex.quote() instead of re.escape().

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
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-2020 Teddy Hogeborn
7
 
# Copyright © 2008-2020 Björn Påhlsson
 
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
#
 
4
# Mandos Monitor - Control and monitor the Mandos server
 
5
#
 
6
# Copyright © 2008-2019 Teddy Hogeborn
 
7
# Copyright © 2008-2019 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
 
26
27
from __future__ import (division, absolute_import, print_function,
27
28
                        unicode_literals)
28
29
 
32
33
    pass
33
34
 
34
35
import sys
35
 
import unittest
36
36
import argparse
37
 
import logging
38
 
import os
39
37
import locale
40
38
import datetime
41
39
import re
 
40
import os
42
41
import collections
43
42
import json
 
43
import unittest
 
44
import logging
44
45
import io
45
46
import tempfile
46
47
import contextlib
48
49
if sys.version_info.major == 2:
49
50
    __metaclass__ = type
50
51
    str = unicode
51
 
    input = raw_input
52
52
 
53
53
class gi:
54
54
    """Dummy gi module, for the tests"""
77
77
    import warnings
78
78
    warnings.simplefilter("default")
79
79
 
80
 
log = logging.getLogger(os.path.basename(sys.argv[0]))
 
80
log = logging.getLogger(sys.argv[0])
81
81
logging.basicConfig(level="INFO", # Show info level messages
82
82
                    format="%(message)s") # Show basic log messages
83
83
 
89
89
 
90
90
locale.setlocale(locale.LC_ALL, "")
91
91
 
92
 
version = "1.8.14"
 
92
version = "1.8.9"
93
93
 
94
94
 
95
95
def main():
102
102
    clientnames = options.client
103
103
 
104
104
    if options.debug:
105
 
        logging.getLogger("").setLevel(logging.DEBUG)
 
105
        log.setLevel(logging.DEBUG)
106
106
 
107
107
    if dbussy is not None and ravel is not None:
108
108
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
256
256
        return rfc3339_duration_to_delta(interval)
257
257
    except ValueError as e:
258
258
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
259
 
                    " ".join(e.args))
 
259
                    ' '.join(e.args))
260
260
    return parse_pre_1_6_1_interval(interval)
261
261
 
262
262
 
395
395
    """Parse an interval string as documented by Mandos before 1.6.1,
396
396
    and return a datetime.timedelta
397
397
 
398
 
    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
399
 
    True
400
 
    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
401
 
    True
402
 
    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
403
 
    True
404
 
    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
405
 
    True
406
 
    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
407
 
    True
408
 
    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
409
 
    True
410
 
    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
 
398
    >>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
 
399
    True
 
400
    >>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
 
401
    True
 
402
    >>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
 
403
    True
 
404
    >>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
 
405
    True
 
406
    >>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
 
407
    True
 
408
    >>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
 
409
    True
 
410
    >>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
411
411
    True
412
412
    >>> # Ignore unknown characters, allow any order and repetitions
413
 
    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") == datetime.timedelta(2, 480, 18000)
 
413
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
414
414
    True
415
415
 
416
416
    """
726
726
            with self.convert_exception(dbus.Error):
727
727
                value =  method(*args)
728
728
            # DBussy returns values either as an empty list or as a
729
 
            # list of one element with the return value
 
729
            # tuple: (signature, value)
730
730
            if value:
731
731
                return self.type_filter(value[0])
732
732
 
738
738
 
739
739
        def type_filter(self, value):
740
740
            """Convert the most bothersome types to Python types"""
741
 
            # A D-Bus Variant value is represented as the Python type
742
 
            # Tuple[dbussy.DBUS.Signature, Any]
743
741
            if isinstance(value, tuple):
744
742
                if (len(value) == 2
745
743
                    and isinstance(value[0],
876
874
                    {key: properties[key]
877
875
                     for key in self.all_keywords}
878
876
                    for properties in clients.values()}
879
 
            print(json.dumps(data, indent=4, separators=(",", ": ")))
 
877
            print(json.dumps(data, indent=4, separators=(',', ': ')))
880
878
 
881
879
 
882
880
    class PrintTable(Output):
2439
2437
        busname = "se.recompile.Mandos"
2440
2438
        client_interface = "se.recompile.Mandos.Client"
2441
2439
        command.Approve().run(self.bus.clients, self.bus)
2442
 
        self.assertTrue(self.bus.clients)
2443
2440
        for clientpath in self.bus.clients:
2444
2441
            self.assertIn(("Approve", busname, clientpath,
2445
2442
                           client_interface, (True,)), self.bus.calls)
2448
2445
        busname = "se.recompile.Mandos"
2449
2446
        client_interface = "se.recompile.Mandos.Client"
2450
2447
        command.Deny().run(self.bus.clients, self.bus)
2451
 
        self.assertTrue(self.bus.clients)
2452
2448
        for clientpath in self.bus.clients:
2453
2449
            self.assertIn(("Approve", busname, clientpath,
2454
2450
                           client_interface, (False,)),
2455
2451
                          self.bus.calls)
2456
2452
 
2457
2453
    def test_Remove(self):
2458
 
        busname = "se.recompile.Mandos"
2459
 
        server_path = "/"
2460
 
        server_interface = "se.recompile.Mandos"
2461
 
        orig_clients = self.bus.clients.copy()
2462
2454
        command.Remove().run(self.bus.clients, self.bus)
2463
 
        self.assertFalse(self.bus.clients)
2464
 
        for clientpath in orig_clients:
2465
 
            self.assertIn(("RemoveClient", busname,
2466
 
                           server_path, server_interface,
 
2455
        for clientpath in self.bus.clients:
 
2456
            self.assertIn(("RemoveClient", dbus_busname,
 
2457
                           dbus_server_path, dbus_server_interface,
2467
2458
                           (clientpath,)), self.bus.calls)
2468
2459
 
2469
2460
    expected_json = {
2671
2662
        else:
2672
2663
            cmd_args = [() for x in range(len(self.values_to_get))]
2673
2664
            values_to_get = self.values_to_get
2674
 
        self.assertTrue(values_to_get)
2675
2665
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2676
2666
            for clientpath in self.bus.clients:
2677
2667
                self.bus.clients[clientpath][self.propname] = (
2678
2668
                    Unique())
2679
2669
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
2680
 
            self.assertTrue(self.bus.clients)
2681
2670
            for clientpath in self.bus.clients:
2682
2671
                value = (self.bus.clients[clientpath]
2683
2672
                         [self.propname])
2742
2731
class TestSetSecretCmd(TestPropertySetterCmd):
2743
2732
    command = command.SetSecret
2744
2733
    propname = "Secret"
2745
 
    def __init__(self, *args, **kwargs):
2746
 
        self.values_to_set = [io.BytesIO(b""),
2747
 
                              io.BytesIO(b"secret\0xyzzy\nbar")]
2748
 
        self.values_to_get = [f.getvalue() for f in
2749
 
                              self.values_to_set]
2750
 
        super(TestSetSecretCmd, self).__init__(*args, **kwargs)
 
2734
    values_to_set = [io.BytesIO(b""),
 
2735
                     io.BytesIO(b"secret\0xyzzy\nbar")]
 
2736
    values_to_get = [f.getvalue() for f in values_to_set]
2751
2737
 
2752
2738
 
2753
2739
class TestSetTimeoutCmd(TestPropertySetterCmd):
2806
2792
 
2807
2793
 
2808
2794
 
2809
 
def parse_test_args():
2810
 
    # type: () -> argparse.Namespace
 
2795
def should_only_run_tests():
2811
2796
    parser = argparse.ArgumentParser(add_help=False)
2812
 
    parser.add_argument("--check", action="store_true")
2813
 
    parser.add_argument("--prefix", )
 
2797
    parser.add_argument("--check", action='store_true')
2814
2798
    args, unknown_args = parser.parse_known_args()
2815
 
    if args.check:
2816
 
        # Remove test options from sys.argv
 
2799
    run_tests = args.check
 
2800
    if run_tests:
 
2801
        # Remove --check argument from sys.argv
2817
2802
        sys.argv[1:] = unknown_args
2818
 
    return args
 
2803
    return run_tests
2819
2804
 
2820
2805
# Add all tests from doctest strings
2821
2806
def load_tests(loader, tests, none):
2824
2809
    return tests
2825
2810
 
2826
2811
if __name__ == "__main__":
2827
 
    options = parse_test_args()
2828
2812
    try:
2829
 
        if options.check:
2830
 
            extra_test_prefix = options.prefix
2831
 
            if extra_test_prefix is not None:
2832
 
                if not (unittest.main(argv=[""], exit=False)
2833
 
                        .result.wasSuccessful()):
2834
 
                    sys.exit(1)
2835
 
                class ExtraTestLoader(unittest.TestLoader):
2836
 
                    testMethodPrefix = extra_test_prefix
2837
 
                # Call using ./scriptname --check [--verbose]
2838
 
                unittest.main(argv=[""], testLoader=ExtraTestLoader())
2839
 
            else:
2840
 
                unittest.main(argv=[""])
 
2813
        if should_only_run_tests():
 
2814
            # Call using ./tdd-python-script --check [--verbose]
 
2815
            unittest.main()
2841
2816
        else:
2842
2817
            main()
2843
2818
    finally:
2844
2819
        logging.shutdown()
2845
 
 
2846
 
# Local Variables:
2847
 
# run-tests:
2848
 
# (lambda (&optional extra)
2849
 
#   (if (not (funcall run-tests-in-test-buffer default-directory
2850
 
#             extra))
2851
 
#       (funcall show-test-buffer-in-test-window)
2852
 
#     (funcall remove-test-window)
2853
 
#     (if extra (message "Extra tests run successfully!"))))
2854
 
# run-tests-in-test-buffer:
2855
 
# (lambda (dir &optional extra)
2856
 
#   (with-current-buffer (get-buffer-create "*Test*")
2857
 
#     (setq buffer-read-only nil
2858
 
#           default-directory dir)
2859
 
#     (erase-buffer)
2860
 
#     (compilation-mode))
2861
 
#   (let ((inhibit-read-only t))
2862
 
#     (= (process-file-shell-command
2863
 
#         (funcall get-command-line extra)
2864
 
#         nil "*Test*") 0)))
2865
 
# get-command-line:
2866
 
# (lambda (&optional extra)
2867
 
#   (let ((quoted-script
2868
 
#          (shell-quote-argument (funcall get-script-name))))
2869
 
#     (format
2870
 
#      (concat "%s --check" (if extra " --prefix=atest" ""))
2871
 
#      quoted-script)))
2872
 
# get-script-name:
2873
 
# (lambda ()
2874
 
#   (if (fboundp 'file-local-name)
2875
 
#       (file-local-name (buffer-file-name))
2876
 
#     (or (file-remote-p (buffer-file-name) 'localname)
2877
 
#         (buffer-file-name))))
2878
 
# remove-test-window:
2879
 
# (lambda ()
2880
 
#   (let ((test-window (get-buffer-window "*Test*")))
2881
 
#     (if test-window (delete-window test-window))))
2882
 
# show-test-buffer-in-test-window:
2883
 
# (lambda ()
2884
 
#   (when (not (get-buffer-window-list "*Test*"))
2885
 
#     (setq next-error-last-buffer (get-buffer "*Test*"))
2886
 
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
2887
 
#            (display-buffer-overriding-action
2888
 
#             `((display-buffer-in-side-window) (side . ,side)
2889
 
#               (window-height . fit-window-to-buffer)
2890
 
#               (window-width . fit-window-to-buffer))))
2891
 
#       (display-buffer "*Test*"))))
2892
 
# eval:
2893
 
# (progn
2894
 
#   (let* ((run-extra-tests (lambda () (interactive)
2895
 
#                             (funcall run-tests t)))
2896
 
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
2897
 
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
2898
 
#     (setq minor-mode-overriding-map-alist
2899
 
#           (cons `(run-tests . ,outer-keymap)
2900
 
#                 minor-mode-overriding-map-alist)))
2901
 
#   (add-hook 'after-save-hook run-tests 90 t))
2902
 
# End: