bidirectional tcp port forwarder

Hi

I 'borrowed' the tcp port forwarder script from the POE cookbook to use as 
a bidirectional port forwarder with a filter built into 1 direction. The 
data stream is actually SOAP, as is the response. The forward filter works 
perfectly, the problem is with the response. For some reason, the 
initiator does not recognize the response. Although the data stream is 
http, I wanted to treat it a strictly a data stream with out any parsing 
or decoding. The filter uses straight regexp to make the necessary 
changes. The response is suppose to be sent thru unchanged.

The script seems to work perfectly except, the originating client does't 
appear to accept the response. When I use a perl script based on 
LWP::Useragent and do a POST, the response is exactly what I would expect. 
What can be the difference?

Anyone see a problem?

My code (redirects blanked):

#!/usr/bin/perl -w
####################################################################################################
# Filename:     EdcEchoToEwocPoe.pl
# Author:       Tom Lines
# Description:  Filters the SOAP messages from ECHO to EWOC, Script taken 
from the Perl POE Cookbook
# Version:      1.00
# Date:         19 Jun 2009
#
# Ver   When          Who             What
# ----- ------------- --------------- 
--------------------------------------------------------------
# 1.0   19 Jun 2009   Tom Lines       Initial
# 1.1   24 Jun 2009   Tom Lines       Added SSL
#
####################################################################################################

use strict;

use Socket;
use POSIX qw(errno_h);
use FindBin;
use POE qw( Wheel::ReadWrite Wheel::SocketFactory Filter::Stream );
use Net::SSLeay;
use POE::Component::SSLify qw( Client_SSLify Server_SSLify SSLify_Options 
);

# The redirection table.
my %redirects = (
 
   );

# Main loop.  Create a new server for each record in %redirects.
# Run POE until all the servers (and their forwarders) shut down.
#
# The client is ECHO
# The server is EWOC

open( LOG, ">", "/usr/ecs/OPS/CUSTOM/logs/EdcEchoToEwoc.log" );
select( LOG );
$|++;

while ( my ( $from, $to ) = each %redirects ) {

   my ( $from_address, $from_port ) = split( /:/, $from );
   my ( $to_address,   $to_port )   = split( /:/, $to );

   &server_create( $from_address, $from_port, $to_address, $to_port );
}

$poe_kernel->run();

close( LOG );

exit;

#---------------------------------------------------------------------------------------------------
#---------------------------------------------------------------------------------------------------
# Create a session that will forward data between two sockets.
sub forwarder_create {

   my ( $handle, $peer_host, $peer_port, $remote_addr, $remote_port ) = 
@_;

   POE::Session->create(
      inline_states => {
                         _start         => \&forwarder_start,
                         _stop          => \&forwarder_stop,
                         client_input   => \&forwarder_client_input, # 
Client sent something.
                         client_error   => \&forwarder_client_error, # 
Error on client socket.
                         server_connect => \&forwarder_server_connect, # 
Connected to server.
                         server_input   => \&forwarder_server_input, # 
Server sent something.
                         server_error   => \&forwarder_server_error, # 
Error on server socket.
         },

      # Pass some things to forwarder_start():
      #         ARG0,    ARG1,       ARG2,       ARG3,         ARG4
      args => [ $handle, $peer_host, $peer_port, $remote_addr, 
$remote_port ]
      );
}

=for cookbook

The forwarder has been created.  This function sets up its initial state. 
Every Session instance
has its own HEAP.  This function stores information about the ports being 
redirected in its instance's heap.

It then logs the redirection to STDOUT and begins two wheels.  The first 
wheel, an instance of
POE::Wheel::ReadWrite, is used to interact with the client.  The second 
wheel is a 
POE::Wheel::SocketFactory instance.  It's used to connect to the server.

=cut

#---------------------------------------------------------------------------------------------------
sub forwarder_start {

   my ( $heap, $session, $socket, $peer_host, $peer_port, $remote_addr, 
$remote_port ) =
      @_[ HEAP, SESSION, ARG0, ARG1, ARG2, ARG3, ARG4 ];

   $heap ->{ log }         = $session->ID;
   $peer_host              = inet_ntoa( $peer_host );
   $heap ->{ peer_host }   = $peer_host;
   $heap ->{ peer_port }   = $peer_port;
   $heap ->{ remote_addr } = $remote_addr;
   $heap ->{ remote_port } = $remote_port;

   print "[$heap->{log}] Accepted connection from 
$peer_host:$peer_port\n";

   $heap ->{ state } = 'connecting';
   $heap ->{ queue } = [];

   $heap ->{ wheel_client } = POE::Wheel::ReadWrite->new( Handle     => 
$socket,
                                                          Driver     => 
POE::Driver::SysRW->new,
                                                          Filter     => 
POE::Filter::Stream->new,
                                                          InputEvent => 
'client_input',
                                                          ErrorEvent => 
'client_error',
                                                          );

   $heap ->{ wheel_server } = POE::Wheel::SocketFactory->new( 
RemoteAddress => $remote_addr,
                                                              RemotePort  
=> $remote_port,
                                                              SuccessEvent 
 => 'server_connect',
                                                              FailureEvent 
 => 'server_error',
                                                              );
}

