AnyEvent-MP/MP/Transport.pm

=head1 NAME

AnyEvent::MP::Transport - actual transport protocol

=head1 SYNOPSIS

   use AnyEvent::MP::Transport;

=head1 DESCRIPTION

This is the superclass for MP transports, most of which is considered an
implementation detail.

Future versions might document the actual protocol.

=head1 FUNCTIONS/METHODS

=over 4

=cut

package AnyEvent::MP::Transport;

use common::sense;

use Scalar::Util;
use MIME::Base64 ();
use Storable ();

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

use base Exporter::;

our $VERSION = '0.0';
our $PROTOCOL_VERSION_MAJOR = 0;
our $PROTOCOL_VERSION_MINOR = 0;

=item $listener = mp_listener $host, $port, <constructor-args>, $cb->($transport)

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, fh and tls are provided.

=cut

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

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

      $cb->(new AnyEvent::MP::Transport
         fh       => $fh,
         peerhost => $host,
         peerport => $port,
         tls      => "accept",
         @args,
      );
   }
}

=item new AnyEvent::MP::Transport

   # immediately starts negotiation
   my $transport = new AnyEvent::MP::Transport
      # fh OR connect is mandatory
      fh       => $filehandle,
      connect  => [$host, $port],

      # mandatory
      on_recv  => sub { receive-callback },
      on_error => sub { error-callback },

      # optional
      local_id => $identifier,
      secret   => "shared secret",
      on_eof   => sub { clean-close-callback },
      on_connect => sub { successful-connect-callback },

      # tls support
      tls      => "accept|connect",
      tls_ctx  => AnyEvent::TLS,
      peername => $peername, # for verification
   ;

=cut

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

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

   $self->{queue} = [];

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

      if (exists $arg{connect}) {
         $arg{tls}     ||= "connect";
         $arg{tls_ctx} ||= { sslv2 => 0, sslv3 => 0, tlsv1 => 1, verify => 1, verify_peername => "https" };
      }

      $self->{hdl} = new AnyEvent::Handle
         (exists $arg{fh} ? (fh => delete $arg{fh}) : (connect => delete $arg{connect})),
         on_error => sub {
            $self->error ($_[2]);
         },
         peername => delete $arg{peername},
      ;

      my $secret = delete $arg{secret} ? delete $arg{secret} : AnyEvent::MP::default_secret;

      # send greeting
      my $lgreeting = "aemp;$PROTOCOL_VERSION_MAJOR;$PROTOCOL_VERSION_MINOR;AnyEvent::MP;$VERSION;"
                    . (MIME::Base64::encode_base64 AnyEvent::MP::nonce 33, "") . ";"
                    . "hmac_md6_64_256;" # hardcoded atm.
                    . "storable;" # hardcoded atm.
                    . "$self->{local_id};"
                    . (exists $arg{tls} && $arg{tls_ctx} ? "tls1.0=$arg{tls};" : "");

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

      # expect greeting
      $self->{hdl}->push_read (line => sub {
         my $rgreeting = $_[1];

         my ($aemp, $major, $minor, $provider, $provider_version, $nonce2, $auth, $framing, $rid, @kv) = split /;/, $rgreeting;

         if ($aemp ne "aemp") {
            return $self->error ("unparsable greeting");
         } elsif ($major != $PROTOCOL_VERSION_MAJOR) {
            return $self->error ("major version mismatch ($PROTOCOL_VERSION_MAJOR vs. $major)");
         } elsif ($auth ne "hmac_md6_64_256") {
            return $self->error ("unsupported auth method ($auth)");
         } elsif ($framing ne "storable") {
            return $self->error ("unsupported auth method ($auth)");
         }

         $self->{remote_id} = $rid;

         $self->{greeting} = {
            provider         => $provider,
            provider_version => $provider_version,
         };

         /^([^=]+)(?:=(.*))?/ and $self->{greeting}{$1} = $2
            for @kv;

         if (exists $self->{tls} and $self->{tls_ctx} and exists $self->{greeting}{"tls1.0"}) {
            if ($self->{tls} ne $self->{greeting}{"tls1.0"}) {
               return $self->error ("TLS server/client mismatch");
            }
            $self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});
         }

         # auth
         require Digest::MD6;
         require Digest::HMAC_MD6;

         my $key   = Digest::MD6::md6_hex ($secret);
         my $lauth = Digest::HMAC_MD6::hmac_md6_base64 ($key, "$lgreeting$rgreeting", 64, 256);
         my $rauth = Digest::HMAC_MD6::hmac_md6_base64 ($key, "$rgreeting$lgreeting", 64, 256);
         $self->{hdl}->push_write ("$lauth\012");

         $self->{hdl}->push_read (line => sub {
            my ($hdl, $rauth2) = @_;

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

            my $queue = delete $self->{queue}; # we are connected

            $self->{on_connect}($self)
               if $self->{on_connect};

            $hdl->push_write (storable => $_)
               for @$queue;

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

               $self->{on_recv}($self, $_[1]);
            };
            $hdl->push_read (storable => $rmsg);
         });
      });
   }

   $self
}

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

   $self->{on_error}($self, $msg);
   $self->{hdl}->destroy;
}

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

   exists $self->{queue}
      ? push @{ $self->{queue} }, $msg
      : $self->{hdl}->push_write (storable => $msg);
}

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

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

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

   $self->destroy;
}

