lib/PublicInbox/POP3.pm

   1 # Copyright (C) all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3 #
   4 # Each instance of this represents a POP3 client connected to
   5 # public-inbox-{netd,pop3d}.  Much of this was taken from IMAP.pm and NNTP.pm
   6 #
   7 # POP3 is one mailbox per-user, so the "USER" command is like the
   8 # format of -imapd and is mapped to $NEWSGROUP.$SLICE (large inboxes
   9 # are sliced into 50K mailboxes in both POP3 and IMAP to avoid overloading
  10 # clients)
  11 #
  12 # Unlike IMAP, the "$NEWSGROUP" mailbox (without $SLICE) is a rolling
  13 # window of the latest messages.  We can do this for POP3 since the
  14 # typical POP3 session is short-lived while long-lived IMAP sessions
  15 # would cause slices to grow on the server side without bounds.
  16 #
  17 # Like IMAP, POP3 also has per-session message sequence numbers (MSN),
  18 # which require mapping to UIDs.  The offset of an entry into our
  19 # per-client cache is: (MSN-1)
  20 #
  21 # fields:
  22 # - uuid - 16-byte (binary) UUID representation (before successful login)
  23 # - cache - one-dimentional arrayref of (UID, bytesize, oidhex)
  24 # - nr_dele - number of deleted messages
  25 # - expire - string of packed unsigned short offsets
  26 # - user_id - user-ID mapped to UUID (on successful login + lock)
  27 # - txn_max_uid - for storing max deleted UID persistently
  28 # - ibx - PublicInbox::Inbox object
  29 # - slice - unsigned integer slice number (0..Inf), -1 => latest
  30 # - salt - pre-auth for APOP
  31 # - uid_dele - maximum deleted from previous session at login (NNTP ARTICLE)
  32 # - uid_base - base UID for mailbox slice (0-based) (same as IMAP)
  33 package PublicInbox::POP3;
  34 use v5.12;
  35 use parent qw(PublicInbox::DS);
  36 use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT);
  37 use PublicInbox::GitAsyncCat;
  38 use PublicInbox::DS qw(now);
  39 use Errno qw(EAGAIN);
  40 use Digest::MD5 qw(md5);
  41 use PublicInbox::IMAP; # for UID slice stuff
  42
  43 use constant {
  44         LINE_MAX => 512, # XXX unsure
  45 };
  46
  47 # XXX FIXME: duplicated stuff from NNTP.pm and IMAP.pm
  48
  49 sub err ($$;@) {
  50         my ($self, $fmt, @args) = @_;
  51         printf { $self->{pop3d}->{err} } $fmt."\n", @args;
  52 }
  53
  54 sub out ($$;@) {
  55         my ($self, $fmt, @args) = @_;
  56         printf { $self->{pop3d}->{out} } $fmt."\n", @args;
  57 }
  58
  59 sub zflush {} # noop
  60
  61 sub requeue_once ($) {
  62         my ($self) = @_;
  63         # COMPRESS users all share the same DEFLATE context.
  64         # Flush it here to ensure clients don't see
  65         # each other's data
  66         $self->zflush;
  67
  68         # no recursion, schedule another call ASAP,
  69         # but only after all pending writes are done.
  70         # autovivify wbuf:
  71         my $new_size = push(@{$self->{wbuf}}, \&long_step);
  72
  73         # wbuf may be populated by $cb, no need to rearm if so:
  74         $self->requeue if $new_size == 1;
  75 }
  76
  77 sub long_step {
  78         my ($self) = @_;
  79         # wbuf is unset or empty, here; {long} may add to it
  80         my ($fd, $cb, $t0, @args) = @{$self->{long_cb}};
  81         my $more = eval { $cb->($self, @args) };
  82         if ($@ || !$self->{sock}) { # something bad happened...
  83                 delete $self->{long_cb};
  84                 my $elapsed = now() - $t0;
  85                 if ($@) {
  86                         err($self,
  87                             "%s during long response[$fd] - %0.6f",
  88                             $@, $elapsed);
  89                 }
  90                 out($self, " deferred[$fd] aborted - %0.6f", $elapsed);
  91                 $self->close;
  92         } elsif ($more) { # $self->{wbuf}:
  93                 # control passed to ibx_async_cat if $more == \undef
  94                 requeue_once($self) if !ref($more);
  95         } else { # all done!
  96                 delete $self->{long_cb};
  97                 my $elapsed = now() - $t0;
  98                 my $fd = fileno($self->{sock});
  99                 out($self, " deferred[$fd] done - %0.6f", $elapsed);
 100                 my $wbuf = $self->{wbuf}; # do NOT autovivify
 101                 $self->requeue unless $wbuf && @$wbuf;
 102         }
 103 }
 104
 105 sub long_response ($$;@) {
 106         my ($self, $cb, @args) = @_; # cb returns true if more, false if done
 107         my $sock = $self->{sock} or return;
 108         # make sure we disable reading during a long response,
 109         # clients should not be sending us stuff and making us do more
 110         # work while we are stream a response to them
 111         $self->{long_cb} = [ fileno($sock), $cb, now(), @args ];
 112         long_step($self); # kick off!
 113         undef;
 114 }
 115
 116 sub _greet ($) {
 117         my ($self) = @_;
 118         my $s = $self->{salt} = sprintf('%x.%x', int(rand(0x7fffffff)), time);
 119         $self->write("+OK POP3 server ready <$s\@public-inbox>\r\n");
 120 }
 121
 122 sub new ($$$) {
 123         my ($class, $sock, $pop3d) = @_;
 124         my $self = bless { pop3d => $pop3d }, __PACKAGE__;
 125         my $ev = EPOLLIN;
 126         my $wbuf;
 127         if ($sock->can('accept_SSL') && !$sock->accept_SSL) {
 128                 return CORE::close($sock) if $! != EAGAIN;
 129                 $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock);
 130                 $wbuf = [ \&PublicInbox::DS::accept_tls_step, \&_greet ];
 131         }
 132         $self->SUPER::new($sock, $ev | EPOLLONESHOT);
 133         if ($wbuf) {
 134                 $self->{wbuf} = $wbuf;
 135         } else {
 136                 _greet($self);
 137         }
 138         $self;
 139 }
 140
 141 # POP user is $UUID1@$NEWSGROUP.$SLICE
 142 sub cmd_user ($$) {
 143         my ($self, $mailbox) = @_;
 144         $self->{salt} // return \"-ERR already authed\r\n";
 145         $mailbox =~ s/\A([a-f0-9\-]+)\@//i or
 146                 return \"-ERR no UUID@ in mailbox name\r\n";
 147         my $user = $1;
 148         $user =~ tr/-//d; # most have dashes, some (dbus-uuidgen) don't
 149         $user =~ m!\A[a-f0-9]{32}\z!i or return \"-ERR user has no UUID\r\n";
 150         my $slice;
 151         $mailbox =~ s/\.([0-9]+)\z// and $slice = $1 + 0;
 152         my $ibx = $self->{pop3d}->{pi_cfg}->lookup_newsgroup($mailbox) //
 153                 return \"-ERR $mailbox does not exist\r\n";
 154         my $uidmax = $ibx->mm(1)->num_highwater // 0;
 155         if (defined $slice) {
 156                 my $max = int($uidmax / PublicInbox::IMAP::UID_SLICE);
 157                 my $tip = "$mailbox.$max";
 158                 return \"-ERR $mailbox.$slice does not exist ($tip does)\r\n"
 159                         if $slice > $max;
 160                 $self->{uid_base} = $slice * PublicInbox::IMAP::UID_SLICE;
 161                 $self->{slice} = $slice;
 162         } else { # latest 50K messages
 163                 my $base = $uidmax - PublicInbox::IMAP::UID_SLICE;
 164                 $self->{uid_base} = $base < 0 ? 0 : $base;
 165                 $self->{slice} = -1;
 166         }
 167         $self->{ibx} = $ibx;
 168         $self->{uuid} = pack('H*', $user); # deleted by _login_ok
 169         $slice //= '(latest)';
 170         \"+OK $ibx->{newsgroup} slice=$slice selected\r\n";
 171 }
 172
 173 sub _login_ok ($) {
 174         my ($self) = @_;
 175         if ($self->{pop3d}->lock_mailbox($self)) {
 176                 $self->{uid_max} = $self->{ibx}->over(1)->max;
 177                 \"+OK logged in\r\n";
 178         } else {
 179                 \"-ERR [IN-USE] unable to lock maildrop\r\n";
 180         }
 181 }
 182
 183 sub cmd_apop {
 184         my ($self, $mailbox, $hex) = @_;
 185         my $res = cmd_user($self, $mailbox); # sets {uuid}
 186         return $res if substr($$res, 0, 1) eq '-';
 187         my $s = delete($self->{salt}) // die 'BUG: salt missing';
 188         return _login_ok($self) if md5("<$s\@public-inbox>anonymous") eq
 189                                 pack('H*', $hex);
 190         $self->{salt} = $s;
 191         \"-ERR APOP password mismatch\r\n";
 192 }
 193
 194 sub cmd_pass {
 195         my ($self, $pass) = @_;
 196         $self->{ibx} // return \"-ERR mailbox unspecified\r\n";
 197         my $s = delete($self->{salt}) // return \"-ERR already authed\r\n";
 198         return _login_ok($self) if $pass eq 'anonymous';
 199         $self->{salt} = $s;
 200         \"-ERR password is not `anonymous'\r\n";
 201 }
 202
 203 sub cmd_stls {
 204         my ($self) = @_;
 205         my $sock = $self->{sock} or return;
 206         return \"-ERR TLS already enabled\r\n" if $sock->can('stop_SSL');
 207         my $opt = $self->{pop3d}->{accept_tls} or
 208                 return \"-ERR can't start TLS negotiation\r\n";
 209         $self->write(\"+OK begin TLS negotiation now\r\n");
 210         $self->{sock} = IO::Socket::SSL->start_SSL($sock, %$opt);
 211         $self->requeue if PublicInbox::DS::accept_tls_step($self);
 212         undef;
 213 }
 214
 215 sub need_txn ($) {
 216         exists($_[0]->{salt}) ? \"-ERR not in TRANSACTION\r\n" : undef;
 217 }
 218
 219 sub _stat_cache ($) {
 220         my ($self) = @_;
 221         my ($beg, $end) = (($self->{uid_dele} // -1) + 1, $self->{uid_max});
 222         PublicInbox::IMAP::uid_clamp($self, \$beg, \$end);
 223         my $opt = { limit => PublicInbox::IMAP::UID_SLICE };
 224         my $m = $self->{ibx}->over(1)->do_get(<<'', $opt, $beg, $end);
 225 SELECT num,ddd FROM over WHERE num >= ? AND num <= ?
 226 ORDER BY num ASC
 227
 228         [ map { ($_->{num}, $_->{bytes} + 0, $_->{blob}) } @$m ];
 229 }
 230
 231 sub cmd_stat {
 232         my ($self) = @_;
 233         my $err; $err = need_txn($self) and return $err;
 234         my $cache = $self->{cache} //= _stat_cache($self);
 235         my $tot = 0;
 236         for (my $i = 1; $i < scalar(@$cache); $i += 3) { $tot += $cache->[$i] }
 237         my $nr = @$cache / 3 - ($self->{nr_dele} // 0);
 238         "+OK $nr $tot\r\n";
 239 }
 240
 241 # for LIST and UIDL
 242 sub _list {
 243         my ($desc, $idx, $self, $msn) = @_;
 244         my $err; $err = need_txn($self) and return $err;
 245         my $cache = $self->{cache} //= _stat_cache($self);
 246         if (defined $msn) {
 247                 my $base_off = ($msn - 1) * 3;
 248                 my $val = $cache->[$base_off + $idx] //
 249                                 return \"-ERR no such message\r\n";
 250                 "+OK $desc listing follows\r\n$msn $val\r\n.\r\n";
 251         } else { # always +OK, even if no messages
 252                 my $res = "+OK $desc listing follows\r\n";
 253                 my $msn = 0;
 254                 for (my $i = 0; $i < scalar(@$cache); $i += 3) {
 255                         ++$msn;
 256                         defined($cache->[$i]) and
 257                                 $res .= "$msn $cache->[$i + $idx]\r\n";
 258                 }
 259                 $res .= ".\r\n";
 260         }
 261 }
 262
 263 sub cmd_list { _list('scan', 1, @_) }
 264 sub cmd_uidl { _list('unique-id', 2, @_) }
 265
 266 sub mark_dele ($$) {
 267         my ($self, $off) = @_;
 268         my $base_off = $off * 3;
 269         my $cache = $self->{cache};
 270         my $uid = $cache->[$base_off] // return; # already deleted
 271
 272         my $old = $self->{txn_max_uid} //= $uid;
 273         $self->{txn_max_uid} = $uid if $uid > $old;
 274
 275         $cache->[$base_off] = undef; # clobber UID
 276         $cache->[$base_off + 1] = 0; # zero bytes (simplifies cmd_stat)
 277         $cache->[$base_off + 2] = undef; # clobber oidhex
 278         ++$self->{nr_dele};
 279 }
 280
 281 sub retr_cb { # called by git->cat_async via ibx_async_cat
 282         my ($bref, $oid, $type, $size, $args) = @_;
 283         my ($self, $off, $top_nr) = @$args;
 284         my $hex = $self->{cache}->[$off * 3 + 2] //
 285                 die "BUG: no hex (oid=$oid)";
 286         if (!defined($oid)) {
 287                 # it's possible to have TOCTOU if an admin runs
 288                 # public-inbox-(edit|purge), just move onto the next message
 289                 warn "E: $hex missing in $self->{ibx}->{inboxdir}\n";
 290                 $self->write(\"-ERR no such message\r\n");
 291                 return $self->requeue;
 292         } elsif ($hex ne $oid) {
 293                 $self->close;
 294                 die "BUG: $hex != $oid";
 295         }
 296         PublicInbox::IMAP::to_crlf_full($bref);
 297         if (defined $top_nr) {
 298                 my ($hdr, $bdy) = split(/\r\n\r\n/, $$bref, 2);
 299                 $bref = \$hdr;
 300                 $hdr .= "\r\n\r\n";
 301                 my @tmp = split(/^/m, $bdy);
 302                 $hdr .= join('', splice(@tmp, 0, $top_nr));
 303         } elsif (exists $self->{expire}) {
 304                 $self->{expire} .= pack('S', $off + 1);
 305         }
 306         $$bref =~ s/^\./../gms;
 307         $$bref .= substr($$bref, -2, 2) eq "\r\n" ? ".\r\n" : "\r\n.\r\n";
 308         $self->msg_more("+OK message follows\r\n");
 309         $self->write($bref);
 310         $self->requeue;
 311 }
 312
 313 sub cmd_retr {
 314         my ($self, $msn, $top_nr) = @_;
 315         return \"-ERR lines must be a non-negative number\r\n" if
 316                         (defined($top_nr) && $top_nr !~ /\A[0-9]+\z/);
 317         my $err; $err = need_txn($self) and return $err;
 318         my $cache = $self->{cache} //= _stat_cache($self);
 319         my $off = $msn - 1;
 320         my $hex = $cache->[$off * 3 + 2] // return \"-ERR no such message\r\n";
 321         ${ibx_async_cat($self->{ibx}, $hex, \&retr_cb,
 322                         [ $self, $off, $top_nr ])};
 323 }
 324
 325 sub cmd_noop { $_[0]->write(\"+OK\r\n") }
 326
 327 sub cmd_rset {
 328         my ($self) = @_;
 329         my $err; $err = need_txn($self) and return $err;
 330         delete $self->{cache};
 331         delete $self->{txn_max_uid};
 332         \"+OK\r\n";
 333 }
 334
 335 sub cmd_dele {
 336         my ($self, $msn) = @_;
 337         my $err; $err = need_txn($self) and return $err;
 338         $self->{cache} //= _stat_cache($self);
 339         $msn =~ /\A[1-9][0-9]*\z/ or return \"-ERR no such message\r\n";
 340         mark_dele($self, $msn - 1) ? \"+OK\r\n" : \"-ERR no such message\r\n";
 341 }
 342
 343 # RFC 2449
 344 sub cmd_capa {
 345         my ($self) = @_;
 346         my $STLS = !$self->{ibx} && !$self->{sock}->can('stop_SSL') &&
 347                         $self->{pop3d}->{accept_tls} ? "\nSTLS\r" : '';
 348         $self->{expire} = ''; # "EXPIRE 0" allows clients to avoid DELE commands
 349         <<EOM;
 350 +OK Capability list follows\r
 351 TOP\r
 352 USER\r
 353 PIPELINING\r
 354 UIDL\r
 355 EXPIRE 0\r
 356 RESP-CODES\r$STLS
 357 .\r
 358 EOM
 359 }
 360
 361 sub close {
 362         my ($self) = @_;
 363         $self->{pop3d}->unlock_mailbox($self);
 364         $self->SUPER::close;
 365 }
 366
 367 sub cmd_quit {
 368         my ($self) = @_;
 369         if (defined(my $txn_id = $self->{txn_id})) {
 370                 my $user_id = $self->{user_id} // die 'BUG: no {user_id}';
 371                 if (my $exp = delete $self->{expire}) {
 372                         mark_dele($self, $_) for unpack('S*', $exp);
 373                 }
 374                 my $dbh = $self->{pop3d}->{-state_dbh};
 375                 my $lk = $self->{pop3d}->lock_for_scope;
 376                 my $sth;
 377                 $dbh->begin_work;
 378
 379                 if (defined $self->{txn_max_uid}) {
 380                         $sth = $dbh->prepare_cached(<<'');
 381 UPDATE deletes SET uid_dele = ? WHERE txn_id = ? AND uid_dele < ?
 382
 383                         $sth->execute($self->{txn_max_uid}, $txn_id,
 384                                         $self->{txn_max_uid});
 385                 }
 386                 $sth = $dbh->prepare_cached(<<'');
 387 UPDATE users SET last_seen = ? WHERE user_id = ?
 388
 389                 $sth->execute(time, $user_id);
 390                 $dbh->commit;
 391         }
 392         $self->write(\"+OK public-inbox POP3 server signing off\r\n");
 393         $self->close;
 394         undef;
 395 }
 396
 397 # returns 1 if we can continue, 0 if not due to buffered writes or disconnect
 398 sub process_line ($$) {
 399         my ($self, $l) = @_;
 400         my ($req, @args) = split(/[ \t]+/, $l);
 401         return 1 unless defined($req); # skip blank line
 402         $req = $self->can('cmd_'.lc($req));
 403         my $res = $req ? eval { $req->($self, @args) } :
 404                 \"-ERR command not recognized\r\n";
 405         my $err = $@;
 406         if ($err && $self->{sock}) {
 407                 chomp($l);
 408                 err($self, 'error from: %s (%s)', $l, $err);
 409                 $res = \"-ERR program fault - command not performed\r\n";
 410         }
 411         defined($res) ? $self->write($res) : 0;
 412 }
 413
 414 # callback used by PublicInbox::DS for any (e)poll (in/out/hup/err)
 415 sub event_step {
 416         my ($self) = @_;
 417         return unless $self->flush_write && $self->{sock} && !$self->{long_cb};
 418
 419         # only read more requests if we've drained the write buffer,
 420         # otherwise we can be buffering infinitely w/o backpressure
 421         my $rbuf = $self->{rbuf} // \(my $x = '');
 422         my $line = index($$rbuf, "\n");
 423         while ($line < 0) {
 424                 return $self->close if length($$rbuf) >= LINE_MAX;
 425                 $self->do_read($rbuf, LINE_MAX, length($$rbuf)) or return;
 426                 $line = index($$rbuf, "\n");
 427         }
 428         $line = substr($$rbuf, 0, $line + 1, '');
 429         $line =~ s/\r?\n\z//s;
 430         return $self->close if $line =~ /[[:cntrl:]]/s;
 431         my $t0 = now();
 432         my $fd = fileno($self->{sock}); # may become invalid after process_line
 433         my $r = eval { process_line($self, $line) };
 434         my $pending = $self->{wbuf} ? ' pending' : '';
 435         out($self, "[$fd] %s - %0.6f$pending - $r", $line, now() - $t0);
 436         return $self->close if $r < 0;
 437         $self->rbuf_idle($rbuf);
 438
 439         # maybe there's more pipelined data, or we'll have
 440         # to register it for socket-readiness notifications
 441         $self->requeue unless $pending;
 442 }
 443
 444 no warnings 'once';
 445 *cmd_top = \&cmd_retr;
 446
 447 1;