AnyEvent-MP/MP/Transport.pm

=head1 NAME

AnyEvent::MP::Transport - actual transport protocol handler

=head1 SYNOPSIS

   use AnyEvent::MP::Transport;

=head1 DESCRIPTION

This implements the actual transport protocol for MP (it represents a
single link), most of which is considered an implementation detail.

See the "PROTOCOL" section below if you want to write another client for
this protocol.

=head1 FUNCTIONS/METHODS

=over 4

=cut

package AnyEvent::MP::Transport;

use common::sense;

use Scalar::Util ();
use List::Util ();
use MIME::Base64 ();
use Storable ();
use JSON::XS ();

use Digest::MD6 ();
use Digest::HMAC_MD6 ();

use AE ();
use AnyEvent::Socket ();
use AnyEvent::Handle 4.92 ();

use AnyEvent::MP::Config ();

our $PROTOCOL_VERSION = 0;

=item $listener = mp_listener $host, $port, <constructor-args>

Creates a listener on the given host/port using
C<AnyEvent::Socket::tcp_server>.

See C<new>, below, for constructor arguments.

Defaults for peerhost, peerport and fh are provided.

=cut

sub mp_server($$@) {
   my ($host, $port, @args) = @_;

   AnyEvent::Socket::tcp_server $host, $port, sub {
      my ($fh, $host, $port) = @_;

      my $tp = new AnyEvent::MP::Transport
         fh       => $fh,
         peerhost => $host,
         peerport => $port,
         @args,
      ;
      $tp->{keepalive} = $tp;
   }
}

=item $guard = mp_connect $host, $port, <constructor-args>, $cb->($transport)

=cut

sub mp_connect {
   my $release = pop;
   my ($host, $port, @args) = @_;

   my $state;

   $state = AnyEvent::Socket::tcp_connect $host, $port, sub {
      my ($fh, $nhost, $nport) = @_;

      return $release->() unless $fh;

      $state = new AnyEvent::MP::Transport
         fh       => $fh,
         peername => $host,
         peerhost => $nhost,
         peerport => $nport,
         release  => $release,
         @args,
      ;
   };

   \$state
}

=item new AnyEvent::MP::Transport

   # immediately starts negotiation
   my $transport = new AnyEvent::MP::Transport
      # mandatory
      fh       => $filehandle,
      local_id => $identifier,
      on_recv  => sub { receive-callback },
      on_error => sub { error-callback },

      # optional
      on_eof   => sub { clean-close-callback },
      on_connect => sub { successful-connect-callback },
      greeting => { key => value },

      # tls support
      tls_ctx  => AnyEvent::TLS,
      peername => $peername, # for verification
   ;

=cut

sub LATENCY() { 3 } # assumed max. network latency

our @FRAMINGS = qw(json storable); # the framing types we accept and send, in order of preference
our @AUTH_SND = qw(tls_md6_64_256 hmac_md6_64_256); # auth types we send
our @AUTH_RCV = (@AUTH_SND, qw(tls_anon cleartext)); # auth types we accept

#AnyEvent::Handle::register_write_type mp_record => sub {
#};

