File Coverage

blib/lib/Net/DNAT.pm

Criterion	Covered	Total	%
statement	21	232	9.0
branch	0	90	0.0
condition	0	35	0.0
subroutine	7	15	46.6
pod	5	6	83.3
total	33	378	8.7

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.15';
@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}\r\n".
        "Remote-Port: $self->{peerport}\r\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__