Add checkpoint store MSI auth support (Azure#19082)

Author: pakrym

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Config/EventHubClientFactory.cs

@@ -166,8 +166,12 @@ internal IEventHubConsumerClient GetEventHubConsumerClient(string eventHubName,
 
         internal BlobContainerClient GetCheckpointStoreClient()
         {
-            // Fall back to default if not explicitly registered
-            return new BlobContainerClient(_configuration.GetWebJobsConnectionString(ConnectionStringNames.Storage), _options.CheckpointContainer);
+            var section = _configuration.GetWebJobsConnectionStringSection(ConnectionStringNames.Storage);
+            var options = _componentFactory.CreateClientOptions(typeof(BlobClientOptions), null, section);
+            var credential = _componentFactory.CreateTokenCredential(section);
+            var client = (BlobServiceClient)_componentFactory.CreateClient(typeof(BlobServiceClient), section, credential, options);
+
+            return client.GetBlobContainerClient(_options.CheckpointContainer);
         }
 
         internal static string NormalizeConnectionString(string originalConnectionString, string eventHubName)

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubEndToEndTests.cs

@@ -160,7 +160,8 @@ await AssertCanSendReceiveMessage(host =>
                 host.ConfigureAppConfiguration(configurationBuilder =>
                     configurationBuilder.AddInMemoryCollection(new Dictionary<string, string>()
                     {
-                        {"TestConnection", EventHubsTestEnvironment.Instance.EventHubsConnectionString}
+                        {"TestConnection", EventHubsTestEnvironment.Instance.EventHubsConnectionString},
+                        {"AzureWebJobsStorage", StorageTestEnvironment.Instance.StorageConnectionString}
                     })));
         }
 
@@ -175,9 +176,18 @@ await AssertCanSendReceiveMessage(host =>
                         {"TestConnection:clientId", EventHubsTestEnvironment.Instance.ClientId},
                         {"TestConnection:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
                         {"TestConnection:tenantId", EventHubsTestEnvironment.Instance.TenantId},
+                        {"AzureWebJobsStorage:serviceUri", GetServiceUri()},
+                        {"AzureWebJobsStorage:clientId", EventHubsTestEnvironment.Instance.ClientId},
+                        {"AzureWebJobsStorage:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
+                        {"AzureWebJobsStorage:tenantId", EventHubsTestEnvironment.Instance.TenantId},
                     })));
         }
 
+        private static string GetServiceUri()
+        {
+            return "https://" + StorageTestEnvironment.Instance.StorageAccountName + ".blob." + StorageTestEnvironment.Instance.StorageEndpointSuffix;
+        }
+
         public async Task AssertCanSendReceiveMessage(Action<IHostBuilder> hostConfiguration)
         {
             var (jobHost, host) = BuildHost<EventHubTestSingleDispatchJobWithConnection>(hostConfiguration);

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubsClientFactoryTests.cs

@@ -9,6 +9,7 @@
 using Azure.Messaging.EventHubs.Consumer;
 using Azure.Messaging.EventHubs.Primitives;
 using Azure.Messaging.EventHubs.Producer;
+using Azure.Storage.Blobs;
 using Microsoft.Azure.WebJobs.EventHubs.Processor;
 using Microsoft.Extensions.Azure;
 using Microsoft.Extensions.Configuration;
@@ -105,7 +106,14 @@ public void UsesDefaultConnectionToStorageAccount()
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("AzureWebJobsStorage", "UseDevelopmentStorage=true"));
 
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration));
+            var factoryMock = new Mock<AzureComponentFactory>();
+            factoryMock.Setup(m => m.CreateClient(
+                        typeof(BlobServiceClient),
+                        It.Is<ConfigurationSection>(c => c.Path == "AzureWebJobsStorage"),
+                        null, null))
+                .Returns(new BlobServiceClient(configuration["AzureWebJobsStorage"]));
+
+            var factory = new EventHubClientFactory(configuration, factoryMock.Object, Options.Create(options), new DefaultNameResolver(configuration));
 
             var client = factory.GetCheckpointStoreClient();
             Assert.AreEqual("azure-webjobs-eventhub", client.Name);

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/WebJobsEventHubTestBase.cs

@@ -70,7 +70,6 @@ protected void ConfigureTestEventHub(IHostBuilder builder)
                     builder.AddInMemoryCollection(new Dictionary<string, string>()
                     {
                         {"webjobstesthub", _eventHubScope.EventHubName},
-                        {"AzureWebJobsStorage", StorageTestEnvironment.Instance.StorageConnectionString}
                     });
                 })
                 .ConfigureDefaultTestHost<T>(b =>

sdk/eventhub/test-resources.json

@@ -66,6 +66,7 @@
   "variables": {
     "contributorRoleId": "b24988ac-6180-42a0-ab88-20f7382dd24c",
     "eventHubsDataOwnerRoleId": "f526a384-b230-433a-b45c-95f59c4a2dec",
+    "storageDataOwnerRoleId": "b7e6dc6d-f1e8-4753-8033-0f276bb0955b",
     "eventHubsNamespace": "[concat('eh-', parameters('baseName'))]",
     "storageAccount": "[concat('blb', parameters('baseName'))]",
     "defaultSASKeyName": "RootManageSharedAccessKey",
@@ -158,6 +159,20 @@
         "principalId": "[parameters('testApplicationOid')]",
         "scope": "[resourceGroup().id]"
       }
+    },
+    {
+      "type": "Microsoft.Authorization/roleAssignments",
+      "apiVersion": "2019-04-01-preview",
+      "name": "[guid(resourceGroup().id, parameters('testApplicationOid'), variables('storageDataOwnerRoleId'))]",
+      "dependsOn": [
+        "[resourceId('Microsoft.EventHub/Namespaces', variables('eventHubsNamespace'))]",
+        "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccount'))]"
+      ],
+      "properties": {
+        "roleDefinitionId": "[resourceId('Microsoft.Authorization/roleDefinitions', variables('storageDataOwnerRoleId'))]",
+        "principalId": "[parameters('testApplicationOid')]",
+        "scope": "[resourceGroup().id]"
+      }
     }
   ],
   "outputs": {