From bb0866b4c384510915e2038d8973e3bc6306f223 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Mon, 10 Aug 2026 22:00:39 +0300 Subject: [PATCH] tcp_proxy: split svc_id into SERVER(0x04) and CLIENT(0x08), fix loopback conn=NULL --- src/proxy/socks_proxy.c | 12 ++++++------ src/proxy/tcp_proxy_client.c | 10 +++++----- src/proxy/tcp_proxy_server.c | 12 ++++++------ src/routing_layer/etcp_router.c | 15 +++++++++++---- src/transport_layer/etcp_api.h | 3 ++- 5 files changed, 30 insertions(+), 22 deletions(-) diff --git a/src/proxy/socks_proxy.c b/src/proxy/socks_proxy.c index 4411fa21..b67c89f4 100644 --- a/src/proxy/socks_proxy.c +++ b/src/proxy/socks_proxy.c @@ -52,7 +52,7 @@ static int send_msg(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst, if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_proxy: queue_entry_new failed subcmd=%02x sid=%08x", subcmd, sid); return -1; } e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + len); if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_proxy: malloc(%zu) failed", TCP_PROXY_HDR_SIZE + len); queue_entry_free(e); return -1; } - e->dgram[0] = ETCP_RT_ID_TCP_PROXY; + e->dgram[0] = ETCP_RT_ID_TCP_PROXY_SERVER; e->dgram[1] = subcmd; memcpy(e->dgram + 2, &sid, 4); if (len > 0) memcpy(e->dgram + TCP_PROXY_HDR_SIZE, data, len); @@ -332,7 +332,7 @@ static void process_http_request(struct socks_proxy_conn* c) { u_free(pkt); } else { c->tx_buf = pkt; c->tx_len = total; - etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c); + etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &c->tx_waiter, tx_waiter_cb, c); } c->buf_len = 0; @@ -396,7 +396,7 @@ static void on_read_cb(struct ll_queue* q, void* arg) { if (c->tx_buf) { memcpy(c->tx_buf, e->dgram, e->len); c->tx_len = e->len; } else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_proxy: tx_buf malloc=%u failed sid=%08x — drop", e->len, c->stream_id); } memory_pool_free(c->tc->data_pool, e->dgram); queue_entry_free(e); - etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c); + etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &c->tx_waiter, tx_waiter_cb, c); DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCKS PROXY BP: sid=%08x tx_buf=%u waiter_reg", c->stream_id, c->tx_len); } } @@ -411,7 +411,7 @@ static void tx_waiter_cb(struct ll_queue* q, void* arg) { u_free(c->tx_buf); c->tx_buf = NULL; c->tx_len = 0; queue_resume_callback(c->tc->read_queue); } else { - etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c); + etcp_router_on_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &c->tx_waiter, tx_waiter_cb, c); } } @@ -525,7 +525,7 @@ int socks_proxy_handle_etcp(struct socks_proxy_conn** head, int* count, if (subcmd == TCP_PROXY_SUBCMD_CLOSE) { DEBUG_INFO(DEBUG_CATEGORY_PROXY, "socks_proxy: REM_CLOSED sid=%08x", stream_id); c->rem_closed = 1; - etcp_router_cancel_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY, &c->tx_waiter); + etcp_router_cancel_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &c->tx_waiter); tcp_conn_push_close(c->tc); return 1; } @@ -533,7 +533,7 @@ int socks_proxy_handle_etcp(struct socks_proxy_conn** head, int* count, if (subcmd == TCP_PROXY_SUBCMD_ERROR) { DEBUG_INFO(DEBUG_CATEGORY_PROXY, "socks_proxy: ERROR from exit sid=%08x", stream_id); c->rem_closed = 1; - etcp_router_cancel_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY, &c->tx_waiter); + etcp_router_cancel_send_ready(c->inst, TOPO_GROUP_UTUN, c->via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &c->tx_waiter); tcp_conn_push_close(c->tc); return 1; } diff --git a/src/proxy/tcp_proxy_client.c b/src/proxy/tcp_proxy_client.c index 54f147bc..99236ad4 100644 --- a/src/proxy/tcp_proxy_client.c +++ b/src/proxy/tcp_proxy_client.c @@ -71,7 +71,7 @@ static int tcp_proxy_client_send_msg(struct UTUN_INSTANCE* inst, uint64_t group_ if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "queue_entry_new failed subcmd=%02x sid=%08x", subcmd, sid); return -1; } e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + len); if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "malloc(%zu) failed subcmd=%02x sid=%08x", TCP_PROXY_HDR_SIZE + len, subcmd, sid); queue_entry_free(e); return -1; } - e->dgram[0] = ETCP_RT_ID_TCP_PROXY; + e->dgram[0] = ETCP_RT_ID_TCP_PROXY_SERVER; e->dgram[1] = subcmd; memcpy(e->dgram + 2, &sid, 4); if (len > 0) memcpy(e->dgram + TCP_PROXY_HDR_SIZE, data, len); @@ -518,7 +518,7 @@ void tcp_proxy_client_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* en struct tcp_proxy_client* proxy = inst ? inst->tcp_proxy_client : NULL; if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) { - if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY) { + if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY_CLIENT) { uint64_t peer_id; memcpy(&peer_id, entry->dgram + 1, 8); DEBUG_INFO(DEBUG_CATEGORY_PROXY, "PROXY CLOSE_ALL from %016llx — clearing client conns for peer", @@ -625,10 +625,10 @@ struct tcp_proxy_client* tcp_proxy_client_create(struct UTUN_INSTANCE* inst, str } if (inst) { - if (etcp_router_bind(inst, ETCP_RT_ID_TCP_PROXY, tcp_proxy_client_router_recv_cb) != 0) { + if (etcp_router_bind(inst, ETCP_RT_ID_TCP_PROXY_CLIENT, tcp_proxy_client_router_recv_cb) != 0) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "TCP proxy client: etcp_router_bind failed"); } else { - DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy client: etcp_router bind registered for ID=0x%02x", ETCP_RT_ID_TCP_PROXY); + DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy client: etcp_router bind registered for ID=0x%02x", ETCP_RT_ID_TCP_PROXY_CLIENT); if (need_tun) { udp_proxy_init(inst, ua); icmp_proxy_init(inst, ua); } } } @@ -644,7 +644,7 @@ void tcp_proxy_client_destroy(struct tcp_proxy_client* p) { int total = p->conn_count + p->socks_conn_count + p->http_conn_count; DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy client destroying: lwip=%d socks=%d http=%d", p->conn_count, p->socks_conn_count, p->http_conn_count); - if (p->inst) etcp_router_unbind(p->inst, ETCP_RT_ID_TCP_PROXY); + if (p->inst) etcp_router_unbind(p->inst, ETCP_RT_ID_TCP_PROXY_CLIENT); udp_proxy_destroy(p->inst); icmp_proxy_destroy(p->inst); diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c index 0b3bf7d6..a3c52da1 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -57,7 +57,7 @@ static int send_msg(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst, if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "tcp_proxy_server: queue_entry_new failed subcmd=%02x sid=%08x", subcmd, sid); return -1; } e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + len); if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "tcp_proxy_server: malloc(%zu) failed", TCP_PROXY_HDR_SIZE + len); queue_entry_free(e); return -1; } - e->dgram[0] = ETCP_RT_ID_TCP_PROXY; + e->dgram[0] = ETCP_RT_ID_TCP_PROXY_CLIENT; e->dgram[1] = subcmd; memcpy(e->dgram + 2, &sid, 4); if (len > 0) memcpy(e->dgram + TCP_PROXY_HDR_SIZE, data, len); @@ -228,7 +228,7 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) { return; } } else { - etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY, + etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY_CLIENT, &rc->pause_waiter, pause_resume_cb, rc); if (!rc->retry_timer) rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry"); DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "TPS TXBUF BP AGAIN: sid=%08x waiter_reg", rc->stream_id); @@ -267,7 +267,7 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) { memory_pool_free(rc->tc->data_pool, e->dgram); queue_entry_free(e); DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:BACKPRESSURE fd=%d sid=%08x — buffered %u for retry", (int)rc->tc->sock, rc->stream_id, rc->tx_len); - etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY, + etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY_CLIENT, &rc->pause_waiter, pause_resume_cb, rc); if (!rc->retry_timer) rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry"); } @@ -334,7 +334,7 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) { struct tcp_proxy_server_conn** prev = &rc->ctx->conns; while (*prev) { if (*prev == rc) { *prev = rc->next; rc->ctx->conn_count--; break; } prev = &(*prev)->next; } } - if (rc->ctx && rc->ctx->inst) etcp_router_cancel_send_ready(rc->ctx->inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY, &rc->pause_waiter); + if (rc->ctx && rc->ctx->inst) etcp_router_cancel_send_ready(rc->ctx->inst, TOPO_GROUP_UTUN, rc->peer_node_id, ETCP_RT_ID_TCP_PROXY_CLIENT, &rc->pause_waiter); if (rc->tx_buf) { u_free(rc->tx_buf); rc->tx_buf = NULL; rc->tx_len = 0; } if (rc->retry_timer) { uasync_cancel_timeout(rc->ua, rc->retry_timer); rc->retry_timer = NULL; } if (rc->close_timer) { uasync_cancel_timeout(rc->ua, rc->close_timer); rc->close_timer = NULL; } @@ -506,7 +506,7 @@ void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) { - if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY) { + if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY_SERVER) { uint64_t peer_id; memcpy(&peer_id, entry->dgram + 1, 8); DEBUG_INFO(DEBUG_CATEGORY_PROXY, "PROXY CLOSE_ALL from %016llx — clearing server conns for peer", @@ -562,7 +562,7 @@ int tcp_proxy_server_init(struct UTUN_INSTANCE* inst) { ctx->inst = inst; if (!ctx->enabled) return 0; g_tcp_proxy_server_ctx = ctx; - etcp_router_bind(inst, ETCP_RT_ID_TCP_PROXY, tcp_proxy_server_recv_cb); + etcp_router_bind(inst, ETCP_RT_ID_TCP_PROXY_SERVER, tcp_proxy_server_recv_cb); if (!inst->config->global.tcp_proxy_client_enabled) { udp_proxy_init(inst, inst->ua); icmp_proxy_init(inst, inst->ua); diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 9436341d..772087b4 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -328,6 +328,13 @@ static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64 } } +static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) { + static struct ETCP_CONN c; + memset(&c, 0, sizeof(c)); + c.instance = inst; c.peer_node_id = inst->node_id; + return &c; +} + static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { struct UTUN_INSTANCE* inst = rconn->inst; etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id]; @@ -341,7 +348,7 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { e->len = 9; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx", rconn->svc_id, (unsigned long long)rconn->remote_node_id); - cb(NULL, e); + cb(loopback_conn(rconn->inst), e); } // Закрыть conn: отправить CLOSE удалённой стороне, уведомить локальный сервис, очистить @@ -1167,7 +1174,7 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_ if (dst_node_id == inst->node_id) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "loopback svc_id=%u len=%zu", svc_id, payload_len); if (inst->router_bindings.callbacks[svc_id]) - inst->router_bindings.callbacks[svc_id](NULL, entry); + inst->router_bindings.callbacks[svc_id](loopback_conn(inst), entry); else { queue_dgram_free(entry); queue_entry_free(entry); } return 0; } @@ -1217,7 +1224,7 @@ int etcp_route_send_encrypted(struct UTUN_INSTANCE* inst, uint64_t group_id, uin if (dst_node_id == inst->node_id) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "E2E send: loopback svc_id=%u len=%zu", svc_id, pl_len); if (inst->router_bindings.callbacks[svc_id]) - inst->router_bindings.callbacks[svc_id](NULL, entry); + inst->router_bindings.callbacks[svc_id](loopback_conn(inst), entry); else { queue_dgram_free(entry); queue_entry_free(entry); } return 0; } @@ -1305,7 +1312,7 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uin e->dgram[0] = svc_id; memcpy(e->dgram + 1, &remote_node_id, 8); e->len = 9; - cb(NULL, e); + cb(loopback_conn(inst), e); } else { queue_entry_free(e); } } } diff --git a/src/transport_layer/etcp_api.h b/src/transport_layer/etcp_api.h index f1539d07..d9cf8ddf 100644 --- a/src/transport_layer/etcp_api.h +++ b/src/transport_layer/etcp_api.h @@ -39,7 +39,8 @@ extern "C" { #define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных #define ETCP_RT_ID_NAT 0x02 // NAT трафик между узлами #define ETCP_RT_ID_SVC_ROUTE 0x03 // etcp_router — транспорт роутера -#define ETCP_RT_ID_TCP_PROXY 0x04 // TCP proxy (клиент ↔ exit) +#define ETCP_RT_ID_TCP_PROXY_SERVER 0x04 // TCP proxy server: принимает CONNECT/DATA от клиентов +#define ETCP_RT_ID_TCP_PROXY_CLIENT 0x08 // TCP proxy client: принимает DATA/CLOSE/ERROR/FIN от сервера #define ETCP_RT_ID_UDP_PROXY 0x05 // UDP datagram прокси (client ↔ exit) #define ETCP_RT_ID_ICMP_PROXY 0x06 // ICMP echo прокси (ping через exit) #define ETCP_RT_ID_MEDIA_DELIVERY 0x07 // распространение медиа (аудио/видео стриминг)