|
| 1 | +// Copyright (c) Microsoft. All rights reserved. |
| 2 | + |
| 3 | +#if NET |
| 4 | +using System; |
| 5 | +using System.Collections.Generic; |
| 6 | +using System.IO; |
| 7 | +using System.Linq; |
| 8 | +using System.Net; |
| 9 | +#endif |
| 10 | +using System.Net.Http; |
| 11 | +#if NET |
| 12 | +using System.Net.Security; |
| 13 | +using System.Net.Sockets; |
| 14 | +using System.Security.Cryptography.X509Certificates; |
| 15 | +using System.Threading; |
| 16 | +using System.Threading.Tasks; |
| 17 | +#else |
| 18 | +using Microsoft.SemanticKernel.Http; |
| 19 | +#endif |
| 20 | + |
| 21 | +#pragma warning disable CA2000 // Ownership is transferred to the shared handler or NetworkStream. |
| 22 | + |
| 23 | +namespace Microsoft.SemanticKernel.Plugins.OpenApi; |
| 24 | + |
| 25 | +/// <summary> |
| 26 | +/// Creates the built-in OpenAPI HTTP client and binds validated DNS addresses to its connections. |
| 27 | +/// </summary> |
| 28 | +internal static class OpenApiHttpClientProvider |
| 29 | +{ |
| 30 | +#if NET |
| 31 | + internal static readonly HttpRequestOptionsKey<IPAddress[]> ValidatedAddressesKey = new("OpenApiValidatedAddresses"); |
| 32 | + |
| 33 | + // RFC 8305's 250 ms delay starts another route (e.g., IPv4) without timing out the first (e.g., IPv6). |
| 34 | + private static readonly TimeSpan s_connectionAttemptDelay = TimeSpan.FromMilliseconds(250); |
| 35 | +#endif |
| 36 | + |
| 37 | + /// <summary> |
| 38 | + /// Creates a non-redirecting client backed by a shared handler with separate pinned and unchecked connection pools. |
| 39 | + /// </summary> |
| 40 | + public static HttpClient CreateHttpClient() |
| 41 | + { |
| 42 | +#if NET |
| 43 | + return new HttpClient(ValidatedAddressHandler.Instance, disposeHandler: false); |
| 44 | +#else |
| 45 | + return HttpClientProvider.GetNonRedirectingHttpClient(); |
| 46 | +#endif |
| 47 | + } |
| 48 | + |
| 49 | +#if NET |
| 50 | + /// <summary> |
| 51 | + /// Keeps pinned operations from reusing unchecked connections opened for document loading or explicit policy bypasses. |
| 52 | + /// </summary> |
| 53 | + private sealed class ValidatedAddressHandler : HttpMessageHandler |
| 54 | + { |
| 55 | + // Plugins have no disposal contract; keep pool ownership at assembly scope rather than allocating pools per plugin. |
| 56 | + public static ValidatedAddressHandler Instance { get; } = new(); |
| 57 | + |
| 58 | + private readonly HttpMessageInvoker _pinnedClient; |
| 59 | + private readonly HttpMessageInvoker _unpinnedClient; |
| 60 | + |
| 61 | + private ValidatedAddressHandler() |
| 62 | + { |
| 63 | + // Share cookies, not sockets, to preserve sessions between document loading and operations. |
| 64 | + var cookies = new CookieContainer(); |
| 65 | + this._pinnedClient = new HttpMessageInvoker(CreateHandler(cookies)); |
| 66 | + this._unpinnedClient = new HttpMessageInvoker(CreateHandler(cookies)); |
| 67 | + } |
| 68 | + |
| 69 | + protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) |
| 70 | + { |
| 71 | + // Reused connections skip ConnectCallback, so pinned requests must never use the unchecked pool. |
| 72 | + var client = request.Options.TryGetValue(ValidatedAddressesKey, out var addresses) && addresses.Length > 0 |
| 73 | + ? this._pinnedClient |
| 74 | + : this._unpinnedClient; |
| 75 | + return client.SendAsync(request, cancellationToken); |
| 76 | + } |
| 77 | + |
| 78 | + protected override void Dispose(bool disposing) |
| 79 | + { |
| 80 | + if (disposing) |
| 81 | + { |
| 82 | + this._pinnedClient.Dispose(); |
| 83 | + this._unpinnedClient.Dispose(); |
| 84 | + } |
| 85 | + base.Dispose(disposing); |
| 86 | + } |
| 87 | + |
| 88 | + private static SocketsHttpHandler CreateHandler(CookieContainer cookies) |
| 89 | + { |
| 90 | + return new SocketsHttpHandler |
| 91 | + { |
| 92 | + AllowAutoRedirect = false, |
| 93 | + CookieContainer = cookies, |
| 94 | + PooledConnectionLifetime = TimeSpan.FromMinutes(2), |
| 95 | + SslOptions = new SslClientAuthenticationOptions |
| 96 | + { |
| 97 | + CertificateRevocationCheckMode = X509RevocationMode.Online, |
| 98 | + }, |
| 99 | + ConnectCallback = ConnectAsync, |
| 100 | + }; |
| 101 | + } |
| 102 | + } |
| 103 | + |
| 104 | + /// <summary> |
| 105 | + /// Connects directly to an address validated for the request, preventing a second DNS lookup from selecting a different destination. |
| 106 | + /// </summary> |
| 107 | + /// <remarks> |
| 108 | + /// For example, <c>https://api.example.com</c> can connect to validated address <c>93.184.216.34</c> |
| 109 | + /// while retaining the hostname for TLS and HTTP. Proxy destinations are left unchanged because the proxy owns DNS resolution. |
| 110 | + /// Address attempts overlap after a short delay, keeping slow routes viable until another address connects. |
| 111 | + /// </remarks> |
| 112 | + private static async ValueTask<Stream> ConnectAsync(SocketsHttpConnectionContext context, CancellationToken cancellationToken) |
| 113 | + { |
| 114 | + var request = context.InitialRequestMessage; |
| 115 | + var destination = context.DnsEndPoint; |
| 116 | + |
| 117 | + // Pin only direct requests with vetted DNS results; proxies and explicit validation bypasses keep their normal routing. |
| 118 | + if (request.RequestUri is null || |
| 119 | + !string.Equals(request.RequestUri.IdnHost, destination.Host, StringComparison.OrdinalIgnoreCase) || |
| 120 | + request.RequestUri.Port != destination.Port || |
| 121 | + !request.Options.TryGetValue(ValidatedAddressesKey, out var validatedAddresses) || |
| 122 | + validatedAddresses.Length == 0) |
| 123 | + { |
| 124 | + return await ConnectSocketAsync(destination, cancellationToken).ConfigureAwait(false); |
| 125 | + } |
| 126 | + |
| 127 | + return await ConnectToValidatedAddressesAsync(validatedAddresses, destination.Port, cancellationToken).ConfigureAwait(false); |
| 128 | + } |
| 129 | + |
| 130 | + /// <summary> |
| 131 | + /// Staggers connections to vetted addresses and disposes losing connections before returning the first success. |
| 132 | + /// </summary> |
| 133 | + /// <remarks> |
| 134 | + /// For example, IPv4 can start while IPv6 is pending, and either can still win. |
| 135 | + /// Only TCP connections are raced; HTTP uses the selected connection afterward. |
| 136 | + /// </remarks> |
| 137 | + private static async ValueTask<Stream> ConnectToValidatedAddressesAsync(IPAddress[] addresses, int port, CancellationToken cancellationToken) |
| 138 | + { |
| 139 | + if (addresses.Length == 1) |
| 140 | + { |
| 141 | + // With no alternative address, racing would only add bookkeeping. |
| 142 | + return await ConnectSocketAsync(new IPEndPoint(addresses[0], port), cancellationToken).ConfigureAwait(false); |
| 143 | + } |
| 144 | + |
| 145 | + // A winner can cancel the losers without cancelling the caller; caller cancellation still stops the race. |
| 146 | + using var raceCancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); |
| 147 | + |
| 148 | + // Keep successes tracked: two addresses may connect before either is selected. |
| 149 | + var attempts = new List<Task<Stream>>(addresses.Length); |
| 150 | + var nextAttemptDelay = Task.CompletedTask; |
| 151 | + int nextAddress = 0; |
| 152 | + Stream? winningStream = null; |
| 153 | + |
| 154 | + try |
| 155 | + { |
| 156 | + while (true) |
| 157 | + { |
| 158 | + cancellationToken.ThrowIfCancellationRequested(); |
| 159 | + |
| 160 | + // Start the next route while this one waits, or immediately if all earlier routes failed. |
| 161 | + if (nextAddress < addresses.Length && (attempts.Count == 0 || nextAttemptDelay.IsCompleted)) |
| 162 | + { |
| 163 | + // AsTask lets both the race and cleanup safely await the same connection result. |
| 164 | + attempts.Add(ConnectSocketAsync(new IPEndPoint(addresses[nextAddress++], port), raceCancellationSource.Token).AsTask()); |
| 165 | + nextAttemptDelay = Task.Delay(s_connectionAttemptDelay, raceCancellationSource.Token); |
| 166 | + } |
| 167 | + |
| 168 | + IEnumerable<Task> candidates = attempts; |
| 169 | + // Include the delay only while another address can be started. |
| 170 | + // Otherwise, an expired delay makes WhenAny return immediately instead of waiting for a connection. |
| 171 | + if (nextAddress < addresses.Length) |
| 172 | + { |
| 173 | + candidates = candidates.Append(nextAttemptDelay); |
| 174 | + } |
| 175 | + |
| 176 | + // Wait for either a connection attempt or the delay task to finish. |
| 177 | + // WhenAny returns the finished task itself, so we can identify which kind of event occurred. |
| 178 | + var completed = await Task.WhenAny(candidates).ConfigureAwait(false); |
| 179 | + if (ReferenceEquals(completed, nextAttemptDelay)) |
| 180 | + { |
| 181 | + // The timer is a launch signal, not an address timeout. |
| 182 | + // The next iteration starts B without cancelling A, or stops if the caller cancelled. |
| 183 | + continue; |
| 184 | + } |
| 185 | + |
| 186 | + // WhenAny reports completion, not success; await the connection below to observe its errors. |
| 187 | + var completedAttempt = attempts.First(attempt => ReferenceEquals(attempt, completed)); |
| 188 | + try |
| 189 | + { |
| 190 | + var stream = await completedAttempt.ConfigureAwait(false); |
| 191 | + |
| 192 | + // If cancellation wins this race, leave the stream tracked so cleanup closes it. |
| 193 | + cancellationToken.ThrowIfCancellationRequested(); |
| 194 | + |
| 195 | + winningStream = stream; |
| 196 | + |
| 197 | + return stream; |
| 198 | + } |
| 199 | + // The failed task is still counted; require another attempt or untried address before ignoring its error. |
| 200 | + catch (SocketException) when (attempts.Count > 1 || nextAddress < addresses.Length) |
| 201 | + { |
| 202 | + // E.g., a refused IPv4 route must not abort a still-pending IPv6 connection. |
| 203 | + attempts.Remove(completedAttempt); |
| 204 | + cancellationToken.ThrowIfCancellationRequested(); |
| 205 | + } |
| 206 | + } |
| 207 | + } |
| 208 | + finally |
| 209 | + { |
| 210 | + // Even return must finish cleanup before transferring ownership of the winning stream. |
| 211 | + raceCancellationSource.Cancel(); |
| 212 | + try |
| 213 | + { |
| 214 | + await DisposeLosingConnectionsAsync(attempts, winningStream).ConfigureAwait(false); |
| 215 | + } |
| 216 | + catch |
| 217 | + { |
| 218 | + // A cleanup failure prevents ownership transfer, even if a connection succeeded. |
| 219 | + winningStream?.Dispose(); |
| 220 | + throw; |
| 221 | + } |
| 222 | + } |
| 223 | + } |
| 224 | + |
| 225 | + /// <summary> |
| 226 | + /// Observes stopped attempts and closes every successful connection except the selected winner. |
| 227 | + /// </summary> |
| 228 | + private static async Task DisposeLosingConnectionsAsync(List<Task<Stream>> attempts, Stream? winningStream) |
| 229 | + { |
| 230 | + var completion = Task.WhenAll(attempts); |
| 231 | + try |
| 232 | + { |
| 233 | + await completion.ConfigureAwait(false); |
| 234 | + } |
| 235 | + catch (SocketException) when (completion.Exception is { } failures) |
| 236 | + { |
| 237 | + // Refused routes are expected; unrelated failures must not turn into a successful request. |
| 238 | + if (failures.InnerExceptions.Any(static exception => exception is not SocketException)) |
| 239 | + { |
| 240 | + throw failures; |
| 241 | + } |
| 242 | + } |
| 243 | + catch (OperationCanceledException) when (completion.IsCanceled) |
| 244 | + { |
| 245 | + } |
| 246 | + finally |
| 247 | + { |
| 248 | + // IPv4 and IPv6 can both connect before selection; cancellation alone does not close either stream. |
| 249 | + foreach (var attempt in attempts) |
| 250 | + { |
| 251 | + if (attempt.IsCompletedSuccessfully) |
| 252 | + { |
| 253 | + var stream = await attempt.ConfigureAwait(false); |
| 254 | + if (!ReferenceEquals(stream, winningStream)) |
| 255 | + { |
| 256 | + stream.Dispose(); |
| 257 | + } |
| 258 | + } |
| 259 | + } |
| 260 | + } |
| 261 | + } |
| 262 | + |
| 263 | + /// <summary> |
| 264 | + /// Opens a TCP stream to the exact endpoint. Passing an <see cref="IPEndPoint"/> avoids another DNS lookup. |
| 265 | + /// </summary> |
| 266 | + private static async ValueTask<Stream> ConnectSocketAsync(EndPoint endpoint, CancellationToken cancellationToken) |
| 267 | + { |
| 268 | + var socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true }; |
| 269 | + try |
| 270 | + { |
| 271 | + await socket.ConnectAsync(endpoint, cancellationToken).ConfigureAwait(false); |
| 272 | + return new NetworkStream(socket, ownsSocket: true); |
| 273 | + } |
| 274 | + catch |
| 275 | + { |
| 276 | + socket.Dispose(); |
| 277 | + throw; |
| 278 | + } |
| 279 | + } |
| 280 | +#endif |
| 281 | +} |
0 commit comments