python/qpid_dispatch_internal/dispatch.py

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License
#

"""Interface between python and libqpid-dispatch.so.

This module contains python ctypes definitions to directly call functions in the
libqpid-dispatch.so library from python.

The C library also adds the following C extension types to this module:

- LogAdapter: Logs to the C logging system.
- IoAdapter: Receives messages from the router into python.

This module also prevents the proton python module from being accidentally loaded.
"""
import builtins
import ctypes
import sys
from ctypes import c_char_p, c_long, py_object


class CError(Exception):
    """Exception raised if there is an error in a C call"""
    pass


class QdDll(ctypes.PyDLL):
    """
    Load the library, set up function prototypes.

    NOTE: We use the python calling convention because the C library
    internally makes python calls.
    """

    def __init__(self, handle):
        super(QdDll, self).__init__("qpid-dispatch", handle=handle)

        # Types
        self.qd_dispatch_p = ctypes.c_void_p

        # No check on qd_error_* functions, it would be recursive
        self._prototype(self.qd_error_code, c_long, [], check=False)
        self._prototype(self.qd_error_message, c_char_p, [], check=False)
        self._prototype(self.qd_log_entity, c_long, [py_object])
        self._prototype(self.qd_dispatch_configure_router, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_prepare, None, [self.qd_dispatch_p])
        self._prototype(self.qd_dispatch_configure_listener, ctypes.c_void_p, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_connector, ctypes.c_void_p, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_ssl_profile, ctypes.c_void_p, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_sasl_plugin, ctypes.c_void_p, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_connection_manager_delete_listener, None, [self.qd_dispatch_p, ctypes.c_void_p])
        self._prototype(self.qd_connection_manager_delete_connector, None, [self.qd_dispatch_p, ctypes.c_void_p])
        self._prototype(self.qd_connection_manager_delete_ssl_profile, ctypes.c_bool, [self.qd_dispatch_p, ctypes.c_void_p])
        self._prototype(self.qd_connection_manager_delete_sasl_plugin, ctypes.c_bool, [self.qd_dispatch_p, ctypes.c_void_p])

        self._prototype(self.qd_dispatch_configure_address, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_link_route, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_auto_link, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_exchange, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_configure_binding, None, [self.qd_dispatch_p, py_object])

        self._prototype(self.qd_dispatch_configure_policy, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_register_policy_manager, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_policy_c_counts_alloc, c_long, [], check=False)
        self._prototype(self.qd_dispatch_policy_c_counts_free, None, [c_long], check=False)
        self._prototype(self.qd_dispatch_policy_c_counts_refresh, None, [c_long, py_object])
        self._prototype(self.qd_dispatch_policy_host_pattern_add, ctypes.c_bool, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_policy_host_pattern_remove, None, [self.qd_dispatch_p, py_object])
        self._prototype(self.qd_dispatch_policy_host_pattern_lookup, c_char_p, [self.qd_dispatch_p, py_object])

        self._prototype(self.qd_dispatch_register_display_name_service, None, [self.qd_dispatch_p, py_object])

        self._prototype(self.qd_dispatch_set_agent, None, [self.qd_dispatch_p, py_object])

        self._prototype(self.qd_router_setup_late, None, [self.qd_dispatch_p])

        self._prototype(self.qd_dispatch_router_lock, None, [self.qd_dispatch_p])
        self._prototype(self.qd_dispatch_router_unlock, None, [self.qd_dispatch_p])

        self._prototype(self.qd_connection_manager_start, None, [self.qd_dispatch_p])
        self._prototype(self.qd_entity_refresh_begin, c_long, [py_object])
        self._prototype(self.qd_entity_refresh_end, None, [])

        self._prototype(self.qd_log_recent_py, py_object, [c_long])

    def _prototype(self, f, restype, argtypes, check=True):
        """Set up the return and argument types and the error checker for a
        ctypes function"""

        def _do_check(result, func, args):
            if check and self.qd_error_code():
                raise CError(self.qd_error_message())
            if restype is c_char_p and result:
                # in python3 c_char_p returns a byte type for the error
                # message. We need to convert that to a string
                result = result.decode('utf-8')
            return result

        f.restype = restype
        f.argtypes = argtypes
        f.errcheck = _do_check
        return f

    def function(self, fname, restype, argtypes, check=True):
        return self._prototype(getattr(self, fname), restype, argtypes, check)


# Prevent accidental loading of the proton python module inside dispatch.
# The proton-C library is linked with the dispatch C library, loading the proton
# python module loads a second copy of the library and mayhem ensues.
#
# Note the FORBIDDEN list is over-written to disable this tests in mock python
# testing code.
FORBIDDEN = ["proton"]


def check_forbidden():
    bad = set(FORBIDDEN) & set(sys.modules)
    if bad:
        raise ImportError("Forbidden modules loaded: '%s'." % "', '".join(bad))


def import_check(name, *args, **kw):
    if name in FORBIDDEN:
        raise ImportError("Python code running inside a dispatch router cannot import '%s', use the 'dispatch' module for internal messaging" % name)
    return builtin_import(name, *args, **kw)


check_forbidden()

builtin_import = builtins.__import__
builtins.__import__ = import_check