Browse Source

fix lwip_tcp port byte order (double-htons), mapping lookup, socket error handling

- Fix port encoding: PCB stores ports in host byte order, tcp_create_segment uses htons
- Fix tcp_bind/tcp_connect: ntohs incoming network-order ports
- Fix demux comparisons: use host-order sport/dport statics instead of in-place ntohs
- Fix tcp_output_alloc_header_common: htons for port output
- Add static sport/dport variables (host byte order) to avoid tcphdr corruption
- Fix sock_transport_error_callback: handle EPOLLERR during non-blocking connect
- Fix mapping local_port: store host-order port for find_mapping_by_port
- Fix tcp_proxy_raw_read/tun_input: use IP total length for packet boundaries
- test_tcp_proxy: reduce size to 2KB, fix port, add debug_init, add echo server log
- test_tcp_proxy handshake and basic data flow now working (known: cwnd/poll throughput tuning needed)
congestion
Evgeny 5 months ago
parent
commit
ad8bc0ae00
  1. 4
      src/lwip_tcp/lwip_tcp.c
  2. 31
      src/lwip_tcp/lwip_tcp_in.c
  3. 4
      src/lwip_tcp/lwip_tcp_out.c
  4. 89
      src/tcp_proxy.c
  5. 6
      tests/test_tcp_proxy.c

4
src/lwip_tcp/lwip_tcp.c

