Commit 4c97bb50 authored by Honfika's avatar Honfika

Stream reading/writing simplified

parent 50129826
...@@ -3,6 +3,7 @@ using System.Globalization; ...@@ -3,6 +3,7 @@ using System.Globalization;
using System.IO; using System.IO;
using System.Threading.Tasks; using System.Threading.Tasks;
using Titanium.Web.Proxy.Exceptions; using Titanium.Web.Proxy.Exceptions;
using Titanium.Web.Proxy.Helpers;
using Titanium.Web.Proxy.StreamExtended.BufferPool; using Titanium.Web.Proxy.StreamExtended.BufferPool;
using Titanium.Web.Proxy.StreamExtended.Network; using Titanium.Web.Proxy.StreamExtended.Network;
...@@ -11,16 +12,16 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -11,16 +12,16 @@ namespace Titanium.Web.Proxy.EventArguments
internal class LimitedStream : Stream internal class LimitedStream : Stream
{ {
private readonly IBufferPool bufferPool; private readonly IBufferPool bufferPool;
private readonly CustomBufferedStream baseStream; private readonly IHttpStreamReader baseReader;
private readonly bool isChunked; private readonly bool isChunked;
private long bytesRemaining; private long bytesRemaining;
private bool readChunkTrail; private bool readChunkTrail;
internal LimitedStream(CustomBufferedStream baseStream, IBufferPool bufferPool, bool isChunked, internal LimitedStream(IHttpStreamReader baseStream, IBufferPool bufferPool, bool isChunked,
long contentLength) long contentLength)
{ {
this.baseStream = baseStream; this.baseReader = baseStream;
this.bufferPool = bufferPool; this.bufferPool = bufferPool;
this.isChunked = isChunked; this.isChunked = isChunked;
bytesRemaining = isChunked bytesRemaining = isChunked
...@@ -49,12 +50,12 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -49,12 +50,12 @@ namespace Titanium.Web.Proxy.EventArguments
if (readChunkTrail) if (readChunkTrail)
{ {
// read the chunk trail of the previous chunk // read the chunk trail of the previous chunk
string? s = baseStream.ReadLineAsync().Result; string? s = baseReader.ReadLineAsync().Result;
} }
readChunkTrail = true; readChunkTrail = true;
string? chunkHead = baseStream.ReadLineAsync().Result!; string? chunkHead = baseReader.ReadLineAsync().Result!;
int idx = chunkHead.IndexOf(";", StringComparison.Ordinal); int idx = chunkHead.IndexOf(";", StringComparison.Ordinal);
if (idx >= 0) if (idx >= 0)
{ {
...@@ -73,7 +74,7 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -73,7 +74,7 @@ namespace Titanium.Web.Proxy.EventArguments
bytesRemaining = -1; bytesRemaining = -1;
// chunk trail // chunk trail
var task = baseStream.ReadLineAsync(); var task = baseReader.ReadLineAsync();
if (!task.IsCompleted) if (!task.IsCompleted)
task.AsTask().Wait(); task.AsTask().Wait();
} }
...@@ -119,7 +120,7 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -119,7 +120,7 @@ namespace Titanium.Web.Proxy.EventArguments
} }
int toRead = (int)Math.Min(count, bytesRemaining); int toRead = (int)Math.Min(count, bytesRemaining);
int res = baseStream.Read(buffer, offset, toRead); int res = baseReader.Read(buffer, offset, toRead);
bytesRemaining -= res; bytesRemaining -= res;
if (res == 0) if (res == 0)
......
...@@ -10,6 +10,7 @@ using Titanium.Web.Proxy.Http; ...@@ -10,6 +10,7 @@ using Titanium.Web.Proxy.Http;
using Titanium.Web.Proxy.Http.Responses; using Titanium.Web.Proxy.Http.Responses;
using Titanium.Web.Proxy.Models; using Titanium.Web.Proxy.Models;
using Titanium.Web.Proxy.Network; using Titanium.Web.Proxy.Network;
using Titanium.Web.Proxy.Network.Tcp;
using Titanium.Web.Proxy.StreamExtended.Network; using Titanium.Web.Proxy.StreamExtended.Network;
namespace Titanium.Web.Proxy.EventArguments namespace Titanium.Web.Proxy.EventArguments
...@@ -35,8 +36,8 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -35,8 +36,8 @@ namespace Titanium.Web.Proxy.EventArguments
/// <summary> /// <summary>
/// Constructor to initialize the proxy /// Constructor to initialize the proxy
/// </summary> /// </summary>
internal SessionEventArgs(ProxyServer server, ProxyEndPoint endPoint, ProxyClient proxyClient, ConnectRequest? connectRequest, CancellationTokenSource cancellationTokenSource) internal SessionEventArgs(ProxyServer server, ProxyEndPoint endPoint, TcpClientConnection clientConnection, HttpClientStream clientStream, ConnectRequest? connectRequest, CancellationTokenSource cancellationTokenSource)
: base(server, endPoint, proxyClient, connectRequest, new Request(), cancellationTokenSource) : base(server, endPoint, clientConnection, clientStream, connectRequest, new Request(), cancellationTokenSource)
{ {
} }
...@@ -64,14 +65,9 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -64,14 +65,9 @@ namespace Titanium.Web.Proxy.EventArguments
/// </summary> /// </summary>
public event EventHandler<MultipartRequestPartSentEventArgs>? MultipartRequestPartSent; public event EventHandler<MultipartRequestPartSentEventArgs>? MultipartRequestPartSent;
private CustomBufferedStream getStreamReader(bool isRequest) private HttpStream getStream(bool isRequest)
{ {
return isRequest ? ProxyClient.ClientStream : HttpClient.Connection.Stream; return isRequest ? (HttpStream)ClientStream : HttpClient.Connection.Stream;
}
private HttpWriter getStreamWriter(bool isRequest)
{
return isRequest ? (HttpWriter)ProxyClient.ClientStreamWriter : HttpClient.Connection.StreamWriter;
} }
/// <summary> /// <summary>
...@@ -197,22 +193,20 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -197,22 +193,20 @@ namespace Titanium.Web.Proxy.EventArguments
private async Task<byte[]> readBodyAsync(bool isRequest, CancellationToken cancellationToken) private async Task<byte[]> readBodyAsync(bool isRequest, CancellationToken cancellationToken)
{ {
using (var bodyStream = new MemoryStream()) using var bodyStream = new MemoryStream();
{ using var http = new HttpStream(bodyStream, BufferPool);
var writer = new HttpWriter(bodyStream, BufferPool);
if (isRequest) if (isRequest)
{ {
await CopyRequestBodyAsync(writer, TransformationMode.Uncompress, cancellationToken); await CopyRequestBodyAsync(http, TransformationMode.Uncompress, cancellationToken);
} }
else else
{ {
await CopyResponseBodyAsync(writer, TransformationMode.Uncompress, cancellationToken); await CopyResponseBodyAsync(http, TransformationMode.Uncompress, cancellationToken);
} }
return bodyStream.ToArray(); return bodyStream.ToArray();
} }
}
/// <summary> /// <summary>
/// Syphon out any left over data in given request/response from backing tcp connection. /// Syphon out any left over data in given request/response from backing tcp connection.
...@@ -229,18 +223,16 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -229,18 +223,16 @@ namespace Titanium.Web.Proxy.EventArguments
return; return;
} }
using (var bodyStream = new MemoryStream()) using var bodyStream = new MemoryStream();
{ using var http = new HttpStream(bodyStream, BufferPool);
var writer = new HttpWriter(bodyStream, BufferPool); await copyBodyAsync(isRequest, true, http, TransformationMode.None, null, cancellationToken);
await copyBodyAsync(isRequest, true, writer, TransformationMode.None, null, cancellationToken);
}
} }
/// <summary> /// <summary>
/// This is called when the request is PUT/POST/PATCH to read the body /// This is called when the request is PUT/POST/PATCH to read the body
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
internal async Task CopyRequestBodyAsync(HttpWriter writer, TransformationMode transformation, CancellationToken cancellationToken) internal async Task CopyRequestBodyAsync(IHttpStreamWriter writer, TransformationMode transformation, CancellationToken cancellationToken)
{ {
var request = HttpClient.Request; var request = HttpClient.Request;
...@@ -249,7 +241,7 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -249,7 +241,7 @@ namespace Titanium.Web.Proxy.EventArguments
// send the request body bytes to server // send the request body bytes to server
if (contentLength > 0 && hasMulipartEventSubscribers && request.IsMultipartFormData) if (contentLength > 0 && hasMulipartEventSubscribers && request.IsMultipartFormData)
{ {
var reader = getStreamReader(true); var reader = getStream(true);
var boundary = HttpHelper.GetBoundaryFromContentType(request.ContentType); var boundary = HttpHelper.GetBoundaryFromContentType(request.ContentType);
using (var copyStream = new CopyStream(reader, writer, BufferPool)) using (var copyStream = new CopyStream(reader, writer, BufferPool))
...@@ -279,14 +271,14 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -279,14 +271,14 @@ namespace Titanium.Web.Proxy.EventArguments
} }
} }
internal async Task CopyResponseBodyAsync(HttpWriter writer, TransformationMode transformation, CancellationToken cancellationToken) internal async Task CopyResponseBodyAsync(IHttpStreamWriter writer, TransformationMode transformation, CancellationToken cancellationToken)
{ {
await copyBodyAsync(false, false, writer, transformation, OnDataReceived, cancellationToken); await copyBodyAsync(false, false, writer, transformation, OnDataReceived, cancellationToken);
} }
private async Task copyBodyAsync(bool isRequest, bool useOriginalHeaderValues, HttpWriter writer, TransformationMode transformation, Action<byte[], int, int>? onCopy, CancellationToken cancellationToken) private async Task copyBodyAsync(bool isRequest, bool useOriginalHeaderValues, IHttpStreamWriter writer, TransformationMode transformation, Action<byte[], int, int>? onCopy, CancellationToken cancellationToken)
{ {
var stream = getStreamReader(isRequest); var stream = getStream(isRequest);
var requestResponse = isRequest ? (RequestResponseBase)HttpClient.Request : HttpClient.Response; var requestResponse = isRequest ? (RequestResponseBase)HttpClient.Request : HttpClient.Response;
...@@ -313,10 +305,8 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -313,10 +305,8 @@ namespace Titanium.Web.Proxy.EventArguments
try try
{ {
using (var bufStream = new CustomBufferedStream(s, BufferPool, true)) var http = new HttpStream(s, BufferPool, true);
{ await writer.CopyBodyAsync(http, false, -1, onCopy, cancellationToken);
await writer.CopyBodyAsync(bufStream, false, -1, onCopy, cancellationToken);
}
} }
finally finally
{ {
......
...@@ -26,7 +26,12 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -26,7 +26,12 @@ namespace Titanium.Web.Proxy.EventArguments
internal TcpServerConnection ServerConnection => HttpClient.Connection; internal TcpServerConnection ServerConnection => HttpClient.Connection;
internal TcpClientConnection ClientConnection => ProxyClient.Connection; /// <summary>
/// Holds a reference to client
/// </summary>
internal TcpClientConnection ClientConnection { get; }
internal HttpClientStream ClientStream { get; }
protected readonly IBufferPool BufferPool; protected readonly IBufferPool BufferPool;
protected readonly ExceptionHandler ExceptionFunc; protected readonly ExceptionHandler ExceptionFunc;
...@@ -41,7 +46,7 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -41,7 +46,7 @@ namespace Titanium.Web.Proxy.EventArguments
/// Initializes a new instance of the <see cref="SessionEventArgsBase" /> class. /// Initializes a new instance of the <see cref="SessionEventArgsBase" /> class.
/// </summary> /// </summary>
private protected SessionEventArgsBase(ProxyServer server, ProxyEndPoint endPoint, private protected SessionEventArgsBase(ProxyServer server, ProxyEndPoint endPoint,
ProxyClient proxyClient, ConnectRequest? connectRequest, Request request, CancellationTokenSource cancellationTokenSource) TcpClientConnection clientConnection, HttpClientStream clientStream, ConnectRequest? connectRequest, Request request, CancellationTokenSource cancellationTokenSource)
{ {
BufferPool = server.BufferPool; BufferPool = server.BufferPool;
ExceptionFunc = server.ExceptionFunc; ExceptionFunc = server.ExceptionFunc;
...@@ -49,17 +54,13 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -49,17 +54,13 @@ namespace Titanium.Web.Proxy.EventArguments
CancellationTokenSource = cancellationTokenSource; CancellationTokenSource = cancellationTokenSource;
ProxyClient = proxyClient; ClientConnection = clientConnection;
HttpClient = new HttpWebClient(connectRequest, request, new Lazy<int>(() => ProxyClient.Connection.GetProcessId(endPoint))); ClientStream = clientStream;
HttpClient = new HttpWebClient(connectRequest, request, new Lazy<int>(() => clientConnection.GetProcessId(endPoint)));
LocalEndPoint = endPoint; LocalEndPoint = endPoint;
EnableWinAuth = server.EnableWinAuth && isWindowsAuthenticationSupported; EnableWinAuth = server.EnableWinAuth && isWindowsAuthenticationSupported;
} }
/// <summary>
/// Holds a reference to client
/// </summary>
internal ProxyClient ProxyClient { get; }
/// <summary> /// <summary>
/// Returns a user data for this request/response session which is /// Returns a user data for this request/response session which is
/// same as the user data of HttpClient. /// same as the user data of HttpClient.
...@@ -93,7 +94,7 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -93,7 +94,7 @@ namespace Titanium.Web.Proxy.EventArguments
/// <summary> /// <summary>
/// Client End Point. /// Client End Point.
/// </summary> /// </summary>
public IPEndPoint ClientEndPoint => (IPEndPoint)ProxyClient.Connection.RemoteEndPoint; public IPEndPoint ClientEndPoint => (IPEndPoint)ClientConnection.RemoteEndPoint;
/// <summary> /// <summary>
/// The web client used to communicate with server for this session. /// The web client used to communicate with server for this session.
......
using System; using System;
using System.Threading; using System.Threading;
using Titanium.Web.Proxy.Helpers;
using Titanium.Web.Proxy.Http; using Titanium.Web.Proxy.Http;
using Titanium.Web.Proxy.Models; using Titanium.Web.Proxy.Models;
using Titanium.Web.Proxy.Network; using Titanium.Web.Proxy.Network;
using Titanium.Web.Proxy.Network.Tcp;
using Titanium.Web.Proxy.StreamExtended.Network; using Titanium.Web.Proxy.StreamExtended.Network;
namespace Titanium.Web.Proxy.EventArguments namespace Titanium.Web.Proxy.EventArguments
...@@ -15,8 +17,8 @@ namespace Titanium.Web.Proxy.EventArguments ...@@ -15,8 +17,8 @@ namespace Titanium.Web.Proxy.EventArguments
private bool? isHttpsConnect; private bool? isHttpsConnect;
internal TunnelConnectSessionEventArgs(ProxyServer server, ProxyEndPoint endPoint, ConnectRequest connectRequest, internal TunnelConnectSessionEventArgs(ProxyServer server, ProxyEndPoint endPoint, ConnectRequest connectRequest,
ProxyClient proxyClient, CancellationTokenSource cancellationTokenSource) TcpClientConnection clientConnection, HttpClientStream clientStream, CancellationTokenSource cancellationTokenSource)
: base(server, endPoint, proxyClient, connectRequest, connectRequest, cancellationTokenSource) : base(server, endPoint, clientConnection, clientStream, connectRequest, connectRequest, cancellationTokenSource)
{ {
} }
......
...@@ -36,8 +36,7 @@ namespace Titanium.Web.Proxy ...@@ -36,8 +36,7 @@ namespace Titanium.Web.Proxy
var cancellationTokenSource = new CancellationTokenSource(); var cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = cancellationTokenSource.Token; var cancellationToken = cancellationTokenSource.Token;
var clientStream = new CustomBufferedStream(clientConnection.GetStream(), BufferPool); var clientStream = new HttpClientStream(clientConnection.GetStream(), BufferPool);
var clientStreamWriter = new HttpResponseWriter(clientStream, BufferPool);
Task<TcpServerConnection>? prefetchConnectionTask = null; Task<TcpServerConnection>? prefetchConnectionTask = null;
bool closeServerConnection = false; bool closeServerConnection = false;
...@@ -70,7 +69,7 @@ namespace Titanium.Web.Proxy ...@@ -70,7 +69,7 @@ namespace Titanium.Web.Proxy
await HeaderParser.ReadHeaders(clientStream, connectRequest.Headers, cancellationToken); await HeaderParser.ReadHeaders(clientStream, connectRequest.Headers, cancellationToken);
connectArgs = new TunnelConnectSessionEventArgs(this, endPoint, connectRequest, connectArgs = new TunnelConnectSessionEventArgs(this, endPoint, connectRequest,
new ProxyClient(clientConnection, clientStream, clientStreamWriter), cancellationTokenSource); clientConnection, clientStream, cancellationTokenSource);
clientStream.DataRead += (o, args) => connectArgs.OnDataSent(args.Buffer, args.Offset, args.Count); clientStream.DataRead += (o, args) => connectArgs.OnDataSent(args.Buffer, args.Offset, args.Count);
clientStream.DataWrite += (o, args) => connectArgs.OnDataReceived(args.Buffer, args.Offset, args.Count); clientStream.DataWrite += (o, args) => connectArgs.OnDataReceived(args.Buffer, args.Offset, args.Count);
...@@ -92,7 +91,7 @@ namespace Titanium.Web.Proxy ...@@ -92,7 +91,7 @@ namespace Titanium.Web.Proxy
} }
// send the response // send the response
await clientStreamWriter.WriteResponseAsync(connectArgs.HttpClient.Response, await clientStream.WriteResponseAsync(connectArgs.HttpClient.Response,
cancellationToken: cancellationToken); cancellationToken: cancellationToken);
return; return;
} }
...@@ -102,7 +101,7 @@ namespace Titanium.Web.Proxy ...@@ -102,7 +101,7 @@ namespace Titanium.Web.Proxy
await endPoint.InvokeBeforeTunnelConnectResponse(this, connectArgs, ExceptionFunc); await endPoint.InvokeBeforeTunnelConnectResponse(this, connectArgs, ExceptionFunc);
// send the response // send the response
await clientStreamWriter.WriteResponseAsync(connectArgs.HttpClient.Response, await clientStream.WriteResponseAsync(connectArgs.HttpClient.Response,
cancellationToken: cancellationToken); cancellationToken: cancellationToken);
return; return;
} }
...@@ -115,7 +114,7 @@ namespace Titanium.Web.Proxy ...@@ -115,7 +114,7 @@ namespace Titanium.Web.Proxy
response.Headers.FixProxyHeaders(); response.Headers.FixProxyHeaders();
connectArgs.HttpClient.Response = response; connectArgs.HttpClient.Response = response;
await clientStreamWriter.WriteResponseAsync(response, cancellationToken: cancellationToken); await clientStream.WriteResponseAsync(response, cancellationToken: cancellationToken);
var clientHelloInfo = await SslTools.PeekClientHello(clientStream, BufferPool, cancellationToken); var clientHelloInfo = await SslTools.PeekClientHello(clientStream, BufferPool, cancellationToken);
...@@ -216,10 +215,9 @@ namespace Titanium.Web.Proxy ...@@ -216,10 +215,9 @@ namespace Titanium.Web.Proxy
#endif #endif
// HTTPS server created - we can now decrypt the client's traffic // HTTPS server created - we can now decrypt the client's traffic
clientStream = new CustomBufferedStream(sslStream, BufferPool); clientStream = new HttpClientStream(sslStream, BufferPool);
clientStream.DataRead += (o, args) => connectArgs.OnDecryptedDataSent(args.Buffer, args.Offset, args.Count); clientStream.DataRead += (o, args) => connectArgs.OnDecryptedDataSent(args.Buffer, args.Offset, args.Count);
clientStream.DataWrite += (o, args) => connectArgs.OnDecryptedDataReceived(args.Buffer, args.Offset, args.Count); clientStream.DataWrite += (o, args) => connectArgs.OnDecryptedDataReceived(args.Buffer, args.Offset, args.Count);
clientStreamWriter = new HttpResponseWriter(clientStream, BufferPool);
} }
catch (Exception e) catch (Exception e)
{ {
...@@ -274,7 +272,7 @@ namespace Titanium.Web.Proxy ...@@ -274,7 +272,7 @@ namespace Titanium.Web.Proxy
{ {
// clientStream.Available should be at most BufferSize because it is using the same buffer size // clientStream.Available should be at most BufferSize because it is using the same buffer size
await clientStream.ReadAsync(data, 0, available, cancellationToken); await clientStream.ReadAsync(data, 0, available, cancellationToken);
await connection.StreamWriter.WriteAsync(data, 0, available, true, cancellationToken); await connection.Stream.WriteAsync(data, 0, available, true, cancellationToken);
} }
finally finally
{ {
...@@ -335,9 +333,9 @@ namespace Titanium.Web.Proxy ...@@ -335,9 +333,9 @@ namespace Titanium.Web.Proxy
{ {
#if NETSTANDARD2_1 #if NETSTANDARD2_1
var connectionPreface = new ReadOnlyMemory<byte>(Http2Helper.ConnectionPreface); var connectionPreface = new ReadOnlyMemory<byte>(Http2Helper.ConnectionPreface);
await connection.StreamWriter.WriteAsync(connectionPreface, cancellationToken); await connection.Stream.WriteAsync(connectionPreface, cancellationToken);
await Http2Helper.SendHttp2(clientStream, connection.Stream, await Http2Helper.SendHttp2(clientStream, connection.Stream,
() => new SessionEventArgs(this, endPoint, new ProxyClient(clientConnection, clientStream, clientStreamWriter), connectArgs?.HttpClient.ConnectRequest, cancellationTokenSource) () => new SessionEventArgs(this, endPoint, clientConnection, clientStream, connectArgs?.HttpClient.ConnectRequest, cancellationTokenSource)
{ {
UserData = connectArgs?.UserData UserData = connectArgs?.UserData
}, },
...@@ -356,8 +354,7 @@ namespace Titanium.Web.Proxy ...@@ -356,8 +354,7 @@ namespace Titanium.Web.Proxy
calledRequestHandler = true; calledRequestHandler = true;
// Now create the request // Now create the request
await handleHttpSessionRequest(endPoint, clientConnection, clientStream, clientStreamWriter, await handleHttpSessionRequest(endPoint, clientConnection, clientStream, cancellationTokenSource, connectArgs, prefetchConnectionTask);
cancellationTokenSource, connectArgs, prefetchConnectionTask);
} }
catch (ProxyException e) catch (ProxyException e)
{ {
......
...@@ -7,9 +7,9 @@ using Titanium.Web.Proxy.StreamExtended.BufferPool; ...@@ -7,9 +7,9 @@ using Titanium.Web.Proxy.StreamExtended.BufferPool;
namespace Titanium.Web.Proxy.Helpers namespace Titanium.Web.Proxy.Helpers
{ {
internal sealed class HttpResponseWriter : HttpWriter internal sealed class HttpClientStream : HttpStream
{ {
internal HttpResponseWriter(Stream stream, IBufferPool bufferPool) internal HttpClientStream(Stream stream, IBufferPool bufferPool)
: base(stream, bufferPool) : base(stream, bufferPool)
{ {
} }
......
...@@ -170,18 +170,18 @@ namespace Titanium.Web.Proxy.Helpers ...@@ -170,18 +170,18 @@ namespace Titanium.Web.Proxy.Helpers
/// Determines whether is connect method. /// Determines whether is connect method.
/// </summary> /// </summary>
/// <returns>1: when CONNECT, 0: when valid HTTP method, -1: otherwise</returns> /// <returns>1: when CONNECT, 0: when valid HTTP method, -1: otherwise</returns>
internal static Task<int> IsConnectMethod(CustomBufferedStream clientStreamReader, IBufferPool bufferPool, CancellationToken cancellationToken = default) internal static Task<int> IsConnectMethod(IPeekStream httpReader, IBufferPool bufferPool, CancellationToken cancellationToken = default)
{ {
return startsWith(clientStreamReader, bufferPool, "CONNECT", cancellationToken); return startsWith(httpReader, bufferPool, "CONNECT", cancellationToken);
} }
/// <summary> /// <summary>
/// Determines whether is pri method (HTTP/2). /// Determines whether is pri method (HTTP/2).
/// </summary> /// </summary>
/// <returns>1: when PRI, 0: when valid HTTP method, -1: otherwise</returns> /// <returns>1: when PRI, 0: when valid HTTP method, -1: otherwise</returns>
internal static Task<int> IsPriMethod(CustomBufferedStream clientStreamReader, IBufferPool bufferPool, CancellationToken cancellationToken = default) internal static Task<int> IsPriMethod(IPeekStream httpReader, IBufferPool bufferPool, CancellationToken cancellationToken = default)
{ {
return startsWith(clientStreamReader, bufferPool, "PRI", cancellationToken); return startsWith(httpReader, bufferPool, "PRI", cancellationToken);
} }
/// <summary> /// <summary>
...@@ -190,7 +190,7 @@ namespace Titanium.Web.Proxy.Helpers ...@@ -190,7 +190,7 @@ namespace Titanium.Web.Proxy.Helpers
/// <returns> /// <returns>
/// 1: when starts with the given string, 0: when valid HTTP method, -1: otherwise /// 1: when starts with the given string, 0: when valid HTTP method, -1: otherwise
/// </returns> /// </returns>
private static async Task<int> startsWith(CustomBufferedStream clientStreamReader, IBufferPool bufferPool, string expectedStart, CancellationToken cancellationToken = default) private static async Task<int> startsWith(IPeekStream httpReader, IBufferPool bufferPool, string expectedStart, CancellationToken cancellationToken = default)
{ {
const int lengthToCheck = 10; const int lengthToCheck = 10;
if (bufferPool.BufferSize < lengthToCheck) if (bufferPool.BufferSize < lengthToCheck)
...@@ -205,7 +205,7 @@ namespace Titanium.Web.Proxy.Helpers ...@@ -205,7 +205,7 @@ namespace Titanium.Web.Proxy.Helpers
int i = 0; int i = 0;
while (i < lengthToCheck) while (i < lengthToCheck)
{ {
int peeked = await clientStreamReader.PeekBytesAsync(buffer, i, i, lengthToCheck - i, cancellationToken); int peeked = await httpReader.PeekBytesAsync(buffer, i, i, lengthToCheck - i, cancellationToken);
if (peeked <= 0) if (peeked <= 0)
return -1; return -1;
......
...@@ -6,9 +6,9 @@ using Titanium.Web.Proxy.StreamExtended.BufferPool; ...@@ -6,9 +6,9 @@ using Titanium.Web.Proxy.StreamExtended.BufferPool;
namespace Titanium.Web.Proxy.Helpers namespace Titanium.Web.Proxy.Helpers
{ {
internal sealed class HttpRequestWriter : HttpWriter internal sealed class HttpServerStream : HttpStream
{ {
internal HttpRequestWriter(Stream stream, IBufferPool bufferPool) internal HttpServerStream(Stream stream, IBufferPool bufferPool)
: base(stream, bufferPool) : base(stream, bufferPool)
{ {
} }
......
using System; using System;
using System.Buffers;
using System.Diagnostics; using System.Diagnostics;
using System.Globalization;
using System.IO; using System.IO;
using System.Net.Sockets; using System.Net.Sockets;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Titanium.Web.Proxy.Helpers; using Titanium.Web.Proxy.Http;
using Titanium.Web.Proxy.Models; using Titanium.Web.Proxy.Models;
using Titanium.Web.Proxy.Shared;
using Titanium.Web.Proxy.StreamExtended.BufferPool; using Titanium.Web.Proxy.StreamExtended.BufferPool;
using Titanium.Web.Proxy.StreamExtended.Network;
namespace Titanium.Web.Proxy.StreamExtended.Network namespace Titanium.Web.Proxy.Helpers
{ {
/// <summary> internal class HttpStream : Stream, IHttpStreamWriter, IHttpStreamReader, IPeekStream
/// A custom network stream inherited from stream
/// with an underlying read buffer supporting both read/write
/// of UTF-8 encoded string or raw bytes asynchronously from last read position.
/// </summary>
/// <seealso cref="System.IO.Stream" />
internal class CustomBufferedStream : Stream, IPeekStream, ILineStream
{ {
private readonly bool leaveOpen; private readonly bool leaveOpen;
private readonly byte[] streamBuffer; private readonly byte[] streamBuffer;
...@@ -44,7 +42,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -44,7 +42,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
public bool IsClosed => closed; public bool IsClosed => closed;
static CustomBufferedStream() static HttpStream()
{ {
// TODO: remove this hack when removing .NET 4.x support // TODO: remove this hack when removing .NET 4.x support
try try
...@@ -62,13 +60,15 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -62,13 +60,15 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
} }
} }
private static readonly byte[] newLine = ProxyConstants.NewLineBytes;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="CustomBufferedStream"/> class. /// Initializes a new instance of the <see cref="HttpStream"/> class.
/// </summary> /// </summary>
/// <param name="baseStream">The base stream.</param> /// <param name="baseStream">The base stream.</param>
/// <param name="bufferPool">Bufferpool.</param> /// <param name="bufferPool">Bufferpool.</param>
/// <param name="leaveOpen"><see langword="true" /> to leave the stream open after disposing the <see cref="T:CustomBufferedStream" /> object; otherwise, <see langword="false" />.</param> /// <param name="leaveOpen"><see langword="true" /> to leave the stream open after disposing the <see cref="T:CustomBufferedStream" /> object; otherwise, <see langword="false" />.</param>
public CustomBufferedStream(Stream baseStream, IBufferPool bufferPool, bool leaveOpen = false) internal HttpStream(Stream baseStream, IBufferPool bufferPool, bool leaveOpen = false)
{ {
BaseStream = baseStream; BaseStream = baseStream;
this.leaveOpen = leaveOpen; this.leaveOpen = leaveOpen;
...@@ -138,8 +138,8 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -138,8 +138,8 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <summary> /// <summary>
/// When overridden in a derived class, writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written. /// When overridden in a derived class, writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written.
/// </summary> /// </summary>
/// <param name="buffer">An array of bytes. This method copies <paramref name="count" /> bytes from <paramref name="buffer" /> to the current stream.</param> /// <param name="buffer">An array of bytes. This method copies count bytes from buffer to the current stream.</param>
/// <param name="offset">The zero-based byte offset in <paramref name="buffer" /> at which to begin copying bytes to the current stream.</param> /// <param name="offset">The zero-based byte offset in buffer at which to begin copying bytes to the current stream.</param>
/// <param name="count">The number of bytes to be written to the current stream.</param> /// <param name="count">The number of bytes to be written to the current stream.</param>
[DebuggerStepThrough] [DebuggerStepThrough]
public override void Write(byte[] buffer, int offset, int count) public override void Write(byte[] buffer, int offset, int count)
...@@ -378,12 +378,9 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -378,12 +378,9 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests. /// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary> /// </summary>
/// <param name="buffer">The buffer to write data from.</param> /// <param name="buffer">The buffer to write data from.</param>
/// <param name="offset">The zero-based byte offset in <paramref name="buffer" /> from which to begin copying bytes to the stream.</param> /// <param name="offset">The zero-based byte offset in buffer from which to begin copying bytes to the stream.</param>
/// <param name="count">The maximum number of bytes to write.</param> /// <param name="count">The maximum number of bytes to write.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param> /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None"></see>.</param>
/// <returns>
/// A task that represents the asynchronous write operation.
/// </returns>
[DebuggerStepThrough] [DebuggerStepThrough]
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{ {
...@@ -756,5 +753,292 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -756,5 +753,292 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
((TaskResult)asyncResult).GetResult(); ((TaskResult)asyncResult).GetResult();
} }
/// <summary>
/// Writes a line async
/// </summary>
/// <param name="cancellationToken">Optional cancellation token for this async task.</param>
/// <returns></returns>
internal Task WriteLineAsync(CancellationToken cancellationToken = default)
{
return WriteAsync(newLine, cancellationToken: cancellationToken);
}
internal Task WriteAsync(string value, CancellationToken cancellationToken = default)
{
return writeAsyncInternal(value, false, cancellationToken);
}
private async Task writeAsyncInternal(string value, bool addNewLine, CancellationToken cancellationToken)
{
int newLineChars = addNewLine ? newLine.Length : 0;
int charCount = value.Length;
if (charCount < bufferPool.BufferSize - newLineChars)
{
var buffer = bufferPool.GetBuffer();
try
{
int idx = encoding.GetBytes(value, 0, charCount, buffer, 0);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
await BaseStream.WriteAsync(buffer, 0, idx, cancellationToken);
}
finally
{
bufferPool.ReturnBuffer(buffer);
}
}
else
{
var buffer = new byte[charCount + newLineChars + 1];
int idx = encoding.GetBytes(value, 0, charCount, buffer, 0);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
await BaseStream.WriteAsync(buffer, 0, idx, cancellationToken);
}
}
internal Task WriteLineAsync(string value, CancellationToken cancellationToken = default)
{
return writeAsyncInternal(value, true, cancellationToken);
}
/// <summary>
/// Write the headers to client
/// </summary>
/// <param name="headerBuilder"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
internal async Task WriteHeadersAsync(HeaderBuilder headerBuilder, CancellationToken cancellationToken = default)
{
var buffer = headerBuilder.GetBuffer();
await WriteAsync(buffer.Array, buffer.Offset, buffer.Count, true, cancellationToken);
}
/// <summary>
/// Writes the data to the stream.
/// </summary>
/// <param name="data">The data.</param>
/// <param name="flush">Should we flush after write?</param>
/// <param name="cancellationToken">The cancellation token.</param>
internal async Task WriteAsync(byte[] data, bool flush = false, CancellationToken cancellationToken = default)
{
await BaseStream.WriteAsync(data, 0, data.Length, cancellationToken);
if (flush)
{
await BaseStream.FlushAsync(cancellationToken);
}
}
internal async Task WriteAsync(byte[] data, int offset, int count, bool flush,
CancellationToken cancellationToken = default)
{
await BaseStream.WriteAsync(data, offset, count, cancellationToken);
if (flush)
{
await BaseStream.FlushAsync(cancellationToken);
}
}
/// <summary>
/// Writes the byte array body to the stream; optionally chunked
/// </summary>
/// <param name="data"></param>
/// <param name="isChunked"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
internal Task WriteBodyAsync(byte[] data, bool isChunked, CancellationToken cancellationToken)
{
if (isChunked)
{
return writeBodyChunkedAsync(data, cancellationToken);
}
return WriteAsync(data, cancellationToken: cancellationToken);
}
/// <summary>
/// Copies the specified content length number of bytes to the output stream from the given inputs stream
/// optionally chunked
/// </summary>
/// <param name="streamReader"></param>
/// <param name="isChunked"></param>
/// <param name="contentLength"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public Task CopyBodyAsync(IHttpStreamReader streamReader, bool isChunked, long contentLength,
Action<byte[], int, int>? onCopy, CancellationToken cancellationToken)
{
// For chunked request we need to read data as they arrive, until we reach a chunk end symbol
if (isChunked)
{
return copyBodyChunkedAsync(streamReader, onCopy, cancellationToken);
}
// http 1.0 or the stream reader limits the stream
if (contentLength == -1)
{
contentLength = long.MaxValue;
}
// If not chunked then its easy just read the amount of bytes mentioned in content length header
return copyBytesFromStream(streamReader, contentLength, onCopy, cancellationToken);
}
/// <summary>
/// Copies the given input bytes to output stream chunked
/// </summary>
/// <param name="data"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task writeBodyChunkedAsync(byte[] data, CancellationToken cancellationToken)
{
var chunkHead = Encoding.ASCII.GetBytes(data.Length.ToString("x2"));
await WriteAsync(chunkHead, cancellationToken: cancellationToken);
await WriteLineAsync(cancellationToken);
await WriteAsync(data, cancellationToken: cancellationToken);
await WriteLineAsync(cancellationToken);
await WriteLineAsync("0", cancellationToken);
await WriteLineAsync(cancellationToken);
}
/// <summary>
/// Copies the streams chunked
/// </summary>
/// <param name="reader"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task copyBodyChunkedAsync(IHttpStreamReader reader, Action<byte[], int, int>? onCopy, CancellationToken cancellationToken)
{
while (true)
{
string chunkHead = (await reader.ReadLineAsync(cancellationToken))!;
int idx = chunkHead.IndexOf(";");
if (idx >= 0)
{
chunkHead = chunkHead.Substring(0, idx);
}
int chunkSize = int.Parse(chunkHead, NumberStyles.HexNumber);
await WriteLineAsync(chunkHead, cancellationToken);
if (chunkSize != 0)
{
await copyBytesFromStream(reader, chunkSize, onCopy, cancellationToken);
}
await WriteLineAsync(cancellationToken);
// chunk trail
await reader.ReadLineAsync(cancellationToken);
if (chunkSize == 0)
{
break;
}
}
}
/// <summary>
/// Copies the specified bytes to the stream from the input stream
/// </summary>
/// <param name="reader"></param>
/// <param name="count"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task copyBytesFromStream(IHttpStreamReader reader, long count, Action<byte[], int, int>? onCopy,
CancellationToken cancellationToken)
{
var buffer = bufferPool.GetBuffer();
try
{
long remainingBytes = count;
while (remainingBytes > 0)
{
int bytesToRead = buffer.Length;
if (remainingBytes < bytesToRead)
{
bytesToRead = (int)remainingBytes;
}
int bytesRead = await reader.ReadAsync(buffer, 0, bytesToRead, cancellationToken);
if (bytesRead == 0)
{
break;
}
remainingBytes -= bytesRead;
await BaseStream.WriteAsync(buffer, 0, bytesRead, cancellationToken);
onCopy?.Invoke(buffer, 0, bytesRead);
}
}
finally
{
bufferPool.ReturnBuffer(buffer);
}
}
/// <summary>
/// Writes the request/response headers and body.
/// </summary>
/// <param name="requestResponse"></param>
/// <param name="headerBuilder"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
protected async Task WriteAsync(RequestResponseBase requestResponse, HeaderBuilder headerBuilder, CancellationToken cancellationToken = default)
{
var body = requestResponse.CompressBodyAndUpdateContentLength();
headerBuilder.WriteHeaders(requestResponse.Headers);
await WriteHeadersAsync(headerBuilder, cancellationToken);
if (body != null)
{
await WriteBodyAsync(body, requestResponse.IsChunked, cancellationToken);
}
}
#if NETSTANDARD2_1
/// <summary>
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>A task that represents the asynchronous write operation.</returns>
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{
return BaseStream.WriteAsync(buffer, cancellationToken);
}
#else
/// <summary>
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>A task that represents the asynchronous write operation.</returns>
public Task WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)
{
var buf = ArrayPool<byte>.Shared.Rent(buffer.Length);
buffer.CopyTo(buf);
return BaseStream.WriteAsync(buf, 0, buf.Length, cancellationToken);
}
#endif
} }
} }
using System;
using System.Buffers;
using System.Globalization;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Titanium.Web.Proxy.Http;
using Titanium.Web.Proxy.Models;
using Titanium.Web.Proxy.Shared;
using Titanium.Web.Proxy.StreamExtended.BufferPool;
using Titanium.Web.Proxy.StreamExtended.Network;
namespace Titanium.Web.Proxy.Helpers
{
internal class HttpWriter : ICustomStreamWriter
{
private readonly Stream stream;
private readonly IBufferPool bufferPool;
private static readonly byte[] newLine = ProxyConstants.NewLineBytes;
private static Encoding encoding => HttpHeader.Encoding;
internal HttpWriter(Stream stream, IBufferPool bufferPool)
{
this.stream = stream;
this.bufferPool = bufferPool;
}
/// <summary>
/// Writes a line async
/// </summary>
/// <param name="cancellationToken">Optional cancellation token for this async task.</param>
/// <returns></returns>
internal Task WriteLineAsync(CancellationToken cancellationToken = default)
{
return WriteAsync(newLine, cancellationToken: cancellationToken);
}
internal Task WriteAsync(string value, CancellationToken cancellationToken = default)
{
return writeAsyncInternal(value, false, cancellationToken);
}
private async Task writeAsyncInternal(string value, bool addNewLine, CancellationToken cancellationToken)
{
int newLineChars = addNewLine ? newLine.Length : 0;
int charCount = value.Length;
if (charCount < bufferPool.BufferSize - newLineChars)
{
var buffer = bufferPool.GetBuffer();
try
{
int idx = encoding.GetBytes(value, 0, charCount, buffer, 0);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
await stream.WriteAsync(buffer, 0, idx, cancellationToken);
}
finally
{
bufferPool.ReturnBuffer(buffer);
}
}
else
{
var buffer = new byte[charCount + newLineChars + 1];
int idx = encoding.GetBytes(value, 0, charCount, buffer, 0);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
await stream.WriteAsync(buffer, 0, idx, cancellationToken);
}
}
internal Task WriteLineAsync(string value, CancellationToken cancellationToken = default)
{
return writeAsyncInternal(value, true, cancellationToken);
}
/// <summary>
/// Write the headers to client
/// </summary>
/// <param name="headerBuilder"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
internal async Task WriteHeadersAsync(HeaderBuilder headerBuilder, CancellationToken cancellationToken = default)
{
var buffer = headerBuilder.GetBuffer();
await WriteAsync(buffer.Array, buffer.Offset, buffer.Count, true, cancellationToken);
}
/// <summary>
/// Writes the data to the stream.
/// </summary>
/// <param name="data">The data.</param>
/// <param name="flush">Should we flush after write?</param>
/// <param name="cancellationToken">The cancellation token.</param>
internal async Task WriteAsync(byte[] data, bool flush = false, CancellationToken cancellationToken = default)
{
await stream.WriteAsync(data, 0, data.Length, cancellationToken);
if (flush)
{
await stream.FlushAsync(cancellationToken);
}
}
internal async Task WriteAsync(byte[] data, int offset, int count, bool flush,
CancellationToken cancellationToken = default)
{
await stream.WriteAsync(data, offset, count, cancellationToken);
if (flush)
{
await stream.FlushAsync(cancellationToken);
}
}
/// <summary>
/// Writes the byte array body to the stream; optionally chunked
/// </summary>
/// <param name="data"></param>
/// <param name="isChunked"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
internal Task WriteBodyAsync(byte[] data, bool isChunked, CancellationToken cancellationToken)
{
if (isChunked)
{
return writeBodyChunkedAsync(data, cancellationToken);
}
return WriteAsync(data, cancellationToken: cancellationToken);
}
/// <summary>
/// Copies the specified content length number of bytes to the output stream from the given inputs stream
/// optionally chunked
/// </summary>
/// <param name="streamReader"></param>
/// <param name="isChunked"></param>
/// <param name="contentLength"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
internal Task CopyBodyAsync(CustomBufferedStream streamReader, bool isChunked, long contentLength,
Action<byte[], int, int>? onCopy, CancellationToken cancellationToken)
{
// For chunked request we need to read data as they arrive, until we reach a chunk end symbol
if (isChunked)
{
return copyBodyChunkedAsync(streamReader, onCopy, cancellationToken);
}
// http 1.0 or the stream reader limits the stream
if (contentLength == -1)
{
contentLength = long.MaxValue;
}
// If not chunked then its easy just read the amount of bytes mentioned in content length header
return copyBytesFromStream(streamReader, contentLength, onCopy, cancellationToken);
}
/// <summary>
/// Copies the given input bytes to output stream chunked
/// </summary>
/// <param name="data"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task writeBodyChunkedAsync(byte[] data, CancellationToken cancellationToken)
{
var chunkHead = Encoding.ASCII.GetBytes(data.Length.ToString("x2"));
await WriteAsync(chunkHead, cancellationToken: cancellationToken);
await WriteLineAsync(cancellationToken);
await WriteAsync(data, cancellationToken: cancellationToken);
await WriteLineAsync(cancellationToken);
await WriteLineAsync("0", cancellationToken);
await WriteLineAsync(cancellationToken);
}
/// <summary>
/// Copies the streams chunked
/// </summary>
/// <param name="reader"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task copyBodyChunkedAsync(CustomBufferedStream reader, Action<byte[], int, int>? onCopy, CancellationToken cancellationToken)
{
while (true)
{
string chunkHead = (await reader.ReadLineAsync(cancellationToken))!;
int idx = chunkHead.IndexOf(";");
if (idx >= 0)
{
chunkHead = chunkHead.Substring(0, idx);
}
int chunkSize = int.Parse(chunkHead, NumberStyles.HexNumber);
await WriteLineAsync(chunkHead, cancellationToken);
if (chunkSize != 0)
{
await copyBytesFromStream(reader, chunkSize, onCopy, cancellationToken);
}
await WriteLineAsync(cancellationToken);
// chunk trail
await reader.ReadLineAsync(cancellationToken);
if (chunkSize == 0)
{
break;
}
}
}
/// <summary>
/// Copies the specified bytes to the stream from the input stream
/// </summary>
/// <param name="reader"></param>
/// <param name="count"></param>
/// <param name="onCopy"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
private async Task copyBytesFromStream(CustomBufferedStream reader, long count, Action<byte[], int, int>? onCopy,
CancellationToken cancellationToken)
{
var buffer = bufferPool.GetBuffer();
try
{
long remainingBytes = count;
while (remainingBytes > 0)
{
int bytesToRead = buffer.Length;
if (remainingBytes < bytesToRead)
{
bytesToRead = (int)remainingBytes;
}
int bytesRead = await reader.ReadAsync(buffer, 0, bytesToRead, cancellationToken);
if (bytesRead == 0)
{
break;
}
remainingBytes -= bytesRead;
await stream.WriteAsync(buffer, 0, bytesRead, cancellationToken);
onCopy?.Invoke(buffer, 0, bytesRead);
}
}
finally
{
bufferPool.ReturnBuffer(buffer);
}
}
/// <summary>
/// Writes the request/response headers and body.
/// </summary>
/// <param name="requestResponse"></param>
/// <param name="headerBuilder"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
protected async Task WriteAsync(RequestResponseBase requestResponse, HeaderBuilder headerBuilder, CancellationToken cancellationToken = default)
{
var body = requestResponse.CompressBodyAndUpdateContentLength();
headerBuilder.WriteHeaders(requestResponse.Headers);
await WriteHeadersAsync(headerBuilder, cancellationToken);
if (body != null)
{
await WriteBodyAsync(body, requestResponse.IsChunked, cancellationToken);
}
}
/// <summary>When overridden in a derived class, writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written.</summary>
/// <param name="buffer">An array of bytes. This method copies count bytes from buffer to the current stream.</param>
/// <param name="offset">The zero-based byte offset in buffer at which to begin copying bytes to the current stream.</param>
/// <param name="count">The number of bytes to be written to the current stream.</param>
/// <exception cref="T:System.ArgumentException">The sum of offset and count is greater than the buffer length.</exception>
/// <exception cref="T:System.ArgumentNullException">buffer is null.</exception>
/// <exception cref="T:System.ArgumentOutOfRangeException">offset or count is negative.</exception>
/// <exception cref="T:System.IO.IOException">An I/O error occured, such as the specified file cannot be found.</exception>
/// <exception cref="T:System.NotSupportedException">The stream does not support writing.</exception>
/// <exception cref="T:System.ObjectDisposedException"><see cref="M:System.IO.Stream.Write(System.Byte[],System.Int32,System.Int32)"></see> was called after the stream was closed.</exception>
public void Write(byte[] buffer, int offset, int count)
{
stream.Write(buffer, offset, count);
}
/// <summary>
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
/// <param name="offset">The zero-based byte offset in <paramref name="buffer" /> from which to begin copying bytes to the stream.</param>
/// <param name="count">The maximum number of bytes to write.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>A task that represents the asynchronous write operation.</returns>
public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
return stream.WriteAsync(buffer, offset, count, cancellationToken);
}
#if NETSTANDARD2_1
/// <summary>
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>A task that represents the asynchronous write operation.</returns>
public ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)
{
return stream.WriteAsync(buffer, cancellationToken);
}
#else
/// <summary>
/// Asynchronously writes a sequence of bytes to the current stream, advances the current position within this stream by the number of bytes written, and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write data from.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>A task that represents the asynchronous write operation.</returns>
public Task WriteAsync2(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)
{
var buf = ArrayPool<byte>.Shared.Rent(buffer.Length);
buffer.CopyTo(buf);
return stream.WriteAsync(buf, 0, buf.Length, cancellationToken);
}
#endif
}
}
...@@ -111,7 +111,7 @@ namespace Titanium.Web.Proxy.Http ...@@ -111,7 +111,7 @@ namespace Titanium.Web.Proxy.Http
bool useUpstreamProxy = upstreamProxy != null && Connection.IsHttps == false; bool useUpstreamProxy = upstreamProxy != null && Connection.IsHttps == false;
var writer = Connection.StreamWriter; var serverStream = Connection.Stream;
string url; string url;
if (useUpstreamProxy || isTransparent) if (useUpstreamProxy || isTransparent)
...@@ -153,7 +153,7 @@ namespace Titanium.Web.Proxy.Http ...@@ -153,7 +153,7 @@ namespace Titanium.Web.Proxy.Http
headerBuilder.WriteLine(); headerBuilder.WriteLine();
await writer.WriteHeadersAsync(headerBuilder, cancellationToken); await serverStream.WriteHeadersAsync(headerBuilder, cancellationToken);
if (enable100ContinueBehaviour && Request.ExpectContinue) if (enable100ContinueBehaviour && Request.ExpectContinue)
{ {
......
using Titanium.Web.Proxy.Helpers;
using Titanium.Web.Proxy.Network.Tcp;
using Titanium.Web.Proxy.StreamExtended.Network;
namespace Titanium.Web.Proxy.Network
{
/// <summary>
/// This class wraps Tcp connection to client
/// </summary>
internal class ProxyClient
{
public ProxyClient(TcpClientConnection connection, CustomBufferedStream clientStream, HttpResponseWriter clientStreamWriter)
{
Connection = connection;
ClientStream = clientStream;
ClientStreamWriter = clientStreamWriter;
}
/// <summary>
/// TcpClient connection used to communicate with client
/// </summary>
internal TcpClientConnection Connection { get; }
/// <summary>
/// Holds the stream to client
/// </summary>
internal CustomBufferedStream ClientStream { get; }
/// <summary>
/// Used to write line by line to client
/// </summary>
internal HttpResponseWriter ClientStreamWriter { get; }
}
}
...@@ -86,7 +86,6 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -86,7 +86,6 @@ namespace Titanium.Web.Proxy.Network.Tcp
proxyServer.UpdateClientConnectionCount(false); proxyServer.UpdateClientConnectionCount(false);
tcpClient.CloseSocket(); tcpClient.CloseSocket();
}); });
} }
} }
} }
...@@ -211,7 +211,7 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -211,7 +211,7 @@ namespace Titanium.Web.Proxy.Network.Tcp
ProxyServer proxyServer, SessionEventArgsBase? session, IPEndPoint? upStreamEndPoint, IExternalProxy? externalProxy, ProxyServer proxyServer, SessionEventArgsBase? session, IPEndPoint? upStreamEndPoint, IExternalProxy? externalProxy,
bool noCache, CancellationToken cancellationToken) bool noCache, CancellationToken cancellationToken)
{ {
var sslProtocol = session?.ProxyClient.Connection.SslProtocol ?? SslProtocols.None; var sslProtocol = session?.ClientConnection.SslProtocol ?? SslProtocols.None;
var cacheKey = GetConnectionCacheKey(remoteHostName, remotePort, var cacheKey = GetConnectionCacheKey(remoteHostName, remotePort,
isHttps, applicationProtocols, upStreamEndPoint, externalProxy); isHttps, applicationProtocols, upStreamEndPoint, externalProxy);
...@@ -297,7 +297,7 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -297,7 +297,7 @@ namespace Titanium.Web.Proxy.Network.Tcp
} }
TcpClient? tcpClient = null; TcpClient? tcpClient = null;
CustomBufferedStream? stream = null; HttpServerStream? stream = null;
SslApplicationProtocol negotiatedApplicationProtocol = default; SslApplicationProtocol negotiatedApplicationProtocol = default;
...@@ -323,6 +323,7 @@ retry: ...@@ -323,6 +323,7 @@ retry:
Array.Sort(ipAddresses, (x, y) => x.AddressFamily.CompareTo(y.AddressFamily)); Array.Sort(ipAddresses, (x, y) => x.AddressFamily.CompareTo(y.AddressFamily));
Exception lastException = null;
for (int i = 0; i < ipAddresses.Length; i++) for (int i = 0; i < ipAddresses.Length; i++)
{ {
try try
...@@ -352,28 +353,29 @@ retry: ...@@ -352,28 +353,29 @@ retry:
} }
catch (Exception e) catch (Exception e)
{ {
if (i == ipAddresses.Length - 1)
{
throw new Exception($"Could not establish connection to {hostname}", e);
}
// dispose the current TcpClient and try the next address // dispose the current TcpClient and try the next address
lastException = e;
tcpClient?.Dispose(); tcpClient?.Dispose();
tcpClient = null;
} }
} }
if (tcpClient == null)
{
throw new Exception($"Could not establish connection to {hostname}", lastException);
}
if (session != null) if (session != null)
{ {
session.TimeLine["Connection Established"] = DateTime.Now; session.TimeLine["Connection Established"] = DateTime.Now;
} }
await proxyServer.InvokeConnectionCreateEvent(tcpClient!, false); await proxyServer.InvokeConnectionCreateEvent(tcpClient, false);
stream = new CustomBufferedStream(tcpClient!.GetStream(), proxyServer.BufferPool); stream = new HttpServerStream(tcpClient.GetStream(), proxyServer.BufferPool);
if (useUpstreamProxy && (isConnect || isHttps)) if (useUpstreamProxy && (isConnect || isHttps))
{ {
var writer = new HttpRequestWriter(stream, proxyServer.BufferPool);
string authority = $"{remoteHostName}:{remotePort}"; string authority = $"{remoteHostName}:{remotePort}";
var connectRequest = new ConnectRequest(authority) var connectRequest = new ConnectRequest(authority)
{ {
...@@ -391,7 +393,7 @@ retry: ...@@ -391,7 +393,7 @@ retry:
HttpHeader.GetProxyAuthorizationHeader(externalProxy.UserName, externalProxy.Password)); HttpHeader.GetProxyAuthorizationHeader(externalProxy.UserName, externalProxy.Password));
} }
await writer.WriteRequestAsync(connectRequest, cancellationToken: cancellationToken); await stream.WriteRequestAsync(connectRequest, cancellationToken: cancellationToken);
string httpStatus = await stream.ReadLineAsync(cancellationToken) string httpStatus = await stream.ReadLineAsync(cancellationToken)
?? throw new ServerConnectionException("Server connection was closed."); ?? throw new ServerConnectionException("Server connection was closed.");
...@@ -411,7 +413,7 @@ retry: ...@@ -411,7 +413,7 @@ retry:
{ {
var sslStream = new SslStream(stream, false, proxyServer.ValidateServerCertificate, var sslStream = new SslStream(stream, false, proxyServer.ValidateServerCertificate,
proxyServer.SelectClientCertificate); proxyServer.SelectClientCertificate);
stream = new CustomBufferedStream(sslStream, proxyServer.BufferPool); stream = new HttpServerStream(sslStream, proxyServer.BufferPool);
var options = new SslClientAuthenticationOptions var options = new SslClientAuthenticationOptions
{ {
...@@ -435,6 +437,9 @@ retry: ...@@ -435,6 +437,9 @@ retry:
} }
catch (IOException ex) when (ex.HResult == unchecked((int)0x80131620) && retry && enabledSslProtocols >= SslProtocols.Tls11) catch (IOException ex) when (ex.HResult == unchecked((int)0x80131620) && retry && enabledSslProtocols >= SslProtocols.Tls11)
{ {
stream?.Dispose();
tcpClient?.Close();
enabledSslProtocols = SslProtocols.Tls; enabledSslProtocols = SslProtocols.Tls;
retry = false; retry = false;
goto retry; goto retry;
......
...@@ -15,15 +15,14 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -15,15 +15,14 @@ namespace Titanium.Web.Proxy.Network.Tcp
/// </summary> /// </summary>
internal class TcpServerConnection : IDisposable internal class TcpServerConnection : IDisposable
{ {
internal TcpServerConnection(ProxyServer proxyServer, TcpClient tcpClient, CustomBufferedStream stream, internal TcpServerConnection(ProxyServer proxyServer, TcpClient tcpClient, HttpServerStream stream,
string hostName, int port, bool isHttps, SslApplicationProtocol negotiatedApplicationProtocol, string hostName, int port, bool isHttps, SslApplicationProtocol negotiatedApplicationProtocol,
Version version, bool useUpstreamProxy, IExternalProxy? upStreamProxy, IPEndPoint? upStreamEndPoint, string cacheKey) Version version, bool useUpstreamProxy, IExternalProxy? upStreamProxy, IPEndPoint? upStreamEndPoint, string cacheKey)
{ {
this.tcpClient = tcpClient; TcpClient = tcpClient;
LastAccess = DateTime.Now; LastAccess = DateTime.Now;
this.proxyServer = proxyServer; this.proxyServer = proxyServer;
this.proxyServer.UpdateServerConnectionCount(true); this.proxyServer.UpdateServerConnectionCount(true);
StreamWriter = new HttpRequestWriter(stream, proxyServer.BufferPool);
Stream = stream; Stream = stream;
HostName = hostName; HostName = hostName;
Port = port; Port = port;
...@@ -63,22 +62,15 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -63,22 +62,15 @@ namespace Titanium.Web.Proxy.Network.Tcp
/// </summary> /// </summary>
internal Version Version { get; set; } = HttpHeader.VersionUnknown; internal Version Version { get; set; } = HttpHeader.VersionUnknown;
private readonly TcpClient tcpClient;
/// <summary> /// <summary>
/// The TcpClient. /// The TcpClient.
/// </summary> /// </summary>
internal TcpClient TcpClient => tcpClient; internal TcpClient TcpClient { get; }
/// <summary> /// <summary>
/// Used to write lines to server /// Used to write lines to server
/// </summary> /// </summary>
internal HttpRequestWriter StreamWriter { get; } internal HttpServerStream Stream { get; }
/// <summary>
/// Server stream
/// </summary>
internal CustomBufferedStream Stream { get; }
/// <summary> /// <summary>
/// Last time this connection was used /// Last time this connection was used
...@@ -107,8 +99,8 @@ namespace Titanium.Web.Proxy.Network.Tcp ...@@ -107,8 +99,8 @@ namespace Titanium.Web.Proxy.Network.Tcp
// This way we can push tcp Time_Wait to server side when possible. // This way we can push tcp Time_Wait to server side when possible.
await Task.Delay(1000); await Task.Delay(1000);
proxyServer.UpdateServerConnectionCount(false); proxyServer.UpdateServerConnectionCount(false);
Stream?.Dispose(); Stream.BaseStream?.Dispose();
tcpClient.CloseSocket(); TcpClient.CloseSocket();
}); });
} }
......
...@@ -803,7 +803,7 @@ namespace Titanium.Web.Proxy ...@@ -803,7 +803,7 @@ namespace Titanium.Web.Proxy
/// </summary> /// </summary>
/// <param name="clientStream">The client stream.</param> /// <param name="clientStream">The client stream.</param>
/// <param name="exception">The exception.</param> /// <param name="exception">The exception.</param>
private void onException(CustomBufferedStream clientStream, Exception exception) private void onException(HttpClientStream clientStream, Exception exception)
{ {
ExceptionFunc(exception); ExceptionFunc(exception);
} }
......
...@@ -31,13 +31,11 @@ namespace Titanium.Web.Proxy ...@@ -31,13 +31,11 @@ namespace Titanium.Web.Proxy
/// <param name="endPoint">The proxy endpoint.</param> /// <param name="endPoint">The proxy endpoint.</param>
/// <param name="clientConnection">The client connection.</param> /// <param name="clientConnection">The client connection.</param>
/// <param name="clientStream">The client stream.</param> /// <param name="clientStream">The client stream.</param>
/// <param name="clientStreamWriter">The client stream writer.</param>
/// <param name="cancellationTokenSource">The cancellation token source for this async task.</param> /// <param name="cancellationTokenSource">The cancellation token source for this async task.</param>
/// <param name="connectArgs">The Connect request if this is a HTTPS request from explicit endpoint.</param> /// <param name="connectArgs">The Connect request if this is a HTTPS request from explicit endpoint.</param>
/// <param name="prefetchConnectionTask">Prefetched server connection for current client using Connect/SNI headers.</param> /// <param name="prefetchConnectionTask">Prefetched server connection for current client using Connect/SNI headers.</param>
private async Task handleHttpSessionRequest(ProxyEndPoint endPoint, TcpClientConnection clientConnection, private async Task handleHttpSessionRequest(ProxyEndPoint endPoint, TcpClientConnection clientConnection,
CustomBufferedStream clientStream, HttpResponseWriter clientStreamWriter, HttpClientStream clientStream, CancellationTokenSource cancellationTokenSource, TunnelConnectSessionEventArgs? connectArgs = null,
CancellationTokenSource cancellationTokenSource, TunnelConnectSessionEventArgs? connectArgs = null,
Task<TcpServerConnection>? prefetchConnectionTask = null) Task<TcpServerConnection>? prefetchConnectionTask = null)
{ {
var connectRequest = connectArgs?.HttpClient.ConnectRequest; var connectRequest = connectArgs?.HttpClient.ConnectRequest;
...@@ -66,7 +64,7 @@ namespace Titanium.Web.Proxy ...@@ -66,7 +64,7 @@ namespace Titanium.Web.Proxy
return; return;
} }
var args = new SessionEventArgs(this, endPoint, new ProxyClient(clientConnection, clientStream, clientStreamWriter), connectRequest, cancellationTokenSource) var args = new SessionEventArgs(this, endPoint, clientConnection, clientStream, connectRequest, cancellationTokenSource)
{ {
UserData = connectArgs?.UserData UserData = connectArgs?.UserData
}; };
...@@ -101,7 +99,7 @@ namespace Titanium.Web.Proxy ...@@ -101,7 +99,7 @@ namespace Titanium.Web.Proxy
await onBeforeResponse(args); await onBeforeResponse(args);
// send the response // send the response
await clientStreamWriter.WriteResponseAsync(args.HttpClient.Response, await clientStream.WriteResponseAsync(args.HttpClient.Response,
cancellationToken: cancellationToken); cancellationToken: cancellationToken);
return; return;
} }
...@@ -276,7 +274,7 @@ namespace Titanium.Web.Proxy ...@@ -276,7 +274,7 @@ namespace Titanium.Web.Proxy
// if upgrading to websocket then relay the request without reading the contents // if upgrading to websocket then relay the request without reading the contents
await handleWebSocketUpgrade(requestHttpMethod, requestHttpUrl, requestVersion, args, args.HttpClient.Request, await handleWebSocketUpgrade(requestHttpMethod, requestHttpUrl, requestVersion, args, args.HttpClient.Request,
args.HttpClient.Response, args.ProxyClient.ClientStream, args.ProxyClient.ClientStreamWriter, args.HttpClient.Response, args.ClientStream,
connection, cancellationTokenSource, cancellationToken); connection, cancellationTokenSource, cancellationToken);
return false; return false;
} }
...@@ -304,13 +302,13 @@ namespace Titanium.Web.Proxy ...@@ -304,13 +302,13 @@ namespace Titanium.Web.Proxy
// If a successful 100 continue request was made, inform that to the client and reset response // If a successful 100 continue request was made, inform that to the client and reset response
if (request.ExpectationSucceeded) if (request.ExpectationSucceeded)
{ {
var clientStreamWriter = args.ProxyClient.ClientStreamWriter; var writer = args.ClientStream;
var response = args.HttpClient.Response; var response = args.HttpClient.Response;
var headerBuilder = new HeaderBuilder(); var headerBuilder = new HeaderBuilder();
headerBuilder.WriteResponseLine(response.HttpVersion, response.StatusCode, response.StatusDescription); headerBuilder.WriteResponseLine(response.HttpVersion, response.StatusCode, response.StatusDescription);
headerBuilder.WriteHeaders(response.Headers); headerBuilder.WriteHeaders(response.Headers);
await clientStreamWriter.WriteHeadersAsync(headerBuilder, cancellationToken); await writer.WriteHeadersAsync(headerBuilder, cancellationToken);
await args.ClearResponse(cancellationToken); await args.ClearResponse(cancellationToken);
} }
...@@ -320,14 +318,12 @@ namespace Titanium.Web.Proxy ...@@ -320,14 +318,12 @@ namespace Titanium.Web.Proxy
{ {
if (request.IsBodyRead) if (request.IsBodyRead)
{ {
var writer = args.HttpClient.Connection.StreamWriter; await args.HttpClient.Connection.Stream.WriteBodyAsync(body!, request.IsChunked, cancellationToken);
await writer.WriteBodyAsync(body!, request.IsChunked, cancellationToken);
} }
else if (!request.ExpectationFailed) else if (!request.ExpectationFailed)
{ {
// get the request body unless an unsuccessful 100 continue request was made // get the request body unless an unsuccessful 100 continue request was made
HttpWriter writer = args.HttpClient.Connection.StreamWriter!; await args.CopyRequestBodyAsync(args.HttpClient.Connection.Stream, TransformationMode.None, cancellationToken);
await args.CopyRequestBodyAsync(writer, TransformationMode.None, cancellationToken);
} }
} }
......
...@@ -64,13 +64,13 @@ namespace Titanium.Web.Proxy ...@@ -64,13 +64,13 @@ namespace Titanium.Web.Proxy
// it may changed in the user event // it may changed in the user event
response = args.HttpClient.Response; response = args.HttpClient.Response;
var clientStreamWriter = args.ProxyClient.ClientStreamWriter; var clientStream = args.ClientStream;
// user set custom response by ignoring original response from server. // user set custom response by ignoring original response from server.
if (response.Locked) if (response.Locked)
{ {
// write custom user response with body and return. // write custom user response with body and return.
await clientStreamWriter.WriteResponseAsync(response, cancellationToken: cancellationToken); await clientStream.WriteResponseAsync(response, cancellationToken: cancellationToken);
if (args.HttpClient.HasConnection && !args.HttpClient.CloseServerConnection) if (args.HttpClient.HasConnection && !args.HttpClient.CloseServerConnection)
{ {
...@@ -108,7 +108,7 @@ namespace Titanium.Web.Proxy ...@@ -108,7 +108,7 @@ namespace Titanium.Web.Proxy
if (response.IsBodyRead) if (response.IsBodyRead)
{ {
await clientStreamWriter.WriteResponseAsync(response, cancellationToken: cancellationToken); await clientStream.WriteResponseAsync(response, cancellationToken: cancellationToken);
} }
else else
{ {
...@@ -116,12 +116,12 @@ namespace Titanium.Web.Proxy ...@@ -116,12 +116,12 @@ namespace Titanium.Web.Proxy
var headerBuilder = new HeaderBuilder(); var headerBuilder = new HeaderBuilder();
headerBuilder.WriteResponseLine(response.HttpVersion, response.StatusCode, response.StatusDescription); headerBuilder.WriteResponseLine(response.HttpVersion, response.StatusCode, response.StatusDescription);
headerBuilder.WriteHeaders(response.Headers); headerBuilder.WriteHeaders(response.Headers);
await clientStreamWriter.WriteHeadersAsync(headerBuilder, cancellationToken); await clientStream.WriteHeadersAsync(headerBuilder, cancellationToken);
// Write body if exists // Write body if exists
if (response.HasBody) if (response.HasBody)
{ {
await args.CopyResponseBodyAsync(clientStreamWriter, TransformationMode.None, await args.CopyResponseBodyAsync(clientStream, TransformationMode.None,
cancellationToken); cancellationToken);
} }
} }
......
using System; using System;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Titanium.Web.Proxy.Helpers;
using Titanium.Web.Proxy.StreamExtended.BufferPool; using Titanium.Web.Proxy.StreamExtended.BufferPool;
namespace Titanium.Web.Proxy.StreamExtended.Network namespace Titanium.Web.Proxy.StreamExtended.Network
...@@ -11,9 +12,9 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -11,9 +12,9 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// </summary> /// </summary>
internal class CopyStream : ILineStream, IDisposable internal class CopyStream : ILineStream, IDisposable
{ {
private readonly CustomBufferedStream reader; private readonly IHttpStreamReader reader;
private readonly ICustomStreamWriter writer; private readonly IHttpStreamWriter writer;
private readonly IBufferPool bufferPool; private readonly IBufferPool bufferPool;
...@@ -27,7 +28,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -27,7 +28,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
public long ReadBytes { get; private set; } public long ReadBytes { get; private set; }
public CopyStream(CustomBufferedStream reader, ICustomStreamWriter writer, IBufferPool bufferPool) public CopyStream(IHttpStreamReader reader, IHttpStreamWriter writer, IBufferPool bufferPool)
{ {
this.reader = reader; this.reader = reader;
this.writer = writer; this.writer = writer;
...@@ -61,7 +62,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -61,7 +62,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
public ValueTask<string?> ReadLineAsync(CancellationToken cancellationToken = default) public ValueTask<string?> ReadLineAsync(CancellationToken cancellationToken = default)
{ {
return CustomBufferedStream.ReadLineInternalAsync(this, bufferPool, cancellationToken); return HttpStream.ReadLineInternalAsync(this, bufferPool, cancellationToken);
} }
public void Dispose() public void Dispose()
......
using System.Threading;
using System.Threading.Tasks;
namespace Titanium.Web.Proxy.StreamExtended.Network
{
public interface IHttpStreamReader : ILineStream
{
int Read(byte[] buffer, int offset, int count);
Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken);
}
}
using System.Threading; using System;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace Titanium.Web.Proxy.StreamExtended.Network namespace Titanium.Web.Proxy.StreamExtended.Network
...@@ -6,10 +7,13 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -6,10 +7,13 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <summary> /// <summary>
/// A concrete implementation of this interface is required when calling CopyStream. /// A concrete implementation of this interface is required when calling CopyStream.
/// </summary> /// </summary>
public interface ICustomStreamWriter internal interface IHttpStreamWriter
{ {
void Write(byte[] buffer, int i, int bufferLength); void Write(byte[] buffer, int i, int bufferLength);
Task WriteAsync(byte[] buffer, int i, int bufferLength, CancellationToken cancellationToken); Task WriteAsync(byte[] buffer, int i, int bufferLength, CancellationToken cancellationToken);
Task CopyBodyAsync(IHttpStreamReader streamReader, bool isChunked, long contentLength,
Action<byte[], int, int>? onCopy, CancellationToken cancellationToken);
} }
} }
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Titanium.Web.Proxy.Helpers;
using Titanium.Web.Proxy.StreamExtended.BufferPool; using Titanium.Web.Proxy.StreamExtended.BufferPool;
using Titanium.Web.Proxy.StreamExtended.Models; using Titanium.Web.Proxy.StreamExtended.Models;
using Titanium.Web.Proxy.StreamExtended.Network; using Titanium.Web.Proxy.StreamExtended.Network;
...@@ -19,7 +20,7 @@ namespace Titanium.Web.Proxy.StreamExtended ...@@ -19,7 +20,7 @@ namespace Titanium.Web.Proxy.StreamExtended
/// <param name="bufferPool"></param> /// <param name="bufferPool"></param>
/// <param name="cancellationToken"></param> /// <param name="cancellationToken"></param>
/// <returns></returns> /// <returns></returns>
public static async Task<ClientHelloInfo?> PeekClientHello(CustomBufferedStream clientStream, IBufferPool bufferPool, CancellationToken cancellationToken = default) public static async Task<ClientHelloInfo?> PeekClientHello(IPeekStream clientStream, IBufferPool bufferPool, CancellationToken cancellationToken = default)
{ {
// detects the HTTPS ClientHello message as it is described in the following url: // detects the HTTPS ClientHello message as it is described in the following url:
// https://stackoverflow.com/questions/3897883/how-to-detect-an-incoming-ssl-https-handshake-ssl-wire-format // https://stackoverflow.com/questions/3897883/how-to-detect-an-incoming-ssl-https-handshake-ssl-wire-format
...@@ -177,7 +178,7 @@ namespace Titanium.Web.Proxy.StreamExtended ...@@ -177,7 +178,7 @@ namespace Titanium.Web.Proxy.StreamExtended
/// <param name="bufferPool"></param> /// <param name="bufferPool"></param>
/// <param name="cancellationToken"></param> /// <param name="cancellationToken"></param>
/// <returns></returns> /// <returns></returns>
public static async Task<bool> IsServerHello(CustomBufferedStream stream, IBufferPool bufferPool, CancellationToken cancellationToken) public static async Task<bool> IsServerHello(IPeekStream stream, IBufferPool bufferPool, CancellationToken cancellationToken)
{ {
var serverHello = await PeekServerHello(stream, bufferPool, cancellationToken); var serverHello = await PeekServerHello(stream, bufferPool, cancellationToken);
return serverHello != null; return serverHello != null;
...@@ -189,7 +190,7 @@ namespace Titanium.Web.Proxy.StreamExtended ...@@ -189,7 +190,7 @@ namespace Titanium.Web.Proxy.StreamExtended
/// <param name="bufferPool"></param> /// <param name="bufferPool"></param>
/// <param name="cancellationToken"></param> /// <param name="cancellationToken"></param>
/// <returns></returns> /// <returns></returns>
public static async Task<ServerHelloInfo?> PeekServerHello(CustomBufferedStream serverStream, IBufferPool bufferPool, CancellationToken cancellationToken = default) public static async Task<ServerHelloInfo?> PeekServerHello(IPeekStream serverStream, IBufferPool bufferPool, CancellationToken cancellationToken = default)
{ {
// detects the HTTPS ClientHello message as it is described in the following url: // detects the HTTPS ClientHello message as it is described in the following url:
// https://stackoverflow.com/questions/3897883/how-to-detect-an-incoming-ssl-https-handshake-ssl-wire-format // https://stackoverflow.com/questions/3897883/how-to-detect-an-incoming-ssl-https-handshake-ssl-wire-format
......
...@@ -33,8 +33,7 @@ namespace Titanium.Web.Proxy ...@@ -33,8 +33,7 @@ namespace Titanium.Web.Proxy
var cancellationTokenSource = new CancellationTokenSource(); var cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = cancellationTokenSource.Token; var cancellationToken = cancellationTokenSource.Token;
var clientStream = new CustomBufferedStream(clientConnection.GetStream(), BufferPool); var clientStream = new HttpClientStream(clientConnection.GetStream(), BufferPool);
var clientStreamWriter = new HttpResponseWriter(clientStream, BufferPool);
SslStream? sslStream = null; SslStream? sslStream = null;
...@@ -75,14 +74,12 @@ namespace Titanium.Web.Proxy ...@@ -75,14 +74,12 @@ namespace Titanium.Web.Proxy
await sslStream.AuthenticateAsServerAsync(certificate, false, SslProtocols.Tls, false); await sslStream.AuthenticateAsServerAsync(certificate, false, SslProtocols.Tls, false);
// HTTPS server created - we can now decrypt the client's traffic // HTTPS server created - we can now decrypt the client's traffic
clientStream = new CustomBufferedStream(sslStream, BufferPool); clientStream = new HttpClientStream(sslStream, BufferPool);
clientStreamWriter = new HttpResponseWriter(clientStream, BufferPool);
} }
catch (Exception e) catch (Exception e)
{ {
var certname = certificate?.GetNameInfo(X509NameType.SimpleName, false); var certname = certificate?.GetNameInfo(X509NameType.SimpleName, false);
var session = new SessionEventArgs(this, endPoint, new ProxyClient(clientConnection, clientStream, clientStreamWriter), null, var session = new SessionEventArgs(this, endPoint, clientConnection, clientStream, null,
cancellationTokenSource); cancellationTokenSource);
throw new ProxyConnectException( throw new ProxyConnectException(
$"Couldn't authenticate host '{httpsHostName}' with certificate '{certname}'.", e, session); $"Couldn't authenticate host '{httpsHostName}' with certificate '{certname}'.", e, session);
...@@ -108,7 +105,7 @@ namespace Titanium.Web.Proxy ...@@ -108,7 +105,7 @@ namespace Titanium.Web.Proxy
{ {
// clientStream.Available should be at most BufferSize because it is using the same buffer size // clientStream.Available should be at most BufferSize because it is using the same buffer size
await clientStream.ReadAsync(data, 0, available, cancellationToken); await clientStream.ReadAsync(data, 0, available, cancellationToken);
await connection.StreamWriter.WriteAsync(data, 0, available, true, cancellationToken); await connection.Stream.WriteAsync(data, 0, available, true, cancellationToken);
} }
finally finally
{ {
...@@ -133,7 +130,7 @@ namespace Titanium.Web.Proxy ...@@ -133,7 +130,7 @@ namespace Titanium.Web.Proxy
// HTTPS server created - we can now decrypt the client's traffic // HTTPS server created - we can now decrypt the client's traffic
// Now create the request // Now create the request
await handleHttpSessionRequest(endPoint, clientConnection, clientStream, clientStreamWriter, cancellationTokenSource); await handleHttpSessionRequest(endPoint, clientConnection, clientStream, cancellationTokenSource);
} }
catch (ProxyException e) catch (ProxyException e)
{ {
......
...@@ -18,15 +18,14 @@ namespace Titanium.Web.Proxy ...@@ -18,15 +18,14 @@ namespace Titanium.Web.Proxy
/// </summary> /// </summary>
private async Task handleWebSocketUpgrade(string requestHttpMethod, string requestHttpUrl, Version requestVersion, private async Task handleWebSocketUpgrade(string requestHttpMethod, string requestHttpUrl, Version requestVersion,
SessionEventArgs args, Request request, Response response, SessionEventArgs args, Request request, Response response,
CustomBufferedStream clientStream, HttpResponseWriter clientStreamWriter, HttpClientStream clientStream, TcpServerConnection serverConnection,
TcpServerConnection serverConnection,
CancellationTokenSource cancellationTokenSource, CancellationToken cancellationToken) CancellationTokenSource cancellationTokenSource, CancellationToken cancellationToken)
{ {
// prepare the prefix content // prepare the prefix content
var headerBuilder = new HeaderBuilder(); var headerBuilder = new HeaderBuilder();
headerBuilder.WriteRequestLine(requestHttpMethod, requestHttpUrl, requestVersion); headerBuilder.WriteRequestLine(requestHttpMethod, requestHttpUrl, requestVersion);
headerBuilder.WriteHeaders(request.Headers); headerBuilder.WriteHeaders(request.Headers);
await serverConnection.StreamWriter.WriteHeadersAsync(headerBuilder, cancellationToken); await serverConnection.Stream.WriteHeadersAsync(headerBuilder, cancellationToken);
string httpStatus; string httpStatus;
try try
...@@ -51,7 +50,7 @@ namespace Titanium.Web.Proxy ...@@ -51,7 +50,7 @@ namespace Titanium.Web.Proxy
if (!args.IsTransparent) if (!args.IsTransparent)
{ {
await clientStreamWriter.WriteResponseAsync(response, await clientStream.WriteResponseAsync(response,
cancellationToken: cancellationToken); cancellationToken: cancellationToken);
} }
......
...@@ -175,7 +175,7 @@ namespace Titanium.Web.Proxy ...@@ -175,7 +175,7 @@ namespace Titanium.Web.Proxy
// Add custom div to body to clarify that the proxy (not the client browser) failed authentication // Add custom div to body to clarify that the proxy (not the client browser) failed authentication
string authErrorMessage = string authErrorMessage =
"<div class=\"inserted-by-proxy\"><h2>NTLM authentication through Titanium.Web.Proxy (" + "<div class=\"inserted-by-proxy\"><h2>NTLM authentication through Titanium.Web.Proxy (" +
args.ProxyClient.Connection.LocalEndPoint + args.ClientConnection.LocalEndPoint +
") failed. Please check credentials.</h2></div>"; ") failed. Please check credentials.</h2></div>";
string originalErrorMessage = string originalErrorMessage =
"<div class=\"inserted-by-proxy\"><h3>Response from remote web server below.</h3></div><br/>"; "<div class=\"inserted-by-proxy\"><h3>Response from remote web server below.</h3></div><br/>";
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment