/mandos/trunk

#!/usr/bin/python3 -bI

# -*- mode: python; 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 -*-

# Copyright © 2008-2020 Teddy Hogeborn

# Copyright © 2008-2020 Björn Påhlsson

import unittest

import random

import shlex

import gi

if sys.version_info.major == 2:

    __metaclass__ = type

    str = unicode

 

# Add collections.abc.Callable if it does not exist

try:

    collections.abc.Callable

except AttributeError:

    class abc:

        Callable = collections.Callable

    collections.abc = abc

    del abc

 

# Add shlex.quote if it does not exist

try:

    shlex.quote

except AttributeError:

    shlex.quote = re.escape

 

# Show warnings by default

if not sys.warnoptions:

    import warnings

    warnings.simplefilter("default")

 

if sys.version_info < (3, 2):

    configparser.Configparser = configparser.SafeConfigParser

version = "1.8.14"

logging.captureWarnings(True)   # Show warnings via the logging system

class PGPEngine:

                if name == b"gpg":

        if self.gpg == b"gpg" or self.gpg.endswith(b"/gpg"):

class avahi:

class AvahiService:

        ret = super(AvahiServiceToSyslog, self).rename(*args,

                                                       **kwargs)

class gnutls:

    class _session_int(ctypes.Structure):

    session_t = ctypes.POINTER(_session_int)

    class _openpgp_crt_int(ctypes.Structure):

    openpgp_crt_t = ctypes.POINTER(_openpgp_crt_int)

                message = gnutls.strerror(code).decode(

                    "utf-8", errors="replace")

    class PointerTo:

        def __init__(self, cls):

            self.cls = cls

 

        def from_param(self, obj):

            if not isinstance(obj, self.cls):

                raise TypeError("Not of type {}: {!r}"

                                .format(self.cls.__name__, obj))

            return ctypes.byref(obj.from_param(obj))

 

    class CastToVoidPointer:

        def __init__(self, cls):

            self.cls = cls

 

        def from_param(self, obj):

            if not isinstance(obj, self.cls):

                raise TypeError("Not of type {}: {!r}"

                                .format(self.cls.__name__, obj))

            return ctypes.cast(obj.from_param(obj), ctypes.c_void_p)

 

    class With_from_param:

        @classmethod

        def from_param(cls, obj):

            return obj._as_parameter_

 

    class Credentials(With_from_param):

            self._as_parameter_ = gnutls.certificate_credentials_t()

            gnutls.certificate_allocate_credentials(self)

            gnutls.certificate_free_credentials(self)

    class ClientSession(With_from_param):

            self._as_parameter_ = gnutls.session_t()

            gnutls.init(self, gnutls_flags)

            gnutls.set_default_priority(self)

            gnutls.transport_set_ptr(self, socket.fileno())

            gnutls.handshake_set_private_extensions(self, True)

            gnutls.credentials_set(self, credentials.type,

                                   credentials)

            gnutls.deinit(self)

            return gnutls.handshake(self)

                data_len -= gnutls.record_send(self, data[-data_len:],

            return gnutls.bye(self, gnutls.SHUT_RDWR)

        if result >= gnutls.E_SUCCESS:

    def _retry_on_error(result, func, arguments,

                        _error_code=_error_code):

        while result < gnutls.E_SUCCESS:

    priority_set_direct.argtypes = [ClientSession, ctypes.c_char_p,

    init.argtypes = [PointerTo(ClientSession), ctypes.c_int]

    set_default_priority.argtypes = [ClientSession]

    record_send.argtypes = [ClientSession, ctypes.c_void_p,

        PointerTo(Credentials)]

    certificate_free_credentials.argtypes = [Credentials]

    handshake_set_private_extensions.argtypes = [ClientSession,

    credentials_set.argtypes = [ClientSession, credentials_type_t,

                                CastToVoidPointer(Credentials)]

    certificate_type_get.argtypes = [ClientSession]

    certificate_get_peers.argtypes = [ClientSession,

    deinit.argtypes = [ClientSession]

    handshake.argtypes = [ClientSession]

    handshake.restype = ctypes.c_int

    transport_set_ptr.argtypes = [ClientSession, transport_ptr_t]

    bye.argtypes = [ClientSession, close_request_t]

    bye.restype = ctypes.c_int

        # All the function declarations below are from

        # gnutls/abstract.h

        # All the function declarations below are from

        # gnutls/openpgp.h

        openpgp_crt_verify_self = \

            _library.gnutls_openpgp_crt_verify_self

        openpgp_crt_verify_self.argtypes = [

            openpgp_crt_t,

            ctypes.c_uint,

            ctypes.POINTER(ctypes.c_uint),

        ]

        certificate_type_get2.argtypes = [ClientSession, ctypes.c_int]

class Client:

            random.randrange(int(self.interval.total_seconds() * 1000

                                 + 1)),

        if self.checker is not None:

            self.checker.join()

                attr: shlex.quote(str(getattr(self, attr)))

                GLib.IOChannel.unix_new(pipe[0].fileno()),

                GLib.PRIORITY_DEFAULT, GLib.IO_IN,

            value = dbus.ByteArray(bytes(value))

class ProxyClient:

            gnutls.priority_set_direct(session,

                                       priority.encode("utf-8"), None)

                    logger.debug("Key ID: %s",

                                 key_id.decode("utf-8",

                                               errors="replace"))

            cert_type = gnutls.certificate_type_get2(

                session, gnutls.CTYPE_PEERS)

            cert_type = gnutls.certificate_type_get(session)

                     (session, ctypes.byref(list_size)))

        gnutls.pubkey_get_key_id(

            pubkey,

            gnutls.KEYID_USE_SHA256,

            ctypes.cast(ctypes.byref(buf),

                        ctypes.POINTER(ctypes.c_ubyte)),

            ctypes.byref(buf_len))

class MultiprocessingMixIn:

class MultiprocessingMixInWithPipe(MultiprocessingMixIn):

                     socketserver.TCPServer):

            GLib.IOChannel.unix_new(parent_pipe.fileno()),

            GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP,

                if key_id == ("E3B0C44298FC1C149AFBF4C8996FB924"

                              "27AE41E4649B934CA495991B7852B855"):

                GLib.IOChannel.unix_new(parent_pipe.fileno()),

                GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP,

                          collections.abc.Callable):

    >>> timedelta = datetime.timedelta

    >>> rfc3339_duration_to_delta("P7D") == timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT60S") == timedelta(0, 60)

    True

    >>> rfc3339_duration_to_delta("PT60M") == timedelta(0, 3600)

    True

    >>> rfc3339_duration_to_delta("PT24H") == timedelta(1)

    True

    >>> rfc3339_duration_to_delta("P1W") == timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT5M30S") == timedelta(0, 330)

    True

    >>> rfc3339_duration_to_delta("P1DT3M20S") == timedelta(1, 200)

    True

    >>> del timedelta

    >>> string_to_delta('7d') == datetime.timedelta(7)

    True

    >>> string_to_delta('60s') == datetime.timedelta(0, 60)

    True

    >>> string_to_delta('60m') == datetime.timedelta(0, 3600)

    True

    >>> string_to_delta('24h') == datetime.timedelta(1)

    True

    >>> string_to_delta('1w') == datetime.timedelta(7)

    True

    >>> string_to_delta('5m 30s') == datetime.timedelta(0, 330)

    True

    server_config = configparser.ConfigParser(server_defaults)

    # Convert the ConfigParser object to a dict

    client_config = configparser.ConfigParser(Client.client_defaults)

            logger.debug("GnuTLS: %s",

                         string[:-1].decode("utf-8",

                                            errors="replace"))

    if gi.version_info < (3, 10, 2):

        # multiprocessing will use threads, so before we use GLib we

        # need to inform GLib that threads will be used.

        GLib.threads_init()

                        # .name, .host, and .checker_command

                        for k in ("name", "host", "checker_command"):

                    # .host and .checker_command

                        for attribute in ("host", "checker_command"):

                            if isinstance(value[attribute], bytes):

                                value[attribute] = (value[attribute]

                                                    .decode("utf-8"))

        GLib.io_add_watch(

            GLib.IOChannel.unix_new(tcp_server.fileno()),

            GLib.PRIORITY_DEFAULT, GLib.IO_IN,

            lambda *args, **kwargs: (tcp_server.handle_request

                                     (*args[2:], **kwargs) or True))

 

def should_only_run_tests():

    parser = argparse.ArgumentParser(add_help=False)

    parser.add_argument("--check", action='store_true')

    args, unknown_args = parser.parse_known_args()

    run_tests = args.check

    if run_tests:

        # Remove --check argument from sys.argv

        sys.argv[1:] = unknown_args

    return run_tests

 

# Add all tests from doctest strings

def load_tests(loader, tests, none):

    import doctest

    tests.addTests(doctest.DocTestSuite())

    return tests

    try:

        if should_only_run_tests():

            # Call using ./mandos --check [--verbose]

            unittest.main()

        else:

            main()

    finally:

        logging.shutdown()