/mandos/trunk

# -*- 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-2019 Teddy Hogeborn

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

#

# This file is part of Mandos.

#

# Mandos is free software: you can redistribute it and/or modify it

# under the terms of the GNU General Public License as published by

#     Mandos is distributed in the hope that it will be useful, but

#     WITHOUT ANY WARRANTY; without even the implied warranty of

# along with Mandos.  If not, see <http://www.gnu.org/licenses/>.

import unittest

import gi

if sys.version_info.major == 2:

    __metaclass__ = type

 

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

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

class PGPEngine:

class avahi:

    """This isn't so much a class as it is a module-like namespace."""

    @staticmethod

    def string_array_to_txt_array(t):

class AvahiService:

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

class gnutls:

    """This isn't so much a class as it is a module-like namespace."""

    CRT_RAWPK = 3

    X509_FMT_DER = 0

    NO_TICKETS = 1<<10

    ENABLE_RAWPK = 1<<18

    CTYPE_PEERS = 3

    KEYID_USE_SHA256 = 1        # gnutls/x509.h

                message = gnutls.strerror(code)

            return super(gnutls.Error, self).__init__(

    class Credentials:

    class ClientSession:

            gnutls_flags = gnutls.CLIENT

            if gnutls.check_version(b"3.5.6"):

                gnutls_flags |= gnutls.NO_TICKETS

            if gnutls.has_rawpk:

                gnutls_flags |= gnutls.ENABLE_RAWPK

            gnutls.init(ctypes.byref(self._c_object), gnutls_flags)

            del gnutls_flags

    _need_version = b"3.3.0"

    if check_version(_need_version) is None:

        raise self.Error("Needs GnuTLS {} or later"

                         .format(_need_version))

 

    _tls_rawpk_version = b"3.6.6"

    has_rawpk = bool(check_version(_tls_rawpk_version))

 

    if has_rawpk:

        # Types

        class pubkey_st(ctypes.Structure):

            _fields = []

        pubkey_t = ctypes.POINTER(pubkey_st)

 

        x509_crt_fmt_t = ctypes.c_int

 

        # All the function declarations below are from gnutls/abstract.h

        pubkey_init = _library.gnutls_pubkey_init

        pubkey_init.argtypes = [ctypes.POINTER(pubkey_t)]

        pubkey_init.restype = _error_code

 

        pubkey_import = _library.gnutls_pubkey_import

        pubkey_import.argtypes = [pubkey_t, ctypes.POINTER(datum_t),

                                  x509_crt_fmt_t]

        pubkey_import.restype = _error_code

 

        pubkey_get_key_id = _library.gnutls_pubkey_get_key_id

        pubkey_get_key_id.argtypes = [pubkey_t, ctypes.c_int,

                                      ctypes.POINTER(ctypes.c_ubyte),

                                      ctypes.POINTER(ctypes.c_size_t)]

        pubkey_get_key_id.restype = _error_code

 

        pubkey_deinit = _library.gnutls_pubkey_deinit

        pubkey_deinit.argtypes = [pubkey_t]

        pubkey_deinit.restype = None

    else:

        # All the function declarations below are from gnutls/openpgp.h

 

        openpgp_crt_init = _library.gnutls_openpgp_crt_init

        openpgp_crt_init.argtypes = [ctypes.POINTER(openpgp_crt_t)]

        openpgp_crt_init.restype = _error_code

 

        openpgp_crt_import = _library.gnutls_openpgp_crt_import

        openpgp_crt_import.argtypes = [openpgp_crt_t,

                                       ctypes.POINTER(datum_t),

                                       openpgp_crt_fmt_t]

        openpgp_crt_import.restype = _error_code

 

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

        openpgp_crt_verify_self.restype = _error_code

 

        openpgp_crt_deinit = _library.gnutls_openpgp_crt_deinit

        openpgp_crt_deinit.argtypes = [openpgp_crt_t]

        openpgp_crt_deinit.restype = None

 

        openpgp_crt_get_fingerprint = (

            _library.gnutls_openpgp_crt_get_fingerprint)

        openpgp_crt_get_fingerprint.argtypes = [openpgp_crt_t,

                                                ctypes.c_void_p,

                                                ctypes.POINTER(

                                                    ctypes.c_size_t)]

        openpgp_crt_get_fingerprint.restype = _error_code

 

    if check_version(b"3.6.4"):

        certificate_type_get2 = _library.gnutls_certificate_type_get2

        certificate_type_get2.argtypes = [session_t, ctypes.c_int]

        certificate_type_get2.restype = _error_code

class Client:

    checker: multiprocessing.Process(); a running checker process used

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

             running.

                 uniquely identify an OpenPGP client

    key_id: string (64 hexadecimal digits); used to uniquely identify

            a client using raw public keys

                          "created", "enabled", "expires", "key_id",

            # Uppercase and remove spaces from key_id and fingerprint

            # for later comparison purposes with return value from the

            # key_id() and fingerprint() functions

            client["key_id"] = (section.get("key_id", "").upper()

                                .replace(" ", ""))

        logger.debug("  Key ID: %s", self.key_id)

        self.checker.join()

        self.checker_callback_tag = None

        self.checker = None

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

                GLib.PRIORITY_DEFAULT, GLib.IO_IN,

    # KeyID - property

    @dbus_annotations(

        {"org.freedesktop.DBus.Property.EmitsChangedSignal": "const"})

    @dbus_service_property(_interface, signature="s", access="read")

    def KeyID_dbus_property(self):

        return dbus.String(self.key_id)

 

class ProxyClient:

    def __init__(self, child_pipe, key_id, fpr, address):

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

            raise KeyError(key_id or fpr)

                if gnutls.has_rawpk:

                    fpr = b""

                    try:

                        key_id = self.key_id(

                            self.peer_certificate(session))

                    except (TypeError, gnutls.Error) as error:

                        logger.warning("Bad certificate: %s", error)

                        return

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

 

                else:

                    key_id = b""

                    try:

                        fpr = self.fingerprint(

                            self.peer_certificate(session))

                    except (TypeError, gnutls.Error) as error:

                        logger.warning("Bad certificate: %s", error)

                        return

                    logger.debug("Fingerprint: %s", fpr)

 

                try:

                    client = ProxyClient(child_pipe, key_id, fpr,

        "Return the peer's certificate as a bytestring"

        try:

            cert_type = gnutls.certificate_type_get2(session._c_object,

                                                     gnutls.CTYPE_PEERS)

        except AttributeError:

            cert_type = gnutls.certificate_type_get(session._c_object)

        if gnutls.has_rawpk:

            valid_cert_types = frozenset((gnutls.CRT_RAWPK,))

        else:

            valid_cert_types = frozenset((gnutls.CRT_OPENPGP,))

        # If not a valid certificate type...

        if cert_type not in valid_cert_types:

            logger.info("Cert type %r not in %r", cert_type,

                        valid_cert_types)

    def key_id(certificate):

        "Convert a certificate bytestring to a hexdigit key ID"

        # New GnuTLS "datum" with the public key

        datum = gnutls.datum_t(

            ctypes.cast(ctypes.c_char_p(certificate),

                        ctypes.POINTER(ctypes.c_ubyte)),

            ctypes.c_uint(len(certificate)))

        # XXX all these need to be created in the gnutls "module"

        # New empty GnuTLS certificate

        pubkey = gnutls.pubkey_t()

        gnutls.pubkey_init(ctypes.byref(pubkey))

        # Import the raw public key into the certificate

        gnutls.pubkey_import(pubkey,

                             ctypes.byref(datum),

                             gnutls.X509_FMT_DER)

        # New buffer for the key ID

        buf = ctypes.create_string_buffer(32)

        buf_len = ctypes.c_size_t(len(buf))

        # Get the key ID from the raw public key into the buffer

        gnutls.pubkey_get_key_id(pubkey,

                                 gnutls.KEYID_USE_SHA256,

                                 ctypes.cast(ctypes.byref(buf),

                                             ctypes.POINTER(ctypes.c_ubyte)),

                                 ctypes.byref(buf_len))

        # Deinit the certificate

        gnutls.pubkey_deinit(pubkey)

 

        # Convert the buffer to a Python bytestring

        key_id = ctypes.string_at(buf, buf_len.value)

        # Convert the bytestring to hexadecimal notation

        hex_key_id = binascii.hexlify(key_id).upper()

        return hex_key_id

 

    @staticmethod

            raise gnutls.CertificateSecurityError(code

                                                  =crtverify.value)

class MultiprocessingMixIn:

class MultiprocessingMixInWithPipe(MultiprocessingMixIn):

                     socketserver.TCPServer):

            if self.server_address[1]:

                self.allow_reuse_address = True

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

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

            key_id = request[1].decode("ascii")

            fpr = request[2].decode("ascii")

            address = request[3]

                if key_id == "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855":

                    continue

                if key_id and c.key_id == key_id:

                    client = c

                    break

                if fpr and c.fingerprint == fpr:

                logger.info("Client not found for key ID: %s, address"

                            ": %s", key_id or fpr, address)

                    mandos_dbus_service.ClientNotFound(key_id or fpr,

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

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

    if gnutls.has_rawpk:

        priority = ("SECURE128:!CTYPE-X.509:+CTYPE-RAWPK:!RSA"

                    ":!VERS-ALL:+VERS-TLS1.3:%PROFILE_ULTRA")

    else:

        priority = ("SECURE256:!CTYPE-X.509:+CTYPE-OPENPGP:!RSA"

                    ":+SIGN-DSA-SHA256")

                       "priority": priority,

    del priority

    server_config = configparser.ConfigParser(server_defaults)

    # Convert the ConfigParser object to a dict

    for option in ("debug", "use_dbus", "use_ipv6", "restore",

                   "foreground", "zeroconf"):

    client_config = configparser.ConfigParser(Client.client_defaults)

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

                        if "key_id" not in value:

                            value["key_id"] = ""

                        elif "fingerprint" not in value:

                            value["fingerprint"] = ""

            def ClientNotFound(self, key_id, address):

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