Net::REPL::Client: Allow read/print portions to be overridden separately.

[thrasher.git] / perl / lib / Thrasher / Plugin / ProxyFileTransfer.pm

package Thrasher::Plugin::ProxyFileTransfer;

use strict;
use warnings;

=pod

=head1 NAME

Thrasher::Plugin::ProxyFileTransfer - using a XEP-0065-compliant
socket proxy, transfer files to and from the legacy service.

=head1 DESCRIPTION

This module provides as much code as possible to implement file
transfer to and from the legacy service, but legacy Protocols will
still need to implement a little bit of stuff to hook everything
together. This will be documented at some point, probably when
somebody asks for it.

Note that this does not permit direct connections; it always uses a
proxy. This works out well for what the authors need. If you need to
permit direct connections, please feel free to submit a patch; it's
certainly possible.

Unfortunately, this is a very complicated interaction. If you have
any interest in working on this, I strongly recommend popping open
something with an XML console (like Psi) and looking at all the XML
flying around, rather than trying to figure out what is going
on by reading this code. This is event-based code at its worst.

=cut

# FIXME: Use bytestream query to see if remote client can support file
# transfers
# FIXME: Verify the account is online when sending a file.

use Thrasher::Callbacks qw(:all);
use Thrasher::Log qw(:all);
use Thrasher::Constants qw(:all);
use Thrasher::XML qw(:all);
use Thrasher::Plugin qw(:all);

use Digest::SHA1 qw(sha1_hex);

use Thrasher::Plugin::Socks;

use Data::Dumper;

# FIXME: It is possible to fire discovery before the proxy is
# configured. If someone wants to transfer files and no proxy
# was found, try again before giving up.
# undef => not yet discovered
# 0 => discovery failed; there is no XEP-0065 proxy
my $proxy_server_jid = undef;
my $proxy_server_port = undef;
my $proxy_server_host = undef;

# This is keyed by stream id
my $file_transfer_data_by_sid = {};

# * Key: Thrasher's file id as returned by thrasher_wrapper_send_file.
# * Value: Thrasher::Plugin::ProxyFileTransfer::FT object.
my $file_transfer_data_by_id = {};

# When the server completes the basic discovery sweep it does,
# finish discovering the proxy's information.
do_when('server_discovery_items',
        sub {
            PROXY_SEARCH:
            for my $item (keys %{$Thrasher::SERVER_INFO}) {
                my ($identities, undef) =
                    @{$Thrasher::SERVER_INFO->{$item}};
                for my $identity (@$identities) {
                    my ($type, $category, $name) = @$identity;
                    if ($type eq 'bytestreams' &&
                        $category eq 'proxy') {
                        $proxy_server_jid = $item;
                        succeeded("proxy_server_found");
                        last PROXY_SEARCH;
                    }
                }
            }

            if (!defined($proxy_server_jid)) {
                log("ERROR: No XEP-0065-compliant SOCKS5 proxy "
                    ."service found on the host server. File "
                    ."transfers WILL NOT WORK.");
                $proxy_server_jid = 0;
                failed("proxy_server_found");
            } else {
                log("Proxy server for file transfers found: ".
                    "$proxy_server_jid; querying for IP");
                $Thrasher::Component::COMPONENT->iq_query
                    ([[$NS_COMPONENT, 'iq'],
                      {to => $proxy_server_jid,
                       from => $Thrasher::Component::COMPONENT->{component_name},
                       type => 'get'},
                      [[[$NS_BYTESTREAMS, 'query'], {}, []]]],
                     sub {
                         my $component = shift;
                         my $iq_params = shift;
                         my $iq_packet = shift;

                         if ($iq_params->{type} eq 'error') {
                             log("ERROR: Bytestreams proxy won't"
                                 ." tell us the IP it is located on."
                                 ." File transfers WILL NOT WORK. "
                                 ."You may need to tweak permissions "
                                 ."to permit "
                                 .$Thrasher::Component::COMPONENT->{component_name}
                                 ." to connect to " .
                                 $proxy_server_jid . ".");
                             return;
                         }

                         # Note that while this query is permitted
                         # to return a zeroconf value, we don't
                         # support that at this time.
                         my $streamhost;
                         {
                             local $@;
                             eval {
                                 $streamhost = extract
                                     ([undef, undef,
                                       save_match(
                                           [[$NS_BYTESTREAMS,
                                             'streamhost'],
                                            {'{}host' => save('host'),
                                             '{}port' => save('port')},
                                            undef])],
                                      $iq_params->{query});
                             };
                             if ($@) {
                                 $streamhost = undef;
                             }
                         }
                         if ($streamhost) {
                             $proxy_server_port = $streamhost->{port};
                             $proxy_server_host = $streamhost->{host};
                             log("Proxy server fully located, file "
                                   . "transfers should work.");
                             succeeded('xep-0065_proxy_server_located');
                             _initialize_the_wrapper();
                         }
                         else {
                             log("Error: Proxy server will not "
                                   . "identify the ip/port it is "
                                   . "on. File transfers WILL NOT "
                                   . "WORK.");
                         }
                     });
            }
        },

        sub {
            log("ERROR: No XEP-0065-compliant SOCKS5 proxy service "
                ."found on the host server, because service discovery "
                ."against the server failed (which is really weird!). "
                ."File transfers WILL NOT WORK.");
            $proxy_server_jid = 0;
            failed("proxy_server_found");
        });

