如何使用 websocket 完成 socks5 网络穿透

internal class Socks5ToWSMiddleware : ITcpProxyMiddleware{    private readonly IForwarderHttpClientFactory httpClientFactory;    private readonly ILoadBalancingPolicyFactory loadBalancing;    private readonly ProxyLogger logger;    private readonly TimeProvider timeProvider;
    public Socks5ToWSMiddleware(IForwarderHttpClientFactory httpClientFactory, ILoadBalancingPolicyFactory loadBalancing, ProxyLogger logger, TimeProvider timeProvider)    {        this.httpClientFactory = httpClientFactory;        this.loadBalancing = loadBalancing;        this.logger = logger;        this.timeProvider = timeProvider;    }
    public Task InitAsync(ConnectionContext context, CancellationToken token, TcpDelegate next)    {        // 过滤符合的路由配置        var feature = context.Features.Get<IL4ReverseProxyFeature>();        if (feature is not null)        {            var route = feature.Route;            if (route is not null && route.Metadata is not null                && route.Metadata.TryGetValue("socks5ToWS", out var b) && bool.TryParse(b, out var isSocks5) && isSocks5)            {                feature.IsDone = true;                route.ClusterConfig?.InitHttp(httpClientFactory);                return Proxy(context, feature, token);            }        }        return next(context, token);    }
    private async Task Proxy(ConnectionContext context, IL4ReverseProxyFeature feature, CancellationToken token)    { // loadBalancing 选取有效 ip        var route = feature.Route;        var cluster = route.ClusterConfig;        DestinationState selectedDestination;        if (cluster is null)        {            selectedDestination = null;        }        else        {            selectedDestination = feature.SelectedDestination;            selectedDestination ??= loadBalancing.PickDestination(feature);        }
        if (selectedDestination is null)        {            logger.NotFoundAvailableUpstream(route.ClusterId);            Abort(context);            return;        }        selectedDestination.ConcurrencyCounter.Increment();        try        {            await SendAsync(context, feature, selectedDestination, cluster, route.Transformer, token);            selectedDestination.ReportSuccessed();        }        catch        {            selectedDestination.ReportFailed();            throw;        }        finally        {            selectedDestination.ConcurrencyCounter.Decrement();        }    }
    private async Task<ForwarderError> SendAsync(ConnectionContext context, IL4ReverseProxyFeature feature, DestinationState selectedDestination, ClusterConfig? cluster, IHttpTransformer transformer, CancellationToken token)    {        // 创建 websocket 请求， 这里为了简单，只创建简单 http1.1 websocket         var destinationPrefix = selectedDestination.Address;        if (destinationPrefix is null || destinationPrefix.Length < 8)        {            throw new ArgumentException("Invalid destination prefix.", nameof(destinationPrefix));        }        var route = feature.Route;        var requestConfig = cluster.HttpRequest ?? ForwarderRequestConfig.Empty;        var httpClient = cluster.HttpMessageHandler ?? throw new ArgumentNullException("httpClient");        var destinationRequest = new HttpRequestMessage();        destinationRequest.Version = HttpVersion.Version11;        destinationRequest.VersionPolicy = HttpVersionPolicy.RequestVersionOrLower;        destinationRequest.Method = HttpMethod.Get;        destinationRequest.RequestUri ??= new Uri(destinationPrefix, UriKind.Absolute);        destinationRequest.Headers.TryAddWithoutValidation(HeaderNames.Connection, HeaderNames.Upgrade);        destinationRequest.Headers.TryAddWithoutValidation(HeaderNames.Upgrade, HttpForwarder.WebSocketName);        destinationRequest.Headers.TryAddWithoutValidation(HeaderNames.SecWebSocketVersion, "13");        destinationRequest.Headers.TryAddWithoutValidation(HeaderNames.SecWebSocketKey, ProtocolHelper.CreateSecWebSocketKey());        destinationRequest.Content = new EmptyHttpContent();        if (!string.IsNullOrWhiteSpace(selectedDestination.Host))        {            destinationRequest.Headers.TryAddWithoutValidation(HeaderNames.Host, selectedDestination.Host);        }                // 建立websocket 链接，成功则直接 复制原始 req/resp 数据，不做任何而外处理        var destinationResponse = await httpClient.SendAsync(destinationRequest, token);        if (destinationResponse.StatusCode == HttpStatusCode.SwitchingProtocols)        {            using var destinationStream = await destinationResponse.Content.ReadAsStreamAsync(token);            var clientStream = new DuplexPipeStreamAdapter<Stream>(null, context.Transport, static i => i);            var activityCancellationSource = ActivityCancellationTokenSource.Rent(route.Timeout);            var requestTask = StreamCopier.CopyAsync(isRequest: true, clientStream, destinationStream, StreamCopier.UnknownLength, timeProvider, activityCancellationSource,                autoFlush: destinationResponse.Version == HttpVersion.Version20, token).AsTask();            var responseTask = StreamCopier.CopyAsync(isRequest: false, destinationStream, clientStream, StreamCopier.UnknownLength, timeProvider, activityCancellationSource, token).AsTask();
            var task = await Task.WhenAny(requestTask, responseTask);            await clientStream.DisposeAsync();            if (task.IsCanceled)            {                Abort(context);                activityCancellationSource.Cancel();                if (task.Exception is not null)                {                    throw task.Exception;                }            }        }        else        {            Abort(context);            return ForwarderError.UpgradeRequestDestination;        }
        return ForwarderError.None;    }
    public Task<ReadOnlyMemory<byte>> OnRequestAsync(ConnectionContext context, ReadOnlyMemory<byte> source, CancellationToken token, TcpProxyDelegate next)    {        return next(context, source, token);    }
    public Task<ReadOnlyMemory<byte>> OnResponseAsync(ConnectionContext context, ReadOnlyMemory<byte> source, CancellationToken token, TcpProxyDelegate next)    {        return next(context, source, token);    }
    private static void Abort(ConnectionContext upstream)    {        upstream.Transport.Input.CancelPendingRead();        upstream.Transport.Output.CancelPendingFlush();        upstream.Abort();    }}

