diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CertificateConfigurationTests.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CertificateConfigurationTests.cs index 0d33a8ec..6973eb7c 100644 --- a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CertificateConfigurationTests.cs +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CertificateConfigurationTests.cs @@ -51,7 +51,7 @@ public void CreateClient_WithDisableSslValidation_LogsWarningAndSetsCallback() DisableSslValidation = true }; - var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object); + using var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object); // Verify warning was logged when DisableSslValidation is true loggerMock.Verify( @@ -79,7 +79,7 @@ public void CreateClient_WithoutDisableSslValidation_DoesNotLogWarning() DisableSslValidation = false }; - var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object); + using var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object); // Verify warning was NOT logged when DisableSslValidation is false loggerMock.Verify( @@ -116,7 +116,8 @@ public void CreateClient_WithEnableNetHttpLogging_LogsWarningAndPreservesClientO AllowBulkExecution = true }; - using var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object); + using var client = CosmosExtensionServices.CreateClient(settings, "TestDisplay", loggerMock.Object, + null, (_, _) => new OfflineTokenCredential()); loggerMock.Verify( x => x.Log( diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosDataSourceExtensionTests.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosDataSourceExtensionTests.cs index 60fcebc3..314f4f60 100644 --- a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosDataSourceExtensionTests.cs +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosDataSourceExtensionTests.cs @@ -3,7 +3,14 @@ using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Azure.Cosmos; using System.Net; +using System.CommandLine; +using System.CommandLine.Invocation; +using System.ComponentModel.Composition.Hosting; +using Cosmos.DataTransfer.Core; using Cosmos.DataTransfer.Interfaces; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; +using Moq; using Newtonsoft.Json; using Newtonsoft.Json.Linq; @@ -198,6 +205,7 @@ private static async Task AssertPartitionKeyRoundTrip(string[] paths, (string js { ["ConnectionString"] = connectionString, ["ConnectionMode"] = "Gateway", + ["UseDefaultCredentials"] = "true", ["AllowBulkExecution"] = "false", ["Database"] = database.Id, ["Container"] = containerName @@ -219,6 +227,7 @@ private static async Task AssertPartitionKeyRoundTrip(string[] paths, (string js var container = database.GetContainer(containerName); foreach (var mode in new[] { DataWriteMode.Insert, DataWriteMode.Upsert, DataWriteMode.InsertStream, DataWriteMode.UpsertStream }) { + Console.WriteLine($"{containerName}: testing {mode} with pooled HTTP transport."); string GetId(int index) => separateNullId && cases[index].key.Equals(PartitionKey.Null) ? $"{mode}-null" : mode.ToString(); @@ -272,6 +281,7 @@ public async Task Test_Json_RoundTrip_WithLongValues() { var (_, database, connectionString) = GetSetup(); var config = TestHelpers.CreateConfig(new Dictionary() { { "UseRbacAuth", "false" }, + { "UseDefaultCredentials", "true" }, { "ConnectionString", connectionString }, { "ConnectionMode", "Gateway" }, { "AllowBulkExecution", "false" }, @@ -305,6 +315,61 @@ await sinkExtension.WriteAsync(items.ToAsyncEnumerable(), config, await jsonSink.FormatDataAsync(fetched.ToAsyncEnumerable(), writer, config, NullLogger.Instance); var res = Encoding.UTF8.GetString(writer.ToArray()); Assert.AreEqual("[{\"partitionKey\":\"pk\",\"id\":\"1\",\"long\":638676324052177500}]", res); + + foreach (var preAuthenticate in new[] { false, true }) + { + await RunSequentialOperationsAsync(connectionString, database.Id, + config["Container"]!, preAuthenticate); + } + } + + private static async Task RunSequentialOperationsAsync( + string connectionString, string database, string sourceContainer, bool preAuthenticate) + { + var source = new CosmosDataSourceExtension(); + var sink = new CosmosDataSinkExtension(); + using var catalog = new CompositionContainer(); + var loader = new Mock(); + loader.Setup(l => l.BuildExtensionCatalog(It.IsAny())).Returns(catalog); + loader.Setup(l => l.LoadExtensions(catalog)).Returns([source]); + loader.Setup(l => l.LoadExtensions(catalog)).Returns([sink]); + var values = new Dictionary(); + foreach (var side in new[] { "SourceSettings", "SinkSettings" }) + { + values[$"{side}:ConnectionString"] = connectionString; + values[$"{side}:Database"] = database; + values[$"{side}:ConnectionMode"] = "Gateway"; + values[$"{side}:LimitToEndpoint"] = "true"; + values[$"{side}:UseDefaultCredentials"] = "true"; + values[$"{side}:PreAuthenticate"] = preAuthenticate.ToString(); + values[$"{side}:AllowBulkExecution"] = "false"; + } + values["SourceSettings:Container"] = sourceContainer; + values["SinkSettings:PartitionKeyPath"] = "/partitionKey"; + // Reuse two destinations across pre-auth variants to stay within the current emulator's partition limit. + values["SinkSettings:WriteMode"] = "Upsert"; + for (var i = 0; i < 2; i++) + { + values[$"Operations:{i}:SinkSettings:Container"] = $"operation-{i}"; + } + var configuration = new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + using var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole()); + var command = new RunCommand.CommandHandler(loader.Object, configuration, loggerFactory) + { + Source = source.DisplayName, + Sink = sink.DisplayName, + }; + var context = new InvocationContext(new RootCommand().Parse(Array.Empty())); + Assert.AreEqual(0, await command.InvokeAsync(context), "Sequential Cosmos Operations must succeed."); + for (var i = 0; i < 2; i++) + { + var outputConfig = new ConfigurationBuilder() + .AddConfiguration(configuration.GetSection("SinkSettings")) + .AddConfiguration(configuration.GetSection($"Operations:{i}:SinkSettings")).Build(); + var output = await source.ReadAsync(outputConfig, NullLogger.Instance).ToArrayAsync(); + Assert.AreEqual(1, output.Length); + Assert.AreEqual(638676324052177500L, output[0].GetValue("long")); + } } [TestMethod] diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosHttpClientLifecycleTests.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosHttpClientLifecycleTests.cs new file mode 100644 index 00000000..697214a3 --- /dev/null +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosHttpClientLifecycleTests.cs @@ -0,0 +1,396 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Logging; +using Microsoft.Azure.Cosmos; +using Microsoft.Azure.Cosmos.Encryption; +using Azure.Security.KeyVault.Keys.Cryptography; +using Moq; +using System.Net; +using System.Net.Security; +using System.Net.Sockets; +using System.Security.Authentication; +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; +using System.Text; + +namespace Cosmos.DataTransfer.CosmosExtension.UnitTests; + +[TestClass] +public class CosmosHttpClientLifecycleTests +{ + [DataTestMethod] + [DataRow(false, false, false)] + [DataRow(false, false, true)] + [DataRow(false, true, false)] + [DataRow(false, true, true)] + [DataRow(true, false, false)] + [DataRow(true, false, true)] + [DataRow(true, true, false)] + [DataRow(true, true, true)] + public async Task CreateClient_AfterTransportUsed_CreatesLaterSourceAndSink( + bool useRbacAuth, bool preAuthenticate, bool differentAccount) + { + var source = CreateSettings(useRbacAuth, preAuthenticate); + var sink = CreateSettings(useRbacAuth, preAuthenticate, + differentAccount ? 2 : 1); + using var first = CreateOfflineClient(source, "source"); + using var transport = first.ClientOptions.HttpClientFactory(); + await StartTransportAsync(transport); + + // Construct through DMT, not by manually assigning HttpClient.Timeout. + using var second = CreateOfflineClient(sink, "sink", "source"); + first.Dispose(); + using var secondTransport = second.ClientOptions.HttpClientFactory(); + await StartTransportAsync(secondTransport); + using var third = CreateOfflineClient(source, "next-operation"); + using var thirdTransport = third.ClientOptions.HttpClientFactory(); + await StartTransportAsync(thirdTransport); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void CreateClient_WithoutDefaultCredentials_LeavesTransportToSdk(bool useRbacAuth) + { + var settings = CreateSettings(useRbacAuth, false); + settings.UseDefaultCredentials = false; + using var first = CreateOfflineClient(settings, "source"); + using var second = CreateOfflineClient(settings, "sink"); + Assert.IsNull(first.ClientOptions.HttpClientFactory); + Assert.IsNull(second.ClientOptions.HttpClientFactory); + } + + [TestMethod] + public async Task Factory_ReturnsIndependentWrappers_AndSnapshotsSettings() + { + var settings = CreateSettings(false, false); + var options = CosmosExtensionServices.CreateClientOptions(settings, "source-agent", NullLogger.Instance); + settings.PreAuthenticate = true; + settings.WebProxy = "http://127.0.0.1:1/"; + settings.DisableSslValidation = true; + using var first = options.HttpClientFactory(); + using var second = options.HttpClientFactory(); + Assert.AreNotSame(first, second); + first.DefaultRequestHeaders.Add("x-test-source", "source"); + first.Timeout = TimeSpan.FromSeconds(17); + Assert.IsFalse(second.DefaultRequestHeaders.Contains("x-test-source")); + Assert.AreNotEqual(first.Timeout, second.Timeout); + await StartTransportAsync(first); + first.Dispose(); + await StartTransportAsync(second); + using var later = options.HttpClientFactory(); + await StartTransportAsync(later); + Assert.AreEqual("source-agent", options.ApplicationName); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task SdkDisposal_DisposesWrapper_NotSharedHandler(bool encryption) + { + var settings = CreateSettings(true, false); + var options = CosmosExtensionServices.CreateClientOptions(settings, "test", NullLogger.Instance); + var factory = options.HttpClientFactory; + HttpClient? ownedTransport = null; + options.HttpClientFactory = () => ownedTransport = factory(); + var credential = new OfflineTokenCredential(); + using var client = new CosmosClient(settings.AccountEndpoint, credential, options); + Assert.IsNotNull(ownedTransport); + await StartTransportAsync(ownedTransport); + + using var owner = encryption + ? client.WithEncryption(new KeyResolver(credential), KeyEncryptionKeyResolverName.AzureKeyVault) + : client; + owner.Dispose(); + await Assert.ThrowsExceptionAsync(() => + ownedTransport.GetAsync("http://127.0.0.1:1/", new CancellationToken(true))); + + using var survivingTransport = factory(); + await StartTransportAsync(survivingTransport); + using var next = CreateOfflineClient(settings, "next"); + } + + [TestMethod] + public async Task CreateClient_ConcurrentCreationAndUse_DoesNotShareWrapperState() + { + await Task.WhenAll(Enumerable.Range(0, 24).Select(i => Task.Run(async () => + { + var settings = CreateSettings(i % 2 == 0, i % 3 == 0, 1 + i % 2); + using var client = CreateOfflineClient(settings, $"sink-{i}"); + using var transport = client.ClientOptions.HttpClientFactory(); + await StartTransportAsync(transport); + using var next = CreateOfflineClient(settings, $"next-{i}"); + }))); + } + + [TestMethod] + public void Handler_PoolsOnlyCompatibleSettings_AndAppliesCredentialsAndTls() + { + var settings = CreateSettings(false, false); + var handler = CosmosExtensionServices.GetHttpHandler(settings); + Assert.AreSame(handler, CosmosExtensionServices.GetHttpHandler( + CreateSettings(true, false, 2))); + Assert.AreSame(CredentialCache.DefaultNetworkCredentials, handler.Credentials); + Assert.IsNull(handler.DefaultProxyCredentials); + Assert.IsNull(handler.Proxy); + Assert.IsFalse(handler.PreAuthenticate); + Assert.IsFalse(handler.UseCookies); + Assert.IsNull(handler.SslOptions.RemoteCertificateValidationCallback); + Assert.AreEqual(TimeSpan.FromMinutes(5), handler.PooledConnectionLifetime); + + settings.PreAuthenticate = true; + var preAuth = CosmosExtensionServices.GetHttpHandler(settings); + Assert.AreNotSame(handler, preAuth); + Assert.IsTrue(preAuth.PreAuthenticate); + settings.DisableSslValidation = true; + var insecure = CosmosExtensionServices.GetHttpHandler(settings); + Assert.AreNotSame(preAuth, insecure); + Assert.IsNotNull(insecure.SslOptions.RemoteCertificateValidationCallback); + settings.WebProxy = "http://127.0.0.1:12345/"; + var proxy = CosmosExtensionServices.GetHttpHandler(settings); + Assert.AreNotSame(insecure, proxy); + Assert.IsInstanceOfType(proxy.Proxy); + Assert.AreEqual(new Uri(settings.WebProxy), ((WebProxy)proxy.Proxy!).Address); + Assert.IsNull(proxy.Proxy.Credentials); + settings.UseDefaultProxyCredentials = true; + var authenticatedProxy = CosmosExtensionServices.GetHttpHandler(settings); + Assert.AreNotSame(proxy, authenticatedProxy); + Assert.AreSame(CredentialCache.DefaultNetworkCredentials, authenticatedProxy.Proxy!.Credentials); + Assert.IsNull(proxy.Proxy.Credentials); + settings.WebProxy = "http://127.0.0.1:12346/"; + Assert.AreNotSame(authenticatedProxy, CosmosExtensionServices.GetHttpHandler(settings)); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void ClientOptions_PreservesProxyTlsAndCosmosSettings(bool customFactory) + { + var settings = CreateSettings(false, true); + settings.UseDefaultCredentials = customFactory; + settings.UseDefaultProxyCredentials = true; + settings.WebProxy = "http://127.0.0.1:12345/"; + settings.DisableSslValidation = true; + settings.ConnectionMode = ConnectionMode.Direct; + settings.LimitToEndpoint = true; + settings.AllowBulkExecution = true; + settings.EnableContentResponseOnWrite = false; + var logger = new Mock(); + var options = CosmosExtensionServices.CreateClientOptions(settings, "test-agent", logger.Object); + Assert.AreEqual(ConnectionMode.Direct, options.ConnectionMode); + Assert.IsTrue(options.LimitToEndpoint); + Assert.IsTrue(options.AllowBulkExecution); + Assert.IsFalse(options.EnableContentResponseOnWrite); + Assert.AreEqual("test-agent", options.ApplicationName); + Assert.IsInstanceOfType(options.Serializer); + Assert.IsNotNull(options.ServerCertificateCustomValidationCallback); + if (customFactory) + { + Assert.IsNotNull(options.HttpClientFactory); + Assert.IsNull(options.WebProxy); // The SDK rejects WebProxy together with HttpClientFactory. + } + else + { + Assert.IsNull(options.HttpClientFactory); + Assert.IsInstanceOfType(options.WebProxy); + Assert.AreSame(CredentialCache.DefaultNetworkCredentials, options.WebProxy.Credentials); + } + logger.Verify(x => x.Log(LogLevel.Warning, It.IsAny(), + It.Is((v, _) => v.ToString()!.Contains("SSL certificate validation is DISABLED")), + It.IsAny(), It.IsAny>()), Times.Once); + } + + [TestMethod] + public async Task Factory_ConcurrentRequestsAndDisposal_KeepOtherClientsUsable() + { + using var server = new LoopbackServer(); + var responses = Enumerable.Range(0, 12).Select(_ => server.ServeAsync(1)).ToArray(); + var clients = Task.WhenAll(Enumerable.Range(0, 12).Select(i => Task.Run(async () => + { + var settings = CreateSettings(i % 2 == 0, false); + settings.WebProxy = server.Address.ToString(); + var options = CosmosExtensionServices.CreateClientOptions(settings, $"sink-{i}", NullLogger.Instance); + using var transport = options.HttpClientFactory(); + using var response = await transport.GetAsync("http://cosmos.invalid/dbs", server.Token); + Assert.AreEqual(HttpStatusCode.OK, response.StatusCode); + }))); + await Task.WhenAll(responses.Cast().Append(clients)); + } + + [TestMethod] + public async Task Factory_UsesConfiguredProxy_AndReusesConnectionAfterWrapperDisposal() + { + using var server = new LoopbackServer(); + var requests = server.ServeAsync(requestCount: 2); + var settings = CreateSettings(false, true); + settings.WebProxy = server.Address.ToString(); + settings.UseDefaultProxyCredentials = true; + var options = CosmosExtensionServices.CreateClientOptions(settings, "test", NullLogger.Instance); + // The closure must not read subsequent mutations to the settings object. + settings.WebProxy = "http://127.0.0.1:1/"; + for (var i = 0; i < 2; i++) + { + using var transport = options.HttpClientFactory(); + using var response = await transport.GetAsync("http://cosmos.invalid/dbs", server.Token); + Assert.AreEqual(HttpStatusCode.OK, response.StatusCode); + } + var received = await requests; + Assert.AreEqual(2, received.Count); + Assert.IsTrue(received.All(r => r.StartsWith("GET http://cosmos.invalid/dbs HTTP/1.1"))); + Assert.IsTrue(received.All(r => !r.Contains("Authorization:", StringComparison.OrdinalIgnoreCase))); + Assert.IsFalse(received[1].Contains("Cookie:", StringComparison.OrdinalIgnoreCase)); + // ServeAsync accepts only one socket: both requests must reuse that connection. + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Factory_SelfSignedTls_RequiresExplicitOptOut(bool disableSslValidation) + { + using var rsa = RSA.Create(2048); + var request = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1); + using var certificate = request.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1)); + // Import the key for Windows Schannel as well as OpenSSL. + using var serverCertificate = new X509Certificate2(certificate.Export(X509ContentType.Pfx)); + using var server = new LoopbackServer(); + var served = server.ServeAsync(1, serverCertificate, expectTlsRejection: !disableSslValidation); + var settings = CreateSettings(false, false); + settings.DisableSslValidation = disableSslValidation; + var options = CosmosExtensionServices.CreateClientOptions(settings, "test", NullLogger.Instance); + using var transport = options.HttpClientFactory(); + var uri = new UriBuilder(server.Address) { Scheme = "https" }.Uri; + try + { + if (disableSslValidation) + { + using var response = await transport.GetAsync(uri, server.Token); + Assert.AreEqual(HttpStatusCode.OK, response.StatusCode); + } + else + { + var error = await Assert.ThrowsExceptionAsync(() => transport.GetAsync(uri, server.Token)); + Assert.IsInstanceOfType(error.InnerException); + } + } + finally + { + await served; + } + } + + private sealed class LoopbackServer : IDisposable + { + private readonly TcpListener listener = new(IPAddress.Loopback, 0); + private readonly CancellationTokenSource timeout = new(TimeSpan.FromSeconds(15)); + public Uri Address { get; } + public CancellationToken Token => timeout.Token; + + public LoopbackServer() + { + listener.Start(); + Address = new Uri($"http://127.0.0.1:{((IPEndPoint)listener.LocalEndpoint).Port}/"); + } + + public async Task> ServeAsync(int requestCount, X509Certificate2? certificate = null, + bool expectTlsRejection = false) + { + using var socket = await listener.AcceptTcpClientAsync(Token); + using var network = socket.GetStream(); + using var tls = certificate is null ? null : new SslStream(network, leaveInnerStreamOpen: true); + if (tls is not null) + { + try + { + await tls.AuthenticateAsServerAsync(new SslServerAuthenticationOptions + { + ServerCertificate = certificate, + }, Token); + } + catch (AuthenticationException) when (expectTlsRejection) + { + return []; + } + catch (IOException) when (expectTlsRejection) + { + return []; + } + if (expectTlsRejection) + { + return []; + } + } + Stream stream = tls is null ? network : tls; + using var reader = new StreamReader(stream, Encoding.ASCII, leaveOpen: true); + var requests = new List(); + for (var i = 0; i < requestCount; i++) + { + var headers = new StringBuilder(); + while (true) + { + var line = await reader.ReadLineAsync(Token); + Assert.IsNotNull(line, "Client closed the pooled connection."); + if (line.Length == 0) + { + break; + } + headers.AppendLine(line); + } + requests.Add(headers.ToString()); + var closeHeader = i == requestCount - 1 ? "Connection: close\r\n" : ""; + await stream.WriteAsync(Encoding.ASCII.GetBytes( + $"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n{closeHeader}Set-Cookie: test=not-shared\r\n\r\n"), Token); + await stream.FlushAsync(Token); + } + return requests; + } + + public void Dispose() + { + timeout.Cancel(); + listener.Stop(); + timeout.Dispose(); + } + } + + [TestMethod] + public async Task CreateClient_RbacConstructor_UsesOnlyOfflineCredentialForEagerTokenAcquisition() + { + var credential = new OfflineTokenCredential(); + var settings = CreateSettings(true, false); + using var client = CosmosExtensionServices.CreateClient(settings, "test", NullLogger.Instance, null, + (_, _) => credential); + var scopes = await credential.RequestedScopes.Task.WaitAsync(TimeSpan.FromSeconds(5)); + CollectionAssert.AreEqual(new[] { "https://127.0.0.1/.default" }, scopes); + } + + private static CosmosClient CreateOfflineClient(CosmosSettingsBase settings, string displayName, + string? sourceDisplayName = null) + { + return CosmosExtensionServices.CreateClient(settings, displayName, NullLogger.Instance, sourceDisplayName, + (_, _) => new OfflineTokenCredential()); + } + + private static T CreateSettings(bool useRbacAuth, bool preAuthenticate, int port = 1) + where T : CosmosSettingsBase, new() + { + var endpoint = $"https://127.0.0.1:{port}/"; + return new T + { + UseDefaultCredentials = true, + PreAuthenticate = preAuthenticate, + UseRbacAuth = useRbacAuth, + AccountEndpoint = useRbacAuth ? endpoint : null, + ConnectionString = useRbacAuth ? null + : $"AccountEndpoint={endpoint};AccountKey={Convert.ToBase64String(new byte[64])};", + Database = "db", + Container = "container", + }; + } + + private static async Task StartTransportAsync(HttpClient transport) + { + // Cancellation marks HttpClient as started without sending credentials or doing network I/O. + using var request = new HttpRequestMessage(HttpMethod.Get, "http://127.0.0.1:1/"); + await Assert.ThrowsExceptionAsync(async () => + await transport.SendAsync(request, new CancellationToken(canceled: true))); + } +} diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/OfflineTokenCredential.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/OfflineTokenCredential.cs new file mode 100644 index 00000000..6fb3c036 --- /dev/null +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/OfflineTokenCredential.cs @@ -0,0 +1,26 @@ +using Azure.Core; +using Azure.Identity; + +namespace Cosmos.DataTransfer.CosmosExtension.UnitTests; + +internal sealed class OfflineTokenCredential : TokenCredential +{ + public TaskCompletionSource RequestedScopes { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public override AccessToken GetToken(TokenRequestContext requestContext, CancellationToken cancellationToken) + { + throw GetTokenFailure(requestContext); + } + + public override ValueTask GetTokenAsync(TokenRequestContext requestContext, CancellationToken cancellationToken) + { + return ValueTask.FromException(GetTokenFailure(requestContext)); + } + + private Exception GetTokenFailure(TokenRequestContext requestContext) + { + RequestedScopes.TrySetResult(requestContext.Scopes); + // Cosmos eagerly acquires a token during construction. Fail locally before any identity or account I/O. + return new CredentialUnavailableException("Token acquisition is disabled in offline transport tests."); + } +} diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs index 69072f22..377f173e 100644 --- a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs @@ -143,7 +143,7 @@ public async Task WriteAsync(IAsyncEnumerable dataItems, IConfigurati var settings = config.Get(); settings.Validate(); - var client = CosmosExtensionServices.CreateClient(settings!, DisplayName, logger, dataSource.DisplayName); + using var client = CosmosExtensionServices.CreateClient(settings!, DisplayName, logger, dataSource.DisplayName); Container container; if (settings!.UseRbacAuth) diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs index 9c2dbcb1..9c3c23ac 100644 --- a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSourceExtension.cs @@ -19,7 +19,7 @@ public async IAsyncEnumerable ReadAsync(IConfiguration config, ILogge var settings = config.Get(); settings.Validate(); - var client = CosmosExtensionServices.CreateClient(settings!, DisplayName, logger); + using var client = CosmosExtensionServices.CreateClient(settings!, DisplayName, logger); Container container; diff --git a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosExtensionServices.cs b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosExtensionServices.cs index bc97a0de..894b38c6 100644 --- a/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosExtensionServices.cs +++ b/Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosExtensionServices.cs @@ -13,6 +13,7 @@ using System.Security.Cryptography.X509Certificates; using System.Text.RegularExpressions; using System.Diagnostics.Tracing; +using System.Collections.Concurrent; namespace Cosmos.DataTransfer.CosmosExtension { @@ -25,27 +26,41 @@ internal enum TokenCredentialSelection ClientCertificateCredential, } - // Static HttpClient instances with different configurations for reuse across connections - // This avoids connection exhaustion and properly handles credentials - private static readonly Lazy _httpClientWithDefaultCredentials = new Lazy(() => - { - var handler = new HttpClientHandler - { - Credentials = CredentialCache.DefaultNetworkCredentials, - PreAuthenticate = false - }; - return new HttpClient(handler); - }); + // The SDK configures and disposes each HttpClient. Only handlers/pools are process-owned. + private static readonly ConcurrentDictionary> _httpHandlers = new(); - private static readonly Lazy _httpClientWithDefaultCredentialsAndPreAuth = new Lazy(() => + private sealed record HttpTransportSettings(string? WebProxy, bool UseDefaultProxyCredentials, + bool PreAuthenticate, bool DisableSslValidation); + + internal static SocketsHttpHandler GetHttpHandler(CosmosSettingsBase settings) { - var handler = new HttpClientHandler + var transportSettings = new HttpTransportSettings( + string.IsNullOrEmpty(settings.WebProxy) ? null : settings.WebProxy, + settings.UseDefaultProxyCredentials, settings.PreAuthenticate, settings.DisableSslValidation); + return _httpHandlers.GetOrAdd(transportSettings, key => new Lazy(() => { - Credentials = CredentialCache.DefaultNetworkCredentials, - PreAuthenticate = true - }; - return new HttpClient(handler); - }); + var handler = new SocketsHttpHandler + { + Credentials = CredentialCache.DefaultNetworkCredentials, + PreAuthenticate = key.PreAuthenticate, + // Refresh DNS for long-running migrations without giving up connection pooling. + PooledConnectionLifetime = TimeSpan.FromMinutes(5), + UseCookies = false, + }; + if (key.WebProxy is not null) + { + handler.Proxy = new WebProxy(key.WebProxy) + { + UseDefaultCredentials = key.UseDefaultProxyCredentials, + }; + } + if (key.DisableSslValidation) + { + handler.SslOptions.RemoteCertificateValidationCallback = (_, _, _, _) => true; + } + return handler; + })).Value; + } private static readonly object _httpEventListenerLock = new(); private static EventListener? _httpEventListener; @@ -168,7 +183,12 @@ internal static CosmosClientOptions CreateClientOptions(CosmosSettingsBase setti LimitToEndpoint = settings.LimitToEndpoint, }; - if (!string.IsNullOrEmpty(settings.WebProxy)) + if (settings.UseDefaultCredentials) + { + var handler = GetHttpHandler(settings); + clientOptions.HttpClientFactory = () => new HttpClient(handler, disposeHandler: false); + } + else if (!string.IsNullOrEmpty(settings.WebProxy)) { var webProxy = new WebProxy(settings.WebProxy); if (settings.UseDefaultProxyCredentials) @@ -178,15 +198,6 @@ internal static CosmosClientOptions CreateClientOptions(CosmosSettingsBase setti clientOptions.WebProxy = webProxy; } - // Configure the HttpClient with default credentials if requested - // This enables authenticated proxy support for the underlying HTTP connections - if (settings.UseDefaultCredentials) - { - clientOptions.HttpClientFactory = settings.PreAuthenticate - ? () => _httpClientWithDefaultCredentialsAndPreAuth.Value - : () => _httpClientWithDefaultCredentials.Value; - } - // Disable SSL certificate validation for development scenarios if (settings.DisableSslValidation) { @@ -198,6 +209,12 @@ internal static CosmosClientOptions CreateClientOptions(CosmosSettingsBase setti } public static CosmosClient CreateClient(CosmosSettingsBase settings, string displayName, ILogger logger, string? sourceDisplayName = null) + { + return CreateClient(settings, displayName, logger, sourceDisplayName, CreateRbacTokenCredential); + } + + internal static CosmosClient CreateClient(CosmosSettingsBase settings, string displayName, ILogger logger, + string? sourceDisplayName, Func credentialFactory) { string userAgentString = CreateUserAgentString(displayName, sourceDisplayName); var clientOptions = CreateClientOptions(settings, userAgentString, logger); @@ -211,7 +228,7 @@ public static CosmosClient CreateClient(CosmosSettingsBase settings, string disp CosmosClient? cosmosClient; if (settings.UseRbacAuth) { - var tokenCredential = CreateRbacTokenCredential(settings, logger); + var tokenCredential = credentialFactory(settings, logger); cosmosClient = settings.InitClientEncryption ? new CosmosClient(settings.AccountEndpoint, tokenCredential, clientOptions) diff --git a/Extensions/Cosmos/README.md b/Extensions/Cosmos/README.md index 4078e38a..515c99e2 100644 --- a/Extensions/Cosmos/README.md +++ b/Extensions/Cosmos/README.md @@ -63,6 +63,20 @@ uses the documented `/ready` probe on port 8080; the current emulator waits for its certificate endpoint. The tests use Gateway mode and explicitly disable bulk execution, which the vNext emulator does not support with the .NET SDK. See the [Linux vNext emulator documentation](https://learn.microsoft.com/azure/cosmos-db/emulator-linux). +The round-trip test also runs sequential Cosmos-to-Cosmos `Operations` with +`UseDefaultCredentials` and both `PreAuthenticate` values. This exercises source +enumeration, sink completion/disposal, and later client creation after real HTTP +requests. These tests do not validate bulk execution, live Entra authentication, +or enterprise proxy credential negotiation. + +`CosmosHttpClientLifecycleTests` runs without an emulator or Azure credentials. +It covers key/RBAC client construction after transport use, disposal (including +the encryption wrapper), concurrent client creation, proxy routing, connection +reuse, and self-signed TLS rejection/explicit opt-out using local loopback servers. +RBAC constructor tests inject an offline credential that fails token requests +locally: the SDK eagerly requests tokens during construction, even before a test +sends a request. These tests never use developer sign-ins or real identity providers. + These tests cover JSON round trips, the missing-field/explicit-null request mismatch, and typed/stream insert, upsert, and delete with single, nested, and hierarchical partition keys. They do not cover bulk execution or authentication enforcement. @@ -187,10 +201,33 @@ Source supports the following optional parameters: - `Query` - Allows further filtering using a Cosmos SQL statement. - `WebProxy` (`null` by default) - Enables connections through a proxy. - `UseDefaultProxyCredentials` (`false` by default) - When `true`, includes default credentials in the WebProxy request. Use this when connecting through an authenticated proxy that returns `407 Proxy Authentication Required`. See [WebProxy.Credentials Property](https://learn.microsoft.com/dotnet/api/system.net.webproxy.credentials). -- `UseDefaultCredentials` (`false` by default) - When `true`, configures the underlying HttpClient with default network credentials. Use this when the connection to CosmosDB requires authentication through a proxy. -- `PreAuthenticate` (`false` by default) - When `true`, enables pre-authentication on the HttpClient, which sends credentials with the initial request rather than waiting for a 401/407 challenge. This can save extra round-trips but should only be used when the endpoint is trusted. +- `UseDefaultCredentials` (`false` by default) - When `true`, configures HTTP server authentication with the current user's default network credentials. This is separate from Cosmos account authentication and from proxy credentials; it is **not required for RBAC**. For an explicit authenticated `WebProxy`, use `UseDefaultProxyCredentials`. +- `PreAuthenticate` (`false` by default) - With `UseDefaultCredentials`, enables HTTP pre-authentication for matching URIs after authentication has succeeded. It does not eliminate the first authentication challenge and does not acquire an Entra token. - `EnableNetHttpLogging` (`false` by default) - When `true`, enables `.NET` `System.Net.Http` diagnostic events to help investigate connection issues. These events are emitted at Debug level and can include endpoint/request details. +### Connection settings and client lifetime + +These connection settings apply independently to sources, sinks, and each entry +in `Operations`, with either key or RBAC authentication. Leave +`UseDefaultCredentials` unset unless your HTTP network environment requires it. +On older versions, removing this setting when it is unnecessary also avoids +the "This instance has already started one or more requests" error. + +When default network credentials are enabled, each Cosmos SDK client receives +a fresh `HttpClient` so its timeout, headers, and disposal cannot affect another +client. Process-owned handlers retain connection pools across operations and +are separated by proxy URL, proxy-credential setting, pre-authentication, and TLS +validation setting. Connections are recycled after five minutes to refresh DNS; +cookies are not shared between clients. Account keys and Entra tokens remain +per Cosmos client and are not stored in the handler configuration. + +Explicit proxies and opt-in `DisableSslValidation` are applied to the custom +HTTP handler; certificate validation stays enabled by default. The TLS opt-out +also applies to SDK Direct-mode connections and retains its development-only +warning. Without `UseDefaultCredentials`, the SDK continues to manage its own +HTTP transport. Source clients live until enumeration completes or is disposed; +sink clients live until writes finish, including failure/cancellation cleanup. + ### Connectivity troubleshooting (.NET HTTP diagnostics) When connection failures are not verbose enough, enable `EnableNetHttpLogging` and run the tool with `Debug` logging. @@ -449,8 +486,8 @@ this setting. See [troubleshooting 429 responses](https://learn.microsoft.com/az - **`WebProxy`**: Optional. Specifies the proxy server URL to use for connections (e.g., `http://yourproxy.server.com/`). - **`UseDefaultProxyCredentials`**: Optional, defaults to `false`. When `true`, includes default credentials in the WebProxy request. Use this when connecting through an authenticated proxy that returns `407 Proxy Authentication Required`. See [WebProxy.Credentials Property](https://learn.microsoft.com/dotnet/api/system.net.webproxy.credentials). -- **`UseDefaultCredentials`**: Optional, defaults to `false`. When `true`, configures the underlying HttpClient with default network credentials. Use this when the connection to CosmosDB requires authentication through a proxy. -- **`PreAuthenticate`**: Optional, defaults to `false`. When `true`, enables pre-authentication on the HttpClient, which sends credentials with the initial request rather than waiting for a 401/407 challenge. This can save extra round-trips but should only be used when the endpoint is trusted. +- **`UseDefaultCredentials`**: Optional, defaults to `false`. Configures HTTP server authentication with the current user's default network credentials, independently of Cosmos account authentication. It is **not required for RBAC**. For an explicit authenticated `WebProxy`, use `UseDefaultProxyCredentials`. +- **`PreAuthenticate`**: Optional, defaults to `false`. With `UseDefaultCredentials`, enables HTTP pre-authentication for matching URIs after authentication has succeeded. It does not eliminate the first authentication challenge and does not acquire an Entra token. - **`EnableNetHttpLogging`**: Optional, defaults to `false`. When `true`, enables `.NET` `System.Net.Http` diagnostic events for troubleshooting. Events are logged at Debug level and may include endpoint/request details. - **`LimitToEndpoint`**: Optional, defaults to `false`. When the value of this property is false, the Cosmos DB SDK will automatically discover