admin/VolcanicCore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(config): JS API subjects belong in publish.allow, not subscribe.allow

Browse Source 
The JetStream request/reply model has the CLIENT publishing requests to
$JS.API.<...> (and $JS.ACK.<...>) and receiving replies on _INBOX.>.
ConfigGenerator was granting these on the subscribe side instead, so
end users could never create consumers, fetch, ack, or query streams —
their JS calls silently timed out on unauthorized publishes. Also add
STREAM.INFO / CONSUMER.INFO / CONSUMER.DURABLE.CREATE / CONSUMER.DELETE,
which end-user clients need but weren't granted at all.

Tests now assert placement (publish vs subscribe block) so this can't
regress silently. The example switches to Test_2 as the destination
stream name to match the real deployment.
This commit is contained in:
EugeneTes
2026-08-27 15:37:56 +00:00
parent ad1a2462e5
commit ddaa43d924
3 changed files with 83 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
Examples/NodeExample/Program.cs
View File

@@ -4,24 +4,21 @@ using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
// Two NATS connections, one per node:
// - "Test" node: publisher, authenticated as TEST_USER
// - "Test 2" node: consumer, authenticated as TEST2_USER
// - "Test" node: publisher (authed as "test")
// - "Test 2" node: consumer (authed as "test")
//
// Assumes the graph in VolcanicCore has: a "Test" node capturing "test.>" (or at
// least "test.income.one"), a "Test 2" node, and an edge Test -> Test 2 that
// routes matching subjects. Also assumes TEST_USER is assigned to Test and
// TEST2_USER is assigned to Test 2 in the admin UI.
// Assumes the graph in VolcanicCore has: a "Test" node capturing "test.>" (or
// at least "test.income.one"), a "Test 2" node, and an edge Test -> Test 2 that
// routes matching subjects. Also assumes the "test" user is assigned to both
// nodes in the admin UI.
var natsUrl = Env("NATS_URL", "nats://127.0.0.1:4222");
var testUser = Env("TEST_USER", "test");
var testPassword = Env("TEST_PASSWORD", "test");
var test2User = Env("TEST2_USER", "test2");
var test2Password= Env("TEST2_PASSWORD", "test2");
var messageCount = int.Parse(Env("MESSAGE_COUNT", "5"));
const string testStream = "Test";
const string test2Stream = "Test 2";
const string natsUrl = "wss://nats-core.tes.gd";
const string username = "test";
const string password = "test";
const string testStream = "Test";
const string test2Stream = "Test_2";
const string publishSubject = "test.income.one";
const int messageCount = 5;
using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };
@@ -30,19 +27,19 @@ await using var pubConn = new NatsConnection(new NatsOpts
{
Url = natsUrl,
Name = $"example-publisher-{testStream}",
AuthOpts = new NatsAuthOpts { Username = testUser, Password = testPassword },
AuthOpts = new NatsAuthOpts { Username = username, Password = password },
});
await pubConn.ConnectAsync();
Console.WriteLine($"[pub] connected to {natsUrl} as {testUser} (node: {testStream})");
Console.WriteLine($"[pub] connected to {natsUrl} as {username} (node: {testStream})");
await using var subConn = new NatsConnection(new NatsOpts
{
Url = natsUrl,
Name = $"example-consumer-{test2Stream}",
AuthOpts = new NatsAuthOpts { Username = test2User, Password = test2Password },
AuthOpts = new NatsAuthOpts { Username = username, Password = password },
});
await subConn.ConnectAsync();
Console.WriteLine($"[sub] connected to {natsUrl} as {test2User} (node: {test2Stream})");
Console.WriteLine($"[sub] connected to {natsUrl} as {username} (node: {test2Stream})");
var subJs = new NatsJSContext(subConn);
var test2 = await subJs.GetStreamAsync(test2Stream, cancellationToken: cts.Token);
@@ -86,6 +83,3 @@ cts.Cancel();
try { await reader; } catch (OperationCanceledException) { }
Console.WriteLine("done.");
static string Env(string key, string fallback) =>
Environment.GetEnvironmentVariable(key) is { Length: > 0 } v ? v : fallback;

30
src/VolcanicCore.Core/NatsConfig/ConfigGenerator.cs
View File