=back

=head1 SEE ALSO

L<AnyEvent>.

=head1 AUTHOR

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

=cut

1

Revision:	1.1
Committed:	Thu Jul 30 08:38:50 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MP::Transport - actual transport protocol
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::MP::Transport;
8
9			=head1 DESCRIPTION
10
11			This is the superclass for MP transports, most of which is considered an
12			implementation detail.
13
14			Future versions might document the actual protocol.
15
16			=head1 FUNCTIONS/METHODS
17
18			=over 4
19
20			=cut
21
22			package AnyEvent::MP::Transport;
23
24			use common::sense;
25
26			use Scalar::Util;
27			use MIME::Base64 ();
28			use Storable ();
29
30			use AE ();
31			use AnyEvent::Socket ();
32			use AnyEvent::Handle ();
33			use AnyEvent::MP ();
34
35			use base Exporter::;
36
37			our $VERSION = '0.0';
38			our $PROTOCOL_VERSION_MAJOR = 0;
39			our $PROTOCOL_VERSION_MINOR = 0;
40
41			=item $listener = mp_listener $host, $port, <constructor-args>, $cb->($transport)
42
43			Creates a listener on the given host/port using
44			C<AnyEvent::Socket::tcp_server>.
45
46			See C<new>, below, for constructor arguments.
47
48			Defaults for peerhost, peerport, fh and tls are provided.
49
50			=cut
51
52			sub mp_server($$@) {
53			my $cb = pop;
54			my ($host, $port, @args) = @_;
55
56			AnyEvent::Socket::tcp_server $host, $port, sub {
57			my ($fh, $host, $port) = @_;
58
59			$cb->(new AnyEvent::MP::Transport
60			fh => $fh,
61			peerhost => $host,
62			peerport => $port,
63			tls => "accept",
64			@args,
65			);
66			}
67			}
68
69			=item new AnyEvent::MP::Transport
70
71			# immediately starts negotiation
72			my $transport = new AnyEvent::MP::Transport
73			# fh OR connect is mandatory
74			fh => $filehandle,
75			connect => [$host, $port],
76
77			# mandatory
78			on_recv => sub { receive-callback },
79			on_error => sub { error-callback },
80
81			# optional
82			local_id => $identifier,
83			secret => "shared secret",
84			on_eof => sub { clean-close-callback },
85			on_connect => sub { successful-connect-callback },
86
87			# tls support
88			tls => "accept\|connect",
89			tls_ctx => AnyEvent::TLS,
90			peername => $peername, # for verification
91			;
92
93			=cut
94
95			sub new {
96			my ($class, %arg) = @_;
97
98			my $self = bless \%arg, $class;
99
100			$self->{queue} = [];
101
102			{
103			Scalar::Util::weaken (my $self = $self);
104
105			if (exists $arg{connect}) {
106			$arg{tls} \|\|= "connect";
107			$arg{tls_ctx} \|\|= { sslv2 => 0, sslv3 => 0, tlsv1 => 1, verify => 1, verify_peername => "https" };
108			}
109
110			$self->{hdl} = new AnyEvent::Handle
111			(exists $arg{fh} ? (fh => delete $arg{fh}) : (connect => delete $arg{connect})),
112			on_error => sub {
113			$self->error ($_[2]);
114			},
115			peername => delete $arg{peername},
116			;
117
118			my $secret = delete $arg{secret} ? delete $arg{secret} : AnyEvent::MP::default_secret;
119
120			# send greeting
121			my $lgreeting = "aemp;$PROTOCOL_VERSION_MAJOR;$PROTOCOL_VERSION_MINOR;AnyEvent::MP;$VERSION;"
122			. (MIME::Base64::encode_base64 AnyEvent::MP::nonce 33, "") . ";"
123			. "hmac_md6_64_256;" # hardcoded atm.
124			. "storable;" # hardcoded atm.
125			. "$self->{local_id};"
126			. (exists $arg{tls} && $arg{tls_ctx} ? "tls1.0=$arg{tls};" : "");
127
128			$self->{hdl}->push_write ("$lgreeting\012");
129
130			# expect greeting
131			$self->{hdl}->push_read (line => sub {
132			my $rgreeting = $_[1];
133
134			my ($aemp, $major, $minor, $provider, $provider_version, $nonce2, $auth, $framing, $rid, @kv) = split /;/, $rgreeting;
135
136			if ($aemp ne "aemp") {
137			return $self->error ("unparsable greeting");
138			} elsif ($major != $PROTOCOL_VERSION_MAJOR) {
139			return $self->error ("major version mismatch ($PROTOCOL_VERSION_MAJOR vs. $major)");
140			} elsif ($auth ne "hmac_md6_64_256") {
141			return $self->error ("unsupported auth method ($auth)");
142			} elsif ($framing ne "storable") {
143			return $self->error ("unsupported auth method ($auth)");
144			}
145
146			$self->{remote_id} = $rid;
147
148			$self->{greeting} = {
149			provider => $provider,
150			provider_version => $provider_version,
151			};
152
153			/^([^=]+)(?:=(.*))?/ and $self->{greeting}{$1} = $2
154			for @kv;
155
156			if (exists $self->{tls} and $self->{tls_ctx} and exists $self->{greeting}{"tls1.0"}) {
157			if ($self->{tls} ne $self->{greeting}{"tls1.0"}) {
158			return $self->error ("TLS server/client mismatch");
159			}
160			$self->{hdl}->starttls ($self->{tls}, $self->{tls_ctx});
161			}
162
163			# auth
164			require Digest::MD6;
165			require Digest::HMAC_MD6;
166
167			my $key = Digest::MD6::md6_hex ($secret);
168			my $lauth = Digest::HMAC_MD6::hmac_md6_base64 ($key, "$lgreeting$rgreeting", 64, 256);
169			my $rauth = Digest::HMAC_MD6::hmac_md6_base64 ($key, "$rgreeting$lgreeting", 64, 256);
170			$self->{hdl}->push_write ("$lauth\012");
171
172			$self->{hdl}->push_read (line => sub {
173			my ($hdl, $rauth2) = @_;
174
175			if ($rauth2 ne $rauth) {
176			return $self->error ("authentication failure/shared secret mismatch");
177			}
178
179			my $queue = delete $self->{queue}; # we are connected
180
181			$self->{on_connect}($self)
182			if $self->{on_connect};
183
184			$hdl->push_write (storable => $_)
185			for @$queue;
186
187			my $rmsg; $rmsg = sub {
188			$_[0]->push_read (storable => $rmsg);
189
190			$self->{on_recv}($self, $_[1]);
191			};
192			$hdl->push_read (storable => $rmsg);
193			});
194			});
195			}
196
197			$self
198			}
199
200			sub error {
201			my ($self, $msg) = @_;
202
203			$self->{on_error}($self, $msg);
204			$self->{hdl}->destroy;
205			}
206
207			sub send {
208			my ($self, $msg) = @_;
209
210			exists $self->{queue}
211			? push @{ $self->{queue} }, $msg
212			: $self->{hdl}->push_write (storable => $msg);
213			}
214
215			sub destroy {
216			my ($self) = @_;
217
218			$self->{hdl}->destroy;
219			}
220
221			sub DESTROY {
222			my ($self) = @_;
223
224			$self->destroy;
225			}
226
227			=back
228
229			=head1 SEE ALSO
230
231			L<AnyEvent>.
232
233			=head1 AUTHOR
234
235			Marc Lehmann <schmorp@schmorp.de>
236			http://home.schmorp.de/
237
238			=cut
239
240			1
241