register_plugin({client_iq_handlers =>
                 {$NS_BYTESTREAMS =>
                  {set => \&handle_set_bytestreams},
                  $NS_FILE_TRANSFER =>
                  {set => \&handle_set_file_transfer},
                  $NS_STREAM_INITIATION =>
                  {set => \&handle_set_stream_initiation}
                 },
                 features => [$NS_BYTESTREAMS,
                              $NS_STREAM_INITIATION,
                              $NS_FILE_TRANSFER]});

#<iq type="set" to="profgilzot@aim.transport" id="ab06a" >
#<query xmlns="http://jabber.org/protocol/bytestreams" mode="tcp" sid="s5b_5666fd39c31ac262" >
#<streamhost port="7777" host="10.2.68.46" jid="proxy65.imtest.barracuda.com" >
#<proxy xmlns="http://affinix.com/jabber/stream"/>
#</streamhost>
#<fast xmlns="http://affinix.com/jabber/stream"/>
#</query>
#</iq>

sub handle_set_bytestreams {
    my $component = shift;
    my $iq_params = shift;
    my $iq_packet = shift;

    if (!$proxy_server_jid ||
        !$proxy_server_port ||
        !$proxy_server_host) {
        log('Cannot set bytestreams; no proxy server yet known.');
        $component->iq_error($iq_params, 'service_unavailable');
        return;
    }

    my $query = $iq_params->{query};

    my $stream_id = $query->[1]->{'{}sid'};
    if (!defined($stream_id)) {
        log("While attempting to deal with bytestream negotiation, "
            ."got a packet with no sid.");
        $component->iq_error($iq_params, 'bad_request');
        return;
    }

    my $file_data = delete $file_transfer_data_by_sid->{$stream_id};
    if (!defined($file_data)) {
        log("While attempting to find the file data, couldn't "
            ."find data about stream id $stream_id.");
        $component->iq_error($iq_params, 'bad_request');
        return;
    }

    # We are looking for a streamhost that isn't using zeroconf
    my $acceptable_streamhosts = [];

    for my $child (@{$iq_params->{query}->[2]}) {
        if (ref($child) &&
            $child->[0]->[1] eq 'streamhost') {
            my $atts = $child->[1];
            if ($atts->{'{}zeroconf'}) {
                next;
            }

            if ((my $jid = $atts->{'{}jid'}) &&
                (my $host = $atts->{'{}host'}) &&
                (my $port = $atts->{'{}port'})) {
                push @$acceptable_streamhosts,
                {jid => $jid,
                 host => $host,
                 port => $port};
            }
        }
    }

    if (!@$acceptable_streamhosts) {
        log("Could not negotiate an acceptable streamhost.");
        $component->iq_error($iq_params, 'not_acceptable');
        return;
    }

    # FIXME: We really ought to try these in sequence. However,
    # I find myself wondering how many clients actually ship
    # out more than one possibility.
    $file_transfer_data_by_sid->{streamhosts} = $acceptable_streamhosts;

    my $streamhost = $file_transfer_data_by_sid->{streamhosts}->[0];

    log("Streamhost chosen: " . Dumper($streamhost));

    my $host = sha1_hex($stream_id .
                        $iq_params->{from} .
                        $iq_params->{to});

    my $ft = Thrasher::Plugin::ProxyFileTransfer::FT->new();

    # Unfortunately, the raw TCP connection seems to be
    # synchronous and I don't see how to make it
    # unsynchronous. Fortunately, it should be local anyhow.
    $ft->{'socks'} = Thrasher::Plugin::Socks->new(
        ProxyAddr => $streamhost->{'host'},
        ProxyPort => $streamhost->{'port'},
        ConnectAddr => $host,
        ConnectPort => 0,
        EventLoop => $component->{'event_loop'},
        SuccessCallback => sub {
            my ($data_read_sub) = @_;
            log('Got SOCKS data reader.');
            $ft->{'data_reader'} = $data_read_sub;
        },
        FailureCallback => sub { log('Proxy file transfer failure.'); },
    );

    if (!defined($ft->{'socks'})) {
        # FIXME: Try any other streams offered.
        $component->iq_error($iq_params, 'item_not_found');
        return;
    }

    $component->iq_reply
        ($iq_params,
         [[$NS_BYTESTREAMS, 'query'], {},
          [[[$NS_BYTESTREAMS, 'streamhost-used'],
            {jid => $streamhost->{jid}},
            []]]]);

    my $legacy_to_id =
        $component->xmpp_name_to_legacy($iq_params->{from},
                                        $iq_params->{to});

    log("About to initiate purple-side file transfer:");
    my $file_transfer_id = THPPW::thrasher_wrapper_send_file(
        strip_resource($iq_params->{'from'}),
        $legacy_to_id,
        $file_data->{'filename'},
        $file_data->{'size'},
        $file_data->{'desc'},
    );
    if (! $file_transfer_id) {
        log('Failed to set up libpurple xfer!');
        $component->iq_error($iq_params, 'service_unavailable');
        return;
    }
    log("Purple-side init for $file_transfer_id done.");

    $file_transfer_data_by_id->{$file_transfer_id} = $ft;
    $ft->{'id'} = $file_transfer_id;

    $ft->{'socks'}->connect();
}

