diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index c6522a625c..c2cfe52955 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/sdk/core/azure-core-amqp/CHANGELOG.md @@ -10,6 +10,9 @@ ### Bugs Fixed +- The connection no longer returns a cached CBS token that is at or near its expiry. It authenticates the audience again instead. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) +- The connection now replaces each cached CBS token before that token expires, so a client that runs for longer than one token lifetime keeps working. This refresh applies to the uAMQP transport. Without this refresh, a send that gets the `amqp:unauthorized-access` condition stops at the first attempt, because the Event Hubs producer treats that condition as not transient. [[#7254]](https://github.com/Azure/azure-sdk-for-cpp/issues/7254) + ### Other Changes ## 1.0.0-beta.12 (2026-05-14) diff --git a/sdk/core/azure-core-amqp/src/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/amqp/connection.cpp index 625af7523b..3fd8633197 100644 --- a/sdk/core/azure-core-amqp/src/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/amqp/connection.cpp @@ -9,12 +9,15 @@ #include "azure/core/amqp/models/amqp_value.hpp" #include "claims_based_security_impl.hpp" #include "connection_impl.hpp" +#include "private/token_refresh.hpp" #include "session_impl.hpp" #include #include +#include #include +#include using namespace Azure::Core::Diagnostics::_internal; using namespace Azure::Core::Diagnostics; @@ -127,6 +130,48 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal { }}}} // namespace Azure::Core::Amqp::_internal namespace Azure { namespace Core { namespace Amqp { namespace _detail { + + namespace { + // Put a token for one audience on the wire. Both the first authentication + // and the proactive refresh use this function, so the two paths stay the + // same. The caller holds the token mutex. + void PutTokenForAudience( + std::shared_ptr session, + CbsTokenType tokenType, + std::string const& audienceUrl, + std::string const& token, + Azure::DateTime const& expiresOn, + Azure::Core::Context const& context) + { + auto claimsBasedSecurity = std::make_shared(session); + auto cbsOpenStatus = claimsBasedSecurity->Open(context); + if (cbsOpenStatus != CbsOpenResult::Ok) + { + throw std::runtime_error("Could not open Claims Based Security object."); + } + + try + { + auto result + = claimsBasedSecurity->PutToken(tokenType, audienceUrl, token, expiresOn, context); + if (std::get<0>(result) != CbsOperationResult::Ok) + { + throw Azure::Core::Credentials::AuthenticationException( + "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result)) + + " reason: " + std::get<2>(result)); + } + Log::Stream(Logger::Level::Verbose) << "Close CBS object"; + claimsBasedSecurity->Close(context); + } + catch (...) + { + // Ensure that the claims based security object is closed before we leave this scope. + claimsBasedSecurity->Close(context); + throw; + } + } + } // namespace + bool ConnectionImpl::IsSasCredential() const { if (GetCredential()) @@ -170,16 +215,34 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { } } - std::unique_lock lock(m_tokenMutex); + std::unique_lock lock(m_tokenState->Mutex); // If we have authenticated this audience, we're done and can return success. - auto token = m_tokenStore.find(audienceUrl); - if (token != m_tokenStore.end()) + // A cached token is only good while it has enough life left to use. A token + // that is at or near its expiry is discarded here, so the audience is + // authenticated again below. + auto token = m_tokenState->TokenStore.find(audienceUrl); + if (token != m_tokenState->TokenStore.end()) { + if (IsCachedTokenUsable(token->second, std::chrono::system_clock::now())) + { + if (m_options.EnableTrace) + { + Log::Stream(Logger::Level::Verbose) << "Using cached token for " << audienceUrl; + } +#if ENABLE_UAMQP + // Point the refresh thread at a session that is in use now. The + // session that first authenticated this audience can be gone while + // another session still uses the token. + m_tokenState->TokenSessions[audienceUrl] = session; +#endif + return token->second; + } if (m_options.EnableTrace) { - Log::Stream(Logger::Level::Verbose) << "Using cached token for " << audienceUrl; + Log::Stream(Logger::Level::Verbose) << "Cached token for " << audienceUrl + << " is at or near expiry, authenticating again."; } - return token->second; + m_tokenState->TokenStore.erase(token); } // We've not authenticated this audience. // Authenticate it with the server @@ -190,55 +253,342 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { << "No cached token for " << audienceUrl << ", Authenticating."; } - auto claimsBasedSecurity = std::make_shared(session); - auto cbsOpenStatus = claimsBasedSecurity->Open(context); - if (cbsOpenStatus != CbsOpenResult::Ok) - { - throw std::runtime_error("Could not open Claims Based Security object."); - } + Credentials::TokenRequestContext requestContext; - try - { - Credentials::TokenRequestContext requestContext; - - requestContext.Scopes = m_options.AuthenticationScopes; - auto accessToken{GetCredential()->GetToken(requestContext, context)}; + requestContext.Scopes = m_options.AuthenticationScopes; + auto accessToken{GetCredential()->GetToken(requestContext, context)}; - auto result = claimsBasedSecurity->PutToken( + { +#if ENABLE_UAMQP + // Only one claims based security object may exist on this connection at + // a time. See m_cbsMutex. + std::lock_guard cbsLock(m_cbsMutex); +#endif + PutTokenForAudience( + session, (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt), audienceUrl, accessToken.Token, accessToken.ExpiresOn, context); - if (std::get<0>(result) != CbsOperationResult::Ok) + } + + if (m_options.EnableTrace) + { + Log::Stream(Logger::Level::Verbose) + << "Authenticated connection for audience " << audienceUrl << " successfully."; + } + + // Assign, do not emplace. A refreshed token must replace the token that is + // already in the cache. + m_tokenState->TokenStore[audienceUrl] = accessToken; +#if ENABLE_UAMQP + // Remember the session that authenticated this audience, so the refresh + // thread can put a new token on the same session. The pointer is weak, so + // the refresh thread never keeps a session alive. + m_tokenState->TokenSessions[audienceUrl] = session; + StartTokenRefresh(); +#endif + return accessToken; + } + else + { + Log::Stream(Logger::Level::Verbose) << "No credential, returning empty token."; + // If the connection is unauthenticated, then just return an empty access token. + return {}; + } + } + +#if ENABLE_UAMQP + // Start the refresh thread if it is not running yet. The caller holds the + // token mutex. + // + // A thread that stopped on an error stays joinable, so this function does not + // start a second one. That is deliberate. The connection then refreshes each + // token when a caller uses it, which is the behavior the error log describes. + void ConnectionImpl::StartTokenRefresh() + { + if (m_tokenState->Stop) + { + return; + } + if (!m_tokenRefreshThread.joinable()) + { + // The thread co-owns the shared state, so the state outlives this + // connection. The raw `this` pointer is good only while the state says + // the connection is alive. See TokenRefreshState. + m_tokenRefreshThread + = std::thread([this, state = m_tokenState]() { TokenRefreshThread(this, state); }); + } + m_tokenState->Cv.notify_all(); + } + + void ConnectionImpl::StopTokenRefresh() + { + // Cancel before the lock, not after it. The refresh thread does its network + // work without the token mutex, but it can hold that mutex at other times. + // A cancel that waits for the mutex could not stop an operation that is in + // flight, which is the operation this call must stop. Context::Cancel is + // safe to call from any thread. + m_tokenRefreshContext.Cancel(); + + std::thread threadToJoin; + { + std::unique_lock lock(m_tokenState->Mutex); + m_tokenState->Stop = true; + m_tokenState->Cv.notify_all(); + threadToJoin = std::move(m_tokenRefreshThread); + } + + // Join outside the lock, because the refresh thread takes the token mutex. + if (threadToJoin.joinable()) + { + if (threadToJoin.get_id() == std::this_thread::get_id()) + { + // The refresh thread is running this call, which means it released the + // last reference to this connection. It cannot join itself, so detach + // it. + // + // The detached thread outlives this connection, and that is safe. It + // holds a shared_ptr to the state block, so the mutex, the condition + // variable, the stop flag, and the token maps all stay alive. Stop is + // set above, so the thread comes back from the release, takes the mutex + // in the state block, sees Stop, and leaves without touching this + // connection. + threadToJoin.detach(); + } + else + { + threadToJoin.join(); + } + } + } + + // The body of the refresh thread. + // + // This is a static function, and `connection` is a raw pointer on purpose. + // The thread must not own the connection, because a thread that owns the + // object it refreshes keeps that object alive forever. The `state` block is + // shared instead. The thread uses `connection` only while it knows the + // connection is alive, which means until RefreshTokenForAudience reports that + // the connection can be gone. + void ConnectionImpl::TokenRefreshThread( + ConnectionImpl* connection, + std::shared_ptr state) + { + // A background thread must not let an exception escape, because that ends + // the process. Every failure of one refresh is handled inside + // RefreshTokenForAudience, so this catch is for the unexpected. + try + { + std::unique_lock lock(state->Mutex); + while (!state->Stop) + { + auto const now = std::chrono::system_clock::now(); + + // Idle wake time when no token is close to its expiry. + auto nextWake = now + IdleTokenRefreshPoll; + std::vector dueAudiences; + for (auto const& entry : state->TokenStore) { - throw Azure::Core::Credentials::AuthenticationException( - "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result)) - + " reason: " + std::get<2>(result)); + if (IsTokenRefreshDue(entry.second, now)) + { + dueAudiences.push_back(entry.first); + } + else if (IsTokenRefreshDue(entry.second, now + IdleTokenRefreshPoll)) + { + // This token is due inside the idle poll time, so wake for it. The + // expiry is close to now, which makes the cast safe. + auto const expiry + = static_cast(entry.second.ExpiresOn); + nextWake = (std::min)(nextWake, expiry - TokenRefreshBuffer); + } } - Log::Stream(Logger::Level::Verbose) << "Close CBS object"; - claimsBasedSecurity->Close(context); - if (m_options.EnableTrace) + + for (auto const& audience : dueAudiences) { - Log::Stream(Logger::Level::Verbose) - << "Authenticated connection for audience " << audienceUrl << " successfully."; + if (state->Stop) + { + break; + } + if (connection->RefreshTokenForAudience(*state, audience, lock)) + { + // That call released a session reference, which can have destroyed + // this connection. `connection` is not safe to use now, so leave. + // The state block stays alive until this thread returns. + return; + } } - m_tokenStore.emplace(audienceUrl, accessToken); - return accessToken; + if (!dueAudiences.empty()) + { + // Keep a minimum time between two refresh passes. A token with a + // lifetime shorter than the refresh buffer is due as soon as it + // arrives, and this wait stops the thread from spinning on it. + nextWake = std::chrono::system_clock::now() + MinimumTokenRefreshInterval; + } + + if (state->TokenStore.empty()) + { + // Nothing to refresh. Wait until an audience is authenticated, or + // until shutdown. + state->Cv.wait(lock, [&state]() { return state->Stop || !state->TokenStore.empty(); }); + } + else + { + state->Cv.wait_until(lock, nextWake, [&state]() { return state->Stop; }); + } } - catch (...) + } + catch (std::exception const& e) + { + Log::Stream(Logger::Level::Error) + << "The token refresh thread stopped on an error: " << e.what() + << ". Tokens are refreshed on use instead."; + } + catch (...) + { + Log::Stream(Logger::Level::Error) + << "The token refresh thread stopped on an unknown error. Tokens are refreshed on use " + "instead."; + } + } + + // Replace the token for one audience. + // + // The caller holds the token mutex through `lock`. This function releases that + // mutex for the credential call and for the CBS operation, because both go to + // the network. Holding the mutex there would block every caller that opens a + // link, and it would stop a shutdown from cancelling this work. + // + // This function holds a strong session pointer while the mutex is free, so it + // can hold the last reference to the session and, through the session, the + // last reference to this connection. ReleaseOutsideLock releases that + // pointer, always with the mutex free. The return value tells the caller + // whether this connection can be gone. + bool ConnectionImpl::RefreshTokenForAudience( + TokenRefreshState& state, + std::string const& audienceUrl, + std::unique_lock& lock) + { + std::shared_ptr promotedSession; + auto sessionEntry = state.TokenSessions.find(audienceUrl); + if (sessionEntry != state.TokenSessions.end()) + { + promotedSession = sessionEntry->second.lock(); + } + if (!promotedSession) + { + // The session that authenticated this audience is gone. Drop the entry, + // and the next link open authenticates the audience again. + state.TokenStore.erase(audienceUrl); + state.TokenSessions.erase(audienceUrl); + return false; + } + + // The hold owns this reference from here on, on every path out of this + // function. + ReleaseOutsideLock sessionHold{std::move(promotedSession), lock}; + + // Remember the token that this refresh replaces, so the result does not + // overwrite a newer token that a caller stored while the mutex was free. + auto const previousEntry = state.TokenStore.find(audienceUrl); + if (previousEntry == state.TokenStore.end()) + { + // A caller dropped this token while this pass was in flight. There is + // nothing to replace. Release the session here, so the flag that this + // function returns is read after the release. + sessionHold.Release(); + return state.Stop; + } + std::string const previousToken{previousEntry->second.Token}; + + auto const tokenType = (IsSasCredential() ? CbsTokenType::Sas : CbsTokenType::Jwt); + auto const credential = GetCredential(); + auto const scopes = m_options.AuthenticationScopes; + auto const traceEnabled = m_options.EnableTrace; + auto const parentContext = m_tokenRefreshContext; + + Credentials::AccessToken accessToken; + bool refreshed{false}; + std::string failureMessage; + + lock.unlock(); + try + { + auto context = parentContext.WithDeadline( + std::chrono::system_clock::now() + TokenRefreshOperationTimeout); + + Credentials::TokenRequestContext requestContext; + requestContext.Scopes = scopes; + accessToken = credential->GetToken(requestContext, context); + { - // Ensure that the claims based security object is closed before we leave this scope. - claimsBasedSecurity->Close(context); - throw; + // Only one claims based security object may exist on this connection at + // a time, so this waits for a caller that authenticates right now. See + // m_cbsMutex. This mutex is released before the token mutex is taken + // again, which keeps the lock order acyclic. + std::lock_guard cbsLock(m_cbsMutex); + PutTokenForAudience( + sessionHold.Get(), + tokenType, + audienceUrl, + accessToken.Token, + accessToken.ExpiresOn, + context); + } + refreshed = true; + } + catch (std::exception const& e) + { + failureMessage = e.what(); + } + catch (...) + { + failureMessage = "unknown error"; + } + // This is the one point where this thread can destroy the connection. A + // session destructor releases the session's reference to the connection, so + // this release can run the connection destructor on this thread. That + // destructor calls StopTokenRefresh, which sets Stop in the state block and + // detaches this thread. Every line after this one uses `state`, which this + // thread co-owns, and no member of this connection. + sessionHold.Release(); + lock.lock(); + + if (state.Stop) + { + // Either a shutdown is in progress, or the release above destroyed this + // connection. Do not touch this connection again. + return true; + } + + auto currentEntry = state.TokenStore.find(audienceUrl); + if (currentEntry == state.TokenStore.end() || currentEntry->second.Token != previousToken) + { + // A caller replaced or dropped this token while the mutex was free. That + // token is newer than this one, so keep it. + return false; + } + + if (refreshed) + { + currentEntry->second = accessToken; + if (traceEnabled) + { + Log::Stream(Logger::Level::Verbose) + << "Refreshed the token for audience " << audienceUrl << " before its expiry."; } } else { - Log::Stream(Logger::Level::Verbose) << "No credential, returning empty token."; - // If the connection is unauthenticated, then just return an empty access token. - return {}; + // Drop the cache entry, so the next link open authenticates again. + Log::Stream(Logger::Level::Warning) + << "Could not refresh the token for audience " << audienceUrl << ": " << failureMessage; + state.TokenStore.erase(currentEntry); + state.TokenSessions.erase(audienceUrl); } + return false; } +#endif // ENABLE_UAMQP }}}} // namespace Azure::Core::Amqp::_detail diff --git a/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp new file mode 100644 index 0000000000..130e9ed18e --- /dev/null +++ b/sdk/core/azure-core-amqp/src/amqp/private/token_refresh.hpp @@ -0,0 +1,149 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#pragma once + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace Azure { namespace Core { namespace Amqp { namespace _detail { + + class SessionImpl; + + // The state that the connection shares with the token refresh thread. + // + // The refresh thread promotes a weak session pointer to a strong one, so it + // can hold the last reference to a session and, through that session, the + // last reference to the connection. When it releases that reference, the + // connection destructor runs on the refresh thread. That destructor sets + // Stop, signals Cv, and detaches the thread, because a thread cannot join + // itself. + // + // The thread holds a shared_ptr to this block, so the block stays alive after + // the connection is gone. The thread comes back from the release, takes + // Mutex, sees Stop, and leaves without touching the connection. + struct TokenRefreshState final + { + // Protects every other member of this block. + std::mutex Mutex; + std::condition_variable Cv; + bool Stop{false}; + // The cached token for each audience. + std::map TokenStore; + // The session that authenticated each audience. The pointer is weak, so the + // refresh thread never keeps a session alive. + std::map> TokenSessions; + }; + + // Holds a shared pointer that a thread must not drop while it holds a lock, + // and releases it with that lock free. + // + // The token refresh thread promotes a weak session pointer for the length of + // one refresh. A session holds the connection, so that promoted pointer can + // be the last reference to both, and the release then runs the connection + // destructor on the refresh thread. That destructor takes the token mutex, so + // a release under the token mutex would lock a mutex that this thread already + // holds, and the mutex is not recursive. The release also runs other + // destructors that take other locks. + // + // Every exit from the scope runs this destructor, so no return path and no + // exception leaves the release under the lock. + template class ReleaseOutsideLock final { + public: + ReleaseOutsideLock(std::shared_ptr held, std::unique_lock& lock) + : m_held{std::move(held)}, m_lock{lock} + { + } + + ~ReleaseOutsideLock() { Release(); } + + ReleaseOutsideLock(ReleaseOutsideLock const&) = delete; + ReleaseOutsideLock& operator=(ReleaseOutsideLock const&) = delete; + ReleaseOutsideLock(ReleaseOutsideLock&&) = delete; + ReleaseOutsideLock& operator=(ReleaseOutsideLock&&) = delete; + + std::shared_ptr const& Get() const { return m_held; } + + // Release the pointer with the lock free, then put the lock back in the + // state it was in. This is safe to call more than once. + void Release() + { + if (!m_held) + { + return; + } + bool const wasLocked = m_lock.owns_lock(); + if (wasLocked) + { + m_lock.unlock(); + } + m_held.reset(); + if (wasLocked) + { + m_lock.lock(); + } + } + + private: + std::shared_ptr m_held; + std::unique_lock& m_lock; + }; + + // The .NET client refreshes a CBS token seven minutes before the token + // expires. See AmqpConnectionScope.cs in Azure.Messaging.EventHubs. Use the + // same buffer here. + constexpr std::chrono::minutes TokenRefreshBuffer{7}; + + // The smallest time between two refresh passes. A service that issues tokens + // with a lifetime shorter than the buffer makes every token due immediately. + // This interval stops the refresh thread from spinning in that case. + constexpr std::chrono::seconds MinimumTokenRefreshInterval{20}; + + // The refresh thread does the early refresh. The cache only has to stop + // itself from giving a caller a token that is about to die, so its margin is + // small. A larger margin here would make the cache authenticate again on + // every call for a token with a short lifetime. + constexpr std::chrono::seconds MinimumTokenLifetimeToUse{30}; + + // How long the refresh thread sleeps when no token is close to its expiry. + constexpr std::chrono::minutes IdleTokenRefreshPoll{1}; + + // The deadline for one refresh, which covers the credential call and the CBS + // operation. + constexpr std::chrono::seconds TokenRefreshOperationTimeout{60}; + + // These functions compare in the Azure::DateTime domain on purpose. The cast + // from Azure::DateTime to std::chrono::system_clock::time_point is explicit + // and it throws when the value is outside the range of the system clock. A + // credential can return any value in ExpiresOn, and a default constructed + // ExpiresOn is year 1, so a cast here could throw on a caller thread or on + // the refresh thread. The conversion in the other direction does not throw. + + // Return true while the cached token has enough life left to give to a + // caller. + inline bool IsCachedTokenUsable( + Azure::Core::Credentials::AccessToken const& token, + std::chrono::system_clock::time_point now) + { + return token.ExpiresOn > Azure::DateTime(now + MinimumTokenLifetimeToUse); + } + + // Return true when the refresh thread must replace this token by the given + // time. The refresh is due one buffer before the token expires, so the test + // adds the buffer to the time instead of subtracting it from the expiry. + // Adding avoids an underflow for a token that reports a very early expiry. + inline bool IsTokenRefreshDue( + Azure::Core::Credentials::AccessToken const& token, + std::chrono::system_clock::time_point now) + { + return token.ExpiresOn <= Azure::DateTime(now + TokenRefreshBuffer); + } + +}}}} // namespace Azure::Core::Amqp::_detail diff --git a/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp index ba4813a624..025c163f23 100644 --- a/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/rust_amqp/amqp/private/connection_impl.hpp @@ -3,6 +3,7 @@ #pragma once +#include "../../../../amqp/private/token_refresh.hpp" #include "azure/core/amqp/internal/common/global_state.hpp" #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/network/transport.hpp" @@ -153,10 +154,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { bool m_connectionOpened{false}; std::atomic m_openCount{0}; - // mutex protecting the token acquisition process. - std::mutex m_tokenMutex; std::shared_ptr m_credential{}; - std::map m_tokenStore; + + // The token mutex and the token cache. The Rust stack has no refresh + // thread, so nothing else shares this block here. + std::shared_ptr m_tokenState{std::make_shared()}; #if ENABLE_UAMQP ConnectionImpl( diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp index 08fe657e1a..15a932562b 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/connection.cpp @@ -148,6 +148,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { ConnectionImpl::~ConnectionImpl() { + // Stop the refresh thread first. It uses this connection, so it must not be + // running while the connection is torn down. + StopTokenRefresh(); + std::unique_lock lock(m_amqpMutex); if (m_openCount.load() != 0) { @@ -435,6 +439,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { throw std::logic_error("Connection not opened."); } + // Stop the refresh thread before polling stops, because a CBS operation + // needs the connection to poll. + StopTokenRefresh(); + // Stop polling on this connection, we're shutting it down. EnableAsyncOperation(false); @@ -463,6 +471,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { throw std::logic_error("Connection not opened."); } + // Stop the refresh thread before polling stops, because a CBS operation + // needs the connection to poll. + StopTokenRefresh(); + // Stop polling on this connection, we're shutting it down. EnableAsyncOperation(false); diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp index 26be189e63..3b67926b9f 100644 --- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp +++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/connection_impl.hpp @@ -3,19 +3,25 @@ #pragma once +#include "../../../../amqp/private/token_refresh.hpp" #include "../../../../amqp/private/unique_handle.hpp" #include "azure/core/amqp/internal/common/global_state.hpp" #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/network/transport.hpp" +#include #include #include #include #include +#include +#include #include +#include #include +#include #if defined(_MSC_VER) #define _azure_ACQUIRES_LOCK(...) _Acquires_exclusive_lock_(__VA_ARGS__) @@ -132,6 +138,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { std::string const& audience, Azure::Core::Context const& context); + // Stop the token refresh thread. This is safe to call more than once, and + // the connection calls it from Close and from the destructor. + void StopTokenRefresh(); + using LockType = std::recursive_mutex; _azure_ACQUIRES_LOCK(m_amqpMutex) std::unique_lock Lock() @@ -159,10 +169,48 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail { bool m_connectionOpened = false; std::atomic m_openCount{0}; - // mutex protecting the token acquisition process. - std::mutex m_tokenMutex; std::shared_ptr m_credential{}; - std::map m_tokenStore; + + // The token mutex, the token cache, and the stop protocol for the refresh + // thread. The connection and the refresh thread both own this block, so the + // thread finds it alive even after the connection is destroyed. See + // TokenRefreshState. + std::shared_ptr m_tokenState{std::make_shared()}; + + // Serializes the CBS operation itself. uAMQP names the CBS links after the + // node, so every claims based security object on this connection attaches a + // link called "$cbs-sender" and one called "$cbs-receiver". AMQP 1.0 section + // 2.6.1 requires a link name to be unique for one direction between two + // containers, so only one of these objects may exist at a time. The refresh + // thread does its work without the token mutex, so this mutex is what keeps + // the refresh and a caller apart. + // + // Lock order: a caller takes the token mutex and then this mutex. The + // refresh thread takes this mutex only while it does not hold the token + // mutex. + std::mutex m_cbsMutex; + + // The thread that replaces each cached token before the token expires. + std::thread m_tokenRefreshThread; + // Cancelled on shutdown, to stop a CBS operation that is in flight. + Azure::Core::Context m_tokenRefreshContext; + + void StartTokenRefresh(); + + // The body of the refresh thread. This is a static function on purpose. The + // thread can outlive the connection, so it must be able to finish without a + // live `this`. It uses `connection` only while the shared state says the + // connection is alive. + static void TokenRefreshThread( + ConnectionImpl* connection, + std::shared_ptr state); + + // Replace the token for one audience. Return true when the caller must stop + // at once, because this call can have destroyed the connection. + bool RefreshTokenForAudience( + TokenRefreshState& state, + std::string const& audienceUrl, + std::unique_lock& lock); ConnectionImpl( _internal::ConnectionEvents* eventHandler, diff --git a/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp b/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp index cec0ff665b..d767f3360b 100644 --- a/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/amqp_value_tests.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -563,6 +564,93 @@ TEST_F(TestValues, TestDescribed) } } +TEST_F(TestValues, ValueTypeOstreamInserter) +{ + // Invalid and Unknown are the two types that no AmqpValue can hold, so the other tests never + // reach them. The inserter still has to name them, because it reports the type of a value that + // the peer sent. + { + std::stringstream ss; + ss << AmqpValueType::Invalid; + EXPECT_EQ("Invalid", ss.str()); + } + { + std::stringstream ss; + ss << AmqpValueType::Unknown; + EXPECT_EQ("Unknown", ss.str()); + } +} + +// AmqpValue is the key type of AmqpMap, so its ordering must work for every value type, and not +// only for the scalar types. +TEST_F(TestValues, ValueOrdering) +{ + // Values of different types order by the numeric order of AmqpValueType. Null comes before Int. + EXPECT_LT(AmqpValue{}, AmqpValue{5}); + EXPECT_FALSE(AmqpValue{5} < AmqpValue{}); + + // Two null values are equal, so neither one is less than the other. + EXPECT_FALSE(AmqpValue{} < AmqpValue{}); + + { + AmqpValue smaller{U'a'}; + AmqpValue larger{U'b'}; + EXPECT_LT(smaller, larger); + EXPECT_FALSE(larger < smaller); + } + { + AmqpValue earlier{AmqpTimestamp{std::chrono::milliseconds{1000}}.AsAmqpValue()}; + AmqpValue later{AmqpTimestamp{std::chrono::milliseconds{2000}}.AsAmqpValue()}; + EXPECT_LT(earlier, later); + EXPECT_FALSE(later < earlier); + } + { + std::array smallerBytes{}; + std::array largerBytes{}; + largerBytes[0] = 1; + AmqpValue smaller{Azure::Core::Uuid::CreateFromArray(smallerBytes)}; + AmqpValue larger{Azure::Core::Uuid::CreateFromArray(largerBytes)}; + EXPECT_LT(smaller, larger); + EXPECT_FALSE(larger < smaller); + } + { + // Described values with the same descriptor order by their value. + AmqpDescribed smaller{AmqpSymbol{"descriptor"}, 1}; + AmqpDescribed larger{AmqpSymbol{"descriptor"}, 2}; + EXPECT_LT(smaller.AsAmqpValue(), larger.AsAmqpValue()); + EXPECT_FALSE(larger.AsAmqpValue() < smaller.AsAmqpValue()); + } + { + // Composite values order by their fields. + AmqpComposite smaller("composite", {1}); + AmqpComposite larger("composite", {2}); + EXPECT_LT(smaller.AsAmqpValue(), larger.AsAmqpValue()); + EXPECT_FALSE(larger.AsAmqpValue() < smaller.AsAmqpValue()); + } +} + +TEST_F(TestValues, CompositeAndDescribedEquality) +{ + // The underlying amqpvalue_are_equal does not handle composite or described values, so + // AmqpValue::operator== unwraps them and compares the unwrapped values. + { + AmqpComposite composite1("composite", {25, 25.0f}); + AmqpComposite composite2("composite", {25, 25.0f}); + AmqpComposite composite3("composite", {26, 25.0f}); + EXPECT_EQ(composite1.AsAmqpValue(), composite2.AsAmqpValue()); + EXPECT_NE(composite1.AsAmqpValue(), composite3.AsAmqpValue()); + } + { + AmqpDescribed described1{AmqpSymbol{"descriptor"}, 25}; + AmqpDescribed described2{AmqpSymbol{"descriptor"}, 25}; + AmqpDescribed described3{AmqpSymbol{"descriptor"}, 26}; + EXPECT_EQ(described1.AsAmqpValue(), described2.AsAmqpValue()); + EXPECT_NE(described1.AsAmqpValue(), described3.AsAmqpValue()); + } + // Values of different types are never equal, even when they read the same. + EXPECT_NE(AmqpValue{5}, AmqpValue{"5"}); +} + class TestValueSerialization : public testing::Test { protected: void SetUp() override {} diff --git a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp index 33d08d51cb..a77f6a4a52 100644 --- a/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp +++ b/sdk/core/azure-core-amqp/test/ut/connection_tests.cpp @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +#include "../../src/amqp/private/token_refresh.hpp" #include "azure/core/amqp/internal/common/async_operation_queue.hpp" #include "azure/core/amqp/internal/connection.hpp" #include "azure/core/amqp/internal/message_receiver.hpp" @@ -17,8 +18,12 @@ #include #include +#include #include +#include +#include #include +#include #include @@ -31,6 +36,273 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests { void TearDown() override {} }; + // Tests for the rules that decide when a cached CBS token is still good, and + // when the connection must replace it. These rules are pure functions, so they + // run on every platform and need no service. + class TestTokenRefresh : public testing::Test { + protected: + static Azure::Core::Credentials::AccessToken TokenExpiringIn(std::chrono::seconds lifetime) + { + Azure::Core::Credentials::AccessToken token; + token.Token = "TestToken"; + token.ExpiresOn = std::chrono::system_clock::now() + lifetime; + return token; + } + }; + + TEST_F(TestTokenRefresh, CachedTokenIsUsableWhileItHasLifeLeft) + { + auto const now = std::chrono::system_clock::now(); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::hours(1)), now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::minutes(2)), now)); + } + + TEST_F(TestTokenRefresh, CachedTokenIsNotUsableNearOrAfterExpiry) + { + auto const now = std::chrono::system_clock::now(); + // Inside the minimum lifetime that a caller may use. + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(10)), now)); + // Already expired. + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable( + TokenExpiringIn(std::chrono::seconds(-30)), now)); + } + + TEST_F(TestTokenRefresh, RefreshIsDueOneBufferBeforeExpiry) + { + auto const now = std::chrono::system_clock::now(); + // A normal token has a long life, so no refresh is due yet. + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(90)), now)); + // Inside the buffer, so the refresh thread must replace the token. + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue( + TokenExpiringIn(std::chrono::minutes(6)), now)); + } + + // A credential can put any value in ExpiresOn, and the cast from + // Azure::DateTime to a system clock time point throws outside the range of + // that clock. A default constructed ExpiresOn is year 1. These rules run on + // the refresh thread, where an exception would end the process, so they must + // not throw for any value. + TEST_F(TestTokenRefresh, ExtremeExpiryValuesDoNotThrow) + { + auto const now = std::chrono::system_clock::now(); + + // A default constructed token reports year 1. + Azure::Core::Credentials::AccessToken defaultToken; + EXPECT_NO_THROW({ + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(defaultToken, now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(defaultToken, now)); + }); + + // A token that reports a year past the range of the system clock. + Azure::Core::Credentials::AccessToken farFutureToken; + farFutureToken.Token = "TestToken"; + farFutureToken.ExpiresOn = Azure::DateTime(9999, 12, 31); + EXPECT_NO_THROW({ + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(farFutureToken, now)); + EXPECT_FALSE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(farFutureToken, now)); + }); + } + + // A token with a lifetime shorter than the buffer is due as soon as it + // arrives. The connection must still hand it to a caller, because refusing it + // would make every call authenticate again. + TEST_F(TestTokenRefresh, ShortLivedTokenIsDueButStillUsable) + { + auto const now = std::chrono::system_clock::now(); + auto const token = TokenExpiringIn(std::chrono::seconds(80)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsTokenRefreshDue(token, now)); + EXPECT_TRUE(Azure::Core::Amqp::_detail::IsCachedTokenUsable(token, now)); + } + + // The state that the connection shares with the refresh thread. + TEST_F(TestTokenRefresh, TheSharedStateStartsEmptyAndRuns) + { + Azure::Core::Amqp::_detail::TokenRefreshState state; + EXPECT_FALSE(state.Stop); + EXPECT_TRUE(state.TokenStore.empty()); + EXPECT_TRUE(state.TokenSessions.empty()); + } + + // The refresh thread co-owns the state block, so the block stays alive after + // the connection that made it is gone. That is what lets the thread come back + // from a release that destroyed the connection, take the mutex, and read the + // stop flag. + TEST_F(TestTokenRefresh, TheSharedStateOutlivesTheOwnerThatMadeIt) + { + std::weak_ptr observer; + std::shared_ptr threadCopy; + { + auto connectionCopy = std::make_shared(); + observer = connectionCopy; + threadCopy = connectionCopy; + } + ASSERT_FALSE(observer.expired()); + + // The owner is gone, so this stands for the connection destructor setting + // the stop flag before it detaches the thread. + { + std::unique_lock lock(threadCopy->Mutex); + threadCopy->Stop = true; + } + EXPECT_TRUE(threadCopy->Stop); + + threadCopy.reset(); + EXPECT_TRUE(observer.expired()); + } + + // Tests for ReleaseOutsideLock, the hold that the refresh thread puts the + // promoted session in. The session can be the last reference to the + // connection, and the connection destructor takes the token mutex, so the + // hold must always drop the pointer with that mutex free. + class TestReleaseOutsideLock : public testing::Test { + protected: + // Records the state of the lock at the moment it is destroyed. + class LockObserver final { + public: + LockObserver(std::unique_lock& lock, bool& destroyed, bool& lockWasHeld) + : m_lock{lock}, m_destroyed{destroyed}, m_lockWasHeld{lockWasHeld} + { + } + + ~LockObserver() + { + m_destroyed = true; + m_lockWasHeld = m_lock.owns_lock(); + } + + private: + std::unique_lock& m_lock; + bool& m_destroyed; + bool& m_lockWasHeld; + }; + + std::mutex m_mutex; + bool m_destroyed{false}; + bool m_lockWasHeld{true}; + + std::shared_ptr MakeObserver(std::unique_lock& lock) + { + return std::make_shared(lock, m_destroyed, m_lockWasHeld); + } + }; + + TEST_F(TestReleaseOutsideLock, ReleaseDropsThePointerWithTheLockFree) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + EXPECT_TRUE(lock.owns_lock()); + EXPECT_FALSE(m_destroyed); + hold.Release(); + EXPECT_TRUE(m_destroyed); + } + EXPECT_FALSE(m_lockWasHeld); + // The lock is back in the state the caller left it in. + EXPECT_TRUE(lock.owns_lock()); + } + + // The early return in the refresh path leaves the scope with the lock held + // and the session still in the hold. The destructor must give up the lock for + // that release too, or it locks a mutex this thread already holds. + TEST_F(TestReleaseOutsideLock, TheDestructorAlsoReleasesWithTheLockFree) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + EXPECT_FALSE(m_destroyed); + } + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_TRUE(lock.owns_lock()); + } + + // An exception on the refresh path must not leave the release under the lock + // either. + TEST_F(TestReleaseOutsideLock, AnExceptionStillReleasesWithTheLockFree) + { + std::unique_lock lock(m_mutex); + bool caught{false}; + try + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + throw std::runtime_error("the refresh failed"); + } + catch (std::runtime_error const&) + { + caught = true; + } + EXPECT_TRUE(caught); + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_TRUE(lock.owns_lock()); + } + + // The refresh path releases on the normal path and then leaves the scope, so + // the pointer is released once and the second call does nothing. + TEST_F(TestReleaseOutsideLock, ASecondReleaseDoesNothing) + { + std::unique_lock lock(m_mutex); + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + hold.Release(); + EXPECT_EQ(nullptr, hold.Get()); + + m_destroyed = false; + hold.Release(); + EXPECT_FALSE(m_destroyed); + EXPECT_TRUE(lock.owns_lock()); + } + + // The refresh path gives up the token mutex for the network work. A release + // that happens then must leave the lock free, not take it. + TEST_F(TestReleaseOutsideLock, AFreeLockStaysFree) + { + std::unique_lock lock(m_mutex); + lock.unlock(); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{MakeObserver(lock), lock}; + hold.Release(); + } + EXPECT_TRUE(m_destroyed); + EXPECT_FALSE(m_lockWasHeld); + EXPECT_FALSE(lock.owns_lock()); + } + + // The hold keeps the pointer usable for the whole refresh, and it does not + // destroy an object that another owner still holds. + TEST_F(TestReleaseOutsideLock, TheHoldKeepsThePointerAndSharesIt) + { + std::unique_lock lock(m_mutex); + auto observer = MakeObserver(lock); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{observer, lock}; + EXPECT_EQ(observer.get(), hold.Get().get()); + hold.Release(); + // This test still owns the object, so the release did not destroy it. + EXPECT_FALSE(m_destroyed); + } + EXPECT_FALSE(m_destroyed); + observer.reset(); + EXPECT_TRUE(m_destroyed); + } + + // An empty hold is what the refresh path never builds, but the class must not + // touch the lock for one. + TEST_F(TestReleaseOutsideLock, AnEmptyHoldLeavesTheLockAlone) + { + std::unique_lock lock(m_mutex); + { + Azure::Core::Amqp::_detail::ReleaseOutsideLock hold{nullptr, lock}; + hold.Release(); + EXPECT_TRUE(lock.owns_lock()); + } + EXPECT_TRUE(lock.owns_lock()); + EXPECT_FALSE(m_destroyed); + } + #if !defined(AZ_PLATFORM_MAC) TEST_F(TestConnections, SimpleConnection) { diff --git a/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp b/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp index 9067b68172..d76fb8a855 100644 --- a/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp +++ b/sdk/core/azure-core-amqp/test/ut/message_source_target.cpp @@ -5,6 +5,8 @@ #include #include +#include + #include class TestSourceTarget : public testing::Test { @@ -271,6 +273,31 @@ TEST_F(TestSourceTarget, TargetThroughValue) EXPECT_EQ(target.GetAddress(), target2.GetAddress()); } +TEST_F(TestSourceTarget, TargetAssignment) +{ + { + MessageTarget target{"address2"}; + { + MessageTarget other{"address1"}; + target = other; + } + // Copy assignment copies the implementation, so the assigned object stays valid after the + // original goes out of scope. + EXPECT_EQ(AmqpValue{"address1"}, target.GetAddress()); + } + { + MessageTarget target{"address1"}; + MessageTarget moved{std::move(target)}; + EXPECT_EQ(AmqpValue{"address1"}, moved.GetAddress()); + } + { + MessageTarget target{"address2"}; + MessageTarget other{"address1"}; + target = std::move(other); + EXPECT_EQ(AmqpValue{"address1"}, target.GetAddress()); + } +} + MessageSource ReturnsSource() { return MessageSource(); } MessageSource ReturnsSource(const char* str) { return MessageSource(str); } MessageSource ReturnsSource(const std::string& str) { return MessageSource(str); } @@ -537,3 +564,57 @@ TEST_F(TestSourceTarget, SourceProperties) EXPECT_EQ(source.GetAddress(), source2.GetAddress()); } } + +TEST_F(TestSourceTarget, SourceAssignment) +{ + { + MessageSource source{"address2"}; + { + MessageSource other{"address1"}; + source = other; + } + // Copy assignment copies the implementation, so the assigned object stays valid after the + // original goes out of scope. + EXPECT_EQ(AmqpValue{"address1"}, source.GetAddress()); + } + { + MessageSource source{"address1"}; + MessageSource moved{std::move(source)}; + EXPECT_EQ(AmqpValue{"address1"}, moved.GetAddress()); + } + { + MessageSource source{"address2"}; + MessageSource other{"address1"}; + source = std::move(other); + EXPECT_EQ(AmqpValue{"address1"}, source.GetAddress()); + } +} + +// uAMQP accepts any value as the default outcome. The Rust stack accepts only the four delivery +// outcomes that the AMQP spec names, so the round trip and the two error cases below are specific +// to that stack. +#if ENABLE_RUST_AMQP +TEST_F(TestSourceTarget, SourceDefaultOutcome) +{ + for (auto const& outcome : + {"amqp:accepted:list", "amqp:rejected:list", "amqp:released:list", "amqp:modified:list"}) + { + MessageSourceOptions options; + options.DefaultOutcome = AmqpSymbol{outcome}; + MessageSource source(options); + EXPECT_EQ(source.GetDefaultOutcome(), AmqpSymbol{outcome}); + } + { + // A default outcome that is a string, and not a symbol, is rejected. + MessageSourceOptions options; + options.DefaultOutcome = AmqpValue{"amqp:accepted:list"}; + EXPECT_ANY_THROW(MessageSource source(options)); + } + { + // A symbol that names no delivery outcome is rejected. + MessageSourceOptions options; + options.DefaultOutcome = AmqpSymbol{"amqp:unknown:list"}; + EXPECT_ANY_THROW(MessageSource source(options)); + } +} +#endif diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp index 24a73d24f1..dac56b52b3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp +++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/producer_client_test.cpp @@ -5,12 +5,17 @@ #include "eventhubs_test_base.hpp" +#include #include +#include #include #include #include +#include +#include #include +#include #include @@ -356,6 +361,297 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test { } } +#if ENABLE_UAMQP + namespace { + // Counts the tokens that the test asked for, and can report a shorter life + // for each token than the real life. + // + // A real Microsoft Entra token lives for 60 to 90 minutes, so a test that + // waits for a true expiry would run for hours. A shorter reported life makes + // the connection refresh the token in seconds. The token itself is a real + // token and stays valid, so the service still accepts it. + class CountingTokenCredential final : public Azure::Core::Credentials::TokenCredential { + public: + CountingTokenCredential( + std::shared_ptr innerCredential, + Azure::Nullable reportedLifetime = {}) + // Report the name of the credential that this one wraps. The + // connection asks for that name to decide whether it puts a shared + // access signature token or a JWT, so a new name here would break a + // connection string credential. + : Azure::Core::Credentials::TokenCredential(innerCredential->GetCredentialName()), + m_innerCredential(std::move(innerCredential)), m_reportedLifetime(reportedLifetime) + { + } + + int GetTokenCount() const { return m_tokenCount.load(); } + + private: + std::shared_ptr m_innerCredential; + Azure::Nullable m_reportedLifetime; + mutable std::atomic m_tokenCount{0}; + + Azure::Core::Credentials::AccessToken GetToken( + Azure::Core::Credentials::TokenRequestContext const& requestContext, + Azure::Core::Context const& context) const override + { + auto token = m_innerCredential->GetToken(requestContext, context); + if (m_reportedLifetime.HasValue()) + { + token.ExpiresOn = std::chrono::system_clock::now() + m_reportedLifetime.Value(); + } + m_tokenCount++; + return token; + } + }; + + // The connection replaces a token one buffer, seven minutes, before the + // token expires. A token that reports seven minutes and thirty seconds of + // life is due in about thirty seconds. That gap is long enough to see the + // refresh and then to check that the refreshed token is the one in use. + constexpr std::chrono::seconds SlowRefreshLifetime{450}; + + // A token that reports less life than the buffer is due as soon as it + // arrives, so the refresh thread replaces it every twenty seconds, which is + // the smallest interval between two refresh passes. This drives the tests + // that need several refreshes in a short run. + constexpr std::chrono::seconds FastRefreshLifetime{80}; + + Azure::Messaging::EventHubs::Models::EventData MakeTestEvent(std::uint8_t tag) + { + Azure::Messaging::EventHubs::Models::EventData event; + event.Body = {'T', tag}; + return event; + } + + // Return the credential that the refresh tests wrap and count. + // + // When the environment gives a connection string, build a shared access + // signature credential from it. That path needs no Microsoft Entra sign in, + // so a developer can run these tests from a workstation. Otherwise use the + // credential of the test harness, which is what the pipeline uses. + std::shared_ptr MakeInnerCredential( + std::shared_ptr harnessCredential) + { + auto const connectionString + = Azure::Core::_internal::Environment::GetVariable("EVENTHUBS_CONNECTION_STRING"); + if (!connectionString.empty()) + { + return std::make_shared< + Azure::Core::Amqp::_internal::ServiceBusSasConnectionStringCredential>( + connectionString); + } + return harnessCredential; + } + + // Wait until the credential issued at least `count` tokens, or until the + // timeout. Returns the count that was reached. + int WaitForTokenCount( + std::shared_ptr const& credential, + int count, + std::chrono::seconds timeout) + { + auto const deadline = std::chrono::steady_clock::now() + timeout; + while (credential->GetTokenCount() < count && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + return credential->GetTokenCount(); + } + } // namespace + + // L1. The connection must replace the CBS token before that token expires, + // and the replacement must reach the service. + TEST_P(ProducerClientTest, TokenRefreshBeforeExpiry_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), SlowRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('1'))); + auto const countAfterFirstSend = credential->GetTokenCount(); + ASSERT_GE(countAfterFirstSend, 1); + + // The test makes no client call here, so only the refresh thread can raise + // the count. + auto const countAfterRefresh + = WaitForTokenCount(credential, countAfterFirstSend + 1, std::chrono::seconds(120)); + ASSERT_GT(countAfterRefresh, countAfterFirstSend) + << "The connection did not replace the CBS token before the token expired."; + + // The count rises when the refresh asks for the token, which is before the + // refresh puts that token on the wire. Wait for the put to finish. A put + // that fails drops the cached token, and the send below then has to ask for + // a new one, which fails this test. A put that takes longer than this wait + // to fail stays invisible to the test. The next refresh is about thirty + // seconds out, so this wait cannot let another refresh raise the count. + std::this_thread::sleep_for(std::chrono::seconds(5)); + ASSERT_EQ(credential->GetTokenCount(), countAfterRefresh) + << "A second refresh ran during the wait, so this test cannot judge the first one."; + + // Make sure the replacement reached the service. A refresh that failed makes + // the connection drop the cached token, and then this send has to ask the + // credential for another token. So a send that does not raise the count + // proves that the refreshed token is in the cache and that the service + // accepted it. + ASSERT_NO_THROW(client.Send(MakeTestEvent('2'))); + EXPECT_EQ(credential->GetTokenCount(), countAfterRefresh) + << "The send asked for a new token, so the refresh did not reach the service."; + client.Close(); + } + + // L2. The client must keep sending past two token lifetimes, while the + // refresh thread replaces the token repeatedly. This is the continuity gate + // from issue #7254, on a compressed clock. L1 is the test that proves a + // refresh reaches the service; this one proves that a long run keeps working. + TEST_P(ProducerClientTest, TokenRefreshContinuityPastTwoLifetimes_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + auto const runFor = FastRefreshLifetime * 2 + std::chrono::seconds(10); + auto const start = std::chrono::steady_clock::now(); + int sendCount = 0; + while (std::chrono::steady_clock::now() - start < runFor) + { + ASSERT_NO_THROW(client.Send(MakeTestEvent('3'))) + << "A send failed after " << sendCount << " sends."; + sendCount++; + std::this_thread::sleep_for(std::chrono::seconds(20)); + } + + EXPECT_GT(sendCount, 0); + EXPECT_GE(credential->GetTokenCount(), 3) + << "The connection did not refresh the token while the client was running."; + client.Close(); + } + + // L3. Shutting the client down must stop the refresh thread, and must not + // hang. + TEST_P(ProducerClientTest, TokenRefreshStopsOnClose_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + { + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('4'))); + EXPECT_GE(WaitForTokenCount(credential, 2, std::chrono::seconds(90)), 2); + + auto const closeStart = std::chrono::steady_clock::now(); + client.Close(); + EXPECT_LT(std::chrono::steady_clock::now() - closeStart, std::chrono::seconds(70)) + << "Close did not return promptly while the refresh thread was running."; + } + + // The client is gone, so the refresh thread is stopped. The count must not + // move after this point. + auto const countAfterShutdown = credential->GetTokenCount(); + std::this_thread::sleep_for(std::chrono::seconds(30)); + EXPECT_EQ(credential->GetTokenCount(), countAfterShutdown) + << "The refresh thread kept running after the client was destroyed."; + } + + // L4. A token with a normal life must be fetched once. This is the control + // case. It fails if the new expiry test makes the cache authenticate again on + // every call. + TEST_P(ProducerClientTest, NormalTokenIsNotRefetched_LIVEONLY_) + { + // No reported lifetime, so the real expiry of the token applies. + auto credential + = std::make_shared(MakeInnerCredential(GetTestCredential())); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + ASSERT_NO_THROW(client.Send(MakeTestEvent('5'))); + auto const countAfterFirstSend = credential->GetTokenCount(); + + std::this_thread::sleep_for(std::chrono::seconds(10)); + ASSERT_NO_THROW(client.Send(MakeTestEvent('6'))); + + EXPECT_EQ(credential->GetTokenCount(), countAfterFirstSend) + << "The connection asked for a new token for an audience that already had a good one."; + client.Close(); + } + + // L5. Calls from many threads must keep working while the refresh thread + // replaces the token. The refresh thread and the calling threads take the same + // lock, so this is the test for a deadlock. + TEST_P(ProducerClientTest, TokenRefreshUnderConcurrentCalls_LIVEONLY_) + { + auto credential = std::make_shared( + MakeInnerCredential(GetTestCredential()), FastRefreshLifetime); + + Azure::Messaging::EventHubs::ProducerClientOptions options; + options.Name = "sender-link"; + options.ApplicationID = testing::UnitTest::GetInstance()->current_test_info()->name(); + + Azure::Messaging::EventHubs::ProducerClient client{ + GetEnv("EVENTHUBS_HOST"), GetEventHubName(), credential, options}; + + std::string const eventHubName{GetEventHubName()}; + std::atomic failed{false}; + std::vector threads; + for (int i = 0; i < 8; i++) + { + threads.emplace_back([&client, &eventHubName, &failed]() { + auto const start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < std::chrono::seconds(60)) + { + try + { + auto result = client.GetEventHubProperties(); + if (result.Name != eventHubName) + { + failed = true; + } + } + catch (std::exception const&) + { + failed = true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + }); + } + for (auto& t : threads) + { + if (t.joinable()) + { + t.join(); + } + } + + EXPECT_FALSE(failed.load()) << "A call failed while the refresh thread replaced the token."; + EXPECT_GE(credential->GetTokenCount(), 2) << "The refresh thread did not run during the test."; + client.Close(); + } +#endif // ENABLE_UAMQP + namespace { static std::string GetSuffix(const testing::TestParamInfo& info) {