sub new {
   my ($class, %arg) = @_;

   my $self = bless \%arg, $class;

   $self->{queue} = [];

   {
      Scalar::Util::weaken (my $self = $self);

      my $config = AnyEvent::MP::Config::config;

      my $latency = $config->{network_latency} || LATENCY;

      $self->{secret} = $config->{secret}
         unless exists $self->{secret};

      $self->{timeout} = $config->{monitor_timeout} || $AnyEvent::MP::Kernel::MONITOR_TIMEOUT
         unless exists $self->{timeout};

      $self->{timeout} -= $latency;

      $self->{timeout} = 1 + $latency
         if $self->{timeout} < 1 + $latency;

      my $secret = $self->{secret};

      if (exists $config->{cert}) {
         $self->{tls_ctx} = {
            sslv2   => 0,
            sslv3   => 0,
            tlsv1   => 1,
            verify  => 1,
            cert    => $config->{cert},
            ca_cert => $config->{cert},
            verify_require_client_cert => 1,
         };
      }

      $self->{hdl} = new AnyEvent::Handle
         fh       => delete $self->{fh},
         autocork => 1,
         no_delay => 1,
         on_error => sub {
            $self->error ($_[2]);
         },
         rtimeout => $latency,
         peername => delete $self->{peername},
      ;

      my $greeting_kv = $self->{greeting} ||= {};

      $self->{local_node} ||= $AnyEvent::MP::Kernel::NODE;

      $greeting_kv->{tls}      = "1.0" if $self->{tls_ctx};
      $greeting_kv->{provider} = "AE-$AnyEvent::MP::Kernel::VERSION";
      $greeting_kv->{peeraddr} = AnyEvent::Socket::format_hostport $self->{peerhost}, $self->{peerport};
      $greeting_kv->{timeout}  = $self->{timeout};

      # send greeting
      my $lgreeting1 = "aemp;$PROTOCOL_VERSION"
                     . ";$self->{local_node}"
                     . ";" . (join ",", @AUTH_RCV)
                     . ";" . (join ",", @FRAMINGS)
                     . (join "", map ";$_=$greeting_kv->{$_}", keys %$greeting_kv);

      my $lgreeting2 = MIME::Base64::encode_base64 AnyEvent::MP::Kernel::nonce (66), "";

      $self->{hdl}->push_write ("$lgreeting1\012$lgreeting2\012");

      # expect greeting
      $self->{hdl}->rbuf_max (4 * 1024);
      $self->{hdl}->push_read (line => sub {
         my $rgreeting1 = $_[1];

         my ($aemp, $version, $rnode, $auths, $framings, @kv) = split /;/, $rgreeting1;

         if ($aemp ne "aemp") {
            return $self->error ("unparsable greeting");
         } elsif ($version != $PROTOCOL_VERSION) {
            return $self->error ("version mismatch (we: $PROTOCOL_VERSION, they: $version)");
         } elsif ($rnode eq $self->{local_node}) {
            AnyEvent::MP::Global::avoid_seed ($self->{seed})
               if exists $self->{seed};

            return $self->error ("I refuse to talk to myself");
         } elsif ($AnyEvent::MP::Kernel::NODE{$rnode} && $AnyEvent::MP::Kernel::NODE{$rnode}{transport}) {
            return $self->error ("$rnode already connected, not connecting again.");
         }

         $self->{remote_node} = $rnode;

         $self->{remote_greeting} = {
            map /^([^=]+)(?:=(.*))?/ ? ($1 => $2) : (),
               @kv
         };

         # read nonce
         $self->{hdl}->push_read (line => sub {
            my $rgreeting2 = $_[1];

            "$lgreeting1\012$lgreeting2" ne "$rgreeting1\012$rgreeting2" # echo attack?
               or return $self->error ("authentication error, echo attack?");

            my $tls = $self->{tls_ctx} && 1 == int $self->{remote_greeting}{tls};

            my $s_auth;
            for my $auth_ (split /,/, $auths) {
               if (grep $auth_ eq $_, @AUTH_SND and ($auth_ !~ /^tls_/ or $tls)) {
                  $s_auth = $auth_;
                  last;
               }
            }

            defined $s_auth
               or return $self->error ("$auths: no common auth type supported");

            my $s_framing;
            for my $framing_ (split /,/, $framings) {
               if (grep $framing_ eq $_, @FRAMINGS) {
                  $s_framing = $framing_;
                  last;
               }
            }

            defined $s_framing
               or return $self->error ("$framings: no common framing method supported");

            my $key;
            my $lauth;

            if ($tls) {
               $self->{tls} = $lgreeting2 lt $rgreeting2 ? "connect" : "accept";
               $self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});

               $lauth =
                  $s_auth eq "tls_anon"       ? ""
                : $s_auth eq "tls_md6_64_256" ? Digest::MD6::md6_hex "$lgreeting1\012$lgreeting2\012$rgreeting1\012$rgreeting2\012"
                : return $self->error ("$s_auth: fatal, selected unsupported snd auth method");

            } elsif (length $secret) {
               return $self->error ("$s_auth: fatal, selected unsupported snd auth method")
                  unless $s_auth eq "hmac_md6_64_256"; # hardcoded atm.

               $key = Digest::MD6::md6 $secret;
               # we currently only support hmac_md6_64_256
               $lauth = Digest::HMAC_MD6::hmac_md6_hex $key, "$lgreeting1\012$lgreeting2\012$rgreeting1\012$rgreeting2\012", 64, 256;

            } else {
               return $self->error ("unable to handshake TLS and no shared secret configured");
            }

            $self->{hdl}->push_write ("$s_auth;$lauth;$s_framing\012");

            # read the authentication response
            $self->{hdl}->push_read (line => sub {
               my ($hdl, $rline) = @_;

               my ($auth_method, $rauth2, $r_framing) = split /;/, $rline;

               my $rauth =
                  $auth_method eq "hmac_md6_64_256" ? Digest::HMAC_MD6::hmac_md6_hex $key, "$rgreeting1\012$rgreeting2\012$lgreeting1\012$lgreeting2\012", 64, 256
                : $auth_method eq "cleartext"       ? unpack "H*", $secret
                : $auth_method eq "tls_anon"        ? ($tls ? "" : "\012\012") # \012\012 never matches
                : $auth_method eq "tls_md6_64_256"  ? ($tls ? Digest::MD6::md6_hex "$rgreeting1\012$rgreeting2\012$lgreeting1\012$lgreeting2\012" : "\012\012")
                : return $self->error ("$auth_method: fatal, selected unsupported rcv auth method");

               if ($rauth2 ne $rauth) {
                  return $self->error ("authentication failure/shared secret mismatch");
               }

               $self->{s_framing} = $s_framing;

               $hdl->rbuf_max (undef);
               my $queue = delete $self->{queue}; # we are connected

               $self->{hdl}->rtimeout ($self->{remote_greeting}{timeout});
               $self->{hdl}->wtimeout ($self->{timeout} - LATENCY);
               $self->{hdl}->on_wtimeout (sub { $self->send ([]) });

               $self->connected;

               # send queued messages
               $self->send ($_)
                  for @$queue;

               # receive handling
               my $src_node = $self->{node};

               my $rmsg; $rmsg = sub {
                  $_[0]->push_read ($r_framing => $rmsg);

                  local $AnyEvent::MP::Kernel::SRCNODE = $src_node;
                  AnyEvent::MP::Kernel::_inject (@{ $_[1] });
               };
               $hdl->push_read ($r_framing => $rmsg);
            });
         });
      });
   }

   $self
}

sub error {
   my ($self, $msg) = @_;

   delete $self->{keepalive};

#   $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} $msg");#d#

   $self->{node}->transport_error (transport_error => $self->{node}{id}, $msg)
      if $self->{node} && $self->{node}{transport} == $self;

   (delete $self->{release})->()
      if exists $self->{release};
   
#   $AnyEvent::MP::Kernel::WARN->(7, "$self->{peerhost}:$self->{peerport}: $msg");
   $self->destroy;
}

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

   delete $self->{keepalive};

   (delete $self->{release})->()
      if exists $self->{release};

   $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} connected as $self->{remote_node}");

   my $node = AnyEvent::MP::Kernel::add_node ($self->{remote_node});
   Scalar::Util::weaken ($self->{node} = $node);
   $node->transport_connect ($self);
}

