feat: add per-account SubList isolation for message routing

Subscriptions and message routing now go through account-specific SubLists
instead of a single global SubList. Clients in different accounts cannot
see each other's messages. When no account is specified (or auth is not
configured), all clients share the global $G account.
This commit is contained in:
Joseph Doherty
2026-02-22 23:00:59 -05:00
parent 2980a343c1
commit 9cb3e2fe0f
3 changed files with 121 additions and 9 deletions

View File

@@ -260,6 +260,13 @@ public sealed class NatsClient : IDisposable
_logger.LogDebug("Client {ClientId} authenticated as {Identity}", Id, result.Identity);
}
// If no account was assigned by auth, assign global account
if (Account == null && Router is NatsServer server2)
{
Account = server2.GetOrCreateAccount(Account.GlobalAccountName);
Account.AddClient(Id);
}
ConnectReceived = true;
_logger.LogDebug("CONNECT received from client {ClientId}, name={ClientName}", Id, ClientOpts?.Name);
}
@@ -286,8 +293,7 @@ public sealed class NatsClient : IDisposable
_logger.LogDebug("SUB {Subject} {Sid} from client {ClientId}", cmd.Subject, cmd.Sid, Id);
if (Router is ISubListAccess sl)
sl.SubList.Insert(sub);
Account?.SubList.Insert(sub);
}
private void ProcessUnsub(ParsedCommand cmd)
@@ -306,8 +312,7 @@ public sealed class NatsClient : IDisposable
_subs.Remove(cmd.Sid!);
if (Router is ISubListAccess sl)
sl.SubList.Remove(sub);
Account?.SubList.Remove(sub);
}
private async ValueTask ProcessPubAsync(ParsedCommand cmd)
@@ -488,6 +493,7 @@ public sealed class NatsClient : IDisposable
public void Dispose()
{
_permissions?.Dispose();
_clientCts?.Dispose();
_stream.Dispose();
_socket.Dispose();

View File

@@ -13,7 +13,6 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
{
private readonly NatsOptions _options;
private readonly ConcurrentDictionary<ulong, NatsClient> _clients = new();
private readonly SubList _subList = new();
private readonly ServerInfo _serverInfo;
private readonly ILogger<NatsServer> _logger;
private readonly ILoggerFactory _loggerFactory;
@@ -24,7 +23,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
private Socket? _listener;
private ulong _nextClientId;
public SubList SubList => _subList;
public SubList SubList => _globalAccount.SubList;
public Task WaitForReadyAsync() => _listeningStarted.Task;
@@ -148,7 +147,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
public void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory<byte> headers,
ReadOnlyMemory<byte> payload, NatsClient sender)
{
var result = _subList.Match(subject);
var subList = sender.Account?.SubList ?? _globalAccount.SubList;
var result = subList.Match(subject);
// Deliver to plain subscribers
foreach (var sub in result.PlainSubs)
@@ -205,7 +205,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
{
_clients.TryRemove(client.Id, out _);
_logger.LogDebug("Removed client {ClientId}", client.Id);
client.RemoveAllSubscriptions(_subList);
var subList = client.Account?.SubList ?? _globalAccount.SubList;
client.RemoveAllSubscriptions(subList);
client.Account?.RemoveClient(client.Id);
}
@@ -214,7 +215,6 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
_listener?.Dispose();
foreach (var client in _clients.Values)
client.Dispose();
_subList.Dispose();
foreach (var account in _accounts.Values)
account.Dispose();
}