sub handle_file_transfer {
    my $component = shift;
    my $iq_params = shift;
    my $iq_packet = shift;

    log("Got a request for a file transfer.");
    $component->iq_error($iq_params, 'service_unavailable');
}

sub handle_set_stream_initiation {
    my $component = shift;
    my $iq_params = shift;
    my $iq_packet = shift;

    log("Got a request for a stream initiation.");

    # Extract the information for the <file>.
    my $file_results = recursive_extract
        ($iq_params->{query},

         # Extract the <file> tag and save the require params
         [undef, {'{}id' => save('stream_id')},
          save_match('rec', [[$NS_FILE_TRANSFER, 'file'],
                             {'{}name' => save('filename'),
                              '{}size' => save('size'),
                              '{}date' => save('date', 1),
                              '{}hash' => save('hash', 1)},
                             undef])],

         # go looking for the description.
         # Note we don't implement <range> support right now.
         [undef, undef,
          save_match('rec', [[undef, 'desc'], undef,
                             save_sub('desc', \&text_extractor)],
                     1)]);
    $file_results->{desc} ||= [];
    $file_results->{desc} = join '', @{$file_results->{desc}};

    # Extract the feature negotiation information.
    my $negotiation_results = recursive_extract
        ($iq_params->{query},

         [undef, undef,
          save_match('rec', [[$NS_FEATURE_NEG, 'feature']])],

         [undef, undef, save_match('rec', [[$NS_DATA, 'x']])],

         [undef, undef, save_match('rec', [[$NS_DATA, 'field'],
                                           {type => 'list-single',
                                            var =>
                                                'stream-method'}])],

         # For each option, extract the possible values
         [undef, undef,
          save_sub('options',
                   sub {
                       my $tag = shift;
                       # "If this is an option, extract the
                       # value's text element."
                       if (ref($tag) && $tag->[0]->[1] eq 'option') {
                           for my $child (@{$tag->[2]}) {
                               if (ref($child) &&
                                   $child->[0]->[1] eq 'value') {
                                   return join '', @{$child->[2]};
                               }
                           }
                       }

                       return undef;
                   }
          )]);

    # Validate the basic information we are looking for
    my $valid = 1;
    if (!exists($file_results->{filename})) {
        log("Offered file transfer to legacy user, but no filename given.");
        $valid = 0;
    }
    if (!exists($file_results->{size})) {
        log("Offered file transfer to legacy user, but no size given.");
        $valid = 0;
    }
    if (!$negotiation_results->{options}) {
        log("Offered file transfer to legacy user, but no feature "
            ."negotiations found.");
        $valid = 0;
    }

    if ($valid == 0) {
        $component->iq_error($iq_params, 'bad_request');
        return;
    }

    # Verify the remote server is offering bytestreams
    my $bytestreams_offered = 0;
    for my $option (@{$negotiation_results->{options}}) {
        if ($option eq $NS_BYTESTREAMS) {
            $bytestreams_offered = 1;
        }
    }
    if ($bytestreams_offered == 0) {
        log("Offered a file transfer to legacy user, but no "
            ."bytestreams offer found.");
        $component->iq_error($iq_params, 'bad_request');
    }

    $file_transfer_data_by_sid->{$file_results->{stream_id}} = $file_results;

    # If we've gotten to here, we accept the file transfer offer.
    my $response =
        [[$NS_STREAM_INITIATION, 'si'], {},
         [#[[$NS_FILE_TRANSFER, 'file'], {}, []],
          [[$NS_FEATURE_NEG, 'feature'], {},
           [[[$NS_DATA, 'x'], {type => 'submit'},
             [[[$NS_DATA, 'field'], {var => 'stream-method'},
               [[[$NS_DATA, 'value'], {}, [$NS_BYTESTREAMS]
                ]]]]]]]]];
    $component->iq_reply($iq_params, $response);
}

