FileTransferSession.py

# Copyright (C) 2009-2011 AG Projects. See LICENSE for details.
#

from Foundation import NSLocalizedString

import datetime
import os
import re
import time
import uuid

from collections import deque

from application.notification import NotificationCenter, IObserver, NotificationData
from application.python import Null
from sipsimple.account import Account, BonjourAccount
from sipsimple.configuration.settings import SIPSimpleSettings
from sipsimple.core import ToHeader, SIPURI, Route
from sipsimple.lookup import DNSLookup
from sipsimple.session import Session
from sipsimple.streams.msrp.filetransfer import FileTransferStream, FileSelector
from sipsimple.threading.green import run_in_green_thread
from sipsimple.util import ISOTimestamp
from zope.interface import implementer

from BlinkLogger import BlinkLogger
from HistoryManager import FileTransferHistory, ChatHistory
from util import call_later, run_in_gui_thread, format_size, format_date, format_identity_to_string


def format_duration(t):
    s = ""
    if t.days > 0:
        s += "%d days, " % t.days
    s += "%02d:%02d:%02d" % (t.seconds / 3600, (t.seconds / 60) % 60, t.seconds % 60)
    return s


class FileTransferInfo(object):
    def __init__(self, transfer_id=None, direction=None, local_uri=None, remote_uri=None, file_path=None, bytes_transfered=0, file_size=0, status=None):
        self.transfer_id       = transfer_id
        self.direction         = direction
        self.local_uri         = local_uri
        self.remote_uri        = remote_uri
        self.file_path         = file_path
        self.bytes_transfered  = bytes_transfered
        self.file_size         = file_size
        self.status            = status


