Net::DNAT - Psuedo Layer7 Packet Processer

  use Net::DNAT;

  run Net::DNAT <settings...>;

 Example: port => 80

 Example: user => "nobody"

 Example: group => "nobody"

 Example: pools => {
    www => "web.server.com",
    dev => "dev.server.com",
  }

 Example: pools => "web.server.com"

 Example: default_pool => www

 Example: host_switch_table => {
    "server.com" => "www",
    "test.com" => "dev",
  }

 Example: switch_filters => [
    qr%^Cookie:.*magic%im => "dev",
    sub { s/^(Host: )www\.%$1%im; 0; } => "dev",
  ]

 Example: connect_timeout => 10

 Example: check_for_dequeue => 30

  PerlModule Apache::DNAT
  PerlInitHandler Apache::DNAT

  Test suite example using server and client though Net::DNAT.
  Test suite example using client and pool of servers.
  Test suite example using Apache::DNAT.
  Support for HTTP/1.1 protocol conversion to 1.0 protocol and back again.
  Support for HTTP/1.1 KeepAlive timeout and KeepAliveRequests.
  Support for SSL conversion to plain text and back (IO::Multiplex).
  Support for html error pages for internal errors like Server outages.
  Support for error logs.
  Support for access logs.
  Support for CVS protocol.
  Support for FTP protocol.
  Support for OOB channel data correctly.
  Support for DNS protocol.

  More information on network layers:

  http://uwsg.iu.edu/usail/network/nfs/network_layers.html

  Copyright (C) 2002-2003,
  Rob Brown, bbb@cpan.org

  This package may be distributed under the same terms as Perl itself.

  All rights reserved.

 L<Apache::DNAT>,
 L<Net::Server>,
 L<IO::Multiplex>

package Net::DNAT;

use strict;
use Exporter;
use vars qw(@ISA $VERSION $listen_port);
use Net::Server::Multiplex 0.85;
use Net::Ping 2.29;
use IO::Socket;
use Carp ();

$VERSION = '0.13';
@ISA = qw(Net::Server::Multiplex);

$listen_port = getservbyname("http", "tcp");

# DEBUG warnings
$SIG{__WARN__} = sub {
  &Carp::cluck((scalar localtime).": [pid $$] WARNING\n : $_[0]");
};

# DEBUG dies
my $dying = 0;
$SIG{__DIE__} = sub {
  $dying++;
  if ($dying > 2) {
    # Safety to avoid recursive or infinite dies
    return exit(1);
  }
  print STDERR ((scalar localtime),": [pid $$] CRASHED\n : ",@_,"\n");
  if ($^S) {
    # Die within eval does not count.
    $dying--;
    # Just use regular die.
    return CORE::die(@_);
  }
  # Stack trace of who crashed.
  &Carp::confess(@_);
};


sub _resolve_it {
  my $string = shift;
  my @result = ();
  my $port = $listen_port;
  if ($string =~ s/:(\d+)//) {
    $port = $1;
  } elsif ($string =~ s/:(\w+)//) {
    $port = getservbyname($1, "tcp");
  }
  if ($string !~ /^\d+\.\d+\.\d+\.\d+$/) {
    my $j;
    ($j, $j, $j, $j, @result) = gethostbyname($string);
    die "Failed to resolve [$string] to an IP address\n"
      unless @result;
    map { $_ = join(".", unpack("C4", $_)); } @result;
  } else {
    @result = ($string);
  }
  map { $_ .= ":$port"; } @result;
  return @result;
}

