Browse Source

tun: socket pair in test_mode, remove ip_fd_mode from tcp_proxy, fix test_remote_proxy stream_id size

congestion
Evgeny 4 months ago
parent
commit
9a5d8797af
  1. 71
      src/tcp_proxy.c
  2. 9
      src/tcp_proxy.h
  3. 62
      src/tun_if.c
  4. 7
      src/tun_if.h
  5. 2
      src/utun_instance.c
  6. 2
      tests/test_remote_proxy.c
  7. 24
      tests/test_tcp_proxy.c

71
src/tcp_proxy.c

@ -122,10 +122,6 @@ static err_t tcp_output_cb(void *arg, struct pbuf *p, uint32_t src_ip, uint32_t
uint32_t sip, dip; memcpy(&sip, buf + 12, 4); memcpy(&dip, buf + 16, 4);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TUN_WRITE_ERR ret=%zd len=%u ip_total=%u", wr, len, ip_total);
}
} else if (proxy->ip_fd >= 0) {
buf[0] = (len >> 8) & 0xFF; buf[1] = len & 0xFF;
pbuf_copy_partial(p, buf + 2, len, 0);
ssize_t n = write(proxy->ip_fd, buf, 2 + len); (void)n;
}
return LERR_OK;
}
@ -340,48 +336,8 @@ static void tcp_proxy_ensure_outbound_listen(struct tcp_proxy* p, uint16_t dport
}
// ====================================================================
// Input: IP packet → lwip_tcp
// Input: IP packet → lwip_tcp (from TUN output_queue)
// ====================================================================
static void tcp_proxy_raw_read(int fd, void* arg) {
(void)fd; struct tcp_proxy* p = (struct tcp_proxy*)arg;
ssize_t n;
if (p->raw_need == 0) {
n = read(p->ip_fd, p->raw_buf + p->raw_pos, 2 - p->raw_pos);
if (n <= 0) { if (n < 0 && errno == EAGAIN) return; p->raw_pos = 0; return; }
p->raw_pos += n;
if (p->raw_pos < 2) return;
p->raw_need = ((uint16_t)p->raw_buf[0] << 8) | p->raw_buf[1];
if (p->raw_need == 0 || p->raw_need > 2000) { p->raw_pos = 0; return; }
p->raw_pos = 0;
}
n = read(p->ip_fd, p->raw_buf + 2 + p->raw_pos, p->raw_need - p->raw_pos);
if (n <= 0) { if (n < 0 && errno == EAGAIN) return; p->raw_pos = p->raw_need = 0; return; }
p->raw_pos += n;
if (p->raw_pos < p->raw_need) return;
uint8_t* pkt = p->raw_buf + 2;
size_t remaining = p->raw_need;
p->raw_pos = 0; p->raw_need = 0;
if (!tcp_proxy_handle_non_tcp(p, pkt, remaining)) {
uint8_t proto = pkt[9];
if (proto == IPPROTO_TCP) {
uint16_t ip_hdr_len = (pkt[0] & 0x0F) * 4;
uint16_t ip_total = ((uint16_t)pkt[2] << 8) | pkt[3];
if (ip_hdr_len >= 20 && ip_total >= ip_hdr_len && (size_t)ip_total <= remaining) {
uint16_t dport_net; memcpy(&dport_net, pkt + ip_hdr_len + 2, 2);
tcp_proxy_ensure_outbound_listen(p, dport_net);
uint32_t src_ip, dst_ip;
memcpy(&src_ip, pkt + 12, 4); memcpy(&dst_ip, pkt + 16, 4);
uint16_t tcp_len = ip_total - ip_hdr_len;
struct pbuf *pb = pbuf_alloc(PBUF_RAW, tcp_len);
if (pb) { pbuf_take(pb, pkt + ip_hdr_len, tcp_len); lwip_tcp_input(p->lwip, pb, src_ip, dst_ip); }
}
}
}
}
static void tcp_proxy_tun_input(struct ll_queue* q, void* arg) {
struct tcp_proxy* p = (struct tcp_proxy*)arg;
struct ll_entry* entry = queue_data_get(q);
@ -552,36 +508,28 @@ void tcp_proxy_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
struct tcp_proxy* tcp_proxy_create(struct UTUN_INSTANCE* inst, struct UASYNC* ua,
const char* tun_name, const char* tun_ip, int mtu, int test_mode,
struct tcp_proxy_mapping_config* mappings, int mapping_count,
int use_tun, int ip_fd, uint64_t via_node_id)
uint64_t via_node_id)
{
if (!ua) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: ua is NULL"); return NULL; }
if (!tun_name || !tun_ip) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: tun name/ip required"); return NULL; }
struct tcp_proxy* p = u_calloc(1, sizeof(struct tcp_proxy));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: u_calloc failed"); return NULL; }
p->inst = inst; p->ua = ua; p->ip_fd = -1; p->next_stream_id = 1;
p->inst = inst; p->ua = ua; p->next_stream_id = 1;
p->via_node_id = via_node_id;
p->ip_fd_mode = !use_tun;
p->mappings = mappings; p->mapping_count = mapping_count;
p->entry_pool = memory_pool_init(sizeof(struct ll_entry));
if (!p->entry_pool) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: memory_pool_init failed"); u_free(p); return NULL; }
if (use_tun) {
if (!tun_name || !tun_ip) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: tun name/ip required"); memory_pool_destroy(p->entry_pool); u_free(p); return NULL; }
p->tun = tun_init_nat(ua, tun_name, tun_ip, mtu > 0 ? mtu : 1500, test_mode);
if (!p->tun) { DEBUG_ERROR(DEBUG_CATEGORY_TUN, "tcp_proxy: failed to create TUN %s", tun_name); memory_pool_destroy(p->entry_pool); u_free(p); return NULL; }
queue_set_callback(p->tun->output_queue, tcp_proxy_tun_input, p);
} else {
p->ip_fd = ip_fd;
p->ip_fd_id = uasync_add_socket(ua, ip_fd, tcp_proxy_raw_read, NULL, NULL, p);
if (!p->ip_fd_id) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy: uasync_add_socket for ip_fd failed"); memory_pool_destroy(p->entry_pool); u_free(p); return NULL; }
}
p->tun = tun_init_nat(ua, tun_name, tun_ip, mtu > 0 ? mtu : 1500, test_mode);
if (!p->tun) { DEBUG_ERROR(DEBUG_CATEGORY_TUN, "tcp_proxy: failed to create TUN %s", tun_name); memory_pool_destroy(p->entry_pool); u_free(p); return NULL; }
queue_set_callback(p->tun->output_queue, tcp_proxy_tun_input, p);
p->lwip = lwip_tcp_init(ua, tcp_output_cb, p);
if (!p->lwip) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_create: lwip_tcp_init failed");
if (p->ip_fd_id) uasync_remove_socket(ua, p->ip_fd_id);
if (p->tun) tun_close(p->tun);
tun_close(p->tun);
memory_pool_destroy(p->entry_pool); u_free(p); return NULL;
}
@ -609,7 +557,7 @@ struct tcp_proxy* tcp_proxy_create(struct UTUN_INSTANCE* inst, struct UASYNC* ua
}
}
DEBUG_INFO(DEBUG_CATEGORY_TUN, "TCP proxy created: use_tun=%d mappings=%d via_node=%016llx", use_tun, mapping_count, (unsigned long long)via_node_id);
DEBUG_INFO(DEBUG_CATEGORY_TUN, "TCP proxy created: mappings=%d via_node=%016llx", mapping_count, (unsigned long long)via_node_id);
return p;
}
@ -636,7 +584,6 @@ void tcp_proxy_destroy(struct tcp_proxy* p) {
}
p->conns = NULL;
if (p->ip_fd_id) { uasync_remove_socket(p->ua, p->ip_fd_id); p->ip_fd_id = NULL; }
if (p->tun) tun_close(p->tun);
if (p->entry_pool) memory_pool_destroy(p->entry_pool);
u_free(p);