@implementer(IObserver)
class FileTransfer(object):

    direction = None    # to be set by subclasses

    session = None
    stream = None
    handler = None

    status = ""

    remote_identity = None
    target_uri = None
    # For a Bonjour transfer: the neighbour's instance id, resolved while
    # they are still on the network. Messages with a neighbour are filed
    # under it, so a transfer filed under the link-local address it happened
    # to use ended up as a second, unreachable conversation with the same
    # person -- which is exactly what happened in the live database.
    remote_device_id = None

    transfer_rate = None
    last_rate_pos = 0
    last_rate_time = None
    rate_history = None
    ft_info = None

    bytes = 0
    total_bytes = 0
    progress = None

    start_time = None
    end_time = None

    @property
    def file_selector(self):
        return self.stream.file_selector if self.stream is not None else None

    @property
    def progress_text(self):
        return self.status

    @property
    def file_size(self):
        return self.file_selector and self.file_selector.size

    @property
    def file_type(self):
        return self.file_selector and self.file_selector.type

    @property
    def file_path(self):
        return self.file_selector.name if self.file_selector is not None else ''

    @property
    def file_pos(self):
        return self.bytes

    def format_progress(self):
        t = NSLocalizedString("Transferred %s of ", "Label") % format_size(self.bytes, 1024) + format_size(self.total_bytes, 1024)
        if self.transfer_rate is not None:
            if self.transfer_rate == 0:
                status = t + " (" + NSLocalizedString("stalled", "Label") + ")"
            else:
                eta = (self.total_bytes - self.bytes) / self.transfer_rate
                if eta < 60:
                    time_left = NSLocalizedString("Less than 1 minute", "Label")
                elif eta < 60*60:
                    e = eta/60
                    time_left = NSLocalizedString("About %i minutes", "Label") % e
                else:
                    time_left = "%s left" % format_duration(datetime.timedelta(seconds=eta))

                status = t + " - %s/s - %s" % (format_size(self.transfer_rate, bits=True), time_left)
        else:
            status = t
        return status

    def update_transfer_rate(self):
        now = time.time()
        if self.last_rate_time is not None:
            dt = now - self.last_rate_time
            if dt >= 1:
                db = self.file_pos - self.last_rate_pos
                transfer_rate = db / dt
                self.rate_history.append(transfer_rate)
                self.last_rate_time = now
                self.last_rate_pos = self.file_pos
                self.transfer_rate = int(sum(self.rate_history) / len(self.rate_history))
        else:
            self.last_rate_time = now
            self.last_rate_pos = self.file_pos
            self.rate_history = deque(maxlen=5)

        notification_center = NotificationCenter()
        notification_center.post_notification("BlinkFileTransferSpeedDidUpdate", sender=self)

    def resolveRemoteDeviceId(self, uri=None):
        """Note which neighbour this is, while they are still discoverable.

        Done when the transfer starts rather than when it is filed: by the
        time a transfer ends the neighbour may have gone, and a device id
        cannot be recovered from a link-local address afterwards -- that
        address belonged to a machine on a network, not to a person.

        The party's own SIP URI is asked first, because it carries the
        instance id as a parameter and that is the answer rather than a
        guess at it. The contact list is the fallback, and it matches on an
        address, which two machines can share.

        Only Bonjour has this problem. A SIP conversation is filed under the
        peer's address, which is the same address the transfer has.
        """
        if self.account is not BonjourAccount():
            return

        try:
            instance_id = uri.parameters.get('instance_id') if uri is not None else None
        except AttributeError:
            instance_id = None
        if instance_id:
            instance_id = str(instance_id)
            if instance_id.startswith('urn:uuid:'):
                instance_id = instance_id[9:]
            self.remote_device_id = instance_id
            return

        try:
            from AppKit import NSApp
            contact = NSApp.delegate().contactsWindowController.getBonjourContact(
                None, self.remote_identity, online_only=True)
        except Exception as e:
            BlinkLogger().log_error('Cannot tell which neighbour %s is: %s'
                                    % (self.remote_identity, e))
            return
        if contact is not None and getattr(contact, 'id', None):
            self.remote_device_id = contact.id

    @property
    def history_remote_uri(self):
        """The address this transfer is filed under in the message history.

        A Bonjour neighbour is one conversation whichever address their
        machine is answering on today, so their transfers belong under the
        same instance id their messages do. Everyone else is already filed
        under the address the transfer used.
        """
        return self.remote_device_id or self.ft_info.remote_uri

    def add_to_history(self):
        FileTransferHistory().add_transfer(transfer_id=self.ft_info.transfer_id, direction=self.ft_info.direction, local_uri=self.ft_info.local_uri, remote_uri=self.ft_info.remote_uri, file_path=self.ft_info.file_path, bytes_transfered=self.ft_info.bytes_transfered, file_size=self.ft_info.file_size or 0, status=self.ft_info.status)

        message  = "<h3>%s File Transfer</h3>" % self.ft_info.direction.capitalize()
        message += "<p>%s (%s)" % (self.ft_info.file_path, format_size(self.ft_info.file_size or 0))
        media_type = 'file-transfer'
        local_uri = self.ft_info.local_uri
        # NOT ft_info.remote_uri: that stays the address the transfer
        # actually used, which is what the File Transfers window shows.
        # This is the conversation the transfer belongs to.
        remote_uri = self.history_remote_uri
        direction = self.ft_info.direction
        status = 'delivered' if self.ft_info.status == 'completed' else 'failed'
        # Both ends used to be the remote party, so an outgoing transfer
        # claimed the peer had sent it to themselves. Anything reading a
        # name off these rows -- the conversation list, the Bonjour group --
        # got the wrong one, or none.
        peer = self.ft_info.remote_uri
        if direction == 'incoming':
            cpim_from, cpim_to = peer, local_uri
        else:
            cpim_from, cpim_to = local_uri, peer
        timestamp = str(ISOTimestamp.now())

        ChatHistory().add_message(self.ft_info.transfer_id, media_type, local_uri, remote_uri, direction, cpim_from, cpim_to, timestamp, message, "html", "0", status)

    def end(self):
        if self._ended:
            return

        self._ended = True

        if self.session is not None:
            print('Session will end %s in state %s' % (self.session, self.session.state))
            self.session.end()

        status = NSLocalizedString("Cancelled", "Label")
        self._terminate(failure_reason="Cancelled", failure_status=status)
        self.log_info("File Transfer has been cancelled")

    def _terminate(self, failure_reason=None, failure_status=None):
        notification_center = NotificationCenter()

        if failure_reason is None:
            self.log_info("File Transfer ended (%i of %i bytes transferred)" % (self.bytes, self.total_bytes))
            self.end_time = datetime.datetime.now()
            t = NSLocalizedString("Completed transfer of ", "Label")
            self.status = t + format_size(self.file_size) + NSLocalizedString(" in ", "Label") + format_duration(self.end_time-self.start_time) + " " + format_date(self.end_time)
            self.ft_info.status = "completed"
            self.ft_info.bytes_transfered = self.file_selector.size
        else:
            if failure_status is None:
                if self.total_bytes:
                    self.log_info("File Transfer was interrupted")
                    self.status = NSLocalizedString("Transferred %s of ", "Label") % format_size(self.bytes, 1024) + format_size(self.total_bytes) + " - " +  failure_reason
                else:
                    self.log_info("File Transfer was cancelled")
                    self.status = failure_reason
            else:
                self.log_info("File Transfer failed: %s" % failure_reason)
                self.status = failure_status
            self.ft_info.status = "failed"
            self.ft_info.bytes_transfered = self.file_pos

        if self.session is not None and self.stream is not None and self.handler is not None:
            notification_center.remove_observer(self, sender=self.stream)
            notification_center.remove_observer(self, sender=self.handler)

        self.session = None
        self.stream = None
        self.handler = None
        self.transfer_rate = None

        notification_center.post_notification("BlinkFileTransferDidEnd", sender=self, data=NotificationData(file_path=self.ft_info.file_path, error=failure_reason is not None))
        self.add_to_history()

    @run_in_gui_thread
    def handle_notification(self, notification):
        handler = getattr(self, '_NH_%s' % notification.name, Null)
        handler(notification)

    def _NH_MediaStreamDidNotInitialize(self, notification):
        self._terminate(failure_reason=notification.data.reason)

    def _NH_MediaStreamDidStart(self, notification):
        self.log_info("File transfer in progress...")
        self.ft_info.status = "transferring"
        self.start_time = datetime.datetime.now()
        notification.center.post_notification("BlinkFileTransferDidStart", sender=self)

    def _NH_FileTransferHandlerDidInitialize(self, notification):
        self.progress = None
        # The filename is now properly initialized
        self.ft_info.file_path = self.file_path
        notification.center.post_notification("BlinkFileTransferDidInitialize", self)

    def _NH_FileTransferHandlerError(self, notification):
        self.log_info("Error while handling file transfer: %s" % notification.data.error)

    def _NH_FileTransferHandlerDidEnd(self, notification):
        if self.direction == 'incoming':
            if self.stream.direction == 'sendonly':
                call_later(3, self.session.end)
            else:
                call_later(1, self.session.end)
        else:
            self.session.end()
        self._terminate(failure_reason=notification.data.reason)

    def _NH_FileTransferHandlerDidStart(self, notification):
        self.status = NSLocalizedString("Starting File Transfer...", "Label")
        notification.center.post_notification("BlinkFileTransferUpdate", sender=self)

    def _NH_FileTransferHandlerProgress(self, notification):
        self.bytes = notification.data.transferred_bytes
        self.total_bytes = notification.data.total_bytes
        progress = int(self.bytes * 100 / self.total_bytes)
        if self.progress is None or progress > self.progress:
            self.progress = progress
            self.update_transfer_rate()
            self.status = self.format_progress()
            notification.center.post_notification("BlinkFileTransferProgress", sender=self, data=NotificationData(progress=progress))

    @property
    def log_identity(self):
        """Who this transfer is with, as the log should name them.

        The address for anyone who has a stable one. For a Bonjour
        neighbour, their instance id as well: the address is whichever
        link-local one their machine answers on this afternoon, while the
        id is the conversation the file actually belongs to -- the key its
        history row, its folder under file_transfers and its unread badge
        all use. Without it a line about a transfer cannot be matched to
        the conversation it lands in.
        """
        if self.remote_device_id:
            return '%s (%s)' % (self.remote_identity, self.remote_device_id)
        return str(self.remote_identity)

    def log_info(self, text):
        BlinkLogger().log_info("[%s file transfer with %s] %s" % (self.direction.title(), self.log_identity, text))

    def lookup_destination(self, uri):
        self.log_info("Lookup destination for %s" % uri)

        is_ip_address = re.match("^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", uri.host.decode()) or ":" in uri.host.decode()

        if self.account is BonjourAccount() and is_ip_address:
            tls_name = self.account.sip.tls_name
            transport = uri.transport.decode() if isinstance(uri.transport, bytes) else uri.transport
            transport = 'tls' if uri.secure else transport.lower()
            port = uri.port or (5061 if transport=='tls' else 5060)
            routes = [Route(address=uri.host, port=port, transport=transport, tls_name=tls_name or uri.host)]
            self.connect(routes)
            return

        self.lookup_dns(uri)

    @run_in_green_thread
    def lookup_dns(self, target_uri):
        settings = SIPSimpleSettings()
        lookup = DNSLookup()
        notification_center = NotificationCenter()
        notification_center.add_observer(self, sender=lookup)

        if self.account is BonjourAccount():
            tls_name = target_uri.host
        else:
            if self.account.id.domain == target_uri.host.decode():
                tls_name = self.account.sip.tls_name or self.account.id.domain
            else:
                tls_name = target_uri.host.decode()

        uri = target_uri
        if self.account is not BonjourAccount():
            if self.account.sip.outbound_proxy is not None:
                proxy = self.account.sip.outbound_proxy
                uri = SIPURI(host=proxy.host, port=proxy.port, parameters={'transport': proxy.transport})
                tls_name = self.account.sip.tls_name or proxy.host
                self.log_info("Starting DNS lookup for %s via proxy %s" % (target_uri.host.decode(), uri))
            elif self.account.sip.always_use_my_proxy:
                uri = SIPURI(host=self.account.id.domain)
                tls_name = self.account.sip.tls_name or self.account.id.domain
                self.log_info("Starting DNS lookup for %s via proxy of account %s" % (target_uri.host.decode(), self.account.id))

        self.log_info("Starting DNS lookup for %s" % target_uri.host.decode())

        lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=tls_name)

    def _NH_DNSLookupDidSucceed(self, notification):
        notification.center.remove_observer(self, sender=notification.sender)
        routes = notification.data.result
        if not routes:
            self._terminate(failure_reason="Destination not found")
            return

        self.connect(routes)

    def _NH_DNSLookupDidFail(self, notification):
        notification.center.remove_observer(self, sender=notification.sender)
        if self._ended:
            return
        self.log_info("DNS Lookup for SIP routes failed: '%s'" % notification.data.error)
        status = NSLocalizedString("DNS Lookup failed", "Label")
        self._terminate(failure_reason=notification.data.error, failure_status=status)

    def connect(self, routes):
        notification_center = NotificationCenter()

        if self._ended:
            self.log_info("File transfer has already ended")
            return
            
        self.session = Session(self.account)
        self.stream = FileTransferStream(self._file_selector, 'sendonly' if self.transfer_type=='push' else 'recvonly', transfer_id=self.ft_info.transfer_id)
        self.handler = self.stream.handler
        notification_center.add_observer(self, sender=self.stream)
        notification_center.add_observer(self, sender=self.handler)

        self.log_info("Sending %s file transfer request via %s..." % (self.transfer_type, routes[0]))
        self.session.connect(ToHeader(self.target_uri), routes, [self.stream])


