From a4cf9f09cfc57c5df2ffa0b97129e496ba69f3a9 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Wed, 17 Jun 2026 13:54:54 +0000 Subject: [PATCH 1/4] Remove indexing::LFS ringbuffer types, replace with ccf::tasks --- src/enclave/enclave.h | 4 +- src/host/lfs_file_handler.h | 98 ----------- src/host/run.cpp | 7 - src/indexing/enclave_lfs_access.h | 252 +++++++++++++++------------- src/indexing/lfs_ringbuffer_types.h | 35 ---- src/indexing/test/indexing.cpp | 28 +--- src/indexing/test/lfs.cpp | 116 +++++-------- 7 files changed, 186 insertions(+), 354 deletions(-) delete mode 100644 src/host/lfs_file_handler.h delete mode 100644 src/indexing/lfs_ringbuffer_types.h diff --git a/src/enclave/enclave.h b/src/enclave/enclave.h index 074aa1eab0a..40ae192bc0b 100644 --- a/src/enclave/enclave.h +++ b/src/enclave/enclave.h @@ -124,7 +124,7 @@ namespace ccf context->install_subsystem(indexer); lfs_access = std::make_shared( - writer_factory->create_writer_to_outside()); + ccf::tasks::get_main_job_board()); context->install_subsystem(lfs_access); context->install_subsystem(std::make_shared(*node)); @@ -270,8 +270,6 @@ namespace ccf // reconstruct oversized messages sent to the enclave oversized::FragmentReconstructor fr(bp.get_dispatcher()); - lfs_access->register_message_handlers(bp.get_dispatcher()); - DISPATCHER_SET_MESSAGE_HANDLER( bp, AdminMessage::stop, [this, &bp](const uint8_t*, size_t) { bp.set_finished(); diff --git a/src/host/lfs_file_handler.h b/src/host/lfs_file_handler.h deleted file mode 100644 index 338869cd940..00000000000 --- a/src/host/lfs_file_handler.h +++ /dev/null @@ -1,98 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the Apache 2.0 License. -#pragma once - -#include "ds/files.h" -#include "ds/messaging.h" -#include "indexing/lfs_ringbuffer_types.h" -#include "time_bound_logger.h" - -#include - -namespace asynchost -{ - struct LFSFileHandler - { - const std::filesystem::path root_dir = ".index"; - - ringbuffer::WriterPtr writer; - - LFSFileHandler(ringbuffer::WriterPtr&& w) : writer(w) - { - if (std::filesystem::is_directory(root_dir)) - { - LOG_INFO_FMT("Clearing contents from existing directory {}", root_dir); - TimeBoundLogger log_if_slow(fmt::format( - "Clearing LFS index directory - remove_all({})", root_dir)); - std::filesystem::remove_all(root_dir); - } - - { - TimeBoundLogger log_if_slow(fmt::format( - "Creating LFS index directory - create_directory({})", root_dir)); - if (!std::filesystem::create_directory(root_dir)) - { - throw std::logic_error( - fmt::format("Could not create directory: {}", root_dir)); - } - } - } - - void register_message_handlers(messaging::RingbufferDispatcher& disp) - { - DISPATCHER_SET_MESSAGE_HANDLER( - disp, - ccf::indexing::LFSMsg::store, - [&](const uint8_t* data, size_t size) { - auto [key, encrypted] = - ringbuffer::read_message(data, size); - - const auto target_path = root_dir / key; - { - TimeBoundLogger log_if_slow(fmt::format( - "Writing LFS file ({} bytes) - {}", - encrypted.size(), - target_path)); - LOG_TRACE_FMT( - "Writing {} byte file to {}", encrypted.size(), target_path); - files::dump(encrypted, target_path); - } - }); - - DISPATCHER_SET_MESSAGE_HANDLER( - disp, - ccf::indexing::LFSMsg::get, - [&](const uint8_t* data, size_t size) { - auto [key] = - ringbuffer::read_message(data, size); - - const auto target_path = root_dir / key; - if (std::filesystem::is_regular_file(target_path)) - { - TimeBoundLogger log_if_slow( - fmt::format("Reading LFS file - ifstream({})", target_path)); - std::ifstream f(target_path, std::ios::binary); - f.seekg(0, f.end); - const auto file_size = f.tellg(); - LOG_TRACE_FMT( - "Reading {} byte file from {}", - static_cast(file_size), - target_path); - f.seekg(0, f.beg); - - ccf::indexing::LFSEncryptedContents blob(file_size); - f.read(reinterpret_cast(blob.data()), blob.size()); - f.close(); - RINGBUFFER_WRITE_MESSAGE( - ccf::indexing::LFSMsg::response, writer, key, blob); - } - else - { - LOG_TRACE_FMT("File {} not found", target_path); - RINGBUFFER_WRITE_MESSAGE( - ccf::indexing::LFSMsg::not_found, writer, key); - } - }); - } - }; -} diff --git a/src/host/run.cpp b/src/host/run.cpp index c9a046e5a38..e3f31d2f0c5 100644 --- a/src/host/run.cpp +++ b/src/host/run.cpp @@ -33,7 +33,6 @@ #include "host/files_cleanup_timer.h" #include "http/curl.h" #include "json_schema.h" -#include "lfs_file_handler.h" #include "node_connections.h" #include "pal/quote_generation.h" #include "rpc_connections.h" @@ -639,12 +638,6 @@ namespace ccf config.files_cleanup.max_committed_ledger_chunks); } - // handle LFS-related messages from the enclave - asynchost::LFSFileHandler lfs_file_handler( - writer_factory.create_writer_to_inside()); - lfs_file_handler.register_message_handlers( - buffer_processor.get_dispatcher()); - // Setup node-to-node connections auto [node_host, node_port] = cli::validate_address(config.network.node_to_node_interface.bind_address); diff --git a/src/indexing/enclave_lfs_access.h b/src/indexing/enclave_lfs_access.h index 18764299c82..8652193f53f 100644 --- a/src/indexing/enclave_lfs_access.h +++ b/src/indexing/enclave_lfs_access.h @@ -7,12 +7,17 @@ #include "ccf/crypto/symmetric_key.h" #include "ccf/ds/hex.h" #include "ccf/pal/locking.h" -#include "ds/messaging.h" +#include "ds/files.h" +#include "ds/internal_logger.h" +#include "ds/serialized.h" +#include "host/time_bound_logger.h" #include "indexing/lfs_interface.h" -#include "indexing/lfs_ringbuffer_types.h" +#include "tasks/ordered_tasks.h" +#include +#include +#include #include -#include #include // Uncomment to disable encryption and obfuscation, writing cache content @@ -87,7 +92,14 @@ namespace ccf::indexing std::unordered_map pending; ccf::pal::Mutex pending_access; - ringbuffer::WriterPtr to_host; + // Directory where encrypted blobs are persisted to disk + const std::filesystem::path root_dir = ".index"; + + // All disk I/O is performed as actions on this single ordered task queue, + // off the calling (indexing) thread. A single ordered queue preserves the + // submission order of stores and fetches, so a fetch issued after a store + // for the same key always observes the written file. + std::shared_ptr tasks; ccf::crypto::EntropyPtr entropy_src; std::unique_ptr encryption_key; @@ -118,123 +130,78 @@ namespace ccf::indexing return gcm.serialise(); } + void write_blob( + const LFSKey& obfuscated, const LFSEncryptedContents& encrypted) + { + const auto target_path = root_dir / obfuscated; + asynchost::TimeBoundLogger log_if_slow(fmt::format( + "Writing LFS file ({} bytes) - {}", + encrypted.size(), + target_path.string())); + LOG_TRACE_FMT( + "Writing {} byte file to {}", encrypted.size(), target_path.string()); + files::dump(encrypted, target_path); + } + + std::optional read_blob(const LFSKey& obfuscated) + { + const auto target_path = root_dir / obfuscated; + if (!std::filesystem::is_regular_file(target_path)) + { + LOG_TRACE_FMT("File {} not found", target_path.string()); + return std::nullopt; + } + + asynchost::TimeBoundLogger log_if_slow( + fmt::format("Reading LFS file - ifstream({})", target_path.string())); + std::ifstream f(target_path, std::ios::binary); + f.seekg(0, std::ios::end); + const auto file_size = f.tellg(); + f.seekg(0, std::ios::beg); + LFSEncryptedContents blob(static_cast(file_size)); + f.read( + reinterpret_cast(blob.data()), + static_cast(blob.size())); + return blob; + } + public: - EnclaveLFSAccess(ringbuffer::WriterPtr writer) : - to_host(std::move(writer)), + EnclaveLFSAccess(ccf::tasks::JobBoard& job_board) : + tasks(ccf::tasks::OrderedTasks::create(job_board, "LFSAccess")), entropy_src(ccf::crypto::get_entropy()) { - // Generate a fresh random key. Only this specific instance, in this - // enclave, can read these files! + // Generate a fresh random key. Only this specific instance can read + // these files! encryption_key = ccf::crypto::make_key_aes_gcm( entropy_src->random(ccf::crypto::GCM_DEFAULT_KEY_SIZE)); + + if (std::filesystem::is_directory(root_dir)) + { + LOG_INFO_FMT( + "Clearing contents from existing directory {}", root_dir.string()); + asynchost::TimeBoundLogger log_if_slow(fmt::format( + "Clearing LFS index directory - remove_all({})", root_dir.string())); + std::filesystem::remove_all(root_dir); + } + + { + asynchost::TimeBoundLogger log_if_slow(fmt::format( + "Creating LFS index directory - create_directory({})", + root_dir.string())); + if (!std::filesystem::create_directory(root_dir)) + { + throw std::logic_error( + fmt::format("Could not create directory: {}", root_dir.string())); + } + } } - void register_message_handlers( - messaging::Dispatcher& dispatcher) + ~EnclaveLFSAccess() override { - DISPATCHER_SET_MESSAGE_HANDLER( - dispatcher, LFSMsg::response, [this](const uint8_t* data, size_t size) { - auto [obfuscated, encrypted] = - ringbuffer::read_message(data, size); - std::lock_guard guard(pending_access); - auto it = pending.find(obfuscated); - if (it != pending.end()) - { - auto result = it->second.lock(); - if (result != nullptr) - { - if ( - result->fetch_result == FetchResult::FetchResultType::Fetching) - { - const auto success = verify_and_decrypt( - *encryption_key, - obfuscated, - std::move(encrypted), - result->contents); - if (success) - { - result->fetch_result = FetchResult::FetchResultType::Loaded; - } - else - { - result->fetch_result = FetchResult::FetchResultType::Corrupt; - LOG_TRACE_FMT( - "Cache was given invalid contents for {} (aka {})", - obfuscated, - result->key); - } - } - else - { - LOG_FAIL_FMT( - "Retained result for {} (aka {}) in state {}", - obfuscated, - result->key, - result->fetch_result); - } - } - else - { - LOG_TRACE_FMT( - "Received response for {}, but caller has already dropped " - "result", - obfuscated); - } - pending.erase(it); - } - else - { - LOG_TRACE_FMT( - "Ignoring response message for unrequested key {}", obfuscated); - } - }); - - DISPATCHER_SET_MESSAGE_HANDLER( - dispatcher, - LFSMsg::not_found, - [this](const uint8_t* data, size_t size) { - auto [obfuscated] = - ringbuffer::read_message(data, size); - std::lock_guard guard(pending_access); - auto it = pending.find(obfuscated); - if (it != pending.end()) - { - auto result = it->second.lock(); - if (result != nullptr) - { - if ( - result->fetch_result == FetchResult::FetchResultType::Fetching) - { - LOG_TRACE_FMT( - "Host has no contents for key {} (aka {})", - obfuscated, - result->key); - result->fetch_result = FetchResult::FetchResultType::NotFound; - } - else - { - LOG_FAIL_FMT( - "Retained result for {} (aka {}) in state {}", - obfuscated, - result->key, - result->fetch_result); - } - } - else - { - LOG_TRACE_FMT( - "Received not_found for {}, but caller has already dropped " - "result", - obfuscated); - } - pending.erase(it); - } - else - { - LOG_TRACE_FMT( - "Ignoring not_found message for unrequested key {}", obfuscated); - } - }); + // Stop processing any further queued I/O actions. In production the task + // threads have already been stopped by this point; in tests actions are + // drained synchronously before destruction. + tasks->cancel_task(); } static LFSKey obfuscate_key(const LFSKey& key) @@ -251,13 +218,14 @@ namespace ccf::indexing void store(const LFSKey& key, LFSContents&& contents) override { const auto obfuscated = obfuscate_key(key); - RINGBUFFER_WRITE_MESSAGE( - LFSMsg::store, - to_host, - // To avoid leaking potentially confidential information to the host, - // all cached data is encrypted and stored at an obfuscated key - obfuscated, - encrypt(obfuscated, std::move(contents))); + // To avoid leaking potentially confidential information, all cached data + // is encrypted and stored at an obfuscated key. Encryption happens on the + // calling thread; the resulting blob is written to disk asynchronously. + auto encrypted = encrypt(obfuscated, std::move(contents)); + tasks->add_action(ccf::tasks::make_basic_action( + [this, obfuscated, encrypted = std::move(encrypted)]() { + write_blob(obfuscated, encrypted); + })); } FetchResultPtr fetch(const LFSKey& key) override @@ -297,7 +265,49 @@ namespace ccf::indexing result->fetch_result = FetchResult::FetchResultType::Fetching; result->key = key; - RINGBUFFER_WRITE_MESSAGE(LFSMsg::get, to_host, obfuscated); + + // Read and decrypt the file from disk asynchronously, updating the + // result's state when complete. + tasks->add_action( + ccf::tasks::make_basic_action([this, obfuscated, result]() { + auto encrypted = read_blob(obfuscated); + if (!encrypted.has_value()) + { + LOG_TRACE_FMT( + "No contents for key {} (aka {})", obfuscated, result->key); + result->fetch_result = FetchResult::FetchResultType::NotFound; + } + else + { + const auto success = verify_and_decrypt( + *encryption_key, + obfuscated, + std::move(*encrypted), + result->contents); + if (success) + { + result->fetch_result = FetchResult::FetchResultType::Loaded; + } + else + { + result->fetch_result = FetchResult::FetchResultType::Corrupt; + LOG_TRACE_FMT( + "Cache was given invalid contents for {} (aka {})", + obfuscated, + result->key); + } + } + + // Drop the pending entry, unless a newer fetch has already replaced + // it with a different result. + std::lock_guard guard(pending_access); + auto it = pending.find(obfuscated); + if (it != pending.end() && it->second.lock() == result) + { + pending.erase(it); + } + })); + return result; } }; diff --git a/src/indexing/lfs_ringbuffer_types.h b/src/indexing/lfs_ringbuffer_types.h deleted file mode 100644 index a6797efe403..00000000000 --- a/src/indexing/lfs_ringbuffer_types.h +++ /dev/null @@ -1,35 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the Apache 2.0 License. -#pragma once - -#include "ds/ring_buffer_types.h" -#include "indexing/lfs_interface.h" - -namespace ccf::indexing -{ - /// LFS-related ringbuffer messages - enum LFSMsg : ringbuffer::Message - { - DEFINE_RINGBUFFER_MSG_TYPE(store), - - DEFINE_RINGBUFFER_MSG_TYPE(get), - - DEFINE_RINGBUFFER_MSG_TYPE(response), - DEFINE_RINGBUFFER_MSG_TYPE(not_found), - }; -} - -DECLARE_RINGBUFFER_MESSAGE_PAYLOAD( - ccf::indexing::LFSMsg::store, - ccf::indexing::LFSKey, - ccf::indexing::LFSEncryptedContents); - -DECLARE_RINGBUFFER_MESSAGE_PAYLOAD( - ccf::indexing::LFSMsg::get, ccf::indexing::LFSKey); - -DECLARE_RINGBUFFER_MESSAGE_PAYLOAD( - ccf::indexing::LFSMsg::response, - ccf::indexing::LFSKey, - ccf::indexing::LFSEncryptedContents); -DECLARE_RINGBUFFER_MESSAGE_PAYLOAD( - ccf::indexing::LFSMsg::not_found, ccf::indexing::LFSKey); \ No newline at end of file diff --git a/src/indexing/test/indexing.cpp b/src/indexing/test/indexing.cpp index ba52bf294bd..f69a8ebb58e 100644 --- a/src/indexing/test/indexing.cpp +++ b/src/indexing/test/indexing.cpp @@ -8,7 +8,6 @@ #include "consensus/aft/test/logging_stub.h" #include "crypto/openssl/hash.h" #include "ds/test/stub_writer.h" -#include "host/lfs_file_handler.h" #include "indexing/enclave_lfs_access.h" #include "indexing/historical_transaction_fetcher.h" #include "indexing/test/common.h" @@ -741,22 +740,9 @@ TEST_CASE( auto indexer_p = std::make_shared(fetcher); auto& indexer = *indexer_p; - messaging::BufferProcessor host_bp("lfs_host"); - messaging::BufferProcessor enclave_bp("lfs_enclave"); + ccf::tasks::JobBoard board; - constexpr size_t buf_size = 1 << 16; - auto inbound_buffer = std::make_unique(buf_size); - ringbuffer::Reader inbound_reader(inbound_buffer->bd); - auto outbound_buffer = std::make_unique(buf_size); - - ringbuffer::Reader outbound_reader(outbound_buffer->bd); - asynchost::LFSFileHandler host_files( - std::make_shared(inbound_reader)); - host_files.register_message_handlers(host_bp.get_dispatcher()); - - auto enclave_lfs = std::make_shared( - std::make_shared(outbound_reader)); - enclave_lfs->register_message_handlers(enclave_bp.get_dispatcher()); + auto enclave_lfs = std::make_shared(board); ccf::AbstractNodeContext node_context; node_context.install_subsystem(enclave_lfs); @@ -919,11 +905,13 @@ TEST_CASE( std::atomic work_done = false; - std::thread ringbuffer_flusher([&]() { + std::thread lfs_flusher([&]() { while (!work_done) { - host_bp.read_all(outbound_reader); - enclave_bp.read_all(inbound_reader); + while (auto task = board.get_task()) + { + task->do_task(); + } std::this_thread::yield(); } }); @@ -956,7 +944,7 @@ TEST_CASE( } work_done = true; - ringbuffer_flusher.join(); + lfs_flusher.join(); index_ticker.join(); watchdog.join(); } diff --git a/src/indexing/test/lfs.cpp b/src/indexing/test/lfs.cpp index 2ba6ddcee66..b1b737f62e9 100644 --- a/src/indexing/test/lfs.cpp +++ b/src/indexing/test/lfs.cpp @@ -2,9 +2,9 @@ // Licensed under the Apache 2.0 License. #include "ccf/indexing/strategies/seqnos_by_key_bucketed.h" -#include "host/lfs_file_handler.h" #include "indexing/enclave_lfs_access.h" #include "indexing/test/common.h" +#include "tasks/job_board.h" #include @@ -91,22 +91,20 @@ static std::vector create_actions( TEST_CASE("Basic cache" * doctest::test_suite("lfs")) { - messaging::BufferProcessor host_bp("lfs_host"); - messaging::BufferProcessor enclave_bp("lfs_enclave"); + ccf::tasks::JobBoard board; + const std::filesystem::path index_dir{".index"}; - constexpr size_t buf_size = 1 << 10; - auto inbound_buffer = std::make_unique(buf_size); - ringbuffer::Reader inbound_reader(inbound_buffer->bd); - auto outbound_buffer = std::make_unique(buf_size); - ringbuffer::Reader outbound_reader(outbound_buffer->bd); + ccf::indexing::EnclaveLFSAccess enclave_lfs(board); - asynchost::LFSFileHandler host_files( - std::make_shared(inbound_reader)); - host_files.register_message_handlers(host_bp.get_dispatcher()); - - ccf::indexing::EnclaveLFSAccess enclave_lfs( - std::make_shared(outbound_reader)); - enclave_lfs.register_message_handlers(enclave_bp.get_dispatcher()); + auto flush_lfs = [&board]() { + size_t count = 0; + while (auto task = board.get_task()) + { + task->do_task(); + ++count; + } + return count; + }; ccf::indexing::LFSKey key_a("Blob A"); ccf::indexing::LFSContents blob_a{0, 1, 2, 3, 4, 5, 6, 7}; @@ -117,7 +115,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) enclave_lfs.store(key_a, ccf::indexing::LFSContents(blob_a)); enclave_lfs.store(key_b, ccf::indexing::LFSContents(blob_b)); - REQUIRE(2 == host_bp.read_all(outbound_reader)); + REQUIRE(flush_lfs() > 0); { INFO("Load entries"); @@ -132,8 +130,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) result_b->fetch_result == ccf::indexing::FetchResult::FetchResultType::Fetching); - host_bp.read_all(outbound_reader); - enclave_bp.read_all(inbound_reader); + flush_lfs(); REQUIRE( result_a->fetch_result == @@ -147,18 +144,15 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) } { - INFO("Host cache provides wrong file"); + INFO("Cache provides wrong file"); REQUIRE(std::filesystem::copy_file( - host_files.root_dir / - ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b), - host_files.root_dir / - ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_a), + index_dir / ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b), + index_dir / ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_a), std::filesystem::copy_options::overwrite_existing)); auto result = enclave_lfs.fetch(key_a); - host_bp.read_all(outbound_reader); - enclave_bp.read_all(inbound_reader); + flush_lfs(); REQUIRE( result->fetch_result == @@ -168,9 +162,9 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) #ifndef PLAINTEXT_CACHE { - INFO("Host cache provides corrupt file"); - const auto b_path = host_files.root_dir / - ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b); + INFO("Cache provides corrupt file"); + const auto b_path = + index_dir / ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b); const auto original_b_contents = read_file(b_path); for (auto i = 0; i < original_b_contents.size(); ++i) @@ -179,8 +173,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) auto result = enclave_lfs.fetch(key_b); - host_bp.read_all(outbound_reader); - enclave_bp.read_all(inbound_reader); + flush_lfs(); REQUIRE( result->fetch_result == @@ -204,29 +197,22 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) auto encryptor = std::make_shared(); kv_store.set_encryptor(encryptor); - messaging::BufferProcessor host_bp("lfs_host"); - messaging::BufferProcessor enclave_bp("lfs_enclave"); - - constexpr size_t buf_size = 1 << 16; - auto inbound_buffer = std::make_unique(buf_size); - ringbuffer::Reader inbound_reader(inbound_buffer->bd); - auto outbound_buffer = std::make_unique(buf_size); - - ringbuffer::Reader outbound_reader(outbound_buffer->bd); - asynchost::LFSFileHandler host_files( - std::make_shared(inbound_reader)); - host_files.register_message_handlers(host_bp.get_dispatcher()); + ccf::tasks::JobBoard board; + const std::filesystem::path index_dir{".index"}; - auto enclave_lfs = std::make_shared( - std::make_shared(outbound_reader)); - enclave_lfs->register_message_handlers(enclave_bp.get_dispatcher()); + auto enclave_lfs = std::make_shared(board); ccf::AbstractNodeContext node_context; node_context.install_subsystem(enclave_lfs); - auto flush_ringbuffers = [&]() { - return host_bp.read_all(outbound_reader) + - enclave_bp.read_all(inbound_reader); + auto flush_ringbuffers = [&board]() { + size_t count = 0; + while (auto task = board.get_task()) + { + task->do_task(); + ++count; + } + return count; }; using StratA = @@ -477,8 +463,7 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) { INFO("Deleted files"); - for (auto const& f : - std::filesystem::directory_iterator(host_files.root_dir)) + for (auto const& f : std::filesystem::directory_iterator(index_dir)) { std::filesystem::remove(f); } @@ -488,8 +473,7 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) { INFO("Corrupted files"); - for (auto const& f : - std::filesystem::directory_iterator(host_files.root_dir)) + for (auto const& f : std::filesystem::directory_iterator(index_dir)) { auto original = read_file(f); @@ -526,29 +510,21 @@ void run_sparse_index_test(size_t bucket_size, size_t num_buckets) auto encryptor = std::make_shared(); kv_store.set_encryptor(encryptor); - messaging::BufferProcessor host_bp("lfs_host"); - messaging::BufferProcessor enclave_bp("lfs_enclave"); - - constexpr size_t buf_size = 1 << 16; - auto inbound_buffer = std::make_unique(buf_size); - ringbuffer::Reader inbound_reader(inbound_buffer->bd); - auto outbound_buffer = std::make_unique(buf_size); + ccf::tasks::JobBoard board; - ringbuffer::Reader outbound_reader(outbound_buffer->bd); - asynchost::LFSFileHandler host_files( - std::make_shared(inbound_reader)); - host_files.register_message_handlers(host_bp.get_dispatcher()); - - auto enclave_lfs = std::make_shared( - std::make_shared(outbound_reader)); - enclave_lfs->register_message_handlers(enclave_bp.get_dispatcher()); + auto enclave_lfs = std::make_shared(board); ccf::AbstractNodeContext node_context; node_context.install_subsystem(enclave_lfs); - auto flush_ringbuffers = [&]() { - return host_bp.read_all(outbound_reader) + - enclave_bp.read_all(inbound_reader); + auto flush_ringbuffers = [&board]() { + size_t count = 0; + while (auto task = board.get_task()) + { + task->do_task(); + ++count; + } + return count; }; using Strat = From 69e7bfeac70dde1b27d38064e2d366ac5347a392 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Wed, 17 Jun 2026 14:10:55 +0000 Subject: [PATCH 2/4] Common flush_lfs --- src/indexing/test/lfs.cpp | 67 +++++++++++++++------------------------ 1 file changed, 25 insertions(+), 42 deletions(-) diff --git a/src/indexing/test/lfs.cpp b/src/indexing/test/lfs.cpp index b1b737f62e9..4095398f34e 100644 --- a/src/indexing/test/lfs.cpp +++ b/src/indexing/test/lfs.cpp @@ -33,6 +33,19 @@ void write_file_corrupted_at( f.close(); } +// Synchronously execute any LFS disk I/O actions (store/fetch) that have been +// queued on the given job board, returning the number of actions executed. +size_t flush_lfs(ccf::tasks::JobBoard& board) +{ + size_t count = 0; + while (auto task = board.get_task()) + { + task->do_task(); + ++count; + } + return count; +} + static std::vector create_actions( ExpectedSeqNos& seqnos_hello, ExpectedSeqNos& seqnos_saluton, @@ -96,16 +109,6 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) ccf::indexing::EnclaveLFSAccess enclave_lfs(board); - auto flush_lfs = [&board]() { - size_t count = 0; - while (auto task = board.get_task()) - { - task->do_task(); - ++count; - } - return count; - }; - ccf::indexing::LFSKey key_a("Blob A"); ccf::indexing::LFSContents blob_a{0, 1, 2, 3, 4, 5, 6, 7}; @@ -115,7 +118,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) enclave_lfs.store(key_a, ccf::indexing::LFSContents(blob_a)); enclave_lfs.store(key_b, ccf::indexing::LFSContents(blob_b)); - REQUIRE(flush_lfs() > 0); + REQUIRE(flush_lfs(board) > 0); { INFO("Load entries"); @@ -130,7 +133,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) result_b->fetch_result == ccf::indexing::FetchResult::FetchResultType::Fetching); - flush_lfs(); + flush_lfs(board); REQUIRE( result_a->fetch_result == @@ -152,7 +155,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) auto result = enclave_lfs.fetch(key_a); - flush_lfs(); + flush_lfs(board); REQUIRE( result->fetch_result == @@ -173,7 +176,7 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) auto result = enclave_lfs.fetch(key_b); - flush_lfs(); + flush_lfs(board); REQUIRE( result->fetch_result == @@ -205,16 +208,6 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) ccf::AbstractNodeContext node_context; node_context.install_subsystem(enclave_lfs); - auto flush_ringbuffers = [&board]() { - size_t count = 0; - while (auto task = board.get_task()) - { - task->do_task(); - ++count; - } - return count; - }; - using StratA = ccf::indexing::strategies::SeqnosByKey_Bucketed; auto index_a = std::make_shared(map_a, node_context, 100, 4); @@ -254,12 +247,12 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) } fetcher->requested.clear(); - flush_ringbuffers(); + flush_lfs(board); } }; tick_until_caught_up(); - REQUIRE(flush_ringbuffers() == 0); + REQUIRE(flush_lfs(board) == 0); auto current_seqno = kv_store.current_version(); const auto max_requestable = index_a->max_requestable_range(); @@ -275,7 +268,7 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) REQUIRE_THROWS(index_a->get_write_txs_in_range( "hello", current_seqno - (max_requestable + 1), current_seqno)); - REQUIRE(flush_ringbuffers() == 0); + REQUIRE(flush_lfs(board) == 0); } auto fetch_all = [&]( @@ -295,7 +288,7 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) if (!results.has_value()) { // This required an async load from disk - REQUIRE(flush_ringbuffers() > 0); + REQUIRE(flush_lfs(board) > 0); results = strat->get_write_txs_in_range(key, range_start, range_end); @@ -344,7 +337,7 @@ TEST_CASE("Integrated cache" * doctest::test_suite("lfs")) if (!results.has_value()) { // This required an async load from disk - REQUIRE(flush_ringbuffers() > 0); + REQUIRE(flush_lfs(board) > 0); results = strat->get_write_txs_in_range(range_start, range_end); @@ -517,16 +510,6 @@ void run_sparse_index_test(size_t bucket_size, size_t num_buckets) ccf::AbstractNodeContext node_context; node_context.install_subsystem(enclave_lfs); - auto flush_ringbuffers = [&board]() { - size_t count = 0; - while (auto task = board.get_task()) - { - task->do_task(); - ++count; - } - return count; - }; - using Strat = ccf::indexing::strategies::SeqnosByKey_Bucketed; const auto many_buckets = bucket_size * (num_buckets + 1); @@ -606,12 +589,12 @@ void run_sparse_index_test(size_t bucket_size, size_t num_buckets) } fetcher->requested.clear(); - flush_ringbuffers(); + flush_lfs(board); } }; tick_until_caught_up(); - REQUIRE(flush_ringbuffers() == 0); + REQUIRE(flush_lfs(board) == 0); auto fetch_write_seqnos = [&](size_t key) { const auto max_range = index->max_requestable_range(); @@ -636,7 +619,7 @@ void run_sparse_index_test(size_t bucket_size, size_t num_buckets) if (!results.has_value()) { // This required an async load from disk - REQUIRE(flush_ringbuffers() > 0); + REQUIRE(flush_lfs(board) > 0); results = index->get_write_txs_in_range(key, range_start, range_end); REQUIRE(results.has_value()); From 83dd7c9e9c9069a09a6937073af857a76a6f91ac Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 11 Sep 2026 15:38:53 +0100 Subject: [PATCH 3/4] Fix LFS read result tidy errors Use a compact enum representation and explicit aggregate initializers for blob read outcomes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/indexing/enclave_lfs_access.h | 39 ++++++++++++++++++++++++------- 1 file changed, 30 insertions(+), 9 deletions(-) diff --git a/src/indexing/enclave_lfs_access.h b/src/indexing/enclave_lfs_access.h index 7a43389477b..4df5b8fb15e 100644 --- a/src/indexing/enclave_lfs_access.h +++ b/src/indexing/enclave_lfs_access.h @@ -146,7 +146,7 @@ namespace ccf::indexing struct BlobReadResult { - enum class Status + enum class Status : std::uint8_t { Success, NotFound, @@ -154,7 +154,7 @@ namespace ccf::indexing }; Status status; - LFSEncryptedContents contents = {}; + LFSEncryptedContents contents; }; BlobReadResult read_blob(const LFSKey& obfuscated) @@ -169,11 +169,17 @@ namespace ccf::indexing "Failed to inspect LFS file {}: {}", target_path.string(), ec.message()); - return {BlobReadResult::Status::Error}; + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; } LOG_TRACE_FMT("File {} not found", target_path.string()); - return {BlobReadResult::Status::NotFound}; + return { + .status = BlobReadResult::Status::NotFound, + .contents = {}, + }; } asynchost::TimeBoundLogger log_if_slow( @@ -182,7 +188,10 @@ namespace ccf::indexing if (!f) { LOG_FAIL_FMT("Failed to open LFS file {}", target_path.string()); - return {BlobReadResult::Status::Error}; + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; } const auto file_size = static_cast(f.tellg()); @@ -197,7 +206,10 @@ namespace ccf::indexing LOG_FAIL_FMT( "Failed to determine a supported size for LFS file {}", target_path.string()); - return {BlobReadResult::Status::Error}; + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; } f.seekg(0, std::ios::beg); @@ -205,7 +217,10 @@ namespace ccf::indexing { LOG_FAIL_FMT( "Failed to seek to the start of LFS file {}", target_path.string()); - return {BlobReadResult::Status::Error}; + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; } blob.resize(static_cast(file_size)); @@ -217,11 +232,17 @@ namespace ccf::indexing { LOG_FAIL_FMT( "Failed to read the complete LFS file {}", target_path.string()); - return {BlobReadResult::Status::Error}; + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; } } - return {BlobReadResult::Status::Success, std::move(blob)}; + return { + .status = BlobReadResult::Status::Success, + .contents = std::move(blob), + }; } public: From 5230fa9448701acb6e37320e433053e3c3d132d7 Mon Sep 17 00:00:00 2001 From: achamayou Date: Sun, 13 Sep 2026 21:09:12 +0100 Subject: [PATCH 4/4] Reject malformed and oversized LFS cache files as corrupt The contents of the .index directory are untrusted. A file truncated below the size of a GCM header made GcmCipher::deserialise throw out of the fetch action, and a file larger than physical memory was read into a vector before decryption. Both escaped try_do_task and aborted the node, instead of being reported as Corrupt so the strategy can re-index. Catch the deserialisation failure in verify_and_decrypt, and bound the admitted file size by the largest blob this instance has stored, which is published before the write is queued. Only this instance can produce a decryptable file, so any larger file is invalid without reading it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/indexing/enclave_lfs_access.h | 62 ++++++++++++++++++++----- src/indexing/test/lfs.cpp | 76 ++++++++++++++++++++++++++++--- 2 files changed, 120 insertions(+), 18 deletions(-) diff --git a/src/indexing/enclave_lfs_access.h b/src/indexing/enclave_lfs_access.h index 4df5b8fb15e..32a783568c9 100644 --- a/src/indexing/enclave_lfs_access.h +++ b/src/indexing/enclave_lfs_access.h @@ -18,7 +18,6 @@ #include #include #include -#include #include // Uncomment to disable encryption and obfuscation, writing cache content @@ -39,7 +38,17 @@ namespace ccf::indexing std::vector& plaintext) { ccf::crypto::GcmCipher gcm; - gcm.deserialise(encrypted); + try + { + gcm.deserialise(encrypted); + } + catch (const serialized::InsufficientSpaceException& e) + { + // The file on disk is untrusted, and may have been truncated below the + // size of a GCM header + LOG_TRACE_FMT("Malformed encrypted contents for {}: {}", key, e.what()); + return false; + } #ifdef PLAINTEXT_CACHE plaintext = gcm.cipher; @@ -105,6 +114,23 @@ namespace ccf::indexing ccf::crypto::EntropyPtr entropy_src; std::unique_ptr encryption_key; + // Size of the largest encrypted blob written by this instance. Only files + // written by this instance can be decrypted, so any larger file on disk + // cannot be valid, and is rejected before it is read into memory. + std::atomic max_stored_size = 0; + + void update_max_stored_size(size_t size) + { + auto current = max_stored_size.load(); + while (size > current) + { + if (max_stored_size.compare_exchange_weak(current, size)) + { + break; + } + } + } + LFSEncryptedContents encrypt(const LFSKey& key, LFSContents&& contents) { // Prefix the contents with the key, to be checked during decryption @@ -195,17 +221,28 @@ namespace ccf::indexing } const auto file_size = static_cast(f.tellg()); - LFSEncryptedContents blob; - if ( - file_size < 0 || - static_cast(file_size) > blob.max_size() || - static_cast(file_size) > - static_cast( - std::numeric_limits::max())) + if (file_size < 0) + { + LOG_FAIL_FMT( + "Failed to determine the size of LFS file {}", target_path.string()); + return { + .status = BlobReadResult::Status::Error, + .contents = {}, + }; + } + + // The file contents are untrusted, so bound the allocation below by the + // largest blob this instance has written. Any store of a larger blob + // publishes its size before its write is queued, so a fetch ordered + // after that store always accepts the resulting file. + const auto max_size = max_stored_size.load(); + if (static_cast(file_size) > max_size) { LOG_FAIL_FMT( - "Failed to determine a supported size for LFS file {}", - target_path.string()); + "LFS file {} is {} bytes, larger than any stored blob ({} bytes)", + target_path.string(), + file_size, + max_size); return { .status = BlobReadResult::Status::Error, .contents = {}, @@ -223,7 +260,7 @@ namespace ccf::indexing }; } - blob.resize(static_cast(file_size)); + LFSEncryptedContents blob(static_cast(file_size)); if (!blob.empty()) { const auto expected_size = static_cast(blob.size()); @@ -302,6 +339,7 @@ namespace ccf::indexing // is encrypted and stored at an obfuscated key. Encryption happens on the // calling thread; the resulting blob is written to disk asynchronously. auto encrypted = encrypt(obfuscated, std::move(contents)); + update_max_stored_size(encrypted.size()); tasks->add_action(ccf::tasks::make_basic_action( [this, obfuscated, encrypted = std::move(encrypted)]() { write_blob(obfuscated, encrypted); diff --git a/src/indexing/test/lfs.cpp b/src/indexing/test/lfs.cpp index 61c94aa2115..ef15d7ad08b 100644 --- a/src/indexing/test/lfs.cpp +++ b/src/indexing/test/lfs.cpp @@ -20,6 +20,14 @@ std::vector read_file(const std::filesystem::path& p) return contents; } +void write_file( + const std::filesystem::path& p, const std::vector& contents) +{ + std::ofstream f(p, std::ios::trunc | std::ios::binary); + f.write((char const*)contents.data(), contents.size()); + f.close(); +} + void write_file_corrupted_at( const std::filesystem::path& p, size_t i, @@ -28,9 +36,7 @@ void write_file_corrupted_at( auto corrupted(original); REQUIRE(i < corrupted.size()); corrupted[i]++; - std::ofstream f(p, std::ios::trunc | std::ios::binary); - f.write((char const*)corrupted.data(), corrupted.size()); - f.close(); + write_file(p, corrupted); } // Synchronously execute any LFS disk I/O actions (store/fetch) that have been @@ -163,12 +169,13 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) REQUIRE(result->contents != blob_a); } + const auto b_path = + index_dir / ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b); + const auto original_b_contents = read_file(b_path); + #ifndef PLAINTEXT_CACHE { INFO("Cache provides corrupt file"); - const auto b_path = - index_dir / ccf::indexing::EnclaveLFSAccess::obfuscate_key(key_b); - const auto original_b_contents = read_file(b_path); for (size_t i = 0; i < original_b_contents.size(); ++i) { @@ -185,6 +192,63 @@ TEST_CASE("Basic cache" * doctest::test_suite("lfs")) } } #endif + + { + INFO("Cache provides truncated file"); + + // Every truncation must be reported as corrupt, including files too short + // to contain a GCM header, rather than escaping as an exception + for (size_t size = 0; size < original_b_contents.size(); ++size) + { + write_file( + b_path, + std::vector( + original_b_contents.begin(), original_b_contents.begin() + size)); + + auto result = lfs_access.fetch(key_b); + + flush_lfs(board); + + REQUIRE( + result->fetch_result == + ccf::indexing::FetchResult::FetchResultType::Corrupt); + REQUIRE(result->contents != blob_b); + } + } + + { + INFO("Cache provides oversized file"); + + // Sparsely extend the file far beyond any blob this instance has written. + // The read must be rejected by size, before any allocation is attempted. + std::error_code ec; + std::filesystem::resize_file( + b_path, static_cast(1) << 40, ec); + REQUIRE_MESSAGE(!ec, ec.message()); + + auto result = lfs_access.fetch(key_b); + + flush_lfs(board); + + REQUIRE( + result->fetch_result == + ccf::indexing::FetchResult::FetchResultType::Corrupt); + REQUIRE(result->contents != blob_b); + } + + { + INFO("Restored file loads again"); + write_file(b_path, original_b_contents); + + auto result = lfs_access.fetch(key_b); + + flush_lfs(board); + + REQUIRE( + result->fetch_result == + ccf::indexing::FetchResult::FetchResultType::Loaded); + REQUIRE(result->contents == blob_b); + } } TEST_CASE("Integrated cache" * doctest::test_suite("lfs"))