9
src/tcp_proxy.h

@ -47,28 +47,21 @@ struct tcp_proxy {
struct UTUN_INSTANCE* inst;
struct UASYNC* ua;
struct tun_if* tun;
int ip_fd;
void* ip_fd_id;
struct proxy_conn* conns;
int conn_count;
struct memory_pool* entry_pool;
uint32_t next_stream_id;
uint64_t via_node_id;
struct lwip_tcp_ctx* lwip;
int ip_fd_mode;
struct tcp_proxy_mapping_config* mappings; // pointer to config
int mapping_count;
uint8_t raw_buf[2002];
size_t raw_pos;
uint16_t raw_need;
};
struct tcp_proxy* tcp_proxy_create(struct UTUN_INSTANCE* inst, struct UASYNC* ua,
const char* tun_name, const char* tun_ip, int mtu, int test_mode,
struct tcp_proxy_mapping_config* mappings, int mapping_count,
int use_tun, int ip_fd, uint64_t via_node_id);
uint64_t via_node_id);
void tcp_proxy_destroy(struct tcp_proxy* p);
void tcp_proxy_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);

62
src/tun_if.c

@ -10,6 +10,8 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/socket.h>
#include "../lib/mem.h"
#ifdef _WIN32
@ -126,6 +128,7 @@ struct tun_if* tun_init(struct UASYNC* ua, struct utun_config* config)
tun->ua = ua;
tun->test_mode = test_mode;
tun->fd = -1;
tun->test_fd = -1;
const char* name = config->global.tun_ifname;
if (name && name[0]) strncpy(tun->ifname, name, sizeof(tun->ifname)-1);
@ -145,8 +148,16 @@ struct tun_if* tun_init(struct UASYNC* ua, struct utun_config* config)
u_free(tun);
return NULL;
}
} else if (tun->ifname[0] == '\0') {
strncpy(tun->ifname, "tun_test", sizeof(tun->ifname)-1);
} else {
if (tun->ifname[0] == '\0') strncpy(tun->ifname, "tun_test", sizeof(tun->ifname)-1);
int fds[2];
if (socketpair(AF_UNIX, SOCK_DGRAM, 0, fds) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_TUN, "TUN %s test_mode socketpair failed: %s", tun->ifname, strerror(errno));
u_free(tun);
return NULL;
}
tun->fd = fds[0];
tun->test_fd = fds[1];
}
tun->pool = memory_pool_init(sizeof(struct ll_entry));
@ -158,36 +169,37 @@ struct tun_if* tun_init(struct UASYNC* ua, struct utun_config* config)
queue_set_callback(tun->input_queue, tun_input_queue_callback, tun);
if (!test_mode) {
int poll_fd = tun_platform_get_poll_fd(tun);
if (poll_fd >= 0) {
#ifndef _WIN32
// ==================== LINUX ====================
{
int poll_fd = test_mode ? tun->fd : tun_platform_get_poll_fd(tun);
if (poll_fd >= 0) {
tun->socket_id = uasync_add_socket(ua, poll_fd, tun_read_callback, NULL, NULL, tun);
if (!tun->socket_id) goto fail;
}
}
#endif
} else {
#ifdef _WIN32
// ==================== WINDOWS ====================
if (!test_mode) {
int poll_fd = tun_platform_get_poll_fd(tun);
if (poll_fd < 0) {
tun->running = 1;
tun->stop_event = CreateEvent(NULL, TRUE, FALSE, NULL);
if (!tun->stop_event) goto fail;
tun->read_thread = CreateThread(NULL, 0, tun_read_thread_proc, tun, 0, NULL);
if (!tun->read_thread) {
CloseHandle(tun->stop_event);
goto fail;
}
#endif
}
}
#endif
DEBUG_INFO(DEBUG_CATEGORY_TUN, "TUN %s initialized (%s mode)", tun->ifname,
test_mode ? "TEST" : "REAL");
return tun;
fail:
if (tun->test_fd >= 0) { close(tun->test_fd); tun->test_fd = -1; }
#ifdef _WIN32
if (tun->stop_event) CloseHandle(tun->stop_event);
if (tun->read_thread) {
@ -218,6 +230,7 @@ struct tun_if* tun_init_nat(struct UASYNC* ua, const char* ifname, const char* i
tun->ua = ua;
tun->test_mode = test_mode;
tun->fd = -1;
tun->test_fd = -1;
strncpy(tun->ifname, ifname, sizeof(tun->ifname) - 1);
char ip_str[64];
@ -228,6 +241,15 @@ struct tun_if* tun_init_nat(struct UASYNC* ua, const char* ifname, const char* i
u_free(tun);
return NULL;
}
} else {
int fds[2];
if (socketpair(AF_UNIX, SOCK_DGRAM, 0, fds) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_TUN, "NAT TUN %s test_mode socketpair failed: %s", tun->ifname, strerror(errno));
u_free(tun);
return NULL;
}
tun->fd = fds[0];
tun->test_fd = fds[1];
}
tun->pool = memory_pool_init(sizeof(struct ll_entry));
@ -239,21 +261,22 @@ struct tun_if* tun_init_nat(struct UASYNC* ua, const char* ifname, const char* i
queue_set_callback(tun->input_queue, tun_input_queue_callback, tun);
if (!test_mode) {
int poll_fd = tun_platform_get_poll_fd(tun);
if (poll_fd >= 0) {
#ifndef _WIN32
{
int poll_fd = test_mode ? tun->fd : tun_platform_get_poll_fd(tun);
if (poll_fd >= 0) {
tun->socket_id = uasync_add_socket(ua, poll_fd, tun_read_callback, NULL, NULL, tun);
if (!tun->socket_id) goto fail2;
#endif
}
}
#endif
DEBUG_INFO(DEBUG_CATEGORY_TUN, "NAT TUN %s initialized (%s mode)", tun->ifname,
test_mode ? "TEST" : "REAL");
return tun;
fail2:
if (tun->test_fd >= 0) { close(tun->test_fd); tun->test_fd = -1; }
if (tun->output_queue) queue_free(tun->output_queue);
if (tun->input_queue) queue_free(tun->input_queue);
if (tun->pool) memory_pool_destroy(tun->pool);
@ -283,6 +306,8 @@ void tun_close(struct tun_if* tun)
uasync_remove_socket(tun->ua, tun->socket_id);
}
if (tun->test_fd >= 0) { close(tun->test_fd); tun->test_fd = -1; }
struct ll_entry* pkt;
while ((pkt = queue_data_get(tun->input_queue)) != NULL) {
queue_dgram_free(pkt);
@ -306,11 +331,9 @@ ssize_t tun_write(struct tun_if* tun, const uint8_t* buf, size_t len)
if (!tun || !buf || len <= 1) return -1;
if (tun->test_mode) {
// Skip prefix byte for test mode injection
return tun_inject_packet(tun, buf + 1, len - 1) == 0 ? (ssize_t)(len - 1) : -1;
}
// Skip prefix byte, write from second byte
ssize_t n = tun_platform_write(tun, buf + 1, len - 1);
if (n > 0) {
tun->bytes_written += n;
@ -327,6 +350,7 @@ ssize_t tun_write(struct tun_if* tun, const uint8_t* buf, size_t len)
struct ll_queue* tun_get_input_queue(struct tun_if* tun) { return tun ? tun->input_queue : NULL; }
struct ll_queue* tun_get_output_queue(struct tun_if* tun) { return tun ? tun->output_queue : NULL; }
int tun_is_test_mode(struct tun_if* tun) { return tun ? tun->test_mode : -1; }
int tun_get_test_fd(struct tun_if* tun) { return tun ? tun->test_fd : -1; }
// ===================================================================
// Callback для обработки пакетов из TUN в main thread
@ -364,10 +388,9 @@ int tun_inject_packet(struct tun_if* tun, const uint8_t* buf, size_t len)
{
if (!tun || !buf || len == 0 || len > TUN_MAX_PACKET_SIZE) return -1;
// Allocate with +1 for prefix byte
uint8_t* data = u_malloc(len + 1);
if (!data) { tun->read_errors++; return -1; }
data[0] = 0; // Prefix byte
data[0] = 0;
memcpy(data + 1, buf, len);
struct ll_entry* pkt = queue_entry_new_from_pool(tun->pool);
@ -402,7 +425,6 @@ ssize_t tun_read_packet(struct tun_if* tun, uint8_t* buf, size_t len)
return -1;
}
// Add prefix byte at start
buf[0] = 0;
memcpy(buf + 1, pkt->dgram, pkt->len);
ssize_t ret = pkt->len + 1;

7
src/tun_if.h

@ -49,6 +49,7 @@ struct tun_if {
struct ll_queue* input_queue; // routing → TUN (пакеты в интерфейс)
int test_mode; // 1 = тестовый режим (без реального TUN)
int test_fd; // в тестовом режиме: вторая сторона socket pair для тестов, -1 иначе
// Статистика
uint64_t bytes_read;
@ -111,6 +112,12 @@ struct ll_queue* tun_get_output_queue(struct tun_if* tun);
*/
int tun_is_test_mode(struct tun_if* tun);
/**
* @brief Получить test_fd (вторая сторона socket pair) для тестов
* @return fd или -1 если не тестовый режим
*/
int tun_get_test_fd(struct tun_if* tun);
/**
* @brief Инъекция пакета в output_queue (для тестов и Windows-уведомлений)
*/

2
src/utun_instance.c

@ -158,7 +158,7 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u
int mtu = config->global.tcp_proxy_mtu > 0 ? config->global.tcp_proxy_mtu : 1500;
instance->tcp_proxy = tcp_proxy_create(instance, ua, tun_name, tun_ip, mtu, g_tun_init_enabled ? 0 : 1,
config->global.tcp_proxy_mappings, config->global.tcp_proxy_mapping_count,
1, -1, config->global.tcp_proxy_via_node_id);
config->global.tcp_proxy_via_node_id);
if (instance->tcp_proxy) {
DEBUG_INFO(DEBUG_CATEGORY_TUN, "TCP proxy enabled: TUN=%s IP=%s MTU=%d mappings=%d",
tun_name, tun_ip, mtu, config->global.tcp_proxy_mapping_count);

2
tests/test_remote_proxy.c

@ -73,7 +73,7 @@ static void test_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return;
}
uint8_t subcmd = entry->dgram[1];
uint64_t sid = 0; memcpy(&sid, entry->dgram + 2, 8);
uint32_t sid = 0; memcpy(&sid, entry->dgram + 2, 4);
uint64_t src = conn ? conn->peer_node_id : i->node_id;
if (subcmd == TCP_PROXY_SUBCMD_CONNECT && sid == stream_id) { remote_proxy_handle_connect(i, entry, sid, src); return; }
if (sid != stream_id) { queue_dgram_free(entry); queue_entry_free(entry); return; }

24
tests/test_tcp_proxy.c

@ -2,10 +2,9 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include "test_utils.h"
#include "../src/tcp_proxy.h"
#include "../src/tun_if.h"
#include "../src/config_parser.h"
#include "../lib/u_async.h"
#include "../lib/debug_config.h"
@ -21,17 +20,14 @@ static void check_no_leaks(const char* label) {
int main(void) {
debug_config_init();
int fds[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) { perror("socketpair"); return 1; }
// Test 1: create proxy with mapping, raw-fd mode
// Test 1: create proxy with mapping
{
struct UASYNC* ua = uasync_create();
if (!ua) { printf("[FAIL] uasync_create\n"); close(fds[0]); close(fds[1]); return 1; }
if (!ua) { printf("[FAIL] uasync_create\n"); return 1; }
struct tcp_proxy_mapping_config m = {.local_port = 9090, .remote_ip = "127.0.0.1", .remote_port = 9999};
struct tcp_proxy* p = tcp_proxy_create(NULL, ua, NULL, NULL, 0, 0, &m, 1, 0, fds[0], 0);
if (!p) { printf("[FAIL] tcp_proxy_create with mapping\n"); uasync_destroy(ua, 0); close(fds[0]); close(fds[1]); return 1; }
struct tcp_proxy* p = tcp_proxy_create(NULL, ua, "tun_tcp", "10.99.0.1", 1500, 1, &m, 1, 0);
if (!p) { printf("[FAIL] tcp_proxy_create with mapping\n"); uasync_destroy(ua, 0); return 1; }
tcp_proxy_destroy(p);
uasync_destroy(ua, 0);
@ -40,21 +36,17 @@ int main(void) {
// Test 2: create proxy without mapping
{
int fds2[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds2) < 0) { perror("socketpair2"); close(fds[1]); return 1; }
struct UASYNC* ua = uasync_create();
if (!ua) { printf("[FAIL] uasync_create\n"); close(fds2[0]); close(fds2[1]); close(fds[1]); return 1; }
if (!ua) { printf("[FAIL] uasync_create\n"); return 1; }
struct tcp_proxy* p = tcp_proxy_create(NULL, ua, NULL, NULL, 0, 0, NULL, 0, 0, fds2[0], 0);
if (!p) { printf("[FAIL] tcp_proxy_create without mapping\n"); uasync_destroy(ua, 0); close(fds2[0]); close(fds2[1]); close(fds[1]); return 1; }
struct tcp_proxy* p = tcp_proxy_create(NULL, ua, "tun_tcp", "10.99.0.1", 1500, 1, NULL, 0, 0);
if (!p) { printf("[FAIL] tcp_proxy_create without mapping\n"); uasync_destroy(ua, 0); return 1; }
tcp_proxy_destroy(p);
uasync_destroy(ua, 0);
close(fds2[1]);
check_no_leaks("create/destroy without mapping");
}
close(fds[1]);
if (g_ok) printf("[PASS] test_tcp_proxy — create/destroy\n");
return g_ok ? 0 : 1;
}

Loading…
Cancel
Save