rspamd/utils/cgp_rspamd.pl

#!/usr/bin/env perl

use warnings;
use strict;
use JSON::XS;
use AnyEvent;
use AnyEvent::HTTP;
use AnyEvent::IO;
use EV;
use Pod::Usage;
use Getopt::Long;
use File::stat;

my $rspamd_host     = "localhost:11333";
my $man             = 0;
my $help            = 0;
my $local           = 0;
my $header          = "X-Spam: yes";
my $max_size        = 10 * 1024 * 1024;          # 10 MB
my $request_timeout = 15;                        # 15 seconds by default
my $reject_message  = "Spam message rejected";

GetOptions(
    "host=s"           => \$rspamd_host,
    "header=s"         => \$header,
    "reject-message=s" => \$reject_message,
    "max-size=i"       => \$max_size,
    "timeout=f"        => \$request_timeout,
    "help|?"           => \$help,
    "man"              => \$man
) or pod2usage(2);

pod2usage(1) if $help;
pod2usage( -exitval => 0, -verbose => 2 ) if $man;

my $main_domain = cgp_main_domain();
my $scanned     = 0;

# Turn off bufferization as required by CGP
$| = 1;

sub cgp_main_domain {
    if ( open( my $fh, 'Settings/Main.settings' ) ) {
        while (<$fh>) {
            if (/^\s+DomainName\s+=\s+([^;]+);/) {
                return $1;
            }
        }
    }
}

sub cgp_string {
    my ($in) = @_;

    $in =~ s/\"/\\"/g;
    $in =~ s/\n/\\n/gms;
    $in =~ s/\r/\\r/mgs;
    $in =~ s/\t/  /g;

    return "\"$in\"";
}

sub rspamd_scan {
    my ( $tag, $file ) = @_;

    my $http_callback = sub {
        my ( $body, $hdr ) = @_;

        if ( $hdr && $hdr->{Status} =~ /^2/ ) {
            my $js = eval('decode_json($body)');
            $scanned++;

            if ( !$js ) {
                print "* Rspamd: Bad response for $file: invalid JSON: parse error\n";
                print "$tag FAILURE\n";
            }
            else {
                my $def     = $js;
                my $headers = "";

                if ( !$def ) {
                    print "* Rspamd: Bad response for $file: invalid JSON: default is missing\n";
                    print "$tag FAILURE\n";
                }
                else {
                    my $action = $def->{'action'};
                    my $id     = $js->{'message-id'};

                    my $symbols = "";
                    while ( my ( $k, $s ) = each( %{ $def->{'symbols'} } ) ) {
                        $symbols .= sprintf "%s(%.2f);", $k, $s->{'score'};
                    }

                    printf
                      "* Rspamd: Scanned %s; id: <%s>; Score: %.2f / %.2f; Symbols: [%s]\n",
                      $file, $id, $def->{'score'}, $def->{'required_score'}, $symbols;

                    if ( $js->{'dkim-signature'} ) {
                        $headers .= "DKIM-Signature: " . $js->{'dkim-signature'};
                    }

                    if ( $js->{'milter'} ) {
                        my $block = $js->{'milter'};

                        if ( $block->{'add_headers'} ) {
                            while ( my ( $h, $v ) = each( %{ $block->{'add_headers'} } ) ) {
                                if ( ref($v) eq 'HASH' ) {
                                    if ( $headers eq "" ) {
                                        $headers .= "$h: $v->{value}";
                                    }
                                    else {
                                        $headers .= "\\e$h: $v->{value}";
                                    }
                                }
                                else {
                                    if ( $headers eq "" ) {
                                        $headers .= "$h: $v";
                                    }
                                    else {
                                        $headers .= "\\e$h: $v";
                                    }
                                }
                            }
                        }
                    }

                    if ( $action eq 'reject' ) {
                        print "$tag DISCARD\n";
                        return;
                    }
                    elsif ( $action eq 'add header' || $action eq 'rewrite subject' ) {
                        if ( $headers eq "" ) {
                            $headers .= "$header";
                        }
                        else {
                            $headers .= "\\e$header";
                        }
                    }
                    elsif ( $action eq 'soft reject' ) {
                        print "$tag REJECTED Try again later\n";
                        return;
                    }

                    if ( $headers eq "" ) {
                        print "$tag OK\n";
                    }
                    else {
                        print "$tag ADDHEADER " . cgp_string($headers) . " OK\n";
                    }
                }
            }
        }
        else {
            if ($hdr) {
                print "* Rspamd: Bad response for $file: HTTP error: $hdr->{Status} $hdr->{Reason}\n";
            }
            else {
                print "* Rspamd: Bad response for $file: IO error: $!\n";
            }
            print "$tag FAILURE\n";
        }
    };

    if ($local) {

        # Use file scan
        # XXX: not implemented now due to CGP queue format
        http_get(
            "http://$rspamd_host/symbols?file=$file",
            timeout => $request_timeout,
            $http_callback
        );
    }
    else {
        my $sb = stat($file);

        if ( !$sb || $sb->size > $max_size ) {
            if ($sb) {
                print "* File $file is too large: " . $sb->size . "\n$tag FAILURE\n";

            }
            else {
                print "* Cannot stat $file: $!\n$tag FAILURE\n";
            }
            return;
        }
        aio_load(
            $file,
            sub {
                my ($data) = @_;

                if ( !$data ) {
                    print "* Cannot open $file: $!\n$tag FAILURE\n";
                    return;
                }

                # Parse CGP format
                $data =~ s/^((?:[^\n]*\n)*?)\n(.*)$/$2/ms;
                my @envelope = split /\n/, $1;
                chomp(@envelope);
                my $from;
                my @rcpts;
                my $ip;
                my $user;

                foreach my $elt (@envelope) {
                    if ( $elt =~ /^P\s[^<]*(<[^>]*>).*$/ ) {
                        $from = $1;
                    }
                    elsif ( $elt =~ /^R\s[^<]*(<[^>]*>).*$/ ) {
                        push @rcpts, $1;
                    }
                    elsif ( $elt =~ /^S (?:<([^>]+)> )?(?:SMTP|HTTPU?|AIRSYNC|XIMSS) \[([0-9a-f.:]+)\]/ ) {
                        if ($1) {
                            $user = $1;
                        }
                        if ($2) {
                            $ip = $2;
                        }
                    }
                    elsif ( $elt =~ /^S (?:<([^>]+)> )?(?:DSN|GROUP|LIST|PBX|PIPE|RULE) \[0\.0\.0\.0\]/ ) {
                        if ($1) {
                            $user = $1;
                        }
                        $ip = '127.2.4.7';
                    }
                }

                my $headers = {};
                if ( $file =~ /\/([^\/.]+)\.msg$/ ) {
                    $headers->{'Queue-ID'} = $1;
                }
                if ($from) {
                    $headers->{From} = $from;
                }
                if ( scalar(@rcpts) > 0 ) {

                    # XXX: Anyevent cannot parse headers with multiple values
                    $headers->{Rcpt} = join( ',', @rcpts );
                }
                if ($ip) {
                    $headers->{IP} = $ip;
                }
                if ($user) {
                    $headers->{User} = $user;
                }
                if ($main_domain) {
                    $headers->{'MTA-Tag'} = $main_domain;
                }

                http_post(
                    "http://$rspamd_host/checkv2", $data,
                    timeout => $request_timeout,
                    headers => $headers,
                    $http_callback
                );
            }
        );
    }
}

