[EventHubs] Retry a CBS authentication failure once on a fresh connection

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -11,6 +11,8 @@
 
 ### Bugs Fixed
 
+- The uAMQP message sender now encodes delivery annotations, message annotations, and the footer as described sections, so a uAMQP receiver can decode a message that carries them. Before, the sender wrote the bare maps, and the receiving link failed with "Error decoding message" and went to the error state. [[#7376]](https://github.com/Azure/azure-sdk-for-cpp/issues/7376)
+- On the uAMQP transport, `MessageSender::Open` and `MessageReceiver::Open` now throw `_detail::CbsPutTokenFailedException` when the service rejects the CBS put-token. The type derives from `std::runtime_error` and carries the original `AuthenticationException`, which `RethrowOriginal()` throws again. The Event Hubs clients use the type to tell a rejected put-token from a credential failure. `ManagementClient::Open` and `ManagementClient::ExecuteOperation` still throw `AuthenticationException`. [[#7376]](https://github.com/Azure/azure-sdk-for-cpp/issues/7376)
 - uAMQP pollable registration and removal no longer block each other while a poll is in flight. The
   polling registry now waits on completion notifications, and sender, receiver, and link setup and
   teardown do not hold connection locks across registry operations. The polling thread now sleeps

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp

@@ -7,8 +7,10 @@
 
 #include <azure/core/context.hpp>
 
+#include <exception>
 #include <stdexcept>
 #include <string>
+#include <utility>
 
 namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   class ClaimsBasedSecurityImpl;
@@ -53,6 +55,23 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     CbsOpenResult Result;
   };
 
+#if ENABLE_UAMQP
+  /** @brief Identifies a failed uAMQP CBS put-token operation. */
+  class CbsPutTokenFailedException final : public std::runtime_error {
+  public:
+    CbsPutTokenFailedException(std::exception_ptr original, std::string const& what)
+        : std::runtime_error(what), m_original{std::move(original)}
+    {
+    }
+
+    std::exception_ptr GetOriginal() const { return m_original; }
+    [[noreturn]] void RethrowOriginal() const { std::rethrow_exception(m_original); }
+
+  private:
+    std::exception_ptr m_original;
+  };
+#endif
+
   enum class CbsTokenType
   {
     Invalid,

sdk/core/azure-core-amqp/src/amqp/connection.cpp

@@ -195,9 +195,14 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
             = claimsBasedSecurity->PutToken(tokenType, audienceUrl, token, expiresOn, context);
         if (std::get<0>(result) != CbsOperationResult::Ok)
         {
-          throw Azure::Core::Credentials::AuthenticationException(
+          auto failure = Azure::Core::Credentials::AuthenticationException(
               "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result))
               + " reason: " + std::get<2>(result));
+#if ENABLE_UAMQP
+          throw CbsPutTokenFailedException(std::make_exception_ptr(failure), failure.what());
+#else
+          throw failure;
+#endif
         }
         Log::Stream(Logger::Level::Verbose) << "Close CBS object";
         claimsBasedSecurity->Close(context);

sdk/core/azure-core-amqp/src/impl/uamqp/amqp/management.cpp

@@ -3,6 +3,7 @@
 
 #include "azure/core/amqp/internal/management.hpp"
 
+#include "azure/core/amqp/internal/claims_based_security.hpp"
 #include "azure/core/amqp/internal/models/messaging_values.hpp"
 #include "azure/core/amqp/models/amqp_message.hpp"
 #include "private/connection_impl.hpp"
@@ -90,6 +91,22 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     }
   }
 
+  // The put-token marker is for the Event Hubs retry loops. Management callers keep
+  // the AuthenticationException contract.
+  Credentials::AccessToken ManagementClientImpl::AuthenticateManagementAudience(
+      Context const& context)
+  {
+    try
+    {
+      return m_session->GetConnection()->AuthenticateAudience(
+          m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context);
+    }
+    catch (CbsPutTokenFailedException const& failure)
+    {
+      failure.RethrowOriginal();
+    }
+  }
+
   _internal::ManagementOpenStatus ManagementClientImpl::Open(Context const& context)
   {
     std::unique_lock<std::mutex> lock(m_openCloseLock);
@@ -107,8 +124,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
        */
       if (m_options.ManagementNodeName == "$management")
       {
-        m_accessToken = m_session->GetConnection()->AuthenticateAudience(
-            m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context);
+        m_accessToken = AuthenticateManagementAudience(context);
       }
       {
         _internal::MessageSenderOptions messageSenderOptions;
@@ -243,8 +259,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       // than one thread and that member has no lock.
       if (!m_accessToken.Token.empty())
       {
-        auto accessToken{m_session->GetConnection()->AuthenticateAudience(
-            m_session, m_managementEntityPath + "/" + m_options.ManagementNodeName, context)};
+        auto accessToken{AuthenticateManagementAudience(context)};
         messageToSend.ApplicationProperties["security_token"]
             = Models::AmqpValue{accessToken.Token};
       }

sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/management_impl.hpp

@@ -109,6 +109,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     bool m_sendCompleted{false};
 
     void CloseSenderAndReceiverAfterFailedOpen() noexcept;
+    Azure::Core::Credentials::AccessToken AuthenticateManagementAudience(Context const& context);
     void SetState(ManagementState newState);
     // Reflect the error state to the OnError callback and return a delivery rejected status.
     Models::AmqpValue IndicateError(

sdk/core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/src/message_sender.c

@@ -241,8 +241,11 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
         size_t body_data_count = 0;
         size_t body_sequence_count = 0;
         AMQP_VALUE msg_annotations = NULL;
+        AMQP_VALUE msg_annotations_value = NULL;
         AMQP_VALUE footer = NULL;
+        AMQP_VALUE footer_value = NULL;
         AMQP_VALUE delivery_annotations = NULL;
+        AMQP_VALUE delivery_annotations_value = NULL;
         bool is_error = false;
 
         // message header
@@ -275,14 +278,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
             (message_get_message_annotations(message, &msg_annotations) == 0) &&
             (msg_annotations != NULL))
         {
-            if (amqpvalue_get_encoded_size(msg_annotations, &encoded_size) != 0)
+            msg_annotations_value = amqpvalue_create_message_annotations(msg_annotations);
+            if (msg_annotations_value == NULL)
             {
-                LogError("Cannot obtain message annotations encoded size");
+                LogError("Cannot create message annotations AMQP value");
                 is_error = true;
             }
             else
             {
-                total_encoded_size += encoded_size;
+                if (amqpvalue_get_encoded_size(msg_annotations_value, &encoded_size) != 0)
+                {
+                    LogError("Cannot obtain message annotations encoded size");
+                    is_error = true;
+                }
+                else
+                {
+                    total_encoded_size += encoded_size;
+                }
             }
         }
 
@@ -341,14 +353,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
             (message_get_footer(message, &footer) == 0) &&
             (footer != NULL))
         {
-            if (amqpvalue_get_encoded_size(footer, &encoded_size) != 0)
+            footer_value = amqpvalue_create_footer(footer);
+            if (footer_value == NULL)
             {
-                LogError("Cannot obtain footer encoded size");
+                LogError("Cannot create footer AMQP value");
                 is_error = true;
             }
             else
             {
-                total_encoded_size += encoded_size;
+                if (amqpvalue_get_encoded_size(footer_value, &encoded_size) != 0)
+                {
+                    LogError("Cannot obtain footer encoded size");
+                    is_error = true;
+                }
+                else
+                {
+                    total_encoded_size += encoded_size;
+                }
             }
         }
 
@@ -357,14 +378,23 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
             (message_get_delivery_annotations(message, &delivery_annotations) == 0) &&
             (delivery_annotations != NULL))
         {
-            if (amqpvalue_get_encoded_size(delivery_annotations, &encoded_size) != 0)
+            delivery_annotations_value = amqpvalue_create_delivery_annotations(delivery_annotations);
+            if (delivery_annotations_value == NULL)
             {
-                LogError("Cannot obtain delivery annotations encoded size");
+                LogError("Cannot create delivery annotations AMQP value");
                 is_error = true;
             }
             else
             {
-                total_encoded_size += encoded_size;
+                if (amqpvalue_get_encoded_size(delivery_annotations_value, &encoded_size) != 0)
+                {
+                    LogError("Cannot obtain delivery annotations encoded size");
+                    is_error = true;
+                }
+                else
+                {
+                    total_encoded_size += encoded_size;
+                }
             }
         }
 
@@ -554,13 +584,13 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
 
                 if ((result == SEND_ONE_MESSAGE_OK) && (msg_annotations != NULL))
                 {
-                    if (amqpvalue_encode(msg_annotations, encode_bytes, &payload) != 0)
+                    if (amqpvalue_encode(msg_annotations_value, encode_bytes, &payload) != 0)
                     {
                         LogError("Cannot encode message annotations value");
                         result = SEND_ONE_MESSAGE_ERROR;
                     }
 
-                    log_message_chunk(message_sender, "Message Annotations:", msg_annotations);
+                    log_message_chunk(message_sender, "Message Annotations:", msg_annotations_value);
                 }
 
                 if ((result == SEND_ONE_MESSAGE_OK) && (properties != NULL))
@@ -587,24 +617,24 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
 
                 if ((result == SEND_ONE_MESSAGE_OK) && (footer != NULL))
                 {
-                    if (amqpvalue_encode(footer, encode_bytes, &payload) != 0)
+                    if (amqpvalue_encode(footer_value, encode_bytes, &payload) != 0)
                     {
                         LogError("Cannot encode footer value");
                         result = SEND_ONE_MESSAGE_ERROR;
                     }
 
-                    log_message_chunk(message_sender, "Footer:", footer);
+                    log_message_chunk(message_sender, "Footer:", footer_value);
                 }
 
                 if ((result == SEND_ONE_MESSAGE_OK) && (delivery_annotations != NULL))
                 {
-                    if (amqpvalue_encode(delivery_annotations, encode_bytes, &payload) != 0)
+                    if (amqpvalue_encode(delivery_annotations_value, encode_bytes, &payload) != 0)
                     {
                         LogError("Cannot encode delivery annotations value");
                         result = SEND_ONE_MESSAGE_ERROR;
                     }
 
-                    log_message_chunk(message_sender, "Delivery annotations:", delivery_annotations);
+                    log_message_chunk(message_sender, "Delivery annotations:", delivery_annotations_value);
                 }
 
                 if (result == SEND_ONE_MESSAGE_OK)
@@ -764,6 +794,11 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
             annotations_destroy(msg_annotations);
         }
 
+        if (msg_annotations_value != NULL)
+        {
+            amqpvalue_destroy(msg_annotations_value);
+        }
+
         if (application_properties != NULL)
         {
             amqpvalue_destroy(application_properties);
@@ -789,10 +824,20 @@ static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message
             annotations_destroy(footer);
         }
 
+        if (footer_value != NULL)
+        {
+            amqpvalue_destroy(footer_value);
+        }
+
         if (delivery_annotations != NULL)
         {
             annotations_destroy(delivery_annotations);
         }
+
+        if (delivery_annotations_value != NULL)
+        {
+            amqpvalue_destroy(delivery_annotations_value);
+        }
     }
 
     return result;

sdk/core/azure-core-amqp/test/ut/message_sender_receiver.cpp

@@ -1565,5 +1565,92 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
     CloseAmqpConnection(connection);
   }
 
