wip: STCP TCP chat integration — diagnostics, invite TCP fallback, test infra

This commit is contained in:
evgeny committed 2026-08-06 18:41:41 +03:00
1 parent e2d8f68c9a
commit dfe9d36783
41 files changed
+2432 -294

No files matched your search

+6 -2
View File
@@ -76,7 +76,9 @@ utun_CORE_SOURCES = \
chat/chat_status.c \
chat/chat_sync.c \
chat/member_sync.c \
chat/merkle_sync.c
chat/merkle_sync.c \
chat/invite_link.c \
chat/chat_headless_control.c
# libutun: all core sources except main()
libutun_a_SOURCES = \
@@ -152,7 +154,9 @@ libutun_a_SOURCES = \
chat/chat_status.c \
chat/chat_sync.c \
chat/member_sync.c \
chat/merkle_sync.c
chat/merkle_sync.c \
chat/invite_link.c \
chat/chat_headless_control.c
libutun_a_CFLAGS = $(utun_CFLAGS)
# Platform-specific TUN libs (Windows only)
+194
View File
@@ -264,3 +264,197 @@ void chat_core_set_setting_trampoline(void* arg) {
chat_core_set_setting(name, value);
u_free(arg);
}
/* ─── headless API ─── */
void chat_core_set_ctx_for_headless(sqlite3* db, struct UTUN_INSTANCE* inst, uint64_t my_node_id) {
if (!g_cc.initialized) g_cc.initialized = 1;
g_cc.db = db; g_cc.inst = inst; g_cc.my_node_id = my_node_id;
g_cc.shared_db = 1;
}
static void json_escape(const char* src, char* dst, size_t dst_sz) {
size_t i = 0; const char* s = src;
while (*s && i < dst_sz - 2) {
if (*s == '"' || *s == '\\') { if (i + 1 < dst_sz - 2) dst[i++] = '\\'; }
dst[i++] = *s++;
}
dst[i] = '\0';
}
int chat_core_get_channels_json(char* buf, size_t buf_size, size_t* out_len) {
if (!g_cc.initialized || !buf || !out_len) return -1;
char* w = buf; char* end = buf + buf_size;
*w++ = '['; const char* sep = "";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db,
"SELECT c.channel_id, c.name, c.owner_node_id, "
"(SELECT COUNT(*) FROM nodes WHERE node_id=c.owner_node_id AND name IS NOT NULL) AS has_owner_name, "
"(SELECT name FROM nodes WHERE node_id=c.owner_node_id) AS owner_name "
"FROM channels c ORDER BY c.created_at ASC",
-1, &st, NULL) != SQLITE_OK) { *out_len = 0; return -1; }
while (sqlite3_step(st) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(st, 0);
const char* ch_name = (const char*)sqlite3_column_text(st, 1);
uint64_t owner_id = (uint64_t)sqlite3_column_int64(st, 2);
const char* owner_name = (const char*)sqlite3_column_text(st, 4);
if (!ch_id) continue;
uint32_t msg_count = chat_core_count(ch_id);
char peers_tbl[80]; peers_table_name(ch_id, peers_tbl, sizeof(peers_tbl));
int peer_count = 0, online_count = 0;
sqlite3_stmt* ps = NULL;
char psql[200]; snprintf(psql, sizeof(psql),
"SELECT COUNT(*), SUM(CASE WHEN n.online=1 THEN 1 ELSE 0 END) FROM \"%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", peers_tbl);
if (sqlite3_prepare_v2(g_cc.db, psql, -1, &ps, NULL) == SQLITE_OK) {
if (sqlite3_step(ps) == SQLITE_ROW) { peer_count = sqlite3_column_int(ps, 0); online_count = sqlite3_column_int(ps, 1); }
sqlite3_finalize(ps);
}
char esc_name[256], esc_owner[256];
json_escape(ch_name ? ch_name : "", esc_name, sizeof(esc_name));
json_escape(owner_name ? owner_name : "", esc_owner, sizeof(esc_owner));
w += snprintf(w, (size_t)(end - w),
"%s{\"id\":\"%s\",\"name\":\"%s\",\"owner_id\":\"0x%016llx\",\"owner_name\":\"%s\","
"\"peers\":%d,\"msgs\":%u,\"online\":%d}",
sep, ch_id, esc_name, (unsigned long long)owner_id, esc_owner, peer_count, msg_count, online_count);
sep = ",";
}
sqlite3_finalize(st);
*w++ = ']';
*w = '\0';
*out_len = (size_t)(w - buf);
return 0;
}
int chat_core_get_messages_json(const char* ch_id, int count, int offset,
char* buf, size_t buf_size, size_t* out_len) {
if (!g_cc.initialized || !ch_id || !buf || !out_len) return -1;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char* w = buf; char* end = buf + buf_size;
*w++ = '['; const char* sep = "";
char sql[256];
snprintf(sql, sizeof(sql),
"SELECT m.id, m.timestamp, m.node_id, CAST(m.data AS TEXT), n.name "
"FROM \"%s\" m LEFT JOIN nodes n ON m.node_id=n.node_id "
"ORDER BY m.timestamp DESC LIMIT ? OFFSET ?", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) != SQLITE_OK) { *out_len = 0; return -1; }
sqlite3_bind_int(st, 1, count > 0 ? count : 100);
sqlite3_bind_int(st, 2, offset);
while (sqlite3_step(st) == SQLITE_ROW) {
int64_t id = sqlite3_column_int64(st, 0);
int64_t ts = sqlite3_column_int64(st, 1);
uint64_t author_id = (uint64_t)sqlite3_column_int64(st, 2);
const char* data = (const char*)sqlite3_column_text(st, 3);
const char* author_name = (const char*)sqlite3_column_text(st, 4);
if (!data) data = "";
const char* ct = "text";
const char* ctpos = strstr(data, "\"ct\":\"");
if (ctpos) { ctpos += 6; char ctbuf[32] = {0}; int ci = 0; while (*ctpos && *ctpos != '"' && ci < 31) ctbuf[ci++] = *ctpos++; ct = ctbuf; }
char esc_data[4096], esc_aname[256];
json_escape(data, esc_data, sizeof(esc_data));
json_escape(author_name ? author_name : "", esc_aname, sizeof(esc_aname));
w += snprintf(w, (size_t)(end - w),
"%s{\"id\":%lld,\"ts\":%lld,\"author_id\":\"0x%016llx\",\"author_name\":\"%s\","
"\"content_type\":\"%s\",\"data\":\"%.2000s\"}",
sep, (long long)id, (long long)ts, (unsigned long long)author_id, esc_aname, ct, esc_data);
sep = ",";
}
sqlite3_finalize(st);
*w++ = ']';
*w = '\0';
*out_len = (size_t)(w - buf);
return 0;
}
int chat_core_get_members_json(const char* ch_id, char* buf, size_t buf_size, size_t* out_len) {
if (!g_cc.initialized || !ch_id || !buf || !out_len) return -1;
char peers_tbl[80]; peers_table_name(ch_id, peers_tbl, sizeof(peers_tbl));
char* w = buf; char* end = buf + buf_size;
*w++ = '['; const char* sep = "";
char sql[256];
snprintf(sql, sizeof(sql),
"SELECT p.node_id, n.name, n.online, n.x25519_pubkey, n.ed25519_pubkey "
"FROM \"%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", peers_tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) != SQLITE_OK) { *out_len = 0; return -1; }
while (sqlite3_step(st) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
const char* name = (const char*)sqlite3_column_text(st, 1);
int online = sqlite3_column_int(st, 2);
const uint8_t* x25519 = (const uint8_t*)sqlite3_column_blob(st, 3);
const uint8_t* ed25519 = sqlite3_column_blob(st, 4);
int x25519_len = sqlite3_column_bytes(st, 3);
int ed25519_len = sqlite3_column_bytes(st, 4);
char esc_name[256], x25519_hex[65], ed25519_hex[65];
json_escape(name ? name : "", esc_name, sizeof(esc_name));
x25519_hex[0] = ed25519_hex[0] = '\0';
if (x25519 && x25519_len >= 32) for (int i = 0; i < 32; i++) snprintf(x25519_hex + i*2, 3, "%02x", x25519[i]);
if (ed25519 && ed25519_len >= 32) for (int i = 0; i < 32; i++) snprintf(ed25519_hex + i*2, 3, "%02x", ed25519[i]);
/* check connected */
int connected = 0;
if (g_cc.inst && g_cc.inst->connections) {
struct ll_entry* e = queue_find_data_by_index(g_cc.inst->connections, (const uint8_t*)&nid);
if (e) connected = 1;
}
w += snprintf(w, (size_t)(end - w),
"%s{\"node_id\":\"0x%016llx\",\"name\":\"%s\",\"online\":%d,\"connected\":%d,"
"\"x25519\":\"%s\",\"ed25519\":\"%s\",\"addrs\":[",
sep, (unsigned long long)nid, esc_name, online, connected, x25519_hex, ed25519_hex);
sep = ",";
/* addresses */
sqlite3_stmt* as = NULL;
char asql[200]; snprintf(asql, sizeof(asql),
"SELECT family, protocol, address, port, rtt FROM node_addresses WHERE node_id=? ORDER BY family, protocol");
if (sqlite3_prepare_v2(g_cc.db, asql, -1, &as, NULL) == SQLITE_OK) {
sqlite3_bind_int64(as, 1, (sqlite3_int64)nid);
const char* asep2 = "";
while (sqlite3_step(as) == SQLITE_ROW) {
int fam = sqlite3_column_int(as, 0);
int proto = sqlite3_column_int(as, 1);
const uint8_t* addr = (const uint8_t*)sqlite3_column_blob(as, 2);
int addr_len = sqlite3_column_bytes(as, 2);
int port = sqlite3_column_int(as, 3);
int rtt = sqlite3_column_int(as, 4);
char ip_str[64]; ip_str[0] = '\0';
if (fam == 4 && addr && addr_len >= 4)
snprintf(ip_str, sizeof(ip_str), "%d.%d.%d.%d", addr[0], addr[1], addr[2], addr[3]);
else if (fam == 6 && addr && addr_len >= 16)
snprintf(ip_str, sizeof(ip_str), "%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x",
addr[0],addr[1],addr[2],addr[3],addr[4],addr[5],addr[6],addr[7],
addr[8],addr[9],addr[10],addr[11],addr[12],addr[13],addr[14],addr[15]);
else { snprintf(ip_str, sizeof(ip_str), "?"); }
w += snprintf(w, (size_t)(end - w),
"%s{\"ip\":\"%s\",\"port\":%d,\"proto\":\"%s\",\"rtt\":%d}",
asep2, ip_str, port, proto == 1 ? "UDP" : proto == 2 ? "TCP" : "?", rtt);
asep2 = ",";
}
sqlite3_finalize(as);
}
w += snprintf(w, (size_t)(end - w), "]}");
}
sqlite3_finalize(st);
*w++ = ']';
*w = '\0';
*out_len = (size_t)(w - buf);
return 0;
}
int chat_core_get_node_name(uint64_t node_id, char* out, size_t sz) {
if (!g_cc.initialized || !out) return -1;
out[0] = '\0';
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, "SELECT name FROM nodes WHERE node_id=?", -1, &st, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
const char* n = (const char*)sqlite3_column_text(st, 0);
if (n) snprintf(out, sz, "%s", n);
}
sqlite3_finalize(st);
return 0;
}
+21
View File
@@ -10,6 +10,7 @@
#include <stdint.h>
#include <stddef.h>
#include <sqlite3.h>
struct sqlite3;
struct UTUN_INSTANCE;
@@ -102,4 +103,24 @@ int chat_core_update_local_attrs(const char* ch_id, uint64_t ts,
/* Сбор статуса (NTP + connections) и отправка в GUI */
void chat_core_collect_status_trampoline(void* arg);
/* ── Headless CLI API ── */
/** Вариант для сборки без UTUN_INSTANCE (использует sqlite3* и node_id) */
void chat_core_set_ctx_for_headless(sqlite3* db, struct UTUN_INSTANCE* inst, uint64_t my_node_id);
/** Список каналов с метаданными в JSON: [{id,name,owner_id,owner_name,peers,msgs,online}] */
int chat_core_get_channels_json(char* buf, size_t buf_size, size_t* out_len);
/** Сообщения канала в JSON: [{id,ts,author_id,author_name,content_type,data}]
* count=0 → все, offset=0 → сначала */
int chat_core_get_messages_json(const char* ch_id, int count, int offset,
char* buf, size_t buf_size, size_t* out_len);
/** Мемберы канала с полным состоянием в JSON:
* [{node_id,name,online,connected,x25519,ed25519,addrs:[{ip,port,proto,rtt}]}] */
int chat_core_get_members_json(const char* ch_id, char* buf, size_t buf_size, size_t* out_len);
/** Имя узла из таблицы nodes */
int chat_core_get_node_name(uint64_t node_id, char* out, size_t sz);
#endif /* CHAT_CORE_H */
+1
View File
@@ -37,6 +37,7 @@ extern "C" {
#define CHAT_EVT_NODEINFO_UPDATED 16 /* [node_id:8][conn_presence:1][conn_up:1][best_rtt:2] */
#define CHAT_EVT_LINK_STATUS_CHANGED 17 /* [peer_id:8][link_id:1][state:1][status:1][links_up:1][initialized:1] */
#define CHAT_EVT_INVITE_RECEIVED 18 /* [ch_id_len:1][ch_id][name_len:1][name][inviter_node_id:8][inviter_name_len:1][inviter_name] */
#define CHAT_EVT_LOCAL_SOCKETS 19 /* JSON: [{"si":id,"nm":"...","ad":"...","po":port,"pt":"TCP|UDP","st":0|1|2,"lk":n},...] */
typedef void (*chat_event_handler_fn)(int type, const uint8_t* data, int len);
+536
View File
@@ -0,0 +1,536 @@
/*
* chat_headless_control.c — TCP control socket for headless chat CLI
*
* JSON line protocol over TCP. Integrated into uasync event loop.
* Default: localhost-only, port from config.
*/
#include "chat_headless_control.h"
#include "invite_link.h"
#include "chat_core.h"
#include "chat_sync.h"
#include "chat_event.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/etcp_connections.h"
#include "../../lib/u_async.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../routing_layer/topo_node_sqlite.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#define MAX_CLIENTS 16
#define RECV_BUF_SIZE 16384
#define SEND_BUF_SIZE 16384
#ifndef DEBUG_CATEGORY_HEADLESS
#define DEBUG_CATEGORY_HEADLESS 27
#endif
struct headless_client {
socket_t fd;
void* socket_id;
char recv_buf[RECV_BUF_SIZE];
int recv_len;
char send_buf[SEND_BUF_SIZE];
int send_len;
int send_offset;
int subscribed;
int closing;
struct headless_client* next;
};
static struct {
struct UASYNC* ua;
struct UTUN_INSTANCE* inst;
struct headless_client* clients;
int client_count;
socket_t listen_fd;
void* listen_sock_id;
int running;
} g_hc;
/* ── Forward declarations ── */
static void hc_handle_command(struct headless_client* cli, const char* json);
/* ── JSON helpers ── */
static int json_get_int(const char* s, const char* key, int def) {
char pat[128]; snprintf(pat, sizeof(pat), "\"%s\"", key);
const char* p = strstr(s, pat); if (!p) return def;
p += strlen(pat);
while (*p == ' ' || *p == '\t' || *p == ':') p++;
return (int)strtol(p, NULL, 10);
}
static int json_get_str(const char* s, const char* key, char* out, size_t out_sz) {
char pat[128]; snprintf(pat, sizeof(pat), "\"%s\"", key);
const char* p = strstr(s, pat); if (!p) return -1;
p += strlen(pat);
while (*p == ' ' || *p == '\t' || *p == ':') p++;
if (*p == '"') p++;
size_t i = 0;
while (*p && *p != '"' && i < out_sz - 1) {
if (*p == '\\' && *(p + 1)) { p++; out[i++] = *(p++); }
else out[i++] = *(p++);
}
out[i] = '\0'; return 0;
}
static const char* json_get_cmd(const char* s) {
const char* p = strstr(s, "\"cmd\"");
if (!p) return NULL;
p += 5;
while (*p == ' ' || *p == '\t' || *p == ':') p++;
if (*p == '"') p++;
return p;
}
static const char* json_get_cmd_end(const char* cmd_start) {
const char* p = cmd_start;
while (*p && *p != '"') { if (*p == '\\' && *(p + 1)) p++; p++; }
return p;
}
/* ── Send helpers ── */
static void cli_send(struct headless_client* cli, const char* data, size_t len) {
if (!cli || cli->closing || cli->fd < 0 || !data || len == 0) return;
if (cli->send_len > 0) {
size_t space = sizeof(cli->send_buf) - (size_t)cli->send_len;
if (len + 1 > space) return;
memcpy(cli->send_buf + cli->send_len, data, len);
cli->send_len += (int)len;
return;
}
ssize_t r = write(cli->fd, data, len);
if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
if (len <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data, len); cli->send_len = (int)len; cli->send_offset = 0; }
if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 1);
return;
}
if (r >= 0 && (size_t)r < len) {
size_t remain = len - (size_t)r;
if (remain <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data + r, remain); cli->send_len = (int)remain; cli->send_offset = 0; }
if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 1);
}
}
static void cli_send_str(struct headless_client* cli, const char* s) {
if (s) cli_send(cli, s, strlen(s));
}
static void send_response(struct headless_client* cli, int id, const char* ok_data, const char* error) {
char buf[4096]; int len;
if (error) len = snprintf(buf, sizeof(buf), "{\"id\":%d,\"ok\":false,\"error\":\"%s\"}\n", id, error);
else if (ok_data) len = snprintf(buf, sizeof(buf), "{\"id\":%d,\"ok\":true,\"data\":%s}\n", id, ok_data);
else len = snprintf(buf, sizeof(buf), "{\"id\":%d,\"ok\":true}\n", id);
if (len > 0) cli_send(cli, buf, (size_t)len);
}
/* ── Broadcast event ── */
static void hc_broadcast_event(const char* json) {
size_t jlen = strlen(json);
char buf[4096]; int blen = snprintf(buf, sizeof(buf), "%s\n", json);
if (blen <= 0) return;
for (struct headless_client* c = g_hc.clients; c; c = c->next)
if (c->subscribed) cli_send(c, buf, (size_t)blen);
}
/* ── chat_event callback ── */
static void hc_on_chat_event(int type, const uint8_t* data, int len) {
if (type == 10) return; /* skip status text */
char json[1024]; json[0] = '\0';
if (type == 1 && data && len >= 10) { /* MSG_RECEIVED */
uint8_t cl = data[0];
uint64_t author; memcpy(&author, data + 1 + cl, 8);
snprintf(json, sizeof(json), "{\"event\":\"msg\",\"ch\":\"%.*s\",\"author_id\":\"0x%016llx\"}", (int)cl, data + 1, (unsigned long long)author);
} else if (type == 5 && data && len >= 2) { /* MEMBERS_CHANGED */
uint8_t cl = data[0];
snprintf(json, sizeof(json), "{\"event\":\"members_changed\",\"ch\":\"%.*s\"}", (int)cl, data + 1);
} else if (type == 4 && data && len >= 2) { /* CHANNEL_UPDATED */
uint8_t cl = data[0];
snprintf(json, sizeof(json), "{\"event\":\"channel_updated\",\"ch\":\"%.*s\"}", (int)cl, data + 1);
} else if (type == 18 && data && len >= 2) { /* INVITE_RECEIVED */
uint8_t cl = data[0];
if (len >= 2 + cl) {
uint8_t nl = data[1 + cl];
uint64_t inviter; int off = 2 + cl + nl;
if (len >= off + 8) { memcpy(&inviter, data + off, 8);
snprintf(json, sizeof(json), "{\"event\":\"invite_received\",\"ch\":\"%.*s\",\"ch_name\":\"%.*s\",\"from_id\":\"0x%016llx\"}",
(int)cl, data + 1, (int)nl, data + 2 + cl, (unsigned long long)inviter);
}
}
}
if (json[0]) hc_broadcast_event(json);
}
/* ── Write callback ── */
static void client_write_callback(socket_t fd, void* arg) {
struct headless_client* cli = (struct headless_client*)arg;
if (!cli || cli->closing || fd != cli->fd) return;
if (cli->send_len <= 0) { if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; }
ssize_t r = write(cli->fd, cli->send_buf + cli->send_offset, (size_t)(cli->send_len - cli->send_offset));
if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) return;
if (r <= 0) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; }
cli->send_offset += (int)r;
if (cli->send_offset >= cli->send_len) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); }
}
/* ── Close client ── */
static void hc_close_client(struct headless_client* cli) {
if (!cli || cli->closing) return;
cli->closing = 1;
if (g_hc.ua && cli->socket_id) uasync_remove_socket_t(g_hc.ua, cli->fd);
if (cli->fd >= 0) { close(cli->fd); cli->fd = -1; }
}
/* ── Read callback ── */
static void client_read_callback(socket_t fd, void* arg) {
struct headless_client* cli = (struct headless_client*)arg;
if (!cli || cli->closing) return;
char tmp[4096];
ssize_t r = read(fd, tmp, sizeof(tmp) - 1);
if (r <= 0) { hc_close_client(cli); return; }
for (ssize_t i = 0; i < r; i++) {
if (tmp[i] == '\n') {
if (cli->recv_len > 0) { cli->recv_buf[cli->recv_len] = '\0'; hc_handle_command(cli, cli->recv_buf); cli->recv_len = 0; }
} else if (cli->recv_len < RECV_BUF_SIZE - 1) {
cli->recv_buf[cli->recv_len++] = tmp[i];
}
}
}
/* ── Command handlers ── */
static void hc_handle_ping(struct headless_client* cli, int id, const char* json) {
(void)json;
uint64_t uptime = 0;
(void)uptime;
char data[128]; snprintf(data, sizeof(data), "{\"pong\":true,\"uptime_sec\":%llu}", (unsigned long long)uptime);
send_response(cli, id, data, NULL);
}
static void hc_handle_status(struct headless_client* cli, int id, const char* json) {
(void)json;
char buf[8192]; int off = 0;
off += snprintf(buf + off, sizeof(buf) - off, "\"");
if (!g_hc.inst) {
off += snprintf(buf + off, sizeof(buf) - off, "ERROR: no instance");
} else {
int conn_count = g_hc.inst->connections ? queue_entry_count(g_hc.inst->connections) : 0;
struct NTP_TIME* ntp = &g_hc.inst->ntp;
off += snprintf(buf + off, sizeof(buf) - off, "NTP: enabled=%s synced=%s offset=%.1fs\\n",
ntp->enabled ? "yes" : "no", ntp->synced ? "yes" : "no", ntp->offset_us / 1000000.0);
off += snprintf(buf + off, sizeof(buf) - off, "Connections: %d active\\n", conn_count);
uint64_t my_id = g_hc.inst->node_id;
struct ll_entry* entry = g_hc.inst->connections ? g_hc.inst->connections->head : NULL;
while (entry) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
if (ce) {
char name[64]; chat_core_get_node_name(ce->peer_node_id, name, sizeof(name));
off += snprintf(buf + off, sizeof(buf) - off, " [%04llX]->[%04llX] %s ETCP:%s\\n",
(unsigned long long)(my_id & 0xFFFF), (unsigned long long)(ce->peer_node_id & 0xFFFF),
name[0] ? name : "?", ce->conn && ce->conn->links_up ? "UP" : "DOWN");
}
entry = entry->next;
}
}
if (off >= (int)sizeof(buf) - 2) off = (int)sizeof(buf) - 2;
off += snprintf(buf + off, sizeof(buf) - off, "\"");
send_response(cli, id, buf, NULL);
}
static void hc_handle_channels(struct headless_client* cli, int id, const char* json) {
(void)json;
char buf[16384]; size_t len = 0;
if (chat_core_get_channels_json(buf, sizeof(buf), &len) < 0)
send_response(cli, id, NULL, "failed to query channels");
else
send_response(cli, id, buf, NULL);
}
static void hc_handle_members(struct headless_client* cli, int id, const char* json) {
char ch[64];
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
char buf[16384]; size_t len = 0;
if (chat_core_get_members_json(ch, buf, sizeof(buf), &len) < 0)
send_response(cli, id, NULL, "failed to query members");
else
send_response(cli, id, buf, NULL);
}
static void hc_handle_messages(struct headless_client* cli, int id, const char* json) {
char ch[64]; int count = 20, offset = 0;
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
count = json_get_int(json, "count", 20);
offset = json_get_int(json, "offset", 0);
char buf[16384]; size_t len = 0;
if (chat_core_get_messages_json(ch, count, offset, buf, sizeof(buf), &len) < 0)
send_response(cli, id, NULL, "failed to query messages");
else
send_response(cli, id, buf, NULL);
}
static void hc_handle_send(struct headless_client* cli, int id, const char* json) {
char ch[64], ct[32], data[4096];
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
if (json_get_str(json, "content_type", ct, sizeof(ct)) < 0) { ct[0] = 't'; ct[1] = 'e'; ct[2] = 'x'; ct[3] = 't'; ct[4] = '\0'; }
if (json_get_str(json, "data", data, sizeof(data)) < 0) { send_response(cli, id, NULL, "missing 'data' param"); return; }
struct chat_msg_submit* req = u_calloc(1, sizeof(*req));
if (!req) { send_response(cli, id, NULL, "out of memory"); return; }
snprintf(req->channel_id, sizeof(req->channel_id), "%s", ch);
snprintf(req->content_type, sizeof(req->content_type), "%s", ct);
req->data = (uint8_t*)u_strdup(data);
req->data_len = (uint32_t)strlen(data);
chat_core_submit_message(req);
if (req->data) u_free(req->data);
u_free(req);
send_response(cli, id, "{\"sent\":true}", NULL);
}
static void hc_handle_invite(struct headless_client* cli, int id, const char* json) {
char ch[64];
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
if (!g_hc.inst) { send_response(cli, id, NULL, "no instance"); return; }
uint64_t my_node_id = g_hc.inst->node_id;
uint8_t my_pubkey[32]; memcpy(my_pubkey, g_hc.inst->my_keys.public_key, 32);
sqlite3* db = chat_core_get_db();
if (!db) { send_response(cli, id, NULL, "no database"); return; }
struct InviteData inv; memset(&inv, 0, sizeof(inv));
inv.channelId = strtoull(ch, NULL, 10);
if (inv.channelId == 0) { send_response(cli, id, NULL, "invalid channel_id"); return; }
memcpy(inv.pubkey, my_pubkey, 32);
inv.nodeId = my_node_id;
sqlite3_stmt* as = NULL;
sqlite3_prepare_v2(db, "SELECT family, protocol, address, port, rtt, socket_id FROM node_addresses WHERE node_id=? AND protocol IN (1,2) ORDER BY family, protocol",
-1, &as, NULL);
if (as) {
sqlite3_bind_int64(as, 1, (sqlite3_int64)my_node_id);
while (sqlite3_step(as) == SQLITE_ROW && inv.addrCount < INVITE_ADDR_MAX) {
struct InviteAddr* a = &inv.addrs[inv.addrCount++];
int fam = sqlite3_column_int(as, 0);
a->proto = (uint8_t)sqlite3_column_int(as, 1);
const uint8_t* addr = (const uint8_t*)sqlite3_column_blob(as, 2);
int addr_len = sqlite3_column_bytes(as, 2);
a->port = (uint16_t)sqlite3_column_int(as, 3);
a->socketId = (uint8_t)sqlite3_column_int(as, 5);
a->family = fam;
memset(a->address, 0, sizeof(a->address));
if (addr && addr_len > 0) memcpy(a->address, addr, addr_len < 16 ? (size_t)addr_len : 16);
}
sqlite3_finalize(as);
}
if (inv.addrCount == 0) {
/* fallback: collect from sockets directly */
struct ETCP_SOCKET* s = g_hc.inst->etcp_sockets;
while (s && inv.addrCount < INVITE_ADDR_MAX) {
if (s->local_addr.ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)&s->local_addr;
struct InviteAddr* a = &inv.addrs[inv.addrCount++];
a->family = 4; a->proto = 1; a->socketId = s->sock_id;
a->port = ntohs(sin->sin_port); memcpy(a->address, &sin->sin_addr, 4);
}
s = s->next;
}
struct TCP_SOCKET* ts = g_hc.inst->tcp_sockets;
while (ts && inv.addrCount < INVITE_ADDR_MAX) {
if (ts->interface_addr.ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)&ts->interface_addr;
struct InviteAddr* a = &inv.addrs[inv.addrCount++];
a->family = 4; a->proto = 2; a->socketId = ts->sock_id;
a->port = ntohs(sin->sin_port); memcpy(a->address, &sin->sin_addr, 4);
}
ts = ts->next;
}
}
if (inv.addrCount == 0) { send_response(cli, id, NULL, "no addresses available to build invite link"); return; }
char link[1024];
int r = invite_link_encode(&inv, NULL, link, sizeof(link));
if (r < 0) { send_response(cli, id, NULL, "failed to encode invite link"); return; }
char resp[1280]; snprintf(resp, sizeof(resp), "{\"link\":\"%s\",\"channel_id\":%llu,\"addrs\":%d}", link,
(unsigned long long)inv.channelId, (int)inv.addrCount);
send_response(cli, id, resp, NULL);
}
static void hc_handle_connect(struct headless_client* cli, int id, const char* json) {
char link[1024];
if (json_get_str(json, "link", link, sizeof(link)) < 0) { send_response(cli, id, NULL, "missing 'link' param"); return; }
struct InviteData d; char err[256];
if (invite_link_decode(link, strlen(link), &d, err, sizeof(err)) < 0) { send_response(cli, id, NULL, err); return; }
if (!g_hc.inst) { send_response(cli, id, NULL, "no instance"); return; }
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: connect ch=%llu node=0x%016llx addrs=%d",
(unsigned long long)d.channelId, (unsigned long long)d.nodeId, (int)d.addrCount);
uint8_t addrs_buf[2048];
int addrs_len = invite_serialize_addrs(&d, addrs_buf, sizeof(addrs_buf));
if (addrs_len <= 0) { send_response(cli, id, NULL, "serialize addrs failed"); return; }
chat_core_ensure_channel_ready((const char*)(&(char[64]){0}));
{ char chstr[64]; snprintf(chstr, sizeof(chstr), "%llu", (unsigned long long)d.channelId);
chat_core_ensure_channel_ready(chstr); }
chat_core_set_my_node_id(g_hc.inst->node_id);
chat_sync_connect_from_invite(d.channelId, d.nodeId, d.pubkey,
addrs_buf, d.addrCount, addrs_len, d.password_len ? d.password : NULL);
char resp[512]; snprintf(resp, sizeof(resp),
"{\"channel_id\":%llu,\"node_id\":\"0x%016llx\",\"addrs\":%d,\"connecting\":true}",
(unsigned long long)d.channelId, (unsigned long long)d.nodeId, (int)d.addrCount);
send_response(cli, id, resp, NULL);
}
static void hc_handle_create_channel(struct headless_client* cli, int id, const char* json) {
char name[256];
if (json_get_str(json, "name", name, sizeof(name)) < 0) { send_response(cli, id, NULL, "missing 'name' param"); return; }
chat_core_create_channel_auto(name);
send_response(cli, id, "{\"created\":true}", NULL);
}
static void hc_handle_subscribe(struct headless_client* cli, int id, const char* json) {
int enable = json_get_int(json, "enable", 1);
cli->subscribed = enable;
char resp[64]; snprintf(resp, sizeof(resp), "{\"subscribed\":%s}", enable ? "true" : "false");
send_response(cli, id, resp, NULL);
}
static void hc_handle_quit(struct headless_client* cli, int id, const char* json) {
(void)json;
send_response(cli, id, "{\"bye\":true}", NULL);
cli_send(cli, "{\"status\":\"closing\"}\n", 20);
hc_close_client(cli);
}
/* ── Command dispatch ── */
typedef void (*hc_cmd_fn)(struct headless_client* cli, int id, const char* json);
static const char* cmd_names[] = { "ping", "status", "channels", "members", "messages", "send", "invite", "connect", "create_channel", "subscribe", "quit", NULL };
static hc_cmd_fn cmd_handlers[] = { hc_handle_ping, hc_handle_status, hc_handle_channels, hc_handle_members, hc_handle_messages, hc_handle_send, hc_handle_invite, hc_handle_connect, hc_handle_create_channel, hc_handle_subscribe, hc_handle_quit };
void hc_handle_command(struct headless_client* cli, const char* json) {
if (!cli || !json) return;
int id = json_get_int(json, "id", 0);
const char* cs = json_get_cmd(json);
if (!cs) { send_response(cli, id, NULL, "missing cmd"); return; }
const char* ce = json_get_cmd_end(cs);
size_t clen = (size_t)(ce - cs);
if (clen == 0 || clen > 31) { send_response(cli, id, NULL, "invalid cmd"); return; }
char cmd[32]; memcpy(cmd, cs, clen); cmd[clen] = '\0';
for (int i = 0; cmd_names[i]; i++) {
if (strcmp(cmd, cmd_names[i]) == 0) { cmd_handlers[i](cli, id, json); return; }
}
send_response(cli, id, NULL, "unknown command");
}
/* ── Accept callback ── */
static void accept_callback(socket_t fd, void* arg) {
(void)arg;
struct sockaddr_storage addr; socklen_t alen = sizeof(addr);
socket_t cfd = accept(fd, (struct sockaddr*)&addr, &alen);
if (cfd < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: accept failed: %s", strerror(errno)); return; }
int flags = fcntl(cfd, F_GETFL, 0); if (flags >= 0) fcntl(cfd, F_SETFL, flags | O_NONBLOCK);
if (g_hc.client_count >= MAX_CLIENTS) { close(cfd); DEBUG_WARN((int)DEBUG_CATEGORY_HEADLESS, "headless: max clients reached"); return; }
struct headless_client* cli = u_calloc(1, sizeof(*cli));
if (!cli) { close(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to allocate client"); return; }
cli->fd = cfd;
cli->socket_id = uasync_add_socket_t(g_hc.ua, cfd, client_read_callback, client_write_callback, NULL, cli);
if (!cli->socket_id) { u_free(cli); close(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to register client fd"); return; }
cli->next = g_hc.clients; g_hc.clients = cli; g_hc.client_count++;
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: client connected fd=%d total=%d", (int)cfd, g_hc.client_count);
}
/* ── Periodic cleanup ── */
static void hc_cleanup(void* arg) {
(void)arg;
struct headless_client** prev = &g_hc.clients;
while (*prev) {
if ((*prev)->closing) {
struct headless_client* dead = *prev; *prev = dead->next;
if (dead->socket_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, dead->fd);
if (dead->fd >= 0) close(dead->fd);
u_free(dead); g_hc.client_count--;
} else prev = &(*prev)->next;
}
if (g_hc.ua) uasync_set_timeout(g_hc.ua, 5000, NULL, hc_cleanup, "hc_cleanup");
}
/* ── Init / Destroy ── */
int chat_headless_control_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
const char* bind_ip, int port) {
if (!ua || !bind_ip || port <= 0 || port > 65535) return -1;
if (g_hc.running) return 0;
memset(&g_hc, 0, sizeof(g_hc));
g_hc.ua = ua; g_hc.inst = inst;
g_hc.listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (g_hc.listen_fd < 0) { DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: socket() failed: %s", strerror(errno)); return -1; }
int reuse = 1; setsockopt(g_hc.listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
int flags = fcntl(g_hc.listen_fd, F_GETFL, 0); if (flags >= 0) fcntl(g_hc.listen_fd, F_SETFL, flags | O_NONBLOCK);
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin));
sin.sin_family = AF_INET; sin.sin_port = htons((uint16_t)port);
if (inet_pton(AF_INET, bind_ip, &sin.sin_addr) != 1) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; }
if (bind(g_hc.listen_fd, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: bind(%s:%d) failed: %s", bind_ip, port, strerror(errno));
close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1;
}
if (listen(g_hc.listen_fd, 5) < 0) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; }
g_hc.listen_sock_id = uasync_add_socket_t(ua, g_hc.listen_fd, accept_callback, NULL, NULL, NULL);
if (!g_hc.listen_sock_id) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; }
chat_event_set_handler(hc_on_chat_event);
uasync_set_timeout(ua, 5000, NULL, hc_cleanup, "hc_cleanup");
g_hc.running = 1;
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: listening on %s:%d", bind_ip, port);
return 0;
}
void chat_headless_control_destroy(void) {
if (!g_hc.running) return;
g_hc.running = 0;
chat_event_set_handler(NULL);
if (g_hc.listen_sock_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, g_hc.listen_fd);
if (g_hc.listen_fd >= 0) { close(g_hc.listen_fd); g_hc.listen_fd = -1; }
struct headless_client* c = g_hc.clients;
while (c) { struct headless_client* n = c->next; if (c->fd >= 0) close(c->fd); u_free(c); c = n; }
g_hc.clients = NULL; g_hc.client_count = 0;
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: destroyed");
}
+25
View File
@@ -0,0 +1,25 @@
/*
* chat_headless_control.h — TCP control socket for headless chat CLI
*
* JSON line protocol:
* Request: {"id":N,"cmd":"<command>",...params}
* Response: {"id":N,"ok":true,"data":{...}} | {"id":N,"ok":false,"error":"..."}
* Event: {"event":"<type>",...}
*
* Commands: ping, status, channels, members, messages, send, invite, connect,
* create_channel, subscribe, quit
*/
#ifndef CHAT_HEADLESS_CONTROL_H
#define CHAT_HEADLESS_CONTROL_H
#include <stdint.h>
#include <stddef.h>
struct UASYNC;
struct UTUN_INSTANCE;
int chat_headless_control_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
const char* bind_ip, int port);
void chat_headless_control_destroy(void);
#endif /* CHAT_HEADLESS_CONTROL_H */
+176
View File
@@ -0,0 +1,176 @@
/*
* invite_link.c — encode/decode invite-ссылок utun://
*
* Перенесено из tools/chatgui-android/libutun_lite/invite_link_c.c
* Формат совместим с десктопной Qt-версией (tools/chatgui/src/invite_link.cpp).
*/
#include "invite_link.h"
#include "../transport_layer/secure_channel.h"
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#define INVITE_PREFIX "utun://"
#define INVITE_PREFIX_LEN 7
static const char base64_table[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
static int base64_char_val(char c) {
if (c >= 'A' && c <= 'Z') return c - 'A';
if (c >= 'a' && c <= 'z') return c - 'a' + 26;
if (c >= '0' && c <= '9') return c - '0' + 52;
if (c == '+') return 62;
if (c == '/') return 63;
return -1;
}
static int base64_decode(const char* in, size_t in_len, uint8_t* out, size_t out_cap) {
size_t opos = 0; int group[4]; int gi = 0;
for (size_t i = 0; i < in_len && in[i] != '='; i++) {
int v = base64_char_val(in[i]);
if (v < 0) return -1;
group[gi++] = v;
if (gi == 4) {
out[opos++] = (uint8_t)((group[0] << 2) | (group[1] >> 4));
if (opos >= out_cap) return -1;
out[opos++] = (uint8_t)(((group[1] & 0xF) << 4) | (group[2] >> 2));
if (opos >= out_cap) return -1;
out[opos++] = (uint8_t)(((group[2] & 0x3) << 6) | group[3]);
if (opos >= out_cap) return -1;
gi = 0;
}
}
if (gi == 2) { out[opos++] = (uint8_t)((group[0] << 2) | (group[1] >> 4)); }
else if (gi == 3) {
out[opos++] = (uint8_t)((group[0] << 2) | (group[1] >> 4));
out[opos++] = (uint8_t)(((group[1] & 0xF) << 4) | (group[2] >> 2));
}
return (int)opos;
}
int invite_link_decode(const char* link, size_t link_len, struct InviteData* out,
char* error_buf, size_t error_buf_size) {
if (!link || !out) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "null argument"); return -1; }
if (link_len < INVITE_PREFIX_LEN || strncmp(link, INVITE_PREFIX, INVITE_PREFIX_LEN) != 0) {
if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "invalid prefix (expected utun://)"); return -1;
}
const char* b64 = link + INVITE_PREFIX_LEN;
size_t b64_len = link_len - INVITE_PREFIX_LEN;
uint8_t raw[4096];
int raw_len = base64_decode(b64, b64_len, raw, sizeof(raw));
if (raw_len < 0) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "base64 decode failed"); return -1; }
if (raw_len < 11) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "blob too short (%d bytes)", raw_len); return -1; }
int off = 0;
uint8_t ver = raw[off++];
if (ver != INVITE_LINK_VERSION && ver != 0x02) {
if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "unsupported version 0x%02x", ver); return -1;
}
out->password[0] = '\0'; out->password_len = 0;
if (ver == 0x02) {
if (off + 1 > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at pass_len"); return -1; }
uint8_t plen = raw[off++];
if (plen > 0) {
if (plen > INVITE_PASS_MAX - 1) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "password too long %d", plen); return -1; }
if (off + plen > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at password"); return -1; }
memcpy(out->password, raw + off, plen); out->password[plen] = '\0'; out->password_len = plen;
off += plen;
}
}
if (off + 8 > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at channel_id"); return -1; }
uint64_t chId = 0;
for (int i = 0; i < 8; i++) chId = (chId << 8) | raw[off++];
out->channelId = chId;
memset(out->pubkey, 0, INVITE_PUBKEY_SIZE);
out->addrCount = 0;
int first_block = 1;
while (off + 1 <= raw_len) {
uint8_t header = raw[off++];
int cnt = (header & 0x03) + 1;
if (cnt < 1 || cnt > 4) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "invalid addr count %d", cnt); return -1; }
if (off + 32 > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at pubkey"); return -1; }
if (first_block) { memcpy(out->pubkey, raw + off, INVITE_PUBKEY_SIZE); out->nodeId = sc_derive_node_id_from_pubkey(out->pubkey); first_block = 0; }
off += 32;
for (int j = 0; j < cnt; j++) {
int is_v6 = (header >> (2 + j)) & 1;
int ip_len = is_v6 ? 16 : 4;
if (off + 2 + ip_len + 2 > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at addr %d", j); return -1; }
if (out->addrCount >= INVITE_ADDR_MAX) break;
struct InviteAddr* a = &out->addrs[out->addrCount++];
a->socketId = raw[off++]; a->proto = raw[off++];
a->family = is_v6 ? 6 : 4;
memcpy(a->address, raw + off, (size_t)ip_len); off += ip_len;
a->port = (uint16_t)((raw[off] << 8) | raw[off + 1]); off += 2;
}
}
return 0;
}
int invite_serialize_addrs(const struct InviteData* data, uint8_t* buf, size_t buf_size) {
if (!data || !buf) return -1;
size_t needed = 0;
for (int i = 0; i < (int)data->addrCount; i++) needed += (data->addrs[i].family == 6) ? 21U : 9U;
if (needed > buf_size) return -1;
size_t pos = 0;
for (int i = 0; i < (int)data->addrCount; i++) {
const struct InviteAddr* a = &data->addrs[i];
int ip_len = (a->family == 6) ? 16 : 4;
buf[pos++] = (uint8_t)a->family; buf[pos++] = a->socketId; buf[pos++] = a->proto;
memcpy(buf + pos, a->address, (size_t)ip_len); pos += ip_len;
buf[pos++] = (uint8_t)(a->port >> 8); buf[pos++] = (uint8_t)(a->port & 0xFF);
}
return (int)pos;
}
int invite_link_encode(const struct InviteData* data, const char* password, char* out, size_t out_size) {
if (!data || !out || data->pubkey[0] == 0) return -1;
if (data->addrCount == 0) return -1;
int has_pass = (password && password[0]) ? 1 : 0;
uint8_t raw[4096]; size_t pos = 0;
raw[pos++] = has_pass ? 0x02 : INVITE_LINK_VERSION;
if (has_pass) {
size_t plen = strlen(password);
if (plen > INVITE_PASS_MAX - 1) return -1;
raw[pos++] = (uint8_t)plen; memcpy(raw + pos, password, plen); pos += plen;
}
for (int i = 7; i >= 0; i--) raw[pos++] = (uint8_t)((data->channelId >> (i * 8)) & 0xFF);
uint8_t pk[32]; memcpy(pk, data->pubkey, 32);
int ia = 0;
while (ia < (int)data->addrCount) {
int rem = (int)data->addrCount - ia;
int cnt = (rem > 4) ? 4 : rem;
uint8_t header = (uint8_t)((cnt - 1) & 0x03);
for (int j = 0; j < cnt; j++) if (data->addrs[ia + j].family == 6) header |= (uint8_t)(1 << (2 + j));
raw[pos++] = header; memcpy(raw + pos, pk, 32); pos += 32;
for (int j = 0; j < cnt; j++) {
const struct InviteAddr* a = &data->addrs[ia + j];
int ip_len = (a->family == 6) ? 16 : 4;
raw[pos++] = a->socketId; raw[pos++] = a->proto;
memcpy(raw + pos, a->address, (size_t)ip_len); pos += (size_t)ip_len;
raw[pos++] = (uint8_t)(a->port >> 8); raw[pos++] = (uint8_t)(a->port & 0xFF);
}
ia += cnt;
}
if (out_size < 12 + (pos * 4 + 2) / 3 + 1) return -1;
memcpy(out, INVITE_PREFIX, INVITE_PREFIX_LEN);
size_t opos = INVITE_PREFIX_LEN;
size_t i = 0;
while (i < pos) {
uint32_t val = (uint32_t)raw[i] << 16;
val |= (i + 1 < pos) ? (uint32_t)raw[i + 1] << 8 : 0;
val |= (i + 2 < pos) ? (uint32_t)raw[i + 2] : 0;
out[opos++] = base64_table[(val >> 18) & 0x3F];
out[opos++] = base64_table[(val >> 12) & 0x3F];
out[opos++] = (i + 1 < pos) ? base64_table[(val >> 6) & 0x3F] : '=';
out[opos++] = (i + 2 < pos) ? base64_table[val & 0x3F] : '=';
i += 3;
}
if (opos >= out_size) return -1;
out[opos] = '\0';
return (int)(opos - INVITE_PREFIX_LEN);
}
+54
View File
@@ -0,0 +1,54 @@
/*
* invite_link.h — invite-ссылки utun:// для каналов
*
* Формат: utun:// + base64(version | password? | channel_id | [header | pubkey | addrs]+)
* Совместим с десктопной (Qt) и Android (Kotlin) версиями.
*/
#ifndef INVITE_LINK_H
#define INVITE_LINK_H
#include <stdint.h>
#include <stddef.h>
#define INVITE_LINK_VERSION 0x01
#define INVITE_PUBKEY_SIZE 32
#define INVITE_ADDR_MAX 32
#define INVITE_PASS_MAX 128
#define INVITE_PROTO_UDP 0x01
#define INVITE_PROTO_TCP 0x02
#define INVITE_PROTO_BOTH 0x03
#ifdef __cplusplus
extern "C" {
#endif
struct InviteAddr {
int family;
uint8_t address[16];
uint8_t socketId;
uint8_t proto;
uint16_t port;
};
struct InviteData {
uint64_t channelId;
uint8_t pubkey[INVITE_PUBKEY_SIZE];
uint64_t nodeId;
uint8_t addrCount;
struct InviteAddr addrs[INVITE_ADDR_MAX];
char password[INVITE_PASS_MAX];
uint8_t password_len;
};
int invite_link_decode(const char* link, size_t link_len, struct InviteData* out,
char* error_buf, size_t error_buf_size);
int invite_link_encode(const struct InviteData* data, const char* password, char* out, size_t out_size);
int invite_serialize_addrs(const struct InviteData* data, uint8_t* buf, size_t buf_size);
#ifdef __cplusplus
}
#endif
#endif /* INVITE_LINK_H */
+5
View File
@@ -767,6 +767,11 @@ static int parse_chatserver(const char *key, const char *value, struct global_co
global->chatserver_storage_unit_size = sz;
return 0;
}
if (strcmp(key, "headless_control_bind") == 0) {
strncpy(global->headless_control_bind, value, sizeof(global->headless_control_bind) - 1);
global->headless_control_bind[sizeof(global->headless_control_bind) - 1] = '\0';
return 0;
}
return -1;
}
+1
View File
@@ -202,6 +202,7 @@ struct global_config {
char chatserver_allowed_groups[1024]; // "all" или hex-ID через запятую
uint64_t chatserver_storage_total_size; // bytes, общий лимит хранилища (0 = безлимит)
uint64_t chatserver_storage_unit_size; // bytes, лимит одного файла (0 = отключено)
char headless_control_bind[64]; // ip:port для headless chat CLI (пусто = отключено)
};
struct utun_config {
+36 -151
View File
@@ -21,7 +21,7 @@
#include "topo_node.h"
#include "topo_node_sqlite.h"
#include "topo_group.h"
#include "../transport_layer/stcp_link.h"
#include "../transport_layer/etcp_connect.h"
#include "route_ping.h"
/* ═══════ forward-декларации ═══════ */
@@ -298,37 +298,6 @@ void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void
/* NCD → invite: UP (отправляем INVITE_INFO_REQ для проверки членства),
* TIMEOUT/DOWN (фейлим invite с CONN_EVENT_TIMEOUT). */
void cm_invite_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (inv->cancelled) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: NCD callback IGNORED — already cancelled node=0x%016llx", (unsigned long long)inv->node_id); return; }
switch (ncd_ev) {
case NCD_EVENT_UP:
if (inv->state != CM_INVITE_CONNECTING) return;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: ETCP handshake OK with 0x%016llx — sending membership check",
(unsigned long long)inv->node_id);
inv->state = CM_INVITE_WAIT_INFO;
{ uint8_t buf[11]; /* dgram[0]=cmd(dispatch), dgram[1]=cmd(handler skip), dgram[2]=subcmd, dgram[3..10]=group_id */
buf[0] = ETCP_RT_ID_CONN_MGR; buf[1] = ETCP_RT_ID_CONN_MGR; buf[2] = CM_SUBCMD_INVITE_INFO_REQ;
memcpy(buf + 3, &inv->mgr->group->group_id, 8);
struct ll_entry* qe = queue_entry_new(0);
if (!qe) { cm_invite_fail(inv); return; }
qe->dgram = u_malloc(sizeof(buf)); if (!qe->dgram) { queue_entry_free(qe); cm_invite_fail(inv); return; }
memcpy(qe->dgram, buf, sizeof(buf)); qe->len = sizeof(buf);
etcp_send(node_conn_direct_get_conn(ncd_h), qe);
}
break;
case NCD_EVENT_TIMEOUT:
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: ETCP connect TIMEOUT to 0x%016llx — no response from peer",
(unsigned long long)inv->node_id);
cm_invite_fail(inv); break;
case NCD_EVENT_DOWN:
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: ETCP link DROPPED to 0x%016llx while connecting",
(unsigned long long)inv->node_id);
if (inv->state == CM_INVITE_CONNECTING) cm_invite_fail(inv);
break;
}
}
/* ═══════ init / destroy ═══════ */
struct CONN_MGR* conn_mgr_init(struct TOPO_GROUP* group) {
@@ -804,23 +773,11 @@ void cm_handle_invite_info_resp(struct CONN_MGR* mgr, struct ETCP_CONN* conn, co
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: membership CONFIRMED by 0x%016llx name=\"%.*s\" — joined to channel, handle registered",
(unsigned long long)inv->node_id, resp->node_name_len, resp->node_name_len ? (const char*)resp->node_name : "");
if (inv->cb) inv->cb(inv->handle, inv->node_id, inv->mgr->group->group_id, CONN_EVENT_JOIN, inv->cb_arg);
if (inv->ncd_handle) {
struct CONN_MGR_ENTRY* old_e = cm_find_entry(inv->mgr, inv->node_id);
if (old_e) {
if (old_e->ncd_handle) { node_conn_direct_force_close(old_e->ncd_handle); old_e->ncd_handle = NULL; }
cm_entry_cleanup(old_e);
}
struct CONN_MGR_ENTRY* cm_entry = cm_ensure_entry(inv->mgr, inv->node_id);
if (cm_entry) {
node_conn_direct_transfer(inv->ncd_handle, cm_ncd_callback, cm_entry);
cm_entry->ncd_handle = inv->ncd_handle;
cm_entry->state = CONN_MGR_STATE_CONNECTED;
cm_entry->conn_type = CONN_TYPE_DIRECT;
cm_update_nodeinfo(cm_entry);
}
inv->ncd_handle = NULL;
if (inv->invite_conn) {
node_conn_direct_open_node(inv->mgr->instance, inv->node_id, cm_ncd_callback, NULL, NULL, NULL, NULL);
inv->invite_conn = NULL;
}
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_SUCCESS] ncd transferred, temp_nq kept in group — calling cancel");
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_SUCCESS] conn transferred to ncd, calling cancel");
inv->temp_nq = NULL;
cm_invite_cancel(inv);
}
@@ -845,43 +802,24 @@ static void cm_invite_free_deferred(void* arg) {
static void cm_invite_cancel(struct cm_invite_pending* inv) {
if (!inv) return;
inv->cancelled = 1;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 0 inv=%p node=0x%016llx mgr=%p",
inv, (unsigned long long)inv->node_id, inv->mgr);
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] inv=%p node=0x%016llx mgr=%p conn=%p",
inv, (unsigned long long)inv->node_id, inv->mgr, (void*)inv->invite_conn);
if (!inv->mgr) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[INV_CANCEL] inv->mgr is NULL!"); goto inv_cancel_free; }
if (!inv->mgr->instance) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[INV_CANCEL] inv->mgr->instance is NULL!"); goto inv_cancel_free; }
if (inv->overall_timer) {
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 1 overall_timer=%p — cancelling", inv->overall_timer);
uasync_cancel_timeout(inv->mgr->instance->ua, inv->overall_timer); inv->overall_timer = NULL;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 1 overall_timer done");
} else
{ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 1 overall_timer skip (NULL)"); }
if (inv->tcp_link) {
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 2 tcp_link=%p — closing", inv->tcp_link);
struct ll_entry* e = queue_find_data_by_index(inv->mgr->instance->tcp_connections, (const uint8_t*)&inv->node_id);
if (e) queue_remove_data(inv->mgr->instance->tcp_connections, e);
stcp_link_close(inv->tcp_link); inv->tcp_link = NULL;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 2 tcp_link done");
} else
{ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 2 tcp_link skip (NULL)"); }
if (inv->ncd_handle) {
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 3 ncd_handle=%p — closing", inv->ncd_handle);
node_conn_direct_close(inv->ncd_handle); inv->ncd_handle = NULL;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 3 ncd done");
} else
{ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 3 ncd skip (NULL)"); }
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] overall_timer cancelled");
}
if (inv->temp_nq) {
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 4 temp_nq=%p group=%p — removing from nodes",
inv->temp_nq, inv->mgr->group);
queue_remove_data(inv->mgr->group->nodes, &inv->temp_nq->ll);
topo_nodeq_free_group_fields(inv->mgr->instance->topo_groups, inv->temp_nq);
queue_entry_free(&inv->temp_nq->ll); inv->temp_nq = NULL;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 4 temp_nq done");
} else
{ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 4 temp_nq skip (NULL)"); }
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] temp_nq removed");
}
{ struct cm_invite_pending** pp = &inv->mgr->invite_list;
while (*pp) { if (*pp == inv) { *pp = inv->next; break; } pp = &(*pp)->next; }
}
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] 5 unlink done");
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] unlinked from invite_list");
inv_cancel_free:
if (inv->mgr && inv->mgr->instance && inv->mgr->instance->ua)
{ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[INV_CANCEL] posting deferred free inv=%p", inv);
@@ -896,9 +834,9 @@ inv_cancel_free:
void cm_invite_fail(struct cm_invite_pending* inv) {
if (!inv) return;
if (inv->cancelled) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "cm_invite_fail: IGNORED — already cancelled node=0x%016llx", (unsigned long long)inv->node_id); return; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "cm_invite_fail: node=0x%016llx state=%d ncd=%p tcp=%p",
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "cm_invite_fail: node=0x%016llx state=%d conn=%p",
(unsigned long long)inv->node_id, (int)inv->state,
(void*)inv->ncd_handle, (void*)inv->tcp_link);
(void*)inv->invite_conn);
conn_mgr_cb_t cb = inv->cb;
struct CONN_MGR_HANDLE* handle = inv->handle;
uint64_t node_id = inv->node_id;
@@ -911,81 +849,29 @@ void cm_invite_fail(struct cm_invite_pending* inv) {
void cm_invite_cleanup(struct cm_invite_pending* inv) { if (inv) u_free(inv); }
static void cm_tcp_ready_cb(struct stcp_link* link, void* arg) {
/* ── etcp_connect callback for invite ── */
static void cm_invite_etcp_cb(void* arg, struct ETCP_CONN* conn, int event) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (!inv) return;
if (inv->cancelled) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: TCP ready IGNORED — already cancelled node=0x%016llx", (unsigned long long)inv->node_id); return; }
if (inv->tcp_ready) return; inv->tcp_ready = 1;
struct ETCP_CONN* etcp = stcp_link_get_etcp_conn(link); etcp->peer_node_id = inv->node_id;
struct ll_entry* qe = queue_entry_new(sizeof(struct tcp_conn_entry));
if (!qe) { cm_invite_fail(inv); return; }
{ struct tcp_conn_entry* te = (struct tcp_conn_entry*)qe->data; te->node_id=inv->node_id; te->link=link; te->etcp_conn=etcp; }
queue_data_put_with_index(inv->mgr->instance->tcp_connections, qe);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: TCP handshake OK with 0x%016llx — sending membership check",
(unsigned long long)inv->node_id);
if (!inv || inv->cancelled) return;
if (event != ETCP_CONNECT_LATE && event != ETCP_CONNECT_EARLY) return;
if (inv->state != CM_INVITE_CONNECTING) return;
if (!conn) { cm_invite_fail(inv); return; }
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: etcp_connect %s for 0x%016llx — sending membership check",
event == ETCP_CONNECT_LATE ? "LATE" : "EARLY", (unsigned long long)inv->node_id);
inv->invite_conn = conn;
inv->state = CM_INVITE_WAIT_INFO;
{ uint8_t buf[11]; /* dgram[0]=cmd(dispatch), dgram[1]=cmd(handler skip), dgram[2]=subcmd, dgram[3..10]=group_id */
{ uint8_t buf[11];
buf[0] = ETCP_RT_ID_CONN_MGR; buf[1] = ETCP_RT_ID_CONN_MGR; buf[2] = CM_SUBCMD_INVITE_INFO_REQ;
memcpy(buf + 3, &inv->mgr->group->group_id, 8);
struct ll_entry* qe2 = queue_entry_new(0); if (!qe2) { cm_invite_fail(inv); return; }
qe2->dgram=u_malloc(sizeof(buf)); if (!qe2->dgram) { queue_entry_free(qe2); cm_invite_fail(inv); return; }
memcpy(qe2->dgram,buf,sizeof(buf)); qe2->len=sizeof(buf); etcp_send(etcp, qe2);
struct ll_entry* qe = queue_entry_new(0);
if (!qe) { cm_invite_fail(inv); return; }
qe->dgram = u_malloc(sizeof(buf)); if (!qe->dgram) { queue_entry_free(qe); cm_invite_fail(inv); return; }
memcpy(qe->dgram, buf, sizeof(buf)); qe->len = sizeof(buf); etcp_send(conn, qe);
}
}
static void cm_tcp_close_cb(struct stcp_link* link, int err, void* arg) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (!inv) return;
if (inv->cancelled) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: TCP close IGNORED — already cancelled node=0x%016llx err=%d", (unsigned long long)inv->node_id, err); return; }
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: TCP link CLOSED to 0x%016llx err=%d", (unsigned long long)inv->node_id, err);
inv->tcp_link = NULL;
}
static int cm_invite_tcp_connect(struct cm_invite_pending* inv, const struct TOPO_NODE* ni) {
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
if (!(a->protocol & TOPO_PROTO_TCP)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "[cm_invite] v4 skip proto=%02x port=%d node=0x%016llx", a->protocol, (int)a->port, (unsigned long long)inv->node_id);
continue;
}
struct sockaddr_in sin; memset(&sin,0,sizeof(sin)); sin.sin_family=AF_INET;
memcpy(&sin.sin_addr.s_addr,a->addr,4); sin.sin_port=htons(a->port);
struct sockaddr_storage sa; memcpy(&sa,&sin,sizeof(sin));
struct stcp_link_config cfg = {.ua=inv->mgr->instance->ua, .my_keys=&inv->mgr->instance->my_keys, .inst=inv->mgr->instance,
.peer_pubkey=ni->public_key, .peer_pubkey_mode=0, .remote_addr=&sa, .remote_port=a->port};
inv->tcp_link = stcp_link_connect(&cfg);
if (inv->tcp_link) { stcp_link_set_on_ready(inv->tcp_link, cm_tcp_ready_cb, inv); stcp_link_set_on_close(inv->tcp_link, cm_tcp_close_cb, inv);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: TCP connecting to %d.%d.%d.%d:%d",
a->addr[0], a->addr[1], a->addr[2], a->addr[3], (int)a->port);
return 1; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "[cm_invite] v4 connect FAIL %d.%d.%d.%d:%d node=0x%016llx",
a->addr[0], a->addr[1], a->addr[2], a->addr[3], (int)a->port, (unsigned long long)inv->node_id);
}
for (const struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) {
if (!(a6->protocol & TOPO_PROTO_TCP)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "[cm_invite] v6 skip proto=%02x port=%d node=0x%016llx", a6->protocol, (int)a6->port, (unsigned long long)inv->node_id);
continue;
}
struct sockaddr_in6 sin6; memset(&sin6,0,sizeof(sin6)); sin6.sin6_family=AF_INET6;
memcpy(&sin6.sin6_addr,a6->addr,16); sin6.sin6_port=htons(a6->port);
if (cm_classify_v6_addr(a6->addr) == CM_V6_LL) {
struct ETCP_SOCKET* sv = inv->mgr->instance->etcp_sockets;
while (sv) { if (sv->local_addr.ss_family == AF_INET6 && sv->netif_index) { sin6.sin6_scope_id = sv->netif_index; break; } sv = sv->next; }
if (sin6.sin6_scope_id == 0)
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "[cm_invite] v6 LL no scope_id port=%d node=0x%016llx", (int)a6->port, (unsigned long long)inv->node_id);
}
struct sockaddr_storage sa; memcpy(&sa,&sin6,sizeof(sin6));
struct stcp_link_config cfg = {.ua=inv->mgr->instance->ua, .my_keys=&inv->mgr->instance->my_keys, .inst=inv->mgr->instance,
.peer_pubkey=ni->public_key, .peer_pubkey_mode=0, .remote_addr=&sa, .remote_port=a6->port};
inv->tcp_link = stcp_link_connect(&cfg);
if (inv->tcp_link) { stcp_link_set_on_ready(inv->tcp_link, cm_tcp_ready_cb, inv); stcp_link_set_on_close(inv->tcp_link, cm_tcp_close_cb, inv);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: TCP6 connecting to port %d",
(int)a6->port);
return 1; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "[cm_invite] v6 connect FAIL port=%d node=0x%016llx", (int)a6->port, (unsigned long long)inv->node_id);
}
return 0;
}
int conn_mgr_open_invite(struct UTUN_INSTANCE* inst,
uint64_t group_id,
struct TOPO_NODE* ni,
@@ -1047,11 +933,10 @@ int conn_mgr_open_invite(struct UTUN_INSTANCE* inst,
if (!h) { cm_invite_fail(inv); return -1; } inv->handle=h; if (out_handle) *out_handle=h;
queue_data_put_with_index(group->nodes, &nq->ll);
/* 6. NCD + TCP */
int r = node_conn_direct_open_node(inst, nid, cm_invite_ncd_callback, inv, &inv->ncd_handle, gni, NULL);
if (r == NCD_ERR) { cm_invite_fail(inv); if (out_handle) *out_handle=NULL; return -1; }
int udp = (r == NCD_NEW || r == NCD_REUSED), tcp = cm_invite_tcp_connect(inv, gni);
if (!udp && !tcp) { cm_invite_fail(inv); if (out_handle) *out_handle=NULL; return -1; }
/* 6. etcp_connect — запускает UDP links + TCP STCP link через единый API */
if (etcp_connect(inst, nq, cm_invite_etcp_cb, inv, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) != 0) {
cm_invite_fail(inv); if (out_handle) *out_handle=NULL; return -1;
}
inv->overall_timer = uasync_set_timeout(inst->ua, CM_INVITE_DEFAULT_TIMEOUT_MS*10, inv, cm_invite_overall_timeout, "conn_mgr_invite");
{
@@ -1063,8 +948,8 @@ int conn_mgr_open_invite(struct UTUN_INSTANCE* inst,
for (const struct TOPO_ADDR6* a6 = gni->v6_addrs; a6; a6 = a6->next)
aoff += snprintf(addrs + aoff, sizeof(addrs) - (size_t)aoff, " [v6]:%d(%s)", (int)a6->port,
a6->protocol == TOPO_PROTO_UDP ? "UDP" : a6->protocol == TOPO_PROTO_TCP ? "TCP" : "?");
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: starting to 0x%016llx via%s udp=%d tcp=%d timeout=%ds",
(unsigned long long)nid, addrs, udp, tcp, CM_INVITE_DEFAULT_TIMEOUT_MS/1000);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: starting to 0x%016llx via%s timeout=%ds",
(unsigned long long)nid, addrs, CM_INVITE_DEFAULT_TIMEOUT_MS/1000);
}
return 0;
}
+2 -3
View File
@@ -109,8 +109,8 @@ enum CM_INVITE_STATE { CM_INVITE_CONNECTING = 0, CM_INVITE_WAIT_INFO = 1, };
struct cm_invite_pending {
struct cm_invite_pending* next; struct CONN_MGR* mgr;
uint64_t node_id; uint8_t state;
struct NODE_CONN_DIRECT* ncd_handle; struct stcp_link* tcp_link;
uint8_t tcp_ready:1, cancelled:1; void* overall_timer;
struct ETCP_CONN* invite_conn;
uint8_t cancelled:1; void* overall_timer;
struct CONN_MGR_HANDLE* handle; struct TOPO_GROUP_NODE* temp_nq;
conn_mgr_cb_t cb; void* cb_arg;
};
@@ -160,7 +160,6 @@ void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event);
void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry);
void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id);
void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg);
void cm_invite_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg);
void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry);
void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry);
void cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry);
+2 -2
View File
@@ -46,7 +46,7 @@ static void wait_ack_cb(struct ll_queue* q, void* arg);
static void send_ack_req_cb(struct ll_queue* q, void* arg);
static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp);
static void etcp_connection_free_deferred(void* arg);
static void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status);
void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status);
struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp);
static void clear_queue(struct ll_queue* q) {
@@ -286,7 +286,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n
}
static void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status) {
void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status) {
if (!etcp || !etcp->instance) return;
static const char* names[] = { "NEW", "UP", "DOWN", "DELETE" };
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[%s] Connection status changed to %s", etcp->log_name,
+3
View File
@@ -312,6 +312,9 @@ void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t d
// Recalculate optimal_inflight and resume input queue after link inflight_lim change
void etcp_conn_on_inflight_lim_changed(struct ETCP_CONN* etcp);
// Fire connection status callback on instance's conn_status_cbks chain
void etcp_fire_conn_status(struct ETCP_CONN* conn, int status);
// Update log_name when peer_node_id becomes known
void etcp_update_log_name(struct ETCP_CONN* etcp);
+4
View File
@@ -139,8 +139,12 @@ static void tcp_link_ready_cb(struct stcp_link* link, void* arg) {
struct ETCP_CONNECT* ctx = (struct ETCP_CONNECT*)arg;
if (!ctx || ctx->done || ctx->tcp_ready) return;
ctx->tcp_ready = 1;
ctx->conn->initialized = 1;
ctx->conn->links_up = 1;
ctx->conn->send_input_q = stcp_link_get_etcp_conn(link) ? stcp_link_get_etcp_conn(link)->send_input_q : NULL;
DEBUG_INFO(DEBUG_CATEGORY_ETCP_CONNECT, "[etcp_connect] TCP link ready for node 0x%016llx", (unsigned long long)ctx->node_id);
connect_init_cb(ctx->conn, ETCP_CBK_EVENT_INIT, ctx);
etcp_fire_conn_status(ctx->conn, ETCP_CONN_STATUS_UP);
}
static void tcp_link_close_cb(struct stcp_link* link, int err, void* arg) {
+22 -5
View File
@@ -42,17 +42,34 @@ static const char* nat_type_str(uint8_t nat_type) {
// TCP server: on new incoming connection → create minimal ETCP_CONN
static void tcp_server_on_link(struct stcp_link *link, void *arg) {
struct UTUN_INSTANCE *inst = (struct UTUN_INSTANCE *)arg;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: start link=%p inst=%p", (void*)link, (void*)inst);
struct ETCP_CONN *conn = etcp_connection_create(inst, NULL);
if (!conn) return;
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: etcp_connection_create FAILED"); return; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: conn=%p log=%s", (void*)conn, conn->log_name);
conn->transport_link = link;
struct ETCP_CONN *lconn = stcp_link_get_etcp_conn(link);
conn->send_input_q = lconn ? lconn->send_input_q : NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: lconn=%p send_input_q=%p", (void*)lconn, (void*)conn->send_input_q);
conn->initialized = 1;
conn->links_up = 1;
snprintf(conn->log_name, sizeof(conn->log_name), "tcp-[%p]", (void*)link);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: firing new_conn_cbks (count=%d)",
({ int c=0; for(struct etcp_inst_cbk_entry* e=inst->new_conn_cbks;e;e=e->next)c++; c; }));
struct etcp_inst_cbk_entry* cbe = inst->new_conn_cbks;
while (cbe) { struct etcp_inst_cbk_entry* n = cbe->next; cbe->fn(conn, cbe->arg); cbe = n; }
etcp_cbk_fire(conn, ETCP_CBK_EVENT_INIT);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "TCP server new conn=%p total=%d pending=%d",
(void*)conn,
queue_entry_count(inst->connections),
queue_entry_count(inst->connections));
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: firing ETCP_CONN_STATUS_UP");
etcp_fire_conn_status(conn, ETCP_CONN_STATUS_UP);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "tcp_server_on_link: DONE conn=%p send_input_q=%p total=%d",
(void*)conn, (void*)conn->send_input_q, queue_entry_count(inst->connections));
}
// Forward declaration
+2
View File
@@ -25,6 +25,8 @@ void stcp_conn_set_on_close(struct stcp_conn *c, void (*cb)(struct stcp_conn *co
void stcp_conn_free(struct stcp_conn *c) {
if (!c) return;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_conn_free: c=%p sock=%d sock_id=%p recv_buf=%p state=%d allocated=%d",
(void*)c, (int)c->sock, (void*)c->socket_id, (void*)c->recv_buf, (int)c->state, (int)c->allocated);
if (c->socket_id) { uasync_remove_socket_t(c->ua, c->sock); c->socket_id = NULL; }
if (c->sock != SOCKET_INVALID) { socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID; }
if (c->recv_buf) { u_free(c->recv_buf); c->recv_buf = NULL; }
+58 -30
View File
@@ -44,7 +44,7 @@ static int encrypt_and_crc(struct stcp_conn *c, const uint8_t *data, size_t data
output[2 + data_len + 2] = (uint8_t)(crc >> 16);
output[2 + data_len + 3] = (uint8_t)(crc >> 24);
if (sc_stream_xor(&c->stream_send, output + 2, data_len + 4) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stream_xor failed len=%zu", data_len);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_xor failed len=%zu", data_len);
return -1;
}
*output_len = 2 + data_len + 4;
@@ -52,12 +52,12 @@ static int encrypt_and_crc(struct stcp_conn *c, const uint8_t *data, size_t data
}
static int decrypt_and_check(uint8_t *data, size_t len, struct sc_stream_state *stream, size_t *out_len) {
if (sc_stream_xor(stream, data, len) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stream_xor failed"); return -1; }
if (sc_stream_xor(stream, data, len) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_xor failed"); return -1; }
size_t data_len = len - SC_CRC32_SIZE;
uint32_t recv_crc = ((uint32_t)data[data_len]) | ((uint32_t)data[data_len + 1] << 8) |
((uint32_t)data[data_len + 2] << 16) | ((uint32_t)data[data_len + 3] << 24);
uint32_t calc_crc = crc32_calc(data, data_len);
if (recv_crc != calc_crc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "CRC mismatch"); return -1; }
if (recv_crc != calc_crc) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "CRC mismatch"); return -1; }
*out_len = data_len;
return 0;
}
@@ -66,12 +66,14 @@ static void client_do_close(struct stcp_conn *c, int err) {
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) return;
int prev = c->state;
c->state = STCP_STATE_CLOSED;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_client: closed is_server=%d err=%d prev_state=%d", c->is_server, err, prev);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_client: closed is_server=%d err=%d prev_state=%d", c->is_server, err, prev);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: do_close sock=%d sock_id=%p recv_buf=%p len=%zu cap=%zu send_buf=%p on_close=%p",
(int)c->sock, (void*)c->socket_id, (void*)c->recv_buf, c->recv_buf_len, c->recv_buf_cap, (void*)c->send_buf, (void*)c->on_close);
if (c->socket_id) { uasync_remove_socket_t(c->ua, c->sock); c->socket_id = NULL; }
if (c->sock != SOCKET_INVALID) { socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID; }
if (c->recv_buf) { u_free(c->recv_buf); c->recv_buf = NULL; c->recv_buf_len = 0; c->recv_buf_cap = 0; }
if (c->send_buf) { u_free(c->send_buf); c->send_buf = NULL; c->send_len = 0; }
if (c->on_close) { void (*cb)(struct stcp_conn*, int, void*) = c->on_close; c->on_close = NULL; cb(c, err, c->close_arg); }
if (c->recv_buf) { void *p = c->recv_buf; u_free(c->recv_buf); c->recv_buf = NULL; c->recv_buf_len = 0; c->recv_buf_cap = 0; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: do_close u_free recv_buf=%p", p); }
if (c->send_buf) { void *p = c->send_buf; u_free(c->send_buf); c->send_buf = NULL; c->send_len = 0; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: do_close u_free send_buf=%p", p); }
if (c->on_close) { void (*cb)(struct stcp_conn*, int, void*) = c->on_close; c->on_close = NULL; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: do_close calling on_close=%p", (void*)cb); cb(c, err, c->close_arg); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: do_close on_close returned"); }
}
static int client_try_send(struct stcp_conn *c, const uint8_t *data, size_t len) {
@@ -81,7 +83,7 @@ static int client_try_send(struct stcp_conn *c, const uint8_t *data, size_t len)
if (sent < 0) {
int e = socket_get_error();
if (e == ERR_AGAIN || e == ERR_WOULDBLOCK) { c->send_buf = (uint8_t *)data; c->send_len = len; c->send_offset = 0; uasync_set_socket_write(c->ua, c->socket_id, 1); return 0; }
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "send failed err=%d", e);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "send failed err=%d", e);
u_free((uint8_t *)data); client_do_close(c, e); return -1;
}
if ((size_t)sent < len) { c->send_buf = (uint8_t *)data; c->send_len = len; c->send_offset = (size_t)sent; uasync_set_socket_write(c->ua, c->socket_id, 1); return 0; }
@@ -120,20 +122,20 @@ static void client_tx_queue_cb(struct ll_queue *q, void *arg) {
static int client_derive_session(struct stcp_conn *c, const uint8_t *peer_pubkey) {
struct secure_channel sc;
sc_init_ctx(&sc, &c->my_keys);
if (sc_set_peer_public_key(&sc, peer_pubkey, SC_PEER_PUBKEY_BIN) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "client ECDH failed"); return -1; }
if (sc_set_peer_public_key(&sc, peer_pubkey, SC_PEER_PUBKEY_BIN) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client ECDH failed"); return -1; }
memcpy(c->session_key, sc.session_key, SC_SESSION_KEY_SIZE);
if (sc_stream_init(&sc, &c->stream_send, STCP_STREAM_CLIENT_SEND) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "client stream_send init failed"); return -1; }
if (sc_stream_init(&sc, &c->stream_recv, STCP_STREAM_SERVER_SEND) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "client stream_recv init failed"); return -1; }
if (sc_stream_init(&sc, &c->stream_send, STCP_STREAM_CLIENT_SEND) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client stream_send init failed"); return -1; }
if (sc_stream_init(&sc, &c->stream_recv, STCP_STREAM_SERVER_SEND) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client stream_recv init failed"); return -1; }
return 0;
}
static void client_send_handshake(struct stcp_conn *c, const uint8_t *server_pubkey) {
uint8_t salt[SC_PUBKEY_ENC_SALT_SIZE];
if (random_bytes(salt, SC_PUBKEY_ENC_SALT_SIZE) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "client random_bytes failed"); client_do_close(c, 1); return; }
if (random_bytes(salt, SC_PUBKEY_ENC_SALT_SIZE) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client random_bytes failed"); client_do_close(c, 1); return; }
uint16_t padding = 8;
size_t total = SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_CLIENT + padding;
uint8_t *hs = u_malloc(total);
if (!hs) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "client hs malloc failed"); client_do_close(c, 1); return; }
if (!hs) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client hs malloc failed"); client_do_close(c, 1); return; }
memcpy(hs, salt, SC_PUBKEY_ENC_SALT_SIZE);
sc_obfuscate_pubkey(salt, server_pubkey, c->my_keys.public_key, hs + SC_PUBKEY_ENC_SALT_SIZE);
@@ -147,7 +149,7 @@ static void client_send_handshake(struct stcp_conn *c, const uint8_t *server_pub
for (int i = 0; i < padding; i++) hs[SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_CLIENT + i] = (uint8_t)(salt[0] ^ i);
c->state = STCP_STATE_HS_CLIENT_SENT;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_client: handshake sent (%zu bytes), entering HS_CLIENT_SENT", total);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_client: handshake sent (%zu bytes), entering HS_CLIENT_SENT", total);
client_try_send(c, hs, total);
}
@@ -166,8 +168,8 @@ static void process_server_response(struct stcp_conn *c) {
uint8_t status = enc_hs[0];
uint16_t padding_size = (uint16_t)enc_hs[1] | ((uint16_t)enc_hs[2] << 8);
c->hs_expected_len = SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_SERVER + padding_size;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_client: server response OK status=%d padding=%u", status, padding_size);
if (status != 0) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "server handshake status=%d", status); client_do_close(c, 4); return; }
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_client: server response OK status=%d padding=%u", status, padding_size);
if (status != 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "server handshake status=%d", status); client_do_close(c, 4); return; }
}
static void finish_server_response(struct stcp_conn *c) {
@@ -176,11 +178,17 @@ static void finish_server_response(struct stcp_conn *c) {
c->hs_expected_len = 0;
c->hs_key_processed = 0;
c->state = STCP_STATE_DATA;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_client: handshake OK, entering DATA state");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_client: handshake OK, entering DATA state");
if (c->on_ready) { void (*cb)(struct stcp_conn*, void*) = c->on_ready; c->on_ready = NULL; cb(c, c->ready_arg); }
}
static void stcp_conn_process_recv(struct stcp_conn *c) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: process_recv entry state=%d sock=%d recv_buf=%p len=%zu cap=%zu",
c->state, (int)c->sock, (void*)c->recv_buf, c->recv_buf_len, c->recv_buf_cap);
if (!c->recv_buf && c->recv_buf_len > 0) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_client: process_recv NULL recv_buf with len=%zu — closing", c->recv_buf_len);
client_do_close(c, EFAULT); return;
}
while (c->recv_buf_len > 0) {
switch (c->state) {
case STCP_STATE_HS_CLIENT_SENT:
@@ -194,7 +202,7 @@ static void stcp_conn_process_recv(struct stcp_conn *c) {
case STCP_STATE_DATA: {
if (c->recv_buf_len < 2) return;
uint16_t msg_size = (uint16_t)c->recv_buf[0] | ((uint16_t)c->recv_buf[1] << 8);
if (msg_size > STCP_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "client msg too large: %u", msg_size); client_do_close(c, 5); return; }
if (msg_size > STCP_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "client msg too large: %u", msg_size); client_do_close(c, 5); return; }
size_t total = 2 + msg_size + SC_CRC32_SIZE;
if (c->recv_buf_len < total) return;
uint8_t *enc_data = c->recv_buf + 2;
@@ -203,6 +211,13 @@ static void stcp_conn_process_recv(struct stcp_conn *c) {
if (c->rx_queue) {
struct ll_entry *e = queue_entry_new(0);
if (e) { e->dgram = u_malloc(data_len); if (e->dgram) { if (data_len) memcpy(e->dgram, enc_data, data_len); e->len = (uint16_t)data_len; queue_data_put(c->rx_queue, e); } else { queue_entry_free(e); } }
if (c->state != STCP_STATE_DATA) return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: memmove recv_buf=%p src=%p(+%zu) len=%zu (total=%zu recv_buf_len=%zu)",
(void*)c->recv_buf, (void*)(c->recv_buf + total), total, c->recv_buf_len - total, total, c->recv_buf_len);
if ((uintptr_t)c->recv_buf < 4096) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_client: suspicious recv_buf=%p (addr < 4K) — closing", (void*)c->recv_buf);
client_do_close(c, EFAULT); return;
}
memmove(c->recv_buf, c->recv_buf + total, c->recv_buf_len - total);
c->recv_buf_len -= total;
@@ -215,7 +230,12 @@ static void stcp_conn_process_recv(struct stcp_conn *c) {
static void client_conn_read_cb(socket_t sock, void *arg) {
struct stcp_conn *c = (struct stcp_conn *)arg;
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) return;
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: read_cb SKIP state=%d sock=%d c=%p recv_buf=%p", c->state, (int)sock, (void*)c, (void*)c->recv_buf);
return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_client: read_cb state=%d sock=%d c=%p recv_buf=%p len=%zu cap=%zu",
c->state, (int)sock, (void*)c, (void*)c->recv_buf, c->recv_buf_len, c->recv_buf_cap);
if (!c->recv_buf) { c->recv_buf_cap = STCP_RECV_BUF_INIT; c->recv_buf = u_malloc(c->recv_buf_cap); if (!c->recv_buf) { client_do_close(c, ENOMEM); return; } }
if (c->recv_buf_len + 4096 > c->recv_buf_cap) {
size_t nc = c->recv_buf_cap * 2; if (nc > STCP_RECV_BUF_MAX) nc = STCP_RECV_BUF_MAX;
@@ -236,10 +256,10 @@ static void client_connect_write_cb(socket_t sock, void *arg) {
int err = 0;
socklen_t len = sizeof(err);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &len) < 0 || err != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_client connect failed err=%d", err);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_client connect failed err=%d", err);
client_do_close(c, err); return;
}
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_client: TCP connected, starting handshake");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_client: TCP connected, starting handshake");
uasync_remove_socket_t(cli->ua, sock);
c->socket_id = uasync_add_socket_t(cli->ua, sock, client_conn_read_cb, client_conn_write_cb, NULL, c);
if (!c->socket_id) { client_do_close(c, ENOMEM); return; }
@@ -252,9 +272,9 @@ struct stcp_client *stcp_client_connect(struct UASYNC *ua, const char *addr, uin
struct SC_MYKEYS *keys, const uint8_t *peer_pubkey,
stcp_ready_cb ready_cb, void *arg,
stcp_close_cb close_cb, void *close_arg) {
if (!ua || !addr || !keys || !peer_pubkey || !ready_cb) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "invalid args"); return NULL; }
if (!ua || !addr || !keys || !peer_pubkey || !ready_cb) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "invalid args"); return NULL; }
struct stcp_client *cli = u_calloc(1, sizeof(struct stcp_client));
if (!cli) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "calloc failed"); return NULL; }
if (!cli) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "calloc failed"); return NULL; }
cli->ua = ua; cli->ready_cb = ready_cb; cli->ready_arg = arg;
memcpy(cli->peer_pubkey, peer_pubkey, SC_PUBKEY_SIZE);
struct stcp_conn *c = &cli->conn;
@@ -269,22 +289,30 @@ struct stcp_client *stcp_client_connect(struct UASYNC *ua, const char *addr, uin
char port_str[16]; snprintf(port_str, sizeof(port_str), "%u", port);
struct addrinfo *res;
if (getaddrinfo(addr, port_str, &hints, &res) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_client getaddrinfo %s:%u failed", addr, port);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_client getaddrinfo %s:%u failed", addr, port);
u_free(cli); return NULL;
}
c->sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (c->sock == SOCKET_INVALID) { freeaddrinfo(res); u_free(cli); return NULL; }
socket_set_nonblocking(c->sock);
if (connect(c->sock, res->ai_addr, res->ai_addrlen) < 0) {
int conn_ret = connect(c->sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
if (conn_ret < 0) {
int sock_err = socket_get_error();
if (sock_err != EINPROGRESS && sock_err != ERR_WOULDBLOCK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_client connect to %s:%u failed err=%d(%s)", addr, port, sock_err, socket_strerror(sock_err));
socket_close_wrapper(c->sock); freeaddrinfo(res); u_free(cli); return NULL;
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_client connect to %s:%u failed err=%d(%s)", addr, port, sock_err, socket_strerror(sock_err));
socket_close_wrapper(c->sock); u_free(cli); return NULL;
}
c->socket_id = uasync_add_socket_t(ua, c->sock, NULL, client_connect_write_cb, NULL, cli);
if (!c->socket_id) { socket_close_wrapper(c->sock); u_free(cli); return NULL; }
} else {
/* immediate connect (localhost) — запускаем handshake сразу */
int opt = 1; setsockopt(c->sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt));
c->socket_id = uasync_add_socket_t(ua, c->sock, client_conn_read_cb, client_conn_write_cb, NULL, c);
if (!c->socket_id) { socket_close_wrapper(c->sock); u_free(cli); return NULL; }
if (client_derive_session(c, cli->peer_pubkey)) { client_do_close(c, 1); return cli; }
client_send_handshake(c, cli->peer_pubkey);
}
freeaddrinfo(res);
c->socket_id = uasync_add_socket_t(ua, c->sock, NULL, client_connect_write_cb, NULL, cli);
if (!c->socket_id) { socket_close_wrapper(c->sock); u_free(cli); return NULL; }
return cli;
}
+28 -10
View File
@@ -27,6 +27,7 @@ struct stcp_link {
struct stcp_conn *conn; // STCP connection
uint8_t ready;
uint8_t closing; // deferred close pending
uint8_t peer_pubkey[SC_PUBKEY_SIZE];
stcp_link_cb on_ready_cb;
@@ -80,7 +81,7 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) {
link->etcp_conn.send_input_q = link->tx_queue;
stcp_conn_set_tx_queue(conn, link->tx_queue);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_link: server accepted connection");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: server accepted connection");
if (link->on_ready_cb) link->on_ready_cb(link, link->ready_arg);
if (ss->on_link) ss->on_link(link, ss->on_link_arg);
}
@@ -104,7 +105,7 @@ static void client_ready_cb(struct stcp_conn *conn, void *arg) {
link->etcp_conn.send_input_q = link->tx_queue;
stcp_conn_set_tx_queue(conn, link->tx_queue);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_link: client connected, peer_pubkey=%02x%02x%02x%02x...",
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: client connected, peer_pubkey=%02x%02x%02x%02x...",
link->peer_pubkey[0], link->peer_pubkey[1], link->peer_pubkey[2], link->peer_pubkey[3]);
if (link->on_ready_cb) link->on_ready_cb(link, link->ready_arg);
}
@@ -121,7 +122,7 @@ struct stcp_server *stcp_server_listen(struct stcp_link_config *cfg, uint16_t po
ss->on_link_arg = arg;
ss->srv = stcp_server_create(cfg->ua, port, cfg->my_keys, server_accept_cb, ss, NULL, NULL, cfg->listen_family);
if (!ss->srv) { u_free(ss); return NULL; }
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "port=%u", port);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "port=%u", port);
return ss;
}
@@ -147,7 +148,7 @@ struct stcp_link *stcp_link_connect(struct stcp_link_config *cfg) {
getnameinfo((struct sockaddr*)sa6, sizeof(*sa6), addr_str, sizeof(addr_str), NULL, 0, NI_NUMERICHOST);
port = cfg->remote_port ? cfg->remote_port : ntohs(sa6->sin6_port);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_link: unsupported address family %d", family);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_link: unsupported address family %d", family);
return NULL;
}
@@ -160,7 +161,7 @@ struct stcp_link *stcp_link_connect(struct stcp_link_config *cfg) {
struct secure_channel sc_tmp;
sc_init_ctx(&sc_tmp, cfg->my_keys);
if (sc_set_peer_public_key(&sc_tmp, cfg->peer_pubkey, 1) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "invalid peer pubkey hex");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "invalid peer pubkey hex");
u_free(link); return NULL;
}
memcpy(link->peer_pubkey, sc_tmp.peer_public_key, SC_PUBKEY_SIZE);
@@ -174,13 +175,13 @@ struct stcp_link *stcp_link_connect(struct stcp_link_config *cfg) {
link->etcp_conn.instance = cfg->inst;
link->etcp_conn.transport_link = (void *)link;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "stcp_link: connecting to %s:%u pubkey=%016llx",
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: connecting to %s:%u pubkey=%016llx",
addr_str, port, (unsigned long long)*(const uint64_t*)link->peer_pubkey);
return link;
}
void stcp_link_close(struct stcp_link *link) {
if (!link) return;
static void stcp_link_close_impl(void *arg) {
struct stcp_link *link = (struct stcp_link *)arg;
struct ll_queue* tq_reg = link->etcp_conn.transit_queues;
if (tq_reg) {
struct ll_entry* te;
@@ -205,6 +206,13 @@ void stcp_link_close(struct stcp_link *link) {
u_free(link);
}
void stcp_link_close(struct stcp_link *link) {
if (!link) return;
if (link->closing) return;
link->closing = 1;
uasync_call_soon(link->cfg.ua, link, stcp_link_close_impl);
}
int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len) {
if (!link || !link->ready) return -1;
struct ll_entry *e = queue_entry_new(0);
@@ -241,11 +249,21 @@ void stcp_link_set_on_close(struct stcp_link *link, void (*cb)(struct stcp_link
link->close_arg = arg;
}
void stcp_link_set_peer_node_id(struct stcp_link *link, uint64_t node_id) {
if (!link) return;
link->etcp_conn.peer_node_id = node_id;
etcp_update_log_name(&link->etcp_conn);
}
const uint8_t *stcp_link_get_peer_pubkey(struct stcp_link *link) {
return link && link->conn ? link->conn->peer_pubkey : NULL;
}
void stcp_server_list_add(struct UTUN_INSTANCE *inst, struct stcp_server *srv) {
if (!inst || !srv) return;
srv->next = inst->stcp_servers;
inst->stcp_servers = srv;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server_list_add: %p total=%d", (void*)srv, stcp_server_list_count(inst));
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server_list_add: %p total=%d", (void*)srv, stcp_server_list_count(inst));
}
void stcp_server_list_destroy_all(struct UTUN_INSTANCE *inst) {
@@ -257,7 +275,7 @@ void stcp_server_list_destroy_all(struct UTUN_INSTANCE *inst) {
inst->stcp_servers = next;
count++;
}
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server_list_destroy_all: destroyed %d servers", count);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server_list_destroy_all: destroyed %d servers", count);
}
int stcp_server_list_count(struct UTUN_INSTANCE *inst) {
+2
View File
@@ -56,6 +56,8 @@ const struct sockaddr_storage *stcp_link_get_remote_addr(struct stcp_link *link)
typedef void (*stcp_link_cb)(struct stcp_link *link, void *arg);
void stcp_link_set_on_ready(struct stcp_link *link, stcp_link_cb cb, void *arg);
void stcp_link_set_on_close(struct stcp_link *link, void (*cb)(struct stcp_link *link, int err, void *arg), void *arg);
void stcp_link_set_peer_node_id(struct stcp_link *link, uint64_t node_id);
const uint8_t *stcp_link_get_peer_pubkey(struct stcp_link *link);
#ifdef __cplusplus
+62 -43
View File
@@ -49,7 +49,7 @@ static int stcp_encrypt_and_crc(struct stcp_conn *c, const uint8_t *data, size_t
output[2 + data_len + 3] = (uint8_t)(crc >> 24);
size_t enc_len = 2 + data_len + 4;
if (sc_stream_xor(&c->stream_send, output + 2, data_len + 4) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stream_xor failed len=%zu", data_len);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_xor failed len=%zu", data_len);
return -1;
}
*output_len = enc_len;
@@ -59,7 +59,7 @@ static int stcp_encrypt_and_crc(struct stcp_conn *c, const uint8_t *data, size_t
static int stcp_decrypt_and_check(uint8_t *plaintext, size_t plaintext_len,
struct sc_stream_state *stream, size_t *out_len) {
if (sc_stream_xor(stream, plaintext, plaintext_len) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stream_xor failed len=%zu", plaintext_len);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_xor failed len=%zu", plaintext_len);
return -1;
}
size_t data_len = plaintext_len - 4;
@@ -67,7 +67,7 @@ static int stcp_decrypt_and_check(uint8_t *plaintext, size_t plaintext_len,
((uint32_t)plaintext[data_len + 2] << 16) | ((uint32_t)plaintext[data_len + 3] << 24);
uint32_t calc_crc = crc32_calc(plaintext, data_len);
if (recv_crc != calc_crc) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "CRC mismatch recv=%08x calc=%08x", recv_crc, calc_crc);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "CRC mismatch recv=%08x calc=%08x", recv_crc, calc_crc);
return -1;
}
*out_len = data_len;
@@ -84,20 +84,22 @@ static void stcp_conn_do_close(struct stcp_conn *c, int err) {
c->state = STCP_STATE_CLOSED;
if (c->socket_id) { uasync_remove_socket_t(c->ua, c->sock); c->socket_id = NULL; }
if (c->sock != SOCKET_INVALID) { socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID; }
if (c->recv_buf) { u_free(c->recv_buf); c->recv_buf = NULL; c->recv_buf_len = 0; c->recv_buf_cap = 0; }
if (c->send_buf) { u_free(c->send_buf); c->send_buf = NULL; c->send_len = 0; }
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_conn closed is_server=%d prev_state=%d err=%d", c->is_server, prev, err);
if (c->on_close) { void (*cb)(struct stcp_conn*, int, void*) = c->on_close; cb(c, err, c->close_arg); }
if (c->recv_buf) { void *p = c->recv_buf; u_free(c->recv_buf); c->recv_buf = NULL; c->recv_buf_len = 0; c->recv_buf_cap = 0; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: do_close u_free recv_buf=%p", p); }
if (c->send_buf) { void *p = c->send_buf; u_free(c->send_buf); c->send_buf = NULL; c->send_len = 0; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: do_close u_free send_buf=%p", p); }
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_conn closed is_server=%d prev_state=%d err=%d sock=%d", c->is_server, prev, err, (int)c->sock);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: do_close sock=%d sock_id=%p recv_buf was=%p on_close=%p",
(int)c->sock, (void*)c->socket_id, (void*)c->recv_buf, (void*)c->on_close);
if (c->on_close) { void (*cb)(struct stcp_conn*, int, void*) = c->on_close; c->on_close = NULL; DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: do_close calling on_close=%p", (void*)cb); cb(c, err, c->close_arg); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: do_close on_close returned"); }
else if (prev == STCP_STATE_HS_SERVER_WAIT) { sc_stream_cleanup(&c->stream_send); sc_stream_cleanup(&c->stream_recv); if (c->allocated) uasync_call_soon(c->ua, c, stcp_conn_deferred_free); }
}
static void stcp_conn_send_message(struct stcp_conn *c, const uint8_t *data, size_t len) {
if (c->state != STCP_STATE_DATA) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "not in DATA state"); return; }
if (c->state != STCP_STATE_DATA) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "not in DATA state"); return; }
size_t max_enc = STCP_MAX_MSG_SIZE;
if (len > max_enc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "data too long %zu", len); return; }
if (len > max_enc) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "data too long %zu", len); return; }
size_t need = 2 + len + 4;
uint8_t *buf = u_malloc(need);
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "malloc(%zu) failed", need); return; }
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "malloc(%zu) failed", need); return; }
if (stcp_encrypt_and_crc(c, data, len, buf, &need)) { u_free(buf); return; }
stcp_conn_try_send(c, buf, need);
}
@@ -105,7 +107,7 @@ static void stcp_conn_send_message(struct stcp_conn *c, const uint8_t *data, siz
static int stcp_conn_try_send(struct stcp_conn *c, const uint8_t *data, size_t len) {
if (c->sock == SOCKET_INVALID) return -1;
if (c->send_buf) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "send_buf already busy");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "send_buf already busy");
return -1;
}
ssize_t sent = send(c->sock, data, len, 0);
@@ -118,7 +120,7 @@ static int stcp_conn_try_send(struct stcp_conn *c, const uint8_t *data, size_t l
uasync_set_socket_write(c->ua, c->socket_id, 1);
return 0;
}
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "send failed err=%d", err);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "send failed err=%d", err);
u_free((uint8_t *)data);
stcp_conn_do_close(c, err);
return -1;
@@ -141,7 +143,7 @@ static void server_conn_write_cb(socket_t sock, void *arg) {
if (sent < 0) {
int err = socket_get_error();
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) return;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "send failed err=%d", err);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "send failed err=%d", err);
stcp_conn_do_close(c, err);
return;
}
@@ -167,16 +169,16 @@ static int derive_session_and_streams(struct stcp_conn *c, const uint8_t *peer_p
struct secure_channel sc;
sc_init_ctx(&sc, &c->my_keys);
if (sc_set_peer_public_key(&sc, peer_pubkey, SC_PEER_PUBKEY_BIN) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "ECDH failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "ECDH failed");
return -1;
}
memcpy(c->session_key, sc.session_key, SC_SESSION_KEY_SIZE);
if (sc_stream_init(&sc, &c->stream_send, STCP_STREAM_SERVER_SEND) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "stream_send init failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_send init failed");
return -1;
}
if (sc_stream_init(&sc, &c->stream_recv, STCP_STREAM_CLIENT_SEND) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "stream_recv init failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stream_recv init failed");
return -1;
}
memcpy(c->peer_pubkey, peer_pubkey, SC_PUBKEY_SIZE);
@@ -194,13 +196,13 @@ static void process_client_handshake(struct stcp_conn *c) {
memcpy(enc_hs, c->recv_buf + SC_PUBKEY_ENC_SIZE, STCP_HS_ENC_CLIENT);
size_t hs_data_len;
if (stcp_decrypt_and_check(enc_hs, STCP_HS_ENC_CLIENT, &c->stream_recv, &hs_data_len)) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "decrypt/CRC failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "decrypt/CRC failed");
stcp_conn_do_close(c, 2); return;
}
uint16_t padding_size = (uint16_t)enc_hs[0] | ((uint16_t)enc_hs[1] << 8);
c->hs_expected_len = SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_CLIENT + padding_size;
c->hs_key_processed = 1;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server: client handshake key processed, expecting %zu bytes padding", c->hs_expected_len - SC_PUBKEY_ENC_SIZE - STCP_HS_ENC_CLIENT);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server: client handshake key processed, expecting %zu bytes padding", c->hs_expected_len - SC_PUBKEY_ENC_SIZE - STCP_HS_ENC_CLIENT);
}
static void finish_client_handshake(struct stcp_conn *c) {
@@ -211,13 +213,13 @@ static void finish_client_handshake(struct stcp_conn *c) {
uint8_t salt2[SC_PUBKEY_ENC_SALT_SIZE];
if (random_bytes(salt2, SC_PUBKEY_ENC_SALT_SIZE) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "random_bytes failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "random_bytes failed");
stcp_conn_do_close(c, 3); return;
}
uint16_t padding = 8;
size_t total_resp = SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_SERVER + padding;
uint8_t *resp = u_malloc(total_resp);
if (!resp) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "malloc failed"); stcp_conn_do_close(c, 3); return; }
if (!resp) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "malloc failed"); stcp_conn_do_close(c, 3); return; }
memcpy(resp, salt2, SC_PUBKEY_ENC_SALT_SIZE);
sc_obfuscate_pubkey(salt2, c->peer_pubkey, c->my_keys.public_key, resp + SC_PUBKEY_ENC_SALT_SIZE);
@@ -228,47 +230,58 @@ static void finish_client_handshake(struct stcp_conn *c) {
enc_dst[3] = (uint8_t)(crc >> 0); enc_dst[4] = (uint8_t)(crc >> 8);
enc_dst[5] = (uint8_t)(crc >> 16); enc_dst[6] = (uint8_t)(crc >> 24);
if (sc_stream_xor(&c->stream_send, enc_dst, STCP_HS_ENC_SERVER) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "encrypt failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "encrypt failed");
u_free(resp); stcp_conn_do_close(c, 3); return;
}
for (int i = 0; i < padding; i++) resp[SC_PUBKEY_ENC_SIZE + STCP_HS_ENC_SERVER + i] = (uint8_t)(salt2[0] ^ i);
c->state = STCP_STATE_DATA;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server: handshake OK, entering DATA state");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server: handshake OK, entering DATA state");
if (c->on_ready) c->on_ready(c, c->ready_arg);
stcp_conn_try_send(c, resp, total_resp);
}
static void server_conn_read_cb(socket_t sock, void *arg) {
struct stcp_conn *c = (struct stcp_conn *)arg;
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) return;
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: read_cb SKIP state=%d sock=%d c=%p recv_buf=%p", c->state, (int)sock, (void*)c, (void*)c->recv_buf);
return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: read_cb state=%d sock=%d c=%p recv_buf=%p len=%zu cap=%zu",
c->state, (int)sock, (void*)c, (void*)c->recv_buf, c->recv_buf_len, c->recv_buf_cap);
if (c->recv_buf_cap == 0) {
c->recv_buf_cap = STCP_RECV_BUF_INIT;
c->recv_buf = u_malloc(c->recv_buf_cap);
if (!c->recv_buf) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "malloc failed"); stcp_conn_do_close(c, ENOMEM); return; }
if (!c->recv_buf) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "malloc failed"); stcp_conn_do_close(c, ENOMEM); return; }
}
if (c->recv_buf_len + 4096 > c->recv_buf_cap) {
size_t new_cap = c->recv_buf_cap * 2;
if (new_cap > STCP_RECV_BUF_MAX) new_cap = STCP_RECV_BUF_MAX;
if (new_cap <= c->recv_buf_cap) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "recv buf full"); stcp_conn_do_close(c, ENOBUFS); return; }
if (new_cap <= c->recv_buf_cap) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "recv buf full"); stcp_conn_do_close(c, ENOBUFS); return; }
uint8_t *nb = u_realloc(c->recv_buf, new_cap);
if (!nb) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "realloc failed"); stcp_conn_do_close(c, ENOMEM); return; }
if (!nb) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "realloc failed"); stcp_conn_do_close(c, ENOMEM); return; }
c->recv_buf = nb; c->recv_buf_cap = new_cap;
}
ssize_t n = recv(sock, c->recv_buf + c->recv_buf_len, c->recv_buf_cap - c->recv_buf_len, 0);
if (n < 0) {
int err = socket_get_error();
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) return;
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "recv failed err=%d", err);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "recv failed err=%d", err);
stcp_conn_do_close(c, err); return;
}
if (n == 0) { DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "EOF"); stcp_conn_do_close(c, 0); return; }
if (n == 0) { DEBUG_INFO(DEBUG_CATEGORY_ETCP, "EOF"); stcp_conn_do_close(c, 0); return; }
c->recv_buf_len += (size_t)n;
stcp_conn_process_recv(c);
}
static void stcp_conn_process_recv(struct stcp_conn *c) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: process_recv entry state=%d sock=%d recv_buf=%p len=%zu cap=%zu",
c->state, (int)c->sock, (void*)c->recv_buf, c->recv_buf_len, c->recv_buf_cap);
if (!c->recv_buf && c->recv_buf_len > 0) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_server: process_recv NULL recv_buf with len=%zu — closing", c->recv_buf_len);
stcp_conn_do_close(c, EFAULT); return;
}
while (c->recv_buf_len > 0) {
switch (c->state) {
case STCP_STATE_HS_SERVER_WAIT:
@@ -285,13 +298,13 @@ static void stcp_conn_process_recv(struct stcp_conn *c) {
case STCP_STATE_DATA: {
if (c->recv_buf_len < 2) return;
uint16_t msg_size = (uint16_t)c->recv_buf[0] | ((uint16_t)c->recv_buf[1] << 8);
if (msg_size > STCP_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "msg too large: %u", msg_size); stcp_conn_do_close(c, 4); return; }
if (msg_size > STCP_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "msg too large: %u", msg_size); stcp_conn_do_close(c, 4); return; }
size_t total = 2 + msg_size + 4;
if (c->recv_buf_len < total) return;
uint8_t *enc_data = c->recv_buf + 2;
size_t data_len;
if (stcp_decrypt_and_check(enc_data, msg_size + 4, &c->stream_recv, &data_len)) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "data decrypt/CRC failed");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "data decrypt/CRC failed");
stcp_conn_do_close(c, 5); return;
}
if (c->rx_queue) {
@@ -299,8 +312,14 @@ static void stcp_conn_process_recv(struct stcp_conn *c) {
if (e) {
e->dgram = u_malloc(data_len);
if (e->dgram) { if (data_len) memcpy(e->dgram, enc_data, data_len); e->len = (uint16_t)data_len; queue_data_put(c->rx_queue, e); }
else { queue_entry_free(e); DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "rx malloc(%zu) failed", data_len); }
} else { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "queue_entry_new failed"); }
else { queue_entry_free(e); DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "rx malloc(%zu) failed", data_len); }
} else { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "queue_entry_new failed"); }
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stcp_server: memmove recv_buf=%p src=%p(+%zu) len=%zu (total=%zu recv_buf_len=%zu)",
(void*)c->recv_buf, (void*)(c->recv_buf + total), total, c->recv_buf_len - total, total, c->recv_buf_len);
if ((uintptr_t)c->recv_buf < 4096) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_server: suspicious recv_buf=%p (addr < 4K) — closing", (void*)c->recv_buf);
stcp_conn_do_close(c, EFAULT); return;
}
memmove(c->recv_buf, c->recv_buf + total, c->recv_buf_len - total);
c->recv_buf_len -= total;
@@ -318,7 +337,7 @@ static void server_accept_cb(socket_t listen_sock, void *arg) {
socket_t cli_sock = accept(listen_sock, (struct sockaddr *)&cli_addr, &addr_len);
if (cli_sock == SOCKET_INVALID) {
int err = socket_get_error();
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_server accept failed err=%d", err);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_server accept failed err=%d", err);
return;
}
socket_set_nonblocking(cli_sock);
@@ -326,7 +345,7 @@ static void server_accept_cb(socket_t listen_sock, void *arg) {
setsockopt(cli_sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt));
struct stcp_conn *c = u_calloc(1, sizeof(struct stcp_conn));
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "stcp_server: calloc conn failed"); socket_close_wrapper(cli_sock); return; }
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_server: calloc conn failed"); socket_close_wrapper(cli_sock); return; }
c->sock = cli_sock;
c->ua = srv->ua;
c->state = STCP_STATE_HS_SERVER_WAIT;
@@ -339,7 +358,7 @@ static void server_accept_cb(socket_t listen_sock, void *arg) {
c->on_close = srv->close_cb;
c->close_arg = srv->close_arg;
c->socket_id = uasync_add_socket_t(srv->ua, cli_sock, server_conn_read_cb, server_conn_write_cb, NULL, c);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server: accepted connection fd=%d", (int)cli_sock);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server: accepted connection fd=%d", (int)cli_sock);
}
struct stcp_server *stcp_server_create(struct UASYNC *ua, uint16_t port,
@@ -347,9 +366,9 @@ struct stcp_server *stcp_server_create(struct UASYNC *ua, uint16_t port,
stcp_connect_cb connect_cb, void *arg,
stcp_close_cb close_cb, void *close_arg,
int family) {
if (!ua || !keys || !connect_cb) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "invalid args"); return NULL; }
if (!ua || !keys || !connect_cb) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "invalid args"); return NULL; }
struct stcp_server *srv = u_calloc(1, sizeof(struct stcp_server));
if (!srv) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "calloc failed"); return NULL; }
if (!srv) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "calloc failed"); return NULL; }
srv->ua = ua;
srv->my_keys = *keys;
srv->connect_cb = connect_cb;
@@ -359,7 +378,7 @@ struct stcp_server *stcp_server_create(struct UASYNC *ua, uint16_t port,
int use_family = (family == AF_INET6) ? AF_INET6 : AF_INET;
srv->listen_sock = socket(use_family, SOCK_STREAM, 0);
if (srv->listen_sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "socket failed"); u_free(srv); return NULL; }
if (srv->listen_sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "socket failed"); u_free(srv); return NULL; }
socket_set_nonblocking(srv->listen_sock);
int reuse = 1;
setsockopt(srv->listen_sock, SOL_SOCKET, SO_REUSEADDR, (const char *)&reuse, sizeof(reuse));
@@ -374,7 +393,7 @@ struct stcp_server *stcp_server_create(struct UASYNC *ua, uint16_t port,
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(port);
if (bind(srv->listen_sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "bind port=%u (v4) failed err=%d", port, socket_get_error());
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "bind port=%u (v4) failed err=%d", port, socket_get_error());
socket_close_wrapper(srv->listen_sock); u_free(srv); return NULL;
}
} else {
@@ -384,16 +403,16 @@ struct stcp_server *stcp_server_create(struct UASYNC *ua, uint16_t port,
addr.sin6_addr = in6addr_any;
addr.sin6_port = htons(port);
if (bind(srv->listen_sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "bind port=%u (v6) failed err=%d", port, socket_get_error());
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "bind port=%u (v6) failed err=%d", port, socket_get_error());
socket_close_wrapper(srv->listen_sock); u_free(srv); return NULL;
}
}
if (listen(srv->listen_sock, 16) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "listen failed err=%d", socket_get_error());
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "listen failed err=%d", socket_get_error());
socket_close_wrapper(srv->listen_sock); u_free(srv); return NULL;
}
srv->listen_id = uasync_add_socket_t(ua, srv->listen_sock, server_accept_cb, NULL, NULL, srv);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server: listening on port %u family=%s", port, use_family == AF_INET ? "v4" : "v6");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server: listening on port %u family=%s", port, use_family == AF_INET ? "v4" : "v6");
return srv;
}
@@ -402,5 +421,5 @@ void stcp_server_destroy(struct stcp_server *srv) {
if (srv->listen_id) uasync_remove_socket_t(srv->ua, srv->listen_sock);
if (srv->listen_sock != SOCKET_INVALID) socket_close_wrapper(srv->listen_sock);
u_free(srv);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_server destroyed");
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server destroyed");
}
+16
View File
@@ -16,6 +16,7 @@
#include "chat/db_sync.h"
#include "chat/chat_core.h"
#include "chat/chat_sync.h"
#include "chat/chat_headless_control.h"
#include "stcp_server.h"
#include "control_server.h"
#include "transport_layer/node_conn_direct.h"
@@ -472,6 +473,7 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] G done — ETCP sockets");
/* Phase H: chat */
chat_headless_control_destroy();
chat_sync_destroy(instance);
chat_core_destroy(instance);
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] H done — chat");
@@ -710,6 +712,20 @@ int utun_instance_init(struct UTUN_INSTANCE *instance) {
}
}
// Initialize headless chat control if configured in [chatserver]
if (instance->config->global.headless_control_bind[0] != '\0') {
char ip_str[64] = "127.0.0.1"; int port = 9999;
const char* colon = strrchr(instance->config->global.headless_control_bind, ':');
if (colon) {
size_t ip_len = (size_t)(colon - instance->config->global.headless_control_bind);
if (ip_len < sizeof(ip_str)) { memcpy(ip_str, instance->config->global.headless_control_bind, ip_len); ip_str[ip_len] = '\0'; }
port = atoi(colon + 1);
if (port <= 0 || port > 65535) port = 9999;
}
if (chat_headless_control_init(instance->ua, instance, ip_str, port) != 0)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "Failed to init headless chat control (non-fatal)");
}
// Start the main loop
instance->running = 1;