// tcp_proxy_server.c — TCP прокси-сервер (exit node) #include "tcp_proxy_server.h" #include "udp_proxy.h" #include "icmp_proxy.h" #include "etcp.h" #include "etcp_api.h" #include "etcp_router.h" #include "utun_instance.h" #include "config_parser.h" #include "../routing_layer/topo_node.h" #include "../lib/u_async.h" #include "../lib/debug_config.h" #include "../lib/ll_queue.h" #include "../lib/mem.h" #include "../lib/tcp_io.h" #include "../lib/socket_compat.h" #include #include #include #ifndef _WIN32 #include #include #include #include #include #include #endif static struct tcp_proxy_server* g_tcp_proxy_server_ctx = NULL; static void on_fin_cb(struct tcp_conn* tc, void* arg); static void on_error_cb(struct tcp_conn* tc, int err, void* arg); static void on_flushed_cb(struct tcp_conn* tc, void* arg); static void on_closed_cb(struct tcp_conn* tc, void* arg); static void read_queue_drain_cb(struct ll_queue* q, void* arg); static void pause_resume_cb(struct ll_queue* q, void* arg); static void close_retry_cb(void* arg); static void diag_timer_cb(void* arg); static void retry_timer_cb(void* arg); static int send_msg(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst, uint8_t subcmd, uint32_t sid, const uint8_t* data, size_t len, int force); static void send_close(struct tcp_proxy_server_conn* rc); void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc); static inline int write_pending(struct tcp_conn* tc) { return tc && (tc->write_buf || tc->write_queue->head); } // ==================================================================== // Отправка сообщений через ETCP // ==================================================================== static int send_msg(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst, uint8_t subcmd, uint32_t sid, const uint8_t* data, size_t len, int force) { struct ll_entry* e = queue_entry_new(0); 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_CLIENT; e->dgram[1] = subcmd; memcpy(e->dgram + 2, &sid, 4); if (len > 0) memcpy(e->dgram + TCP_PROXY_HDR_SIZE, data, len); if (TCP_PROXY_HDR_SIZE + len > UINT16_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "tcp_proxy_server: msg too large len=%zu subcmd=%02x", len, subcmd); queue_dgram_free(e); queue_entry_free(e); return -1; } e->len = TCP_PROXY_HDR_SIZE + len; return etcp_route_send(inst, group_id, dst, e, force); } static void close_retry_cb(void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (!rc) return; rc->close_timer = NULL; if (!rc->close_pending) return; struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; if (!inst) return; uint8_t subcmd = (rc->tc && rc->tc->error) ? TCP_PROXY_SUBCMD_ERROR : TCP_PROXY_SUBCMD_CLOSE; if (send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, subcmd, rc->stream_id, NULL, 0, 1) < 0) { rc->close_backoff = rc->close_backoff < 5000 ? rc->close_backoff * 2 : 5000; rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, close_retry_cb, "tps_close_retry"); return; } rc->close_pending = 0; } static void send_close(struct tcp_proxy_server_conn* rc) { struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; if (!inst) return; if (send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, TCP_PROXY_SUBCMD_CLOSE, rc->stream_id, NULL, 0, 1) < 0) { rc->close_pending = 1; rc->close_backoff = 50; if (!rc->close_timer) rc->close_timer = uasync_set_timeout(rc->ua, rc->close_backoff, rc, close_retry_cb, "tps_close_retry"); } } static void send_fin(struct tcp_proxy_server_conn* rc) { struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; if (!inst) return; send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, TCP_PROXY_SUBCMD_FIN, rc->stream_id, NULL, 0, 1); } // ==================================================================== // Коллбэки tcp_io // ==================================================================== static void on_fin_cb(struct tcp_conn* tc, void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (rc->cli_closed) return; int pend_w = write_pending(tc); int pend_r = (rc->tx_buf != NULL) || (tc->read_queue && tc->read_queue->head); DEBUG_INFO(DEBUG_CATEGORY_PROXY, "SOCK:DST_FIN fd=%d sid=%08x fin_l=%d pend_w=%d pend_r=%d " "sent=%u recv=%u relayed=%u drain#=%u data#=%u " "rq=%d(%zub) wq=%d(%zub) wbuf=%s err=%d tx_buf=%s", (int)tc->sock, rc->stream_id, tc->fin_local, pend_w, pend_r, rc->bytes_sent, rc->bytes_recv, rc->bytes_relayed, rc->drain_count, rc->data_count, tc->read_queue->count, queue_total_bytes(tc->read_queue), tc->write_queue->count, queue_total_bytes(tc->write_queue), tc->write_buf ? "y" : "n", tc->error, rc->tx_buf ? "y" : "n"); if (tc->fin_local) { send_close(rc); tcp_conn_push_close(tc); } else if (pend_w) { tcp_conn_set_flushed(tc, on_flushed_cb); } else if (pend_r) { rc->dst_fin_deferred = 1; } else { send_fin(rc); } } static void on_flushed_cb(struct tcp_conn* tc, void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (rc->cli_closed) return; if (tc->fin_local) { DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FLUSHED fd=%d sid=%08x fin_local=%d — both FINs, closing", (int)tc->sock, rc->stream_id, tc->fin_local); send_close(rc); tcp_conn_push_close(tc); return; } send_fin(rc); } static void on_closed_cb(struct tcp_conn* tc, void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:CLOSED fd=%d sid=%08x — freeing", tc ? (int)tc->sock : -1, rc->stream_id); tcp_proxy_server_conn_free(rc); } static void on_error_cb(struct tcp_conn* tc, int err, void* arg) { (void)err; if (tc->closed) return; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "SOCK:ERR fd=%d sid=%08x err=%d fin_r=%d fin_l=%d " "sent=%u recv=%u relayed=%u drain#=%u data#=%u " "rq=%d wq=%d wbuf=%s conn=%d", tc ? (int)tc->sock : -1, rc->stream_id, err, tc ? tc->fin_remote : 0, tc ? tc->fin_local : 0, rc->bytes_sent, rc->bytes_recv, rc->bytes_relayed, rc->drain_count, rc->data_count, tc ? tc->read_queue->count : 0, tc ? tc->write_queue->count : 0, tc && tc->write_buf ? "y" : "n", tc ? tc->connected : 0); struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; uint64_t peer = rc->peer_node_id; uint32_t sid = rc->stream_id; tcp_proxy_server_conn_free(rc); if (inst) send_msg(inst, TOPO_GROUP_UTUN, peer, TCP_PROXY_SUBCMD_ERROR, sid, NULL, 0, 1); } static void diag_timer_cb(void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; rc->diag_timer = NULL; struct tcp_conn* tc = rc->tc; if (!tc || tc->sock == SOCKET_INVALID) return; int ent_f = tc->entry_pool->free_count; int dat_f = tc->data_pool->free_count; int rcv_buf = 0, snd_buf = 0; socklen_t optlen = sizeof(int); getsockopt(tc->sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcv_buf, &optlen); getsockopt(tc->sock, SOL_SOCKET, SO_SNDBUF, (char*)&snd_buf, &optlen); DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "SOCK:DIAG fd=%d sid=%08x conn=%d fin=%d " "rq=%d(%zub) wq=%d(%zub) wbuf=%s " "ent_f=%d dat_f=%d tcp_rb=%zu tcp_sb=%zu", (int)tc->sock, rc->stream_id, tc->connected, tc->fin_remote, tc->read_queue->count, queue_total_bytes(tc->read_queue), tc->write_queue->count, queue_total_bytes(tc->write_queue), tc->write_buf ? "y" : "n", ent_f, dat_f, (size_t)rcv_buf, (size_t)snd_buf); rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag"); } // ==================================================================== // Дрейн read_queue → ETCP (автозабор + backpressure) // ==================================================================== static void read_queue_drain_cb(struct ll_queue* q, void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; struct tcp_conn* tc = rc->tc; struct ll_entry* e; if (rc->tx_buf) { int ret = send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->tx_buf, rc->tx_len, 0); if (!tc->read_queue) { u_free(rc->tx_buf); return; } DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "TPS TXBUF RETRY: sid=%08x len=%u ret=%d", rc->stream_id, rc->tx_len, ret); if (ret == 0) { 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->dst_fin_deferred && !q->head) { rc->dst_fin_deferred = 0; send_fin(rc); return; } } else { 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); return; } } e = queue_data_get(q); if (!e) { if (rc->dst_fin_deferred) { rc->dst_fin_deferred = 0; send_fin(rc); } queue_resume_callback(q); return; } if (rc->cli_closed) { do { memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); e = queue_data_get(q); } while (e); queue_resume_callback(q); return; } int ret = send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, e->dgram, e->len, 0); if (!tc->read_queue) { memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); return; } rc->bytes_relayed += e->len; rc->drain_count++; DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "TPS DRAIN #%u sid=%08x len=%u → ret=%d rq=%d", rc->drain_count, rc->stream_id, e->len, ret, q->count); if (ret == 0) { memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); queue_resume_callback(q); } else { rc->tx_buf = u_malloc(e->len); if (rc->tx_buf) { memcpy(rc->tx_buf, e->dgram, e->len); rc->tx_len = e->len; } else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TPS tx_buf malloc=%u failed sid=%08x — drop", e->len, rc->stream_id); } memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); if (!tc->read_queue) return; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:BACKPRESSURE fd=%d sid=%08x — buffered %u for retry", (int)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_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"); } } static void pause_resume_cb(struct ll_queue* q, void* arg) { (void)q; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (!rc->tc || rc->tc->sock == SOCKET_INVALID || rc->cli_closed) return; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "TPS RESUME sid=%08x tx_buf=%s rq=%d", rc->stream_id, rc->tx_buf ? "y" : "n", rc->tc->read_queue ? rc->tc->read_queue->count : 0); queue_resume_callback(rc->tc->read_queue); } static void retry_timer_cb(void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; rc->retry_timer = NULL; if (!rc->tx_buf || !rc->tc || rc->tc->sock == SOCKET_INVALID || rc->cli_closed) return; struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; int ret = send_msg(inst, TOPO_GROUP_UTUN, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->tx_buf, rc->tx_len, 1); DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "TPS RETRY TIMER: sid=%08x len=%u force=1 ret=%d", rc->stream_id, rc->tx_len, ret); if (ret == 0) { u_free(rc->tx_buf); rc->tx_buf = NULL; rc->tx_len = 0; if (rc->dst_fin_deferred && rc->tc->read_queue && !rc->tc->read_queue->head) { rc->dst_fin_deferred = 0; send_fin(rc); return; } queue_resume_callback(rc->tc->read_queue); } else { rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry"); } } // ==================================================================== // Управление жизненным циклом коннекта // ==================================================================== static int conn_total(struct tcp_proxy_server_conn* rc) { if (!rc || !rc->ctx) return 0; int n = 0; struct tcp_proxy_server_conn* c; for (c = rc->ctx->conns; c; c = c->next) n++; return n; } void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) { if (!rc) return; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FREE enter rc=%p freed=%d sid=%08x", (void*)rc, rc->freed, rc->stream_id); if (rc->freed) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "SOCK:FREE double rc=%p — IGNORED", (void*)rc); return; } rc->freed = 1; int total = conn_total(rc); DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FREE fd=%d sid=%08x total=%d cli_closed=%d fin=%d error=%d", rc->tc ? (int)rc->tc->sock : -1, rc->stream_id, total, rc->cli_closed, rc->tc ? rc->tc->fin_remote : 0, rc->tc ? rc->tc->error : 0); if (rc->close_pending && rc->ctx && rc->ctx->inst) { rc->close_pending = 0; uint8_t subcmd = (rc->tc && rc->tc->error) ? TCP_PROXY_SUBCMD_ERROR : TCP_PROXY_SUBCMD_CLOSE; send_msg(rc->ctx->inst, TOPO_GROUP_UTUN, rc->peer_node_id, subcmd, rc->stream_id, NULL, 0, 1); } if (rc->ctx) { 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_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; } if (rc->diag_timer) { uasync_cancel_timeout(rc->ua, rc->diag_timer); rc->diag_timer = NULL; } if (rc->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; } DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FREE u_free rc=%p", (void*)rc); u_free(rc); } struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint32_t stream_id) { struct tcp_proxy_server_conn* c; for (c = ctx->conns; c; c = c->next) if (c->stream_id == stream_id) return c; return NULL; } // ==================================================================== // Входные точки из ETCP-диспетчера // ==================================================================== int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry, uint32_t stream_id, uint64_t src_node_id) { if (!inst || !inst->tcp_proxy_server.enabled) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return -1; } struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; if (entry->len < TCP_PROXY_CONNECT_HDR_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: CONNECT too short len=%u", entry->len); queue_dgram_free(entry); queue_entry_free(entry); return -1; } uint8_t* dest_ip = entry->dgram + TCP_PROXY_HDR_SIZE; uint16_t dest_port = 0; memcpy(&dest_port, dest_ip + 4, 2); struct tcp_proxy_server_conn* rc = u_calloc(1, sizeof(struct tcp_proxy_server_conn)); if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: u_calloc failed sid=%08x", stream_id); queue_dgram_free(entry); queue_entry_free(entry); return -1; } rc->ctx = ctx; rc->stream_id = stream_id; rc->peer_node_id = src_node_id; memcpy(rc->dest_ip, dest_ip, 4); rc->dest_port = dest_port; rc->ua = inst->ua; socket_t sock = socket(AF_INET, SOCK_STREAM, 0); if (sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: socket() failed"); u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; } ctx->conn_count++; DEBUG_INFO(DEBUG_CATEGORY_PROXY, "SOCK:NEW fd=%d sid=%08x dest=%d.%d.%d.%d:%d total=%d", (int)sock, stream_id, dest_ip[0],dest_ip[1],dest_ip[2],dest_ip[3],ntohs(dest_port), ctx->conn_count); socket_set_nonblocking(sock); struct global_config* g = &inst->config->global; if (g->tcp_proxy_server_bind_ip[0]) { struct sockaddr_in local_addr = {.sin_family = AF_INET, .sin_port = 0}; if (inet_pton(AF_INET, g->tcp_proxy_server_bind_ip, &local_addr.sin_addr) == 1) { if (bind(sock, (struct sockaddr*)&local_addr, sizeof(local_addr)) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: bind(%s) failed: %s", g->tcp_proxy_server_bind_ip, strerror(errno)); socket_close_wrapper(sock); ctx->conn_count--; u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; } } else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: invalid bind_ip=%s", g->tcp_proxy_server_bind_ip); } } if (g->tcp_proxy_server_so_mark > 0) socket_set_mark(sock, g->tcp_proxy_server_so_mark); #ifdef __FreeBSD__ if (g->tcp_proxy_server_fib > 0) { setsockopt(sock, SOL_SOCKET, SO_SETFIB, &g->tcp_proxy_server_fib, sizeof(g->tcp_proxy_server_fib)); } #endif // Bind to interface if specified (Linux only) #ifndef _WIN32 if (g->tcp_proxy_server_netif_index > 0) { char ifname[IF_NAMESIZE]; if (if_indextoname(g->tcp_proxy_server_netif_index, ifname)) socket_bind_to_device(sock, ifname); } #endif rc->tc = tcp_conn_create(inst->ua, sock, 1500, 8192, 4, 0, inst->config->global.tcp_recv_buf, on_fin_cb, on_error_cb, rc); if (!rc->tc) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: tcp_conn_create failed"); socket_close_wrapper(sock); ctx->conn_count--; u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; } rc->tc->on_closed = on_closed_cb; queue_set_callback(rc->tc->read_queue, read_queue_drain_cb, rc); queue_set_waiter_defer(rc->tc->read_queue, 1); struct sockaddr_in addr; memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; memcpy(&addr.sin_addr.s_addr, dest_ip, 4); addr.sin_port = dest_port; int ret = connect(sock, (struct sockaddr*)&addr, sizeof(addr)); if (ret < 0 && errno != EINPROGRESS) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "TCP proxy server: connect() to %d.%d.%d.%d:%d failed: %s", dest_ip[0], dest_ip[1], dest_ip[2], dest_ip[3], ntohs(dest_port), strerror(errno)); send_msg(inst, TOPO_GROUP_UTUN, src_node_id, TCP_PROXY_SUBCMD_ERROR, stream_id, NULL, 0, 1); ctx->conn_count--; tcp_conn_destroy(rc->tc); u_free(rc); queue_dgram_free(entry); queue_entry_free(entry); return -1; } rc->next = ctx->conns; ctx->conns = rc; rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag"); queue_dgram_free(entry); queue_entry_free(entry); return 0; } int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id) { (void)conn; if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); if (!rc || !rc->tc || rc->tc->sock == SOCKET_INVALID || rc->tc->error || rc->cli_closed) { DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TPS handle_data: no/closed conn sid=%08x, dropping", stream_id); queue_dgram_free(entry); queue_entry_free(entry); return -1; } size_t data_len = entry->len - TCP_PROXY_HDR_SIZE; rc->bytes_sent += (uint32_t)data_len; rc->data_count++; DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "TPS DATA #%u sid=%08x len=%zu", rc->data_count, stream_id, data_len); if (data_len > 0) { if (data_len > rc->tc->data_pool->object_size) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "tcp_proxy_server: data_len=%zu > pool_sz=%zu, dropping sid=%08x", data_len, rc->tc->data_pool->object_size, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return -1; } struct ll_entry* e = queue_entry_new_from_pool(rc->tc->entry_pool); uint8_t* buf = memory_pool_alloc(rc->tc->data_pool); if (e && buf) { memcpy(buf, entry->dgram + TCP_PROXY_HDR_SIZE, data_len); e->dgram = buf; e->len = (uint16_t)data_len; queue_data_put(rc->tc->write_queue, e); } else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "tcp_proxy_server: handle_data alloc failed sid=%08x", stream_id); if (e) queue_entry_free(e); if (buf) memory_pool_free(rc->tc->data_pool, buf); } } queue_dgram_free(entry); queue_entry_free(entry); return 0; } void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_id) { if (!inst) return; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); if (!rc) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "TCP proxy server: CLOSE sid=%08x — no conn", stream_id); return; } DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:CLOSE_RECV fd=%d sid=%08x total=%d fin=%d write_pend=%d", rc->tc ? (int)rc->tc->sock : -1, stream_id, conn_total(rc), rc->tc ? rc->tc->fin_remote : 0, rc->tc ? write_pending(rc->tc) : 0); rc->cli_closed = 1; if (!rc->tc) { tcp_proxy_server_conn_free(rc); return; } tcp_conn_pause_read(rc->tc); queue_set_callback(rc->tc->read_queue, NULL, NULL); { struct ll_entry* e; while ((e = queue_data_get(rc->tc->read_queue))) { memory_pool_free(rc->tc->data_pool, e->dgram); queue_entry_free(e); } } if (rc->tc->connected) tcp_conn_push_close(rc->tc); else tcp_proxy_server_conn_free(rc); } void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint32_t stream_id) { if (!inst) return; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); if (!rc) { DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy server: ERROR sid=%08x — no conn", stream_id); return; } DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:ERROR_RECV fd=%d sid=%08x total=%d fin=%d write_pend=%d", rc->tc ? (int)rc->tc->sock : -1, stream_id, conn_total(rc), rc->tc ? rc->tc->fin_remote : 0, rc->tc ? write_pending(rc->tc) : 0); tcp_proxy_server_handle_close(inst, stream_id); } void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint32_t stream_id) { if (!inst) return; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); if (!rc || !rc->tc || !rc->tc->connected) return; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FIN_RECV fd=%d sid=%08x — pushing FIN to wq", (int)rc->tc->sock, stream_id); tcp_conn_push_fin(rc->tc); } // ==================================================================== // ETCP коллбэк серверной стороны (принимает CONNECT/DATA/CLOSE/ERROR/FIN от клиента) // ==================================================================== 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_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", (unsigned long long)peer_id); if (inst && inst->tcp_proxy_server.enabled) { struct tcp_proxy_server_conn *rc, *next; for (rc = inst->tcp_proxy_server.conns; rc; rc = next) { next = rc->next; if (rc->peer_node_id == peer_id) tcp_proxy_server_conn_free(rc); } } } if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; } uint8_t subcmd = entry->dgram[1]; uint32_t stream_id; memcpy(&stream_id, entry->dgram + 2, 4); DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "PROXY SERVER RECV subcmd=%02x sid=%08x pkt_len=%u", subcmd, stream_id, entry->len); if (subcmd == TCP_PROXY_SUBCMD_CONNECT) { uint64_t src_node_id = conn ? conn->peer_node_id : (inst ? inst->node_id : 0); tcp_proxy_server_handle_connect(inst, entry, stream_id, src_node_id); return; } if (inst && inst->tcp_proxy_server.enabled) { struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id); if (rc) { if (subcmd == TCP_PROXY_SUBCMD_DATA) { tcp_proxy_server_handle_data(inst, conn, entry, stream_id); return; } if (subcmd == TCP_PROXY_SUBCMD_CLOSE) { tcp_proxy_server_handle_close(inst, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; } if (subcmd == TCP_PROXY_SUBCMD_ERROR) { tcp_proxy_server_handle_error(inst, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; } if (subcmd == TCP_PROXY_SUBCMD_FIN) { tcp_proxy_server_handle_fin(inst, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; } } } if (subcmd == TCP_PROXY_SUBCMD_DATA || subcmd == TCP_PROXY_SUBCMD_CLOSE || subcmd == TCP_PROXY_SUBCMD_FIN || subcmd == TCP_PROXY_SUBCMD_ERROR) { DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "PROXY subcmd=%02x sid=%08x — server conn not found, silent drop", subcmd, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; } DEBUG_WARN(DEBUG_CATEGORY_PROXY, "TCP proxy server: unhandled subcmd=%02x sid=%08x", subcmd, stream_id); queue_dgram_free(entry); queue_entry_free(entry); } // ==================================================================== // Инициализация / деинициализация // ==================================================================== int tcp_proxy_server_init(struct UTUN_INSTANCE* inst) { if (!inst) return -1; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; memset(ctx, 0, sizeof(*ctx)); if (!inst->config) return 0; ctx->enabled = inst->config->global.tcp_proxy_server_enabled; ctx->inst = inst; if (!ctx->enabled) return 0; g_tcp_proxy_server_ctx = ctx; 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); } DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy server initialized node=%016llx", (unsigned long long)inst->node_id); return 0; } void tcp_proxy_server_destroy(struct UTUN_INSTANCE* inst) { if (!inst) return; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy server destroying: conn_count=%d", ctx->conn_count); struct tcp_proxy_server_conn* rc = ctx->conns; while (rc) { struct tcp_proxy_server_conn* next = rc->next; tcp_proxy_server_conn_free(rc); rc = next; } ctx->conns = NULL; ctx->enabled = 0; if (g_tcp_proxy_server_ctx == ctx) g_tcp_proxy_server_ctx = NULL; udp_proxy_destroy(inst); icmp_proxy_destroy(inst); DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy server destroyed"); }