class IncomingFileTransferHandler(FileTransfer):
    direction = 'incoming'

    def __init__(self, session, stream):
        self.session = session
        self.stream = stream
        self.handler = stream.handler
        self.account = session.account
        self.remote_identity = format_identity_to_string(session.remote_identity)
        self.resolveRemoteDeviceId(getattr(session.remote_identity, 'uri', None))

    @property
    def target_text(self):
        return "From " + self.remote_identity

    def start(self):
        notification_center = NotificationCenter()
        file_path = self.file_path.decode() if isinstance(self.file_path, bytes) else self.file_path
        self.ft_info = FileTransferInfo(transfer_id=self.stream.transfer_id,
                                        direction='incoming',
                                        local_uri=format_identity_to_string(self.account) if self.account is not BonjourAccount() else 'bonjour@local' ,
                                        file_size=self.file_selector.size,
                                        remote_uri=self.remote_identity,
                                        file_path=file_path)

        self.log_info("Will write file to %s" % file_path)
        self.status = NSLocalizedString("Accepting File Transfer...", "Label")

        notification_center.add_observer(self, sender=self.stream)
        notification_center.add_observer(self, sender=self.handler)

        self.log_info("Initiating Incoming File Transfer")

        notification_center.post_notification("BlinkFileTransferNewIncoming", sender=self)

    def end(self):
        if self.session is not None:
            self.session.end()


