diff --git a/src/proxy/tcp_proxy_server.h b/src/proxy/tcp_proxy_server.h index df7ee63d..8dba6be7 100644 --- a/src/proxy/tcp_proxy_server.h +++ b/src/proxy/tcp_proxy_server.h @@ -1,6 +1,6 @@ /* Exit TCP-прокси: принимает CONNECT/DATA через UTUN-router и открывает обычный TCP к назначению. * Поток определяется парой peer_node_id/stream_id; proxy_flow согласует окна и half-close. - * init учитывает tcp_proxy_server_enabled в конфиге; destroy закрывает потоки и общий UDP/ICMP-прокси. + * init учитывает tcp_proxy_server_enabled в конфиге; destroy закрывает только TCP-потоки и свой binding. * Сначала init, затем приём через router callback, в конце destroy; повторный init живого сервера недопустим. * Состояние принадлежит UTUN_INSTANCE, операции и callbacks выполняются в его uasync-потоке. * CLOSE дренирует запись и прекращает чтение; ERROR немедленно уничтожает поток, FIN — half-close. */ @@ -58,11 +58,11 @@ struct tcp_proxy_server { }; // Обнулить состояние и учесть config->global.tcp_proxy_server_enabled; NULL inst → -1, иначе 0. -// Без конфига/при отключении не bind-ит сервис. При включении bind SERVER и вызывает UDP/ICMP init. -// 0 не гарантирует успешный bind/UDP/ICMP: их ошибки не пробрасываются вызывающему коду. +// Без конфига/при отключении не bind-ит сервис. При включении bind SERVER обязателен; отказ возвращает -1. +// Общими UDP/ICMP владеет instance; init/destroy TCP-сервера их не затрагивают. int tcp_proxy_server_init(struct UTUN_INSTANCE* inst); -// Удалить потоки, выключить сервер, снять binding и уничтожить общие UDP/ICMP; NULL inst допустим. -// Перед этим остановить tcp_proxy_client того же inst; дать ua обработать отложенные освобождения. +// Удалить потоки, выключить сервер и снять свой binding; NULL inst и повторный destroy допустимы. +// Дать ua обработать отложенные освобождения до уничтожения instance. void tcp_proxy_server_destroy(struct UTUN_INSTANCE* inst); // Линейный поиск в ненулевом ctx по паре peer/sid; заимствованный указатель или NULL. diff --git a/src/proxy/udp_proxy.c b/src/proxy/udp_proxy.c index fae04a9d..2247166b 100644 --- a/src/proxy/udp_proxy.c +++ b/src/proxy/udp_proxy.c @@ -10,14 +10,13 @@ #include "../lib/debug_config.h" #include "../lib/ll_queue.h" #include "../lib/mem.h" +#include "../lib/platform_compat.h" #include #include #include #ifndef _WIN32 -#include #include #include -#include #include #endif @@ -27,6 +26,19 @@ static void flow_expire_timer_cb(void* arg); static void flow_expire(struct udp_proxy_ctx* ctx); +// Удаляет опубликованный или ещё создаваемый flow, закрывая все его ресурсы. +static void flow_free(struct udp_flow* f) { + struct udp_proxy_ctx* ctx = f->ctx; + struct udp_flow** prev = &ctx->flows; + while (*prev && *prev != f) prev = &(*prev)->next; + if (*prev) { *prev = f->next; ctx->flow_count--; } + DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "udp_proxy: free flow peer=%016llx fd=%d remaining=%d", + (unsigned long long)f->client_node_id, (int)f->sock, ctx->flow_count); + if (f->read_id) uasync_remove_socket_t(ctx->ua, f->sock); + if (f->sock != SOCKET_INVALID) socket_close_wrapper(f->sock); + u_free(f); +} + static struct udp_flow* flow_find(struct udp_flow* head, uint64_t client_node_id, uint32_t src_ip, uint16_t src_port, uint32_t dst_ip, uint16_t dst_port) { @@ -44,7 +56,16 @@ static void flow_read_cb(socket_t sock, void* arg) { uint8_t buf[65507]; struct sockaddr_in from; socklen_t flen = sizeof(from); ssize_t n = recvfrom(f->sock, buf, sizeof(buf), 0, (struct sockaddr*)&from, &flen); - if (n < 0) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: recv failed errno=%d", errno); return; } + if (n < 0) { + int err = socket_get_error(); + if (err == EINTR || err == EAGAIN || err == EWOULDBLOCK) { + DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "udp_proxy: read not ready fd=%d error=%d", (int)f->sock, err); return; + } + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: recv failed peer=%016llx fd=%d error=%d (%s)", + (unsigned long long)f->client_node_id, (int)f->sock, err, socket_strerror(err)); + if (err == EBADF || err == ENOTSOCK) flow_free(f); + return; + } if (from.sin_addr.s_addr != f->dst_ip || from.sin_port != f->dst_port) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: unexpected reply source"); return; } @@ -53,9 +74,9 @@ static void flow_read_cb(socket_t sock, void* arg) { // Обернуть ответ: меняем src/dst для восстановления на стороне клиента struct ll_entry* e = queue_entry_new(0); - if (!e) return; + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: reply entry allocation failed"); return; } e->dgram = u_malloc(UDP_PROXY_HDR_SIZE + n); - if (!e->dgram) { queue_entry_free(e); return; } + if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: reply allocation failed bytes=%zd", n); queue_entry_free(e); return; } e->dgram[0] = ETCP_RT_ID_UDP_PROXY; e->dgram[1] = UDP_PROXY_SUBCMD_REPLY; // Ответ src = оригинальный dst_ip:dest_port @@ -67,7 +88,8 @@ static void flow_read_cb(socket_t sock, void* arg) { memcpy(e->dgram + UDP_PROXY_HDR_SIZE, buf, n); e->len = UDP_PROXY_HDR_SIZE + n; - etcp_route_send(ctx->inst, TOPO_GROUP_UTUN, f->client_node_id, e, 0, 0); + if (etcp_route_send(ctx->inst, TOPO_GROUP_UTUN, f->client_node_id, e, 0, 0) < 0) + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: reply router send failed bytes=%zd", n); } // ==================================================================== @@ -76,7 +98,9 @@ static void flow_read_cb(socket_t sock, void* arg) { static void exit_handle_data(struct ETCP_CONN* conn, struct ll_entry* entry) { struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; struct udp_proxy_ctx* ctx = inst ? inst->udp_proxy : NULL; - if (!inst || !ctx || !inst->tcp_proxy_server.enabled || entry->len < UDP_PROXY_RECV_HDR_SIZE) goto drop; + if (!inst || !ctx || !inst->tcp_proxy_server.enabled || entry->len < UDP_PROXY_RECV_HDR_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: invalid request len=%u context=%p", entry->len, (void*)ctx); goto drop; + } uint64_t client_node_id; memcpy(&client_node_id, entry->dgram + ROUTER_SVC_SRC_OFF, 8); uint32_t src_ip; memcpy(&src_ip, entry->dgram + ROUTER_SVC_PAYLOAD_OFF + 1, 4); @@ -85,31 +109,54 @@ static void exit_handle_data(struct ETCP_CONN* conn, struct ll_entry* entry) { uint16_t dst_port; memcpy(&dst_port, entry->dgram + ROUTER_SVC_PAYLOAD_OFF + 11, 2); uint8_t* payload = entry->dgram + UDP_PROXY_RECV_HDR_SIZE; size_t payload_len = entry->len - UDP_PROXY_RECV_HDR_SIZE; + if (payload_len > 65507) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: oversized request len=%zu", payload_len); goto drop; } struct udp_flow* f = flow_find(ctx->flows, client_node_id, src_ip, src_port, dst_ip, dst_port); if (!f) { f = u_calloc(1, sizeof(struct udp_flow)); - if (!f) goto drop; + if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: flow allocation failed"); goto drop; } f->ctx = ctx; + f->sock = SOCKET_INVALID; f->client_node_id = client_node_id; f->src_ip = src_ip; f->src_port = src_port; f->dst_ip = dst_ip; f->dst_port = dst_port; f->ua = ctx->ua; f->created_tb = get_time_tb(); f->last_activity_tb = f->created_tb; f->sock = socket(AF_INET, SOCK_DGRAM, 0); - if (f->sock == SOCKET_INVALID) { u_free(f); DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: socket failed"); goto drop; } - socket_set_nonblocking(f->sock); + if (f->sock == SOCKET_INVALID) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: socket failed error=%d", socket_get_error()); flow_free(f); goto drop; + } + if (socket_set_nonblocking(f->sock) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: nonblocking failed fd=%d error=%d", (int)f->sock, socket_get_error()); + flow_free(f); goto drop; + } struct sockaddr_in bind_addr = {.sin_family = AF_INET, .sin_addr = {.s_addr = INADDR_ANY}, .sin_port = 0}; - bind(f->sock, (struct sockaddr*)&bind_addr, sizeof(bind_addr)); + if (bind(f->sock, (struct sockaddr*)&bind_addr, sizeof(bind_addr)) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: bind ANY:0 failed fd=%d error=%d", (int)f->sock, socket_get_error()); + flow_free(f); goto drop; + } f->read_id = uasync_add_socket_t(ctx->ua, f->sock, flow_read_cb, NULL, NULL, "udp_flow", f); - if (!f->read_id) { socket_close_wrapper(f->sock); u_free(f); goto drop; } - f->next = ctx->flows; ctx->flows = f; ctx->flow_count++; - if (!ctx->expire_timer) + if (!f->read_id) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: read registration failed"); flow_free(f); goto drop; } + if (!ctx->expire_timer) { ctx->expire_timer = uasync_set_timeout(ctx->ua, ctx->flow_timeout_tb, ctx, flow_expire_timer_cb, "udp_expire"); + if (!ctx->expire_timer) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: expiry timer failed"); flow_free(f); goto drop; } + } + f->next = ctx->flows; ctx->flows = f; ctx->flow_count++; + DEBUG_INFO(DEBUG_CATEGORY_PROXY, "udp_proxy: flow ready peer=%016llx src=%08x:%u dst=%08x:%u fd=%d flows=%d", + (unsigned long long)client_node_id, ntohl(src_ip), ntohs(src_port), ntohl(dst_ip), ntohs(dst_port), + (int)f->sock, ctx->flow_count); } f->last_activity_tb = get_time_tb(); struct sockaddr_in addr = {.sin_family = AF_INET, .sin_addr = {.s_addr = dst_ip}, .sin_port = dst_port}; - sendto(f->sock, payload, payload_len, 0, (struct sockaddr*)&addr, sizeof(addr)); + ssize_t sent; + do { sent = sendto(f->sock, payload, payload_len, 0, (struct sockaddr*)&addr, sizeof(addr)); } while (sent < 0 && errno == EINTR); + if (sent != (ssize_t)payload_len) { + int err = sent < 0 ? socket_get_error() : 0; + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: dropped datagram peer=%016llx src=%08x:%u dst=%08x:%u len=%zu sent=%zd error=%d", + (unsigned long long)client_node_id, ntohl(src_ip), ntohs(src_port), ntohl(dst_ip), ntohs(dst_port), + payload_len, sent, err); + if (err == EBADF || err == ENOTSOCK) flow_free(f); + } else DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "udp_proxy: sent peer=%016llx bytes=%zu", (unsigned long long)client_node_id, payload_len); drop: queue_dgram_free(entry); queue_entry_free(entry); @@ -120,7 +167,7 @@ drop: // ==================================================================== static void client_handle_reply(struct ETCP_CONN* conn, struct ll_entry* entry) { if (entry->len < UDP_PROXY_RECV_HDR_SIZE) { queue_dgram_free(entry); queue_entry_free(entry); return; } - const uint8_t* d = entry->dgram; + uint8_t* d = entry->dgram; uint32_t src_ip, dst_ip; uint16_t src_port, dst_port; // В сообщении: src_ip:src_port = оригинальный dst, dst_ip:dst_port = оригинальный src @@ -132,7 +179,8 @@ static void client_handle_reply(struct ETCP_CONN* conn, struct ll_entry* entry) size_t payload_len = entry->len - UDP_PROXY_RECV_HDR_SIZE; struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; - udp_proxy_deliver_reply(inst, src_ip, src_port, dst_ip, dst_port, payload, payload_len); + if (udp_proxy_deliver_reply(inst, src_ip, src_port, dst_ip, dst_port, payload, payload_len) < 0) + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: TUN delivery failed bytes=%zu", payload_len); queue_dgram_free(entry); queue_entry_free(entry); } @@ -164,11 +212,13 @@ int udp_proxy_send_to_exit(struct UTUN_INSTANCE* inst, uint64_t exit_node_id, uint32_t src_ip, uint16_t src_port, uint32_t dst_ip, uint16_t dst_port, const uint8_t* payload, size_t payload_len) { - if (!inst || !inst->udp_proxy || payload_len > 65507) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: invalid send"); return -1; } + if (!inst || !inst->udp_proxy || payload_len > 65507 || (payload_len && !payload)) { + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: invalid send len=%zu", payload_len); return -1; + } struct ll_entry* e = queue_entry_new(0); - if (!e) return -1; + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: request entry allocation failed"); return -1; } e->dgram = u_malloc(UDP_PROXY_HDR_SIZE + payload_len); - if (!e->dgram) { queue_entry_free(e); return -1; } + if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: request allocation failed"); queue_entry_free(e); return -1; } e->dgram[0] = ETCP_RT_ID_UDP_PROXY; e->dgram[1] = UDP_PROXY_SUBCMD_REQUEST; memcpy(e->dgram + 2, &src_ip, 4); @@ -187,12 +237,14 @@ int udp_proxy_deliver_reply(struct UTUN_INSTANCE* inst, uint32_t src_ip, uint16_t src_port, uint32_t dst_ip, uint16_t dst_port, const uint8_t* payload, size_t payload_len) { - if (!inst || !inst->tcp_proxy_client || !inst->tcp_proxy_client->tun || payload_len > 65506) return -1; + if (!inst || !inst->tcp_proxy_client || !inst->tcp_proxy_client->tun || payload_len > 65506 || (payload_len && !payload)) { + DEBUG_WARN(DEBUG_CATEGORY_PROXY, "udp_proxy: invalid TUN delivery len=%zu", payload_len); return -1; + } // Собрать IP/UDP ответный пакет size_t ip_len = 20 + 8 + payload_len; uint8_t* pkt = u_malloc(ip_len); - if (!pkt) return -1; + if (!pkt) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: IP packet allocation failed"); return -1; } memset(pkt, 0, 28); // IPv4 UDP checksum=0: проверка отключена явно pkt[0] = 0x45; pkt[8] = 64; pkt[9] = IPPROTO_UDP; @@ -212,7 +264,7 @@ int udp_proxy_deliver_reply(struct UTUN_INSTANCE* inst, if (payload_len > 0) memcpy(pkt + 28, payload, payload_len); struct ll_entry* e = queue_entry_new(0); - if (!e) { u_free(pkt); return -1; } + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: TUN entry allocation failed"); u_free(pkt); return -1; } e->dgram = u_malloc(1 + ip_len); if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: reply allocation failed"); u_free(pkt); queue_entry_free(e); return -1; } e->dgram[0] = 4; memcpy(e->dgram + 1, pkt, ip_len); e->len = 1 + ip_len; @@ -223,10 +275,15 @@ int udp_proxy_deliver_reply(struct UTUN_INSTANCE* inst, static void flow_expire_timer_cb(void* arg) { struct udp_proxy_ctx* ctx = (struct udp_proxy_ctx*)arg; - flow_expire(ctx); ctx->expire_timer = NULL; - if (ctx->flows) + flow_expire(ctx); + if (ctx->flows) { ctx->expire_timer = uasync_set_timeout(ctx->ua, ctx->flow_timeout_tb, ctx, flow_expire_timer_cb, "udp_expire"); + if (!ctx->expire_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: cannot renew expiry timer, closing %d flows", ctx->flow_count); + while (ctx->flows) flow_free(ctx->flows); + } + } } static void flow_expire(struct udp_proxy_ctx* ctx) { @@ -234,9 +291,7 @@ static void flow_expire(struct udp_proxy_ctx* ctx) { while (*prev) { struct udp_flow* f = *prev; if (now - f->last_activity_tb > ctx->flow_timeout_tb) { - *prev = f->next; ctx->flow_count--; - if (f->read_id) { uasync_remove_socket_t(f->ua, f->sock); f->read_id = NULL; } - socket_close_wrapper(f->sock); u_free(f); + flow_free(f); } else prev = &f->next; } } @@ -245,19 +300,21 @@ static void flow_expire(struct udp_proxy_ctx* ctx) { // Публичные init/destroy // ==================================================================== int udp_proxy_init(struct UTUN_INSTANCE* inst, struct UASYNC* ua) { - if (!inst) return -1; - if (inst->udp_proxy) return 0; + if (!inst || !ua) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: invalid init"); return -1; } + if (inst->udp_proxy) { + if (inst->udp_proxy->ua == ua) return 0; + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: init attempted with another event loop"); return -1; + } struct udp_proxy_ctx* ctx = u_calloc(1, sizeof(struct udp_proxy_ctx)); - if (!ctx) return -1; + if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: context allocation failed"); return -1; } ctx->inst = inst; ctx->ua = ua; ctx->flow_timeout_tb = UDP_FLOW_TIMEOUT_TB; ctx->expire_timer = NULL; - ctx->is_exit = inst->tcp_proxy_server.enabled; + if (etcp_router_bind(inst, ETCP_RT_ID_UDP_PROXY, udp_proxy_recv_cb) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "udp_proxy: router binding failed"); u_free(ctx); return -1; + } inst->udp_proxy = ctx; - - etcp_router_bind(inst, ETCP_RT_ID_UDP_PROXY, udp_proxy_recv_cb); - ctx->initialized = 1; - DEBUG_INFO(DEBUG_CATEGORY_PROXY, "udp_proxy initialized (exit=%d)", ctx->is_exit); + DEBUG_INFO(DEBUG_CATEGORY_PROXY, "udp_proxy: shared context ready node=%016llx", (unsigned long long)inst->node_id); return 0; } @@ -266,10 +323,7 @@ void udp_proxy_destroy(struct UTUN_INSTANCE* inst) { if (!ctx) return; etcp_router_unbind(inst, ETCP_RT_ID_UDP_PROXY); if (ctx->expire_timer) { uasync_cancel_timeout(ctx->ua, ctx->expire_timer); ctx->expire_timer = NULL; } - struct udp_flow* f = ctx->flows; - while (f) { struct udp_flow* n = f->next; - if (f->read_id) { uasync_remove_socket_t(ctx->ua, f->sock); f->read_id = NULL; } - socket_close_wrapper(f->sock); u_free(f); f = n; } + while (ctx->flows) flow_free(ctx->flows); u_free(ctx); inst->udp_proxy = NULL; DEBUG_INFO(DEBUG_CATEGORY_PROXY, "udp_proxy destroyed"); } diff --git a/src/proxy/udp_proxy.h b/src/proxy/udp_proxy.h index e1676545..4ff86bca 100644 --- a/src/proxy/udp_proxy.h +++ b/src/proxy/udp_proxy.h @@ -1,7 +1,7 @@ /* UDP из TUN через UTUN-router: клиент отправляет REQUEST, exit открывает сокет для кортежа адресов * и возвращает REPLY. Потоки exit удаляются по таймауту неактивности. * init/destroy и callbacks — в uasync; inst владеет контекстом в inst->udp_proxy. - * init сохраняет роль в is_exit для диагностики; приём REQUEST проверяет текущее server.enabled, + * Приём REQUEST проверяет текущее server.enabled, * а REPLY — наличие tcp_proxy_client и совпадение peer с via_node_id. Один inst может быть и клиентом, и exit. * API принимает IP/порты в network byte order, payload заимствован только на время вызова. */ #ifndef UDP_PROXY_H @@ -49,8 +49,6 @@ struct udp_flow { }; struct udp_proxy_ctx { - int initialized; // init завершён, binding запрошен - int is_exit; // снимок server.enabled на момент init, не фильтр recv_cb struct UTUN_INSTANCE* inst; // заимствованный instance-владелец struct UASYNC* ua; // заимствованный event loop struct udp_flow* flows; // собственные flows exit @@ -61,7 +59,8 @@ struct udp_proxy_ctx { // Создать per-instance контекст и bind UDP service; повторный init существующего контекста даёт 0. -// inst/ua должны жить до destroy. 0 = контекст есть (результат bind не проверяется), -1 = ошибка inst/памяти. +// inst/ua должны жить до destroy. 0 = контекст и binding готовы, -1 = ошибка без публикации нового контекста. +// TCP client/server этот контекст не создают/уничтожают: им владеет instance. int udp_proxy_init(struct UTUN_INSTANCE* inst, struct UASYNC* ua); // Снять binding, отменить таймер, закрыть flows и обнулить inst->udp_proxy; NULL/нет контекста допустимы. void udp_proxy_destroy(struct UTUN_INSTANCE* inst);