From 62421e5f2c61c8911ccb064615ef72678689a553 Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 9 Oct 2026 11:14:45 +0200 Subject: [PATCH 1/4] Secure the UADP DataSetMetaData announcement on secured connections MetaDataPublisher encoded the DataSetMetaData announcement itself and called IPubSubTransport.SendAsync directly. That bypassed the connection's security wrapper and its fail-closed check, so a WriterGroup configured for SignAndEncrypt announced its DataSet layout (field names, types, version) unsigned and unencrypted. A secured subscriber dropped the announcement ("Dropping unsecured inbound frame ... requiring SignAndEncrypt"), and any listener could read it. OPC 10000-14 6.2.5.2 applies the WriterGroup's SecurityMode to the NetworkMessages it publishes, and 7.2.4.6.3 carries the security of a discovery announcement in its SecurityHeader. Cause, confirmed on the wire: the documented udp-uadp exchange sent eight secured DataSetMessages and one DiscoveryResponse (announcement type 2, DataSetMetaData) with ExtendedFlags1 = 0x81, i.e. without the SecurityHeader bit. Fix: - MetaDataPublisher builds the announcement and hands it to the connection (PubSubConnection.SendMetaDataAnnouncementAsync), which applies the connection's message security, or refuses to send when it cannot secure the message. - UadpDiscoveryCoder can encode a discovery message with the SecurityHeader bit and reports the security boundary after the PublisherId (and DataSetClassId), where UadpDecoder.TryReadOuterPrefix ends the prefix. UadpEncoder.EncodeWithSecurityBoundary accepts discovery messages. - PubSubConnection.SendNetworkMessageAsync wraps UADP discovery responses. Before, a secured connection refused them (metadata and writer configuration probe responses, and the MQTT discovery topics); now it sends them secured. The receive path is unchanged: it already unwraps secured discovery frames and registers authenticated metadata. Tests: - MetaDataPublisherTests.UadpPathSecuresTheAnnouncementOfASecuredWriterGroupAsync: the announcement of a SignAndEncrypt WriterGroup carries a SecurityHeader, does not contain the DataSet name in cleartext, and unwraps and decodes with the subscriber's keys. Red before the fix (no SecurityHeader). - UadpDiscoveryTests.SecuredDiscoveryMessagesSplitWhereTheReceiverExpects: for requests and responses, with and without a DataSetClassId, the secured encoding differs from the unsecured one only by the SecurityHeader flag, and its boundary equals the receiver's prefix length. Validation: the documented exchange (with the sample KeyNonce fix applied) now sends the announcement secured, logs no unsecured-frame warning, and the subscriber logs "Registered inbound metadata for writer 1". docs/PubSub.md no longer lists the unsecured-frame warning and describes the secured announcement. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/PubSub.md | 22 ++-- .../Application/MetaDataPublisher.cs | 107 ++++-------------- .../Application/PubSubApplication.cs | 2 - .../Connections/PubSubConnection.cs | 23 +++- .../Encoding/Uadp/UadpDiscoveryCoder.cs | 69 +++++++---- .../Encoding/Uadp/UadpEncoder.cs | 12 +- .../Application/MetaDataPublisherTests.cs | 86 +++++++++++++- .../Encoding/Uadp/UadpDiscoveryTests.cs | 55 +++++++++ 8 files changed, 248 insertions(+), 128 deletions(-) diff --git a/docs/PubSub.md b/docs/PubSub.md index d7333b2610..bbd847d2c4 100644 --- a/docs/PubSub.md +++ b/docs/PubSub.md @@ -95,17 +95,13 @@ DataSet #1 received (3 fields): BoolToggle=False, Int32=1, DateTime=09/28/2026 1 DataSet #2 received (3 fields): BoolToggle=True, Int32=2, DateTime=09/28/2026 12:18:51 ``` -The `udp-uadp` profile signs and encrypts the UADP messages with demo keys -that are built into the sample. A production deployment obtains its keys from -a [Security Key Service](#security-key-service-sks). Two warnings are expected: - -- `DataSetReader Reader 1 faulted on MessageReceiveTimeout (>00:00:05)` - appears when no matching message arrives for five seconds, for example - before the publisher starts. The reader resumes when messages arrive. -- `Dropping unsecured inbound frame ... requiring SignAndEncrypt` appears once. - The publisher announces its DataSet metadata without message security, and - the secured subscriber discards the announcement. It decodes the DataSets - with the metadata configured on its DataSetReader instead. +The `udp-uadp` profile signs and encrypts the UADP messages, including the +DataSet metadata announcement, with demo keys that are built into the sample. +A production deployment obtains its keys from a +[Security Key Service](#security-key-service-sks). +`DataSetReader Reader 1 faulted on MessageReceiveTimeout (>00:00:05)` can +appear when no matching message arrives for five seconds, for example before +the publisher starts. The reader resumes when messages arrive. The sample's default endpoint, `opc.udp://239.0.0.1:4840`, is a multicast address. On one computer, the subscriber can miss multicast messages: the UDP @@ -289,7 +285,9 @@ The publisher-side `MetaDataPublisher` ([§6.2.2.5](https://reference.opcfoundat emits a retained `JsonMetaDataMessage` / `UadpDiscoveryResponseMessage` on the well-known `ua-metadata` topic at startup and after each configuration version bump; subscribers cache it before the first -KeyFrame arrives. +KeyFrame arrives. A connection configured for message security signs and +encrypts the UADP announcement with the same keys as its DataSetMessages, +so only subscribers that hold the keys can read the DataSet layout. A subscriber also registers the `DataSetMetaData` configured on each DataSetReader whose filter names an exact PublisherId, WriterGroupId, and diff --git a/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs b/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs index 9c162a9c3f..1669edcbaf 100644 --- a/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs +++ b/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs @@ -75,6 +75,12 @@ namespace Opc.Ua.PubSub.Application /// . /// /// + /// Announcements go through the connection's send pipeline, so a + /// connection configured for message security signs and encrypts + /// them with the same keys as its DataSetMessages, and refuses to + /// send them when it cannot secure them. + /// + /// /// Lifetime is owned by : started /// after EnableConnectionsAsync returns, disposed before /// the connections are torn down. @@ -84,9 +90,6 @@ internal sealed class MetaDataPublisher : IAsyncDisposable { private readonly PubSubApplication m_application; private readonly IDataSetMetaDataRegistry m_registry; - private readonly IReadOnlyDictionary m_encoders; - private readonly IPubSubDiagnostics m_diagnostics; - private readonly ITelemetryContext m_telemetry; private readonly BackgroundTaskScope m_backgroundWork; private readonly TimeProvider m_timeProvider; private readonly ILogger m_logger; @@ -106,17 +109,11 @@ internal sealed class MetaDataPublisher : IAsyncDisposable /// the matching transport per writer group. /// /// Shared metadata registry. - /// - /// Network-message encoders keyed by transport profile URI. - /// - /// Diagnostics sink. /// Telemetry context. /// Clock used to stamp MessageIds. public MetaDataPublisher( PubSubApplication application, IDataSetMetaDataRegistry metaDataRegistry, - IReadOnlyDictionary encoders, - IPubSubDiagnostics diagnostics, ITelemetryContext telemetry, TimeProvider timeProvider) { @@ -128,14 +125,6 @@ public MetaDataPublisher( { throw new ArgumentNullException(nameof(metaDataRegistry)); } - if (encoders is null) - { - throw new ArgumentNullException(nameof(encoders)); - } - if (diagnostics is null) - { - throw new ArgumentNullException(nameof(diagnostics)); - } if (telemetry is null) { throw new ArgumentNullException(nameof(telemetry)); @@ -146,9 +135,6 @@ public MetaDataPublisher( } m_application = application; m_registry = metaDataRegistry; - m_encoders = encoders; - m_diagnostics = diagnostics; - m_telemetry = telemetry; m_backgroundWork = new BackgroundTaskScope(nameof(MetaDataPublisher), telemetry); m_timeProvider = timeProvider; m_logger = telemetry.CreateLogger(); @@ -497,22 +483,16 @@ private async ValueTask PublishMetaDataAsync( { return; } - string profile = connection.TransportProfileUri; - string family = TransportProfileFamily(profile); Uuid classId = metaData.DataSetClassId == Guid.Empty ? Uuid.Empty : new Uuid(metaData.DataSetClassId); - ReadOnlyMemory payload; - string? topic; - if (string.Equals(family, "Json", StringComparison.Ordinal)) + PubSubNetworkMessage message; + if (string.Equals( + TransportProfileFamily(connection.TransportProfileUri), + "Json", + StringComparison.Ordinal)) { - if (!TryResolveEncoder(profile, family, out INetworkMessageEncoder? encoder) || - encoder is null) - { - m_logger.NoJsonEncoderRegistered(profile); - return; - } - var message = new JsonMetaDataMessage + message = new JsonMetaDataMessage { MessageId = NewMessageId(), PublisherId = connection.PublisherId, @@ -530,22 +510,10 @@ private async ValueTask PublishMetaDataAsync( MetaData = metaData, MetaDataPayload = metaData }; - var context = new PubSubNetworkMessageContext( - ServiceMessageContext.CreateEmpty(m_telemetry), - m_registry, - m_diagnostics, - m_timeProvider); - payload = await encoder.EncodeAsync(message, context, cancellationToken) - .ConfigureAwait(false); - topic = ResolveMetaDataTopic( - transport, - connection.PublisherId, - writerGroup.WriterGroupId, - writer.DataSetWriterId); } else { - var message = new UadpDiscoveryResponseMessage + message = new UadpDiscoveryResponseMessage { PublisherId = connection.PublisherId, WriterGroupId = writerGroup.WriterGroupId, @@ -556,44 +524,17 @@ private async ValueTask PublishMetaDataAsync( SequenceNumber = NewSequenceNumber(), StatusCode = StatusCodes.Good }; - var context = new PubSubNetworkMessageContext( - ServiceMessageContext.CreateEmpty(m_telemetry), - m_registry, - m_diagnostics, - m_timeProvider); - payload = UadpDiscoveryCoder.Encode(message, context); - topic = ResolveMetaDataTopic( - transport, - connection.PublisherId, - writerGroup.WriterGroupId, - writer.DataSetWriterId); } + string? topic = ResolveMetaDataTopic( + transport, + connection.PublisherId, + writerGroup.WriterGroupId, + writer.DataSetWriterId); - await transport.SendAsync(payload, topic, cancellationToken).ConfigureAwait(false); - } - - private bool TryResolveEncoder( - string profile, - string family, - out INetworkMessageEncoder? encoder) - { - if (m_encoders.TryGetValue(profile, out encoder)) - { - return true; - } - foreach (KeyValuePair entry in m_encoders) - { - if (string.Equals( - TransportProfileFamily(entry.Key), - family, - StringComparison.Ordinal)) - { - encoder = entry.Value; - return true; - } - } - encoder = null; - return false; + // The connection applies its message security, so a secured + // connection never announces the DataSet layout in the clear. + await connection.SendMetaDataAnnouncementAsync(message, topic, cancellationToken) + .ConfigureAwait(false); } private static string? ResolveMetaDataTopic( @@ -681,10 +622,6 @@ public static partial void FailedToPublishMetadataChange( Exception exception, ushort writer, ushort group); - - [LoggerMessage(EventId = PubSubEventIds.MetaDataPublisher + 2, Level = LogLevel.Debug, - Message = "No JSON encoder registered for {Profile}; metadata publish skipped.")] - public static partial void NoJsonEncoderRegistered(this ILogger logger, string? profile); } } diff --git a/src/Opc.Ua.PubSub/Application/PubSubApplication.cs b/src/Opc.Ua.PubSub/Application/PubSubApplication.cs index 1b8a3f3ee7..f31bef344e 100644 --- a/src/Opc.Ua.PubSub/Application/PubSubApplication.cs +++ b/src/Opc.Ua.PubSub/Application/PubSubApplication.cs @@ -662,8 +662,6 @@ public async ValueTask StartAsync(CancellationToken cancellationToken = default) var metaDataPublisher = new MetaDataPublisher( this, MetaDataRegistry, - m_encoderMap, - m_aggregatingDiagnostics, m_telemetry, m_timeProvider); try diff --git a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs index 2ad26f42e0..cd8b78ec5c 100644 --- a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs +++ b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs @@ -2454,6 +2454,23 @@ private ValueTask SendNetworkMessageAsync( return SendNetworkMessageAsync(networkMessage, topic, writerGroup: null, cancellationToken); } + /// + /// Sends a DataSetMetaData announcement through the connection's send + /// pipeline, so a connection configured for message security signs and + /// encrypts the announcement like its DataSetMessages and never sends it + /// in the clear. + /// + /// UADP discovery response or JSON metadata message. + /// Broker topic, or for datagram transports. + /// Cancellation token. + internal ValueTask SendMetaDataAnnouncementAsync( + PubSubNetworkMessage announcement, + string? topic, + CancellationToken cancellationToken) + { + return SendNetworkMessageAsync(announcement, topic, writerGroup: null, cancellationToken); + } + private async ValueTask SendNetworkMessageAsync( PubSubNetworkMessage networkMessage, string? topic, @@ -2483,9 +2500,9 @@ private async ValueTask SendNetworkMessageAsync( ReadOnlyMemory payload; if (m_securityWrapper is not null && - networkMessage is UadpNetworkMessage uadp) + networkMessage is UadpNetworkMessage or UadpDiscoveryResponseMessage) { - payload = await EncodeAndWrapUadpAsync(uadp, context, cancellationToken) + payload = await EncodeAndWrapUadpAsync(networkMessage, context, cancellationToken) .ConfigureAwait(false); } else if (RequiresInboundSecurity || @@ -2685,7 +2702,7 @@ await transport.SendAsync(frame, topic, cancellationToken) } private async ValueTask> EncodeAndWrapUadpAsync( - UadpNetworkMessage message, + PubSubNetworkMessage message, PubSubNetworkMessageContext context, CancellationToken cancellationToken) { diff --git a/src/Opc.Ua.PubSub/Encoding/Uadp/UadpDiscoveryCoder.cs b/src/Opc.Ua.PubSub/Encoding/Uadp/UadpDiscoveryCoder.cs index 688c9ef345..caa0a0c986 100644 --- a/src/Opc.Ua.PubSub/Encoding/Uadp/UadpDiscoveryCoder.cs +++ b/src/Opc.Ua.PubSub/Encoding/Uadp/UadpDiscoveryCoder.cs @@ -119,6 +119,31 @@ public static class UadpDiscoveryCoder public static byte[] Encode( PubSubNetworkMessage message, PubSubNetworkMessageContext context) + { + return Encode(message, context, securityEnabled: false, out _); + } + + /// + /// Encodes a discovery NetworkMessage and reports the byte offset + /// at which the security wrapper inserts the SecurityHeader: the + /// discovery header and payload that follow the PublisherId are + /// signed and encrypted (Part 14 §7.2.4.6.3). + /// + /// Source message; must be a + /// or + /// . + /// Network message context. + /// Sets the SecurityHeader bit in + /// ExtendedFlags1. + /// Boundary between the outer prefix + /// and the inner payload. + /// + /// + internal static byte[] Encode( + PubSubNetworkMessage message, + PubSubNetworkMessageContext context, + bool securityEnabled, + out int payloadOffset) { if (message is null) { @@ -131,9 +156,9 @@ public static byte[] Encode( return message switch { UadpDiscoveryRequestMessage request => - EncodeRequest(request, context), + EncodeRequest(request, context, securityEnabled, out payloadOffset), UadpDiscoveryResponseMessage response => - EncodeResponse(response, context), + EncodeResponse(response, context, securityEnabled, out payloadOffset), _ => throw new InvalidOperationException( "Discovery encoding requires a UadpDiscoveryRequestMessage " + "or UadpDiscoveryResponseMessage instance.") @@ -177,13 +202,17 @@ public static byte[] Encode( private static byte[] EncodeRequest( UadpDiscoveryRequestMessage message, - PubSubNetworkMessageContext context) + PubSubNetworkMessageContext context, + bool securityEnabled, + out int payloadOffset) { byte[] buffer = new byte[1024]; var writer = new UadpBinaryWriter(buffer, 0, buffer.Length); UadpDiscoveryWire.WriteCommonHeader( ref writer, message, - ExtendedFlags2EncodingMask.NetworkMessageWithDiscoveryRequest); + ExtendedFlags2EncodingMask.NetworkMessageWithDiscoveryRequest, + securityEnabled); + payloadOffset = writer.Position; writer.WriteByte((byte)message.DiscoveryType); writer.WriteUInt32Le((uint)message.DataSetWriterIds.Count); @@ -201,13 +230,17 @@ private static byte[] EncodeRequest( private static byte[] EncodeResponse( UadpDiscoveryResponseMessage message, - PubSubNetworkMessageContext context) + PubSubNetworkMessageContext context, + bool securityEnabled, + out int payloadOffset) { byte[] buffer = new byte[8192]; var writer = new UadpBinaryWriter(buffer, 0, buffer.Length); UadpDiscoveryWire.WriteCommonHeader( ref writer, message, - ExtendedFlags2EncodingMask.NetworkMessageWithDiscoveryResponse); + ExtendedFlags2EncodingMask.NetworkMessageWithDiscoveryResponse, + securityEnabled); + payloadOffset = writer.Position; writer.WriteByte((byte)message.DiscoveryType); writer.WriteUInt16Le(message.SequenceNumber); @@ -790,21 +823,25 @@ internal static class UadpDiscoveryWire public static void WriteCommonHeader( ref UadpBinaryWriter writer, UadpDiscoveryRequestMessage message, - ExtendedFlags2EncodingMask discoveryBit) + ExtendedFlags2EncodingMask discoveryBit, + bool securityEnabled) { WriteCommonHeader( ref writer, message.UadpVersion, message.PublisherId, - message.DataSetClassId, discoveryBit); + message.DataSetClassId, discoveryBit, securityEnabled, + payloadHeaderEnabled: false, writerGroupId: null); } public static void WriteCommonHeader( ref UadpBinaryWriter writer, UadpDiscoveryResponseMessage message, - ExtendedFlags2EncodingMask discoveryBit) + ExtendedFlags2EncodingMask discoveryBit, + bool securityEnabled) { WriteCommonHeader( ref writer, message.UadpVersion, message.PublisherId, - message.DataSetClassId, discoveryBit); + message.DataSetClassId, discoveryBit, securityEnabled, + payloadHeaderEnabled: false, writerGroupId: null); } public static void WriteCommonHeader( @@ -822,18 +859,6 @@ public static void WriteCommonHeader( securityEnabled, payloadHeaderEnabled, writerGroupId); } - private static void WriteCommonHeader( - ref UadpBinaryWriter writer, - byte uadpVersion, - PublisherId publisherId, - Uuid dataSetClassId, - ExtendedFlags2EncodingMask discoveryBit) - { - WriteCommonHeaderCore( - ref writer, uadpVersion, publisherId, dataSetClassId, discoveryBit, - securityEnabled: false, payloadHeaderEnabled: false, writerGroupId: null); - } - private static void WriteCommonHeaderCore( ref UadpBinaryWriter writer, byte uadpVersion, diff --git a/src/Opc.Ua.PubSub/Encoding/Uadp/UadpEncoder.cs b/src/Opc.Ua.PubSub/Encoding/Uadp/UadpEncoder.cs index d2f5ab9541..1950ae216f 100644 --- a/src/Opc.Ua.PubSub/Encoding/Uadp/UadpEncoder.cs +++ b/src/Opc.Ua.PubSub/Encoding/Uadp/UadpEncoder.cs @@ -147,12 +147,12 @@ public static ReadOnlyMemory EncodeWithSecurityBoundary( } /// - /// Encodes a UADP data or action NetworkMessage with the + /// Encodes a UADP data, action or discovery NetworkMessage with the /// ExtendedFlags1.SecurityEnabled bit set and reports the /// byte offset at which the security wrapper must insert the /// SecurityHeader. /// - /// UADP data or action message. + /// UADP data, action or discovery message. /// Network message context. /// Boundary between outer prefix and inner payload. /// @@ -171,8 +171,14 @@ public static ReadOnlyMemory EncodeWithSecurityBoundary( return UadpActionCoder.Encode( networkMessage, context, securityEnabled: true, out payloadOffset); } + if (networkMessage is UadpDiscoveryRequestMessage + or UadpDiscoveryResponseMessage) + { + return UadpDiscoveryCoder.Encode( + networkMessage, context, securityEnabled: true, out payloadOffset); + } throw new ArgumentException( - "Security wrapping is supported for UADP data and action messages.", + "Security wrapping is supported for UADP data, action and discovery messages.", nameof(networkMessage)); } diff --git a/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs b/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs index b1b1f7fed5..de049ec946 100644 --- a/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs @@ -42,6 +42,9 @@ using Opc.Ua.PubSub.Encoding.Json; using Opc.Ua.PubSub.Encoding.Uadp; using Opc.Ua.PubSub.MetaData; +using Opc.Ua.PubSub.Security; +using Opc.Ua.PubSub.Security.Policies; +using Opc.Ua.PubSub.Tests.Security; using Opc.Ua.PubSub.Transports; using Opc.Ua.Tests; @@ -73,6 +76,8 @@ public class MetaDataPublisherTests private const ushort PublisherIdValue = 17; private const ushort WriterGroupIdValue = 7; private const ushort DataSetWriterIdValue = 42; + private const string SecurityGroupIdValue = "metadata-group"; + private const uint SecurityTokenIdValue = 1; [Test] public async Task OnStartup_PublishesMetaData_ToMatchingTransport() @@ -166,6 +171,60 @@ await WaitUntilAsync( Assert.That(response.DataSetWriterId, Is.EqualTo(DataSetWriterIdValue)); } + /// + /// The announcement of a WriterGroup configured for SignAndEncrypt is + /// secured with the connection's keys like its DataSetMessages, so a + /// subscriber holding the keys can verify and decrypt it and nobody + /// else can read the DataSet layout. + /// + [Test] + public async Task UadpPathSecuresTheAnnouncementOfASecuredWriterGroupAsync() + { + using PubSubSecurityKeyRing publisherRing = NewKeyRing(); + using PubSubSecurityKeyRing subscriberRing = NewKeyRing(); + var factory = new RecordingTransportFactory(UadpProfile, supportsTopics: false); + await using IPubSubApplication app = BuildApp( + UadpProfile, + factory, + securityKeyProvider: new StaticSecurityKeyProvider(SecurityGroupIdValue, publisherRing)); + + await app.StartAsync(CancellationToken.None).ConfigureAwait(false); + await WaitUntilAsync( + () => factory.Transport is { } t && t.Sends.Count >= 1, + TimeSpan.FromSeconds(2)).ConfigureAwait(false); + byte[] payload = factory.Transport!.Sends[0].Payload.ToArray(); + + Assert.That( + UadpDecoder.TryReadOuterPrefix(payload, out int prefixLength, out bool securityEnabled, out _, out _), + Is.True); + Assert.That(securityEnabled, Is.True, "The announcement must carry a SecurityHeader."); + Assert.That( + payload.AsSpan().IndexOf(System.Text.Encoding.UTF8.GetBytes("pds-1")), + Is.EqualTo(-1), + "The DataSet name must not appear in cleartext."); + + var window = new SecurityTokenWindow(); + window.RegisterToken(SecurityTokenIdValue); + using var nonceProvider = new RandomNonceProvider(PublisherId.FromUInt16(PublisherIdValue)); + var subscriber = new UadpSecurityWrapper( + PubSubAes256CtrPolicy.Instance, + new StaticSecurityKeyProvider(SecurityGroupIdValue, subscriberRing), + nonceProvider, + window, + NUnitTelemetryContext.Create()); + UadpSecurityWrapper.UnwrapResult unwrapped = await subscriber + .TryUnwrapAsync(payload.AsMemory(0, prefixLength), payload.AsMemory(prefixLength)) + .ConfigureAwait(false); + Assert.That(unwrapped.IsSuccess, Is.True, unwrapped.Reason); + + byte[] cleartext = [.. payload.AsSpan(0, prefixLength), .. unwrapped.InnerPayload!.Value.Span]; + PubSubNetworkMessage? decoded = UadpDecoder.Decode(cleartext, NewDecodeContext()); + Assert.That(decoded, Is.InstanceOf()); + var response = (UadpDiscoveryResponseMessage)decoded!; + Assert.That(response.DiscoveryType, Is.EqualTo(UadpDiscoveryType.DataSetMetaData)); + Assert.That(response.DataSetMetaData?.Name, Is.EqualTo("pds-1")); + } + [Test] public async Task DisposeAsync_UnsubscribesFromRegistry() { @@ -341,7 +400,8 @@ private static IPubSubApplication BuildApp( RecordingTransportFactory factory, IPublishedDataSetSource? source = null, INetworkMessageEncoder? encoder = null, - ushort? secondWriterId = null) + ushort? secondWriterId = null, + IPubSubSecurityKeyProvider? securityKeyProvider = null) { string addressUrl = transportProfileUri == JsonMqttProfile ? "mqtt://localhost:1883" @@ -392,6 +452,16 @@ secondWriterId is null } }) }; + if (securityKeyProvider is not null) + { + WriterGroupDataType writerGroup = connection.WriterGroups[0]; + writerGroup.SecurityMode = MessageSecurityMode.SignAndEncrypt; + writerGroup.SecurityGroupId = SecurityGroupIdValue; + writerGroup.SecurityKeyServices = new ArrayOf(new[] + { + new EndpointDescription { EndpointUrl = "opc.tcp://localhost:4840/SecurityKeyService" } + }); + } var pds = new PublishedDataSetDataType { Name = "pds-1", @@ -426,9 +496,23 @@ secondWriterId is null { builder.AddEncoder(encoder); } + if (securityKeyProvider is not null) + { + builder.AddSecurityKeyProvider(securityKeyProvider); + } return builder.Build(); } + private static PubSubSecurityKeyRing NewKeyRing() + { + var ring = new PubSubSecurityKeyRing(SecurityGroupIdValue); + ring.SetCurrent(TestSecurityKeyFactory.Create( + SecurityTokenIdValue, + PubSubAes256CtrPolicy.Instance.SigningKeyLength, + PubSubAes256CtrPolicy.Instance.EncryptingKeyLength)); + return ring; + } + private static DataSetMetaDataKey NewKey() { return new DataSetMetaDataKey( diff --git a/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs b/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs index 35f1024d82..bfbb905617 100644 --- a/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs @@ -47,6 +47,61 @@ namespace Opc.Ua.PubSub.Tests.Encoding.Uadp [TestSpec("7.2.4.6.9")] public class UadpDiscoveryTests { + /// + /// A secured discovery message differs from the unsecured encoding only + /// by the SecurityHeader flag, and its security boundary is where a + /// receiver's prefix parse ends, so the receiver can verify and decrypt + /// the discovery header and payload. + /// + [Test] + [TestSpec("7.2.4.6.3")] + [TestCase(false)] + [TestCase(true)] + public void SecuredDiscoveryMessagesSplitWhereTheReceiverExpects(bool withDataSetClassId) + { + PubSubNetworkMessageContext context = UadpTestUtilities.NewContext(); + Uuid classId = withDataSetClassId ? (Uuid)Guid.NewGuid() : Uuid.Empty; + PubSubNetworkMessage[] messages = + [ + new UadpDiscoveryResponseMessage + { + PublisherId = PublisherId.FromUInt16(0x4242), + DataSetClassId = classId, + SequenceNumber = 7, + DiscoveryType = UadpDiscoveryType.DataSetMetaData, + DataSetWriterId = 3, + DataSetMetaData = new DataSetMetaDataType { Name = "Secured" }, + StatusCode = StatusCodes.Good + }, + new UadpDiscoveryRequestMessage + { + PublisherId = PublisherId.FromUInt16(0x4242), + DataSetClassId = classId, + DiscoveryType = UadpDiscoveryType.DataSetMetaData, + DataSetWriterIds = new ushort[] { 3 } + } + ]; + + foreach (PubSubNetworkMessage message in messages) + { + byte[] plain = UadpDiscoveryCoder.Encode(message, context); + byte[] secured = UadpEncoder + .EncodeWithSecurityBoundary(message, context, out int payloadOffset) + .ToArray(); + + Assert.That( + UadpDecoder.TryReadOuterPrefix( + secured, out int prefixLength, out bool securityEnabled, out _, out _), + Is.True); + Assert.That(securityEnabled, Is.True); + Assert.That(prefixLength, Is.EqualTo(payloadOffset)); + Assert.That( + secured[1], + Is.EqualTo((byte)(plain[1] | (byte)ExtendedFlags1EncodingMask.SecurityEnabled))); + Assert.That(secured.AsSpan(2).ToArray(), Is.EqualTo(plain.AsSpan(2).ToArray())); + } + } + [Test] public void DiscoveryRequest_DataSetMetaData_RoundTrips() { From 9417ccbbb91c95edc0101c051d2aefb2dda3d005 Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 9 Oct 2026 13:46:40 +0200 Subject: [PATCH 2/4] Secure discovery requests and transport discovery announcements Review feedback: - A secured connection dropped its own discovery requests: the fail-closed check refused every UadpDiscoveryRequestMessage, so a secured subscriber could not ask a publisher for metadata, writer configuration, or endpoints. Requests are now secured like responses on connections that use the UADP encoding. - The ApplicationInformation, PublisherEndpoints, and PubSubConnection announcements that UDP and Ethernet send to their transport-specific discovery destination were still encoded in plaintext. SendDiscoveryResponseAsync now uses the same encode-and-secure step as SendNetworkMessageAsync (EncodeOutboundAsync), which also refuses to send a message that a secured connection cannot protect. The MQTT last-will status message is unchanged: it is registered with the broker when the connection opens and published later, so it cannot carry a current SecurityHeader. docs/PubSub.md describes the secured discovery and this exception. Tests: - PubSubDiscoveryTests.SecuredSubscriberDiscoversSecuredPublisherWithoutPlaintextFramesAsync connects a secured publisher and a secured subscriber through an in-memory transport that also sends announcements to a separate destination. Metadata, writer configuration, and endpoints are discovered, and no frame without a SecurityHeader is sent. Before the fix the metadata request was dropped. - UadpDiscoveryTests.EncodeWithSecurityBoundaryRejectsAMessageItCannotSecure. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/PubSub.md | 5 + .../Connections/PubSubConnection.cs | 95 +++-- .../Application/PubSubDiscoveryTests.cs | 392 ++++++++++++++++-- .../Encoding/Uadp/UadpDiscoveryTests.cs | 15 + 4 files changed, 435 insertions(+), 72 deletions(-) diff --git a/docs/PubSub.md b/docs/PubSub.md index bbd847d2c4..739e3396ea 100644 --- a/docs/PubSub.md +++ b/docs/PubSub.md @@ -1222,6 +1222,11 @@ subscribers that need to find publishers and bind to metadata at runtime. duplicate probes are suppressed, and identical responses are throttled. - MQTT publishes retained discovery messages on the standard status, connection, application, endpoint, and metadata topics. +- A UADP connection configured for message security signs and encrypts its + discovery requests, responses, and announcements with the keys of its + SecurityGroup, so only applications that hold those keys can discover each + other. The MQTT last-will status message is the exception: the broker + publishes it later, so it is sent unsecured. ```csharp PubSubDiscoveryResult result = await application.RequestDiscoveryAsync( diff --git a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs index cd8b78ec5c..48a8dbd594 100644 --- a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs +++ b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs @@ -1873,11 +1873,21 @@ private async ValueTask SendDiscoveryResponseAsync( out IPubSubDiscoveryAnnouncementTransport? announcementTransport) && encoder is not null) { - ReadOnlyMemory payload = await EncodeNetworkMessageAsync( + var context = new PubSubNetworkMessageContext( + ServiceMessageContext.CreateEmpty(m_telemetry), + m_metaDataRegistry, + m_diagnostics, + m_timeProvider); + ReadOnlyMemory? payload = await EncodeOutboundAsync( networkMessage, encoder, + context, cancellationToken).ConfigureAwait(false); - await announcementTransport!.SendDiscoveryAnnouncementAsync(payload, cancellationToken) + if (payload is null) + { + return; + } + await announcementTransport!.SendDiscoveryAnnouncementAsync(payload.Value, cancellationToken) .ConfigureAwait(false); return; } @@ -2498,48 +2508,79 @@ private async ValueTask SendNetworkMessageAsync( m_diagnostics, m_timeProvider); - ReadOnlyMemory payload; - if (m_securityWrapper is not null && - networkMessage is UadpNetworkMessage or UadpDiscoveryResponseMessage) + ReadOnlyMemory? encoded = await EncodeOutboundAsync( + networkMessage, + encoder, + context, + cancellationToken).ConfigureAwait(false); + if (encoded is null) + { + return; + } + ReadOnlyMemory payload = encoded.Value; + + if (m_maxNetworkMessageSize > 0 && + payload.Length > m_maxNetworkMessageSize && + networkMessage is UadpNetworkMessage uadpForChunk) { - payload = await EncodeAndWrapUadpAsync(networkMessage, context, cancellationToken) + await SendChunkedAsync(transport, uadpForChunk, context, writerGroup, cancellationToken) .ConfigureAwait(false); + return; } - else if (RequiresInboundSecurity || + + await transport.SendAsync(payload, topic, cancellationToken) + .ConfigureAwait(false); + } + + /// + /// Encodes a NetworkMessage for sending and applies the message security of + /// the connection. Returns , and the message must not be + /// sent, when the connection requires message security that the message + /// cannot carry. + /// + private async ValueTask?> EncodeOutboundAsync( + PubSubNetworkMessage networkMessage, + INetworkMessageEncoder encoder, + PubSubNetworkMessageContext context, + CancellationToken cancellationToken) + { + if (m_securityWrapper is not null && IsSecurableUadpMessage(networkMessage)) + { + return await EncodeAndWrapUadpAsync(networkMessage, context, cancellationToken) + .ConfigureAwait(false); + } + if (RequiresInboundSecurity || (m_securityWrapper is not null && m_requiredSecurityMode is MessageSecurityMode.Sign or MessageSecurityMode.SignAndEncrypt)) { - // Fail-closed: never emit plaintext for a secured group. - // This path is only reachable for non-UADP messages, which - // the UADP security wrapper cannot protect. + // Fail-closed: never emit plaintext for a secured group. This path is + // reached for messages that the UADP security wrapper cannot protect. m_diagnostics.Increment(PubSubDiagnosticsCounterKind.EncryptionErrors); m_diagnostics.RecordError( StatusCodes.BadSecurityModeRejected, "Refusing to publish an unsecured NetworkMessage on a connection " + "configured for message security."); m_logger.DroppingOutboundMessage(Name, m_requiredSecurityMode); - return; - } - else - { - payload = await encoder.EncodeAsync( - networkMessage, - context, - cancellationToken).ConfigureAwait(false); + return null; } + return await encoder.EncodeAsync(networkMessage, context, cancellationToken) + .ConfigureAwait(false); + } - if (m_maxNetworkMessageSize > 0 && - payload.Length > m_maxNetworkMessageSize && - networkMessage is UadpNetworkMessage uadpForChunk) + /// + /// Returns true for the NetworkMessages that the UADP security wrapper + /// protects: UADP data messages, and the discovery requests and responses + /// of a connection that uses the UADP encoding. + /// + private bool IsSecurableUadpMessage(PubSubNetworkMessage networkMessage) + { + if (networkMessage is UadpNetworkMessage) { - await SendChunkedAsync(transport, uadpForChunk, context, writerGroup, cancellationToken) - .ConfigureAwait(false); - return; + return true; } - - await transport.SendAsync(payload, topic, cancellationToken) - .ConfigureAwait(false); + return networkMessage is UadpDiscoveryRequestMessage or UadpDiscoveryResponseMessage && + TransportProfileFamily(TransportProfileUri) == "Uadp"; } private ValueTask> EncodeNetworkMessageAsync( diff --git a/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs b/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs index 7023b489d1..201fd35a6d 100644 --- a/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs @@ -39,6 +39,9 @@ using Opc.Ua.PubSub.Diagnostics; using Opc.Ua.PubSub.Encoding; using Opc.Ua.PubSub.Encoding.Uadp; +using Opc.Ua.PubSub.Security; +using Opc.Ua.PubSub.Security.Policies; +using Opc.Ua.PubSub.Tests.Security; using Opc.Ua.PubSub.Transports; using Opc.Ua.PubSub.Udp; using Opc.Ua.Tests; @@ -56,6 +59,7 @@ public class PubSubDiscoveryTests private const ushort WriterGroupIdValue = 7; private const ushort DataSetWriterIdValue = 42; private const string PublishedDataSetName = "pds-1"; + private const string SecurityGroupIdValue = "discovery-group"; [Test] public async Task RequestDiscoveryAsyncEncodesRequestAndCollectsResponse() @@ -176,6 +180,71 @@ public async Task UdpLoopbackDiscoveryPublisherAnswersSubscriberRequests() Assert.That(endpoints.PublisherEndpoints[0].EndpointUrl, Is.EqualTo(url)); } + /// + /// A subscriber and a publisher whose connections are secured with the same + /// SecurityGroup exchange discovery requests, responses, and the publisher's + /// transport-specific announcements, and no frame leaves either connection + /// without a SecurityHeader. + /// + [Test] + [TestSpec("7.2.4.6.3")] + public async Task SecuredSubscriberDiscoversSecuredPublisherWithoutPlaintextFramesAsync() + { + const string url = "opc.udp://239.0.0.1:4840"; + using PubSubSecurityKeyRing publisherRing = NewKeyRing(); + using PubSubSecurityKeyRing subscriberRing = NewKeyRing(); + var bus = new LoopbackBus(); + await using IPubSubApplication publisher = BuildPublisherApp( + url, + new LoopbackTransportFactory(bus), + new StaticSecurityKeyProvider(SecurityGroupIdValue, publisherRing)); + await using IPubSubApplication subscriber = BuildSubscriberApp( + url, + new LoopbackTransportFactory(bus), + new StaticSecurityKeyProvider(SecurityGroupIdValue, subscriberRing)); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await publisher.StartAsync(cts.Token).ConfigureAwait(false); + await subscriber.StartAsync(cts.Token).ConfigureAwait(false); + + // Each request collects responses for its whole timeout, so the endpoints + // request is sent after the publisher's startup announcement of its + // endpoints has left the 500 ms duplicate-response window. + PubSubDiscoveryResult metaData = await subscriber.RequestDiscoveryAsync( + new PubSubDiscoveryRequest + { + DiscoveryType = UadpDiscoveryType.DataSetMetaData, + DataSetWriterIds = [DataSetWriterIdValue] + }, + TimeSpan.FromMilliseconds(600), + cts.Token).ConfigureAwait(false); + PubSubDiscoveryResult writerConfiguration = await subscriber.RequestDiscoveryAsync( + new PubSubDiscoveryRequest + { + DiscoveryType = UadpDiscoveryType.DataSetWriterConfiguration, + DataSetWriterIds = [DataSetWriterIdValue] + }, + TimeSpan.FromMilliseconds(600), + cts.Token).ConfigureAwait(false); + PubSubDiscoveryResult endpoints = await subscriber.RequestDiscoveryAsync( + new PubSubDiscoveryRequest + { + DiscoveryType = UadpDiscoveryType.PublisherEndpoints + }, + TimeSpan.FromMilliseconds(600), + cts.Token).ConfigureAwait(false); + + Assert.That(metaData.DataSetMetaDataEntries, Has.Count.EqualTo(1)); + Assert.That( + metaData.DataSetMetaDataEntries[0].DataSetMetaData?.Name, + Is.EqualTo(PublishedDataSetName)); + Assert.That(writerConfiguration.WriterConfigurations, Has.Count.EqualTo(1)); + Assert.That(endpoints.PublisherEndpoints, Is.Not.Empty); + Assert.That(bus.AnnouncementCount, Is.GreaterThan(0), + "The publisher sent no transport-specific discovery announcement."); + Assert.That(bus.PlaintextFrames, Is.Empty, + "Every frame of a secured connection must carry a SecurityHeader."); + } + private static IPubSubApplication BuildDiscoveryOnlyApp(IPubSubTransportFactory factory) { return new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) @@ -203,10 +272,35 @@ private static IPubSubApplication BuildDiscoveryOnlyApp(IPubSubTransportFactory private static IPubSubApplication BuildPublisherApp( string url, - IPubSubTransportFactory factory) + IPubSubTransportFactory factory, + IPubSubSecurityKeyProvider? securityKeyProvider = null) { DataSetMetaDataType metaData = NewMetaData(); - return new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) + var writerGroup = new WriterGroupDataType + { + Name = "writer-group", + WriterGroupId = WriterGroupIdValue, + PublishingInterval = 600_000, + DataSetWriters = + [ + new DataSetWriterDataType + { + Name = "writer", + DataSetWriterId = DataSetWriterIdValue, + DataSetName = PublishedDataSetName + } + ] + }; + var readerGroup = new ReaderGroupDataType + { + Name = "discovery-listener" + }; + if (securityKeyProvider is not null) + { + SecureGroup(writerGroup); + SecureGroup(readerGroup); + } + PubSubApplicationBuilder builder = new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) .WithApplicationId("discovery-publisher") .UseConfiguration(new PubSubConfigurationDataType { @@ -221,31 +315,8 @@ private static IPubSubApplication BuildPublisherApp( { Url = url }), - WriterGroups = - [ - new WriterGroupDataType - { - Name = "writer-group", - WriterGroupId = WriterGroupIdValue, - PublishingInterval = 600_000, - DataSetWriters = - [ - new DataSetWriterDataType - { - Name = "writer", - DataSetWriterId = DataSetWriterIdValue, - DataSetName = PublishedDataSetName - } - ] - } - ], - ReaderGroups = - [ - new ReaderGroupDataType - { - Name = "discovery-listener" - } - ] + WriterGroups = [writerGroup], + ReaderGroups = [readerGroup] } ], PublishedDataSets = @@ -259,35 +330,71 @@ private static IPubSubApplication BuildPublisherApp( }) .AddDataSetSource(PublishedDataSetName, new MetaDataOnlySource(metaData)) .UseAllStandardEncoders() - .AddTransportFactory(factory) - .Build(); + .AddTransportFactory(factory); + if (securityKeyProvider is not null) + { + builder.AddSecurityKeyProvider(securityKeyProvider); + } + return builder.Build(); } private static IPubSubApplication BuildSubscriberApp( string url, - IPubSubTransportFactory factory) + IPubSubTransportFactory factory, + IPubSubSecurityKeyProvider? securityKeyProvider = null) { - return new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) + var connection = new PubSubConnectionDataType + { + Name = "subscriber", + TransportProfileUri = Profiles.PubSubUdpUadpTransport, + Address = new ExtensionObject(new NetworkAddressUrlDataType + { + Url = url + }) + }; + if (securityKeyProvider is not null) + { + var readerGroup = new ReaderGroupDataType + { + Name = "secured-listener" + }; + SecureGroup(readerGroup); + connection.ReaderGroups = [readerGroup]; + } + PubSubApplicationBuilder builder = new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) .WithApplicationId("discovery-subscriber") .UseConfiguration(new PubSubConfigurationDataType { - Connections = - [ - new PubSubConnectionDataType - { - Name = "subscriber", - TransportProfileUri = Profiles.PubSubUdpUadpTransport, - Address = new ExtensionObject(new NetworkAddressUrlDataType - { - Url = url - }) - } - ], + Connections = [connection], PublishedDataSets = [] }) .UseAllStandardEncoders() - .AddTransportFactory(factory) - .Build(); + .AddTransportFactory(factory); + if (securityKeyProvider is not null) + { + builder.AddSecurityKeyProvider(securityKeyProvider); + } + return builder.Build(); + } + + private static void SecureGroup(PubSubGroupDataType group) + { + group.SecurityMode = MessageSecurityMode.SignAndEncrypt; + group.SecurityGroupId = SecurityGroupIdValue; + group.SecurityKeyServices = new ArrayOf(new[] + { + new EndpointDescription { EndpointUrl = "opc.tcp://localhost:4840/SecurityKeyService" } + }); + } + + private static PubSubSecurityKeyRing NewKeyRing() + { + var ring = new PubSubSecurityKeyRing(SecurityGroupIdValue); + ring.SetCurrent(TestSecurityKeyFactory.Create( + 1, + PubSubAes256CtrPolicy.Instance.SigningKeyLength, + PubSubAes256CtrPolicy.Instance.EncryptingKeyLength)); + return ring; } private static DataSetMetaDataType NewMetaData() @@ -437,6 +544,201 @@ private void Enqueue(ReadOnlyMemory payload) } } + /// + /// Delivers every frame that one transport sends to all other transports of + /// the bus, and records the frames that carry no SecurityHeader. + /// + private sealed class LoopbackBus + { + private readonly List m_transports = []; + private readonly List m_plaintextFrames = []; + private readonly Lock m_gate = new(); + private int m_announcementCount; + + public int AnnouncementCount => Volatile.Read(ref m_announcementCount); + + public IReadOnlyList PlaintextFrames + { + get + { + lock (m_gate) + { + return [.. m_plaintextFrames]; + } + } + } + + public void Attach(LoopbackTransport transport) + { + lock (m_gate) + { + m_transports.Add(transport); + } + } + + public void Deliver(LoopbackTransport sender, ReadOnlyMemory payload, bool isAnnouncement) + { + if (isAnnouncement) + { + Interlocked.Increment(ref m_announcementCount); + } + var receivers = new List(); + lock (m_gate) + { + if (!UadpDecoder.TryReadOuterPrefix(payload, out _, out bool securityEnabled, out _, out _) || + !securityEnabled) + { + byte[] head = payload.Slice(0, Math.Min(8, payload.Length)).ToArray(); + m_plaintextFrames.Add(BitConverter.ToString(head)); + } + foreach (LoopbackTransport transport in m_transports) + { + if (!ReferenceEquals(transport, sender)) + { + receivers.Add(transport); + } + } + } + foreach (LoopbackTransport receiver in receivers) + { + receiver.Enqueue(payload); + } + } + } + + private sealed class LoopbackTransportFactory : IPubSubTransportFactory + { + private readonly LoopbackBus m_bus; + + public LoopbackTransportFactory(LoopbackBus bus) + { + m_bus = bus; + } + + public string TransportProfileUri => Profiles.PubSubUdpUadpTransport; + + public IPubSubTransport Create( + PubSubConnectionDataType connection, + ITelemetryContext telemetry, + TimeProvider timeProvider) + { + _ = connection; + _ = telemetry; + _ = timeProvider; + var transport = new LoopbackTransport(m_bus); + m_bus.Attach(transport); + return transport; + } + } + + /// + /// An in-memory datagram transport that, like the UDP transport, sends + /// transport-specific discovery announcements to a separate destination. + /// + private sealed class LoopbackTransport : IPubSubTransport, IPubSubDiscoveryAnnouncementTransport + { + private readonly LoopbackBus m_bus; + private readonly Queue m_frames = new(); + private readonly SemaphoreSlim m_signal = new(0, int.MaxValue); + private readonly Lock m_gate = new(); + private bool m_disposed; + + public LoopbackTransport(LoopbackBus bus) + { + m_bus = bus; + } + + public string TransportProfileUri => Profiles.PubSubUdpUadpTransport; + + public PubSubTransportDirection Direction => PubSubTransportDirection.SendReceive; + + public bool IsConnected { get; private set; } + + public uint DiscoveryAnnounceRate => 0; + + public event EventHandler? StateChanged + { + add { } + remove { } + } + + public ValueTask OpenAsync(CancellationToken cancellationToken = default) + { + _ = cancellationToken; + IsConnected = true; + return default; + } + + public ValueTask CloseAsync(CancellationToken cancellationToken = default) + { + _ = cancellationToken; + IsConnected = false; + return default; + } + + public ValueTask SendAsync( + ReadOnlyMemory payload, + string? topic = null, + CancellationToken cancellationToken = default) + { + _ = topic; + cancellationToken.ThrowIfCancellationRequested(); + m_bus.Deliver(this, payload, isAnnouncement: false); + return default; + } + + public ValueTask SendDiscoveryAnnouncementAsync( + ReadOnlyMemory payload, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + m_bus.Deliver(this, payload, isAnnouncement: true); + return default; + } + + public async IAsyncEnumerable ReceiveAsync( + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + while (!cancellationToken.IsCancellationRequested) + { + await m_signal.WaitAsync(cancellationToken).ConfigureAwait(false); + PubSubTransportFrame frame; + lock (m_gate) + { + frame = m_frames.Dequeue(); + } + yield return frame; + } + } + + public ValueTask DisposeAsync() + { + lock (m_gate) + { + m_disposed = true; + IsConnected = false; + } + m_signal.Dispose(); + return default; + } + + public void Enqueue(ReadOnlyMemory payload) + { + lock (m_gate) + { + if (m_disposed) + { + return; + } + m_frames.Enqueue(new PubSubTransportFrame( + payload.ToArray(), + topic: null, + DateTimeUtc.From(DateTimeOffset.UtcNow))); + m_signal.Release(); + } + } + } + private sealed class MetaDataOnlySource : IPublishedDataSetSource { private readonly DataSetMetaDataType m_metaData; diff --git a/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs b/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs index bfbb905617..be651b8dc0 100644 --- a/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Encoding/Uadp/UadpDiscoveryTests.cs @@ -102,6 +102,21 @@ public void SecuredDiscoveryMessagesSplitWhereTheReceiverExpects(bool withDataSe } } + /// + /// The security boundary is defined for UADP data, action, and discovery + /// messages only; any other message cannot be secured. + /// + [Test] + public void EncodeWithSecurityBoundaryRejectsAMessageItCannotSecure() + { + PubSubNetworkMessageContext context = UadpTestUtilities.NewContext(); + var message = new Opc.Ua.PubSub.Encoding.Json.JsonNetworkMessage { MessageId = "json" }; + + Assert.That( + () => UadpEncoder.EncodeWithSecurityBoundary(message, context, out _), + Throws.ArgumentException); + } + [Test] public void DiscoveryRequest_DataSetMetaData_RoundTrips() { From 94d43919a98512fb2dd5019044792d62b9615b19 Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 9 Oct 2026 14:25:04 +0200 Subject: [PATCH 3/4] Count secured sends and describe Sign-only security accurately Review feedback: - Secured NetworkMessages are encoded at the security boundary, outside UadpEncoder.EncodeAsync, which is where SentNetworkMessages and SentDataSetMessages are counted. Secured data messages were therefore never counted, and the discovery messages this branch now secures stopped being counted. EncodeAndWrapUadpAsync counts them the same way the encoder counts unsecured messages. - MessageSecurityMode.Sign signs without encrypting, so the DataSet layout of a Sign-only announcement stays readable. docs/PubSub.md, the MetaDataPublisher remarks, and SendMetaDataAnnouncementAsync now describe applying the configured SecurityMode, and limit confidentiality to SignAndEncrypt. Test: MetaDataPublisherTests.SecuredNetworkMessagesAreCountedAsSentAsync (the counter read 0 before the fix). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/PubSub.md | 19 ++++++----- .../Application/MetaDataPublisher.cs | 9 +++--- .../Connections/PubSubConnection.cs | 20 +++++++++--- .../Application/MetaDataPublisherTests.cs | 32 +++++++++++++++++++ 4 files changed, 64 insertions(+), 16 deletions(-) diff --git a/docs/PubSub.md b/docs/PubSub.md index 739e3396ea..cfbe2cfe57 100644 --- a/docs/PubSub.md +++ b/docs/PubSub.md @@ -285,9 +285,11 @@ The publisher-side `MetaDataPublisher` ([§6.2.2.5](https://reference.opcfoundat emits a retained `JsonMetaDataMessage` / `UadpDiscoveryResponseMessage` on the well-known `ua-metadata` topic at startup and after each configuration version bump; subscribers cache it before the first -KeyFrame arrives. A connection configured for message security signs and -encrypts the UADP announcement with the same keys as its DataSetMessages, -so only subscribers that hold the keys can read the DataSet layout. +KeyFrame arrives. A connection configured for message security applies its +SecurityMode to the UADP announcement, with the same keys as its +DataSetMessages: `Sign` lets subscribers verify the announcement, and +`SignAndEncrypt` also hides the DataSet layout from applications that do not +hold the keys. A subscriber also registers the `DataSetMetaData` configured on each DataSetReader whose filter names an exact PublisherId, WriterGroupId, and @@ -1222,11 +1224,12 @@ subscribers that need to find publishers and bind to metadata at runtime. duplicate probes are suppressed, and identical responses are throttled. - MQTT publishes retained discovery messages on the standard status, connection, application, endpoint, and metadata topics. -- A UADP connection configured for message security signs and encrypts its - discovery requests, responses, and announcements with the keys of its - SecurityGroup, so only applications that hold those keys can discover each - other. The MQTT last-will status message is the exception: the broker - publishes it later, so it is sent unsecured. +- A UADP connection configured for message security applies its SecurityMode + to its discovery requests, responses, and announcements, with the keys of + its SecurityGroup: they are signed, and with `SignAndEncrypt` also + encrypted. A secured connection therefore discovers only applications that + hold the same keys. The MQTT last-will status message is the exception: the + broker publishes it later, so it is sent unsecured. ```csharp PubSubDiscoveryResult result = await application.RequestDiscoveryAsync( diff --git a/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs b/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs index 1669edcbaf..07c81830c8 100644 --- a/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs +++ b/src/Opc.Ua.PubSub/Application/MetaDataPublisher.cs @@ -76,9 +76,10 @@ namespace Opc.Ua.PubSub.Application /// /// /// Announcements go through the connection's send pipeline, so a - /// connection configured for message security signs and encrypts - /// them with the same keys as its DataSetMessages, and refuses to - /// send them when it cannot secure them. + /// connection configured for message security applies its + /// SecurityMode to them with the same keys as its DataSetMessages + /// (signing, and with SignAndEncrypt also encryption), and refuses + /// to send them when it cannot secure them. /// /// /// Lifetime is owned by : started @@ -532,7 +533,7 @@ private async ValueTask PublishMetaDataAsync( writer.DataSetWriterId); // The connection applies its message security, so a secured - // connection never announces the DataSet layout in the clear. + // connection never sends the announcement unsecured. await connection.SendMetaDataAnnouncementAsync(message, topic, cancellationToken) .ConfigureAwait(false); } diff --git a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs index 48a8dbd594..48fe716d89 100644 --- a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs +++ b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs @@ -2466,9 +2466,9 @@ private ValueTask SendNetworkMessageAsync( /// /// Sends a DataSetMetaData announcement through the connection's send - /// pipeline, so a connection configured for message security signs and - /// encrypts the announcement like its DataSetMessages and never sends it - /// in the clear. + /// pipeline, so a connection configured for message security secures the + /// announcement with its SecurityMode, like its DataSetMessages, and never + /// sends it unsecured. /// /// UADP discovery response or JSON metadata message. /// Broker topic, or for datagram transports. @@ -2747,13 +2747,14 @@ private async ValueTask> EncodeAndWrapUadpAsync( PubSubNetworkMessageContext context, CancellationToken cancellationToken) { + ReadOnlyMemory wrapped; try { ReadOnlyMemory encoded = UadpEncoder.EncodeWithSecurityBoundary( message, context, out int payloadOffset); ReadOnlyMemory prefix = encoded[..payloadOffset]; ReadOnlyMemory inner = encoded[payloadOffset..]; - return await m_securityWrapper! + wrapped = await m_securityWrapper! .WrapAsync(prefix, inner, m_securityWrapOptions, cancellationToken) .ConfigureAwait(false); } @@ -2770,6 +2771,17 @@ private async ValueTask> EncodeAndWrapUadpAsync( m_logger.UadpSecurityWrapFailed(ex); throw; } + + // The encoder counts the messages it encodes; the secured encoding bypasses + // it, so count the secured message in the same way. + context.Diagnostics.Increment(PubSubDiagnosticsCounterKind.SentNetworkMessages); + if (message is UadpNetworkMessage { DataSetMessages.Count: > 0 } data) + { + context.Diagnostics.Increment( + PubSubDiagnosticsCounterKind.SentDataSetMessages, + data.DataSetMessages.Count); + } + return wrapped; } private INetworkMessageEncoder? ResolveEncoder() diff --git a/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs b/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs index de049ec946..72abce97ca 100644 --- a/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Application/MetaDataPublisherTests.cs @@ -225,6 +225,38 @@ await WaitUntilAsync( Assert.That(response.DataSetMetaData?.Name, Is.EqualTo("pds-1")); } + /// + /// A secured NetworkMessage is encoded at the security boundary, outside the + /// encoder that counts the messages it encodes, and must still be counted as + /// sent. + /// + [Test] + public async Task SecuredNetworkMessagesAreCountedAsSentAsync() + { + using PubSubSecurityKeyRing ring = NewKeyRing(); + var factory = new RecordingTransportFactory(UadpProfile, supportsTopics: false); + await using IPubSubApplication app = BuildApp( + UadpProfile, + factory, + securityKeyProvider: new StaticSecurityKeyProvider(SecurityGroupIdValue, ring)); + + await app.StartAsync(CancellationToken.None).ConfigureAwait(false); + await WaitUntilAsync( + () => factory.Transport is { } t && t.Sends.Count >= 1, + TimeSpan.FromSeconds(2)).ConfigureAwait(false); + + // Each message is counted before it reaches the transport, so the counter + // read after the sends is at least the number of sends. + int sends; + lock (factory.Transport!.Sends) + { + sends = factory.Transport.Sends.Count; + } + Assert.That( + app.Diagnostics.Read(PubSubDiagnosticsCounterKind.SentNetworkMessages), + Is.GreaterThanOrEqualTo(sends)); + } + [Test] public async Task DisposeAsync_UnsubscribesFromRegistry() { From 58d07e0834e045b43f25c11abd9389ad9772cf4c Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 9 Oct 2026 14:55:51 +0200 Subject: [PATCH 4/4] Keep discovery send failures from failing the connection Securing the discovery announcements (previous commits) made them depend on a current security key. When the key provider has none yet, as before a Security Key Service delivers the first key, or when the key does not fit the policy (the reference sample's 12-byte KeyNonce, fixed in #4710), wrapping the startup announcements threw out of EnableAsync, and StartAsync failed. The aot-ubuntu-latest job failed this way in PubSubAotTests.StartsAndStopsPublisher_UdpUadp. A response to a discovery probe that could not be secured threw into the receive loop and ended it. SendDiscoveryResponseAsync now logs a discovery message that cannot be secured or sent (DiscoveryMessageNotSent, Warning) and drops it, as the UDP transport already does for network errors of discovery announcements. Requests sent through RequestDiscoveryAsync still throw to the caller. Tests (PubSubDiscoveryTests), both red before the fix: - StartupAnnouncementsWithoutACurrentKeyDoNotFailTheStartAsync - ReceiveLoopSurvivesADiscoveryResponseThatCannotBeSecuredAsync Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Connections/PubSubConnection.cs | 26 ++++ .../Application/PubSubDiscoveryTests.cs | 119 ++++++++++++++++++ 2 files changed, 145 insertions(+) diff --git a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs index 48fe716d89..e96b53aec4 100644 --- a/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs +++ b/src/Opc.Ua.PubSub/Connections/PubSubConnection.cs @@ -1866,6 +1866,24 @@ private async ValueTask SendDiscoveryResponseAsync( { return; } + try + { + await SendDiscoveryMessageAsync(response, topic, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // A discovery message that cannot be secured or sent, for example while + // no security key is available, must neither fail the start of the + // connection nor end its receive loop. + m_logger.DiscoveryMessageNotSent(ex, response.DiscoveryType, Name); + } + } + + private async ValueTask SendDiscoveryMessageAsync( + UadpDiscoveryResponseMessage response, + string? topic, + CancellationToken cancellationToken) + { PubSubNetworkMessage networkMessage = ConvertDiscoveryMessageForTransport(response); INetworkMessageEncoder? encoder = ResolveEncoder(); if (ShouldUseDiscoveryAnnouncementDestination( @@ -3387,6 +3405,14 @@ public static partial void DroppingOutboundMessage( [LoggerMessage(EventId = PubSubEventIds.PubSubConnection + 20, Level = LogLevel.Error, Message = "UADP unwrap threw on inbound frame.")] public static partial void UadpUnwrapThrewOnInboundFrame(this ILogger logger, Exception exception); + + [LoggerMessage(EventId = PubSubEventIds.PubSubConnection + 22, Level = LogLevel.Warning, + Message = "Could not send the {DiscoveryType} discovery message on connection '{Connection}'.")] + public static partial void DiscoveryMessageNotSent( + this ILogger logger, + Exception exception, + UadpDiscoveryType discoveryType, + string connection); } } diff --git a/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs b/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs index 201fd35a6d..a446f143fb 100644 --- a/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs +++ b/tests/Opc.Ua.PubSub.Tests/Application/PubSubDiscoveryTests.cs @@ -245,6 +245,79 @@ public async Task SecuredSubscriberDiscoversSecuredPublisherWithoutPlaintextFram "Every frame of a secured connection must carry a SecurityHeader."); } + /// + /// Until the key provider has a current key, as before a Security Key Service + /// delivers the first key, the startup discovery announcements cannot be + /// secured. They are dropped, and the application still starts. + /// + [Test] + public async Task StartupAnnouncementsWithoutACurrentKeyDoNotFailTheStartAsync() + { + using PubSubSecurityKeyRing ring = NewKeyRing(); + var keys = new InterruptibleKeyProvider(new StaticSecurityKeyProvider(SecurityGroupIdValue, ring)) + { + IsCurrentKeyAvailable = false + }; + var bus = new LoopbackBus(); + await using IPubSubApplication publisher = BuildPublisherApp( + "opc.udp://239.0.0.1:4840", + new LoopbackTransportFactory(bus), + keys); + + await publisher.StartAsync(CancellationToken.None).ConfigureAwait(false); + + Assert.That(publisher.State.State, Is.EqualTo(PubSubState.Operational)); + Assert.That(bus.PlaintextFrames, Is.Empty); + } + + /// + /// A discovery response that cannot be secured is dropped without ending the + /// receive loop of the publisher, which answers the next request. + /// + [Test] + public async Task ReceiveLoopSurvivesADiscoveryResponseThatCannotBeSecuredAsync() + { + const string url = "opc.udp://239.0.0.1:4840"; + using PubSubSecurityKeyRing publisherRing = NewKeyRing(); + using PubSubSecurityKeyRing subscriberRing = NewKeyRing(); + var publisherKeys = new InterruptibleKeyProvider( + new StaticSecurityKeyProvider(SecurityGroupIdValue, publisherRing)); + var bus = new LoopbackBus(); + await using IPubSubApplication publisher = BuildPublisherApp( + url, + new LoopbackTransportFactory(bus), + publisherKeys); + await using IPubSubApplication subscriber = BuildSubscriberApp( + url, + new LoopbackTransportFactory(bus), + new StaticSecurityKeyProvider(SecurityGroupIdValue, subscriberRing)); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await publisher.StartAsync(cts.Token).ConfigureAwait(false); + await subscriber.StartAsync(cts.Token).ConfigureAwait(false); + + publisherKeys.IsCurrentKeyAvailable = false; + PubSubDiscoveryResult unanswered = await subscriber.RequestDiscoveryAsync( + new PubSubDiscoveryRequest + { + DiscoveryType = UadpDiscoveryType.DataSetMetaData, + DataSetWriterIds = [DataSetWriterIdValue] + }, + TimeSpan.FromMilliseconds(300), + cts.Token).ConfigureAwait(false); + publisherKeys.IsCurrentKeyAvailable = true; + PubSubDiscoveryResult answered = await subscriber.RequestDiscoveryAsync( + new PubSubDiscoveryRequest + { + DiscoveryType = UadpDiscoveryType.DataSetWriterConfiguration, + DataSetWriterIds = [DataSetWriterIdValue] + }, + TimeSpan.FromMilliseconds(300), + cts.Token).ConfigureAwait(false); + + Assert.That(unanswered.DataSetMetaDataEntries, Is.Empty); + Assert.That(answered.WriterConfigurations, Has.Count.EqualTo(1)); + } + private static IPubSubApplication BuildDiscoveryOnlyApp(IPubSubTransportFactory factory) { return new PubSubApplicationBuilder(NUnitTelemetryContext.Create()) @@ -739,6 +812,52 @@ public void Enqueue(ReadOnlyMemory payload) } } + /// + /// A key provider whose current key can be made unavailable, as before a + /// Security Key Service delivers the first or the next key. Keys that were + /// already issued can still be looked up by their token. + /// + private sealed class InterruptibleKeyProvider : IPubSubSecurityKeyProvider + { + private readonly IPubSubSecurityKeyProvider m_inner; + private int m_isCurrentKeyAvailable = 1; + + public InterruptibleKeyProvider(IPubSubSecurityKeyProvider inner) + { + m_inner = inner; + } + + public bool IsCurrentKeyAvailable + { + get => Volatile.Read(ref m_isCurrentKeyAvailable) != 0; + set => Volatile.Write(ref m_isCurrentKeyAvailable, value ? 1 : 0); + } + + public string SecurityGroupId => m_inner.SecurityGroupId; + + public event EventHandler? KeyRotated + { + add => m_inner.KeyRotated += value; + remove => m_inner.KeyRotated -= value; + } + + public ValueTask GetCurrentKeyAsync(CancellationToken cancellationToken = default) + { + if (!IsCurrentKeyAvailable) + { + throw new InvalidOperationException("No current security key is available."); + } + return m_inner.GetCurrentKeyAsync(cancellationToken); + } + + public ValueTask TryGetKeyAsync( + uint tokenId, + CancellationToken cancellationToken = default) + { + return m_inner.TryGetKeyAsync(tokenId, cancellationToken); + } + } + private sealed class MetaDataOnlySource : IPublishedDataSetSource { private readonly DataSetMetaDataType m_metaData;