dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhance ConfigProcessor & add 110 Go-parity opts tests (Task 22)

Browse Source 
Port configuration parsing for NKey users, gateway remotes, leaf node
remotes, auth timeout, write_deadline, websocket ping_interval, and
token+users conflict validation. Add RemoteGatewayOptions, enhanced
LeafNodeOptions with remotes support.

110 new tests ported from opts_test.go.
This commit is contained in:
Joseph Doherty
2026-02-24 20:17:48 -05:00
parent f35961abea
commit 1a3fe91611
5 changed files with 2201 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/NATS.Server/Configuration/ClusterOptions.cs
View File

@@ -9,4 +9,7 @@ public sealed class ClusterOptions
public List<string> Routes { get; set; } = [];
public List<string> Accounts { get; set; } = [];
public RouteCompression Compression { get; set; } = RouteCompression.None;
// Go: opts.go — cluster write_deadline
public TimeSpan WriteDeadline { get; set; }
}

457
src/NATS.Server/Configuration/ConfigProcessor.cs
View File

@@ -271,7 +271,13 @@ public static class ConfigProcessor
ParseMqtt(mqttDict, opts, errors);
break;
// Unknown keys silently ignored (cluster, jetstream, gateway, leafnode, etc.)
// WebSocket
case "websocket" or "ws":
if (value is Dictionary<string, object?> wsDict)
ParseWebSocket(wsDict, opts, errors);
break;
// Unknown keys silently ignored (accounts, resolver, operator, etc.)
default:
break;
}
@@ -417,6 +423,26 @@ public static class ConfigProcessor
errors.Add($"Invalid cluster.listen: {ex.Message}");
}
break;
case "write_deadline":
try
{
options.WriteDeadline = ParseDuration(value);
}
catch (Exception ex)
{
errors.Add($"Invalid cluster.write_deadline: {ex.Message}");
}
break;
case "routes":
if (value is List<object?> routeList)
options.Routes = ToStringList(routeList).ToList();
break;
case "pool_size":
options.PoolSize = ToInt(value);
break;
default:
break;
}
}
@@ -434,6 +460,12 @@ public static class ConfigProcessor
case "name":
options.Name = ToString(value);
break;
case "host" or "net":
options.Host = ToString(value);
break;
case "port":
options.Port = ToInt(value);
break;
case "listen":
try
{
@@ -448,6 +480,51 @@ public static class ConfigProcessor
errors.Add($"Invalid gateway.listen: {ex.Message}");
}
break;
case "reject_unknown_cluster" or "reject_unknown":
options.RejectUnknown = ToBool(value);
break;
case "advertise":
options.Advertise = ToString(value);
break;
case "connect_retries":
options.ConnectRetries = ToInt(value);
break;
case "connect_backoff":
options.ConnectBackoff = ToBool(value);
break;
case "write_deadline":
try
{
options.WriteDeadline = ParseDuration(value);
}
catch (Exception ex)
{
errors.Add($"Invalid gateway.write_deadline: {ex.Message}");
}
break;
case "authorization" or "authentication":
if (value is Dictionary<string, object?> authDict)
ParseGatewayAuthorization(authDict, options, errors);
break;
case "gateways":
// Must be a list, not a map
if (value is List<object?> gwList)
{
foreach (var item in gwList)
{
if (item is Dictionary<string, object?> gwDict)
options.RemoteGateways.Add(ParseRemoteGateway(gwDict, errors));
}
}
else if (value is Dictionary<string, object?>)
{
errors.Add("gateway.gateways must be an array, not a map");
}
break;
default:
break;
}
}
@@ -455,31 +532,209 @@ public static class ConfigProcessor
return options;
}
private static void ParseGatewayAuthorization(Dictionary<string, object?> dict, GatewayOptions options, List<string> errors)
{
// Gateway authorization only supports a single user — users array is not allowed.
// Go reference: opts.go parseGateway — "does not allow multiple users"
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
{
case "user" or "username":
options.Username = ToString(value);
break;
case "pass" or "password":
options.Password = ToString(value);
break;
case "token":
// Token-only auth
options.Username = ToString(value);
break;
case "timeout":
options.AuthTimeout = ToDouble(value);
break;
case "users":
// Not supported in gateway auth
errors.Add("gateway authorization does not allow multiple users");
break;
default:
break;
}
}
}
private static RemoteGatewayOptions ParseRemoteGateway(Dictionary<string, object?> dict, List<string> errors)
{
var remote = new RemoteGatewayOptions();
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
{
case "name":
remote.Name = ToString(value);
break;
case "url":
remote.Urls.Add(ToString(value));
break;
case "urls":
if (value is List<object?> urlList)
remote.Urls.AddRange(ToStringList(urlList));
break;
default:
break;
}
}
return remote;
}
private static LeafNodeOptions ParseLeafNode(Dictionary<string, object?> dict, List<string> errors)
{
var options = new LeafNodeOptions();
foreach (var (key, value) in dict)
{
if (key.Equals("listen", StringComparison.OrdinalIgnoreCase))
switch (key.ToLowerInvariant())
{
try
{
var (host, port) = ParseHostPort(value);
if (host is not null)
options.Host = host;
if (port is not null)
options.Port = port.Value;
}
catch (Exception ex)
{
errors.Add($"Invalid leafnode.listen: {ex.Message}");
}
case "host" or "net":
options.Host = ToString(value);
break;
case "port":
options.Port = ToInt(value);
break;
case "listen":
try
{
var (host, port) = ParseHostPort(value);
if (host is not null)
options.Host = host;
if (port is not null)
options.Port = port.Value;
}
catch (Exception ex)
{
errors.Add($"Invalid leafnode.listen: {ex.Message}");
}
break;
case "advertise":
options.Advertise = ToString(value);
break;
case "write_deadline":
try
{
options.WriteDeadline = ParseDuration(value);
}
catch (Exception ex)
{
errors.Add($"Invalid leafnode.write_deadline: {ex.Message}");
}
break;
case "authorization" or "authentication":
if (value is Dictionary<string, object?> authDict)
ParseLeafNodeAuthorization(authDict, options, errors);
break;
case "remotes":
if (value is List<object?> remoteList)
{
foreach (var item in remoteList)
{
if (item is Dictionary<string, object?> remoteDict)
options.RemoteLeaves.Add(ParseRemoteLeaf(remoteDict, errors));
}
}
break;
case "no_advertise":
case "compress":
case "tls":
case "deny_exports":
case "deny_imports":
// Silently accepted fields (some are handled elsewhere)
break;
default:
break;
}
}
return options;
}
private static void ParseLeafNodeAuthorization(Dictionary<string, object?> dict, LeafNodeOptions options, List<string> errors)
{
// Go reference: opts.go parseLeafNode authorization block
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
{
case "user" or "username":
options.Username = ToString(value);
break;
case "pass" or "password":
options.Password = ToString(value);
break;
case "token":
options.Username = ToString(value);
break;
case "timeout":
// Go stores leafnode auth_timeout as float64 seconds.
// Supports plain numbers and duration strings like "1m".
options.AuthTimeout = value switch
{
long l => (double)l,
double d => d,
string s => ParseDuration(s).TotalSeconds,
_ => throw new FormatException($"Invalid leafnode auth timeout: {value?.GetType().Name}"),
};
break;
default:
break;
}
}
}
private static RemoteLeafOptions ParseRemoteLeaf(Dictionary<string, object?> dict, List<string> errors)
{
var remote = new RemoteLeafOptions();
var urls = new List<string>();
string? localAccount = null;
string? credentials = null;
var dontRandomize = false;
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
{
case "url":
urls.Add(ToString(value));
break;
case "urls":
if (value is List<object?> urlList)
urls.AddRange(ToStringList(urlList));
break;
case "account":
localAccount = ToString(value);
break;
case "credentials" or "creds":
credentials = ToString(value);
break;
case "dont_randomize" or "no_randomize":
dontRandomize = ToBool(value);
break;
default:
break;
}
}
return new RemoteLeafOptions
{
LocalAccount = localAccount,
Credentials = credentials,
Urls = urls,
DontRandomize = dontRandomize,
};
}
private static JetStreamOptions ParseJetStream(Dictionary<string, object?> dict, List<string> errors)
{
var options = new JetStreamOptions();
@@ -522,6 +777,9 @@ public static class ConfigProcessor
private static void ParseAuthorization(Dictionary<string, object?> dict, NatsOptions opts, List<string> errors)
{
string? token = null;
List<object?>? userList = null;
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
@@ -533,7 +791,8 @@ public static class ConfigProcessor
opts.Password = ToString(value);
break;
case "token":
opts.Authorization = ToString(value);
token = ToString(value);
opts.Authorization = token;
break;
case "timeout":
opts.AuthTimeout = value switch
@@ -545,19 +804,43 @@ public static class ConfigProcessor
};
break;
case "users":
if (value is List<object?> userList)
opts.Users = ParseUsers(userList, errors);
if (value is List<object?> ul)
userList = ul;
break;
default:
// Unknown auth keys silently ignored
break;
}
}
// Validate: token cannot be combined with users array.
// Go reference: opts.go — "cannot have a token with a users array"
if (token is not null && userList is not null)
{
errors.Add("Cannot have a token with a users array");
return;
}
if (userList is not null)
{
var (plainUsers, nkeyUsers) = ParseUsersAndNkeys(userList, errors);
if (plainUsers.Count > 0)
opts.Users = plainUsers;
if (nkeyUsers.Count > 0)
opts.NKeys = nkeyUsers;
}
}
private static List<User> ParseUsers(List<object?> list, List<string> errors)
/// <summary>
/// Splits a users array into plain users and NKey users.
/// An entry with an "nkey" field is an NKey user; entries with "user" are plain users.
/// Go reference: opts.go — parseUsers (lines ~2500-2700).
/// </summary>
private static (List<User> PlainUsers, List<Auth.NKeyUser> NkeyUsers) ParseUsersAndNkeys(List<object?> list, List<string> errors)
{
var users = new List<User>();
var plainUsers = new List<User>();
var nkeyUsers = new List<Auth.NKeyUser>();
foreach (var item in list)
{
if (item is not Dictionary<string, object?> userDict)
@@ -568,6 +851,7 @@ public static class ConfigProcessor
string? username = null;
string? password = null;
string? nkey = null;
string? account = null;
Permissions? permissions = null;
@@ -581,6 +865,9 @@ public static class ConfigProcessor
case "pass" or "password":
password = ToString(value);
break;
case "nkey":
nkey = ToString(value);
break;
case "account":
account = ToString(value);
break;
@@ -591,13 +878,28 @@ public static class ConfigProcessor
}
}
if (nkey is not null)
{
// NKey user: validate no password and valid NKey format
if (!ValidateNkey(nkey, password is not null, errors))
continue;
nkeyUsers.Add(new Auth.NKeyUser
{
Nkey = nkey,
Permissions = permissions,
Account = account,
});
continue;
}
if (username is null)
{
errors.Add("User entry missing 'user' field");
continue;
}
users.Add(new User
plainUsers.Add(new User
{
Username = username,
Password = password ?? string.Empty,
@@ -606,7 +908,36 @@ public static class ConfigProcessor
});
}
return users;
return (plainUsers, nkeyUsers);
}
/// <summary>
/// Validates an NKey public key string.
/// Go reference: opts.go — nkey must start with 'U' and be at least 56 chars.
/// </summary>
private const int NKeyMinLen = 56;
private static bool ValidateNkey(string nkey, bool hasPassword, List<string> errors)
{
if (nkey.Length < NKeyMinLen || !nkey.StartsWith('U'))
{
errors.Add($"Not a valid public NKey: {nkey}");
return false;
}
if (hasPassword)
{
errors.Add("NKey user entry cannot have a password");
return false;
}
return true;
}
private static List<User> ParseUsers(List<object?> list, List<string> errors)
{
var (plainUsers, _) = ParseUsersAndNkeys(list, errors);
return plainUsers;
}
private static Permissions ParsePermissions(Dictionary<string, object?> dict, List<string> errors)
@@ -869,6 +1200,90 @@ public static class ConfigProcessor
}
}
// ─── WebSocket parsing ────────────────────────────────────────────────────
// Reference: Go server/opts.go parseWebsocket (lines ~5600-5700)
private static void ParseWebSocket(Dictionary<string, object?> dict, NatsOptions opts, List<string> errors)
{
var ws = opts.WebSocket ?? new WebSocketOptions();
foreach (var (key, value) in dict)
{
switch (key.ToLowerInvariant())
{
case "listen":
try
{
var (host, port) = ParseHostPort(value);
if (host is not null) ws.Host = host;
if (port is not null) ws.Port = port.Value;
}
catch (Exception ex)
{
errors.Add($"Invalid websocket.listen: {ex.Message}");
}
break;
case "port":
ws.Port = ToInt(value);
break;
case "host" or "net":
ws.Host = ToString(value);
break;
case "advertise":
ws.Advertise = ToString(value);
break;
case "no_auth_user":
ws.NoAuthUser = ToString(value);
break;
case "no_tls":
ws.NoTls = ToBool(value);
break;
case "same_origin":
ws.SameOrigin = ToBool(value);
break;
case "compression":
ws.Compression = ToBool(value);
break;
case "ping_interval":
try
{
ws.PingInterval = ParseDuration(value);
}
catch (Exception ex)
{
errors.Add($"Invalid websocket.ping_interval: {ex.Message}");
}
break;
case "handshake_timeout":
try
{
ws.HandshakeTimeout = ParseDuration(value);
}
catch (Exception ex)
{
errors.Add($"Invalid websocket.handshake_timeout: {ex.Message}");
}
break;
case "jwt_cookie":
ws.JwtCookie = ToString(value);
break;
case "username_header" or "username_cookie":
ws.UsernameCookie = ToString(value);
break;
case "token_cookie":
ws.TokenCookie = ToString(value);
break;
default:
break;
}
}
opts.WebSocket = ws;
}
private static void ParseMqttTls(Dictionary<string, object?> dict, MqttOptions mqtt, List<string> errors)
{
foreach (var (key, value) in dict)

22
src/NATS.Server/Configuration/GatewayOptions.cs
View File

@@ -6,4 +6,26 @@ public sealed class GatewayOptions
public string Host { get; set; } = "0.0.0.0";
public int Port { get; set; }
public List<string> Remotes { get; set; } = [];
// Go: opts.go — gateway authorization fields
public bool RejectUnknown { get; set; }
public string? Username { get; set; }
public string? Password { get; set; }
public double AuthTimeout { get; set; }
public string? Advertise { get; set; }
public int ConnectRetries { get; set; }
public bool ConnectBackoff { get; set; }
public TimeSpan WriteDeadline { get; set; }
// Go: opts.go — gateways remotes list (RemoteGatewayOpts)
public List<RemoteGatewayOptions> RemoteGateways { get; set; } = [];
}
/// <summary>
/// Go: opts.go RemoteGatewayOpts struct — a single remote gateway entry.
/// </summary>
public sealed class RemoteGatewayOptions
{
public string? Name { get; set; }
public List<string> Urls { get; set; } = [];
}

