romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-07-09 11:07:15 +03:00
Code Issues Projects Releases Packages Wiki Activity

websocket: improve handling of 0-len frames

Browse source 
Write out 9-length frames to client's WRITEFUNCTION
Read 0-length frames from READFUNCTION *if* the function
started a new frame via `curl_ws_start_frame()`.

TODO: prevent base client reader to get stuck in EOS

refs #18286
This commit is contained in:
Stefan Eissing 2025-08-18 17:12:35 +02:00
parent 9a37dd3494
commit 476076e6c2
No known key found for this signature in database
5 changed files with 100 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

74
lib/cw-out.c
View file

@ -74,6 +74,7 @@
typedef enum {
CW_OUT_NONE,
CW_OUT_BODY,
CW_OUT_BODY_0LEN,
CW_OUT_HDS
} cw_out_type;
@ -170,6 +171,7 @@ static void cw_get_writefunc(struct Curl_easy *data, cw_out_type otype,
{
switch(otype) {
case CW_OUT_BODY:
case CW_OUT_BODY_0LEN:
*pwcb = data->set.fwrite_func;
*pwcb_data = data->set.out;
*pmax_write = CURL_MAX_WRITE_SIZE;
@ -217,40 +219,50 @@ static CURLcode cw_out_ptr_flush(struct cw_out_ctx *ctx,
}
*pconsumed = 0;
while(blen && !ctx->paused) {
if(!flush_all && blen < min_write)
break;
wlen = max_write ? CURLMIN(blen, max_write) : blen;
if(otype == CW_OUT_BODY_0LEN) {
DEBUGASSERT(!blen);
Curl_set_in_callback(data, TRUE);
nwritten = wcb((char *)CURL_UNCONST(buf), 1, wlen, wcb_data);
nwritten = wcb((char *)CURL_UNCONST(buf), 1, blen, wcb_data);
Curl_set_in_callback(data, FALSE);
CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu",
wlen, (otype == CW_OUT_BODY) ? "body" : "header",
nwritten);
if(CURL_WRITEFUNC_PAUSE == nwritten) {
if(data->conn && data->conn->handler->flags & PROTOPT_NONETWORK) {
/* Protocols that work without network cannot be paused. This is
actually only FILE:// just now, and it cannot pause since the
transfer is not done using the "normal" procedure. */
failf(data, "Write callback asked for PAUSE when not supported");
CURL_TRC_WRITE(data, "[OUT] wrote %zu BODY bytes -> %zu",
blen, nwritten);
}
else {
while(blen && !ctx->paused) {
if(!flush_all && blen < min_write)
break;
wlen = max_write ? CURLMIN(blen, max_write) : blen;
Curl_set_in_callback(data, TRUE);
nwritten = wcb((char *)CURL_UNCONST(buf), 1, wlen, wcb_data);
Curl_set_in_callback(data, FALSE);
CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu",
wlen, (otype == CW_OUT_BODY) ? "body" : "header",
nwritten);
if(CURL_WRITEFUNC_PAUSE == nwritten) {
if(data->conn && data->conn->handler->flags & PROTOPT_NONETWORK) {
/* Protocols that work without network cannot be paused. This is
actually only FILE:// just now, and it cannot pause since the
transfer is not done using the "normal" procedure. */
failf(data, "Write callback asked for PAUSE when not supported");
return CURLE_WRITE_ERROR;
}
ctx->paused = TRUE;
CURL_TRC_WRITE(data, "[OUT] PAUSE requested by client");
return Curl_xfer_pause_recv(data, TRUE);
}
else if(CURL_WRITEFUNC_ERROR == nwritten) {
failf(data, "client returned ERROR on write of %zu bytes", wlen);
return CURLE_WRITE_ERROR;
}
ctx->paused = TRUE;
CURL_TRC_WRITE(data, "[OUT] PAUSE requested by client");
return Curl_xfer_pause_recv(data, TRUE);
else if(nwritten != wlen) {
failf(data, "Failure writing output to destination, "
"passed %zu returned %zd", wlen, nwritten);
return CURLE_WRITE_ERROR;
}
*pconsumed += nwritten;
blen -= nwritten;
buf += nwritten;
}
else if(CURL_WRITEFUNC_ERROR == nwritten) {
failf(data, "client returned ERROR on write of %zu bytes", wlen);
return CURLE_WRITE_ERROR;
}
else if(nwritten != wlen) {
failf(data, "Failure writing output to destination, "
"passed %zu returned %zd", wlen, nwritten);
return CURLE_WRITE_ERROR;
}
*pconsumed += nwritten;
blen -= nwritten;
buf += nwritten;
}
return CURLE_OK;
}
@ -413,7 +425,9 @@ static CURLcode cw_out_write(struct Curl_easy *data,
if((type & CLIENTWRITE_BODY) ||
((type & CLIENTWRITE_HEADER) && data->set.include_header)) {
result = cw_out_do_write(ctx, data, CW_OUT_BODY, flush_all, buf, blen);
cw_out_type otype = (!blen && (type & CLIENTWRITE_0LEN)) ?
CW_OUT_BODY_0LEN : CW_OUT_BODY;
result = cw_out_do_write(ctx, data, otype, flush_all, buf, blen);
if(result)
return result;
}

1
lib/sendf.h
View file

@ -50,6 +50,7 @@
#define CLIENTWRITE_1XX (1<<5) /* a 1xx response related HEADER */
#define CLIENTWRITE_TRAILER (1<<6) /* a trailer HEADER */
#define CLIENTWRITE_EOS (1<<7) /* End Of transfer download Stream */
#define CLIENTWRITE_0LEN (1<<8) /* write even 0-length buffers */
/**
* Write `len` bytes at `prt` to the client. `type` indicates what

29
lib/ws.c
View file

@ -474,7 +474,7 @@ static CURLcode ws_dec_read_head(struct ws_decoder *dec,
static CURLcode ws_dec_pass_payload(struct ws_decoder *dec,
struct Curl_easy *data,
struct bufq *inraw,
ws_write_payload *write_payload,
ws_write_payload *write_cb,
void *write_ctx)
{
const unsigned char *inbuf;
@ -487,9 +487,9 @@ static CURLcode ws_dec_pass_payload(struct ws_decoder *dec,
while(remain && Curl_bufq_peek(inraw, &inbuf, &inlen)) {
if((curl_off_t)inlen > remain)
inlen = (size_t)remain;
nwritten = write_payload(inbuf, inlen, dec->frame_age, dec->frame_flags,
dec->payload_offset, dec->payload_len,
write_ctx, &result);
nwritten = write_cb(inbuf, inlen, dec->frame_age, dec->frame_flags,
dec->payload_offset, dec->payload_len,
write_ctx, &result);
if(nwritten < 0)
return result;
Curl_bufq_skip(inraw, (size_t)nwritten);
@ -505,7 +505,7 @@ static CURLcode ws_dec_pass_payload(struct ws_decoder *dec,
static CURLcode ws_dec_pass(struct ws_decoder *dec,
struct Curl_easy *data,
struct bufq *inraw,
ws_write_payload *write_payload,
ws_write_payload *write_cb,
void *write_ctx)
{
CURLcode result;
@ -535,8 +535,8 @@ static CURLcode ws_dec_pass(struct ws_decoder *dec,
ssize_t nwritten;
const unsigned char tmp = '\0';
/* special case of a 0 length frame, need to write once */
nwritten = write_payload(&tmp, 0, dec->frame_age, dec->frame_flags,
0, 0, write_ctx, &result);
nwritten = write_cb(&tmp, 0, dec->frame_age, dec->frame_flags,
0, 0, write_ctx, &result);
if(nwritten < 0)
return result;
dec->state = WS_DEC_INIT;
@ -544,7 +544,7 @@ static CURLcode ws_dec_pass(struct ws_decoder *dec,
}
FALLTHROUGH();
case WS_DEC_PAYLOAD:
result = ws_dec_pass_payload(dec, data, inraw, write_payload, write_ctx);
result = ws_dec_pass_payload(dec, data, inraw, write_cb, write_ctx);
ws_dec_info(dec, data, "passing");
if(result)
return result;
@ -631,7 +631,8 @@ static ssize_t ws_cw_dec_next(const unsigned char *buf, size_t buflen,
update_meta(ws, frame_age, frame_flags, payload_offset,
payload_len, buflen);
*err = Curl_cwriter_write(data, ctx->next_writer, ctx->cw_type,
*err = Curl_cwriter_write(data, ctx->next_writer,
(ctx->cw_type | CLIENTWRITE_0LEN),
(const char *)buf, buflen);
if(*err)
return -1;
@ -943,7 +944,11 @@ static CURLcode cr_ws_read(struct Curl_easy *data,
return result;
ctx->read_eos = eos;
if(!nread) {
if(!Curl_bufq_is_empty(&ws->sendbuf)) {
/* client_read started a new frame, we disregard any eos reported */
ctx->read_eos = FALSE;
}
else if(!nread) {
/* nothing to convert, return this right away */
if(ctx->read_eos)
ctx->eos = TRUE;
@ -952,7 +957,7 @@ static CURLcode cr_ws_read(struct Curl_easy *data,
goto out;
}
if(!ws->enc.payload_remain) {
if(!ws->enc.payload_remain && Curl_bufq_is_empty(&ws->sendbuf)) {
/* encode the data as a new BINARY frame */
result = ws_enc_write_head(data, &ws->enc, CURLWS_BINARY, nread,
&ws->sendbuf);
@ -1732,7 +1737,7 @@ CURL_EXTERN CURLcode curl_ws_start_frame(CURL *d,
return CURLE_FAILED_INIT;
}
CURL_TRC_WS(data, "curl_start_frame(flags=%x, frame_len=%" FMT_OFF_T,
CURL_TRC_WS(data, "curl_ws_start_frame(flags=%x, frame_len=%" FMT_OFF_T,
flags, frame_len);
if(!data->conn) {

16
tests/http/test_20_websockets.py
View file

@ -135,7 +135,7 @@ class TestWebsockets:
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
url = f'ws://localhost:{env.ws_port}/'
r = client.run(args=[f'-{model}', '-m', str(0), '-M', str(10), url])
r = client.run(args=[f'-{model}', '-m', str(1), '-M', str(10), url])
r.check_exit_code(0)
@pytest.mark.parametrize("model", [
@ -193,3 +193,17 @@ class TestWebsockets:
large = 20000
r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url])
r.check_exit_code(0)
@pytest.mark.parametrize("model", [
pytest.param(1, id='multi_perform'),
pytest.param(2, id='curl_ws_send+recv'),
])
def test_20_09_data_empty(self, env: Env, ws_echo, model):
client = LocalClient(env=env, name='cli_ws_data')
if not client.exists():
pytest.skip(f'example client not built: {client.name}')
url = f'ws://localhost:{env.ws_port}/'
count = 10
large = 0
r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url])
r.check_exit_code(0)

27
tests/libtest/cli_ws_data.c
View file

@ -195,6 +195,8 @@ struct test_ws_m1_ctx {
char *recv_buf;
size_t send_len, nsent;
size_t recv_len, nrcvd;
int read_calls;
int write_calls;
};
static size_t test_ws_data_m1_read(char *buf, size_t nitems, size_t buflen,
@ -204,14 +206,21 @@ static size_t test_ws_data_m1_read(char *buf, size_t nitems, size_t buflen,
size_t len = nitems * buflen;
size_t left = ctx->send_len - ctx->nsent;
curl_mfprintf(stderr, "m1_read(len=%zu, left=%zu)\n", len, left);
if(left) {
ctx->read_calls++;
curl_mfprintf(stderr, "m1_read(len=%zu, left=%zu, call #%d)\n",
len, left, ctx->read_calls);
if(ctx->read_calls == 1)
curl_ws_start_frame(ctx->easy, CURLWS_BINARY, ctx->send_len);
if(left || (ctx->read_calls == 1)) {
if(left > len)
left = len;
memcpy(buf, ctx->send_buf + ctx->nsent, left);
ctx->nsent += left;
return left;
}
curl_mfprintf(stderr, "m1_read, pausing, call #%d\n",
ctx->read_calls);
return CURL_READFUNC_PAUSE;
}
@ -221,7 +230,9 @@ static size_t test_ws_data_m1_write(char *buf, size_t nitems, size_t buflen,
struct test_ws_m1_ctx *ctx = userdata;
size_t len = nitems * buflen;
curl_mfprintf(stderr, "m1_write(len=%zu)\n", len);
ctx->write_calls++;
curl_mfprintf(stderr, "m1_write(len=%zu, call +%d)\n",
len, ctx->write_calls);
if(len > (ctx->recv_len - ctx->nrcvd))
return CURL_WRITEFUNC_ERROR;
memcpy(ctx->recv_buf + ctx->nrcvd, buf, len);
@ -240,6 +251,8 @@ static CURLcode test_ws_data_m1_echo(const char *url,
struct test_ws_m1_ctx m1_ctx;
size_t i, len;
curl_mfprintf(stderr, "test_ws_data_m1_echo(min=%zu, max=%zu)\n",
plen_min, plen_max);
memset(&m1_ctx, 0, sizeof(m1_ctx));
m1_ctx.send_buf = calloc(1, plen_max + 1);
m1_ctx.recv_buf = calloc(1, plen_max + 1);
@ -278,19 +291,25 @@ static CURLcode test_ws_data_m1_echo(const char *url,
for(len = plen_min; len <= plen_max; ++len) {
/* init what we want to send and expect to receive */
curl_mfprintf(stderr, "m1_echo, iter len=%zu\n", len);
m1_ctx.send_len = len;
m1_ctx.nsent = 0;
m1_ctx.recv_len = len;
m1_ctx.nrcvd = 0;
m1_ctx.read_calls = 0;
m1_ctx.write_calls = 0;
memset(m1_ctx.recv_buf, 0, plen_max);
curl_easy_pause(m1_ctx.easy, CURLPAUSE_CONT);
for(i = 0; i < count; ++i) {
curl_mfprintf(stderr, "m1_echo, iter len=%zu, count=%zu\n", len, i);
while(1) {
int still_running; /* keep number of running handles */
CURLMcode mc = curl_multi_perform(multi, &still_running);
if(!still_running || (m1_ctx.nrcvd == m1_ctx.recv_len)) {
if(!still_running ||
(m1_ctx.read_calls && m1_ctx.write_calls &&
m1_ctx.nrcvd == m1_ctx.recv_len)) {
/* got the full echo back or failed */
break;
}