sub _initialize_the_wrapper {
    THPPW::thrasher_wrapper_ft_init(
        Thrasher::error_wrap("ft_recv_request", \&_ft_recv_request),
        Thrasher::error_wrap("ft_recv_accept", \&_ft_recv_accept),
        Thrasher::error_wrap("ft_recv_start", \&_ft_recv_start),
        Thrasher::error_wrap("ft_recv_cancel", \&_ft_recv_cancel),
        Thrasher::error_wrap("ft_recv_complete", \&_ft_recv_complete),
        Thrasher::error_wrap("ft_send_accept", \&_ft_send_accept),
        Thrasher::error_wrap("ft_send_start", \&_ft_send_start),
        Thrasher::error_wrap("ft_send_cancel", \&_ft_send_cancel),
        Thrasher::error_wrap("ft_send_complete", \&_ft_send_complete),
        Thrasher::error_wrap("ft_read", \&_ft_read),
        Thrasher::error_wrap("ft_write", \&_ft_write),
        Thrasher::error_wrap("ft_data_not_sent", \&_ft_data_not_sent),
    );
}

sub _ft_recv_request {
    my ($id, $jid, $legacy_name, $filename, $size) = @_;
    debug("_ft_recv_request($id, $jid, $legacy_name"
            . ", $filename, $size) called");

    $jid = strip_resource($jid);
    my $session = $Thrasher::Component::COMPONENT->session_for($jid);
    my $stream_id = 's_' . $Thrasher::Component::COMPONENT->get_id();
    my $legacy_jid = $Thrasher::Component::COMPONENT->legacy_name_to_xmpp(
        $jid,
        $legacy_name,
    );
    my $ft = Thrasher::Plugin::ProxyFileTransfer::FT->new(
       'sid' => $stream_id,
       'id' => $id,
       'target_full_jid' => $session->{'full_jid'},
       'initiator_legacy_jid' => $legacy_jid,
    );
    $file_transfer_data_by_id->{$ft->{'id'}} = $ft;

    # Refer to http://xmpp.org/extensions/xep-0065.html &c.
    #
    # FT setup stage 1: offer the file and get "declined" or feature
    # negotiation back.
    #
    # <iq from='c%d@xmpp.transport' to='a@b/R' type='set'>
    #   <si xmlns='http://jabber.org/protocol/si' profile='http://jabber.org/protocol/si/profile/file-transfer' id='s_id937'>
    #     <file xmlns='http://jabber.org/protocol/si/profile/file-transfer' size='694' name='small.file'>
    #       <range/>
    #     </file>
    #     <feature xmlns='http://jabber.org/protocol/feature-neg'>
    #       <x xmlns='jabber:x:data' type='form'>
    #         <field type='list-single' var='stream-method'>
    #           <option>
    #             <value>http://jabber.org/protocol/bytestreams</value>
    #          </option>
    #         </field>
    #       </x>
    #     </feature>
    #   </si>
    # </iq>
    $Thrasher::Component::COMPONENT->iq_query(
        [[$NS_COMPONENT, 'iq'], { 'to' => $ft->{'target_full_jid'},
                                  'from' => $ft->{'initiator_legacy_jid'},
                                  'type' => 'set',
                                },
         [[[$NS_STREAM_INITIATION, 'si'], { 'profile' => $NS_FILE_TRANSFER,
                                            'id' => $stream_id, },
           [[[$NS_FILE_TRANSFER, 'file'], { 'size' => $size,
                                            'name' => $filename, },
             [ [[$NS_FILE_TRANSFER, 'range'], {}, []] ]], # </file>
            [[$NS_FEATURE_NEG, 'feature'], {},
             [[[$NS_DATA, 'x'], { 'type' => 'form' }, [
                 [[$NS_DATA, 'field'], { 'type' => 'list-single',
                                   'var' => 'stream-method', },
                  [[[$NS_DATA, 'option'], {},
                    [[[$NS_DATA, 'value'], {},
                      [ $NS_BYTESTREAMS ],
                     []], # </value>
                   ]], # </option
                 ]], # </field>
              ]], # </x>
            ]], # </feature>
          ]], # </si>
        ]], # </iq>
        sub { _ft_recv_request2($ft, @_); });

    return;
}

