Merge branch 'feat/transport-g1' into feat/transport-xattr

src/shared/protocol.c

@@ -99,8 +99,10 @@ static ssize_t tls_io_send(ProtocolSession* session, const void* data, size_t si
       return PROTOCOL_IO_RETRY;
     }
     /* A signal (e.g. Ctrl-C) interrupts the blocking TLS write: retry so the
-     * send loop can observe the abort flag at the next checkpoint. */
-    if (ssl_err == SSL_ERROR_SYSCALL && errno == EINTR)
+     * send loop can observe the abort flag at the next checkpoint.  Only an
+     * actual negative return is an interrupted syscall; a 0-byte SSL_write is
+     * not a valid EINTR retry. */
+    if (written < 0 && ssl_err == SSL_ERROR_SYSCALL && errno == EINTR)
       return PROTOCOL_IO_RETRY;
     return PROTOCOL_IO_ERROR;
   }
@@ -125,8 +127,13 @@ static ssize_t tls_io_recv(ProtocolSession* session, void* data, size_t size, sh
       return PROTOCOL_IO_RETRY;
     }
     /* A signal interrupts the blocking TLS read: retry (mirrors the send path)
-     * so the loop reaches its next abort/deadline checkpoint. */
-    if (ssl_err == SSL_ERROR_SYSCALL && errno == EINTR)
+     * so the loop reaches its next abort/deadline checkpoint.  Only an actual
+     * negative return is an interrupted syscall: a 0-byte SSL_read is an
+     * unexpected EOF (the peer closed without close_notify), which OpenSSL also
+     * reports as SSL_ERROR_SYSCALL with errno possibly still EINTR from an
+     * earlier interrupted poll/read.  Retrying that would busy-spin the
+     * status-read loop until its deadline, so classify it as closed instead. */
+    if (received < 0 && ssl_err == SSL_ERROR_SYSCALL && errno == EINTR)
       return PROTOCOL_IO_RETRY;
     /* A zero-length SSL_read is the peer's clean close_notify (or EOF without
      * one); report it distinctly so the caller can log it as a close. */
@@ -396,9 +403,12 @@ SSL* protocol_current_ssl(void) {
   /* The bound session is the authoritative transport for a worker thread: it
    * was explicitly handed to protocol_session_bind() and carries its own SSL,
    * whereas io_ssl is thread-local and NULL in a thread that never performed
-   * the handshake.  With no session bound (the fd-shim path), fall back to the
-   * legacy thread-local SSL. */
-  if (bound_session)
+   * the handshake.  Only a session whose selected dispatch is TLS may supply
+   * the SSL: a bound plaintext session has ssl == NULL and must not shadow a
+   * live thread-local io_ssl, or file_send.c would take the raw sendfile(2)
+   * path on a socket this thread is driving with TLS.  With no TLS session
+   * bound (plaintext session, or the fd-shim path), fall back to io_ssl. */
+  if (bound_session && bound_session->ops == &tls_io_ops && bound_session->ssl)
     return bound_session->ssl;
   return io_ssl;
 }

src/shared/protocol.h

