From a9a6dab73bf422b2a8a7f7b8beb5920284a7e812 Mon Sep 17 00:00:00 2001 From: js6pak Date: Fri, 2 Apr 2021 20:28:06 +0200 Subject: [PATCH] Bring back async changes --- Hazel.UnitTests/TestHelper.cs | 15 +- Hazel.UnitTests/UdpConnectionTestHarness.cs | 25 +- Hazel.UnitTests/UnityUdpConnectionTests.cs | 8 +- Hazel/Connection.cs | 50 +- Hazel/ConnectionListener.cs | 39 +- Hazel/DataReceivedEventArgs.cs | 12 +- Hazel/Dtls/ConnectionId.cs | 45 + Hazel/Dtls/DtlsConnectionListener.cs | 391 +++--- Hazel/Dtls/DtlsUnityConnection.cs | 1077 ----------------- Hazel/Extensions/ServiceProviderExtensions.cs | 21 + Hazel/FewerThreads/HazelThreadPool.cs | 39 - .../ThreadLimitedUdpConnectionListener.cs | 404 ------- .../ThreadLimitedUdpServerConnection.cs | 106 -- Hazel/MessageReader.cs | 353 ++---- Hazel/MessageReaderPolicy.cs | 27 + Hazel/MessageWriter.cs | 118 +- Hazel/NetworkConnection.cs | 23 +- Hazel/NewConnectionEventArgs.cs | 8 +- Hazel/{ObjectPool.cs => ObjectPoolCustom.cs} | 4 +- Hazel/SendOption.cs | 32 - Hazel/Udp/UdpClientConnection.cs | 276 ++--- Hazel/Udp/UdpConnection.KeepAlive.cs | 17 +- Hazel/Udp/UdpConnection.Reliable.cs | 47 +- Hazel/Udp/UdpConnection.cs | 188 ++- Hazel/Udp/UdpConnectionListener.cs | 341 ++---- Hazel/Udp/UdpConnectionRateLimit.cs | 75 ++ Hazel/Udp/UdpServerConnection.cs | 32 +- Hazel/Udp/UnityUdpClientConnection.cs | 325 ----- 28 files changed, 1051 insertions(+), 3047 deletions(-) create mode 100644 Hazel/Dtls/ConnectionId.cs delete mode 100644 Hazel/Dtls/DtlsUnityConnection.cs create mode 100644 Hazel/Extensions/ServiceProviderExtensions.cs delete mode 100644 Hazel/FewerThreads/HazelThreadPool.cs delete mode 100644 Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs delete mode 100644 Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs create mode 100644 Hazel/MessageReaderPolicy.cs rename Hazel/{ObjectPool.cs => ObjectPoolCustom.cs} (96%) delete mode 100644 Hazel/SendOption.cs create mode 100644 Hazel/Udp/UdpConnectionRateLimit.cs delete mode 100644 Hazel/Udp/UnityUdpClientConnection.cs diff --git a/Hazel.UnitTests/TestHelper.cs b/Hazel.UnitTests/TestHelper.cs index df22a19..0206a65 100644 --- a/Hazel.UnitTests/TestHelper.cs +++ b/Hazel.UnitTests/TestHelper.cs @@ -5,6 +5,7 @@ using Hazel; using System.Net; using System.Threading; using System.Diagnostics; +using Impostor.Api.Net.Messages; using Impostor.Hazel; using Impostor.Hazel.FewerThreads; using DataReceivedEventArgs = Impostor.Hazel.DataReceivedEventArgs; @@ -19,7 +20,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunServerToClientTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunServerToClientTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, MessageType sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); @@ -61,7 +62,7 @@ namespace Hazel.UnitTests Assert.AreEqual(data[i], args.Value.Message.ReadByte()); } - Assert.AreEqual(sendOption, args.Value.SendOption); + Assert.AreEqual(sendOption, args.Value.Type); } /// @@ -69,7 +70,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunServerToClientTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunServerToClientTest(NetworkConnectionListener listener, Connection connection, int dataSize, MessageType sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); @@ -111,7 +112,7 @@ namespace Hazel.UnitTests Assert.AreEqual(data[i], args.Value.Message.ReadByte()); } - Assert.AreEqual(sendOption, args.Value.SendOption); + Assert.AreEqual(sendOption, args.Value.Type); } /// @@ -119,7 +120,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientToServerTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunClientToServerTest(NetworkConnectionListener listener, Connection connection, int dataSize, MessageType sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); @@ -161,7 +162,7 @@ namespace Hazel.UnitTests Assert.AreEqual(data[i], result.Value.Message.ReadByte()); } - Assert.AreEqual(sendOption, result.Value.SendOption); + Assert.AreEqual(sendOption, result.Value.Type); } @@ -170,7 +171,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientToServerTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunClientToServerTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, MessageType sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); diff --git a/Hazel.UnitTests/UdpConnectionTestHarness.cs b/Hazel.UnitTests/UdpConnectionTestHarness.cs index b1187c2..3350b80 100644 --- a/Hazel.UnitTests/UdpConnectionTestHarness.cs +++ b/Hazel.UnitTests/UdpConnectionTestHarness.cs @@ -1,10 +1,9 @@ using System; using System.Collections.Generic; -using System.Linq; -using System.Text; using System.Threading.Tasks; using Impostor.Hazel; using Impostor.Hazel.Udp; +using Microsoft.Extensions.ObjectPool; namespace Hazel.UnitTests { @@ -13,35 +12,36 @@ namespace Hazel.UnitTests public List BytesSent = new List(); public ushort ReliableReceiveLast => this.reliableReceiveLast; - - public override void Connect(byte[] bytes = null, int timeout = 5000) + public UdpConnectionTestHarness(ConnectionListener listener, ObjectPool readerPool) : base(listener, readerPool) { - this.State = ConnectionState.Connected; } - public override void ConnectAsync(byte[] bytes = null) + public override ValueTask ConnectAsync(byte[] bytes = null, int timeout = 5000) { this.State = ConnectionState.Connected; + return default; } - protected override bool SendDisconnect(MessageWriter writer) + protected override ValueTask SendDisconnect(MessageWriter writer) { lock (this) { if (this.State != ConnectionState.Connected) { - return false; + return ValueTask.FromResult(false); } this.State = ConnectionState.NotConnected; } - return true; + return ValueTask.FromResult(true); } - protected override void WriteBytesToConnection(byte[] bytes, int length) + protected override ValueTask WriteBytesToConnection(byte[] bytes, int length) { - this.BytesSent.Add(MessageReader.Get(bytes)); + var data = _readerPool.Get(); + data.Update(bytes); + this.BytesSent.Add(data); } public void Test_Receive(MessageWriter msg) @@ -49,7 +49,8 @@ namespace Hazel.UnitTests byte[] buffer = new byte[msg.Length]; Buffer.BlockCopy(msg.Buffer, 0, buffer, 0, msg.Length); - var data = MessageReader.Get(buffer); + var data = _readerPool.Get(); + data.Update(buffer); this.HandleReceive(data, data.Length); } } diff --git a/Hazel.UnitTests/UnityUdpConnectionTests.cs b/Hazel.UnitTests/UnityUdpConnectionTests.cs index 25733b6..6b5127f 100644 --- a/Hazel.UnitTests/UnityUdpConnectionTests.cs +++ b/Hazel.UnitTests/UnityUdpConnectionTests.cs @@ -444,10 +444,10 @@ namespace Hazel.UnitTests /// Tests disconnection from the server. /// [TestMethod] - public void ServerExtraDataDisconnectTest() + public async Task ServerExtraDataDisconnectTest() { using (UdpConnectionListener listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296))) - using (UdpConnection connection = new UnityUdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296))) + using (UdpConnection connection = new UdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296))) { MessageReader received = null; ManualResetEvent mutex = new ManualResetEvent(false); @@ -465,9 +465,9 @@ namespace Hazel.UnitTests args.Connection.Disconnect("Testing", writer); }; - listener.Start(); + await listener.StartAsync(); - connection.Connect(); + await connection.ConnectAsync(); mutex.WaitOne(); diff --git a/Hazel/Connection.cs b/Hazel/Connection.cs index 6e6bd61..4d99a2b 100644 --- a/Hazel/Connection.cs +++ b/Hazel/Connection.cs @@ -1,5 +1,8 @@ using System; using System.Net; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; +using Serilog; namespace Impostor.Hazel { @@ -29,6 +32,8 @@ namespace Impostor.Hazel /// public abstract class Connection : IDisposable { + private static readonly ILogger Logger = Log.ForContext(); + /// /// Called when a message has been received. /// @@ -43,7 +48,7 @@ namespace Impostor.Hazel /// /// /// - public event Action DataReceived; + public Func DataReceived; public int TestLagMs = -1; public int TestDropRate = 0; @@ -63,7 +68,7 @@ namespace Impostor.Hazel /// /// /// - public event EventHandler Disconnected; + public Func Disconnected; /// /// The remote end point of this Connection. @@ -134,7 +139,7 @@ namespace Impostor.Hazel /// general any implementer should aim to always follow the user's request. /// /// - public abstract void Send(MessageWriter msg); + public abstract ValueTask SendAsync(IMessageWriter msg); /// /// Sends a number of bytes to the end point of the connection using the specified . @@ -149,7 +154,7 @@ namespace Impostor.Hazel /// general any implementer should aim to always follow the user's request. /// /// - public abstract void SendBytes(byte[] bytes, SendOption sendOption = SendOption.None); + public abstract ValueTask SendBytes(byte[] bytes, MessageType sendOption = MessageType.Unreliable); /// /// Connects the connection to a server and begins listening. @@ -157,14 +162,7 @@ namespace Impostor.Hazel /// /// The bytes of data to send in the handshake. /// The number of milliseconds to wait before giving up on the connect attempt. - public abstract void Connect(byte[] bytes = null, int timeout = 5000); - - /// - /// Connects the connection to a server and begins listening. - /// This method does not block. - /// - /// The bytes of data to send in the handshake. - public abstract void ConnectAsync(byte[] bytes = null); + public abstract ValueTask ConnectAsync(byte[] bytes = null, int timeout = 5000); /// /// Invokes the DataReceived event. @@ -176,21 +174,21 @@ namespace Impostor.Hazel /// received. The bytes and the send option that the message was sent with should be passed in to give to the /// subscribers. /// - protected void InvokeDataReceived(MessageReader msg, SendOption sendOption) + protected async ValueTask InvokeDataReceived(MessageReader msg, MessageType messageType) { // Make a copy to avoid race condition between null check and invocation - Action handler = DataReceived; + var handler = DataReceived; if (handler != null) { try { - handler(new DataReceivedEventArgs(this, msg, sendOption)); + await handler(new DataReceivedEventArgs(this, msg, messageType)); + } + catch (Exception e) + { + Logger.Error(e, "Invoking data received failed"); + await Disconnect("Invoking data received failed"); } - catch { } - } - else - { - msg.Recycle(); } } @@ -204,19 +202,19 @@ namespace Impostor.Hazel /// by the end point or because an error occurred. If an error occurred the error should be passed in in order to /// pass to the subscribers, otherwise null can be passed in. /// - protected void InvokeDisconnected(string e, MessageReader reader) + protected async ValueTask InvokeDisconnected(string e, MessageReader reader) { // Make a copy to avoid race condition between null check and invocation - EventHandler handler = Disconnected; + var handler = Disconnected; if (handler != null) { - DisconnectedEventArgs args = new DisconnectedEventArgs(e, reader); try { - handler(this, args); + await handler(new DisconnectedEventArgs(e, reader)); } - catch + catch (Exception ex) { + Logger.Error(ex, "Error in InvokeDisconnected"); } } } @@ -225,7 +223,7 @@ namespace Impostor.Hazel /// For times when you want to force the disconnect handler to fire as well as close it. /// If you only want to close it, just use Dispose. /// - public abstract void Disconnect(string reason, MessageWriter writer = null); + public abstract ValueTask Disconnect(string reason, MessageWriter writer = null); /// /// Disposes of this NetworkConnection. diff --git a/Hazel/ConnectionListener.cs b/Hazel/ConnectionListener.cs index b83f968..0e231f9 100644 --- a/Hazel/ConnectionListener.cs +++ b/Hazel/ConnectionListener.cs @@ -1,4 +1,7 @@ using System; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; +using Serilog; namespace Impostor.Hazel { @@ -17,8 +20,10 @@ namespace Impostor.Hazel /// /// /// - public abstract class ConnectionListener : IDisposable + public abstract class ConnectionListener : IAsyncDisposable { + private static readonly ILogger Logger = Log.ForContext(); + /// /// Invoked when a new client connects. /// @@ -37,7 +42,7 @@ namespace Impostor.Hazel /// /// /// - public event Action NewConnection; + public Func NewConnection; /// /// Makes this connection listener begin listening for connections. @@ -54,7 +59,7 @@ namespace Impostor.Hazel /// /// /// - public abstract void Start(); + public abstract Task StartAsync(); /// /// Invokes the NewConnection event with the supplied connection. @@ -65,39 +70,31 @@ namespace Impostor.Hazel /// Implementers should call this to invoke the event before data is received so that /// subscribers do not miss any data that may have been sent immediately after connecting. /// - protected void InvokeNewConnection(MessageReader msg, Connection connection) + internal async Task InvokeNewConnection(IMessageReader msg, Connection connection) { // Make a copy to avoid race condition between null check and invocation - Action handler = NewConnection; + var handler = NewConnection; if (handler != null) { try { - handler(new NewConnectionEventArgs(msg, connection)); + await handler(new NewConnectionEventArgs(msg, connection)); + } + catch (Exception e) + { + Logger.Error(e, "Accepting connection failed"); + await connection.Disconnect("Accepting connection failed"); } - catch { } - } - else - { - msg.Recycle(); } } /// /// Call to dispose of the connection listener. /// - public void Dispose() - { - Dispose(true); - } - - /// - /// Called when the object is being disposed. - /// - /// Are we disposing? - protected virtual void Dispose(bool disposing) + public virtual ValueTask DisposeAsync() { this.NewConnection = null; + return ValueTask.CompletedTask; } } } diff --git a/Hazel/DataReceivedEventArgs.cs b/Hazel/DataReceivedEventArgs.cs index 7605746..6e7d880 100644 --- a/Hazel/DataReceivedEventArgs.cs +++ b/Hazel/DataReceivedEventArgs.cs @@ -1,4 +1,6 @@ -namespace Impostor.Hazel +using Impostor.Api.Net.Messages; + +namespace Impostor.Hazel { public struct DataReceivedEventArgs { @@ -7,18 +9,18 @@ /// /// The bytes received from the client. /// - public readonly MessageReader Message; + public readonly IMessageReader Message; /// /// The the data was sent with. /// - public readonly SendOption SendOption; + public readonly MessageType Type; - public DataReceivedEventArgs(Connection sender, MessageReader msg, SendOption sendOption) + public DataReceivedEventArgs(Connection sender, IMessageReader msg, MessageType type) { this.Sender = sender; this.Message = msg; - this.SendOption = sendOption; + this.Type = type; } } } diff --git a/Hazel/Dtls/ConnectionId.cs b/Hazel/Dtls/ConnectionId.cs new file mode 100644 index 0000000..5fdbb3f --- /dev/null +++ b/Hazel/Dtls/ConnectionId.cs @@ -0,0 +1,45 @@ +using System; +using System.Net; + +namespace Impostor.Hazel.Dtls +{ + public struct ConnectionId : IEquatable + { + public IPEndPoint EndPoint; + public int Serial; + + public static ConnectionId Create(IPEndPoint endPoint, int serial) + { + return new ConnectionId + { + EndPoint = endPoint, + Serial = serial, + }; + } + + public bool Equals(ConnectionId other) + { + return this.Serial == other.Serial + && this.EndPoint.Equals(other.EndPoint) + ; + } + + public override bool Equals(object obj) + { + if (obj is ConnectionId) + { + return this.Equals((ConnectionId)obj); + } + + return false; + } + + public override int GetHashCode() + { + ///NOTE(mendsley): We're only hashing the endpoint + /// here, as the common case will have one + /// connection per address+port tuple. + return this.EndPoint.GetHashCode(); + } + } +} diff --git a/Hazel/Dtls/DtlsConnectionListener.cs b/Hazel/Dtls/DtlsConnectionListener.cs index c3cd51c..c3b6cce 100644 --- a/Hazel/Dtls/DtlsConnectionListener.cs +++ b/Hazel/Dtls/DtlsConnectionListener.cs @@ -4,12 +4,15 @@ using System.Collections.Generic; using System.Diagnostics; using System.IO; using System.Net; +using System.Net.Sockets; using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; using System.Threading; +using System.Threading.Tasks; using Impostor.Hazel.Crypto; -using Impostor.Hazel.FewerThreads; -using Impostor.Hazel.UPnP; +using Impostor.Hazel.Udp; +using Microsoft.Extensions.ObjectPool; +using Serilog; namespace Impostor.Hazel.Dtls { @@ -17,8 +20,10 @@ namespace Impostor.Hazel.Dtls /// Listens for new UDP-DTLS connections and creates UdpConnections for them. /// /// - public class DtlsConnectionListener : ThreadLimitedUdpConnectionListener + public class DtlsConnectionListener : UdpConnectionListener { + private static readonly ILogger Logger = Log.ForContext(); + const int MaxDatagramSize = 1200; /// @@ -75,7 +80,6 @@ namespace Impostor.Hazel.Dtls public ByteSpan ClientVerification; public ByteSpan ServerVerification; - } /// @@ -87,7 +91,7 @@ namespace Impostor.Hazel.Dtls public bool CanHandleApplicationData; public CurrentEpoch CurrentEpoch; - public NextEpoch NextEpoch; + public NextEpoch NextEpoch; public ConnectionId ConnectionId; @@ -95,13 +99,15 @@ namespace Impostor.Hazel.Dtls public DateTime StartOfNegotiation; + public SemaphoreSlim Semaphore = new SemaphoreSlim(1, 1); + public PeerData() { ByteSpan block = new byte[2 * Finished.Size]; this.CurrentEpoch.ServerFinishedVerification = block.Slice(0, Finished.Size); this.CurrentEpoch.ExpectedClientFinishedVerification = block.Slice(Finished.Size, Finished.Size); - ResetPeer(ConnectionId.Create(new IPEndPoint(0,0), 0), 1); + ResetPeer(ConnectionId.Create(new IPEndPoint(0, 0), 0), 1); } public void ResetPeer(ConnectionId connectionId, ulong nextExpectedSequenceNumber) @@ -158,17 +164,16 @@ namespace Impostor.Hazel.Dtls private readonly ConcurrentDictionary existingPeers = new ConcurrentDictionary(); - private int connectionSerial_unsafe = 0; + private int connectionSerial_unsafe = 0; /// /// Create a new instance of the DTLS listener /// - /// /// - /// /// - public DtlsConnectionListener(int numWorkers, IPEndPoint endPoint, ILogger logger, IPMode ipMode = IPMode.IPv4) - : base(numWorkers, endPoint, logger, ipMode) + /// + public DtlsConnectionListener(IPEndPoint endPoint, ObjectPool readerPool, IPMode ipMode = IPMode.IPv4) + : base(endPoint, readerPool, ipMode) { this.random = RandomNumberGenerator.Create(); @@ -177,10 +182,98 @@ namespace Impostor.Hazel.Dtls this.nextCookieHmacRotation = DateTime.UtcNow + CookieHmacRotationTimeout; } + internal async ValueTask SendData(ByteSpan span, IPEndPoint endPoint) + { + var array = span.ToArray(); + await base.SendData(array, array.Length, endPoint); + } + + internal override async ValueTask SendData(byte[] bytes, int length, IPEndPoint remoteEndPoint) + { + var span = new ByteSpan(bytes); + + PeerData peer; + if (!this.existingPeers.TryGetValue(remoteEndPoint, out peer)) + { + Logger.Warning("Peer not found"); + // Drop messages if we don't know how to send them + return; + } + + await peer.Semaphore.WaitAsync(); + { + // If we're negotiating a new epoch, queue data + if (peer.Epoch == 0 || peer.NextEpoch.State != HandshakeState.ExpectingHello) + { + ByteSpan copyOfSpan = new byte[span.Length]; + span.CopyTo(copyOfSpan); + + peer.QueuedApplicationDataMessage.Add(copyOfSpan); + return; + } + + // Send any queued application data now + for (int ii = 0, nn = peer.QueuedApplicationDataMessage.Count; ii != nn; ++ii) + { + ByteSpan queuedSpan = peer.QueuedApplicationDataMessage[ii]; + + Record outgoingRecord = new Record(); + outgoingRecord.ContentType = ContentType.ApplicationData; + outgoingRecord.Epoch = peer.Epoch; + outgoingRecord.SequenceNumber = peer.CurrentEpoch.NextOutgoingSequence; + outgoingRecord.Length = (ushort)peer.CurrentEpoch.RecordProtection.GetEncryptedSize(queuedSpan.Length); + ++peer.CurrentEpoch.NextOutgoingSequence; + + // Encode the record to wire format + ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; + ByteSpan writer = packet; + outgoingRecord.Encode(writer); + writer = writer.Slice(Record.Size); + queuedSpan.CopyTo(writer); + + // Protect the record + peer.CurrentEpoch.RecordProtection.EncryptServerPlaintext( + packet.Slice(Record.Size, outgoingRecord.Length) + , packet.Slice(Record.Size, queuedSpan.Length) + , ref outgoingRecord); + + await this.SendData(packet, remoteEndPoint); + } + + peer.QueuedApplicationDataMessage.Clear(); + + { + Record outgoingRecord = new Record(); + outgoingRecord.ContentType = ContentType.ApplicationData; + outgoingRecord.Epoch = peer.Epoch; + outgoingRecord.SequenceNumber = peer.CurrentEpoch.NextOutgoingSequence; + outgoingRecord.Length = (ushort)peer.CurrentEpoch.RecordProtection.GetEncryptedSize(span.Length); + ++peer.CurrentEpoch.NextOutgoingSequence; + + // Encode the record to wire format + ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; + ByteSpan writer = packet; + outgoingRecord.Encode(writer); + writer = writer.Slice(Record.Size); + span.CopyTo(writer); + + // Protect the record + peer.CurrentEpoch.RecordProtection.EncryptServerPlaintext( + packet.Slice(Record.Size, outgoingRecord.Length) + , packet.Slice(Record.Size, span.Length) + , ref outgoingRecord + ); + + await this.SendData(packet, remoteEndPoint); + } + } + peer.Semaphore.Release(); + } + /// - protected override void Dispose(bool disposing) + public override async ValueTask DisposeAsync() { - base.Dispose(disposing); + await base.DisposeAsync(); this.random?.Dispose(); this.random = null; @@ -194,6 +287,7 @@ namespace Impostor.Hazel.Dtls { pair.Value.Dispose(); } + this.existingPeers.Clear(); } @@ -247,26 +341,25 @@ namespace Impostor.Hazel.Dtls /// This is primarily a wrapper around ProcessIncomingMessage /// to ensure `reader.Recycle()` is always called /// - protected override void ProcessIncomingMessageFromOtherThread(MessageReader reader, IPEndPoint peerAddress, ConnectionId connectionId) + protected override ValueTask ProcessData(UdpReceiveResult data) { - ByteSpan message = new ByteSpan(reader.Buffer, reader.Offset + reader.Position, reader.BytesRemaining); - this.ProcessIncomingMessage(message, peerAddress); - reader.Recycle(); + ByteSpan message = new ByteSpan(data.Buffer); + return this.ProcessIncomingMessage(message, data.RemoteEndPoint); } /// /// Handle an incoming datagram from the network /// - private void ProcessIncomingMessage(ByteSpan message, IPEndPoint peerAddress) + private async ValueTask ProcessIncomingMessage(ByteSpan message, IPEndPoint peerAddress) { PeerData peer = null; if (!this.existingPeers.TryGetValue(peerAddress, out peer)) { - HandleNonPeerRecord(message, peerAddress); + await HandleNonPeerRecord(message, peerAddress); return; } - lock (peer) + await peer.Semaphore.WaitAsync(); { // Each incoming packet may contain multiple DTLS // records @@ -275,14 +368,15 @@ namespace Impostor.Hazel.Dtls Record record; if (!Record.Parse(out record, message)) { - this.Logger.WriteError($"Dropping malformed record from `{peerAddress}`"); + Logger.Error($"Dropping malformed record from `{peerAddress}`"); return; } + message = message.Slice(Record.Size); if (message.Length < record.Length) { - this.Logger.WriteError($"Dropping malformed record from `{peerAddress}` Length({record.Length}) AvailableBytes({message.Length})"); + Logger.Error($"Dropping malformed record from `{peerAddress}` Length({record.Length}) AvailableBytes({message.Length})"); return; } @@ -292,7 +386,7 @@ namespace Impostor.Hazel.Dtls // Early-out and drop ApplicationData records if (record.ContentType == ContentType.ApplicationData && !peer.CanHandleApplicationData) { - this.Logger.WriteInfo($"Dropping ApplicationData record from `{peerAddress}` Cannot process yet"); + Logger.Information($"Dropping ApplicationData record from `{peerAddress}` Cannot process yet"); continue; } @@ -307,34 +401,35 @@ namespace Impostor.Hazel.Dtls Handshake handshake; if (!Handshake.Parse(out handshake, recordPayload)) { - this.Logger.WriteError($"Dropping malformed re-negotiation Handshake from `{peerAddress}`"); + Logger.Error($"Dropping malformed re-negotiation Handshake from `{peerAddress}`"); continue; } + handshakePayload = handshakePayload.Slice(Handshake.Size); if (handshake.FragmentOffset != 0 || handshake.Length != handshake.FragmentLength) { - this.Logger.WriteError($"Dropping fragmented re-negotiation Handshake from `{peerAddress}`"); + Logger.Error($"Dropping fragmented re-negotiation Handshake from `{peerAddress}`"); continue; } else if (handshake.MessageType != HandshakeType.ClientHello) { - this.Logger.WriteVerbose($"Dropping non-ClientHello re-negotiation Handshake from `{peerAddress}`"); + Logger.Error($"Dropping non-ClientHello re-negotiation Handshake from `{peerAddress}`"); continue; } else if (handshakePayload.Length < handshake.Length) { - this.Logger.WriteError($"Dropping malformed re-negotiation Handshake from `{peerAddress}`: Length({handshake.Length}) AvailableBytes({handshakePayload.Length})"); + Logger.Error($"Dropping malformed re-negotiation Handshake from `{peerAddress}`: Length({handshake.Length}) AvailableBytes({handshakePayload.Length})"); } - if (!this.HandleClientHello(peer, peerAddress, ref record, ref handshake, recordPayload, handshakePayload)) + if (!await this.HandleClientHello(peer, peerAddress, record, handshake, recordPayload, handshakePayload)) { return; } continue; } - this.Logger.WriteVerbose($"Dropping bad-epoch record from `{peerAddress}` RecordEpoch({record.Epoch}) CurrentEpoch({peer.Epoch})"); + Logger.Error($"Dropping bad-epoch record from `{peerAddress}` RecordEpoch({record.Epoch}) CurrentEpoch({peer.Epoch})"); continue; } @@ -346,13 +441,13 @@ namespace Impostor.Hazel.Dtls { if (windowIndex >= 64) { - this.Logger.WriteInfo($"Dropping too-old record from `{peerAddress}` Sequence({record.SequenceNumber}) Expected({peer.CurrentEpoch.NextExpectedSequence})"); + Logger.Information($"Dropping too-old record from `{peerAddress}` Sequence({record.SequenceNumber}) Expected({peer.CurrentEpoch.NextExpectedSequence})"); continue; } if ((peer.CurrentEpoch.PreviousSequenceWindowBitmask & windowMask) != 0) { - this.Logger.WriteInfo($"Dropping duplicate record from `{peerAddress}`"); + Logger.Information($"Dropping duplicate record from `{peerAddress}`"); continue; } } @@ -361,7 +456,7 @@ namespace Impostor.Hazel.Dtls int decryptedSize = peer.CurrentEpoch.RecordProtection.GetDecryptedSize(recordPayload.Length); if (decryptedSize < 0) { - this.Logger.WriteInfo($"Dropping malformed record: Length {recordPayload.Length} Decrypted length: {decryptedSize}"); + Logger.Information($"Dropping malformed record: Length {recordPayload.Length} Decrypted length: {decryptedSize}"); continue; } @@ -369,7 +464,7 @@ namespace Impostor.Hazel.Dtls if (!peer.CurrentEpoch.RecordProtection.DecryptCiphertextFromClient(decryptedPayload, recordPayload, ref record)) { - this.Logger.WriteVerbose($"Dropping non-authentic record from `{peerAddress}`"); + Logger.Error($"Dropping non-authentic record from `{peerAddress}`"); return; } @@ -392,7 +487,7 @@ namespace Impostor.Hazel.Dtls case ContentType.ChangeCipherSpec: if (peer.NextEpoch.State != HandshakeState.ExpectingChangeCipherSpec) { - this.Logger.WriteError($"Dropping unexpected ChangeChiperSpec record from `{peerAddress}` State({peer.NextEpoch.State})"); + Logger.Error($"Dropping unexpected ChangeChiperSpec record from `{peerAddress}` State({peer.NextEpoch.State})"); break; } else if (peer.NextEpoch.RecordProtection == null) @@ -401,13 +496,13 @@ namespace Impostor.Hazel.Dtls /// happen on a well-formed server. Debug.Assert(false, "How did we receive a ChangeCipherSpec message without a pending record protection instance?"); - this.Logger.WriteError($"Dropping ChangeCipherSpec message from `{peerAddress}`: No pending record protection"); + Logger.Error($"Dropping ChangeCipherSpec message from `{peerAddress}`: No pending record protection"); break; } if (!ChangeCipherSpec.Parse(recordPayload)) { - this.Logger.WriteError($"Dropping malformed ChangeCipherSpec message from `{peerAddress}`"); + Logger.Error($"Dropping malformed ChangeCipherSpec message from `{peerAddress}`"); break; } @@ -435,27 +530,25 @@ namespace Impostor.Hazel.Dtls break; case ContentType.Alert: - this.Logger.WriteError($"Dropping unsupported Alert record from `{peerAddress}`"); + Logger.Error($"Dropping unsupported Alert record from `{peerAddress}`"); break; case ContentType.Handshake: - if (!ProcessHandshake(peer, peerAddress, ref record, recordPayload)) + if (!await ProcessHandshake(peer, peerAddress, record, recordPayload)) { return; } + break; case ContentType.ApplicationData: // Forward data to the application - MessageReader reader = MessageReader.GetSized(recordPayload.Length); - reader.Length = recordPayload.Length; - recordPayload.CopyTo(reader.Buffer); - - base.ProcessIncomingMessageFromOtherThread(reader, peerAddress, peer.ConnectionId); + await base.ProcessData(new UdpReceiveResult(recordPayload.ToArray(), peerAddress)); break; } } } + peer.Semaphore.Release(); } /// @@ -469,7 +562,7 @@ namespace Impostor.Hazel.Dtls /// True if further processing of the underlying datagram /// should be continues. Otherwise, false. /// - private bool ProcessHandshake(PeerData peer, IPEndPoint peerAddress, ref Record record, ByteSpan message) + private async ValueTask ProcessHandshake(PeerData peer, IPEndPoint peerAddress, Record record, ByteSpan message) { // Each record may have multiple handshake payloads while (message.Length > 0) @@ -479,14 +572,14 @@ namespace Impostor.Hazel.Dtls Handshake handshake; if (!Handshake.Parse(out handshake, message)) { - this.Logger.WriteError($"Dropping malformed Handshake message from `{peerAddress}`"); + Logger.Error($"Dropping malformed Handshake message from `{peerAddress}`"); return false; } message = message.Slice(Handshake.Size); if (message.Length < handshake.Length) { - this.Logger.WriteError($"Dropping malformed Handshake message from `{peerAddress}`"); + Logger.Error($"Dropping malformed Handshake message from `{peerAddress}`"); return false; } @@ -498,7 +591,7 @@ namespace Impostor.Hazel.Dtls // from the client if (handshake.FragmentOffset != 0 || handshake.FragmentLength != handshake.Length) { - this.Logger.WriteError($"Dropping fragmented Handshake message from `{peerAddress}` Offset({handshake.FragmentOffset}) FragmentLength({handshake.FragmentLength}) Length({handshake.Length})"); + Logger.Error($"Dropping fragmented Handshake message from `{peerAddress}` Offset({handshake.FragmentOffset}) FragmentLength({handshake.FragmentLength}) Length({handshake.Length})"); continue; } @@ -508,7 +601,7 @@ namespace Impostor.Hazel.Dtls switch (handshake.MessageType) { case HandshakeType.ClientHello: - if (!this.HandleClientHello(peer, peerAddress, ref record, ref handshake, originalMessage, payload)) + if (!await this.HandleClientHello(peer, peerAddress, record, handshake, originalMessage, payload)) { return false; } @@ -517,19 +610,19 @@ namespace Impostor.Hazel.Dtls case HandshakeType.ClientKeyExchange: if (peer.NextEpoch.State != HandshakeState.ExpectingClientKeyExchange) { - this.Logger.WriteError($"Dropping unexpected ClientKeyExchange message form `{peerAddress}` State({peer.NextEpoch.State})"); + Logger.Error($"Dropping unexpected ClientKeyExchange message form `{peerAddress}` State({peer.NextEpoch.State})"); continue; } else if (handshake.MessageSequence != 5) { - this.Logger.WriteError($"Dropping bad-sequence ClientKeyExchange message from `{peerAddress}` MessageSequence({handshake.MessageSequence})"); + Logger.Error($"Dropping bad-sequence ClientKeyExchange message from `{peerAddress}` MessageSequence({handshake.MessageSequence})"); continue; } ByteSpan sharedSecret = new byte[peer.NextEpoch.Handshake.SharedKeySize()]; if (!peer.NextEpoch.Handshake.VerifyClientMessageAndGenerateSharedKey(sharedSecret, payload)) { - this.Logger.WriteError($"Dropping malformed ClientKeyExchange message from `{peerAddress}`"); + Logger.Error($"Dropping malformed ClientKeyExchange message from `{peerAddress}`"); return false; } @@ -566,7 +659,7 @@ namespace Impostor.Hazel.Dtls default: Debug.Assert(false, $"How did we agree to a cipher suite {peer.NextEpoch.SelectedCipherSuite} we can't create?"); - this.Logger.WriteError($"Dropping ClientKeyExchange message from `{peerAddress}` Unsuppored cipher suite"); + Logger.Error($"Dropping ClientKeyExchange message from `{peerAddress}` Unsuppored cipher suite"); return false; } @@ -605,14 +698,14 @@ namespace Impostor.Hazel.Dtls // epoch 0 if (peer.Epoch == 0) { - this.Logger.WriteError($"Dropping Finished message for 0-epoch from `{peerAddress}`"); + Logger.Error($"Dropping Finished message for 0-epoch from `{peerAddress}`"); continue; } // Cannot process a Finished message when we // are negotiating the next epoch else if (peer.NextEpoch.State != HandshakeState.ExpectingHello) { - this.Logger.WriteError($"Dropping Finished message while negotiating new epoch from `{peerAddress}`"); + Logger.Error($"Dropping Finished message while negotiating new epoch from `{peerAddress}`"); continue; } // Cannot process a Finished message without @@ -623,7 +716,7 @@ namespace Impostor.Hazel.Dtls /// happen on a well-formed server. Debug.Assert(false, "How do we have an established non-zero epoch without verify data?"); - this.Logger.WriteError($"Dropping Finished message (no verify data) from `{peerAddress}`"); + Logger.Error($"Dropping Finished message (no verify data) from `{peerAddress}`"); return false; } // Cannot process a Finished message without @@ -634,14 +727,14 @@ namespace Impostor.Hazel.Dtls /// happen on a well-formed server. Debug.Assert(false, "How do we have an established non-zero epoch with record protection for the previous epoch?"); - this.Logger.WriteError($"Dropping Finished message from `{peerAddress}`: No previous epoch record protection"); + Logger.Error($"Dropping Finished message from `{peerAddress}`: No previous epoch record protection"); return false; } // Verify message sequence if (handshake.MessageSequence != 6) { - this.Logger.WriteError($"Dropping bad-sequence Finished message from `{peerAddress}` MessageSequence({handshake.MessageSequence})"); + Logger.Error($"Dropping bad-sequence Finished message from `{peerAddress}` MessageSequence({handshake.MessageSequence})"); continue; } @@ -649,12 +742,12 @@ namespace Impostor.Hazel.Dtls // handshake sequence if (payload.Length != Finished.Size) { - this.Logger.WriteError($"Dropping malformed Finished message from `{peerAddress}`"); + Logger.Error($"Dropping malformed Finished message from `{peerAddress}`"); return false; } else if (1 != Crypto.Const.ConstantCompareSpans(payload, peer.CurrentEpoch.ExpectedClientFinishedVerification)) { - this.Logger.WriteError($"Dropping non-verified Finished Handshake from `{peerAddress}`"); + Logger.Error($"Dropping non-verified Finished Handshake from `{peerAddress}`"); // Abort the connection here // @@ -663,7 +756,7 @@ namespace Impostor.Hazel.Dtls // // Either way, there is not a feasible // way to progress the connection. - base.MarkConnectionAsStale(peer.ConnectionId); + // base.MarkConnectionAsStale(peer.ConnectionId); this.existingPeers.TryRemove(peerAddress, out peer); return false; } @@ -723,12 +816,12 @@ namespace Impostor.Hazel.Dtls // Current epoch can now handle application data peer.CanHandleApplicationData = true; - base.QueueRawData(packet, peerAddress); + await SendData(packet, peerAddress); break; // Drop messages that we do not support case HandshakeType.CertificateVerify: - this.Logger.WriteError($"Dropping unsupported Handshake message from `{peerAddress}` MessageType({handshake.MessageType})"); + Logger.Error($"Dropping unsupported Handshake message from `{peerAddress}` MessageType({handshake.MessageType})"); continue; // Drop messages that originate from the server @@ -739,7 +832,7 @@ namespace Impostor.Hazel.Dtls case HandshakeType.ServerKeyExchange: case HandshakeType.CertificateRequest: case HandshakeType.ServerHelloDone: - this.Logger.WriteError($"Dropping server Handshake message from `{peerAddress}` MessageType({handshake.MessageType})"); + Logger.Error($"Dropping server Handshake message from `{peerAddress}` MessageType({handshake.MessageType})"); continue; } } @@ -755,12 +848,12 @@ namespace Impostor.Hazel.Dtls /// Parent record /// Parent Handshake header /// Handshake payload - private bool HandleClientHello(PeerData peer, IPEndPoint peerAddress, ref Record record, ref Handshake handshake, ByteSpan originalMessage, ByteSpan payload) + private async ValueTask HandleClientHello(PeerData peer, IPEndPoint peerAddress, Record record, Handshake handshake, ByteSpan originalMessage, ByteSpan payload) { // Verify message sequence if (handshake.MessageSequence != 0) { - this.Logger.WriteError($"Dropping bad-sequence ClientHello from `{peerAddress}` MessageSequence({handshake.MessageSequence})`"); + Logger.Error($"Dropping bad-sequence ClientHello from `{peerAddress}` MessageSequence({handshake.MessageSequence})`"); return true; } @@ -770,7 +863,7 @@ namespace Impostor.Hazel.Dtls // Always handle ClientHello for epoch 0 if (record.Epoch != 0) { - this.Logger.WriteError($"Dropping ClientHello from `{peer}` Not expecting ClientHello"); + Logger.Error($"Dropping ClientHello from `{peer}` Not expecting ClientHello"); return true; } } @@ -778,7 +871,7 @@ namespace Impostor.Hazel.Dtls ClientHello clientHello; if (!ClientHello.Parse(out clientHello, payload)) { - this.Logger.WriteError($"Dropping malformed ClientHello Handshake message from `{peerAddress}`"); + Logger.Error($"Dropping malformed ClientHello Handshake message from `{peerAddress}`"); return false; } @@ -786,7 +879,7 @@ namespace Impostor.Hazel.Dtls CipherSuite selectedCipherSuite = CipherSuite.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256; if (!clientHello.ContainsCipherSuite(CipherSuite.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256) || !clientHello.ContainsCurve(NamedCurve.x25519)) { - this.Logger.WriteError($"Dropping ClientHello from `{peerAddress}` No compatible cipher suite"); + Logger.Error($"Dropping ClientHello from `{peerAddress}` No compatible cipher suite"); return false; } @@ -806,7 +899,7 @@ namespace Impostor.Hazel.Dtls recordProtection = peer.CurrentEpoch.RecordProtection; } - this.SendHelloVerifyRequest(peerAddress, outgoingSequence, record.Epoch, recordProtection); + await this.SendHelloVerifyRequest(peerAddress, outgoingSequence, record.Epoch, recordProtection); return true; } } @@ -821,7 +914,7 @@ namespace Impostor.Hazel.Dtls // Inform the parent layer that the existing // connection should be abandoned. - base.MarkConnectionAsStale(oldConnectionId); + // base.MarkConnectionAsStale(oldConnectionId); } // Determine if this is an original message, or a retransmission @@ -839,14 +932,14 @@ namespace Impostor.Hazel.Dtls } else { - this.Logger.WriteError($"Dropping ClientHello from `{peerAddress}` Could not create TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 cipher suite"); + Logger.Error($"Dropping ClientHello from `{peerAddress}` Could not create TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 cipher suite"); return false; } break; default: - this.Logger.WriteError($"Dropping ClientHello from `{peerAddress}` Could not create handshake cipher suite"); + Logger.Error($"Dropping ClientHello from `{peerAddress}` Could not create handshake cipher suite"); return false; } @@ -945,7 +1038,7 @@ namespace Impostor.Hazel.Dtls , ref initialRecord ); - base.QueueRawData(packet, peerAddress); + await SendData(packet, peerAddress); // Record record payload for verification if (recordMessagesForVerifyData) @@ -1007,7 +1100,7 @@ namespace Impostor.Hazel.Dtls , ref additionalRecord ); - base.QueueRawData(packet, peerAddress); + await SendData(packet, peerAddress); } // Describe final record of the flight @@ -1065,7 +1158,7 @@ namespace Impostor.Hazel.Dtls , ref finalRecord ); - base.QueueRawData(packet, peerAddress); + await SendData(packet, peerAddress); return true; } @@ -1075,12 +1168,12 @@ namespace Impostor.Hazel.Dtls /// /// Incoming datagram /// Originating address - private void HandleNonPeerRecord(ByteSpan message, IPEndPoint peerAddress) + private async ValueTask HandleNonPeerRecord(ByteSpan message, IPEndPoint peerAddress) { Record record; if (!Record.Parse(out record, message)) { - this.Logger.WriteError($"Dropping malformed record from non-peer `{peerAddress}`"); + Logger.Error($"Dropping malformed record from non-peer `{peerAddress}`"); return; } message = message.Slice(Record.Size); @@ -1097,7 +1190,7 @@ namespace Impostor.Hazel.Dtls /// worst case we're dealing with a malicious actor. /// In the malicious case, we'll end up dropping the /// connection later in the process. - this.Logger.WriteInfo($"Received multiple record from non-peer `{peerAddress}`. Dropping all but first"); + Logger.Information($"Received multiple record from non-peer `{peerAddress}`. Dropping all but first"); if (message.Length < record.Length) { return; @@ -1116,7 +1209,7 @@ namespace Impostor.Hazel.Dtls // We only accept Handshake protocol messages from non-peers if (record.ContentType != ContentType.Handshake) { - this.Logger.WriteError($"Dropping non-handhsake message from non-peer `{peerAddress}`"); + Logger.Error($"Dropping non-handhsake message from non-peer `{peerAddress}`"); return; } @@ -1125,22 +1218,23 @@ namespace Impostor.Hazel.Dtls Handshake handshake; if (!Handshake.Parse(out handshake, message)) { - this.Logger.WriteError($"Dropping malformed handshake message from non-peer `{peerAddress}`"); + Logger.Error($"Dropping malformed handshake message from non-peer `{peerAddress}`"); return; } // We only accept ClientHello messages from non-peers if (handshake.MessageType != HandshakeType.ClientHello) { - this.Logger.WriteError($"Dropping non-ClientHello ({handshake.MessageType}) message from non-peer `{peerAddress}`"); + Logger.Error($"Dropping non-ClientHello ({handshake.MessageType}) message from non-peer `{peerAddress}`"); return; } + message = message.Slice(Handshake.Size); ClientHello clientHello; if (!ClientHello.Parse(out clientHello, message)) { - this.Logger.WriteError($"Dropping malformed ClientHello message from non-peer `{peerAddress}`"); + Logger.Error($"Dropping malformed ClientHello message from non-peer `{peerAddress}`"); return; } @@ -1150,7 +1244,7 @@ namespace Impostor.Hazel.Dtls { if (!HelloVerifyRequest.VerifyCookie(clientHello.Cookie, peerAddress, this.previousCookieHmac)) { - this.SendHelloVerifyRequest(peerAddress, 1, 0, NullRecordProtection.Instance); + await this.SendHelloVerifyRequest(peerAddress, 1, 0, NullRecordProtection.Instance); return; } } @@ -1161,14 +1255,15 @@ namespace Impostor.Hazel.Dtls this.existingPeers[peerAddress] = peer; - lock (peer) + await peer.Semaphore.WaitAsync(); { - this.ProcessHandshake(peer, peerAddress, ref record, originalMessage); + await this.ProcessHandshake(peer, peerAddress, record, originalMessage); } + peer.Semaphore.Release(); } //Send a HelloVerifyRequest handshake message to a peer - private void SendHelloVerifyRequest(IPEndPoint peerAddress, ulong recordSequence, ushort epoch, IRecordProtection recordProtection) + private ValueTask SendHelloVerifyRequest(IPEndPoint peerAddress, ulong recordSequence, ushort epoch, IRecordProtection recordProtection) { // Do we need to rotate the HMAC key? DateTime now = DateTime.UtcNow; @@ -1211,111 +1306,31 @@ namespace Impostor.Hazel.Dtls , ref record ); - base.QueueRawData(packet, peerAddress); - } - - /// - /// Handle a requrest to send a datagram to the network - /// - protected override void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint) - { - PeerData peer; - if (!this.existingPeers.TryGetValue(remoteEndPoint, out peer)) - { - // Drop messages if we don't know how to send them - return; - } - - lock (peer) - { - // If we're negotiating a new epoch, queue data - if (peer.Epoch == 0 || peer.NextEpoch.State != HandshakeState.ExpectingHello) - { - ByteSpan copyOfSpan = new byte[span.Length]; - span.CopyTo(copyOfSpan); - - peer.QueuedApplicationDataMessage.Add(copyOfSpan); - return; - } - - // Send any queued application data now - for (int ii = 0, nn = peer.QueuedApplicationDataMessage.Count; ii != nn; ++ii) - { - ByteSpan queuedSpan = peer.QueuedApplicationDataMessage[ii]; - - Record outgoingRecord = new Record(); - outgoingRecord.ContentType = ContentType.ApplicationData; - outgoingRecord.Epoch = peer.Epoch; - outgoingRecord.SequenceNumber = peer.CurrentEpoch.NextOutgoingSequence; - outgoingRecord.Length = (ushort)peer.CurrentEpoch.RecordProtection.GetEncryptedSize(queuedSpan.Length); - ++peer.CurrentEpoch.NextOutgoingSequence; - - // Encode the record to wire format - ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; - ByteSpan writer = packet; - outgoingRecord.Encode(writer); - writer = writer.Slice(Record.Size); - queuedSpan.CopyTo(writer); - - // Protect the record - peer.CurrentEpoch.RecordProtection.EncryptServerPlaintext( - packet.Slice(Record.Size, outgoingRecord.Length) - , packet.Slice(Record.Size, queuedSpan.Length) - , ref outgoingRecord); - - base.QueueRawData(packet, remoteEndPoint); - } - peer.QueuedApplicationDataMessage.Clear(); - - { - Record outgoingRecord = new Record(); - outgoingRecord.ContentType = ContentType.ApplicationData; - outgoingRecord.Epoch = peer.Epoch; - outgoingRecord.SequenceNumber = peer.CurrentEpoch.NextOutgoingSequence; - outgoingRecord.Length = (ushort)peer.CurrentEpoch.RecordProtection.GetEncryptedSize(span.Length); - ++peer.CurrentEpoch.NextOutgoingSequence; - - // Encode the record to wire format - ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; - ByteSpan writer = packet; - outgoingRecord.Encode(writer); - writer = writer.Slice(Record.Size); - span.CopyTo(writer); - - // Protect the record - peer.CurrentEpoch.RecordProtection.EncryptServerPlaintext( - packet.Slice(Record.Size, outgoingRecord.Length) - , packet.Slice(Record.Size, span.Length) - , ref outgoingRecord - ); - - base.QueueRawData(packet, remoteEndPoint); - } - } + return SendData(packet, peerAddress); } /// - public override void DisconnectOldConnections(TimeSpan maxAge, MessageWriter disconnectMessage) - { - DateTime now = DateTime.UtcNow; - foreach (KeyValuePair kvp in this.existingPeers) - { - PeerData peer = kvp.Value; - lock(peer) - { - if (peer.Epoch == 0 || peer.NextEpoch.State != HandshakeState.ExpectingHello) - { - TimeSpan negotiationAge = now - peer.StartOfNegotiation; - if (negotiationAge > maxAge) - { - base.MarkConnectionAsStale(peer.ConnectionId); - } - } - } - } - - base.DisconnectOldConnections(maxAge, disconnectMessage); - } + // public override ValueTask DisconnectOldConnections(TimeSpan maxAge, MessageWriter disconnectMessage) + // { + // DateTime now = DateTime.UtcNow; + // foreach (KeyValuePair kvp in this.existingPeers) + // { + // PeerData peer = kvp.Value; + // lock (peer) + // { + // if (peer.Epoch == 0 || peer.NextEpoch.State != HandshakeState.ExpectingHello) + // { + // TimeSpan negotiationAge = now - peer.StartOfNegotiation; + // if (negotiationAge > maxAge) + // { + // base.MarkConnectionAsStale(peer.ConnectionId); + // } + // } + // } + // } + // + // return base.DisconnectOldConnections(maxAge, disconnectMessage); + // } /// /// Allocate a new connection id diff --git a/Hazel/Dtls/DtlsUnityConnection.cs b/Hazel/Dtls/DtlsUnityConnection.cs deleted file mode 100644 index bcd9ef2..0000000 --- a/Hazel/Dtls/DtlsUnityConnection.cs +++ /dev/null @@ -1,1077 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Diagnostics; -using System.IO; -using System.Net; -using System.Security.Cryptography; -using System.Security.Cryptography.X509Certificates; -using Impostor.Hazel.Crypto; -using Impostor.Hazel.Udp; -using Impostor.Hazel.UPnP; - -namespace Impostor.Hazel.Dtls -{ - /// - /// Connects to a UDP-DTLS server - /// - /// - public class DtlsUnityConnection : UnityUdpClientConnection - { - /// - /// Current state of the handshake sequence - /// - enum HandshakeState - { - Established, - - ExpectingServerHello, - ExpectingCertificate, - ExpectingServerKeyExchange, - ExpectingServerHelloDone, - ExpectingChangeCipherSpec, - ExpectingFinished, - - Initializing, - } - - /// - /// State data for the current epoch - /// - struct CurrentEpoch - { - public ulong NextOutgoingSequence; - - public ulong NextExpectedSequence; - public ulong PreviousSequenceWindowBitmask; - - public IRecordProtection RecordProtection; - } - - struct FragmentRange - { - public int Offset; - public int Length; - } - - /// - /// State data for the next epoch - /// - struct NextEpoch - { - public ushort Epoch; - - public HandshakeState State; - - public ulong NextOutgoingSequence; - - public DateTime NextPacketResendTime; - - public CipherSuite SelectedCipherSuite; - public IRecordProtection RecordProtection; - public IHandshakeCipherSuite Handshake; - public ByteSpan Cookie; - public MemoryStream VerificationStream; - public RSA ServerPublicKey; - - public ByteSpan ClientRandom; - public ByteSpan ServerRandom; - - public ByteSpan MasterSecret; - public ByteSpan ServerVerification; - - public List CertificateFragments; - public ByteSpan CertificatePayload; - } - - private readonly object syncRoot = new object(); - private readonly RandomNumberGenerator random = RandomNumberGenerator.Create(); - - private ushort epoch; - private CurrentEpoch currentEpoch; - private NextEpoch nextEpoch; - private TimeSpan handshakeResendTimeout = TimeSpan.FromMilliseconds(200); - - private readonly List queuedApplicationData = new List(); - - private X509Certificate2Collection serverCertificates = new X509Certificate2Collection(); - - private readonly ILogger logger = null; - - /// - /// Create a new instance of the DTLS connection - /// - /// - public DtlsUnityConnection(ILogger logger, IPEndPoint remoteEndPoint, IPMode ipMode = IPMode.IPv4) - : base(remoteEndPoint, ipMode) - { - this.logger = logger; - this.nextEpoch.ServerRandom = new byte[Random.Size]; - this.nextEpoch.ClientRandom = new byte[Random.Size]; - this.nextEpoch.ServerVerification = new byte[Finished.Size]; - this.nextEpoch.CertificateFragments = new List(); - - this.ResetConnectionState(); - } - - /// - protected override void Dispose(bool disposing) - { - base.Dispose(disposing); - - lock (this.syncRoot) - { - this.ResetConnectionState(); - } - } - - /// - /// Set the list of valid server certificates - /// - /// - /// List of certificates of authentic servers - /// - public void SetValidServerCertificates(X509Certificate2Collection certificateCollection) - { - lock (this.syncRoot) - { - foreach (X509Certificate2 certificate in certificateCollection) - { - if (!(certificate.PublicKey.Key is RSA)) - { - throw new ArgumentException("Certificate must be signed with an RSA key", nameof(certificateCollection)); - } - } - - this.serverCertificates = certificateCollection; - } - } - - /// - /// Set the packet resend timer for handshake messages - /// - public void SetHandshakeResendTimeout(TimeSpan timeout) - { - lock (this.syncRoot) - { - this.handshakeResendTimeout = timeout; - } - } - - /// - /// Reset existing connection state - /// - private void ResetConnectionState() - { - this.currentEpoch.NextOutgoingSequence = 1; - this.currentEpoch.NextExpectedSequence = 1; - this.currentEpoch.PreviousSequenceWindowBitmask = 0; - this.currentEpoch.RecordProtection?.Dispose(); - this.currentEpoch.RecordProtection = NullRecordProtection.Instance; - - this.nextEpoch.Epoch = 1; - this.nextEpoch.State = HandshakeState.Initializing; - this.nextEpoch.NextOutgoingSequence = 1; - this.nextEpoch.NextPacketResendTime = DateTime.MinValue; - this.nextEpoch.SelectedCipherSuite = CipherSuite.TLS_NULL_WITH_NULL_NULL; - this.nextEpoch.RecordProtection?.Dispose(); - this.nextEpoch.RecordProtection = null; - this.nextEpoch.Handshake?.Dispose(); - this.nextEpoch.Handshake = null; - this.nextEpoch.Cookie = ByteSpan.Empty; - this.nextEpoch.VerificationStream?.Dispose(); - this.nextEpoch.VerificationStream = new MemoryStream(); - this.nextEpoch.ServerPublicKey = null; - this.nextEpoch.ServerRandom.SecureClear(); - this.nextEpoch.ClientRandom.SecureClear(); - this.nextEpoch.MasterSecret.SecureClear(); - this.nextEpoch.ServerVerification.SecureClear(); - this.nextEpoch.CertificateFragments.Clear(); - this.nextEpoch.CertificatePayload = ByteSpan.Empty; - - this.epoch = 0; - this.queuedApplicationData.Clear(); - } - - /// - /// Abort the existing connection and restart the process - /// - protected override void RestartConnection() - { - lock (this.syncRoot) - { - this.ResetConnectionState(); - this.nextEpoch.ClientRandom.FillWithRandom(this.random); - this.SendClientHello(); - } - - base.RestartConnection(); - } - - /// - protected override void ResendPacketsIfNeeded() - { - lock (this.syncRoot) - { - // Check if we need to resend handshake message - if (this.nextEpoch.State != HandshakeState.Established) - { - DateTime now = DateTime.UtcNow; - if (now >= this.nextEpoch.NextPacketResendTime) - { - switch (this.nextEpoch.State) - { - case HandshakeState.ExpectingServerHello: - case HandshakeState.ExpectingCertificate: - case HandshakeState.ExpectingServerKeyExchange: - case HandshakeState.ExpectingServerHelloDone: - this.SendClientHello(); - break; - - case HandshakeState.ExpectingChangeCipherSpec: - case HandshakeState.ExpectingFinished: - this.SendClientKeyExchangeFlight(true); - break; - - case HandshakeState.Established: - default: - break; - } - } - } - } - - base.ResendPacketsIfNeeded(); - } - - /// - /// Flush any queued application data packets - /// - private void FlushQueuedApplicationData() - { - foreach (ByteSpan queuedSpan in this.queuedApplicationData) - { - Record outgoingRecord = new Record(); - outgoingRecord.ContentType = ContentType.ApplicationData; - outgoingRecord.Epoch = this.epoch; - outgoingRecord.SequenceNumber = this.currentEpoch.NextOutgoingSequence; - outgoingRecord.Length = (ushort)this.currentEpoch.RecordProtection.GetEncryptedSize(queuedSpan.Length); - ++this.currentEpoch.NextOutgoingSequence; - - // Encode the record to wire format - ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; - ByteSpan writer = packet; - outgoingRecord.Encode(writer); - writer = writer.Slice(Record.Size); - queuedSpan.CopyTo(writer); - - // Protect the record - this.currentEpoch.RecordProtection.EncryptClientPlaintext( - packet.Slice(Record.Size, outgoingRecord.Length) - , packet.Slice(Record.Size, queuedSpan.Length) - , ref outgoingRecord - ); - - base.WriteBytesToConnection(packet.GetUnderlyingArray(), packet.Length); - } - this.queuedApplicationData.Clear(); - } - - /// - /// Request from the application to write data to the DTLS - /// stream. If appropriate, returns a byte span to send to - /// the wire. - /// - /// Plaintext bytes to write - /// Length of the bytes to write - /// - /// Encrypted data to put on the wire if appropriate, - /// otherwise an empty span - /// - private ByteSpan WriteBytesToConnectionInternal(byte[] bytes, int length) - { - lock (this.syncRoot) - { - // If we're negotiating a new epoch, queue data - if (this.nextEpoch.State != HandshakeState.Established) - { - ByteSpan copyOfSpan = new byte[length]; - new ByteSpan(bytes, 0, length).CopyTo(copyOfSpan); - - this.queuedApplicationData.Add(copyOfSpan); - return ByteSpan.Empty; - } - - // Send any queued application data now - this.FlushQueuedApplicationData(); - - Record outgoinRecord = new Record(); - outgoinRecord.ContentType = ContentType.ApplicationData; - outgoinRecord.Epoch = this.epoch; - outgoinRecord.SequenceNumber = this.currentEpoch.NextOutgoingSequence; - outgoinRecord.Length = (ushort)this.currentEpoch.RecordProtection.GetEncryptedSize(length); - ++this.currentEpoch.NextOutgoingSequence; - - // Encode the record to wire format - ByteSpan packet = new byte[Record.Size + outgoinRecord.Length]; - ByteSpan writer = packet; - outgoinRecord.Encode(writer); - writer = writer.Slice(Record.Size); - new ByteSpan(bytes, 0, length).CopyTo(writer); - - // Protect the record - this.currentEpoch.RecordProtection.EncryptClientPlaintext( - packet.Slice(Record.Size, outgoinRecord.Length) - , packet.Slice(Record.Size, length) - , ref outgoinRecord - ); - - return packet; - } - } - - /// - protected override void WriteBytesToConnection(byte[] bytes, int length) - { - ByteSpan wireData = this.WriteBytesToConnectionInternal(bytes, length); - if (wireData.Length > 0) - { - Debug.Assert(wireData.Offset == 0, "Got a non-zero write data offset"); - base.WriteBytesToConnection(wireData.GetUnderlyingArray(), wireData.Length); - } - } - - /// - protected override void WriteBytesToConnectionSync(byte[] bytes, int length) - { - ByteSpan wireData = this.WriteBytesToConnectionInternal(bytes, length); - if (wireData.Length > 0) - { - Debug.Assert(wireData.Offset == 0, "Got a non-zero write data offset"); - base.WriteBytesToConnectionSync(wireData.GetUnderlyingArray(), wireData.Length); - } - } - - /// - protected internal override void HandleReceive(MessageReader reader, int bytesReceived) - { - ByteSpan message = new ByteSpan(reader.Buffer, reader.Offset + reader.Position, reader.BytesRemaining); - lock (this.syncRoot) - { - this.HandleReceive(message); - } - - reader.Recycle(); - } - - /// - /// Handle an incoming datagram - /// - /// Bytes of the datagram - private void HandleReceive(ByteSpan span) - { - // Each incoming packet may contain multiple DTLS - // records - while (span.Length > 0) - { - Record record; - if (!Record.Parse(out record, span)) - { - this.logger.WriteError("Dropping malformed record"); - return; - } - span = span.Slice(Record.Size); - - if (span.Length < record.Length) - { - this.logger.WriteError($"Dropping malformed record. Length({record.Length}) Available Bytes({span.Length})"); - return; - } - - ByteSpan recordPayload = span.Slice(0, record.Length); - span = span.Slice(record.Length); - - // Early out and drop ApplicationData records - if (record.ContentType == ContentType.ApplicationData && this.nextEpoch.State != HandshakeState.Established) - { - this.logger.WriteError("Dropping ApplicationData record. Cannot process yet"); - continue; - } - - // Drop records from a different epoch - if (record.Epoch != this.epoch) - { - this.logger.WriteError($"Dropping bad-epoch record. RecordEpoch({record.Epoch}) Epoch({this.epoch})"); - continue; - } - - // Prevent replay attacks by dropping records - // we've already processed - int windowIndex = (int)(this.currentEpoch.NextExpectedSequence - record.SequenceNumber - 1); - ulong windowMask = 1ul << windowIndex; - if (record.SequenceNumber < this.currentEpoch.NextExpectedSequence) - { - if (windowIndex >= 64) - { - this.logger.WriteError($"Dropping too-old record: Sequnce({record.SequenceNumber}) Expected({this.currentEpoch.NextExpectedSequence})"); - continue; - } - - if ((this.currentEpoch.PreviousSequenceWindowBitmask & windowMask) != 0) - { - this.logger.WriteError("Dropping duplicate record"); - continue; - } - } - - // Verify record authenticity - int decryptedSize = this.currentEpoch.RecordProtection.GetDecryptedSize(recordPayload.Length); - ByteSpan decryptedPayload = recordPayload.ReuseSpanIfPossible(decryptedSize); - - if (!this.currentEpoch.RecordProtection.DecryptCiphertextFromServer(decryptedPayload, recordPayload, ref record)) - { - this.logger.WriteError("Dropping non-authentic record"); - return; - } - - recordPayload = decryptedPayload; - - // Update out sequence number bookkeeping - if (record.SequenceNumber >= this.currentEpoch.NextExpectedSequence) - { - int windowShift = (int)(record.SequenceNumber + 1 - this.currentEpoch.NextExpectedSequence); - this.currentEpoch.PreviousSequenceWindowBitmask <<= windowShift; - this.currentEpoch.NextExpectedSequence = record.SequenceNumber + 1; - } - else - { - this.currentEpoch.PreviousSequenceWindowBitmask |= windowMask; - } - - switch (record.ContentType) - { - case ContentType.ChangeCipherSpec: - if (this.nextEpoch.State != HandshakeState.ExpectingChangeCipherSpec) - { - this.logger.WriteError($"Dropping unexpected ChangeCipherSpec State({this.nextEpoch.State})"); - break; - } - else if (this.nextEpoch.RecordProtection == null) - { - ///NOTE(mendsley): This _should_ not - /// happen on a well-formed client. - Debug.Assert(false, "How did we receive a ChangeCipherSpec message without a pending record protection instance?"); - break; - } - - if (!ChangeCipherSpec.Parse(recordPayload)) - { - this.logger.WriteError("Dropping malformed ChangeCipherSpec message"); - break; - } - - // Migrate to the next epoch - this.epoch = this.nextEpoch.Epoch; - this.currentEpoch.RecordProtection = this.nextEpoch.RecordProtection; - this.currentEpoch.NextOutgoingSequence = this.nextEpoch.NextOutgoingSequence; - this.currentEpoch.NextExpectedSequence = 1; - this.currentEpoch.PreviousSequenceWindowBitmask = 0; - - this.nextEpoch.State = HandshakeState.ExpectingFinished; - this.nextEpoch.SelectedCipherSuite = CipherSuite.TLS_NULL_WITH_NULL_NULL; - this.nextEpoch.RecordProtection = null; - this.nextEpoch.Handshake?.Dispose(); - this.nextEpoch.Cookie = ByteSpan.Empty; - this.nextEpoch.VerificationStream.SetLength(0); - this.nextEpoch.ServerPublicKey = null; - this.nextEpoch.ServerRandom.SecureClear(); - this.nextEpoch.ClientRandom.SecureClear(); - this.nextEpoch.MasterSecret.SecureClear(); - break; - - case ContentType.Alert: - this.logger.WriteError("Dropping unsupported alert record"); - continue; - - case ContentType.Handshake: - if (!ProcessHandshake(ref record, recordPayload)) - { - return; - } - break; - - case ContentType.ApplicationData: - // Forward data to the application - MessageReader reader = MessageReader.GetSized(recordPayload.Length); - reader.Length = recordPayload.Length; - recordPayload.CopyTo(reader.Buffer); - - base.HandleReceive(reader, recordPayload.Length); - break; - } - } - } - - /// - /// Process an incoming Handshake protocol message - /// - /// Parent record - /// Record payload - /// - /// True if further processing of the underlying datagram - /// should be continues. Otherwise, false. - /// - private bool ProcessHandshake(ref Record record, ByteSpan message) - { - // Each record may have multiple Handshake messages - while (message.Length > 0) - { - ByteSpan originalPayload = message; - - Handshake handshake; - if (!Handshake.Parse(out handshake, message)) - { - this.logger.WriteError("Dropping malformed handshake message"); - return false; - } - message = message.Slice(Handshake.Size); - - if (message.Length < handshake.Length) - { - this.logger.WriteError($"Dropping malformed handshake message: AvailableBytes({message.Length}) Size({handshake.Length})"); - return false; - } - - originalPayload = originalPayload.Slice(0, (int)(Handshake.Size + handshake.Length)); - ByteSpan payload = originalPayload.Slice(Handshake.Size); - message = message.Slice((int)handshake.Length); - - // We only support fragmented Certificate messages - // from the server - if (handshake.MessageType != HandshakeType.Certificate && (handshake.FragmentOffset != 0 || handshake.FragmentLength != handshake.Length)) - { - this.logger.WriteError($"Dropping fragmented handshake message Type({handshake.MessageType}) Offset({handshake.FragmentOffset}) FragmentLength({handshake.FragmentLength}) Length({handshake.Length})"); - continue; - } - - switch (handshake.MessageType) - { - case HandshakeType.HelloVerifyRequest: - if (this.nextEpoch.State != HandshakeState.ExpectingServerHello) - { - this.logger.WriteError($"Dropping unexpected HelloVerifyRequest handshake message State({this.nextEpoch.State})"); - continue; - } - else if (handshake.MessageSequence != 0) - { - this.logger.WriteError($"Dropping bad-sequence HelloVerifyRequest MessageSequence({handshake.MessageSequence})"); - continue; - } - - HelloVerifyRequest helloVerifyRequest; - if (!HelloVerifyRequest.Parse(out helloVerifyRequest, payload)) - { - this.logger.WriteError("Dropping malformed HelloVerifyRequest handshake message"); - continue; - } - - // Save the cookie - this.nextEpoch.Cookie = new byte[helloVerifyRequest.Cookie.Length]; - helloVerifyRequest.Cookie.CopyTo(this.nextEpoch.Cookie); - - // Restart the handshake - this.nextEpoch.ClientRandom.FillWithRandom(this.random); - this.SendClientHello(); - break; - - case HandshakeType.ServerHello: - if (this.nextEpoch.State != HandshakeState.ExpectingServerHello) - { - this.logger.WriteError($"Dropping unexpected ServerHello handshake message State({this.nextEpoch.State})"); - continue; - } - else if (handshake.MessageSequence != 1) - { - this.logger.WriteError($"Dropping bad-sequence ServerHello MessageSequence({handshake.MessageSequence})"); - continue; - } - - ServerHello serverHello; - if (!ServerHello.Parse(out serverHello, payload)) - { - this.logger.WriteError("Dropping malformed ServerHello message"); - continue; - } - - switch (serverHello.CipherSuite) - { - case CipherSuite.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256: - this.nextEpoch.Handshake = new X25519EcdheRsaSha256(this.random); - break; - - default: - this.logger.WriteError($"Dropping malformed ServerHello message. Unsupported CipherSuite({serverHello.CipherSuite})"); - continue; - } - - // Save server parameters - this.nextEpoch.SelectedCipherSuite = serverHello.CipherSuite; - serverHello.Random.CopyTo(this.nextEpoch.ServerRandom); - this.nextEpoch.State = HandshakeState.ExpectingCertificate; - this.nextEpoch.CertificateFragments.Clear(); - this.nextEpoch.CertificatePayload = ByteSpan.Empty; - - // Append ServerHelllo message to the verification stream - this.nextEpoch.VerificationStream.Write( - originalPayload.GetUnderlyingArray() - , originalPayload.Offset - , originalPayload.Length - ); - break; - - case HandshakeType.Certificate: - if (this.nextEpoch.State != HandshakeState.ExpectingCertificate) - { - this.logger.WriteError($"Dropping unexpected Certificate handshake message State({this.nextEpoch.State})"); - continue; - } - else if (handshake.MessageSequence != 2) - { - this.logger.WriteError($"Dropping bad-sequence Certificate MessageSequence({handshake.MessageSequence})"); - continue; - } - - // If this is a fragmented message - if (handshake.FragmentLength != handshake.Length) - { - if (this.nextEpoch.CertificatePayload.Length != handshake.Length) - { - this.nextEpoch.CertificatePayload = new byte[handshake.Length]; - this.nextEpoch.CertificateFragments.Clear(); - } - - // Add this fragment - payload.CopyTo(this.nextEpoch.CertificatePayload.Slice((int)handshake.FragmentOffset, (int)handshake.FragmentLength)); - this.nextEpoch.CertificateFragments.Add(new FragmentRange {Offset = (int)handshake.FragmentOffset, Length = (int)handshake.FragmentLength }); - this.nextEpoch.CertificateFragments.Sort((FragmentRange lhs, FragmentRange rhs) => { - return lhs.Offset.CompareTo(rhs.Offset); - }); - - // Have we completed the message? - int currentOffset = 0; - bool valid = true; - foreach (FragmentRange range in this.nextEpoch.CertificateFragments) - { - if (range.Offset != currentOffset) - { - valid = false; - break; - } - - currentOffset += range.Length; - } - - if (currentOffset != this.nextEpoch.CertificatePayload.Length) - { - valid = false; - } - - // Still waiting on more fragments? - if (!valid) - { - continue; - } - - // Replace the message payload, and continue - this.nextEpoch.CertificateFragments.Clear(); - payload = this.nextEpoch.CertificatePayload; - } - - X509Certificate2 certificate; - if (!Certificate.Parse(out certificate, payload)) - { - this.logger.WriteError("Dropping malformed Certificate message"); - continue; - } - - // Verify the certificate is authenticate - if (!this.serverCertificates.Contains(certificate)) - { - this.logger.WriteError("Dropping malformed Certificate message: Certificate not authentic"); - continue; - } - - RSA publicKey = certificate.PublicKey.Key as RSA; - if (publicKey == null) - { - this.logger.WriteError("Dropping malfomed Certificate message: Certificate is not RSA signed"); - continue; - } - - // Add the final Certificate message to the verification stream - Handshake fullCertificateHandhake = handshake; - fullCertificateHandhake.FragmentOffset = 0; - fullCertificateHandhake.FragmentLength = fullCertificateHandhake.Length; - - byte[] serializedCertificateHandshake = new byte[Handshake.Size]; - fullCertificateHandhake.Encode(serializedCertificateHandshake); - this.nextEpoch.VerificationStream.Write(serializedCertificateHandshake, 0, serializedCertificateHandshake.Length); - this.nextEpoch.VerificationStream.Write(payload.GetUnderlyingArray(), payload.Offset, payload.Length); - - this.nextEpoch.ServerPublicKey = publicKey; - this.nextEpoch.State = HandshakeState.ExpectingServerKeyExchange; - break; - - case HandshakeType.ServerKeyExchange: - if (this.nextEpoch.State != HandshakeState.ExpectingServerKeyExchange) - { - this.logger.WriteError($"Dropping unexpected ServerKeyExchange handshake message State({this.nextEpoch.State})"); - continue; - } - else if (this.nextEpoch.ServerPublicKey == null) - { - ///NOTE(mendsley): This _should_ not - /// happen on a well-formed client - Debug.Assert(false, "How are we processing a ServerKeyExchange message without a server public key?"); - - this.logger.WriteError($"Dropping unexpected ServerKeyExchange handshake message: No server public key"); - continue; - } - else if (this.nextEpoch.Handshake == null) - { - ///NOTE(mendsley): This _should_ not - /// happen on a well-formed client - Debug.Assert(false, "How did we receive a ServerKeyExchange message without a handshake instance?"); - - this.logger.WriteError($"Dropping unexpected ServerKeyExchange handshake message: No key agreement interface"); - continue; - } - else if (handshake.MessageSequence != 3) - { - this.logger.WriteError($"Dropping bad-sequence ServerKeyExchange MessageSequence({handshake.MessageSequence})"); - continue; - } - - ByteSpan sharedSecret = new byte[this.nextEpoch.Handshake.SharedKeySize()]; - if (!this.nextEpoch.Handshake.VerifyServerMessageAndGenerateSharedKey(sharedSecret, payload, this.nextEpoch.ServerPublicKey)) - { - this.logger.WriteError("Dropping malformed ServerKeyExchangeMessage"); - return false; - } - - // Generate the session master secret - ByteSpan randomSeed = new byte[2 * Random.Size]; - this.nextEpoch.ClientRandom.CopyTo(randomSeed); - this.nextEpoch.ServerRandom.CopyTo(randomSeed.Slice(Random.Size)); - - const int MasterSecretSize = 48; - ByteSpan masterSecret = new byte[MasterSecretSize]; - PrfSha256.ExpandSecret( - masterSecret - , sharedSecret - , PrfLabel.MASTER_SECRET - , randomSeed - ); - - // Create record protection for the upcoming epoch - switch (this.nextEpoch.SelectedCipherSuite) - { - case CipherSuite.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256: - this.nextEpoch.RecordProtection = new Aes128GcmRecordProtection( - masterSecret - , this.nextEpoch.ServerRandom - , this.nextEpoch.ClientRandom - ); - break; - - default: - ///NOTE(mendsley): this _should_ not - /// happen on a well-formed client. - Debug.Assert(false, "SeverHello processing already approved this ciphersuite"); - - this.logger.WriteError($"Dropping malformed ServerKeyExchangeMessage: Could not create record protection"); - return false; - } - - this.nextEpoch.State = HandshakeState.ExpectingServerHelloDone; - this.nextEpoch.MasterSecret = masterSecret; - - // Append ServerKeyExchange to the verification stream - this.nextEpoch.VerificationStream.Write( - originalPayload.GetUnderlyingArray() - , originalPayload.Offset - , originalPayload.Length - ); - break; - - case HandshakeType.ServerHelloDone: - if (this.nextEpoch.State != HandshakeState.ExpectingServerHelloDone) - { - this.logger.WriteError($"Dropping unexpected ServerHelloDone handshake message State({this.nextEpoch.State})"); - continue; - } - else if (handshake.MessageSequence != 4) - { - this.logger.WriteError($"Dropping bad-sequence ServerHelloDone MessageSequence({handshake.MessageSequence})"); - continue; - } - - this.nextEpoch.State = HandshakeState.ExpectingChangeCipherSpec; - - // Append ServerHelloDone to the verification stream - this.nextEpoch.VerificationStream.Write( - originalPayload.GetUnderlyingArray() - , originalPayload.Offset - , originalPayload.Length - ); - - this.SendClientKeyExchangeFlight(false); - break; - - case HandshakeType.Finished: - if (this.nextEpoch.State != HandshakeState.ExpectingFinished) - { - this.logger.WriteError($"Dropping unexpected Finished handshake message State({this.nextEpoch.State})"); - continue; - } - else if (payload.Length != Finished.Size) - { - this.logger.WriteError($"Dropping malformed Finished handshake message Size({payload.Length})"); - continue; - } - else if (handshake.MessageSequence != 7) - { - this.logger.WriteError($"Dropping bad-sequence Finished MessageSequence({handshake.MessageSequence})"); - continue; - } - - // Verify the digest from the server - if (1 != Crypto.Const.ConstantCompareSpans(payload, this.nextEpoch.ServerVerification)) - { - this.logger.WriteError("Dropping non-verified Finished handshake message"); - return false; - } - - ++this.nextEpoch.Epoch; - this.nextEpoch.State = HandshakeState.Established; - this.nextEpoch.NextPacketResendTime = DateTime.MinValue; - this.nextEpoch.ServerVerification.SecureClear(); - this.nextEpoch.MasterSecret.SecureClear(); - - this.FlushQueuedApplicationData(); - break; - - // Drop messages we do not support - case HandshakeType.CertificateRequest: - case HandshakeType.HelloRequest: - this.logger.WriteError($"Dropping unsupported handshake message MessageType({handshake.MessageType})"); - break; - - // Drop messages that originate from the client - case HandshakeType.ClientHello: - case HandshakeType.ClientKeyExchange: - case HandshakeType.CertificateVerify: - this.logger.WriteError($"Dropping client handshake message MessageType({handshake.MessageType})"); - break; - } - } - - return true; - } - - /// - /// Send (resend) a ClientHello message to the server - /// - private void SendClientHello() - { - // Reset our verification stream - this.nextEpoch.VerificationStream.SetLength(0); - - // Describe our ClientHello flight - ClientHello clientHello = new ClientHello(); - clientHello.Random = this.nextEpoch.ClientRandom; - clientHello.Cookie = this.nextEpoch.Cookie; - clientHello.CipherSuites = new byte[2]; - clientHello.CipherSuites.WriteBigEndian16((ushort)CipherSuite.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256); - clientHello.SupportedCurves = new byte[2]; - clientHello.SupportedCurves.WriteBigEndian16((ushort)NamedCurve.x25519); - - Handshake handshake = new Handshake(); - handshake.MessageType = HandshakeType.ClientHello; - handshake.Length = (uint)clientHello.CalculateSize(); - handshake.MessageSequence = 0; - handshake.FragmentOffset = 0; - handshake.FragmentLength = handshake.Length; - - // Describe the record - int plaintextLength = (int)(Handshake.Size + handshake.Length); - Record outgoingRecord = new Record(); - outgoingRecord.ContentType = ContentType.Handshake; - outgoingRecord.Epoch = this.epoch; - outgoingRecord.SequenceNumber = this.currentEpoch.NextOutgoingSequence; - outgoingRecord.Length = (ushort)this.currentEpoch.RecordProtection.GetEncryptedSize(plaintextLength); - ++this.currentEpoch.NextOutgoingSequence; - - // Convert the record to wire format - ByteSpan packet = new byte[Record.Size + outgoingRecord.Length]; - ByteSpan writer = packet; - outgoingRecord.Encode(packet); - writer = writer.Slice(Record.Size); - handshake.Encode(writer); - writer = writer.Slice(Handshake.Size); - clientHello.Encode(writer); - - // Write ClientHello to the verification stream - this.nextEpoch.VerificationStream.Write( - packet.GetUnderlyingArray() - , Record.Size - , Handshake.Size + (int)handshake.Length - ); - - // Protect the record - this.currentEpoch.RecordProtection.EncryptClientPlaintext( - packet.Slice(Record.Size, outgoingRecord.Length) - , packet.Slice(Record.Size, plaintextLength) - , ref outgoingRecord - ); - - this.nextEpoch.State = HandshakeState.ExpectingServerHello; - this.nextEpoch.NextPacketResendTime = DateTime.UtcNow + this.handshakeResendTimeout; - base.WriteBytesToConnection(packet.GetUnderlyingArray(), packet.Length); - } - - /// - /// Send (resend) the ClientKeyExchange flight - /// - /// - /// True if this is a retransmit of the flight. Otherwise, - /// false - /// - private void SendClientKeyExchangeFlight(bool isRetransmit) - { - // Describe our flight - Handshake keyExchangeHandshake = new Handshake(); - keyExchangeHandshake.MessageType = HandshakeType.ClientKeyExchange; - keyExchangeHandshake.Length = (ushort)this.nextEpoch.Handshake.CalculateClientMessageSize(); - keyExchangeHandshake.MessageSequence = 5; - keyExchangeHandshake.FragmentOffset = 0; - keyExchangeHandshake.FragmentLength = keyExchangeHandshake.Length; - - Record keyExchangeRecord = new Record(); - keyExchangeRecord.ContentType = ContentType.Handshake; - keyExchangeRecord.Epoch = this.epoch; - keyExchangeRecord.SequenceNumber = this.currentEpoch.NextOutgoingSequence; - keyExchangeRecord.Length = (ushort)this.currentEpoch.RecordProtection.GetEncryptedSize(Handshake.Size + (int)keyExchangeHandshake.Length); - ++this.currentEpoch.NextOutgoingSequence; - - Record changeCipherSpecRecord = new Record(); - changeCipherSpecRecord.ContentType = ContentType.ChangeCipherSpec; - changeCipherSpecRecord.Epoch = this.epoch; - changeCipherSpecRecord.SequenceNumber = this.currentEpoch.NextOutgoingSequence; - changeCipherSpecRecord.Length = (ushort)this.currentEpoch.RecordProtection.GetEncryptedSize(ChangeCipherSpec.Size); - ++this.currentEpoch.NextOutgoingSequence; - - Handshake finishedHandshake = new Handshake(); - finishedHandshake.MessageType = HandshakeType.Finished; - finishedHandshake.Length = Finished.Size; - finishedHandshake.MessageSequence = 6; - finishedHandshake.FragmentOffset = 0; - finishedHandshake.FragmentLength = finishedHandshake.Length; - - Record finishedRecord = new Record(); - finishedRecord.ContentType = ContentType.Handshake; - finishedRecord.Epoch = this.nextEpoch.Epoch; - finishedRecord.SequenceNumber = this.nextEpoch.NextOutgoingSequence; - finishedRecord.Length = (ushort)this.nextEpoch.RecordProtection.GetEncryptedSize(Handshake.Size + (int)finishedHandshake.Length); - ++this.nextEpoch.NextOutgoingSequence; - - // Encode flight to wire format - int packetLength = 0 - + Record.Size + keyExchangeRecord.Length - + Record.Size + changeCipherSpecRecord.Length - + Record.Size + finishedRecord.Length; - ; - ByteSpan packet = new byte[packetLength]; - ByteSpan writer = packet; - - keyExchangeRecord.Encode(writer); - writer = writer.Slice(Record.Size); - keyExchangeHandshake.Encode(writer); - writer = writer.Slice(Handshake.Size); - this.nextEpoch.Handshake.EncodeClientKeyExchangeMessage(writer); - - ByteSpan startOfChangeCipherSpecRecord = packet.Slice(Record.Size + keyExchangeRecord.Length); - writer = startOfChangeCipherSpecRecord; - changeCipherSpecRecord.Encode(writer); - writer = writer.Slice(Record.Size); - ChangeCipherSpec.Encode(writer); - writer = writer.Slice(ChangeCipherSpec.Size); - - ByteSpan startOfFinishedRecord = startOfChangeCipherSpecRecord.Slice(Record.Size + changeCipherSpecRecord.Length); - writer = startOfFinishedRecord; - finishedRecord.Encode(writer); - writer = writer.Slice(Record.Size); - finishedHandshake.Encode(writer); - writer = writer.Slice(Handshake.Size); - - // Interject here to writer our client key exchange - // message into the verification stream - if (!isRetransmit) - { - this.nextEpoch.VerificationStream.Write( - packet.GetUnderlyingArray() - , Record.Size - , Handshake.Size + (int)keyExchangeHandshake.Length - ); - } - - // Calculate the hash of the verification stream - ByteSpan handshakeHash; - using (SHA256 sha256 = SHA256.Create()) - { - this.nextEpoch.VerificationStream.Position = 0; - handshakeHash = sha256.ComputeHash(this.nextEpoch.VerificationStream); - } - - // Expand our master secret into Finished digests for the client and server - PrfSha256.ExpandSecret( - this.nextEpoch.ServerVerification - , this.nextEpoch.MasterSecret - , PrfLabel.SERVER_FINISHED - , handshakeHash - ); - - PrfSha256.ExpandSecret( - writer.Slice(0, Finished.Size) - , this.nextEpoch.MasterSecret - , PrfLabel.CLIENT_FINISHED - , handshakeHash - ); - writer = writer.Slice(Finished.Size); - - // Protect the ClientKeyExchange record - this.currentEpoch.RecordProtection.EncryptClientPlaintext( - packet.Slice(Record.Size, keyExchangeRecord.Length) - , packet.Slice(Record.Size, Handshake.Size + (int)keyExchangeHandshake.Length) - , ref keyExchangeRecord - ); - - // Protect the ChangeCipherSpec record - this.currentEpoch.RecordProtection.EncryptClientPlaintext( - startOfChangeCipherSpecRecord.Slice(Record.Size, changeCipherSpecRecord.Length) - , startOfChangeCipherSpecRecord.Slice(Record.Size, ChangeCipherSpec.Size) - , ref changeCipherSpecRecord - ); - - // Protect the Finished record - this.nextEpoch.RecordProtection.EncryptClientPlaintext( - startOfFinishedRecord.Slice(Record.Size, finishedRecord.Length) - , startOfFinishedRecord.Slice(Record.Size, Handshake.Size + (int)finishedHandshake.Length) - , ref finishedRecord - ); - - this.nextEpoch.State = HandshakeState.ExpectingChangeCipherSpec; - this.nextEpoch.NextPacketResendTime = DateTime.UtcNow + this.handshakeResendTimeout; - base.WriteBytesToConnection(packet.GetUnderlyingArray(), packet.Length); - } - } -} diff --git a/Hazel/Extensions/ServiceProviderExtensions.cs b/Hazel/Extensions/ServiceProviderExtensions.cs new file mode 100644 index 0000000..56c7380 --- /dev/null +++ b/Hazel/Extensions/ServiceProviderExtensions.cs @@ -0,0 +1,21 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.ObjectPool; + +namespace Impostor.Hazel.Extensions +{ + public static class ServiceProviderExtensions + { + public static void AddHazel(this IServiceCollection services) + { + services.TryAddSingleton(new DefaultObjectPoolProvider()); + + services.AddSingleton(serviceProvider => + { + var provider = serviceProvider.GetRequiredService(); + var policy = ActivatorUtilities.CreateInstance(serviceProvider); + return provider.Create(policy); + }); + } + } +} diff --git a/Hazel/FewerThreads/HazelThreadPool.cs b/Hazel/FewerThreads/HazelThreadPool.cs deleted file mode 100644 index 97a37d8..0000000 --- a/Hazel/FewerThreads/HazelThreadPool.cs +++ /dev/null @@ -1,39 +0,0 @@ -using System.Threading; - -namespace Impostor.Hazel.FewerThreads -{ - internal class HazelThreadPool - { - private Thread[] threads; - - public HazelThreadPool(int numThreads, ThreadStart action) - { - this.threads = new Thread[numThreads]; - for (int i = 0; i < this.threads.Length; ++i) - { - this.threads[i] = new Thread(action); - } - } - - public void Start() - { - for (int i = 0; i < this.threads.Length; ++i) - { - this.threads[i].Start(); - } - } - - public void Join() - { - for (int i = 0; i < this.threads.Length; ++i) - { - var thread = this.threads[i]; - try - { - thread.Join(); - } - catch { } - } - } - } -} \ No newline at end of file diff --git a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs b/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs deleted file mode 100644 index 1cb4fd8..0000000 --- a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs +++ /dev/null @@ -1,404 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using Impostor.Hazel.Udp; -using Impostor.Hazel.UPnP; - -namespace Impostor.Hazel.FewerThreads -{ - /// - /// Listens for new UDP connections and creates UdpConnections for them. - /// - /// - public class ThreadLimitedUdpConnectionListener : IDisposable - { - private struct SendMessageInfo - { - public ByteSpan Span; - public IPEndPoint Recipient; - } - - private struct ReceiveMessageInfo - { - public MessageReader Message; - public IPEndPoint Sender; - public ConnectionId ConnectionId; - } - - private const int SendReceiveBufferSize = 1024 * 1024; - private const int BufferSize = ushort.MaxValue; - - public event Action NewConnection; - - /// - /// A callback for early connection rejection. - /// * Return false to reject connection. - /// * A null response is ok, we just won't send anything. - /// - public AcceptConnectionCheck AcceptConnection; - public delegate bool AcceptConnectionCheck(IPEndPoint endPoint, byte[] input, out byte[] response); - - private Socket socket; - protected ILogger Logger; - - public IPEndPoint EndPoint { get; } - public IPMode IPMode { get; } - - private Thread reliablePacketThread; - private Thread receiveThread; - private Thread sendThread; - private HazelThreadPool processThreads; - - public struct ConnectionId : IEquatable - { - public IPEndPoint EndPoint; - public int Serial; - - public static ConnectionId Create(IPEndPoint endPoint, int serial) - { - return new ConnectionId{ - EndPoint = endPoint, - Serial = serial, - }; - } - - public bool Equals(ConnectionId other) - { - return this.Serial == other.Serial - && this.EndPoint.Equals(other.EndPoint) - ; - } - - public override bool Equals(object obj) - { - if (obj is ConnectionId) - { - return this.Equals((ConnectionId)obj); - } - - return false; - } - - public override int GetHashCode() - { - ///NOTE(mendsley): We're only hashing the endpoint - /// here, as the common case will have one - /// connection per address+port tuple. - return this.EndPoint.GetHashCode(); - } - } - - private ConcurrentDictionary allConnections = new ConcurrentDictionary(); - private ConcurrentStack staleConnections = new ConcurrentStack(); - - private BlockingCollection receiveQueue; - private BlockingCollection sendQueue = new BlockingCollection(); - - public int MaxAge - { - get - { - var now = DateTime.UtcNow; - TimeSpan max = new TimeSpan(); - foreach (var con in allConnections.Values) - { - var val = now - con.CreationTime; - if (val > max) max = val; - } - - return (int)max.TotalSeconds; - } - } - - public int ConnectionCount { get { return this.allConnections.Count; } } - public int SendQueueLength { get { return this.sendQueue.Count; } } - public int ReceiveQueueLength { get { return this.receiveQueue.Count; } } - - private bool isActive; - - public ThreadLimitedUdpConnectionListener(int numWorkers, IPEndPoint endPoint, ILogger logger, IPMode ipMode = IPMode.IPv4) - { - this.Logger = logger; - this.EndPoint = endPoint; - this.IPMode = ipMode; - - this.receiveQueue = new BlockingCollection(10000); - - this.socket = UdpConnection.CreateSocket(this.IPMode); - this.socket.ExclusiveAddressUse = true; - this.socket.Blocking = false; - - this.socket.ReceiveBufferSize = SendReceiveBufferSize; - this.socket.SendBufferSize = SendReceiveBufferSize; - - this.reliablePacketThread = new Thread(ManageReliablePackets); - this.sendThread = new Thread(SendLoop); - this.receiveThread = new Thread(ReceiveLoop); - this.processThreads = new HazelThreadPool(numWorkers, ProcessingLoop); - } - - ~ThreadLimitedUdpConnectionListener() - { - this.Dispose(false); - } - - protected void MarkConnectionAsStale(ConnectionId connectionId) - { - if (this.allConnections.ContainsKey(connectionId)) - { - this.staleConnections.Push(connectionId); - } - } - - public virtual void DisconnectOldConnections(TimeSpan maxAge, MessageWriter disconnectMessage) - { - var now = DateTime.UtcNow; - foreach (var conn in this.allConnections.Values) - { - if (now - conn.CreationTime > maxAge) - { - conn.Disconnect("Stale Connection", disconnectMessage); - } - } - - ConnectionId connectionId; - while (this.staleConnections.TryPop(out connectionId)) - { - ThreadLimitedUdpServerConnection connection; - if (this.allConnections.TryGetValue(connectionId, out connection)) - { - connection.Disconnect("Stale Connection", disconnectMessage); - } - } - } - - private void ManageReliablePackets() - { - while (this.isActive) - { - foreach (var kvp in this.allConnections) - { - var sock = kvp.Value; - sock.ManageReliablePackets(); - } - - Thread.Sleep(100); - } - } - - public void Start() - { - try - { - socket.Bind(EndPoint); - } - catch (SocketException e) - { - throw new HazelException("Could not start listening as a SocketException occurred", e); - } - - this.isActive = true; - this.reliablePacketThread.Start(); - this.sendThread.Start(); - this.receiveThread.Start(); - this.processThreads.Start(); - } - - private void ReceiveLoop() - { - while (this.isActive) - { - if (this.socket.Poll(Timeout.Infinite, SelectMode.SelectRead)) - { - EndPoint remoteEP = new IPEndPoint(this.EndPoint.Address, this.EndPoint.Port); - MessageReader message = MessageReader.GetSized(BufferSize); - try - { - message.Length = socket.ReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP); - } - catch (SocketException sx) - { - message.Recycle(); - this.Logger.WriteError("Socket Ex in ReceiveLoop: " + sx.Message); - continue; - } - catch (Exception ex) - { - message.Recycle(); - this.Logger.WriteError("Stopped due to: " + ex.Message); - return; - } - - ConnectionId connectionId = ConnectionId.Create((IPEndPoint)remoteEP, 0); - this.ProcessIncomingMessageFromOtherThread(message, (IPEndPoint)remoteEP, connectionId); - } - } - } - private void ProcessingLoop() - { - foreach (ReceiveMessageInfo msg in this.receiveQueue.GetConsumingEnumerable()) - { - try - { - this.ReadCallback(msg.Message, msg.Sender, msg.ConnectionId); - } - catch - { - - } - } - } - protected virtual void ProcessIncomingMessageFromOtherThread(MessageReader message, IPEndPoint remoteEndPoint, ConnectionId connectionId) - { - this.receiveQueue.Add(new ReceiveMessageInfo() { Message = message, Sender = remoteEndPoint, ConnectionId = connectionId }); - } - - private void SendLoop() - { - foreach (SendMessageInfo msg in this.sendQueue.GetConsumingEnumerable()) - { - try - { - if (this.socket.Poll(Timeout.Infinite, SelectMode.SelectWrite)) - { - this.socket.SendTo(msg.Span.GetUnderlyingArray(), msg.Span.Offset, msg.Span.Length, SocketFlags.None, msg.Recipient); - } - } - catch (Exception e) - { - this.Logger.WriteError("Error in loop while sending: " + e.Message); - Thread.Sleep(1); - } - } - } - - void ReadCallback(MessageReader message, IPEndPoint remoteEndPoint, ConnectionId connectionId) - { - int bytesReceived = message.Length; - bool aware = true; - bool isHello = message.Buffer[0] == (byte)UdpSendOption.Hello; - - // If we're aware of this connection use the one already - // If this is a new client then connect with them! - ThreadLimitedUdpServerConnection connection; - if (!this.allConnections.TryGetValue(connectionId, out connection)) - { - lock (this.allConnections) - { - if (!this.allConnections.TryGetValue(connectionId, out connection)) - { - // Check for malformed connection attempts - if (!isHello) - { - message.Recycle(); - return; - } - - if (AcceptConnection != null) - { - if (!AcceptConnection((IPEndPoint)remoteEndPoint, message.Buffer, out var response)) - { - message.Recycle(); - if (response != null) - { - SendDataRaw(response, remoteEndPoint); - } - - return; - } - } - - aware = false; - connection = new ThreadLimitedUdpServerConnection(this, connectionId, (IPEndPoint)remoteEndPoint, this.IPMode); - if (!this.allConnections.TryAdd(connectionId, connection)) - { - throw new HazelException("Failed to add a connection. This should never happen."); - } - } - } - } - - // If it's a new connection invoke the NewConnection event. - // This needs to happen before handling the message because in localhost scenarios, the ACK and - // subsequent messages can happen before the NewConnection event sets up OnDataRecieved handlers - if (!aware) - { - // Skip header and hello byte; - message.Offset = 4; - message.Length = bytesReceived - 4; - message.Position = 0; - this.NewConnection?.Invoke(new NewConnectionEventArgs(message, connection)); - } - - // Inform the connection of the buffer (new connections need to send an ack back to client) - connection.HandleReceive(message, bytesReceived); - - if (isHello && aware) - { - message.Recycle(); - } - } - - internal void SendDataRaw(byte[] response, IPEndPoint remoteEndPoint) - { - QueueRawData(response, remoteEndPoint); - } - - protected virtual void QueueRawData(ByteSpan span, IPEndPoint remoteEndPoint) - { - this.sendQueue.TryAdd(new SendMessageInfo() { Span = span, Recipient = remoteEndPoint }); - } - - /// - /// Removes a virtual connection from the list. - /// - /// Connection key of the virtual connection. - internal bool RemoveConnectionTo(ConnectionId connectionId) - { - return this.allConnections.TryRemove(connectionId, out var conn); - } - - protected virtual void Dispose(bool disposing) - { - foreach (var kvp in this.allConnections) - { - kvp.Value.Dispose(); - } - - bool wasActive = this.isActive; - this.isActive = false; - - // Flush outgoing packets - this.sendQueue?.CompleteAdding(); - if (wasActive) - { - this.sendThread.Join(); - } - - try { this.socket.Shutdown(SocketShutdown.Both); } catch { } - try { this.socket.Close(); } catch { } - try { this.socket.Dispose(); } catch { } - - this.receiveQueue?.CompleteAdding(); - - if (wasActive) - { - this.reliablePacketThread.Join(); - this.receiveThread.Join(); - this.processThreads.Join(); - } - - this.receiveQueue?.Dispose(); - this.receiveQueue = null; - this.sendQueue?.Dispose(); - this.sendQueue = null; - } - - public void Dispose() - { - this.Dispose(true); - } - } -} diff --git a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs b/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs deleted file mode 100644 index efa554d..0000000 --- a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs +++ /dev/null @@ -1,106 +0,0 @@ -using System; -using System.Net; -using Impostor.Hazel.Udp; - -namespace Impostor.Hazel.FewerThreads -{ - /// - /// Represents a servers's connection to a client that uses the UDP protocol. - /// - /// - internal sealed class ThreadLimitedUdpServerConnection : UdpConnection - { - public readonly DateTime CreationTime = DateTime.UtcNow; - - /// - /// The connection listener that we use the socket of. - /// - /// - /// Udp server connections utilize the same socket in the listener for sends/receives, this is the listener that - /// created this connection and is hence the listener this conenction sends and receives via. - /// - public ThreadLimitedUdpConnectionListener Listener { get; private set; } - - public ThreadLimitedUdpConnectionListener.ConnectionId ConnectionId { get; private set; } - - /// - /// Creates a UdpConnection for the virtual connection to the endpoint. - /// - /// The listener that created this connection. - /// The endpoint that we are connected to. - /// The IPMode we are connected using. - internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, ThreadLimitedUdpConnectionListener.ConnectionId connectionId, IPEndPoint endPoint, IPMode IPMode) - : base() - { - this.Listener = listener; - this.ConnectionId = connectionId; - this.EndPoint = endPoint; - this.IPMode = IPMode; - - State = ConnectionState.Connected; - this.InitializeKeepAliveTimer(); - } - - /// - protected override void WriteBytesToConnection(byte[] bytes, int length) - { - if (bytes.Length != length) throw new ArgumentException("I made an assumption here. I hope you see this error."); - - Listener.SendDataRaw(bytes, EndPoint); - } - - /// - /// - /// This will always throw a HazelException. - /// - public override void Connect(byte[] bytes = null, int timeout = 5000) - { - throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); - } - - /// - /// - /// This will always throw a HazelException. - /// - public override void ConnectAsync(byte[] bytes = null) - { - throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); - } - - /// - /// Sends a disconnect message to the end point. - /// - protected override bool SendDisconnect(MessageWriter data = null) - { - if (!Listener.RemoveConnectionTo(this.ConnectionId)) return false; - this._state = ConnectionState.NotConnected; - - var bytes = EmptyDisconnectBytes; - if (data != null && data.Length > 0) - { - if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable."); - - bytes = data.ToByteArray(true); - bytes[0] = (byte)UdpSendOption.Disconnect; - } - - try - { - Listener.SendDataRaw(bytes, EndPoint); - } - catch { } - - return true; - } - - protected override void Dispose(bool disposing) - { - if (disposing) - { - SendDisconnect(); - } - - base.Dispose(disposing); - } - } -} diff --git a/Hazel/MessageReader.cs b/Hazel/MessageReader.cs index 0c084ca..6e2a555 100644 --- a/Hazel/MessageReader.cs +++ b/Hazel/MessageReader.cs @@ -1,198 +1,105 @@ using System; -using System.IO; +using System.Buffers.Binary; +using System.Numerics; using System.Runtime.CompilerServices; using System.Text; +using Impostor.Api; +using Impostor.Api.Games; +using Impostor.Api.Net.Inner; +using Impostor.Api.Net.Messages; +using Impostor.Api.Unity; +using Microsoft.Extensions.ObjectPool; namespace Impostor.Hazel { - public class MessageReader : IRecyclable + public class MessageReader : IMessageReader { - public static readonly ObjectPool ReaderPool = new ObjectPool(() => new MessageReader()); + private readonly ObjectPool _pool; + private bool _inUse; - public byte[] Buffer; - public byte Tag; - - public int Length; - public int Offset; - - public int BytesRemaining => this.Length - this.Position; - - private MessageReader Parent; - - public int Position - { - get { return this._position; } - set - { - this._position = value; - this.readHead = value + Offset; - } - } - - private int _position; - private int readHead; - - public static MessageReader GetSized(int minSize) + internal MessageReader(ObjectPool pool) { - var output = ReaderPool.GetObject(); - - if (output.Buffer == null || output.Buffer.Length < minSize) - { - output.Buffer = new byte[minSize]; - } - else - { - Array.Clear(output.Buffer, 0, output.Buffer.Length); - } - - output.Offset = 0; - output.Position = 0; - output.Tag = byte.MaxValue; - return output; + _pool = pool; } - public static MessageReader Get(byte[] buffer) - { - var output = ReaderPool.GetObject(); + public byte[] Buffer { get; private set; } - output.Buffer = buffer; - output.Offset = 0; - output.Position = 0; - output.Length = buffer.Length; - output.Tag = byte.MaxValue; + public int Offset { get; internal set; } - return output; - } + public int Position { get; internal set; } - public static MessageReader CopyMessageIntoParent(MessageReader source) - { - var output = MessageReader.GetSized(source.Length + 3); - System.Buffer.BlockCopy(source.Buffer, source.Offset - 3, output.Buffer, 0, source.Length + 3); + public int Length { get; internal set; } + + public int BytesRemaining => this.Length - this.Position; - output.Offset = 0; - output.Position = 0; - output.Length = source.Length + 3; + public byte Tag { get; private set; } - return output; - } + public MessageReader Parent { get; private set; } - public static MessageReader Get(MessageReader source) + private int ReadPosition => Offset + Position; + public void Update(byte[] buffer, int offset = 0, int position = 0, int? length = null, byte tag = byte.MaxValue, MessageReader parent = null) { - var output = MessageReader.GetSized(source.Buffer.Length); - System.Buffer.BlockCopy(source.Buffer, 0, output.Buffer, 0, source.Buffer.Length); - - output.Offset = source.Offset; - - output._position = source._position; - output.readHead = source.readHead; - - output.Length = source.Length; - output.Tag = source.Tag; - - return output; + _inUse = true; + + Buffer = buffer; + Offset = offset; + Position = position; + Length = length ?? buffer.Length; + Tag = tag; + Parent = parent; } - public static MessageReader Get(byte[] buffer, int offset) + internal void Reset() { - // Ensure there is at least a header - if (offset + 3 > buffer.Length) return null; - - var output = ReaderPool.GetObject(); - - output.Buffer = buffer; - output.Offset = offset; - output.Position = 0; - - output.Length = output.ReadUInt16(); - output.Tag = output.ReadByte(); - - output.Offset += 3; - output.Position = 0; - - return output; + _inUse = false; + + Tag = byte.MaxValue; + Buffer = null; + Offset = 0; + Position = 0; + Length = 0; + Parent = null; } - /// - /// Produces a MessageReader using the parent's buffer. This MessageReader should **NOT** be recycled. - /// - public MessageReader ReadMessage() + public IMessageReader ReadMessage() { - // Ensure there is at least a header - if (this.BytesRemaining < 3) throw new InvalidDataException($"ReadMessage header is longer than message length: 3 of {this.BytesRemaining}"); + var length = ReadUInt16(); + var tag = FastByte(); + var pos = ReadPosition; - var output = new MessageReader(); + Position += length; - output.Parent = this; - output.Buffer = this.Buffer; - output.Offset = this.readHead; - output.Position = 0; - - output.Length = output.ReadUInt16(); - output.Tag = output.ReadByte(); - - output.Offset += 3; - output.Position = 0; - - if (this.BytesRemaining < output.Length + 3) throw new InvalidDataException($"Message Length at Position {this.readHead} is longer than message length: {output.Length + 3} of {this.BytesRemaining}"); - - this.Position += output.Length + 3; - return output; + var reader = _pool.Get(); + reader.Update(Buffer, pos, 0, length, tag, this); + return reader; } - /// - /// Produces a MessageReader with a new buffer. This MessageReader should be recycled. - /// - public MessageReader ReadMessageAsNewBuffer() + public void RemoveMessage(IMessageReader message) { - if (this.BytesRemaining < 3) throw new InvalidDataException($"ReadMessage header is longer than message length: 3 of {this.BytesRemaining}"); - - var len = this.ReadUInt16(); - var tag = this.ReadByte(); + if (message.Buffer != Buffer) + { + throw new ImpostorProtocolException("Tried to remove message from a message that does not have the same buffer."); + } - if (this.BytesRemaining < len) throw new InvalidDataException($"Message Length at Position {this.readHead} is longer than message length: {len} of {this.BytesRemaining}"); + // Offset of where to start removing. + var offsetStart = message.Offset - 3; - var output = MessageReader.GetSized(len); + // Offset of where to end removing. + var offsetEnd = message.Offset + message.Length; - output.Parent = this; - Array.Copy(this.Buffer, this.readHead, output.Buffer, 0, len); + // The amount of bytes to copy over ourselves. + var lengthToCopy = message.Buffer.Length - offsetEnd; - output.Length = len; - output.Tag = tag; + System.Buffer.BlockCopy(Buffer, offsetEnd, Buffer, offsetStart, lengthToCopy); - this.Position += output.Length; - return output; - } - - public MessageWriter StartWriter() - { - var output = new MessageWriter(this.Buffer); - output.Position = this.readHead; - return output; - } - - public void RemoveMessage(MessageReader reader) - { - var temp = MessageReader.GetSized(reader.Buffer.Length); - try - { - var headerOffset = reader.Offset - 3; - var endOfMessage = reader.Offset + reader.Length; - var len = reader.Buffer.Length - endOfMessage; - - Array.Copy(reader.Buffer, endOfMessage, temp.Buffer, 0, len); - Array.Copy(temp.Buffer, 0, this.Buffer, headerOffset, len); - - this.AdjustLength(reader.Offset, reader.Length + 3); - } - finally - { - temp.Recycle(); - } + ((MessageReader) message).Parent.AdjustLength(message.Offset, message.Length + 3); } private void AdjustLength(int offset, int amount) { - if (this.readHead > offset) + this.Length -= amount; + + if (this.ReadPosition > offset) { this.Position -= amount; } @@ -200,11 +107,10 @@ namespace Impostor.Hazel if (Parent != null) { var lengthOffset = this.Offset - 3; - var curLen = this.Buffer[lengthOffset] - | (this.Buffer[lengthOffset + 1] << 8); + var curLen = this.Buffer[lengthOffset] | + (this.Buffer[lengthOffset + 1] << 8); curLen -= amount; - this.Length -= amount; this.Buffer[lengthOffset] = (byte)curLen; this.Buffer[lengthOffset + 1] = (byte)(this.Buffer[lengthOffset + 1] >> 8); @@ -213,10 +119,12 @@ namespace Impostor.Hazel } } - public void Recycle() + public void Dispose() { - this.Parent = null; - ReaderPool.PutObject(this); + if (_inUse) + { + _pool.Return(this); + } } #region Read Methods @@ -238,83 +146,61 @@ namespace Impostor.Hazel public ushort ReadUInt16() { - ushort output = - (ushort)(this.FastByte() - | this.FastByte() << 8); + var output = BinaryPrimitives.ReadUInt16LittleEndian(Buffer.AsSpan(ReadPosition)); + Position += sizeof(ushort); return output; } public short ReadInt16() { - short output = - (short)(this.FastByte() - | this.FastByte() << 8); + var output = BinaryPrimitives.ReadInt16LittleEndian(Buffer.AsSpan(ReadPosition)); + Position += sizeof(short); return output; } public uint ReadUInt32() { - uint output = this.FastByte() - | (uint)this.FastByte() << 8 - | (uint)this.FastByte() << 16 - | (uint)this.FastByte() << 24; - + var output = BinaryPrimitives.ReadUInt32LittleEndian(Buffer.AsSpan(ReadPosition)); + Position += sizeof(uint); return output; } public int ReadInt32() { - int output = this.FastByte() - | this.FastByte() << 8 - | this.FastByte() << 16 - | this.FastByte() << 24; - + var output = BinaryPrimitives.ReadInt32LittleEndian(Buffer.AsSpan(ReadPosition)); + Position += sizeof(int); return output; } public unsafe float ReadSingle() { - float output = 0; - fixed (byte* bufPtr = &this.Buffer[this.readHead]) - { - byte* outPtr = (byte*)&output; - - *outPtr = *bufPtr; - *(outPtr + 1) = *(bufPtr + 1); - *(outPtr + 2) = *(bufPtr + 2); - *(outPtr + 3) = *(bufPtr + 3); - } - - this.Position += 4; + var output = BinaryPrimitives.ReadSingleLittleEndian(Buffer.AsSpan(ReadPosition)); + Position += sizeof(float); return output; } - public string ReadString() + public string ReadString(int length) { - int len = this.ReadPackedInt32(); - if (this.BytesRemaining < len) throw new InvalidDataException($"Read length is longer than message length: {len} of {this.BytesRemaining}"); - - string output = UTF8Encoding.UTF8.GetString(this.Buffer, this.readHead, len); - - this.Position += len; + var output = Encoding.UTF8.GetString(Buffer.AsSpan(ReadPosition, length)); + Position += length; return output; } - public byte[] ReadBytesAndSize() + public string ReadString() { - int len = this.ReadPackedInt32(); - if (this.BytesRemaining < len) throw new InvalidDataException($"Read length is longer than message length: {len} of {this.BytesRemaining}"); - - return this.ReadBytes(len); + return ReadString(ReadPackedInt32()); } - public byte[] ReadBytes(int length) + public ReadOnlyMemory ReadBytesAndSize() { - if (this.BytesRemaining < length) throw new InvalidDataException($"Read length is longer than message length: {length} of {this.BytesRemaining}"); + var len = ReadPackedInt32(); + return ReadBytes(len); + } - byte[] output = new byte[length]; - Array.Copy(this.Buffer, this.readHead, output, 0, output.Length); - this.Position += output.Length; + public ReadOnlyMemory ReadBytes(int length) + { + var output = Buffer.AsMemory(ReadPosition, length); + Position += length; return output; } @@ -333,9 +219,7 @@ namespace Impostor.Hazel while (readMore) { - if (this.BytesRemaining < 1) throw new InvalidDataException($"Read length is longer than message length."); - - byte b = this.ReadByte(); + byte b = FastByte(); if (b >= 0x80) { readMore = true; @@ -352,26 +236,47 @@ namespace Impostor.Hazel return output; } + + public T ReadNetObject(IGame game) where T : IInnerNetObject + { + return game.FindObjectByNetId(ReadPackedUInt32()); + } + + public Vector2 ReadVector2() + { + const float range = 50f; + + var x = ReadUInt16() / (float)ushort.MaxValue; + var y = ReadUInt16() / (float)ushort.MaxValue; + + return new Vector2(Mathf.Lerp(-range, range, x), Mathf.Lerp(-range, range, y)); + } + #endregion - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private byte FastByte() + public void CopyTo(IMessageWriter writer) { - this._position++; - return this.Buffer[this.readHead++]; + writer.Write((ushort)Length); + writer.Write((byte)Tag); + writer.Write(Buffer.AsMemory(Offset, Length)); } - public unsafe static bool IsLittleEndian() + public IMessageReader Copy(int offset = 0) { - byte b; - unsafe - { - int i = 1; - byte* bp = (byte*)&i; - b = *bp; - } + var reader = _pool.Get(); + reader.Update(Buffer, Offset + offset, Position, Length - offset, Tag, Parent); + return reader; + } - return b == 1; + public void Seek(int position) + { + Position = position; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private byte FastByte() + { + return Buffer[Offset + Position++]; } } } diff --git a/Hazel/MessageReaderPolicy.cs b/Hazel/MessageReaderPolicy.cs new file mode 100644 index 0000000..ef3939a --- /dev/null +++ b/Hazel/MessageReaderPolicy.cs @@ -0,0 +1,27 @@ +using System; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.ObjectPool; + +namespace Impostor.Hazel +{ + public class MessageReaderPolicy : IPooledObjectPolicy + { + private readonly IServiceProvider _serviceProvider; + + public MessageReaderPolicy(IServiceProvider serviceProvider) + { + _serviceProvider = serviceProvider; + } + + public MessageReader Create() + { + return new MessageReader(_serviceProvider.GetRequiredService>()); + } + + public bool Return(MessageReader obj) + { + obj.Reset(); + return true; + } + } +} diff --git a/Hazel/MessageWriter.cs b/Hazel/MessageWriter.cs index 7de621a..d2d104e 100644 --- a/Hazel/MessageWriter.cs +++ b/Hazel/MessageWriter.cs @@ -1,20 +1,21 @@ using System; using System.Collections.Generic; +using System.Net; +using System.Numerics; using System.Text; +using Impostor.Api.Games; +using Impostor.Api.Net.Inner; +using Impostor.Api.Net.Messages; +using Impostor.Api.Unity; namespace Impostor.Hazel { - /// - public class MessageWriter : IRecyclable + public class MessageWriter : IMessageWriter, IRecyclable { public static int BufferSize = 64000; - public static readonly ObjectPool WriterPool = new ObjectPool(() => new MessageWriter(BufferSize)); + private static readonly ObjectPoolCustom WriterPool = new ObjectPoolCustom(() => new MessageWriter(BufferSize)); - public byte[] Buffer; - public int Length; - public int Position; - - public SendOption SendOption { get; private set; } + public MessageType SendOption { get; private set; } private Stack messageStarts = new Stack(); @@ -29,6 +30,10 @@ namespace Impostor.Hazel { this.Buffer = new byte[bufferSize]; } + + public byte[] Buffer { get; } + public int Length { get; set; } + public int Position { get; set; } public byte[] ToByteArray(bool includeHeader) { @@ -42,13 +47,13 @@ namespace Impostor.Hazel { switch (this.SendOption) { - case SendOption.Reliable: + case MessageType.Reliable: { byte[] output = new byte[this.Length - 3]; System.Buffer.BlockCopy(this.Buffer, 3, output, 0, this.Length - 3); return output; } - case SendOption.None: + case MessageType.Unreliable: { byte[] output = new byte[this.Length - 1]; System.Buffer.BlockCopy(this.Buffer, 1, output, 0, this.Length - 1); @@ -62,7 +67,7 @@ namespace Impostor.Hazel /// /// The option specifying how the message should be sent. - public static MessageWriter Get(SendOption sendOption = SendOption.None) + public static MessageWriter Get(MessageType sendOption = MessageType.Unreliable) { var output = WriterPool.GetObject(); output.Clear(sendOption); @@ -72,7 +77,7 @@ namespace Impostor.Hazel public bool HasBytes(int expected) { - if (this.SendOption == SendOption.None) + if (this.SendOption == MessageType.Unreliable) { return this.Length > 1 + expected; } @@ -83,11 +88,8 @@ namespace Impostor.Hazel /// public void StartMessage(byte typeFlag) { - var messageStart = this.Position; - messageStarts.Push(messageStart); - this.Buffer[messageStart] = 0; - this.Buffer[messageStart + 1] = 0; - this.Position += 2; + messageStarts.Push(this.Position); + this.Position += 2; // Skip for size this.Write(typeFlag); } @@ -107,19 +109,19 @@ namespace Impostor.Hazel this.Length = this.Position; } - public void Clear(SendOption sendOption) + public void Clear(MessageType sendOption) { - Array.Clear(this.Buffer, 0, this.Buffer.Length); this.messageStarts.Clear(); this.SendOption = sendOption; this.Buffer[0] = (byte)sendOption; switch (sendOption) { default: - case SendOption.None: + case MessageType.Unreliable: this.Length = this.Position = 1; break; - case SendOption.Reliable: + + case MessageType.Reliable: this.Length = this.Position = 3; break; } @@ -134,25 +136,6 @@ namespace Impostor.Hazel #region WriteMethods - public void CopyFrom(MessageReader target) - { - int offset, length; - if (target.Tag == byte.MaxValue) - { - offset = target.Offset; - length = target.Length; - } - else - { - offset = target.Offset - 3; - length = target.Length + 3; - } - - System.Buffer.BlockCopy(target.Buffer, offset, this.Buffer, this.Position, length); - this.Position += length; - if (this.Position > this.Length) this.Length = this.Position; - } - public void Write(bool value) { this.Buffer[this.Position++] = (byte)(value ? 1 : 0); @@ -244,6 +227,19 @@ namespace Impostor.Hazel this.Write(bytes, offset, length); } + public void Write(ReadOnlyMemory data) + { + Write(data.Span); + } + + public void Write(ReadOnlySpan bytes) + { + bytes.CopyTo(this.Buffer.AsSpan(this.Position, bytes.Length)); + + this.Position += bytes.Length; + if (this.Position > this.Length) this.Length = this.Position; + } + public void Write(byte[] bytes) { Array.Copy(bytes, 0, this.Buffer, this.Position, bytes.Length); @@ -286,8 +282,7 @@ namespace Impostor.Hazel value >>= 7; } while (value > 0); } - #endregion - + public void Write(MessageWriter msg, bool includeHeader) { int offset = 0; @@ -295,10 +290,10 @@ namespace Impostor.Hazel { switch (msg.SendOption) { - case SendOption.None: + case MessageType.Unreliable: offset = 1; break; - case SendOption.Reliable: + case MessageType.Reliable: offset = 3; break; } @@ -307,6 +302,36 @@ namespace Impostor.Hazel this.Write(msg.Buffer, offset, msg.Length - offset); } + public void Write(IPAddress value) + { + this.Write(value.GetAddressBytes()); + } + + public void Write(GameCode value) + { + this.Write(value.Value); + } + + public void Write(IInnerNetObject innerNetObject) + { + if (innerNetObject == null) + { + this.Write(0); + } + else + { + this.WritePacked(innerNetObject.NetId); + } + } + + public void Write(Vector2 vector) + { + Write((ushort)(Mathf.ReverseLerp(vector.X) * (double) ushort.MaxValue)); + Write((ushort)(Mathf.ReverseLerp(vector.Y) * (double) ushort.MaxValue)); + } + + #endregion + public unsafe static bool IsLittleEndian() { byte b; @@ -319,5 +344,10 @@ namespace Impostor.Hazel return b == 1; } + + public void Dispose() + { + Recycle(); + } } } diff --git a/Hazel/NetworkConnection.cs b/Hazel/NetworkConnection.cs index 3755a09..a2a2adc 100644 --- a/Hazel/NetworkConnection.cs +++ b/Hazel/NetworkConnection.cs @@ -1,4 +1,5 @@ using System; +using System.Threading.Tasks; namespace Impostor.Hazel { @@ -43,18 +44,18 @@ namespace Impostor.Hazel /// /// Sends a disconnect message to the end point. /// - protected abstract bool SendDisconnect(MessageWriter writer); + protected abstract ValueTask SendDisconnect(MessageWriter writer); /// /// Called when the socket has been disconnected at the remote host. /// - protected void DisconnectRemote(string reason, MessageReader reader) + protected async ValueTask DisconnectRemote(string reason, MessageReader reader) { - if (this.SendDisconnect(null)) + if (await this.SendDisconnect(null)) { try { - InvokeDisconnected(reason, reader); + await InvokeDisconnected(reason, reader); } catch { } } @@ -65,7 +66,7 @@ namespace Impostor.Hazel /// /// Called when socket is disconnected internally /// - internal void DisconnectInternal(HazelInternalErrors error, string reason) + internal async ValueTask DisconnectInternal(HazelInternalErrors error, string reason) { var handler = this.OnInternalDisconnect; if (handler != null) @@ -75,7 +76,7 @@ namespace Impostor.Hazel { try { - Disconnect(reason, messageToRemote); + await Disconnect(reason, messageToRemote); } finally { @@ -84,25 +85,25 @@ namespace Impostor.Hazel } else { - Disconnect(reason); + await Disconnect(reason); } } else { - Disconnect(reason); + await Disconnect(reason); } } /// /// Called when the socket has been disconnected locally. /// - public override void Disconnect(string reason, MessageWriter writer = null) + public override async ValueTask Disconnect(string reason, MessageWriter writer = null) { - if (this.SendDisconnect(writer)) + if (await this.SendDisconnect(writer)) { try { - InvokeDisconnected(reason, null); + await InvokeDisconnected(reason, null); } catch { } } diff --git a/Hazel/NewConnectionEventArgs.cs b/Hazel/NewConnectionEventArgs.cs index e606e3f..be9e7a2 100644 --- a/Hazel/NewConnectionEventArgs.cs +++ b/Hazel/NewConnectionEventArgs.cs @@ -1,4 +1,6 @@ -namespace Impostor.Hazel +using Impostor.Api.Net.Messages; + +namespace Impostor.Hazel { public struct NewConnectionEventArgs { @@ -6,14 +8,14 @@ /// The data received from the client in the handshake. /// This data is yours. Remember to recycle it. /// - public readonly MessageReader HandshakeData; + public readonly IMessageReader HandshakeData; /// /// The to the new client. /// public readonly Connection Connection; - public NewConnectionEventArgs(MessageReader handshakeData, Connection connection) + public NewConnectionEventArgs(IMessageReader handshakeData, Connection connection) { this.HandshakeData = handshakeData; this.Connection = connection; diff --git a/Hazel/ObjectPool.cs b/Hazel/ObjectPoolCustom.cs similarity index 96% rename from Hazel/ObjectPool.cs rename to Hazel/ObjectPoolCustom.cs index 6d6ce6e..5c9ef9b 100644 --- a/Hazel/ObjectPool.cs +++ b/Hazel/ObjectPoolCustom.cs @@ -9,7 +9,7 @@ namespace Impostor.Hazel /// /// The type that is pooled. /// - public sealed class ObjectPool where T : IRecyclable + public sealed class ObjectPoolCustom where T : IRecyclable { private int numberCreated; public int NumberCreated { get { return numberCreated; } } @@ -36,7 +36,7 @@ namespace Impostor.Hazel /// /// Internal constructor for our ObjectPool. /// - internal ObjectPool(Func objectFactory) + internal ObjectPoolCustom(Func objectFactory) { this.objectFactory = objectFactory; } diff --git a/Hazel/SendOption.cs b/Hazel/SendOption.cs deleted file mode 100644 index cd32c95..0000000 --- a/Hazel/SendOption.cs +++ /dev/null @@ -1,32 +0,0 @@ -using System; - -namespace Impostor.Hazel -{ - /// - /// Specifies how a message should be sent between connections. - /// - [Flags] - public enum SendOption : byte - { - /// - /// Requests unreliable delivery with no framentation. - /// - /// - /// Sending data using unreliable delivery means that data is not guaranteed to arrive at it's destination nor is - /// it guarenteed to arrive only once. However, unreliable delivery can be faster than other methods and it - /// typically requires a smaller number of protocol bytes than other methods. There is also typically less - /// processing involved and less memory needed as packets are not stored once sent. - /// - None = 0, - - /// - /// Requests data be sent reliably but with no fragmentation. - /// - /// - /// Sending data reliably means that data is guarenteed to arrive and to arrive only once. Reliable delivery - /// typically requires more processing, more memory (as packets need to be stored in case they need resending), - /// a larger number of protocol bytes and can be slower than unreliable delivery. - /// - Reliable = 1, - } -} diff --git a/Hazel/Udp/UdpClientConnection.cs b/Hazel/Udp/UdpClientConnection.cs index 5a92f4c..01f003f 100644 --- a/Hazel/Udp/UdpClientConnection.cs +++ b/Hazel/Udp/UdpClientConnection.cs @@ -2,6 +2,9 @@ using System; using System.Net; using System.Net.Sockets; using System.Threading; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; +using Microsoft.Extensions.ObjectPool; namespace Impostor.Hazel.Udp { @@ -9,104 +12,73 @@ namespace Impostor.Hazel.Udp /// Represents a client's connection to a server that uses the UDP protocol. /// /// - public sealed class UdpClientConnection : UdpConnection + public class UdpClientConnection : UdpConnection { /// /// The socket we're connected via. /// - private Socket socket; + private readonly UdpClient _socket; /// /// Reset event that is triggered when the connection is marked Connected. /// - private ManualResetEvent connectWaitLock = new ManualResetEvent(false); + private readonly SemaphoreSlim _connectWaitLock; - private Timer reliablePacketTimer; + private Task _listenTask; -#if DEBUG - public event Action DataSentRaw; - public event Action DataReceivedRaw; -#endif + private Timer reliablePacketTimer; /// /// Creates a new UdpClientConnection. /// /// A to connect to. - public UdpClientConnection(IPEndPoint remoteEndPoint, IPMode ipMode = IPMode.IPv4) - : base() + public UdpClientConnection(IPEndPoint remoteEndPoint, ObjectPool readerPool, IPMode ipMode = IPMode.IPv4) : base(null, readerPool) { this.EndPoint = remoteEndPoint; this.IPMode = ipMode; - this.socket = CreateSocket(ipMode); + _socket = new UdpClient + { + DontFragment = false + }; reliablePacketTimer = new Timer(ManageReliablePacketsInternal, null, 100, Timeout.Infinite); + _connectWaitLock = new SemaphoreSlim(0, 1); + this.InitializeKeepAliveTimer(); } - + ~UdpClientConnection() { this.Dispose(false); } - private void ManageReliablePacketsInternal(object state) + private async void ManageReliablePacketsInternal(object state) { - base.ManageReliablePackets(); + await base.ManageReliablePackets(); try { reliablePacketTimer.Change(100, Timeout.Infinite); } catch { } } - - /// - protected override void WriteBytesToConnection(byte[] bytes, int length) + + protected virtual async ValueTask ResendPacketsIfNeeded() { -#if DEBUG - if (TestLagMs > 0) - { - ThreadPool.QueueUserWorkItem(a => { Thread.Sleep(this.TestLagMs); WriteBytesToConnectionReal(bytes, length); }); - } - else -#endif - { - WriteBytesToConnectionReal(bytes, length); - } + await base.ManageReliablePackets(); } - private void WriteBytesToConnectionReal(byte[] bytes, int length) + /// + protected override ValueTask WriteBytesToConnection(byte[] bytes, int length) { -#if DEBUG - DataSentRaw?.Invoke(bytes, length); -#endif - - try - { - socket.BeginSendTo( - bytes, - 0, - length, - SocketFlags.None, - EndPoint, - HandleSendTo, - null); - } - catch (NullReferenceException) { } - catch (ObjectDisposedException) - { - // Already disposed and disconnected... - } - catch (SocketException ex) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); - } + return WriteBytesToConnectionReal(bytes, length); } - private void HandleSendTo(IAsyncResult result) + private async ValueTask WriteBytesToConnectionReal(byte[] bytes, int length) { try { - socket.EndSendTo(result); + await _socket.SendAsync(bytes, length); } catch (NullReferenceException) { } catch (ObjectDisposedException) @@ -115,53 +87,36 @@ namespace Impostor.Hazel.Udp } catch (SocketException ex) { - DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); + await DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); } } /// - public override void Connect(byte[] bytes = null, int timeout = 5000) + public override async ValueTask ConnectAsync(byte[] bytes = null, int timeout = 5000) { - this.ConnectAsync(bytes); - - //Wait till hello packet is acknowledged and the state is set to Connected - bool timedOut = !WaitOnConnect(timeout); - - //If we timed out raise an exception - if (timedOut) - { - Dispose(); - throw new HazelException("Connection attempt timed out."); - } - } - - /// - public override void ConnectAsync(byte[] bytes = null) - { - this.State = ConnectionState.Connecting; + State = ConnectionState.Connecting; try { - if (IPMode == IPMode.IPv4) - socket.Bind(new IPEndPoint(IPAddress.Any, 0)); - else - socket.Bind(new IPEndPoint(IPAddress.IPv6Any, 0)); + _socket.Connect(EndPoint); } catch (SocketException e) { - this.State = ConnectionState.NotConnected; + State = ConnectionState.NotConnected; throw new HazelException("A SocketException occurred while binding to the port.", e); } + + this.RestartConnection(); try { - StartListeningForData(); + _listenTask = Task.Factory.StartNew(ListenAsync, TaskCreationOptions.LongRunning); } catch (ObjectDisposedException) { // If the socket's been disposed then we can just end there but make sure we're in NotConnected state. // If we end up here I'm really lost... - this.State = ConnectionState.NotConnected; + State = ConnectionState.NotConnected; return; } catch (SocketException e) @@ -172,127 +127,66 @@ namespace Impostor.Hazel.Udp // Write bytes to the server to tell it hi (and to punch a hole in our NAT, if present) // When acknowledged set the state to connected - SendHello(bytes, () => + await SendHello(bytes, () => { - this.State = ConnectionState.Connected; - this.InitializeKeepAliveTimer(); + State = ConnectionState.Connected; + InitializeKeepAliveTimer(); }); - } - - /// - /// Instructs the listener to begin listening. - /// - void StartListeningForData() - { -#if DEBUG - if (this.TestLagMs > 0) - { - Thread.Sleep(this.TestLagMs); - } -#endif - - var msg = MessageReader.GetSized(ushort.MaxValue); - try - { - socket.BeginReceive(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, ReadCallback, msg); - } - catch - { - msg.Recycle(); - this.Dispose(); - } - } - - protected override void SetState(ConnectionState state) - { - try - { - if (state == ConnectionState.Connected) - connectWaitLock.Set(); - else - connectWaitLock.Reset(); - } - catch (ObjectDisposedException) - { - } + await _connectWaitLock.WaitAsync(TimeSpan.FromSeconds(10)); } - /// - /// Blocks until the Connection is connected. - /// - /// The number of milliseconds to wait before timing out. - public bool WaitOnConnect(int timeout) + protected virtual void RestartConnection() { - return connectWaitLock.WaitOne(timeout); } - /// - /// Called when data has been received by the socket. - /// - /// The asyncronous operation's result. - void ReadCallback(IAsyncResult result) + private async Task ListenAsync() { - var msg = (MessageReader)result.AsyncState; + // Start packet handler. + await StartAsync(); - try + // Listen. + while (State != ConnectionState.NotConnected) { - msg.Length = socket.EndReceive(result); - } - catch (SocketException e) - { - msg.Recycle(); - DisconnectInternal(HazelInternalErrors.SocketExceptionReceive, "Socket exception while reading data: " + e.Message); - return; - } - catch (Exception) - { - msg.Recycle(); - return; - } + UdpReceiveResult data; - //Exit if no bytes read, we've failed. - if (msg.Length == 0) - { - msg.Recycle(); - DisconnectInternal(HazelInternalErrors.ReceivedZeroBytes, "Received 0 bytes"); - return; - } - - //Begin receiving again - try - { - StartListeningForData(); - } - catch (SocketException e) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionReceive, "Socket exception during receive: " + e.Message); - } - catch (ObjectDisposedException) - { - //If the socket's been disposed then we can just end there. - return; - } + try + { + data = await _socket.ReceiveAsync(); + } + catch (SocketException e) + { + await DisconnectInternal(HazelInternalErrors.SocketExceptionReceive, "Socket exception while reading data: " + e.Message); + return; + } + catch (Exception) + { + return; + } -#if DEBUG - if (this.TestDropRate > 0) - { - if ((this.testDropCount++ % this.TestDropRate) == 0) + if (data.Buffer.Length == 0) { + await DisconnectInternal(HazelInternalErrors.ReceivedZeroBytes, "Received 0 bytes"); return; } - } - DataReceivedRaw?.Invoke(msg.Buffer, msg.Length); -#endif - HandleReceive(msg, msg.Length); + // Write to client. + await Pipeline.Writer.WriteAsync(data.Buffer); + } } + protected override void SetState(ConnectionState state) + { + if (state == ConnectionState.Connected) + { + _connectWaitLock.Release(); + } + } /// /// Sends a disconnect message to the end point. /// You may include optional disconnect data. The SendOption must be unreliable. /// - protected override bool SendDisconnect(MessageWriter data = null) + protected override async ValueTask SendDisconnect(MessageWriter data = null) { lock (this) { @@ -303,7 +197,7 @@ namespace Impostor.Hazel.Udp var bytes = EmptyDisconnectBytes; if (data != null && data.Length > 0) { - if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable."); + if (data.SendOption != MessageType.Unreliable) throw new ArgumentException("Disconnect messages can only be unreliable."); bytes = data.ToByteArray(true); bytes[0] = (byte)UdpSendOption.Disconnect; @@ -311,12 +205,7 @@ namespace Impostor.Hazel.Udp try { - socket.SendTo( - bytes, - 0, - bytes.Length, - SocketFlags.None, - EndPoint); + await _socket.SendAsync(bytes, bytes.Length, EndPoint); } catch { } @@ -326,17 +215,16 @@ namespace Impostor.Hazel.Udp /// protected override void Dispose(bool disposing) { - if (disposing) - { - SendDisconnect(); - } + State = ConnectionState.NotConnected; - try { this.socket.Shutdown(SocketShutdown.Both); } catch { } - try { this.socket.Close(); } catch { } - try { this.socket.Dispose(); } catch { } + try { _socket.Close(); } + catch { } + + try { _socket.Dispose(); } + catch { } - this.reliablePacketTimer.Dispose(); - this.connectWaitLock.Dispose(); + reliablePacketTimer.Dispose(); + _connectWaitLock.Dispose(); base.Dispose(disposing); } diff --git a/Hazel/Udp/UdpConnection.KeepAlive.cs b/Hazel/Udp/UdpConnection.KeepAlive.cs index a145534..c8d6bb5 100644 --- a/Hazel/Udp/UdpConnection.KeepAlive.cs +++ b/Hazel/Udp/UdpConnection.KeepAlive.cs @@ -2,6 +2,7 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.Threading; +using System.Threading.Tasks; namespace Impostor.Hazel.Udp { @@ -13,7 +14,7 @@ namespace Impostor.Hazel.Udp /// public class PingPacket : IRecyclable { - private static readonly ObjectPool PacketPool = new ObjectPool(() => new PingPacket()); + private static readonly ObjectPoolCustom PacketPool = new ObjectPoolCustom(() => new PingPacket()); public readonly Stopwatch Stopwatch = new Stopwatch(); @@ -80,21 +81,21 @@ namespace Impostor.Hazel.Udp ); } - private void HandleKeepAlive(object state) + private async void HandleKeepAlive(object state) { if (this.State != ConnectionState.Connected) return; if (this.pingsSinceAck >= this.MissingPingsUntilDisconnect) { this.DisposeKeepAliveTimer(); - this.DisconnectInternal(HazelInternalErrors.PingsWithoutResponse, $"Sent {this.pingsSinceAck} pings that remote has not responded to."); + await this.DisconnectInternal(HazelInternalErrors.PingsWithoutResponse, $"Sent {this.pingsSinceAck} pings that remote has not responded to."); return; } try { - this.pingsSinceAck++; - SendPing(); + Interlocked.Increment(ref pingsSinceAck); + await SendPing(); } catch { @@ -106,7 +107,7 @@ namespace Impostor.Hazel.Udp // An unacked ping should never be the sole cause of a disconnect. // Rather, the responses will reset our pingsSinceAck, enough unacked // pings should cause a disconnect. - private void SendPing() + private async ValueTask SendPing() { ushort id = (ushort)Interlocked.Increment(ref lastIDAllocated); @@ -127,7 +128,7 @@ namespace Impostor.Hazel.Udp pkt.Stopwatch.Restart(); - WriteBytesToConnection(bytes, bytes.Length); + await WriteBytesToConnection(bytes, bytes.Length); Statistics.LogReliableSend(0, bytes.Length); } @@ -163,4 +164,4 @@ namespace Impostor.Hazel.Udp } } } -} \ No newline at end of file +} diff --git a/Hazel/Udp/UdpConnection.Reliable.cs b/Hazel/Udp/UdpConnection.Reliable.cs index d4f16be..7cf3858 100644 --- a/Hazel/Udp/UdpConnection.Reliable.cs +++ b/Hazel/Udp/UdpConnection.Reliable.cs @@ -3,6 +3,8 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Threading; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; namespace Impostor.Hazel.Udp { @@ -85,7 +87,7 @@ namespace Impostor.Hazel.Udp /// /// Object pool for this event. /// - public static readonly ObjectPool PacketPool = new ObjectPool(() => new Packet()); + public static readonly ObjectPoolCustom PacketPool = new ObjectPoolCustom(() => new Packet()); /// /// Returns an instance of this object from the pool. @@ -129,7 +131,7 @@ namespace Impostor.Hazel.Udp } // Packets resent - public int Resend() + public async ValueTask Resend() { var connection = this.Connection; if (!this.Acknowledged && connection != null) @@ -139,7 +141,7 @@ namespace Impostor.Hazel.Udp { if (connection.reliableDataPacketsSent.TryRemove(this.Id, out Packet self)) { - connection.DisconnectInternal(HazelInternalErrors.ReliablePacketWithoutResponse, $"Reliable packet {self.Id} (size={this.Length}) was not ack'd after {lifetime}ms ({self.Retransmissions} resends)"); + await connection.DisconnectInternal(HazelInternalErrors.ReliablePacketWithoutResponse, $"Reliable packet {self.Id} (size={this.Length}) was not ack'd after {lifetime}ms ({self.Retransmissions} resends)"); self.Recycle(); } @@ -155,7 +157,7 @@ namespace Impostor.Hazel.Udp { if (connection.reliableDataPacketsSent.TryRemove(this.Id, out Packet self)) { - connection.DisconnectInternal(HazelInternalErrors.ReliablePacketWithoutResponse, $"Reliable packet {self.Id} (size={this.Length}) was not ack'd after {self.Retransmissions} resends ({lifetime}ms)"); + await connection.DisconnectInternal(HazelInternalErrors.ReliablePacketWithoutResponse, $"Reliable packet {self.Id} (size={this.Length}) was not ack'd after {self.Retransmissions} resends ({lifetime}ms)"); self.Recycle(); } @@ -166,13 +168,13 @@ namespace Impostor.Hazel.Udp this.NextTimeout += (int)Math.Min(this.NextTimeout * connection.ResendPingMultiplier, 1000); try { - connection.WriteBytesToConnection(this.Data, this.Length); + await connection.WriteBytesToConnection(this.Data, this.Length); connection.Statistics.LogMessageResent(); return 1; } catch (InvalidOperationException) { - connection.DisconnectInternal(HazelInternalErrors.ConnectionDisconnected, "Could not resend data as connection is no longer connected"); + await connection.DisconnectInternal(HazelInternalErrors.ConnectionDisconnected, "Could not resend data as connection is no longer connected"); } } } @@ -192,7 +194,7 @@ namespace Impostor.Hazel.Udp } } - internal int ManageReliablePackets() + internal async ValueTask ManageReliablePackets() { int output = 0; if (this.reliableDataPacketsSent.Count > 0) @@ -203,7 +205,7 @@ namespace Impostor.Hazel.Udp try { - output += pkt.Resend(); + output += await pkt.Resend(); } catch { } } @@ -253,7 +255,7 @@ namespace Impostor.Hazel.Udp /// /// The byte array to write to. /// The callback to make once the packet has been acknowledged. - private void ReliableSend(byte sendOption, byte[] data, Action ackCallback = null) + private async ValueTask ReliableSend(byte sendOption, byte[] data, Action ackCallback = null) { //Inform keepalive not to send for a while ResetKeepAliveTimer(); @@ -270,7 +272,7 @@ namespace Impostor.Hazel.Udp Buffer.BlockCopy(data, 0, bytes, bytes.Length - data.Length, data.Length); //Write to connection - WriteBytesToConnection(bytes, bytes.Length); + await WriteBytesToConnection(bytes, bytes.Length); Statistics.LogReliableSend(data.Length, bytes.Length); } @@ -279,16 +281,11 @@ namespace Impostor.Hazel.Udp /// Handles a reliable message being received and invokes the data event. /// /// The buffer received. - private void ReliableMessageReceive(MessageReader message, int bytesReceived) + private async ValueTask ReliableMessageReceive(MessageReader message, int bytesReceived) { - ushort id; - if (ProcessReliableReceive(message.Buffer, 1, out id)) + if (await ProcessReliableReceive(message.Buffer, 1)) { - InvokeDataReceived(SendOption.Reliable, message, 3, bytesReceived); - } - else - { - message.Recycle(); + await InvokeDataReceived(MessageType.Reliable, message, 3, bytesReceived); } Statistics.LogReliableReceive(message.Length - 3, message.Length); @@ -300,13 +297,13 @@ namespace Impostor.Hazel.Udp /// The buffer containing the data. /// The offset of the reliable header. /// Whether the packet was a new packet or not. - private bool ProcessReliableReceive(byte[] bytes, int offset, out ushort id) + private async ValueTask ProcessReliableReceive(ReadOnlyMemory bytes, int offset) { - byte b1 = bytes[offset]; - byte b2 = bytes[offset + 1]; + var b1 = bytes.Span[offset]; + var b2 = bytes.Span[offset + 1]; //Get the ID form the packet - id = (ushort)((b1 << 8) + b2); + var id = (ushort)((b1 << 8) + b2); /* * It gets a little complicated here (note the fact I'm actually using a multiline comment for once...) @@ -383,7 +380,7 @@ namespace Impostor.Hazel.Udp } // Send an acknowledgement - SendAck(id); + await SendAck(id); return result; } @@ -449,7 +446,7 @@ namespace Impostor.Hazel.Udp /// /// The first identification byte. /// The second identification byte. - private void SendAck(ushort id) + private async ValueTask SendAck(ushort id) { byte recentPackets = 0; lock (this.reliableDataPacketsMissing) @@ -473,7 +470,7 @@ namespace Impostor.Hazel.Udp try { - WriteBytesToConnection(bytes, bytes.Length); + await WriteBytesToConnection(bytes, bytes.Length); } catch (InvalidOperationException) { } } diff --git a/Hazel/Udp/UdpConnection.cs b/Hazel/Udp/UdpConnection.cs index d3c1d29..5221d88 100644 --- a/Hazel/Udp/UdpConnection.cs +++ b/Hazel/Udp/UdpConnection.cs @@ -1,5 +1,10 @@ using System; -using System.Net.Sockets; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; +using Microsoft.Extensions.ObjectPool; +using Serilog; namespace Impostor.Hazel.Udp { @@ -9,52 +14,111 @@ namespace Impostor.Hazel.Udp /// public abstract partial class UdpConnection : NetworkConnection { + private static readonly ILogger Logger = Log.ForContext(); + public override float AveragePingMs => this._pingMs; private const int SioUdpConnectionReset = -1744830452; public static readonly byte[] EmptyDisconnectBytes = new byte[] { (byte)UdpSendOption.Disconnect }; - internal static Socket CreateSocket(IPMode ipMode) + private readonly ConnectionListener _listener; + protected readonly ObjectPool _readerPool; + private readonly CancellationTokenSource _stoppingCts; + + private bool _isDisposing; + private bool _isFirst = true; + private Task _executingTask; + + protected UdpConnection(ConnectionListener listener, ObjectPool readerPool) { - Socket socket; - if (ipMode == IPMode.IPv4) + _listener = listener; + _readerPool = readerPool; + _stoppingCts = new CancellationTokenSource(); + + Pipeline = Channel.CreateUnbounded(new UnboundedChannelOptions { - socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); - } - else + SingleReader = true, + SingleWriter = true + }); + } + + internal Channel Pipeline { get; } + + public Task StartAsync() + { + // Store the task we're executing + _executingTask = Task.Factory.StartNew(ReadAsync, TaskCreationOptions.LongRunning); + + // If the task is completed then return it, this will bubble cancellation and failure to the caller + if (_executingTask.IsCompleted) { - if (!Socket.OSSupportsIPv6) - throw new InvalidOperationException("IPV6 not supported!"); + return _executingTask; + } + + // Otherwise it's running + return Task.CompletedTask; + } - socket = new Socket(AddressFamily.InterNetworkV6, SocketType.Dgram, ProtocolType.Udp); - socket.SetSocketOption(SocketOptionLevel.IPv6, SocketOptionName.IPv6Only, false); + public void Stop() + { + // Stop called without start + if (_executingTask == null) + { + return; } + // Signal cancellation to methods. + _stoppingCts.Cancel(); + try { - socket.DontFragment = false; + // Cancel reader. + Pipeline.Writer.Complete(); + } + catch (ChannelClosedException) + { + // Already done. } - catch { } - try + // Remove references. + if (!_isDisposing) { - const int SIO_UDP_CONNRESET = -1744830452; - socket.IOControl(SIO_UDP_CONNRESET, new byte[1], null); + Dispose(true); } - catch { } // Only necessary on Windows + } + + private async Task ReadAsync() + { + var reader = new MessageReader(_readerPool); - return socket; + while (!_stoppingCts.IsCancellationRequested) + { + var result = await Pipeline.Reader.ReadAsync(_stoppingCts.Token); + + try + { + reader.Update(result); + + await HandleReceive(reader, reader.Length); + } + catch (Exception e) + { + Logger.Error(e, "Exception during ReadAsync"); + Dispose(true); + break; + } + } } /// /// Writes the given bytes to the connection. /// /// The bytes to write. - protected abstract void WriteBytesToConnection(byte[] bytes, int length); + protected abstract ValueTask WriteBytesToConnection(byte[] bytes, int length); /// - public override void Send(MessageWriter msg) + public override async ValueTask SendAsync(IMessageWriter msg) { if (this._state != ConnectionState.Connected) throw new InvalidOperationException("Could not send data as this Connection is not connected. Did you disconnect?"); @@ -64,16 +128,16 @@ namespace Impostor.Hazel.Udp switch (msg.SendOption) { - case SendOption.Reliable: + case MessageType.Reliable: ResetKeepAliveTimer(); AttachReliableID(buffer, 1); - WriteBytesToConnection(buffer, buffer.Length); + await WriteBytesToConnection(buffer, buffer.Length); Statistics.LogReliableSend(buffer.Length - 3, buffer.Length); break; default: - WriteBytesToConnection(buffer, buffer.Length); + await WriteBytesToConnection(buffer, buffer.Length); Statistics.LogUnreliableSend(buffer.Length - 1, buffer.Length); break; } @@ -83,17 +147,17 @@ namespace Impostor.Hazel.Udp /// /// /// - /// Udp connections can currently send messages using and - /// . Fragmented messages are not currently supported and will default to - /// until implemented. + /// Udp connections can currently send messages using and + /// . Fragmented messages are not currently supported and will default to + /// until implemented. /// /// - public override void SendBytes(byte[] bytes, SendOption sendOption = SendOption.None) + public override async ValueTask SendBytes(byte[] bytes, MessageType sendOption = MessageType.Unreliable) { //Add header information and send - HandleSend(bytes, (byte)sendOption); + await HandleSend(bytes, (byte)sendOption); } - + /// /// Handles the reliable/fragmented sending from this connection. /// @@ -101,19 +165,19 @@ namespace Impostor.Hazel.Udp /// The specified as its byte value. /// The callback to invoke when this packet is acknowledged. /// The bytes that should actually be sent. - protected void HandleSend(byte[] data, byte sendOption, Action ackCallback = null) + protected async ValueTask HandleSend(byte[] data, byte sendOption, Action ackCallback = null) { switch (sendOption) { case (byte)UdpSendOption.Ping: - case (byte)SendOption.Reliable: + case (byte)MessageType.Reliable: case (byte)UdpSendOption.Hello: - ReliableSend(sendOption, data, ackCallback); + await ReliableSend(sendOption, data, ackCallback); break; - + //Treat all else as unreliable default: - UnreliableSend(sendOption, data); + await UnreliableSend(sendOption, data); break; } } @@ -122,43 +186,54 @@ namespace Impostor.Hazel.Udp /// Handles the receiving of data. /// /// The buffer containing the bytes received. - protected internal virtual void HandleReceive(MessageReader message, int bytesReceived) + internal virtual async ValueTask HandleReceive(MessageReader message, int bytesReceived) { - ushort id; + // Check if the first message received is the hello packet. + if (_isFirst) + { + _isFirst = false; + + // Slice 4 bytes to get handshake data. + if (_listener != null) + { + using (var handshake = message.Copy(4)) + { + await _listener.InvokeNewConnection(handshake, this); + } + } + } + switch (message.Buffer[0]) { //Handle reliable receives - case (byte)SendOption.Reliable: - ReliableMessageReceive(message, bytesReceived); + case (byte)MessageType.Reliable: + await ReliableMessageReceive(message, bytesReceived); break; //Handle acknowledgments case (byte)UdpSendOption.Acknowledgement: AcknowledgementMessageReceive(message.Buffer, bytesReceived); - message.Recycle(); break; //We need to acknowledge hello and ping messages but dont want to invoke any events! case (byte)UdpSendOption.Ping: - ProcessReliableReceive(message.Buffer, 1, out id); + await ProcessReliableReceive(message.Buffer, 1); Statistics.LogHelloReceive(bytesReceived); - message.Recycle(); break; case (byte)UdpSendOption.Hello: - ProcessReliableReceive(message.Buffer, 1, out id); + await ProcessReliableReceive(message.Buffer, 1); Statistics.LogHelloReceive(bytesReceived); break; case (byte)UdpSendOption.Disconnect: message.Offset = 1; message.Position = 0; - DisconnectRemote("The remote sent a disconnect request", message); - message.Recycle(); + await DisconnectRemote("The remote sent a disconnect request", message); break; - + //Treat everything else as unreliable default: - InvokeDataReceived(SendOption.None, message, 1, bytesReceived); + await InvokeDataReceived(MessageType.Unreliable, message, 1, bytesReceived); Statistics.LogUnreliableReceive(bytesReceived - 1, bytesReceived); break; } @@ -169,9 +244,9 @@ namespace Impostor.Hazel.Udp /// /// The SendOption to attach. /// The data. - void UnreliableSend(byte sendOption, byte[] data) + ValueTask UnreliableSend(byte sendOption, byte[] data) { - this.UnreliableSend(sendOption, data, 0, data.Length); + return this.UnreliableSend(sendOption, data, 0, data.Length); } /// @@ -181,7 +256,7 @@ namespace Impostor.Hazel.Udp /// The SendOption to attach. /// /// - void UnreliableSend(byte sendOption, byte[] data, int offset, int length) + async ValueTask UnreliableSend(byte sendOption, byte[] data, int offset, int length) { byte[] bytes = new byte[length + 1]; @@ -192,7 +267,7 @@ namespace Impostor.Hazel.Udp Buffer.BlockCopy(data, offset, bytes, bytes.Length - length, length); //Write to connection - WriteBytesToConnection(bytes, bytes.Length); + await WriteBytesToConnection(bytes, bytes.Length); Statistics.LogUnreliableSend(length, bytes.Length); } @@ -203,20 +278,20 @@ namespace Impostor.Hazel.Udp /// The send option the message was received with. /// The buffer received. /// The offset of data in the buffer. - void InvokeDataReceived(SendOption sendOption, MessageReader buffer, int dataOffset, int bytesReceived) + ValueTask InvokeDataReceived(MessageType sendOption, MessageReader buffer, int dataOffset, int bytesReceived) { buffer.Offset = dataOffset; buffer.Length = bytesReceived - dataOffset; buffer.Position = 0; - InvokeDataReceived(buffer, sendOption); + return InvokeDataReceived(buffer, sendOption); } /// /// Sends a hello packet to the remote endpoint. /// /// The callback to invoke when the hello packet is acknowledged. - protected void SendHello(byte[] bytes, Action acknowledgeCallback) + protected ValueTask SendHello(byte[] bytes, Action acknowledgeCallback) { //First byte of handshake is version indicator so add data after byte[] actualBytes; @@ -230,14 +305,17 @@ namespace Impostor.Hazel.Udp Buffer.BlockCopy(bytes, 0, actualBytes, 1, bytes.Length); } - HandleSend(actualBytes, (byte)UdpSendOption.Hello, acknowledgeCallback); + return HandleSend(actualBytes, (byte)UdpSendOption.Hello, acknowledgeCallback); } - + /// protected override void Dispose(bool disposing) { if (disposing) { + _isDisposing = true; + + Stop(); DisposeKeepAliveTimer(); DisposeReliablePackets(); } diff --git a/Hazel/Udp/UdpConnectionListener.cs b/Hazel/Udp/UdpConnectionListener.cs index a74da2a..23c7a6a 100644 --- a/Hazel/Udp/UdpConnectionListener.cs +++ b/Hazel/Udp/UdpConnectionListener.cs @@ -3,6 +3,9 @@ using System.Collections.Concurrent; using System.Net; using System.Net.Sockets; using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.ObjectPool; +using Serilog; namespace Impostor.Hazel.Udp { @@ -12,8 +15,7 @@ namespace Impostor.Hazel.Udp /// public class UdpConnectionListener : NetworkConnectionListener { - private const int SendReceiveBufferSize = 1024 * 1024; - private const int BufferSize = ushort.MaxValue; + private static readonly ILogger Logger = Log.ForContext(); /// /// A callback for early connection rejection. @@ -21,257 +23,193 @@ namespace Impostor.Hazel.Udp /// * A null response is ok, we just won't send anything. /// public AcceptConnectionCheck AcceptConnection; - public delegate bool AcceptConnectionCheck(IPEndPoint endPoint, byte[] input, out byte[] response); - private Socket socket; - private Action Logger; - private Timer reliablePacketTimer; + public delegate bool AcceptConnectionCheck(IPEndPoint endPoint, byte[] input, out byte[] response); - private ConcurrentDictionary allConnections = new ConcurrentDictionary(); - - public int ConnectionCount { get { return this.allConnections.Count; } } + private readonly UdpClient _socket; + protected readonly ObjectPool _readerPool; + private readonly Timer _reliablePacketTimer; + private readonly ConcurrentDictionary _allConnections; + private readonly CancellationTokenSource _stoppingCts; + private readonly UdpConnectionRateLimit _connectionRateLimit; + private Task _executingTask; /// /// Creates a new UdpConnectionListener for the given , port and . /// /// The endpoint to listen on. - public UdpConnectionListener(IPEndPoint endPoint, IPMode ipMode = IPMode.IPv4, Action logger = null) + public UdpConnectionListener(IPEndPoint endPoint, ObjectPool readerPool, IPMode ipMode = IPMode.IPv4) { - this.Logger = logger; this.EndPoint = endPoint; this.IPMode = ipMode; - this.socket = UdpConnection.CreateSocket(this.IPMode); - - socket.ReceiveBufferSize = SendReceiveBufferSize; - socket.SendBufferSize = SendReceiveBufferSize; - - reliablePacketTimer = new Timer(ManageReliablePackets, null, 100, Timeout.Infinite); - } + _readerPool = readerPool; + _socket = new UdpClient(endPoint); - ~UdpConnectionListener() - { - this.Dispose(false); + try + { + _socket.DontFragment = false; + } + catch (SocketException) + { + } + + _reliablePacketTimer = new Timer(ManageReliablePackets, null, 100, Timeout.Infinite); + + _allConnections = new ConcurrentDictionary(); + + _stoppingCts = new CancellationTokenSource(); + _stoppingCts.Token.Register(() => + { + _socket.Dispose(); + }); + + _connectionRateLimit = new UdpConnectionRateLimit(); } - - private void ManageReliablePackets(object state) + + private async void ManageReliablePackets(object state) { - foreach (var kvp in this.allConnections) + foreach (var kvp in this._allConnections) { var sock = kvp.Value; - sock.ManageReliablePackets(); + await sock.ManageReliablePackets(); } try { - this.reliablePacketTimer.Change(100, Timeout.Infinite); + this._reliablePacketTimer.Change(100, Timeout.Infinite); } catch { } } /// - public override void Start() + public override Task StartAsync() { - try - { - socket.Bind(EndPoint); - } - catch (SocketException e) + // Store the task we're executing + _executingTask = Task.Factory.StartNew(ListenAsync, TaskCreationOptions.LongRunning); + + // If the task is completed then return it, this will bubble cancellation and failure to the caller + if (_executingTask.IsCompleted) { - throw new HazelException("Could not start listening as a SocketException occurred", e); + return _executingTask; } - StartListeningForData(); + // Otherwise it's running + return Task.CompletedTask; } - /// - /// Instructs the listener to begin listening. - /// - private void StartListeningForData() + private async Task StopAsync() { - EndPoint remoteEP = EndPoint; - - MessageReader message = null; - try + // Stop called without start + if (_executingTask == null) { - message = MessageReader.GetSized(BufferSize); - socket.BeginReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP, ReadCallback, message); + return; } - catch (SocketException sx) - { - message?.Recycle(); - this.Logger?.Invoke("Socket Ex in StartListening: " + sx.Message); - - Thread.Sleep(10); - StartListeningForData(); - return; + try + { + // Signal cancellation to the executing method + _stoppingCts.Cancel(); } - catch (Exception ex) + finally { - message.Recycle(); - this.Logger?.Invoke("Stopped due to: " + ex.Message); - return; + // Wait until the task completes or the timeout triggers + await Task.WhenAny(_executingTask, Task.Delay(TimeSpan.FromSeconds(5))); } } - void ReadCallback(IAsyncResult result) + /// + /// Instructs the listener to begin listening. + /// + private async Task ListenAsync() { - var message = (MessageReader)result.AsyncState; - int bytesReceived; - EndPoint remoteEndPoint = new IPEndPoint(this.EndPoint.Address, this.EndPoint.Port); - - //End the receive operation try { - bytesReceived = socket.EndReceiveFrom(result, ref remoteEndPoint); + while (!_stoppingCts.IsCancellationRequested) + { + UdpReceiveResult data; - message.Offset = 0; - message.Length = bytesReceived; - } - catch (ObjectDisposedException) - { - message.Recycle(); - return; - } - catch (SocketException sx) - { - // Client no longer reachable, pretend it didn't happen - // TODO should this not inform the connection this client is lost??? + try + { + data = await _socket.ReceiveAsync(); - // This thread suggests the IP is not passed out from WinSoc so maybe not possible - // http://stackoverflow.com/questions/2576926/python-socket-error-on-udp-data-receive-10054 - message.Recycle(); - this.Logger?.Invoke($"Socket Ex {sx.SocketErrorCode} in ReadCallback: {sx.Message}"); + if (data.Buffer.Length == 0) + { + Logger.Fatal("Hazel read 0 bytes from UDP server socket."); + continue; + } + } + catch (SocketException) + { + // Client no longer reachable, pretend it didn't happen + continue; + } + catch (ObjectDisposedException) + { + // Socket was disposed, don't care. + return; + } - Thread.Sleep(10); - StartListeningForData(); - return; + await ProcessData(data); + } } - catch (Exception ex) + catch (Exception e) { - //If the socket's been disposed then we can just end there. - message.Recycle(); - this.Logger?.Invoke("Stopped due to: " + ex.Message); - return; + Logger.Error(e, "Listen loop error"); } + } - // I'm a little concerned about a infinite loop here, but it seems like it's possible - // to get 0 bytes read on UDP without the socket being shut down. - if (bytesReceived == 0) + protected virtual async ValueTask ProcessData(UdpReceiveResult data) + { + // Get client from active clients + if (!_allConnections.TryGetValue(data.RemoteEndPoint, out var client)) { - message.Recycle(); - this.Logger?.Invoke("Received 0 bytes"); - Thread.Sleep(10); - StartListeningForData(); - return; - } - - //Begin receiving again - StartListeningForData(); - - bool aware = true; - bool isHello = message.Buffer[0] == (byte)UdpSendOption.Hello; + // Check for malformed connection attempts + if (data.Buffer[0] != (byte)UdpSendOption.Hello) + { + return; + } - // If we're aware of this connection use the one already - // If this is a new client then connect with them! - UdpServerConnection connection; - if (!this.allConnections.TryGetValue(remoteEndPoint, out connection)) - { - lock (this.allConnections) + // Check rateLimit. + if (!_connectionRateLimit.IsAllowed(data.RemoteEndPoint.Address)) { - if (!this.allConnections.TryGetValue(remoteEndPoint, out connection)) - { - // Check for malformed connection attempts - if (!isHello) - { - message.Recycle(); - return; - } + Logger.Warning("Ratelimited connection attempt from {0}.", data.RemoteEndPoint); + return; + } - if (AcceptConnection != null) - { - if (!AcceptConnection((IPEndPoint)remoteEndPoint, message.Buffer, out var response)) - { - message.Recycle(); - if (response != null) - { - SendData(response, response.Length, remoteEndPoint); - } - - return; - } - } + // Create new client + client = new UdpServerConnection(this, data.RemoteEndPoint, IPMode, _readerPool); - aware = false; - connection = new UdpServerConnection(this, (IPEndPoint)remoteEndPoint, this.IPMode); - if (!this.allConnections.TryAdd(remoteEndPoint, connection)) - { - throw new HazelException("Failed to add a connection. This should never happen."); - } - } + // Store the client + if (!_allConnections.TryAdd(data.RemoteEndPoint, client)) + { + throw new HazelException("Failed to add a connection. This should never happen."); } - } - // If it's a new connection invoke the NewConnection event. - // This needs to happen before handling the message because in localhost scenarios, the ACK and - // subsequent messages can happen before the NewConnection event sets up OnDataRecieved handlers - if (!aware) - { - // Skip header and hello byte; - message.Offset = 4; - message.Length = bytesReceived - 4; - message.Position = 0; - InvokeNewConnection(message, connection); + // Activate the reader loop of the client + await client.StartAsync(); } - // Inform the connection of the buffer (new connections need to send an ack back to client) - connection.HandleReceive(message, bytesReceived); - - if (aware && isHello) - { - message.Recycle(); - } + // Write to client. + await client.Pipeline.Writer.WriteAsync(data.Buffer); } -#if DEBUG - public int TestDropRate = -1; - private int dropCounter = 0; -#endif - /// /// Sends data from the listener socket. /// /// The bytes to send. /// The endpoint to send to. - internal void SendData(byte[] bytes, int length, EndPoint endPoint) + internal virtual async ValueTask SendData(byte[] bytes, int length, IPEndPoint endPoint) { if (length > bytes.Length) return; -#if DEBUG - if (TestDropRate > 0) - { - if (Interlocked.Increment(ref dropCounter) % TestDropRate == 0) - { - return; - } - } -#endif - try { - socket.BeginSendTo( - bytes, - 0, - length, - SocketFlags.None, - endPoint, - SendCallback, - null); + await _socket.SendAsync(bytes, length, endPoint); } catch (SocketException e) { - this.Logger?.Invoke("Could not send data as a SocketException occurred: " + e); + Logger.Error(e, "Could not send data as a SocketException occurred"); } catch (ObjectDisposedException) { @@ -280,59 +218,30 @@ namespace Impostor.Hazel.Udp } } - private void SendCallback(IAsyncResult result) - { - try - { - socket.EndSendTo(result); - } - catch { } - } - - /// - /// Sends data from the listener socket. - /// - /// The bytes to send. - /// The endpoint to send to. - internal void SendDataSync(byte[] bytes, int length, EndPoint endPoint) - { - try - { - socket.SendTo( - bytes, - 0, - length, - SocketFlags.None, - endPoint - ); - } - catch { } - } - /// /// Removes a virtual connection from the list. /// /// The endpoint of the virtual connection. internal void RemoveConnectionTo(EndPoint endPoint) { - this.allConnections.TryRemove(endPoint, out var conn); + this._allConnections.TryRemove(endPoint, out var conn); } /// - protected override void Dispose(bool disposing) + public override async ValueTask DisposeAsync() { - foreach (var kvp in this.allConnections) + foreach (var kvp in _allConnections) { kvp.Value.Dispose(); } - try { this.socket.Shutdown(SocketShutdown.Both); } catch { } - try { this.socket.Close(); } catch { } - try { this.socket.Dispose(); } catch { } + await StopAsync(); + + await _reliablePacketTimer.DisposeAsync(); - this.reliablePacketTimer.Dispose(); + _connectionRateLimit.Dispose(); - base.Dispose(disposing); + await base.DisposeAsync(); } } } diff --git a/Hazel/Udp/UdpConnectionRateLimit.cs b/Hazel/Udp/UdpConnectionRateLimit.cs new file mode 100644 index 0000000..64881d3 --- /dev/null +++ b/Hazel/Udp/UdpConnectionRateLimit.cs @@ -0,0 +1,75 @@ +using System; +using System.Collections.Concurrent; +using System.Net; +using System.Threading; +using Serilog; + +namespace Impostor.Hazel.Udp +{ + public class UdpConnectionRateLimit : IDisposable + { + private static readonly ILogger Logger = Log.ForContext(); + + // Allow burst to 5 connections. + // Decrease by 1 every second. + private const int MaxConnections = 5; + private const int FalloffMs = 1000; + + private readonly ConcurrentDictionary _connectionCount; + private readonly Timer _timer; + private bool _isDisposed; + + public UdpConnectionRateLimit() + { + _connectionCount = new ConcurrentDictionary(); + _timer = new Timer(UpdateRateLimit, null, FalloffMs, Timeout.Infinite); + } + + private void UpdateRateLimit(object state) + { + try + { + foreach (var pair in _connectionCount) + { + var count = pair.Value - 1; + if (count > 0) + { + _connectionCount.TryUpdate(pair.Key, count, pair.Value); + } + else + { + _connectionCount.TryRemove(pair); + } + } + } + catch (Exception e) + { + Logger.Error(e, "Exception caught in UpdateRateLimit."); + } + finally + { + if (!_isDisposed) + { + _timer.Change(FalloffMs, Timeout.Infinite); + } + } + } + + public bool IsAllowed(IPAddress key) + { + if (_connectionCount.TryGetValue(key, out var value) && value >= MaxConnections) + { + return false; + } + + _connectionCount.AddOrUpdate(key, _ => 1, (_, i) => i + 1); + return true; + } + + public void Dispose() + { + _isDisposed = true; + _timer.Dispose(); + } + } +} \ No newline at end of file diff --git a/Hazel/Udp/UdpServerConnection.cs b/Hazel/Udp/UdpServerConnection.cs index 00d5e70..6511375 100644 --- a/Hazel/Udp/UdpServerConnection.cs +++ b/Hazel/Udp/UdpServerConnection.cs @@ -1,5 +1,8 @@ using System; using System.Net; +using System.Threading.Tasks; +using Impostor.Api.Net.Messages; +using Microsoft.Extensions.ObjectPool; namespace Impostor.Hazel.Udp { @@ -24,8 +27,8 @@ namespace Impostor.Hazel.Udp /// The listener that created this connection. /// The endpoint that we are connected to. /// The IPMode we are connected using. - internal UdpServerConnection(UdpConnectionListener listener, IPEndPoint endPoint, IPMode IPMode) - : base() + internal UdpServerConnection(UdpConnectionListener listener, IPEndPoint endPoint, IPMode IPMode, ObjectPool readerPool) + : base(listener, readerPool) { this.Listener = listener; this.EndPoint = endPoint; @@ -36,25 +39,16 @@ namespace Impostor.Hazel.Udp } /// - protected override void WriteBytesToConnection(byte[] bytes, int length) + protected override ValueTask WriteBytesToConnection(byte[] bytes, int length) { - Listener.SendData(bytes, length, EndPoint); + return Listener.SendData(bytes, length, EndPoint); } /// /// /// This will always throw a HazelException. /// - public override void Connect(byte[] bytes = null, int timeout = 5000) - { - throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); - } - - /// - /// - /// This will always throw a HazelException. - /// - public override void ConnectAsync(byte[] bytes = null) + public override ValueTask ConnectAsync(byte[] bytes = null, int timeout = 5000) { throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); } @@ -62,18 +56,18 @@ namespace Impostor.Hazel.Udp /// /// Sends a disconnect message to the end point. /// - protected override bool SendDisconnect(MessageWriter data = null) + protected override async ValueTask SendDisconnect(MessageWriter data = null) { lock (this) { if (this._state != ConnectionState.Connected) return false; this._state = ConnectionState.NotConnected; } - + var bytes = EmptyDisconnectBytes; if (data != null && data.Length > 0) { - if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable."); + if (data.SendOption != MessageType.Unreliable) throw new ArgumentException("Disconnect messages can only be unreliable."); bytes = data.ToByteArray(true); bytes[0] = (byte)UdpSendOption.Disconnect; @@ -81,7 +75,7 @@ namespace Impostor.Hazel.Udp try { - Listener.SendDataSync(bytes, bytes.Length, EndPoint); + await Listener.SendData(bytes, bytes.Length, EndPoint); } catch { } @@ -94,7 +88,7 @@ namespace Impostor.Hazel.Udp if (disposing) { - SendDisconnect(); + _ = SendDisconnect(); } base.Dispose(disposing); diff --git a/Hazel/Udp/UnityUdpClientConnection.cs b/Hazel/Udp/UnityUdpClientConnection.cs deleted file mode 100644 index 76bc342..0000000 --- a/Hazel/Udp/UnityUdpClientConnection.cs +++ /dev/null @@ -1,325 +0,0 @@ -using System; -using System.Net; -using System.Net.Sockets; -using System.Threading; - -namespace Impostor.Hazel.Udp -{ - /// - /// Unity doesn't always get along with thread pools well, so this interface will hopefully suit that case better. - /// Be very careful since this interface is likely unstable or actively changing - /// - /// - public class UnityUdpClientConnection : UdpConnection - { - private Socket socket; - - public UnityUdpClientConnection(IPEndPoint remoteEndPoint, IPMode ipMode = IPMode.IPv4) - : base() - { - this.EndPoint = remoteEndPoint; - this.IPMode = ipMode; - - this.socket = CreateSocket(ipMode); - this.socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse, true); - } - - ~UnityUdpClientConnection() - { - this.Dispose(false); - } - - public void FixedUpdate() - { - this.ResendPacketsIfNeeded(); - } - - protected virtual void RestartConnection() - { - } - - protected virtual void ResendPacketsIfNeeded() - { - base.ManageReliablePackets(); - } - - - /// - protected override void WriteBytesToConnection(byte[] bytes, int length) - { -#if DEBUG - if (TestLagMs > 0) - { - ThreadPool.QueueUserWorkItem(a => { Thread.Sleep(this.TestLagMs); WriteBytesToConnectionReal(bytes, length); }); - } - else -#endif - { - WriteBytesToConnectionReal(bytes, length); - } - } - - private void WriteBytesToConnectionReal(byte[] bytes, int length) - { - try - { - socket.BeginSendTo( - bytes, - 0, - length, - SocketFlags.None, - EndPoint, - HandleSendTo, - null); - } - catch (NullReferenceException) { } - catch (ObjectDisposedException) - { - // Already disposed and disconnected... - } - catch (SocketException ex) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); - } - } - - /// - /// Synchronously writes the given bytes to the connection. - /// - /// The bytes to write. - protected virtual void WriteBytesToConnectionSync(byte[] bytes, int length) - { - try - { - socket.SendTo( - bytes, - 0, - length, - SocketFlags.None, - EndPoint); - } - catch (NullReferenceException) { } - catch (ObjectDisposedException) - { - // Already disposed and disconnected... - } - catch (SocketException ex) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); - } - } - - private void HandleSendTo(IAsyncResult result) - { - try - { - socket.EndSendTo(result); - } - catch (NullReferenceException) { } - catch (ObjectDisposedException) - { - // Already disposed and disconnected... - } - catch (SocketException ex) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionSend, "Could not send data as a SocketException occurred: " + ex.Message); - } - } - - public override void Connect(byte[] bytes = null, int timeout = 5000) - { - this.ConnectAsync(bytes); - for(int timer = 0; timer < timeout; timeout += 100) - { - if (this.State != ConnectionState.Connecting) return; - Thread.Sleep(100); - this.ResendPacketsIfNeeded(); - } - } - - /// - public override void ConnectAsync(byte[] bytes = null) - { - this.State = ConnectionState.Connecting; - - try - { - if (IPMode == IPMode.IPv4) - socket.Bind(new IPEndPoint(IPAddress.Any, 0)); - else - socket.Bind(new IPEndPoint(IPAddress.IPv6Any, 0)); - } - catch (SocketException e) - { - this.State = ConnectionState.NotConnected; - throw new HazelException("A SocketException occurred while binding to the port.", e); - } - - this.RestartConnection(); - - try - { - StartListeningForData(); - } - catch (ObjectDisposedException) - { - // If the socket's been disposed then we can just end there but make sure we're in NotConnected state. - // If we end up here I'm really lost... - this.State = ConnectionState.NotConnected; - return; - } - catch (SocketException e) - { - Dispose(); - throw new HazelException("A SocketException occurred while initiating a receive operation.", e); - } - - this.InitializeKeepAliveTimer(); - - // Write bytes to the server to tell it hi (and to punch a hole in our NAT, if present) - // When acknowledged set the state to connected - SendHello(bytes, () => - { - this.State = ConnectionState.Connected; - }); - } - - /// - /// Instructs the listener to begin listening. - /// - void StartListeningForData() - { - var msg = MessageReader.GetSized(ushort.MaxValue); - try - { - EndPoint ep = this.EndPoint; - socket.BeginReceiveFrom(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, ref ep, ReadCallback, msg); - } - catch - { - msg.Recycle(); - this.Dispose(); - } - } - - /// - /// Called when data has been received by the socket. - /// - /// The asyncronous operation's result. - void ReadCallback(IAsyncResult result) - { -#if DEBUG - if (this.TestLagMs > 0) - { - Thread.Sleep(this.TestLagMs); - } -#endif - - var msg = (MessageReader)result.AsyncState; - - try - { - EndPoint ep = this.EndPoint; - msg.Length = socket.EndReceiveFrom(result, ref ep); - } - catch (SocketException e) - { - msg.Recycle(); - DisconnectInternal(HazelInternalErrors.SocketExceptionReceive, "Socket exception while reading data: " + e.Message); - return; - } - catch (ObjectDisposedException) - { - // Weirdly, it seems that this method can be called twice on the same AsyncState when object is disposed... - // So this just keeps us from hitting Duplicate Add errors at the risk of if this is a platform - // specific bug, we leak a MessageReader while the socket is disposing. Not a bad trade off. - return; - } - catch (Exception) - { - msg.Recycle(); - return; - } - - //Exit if no bytes read, we've failed. - if (msg.Length == 0) - { - msg.Recycle(); - DisconnectInternal(HazelInternalErrors.ReceivedZeroBytes, "Received 0 bytes"); - return; - } - - //Begin receiving again - try - { - StartListeningForData(); - } - catch (SocketException e) - { - DisconnectInternal(HazelInternalErrors.SocketExceptionReceive, "Socket exception during receive: " + e.Message); - } - catch (ObjectDisposedException) - { - //If the socket's been disposed then we can just end there. - return; - } - -#if DEBUG - if (this.TestDropRate > 0) - { - if ((this.testDropCount++ % this.TestDropRate) == 0) - { - return; - } - } -#endif - - HandleReceive(msg, msg.Length); - } - - /// - /// Sends a disconnect message to the end point. - /// You may include optional disconnect data. The SendOption must be unreliable. - /// - protected override bool SendDisconnect(MessageWriter data = null) - { - lock (this) - { - if (this._state == ConnectionState.NotConnected) return false; - this._state = ConnectionState.NotConnected; - } - - var bytes = EmptyDisconnectBytes; - if (data != null && data.Length > 0) - { - if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable."); - - bytes = data.ToByteArray(true); - bytes[0] = (byte)UdpSendOption.Disconnect; - } - - try - { - this.WriteBytesToConnectionSync(bytes, bytes.Length); - } - catch { } - - return true; - } - - /// - protected override void Dispose(bool disposing) - { - if (disposing) - { - SendDisconnect(); - } - - try { this.socket.Shutdown(SocketShutdown.Both); } catch { } - try { this.socket.Close(); } catch { } - try { this.socket.Dispose(); } catch { } - - base.Dispose(disposing); - } - } -} -- 2.39.5