at least the monitoring now works better
[spider.git] / perl / AGWMsg.pm
1 #
2 # This class is the internal subclass that deals with AGW Engine connections
3 #
4 # The complication here is that there only one 'real' (and from the node's point
5 # of view, invisible) IP connection. This connection then has multiplexed 
6 # connections passed down it, a la BPQ native host ports (but not as nicely).
7 #
8 # It is a shame that the author has chosen an inherently dangerous binary format
9 # which is non-framed and has the potential for getting out of sync and not
10 # being able to recover. Relying on length fields is recipe for disaster (esp.
11 # for him!). DoS attacks are a wonderful thing....
12 #
13 # Also making the user handle the distinction between a level 2 and 4 connection
14 # and especially Digis, in the way that he has, is a bit of a cop out! If I can
15 # be arsed to do anything other than straight ax25 connects then it will only
16 # because I have the 'power of perl' available that avoids me getting 
17 # terminally bored sorting out other people's sloppyness.
18 #
19 # $Id$
20 #
21 # Copyright (c) 2001 - Dirk Koopman G1TLH
22 #
23
24 package AGWMsg;
25
26 use strict;
27 use IO::Socket;
28 use Msg;
29 use AGWConnect;
30 use DXDebug;
31
32 use vars qw(@ISA $sock @outqueue $send_offset $inmsg $rproc $noports $lasttime);
33
34 @ISA = qw(Msg ExtMsg);
35 $sock = undef;
36 @outqueue = ();
37 $send_offset = 0;
38 $inmsg = '';
39 $rproc = undef;
40 $noports = 0;
41 $lasttime = time;
42
43 sub init
44 {
45         return unless $enable;
46         $rproc = shift;
47         
48         finish();
49         dbg('err', "AGW initialising and connecting to $addr/$port ...");
50         $sock = IO::Socket::INET->new(PeerAddr => $addr, PeerPort => $port, Proto=>'tcp', Timeout => 30);
51         unless ($sock) {
52                 dbg('err', "Cannot connect to AGW Engine at $addr/$port $!");
53                 return;
54         }
55         Msg::blocking($sock, 0);
56         Msg::set_event_handler($sock, read=>\&_rcv, error=>\&_error);
57         
58         # send a P frame for the login if required
59         if ($login) {
60                 my $data = pack "a255 a255", $login, $passwd;
61                 _sendf('P', undef, undef, undef, undef, $data);
62         }
63
64         # send:
65         # R frame for the release number
66         # G frame to ask for ports
67         # X frame to say who we are
68         # optional m frame to enable monitoring
69         _sendf('R');
70         _sendf('G');
71         _sendf('X', $main::mycall);
72         _sendf('m') if $monitor;
73 }
74
75 sub finish
76 {
77         if ($sock) {
78                 dbg('err', "AGW ending...");
79                 for (values %Msg::conns) {
80                         next unless $_->isa('AGWMsg');
81                         $_->disconnect;
82                 }
83                 # say we are going
84                 _sendf('m') if $monitor;
85                 _sendf('x', $main::mycall);
86                 Msg->sleep(2);
87                 Msg::set_event_handler($sock, read=>undef, write=>undef, error=>undef);
88                 $sock->close;
89         }
90 }
91
92 sub _sendf
93 {
94         my $sort = shift || confess "need a valid AGW command letter";
95         my $from = shift || '';
96         my $to   = shift || '';
97         my $port = shift || 0;
98         my $pid  = shift || 0;
99         my $data = shift || '';
100         my $len  = 0;
101         
102         $len = length $data; 
103         dbg('agw', "AGW sendf: $sort '${from}'->'${to}' port: $port pid: $pid \"$data\"");
104         push @outqueue, pack('C x3 a1 x1 C x1 a10 a10 V x4 a*', $port, $sort, $pid, $from, $to, $len, $data);
105         Msg::set_event_handler($sock, write=>\&_send);
106 }
107
108 sub _send 
109 {
110     return unless $sock;
111
112     # If $flush is set, set the socket to blocking, and send all
113     # messages in the queue - return only if there's an error
114     # If $flush is 0 (deferred mode) make the socket non-blocking, and
115     # return to the event loop only after every message, or if it
116     # is likely to block in the middle of a message.
117
118     my $offset = $send_offset;
119
120     while (@outqueue) {
121         my $msg            = $outqueue[0];
122                 my $mlth           = length($msg);
123         my $bytes_to_write = $mlth - $offset;
124         my $bytes_written  = 0;
125                 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
126         while ($bytes_to_write > 0) {
127             $bytes_written = syswrite ($sock, $msg,
128                                        $bytes_to_write, $offset);
129             if (!defined($bytes_written)) {
130                 if (Msg::_err_will_block($!)) {
131                     # Should happen only in deferred mode. Record how
132                     # much we have already sent.
133                     $send_offset = $offset;
134                     # Event handler should already be set, so we will
135                     # be called back eventually, and will resume sending
136                     return 1;
137                 } else {    # Uh, oh
138                                         _error();
139                     return 0; # fail. Message remains in queue ..
140                 }
141             }
142             $offset         += $bytes_written;
143             $bytes_to_write -= $bytes_written;
144         }
145         $send_offset = $offset = 0;
146         shift @outqueue;
147         last;  # Go back to select and wait
148                        # for it to fire again.
149     }
150
151     # Call me back if queue has not been drained.
152     if (@outqueue) {
153         Msg::set_event_handler ($sock, write => \&_send);
154     } else {
155         Msg::set_event_handler ($sock, write => undef);
156     }
157     1;  # Success
158 }
159
160 sub _rcv {                     # Complement to _send
161     return unless $sock;
162     my ($msg, $offset, $bytes_read);
163
164         $bytes_read = sysread ($sock, $msg, 1024, 0);
165         if (defined ($bytes_read)) {
166                 if ($bytes_read > 0) {
167                         $inmsg .= $msg;
168                 } 
169         } else {
170                 if (Msg::_err_will_block($!)) {
171                         return; 
172                 } else {
173                         $bytes_read = 0;
174                 }
175     }
176
177 FINISH:
178     if (defined $bytes_read && $bytes_read == 0) {
179                 finish();
180     } else {
181                 _decode() if length $inmsg > 36;
182         }
183 }
184
185 sub _error
186 {
187         dbg('agw', "error on AGW connection $addr/$port $!");
188         Msg::set_event_handler($sock, read=>undef, write=>undef, error=>undef);
189         $sock = undef;
190         for (values %Msg::conns) {
191                 next unless $_->isa('AGWMsg');
192                 $_->disconnect;
193         }
194 }
195
196 sub _decode
197 {
198         return unless $sock;
199
200         # we have at least 36 bytes of data (ugh!)
201         my ($port, $sort, $pid, $from, $to, $len) = unpack('C x3 a1 x1 C x1 Z10 Z10 V x4', $inmsg);
202         my $data;
203
204         # do a sanity check on the length
205         if ($len > 2000) {
206                 dbg('err', "AGW: invalid length $len > 2000 received ($sort $port $pid '$from'->'$to')");
207                 finish();
208                 return;
209         }
210         if ($len == 0){
211                 if (length $inmsg > 36) {
212                         $inmsg = substr($inmsg, 36);
213                 } else {
214                         $inmsg = '';
215                 }
216         } elsif (length $inmsg > $len + 36) {
217                 $data = substr($inmsg, 36, $len);
218                 $inmsg = substr($inmsg, $len + 36);
219         } elsif (length $inmsg == $len + 36) {
220                 $data = substr($inmsg, 36);
221                 $inmsg = '';
222         } else {
223                 # we don't have enough data or something
224                 # or we have screwed up
225                 return;
226         }
227
228         $data = '' unless defined $data;
229         if ($sort eq 'D') {
230                 my $d = unpack "Z*", $data;
231                 $d =~ s/\cM$//;
232                 dbg('agw', "AGW Data In port: $port pid: $pid '$from'->'$to' length: $len \"$d\"");
233                 my $conn = Msg->conns($from eq $main::mycall ? $to : $from);
234                 if ($conn) {
235                         if ($conn->{state} eq 'WC') {
236                                 if (exists $conn->{cmd}) {
237                                         if (@{$conn->{cmd}}) {
238                                                 dbg('connect', $d);
239                                                 $conn->_docmd($d);
240                                         }
241                                 }
242                                 if ($conn->{state} eq 'WC' && exists $conn->{cmd} && @{$conn->{cmd}} == 0) {
243                                         $conn->to_connected($conn->{call}, 'O', $conn->{csort});
244                                 }
245                         } else {
246                                 &{$conn->{rproc}}($conn, "I$conn->{call}|$data");
247                         }
248                 } else {
249                         dbg('err', "AGW error Unsolicited Data!");
250                 }
251         } elsif ($sort eq 'I' || $sort eq 'S' || $sort eq 'U' || $sort eq 'M' || $sort eq 'T') {
252                 my $d = unpack "Z*", $data;
253                 $d =~ s/\cM$//;
254                 dbg('agw', "AGW Monitor port: $port \"$d\"");
255         } elsif ($sort eq 'C') {
256                 my $d = unpack "Z*", $data;
257                 $d =~ s/\cM$//;
258                 dbg('agw', "AGW Connect port: $port pid: $pid '$from'->'$to' \"$d\"");
259                 my $call = $from eq $main::mycall ? $to : $from;
260                 my $conn = Msg->conns($call);
261                 if ($conn) {
262                         if ($conn->{state} eq 'WC') {
263                                 if (exists $conn->{cmd} && @{$conn->{cmd}}) {
264                                         $conn->_docmd($d);
265                                         if ($conn->{state} eq 'WC' && exists $conn->{cmd} &&  @{$conn->{cmd}} == 0) {
266                                                 $conn->to_connected($conn->{call}, 'O', $conn->{csort});
267                                         }
268                                 }
269                         }
270                 } else {
271                         $conn = AGWMsg->new($rproc);
272                         $conn->{agwpid} = $pid;
273                         $conn->{agwport} = $port;
274                         $conn->{lineend} = "\cM";
275                         $conn->{incoming} = 1;
276                         $conn->{agwcall} = $call;
277                         $conn->to_connected($call, 'A', $conn->{csort} = 'ax25');
278                 }
279         } elsif ($sort eq 'd') {
280                 dbg('agw', "AGW '$from'->'$to' port: $port Disconnected");
281                 my $conn = Msg->conns($from eq $main::mycall ? $to : $from);
282                 $conn->in_disconnect if $conn;
283         } elsif ($sort eq 'y') {
284                 my ($frames) = unpack "V", $data;
285                 dbg('agw', "AGW Frames Outstanding on port $port = $frames");
286                 my $conn = Msg->conns($from);
287                 $conn->{oframes} = $frames if $conn;
288         } elsif ($sort eq 'Y') {
289                 my ($frames) = unpack "V", $data;
290                 dbg('agw', "AGW Frames Outstanding on circuit '$from'->'$to' = $frames");
291                 my $conn = Msg->conns($from eq $main::mycall ? $to : $from);
292                 $conn->{oframes} = $frames if $conn;
293         } elsif ($sort eq 'X') {
294                 my ($r) = unpack "C", $data;
295                 $r = $r ? "Successful" : "Failed";
296                 dbg('err', "AGW Register $from $r");
297                 finish() unless $r;
298         } elsif ($sort eq 'R') {
299                 my ($major, $minor) = unpack "v x2 v x2", $data;
300                 dbg('agw', "AGW Version $major.$minor");
301         } elsif ($sort eq 'G') {
302                 my @ports = split /;/, $data;
303             $noports = shift @ports || '0';
304                 dbg('agw', "AGW $noports Ports available");
305                 pop @ports while @ports > $noports;
306                 for (@ports) {
307                         next unless $_;
308                         dbg('agw', "AGW Port: $_");
309                 }
310                 for (my $i = 0; $i < $noports; $i++) {
311                         _sendf('y', undef, undef, $i );
312                 }
313         } else {
314                 my $d = unpack "Z*", $data;
315                 dbg('agw', "AGW decode $sort port: $port pid: $pid '$from'->'$to' length: $len \"$d\"");
316         }
317 }
318
319 sub connect
320 {
321         my ($conn, $line) = @_;
322
323         my ($port, $call) = split /\s+/, $line;
324         $conn->{agwpid} = ord "\xF0";
325         $conn->{agwport} = $port - 1;
326         $conn->{lineend} = "\cM";
327         $conn->{incoming} = 0;
328         $conn->{csort} = 'ax25';
329         $conn->{agwcall} = uc $call;
330         
331         _sendf('C', $main::mycall, $conn->{agwcall}, $conn->{agwport}, $conn->{agwpid});
332 }
333
334 sub in_disconnect
335 {
336         my $conn = shift;
337         $conn->SUPER::disconnect;
338 }
339
340 sub disconnect
341 {
342         my $conn = shift;
343         _sendf('d', $main::mycall, $conn->{agwcall}, $conn->{agwport}, $conn->{agwpid});
344         $conn->SUPER::disconnect;
345 }
346
347 sub enqueue
348 {
349         my ($conn, $msg) = @_;
350         if ($msg =~ /^[D]/) {
351                 $msg =~ s/^[-\w]+\|//;
352                 _sendf('Y', $main::mycall, $conn->{call}, $conn->{agwport}, $conn->{agwpid});
353                 _sendf('D', $main::mycall, $conn->{call}, $conn->{agwport}, $conn->{agwpid}, $msg . $conn->{lineend});
354         }
355 }
356
357 sub process
358 {
359         return unless $sock;
360         if ($main::systime - $lasttime >= 60) {
361                 for (my $i = 0; $i < $noports; $i++) {
362                         _sendf('y', undef, undef, $i );
363 #                       _sendf('H', undef, undef, $i );
364                 }
365                 $lasttime = $main::systime;
366         }
367 }
368 1;
369