From 3195167ff3de0c9280e8edd2ad4c64171ba35b08 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 06:23:48 +0100 Subject: [PATCH] fix(chat): commit signed profile updates under one version lock --- src/chat/chat_profile.c | 259 +++++++++++++++++++++++++--------------- 1 file changed, 162 insertions(+), 97 deletions(-) diff --git a/src/chat/chat_profile.c b/src/chat/chat_profile.c index 953a3e31..b3141fb5 100644 --- a/src/chat/chat_profile.c +++ b/src/chat/chat_profile.c @@ -63,20 +63,14 @@ static int profile_string(struct profile_json* p, const char* text) { if (p->used >= MS_USERINFO_MAX) return -1; p->text[p->used++] = '"'; for (const unsigned char* c = (const unsigned char*)text; *c; c++) { - char escape = 0; - if (*c == '"' || *c == '\\') escape = *c; - if (*c == '\n') escape = 'n'; - if (*c == '\r') escape = 'r'; - if (*c == '\t') escape = 't'; - if (*c == '\b') escape = 'b'; - if (*c == '\f') escape = 'f'; - if ((*c < 32 && !escape) || p->used + (escape ? 2 : 1) >= MS_USERINFO_MAX) return -1; - if (escape) p->text[p->used++] = '\\'; - p->text[p->used++] = escape ? escape : *c; + char encoded[7]; size_t bytes = 1; encoded[0] = (char)*c; + if (*c == '"' || *c == '\\') { encoded[0] = '\\'; encoded[1] = (char)*c; bytes = 2; } + else if (*c < 32) { snprintf(encoded, sizeof(encoded), "\\u%04x", *c); bytes = 6; } + if (p->used + bytes >= MS_USERINFO_MAX) return -1; + memcpy(p->text + p->used, encoded, bytes); p->used += bytes; } if (p->used >= MS_USERINFO_MAX) return -1; - p->text[p->used++] = '"'; p->text[p->used] = 0; - return 0; + p->text[p->used++] = '"'; p->text[p->used] = 0; return 0; } static int profile_pair(struct profile_json* p, const char* key, const char* value) { @@ -140,84 +134,154 @@ error: return -1; } -/* Обновить профиль целиком одной подписанной записью, сохранив не затронутые поля. */ -static int profile_update_channel(struct UTUN_INSTANCE* inst, const char* ch, int avatar, uint64_t server, const char* file, - const struct chat_profile_option* options, size_t option_count) { - struct chat_core_ctx* cc = CC(inst); - if (!cc || !cc->initialized || !cc->db) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: update before initialization"); return -1; - } - uint64_t myid = cc->my_node_id; - char tbl[80]; peers_table_name(ch, tbl, sizeof(tbl)); - uint8_t join_sig[64]; uint64_t join_ts = 0; - if (topo_node_sqlite_member_get_join(cc->db, ch, myid, join_sig, &join_ts) != 0) { - DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: own signed member missing channel=%s", ch); return -1; - } - struct profile_json profile = { .text = "{", .used = 1, .avatar = avatar, .options = options, .option_count = option_count }; - if (profile_pair(&profile, "name", inst->name) < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: display name exceeds profile limit channel=%s", ch); return -1; - } - char query[256]; snprintf(query, sizeof(query), "SELECT update_ts,userinfo,update_sig FROM \"%s\" WHERE node_id=?", tbl); - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(cc->db, query, -1, &st, NULL) != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: query failed channel=%s: %s", ch, sqlite3_errmsg(cc->db)); return -1; - } - sqlite3_bind_int64(st, 1, (sqlite3_int64)myid); - if (sqlite3_step(st) != SQLITE_ROW) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: cannot read own member channel=%s: %s", ch, sqlite3_errmsg(cc->db)); - sqlite3_finalize(st); return -1; - } - uint64_t local_uts = (uint64_t)sqlite3_column_int64(st, 0); - int signed_profile = local_uts && sqlite3_column_bytes(st, 2) == 64 - && memcmp(sqlite3_column_blob(st, 2), (uint8_t[64]){0}, 64); - const char* old = (const char*)sqlite3_column_text(st, 1); - char before[MS_USERINFO_MAX + 1] = ""; - if (old && strlen(old) > MS_USERINFO_MAX) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: stored attributes exceed limit channel=%s", ch); - sqlite3_finalize(st); return -1; - } - if (old) memcpy(before, old, strlen(old) + 1); - if (old && old[0] && json_flat_parse(old, profile_copy_pair, &profile) < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: cannot preserve attributes channel=%s", ch); - sqlite3_finalize(st); return -1; - } - sqlite3_finalize(st); - for (size_t i = 0; i < option_count; i++) { - if (options[i].value && profile_pair(&profile, options[i].key, options[i].value) < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: option exceeds profile limit channel=%s key=%s", ch, options[i].key); - return -1; +struct profile_current { + uint8_t join[64], signature[64]; + uint64_t join_ts, version; + char json[MS_USERINFO_MAX + 1]; + int signed_profile; +}; + +/* 1 — свой полный member,0 — отсутствует/placeholder,-1 — ошибка. Cursor закрыт до подписи/изменения. */ +static int profile_read_current(struct UTUN_INSTANCE* inst, const char* channel, struct profile_current* out) { + sqlite3* db = CC(inst)->db; char table[80], sql[384]; peers_table_name(channel, table, sizeof(table)); + snprintf(sql, sizeof(sql), "SELECT join_sig,join_ts,update_ts,userinfo,update_sig,x25519_pubkey,ed25519_pubkey,source" + " FROM \"%s\" WHERE node_id=?", table); + sqlite3_stmt* st = NULL; const char* stage = "prepare"; int found = 0; + int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL); + if (rc == SQLITE_OK) { stage = "bind"; rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); } + if (rc == SQLITE_OK) { stage = "step"; rc = sqlite3_step(st); } + if (rc == SQLITE_DONE) goto done; + if (rc != SQLITE_ROW) goto error; + if (sqlite3_column_int(st, 7)) goto done; + stage = "fields"; rc = SQLITE_CORRUPT; + int bytes = sqlite3_column_bytes(st, 3), sig_bytes = sqlite3_column_bytes(st, 4); + const char* json = (const char*)sqlite3_column_text(st, 3); + const void *join = sqlite3_column_blob(st, 0), *signature = sqlite3_column_blob(st, 4); + const void *x = sqlite3_column_blob(st, 5), *ed = sqlite3_column_blob(st, 6); + if (sqlite3_column_type(st, 0) != SQLITE_BLOB || sqlite3_column_bytes(st, 0) != 64 || !join || + sqlite3_column_type(st, 1) != SQLITE_INTEGER || sqlite3_column_int64(st, 1) <= 0 || + sqlite3_column_type(st, 2) != SQLITE_INTEGER || sqlite3_column_int64(st, 2) < 0 || + sqlite3_column_type(st, 3) != SQLITE_TEXT || !json || bytes > MS_USERINFO_MAX || strlen(json) != (size_t)bytes || + sqlite3_column_type(st, 5) != SQLITE_BLOB || sqlite3_column_bytes(st, 5) != 32 || !x || + sqlite3_column_type(st, 6) != SQLITE_BLOB || sqlite3_column_bytes(st, 6) != 32 || !ed || + memcmp(x, inst->my_keys.public_key, 32) || memcmp(ed, inst->my_ed25519_pubkey, 32)) goto error; + out->join_ts = (uint64_t)sqlite3_column_int64(st, 1); out->version = (uint64_t)sqlite3_column_int64(st, 2); + if (out->version) { + if (sqlite3_column_type(st, 4) != SQLITE_BLOB || sig_bytes != 64 || !signature) goto error; + memcpy(out->signature, signature, 64); out->signed_profile = 1; + } else if (sig_bytes && (sig_bytes != 64 || !signature || memcmp(signature, (uint8_t[64]){0}, 64))) goto error; + memcpy(out->json, json, (size_t)bytes + 1); memcpy(out->join, join, 64); found = 1; +done: + stage = "finalize"; rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto error; + if (found && out->signed_profile) { + uint8_t body[MS_USERINFO_MAX + 73]; + int bytes = member_sync_build_update_msg(out->join, out->version, out->json, body, sizeof(body)); + if (bytes < 0 || sc_ed25519_verify(inst->my_ed25519_pubkey, body, (size_t)bytes, out->signature) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: stored signature invalid channel=%s", channel); return -1; } } + return found; +error: + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: current record ch=%s stage=%s rc=%d: %s", channel, stage, rc, sqlite3_errmsg(db)); + rc = sqlite3_finalize(st); + if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: current record cleanup finalize rc=%d", rc); + return -1; +} + +/* Одинаковые поля не меняют версию из-за порядка ключей или способа JSON escaping. */ +static int profile_equal(sqlite3* db, const char* a, const char* b) { + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "SELECT NOT EXISTS(SELECT key,value FROM json_each(?1) EXCEPT SELECT key,value FROM json_each(?2))" + " AND NOT EXISTS(SELECT key,value FROM json_each(?2) EXCEPT SELECT key,value FROM json_each(?1))", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, a, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, b, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int same = rc == SQLITE_ROW && sqlite3_column_int(st, 0), final = sqlite3_finalize(st); + if (rc != SQLITE_ROW || final != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: comparison rc=%d finalize=%d: %s", rc, final, sqlite3_errmsg(db)); return -1; + } + return same; +} + +/* Вызывается под собственной блокировкой;возвращает commit-флаги member_sync либо-1,без callbacks. */ +static int profile_update_locked(struct UTUN_INSTANCE* inst, const char* ch, int avatar, uint64_t server, const char* file, + const struct chat_profile_option* options, size_t option_count) { + struct profile_current current = {0}; + int found = profile_read_current(inst, ch, ¤t); + if (found <= 0) { + if (!found) DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: own full member unavailable channel=%s", ch); + return -1; + } + if (avatar && server && !member_files_member(inst, strtoull(ch, NULL, 10), server, 1)) { + DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: avatar server has no storage role channel=%s", ch); return -1; + } + struct profile_json profile = {.text = "{", .used = 1, .avatar = avatar, .options = options, .option_count = option_count}; + if (profile_pair(&profile, "name", inst->name) < 0 || + (current.signed_profile && chat_json_strings(CC(inst)->db, current.json, MS_USERINFO_MAX, profile_copy_pair, &profile))) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: cannot preserve/encode signed attributes channel=%s", ch); return -1; + } + for (size_t i = 0; i < option_count; i++) if (options[i].value && profile_pair(&profile, options[i].key, options[i].value) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: option exceeds limit channel=%s index=%zu", ch, i); return -1; + } if (avatar && server) { char id[17]; snprintf(id, sizeof(id), "%016llx", (unsigned long long)server); if (profile_pair(&profile, "avatar_server", id) < 0 || profile_pair(&profile, "avatar_file", file) < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: avatar exceeds profile limit channel=%s", ch); return -1; + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: avatar exceeds limit channel=%s", ch); return -1; } } - if (profile.used >= MS_USERINFO_MAX) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: attributes exceed profile limit channel=%s", ch); return -1; - } + if (profile.used >= MS_USERINFO_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: attributes exceed limit channel=%s", ch); return -1; } profile.text[profile.used++] = '}'; profile.text[profile.used] = 0; - if (signed_profile && !strcmp(before, profile.text)) { - DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "profile: unchanged channel=%s name=%s ts=%llu", ch, inst->name, - (unsigned long long)local_uts); return 0; + if (current.signed_profile) { + int equal = profile_equal(CC(inst)->db, current.json, profile.text); + if (equal < 0) return -1; + if (equal) { DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "profile: unchanged channel=%s ts=%llu", ch, (unsigned long long)current.version); return 0; } } - if (local_uts >= INT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: version exhausted channel=%s", ch); return -1; } - time_t seconds = ntp_time_get_seconds(inst); - uint64_t now = seconds > 0 ? (uint64_t)seconds : 0; - uint64_t update_ts = now > local_uts ? now : local_uts + 1; - uint8_t umsg[MS_USERINFO_MAX + 73], update_sig[64]; - int ulen = member_sync_build_update_msg(join_sig, update_ts, profile.text, umsg, sizeof(umsg)); - if (ulen < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: cannot build signature channel=%s", ch); return -1; } - if (sc_ed25519_sign(inst->my_ed25519_privkey, umsg, ulen, update_sig) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: signature failed channel=%s", ch); return -1; + if (current.version == INT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: version exhausted channel=%s", ch); return -1; } + time_t seconds = ntp_time_get_seconds(inst); uint64_t now = seconds > 0 ? (uint64_t)seconds : 0; + uint64_t version = now > current.version ? now : current.version + 1; + uint8_t body[MS_USERINFO_MAX + 73], signature[64]; + int bytes = member_sync_build_update_msg(current.join, version, profile.text, body, sizeof(body)); + if (bytes < 0 || sc_ed25519_sign(inst->my_ed25519_privkey, body, (size_t)bytes, signature) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: signing failed channel=%s", ch); return -1; } - int rc = member_sync_put(inst, ch, myid, inst->my_keys.public_key, inst->my_ed25519_pubkey, join_sig, join_ts, - update_sig, update_ts, profile.text, NULL, NULL, 0, 0, NULL); - if (rc < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: update failed channel=%s result=%d", ch, rc); return -1; } - DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: update my member ch=%s name=%s ts=%llu", CC_ID, ch, inst->name, - (unsigned long long)update_ts); - return 0; + int changes = member_sync_put(inst, ch, inst->node_id, inst->my_keys.public_key, inst->my_ed25519_pubkey, + current.join, current.join_ts, signature, version, profile.text, NULL, NULL, 0, 0, NULL); + if (changes < 0 || !(changes & MS_APPLY_CHANGED)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: new version not saved channel=%s flags=%d", ch, changes); return -1; + } + DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "profile: staged channel=%s ts=%llu bytes=%zu", ch, (unsigned long long)version, profile.used); + return changes; +} + +static int profile_update_channel(struct UTUN_INSTANCE* inst, const char* channel, int avatar, uint64_t server, const char* file, + const struct chat_profile_option* options, size_t option_count) { + struct chat_core_ctx* cc = inst ? CC(inst) : NULL; + if (!cc || !cc->initialized || !cc->db || cc->my_node_id != inst->node_id || !chat_channel_id(channel) || + !sqlite3_get_autocommit(cc->db)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: inactive core/invalid channel/external transaction"); return -1; + } + char ch[20]; memcpy(ch, channel, strlen(channel) + 1); + int rc = sqlite3_exec(cc->db, "BEGIN IMMEDIATE /* profile */", NULL, NULL, NULL); + if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: begin ch=%s rc=%d: %s", ch, rc, sqlite3_errmsg(cc->db)); return -1; } + int changes = profile_update_locked(inst, ch, avatar, server, file, options, option_count); + if (changes > 0) { + rc = sqlite3_exec(cc->db, "COMMIT /* profile */", NULL, NULL, NULL); + if (rc == SQLITE_OK) { + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "profile: committed ch=%s node=%016llx avatar=%d", + ch, (unsigned long long)inst->node_id, avatar); + member_sync_record_committed(inst, ch, inst->node_id, (unsigned)changes); return 0; + } + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: commit ch=%s rc=%d: %s", ch, rc, sqlite3_errmsg(cc->db)); changes = -1; + } + if (!sqlite3_get_autocommit(cc->db)) { + rc = sqlite3_exec(cc->db, "ROLLBACK", NULL, NULL, NULL); + if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: rollback ch=%s rc=%d: %s", ch, rc, sqlite3_errmsg(cc->db)); changes = -1; } + if (!sqlite3_get_autocommit(cc->db)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: transaction remains open ch=%s;requesting owner stop/DB close", ch); + utun_instance_stop(inst); changes = -1; + } + } + return changes < 0 ? -1 : 0; } /* Проверить заменяемые ключи до построения/подписи нового блока мембера. */ @@ -243,17 +307,24 @@ int chat_core_set_my_profile_options(struct UTUN_INSTANCE* inst, const char* cha } void chat_core_update_my_member(struct UTUN_INSTANCE* inst) { - struct chat_core_ctx* cc = CC(inst); - if (!cc || !cc->initialized || !cc->db) return; - sqlite3_stmt* cs = NULL; - if (sqlite3_prepare_v2(cc->db, "SELECT channel_id FROM channels", -1, &cs, NULL) != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: channel query failed: %s", sqlite3_errmsg(cc->db)); return; + struct chat_core_ctx* cc = inst ? CC(inst) : NULL; + if (!cc || !cc->initialized || !cc->db || !sqlite3_get_autocommit(cc->db)) { + DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: refresh requires active chat without external transaction"); return; } - while (sqlite3_step(cs) == SQLITE_ROW) { - const char* ch = (const char*)sqlite3_column_text(cs, 0); - if (ch) profile_update_channel(inst, ch, 0, 0, NULL, NULL, 0); + char** rows = NULL; int count = 0, columns = 0; + int rc = sqlite3_get_table(cc->db, "SELECT channel_id,length(CAST(channel_id AS BLOB)) FROM channels ORDER BY channel_id", + &rows, &count, &columns, NULL); + if (rc != SQLITE_OK || (count && columns != 2)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: channel scan rc=%d: %s", rc, sqlite3_errmsg(cc->db)); sqlite3_free_table(rows); return; } - sqlite3_finalize(cs); + for (int i = 0; i < count; i++) { + const char *ch = rows[(i + 1) * columns], *bytes = rows[(i + 1) * columns + 1]; + if (!chat_channel_id(ch) || !bytes || strlen(ch) != strtoul(bytes, NULL, 10)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: invalid stored channel index=%d", i); continue; + } + profile_update_channel(inst, ch, 0, 0, NULL, NULL, 0); + } + sqlite3_free_table(rows); } int chat_core_set_my_avatar(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t server, const char* file) { @@ -264,19 +335,13 @@ int chat_core_set_my_avatar(struct UTUN_INSTANCE* inst, const char* ch_id, uint6 for (int i = 0; i < 64; i++) if (!((file[i] >= '0' && file[i] <= '9') || (file[i] >= 'a' && file[i] <= 'f'))) { DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: invalid avatar hash"); return -1; } - uint64_t group = strtoull(ch_id, NULL, 10); - if (!member_files_member(inst, group, server, 1)) { - DEBUG_WARN(DEBUG_CATEGORY_CHAT, "profile: avatar server has no storage role channel=%s", ch_id); return -1; - } } - int rc = profile_update_channel(inst, ch_id, 1, server, file, NULL, 0); - if (!rc) DEBUG_INFO(DEBUG_CATEGORY_CHAT, "profile: avatar published channel=%s server=%016llx file=%s", - ch_id, (unsigned long long)server, server ? file : "none"); - return rc; + return profile_update_channel(inst, ch_id, 1, server, file, NULL, 0); } void chat_core_set_my_avatar_trampoline(void* arg) { struct chat_avatar_req* q = arg; + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "profile: null avatar request"); return; } char ch[32]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)q->group); int rc = chat_core_set_my_avatar(q->inst, ch, q->server, q->file); uint8_t data[17]; mf_put64(data, q->id); mf_put64(data + 8, q->group); data[16] = rc ? 1 : 0;