diff --git a/src/transport_layer/reality_relay.c b/src/transport_layer/reality_relay.c index bf11588e..19169af7 100644 --- a/src/transport_layer/reality_relay.c +++ b/src/transport_layer/reality_relay.c @@ -131,7 +131,12 @@ static void relay_flush_c2d(struct reality_relay *r) { if (sent < 0) { int err = socket_get_error(); if (err == ERR_INTR) continue; - if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { uasync_set_socket_write(r->ua, r->dest_sid, 1); return; } + if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { + if (uasync_set_socket_write(r->ua, r->dest_sid, 1) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot wait for dest write"); relay_free(r); + } + return; + } DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay send to dest failed err=%d", err); relay_free(r); return; } @@ -141,8 +146,10 @@ static void relay_flush_c2d(struct reality_relay *r) { if (relay_touch(r) < 0) { relay_free(r); return; } } u_free(r->c2d); r->c2d = NULL; - uasync_set_socket_write(r->ua, r->dest_sid, 0); - if (!r->client_read_eof) uasync_set_socket_read(r->ua, r->client_sid, 1); + if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK || + (!r->client_read_eof && uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK)) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot resume client after dest flush"); relay_free(r); + } } static void relay_flush_d2c(struct reality_relay *r) { @@ -152,7 +159,12 @@ static void relay_flush_d2c(struct reality_relay *r) { if (sent < 0) { int err = socket_get_error(); if (err == ERR_INTR) continue; - if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { uasync_set_socket_write(r->ua, r->client_sid, 1); return; } + if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { + if (uasync_set_socket_write(r->ua, r->client_sid, 1) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot wait for client write"); relay_free(r); + } + return; + } DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay send to client failed err=%d", err); relay_free(r); return; } @@ -162,14 +174,16 @@ static void relay_flush_d2c(struct reality_relay *r) { if (relay_touch(r) < 0) { relay_free(r); return; } } u_free(r->d2c); r->d2c = NULL; - uasync_set_socket_write(r->ua, r->client_sid, 0); - if (!r->dest_read_eof) uasync_set_socket_read(r->ua, r->dest_sid, 1); + if (uasync_set_socket_write(r->ua, r->client_sid, 0) != ERR_OK || + (!r->dest_read_eof && uasync_set_socket_read(r->ua, r->dest_sid, 1) != ERR_OK)) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot resume dest after client flush"); relay_free(r); + } } static void relay_client_read_cb(socket_t sock, void *arg) { struct reality_relay *r = (struct reality_relay *)arg; (void)sock; - if (r->c2d) return; + if (r->closed || r->c2d) return; uint8_t *buf = u_malloc(RELAY_IO_BUF); if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay client receive allocation failed"); relay_free(r); return; } ssize_t n = recv(r->client_sock, buf, RELAY_IO_BUF, 0); @@ -183,7 +197,9 @@ static void relay_client_read_cb(socket_t sock, void *arg) { u_free(buf); r->client_read_eof = 1; DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay client EOF: dest_eof=%d", r->dest_read_eof); - uasync_set_socket_read(r->ua, r->client_sid, 0); + if (uasync_set_socket_read(r->ua, r->client_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop client read at EOF"); relay_free(r); return; + } if (shutdown(r->dest_sock, RELAY_SHUT_WR) < 0) DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay dest half-close failed: err=%d", socket_get_error()); if (r->dest_read_eof) relay_free(r); @@ -200,7 +216,9 @@ static void relay_client_read_cb(socket_t sock, void *arg) { } #endif r->c2d = buf; r->c2d_len = (size_t)n; r->c2d_off = 0; - uasync_set_socket_read(r->ua, r->client_sid, 0); // backpressure + if (uasync_set_socket_read(r->ua, r->client_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot pause client for backpressure"); relay_free(r); return; + } if (relay_touch(r) < 0) { relay_free(r); return; } relay_flush_c2d(r); } @@ -208,7 +226,7 @@ static void relay_client_read_cb(socket_t sock, void *arg) { static void relay_dest_read_cb(socket_t sock, void *arg) { struct reality_relay *r = (struct reality_relay *)arg; (void)sock; - if (r->d2c) return; + if (r->closed || r->d2c) return; uint8_t *buf = u_malloc(RELAY_IO_BUF); if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay dest receive allocation failed"); relay_free(r); return; } ssize_t n = recv(r->dest_sock, buf, RELAY_IO_BUF, 0); @@ -222,14 +240,18 @@ static void relay_dest_read_cb(socket_t sock, void *arg) { u_free(buf); r->dest_read_eof = 1; DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay dest EOF: client_eof=%d", r->client_read_eof); - uasync_set_socket_read(r->ua, r->dest_sid, 0); + if (uasync_set_socket_read(r->ua, r->dest_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop dest read at EOF"); relay_free(r); return; + } if (shutdown(r->client_sock, RELAY_SHUT_WR) < 0) DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay client half-close failed: err=%d", socket_get_error()); if (r->client_read_eof) relay_free(r); return; } r->d2c = buf; r->d2c_len = (size_t)n; r->d2c_off = 0; - uasync_set_socket_read(r->ua, r->dest_sid, 0); // backpressure + if (uasync_set_socket_read(r->ua, r->dest_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot pause dest for backpressure"); relay_free(r); return; + } if (relay_touch(r) < 0) { relay_free(r); return; } relay_flush_d2c(r); } @@ -237,12 +259,19 @@ static void relay_dest_read_cb(socket_t sock, void *arg) { static void relay_client_write_cb(socket_t sock, void *arg) { struct reality_relay *r = (struct reality_relay *)arg; (void)sock; - if (!r->d2c) { uasync_set_socket_write(r->ua, r->client_sid, 0); return; } + if (r->closed) return; + if (!r->d2c) { + if (uasync_set_socket_write(r->ua, r->client_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop idle client write"); relay_free(r); + } + return; + } relay_flush_d2c(r); } static void relay_dest_write_cb(socket_t sock, void *arg) { struct reality_relay *r = (struct reality_relay *)arg; + if (r->closed) return; if (r->connecting) { int err = 0; socklen_t elen = sizeof(err); if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0) err = socket_get_error(); @@ -255,10 +284,18 @@ static void relay_dest_write_cb(socket_t sock, void *arg) { if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt)) < 0) DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay dest TCP_NODELAY failed: err=%d", socket_get_error()); if (r->c2d) relay_flush_c2d(r); - else { uasync_set_socket_write(r->ua, r->dest_sid, 0); uasync_set_socket_read(r->ua, r->client_sid, 1); } + else if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK || + uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot enable connected stream"); relay_free(r); + } + return; + } + if (!r->c2d) { + if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop idle dest write"); relay_free(r); + } return; } - if (!r->c2d) { uasync_set_socket_write(r->ua, r->dest_sid, 0); return; } relay_flush_c2d(r); } @@ -274,6 +311,7 @@ static void relay_except_cb(socket_t sock, void *arg) { relay_free(r); return; } if (sock == r->client_sock) { + if (r->connecting) { DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay client closed before dest connection"); relay_free(r); return; } if (!r->client_read_eof) relay_client_read_cb(sock, r); } else if (!r->connecting && !r->dest_read_eof) relay_dest_read_cb(sock, r); } @@ -312,7 +350,10 @@ static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in if (!r->connecting) { if (r->c2d) relay_flush_c2d(r); - else uasync_set_socket_read(r->ua, r->client_sid, 1); + else if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK || + uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot enable immediately connected stream"); relay_free(r); + } } return r->closed ? -1 : 0; } @@ -364,7 +405,9 @@ int reality_relay_start(struct reality_owner *owner, struct UASYNC *ua, socket_t r->client_sid = uasync_add_socket_t(ua, client_sock, relay_client_read_cb, relay_client_write_cb, relay_except_cb, "reality_client", r); if (!r->client_sid) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay client registration failed"); relay_free(r); return -1; } // не читаем от клиента, пока dest не подключён (иначе буферили бы без границ во время DNS) - uasync_set_socket_read(ua, r->client_sid, 0); + if (uasync_set_socket_read(ua, r->client_sid, 0) != ERR_OK || uasync_set_socket_write(ua, r->client_sid, 0) != ERR_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot suspend client during connect"); relay_free(r); return -1; + } // парсим "host:port" { diff --git a/tests/Makefile.am b/tests/Makefile.am index be27ceb9..017de72b 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -271,7 +271,8 @@ check_PROGRAMS += test_socket_monitor check_PROGRAMS += test_reality_relay_faults test_reality_relay_faults_SOURCES = test_reality_relay_faults.c test_reality_relay_faults_LDADD = $(LIBUTUN) $(COMMON_LIBS) -test_reality_relay_faults_LDFLAGS = -Wl,--wrap=uasync_set_timeout -Wl,--wrap=uasync_call_soon -Wl,--wrap=socket_set_nonblocking +test_reality_relay_faults_LDFLAGS = -Wl,--wrap=uasync_set_timeout -Wl,--wrap=uasync_call_soon \ + -Wl,--wrap=socket_set_nonblocking -Wl,--wrap=uasync_set_socket_read check_PROGRAMS += test_call_ring test_call_ring_SOURCES = test_call_ring.c ../src/transport_layer/call_ring.c test_call_ring_LDADD = $(COMMON_LIBS) diff --git a/tests/test_reality_relay_faults.c b/tests/test_reality_relay_faults.c index 9ec73ce5..8d99b87b 100644 --- a/tests/test_reality_relay_faults.c +++ b/tests/test_reality_relay_faults.c @@ -11,7 +11,12 @@ #include #include -static int fail_idle, fail_free, fail_nonblock; +static int fail_idle, fail_free, fail_nonblock, fail_monitor; +extern err_t __real_uasync_set_socket_read(struct UASYNC*, void*, int); +err_t __wrap_uasync_set_socket_read(struct UASYNC* ua, void* sid, int enabled) { + if (fail_monitor) { fail_monitor = 0; return ERR_FAIL; } + return __real_uasync_set_socket_read(ua, sid, enabled); +} extern void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*); void* __wrap_uasync_set_timeout(struct UASYNC* ua, int delay, void* arg, timeout_callback_t cb, const char* name) { if (fail_idle && !strcmp(name, "relay_idle") && !--fail_idle) return NULL; @@ -57,9 +62,13 @@ static void test_failures(void) { client_pair(pair); fail_nonblock = 1; assert(reality_relay_start(&owner, ua, pair[0], "127.0.0.1:1", NULL, 0, 1) == -1); socket_close_wrapper(pair[1]); reality_owner_cleanup(&owner); + client_pair(pair); fail_monitor = 1; + assert(reality_relay_start(&owner, ua, pair[0], "127.0.0.1:1", NULL, 0, 1) == -1); + assert(!owner.relay_count && !owner.relays); + socket_close_wrapper(pair[1]); reality_owner_cleanup(&owner); uasync_drain_immediate(ua); uasync_destroy(ua, 1); assert(u_get_allocated_count() == initial); - DEBUG_INFO(DEBUG_CATEGORY_REALITY, "PASS relay invalid ports, timeout range, timer/nonblocking failures and retained free context cleanup"); + DEBUG_INFO(DEBUG_CATEGORY_REALITY, "PASS relay invalid ports, timeout range, timer/nonblocking/monitor failures and retained free cleanup"); } static void test_stream(void) {