Guard StreamEngine handshake against null peer identity when RecvIdentity is enabled

src/NetMQ.Tests/MechanismTests.cs

@@ -1,5 +1,9 @@
-using System.Text;
+using System.Reflection;
+using System.Runtime.Serialization;
+using System.Text;
+using NetMQ.Core;
 using NetMQ.Core.Mechanisms;
+using NetMQ.Core.Transports;
 using Xunit;
 
 namespace NetMQ.Tests
@@ -43,6 +47,51 @@ public void IsCommandShouldReturnFalseForInvalidCommand()
             Assert.False(mechanism.IsCommand("READY", ref msg));
         }
 
+        [Fact]
+        public void MechanismReadyShouldHandleNullPeerIdentityWhenRecvIdentityIsEnabled()
+        {
+#pragma warning disable SYSLIB0050
+            var streamEngine = (StreamEngine)FormatterServices.GetUninitializedObject(typeof(StreamEngine));
+#pragma warning restore SYSLIB0050
+            var options = new Options { RecvIdentity = true, HeartbeatInterval = 0 };
+            var session = new RecordingSession();
+            var mechanism = new NullMechanism(session, options) { PeerIdentity = null };
+
+            SetPrivateField(streamEngine, "m_options", options);
+            SetPrivateField(streamEngine, "m_session", session);
+            SetPrivateField(streamEngine, "m_mechanism", mechanism);
+
+            var method = typeof(StreamEngine).GetMethod("MechanismReady", BindingFlags.Instance | BindingFlags.NonPublic);
+            Assert.NotNull(method);
+
+            var exception = Record.Exception(() => method!.Invoke(streamEngine, null));
+            Assert.Null(exception);
+            Assert.Equal(0, session.LastPushedMessageSize);
+        }
+
+        private static void SetPrivateField(object instance, string fieldName, object value)
+        {
+            var field = instance.GetType().GetField(fieldName, BindingFlags.Instance | BindingFlags.NonPublic);
+            Assert.NotNull(field);
+            field!.SetValue(instance, value);
+        }
+
+        private sealed class RecordingSession : SessionBase
+        {
+            public int LastPushedMessageSize { get; private set; } = -1;
+
+            public RecordingSession()
+                : base(new IOThread(new Ctx(), 0), false, null!, new Options(), null!)
+            {
+            }
+
+            public override PushMsgResult PushMsg(ref Msg msg)
+            {
+                LastPushedMessageSize = msg.Size;
+                return PushMsgResult.Ok;
+            }
+        }
+
         // this test was used to validate the behavior prior to changing the validation logic in Mechanism.IsCommand
         // [Fact]
         // public void IsCommandShouldThrowWhenLengthByteExceedsSize()

src/NetMQ.Tests/ThreadSafeSocketPollerCleanupTests.cs

@@ -25,12 +25,18 @@ public class ThreadSafeSocketPollerCleanupTests : IClassFixture<CleanupAfterFixt
 
         /// <summary>
         /// Repeatedly creates a <see cref="ServerSocket"/>, connects several
-        /// <see cref="ClientSocket"/>s to it, sends messages concurrently from multiple
-        /// threads (to keep the user thread processing commands), then disposes the server
-        /// while new clients are still connecting.
+        /// <see cref="ClientSocket"/>s to it, reads from it with a small positive timeout
+        /// (so the reader thread holds <c>m_threadSafeSync</c> while blocked in
+        /// <c>ProcessCommands</c>), then disposes the server while new clients are still
+        /// connecting.
         ///
-        /// Without the fix this reliably throws ArgumentOutOfRangeException inside
-        /// SocketBase.ProcessTerm when run enough iterations.
+        /// Without the fix this reliably throws <see cref="ArgumentOutOfRangeException"/>
+        /// inside <c>SocketBase.ProcessTerm</c> when run enough iterations.
+        ///
+        /// The reader task also catches <see cref="TerminatingException"/> because
+        /// <c>ProcessCommands</c> calls <c>CheckContextTerminated()</c> after draining the
+        /// mailbox, and that check can fire once the reaper has set <c>m_isStopped</c> on
+        /// the socket — a normal part of socket shutdown that the reader must tolerate.
         /// </summary>
         [Fact]
         public async Task ClosingServerWhileClientsConnectDoesNotCrash()
@@ -62,6 +68,7 @@ public async Task ClosingServerWhileClientsConnectDoesNotCrash()
                     }
                     catch (ObjectDisposedException) { }
                     catch (OperationCanceledException) { }
+                    catch (TerminatingException) { }
                 });
 
                 // Connect several clients.  Each successful TCP handshake will enqueue a

src/NetMQ/Core/Transports/StreamEngine.cs

@@ -1189,6 +1189,7 @@ PushMsgResult ProcessHandshakeCommand (ref Msg msg)
             return result;
         }
 
+#nullable enable
         void MechanismReady ()
         {
             if (m_options.HeartbeatInterval > 0)
@@ -1199,8 +1200,14 @@ void MechanismReady ()
 
             if (m_options.RecvIdentity) {
                 Msg identity = new Msg();
-                identity.InitPool(m_mechanism.PeerIdentity.Length);
-                identity.Put(m_mechanism.PeerIdentity, 0, m_mechanism.PeerIdentity.Length);
+                byte[]? peerIdentity = m_mechanism.PeerIdentity;
+                if (peerIdentity is null)
+                    identity.InitEmpty();
+                else
+                {
+                    identity.InitPool(peerIdentity.Length);
+                    identity.Put(peerIdentity, 0, peerIdentity.Length);
+                }
                 var pushResult = m_session.PushMsg(ref identity);
                 if (pushResult == PushMsgResult.Full) {
                     // If the write is failing at this stage with
@@ -1215,6 +1222,7 @@ void MechanismReady ()
             m_nextMsg = PullAndEncode;
             m_processMsg = DecodeAndPush;
         }
+#nullable disable
 
         PullMsgResult PullAndEncode (ref Msg msg)
         {
@@ -1384,4 +1392,4 @@ public void TimerEvent(int id)
 
         }
     }
-}
+}