/* * msg_transport.c — TCP-сервер транспорта сообщений (local IPC → ETCP) * * Принимает TCP-соединения от локальных приложений (протокол msg_transport.h), * привязывает к remote peer и пересылает данные через etcp_router. */ #include "msg_transport.h" #include "utun_instance.h" #include "etcp_router.h" #include "etcp.h" #include "secure_channel.h" #include "route_bgp.h" #include "../lib/u_async.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include #include #ifdef _WIN32 #include #include #else #include #include #include #include #include #include #endif #ifndef DEBUG_CATEGORY_MSGTRANSPORT #define DEBUG_CATEGORY_MSGTRANSPORT DEBUG_CATEGORY_GENERAL #endif /* ============================================================================ * Внутренние структуры * ============================================================================ */ struct msg_client { struct msg_client* next; struct msg_transport* transport; socket_t fd; void* socket_id; uint8_t recv_buffer[MSG_MAX_MSG_SIZE]; uint16_t recv_len; uint64_t bound_peer_id; /* 0 = не забинден */ uint8_t is_localhost; /* 1 = подключение с loopback */ }; struct msg_transport { struct UTUN_INSTANCE* instance; struct UASYNC* ua; socket_t listen_fd; void* listen_socket_id; struct sockaddr_storage bind_addr; struct msg_client* clients; uint32_t client_count; uint32_t max_clients; uint8_t svc_id; struct msg_transport_counters counters; }; /* ============================================================================ * Forward declarations * ============================================================================ */ static void accept_callback(socket_t fd, void* arg); static void client_read_callback(socket_t fd, void* arg); static void client_except_callback(socket_t fd, void* arg); static void close_client(struct msg_transport* t, struct msg_client* client); static void handle_client_data(struct msg_transport* t, struct msg_client* client); static void on_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry); static int forward_to_peer(struct msg_transport* t, uint64_t dst_peer_id, const uint8_t* data, size_t data_len, int is_signed); static void send_response(struct msg_client* client, uint8_t type, uint64_t id, const uint8_t* data, size_t data_len); static void send_error(struct msg_client* client, uint8_t error_code, const char* msg); /* ============================================================================ * Инициализация / завершение * ============================================================================ */ int msg_transport_init(struct msg_transport** out, struct UTUN_INSTANCE* instance, struct UASYNC* ua, struct sockaddr_storage* bind_addr, uint32_t max_clients, uint8_t svc_id) { if (!out || !instance || !ua || !bind_addr) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport_init: invalid args out=%p inst=%p ua=%p addr=%p", (void*)out, (void*)instance, (void*)ua, (void*)bind_addr); return -1; } if (bind_addr->ss_family == 0) { DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport_init: bind_addr has no family, skipping"); *out = NULL; return 0; } struct msg_transport* t = u_calloc(1, sizeof(*t)); if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport_init: u_calloc failed"); return -1; } t->instance = instance; t->ua = ua; t->max_clients = max_clients ? max_clients : 8; t->svc_id = svc_id; memcpy(&t->bind_addr, bind_addr, sizeof(*bind_addr)); int family = bind_addr->ss_family; t->listen_fd = socket(family, SOCK_STREAM, IPPROTO_TCP); #ifdef _WIN32 if (t->listen_fd == INVALID_SOCKET) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "socket() failed: %d", WSAGetLastError()); u_free(t); return -1; } u_long nonblock = 1; ioctlsocket(t->listen_fd, FIONBIO, &nonblock); int reuse = 1; setsockopt(t->listen_fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&reuse, sizeof(reuse)); #else if (t->listen_fd < 0) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "socket() failed: %s", strerror(errno)); u_free(t); return -1; } int reuse = 1; setsockopt(t->listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); int flags = fcntl(t->listen_fd, F_GETFL, 0); if (flags >= 0) fcntl(t->listen_fd, F_SETFL, flags | O_NONBLOCK); #endif socklen_t addr_len = (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6); #ifdef _WIN32 if (bind(t->listen_fd, (struct sockaddr*)bind_addr, addr_len) == SOCKET_ERROR) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "bind() failed: %d", WSAGetLastError()); closesocket(t->listen_fd); u_free(t); return -1; } if (listen(t->listen_fd, 5) == SOCKET_ERROR) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "listen() failed: %d", WSAGetLastError()); closesocket(t->listen_fd); u_free(t); return -1; } #else if (bind(t->listen_fd, (struct sockaddr*)bind_addr, addr_len) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "bind() failed: %s", strerror(errno)); close(t->listen_fd); u_free(t); return -1; } if (listen(t->listen_fd, 5) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "listen() failed: %s", strerror(errno)); close(t->listen_fd); u_free(t); return -1; } #endif t->listen_socket_id = uasync_add_socket_t(ua, t->listen_fd, accept_callback, NULL, NULL, t); if (!t->listen_socket_id) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "uasync_add_socket_t failed for listen_fd"); #ifdef _WIN32 closesocket(t->listen_fd); #else close(t->listen_fd); #endif u_free(t); return -1; } if (etcp_router_bind(instance, svc_id, on_etcp_recv) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "etcp_router_bind(svc_id=%u) failed", svc_id); uasync_remove_socket_t(ua, t->listen_fd); #ifdef _WIN32 closesocket(t->listen_fd); #else close(t->listen_fd); #endif u_free(t); return -1; } if (family == AF_INET) { struct sockaddr_in* sin = (struct sockaddr_in*)bind_addr; DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport listening on %s:%d svc_id=%u", ip_to_str(&sin->sin_addr, AF_INET).str, ntohs(sin->sin_port), svc_id); } else { struct sockaddr_in6* sin6 = (struct sockaddr_in6*)bind_addr; DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport listening on [%s]:%d svc_id=%u", ip_to_str(&sin6->sin6_addr, AF_INET6).str, ntohs(sin6->sin6_port), svc_id); } *out = t; return 0; } void msg_transport_shutdown(struct msg_transport* t) { if (!t) return; while (t->clients) close_client(t, t->clients); #ifdef _WIN32 if (t->listen_fd != INVALID_SOCKET) { uasync_remove_socket_t(t->ua, t->listen_fd); t->listen_socket_id = NULL; closesocket(t->listen_fd); t->listen_fd = INVALID_SOCKET; } #else if (t->listen_fd >= 0) { uasync_remove_socket_t(t->ua, t->listen_fd); t->listen_socket_id = NULL; close(t->listen_fd); t->listen_fd = -1; } #endif DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport shutdown complete svc_id=%u", t->svc_id); memset(t, 0, sizeof(*t)); u_free(t); } void msg_transport_get_counters(struct msg_transport* t, struct msg_transport_counters* c) { if (t && c) memcpy(c, &t->counters, sizeof(*c)); } /* ============================================================================ * TCP-клиенты: accept / read / close * ============================================================================ */ static void accept_callback(socket_t fd, void* arg) { struct msg_transport* t = (struct msg_transport*)arg; struct sockaddr_storage client_addr; socklen_t addr_len = sizeof(client_addr); #ifdef _WIN32 socket_t client_fd = accept(fd, (struct sockaddr*)&client_addr, &addr_len); if (client_fd == INVALID_SOCKET) { int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "accept() failed: %d", err); return; } u_long nonblock = 1; ioctlsocket(client_fd, FIONBIO, &nonblock); #else socket_t client_fd = accept(fd, (struct sockaddr*)&client_addr, &addr_len); if (client_fd < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "accept() failed: %s", strerror(errno)); return; } int flags = fcntl(client_fd, F_GETFL, 0); if (flags >= 0) fcntl(client_fd, F_SETFL, flags | O_NONBLOCK); #endif if (t->client_count >= t->max_clients) { DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "max_clients=%u reached, rejecting", t->max_clients); #ifdef _WIN32 closesocket(client_fd); #else close(client_fd); #endif return; } struct msg_client* client = u_calloc(1, sizeof(*client)); if (!client) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "u_calloc(msg_client) failed"); #ifdef _WIN32 closesocket(client_fd); #else close(client_fd); #endif return; } client->fd = client_fd; client->transport = t; /* Определяем loopback */ if (client_addr.ss_family == AF_INET) { struct sockaddr_in* sin = (struct sockaddr_in*)&client_addr; client->is_localhost = (ntohl(sin->sin_addr.s_addr) == 0x7F000001u); /* 127.0.0.1 */ } else if (client_addr.ss_family == AF_INET6) { struct sockaddr_in6* sin6 = (struct sockaddr_in6*)&client_addr; static const uint8_t loopback_v6[16] = {0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1}; client->is_localhost = (memcmp(&sin6->sin6_addr, loopback_v6, 16) == 0); } else { client->is_localhost = 0; } client->socket_id = uasync_add_socket_t(t->ua, client_fd, client_read_callback, NULL, client_except_callback, client); if (!client->socket_id) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "uasync_add_socket_t failed for client"); u_free(client); #ifdef _WIN32 closesocket(client_fd); #else close(client_fd); #endif return; } client->next = t->clients; t->clients = client; t->client_count++; t->counters.accepted++; DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport client connected (total=%u)", t->client_count); } static void client_read_callback(socket_t fd, void* arg) { struct msg_client* client = (struct msg_client*)arg; struct msg_transport* t = client->transport; uint8_t* buf = client->recv_buffer + client->recv_len; size_t buf_space = MSG_MAX_MSG_SIZE - client->recv_len; #ifdef _WIN32 int received = recv(fd, (char*)buf, (int)buf_space, 0); if (received == SOCKET_ERROR) { int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) { if (err == 10054) { close_client(t, client); return; } DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "recv() error: %d", err); close_client(t, client); } return; } #else ssize_t received = recv(fd, buf, buf_space, 0); if (received < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "recv() error: %s", strerror(errno)); close_client(t, client); } return; } #endif if (received == 0) { close_client(t, client); return; } client->recv_len += (uint16_t)received; if (client->recv_len > MSG_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "recv buffer overflow"); close_client(t, client); return; } handle_client_data(t, client); } static void client_except_callback(socket_t fd, void* arg) { struct msg_client* client = (struct msg_client*)arg; DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "client socket exception"); close_client(client->transport, client); (void)fd; } static void close_client(struct msg_transport* t, struct msg_client* client) { if (!t || !client) return; #ifdef _WIN32 if (client->fd != INVALID_SOCKET) { uasync_remove_socket_t(t->ua, client->fd); closesocket(client->fd); } #else if (client->fd >= 0) { uasync_remove_socket_t(t->ua, client->fd); close(client->fd); } #endif struct msg_client** curr = &t->clients; while (*curr) { if (*curr == client) { *curr = client->next; break; } curr = &(*curr)->next; } t->client_count--; t->counters.closed++; u_free(client); DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "msg_transport client disconnected (total=%u)", t->client_count); } /* ============================================================================ * Обработка входящих сообщений от TCP-клиента * ============================================================================ */ static void handle_client_data(struct msg_transport* t, struct msg_client* client) { while (client->recv_len >= MSG_HDR_SIZE) { struct msg_header* hdr = (struct msg_header*)client->recv_buffer; if (msg_validate_header(hdr) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "invalid header size=%u type=0x%02x", hdr->size, hdr->type); send_error(client, MSG_ERR_INVALID_MSG, "Invalid message header"); close_client(t, client); return; } if (client->recv_len < hdr->size) break; uint8_t* payload = client->recv_buffer + MSG_HDR_SIZE; uint16_t payload_size = hdr->size - MSG_HDR_SIZE; switch (hdr->type) { case MSG_CTL_BIND: { uint64_t peer_id = hdr->id; if (peer_id == 0) { send_error(client, MSG_ERR_INVALID_MSG, "BIND with zero peer_id"); break; } /* Проверить конфликт: этот peer_id уже занят другим клиентом */ struct msg_client* other = t->clients; while (other) { if (other != client && other->bound_peer_id == peer_id) { send_error(client, MSG_ERR_ID_TAKEN, "peer_id already bound by another client"); DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "BIND peer=%016llx rejected: already bound", (unsigned long long)peer_id); break; } other = other->next; } if (other) break; /* конфликт */ client->bound_peer_id = peer_id; send_response(client, MSG_RSP_BIND_OK, peer_id, NULL, 0); t->counters.bind_count++; DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "BIND peer=%016llx", (unsigned long long)peer_id); break; } case MSG_CTL_UNBIND: client->bound_peer_id = 0; t->counters.unbind_count++; DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "UNBIND"); break; case MSG_CTL_GET_PUBKEY: send_response(client, MSG_RSP_PUBKEY, 0, t->instance->my_keys.public_key, SC_PUBKEY_SIZE); break; case MSG_CTL_GET_PRIVKEY: if (!client->is_localhost) { send_error(client, MSG_ERR_ACCESS_DENIED, "private key only available from localhost"); DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "GET_PRIVKEY denied: not localhost"); break; } send_response(client, MSG_RSP_PRIVKEY, 0, t->instance->my_keys.private_key, SC_PRIVKEY_SIZE); break; case MSG_CTL_GET_PEER_INFO: { uint64_t peer_id = hdr->id; struct ROUTE_BGP* bgp = t->instance->bgp; if (!bgp) { send_error(client, MSG_ERR_PEER_NOT_FOUND, "BGP not initialized"); break; } struct NODEINFO_Q* nq = route_bgp_get_node(bgp, peer_id); if (!nq) { send_error(client, MSG_ERR_PEER_NOT_FOUND, "peer not found"); DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "GET_PEER_INFO: peer=%016llx not found", (unsigned long long)peer_id); break; } uint8_t buf[SC_PUBKEY_SIZE * 2]; memcpy(buf, nq->node.public_key, SC_PUBKEY_SIZE); memcpy(buf + SC_PUBKEY_SIZE, nq->node.ed25519_public_key, SC_PUBKEY_SIZE); send_response(client, MSG_RSP_PEER_INFO, peer_id, buf, SC_PUBKEY_SIZE * 2); DEBUG_INFO(DEBUG_CATEGORY_MSGTRANSPORT, "GET_PEER_INFO: peer=%016llx sent pubkey+edpub", (unsigned long long)peer_id); break; } case MSG_DATA_SEND: { uint64_t dst = hdr->id ? hdr->id : client->bound_peer_id; if (dst == 0) { send_error(client, MSG_ERR_NOT_BOUND, "Not bound to any peer"); break; } if (payload_size == 0) break; int is_signed = (hdr->options & MSG_OPT_SIGNED) ? 1 : 0; if (forward_to_peer(t, dst, payload, payload_size, is_signed) != 0) { send_error(client, MSG_ERR_SEND_FAILED, "Failed to forward message to peer"); t->counters.drops_no_route++; } break; } default: DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "unknown message type 0x%02x from client", hdr->type); send_error(client, MSG_ERR_INVALID_MSG, "Unknown message type"); break; } uint16_t msg_size = hdr->size; if (client->recv_len > msg_size) memmove(client->recv_buffer, client->recv_buffer + msg_size, client->recv_len - msg_size); client->recv_len -= msg_size; } } /* ============================================================================ * Отправка пиру через etcp_router * ============================================================================ */ static int forward_to_peer(struct msg_transport* t, uint64_t dst_peer_id, const uint8_t* data, size_t data_len, int is_signed) { struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(t->instance, dst_peer_id, t->svc_id); if (!rconn) { DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "forward_to_peer: no route to %016llx", (unsigned long long)dst_peer_id); return -1; } int ret; if (is_signed) { ret = etcp_router_conn_send_signed(rconn, data, data_len); } else { ret = etcp_router_conn_send(rconn, data, data_len); } if (ret != 0) { DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "forward_to_peer: send failed to %016llx (signed=%d)", (unsigned long long)dst_peer_id, is_signed); return -1; } t->counters.msg_sent_to_peer++; return 0; } /* ============================================================================ * Приём от пира через etcp_router → доставка TCP-клиенту * ============================================================================ */ static void on_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { if (!conn || !entry) return; struct UTUN_INSTANCE* inst = conn->instance; if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return; } struct msg_transport* t = inst->msg_t; if (!t) { DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "on_etcp_recv: msg_transport not initialized on instance"); queue_dgram_free(entry); queue_entry_free(entry); return; } uint8_t* payload = entry->dgram + 1; size_t payload_len = entry->len - 1; uint64_t src_node_id = conn->peer_node_id; DEBUG_TRACE(DEBUG_CATEGORY_MSGTRANSPORT, "on_etcp_recv: from %016llx len=%zu", (unsigned long long)src_node_id, payload_len); /* Ищем клиента привязанного к этому пиру */ struct msg_client* client = t->clients; while (client) { if (client->bound_peer_id == src_node_id) break; client = client->next; } if (!client) { DEBUG_WARN(DEBUG_CATEGORY_MSGTRANSPORT, "on_etcp_recv: no client bound to %016llx, dropping", (unsigned long long)src_node_id); t->counters.drops_no_client++; queue_dgram_free(entry); queue_entry_free(entry); return; } t->counters.msg_recv_from_peer++; /* Формируем MSG_DATA_RECV */ size_t total = MSG_HDR_SIZE + payload_len; uint8_t* buf = u_malloc(total); if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_MSGTRANSPORT, "on_etcp_recv: u_malloc failed"); queue_dgram_free(entry); queue_entry_free(entry); return; } struct msg_header* hdr = (struct msg_header*)buf; msg_build_header(hdr, (uint16_t)payload_len, MSG_DATA_RECV, src_node_id, 0); memcpy(buf + MSG_HDR_SIZE, payload, payload_len); #ifdef _WIN32 send(client->fd, (const char*)buf, (int)total, 0); #else send(client->fd, buf, total, 0); #endif u_free(buf); t->counters.msg_sent_to_client++; queue_dgram_free(entry); queue_entry_free(entry); } /* ============================================================================ * Помощники отправки ответов TCP-клиенту * ============================================================================ */ static void send_response(struct msg_client* client, uint8_t type, uint64_t id, const uint8_t* data, size_t data_len) { size_t total = MSG_HDR_SIZE + data_len; uint8_t* buf = u_malloc(total); if (!buf) return; struct msg_header* hdr = (struct msg_header*)buf; msg_build_header(hdr, (uint16_t)data_len, type, id, 0); if (data_len > 0) memcpy(buf + MSG_HDR_SIZE, data, data_len); #ifdef _WIN32 send(client->fd, (const char*)buf, (int)total, 0); #else send(client->fd, buf, total, 0); #endif u_free(buf); } static void send_error(struct msg_client* client, uint8_t error_code, const char* msg) { size_t msg_len = strlen(msg); size_t total = MSG_HDR_SIZE + 1 + msg_len; uint8_t* buf = u_malloc(total); if (!buf) return; struct msg_header* hdr = (struct msg_header*)buf; msg_build_header(hdr, (uint16_t)(1 + msg_len), MSG_RSP_ERROR, 0, 0); buf[MSG_HDR_SIZE] = error_code; memcpy(buf + MSG_HDR_SIZE + 1, msg, msg_len); #ifdef _WIN32 send(client->fd, (const char*)buf, (int)total, 0); #else send(client->fd, buf, total, 0); #endif u_free(buf); }