/* ======================================================================== * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Buffers; using System.Buffers.Binary; using System.Collections.Generic; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Opc.Ua.Bindings; using Opc.Ua.Bindings.Pcap.Capture; namespace Opc.Ua.Bindings.Pcap.Frame { /// /// Splits TCP byte streams into discrete OPC UA UA-SC chunks. /// public sealed class OpcUaFrameParser { /// /// Constructs an OPC UA frame parser. /// public OpcUaFrameParser(ILogger? logger = null) { m_logger = logger ?? NullLogger.Instance; } /// /// Processes one TCP segment and yields all complete OPC UA chunks that become available. /// public IEnumerable Process(TcpFlowSegment segment) { if (segment.Data.IsEmpty) { yield break; } FlowBuffer buffer = GetBuffer(segment.FlowKey); buffer.Append(segment.Data.Span); while (buffer.Length >= HeaderLength) { ReadOnlySpan span = buffer.WrittenSpan; uint messageType = BinaryPrimitives.ReadUInt32LittleEndian(span); uint size = BinaryPrimitives.ReadUInt32LittleEndian(span[4..]); if (TcpMessageType.IsValid(messageType) && size >= HeaderLength && size <= MaxChunkSize) { if (buffer.Length < size) { yield break; } byte[] chunk = span[..checked((int)size)].ToArray(); buffer.Consume(checked((int)size)); yield return new OpcUaChunk( segment.Timestamp, segment.FlowKey, segment.SourceEndpoint, segment.DestinationEndpoint, messageType, IsClientToServer(segment, messageType), chunk); continue; } int resyncOffset = FindNextValidStart(span[1..]); int consume = resyncOffset < 0 ? Math.Max(1, buffer.Length - HeaderLength + 1) : resyncOffset + 1; m_logger.LogWarning("Skipped {ByteCount} bytes while resynchronizing OPC UA chunk stream.", consume); buffer.Consume(consume); } } private FlowBuffer GetBuffer(string flowKey) { if (!m_buffers.TryGetValue(flowKey, out FlowBuffer? buffer)) { buffer = new FlowBuffer(); m_buffers.Add(flowKey, buffer); } return buffer; } private static bool IsClientToServer(TcpFlowSegment segment, uint messageType) { if (messageType == TcpMessageType.Hello) { return true; } if (messageType == TcpMessageType.Acknowledge) { return false; } return IsOpcUaServerEndpoint(segment.DestinationEndpoint); } private static bool IsOpcUaServerEndpoint(string endpoint) { int separator = endpoint.LastIndexOf(':'); if (separator < 0 || !ushort.TryParse(endpoint[(separator + 1)..], out ushort port)) { return false; } return port == 4840 || port is >= 48010 and <= 48020; } private static int FindNextValidStart(ReadOnlySpan span) { for (int index = 0; index + HeaderLength <= span.Length; index++) { uint messageType = BinaryPrimitives.ReadUInt32LittleEndian(span[index..]); uint size = BinaryPrimitives.ReadUInt32LittleEndian(span[(index + 4)..]); if (TcpMessageType.IsValid(messageType) && size >= HeaderLength && size <= MaxChunkSize) { return index; } } return -1; } private const int HeaderLength = 8; private const uint MaxChunkSize = 64U * 1024U * 1024U; private readonly ILogger m_logger; private readonly Dictionary m_buffers = []; internal sealed class FlowBuffer { /// /// Maximum total size of the per-flow reassembly buffer. Defends /// against a malicious source that emits fragmented TCP streams /// designed to force unbounded buffer growth. /// public const int MaxBufferBytes = 256 * 1024 * 1024; // 256 MB internal FlowBuffer() : this(DefaultBufferBytes) { } internal FlowBuffer(int initialCapacity) { m_buffer = ArrayPool.Shared.Rent(initialCapacity); } internal int Length { get; private set; } internal int Capacity => m_buffer.Length; internal ReadOnlySpan WrittenSpan => m_buffer.AsSpan(0, Length); internal void Append(ReadOnlySpan data) { EnsureCapacity(Length + data.Length); data.CopyTo(m_buffer.AsSpan(Length)); Length += data.Length; } internal void Consume(int count) { if (count >= Length) { Length = 0; return; } m_buffer.AsSpan(count, Length - count).CopyTo(m_buffer); Length -= count; } internal void EnsureCapacity(int capacity) { if (capacity <= m_buffer.Length) { return; } if (capacity > MaxBufferBytes) { throw new PcapDiagnosticsException( $"FlowBuffer capacity {capacity} would exceed " + $"MaxBufferBytes ({MaxBufferBytes}). A TCP flow has " + "produced more pending reassembly bytes than the parser " + "allows; refusing to grow further."); } int newLength = Math.Min( MaxBufferBytes, Math.Max(capacity, m_buffer.Length * 2)); byte[] newBuffer = ArrayPool.Shared.Rent(newLength); m_buffer.AsSpan(0, Length).CopyTo(newBuffer); ArrayPool.Shared.Return(m_buffer); m_buffer = newBuffer; } private const int DefaultBufferBytes = 8192; private byte[] m_buffer; } } /// /// A complete OPC UA UA-SC chunk extracted from a TCP byte stream. /// public readonly struct OpcUaChunk : IEquatable { /// /// Constructs an OPC UA chunk record. /// public OpcUaChunk( DateTimeOffset timestamp, string flowKey, string sourceEndpoint, string destinationEndpoint, uint messageType, bool isClientToServer, ReadOnlyMemory data) { Timestamp = timestamp; FlowKey = flowKey; SourceEndpoint = sourceEndpoint; DestinationEndpoint = destinationEndpoint; MessageType = messageType; IsClientToServer = isClientToServer; Data = data; } /// /// Chunk timestamp. /// public DateTimeOffset Timestamp { get; } /// /// TCP flow key. /// public string FlowKey { get; } /// /// Source endpoint. /// public string SourceEndpoint { get; } /// /// Destination endpoint. /// public string DestinationEndpoint { get; } /// /// OPC UA TCP message type. /// public uint MessageType { get; } /// /// Whether this chunk appears to flow from client to server. /// public bool IsClientToServer { get; } /// /// Full chunk bytes including the 8-byte UA-SC header. /// public ReadOnlyMemory Data { get; } /// public bool Equals(OpcUaChunk other) { return Timestamp.Equals(other.Timestamp) && string.Equals(FlowKey, other.FlowKey, StringComparison.Ordinal) && string.Equals(SourceEndpoint, other.SourceEndpoint, StringComparison.Ordinal) && string.Equals(DestinationEndpoint, other.DestinationEndpoint, StringComparison.Ordinal) && MessageType == other.MessageType && IsClientToServer == other.IsClientToServer && Data.Span.SequenceEqual(other.Data.Span); } /// public override bool Equals(object? obj) { return obj is OpcUaChunk other && Equals(other); } /// public override int GetHashCode() { return HashCode.Combine( Timestamp, FlowKey, SourceEndpoint, DestinationEndpoint, MessageType, IsClientToServer, Data.Length); } /// /// Equality comparison. /// public static bool operator ==(OpcUaChunk left, OpcUaChunk right) => left.Equals(right); /// /// Inequality comparison. /// public static bool operator !=(OpcUaChunk left, OpcUaChunk right) => !left.Equals(right); } }