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

Define signed stream member options and BGP eligibility

Browse Source
This commit is contained in:
evgeny committed 2026-10-12 00:01:38 +02:00
1 parent a33666b778
commit b6f80668d1
3 files changed
+271 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_member.c
+219
View File
@@ -0,0 +1,219 @@
#include <stdio.h>
#include <string.h>
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "../utun_instance.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "stream_block.h"
#include "stream_member.h"
/* Канонические IDs не проходят через JSON number и не теряют старшие биты. */
static int member_hex(const char **text, uint64_t *value) {
const char *p = *text;
uint64_t result = 0;
for (unsigned i = 0; i < 16; i++) {
unsigned digit;
if (p[i] >= '0' && p[i] <= '9') digit = (unsigned)(p[i] - '0');
else if (p[i] >= 'a' && p[i] <= 'f') digit = (unsigned)(p[i] - 'a') + 10;
else return STREAM_ERR_FORMAT;
result = (result << 4) | digit;
}
if (!result) return STREAM_ERR_FORMAT;
*text = p + 16;
*value = result;
return STREAM_OK;
}
/* Десятичные параметры имеют ограничение ещё до сужения до uint8/uint16. */
static int member_decimal(const char **text, unsigned maximum, unsigned *value) {
const char *p = *text;
unsigned result = 0;
if (*p < '0' || *p > '9' || (*p == '0' && p[1] >= '0' && p[1] <= '9')) return STREAM_ERR_FORMAT;
while (*p >= '0' && *p <= '9') {
unsigned digit = (unsigned)(*p++ - '0');
if (result > (maximum - digit) / 10) return STREAM_ERR_FORMAT;
result = result * 10 + digit;
}
if (result > maximum) return STREAM_ERR_FORMAT;
*text = p;
*value = result;
return STREAM_OK;
}
struct member_parse {
struct stream_member_state state;
unsigned seen;
};
/* Полностью проверить обе опции, в том числе повторы ключей и порядок списка. */
static int member_option(const char *key, const char *value, void *arg) {
struct member_parse *parse = arg;
if (!strcmp(key, STREAM_MEMBER_SOURCE)) {
if (parse->seen & 1) return -1;
parse->seen |= 1;
if (!value[0]) return 0;
const char *p = value;
unsigned n, r, size;
if (member_hex(&p, &parse->state.source.source_epoch) != STREAM_OK || *p++ != ':'
|| member_decimal(&p, 255, &n) != STREAM_OK || *p++ != ':'
|| member_decimal(&p, 254, &r) != STREAM_OK || *p++ != ':'
|| member_decimal(&p, RS_MAX_BLOCK_SIZE, &size) != STREAM_OK || *p) return -1;
parse->state.source.n = (uint8_t)n;
parse->state.source.r = (uint8_t)r;
parse->state.source.block_size = (uint16_t)size;
if (stream_descriptor_validate(&parse->state.source) != STREAM_OK) return -1;
parse->state.published = 1;
} else if (!strcmp(key, STREAM_MEMBER_SUBSCRIPTIONS)) {
if (parse->seen & 2) return -1;
parse->seen |= 2;
const char *p = value;
while (*p) {
if (parse->state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) return -1;
struct stream_subscription_key entry;
if (member_hex(&p, &entry.source_node_id) != STREAM_OK || *p++ != ':'
|| member_hex(&p, &entry.source_epoch) != STREAM_OK) return -1;
if (parse->state.count) {
const struct stream_subscription_key *previous = &parse->state.subscriptions[parse->state.count - 1];
if (entry.source_node_id < previous->source_node_id
|| (entry.source_node_id == previous->source_node_id && entry.source_epoch <= previous->source_epoch)) return -1;
}
parse->state.subscriptions[parse->state.count++] = entry;
if (*p && (*p++ != ',' || !*p)) return -1;
}
}
return 0;
}
int stream_member_has_subscription(const struct stream_member_state *state, uint64_t source_id, uint64_t epoch) {
if (!state || state->count > STREAM_MEMBER_MAX_SUBSCRIPTIONS) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member subscription lookup invalid state"); return 0;
}
for (size_t i = 0; i < state->count; i++)
if (state->subscriptions[i].source_node_id == source_id && state->subscriptions[i].source_epoch == epoch) return 1;
return 0;
}
/* Отсутствие stream-ключей — обычный мембер, оно не создаёт экземпляр или буфер. */
int stream_member_parse(uint64_t group_id, uint64_t member_id, const char *userinfo, struct stream_member_state *state) {
if (!group_id || !member_id || !userinfo || !state || strlen(userinfo) > MS_USERINFO_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member parse invalid arguments group=%llx member=%llx",
(unsigned long long)group_id, (unsigned long long)member_id);
return STREAM_ERR_ARGUMENT;
}
struct member_parse parse = { .state.source = { .group_id = group_id, .source_node_id = member_id } };
if (json_flat_parse(userinfo, member_option, &parse) < 0
|| (parse.state.published && !stream_member_has_subscription(&parse.state, member_id, parse.state.source.source_epoch))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "member stream options malformed/inconsistent group=%llx member=%llx",
(unsigned long long)group_id, (unsigned long long)member_id);
return STREAM_ERR_FORMAT;
}
*state = parse.state;
return STREAM_OK;
}
/* Сериализовать канонические строки до обновления подписанного JSON-профиля. */
int stream_member_format(const struct stream_member_state *state, char source[MS_USERINFO_MAX + 1],
char subscriptions[MS_USERINFO_MAX + 1]) {
if (!state || !source || !subscriptions || state->count > STREAM_MEMBER_MAX_SUBSCRIPTIONS
|| (state->published && (stream_descriptor_validate(&state->source) != STREAM_OK
|| !stream_member_has_subscription(state, state->source.source_node_id, state->source.source_epoch)))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member format invalid state"); return STREAM_ERR_ARGUMENT;
}
char encoded_source[MS_USERINFO_MAX + 1] = "", encoded_subscriptions[MS_USERINFO_MAX + 1] = "";
if (state->published) snprintf(encoded_source, sizeof(encoded_source), "%016llx:%u:%u:%u",
(unsigned long long)state->source.source_epoch, state->source.n, state->source.r, state->source.block_size);
size_t used = 0;
for (size_t i = 0; i < state->count; i++) {
const struct stream_subscription_key *entry = &state->subscriptions[i];
if (!entry->source_node_id || !entry->source_epoch || (i &&
(entry->source_node_id < state->subscriptions[i - 1].source_node_id ||
(entry->source_node_id == state->subscriptions[i - 1].source_node_id
&& entry->source_epoch <= state->subscriptions[i - 1].source_epoch)))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member format invalid subscription index=%zu", i); return STREAM_ERR_ARGUMENT;
}
int length = snprintf(encoded_subscriptions + used, sizeof(encoded_subscriptions) - used, "%s%016llx:%016llx",
i ? "," : "", (unsigned long long)entry->source_node_id, (unsigned long long)entry->source_epoch);
if (length < 0 || (size_t)length >= sizeof(encoded_subscriptions) - used) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member subscriptions exceed limit count=%zu", state->count); return STREAM_ERR_ARGUMENT;
}
used += (size_t)length;
}
memcpy(source, encoded_source, strlen(encoded_source) + 1);
memcpy(subscriptions, encoded_subscriptions, used + 1);
return STREAM_OK;
}
/* Прочитать именно подписанный блок данного канала, а не общий online/name из nodes. */
int stream_member_load(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id,
struct stream_member_state *state, uint8_t public_key[32]) {
struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT || !inst->topo_sqlite_db || !state || !member_id) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member load invalid group=%llx member=%llx", (unsigned long long)group_id,
(unsigned long long)member_id); return STREAM_ERR_ARGUMENT;
}
for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member load invalid channel namespace group=%llx", (unsigned long long)group_id);
return STREAM_ERR_ARGUMENT;
}
char query[256];
snprintf(query, sizeof(query), "SELECT source,x25519_pubkey,ed25519_pubkey,join_sig,join_ts,update_sig,update_ts,"
"userinfo,signed_by,signature FROM \"peers_%s\" WHERE node_id=?", group->channel_id);
sqlite3_stmt *statement = NULL;
sqlite3 *db = inst->topo_sqlite_db;
if (sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member load query failed group=%llx: %s", (unsigned long long)group_id, sqlite3_errmsg(db));
return STREAM_ERR_STATE;
}
sqlite3_bind_int64(statement, 1, (sqlite3_int64)member_id);
int result = STREAM_WAIT, step = sqlite3_step(statement);
if (step == SQLITE_ROW && sqlite3_column_int(statement, 0) == 0
&& sqlite3_column_bytes(statement, 1) == 32 && sqlite3_column_bytes(statement, 2) == 32
&& sqlite3_column_bytes(statement, 3) == 64 && sqlite3_column_bytes(statement, 5) == 64
&& sqlite3_column_bytes(statement, 9) == 64 && sqlite3_column_int64(statement, 6) > 0) {
const uint8_t *x25519 = sqlite3_column_blob(statement, 1), *ed25519 = sqlite3_column_blob(statement, 2);
const char *userinfo = (const char *)sqlite3_column_text(statement, 7);
const uint8_t *join = sqlite3_column_blob(statement, 3), *signature = sqlite3_column_blob(statement, 5);
uint64_t version = (uint64_t)sqlite3_column_int64(statement, 6);
uint8_t body[MS_USERINFO_MAX + 73];
struct ms_member_rec record = { .node_id = member_id, .x25519 = x25519, .ed25519 = ed25519,
.join_sig = join, .join_ts = (uint64_t)sqlite3_column_int64(statement, 4), .update_sig = signature,
.update_ts = version, .userinfo = userinfo, .signed_by = (uint64_t)sqlite3_column_int64(statement, 8),
.signature = sqlite3_column_blob(statement, 9) };
int tree = member_sync_verify_signature(db, group->channel_id, &record);
int bytes = userinfo && sqlite3_column_bytes(statement, 7) <= MS_USERINFO_MAX
&& (size_t)sqlite3_column_bytes(statement, 7) == strlen(userinfo)
? member_sync_build_update_msg(join, version, userinfo, body, sizeof(body)) : -1;
if (tree < 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "member source trust pending group=%llx member=%llx", (unsigned long long)group_id,
(unsigned long long)member_id);
} else if (!tree || sc_derive_node_id_from_pubkey(x25519) != member_id || bytes < 0
|| sc_ed25519_verify(ed25519, body, (size_t)bytes, signature) != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "member profile signature/identity invalid group=%llx member=%llx version=%llu",
(unsigned long long)group_id, (unsigned long long)member_id, (unsigned long long)version);
result = STREAM_ERR_SIGNATURE;
} else {
struct stream_member_state parsed;
result = stream_member_parse(group_id, member_id, userinfo, &parsed);
if (result == STREAM_OK) { parsed.version = version; *state = parsed; if (public_key) memcpy(public_key, ed25519, 32); }
}
} else if (step == SQLITE_DONE) {
result = STREAM_ERR_NOT_MEMBER;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "member absent group=%llx member=%llx", (unsigned long long)group_id,
(unsigned long long)member_id);
} else if (step != SQLITE_ROW) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member load failed group=%llx: %s", (unsigned long long)group_id, sqlite3_errmsg(db));
result = STREAM_ERR_STATE;
}
sqlite3_finalize(statement);
return result;
}
/* Online относится к пути этой группы; транспорт или запись в общей БД не заменяют его. */
int stream_member_online(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id) {
struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return 0;
if (member_id == inst->node_id) return inst->core_started && group->local_node != NULL;
struct TOPO_GROUP_NODE *node = topo_node_find_by_id(group, member_id);
return node && topo_node_best_path(node) != NULL;
}
src/stream/stream_member.h
+50
View File
@@ -0,0 +1,50 @@
/* Строковые опции подписанного блока мембера, без отдельного протокола объявлений.
* stream_source = epoch(16 hex):n:r:RS-size; отсутствие/пустое значение — не публикует.
* stream_subscriptions = source(16 hex):epoch(16 hex),...; сортировка по source/epoch.
* Указанная в stream_source собственная трансляция также должна быть в subscriptions.
* load проверяет подпись профиля и принадлежность ключа источника, BGP-доступность — отдельно.
* parse/format не меняют выход при ошибке. Все вызовы — один uasync-поток. */
#ifndef STREAM_MEMBER_H
#define STREAM_MEMBER_H
#include "stream_proto.h"
#include "../chat/member_sync.h"
#ifdef __cplusplus
extern "C" {
#endif
#define STREAM_MEMBER_SOURCE "stream_source"
#define STREAM_MEMBER_SUBSCRIPTIONS "stream_subscriptions"
#define STREAM_MEMBER_MAX_SUBSCRIPTIONS (MS_USERINFO_MAX / 34)
struct UTUN_INSTANCE;
struct stream_subscription_key {
uint64_t source_node_id;
uint64_t source_epoch;
};
struct stream_member_state {
struct stream_descriptor source;
struct stream_subscription_key subscriptions[STREAM_MEMBER_MAX_SUBSCRIPTIONS];
size_t count;
uint64_t version;
int published;
};
int stream_member_parse(uint64_t group_id, uint64_t member_id, const char *userinfo, struct stream_member_state *state);
int stream_member_format(const struct stream_member_state *state, char source[MS_USERINFO_MAX + 1],
char subscriptions[MS_USERINFO_MAX + 1]);
int stream_member_has_subscription(const struct stream_member_state *state, uint64_t source_id, uint64_t epoch);
/* WAIT — нет подписанного профиля/подтверждения родителя; ошибка подписи/формата <0.
* NOT_MEMBER отличает удалённого участника от временно неполного профиля. */
int stream_member_load(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id,
struct stream_member_state *state, uint8_t public_key[SC_PUBKEY_SIZE]);
int stream_member_online(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id);
#ifdef __cplusplus
}
#endif
#endif
src/stream/stream_proto.h
+2 -1
View File
@@ -56,7 +56,8 @@ enum stream_result {
STREAM_ERR_CONFLICT = -5,
STREAM_ERR_STATE = -6,
STREAM_ERR_CRYPTO = -7,
STREAM_ERR_FEC = -8
STREAM_ERR_FEC = -8,
STREAM_ERR_NOT_MEMBER = -9
};
/* Короткие примитивы без доступа за границы: размер проверяет вызывающий parser. */