Exercise DM reordering, rollback and signed acknowledgments

tests/test_dm_e2e.c

@@ -31,6 +31,7 @@
 #include <sqlite3.h>
 #define OPENSSL_API_COMPAT 0x10100000L
 #include <openssl/evp.h>
+#include <openssl/sha.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -219,6 +220,68 @@ static int dm_mail_count(struct UTUN_INSTANCE* inst) {
     return n;
 }
 
+/* Точная проверка durable исходящей очереди; ошибка SQL не маскируется нулём. */
+static int outbox_count(struct UTUN_INSTANCE* inst) {
+    sqlite3_stmt* st = NULL;
+    int count = -1;
+    if (sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT COUNT(*) FROM dm_outbox", -1, &st, NULL) == SQLITE_OK &&
+        sqlite3_step(st) == SQLITE_ROW) count = sqlite3_column_int(st, 0);
+    sqlite3_finalize(st);
+    return count;
+}
+
+/* Каноническое сообщение реального автора для проверок порядка, подписи и commit. */
+static size_t test_body(struct UTUN_INSTANCE* author, uint64_t seq, const char* text, uint8_t body[512]) {
+    uint64_t conv = strtoull(g_sh.conv_id, NULL, 10), ts = 1;
+    uint8_t key[32], nonce[DM_NONCE_SIZE];
+    if (dm_derive_content_key(author->my_keys.private_key, g_sh.x_pub[IDX_B], key) != 0) return 0;
+    dm_build_nonce(conv, author->node_id, seq, nonce);
+    memcpy(body, &conv, 8);
+    memcpy(body + 8, &seq, 8);
+    memcpy(body + 16, &ts, 8);
+    memcpy(body + 24, &author->node_id, 8);
+    body[32] = 4;
+    memcpy(body + 33, "text", 4);
+    size_t enc_len = 0;
+    if (dm_encrypt(key, nonce, (const uint8_t*)text, strlen(text), body + 39, &enc_len) != 0) return 0;
+    uint16_t n = (uint16_t)enc_len;
+    memcpy(body + 37, &n, 2);
+    if (sc_ed25519_sign(author->my_ed25519_privkey, body, 39 + enc_len, body + 39 + enc_len) != SC_OK) return 0;
+    return 39 + enc_len + 64;
+}
+
+/* Ошибочный commit, tamper, точные повторы и ACK чужого тела не теряют pending. */
+static int protocol_checks(struct UTUN_INSTANCE* A, struct UTUN_INSTANCE* B) {
+    uint8_t body[512], receipt[DM_RECEIPT_SIZE], again[DM_RECEIPT_SIZE];
+    size_t len = test_body(A, 100, "higher", body);
+    if (!len || dm_accept_message(B, body, len, receipt) != 0 ||
+        dm_accept_message(B, body, len, again) != 0 || memcmp(receipt, again, sizeof(receipt))) return -1;
+    body[len - 1] ^= 1;
+    if (dm_accept_message(B, body, len, again) == 0) return -1;
+    body[len - 1] ^= 1;
+    if (dm_accept_message(B, body, len - 1, again) == 0 ||
+        dm_verify_message(A, g_sh.nid[IDX_C], body, len) == 0) return -1;
+    len = test_body(A, 99, "lower", body);
+    if (!len || dm_accept_message(B, body, len, receipt) != 0 || !dm_msg_has(B, "lower")) return -1;
+    len = test_body(A, 99, "conflicting", body);
+    if (!len || dm_accept_message(B, body, len, again) == 0) return -1;
+    if (sqlite3_exec(B->topo_sqlite_db, "CREATE TEMP TRIGGER dm_fail BEFORE INSERT ON dm_messages"
+                     " WHEN NEW.seq=98 BEGIN SELECT RAISE(ABORT,'test commit failure'); END", NULL, NULL, NULL) != SQLITE_OK) return -1;
+    len = test_body(A, 98, "retry-after-db-error", body);
+    if (!len || dm_accept_message(B, body, len, again) == 0) return -1;
+    if (sqlite3_exec(B->topo_sqlite_db, "DROP TRIGGER dm_fail", NULL, NULL, NULL) != SQLITE_OK ||
+        dm_accept_message(B, body, len, again) != 0) return -1;
+    /* Правильная подпись B, правильные conv/seq, но хеш другого тела. */
+    uint64_t seq = 1;
+    memcpy(receipt + 16, &seq, 8);
+    if (sc_ed25519_sign(B->my_ed25519_privkey, receipt, 72, receipt + 72) != SC_OK ||
+        dm_accept_receipt(A, receipt) == 0 || outbox_count(A) != 1) return -1;
+    receipt[DM_RECEIPT_SIZE - 1] ^= 1;
+    if (dm_accept_receipt(A, receipt) == 0 || outbox_count(A) != 1) return -1;
+    DEBUG_INFO(DEBUG_CATEGORY_DM, "protocol checks passed: reorder, dedup, tamper, DB rollback, exact receipt");
+    return 0;
+}
+
 /* ── настройка канала и мемберов (однократно) ── */
 
 static void channel_put_shared(struct UTUN_INSTANCE* inst, int has_priv) {
@@ -318,7 +381,7 @@ static void dm_tick(void* arg) {
         break;
     case P_SEND1:
         if (dm_start(A, nb, g_sh.x_pub[IDX_B], g_sh.ed_pub[IDX_B], "B", CH_ID) != 0
-            || dm_send(A, g_sh.conv_id, "text", (const uint8_t*)"hello1", 6) != 0) {
+            || dm_send(A, g_sh.conv_id, "text", (const uint8_t*)"hello1", 6) != 0 || protocol_checks(A, B) != 0) {
             fprintf(stderr, "A: dm_start/send hello1 failed\n");
             t->result = 2;
         } else {
@@ -326,7 +389,7 @@ static void dm_tick(void* arg) {
         }
         break;
     case P_WAIT_MSG1:
-        if (dm_msg_has(B, "hello1")) t->phase = P_B_OFFLINE;
+        if (dm_msg_has(B, "hello1") && outbox_count(A) == 0 && dm_mail_count(C) == 0) t->phase = P_B_OFFLINE;
         break;
     case P_B_OFFLINE:
         utun_instance_destroy(B);
@@ -345,7 +408,20 @@ static void dm_tick(void* arg) {
         }
         break;
     case P_WAIT_MAIL:
-        if (dm_mail_count(C) > 0) t->phase = P_RECONNECT;
+        if (dm_mail_count(C) > 0) {
+            utun_instance_destroy(A);
+            t->inst[IDX_A] = NULL;
+            char cfg[512];
+            snprintf(cfg, sizeof(cfg), "%s/a.conf", getenv("UTUN_TEST_DIR"));
+            t->inst[IDX_A] = utun_instance_create(t->ua, cfg);
+            if (!t->inst[IDX_A] || utun_instance_init(t->inst[IDX_A]) != 0 || outbox_count(t->inst[IDX_A]) != 1) {
+                DEBUG_ERROR(DEBUG_CATEGORY_DM, "outbox restart failed");
+                t->result = 2;
+                uasync_stop(t->ua);
+                return;
+            }
+            t->phase = P_RECONNECT;
+        }
         break;
     case P_RECONNECT: {
         char cfg[512];
@@ -358,7 +434,7 @@ static void dm_tick(void* arg) {
         break;
     }
     case P_WAIT_B2:
-        if (dm_msg_has(B, "hello2") && dm_mail_count(C) == 0) t->phase = P_DONE;
+        if (dm_msg_has(B, "hello2") && dm_mail_count(C) == 0 && outbox_count(A) == 0) t->phase = P_DONE;
         break;
     case P_DONE:
         t->result = 1;