Add local UDP discovery and invitation delivery for chat clients

This commit is contained in:
evgeny committed 2026-10-18 14:46:04 +02:00
1 parent aad295c9d0
commit 28826e050c
43 files changed
+1934 -43

No files matched your search

+14
View File
@@ -153,6 +153,13 @@ utun_CORE_SOURCES = \
lwip_tcp/lwip_tcp.c \
lwip_tcp/lwip_tcp_in.c \
lwip_tcp/lwip_tcp_out.c \
local_transport/local_transport.c \
local_transport/local_transport.h \
local_transport/platform.c \
local_transport/platform.h \
local_transport/protocol.h \
chat/chat_local.c \
chat/chat_local.h \
chat/chat_event.c \
chat/chat_setting.c \
chat/chat_core.c \
@@ -327,6 +334,13 @@ libutun_a_SOURCES = \
lwip_tcp/lwip_tcp.c \
lwip_tcp/lwip_tcp_in.c \
lwip_tcp/lwip_tcp_out.c \
local_transport/local_transport.c \
local_transport/local_transport.h \
local_transport/platform.c \
local_transport/platform.h \
local_transport/protocol.h \
chat/chat_local.c \
chat/chat_local.h \
chat/chat_event.c \
chat/chat_setting.c \
chat/chat_core.c \
+2 -1
View File
@@ -54,10 +54,11 @@ void chat_event_post(struct UTUN_INSTANCE* inst, int type, const uint8_t* data,
[56]="MSG_SUBMITTED",
[57]="CPU_LOAD",
[59]="DM_DELETE_RESULT",
[61]="LOCAL_DEVICES", [62]="LOCAL_INVITE", [63]="LOCAL_DELIVERY", [64]="LOCAL_DISCOVERY_ACTIVE",
};
const char* n = (type >= 1 && type < (int)(sizeof(names) / sizeof(names[0]))) ? names[type] : "?";
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "chat_event: %s(%d) data=%d bytes", n, type, len);
if (data && len > 0) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CHAT, " ", data, len);
if (data && len > 0 && type != CHAT_EVT_LOCAL_INVITE) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CHAT, " ", data, len);
}
/* Идентификатор связывает результат с конкретной отправкой, независимо от текущего канала GUI. */
+5
View File
@@ -83,6 +83,11 @@ extern "C" {
#define CHAT_EVT_DM_DELETE_RESULT 59 /* [request:8LE][conv:8LE][generation:8LE][success:1] — после транзакции */
#define CHAT_EVT_ATTACHMENT_KEYS_CHANGED 60 /* [channel_len:1][channel][success?:1], без секретов; success — результат команды GUI */
#define CHAT_EVT_LOCAL_DEVICES 61 /* [count:2BE][node:8BE,name_len:1,name:UTF8]* */
#define CHAT_EVT_LOCAL_INVITE 62 /* [node:8BE][name_len:1][name:UTF8][invite_link:UTF8], без NUL */
#define CHAT_EVT_LOCAL_DELIVERY 63 /* [request:8BE][node:8BE][status:1], local_delivery_status, не результат JOIN */
#define CHAT_EVT_LOCAL_DISCOVERY_ACTIVE 64 /* [active:1], Android MulticastLock */
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
void chat_event_set_handler(struct UTUN_INSTANCE* inst, chat_event_handler_fn handler);
+109
View File
@@ -0,0 +1,109 @@
#include "chat_local.h"
#include "chat_event.h"
#include "invite_link.h"
#include "chat_setting.h"
#include "../utun_instance.h"
#include "../local_transport/protocol.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include <string.h>
void chat_local_peers_trampoline(void* arg) {
struct UTUN_INSTANCE* inst = arg;
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "device list request without instance"); return; }
struct local_peer peers[LOCAL_TRANSPORT_PEERS_MAX];
size_t count = local_transport_peers(inst->local_transport, peers, LOCAL_TRANSPORT_PEERS_MAX), offset = 2;
uint8_t data[2 + LOCAL_TRANSPORT_PEERS_MAX * (8 + 1 + LOCAL_TRANSPORT_NAME_MAX - 1)];
data[0] = (uint8_t)(count >> 8); data[1] = (uint8_t)count;
for (size_t i = 0; i < count; i++) {
size_t len = strlen(peers[i].name);
local_put64(data + offset, peers[i].node_id); data[offset + 8] = (uint8_t)len;
memcpy(data + offset + 9, peers[i].name, len); offset += 9 + len;
}
chat_event_post(inst, CHAT_EVT_LOCAL_DEVICES, data, (int)offset);
}
static void peers_changed(void* arg) { chat_local_peers_trampoline(arg); }
static int valid_link(const char* link, size_t len) {
struct InviteData invite;
char error[128];
if (!len || len > LOCAL_TRANSPORT_DATA_MAX || memchr(link, 0, len) || len < 7 || memcmp(link, "utun://", 7)) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation length/prefix bytes=%zu", len); return 0;
}
if (invite_link_decode(link, len, &invite, error, sizeof(error)) < 0) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation: %s", error); return 0;
}
return 1;
}
static int received(void* arg, const struct local_peer* peer, const uint8_t* link, size_t len) {
struct UTUN_INSTANCE* inst = arg;
if (!chat_setting_get_int(inst, "local_invites_enabled", 1)) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local invitation rejected peer=%016llx reason=receiving disabled",
(unsigned long long)peer->node_id); return LOCAL_DELIVERY_FORBIDDEN;
}
if (!inst->chat_event_handler || !valid_link((const char*)link, len)) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local invitation rejected peer=%016llx handler=%d",
(unsigned long long)peer->node_id, inst->chat_event_handler != NULL); return -1;
}
uint8_t data[9 + LOCAL_TRANSPORT_NAME_MAX + LOCAL_TRANSPORT_DATA_MAX];
size_t name_len = strlen(peer->name);
local_put64(data, peer->node_id); data[8] = (uint8_t)name_len;
memcpy(data + 9, peer->name, name_len); memcpy(data + 9 + name_len, link, len);
chat_event_post(inst, CHAT_EVT_LOCAL_INVITE, data, (int)(9 + name_len + len));
return 0;
}
static void delivered(void* arg, uint64_t request, uint64_t peer, int status) {
uint8_t data[17];
local_put64(data, request); local_put64(data + 8, peer); data[16] = (uint8_t)status;
chat_event_post(arg, CHAT_EVT_LOCAL_DELIVERY, data, sizeof(data));
}
static void activity_changed(void* arg, int active) {
uint8_t value = active != 0;
chat_event_post(arg, CHAT_EVT_LOCAL_DISCOVERY_ACTIVE, &value, 1);
}
static void activity(struct UTUN_INSTANCE* inst, int active, void* arg) {
(void)arg;
if (local_transport_set_active(inst->local_transport, inst->client_type != CLIENT_TYPE_MOBILE || active) < 0)
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "failed to apply discovery activity node=%016llx", (unsigned long long)inst->node_id);
}
int chat_local_start(struct UTUN_INSTANCE* inst) {
if (inst->local_transport || inst->client_type == CLIENT_TYPE_SERVER) return 0;
struct local_transport_callbacks callbacks = { peers_changed, received, delivered, activity_changed };
inst->local_transport = local_transport_create(inst->ua, inst->node_id, inst->name, &callbacks, inst);
if (!inst->local_transport) return -1;
if (chat_local_apply_settings(inst) < 0) { chat_local_stop(inst); return -1; }
utun_add_activity_cbk(inst, activity, inst);
activity(inst, inst->client_activity == CLIENT_ACTIVITY_ACTIVE, inst);
return 0;
}
int chat_local_apply_settings(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->local_transport) return 0;
int receive = chat_setting_get_int(inst, "local_invites_enabled", 1);
int broadcasts = chat_setting_get_int(inst, "local_broadcasts_enabled", 1);
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local permissions applied node=%016llx receive_invites=%d broadcast=%d",
(unsigned long long)inst->node_id, receive, broadcasts);
return local_transport_set_broadcasts(inst->local_transport, broadcasts);
}
void chat_local_stop(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->local_transport) return;
utun_remove_activity_cbk(inst, activity, inst);
local_transport_destroy(inst->local_transport); inst->local_transport = NULL;
}
void chat_local_send_trampoline(void* arg) {
struct chat_local_send_request* req = arg;
if (!req) return;
size_t len = strnlen(req->link, sizeof(req->link));
if (!req->inst) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "invitation send without instance"); u_free(req); return; }
if (!valid_link(req->link, len) || local_transport_send(req->inst->local_transport, req->peer_id,
req->request_id, req->link, len) < 0) delivered(req->inst, req->request_id, req->peer_id, 0);
u_free(req);
}
+32
View File
@@ -0,0 +1,32 @@
/* Адаптер локального транспорта: только доставка готовых invite-ссылок, без запуска JOIN. */
#ifndef CHAT_LOCAL_H
#define CHAT_LOCAL_H
#include "../local_transport/local_transport.h"
#ifdef __cplusplus
extern "C" {
#endif
struct UTUN_INSTANCE;
/* Создать транспорт для desktop/mobile, подписать активность. Повтор и server — 0; ошибка — -1. */
int chat_local_start(struct UTUN_INSTANCE* inst);
/* Снять подписку и уничтожить транспорт чат-сервиса; NULL/repeat допустимы. */
void chat_local_stop(struct UTUN_INSTANCE* inst);
/* Применить local_invites_enabled/local_broadcasts_enabled из chat_setting без рестарта.
* При отсутствии транспорта ничего не делает. Возвращает 0/-1. */
int chat_local_apply_settings(struct UTUN_INSTANCE* inst);
/* Трамплины uasync. peers получает заимствованный inst; send освобождает u_malloc request. */
void chat_local_peers_trampoline(void* inst);
struct chat_local_send_request {
struct UTUN_INSTANCE* inst; // заимствованный instance, которому передаётся команда
uint64_t peer_id; // устройство из CHAT_EVT_LOCAL_DEVICES
uint64_t request_id; // ненулевой ID UI, возвращается LOCAL_DELIVERY
char link[LOCAL_TRANSPORT_DATA_MAX + 1]; // готовая зарегистрированная ссылка, завершается NUL
};
void chat_local_send_trampoline(void* request);
#ifdef __cplusplus
}
#endif
#endif
+2
View File
@@ -7,6 +7,7 @@
#include "chat_core_priv.h"
#include "chat_event.h"
#include "chat_member.h"
#include "chat_local.h"
#include "../../lib/json_flat.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_node.h"
@@ -36,6 +37,7 @@ void chat_core_update_my_name(struct UTUN_INSTANCE* inst, const char* name) {
name, inst->my_keys.public_key, inst->my_ed25519_pubkey,
(uint64_t)ntp_time_get_seconds(inst), 0);
snprintf(inst->name, sizeof(inst->name), "%s", name);
if (inst->local_transport) local_transport_set_name(inst->local_transport, inst->name);
chat_core_update_my_member(inst);
}
+7 -1
View File
@@ -6,6 +6,7 @@
*/
#include "chat_setting.h"
#include "chat_local.h"
#include "../call/call_video_rate.h"
#include "../utun_instance.h"
#include "../../lib/debug_config.h"
@@ -73,6 +74,8 @@ static const struct chat_setting_def g_setting_defs[] = {
{"member_files_path", CHAT_SETTING_STRING, 0, 0, 0},
{"member_files_user_quota_mb", CHAT_SETTING_INT, 1024, 1, 1048576},
{"member_files_upload_ttl_sec", CHAT_SETTING_INT, 3600, 60, 604800},
{"local_invites_enabled", CHAT_SETTING_BOOL, 1, 0, 1},
{"local_broadcasts_enabled", CHAT_SETTING_BOOL, 1, 0, 1},
};
#define CHAT_SETTING_COUNT (sizeof(g_setting_defs) / sizeof(g_setting_defs[0]))
@@ -204,7 +207,10 @@ static struct chat_setting_state* cs_state(struct UTUN_INSTANCE* inst) {
int chat_setting_set(struct UTUN_INSTANCE* inst, const char* name, const char* value) {
struct chat_setting_state* s = cs_state(inst);
if (!s) return -2;
return chat_setting_set_state(s, name, value);
int result = chat_setting_set_state(s, name, value);
if (!result && (!strcmp(name, "local_invites_enabled") || !strcmp(name, "local_broadcasts_enabled")))
return chat_local_apply_settings(inst);
return result;
}
const char* chat_setting_get(struct UTUN_INSTANCE* inst, const char* name) {
+506
View File
@@ -0,0 +1,506 @@
#include "local_transport.h"
#include "platform.h"
#include "protocol.h"
#include "../../lib/u_async.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include <openssl/rand.h>
#include <string.h>
#define ANNOUNCE_TB 100000
#define EXPIRE_TB 300000
#define TICK_TB 10000
#define ENDPOINT_MAX 8
#define PENDING_MAX 8
#define SEEN_MAX 128
struct local_socket {
struct local_socket* next;
struct local_transport* owner;
struct local_interface interface;
socket_t fd;
void* handle;
int present;
};
struct local_endpoint { struct sockaddr_storage address; uint32_t index; uint64_t seen; };
struct peer_entry {
struct local_peer peer;
uint64_t epoch;
struct local_endpoint endpoints[ENDPOINT_MAX];
size_t count;
};
struct pending_entry {
uint64_t request, peer_id, epoch;
size_t size;
unsigned attempts;
uint8_t data[LOCAL_TRANSPORT_DATA_MAX];
};
struct seen_entry { uint64_t peer, epoch, request, time; };
struct local_transport {
struct UASYNC* ua;
uint64_t node_id, epoch, next_announce;
char name[LOCAL_TRANSPORT_NAME_MAX];
struct local_transport_callbacks callbacks;
void* arg;
void* timer;
struct local_socket* sockets;
uint32_t joined6[LOCAL_INTERFACE_MAX];
size_t joined6_count;
struct peer_entry peers[LOCAL_TRANSPORT_PEERS_MAX];
size_t peer_count;
struct pending_entry pending[PENDING_MAX];
struct seen_entry seen[SEEN_MAX];
unsigned seen_next;
int active, broadcasts;
};
static void socket_read(socket_t fd, void* arg);
static void tick(void* arg);
static void peers_changed(struct local_transport* t) {
if (t->callbacks.peers_changed) t->callbacks.peers_changed(t->arg);
}
static struct peer_entry* find_peer(struct local_transport* t, uint64_t node_id) {
for (size_t i = 0; i < t->peer_count; i++) if (t->peers[i].peer.node_id == node_id) return &t->peers[i];
return NULL;
}
static int same_address(const struct sockaddr_storage* a, const struct sockaddr_storage* b) {
if (a->ss_family != b->ss_family || ss_get_port(a) != ss_get_port(b)) return 0;
if (a->ss_family == AF_INET)
return ((const struct sockaddr_in*)a)->sin_addr.s_addr == ((const struct sockaddr_in*)b)->sin_addr.s_addr;
return !memcmp(&((const struct sockaddr_in6*)a)->sin6_addr, &((const struct sockaddr_in6*)b)->sin6_addr, 16);
}
static struct local_socket* find_socket(struct local_transport* t, const struct sockaddr_storage* source, uint32_t index) {
for (struct local_socket* s = t->sockets; s; s = s->next)
if (s->interface.index == index && index && local_on_interface(&s->interface, source)) return s;
return NULL;
}
static int send_packet(struct local_socket* s, const struct sockaddr_storage* address, uint8_t type,
uint64_t peer, uint64_t epoch, uint64_t request, const void* payload, size_t len) {
struct local_transport* t = s->owner;
uint8_t data[LOCAL_PACKET_MAX];
memcpy(data, LOCAL_MAGIC, 16); data[16] = 1; data[17] = type;
data[18] = (uint8_t)(len >> 8); data[19] = (uint8_t)len;
local_put64(data + 20, t->node_id); local_put64(data + 28, t->epoch);
local_put64(data + 36, peer); local_put64(data + 44, epoch); local_put64(data + 52, request);
if (len) memcpy(data + LOCAL_HEADER_SIZE, payload, len);
socklen_t address_len = address->ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);
ssize_t result = socket_sendto(s->fd, data, LOCAL_HEADER_SIZE + len, (const struct sockaddr*)address, address_len);
if (result != (ssize_t)(LOCAL_HEADER_SIZE + len)) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "send failed type=%u interface=%u destination=%s bytes=%zu result=%zd error=%s",
type, s->interface.index, sockaddr_storage_to_str(address).str, len, result, socket_strerror(socket_get_error()));
return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "sent type=%u interface=%u destination=%s peer=%016llx request=%llu bytes=%zu",
type, s->interface.index, sockaddr_storage_to_str(address).str, (unsigned long long)peer, (unsigned long long)request, len);
return 0;
}
static void announce(struct local_socket* s, const struct sockaddr_storage* address) {
send_packet(s, address, LOCAL_ANNOUNCE, 0, 0, 0, s->owner->name, strlen(s->owner->name));
}
static void announce_all(struct local_transport* t) {
if (!t->active || !t->broadcasts) return;
for (struct local_socket* s = t->sockets; s; s = s->next)
if (s->interface.index) announce(s, &s->interface.broadcast);
}
static void delivery(struct local_transport* t, struct pending_entry* p, int status) {
uint64_t request = p->request, peer = p->peer_id;
memset(p, 0, sizeof(*p));
if (status == LOCAL_DELIVERY_RECEIVED) DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "message delivered peer=%016llx request=%llu",
(unsigned long long)peer, (unsigned long long)request);
else if (status == LOCAL_DELIVERY_FORBIDDEN) DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY,
"message receipt forbidden by peer=%016llx request=%llu; enable receiving in recipient settings",
(unsigned long long)peer, (unsigned long long)request);
else DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "message delivery failed peer=%016llx request=%llu",
(unsigned long long)peer, (unsigned long long)request);
if (t->callbacks.delivery) t->callbacks.delivery(t->arg, request, peer, status);
}
static void expire(struct local_transport* t, uint64_t now) {
int changed = 0;
for (size_t i = 0; i < t->peer_count;) {
struct peer_entry* p = &t->peers[i];
for (size_t j = 0; j < p->count;) {
struct local_endpoint* e = &p->endpoints[j];
if (now - e->seen < EXPIRE_TB && find_socket(t, &e->address, e->index)) { j++; continue; }
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "endpoint expired peer=%016llx interface=%u address=%s age_ms=%llu",
(unsigned long long)p->peer.node_id, e->index, sockaddr_storage_to_str(&e->address).str,
(unsigned long long)((now - e->seen) / 10));
p->endpoints[j] = p->endpoints[--p->count];
}
if (p->count) { i++; continue; }
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "device expired peer=%016llx name=%s", (unsigned long long)p->peer.node_id, p->peer.name);
t->peers[i] = t->peers[--t->peer_count]; changed = 1;
}
if (changed) peers_changed(t);
}
static void receive_packet(struct local_socket* s, struct sockaddr_storage* source, const uint8_t* data, size_t size) {
struct local_transport* t = s->owner;
if (size < 16 || memcmp(data, LOCAL_MAGIC, 16)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "foreign UDP ignored interface=%u source=%s bytes=%zu",
s->interface.index, sockaddr_storage_to_str(source).str, size); return;
}
if (size < LOCAL_HEADER_SIZE || size > LOCAL_PACKET_MAX || data[16] != 1 ||
(size_t)((unsigned)data[18] << 8 | data[19]) != size - LOCAL_HEADER_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local header source=%s bytes=%zu", sockaddr_storage_to_str(source).str, size); return;
}
uint64_t peer_id = local_get64(data + 20), epoch = local_get64(data + 28), request = local_get64(data + 52);
uint64_t destination = local_get64(data + 36), destination_epoch = local_get64(data + 44);
if (!peer_id || !epoch) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "zero sender identity in local packet"); return; }
if (peer_id == t->node_id) return;
uint8_t type = data[17];
const uint8_t* payload = data + LOCAL_HEADER_SIZE;
size_t len = size - LOCAL_HEADER_SIZE;
uint64_t now = get_time_tb();
struct peer_entry* p = find_peer(t, peer_id);
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "received type=%u peer=%016llx interface=%u source=%s request=%llu bytes=%zu",
type, (unsigned long long)peer_id, s->interface.index, sockaddr_storage_to_str(source).str, (unsigned long long)request, len);
if (type == LOCAL_ANNOUNCE) {
if (destination || destination_epoch || request || !local_name_valid(payload, len)) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid announcement peer=%016llx name_bytes=%zu", (unsigned long long)peer_id, len); return;
}
int changed = !p;
if (!p) {
if (t->peer_count == LOCAL_TRANSPORT_PEERS_MAX) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "device table full peer=%016llx", (unsigned long long)peer_id); return;
}
p = &t->peers[t->peer_count++]; memset(p, 0, sizeof(*p)); p->peer.node_id = peer_id;
}
if (p->epoch != epoch) { p->epoch = epoch; p->count = 0; changed = 1; }
if (strlen(p->peer.name) != len || memcmp(p->peer.name, payload, len)) {
memcpy(p->peer.name, payload, len); p->peer.name[len] = 0; changed = 1;
}
size_t j;
for (j = 0; j < p->count; j++)
if (p->endpoints[j].index == s->interface.index && same_address(&p->endpoints[j].address, source)) break;
int new_endpoint = j == p->count;
if (new_endpoint) {
if (j == ENDPOINT_MAX) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "endpoint limit peer=%016llx", (unsigned long long)peer_id); return;
}
p->count++;
}
p->endpoints[j] = (struct local_endpoint){ .address = *source, .index = s->interface.index, .seen = now };
if (changed) {
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "device discovered/updated peer=%016llx name=%s endpoints=%zu",
(unsigned long long)peer_id, p->peer.name, p->count); peers_changed(t);
}
if (new_endpoint) announce(s, source);
return;
}
if ((type != LOCAL_MESSAGE && type != LOCAL_ACK) || !request || !p || p->epoch != epoch ||
destination != t->node_id || destination_epoch != t->epoch) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local packet rejected type=%u peer=%016llx reason=type/identity/destination",
type, (unsigned long long)peer_id); return;
}
size_t j;
for (j = 0; j < p->count; j++)
if (p->endpoints[j].index == s->interface.index && same_address(&p->endpoints[j].address, source) &&
now - p->endpoints[j].seen < EXPIRE_TB) break;
if (j == p->count) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "unknown/expired message endpoint peer=%016llx", (unsigned long long)peer_id); return; }
if (type == LOCAL_ACK) {
if (len != 1 || payload[0] > LOCAL_DELIVERY_FORBIDDEN) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid ACK payload"); return; }
for (int i = 0; i < PENDING_MAX; i++) {
struct pending_entry* pending = &t->pending[i];
if (pending->request == request && pending->peer_id == peer_id && pending->epoch == epoch) { delivery(t, pending, payload[0]); return; }
}
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "late ACK peer=%016llx request=%llu", (unsigned long long)peer_id, (unsigned long long)request);
return;
}
if (!len) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "empty local message peer=%016llx", (unsigned long long)peer_id); return; }
uint8_t status = LOCAL_DELIVERY_FAILED;
for (int i = 0; i < SEEN_MAX; i++) {
const struct seen_entry* entry = &t->seen[i];
if (entry->peer == peer_id && entry->epoch == epoch && entry->request == request && now - entry->time < EXPIRE_TB) {
status = LOCAL_DELIVERY_RECEIVED; break;
}
}
if (status != LOCAL_DELIVERY_RECEIVED && t->callbacks.received) {
int result = t->callbacks.received(t->arg, &p->peer, payload, len);
if (!result) {
status = LOCAL_DELIVERY_RECEIVED;
t->seen[t->seen_next++ % SEEN_MAX] = (struct seen_entry){ peer_id, epoch, request, now };
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "message accepted peer=%016llx request=%llu bytes=%zu",
(unsigned long long)peer_id, (unsigned long long)request, len);
} else if (result == LOCAL_DELIVERY_FORBIDDEN) status = LOCAL_DELIVERY_FORBIDDEN;
}
if (status == LOCAL_DELIVERY_FAILED)
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "application rejected local message peer=%016llx", (unsigned long long)peer_id);
send_packet(s, source, LOCAL_ACK, peer_id, epoch, request, &status, 1);
}
static void socket_read(socket_t fd, void* arg) {
struct local_socket* socket = arg;
struct local_transport* t = socket->owner;
/* Ограничение burst не позволяет локальному трафику вытеснить остальные callbacks ядра. */
for (int i = 0; i < 32; i++) {
uint8_t data[LOCAL_PACKET_MAX];
struct sockaddr_storage source;
uint32_t index;
ssize_t size = local_receive(fd, data, sizeof(data), &source, &index);
if (size == -1) {
int error = socket_get_error();
if (error != ERR_WOULDBLOCK && error != ERR_AGAIN && error != ERR_INTR)
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "receive failed: %s", socket_strerror(error));
return;
}
if (size == -2) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "truncated local datagram dropped"); continue; }
/* Unicast-сокет уже привязан к адресу интерфейса. При обмене двух процессов на одном
* хосте ОС может указать loopback как ingress, но IPv6 scope нужен от локальной привязки. */
if (socket->interface.index) index = socket->interface.index;
struct local_socket* s = find_socket(t, &source, index);
if (!s) {
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "non-local source ignored interface=%u source=%s", index, sockaddr_storage_to_str(&source).str);
continue;
}
if (source.ss_family == AF_INET6) ((struct sockaddr_in6*)&source)->sin6_scope_id = index;
receive_packet(s, &source, data, (size_t)size);
}
}
static void close_socket(struct local_socket* s) {
if (s->handle) uasync_remove_socket(s->owner->ua, s->handle);
socket_close_wrapper(s->fd); u_free(s);
}
static struct local_socket* open_socket(struct local_transport* t, const struct local_interface* interface) {
int family = interface->address.ss_family, enabled = 1;
struct local_socket* s = u_calloc(1, sizeof(*s));
if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "socket record allocation failed"); return NULL; }
s->owner = t; s->interface = *interface; s->present = 1; s->fd = socket_create_udp(family);
if (s->fd == SOCKET_INVALID) goto error;
/* Общий порт используется только broadcast/multicast-приёмниками; unicast уходит на уникальные ephemeral-порты. */
if ((!interface->index && socket_set_reuseaddr(s->fd, 1)) || socket_set_nonblocking(s->fd) ||
local_receive_options(s->fd, family)) goto error;
if (family == AF_INET && setsockopt(s->fd, SOL_SOCKET, SO_BROADCAST, (const char*)&enabled, sizeof(enabled))) goto error;
if (family == AF_INET && interface->index) {
int ttl = 1;
if (setsockopt(s->fd, IPPROTO_IP, IP_TTL, (const char*)&ttl, sizeof(ttl))) goto error;
#ifdef IP_UNICAST_IF
/* bind(IP) фиксирует источник, но не egress при перекрывающихся подсетях/VPN. */
uint32_t index = htonl(interface->index);
if (setsockopt(s->fd, IPPROTO_IP, IP_UNICAST_IF, (const char*)&index, sizeof(index))) goto error;
#endif
}
if (family == AF_INET6) {
int hops = 1;
if (setsockopt(s->fd, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&enabled, sizeof(enabled))) goto error;
if (interface->index && (setsockopt(s->fd, IPPROTO_IPV6, IPV6_MULTICAST_IF,
(const char*)&interface->index, sizeof(interface->index)) ||
setsockopt(s->fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, (const char*)&hops, sizeof(hops)))) goto error;
}
struct sockaddr_storage address = interface->address;
if (!interface->index) {
if (family == AF_INET) ((struct sockaddr_in*)&address)->sin_port = htons(LOCAL_TRANSPORT_PORT);
else ((struct sockaddr_in6*)&address)->sin6_port = htons(LOCAL_TRANSPORT_PORT);
}
socklen_t len = family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);
if (bind(s->fd, (struct sockaddr*)&address, len)) goto error;
s->handle = uasync_add_socket_t(t->ua, s->fd, socket_read, NULL, NULL, "local_transport", s);
if (!s->handle) goto error;
s->next = t->sockets; t->sockets = s;
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local UDP socket opened family=%d interface=%u address=%s port=%u",
family, interface->index, sockaddr_storage_to_str(&address).str, socket_get_bound_port(s->fd));
return s;
error:
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local UDP socket failed family=%d interface=%u: %s",
family, interface->index, socket_strerror(socket_get_error()));
if (s->fd != SOCKET_INVALID) socket_close_wrapper(s->fd);
u_free(s); return NULL;
}
static int membership(struct local_socket* receiver, uint32_t index, int join) {
struct ipv6_mreq group = {0};
inet_pton(AF_INET6, LOCAL_MULTICAST6, &group.ipv6mr_multiaddr); group.ipv6mr_interface = index;
if (setsockopt(receiver->fd, IPPROTO_IPV6, join ? IPV6_JOIN_GROUP : IPV6_LEAVE_GROUP, (const char*)&group, sizeof(group))) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "multicast %s failed interface=%u: %s", join ? "join" : "leave", index,
socket_strerror(socket_get_error()));
return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "multicast %s interface=%u group=%s", join ? "joined" : "left", index, LOCAL_MULTICAST6);
return 0;
}
static void reconcile(struct local_transport* t) {
struct local_interface interfaces[LOCAL_INTERFACE_MAX];
int count = local_interfaces(interfaces, LOCAL_INTERFACE_MAX);
if (count < 0) return;
struct local_socket* receiver6 = NULL;
int receiver4_found = 0;
for (struct local_socket* s = t->sockets; s; s = s->next) {
s->present = !s->interface.index;
if (!s->interface.index && s->interface.address.ss_family == AF_INET6) receiver6 = s;
if (!s->interface.index && s->interface.address.ss_family == AF_INET) receiver4_found = 1;
}
if (!receiver4_found) {
struct local_interface listener = {0}; listener.address.ss_family = AF_INET;
open_socket(t, &listener);
}
if (!receiver6) {
t->joined6_count = 0;
struct local_interface listener = {0}; listener.address.ss_family = AF_INET6;
receiver6 = open_socket(t, &listener);
}
for (int i = 0; i < count; i++) {
struct local_socket* found = NULL;
for (struct local_socket* s = t->sockets; s; s = s->next)
if (s->interface.index == interfaces[i].index && same_address(&s->interface.address, &interfaces[i].address)) { found = s; break; }
if (found) {
found->present = 1; found->interface = interfaces[i];
continue;
}
open_socket(t, &interfaces[i]);
}
struct local_socket** link = &t->sockets;
while (*link) {
struct local_socket* s = *link;
if (s->present) { link = &s->next; continue; }
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local UDP socket removed interface=%u address=%s",
s->interface.index, sockaddr_storage_to_str(&s->interface.address).str);
*link = s->next; close_socket(s);
}
if (!receiver6) return;
for (size_t i = 0; i < t->joined6_count;) {
int present = 0;
for (struct local_socket* s = t->sockets; s; s = s->next)
if (s->interface.index == t->joined6[i] && s->interface.address.ss_family == AF_INET6) present = 1;
if (present) { i++; continue; }
membership(receiver6, t->joined6[i], 0); t->joined6[i] = t->joined6[--t->joined6_count];
}
for (struct local_socket* s = t->sockets; s; s = s->next) {
if (!s->interface.index || s->interface.address.ss_family != AF_INET6) continue;
size_t i;
for (i = 0; i < t->joined6_count; i++) if (t->joined6[i] == s->interface.index) break;
if (i == t->joined6_count && !membership(receiver6, s->interface.index, 1)) t->joined6[t->joined6_count++] = s->interface.index;
}
}
static void try_send(struct local_transport* t, struct pending_entry* p) {
struct peer_entry* peer = find_peer(t, p->peer_id);
if (!peer || peer->epoch != p->epoch || !peer->count || p->attempts >= 3) { delivery(t, p, 0); return; }
p->attempts++;
int sent = 0;
for (size_t i = 0; i < peer->count; i++) {
struct local_endpoint* e = &peer->endpoints[i];
struct local_socket* s = find_socket(t, &e->address, e->index);
if (s) {
/* Получатель узнаёт отправителя и без его broadcast-объявлений. */
announce(s, &e->address);
if (!send_packet(s, &e->address, LOCAL_MESSAGE, p->peer_id, p->epoch, p->request, p->data, p->size)) sent = 1;
}
}
if (!sent) DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "no usable endpoint peer=%016llx attempt=%u",
(unsigned long long)p->peer_id, p->attempts);
}
static void tick(void* arg) {
struct local_transport* t = arg;
t->timer = NULL;
uint64_t now = get_time_tb();
if (now >= t->next_announce) {
reconcile(t); t->next_announce = now + ANNOUNCE_TB;
announce_all(t);
}
expire(t, now);
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) try_send(t, &t->pending[i]);
t->timer = uasync_set_timeout(t->ua, TICK_TB, t, tick, "local_transport");
if (!t->timer) {
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local transport timer allocation failed; disabling discovery");
local_transport_set_active(t, 0);
}
}
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
const struct local_transport_callbacks* callbacks, void* arg) {
if (!ua || !node_id || !name || !callbacks || !local_name_valid((const uint8_t*)name, strlen(name))) {
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "invalid local transport configuration"); return NULL;
}
struct local_transport* t = u_calloc(1, sizeof(*t));
if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local transport allocation failed"); return NULL; }
if (RAND_bytes((uint8_t*)&t->epoch, sizeof(t->epoch)) != 1 || !t->epoch) {
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local transport epoch generation failed"); u_free(t); return NULL;
}
t->ua = ua; t->node_id = node_id; t->callbacks = *callbacks; t->arg = arg;
t->broadcasts = 1;
memcpy(t->name, name, strlen(name) + 1);
return t;
}
int local_transport_set_active(struct local_transport* t, int active) {
if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "activity update without transport"); return -1; }
if (t->active == !!active) return 0;
t->active = !!active;
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local discovery %s node=%016llx port=%u",
active ? "active" : "paused", (unsigned long long)t->node_id, LOCAL_TRANSPORT_PORT);
if (t->callbacks.activity_changed) t->callbacks.activity_changed(t->arg, t->active);
if (active) { t->next_announce = 0; tick(t); return t->active ? 0 : -1; }
if (t->timer) { uasync_cancel_timeout(t->ua, t->timer); t->timer = NULL; }
while (t->sockets) { struct local_socket* s = t->sockets; t->sockets = s->next; close_socket(s); }
t->joined6_count = 0;
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) delivery(t, &t->pending[i], 0);
if (t->peer_count) { t->peer_count = 0; peers_changed(t); }
return 0;
}
void local_transport_destroy(struct local_transport* t) {
if (!t) return;
local_transport_set_active(t, 0); u_free(t);
}
int local_transport_set_broadcasts(struct local_transport* t, int enabled) {
if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "broadcast update without transport"); return -1; }
if (t->broadcasts == !!enabled) return 0;
t->broadcasts = !!enabled;
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local broadcast announcements %s node=%016llx",
enabled ? "enabled" : "disabled", (unsigned long long)t->node_id);
if (enabled) announce_all(t);
return 0;
}
int local_transport_set_name(struct local_transport* t, const char* name) {
if (!t || !name || !local_name_valid((const uint8_t*)name, strlen(name))) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid device name"); return -1;
}
if (!strcmp(t->name, name)) return 0;
memcpy(t->name, name, strlen(name) + 1);
announce_all(t);
return 0;
}
size_t local_transport_peers(struct local_transport* t, struct local_peer* out, size_t capacity) {
if (!t || !out) return 0;
expire(t, get_time_tb());
size_t count = t->peer_count < capacity ? t->peer_count : capacity;
for (size_t i = 0; i < count; i++) out[i] = t->peers[i].peer;
return count;
}
int local_transport_send(struct local_transport* t, uint64_t peer_id, uint64_t request_id, const void* data, size_t len) {
if (!t || !t->active || !request_id || !data || !len || len > LOCAL_TRANSPORT_DATA_MAX) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local send peer=%016llx request=%llu bytes=%zu",
(unsigned long long)peer_id, (unsigned long long)request_id, len); return -1;
}
expire(t, get_time_tb());
struct peer_entry* peer = find_peer(t, peer_id);
if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local device unavailable peer=%016llx", (unsigned long long)peer_id); return -1; }
struct pending_entry* pending = NULL;
for (int i = 0; i < PENDING_MAX; i++) {
if (t->pending[i].request == request_id) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "duplicate local request=%llu", (unsigned long long)request_id); return -1;
}
if (!t->pending[i].request) pending = &t->pending[i];
}
if (!pending) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local delivery queue full"); return -1; }
pending->request = request_id; pending->peer_id = peer_id; pending->epoch = peer->epoch;
pending->size = len; pending->attempts = 0; memcpy(pending->data, data, len);
try_send(t, pending);
return 0;
}
+63
View File
@@ -0,0 +1,63 @@
/* Открытый UDP-транспорт локальной сети. Не создаёт ETCP-сокетов и не подключает пиры.
* Все вызовы и callbacks — в потоке ua. destroy/set_active — вне callbacks транспорта.
* Данные callback заимствованы до возврата; received возвращает 0 после принятия сообщения.
* delivery подтверждает приём сообщения приложением, а не выполнение его содержимого. */
#ifndef LOCAL_TRANSPORT_H
#define LOCAL_TRANSPORT_H
#include <stddef.h>
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
#define LOCAL_TRANSPORT_PORT 45731
#define LOCAL_TRANSPORT_NAME_MAX 128
#define LOCAL_TRANSPORT_DATA_MAX 1100
#define LOCAL_TRANSPORT_PEERS_MAX 128
struct UASYNC;
struct local_transport;
/* Один байт в ACK и результате доставки; FORBIDDEN означает явный запрет приёма у получателя. */
enum local_delivery_status {
LOCAL_DELIVERY_FAILED = 0,
LOCAL_DELIVERY_RECEIVED = 1,
LOCAL_DELIVERY_FORBIDDEN = 2
};
struct local_peer {
uint64_t node_id; // идентификатор из объявления, не подтверждает подлинность
char name[LOCAL_TRANSPORT_NAME_MAX]; // UTF-8 ник, завершается NUL
};
struct local_transport_callbacks {
void (*peers_changed)(void* arg); // таблица изменилась; получить копию через peers()
int (*received)(void* arg, const struct local_peer* peer, const uint8_t* data, size_t len); // 0 = принято, FORBIDDEN = запрет, -1 = ошибка
void (*delivery)(void* arg, uint64_t request_id, uint64_t peer_id, int status); // один результат: enum local_delivery_status
void (*activity_changed)(void* arg, int active); // сокеты включаются/выключаются, в том числе при ошибке таймера
};
/* Создать выключенный транспорт: ua заимствован, name/callbacks копируются, arg передаётся callbacks.
* node_id != 0, name — 1..127 байтов UTF-8. Возвращает владельческий handle либо NULL с ошибкой в логе. */
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
const struct local_transport_callbacks* callbacks, void* arg);
/* Закрыть сокеты, отменить таймеры/доставки и освободить handle. NULL допустим. */
void local_transport_destroy(struct local_transport* transport);
/* ACTIVE включает сокеты и таймер; выключение отменяет доставки, очищает таблицу и не будит standby.
* Повтор допустим. Возвращает 0/-1; отсутствие локальной сети не является ошибкой активации. */
int local_transport_set_active(struct local_transport* transport, int active);
/* Разрешить периодические broadcast/multicast-объявления (по умолчанию включены).
* Unicast, приём и таблица устройств сохраняются; включение сразу отправляет объявление. Возвращает 0/-1. */
int local_transport_set_broadcasts(struct local_transport* transport, int enabled);
/* Обновить ник (те же ограничения, что у create), сразу отправить объявление. Возвращает 0/-1. */
int local_transport_set_name(struct local_transport* transport, const char* name);
/* Скопировать не более capacity живых устройств в out; возвращает количество, 0 для NULL transport/out. */
size_t local_transport_peers(struct local_transport* transport, struct local_peer* out, size_t capacity);
/* Скопировать 1..DATA_MAX байтов data для peer_id из текущей таблицы. request_id != 0, уникален среди pending.
* 0 = запрос принят, -1 = callback не будет. До 3 попыток на всех endpoints; дубли получатель не выдаёт приложению. */
int local_transport_send(struct local_transport* transport, uint64_t peer_id, uint64_t request_id,
const void* data, size_t len);
#ifdef __cplusplus
}
#endif
#endif
+178
View File
@@ -0,0 +1,178 @@
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include "platform.h"
#include "local_transport.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include <string.h>
#ifdef _WIN32
#include <iphlpapi.h>
#include <mswsock.h>
#else
#include <ifaddrs.h>
#include <net/if.h>
#ifdef __FreeBSD__
#include <net/if_dl.h>
#endif
#endif
static int add_interface(struct local_interface* out, size_t capacity, int count, uint32_t index,
const struct sockaddr* address, uint32_t mask) {
if (!index || !address) return count;
int family = address->sa_family;
if (family != AF_INET && family != AF_INET6) return count;
if (family == AF_INET6 && !IN6_IS_ADDR_LINKLOCAL(&((const struct sockaddr_in6*)address)->sin6_addr)) return count;
if (family == AF_INET && (!mask || mask == UINT32_MAX)) return count;
struct local_interface entry = {0};
entry.index = index; entry.mask4 = mask;
if (family == AF_INET) {
struct sockaddr_in* addr = (struct sockaddr_in*)&entry.address;
*addr = *(const struct sockaddr_in*)address; addr->sin_port = 0;
struct sockaddr_in* dest = (struct sockaddr_in*)&entry.broadcast;
dest->sin_family = AF_INET; dest->sin_port = htons(LOCAL_TRANSPORT_PORT);
dest->sin_addr.s_addr = addr->sin_addr.s_addr | ~mask;
} else {
struct sockaddr_in6* addr = (struct sockaddr_in6*)&entry.address;
*addr = *(const struct sockaddr_in6*)address; addr->sin6_port = 0; addr->sin6_scope_id = index;
struct sockaddr_in6* dest = (struct sockaddr_in6*)&entry.broadcast;
dest->sin6_family = AF_INET6; dest->sin6_port = htons(LOCAL_TRANSPORT_PORT); dest->sin6_scope_id = index;
inet_pton(AF_INET6, LOCAL_MULTICAST6, &dest->sin6_addr);
}
for (int i = 0; i < count; i++) {
if (out[i].index != index || out[i].address.ss_family != family) continue;
if (family == AF_INET6 || ((struct sockaddr_in*)&out[i].address)->sin_addr.s_addr ==
((struct sockaddr_in*)&entry.address)->sin_addr.s_addr) return count;
}
if ((size_t)count == capacity) {
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "interface limit reached count=%d index=%u family=%d", count, index, family);
return count;
}
out[count] = entry;
return count + 1;
}
int local_interfaces(struct local_interface* out, size_t capacity) {
int count = 0;
#ifdef _WIN32
ULONG size = 16384, result;
IP_ADAPTER_ADDRESSES* adapters = NULL;
for (int attempt = 0; attempt < 3; attempt++) {
u_free(adapters); adapters = u_malloc(size);
if (!adapters) { DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "adapter allocation failed bytes=%lu", size); return -1; }
result = GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST |
GAA_FLAG_SKIP_DNS_SERVER, NULL, adapters, &size);
if (result != ERROR_BUFFER_OVERFLOW) break;
}
if (result != NO_ERROR && result != ERROR_NO_DATA) {
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "GetAdaptersAddresses failed error=%lu", result); u_free(adapters); return -1;
}
for (IP_ADAPTER_ADDRESSES* a = result == NO_ERROR ? adapters : NULL; a; a = a->Next) {
if (a->OperStatus != IfOperStatusUp || a->IfType == IF_TYPE_SOFTWARE_LOOPBACK || a->IfType == IF_TYPE_PPP ||
a->IfType == IF_TYPE_TUNNEL) continue;
for (IP_ADAPTER_UNICAST_ADDRESS* p = a->FirstUnicastAddress; p; p = p->Next) {
struct sockaddr* address = p->Address.lpSockaddr;
uint32_t mask = p->OnLinkPrefixLength && p->OnLinkPrefixLength < 32 ?
htonl(UINT32_MAX << (32 - p->OnLinkPrefixLength)) : 0;
count = add_interface(out, capacity, count, address->sa_family == AF_INET ? a->IfIndex : a->Ipv6IfIndex, address, mask);
}
}
u_free(adapters);
#else
struct ifaddrs* addresses = NULL;
if (getifaddrs(&addresses)) {
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "getifaddrs failed: %s", strerror(errno)); return -1;
}
for (struct ifaddrs* a = addresses; a; a = a->ifa_next) {
if (!a->ifa_addr || !(a->ifa_flags & IFF_UP) || (a->ifa_flags & (IFF_LOOPBACK | IFF_POINTOPOINT))) continue;
int family = a->ifa_addr->sa_family;
if (family == AF_INET && !(a->ifa_flags & IFF_BROADCAST)) continue;
if (family == AF_INET6 && !(a->ifa_flags & IFF_MULTICAST)) continue;
uint32_t mask = family == AF_INET && a->ifa_netmask ? ((struct sockaddr_in*)a->ifa_netmask)->sin_addr.s_addr : 0;
count = add_interface(out, capacity, count, if_nametoindex(a->ifa_name), a->ifa_addr, mask);
}
freeifaddrs(addresses);
#endif
return count;
}
int local_receive_options(socket_t fd, int family) {
int enabled = 1, option, level;
if (family == AF_INET6) {
level = IPPROTO_IPV6;
#ifdef _WIN32
option = IPV6_PKTINFO;
#else
option = IPV6_RECVPKTINFO;
#endif
}
else {
level = IPPROTO_IP;
#ifdef __FreeBSD__
option = IP_RECVIF;
#else
option = IP_PKTINFO;
#endif
}
if (!setsockopt(fd, level, option, (const char*)&enabled, sizeof(enabled))) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "receive interface option failed family=%d: %s",
family, socket_strerror(socket_get_error()));
return -1;
}
ssize_t local_receive(socket_t fd, void* data, size_t capacity, struct sockaddr_storage* source, uint32_t* index) {
*index = 0; memset(source, 0, sizeof(*source));
#ifdef _WIN32
GUID guid = WSAID_WSARECVMSG;
LPFN_WSARECVMSG receive = NULL;
DWORD bytes = 0;
if (WSAIoctl(fd, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), &receive, sizeof(receive), &bytes, NULL, NULL)) return -1;
WSABUF buffer = { (ULONG)capacity, data };
union { WSACMSGHDR alignment; char bytes[256]; } control;
WSAMSG message = {0};
message.name = (struct sockaddr*)source; message.namelen = sizeof(*source);
message.lpBuffers = &buffer; message.dwBufferCount = 1;
message.Control.buf = control.bytes; message.Control.len = sizeof(control.bytes);
if (receive(fd, &message, &bytes, NULL, NULL)) return socket_get_error() == WSAEMSGSIZE ? -2 : -1;
if (message.dwFlags & (MSG_TRUNC | MSG_CTRUNC)) return -2;
for (WSACMSGHDR* c = WSA_CMSG_FIRSTHDR(&message); c; c = WSA_CMSG_NXTHDR(&message, c)) {
if (c->cmsg_level == IPPROTO_IP && c->cmsg_type == IP_PKTINFO)
*index = ((IN_PKTINFO*)WSA_CMSG_DATA(c))->ipi_ifindex;
if (c->cmsg_level == IPPROTO_IPV6 && c->cmsg_type == IPV6_PKTINFO)
*index = ((IN6_PKTINFO*)WSA_CMSG_DATA(c))->ipi6_ifindex;
}
return (ssize_t)bytes;
#else
struct iovec buffer = { data, capacity };
union { struct cmsghdr alignment; char bytes[256]; } control;
struct msghdr message = {0};
message.msg_name = source; message.msg_namelen = sizeof(*source);
message.msg_iov = &buffer; message.msg_iovlen = 1;
message.msg_control = control.bytes; message.msg_controllen = sizeof(control.bytes);
ssize_t size = recvmsg(fd, &message, 0);
if (size < 0) return -1;
if (message.msg_flags & (MSG_TRUNC | MSG_CTRUNC)) return -2;
for (struct cmsghdr* c = CMSG_FIRSTHDR(&message); c; c = CMSG_NXTHDR(&message, c)) {
#ifdef __FreeBSD__
if (c->cmsg_level == IPPROTO_IP && c->cmsg_type == IP_RECVIF)
*index = ((struct sockaddr_dl*)CMSG_DATA(c))->sdl_index;
#else
if (c->cmsg_level == IPPROTO_IP && c->cmsg_type == IP_PKTINFO)
*index = ((struct in_pktinfo*)CMSG_DATA(c))->ipi_ifindex;
#endif
if (c->cmsg_level == IPPROTO_IPV6 && c->cmsg_type == IPV6_PKTINFO)
*index = ((struct in6_pktinfo*)CMSG_DATA(c))->ipi6_ifindex;
}
return size;
#endif
}
int local_on_interface(const struct local_interface* interface, const struct sockaddr_storage* source) {
if (source->ss_family != interface->address.ss_family || !ss_get_port(source)) return 0;
if (source->ss_family == AF_INET6) return IN6_IS_ADDR_LINKLOCAL(&((const struct sockaddr_in6*)source)->sin6_addr);
uint32_t address = ((const struct sockaddr_in*)source)->sin_addr.s_addr;
uint32_t own = ((const struct sockaddr_in*)&interface->address)->sin_addr.s_addr;
return address && address != UINT32_MAX && (address & interface->mask4) == (own & interface->mask4) &&
address != (own | ~interface->mask4) && address != (own & interface->mask4);
}
+27
View File
@@ -0,0 +1,27 @@
/* Внутренний API ОС: снимок локальных адресов и метаданные входящих UDP-пакетов. */
#ifndef LOCAL_TRANSPORT_PLATFORM_H
#define LOCAL_TRANSPORT_PLATFORM_H
#include "../../lib/platform_compat.h"
#include "../../lib/socket_compat.h"
#define LOCAL_INTERFACE_MAX 64
#define LOCAL_MULTICAST6 "ff12::7574:756e"
struct local_interface {
uint32_t index; // индекс интерфейса ОС; задаёт IPv6 scope
struct sockaddr_storage address; // локальный unicast IP, порт 0 для отдельного сокета отправки
struct sockaddr_storage broadcast; // broadcast IPv4 или multicast IPv6 с фиксированным discovery-портом
uint32_t mask4; // IPv4 netmask в network byte order; 0 для IPv6
};
/* Снимок UP-интерфейсов: IPv4 broadcast и IPv6 link-local, без loopback/P2P. count (<= capacity) либо -1. */
int local_interfaces(struct local_interface* out, size_t capacity);
/* Включить получение interface index через ancillary data для family. Возвращает 0/-1. */
int local_receive_options(socket_t fd, int family);
/* -2 = пакет усечён, -1 = ошибка ОС. index — входящий интерфейс (в том числе для IPv4 broadcast). */
ssize_t local_receive(socket_t fd, void* data, size_t capacity, struct sockaddr_storage* source, uint32_t* index);
/* Проверить адрес/порт источника: IPv4 той же подсети либо IPv6 link-local; 1 = подходит. */
int local_on_interface(const struct local_interface* interface, const struct sockaddr_storage* source);
#endif
+41
View File
@@ -0,0 +1,41 @@
/* Внутренний wire-формат: magic(16), version/type(1+1), length(2), пять uint64 BE.
* source/epoch, destination/epoch, request. Объявление не имеет destination/request. */
#ifndef LOCAL_TRANSPORT_PROTOCOL_H
#define LOCAL_TRANSPORT_PROTOCOL_H
#include "local_transport.h"
#include <string.h>
#define LOCAL_MAGIC "uTunChatLocalUDP"
#define LOCAL_HEADER_SIZE 60
#define LOCAL_PACKET_MAX (LOCAL_HEADER_SIZE + LOCAL_TRANSPORT_DATA_MAX)
#define LOCAL_ANNOUNCE 1
#define LOCAL_MESSAGE 2
#define LOCAL_ACK 3
static inline void local_put64(uint8_t* p, uint64_t value) {
for (int i = 7; i >= 0; i--) { p[i] = (uint8_t)value; value >>= 8; }
}
static inline uint64_t local_get64(const uint8_t* p) {
uint64_t value = 0;
for (int i = 0; i < 8; i++) value = value << 8 | p[i];
return value;
}
/* Непустой UTF-8 ник без NUL, управляющих ASCII, overlong, surrogate и кодов > U+10FFFF. */
static inline int local_name_valid(const uint8_t* p, size_t len) {
if (!len || len >= LOCAL_TRANSPORT_NAME_MAX) return 0;
for (size_t i = 0; i < len;) {
unsigned c = p[i++], value, count, minimum;
if (c < 0x80) { if (c < 0x20 || c == 0x7f) return 0; continue; }
if (c >= 0xc2 && c <= 0xdf) { value = c & 31; count = 1; minimum = 0x80; }
else if (c >= 0xe0 && c <= 0xef) { value = c & 15; count = 2; minimum = 0x800; }
else if (c >= 0xf0 && c <= 0xf4) { value = c & 7; count = 3; minimum = 0x10000; }
else return 0;
if (count > len - i) return 0;
while (count--) { c = p[i++]; if ((c & 0xc0) != 0x80) return 0; value = value << 6 | (c & 63); }
if (value < minimum || value > 0x10ffff || (value >= 0xd800 && value <= 0xdfff)) return 0;
}
return 1;
}
#endif
+4 -1
View File
@@ -1,5 +1,6 @@
// utun_instance.c - Root instance implementation
#include "utun_instance.h"
#include "chat/chat_local.h"
#include "stream/stream.h"
#include "etcp_router.h"
#include "config_parser.h"
@@ -840,6 +841,7 @@ int chat_service_start(struct UTUN_INSTANCE* instance) {
}
if (chat_local_start(instance) < 0) goto fail;
instance->chat_started = 1;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "chat service started"); return 0;
fail:
@@ -850,6 +852,7 @@ fail:
void chat_service_stop(struct UTUN_INSTANCE* instance) {
if (!instance) return;
instance->chat_started = 0;
chat_local_stop(instance);
chat_headless_control_destroy(instance);
call_headless_destroy(instance); radio_headless_destroy(instance);
stream_service_destroy(instance);
@@ -1106,7 +1109,7 @@ struct UTUN_INSTANCE *utun_instance_reload(struct UTUN_INSTANCE *instance, struc
void utun_add_activity_cbk(struct UTUN_INSTANCE* instance, utun_activity_cbk_fn fn, void* arg) {
if (!instance || !fn) return;
struct utun_activity_cbk_entry* e = u_malloc(sizeof(struct utun_activity_cbk_entry));
if (!e) return;
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "client activity subscription allocation failed"); return; }
e->fn = fn; e->arg = arg; e->next = instance->activity_cbks;
instance->activity_cbks = e;
}
+1
View File
@@ -229,6 +229,7 @@ struct UTUN_INSTANCE {
// Контексты чат-сервиса; освобождаются при chat_service_stop.
struct chat_core_ctx* chat_core; // каналы, сообщения и доступ к общей БД
struct local_transport* local_transport; // отдельный открытый LAN UDP, принадлежит чат-сервису
struct chat_sync* chat_sync; // join и синхронизация каналов
struct join_key_entry* join_keys; // зарегистрированные ключи приглашений
struct chat_join_registration* join_registrations; // ожидания KEY_REGISTER_ACK