@@ -75,20 +75,30 @@ public static class ConfigGenerator
{
var user = users[i];
var assignedNodeIds = assignmentsByUser.GetValueOrDefault(user.Id, new List<string>());
var publishAllow = assignedNodeIds
.Where(subjectsByNodeId.ContainsKey)
.SelectMany(id => subjectsByNodeId[id])
// JetStream request/reply model: the client PUBLISHES a request to
// $JS.API.<...> (or $JS.ACK.<...>) and SUBSCRIBES to _INBOX.> to
// receive the reply. So the API grants belong on the publish side.
var publishAllow = new List<string>();
publishAllow.AddRange(
assignedNodeIds
.Where(subjectsByNodeId.ContainsKey)
.SelectMany(id => subjectsByNodeId[id]));
foreach (var id in assignedNodeIds)
{
if (!streamByNodeId.TryGetValue(id, out var stream)) continue;
publishAllow.Add($"$JS.API.STREAM.INFO.{stream}");
publishAllow.Add($"$JS.API.CONSUMER.CREATE.{stream}.>");
publishAllow.Add($"$JS.API.CONSUMER.DURABLE.CREATE.{stream}.>");
publishAllow.Add($"$JS.API.CONSUMER.INFO.{stream}.>");
publishAllow.Add($"$JS.API.CONSUMER.DELETE.{stream}.>");
publishAllow.Add($"$JS.API.CONSUMER.MSG.NEXT.{stream}.>");
publishAllow.Add($"$JS.ACK.{stream}.>");
}
publishAllow = publishAllow
.Distinct(StringComparer.Ordinal)
.OrderBy(s => s, StringComparer.Ordinal)
.ToList();
var subscribeAllow = new List<string> { "_INBOX.>" };
foreach (var id in assignedNodeIds.OrderBy(id => streamByNodeId.GetValueOrDefault(id, ""), StringComparer.Ordinal))
{
if (!streamByNodeId.TryGetValue(id, out var stream)) continue;
subscribeAllow.Add($"$JS.API.CONSUMER.CREATE.{stream}.>");
subscribeAllow.Add($"$JS.API.CONSUMER.MSG.NEXT.{stream}.>");
subscribeAllow.Add($"$JS.ACK.{stream}.>");
}
sb.AppendLine(" {");
sb.AppendLine($" user: \"{user.Username}\"");

58
tests/VolcanicCore.Core.Tests/ConfigGeneratorTests.cs
View File

@@ -50,13 +50,25 @@ public class ConfigGeneratorTests
Assert.Contains("user: \"alice\"", text);
Assert.Contains("password: \"alice-pass\"", text);
Assert.Contains("publish {", text);
// publish subjects from the node's subjects list
Assert.Contains("\"mail.inbox.test-sender.*\"", text);
// subscribe grants for that stream
Assert.Contains("\"$JS.API.CONSUMER.CREATE.SENDER_STREAM.>\"", text);
Assert.Contains("\"$JS.API.CONSUMER.MSG.NEXT.SENDER_STREAM.>\"", text);
Assert.Contains("\"$JS.ACK.SENDER_STREAM.>\"", text);
// JetStream request/reply uses the client's PUBLISH side for API calls,
// and only _INBOX.> on the subscribe side to receive replies.
var (pub, sub) = ExtractPermissionBlocks(text, "alice");
// Ingress subjects from the assigned node.
Assert.Contains("\"mail.inbox.test-sender.*\"", pub);
// Every JS API call this user needs to make is an outbound publish.
Assert.Contains("\"$JS.API.STREAM.INFO.SENDER_STREAM\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.CREATE.SENDER_STREAM.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.DURABLE.CREATE.SENDER_STREAM.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.INFO.SENDER_STREAM.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.DELETE.SENDER_STREAM.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.MSG.NEXT.SENDER_STREAM.>\"", pub);
Assert.Contains("\"$JS.ACK.SENDER_STREAM.>\"", pub);
// Subscribe side is _INBOX.> only — no $JS subjects.
Assert.Contains("\"_INBOX.>\"", sub);
Assert.DoesNotContain("$JS", sub);
}
[Fact]
@@ -72,10 +84,13 @@ public class ConfigGeneratorTests
Assignments: new[] { new Assignment("n1", "u"), new Assignment("n2", "u") },
SysPassword: "sys");
var text = ConfigGenerator.Generate(SampleInput(snap));
Assert.Contains("\"a.>\"", text);
Assert.Contains("\"b.>\"", text);
Assert.Contains("\"$JS.API.CONSUMER.CREATE.S1.>\"", text);
Assert.Contains("\"$JS.API.CONSUMER.CREATE.S2.>\"", text);
var (pub, _) = ExtractPermissionBlocks(text, "bob");
Assert.Contains("\"a.>\"", pub);
Assert.Contains("\"b.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.CREATE.S1.>\"", pub);
Assert.Contains("\"$JS.API.CONSUMER.CREATE.S2.>\"", pub);
Assert.Contains("\"$JS.API.STREAM.INFO.S1\"", pub);
Assert.Contains("\"$JS.API.STREAM.INFO.S2\"", pub);
}
[Fact]
@@ -96,6 +111,27 @@ public class ConfigGeneratorTests
Assert.DoesNotContain("$JS.API", text.Substring(text.IndexOf("user: \"solo\"", StringComparison.Ordinal)));
}
// Returns the raw text between `publish { allow: [ ... ] }` and
// `subscribe { allow: [ ... ] }` for the named user, so a test can assert
// where a subject appears (publish side vs subscribe side).
private static (string Publish, string Subscribe) ExtractPermissionBlocks(string text, string username)
{
var marker = $"user: \"{username}\"";
var start = text.IndexOf(marker, StringComparison.Ordinal);
Assert.NotEqual(-1, start);
var section = text.Substring(start);
// Cut at the next user boundary so a later user's grants don't leak in.
var next = section.IndexOf(" }", StringComparison.Ordinal);
if (next > 0) section = section.Substring(0, next);
var pubIdx = section.IndexOf("publish {", StringComparison.Ordinal);
var subIdx = section.IndexOf("subscribe {", StringComparison.Ordinal);
Assert.True(pubIdx >= 0 && subIdx > pubIdx, "expected publish + subscribe blocks in order");
var pub = section.Substring(pubIdx, subIdx - pubIdx);
var sub = section.Substring(subIdx);
return (pub, sub);
}
private static Snapshot Fixture()
{
var sender = new Node("n1", "sender", "SENDER_STREAM",