#---------------------------------------------------------------------------------------------------
# The forwarder has stopped.  Log that it's done.
sub forwarder_stop {

   my $heap = $_[ HEAP ];
   print "[$heap->{log}] Closing redirection session\n";
}

#---------------------------------------------------------------------------------------------------
# The forwarder has received data from its client side.  Pass the data 
through to the server
# if it's connected.  Otherwise hold the data in a queue until the server 
connects.
sub forwarder_client_input {

   my ( $heap, $input ) = @_[ HEAP, ARG0 ];

   if ( $heap ->{ state } eq 'connecting' ) {

      push @{ $heap ->{ queue } }, $input;
   }
   else {

      # Filter the input from ECHO to replace FtpPull w/DVD
      $input = &Filter( $heap, $input );

      if ( ( exists $heap ->{ wheel_server } ) && $input ) {

         print "[$heap->{log}] To EWOC:\n$input\n";
         $heap ->{ wheel_server }->put( $input );
      }
   }
}

#---------------------------------------------------------------------------------------------------
# The forwarder has received an error from the client.  Shut down both 
sides of the connection.
# Log the error in a manner appropriate to its type.
sub forwarder_client_error {

   my ( $kernel, $heap, $operation, $errnum, $errstr ) = @_[ KERNEL, HEAP, 
ARG0, ARG1, ARG2 ];

   if ( $errnum ) {

      print( "[$heap->{log}] Client connection encountered ", "$operation 
error $errnum: $errstr\n" );
   }
   else {

      print "[$heap->{log}] Client closed connection.\n";
   }
   delete $heap ->{ wheel_client };
   delete $heap ->{ wheel_server };
}

#---------------------------------------------------------------------------------------------------
# The forwarder's SocketFactory has successfully connected to the server. 
Log the success,
# and create a ReadWrite wheel to interact with the server socket.  If the 
client sent anything
# during the connection process, pass it through to the server now.
sub forwarder_server_connect {

   my ( $kernel, $session, $heap, $socket ) = @_[ KERNEL, SESSION, HEAP, 
ARG0 ];

   my ( $local_port, $local_addr ) = unpack_sockaddr_in( getsockname( 
$socket ) );
   $local_addr = inet_ntoa( $local_addr );
   print( "[$heap->{log}] Established forward from local ",
          "$local_addr:$local_port to remote ",
          $heap ->{ remote_addr },
          ':', $heap ->{ remote_port }, "\n" );

   # Replace the SocketFactory wheel with a ReadWrite wheel.
   $heap ->{ wheel_server } = POE::Wheel::ReadWrite->new( Handle     => 
$socket,
                                                          Driver     => 
POE::Driver::SysRW->new,
                                                          Filter     => 
POE::Filter::Stream->new,
                                                          InputEvent => 
'server_input',
                                                          ErrorEvent => 
'server_error',
                                                          );

   $heap ->{ state } = 'connected';
   foreach my $pending ( @{ $heap ->{ queue } } ) {

      $kernel->call( $session, 'client_input', $pending );
   }
   $heap ->{ queue } = [];
}

#---------------------------------------------------------------------------------------------------
# The forwarder has received data from its server side.  Pass that through 
to the client.
sub forwarder_server_input {

   my ( $heap, $input ) = @_[ HEAP, ARG0 ];

   print "[$heap->{log}] From EWOC:\n$input\n";
   exists( $heap ->{ wheel_client } ) and $heap ->{ wheel_client }->put( 
$input );
}

#---------------------------------------------------------------------------------------------------
# The forwarder has received an error from the server.  Shut down both 
sides of the connection.
# Log the error in a manner appropriate to its type.
sub forwarder_server_error {

   my ( $kernel, $heap, $operation, $errnum, $errstr ) = @_[ KERNEL, HEAP, 
ARG0, ARG1, ARG2 ];

   if ( $errnum ) {

      print( "[$heap->{log}] Server connection encountered ", "$operation 
error $errnum: $errstr\n" );
   }
   else {

      print "[$heap->{log}] Server closed connection.\n";
   }
   delete $heap ->{ wheel_client };
   delete $heap ->{ wheel_server };
}

#---------------------------------------------------------------------------------------------------
# This is a stream-based forwarder server. It listens on TCP ports, and it 
spawns new forwarders
# to redirect incoming connections.

# Create a session that acts as the forwarder server.
sub server_create {

   my ( $local_address, $local_port, $remote_address, $remote_port ) = @_;

   POE::Session->create(
      inline_states => {
                         _start         => \&server_start,
                         _stop          => \&server_stop,
                         accept_success => \&server_accept_success,
                         accept_failure => \&server_accept_failure,
         },

      # Pass this function's parameters to the server_start().
      #         ARG0,            ARG1,        ARG2,            ARG3
      args => [ $local_address, $local_port, $remote_address, $remote_port 
]
      );
}

#---------------------------------------------------------------------------------------------------
# Start the server.
# This records where the server should connect, and it creates the 
listening socket factory.
sub server_start {

   my ( $heap, $local_addr, $local_port, $remote_addr, $remote_port ) = 
@_[ HEAP, ARG0, ARG1, ARG2, ARG3 ];

   print "+ Redirecting $local_addr:$local_port to 
$remote_addr:$remote_port\n";

   $heap ->{ local_addr }  = $local_addr;
   $heap ->{ local_port }  = $local_port;
   $heap ->{ remote_addr } = $remote_addr;
   $heap ->{ remote_port } = $remote_port;

   $heap ->{ server_wheel } = POE::Wheel::SocketFactory->new(
                                                BindAddress  => 
$local_addr,         # bind to this address
                                                BindPort     => 
$local_port,         # and bind to this port
                                                Reuse        => 'yes',     
# reuse immediately
                                                SuccessEvent => 
'accept_success',    # generate this event on connection
                                                FailureEvent => 
'accept_failure',    # generate this event on error
                                                );

}

#---------------------------------------------------------------------------------------------------
# The server is stopping.  Log that fact.
sub server_stop {

   my $heap = $_[ HEAP ];
   print( "- Redirection from $heap->{local_addr}:$heap->{local_port} to 
",
          "$heap->{remote_addr}:$heap->{remote_port} has stopped.\n" );
}

#---------------------------------------------------------------------------------------------------
# The server has accepted a client connection.  Pass the details about it 
to the function that
# creates a new forwarder.  This is as unnecessary step.  The contents of 
forwarder_create()
# could have been placed directly into server_accept_success().
sub server_accept_success {

   my ( $heap, $socket, $peer_addr, $peer_port ) = @_[ HEAP, ARG0, ARG1, 
ARG2 ];
   &forwarder_create( $socket, $peer_addr, $peer_port, $heap ->{ 
remote_addr }, $heap ->{ remote_port } );
}

#---------------------------------------------------------------------------------------------------
# The server encountered an error.  Log the error.  If we've run out of 
file descriptors,
# we'll have to shut down the server.  A serious port redirector should 
just restart the server here.
sub server_accept_failure {

   my ( $heap, $operation, $errnum, $errstr ) = @_[ HEAP, ARG0, ARG1, ARG2 
];

   print( "! Redirection from $heap->{local_addr}:$heap->{local_port} to 
",
          "$heap->{remote_addr}:$heap->{remote_port} encountered 
$operation ",
          "error $errnum: $errstr\n"
          );

   delete $heap ->{ server_wheel } if $errnum == ENFILE or $errnum == 
EMFILE;
}

#---------------------------------------------------------------------------------------------------
sub Filter {

   my ( $heap, $input ) = @_;

   my $eol = "\x0D\x0A";

   # ECHO sends data in chunks so I have to re-assemble it into 1 data 
stream so I can filter it.
   if ( $input =~ /^POST\s+.*/ ) {

      ( $heap ->{ header }, $heap ->{ content } ) = split( /(?:$eol){2}/, 
$input, 2 );

      if ( $heap ->{ header } =~ /.*?Content-Length:\s+(\d+).*/ ) {

         $heap ->{ content_length } = $1;
      }
   }
   else {

      $heap ->{ content } .= $input;
   }

   # return if I don't have the whole content
   return "" unless ( defined( $heap ->{ content } ) && defined( $heap ->{ 
content_length } ) );
   return "" unless ( length( $heap ->{ content } ) == $heap ->{ 
content_length } );

   open( IN, "<", "$FindBin::Bin/AuthorizedMediaUsers" );
   my @mediaUsers = <IN>;
   close( IN );
   chomp @mediaUsers;
   my $targetIds = join( "|", @mediaUsers );

   if ( $heap ->{ content } =~ 
/.*?\<UserId\>(?:$targetIds)\<\/UserId\>.*/gs ) {

      $heap ->{ content } =~ s/FtpPull/DVD/gs;
      $heap ->{ content } =~ s/ftppull-format/dvd-format/gs;
      $heap ->{ content } =~ s/FILEFORMAT/RockRidge/gs;

      my $len = length( $heap ->{ content } );
      $heap ->{ header } =~ s/(Content-Length:\s*)\d+/$1$len/s;
      $heap ->{ header } =~ s/(User-Agent:\s*)\S+/$1ECHO-Filtered/s;
   }
   return sprintf( "%s%s%s%s", $heap ->{ header }, $eol, $eol, $heap ->{ 
content } );
}




Thomas G Lines
NASA LP DAAC Core Systems Software Lead
Stinger Ghaffarian Technologies (SGT, Inc.)
Contractor to the U.S. Geological Survey (USGS) 
Earth Resources Observation and Science (EROS) Center
47914 252nd Street
Sioux Falls, South Dakota 57198

Phone: 605-594-2602
Email: [email protected]

A truly great computer programmer is lazy, impatient and full of hubris. 
Laziness drives one to work very hard to avoid future work for a future 
self. Impatience has the same endgame. And hubris is required for the 
newest Promethean fire -- inventing computer languages.
- Larry Wall