class OutgoingPushFileTransferHandler(FileTransfer):
    direction = 'outgoing'
    transfer_type = 'push'

    def __init__(self, account, target_uri, file_path, transfer_id=None):
        self.account = account
        self._file_selector = FileSelector.for_file(file_path)
        self.remote_identity = format_identity_to_string(target_uri)
        self.resolveRemoteDeviceId(target_uri)
        self.target_uri = target_uri
        self._ended = False
        # The id this transfer is known by ON BOTH SIDES. RFC 5547 puts it
        # in the SDP as a=file-transfer-id and the receiving stream adopts
        # it, so whatever is set here is what the peer will file the file
        # under. A conversation sending a file has already built its
        # envelope, its bubble and its history row around an id; without
        # passing it in, the handler minted a second one, the peer stored
        # the file under THAT, and the two sides had no id in common --
        # which is why a removal sent by the peer named a message this
        # side had never heard of and took nothing off the screen.
        self.transfer_id = str(transfer_id) if transfer_id else None

    @property
    def target_text(self):
        t = NSLocalizedString("To %s", "Label") % self.remote_identity
        f = NSLocalizedString("from account %s", "Label") % self.account.id
        return t + " " + f

    def retry(self):
        self._ended = False
        self._file_selector = FileSelector.for_file(self._file_selector.name)
        self.bytes = 0
        self.total_bytes = 0
        self.progress = None
        self.last_rate_pos = 0
        self.last_rate_time = None
        self.rate_history = None
        self.session = None
        self.stream = None
        self.handler = None
        self.transfer_rate = None
        self.start(restart=True)

    def start(self, restart=False):
        notification_center = NotificationCenter()
        file_path = self._file_selector.name.decode() if isinstance(self._file_selector.name, bytes) else self._file_selector.name

        self.ft_info = FileTransferInfo(transfer_id=self.transfer_id or str(uuid.uuid4()),
                                        direction='outgoing',
                                        file_size=self._file_selector.size,
                                        local_uri=format_identity_to_string(self.account) if self.account is not BonjourAccount() else 'bonjour@local',
                                        remote_uri=self.remote_identity,
                                        file_path=file_path)

        self.status = NSLocalizedString("Offering File...", "Label")
        self.ft_info.status = "proposing"

        self.lookup_destination(self.target_uri)

        if restart:
            notification_center.post_notification("BlinkFileTransferWillRestart", self)
        else:
            notification_center.post_notification("BlinkFileTransferNewOutgoing", sender=self)

    def _NH_FileTransferHandlerHashProgress(self, notification):
        progress = int(notification.data.processed * 100 / notification.data.total)
        if self.progress is None or progress > self.progress:
            self.progress = progress
            notification.center.post_notification('BlinkFileTransferHashProgress', sender=self, data=NotificationData(progress=progress))


