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

tcp_proxy: split svc_id into SERVER(0x04) and CLIENT(0x08), fix loopback conn=NULL

topo_upd
Evgeny 2 months ago
parent
5a404fbc97
commit
bb0866b4c3
5 changed files with 30 additions and 22 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 12
      src/proxy/socks_proxy.c
  2. 10
      src/proxy/tcp_proxy_client.c
  3. 12
      src/proxy/tcp_proxy_server.c
  4. 15
      src/routing_layer/etcp_router.c
  5. 3
      src/transport_layer/etcp_api.h

12
src/proxy/socks_proxy.c
Unescape View File

@ -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;
}

10
src/proxy/tcp_proxy_client.c
Unescape View File

@ -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);

12
src/proxy/tcp_proxy_server.c
Unescape View File

@ -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);

15
src/routing_layer/etcp_router.c
Unescape View File

@ -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); }
}
}

3
src/transport_layer/etcp_api.h
Unescape View File

@ -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 // распространение медиа (аудио/видео стриминг)

Write Preview
Loading…
Cancel
Save