Send binary WebSocket frames in one write

tests/web_cookie_auth/run.py

@@ -20,6 +20,7 @@ from serial_headers import SERIAL_HEADERS
 
 HEADERS.update(SERIAL_HEADERS)
 HEADERS["esp_http_server.h"] += """
+#define HTTPD_SOCK_ERR_FAIL -1
 #define ESP_ERR_HTTPD_INVALID_REQ 0x200
 #define ESP_ERR_HTTPD_RESULT_TRUNC 0x201
 #define ESP_ERR_HTTPD_RESP_HDR 0x202
@@ -35,6 +36,7 @@ HEADERS["esp_httpd_priv.h"] = """#pragma once
 #include "esp_http_server.h"
 static inline void *httpd_os_thread_handle(void) { return (void *)1; }
 struct sock_db { int fd; bool for_async_req, ws_close; uint64_t lru_counter;
+ void *ctx; int (*send_fn)(httpd_handle_t, int, const char *, size_t, int);
  int (*pending_fn)(httpd_handle_t, int);
  bool ws_handshake_done; esp_err_t (*ws_handler)(httpd_req_t *);
  bool ws_control_frames; void *ws_user_ctx; char pending_data[128]; size_t pending_len; };
@@ -44,6 +46,12 @@ struct httpd_req_aux { struct sock_db *sd; char *scratch; size_t scratch_cur_siz
 struct httpd_data { struct { unsigned max_resp_headers, max_uri_handlers, max_open_sockets; void *uri_match_fn; } config;
  struct { void *handle; } hd_td; struct httpd_req_aux hd_req_aux; struct sock_db *hd_sd;
  httpd_uri_t **hd_calls; };
+static inline struct sock_db *httpd_sess_get(httpd_handle_t h, int fd) {
+ struct httpd_data *hd = h;
+ for (unsigned i = 0; i < hd->config.max_open_sockets; ++i)
+  if (hd->hd_sd[i].fd == fd) return &hd->hd_sd[i];
+ return NULL;
+}
 esp_err_t httpd_ws_respond_server_handshake(httpd_req_t *, const char *);
 """
 

tests/web_serial_performance/run.py

@@ -2,6 +2,7 @@
 """Compile actual transport callbacks using the existing serial/store doubles."""
 import os
 import pathlib
+import re
 import subprocess
 import sys
 import tempfile
@@ -13,9 +14,32 @@ sys.path.insert(0, str(BASE))
 from run import HEADERS
 from serial_headers import SERIAL_HEADERS
 os.environ['CCACHE_DISABLE'] = '1'
+IDF = pathlib.Path(os.environ.get('IDF_PATH', str(pathlib.Path.home() / '.platformio/packages/framework-espidf')))
+
+def function(source, name):
+    match = re.search(r'^(?:static )?(?:esp_err_t|int|ssize_t) ' + name + r'\(.*?^\}', source, re.M | re.S)
+    assert match, name
+    return match.group() + '\n'
+
+adapter = (ROOT / 'src/web_httpd_adapter.c').read_text()
+assert 'ESP_IDF_VERSION_VAL(5, 5, 0)' in adapter
+ws_source = (IDF / 'components/esp_http_server/src/httpd_ws.c').read_text()
+ws = function(ws_source, 'httpd_ws_send_frame_async')
+main = (IDF / 'components/esp_http_server/src/httpd_main.c').read_text()
+assert main.index('/* Case0:') < main.index('httpd_process_ctrl_msg(hd);') < main.index('/* Case1:')
+parse = (IDF / 'components/esp_http_server/src/httpd_parse.c').read_text()
+assert parse.index('if (sd->ws_close == true)') < parse.index('ret = httpd_ws_get_frame_type(r);')
+ssl_msg = (IDF / 'components/mbedtls/mbedtls/library/ssl_msg.c').read_text()
+assert 'to be called with the same parameters' in ssl_msg
+assert ws.count('sess->send_fn(') == 2
+https = function((IDF / 'components/esp_https_server/src/https_server.c').read_text(), 'httpd_ssl_send')
+assert 'esp_tls_conn_write(tls, buf, buf_len)' in https and 'return ret;' in https
+tls = function((IDF / 'components/esp-tls/esp_tls_mbedtls.c').read_text(), 'esp_mbedtls_write')
+assert 'return (written > 0) ? written : ret;' in tls and 'return written;' in tls
 with tempfile.TemporaryDirectory(prefix='web-performance-') as directory:
     tmp = pathlib.Path(directory)
     headers = HEADERS | SERIAL_HEADERS
+    headers['esp_http_server.h'] += '\n#define HTTPD_SOCK_ERR_FAIL -1\n'
     headers['freertos/FreeRTOS.h'] = headers['freertos/FreeRTOS.h'].replace(
         'extern int host_lock_depth;', 'extern int host_lock_depth; void host_before_lock(void);').replace(
         '(void)(m); assert(host_lock_depth++', '(void)(m); host_before_lock(); assert(host_lock_depth++')
@@ -27,6 +51,14 @@ with tempfile.TemporaryDirectory(prefix='web-performance-') as directory:
     fixture = fixture.replace('#include "test.c"', '#include "' + str(BASE / 'test.c') + '"')
     fixture = fixture.replace('../../src/web_serial_transport.c', str(ROOT / 'src/web_serial_transport.c'))
     (tmp / 'fixture.c').write_text(fixture)
+    (tmp / 'binary.inc').write_text(function(adapter, 'web_httpd_aborted_send') + function(adapter, 'web_httpd_ws_send_binary'))
+    (tmp / 'sdk_ws.inc').write_text(ws.replace('httpd_ws_send_frame_async(', 'sdk_ws_send_frame('))
+    automatic = ''.join(function(ws_source, name) for name in
+                        ('httpd_ws_check_req', 'httpd_ws_send_frame', 'httpd_ws_get_frame_type'))
+    automatic = automatic.replace('httpd_ws_send_frame_async(', 'sdk_ws_send_frame(')
+    automatic = automatic.replace('httpd_ws_send_frame(', 'sdk_ws_send_request(')
+    automatic = automatic.replace('httpd_ws_recv_frame(', 'sdk_control_recv(')
+    (tmp / 'sdk_automatic.inc').write_text(automatic)
     console = (ROOT / 'src/web_console.c').read_text()
     start = console.index('static void print_performance_time(')
     end = console.index('static int command_web(', start)
@@ -39,3 +71,4 @@ with tempfile.TemporaryDirectory(prefix='web-performance-') as directory:
         '-Wl,--gc-sections', str(HERE / 'test.c'), str(ROOT / 'src/web_session_store.c'),
         str(ROOT / 'src/web_auth_parse.c'), '-o', str(tmp / 'test')], check=True, timeout=30)
     subprocess.run([str(tmp / 'test')], check=True, timeout=15)
+print('PASS installed IDF WS two-write / HTTPS forwarding / TLS partial-return contract guards')

tests/web_serial_performance/test.c

@@ -6,7 +6,83 @@ static void *queued_arg;
 static bool early;
 static esp_err_t queue_result, send_result;
 static size_t read_bytes;
-static unsigned sends;
+static unsigned sends, control_sends, lru_updates;
+/* Only private HTTPD layout/owner and the low-level TLS send are doubled. */
+struct sock_db {
+    int fd;
+    void *ctx;
+    bool ws_handshake_done, ws_close, for_async_req, ws_control_frames;
+    int (*send_fn)(httpd_handle_t, int, const char *, size_t, int);
+};
+struct httpd_req_aux {
+    struct sock_db *sd;
+    bool ws_final;
+    httpd_ws_type_t ws_type;
+};
+struct httpd_data {
+    struct { unsigned handle; } hd_td;
+    struct { struct sock_db *sd; } hd_req_aux;
+};
+static struct httpd_data host_hd;
+static struct sock_db host_sd;
+static unsigned owner;
+static int raw_result;
+static bool override_result;
+static uint8_t captured[516];
+static size_t captured_length;
+static unsigned httpd_os_thread_handle(void) { return owner; }
+static struct sock_db *httpd_sess_get(httpd_handle_t h, int fd)
+{ assert(h == &host_hd); return fd == host_sd.fd ? &host_sd : NULL; }
+static unsigned shutdown_calls, shutdown_delay;
+static int shutdown_result;
+#define SHUT_RDWR 2
+static int host_shutdown(int fd, int how)
+{
+    assert(!host_lock_depth && owner == host_hd.hd_td.handle);
+    assert(fd == host_sd.fd && how == SHUT_RDWR && host_sd.ws_close);
+    /* The send barrier must already hold even when shutdown fails. */
+    assert(host_sd.send_fn(&host_hd, fd, "ignored", 7, 0) == HTTPD_SOCK_ERR_FAIL);
+    ++shutdown_calls; now += shutdown_delay;
+    return shutdown_result;
+}
+#define shutdown host_shutdown
+#include "binary.inc"
+#undef shutdown
+#define HTTPD_WS_FIN_BIT 0x80
+#define HTTPD_WS_CONTINUE 0
+#define HTTPD_WS_MASK_BIT 0x80
+#ifndef ESP_LOGW
+#define ESP_LOGW(...) ((void)0)
+#endif
+#include "sdk_ws.inc"
+#define HTTPD_WS_OPCODE_BITS 0x0f
+#define HTTPD_WS_TYPE_CLOSE 8
+#define HTTPD_WS_TYPE_PING 9
+#define HTTPD_WS_TYPE_PONG 10
+static uint8_t incoming_opcode;
+static unsigned automatic_reads;
+static int httpd_recv_with_opt(httpd_req_t *r, char *out, size_t n, bool peek)
+{
+    (void)r; assert(n == 1 && !peek); *out = 0x80 | incoming_opcode;
+    ++automatic_reads; return 1;
+}
+static esp_err_t sdk_control_recv(httpd_req_t *r, httpd_ws_frame_t *f, size_t n)
+{
+    (void)r; assert(n == 126); f->len = 1; f->payload[0] = 0x5a;
+    ++automatic_reads; return ESP_OK;
+}
+#define ESP_LOGD(...) ((void)0)
+#include "sdk_automatic.inc"
+
+static int tls_send(httpd_handle_t h, int fd, const char *data, size_t n, int flags)
+{
+    assert(!host_lock_depth && h == &host_hd && fd == host_sd.fd && flags == 0);
+    assert(owner == host_hd.hd_td.handle && captured_length + n <= sizeof(captured));
+    memcpy(captured + captured_length, data, n); captured_length += n;
+    ++sends; now += 11;
+    if (send_hook) send_hook();
+    return override_result ? raw_result : send_result == ESP_OK ? (int)n : -1;
+}
 static unsigned lock_delay;
 static bool disconnect_stalled;
 void host_before_lock(void) { now += lock_delay; lock_delay = 0; }
@@ -25,18 +101,19 @@ esp_err_t httpd_queue_work(httpd_handle_t h, void (*fn)(void *), void *arg)
     (void)h; assert(!host_lock_depth);
     queued = fn; queued_arg = arg;
     if (queue_hook) queue_hook();
-    if (early && queue_result == ESP_OK) { now += 7; fn(arg); }
+    if (early && queue_result == ESP_OK) { captured_length = 0; now += 7; fn(arg); }
     return queue_result;
 }
 esp_err_t httpd_ws_send_frame_async(httpd_handle_t h, int fd, httpd_ws_frame_t *f)
 {
     (void)h; (void)fd; assert(!host_lock_depth); assert(f->payload == s_slots[0].tx_data);
-    ++sends; now += 11;
+    assert(f->type == HTTPD_WS_TYPE_TEXT);
+    ++control_sends; now += 11;
     if (send_hook) send_hook();
     return send_result;
 }
 esp_err_t httpd_sess_update_lru_counter(httpd_handle_t h, int fd)
-{ (void)h; (void)fd; assert(!host_lock_depth); now += 3; return ESP_OK; }
+{ (void)h; (void)fd; assert(!host_lock_depth); ++lru_updates; now += 3; return ESP_OK; }
 esp_err_t session_broker_read(session_broker_client_id_t id, uint8_t *data, size_t n, size_t *out)
 {
     (void)id; (void)data; assert(!host_lock_depth && read_bytes <= n);
@@ -45,7 +122,13 @@ esp_err_t session_broker_read(session_broker_client_id_t id, uint8_t *data, size
 }
 static web_serial_slot_t *setup(void)
 {
+    request.handle = &host_hd;
     serial_reset(); issued_t a = mint(&alice); web_serial_slot_t *s = connect_session(&a);
+    host_hd = (struct httpd_data){.hd_td.handle = 1}; owner = 1;
+    host_sd = (struct sock_db){.fd = 10, .ctx = s, .ws_handshake_done = true, .send_fn = tls_send};
+    override_result = false; captured_length = 0; control_sends = 0; lru_updates = 0;
+    shutdown_calls = shutdown_delay = automatic_reads = 0; shutdown_result = 0;
+    memset(&s_counters, 0, sizeof(s_counters));
     early = false; disconnect_stalled = false; lock_delay = 0; queue_result = send_result = ESP_OK;
     send_hook = queue_hook = read_hook = NULL; read_bytes = 0; sends = 0;
     assert(web_serial_performance_enable(false) == ESP_OK);
@@ -56,7 +139,7 @@ static void enable(void) { assert(web_serial_performance_enable(true) == ESP_OK)
 static void clear(void) { assert(web_serial_performance_clear() == ESP_OK); }
 static void disable(void) { assert(web_serial_performance_enable(false) == ESP_OK); }
 static void cycle(void) { disable(); enable(); }
-static void run_work(void) { assert(queued); queued(queued_arg); }
+static void run_work(void) { assert(queued); captured_length = 0; queued(queued_arg); }
 static void queue_binary(web_serial_slot_t *s)
 { assert(queue_slot_frame(s, s->generation, HTTPD_WS_TYPE_BINARY, 512) == ESP_OK); }
 static web_serial_performance_session_t row(void)
@@ -65,9 +148,133 @@ static web_serial_performance_session_t row(void)
     assert(web_serial_performance_snapshot(&s) == ESP_OK);
     return s.sessions[0];
 }
+static void binary_tests(void)
+{
+    /* Compare every supported length and arbitrary bytes with installed IDF's
+     * actual serializer, but require one override call instead of two. */
+    for (size_t length = 0; length <= 512; ++length) {
+        web_serial_slot_t *s = setup();
+        for (size_t i = 0; i < sizeof(s->tx_data); ++i) s->tx_data[i] = (uint8_t)i;
+        httpd_ws_frame_t frame = {.final = true, .type = HTTPD_WS_TYPE_BINARY,
+                                 .payload = s->tx_data, .len = length};
+        assert(sdk_ws_send_frame(request.handle, 10, &frame) == ESP_OK);
+        uint8_t expected[516]; size_t expected_length = captured_length;
+        memcpy(expected, captured, expected_length);
+        assert(sends == (length ? 2U : 1U));
+        captured_length = 0; sends = 0;
+        assert(web_httpd_ws_send_binary(request.handle, 10, s, s->tx_data, length) == ESP_OK);
+        assert(sends == 1 && captured_length == expected_length);
+        assert(!memcmp(captured, expected, expected_length));
+        for (size_t i = 0; i < sizeof(s->tx_data); ++i) assert(s->tx_data[i] == (uint8_t)i);
+    }
+    /* Every short result, including an entirely consumed header, fails closed.
+     * No second call/replay, no successful byte/frame or LRU accounting. */
+    for (int result = -4; result <= 517; ++result) {
+        if (result == 516) continue;
+        web_serial_slot_t *s = setup(); enable(); override_result = true; raw_result = result;
+        queue_binary(s); run_work();
+        assert(sends == 1 && !control_sends && s->close_requested && !s->work_pending);
+        assert(row().send_errors == 1 && row().send_call.sum_us == 11 && !row().sent_frames);
+        assert(s_counters.send_failures == 1 && !s_counters.tx_binary_bytes && !lru_updates);
+        assert(queue_slot_frame(s, s->generation, HTTPD_WS_TYPE_BINARY, 1) == ESP_ERR_INVALID_STATE);
+        run_work(); assert(sends == 1);
+    }
+    const int tls_errors[] = {-0x6880, -0x6900, -0x0050, -0x004e};
+    for (size_t i = 0; i < sizeof(tls_errors)/sizeof(tls_errors[0]); ++i) {
+        web_serial_slot_t *s = setup(); enable(); override_result = true; raw_result = tls_errors[i];
+        queue_binary(s); run_work(); assert(sends == 1 && s->close_requested && row().send_errors == 1);
+    }
+    for (unsigned invalid = 0; invalid < 7; ++invalid) {
+        web_serial_slot_t *s = setup(); enable(); queue_binary(s);
+        switch (invalid) {
+        case 0: owner = 2; break;
+        case 1: host_hd.hd_req_aux.sd = &host_sd; break;
+        case 2: host_sd.fd = 11; break;
+        case 3: host_sd.ctx = &s_slots[1]; break;
+        case 4: host_sd.ws_handshake_done = false; break;
+        case 5: host_sd.ws_close = true; break;
+        case 6: host_sd.for_async_req = true; break;
+        }
+        run_work(); assert(!sends && s->close_requested && row().send_errors == 1);
+        assert(!shutdown_calls && host_sd.send_fn == tls_send);
+    }
+    web_serial_slot_t *s = setup();
+    host_sd.send_fn = NULL;
+    assert(web_httpd_ws_send_binary(request.handle, 10, s, s->tx_data, 1) == ESP_ERR_INVALID_STATE);
+    host_sd.send_fn = tls_send;
+    assert(web_httpd_ws_send_binary(NULL, 10, s, s->tx_data, 1) == ESP_ERR_INVALID_ARG);
+    assert(web_httpd_ws_send_binary(request.handle, -1, s, s->tx_data, 1) == ESP_ERR_INVALID_ARG);
+    assert(web_httpd_ws_send_binary(request.handle, 10, NULL, s->tx_data, 1) == ESP_ERR_INVALID_ARG);
+    assert(web_httpd_ws_send_binary(request.handle, 10, s, NULL, 1) == ESP_ERR_INVALID_ARG);
+    assert(web_httpd_ws_send_binary(request.handle, 10, s, s->tx_data, 513) == ESP_ERR_INVALID_ARG);
+    assert(!sends);
+    s = setup(); enable(); queue_binary(s);
+    assert(queue_slot_frame(s, s->generation, HTTPD_WS_TYPE_BINARY, 512) == ESP_ERR_INVALID_STATE);
+    run_work(); assert(sends == 1 && row().sent_bytes == 512 && lru_updates == 1);
+    assert(queue_slot_frame(s, s->generation, HTTPD_WS_TYPE_TEXT, 4) == ESP_OK); run_work();
+    assert(control_sends == 1 && sends == 1 && row().sent_frames == 1);
+    /* Both fixed slots can retain one independent payload. HTTPD still sends
+     * serially; failure of the first must not retire the second reservation. */
+    s = setup(); issued_t other = mint(&bob); web_serial_slot_t *second = connect_session(&other);
+    memset(s->tx_data, 0xa5, sizeof(s->tx_data));
+    memset(second->tx_data, 0x5a, sizeof(second->tx_data));
+    queue_binary(s); queue_binary(second);
+    request.sess_ctx = s; host_sd.ctx = s; override_result = true; raw_result = 2;
+    web_serial_send_work(&s->work);
+    assert(s->close_requested && second->work_pending && !second->close_requested);
+    captured_length = 0; request.sess_ctx = second; override_result = false;
+    host_sd = (struct sock_db){.fd = 10, .ctx = second, .ws_handshake_done = true, .send_fn = tls_send};
+    web_serial_send_work(&second->work);
+    assert(sends == 2 && !second->work_pending && !second->close_requested);
+    for (size_t i = 4; i < captured_length; ++i) assert(captured[i] == 0x5a);
+    assert(s_counters.tx_binary_bytes == 512 && s_counters.send_failures == 1);
+    printf("PASS binary single-write: 513 SDK wire comparisons, short/zero/TLS errors, owner/session gates, two-slot isolation, one outstanding, text isolation\n");
+}
+
+static void abort_tests(void)
+{
+    const int results[] = {-0x6880, -0x6900, -0x0050, 0, 2, 515};
+    for (unsigned i = 0; i < sizeof(results)/sizeof(results[0]); ++i) {
+        for (int failure = 0; failure < 2; ++failure) {
+            web_serial_slot_t *s = setup(); enable();
+            override_result = true; raw_result = results[i];
+            shutdown_result = failure ? -1 : 0; shutdown_delay = 5;
+            unsigned closes_before = closes;
+            queue_binary(s); run_work();
+            assert(sends == 1 && shutdown_calls == 1 && closes == closes_before);
+            assert(s->close_requested && !s->work_pending && host_sd.ws_close);
+            assert(row().send_call.sum_us == 16 && row().send_errors == 1 && !lru_updates);
+            /* Deliberately invoke actual SDK automatic control processing even
+             * though ws_close would normally skip it in httpd_req_new. Model
+             * buffered input after shutdown, including failed shutdown. No
+             * deferred close or transport-owner iteration has run yet. */
+            struct httpd_req_aux aux = {.sd = &host_sd};
+            httpd_req_t req = {.handle = &host_hd, .aux = &aux};
+            const uint8_t opcodes[] = {HTTPD_WS_TYPE_PING, HTTPD_WS_TYPE_CLOSE};
+            override_result = false; captured_length = 0;
+            for (unsigned j = 0; j < sizeof(opcodes); ++j) {
+                incoming_opcode = opcodes[j];
+                assert(httpd_ws_get_frame_type(&req) == ESP_FAIL);
+                assert(sends == 1 && !captured_length && closes == closes_before);
+                assert(aux.ws_type == incoming_opcode && aux.ws_final);
+            }
+            assert(automatic_reads == 4 && shutdown_calls == 1);
+            /* Ordinary HTTPD teardown/new-session initialization resets the
+             * per-session override: no global fd poison survives reuse. */
+            host_sd = (struct sock_db){.fd = 10, .ctx = &s_slots[1],
+                .ws_handshake_done = true, .send_fn = tls_send};
+            incoming_opcode = HTTPD_WS_TYPE_PING;
+            assert(httpd_ws_get_frame_type(&req) == ESP_OK && sends == 3);
+        }
+    }
+    puts("PASS owner abort before return: WANT/error/zero/short, shutdown success/failure, actual SDK automatic PING/CLOSE blocked before deferred cleanup, fd reuse");
+}
+
 int main(void)
 {
     assert(serial_tests() == 0);
+    binary_tests();
+    abort_tests();
     web_serial_slot_t *s = setup();
     queue_binary(s); run_work(); assert(row().sent_frames == 0); /* default off */
     enable(); early = true; queue_binary(s);