Merge branch 'develop' into dependabot/nuget/dependencies-0e8a0d9cc7

This commit is contained in:
Rob Hague
2026-10-19 21:12:09 +02:00
12 changed files with 323 additions and 161 deletions
@@ -3,7 +3,6 @@ using System.Globalization;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Renci.SshNet.Common;
using Renci.SshNet.Messages.Transport;
@@ -26,11 +25,6 @@ namespace Renci.SshNet.Abstractions
ConnectCore(socket, remoteEndpoint, connectTimeout, ownsSocket: false);
}
public static async Task ConnectAsync(Socket socket, EndPoint remoteEndpoint, CancellationToken cancellationToken)
{
await socket.ConnectAsync(remoteEndpoint, cancellationToken).ConfigureAwait(false);
}
private static void ConnectCore(Socket socket, EndPoint remoteEndpoint, TimeSpan connectTimeout, bool ownsSocket)
{
var connectCompleted = new ManualResetEvent(initialState: false);
@@ -254,13 +248,6 @@ namespace Renci.SshNet.Abstractions
return totalBytesRead;
}
#if !NET
public static Task<int> ReadAsync(Socket socket, byte[] buffer, CancellationToken cancellationToken)
{
return socket.ReceiveAsync(buffer, 0, buffer.Length, cancellationToken);
}
#endif
public static void Send(Socket socket, byte[] data)
{
Send(socket, data, 0, data.Length);
@@ -2,118 +2,69 @@
using System;
using System.Net;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace Renci.SshNet.Abstractions
{
// Async helpers based on https://devblogs.microsoft.com/pfxteam/awaiting-socket-operations/
internal static class SocketExtensions
{
private sealed class AwaitableSocketAsyncEventArgs : SocketAsyncEventArgs, INotifyCompletion
public static Task ConnectAsync(this Socket socket, EndPoint remoteEndpoint, CancellationToken cancellationToken)
{
private static readonly Action SENTINEL = () => { };
private bool _isCancelled;
private Action _continuationAction;
public AwaitableSocketAsyncEventArgs()
if (cancellationToken.IsCancellationRequested)
{
Completed += (sender, e) => SetCompleted();
return Task.FromCanceled(cancellationToken);
}
public AwaitableSocketAsyncEventArgs ExecuteAsync(Func<SocketAsyncEventArgs, bool> func)
var connectTask = Task.Factory.FromAsync(
static (s, remoteEndpoint, callback, state) => s.BeginConnect(remoteEndpoint, callback, state),
static result => ((Socket)result.AsyncState).EndConnect(result),
socket,
remoteEndpoint,
state: socket);
return connectTask.IsCompleted || !cancellationToken.CanBeCanceled
? connectTask
: WaitWithCancellation(connectTask, socket, cancellationToken);
static async Task WaitWithCancellation(Task connectTask, Socket socket, CancellationToken cancellationToken)
{
if (!func(this))
using (cancellationToken.Register(static s => ((Socket)s).Dispose(), socket, useSynchronizationContext: false))
{
SetCompleted();
}
return this;
}
private void SetCompleted()
{
IsCompleted = true;
var continuation = Interlocked.Exchange(ref _continuationAction, SENTINEL);
if (continuation is not null)
{
continuation();
}
}
public void SetCancelled()
{
_isCancelled = true;
SetCompleted();
}
public AwaitableSocketAsyncEventArgs GetAwaiter()
{
return this;
}
public bool IsCompleted { get; private set; }
void INotifyCompletion.OnCompleted(Action continuation)
{
if (_continuationAction == SENTINEL || Interlocked.CompareExchange(ref _continuationAction, continuation, comparand: null) == SENTINEL)
{
// We have already completed; run continuation asynchronously
_ = Task.Run(continuation);
}
}
public void GetResult()
{
if (_isCancelled)
{
throw new TaskCanceledException();
}
if (!IsCompleted)
{
// We don't support sync/async
throw new InvalidOperationException("The asynchronous operation has not yet completed.");
}
if (SocketError != SocketError.Success)
{
throw new SocketException((int)SocketError);
await connectTask.ConfigureAwait(false);
}
}
}
public static async Task ConnectAsync(this Socket socket, EndPoint remoteEndpoint, CancellationToken cancellationToken)
public static Task<int> ReceiveAsync(this Socket socket, byte[] buffer, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
using (var args = new AwaitableSocketAsyncEventArgs())
{
args.RemoteEndPoint = remoteEndpoint;
using (cancellationToken.Register(o => ((AwaitableSocketAsyncEventArgs)o).SetCancelled(), args, useSynchronizationContext: false))
{
await args.ExecuteAsync(socket.ConnectAsync);
}
}
return ReceiveAsync(socket, buffer, 0, buffer.Length, cancellationToken);
}
public static async Task<int> ReceiveAsync(this Socket socket, byte[] buffer, int offset, int length, CancellationToken cancellationToken)
public static Task<int> ReceiveAsync(this Socket socket, byte[] buffer, int offset, int length, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
using (var args = new AwaitableSocketAsyncEventArgs())
if (cancellationToken.IsCancellationRequested)
{
args.SetBuffer(buffer, offset, length);
return Task.FromCanceled<int>(cancellationToken);
}
using (cancellationToken.Register(o => ((AwaitableSocketAsyncEventArgs)o).SetCancelled(), args, useSynchronizationContext: false))
var receiveTask = Task.Factory.FromAsync(
static (s, buffer, callback, state) => s.BeginReceive(buffer.Array, buffer.Offset, buffer.Count, SocketFlags.None, callback, state),
static result => ((Socket)result.AsyncState).EndReceive(result),
socket,
new ArraySegment<byte>(buffer, offset, length),
state: socket);
return receiveTask.IsCompleted || !cancellationToken.CanBeCanceled
? receiveTask
: WaitWithCancellation(receiveTask, socket, cancellationToken);
static async Task<int> WaitWithCancellation(Task<int> receiveTask, Socket socket, CancellationToken cancellationToken)
{
using (cancellationToken.Register(static s => ((Socket)s).Dispose(), socket, useSynchronizationContext: false))
{
await args.ExecuteAsync(socket.ReceiveAsync);
return await receiveTask.ConfigureAwait(false);
}
return args.BytesTransferred;
}
}
}
+86 -15
View File
@@ -65,13 +65,7 @@ namespace Renci.SshNet.Common
_ = Monitor.Wait(_sync);
}
var bytesRead = Math.Min(buffer.Length, _buffer.ActiveLength);
_buffer.ActiveReadOnlySpan.Slice(0, bytesRead).CopyTo(buffer);
_buffer.Discard(bytesRead);
return bytesRead;
return ReadCore(buffer);
}
}
@@ -85,6 +79,20 @@ namespace Renci.SshNet.Common
}
#endif
private int ReadCore(Span<byte> buffer)
{
Debug.Assert(Monitor.IsEntered(_sync));
Debug.Assert(_buffer.ActiveLength > 0 || _disposed);
var bytesRead = Math.Min(buffer.Length, _buffer.ActiveLength);
_buffer.ActiveReadOnlySpan.Slice(0, bytesRead).CopyTo(buffer);
_buffer.Discard(bytesRead);
return bytesRead;
}
/// <inheritdoc/>
public override void Write(byte[] buffer, int offset, int count)
{
@@ -134,17 +142,80 @@ namespace Renci.SshNet.Common
Monitor.PulseAll(_sync);
}
// We provide overrides for async Write methods but not async Read.
// The default implementations from the base class effectively call the
// sync methods on a threadpool thread, but only allowing one async
// operation at a time (for protecting thread-unsafe implementations).
// This constraint is desirable for reads because if there were multiple
// readers and no data coming in, our current Monitor.Wait implementation
// would just block as many threadpool threads as there are readers.
// But since a write is just short-lived buffer copying and can unblock
// We provide overrides for async Write methods, but the async Read overrides
// have only partial implementations (for the happy path when data is available
// to be read), and otherwise defer to the base implementation.
// The default implementations from the base class effectively call the sync
// methods on a threadpool thread, but only allowing one such async-over-sync
// operation at a time.
// This constraint is desirable for blocking reads because if there were
// multiple readers and no data coming in, our current Monitor.Wait
// implementation would just block as many threadpool threads as there
// are readers. But when data is already available, a read is just
// short-lived buffer copying, so we can complete synchronously and avoid
// the overhead of the base.
// Likewise, since a write is just short-lived buffer copying and can unblock
// readers, it is beneficial to circumvent the one-at-a-time constraint,
// as otherwise a waiting async read will block the async write that could
// unblock it.
// Note that we do not override BeginRead/EndRead because the base ReadAsync
// takes a faster path in this way (and it can introduce circular dependency).
/// <inheritdoc/>
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
#if !NET
ThrowHelper.
#endif
ValidateBufferArguments(buffer, offset, count);
if (!cancellationToken.IsCancellationRequested &&
TryReadWithoutBlocking(buffer.AsSpan(offset, count), out var bytesRead))
{
return Task.FromResult(bytesRead);
}
return base.ReadAsync(buffer, offset, count, cancellationToken);
}
#if NET
/// <inheritdoc/>
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
if (!cancellationToken.IsCancellationRequested &&
TryReadWithoutBlocking(buffer.Span, out var bytesRead))
{
return new ValueTask<int>(bytesRead);
}
return base.ReadAsync(buffer, cancellationToken);
}
#endif
private bool TryReadWithoutBlocking(Span<byte> buffer, out int bytesRead)
{
if (!Monitor.TryEnter(_sync))
{
bytesRead = 0;
return false;
}
try
{
if (_buffer.ActiveLength == 0 && !_disposed)
{
bytesRead = 0;
return false;
}
bytesRead = ReadCore(buffer);
return true;
}
finally
{
Monitor.Exit(_sync);
}
}
/// <inheritdoc/>
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+3 -6
View File
@@ -371,12 +371,9 @@ namespace Renci.SshNet.Common
/// <exception cref="ArgumentOutOfRangeException"><paramref name="length"/> is greater than the internal buffer size.</exception>
internal byte[] ReadBytes(int length)
{
var data = new byte[length];
var bytesRead = Read(data, 0, length);
if (bytesRead < length)
{
throw new ArgumentOutOfRangeException(nameof(length), string.Format(CultureInfo.InvariantCulture, "The requested length ({0}) is greater than the actual number of bytes read ({1}).", length, bytesRead));
}
var data = GetRemainingBuffer().Slice(0, length).ToArray();
Position += length;
return data;
}
+1 -25
View File
@@ -72,7 +72,7 @@ namespace Renci.SshNet.Connection
var socket = SocketFactory.Create(SocketType.Stream, ProtocolType.Tcp);
try
{
await SocketAbstraction.ConnectAsync(socket, endPoint, cancellationToken).ConfigureAwait(false);
await socket.ConnectAsync(endPoint, cancellationToken).ConfigureAwait(false);
return socket;
}
@@ -83,13 +83,6 @@ namespace Renci.SshNet.Connection
}
}
protected static byte SocketReadByte(Socket socket)
{
var buffer = new byte[1];
_ = SocketRead(socket, buffer, 0, 1, Timeout.InfiniteTimeSpan);
return buffer[0];
}
protected static byte SocketReadByte(Socket socket, TimeSpan readTimeout)
{
var buffer = new byte[1];
@@ -97,23 +90,6 @@ namespace Renci.SshNet.Connection
return buffer[0];
}
/// <summary>
/// Performs a blocking read on the socket until <paramref name="length"/> bytes are received.
/// </summary>
/// <param name="socket">The <see cref="Socket"/> to read from.</param>
/// <param name="buffer">An array of type <see cref="byte"/> that is the storage location for the received data.</param>
/// <param name="offset">The position in <paramref name="buffer"/> parameter to store the received data.</param>
/// <param name="length">The number of bytes to read.</param>
/// <returns>
/// The number of bytes read.
/// </returns>
/// <exception cref="SshConnectionException">The socket is closed.</exception>
/// <exception cref="SocketException">The read failed.</exception>
protected static int SocketRead(Socket socket, byte[] buffer, int offset, int length)
{
return SocketRead(socket, buffer, offset, length, Timeout.InfiniteTimeSpan);
}
/// <summary>
/// Performs a blocking read on the socket until <paramref name="length"/> bytes are received.
/// </summary>
@@ -181,7 +181,7 @@ namespace Renci.SshNet.Connection
while (buffer.Count < MaximumBannerLineLength)
{
var bytesRead = await SocketAbstraction.ReadAsync(socket, data, cancellationToken).ConfigureAwait(false);
var bytesRead = await socket.ReceiveAsync(data, cancellationToken).ConfigureAwait(false);
if (bytesRead == 0)
{
throw CreateConnectionLostException();
@@ -1,7 +1,10 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Text;
using Renci.SshNet.Common;
namespace Renci.SshNet.Sftp.Responses
{
internal sealed class SftpNameResponse : SftpResponse
@@ -29,6 +32,13 @@ namespace Renci.SshNet.Sftp.Responses
base.LoadData();
Count = ReadUInt32();
// Each entry is at least two uints in size.
if (Count > (DataStream.Length - DataStream.Position) / 8)
{
throw new SshException(string.Format(CultureInfo.InvariantCulture, "Invalid SSH_FXP_NAME count: {0}.", Count));
}
Files = new KeyValuePair<string, SftpFileAttributes>[Count];
for (var i = 0; i < Count; i++)
+10 -2
View File
@@ -3,6 +3,7 @@ using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.Linq;
using System.Text;
@@ -820,8 +821,15 @@ namespace Renci.SshNet.Sftp
if ((flag & SSH_FILEXFER_ATTR_EXTENDED) == SSH_FILEXFER_ATTR_EXTENDED)
{
var extendedCount = (int)stream.ReadUInt32();
extensions = new Dictionary<string, string>(extendedCount);
var extendedCount = stream.ReadUInt32();
// Each extension is at least two string lengths.
if (extendedCount > (stream.Length - stream.Position) / 8)
{
throw new SshException(string.Format(CultureInfo.InvariantCulture, "Invalid extended attribute count: {0}.", extendedCount));
}
extensions = new Dictionary<string, string>((int)extendedCount);
for (var i = 0; i < extendedCount; i++)
{
var extensionName = stream.ReadString(Encoding.UTF8);
@@ -1,5 +1,6 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -157,6 +158,115 @@ namespace Renci.SshNet.Tests.Classes.Common
}
#endif
[TestMethod]
public async Task ReadAsync_DataAvailable_CompletesSynchronously()
{
var pipeStream = new PipeStream();
pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4);
var readBuffer = new byte[3];
Task<int> readTask = pipeStream.ReadAsync(readBuffer, 0, 3);
Assert.IsTrue(readTask.IsCompleted);
Assert.AreEqual(3, await readTask);
CollectionAssert.AreEqual(new byte[] { 1, 2, 3 }, readBuffer);
readTask = pipeStream.ReadAsync(readBuffer, 0, 3);
Assert.IsTrue(readTask.IsCompleted);
Assert.AreEqual(1, await readTask);
Assert.AreEqual(4, readBuffer[0]);
// No data left, so the next read should block.
readTask = pipeStream.ReadAsync(readBuffer, 0, 3);
await Task.Delay(50);
Assert.IsFalse(readTask.IsCompleted);
await pipeStream.WriteAsync(new byte[] { 5 }, 0, 1);
Assert.AreEqual(1, await readTask);
Assert.AreEqual(5, readBuffer[0]);
}
[TestMethod]
public async Task ReadAsync_AfterDispose_CompletesSynchronously()
{
var pipeStream = new PipeStream();
pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4);
pipeStream.Dispose();
Task<int> readTask = pipeStream.ReadAsync(new byte[5], 0, 5);
Assert.IsTrue(readTask.IsCompleted);
Assert.AreEqual(4, await readTask);
readTask = pipeStream.ReadAsync(new byte[5], 0, 5);
Assert.IsTrue(readTask.IsCompleted);
Assert.AreEqual(0, await readTask);
}
[TestMethod]
public async Task ReadAsync_DataAvailable_CancelledToken_TaskIsCancelled()
{
var pipeStream = new PipeStream();
pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4);
var token = new CancellationToken(canceled: true);
Task<int> readTask = pipeStream.ReadAsync(new byte[4], 0, 4, token);
Assert.IsTrue(readTask.IsCanceled);
var ex = await Assert.ThrowsAsync<OperationCanceledException>(() => readTask);
Assert.AreEqual(token, ex.CancellationToken);
Assert.AreEqual(4, pipeStream.Length);
}
#if NET
[TestMethod]
public async Task ReadAsync_Memory_DataAvailable_CompletesSynchronously()
{
var pipeStream = new PipeStream();
pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4);
var readBuffer = new byte[3];
ValueTask<int> readTask = pipeStream.ReadAsync(readBuffer.AsMemory());
Assert.IsTrue(readTask.IsCompletedSuccessfully);
Assert.AreEqual(3, await readTask);
CollectionAssert.AreEqual(new byte[] { 1, 2, 3 }, readBuffer);
ValueTask<int> readTask2 = pipeStream.ReadAsync(readBuffer.AsMemory());
Assert.IsTrue(readTask2.IsCompletedSuccessfully);
Assert.AreEqual(1, await readTask2);
Assert.AreEqual(4, readBuffer[0]);
// No data left, so the next read should block.
ValueTask<int> readTask3 = pipeStream.ReadAsync(readBuffer.AsMemory());
await Task.Delay(50);
Assert.IsFalse(readTask3.IsCompleted);
await pipeStream.WriteAsync(new byte[] { 5 });
Assert.AreEqual(1, await readTask3);
Assert.AreEqual(5, readBuffer[0]);
}
#endif
[TestMethod]
public void Read_AfterDispose_StillWorks()
{
@@ -102,23 +102,15 @@ namespace Renci.SshNet.Tests.Classes.Common
}
[TestMethod]
public void ReadBytes_Length_LengthIsGreaterThanNumberOfBytesAvailable()
[DataRow(4)]
[DataRow(int.MaxValue)]
public void ReadBytes_Length_LengthIsGreaterThanNumberOfBytesAvailable(int length)
{
MySshData sshData = new MySshData();
sshData.Load(new byte[] { 0x01, 0x0d, 0x0a }, 0, 3);
try
{
sshData.ReadBytes(4);
Assert.Fail();
}
catch (ArgumentOutOfRangeException ex)
{
Assert.AreEqual(typeof(ArgumentOutOfRangeException), ex.GetType());
Assert.IsNull(ex.InnerException);
Assert.AreEqual("length", ex.ParamName);
}
Assert.Throws<ArgumentException>(() => sshData.ReadBytes(length));
}
[TestMethod]
@@ -0,0 +1,47 @@
using System.Text;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Renci.SshNet.Common;
using Renci.SshNet.Sftp.Responses;
namespace Renci.SshNet.Tests.Classes.Sftp.Responses
{
[TestClass]
public class SftpNameResponseTest
{
[TestMethod]
public void Load()
{
var target = new SftpNameResponse(3, Encoding.UTF8);
target.Load([
0, 0, 0, 7, // response id
0, 0, 0, 1, // count
0, 0, 0, 1, (byte)'a', // filename
0, 0, 0, 0, // longname
0, 0, 0, 0, // attribute flags
]);
Assert.AreEqual(7u, target.ResponseId);
Assert.HasCount(1, target.Files);
Assert.AreEqual("a", target.Files[0].Key);
}
[TestMethod]
public void Load_CountExceedsData_Throws()
{
var target = new SftpNameResponse(3, Encoding.UTF8);
var ex = Assert.ThrowsExactly<SshException>(() => target.Load([
0, 0, 0, 7, // response id
0xFF, 0xFF, 0xFF, 0xFF, // count
0, 0, 0, 1, (byte)'a', // filename
0, 0, 0, 0, // longname
0, 0, 0, 0, // attribute flags
]));
Assert.AreEqual("Invalid SSH_FXP_NAME count: 4294967295.", ex.Message);
}
}
}
@@ -44,6 +44,19 @@ namespace Renci.SshNet.Tests.Classes.Sftp
Assert.AreEqual(isNamedPipe, attributes.IsNamedPipe);
}
[TestMethod]
public void FromBytes_ExtendedCountExceedsData_Throws()
{
var ex = Assert.ThrowsExactly<SshException>(() => SftpFileAttributes.FromBytes([
0x80, 0, 0, 0, // flags: extended
0x7F, 0xFF, 0xFF, 0xFF, // extended count
0, 0, 0, 1, (byte)'a', // extension name
0, 0, 0, 1, (byte)'b', // extension data
]));
Assert.AreEqual("Invalid extended attribute count: 2147483647.", ex.Message);
}
[TestMethod]
public void FromBytesGetBytes()
{