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

Scope TCP proxy streams and restart messages to their peer

proxy
evgeny 2 days ago
parent
268243feeb
commit
f19ff11b36
3 changed files with 34 additions and 18 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 9
      src/proxy/tcp_proxy_client.c
  2. 35
      src/proxy/tcp_proxy_server.c
  3. 8
      src/proxy/tcp_proxy_server.h

9
src/proxy/tcp_proxy_client.c
Unescape View File

@ -682,6 +682,15 @@ void tcp_proxy_client_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* en
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return; return;
} }
if (entry->len < ROUTER_SVC_HDR_SIZE) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t peer, group;
memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
if (!proxy || peer != proxy->via_node_id || group != TOPO_GROUP_UTUN) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "proxy client: unexpected peer=%016llx group=%016llx",
(unsigned long long)peer, (unsigned long long)group);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
// CLOSE_ALL / restart-уведомление: [svc_id][src][dst] без payload // CLOSE_ALL / restart-уведомление: [svc_id][src][dst] без payload
if (entry->len == ROUTER_SVC_HDR_SIZE && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY_CLIENT) { if (entry->len == ROUTER_SVC_HDR_SIZE && entry->dgram[0] == ETCP_RT_ID_TCP_PROXY_CLIENT) {
uint64_t peer_id; uint64_t peer_id;

35
src/proxy/tcp_proxy_server.c
Unescape View File

@ -338,9 +338,9 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) {
u_free(rc); u_free(rc);
} }
struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint32_t 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; struct tcp_proxy_server_conn* c;
for (c = ctx->conns; c; c = c->next) if (c->stream_id == stream_id) return c; for (c = ctx->conns; c; c = c->next) if (c->peer_node_id == peer && c->stream_id == stream_id) return c;
return NULL; return NULL;
} }
@ -351,6 +351,11 @@ struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server
int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry, uint32_t stream_id, uint64_t src_node_id) { 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; } 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; 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; } 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; uint8_t* dest_ip = entry->dgram + TCP_PROXY_RECV_HDR_SIZE;
uint16_t dest_port = 0; memcpy(&dest_port, dest_ip + 4, 2); uint16_t dest_port = 0; memcpy(&dest_port, dest_ip + 4, 2);
@ -422,9 +427,10 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry*
int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id) { int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id) {
(void)conn; (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; } 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* ctx = &inst->tcp_proxy_server;
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); 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) { 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); 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; queue_dgram_free(entry); queue_entry_free(entry); return -1;
@ -455,10 +461,10 @@ int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* c
return 0; return 0;
} }
void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_id) { void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id) {
if (!inst) return; if (!inst) return;
struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); 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; } 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", 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 ? (int)rc->tc->sock : -1, stream_id, conn_total(rc),
@ -477,21 +483,21 @@ void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_i
if (rc->tc->connected) tcp_conn_push_close(rc->tc); else tcp_proxy_server_conn_free(rc); if (rc->tc->connected) tcp_conn_push_close(rc->tc); else tcp_proxy_server_conn_free(rc);
} }
void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint32_t stream_id) { void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id) {
if (!inst) return; if (!inst) return;
struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); 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; } 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", 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 ? (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->tc ? rc->tc->fin_remote : 0, rc->tc ? write_pending(rc->tc) : 0);
tcp_proxy_server_handle_close(inst, stream_id); tcp_proxy_server_handle_close(inst, peer, stream_id);
} }
void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint32_t stream_id) { void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id) {
if (!inst) return; if (!inst) return;
struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; struct tcp_proxy_server* ctx = &inst->tcp_proxy_server;
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, stream_id); struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(ctx, peer, stream_id);
if (!rc || !rc->tc || !rc->tc->connected) return; if (!rc || !rc->tc || !rc->tc->connected) return;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FIN_RECV fd=%d sid=%08x — pushing FIN to wq", DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "SOCK:FIN_RECV fd=%d sid=%08x — pushing FIN to wq",
(int)rc->tc->sock, stream_id); (int)rc->tc->sock, stream_id);
@ -529,6 +535,7 @@ void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
queue_dgram_free(entry); queue_entry_free(entry); queue_dgram_free(entry); queue_entry_free(entry);
return; return;
} }
uint64_t peer; memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
uint8_t subcmd = entry->dgram[ROUTER_SVC_PAYLOAD_OFF]; uint8_t subcmd = entry->dgram[ROUTER_SVC_PAYLOAD_OFF];
uint32_t stream_id; memcpy(&stream_id, entry->dgram + ROUTER_SVC_PAYLOAD_OFF + 1, 4); uint32_t stream_id; memcpy(&stream_id, entry->dgram + ROUTER_SVC_PAYLOAD_OFF + 1, 4);
@ -541,12 +548,12 @@ void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
return; return;
} }
if (inst && inst->tcp_proxy_server.enabled) { if (inst && inst->tcp_proxy_server.enabled) {
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id); struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, peer, stream_id);
if (rc) { if (rc) {
if (subcmd == TCP_PROXY_SUBCMD_DATA) { tcp_proxy_server_handle_data(inst, conn, entry, stream_id); 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, stream_id); queue_dgram_free(entry); queue_entry_free(entry); 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, 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, 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) { if (subcmd == TCP_PROXY_SUBCMD_DATA || subcmd == TCP_PROXY_SUBCMD_CLOSE || subcmd == TCP_PROXY_SUBCMD_FIN || subcmd == TCP_PROXY_SUBCMD_ERROR) {

8
src/proxy/tcp_proxy_server.h
Unescape View File

@ -70,7 +70,7 @@ struct tcp_proxy_server {
int tcp_proxy_server_init(struct UTUN_INSTANCE* inst); int tcp_proxy_server_init(struct UTUN_INSTANCE* inst);
void tcp_proxy_server_destroy(struct UTUN_INSTANCE* inst); void tcp_proxy_server_destroy(struct UTUN_INSTANCE* inst);
struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint32_t stream_id); struct tcp_proxy_server_conn* tcp_proxy_server_find_conn(struct tcp_proxy_server* ctx, uint64_t peer, uint32_t stream_id);
void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc); void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc);
void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry); void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);
@ -78,9 +78,9 @@ void tcp_proxy_server_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);
int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry, int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* entry,
uint32_t stream_id, uint64_t src_node_id); uint32_t stream_id, uint64_t src_node_id);
int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id); int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* conn, struct ll_entry* entry, uint32_t stream_id);
void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint32_t stream_id); void tcp_proxy_server_handle_close(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id);
void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint32_t stream_id); void tcp_proxy_server_handle_error(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id);
void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint32_t stream_id); void tcp_proxy_server_handle_fin(struct UTUN_INSTANCE* inst, uint64_t peer, uint32_t stream_id);
#ifdef __cplusplus #ifdef __cplusplus

Write Preview
Loading…
Cancel
Save