# Show informational message
print "* Rspamd CGP filter has been started\n";

my $w = AnyEvent->io(
    fh   => \*STDIN,
    poll => 'r',
    cb   => sub {
        chomp( my $input = <STDIN> );

        if ( $input =~ /^(\d+)\s+(\S+)(\s+(\S+)\s*)?$/ ) {
            my $tag = $1;
            my $cmd = $2;

            if ( $cmd eq "INTF" ) {
                print "$input\n";
            }
            elsif ( $cmd eq "FILE" && $4 ) {
                my $file = $4;
                print "* Scanning file $file\n";
                rspamd_scan $tag, $file;
            }
            elsif ( $cmd eq "QUIT" ) {
                print "* Terminating after scanning of $scanned files\n";
                print "$tag OK\n";
                exit 0;
            }
            else {
                print "* Unknown command $cmd\n";
                print "$tag FAILURE\n";
            }
        }
    }
);

EV::run;

__END__

=head1 NAME

cgp_rspamd - implements Rspamd filter for CommunigatePro MTA

=head1 SYNOPSIS

cgp_rspamd [options]

 Options:
   --host=hostport        Rspamd host to connect (localhost:11333 by default)
   --header               Add specific header for a spam message ("X-Spam: yes" by default)
   --reject-message       Rejection message for spam mail ("Spam message rejected" by default)
   --timeout              Timeout to read response from Rspamd (15 seconds by default)
   --max-size             Maximum size of message to scan (10 megabytes by default)
   --help                 brief help message
   --man                  full documentation

=head1 OPTIONS

=over 8

=item B<--host>

Specifies Rspamd host to use for scanning

=item B<--header>

Specifies the header that should be added when Rspamd action is B<add header> or B<rewrite subject>.

=item B<--reject-message>

Specifies the rejection message for spam.

=item B<--timeout>

Sets timeout in seconds for waiting Rspamd reply for a message.

=item B<--max-size>

Define the maximum messages size to be processed by Rspamd in bytes.

=item B<--help>

Print a brief help message and exits.

=item B<--man>

Prints the manual page and exits.

=back

=head1 DESCRIPTION

B<cgp_rspamd> is intended to scan messages processed with B<CommunigatePro> MTA on some Rspamd scanner. It reads
standard input and parses CGP helpers protocol.  On scan requests, this filter can query Rspamd to process a message.
B<cgp_rspamd> can tell CGP to add header or reject SPAM messages depending on Rspamd scan result.

=cut