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

Reconcile group policy with members and prefer group signatures

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 15:48:34 +01:00
1 parent 94ae8231aa
commit b5d008ee87
2 files changed
+75 -16
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/group_policy.c
+5 -2
View File
@@ -9,6 +9,8 @@
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_group_connect.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
@@ -105,7 +107,7 @@ int group_policy_apply(sqlite3* db, const char* channel, const struct group_poli
struct gp_member {
uint64_t node, signer;
uint8_t x[32], valid, admin, previous, allowed, keep;
uint8_t x[32], ed[32], valid, admin, previous, allowed, keep;
size_t child, sibling; /* Индексы + 1, 0 — конец списка детей. */
char name[128];
};
@@ -146,6 +148,7 @@ static int gp_load(struct UTUN_INSTANCE* inst, const char* ch, struct gp_member*
.signed_by = p->signer, .signature = sqlite3_column_blob(st, 11)};
if (!m.x25519 || !m.ed25519 || !m.join_sig || !m.signature) { rc = SQLITE_NOMEM; goto done; }
memcpy(p->x, m.x25519, 32);
memcpy(p->ed, m.ed25519, 32);
int verified = member_sync_verify_local_record(db, ch, &m);
if (verified < 0) goto done;
int tree = member_sync_verify_signature(db, ch, &m);
@@ -292,7 +295,7 @@ int group_policy_command(struct UTUN_INSTANCE* inst, const char* channel, int al
for (size_t i = 0; i < count; i++) if (m[i].keep && m[i].signer) {
uint8_t signature[64];
if (member_sync_sign_pubkey(key, m[i].x, signature) ||
member_sync_put(inst, channel, m[i].node, m[i].x, NULL, NULL, 0, NULL, 0, NULL, NULL, NULL, 0, 0, signature) < 0) goto done;
member_sync_put(inst, channel, m[i].node, m[i].x, m[i].ed, NULL, 0, NULL, 0, NULL, NULL, NULL, 0, 0, signature) < 0) goto done;
}
if (policy.version == INT64_MAX) { error = "version_exhausted"; goto done; }
int64_t now = ntp_time_get_us(inst); uint64_t version = now > 0 ? (uint64_t)now : 1;
src/chat/member_sync.c
+70 -14
View File
@@ -2,6 +2,7 @@
#include "member_sync.h"
#include "db_sync.h"
#include "chat_feedback.h"
#include "group_policy.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
@@ -194,6 +195,17 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bucket_hash prepare failed ns=%s", MS_ID, ns); return -1; }
int count = 0;
if (!prefix64) {
struct group_policy policy; uint8_t wire[GROUP_POLICY_WIRE_SIZE];
if (group_policy_get(db, ns, &policy)) { sqlite3_finalize(stmt); return -1; }
if (policy.version) {
group_policy_encode(&policy, wire);
if (EVP_DigestUpdate(sha_ctx, wire, sizeof(wire)) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group policy hash failed ns=%s", ns); sqlite3_finalize(stmt); return -1;
}
count++;
}
}
int step;
while ((step = sqlite3_step(stmt)) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
@@ -261,9 +273,12 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte
sqlite3_bind_int(stmt, 1, after_valid);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after_id);
size_t off = 0;
if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; }
uint16_t cnt = 0; off += 2;
size_t off = 2 + GROUP_POLICY_WIRE_SIZE;
if (off > *len) { sqlite3_finalize(stmt); return -2; }
struct group_policy policy;
if (group_policy_get(db, ns, &policy)) { sqlite3_finalize(stmt); return -1; }
group_policy_encode(&policy, buf + 2);
uint16_t cnt = 0;
int step;
while ((step = sqlite3_step(stmt)) == SQLITE_ROW) {
@@ -732,7 +747,10 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
}
/* Подпись приглашения не зависит от версии профиля: подтверждаем уже сохранённого мембера. */
int signature_changed = has_local && has_signature && _sig_is_zero64(local_signature);
int same_join = has_local && (!m->join_sig ||
(m->join_ts == local_join_ts && !memcmp(m->join_sig, local_join_sig, 64)));
int signature_changed = same_join && has_signature &&
(_sig_is_zero64(local_signature) || m->signed_by < local_signed_by);
if (signature_changed) {
int verified = member_sync_verify_signature(db, ch_id, m);
if (verified == MS_VERIFY_ERROR) return MT_ERR_IO;
@@ -743,8 +761,9 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
if (block_a_changed) {
const uint8_t* jsig = m->join_sig ? m->join_sig : (has_local ? local_join_sig : NULL);
/* Нулевой блок на проводе означает отсутствие подписи и не стирает сохранённое подтверждение. */
uint64_t w_signed_by = has_signature ? m->signed_by : local_signed_by;
const uint8_t* w_signature = has_signature ? m->signature : (has_local ? local_signature : NULL);
int keep_signature = same_join && !_sig_is_zero64(local_signature) && (!has_signature || local_signed_by <= m->signed_by);
uint64_t w_signed_by = keep_signature ? local_signed_by : m->signed_by;
const uint8_t* w_signature = keep_signature ? local_signature : m->signature;
if (topo_node_sqlite_member_block_put(db, ch_id, m->node_id, jsig, m->join_ts,
m->update_sig, m->update_ts, m->x25519, m->ed25519,
m->userinfo ? m->userinfo : "", NULL, w_signed_by, w_signature) != 0) {
@@ -860,6 +879,9 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
if (sql_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record commit ch=%s rc=%d: %s", MS_ID, ch_id, sql_rc, sqlite3_errmsg(db)); goto error;
}
if (own && rc >= 0 && (rc & (MS_APPLY_CHANGED | MS_APPLY_FEEDBACK)))
if (from_peer == inst->node_id && group_policy_refresh(inst, ch_id))
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "local member committed; admission refresh failed ch=%s", ch_id);
if (own && rc >= 0 && (rc & (MS_APPLY_CHANGED | MS_APPLY_FEEDBACK)))
member_sync_record_committed(inst, ch_id, m->node_id, (unsigned)rc);
return rc;
@@ -1150,14 +1172,16 @@ static int member_decode(const uint8_t* p, size_t len, struct member_wire* w, si
static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len) {
struct UTUN_INSTANCE* inst = ctx;
sqlite3* db = _db(inst);
if (!db || len < 2) {
if (!db || len < 2 + GROUP_POLICY_WIRE_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid member page ns=%s bytes=%zu", MS_ID, ns, len); return MT_ERR_DATA;
}
uint16_t count = ((uint16_t)data[0] << 8) | data[1];
if (count > (len - 2) / 285) {
if (count > (len - 2 - GROUP_POLICY_WIRE_SIZE) / 285) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: impossible page count=%u bytes=%zu", MS_ID, count, len); return MT_ERR_DATA;
}
size_t off = 2;
struct group_policy policy;
if (group_policy_decode(db, ns, data + 2, &policy)) return MT_ERR_DATA;
size_t off = 2 + GROUP_POLICY_WIRE_SIZE;
for (unsigned i = 0; i < count; i++) {
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) {
@@ -1167,15 +1191,16 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, co
off += used;
}
if (off != len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: trailing page bytes ns=%s bytes=%zu", MS_ID, ns, len - off); return MT_ERR_DATA; }
if (!count) return 0;
uint8_t* changed = u_calloc(count, 1);
uint8_t* changed = u_calloc(count ? count : 1, 1);
if (!changed) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page allocation failed", MS_ID); return MT_ERR_IO; }
if (sqlite3_exec(db, "SAVEPOINT member_page", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page begin: %s", MS_ID, sqlite3_errmsg(db)); u_free(changed); return MT_ERR_IO;
}
int rc = 0;
off = 2;
int rc = group_policy_apply(db, ns, &policy);
if (rc > 0) rc = merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ns, 0) < 0 ? MT_ERR_IO : 0;
off = 2 + GROUP_POLICY_WIRE_SIZE;
for (unsigned i = 0; i < count; i++) {
if (rc < 0) break;
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) { rc = MT_ERR_DATA; break; }
if (!w.rec.join_sig && topo_node_sqlite_member_get_join(db, ns, w.rec.node_id, w.local_join, &w.rec.join_ts) == 0)
@@ -1193,7 +1218,8 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, co
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page cleanup: %s", MS_ID, sqlite3_errmsg(db));
}
if (rc == 0) {
off = 2;
merkle_sync_changed(inst, MT_DATA_MEMBERS, ns);
off = 2 + GROUP_POLICY_WIRE_SIZE;
for (unsigned i = 0; i < count; i++) {
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) { rc = MT_ERR_DATA; break; }
@@ -1249,12 +1275,38 @@ static int member_finish(void* ctx, const char* ns) {
return 0;
}
/* Допуск меняется только после общей контрольной точки политики и всех записей MEMBERS.
* Список для JOIN callbacks копируется: получатель вправе изменить группу/таблицу. */
static void member_policy_result(void* ctx, const char* ns, uint64_t peer, int result) {
(void)peer;
struct UTUN_INSTANCE* inst = ctx;
if (result != MT_OK || group_policy_refresh(inst, ns)) return;
struct ready_member { uint64_t node, signer; uint8_t x[32], signature[64]; };
struct ready_member* rows = NULL; size_t count = 0;
char sql[192]; sqlite3_stmt* st = NULL;
snprintf(sql, sizeof(sql), "SELECT node_id,signed_by,x25519_pubkey,signature FROM peers_%s WHERE admitted=1 AND deleted=0", ns);
int rc = sqlite3_prepare_v2(_db(inst), sql, -1, &st, NULL);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
if (sqlite3_column_bytes(st, 2) != 32 || sqlite3_column_bytes(st, 3) != 64) { rc = SQLITE_CORRUPT; break; }
if (count >= UINT32_MAX / sizeof(*rows) - 1) { rc = SQLITE_TOOBIG; break; }
struct ready_member* p = u_realloc(rows, (count + 1) * sizeof(*rows));
if (!p) { rc = SQLITE_NOMEM; break; } rows = p;
rows[count].node = (uint64_t)sqlite3_column_int64(st, 0); rows[count].signer = (uint64_t)sqlite3_column_int64(st, 1);
memcpy(rows[count].x, sqlite3_column_blob(st, 2), 32); memcpy(rows[count].signature, sqlite3_column_blob(st, 3), 64); count++;
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "admission callbacks read ns=%s rc=%d", ns, rc);
else for (size_t i = 0; i < count; i++) _fire_apply_cbk(inst, ns, rows[i].node, rows[i].x, rows[i].signer, rows[i].signature);
u_free(rows);
}
static const struct merkle_sync_data_ops g_member_ops = {
.update_bucket_hash = _member_update_bucket_hash,
.get_page = _member_get_page,
.apply_items = _member_apply_items,
.validate_peer = _member_validate_peer,
.finish = member_finish,
.on_result = member_policy_result,
};
/* ── Public API ── */
@@ -1269,6 +1321,9 @@ static int _rebuild_all_trees(struct UTUN_INSTANCE* inst) {
const char* table = (const char*)sqlite3_column_text(tables, 0);
if (!table || strncmp(table, "peers_", 6)) continue;
const char* ns = table + 6;
if (topo_node_sqlite_member_local_init(db, ns) || merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ns, 0) < 0) {
rc = SQLITE_ERROR; break;
}
uint64_t mask = merkle_sync_level_prefix(UINT64_MAX, MT_MAX_LEVEL);
char sql[256];
snprintf(sql, sizeof(sql), "SELECT DISTINCT (node_id & %lld) FROM \"%s\" WHERE source=0", (long long)mask, table);
@@ -1316,6 +1371,7 @@ int member_sync_init(struct UTUN_INSTANCE* inst) {
if (!db || !sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init requires database outside transaction", MS_ID); return -1;
}
if (group_policy_schema(db)) return -1;
int rc = merkle_sync_init(inst, ETCP_ID_MEMBER_SYNC, &g_member_ops, inst);
if (rc != 0) return rc;
/* очистка битых записей во всех каналах перед построением дерева */