# FT setup stage 2:
# * got error (including "Declined") => tell libpurple FT was refused.
# * bytestreams not available => tell libpurple FT was refused.
# * feature negotiation has bytestreams => send streamhost and wait
#   for streamhost-used.
sub _ft_recv_request2 {
    my ($ft, $component, $iq_params, $iq_packet) = @_;

    if ($iq_params->{'type'} eq 'error') {
        # including "Declined"
        THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
        return;
    }

    my $neg = recursive_extract(
        $iq_params->{query},
        [undef, undef, save_match('rec', [[$NS_FEATURE_NEG, 'feature']])],
        [undef, undef, save_match('rec', [[$NS_DATA, 'x']])],
        [undef, undef,
         save_sub('stream-method',
                  sub {
                      my ($tag) = @_;
                      # "If $tag is the stream-method field, extract
                      # the value's text element."
                      if (ref($tag)
                            && $tag->[0]->[0] eq $NS_DATA
                            && $tag->[0]->[1] eq 'field') {
                          while (my ($attr, $val) = each(%{$tag->[1]})) {
                              if ($attr =~ /}var$/ && $val eq 'stream-method') {
                                  for my $child (@{$tag->[2]}) {
                                      if (ref($child)
                                            && $child->[0]->[1] eq 'value') {
                                          return join('', @{$child->[2]});
                                      }
                                  }
                              }
                              return undef;
                          }
                      }
                  })],
    );
    $neg->{'stream-method'} ||= [];
    if (! grep(/^$NS_BYTESTREAMS/, @{$neg->{'stream-method'}})) {
        # If bytestreams aren't available, refuse to FT.
        THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
        return;
    }

    # Send streamhost and wait for streamhost-used
    #
    # <iq from='c%d@xmpp.transport' to='a@b/R' type='set'>
    #   <query xmlns="http://jabber.org/protocol/bytestreams" sid="s_id937" >
    #     <streamhost port="$proxy_server_port" host="$proxy_server_host" jid="$proxy_server_jid">
    #       <proxy xmlns="http://affinix.com/jabber/stream"/>
    #     </streamhost>
    #   </query>
    # </iq>
    $Thrasher::Component::COMPONENT->iq_query(
        [[$NS_COMPONENT, 'iq'], { 'to' => $ft->{'target_full_jid'},
                                  'from' => $ft->{'initiator_legacy_jid'},
                                  'type' => 'set',
                                },
         [[[$NS_BYTESTREAMS, 'query'], { 'sid' => $ft->{'sid'}, },
           [[[$NS_BYTESTREAMS, 'streamhost'], { 'port' => $proxy_server_port,
                                                'host' => $proxy_server_host,
                                                'jid' => $proxy_server_jid,
                                              },
             []], # </streamhost>
           ]], # </query>
        ]], # </iq>
        sub { _ft_recv_request3($ft, @_); });

    return;
}