@@ -249,13 +249,15 @@ unsigned long long io_get_bwlimit(void);
 void io_set_ssl(SSL* ssl);
 SSL* io_get_ssl(void);
 /* SSL object of the transport in effect on this thread: the currently bound
- * session's SSL when a session is bound, otherwise the legacy thread-local
+ * session's SSL when a TLS session is bound, otherwise the legacy thread-local
  * io_ssl.  NULL for a plaintext transport.  Unlike io_get_ssl(), this resolves
  * worker threads that bound a TLS session via protocol_session_set_ssl()/
  * protocol_session_bind() but never called io_set_ssl() themselves (C11
- * _Thread_local state is not inherited by a new thread).  Callers that must
- * choose a TLS-only code path (e.g. file_send.c's sendfile fallback) must use
- * this instead of io_get_ssl(). */
+ * _Thread_local state is not inherited by a new thread).  A bound session only
+ * wins when its selected dispatch is TLS; a bound plaintext session (ssl ==
+ * NULL) falls back to io_ssl so it can never mask a live encrypted transport.
+ * Callers that must choose a TLS-only code path (e.g. file_send.c's sendfile
+ * fallback) must use this instead of io_get_ssl(). */
 SSL* protocol_current_ssl(void);
 
 /* Process-wide wire byte counters.  protocol_send_n_data/protocol_receive_n_data

tests/integration/test_tls.py

@@ -146,8 +146,16 @@ class TestTLSBasic:
         assert not missing, f"Missing files: {missing}"
         assert not mismatches, f"Mismatched files: {mismatches}"
 
+    @pytest.mark.ci
     def test_tls_with_multithreading(self, certs):
-        """TLS + multithreading."""
+        """TLS + multithreading + --sendfile.
+
+        Exercises the TLS/sendfile interaction end to end: with --sendfile the
+        sender must route the file body through the buffered TLS path rather
+        than raw sendfile(2) on the encrypted socket.  The focused decision
+        guard lives in tests/test_protocol.c
+        (test_tls_sendfile_decision_uses_buffered_path).
+        """
         clean_dir(DEST_DIR)
         with ServerManager() as server:
             server.start(extra_args=[
@@ -156,7 +164,7 @@ class TestTLSBasic:
             ])
             result, dur = run_client(
                 SOURCE_DIR, DEST_DIR,
-                flags=["--threads", "--tls",
+                flags=["--threads", "--sendfile", "--tls",
                        "--cert", certs["client_cert"], "--key", certs["client_key"],
                        "--ca", certs["ca"]],
                 port=server.port,

tests/test_protocol.c

@@ -1,9 +1,13 @@
 #include "protocol.h"
+#include "file.h"
 #include "test_utils.h"
+#include "utils.h"
 #include <errno.h>
 #include <fcntl.h>
 #include <limits.h>
+#include <openssl/evp.h>
 #include <openssl/ssl.h>
+#include <openssl/x509.h>
 #include <poll.h>
 #include <stdlib.h>
 #include <string.h>
@@ -865,6 +869,106 @@ static void test_protocol_dispatch_via_ops() {
   close(sv[1]);
 }
 
+/* Retry-contract tests: an op that reports PROTOCOL_IO_RETRY once (and hands the
+ * loop a switched wait event) must be retried rather than treated as a fatal
+ * error or a close.  The send/receive loops had no unit coverage for this path
+ * even though every TLS WANT_READ/WANT_WRITE and EINTR retry relies on it. */
+static int retry_send_calls;
+static short retry_send_last_wait;
+static int retry_recv_calls;
+static short retry_recv_last_wait;
+
+static ssize_t retry_once_send(ProtocolSession* session, const void* data, size_t size,
+                               short* wait_events) {
+  retry_send_calls++;
+  if (retry_send_calls == 1) {
+    /* Simulate a WANT_READ-style retry: switch the poll event and make no
+     * progress.  The send loop must consume this and retry. */
+    *wait_events = POLLIN;
+    return PROTOCOL_IO_RETRY;
+  }
+  ssize_t written = write(session->write_fd, data, size);
+  if (written < 0)
+    return PROTOCOL_IO_ERROR;
+  if (written == 0)
+    return PROTOCOL_IO_ERROR;
+  *wait_events = POLLOUT;
+  retry_send_last_wait = *wait_events;
+  return written;
+}
+
+static ssize_t retry_once_recv(ProtocolSession* session, void* data, size_t size,
+                               short* wait_events) {
+  retry_recv_calls++;
+  if (retry_recv_calls == 1) {
+    *wait_events = POLLOUT;
+    return PROTOCOL_IO_RETRY;
+  }
+  ssize_t received = read(session->read_fd, data, size);
+  if (received < 0)
+    return PROTOCOL_IO_ERROR;
+  if (received == 0)
+    return PROTOCOL_IO_CLOSED;
+  *wait_events = POLLIN;
+  retry_recv_last_wait = *wait_events;
+  return received;
+}
+
+static const ProtocolIoOps retry_send_ops = {
+    .send = retry_once_send,
+    .recv = counting_recv,
+    .has_pending = counting_has_pending,
+};
+
+static const ProtocolIoOps retry_recv_ops = {
+    .send = counting_send,
+    .recv = retry_once_recv,
+    .has_pending = counting_has_pending,
+};
+
+static void test_protocol_io_retry_contract() {
+  const char payload[] = "retry-contract";
+
+  /* The send loop: the first attempt reports RETRY and switches the poll event
+   * to POLLIN.  A pre-seeded readable byte on the *opposite* end of the
+   * socketpair keeps that poll immediately satisfiable, so the retry is the
+   * only thing under test. */
+  int send_sv[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, send_sv), 0);
+  char seed = 'x';
+  EXPECT_EQ_INT(write(send_sv[1], &seed, 1), 1);
+  ProtocolSession sender;
+  protocol_session_init(&sender, send_sv[0], send_sv[0]);
+  protocol_session_set_bwlimit(&sender, 0);
+  sender.ops = &retry_send_ops;
+  retry_send_calls = 0;
+  retry_send_last_wait = 0;
+  EXPECT_TRUE(protocol_send_n_data(&sender, payload, sizeof(payload)));
+  EXPECT_EQ_INT(retry_send_calls, 2);
+  EXPECT_EQ_INT(retry_send_last_wait, POLLOUT);
+  close(send_sv[0]);
+  close(send_sv[1]);
+
+  /* The receive loop: the first attempt reports RETRY and switches the poll
+   * event to POLLOUT, which a socketpair read fd satisfies immediately. */
+  int recv_sv[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, recv_sv), 0);
+  EXPECT_EQ_INT((int)write(recv_sv[0], payload, sizeof(payload)), (int)sizeof(payload));
+  ProtocolSession receiver;
+  protocol_session_init(&receiver, recv_sv[1], recv_sv[1]);
+  protocol_session_set_bwlimit(&receiver, 0);
+  receiver.ops = &retry_recv_ops;
+  retry_recv_calls = 0;
+  retry_recv_last_wait = 0;
+  char received[sizeof(payload)] = {0};
+  EXPECT_TRUE(protocol_receive_n_data(&receiver, received, sizeof(received)));
+  EXPECT_EQ_INT(memcmp(payload, received, sizeof(payload)), 0);
+  EXPECT_EQ_INT(retry_recv_calls, 2);
+  EXPECT_EQ_INT(retry_recv_last_wait, POLLIN);
+  close(recv_sv[0]);
+  close(recv_sv[1]);
+}
+
 typedef struct {
   ProtocolSession* session;
   SSL* expected_ssl;
@@ -894,7 +998,14 @@ static void test_protocol_current_ssl_prefers_bound_session() {
   EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
   ProtocolSession session;
   protocol_session_init(&session, sv[0], sv[0]);
+  const ProtocolIoOps* plain_ops = session.ops;
   protocol_session_set_ssl(&session, ssl);
+  /* set_ssl must select a distinct (TLS) dispatch table; protocol_current_ssl
+   * only returns a bound session's SSL for TLS ops, so arg.resolved_ssl == ssl
+   * below also proves the bound session's ops are the TLS ops. */
+  EXPECT_NOT_NULL(plain_ops);
+  EXPECT_TRUE(session.ops != plain_ops);
+  EXPECT_TRUE(session.ssl == ssl);
 
   /* Clear the calling thread's legacy SSL: only the bound session carries it. */
   io_set_fds(-1, -1);
@@ -904,6 +1015,7 @@ static void test_protocol_current_ssl_prefers_bound_session() {
   thrd_t worker;
   EXPECT_EQ_INT(thrd_create(&worker, ssl_resolver_worker, &arg), thrd_success);
   EXPECT_EQ_INT(thrd_join(worker, NULL), thrd_success);
+  EXPECT_TRUE(arg.resolved_ssl == arg.expected_ssl);
   EXPECT_TRUE(arg.resolved_ssl == ssl);
   EXPECT_NULL(arg.thread_local_ssl);
 
@@ -913,6 +1025,221 @@ static void test_protocol_current_ssl_prefers_bound_session() {
   SSL_CTX_free(ctx);
 }
 
+/* A bound plaintext session must NOT mask a live thread-local TLS transport:
+ * protocol_current_ssl() only trusts a bound session whose dispatch is TLS, so
+ * it falls back to io_ssl here.  This is the safe direction for the sendfile
+ * decision -- returning NULL would let file_send.c take raw sendfile(2) on a
+ * socket this thread is encrypting. */
+static void test_protocol_current_ssl_plaintext_bound_falls_back() {
+  SSL_CTX* ctx = SSL_CTX_new(TLS_method());
+  EXPECT_NOT_NULL(ctx);
+  SSL* ssl = SSL_new(ctx);
+  EXPECT_NOT_NULL(ssl);
+
+  int sv[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  /* Live thread-local TLS, then a bound plaintext session: the plaintext
+   * session's NULL ssl must not shadow the encrypted transport. */
+  io_set_ssl(ssl);
+  ProtocolSession plain;
+  protocol_session_init(&plain, sv[0], sv[0]);
+  protocol_session_bind(&plain);
+  EXPECT_TRUE(protocol_current_ssl() == ssl);
+  protocol_session_unbind();
+
+  /* A bound TLS session still wins over a different thread-local TLS object. */
+  SSL* other = SSL_new(ctx);
+  EXPECT_NOT_NULL(other);
+  io_set_ssl(other);
+  ProtocolSession tls;
+  protocol_session_init(&tls, sv[0], sv[0]);
+  protocol_session_set_ssl(&tls, ssl);
+  protocol_session_bind(&tls);
+  EXPECT_TRUE(protocol_current_ssl() == ssl);
+  EXPECT_TRUE(protocol_current_ssl() != other);
+  protocol_session_unbind();
+
+  io_set_fds(-1, -1);
+  close(sv[0]);
+  close(sv[1]);
+  SSL_free(other);
+  SSL_free(ssl);
+  SSL_CTX_free(ctx);
+}
+
+/* ------------------------------------------------------------------------- *
+ * Genuine TLS + sendfile regression test.
+ *
+ * file_send_sendfile_with_skip() must route a TLS transfer through the
+ * buffered SSL path, resolved from the bound session, even in a worker thread
+ * whose thread-local io_ssl was never installed.  This drives a real TLS
+ * handshake between two in-memory endpoints and calls the production
+ * file_send entry from a worker that bound a TLS session only: if the sendfile
+ * decision regresses to io_get_ssl() it sees NULL, takes raw sendfile(2), and
+ * copies the file's plaintext into the encrypted stream, so the peer's final
+ * SSL_read here fails.  A tautology-free end-to-end decision guard.
+ * ------------------------------------------------------------------------- */
+
+static void test_set_fd_nonblocking(int fd) {
+  int flags = fcntl(fd, F_GETFL, 0);
+  if (flags != -1)
+    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+}
+
+static SSL_CTX* test_tls_context_with_self_signed_cert(void) {
+  EVP_PKEY* key = EVP_PKEY_new();
+  EVP_PKEY_CTX* key_ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_RSA, NULL);
+  if (!key || !key_ctx) {
+    EVP_PKEY_free(key);
+    EVP_PKEY_CTX_free(key_ctx);
+    return NULL;
+  }
+  bool key_ok = EVP_PKEY_keygen_init(key_ctx) == 1 &&
+                EVP_PKEY_CTX_set_rsa_keygen_bits(key_ctx, 2048) == 1 &&
+                EVP_PKEY_keygen(key_ctx, &key) == 1;
+  EVP_PKEY_CTX_free(key_ctx);
+
+  X509* cert = key_ok ? X509_new() : NULL;
+  bool cert_ok = cert != NULL && X509_set_version(cert, 2) == 1 &&
+                 ASN1_INTEGER_set(X509_get_serialNumber(cert), 1) == 1 &&
+                 X509_gmtime_adj(X509_getm_notBefore(cert), 0) != NULL &&
+                 X509_gmtime_adj(X509_getm_notAfter(cert), 3600) != NULL &&
+                 X509_set_pubkey(cert, key) == 1;
+  if (cert_ok) {
+    X509_NAME* name = X509_get_subject_name(cert);
+    cert_ok = X509_NAME_add_entry_by_txt(name, "CN", MBSTRING_ASC, (unsigned char*)"localhost", -1,
+                                         -1, 0) == 1 &&
+              X509_set_issuer_name(cert, name) == 1 && X509_sign(cert, key, EVP_sha256()) > 0;
+  }
+
+  SSL_CTX* ctx = cert_ok ? SSL_CTX_new(TLS_method()) : NULL;
+  bool installed = ctx != NULL && SSL_CTX_use_certificate(ctx, cert) == 1 &&
+                   SSL_CTX_use_PrivateKey(ctx, key) == 1;
+  if (ctx && !installed) {
+    SSL_CTX_free(ctx);
+    ctx = NULL;
+  }
+  if (ctx)
+    SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL);
+
+  X509_free(cert);
+  EVP_PKEY_free(key);
+  return ctx;
+}
+
+static bool test_tls_pump_handshake(SSL* ssl, int* done) {
+  int result = SSL_do_handshake(ssl);
+  if (result == 1) {
+    *done = 1;
+    return true;
+  }
+  int err = SSL_get_error(ssl, result);
+  return err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE;
+}
+
+static bool test_tls_read_exact(SSL* ssl, void* out, size_t size) {
+  char* bytes = out;
+  size_t got = 0;
+  while (got < size) {
+    int result = SSL_read(ssl, bytes + got, (int)(size - got));
+    if (result > 0) {
+      got += (size_t)result;
+      continue;
+    }
+    int err = SSL_get_error(ssl, result);
+    if (err != SSL_ERROR_WANT_READ && err != SSL_ERROR_WANT_WRITE)
+      return false;
+    struct pollfd pfd = {.fd = SSL_get_fd(ssl),
+                         .events = err == SSL_ERROR_WANT_READ ? POLLIN : POLLOUT};
+    if (poll(&pfd, 1, 5000) <= 0)
+      return false;
+  }
+  return true;
+}
+
+typedef struct {
+  ProtocolSession* session;
+  File* file;
+  int fd;
+  bool ok;
+} TlsSendfileWorkerArg;
+
+static int tls_sendfile_worker(void* arg) {
+  TlsSendfileWorkerArg* worker = arg;
+  /* Deliberately never call io_set_ssl(): the bound session is the only
+   * transport this thread has, exactly like a worker in the -m pipeline. */
+  protocol_session_bind(worker->session);
+  worker->ok =
+      file_send_sendfile_with_skip(worker->file, worker->fd, false, 0, false, NULL, 0, 0, false);
+  protocol_session_unbind();
+  return thrd_success;
+}
+
+static void test_tls_sendfile_decision_uses_buffered_path() {
+  const char content[] = "tls-sendfile-regression-payload";
+  const char* path = "test_tls_sendfile_regression.bin";
+  EXPECT_TRUE(file_write_to_disk(path, content, sizeof(content), false, false));
+
+  SSL_CTX* ctx = test_tls_context_with_self_signed_cert();
+  EXPECT_NOT_NULL(ctx);
+  SSL* server_ssl = SSL_new(ctx);
+  SSL* client_ssl = SSL_new(ctx);
+  EXPECT_NOT_NULL(server_ssl);
+  EXPECT_NOT_NULL(client_ssl);
+
+  int sv[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+  test_set_fd_nonblocking(sv[0]);
+  test_set_fd_nonblocking(sv[1]);
+  EXPECT_EQ_INT(SSL_set_fd(server_ssl, sv[0]), 1);
+  EXPECT_EQ_INT(SSL_set_fd(client_ssl, sv[1]), 1);
+  SSL_set_accept_state(server_ssl);
+  SSL_set_connect_state(client_ssl);
+
+  int server_done = 0;
+  int client_done = 0;
+  for (int i = 0; i < 1000 && !(server_done && client_done); i++) {
+    bool server_ok = server_done || test_tls_pump_handshake(server_ssl, &server_done);
+    bool client_ok = client_done || test_tls_pump_handshake(client_ssl, &client_done);
+    if (!server_ok || !client_ok)
+      break;
+  }
+  EXPECT_TRUE(server_done && client_done);
+
+  File* file = file_create(path);
+  EXPECT_NOT_NULL(file);
+  file->data->size = sizeof(content);
+
+  ProtocolSession session;
+  protocol_session_init(&session, sv[0], sv[0]);
+  protocol_session_set_bwlimit(&session, 0);
+  protocol_session_set_ssl(&session, server_ssl);
+
+  TlsSendfileWorkerArg arg = {.session = &session, .file = file, .fd = sv[0], .ok = false};
+  thrd_t worker;
+  EXPECT_EQ_INT(thrd_create(&worker, tls_sendfile_worker, &arg), thrd_success);
+  EXPECT_EQ_INT(thrd_join(worker, NULL), thrd_success);
+  EXPECT_TRUE(arg.ok);
+
+  /* The peer must be able to decrypt the whole framing: size header and the
+   * file body, both produced through the TLS transport. */
+  unsigned long long wire_size = 0;
+  EXPECT_TRUE(test_tls_read_exact(client_ssl, &wire_size, sizeof(wire_size)));
+  EXPECT_EQ_INT((int)wire_size, (int)sizeof(content));
+  char received[sizeof(content)] = {0};
+  EXPECT_TRUE(test_tls_read_exact(client_ssl, received, sizeof(received)));
+  EXPECT_EQ_INT(memcmp(received, content, sizeof(content)), 0);
+
+  file_destroy(file);
+  close(sv[0]);
+  close(sv[1]);
+  SSL_free(server_ssl);
+  SSL_free(client_ssl);
+  SSL_CTX_free(ctx);
+  unlink(path);
+}
+
 void test_protocol() {
   test_send_receive_n_data();
   test_send_receive_n_data_zero();
@@ -947,5 +1274,8 @@ void test_protocol() {
   test_protocol_throttle_bytes_unlimited();
   test_protocol_throttle_bytes_legacy_same_session();
   test_protocol_dispatch_via_ops();
+  test_protocol_io_retry_contract();
   test_protocol_current_ssl_prefers_bound_session();
+  test_protocol_current_ssl_plaintext_bound_falls_back();
+  test_tls_sendfile_decision_uses_buffered_path();
 }