diff options
Diffstat (limited to 'lib/PublicInbox/POP3.pm')
| -rw-r--r-- | lib/PublicInbox/POP3.pm | 428 |
1 files changed, 428 insertions, 0 deletions
diff --git a/lib/PublicInbox/POP3.pm b/lib/PublicInbox/POP3.pm new file mode 100644 index 00000000..06772069 --- /dev/null +++ b/lib/PublicInbox/POP3.pm @@ -0,0 +1,428 @@ +# Copyright (C) all contributors <meta@public-inbox.org> +# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt> +# +# Each instance of this represents a POP3 client connected to +# public-inbox-{netd,pop3d}. Much of this was taken from IMAP.pm and NNTP.pm +# +# POP3 is one mailbox per-user, so the "USER" command is like the +# format of -imapd and is mapped to $NEWSGROUP.$SLICE (large inboxes +# are sliced into 50K mailboxes in both POP3 and IMAP to avoid overloading +# clients) +# +# Unlike IMAP, the "$NEWSGROUP" mailbox (without $SLICE) is a rolling +# window of the latest messages. We can do this for POP3 since the +# typical POP3 session is short-lived while long-lived IMAP sessions +# would cause slices to grow on the server side without bounds. +# +# Like IMAP, POP3 also has per-session message sequence numbers (MSN), +# which require mapping to UIDs. The offset of an entry into our +# per-client cache is: (MSN-1) +# +# fields: +# - uuid - 16-byte (binary) UUID representation (before successful login) +# - cache - one-dimentional arrayref of (UID, bytesize, oidhex) +# - nr_dele - number of deleted messages +# - expire - string of packed unsigned short offsets +# - user_id - user-ID mapped to UUID (on successful login + lock) +# - txn_max_uid - for storing max deleted UID persistently +# - ibx - PublicInbox::Inbox object +# - slice - unsigned integer slice number (0..Inf), -1 => latest +# - salt - pre-auth for APOP +# - uid_dele - maximum deleted from previous session at login (NNTP ARTICLE) +# - uid_base - base UID for mailbox slice (0-based) (same as IMAP) +package PublicInbox::POP3; +use v5.12; +use parent qw(PublicInbox::DS); +use PublicInbox::GitAsyncCat; +use PublicInbox::DS qw(now); +use Errno qw(EAGAIN); +use Digest::MD5 qw(md5); +use PublicInbox::IMAP; # for UID slice stuff + +use constant { + LINE_MAX => 512, # XXX unsure + UID_SLICE => PublicInbox::IMAP::UID_SLICE, +}; + +# XXX FIXME: duplicated stuff from NNTP.pm and IMAP.pm + +sub out ($$;@) { + my ($self, $fmt, @args) = @_; + printf { $self->{pop3d}->{out} } $fmt."\n", @args; +} + +sub do_greet { + my ($self) = @_; + my $s = $self->{salt} = sprintf('%x.%x', int(rand(0x7fffffff)), time); + $self->write("+OK POP3 server ready <$s\@public-inbox>\r\n"); +} + +sub new { + my ($cls, $sock, $pop3d) = @_; + (bless { pop3d => $pop3d }, $cls)->greet($sock) +} + +# POP user is $UUID1@$NEWSGROUP[.$SLICE][?QUERY_ARGS] +sub cmd_user ($$) { + my ($self, $mailbox) = @_; + $self->{salt} // return \"-ERR already authed\r\n"; + $mailbox =~ s/\A([a-f0-9\-]+)\@//i or + return \"-ERR no UUID@ in mailbox name\r\n"; + my $user = $1; + $user =~ tr/-//d; # most have dashes, some (dbus-uuidgen) don't + $user =~ m!\A[a-f0-9]{32}\z!i or return \"-ERR user has no UUID\r\n"; + + my %l; + if ($mailbox =~ s/\?(.*)\z//) { # query args + for (split(/&+/, $1)) { + /\A(initial_limit|limit)=([0-9]+)\z/ and $l{$1} = $2; + } + $self->{limits} = \%l; + } + my $slice = $mailbox =~ s/\.([0-9]+)\z// ? $1 + 0 : undef; + + my $ibx = $self->{pop3d}->{pi_cfg}->lookup_newsgroup($mailbox) // + return \"-ERR $mailbox does not exist\r\n"; + my $uidmax = $self->{uidmax} = $ibx->mm(1)->num_highwater // 0; + if (defined $slice) { + my $max = int($uidmax / UID_SLICE); + my $tip = "$mailbox.$max"; + return \"-ERR $mailbox.$slice does not exist ($tip does)\r\n" + if $slice > $max; + $self->{slice} = $slice; + } else { # latest messages: + $self->{slice} = -1; + } + $self->{ibx} = $ibx; + $self->{uuid} = pack('H*', $user); # deleted by _login_ok + $slice //= '(latest)'; + \"+OK $ibx->{newsgroup} slice=$slice selected\r\n"; +} + +sub _login_ok ($) { + my ($self) = @_; + $self->{pop3d}->lock_mailbox($self) or + return \"-ERR [IN-USE] unable to lock maildrop\r\n"; + + my $l = delete $self->{limits}; + $l = defined($self->{uid_dele}) ? $l->{limit} + : ($l->{initial_limit} // $l->{limit}); + my $uidmax = delete $self->{uidmax}; + if ($self->{slice} >= 0) { + $self->{uid_base} = $self->{slice} * UID_SLICE; + if (defined $l) { # n.b: the last slice is not full: + my $max = int($uidmax/UID_SLICE) == $self->{slice} ? + ($uidmax % UID_SLICE) : UID_SLICE; + my $off = $max - $l; + $self->{uid_base} += $off if $off > 0; + } + } else { # latest $l messages, or 1k if unspecified + my $base = $uidmax - ($l // 1000); + $self->{uid_base} = $base < 0 ? 0 : $base; + } + $self->{uid_max} = $self->{ibx}->over(1)->max; + \"+OK logged in\r\n"; +} + +sub cmd_apop { + my ($self, $mailbox, $hex) = @_; + my $res = cmd_user($self, $mailbox); # sets {uuid} + return $res if substr($$res, 0, 1) eq '-'; + my $s = delete($self->{salt}) // die 'BUG: salt missing'; + return _login_ok($self) if md5("<$s\@public-inbox>anonymous") eq + pack('H*', $hex); + $self->{salt} = $s; + \"-ERR APOP password mismatch\r\n"; +} + +sub cmd_pass { + my ($self, $pass) = @_; + $self->{ibx} // return \"-ERR mailbox unspecified\r\n"; + my $s = delete($self->{salt}) // return \"-ERR already authed\r\n"; + return _login_ok($self) if $pass eq 'anonymous'; + $self->{salt} = $s; + \"-ERR password is not `anonymous'\r\n"; +} + +sub cmd_stls { + my ($self) = @_; + ($self->{sock} // return)->can('stop_SSL') and + return \"-ERR TLS already enabled\r\n"; + $self->{pop3d}->{ssl_ctx_opt} or + return \"-ERR can't start TLS negotiation\r\n"; + $self->write(\"+OK begin TLS negotiation now\r\n"); + PublicInbox::TLS::start($self->{sock}, $self->{pop3d}); + $self->requeue if PublicInbox::DS::accept_tls_step($self); + undef; +} + +sub need_txn ($) { + exists($_[0]->{salt}) ? \"-ERR not in TRANSACTION\r\n" : undef; +} + +sub _stat_cache ($) { + my ($self) = @_; + my ($beg, $end) = (($self->{uid_dele} // -1) + 1, $self->{uid_max}); + PublicInbox::IMAP::uid_clamp($self, \$beg, \$end); + my (@cache, $m); + my $sth = $self->{ibx}->over(1)->dbh->prepare_cached(<<'', undef, 1); +SELECT num,ddd FROM over WHERE num >= ? AND num <= ? +ORDER BY num ASC + + $sth->execute($beg, $end); + my $tot = 0; + while (defined($m = $sth->fetchall_arrayref({}, 1000))) { + for my $x (@$m) { + PublicInbox::Over::load_from_row($x); + push(@cache, $x->{num}, $x->{bytes} + 0, $x->{blob}); + undef $x; # saves ~1.5M memory w/ 50k messages + $tot += $cache[-2]; + } + } + $self->{total_bytes} = $tot; + $self->{cache} = \@cache; +} + +sub cmd_stat { + my ($self) = @_; + my $err; $err = need_txn($self) and return $err; + my $cache = $self->{cache} // _stat_cache($self); + my $nr = @$cache / 3 - ($self->{nr_dele} // 0); + "+OK $nr $self->{total_bytes}\r\n"; +} + +# for LIST and UIDL +sub _list { + my ($desc, $idx, $self, $msn) = @_; + my $err; $err = need_txn($self) and return $err; + my $cache = $self->{cache} // _stat_cache($self); + if (defined $msn) { + my $base_off = ($msn - 1) * 3; + my $val = $cache->[$base_off + $idx] // + return \"-ERR no such message\r\n"; + "+OK $desc listing follows\r\n$msn $val\r\n.\r\n"; + } else { # always +OK, even if no messages + my $res = "+OK $desc listing follows\r\n"; + my $msn = 0; + for (my $i = 0; $i < scalar(@$cache); $i += 3) { + ++$msn; + defined($cache->[$i]) and + $res .= "$msn $cache->[$i + $idx]\r\n"; + } + $res .= ".\r\n"; + } +} + +sub cmd_list { _list('scan', 1, @_) } +sub cmd_uidl { _list('unique-id', 2, @_) } + +sub mark_dele ($$) { + my ($self, $off) = @_; + my $base_off = $off * 3; + my $cache = $self->{cache}; + my $uid = $cache->[$base_off] // return; # already deleted + + my $old = $self->{txn_max_uid} //= $uid; + $self->{txn_max_uid} = $uid if $uid > $old; + + $self->{total_bytes} -= $cache->[$base_off + 1]; + $cache->[$base_off] = undef; # clobber UID + $cache->[$base_off + 1] = undef; # clobber bytes + $cache->[$base_off + 2] = undef; # clobber oidhex + ++$self->{nr_dele}; +} + +sub retr_cb { # called by git->cat_async via ibx_async_cat + my ($bref, $oid, $type, $size, $args) = @_; + my ($self, $off, $top_nr) = @$args; + my $hex = $self->{cache}->[$off * 3 + 2] // + die "BUG: no hex (oid=$oid)"; + if (!defined($type)) { + warn "E: git aborted on $oid / $hex $self->{ibx}->{inboxdir}"; + return $self->close; + } elsif ($type ne 'blob') { + # it's possible to have TOCTOU if an admin runs + # public-inbox-(edit|purge), just move onto the next message + warn "E: $hex missing in $self->{ibx}->{inboxdir}\n"; + $self->write(\"-ERR no such message\r\n"); + return $self->requeue; + } elsif ($hex ne $oid) { + $self->close; + die "BUG: $hex != $oid"; + } + PublicInbox::IMAP::to_crlf_full($bref); + if (defined $top_nr) { + my ($hdr, $bdy) = split(/\r\n\r\n/, $$bref, 2); + $bref = \$hdr; + $hdr .= "\r\n\r\n"; + my @tmp = split(/^/m, $bdy); + $hdr .= join('', splice(@tmp, 0, $top_nr)); + } elsif (exists $self->{expire}) { + $self->{expire} .= pack('S', $off); + } + $$bref =~ s/^\./../gms; + $$bref .= substr($$bref, -2, 2) eq "\r\n" ? ".\r\n" : "\r\n.\r\n"; + $self->msg_more("+OK message follows\r\n"); + $self->write($bref); + $self->requeue; +} + +sub cmd_retr { + my ($self, $msn, $top_nr) = @_; + return \"-ERR lines must be a non-negative number\r\n" if + (defined($top_nr) && $top_nr !~ /\A[0-9]+\z/); + my $err; $err = need_txn($self) and return $err; + my $cache = $self->{cache} // _stat_cache($self); + my $off = $msn - 1; + my $hex = $cache->[$off * 3 + 2] // return \"-ERR no such message\r\n"; + ${ibx_async_cat($self->{ibx}, $hex, \&retr_cb, + [ $self, $off, $top_nr ])}; +} + +sub cmd_noop { $_[0]->write(\"+OK\r\n") } + +sub cmd_rset { + my ($self) = @_; + my $err; $err = need_txn($self) and return $err; + delete $self->{cache}; + delete $self->{txn_max_uid}; + \"+OK\r\n"; +} + +sub cmd_dele { + my ($self, $msn) = @_; + my $err; $err = need_txn($self) and return $err; + $self->{cache} // _stat_cache($self); + $msn =~ /\A[1-9][0-9]*\z/ or return \"-ERR no such message\r\n"; + mark_dele($self, $msn - 1) ? \"+OK\r\n" : \"-ERR no such message\r\n"; +} + +# RFC 2449 +sub cmd_capa { + my ($self) = @_; + my $STLS = !$self->{ibx} && !$self->{sock}->can('stop_SSL') && + $self->{pop3d}->{ssl_ctx_opt} ? "\nSTLS\r" : ''; + $self->{expire} = ''; # "EXPIRE 0" allows clients to avoid DELE commands + <<EOM; ++OK Capability list follows\r +TOP\r +USER\r +PIPELINING\r +UIDL\r +EXPIRE 0\r +RESP-CODES\r$STLS +.\r +EOM +} + +sub close { + my ($self) = @_; + $self->{pop3d}->unlock_mailbox($self); + $self->SUPER::close; +} + +# must be called inside a state_dbh transaction with flock held +sub __cleanup_state { + my ($self, $txn_id) = @_; + my $user_id = $self->{user_id} // die 'BUG: no {user_id}'; + $self->{pop3d}->{-state_dbh}->prepare_cached(<<'')->execute($txn_id); +DELETE FROM deletes WHERE txn_id = ? AND uid_dele = -1 + + my $sth = $self->{pop3d}->{-state_dbh}->prepare_cached(<<'', undef, 1); +SELECT COUNT(*) FROM deletes WHERE user_id = ? + + $sth->execute($user_id); + my $nr = $sth->fetchrow_array; + if ($nr == 0) { + $sth = $self->{pop3d}->{-state_dbh}->prepare_cached(<<''); +DELETE FROM users WHERE user_id = ? + + $sth->execute($user_id); + } + $nr; +} + +sub cmd_quit { + my ($self) = @_; + if (defined(my $txn_id = $self->{txn_id})) { + my $user_id = $self->{user_id} // die 'BUG: no {user_id}'; + if (my $exp = delete $self->{expire}) { + mark_dele($self, $_) for unpack('S*', $exp); + } + my $keep = 1; + my $dbh = $self->{pop3d}->{-state_dbh}; + my $lk = $self->{pop3d}->lock_for_scope; + $dbh->begin_work; + + if (defined(my $max = $self->{txn_max_uid})) { + $dbh->prepare_cached(<<'')->execute($max, $txn_id, $max) +UPDATE deletes SET uid_dele = ? WHERE txn_id = ? AND uid_dele < ? + + } else { + $keep = $self->__cleanup_state($txn_id); + } + $dbh->prepare_cached(<<'')->execute(time, $user_id) if $keep; +UPDATE users SET last_seen = ? WHERE user_id = ? + + $dbh->commit; + # we MUST do txn_id F_UNLCK here inside ->lock_for_scope: + $self->{did_quit} = 1; + $self->{pop3d}->unlock_mailbox($self); + } + $self->write(\"+OK public-inbox POP3 server signing off\r\n"); + $self->shutdn; + undef; +} + +# returns 1 if we can continue, 0 if not due to buffered writes or disconnect +sub process_line ($$) { + my ($self, $l) = @_; + my ($req, @args) = split(/[ \t]+/, $l); + return 1 unless defined($req); # skip blank line + $req = $self->can('cmd_'.lc($req)); + my $res = $req ? eval { $req->($self, @args) } : + \"-ERR command not recognized\r\n"; + my $err = $@; + if ($err && $self->{sock}) { + $l =~ s/\r?\n//s; + warn("error from: $l ($err)\n"); + $res = \"-ERR program fault - command not performed\r\n"; + } + defined($res) ? $self->write($res) : 0; +} + +# callback used by PublicInbox::DS for any (e)poll (in/out/hup/err) +sub event_step { + my ($self) = @_; + local $SIG{__WARN__} = $self->{pop3d}->{warn_cb}; + return unless $self->flush_write && $self->{sock} && !$self->{long_cb}; + + # only read more requests if we've drained the write buffer, + # otherwise we can be buffering infinitely w/o backpressure + my $rbuf = $self->{rbuf} // \(my $x = ''); + my $line = index($$rbuf, "\n"); + while ($line < 0) { + return $self->close if length($$rbuf) >= LINE_MAX; + $self->do_read($rbuf, LINE_MAX, length($$rbuf)) or return; + $line = index($$rbuf, "\n"); + } + $line = substr($$rbuf, 0, $line + 1, ''); + $line =~ s/\r?\n\z//s; + return $self->close if $line =~ /[[:cntrl:]]/s; + my $t0 = now(); + my $fd = fileno($self->{sock}); # may become invalid after process_line + my $r = eval { process_line($self, $line) }; + my $pending = $self->{wbuf} ? ' pending' : ''; + out($self, "[$fd] %s - %0.6f$pending - $r", $line, now() - $t0); + return $self->close if $r < 0; + $self->rbuf_idle($rbuf); + + # maybe there's more pipelined data, or we'll have + # to register it for socket-readiness notifications + $self->requeue unless $pending; +} + +no warnings 'once'; +*cmd_top = \&cmd_retr; + +1; |