internal class WSToSocks5HttpMiddleware : IMiddleware{    private static ReadOnlySpan<byte> EncodedWebSocketKey => "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"u8;    private WebSocketMiddleware middleware;    private readonly Socks5Middleware socks5Middleware;
    public WSToSocks5HttpMiddleware(IOptions<WebSocketOptions> options, ILoggerFactory loggerFactory, Socks5Middleware socks5Middleware)    {        middleware = new WebSocketMiddleware(Scoks5, options, loggerFactory);        this.socks5Middleware = socks5Middleware;    }
    private async Task Scoks5(HttpContext context)    {        var upgradeFeature = context.Features.Get<IHttpUpgradeFeature>();        // 检查是否未正确 websocket 请求        var f = context.Features.Get<IHttpWebSocketFeature>();        if (f.IsWebSocketRequest)        {            // 返回 websocket 接受信息            var responseHeaders = context.Response.Headers;            responseHeaders.Connection = HeaderNames.Upgrade;            responseHeaders.Upgrade = HttpForwarder.WebSocketName;            responseHeaders.SecWebSocketAccept = CreateResponseKey(context.Request.Headers.SecWebSocketKey.ToString());
            var stream = await upgradeFeature!.UpgradeAsync(); // Sets status code to 101                        // 建原始 websocket stream 包装成 pipe 方便使用原来的 socks5Middleware 实现            var memoryPool = context is IMemoryPoolFeature s ? s.MemoryPool : MemoryPool<byte>.Shared;            StreamPipeReaderOptions readerOptions = new StreamPipeReaderOptions            (                pool: memoryPool,                bufferSize: memoryPool.GetMinimumSegmentSize(),                minimumReadSize: memoryPool.GetMinimumAllocSize(),                leaveOpen: true,                useZeroByteReads: true            );
            var writerOptions = new StreamPipeWriterOptions            (                pool: memoryPool,                leaveOpen: true            );
            var input = PipeReader.Create(stream, readerOptions);            var output = PipeWriter.Create(stream, writerOptions);            var feature = context.Features.Get<IReverseProxyFeature>();            var route = feature.Route;            using var cts = CancellationTokenSourcePool.Default.Rent(route.Timeout);            var token = cts.Token;            context.Features.Set<IL4ReverseProxyFeature>(new L4ReverseProxyFeature() { IsDone = true, Route = route });            // socks5Middleware 进行转发            await socks5Middleware.Proxy(new WebSocketConnection(context.Features)            {                Transport = new WebSocketDuplexPipe() { Input = input, Output = output },                ConnectionId = context.Connection.Id,                Items = context.Items,            }, null, token);        }        else        {            context.Response.StatusCode = StatusCodes.Status400BadRequest;        }    }
    public static string CreateResponseKey(string requestKey)    {        // "The value of this header field is constructed by concatenating /key/, defined above in step 4        // in Section 4.2.2, with the string "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", taking the SHA-1 hash of        // this concatenated value to obtain a 20-byte value and base64-encoding"        // https://tools.ietf.org/html/rfc6455#section-4.2.2
        // requestKey is already verified to be small (24 bytes) by 'IsRequestKeyValid()' and everything is 1:1 mapping to UTF8 bytes        // so this can be hardcoded to 60 bytes for the requestKey + static websocket string        Span<byte> mergedBytes = stackalloc byte[60];        Encoding.UTF8.GetBytes(requestKey, mergedBytes);        EncodedWebSocketKey.CopyTo(mergedBytes[24..]);
        Span<byte> hashedBytes = stackalloc byte[20];        var written = SHA1.HashData(mergedBytes, hashedBytes);        if (written != 20)        {            throw new InvalidOperationException("Could not compute the hash for the 'Sec-WebSocket-Accept' header.");        }
        return Convert.ToBase64String(hashedBytes);    }
    public Task InvokeAsync(HttpContext context, RequestDelegate next)    {       // 过滤符合的路由配置        var feature = context.Features.Get<IReverseProxyFeature>();        if (feature is not null)        {            var route = feature.Route;            if (route is not null && route.Metadata is not null                && route.Metadata.TryGetValue("WSToSocks5", out var b) && bool.TryParse(b, out var isSocks5) && isSocks5)            {                // 这里偷个懒，利用现成的 WebSocketMiddleware 检查 websocket 请求，                return middleware.Invoke(context);            }        }        return next(context);    }}
internal class WebSocketConnection : ConnectionContext{    public WebSocketConnection(IFeatureCollection features)    {        this.features = features;    }
    public override IDuplexPipe Transport { get; set; }    public override string ConnectionId { get; set; }
    private IFeatureCollection features;    public override IFeatureCollection Features => features;
    public override IDictionary<object, object?> Items { get; set; }}
internal class WebSocketDuplexPipe : IDuplexPipe{    public PipeReader Input { get; set; }
    public PipeWriter Output { get; set; }}