@ -407,7 +407,7 @@ err_t tcp_bind(struct tcp_pcb *pcb, uint32_t ipaddr, uint16_t port)
}
if (ipaddr != 0) pcb->local_ip = ipaddr;
pcb->local_port = new_port;
pcb->local_port = ntohs(new_port);
TCP_REG(&ctx->bound_pcbs, pcb);
return LERR_OK;
}
@ -563,7 +563,7 @@ err_t tcp_connect(struct tcp_pcb *pcb, uint32_t ipaddr, uint16_t port, tcp_conne
if (pcb->state != CLOSED) return LERR_ISCONN;
pcb->remote_ip = ipaddr;
pcb->remote_port = port;
pcb->remote_port = ntohs(port);
old_local_port = pcb->local_port;
if (pcb->local_port == 0) {

31
src/lwip_tcp/lwip_tcp_in.c

@ -26,6 +26,7 @@ static uint16_t tcphdr_opt1len;
static uint8_t *tcphdr_opt2;
static uint16_t tcp_optidx;
static uint32_t seqno, ackno;
static uint16_t sport, dport; // host byte order
static tcpwnd_size_t recv_acked;
static uint16_t tcplen;
static uint8_t flags;
@ -133,6 +134,9 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
tcp_dst_ip = dst_ip;
tcphdr = (struct tcp_hdr *)p->payload;
// Save port values in network byte order before pbuf_header modifies pbuf state
uint16_t wire_src_port = tcphdr->src;
uint16_t wire_dst_port = tcphdr->dest;
if (p->len < TCP_HLEN) {
DEBUG_ERROR(DEBUG_CATEGORY_ALL, "tcp_input: short packet (%u bytes) discarded", p->tot_len);
@ -168,8 +172,9 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
p->tot_len = (uint16_t)(p->tot_len - opt2len);
}
tcphdr->src = ntohs(tcphdr->src);
tcphdr->dest = ntohs(tcphdr->dest);
sport = ntohs(tcphdr->src);
dport = ntohs(tcphdr->dest);
// Keep tcphdr->src and tcphdr->dest in network byte order for demux
seqno = tcphdr->seqno = ntohl(tcphdr->seqno);
ackno = tcphdr->ackno = ntohl(tcphdr->ackno);
tcphdr->wnd = ntohs(tcphdr->wnd);
@ -187,8 +192,8 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
prev = NULL;
for (pcb = ctx->active_pcbs; pcb != NULL; pcb = pcb->next) {
if (pcb->remote_port == tcphdr->src &&
pcb->local_port == tcphdr->dest &&
if (pcb->remote_port == sport &&
pcb->local_port == dport &&
pcb->remote_ip == src_ip &&
pcb->local_ip == dst_ip) {
if (prev != NULL) {
@ -203,8 +208,8 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
if (pcb == NULL) {
for (pcb = ctx->tw_pcbs; pcb != NULL; pcb = pcb->next) {
if (pcb->remote_port == tcphdr->src &&
pcb->local_port == tcphdr->dest &&
if (pcb->remote_port == sport &&
pcb->local_port == dport &&
pcb->remote_ip == src_ip &&
pcb->local_ip == dst_ip) {
tcp_timewait_input(pcb);
@ -216,7 +221,7 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
prev = NULL;
for (lpcb = (struct tcp_pcb_listen *)ctx->listen_pcbs;
lpcb != NULL; lpcb = (struct tcp_pcb_listen *)lpcb->next) {
if (lpcb->local_port == tcphdr->dest) {
if (lpcb->local_port == dport) {
if (lpcb->local_ip == 0 || lpcb->local_ip == dst_ip) {
break;
}
@ -315,7 +320,7 @@ void lwip_tcp_input(struct lwip_tcp_ctx *ctx, struct pbuf *p,
} else {
if (!(TCPH_FLAGS(tcphdr) & TCP_RST)) {
tcp_rst(NULL, ackno, seqno + tcplen, dst_ip, src_ip,
tcphdr->dest, tcphdr->src);
dport, sport);
}
pbuf_free(p);
}
@ -353,7 +358,7 @@ static void tcp_listen_input(struct tcp_pcb_listen *pcb)
if (flags & TCP_ACK) {
tcp_rst((struct tcp_pcb *)pcb, ackno, seqno + tcplen,
tcp_dst_ip, tcp_src_ip, tcphdr->dest, tcphdr->src);
tcp_dst_ip, tcp_src_ip, dport, sport);
} else if (flags & TCP_SYN) {
DEBUG_INFO(DEBUG_CATEGORY_ALL, "TCP connection request %u -> %u",
(unsigned)tcphdr->src, (unsigned)tcphdr->dest);
@ -366,7 +371,7 @@ static void tcp_listen_input(struct tcp_pcb_listen *pcb)
npcb->local_ip = tcp_dst_ip;
npcb->remote_ip = tcp_src_ip;
npcb->local_port = pcb->local_port;
npcb->remote_port = tcphdr->src;
npcb->remote_port = sport;
npcb->state = SYN_RCVD;
npcb->rcv_nxt = seqno + 1;
npcb->rcv_ann_right_edge = npcb->rcv_nxt;
@ -405,7 +410,7 @@ static void tcp_timewait_input(struct tcp_pcb *pcb)
if (flags & TCP_SYN) {
if (TCP_SEQ_BETWEEN(seqno, pcb->rcv_nxt, pcb->rcv_nxt + pcb->rcv_wnd)) {
tcp_rst(pcb, ackno, seqno + tcplen, tcp_dst_ip, tcp_src_ip,
tcphdr->dest, tcphdr->src);
dport, sport);
return;
}
} else if (flags & TCP_FIN) {
@ -496,7 +501,7 @@ static err_t tcp_process(struct tcp_pcb *pcb)
tcp_ack_now(pcb);
} else if (flags & TCP_ACK) {
tcp_rst(pcb, ackno, seqno + tcplen, pcb->local_ip, pcb->remote_ip,
tcphdr->dest, tcphdr->src);
dport, sport);
if (pcb->nrtx < TCP_SYNMAXRTX) {
pcb->rtime = 0;
tcp_rexmit_rto(pcb);
@ -531,7 +536,7 @@ static err_t tcp_process(struct tcp_pcb *pcb)
}
} else {
tcp_rst(pcb, ackno, seqno + tcplen, pcb->local_ip, pcb->remote_ip,
tcphdr->dest, tcphdr->src);
dport, sport);
}
} else if ((flags & TCP_SYN) && (seqno == pcb->rcv_nxt - 1)) {
tcp_rexmit(pcb);

4
src/lwip_tcp/lwip_tcp_out.c

@ -479,6 +479,8 @@ tcp_output(struct tcp_pcb *pcb)
if (pcb->local_ip == 0) return LERR_RTE;
if ((ntohl(seg->tcphdr->seqno) - pcb->lastack + seg->len) > wnd) {
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP_OUTPUT cwnd limited: seqno_diff=%u seg_len=%u wnd=%u cwnd=%u snd_wnd=%u",
ntohl(seg->tcphdr->seqno) - pcb->lastack, seg->len, wnd, pcb->cwnd, pcb->snd_wnd);
if (wnd == pcb->snd_wnd && pcb->unacked == NULL && pcb->persist_backoff == 0) {
pcb->persist_cnt = 0;
pcb->persist_backoff = 1;
@ -504,6 +506,8 @@ tcp_output(struct tcp_pcb *pcb)
}
err = tcp_output_segment(seg, pcb);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP_OUTPUT seg seqno=%u len=%u wnd=%u cwnd=%u ret=%d",
ntohl(seg->tcphdr->seqno), seg->len, wnd, pcb->cwnd, err);
if (err != LERR_OK) {
tcp_set_flags(pcb, TF_NAGLEMEMERR);
return err;

89
src/tcp_proxy.c

@ -126,6 +126,10 @@ static err_t tcp_output_cb(void *arg, struct pbuf *p, uint32_t src_ip, uint32_t
if (len > 2000) return LERR_BUF;
uint8_t buf[2000];
pbuf_copy_partial(p, buf, len, 0);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP OUT %u.%u.%u.%u→%u.%u.%u.%u len=%u fd=%d",
(uint8_t)(src_ip), (uint8_t)(src_ip>>8), (uint8_t)(src_ip>>16), (uint8_t)(src_ip>>24),
(uint8_t)(dst_ip), (uint8_t)(dst_ip>>8), (uint8_t)(dst_ip>>16), (uint8_t)(dst_ip>>24),
len, proxy->ip_fd);
if (proxy->tun) tun_platform_write(proxy->tun, buf, len);
else if (proxy->ip_fd >= 0) { ssize_t n = write(proxy->ip_fd, buf, len); (void)n; }
return LERR_OK;
@ -168,14 +172,16 @@ static int tcp_proxy_handle_non_tcp(struct tcp_proxy* p, uint8_t* buf, size_t le
static void proxy_feed_from_transport(struct proxy_conn *pc) {
if (!pc->transport_to_uip || !pc->pcb) return;
int sent_any = 0;
int entries_fed = 0;
while (1) {
uint16_t space = tcp_sndbuf(pc->pcb);
if (space < TCP_MSS / 2) break;
if (space < TCP_MSS / 2) { DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "FEED space=%u<ssthresh break active=%d", space, pc->active); break; }
struct ll_entry *e = queue_data_get(pc->transport_to_uip);
if (!e) break;
uint16_t len = e->len;
if (len > space) len = space;
err_t ret = tcp_write(pc->pcb, e->dgram, len, TCP_WRITE_FLAG_COPY);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "FEED tcp_write len=%u ret=%d active=%d cwnd=%u snd_wnd=%u unsent=%p", len, ret, pc->active, pc->pcb->cwnd, pc->pcb->snd_wnd, (void*)pc->pcb->unsent);
if (ret == LERR_OK) {
sent_any = 1;
if (len >= e->len) { queue_dgram_free(e); queue_entry_free(e); }
@ -209,7 +215,7 @@ static err_t proxy_accept_cb(void *arg, struct tcp_pcb *newpcb, err_t err) {
tcp_recv(newpcb, proxy_recv_cb);
tcp_sent(newpcb, proxy_sent_cb);
tcp_err(newpcb, proxy_err_cb);
tcp_poll(newpcb, proxy_poll_cb, 2);
tcp_poll(newpcb, proxy_poll_cb, 0);
tcp_nagle_disable(newpcb);
pc->uip_to_transport = queue_new(p->ua, 0, 0, 0, "uip_to_transport");
@ -255,12 +261,14 @@ static err_t proxy_recv_cb(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t
}
if (pc->active) {
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "RECV active pc=%p len=%u", (void*)pc, p->tot_len);
struct ll_entry *e = entry_from_data(pc->proxy->entry_pool, p->payload, p->tot_len);
if (e) queue_data_put(pc->uip_to_transport, e);
tcp_recved(pcb, p->tot_len);
pbuf_free(p);
} else if (pc->transport && !pc->half_closed) {
uint16_t len = p->tot_len;
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "RECV passive pc=%p len=%u transport=%p connected=%d", (void*)pc, len, (void*)pc->transport, ((struct sock_transport*)pc->transport)->connected);
uint8_t *data = u_malloc(len);
if (data) {
pbuf_copy_partial(p, data, len, 0);
@ -313,8 +321,9 @@ static err_t proxy_connected_cb(void *arg, struct tcp_pcb *pcb, err_t err) {
pc->closing = 1;
return LERR_ABRT;
}
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TCP proxy: active connect established to %d.%d.%d.%d:%d",
pc->dest_ip[0], pc->dest_ip[1], pc->dest_ip[2], pc->dest_ip[3], ntohs(pc->dest_port));
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP proxy: active connect established to %d.%d.%d.%d:%d cwnd=%u snd_wnd=%u snd_buf=%u",
pc->dest_ip[0], pc->dest_ip[1], pc->dest_ip[2], pc->dest_ip[3], ntohs(pc->dest_port),
pcb->cwnd, pcb->snd_wnd, pcb->snd_buf);
proxy_feed_from_transport(pc);
return LERR_OK;
}
@ -326,18 +335,31 @@ static void tcp_proxy_raw_read(int fd, void* arg) {
(void)fd; struct tcp_proxy* p = (struct tcp_proxy*)arg;
uint8_t buf[2000]; ssize_t n = read(p->ip_fd, buf, sizeof(buf));
if (n <= 0) return;
if (tcp_proxy_handle_non_tcp(p, buf, n)) return;
if ((size_t)n < 20) return;
uint8_t proto = buf[9];
uint32_t src_ip, dst_ip;
memcpy(&src_ip, buf + 12, 4); memcpy(&dst_ip, buf + 16, 4);
if (proto != IPPROTO_TCP) return;
uint16_t ip_hdr_len = (buf[0] & 0x0F) * 4;
if ((size_t)n < ip_hdr_len) return;
struct pbuf *pb = pbuf_alloc(PBUF_RAW, (uint16_t)(n - ip_hdr_len));
if (!pb) return;
pbuf_take(pb, buf + ip_hdr_len, (uint16_t)(n - ip_hdr_len));
lwip_tcp_input(p->lwip, pb, src_ip, dst_ip);
size_t offset = 0;
while (offset < (size_t)n) {
size_t remaining = (size_t)n - offset;
if (remaining < 20) break;
uint8_t *pkt = buf + offset;
if (tcp_proxy_handle_non_tcp(p, pkt, remaining)) { offset += remaining; continue; }
uint8_t proto = pkt[9];
if (proto != IPPROTO_TCP) { offset += remaining; continue; }
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) break;
uint32_t src_ip, dst_ip;
memcpy(&src_ip, pkt + 12, 4); memcpy(&dst_ip, pkt + 16, 4);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP IN %u.%u.%u.%u→%u.%u.%u.%u proto=%u iplen=%u",
(uint8_t)(src_ip), (uint8_t)(src_ip>>8), (uint8_t)(src_ip>>16), (uint8_t)(src_ip>>24),
(uint8_t)(dst_ip), (uint8_t)(dst_ip>>8), (uint8_t)(dst_ip>>16), (uint8_t)(dst_ip>>24),
proto, ip_total);
uint16_t tcp_len = ip_total - ip_hdr_len;
struct pbuf *pb = pbuf_alloc(PBUF_RAW, tcp_len);
if (!pb) break;
pbuf_take(pb, pkt + ip_hdr_len, tcp_len);
lwip_tcp_input(p->lwip, pb, src_ip, dst_ip);
offset += ip_total;
}
}
static void tcp_proxy_tun_input(struct ll_queue* q, void* arg) {
@ -351,11 +373,13 @@ static void tcp_proxy_tun_input(struct ll_queue* q, void* arg) {
uint8_t proto = ip[9];
if (proto == IPPROTO_TCP) {
uint16_t ip_hdr_len = (ip[0] & 0x0F) * 4;
if (len >= ip_hdr_len) {
uint16_t ip_total = ((uint16_t)ip[2] << 8) | ip[3];
if (ip_hdr_len >= 20 && ip_total >= ip_hdr_len && len >= ip_total) {
uint32_t src_ip, dst_ip;
memcpy(&src_ip, ip + 12, 4); memcpy(&dst_ip, ip + 16, 4);
struct pbuf *pb = pbuf_alloc(PBUF_RAW, (uint16_t)(len - ip_hdr_len));
if (pb) { pbuf_take(pb, ip + ip_hdr_len, (uint16_t)(len - ip_hdr_len)); lwip_tcp_input(p->lwip, pb, src_ip, dst_ip); }
uint16_t tcp_len = ip_total - ip_hdr_len;
struct pbuf *pb = pbuf_alloc(PBUF_RAW, tcp_len);
if (pb) { pbuf_take(pb, ip + ip_hdr_len, tcp_len); lwip_tcp_input(p->lwip, pb, src_ip, dst_ip); }
}
}
}
@ -445,6 +469,7 @@ static void sock_transport_read_callback(socket_t sock, void* arg) {
static void sock_transport_write_callback(socket_t sock, void* arg) {
(void)sock; struct sock_transport* st = (struct sock_transport*)arg;
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "WRITE_CB st=%p connected=%d connect_called=%d", (void*)st, st->connected, st->connect_called);
if(!st->connected) {
if(!st->connect_called) return;
int err = 0; socklen_t len = sizeof(err);
@ -475,7 +500,23 @@ static void sock_transport_write_callback(socket_t sock, void* arg) {
static void sock_transport_error_callback(socket_t sock, void* arg) {
(void)sock; struct sock_transport* st = (struct sock_transport*)arg;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: socket error"); sock_transport_close(&st->base);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "ERROR_CB st=%p sock=%d connect_called=%d connected=%d", (void*)st, st->sock, st->connect_called, st->connected);
// For non-blocking connect, EPOLLERR can fire before EPOLLOUT.
// Check SO_ERROR: if connect succeeded (err==0), trigger write callback.
if (st->connect_called && !st->connected) {
int err = 0; socklen_t len = sizeof(err);
if (getsockopt(st->sock, SOL_SOCKET, SO_ERROR, &err, &len) == 0) {
if (err == 0) {
sock_transport_write_callback(st->sock, st);
return;
}
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: connect error fd=%d err=%d (%s)", st->sock, err, strerror(err));
sock_transport_close(&st->base);
return;
}
}
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "TCP proxy: socket error fd=%d", st->sock);
sock_transport_close(&st->base);
}
// ====================================================================
@ -668,13 +709,13 @@ struct tcp_proxy* tcp_proxy_create(struct UTUN_INSTANCE* inst, struct UASYNC* ua
if (listen_pcb) {
tcp_arg(listen_pcb, p);
tcp_accept(listen_pcb, proxy_accept_cb);
DEBUG_ERROR(DEBUG_CATEGORY_TRAFFIC, "TCP proxy: listen on %d ok lpcb=%p ctx_listen=%p",
mappings[j].local_port, (void*)listen_pcb, (void*)p->lwip->listen_pcbs);
}
struct tcp_proxy_mapping* m = u_calloc(1, sizeof(struct tcp_proxy_mapping));
if (m) { m->local_port = port_net; struct in_addr ra; ra.s_addr = inet_addr(mappings[j].remote_ip);
if (m) { m->local_port = mappings[j].local_port; struct in_addr ra; ra.s_addr = inet_addr(mappings[j].remote_ip);
memcpy(m->remote_ip, &ra.s_addr, 4); m->remote_port = htons(mappings[j].remote_port); m->dynamic = 0;
m->next = p->mappings; p->mappings = m;
DEBUG_INFO(DEBUG_CATEGORY_TUN, "TCP proxy mapping: %d -> %s:%d%s", mappings[j].local_port, mappings[j].remote_ip, mappings[j].remote_port,
p->has_remote_mappings ? " (remote)" : ""); }
m->next = p->mappings; p->mappings = m; }
}
}

6
tests/test_tcp_proxy.c

@ -19,8 +19,8 @@
#include "../lib/debug_config.h"
#define TEST_PORT 9090
#define ECHO_PORT 9999
#define TEST_SIZE (1024 * 1024)
#define ECHO_PORT 19999
#define TEST_SIZE (2 * 1024) // 2KB — minimum viable test
#define POLL_TIMEOUT_MS 10000
#define DATA_TIMEOUT_MS 120000
@ -34,6 +34,7 @@ static void echo_server(uint16_t port) {
struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons(port)}; addr.sin_addr.s_addr = inet_addr("127.0.0.1");
if(bind(srv, (struct sockaddr*)&addr, sizeof(addr)) < 0) { perror("echo bind"); close(srv); exit(1); }
if(listen(srv, 1) < 0) { perror("echo listen"); close(srv); exit(1); }
fprintf(stderr, "ECHO: listening on port %d\n", port);
int cli = accept(srv, NULL, NULL);
if(cli < 0) { perror("echo accept"); close(srv); exit(1); }
uint8_t buf[65536]; ssize_t n;
@ -51,6 +52,7 @@ static void run_instance_b(int ip_fd) {
}
int main(void) {
debug_config_init();
// 1. Fork echo server
echo_pid = fork();
if(echo_pid == 0) { echo_server(ECHO_PORT); _exit(0); }

Loading…
Cancel
Save