jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
c0422a73c0c78ad9a6cae67996905a90b8e0fb18
utun2/src/proxy/tcp_proxy_server.c
T

527 lines
26 KiB
C
Raw Blame History

// 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 <stdlib.h>
#include <string.h>
#include <errno.h>
#ifndef _WIN32
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <net/if.h>
#include <fcntl.h>
#endif
static void server_wake(void* arg);
static void on_connected_cb(struct tcp_conn* tc, void* arg);
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 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);
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) {
(void)group_id;
return proxy_send(inst, dst, ETCP_RT_ID_TCP_PROXY_CLIENT, subcmd, sid, data, len, force);
}
// CLOSE может синхронно освободить rc через on_closed: резервируем состояние до push.
static void server_queue_close(struct tcp_proxy_server_conn* rc) {
if (rc->close_queued) return;
rc->close_queued = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "proxy exit: queue CLOSE peer=%016llx sid=%08x fd=%d",
(unsigned long long)rc->peer_node_id, rc->stream_id, (int)rc->tc->sock);
if (tcp_conn_push_close(rc->tc) == 0) return;
rc->close_queued = 0;
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy exit: queue CLOSE failed peer=%016llx sid=%08x",
(unsigned long long)rc->peer_node_id, rc->stream_id);
}
// FIN относится только к прочитанному направлению: сначала передаём весь read_queue.
static void server_maybe_finish(struct tcp_proxy_server_conn* rc) {
struct tcp_conn* tc = rc->tc;
if (!tc || rc->cli_closed || rc->close_queued) return;
if (tc->fin_remote && !tc->read_queue->head && !rc->flow.fin_sent) {
rc->flow.fin_pending = 1;
proxy_flow_flush(&rc->flow);
if (rc->freed) return;
}
if (rc->flow.fin_sent && rc->flow.fin_received && tc->fin_local) {
server_queue_close(rc);
}
}
static void on_fin_cb(struct tcp_conn* tc, void* arg) {
(void)tc;
server_maybe_finish(arg);
}
// Кредит возвращается после записи всей принятой порции в OS TCP, включая write_buf.
static void on_flushed_cb(struct tcp_conn* tc, void* arg) {
struct tcp_proxy_server_conn* rc = arg;
if (rc->cli_closed) return;
if (!write_pending(tc)) proxy_flow_consume(&rc->flow, TCP_PROXY_WINDOW - rc->flow.rx_credit - rc->flow.consumed);
server_maybe_finish(rc);
}
static void on_connected_cb(struct tcp_conn* tc, void* arg) {
struct tcp_proxy_server_conn* rc = arg;
DEBUG_INFO(DEBUG_CATEGORY_PROXY, "proxy exit connected peer=%016llx sid=%08x fd=%d",
(unsigned long long)rc->peer_node_id, rc->stream_id, (int)tc->sock);
rc->flow.connected_pending = 1;
proxy_flow_flush(&rc->flow);
server_wake(rc);
}
static void server_wake(void* arg) {
struct tcp_proxy_server_conn* rc = arg;
if (!rc->tc || rc->cli_closed) return;
queue_resume_callback(rc->tc->read_queue);
server_maybe_finish(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 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_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 = arg;
if (!rc->tc || rc->cli_closed) return;
struct tcp_conn* tc = rc->tc;
struct ll_entry* e = queue_data_get(q);
if (!e) { queue_resume_callback(q); server_maybe_finish(rc); return; }
int ret = proxy_flow_send(&rc->flow, e->dgram, e->len, 0);
// Loopback ERROR может отменить exit внутри send; pools tc освобождаются отложенно.
if (rc->freed) {
memory_pool_free(tc->data_pool, e->dgram); queue_entry_free(e); return;
}
if (ret == 0) {
rc->bytes_relayed += e->len; rc->drain_count++;
memory_pool_free(rc->tc->data_pool, e->dgram); queue_entry_free(e);
queue_resume_callback(q);
server_maybe_finish(rc);
return;
}
queue_data_put_first(q, e);
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "proxy exit: paused sid=%08x ret=%d credit=%u queued=%zu",
rc->stream_id, ret, rc->flow.tx_credit, queue_total_bytes(q));
if (ret != -2)
etcp_router_on_send_ready(rc->ctx->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;
server_wake(arg);
}
static void retry_timer_cb(void* arg) {
struct tcp_proxy_server_conn* rc = arg;
rc->retry_timer = NULL;
server_wake(rc);
}
// ====================================================================
// Управление жизненным циклом коннекта
// ====================================================================
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;
}
static void server_conn_free_deferred(void* arg) { u_free(arg); }
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->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);
proxy_flow_destroy(&rc->flow);
if (rc->retry_timer) { uasync_cancel_timeout(rc->ua, rc->retry_timer); rc->retry_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 deferred rc=%p", (void*)rc);
if (!uasync_call_soon(rc->ua, rc, server_conn_free_deferred))
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy exit: deferred free allocation failed sid=%08x", rc->stream_id);
}
struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint64_t peer, uint32_t stream_id) {
struct tcp_proxy_server_conn* c;
for (c = ctx->conns; c; c = c->next) if (c->peer_node_id == peer && 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 (tcp_proxy_server_find_conn(ctx, src_node_id, stream_id)) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "proxy exit: duplicate CONNECT peer=%016llx sid=%08x", (unsigned long long)src_node_id, stream_id);
send_msg(inst, TOPO_GROUP_UTUN, src_node_id, TCP_PROXY_SUBCMD_ERROR, stream_id, NULL, 0, 1);
queue_dgram_free(entry); queue_entry_free(entry); return -1;
}
if (entry->len < TCP_PROXY_RECV_HDR_SIZE + 6) { 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_RECV_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;
proxy_flow_init(&rc->flow, inst, inst->ua, src_node_id, ETCP_RT_ID_TCP_PROXY_CLIENT, stream_id, server_wake, rc);
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 src=%016llx dest=%d.%d.%d.%d:%d total=%d",
(int)sock, stream_id, (unsigned long long)src_node_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;
rc->tc->on_connected = on_connected_cb;
rc->tc->on_fin_sent = on_fin_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;
if (ret == 0) { rc->tc->connected = 1; on_connected_cb(rc->tc, rc); }
tcp_conn_set_connect_timeout(rc->tc, g->tcp_proxy_server_connect_timeout_ms);
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;
uint64_t peer; memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
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, peer, 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_RECV_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 (proxy_flow_receive(&rc->flow, data_len) < 0) {
on_error_cb(rc->tc, EPROTO, rc); queue_dgram_free(entry); queue_entry_free(entry); return -1;
}
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_RECV_HDR_SIZE, data_len);
e->dgram = buf; e->len = (uint16_t)data_len;
tcp_conn_set_flushed(rc->tc, on_flushed_cb);
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);
on_error_cb(rc->tc, ENOMEM, rc);
}
}
queue_dgram_free(entry); queue_entry_free(entry);
return 0;
}
void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint64_t peer, 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, peer, 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) {
server_queue_close(rc);
} else tcp_proxy_server_conn_free(rc);
}
void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint64_t peer, 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, peer, 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);
// ERROR отменяет поток: очередь назначения может ждать бесконечно, дренировать её нельзя.
DEBUG_INFO(DEBUG_CATEGORY_PROXY, "proxy exit: abort peer=%016llx sid=%08x queued=%zu",
(unsigned long long)peer, stream_id, rc->tc ? queue_total_bytes(rc->tc->write_queue) : 0);
tcp_proxy_server_conn_free(rc);
}
void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint64_t peer, 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, peer, stream_id);
if (!rc || !rc->tc || rc->flow.fin_received) return;
rc->flow.fin_received = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FIN_RECV fd=%d sid=%08x — pushing FIN to wq",
(int)rc->tc->sock, stream_id);
if (tcp_conn_push_fin(rc->tc) < 0) on_error_cb(rc->tc, ENOMEM, rc);
}
// ====================================================================
// 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) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
// CLOSE_ALL / restart-уведомление: [svc_id][src][dst] без payload
if (entry->len == ROUTER_SVC_HDR_SIZE && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY_SERVER) {
uint64_t peer_id;
memcpy(&peer_id, entry->dgram + ROUTER_SVC_SRC_OFF, 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);
}
}
queue_dgram_free(entry); queue_entry_free(entry);
return;
}
if (entry->len < TCP_PROXY_RECV_HDR_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "TCP proxy server: short packet len=%u", entry->len);
queue_dgram_free(entry); queue_entry_free(entry);
return;
}
uint64_t peer; memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
uint8_t subcmd = entry->dgram[ROUTER_SVC_PAYLOAD_OFF];
uint32_t stream_id; memcpy(&stream_id, entry->dgram + ROUTER_SVC_PAYLOAD_OFF + 1, 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;
memcpy(&src_node_id, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
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, peer, stream_id);
if (rc) {
if (subcmd == TCP_PROXY_SUBCMD_WINDOW) {
if (proxy_flow_window(&rc->flow, entry->dgram + TCP_PROXY_RECV_HDR_SIZE, entry->len - TCP_PROXY_RECV_HDR_SIZE) < 0)
on_error_cb(rc->tc, EPROTO, rc);
else server_wake(rc);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
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, peer, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; }
if (subcmd == TCP_PROXY_SUBCMD_ERROR) { tcp_proxy_server_handle_error(inst, peer, stream_id); queue_dgram_free(entry); queue_entry_free(entry); return; }
if (subcmd == TCP_PROXY_SUBCMD_FIN) { tcp_proxy_server_handle_fin(inst, peer, 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;
etcp_router_bind(inst, ETCP_RT_ID_TCP_PROXY_SERVER, tcp_proxy_server_recv_cb);
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;
etcp_router_unbind(inst, ETCP_RT_ID_TCP_PROXY_SERVER);
udp_proxy_destroy(inst);
icmp_proxy_destroy(inst);
DEBUG_INFO(DEBUG_CATEGORY_PROXY, "TCP proxy server destroyed");
}
Reference in new issue View Git Blame Copy Permalink