From cde51ee403101d1d619ac9ac816cde8a43fcb845 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 19:16:57 +0000 Subject: [PATCH 1/3] Add packet capturing to the connections and make the analyzer reusable This is the infrastructure for the network analyzer page of the admin panel: it allows to watch the decrypted traffic of a connection which is handled by one of our servers, without an external proxy. All three server types create their connections through the same Listener, so one hook in the Connection covers the connect server, the game server and the chat server. Both directions are captured unencrypted: the incoming packets are read after the decryptor, and the Output writes into the encryptor, so it still sees plain text. * IPacketCaptureSink gets the complete, decrypted data packets of a connection. Sinks are registered with IConnection.AddCaptureSink and removed again with RemoveCaptureSink. As long as no sink is registered, a connection captures nothing - it just checks a field for null per packet and direction. * Because a write to the pipe writer isn't necessarily one data packet, the written data is buffered by the OutgoingPacketCollector and split into packets again, based on the packet header. * IConnection has an Id now, so a connection can be addressed by the admin panel and correlated in the logs. The PacketAnalyzer got three changes to be usable for more than the WinForms tool: * The client version is a parameter of the extraction methods instead of a property. One instance can therefore analyze the traffic of connections with different client versions at the same time. * The packet definitions are selected by their Direction element and a PacketDefinitionSet, instead of one file per direction. That's what allows to analyze the traffic of the connect server and the chat server, too - their definitions are included in the output now. * The definition files are only watched for changes when it's requested. That's a development feature, so only the WinForms tool uses it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Pb82LmoaUVdZtBtQs7xrtA --- src/Network/Analyzer.WinForms/MainForm.cs | 13 +- .../MUnique.OpenMU.Network.Analyzer.csproj | 6 + src/Network/Analyzer/PacketAnalyzer.cs | 187 +++++++----- src/Network/Analyzer/PacketDefinitionSet.cs | 27 ++ src/Network/Connection.cs | 139 ++++++++- src/Network/ExtendedPipeWriter.cs | 41 ++- src/Network/IConnection.cs | 23 ++ src/Network/IPacketCaptureSink.cs | 31 ++ src/Network/OutgoingPacketCollector.cs | 115 +++++++ .../MUnique.OpenMU.Network.Tests.csproj | 1 + .../PacketAnalyzerTest.cs | 107 +++++++ .../PacketCaptureTest.cs | 287 ++++++++++++++++++ 12 files changed, 889 insertions(+), 88 deletions(-) create mode 100644 src/Network/Analyzer/PacketDefinitionSet.cs create mode 100644 src/Network/IPacketCaptureSink.cs create mode 100644 src/Network/OutgoingPacketCollector.cs create mode 100644 tests/MUnique.OpenMU.Network.Tests/PacketAnalyzerTest.cs create mode 100644 tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs diff --git a/src/Network/Analyzer.WinForms/MainForm.cs b/src/Network/Analyzer.WinForms/MainForm.cs index 22876a317e..f9fc3f9870 100644 --- a/src/Network/Analyzer.WinForms/MainForm.cs +++ b/src/Network/Analyzer.WinForms/MainForm.cs @@ -58,7 +58,9 @@ public MainForm() this.connectedClientsListBox.DisplayMember = nameof(ICapturedConnection.Name); this.connectedClientsListBox.Update(); - this._analyzer = new PacketAnalyzer(); + // The analyzer watches the packet definition files, so that changed definitions are + // applied without restarting the tool. + this._analyzer = new PacketAnalyzer(PacketDefinitionSet.GameServer, watchFiles: true); this.Disposed += (_, _) => this._analyzer.Dispose(); this.clientVersionComboBox.SelectedIndexChanged += this.OnSelectedClientVersionChanged; @@ -236,8 +238,9 @@ private void SetPacketDataSource() private void OnPacketAnalyzingRequested(object? sender, Packet.AnalyzingRequestedEventArgs e) { - e.ClientVersion = this._analyzer.ClientVersion; - (e.Message, e.Definition) = this._analyzer.ExtractShortInformation(e.Packet); + var clientVersion = this.SelectedClientVersion; + e.ClientVersion = clientVersion; + (e.Message, e.Definition) = this._analyzer.ExtractShortInformation(e.Packet, clientVersion); } private void OnUnfilteredListChanged(object? sender, ListChangedEventArgs e) @@ -267,7 +270,6 @@ private void OnSelectedClientVersionChanged(object? o, EventArgs eventArgs) listener.ClientVersion = this.SelectedClientVersion; } - this._analyzer.ClientVersion = this.SelectedClientVersion; if (this._unfilteredList is { } unfilteredList) { foreach (var packet in unfilteredList) @@ -315,7 +317,6 @@ private void StartProxy(object sender, System.EventArgs e) ClientVersion = this.SelectedClientVersion, }; - this._analyzer.ClientVersion = this.SelectedClientVersion; this._clientListener.ClientConnected += this.ClientListenerOnClientConnected; this._clientListener.Start(); this.btnStartProxy.Text = "Stop Proxy"; @@ -364,7 +365,7 @@ private void OnPacketSelected(object sender, EventArgs e) if (rows.Count > 0 && this.packetGridView.SelectedRows[0].DataBoundItem is Packet packet) { this.rawDataTextBox.Text = packet.PacketData; - this.extractedInfoTextBox.Text = this._analyzer.ExtractInformation(packet); + this.extractedInfoTextBox.Text = this._analyzer.ExtractInformation(packet, this.SelectedClientVersion); this.packetInfoGroup.Enabled = true; } else diff --git a/src/Network/Analyzer/MUnique.OpenMU.Network.Analyzer.csproj b/src/Network/Analyzer/MUnique.OpenMU.Network.Analyzer.csproj index 0213d7ad1f..e2ad75c00d 100644 --- a/src/Network/Analyzer/MUnique.OpenMU.Network.Analyzer.csproj +++ b/src/Network/Analyzer/MUnique.OpenMU.Network.Analyzer.csproj @@ -28,6 +28,12 @@ PreserveNewest + + PreserveNewest + + + PreserveNewest + diff --git a/src/Network/Analyzer/PacketAnalyzer.cs b/src/Network/Analyzer/PacketAnalyzer.cs index 4007a471e4..c1fed9b3cf 100644 --- a/src/Network/Analyzer/PacketAnalyzer.cs +++ b/src/Network/Analyzer/PacketAnalyzer.cs @@ -1,4 +1,4 @@ -// +// // Licensed under the MIT License. See LICENSE file in the project root for full license information. // @@ -16,82 +16,91 @@ namespace MUnique.OpenMU.Network.Analyzer; /// public sealed class PacketAnalyzer : IDisposable { - private const string ClientToServerPacketsFile = "ClientToServerPackets.xml"; - private const string ServerToClientPacketsFile = "ServerToClientPackets.xml"; private const string CommonFile = "CommonEnums.xml"; private const int DefaultVersionValue = 100; private const int ExtendedVersionValue = (106 * 100) + 3; private readonly IList _watchers = new List(); - private PacketDefinitions? _clientPacketDefinitions; - private PacketDefinitions? _serverPacketDefinitions; - private PacketDefinitions? _commonDefinitions; - private ClientVersion _clientVersion; - private int _clientVersionValue = DefaultVersionValue; /// - /// Initializes a new instance of the class. - /// The configuration is automatically loaded from the configuration files. + /// The loaded packet definitions of the . The array has one + /// slot per file, so that a reloaded file can simply replace its previous content. /// - public PacketAnalyzer() - { - this.LoadAndWatchConfiguration(def => this._serverPacketDefinitions = def, ServerToClientPacketsFile); - this.LoadAndWatchConfiguration(def => this._clientPacketDefinitions = def, ClientToServerPacketsFile); - this.LoadAndWatchConfiguration(def => this._commonDefinitions = def, CommonFile); - } + private readonly PacketDefinitions?[] _packetDefinitions; + + private PacketDefinitions? _commonDefinitions; /// - /// Gets or sets the client version. + /// Initializes a new instance of the class. + /// The definitions are automatically loaded from the configuration files. /// - public ClientVersion ClientVersion + /// The set of packet definitions which should be used. + /// If set to true, the definition files are watched and + /// automatically reloaded when they change. That's useful when new packet definitions are + /// developed, but usually not required when just analyzing the traffic. + public PacketAnalyzer(PacketDefinitionSet definitionSet = PacketDefinitionSet.GameServer, bool watchFiles = false) { - get => this._clientVersion; - set + this.DefinitionSet = definitionSet; + + var files = GetDefinitionFiles(definitionSet); + this._packetDefinitions = new PacketDefinitions?[files.Length]; + for (int i = 0; i < files.Length; i++) { - this._clientVersion = value; - this._clientVersionValue = (value.Season * 100) + value.Episode; + var index = i; + this.LoadConfiguration(def => this._packetDefinitions[index] = def, files[index], watchFiles); } + + this.LoadConfiguration(def => this._commonDefinitions = def, CommonFile, watchFiles); } + /// + /// Gets the set of packet definitions which is used by this instance. + /// + public PacketDefinitionSet DefinitionSet { get; } + /// /// Extracts the information of the packet and returns it as a formatted string. /// /// The packet. + /// The client version of the connection, which decides which + /// packet definition applies when more than one matches. /// The formatted string with the extracted information. - public string ExtractInformation(Packet packet) + public string ExtractInformation(Packet packet, ClientVersion clientVersion) { - var definitions = packet.ToServer ? this._clientPacketDefinitions : this._serverPacketDefinitions; - var definition = this.DeterminePacketDefinition(packet); - if (definition != null) + if (this.DeterminePacketDefinition(packet, clientVersion) is not { } match) { - var stringBuilder = new StringBuilder() - .Append(definition.Caption ?? definition.Name); - foreach (var field in definition.Fields ?? Enumerable.Empty()) - { - stringBuilder.Append(Environment.NewLine) - .Append(field.Name).Append(": ").Append(this.ExtractFieldValueOrGetError(packet.Data.AsSpan(), field, definition, definitions!)); - } + return string.Empty; + } - return stringBuilder.ToString(); + var (definition, definitions) = match; + var clientVersionValue = GetVersionValue(clientVersion); + var stringBuilder = new StringBuilder() + .Append(definition.Caption ?? definition.Name); + foreach (var field in definition.Fields ?? Enumerable.Empty()) + { + stringBuilder.Append(Environment.NewLine) + .Append(field.Name).Append(": ").Append(this.ExtractFieldValueOrGetError(packet.Data.AsSpan(), field, definition, definitions, clientVersionValue)); } - return string.Empty; + return stringBuilder.ToString(); } /// /// Extracts the information of the packet and returns it as a short, formatted string. /// /// The packet. + /// The client version of the connection, which decides which + /// packet definition applies when more than one matches. /// The formatted string with the extracted information. - public (string Data, PacketDefinition? Definition) ExtractShortInformation(Packet packet) + public (string Data, PacketDefinition? Definition) ExtractShortInformation(Packet packet, ClientVersion clientVersion) { - var definitions = packet.ToServer ? this._clientPacketDefinitions : this._serverPacketDefinitions; - var definition = this.DeterminePacketDefinition(packet); - if (definition is null) + if (this.DeterminePacketDefinition(packet, clientVersion) is not { } match) { return (packet.PacketData, null); } + var (definition, definitions) = match; + var clientVersionValue = GetVersionValue(clientVersion); var stringBuilder = new StringBuilder(definition.Caption ?? definition.Name ?? string.Empty); var relevantFields = definition.Fields? .Where(f => f.Type != FieldType.Binary && f.Type != FieldType.StructureArray) @@ -112,7 +121,7 @@ public string ExtractInformation(Packet packet) stringBuilder.Append(field.Name) .Append(": ") - .Append(this.ExtractFieldValueOrGetError(packet.Data.AsSpan(), field, definition, definitions!)); + .Append(this.ExtractFieldValueOrGetError(packet.Data.AsSpan(), field, definition, definitions, clientVersionValue)); } stringBuilder.Append(")"); @@ -134,13 +143,26 @@ public void Dispose() this._watchers.Clear(); } - private PacketDefinition? DeterminePacketDefinition(Packet packet) + private static int GetVersionValue(ClientVersion clientVersion) + { + return (clientVersion.Season * 100) + clientVersion.Episode; + } + + private static string[] GetDefinitionFiles(PacketDefinitionSet definitionSet) { - var allDefinitions = packet.ToServer ? this._clientPacketDefinitions : this._serverPacketDefinitions; - if (allDefinitions is null) + return definitionSet switch { - return null; - } + PacketDefinitionSet.GameServer => ["ClientToServerPackets.xml", "ServerToClientPackets.xml"], + PacketDefinitionSet.ConnectServer => ["ConnectServerPackets.xml"], + PacketDefinitionSet.ChatServer => ["ChatServerPackets.xml"], + _ => throw new ArgumentOutOfRangeException(nameof(definitionSet), definitionSet, "Unknown packet definition set."), + }; + } + + private (PacketDefinition Definition, PacketDefinitions Owner)? DeterminePacketDefinition(Packet packet, ClientVersion clientVersion) + { + var direction = packet.ToServer ? Direction.ClientToServer : Direction.ServerToClient; + var clientVersionValue = GetVersionValue(clientVersion); int GetVersion(string name) { @@ -158,51 +180,55 @@ int GetVersion(string name) return DefaultVersionValue; } - var filteredDefinitions = allDefinitions.Packets? - .Where(p => (byte)p.Type == packet.Type && p.Code == packet.Code && (!p.SubCodeSpecified || p.SubCode == packet.SubCode)) - .Select(p => (Version: GetVersion(p.Name ?? string.Empty), Definition: p)) + var filteredDefinitions = this._packetDefinitions + .Where(definitions => definitions is not null) + .SelectMany(definitions => (definitions!.Packets ?? Enumerable.Empty()) + .Select(p => (Definition: p, Owner: definitions))) + .Where(pair => pair.Definition.Direction == direction || pair.Definition.Direction == Direction.Bidirectional) + .Where(pair => (byte)pair.Definition.Type == packet.Type && pair.Definition.Code == packet.Code && (!pair.Definition.SubCodeSpecified || pair.Definition.SubCode == packet.SubCode)) + .Select(pair => (Version: GetVersion(pair.Definition.Name ?? string.Empty), pair.Definition, pair.Owner)) .OrderBy(pair => pair.Version) .ToList(); - if (filteredDefinitions is null || !filteredDefinitions.Any()) + if (filteredDefinitions.Count == 0) { return null; } if (filteredDefinitions.Count == 1) { - return filteredDefinitions[0].Definition; + return (filteredDefinitions[0].Definition, filteredDefinitions[0].Owner); } - if (filteredDefinitions.FirstOrDefault(d => d.Version == this._clientVersionValue) is { Definition: { Name: { } } } exactMatch) + if (filteredDefinitions.FirstOrDefault(d => d.Version == clientVersionValue) is { Definition: { Name: { } } } exactMatch) { - return exactMatch.Definition; + return (exactMatch.Definition, exactMatch.Owner); } - var sameLengthPackets = filteredDefinitions.Where(d => d.Definition.Length == packet.Size).Select(d => d.Definition).ToList(); + var sameLengthPackets = filteredDefinitions.Where(d => d.Definition.Length == packet.Size).ToList(); if (sameLengthPackets.Count > 0) { - if (sameLengthPackets.Count == 1 && sameLengthPackets.First() is { Name: { } } sameLengthMatch) + if (sameLengthPackets.Count == 1 && sameLengthPackets[0] is { Definition.Name: { } } sameLengthMatch) { - return sameLengthMatch; + return (sameLengthMatch.Definition, sameLengthMatch.Owner); } - var filteredByDefaults = this.GetPacketDefinitionsFilteredByDefaultValues(packet, sameLengthPackets, allDefinitions).ToList(); + var filteredByDefaults = this.GetPacketDefinitionsFilteredByDefaultValues(packet, sameLengthPackets, clientVersionValue).ToList(); if (filteredByDefaults.Count == 1) { - return filteredByDefaults.First(); + return (filteredByDefaults[0].Definition, filteredByDefaults[0].Owner); } if (filteredByDefaults.Count > 0) { - filteredDefinitions.RemoveAll(def => !filteredByDefaults.Contains(def.Definition)); + filteredDefinitions.RemoveAll(def => !filteredByDefaults.Any(f => ReferenceEquals(f.Definition, def.Definition))); } } - var current = filteredDefinitions.First(); + var current = filteredDefinitions[0]; foreach (var def in filteredDefinitions.Skip(1)) { - if (def.Version > this._clientVersionValue) + if (def.Version > clientVersionValue) { break; } @@ -210,31 +236,37 @@ int GetVersion(string name) current = def; } - return current.Definition; + return (current.Definition, current.Owner); } - private IEnumerable GetPacketDefinitionsFilteredByDefaultValues(Packet packet, IEnumerable definitions, PacketDefinitions allDefinitions) + private IEnumerable<(int Version, PacketDefinition Definition, PacketDefinitions Owner)> GetPacketDefinitionsFilteredByDefaultValues(Packet packet, IEnumerable<(int Version, PacketDefinition Definition, PacketDefinitions Owner)> definitions, int clientVersionValue) { - foreach (var def in definitions) + foreach (var candidate in definitions) { + var def = candidate.Definition; var defaultFields = def.Fields?.TakeWhile(f => !string.IsNullOrWhiteSpace(f.DefaultValue)).ToList(); if (defaultFields is null or { Count: 0 }) { break; } - if (defaultFields.TrueForAll(field => int.TryParse(this.ExtractFieldValueOrGetError(packet.Data, field, def, allDefinitions), out var actual) + if (defaultFields.TrueForAll(field => int.TryParse(this.ExtractFieldValueOrGetError(packet.Data, field, def, candidate.Owner, clientVersionValue), out var actual) && (int.TryParse(field.DefaultValue, out var target) || int.TryParse(field.DefaultValue!.Replace("0x", string.Empty), NumberStyles.HexNumber, CultureInfo.InvariantCulture, out target)) && actual == target)) { - yield return def; + yield return candidate; } } } - private void LoadAndWatchConfiguration(Action assignAction, string fileName) + private void LoadConfiguration(Action assignAction, string fileName, bool watchFile) { assignAction(PacketDefinitions.Load(fileName)); + if (!watchFile) + { + return; + } + var watcher = new FileSystemWatcher(Environment.CurrentDirectory, fileName); watcher.Changed += (_, _) => @@ -268,14 +300,15 @@ private void LoadAndWatchConfiguration(Action assignAction, /// The field definition. /// The packet. /// The definitions. + /// The numeric client version of the connection. /// /// The value of the field or the error message. /// - private string ExtractFieldValueOrGetError(Span data, Field field, PacketDefinition packet, PacketDefinitions definitions) + private string ExtractFieldValueOrGetError(Span data, Field field, PacketDefinition packet, PacketDefinitions definitions, int clientVersionValue) { try { - return this.ExtractFieldValue(data, field, packet, definitions); + return this.ExtractFieldValue(data, field, packet, definitions, clientVersionValue); } catch (Exception e) { @@ -290,10 +323,11 @@ private string ExtractFieldValueOrGetError(Span data, Field field, PacketD /// The field definition. /// The packet. /// The definitions. + /// The numeric client version of the connection. /// /// The value of the field. /// - private string ExtractFieldValue(Span data, Field field, PacketDefinition packet, PacketDefinitions definitions) + private string ExtractFieldValue(Span data, Field field, PacketDefinition packet, PacketDefinitions definitions, int clientVersionValue) { var fieldSize = field.GetFieldSizeInBytes(); if (field.Type == FieldType.String && field.Index < data.Length) @@ -326,14 +360,14 @@ private string ExtractFieldValue(Span data, Field field, PacketDefinition FieldType.LongLittleEndian => ReadUInt64LittleEndian(data[field.Index..]).ToString(CultureInfo.InvariantCulture), FieldType.LongBigEndian => ReadUInt64BigEndian(data[field.Index..]).ToString(CultureInfo.InvariantCulture), FieldType.Enum => this.ExtractEnumValue(data, field, packet, definitions), - FieldType.StructureArray => this.ExtractStructureArrayValues(data, field, packet, definitions), + FieldType.StructureArray => this.ExtractStructureArrayValues(data, field, packet, definitions, clientVersionValue), FieldType.Float => ReadSingleLittleEndian(data[field.Index..]).ToString(CultureInfo.InvariantCulture), FieldType.Double => ReadDoubleBigEndian(data[field.Index..]).ToString(CultureInfo.InvariantCulture), _ => string.Empty, }; } - private string ExtractStructureArrayValues(Span data, Field arrayField, PacketDefinition packet, PacketDefinitions definitions) + private string ExtractStructureArrayValues(Span data, Field arrayField, PacketDefinition packet, PacketDefinitions definitions, int clientVersionValue) { var elementType = packet.Structures?.FirstOrDefault(s => s.Name == arrayField.TypeName) ?? definitions.Structures?.FirstOrDefault(s => s.Name == arrayField.TypeName) @@ -345,7 +379,7 @@ private string ExtractStructureArrayValues(Span data, Field arrayField, Pa var countField = packet.Fields?.FirstOrDefault(f => f.Name == arrayField.ItemCountField) ?? packet.Structures?.SelectMany(s => s.Fields ?? Enumerable.Empty()).FirstOrDefault(f => f.Name == arrayField.ItemCountField); - int count = countField is null ? 0 : int.Parse(this.ExtractFieldValue(data, countField, packet, definitions), CultureInfo.InvariantCulture); + int count = countField is null ? 0 : int.Parse(this.ExtractFieldValue(data, countField, packet, definitions, clientVersionValue), CultureInfo.InvariantCulture); if (count == 0) { return string.Empty; @@ -358,7 +392,7 @@ private string ExtractStructureArrayValues(Span data, Field arrayField, Pa for (int i = 0; i < count; i++) { - var currentLength = typeLength ?? this.DetermineDynamicStructLength(restData, elementType, packet) ?? fixedLengthByCount; + var currentLength = typeLength ?? this.DetermineDynamicStructLength(restData, elementType, packet, clientVersionValue) ?? fixedLengthByCount; if (currentLength is null) { break; @@ -374,7 +408,7 @@ private string ExtractStructureArrayValues(Span data, Field arrayField, Pa foreach (var structField in elementType.Fields ?? Enumerable.Empty()) { stringBuilder.Append(Environment.NewLine) - .Append(" ").Append(structField.Name).Append(": ").Append(this.ExtractFieldValue(elementData, structField, packet, definitions)); + .Append(" ").Append(structField.Name).Append(": ").Append(this.ExtractFieldValue(elementData, structField, packet, definitions, clientVersionValue)); } } @@ -418,8 +452,9 @@ private string ExtractEnumValue(Span data, Field field, PacketDefinition p /// The rest data. /// The type. /// Type of the packet. + /// The numeric client version of the connection. /// The dynamic length of a struct with a nested structure array. - private int? DetermineDynamicStructLength(Span restData, Structure type, PacketDefinition packetType) + private int? DetermineDynamicStructLength(Span restData, Structure type, PacketDefinition packetType, int clientVersionValue) { if (type.Fields is null) { @@ -435,7 +470,7 @@ private string ExtractEnumValue(Span data, Field field, PacketDefinition p return nestedStructField.Index + (count * nestedStructType.Length); } - if (this._clientVersionValue == ExtendedVersionValue + if (clientVersionValue == ExtendedVersionValue && type.Fields.FirstOrDefault(f => f.Type == FieldType.Binary) is { } binaryField && binaryField.Name?.EndsWith("ItemData") is true) { diff --git a/src/Network/Analyzer/PacketDefinitionSet.cs b/src/Network/Analyzer/PacketDefinitionSet.cs new file mode 100644 index 0000000000..0979eb8be1 --- /dev/null +++ b/src/Network/Analyzer/PacketDefinitionSet.cs @@ -0,0 +1,27 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network.Analyzer; + +/// +/// The set of packet definitions which is used to analyze the traffic of a connection. +/// Which one applies depends on the server the client is connected to. +/// +public enum PacketDefinitionSet +{ + /// + /// The packets which are exchanged between game client and game server. + /// + GameServer, + + /// + /// The packets which are exchanged between game client and connect server. + /// + ConnectServer, + + /// + /// The packets which are exchanged between game client and chat server. + /// + ChatServer, +} diff --git a/src/Network/Connection.cs b/src/Network/Connection.cs index abc5dd65c9..59b0301fde 100644 --- a/src/Network/Connection.cs +++ b/src/Network/Connection.cs @@ -43,9 +43,13 @@ public sealed class Connection : PacketPipeReaderBase, IConnection private readonly ILogger _logger; private readonly EndPoint _remoteEndPoint; + private readonly object _captureLock = new(); + private IDuplexPipe? _duplexPipe; private bool _disconnected; - private PipeWriter? _outputWriter; + private ExtendedPipeWriter? _outputWriter; + + private volatile IPacketCaptureSink[]? _captureSinks; /// /// Initializes a new instance of the class. @@ -71,6 +75,9 @@ public Connection(IDuplexPipe duplexPipe, IPipelinedDecryptor? decryptionPipe, I /// public event AsyncEventHandler? Disconnected; + /// + public Guid Id { get; } = Guid.NewGuid(); + /// public bool Connected => this.SocketConnection != null ? this.SocketConnection.ShutdownKind == PipeShutdownKind.None && !this._disconnected : !this._disconnected; @@ -81,7 +88,7 @@ public Connection(IDuplexPipe duplexPipe, IPipelinedDecryptor? decryptionPipe, I public EndPoint? LocalEndPoint { get; } /// - public PipeWriter Output => this._outputWriter ??= new ExtendedPipeWriter(this._encryptionPipe?.Writer ?? this._duplexPipe!.Output, OutgoingBytesCounter); + public PipeWriter Output => this.OutputWriter; /// public AsyncLock OutputLock { get; } @@ -96,6 +103,12 @@ public Connection(IDuplexPipe duplexPipe, IPipelinedDecryptor? decryptionPipe, I /// private SocketConnection? SocketConnection => this._duplexPipe as SocketConnection; + /// + /// Gets the of the , which is also + /// the place where the outgoing data packets are captured. + /// + private ExtendedPipeWriter OutputWriter => this._outputWriter ??= this.CreateOutputWriter(this._duplexPipe!); + /// public override string ToString() => this._remoteEndPoint?.ToString() ?? $"{base.ToString()} {this.GetHashCode()}"; @@ -146,12 +159,80 @@ public async ValueTask DisconnectAsync() await this.Disconnected.SafeInvokeAsync().ConfigureAwait(false); } + /// + public void AddCaptureSink(IPacketCaptureSink sink) + { + lock (this._captureLock) + { + var current = this._captureSinks; + if (current is null) + { + this._captureSinks = new[] { sink }; + } + else + { + if (Array.IndexOf(current, sink) >= 0) + { + return; + } + + var updated = new IPacketCaptureSink[current.Length + 1]; + current.CopyTo(updated, 0); + updated[^1] = sink; + this._captureSinks = updated; + } + + // We don't use the OutputWriter property here, because the connection may already + // be disconnected - in that case there is no outgoing traffic to capture anymore. + var outputWriter = this._outputWriter + ?? (this._duplexPipe is { } duplexPipe ? this._outputWriter = this.CreateOutputWriter(duplexPipe) : null); + if (outputWriter is not null) + { + outputWriter.PacketCollector ??= new OutgoingPacketCollector(this.OnPacketSent); + } + } + } + + /// + public void RemoveCaptureSink(IPacketCaptureSink sink) + { + lock (this._captureLock) + { + var current = this._captureSinks; + if (current is null) + { + return; + } + + var index = Array.IndexOf(current, sink); + if (index < 0) + { + return; + } + + if (current.Length == 1) + { + this.StopCapturing(); + return; + } + + var updated = new IPacketCaptureSink[current.Length - 1]; + Array.Copy(current, updated, index); + Array.Copy(current, index + 1, updated, index, current.Length - index - 1); + this._captureSinks = updated; + } + } + /// public void Dispose() { this.DisconnectAsync().AsTask().WaitAndUnwrapException(); this.PacketReceived = null; this.Disconnected = null; + lock (this._captureLock) + { + this.StopCapturing(); + } } /// @@ -207,6 +288,18 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence .Start(); try { + if (this._captureSinks is not null) + { + if (packet.IsSingleSegment) + { + this.RaisePacketCaptured(packet.FirstSpan, false); + } + else + { + this.RaisePacketCaptured(packet.ToArray(), false); + } + } + await this.PacketReceived.SafeInvokeAsync(packet).ConfigureAwait(false); return true; } @@ -215,4 +308,46 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence activity?.Stop(); } } + + private ExtendedPipeWriter CreateOutputWriter(IDuplexPipe duplexPipe) + { + return new ExtendedPipeWriter(this._encryptionPipe?.Writer ?? duplexPipe.Output, OutgoingBytesCounter); + } + + /// + /// Stops the capturing. Must be called within a lock of the . + /// + private void StopCapturing() + { + this._captureSinks = null; + if (this._outputWriter is { } outputWriter) + { + outputWriter.PacketCollector = null; + } + } + + private void OnPacketSent(ReadOnlySpan packet) + { + this.RaisePacketCaptured(packet, true); + } + + private void RaisePacketCaptured(ReadOnlySpan packet, bool sent) + { + if (this._captureSinks is not { } sinks) + { + return; + } + + foreach (var sink in sinks) + { + try + { + sink.PacketCaptured(packet, sent); + } + catch (Exception ex) + { + this._logger.LogWarning(ex, "Error in a packet capture sink of connection {connectionId}.", this.Id); + } + } + } } \ No newline at end of file diff --git a/src/Network/ExtendedPipeWriter.cs b/src/Network/ExtendedPipeWriter.cs index 850daa6ae3..206c112510 100644 --- a/src/Network/ExtendedPipeWriter.cs +++ b/src/Network/ExtendedPipeWriter.cs @@ -16,6 +16,8 @@ public class ExtendedPipeWriter : PipeWriter private readonly PipeWriter _target; private readonly Counter _writeCounter; + private Memory _lastBuffer; + /// /// Initializes a new instance of the class. /// @@ -27,6 +29,12 @@ public ExtendedPipeWriter(PipeWriter target, Counter writeCounter) this._writeCounter = writeCounter; } + /// + /// Gets or sets the collector which gets the written data, to capture the outgoing data + /// packets. If it's , nothing is captured. + /// + internal OutgoingPacketCollector? PacketCollector { get; set; } + /// public override void Complete(Exception? exception = null) { @@ -48,6 +56,17 @@ public override ValueTask FlushAsync(CancellationToken cancellation /// public override void Advance(int bytes) { + if (this.PacketCollector is { } collector && bytes > 0 && this._lastBuffer.Length >= bytes) + { + // The data has to be collected before it's advanced, because the target may + // recycle the buffer afterwards. + collector.DataWritten(this._lastBuffer.Span[..bytes]); + } + + // A new buffer has to be requested before writing again, so we forget this one. + // That way, the same data can't be reported twice. + this._lastBuffer = default; + this._target.Advance(bytes); this._writeCounter.Add(bytes); } @@ -55,14 +74,28 @@ public override void Advance(int bytes) /// public override Memory GetMemory(int sizeHint = 0) { - return this._target.GetMemory(sizeHint); + var memory = this._target.GetMemory(sizeHint); + this._lastBuffer = memory; + return memory; } /// public override Span GetSpan(int sizeHint = 0) { - var span = this._target.GetSpan(sizeHint); - span.Clear(); - return span; + if (this.PacketCollector is null) + { + // We don't remember the buffer in this case. That way, a collector which gets + // attached between this call and the next Advance doesn't report stale data. + this._lastBuffer = default; + var span = this._target.GetSpan(sizeHint); + span.Clear(); + return span; + } + + var memory = this._target.GetMemory(sizeHint); + this._lastBuffer = memory; + var memorySpan = memory.Span; + memorySpan.Clear(); + return memorySpan; } } \ No newline at end of file diff --git a/src/Network/IConnection.cs b/src/Network/IConnection.cs index 7ff3b24a36..2bb6f8e54c 100644 --- a/src/Network/IConnection.cs +++ b/src/Network/IConnection.cs @@ -29,6 +29,11 @@ public interface IConnection : IDisposable /// event AsyncEventHandler? Disconnected; + /// + /// Gets the identifier of this connection, which is unique for the lifetime of the process. + /// + Guid Id { get; } + /// /// Gets a value indicating whether this is connected. /// @@ -70,4 +75,22 @@ public interface IConnection : IDisposable /// Disconnects this instance. /// ValueTask DisconnectAsync(); + + /// + /// Adds a sink which gets all decrypted data packets of this connection, until it's + /// removed again by . + /// + /// The sink which should get the data packets. + /// + /// Adding the same sink more than once has no effect. As long as no sink is added, the + /// connection doesn't capture anything. + /// + void AddCaptureSink(IPacketCaptureSink sink); + + /// + /// Removes a previously added capture sink. When the last sink is removed, the connection + /// stops capturing. + /// + /// The sink which should not get the data packets anymore. + void RemoveCaptureSink(IPacketCaptureSink sink); } \ No newline at end of file diff --git a/src/Network/IPacketCaptureSink.cs b/src/Network/IPacketCaptureSink.cs new file mode 100644 index 0000000000..3004d93d95 --- /dev/null +++ b/src/Network/IPacketCaptureSink.cs @@ -0,0 +1,31 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network; + +/// +/// A sink which gets the decrypted data packets of a connection, as long as it's registered +/// at . +/// +/// +/// This allows to watch the traffic of a connection without an external proxy, e.g. to show +/// it in the admin panel. As long as no sink is registered at a connection, it doesn't +/// capture anything. +/// +public interface IPacketCaptureSink +{ + /// + /// Is called when a complete, decrypted data packet was received from or sent to the + /// remote endpoint of the connection. + /// + /// The complete, decrypted data packet. + /// , if the packet was sent to the remote endpoint; + /// , if it was received from it. + /// + /// This is called on the network thread of the connection, so an implementation should + /// return as fast as possible. The is only valid during the call, + /// so it must be copied if it's required afterwards. + /// + void PacketCaptured(ReadOnlySpan packet, bool sent); +} diff --git a/src/Network/OutgoingPacketCollector.cs b/src/Network/OutgoingPacketCollector.cs new file mode 100644 index 0000000000..9caa40aa43 --- /dev/null +++ b/src/Network/OutgoingPacketCollector.cs @@ -0,0 +1,115 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network; + +/// +/// Handler for a collected data packet. +/// +/// The complete data packet. +internal delegate void PacketCollectedHandler(ReadOnlySpan packet); + +/// +/// Collects the data which is written to an and forwards +/// complete data packets to a handler. +/// +/// +/// One write to a is not necessarily one data +/// packet: a bigger message may be written in several chunks, and it's also possible that +/// more than one packet is written before the writer gets flushed. The written data is +/// therefore buffered and split into packets again, based on the packet header. +/// +internal sealed class OutgoingPacketCollector +{ + private const int InitialBufferSize = 256; + + private readonly PacketCollectedHandler _packetCollected; + + private byte[] _buffer = new byte[InitialBufferSize]; + + private int _bufferedLength; + + /// + /// Initializes a new instance of the class. + /// + /// The handler which is called for each complete data packet. + public OutgoingPacketCollector(PacketCollectedHandler packetCollected) + { + this._packetCollected = packetCollected; + } + + /// + /// Adds the written data and forwards each complete data packet to the handler. + /// + /// The data which has been written to the pipe writer. + public void DataWritten(ReadOnlySpan data) + { + this.Append(data); + + var offset = 0; + while (offset < this._bufferedLength) + { + var rest = this._buffer.AsSpan(offset, this._bufferedLength - offset); + var headerSize = ArrayExtensions.GetPacketHeaderSize(rest[0]); + if (headerSize == 0) + { + // It's not a packet we know, so we're not able to determine the packet + // boundaries of the subsequent data anymore. We drop what we have instead + // of reporting garbage. + this._bufferedLength = 0; + return; + } + + if (rest.Length < headerSize) + { + break; + } + + var packetSize = rest.GetPacketSize(); + if (packetSize < headerSize) + { + this._bufferedLength = 0; + return; + } + + if (rest.Length < packetSize) + { + break; + } + + this._packetCollected(rest[..packetSize]); + offset += packetSize; + } + + this.RemoveFromBuffer(offset); + } + + private void Append(ReadOnlySpan data) + { + var requiredLength = this._bufferedLength + data.Length; + if (this._buffer.Length < requiredLength) + { + Array.Resize(ref this._buffer, Math.Max(requiredLength, this._buffer.Length * 2)); + } + + data.CopyTo(this._buffer.AsSpan(this._bufferedLength)); + this._bufferedLength = requiredLength; + } + + private void RemoveFromBuffer(int count) + { + if (count == 0) + { + return; + } + + var rest = this._bufferedLength - count; + if (rest > 0) + { + Array.Copy(this._buffer, count, this._buffer, 0, rest); + } + + this._bufferedLength = rest; + } +} diff --git a/tests/MUnique.OpenMU.Network.Tests/MUnique.OpenMU.Network.Tests.csproj b/tests/MUnique.OpenMU.Network.Tests/MUnique.OpenMU.Network.Tests.csproj index 6937f452d5..a8c38dfc19 100644 --- a/tests/MUnique.OpenMU.Network.Tests/MUnique.OpenMU.Network.Tests.csproj +++ b/tests/MUnique.OpenMU.Network.Tests/MUnique.OpenMU.Network.Tests.csproj @@ -38,5 +38,6 @@ + diff --git a/tests/MUnique.OpenMU.Network.Tests/PacketAnalyzerTest.cs b/tests/MUnique.OpenMU.Network.Tests/PacketAnalyzerTest.cs new file mode 100644 index 0000000000..0b2a6e3d70 --- /dev/null +++ b/tests/MUnique.OpenMU.Network.Tests/PacketAnalyzerTest.cs @@ -0,0 +1,107 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network.Tests; + +using MUnique.OpenMU.Network.Analyzer; +using MUnique.OpenMU.Network.PlugIns; + +/// +/// Tests for the , especially for the selection of the packet +/// definition by direction, definition set and client version. +/// +[TestFixture] +public class PacketAnalyzerTest +{ + private static readonly ClientVersion Season6 = new(6, 3, ClientLanguage.English); + + private static readonly ClientVersion Version075 = new(0, 75, ClientLanguage.Invariant); + + /// + /// Tests if the same packet code is resolved to a different definition, depending on the + /// direction of the packet. + /// + [Test] + public void DefinitionIsSelectedByDirection() + { + using var analyzer = new PacketAnalyzer(); + + var toServer = new Packet(TimeSpan.Zero, [0xC1, 0x05, 0x15, 0x01, 0x02], true); + var toClient = new Packet(TimeSpan.Zero, [0xC1, 0x08, 0x15, 0x00, 0x01, 0x02, 0x03, 0x04], false); + + Assert.That(analyzer.ExtractShortInformation(toServer, Season6).Definition?.Name, Is.EqualTo("InstantMoveRequest")); + Assert.That(analyzer.ExtractShortInformation(toClient, Season6).Definition?.Name, Is.EqualTo("ObjectMoved")); + } + + /// + /// Tests if the client version decides which of the definitions of a packet code applies. + /// The version is passed per call, so one instance can serve connections of different + /// client versions. + /// + [Test] + public void DefinitionIsSelectedByClientVersion() + { + using var analyzer = new PacketAnalyzer(); + var packet = new Packet(TimeSpan.Zero, [0xC2, 0x00, 0x06, 0x13, 0x01, 0x00], false); + + Assert.That(analyzer.ExtractShortInformation(packet, Version075).Definition?.Name, Is.EqualTo("AddNpcsToScope075")); + Assert.That(analyzer.ExtractShortInformation(packet, Season6).Definition?.Name, Is.EqualTo("AddNpcsToScope")); + } + + /// + /// Tests if the definitions of the connect server are used when the corresponding + /// definition set is selected. + /// + [Test] + public void ConnectServerDefinitionsAreSelectable() + { + using var analyzer = new PacketAnalyzer(PacketDefinitionSet.ConnectServer); + var packet = new Packet(TimeSpan.Zero, [0xC1, 0x04, 0xF4, 0x06], true); + + Assert.That(analyzer.ExtractShortInformation(packet, Season6).Definition?.Name, Is.EqualTo("ServerListRequest")); + } + + /// + /// Tests if a packet of the game server definitions is not found in the connect server + /// definition set. + /// + [Test] + public void DefinitionsOfOtherSetsAreNotUsed() + { + using var analyzer = new PacketAnalyzer(PacketDefinitionSet.ConnectServer); + var packet = new Packet(TimeSpan.Zero, [0xC1, 0x05, 0x15, 0x01, 0x02], true); + + Assert.That(analyzer.ExtractShortInformation(packet, Season6).Definition, Is.Null); + } + + /// + /// Tests if a bidirectional packet definition is found for both directions. + /// + [Test] + public void BidirectionalDefinitionIsFoundInBothDirections() + { + using var analyzer = new PacketAnalyzer(PacketDefinitionSet.ChatServer); + var toServer = new Packet(TimeSpan.Zero, [0xC1, 0x05, 0x04, 0x00, 0x01], true); + var toClient = new Packet(TimeSpan.Zero, [0xC1, 0x05, 0x04, 0x00, 0x01], false); + + Assert.That(analyzer.ExtractShortInformation(toServer, Season6).Definition?.Name, Is.EqualTo("ChatMessage")); + Assert.That(analyzer.ExtractShortInformation(toClient, Season6).Definition?.Name, Is.EqualTo("ChatMessage")); + } + + /// + /// Tests if the extracted information contains the field values of the packet. + /// + [Test] + public void InformationContainsFieldValues() + { + using var analyzer = new PacketAnalyzer(); + var packet = new Packet(TimeSpan.Zero, [0xC1, 0x05, 0x15, 0x0A, 0x14], true); + + var information = analyzer.ExtractInformation(packet, Season6); + + Assert.That(information, Does.Contain("InstantMoveRequest")); + Assert.That(information, Does.Contain("10")); + Assert.That(information, Does.Contain("20")); + } +} diff --git a/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs new file mode 100644 index 0000000000..f9e06e10a8 --- /dev/null +++ b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs @@ -0,0 +1,287 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network.Tests; + +using System.Buffers; +using System.Collections.Generic; +using System.Linq; +using Microsoft.Extensions.Logging.Abstractions; + +/// +/// Tests for the packet capturing of a . +/// +[TestFixture] +public class PacketCaptureTest +{ + /// + /// Tests if a received data packet is reported to a registered sink. + /// + /// The async task. + [Test] + public async Task ReceivedPacketIsCapturedAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + var duplexPipe = new DuplexPipe(); + using var connection = new Connection(duplexPipe, null, null, new NullLogger()); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + _ = connection.BeginReceiveAsync(); + + await duplexPipe.ReceivePipe.Writer.WriteAsync(packet).ConfigureAwait(false); + await WaitForPacketsAsync(sink, 1).ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(1)); + Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + Assert.That(sink.Captured[0].Sent, Is.False); + } + + /// + /// Tests if a sent data packet is reported to a registered sink. + /// + /// The async task. + [Test] + public async Task SentPacketIsCapturedAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(1)); + Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + Assert.That(sink.Captured[0].Sent, Is.True); + } + + /// + /// Tests if a data packet which is written in more than one chunk is reported as one packet. + /// + /// The async task. + [Test] + public async Task FragmentedPacketIsCapturedAsOnePacketAsync() + { + var packet = new byte[] { 0xC1, 0x06, 0xF1, 0x00, 0x11, 0x22 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + Write(connection, packet.AsSpan(0, 3)); + Assert.That(sink.Captured, Is.Empty, "The packet is not complete yet."); + + Write(connection, packet.AsSpan(3)); + await connection.Output.FlushAsync().ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(1)); + Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + } + + /// + /// Tests if more than one data packet which is written at once is reported as separate packets. + /// + /// The async task. + [Test] + public async Task MultiplePacketsInOneWriteAreCapturedSeparatelyAsync() + { + var first = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + var second = new byte[] { 0xC1, 0x03, 0xF3 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + await connection.Output.WriteAsync(first.Concat(second).ToArray()).ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(2)); + Assert.That(sink.Captured[0].Packet, Is.EqualTo(first)); + Assert.That(sink.Captured[1].Packet, Is.EqualTo(second)); + } + + /// + /// Tests if a packet with a two byte length header (C2) is captured with its correct length. + /// + /// The async task. + [Test] + public async Task BigPacketIsCapturedAsync() + { + var packet = new byte[300]; + packet[0] = 0xC2; + packet[1] = 0x01; + packet[2] = 0x2C; + packet[3] = 0xF1; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(1)); + Assert.That(sink.Captured[0].Packet, Has.Length.EqualTo(300)); + Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + } + + /// + /// Tests if an incomplete data packet is reported as soon as the rest is written, + /// even when a complete packet was written before it. + /// + /// The async task. + [Test] + public async Task IncompleteTrailingPacketIsCapturedWhenCompletedAsync() + { + var complete = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + var trailing = new byte[] { 0xC1, 0x04, 0xF3, 0x01 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + Write(connection, complete.Concat(trailing.Take(2)).ToArray()); + Assert.That(sink.Captured, Has.Count.EqualTo(1), "Only the complete packet should be reported."); + + Write(connection, trailing.AsSpan(2)); + await connection.Output.FlushAsync().ConfigureAwait(false); + + Assert.That(sink.Captured, Has.Count.EqualTo(2)); + Assert.That(sink.Captured[1].Packet, Is.EqualTo(trailing)); + } + + /// + /// Tests if all registered sinks get the data packets, and that a removed sink doesn't + /// get them anymore. + /// + /// The async task. + [Test] + public async Task AllRegisteredSinksAreNotifiedAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + using var connection = CreateConnection(); + var first = new CapturingSink(); + var second = new CapturingSink(); + connection.AddCaptureSink(first); + connection.AddCaptureSink(second); + connection.AddCaptureSink(first); // registering twice should have no effect + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(first.Captured, Has.Count.EqualTo(1)); + Assert.That(second.Captured, Has.Count.EqualTo(1)); + + connection.RemoveCaptureSink(first); + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(first.Captured, Has.Count.EqualTo(1), "The removed sink should not get further packets."); + Assert.That(second.Captured, Has.Count.EqualTo(2)); + } + + /// + /// Tests if nothing is captured anymore after the last sink has been removed. + /// + /// The async task. + [Test] + public async Task NothingIsCapturedAfterLastSinkWasRemovedAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + connection.RemoveCaptureSink(sink); + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(sink.Captured, Is.Empty); + } + + /// + /// Tests if the written data still arrives at the target pipe unchanged, with and without + /// a registered capture sink. + /// + /// If set to true, a capture sink is registered. + /// The async task. + [TestCase(true)] + [TestCase(false)] + public async Task WrittenDataIsForwardedUnchangedAsync(bool withSink) + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + var duplexPipe = new DuplexPipe(); + using var connection = new Connection(duplexPipe, null, null, new NullLogger()); + if (withSink) + { + connection.AddCaptureSink(new CapturingSink()); + } + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + var result = await duplexPipe.SendPipe.Reader.ReadAsync().ConfigureAwait(false); + Assert.That(result.Buffer.ToArray(), Is.EqualTo(packet)); + } + + /// + /// Tests if an exception of a sink doesn't bubble up to the connection. + /// + /// The async task. + [Test] + public async Task ExceptionInSinkIsCaughtAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + using var connection = CreateConnection(); + var working = new CapturingSink(); + connection.AddCaptureSink(new ThrowingSink()); + connection.AddCaptureSink(working); + + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + + Assert.That(connection.Connected, Is.True); + Assert.That(working.Captured, Has.Count.EqualTo(1)); + } + + /// + /// Tests if each connection has its own identifier. + /// + [Test] + public void EachConnectionHasItsOwnId() + { + using var first = CreateConnection(); + using var second = CreateConnection(); + + Assert.That(first.Id, Is.Not.EqualTo(Guid.Empty)); + Assert.That(first.Id, Is.Not.EqualTo(second.Id)); + } + + private static Connection CreateConnection() + { + return new Connection(new DuplexPipe(), null, null, new NullLogger()); + } + + private static void Write(Connection connection, ReadOnlySpan data) + { + var span = connection.Output.GetSpan(data.Length); + data.CopyTo(span); + connection.Output.Advance(data.Length); + } + + private static async Task WaitForPacketsAsync(CapturingSink sink, int count) + { + for (int i = 0; i < 100 && sink.Captured.Count < count; i++) + { + await Task.Delay(10).ConfigureAwait(false); + } + } + + private sealed class CapturingSink : IPacketCaptureSink + { + public IList<(byte[] Packet, bool Sent)> Captured { get; } = new List<(byte[] Packet, bool Sent)>(); + + public void PacketCaptured(ReadOnlySpan packet, bool sent) + { + this.Captured.Add((packet.ToArray(), sent)); + } + } + + private sealed class ThrowingSink : IPacketCaptureSink + { + public void PacketCaptured(ReadOnlySpan packet, bool sent) + { + throw new InvalidOperationException("This sink is broken."); + } + } +} From 7a41fa22dc61f06186598ce5da4acf533b19f8f8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 05:25:13 +0000 Subject: [PATCH 2/3] Use a pipe and immutable arrays for the packet capturing Two improvements over the first version of the capture infrastructure: * The outgoing data is no longer split into packets by a hand written collector. Instead, it's copied into an own pipe, which is read by the CapturedPacketReader - a PacketPipeReaderBase, like every other packet source of the network layer. That reuses the packet splitting we already have, including the handling of malformed data: when the captured data is malformed, the reader stops and detaches itself, without affecting the connection. The data is copied when it's advanced, and flushed to the capture when the connection flushes, so the capture never blocks the connection. * The registered sinks are kept in an ImmutableArray which is updated with ImmutableInterlocked, instead of copying arrays under a lock. The lock is gone, and the array keeps the iteration in the hot path allocation free. The creation of the ExtendedPipeWriter is atomic now, so the capture is always attached to the same instance which is used by the Output. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Pb82LmoaUVdZtBtQs7xrtA --- src/Network/CapturedPacketReader.cs | 131 +++++++++++++++++ src/Network/Connection.cs | 137 ++++++++---------- src/Network/ExtendedPipeWriter.cs | 46 ++++-- src/Network/OutgoingPacketCollector.cs | 115 --------------- .../PacketCaptureTest.cs | 130 ++++++++++++----- 5 files changed, 322 insertions(+), 237 deletions(-) create mode 100644 src/Network/CapturedPacketReader.cs delete mode 100644 src/Network/OutgoingPacketCollector.cs diff --git a/src/Network/CapturedPacketReader.cs b/src/Network/CapturedPacketReader.cs new file mode 100644 index 0000000000..10490dc01f --- /dev/null +++ b/src/Network/CapturedPacketReader.cs @@ -0,0 +1,131 @@ +// +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// + +namespace MUnique.OpenMU.Network; + +using System.Buffers; +using System.IO.Pipelines; +using Microsoft.Extensions.Logging; + +/// +/// Handler for a captured data packet. +/// +/// The complete data packet. +internal delegate void PacketCapturedHandler(ReadOnlySpan packet); + +/// +/// Splits the captured outgoing data of a connection into data packets again. +/// +/// +/// One write to a is not necessarily one data packet: a bigger message +/// may be written in several chunks, and more than one packet may be written before the writer +/// gets flushed. The written data is therefore copied into an own pipe, which is read by this +/// class like any other packet source of the network layer. +/// +internal sealed class CapturedPacketReader : PacketPipeReaderBase +{ + private readonly PacketCapturedHandler _packetCaptured; + + private readonly Action _completed; + + private readonly ILogger _logger; + + private readonly Pipe _pipe; + + private bool _isCompleted; + + /// + /// Initializes a new instance of the class. + /// + /// The handler which is called for each complete data packet. + /// Is called when this reader stopped reading, e.g. because the + /// captured data was malformed. The caller should then stop writing into the . + /// The logger. + public CapturedPacketReader(PacketCapturedHandler packetCaptured, Action completed, ILogger logger) + { + this._packetCaptured = packetCaptured; + this._completed = completed; + this._logger = logger; + + // The capturing must never slow down or block the connection itself, so the writer is + // never paused. The reader just forwards the packets to the sinks, so it keeps up. + this._pipe = new Pipe(new PipeOptions(useSynchronizationContext: false, pauseWriterThreshold: 0, resumeWriterThreshold: 0)); + this.Source = this._pipe.Reader; + } + + /// + /// Gets the writer, into which the captured data is written. + /// + public PipeWriter Writer => this._pipe.Writer; + + /// + /// Starts reading the captured data. + /// + public void Start() + { + _ = this.ReadCapturedDataAsync(); + } + + /// + /// Stops the reader by completing the . + /// + public void Stop() + { + try + { + this._pipe.Writer.Complete(); + } + catch (Exception ex) + { + this._logger.LogDebug(ex, "Error when completing the packet capture."); + } + } + + /// + protected override ValueTask ReadPacketAsync(ReadOnlySequence packet) + { + if (packet.IsSingleSegment) + { + this._packetCaptured(packet.FirstSpan); + } + else + { + this._packetCaptured(packet.ToArray()); + } + + return ValueTask.FromResult(true); + } + + /// + protected override async ValueTask OnCompleteAsync(Exception? exception) + { + if (this._isCompleted) + { + return; + } + + this._isCompleted = true; + if (exception is not null) + { + this._logger.LogWarning(exception, "Error while reading the captured data packets. The capturing is stopped."); + } + + await this._pipe.Reader.CompleteAsync(exception).ConfigureAwait(false); + this._completed(); + } + + private async Task ReadCapturedDataAsync() + { + try + { + await this.ReadSourceAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + // ReadSourceAsync already reported it to OnCompleteAsync; the capturing of this + // connection is over, but the connection itself is not affected. + this._logger.LogDebug(ex, "The packet capture reader stopped."); + } + } +} diff --git a/src/Network/Connection.cs b/src/Network/Connection.cs index 59b0301fde..93188cef02 100644 --- a/src/Network/Connection.cs +++ b/src/Network/Connection.cs @@ -5,6 +5,7 @@ namespace MUnique.OpenMU.Network; using System.Buffers; +using System.Collections.Immutable; using System.Diagnostics; using System.Diagnostics.Metrics; using System.IO.Pipelines; @@ -43,13 +44,13 @@ public sealed class Connection : PacketPipeReaderBase, IConnection private readonly ILogger _logger; private readonly EndPoint _remoteEndPoint; - private readonly object _captureLock = new(); - private IDuplexPipe? _duplexPipe; private bool _disconnected; private ExtendedPipeWriter? _outputWriter; - private volatile IPacketCaptureSink[]? _captureSinks; + private ImmutableArray _captureSinks = ImmutableArray.Empty; + + private CapturedPacketReader? _outgoingCapture; /// /// Initializes a new instance of the class. @@ -107,7 +108,7 @@ public Connection(IDuplexPipe duplexPipe, IPipelinedDecryptor? decryptionPipe, I /// Gets the of the , which is also /// the place where the outgoing data packets are captured. /// - private ExtendedPipeWriter OutputWriter => this._outputWriter ??= this.CreateOutputWriter(this._duplexPipe!); + private ExtendedPipeWriter OutputWriter => this.GetOrCreateOutputWriter(this._duplexPipe!); /// public override string ToString() => this._remoteEndPoint?.ToString() ?? $"{base.ToString()} {this.GetHashCode()}"; @@ -162,65 +163,14 @@ public async ValueTask DisconnectAsync() /// public void AddCaptureSink(IPacketCaptureSink sink) { - lock (this._captureLock) - { - var current = this._captureSinks; - if (current is null) - { - this._captureSinks = new[] { sink }; - } - else - { - if (Array.IndexOf(current, sink) >= 0) - { - return; - } - - var updated = new IPacketCaptureSink[current.Length + 1]; - current.CopyTo(updated, 0); - updated[^1] = sink; - this._captureSinks = updated; - } - - // We don't use the OutputWriter property here, because the connection may already - // be disconnected - in that case there is no outgoing traffic to capture anymore. - var outputWriter = this._outputWriter - ?? (this._duplexPipe is { } duplexPipe ? this._outputWriter = this.CreateOutputWriter(duplexPipe) : null); - if (outputWriter is not null) - { - outputWriter.PacketCollector ??= new OutgoingPacketCollector(this.OnPacketSent); - } - } + ImmutableInterlocked.Update(ref this._captureSinks, static (sinks, added) => sinks.Contains(added) ? sinks : sinks.Add(added), sink); + this.StartCapturingOutgoingData(); } /// public void RemoveCaptureSink(IPacketCaptureSink sink) { - lock (this._captureLock) - { - var current = this._captureSinks; - if (current is null) - { - return; - } - - var index = Array.IndexOf(current, sink); - if (index < 0) - { - return; - } - - if (current.Length == 1) - { - this.StopCapturing(); - return; - } - - var updated = new IPacketCaptureSink[current.Length - 1]; - Array.Copy(current, updated, index); - Array.Copy(current, index + 1, updated, index, current.Length - index - 1); - this._captureSinks = updated; - } + ImmutableInterlocked.Update(ref this._captureSinks, static (sinks, removed) => sinks.Remove(removed), sink); } /// @@ -229,10 +179,7 @@ public void Dispose() this.DisconnectAsync().AsTask().WaitAndUnwrapException(); this.PacketReceived = null; this.Disconnected = null; - lock (this._captureLock) - { - this.StopCapturing(); - } + this.StopCapturing(); } /// @@ -288,7 +235,7 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence .Start(); try { - if (this._captureSinks is not null) + if (!this._captureSinks.IsEmpty) { if (packet.IsSingleSegment) { @@ -309,20 +256,65 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence } } - private ExtendedPipeWriter CreateOutputWriter(IDuplexPipe duplexPipe) + /// + /// Gets the , or creates it, if it doesn't exist yet. + /// It's created atomically, so that the capturing is attached to the same instance which + /// is used by the . + /// + /// The duplex pipe of the connection. + /// The of this connection. + private ExtendedPipeWriter GetOrCreateOutputWriter(IDuplexPipe duplexPipe) { - return new ExtendedPipeWriter(this._encryptionPipe?.Writer ?? duplexPipe.Output, OutgoingBytesCounter); + if (this._outputWriter is { } existingWriter) + { + return existingWriter; + } + + var createdWriter = new ExtendedPipeWriter(this._encryptionPipe?.Writer ?? duplexPipe.Output, OutgoingBytesCounter); + return Interlocked.CompareExchange(ref this._outputWriter, createdWriter, null) ?? createdWriter; + } + + /// + /// Starts to capture the outgoing data packets, if it's not running yet. The incoming + /// packets don't need that, because they arrive as complete packets anyway. + /// + private void StartCapturingOutgoingData() + { + if (this._outgoingCapture is not null || this._disconnected || this._duplexPipe is not { } duplexPipe) + { + return; + } + + var capture = new CapturedPacketReader(this.OnPacketSent, this.StopCapturing, this._logger); + if (Interlocked.CompareExchange(ref this._outgoingCapture, capture, null) is not null) + { + // Another thread was faster. + capture.Stop(); + return; + } + + // We don't use the OutputWriter property here, because the connection may already be + // disconnected - in that case there is no outgoing traffic to capture anymore. + var outputWriter = this.GetOrCreateOutputWriter(duplexPipe); + outputWriter.CaptureWriter = capture.Writer; + capture.Start(); } /// - /// Stops the capturing. Must be called within a lock of the . + /// Stops the capturing of the outgoing data packets. The capture of a connection is not + /// stopped when the last sink is removed, because the data could then only be captured + /// again at a packet boundary, which we can't determine from the outside. /// private void StopCapturing() { - this._captureSinks = null; - if (this._outputWriter is { } outputWriter) + if (Interlocked.Exchange(ref this._outgoingCapture, null) is { } capture) { - outputWriter.PacketCollector = null; + if (this._outputWriter is { } outputWriter) + { + outputWriter.CaptureWriter = null; + } + + capture.Stop(); } } @@ -333,12 +325,7 @@ private void OnPacketSent(ReadOnlySpan packet) private void RaisePacketCaptured(ReadOnlySpan packet, bool sent) { - if (this._captureSinks is not { } sinks) - { - return; - } - - foreach (var sink in sinks) + foreach (var sink in this._captureSinks) { try { diff --git a/src/Network/ExtendedPipeWriter.cs b/src/Network/ExtendedPipeWriter.cs index 206c112510..09100ad49e 100644 --- a/src/Network/ExtendedPipeWriter.cs +++ b/src/Network/ExtendedPipeWriter.cs @@ -30,10 +30,10 @@ public ExtendedPipeWriter(PipeWriter target, Counter writeCounter) } /// - /// Gets or sets the collector which gets the written data, to capture the outgoing data - /// packets. If it's , nothing is captured. + /// Gets or sets the writer of the packet capture, into which the written data is copied. + /// If it's , nothing is captured. /// - internal OutgoingPacketCollector? PacketCollector { get; set; } + internal PipeWriter? CaptureWriter { get; set; } /// public override void Complete(Exception? exception = null) @@ -50,17 +50,31 @@ public override void CancelPendingFlush() /// public override ValueTask FlushAsync(CancellationToken cancellationToken = default) { + if (this.CaptureWriter is { } captureWriter) + { + return this.FlushWithCaptureAsync(captureWriter, cancellationToken); + } + return this._target.FlushAsync(cancellationToken); } /// public override void Advance(int bytes) { - if (this.PacketCollector is { } collector && bytes > 0 && this._lastBuffer.Length >= bytes) + if (this.CaptureWriter is { } captureWriter && bytes > 0 && this._lastBuffer.Length >= bytes) { - // The data has to be collected before it's advanced, because the target may - // recycle the buffer afterwards. - collector.DataWritten(this._lastBuffer.Span[..bytes]); + // The data has to be copied before it's advanced, because the target may recycle + // the buffer afterwards. The capturing must never break the connection, so a + // failing capture is silently ignored here. + try + { + this._lastBuffer.Span[..bytes].CopyTo(captureWriter.GetSpan(bytes)); + captureWriter.Advance(bytes); + } + catch (InvalidOperationException) + { + // The capture has been completed in the meantime. + } } // A new buffer has to be requested before writing again, so we forget this one. @@ -82,9 +96,9 @@ public override Memory GetMemory(int sizeHint = 0) /// public override Span GetSpan(int sizeHint = 0) { - if (this.PacketCollector is null) + if (this.CaptureWriter is null) { - // We don't remember the buffer in this case. That way, a collector which gets + // We don't remember the buffer in this case. That way, a capture which gets // attached between this call and the next Advance doesn't report stale data. this._lastBuffer = default; var span = this._target.GetSpan(sizeHint); @@ -98,4 +112,18 @@ public override Span GetSpan(int sizeHint = 0) memorySpan.Clear(); return memorySpan; } + + private async ValueTask FlushWithCaptureAsync(PipeWriter captureWriter, CancellationToken cancellationToken) + { + try + { + await captureWriter.FlushAsync(cancellationToken).ConfigureAwait(false); + } + catch (InvalidOperationException) + { + // The capture has been completed in the meantime. + } + + return await this._target.FlushAsync(cancellationToken).ConfigureAwait(false); + } } \ No newline at end of file diff --git a/src/Network/OutgoingPacketCollector.cs b/src/Network/OutgoingPacketCollector.cs deleted file mode 100644 index 9caa40aa43..0000000000 --- a/src/Network/OutgoingPacketCollector.cs +++ /dev/null @@ -1,115 +0,0 @@ -// -// Licensed under the MIT License. See LICENSE file in the project root for full license information. -// - -namespace MUnique.OpenMU.Network; - -/// -/// Handler for a collected data packet. -/// -/// The complete data packet. -internal delegate void PacketCollectedHandler(ReadOnlySpan packet); - -/// -/// Collects the data which is written to an and forwards -/// complete data packets to a handler. -/// -/// -/// One write to a is not necessarily one data -/// packet: a bigger message may be written in several chunks, and it's also possible that -/// more than one packet is written before the writer gets flushed. The written data is -/// therefore buffered and split into packets again, based on the packet header. -/// -internal sealed class OutgoingPacketCollector -{ - private const int InitialBufferSize = 256; - - private readonly PacketCollectedHandler _packetCollected; - - private byte[] _buffer = new byte[InitialBufferSize]; - - private int _bufferedLength; - - /// - /// Initializes a new instance of the class. - /// - /// The handler which is called for each complete data packet. - public OutgoingPacketCollector(PacketCollectedHandler packetCollected) - { - this._packetCollected = packetCollected; - } - - /// - /// Adds the written data and forwards each complete data packet to the handler. - /// - /// The data which has been written to the pipe writer. - public void DataWritten(ReadOnlySpan data) - { - this.Append(data); - - var offset = 0; - while (offset < this._bufferedLength) - { - var rest = this._buffer.AsSpan(offset, this._bufferedLength - offset); - var headerSize = ArrayExtensions.GetPacketHeaderSize(rest[0]); - if (headerSize == 0) - { - // It's not a packet we know, so we're not able to determine the packet - // boundaries of the subsequent data anymore. We drop what we have instead - // of reporting garbage. - this._bufferedLength = 0; - return; - } - - if (rest.Length < headerSize) - { - break; - } - - var packetSize = rest.GetPacketSize(); - if (packetSize < headerSize) - { - this._bufferedLength = 0; - return; - } - - if (rest.Length < packetSize) - { - break; - } - - this._packetCollected(rest[..packetSize]); - offset += packetSize; - } - - this.RemoveFromBuffer(offset); - } - - private void Append(ReadOnlySpan data) - { - var requiredLength = this._bufferedLength + data.Length; - if (this._buffer.Length < requiredLength) - { - Array.Resize(ref this._buffer, Math.Max(requiredLength, this._buffer.Length * 2)); - } - - data.CopyTo(this._buffer.AsSpan(this._bufferedLength)); - this._bufferedLength = requiredLength; - } - - private void RemoveFromBuffer(int count) - { - if (count == 0) - { - return; - } - - var rest = this._bufferedLength - count; - if (rest > 0) - { - Array.Copy(this._buffer, count, this._buffer, 0, rest); - } - - this._bufferedLength = rest; - } -} diff --git a/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs index f9e06e10a8..8941cb364c 100644 --- a/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs +++ b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs @@ -30,11 +30,11 @@ public async Task ReceivedPacketIsCapturedAsync() _ = connection.BeginReceiveAsync(); await duplexPipe.ReceivePipe.Writer.WriteAsync(packet).ConfigureAwait(false); - await WaitForPacketsAsync(sink, 1).ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(1).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(1)); - Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); - Assert.That(sink.Captured[0].Sent, Is.False); + Assert.That(captured, Has.Count.EqualTo(1)); + Assert.That(captured[0].Packet, Is.EqualTo(packet)); + Assert.That(captured[0].Sent, Is.False); } /// @@ -50,10 +50,11 @@ public async Task SentPacketIsCapturedAsync() connection.AddCaptureSink(sink); await connection.Output.WriteAsync(packet).ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(1).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(1)); - Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); - Assert.That(sink.Captured[0].Sent, Is.True); + Assert.That(captured, Has.Count.EqualTo(1)); + Assert.That(captured[0].Packet, Is.EqualTo(packet)); + Assert.That(captured[0].Sent, Is.True); } /// @@ -69,13 +70,15 @@ public async Task FragmentedPacketIsCapturedAsOnePacketAsync() connection.AddCaptureSink(sink); Write(connection, packet.AsSpan(0, 3)); - Assert.That(sink.Captured, Is.Empty, "The packet is not complete yet."); + await connection.Output.FlushAsync().ConfigureAwait(false); + Assert.That(sink.Snapshot(), Is.Empty, "The packet is not complete yet."); Write(connection, packet.AsSpan(3)); await connection.Output.FlushAsync().ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(1).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(1)); - Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + Assert.That(captured, Has.Count.EqualTo(1)); + Assert.That(captured[0].Packet, Is.EqualTo(packet)); } /// @@ -92,10 +95,11 @@ public async Task MultiplePacketsInOneWriteAreCapturedSeparatelyAsync() connection.AddCaptureSink(sink); await connection.Output.WriteAsync(first.Concat(second).ToArray()).ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(2).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(2)); - Assert.That(sink.Captured[0].Packet, Is.EqualTo(first)); - Assert.That(sink.Captured[1].Packet, Is.EqualTo(second)); + Assert.That(captured, Has.Count.EqualTo(2)); + Assert.That(captured[0].Packet, Is.EqualTo(first)); + Assert.That(captured[1].Packet, Is.EqualTo(second)); } /// @@ -115,14 +119,15 @@ public async Task BigPacketIsCapturedAsync() connection.AddCaptureSink(sink); await connection.Output.WriteAsync(packet).ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(1).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(1)); - Assert.That(sink.Captured[0].Packet, Has.Length.EqualTo(300)); - Assert.That(sink.Captured[0].Packet, Is.EqualTo(packet)); + Assert.That(captured, Has.Count.EqualTo(1)); + Assert.That(captured[0].Packet, Has.Length.EqualTo(300)); + Assert.That(captured[0].Packet, Is.EqualTo(packet)); } /// - /// Tests if an incomplete data packet is reported as soon as the rest is written, + /// Tests if an incomplete data packet is only reported as soon as the rest is written, /// even when a complete packet was written before it. /// /// The async task. @@ -135,14 +140,15 @@ public async Task IncompleteTrailingPacketIsCapturedWhenCompletedAsync() var sink = new CapturingSink(); connection.AddCaptureSink(sink); - Write(connection, complete.Concat(trailing.Take(2)).ToArray()); - Assert.That(sink.Captured, Has.Count.EqualTo(1), "Only the complete packet should be reported."); + await connection.Output.WriteAsync(complete.Concat(trailing.Take(2)).ToArray()).ConfigureAwait(false); + var captured = await sink.WaitForPacketsAsync(1).ConfigureAwait(false); + Assert.That(captured, Has.Count.EqualTo(1), "Only the complete packet should be reported."); - Write(connection, trailing.AsSpan(2)); - await connection.Output.FlushAsync().ConfigureAwait(false); + await connection.Output.WriteAsync(trailing.AsMemory(2)).ConfigureAwait(false); + captured = await sink.WaitForPacketsAsync(2).ConfigureAwait(false); - Assert.That(sink.Captured, Has.Count.EqualTo(2)); - Assert.That(sink.Captured[1].Packet, Is.EqualTo(trailing)); + Assert.That(captured, Has.Count.EqualTo(2)); + Assert.That(captured[1].Packet, Is.EqualTo(trailing)); } /// @@ -162,15 +168,18 @@ public async Task AllRegisteredSinksAreNotifiedAsync() connection.AddCaptureSink(first); // registering twice should have no effect await connection.Output.WriteAsync(packet).ConfigureAwait(false); + await first.WaitForPacketsAsync(1).ConfigureAwait(false); + await second.WaitForPacketsAsync(1).ConfigureAwait(false); - Assert.That(first.Captured, Has.Count.EqualTo(1)); - Assert.That(second.Captured, Has.Count.EqualTo(1)); + Assert.That(first.Snapshot(), Has.Count.EqualTo(1)); + Assert.That(second.Snapshot(), Has.Count.EqualTo(1)); connection.RemoveCaptureSink(first); await connection.Output.WriteAsync(packet).ConfigureAwait(false); + await second.WaitForPacketsAsync(2).ConfigureAwait(false); - Assert.That(first.Captured, Has.Count.EqualTo(1), "The removed sink should not get further packets."); - Assert.That(second.Captured, Has.Count.EqualTo(2)); + Assert.That(first.Snapshot(), Has.Count.EqualTo(1), "The removed sink should not get further packets."); + Assert.That(second.Snapshot(), Has.Count.EqualTo(2)); } /// @@ -187,8 +196,9 @@ public async Task NothingIsCapturedAfterLastSinkWasRemovedAsync() connection.RemoveCaptureSink(sink); await connection.Output.WriteAsync(packet).ConfigureAwait(false); + await Task.Delay(50).ConfigureAwait(false); - Assert.That(sink.Captured, Is.Empty); + Assert.That(sink.Snapshot(), Is.Empty); } /// @@ -229,9 +239,33 @@ public async Task ExceptionInSinkIsCaughtAsync() connection.AddCaptureSink(working); await connection.Output.WriteAsync(packet).ConfigureAwait(false); + var captured = await working.WaitForPacketsAsync(1).ConfigureAwait(false); + + Assert.That(connection.Connected, Is.True); + Assert.That(captured, Has.Count.EqualTo(1)); + } + + /// + /// Tests if malformed captured data stops the capturing without affecting the connection. + /// + /// The async task. + [Test] + public async Task MalformedDataStopsCapturingWithoutBreakingTheConnectionAsync() + { + var malformed = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + connection.AddCaptureSink(sink); + + await connection.Output.WriteAsync(malformed).ConfigureAwait(false); + await Task.Delay(50).ConfigureAwait(false); + + Assert.That(sink.Snapshot(), Is.Empty); + Assert.That(connection.Connected, Is.True); + // The connection still works, it's just not captured anymore. + await connection.Output.WriteAsync(new byte[] { 0xC1, 0x04, 0xF1, 0x00 }).ConfigureAwait(false); Assert.That(connection.Connected, Is.True); - Assert.That(working.Captured, Has.Count.EqualTo(1)); } /// @@ -259,21 +293,41 @@ private static void Write(Connection connection, ReadOnlySpan data) connection.Output.Advance(data.Length); } - private static async Task WaitForPacketsAsync(CapturingSink sink, int count) + private sealed class CapturingSink : IPacketCaptureSink { - for (int i = 0; i < 100 && sink.Captured.Count < count; i++) + private readonly List<(byte[] Packet, bool Sent)> _captured = new(); + + public void PacketCaptured(ReadOnlySpan packet, bool sent) { - await Task.Delay(10).ConfigureAwait(false); + var entry = (packet.ToArray(), sent); + lock (this._captured) + { + this._captured.Add(entry); + } } - } - private sealed class CapturingSink : IPacketCaptureSink - { - public IList<(byte[] Packet, bool Sent)> Captured { get; } = new List<(byte[] Packet, bool Sent)>(); + public IList<(byte[] Packet, bool Sent)> Snapshot() + { + lock (this._captured) + { + return this._captured.ToList(); + } + } - public void PacketCaptured(ReadOnlySpan packet, bool sent) + public async Task> WaitForPacketsAsync(int count) { - this.Captured.Add((packet.ToArray(), sent)); + for (int i = 0; i < 100; i++) + { + var snapshot = this.Snapshot(); + if (snapshot.Count >= count) + { + return snapshot; + } + + await Task.Delay(10).ConfigureAwait(false); + } + + return this.Snapshot(); } } From f0c949bb6a3c8b9f96596dae7c3e15fa1181032f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 16:02:06 +0000 Subject: [PATCH 3/3] Stop the packet capture when the last sink was removed The capture of a connection was kept running until it disconnected, because a capture may only start or end at a packet boundary - otherwise the captured data would begin or end in the middle of a data packet. The ExtendedPipeWriter knows where those boundaries are, so it applies a requested change itself now: a pending capture writer is applied when nothing is written to the target yet, or right after a flush. Nothing is copied anymore when nobody is watching, and the capture can be started again later. The lifecycle of the capture is guarded by a lock now. It's only held when a capture is started or stopped, which is a rare operation and never happens on the path of a data packet - the sinks themselves stay lock-free. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Pb82LmoaUVdZtBtQs7xrtA --- src/Network/Connection.cs | 55 ++++++++++++------- src/Network/ExtendedPipeWriter.cs | 42 ++++++++++++-- .../PacketCaptureTest.cs | 31 +++++++++++ 3 files changed, 102 insertions(+), 26 deletions(-) diff --git a/src/Network/Connection.cs b/src/Network/Connection.cs index 93188cef02..917a16ec6b 100644 --- a/src/Network/Connection.cs +++ b/src/Network/Connection.cs @@ -44,6 +44,13 @@ public sealed class Connection : PacketPipeReaderBase, IConnection private readonly ILogger _logger; private readonly EndPoint _remoteEndPoint; + /// + /// Lock for the lifecycle of the . The sinks themselves are + /// lock-free; this one is just held when a capture is started or stopped, which is a rare + /// operation and never happens on the path of a data packet. + /// + private readonly object _captureLifecycleLock = new(); + private IDuplexPipe? _duplexPipe; private bool _disconnected; private ExtendedPipeWriter? _outputWriter; @@ -171,6 +178,10 @@ public void AddCaptureSink(IPacketCaptureSink sink) public void RemoveCaptureSink(IPacketCaptureSink sink) { ImmutableInterlocked.Update(ref this._captureSinks, static (sinks, removed) => sinks.Remove(removed), sink); + if (this._captureSinks.IsEmpty) + { + this.StopCapturing(); + } } /// @@ -280,38 +291,42 @@ private ExtendedPipeWriter GetOrCreateOutputWriter(IDuplexPipe duplexPipe) /// private void StartCapturingOutgoingData() { - if (this._outgoingCapture is not null || this._disconnected || this._duplexPipe is not { } duplexPipe) + lock (this._captureLifecycleLock) { - return; - } + if (this._outgoingCapture is not null || this._captureSinks.IsEmpty || this._disconnected + || this._duplexPipe is not { } duplexPipe) + { + return; + } - var capture = new CapturedPacketReader(this.OnPacketSent, this.StopCapturing, this._logger); - if (Interlocked.CompareExchange(ref this._outgoingCapture, capture, null) is not null) - { - // Another thread was faster. - capture.Stop(); - return; - } + var capture = new CapturedPacketReader(this.OnPacketSent, this.StopCapturing, this._logger); + this._outgoingCapture = capture; - // We don't use the OutputWriter property here, because the connection may already be - // disconnected - in that case there is no outgoing traffic to capture anymore. - var outputWriter = this.GetOrCreateOutputWriter(duplexPipe); - outputWriter.CaptureWriter = capture.Writer; - capture.Start(); + // We don't use the OutputWriter property here, because the connection may already + // be disconnected - in that case there is no outgoing traffic to capture anymore. + this.GetOrCreateOutputWriter(duplexPipe).PendingCaptureWriter = capture.Writer; + capture.Start(); + } } /// - /// Stops the capturing of the outgoing data packets. The capture of a connection is not - /// stopped when the last sink is removed, because the data could then only be captured - /// again at a packet boundary, which we can't determine from the outside. + /// Stops the capturing of the outgoing data packets, so that nothing is copied anymore + /// when nobody is watching. The applies this at the next + /// packet boundary, so a capture never starts or ends in the middle of a data packet. /// private void StopCapturing() { - if (Interlocked.Exchange(ref this._outgoingCapture, null) is { } capture) + lock (this._captureLifecycleLock) { + if (this._outgoingCapture is not { } capture) + { + return; + } + + this._outgoingCapture = null; if (this._outputWriter is { } outputWriter) { - outputWriter.CaptureWriter = null; + outputWriter.PendingCaptureWriter = null; } capture.Stop(); diff --git a/src/Network/ExtendedPipeWriter.cs b/src/Network/ExtendedPipeWriter.cs index 09100ad49e..30c14506f5 100644 --- a/src/Network/ExtendedPipeWriter.cs +++ b/src/Network/ExtendedPipeWriter.cs @@ -18,6 +18,8 @@ public class ExtendedPipeWriter : PipeWriter private Memory _lastBuffer; + private PipeWriter? _activeCaptureWriter; + /// /// Initializes a new instance of the class. /// @@ -30,10 +32,15 @@ public ExtendedPipeWriter(PipeWriter target, Counter writeCounter) } /// - /// Gets or sets the writer of the packet capture, into which the written data is copied. - /// If it's , nothing is captured. + /// Gets or sets the writer of the packet capture, into which the written data should be + /// copied. If it's , nothing should be captured. /// - internal PipeWriter? CaptureWriter { get; set; } + /// + /// It's only applied at a packet boundary, because the capture would start or end in the + /// middle of a data packet otherwise. It can be set by any thread; the switch itself + /// happens on the thread which writes to this instance. + /// + internal PipeWriter? PendingCaptureWriter { get; set; } /// public override void Complete(Exception? exception = null) @@ -50,18 +57,21 @@ public override void CancelPendingFlush() /// public override ValueTask FlushAsync(CancellationToken cancellationToken = default) { - if (this.CaptureWriter is { } captureWriter) + if (this._activeCaptureWriter is { } captureWriter) { return this.FlushWithCaptureAsync(captureWriter, cancellationToken); } + // After a flush, the next written data starts a new packet, so a requested capture + // can start here. + this._activeCaptureWriter = this.PendingCaptureWriter; return this._target.FlushAsync(cancellationToken); } /// public override void Advance(int bytes) { - if (this.CaptureWriter is { } captureWriter && bytes > 0 && this._lastBuffer.Length >= bytes) + if (this._activeCaptureWriter is { } captureWriter && bytes > 0 && this._lastBuffer.Length >= bytes) { // The data has to be copied before it's advanced, because the target may recycle // the buffer afterwards. The capturing must never break the connection, so a @@ -88,6 +98,7 @@ public override void Advance(int bytes) /// public override Memory GetMemory(int sizeHint = 0) { + this.ApplyPendingCaptureWriterAtPacketBoundary(); var memory = this._target.GetMemory(sizeHint); this._lastBuffer = memory; return memory; @@ -96,7 +107,8 @@ public override Memory GetMemory(int sizeHint = 0) /// public override Span GetSpan(int sizeHint = 0) { - if (this.CaptureWriter is null) + this.ApplyPendingCaptureWriterAtPacketBoundary(); + if (this._activeCaptureWriter is null) { // We don't remember the buffer in this case. That way, a capture which gets // attached between this call and the next Advance doesn't report stale data. @@ -113,6 +125,20 @@ public override Span GetSpan(int sizeHint = 0) return memorySpan; } + /// + /// Applies a requested change of the capture, when nothing is written to the target yet. + /// In that case we're at a packet boundary, so a starting capture doesn't begin in the + /// middle of a data packet. + /// + private void ApplyPendingCaptureWriterAtPacketBoundary() + { + if (!ReferenceEquals(this._activeCaptureWriter, this.PendingCaptureWriter) + && this._target is { CanGetUnflushedBytes: true, UnflushedBytes: 0 }) + { + this._activeCaptureWriter = this.PendingCaptureWriter; + } + } + private async ValueTask FlushWithCaptureAsync(PipeWriter captureWriter, CancellationToken cancellationToken) { try @@ -124,6 +150,10 @@ private async ValueTask FlushWithCaptureAsync(PipeWriter captureWri // The capture has been completed in the meantime. } + // After a flush, the next written data starts a new packet, so a requested change of + // the capture can be applied here. + this._activeCaptureWriter = this.PendingCaptureWriter; + return await this._target.FlushAsync(cancellationToken).ConfigureAwait(false); } } \ No newline at end of file diff --git a/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs index 8941cb364c..17bafa4af4 100644 --- a/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs +++ b/tests/MUnique.OpenMU.Network.Tests/PacketCaptureTest.cs @@ -201,6 +201,37 @@ public async Task NothingIsCapturedAfterLastSinkWasRemovedAsync() Assert.That(sink.Snapshot(), Is.Empty); } + /// + /// Tests if the capturing can be started again after it has been stopped, because the + /// last sink was removed. + /// + /// The async task. + [Test] + public async Task CapturingCanBeRestartedAsync() + { + var packet = new byte[] { 0xC1, 0x04, 0xF1, 0x00 }; + using var connection = CreateConnection(); + var sink = new CapturingSink(); + + connection.AddCaptureSink(sink); + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + await sink.WaitForPacketsAsync(1).ConfigureAwait(false); + Assert.That(sink.Snapshot(), Has.Count.EqualTo(1)); + + connection.RemoveCaptureSink(sink); + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + await Task.Delay(50).ConfigureAwait(false); + Assert.That(sink.Snapshot(), Has.Count.EqualTo(1), "Nothing should be captured without a sink."); + + var secondSink = new CapturingSink(); + connection.AddCaptureSink(secondSink); + await connection.Output.WriteAsync(packet).ConfigureAwait(false); + var captured = await secondSink.WaitForPacketsAsync(1).ConfigureAwait(false); + + Assert.That(captured, Has.Count.EqualTo(1)); + Assert.That(captured[0].Packet, Is.EqualTo(packet)); + } + /// /// Tests if the written data still arrives at the target pipe unchanged, with and without /// a registered capture sink.