+#if !defined(USE_NATIVE_BROKER)
+  // A message that carries these three sections must decode on the receiving link.
+  TEST_F(TestMessageSendReceive, ReceiverDecodesAnnotationsAndFooter)
+  {
+    std::string brokerEndpoint = GetBrokerEndpoint() + "/annotations";
+
+    class AnnotatingEndpoint : public MessageTests::MockServiceEndpoint {
+    public:
+      AnnotatingEndpoint(
+          std::string const& name,
+          MessageTests::MockServiceEndpointOptions const& options)
+          : MockServiceEndpoint(name, options)
+      {
+      }
+      virtual ~AnnotatingEndpoint() = default;
+
+      void SendOnce(Azure::Core::Amqp::Models::AmqpMessage message)
+      {
+        m_message = std::move(message);
+        m_shouldSend = true;
+      }
+
+    private:
+      mutable bool m_shouldSend{false};
+      Azure::Core::Amqp::Models::AmqpMessage m_message;
+
+      void Poll() const override
+      {
+        if (m_shouldSend && HasMessageSender())
+        {
+          m_shouldSend = false;
+          EXPECT_EQ(MessageSendStatus::Ok, std::get<0>(GetMessageSender().Send(m_message)));
+        }
+      }
+
+      void MessageReceived(
+          std::string const&,
+          std::shared_ptr<Azure::Core::Amqp::Models::AmqpMessage> const&) override
+      {
+      }
+    };
+    auto serviceEndpoint = std::make_shared<AnnotatingEndpoint>(
+        brokerEndpoint, MessageTests::MockServiceEndpointOptions{});
+    m_mockServer.AddServiceEndpoint(serviceEndpoint);
+
+    auto connection{CreateAmqpConnection({})};
+    auto session{CreateAmqpSession(connection)};
+    StartServerListening();
+
+    MessageReceiverOptions receiverOptions;
+    receiverOptions.Name = "annotations-receiver";
+    receiverOptions.MessageTarget = "egress";
+    receiverOptions.SettleMode = Azure::Core::Amqp::_internal::ReceiverSettleMode::First;
+    receiverOptions.MaxLinkCredit = 10;
+    MessageReceiver receiver(session.CreateMessageReceiver(brokerEndpoint, receiverOptions));
+    receiver.Open();
+
+    Azure::Core::Amqp::Models::AmqpMessage sent;
+    sent.DeliveryAnnotations[Models::AmqpSymbol{"x-opt-delivery"}] = Models::AmqpValue{"delivery"};
+    sent.MessageAnnotations[Models::AmqpSymbol{"x-opt-offset"}] = Models::AmqpValue{"10"};
+    sent.MessageAnnotations[Models::AmqpSymbol{"x-opt-partition-key"}]
+        = Models::AmqpValue{"partition"};
+    sent.Footer[Models::AmqpSymbol{"x-opt-footer"}] = Models::AmqpValue{"footer"};
+    sent.SetBody(Models::AmqpValue{"annotated body"});
+    serviceEndpoint->SendOnce(sent);
+
+    Azure::Core::Context receiveContext{Azure::DateTime::clock::now() + std::chrono::seconds(10)};
+    auto received = receiver.WaitForIncomingMessage(receiveContext);
+    if (received.first)
+    {
+      EXPECT_TRUE(sent.DeliveryAnnotations == received.first->DeliveryAnnotations);
+      EXPECT_TRUE(sent.MessageAnnotations == received.first->MessageAnnotations);
+      EXPECT_TRUE(sent.Footer == received.first->Footer);
+      EXPECT_EQ("annotated body", static_cast<std::string>(received.first->GetBodyAsAmqpValue()));
+    }
+    else
+    {
+      ADD_FAILURE() << "The receiver returned no message: " << received.second;
+    }
+
+    receiver.Close();
+    StopServerListening();
+    EndAmqpSession(session);
+    CloseAmqpConnection(connection);
+  }
+#endif
+
 #endif // !defined(AZ_PLATFORM_MAC)
 }}}} // namespace Azure::Core::Amqp::Tests