Bound local discovery traffic and validate invitations
This commit is contained in:
1 parent
80b23997c0
commit
4c62f45984
8 files changed
+383
-23
No files matched your search
@@ -32,6 +32,11 @@ This file contains essential information for AI coding agents working in the uTu
|
|||||||
|
|
||||||
Доствп на узел colo2: ssh -p44322 logs@colo2
|
Доствп на узел colo2: ssh -p44322 logs@colo2
|
||||||
|
|
||||||
|
Структура .h файлов:
|
||||||
|
- в начале краткое и понятное описание: что делает модуль, как пользоваться.
|
||||||
|
- поля структур также кратко описываем - что, для его, если есть нюансы/значения
|
||||||
|
- для функций: описываем назначение, как пользоваться, описание аргументов, возврата. кратко но понятно.
|
||||||
|
|
||||||
## Quick Reference
|
## Quick Reference
|
||||||
|
|
||||||
**Repository:** uTun — общее сетевое ядро ETCP, независимо запускаемые UTUN и P2P-чат
|
**Repository:** uTun — общее сетевое ядро ETCP, независимо запускаемые UTUN и P2P-чат
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ void chat_local_peers_trampoline(void* arg) {
|
|||||||
static void peers_changed(void* arg) { chat_local_peers_trampoline(arg); }
|
static void peers_changed(void* arg) { chat_local_peers_trampoline(arg); }
|
||||||
|
|
||||||
static int valid_link(const char* link, size_t len) {
|
static int valid_link(const char* link, size_t len) {
|
||||||
struct InviteData invite;
|
struct InviteData invite = {0};
|
||||||
char error[128];
|
char error[128];
|
||||||
if (!len || len > LOCAL_TRANSPORT_DATA_MAX || memchr(link, 0, len) || len < 7 || memcmp(link, "utun://", 7)) {
|
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;
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation length/prefix bytes=%zu", len); return 0;
|
||||||
@@ -34,6 +34,10 @@ static int valid_link(const char* link, size_t len) {
|
|||||||
if (invite_link_decode(link, len, &invite, error, sizeof(error)) < 0) {
|
if (invite_link_decode(link, len, &invite, error, sizeof(error)) < 0) {
|
||||||
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation: %s", error); return 0;
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation: %s", error); return 0;
|
||||||
}
|
}
|
||||||
|
static const uint8_t empty_key[INVITE_PUBKEY_SIZE] = {0};
|
||||||
|
if (!invite.channelId || !invite.join_key || !invite.addrCount || !memcmp(invite.pubkey, empty_key, sizeof(empty_key))) {
|
||||||
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid local invitation: missing channel, join key or entry node"); return 0;
|
||||||
|
}
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,6 +5,7 @@
|
|||||||
#include "../../lib/debug_config.h"
|
#include "../../lib/debug_config.h"
|
||||||
#include "../../lib/mem.h"
|
#include "../../lib/mem.h"
|
||||||
#include <openssl/rand.h>
|
#include <openssl/rand.h>
|
||||||
|
#include <stdio.h>
|
||||||
#include <string.h>
|
#include <string.h>
|
||||||
|
|
||||||
#define ANNOUNCE_TB 100000
|
#define ANNOUNCE_TB 100000
|
||||||
@@ -13,6 +14,13 @@
|
|||||||
#define ENDPOINT_MAX 8
|
#define ENDPOINT_MAX 8
|
||||||
#define PENDING_MAX 8
|
#define PENDING_MAX 8
|
||||||
#define SEEN_MAX 128
|
#define SEEN_MAX 128
|
||||||
|
#define RX_PER_TICK 256
|
||||||
|
#define MESSAGES_PER_TICK 8
|
||||||
|
|
||||||
|
enum packet_drop {
|
||||||
|
DROP_FOREIGN, DROP_NONLOCAL, DROP_HEADER, DROP_ANNOUNCE, DROP_IDENTITY,
|
||||||
|
DROP_ENDPOINT, DROP_ACK, DROP_MESSAGE, DROP_CAPACITY, DROP_LIMIT, DROP_COUNT
|
||||||
|
};
|
||||||
|
|
||||||
struct local_socket {
|
struct local_socket {
|
||||||
struct local_socket* next;
|
struct local_socket* next;
|
||||||
@@ -51,16 +59,55 @@ struct local_transport {
|
|||||||
struct pending_entry pending[PENDING_MAX];
|
struct pending_entry pending[PENDING_MAX];
|
||||||
struct seen_entry seen[SEEN_MAX];
|
struct seen_entry seen[SEEN_MAX];
|
||||||
unsigned seen_next;
|
unsigned seen_next;
|
||||||
|
unsigned rx_left, messages_left, dropped[DROP_COUNT];
|
||||||
|
int rx_paused, peers_dirty;
|
||||||
int active, broadcasts;
|
int active, broadcasts;
|
||||||
};
|
};
|
||||||
|
|
||||||
static void socket_read(socket_t fd, void* arg);
|
static void socket_read(socket_t fd, void* arg);
|
||||||
static void tick(void* arg);
|
static void tick(void* arg);
|
||||||
|
|
||||||
|
/* Пустой ник допустим в конфиге ядра; в LAN показываем устойчивый ID вместо пустого имени. */
|
||||||
|
static const char* device_name(uint64_t node_id, const char* name, char fallback[17]) {
|
||||||
|
if (!name || name[0]) return name;
|
||||||
|
snprintf(fallback, 17, "%016llx", (unsigned long long)node_id);
|
||||||
|
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "device nickname is empty; using node_id=%s", fallback);
|
||||||
|
return fallback;
|
||||||
|
}
|
||||||
|
|
||||||
static void peers_changed(struct local_transport* t) {
|
static void peers_changed(struct local_transport* t) {
|
||||||
|
t->peers_dirty = 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void flush_peers(struct local_transport* t) {
|
||||||
|
if (!t->peers_dirty) return;
|
||||||
|
t->peers_dirty = 0;
|
||||||
|
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "local device table updated devices=%zu", t->peer_count);
|
||||||
if (t->callbacks.peers_changed) t->callbacks.peers_changed(t->arg);
|
if (t->callbacks.peers_changed) t->callbacks.peers_changed(t->arg);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* Враждебный UDP не должен порождать неограниченный поток логов. Причины считаются до следующего tick. */
|
||||||
|
static void drop_packet(struct local_transport* t, enum packet_drop reason) { t->dropped[reason]++; }
|
||||||
|
|
||||||
|
static void reset_receive(struct local_transport* t) {
|
||||||
|
static const char* reasons[DROP_COUNT] = {
|
||||||
|
"foreign", "non-local source", "header/discovery type", "announcement/name", "identity/destination",
|
||||||
|
"unknown/expired endpoint", "ACK payload", "message/application limit", "table capacity", "receive budget"
|
||||||
|
};
|
||||||
|
for (int i = 0; i < DROP_COUNT; i++) {
|
||||||
|
if (t->dropped[i]) {
|
||||||
|
if (i <= DROP_NONLOCAL) DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "local UDP dropped reason=%s count=%u", reasons[i], t->dropped[i]);
|
||||||
|
else DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "local UDP dropped reason=%s count=%u", reasons[i], t->dropped[i]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
memset(t->dropped, 0, sizeof(t->dropped));
|
||||||
|
t->rx_left = RX_PER_TICK; t->messages_left = MESSAGES_PER_TICK;
|
||||||
|
if (!t->rx_paused) return;
|
||||||
|
t->rx_paused = 0;
|
||||||
|
for (struct local_socket* s = t->sockets; s; s = s->next)
|
||||||
|
if (uasync_set_socket_read(t->ua, s->handle, 1)) DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "UDP read resume failed");
|
||||||
|
}
|
||||||
|
|
||||||
static struct peer_entry* find_peer(struct local_transport* t, uint64_t node_id) {
|
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];
|
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;
|
return NULL;
|
||||||
@@ -142,19 +189,20 @@ static void expire(struct local_transport* t, uint64_t now) {
|
|||||||
if (changed) peers_changed(t);
|
if (changed) peers_changed(t);
|
||||||
}
|
}
|
||||||
|
|
||||||
static void receive_packet(struct local_socket* s, struct sockaddr_storage* source, const uint8_t* data, size_t size) {
|
static void receive_packet(struct local_socket* s, struct sockaddr_storage* source, const uint8_t* data, size_t size, int discovery) {
|
||||||
struct local_transport* t = s->owner;
|
struct local_transport* t = s->owner;
|
||||||
if (size < 16 || memcmp(data, LOCAL_MAGIC, 16)) {
|
if (size < 16 || memcmp(data, LOCAL_MAGIC, 16)) {
|
||||||
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "foreign UDP ignored interface=%u source=%s bytes=%zu",
|
drop_packet(t, DROP_FOREIGN); return;
|
||||||
s->interface.index, sockaddr_storage_to_str(source).str, size); return;
|
|
||||||
}
|
}
|
||||||
if (size < LOCAL_HEADER_SIZE || size > LOCAL_PACKET_MAX || data[16] != 1 ||
|
if (size < LOCAL_HEADER_SIZE || size > LOCAL_PACKET_MAX || data[16] != 1 ||
|
||||||
(size_t)((unsigned)data[18] << 8 | data[19]) != size - LOCAL_HEADER_SIZE) {
|
(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;
|
drop_packet(t, DROP_HEADER); return;
|
||||||
}
|
}
|
||||||
|
/* Общий discovery-порт принимает только объявления, а доставка/ACK идут на отдельные unicast-сокеты. */
|
||||||
|
if (discovery && data[17] != LOCAL_ANNOUNCE) { drop_packet(t, DROP_HEADER); return; }
|
||||||
uint64_t peer_id = local_get64(data + 20), epoch = local_get64(data + 28), request = local_get64(data + 52);
|
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);
|
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 || !epoch) { drop_packet(t, DROP_IDENTITY); return; }
|
||||||
if (peer_id == t->node_id) return;
|
if (peer_id == t->node_id) return;
|
||||||
uint8_t type = data[17];
|
uint8_t type = data[17];
|
||||||
const uint8_t* payload = data + LOCAL_HEADER_SIZE;
|
const uint8_t* payload = data + LOCAL_HEADER_SIZE;
|
||||||
@@ -165,12 +213,12 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
type, (unsigned long long)peer_id, s->interface.index, sockaddr_storage_to_str(source).str, (unsigned long long)request, len);
|
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 (type == LOCAL_ANNOUNCE) {
|
||||||
if (destination || destination_epoch || request || !local_name_valid(payload, len)) {
|
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;
|
drop_packet(t, DROP_ANNOUNCE); return;
|
||||||
}
|
}
|
||||||
int changed = !p;
|
int changed = !p;
|
||||||
if (!p) {
|
if (!p) {
|
||||||
if (t->peer_count == LOCAL_TRANSPORT_PEERS_MAX) {
|
if (t->peer_count == LOCAL_TRANSPORT_PEERS_MAX) {
|
||||||
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "device table full peer=%016llx", (unsigned long long)peer_id); return;
|
drop_packet(t, DROP_CAPACITY); return;
|
||||||
}
|
}
|
||||||
p = &t->peers[t->peer_count++]; memset(p, 0, sizeof(*p)); p->peer.node_id = peer_id;
|
p = &t->peers[t->peer_count++]; memset(p, 0, sizeof(*p)); p->peer.node_id = peer_id;
|
||||||
}
|
}
|
||||||
@@ -184,13 +232,13 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
int new_endpoint = j == p->count;
|
int new_endpoint = j == p->count;
|
||||||
if (new_endpoint) {
|
if (new_endpoint) {
|
||||||
if (j == ENDPOINT_MAX) {
|
if (j == ENDPOINT_MAX) {
|
||||||
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "endpoint limit peer=%016llx", (unsigned long long)peer_id); return;
|
drop_packet(t, DROP_CAPACITY); return;
|
||||||
}
|
}
|
||||||
p->count++;
|
p->count++;
|
||||||
}
|
}
|
||||||
p->endpoints[j] = (struct local_endpoint){ .address = *source, .index = s->interface.index, .seen = now };
|
p->endpoints[j] = (struct local_endpoint){ .address = *source, .index = s->interface.index, .seen = now };
|
||||||
if (changed) {
|
if (changed) {
|
||||||
DEBUG_INFO(DEBUG_CATEGORY_DISCOVERY, "device discovered/updated peer=%016llx name=%s endpoints=%zu",
|
DEBUG_DEBUG(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);
|
(unsigned long long)peer_id, p->peer.name, p->count); peers_changed(t);
|
||||||
}
|
}
|
||||||
if (new_endpoint) announce(s, source);
|
if (new_endpoint) announce(s, source);
|
||||||
@@ -198,16 +246,15 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
}
|
}
|
||||||
if ((type != LOCAL_MESSAGE && type != LOCAL_ACK) || !request || !p || p->epoch != epoch ||
|
if ((type != LOCAL_MESSAGE && type != LOCAL_ACK) || !request || !p || p->epoch != epoch ||
|
||||||
destination != t->node_id || destination_epoch != t->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",
|
drop_packet(t, DROP_IDENTITY); return;
|
||||||
type, (unsigned long long)peer_id); return;
|
|
||||||
}
|
}
|
||||||
size_t j;
|
size_t j;
|
||||||
for (j = 0; j < p->count; j++)
|
for (j = 0; j < p->count; j++)
|
||||||
if (p->endpoints[j].index == s->interface.index && same_address(&p->endpoints[j].address, source) &&
|
if (p->endpoints[j].index == s->interface.index && same_address(&p->endpoints[j].address, source) &&
|
||||||
now - p->endpoints[j].seen < EXPIRE_TB) break;
|
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 (j == p->count) { drop_packet(t, DROP_ENDPOINT); return; }
|
||||||
if (type == LOCAL_ACK) {
|
if (type == LOCAL_ACK) {
|
||||||
if (len != 1 || payload[0] > LOCAL_DELIVERY_FORBIDDEN) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid ACK payload"); return; }
|
if (len != 1 || payload[0] > LOCAL_DELIVERY_FORBIDDEN) { drop_packet(t, DROP_ACK); return; }
|
||||||
for (int i = 0; i < PENDING_MAX; i++) {
|
for (int i = 0; i < PENDING_MAX; i++) {
|
||||||
struct pending_entry* pending = &t->pending[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; }
|
if (pending->request == request && pending->peer_id == peer_id && pending->epoch == epoch) { delivery(t, pending, payload[0]); return; }
|
||||||
@@ -215,7 +262,7 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "late ACK peer=%016llx request=%llu", (unsigned long long)peer_id, (unsigned long long)request);
|
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "late ACK peer=%016llx request=%llu", (unsigned long long)peer_id, (unsigned long long)request);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (!len) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "empty local message peer=%016llx", (unsigned long long)peer_id); return; }
|
if (!len) { drop_packet(t, DROP_MESSAGE); return; }
|
||||||
uint8_t status = LOCAL_DELIVERY_FAILED;
|
uint8_t status = LOCAL_DELIVERY_FAILED;
|
||||||
for (int i = 0; i < SEEN_MAX; i++) {
|
for (int i = 0; i < SEEN_MAX; i++) {
|
||||||
const struct seen_entry* entry = &t->seen[i];
|
const struct seen_entry* entry = &t->seen[i];
|
||||||
@@ -224,6 +271,8 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (status != LOCAL_DELIVERY_RECEIVED && t->callbacks.received) {
|
if (status != LOCAL_DELIVERY_RECEIVED && t->callbacks.received) {
|
||||||
|
if (!t->messages_left) { drop_packet(t, DROP_MESSAGE); return; }
|
||||||
|
t->messages_left--;
|
||||||
int result = t->callbacks.received(t->arg, &p->peer, payload, len);
|
int result = t->callbacks.received(t->arg, &p->peer, payload, len);
|
||||||
if (!result) {
|
if (!result) {
|
||||||
status = LOCAL_DELIVERY_RECEIVED;
|
status = LOCAL_DELIVERY_RECEIVED;
|
||||||
@@ -240,8 +289,16 @@ static void receive_packet(struct local_socket* s, struct sockaddr_storage* sour
|
|||||||
static void socket_read(socket_t fd, void* arg) {
|
static void socket_read(socket_t fd, void* arg) {
|
||||||
struct local_socket* socket = arg;
|
struct local_socket* socket = arg;
|
||||||
struct local_transport* t = socket->owner;
|
struct local_transport* t = socket->owner;
|
||||||
|
if (t->rx_paused) return;
|
||||||
/* Ограничение burst не позволяет локальному трафику вытеснить остальные callbacks ядра. */
|
/* Ограничение burst не позволяет локальному трафику вытеснить остальные callbacks ядра. */
|
||||||
for (int i = 0; i < 32; i++) {
|
for (int i = 0; i < 32; i++) {
|
||||||
|
if (!t->rx_left) {
|
||||||
|
/* Выключаем readiness до tick: простое чтение с отбрасыванием всё равно занимало бы CPU под flood. */
|
||||||
|
t->rx_paused = 1; drop_packet(t, DROP_LIMIT);
|
||||||
|
for (struct local_socket* s = t->sockets; s; s = s->next)
|
||||||
|
if (uasync_set_socket_read(t->ua, s->handle, 0)) DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "UDP read pause failed");
|
||||||
|
return;
|
||||||
|
}
|
||||||
uint8_t data[LOCAL_PACKET_MAX];
|
uint8_t data[LOCAL_PACKET_MAX];
|
||||||
struct sockaddr_storage source;
|
struct sockaddr_storage source;
|
||||||
uint32_t index;
|
uint32_t index;
|
||||||
@@ -252,17 +309,18 @@ static void socket_read(socket_t fd, void* arg) {
|
|||||||
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "receive failed: %s", socket_strerror(error));
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "receive failed: %s", socket_strerror(error));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (size == -2) { DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "truncated local datagram dropped"); continue; }
|
t->rx_left--;
|
||||||
|
if (size == -2) { drop_packet(t, DROP_HEADER); continue; }
|
||||||
/* Unicast-сокет уже привязан к адресу интерфейса. При обмене двух процессов на одном
|
/* Unicast-сокет уже привязан к адресу интерфейса. При обмене двух процессов на одном
|
||||||
* хосте ОС может указать loopback как ingress, но IPv6 scope нужен от локальной привязки. */
|
* хосте ОС может указать loopback как ingress, но IPv6 scope нужен от локальной привязки. */
|
||||||
if (socket->interface.index) index = socket->interface.index;
|
if (socket->interface.index) index = socket->interface.index;
|
||||||
struct local_socket* s = find_socket(t, &source, index);
|
struct local_socket* s = find_socket(t, &source, index);
|
||||||
if (!s) {
|
if (!s) {
|
||||||
DEBUG_DEBUG(DEBUG_CATEGORY_DISCOVERY, "non-local source ignored interface=%u source=%s", index, sockaddr_storage_to_str(&source).str);
|
drop_packet(t, DROP_NONLOCAL);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if (source.ss_family == AF_INET6) ((struct sockaddr_in6*)&source)->sin6_scope_id = index;
|
if (source.ss_family == AF_INET6) ((struct sockaddr_in6*)&source)->sin6_scope_id = index;
|
||||||
receive_packet(s, &source, data, (size_t)size);
|
receive_packet(s, &source, data, (size_t)size, !socket->interface.index);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -404,6 +462,7 @@ static void try_send(struct local_transport* t, struct pending_entry* p) {
|
|||||||
static void tick(void* arg) {
|
static void tick(void* arg) {
|
||||||
struct local_transport* t = arg;
|
struct local_transport* t = arg;
|
||||||
t->timer = NULL;
|
t->timer = NULL;
|
||||||
|
reset_receive(t);
|
||||||
uint64_t now = get_time_tb();
|
uint64_t now = get_time_tb();
|
||||||
if (now >= t->next_announce) {
|
if (now >= t->next_announce) {
|
||||||
reconcile(t); t->next_announce = now + ANNOUNCE_TB;
|
reconcile(t); t->next_announce = now + ANNOUNCE_TB;
|
||||||
@@ -411,6 +470,7 @@ static void tick(void* arg) {
|
|||||||
}
|
}
|
||||||
expire(t, now);
|
expire(t, now);
|
||||||
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) try_send(t, &t->pending[i]);
|
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) try_send(t, &t->pending[i]);
|
||||||
|
flush_peers(t);
|
||||||
t->timer = uasync_set_timeout(t->ua, TICK_TB, t, tick, "local_transport");
|
t->timer = uasync_set_timeout(t->ua, TICK_TB, t, tick, "local_transport");
|
||||||
if (!t->timer) {
|
if (!t->timer) {
|
||||||
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local transport timer allocation failed; disabling discovery");
|
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "local transport timer allocation failed; disabling discovery");
|
||||||
@@ -420,6 +480,7 @@ static void tick(void* arg) {
|
|||||||
|
|
||||||
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
|
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
|
||||||
const struct local_transport_callbacks* callbacks, void* arg) {
|
const struct local_transport_callbacks* callbacks, void* arg) {
|
||||||
|
char fallback[17]; name = device_name(node_id, name, fallback);
|
||||||
if (!ua || !node_id || !name || !callbacks || !local_name_valid((const uint8_t*)name, strlen(name))) {
|
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;
|
DEBUG_ERROR(DEBUG_CATEGORY_DISCOVERY, "invalid local transport configuration"); return NULL;
|
||||||
}
|
}
|
||||||
@@ -446,7 +507,9 @@ int local_transport_set_active(struct local_transport* t, int active) {
|
|||||||
while (t->sockets) { struct local_socket* s = t->sockets; t->sockets = s->next; close_socket(s); }
|
while (t->sockets) { struct local_socket* s = t->sockets; t->sockets = s->next; close_socket(s); }
|
||||||
t->joined6_count = 0;
|
t->joined6_count = 0;
|
||||||
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) delivery(t, &t->pending[i], 0);
|
for (int i = 0; i < PENDING_MAX; i++) if (t->pending[i].request) delivery(t, &t->pending[i], 0);
|
||||||
|
t->rx_paused = 0;
|
||||||
if (t->peer_count) { t->peer_count = 0; peers_changed(t); }
|
if (t->peer_count) { t->peer_count = 0; peers_changed(t); }
|
||||||
|
flush_peers(t);
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -466,6 +529,7 @@ int local_transport_set_broadcasts(struct local_transport* t, int enabled) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
int local_transport_set_name(struct local_transport* t, const char* name) {
|
int local_transport_set_name(struct local_transport* t, const char* name) {
|
||||||
|
char fallback[17]; name = device_name(t ? t->node_id : 0, name, fallback);
|
||||||
if (!t || !name || !local_name_valid((const uint8_t*)name, strlen(name))) {
|
if (!t || !name || !local_name_valid((const uint8_t*)name, strlen(name))) {
|
||||||
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid device name"); return -1;
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "invalid device name"); return -1;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
/* Открытый UDP-транспорт локальной сети. Не создаёт ETCP-сокетов и не подключает пиры.
|
/* Открытый UDP-транспорт локальной сети. Не создаёт ETCP-сокетов и не подключает пиры.
|
||||||
* Все вызовы и callbacks — в потоке ua. destroy/set_active — вне callbacks транспорта.
|
* Все вызовы и callbacks — в потоке ua. destroy/set_active — вне callbacks транспорта.
|
||||||
* Данные callback заимствованы до возврата; received возвращает 0 после принятия сообщения.
|
* Данные callback заимствованы до возврата; received возвращает 0 после принятия сообщения.
|
||||||
* delivery подтверждает приём сообщения приложением, а не выполнение его содержимого. */
|
* delivery подтверждает приём сообщения приложением, а не выполнение его содержимого.
|
||||||
|
* Входящий поток ограничен 256 UDP/сек и 8 новыми сообщениями/сек на транспорт; при flood возможны потери.
|
||||||
|
* peers_changed объединяет изменения таблицы до одного уведомления за секунду. */
|
||||||
#ifndef LOCAL_TRANSPORT_H
|
#ifndef LOCAL_TRANSPORT_H
|
||||||
#define LOCAL_TRANSPORT_H
|
#define LOCAL_TRANSPORT_H
|
||||||
|
|
||||||
@@ -37,7 +39,8 @@ struct local_transport_callbacks {
|
|||||||
};
|
};
|
||||||
|
|
||||||
/* Создать выключенный транспорт: ua заимствован, name/callbacks копируются, arg передаётся callbacks.
|
/* Создать выключенный транспорт: ua заимствован, name/callbacks копируются, arg передаётся callbacks.
|
||||||
* node_id != 0, name — 1..127 байтов UTF-8. Возвращает владельческий handle либо NULL с ошибкой в логе. */
|
* node_id != 0, name — 0..127 байтов UTF-8; пустой ник заменяется hex node_id.
|
||||||
|
* Возвращает владельческий handle либо NULL с ошибкой в логе. */
|
||||||
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
|
struct local_transport* local_transport_create(struct UASYNC* ua, uint64_t node_id, const char* name,
|
||||||
const struct local_transport_callbacks* callbacks, void* arg);
|
const struct local_transport_callbacks* callbacks, void* arg);
|
||||||
/* Закрыть сокеты, отменить таймеры/доставки и освободить handle. NULL допустим. */
|
/* Закрыть сокеты, отменить таймеры/доставки и освободить handle. NULL допустим. */
|
||||||
|
|||||||
@@ -250,6 +250,11 @@ check_PROGRAMS += test_local_transport
|
|||||||
test_local_transport_SOURCES = test_local_transport.c $(test_clock_sources)
|
test_local_transport_SOURCES = test_local_transport.c $(test_clock_sources)
|
||||||
test_local_transport_LDADD = $(LIBUTUN) $(COMMON_LIBS)
|
test_local_transport_LDADD = $(LIBUTUN) $(COMMON_LIBS)
|
||||||
test_local_transport_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=local_interfaces -Wl,--wrap=socket_sendto
|
test_local_transport_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=local_interfaces -Wl,--wrap=socket_sendto
|
||||||
|
check_PROGRAMS += test_local_transport_packets
|
||||||
|
test_local_transport_packets_SOURCES = test_local_transport_packets.c $(test_clock_sources)
|
||||||
|
test_local_transport_packets_LDADD = $(LIBUTUN) $(COMMON_LIBS)
|
||||||
|
test_local_transport_packets_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=local_interfaces -Wl,--wrap=socket_sendto \
|
||||||
|
-Wl,--wrap=local_receive -Wl,--wrap=uasync_add_socket_t
|
||||||
test_stcp_LDFLAGS = -Wl,--wrap=send -Wl,--wrap=recv -Wl,--wrap=random_bytes -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl
|
test_stcp_LDFLAGS = -Wl,--wrap=send -Wl,--wrap=recv -Wl,--wrap=random_bytes -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl
|
||||||
standby_test_sources = ../tools/chatgui-android/libutun_lite/standby.c ../src/utun_instance.c ../src/transport_layer/etcp_keepalive.c
|
standby_test_sources = ../tools/chatgui-android/libutun_lite/standby.c ../src/utun_instance.c ../src/transport_layer/etcp_keepalive.c
|
||||||
standby_test_cppflags = $(AM_CPPFLAGS) -I$(top_srcdir)/src/chat -I$(top_srcdir)/tools/chatgui-android/libutun_lite -DUTUN_HAVE_STANDBY
|
standby_test_cppflags = $(AM_CPPFLAGS) -I$(top_srcdir)/src/chat -I$(top_srcdir)/tools/chatgui-android/libutun_lite -DUTUN_HAVE_STANDBY
|
||||||
|
|||||||
@@ -0,0 +1,270 @@
|
|||||||
|
/* Повреждённые кодограммы и flood проходят штатный socket callback и chat-адаптер.
|
||||||
|
* Подменён recvmsg boundary: воспроизводимый корпус без потерь UDP и без рассылки в пользовательскую LAN.
|
||||||
|
* Реальные recvmsg, усечение UDP и IPv6 дополнительно проверяет test_local_transport. */
|
||||||
|
#include "../src/local_transport/local_transport.h"
|
||||||
|
#include "../src/local_transport/platform.h"
|
||||||
|
#include "../src/local_transport/protocol.h"
|
||||||
|
#include "../src/chat/chat_local.h"
|
||||||
|
#include "../src/chat/chat_event.h"
|
||||||
|
#include "../src/chat/invite_link.h"
|
||||||
|
#include "../src/utun_instance.h"
|
||||||
|
#include "../lib/debug_config.h"
|
||||||
|
#include "../lib/mem.h"
|
||||||
|
#include "test_clock.h"
|
||||||
|
#include <net/if.h>
|
||||||
|
#include <stdio.h>
|
||||||
|
#include <stdlib.h>
|
||||||
|
#include <string.h>
|
||||||
|
|
||||||
|
#define REQUIRE(x) do { if (!(x)) { fprintf(stderr, "line=%d failed: %s\n", __LINE__, #x); exit(1); } } while (0)
|
||||||
|
|
||||||
|
struct test_socket { socket_t fd; socket_t_callback_t callback; void* arg; void* handle; };
|
||||||
|
static struct test_socket sockets[3];
|
||||||
|
static int socket_count, receiving, consumed, changes, invitations, log_lines, wire_packets;
|
||||||
|
static uint32_t interface_index, random_state = 0x938d26af;
|
||||||
|
static uint64_t own_epoch;
|
||||||
|
static uint8_t input[65507];
|
||||||
|
static size_t input_size;
|
||||||
|
static struct sockaddr_storage source;
|
||||||
|
|
||||||
|
int __wrap_local_interfaces(struct local_interface* interfaces, size_t capacity) {
|
||||||
|
REQUIRE(capacity > 0);
|
||||||
|
memset(interfaces, 0, sizeof(*interfaces));
|
||||||
|
interfaces->index = interface_index; interfaces->mask4 = htonl(0xff000000);
|
||||||
|
struct sockaddr_in* address = (struct sockaddr_in*)&interfaces->address;
|
||||||
|
address->sin_family = AF_INET; address->sin_addr.s_addr = htonl(0x7f000001);
|
||||||
|
struct sockaddr_in* broadcast = (struct sockaddr_in*)&interfaces->broadcast;
|
||||||
|
broadcast->sin_family = AF_INET; broadcast->sin_addr.s_addr = htonl(0x7fffffff);
|
||||||
|
broadcast->sin_port = htons(LOCAL_TRANSPORT_PORT);
|
||||||
|
return 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
void* __real_uasync_add_socket_t(struct UASYNC*, socket_t, socket_t_callback_t, socket_t_callback_t,
|
||||||
|
socket_t_callback_t, const char*, void*);
|
||||||
|
void* __wrap_uasync_add_socket_t(struct UASYNC* ua, socket_t fd, socket_t_callback_t read, socket_t_callback_t write,
|
||||||
|
socket_t_callback_t error, const char* name, void* arg) {
|
||||||
|
void* handle = __real_uasync_add_socket_t(ua, fd, read, write, error, name, arg);
|
||||||
|
if (handle && name && !strcmp(name, "local_transport")) {
|
||||||
|
REQUIRE(socket_count < 3); sockets[socket_count++] = (struct test_socket){fd, read, arg, handle};
|
||||||
|
}
|
||||||
|
return handle;
|
||||||
|
}
|
||||||
|
|
||||||
|
ssize_t __real_local_receive(socket_t, void*, size_t, struct sockaddr_storage*, uint32_t*);
|
||||||
|
ssize_t __wrap_local_receive(socket_t fd, void* data, size_t capacity, struct sockaddr_storage* address, uint32_t* index) {
|
||||||
|
if (!receiving) {
|
||||||
|
ssize_t result = __real_local_receive(fd, data, capacity, address, index);
|
||||||
|
if (result >= 0 || result == -2) wire_packets++;
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
if (consumed) { errno = EAGAIN; return -1; }
|
||||||
|
consumed = 1; *address = source; *index = interface_index;
|
||||||
|
size_t size = input_size < capacity ? input_size : capacity;
|
||||||
|
memcpy(data, input, size);
|
||||||
|
return input_size > capacity ? -2 : (ssize_t)input_size;
|
||||||
|
}
|
||||||
|
|
||||||
|
ssize_t __wrap_socket_sendto(socket_t fd, const void* buffer, size_t len, const struct sockaddr* address, socklen_t address_size) {
|
||||||
|
(void)fd; (void)address; (void)address_size;
|
||||||
|
const uint8_t* data = buffer;
|
||||||
|
REQUIRE(len >= LOCAL_HEADER_SIZE && len <= LOCAL_PACKET_MAX && !memcmp(data, LOCAL_MAGIC, 16));
|
||||||
|
REQUIRE(local_get64(data + 20) == 1);
|
||||||
|
if (data[17] == LOCAL_ANNOUNCE) own_epoch = local_get64(data + 28);
|
||||||
|
return (ssize_t)len;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) {
|
||||||
|
(void)inst;
|
||||||
|
REQUIRE(data && len > 0);
|
||||||
|
if (type == CHAT_EVT_LOCAL_DEVICES) {
|
||||||
|
REQUIRE(len >= 2 && len <= 2 + LOCAL_TRANSPORT_PEERS_MAX * (9 + LOCAL_TRANSPORT_NAME_MAX - 1));
|
||||||
|
size_t count = (size_t)data[0] << 8 | data[1], offset = 2;
|
||||||
|
REQUIRE(count <= LOCAL_TRANSPORT_PEERS_MAX);
|
||||||
|
for (size_t i = 0; i < count; i++) {
|
||||||
|
REQUIRE(offset + 9 <= (size_t)len);
|
||||||
|
size_t size = data[offset + 8]; offset += 9;
|
||||||
|
REQUIRE(offset + size <= (size_t)len && local_name_valid(data + offset, size)); offset += size;
|
||||||
|
}
|
||||||
|
REQUIRE(offset == (size_t)len); changes++;
|
||||||
|
}
|
||||||
|
if (type == CHAT_EVT_LOCAL_INVITE) {
|
||||||
|
REQUIRE(len > 9 + data[8] && len <= 9 + LOCAL_TRANSPORT_NAME_MAX + LOCAL_TRANSPORT_DATA_MAX);
|
||||||
|
struct InviteData invite = {0};
|
||||||
|
REQUIRE(!invite_link_decode((const char*)data + 9 + data[8], (size_t)len - 9 - data[8], &invite, NULL, 0));
|
||||||
|
REQUIRE(invite.channelId && invite.join_key && invite.addrCount); invitations++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void log_hook(int level, const char* category, const char* message) {
|
||||||
|
(void)level; (void)category; (void)message; log_lines++;
|
||||||
|
}
|
||||||
|
|
||||||
|
static uint32_t random32(void) {
|
||||||
|
uint32_t value = random_state;
|
||||||
|
value ^= value << 13; value ^= value >> 17; value ^= value << 5;
|
||||||
|
return random_state = value;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void packet(uint8_t type, uint64_t peer, uint64_t request, const void* payload, size_t size) {
|
||||||
|
REQUIRE(size <= sizeof(input) - LOCAL_HEADER_SIZE);
|
||||||
|
memset(input, 0, LOCAL_HEADER_SIZE); memcpy(input, LOCAL_MAGIC, 16);
|
||||||
|
input[16] = 1; input[17] = type; input[18] = (uint8_t)(size >> 8); input[19] = (uint8_t)size;
|
||||||
|
local_put64(input + 20, peer); local_put64(input + 28, 100);
|
||||||
|
if (type != LOCAL_ANNOUNCE) {
|
||||||
|
local_put64(input + 36, 1); local_put64(input + 44, own_epoch); local_put64(input + 52, request);
|
||||||
|
}
|
||||||
|
if (size) memcpy(input + LOCAL_HEADER_SIZE, payload, size);
|
||||||
|
input_size = LOCAL_HEADER_SIZE + size;
|
||||||
|
}
|
||||||
|
|
||||||
|
static int inject(struct test_socket* socket) {
|
||||||
|
receiving = 1; consumed = 0;
|
||||||
|
socket->callback(socket->fd, socket->arg);
|
||||||
|
receiving = 0;
|
||||||
|
return consumed;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void next_window(struct UASYNC* ua) { test_clock_next_timer(ua); }
|
||||||
|
static void heartbeat(void* arg) { (*(int*)arg)++; }
|
||||||
|
|
||||||
|
static void wire_boundaries(struct UASYNC* ua, struct test_socket* unicast) {
|
||||||
|
socket_t raw = socket_create_udp(AF_INET); REQUIRE(raw != SOCKET_INVALID);
|
||||||
|
int enabled = 1;
|
||||||
|
REQUIRE(!setsockopt(raw, SOL_SOCKET, SO_BROADCAST, &enabled, sizeof(enabled)));
|
||||||
|
struct sockaddr_in broadcast = { .sin_family = AF_INET, .sin_port = htons(LOCAL_TRANSPORT_PORT),
|
||||||
|
.sin_addr.s_addr = htonl(0x7fffffff) };
|
||||||
|
struct sockaddr_storage endpoint; socklen_t size = sizeof(endpoint);
|
||||||
|
REQUIRE(!getsockname(unicast->fd, (struct sockaddr*)&endpoint, &size));
|
||||||
|
const size_t sizes[] = {LOCAL_PACKET_MAX, LOCAL_PACKET_MAX + 1, sizeof(input)};
|
||||||
|
packet(LOCAL_ANNOUNCE, 2, 0, "Peer", 4);
|
||||||
|
int before = invitations;
|
||||||
|
/* Необёрнутый sendto проверяет настоящий recvmsg, нулевые/короткие UDP и MSG_TRUNC, включая 65507 байт. */
|
||||||
|
for (size_t i = 0; i < LOCAL_HEADER_SIZE + 1 + sizeof(sizes) / sizeof(sizes[0]); i++) {
|
||||||
|
size_t len = i <= LOCAL_HEADER_SIZE ? i : sizes[i - LOCAL_HEADER_SIZE - 1];
|
||||||
|
REQUIRE(sendto(raw, input, len, 0, (struct sockaddr*)&broadcast, sizeof(broadcast)) == (ssize_t)len);
|
||||||
|
REQUIRE(sendto(raw, input, len, 0, (struct sockaddr*)&endpoint, size) == (ssize_t)len);
|
||||||
|
uasync_poll(ua, 0);
|
||||||
|
}
|
||||||
|
REQUIRE(invitations == before);
|
||||||
|
next_window(ua);
|
||||||
|
int beat = 0, packets_before = wire_packets;
|
||||||
|
REQUIRE(uasync_set_timeout(ua, 1, &beat, heartbeat, "hostile_udp_heartbeat"));
|
||||||
|
uint64_t start = get_time_tb();
|
||||||
|
for (int i = 0; i < 10000; i++) {
|
||||||
|
REQUIRE(sendto(raw, input, 16, 0, (struct sockaddr*)&broadcast, sizeof(broadcast)) == 16);
|
||||||
|
if (!(i % 32)) uasync_poll(ua, 0);
|
||||||
|
}
|
||||||
|
test_clock_run_for(ua, 100);
|
||||||
|
REQUIRE(beat == 1 && invitations == before);
|
||||||
|
REQUIRE((uint64_t)(wire_packets - packets_before) <= 256 * ((get_time_tb() - start) / 10000 + 2));
|
||||||
|
socket_close_wrapper(raw); next_window(ua);
|
||||||
|
}
|
||||||
|
|
||||||
|
int main(void) {
|
||||||
|
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
|
||||||
|
debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO);
|
||||||
|
debug_enable_console(0); debug_set_log_hook(log_hook);
|
||||||
|
test_clock_init(10000);
|
||||||
|
interface_index = if_nametoindex("lo"); REQUIRE(interface_index);
|
||||||
|
struct sockaddr_in* address = (struct sockaddr_in*)&source;
|
||||||
|
address->sin_family = AF_INET; address->sin_addr.s_addr = htonl(0x7f000002); address->sin_port = htons(55000);
|
||||||
|
struct UASYNC* ua = uasync_create(); REQUIRE(ua);
|
||||||
|
struct UTUN_INSTANCE inst = { .ua = ua, .node_id = 1, .name = "", .client_type = CLIENT_TYPE_DESKTOP,
|
||||||
|
.chat_event_handler = event };
|
||||||
|
REQUIRE(!chat_local_start(&inst) && socket_count == 3 && own_epoch);
|
||||||
|
struct test_socket *unicast = NULL, *discovery = NULL;
|
||||||
|
for (int i = 0; i < socket_count; i++) {
|
||||||
|
struct sockaddr_storage bound; socklen_t size = sizeof(bound);
|
||||||
|
REQUIRE(!getsockname(sockets[i].fd, (struct sockaddr*)&bound, &size));
|
||||||
|
if (bound.ss_family != AF_INET) continue;
|
||||||
|
if (ss_get_port(&bound) == LOCAL_TRANSPORT_PORT) discovery = &sockets[i]; else unicast = &sockets[i];
|
||||||
|
}
|
||||||
|
REQUIRE(unicast && discovery);
|
||||||
|
struct InviteData invite = { .channelId = 123, .join_key = 456, .pubkey = {1}, .addrCount = 1 };
|
||||||
|
invite.addrs[0] = (struct InviteAddr){ .family = 4, .address = {127, 0, 0, 1}, .proto = INVITE_PROTO_UDP, .port = 12345 };
|
||||||
|
char link[LOCAL_TRANSPORT_DATA_MAX + 1]; REQUIRE(invite_link_encode(&invite, NULL, link, sizeof(link)) > 0);
|
||||||
|
|
||||||
|
/* На discovery-порт нельзя доставить сообщение/ACK даже от ранее объявленного устройства. */
|
||||||
|
packet(LOCAL_ANNOUNCE, 2, 0, "Peer", 4); REQUIRE(inject(discovery));
|
||||||
|
packet(LOCAL_MESSAGE, 2, 1, link, strlen(link)); REQUIRE(inject(discovery)); REQUIRE(!invitations);
|
||||||
|
packet(LOCAL_MESSAGE, 2, 1, link, strlen(link)); REQUIRE(inject(unicast)); REQUIRE(invitations == 1);
|
||||||
|
packet(LOCAL_MESSAGE, 2, 1, link, strlen(link)); REQUIRE(inject(unicast)); REQUIRE(invitations == 1);
|
||||||
|
next_window(ua);
|
||||||
|
|
||||||
|
int before = invitations;
|
||||||
|
for (int i = 0; i < 64; i++) {
|
||||||
|
packet(LOCAL_MESSAGE, 2, 100 + i, link, strlen(link)); REQUIRE(inject(unicast));
|
||||||
|
}
|
||||||
|
REQUIRE(invitations - before == 8);
|
||||||
|
next_window(ua);
|
||||||
|
|
||||||
|
/* Flood не вызывает ни busy-loop, ни отдельного события/лога на каждое объявление. */
|
||||||
|
int processed = 0, changes_before = changes, logs_before = log_lines;
|
||||||
|
for (int i = 0; i < 10000; i++) {
|
||||||
|
packet(LOCAL_ANNOUNCE, 2, 0, i & 1 ? "Peer A" : "Peer B", 6);
|
||||||
|
processed += inject(discovery);
|
||||||
|
}
|
||||||
|
REQUIRE(processed == 256 && changes == changes_before && log_lines == logs_before);
|
||||||
|
next_window(ua); REQUIRE(changes == changes_before + 1 && log_lines - logs_before <= 10);
|
||||||
|
packet(LOCAL_MESSAGE, 2, 999, link, strlen(link)); REQUIRE(inject(unicast)); REQUIRE(invitations == before + 9);
|
||||||
|
|
||||||
|
/* Исчерпание таблиц узлов и endpoints остаётся внутри предвыделенной памяти. */
|
||||||
|
next_window(ua);
|
||||||
|
for (unsigned i = 0; i < 160; i++) {
|
||||||
|
packet(LOCAL_ANNOUNCE, i + 2, 0, "Peer", 4); REQUIRE(inject(discovery));
|
||||||
|
}
|
||||||
|
struct local_peer peers[LOCAL_TRANSPORT_PEERS_MAX];
|
||||||
|
REQUIRE(local_transport_peers(inst.local_transport, peers, LOCAL_TRANSPORT_PEERS_MAX) == LOCAL_TRANSPORT_PEERS_MAX);
|
||||||
|
for (int i = 0; i < 16; i++) {
|
||||||
|
address->sin_port = htons(55000 + i);
|
||||||
|
packet(LOCAL_ANNOUNCE, 2, 0, "Peer", 4); REQUIRE(inject(discovery));
|
||||||
|
}
|
||||||
|
address->sin_port = htons(55000); next_window(ua);
|
||||||
|
wire_boundaries(ua, unicast);
|
||||||
|
/* Дальше корпус подаётся вручную: настоящие UDP параллельных тестов не должны тратить его receive budget. */
|
||||||
|
for (int i = 0; i < socket_count; i++) REQUIRE(!uasync_set_socket_read(ua, sockets[i].handle, 0));
|
||||||
|
|
||||||
|
size_t allocated = u_get_allocated_count();
|
||||||
|
unsigned long cases = 0;
|
||||||
|
debug_set_category_level(DEBUG_CATEGORY_DISCOVERY, DEBUG_LEVEL_NONE);
|
||||||
|
uint8_t payload[LOCAL_TRANSPORT_DATA_MAX];
|
||||||
|
/* Миллион воспроизводимых мутаций: все длины заголовка, версии/типы/ID, UTF-8, тела invite и ACK. */
|
||||||
|
for (unsigned i = 0; i < 1000000; i++) {
|
||||||
|
if (!(i % 128)) next_window(ua);
|
||||||
|
uint32_t mode = random32() % 8;
|
||||||
|
size_t size = random32() % (sizeof(payload) + 1);
|
||||||
|
for (size_t j = 0; j < size; j++) payload[j] = (uint8_t)random32();
|
||||||
|
packet(mode < 3 ? LOCAL_ANNOUNCE : mode < 6 ? LOCAL_MESSAGE : LOCAL_ACK, 2, i + 1000, payload, size);
|
||||||
|
if (mode == 0) input_size = i % (LOCAL_HEADER_SIZE + 1);
|
||||||
|
if (mode == 1) {
|
||||||
|
for (unsigned j = 0, n = random32() % 8 + 1; j < n; j++) input[random32() % input_size] ^= (uint8_t)random32();
|
||||||
|
}
|
||||||
|
if (mode == 2) { local_put64(input + 20, random32() % 160 + 2); local_put64(input + 28, random32()); }
|
||||||
|
if (mode == 3 && size >= 7) memcpy(input + LOCAL_HEADER_SIZE, "utun://", 7);
|
||||||
|
if (mode == 4) {
|
||||||
|
packet(LOCAL_MESSAGE, 2, i + 1000, link, strlen(link));
|
||||||
|
for (unsigned j = 0, n = random32() % 8 + 1; j < n; j++)
|
||||||
|
input[LOCAL_HEADER_SIZE + random32() % strlen(link)] ^= (uint8_t)random32();
|
||||||
|
}
|
||||||
|
if (mode == 5) { local_put64(input + 36, random32()); local_put64(input + 44, random32()); }
|
||||||
|
if (mode == 6) { input_size = random32() % sizeof(input); }
|
||||||
|
if (mode == 7) { input[18] = (uint8_t)random32(); input[19] = (uint8_t)random32(); }
|
||||||
|
REQUIRE(inject(i & 1 ? unicast : discovery)); cases++;
|
||||||
|
/* Сохраняем достижимость MESSAGE/ACK после мутаций epoch/TTL и смены ника. */
|
||||||
|
if (!(i % 32)) { packet(LOCAL_ANNOUNCE, 2, 0, "Peer", 4); REQUIRE(inject(discovery)); }
|
||||||
|
}
|
||||||
|
next_window(ua);
|
||||||
|
REQUIRE(u_get_allocated_count() <= allocated + 2);
|
||||||
|
debug_set_category_level(DEBUG_CATEGORY_DISCOVERY, DEBUG_LEVEL_WARN);
|
||||||
|
packet(LOCAL_ANNOUNCE, 2, 0, "Peer", 4); REQUIRE(inject(discovery));
|
||||||
|
before = invitations;
|
||||||
|
packet(LOCAL_MESSAGE, 2, 2000000, link, strlen(link)); REQUIRE(inject(unicast)); REQUIRE(invitations == before + 1);
|
||||||
|
chat_local_stop(&inst); uasync_poll(ua, 0);
|
||||||
|
size_t timers, timers_freed, registered, removed;
|
||||||
|
uasync_get_stats(ua, &timers, &timers_freed, ®istered, &removed);
|
||||||
|
REQUIRE(timers == timers_freed && registered == removed);
|
||||||
|
uasync_destroy(ua, 0); test_clock_finish();
|
||||||
|
debug_set_log_hook(NULL); debug_enable_console(1);
|
||||||
|
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS: %lu hostile packets, bounded flood/events/memory, discovery isolation and recovery", cases);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
@@ -22,16 +22,21 @@ extern "C" {
|
|||||||
/* Счётчик и список строятся из durable таблицы, включая уже принятые решения. */
|
/* Счётчик и список строятся из durable таблицы, включая уже принятые решения. */
|
||||||
void MessageList::refreshKeyRequests() {
|
void MessageList::refreshKeyRequests() {
|
||||||
const bool group = !m_currentChannelId.isEmpty() && !isDmChannel(m_currentChannelId);
|
const bool group = !m_currentChannelId.isEmpty() && !isDmChannel(m_currentChannelId);
|
||||||
m_keyRequestsButton->setVisible(group);
|
if (!group || !m_db || !m_db->isOpen()) { m_keyRequestsButton->hide(); return; }
|
||||||
if (!group || !m_db || !m_db->isOpen()) return;
|
|
||||||
const QByteArray channel = m_currentChannelId.toUtf8();
|
const QByteArray channel = m_currentChannelId.toUtf8();
|
||||||
sqlite3_stmt* st = m_db->prepareOrNull("SELECT count(*) FROM attachment_key_requests WHERE channel=? AND incoming=1 AND state=0");
|
sqlite3_stmt* st = m_db->prepareOrNull("SELECT count(*) FROM attachment_key_requests WHERE channel=? AND incoming=1 AND state=0");
|
||||||
int count = 0;
|
int count = 0;
|
||||||
if (st) {
|
if (st) {
|
||||||
sqlite3_bind_text(st, 1, channel.constData(), -1, SQLITE_STATIC);
|
sqlite3_bind_text(st, 1, channel.constData(), -1, SQLITE_STATIC);
|
||||||
if (sqlite3_step(st) == SQLITE_ROW) count = sqlite3_column_int(st, 0);
|
const int rc = sqlite3_step(st);
|
||||||
|
if (rc == SQLITE_ROW) count = sqlite3_column_int(st, 0);
|
||||||
|
else GUI_ERROR("attachment keys: request count read failed channel=%s rc=%d", channel.constData(), rc);
|
||||||
sqlite3_finalize(st);
|
sqlite3_finalize(st);
|
||||||
}
|
}
|
||||||
|
const bool visible = count > 0;
|
||||||
|
if (m_keyRequestsButton->isHidden() == visible)
|
||||||
|
GUI_DEBUG("attachment keys: request button channel=%s pending=%d visible=%d", channel.constData(), count, visible);
|
||||||
|
m_keyRequestsButton->setVisible(visible);
|
||||||
QString title = QCoreApplication::translate("ChatGui", "Decryption requests");
|
QString title = QCoreApplication::translate("ChatGui", "Decryption requests");
|
||||||
m_keyRequestsButton->setText(count ? title + QStringLiteral(" · %1").arg(count) : title);
|
m_keyRequestsButton->setText(count ? title + QStringLiteral(" · %1").arg(count) : title);
|
||||||
m_keyRequestsButton->setAccessibleName(m_keyRequestsButton->text());
|
m_keyRequestsButton->setAccessibleName(m_keyRequestsButton->text());
|
||||||
|
|||||||
@@ -449,10 +449,14 @@ void MainWindow::setupBridgeCallbacks() {
|
|||||||
QString name = QString::fromUtf8((const char*)data + 9, data[8]);
|
QString name = QString::fromUtf8((const char*)data + 9, data[8]);
|
||||||
QString link = QString::fromUtf8((const char*)data + 9 + data[8], len - 9 - data[8]);
|
QString link = QString::fromUtf8((const char*)data + 9 + data[8], len - 9 - data[8]);
|
||||||
auto* window = s_mainWindow;
|
auto* window = s_mainWindow;
|
||||||
|
if (window->findChildren<QMessageBox*>(QStringLiteral("localInvitePrompt")).size() >= 8) {
|
||||||
|
DEBUG_WARN(DEBUG_CATEGORY_DISCOVERY, "incoming local invitation prompt limit reached"); return;
|
||||||
|
}
|
||||||
auto* prompt = new QMessageBox(QMessageBox::Question, QCoreApplication::translate("ChatGui", "Invitation"),
|
auto* prompt = new QMessageBox(QMessageBox::Question, QCoreApplication::translate("ChatGui", "Invitation"),
|
||||||
QCoreApplication::translate("ChatGui", "%1 sent an invitation over the local network. Connect?").arg(name),
|
QCoreApplication::translate("ChatGui", "%1 sent an invitation over the local network. Connect?").arg(name),
|
||||||
QMessageBox::Yes | QMessageBox::No, window);
|
QMessageBox::Yes | QMessageBox::No, window);
|
||||||
prompt->setTextFormat(Qt::PlainText); prompt->setAttribute(Qt::WA_DeleteOnClose);
|
prompt->setTextFormat(Qt::PlainText); prompt->setAttribute(Qt::WA_DeleteOnClose);
|
||||||
|
prompt->setObjectName(QStringLiteral("localInvitePrompt"));
|
||||||
QObject::connect(prompt, &QMessageBox::finished, window, [window, link](int result) {
|
QObject::connect(prompt, &QMessageBox::finished, window, [window, link](int result) {
|
||||||
if (result == QMessageBox::Yes) window->onMessageLinkActivated(link);
|
if (result == QMessageBox::Yes) window->onMessageLinkActivated(link);
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in new issue
Block a user