# FT setup stage 3:
# * got streamhost-used => open SOCKS5 connection to streamhost.
# * otherwise, or if TCP connection fails => tell libpurple FT was refused.
sub _ft_recv_request3 {
    my ($ft, $component, $iq_params, $iq_packet) = @_;

    my $used = extract([undef, undef,
                        save_match([[$NS_BYTESTREAMS, 'streamhost-used'],
                                    { '{}jid' => save('jid'), },
                                    undef])],
                       $iq_params->{query});
    if (! $used->{'jid'}) {
        THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
    }

    # TODO: Currently can only use the one discovered proxy, which was
    # also the only streamhost _ft_recv_request2 offered. If any was
    # picked it better be that one! Check it just in case.
    if ($used->{'jid'} ne $proxy_server_jid) {
        debug("FT $ft->{id}: Wrong streamhost-used $used->{jid}");
        THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
    }

    log("FT $ft->{id}: TCP connecting.");
    $ft->{'socks'} = Thrasher::Plugin::Socks->new(
        Debug => 1,
        ProxyAddr => $proxy_server_host,
        ProxyPort => $proxy_server_port,
        ConnectAddr => sha1_hex(join('',
                                     $ft->{'sid'},
                                     $ft->{'initiator_legacy_jid'},
                                     $ft->{'target_full_jid'})),
        ConnectPort => 0,
        EventLoop => $component->{'event_loop'},
        SuccessCallback => sub { _ft_recv_request4($ft); },
        FailureCallback => sub {
            debug("FT $ft->{id}: streamhost connection failed");
            THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
        },
    );
    if (defined($ft->{'socks'})) {
        log("FT $ft->{id}: SOCKS5 connecting.");
        $ft->{'socks'}->connect();
    }
    else {
        log("FT $ft->{id}: TCP connection to SOCKS5 proxy failed.");
        THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 0);
    }
    return;
}

# FT setup stage 4:
# SOCKS5 connection to streamhost succeeded. Activate streamhost.
# Upon result IQ, tell libpurple FT was accepted.
sub _ft_recv_request4 {
    my ($ft) = @_;

    log("FT $ft->{id}: Got write socket from SOCKS.");
    $ft->{'write_file'} = $ft->{'socks'}->{'socket'};

    # <iq type="set" to="$used->{'jid'}" >
    #   <query xmlns="http://jabber.org/protocol/bytestreams" sid="s_id937">
    #     <activate>$ft->{'target_full_jid'}</activate>
    #   </query>
    # </iq>
    $Thrasher::Component::COMPONENT->iq_query(
        [[$NS_COMPONENT, 'iq'], { 'to' => $proxy_server_jid,
                                  'from' => $ft->{'initiator_legacy_jid'},
                                  'type' => 'set',
                                },
         [[[$NS_BYTESTREAMS, 'query'], { 'sid' => $ft->{'sid'}, },
           [[[$NS_BYTESTREAMS, 'activate'], {},
             [ $ft->{'target_full_jid'} ]], # </activate>
           ]], # </query>
        ]], # </iq>
        sub {
            THPPW::thrasher_action_ft_recv_request_respond($ft->{'id'}, 1);
            # Don't add the FD watch or send ui_ready yet--must wait
            # for recv_start or write loop. If the public IM side
            # isn't ready to read, FT gets canceled before it starts.
        },
    );

    return;
}

sub _ft_recv_accept {
    debug("_ft_recv_accept called");
    return 1;
}

sub _ft_recv_start {
    debug("_ft_recv_start called");
    return 1;
}

sub _ft_recv_cancel {
    my ($id) = @_;
    debug("_ft_recv_cancel($id) called");

    my $ft = delete($file_transfer_data_by_id->{$id});
    # FIXME: End transfer correctly.
    if ($ft->{'socks'}) {
        $ft->{'socks'}->close();
    }

    return;
}

sub _ft_recv_complete {
    my ($id) = @_;
    debug("_ft_recv_complete($id) called");
    my $ft = $file_transfer_data_by_id->{$id};

    # Ensure any buffered data drains into SOCKS eventually.
    if ($ft->{'leftover_data'}) {
        # Done with the FD watch that interacts with the protocol.
        $ft->{'socks'}->remove_fd_watch();
        # Write until leftovers are gone.
        $ft->add_fd_watch(sub {
                              my $written = $ft->write('');
                              if (! $ft->{'leftover_data'} || $written < 0) {
                                  # Completely written or a permanent error.
                                  _ft_recv_complete($id);
                                  return 0;
                              }
                              else {
                                  return 1;
                              }
                          });
    }

    # else done with this side of the FT, too.
    else {
        delete($file_transfer_data_by_id->{$id});
        $ft->{'socks'}->close();
    }
    return;
}

