diff --git a/CHANGELOG.md b/CHANGELOG.md index d48ef750fe5a..cde8bbb88dd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- JWT/JWK auto-refresh failures are now retried after the lesser of 5 seconds and the configured key refresh interval, with exponential backoff capped at that interval (#8226, #3869). - A signature transaction decided whether to end a ledger chunk, and recorded that decision on the chunker, outside the version lock. A rollback landing in that window discarded the signature but left the chunk marker behind. The decision and the record are now made atomically, and skipped when the signature's view or rollback epoch no longer holds (#8246). - A transaction's `force_ledger_chunk` and `snapshot_at_next_signature` flags are no longer applied once a concurrent view change has discarded the transaction's writes, which previously left a chunk boundary, or an armed snapshot, for a transaction no longer present in the ledger. The forced chunk is also attached to the transaction's own version rather than whichever version the store had reached (#8245). - A rollback whose target is at or beyond the store's own version no longer moves ledger chunk metadata forward past it, which previously left a permanent offset skewing later chunk boundaries (#8244). diff --git a/CMakeLists.txt b/CMakeLists.txt index dee27f1f816e..e4b372f4661e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -983,6 +983,15 @@ if(BUILD_TESTS) ) target_link_libraries(jwt_auth_test PRIVATE ccf_endpoints) + add_unit_test( + jwt_key_auto_refresh_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/node/test/jwt_key_auto_refresh.cpp + ) + target_link_libraries( + jwt_key_auto_refresh_test + PRIVATE http_parser ccf_endpoints ccf_kv ccf_tasks curl uv + ) + add_unit_test( tx_status_test ${CMAKE_CURRENT_SOURCE_DIR}/src/node/rpc/test/tx_status_test.cpp diff --git a/doc/build_apps/auth/jwt.rst b/doc/build_apps/auth/jwt.rst index e800255677dc..34d17058b92e 100644 --- a/doc/build_apps/auth/jwt.rst +++ b/doc/build_apps/auth/jwt.rst @@ -107,7 +107,7 @@ Now the issuer can be created with auto-refresh enabled: .. note:: - The key refresh interval is set via the ``jwt.key_refresh_interval`` configuration entry, where the default is 30 min (1800 seconds). The maximum response body size accepted when fetching OpenID metadata and JWKS is set via ``jwt.key_refresh_max_response_size``, where the default is 1 MB. + The key refresh interval is set via the ``jwt.key_refresh_interval`` configuration entry, where the default is 30 min (1800 seconds). Failed refreshes are retried after the lesser of 5 seconds and the configured interval, with exponential backoff capped at the configured interval. The maximum response body size accepted when fetching OpenID metadata and JWKS is set via ``jwt.key_refresh_max_response_size``, where the default is 1 MB. Removing a token issuer ----------------------- diff --git a/src/node/jwt_key_auto_refresh.h b/src/node/jwt_key_auto_refresh.h index f1b67abf575d..ee1e290e6c83 100644 --- a/src/node/jwt_key_auto_refresh.h +++ b/src/node/jwt_key_auto_refresh.h @@ -3,8 +3,11 @@ #pragma once #include "ccf/ds/json.h" +#include "ccf/ds/locking.h" #include "ccf/ds/nonstd.h" +#include "ccf/service/tables/cert_bundles.h" #include "ccf/service/tables/jwt.h" +#include "enclave/rpc_map.h" #include "http/curl.h" #include "http/http_builder.h" #include "http/http_rpc_context.h" @@ -13,8 +16,10 @@ #include "tasks/task_system.h" #define FMT_HEADER_ONLY +#include #include #include +#include namespace ccf { @@ -34,9 +39,137 @@ namespace ccf ccf::tasks::Task periodic_refresh_task; + struct RetryState + { + size_t delay_s; + // Identifies the scheduled callback, not the issuer. Cancellation cannot + // stop a callback that has already started, so it must also check this + // ID. + size_t generation = 0; + ccf::tasks::Task task = nullptr; + }; + + ccf::ds::Mutex retry_states_lock; + std::map retry_states + CCF_GUARDED_BY(retry_states_lock); + // Never reuse an ID, even after an issuer's retry state is erased. + size_t next_retry_generation CCF_GUARDED_BY(retry_states_lock) = 0; + + static constexpr size_t initial_retry_delay_s = 5; static constexpr long request_connection_timeout_s = 5; static constexpr long request_response_timeout_s = 5; + bool should_begin_retry(const JwtIssuer& issuer, size_t generation) + { + ccf::ds::MutexGuard guard(retry_states_lock); + const auto it = retry_states.find(issuer); + if ( + it == retry_states.end() || it->second.generation != generation || + it->second.task == nullptr) + { + return false; + } + + // Consume this callback's slot so a failed refresh can schedule the next + // retry, retaining the increased delay. + it->second.task = nullptr; + return true; + } + + void cancel_retry(const JwtIssuer& issuer) + { + ccf::ds::MutexGuard guard(retry_states_lock); + const auto it = retry_states.find(issuer); + if (it == retry_states.end()) + { + return; + } + + if (it->second.task != nullptr) + { + it->second.task->cancel_task(); + } + retry_states.erase(it); + } + + void cancel_retries() + { + ccf::ds::MutexGuard guard(retry_states_lock); + for (auto& retry_state_entry : retry_states) + { + auto& retry_state = retry_state_entry.second; + if (retry_state.task != nullptr) + { + retry_state.task->cancel_task(); + } + } + retry_states.clear(); + } + + void schedule_retry(const JwtIssuer& issuer) + { + ccf::tasks::Task retry_task; + size_t delay_s = 0; + { + ccf::ds::MutexGuard guard(retry_states_lock); + // Keep this check under the lock: otherwise stop() could clear retries + // between the check and insertion, leaving a new retry after shutdown. + if (stopped.load()) + { + return; + } + + const auto initial_delay_s = + std::min(initial_retry_delay_s, refresh_interval_s); + const auto it = + retry_states + .try_emplace(issuer, RetryState{initial_delay_s, 0, nullptr}) + .first; + auto& retry_state = it->second; + if (retry_state.task != nullptr && !retry_state.task->is_cancelled()) + { + return; + } + + delay_s = retry_state.delay_s; + const auto generation = ++next_retry_generation; + retry_state.generation = generation; + const auto self = weak_from_this(); + retry_task = ccf::tasks::make_basic_task([self, issuer, generation]() { + const auto self_sp = self.lock(); + if ( + self_sp == nullptr || self_sp->stopped.load() || + !self_sp->should_begin_retry(issuer, generation)) + { + return; + } + + if (!self_sp->consensus->can_replicate()) + { + LOG_DEBUG_FMT( + "JWT key auto-refresh retry: Node is not primary, skipping"); + self_sp->cancel_retry(issuer); + return; + } + + self_sp->refresh_jwt_keys(issuer); + }); + retry_state.task = retry_task; + + if (retry_state.delay_s < refresh_interval_s) + { + retry_state.delay_s = + std::min(retry_state.delay_s * 2, refresh_interval_s); + } + } + + LOG_DEBUG_FMT( + "JWT key auto-refresh: Scheduling retry for issuer '{}' in {}s", + issuer, + delay_s); + ccf::tasks::add_delayed_task(retry_task, std::chrono::seconds(delay_s)); + } + void send_curl_get( const std::string& url, const std::string& ca_bundle_pem, @@ -137,6 +270,7 @@ namespace ccf { periodic_refresh_task->cancel_task(); } + cancel_retries(); } void schedule_once() @@ -163,7 +297,7 @@ namespace ccf } template - void send_refresh_jwt_keys(T msg) + bool send_refresh_jwt_keys(T msg) { ::http::Request request(fmt::format( "/{}/{}", @@ -185,14 +319,17 @@ namespace ccf ::http::fetch_rpc_handler(ctx, this->rpc_map); search->process(ctx); + return ::http::status_success( + static_cast(ctx->get_response_status())); } - void send_refresh_jwt_keys_error() + void send_refresh_jwt_keys_error(const JwtIssuer& issuer) { // A message that the endpoint fails to parse, leading to 500. // This is done purely for exposing errors as endpoint metrics. auto msg = false; send_refresh_jwt_keys(msg); + schedule_retry(issuer); } void handle_jwt_jwks_response( @@ -210,7 +347,7 @@ namespace ccf data.empty() ? "" : fmt::format(" '{}'", std::string(data.begin(), data.end()))); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } @@ -228,7 +365,7 @@ namespace ccf "JWT key auto-refresh: Cannot parse JWKS for issuer '{}': {}", issuer, e.what()); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } @@ -248,7 +385,14 @@ namespace ccf // call internal endpoint to update keys auto msg = SetJwtPublicSigningKeys{issuer, jwks}; - send_refresh_jwt_keys(msg); + if (send_refresh_jwt_keys(msg)) + { + cancel_retry(issuer); + } + else + { + schedule_retry(issuer); + } } void handle_jwt_metadata_response( @@ -267,7 +411,7 @@ namespace ccf data.empty() ? "" : fmt::format(" '{}'", std::string(data.begin(), data.end()))); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } @@ -295,7 +439,7 @@ namespace ccf "{}", issuer, e.what()); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } // Validate jwks_uri before handing it to libcurl; the parsed result is @@ -315,7 +459,7 @@ namespace ccf issuer, jwks_url_str, e.what()); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } @@ -326,7 +470,7 @@ namespace ccf "JWT key auto-refresh: jwks_uri for issuer '{}' must use https: {}", issuer, jwks_url_str); - send_refresh_jwt_keys_error(); + send_refresh_jwt_keys_error(issuer); return; } @@ -365,7 +509,7 @@ namespace ccf issuer, curl_easy_strerror(curl_response), curl_response); - self_sp->send_refresh_jwt_keys_error(); + self_sp->send_refresh_jwt_keys_error(issuer); return; } self_sp->handle_jwt_jwks_response( @@ -379,6 +523,98 @@ namespace ccf send_curl_get(jwks_url_str, ca_bundle_pem, std::move(response_callback)); } + void refresh_issuer_jwt_keys( + const JwtIssuer& issuer, + const JwtIssuerMetadata& metadata, + ccf::CACertBundlePEMs::ReadOnlyHandle* ca_cert_bundles) + { + if (!metadata.auto_refresh) + { + LOG_DEBUG_FMT( + "JWT key auto-refresh: Skipping issuer '{}', auto-refresh is " + "disabled", + issuer); + cancel_retry(issuer); + return; + } + + // Increment attempts, only when auto-refresh is enabled. + attempts++; + + LOG_DEBUG_FMT( + "JWT key auto-refresh: Refreshing keys for issuer '{}'", issuer); + if (!metadata.ca_cert_bundle_name.has_value()) + { + LOG_INFO_FMT( + "JWT key auto-refresh: Issuer '{}' has auto-refresh enabled but no " + "CA cert bundle name", + issuer); + send_refresh_jwt_keys_error(issuer); + return; + } + const auto& ca_cert_bundle_name = metadata.ca_cert_bundle_name.value(); + auto ca_cert_bundle_pem = ca_cert_bundles->get(ca_cert_bundle_name); + if (!ca_cert_bundle_pem.has_value()) + { + LOG_INFO_FMT( + "JWT key auto-refresh: CA cert bundle with name '{}' for issuer " + "'{}' not " + "found", + ca_cert_bundle_name, + issuer); + send_refresh_jwt_keys_error(issuer); + return; + } + + auto metadata_url = issuer + "/.well-known/openid-configuration"; + + LOG_DEBUG_FMT( + "JWT key auto-refresh: Requesting OpenID metadata at {}", metadata_url); + + auto ca_bundle_pem = ca_cert_bundle_pem.value(); + + const auto self = weak_from_this(); + auto response_callback = [self, issuer, ca_bundle_pem]( + std::unique_ptr&& + request, + CURLcode curl_response, + long status_code) { + auto http_status = static_cast(status_code); + auto response_body_sp = std::make_shared>( + request->get_response_body() != nullptr ? + std::move(request->get_response_body()->buffer) : + std::vector{}); + ccf::tasks::add_task(ccf::tasks::make_basic_task([self, + issuer, + ca_bundle_pem, + curl_response, + http_status, + response_body_sp]() { + const auto self_sp = self.lock(); + if (self_sp == nullptr || self_sp->stopped.load()) + { + return; + } + + if (curl_response != CURLE_OK) + { + LOG_INFO_FMT( + "JWT key auto-refresh: Failed to fetch OpenID metadata for " + "issuer '{}': {} ({})", + issuer, + curl_easy_strerror(curl_response), + curl_response); + self_sp->send_refresh_jwt_keys_error(issuer); + return; + } + self_sp->handle_jwt_metadata_response( + issuer, ca_bundle_pem, http_status, std::move(*response_body_sp)); + })); + }; + + send_curl_get(metadata_url, ca_bundle_pem, std::move(response_callback)); + } + void refresh_jwt_keys() { if (stopped.load()) @@ -389,97 +625,41 @@ namespace ccf auto tx = network.tables->create_read_only_tx(); auto* jwt_issuers = tx.ro(network.jwt_issuers); auto* ca_cert_bundles = tx.ro(network.ca_cert_bundles); - jwt_issuers->foreach([this, &ca_cert_bundles]( - const JwtIssuer& issuer, - const JwtIssuerMetadata& metadata) { - if (stopped.load()) - { - return false; - } - - if (!metadata.auto_refresh) - { - LOG_DEBUG_FMT( - "JWT key auto-refresh: Skipping issuer '{}', auto-refresh is " - "disabled", - issuer); - return true; - } - - // Increment attempts, only when auto-refresh is enabled. - attempts++; + jwt_issuers->foreach( + [this, &ca_cert_bundles]( + const JwtIssuer& issuer, const JwtIssuerMetadata& metadata) { + if (stopped.load()) + { + return false; + } - LOG_DEBUG_FMT( - "JWT key auto-refresh: Refreshing keys for issuer '{}'", issuer); - const auto& ca_cert_bundle_name = metadata.ca_cert_bundle_name.value(); - auto ca_cert_bundle_pem = ca_cert_bundles->get(ca_cert_bundle_name); - if (!ca_cert_bundle_pem.has_value()) - { - LOG_INFO_FMT( - "JWT key auto-refresh: CA cert bundle with name '{}' for issuer " - "'{}' not " - "found", - ca_cert_bundle_name, - issuer); - send_refresh_jwt_keys_error(); + refresh_issuer_jwt_keys(issuer, metadata, ca_cert_bundles); return true; - } + }); + } - auto metadata_url = issuer + "/.well-known/openid-configuration"; + void refresh_jwt_keys(const JwtIssuer& issuer) + { + if (stopped.load()) + { + return; + } + auto tx = network.tables->create_read_only_tx(); + auto* jwt_issuers = tx.ro(network.jwt_issuers); + const auto metadata = jwt_issuers->get(issuer); + if (!metadata.has_value()) + { LOG_DEBUG_FMT( - "JWT key auto-refresh: Requesting OpenID metadata at {}", - metadata_url); - - auto ca_bundle_pem = ca_cert_bundle_pem.value(); + "JWT key auto-refresh: Issuer '{}' is no longer registered, " + "abandoning retries", + issuer); + cancel_retry(issuer); + return; + } - const auto self = weak_from_this(); - auto response_callback = - [self, issuer, ca_bundle_pem]( - std::unique_ptr&& request, - CURLcode curl_response, - long status_code) { - auto http_status = static_cast(status_code); - auto response_body_sp = std::make_shared>( - request->get_response_body() != nullptr ? - std::move(request->get_response_body()->buffer) : - std::vector{}); - ccf::tasks::add_task( - ccf::tasks::make_basic_task([self, - issuer, - ca_bundle_pem, - curl_response, - http_status, - response_body_sp]() { - const auto self_sp = self.lock(); - if (self_sp == nullptr || self_sp->stopped.load()) - { - return; - } - - if (curl_response != CURLE_OK) - { - LOG_INFO_FMT( - "JWT key auto-refresh: Failed to fetch OpenID metadata for " - "issuer '{}': {} ({})", - issuer, - curl_easy_strerror(curl_response), - curl_response); - self_sp->send_refresh_jwt_keys_error(); - return; - } - self_sp->handle_jwt_metadata_response( - issuer, - ca_bundle_pem, - http_status, - std::move(*response_body_sp)); - })); - }; - - send_curl_get( - metadata_url, ca_bundle_pem, std::move(response_callback)); - return true; - }); + refresh_issuer_jwt_keys( + issuer, metadata.value(), tx.ro(network.ca_cert_bundles)); } // Returns a copy of the current attempts diff --git a/src/node/test/jwt_key_auto_refresh.cpp b/src/node/test/jwt_key_auto_refresh.cpp new file mode 100644 index 000000000000..319b9b52f3d0 --- /dev/null +++ b/src/node/test/jwt_key_auto_refresh.cpp @@ -0,0 +1,352 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +#include "node/jwt_key_auto_refresh.h" + +#include "kv/test/null_encryptor.h" +#include "kv/test/stub_consensus.h" + +#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN +#include + +using namespace std::chrono_literals; + +namespace +{ + class RefreshEndpoint : public ccf::RpcHandler + { + public: + bool accept_keys = true; + size_t key_updates = 0; + + void set_sig_intervals(size_t, size_t) override {} + void set_cmd_forwarder(std::shared_ptr) override {} + void open() override {} + bool is_open() override + { + return true; + } + void set_consensus_and_history( + ccf::kv::Consensus*, ccf::kv::TxHistory*) override + {} + + void process(std::shared_ptr ctx) override + { + const auto body = ccf::parse_json_safe(ctx->get_request_body()); + if (body.is_object()) + { + ++key_updates; + } + ctx->set_response_status( + body.is_object() && accept_keys ? HTTP_STATUS_OK : + HTTP_STATUS_INTERNAL_SERVER_ERROR); + } + }; + + struct Fixture + { + ccf::NetworkState network; + std::shared_ptr consensus = + std::make_shared(); + std::shared_ptr endpoint = + std::make_shared(); + std::shared_ptr refresh; + const ccf::JwtIssuer issuer = "https://issuer.example"; + + Fixture(size_t refresh_interval_s = 30) + { + network.tables->set_encryptor( + std::make_shared()); + consensus->force_become_primary(); + auto rpc_map = std::make_shared(); + rpc_map->register_frontend(endpoint); + refresh = std::make_shared( + refresh_interval_s, + network, + consensus, + rpc_map, + nullptr, + ccf::crypto::Pem{}, + 4096); + set_issuer(issuer); + } + + ~Fixture() + { + refresh.reset(); + advance(1h); + } + + void set_issuer(const ccf::JwtIssuer& name, bool auto_refresh = true) + { + auto tx = network.tables->create_tx(); + // A missing CA bundle makes each attempt fail synchronously, without + // wall-clock waits or external HTTP servers. + tx.rw(network.jwt_issuers) + ->put(name, ccf::JwtIssuerMetadata{std::nullopt, auto_refresh}); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + void advance(std::chrono::milliseconds elapsed) + { + ccf::tasks::tick(elapsed); + auto& job_board = ccf::tasks::get_main_job_board(); + while (auto task = job_board.get_task()) + { + task->do_task(); + } + } + + void expect_attempt_after(std::chrono::milliseconds delay) + { + const auto attempts = refresh->get_attempts(); + CAPTURE(delay.count()); + CAPTURE(attempts); + advance(delay - 1ms); + REQUIRE(refresh->get_attempts() == attempts); + advance(1ms); + REQUIRE(refresh->get_attempts() == attempts + 1); + } + + void respond_with_keys(const ccf::JwtIssuer& name) + { + const auto body = nlohmann::json(ccf::JsonWebKeySet{}).dump(); + refresh->handle_jwt_jwks_response( + name, + std::nullopt, + HTTP_STATUS_OK, + std::vector(body.begin(), body.end())); + } + + std::shared_ptr take_ready_retry() + { + auto task = std::dynamic_pointer_cast( + ccf::tasks::get_main_job_board().get_task()); + REQUIRE(task != nullptr); + return task; + } + }; +} + +TEST_CASE("JWT retries double their delay up to the configured maximum") +{ + Fixture f; + f.refresh->refresh_jwt_keys(f.issuer); + REQUIRE(f.refresh->get_attempts() == 1); + for (const auto delay : {5s, 10s, 20s, 30s, 30s}) + { + f.expect_attempt_after(delay); + } +} + +TEST_CASE("JWT retries respect a maximum below the initial retry delay") +{ + Fixture f(3); + f.refresh->refresh_jwt_keys(f.issuer); + f.expect_attempt_after(3s); + f.expect_attempt_after(3s); +} + +TEST_CASE("JWT refresh failures do not replace or advance a pending retry") +{ + Fixture f; + f.refresh->schedule_once(); + f.advance(0ms); + REQUIRE(f.refresh->get_attempts() == 1); + f.advance(2s); + f.refresh->schedule_once(); + f.advance(0ms); + REQUIRE(f.refresh->get_attempts() == 2); + f.expect_attempt_after(3s); + f.expect_attempt_after(10s); +} + +TEST_CASE("Periodic JWT refresh failures preserve the pending retry deadline") +{ + Fixture f(12); + f.refresh->start(); + f.refresh->schedule_once(); + f.advance(0ms); + REQUIRE(f.refresh->get_attempts() == 1); + f.expect_attempt_after(5s); + + // The periodic failure at 12s must not replace the retry due at 15s. + f.expect_attempt_after(7s); + f.expect_attempt_after(3s); + // The same holds at 24s, with the retry delay now capped at 12s. + f.expect_attempt_after(9s); + f.expect_attempt_after(3s); +} + +TEST_CASE("JWT response failures schedule an initial retry") +{ + Fixture f; + SUBCASE("Metadata HTTP error") + { + f.refresh->handle_jwt_metadata_response( + f.issuer, "", HTTP_STATUS_SERVICE_UNAVAILABLE, {}); + } + SUBCASE("Malformed metadata") + { + f.refresh->handle_jwt_metadata_response(f.issuer, "", HTTP_STATUS_OK, {}); + } + SUBCASE("JWKS HTTP error") + { + f.refresh->handle_jwt_jwks_response( + f.issuer, std::nullopt, HTTP_STATUS_SERVICE_UNAVAILABLE, {}); + } + SUBCASE("Malformed JWKS") + { + f.refresh->handle_jwt_jwks_response( + f.issuer, std::nullopt, HTTP_STATUS_OK, {}); + } + + REQUIRE(f.endpoint->key_updates == 0); + REQUIRE(f.refresh->get_attempts() == 0); + f.expect_attempt_after(5s); + f.expect_attempt_after(10s); +} + +TEST_CASE("A JWT key update schedules an initial retry only if rejected") +{ + Fixture f; + SUBCASE("Accepted") + { + f.respond_with_keys(f.issuer); + ccf::tasks::tick(30s); + REQUIRE(f.refresh->get_attempts() == 0); + REQUIRE(ccf::tasks::get_main_job_board().get_task() == nullptr); + } + SUBCASE("Rejected") + { + f.endpoint->accept_keys = false; + f.respond_with_keys(f.issuer); + REQUIRE(f.refresh->get_attempts() == 0); + f.expect_attempt_after(5s); + f.expect_attempt_after(10s); + } + REQUIRE(f.endpoint->key_updates == 1); +} + +TEST_CASE("JWT retry backoff and successful resets are independent per issuer") +{ + Fixture f; + const ccf::JwtIssuer other = "https://other.example"; + f.set_issuer(other); + f.refresh->refresh_jwt_keys(f.issuer); + f.expect_attempt_after(5s); + + f.refresh->refresh_jwt_keys(other); + f.expect_attempt_after(5s); + f.respond_with_keys(other); + REQUIRE(f.endpoint->key_updates == 1); + + f.expect_attempt_after(5s); + f.refresh->refresh_jwt_keys(other); + f.expect_attempt_after(5s); + f.respond_with_keys(other); + f.expect_attempt_after(15s); +} + +TEST_CASE("JWT retries reset only after the keys are accepted") +{ + Fixture f; + f.refresh->refresh_jwt_keys(f.issuer); + f.expect_attempt_after(5s); + + f.endpoint->accept_keys = false; + f.respond_with_keys(f.issuer); + f.expect_attempt_after(10s); + + f.endpoint->accept_keys = true; + f.respond_with_keys(f.issuer); + REQUIRE(f.endpoint->key_updates == 2); + const auto attempts = f.refresh->get_attempts(); + f.advance(30s); + REQUIRE(f.refresh->get_attempts() == attempts); + + f.refresh->refresh_jwt_keys(f.issuer); + f.expect_attempt_after(5s); +} + +TEST_CASE("Stale JWT retry callbacks cannot consume a newer retry") +{ + Fixture f; + f.refresh->refresh_jwt_keys(f.issuer); + ccf::tasks::tick(5s); + auto stale = f.take_ready_retry(); + + SUBCASE("A completed callback is invoked again") + { + stale->do_task(); + REQUIRE(f.refresh->get_attempts() == 2); + } + SUBCASE("Success cancels the callback and a later failure recreates state") + { + f.respond_with_keys(f.issuer); + REQUIRE(stale->is_cancelled()); + f.refresh->refresh_jwt_keys(f.issuer); + } + + const auto attempts = f.refresh->get_attempts(); + // Invoke the callback directly to model a worker that has already passed + // BaseTask's cancellation check before the newer retry was scheduled. + stale->fn(); + REQUIRE(f.refresh->get_attempts() == attempts); + f.expect_attempt_after(stale->is_cancelled() ? 5s : 10s); +} + +TEST_CASE("JWT retries stop when the issuer or primary role is lost") +{ + Fixture f; + f.refresh->refresh_jwt_keys(f.issuer); + + SUBCASE("Issuer removed") + { + auto tx = f.network.tables->create_tx(); + tx.rw(f.network.jwt_issuers)->remove(f.issuer); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + SUBCASE("Auto-refresh disabled") + { + f.set_issuer(f.issuer, false); + } + SUBCASE("Node is no longer primary") + { + f.consensus->state = ccf::kv::test::StubConsensus::Backup; + } + + f.advance(5s); + f.advance(30s); + REQUIRE(f.refresh->get_attempts() == 1); + + f.set_issuer(f.issuer); + f.consensus->force_become_primary(); + f.refresh->refresh_jwt_keys(f.issuer); + f.expect_attempt_after(5s); +} + +TEST_CASE("Stopping JWT refresh cancels callbacks and prevents new retries") +{ + Fixture f; + const ccf::JwtIssuer other = "https://other.example"; + f.set_issuer(other); + f.refresh->start(); + f.refresh->refresh_jwt_keys(); + REQUIRE(f.refresh->get_attempts() == 2); + ccf::tasks::tick(5s); + auto retry = f.take_ready_retry(); + auto other_retry = f.take_ready_retry(); + f.refresh->stop(); + REQUIRE(retry->is_cancelled()); + REQUIRE(other_retry->is_cancelled()); + retry->fn(); + other_retry->fn(); + f.refresh->send_refresh_jwt_keys_error(f.issuer); + f.refresh->send_refresh_jwt_keys_error(other); + ccf::tasks::tick(30s); + // Check the queue before running callbacks: their stopped checks could hide + // an incorrectly scheduled task from the attempts counter. + REQUIRE(ccf::tasks::get_main_job_board().get_task() == nullptr); + REQUIRE(f.refresh->get_attempts() == 2); +} diff --git a/tests/jwt_test.py b/tests/jwt_test.py index 8921369461eb..9c9b3622057d 100644 --- a/tests/jwt_test.py +++ b/tests/jwt_test.py @@ -410,9 +410,12 @@ def get_jwt_refresh_endpoint_metrics(primary) -> dict: @contextmanager def reserve_unlistened_local_port(): + # The socket is bound but never listened on, so connections to it are + # refused. It is yielded rather than just its port number so that callers + # can release the reservation early, e.g. to let a server bind that port. with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) - yield s.getsockname()[1] + yield s def add_auto_refresh_jwt_issuer(network, primary, issuer, ca_cert_bundle_name): @@ -451,9 +454,11 @@ def test_jwt_key_auto_refresh_connection_failure(network, args): remove_all_jwt_issuers(network, args, primary) failures_before = get_jwt_refresh_endpoint_metrics(primary)["failures"] issuer_host = "127.0.0.1" + kid = "connection_failure" LOG.info("Add JWT issuer with auto-refresh pointing at an unavailable endpoint") - with reserve_unlistened_local_port() as issuer_port: + with reserve_unlistened_local_port() as reserved_socket: + issuer_port = reserved_socket.getsockname()[1] issuer = infra.jwt_issuer.JwtIssuer( f"https://{issuer_host}:{issuer_port}", cn=issuer_host ) @@ -463,6 +468,18 @@ def test_jwt_key_auto_refresh_connection_failure(network, args): lambda: check_refresh_failures_increased(primary, failures_before), timeout=5, ) + + LOG.info("Start the OpenID endpoint and check that the refresh is retried") + # Only release the port now, so that nothing else can claim it while + # the initial refresh failures are observed. + reserved_socket.close() + with issuer.start_openid_server(issuer_port, kid): + with_timeout( + lambda: check_kv_jwt_key_matches( + args, network, kid, issuer.key_pub_pem + ), + timeout=15, + ) finally: network.consortium.remove_jwt_issuer(primary, issuer.name)