class OutgoingPullFileTransferHandler(FileTransfer):
    direction = 'outgoing'
    transfer_type = 'pull'

    def __init__(self, account, target_uri, filename, hash):
        self.account = account
        self._file_selector = FileSelector(name=os.path.basename(filename), hash=hash)
        self.remote_identity = format_identity_to_string(target_uri)
        self.resolveRemoteDeviceId(target_uri)
        self.target_uri = SIPURI.new(target_uri)
        self._ended = False

    @property
    def target_text(self):
        f = NSLocalizedString("From %s", "Label") % self.target_uri
        t = NSLocalizedString("to account %s", "Label") % self.account.id
        return f + " " + t

    def start(self):
        notification_center = NotificationCenter()
        file_path = self._file_selector.name.decode() if isinstance(self._file_selector.name, bytes) else self._file_selector.name
        self.ft_info = FileTransferInfo(transfer_id=str(uuid.uuid4()),
                                        direction='incoming',
                                        local_uri=format_identity_to_string(self.account) if self.account is not BonjourAccount() else 'bonjour@local',
                                        file_size=0,
                                        remote_uri=self.remote_identity,
                                        file_path=file_path)

        self.log_info("Pull File Transfer Request started %s" % file_path)
        notification_center.post_notification("BlinkFileTransferNewOutgoing", self)

        self.status = NSLocalizedString("Requesting File...", "Label")
        self.ft_info.status = "requesting"

        self.lookup_destination(self.target_uri)