sub send {
   $_[0]{hdl}->push_write ($_[0]{s_framing} => $_[1]);
}

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

   (delete $self->{release})->()
      if exists $self->{release};

   $self->{hdl}->destroy
      if $self->{hdl};
}

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

   $self->destroy;
}

=back

=head1 PROTOCOL

The protocol is relatively simple, and consists of three phases which are
symmetrical for both sides: greeting (followed by optionally switching to
TLS mode), authentication and packet exchange.

the protocol is designed to allow both full-text and binary streams.

The greeting consists of two text lines that are ended by either an ASCII
CR LF pair, or a single ASCII LF (recommended).

=head2 GREETING

All the lines until after authentication must not exceed 4kb in length,
including delimiter. Afterwards there is no limit on the packet size that
can be received.

=head3 First Greeting Line

Example:

   aemp;0;fec.4a7720fc;127.0.0.1:1235,[::1]:1235;hmac_md6_64_256;json,storable;provider=AE-0.0

The first line contains strings separated (not ended) by C<;>
characters. The first even ixtrings are fixed by the protocol, the
remaining strings are C<KEY=VALUE> pairs. None of them may contain C<;>
characters themselves.

The fixed strings are:

=over 4

=item protocol identification

The constant C<aemp> to identify the protocol.

=item protocol version

The protocol version supported by this end, currently C<0>. If the
versions don't match then no communication is possible. Minor extensions
are supposed to be handled through additional key-value pairs.

=item the node id

This is the node ID of the connecting node.

=item the acceptable authentication methods

A comma-separated list of authentication methods supported by the
node. Note that AnyEvent::MP supports a C<hex_secret> authentication
method that accepts a cleartext password (hex-encoded), but will not use
this auth method itself.

The receiving side should choose the first auth method it supports.

=item the acceptable framing formats

A comma-separated list of packet encoding/framign formats understood. The
receiving side should choose the first framing format it supports for
sending packets (which might be different from the format it has to accept).

=back

The remaining arguments are C<KEY=VALUE> pairs. The following key-value
pairs are known at this time:

=over 4

=item provider=<module-version>

The software provider for this implementation. For AnyEvent::MP, this is
C<AE-0.0> or whatever version it currently is at.

=item peeraddr=<host>:<port>

The peer address (socket address of the other side) as seen locally.

=item tls=<major>.<minor>

Indicates that the other side supports TLS (version should be 1.0) and
wishes to do a TLS handshake.

=item timeout=<seconds>

The amount of time after which this node should be detected as dead unless
some data has been received. The node is responsible to send traffic
reasonably more often than this interval (such as every timeout minus five
seconds).

=back

=head3 Second Greeting Line

After this greeting line there will be a second line containing a
cryptographic nonce, i.e. random data of high quality. To keep the
protocol text-only, these are usually 32 base64-encoded octets, but
it could be anything that doesn't contain any ASCII CR or ASCII LF
characters.

I<< The two nonces B<must> be different, and an aemp implementation
B<must> check and fail when they are identical >>.

Example of a nonce line:

   p/I122ql7kJR8lumW3lXlXCeBnyDAvz8NQo3x5IFowE4

=head2 TLS handshake

I<< If, after the handshake, both sides indicate interest in TLS, then the
connection B<must> use TLS, or fail. >>

Both sides compare their nonces, and the side who sent the lower nonce
value ("string" comparison on the raw octet values) becomes the client,
and the one with the higher nonce the server.

=head2 AUTHENTICATION PHASE

After the greeting is received (and the optional TLS handshake),
the authentication phase begins, which consists of sending a single
C<;>-separated line with three fixed strings and any number of
C<KEY=VALUE> pairs.

The three fixed strings are:

=over 4

=item the authentication method chosen

This must be one of the methods offered by the other side in the greeting.

Note that all methods starting with C<tls_> are only valid I<iff> TLS was
successfully handshaked (and to be secure the implementation must enforce
this).

The currently supported authentication methods are:

=over 4

=item cleartext

This is simply the shared secret, lowercase-hex-encoded. This method is of
course very insecure, unless TLS is used, which is why this module will
accept, but not generate, cleartext auth replies.

=item hmac_md6_64_256

This method uses an MD6 HMAC with 64 bit blocksize and 256 bit hash. First, the shared secret
is hashed with MD6:

   key = MD6 (secret)

This secret is then used to generate the "local auth reply", by taking
the two local greeting lines and the two remote greeting lines (without
line endings), appending \012 to all of them, concatenating them and
calculating the MD6 HMAC with the key.

   lauth = HMAC_MD6 key, "lgreeting1\012lgreeting2\012rgreeting1\012rgreeting2\012"

This authentication token is then lowercase-hex-encoded and sent to the
other side.

Then the remote auth reply is generated using the same method, but local
and remote greeting lines swapped:

   rauth = HMAC_MD6 key, "rgreeting1\012rgreeting2\012lgreeting1\012lgreeting2\012"

This is the token that is expected from the other side.

=item tls_anon

This type is only valid iff TLS was enabled and the TLS handshake
was successful. It has no authentication data, as the server/client
certificate was successfully verified.

