paivana

early_response_upstream.c (15902B)

---

 /*
   This file is part of paivana tests.
   Copyright (C) 2026 Taler Systems SA
 
   Paivana is free software; you can redistribute it and/or
   modify it under the terms of the GNU Affero General Public License
   as published by the Free Software Foundation; either version
   3, or (at your option) any later version.
 
   Paivana is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty
   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
   the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public
   License along with Paivana; see the file COPYING.  If not,
   write to the Free Software Foundation, Inc., 51 Franklin
   Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
 
 /**
  * @file early_response_upstream.c
  * @brief Raw-socket HTTP upstream that responds before reading the
  *        request body.  Used to exercise paivana's handling of an
  *        early upstream response: when the upstream answers while
  *        the client (paivana) is still uploading, paivana must
  *        deliver that response to its own client rather than treat
  *        the interrupted forward as a failure.
  *
  *        After each connection finishes the server writes the number
  *        of body bytes it received (decimal, trailing newline) to the
  *        file given as the second argument, atomically via rename(2).
  *        That count is a diagnostic, not a contract: RFC 9110 §9.3
  *        lets a client stop sending its body once it has a final
  *        response, and libcurl does exactly that, so the upstream
  *        normally sees only a prefix of what was sent.
  *
  *        With `--no-drain` the server instead *stops* reading once it
  *        has answered, and parks the connection with the rest of the
  *        request still queued in its receive buffer.  That is the
  *        nastier half of the same situation: paivana's outbound
  *        socket backs up and its write() can no longer make progress,
  *        so it can only finish the request if it acts on the response
  *        that is already in hand rather than waiting for the upload
  *        to drain.  A proxy that gets this wrong hangs until its
  *        transfer timeout instead of failing outright, which is why
  *        the driver bounds the case with timeout(1).
  */
 #ifndef _GNU_SOURCE
 /* for POLLRDHUP, which is how we notice the peer hanging up
    without ever reading from the socket */
 #define _GNU_SOURCE
 #endif
 #include "platform.h"
 #include <gnunet/gnunet_util_lib.h>
 #include <limits.h>
 #include <poll.h>
 #include <strings.h>
 
 /**
  * How long (ms) a `--no-drain` connection is parked before we give up
  * on the peer ever hanging up.  Only reached when paivana is stuck,
  * and the driver's timeout(1) will have failed the case by then; the
  * bound is here so a wedged run still terminates.
  */
 #define PARK_TIMEOUT_MS 30000
 
 /**
  * Receive buffer (bytes) requested for `--no-drain` connections.  The
  * kernel doubles this and enforces its own floor; the point is only
  * to keep it far below the body the driver sends, so that refusing to
  * read reliably stalls the sender no matter how TCP autotuning is
  * configured on the machine running the tests.
  */
 #define NO_DRAIN_RCVBUF 8192
 
 
 static const char *receipt_path;
 static volatile sig_atomic_t run_flag = 1;
 
 /**
  * Keep reading the request body after responding?  False under
  * `--no-drain`, where refusing to read is the whole point.
  */
 static bool drain_body = true;
 
 
 static void
 on_sig (int sig)
 {
   (void) sig;
   run_flag = 0;
 }
 
 
 /**
  * Read from @a fd into @a buf until "\r\n\r\n" appears.  Returns the
  * total number of bytes read (headers + any body bytes carried in the
  * same recv) on success, with @a *eoh set to the offset of the first
  * body byte (i.e. just past the CRLFCRLF).  Returns -1 on read error
  * or if @a buf fills up before headers end.
  */
 static ssize_t
 read_until_eoh (int fd,
                 char *buf,
                 size_t cap,
                 size_t *eoh)
 {
   size_t pos = 0;
   /* First offset not yet examined.  Carried across the read loop:
      restarting the scan at 0 after every read would be quadratic in
      the size of the header block. */
   size_t scan = 0;
 
   while (pos < cap)
   {
     ssize_t n = read (fd,
                       buf + pos,
                       cap - pos);
     if (n <= 0)
       return -1;
     pos += (size_t) n;
     while (scan + 3 < pos)
     {
       if ( ('\r' == buf[scan]) &&
            ('\n' == buf[scan + 1]) &&
            ('\r' == buf[scan + 2]) &&
            ('\n' == buf[scan + 3]) )
       {
         *eoh = scan + 4;
         return (ssize_t) pos;
       }
       scan++;
     }
   }
   return -1;
 }
 
 
 /**
  * Find the Content-Length announced in the request header block held
  * in the first @a len bytes of @a hdr.
  *
  * We need it because we cannot rely on end-of-file to tell us the
  * request is over: the peer is free to keep a completed connection in
  * a pool, and then no EOF ever comes.
  *
  * @param hdr start of the (not NUL-terminated) header block
  * @param len number of bytes in @a hdr
  * @return the announced length, or -1 if there was none
  */
 static long long
 find_content_length (const char *hdr,
                      size_t len)
 {
   static const char name[] = "content-length:";
   const size_t nlen = sizeof (name) - 1;
 
   for (size_t i = 0; i + nlen <= len; i++)
   {
     size_t j;
     long long v = 0;
     bool digits = false;
 
     if ( (0 == i) ||
          ('\n' != hdr[i - 1]) )
       continue; /* header names start a line */
     if (0 != strncasecmp (&hdr[i],
                           name,
                           nlen))
       continue;
     j = i + nlen;
     while ( (j < len) &&
             ( (' ' == hdr[j]) || ('\t' == hdr[j]) ) )
       j++;
     while ( (j < len) &&
             ('0' <= hdr[j]) &&
             ('9' >= hdr[j]) )
     {
       /* Signed overflow is undefined, and the value comes off the
          wire: refuse a length we cannot represent rather than wrap
          into a negative (or, with optimisation, into anything at
          all).  We have no use for one either -- the caller reads at
          most what it announced. */
       if (v > (LLONG_MAX - (hdr[j] - '0')) / 10)
         return -1;
       v = v * 10 + (hdr[j] - '0');
       digits = true;
       j++;
     }
     return digits ? v : -1;
   }
   return -1;
 }
 
 
 /**
  * Atomically write @a count to the receipt file (write to <path>.tmp,
  * fsync, rename).  No-op if no receipt path was configured.
  */
 static void
 write_receipt (size_t count)
 {
   FILE *f;
   char tmp[4096];
   int rc;
 
   if (NULL == receipt_path)
     return;
   rc = snprintf (tmp,
                  sizeof (tmp),
                  "%s.tmp",
                  receipt_path);
   if ( (rc < 0) ||
        ((size_t) rc >= sizeof (tmp)) )
     return;
   f = fopen (tmp, "w");
   if (NULL == f)
     return;
   fprintf (f, "%zu\n", count);
   fflush (f);
   fsync (fileno (f));
   fclose (f);
   if (0 != rename (tmp, receipt_path))
     unlink (tmp);
 }
 
 
 /**
  * The early response.  Deliberately *without* `Connection: close`:
  * announcing a close makes libcurl give up on the rest of the request
  * body it was sending, which is legitimate but would rob the
  * drain-mode test of the very thing it checks.
  */
 static const char EARLY_RESPONSE[] =
   "HTTP/1.1 413 Payload Too Large\r\n"
   "Content-Type: text/plain\r\n"
   "X-Upstream: early\r\n"
   "Content-Length: 23\r\n"
   "\r\n"
   "early-response-payload\n";
 
 
 /**
  * Best-effort send the entire @a buf, retrying on partial writes.
  * Returns 0 on success, -1 if the socket failed.
  */
 static int
 write_all (int fd,
            const char *buf,
            size_t len)
 {
   size_t off = 0;
 
   while (off < len)
   {
     ssize_t w = write (fd,
                        buf + off,
                        len - off);
     if (w < 0)
     {
       if (EINTR == errno)
         continue;
       return -1;
     }
     if (0 == w)
       return -1;
     off += (size_t) w;
   }
   return 0;
 }
 
 
 /**
  * Wait for the peer to shut down its sending side of @a fd without
  * ever reading from the socket, so that the unconsumed request body
  * keeps the peer's send buffer full.  Gives up after
  * #PARK_TIMEOUT_MS.
  *
  * @param fd the connection to park
  */
 static void
 park_until_hangup (int fd)
 {
   struct pollfd pfd = {
     .fd = fd,
     /* Deliberately not POLLIN: we do not want to know that body
        bytes arrived, only that the peer is done sending them. */
     .events = POLLRDHUP
   };
   int elapsed = 0;
 
   while (elapsed < PARK_TIMEOUT_MS)
   {
     int rv;
 
     if (! run_flag)
       return; /* shutting down, nothing to report */
     rv = poll (&pfd,
                1,
                250);
     if (rv < 0)
     {
       if (EINTR == errno)
         continue;
       return;
     }
     if (0 != (pfd.revents & (POLLRDHUP | POLLHUP | POLLERR)))
       return;
     elapsed += 250;
   }
   fprintf (stderr,
            "no-drain connection parked for %d ms without a hangup;"
            " the peer is most likely stuck\n",
            elapsed);
   fflush (stderr);
 }
 
 
 static void
 handle_client (int fd)
 {
   char buf[8192];
   size_t eoh = 0;
   ssize_t got;
   size_t body_bytes;
   long long content_length;
 
   got = read_until_eoh (fd,
                         buf,
                         sizeof (buf),
                         &eoh);
   if (got < 0)
   {
     close (fd);
     return;
   }
   body_bytes = (size_t) got - eoh;
   content_length = find_content_length (buf,
                                         eoh);
 
   /* Send the 413 response immediately.  This is the whole point
      of this upstream: respond before consuming the request body. */
   if (0 != write_all (fd,
                       EARLY_RESPONSE,
                       sizeof (EARLY_RESPONSE) - 1))
   {
     close (fd);
     /* Receipt still useful for diagnostics: how far did we get? */
     write_receipt (body_bytes);
     return;
   }
 
   if (! drain_body)
   {
     /* The interesting case: having answered, refuse to read another
        byte.  The rest of the request stays in our receive buffer,
        our window closes, and paivana's send() stops making progress
        with the response already sitting unread on its side.  Park
        until the peer hangs up (which is what a proxy that handles
        this correctly does) rather than closing ourselves: closing
        with unread data queued would send an RST, and the peer would
        be entitled to discard the response we just sent -- turning a
        proxy bug into a transport error and vice versa. */
     park_until_hangup (fd);
     close (fd);
     write_receipt (body_bytes);
     return;
   }
 
   /* Now drain the rest of the request body.  We must keep reading
      until the announced Content-Length is in (or the client closes,
      for a request that had none) — otherwise its CURL would block on
      send() once the kernel recv buffer fills, which would mask the
      very behavior we are trying to test.  Stopping at
      Content-Length rather than at EOF matters: having answered the
      request in full we have given the peer every right to keep the
      connection for reuse, and then the EOF never comes. */
   while ( (content_length < 0) ||
           (body_bytes < (size_t) content_length) )
   {
     ssize_t n = read (fd,
                       buf,
                       sizeof (buf));
     if (n < 0)
     {
       if (EINTR == errno)
         continue;
       break;
     }
     if (0 == n)
       break;
     body_bytes += (size_t) n;
   }
 
   close (fd);
   write_receipt (body_bytes);
 }
 
 
 /**
  * Parse @a arg as a TCP port number.  Rejects trailing garbage and
  * anything outside 1-65535; port 0 in particular is refused, as we
  * would have no way to tell the caller which port the kernel picked.
  *
  * @param arg command-line argument to parse
  * @param[out] port set to the parsed port on success
  * @return 0 on success, -1 if @a arg is not a valid port
  */
 static int
 parse_port (const char *arg,
             int *port)
 {
   char *end;
   long long v;
 
   errno = 0;
   v = strtoll (arg,
                &end,
                10);
   if ( (0 != errno) ||
        (end == arg) ||
        ('\0' != *end) ||
        (v < 1) ||
        (v > 65535) )
     return -1;
   *port = (int) v;
   return 0;
 }
 
 
 int
 main (int argc,
       char **argv)
 {
   int srv;
   int port = 0;
   int yes = 1;
   int have_port = 0;
   struct sockaddr_in addr;
 
   for (int i = 1; i < argc; i++)
   {
     if (0 == strcmp (argv[i],
                      "--no-drain"))
     {
       drain_body = false;
       continue;
     }
     if (! have_port)
     {
       if (0 != parse_port (argv[i],
                            &port))
       {
         fprintf (stderr,
                  "invalid port `%s'\n",
                  argv[i]);
         return 1;
       }
       have_port = 1;
       continue;
     }
     if (NULL == receipt_path)
     {
       receipt_path = argv[i];
       continue;
     }
     fprintf (stderr,
              "unexpected argument `%s'\n",
              argv[i]);
     have_port = 0;
     break;
   }
   if (! have_port)
   {
     fprintf (stderr,
              "usage: %s <port> [receipt-file] [--no-drain]\n",
              argv[0]);
     return 1;
   }
 
   {
     /* sigaction(2), not signal(3): glibc's signal() implies
        SA_RESTART, which would restart the accept(2) we spend most of
        our life in and leave us ignoring SIGTERM forever.  We need the
        EINTR so the loops below can notice `run_flag'. */
     struct sigaction sa = {
       .sa_handler = &on_sig,
       .sa_flags = 0
     };
 
     sigemptyset (&sa.sa_mask);
     sigaction (SIGINT, &sa, NULL);
     sigaction (SIGTERM, &sa, NULL);
   }
   signal (SIGPIPE, SIG_IGN);
 
   srv = socket (AF_INET, SOCK_STREAM, 0);
   if (srv < 0)
   {
     perror ("socket");
     return 1;
   }
   /* Only fails on a bad fd/level/optname, i.e. never here; the
      assertion is to keep a silent failure from turning into a
      spurious EADDRINUSE two test cases later. */
   GNUNET_assert (0 == setsockopt (srv,
                                   SOL_SOCKET,
                                   SO_REUSEADDR,
                                   &yes,
                                   sizeof (yes)));
   if (! drain_body)
   {
     int rcvbuf = NO_DRAIN_RCVBUF;
 
     /* Set on the listening socket so accepted connections inherit
        it, and before listen(2) so it is in effect for the SYN-ACK
        window.  Without this the peer could park a multi-megabyte
        autotuned window's worth of body with us and finish its
        write() after all, which would quietly turn the no-drain case
        back into the drain case. */
     if (0 != setsockopt (srv,
                          SOL_SOCKET,
                          SO_RCVBUF,
                          &rcvbuf,
                          sizeof (rcvbuf)))
       perror ("setsockopt(SO_RCVBUF)"); /* not fatal, just less sharp */
   }
   memset (&addr, 0, sizeof (addr));
   addr.sin_family = AF_INET;
   addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
   addr.sin_port = htons ((uint16_t) port);
   if (0 != bind (srv,
                  (struct sockaddr *) &addr,
                  sizeof (addr)))
   {
     perror ("bind");
     close (srv);
     return 1;
   }
   if (0 != listen (srv, 16))
   {
     perror ("listen");
     close (srv);
     return 1;
   }
   fprintf (stderr,
            "early_response_upstream listening on port %d (%s)\n",
            port,
            drain_body
            ? "draining the body after responding"
            : "refusing to read the body after responding");
   fflush (stderr);
 
   while (run_flag)
   {
     int cfd = accept (srv, NULL, NULL);
 
     if (cfd < 0)
     {
       if (EINTR == errno)
         continue;
       perror ("accept");
       break;
     }
     /* Single-threaded: serialise connections.  The reverse-proxy
        test makes one request per case and Connection: close keeps
        paivana from reusing the socket. */
     handle_client (cfd);
   }
   close (srv);
   return 0;
 }