]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/POP3.pm
pop3: reduce memory use while generating the mailbox cache
[public-inbox.git] / 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::GitAsyncCat;
37 use PublicInbox::DS qw(now);
38 use Errno qw(EAGAIN);
39 use Digest::MD5 qw(md5);
40 use PublicInbox::IMAP; # for UID slice stuff
41
42 use constant {
43         LINE_MAX => 512, # XXX unsure
44 };
45
46 # XXX FIXME: duplicated stuff from NNTP.pm and IMAP.pm
47
48 sub err ($$;@) {
49         my ($self, $fmt, @args) = @_;
50         printf { $self->{pop3d}->{err} } $fmt."\n", @args;
51 }
52
53 sub out ($$;@) {
54         my ($self, $fmt, @args) = @_;
55         printf { $self->{pop3d}->{out} } $fmt."\n", @args;
56 }
57
58 sub do_greet {
59         my ($self) = @_;
60         my $s = $self->{salt} = sprintf('%x.%x', int(rand(0x7fffffff)), time);
61         $self->write("+OK POP3 server ready <$s\@public-inbox>\r\n");
62 }
63
64 sub new {
65         my ($cls, $sock, $pop3d) = @_;
66         (bless { pop3d => $pop3d }, $cls)->greet($sock)
67 }
68
69 # POP user is $UUID1@$NEWSGROUP.$SLICE
70 sub cmd_user ($$) {
71         my ($self, $mailbox) = @_;
72         $self->{salt} // return \"-ERR already authed\r\n";
73         $mailbox =~ s/\A([a-f0-9\-]+)\@//i or
74                 return \"-ERR no UUID@ in mailbox name\r\n";
75         my $user = $1;
76         $user =~ tr/-//d; # most have dashes, some (dbus-uuidgen) don't
77         $user =~ m!\A[a-f0-9]{32}\z!i or return \"-ERR user has no UUID\r\n";
78         my $slice;
79         $mailbox =~ s/\.([0-9]+)\z// and $slice = $1 + 0;
80         my $ibx = $self->{pop3d}->{pi_cfg}->lookup_newsgroup($mailbox) //
81                 return \"-ERR $mailbox does not exist\r\n";
82         my $uidmax = $ibx->mm(1)->num_highwater // 0;
83         if (defined $slice) {
84                 my $max = int($uidmax / PublicInbox::IMAP::UID_SLICE);
85                 my $tip = "$mailbox.$max";
86                 return \"-ERR $mailbox.$slice does not exist ($tip does)\r\n"
87                         if $slice > $max;
88                 $self->{uid_base} = $slice * PublicInbox::IMAP::UID_SLICE;
89                 $self->{slice} = $slice;
90         } else { # latest 50K messages
91                 my $base = $uidmax - PublicInbox::IMAP::UID_SLICE;
92                 $self->{uid_base} = $base < 0 ? 0 : $base;
93                 $self->{slice} = -1;
94         }
95         $self->{ibx} = $ibx;
96         $self->{uuid} = pack('H*', $user); # deleted by _login_ok
97         $slice //= '(latest)';
98         \"+OK $ibx->{newsgroup} slice=$slice selected\r\n";
99 }
100
101 sub _login_ok ($) {
102         my ($self) = @_;
103         if ($self->{pop3d}->lock_mailbox($self)) {
104                 $self->{uid_max} = $self->{ibx}->over(1)->max;
105                 \"+OK logged in\r\n";
106         } else {
107                 \"-ERR [IN-USE] unable to lock maildrop\r\n";
108         }
109 }
110
111 sub cmd_apop {
112         my ($self, $mailbox, $hex) = @_;
113         my $res = cmd_user($self, $mailbox); # sets {uuid}
114         return $res if substr($$res, 0, 1) eq '-';
115         my $s = delete($self->{salt}) // die 'BUG: salt missing';
116         return _login_ok($self) if md5("<$s\@public-inbox>anonymous") eq
117                                 pack('H*', $hex);
118         $self->{salt} = $s;
119         \"-ERR APOP password mismatch\r\n";
120 }
121
122 sub cmd_pass {
123         my ($self, $pass) = @_;
124         $self->{ibx} // return \"-ERR mailbox unspecified\r\n";
125         my $s = delete($self->{salt}) // return \"-ERR already authed\r\n";
126         return _login_ok($self) if $pass eq 'anonymous';
127         $self->{salt} = $s;
128         \"-ERR password is not `anonymous'\r\n";
129 }
130
131 sub cmd_stls {
132         my ($self) = @_;
133         my $sock = $self->{sock} or return;
134         return \"-ERR TLS already enabled\r\n" if $sock->can('stop_SSL');
135         my $opt = $self->{pop3d}->{accept_tls} or
136                 return \"-ERR can't start TLS negotiation\r\n";
137         $self->write(\"+OK begin TLS negotiation now\r\n");
138         $self->{sock} = IO::Socket::SSL->start_SSL($sock, %$opt);
139         $self->requeue if PublicInbox::DS::accept_tls_step($self);
140         undef;
141 }
142
143 sub need_txn ($) {
144         exists($_[0]->{salt}) ? \"-ERR not in TRANSACTION\r\n" : undef;
145 }
146
147 sub _stat_cache ($) {
148         my ($self) = @_;
149         my ($beg, $end) = (($self->{uid_dele} // -1) + 1, $self->{uid_max});
150         PublicInbox::IMAP::uid_clamp($self, \$beg, \$end);
151         my (@cache, $m);
152         my $sth = $self->{ibx}->over(1)->dbh->prepare_cached(<<'', undef, 1);
153 SELECT num,ddd FROM over WHERE num >= ? AND num <= ?
154 ORDER BY num ASC
155
156         $sth->execute($beg, $end);
157         do {
158                 $m = $sth->fetchall_arrayref({}, 1000);
159                 for my $x (@$m) {
160                         PublicInbox::Over::load_from_row($x);
161                         push(@cache, $x->{num}, $x->{bytes} + 0, $x->{blob});
162                         undef $x; # saves ~1.5M memory w/ 50k messages
163                 }
164         } while (scalar(@$m) && ($beg = $cache[-3] + 1));
165         \@cache;
166 }
167
168 sub cmd_stat {
169         my ($self) = @_;
170         my $err; $err = need_txn($self) and return $err;
171         my $cache = $self->{cache} //= _stat_cache($self);
172         my $tot = 0;
173         for (my $i = 1; $i < scalar(@$cache); $i += 3) { $tot += $cache->[$i] }
174         my $nr = @$cache / 3 - ($self->{nr_dele} // 0);
175         "+OK $nr $tot\r\n";
176 }
177
178 # for LIST and UIDL
179 sub _list {
180         my ($desc, $idx, $self, $msn) = @_;
181         my $err; $err = need_txn($self) and return $err;
182         my $cache = $self->{cache} //= _stat_cache($self);
183         if (defined $msn) {
184                 my $base_off = ($msn - 1) * 3;
185                 my $val = $cache->[$base_off + $idx] //
186                                 return \"-ERR no such message\r\n";
187                 "+OK $desc listing follows\r\n$msn $val\r\n.\r\n";
188         } else { # always +OK, even if no messages
189                 my $res = "+OK $desc listing follows\r\n";
190                 my $msn = 0;
191                 for (my $i = 0; $i < scalar(@$cache); $i += 3) {
192                         ++$msn;
193                         defined($cache->[$i]) and
194                                 $res .= "$msn $cache->[$i + $idx]\r\n";
195                 }
196                 $res .= ".\r\n";
197         }
198 }
199
200 sub cmd_list { _list('scan', 1, @_) }
201 sub cmd_uidl { _list('unique-id', 2, @_) }
202
203 sub mark_dele ($$) {
204         my ($self, $off) = @_;
205         my $base_off = $off * 3;
206         my $cache = $self->{cache};
207         my $uid = $cache->[$base_off] // return; # already deleted
208
209         my $old = $self->{txn_max_uid} //= $uid;
210         $self->{txn_max_uid} = $uid if $uid > $old;
211
212         $cache->[$base_off] = undef; # clobber UID
213         $cache->[$base_off + 1] = 0; # zero bytes (simplifies cmd_stat)
214         $cache->[$base_off + 2] = undef; # clobber oidhex
215         ++$self->{nr_dele};
216 }
217
218 sub retr_cb { # called by git->cat_async via ibx_async_cat
219         my ($bref, $oid, $type, $size, $args) = @_;
220         my ($self, $off, $top_nr) = @$args;
221         my $hex = $self->{cache}->[$off * 3 + 2] //
222                 die "BUG: no hex (oid=$oid)";
223         if (!defined($oid)) {
224                 # it's possible to have TOCTOU if an admin runs
225                 # public-inbox-(edit|purge), just move onto the next message
226                 warn "E: $hex missing in $self->{ibx}->{inboxdir}\n";
227                 $self->write(\"-ERR no such message\r\n");
228                 return $self->requeue;
229         } elsif ($hex ne $oid) {
230                 $self->close;
231                 die "BUG: $hex != $oid";
232         }
233         PublicInbox::IMAP::to_crlf_full($bref);
234         if (defined $top_nr) {
235                 my ($hdr, $bdy) = split(/\r\n\r\n/, $$bref, 2);
236                 $bref = \$hdr;
237                 $hdr .= "\r\n\r\n";
238                 my @tmp = split(/^/m, $bdy);
239                 $hdr .= join('', splice(@tmp, 0, $top_nr));
240         } elsif (exists $self->{expire}) {
241                 $self->{expire} .= pack('S', $off + 1);
242         }
243         $$bref =~ s/^\./../gms;
244         $$bref .= substr($$bref, -2, 2) eq "\r\n" ? ".\r\n" : "\r\n.\r\n";
245         $self->msg_more("+OK message follows\r\n");
246         $self->write($bref);
247         $self->requeue;
248 }
249
250 sub cmd_retr {
251         my ($self, $msn, $top_nr) = @_;
252         return \"-ERR lines must be a non-negative number\r\n" if
253                         (defined($top_nr) && $top_nr !~ /\A[0-9]+\z/);
254         my $err; $err = need_txn($self) and return $err;
255         my $cache = $self->{cache} //= _stat_cache($self);
256         my $off = $msn - 1;
257         my $hex = $cache->[$off * 3 + 2] // return \"-ERR no such message\r\n";
258         ${ibx_async_cat($self->{ibx}, $hex, \&retr_cb,
259                         [ $self, $off, $top_nr ])};
260 }
261
262 sub cmd_noop { $_[0]->write(\"+OK\r\n") }
263
264 sub cmd_rset {
265         my ($self) = @_;
266         my $err; $err = need_txn($self) and return $err;
267         delete $self->{cache};
268         delete $self->{txn_max_uid};
269         \"+OK\r\n";
270 }
271
272 sub cmd_dele {
273         my ($self, $msn) = @_;
274         my $err; $err = need_txn($self) and return $err;
275         $self->{cache} //= _stat_cache($self);
276         $msn =~ /\A[1-9][0-9]*\z/ or return \"-ERR no such message\r\n";
277         mark_dele($self, $msn - 1) ? \"+OK\r\n" : \"-ERR no such message\r\n";
278 }
279
280 # RFC 2449
281 sub cmd_capa {
282         my ($self) = @_;
283         my $STLS = !$self->{ibx} && !$self->{sock}->can('stop_SSL') &&
284                         $self->{pop3d}->{accept_tls} ? "\nSTLS\r" : '';
285         $self->{expire} = ''; # "EXPIRE 0" allows clients to avoid DELE commands
286         <<EOM;
287 +OK Capability list follows\r
288 TOP\r
289 USER\r
290 PIPELINING\r
291 UIDL\r
292 EXPIRE 0\r
293 RESP-CODES\r$STLS
294 .\r
295 EOM
296 }
297
298 sub close {
299         my ($self) = @_;
300         $self->{pop3d}->unlock_mailbox($self);
301         $self->SUPER::close;
302 }
303
304 sub cmd_quit {
305         my ($self) = @_;
306         if (defined(my $txn_id = $self->{txn_id})) {
307                 my $user_id = $self->{user_id} // die 'BUG: no {user_id}';
308                 if (my $exp = delete $self->{expire}) {
309                         mark_dele($self, $_) for unpack('S*', $exp);
310                 }
311                 my $dbh = $self->{pop3d}->{-state_dbh};
312                 my $lk = $self->{pop3d}->lock_for_scope;
313                 my $sth;
314                 $dbh->begin_work;
315
316                 if (defined $self->{txn_max_uid}) {
317                         $sth = $dbh->prepare_cached(<<'');
318 UPDATE deletes SET uid_dele = ? WHERE txn_id = ? AND uid_dele < ?
319
320                         $sth->execute($self->{txn_max_uid}, $txn_id,
321                                         $self->{txn_max_uid});
322                 }
323                 $sth = $dbh->prepare_cached(<<'');
324 UPDATE users SET last_seen = ? WHERE user_id = ?
325
326                 $sth->execute(time, $user_id);
327                 $dbh->commit;
328         }
329         $self->write(\"+OK public-inbox POP3 server signing off\r\n");
330         $self->close;
331         undef;
332 }
333
334 # returns 1 if we can continue, 0 if not due to buffered writes or disconnect
335 sub process_line ($$) {
336         my ($self, $l) = @_;
337         my ($req, @args) = split(/[ \t]+/, $l);
338         return 1 unless defined($req); # skip blank line
339         $req = $self->can('cmd_'.lc($req));
340         my $res = $req ? eval { $req->($self, @args) } :
341                 \"-ERR command not recognized\r\n";
342         my $err = $@;
343         if ($err && $self->{sock}) {
344                 chomp($l);
345                 err($self, 'error from: %s (%s)', $l, $err);
346                 $res = \"-ERR program fault - command not performed\r\n";
347         }
348         defined($res) ? $self->write($res) : 0;
349 }
350
351 # callback used by PublicInbox::DS for any (e)poll (in/out/hup/err)
352 sub event_step {
353         my ($self) = @_;
354         return unless $self->flush_write && $self->{sock} && !$self->{long_cb};
355
356         # only read more requests if we've drained the write buffer,
357         # otherwise we can be buffering infinitely w/o backpressure
358         my $rbuf = $self->{rbuf} // \(my $x = '');
359         my $line = index($$rbuf, "\n");
360         while ($line < 0) {
361                 return $self->close if length($$rbuf) >= LINE_MAX;
362                 $self->do_read($rbuf, LINE_MAX, length($$rbuf)) or return;
363                 $line = index($$rbuf, "\n");
364         }
365         $line = substr($$rbuf, 0, $line + 1, '');
366         $line =~ s/\r?\n\z//s;
367         return $self->close if $line =~ /[[:cntrl:]]/s;
368         my $t0 = now();
369         my $fd = fileno($self->{sock}); # may become invalid after process_line
370         my $r = eval { process_line($self, $line) };
371         my $pending = $self->{wbuf} ? ' pending' : '';
372         out($self, "[$fd] %s - %0.6f$pending - $r", $line, now() - $t0);
373         return $self->close if $r < 0;
374         $self->rbuf_idle($rbuf);
375
376         # maybe there's more pipelined data, or we'll have
377         # to register it for socket-readiness notifications
378         $self->requeue unless $pending;
379 }
380
381 no warnings 'once';
382 *cmd_top = \&cmd_retr;
383
384 1;