This authentication type is slightly less secure than the others, as it
allows a man-in-the-middle attacker to change some of the connection
parameters (such as the framing format), although there is no known attack
that exploits this in a way that is worse than just denying the service.

By default, this implementation accepts but uses this authentication
method.

=item tls_md6_64_256

This type is only valid iff TLS was enabled and the TLS handshake
was successful.

This authentication type simply calculates:

   lauth = MD6 "rgreeting1\012rgreeting2\012lgreeting1\012lgreeting2\012"

and lowercase-hex encodes the result and sends it as authentication
data. No shared secret is required (authentication is done by TLS). The
checksum solely exists to make tinkering with the greeting hard.

=back

=item the authentication data

The authentication data itself, usually base64 or hex-encoded data, see
above.

=item the framing protocol chosen

This must be one of the framing protocols offered by the other side in the
greeting. Each side must accept the choice of the other side.

=back

Example of an authentication reply:

   hmac_md6_64_256;363d5175df38bd9eaddd3f6ca18aa1c0c4aa22f0da245ac638d048398c26b8d3;json

=head2 DATA PHASE

After this, packets get exchanged using the chosen framing protocol. It is
quite possible that both sides use a different framing protocol.

=head2 FULL EXAMPLE

This is an actual protocol dump of a handshake, followed by a single data
packet. The greater than/less than lines indicate the direction of the
transfer only.

   > aemp;0;nndKd+gn;10.0.0.1:4040;hmac_md6_64_256,cleartext;json,storable;provider=AE-0.0;peeraddr=127.0.0.1:1235
   > sRG8bbc4TDbkpvH8FTP4HBs87OhepH6VuApoZqXXskuG
   < aemp;0;nmpKd+gh;127.0.0.1:1235,[::1]:1235;hmac_md6_64_256,cleartext;json,storable;provider=AE-0.0;peeraddr=127.0.0.1:58760
   < dCEUcL/LJVSTJcx8byEsOzrwhzJYOq+L3YcopA5T6EAo
   > hmac_md6_64_256;9513d4b258975accfcb2ab7532b83690e9c119a502c612203332a591c7237788;json
   < hmac_md6_64_256;0298d6ba2240faabb2b2e881cf86b97d70a113ca74a87dc006f9f1e9d3010f90;json
   > ["","lookup","pinger","10.0.0.1:4040#nndKd+gn.a","resolved"]

=head1 SEE ALSO

L<AnyEvent::MP>.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1