sub _ft_send_accept {
    debug("_ft_send_accept called");
    return 1;
}

sub _ft_send_start {
    my ($id) = @_;
    debug("_ft_send_start($id) called");

    my $ft = $file_transfer_data_by_id->{$id};
    $ft->add_fd_watch();

    return 1;
}

sub _ft_send_cancel {
    my ($id) = @_;
    debug("_ft_send_cancel($id) called");

    my $ft = delete($file_transfer_data_by_id->{$id});
    # FIXME: End transfer correctly.
    if ($ft->{'socks'}) {
        $ft->{'socks'}->close();
    }

    return;
}

sub _ft_send_complete {
    my ($id) = @_;
    debug("_ft_send_complete($id) called");

    my $ft = delete($file_transfer_data_by_id->{$id});
    # FIXME: End transfer correctly.
    $ft->{'socks'}->close();

    return;
}

sub _ft_read {
    my ($id, $size) = @_;
    debug("_ft_read($id, $size) called");

    my $ft = $file_transfer_data_by_id->{$id};
    if (! $ft) {
        debug("No such FT: $id");
        return ('', 0);
    }

    my $data = '';
    my $need_cancel = 0;

    # Take from leftover_data, if any.
    if ($ft->{'leftover_data'}) {
        $data = $ft->{'leftover_data'};
        $ft->{'leftover_data'} = '';
        if (bytes::length($data) > $size) {
            my $rest = substr($data, $size);
            $ft->{'leftover_data'} = $rest;
            $data = substr($data, 0, $size);
        }
        $size -= bytes::length($data);
    }

    # Read remainder of chunk
    if ($size && $ft->{'data_reader'}) {
        log("FT $ft->{id}: about to read.");

        my ($eof, $error, $chunk) = $ft->{'data_reader'}->($size);
        if (defined($chunk) && bytes::length($chunk) > 0) {
            log("FT $ft->{id}: read " . bytes::length($chunk) . ' bytes.');
            $data .= $chunk;
        }
        elsif ($eof) {
            log("FT $ft->{id}: Reading done.");
            $ft->{'data_reader'} = undef;
            $ft->socks_ended();
        }
        elsif ($error) {
            if ($!{'EAGAIN'}) {
                log("FT $ft->{id}: underflow; waiting for SOCKS data.");
            }
            else {
                log("FT $ft->{id}: Read error: $!");
                $ft->{'data_reader'} = undef;
                $need_cancel = 1;
            }
        }
    }
    elsif (! $ft->{'data_reader'}) {
        log("FT $ft->{id}: nowhere to read!");
        if ($ft->{'socks_ended'}) {
            $need_cancel = 1;
        }
    }

    if ($need_cancel) {
        THPPW::thrasher_action_ft_cancel_local($ft->{'id'});
        # If read() returns any data after canceling, do_transfer() in
        # libpurple/ft.c writes it anyway, potentially triggering
        # another cancel.
        #
        # An xfer must *never* be canceled twice! Some prpls
        # (currently at least jabber, msn, and yahoo) aren't
        # double-cancel safe (e.g. null pointer deference when
        # jabber_si_xfer_cancel_send() is called twice).
        #
        # This cancel should probably be moved out of read() entirely.
        return ('', 0);
    }
    else {
        return ($data, bytes::length($data));
    }
}

sub _ft_data_not_sent {
    my ($id, $rest) = @_;
    my $rest_sz = bytes::length($rest);
    debug("_ft_data_not_sent($id) called with $rest_sz bytes.");

    my $ft = $file_transfer_data_by_id->{$id};
    $ft->{'leftover_data'} .= $rest;

    return;
}

sub _ft_write {
    my ($id, $data) = @_;
    debug("_ft_write($id) called with " . bytes::length($data) . ' bytes.');

    my $ft = $file_transfer_data_by_id->{$id};
    if (! $ft) {
        debug("No such FT: $id");
        return 0;
    }

    my $written = $ft->write($data);
    if ($written == 0) {
        $ft->add_fd_watch();
    }
    elsif ($written < 0) {
        THPPW::thrasher_action_ft_cancel_local($id);
    }
    return $written;
}

1;