sub post_configure_hook {
  my $self = shift;
  my $conf_hash = {
    @{ $self->{server}->{configure_args} }
  };
  my $old_pools_ref = $conf_hash->{pools} ||
    die "The 'pools' setting is missing!\n";
  unless (ref $old_pools_ref &&
          ref $old_pools_ref eq "HASH") {
    $old_pools_ref = { default => $old_pools_ref };
  }

  my $new_pools_ref = {};
  foreach my $poolname (keys %{ $old_pools_ref }) {
    # The first element is the cycle index
    my @list = (0);
    my $dest = $old_pools_ref->{$poolname};
    if (!ref $dest) {
      push(@list, _resolve_it($dest));
    } elsif (ref $dest eq "ARRAY") {
      foreach my $i (@{ $dest }) {
        push(@list, _resolve_it($i));
      }
    } else {
      die "Unimplemented type of pool destination [".(ref $dest)."]\n";
    }
    $new_pools_ref->{$poolname} = [ @list ];
  }
  $self->{orig_pools} = $self->{pools} = $new_pools_ref;

  my $old_switch_table_ref = $conf_hash->{host_switch_table} || {};
  my $new_switch_table_ref = {};
  foreach my $old_host (keys %{ $old_switch_table_ref }) {
    my $new_host = $old_host;
    if ($new_host =~ s/^([a-z0-9\-\.]*[a-z])\.?$/\L$1/i) {
      $new_switch_table_ref->{$new_host} = $old_switch_table_ref->{$old_host};
    } else {
      die "Invalid hostname [$old_host] in host_switch_table\n";
    }
  }
  $self->{host_switch_table} = $new_switch_table_ref;

  $self->{switch_filters} = $conf_hash->{switch_filters} || [];
  # Run a quick sanity check on each pool destination
  for (my $i = scalar $#{ $self->{switch_filters} };
       $i > 0; $i-=2) {
    if (!$self->{pools}->{$self->{switch_filters}->[$i]}) {
      die "No such 'switch_filters' pool [".($self->{switch_filters}->[$i])."]\n";
    }
  }

  $self->{default_pool} = $conf_hash->{default_pool} || undef;
  if (!defined $self->{default_pool}) {
    if (( scalar keys %{ $self->{pools} } ) == 1) {
      # Only one pool?  Guess that should be the default.
      ($self->{default_pool}) = keys %{ $self->{pools} };
    } else {
      die "The 'default_pool' setting must be specified with multiple pools!\n";
    }
  }
  if (!$self->{pools}->{$self->{default_pool}}) {
    die "The 'default_pool' [$self->{default_pool}] has not been defined!\n";
  }

  # Plenty of time to establish the tcp three-way handshake
  # for a connection to a destination node in a pool.
  $self->{connect_timeout} =
    defined $conf_hash->{connect_timeout} ?
      $conf_hash->{connect_timeout} : 3;

  if (exists $conf_hash->{check_for_dequeue}) {
    if (defined $conf_hash->{check_for_dequeue} &&
        $conf_hash->{check_for_dequeue} > 0) {
      $self->{server}->{check_for_dequeue} =
        $conf_hash->{check_for_dequeue};
    }
  } else {
    $self->{server}->{check_for_dequeue} = 60;
  }

  $self->check_pools if $self->{server}->{check_for_dequeue};
}

sub run_dequeue {
  my $self = shift;
  $self->check_pools;
}

sub check_pools {
  my $self = shift;
  my $new_pools = {};
  my $ping_cache = {};
  my $pinger = new Net::Ping "tcp", $self->{connect_timeout};
  $pinger->tcp_service_check(1);
  foreach my $pool (keys %{ $self->{orig_pools} }) {
    my $index = $self->{pools}->{$pool} ? $self->{pools}->{$pool}->[0] : 0;
    for(my $i = 1; $i < @{ $self->{orig_pools}->{$pool} }; $i++) {
      $self->log(4, "Checking pool [$pool] index [$i]...");
      my ($host, $port) = $self->{orig_pools}->{$pool}->[$i] =~ /^(.+):(\d+)$/;
      next unless($host && $port);

      my $alive;
      if(exists $ping_cache->{"$host:$port"}) {
        $alive = $ping_cache->{"$host:$port"};
        $self->log(4, "Cached  pool [$pool] index [$i] at [$host:$port] is [$alive]");
      } else {
        $self->log(4, "Testing pool [$pool] index [$i] at [$host:$port]...");
        $pinger->{port_num} = $port;
        $alive = $ping_cache->{"$host:$port"} = $pinger->ping($host);
        if (!$alive) {
          $self->log(1, "WARNING: [$host:$port] is down!");
        }
      }
      next unless($alive);
      if (!$new_pools->{$pool}) {
        $new_pools->{$pool} = [$index];
      }
      push @{$new_pools->{$pool}}, $self->{orig_pools}->{$pool}->[$i];
    }
  }
  $pinger->close;
  $self->{pools} = $new_pools;
}

