X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=dbcd8097bafce544231a37f7e58565f55999e3c3;hb=42846321acd959aa93c2c2afd2b1f0a67f8accaf;hp=75a0c452c0e9034d830afc042820dc02be28b801;hpb=732e58eaf2e7ef261102aed4f03185fcc775668f;p=spider.git diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm index 75a0c452..dbcd8097 100644 --- a/perl/DXMsg.pm +++ b/perl/DXMsg.pm @@ -26,6 +26,10 @@ use DXLog; use IO::File; use Fcntl; +eval { + require Net::SMTP; +}; + use strict; use vars qw($VERSION $BRANCH); @@ -36,6 +40,7 @@ $main::branch += $BRANCH; use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean $residencetime @badmsg @swop $swopfn $badmsgfn $forwardfn @forward $timeout $waittime + $email_server $email_prog $email_from $queueinterval $lastq $importfn $minchunk $maxchunk $bulltopriv); %work = (); # outstanding jobs @@ -56,7 +61,9 @@ $minchunk = 4800; # minimum chunk size for a split message $maxchunk = 6000; # maximum chunk size $bulltopriv = 1; # convert msgs with callsigns to private if they are bulls $residencetime = 2*86400; # keep deleted messages for this amount of time - +$email_server = undef; # DNS address of smtp server if 'smtp' +$email_prog = undef; # program name + args for sending mail +$email_from = undef; # the from address the email will appear to be from $badmsgfn = "$msgdir/badmsg.pl"; # list of TO address we wont store $forwardfn = "$msgdir/forward.pl"; # the forwarding table @@ -91,6 +98,14 @@ $importfn = "$msgdir/import"; # import directory deletetime => '5,Deletion Time,cldatetime', ); +# fix up the default sendmail if available +for (qw(/usr/sbin/sendmail /usr/lib/sendmail /usr/sbin/sendmail)) { + if (-e $_) { + $email_prog = $_; + last; + } +} + # allocate a new object # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper sub alloc @@ -102,7 +117,6 @@ sub alloc # $to =~ s/-\d+$//o; $self->{to} = ($to eq $main::mycall) ? $main::myalias : $to; my $from = shift; - $from =~ s/-\d+$//o; $self->{from} = uc $from; $self->{t} = shift; $self->{private} = shift; @@ -110,6 +124,8 @@ sub alloc $self->{origin} = shift; $self->{'read'} = shift; $self->{rrreq} = shift; + $self->{delete} = shift; + $self->{deletetime} = shift; $self->{gotit} = []; # $self->{lastt} = $main::systime; $self->{lines} = []; @@ -280,7 +296,7 @@ sub process # does an identical message already exist? my $m; for $m (@msg) { - if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) { + if (substr($ref->{subject},0,28) eq substr($m->{subject},0,28) && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from} && $ref->{to} eq $m->{to}) { $ref->stop_msg($fromnode); my $msgno = $m->{msgno}; dbg("duplicate message from $ref->{from} -> $ref->{to} to msg: $msgno") if isdbg('msg'); @@ -307,8 +323,8 @@ sub process } push @words, BadWords::check($ref->{subject}); if (@words) { - dbg("message with badwords '@words' $ref->{from} -> $ref->{to} '$ref->{subject}' origin: $ref->{origin} via " . $self->call) if isdbg('msg'); - Log('msg',"message with badwords '@words' $ref->{from} -> $ref->{to} origin: $ref->{origin} via " . $self->call); + dbg("$ref->{from} swore: '@words' -> $ref->{to} '$ref->{subject}' origin: $ref->{origin} via " . $self->call) if isdbg('msg'); + Log('msg',"$ref->{from} swore: '@words' -> $ref->{to} origin: $ref->{origin} via " . $self->call); Log('msg',"subject: $ref->{subject}"); for (@{$ref->{lines}}) { Log('msg', "line: $_"); @@ -320,9 +336,8 @@ sub process $ref->{msgno} = next_transno("Msgno"); push @{$ref->{gotit}}, $fromnode; # mark this up as being received $ref->store($ref->{lines}); + $ref->notify; add_dir($ref); - my $dxchan = DXChannel->get($ref->{to}); - $dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user; Log('msg', "Message $ref->{msgno} from $ref->{from} received from $fromnode for $ref->{to}"); } } @@ -340,8 +355,7 @@ sub process if ($ref) { if ($ref->{private}) { # remove it if it private and gone off site# Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode and deleted"); - $ref->{delete}++; - $ref->{deletetime} = $main::systime + $residencetime; + $ref->mark_delete; } else { Log('msg', "Message $ref->{msgno} from $ref->{from} sent to $fromnode"); push @{$ref->{gotit}}, $fromnode; # mark this up as being received @@ -407,8 +421,7 @@ sub process if ($pcno == 49) { # global delete on subject for (@msg) { if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) { - $_->{delete}++; - $_->{deletetime} = $main::systime + $residencetime; + $_->mark_delete; Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted"); DXChannel::broadcast_nodes($line, $self); } @@ -418,6 +431,51 @@ sub process } +sub notify +{ + my $ref = shift; + my $to = $ref->{to}; + my $uref = DXUser->get($to); + my $dxchan = DXChannel->get($to); + if (((*Net::SMTP && $email_server) || $email_prog) && $uref && $uref->wantemail) { + my $email = $uref->email; + if ($email) { + my @rcpt = ref $email ? @{$email} : $email; + my $fromaddr = $email_from || $main::myemail; + my @headers = ("To: $ref->{to}", + "From: $fromaddr", + "Subject: [DXSpider: $ref->{from}] $ref->{subject}", + "X-DXSpider-To: $ref->{to}", + "X-DXSpider-From: $ref->{from}\@$ref->{origin}", + "X-DXSpider-Gateway: $main::mycall" + ); + my @data = ("Msgno: $ref->{msgno} To: $to From: $ref->{from}\@$ref->{origin} Gateway: $main::mycall", + "", + $ref->read_msg_body + ); + my $msg; + undef $!; + if (*Net::SMTP && $email_server) { + $msg = Net::SMTP->new($email_server); + if ($msg) { + $msg->mail($fromaddr); + $msg->to(@rcpt); + $msg->data(map {"$_\n"} @headers, '', @data); + $msg->quit; + } + } elsif ($email_prog) { + $msg = new IO::File "|$email_prog " . join(' ', @rcpt); + if ($msg) { + print $msg map {"$_\r\n"} @headers, '', @data, '.'; + $msg->close; + } + } + dbg("email forwarding error $!") if isdbg('msg') && !$msg && defined $!; + } + } + $dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user; +} + # store a message away on disc or whatever # # NOTE the second arg is a REFERENCE not a list @@ -453,7 +511,9 @@ sub store if (defined $fh) { my $rr = $ref->{rrreq} ? '1' : '0'; my $priv = $ref->{private} ? '1': '0'; - print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr\n"; + my $del = $ref->{delete} ? '1' : '0'; + my $delt = $ref->{deletetime} || '0'; + print $fh "=== $ref->{msgno}^$ref->{to}^$ref->{from}^$ref->{t}^$priv^$ref->{subject}^$ref->{origin}^$ref->{'read'}^$rr^$del^$delt\n"; print $fh "=== ", join('^', @{$ref->{gotit}}), "\n"; my $line; $ref->{size} = 0; @@ -468,6 +528,14 @@ sub store confess "can't open msg file $fn $!"; } } + + # actual remove all the 'deleted' messages in one hit. + # this has to me delayed until here otherwise it only does one at + # a time because @msg is rewritten everytime del_msg is called. + my @del = grep {!$_->{tonode} && $_->{delete} && $_->{deletetime} < $main::systime} @msg; + for (@del) { + $_->del_msg; + } } # delete a message @@ -475,26 +543,44 @@ sub del_msg { my $self = shift; my $dxchan = shift; + my $call = ''; + $call = ' by ' . $dxchan->call if $dxchan; if ($self->{tonode}) { $self->{delete}++; $self->{deletetime} = 0; + dbg("Msgno $self->{msgno} but marked as expunged$call") if isdbg('msg'); } else { - my $call; - if ($dxchan) { - $call = " by " . $dxchan->call; - } else { - $call = ''; - } - # remove it from the active message list @msg = grep { $_ != $self } @msg; + + Log('msg', "Msgno $self->{msgno} expunged$call"); + dbg("Msgno $self->{msgno} expunged$call") if isdbg('msg'); # remove the file unlink filename($self->{msgno}); } } +sub mark_delete +{ + my $ref = shift; + my $t = shift; + $t = $main::systime + $residencetime unless defined $t; + + $ref->{delete}++; + $ref->{deletetime} = $t; + $ref->store( [$ref->read_msg_body] ); +} + +sub unmark_delete +{ + my $ref = shift; + my $t = shift; + delete $ref->{delete}; + delete $ref->{deletetime}; +} + # clean out old messages from the message queue sub clean_old { @@ -503,7 +589,9 @@ sub clean_old # mark old messages for deletion foreach $ref (@msg) { if (ref($ref) && !$ref->{keep} && $ref->{t} < $main::systime - $maxage) { - $ref->{delete} = 1; + + # this is for IMMEDIATE destruction + $ref->{delete}++; $ref->{deletetime} = 0; } } @@ -644,10 +732,7 @@ sub queue_msg next if $ref->{tonode}; # ignore it if it already being processed # is it awaiting deletion? - if ($ref->{delete} && $main::systime > $ref->{deletetime}) { - $ref->del_msg; - next; - } + next if $ref->{delete}; # firstly, is it private and unread? if so can I find the recipient # in my cluster node list offsite? @@ -694,6 +779,8 @@ sub queue_msg # if all the available nodes are busy then stop last if @nodelist == scalar grep { get_busy($_->call) } @nodelist; } + + } # is there a message for me? @@ -946,7 +1033,7 @@ sub do_send_stuff if (my @ans = BadWords::check($line)) { $self->{badcount} += @ans; Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} in msg"); - return ($self->msg('e17', @ans), $self->msg('m1')); + $loc->{reject}++; } $loc->{subject} = $line; $loc->{lines} = []; @@ -958,33 +1045,31 @@ sub do_send_stuff my $loc = $self->{loc}; if ($line eq "\032" || $line eq '%1A' || uc $line eq "/EX") { my $to; - - foreach $to (@{$loc->{to}}) { - my $ref; - my $systime = $main::systime; - my $mycall = $main::mycall; - $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'), - uc $to, - exists $loc->{from} ? $loc->{from} : $self->call, - $systime, - $loc->{private}, - $loc->{subject}, - exists $loc->{origin} ? $loc->{origin} : $mycall, - '0', - $loc->{rrreq}); - $ref->swop_it($self->call); - $ref->store($loc->{lines}); - $ref->add_dir(); - push @out, $self->msg('m11', $ref->{msgno}, $to); - #push @out, "msgno $ref->{msgno} sent to $to"; - my $dxchan = DXChannel->get(uc $to); - if ($dxchan) { - if ($dxchan->is_user()) { - $dxchan->send($dxchan->msg('m9')); - } + unless ($loc->{reject}) { + foreach $to (@{$loc->{to}}) { + my $ref; + my $systime = $main::systime; + my $mycall = $main::mycall; + $ref = DXMsg->alloc(DXMsg::next_transno('Msgno'), + uc $to, + exists $loc->{from} ? $loc->{from} : $self->call, + $systime, + $loc->{private}, + $loc->{subject}, + exists $loc->{origin} ? $loc->{origin} : $mycall, + '0', + $loc->{rrreq}); + $ref->swop_it($self->call); + $ref->store($loc->{lines}); + $ref->add_dir(); + push @out, $self->msg('m11', $ref->{msgno}, $to); + #push @out, "msgno $ref->{msgno} sent to $to"; + $ref->notify; } + } else { + Log('msg', $self->call . " swore to @{$loc->{to}} subject: '$loc->{subject}' in msg, REJECTED"); } - + delete $loc->{lines}; delete $loc->{to}; delete $self->{loc}; @@ -1002,9 +1087,9 @@ sub do_send_stuff } else { if (my @ans = BadWords::check($line)) { $self->{badcount} += @ans; - Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} subject: '$loc->{subject}' in msg"); + Log('msg', $self->call . " used badwords: @ans to @{$loc->{to}} subject: '$loc->{subject}' in msg") unless $loc->{reject}; Log('msg', "line: $line"); - return ($self->msg('e17', @ans)); + $loc->{reject}++; } # i.e. it ain't and end or abort, therefore store the line @@ -1018,11 +1103,14 @@ sub do_send_stuff sub dir { my $ref = shift; - my $flag = $ref->read ? '-' : ' '; - $flag = 'D' if $ref->delete; - return sprintf "%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s", - $ref->msgno, $flag, $ref->private ? 'p' : ' ', $ref->size, - $ref->to, $ref->from, cldate($ref->t), ztime($ref->t), $ref->subject; + my $flag = $ref->{private} && $ref->{read} ? '-' : ' '; + if ($ref->{delete}) { + $flag = $ref->{deletetime} > $main::systime ? 'D' : 'E'; + } + return sprintf("%6d%s%s%5d %8.8s %8.8s %-6.6s %5.5s %-30.30s", + $ref->{msgno}, $flag, $ref->{private} ? 'p' : ' ', + $ref->{size}, $ref->{to}, $ref->{from}, cldate($ref->{t}), + ztime($ref->{t}), $ref->{subject}); } # load the forward table @@ -1094,6 +1182,32 @@ sub forward_it return 0; } +# +# look down the forward table to see whether this is a valid bull +# or not (ie it will forward somewhere even if it is only here) +# +sub valid_bull_addr +{ + my $call = shift; + my $i; + + unless (@forward) { + return 1 if $call =~ /^ALL/; + return 1 if $call =~ /^DX/; + return 0; + } + + for ($i = 0; $i < @forward; $i += 5) { + my ($sort, $field, $pattern, $action, $bbs) = @forward[$i..($i+4)]; + if ($field eq 'T') { + if (!$pattern || $call =~ m{$pattern}i) { + return 1; + } + } + } + return 0; +} + sub dump_it { my $ref = shift; @@ -1202,7 +1316,6 @@ sub import_msgs my @out = import_one($main::me, \@msg, $splitit); Log('msg', @out); } - queue_msg(0); } # import one message as a list in bbs (as extended) mode @@ -1222,7 +1335,7 @@ sub import_one # first line; my $line = shift @$ref; - my @f = split /\b/, $line; + my @f = split /([\s\@\$])/, $line; @f = map {s/\s+//g; length $_ ? $_ : ()} @f; unless (@f && $f[0] =~ /^(:?S|SP|SB|SEND)$/ ) { @@ -1310,7 +1423,18 @@ sub import_one } else { push @chunk, $ref; } - + + # does an identical message already exist? + my $m; + for $m (@msg) { + if (substr($subject,0,28) eq substr($m->{subject},0,28) && $from eq $m->{from} && grep $m->{to} eq $_, @to) { + my $msgno = $m->{msgno}; + dbg("duplicate message from $from -> $m->{to} to msg: $msgno") if isdbg('msg'); + Log('msg', "duplicate message from $from -> $m->{to} to msg: $msgno"); + return; + } + } + # write all the messages away my $i; for ( $i = 0; $i < @chunk; $i++) { @@ -1340,12 +1464,7 @@ sub import_one $mref->add_dir(); push @out, $dxchan->msg('m11', $mref->{msgno}, $to); #push @out, "msgno $ref->{msgno} sent to $to"; - my $todxchan = DXChannel->get(uc $to); - if ($todxchan) { - if ($todxchan->is_user()) { - $todxchan->send($todxchan->msg('m9')); - } - } + $mref->notify; } } return @out;