feat: add no-responders CONNECT validation and tests
Reject connections that send no_responders:true without headers:true, since the 503 HMSG response requires header support. Add three tests: connection rejection, acceptance with headers, and 503 delivery flow.
This commit is contained in:
@@ -374,6 +374,15 @@ public sealed class NatsClient : IDisposable
|
|||||||
Account.AddClient(Id);
|
Account.AddClient(Id);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Validate no_responders requires headers
|
||||||
|
if (ClientOpts.NoResponders && !ClientOpts.Headers)
|
||||||
|
{
|
||||||
|
_logger.LogDebug("Client {ClientId} no_responders requires headers", Id);
|
||||||
|
await CloseWithReasonAsync(ClientClosedReason.NoRespondersRequiresHeaders,
|
||||||
|
NatsProtocol.ErrNoRespondersRequiresHeaders);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
_flags.SetFlag(ClientFlags.ConnectReceived);
|
_flags.SetFlag(ClientFlags.ConnectReceived);
|
||||||
_flags.SetFlag(ClientFlags.ConnectProcessFinished);
|
_flags.SetFlag(ClientFlags.ConnectProcessFinished);
|
||||||
_logger.LogDebug("CONNECT received from client {ClientId}, name={ClientName}", Id, ClientOpts?.Name);
|
_logger.LogDebug("CONNECT received from client {ClientId}, name={ClientName}", Id, ClientOpts?.Name);
|
||||||
|
|||||||
117
tests/NATS.Server.Tests/NoRespondersTests.cs
Normal file
117
tests/NATS.Server.Tests/NoRespondersTests.cs
Normal file
@@ -0,0 +1,117 @@
|
|||||||
|
using System.Net;
|
||||||
|
using System.Net.Sockets;
|
||||||
|
using System.Text;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using NATS.Server;
|
||||||
|
|
||||||
|
namespace NATS.Server.Tests;
|
||||||
|
|
||||||
|
public class NoRespondersTests : IAsyncLifetime
|
||||||
|
{
|
||||||
|
private readonly NatsServer _server;
|
||||||
|
private readonly int _port;
|
||||||
|
private readonly CancellationTokenSource _cts = new();
|
||||||
|
|
||||||
|
public NoRespondersTests()
|
||||||
|
{
|
||||||
|
_port = GetFreePort();
|
||||||
|
_server = new NatsServer(new NatsOptions { Port = _port }, NullLoggerFactory.Instance);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async Task InitializeAsync()
|
||||||
|
{
|
||||||
|
_ = _server.StartAsync(_cts.Token);
|
||||||
|
await _server.WaitForReadyAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
public async Task DisposeAsync()
|
||||||
|
{
|
||||||
|
await _cts.CancelAsync();
|
||||||
|
_server.Dispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static int GetFreePort()
|
||||||
|
{
|
||||||
|
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||||
|
sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||||
|
return ((IPEndPoint)sock.LocalEndPoint!).Port;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<Socket> ConnectClientAsync()
|
||||||
|
{
|
||||||
|
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||||
|
await sock.ConnectAsync(IPAddress.Loopback, _port);
|
||||||
|
return sock;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static async Task<string> ReadUntilAsync(Socket sock, string expected, int timeoutMs = 5000)
|
||||||
|
{
|
||||||
|
using var cts = new CancellationTokenSource(timeoutMs);
|
||||||
|
var sb = new StringBuilder();
|
||||||
|
var buf = new byte[4096];
|
||||||
|
while (!sb.ToString().Contains(expected))
|
||||||
|
{
|
||||||
|
var n = await sock.ReceiveAsync(buf, SocketFlags.None, cts.Token);
|
||||||
|
if (n == 0) break;
|
||||||
|
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||||
|
}
|
||||||
|
return sb.ToString();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task NoResponders_without_headers_closes_connection()
|
||||||
|
{
|
||||||
|
using var client = await ConnectClientAsync();
|
||||||
|
|
||||||
|
// Read INFO
|
||||||
|
await ReadUntilAsync(client, "\r\n");
|
||||||
|
|
||||||
|
// Send CONNECT with no_responders:true but headers:false
|
||||||
|
await client.SendAsync(Encoding.ASCII.GetBytes(
|
||||||
|
"CONNECT {\"no_responders\":true,\"headers\":false}\r\n"));
|
||||||
|
|
||||||
|
// Should receive -ERR and connection should close
|
||||||
|
var response = await ReadUntilAsync(client, "-ERR");
|
||||||
|
response.ShouldContain("-ERR");
|
||||||
|
response.ShouldContain("No Responders Requires Headers Support");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task NoResponders_with_headers_accepted()
|
||||||
|
{
|
||||||
|
using var client = await ConnectClientAsync();
|
||||||
|
|
||||||
|
// Read INFO
|
||||||
|
await ReadUntilAsync(client, "\r\n");
|
||||||
|
|
||||||
|
// Send CONNECT with both no_responders and headers true, then PING
|
||||||
|
await client.SendAsync(Encoding.ASCII.GetBytes(
|
||||||
|
"CONNECT {\"no_responders\":true,\"headers\":true}\r\nPING\r\n"));
|
||||||
|
|
||||||
|
// Should receive PONG (connection stays alive)
|
||||||
|
var response = await ReadUntilAsync(client, "PONG\r\n");
|
||||||
|
response.ShouldContain("PONG\r\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task NoResponders_sends_503_when_no_subscribers()
|
||||||
|
{
|
||||||
|
using var client = await ConnectClientAsync();
|
||||||
|
|
||||||
|
// Read INFO
|
||||||
|
await ReadUntilAsync(client, "\r\n");
|
||||||
|
|
||||||
|
// CONNECT with no_responders and headers enabled
|
||||||
|
// SUB to the reply inbox so we can receive the 503
|
||||||
|
// PUB to a subject with no subscribers, using a reply-to subject
|
||||||
|
await client.SendAsync(Encoding.ASCII.GetBytes(
|
||||||
|
"CONNECT {\"no_responders\":true,\"headers\":true}\r\n" +
|
||||||
|
"SUB _INBOX.reply 1\r\n" +
|
||||||
|
"PUB no.subscribers _INBOX.reply 5\r\nHello\r\n"));
|
||||||
|
|
||||||
|
// Should receive HMSG with 503 status on the reply subject
|
||||||
|
var response = await ReadUntilAsync(client, "HMSG");
|
||||||
|
response.ShouldContain("HMSG _INBOX.reply 1");
|
||||||
|
response.ShouldContain("503");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user