sub mux_connection {
  my $self = shift;
  shift; # I do not need mux
  my $fh   = shift;
  $self->{net_server}->log(4, "Connection on fileno [".fileno($fh)."]");
  $self->{state} = "REQUEST";
  # Store tied file handle within object
  $self->{fh} = $fh;
  # Grab peer information before it's gone
  $self->{peeraddr} = $self->{net_server}->{server}->{peeraddr};
  $self->{peerport} = $self->{net_server}->{server}->{peerport};
}


sub mux_input {
  my $self = shift;
  my $mux  = shift;
  my $fh   = shift;
  my $data = shift;

  my $pool = undef; # Which pool to redirect to

  unless (defined $fh and defined fileno($fh)) {
    $self->{net_server}->log(4, "mux_input: WEIRD fh! Trashing (".length($$data)." bytes) input.  (This should never happen.)");
    $$data = "";
    return;
  }

  if ($self->{state} eq "REQUEST") {
    $self->{net_server}->log(4, "input on [REQUEST] ($$data)");
    # Ignore leading whitespace and blank lines
    while ($$data =~ s/^\s+//) {}
    if ($$data =~ s%^([^\r\n]*)\r?\n%%) {
      # First newline reached.
      my $request = $1;
      if ($request =~ m%
                    (\w+)\s+        # method
                    (/.*)\s+        # path
                    HTTP/(1\.[01])  # protocol
                    $%ix) {
        $self->{request_method}  = $1;  # GET or POST
        $self->{request_path}    = $2;  # URL path
        $self->{request_proto}   = $3;  # 1.0 or 1.1
        $self->{state} = "HEADERS";
      } else {
        $self->{state} = "CONTENT";
        $_ = $request;
        goto POOL_DETERMINED;
      }
    }
  }

  if ($self->{state} eq "HEADERS" && $$data) {
    $self->{net_server}->log(4, "input on [HEADERS] ($$data)");
    # Search for the "nothing" line
    if ($$data =~ s/^((.*\n)*)\r?\n//) {
      # Found! Jump to next state.
      $self->{request_headers_block} = $1;
      # Wipe some headers for cleaner protocol
      # conversion and for security reasons.
      $self->{request_headers_block} =~
        s%^(Connection|
                        Keep-Alive|
                        Remote-Addr|
                        Remote-Port|
                        ):.*\n
                            %%gmix;

      # Add headers for Apache::DNAT
      $self->{request_headers_block} .=
        "Remote-Addr: $self->{peeraddr}\n".
          "Remote-Port: $self->{peerport}\n";

      $self->{state} = "CONTENT";
      # Determine correct pool destination
      # based on the request $_
      $_ = "$self->{request_method} $self->{request_path} HTTP/1.0\r\n$self->{request_headers_block}";
      # Rectify host header for simplicity
      s/^Host:\s*([\w\-\.]*\w)\.?((:\d+)?)\r?\n/Host: \L$1$2\r\n/im;

      # First run through the switch_filters
      my @switch_filters = @{ $self->{net_server}->{switch_filters} };
      while (@switch_filters) {
        my ($ref, $then_pool) = splice(@switch_filters, 0, 2);
        if (my $how = ref $ref) {
          if ($how eq "CODE") {
            if (&$ref()) {
              $pool = $then_pool;
              last;
            }
          } elsif ($how eq "Regexp") {
            if ($_ =~ $ref) {
              $pool = $then_pool;
              last;
            }
          } else {
            die "Switch filter to [$then_pool] smells too weird!\n";
          }
        } else {
          die "Switch filter [$ref] is not a ref!\n";
        }
      }

      # Then run through the host_switch_table
      if (!defined($pool) && m%^Host: ([\w\-\.]+)%m) {
        my $request_host = $1;

        foreach my $host (keys %{ $self->{net_server}->{host_switch_table} }) {
          if ( $request_host eq $host ) {
            $pool = $self->{net_server}->{host_switch_table}->{$host};
            last;
          }
        }
      }

    POOL_DETERMINED:
      # Otherwise, just use the default
      if (!defined($pool)) {
        $pool = $self->{net_server}->{default_pool};
      }

      $self->{net_server}->log(4, "POOL DETERMINED: [$pool]");
      my $pool_ref = $self->{net_server}->{pools}->{$pool};
      if (!$pool_ref) {
        $self->{net_server}->log(4, "Pool [$pool] is down.");
        $mux->write($fh, "ERROR: Pool [$pool] is down.\n");
        $$data = "";
        $mux->shutdown($fh, 2);
        return;
      }

      # Increment cycle counter.
      # If it exceeds pool size
      if (++($pool_ref->[0]) > $#{ $pool_ref }) {
        # Start over with 1 again.
        $pool_ref->[0] = 1;
      }
      $self->{net_server}->log(4, "POOL CYCLE INDEX [$pool_ref->[0]]");
      my $peeraddr = $pool_ref->[$pool_ref->[0]];
      $self->{net_server}->log(4, "Connecting to destination [$peeraddr]");

      $@ = "";
      my $peersock = eval {
        local $SIG{__DIE__} = 'DEFAULT';
        local $SIG{ALRM} = sub { die "Timed out!\n"; };
        alarm ($self->{net_server}->{connect_timeout});
        new IO::Socket::INET $peeraddr or die "$!\n";
      };
      alarm(0); # Reset alarm
      $peersock = undef if $@;
      if ($peersock) {
        $self->{net_server}->log(4, "Connected successfully with fileno [".fileno($peersock)."]");
        $mux->add($peersock);
        my $proxy_object = bless {
          state => "CONTENT",
          fh => $peersock,
          proto => $self->{request_proto},
          complement_object => $self,
          net_server => $self->{net_server},
        }, (ref $self);
        $self->{net_server}->log(4, "Complement for socket on fileno [".fileno($fh)."] created on fileno [".fileno($peersock)."]");
        $self->{complement_object} = $proxy_object;
        $mux->set_callback_object($proxy_object, $peersock);
        $mux->write($peersock, "$_\r\n");
        #$_ = "$self->{request_method} $self->{request_path} HTTP/1.0\r\n$self->{request_headers_block}";
      } else {
        $self->{net_server}->log(4, "Could not connect to [$peeraddr]: $@");
        $mux->write($fh, "ERROR: Pool [$pool] Index [$pool_ref->[0]] (Peer $peeraddr) is down: $!\n");
        $$data = "";
        $mux->shutdown($fh, 2);
        $self->{net_server}->check_pools if $self->{net_server}->{server}->{check_for_dequeue};
      }
    }
  }

  if ($self->{state} eq "CONTENT" && $$data) {
    # Test to make sure complement is up
    if ($self->{complement_object} and $self->{complement_object}->{fh} and
        defined fileno($self->{complement_object}->{fh})) {
      $self->{net_server}->log(4, "input on [CONTENT] on fileno [".fileno($fh)."] (".length($$data)." bytes) to socket on fileno [".fileno($self->{complement_object}->{fh})."]");
      $mux->write($self->{complement_object}->{fh}, $$data);
    } else {
      $self->{net_server}->log(4, "mux_input: Complement CONTENT socket is gone! Trashing (".length($$data)." bytes) input.");
      # close() is a bit stronger than shutdown()
      $mux->kill_output($fh);
      $mux->close($fh);
    }
    # Consumed everything
    $$data = "";
  }

}

sub mux_eof {
  my $self = shift;
  my $mux  = shift;
  my $fh   = shift;
  my $data = shift;
  $self->{net_server}->log(4, "EOF received on fileno [".fileno($fh)."] ($$data)");

  # If it hasn't been consumed by now,
  # then too bad, wipe it anyways.
  $$data = "";
  if ($self->{complement_object}) {
    $self->{net_server}->log(4, "Shutting down complement on fileno [".fileno($self->{complement_object}->{fh})."]");
    # If this end was closed, then tell the
    # complement socket to close.
    $mux->shutdown($self->{complement_object}->{fh}, 2);
    # Make sure that when the complement
    # socket finishes via mux_eof, that
    # it doesn't waste its time trying
    # to shutdown my socket, because I'm
    # already finished.
    delete $self->{complement_object}->{complement_object};
  }
}


1;
__END__