From c7068aa429c4658b4f78ae0a7d72bacf7b574ba3 Mon Sep 17 00:00:00 2001 From: Ian Kerins Date: Sun, 23 Aug 2026 18:23:40 -0400 Subject: [PATCH] aws: remove hang when signing a bootstrap extension callout It is not currently possible to make HTTP callouts from a dynamic modules bootstrap extension that block server initialization to a cluster with an upstream AWS signing filter on it that asynchronously resolves credentials. This is because `MetadataCredentialsProviderBase::setCredentialsToAllThreads` does not notify signers that credentials are available until all worker threads have had their ThreadLocalCredentialsCache updated, but this operation is blocked on worker threads being started, which is itself blocked on the pending server initialization. A cycle! Change setCredentialsToAllThreads to notify signers immediately after starting `runOnAllThreads`, which synchronously updates only the main thread, in addition to doing it after the completion of that operation across all threads. I believe this to be safe because that notification will itself only be handled in worker threads after the `runOnAllThreads` dispatch is handled. However, one wrinkle with this strategy is that it's now possible for workers to observe tears between the per-thread credentials and the global "pending" flag. So, I've made the pending flag also be per-thread, next to the credentials. Updates that set that value to `true` are broadcast to all threads from the main thread - all such updates were already happening there, and now `runOnAllThreads` asserts as much. I've added an integration test covering the stated use case, some more targeted tests in credentials_provider_test, and had to replace some no-longer-valid tests on specific credentials providers with ones that I think are equivalent in spirit. All the added tests fail without the changes in `source/`. AI disclosure: this fix comes from a real situation I found trying to use Envoy, but it was investigated by, and the solution and the tests were written by, Claude. I do not have expertise in any of the systems involved (Envoy approach to propagating state across worker threads, Envoy's AWS credentials implementation, or... C++ in general), so I appreciate careful reviews. Signed-off-by: Ian Kerins --- ...ed-signing-deadlock-during-server-init.rst | 3 + source/extensions/common/aws/BUILD | 3 +- .../assume_role_credentials_provider.cc | 2 +- .../container_credentials_provider.cc | 2 +- ...iam_roles_anywhere_credentials_provider.cc | 2 +- .../instance_profile_credentials_provider.cc | 6 +- .../webidentity_credentials_provider.cc | 2 +- .../common/aws/credentials_provider.h | 7 +- .../aws/metadata_credentials_provider_base.cc | 108 ++++-- .../aws/metadata_credentials_provider_base.h | 23 +- test/extensions/common/aws/BUILD | 3 + .../assume_role_credentials_provider_test.cc | 6 +- ...tance_profile_credentials_provider_test.cc | 6 +- .../common/aws/credentials_provider_test.cc | 322 +++++++++++++++++- test/extensions/common/aws/mocks.h | 2 + .../dynamic_modules/bootstrap/BUILD | 3 + .../bootstrap/integration_test.cc | 61 ++++ .../dynamic_modules/test_data/rust/BUILD | 2 + .../dynamic_modules/test_data/rust/Cargo.toml | 6 + .../rust/bootstrap_signed_callout_test.rs | 214 ++++++++++++ 20 files changed, 736 insertions(+), 47 deletions(-) create mode 100644 changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst create mode 100644 test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs diff --git a/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst b/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst new file mode 100644 index 0000000000000..95b4349d1eb22 --- /dev/null +++ b/changelogs/current/bug_fixes/aws__fixed-signing-deadlock-during-server-init.rst @@ -0,0 +1,3 @@ +Fixed an initialization hang when an extension that gates server initialization +sends an HTTP callout to a cluster whose AWS request signing filter resolves +credentials asynchronously. diff --git a/source/extensions/common/aws/BUILD b/source/extensions/common/aws/BUILD index fe5caa2ce0d65..96b7a48588b01 100644 --- a/source/extensions/common/aws/BUILD +++ b/source/extensions/common/aws/BUILD @@ -134,7 +134,8 @@ envoy_cc_library( ":metadata_fetcher_lib", ":utility_lib", "//envoy/common:time_interface", - "//source/common/common:cancel_wrapper_lib", + "//source/common/common:assert_lib", + "//source/common/common:thread_lib", ], ) diff --git a/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc b/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc index 2df12fb9d2323..5a3edec2fb3b4 100644 --- a/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/assume_role_credentials_provider.cc @@ -105,7 +105,7 @@ void AssumeRoleCredentialsProvider::continueRefresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/container_credentials_provider.cc b/source/extensions/common/aws/credential_providers/container_credentials_provider.cc index 1572bdbdf1c48..9944bcc988b08 100644 --- a/source/extensions/common/aws/credential_providers/container_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/container_credentials_provider.cc @@ -66,7 +66,7 @@ void ContainerCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc b/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc index c59516cb9ee23..4fe30a99c8db3 100644 --- a/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/iam_roles_anywhere_credentials_provider.cc @@ -110,7 +110,7 @@ void IAMRolesAnywhereCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc b/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc index 05a273ba787a9..8b308856fefb6 100644 --- a/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/instance_profile_credentials_provider.cc @@ -47,7 +47,7 @@ void InstanceProfileCredentialsProvider::refresh() { continue_on_async_fetch_failure_reason_ = "Token fetch failed, falling back to IMDSv1"; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(token_req_message, Tracing::NullSpan::instance(), *this); } @@ -71,7 +71,7 @@ void InstanceProfileCredentialsProvider::fetchInstanceRoleAsync(const std::strin }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } @@ -116,7 +116,7 @@ void InstanceProfileCredentialsProvider::fetchCredentialFromInstanceRoleAsync( }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc b/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc index 1bd63b5c08e94..d1261b2f02a37 100644 --- a/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc +++ b/source/extensions/common/aws/credential_providers/webidentity_credentials_provider.cc @@ -79,7 +79,7 @@ void WebIdentityCredentialsProvider::refresh() { }; // mark credentials as pending while async completes - credentials_pending_.store(true); + setCredentialsPendingToAllThreads(); metadata_fetcher_->fetch(message, Tracing::NullSpan::instance(), *this); } diff --git a/source/extensions/common/aws/credentials_provider.h b/source/extensions/common/aws/credentials_provider.h index 7ee281882111e..b214acd467836 100644 --- a/source/extensions/common/aws/credentials_provider.h +++ b/source/extensions/common/aws/credentials_provider.h @@ -168,8 +168,11 @@ using CredentialSubscriberCallbacksSharedPtr = std::shared_ptrsubscribeToCredentialUpdates // to register itself for updates via onCredentialUpdate callback. When a credential provider has -// successfully updated all threads with new credentials, via the setCredentialsToAllThreads method -// it will notify all subscribers that credentials have been retrieved. +// posted new credentials to all threads, via the setCredentialsToAllThreads method it will notify +// all subscribers that credentials have been retrieved. +// +// Notification happens as soon as the update has been posted, not once every thread has applied it, +// because the main thread may need credentials before worker threads are running at all. // // Subscription is only relevant for metadata credentials providers, as these are the only // credential providers that implement async credential retrieval functionality. diff --git a/source/extensions/common/aws/metadata_credentials_provider_base.cc b/source/extensions/common/aws/metadata_credentials_provider_base.cc index e3e8039a46189..97cf39217975c 100644 --- a/source/extensions/common/aws/metadata_credentials_provider_base.cc +++ b/source/extensions/common/aws/metadata_credentials_provider_base.cc @@ -4,6 +4,9 @@ #include "envoy/server/factory_context.h" +#include "source/common/common/assert.h" +#include "source/common/common/thread.h" + namespace Envoy { namespace Extensions { namespace Common { @@ -34,7 +37,6 @@ MetadataCredentialsProviderBase::MetadataCredentialsProviderBase( }; MetadataCredentialsProviderBase::~MetadataCredentialsProviderBase() { - cancel_credentials_update_callback_(); if (metadata_fetcher_) { metadata_fetcher_->cancel(); } @@ -68,7 +70,15 @@ void MetadataCredentialsProviderBase::credentialsRetrievalError() { handleFetchDone(); } -bool MetadataCredentialsProviderBase::credentialsPending() { return credentials_pending_; } +bool MetadataCredentialsProviderBase::credentialsPending() { + if (!tls_slot_->currentThreadRegistered()) { + ASSERT(false, "AWS credentials provider queried from a thread with no thread local storage"); + return true; + } + auto cache = tls_slot_->get(); + ASSERT(cache.has_value()); + return !cache.has_value() || cache->credentials_pending_; +} Credentials MetadataCredentialsProviderBase::getCredentials() { return *(*tls_slot_)->credentials_.get(); @@ -140,28 +150,82 @@ void MetadataCredentialsProviderBase::setCredentialsToAllThreads( CredentialsConstSharedPtr shared_credentials = std::move(creds); if (tls_slot_ && !tls_slot_->isShutdown()) { + // A weak_ptr rather than a raw `this`, so that a completion callback still queued when the + // provider goes away becomes a no-op instead of a use-after-free. + std::weak_ptr weak_self = weak_from_this(); + + // Set the credentials and clear the pending flag as a single update, so that no thread can + // observe one without the other. This writes the main thread's slot synchronously and posts the + // same update to every registered worker dispatcher. tls_slot_->runOnAllThreads( - /* Set the credentials */ [shared_credentials]( - OptRef - obj) { obj->credentials_ = shared_credentials; }, - /* Notify waiting signers on completion of credential setting above */ - CancelWrapper::cancelWrapped( - [this]() { - credentials_pending_.store(false); - std::list> subscribers_copy; - { - Thread::LockGuard guard(mu_); - subscribers_copy = credentials_subscribers_; - } - for (auto& weak_cb : subscribers_copy) { - if (auto cb = weak_cb.lock()) { - ENVOY_LOG(debug, "Notifying subscriber of credential update"); - cb->onCredentialUpdate(); - } - } - }, - &cancel_credentials_update_callback_)); + [shared_credentials](OptRef obj) { + obj->credentials_ = shared_credentials; + obj->credentials_pending_ = false; + }, + // Notify a second time once every worker has applied the update. Between the immediate + // notification below and a worker applying its update, that worker still reads + // `credentials_pending_ == true` from its own slot, so it can queue a pending callback + // after the immediate notification has already drained the queue. This notification is + // what wakes such a callback; without it the request stalls until the next successful + // refresh. + [weak_self]() { + if (auto self = weak_self.lock()) { + self->notifySubscribers(); + } + }); + + // Notify waiting signers from this thread as well, rather than relying only on the + // all-threads-complete callback above. That callback does not run until every registered worker + // dispatcher has handled the posted update, and worker dispatchers do not start running until + // `startWorkers()`, which waits on server initialization. The main thread might like to use + // credentials before that point, and waiting for the workers would deadlock. (For example, a + // dynamic modules bootstrap extension might want to make an HTTP callout before it signals + // server init is complete.) + // + // For subscribers that post their wakeup to their own dispatcher (the AWS request signing and + // Lambda filters do), notifying here is ordered correctly: the credential update above is + // posted to each worker first, and dispatcher post queues are FIFO, so a worker applies the + // update before it runs the wakeup that reads it. Subscribers that instead run their callback + // inline rely on this being the main thread, whose slot `runOnAllThreads` has already updated. + // + // Note that unlike the completion callback, this notification runs on the caller's stack, which + // for a credential refresh is inside MetadataFetcher::onSuccess()/onMetadataError() and ahead + // of handleFetchDone(). No subscriber re-enters the provider today, but one that did would see + // a half-finished refresh. + notifySubscribers(); + } +} + +void MetadataCredentialsProviderBase::notifySubscribers() { + std::list> subscribers_copy; + { + Thread::LockGuard guard(mu_); + subscribers_copy = credentials_subscribers_; + } + for (auto& weak_cb : subscribers_copy) { + if (auto cb = weak_cb.lock()) { + ENVOY_LOG(debug, "Notifying subscriber of credential update"); + cb->onCredentialUpdate(); + } + } +} + +void MetadataCredentialsProviderBase::setCredentialsPendingToAllThreads() { + // The dedup below reads the main thread's slot, so it is only meaningful on the main thread. Not + // relying on the assertion inside runOnAllThreads(), because the dedup can return before ever + // reaching it. + ASSERT_IS_MAIN_OR_TEST_THREAD(); + if (!tls_slot_ || tls_slot_->isShutdown()) { + return; + } + // The main thread's slot is written synchronously by runOnAllThreads, so it always holds the most + // recently initiated update. If it already says pending then every other thread has the same + // update applied or queued, and there is nothing to broadcast. + if ((*tls_slot_)->credentials_pending_) { + return; } + tls_slot_->runOnAllThreads( + [](OptRef obj) { obj->credentials_pending_ = true; }); } CredentialSubscriberCallbacksHandlePtr diff --git a/source/extensions/common/aws/metadata_credentials_provider_base.h b/source/extensions/common/aws/metadata_credentials_provider_base.h index 65430d857485c..2eb8a7eae638e 100644 --- a/source/extensions/common/aws/metadata_credentials_provider_base.h +++ b/source/extensions/common/aws/metadata_credentials_provider_base.h @@ -1,6 +1,5 @@ #pragma once -#include "source/common/common/cancel_wrapper.h" #include "source/extensions/common/aws/aws_cluster_manager.h" #include "source/extensions/common/aws/credentials_provider.h" #include "source/extensions/common/aws/metadata_fetcher.h" @@ -61,13 +60,17 @@ class MetadataCredentialsProviderBase subscribeToCredentialUpdates(CredentialSubscriberCallbacksSharedPtr cs); protected: + // Per-thread credential state. Both members are only ever touched on the thread that owns the + // slot: the main thread writes its own copy synchronously from setCredentialsToAllThreads and + // setCredentialsPendingToAllThreads, and every other thread applies the same update from its own + // dispatcher. struct ThreadLocalCredentialsCache : public ThreadLocal::ThreadLocalObject { ThreadLocalCredentialsCache() : credentials_(std::make_shared()) {}; // The credentials object. CredentialsConstSharedPtr credentials_; - // Lock guard. - Thread::MutexBasicLockable lock_; + // Are credentials pending on this thread? + bool credentials_pending_{true}; }; // Set anonymous credentials to all threads, update stats and close async @@ -81,9 +84,18 @@ class MetadataCredentialsProviderBase // Handle fetch done. void handleFetchDone(); - // Set Credentials shared_ptr on all threads. + // Set Credentials shared_ptr on all threads, and mark credentials as no longer pending. void setCredentialsToAllThreads(CredentialsConstUniquePtr&& creds); + // Tell every subscriber that credentials have been retrieved. Called both from the thread that + // initiates a credential update and again once all threads have applied it; see + // setCredentialsToAllThreads. Subscriber notification is idempotent. + void notifySubscribers(); + + // Mark credentials as pending on all threads, ahead of an async credential fetch. Must be called + // on the main thread. + void setCredentialsPendingToAllThreads(); + virtual void refresh() PURE; Server::Configuration::ServerFactoryContext& context_; @@ -122,12 +134,9 @@ class MetadataCredentialsProviderBase AwsClusterManagerPtr aws_cluster_manager_; // RAII handle for callbacks from AWS cluster manager AwsManagedClusterUpdateCallbacksHandlePtr callback_handle_; - // Are credentials pending? - std::atomic credentials_pending_ = true; Thread::MutexBasicLockable mu_; std::list> credentials_subscribers_ ABSL_GUARDED_BY(mu_); - CancelWrapper::CancelFunction cancel_credentials_update_callback_ = []() {}; }; } // namespace Aws diff --git a/test/extensions/common/aws/BUILD b/test/extensions/common/aws/BUILD index b1ac67e22d2ef..5a1d1c456891c 100644 --- a/test/extensions/common/aws/BUILD +++ b/test/extensions/common/aws/BUILD @@ -105,11 +105,14 @@ envoy_cc_test( srcs = ["credentials_provider_test.cc"], rbe_pool = "6gig", deps = [ + "//source/common/thread_local:thread_local_lib", "//source/extensions/common/aws:credentials_provider_interface", "//source/extensions/common/aws:utility_lib", "//source/extensions/common/aws/signers:sigv4_signer_impl_lib", "//test/extensions/common/aws:aws_mocks", "//test/mocks/server:factory_context_mocks", "//test/test_common:status_utility_lib", + "//test/test_common:utility_lib", + "@abseil-cpp//absl/synchronization", ], ) diff --git a/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc b/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc index d515134eae286..e88823b28ef71 100644 --- a/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc +++ b/test/extensions/common/aws/credential_providers/assume_role_credentials_provider_test.cc @@ -973,8 +973,10 @@ TEST_F(AssumeRoleCredentialsProviderTest, WithExternalId) { EXPECT_EQ("test-access-key", credentials.accessKeyId().value()); } -// Tests ASAN failure when cancel wrapper is not used -TEST_F(AssumeRoleCredentialsProviderTest, CancelWrapperPreventsUseAfterFree) { +// setCredentialsToAllThreads() leaves an all-threads-complete callback outstanding, which may run +// after the provider is gone. It captures a weak_ptr, so firing it then must be a no-op rather than +// a use-after-free (this test fails under ASAN if the callback captures the provider directly). +TEST_F(AssumeRoleCredentialsProviderTest, DeferredCompletionCallbackSafeAfterDestruction) { std::function captured_callback; EXPECT_CALL(context_.thread_local_, runOnAllThreads(testing::_, testing::_)) diff --git a/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc b/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc index a034b8b689413..ceff72bc232d0 100644 --- a/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc +++ b/test/extensions/common/aws/credential_providers/instance_profile_credentials_provider_test.cc @@ -769,8 +769,10 @@ not json delete (raw_metadata_fetcher_); } -// Tests ASAN failure when cancel wrapper is not used -TEST_F(InstanceProfileCredentialsProviderTest, CancelWrapperPreventsUseAfterFree) { +// setCredentialsToAllThreads() leaves an all-threads-complete callback outstanding, which may run +// after the provider is gone. It captures a weak_ptr, so firing it then must be a no-op rather than +// a use-after-free (this test fails under ASAN if the callback captures the provider directly). +TEST_F(InstanceProfileCredentialsProviderTest, DeferredCompletionCallbackSafeAfterDestruction) { std::function captured_callback; EXPECT_CALL(context_.thread_local_, runOnAllThreads(testing::_, testing::_)) diff --git a/test/extensions/common/aws/credentials_provider_test.cc b/test/extensions/common/aws/credentials_provider_test.cc index 34f00df16b847..f5d1c65e095c7 100644 --- a/test/extensions/common/aws/credentials_provider_test.cc +++ b/test/extensions/common/aws/credentials_provider_test.cc @@ -1,4 +1,5 @@ #include "source/common/http/message_impl.h" +#include "source/common/thread_local/thread_local_impl.h" #include "source/extensions/common/aws/credentials_provider.h" #include "source/extensions/common/aws/signers/sigv4_signer_impl.h" @@ -6,12 +7,15 @@ #include "test/mocks/event/mocks.h" #include "test/mocks/server/server_factory_context.h" #include "test/test_common/status_utility.h" +#include "test/test_common/utility.h" +#include "absl/synchronization/notification.h" #include "gtest/gtest.h" using Envoy::Extensions::Common::Aws::MetadataFetcherPtr; using testing::MockFunction; using testing::Return; +using testing::ReturnRef; namespace Envoy { namespace Extensions { @@ -163,8 +167,10 @@ TEST_F(AsyncCredentialHandlingTest, ChainCallbackCalledWhenCredentialsReturned) timer_ = new NiceMock(&context_.dispatcher_); timer_->enableTimer(std::chrono::milliseconds(1), nullptr); + // Subscribers are notified once from the thread that initiates the update and again once + // every thread has applied it. ThreadLocal::MockInstance runs both synchronously. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); EXPECT_CALL(*chain, chainGetCredentials()).WillRepeatedly(Return(Credentials("akid", "skid"))); auto document = R"EOF( @@ -351,8 +357,10 @@ TEST_F(AsyncCredentialHandlingTest, SubscriptionsCleanedUp) { timer_ = new NiceMock(&context_.dispatcher_); timer_->enableTimer(std::chrono::milliseconds(1), nullptr); + // Subscribers are notified once from the thread that initiates the update and again once + // every thread has applied it. ThreadLocal::MockInstance runs both synchronously. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); EXPECT_CALL(*chain, chainGetCredentials()).WillRepeatedly(Return(Credentials("akid", "skid"))); auto chain2 = std::make_shared(); @@ -509,9 +517,10 @@ TEST_F(AsyncCredentialHandlingTest, WeakPtrProtectionInSubscriberCallback) { auto provider_friend = MetadataCredentialsProviderBaseFriend(provider_); - // Test 1: When subscriber is alive, onCredentialUpdate should be called + // Test 1: When subscriber is alive, onCredentialUpdate should be called. It is called once from + // the updating thread and again once all threads have applied the update. auto chain = std::make_shared(); - EXPECT_CALL(*chain, onCredentialUpdate()); + EXPECT_CALL(*chain, onCredentialUpdate()).Times(2); auto handle = provider_->subscribeToCredentialUpdates(chain); // Trigger credential update @@ -649,6 +658,311 @@ TEST(CredentialsProviderChainTest, CheckChainReturnsPendingInCorrectOrder) { EXPECT_EQ(creds.secretAccessKey(), "1"); } +// Regression tests for a deadlock between credential resolution and worker thread startup. +// +// credentials_pending_ used to be cleared from the all-threads-complete callback of +// ThreadLocal::Slot::runOnAllThreads(). That callback only fires once every registered worker +// dispatcher has run the update, and worker dispatchers are registered when the ListenerManager is +// constructed but do not start running until startWorkers(), which itself waits on the server init +// manager. Any credential refresh that resolved before that point latched credentials_pending_ at +// true forever and stalled every signer depending on it. +// +// These tests use a real ThreadLocal::InstanceImpl with a registered worker dispatcher that is +// deliberately never run(), which is exactly the pre-startWorkers() state. Every other AWS test +// uses ThreadLocal::MockInstance, whose runOnAllThreads() invokes both callbacks synchronously, so +// the barrier is never real there. +class CredentialsPendingBeforeWorkerStartupTest : public testing::Test { +public: + CredentialsPendingBeforeWorkerStartupTest() + : api_(Api::createApiForTest()), main_dispatcher_(api_->allocateDispatcher("test_main")), + worker_dispatcher_(api_->allocateDispatcher("test_worker")), + mock_manager_(std::make_shared()) { + tls_.registerThread(*main_dispatcher_, /*main_thread=*/true); + // Registered but never run(), modelling a worker thread that has not been started yet. + tls_.registerThread(*worker_dispatcher_, /*main_thread=*/false); + + // Must be in place before the provider is constructed: MetadataCredentialsProviderBase's + // constructor allocates its tls slot from context_.threadLocal(). + ON_CALL(context_, threadLocal()).WillByDefault(ReturnRef(tls_)); + ON_CALL(*mock_manager_, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + } + + ~CredentialsPendingBeforeWorkerStartupTest() override { + // The provider owns a slot in tls_, so it has to go away while tls_ is still alive and not yet + // shut down. + provider_.reset(); + tls_.shutdownGlobalThreading(); + tls_.shutdownThread(); + } + + void createProvider() { + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + // These tests drive setCredentialsToAllThreads() directly rather than through a fetch, so the + // metadata fetcher is never created. + provider_ = std::make_shared( + context_, mock_manager_, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + } + + Api::ApiPtr api_; + Event::DispatcherPtr main_dispatcher_; + Event::DispatcherPtr worker_dispatcher_; + ThreadLocal::InstanceImpl tls_; + NiceMock context_; + std::shared_ptr mock_manager_; + WebIdentityCredentialsProviderPtr provider_; +}; + +// A successful credential refresh must un-pend the provider and notify its subscribers even though +// no worker dispatcher is running. +TEST_F(CredentialsPendingBeforeWorkerStartupTest, UnpendsBeforeWorkerDispatchersRun) { + createProvider(); + EXPECT_TRUE(provider_->credentialsPending()); + + auto chain = std::make_shared(); + EXPECT_CALL(*chain, onCredentialUpdate()); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + + EXPECT_FALSE(provider_->credentialsPending()); + // The main thread's slot is written synchronously by runOnAllThreads(), so the credentials are + // readable here as well. + const auto credentials = provider_->getCredentials(); + EXPECT_EQ("akid", credentials.accessKeyId()); + EXPECT_EQ("secret", credentials.secretAccessKey()); + EXPECT_EQ("token", credentials.sessionToken()); +} + +// The failure path must un-pend too. +TEST_F(CredentialsPendingBeforeWorkerStartupTest, UnpendsOnRetrievalErrorBeforeWorkersRun) { + createProvider(); + EXPECT_TRUE(provider_->credentialsPending()); + + auto chain = std::make_shared(); + EXPECT_CALL(*chain, onCredentialUpdate()); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + MetadataCredentialsProviderBaseFriend(provider_).credentialsRetrievalError(); + + EXPECT_FALSE(provider_->credentialsPending()); + EXPECT_FALSE(provider_->getCredentials().accessKeyId().has_value()); +} + +// A thread must never observe "credentials are no longer pending" before the credential update +// itself has been applied to that thread. The pending flag and the credentials both live in the +// thread local cache and move as one update, so the pair is always consistent per thread. +// +// This test parks a running worker dispatcher inside one of its own callbacks, performs a +// credential update from the main thread while the worker is stuck there, and then lets the worker +// observe both halves. +class CredentialsPendingWorkerVisibilityTest : public testing::Test { +public: + CredentialsPendingWorkerVisibilityTest() + : api_(Api::createApiForTest()), main_dispatcher_(api_->allocateDispatcher("test_main")), + worker_dispatcher_(api_->allocateDispatcher("test_worker")), + mock_manager_(std::make_shared()) { + tls_.registerThread(*main_dispatcher_, /*main_thread=*/true); + tls_.registerThread(*worker_dispatcher_, /*main_thread=*/false); + + ON_CALL(context_, threadLocal()).WillByDefault(ReturnRef(tls_)); + ON_CALL(*mock_manager_, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + } + + ~CredentialsPendingWorkerVisibilityTest() override { + if (worker_thread_ != nullptr) { + worker_dispatcher_->exit(); + worker_thread_->join(); + } + provider_.reset(); + tls_.shutdownGlobalThreading(); + tls_.shutdownThread(); + } + + void createProvider() { + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + provider_ = std::make_shared( + context_, mock_manager_, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + } + + // Started after the provider exists, so the slot initializer that the provider's constructor + // posted is already queued ahead of anything this test posts. RunUntilExit is what production + // worker threads use: RunType::Block returns as soon as the event set drains, which for a + // dispatcher with no listeners on it means the thread would exit between posts. + void startWorker() { + worker_thread_ = api_->threadFactory().createThread( + [this]() { worker_dispatcher_->run(Event::Dispatcher::RunType::RunUntilExit); }); + } + + // Runs cb on the worker thread and waits for it to finish. + void runOnWorker(std::function cb) { + absl::Notification done; + worker_dispatcher_->post([&]() { + cb(); + done.Notify(); + }); + done.WaitForNotification(); + } + + Api::ApiPtr api_; + Event::DispatcherPtr main_dispatcher_; + Event::DispatcherPtr worker_dispatcher_; + Thread::ThreadPtr worker_thread_; + ThreadLocal::InstanceImpl tls_; + NiceMock context_; + std::shared_ptr mock_manager_; + WebIdentityCredentialsProviderPtr provider_; +}; + +TEST_F(CredentialsPendingWorkerVisibilityTest, WorkerNeverSeesUnpendedStaleCredentials) { + createProvider(); + startWorker(); + + absl::Notification worker_parked; + absl::Notification release_worker; + absl::Notification observation_done; + bool observed_pending = false; + bool observed_has_credentials = false; + + // Occupies the worker's event loop, so the credential update posted below is queued behind this + // callback and cannot be applied until it returns. + worker_dispatcher_->post([&]() { + worker_parked.Notify(); + release_worker.WaitForNotification(); + observed_pending = provider_->credentialsPending(); + observed_has_credentials = provider_->getCredentials().hasCredentials(); + observation_done.Notify(); + }); + worker_parked.WaitForNotification(); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + // Visible on the main thread immediately, which is what lets a callout made before startWorkers() + // be signed. + EXPECT_FALSE(provider_->credentialsPending()); + EXPECT_TRUE(provider_->getCredentials().hasCredentials()); + + release_worker.Notify(); + observation_done.WaitForNotification(); + + // The worker had not applied the update yet, so it has to still report pending. + EXPECT_TRUE(observed_pending); + EXPECT_FALSE(observed_has_credentials); + + // Once it drains its post queue, both halves are visible together. + bool drained_pending = true; + bool drained_has_credentials = false; + runOnWorker([&]() { + drained_pending = provider_->credentialsPending(); + drained_has_credentials = provider_->getCredentials().hasCredentials(); + }); + EXPECT_FALSE(drained_pending); + EXPECT_TRUE(drained_has_credentials); +} + +// A pending callback registered by a worker after the updating thread has already notified +// subscribers must still be woken. The worker keeps reading credentials_pending_ == true from its +// own slot until it applies the posted update, so it can queue a callback onto a queue that has +// just been drained. The all-threads-complete notification is what rescues it; without it the +// request stalls until the next successful refresh, ~1 hour later. +TEST_F(CredentialsPendingWorkerVisibilityTest, PendingCallbackRegisteredDuringStaleWindowIsWoken) { + createProvider(); + startWorker(); + + auto chain = std::make_shared(); + chain->add(provider_); + auto handle = provider_->subscribeToCredentialUpdates(chain); + + absl::Notification worker_parked; + absl::Notification release_worker; + absl::Notification registered; + bool was_pending = false; + std::atomic callback_fired{false}; + + // Parks the worker's event loop so that the credential update posted below cannot be applied to + // the worker's slot until we let it go. + worker_dispatcher_->post([&]() { + worker_parked.Notify(); + release_worker.WaitForNotification(); + // The worker's slot still says pending, so the chain queues this callback - after the main + // thread has already drained the queue. + was_pending = + chain->addCallbackIfChainCredentialsPending([&callback_fired]() { callback_fired = true; }); + registered.Notify(); + }); + worker_parked.WaitForNotification(); + + MetadataCredentialsProviderBaseFriend(provider_).setCredentialsToAllThreads( + std::make_unique("akid", "secret", "token")); + + release_worker.Notify(); + registered.WaitForNotification(); + EXPECT_TRUE(was_pending); + EXPECT_FALSE(callback_fired); + + // Let the worker drain the credential update. Releasing the last reference to the update callback + // posts the all-threads-complete callback onto the main dispatcher. + runOnWorker([]() {}); + main_dispatcher_->run(Event::Dispatcher::RunType::NonBlock); + + EXPECT_TRUE(callback_fired); +} + +// Marking pending again while already pending must not broadcast. +// InstanceProfileCredentialsProvider marks pending once per stage of its three stage fetch, and +// only the first transition needs an update posted to every thread. +TEST(MetadataCredentialsProviderPendingTest, MarkingPendingIsDeduplicated) { + NiceMock context; + auto mock_manager = std::make_shared(); + ON_CALL(*mock_manager, getUriFromClusterName(_)).WillByDefault(Return("uri_2")); + + envoy::extensions::common::aws::v3::AssumeRoleWithWebIdentityCredentialProvider cred_provider = + {}; + cred_provider.mutable_web_identity_token_data_source()->set_inline_string("abced"); + cred_provider.set_role_arn("aws:iam::123456789012:role/arn"); + cred_provider.set_role_session_name("role-session-name"); + + auto provider = std::make_shared( + context, mock_manager, "cluster_2", + [](Upstream::ClusterManager&, absl::string_view) { return MetadataFetcherPtr{}; }, + MetadataFetcher::MetadataReceiver::RefreshState::Ready, std::chrono::seconds(2), + cred_provider); + auto provider_friend = MetadataCredentialsProviderBaseFriend(provider); + + // A provider starts out pending, so there is nothing to broadcast. + EXPECT_CALL(context.thread_local_, runOnAllThreads(_)).Times(0); + provider_friend.setCredentialsPendingToAllThreads(); + EXPECT_TRUE(provider->credentialsPending()); + testing::Mock::VerifyAndClearExpectations(&context.thread_local_); + + // Delivering the credentials and clearing the flag uses the two-argument overload, so that + // subscribers are notified again once every thread has applied the update. Setting the flag again + // is then the only single-argument broadcast: the second, redundant call posts nothing. + EXPECT_CALL(context.thread_local_, runOnAllThreads(_, _)); + EXPECT_CALL(context.thread_local_, runOnAllThreads(_)); + provider_friend.setCredentialsToAllThreads(std::make_unique("akid", "secret")); + EXPECT_FALSE(provider->credentialsPending()); + provider_friend.setCredentialsPendingToAllThreads(); + provider_friend.setCredentialsPendingToAllThreads(); + EXPECT_TRUE(provider->credentialsPending()); +} + } // namespace Aws } // namespace Common } // namespace Extensions diff --git a/test/extensions/common/aws/mocks.h b/test/extensions/common/aws/mocks.h index cb2a9f26ae839..9f015f767c64b 100644 --- a/test/extensions/common/aws/mocks.h +++ b/test/extensions/common/aws/mocks.h @@ -163,6 +163,8 @@ class MetadataCredentialsProviderBaseFriend { void setCredentialsToAllThreads(CredentialsConstUniquePtr&& creds) { provider_->setCredentialsToAllThreads(std::move(creds)); } + void setCredentialsPendingToAllThreads() { provider_->setCredentialsPendingToAllThreads(); } + void credentialsRetrievalError() { provider_->credentialsRetrievalError(); } void invalidateStats() { provider_->stats_.reset(); } size_t getSubscribersCount() { Thread::LockGuard lock(provider_->mu_); diff --git a/test/extensions/dynamic_modules/bootstrap/BUILD b/test/extensions/dynamic_modules/bootstrap/BUILD index e83e7a4668abb..398d5fa016f2d 100644 --- a/test/extensions/dynamic_modules/bootstrap/BUILD +++ b/test/extensions/dynamic_modules/bootstrap/BUILD @@ -129,11 +129,14 @@ envoy_cc_test( "//test/extensions/dynamic_modules/test_data/rust:bootstrap_integration_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_listener_lifecycle_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_shared_data_test", + "//test/extensions/dynamic_modules/test_data/rust:bootstrap_signed_callout_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_stats_test", "//test/extensions/dynamic_modules/test_data/rust:bootstrap_timer_test", ], deps = [ + "//source/common/router:upstream_codec_filter_lib", "//source/extensions/bootstrap/dynamic_modules:config", + "//source/extensions/filters/http/aws_request_signing:config", "//source/extensions/filters/http/dynamic_modules:factory_registration", "//test/integration:http_integration_lib", "//test/test_common:environment_lib", diff --git a/test/extensions/dynamic_modules/bootstrap/integration_test.cc b/test/extensions/dynamic_modules/bootstrap/integration_test.cc index b6e6e9044f3d9..f277b79b077df 100644 --- a/test/extensions/dynamic_modules/bootstrap/integration_test.cc +++ b/test/extensions/dynamic_modules/bootstrap/integration_test.cc @@ -316,6 +316,67 @@ name: envoy.extensions.filters.http.dynamic_modules } } +const std::string AWS_REQUEST_SIGNING_UPSTREAM_FILTER = R"EOF( +name: envoy.filters.http.aws_request_signing +typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.aws_request_signing.v3.AwsRequestSigning + service_name: execute-api + region: us-east-1 + signing_algorithm: aws_sigv4 + credential_provider: + custom_credential_provider_chain: true + container_credential_provider: {} +)EOF"; + +class DynamicModulesBootstrapAwsSigningIntegrationTest + : public DynamicModulesBootstrapIntegrationTest { +public: + DynamicModulesBootstrapAwsSigningIntegrationTest() { + // Nothing is listening here, so the fetch fails, anonymous credentials are installed, and the + // callout goes out unsigned. That doesn't matter. The deadlock is in clearing the pending flag, + // not in the signature, and both the success and the failure path of the fetch reach it + // through setCredentialsToAllThreads(). + TestEnvironment::setEnvVar("AWS_CONTAINER_CREDENTIALS_FULL_URI", + "http://127.0.0.1:1/path/to/creds", 1); + } + + ~DynamicModulesBootstrapAwsSigningIntegrationTest() override { + // Undo environment changes. + TestEnvironment::unsetEnvVar("AWS_CONTAINER_CREDENTIALS_FULL_URI"); + } +}; + +INSTANTIATE_TEST_SUITE_P(IpVersions, DynamicModulesBootstrapAwsSigningIntegrationTest, + testing::ValuesIn(TestEnvironment::getIpVersionsForTest()), + TestUtility::ipTestParamsToString); + +// Regression test for a deadlock between the bootstrap init target and AWS credential resolution. +// The module holds its init target open until an HTTP callout through cluster_0 completes, and +// cluster_0 carries an upstream aws_request_signing filter whose credentials chain has no +// synchronous provider, so signing has to wait on an async metadata fetch. +// +// That fetch resolves on the main thread before any worker thread exists. While the resulting +// "credentials are no longer pending" notification was driven from the all-threads-complete +// callback of runOnAllThreads(), it could never fire here: that callback waits for every +// registered worker dispatcher to run the update, worker dispatchers do not run until +// startWorkers(), and startWorkers() waits on the very init manager this module is holding open. +// The signing filter then held the callout until its deadline. Reaching the module's success log at +// all is the assertion; a regression instead fails on the harness giving up waiting for listeners, +// and the module budgets its retries so it cannot spin indefinitely behind that. +TEST_P(DynamicModulesBootstrapAwsSigningIntegrationTest, SignedCalloutGatingInitTarget) { + // Nothing is servicing the fake upstream while the server is still initializing, so it has to + // answer the callout on its own. + autonomous_upstream_ = true; + config_helper_.prependFilter(AWS_REQUEST_SIGNING_UPSTREAM_FILTER, /*downstream=*/false); + // cluster_0 carries no protocol options in the base config, and upstream_protocol_options is a + // required field once any are set. This fills it in without disturbing the filter chain above. + setUpstreamProtocol(Http::CodecType::HTTP1); + + EXPECT_LOG_CONTAINS( + "info", "Bootstrap signed callout test completed successfully!", + initializeWithBootstrapExtension(testDataDir("rust"), "bootstrap_signed_callout_test")); +} + } // namespace DynamicModules } // namespace Bootstrap } // namespace Extensions diff --git a/test/extensions/dynamic_modules/test_data/rust/BUILD b/test/extensions/dynamic_modules/test_data/rust/BUILD index 3bb811cdf6ba5..316fa652fd9ab 100644 --- a/test/extensions/dynamic_modules/test_data/rust/BUILD +++ b/test/extensions/dynamic_modules/test_data/rust/BUILD @@ -49,6 +49,8 @@ test_program(name = "bootstrap_init_target_test") test_program(name = "bootstrap_timer_test") +test_program(name = "bootstrap_signed_callout_test") + test_program(name = "bootstrap_file_watcher_test") test_program(name = "bootstrap_admin_handler_test") diff --git a/test/extensions/dynamic_modules/test_data/rust/Cargo.toml b/test/extensions/dynamic_modules/test_data/rust/Cargo.toml index 320980f29cd54..4b2e3e957dcbb 100644 --- a/test/extensions/dynamic_modules/test_data/rust/Cargo.toml +++ b/test/extensions/dynamic_modules/test_data/rust/Cargo.toml @@ -74,6 +74,12 @@ path = "bootstrap_timer_test.rs" crate-type = ["cdylib"] test = true +[[example]] +name = "bootstrap_signed_callout_test" +path = "bootstrap_signed_callout_test.rs" +crate-type = ["cdylib"] +test = true + [[example]] name = "bootstrap_admin_handler_test" path = "bootstrap_admin_handler_test.rs" diff --git a/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs b/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs new file mode 100644 index 0000000000000..addcc11f4b168 --- /dev/null +++ b/test/extensions/dynamic_modules/test_data/rust/bootstrap_signed_callout_test.rs @@ -0,0 +1,214 @@ +//! Test module reproducing the shape of a bootstrap extension that gates server initialization on +//! an AWS-signed HTTP callout. +//! +//! The module holds its init target open from `new_bootstrap_extension_config` and only signals +//! completion once an HTTP callout through `cluster_0` has returned a 2xx. The callout is issued +//! with a request body so that an upstream `aws_request_signing` filter on that cluster takes its +//! `decodeData` signing path, which is where a request blocks while AWS credentials are still +//! pending. +//! +//! Combined with a credentials chain that has no synchronous provider, this closes a loop that +//! used to deadlock: signing waited on an async credential refresh, that refresh could not +//! resolve until worker threads were running, and worker threads could not start until this init +//! target completed. +//! +//! Failures are retried rather than given up on immediately, so that a slow environment does not +//! turn into a flake, but the retries are budgeted so the module cannot loop forever. Once the +//! budget is exhausted it signals init complete without logging its success message, which fails +//! the integration test's log assertion. (A regression is bounded either way, because the test +//! harness gives up waiting for listeners after 20 seconds.) + +use envoy_proxy_dynamic_modules_rust_sdk::*; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +declare_bootstrap_init_functions!(my_program_init, my_new_bootstrap_extension_config_fn); + +/// Cluster the callout is sent to. This is the default static cluster of the integration test, +/// which carries the upstream `aws_request_signing` filter. +const CALLOUT_CLUSTER: &str = "cluster_0"; + +/// Callout deadline. Generous relative to the credential refresh it waits behind, so that a +/// timeout here means the request never got signed rather than that it was merely slow. +const CALLOUT_TIMEOUT_MS: u64 = 5000; + +/// Delay before the first callout attempt and between retries. +const CALLOUT_DELAY: Duration = Duration::from_millis(100); + +/// How many callouts may complete unsuccessfully before the module stops retrying. With the bug +/// this reproduces every attempt burns the full `CALLOUT_TIMEOUT_MS`, so this keeps the total under +/// half a minute. +const MAX_CALLOUT_FAILURES: u32 = 4; + +/// How many times the callout may fail to start, which happens while `cluster_0` is still warming +/// up. These retries cost only `CALLOUT_DELAY` each, so the budget is larger. +const MAX_START_FAILURES: u32 = 50; + +/// Scheduler event id used to hop from `on_server_initialized` back onto the config. +const EVENT_SERVER_INITIALIZED: u64 = 1; + +fn my_program_init() -> bool { + true +} + +fn my_new_bootstrap_extension_config_fn( + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _name: &str, + _config: &[u8], +) -> Option> { + // Deliberately do NOT signal init complete here. The init target stays open until the callout + // below succeeds, which is the whole point of this module. + let timer = envoy_extension_config.new_timer(); + let scheduler = Arc::new(envoy_extension_config.new_scheduler()); + envoy_log_info!("signed callout module: config created, init target held open"); + Some(Box::new(SignedCalloutConfig { + timer, + scheduler, + init_signaled: AtomicBool::new(false), + callout_failures: AtomicU32::new(0), + start_failures: AtomicU32::new(0), + })) +} + +struct SignedCalloutConfig { + timer: Box, + scheduler: Arc>, + init_signaled: AtomicBool, + callout_failures: AtomicU32, + start_failures: AtomicU32, +} + +impl SignedCalloutConfig { + /// Signals init complete at most once, so a retry that races a success cannot double-signal. + /// Returns whether this call was the one that signalled. + fn signal_init_complete_once( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + ) -> bool { + let signalled = self + .init_signaled + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(); + if signalled { + envoy_extension_config.signal_init_complete(); + } + signalled + } + + /// Unblocks server initialization without logging the success message. The integration test + /// asserts on that message, so it fails on the assertion rather than waiting out the test + /// timeout. + fn give_up(&self, envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, reason: &str) { + envoy_log_info!("signed callout module: giving up, {}", reason); + self.signal_init_complete_once(envoy_extension_config); + } +} + +impl BootstrapExtensionConfig for SignedCalloutConfig { + fn new_bootstrap_extension( + &self, + _envoy_extension: &mut dyn EnvoyBootstrapExtension, + ) -> Box { + Box::new(SignedCalloutExtension { + scheduler: self.scheduler.clone(), + }) + } + + fn on_scheduled( + &self, + _envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + event_id: u64, + ) { + if event_id == EVENT_SERVER_INITIALIZED { + self.timer.enable(CALLOUT_DELAY); + } + } + + fn on_timer_fired( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _timer: &dyn EnvoyBootstrapExtensionTimer, + ) { + // A body is sent so that the upstream aws_request_signing filter signs from decodeData + // (use_unsigned_payload defaults to false). + let (result, callout_id) = envoy_extension_config.send_http_callout( + CALLOUT_CLUSTER, + vec![ + (":method", b"POST"), + (":path", b"/oauth2/token"), + (":authority", b"authorizer"), + ], + Some(b"grant_type=client_credentials"), + CALLOUT_TIMEOUT_MS, + ); + if result != abi::envoy_dynamic_module_type_http_callout_init_result::Success { + // The cluster may not be warm yet on the first attempt; keep retrying within budget. + if self.start_failures.fetch_add(1, Ordering::SeqCst) + 1 > MAX_START_FAILURES { + self.give_up(envoy_extension_config, "callout never started"); + return; + } + envoy_log_info!("signed callout module: callout could not be started, retrying"); + self.timer.enable(CALLOUT_DELAY); + return; + } + envoy_log_info!("signed callout module: callout {} in flight", callout_id); + } + + fn on_http_callout_done( + &self, + envoy_extension_config: &mut dyn EnvoyBootstrapExtensionConfig, + _callout_id: u64, + result: abi::envoy_dynamic_module_type_http_callout_result, + response_headers: Option<&[(EnvoyBuffer, EnvoyBuffer)]>, + _response_body: Option<&[EnvoyBuffer]>, + ) { + let status = if result == abi::envoy_dynamic_module_type_http_callout_result::Success { + response_headers + .unwrap_or_default() + .iter() + .find(|(name, _)| name.as_slice() == b":status") + .and_then(|(_, value)| std::str::from_utf8(value.as_slice()).ok()) + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) + } else { + 0 + }; + + if (200..300).contains(&status) { + if self.signal_init_complete_once(envoy_extension_config) { + envoy_log_info!("Bootstrap signed callout test completed successfully!"); + } + return; + } + + // Anything else -- a reset, or the 504 the router emits when signing never completes and the + // callout deadline expires -- is retried within budget, so a slow environment does not flake. + if self.callout_failures.fetch_add(1, Ordering::SeqCst) + 1 > MAX_CALLOUT_FAILURES { + self.give_up( + envoy_extension_config, + "callout never returned a successful status", + ); + return; + } + envoy_log_info!( + "signed callout module: callout failed with status {}, retrying", + status + ); + self.timer.enable(CALLOUT_DELAY); + } +} + +struct SignedCalloutExtension { + scheduler: Arc>, +} + +impl BootstrapExtension for SignedCalloutExtension { + fn on_server_initialized(&mut self, _envoy_extension: &mut dyn EnvoyBootstrapExtension) { + // The cluster manager is only reachable from the config once the server is initialized, so + // the callout cannot be issued any earlier than this. Hop back onto the config to arm the + // timer that sends it. + envoy_log_info!("signed callout module: server initialized, scheduling callout"); + self.scheduler.commit(EVENT_SERVER_INITIALIZED); + } +}