qpsmtpd/lib/Qpsmtpd/Postfix.pm

package Qpsmtpd::Postfix;

=head1 NAME

Qpsmtpd::Postfix - postfix queueing support for qpsmtpd

=head2 DESCRIPTION

This package implements the protocol Postfix servers use to communicate
with each other. See src/global/rec_type.h in the postfix source for
details.

=cut

use strict;
use IO::Socket::UNIX;
use IO::Socket::INET;
use vars qw(@ISA);
@ISA = qw(IO::Socket::UNIX);

my %rec_types;

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

    %rec_types = (
        REC_TYPE_SIZE => 'C',    # first record, created by cleanup
        REC_TYPE_TIME => 'T',    # time stamp, required
        REC_TYPE_FULL => 'F',    # full name, optional
        REC_TYPE_INSP => 'I',    # inspector transport
        REC_TYPE_FILT => 'L',    # loop filter transport
        REC_TYPE_FROM => 'S',    # sender, required
        REC_TYPE_DONE => 'D',    # delivered recipient, optional
        REC_TYPE_RCPT => 'R',    # todo recipient, optional
        REC_TYPE_ORCP => 'O',    # original recipient, optional
        REC_TYPE_WARN => 'W',    # warning message time
        REC_TYPE_ATTR => 'A',    # named attribute for extensions

        REC_TYPE_MESG => 'M',    # start message records

        REC_TYPE_CONT => 'L',    # long data record
        REC_TYPE_NORM => 'N',    # normal data record

        REC_TYPE_XTRA => 'X',    # start extracted records

        REC_TYPE_RRTO => 'r',    # return-receipt, from headers
        REC_TYPE_ERTO => 'e',    # errors-to, from headers
        REC_TYPE_PRIO => 'P',    # priority
        REC_TYPE_VERP => 'V',    # VERP delimiters

        REC_TYPE_END => 'E',     # terminator, required

                 );

}

sub print_rec {
    my ($self, $type, @list) = @_;

    die "unknown record type" unless ($rec_types{$type});
    $self->print($rec_types{$type});

    # the length is a little endian base-128 number where each
    # byte except the last has the high bit set:
    my $s  = "@list";
    my $ln = length($s);
    while ($ln >= 0x80) {
        my $lnl = $ln & 0x7F;
        $ln >>= 7;
        $self->print(chr($lnl | 0x80));
    }
    $self->print(chr($ln));

    $self->print($s);
}

sub print_rec_size {
    my ($self, $content_size, $data_offset, $rcpt_count) = @_;

    my $s =
      sprintf("%15ld %15ld %15ld", $content_size, $data_offset, $rcpt_count);
    $self->print_rec('REC_TYPE_SIZE', $s);
}

sub print_rec_time {
    my ($self, $time) = @_;

    $time = time() unless (defined($time));

    my $s = sprintf("%d", $time);
    $self->print_rec('REC_TYPE_TIME', $s);
}

sub open_cleanup {
    my ($class, $socket) = @_;

    my $self;
    if ($socket =~ m#^(/.+)#) {
        $socket = $1;    # un-taint socket path
        $self = IO::Socket::UNIX->new(Type => SOCK_STREAM,
                                      Peer => $socket)
          if $socket;

    }
    elsif ($socket =~ /(.*):(\d+)/) {
        my ($host, $port) = ($1, $2);    # un-taint address and port
        $self = IO::Socket::INET->new(
                                      Proto    => 'tcp',
                                      PeerAddr => $host,
                                      PeerPort => $port
                                     )
          if $host and $port;
    }
    unless (ref $self) {
        warn "Couldn't open \"$socket\": $!";
        return;
    }

    # allow buffered writes
    $self->autoflush(0);
    bless($self, $class);
    $self->init();
    return $self;
}

sub print_attr {
    my ($self, @kv) = @_;
    for (@kv) {
        $self->print("$_\0");
    }
    $self->print("\0");
}

sub get_attr {
    my ($self) = @_;
    local $/ = "\0";
    my %kv;
    for (; ;) {
        my $k = $self->getline;
        chomp($k);
        last unless ($k);
        my $v = $self->getline;
        chomp($v);
        $kv{$k} = $v;
    }
    return %kv;
}

=head2 print_msg_line($line)

print one line of a message to cleanup.

This removes any linefeed characters from the end of the line
and splits the line across several records if it is longer than
1024 chars. 

=cut

sub print_msg_line {
    my ($self, $line) = @_;

    $line =~ s/\r?\n$//s;

    # split into 1k chunks.
    while (length($line) > 1024) {
        my $s = substr($line, 0, 1024);
        $line = substr($line, 1024);
        $self->print_rec('REC_TYPE_CONT', $s);
    }
    $self->print_rec('REC_TYPE_NORM', $line);
}

=head2 inject_mail($transaction)

(class method) inject mail in $transaction into postfix queue via cleanup.
$transaction is supposed to be a Qpsmtpd::Transaction object.

=cut

sub inject_mail {
    my ($class, $transaction) = @_;

    my @sockets = @{$transaction->notes('postfix-queue-sockets')
          // ['/var/spool/postfix/public/cleanup']};
    my $strm;
    $strm = $class->open_cleanup($_) and last for @sockets;
    die "Unable to open any cleanup sockets!" unless $strm;

    my %at  = $strm->get_attr;
    my $qid = $at{queue_id};
    print STDERR "qid=$qid\n";
    $strm->print_attr('flags' => $transaction->notes('postfix-queue-flags'));
    $strm->print_rec_time();
    $strm->print_rec('REC_TYPE_FROM', $transaction->sender->address || "");
    for (map { $_->address } $transaction->recipients) {
        $strm->print_rec('REC_TYPE_RCPT', $_);
    }

    # add an empty message length record.
    # cleanup is supposed to understand that.
    # see src/pickup/pickup.c
    $strm->print_rec('REC_TYPE_MESG', "");

    # a received header has already been added in SMTP.pm
    # so we can just copy the message:

    my $hdr = $transaction->header->as_string;
    for (split(/\r?\n/, $hdr)) {
        print STDERR "hdr: $_\n";
        $strm->print_msg_line($_);
    }
    $transaction->body_resetpos;
    while (my $line = $transaction->body_getline) {

        # print STDERR "body: $line\n";
        $strm->print_msg_line($line);
    }

    # finish it.
    $strm->print_rec('REC_TYPE_XTRA', "");
    $strm->print_rec('REC_TYPE_END',  "");
    $strm->flush();
    %at = $strm->get_attr;
    my $status = $at{status};
    my $reason = $at{reason};
    $strm->close();
    return wantarray ? ($status, $qid, $reason || "") : $status;
}

1;

# vim:sw=2