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

# Content
1 =head1 NAME
2
3 AnyEvent::MP::Transport - actual transport protocol handler
4
5 =head1 SYNOPSIS
6
7 use AnyEvent::MP::Transport;
8
9 =head1 DESCRIPTION
10
11 This implements the actual transport protocol for MP (it represents a
12 single link), most of which is considered an implementation detail.
13
14 See the "PROTOCOL" section below if you want to write another client for
15 this protocol.
16
17 =head1 FUNCTIONS/METHODS
18
19 =over 4
20
21 =cut
22
23 package AnyEvent::MP::Transport;
24
25 use common::sense;
26
27 use Scalar::Util ();
28 use List::Util ();
29 use MIME::Base64 ();
30 use Storable ();
31 use JSON::XS ();
32
33 use Digest::MD6 ();
34 use Digest::HMAC_MD6 ();
35
36 use AE ();
37 use AnyEvent::Socket ();
38 use AnyEvent::Handle 4.92 ();
39
40 use AnyEvent::MP::Config ();
41
42 our $PROTOCOL_VERSION = 0;
43
44 =item $listener = mp_listener $host, $port, <constructor-args>
45
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 Defaults for peerhost, peerport and fh are provided.
52
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 my $tp = new AnyEvent::MP::Transport
62 fh => $fh,
63 peerhost => $host,
64 peerport => $port,
65 @args,
66 ;
67 $tp->{keepalive} = $tp;
68 }
69 }
70
71 =item $guard = mp_connect $host, $port, <constructor-args>, $cb->($transport)
72
73 =cut
74
75 sub mp_connect {
76 my $release = pop;
77 my ($host, $port, @args) = @_;
78
79 my $state;
80
81 $state = AnyEvent::Socket::tcp_connect $host, $port, sub {
82 my ($fh, $nhost, $nport) = @_;
83
84 return $release->() unless $fh;
85
86 $state = new AnyEvent::MP::Transport
87 fh => $fh,
88 peername => $host,
89 peerhost => $nhost,
90 peerport => $nport,
91 release => $release,
92 @args,
93 ;
94 };
95
96 \$state
97 }
98
99 =item new AnyEvent::MP::Transport
100
101 # immediately starts negotiation
102 my $transport = new AnyEvent::MP::Transport
103 # mandatory
104 fh => $filehandle,
105 local_id => $identifier,
106 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 greeting => { key => value },
113
114 # tls support
115 tls_ctx => AnyEvent::TLS,
116 peername => $peername, # for verification
117 ;
118
119 =cut
120
121 sub LATENCY() { 3 } # assumed max. network latency
122
123 our @FRAMINGS = qw(json storable); # the framing types we accept and send, in order of preference
124 our @AUTH_SND = qw(hmac_md6_64_256); # auth types we send
125 our @AUTH_RCV = (@AUTH_SND, qw(cleartext)); # auth types we accept
126
127 #AnyEvent::Handle::register_write_type mp_record => sub {
128 #};
129
130 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 my $config = AnyEvent::MP::Config::config;
141
142 my $latency = $config->{network_latency} || LATENCY;
143
144 $arg{secret} = $config->{secret}
145 unless exists $arg{secret};
146
147 $arg{timeout} = $config->{monitor_timeout} || $AnyEvent::MP::Kernel::MONITOR_TIMEOUT
148 unless exists $arg{timeout};
149
150 $arg{timeout} -= $latency;
151
152 $arg{timeout} = 1 + $latency
153 if $arg{timeout} < 1 + $latency;
154
155 my $secret = $arg{secret};
156
157 if (exists $config->{cert}) {
158 $arg{tls_ctx} = {
159 sslv2 => 0,
160 sslv3 => 0,
161 tlsv1 => 1,
162 verify => 1,
163 cert => $config->{cert},
164 ca_cert => $config->{cert},
165 verify_require_client_cert => 1,
166 };
167 }
168
169 $self->{hdl} = new AnyEvent::Handle
170 fh => delete $arg{fh},
171 autocork => 1,
172 no_delay => 1,
173 on_error => sub {
174 $self->error ($_[2]);
175 },
176 rtimeout => $latency,
177 peername => delete $arg{peername},
178 ;
179
180 my $greeting_kv = $self->{greeting} ||= {};
181
182 $self->{local_node} ||= $AnyEvent::MP::Kernel::NODE;
183
184 $greeting_kv->{"tls"} = "1.0" if $arg{tls_ctx};
185 $greeting_kv->{provider} = "AE-$AnyEvent::MP::Kernel::VERSION";
186 $greeting_kv->{peeraddr} = AnyEvent::Socket::format_hostport $self->{peerhost}, $self->{peerport};
187 $greeting_kv->{timeout} = $arg{timeout};
188
189 # send greeting
190 my $lgreeting1 = "aemp;$PROTOCOL_VERSION"
191 . ";$self->{local_node}"
192 . ";" . (join ",", @AUTH_RCV)
193 . ";" . (join ",", @FRAMINGS)
194 . (join "", map ";$_=$greeting_kv->{$_}", keys %$greeting_kv);
195
196 my $lgreeting2 = MIME::Base64::encode_base64 AnyEvent::MP::Kernel::nonce (66), "";
197
198 $self->{hdl}->push_write ("$lgreeting1\012$lgreeting2\012");
199
200 # expect greeting
201 $self->{hdl}->rbuf_max (4 * 1024);
202 $self->{hdl}->push_read (line => sub {
203 my $rgreeting1 = $_[1];
204
205 my ($aemp, $version, $rnode, $auths, $framings, @kv) = split /;/, $rgreeting1;
206
207 if ($aemp ne "aemp") {
208 return $self->error ("unparsable greeting");
209 } elsif ($version != $PROTOCOL_VERSION) {
210 return $self->error ("version mismatch (we: $PROTOCOL_VERSION, they: $version)");
211 } 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 }
216
217 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 $self->{remote_node} = $rnode;
242
243 $self->{remote_greeting} = {
244 map /^([^=]+)(?:=(.*))?/ ? ($1 => $2) : (),
245 @kv
246 };
247
248 # read nonce
249 $self->{hdl}->push_read (line => sub {
250 my $rgreeting2 = $_[1];
251
252 "$lgreeting1\012$lgreeting2" ne "$rgreeting1\012$rgreeting2" # echo attack?
253 or return $self->error ("authentication error, echo attack?");
254
255 my $key;
256 my $lauth;
257
258 if ($self->{tls_ctx} and 1 == int $self->{remote_greeting}{tls}) {
259 $self->{tls} = $lgreeting2 lt $rgreeting2 ? "connect" : "accept";
260 $self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});
261 $s_auth = "tls";
262 $lauth = "";
263 } elsif (length $secret) {
264 $key = Digest::MD6::md6 $secret;
265 # 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 } else {
268 return $self->error ("unable to handshake TLS and no shared secret configured");
269 }
270
271 $self->{hdl}->push_write ("$s_auth;$lauth;$s_framing\012");
272
273 # read the authentication response
274 $self->{hdl}->push_read (line => sub {
275 my ($hdl, $rline) = @_;
276
277 my ($auth_method, $rauth2, $r_framing) = split /;/, $rline;
278
279 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 if ($rauth2 ne $rauth) {
286 return $self->error ("authentication failure/shared secret mismatch");
287 }
288
289 $self->{s_framing} = $s_framing;
290
291 $hdl->rbuf_max (undef);
292 my $queue = delete $self->{queue}; # we are connected
293
294 $self->{hdl}->rtimeout ($self->{remote_greeting}{timeout});
295 $self->{hdl}->wtimeout ($arg{timeout} - LATENCY);
296 $self->{hdl}->on_wtimeout (sub { $self->send ([]) });
297
298 $self->connected;
299
300 # send queued messages
301 $self->send ($_)
302 for @$queue;
303
304 # receive handling
305 my $src_node = $self->{node};
306
307 my $rmsg; $rmsg = sub {
308 $_[0]->push_read ($r_framing => $rmsg);
309
310 local $AnyEvent::MP::Kernel::SRCNODE = $src_node;
311 AnyEvent::MP::Kernel::_inject (@{ $_[1] });
312 };
313 $hdl->push_read ($r_framing => $rmsg);
314 });
315 });
316 });
317 }
318
319 $self
320 }
321
322 sub error {
323 my ($self, $msg) = @_;
324
325 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 if $self->{node} && $self->{node}{transport} == $self;
331
332 (delete $self->{release})->()
333 if exists $self->{release};
334
335 # $AnyEvent::MP::Kernel::WARN->(7, "$self->{peerhost}:$self->{peerport}: $msg");
336 $self->destroy;
337 }
338
339 sub connected {
340 my ($self) = @_;
341
342 delete $self->{keepalive};
343
344 (delete $self->{release})->()
345 if exists $self->{release};
346
347 $AnyEvent::MP::Kernel::WARN->(9, "$self->{peerhost}:$self->{peerport} connected as $self->{remote_node}");
348
349 my $node = AnyEvent::MP::Kernel::add_node ($self->{remote_node});
350 Scalar::Util::weaken ($self->{node} = $node);
351 $node->transport_connect ($self);
352 }
353
354 sub send {
355 $_[0]{hdl}->push_write ($_[0]{s_framing} => $_[1]);
356 }
357
358 sub destroy {
359 my ($self) = @_;
360
361 $self->{hdl}->destroy
362 if $self->{hdl};
363 }
364
365 sub DESTROY {
366 my ($self) = @_;
367
368 $self->destroy;
369 }
370
371 =back
372
373 =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 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
392 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 The fixed strings are:
402
403 =over 4
404
405 =item protocol identification
406
407 The constant C<aemp> to identify the protocol.
408
409 =item protocol version
410
411 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 are supposed to be handled through additional key-value pairs.
414
415 =item the node id
416
417 This is the node ID of the connecting node.
418
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 =back
435
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 The peer address (socket address of the other side) as seen locally.
449
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 =item timeout=<seconds>
456
457 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
462 =back
463
464 =head3 Second Greeting Line
465
466 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 I<< The two nonces B<must> be different, and an aemp implementation
473 B<must> check and fail when they are identical >>.
474
475 Example of a nonce line:
476
477 p/I122ql7kJR8lumW3lXlXCeBnyDAvz8NQo3x5IFowE4
478
479 =head2 TLS handshake
480
481 I<< If, after the handshake, both sides indicate interest in TLS, then the
482 connection B<must> use TLS, or fail. >>
483
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 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 =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 =back
546
547 =item the authentication data
548
549 The authentication data itself, usually base64 or hex-encoded data, see
550 above.
551
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 Example of an authentication reply:
560
561 hmac_md6_64_256;363d5175df38bd9eaddd3f6ca18aa1c0c4aa22f0da245ac638d048398c26b8d3;json
562
563 =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 =head2 FULL EXAMPLE
569
570 This is an actual protocol dump of a handshake, followed by a single data
571 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 > ["","lookup","pinger","10.0.0.1:4040#nndKd+gn.a","resolved"]
581
582 =head1 SEE ALSO
583
584 L<AnyEvent::MP>.
585
586 =head1 AUTHOR
587
588 Marc Lehmann <schmorp@schmorp.de>
589 http://home.schmorp.de/
590
591 =cut
592
593 1
594