254 lines
15 KiB
C
254 lines
15 KiB
C
#include <stdio.h>
|
|
#include <string.h>
|
|
|
|
#include "../../lib/debug_config.h"
|
|
#include "../../lib/json_flat.h"
|
|
#include "../../lib/mem.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, audio;
|
|
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++ != ':'
|
|
|| member_decimal(&p, 1, &audio) != 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;
|
|
parse->state.audio_enabled = (int)audio;
|
|
} 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 = {0};
|
|
if (member_hex(&p, &entry.source_node_id) != STREAM_OK || *p++ != ':'
|
|
|| member_hex(&p, &entry.source_epoch) != STREAM_OK) return -1;
|
|
if (*p == ':') { p++; if (*p++ != 'r') return -1; entry.relay_only = 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->audio_enabled != 0 && state->audio_enabled != 1) || (!state->published && state->audio_enabled)
|
|
|| (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:%d",
|
|
(unsigned long long)state->source.source_epoch, state->source.n, state->source.r, state->source.block_size, state->audio_enabled);
|
|
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%s",
|
|
i ? "," : "", (unsigned long long)entry->source_node_id, (unsigned long long)entry->source_epoch,
|
|
entry->relay_only ? ":r" : "");
|
|
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;
|
|
}
|
|
|
|
/* Роль читается по ID из блока владельца канала, без загрузки остальных участников и снимков узлов. */
|
|
int stream_member_supernode(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 (!inst->topo_sqlite_db || !group->channel_id[0]) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member role missing DB/channel group=%llx", (unsigned long long)group_id); return -1;
|
|
}
|
|
for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member role invalid channel namespace group=%llx", (unsigned long long)group_id); return -1;
|
|
}
|
|
char query[192];
|
|
snprintf(query, sizeof(query), "SELECT adm_tags FROM \"peers_%s\" WHERE node_id=? AND deleted=0", group->channel_id);
|
|
sqlite3 *db = inst->topo_sqlite_db; sqlite3_stmt *statement = NULL;
|
|
int result = -1, rc = sqlite3_prepare_v2(db, query, -1, &statement, NULL);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(statement, 1, (sqlite3_int64)member_id);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(statement);
|
|
if (rc == SQLITE_ROW) {
|
|
const char *tags = (const char *)sqlite3_column_text(statement, 0); char value[16];
|
|
result = tags && json_flat_get(tags, "supernode", value, sizeof(value)) == 0 && !strcmp(value, "yes");
|
|
} else if (rc == SQLITE_DONE) result = 0;
|
|
else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member role query failed group=%llx member=%llx rc=%d: %s",
|
|
(unsigned long long)group_id, (unsigned long long)member_id, rc, sqlite3_errmsg(db));
|
|
sqlite3_finalize(statement);
|
|
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "member role group=%llx member=%llx supernode=%d",
|
|
(unsigned long long)group_id, (unsigned long long)member_id, result);
|
|
return result;
|
|
}
|