78
src/NATS.Server/Configuration/LeafNodeOptions.cs
View File

@@ -1,49 +1,63 @@
namespace NATS.Server.Configuration;
/// <summary>
/// Remote leaf node entry parsed from the remotes[] array inside a leafnodes {} block.
/// Go reference: opts.go RemoteLeafOpts struct.
/// </summary>
public sealed class RemoteLeafOptions
{
/// <summary>Local account to bind this remote to.</summary>
public string? LocalAccount { get; init; }
/// <summary>Path to credentials file.</summary>
public string? Credentials { get; init; }
/// <summary>URLs for this remote entry.</summary>
public List<string> Urls { get; init; } = [];
/// <summary>Whether to not randomize URL order.</summary>
public bool DontRandomize { get; init; }
}
public sealed class LeafNodeOptions
{
public string Host { get; set; } = "0.0.0.0";
public int Port { get; set; }
// Auth for leaf listener
public string? Username { get; set; }
public string? Password { get; set; }
public double AuthTimeout { get; set; }
// Advertise address
public string? Advertise { get; set; }
// Per-subsystem write deadline
public TimeSpan WriteDeadline { get; set; }
/// <summary>
/// Simple URL list for programmatic setup (tests, server-code wiring).
/// Parsed config populates RemoteLeaves instead.
/// </summary>
public List<string> Remotes { get; set; } = [];
/// <summary>
/// JetStream domain for this leaf node. When set, the domain is propagated
/// during the leaf handshake for domain-aware JetStream routing.
/// Go reference: leafnode.go — JsDomain in leafNodeCfg.
/// Remote leaf node entries parsed from a config file (remotes: [] array).
/// Each entry has a local account, credentials, and a list of URLs.
/// </summary>
public List<RemoteLeafOptions> RemoteLeaves { get; set; } = [];
/// <summary>
/// JetStream domain for this leaf node.
/// Go reference: leafnode.go -- JsDomain in leafNodeCfg.
/// </summary>
public string? JetStreamDomain { get; set; }
/// <summary>
/// Subjects to deny exporting (hub→leaf direction). Messages matching any of
/// these patterns will not be forwarded from the hub to the leaf.
/// Supports wildcards (* and >).
/// Go reference: leafnode.go — DenyExports in RemoteLeafOpts (opts.go:231).
/// </summary>
public List<string> DenyExports { get; set; } = [];
/// <summary>
/// Subjects to deny importing (leaf→hub direction). Messages matching any of
/// these patterns will not be forwarded from the leaf to the hub.
/// Supports wildcards (* and >).
/// Go reference: leafnode.go — DenyImports in RemoteLeafOpts (opts.go:230).
/// </summary>
public List<string> DenyImports { get; set; } = [];
/// <summary>
/// Explicit allow-list for exported subjects (hub→leaf direction). When non-empty,
/// only messages matching at least one of these patterns will be forwarded from
/// the hub to the leaf. Deny patterns (<see cref="DenyExports"/>) take precedence.
/// Supports wildcards (* and >).
/// Go reference: auth.go — SubjectPermission.Allow (Publish allow list).
/// </summary>
public List<string> ExportSubjects { get; set; } = [];
/// <summary>
/// Explicit allow-list for imported subjects (leaf→hub direction). When non-empty,
/// only messages matching at least one of these patterns will be forwarded from
/// the leaf to the hub. Deny patterns (<see cref="DenyImports"/>) take precedence.
/// Supports wildcards (* and >).
/// Go reference: auth.go — SubjectPermission.Allow (Subscribe allow list).
/// </summary>
public List<string> ImportSubjects { get; set; } = [];
/// <summary>List of users for leaf listener authentication (from authorization.users).</summary>
public List<NATS.Server.Auth.User>? Users { get; set; }
}