deluge/deluge/ui/client.py

# -*- coding: utf-8 -*-
#
# Copyright (C) 2009 Andrew Resch <andrewresch@gmail.com>
# Copyright (C) 2011 Pedro Algarvio <pedro@algarvio.me>
#
# This file is part of Deluge and is licensed under GNU General Public License 3.0, or later, with
# the additional special exception to link portions of this program with the OpenSSL library.
# See LICENSE for more details.
#

from __future__ import unicode_literals

import logging
import subprocess
import sys

from twisted.internet import defer, reactor, ssl
from twisted.internet.protocol import ClientFactory

from deluge import error
from deluge.common import get_localhost_auth, get_version
from deluge.decorators import deprecated
from deluge.transfer import DelugeTransferProtocol

RPC_RESPONSE = 1
RPC_ERROR = 2
RPC_EVENT = 3

log = logging.getLogger(__name__)


def format_kwargs(kwargs):
    return ', '.join([key + '=' + str(value) for key, value in kwargs.items()])


class DelugeRPCRequest(object):
    """
    This object is created whenever there is a RPCRequest to be sent to the
    daemon.  It is generally only used by the DaemonProxy's call method.
    """

    request_id = None
    method = None
    args = None
    kwargs = None

    def __repr__(self):
        """
        Returns a string of the RPCRequest in the following form:
            method(arg, kwarg=foo, ...)
        """
        s = self.method + '('
        if self.args:
            s += ', '.join([str(x) for x in self.args])
        if self.kwargs:
            if self.args:
                s += ', '
            s += format_kwargs(self.kwargs)
        s += ')'

        return s

    def format_message(self):
        """
        Returns a properly formatted RPCRequest based on the properties.  Will
        raise a TypeError if the properties haven't been set yet.

        :returns: a properly formatted RPCRequest
        """
        if (
            self.request_id is None
            or self.method is None
            or self.args is None
            or self.kwargs is None
        ):
            raise TypeError(
                'You must set the properties of this object before calling format_message!'
            )

        return (self.request_id, self.method, self.args, self.kwargs)


class DelugeRPCProtocol(DelugeTransferProtocol):
    def connectionMade(self):  # NOQA: N802
        self.__rpc_requests = {}
        # Set the protocol in the daemon so it can send data
        self.factory.daemon.protocol = self
        # Get the address of the daemon that we've connected to
        peer = self.transport.getPeer()
        self.factory.daemon.host = peer.host
        self.factory.daemon.port = peer.port
        self.factory.daemon.connected = True
        log.debug('Connected to daemon at %s:%s..', peer.host, peer.port)
        self.factory.daemon.connect_deferred.callback((peer.host, peer.port))

    def message_received(self, request):
        """
        This method is called whenever we receive a message from the daemon.

        :param request: a tuple that should be either a RPCResponse, RCPError or RPCSignal

        """
        if not isinstance(request, tuple):
            log.debug('Received invalid message: type is not tuple')
            return
        if len(request) < 3:
            log.debug(
                'Received invalid message: number of items in ' 'response is %s',
                len(request),
            )
            return

        message_type = request[0]

        if message_type == RPC_EVENT:
            event = request[1]
            # log.debug('Received RPCEvent: %s', event)
            # A RPCEvent was received from the daemon so run any handlers
            # associated with it.
            if event in self.factory.event_handlers:
                for handler in self.factory.event_handlers[event]:
                    reactor.callLater(0, handler, *request[2])
            return

        request_id = request[1]

        # We get the Deferred object for this request_id to either run the
        # callbacks or the errbacks dependent on the response from the daemon.
        d = self.factory.daemon.pop_deferred(request_id)

        if message_type == RPC_RESPONSE:
            # Run the callbacks registered with this Deferred object
            d.callback(request[2])
        elif message_type == RPC_ERROR:
            # Recreate exception and errback'it
            try:
                # The exception class is located in deluge.error
                try:
                    exception_cls = getattr(error, request[2])
                    exception = exception_cls(*request[3], **request[4])
                except TypeError:
                    log.warning(
                        'Received invalid RPC_ERROR (Old daemon?): %s', request[2]
                    )
                    return

                # Ideally we would chain the deferreds instead of instance
                # checking just to log them. But, that would mean that any
                # errback on the fist deferred should returns it's failure
                # so it could pass back to the 2nd deferred on the chain. But,
                # that does not always happen.
                # So, just do some instance checking and just log rpc error at
                # different levels.
                r = self.__rpc_requests[request_id]
                msg = 'RPCError Message Received!'
                msg += '\n' + '-' * 80
                msg += '\n' + 'RPCRequest: ' + r.__repr__()
                msg += '\n' + '-' * 80
                if isinstance(exception, error.WrappedException):
                    msg += '\n' + exception.type + '\n' + exception.message + ': '
                    msg += exception.traceback
                else:
                    msg += '\n' + request[5] + '\n' + request[2] + ': '
                    msg += str(exception)
                msg += '\n' + '-' * 80

                if not isinstance(exception, error._ClientSideRecreateError):
                    # Let's log these as errors
                    log.error(msg)
                else:
                    # The rest just gets logged in debug level, just to log
                    # what's happening
                    log.debug(msg)
            except Exception:
                import traceback

                log.error(
                    'Failed to handle RPC_ERROR (Old daemon?): %s\nLocal error: %s',
                    request[2],
                    traceback.format_exc(),
                )
            d.errback(exception)
        del self.__rpc_requests[request_id]

    def send_request(self, request):
        """
        Sends a RPCRequest to the server.

        :param request: RPCRequest

        """
        try:
            # Store the DelugeRPCRequest object just in case a RPCError is sent in
            # response to this request.  We use the extra information when printing
            # out the error for debugging purposes.
            self.__rpc_requests[request.request_id] = request
            # log.debug('Sending RPCRequest %s: %s', request.request_id, request)
            # Send the request in a tuple because multiple requests can be sent at once
            self.transfer_message((request.format_message(),))
        except Exception as ex:
            log.warning('Error occurred when sending message: %s', ex)


class DelugeRPCClientFactory(ClientFactory):
    protocol = DelugeRPCProtocol

    def __init__(self, daemon, event_handlers):
        self.daemon = daemon
        self.event_handlers = event_handlers

    def startedConnecting(self, connector):  # NOQA: N802
        log.debug('Connecting to daemon at "%s:%s"...', connector.host, connector.port)

    def clientConnectionFailed(self, connector, reason):  # NOQA: N802
        log.debug(
            'Connection to daemon at "%s:%s" failed: %s',
            connector.host,
            connector.port,
            reason.value,
        )
        self.daemon.connect_deferred.errback(reason)

    def clientConnectionLost(self, connector, reason):  # NOQA: N802
        log.debug(
            'Connection lost to daemon at "%s:%s" reason: %s',
            connector.host,
            connector.port,
            reason.value,
        )
        self.daemon.host = None
        self.daemon.port = None
        self.daemon.username = None
        self.daemon.connected = False

        if (
            self.daemon.disconnect_deferred
            and not self.daemon.disconnect_deferred.called
        ):
            self.daemon.disconnect_deferred.callback(reason.value)
            self.daemon.disconnect_deferred = None

        if self.daemon.disconnect_callback:
            self.daemon.disconnect_callback()


class DaemonProxy(object):
    pass


class DaemonSSLProxy(DaemonProxy):
    def __init__(self, event_handlers=None):
        if event_handlers is None:
            event_handlers = {}
        self.__factory = DelugeRPCClientFactory(self, event_handlers)
        self.__factory.noisy = False
        self.__request_counter = 0
        self.__deferred = {}

        # This is set when a connection is made to the daemon
        self.protocol = None

        # This is set when a connection is made
        self.host = None
        self.port = None
        self.username = None
        self.authentication_level = 0

        self.connected = False

        self.disconnect_deferred = None
        self.disconnect_callback = None

        self.auth_levels_mapping = None
        self.auth_levels_mapping_reverse = None

    def connect(self, host, port):
        """
        Connects to a daemon at host:port

        :param host: str, the host to connect to
        :param port: int, the listening port on the daemon

        :returns: twisted.Deferred

        """
        log.debug('sslproxy.connect()')
        self.host = host
        self.port = port
        self.__connector = reactor.connectSSL(
            self.host, self.port, self.__factory, ssl.ClientContextFactory()
        )
        self.connect_deferred = defer.Deferred()
        self.daemon_info_deferred = defer.Deferred()

        # Upon connect we do a 'daemon.login' RPC
        self.connect_deferred.addCallback(self.__on_connect)
        self.connect_deferred.addErrback(self.__on_connect_fail)

        return self.daemon_info_deferred

    def disconnect(self):
        log.debug('sslproxy.disconnect()')
        self.disconnect_deferred = defer.Deferred()
        self.__connector.disconnect()
        return self.disconnect_deferred

    def call(self, method, *args, **kwargs):
        """
        Makes a RPCRequest to the daemon.  All methods should be in the form of
        'component.method'.

        :params method: str, the method to call in the form of 'component.method'
        :params args: the arguments to call the remote method with
        :params kwargs: the keyword arguments to call the remote method with

        :return: a twisted.Deferred object that will be activated when a RPCResponse
            or RPCError is received from the daemon

        """
        # Create the DelugeRPCRequest to pass to protocol.send_request()
        request = DelugeRPCRequest()
        request.request_id = self.__request_counter
        request.method = method
        request.args = args
        request.kwargs = kwargs
        # Send the request to the server
        self.protocol.send_request(request)
        # Create a Deferred object to return and add a default errback to print
        # the error.
        d = defer.Deferred()

        # Store the Deferred until we receive a response from the daemon
        self.__deferred[self.__request_counter] = d

        # Increment the request counter since we don't want to use the same one
        # before a response is received.
        self.__request_counter += 1

        return d

    def pop_deferred(self, request_id):
        """
        Pops a Deferred object.  This is generally called once we receive the
        reply we were waiting on from the server.

        :param request_id: the request_id of the Deferred to pop
        :type request_id: int

        """
        return self.__deferred.pop(request_id)

    def register_event_handler(self, event, handler):
        """
        Registers a handler function to be called when `:param:event` is received
        from the daemon.

        :param event: the name of the event to handle
        :type event: str
        :param handler: the function to be called when `:param:event`
            is emitted from the daemon
        :type handler: function

        """
        if event not in self.__factory.event_handlers:
            # This is a new event to handle, so we need to tell the daemon
            # that we're interested in receiving this type of event
            self.__factory.event_handlers[event] = []
            if self.connected:
                self.call('daemon.set_event_interest', [event])

        # Only add the handler if it's not already registered
        if handler not in self.__factory.event_handlers[event]:
            self.__factory.event_handlers[event].append(handler)

    def deregister_event_handler(self, event, handler):
        """
        Deregisters a event handler.

        :param event: the name of the event
        :type event: str
        :param handler: the function registered
        :type handler: function

        """
        if (
            event in self.__factory.event_handlers
            and handler in self.__factory.event_handlers[event]
        ):
            self.__factory.event_handlers[event].remove(handler)

    def __on_connect(self, result):
        log.debug('__on_connect called')

        def on_info(daemon_info):
            self.daemon_info = daemon_info
            log.debug('Got info from daemon: %s', daemon_info)
            self.daemon_info_deferred.callback(daemon_info)

        def on_info_fail(reason):
            log.debug('Failed to get info from daemon')
            log.exception(reason)
            self.daemon_info_deferred.errback(reason)

        self.call('daemon.info').addCallback(on_info).addErrback(on_info_fail)
        return self.daemon_info_deferred

    def __on_connect_fail(self, reason):
        self.daemon_info_deferred.errback(reason)

    def authenticate(self, username, password):
        log.debug('%s.authenticate: %s', self.__class__.__name__, username)
        login_deferred = defer.Deferred()
        d = self.call('daemon.login', username, password, client_version=get_version())
        d.addCallbacks(
            self.__on_login,
            self.__on_login_fail,
            callbackArgs=[username, login_deferred],
            errbackArgs=[login_deferred],
        )
        return login_deferred

    def __on_login(self, result, username, login_deferred):
        log.debug('__on_login called: %s %s', username, result)
        self.username = username
        self.authentication_level = result
        # We need to tell the daemon what events we're interested in receiving
        if self.__factory.event_handlers:
            self.call('daemon.set_event_interest', list(self.__factory.event_handlers))
            self.call('core.get_auth_levels_mappings').addCallback(
                self.__on_auth_levels_mappings
            )

        login_deferred.callback(result)

    def __on_login_fail(self, result, login_deferred):
        login_deferred.errback(result)

    def __on_auth_levels_mappings(self, result):
        auth_levels_mapping, auth_levels_mapping_reverse = result
        self.auth_levels_mapping = auth_levels_mapping
        self.auth_levels_mapping_reverse = auth_levels_mapping_reverse

    def set_disconnect_callback(self, cb):
        """
        Set a function to be called when the connection to the daemon is lost
        for any reason.
        """
        self.disconnect_callback = cb

    def get_bytes_recv(self):
        return self.protocol.get_bytes_recv()

    def get_bytes_sent(self):
        return self.protocol.get_bytes_sent()


class DaemonStandaloneProxy(DaemonProxy):
    def __init__(self, event_handlers=None):
        if event_handlers is None:
            event_handlers = {}
        from deluge.core import daemon

        self.__daemon = daemon.Daemon(standalone=True)
        self.__daemon.start()
        log.debug('daemon created!')
        self.connected = True
        self.host = 'localhost'
        self.port = 58846
        # Running in standalone mode, it's safe to import auth level
        from deluge.core.authmanager import (
            AUTH_LEVEL_ADMIN,
            AUTH_LEVELS_MAPPING,
            AUTH_LEVELS_MAPPING_REVERSE,
        )

        self.username = 'localclient'
        self.authentication_level = AUTH_LEVEL_ADMIN
        self.auth_levels_mapping = AUTH_LEVELS_MAPPING
        self.auth_levels_mapping_reverse = AUTH_LEVELS_MAPPING_REVERSE
        # Register the event handlers
        for event in event_handlers:
            for handler in event_handlers[event]:
                self.__daemon.core.eventmanager.register_event_handler(event, handler)

    def disconnect(self):
        self.connected = False
        self.__daemon = None

    def call(self, method, *args, **kwargs):
        # log.debug('call: %s %s %s', method, args, kwargs)

        import copy

        try:
            m = self.__daemon.rpcserver.get_object_method(method)
        except Exception as ex:
            log.exception(ex)
            return defer.fail(ex)
        else:
            return defer.maybeDeferred(m, *copy.deepcopy(args), **copy.deepcopy(kwargs))

    def register_event_handler(self, event, handler):
        """
        Registers a handler function to be called when `:param:event` is
        received from the daemon.

        :param event: the name of the event to handle
        :type event: str
        :param handler: the function to be called when `:param:event`
            is emitted from the daemon
        :type handler: function

        """
        self.__daemon.core.eventmanager.register_event_handler(event, handler)

    def deregister_event_handler(self, event, handler):
        """
        Deregisters a event handler.

        :param event: the name of the event
        :type event: str
        :param handler: the function registered
        :type handler: function

        """
        self.__daemon.core.eventmanager.deregister_event_handler(event, handler)


class DottedObject(object):
    """
    This is used for dotted name calls to client
    """

    def __init__(self, daemon, method):
        self.daemon = daemon
        self.base = method

    def __call__(self, *args, **kwargs):
        raise Exception('You must make calls in the form of "component.method"')

    def __getattr__(self, name):
        return RemoteMethod(self.daemon, self.base + '.' + name)


class RemoteMethod(DottedObject):
    """
    This is used when something like 'client.core.get_something()' is attempted.
    """

    def __call__(self, *args, **kwargs):
        return self.daemon.call(self.base, *args, **kwargs)


class Client(object):
    """
    This class is used to connect to a daemon process and issue RPC requests.
    """

    __event_handlers = {}

    def __init__(self):
        self._daemon_proxy = None
        self.disconnect_callback = None
        self.__started_standalone = False

    def connect(
        self,
        host='127.0.0.1',
        port=58846,
        username='',
        password='',
        skip_authentication=False,
    ):
        """
        Connects to a daemon process.

        :param host: str, the hostname of the daemon
        :param port: int, the port of the daemon
        :param username: str, the username to login with
        :param password: str, the password to login with

        :returns: a Deferred object that will be called once the connection
            has been established or fails
        """

        self._daemon_proxy = DaemonSSLProxy(dict(self.__event_handlers))
        self._daemon_proxy.set_disconnect_callback(self.__on_disconnect)

        d = self._daemon_proxy.connect(host, port)

        def on_connected(daemon_version):
            log.debug('on_connected. Daemon version: %s', daemon_version)
            return daemon_version

        def on_connect_fail(reason):
            log.debug('on_connect_fail: %s', reason)
            self.disconnect()
            return reason

        def on_authenticate(result, daemon_info):
            log.debug('Authentication successful: %s', result)
            return result

        def on_authenticate_fail(reason):
            log.debug('Failed to authenticate: %s', reason.value)
            return reason

        def authenticate(daemon_version, username, password):
            if not username and host in ('127.0.0.1', 'localhost'):
                # No username provided and it's localhost, so attempt to get credentials from auth file.
                username, password = get_localhost_auth()

            d = self._daemon_proxy.authenticate(username, password)
            d.addCallback(on_authenticate, daemon_version)
            d.addErrback(on_authenticate_fail)
            return d

        d.addCallback(on_connected)
        d.addErrback(on_connect_fail)
        if not skip_authentication:
            d.addCallback(authenticate, username, password)
        return d

    def disconnect(self):
        """
        Disconnects from the daemon.
        """
        if self.is_standalone():
            self._daemon_proxy.disconnect()
            self.stop_standalone()
            return defer.succeed(True)

        if self._daemon_proxy:
            return self._daemon_proxy.disconnect()

    def start_standalone(self):
        """
        Starts a daemon in the same process as the client.
        """
        self._daemon_proxy = DaemonStandaloneProxy(self.__event_handlers)
        self.__started_standalone = True

    def stop_standalone(self):
        """
        Stops the daemon process in the client.
        """
        self._daemon_proxy = None
        self.__started_standalone = False

    @deprecated
    def start_classic_mode(self):
        """Deprecated: Use start_standalone"""
        self.start_standalone()

    @deprecated
    def stop_classic_mode(self):
        """Deprecated: Use stop_standalone"""
        self.stop_standalone()

    def start_daemon(self, port, config):
        """Starts a daemon process.

        Args:
            port (int): Port for the daemon to listen on.
            config (str): Config path to pass to daemon.

        Returns:
            bool: True is successfully started the daemon, False otherwise.

        """
        # subprocess.popen does not work with unicode args (with non-ascii characters) on windows
        config = config.encode(sys.getfilesystemencoding())
        try:
            subprocess.Popen(['deluged', '--port=%s' % port, b'--config=%s' % config])
        except OSError as ex:
            from errno import ENOENT

            if ex.errno == ENOENT:
                log.error(
                    _(
                        'Deluge cannot find the `deluged` executable, check that '
                        'the deluged package is installed, or added to your PATH.'
                    )
                )
            else:
                log.exception(ex)
        except Exception as ex:
            log.error('Unable to start daemon!')
            log.exception(ex)
        else:
            return True
        return False

    def is_localhost(self):
        """
        Checks if the current connected host is a localhost or not.

        :returns: bool, True if connected to a localhost

        """
        if (
            self._daemon_proxy
            and self._daemon_proxy.host in ('127.0.0.1', 'localhost')
            or isinstance(self._daemon_proxy, DaemonStandaloneProxy)
        ):
            return True
        return False

    def is_standalone(self):
        """
        Checks to see if the client has been started in standalone mode.

        :returns: bool, True if in standalone mode

        """
        return self.__started_standalone

    @deprecated
    def is_classicmode(self):
        """Deprecated: Use is_standalone"""
        self.is_standalone()

    def connected(self):
        """
        Check to see if connected to a daemon.

        :returns: bool, True if connected

        """
        return self._daemon_proxy.connected if self._daemon_proxy else False

    def connection_info(self):
        """
        Get some info about the connection or return None if not connected.

        :returns: a tuple of (host, port, username) or None if not connected
        """
        if self.connected():
            return (
                self._daemon_proxy.host,
                self._daemon_proxy.port,
                self._daemon_proxy.username,
            )

        return None

    def register_event_handler(self, event, handler):
        """
        Registers a handler that will be called when an event is received from the daemon.

        :params event: str, the event to handle
        :params handler: func, the handler function, f(args)
        """
        if event not in self.__event_handlers:
            self.__event_handlers[event] = []
        self.__event_handlers[event].append(handler)
        # We need to replicate this in the daemon proxy
        if self._daemon_proxy:
            self._daemon_proxy.register_event_handler(event, handler)

    def deregister_event_handler(self, event, handler):
        """
        Deregisters a event handler.

        :param event: str, the name of the event
        :param handler: function, the function registered

        """
        if event in self.__event_handlers and handler in self.__event_handlers[event]:
            self.__event_handlers[event].remove(handler)
        if self._daemon_proxy:
            self._daemon_proxy.deregister_event_handler(event, handler)

    def force_call(self, block=False):
        # no-op for now.. we'll see if we need this in the future
        pass

    def __getattr__(self, method):
        return DottedObject(self._daemon_proxy, method)

    def set_disconnect_callback(self, cb):
        """
        Set a function to be called whenever the client is disconnected from
        the daemon for any reason.
        """
        self.disconnect_callback = cb

    def __on_disconnect(self):
        if self.disconnect_callback:
            self.disconnect_callback()

    def get_bytes_recv(self):
        """
        Returns the number of bytes received from the daemon.

        :returns: the number of bytes received
        :rtype: int
        """
        return self._daemon_proxy.get_bytes_recv()

    def get_bytes_sent(self):
        """
        Returns the number of bytes sent to the daemon.

        :returns: the number of bytes sent
        :rtype: int
        """
        return self._daemon_proxy.get_bytes_sent()

    def get_auth_user(self):
        """
        Returns the current authenticated username.

        :returns: the authenticated username
        :rtype: str
        """
        return self._daemon_proxy.username

    def get_auth_level(self):
        """
        Returns the authentication level the daemon returned upon authentication.

        :returns: the authentication level
        :rtype: int
        """
        return self._daemon_proxy.authentication_level

    @property
    def auth_levels_mapping(self):
        return self._daemon_proxy.auth_levels_mapping

    @property
    def auth_levels_mapping_reverse(self):
        return self._daemon_proxy.auth_levels_mapping_reverse


# This is the object clients will use
client = Client()