/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

#

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

 

# Try to find the value of SO_BINDTODEVICE:

try:

    # This is where SO_BINDTODEVICE is in Python 3.3 (or 3.4?) and

    # newer, and it is also the most natural place for it:

        # This is where SO_BINDTODEVICE was up to and including Python

        # 2.6, and also 3.2:

        # In Python 2.7 it seems to have been removed entirely.

        # Try running the C preprocessor:

        try:

            cc = subprocess.Popen(["cc", "--language=c", "-E",

                                   "/dev/stdin"],

                                  stdin=subprocess.PIPE,

                                  stdout=subprocess.PIPE)

            stdout = cc.communicate(

                "#include <sys/socket.h>\nSO_BINDTODEVICE\n")[0]

            SO_BINDTODEVICE = int(stdout.splitlines()[-1])

        except (OSError, ValueError, IndexError):

            # No value found

            SO_BINDTODEVICE = None

 

if sys.version_info < (3, 2):

    configparser.Configparser = configparser.SafeConfigParser

 

version = "1.8.11"

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

 

 

        facility=logging.handlers.SysLogHandler.LOG_DAEMON,

        address="/dev/log"))

 

class PGPEngine:

 

                if name == b"gpg":

                          '--quiet']

        # Only GPG version 1 has the --no-use-agent option.

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

            self.gnupgargs.append("--no-use-agent")

 

 

 

 

                                             topdown=False):

 

 

                                    stdin=subprocess.PIPE,

                                    stdout=subprocess.PIPE,

                                    stderr=subprocess.PIPE)

            ciphertext, err = proc.communicate(input=data)

 

                dir=self.tempdir) as passfile:

                                    stdin=subprocess.PIPE,

                                    stdout=subprocess.PIPE,

                                    stderr=subprocess.PIPE)

            decrypted_plaintext, err = proc.communicate(input=data)

 

class avahi:

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

    IF_UNSPEC = -1               # avahi-common/address.h

    PROTO_UNSPEC = -1            # avahi-common/address.h

    PROTO_INET = 0               # avahi-common/address.h

    PROTO_INET6 = 1              # avahi-common/address.h

 

    @staticmethod

    def string_array_to_txt_array(t):

    ENTRY_GROUP_ESTABLISHED = 2  # avahi-common/defs.h

    ENTRY_GROUP_COLLISION = 3    # avahi-common/defs.h

    ENTRY_GROUP_FAILURE = 4      # avahi-common/defs.h

    SERVER_INVALID = 0           # avahi-common/defs.h

    SERVER_REGISTERING = 1       # avahi-common/defs.h

    SERVER_RUNNING = 2           # avahi-common/defs.h

    SERVER_COLLISION = 3         # avahi-common/defs.h

    SERVER_FAILURE = 4           # avahi-common/defs.h

 

class AvahiService:

 

 

                 interface=avahi.IF_UNSPEC,

                 name=None,

                 servicetype=None,

                 port=None,

                 TXT=None,

                 domain="",

                 host="",

                 max_renames=32768,

                 protocol=avahi.PROTO_UNSPEC,

                 bus=None):

 

 

 

 

 

 

 

 

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

 

class gnutls:

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

 

    library = ctypes.util.find_library("gnutls")

    if library is None:

        library = ctypes.util.find_library("gnutls-deb0")

    _library = ctypes.cdll.LoadLibrary(library)

    del library

 

 

    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

 

 

 

 

    openpgp_crt_fmt_t = ctypes.c_int  # gnutls/openpgp.h

 

        def __init__(self, message=None, code=None, args=()):

                message = gnutls.strerror(code)

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

 

 

    class Credentials:

 

 

    class ClientSession:

        def __init__(self, socket, credentials=None):

            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

 

 

 

 

 

            raise gnutls.CertificateSecurityError(code=result)

        raise gnutls.Error(code=result)

 

 

 

 

 

 

 

 

    certificate_free_credentials.argtypes = [

        certificate_credentials_t]

 

 

 

 

 

 

 

 

 

 

 

 

 

    _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(" ", ""))

 

 

    def __init__(self, settings, name=None, server_settings=None):

 

 

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

 

 

 

 

 

 

 

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

                                 + 1)),

 

        if self.checker is not None:

            self.checker.join()

        self.checker_callback_tag = None

        self.checker = None

 

 

 

 

 

 

 

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

                for attr in self.runtime_expansions}

            popen_args = {"close_fds": True,

                          "shell": True,

                          "cwd": "/"}

                                   "stderr": wnull})

            pipe = multiprocessing.Pipe(duplex=False)

                target=call_pipe,

                args=(pipe[1], subprocess.call, command),

                kwargs=popen_args)

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

                GLib.PRIORITY_DEFAULT, GLib.IO_IN,

 

 

 

 

        func._dbus_get_args_options = {'byte_arrays': byte_arrays}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

                         out_signature="s",

                         path_keyword='object_path',

                         connection_keyword='connection')

 

 

                    == dbus.INTROSPECTABLE_IFACE):

 

 

 

 

 

 

            value = dbus.ByteArray(bytes(value))

 

 

                value, variant_level=value.variant_level + 1)

 

 

 

 

 

 

 

 

 

                         out_signature="a{oa{sa{sv}}}")

 

                         signature="oa{sa{sv}}")

 

    @dbus.service.signal(dbus.OBJECT_MANAGER_IFACE, signature="oas")

 

                         out_signature="s",

                         path_keyword='object_path',

                         connection_keyword='connection')

 

 

                    == dbus.OBJECT_MANAGER_IFACE):

                         exc_info=error)

 

        return dbus.String("", variant_level=variant_level)