package Thrasher::Plugin::ProxyFileTransfer::FT;
use strict;
use warnings;

use List::Util qw(max);

use Thrasher::Log qw(:all);

# Data members:
# * "id": Thrasher's file id as returned by thrasher_wrapper_send_file.
# * "leftover_data": bytestring left by data_not_sent.
# * "socks": Thrasher::Plugin::Socks object connected to streamhost.
# * "socks_ended": Set to true after the SOCKS connection is finished
#   and closed but more events are expected. Distinguishes from the
#   state where the SOCKS reader/writer is not yet setup but events
#   are happening.
#
# Data members for sending files:
# * "data_reader": if defined, is a function that will
#    actually do the read. If undef, the stream is done.
#    $data_reader->($size) returns ($eof, $error, $chunk).
#
# Data members for receiving files:
# * "sid": Stream ID.
# * "target_full_jid": Recipient's JID.
# * "initiator_legacy_jid": legacy JID of sender.
# * "write_file": FD to write data to. If undef, the stream is done.

sub new {
    my $class = shift;

    my $self = {
        'data_reader' => undef,
        'write_file' => undef,
        'data' => [],
        @_,
    };
    bless($self, $class);
    return $self;
}

sub add_fd_watch {
    my ($self, $callback) = @_;

    if ($self->{'socks'}->{'watch_id'}) {
        # Only add one default FD watch at a time so remove_fd_watch() works.
        # debug("FT $self->{id}: FD watch already added.");
        return;
    }

    # When sending a file and the public IM side cancels, this watch
    # might fire and generate warnings before libpurple figures out
    # the cancel event.
    my $protocol = $Thrasher::Component::COMPONENT->{'protocol'};
    $callback ||= sub {
        # Logs a LOT.
        # log("FT $self->{id}: sending local_ready.");
        return $protocol->ft_local_ready($self->{'id'});
    };
    my $directions;
    if ($self->{'data_reader'}) {
        $directions = $Thrasher::EventLoop::IN;
    }
    elsif ($self->{'write_file'}) {
        $directions = $Thrasher::EventLoop::OUT;
    }
    else {
        log("FT $self->{id}: Neither a reader nor a writer be?!!");
        return;
    }
    # Thrasher must add its own watcher to poke libpurple's
    # do_transfer. Would be nice if the UI could say "I know I
    # underflowed this time, but go back to managing the watcher,
    # please" because if ui_read underflows even once transfer_cb
    # will be removed and there's no way to set it back.
    $self->{'socks'}->add_fd_watch($directions, $callback);
}

sub socks_ended {
    my ($self) = @_;

    $self->{'socks_ended'} = 1;
    # Don't close the SOCKS socket yet as that would take away its FD watch.
}

# $ft->write($data_bytes)
#
# Returns:
# * > 0 if that number of bytes were actually written successfully.
# * == 0 if a temporary condition resulting in no data being written this time.
# * < 0 if a permanent error occurred and this FT will never work data again.
sub write {
    my ($self, $data) = @_;
    my $write_sz = bytes::length($data);

    my $leftover_sz = bytes::length($self->{'leftover_data'});
    if ($leftover_sz) {
        $data = $self->{'leftover_data'} . $data;
        $self->{'leftover_data'} = '';
        $write_sz += $leftover_sz;
        debug("FT $self->{id}: Adding $leftover_sz leftover bytes to write.");
    }

    my $written = 0;
    if ($write_sz && $self->{'write_file'}) {
        # Must do own buffering here, anyway, so might as well syswrite().
        $written = syswrite($self->{'write_file'}, $data);
        if (! defined($written)) {
            $written = 0;
            if (! $!{'EAGAIN'}) {
                log("FT $self->{id}: Write error: $!");
                $self->{'leftover_data'} = undef;
                $self->{'write_file'} = undef;
                return -1;
            }
        }
    }
    elsif (! $self->{'write_file'}) {
        log("FT $self->{id}: nowhere to write!");
        if ($self->{'socks_ended'}) {
            return -1;
        }
    }

    if ($written < $write_sz) {
        my $rest = substr($data, $written);
        log("FT $self->{id}: "
              . bytes::length($rest) . ' bytes yet unwritten.');
        $self->{'leftover_data'} = $rest;
    }

    # Return the amount written from this request, not including leftovers used.
    my $ret = max($written - $leftover_sz,
                  0);
    return $ret;
}

1;