Revision:	1.42
Committed:	Fri Aug 28 20:57:42 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
Changes since 1.41:	+20 -14 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3	root	1.13	AnyEvent::MP::Transport - actual transport protocol handler
4	root	1.1
5			=head1 SYNOPSIS
6
7			use AnyEvent::MP::Transport;
8
9			=head1 DESCRIPTION
10
11	root	1.13	This implements the actual transport protocol for MP (it represents a
12			single link), most of which is considered an implementation detail.
13	root	1.1
14	root	1.7	See the "PROTOCOL" section below if you want to write another client for
15			this protocol.
16	root	1.1
17			=head1 FUNCTIONS/METHODS
18
19			=over 4
20
21			=cut
22
23			package AnyEvent::MP::Transport;
24
25			use common::sense;
26
27	root	1.27	use Scalar::Util ();
28			use List::Util ();
29	root	1.1	use MIME::Base64 ();
30			use Storable ();
31	root	1.2	use JSON::XS ();
32	root	1.1
33	root	1.19	use Digest::MD6 ();
34			use Digest::HMAC_MD6 ();
35
36	root	1.1	use AE ();
37			use AnyEvent::Socket ();
38	root	1.27	use AnyEvent::Handle 4.92 ();
39	root	1.2
40	root	1.30	use AnyEvent::MP::Config ();
41
42	root	1.2	our $PROTOCOL_VERSION = 0;
43	root	1.1
44	root	1.39	=item $listener = mp_listener $host, $port, <constructor-args>
45	root	1.1
46			Creates a listener on the given host/port using
47			C<AnyEvent::Socket::tcp_server>.
48
49			See C<new>, below, for constructor arguments.
50
51	root	1.10	Defaults for peerhost, peerport and fh are provided.
52	root	1.1
53			=cut
54
55			sub mp_server($$@) {
56			my ($host, $port, @args) = @_;
57
58			AnyEvent::Socket::tcp_server $host, $port, sub {
59			my ($fh, $host, $port) = @_;
60
61	root	1.39	my $tp = new AnyEvent::MP::Transport
62	root	1.1	fh => $fh,
63			peerhost => $host,
64			peerport => $port,
65			@args,
66	root	1.39	;
67			$tp->{keepalive} = $tp;
68	root	1.1	}
69			}
70
71	root	1.2	=item $guard = mp_connect $host, $port, <constructor-args>, $cb->($transport)
72
73			=cut
74
75			sub mp_connect {
76	root	1.31	my $release = pop;
77	root	1.2	my ($host, $port, @args) = @_;
78
79	root	1.31	my $state;
80
81			$state = AnyEvent::Socket::tcp_connect $host, $port, sub {
82	root	1.2	my ($fh, $nhost, $nport) = @_;
83
84	root	1.31	return $release->() unless $fh;
85	root	1.2
86	root	1.31	$state = new AnyEvent::MP::Transport
87	root	1.2	fh => $fh,
88			peername => $host,
89			peerhost => $nhost,
90			peerport => $nport,
91	root	1.31	release => $release,
92	root	1.2	@args,
93	root	1.31	;
94			};
95
96			\$state
97	root	1.2	}
98
99	root	1.1	=item new AnyEvent::MP::Transport
100
101			# immediately starts negotiation
102			my $transport = new AnyEvent::MP::Transport
103	root	1.2	# mandatory
104	root	1.1	fh => $filehandle,
105	root	1.2	local_id => $identifier,
106	root	1.1	on_recv => sub { receive-callback },
107			on_error => sub { error-callback },
108
109			# optional
110			on_eof => sub { clean-close-callback },
111			on_connect => sub { successful-connect-callback },
112	root	1.2	greeting => { key => value },
113	root	1.1
114			# tls support
115			tls_ctx => AnyEvent::TLS,
116			peername => $peername, # for verification
117			;
118
119			=cut
120
121	root	1.27	sub LATENCY() { 3 } # assumed max. network latency
122
123	root	1.34	our @FRAMINGS = qw(json storable); # the framing types we accept and send, in order of preference
124	root	1.41	our @AUTH_SND = qw(tls_md6_64_256 hmac_md6_64_256); # auth types we send
125			our @AUTH_RCV = (@AUTH_SND, qw(tls_anon cleartext)); # auth types we accept
126	root	1.7
127			#AnyEvent::Handle::register_write_type mp_record => sub {
128			#};
129	root	1.4
130	root	1.1	sub new {
131			my ($class, %arg) = @_;
132
133			my $self = bless \%arg, $class;
134
135			$self->{queue} = [];
136
137			{
138			Scalar::Util::weaken (my $self = $self);
139
140	root	1.33	my $config = AnyEvent::MP::Config::config;
141	root	1.30
142	root	1.31	my $latency = $config->{network_latency} \|\| LATENCY;
143
144	root	1.42	$self->{secret} = $config->{secret}
145			unless exists $self->{secret};
146	root	1.2
147	root	1.42	$self->{timeout} = $config->{monitor_timeout} \|\| $AnyEvent::MP::Kernel::MONITOR_TIMEOUT
148			unless exists $self->{timeout};
149	root	1.24
150	root	1.42	$self->{timeout} -= $latency;
151	root	1.31
152	root	1.42	$self->{timeout} = 1 + $latency
153			if $self->{timeout} < 1 + $latency;
154	root	1.27
155	root	1.42	my $secret = $self->{secret};
156	root	1.19
157	root	1.30	if (exists $config->{cert}) {
158	root	1.42	$self->{tls_ctx} = {
159	root	1.19	sslv2 => 0,
160			sslv3 => 0,
161			tlsv1 => 1,
162			verify => 1,
163	root	1.30	cert => $config->{cert},
164			ca_cert => $config->{cert},
165	root	1.19	verify_require_client_cert => 1,
166			};
167			}
168
169	root	1.1	$self->{hdl} = new AnyEvent::Handle
170	root	1.42	fh => delete $self->{fh},
171	root	1.4	autocork => 1,
172			no_delay => 1,
173	root	1.1	on_error => sub {
174			$self->error ($_[2]);
175			},
176	root	1.31	rtimeout => $latency,
177	root	1.42	peername => delete $self->{peername},
178	root	1.1	;
179
180	root	1.2	my $greeting_kv = $self->{greeting} \|\|= {};
181	root	1.24
182	root	1.38	$self->{local_node} \|\|= $AnyEvent::MP::Kernel::NODE;
183	root	1.24
184	root	1.42	$greeting_kv->{tls} = "1.0" if $self->{tls_ctx};
185	root	1.31	$greeting_kv->{provider} = "AE-$AnyEvent::MP::Kernel::VERSION";
186	root	1.7	$greeting_kv->{peeraddr} = AnyEvent::Socket::format_hostport $self->{peerhost}, $self->{peerport};
187	root	1.42	$greeting_kv->{timeout} = $self->{timeout};
188	root	1.23
189	root	1.1	# send greeting
190	root	1.12	my $lgreeting1 = "aemp;$PROTOCOL_VERSION"
191	root	1.24	. ";$self->{local_node}"
192	root	1.7	. ";" . (join ",", @AUTH_RCV)
193			. ";" . (join ",", @FRAMINGS)
194			. (join "", map ";$_=$greeting_kv->{$_}", keys %$greeting_kv);
195	root	1.12
196	root	1.31	my $lgreeting2 = MIME::Base64::encode_base64 AnyEvent::MP::Kernel::nonce (66), "";
197	root	1.1
198	root	1.7	$self->{hdl}->push_write ("$lgreeting1\012$lgreeting2\012");
199	root	1.1
200			# expect greeting
201	root	1.12	$self->{hdl}->rbuf_max (4 * 1024);
202	root	1.1	$self->{hdl}->push_read (line => sub {
203	root	1.7	my $rgreeting1 = $_[1];
204	root	1.1
205	root	1.26	my ($aemp, $version, $rnode, $auths, $framings, @kv) = split /;/, $rgreeting1;
206	root	1.1
207			if ($aemp ne "aemp") {
208			return $self->error ("unparsable greeting");
209	root	1.12	} elsif ($version != $PROTOCOL_VERSION) {
210			return $self->error ("version mismatch (we: $PROTOCOL_VERSION, they: $version)");
211	root	1.39	} elsif ($rnode eq $self->{local_node}) {
212	root	1.42	AnyEvent::MP::Global::avoid_seed ($self->{seed})
213			if exists $self->{seed};
214
215	root	1.39	return $self->error ("I refuse to talk to myself");
216			} elsif ($AnyEvent::MP::Kernel::NODE{$rnode} && $AnyEvent::MP::Kernel::NODE{$rnode}{transport}) {
217			return $self->error ("$rnode already connected, not connecting again.");
218	root	1.1	}
219
220	root	1.2	$self->{remote_node} = $rnode;
221	root	1.1
222	root	1.2	$self->{remote_greeting} = {
223			map /^([^=]+)(?:=(.*))?/ ? ($1 => $2) : (),
224			@kv
225	root	1.1	};
226
227	root	1.7	# read nonce
228			$self->{hdl}->push_read (line => sub {
229			my $rgreeting2 = $_[1];
230
231	root	1.19	"$lgreeting1\012$lgreeting2" ne "$rgreeting1\012$rgreeting2" # echo attack?
232			or return $self->error ("authentication error, echo attack?");
233
234	root	1.41	my $tls = $self->{tls_ctx} && 1 == int $self->{remote_greeting}{tls};
235
236			my $s_auth;
237			for my $auth_ (split /,/, $auths) {
238			if (grep $auth_ eq $_, @AUTH_SND and ($auth_ !~ /^tls_/ or $tls)) {
239			$s_auth = $auth_;
240			last;
241			}
242			}
243
244			defined $s_auth
245			or return $self->error ("$auths: no common auth type supported");
246
247			my $s_framing;
248			for my $framing_ (split /,/, $framings) {
249			if (grep $framing_ eq $_, @FRAMINGS) {
250			$s_framing = $framing_;
251			last;
252			}
253			}
254
255			defined $s_framing
256			or return $self->error ("$framings: no common framing method supported");
257
258	root	1.30	my $key;
259	root	1.19	my $lauth;
260
261	root	1.41	if ($tls) {
262	root	1.8	$self->{tls} = $lgreeting2 lt $rgreeting2 ? "connect" : "accept";
263			$self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});
264	root	1.41
265			$lauth =
266			$s_auth eq "tls_anon" ? ""
267			: $s_auth eq "tls_md6_64_256" ? Digest::MD6::md6_hex "$lgreeting1\012$lgreeting2\012$rgreeting1\012$rgreeting2\012"
268			: return $self->error ("$s_auth: fatal, selected unsupported snd auth method");
269
270	root	1.30	} elsif (length $secret) {
271	root	1.41	return $self->error ("$s_auth: fatal, selected unsupported snd auth method")
272			unless $s_auth eq "hmac_md6_64_256"; # hardcoded atm.
273
274	root	1.30	$key = Digest::MD6::md6 $secret;
275	root	1.19	# we currently only support hmac_md6_64_256
276			$lauth = Digest::HMAC_MD6::hmac_md6_hex $key, "$lgreeting1\012$lgreeting2\012$rgreeting1\012$rgreeting2\012", 64, 256;
277	root	1.41
278	root	1.30	} else {
279			return $self->error ("unable to handshake TLS and no shared secret configured");
280	root	1.8	}
281	root	1.2
282	root	1.7	$self->{hdl}->push_write ("$s_auth;$lauth;$s_framing\012");
283	root	1.2
284	root	1.19	# read the authentication response
285	root	1.7	$self->{hdl}->push_read (line => sub {
286			my ($hdl, $rline) = @_;
287	root	1.2
288	root	1.7	my ($auth_method, $rauth2, $r_framing) = split /;/, $rline;
289	root	1.1
290	root	1.19	my $rauth =
291			$auth_method eq "hmac_md6_64_256" ? Digest::HMAC_MD6::hmac_md6_hex $key, "$rgreeting1\012$rgreeting2\012$lgreeting1\012$lgreeting2\012", 64, 256
292			: $auth_method eq "cleartext" ? unpack "H*", $secret
293	root	1.41	: $auth_method eq "tls_anon" ? ($tls ? "" : "\012\012") # \012\012 never matches
294			: $auth_method eq "tls_md6_64_256" ? ($tls ? Digest::MD6::md6_hex "$rgreeting1\012$rgreeting2\012$lgreeting1\012$lgreeting2\012" : "\012\012")
295			: return $self->error ("$auth_method: fatal, selected unsupported rcv auth method");
296	root	1.19
297	root	1.7	if ($rauth2 ne $rauth) {
298			return $self->error ("authentication failure/shared secret mismatch");
299			}
300	root	1.1
301	root	1.7	$self->{s_framing} = $s_framing;
302	root	1.2
303	root	1.7	$hdl->rbuf_max (undef);
304			my $queue = delete $self->{queue}; # we are connected
305	root	1.1
306	root	1.27	$self->{hdl}->rtimeout ($self->{remote_greeting}{timeout});
307	root	1.42	$self->{hdl}->wtimeout ($self->{timeout} - LATENCY);
308	root	1.36	$self->{hdl}->on_wtimeout (sub { $self->send ([]) });
309	root	1.24
310	root	1.7	$self->connected;
311	root	1.1
312	root	1.27	# send queued messages
313	root	1.23	$self->send ($_)
314	root	1.7	for @$queue;
315	root	1.1
316	root	1.27	# receive handling
317			my $src_node = $self->{node};
318
319	root	1.22	my $rmsg; $rmsg = sub {
320	root	1.7	$_[0]->push_read ($r_framing => $rmsg);
321	root	1.1
322	root	1.31	local $AnyEvent::MP::Kernel::SRCNODE = $src_node;
323			AnyEvent::MP::Kernel::_inject (@{ $_[1] });
324	root	1.7	};
325			$hdl->push_read ($r_framing => $rmsg);
326			});
327	root	1.1	});
328			});
329			}
330
331			$self
332			}
333
334			sub error {
335			my ($self, $msg) = @_;
336
337	root	1.39	delete $self->{keepalive};
338
339	root	1.40	# $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} $msg");#d#
340	root	1.39
341			$self->{node}->transport_error (transport_error => $self->{node}{id}, $msg)
342	root	1.31	if $self->{node} && $self->{node}{transport} == $self;
343
344			(delete $self->{release})->()
345			if exists $self->{release};
346
347	root	1.37	# $AnyEvent::MP::Kernel::WARN->(7, "$self->{peerhost}:$self->{peerport}: $msg");
348	root	1.4	$self->destroy;
349	root	1.1	}
350
351	root	1.2	sub connected {
352			my ($self) = @_;
353
354	root	1.39	delete $self->{keepalive};
355
356	root	1.31	(delete $self->{release})->()
357			if exists $self->{release};
358	root	1.23
359	root	1.39	$AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} connected as $self->{remote_node}");
360
361	root	1.31	my $node = AnyEvent::MP::Kernel::add_node ($self->{remote_node});
362	root	1.4	Scalar::Util::weaken ($self->{node} = $node);
363	root	1.31	$node->transport_connect ($self);
364	root	1.2	}
365
366	root	1.1	sub send {
367	root	1.2	$_[0]{hdl}->push_write ($_[0]{s_framing} => $_[1]);
368	root	1.1	}
369
370			sub destroy {
371			my ($self) = @_;
372
373	root	1.42	(delete $self->{release})->()
374			if exists $self->{release};
375
376	root	1.2	$self->{hdl}->destroy
377			if $self->{hdl};
378	root	1.1	}
379
380			sub DESTROY {
381			my ($self) = @_;
382
383			$self->destroy;
384			}
385
386			=back
387
388	root	1.7	=head1 PROTOCOL
389
390			The protocol is relatively simple, and consists of three phases which are
391			symmetrical for both sides: greeting (followed by optionally switching to
392			TLS mode), authentication and packet exchange.
393
394			the protocol is designed to allow both full-text and binary streams.
395
396			The greeting consists of two text lines that are ended by either an ASCII
397			CR LF pair, or a single ASCII LF (recommended).
398
399			=head2 GREETING
400
401	root	1.15	All the lines until after authentication must not exceed 4kb in length,
402			including delimiter. Afterwards there is no limit on the packet size that
403			can be received.
404
405			=head3 First Greeting Line
406	root	1.12
407	root	1.16	Example:
408
409			aemp;0;fec.4a7720fc;127.0.0.1:1235,[::1]:1235;hmac_md6_64_256;json,storable;provider=AE-0.0
410
411			The first line contains strings separated (not ended) by C<;>
412			characters. The first even ixtrings are fixed by the protocol, the
413			remaining strings are C<KEY=VALUE> pairs. None of them may contain C<;>
414			characters themselves.
415
416	root	1.12	The fixed strings are:
417	root	1.7
418			=over 4
419
420	root	1.18	=item protocol identification
421	root	1.7
422			The constant C<aemp> to identify the protocol.
423
424			=item protocol version
425
426	root	1.12	The protocol version supported by this end, currently C<0>. If the
427			versions don't match then no communication is possible. Minor extensions
428	root	1.18	are supposed to be handled through additional key-value pairs.
429	root	1.7
430	root	1.39	=item the node id
431	root	1.7
432	root	1.39	This is the node ID of the connecting node.
433	root	1.7
434			=item the acceptable authentication methods
435
436			A comma-separated list of authentication methods supported by the
437			node. Note that AnyEvent::MP supports a C<hex_secret> authentication
438			method that accepts a cleartext password (hex-encoded), but will not use
439			this auth method itself.
440
441			The receiving side should choose the first auth method it supports.
442
443			=item the acceptable framing formats
444
445			A comma-separated list of packet encoding/framign formats understood. The
446			receiving side should choose the first framing format it supports for
447			sending packets (which might be different from the format it has to accept).
448
449	root	1.10	=back
450	root	1.8
451			The remaining arguments are C<KEY=VALUE> pairs. The following key-value
452			pairs are known at this time:
453
454			=over 4
455
456			=item provider=<module-version>
457
458			The software provider for this implementation. For AnyEvent::MP, this is
459			C<AE-0.0> or whatever version it currently is at.
460
461			=item peeraddr=<host>:<port>
462
463	root	1.39	The peer address (socket address of the other side) as seen locally.
464	root	1.8
465			=item tls=<major>.<minor>
466
467			Indicates that the other side supports TLS (version should be 1.0) and
468			wishes to do a TLS handshake.
469
470	root	1.26	=item timeout=<seconds>
471	root	1.24
472	root	1.26	The amount of time after which this node should be detected as dead unless
473			some data has been received. The node is responsible to send traffic
474			reasonably more often than this interval (such as every timeout minus five
475			seconds).
476	root	1.24
477	root	1.8	=back
478
479	root	1.15	=head3 Second Greeting Line
480
481	root	1.8	After this greeting line there will be a second line containing a
482			cryptographic nonce, i.e. random data of high quality. To keep the
483			protocol text-only, these are usually 32 base64-encoded octets, but
484			it could be anything that doesn't contain any ASCII CR or ASCII LF
485			characters.
486
487	root	1.14	I<< The two nonces B<must> be different, and an aemp implementation
488			B<must> check and fail when they are identical >>.
489
490	root	1.16	Example of a nonce line:
491	root	1.8
492	root	1.12	p/I122ql7kJR8lumW3lXlXCeBnyDAvz8NQo3x5IFowE4
493	root	1.8
494			=head2 TLS handshake
495
496	root	1.14	I<< If, after the handshake, both sides indicate interest in TLS, then the
497	root	1.20	connection B<must> use TLS, or fail. >>
498	root	1.8
499			Both sides compare their nonces, and the side who sent the lower nonce
500			value ("string" comparison on the raw octet values) becomes the client,
501			and the one with the higher nonce the server.
502
503			=head2 AUTHENTICATION PHASE
504
505			After the greeting is received (and the optional TLS handshake),
506			the authentication phase begins, which consists of sending a single
507			C<;>-separated line with three fixed strings and any number of
508			C<KEY=VALUE> pairs.
509
510			The three fixed strings are:
511
512			=over 4
513
514			=item the authentication method chosen
515
516			This must be one of the methods offered by the other side in the greeting.
517
518	root	1.41	Note that all methods starting with C<tls_> are only valid I<iff> TLS was
519			successfully handshaked (and to be secure the implementation must enforce
520			this).
521
522	root	1.13	The currently supported authentication methods are:
523
524			=over 4
525
526			=item cleartext
527
528			This is simply the shared secret, lowercase-hex-encoded. This method is of
529			course very insecure, unless TLS is used, which is why this module will
530			accept, but not generate, cleartext auth replies.
531
532			=item hmac_md6_64_256
533
534			This method uses an MD6 HMAC with 64 bit blocksize and 256 bit hash. First, the shared secret
535			is hashed with MD6:
536
537			key = MD6 (secret)
538
539			This secret is then used to generate the "local auth reply", by taking
540			the two local greeting lines and the two remote greeting lines (without
541			line endings), appending \012 to all of them, concatenating them and
542			calculating the MD6 HMAC with the key.
543
544			lauth = HMAC_MD6 key, "lgreeting1\012lgreeting2\012rgreeting1\012rgreeting2\012"
545
546			This authentication token is then lowercase-hex-encoded and sent to the
547			other side.
548
549			Then the remote auth reply is generated using the same method, but local
550			and remote greeting lines swapped:
551
552			rauth = HMAC_MD6 key, "rgreeting1\012rgreeting2\012lgreeting1\012lgreeting2\012"
553
554			This is the token that is expected from the other side.
555
556	root	1.41	=item tls_anon
557	root	1.19
558			This type is only valid iff TLS was enabled and the TLS handshake
559			was successful. It has no authentication data, as the server/client
560			certificate was successfully verified.
561
562	root	1.41	This authentication type is slightly less secure than the others, as it
563			allows a man-in-the-middle attacker to change some of the connection
564			parameters (such as the framing format), although there is no known attack
565			that exploits this in a way that is worse than just denying the service.
566
567			By default, this implementation accepts but uses this authentication
568			method.
569
570			=item tls_md6_64_256
571
572			This type is only valid iff TLS was enabled and the TLS handshake
573			was successful.
574
575			This authentication type simply calculates:
576
577			lauth = MD6 "rgreeting1\012rgreeting2\012lgreeting1\012lgreeting2\012"
578
579			and lowercase-hex encodes the result and sends it as authentication
580			data. No shared secret is required (authentication is done by TLS). The
581			checksum solely exists to make tinkering with the greeting hard.
582	root	1.19
583	root	1.13	=back
584
585	root	1.8	=item the authentication data
586
587	root	1.13	The authentication data itself, usually base64 or hex-encoded data, see
588			above.
589	root	1.8
590			=item the framing protocol chosen
591
592			This must be one of the framing protocols offered by the other side in the
593			greeting. Each side must accept the choice of the other side.
594
595			=back
596
597	root	1.16	Example of an authentication reply:
598	root	1.9
599	root	1.13	hmac_md6_64_256;363d5175df38bd9eaddd3f6ca18aa1c0c4aa22f0da245ac638d048398c26b8d3;json
600	root	1.9
601	root	1.8	=head2 DATA PHASE
602
603			After this, packets get exchanged using the chosen framing protocol. It is
604			quite possible that both sides use a different framing protocol.
605
606	root	1.16	=head2 FULL EXAMPLE
607
608	root	1.17	This is an actual protocol dump of a handshake, followed by a single data
609	root	1.16	packet. The greater than/less than lines indicate the direction of the
610			transfer only.
611
612			> aemp;0;nndKd+gn;10.0.0.1:4040;hmac_md6_64_256,cleartext;json,storable;provider=AE-0.0;peeraddr=127.0.0.1:1235
613			> sRG8bbc4TDbkpvH8FTP4HBs87OhepH6VuApoZqXXskuG
614			< aemp;0;nmpKd+gh;127.0.0.1:1235,[::1]:1235;hmac_md6_64_256,cleartext;json,storable;provider=AE-0.0;peeraddr=127.0.0.1:58760
615			< dCEUcL/LJVSTJcx8byEsOzrwhzJYOq+L3YcopA5T6EAo
616			> hmac_md6_64_256;9513d4b258975accfcb2ab7532b83690e9c119a502c612203332a591c7237788;json
617			< hmac_md6_64_256;0298d6ba2240faabb2b2e881cf86b97d70a113ca74a87dc006f9f1e9d3010f90;json
618	root	1.18	> ["","lookup","pinger","10.0.0.1:4040#nndKd+gn.a","resolved"]
619	root	1.16
620	root	1.1	=head1 SEE ALSO
621
622	root	1.29	L<AnyEvent::MP>.
623	root	1.1
624			=head1 AUTHOR
625
626			Marc Lehmann <schmorp@schmorp.de>
627			http://home.schmorp.de/
628
629			=cut
630
631			1
632