Terminal: deliver large pastes whole, keep keystrokes in order, and keep one session's failure off the shared socket

- A message larger than the 64 KiB receive buffer was skipped and its tail dispatched as a frame of its
  own. Messages are now reassembled up to 16 MiB; one beyond that is dropped whole.
- Input frames are handed to the pump without waiting, and ShellStream takes no lock around its write
  buffer, so a snippet's text and its carriage return could interleave or lose bytes. Writes are
  serialised per pump in arrival order.
- A resize reaching a shell SSH.NET had already disposed threw out of the receive loop and closed the
  socket every tab shares. Each frame's failure now stays with its session.
This commit is contained in:
jaap-jan committed 2026-10-02 13:16:48 +02:00
1 parent 6846f67c20
commit 2ff9dff560
4 files changed
+205 -11

No files matched your search

@@ -56,6 +56,16 @@ public sealed class TerminalDataPlane : ITerminalTransport, IAsyncDisposable
private const int MaximumRequestBytes = 16 * 1024; private const int MaximumRequestBytes = 16 * 1024;
private const int ReceiveBufferBytes = 64 * 1024; private const int ReceiveBufferBytes = 64 * 1024;
/// <summary>The largest message from the page this host will reassemble.</summary>
/// <remarks>
/// A paste is one input frame however long it is, and so is a copied selection, so the receive buffer
/// is the size of a read and not the size of a message. Sixteen megabytes is far beyond anything a
/// person pastes into a shell on purpose and still small enough that a page sending garbage cannot make
/// this process hold an unbounded amount of it. A message beyond it is dropped whole — never its tail,
/// which would be parsed as a frame whose first bytes are somebody's paste.
/// </remarks>
private const int MaximumMessageBytes = 16 * 1024 * 1024;
private readonly TcpListener listener; private readonly TcpListener listener;
private readonly ITerminalAssetProvider assets; private readonly ITerminalAssetProvider assets;
private readonly Dictionary<uint, TerminalSessionPump> pumps = []; private readonly Dictionary<uint, TerminalSessionPump> pumps = [];
@@ -451,6 +461,11 @@ public sealed class TerminalDataPlane : ITerminalTransport, IAsyncDisposable
{ {
var buffer = new byte[ReceiveBufferBytes]; var buffer = new byte[ReceiveBufferBytes];
// Grown only when a message spans more than one read, and kept for the next one that does.
var message = Array.Empty<byte>();
var length = 0;
var oversized = false;
while (webSocket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested) while (webSocket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested)
{ {
var result = await webSocket var result = await webSocket
@@ -462,14 +477,73 @@ public sealed class TerminalDataPlane : ITerminalTransport, IAsyncDisposable
return; return;
} }
if (!result.EndOfMessage) // The common case — a keystroke, an acknowledgement — is one read, and goes straight through.
if (result.EndOfMessage && length == 0 && !oversized)
{ {
// A frame larger than the receive buffer. Terminal input and acknowledgements are tiny, DispatchSafely(buffer.AsSpan(0, result.Count));
// so this is either a bug or something hostile; dropping is the safe response.
continue; continue;
} }
Dispatch(buffer.AsSpan(0, result.Count)); if (!oversized && length + result.Count <= MaximumMessageBytes)
{
if (length + result.Count > message.Length)
{
Array.Resize(ref message, Math.Min(MaximumMessageBytes, Math.Max(message.Length * 2, length + result.Count)));
}
buffer.AsSpan(0, result.Count).CopyTo(message.AsSpan(length));
length += result.Count;
}
else
{
oversized = true;
}
if (!result.EndOfMessage)
{
continue;
}
if (!oversized)
{
DispatchSafely(message.AsSpan(0, length));
}
length = 0;
oversized = false;
}
}
/// <summary>Dispatches one frame, so that what goes wrong with one session stays with that session.</summary>
/// <remarks>
/// Every session shares this socket, and an exception escaping the receive loop closes it — so a resize
/// that reached a shell SSH.NET had already disposed, which is an ordinary moment at the end of every
/// session, used to drop every other tab to "reconnecting" and lose the output sent meanwhile. A frame
/// that cannot be acted on is a frame not acted on; the session it was for reports its own ending.
/// </remarks>
private void DispatchSafely(ReadOnlySpan<byte> frame)
{
try
{
Dispatch(frame);
}
catch (Exception exception) when (exception is not OutOfMemoryException and not OperationCanceledException)
{
// See the remarks.
}
}
/// <summary>Writes one input frame, without letting a session that is going away fault anything.</summary>
private async Task WriteInputAsync(TerminalSessionPump pump, byte[] input)
{
try
{
await pump.WriteInputAsync(input, lifetime.Token).ConfigureAwait(false);
}
catch (Exception exception) when (exception is not OutOfMemoryException)
{
// The shell ended or the connection dropped between the keystroke and the write. The session's
// own read loop reports that; a keystroke that had nowhere to go has nothing to add.
} }
} }
@@ -502,7 +576,7 @@ public sealed class TerminalDataPlane : ITerminalTransport, IAsyncDisposable
case TerminalClientOpcode.Input: case TerminalClientOpcode.Input:
// Fire and forget: a blocked write must not stall acknowledgements for other sessions // Fire and forget: a blocked write must not stall acknowledgements for other sessions
// sharing this socket. // sharing this socket.
_ = pump.WriteInputAsync(payload.ToArray(), lifetime.Token).AsTask(); _ = WriteInputAsync(pump, payload.ToArray());
break; break;
case TerminalClientOpcode.Acknowledge: case TerminalClientOpcode.Acknowledge:
@@ -107,6 +107,19 @@ public sealed class TerminalSessionPump : IAsyncDisposable
private readonly OutputBacklog backlog; private readonly OutputBacklog backlog;
private readonly CancellationTokenSource lifetime = new(); private readonly CancellationTokenSource lifetime = new();
/// <summary>One write to the remote at a time, in the order they were asked for.</summary>
/// <remarks>
/// The data plane hands each input frame over without waiting for the one before it — a blocked write
/// must not hold up the acknowledgements of every other session on the socket — so without this, two
/// frames' writes run at once. SSH.NET's <c>ShellStream</c> takes no lock around its write buffer, and its
/// flush sends what is there and then empties it by length: bytes added mid-flush are discarded unsent.
/// A snippet's RUN is its text and then a carriage return as two frames, which is exactly the shape that lost or
/// reordered keystrokes. Waiters on a <see cref="SemaphoreSlim"/> are released in the order they arrived,
/// and the data plane calls this from one receive loop, so arrival order is the order typed.
/// </remarks>
private readonly SemaphoreSlim inputGate = new(1, 1);
private int disposed; private int disposed;
/// <param name="sessionId">Identifies this session in every frame.</param> /// <param name="sessionId">Identifies this session in every frame.</param>
@@ -267,9 +280,20 @@ public sealed class TerminalSessionPump : IAsyncDisposable
} }
} }
/// <summary>Forwards keystrokes to the remote.</summary> /// <summary>Forwards keystrokes to the remote, after any still being written.</summary>
public ValueTask WriteInputAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken) => public async ValueTask WriteInputAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
session.WriteAsync(data, cancellationToken); {
await inputGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await session.WriteAsync(data, cancellationToken).ConfigureAwait(false);
}
finally
{
inputGate.Release();
}
}
/// <summary>Tells the remote the terminal was resized.</summary> /// <summary>Tells the remote the terminal was resized.</summary>
public void Resize(TerminalSize size) => session.Resize(size); public void Resize(TerminalSize size) => session.Resize(size);
@@ -78,6 +78,14 @@ internal sealed class FakeShellSession : ISshShellSession
/// <summary>The last size the pump forwarded, or null if it forwarded none.</summary> /// <summary>The last size the pump forwarded, or null if it forwarded none.</summary>
public TerminalSize? LastResize { get; private set; } public TerminalSize? LastResize { get; private set; }
/// <summary>How long the first write takes, so a test can see whether a second one overtakes it.</summary>
public TimeSpan FirstWriteDelay { get; init; }
/// <summary>Whether a resize throws, as SSH.NET's does once it has disposed the shell's stream.</summary>
public bool ResizeThrows { get; init; }
private int writes;
/// <summary>How many resizes were forwarded, so a dropped one is observable.</summary> /// <summary>How many resizes were forwarded, so a dropped one is observable.</summary>
public int ResizeCount { get; private set; } public int ResizeCount { get; private set; }
@@ -119,19 +127,27 @@ internal sealed class FakeShellSession : ISshShellSession
} }
/// <inheritdoc /> /// <inheritdoc />
public ValueTask WriteAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken) public async ValueTask WriteAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
{ {
if (Interlocked.Increment(ref writes) == 1 && FirstWriteDelay > TimeSpan.Zero)
{
await Task.Delay(FirstWriteDelay, cancellationToken).ConfigureAwait(false);
}
lock (gate) lock (gate)
{ {
written.AddRange(data.ToArray()); written.AddRange(data.ToArray());
} }
return ValueTask.CompletedTask;
} }
/// <inheritdoc /> /// <inheritdoc />
public void Resize(TerminalSize size) public void Resize(TerminalSize size)
{ {
if (ResizeThrows)
{
throw new ObjectDisposedException("ShellStream");
}
// Mirrors the real session: an unusable size is dropped rather than forwarded. // Mirrors the real session: an unusable size is dropped rather than forwarded.
if (!size.IsUsable) if (!size.IsUsable)
{ {
@@ -285,6 +285,86 @@ public sealed class TerminalDataPlaneTests : IAsyncDisposable
Encoding.UTF8.GetString(session.Written).ShouldBe("whoami\r"); Encoding.UTF8.GetString(session.Written).ShouldBe("whoami\r");
} }
/// <summary>
/// A paste is one input frame however long it is, and one longer than a single read used to be dropped
/// — or worse, its tail parsed as a frame of its own.
/// </summary>
[Fact]
public async Task APasteLargerThanOneRead_ReachesTheRemoteWhole()
{
Start();
using var socket = await ConnectAsync();
await using var session = new FakeShellSession();
await using var pump = CreatePump(session);
plane.Register(SessionId, pump);
var paste = new byte[300 * 1024];
Random.Shared.NextBytes(paste);
await SendAsync(socket, (byte)TerminalClientOpcode.Input, paste);
await WaitUntilAsync(() => session.Written.Length >= paste.Length);
session.Written.ShouldBe(paste);
}
/// <summary>
/// Frames are handed over without waiting for the write before them, so a slow write must still be
/// followed rather than overtaken: a snippet's RUN is its text and then a carriage return.
/// </summary>
[Fact]
public async Task InputFrames_AreWrittenInTheOrderTheyArrived_EvenWhenAWriteIsSlow()
{
Start();
using var socket = await ConnectAsync();
await using var session = new FakeShellSession { FirstWriteDelay = TimeSpan.FromMilliseconds(300) };
await using var pump = CreatePump(session);
plane.Register(SessionId, pump);
await SendAsync(socket, (byte)TerminalClientOpcode.Input, "sudo systemctl restart api"u8.ToArray());
await SendAsync(socket, (byte)TerminalClientOpcode.Input, "\r"u8.ToArray());
await WaitUntilAsync(() => session.Written.Length >= 27);
Encoding.UTF8.GetString(session.Written).ShouldBe("sudo systemctl restart api\r");
}
/// <summary>
/// A resize that reaches a shell SSH.NET has already disposed is an ordinary moment at the end of a
/// session, and it must not close the socket every other session shares.
/// </summary>
[Fact]
public async Task AResizeThatThrows_LeavesTheSocketWorkingForEverythingElse()
{
Start();
using var socket = await ConnectAsync();
await using var ending = new FakeShellSession { ResizeThrows = true };
await using var endingPump = CreatePump(ending);
plane.Register(SessionId, endingPump);
await using var other = new FakeShellSession();
await using var otherPump = CreatePump(other);
plane.Register(SessionId + 1, otherPump);
await SendAsync(
socket,
(byte)TerminalClientOpcode.Resize,
TerminalFrame.CreateResizePayload(new TerminalSize(80, 24, 800, 600)));
await SendAsync(socket, (byte)TerminalClientOpcode.Input, "still here"u8.ToArray(), SessionId + 1);
await WaitUntilAsync(() => other.Written.Length > 0);
Encoding.UTF8.GetString(other.Written).ShouldBe("still here");
socket.State.ShouldBe(WebSocketState.Open);
}
[Fact] [Fact]
public async Task AResize_ReachesTheRemote() public async Task AResize_ReachesTheRemote()
{ {