From b6f80668d14e404166e183c50bdf80d566c1eeac Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 10 Oct 2026 23:33:37 +0200 Subject: [PATCH] Define signed stream member options and BGP eligibility --- src/stream/stream_member.c | 219 +++++++++++++++++++++++++++++++++++++ src/stream/stream_member.h | 50 +++++++++ src/stream/stream_proto.h | 3 +- 3 files changed, 271 insertions(+), 1 deletion(-) create mode 100644 src/stream/stream_member.c create mode 100644 src/stream/stream_member.h diff --git a/src/stream/stream_member.c b/src/stream/stream_member.c new file mode 100644 index 00000000..28c0380d --- /dev/null +++ b/src/stream/stream_member.c @@ -0,0 +1,219 @@ +#include +#include + +#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; +} diff --git a/src/stream/stream_member.h b/src/stream/stream_member.h new file mode 100644 index 00000000..fd46b2c4 --- /dev/null +++ b/src/stream/stream_member.h @@ -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 diff --git a/src/stream/stream_proto.h b/src/stream/stream_proto.h index d77fb601..f75570a6 100644 --- a/src/stream/stream_proto.h +++ b/src/stream/stream_proto.h @@ -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. */