From: Eric Wong Date: Mon, 27 Jul 2026 12:30:09 +0000 (+0000) Subject: ipc: more error-checking for sendmsg(..., MSG_EOR) X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=HEAD;p=thirdparty%2Fpublic-inbox.git ipc: more error-checking for sendmsg(..., MSG_EOR) Introduce `sendmsg_eor' for error checking to ensure we don't hit short writes when writing to SOCK_SEQPACKET sockets. Reducing imports of MSG_EOR will also make it easier for us to drop it in the future if certain OSes end up being too buggy with it (as was the case with OpenBSD 7.3). It'll also make it easier to switch to SOCK_DGRAM in the future if necessary, since Unix datagram sockets seem reliable in practice. --- diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index efbb21046..65e081638 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -398,16 +398,22 @@ sub wq_broadcast { croak "@exc" if @exc; } +sub sendmsg_eor ($$$;$) { + my $n = $send_cmd->($_[0], $_[1], $_[2], MSG_EOR, $_[3] // 50) // + return; + $n == length($_[2]) ? $n : croak('sendmsg('.length($_[2])." > $n)"); +} + sub stream_in_full ($$$) { my ($s1, $io, $buf) = @_; socketpair(my $r, my $w, AF_UNIX, SOCK_STREAM, 0); - my $n = $send_cmd->($s1, [ $r ], - ipc_freeze(['do_sock_stream', length($buf)]), - MSG_EOR) // croak "sendmsg: $!"; + my $n = sendmsg_eor($s1, [ $r ], + ipc_freeze(['do_sock_stream', length($buf)])) + // croak "sendmsg: $!"; undef $r; $n = $send_cmd->($w, $io, $buf, 0) // croak "sendmsg: $!"; print $w substr($buf, $n) if $n < length($buf); # need > 2G on Linux - close $w; # autodies + close $w; # autodies if print failed } sub wq_io_do { # always async @@ -416,9 +422,9 @@ sub wq_io_do { # always async my $buf = ipc_freeze([$sub, @args]); if (length($buf) > $MY_MAX_ARG_LEN) { stream_in_full($s1, $io, $buf); + } elsif (defined sendmsg_eor($s1, $io, $buf)) { + # success } else { - my $n = $send_cmd->($s1, $io, $buf, MSG_EOR); - return if defined($n); # likely $!{ETOOMANYREFS} and croak "sendmsg: $! (check RLIMIT_NOFILE)"; $!{EMSGSIZE} ? stream_in_full($s1, $io, $buf) : croak("sendmsg: $!"); @@ -458,13 +464,13 @@ sub wq_nonblock_do { # always async my $buf = ipc_freeze([$sub, @args]); if ($self->{wqb}) { # saturated once, assume saturated forever $self->{wqb}->flush_send($buf); - } elsif (!defined $send_cmd->($self->{-wq_s1}, [], $buf, MSG_EOR)) { - if ($!{EAGAIN} || $!{ENOBUFS} || $!{ENOMEM}) { - PublicInbox::WQBlocked->new($self, $buf); - } else { - croak "sendmsg: $!"; - } - } # else success + } elsif (defined sendmsg_eor($self->{-wq_s1}, [], $buf)) { + # success! + } elsif ($!{EAGAIN} || $!{ENOBUFS} || $!{ENOMEM}) { + PublicInbox::WQBlocked->new($self, $buf); + } else { + croak "sendmsg: $!"; + } } sub _wq_worker_start { diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index cc62c6cfb..9cf77fe21 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -11,7 +11,7 @@ use parent qw(PublicInbox::DS PublicInbox::LeiExternal PublicInbox::LeiQuery); use autodie qw(bind chdir listen open pipe socket socketpair syswrite unlink); use Getopt::Long (); -use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un MSG_EOR); +use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un); use Errno qw(EPIPE EAGAIN ECONNREFUSED ENOENT ECONNRESET EINTR); use Cwd qw(getcwd); use POSIX qw(strftime); @@ -1057,9 +1057,9 @@ sub start_mua { sub send_exec_cmd { # tell script/lei to execute a command my ($self, $io, $cmd, $env) = @_; - $PublicInbox::IPC::send_cmd->( + PublicInbox::IPC::sendmsg_eor( $self->{sock} // die('lei client gone'), - $io, exec_buf($cmd, $env), MSG_EOR) // + $io, exec_buf($cmd, $env)) // Carp::croak("sendmsg: $!"); } diff --git a/lib/PublicInbox/WQBlocked.pm b/lib/PublicInbox/WQBlocked.pm index e733f446f..d01d753cf 100644 --- a/lib/PublicInbox/WQBlocked.pm +++ b/lib/PublicInbox/WQBlocked.pm @@ -6,7 +6,6 @@ package PublicInbox::WQBlocked; use v5.12; use parent qw(PublicInbox::DS); use PublicInbox::Syscall qw(EPOLLOUT EPOLLONESHOT); -use Socket qw(MSG_EOR); use PublicInbox::IPC; use Carp (); @@ -22,13 +21,12 @@ sub flush_send { while (defined(my $buf = shift @{$self->{msgq}})) { if (ref($buf) eq 'CODE') { $buf->($self); # could be \&PublicInbox::DS::close + } elsif (defined(PublicInbox::IPC::sendmsg_eor( + $self->{sock}, [], $buf))) { + # success } else { - my $wq_s1 = $self->{sock}; - my $n = $PublicInbox::IPC::send_cmd->($wq_s1, [], $buf, - MSG_EOR); - next if defined($n); if ($!{EAGAIN}) { - PublicInbox::DS::epwait($wq_s1, + PublicInbox::DS::epwait($self->{sock}, EPOLLOUT|EPOLLONESHOT); } elsif ($!{ENOBUFS} || $!{ENOMEM}) { PublicInbox::DS::add_uniq_timer($self + 0, diff --git a/lib/PublicInbox/XapClient.pm b/lib/PublicInbox/XapClient.pm index 88638aed9..7f410ff97 100644 --- a/lib/PublicInbox/XapClient.pm +++ b/lib/PublicInbox/XapClient.pm @@ -9,7 +9,8 @@ package PublicInbox::XapClient; use v5.12; use PublicInbox::Spawn qw(spawn); -use Socket qw(AF_UNIX SOCK_SEQPACKET MSG_EOR); +use Carp qw(croak); +use Socket qw(AF_UNIX SOCK_SEQPACKET); use PublicInbox::IPC; use autodie qw(pipe socketpair); our $tries = -1; # set to zero by read-only daemon @@ -17,10 +18,8 @@ our $tries = -1; # set to zero by read-only daemon sub mkreq { my ($self, $io, @arg) = @_; my $buf = join("\0", @arg, ''); - my $n = $PublicInbox::IPC::send_cmd->($self->{io}, - $io, $buf, MSG_EOR, $tries) - // die "send_cmd: $!"; - $n == length($buf) or die "send_cmd: $n != ".length($buf); + PublicInbox::IPC::sendmsg_eor($self->{io}, $io, $buf, $tries) // + croak "sendmsg_eor: $!"; } sub start_helper (@) { diff --git a/t/lei-daemon.t b/t/lei-daemon.t index 0097c6f0a..f1b052ec2 100644 --- a/t/lei-daemon.t +++ b/t/lei-daemon.t @@ -2,15 +2,10 @@ # Copyright (C) all contributors # License: AGPL-3.0+ use strict; use v5.10.1; use PublicInbox::TestCommon; -use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un MSG_EOR); +use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un); +require PublicInbox::IPC; test_lei({ daemon_only => 1 }, sub { - my $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do { - require PublicInbox::Syscall; - PublicInbox::Syscall->can('send_cmd4'); - }; - $send_cmd or BAIL_OUT 'started testing lei-daemon w/o send_cmd4!'; - my $sock = "$ENV{XDG_RUNTIME_DIR}/lei/5.seq.sock"; my $err_log = "$ENV{XDG_RUNTIME_DIR}/lei/errors.log"; lei_ok('daemon-pid'); @@ -36,7 +31,8 @@ test_lei({ daemon_only => 1 }, sub { socket(my $c, AF_UNIX, SOCK_SEQPACKET, 0) or BAIL_OUT "socket: $!"; connect($c, $addr) or BAIL_OUT "connect: $!"; - $send_cmd->($c, [ $null, $null, $null ], 'hi', MSG_EOR); + PublicInbox::IPC::sendmsg_eor($c, + [ $null, $null, $null ], 'hi'); } lei_ok('daemon-pid'); chomp($pid = $lei_out); diff --git a/t/xap_helper.t b/t/xap_helper.t index 4e7b7f017..c5f1411f4 100644 --- a/t/xap_helper.t +++ b/t/xap_helper.t @@ -5,7 +5,7 @@ use v5.12; use PublicInbox::TestCommon; require_mods(qw(DBD::SQLite Xapian +SCM_RIGHTS)); # TODO: FIFO support? use PublicInbox::Spawn qw(spawn); -use Socket qw(AF_UNIX SOCK_SEQPACKET SOCK_STREAM MSG_EOR); +use Socket qw(AF_UNIX SOCK_SEQPACKET SOCK_STREAM); require PublicInbox::AutoReap; use PublicInbox::IPC; require PublicInbox::XapClient; @@ -91,8 +91,8 @@ my $doreq = sub { pipe(my $x, my $y); my $buf = join("\0", @arg, ''); my @io = ($y, $err); - my $n = $PublicInbox::IPC::send_cmd->($s, \@io, $buf, MSG_EOR) // - xbail "send: $!"; + my $n = PublicInbox::IPC::sendmsg_eor($s, \@io, $buf) // + xbail "sendmsg: $!"; my $exp = length($buf); $exp == $n or xbail "req @arg sent short ($n != $exp)"; $x;