Cover terminal media registration failures and relay admission lifetime

src/media_delivery/media_download.c

@@ -91,7 +91,7 @@ static struct media_download_peer* md_dl_get_peer(struct media_download* dl, uin
     return p;
 }
 
-static void md_mkdir_parent(const char* filepath) {
+static int md_mkdir_parent(const char* filepath) {
     char path[1024]; snprintf(path, sizeof(path), "%s", filepath);
     char* p = path;
     if (((p[0] >= 'A' && p[0] <= 'Z') || (p[0] >= 'a' && p[0] <= 'z')) && p[1] == ':')
@@ -101,10 +101,11 @@ static void md_mkdir_parent(const char* filepath) {
         *p = '\0';
         if (utun_mkdir(path, 0755) != 0 && errno != EEXIST) {
             DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: mkdir(%s) failed: %s", MDL_ID, path, strerror(errno));
-            *p = '/'; return;
+            *p = '/'; return -1;
         }
         *p = '/'; p++;
     }
+    return 0;
 }
 
 /* ── send helpers ── */
@@ -594,7 +595,7 @@ void media_download_handle_chunk(struct UTUN_INSTANCE* inst, uint64_t group_id,
                    MDL_ID, ch->chunk, ch->offset, b->bytes_received, ch->data_len);
         md_dl_reject_block(dl, (int)ch->chunk); return;
     }
-    FILE* file = b->source ? file_io_open_update(b->source->storage.temporary) : NULL;
+    FILE* file = b->source && !b->source->failed ? file_io_open_update(b->source->storage.temporary) : NULL;
     int failed = !file;
     if (file) {
         if (fseeko(file, (off_t)ch->offset, SEEK_SET) || fwrite(data + sizeof(*ch), 1, ch->data_len, file) != ch->data_len)
@@ -707,6 +708,10 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst, uint64_t group_id, u
     int signature = sc_ed25519_verify(dl->author_pubkey, buffer, b->expected_size, dl->block_sigs + (size_t)bi * 64);
     u_free(buffer);
     if (signature != SC_OK) {
+        if (signature != SC_ERR_AUTH_FAILED) {
+            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: local signature verification failed block=%d rc=%d", MDL_ID, bi, signature);
+            md_dl_finish(dl, -1); return;
+        }
         DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: DONE author signature invalid block=%d rc=%d", MDL_ID, bi, signature);
         md_dl_reject_block(dl, bi); return;
     }
@@ -871,8 +876,8 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id,
         media_delivery_release(inst, dl->destination);
         u_free(dl->blocks); u_free(dl->block_sigs); queue_entry_free(qe); return -1;
     }
-    md_mkdir_parent(dl->dest_path);
     queue_data_put_with_index(md->downloads, qe);
+    if (md_mkdir_parent(dl->dest_path)) { md_dl_finish(dl, -1); return 0; }
 
     md_dl_collect_supernodes(inst, dl);
 

tests/test_media_blocks.c

@@ -385,6 +385,57 @@ static void test_destinations(void) {
     puts("normalized destination reservation is shared by incoming and outgoing writers and released on cancel: PASS");
 }
 
