/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-11-29 22:22:45 UTC
  • Revision ID: teddy@recompile.se-20201129222245-9xyto95047gnwwm6
Fix debconf template syntax

* debian/mandos-client.templates: Rename "_description" header to
  "Description".
* debian/mandos.templates: - '' -

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 -*-
 
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 Control - Control or query the Mandos server
5
5
#
6
 
# Copyright © 2008-2022 Teddy Hogeborn
7
 
# Copyright © 2008-2022 Björn Påhlsson
 
6
# Copyright © 2008-2020 Teddy Hogeborn
 
7
# Copyright © 2008-2020 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.16"
 
92
version = "1.8.12"
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
    """
876
876
                    {key: properties[key]
877
877
                     for key in self.all_keywords}
878
878
                    for properties in clients.values()}
879
 
            print(json.dumps(data, indent=4, separators=(",", ": ")))
 
879
            print(json.dumps(data, indent=4, separators=(',', ': ')))
880
880
 
881
881
 
882
882
    class PrintTable(Output):
2439
2439
        busname = "se.recompile.Mandos"
2440
2440
        client_interface = "se.recompile.Mandos.Client"
2441
2441
        command.Approve().run(self.bus.clients, self.bus)
2442
 
        self.assertTrue(self.bus.clients)
2443
2442
        for clientpath in self.bus.clients:
2444
2443
            self.assertIn(("Approve", busname, clientpath,
2445
2444
                           client_interface, (True,)), self.bus.calls)
2448
2447
        busname = "se.recompile.Mandos"
2449
2448
        client_interface = "se.recompile.Mandos.Client"
2450
2449
        command.Deny().run(self.bus.clients, self.bus)
2451
 
        self.assertTrue(self.bus.clients)
2452
2450
        for clientpath in self.bus.clients:
2453
2451
            self.assertIn(("Approve", busname, clientpath,
2454
2452
                           client_interface, (False,)),
2455
2453
                          self.bus.calls)
2456
2454
 
2457
2455
    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
2456
        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,
 
2457
        for clientpath in self.bus.clients:
 
2458
            self.assertIn(("RemoveClient", dbus_busname,
 
2459
                           dbus_server_path, dbus_server_interface,
2467
2460
                           (clientpath,)), self.bus.calls)
2468
2461
 
2469
2462
    expected_json = {
2671
2664
        else:
2672
2665
            cmd_args = [() for x in range(len(self.values_to_get))]
2673
2666
            values_to_get = self.values_to_get
2674
 
        self.assertTrue(values_to_get)
2675
2667
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2676
2668
            for clientpath in self.bus.clients:
2677
2669
                self.bus.clients[clientpath][self.propname] = (
2678
2670
                    Unique())
2679
2671
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
2680
 
            self.assertTrue(self.bus.clients)
2681
2672
            for clientpath in self.bus.clients:
2682
2673
                value = (self.bus.clients[clientpath]
2683
2674
                         [self.propname])
2742
2733
class TestSetSecretCmd(TestPropertySetterCmd):
2743
2734
    command = command.SetSecret
2744
2735
    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)
 
2736
    values_to_set = [io.BytesIO(b""),
 
2737
                     io.BytesIO(b"secret\0xyzzy\nbar")]
 
2738
    values_to_get = [f.getvalue() for f in values_to_set]
2751
2739
 
2752
2740
 
2753
2741
class TestSetTimeoutCmd(TestPropertySetterCmd):
2806
2794
 
2807
2795
 
2808
2796
 
2809
 
def parse_test_args():
2810
 
    # type: () -> argparse.Namespace
 
2797
def should_only_run_tests():
2811
2798
    parser = argparse.ArgumentParser(add_help=False)
2812
 
    parser.add_argument("--check", action="store_true")
2813
 
    parser.add_argument("--prefix", )
 
2799
    parser.add_argument("--check", action='store_true')
2814
2800
    args, unknown_args = parser.parse_known_args()
2815
 
    if args.check:
2816
 
        # Remove test options from sys.argv
 
2801
    run_tests = args.check
 
2802
    if run_tests:
 
2803
        # Remove --check argument from sys.argv
2817
2804
        sys.argv[1:] = unknown_args
2818
 
    return args
 
2805
    return run_tests
2819
2806
 
2820
2807
# Add all tests from doctest strings
2821
2808
def load_tests(loader, tests, none):
2824
2811
    return tests
2825
2812
 
2826
2813
if __name__ == "__main__":
2827
 
    options = parse_test_args()
2828
2814
    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=[""])
 
2815
        if should_only_run_tests():
 
2816
            # Call using ./tdd-python-script --check [--verbose]
 
2817
            unittest.main()
2841
2818
        else:
2842
2819
            main()
2843
2820
    finally:
2844
2821
        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 ((process-result
2862
 
#          (let ((inhibit-read-only t))
2863
 
#            (process-file-shell-command
2864
 
#             (funcall get-command-line extra) nil "*Test*"))))
2865
 
#     (and (numberp process-result)
2866
 
#          (= process-result 0))))
2867
 
# get-command-line:
2868
 
# (lambda (&optional extra)
2869
 
#   (let ((quoted-script
2870
 
#          (shell-quote-argument (funcall get-script-name))))
2871
 
#     (format
2872
 
#      (concat "%s --check" (if extra " --prefix=atest" ""))
2873
 
#      quoted-script)))
2874
 
# get-script-name:
2875
 
# (lambda ()
2876
 
#   (if (fboundp 'file-local-name)
2877
 
#       (file-local-name (buffer-file-name))
2878
 
#     (or (file-remote-p (buffer-file-name) 'localname)
2879
 
#         (buffer-file-name))))
2880
 
# remove-test-window:
2881
 
# (lambda ()
2882
 
#   (let ((test-window (get-buffer-window "*Test*")))
2883
 
#     (if test-window (delete-window test-window))))
2884
 
# show-test-buffer-in-test-window:
2885
 
# (lambda ()
2886
 
#   (when (not (get-buffer-window-list "*Test*"))
2887
 
#     (setq next-error-last-buffer (get-buffer "*Test*"))
2888
 
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
2889
 
#            (display-buffer-overriding-action
2890
 
#             `((display-buffer-in-side-window) (side . ,side)
2891
 
#               (window-height . fit-window-to-buffer)
2892
 
#               (window-width . fit-window-to-buffer))))
2893
 
#       (display-buffer "*Test*"))))
2894
 
# eval:
2895
 
# (progn
2896
 
#   (let* ((run-extra-tests (lambda () (interactive)
2897
 
#                             (funcall run-tests t)))
2898
 
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
2899
 
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
2900
 
#     (setq minor-mode-overriding-map-alist
2901
 
#           (cons `(run-tests . ,outer-keymap)
2902
 
#                 minor-mode-overriding-map-alist)))
2903
 
#   (add-hook 'after-save-hook run-tests 90 t))
2904
 
# End: