/mandos/trunk : revision 1215

1

#!/usr/bin/python3 -bbI

2

# -*- coding: utf-8; lexical-binding: t -*-

3

#

4

# Mandos Control - Control or query the Mandos server

5

#

6

7

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

7

8

#

9

# This file is part of Mandos.

10

#

23

#

24

# Contact the authors at <mandos@recompile.se>.

25

#

26

27

from __future__ import (division, absolute_import, print_function,

27

28

unicode_literals)

28

29

32

33

pass

33

34

35

import sys

35

import unittest

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

53

class gi:

54

"""Dummy gi module, for the tests"""

77

import warnings

78

warnings.simplefilter("default")

79

80

log = logging.getLogger(os.path.basename(sys.argv[0]))

80

log = logging.getLogger(sys.argv[0])

81

logging.basicConfig(level="INFO", # Show info level messages

82

format="%(message)s") # Show basic log messages

83

89

90

locale.setlocale(locale.LC_ALL, "")

91

92

version = "1.8.14"

92

version = "1.8.11"

93

94

95

def main():

102

clientnames = options.client

103

104

if options.debug:

105

logging.getLogger("").setLevel(logging.DEBUG)

105

log.setLevel(logging.DEBUG)

106

107

if dbussy is not None and ravel is not None:

108

bus = dbussy_adapter.CachingBus(dbussy, ravel)

256

return rfc3339_duration_to_delta(interval)

257

except ValueError as e:

258

log.warning("%s - Parsing as pre-1.6.1 interval instead",

259

" ".join(e.args))

259

' '.join(e.args))

260

return parse_pre_1_6_1_interval(interval)

261

262

395

"""Parse an interval string as documented by Mandos before 1.6.1,

396

and return a datetime.timedelta

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

True

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

True

415

416

"""

726

with self.convert_exception(dbus.Error):

727

value = method(*args)

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

if value:

731

return self.type_filter(value[0])

732

738

739

def type_filter(self, value):

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 ((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: