~chicharreros/magicicada-protocol/1.0

# -*- coding: utf-8 -*-

#

# Copyright 2009-2015 Canonical Ltd.

# Copyright 2015-2018 Chicharreros (https://launchpad.net/~chicharreros)

#

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

# under the terms of the GNU Affero General Public License version 3,

# as published by the Free Software Foundation.

#

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

# WITHOUT ANY WARRANTY; without even the implied warranties of

# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR

# PURPOSE.  See the GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

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

#

# In addition, as a special exception, the copyright holders give

# permission to link the code of portions of this program with the

# OpenSSL library under certain conditions as described in each

# individual source file, and distribute linked combinations

# including the two.

# You must obey the GNU General Public License in all respects

# for all of the code used other than OpenSSL.  If you modify

# file(s) with this exception, you may extend this exception to your

# version of the file(s), but you are not obligated to do so.  If you

# do not wish to do so, delete this exception statement from your

# version.  If you delete this exception statement from all source

# files in the program, then also delete it here.

"""The storage protocol client."""

import logging

import uuid

from functools import partial

from itertools import chain

from twisted.internet.protocol import ClientFactory

from twisted.internet import reactor, defer

from twisted.python import log

from ubuntuone.storageprotocol import (

    delta,

    protocol_pb2,

    public_file_info,

    request,

    sharersp,

    volumes,

)

log_debug = partial(log.msg, loglevel=logging.DEBUG)

class StorageClient(request.RequestHandler):

    """A Basic Storage Protocol client."""

    # we are a client, we do odd requests

    REQUEST_ID_START = 1

    def __init__(self):

        """Create the client. done by the factory."""

        request.RequestHandler.__init__(self)

        self.root_id = None

        self.root_id_defers = []

        self._node_state_callback = None

        self._share_change_callback = None

        self._share_delete_callback = None

        self._share_answer_callback = None

        self._free_space_callback = None

        self._account_info_callback = None

        self._volume_created_callback = None

        self._volume_deleted_callback = None

        self._volume_new_generation_callback = None

        self.line_mode = True

        self.max_payload_size = request.MAX_PAYLOAD_SIZE

    def protocol_version(self):

        """Ask for the protocol version

        will return a deferred that will get called with

        the request object when completed.

        """

        p = ProtocolVersion(self)

        p.start()

        return p.deferred

    def dataReceived(self, data):

        """Extend dataReceived.

        First reads the protocol hello line then switch back to len+data.

        """

        if self.line_mode:

            # first read the hello line, then back to binary.

            try:

                pos = data.index("\r\n")

            except ValueError:

                return

            self.line_mode = False

            data = data[pos + 2:]

        request.RequestHandler.dataReceived(self, data)

    def dummy_authenticate(self, credentials, metadata=None):

        """Authenticate to a server using the 'dummy auth' provider.

        Return a deferred that will get called with the request

        object when completed.

        """

        p = Authenticate(self, {'dummy_token': credentials},

                         metadata=metadata)

        p.start()

        return p.deferred

    def simple_authenticate(self, username, password, metadata=None):

        """Authenticate to a server using the username and password."""

        auth_parameters = {

            'username': username,

            'password': password,

        }

        p = Authenticate(self, auth_parameters, metadata=metadata)

        p.start()

        return p.deferred

    def handle_ROOT(self, message):

        """Handle incoming ROOT message.

        Will notify if someone is waiting for this information.

        """

        self.root_id = message.root.node

        if self.root_id_defers:

            for d in self.root_id_defers:

                d.callback(self.root_id)

            self.root_id_defers = []

    def handle_NODE_STATE(self, message):

        """Handle incoming NODE_STATE."""

        self.notify_node_state(message.node_state)

    def handle_NOTIFY_SHARE(self, message):

        """Handle incoming NOTIFY_SHARE."""

        self.notify_share_change(message.notify_share)

    def handle_SHARE_DELETED(self, message):

        """Handle incoming SHARE_DELETED."""

        self.notify_share_deleted(message.share_deleted)

    def handle_SHARE_ACCEPTED(self, message):

        """Handle incoming SHARE_ACCEPTED."""

        self.notify_share_answer(message.share_accepted)

    def handle_VOLUME_CREATED(self, message):

        """Handle incoming VOLUME_CREATED"""

        assert message.type == protocol_pb2.Message.VOLUME_CREATED

        msg = message.volume_created

        vol = None

        if self._volume_created_callback is not None:

            if msg.type == protocol_pb2.Volumes.ROOT:

                vol = volumes.RootVolume.from_msg(msg.root)

            elif msg.type == protocol_pb2.Volumes.SHARE:

                vol = volumes.ShareVolume.from_msg(msg.share)

            elif msg.type == protocol_pb2.Volumes.UDF:

                vol = volumes.UDFVolume.from_msg(msg.udf)

            else:

                msg = "Message.volume_created's type is not valid: %s" % \

                      message.volume_created.type

                raise TypeError(msg)

            self._volume_created_callback(vol)

    def handle_VOLUME_DELETED(self, message):

        """Handle incoming VOLUME_DELETED."""

        assert message.type == protocol_pb2.Message.VOLUME_DELETED

        if self._volume_deleted_callback is not None:

            vol_id = uuid.UUID(message.volume_deleted.volume)

            self._volume_deleted_callback(vol_id)

    def handle_VOLUME_NEW_GENERATION(self, message):

        """Handle incoming VOLUME_NEW_GENERATION."""

        assert message.type == protocol_pb2.Message.VOLUME_NEW_GENERATION

        if self._volume_new_generation_callback is not None:

            volume = message.volume_new_generation.volume

            if volume != request.ROOT:

                volume = uuid.UUID(volume)

            generation = message.volume_new_generation.generation

            self._volume_new_generation_callback(volume, generation)

    def handle_BEGIN_CONTENT(self, message):

        """Accept and discard a misplaced BEGIN_CONTENT.

        It can happen that while the server receives a PUT_CONTENT request

        and that it tells us to BEGIN_CONTENT, we cancelled the request,

        received the OK, and this side's request is gone, so receive this

        message here.

        """

    def handle_FREE_SPACE_INFO(self, message):

        """Handle unsolicited FREE_SPACE_INFO."""

        self.notify_free_space(message.free_space_info)

    def handle_ACCOUNT_INFO(self, message):

        """Handle unsolicited ACCOUNT_INFO."""

        self.notify_account_info(message.account_info)

    def get_root(self):

        """Get the root id through a deferred."""

        if self.root_id is not None:

            return defer.succeed(self.root_id)

        else:

            d = defer.Deferred()

            self.root_id_defers.append(d)

            return d

    def make_dir(self, share, parent, name):

        """Create a directory named name on the node parent

        the new node id will be on request.new_dir_id.

        """

        p = MakeDir(self, share, parent, name)

        p.start()

        return p.deferred

    def make_file(self, share, parent, name):

        """Create a file named name on the node parent

        the new node id will be on request.new_file_id.

        """

        p = MakeFile(self, share, parent, name)

        p.start()

        return p.deferred

    def move(self, share, node, new_parent, new_name):

        """Move a node."""

        p = Move(self, share, node, new_parent, new_name)

        p.start()

        return p.deferred

    def unlink(self, share, node):

        """Unlink a node."""

        p = Unlink(self, share, node)

        p.start()

        return p.deferred

    def get_content(self, share, node, a_hash, offset=0,

                    callback=None, node_attr_callback=None):

        """Get the content of node with 'a_hash'.

        the content will be on request.content

        or callback will be called for every piece that arrives.

        """

        req = self.get_content_request(share, node, a_hash, offset,

                                       callback, node_attr_callback)

        return req.deferred

    def get_content_request(self, share, node, a_hash, offset=0,

                            callback=None, node_attr_callback=None):

        """Get the content of node with 'a_hash', return the request.

        The content will be on request.content, or callback will be

        called for every piece that arrives.

        """

        p = GetContent(self, share, node, a_hash, offset,

                       callback, node_attr_callback)

        p.start()

        return p

    def put_content(self, share, node, previous_hash, new_hash,

                    crc32, size, deflated_size, fd, upload_id=None,

                    upload_id_cb=None, magic_hash=None):

        """Put the content of fd into file node."""

        req = self.put_content_request(share, node, previous_hash, new_hash,

                                       crc32, size, deflated_size, fd,

                                       upload_id=upload_id,

                                       upload_id_cb=upload_id_cb,

                                       magic_hash=magic_hash)

        return req.deferred

    def put_content_request(self, share, node, previous_hash, new_hash,

                            crc32, size, deflated_size, fd, upload_id=None,

                            upload_id_cb=None, magic_hash=None):

        """Put the content of fd into file node, return the request."""

        p = PutContent(self, share, node, previous_hash, new_hash,

                       crc32, size, deflated_size, fd,

                       upload_id=upload_id,

                       upload_id_cb=upload_id_cb,

                       magic_hash=magic_hash)

        p.start()

        return p

    def query(self, items):

        """Get the current hash for items if changed.

        'items' is a list of (node, hash) tuples.

        """

        r = MultiQuery(self, items)

        r.start()

        return r.deferred

    def get_delta(self, share_id, from_generation=None, callback=None,

                  from_scratch=False):

        """Get a delta for a share_id

        'share_id' is the share_id which we want to query.

        'from_generation' is the generation which we are at.

        'callback' can be specified to get deltas as they get instead of

            getting them all at once at the end.

        'from_scratch' at True means list all live nodes.

        """

        r = GetDelta(self, share_id, from_generation, callback, from_scratch)

        r.start()

        return r.deferred

    def get_free_space(self, share_id):

        """Get quota info for the given share (or the user's own space)."""

        r = FreeSpaceInquiry(self, share_id)

        r.start()

        return r.deferred

    def get_account_info(self):

        """Get account information (like purchased space etc.)."""

        r = AccountInquiry(self)

        r.start()

        return r.deferred

    def set_node_state_callback(self, callback):

        """Define the function to be called when a node_state message arrives

        The function will be called with the message as argument.

        """

        self._node_state_callback = callback

    def notify_node_state(self, node_state):

        """Call the current node state callback, if any, with the share, node,

        and hash given in the message.

        @param: node_state - a (raw) NodeState message

        """

        if self._node_state_callback:

            self._node_state_callback(node_state.share, node_state.node,

                                      node_state.hash)

    def set_free_space_callback(self, callback):

        """Set the quota notification callback.

        It will be called with the share id and free bytes.

        """

        self._free_space_callback = callback

    def notify_free_space(self, free_space_info):

        """Call the current quota info callback, if any, with the share

        and available bytes.

        @param: free_space_info - a (raw) FreeSpaceInfo message

        """

        if self._free_space_callback:

            self._free_space_callback(free_space_info.share_id,

                                      free_space_info.free_bytes)

    def set_account_info_callback(self, callback):

        """Set the account info notification callback; the callback

        will be passed a raw AccountInfo structure when it is called.

        """

        self._account_info_callback = callback

    def notify_account_info(self, account_info):

        """Call the current account info callback, if any."""

        if self._account_info_callback:

            self._account_info_callback(account_info)

    def set_share_change_callback(self, callback):

        """Set the callback when something changed regarding a share."""

        if callable(callback):

            self._share_change_callback = callback

        else:

            raise TypeError('callback for share_change must be callable')

    def notify_share_change(self, notify_share):

        """Call the current changed share callback, if any, with the

        notify info.

        @param notify_share: - a NotifyShare message

        """

        if self._share_change_callback:

            info = sharersp.NotifyShareHolder.load_from_msg(notify_share)

            self._share_change_callback(info)

    def set_share_delete_callback(self, callback):

        """Set the callback when something changed regarding a share."""

        if callable(callback):

            self._share_delete_callback = callback

        else:

            raise TypeError('callback for share_delete must be callable')

    def notify_share_deleted(self, share_deleted):

        """Call the current changed share callback, if any, with the

        notify info.

        @param msg: - a (raw) NotifyShare message

        """

        if self._share_delete_callback:

            self._share_delete_callback(uuid.UUID(share_deleted.share_id))

    def set_share_answer_callback(self, callback):

        """Define the function to be called when a share answer is received."""

        if callable(callback):

            self._share_answer_callback = callback

        else:

            raise TypeError('callback for share_answer must be callable')

    def notify_share_answer(self, msg):

        """Call the current share answer callback, if any, with the info.

        @param msg: - a (raw) ShareAccepted message

        """

        if self._share_answer_callback:

            if msg.answer == protocol_pb2.ShareAccepted.YES:

                answer = "Yes"

            elif msg.answer == protocol_pb2.ShareAccepted.NO:

                answer = "No"

            else:

                raise ValueError("Not supported ShareAccepted answer")

            self._share_answer_callback(uuid.UUID(msg.share_id), answer)

    def set_volume_created_callback(self, callback):

        """Set the callback for volume creation notification."""

        if callable(callback):

            self._volume_created_callback = callback

        else:

            raise TypeError('callback for volume_created must be callable')

    def set_volume_deleted_callback(self, callback):

        """Set the callback for volume deletion notification."""

        if callable(callback):

            self._volume_deleted_callback = callback

        else:

            raise TypeError('callback for volume_deleted must be callable')

    def set_volume_new_generation_callback(self, callback):

        """Set the callback for volume new generation notification."""

        if callable(callback):

            self._volume_new_generation_callback = callback

        else:

            raise TypeError('callback for volume_new_gen must be callable')

    def create_share(self, node, share_to, name, access_level):

        """Create a share to other user.

        node: which node to share.

        share_to: the id of the receiving user.

        name: the name of the share

        access_level: the permissions on the share

        There's no need to indicate where the node lives, as it only can be

        in own root (there's no re-sharing).

        """

        r = CreateShare(self, node, share_to, name, access_level)

        r.start()

        return r.deferred

    def delete_share(self, share_id):

        """Delete a share we have offered.

        @param share_id: the id of the share to delete

        """

        r = DeleteShare(self, share_id)

        r.start()

        return r.deferred

    def accept_share(self, share_id, answer):

        """Accept (or not) a share from other user.

        share_id: the share id

        answer: if it was accepted ("Yes") or not ("No")

        """

        r = AcceptShare(self, share_id, answer)

        r.start()

        return r.deferred

    def list_shares(self):

        """List all the shares the user is involved.

        This includes the shares the user created, and those that were

        shared to her.

        """

        p = ListShares(self)

        p.start()

        return p.deferred

    def create_udf(self, path, name):

        """Create a User Defined Folder.

        @param path: the path in disk to the UDF.

        @param name: the name of the UDF.

        """

        p = CreateUDF(self, path, name)

        p.start()

        return p.deferred

    def list_volumes(self):

        """List all the volumes the user has.

        This includes the volumes:

            - all the user's UDFs.

            - all the shares the user has accepted.

            - the root-root volume.

        """

        p = ListVolumes(self)

        p.start()

        return p.deferred

    def change_public_access(self, share_id, node_id, is_public):

        """Change the public access of a file."""

        p = ChangePublicAccess(self, share_id, node_id, is_public)

        p.start()

        return p.deferred

    def list_public_files(self):

        """List the public files."""

        p = ListPublicFiles(self)

        p.start()

        return p.deferred

    def delete_volume(self, volume_id):

        """Delete 'volume' on the server, removing the associated tree.

        @param volume: the id of the volume to delete.

        """

        p = DeleteVolume(self, volume_id)

        p.start()

        return p.deferred

    def query_caps(self, caps):

        """Query the server to discover its capabilitis.

        The server should answer if it supports or not all the given

        caps.

        """

        r = QuerySetCaps(self, caps)

        r.start()

        return r.deferred

    def set_caps(self, caps):

        """Set the server to this capabilities."""

        r = QuerySetCaps(self, caps, set_mode=True)

        r.start()

        return r.deferred

class GetContent(request.Request):

    """A Request to get the content of a node id.

    @ivar data: the content of the node (available upon success)

    """

    __slots__ = ('share', 'node_id', 'hash', 'offset', 'callback',

                 'node_attr_callback', 'parts', 'data')

    def __init__(self, protocol, share, node_id, a_hash,

                 offset=0, callback=None, node_attr_callback=None):

        """Request the content of node with 'a_hash'.

        @param protocol: the request handler

        @param share: the share node or root

        @param node_id: the node id of the node we want to read

        @param a_hash: the hash of the content of the version we have

        @param offset: offset for reading

        @param callback: function to call when data arrives

        """

        request.Request.__init__(self, protocol)

        self.share = share

        self.node_id = node_id

        self.hash = a_hash

        self.offset = offset

        self.callback = callback

        self.node_attr_callback = node_attr_callback

        self.parts = []

    def _start(self):

        """Send GET_CONTENT."""

        message = protocol_pb2.Message()

        message.type = protocol_pb2.Message.GET_CONTENT

        message.get_content.node = str(self.node_id)

        message.get_content.hash = str(self.hash)

        message.get_content.share = self.share

        message.get_content.offset = self.offset

        self.sendMessage(message)

    def processMessage(self, message):

        """Process messages."""

        if message.type == protocol_pb2.Message.NODE_ATTR:

            if self.node_attr_callback is not None:

                self.node_attr_callback(

                    deflated_size=message.node_attr.deflated_size,

                    size=message.node_attr.size,

                    hash=message.node_attr.hash,

                    crc32=message.node_attr.crc32)

        elif message.type == protocol_pb2.Message.BYTES:

            if self.cancelled:

                # don't care about more bytes if already cancelled

                return

            if self.callback is not None:

                self.callback(message.bytes.bytes)

            else:

                self.parts.append(message.bytes.bytes)

        elif message.type == protocol_pb2.Message.EOF:

            if self.cancelled:

                # eof means that the cancel request arrived late. this is the

                # end.

                self.done()

                return

            if self.callback is None:

                self.data = "".join(self.parts)

            self.done()

        elif message.type == protocol_pb2.Message.OK:

            self.done()

        elif message.type == protocol_pb2.Message.CANCELLED:

            self.error(request.RequestCancelledError("CANCELLED"))

        else:

            self._default_process_message(message)

    def _cancel(self):

        """Cancel the current download."""

        message = protocol_pb2.Message()

        message.type = protocol_pb2.Message.CANCEL_REQUEST

        self.sendMessage(message)

class ListShares(request.Request):

    """List all the shares the user is involved.

    This includes the shares the user created, and those that were

    shared to her.

    """

    __slots__ = ('shares',)

    def _start(self):

        """Send the LIST_SHARES message to the server."""

        message = protocol_pb2.Message()

        message.type = protocol_pb2.Message.LIST_SHARES

        self.sendMessage(message)

        self.shares = []

    def processMessage(self, message):

        """Process the answer from the server."""

        if message.type == protocol_pb2.Message.SHARES_INFO:

            share = sharersp.ShareResponse.load_from_msg(message.shares)