diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index 8edb9f45..bcf701c0 100644 --- a/src/chat/merkle_sync.c +++ b/src/chat/merkle_sync.c @@ -11,10 +11,13 @@ #include #include -/* Wire v3: svc:1, namespace:8, type:1, round:8, request:4, dataset:1, payload. +/* Wire v4: svc:1, namespace:8, type:1, round:8, request:4, dataset:1, + * round_id:16, accept_id:16, payload. round — диагностический счётчик; + * пара случайных идентификаторов связывает запрос с текущим handshake обеих сторон. * Все целые заголовка big-endian. Один запрос на проход и один исходящий пакет на сессию. * Меньший node_id ведёт раунд: pull -> TURN -> reverse pull -> CHECK -> DONE. */ -#define MS_HEADER 23 +#define MS_HEADER 55 +#define MS_TOKEN_SIZE 16 #define MS_QUERY_SIZE (10 + MT_KEY_SIZE) #define MS_PAGE_HEADER (1 + MT_KEY_SIZE) #define MS_PROGRESS_TIMEOUT_TB (30u * 10000u) @@ -53,6 +56,7 @@ struct ms_session { int failure_error; struct ms_request* requests; uint64_t ns_id, peer, round, revision, verified_revision; + uint8_t round_id[MS_TOKEN_SIZE], accept_id[MS_TOKEN_SIZE]; char ns[21], tree_ns[25]; int state, leader, dirty, depth, after_valid, waiting; uint32_t request_id, remote_request_id; @@ -77,6 +81,16 @@ static void ms_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8 static uint32_t ms_read32(const uint8_t* p) { uint32_t n; memcpy(&n, p, 4); return ntohl(n); } static void ms_write32(uint8_t* p, uint32_t n) { n = htonl(n); memcpy(p, &n, 4); } +static int ms_token_empty(const uint8_t token[MS_TOKEN_SIZE]) { + static const uint8_t zero[MS_TOKEN_SIZE]; + return !memcmp(token, zero, sizeof(zero)); +} + +static int ms_new_token(uint8_t token[MS_TOKEN_SIZE]) { + if (random_bytes(token, MS_TOKEN_SIZE) == 0 && !ms_token_empty(token)) return 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: handshake RNG failed"); return MT_ERR_IO; +} + static int ms_namespace(const char* ns, uint64_t* id) { if (!ns || !ns[0] || strlen(ns) > 20) return -1; for (const char* p = ns; *p; p++) if (*p < '0' || *p > '9') return -1; @@ -245,6 +259,9 @@ static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8 ms_write64(e->dgram + 10, type == MS_WAKE ? 0 : s->round); ms_write32(e->dgram + 18, request); e->dgram[22] = s->dataset->id; + memset(e->dgram + 23, 0, 2 * MS_TOKEN_SIZE); + if (type != MS_WAKE) memcpy(e->dgram + 23, s->round_id, MS_TOKEN_SIZE); + if (type != MS_WAKE && type != MS_BEGIN) memcpy(e->dgram + 39, s->accept_id, MS_TOKEN_SIZE); if (len) memcpy(e->dgram + MS_HEADER, payload, len); s->tx = e; /* Transport normalizer configures deferred waiters. Не вызываем callback рекурсивно. */ @@ -328,9 +345,13 @@ static int ms_walk_next(struct ms_session* s) { } static int ms_begin(struct ms_session* s) { + uint8_t token[MS_TOKEN_SIZE]; + if (ms_new_token(token) < 0) return MT_ERR_IO; ms_abort_pull(s); s->round = ++s->ms->round_seq; if (!s->round) return -1; + memcpy(s->round_id, token, sizeof(token)); + memset(s->accept_id, 0, sizeof(s->accept_id)); s->request_id = s->remote_request_id = 0; s->state = MS_BEGIN_WAIT; s->dirty = 0; s->waiting = 0; DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: begin dataset=%u ns=%s peer=%016llx round=%llu", @@ -512,13 +533,15 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) { dataset, ns, (unsigned long long)s->peer, (unsigned long long)round, type, request, len, s->state); int rc = 0, notify = 0; if (s->state == MS_FAILING) goto done; - if (type != MS_WAKE && type != MS_BEGIN && round != s->round) { - DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard old round=%llu current=%llu", (unsigned long long)round, - (unsigned long long)s->round); + if (type != MS_WAKE && type != MS_BEGIN && + (round != s->round || memcmp(d + 23, s->round_id, MS_TOKEN_SIZE) || ms_token_empty(s->round_id) || + (type != MS_READY && (memcmp(d + 39, s->accept_id, MS_TOKEN_SIZE) || ms_token_empty(s->accept_id))))) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard stale handshake ns=%s round=%llu current=%llu type=%u", + ns, (unsigned long long)round, (unsigned long long)s->round, type); goto done; } if (type == MS_WAKE) { - if (!s->leader || round || request || len) rc = MT_ERR_PROTOCOL; + if (!s->leader || round || request || len || !ms_token_empty(d + 23) || !ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL; else { /* Follower мог отменить и заново подключить namespace. Старый запрос уже некому обслужить. */ ms_discard_tx(s); @@ -526,19 +549,28 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) { rc = ms_begin(s); } } else if (type == MS_BEGIN) { - if (s->leader || !round || request || len) rc = MT_ERR_PROTOCOL; - else if (round <= s->round) { + if (s->leader || !round || request || len || ms_token_empty(d + 23) || !ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL; + else if (!memcmp(d + 23, s->round_id, MS_TOKEN_SIZE)) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard duplicate BEGIN ns=%s", ns); } else { + uint8_t token[MS_TOKEN_SIZE]; + if (ms_new_token(token) < 0) { rc = MT_ERR_IO; goto done; } ms_discard_tx(s); ms_abort_pull(s); + memcpy(s->round_id, d + 23, MS_TOKEN_SIZE); + memcpy(s->accept_id, token, sizeof(token)); s->round = round; s->state = MS_SERVE; s->waiting = 0; s->request_id = s->remote_request_id = 0; s->dirty = 0; rc = ms_send(s, MS_READY, 0, NULL, 0) < 0 ? MT_ERR_IO : 0; } } else if (type == MS_READY) { - if (!s->leader || s->state != MS_BEGIN_WAIT || request || len) rc = MT_ERR_PROTOCOL; - else rc = ms_pull_start(s); + if (!s->leader || request || len || ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL; + else if (s->state == MS_BEGIN_WAIT) { + memcpy(s->accept_id, d + 39, MS_TOKEN_SIZE); + rc = ms_pull_start(s); + } else if (!memcmp(d + 39, s->accept_id, MS_TOKEN_SIZE)) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard duplicate READY ns=%s", ns); + } else rc = MT_ERR_PROTOCOL; } else if (type == MS_QUERY) rc = ms_serve_query(s, request, p, len); else if (type == MS_HASHES) rc = ms_receive_hashes(s, request, p, len); else if (type == MS_PAGE) rc = ms_receive_page(s, request, p, len); @@ -629,7 +661,7 @@ int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id, const struct me if (merkle_sync_register(inst, MT_DATA_MEMBERS, ops, data_ctx) < 0) { inst->msync = NULL; u_free(ms); return -1; } etcp_bind(inst, svc_id, ms_receive); etcp_add_conn_status_cbk(inst, ms_conn_status, ms); - DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: initialized service=%u protocol=3", svc_id); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: initialized service=%u protocol=4", svc_id); return 0; } diff --git a/src/chat/merkle_sync.h b/src/chat/merkle_sync.h index edf3862a..4d3d1959 100644 --- a/src/chat/merkle_sync.h +++ b/src/chat/merkle_sync.h @@ -2,6 +2,8 @@ * Движок сверяет хеши и передаёт страницы изменившихся листьев; модель данных * через data_ops читает записи, проверяет подписи/права и сохраняет изменения. * Сессия относится к (пир, dataset, namespace), отправка учитывает backpressure. + * Wire v4 привязывает запросы к свежим round_id и accept_id обеих сторон; + * отменённая сессия не может подтвердить новый запрос прежними пакетами. * Все операции — в потоке uasync; один движок на UTUN_INSTANCE. */ #ifndef MERKLE_SYNC_H #define MERKLE_SYNC_H