ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP/Transport.pm
Revision: 1.39
Committed: Thu Aug 27 21:29:37 2009 UTC (14 years, 9 months ago) by root
Branch: MAIN
Changes since 1.38: +20 -12 lines
Log Message:
*** empty log message ***

File Contents

# 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.7 our @AUTH_SND = qw(hmac_md6_64_256); # auth types we send
125 root 1.13 our @AUTH_RCV = (@AUTH_SND, qw(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.30 $arg{secret} = $config->{secret}
145 root 1.2 unless exists $arg{secret};
146    
147 root 1.31 $arg{timeout} = $config->{monitor_timeout} || $AnyEvent::MP::Kernel::MONITOR_TIMEOUT
148 root 1.24 unless exists $arg{timeout};
149    
150 root 1.31 $arg{timeout} -= $latency;
151    
152     $arg{timeout} = 1 + $latency
153     if $arg{timeout} < 1 + $latency;
154 root 1.27
155 root 1.19 my $secret = $arg{secret};
156    
157 root 1.30 if (exists $config->{cert}) {
158 root 1.19 $arg{tls_ctx} = {
159     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.2 fh => delete $arg{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.1 peername => delete $arg{peername},
178     ;
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     $greeting_kv->{"tls"} = "1.0" if $arg{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.26 $greeting_kv->{timeout} = $arg{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     return $self->error ("I refuse to talk to myself");
213     } elsif ($AnyEvent::MP::Kernel::NODE{$rnode} && $AnyEvent::MP::Kernel::NODE{$rnode}{transport}) {
214     return $self->error ("$rnode already connected, not connecting again.");
215 root 1.1 }
216    
217 root 1.7 my $s_auth;
218     for my $auth_ (split /,/, $auths) {
219     if (grep $auth_ eq $_, @AUTH_SND) {
220     $s_auth = $auth_;
221     last;
222     }
223     }
224    
225     defined $s_auth
226     or return $self->error ("$auths: no common auth type supported");
227    
228     die unless $s_auth eq "hmac_md6_64_256"; # hardcoded atm.
229    
230     my $s_framing;
231     for my $framing_ (split /,/, $framings) {
232     if (grep $framing_ eq $_, @FRAMINGS) {
233     $s_framing = $framing_;
234     last;
235     }
236     }
237    
238     defined $s_framing
239     or return $self->error ("$framings: no common framing method supported");
240    
241 root 1.2 $self->{remote_node} = $rnode;
242 root 1.1
243 root 1.2 $self->{remote_greeting} = {
244     map /^([^=]+)(?:=(.*))?/ ? ($1 => $2) : (),
245     @kv
246 root 1.1 };
247    
248 root 1.7 # read nonce
249     $self->{hdl}->push_read (line => sub {
250     my $rgreeting2 = $_[1];
251    
252 root 1.19 "$lgreeting1\012$lgreeting2" ne "$rgreeting1\012$rgreeting2" # echo attack?
253     or return $self->error ("authentication error, echo attack?");
254    
255 root 1.30 my $key;
256 root 1.19 my $lauth;
257    
258 root 1.10 if ($self->{tls_ctx} and 1 == int $self->{remote_greeting}{tls}) {
259 root 1.8 $self->{tls} = $lgreeting2 lt $rgreeting2 ? "connect" : "accept";
260     $self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});
261 root 1.19 $s_auth = "tls";
262     $lauth = "";
263 root 1.30 } elsif (length $secret) {
264     $key = Digest::MD6::md6 $secret;
265 root 1.19 # we currently only support hmac_md6_64_256
266     $lauth = Digest::HMAC_MD6::hmac_md6_hex $key, "$lgreeting1\012$lgreeting2\012$rgreeting1\012$rgreeting2\012", 64, 256;
267 root 1.30 } else {
268     return $self->error ("unable to handshake TLS and no shared secret configured");
269 root 1.8 }
270 root 1.2
271 root 1.7 $self->{hdl}->push_write ("$s_auth;$lauth;$s_framing\012");
272 root 1.2
273 root 1.19 # read the authentication response
274 root 1.7 $self->{hdl}->push_read (line => sub {
275     my ($hdl, $rline) = @_;
276 root 1.2
277 root 1.7 my ($auth_method, $rauth2, $r_framing) = split /;/, $rline;
278 root 1.1
279 root 1.19 my $rauth =
280     $auth_method eq "hmac_md6_64_256" ? Digest::HMAC_MD6::hmac_md6_hex $key, "$rgreeting1\012$rgreeting2\012$lgreeting1\012$lgreeting2\012", 64, 256
281     : $auth_method eq "cleartext" ? unpack "H*", $secret
282     : $auth_method eq "tls" ? ($self->{tls} ? "" : "\012\012") # \012\012 never matches
283     : return $self->error ("$auth_method: fatal, selected unsupported auth method");
284    
285 root 1.7 if ($rauth2 ne $rauth) {
286     return $self->error ("authentication failure/shared secret mismatch");
287     }
288 root 1.1
289 root 1.7 $self->{s_framing} = $s_framing;
290 root 1.2
291 root 1.7 $hdl->rbuf_max (undef);
292     my $queue = delete $self->{queue}; # we are connected
293 root 1.1
294 root 1.27 $self->{hdl}->rtimeout ($self->{remote_greeting}{timeout});
295     $self->{hdl}->wtimeout ($arg{timeout} - LATENCY);
296 root 1.36 $self->{hdl}->on_wtimeout (sub { $self->send ([]) });
297 root 1.24
298 root 1.7 $self->connected;
299 root 1.1
300 root 1.27 # send queued messages
301 root 1.23 $self->send ($_)
302 root 1.7 for @$queue;
303 root 1.1
304 root 1.27 # receive handling
305     my $src_node = $self->{node};
306    
307 root 1.22 my $rmsg; $rmsg = sub {
308 root 1.7 $_[0]->push_read ($r_framing => $rmsg);
309 root 1.1
310 root 1.31 local $AnyEvent::MP::Kernel::SRCNODE = $src_node;
311     AnyEvent::MP::Kernel::_inject (@{ $_[1] });
312 root 1.7 };
313     $hdl->push_read ($r_framing => $rmsg);
314     });
315 root 1.1 });
316     });
317     }
318    
319     $self
320     }
321    
322     sub error {
323     my ($self, $msg) = @_;
324    
325 root 1.39 delete $self->{keepalive};
326    
327     $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} $msg");#d#
328    
329     $self->{node}->transport_error (transport_error => $self->{node}{id}, $msg)
330 root 1.31 if $self->{node} && $self->{node}{transport} == $self;
331    
332     (delete $self->{release})->()
333     if exists $self->{release};
334    
335 root 1.37 # $AnyEvent::MP::Kernel::WARN->(7, "$self->{peerhost}:$self->{peerport}: $msg");
336 root 1.4 $self->destroy;
337 root 1.1 }
338    
339 root 1.2 sub connected {
340     my ($self) = @_;
341    
342 root 1.39 delete $self->{keepalive};
343    
344 root 1.31 (delete $self->{release})->()
345     if exists $self->{release};
346 root 1.23
347 root 1.39 $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} connected as $self->{remote_node}");
348    
349 root 1.31 my $node = AnyEvent::MP::Kernel::add_node ($self->{remote_node});
350 root 1.4 Scalar::Util::weaken ($self->{node} = $node);
351 root 1.31 $node->transport_connect ($self);
352 root 1.2 }
353    
354 root 1.1 sub send {
355 root 1.2 $_[0]{hdl}->push_write ($_[0]{s_framing} => $_[1]);
356 root 1.1 }
357    
358     sub destroy {
359     my ($self) = @_;
360    
361 root 1.2 $self->{hdl}->destroy
362     if $self->{hdl};
363 root 1.1 }
364    
365     sub DESTROY {
366     my ($self) = @_;
367    
368     $self->destroy;
369     }
370    
371     =back
372    
373 root 1.7 =head1 PROTOCOL
374    
375     The protocol is relatively simple, and consists of three phases which are
376     symmetrical for both sides: greeting (followed by optionally switching to
377     TLS mode), authentication and packet exchange.
378    
379     the protocol is designed to allow both full-text and binary streams.
380    
381     The greeting consists of two text lines that are ended by either an ASCII
382     CR LF pair, or a single ASCII LF (recommended).
383    
384     =head2 GREETING
385    
386 root 1.15 All the lines until after authentication must not exceed 4kb in length,
387     including delimiter. Afterwards there is no limit on the packet size that
388     can be received.
389    
390     =head3 First Greeting Line
391 root 1.12
392 root 1.16 Example:
393    
394     aemp;0;fec.4a7720fc;127.0.0.1:1235,[::1]:1235;hmac_md6_64_256;json,storable;provider=AE-0.0
395    
396     The first line contains strings separated (not ended) by C<;>
397     characters. The first even ixtrings are fixed by the protocol, the
398     remaining strings are C<KEY=VALUE> pairs. None of them may contain C<;>
399     characters themselves.
400    
401 root 1.12 The fixed strings are:
402 root 1.7
403     =over 4
404    
405 root 1.18 =item protocol identification
406 root 1.7
407     The constant C<aemp> to identify the protocol.
408    
409     =item protocol version
410    
411 root 1.12 The protocol version supported by this end, currently C<0>. If the
412     versions don't match then no communication is possible. Minor extensions
413 root 1.18 are supposed to be handled through additional key-value pairs.
414 root 1.7
415 root 1.39 =item the node id
416 root 1.7
417 root 1.39 This is the node ID of the connecting node.
418 root 1.7
419     =item the acceptable authentication methods
420    
421     A comma-separated list of authentication methods supported by the
422     node. Note that AnyEvent::MP supports a C<hex_secret> authentication
423     method that accepts a cleartext password (hex-encoded), but will not use
424     this auth method itself.
425    
426     The receiving side should choose the first auth method it supports.
427    
428     =item the acceptable framing formats
429    
430     A comma-separated list of packet encoding/framign formats understood. The
431     receiving side should choose the first framing format it supports for
432     sending packets (which might be different from the format it has to accept).
433    
434 root 1.10 =back
435 root 1.8
436     The remaining arguments are C<KEY=VALUE> pairs. The following key-value
437     pairs are known at this time:
438    
439     =over 4
440    
441     =item provider=<module-version>
442    
443     The software provider for this implementation. For AnyEvent::MP, this is
444     C<AE-0.0> or whatever version it currently is at.
445    
446     =item peeraddr=<host>:<port>
447    
448 root 1.39 The peer address (socket address of the other side) as seen locally.
449 root 1.8
450     =item tls=<major>.<minor>
451    
452     Indicates that the other side supports TLS (version should be 1.0) and
453     wishes to do a TLS handshake.
454    
455 root 1.26 =item timeout=<seconds>
456 root 1.24
457 root 1.26 The amount of time after which this node should be detected as dead unless
458     some data has been received. The node is responsible to send traffic
459     reasonably more often than this interval (such as every timeout minus five
460     seconds).
461 root 1.24
462 root 1.8 =back
463    
464 root 1.15 =head3 Second Greeting Line
465    
466 root 1.8 After this greeting line there will be a second line containing a
467     cryptographic nonce, i.e. random data of high quality. To keep the
468     protocol text-only, these are usually 32 base64-encoded octets, but
469     it could be anything that doesn't contain any ASCII CR or ASCII LF
470     characters.
471    
472 root 1.14 I<< The two nonces B<must> be different, and an aemp implementation
473     B<must> check and fail when they are identical >>.
474    
475 root 1.16 Example of a nonce line:
476 root 1.8
477 root 1.12 p/I122ql7kJR8lumW3lXlXCeBnyDAvz8NQo3x5IFowE4
478 root 1.8
479     =head2 TLS handshake
480    
481 root 1.14 I<< If, after the handshake, both sides indicate interest in TLS, then the
482 root 1.20 connection B<must> use TLS, or fail. >>
483 root 1.8
484     Both sides compare their nonces, and the side who sent the lower nonce
485     value ("string" comparison on the raw octet values) becomes the client,
486     and the one with the higher nonce the server.
487    
488     =head2 AUTHENTICATION PHASE
489    
490     After the greeting is received (and the optional TLS handshake),
491     the authentication phase begins, which consists of sending a single
492     C<;>-separated line with three fixed strings and any number of
493     C<KEY=VALUE> pairs.
494    
495     The three fixed strings are:
496    
497     =over 4
498    
499     =item the authentication method chosen
500    
501     This must be one of the methods offered by the other side in the greeting.
502    
503 root 1.13 The currently supported authentication methods are:
504    
505     =over 4
506    
507     =item cleartext
508    
509     This is simply the shared secret, lowercase-hex-encoded. This method is of
510     course very insecure, unless TLS is used, which is why this module will
511     accept, but not generate, cleartext auth replies.
512    
513     =item hmac_md6_64_256
514    
515     This method uses an MD6 HMAC with 64 bit blocksize and 256 bit hash. First, the shared secret
516     is hashed with MD6:
517    
518     key = MD6 (secret)
519    
520     This secret is then used to generate the "local auth reply", by taking
521     the two local greeting lines and the two remote greeting lines (without
522     line endings), appending \012 to all of them, concatenating them and
523     calculating the MD6 HMAC with the key.
524    
525     lauth = HMAC_MD6 key, "lgreeting1\012lgreeting2\012rgreeting1\012rgreeting2\012"
526    
527     This authentication token is then lowercase-hex-encoded and sent to the
528     other side.
529    
530     Then the remote auth reply is generated using the same method, but local
531     and remote greeting lines swapped:
532    
533     rauth = HMAC_MD6 key, "rgreeting1\012rgreeting2\012lgreeting1\012lgreeting2\012"
534    
535     This is the token that is expected from the other side.
536    
537 root 1.19 =item tls
538    
539     This type is only valid iff TLS was enabled and the TLS handshake
540     was successful. It has no authentication data, as the server/client
541     certificate was successfully verified.
542    
543     Implementations supporting TLS I<must> accept this authentication type.
544    
545 root 1.13 =back
546    
547 root 1.8 =item the authentication data
548    
549 root 1.13 The authentication data itself, usually base64 or hex-encoded data, see
550     above.
551 root 1.8
552     =item the framing protocol chosen
553    
554     This must be one of the framing protocols offered by the other side in the
555     greeting. Each side must accept the choice of the other side.
556    
557     =back
558    
559 root 1.16 Example of an authentication reply:
560 root 1.9
561 root 1.13 hmac_md6_64_256;363d5175df38bd9eaddd3f6ca18aa1c0c4aa22f0da245ac638d048398c26b8d3;json
562 root 1.9
563 root 1.8 =head2 DATA PHASE
564    
565     After this, packets get exchanged using the chosen framing protocol. It is
566     quite possible that both sides use a different framing protocol.
567    
568 root 1.16 =head2 FULL EXAMPLE
569    
570 root 1.17 This is an actual protocol dump of a handshake, followed by a single data
571 root 1.16 packet. The greater than/less than lines indicate the direction of the
572     transfer only.
573    
574     > 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
575     > sRG8bbc4TDbkpvH8FTP4HBs87OhepH6VuApoZqXXskuG
576     < 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
577     < dCEUcL/LJVSTJcx8byEsOzrwhzJYOq+L3YcopA5T6EAo
578     > hmac_md6_64_256;9513d4b258975accfcb2ab7532b83690e9c119a502c612203332a591c7237788;json
579     < hmac_md6_64_256;0298d6ba2240faabb2b2e881cf86b97d70a113ca74a87dc006f9f1e9d3010f90;json
580 root 1.18 > ["","lookup","pinger","10.0.0.1:4040#nndKd+gn.a","resolved"]
581 root 1.16
582 root 1.1 =head1 SEE ALSO
583    
584 root 1.29 L<AnyEvent::MP>.
585 root 1.1
586     =head1 AUTHOR
587    
588     Marc Lehmann <schmorp@schmorp.de>
589     http://home.schmorp.de/
590    
591     =cut
592    
593     1
594