// Общий TCP proxy протокол. DATA ограничен окном каждого потока; WINDOW возвращает // кредит после потребления данных локальным TCP. FIN следует за всеми DATA своего // направления. CONNECTED подтверждает реальный connect exit → destination. #ifndef PROXY_PROTOCOL_H #define PROXY_PROTOCOL_H #include #include #include "../routing_layer/etcp_router.h" #include "../routing_layer/topo_node.h" #include "../../lib/u_async.h" #include "../../lib/ll_queue.h" #include "../../lib/mem.h" #include "../../lib/debug_config.h" #define TCP_PROXY_SUBCMD_CONNECT 0x01 #define TCP_PROXY_SUBCMD_CONNECTED 0x02 #define TCP_PROXY_SUBCMD_DATA 0x03 #define TCP_PROXY_SUBCMD_CLOSE 0x04 #define TCP_PROXY_SUBCMD_ERROR 0x05 #define TCP_PROXY_SUBCMD_FIN 0x06 #define TCP_PROXY_SUBCMD_WINDOW 0x07 #define TCP_PROXY_HDR_SIZE 6 #define TCP_PROXY_CONNECT_HDR_SIZE 12 #define TCP_PROXY_RECV_HDR_SIZE (ROUTER_SVC_PAYLOAD_OFF + 5) #define TCP_PROXY_CHUNK 4096 #define TCP_PROXY_WINDOW 65536u struct proxy_flow { struct UTUN_INSTANCE* inst; struct UASYNC* ua; uint64_t peer; uint32_t sid; uint8_t svc; uint8_t ready; uint8_t connected_pending; uint8_t fin_pending; uint8_t fin_sent; uint8_t fin_received; uint32_t tx_credit; uint32_t rx_credit; uint32_t consumed; void* retry; void (*wake)(void* arg); void* arg; }; // Сообщение целиком передаётся маршрутизатору, который освобождает entry при любом результате. static inline int proxy_send(struct UTUN_INSTANCE* inst, uint64_t peer, uint8_t svc, uint8_t cmd, uint32_t sid, const uint8_t* data, size_t len, int force) { if (len > TCP_PROXY_CHUNK) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: oversized message sid=%08x cmd=%u len=%zu", sid, cmd, len); return -1; } struct ll_entry* e = queue_entry_new(0); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: entry allocation failed sid=%08x", sid); return -1; } e->dgram = (uint8_t*)u_malloc(TCP_PROXY_HDR_SIZE + len); if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: payload allocation failed sid=%08x", sid); queue_entry_free(e); return -1; } e->dgram[0] = svc; e->dgram[1] = cmd; memcpy(e->dgram + 2, &sid, 4); if (len) memcpy(e->dgram + TCP_PROXY_HDR_SIZE, data, len); e->len = TCP_PROXY_HDR_SIZE + len; return etcp_route_send(inst, TOPO_GROUP_UTUN, peer, e, force, 0); } static inline void proxy_flow_flush(struct proxy_flow* f); // Повторяем только недоставленные control-сообщения; DATA остаётся у владельца потока. static inline void proxy_flow_retry(void* arg) { struct proxy_flow* f = (struct proxy_flow*)arg; f->retry = NULL; proxy_flow_flush(f); if (f->wake) f->wake(f->arg); } static inline void proxy_flow_init(struct proxy_flow* f, struct UTUN_INSTANCE* inst, struct UASYNC* ua, uint64_t peer, uint8_t svc, uint32_t sid, void (*wake)(void*), void* arg) { memset(f, 0, sizeof(*f)); f->inst = inst; f->ua = ua; f->peer = peer; f->svc = svc; f->sid = sid; f->tx_credit = TCP_PROXY_WINDOW; f->rx_credit = TCP_PROXY_WINDOW; f->wake = wake; f->arg = arg; } static inline void proxy_flow_destroy(struct proxy_flow* f) { if (f->retry) { uasync_cancel_timeout(f->ua, f->retry); f->retry = NULL; } } // Сначала CONNECTED, затем WINDOW и FIN. Force обходит лишь очередь router, не окно потока. static inline void proxy_flow_flush(struct proxy_flow* f) { if (f->connected_pending) { if (proxy_send(f->inst, f->peer, f->svc, TCP_PROXY_SUBCMD_CONNECTED, f->sid, NULL, 0, 1) < 0) goto retry; f->connected_pending = 0; f->ready = 1; } if (f->consumed) { uint32_t credit = f->consumed; // Резервируем до синхронной loopback-доставки. f->consumed = 0; f->rx_credit += credit; if (proxy_send(f->inst, f->peer, f->svc, TCP_PROXY_SUBCMD_WINDOW, f->sid, (uint8_t*)&credit, 4, 1) < 0) { f->consumed += credit; f->rx_credit -= credit; goto retry; } } if (f->fin_pending && f->ready) { if (proxy_send(f->inst, f->peer, f->svc, TCP_PROXY_SUBCMD_FIN, f->sid, NULL, 0, 1) < 0) goto retry; f->fin_pending = 0; f->fin_sent = 1; DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "proxy: FIN queued peer=%016llx sid=%08x", (unsigned long long)f->peer, f->sid); } return; retry: DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "proxy: control retry sid=%08x connected=%u credit=%u fin=%u", f->sid, f->connected_pending, f->consumed, f->fin_pending); if (!f->retry) f->retry = uasync_set_timeout(f->ua, 5000, f, proxy_flow_retry, "proxy_control"); } // 0 = отправлено, -2 = ждём окно/CONNECTED, -1 = router backpressure/ошибка выделения. static inline int proxy_flow_send(struct proxy_flow* f, const uint8_t* data, size_t len, int force) { if (!f->ready || len > f->tx_credit) return -2; if (!len || len > TCP_PROXY_CHUNK || f->fin_sent) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: invalid DATA sid=%08x len=%zu fin=%u", f->sid, len, f->fin_sent); return -1; } f->tx_credit -= len; int ret = proxy_send(f->inst, f->peer, f->svc, TCP_PROXY_SUBCMD_DATA, f->sid, data, len, force); if (ret < 0) f->tx_credit += len; return ret; } static inline int proxy_flow_receive(struct proxy_flow* f, size_t len) { if (!len || len > TCP_PROXY_CHUNK || len > f->rx_credit || f->fin_received) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: invalid receive sid=%08x len=%zu credit=%u fin=%u", f->sid, len, f->rx_credit, f->fin_received); return -1; } f->rx_credit -= len; return 0; } static inline int proxy_flow_window(struct proxy_flow* f, const uint8_t* data, size_t len) { uint32_t credit = 0; if (len == 4) memcpy(&credit, data, 4); if (!credit || credit > TCP_PROXY_WINDOW - f->tx_credit) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "proxy: invalid WINDOW sid=%08x len=%zu credit=%u tx=%u", f->sid, len, credit, f->tx_credit); return -1; } f->tx_credit += credit; DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "proxy: WINDOW sid=%08x add=%u available=%u", f->sid, credit, f->tx_credit); return 0; } // Вызывать только после передачи байтов локальному TCP, не после получения ETCP DATA. static inline void proxy_flow_consume(struct proxy_flow* f, uint32_t len) { f->consumed += len; proxy_flow_flush(f); } #endif