jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Complete REALITY relay half-close on socket hangup

Browse Source
This commit is contained in:
evgeny committed 2026-10-25 23:49:47 +01:00
1 parent 7e526521eb
commit f37ab0824d
2 files changed
+39 -13
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/reality_relay.c
+38 -12
View File
@@ -72,6 +72,7 @@ static void relay_client_read_cb(socket_t sock, void *arg);
static void relay_dest_read_cb(socket_t sock, void *arg);
static void relay_client_write_cb(socket_t sock, void *arg);
static void relay_dest_write_cb(socket_t sock, void *arg);
static void relay_except_cb(socket_t sock, void *arg);
static void relay_free_cb(void *arg) {
struct reality_relay* r = arg;
@@ -105,7 +106,7 @@ static void relay_free(struct reality_relay *r) {
if (!r->free_call) DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay free scheduling failed; context retained until owner cleanup");
}
// Перезапустить таймер простоя: сброс на любую активность (чтение данных).
// Перезапускает таймер простоя после чтения или успешной отправки данных.
static int relay_touch(struct reality_relay *r) {
if (r->idle_timeout_tb <= 0) return 0;
void* timer = uasync_set_timeout(r->ua, r->idle_timeout_tb, r, relay_idle_timeout_cb, "relay_idle");
@@ -136,9 +137,9 @@ static void relay_flush_c2d(struct reality_relay *r) {
}
if (sent == 0) { DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay send to dest made no progress"); relay_free(r); return; }
r->c2d_off += (size_t)sent;
r->bytes_c2d += (size_t)sent;
if (relay_touch(r) < 0) { relay_free(r); return; }
}
r->bytes_c2d += r->c2d_len;
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);
@@ -157,9 +158,9 @@ static void relay_flush_d2c(struct reality_relay *r) {
}
if (sent == 0) { DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay send to client made no progress"); relay_free(r); return; }
r->d2c_off += (size_t)sent;
r->bytes_d2c += (size_t)sent;
if (relay_touch(r) < 0) { relay_free(r); return; }
}
r->bytes_d2c += r->d2c_len;
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);
@@ -170,7 +171,7 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
(void)sock;
if (r->c2d) return;
uint8_t *buf = u_malloc(RELAY_IO_BUF);
if (!buf) { relay_free(r); return; }
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);
if (n < 0) {
int err = socket_get_error();
@@ -181,8 +182,10 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
if (n == 0) {
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);
shutdown(r->dest_sock, RELAY_SHUT_WR);
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);
return;
}
@@ -207,7 +210,7 @@ static void relay_dest_read_cb(socket_t sock, void *arg) {
(void)sock;
if (r->d2c) return;
uint8_t *buf = u_malloc(RELAY_IO_BUF);
if (!buf) { relay_free(r); return; }
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);
if (n < 0) {
int err = socket_get_error();
@@ -218,8 +221,10 @@ static void relay_dest_read_cb(socket_t sock, void *arg) {
if (n == 0) {
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);
shutdown(r->client_sock, RELAY_SHUT_WR);
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;
}
@@ -240,12 +245,15 @@ static void relay_dest_write_cb(socket_t sock, void *arg) {
struct reality_relay *r = (struct reality_relay *)arg;
if (r->connecting) {
int err = 0; socklen_t elen = sizeof(err);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0 || err != 0) {
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0) err = socket_get_error();
if (err != 0) {
DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay connect to dest failed err=%d", err);
relay_free(r); return;
}
r->connecting = 0;
int opt = 1; setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt));
int opt = 1;
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); }
return;
@@ -254,6 +262,22 @@ static void relay_dest_write_cb(socket_t sock, void *arg) {
relay_flush_c2d(r);
}
/* HUP может прийти с последними данными: сначала дочитать поток, сохранив backpressure. */
static void relay_except_cb(socket_t sock, void *arg) {
struct reality_relay* r = arg;
if (r->closed) return;
int err = 0; socklen_t size = sizeof(err);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char*)&err, &size) < 0) err = socket_get_error();
if (err) {
DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay socket exception: side=%s err=%d",
sock == r->client_sock ? "client" : "dest", err);
relay_free(r); return;
}
if (sock == r->client_sock) {
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);
}
// Создать неблокирующий dest-сокет и начать connect() к уже разрезолвенному адресу.
// Вызывается сразу (literal IP) или из коллбэка async DNS.
static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in *addr) {
@@ -278,10 +302,12 @@ static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in
}
} else {
r->connecting = 0;
int opt = 1; setsockopt(dest_sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt));
int opt = 1;
if (setsockopt(dest_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());
}
r->dest_sid = uasync_add_socket_t(r->ua, dest_sock, relay_dest_read_cb, relay_dest_write_cb, NULL, "reality_dest", r);
r->dest_sid = uasync_add_socket_t(r->ua, dest_sock, relay_dest_read_cb, relay_dest_write_cb, relay_except_cb, "reality_dest", r);
if (!r->dest_sid) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay dest registration failed"); relay_free(r); return -1; }
if (!r->connecting) {
@@ -335,7 +361,7 @@ int reality_relay_start(struct reality_owner *owner, struct UASYNC *ua, socket_t
// берём владение client_sock сразу (снимаем старую регистрацию) — до парсинга,
// чтобы любая ошибка ниже не оставила сокет зарегистрированным в uasync.
r->client_sid = uasync_add_socket_t(ua, client_sock, relay_client_read_cb, relay_client_write_cb, NULL, "reality_client", r);
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);
tests/test_reality_relay_faults.c
+1 -1
View File
@@ -72,7 +72,7 @@ static void test_stream(void) {
assert(!socket_set_nonblocking(listener));
char destination[64]; snprintf(destination, sizeof(destination), "127.0.0.1:%u", test_bound_port(listener));
socket_t pair[2]; client_pair(pair);
assert(!reality_relay_start(&owner, ua, pair[0], destination, (const uint8_t*)"hello", 5, 1));
assert(!reality_relay_start(&owner, ua, pair[0], destination, (const uint8_t*)"hello", 5, 0));
socket_t remote = SOCKET_INVALID;
for (unsigned i = 0; remote == SOCKET_INVALID && i < 1000; i++) { remote = accept(listener, NULL, NULL); uasync_poll(ua, 10); }
assert(remote != SOCKET_INVALID && !socket_set_nonblocking(remote));