/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

    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")

version = "1.8.14"

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

                           ("Mandos [%(process)d]: %(levelname)s:"

                            " %(message)s"))

        console.setFormatter(logging.Formatter("%(asctime)s %(name)s"

                                               " [%(process)d]:"

                                               " %(levelname)s:"

                                               " %(message)s"))

                if name == b"gpg":

        self.gnupgargs = ["--batch",

                          "--homedir", self.tempdir,

                          "--force-mdc",

                          "--quiet"]

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

            proc = subprocess.Popen([self.gpg, "--symmetric",

                                     "--passphrase-file",

            proc = subprocess.Popen([self.gpg, "--decrypt",

                                     "--passphrase-file",

    name: string; Example: "Mandos"

    type: string; Example: "_mandos._tcp".

                "StateChanged", self.entry_group_state_changed))

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

                                                       **kwargs)

            "Mandos ({}) [%(process)d]: %(levelname)s: %(message)s"

    class _session_int(ctypes.Structure):

    session_t = ctypes.POINTER(_session_int)

        _fields_ = [("data", ctypes.POINTER(ctypes.c_ubyte)),

                    ("size", ctypes.c_uint)]

    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)

        for the "restype" attribute on ctypes functions"""

        if result >= gnutls.E_SUCCESS:

    def _retry_on_error(result, func, arguments,

                        _error_code=_error_code):

        for the "errcheck" attribute on ctypes functions"""

        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]

    approved:   bool(); None if not yet approved/disapproved

             to see if the client lives. None if no process is

            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,

        func._dbus_get_args_options = {"byte_arrays": byte_arrays}

                         path_keyword="object_path",

                         connection_keyword="connection")

            value = dbus.ByteArray(bytes(value))

                         path_keyword="object_path",

                         connection_keyword="connection")

                         path_keyword="object_path",

                         connection_keyword="connection")

        self._pipe.send(("init", key_id, fpr, address))

        if name == "_pipe":

        self._pipe.send(("getattr", name))

        if data[0] == "data":

        if data[0] == "function":

                self._pipe.send(("funcall", name, args, kwargs))

        if name == "_pipe":

        self._pipe.send(("setattr", name, value))

            # priority = ":".join(("NONE", "+VERS-TLS1.1",

            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))

    """IPv6-capable TCP server.  Accepts None as address and/or port

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

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

        if command == "init":

                if key_id == ("E3B0C44298FC1C149AFBF4C8996FB924"

                              "27AE41E4649B934CA495991B7852B855"):

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

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

        if command == "funcall":

            parent_pipe.send(("data", getattr(client_object,

        if command == "getattr":

                          collections.abc.Callable):

                parent_pipe.send(("function", ))

                    "data", client_object.__getattribute__(attrname)))

        if command == "setattr":

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

            logging.Formatter("Mandos ({}) [%(process)d]:"

                              " %(levelname)s: %(message)s".format(

            logger.debug("GnuTLS: %s",

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

                                            errors="replace"))

                        # .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"))

                    mode="wb",

                    prefix="clients-",

        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

 

if __name__ == "__main__":

    try:

        if should_only_run_tests():

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

            unittest.main()

        else:

            main()

    finally:

        logging.shutdown()