+static int deny_registration(void* arg, int action, const char* table, const char* column, const char* db, const char* trigger) {
+    (void)arg; (void)column; (void)db; (void)trigger;
+    return action == SQLITE_INSERT && !strcmp(table, "media_files") ? SQLITE_DENY : SQLITE_OK;
+}
+
+static void test_registration_failure_and_keys(void) {
+    struct fixture f = ready(4); old_destination(f.client);
+    struct packet chunk = chunk_packet(f.client, 0, 4); receive(f.client, &chunk);
+    REQUIRE(!sqlite3_set_authorizer(f.client->topo_sqlite_db, deny_registration, NULL));
+    struct packet done = done_packet(f.client); receive(f.client, &done);
+    REQUIRE(!sqlite3_set_authorizer(f.client->topo_sqlite_db, NULL, NULL));
+    REQUIRE(!download(f.client) && done_count == 1 && done_err == -1 && !f.client->md.destinations);
+    char path[512]; snprintf(path, sizeof(path), "%s/result.bin", directory);
+    uint8_t hash[32]; REQUIRE(!ma_sha256_file(path, hash) && !memcmp(hash, f.metadata.content_hash, 32));
+    sqlite3_stmt* statement = NULL;
+    REQUIRE(!sqlite3_prepare_v2(f.client->topo_sqlite_db,
+        "SELECT (SELECT COUNT(*) FROM media_files),(SELECT COUNT(*) FROM media_preparations)", -1, &statement, NULL));
+    REQUIRE(sqlite3_step(statement) == SQLITE_ROW && !sqlite3_column_int(statement, 0) && sqlite3_column_int(statement, 1) == 1);
+    REQUIRE(!sqlite3_finalize(statement));
+    for (int variant = 0; variant < 4; variant++) {
+        const char* values[] = {"NULL", "zeroblob(31)", "zeroblob(33)", "x''"};
+        char sql[128]; snprintf(sql, sizeof(sql), "UPDATE nodes SET ed25519_pubkey=%s", values[variant]);
+        REQUIRE(!sqlite3_exec(f.client->topo_sqlite_db, sql, NULL, NULL, NULL));
+        REQUIRE(media_download_start(f.client, GROUP, &f.metadata, path, directory, SERVER, completed, NULL, NULL, NULL) == -1);
+        REQUIRE(done_count == 1 && !f.client->md.destinations);
+    }
+    finish(&f);
+    puts("registration failure retains published/prepared file; missing and malformed author keys reject before callbacks: PASS");
+}
+
+static void test_relay_limit_and_idle(void) {
+    struct fixture f = ready(1); capture_blocks = 1;
+    struct media_download* dl = download(f.client); struct media_block_source* source = dl->blocks[0].source;
+    for (int i = 0; i < MD_MAX_RELAY_DOWNSTREAM; i++) {
+        struct packet request = request_packet(dl, 0x300 + i); receive(f.client, &request);
+    }
+    REQUIRE(source->refs == 1 + MD_MAX_RELAY_DOWNSTREAM && !f.client->md.active_streams);
+    struct packet overflow = request_packet(dl, 0x900); receive(f.client, &overflow);
+    REQUIRE(packet_count == 1 && packets[0].data[0] == MEDIA_SUBCMD_BLOCK_RELAY_FULL);
+    const struct media_pkt_block_relay_full* full = (const struct media_pkt_block_relay_full*)packets[0].data;
+    REQUIRE(full->num_relay_nodes == MD_MAX_RELAY_DOWNSTREAM && packets[0].len == sizeof(*full) + MD_MAX_RELAY_DOWNSTREAM * 8);
+    REQUIRE(!memcmp(full->transfer_id, ((struct media_pkt_block_req*)overflow.data)->transfer_id, 16));
+    packet_count = 0;
+    f.client->md.dl_stall_timeout_tb = 1000;
+    uint64_t deadline = get_time_tb() + 5000;
+    while (f.client->md.streams && get_time_tb() < deadline) uasync_poll(ua, 5);
+    REQUIRE(!f.client->md.streams && source->refs == 1 && download(f.client) == dl && !done_count);
+    finish(&f);
+    puts("relay admission permits five readers, echoes redirect attempt and releases idle readers: PASS");
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
     if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG);
@@ -398,7 +449,7 @@ int main(void) {
     ua = uasync_create(); REQUIRE(ua);
     snprintf(directory, sizeof(directory), "/tmp/utun-blocks-XXXXXX"); REQUIRE(!test_mkdtemp(directory));
     test_identity_and_bounds(); test_rejected_attempts(); test_io_failures(); test_reentrant_cancel();
-    test_growing_relay(); test_relay_cancel(); test_destinations();
+    test_growing_relay(); test_relay_cancel(); test_destinations(); test_registration_failure_and_keys(); test_relay_limit_and_idle();
     REQUIRE(!rmdir(directory)); uasync_destroy(ua, 0);
     puts("media block regressions: PASS"); return 0;
 }